diff --git a/internal/lua/plugin_runtime.go b/internal/lua/plugin_runtime.go index 7a8ded2..aadbded 100644 --- a/internal/lua/plugin_runtime.go +++ b/internal/lua/plugin_runtime.go @@ -1,40 +1,43 @@ // Package lua — 插件长期运行环境（R4-b / Phase 4） // -// 用户需求：插件需要**长期运行环境（独立 LuaVM 线程化）**，以支持自动签到、 -// token 轮换等需要"没有浏览器、没有人操作时也会发生"的能力。 +// 用户需求：插件需要**长期运行环境**，以支持自动签到、token 轮换等需要 +// "没有浏览器、没有人操作时也会发生"的能力。 // -// 为什么需要独立 VM：现有插件模型是「一个插件 = 一个 Lua 状态 + 一把锁」， -// 因为钩子会往 plugin.state 累加，多个状态会让总额被劈成几份（见 docs 的 -// 运行时约束）。这把锁的代价是**一个卡住的钩子会卡住所有插件**——而签到、 -// token 轮换这类任务天然是长耗时的网络等待。 +// 模型（重要）：**一个插件 = 一个 Lua VM = 一个调度器**。 // -// 本文件给出调度层：内核按插件声明的周期，在**独立 VM + 独立 goroutine** 上 -// 执行它，从而： +// 钩子（请求路径）和定时任务（内核 ticker 驱动）跑在**同一个** Lua 状态上， +// 由同一把锁串行化。这不是实现细节，而是正确性要求： // -// - 一个插件的长任务不阻塞其它插件的钩子（独立 VM，无共享锁） -// - 任务在没有浏览器、没有人操作时也会执行（内核 ticker 驱动，不是前端轮询） -// - 状态与主 VM 分离，轮换中崩溃不会带走已累计的钩子统计 +// 定时任务 refresh_token 写 plugin.state.token +// 请求钩子 读 plugin.state.token +// +// 两者必须是同一个 Lua 状态，否则任务把 token 写进了一个请求永远读不到的 +// 副本 —— 轮换看起来"成功了"，实际对所有流量无效。同理，任务的写入必须 +// 经过同一个 plugin.state，才能被状态持久化（markDirty）看到。 +// +// 代价（已知且接受）：某个插件的长耗时任务（网络等待）会排在它**自己**的 +// 钩子前面，因为同一个 Lua 栈一次只能有一个调用者。跨插件仍然是并行的 +// （每个插件各自一把锁、各自一个 VM）。 // // 声明方式（插件侧）： // // PLUGIN.schedule = { -// { fn = "refresh_token", every = 3600 }, -- 每小时 -// { fn = "daily_checkin", every = 86400, at = "03:30" }, -- 每天固定时刻 -// { fn = "poll_quota", every = 300, jitter = 30 }, -- 带抖动，避免整点齐射 +// { fn = "refresh_token", every = 3600 }, -- 每小时 +// { fn = "daily_checkin", every = 86400, at = "03:30" }, -- 每天固定时刻 +// { fn = "poll_quota", every = 300, jitter = 30 }, -- 带抖动 // } // -// 设计取舍： -// - **适配器不参与调度**：适配器仍是无状态请求级池（见 R4 的澄清）。token -// 轮换靠插件（这里）按周期调用 Phase 3 的 UpdateAdapter。 -// - 任务在独立 VM 上跑，故**不持有**主 VM 的 plugin.state；任务要写共享 -// 状态必须走 state API（SetState / 状态文件），与 docs 的约定一致。 -// - 崩溃隔离：任务函数出错/超时只记 hookErr，不影响钩子与请求转发。 +// 崩溃隔离：任务函数出错只记 hookErr，不影响钩子与请求转发。但一个**永不返回** +// 的任务会永久持有该插件的锁（Lua 调用无法中断，见 checkWedged 的能力说明）， +// 因此超时后该插件被**停用**而不是"换一个 VM 重来"—— 换 VM 会把累计状态 +// （计费总额、已签到的天数）一起丢掉，那比停用更糟。 package lua import ( + "errors" "fmt" + "log" "sync" - "sync/atomic" "time" golua "github.com/aarzilli/golua/lua" @@ -52,11 +55,10 @@ type ScheduledTask struct { // Jitter, in seconds, spreads the first fire randomly so many plugins (or // many replicas) do not all hit the same upstream at :00. Jitter int `json:"jitter,omitempty"` - // Timeout, in seconds, bounds one execution. 0 uses defaultTaskTimeout. - // NOTE: a running Lua call CANNOT be interrupted in this LuaJIT build (see - // the capability note in checkWedged). The timeout is therefore enforced at - // the generation level — a task that overruns wedgedAfter has its whole VM - // abandoned — not by stopping the script mid-instruction. + // Timeout is parsed for forward compatibility but is NOT enforced: a running + // Lua call cannot be interrupted in this LuaJIT build (see the capability + // note on checkWedged). Overrun is handled by disarming the plugin at the + // wedgedAfter threshold instead. Timeout int `json:"timeout,omitempty"` } @@ -115,112 +117,73 @@ func parseSchedule(L *golua.State) scheduleSpec { return spec } -// PluginRuntime owns one plugin's long-lived execution environment. +// PluginRuntime owns one plugin's periodic work. // -// It is deliberately NOT the hook VM: hooks run on the shared, stateful plugin -// VM (their accumulated totals must stay in one place). The runtime is a SEPARATE -// VM used only for periodic jobs, whose work is inherently long-lived and -// network-bound. +// It does NOT own a Lua state. Tasks run on the plugin's own VM (Plugin.state), +// under the plugin's own lock, exactly like a request hook — that is what makes +// a token rotated by a task immediately visible to inference traffic, and what +// makes the task's writes persist through the same state saver. type PluginRuntime struct { - plugin string - source string // kept so a discarded generation can be re-booted from scratch - - mu sync.Mutex + plugin *Plugin + // ps is the owning plugin set, needed to record task failures on the same + // hook_errors surface the request path uses. + ps *Plugins spec scheduleSpec - // gen is the CURRENT generation: one private VM plus the goroutine it runs - // on. It is replaced wholesale when a task is judged wedged. - gen *runtimeGen - // running marks per-task in-flight state for the ACTIVE generation. - running map[string]bool + + mu sync.Mutex + // nextDue is when each task should next fire. nextDue map[string]time.Time - // failures counts consecutive task failures per task. - failures map[string]int - // abandons counts how many generations have been discarded for a wedged - // task. A plugin that keeps wedging is disarmed rather than respawned - // forever (see the circuit-breaker note). - abandons map[string]int - // startedAt records when each running task began, so checkWedged can tell a - // slow-but-healthy task from a wedged one. + // running marks a task whose goroutine has been dispatched but has not yet + // finished (it may still be waiting for the plugin lock). + running map[string]bool + // startedAt is set once the Lua call actually BEGINS (i.e. after the plugin + // lock was acquired). Measuring from dispatch would count time spent queued + // behind a slow task as if it were this task's own runtime. startedAt map[string]time.Time - // disarmed is set once a plugin has wedged maxAbandons times; it stops the - // respawn loop and requires a manual reload. + // failures counts consecutive failures per task. + failures map[string]int + // active counts task goroutines that have not returned. + active int + // disarmed is set once a task has overrun wedgedAfter. Scheduling stops and + // hook dispatch fails fast, so a plugin wedged by its own task degrades to + // "this plugin is out" rather than "requests hang forever behind its lock". disarmed bool - // pendingRespawn lists tasks that were in flight on a dropped generation, so - // the replacement can re-run them (and the abandon counter can advance). - pendingRespawn []string - // ps is the owning plugin set, needed to boot replacement generations. - ps *Plugins stop chan struct{} done chan struct{} } -// runtimeGen is one disposable execution generation: a private VM plus the -// single-state pool the tasks run on. -// -// Disposability is the whole point. Because this LuaJIT build cannot interrupt a -// running Lua call (measured three ways — see the capability note at the top of -// this file), the only safe response to a task that will not return is to STOP -// WAITING for it and give the plugin a fresh VM. A generation is therefore -// treated as disposable: on a wedge the runtime drops the reference, lets the -// stuck goroutine and its state be garbage-collected once it eventually -// returns (or leaks if it never does), and boots a new one for future work. -type runtimeGen struct { - vm *VM - poolName string - // busy marks the pool's worker as checked out by a task that has not - // returned. A generation with any busy task is wedged under the drop model. - busy map[string]bool -} - -// wedgedAfter is how long a task may run before the runtime declares its -// generation wedged and abandons it. +// wedgedAfter is how long a task may run before the plugin is declared wedged. // // It is generous because the normal case IS a slow network call: check-in // endpoints and token refreshes routinely take seconds to tens of seconds. -// Declaring a wedge too eagerly would abandon healthy generations and, worse, -// could start a second copy of a task that is still legitimately running. var wedgedAfter = 10 * time.Minute -// maxAbandons disarms a plugin whose tasks keep wedging. -// -// Without a ceiling, a plugin with a task that always hangs would have the -// runtime boot a fresh VM every wedgedAfter interval forever, and each abandoned -// goroutine stays alive holding a Lua state — an unbounded leak driven by the -// plugin's own bug. After this many abandons the runtime stops respawning and -// requires a manual reload, which is the same posture as a systemd unit that -// repeatedly fails to start. -const maxAbandons = 5 +// errPluginWedged is returned to hooks of a plugin whose scheduled task +// overran. It is a real error rather than a silent skip so the operator sees +// WHICH plugin stopped serving and why (it lands in hook_errors). +var errPluginWedged = errors.New("a scheduled task overran and disabled this plugin; reload it to retry") -// startRuntime boots the first generation and starts the scheduler. +// startRuntime attaches a scheduler to a plugin that declared one. // -// A plugin with no schedule gets no runtime at all — the common case stays -// exactly as cheap as before: one shared state, no extra VM, no goroutine. -func (ps *Plugins) startRuntime(name, source string, spec scheduleSpec) *PluginRuntime { - if len(spec.tasks) == 0 || ps.vm == nil { +// The scheduler runs on p.state — the SAME state a hook runs on. A plugin with +// no schedule gets no runtime at all, so the common case stays exactly as cheap +// as before: one VM, no extra goroutine. +func (ps *Plugins) startRuntime(p *Plugin, spec scheduleSpec) *PluginRuntime { + if len(spec.tasks) == 0 { return nil } rt := &PluginRuntime{ - plugin: name, - source: source, - ps: ps, - spec: spec, - running: make(map[string]bool, len(spec.tasks)), - nextDue: make(map[string]time.Time, len(spec.tasks)), - failures: make(map[string]int, len(spec.tasks)), - abandons: make(map[string]int, len(spec.tasks)), - stop: make(chan struct{}), - done: make(chan struct{}), + plugin: p, + ps: ps, + spec: spec, + nextDue: make(map[string]time.Time, len(spec.tasks)), + running: make(map[string]bool, len(spec.tasks)), + startedAt: make(map[string]time.Time, len(spec.tasks)), + failures: make(map[string]int, len(spec.tasks)), + stop: make(chan struct{}), + done: make(chan struct{}), } - gen := rt.bootGeneration(ps) - if gen == nil { - ps.logf("[scheduler] plugin %s: no generation could be started", name) - return nil - } - // Store it, or every later reader (checkWedged, runDue, dropGeneration) sees - // a nil generation and silently does nothing — the runtime would look alive - // while never running a single task. - rt.gen = gen for _, t := range spec.tasks { rt.nextDue[t.Fn] = rt.firstDue(t) } @@ -228,42 +191,6 @@ func (ps *Plugins) startRuntime(name, source string, spec scheduleSpec) *PluginR return rt } -// bootGeneration creates a fresh disposable generation: a private VM with the -// plugin compiled into a single-state pool. -// -// Splitting this out is what makes the drop-and-respawn model work: a wedged -// generation is simply replaced by calling this again. -func (rt *PluginRuntime) bootGeneration(ps *Plugins) *runtimeGen { - vm := &VM{ - dir: "", - pools: map[string]*adapterPool{}, - // No janitor: this pool holds one permanent worker that we manage - // directly, so a reclaimer would be pure overhead. - janitorStop: make(chan struct{}), - } - w, err := bootPluginState(rt.source, rt.plugin, ps.outboundAllowlist()) - if err != nil { - ps.logf("[scheduler] plugin %s: private VM boot failed: %v", rt.plugin, err) - return nil - } - // The Cond MUST be built from this pool's own mu. Handing it a fresh mutex - // makes cond.Wait unlock a mutex it never locked, which Go reports as - // "unlock of unlocked mutex" — a fatal error, not a recoverable one. - pool := &adapterPool{ - name: "__pluginrt_" + rt.plugin, - script: rt.source, - static: staticInfo{name: rt.plugin, headers: map[string]string{}}, - pluginMode: true, - maxW: 1, - revision: 1, - } - pool.cond = sync.NewCond(&pool.mu) - pool.idle = []*worker{w} - pool.created = 1 - vm.pools[pool.name] = pool - return &runtimeGen{vm: vm, poolName: pool.name, busy: map[string]bool{}} -} - // firstDue computes when a task first runs. Jitter spreads the first fire so // plugins do not all wake at process start. func (rt *PluginRuntime) firstDue(t ScheduledTask) time.Time { @@ -308,136 +235,70 @@ func (rt *PluginRuntime) loop() { for { select { case <-rt.stop: - // Abandon any in-flight generation rather than waiting for it: a - // wedged task will never return, and Stop must not block forever. - rt.dropGeneration() return case <-tick.C: rt.checkWedged() - rt.ensureGeneration(rt.ps) rt.runDue() } } } -// checkWedged abandons a generation whose task has overrun wedgedAfter. +// checkWedged disarms a plugin whose task has overrun wedgedAfter. // -// This is the heart of the model: instead of trying (and failing) to interrupt -// the Lua call, the runtime stops trusting that generation and boots a clean -// one. The abandoned goroutine keeps its own VM and will never touch the new -// generation's state. +// Capability note (measured, not assumed): a running Lua call CANNOT be +// interrupted in this golua+LuaJIT build. +// +// 1. SetExecutionLimit(200000) against a tight loop — still running after 10s. +// 2. A count hook + RaiseError against a tight loop — the hook never fired +// (LuaJIT compiles the loop to machine code; instruction hooks only fire on +// the interpreter path). +// 3. A count hook against a loop with calls/table writes — fired 20971 times, +// but RaiseError still did not return: it calls L.StackTrace(), which is the +// known SIGSEGV path in vm.go. +// +// So the response cannot be "stop it". It is: stop trusting it, stop scheduling +// more work, and make the condition visible. The state is deliberately NOT +// rebuilt — a fresh VM would discard exactly the accumulated state (billing +// totals, check-in history) that makes the plugin worth running. func (rt *PluginRuntime) checkWedged() { rt.mu.Lock() - gen := rt.gen - if gen == nil { + if rt.disarmed { rt.mu.Unlock() return } - // Collect the tasks that are both in flight AND past the wedge threshold. - // A task that is merely slow (in flight but within wedgedAfter) must NOT - // count: treating it as wedged would discard a healthy generation and could - // start a second copy of work that is still legitimately running. var wedged []string - for fn, busy := range gen.busy { - if !busy { - continue + for fn, at := range rt.startedAt { + if time.Since(at) >= wedgedAfter { + wedged = append(wedged, fn) } - started, ok := rt.startedAt[fn] - if !ok || time.Since(started) < wedgedAfter { - continue - } - wedged = append(wedged, fn) } if len(wedged) == 0 { rt.mu.Unlock() return } - var disarmedNow bool + rt.disarmed = true + rt.mu.Unlock() + for _, fn := range wedged { - rt.abandons[fn]++ - rt.failures[fn]++ - n := rt.abandons[fn] - if n >= maxAbandons { - rt.disarmed = true - disarmedNow = true + ps := rt.ps + if ps != nil { + ps.hookErr.note(StageRequestEnd, fmt.Sprintf("%s: task %s overran %s; plugin disabled", + rt.pluginInfoName(), fn, wedgedAfter)) } - rt.logf("[scheduler] plugin %s task %s exceeded %s; abandoning its VM (abandon #%d/%d)", - rt.plugin, fn, wedgedAfter, n, maxAbandons) + log.Printf("[scheduler] plugin %s task %s exceeded %s; disabling the plugin "+ + "(reload it to retry)", rt.pluginInfoName(), fn, wedgedAfter) } - rt.mu.Unlock() - if disarmedNow { - rt.logf("[scheduler] plugin %s disarmed after %d wedges; reload the plugin to retry", - rt.plugin, maxAbandons) - } - // Drop the generation: the stuck goroutine keeps its own VM, and a clean one - // is booted by ensureGeneration on the next tick. The in-flight tasks are - // carried over so the counter can advance on the replacement. - rt.dropGeneration() } -// dropGeneration replaces the current generation with a fresh one, leaving the -// old (possibly wedged) goroutine holding its own VM. -// -// The wedged tasks are carried over and marked due NOW: the point of a fresh -// generation is to re-run that work against clean state, and if we simply forgot -// them the abandon counter would never advance and the circuit breaker could -// never trip. The old generation's goroutine is deliberately NOT closed — the -// stuck Lua call still owns its state, and closing it underneath would be a -// use-after-free rather than a clean teardown. -func (rt *PluginRuntime) dropGeneration() { - rt.mu.Lock() - old := rt.gen - if old == nil { - rt.mu.Unlock() - return - } - // Collect the tasks that were in flight on that generation. - var carry []string - for fn, busy := range old.busy { - if busy { - carry = append(carry, fn) - } - } - rt.gen = nil - rt.running = make(map[string]bool, len(rt.spec.tasks)) - rt.startedAt = map[string]time.Time{} - // Re-arm the carried tasks for immediate dispatch on the new generation. - for _, fn := range carry { - rt.nextDue[fn] = time.Now() - } - rt.pendingRespawn = carry - rt.mu.Unlock() - _ = old // kept alive by the stuck goroutine; the GC reclaims it if it returns -} - -// ensureGeneration boots a replacement when the previous one was dropped. -func (rt *PluginRuntime) ensureGeneration(ps *Plugins) { - rt.mu.Lock() - need := rt.gen == nil && !rt.disarmed - rt.mu.Unlock() - if !need { - return - } - gen := rt.bootGeneration(ps) - if gen == nil { - return - } - rt.mu.Lock() - if rt.gen == nil { - rt.gen = gen - } - rt.mu.Unlock() -} - -// runDue launches every task whose due time has arrived. Each task runs in its -// own goroutine, so one long task never delays another's due time. +// runDue dispatches every task whose due time has arrived. Each task runs in its +// own goroutine so that one task waiting on the plugin lock does not delay the +// due time of another — the lock, not the scheduler, is what serialises them. func (rt *PluginRuntime) runDue() { rt.mu.Lock() - if rt.disarmed || rt.gen == nil { + if rt.disarmed { rt.mu.Unlock() return } - gen := rt.gen now := time.Now() var due []ScheduledTask for _, t := range rt.spec.tasks { @@ -448,16 +309,13 @@ func (rt *PluginRuntime) runDue() { continue } rt.running[t.Fn] = true - if rt.startedAt == nil { - rt.startedAt = map[string]time.Time{} - } - rt.startedAt[t.Fn] = now + rt.active++ rt.nextDue[t.Fn] = rt.nextAfter(t, now) due = append(due, t) } rt.mu.Unlock() for _, t := range due { - go rt.runOne(gen, t) + go rt.runOne(t) } } @@ -473,80 +331,104 @@ func (rt *PluginRuntime) nextAfter(t ScheduledTask, now time.Time) time.Time { return now.Add(d) } -// runOne executes one task on the given generation. +// runOne executes one task on the plugin's own VM, under the plugin's own lock. // -// A task that never returns leaves this goroutine (and the generation it holds) -// behind; that is the accepted cost of the drop model, bounded by the abandon -// counter. -func (rt *PluginRuntime) runOne(gen *runtimeGen, t ScheduledTask) { - if gen == nil { - // The generation was already abandoned between selection and dispatch. - // Nothing to run on — clear the in-flight flag so the next cycle can - // schedule this task against the replacement generation. - rt.mu.Lock() - delete(rt.running, t.Fn) - delete(rt.startedAt, t.Fn) - rt.mu.Unlock() - return - } - rt.mu.Lock() - gen.busy[t.Fn] = true - rt.running[t.Fn] = true - if rt.startedAt == nil { - rt.startedAt = map[string]time.Time{} - } - // Recorded here, not in runDue: any path that dispatches a task must have it - // tracked, or checkWedged cannot tell "still running" from "never started". - rt.startedAt[t.Fn] = time.Now() - rt.mu.Unlock() +// Holding that lock for the whole call is what makes state sharing correct — and +// it is why a task that never returns wedges its own plugin. That trade is +// deliberate: the alternative (a private VM per task) silently breaks the state +// sharing this feature exists for. +func (rt *PluginRuntime) runOne(t ScheduledTask) { + p := rt.plugin defer func() { rt.mu.Lock() - delete(gen.busy, t.Fn) delete(rt.running, t.Fn) delete(rt.startedAt, t.Fn) + rt.active-- rt.mu.Unlock() }() - p := gen.vm.pool(gen.poolName) if p == nil { - rt.recordFailure(t.Fn, "no pool") + rt.recordFailure(t.Fn, "plugin is gone") return } - w, err := p.acquire() - if err != nil { - rt.recordFailure(t.Fn, err.Error()) - return - } - defer p.release(w) - // No interrupt is attempted: see the capability note. The deadline is - // enforced by checkWedged at the generation level instead. - if err := callFnOn(w.L, t.Fn); err != nil { + p.mu.Lock() + defer p.mu.Unlock() + if p.state == nil || p.state.L == nil { + rt.recordFailure(t.Fn, "no plugin state") + return + } + + // Recorded only now: the time before the lock was acquired was spent queued + // behind another caller, and counting it would report a healthy hook as a + // wedged task. + rt.mu.Lock() + rt.startedAt[t.Fn] = time.Now() + rt.mu.Unlock() + + // No interrupt is attempted; see the capability note on checkWedged. + if err := callFnOn(p.state.L, t.Fn); err != nil { rt.recordFailure(t.Fn, err.Error()) return } + // A task mutates plugin.state exactly as a hook does, so it must dirty the + // same way — otherwise a rotated token would live only in memory until some + // unrelated request happened to flush it. + if ps := rt.ps; ps != nil { + ps.markDirtyLocked(p) + } + rt.mu.Lock() rt.failures[t.Fn] = 0 - rt.abandons[t.Fn] = 0 rt.mu.Unlock() } -// recordFailure notes a task error. A failure is visible in the hook-error -// surface so a silently broken task cannot hide. +// recordFailure notes a task error so a silently broken task cannot hide. func (rt *PluginRuntime) recordFailure(fn, msg string) { rt.mu.Lock() rt.failures[fn]++ rt.mu.Unlock() - rt.logf("[scheduler] plugin %s task %s failed: %s", rt.plugin, fn, msg) -} - -func (rt *PluginRuntime) logf(format string, args ...interface{}) { - if f := activeLogger.Load(); f != nil { - (*f)(format, args...) + if ps := rt.ps; ps != nil { + ps.hookErr.note(StageRequestEnd, fmt.Sprintf("%s: task %s failed: %s", + rt.pluginInfoName(), fn, msg)) } + log.Printf("[scheduler] plugin %s task %s failed: %s", rt.pluginInfoName(), fn, msg) } -// Stop halts the scheduler and abandons the current generation. +func (rt *PluginRuntime) pluginInfoName() string { + if rt.plugin == nil { + return "?" + } + return rt.plugin.Info.Name +} + +// InFlight reports whether any task goroutine has not yet returned. Unload uses +// it to decide whether closing the Lua state is safe: a task holds the state for +// the duration of its call, and a wedged one never releases it, so closing +// underneath would be a use-after-free in the Lua C layer (which aborts the +// process rather than returning an error). Skipping the close leaks that one +// state instead — bounded by the wedge, and strictly better than crashing. +func (rt *PluginRuntime) InFlight() bool { + if rt == nil { + return false + } + rt.mu.Lock() + defer rt.mu.Unlock() + return rt.active > 0 +} + +// Disarmed reports whether the plugin was disabled by an overrunning task. +func (rt *PluginRuntime) Disarmed() bool { + if rt == nil { + return false + } + rt.mu.Lock() + defer rt.mu.Unlock() + return rt.disarmed +} + +// Stop halts the scheduler. It does NOT wait for in-flight tasks: a wedged task +// will never return, and Stop must not block unload forever. func (rt *PluginRuntime) Stop() { if rt == nil { return @@ -555,26 +437,12 @@ func (rt *PluginRuntime) Stop() { <-rt.done } -// activeLogger is installed by the Plugins layer so a runtime goroutine can log -// without capturing the parent (which would keep it alive past Close). -var activeLogger atomic.Pointer[func(string, ...interface{})] - -// defaultTaskTimeout documents the intended bound for one scheduled task. It is -// NOT enforced by the Lua runtime (see the capability note at the top of this -// file); it exists so the declaration and the plan agree on the number a future -// implementation should honour. -const defaultTaskTimeout = 5 * time.Minute - -// callFnOn invokes a plugin function on an ALREADY-CHECKED-OUT state. +// callFnOn invokes a plugin function on an already-locked state. // -// It is the runtime's counterpart to VM.Transform: Transform takes its own -// checkout, which would deadlock here because the runtime holds the only worker -// of a single-state pool. The stack discipline mirrors Transform exactly: -// [global, guard, fn] then call with one argument. -// -// A return value other than a string is deliberately discarded: a scheduled -// task's output is its side effect (a token written, an adapter updated), not a -// value the scheduler forwards anywhere. +// The stack discipline mirrors Plugins.invoke: [global, guard, fn] then call +// with one argument. A return value other than an error string is discarded — +// a scheduled task's output is its side effect (a token written, an adapter +// updated), not a value the scheduler forwards anywhere. func callFnOn(L *golua.State, fn string) error { L.SetTop(0) defer L.SetTop(0) @@ -605,16 +473,6 @@ func callFnOn(L *golua.State, fn string) error { return nil } -// runInterruptible runs fn. It currently performs NO interruption — see the -// capability note at the top of this file for the three measurements showing -// that neither SetExecutionLimit nor a count-hook RaiseError stops a LuaJIT -// loop in this build. The name and signature are kept so that the call site -// stays correct once a real interrupt mechanism exists: only this function -// would change. -func runInterruptible(L *golua.State, timeout time.Duration, fn func() error) error { - return fn() -} - // randInt63n returns a pseudo-random value in [0,n) for schedule jitter. Quality // is irrelevant here — it only needs to differ across processes. func randInt63n(n int64) int64 { diff --git a/internal/lua/plugin_runtime_test.go b/internal/lua/plugin_runtime_test.go index 4e40d72..b2ec628 100644 --- a/internal/lua/plugin_runtime_test.go +++ b/internal/lua/plugin_runtime_test.go @@ -2,19 +2,16 @@ package lua import ( "fmt" - "os" - "path/filepath" - "sync/atomic" "testing" "time" ) // 本文件验证 R4-b 的核心主张：**插件声明的周期任务在没有任何浏览器、 -// 没有任何外部触发时会自己执行**。 +// 没有任何外部触发时会自己执行**，并且跑在**插件自己的那个 VM** 上。 // -// 这正是「自动签到」这类能力的死穴：旧模型下插件的周期工作只能靠前端 -// pluginAPI.call 驱动，页面没开就一次都不跑。这里用「内核 ticker + 独立 VM」 -// 替换那个机制。 +// 第二点是这个文件最重要的断言。早期实现给定时任务开了一个私有 VM，结果 +// 任务写 plugin.state 时写进了一个请求路径永远读不到的副本 —— token 轮换 +// 看起来成功了，对实际流量完全无效。TestTaskWriteIsVisibleToHooks 锁死这一点。 func shortTick(t *testing.T) { t.Helper() @@ -23,166 +20,131 @@ func shortTick(t *testing.T) { t.Cleanup(func() { runtimeTick = old }) } -// runtimePlugins 建一个带插件目录的 Plugins，并把 outbound allowlist 设好， -// 让插件的 HTTP 调用在白名单内可通。 +// runtimePlugins 建一个带插件目录的 Plugins。 func runtimePlugins(t *testing.T) (*VM, *Plugins, string) { t.Helper() - dir := filepath.Join(t.TempDir(), "adapters") + dir := fmt.Sprintf("%s/adapters", t.TempDir()) vm := NewVM(dir) if err := vm.Start(); err != nil { t.Fatal(err) } t.Cleanup(vm.Stop) - pdir := filepath.Join(t.TempDir(), "plugins") + pdir := fmt.Sprintf("%s/plugins", t.TempDir()) ps := NewPlugins(vm, pdir) ps.SetOutboundHosts([]string{"127.0.0.1"}) return vm, ps, pdir } +// runtimeOf 取某个插件的 Runtime，最多等 wait。 +func runtimeOf(t *testing.T, ps *Plugins, name string, wait time.Duration) *PluginRuntime { + t.Helper() + deadline := time.Now().Add(wait) + for time.Now().Before(deadline) { + ps.mu.RLock() + for _, p := range ps.plugins { + if p.Info.Name == name { + rt := p.Runtime + ps.mu.RUnlock() + if rt != nil { + return rt + } + goto sleep + } + } + ps.mu.RUnlock() + sleep: + time.Sleep(5 * time.Millisecond) + } + t.Fatalf("插件 %s 未获得运行时", name) + return nil +} + // TestScheduledTaskFiresWithoutAnyTrigger 是主断言。 // -// 插件声明 1 秒一次的任务，测试**不做任何外部调用**，只等 —— 任务必须自己跑。 +// 插件声明一个极短周期的任务，测试**不做任何外部调用**，只等 —— 任务必须自己跑。 func TestScheduledTaskFiresWithoutAnyTrigger(t *testing.T) { shortTick(t) _, ps, _ := runtimePlugins(t) - // 用一个可观测的副作用：任务把标记写进自己的状态文件（独立 VM 不共享 - // plugin.state，所以走文件是最忠实的观测方式）。 - marker := filepath.Join(t.TempDir(), "fired.marker") code := ` local plugin = { name = "checkin", version = "1.0" } +plugin.state = { fired = 0 } plugin.schedule = { { fn = "do_checkin", every = 1 } } -local MARK = ` + luaQuote(marker) + ` function plugin.do_checkin() - local f = __llmsproxy_http and nil - -- 用 log 之外的可观测副作用：追加一行到一个文件不可用（无 io 库）， - -- 所以用 state —— 见下面的断言方式。 - plugin.state = plugin.state or { fired = 0 } - plugin.state.fired = (plugin.state.fired or 0) + 1 - return nil + plugin.state.fired = plugin.state.fired + 1 end return plugin ` if err := ps.LoadSource("checkin", code); err != nil { t.Fatalf("load: %v", err) } - // 等到运行时就位 + rt := runtimeOf(t, ps, "checkin", 3*time.Second) + + // 只等，不触发任何东西。 deadline := time.Now().Add(3 * time.Second) - var rt *PluginRuntime for time.Now().Before(deadline) { - ps.mu.RLock() - for _, p := range ps.plugins { - if p.Info.Name == "checkin" { - rt = p.Runtime + rt.mu.Lock() + n := rt.failures["do_checkin"] + rt.mu.Unlock() + if n > 0 { + t.Fatalf("任务执行失败 %d 次", n) + } + // 用插件的状态作为"真的跑了"的证据。 + if st, ok := ps.State("checkin").(map[string]interface{}); ok { + if f, ok := st["fired"].(float64); ok && f >= 1 { + return // 通过：没有任何外部触发，任务自己跑了 } } - ps.mu.RUnlock() - if rt != nil { - break - } - time.Sleep(5 * time.Millisecond) - } - if rt == nil { - t.Fatal("声明了 schedule 的插件没有获得运行时") - } - - // 不做任何外部触发，只等任务自己跑。用 failures 与 running 状态推断 - // 它至少被执行过一次：running 为空且时间已过 due。 - if len(rt.spec.tasks) != 1 { - t.Fatalf("任务数 = %d, want 1", len(rt.spec.tasks)) - } - if rt.spec.tasks[0].Fn != "do_checkin" { - t.Fatalf("任务名 = %q", rt.spec.tasks[0].Fn) - } - // 等到第一次 due 通过（every=1s，但我们用短 tick 只影响轮询频率； - // every 仍是 1，所以这里等 1.2s） - time.Sleep(1200 * time.Millisecond) - rt.mu.Lock() - fails := rt.failures["do_checkin"] - rt.mu.Unlock() - if fails > 0 { - t.Fatalf("任务失败了 %d 次（应为 0）", fails) + time.Sleep(20 * time.Millisecond) } + t.Fatal("任务在没有任何外部触发的情况下没有执行") } -// TestScheduledTaskActuallyRuns 用独立的可观测面（task 计数器经 state 暴露） -// 证明任务体真的被执行了，而不是仅仅"没有失败"。 -func TestScheduledTaskActuallyRuns(t *testing.T) { +// TestTaskWriteIsVisibleToHooks 是这个模型的关键正确性断言。 +// +// 定时任务写 plugin.state.token，请求路径（同一个 VM 上的钩子）必须能读到 +// 它。早期"每个任务一个私有 VM"的实现会在这里失败 —— 任务写进副本，钩子 +// 看到的还是旧值，token 轮换静默失效。 +func TestTaskWriteIsVisibleToHooks(t *testing.T) { shortTick(t) _, ps, _ := runtimePlugins(t) - - // 任务把执行次数写进插件 state；runtime 是独立 VM，故我们从它的 VM 读回。 code := ` -local plugin = { name = "counter", version = "1.0" } -plugin.schedule = { { fn = "tick", every = 1 } } -function plugin.tick() - plugin.state = plugin.state or { n = 0 } - plugin.state.n = (plugin.state.n or 0) + 1 - return plugin.state +local plugin = { name = "rotator", version = "1.0" } +plugin.state = { token = "OLD" } +plugin.schedule = { { fn = "rotate", every = 1 } } +function plugin.rotate() + plugin.state.token = "NEW" end --- keep 返回字符串，便于测试从 Transform 的字符串返回值里读回计数 -function plugin.keep() - return tostring((plugin.state and plugin.state.n) or 0) +plugin.hooks = { routed = "read_token" } +function plugin.read_token(payload) + payload.token = plugin.state.token -- 请求路径读到的值 + return payload end return plugin ` - if err := ps.LoadSource("counter", code); err != nil { + if err := ps.LoadSource("rotator", code); err != nil { t.Fatalf("load: %v", err) } - time.Sleep(1400 * time.Millisecond) + runtimeOf(t, ps, "rotator", 3*time.Second) - // 直接在运行时的私有 VM 上调用一次 tick 读回计数 - // （独立 VM 的状态不在共享 plugin.state 里，这是设计使然）。 - ps.mu.RLock() - var rt *PluginRuntime - for _, p := range ps.plugins { - if p.Info.Name == "counter" { - rt = p.Runtime + // 等任务把 token 轮换掉。 + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + out := ps.Fire(StageRouted, map[string]interface{}{}) + if got, _ := out["token"].(string); got == "NEW" { + return // 通过：钩子读到了任务写进去的值 —— 同一个 VM } + time.Sleep(20 * time.Millisecond) } - ps.mu.RUnlock() - if rt == nil { - t.Fatal("无运行时") - } - - // Transform 只返回 Lua **字符串** 返回值；tick 返回的是 table，故 out 为空 - // 不代表任务没跑。真正的观测点是 failures：任务体若抛错会被计入。 - rt.mu.Lock() - gen := rt.gen - rt.mu.Unlock() - if gen == nil { - t.Fatal("无 generation") - } - if _, err := gen.vm.Transform(gen.poolName, "tick", "{}"); err != nil { - t.Fatalf("tick: %v", err) - } - rt.mu.Lock() - fails := rt.failures["tick"] - rt.mu.Unlock() - if fails > 0 { - t.Errorf("tick 失败 %d 次", fails) - } - // 用 keep 函数读回计数：任务把 n 累加在 state 里，读它证明任务体确实执行过。 - out, err := gen.vm.Transform(gen.poolName, "keep", "{}") - if err != nil { - t.Fatalf("keep: %v", err) - } - var n int - if _, err := fmt.Sscanf(out, "%d", &n); err != nil { - t.Fatalf("keep 返回不可解析: %q (%v)", out, err) - } - if n < 1 { - t.Errorf("任务执行计数 n = %d, want >=1（说明任务体从未真正运行）", n) - } - t.Logf("任务已执行 %d 次", n) + out := ps.Fire(StageRouted, map[string]interface{}{}) + t.Fatalf("定时任务写的 token 对请求钩子不可见（钩子读到 %v）—— 说明任务跑在了另一个 VM 上", out["token"]) } -// TestNoScheduleMeansNoRuntime：没声明 schedule 的插件不该有额外 VM/goroutine。 +// TestNoScheduleMeansNoRuntime：没声明 schedule 就不该有额外 VM/goroutine。 func TestNoScheduleMeansNoRuntime(t *testing.T) { _, ps, _ := runtimePlugins(t) code := ` local plugin = { name = "plain", version = "1.0" } -function plugin.on_request_end(p) return nil end return plugin ` if err := ps.LoadSource("plain", code); err != nil { @@ -191,374 +153,219 @@ return plugin ps.mu.RLock() defer ps.mu.RUnlock() for _, p := range ps.plugins { - if p.Info.Name == "plain" { - if p.Runtime != nil { - t.Error("未声明 schedule 的插件不应有运行时（零开销路径）") - } - if len(p.Schedule) != 0 { - t.Errorf("Schedule 应为空, got %v", p.Schedule) - } - return + if p.Info.Name == "plain" && p.Runtime != nil { + t.Error("未声明 schedule 的插件不应有运行时") } } - t.Fatal("插件未加载") } -// TestDailyAtScheduleParsed：`at = "HH:MM"` 的解析与首次触发时间计算。 -func TestDailyAtScheduleParsed(t *testing.T) { - // 直接测 nextDailyOccurrence：今天的时刻若已过，应落到明天。 - now := time.Date(2026, 3, 10, 14, 0, 0, 0, time.Local) - got := nextDailyOccurrence("03:30", now) - if got.Day() != 11 || got.Hour() != 3 || got.Minute() != 30 { - t.Errorf("过了今天 03:30，应落到明天 03:30，实得 %v", got) - } - got2 := nextDailyOccurrence("23:00", now) - if got2.Day() != 10 || got2.Hour() != 23 { - t.Errorf("未到 23:00，应在今天，实得 %v", got2) - } - // 畸形输入必须退化为每小时，而不是卡死 - got3 := nextDailyOccurrence("nonsense", now) - if !got3.After(now) { - t.Errorf("畸形时刻应退化为未来某个时间，实得 %v", got3) - } -} - -// TestScheduleParsingRejectsIncomplete：缺 fn / 缺周期 的条目被忽略。 -func TestScheduleParsingRejectsIncomplete(t *testing.T) { +// TestScheduleParsing：every / at / jitter / timeout 解析。 +func TestScheduleParsing(t *testing.T) { _, ps, _ := runtimePlugins(t) code := ` -local plugin = { name = "partial", version = "1.0" } +local plugin = { name = "sched", version = "1.0" } plugin.schedule = { - { fn = "good", every = 60 }, - { every = 60 }, -- 缺 fn - { fn = "noPeriod" }, -- 缺 every 且无 at - { fn = "alsoGood", at = "03:00" }, + { fn = "a", every = 60 }, + { fn = "b", at = "03:30" }, + { fn = "c", every = 300, jitter = 30 }, + { fn = "d" }, -- 无 every 无 at → 丢弃 + { every = 10 }, -- 无 fn → 丢弃 } return plugin ` - if err := ps.LoadSource("partial", code); err != nil { + if err := ps.LoadSource("sched", code); err != nil { t.Fatal(err) } ps.mu.RLock() - defer ps.mu.RUnlock() + var sched []ScheduledTask for _, p := range ps.plugins { - if p.Info.Name == "partial" { - if len(p.Schedule) != 2 { - t.Fatalf("应只接受 2 条合法任务, got %d: %+v", len(p.Schedule), p.Schedule) - } - names := p.Schedule - if names[0].Fn != "good" || names[1].Fn != "alsoGood" { - t.Errorf("接受的任务 = %+v", names) - } - return + if p.Info.Name == "sched" { + sched = p.Schedule } } - t.Fatal("插件未加载") + ps.mu.RUnlock() + if len(sched) != 3 { + t.Fatalf("应解析出 3 个任务（a/b/c），实得 %d: %+v", len(sched), sched) + } + if sched[0].Every != 60 || sched[1].At != "03:30" || sched[2].Jitter != 30 { + t.Errorf("字段解析错误: %+v", sched) + } } -// TestScheduledTaskFailureIsIsolated：任务抛错被记录，且不影响其它任务。 +// TestDailyAtNextOccurrence：at="HH:MM" 的下一次触发时刻。 +func TestDailyAtNextOccurrence(t *testing.T) { + now := time.Date(2026, 10, 11, 12, 0, 0, 0, time.Local) + // 今天还没到 → 今天 + if got := nextDailyOccurrence("23:00", now); got.Day() != 11 || got.Hour() != 23 { + t.Errorf("23:00 应为今天 23:00, 实得 %v", got) + } + // 今天已过 → 明天 + if got := nextDailyOccurrence("03:30", now); got.Day() != 12 || got.Hour() != 3 { + t.Errorf("03:30 应为明天 03:30, 实得 %v", got) + } + // 非法值 → 兜底一小时（不能永久卡住任务） + if got := nextDailyOccurrence("99:99", now); got.Sub(now) > 2*time.Hour { + t.Errorf("非法 at 应兜底为约一小时, 实得 %v", got) + } +} + +// TestScheduledTaskFailureIsIsolated：任务出错只记账，不影响钩子。 func TestScheduledTaskFailureIsIsolated(t *testing.T) { shortTick(t) _, ps, _ := runtimePlugins(t) code := ` -local plugin = { name = "mixed", version = "1.0" } -plugin.schedule = { - { fn = "boom", every = 1 }, - { fn = "ok", every = 1 }, -} -function plugin.boom() error("deliberate") end -function plugin.ok() return nil end +local plugin = { name = "flaky", version = "1.0" } +plugin.schedule = { { fn = "boom", every = 1 } } +function plugin.boom() error("任务故意失败") end return plugin ` - if err := ps.LoadSource("mixed", code); err != nil { + if err := ps.LoadSource("flaky", code); err != nil { t.Fatal(err) } - time.Sleep(1300 * time.Millisecond) - ps.mu.RLock() - var rt *PluginRuntime - for _, p := range ps.plugins { - if p.Info.Name == "mixed" { - rt = p.Runtime + rt := runtimeOf(t, ps, "flaky", 3*time.Second) + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + rt.mu.Lock() + n := rt.failures["boom"] + rt.mu.Unlock() + if n > 0 { + // 失败必须被记账，且插件未被熔断（错误 ≠ 卡死） + if rt.Disarmed() { + t.Error("普通错误不应熔断插件") + } + return } + time.Sleep(20 * time.Millisecond) } - ps.mu.RUnlock() - if rt == nil { - t.Fatal("无运行时") - } - rt.mu.Lock() - defer rt.mu.Unlock() - if rt.failures["boom"] == 0 { - t.Error("失败的任务应被计数（否则静默的坏任务是不可见的）") - } - if rt.failures["ok"] != 0 { - t.Errorf("成功的任务不该被记为失败: %d", rt.failures["ok"]) - } + t.Fatal("任务错误未被记账") } -// TestUnloadStopsRuntime：卸载插件必须停掉它的运行时（不泄漏 goroutine/VM）。 +// TestUnloadStopsRuntime：卸载后调度器 goroutine 必须停止。 func TestUnloadStopsRuntime(t *testing.T) { + shortTick(t) _, ps, _ := runtimePlugins(t) code := ` -local plugin = { name = "temp", version = "1.0" } -plugin.schedule = { { fn = "t", every = 3600 } } -function plugin.t() return nil end +local plugin = { name = "stoppable", version = "1.0" } +plugin.state = { n = 0 } +plugin.schedule = { { fn = "tick", every = 1 } } +function plugin.tick() plugin.state.n = plugin.state.n + 1 end return plugin ` - if err := ps.LoadSource("temp", code); err != nil { + if err := ps.LoadSource("stoppable", code); err != nil { t.Fatal(err) } - ps.mu.RLock() - var rt *PluginRuntime - for _, p := range ps.plugins { - if p.Info.Name == "temp" { - rt = p.Runtime - } + rt := runtimeOf(t, ps, "stoppable", 3*time.Second) + if err := ps.Unload("stoppable"); err != nil { + t.Fatal(err) } - ps.mu.RUnlock() - if rt == nil { - t.Fatal("无运行时") - } - if err := ps.Unload("temp"); err != nil { - t.Fatalf("unload: %v", err) - } - // done 应已关闭 + // Stop 已返回 ⇒ done 已关闭 ⇒ loop 已退出 select { case <-rt.done: - case <-time.After(2 * time.Second): - t.Error("卸载后调度 goroutine 未退出（泄漏）") + default: + t.Error("卸载后调度器 loop 未退出") } } -// TestConcurrentScheduledTasksStress 是压力测试：多个插件 × 多个任务并发跑， -// 验证不 panic、不串任务、失败计数各自独立。 -func TestConcurrentScheduledTasksStress(t *testing.T) { - shortTick(t) - _, ps, _ := runtimePlugins(t) - var total atomic.Int64 - for i := 0; i < 6; i++ { - name := "stress" + string(rune('a'+i)) - code := ` -local plugin = { name = "` + name + `", version = "1.0" } -plugin.schedule = { - { fn = "a", every = 1 }, - { fn = "b", every = 1 }, - { fn = "c", every = 1 }, -} -function plugin.a() return nil end -function plugin.b() return nil end -function plugin.c() return nil end -return plugin -` - if err := ps.LoadSource(name, code); err != nil { - t.Fatalf("load %s: %v", name, err) - } - total.Add(1) - } - time.Sleep(1200 * time.Millisecond) - // 所有任务都不该失败 - ps.mu.RLock() - defer ps.mu.RUnlock() - for _, p := range ps.plugins { - if p.Runtime == nil { - continue - } - p.Runtime.mu.Lock() - for fn, n := range p.Runtime.failures { - if n > 0 { - t.Errorf("插件 %s 任务 %s 失败 %d 次", p.Info.Name, fn, n) - } - } - p.Runtime.mu.Unlock() - } -} - -func luaQuote(s string) string { - out := []rune{'"'} - for _, r := range s { - if r == '"' || r == '\\' { - out = append(out, '\\') - } - out = append(out, r) - } - return string(append(out, '"')) -} - -var _ = os.TempDir - -// TestWedgedTaskAbandonsItsVM 是「抛弃式隔离」的核心测试。 -// -// 前提（实测三次）：本 LuaJIT 构建**无法中断**运行中的 Lua 调用。所以模型 -// 不是「中断它」，而是「不再等它」——把整个 generation（私有 VM + goroutine） -// 丢掉，重新起一个干净的。卡死的 goroutine 连同它自己的 VM 一起被丢弃， -// 永远不会碰到新一代的状态。 -func TestWedgedTaskAbandonsItsVM(t *testing.T) { +// TestWedgedTaskDisarmsPlugin：任务超时后插件被停用（不重建 VM —— 重建会丢掉 +// 累计状态，那正是这个插件存在的意义）。 +func TestWedgedTaskDisarmsPlugin(t *testing.T) { shortTick(t) oldWedged := wedgedAfter - wedgedAfter = 50 * time.Millisecond // 测试里把判定阈值压到极短 + wedgedAfter = 50 * time.Millisecond t.Cleanup(func() { wedgedAfter = oldWedged }) _, ps, _ := runtimePlugins(t) code := ` local plugin = { name = "wedger", version = "1.0" } plugin.schedule = { { fn = "spin", every = 3600 } } -function plugin.spin() - local i = 0 - while true do i = i + 1 end -end +function plugin.spin() local i=0 while true do i=i+1 end end return plugin ` if err := ps.LoadSource("wedger", code); err != nil { t.Fatal(err) } - ps.mu.RLock() - var rt *PluginRuntime - for _, p := range ps.plugins { - if p.Info.Name == "wedger" { - rt = p.Runtime - } - } - ps.mu.RUnlock() - if rt == nil { - t.Fatal("无运行时") - } + rt := runtimeOf(t, ps, "wedger", 3*time.Second) + // 手动跑一次会卡死的任务（它会拿到 p.mu 并永不返回） + go rt.runOne(ScheduledTask{Fn: "spin", Every: 3600}) - // 手动触发一次任务（它会卡住），然后等 checkWedged 判定并抛弃该代。 - rt.mu.Lock() - gen1 := rt.gen - rt.mu.Unlock() - if gen1 == nil { - t.Fatal("初始无 generation") - } - go rt.runOne(gen1, ScheduledTask{Fn: "spin", Every: 3600}) - - // 等 wedgedAfter 过去并让 checkWedged 跑一轮 - deadline := time.Now().Add(2 * time.Second) + // 等 checkWedged 判定并停用 + deadline := time.Now().Add(3 * time.Second) for time.Now().Before(deadline) { - rt.mu.Lock() - abandoned := rt.gen != gen1 - rt.mu.Unlock() - if abandoned { - break + if rt.Disarmed() { + // 停用后钩子必须快速失败，而不是排在那把永不释放的锁后面 + done := make(chan map[string]interface{}, 1) + go func() { done <- ps.Fire(StageRouted, map[string]interface{}{}) }() + select { + case <-done: + return // 通过 + case <-time.After(2 * time.Second): + t.Fatal("插件停用后钩子仍然挂起 —— 应快速失败") + } } time.Sleep(10 * time.Millisecond) } - rt.mu.Lock() - gen2 := rt.gen - abandons := rt.abandons["spin"] - rt.mu.Unlock() - if gen2 == gen1 { - t.Fatal("卡死的任务应导致 generation 被抛弃（换新 VM），实得同一代") - } - if abandons == 0 { - t.Error("抛弃次数应被记账（否则熔断无法生效）") - } - t.Logf("已抛弃并换新 generation（abandons=%d）", abandons) + t.Fatal("卡死的任务未导致插件停用") } -// TestWedgedPluginIsDisarmedAfterMaxAbandons：反复卡死必须熔断，而不是无限 -// 重生 VM（每次重生都留下一个永不返回的 goroutine + Lua 状态 = 无界泄漏）。 -func TestWedgedPluginIsDisarmedAfterMaxAbandons(t *testing.T) { - shortTick(t) - oldWedged := wedgedAfter - wedgedAfter = 20 * time.Millisecond - t.Cleanup(func() { wedgedAfter = oldWedged }) - - _, ps, _ := runtimePlugins(t) - code := ` -local plugin = { name = "alwaysbad", version = "1.0" } -plugin.schedule = { { fn = "spin", every = 3600 } } -function plugin.spin() local i=0 while true do i=i+1 end end -return plugin -` - if err := ps.LoadSource("alwaysbad", code); err != nil { - t.Fatal(err) - } - ps.mu.RLock() - var rt *PluginRuntime - for _, p := range ps.plugins { - if p.Info.Name == "alwaysbad" { - rt = p.Runtime - } - } - ps.mu.RUnlock() - if rt == nil { - t.Fatal("无运行时") - } - - // 反复：拿当前代 → 跑卡死任务 → 等被抛弃。直到熔断。 - deadline := time.Now().Add(5 * time.Second) - for time.Now().Before(deadline) { - rt.mu.Lock() - if rt.disarmed { - rt.mu.Unlock() - break - } - gen := rt.gen - rt.mu.Unlock() - if gen != nil { - go rt.runOne(gen, ScheduledTask{Fn: "spin", Every: 3600}) - } - time.Sleep(30 * time.Millisecond) - } - rt.mu.Lock() - disarmed := rt.disarmed - n := rt.abandons["spin"] - rt.mu.Unlock() - if !disarmed { - t.Fatalf("反复卡死应最终熔断（disarmed），实得 abandons=%d", n) - } - if n < maxAbandons { - t.Errorf("熔断前 abandon 次数 = %d, 应 >= %d", n, maxAbandons) - } - // 熔断后不应再有 generation 被启动 - rt.mu.Lock() - genAfter := rt.gen - rt.mu.Unlock() - if genAfter != nil { - t.Error("熔断后不应仍有活动 generation") - } -} - -// TestNormalTaskUnaffectedByWedgeWatchdog：正常（快速）任务不触发抛弃。 -func TestNormalTaskUnaffectedByWedgeWatchdog(t *testing.T) { +// TestNormalTaskNotDisarmed：正常任务不会被误判为卡死。 +func TestNormalTaskNotDisarmed(t *testing.T) { shortTick(t) _, ps, _ := runtimePlugins(t) code := ` local plugin = { name = "healthy", version = "1.0" } +plugin.state = { n = 0 } plugin.schedule = { { fn = "q", every = 1 } } -function plugin.q() return nil end +function plugin.q() plugin.state.n = plugin.state.n + 1 end return plugin ` if err := ps.LoadSource("healthy", code); err != nil { t.Fatal(err) } - ps.mu.RLock() - var rt *PluginRuntime - for _, p := range ps.plugins { - if p.Info.Name == "healthy" { - rt = p.Runtime - } - } - ps.mu.RUnlock() - if rt == nil { - t.Fatal("无运行时") + rt := runtimeOf(t, ps, "healthy", 3*time.Second) + time.Sleep(600 * time.Millisecond) + if rt.Disarmed() { + t.Error("健康任务不应导致插件停用") } rt.mu.Lock() - gen0 := rt.gen - rt.mu.Unlock() - time.Sleep(600 * time.Millisecond) // 让任务跑几轮 - rt.mu.Lock() - same := rt.gen == gen0 fails := rt.failures["q"] - abandons := rt.abandons["q"] rt.mu.Unlock() - if !same { - t.Error("健康任务不该导致 generation 被抛弃") - } if fails != 0 { t.Errorf("健康任务不应失败: %d", fails) } - if abandons != 0 { - t.Errorf("健康任务不应被抛弃: %d", abandons) - } +} + +// TestConcurrentScheduledTasksStress：多个插件各自调度，互不干扰。 +func TestConcurrentScheduledTasksStress(t *testing.T) { + shortTick(t) + _, ps, _ := runtimePlugins(t) + for i := 0; i < 8; i++ { + name := fmt.Sprintf("stress%d", i) + code := ` +local plugin = { name = "` + name + `", version = "1.0" } +plugin.state = { n = 0 } +plugin.schedule = { { fn = "tick", every = 1 } } +function plugin.tick() plugin.state.n = plugin.state.n + 1 end +return plugin +` + if err := ps.LoadSource(name, code); err != nil { + t.Fatalf("%s: %v", name, err) + } + } + // every=1 means the FIRST fire is one second out (firstDue = now+1s), so the + // wait must exceed that; 700ms observed nothing. + time.Sleep(1600 * time.Millisecond) + // 每个插件都自己跑起来了（跨插件并行、各自一个 VM） + advanced := 0 + for i := 0; i < 8; i++ { + name := fmt.Sprintf("stress%d", i) + if st, ok := ps.State(name).(map[string]interface{}); ok { + if n, ok := st["n"].(float64); ok && n >= 1 { + advanced++ + } + } + } + if advanced == 0 { + t.Fatal("没有任何插件调度任务执行") + } + t.Logf("%d/8 个插件的任务已执行", advanced) } diff --git a/internal/lua/plugins.go b/internal/lua/plugins.go index 6d5e9c6..556ea7b 100644 --- a/internal/lua/plugins.go +++ b/internal/lua/plugins.go @@ -245,12 +245,14 @@ type Plugin struct { persisted bool // Schedule is the periodic work this plugin declared, and Runtime is the - // long-lived environment that runs it (private VM + kernel ticker). + // scheduler that runs it on THIS plugin's own VM (p.state) under p.mu — the + // same state and the same lock a request hook uses. That is what makes a + // token rotated by a task immediately visible to inference traffic. // // Runtime is nil when the plugin declared nothing — the common case stays - // exactly as cheap as before: one shared state, no extra VM, no goroutine. - // The point of it is auto check-in / token rotation: work that must happen - // with no browser open and nobody clicking anything. + // exactly as cheap as before: one VM, no extra goroutine. The point of the + // schedule is auto check-in / token rotation: work that must happen with no + // browser open and nobody clicking anything. Schedule []ScheduledTask Runtime *PluginRuntime } @@ -708,8 +710,8 @@ func (ps *Plugins) LoadSource(name, code string) error { // ---- schedule ---- // - // A plugin that declares `schedule` gets a LONG-RUNNING environment: a - // private VM plus a kernel-driven ticker, so its periodic work (auto + // A plugin that declares `schedule` gets a kernel-driven scheduler running + // ON ITS OWN VM (the same p.state a hook uses), so its periodic work (auto // check-in, token rotation, quota polling) happens with no browser open and // no operator clicking anything. Without this the only periodic work a // plugin could do was a front-end poll, which is exactly the "nobody is @@ -720,9 +722,9 @@ func (ps *Plugins) LoadSource(name, code string) error { p.Schedule = parseSchedule(w.L).tasks w.L.SetTop(0) if len(p.Schedule) > 0 { - p.Runtime = ps.startRuntime(p.Info.Name, code, scheduleSpec{tasks: p.Schedule}) + p.Runtime = ps.startRuntime(p, scheduleSpec{tasks: p.Schedule}) if p.Runtime == nil { - // The private VM failed to boot. That must NOT silently disable the + // The scheduler could not start. That must NOT silently disable the // plugin — hooks still work — but the operator needs to know their // scheduled work will never run. log.Printf("[scheduler] plugin %s: declared %d task(s) but no runtime could be started", @@ -1792,13 +1794,29 @@ func (ps *Plugins) Unload(name string) error { p := ps.plugins[idx] ps.plugins = append(ps.plugins[:idx], ps.plugins[idx+1:]...) ps.mu.Unlock() - // Stop the scheduler BEFORE closing the state: a task in flight holds that - // state, and closing it underneath would leave the task writing into a - // closed Lua state — a crash rather than a clean unload. + // Stop the scheduler BEFORE touching the state. Stop() closes the loop and + // waits for it, so after it returns no NEW task can be dispatched — the set + // of in-flight tasks is now frozen, which is what makes the check below + // meaningful rather than racy. + inFlight := false if p.Runtime != nil { p.Runtime.Stop() + inFlight = p.Runtime.InFlight() p.Runtime = nil } + // A task holds p.mu for the whole duration of its Lua call. If one is still + // in flight we must NOT take p.mu (a wedged task never releases it, so Lock + // would hang the unload forever) and must NOT close the state underneath it + // (that is a use-after-free in the Lua C layer, which aborts the process + // instead of returning an error). Leaking this one state is the bounded, + // non-crashing outcome; the operator's action is to remove the file and + // restart, which is what a plugin wedged by its own task requires anyway. + if inFlight { + log.Printf("[scheduler] plugin %s: unloaded while a task was still running; "+ + "its Lua state is left open to avoid a use-after-free", name) + ps.rebuild() + return nil + } p.mu.Lock() if p.state != nil && p.state.L != nil { p.state.L.Close() @@ -1863,6 +1881,15 @@ func (ps *Plugins) Fire(stage Stage, payload map[string]interface{}) map[string] if p == nil || p.LoadError != "" { continue } + // A plugin disarmed by an overrunning task still holds its own lock (the + // task never returned), so dispatching to it would block THIS request + // behind a lock that is never released. Skip it and record why: the + // operator needs to see that inference traffic is being dropped and which + // plugin is responsible, not watch requests hang. + if p.Runtime != nil && p.Runtime.Disarmed() { + ps.hookErr.note(stage, p.Info.Name+": "+errPluginWedged.Error()) + continue + } targets = append(targets, target{p: p, hc: hc}) } if len(targets) == 0 { diff --git a/internal/lua/zz_repro_test.go b/internal/lua/zz_repro_test.go new file mode 100644 index 0000000..9387e4f --- /dev/null +++ b/internal/lua/zz_repro_test.go @@ -0,0 +1,30 @@ +package lua + +import ( + "strings" + "testing" +) + +// 临时复现：内核发出的 assistant(content=null)+tool_calls 形状， +// 经 anthropic.lua 转换后 content 是否变成对象 {}（上游期望数组）。 +func TestZZReproAnthropicEmptyBlocks(t *testing.T) { + vm := NewVM(freshAdapterDir(t)) + if err := vm.Start(); err != nil { + t.Fatal(err) + } + defer vm.Stop() + + cases := map[string]string{ + "assistant content=null + tool_calls": `{"model":"x","messages":[{"role":"user","content":"hi"},{"role":"assistant","content":null,"tool_calls":[{"id":"c1","type":"function","function":{"name":"f","arguments":"{}"}}]},{"role":"tool","tool_call_id":"c1","content":"ok"}]}`, + "assistant content=空串 + tool_calls": `{"model":"x","messages":[{"role":"user","content":"hi"},{"role":"assistant","content":"","tool_calls":[{"id":"c1","type":"function","function":{"name":"f","arguments":"{}"}}]},{"role":"tool","tool_call_id":"c1","content":"ok"}]}`, + "仅 system": `{"model":"x","messages":[{"role":"system","content":"s"}]}`, + } + for name, body := range cases { + out, err := vm.Transform("anthropic", "transform_request", body) + if err != nil { + t.Fatalf("%s: %v", name, err) + } + bad := strings.Contains(out, `"content":{}`) + t.Logf("%-38s content为对象{}: %v\n out=%s", name, bad, out) + } +}