From 9016eda9059458538ead622de30d6e42f5988e5e Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Sun, 11 Oct 2026 10:40:58 +0800 Subject: [PATCH] =?UTF-8?q?feat(lua):=20=E5=8D=A1=E6=AD=BB=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=E6=94=B9=E4=B8=BA=E3=80=8C=E6=8A=9B=E5=BC=83=E6=95=B4?= =?UTF-8?q?=E4=B8=AA=20VM=E3=80=8D=E2=80=94=E2=80=94=20=E4=B8=8D=E4=B8=AD?= =?UTF-8?q?=E6=96=AD=EF=BC=8C=E7=9B=B4=E6=8E=A5=E6=8D=A2=E4=B8=80=E4=BB=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 用户给的架构决策：与其想尽办法中断卡死的插件任务，不如直接抛弃它所在的 VM，让任务的 goroutine 跟 VM 一起被处置。这比中断干净 —— VM 重建只要 ~0.1ms，抛弃比修复便宜。 ## 为什么必须这样做（三次实测） 本 golua + LuaJIT 构建**无法中断**运行中的 Lua 调用： 1. SetExecutionLimit(200000) 对紧密循环 10s 未返回 —— 无效 2. 计数 hook(2000) + RaiseError 对紧密循环 fired=0 —— hook 未触发 3. 计数 hook(1000) 对「函数调用+表写」循环 fired=20971 触发， 但 RaiseError 后仍未返回 —— RaiseError 走 L.StackTrace()， 正是 vm.go 已记录的 SIGSEGV / 无法跨 C 栈路径 原因：LuaJIT 把紧密循环编译成机器码，指令计数 hook 只在解释器路径触发； 而唯一能在 C 栈上中断的 golua 入口恰好依赖 StackTrace()。 ## 模型：generation 可抛弃 PluginRuntime 持有一个 runtimeGen（私有 VM + 单状态池）。任务超过 wedgedAfter（10min，因为正常情况就是慢网络调用）仍不返回时： - 抛弃该 generation，由 ensureGeneration 在下一 tick 起一个干净的 - 卡死的 goroutine 留在旧 VM 上，永远碰不到新一代的状态 - 在飞任务被 carry 到新 generation 并重新排期（否则 abandon 计数永不前进， 熔断也就永远不触发） ## 熔断（必须有） 每次抛弃都留下一个永不返回的 goroutine + Lua 状态。没有上限的话，一个 必然卡死的任务会让运行时每 10 分钟建一个新 VM —— 由插件自身 bug 驱动的 无界泄漏。maxAbandons=5 后熔断停止重生，需人工 reload（同 systemd 单元 反复启动失败后的姿态）。 ## 开发中修掉的四个缺陷 1. rt.gen 未赋值：bootGeneration 的结果只赋给局部变量，导致 checkWedged/ runDue/dropGeneration 全部读到 nil，运行时会「活着但一个任务都不跑」。 2. checkWedged 无条件抛弃：即使没有任何任务判定为 wedged 也 drop， 让健康插件不断换代。 3. startedAt 由 runDue 设置而非 runOne：任何其它 dispatch 路径的任务 都不会被追踪，checkWedged 无法区分「在跑」与「从未启动」。 4. runOne 收到 nil gen 会 panic（抛弃与派发竞争）—— 改为安全返回并清标志。 测试：13 个（无触发自动执行/真正执行/零开销/at 解析/条目校验/失败隔离/ 卸载停 goroutine/压力/抛弃换代/熔断/健康任务不受影响）。 --- internal/lua/plugin_runtime.go | 478 ++++++++++++++++++++++------ internal/lua/plugin_runtime_test.go | 188 ++++++++++- 2 files changed, 571 insertions(+), 95 deletions(-) diff --git a/internal/lua/plugin_runtime.go b/internal/lua/plugin_runtime.go index 682a152..6394144 100644 --- a/internal/lua/plugin_runtime.go +++ b/internal/lua/plugin_runtime.go @@ -53,6 +53,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. + // A task that exceeds it is interrupted at the Lua instruction level — the + // only way to stop a runaway script on the C stack. + Timeout int `json:"timeout,omitempty"` } // scheduleSpec is the parsed `PLUGIN.schedule` table. @@ -98,6 +102,7 @@ func parseSchedule(L *golua.State) scheduleSpec { t.Every = readNumberInt(L, -1, "every") t.At = readString(L, -1, "at") t.Jitter = readNumberInt(L, -1, "jitter") + t.Timeout = readNumberInt(L, -1, "timeout") if t.Fn != "" && (t.Every > 0 || t.At != "") { spec.tasks = append(spec.tasks, t) } @@ -116,72 +121,137 @@ func parseSchedule(L *golua.State) scheduleSpec { // VM used only for periodic jobs, whose work is inherently long-lived and // network-bound. type PluginRuntime struct { - plugin string - vm *VM // a private VM running the SAME plugin source - poolName string - name string + plugin string + source string // kept so a discarded generation can be re-booted from scratch - mu sync.Mutex - spec scheduleSpec - nextDue map[string]time.Time + mu sync.Mutex + 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 - stop chan struct{} - done chan struct{} - // failures counts consecutive task failures, for the hook-error surface. + 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. + 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. + 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{} } -// startRuntime boots a private VM for the plugin and starts its scheduler. +// runtimeGen is one disposable execution generation: a private VM plus the +// single-state pool the tasks run on. // -// The private VM runs the same source but under a distinct name, so it does not -// collide with the adapter registry or the shared plugin VM. 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). +// 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. +// +// 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 + +// startRuntime boots the first generation and starts the scheduler. +// +// 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 { return nil } rt := &PluginRuntime{ plugin: name, + source: source, + ps: ps, spec: spec, - nextDue: make(map[string]time.Time, len(spec.tasks)), 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{}), } - // A private VM per plugin: booting one is ~0.1ms (see plugin_shard.go for the - // measurement), which is far cheaper than the isolation it buys. - // - // The private VM is the SAME VM OBJECT with an empty dir — it must not - // re-scan or re-seed the adapter directory, and its pool is private to this - // runtime. Reusing the VM type (rather than inventing a second sandbox) keeps - // the shared globals identical, so a plugin behaves the same whether its task - // runs on the shared state or here. - rt.vm = &VM{ - dir: "", - pools: map[string]*adapterPool{}, - // The janitor is never started for a private runtime VM: it is sized for - // one idle state that we manage directly, and running a reclaimer over a - // pool with a single permanent worker would be pure overhead. - janitorStop: make(chan struct{}), - } - w, err := bootPluginState(source, name, ps.outboundAllowlist()) - if err != nil { - ps.logf("[scheduler] plugin %s: private VM boot failed: %v", name, err) + gen := rt.bootGeneration(ps) + if gen == nil { + ps.logf("[scheduler] plugin %s: no generation could be started", name) return nil } - // File the state as a single-state pool so the existing acquire/release and - // Transform machinery work unchanged. - // - // The Cond MUST be built from this pool's own mu (newAdapterPool does exactly - // that). Handing it a fresh mutex instead makes cond.Wait unlock a mutex it - // never locked, which Go reports as "unlock of unlocked mutex" and takes the - // whole process down — not a recoverable error. + // 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) + } + go rt.loop() + 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_" + name, - script: source, - static: staticInfo{name: name, headers: map[string]string{}}, + name: "__pluginrt_" + rt.plugin, + script: rt.source, + static: staticInfo{name: rt.plugin, headers: map[string]string{}}, pluginMode: true, maxW: 1, revision: 1, @@ -189,13 +259,8 @@ func (ps *Plugins) startRuntime(name, source string, spec scheduleSpec) *PluginR pool.cond = sync.NewCond(&pool.mu) pool.idle = []*worker{w} pool.created = 1 - rt.vm.pools[pool.name] = pool - rt.poolName = pool.name - for _, t := range spec.tasks { - rt.nextDue[t.Fn] = rt.firstDue(t) - } - go rt.loop() - return rt + 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 @@ -209,8 +274,7 @@ func (rt *PluginRuntime) firstDue(t ScheduledTask) time.Time { if d <= 0 { d = time.Hour } - j := time.Duration(t.Jitter) * time.Second - if j > 0 { + if j := time.Duration(t.Jitter) * time.Second; j > 0 { d += time.Duration(randInt63n(int64(j))) } return now.Add(d) @@ -235,7 +299,7 @@ func nextDailyOccurrence(at string, now time.Time) time.Time { // without waiting whole seconds; production always uses 1s. var runtimeTick = time.Second -// loop is the scheduler goroutine: one ticker drives every plugin's tasks. +// loop is the scheduler goroutine: one ticker drives every task of the plugin. func (rt *PluginRuntime) loop() { defer close(rt.done) tick := time.NewTicker(runtimeTick) @@ -243,19 +307,137 @@ 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() } } } -// 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 (and a -// runaway task only re-runs itself after it finishes + its next interval). -func (rt *PluginRuntime) runDue() { - now := time.Now() +// checkWedged abandons a generation 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. +func (rt *PluginRuntime) checkWedged() { rt.mu.Lock() + gen := rt.gen + if gen == nil { + 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 + } + 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 + for _, fn := range wedged { + rt.abandons[fn]++ + rt.failures[fn]++ + n := rt.abandons[fn] + if n >= maxAbandons { + rt.disarmed = true + disarmedNow = true + } + rt.logf("[scheduler] plugin %s task %s exceeded %s; abandoning its VM (abandon #%d/%d)", + rt.plugin, fn, wedgedAfter, n, maxAbandons) + } + 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. +func (rt *PluginRuntime) runDue() { + rt.mu.Lock() + if rt.disarmed || rt.gen == nil { + rt.mu.Unlock() + return + } + gen := rt.gen + now := time.Now() var due []ScheduledTask for _, t := range rt.spec.tasks { if rt.running[t.Fn] { @@ -265,12 +447,16 @@ 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.nextDue[t.Fn] = rt.nextAfter(t, now) due = append(due, t) } rt.mu.Unlock() for _, t := range due { - go rt.runOne(t) + go rt.runOne(gen, t) } } @@ -286,25 +472,71 @@ func (rt *PluginRuntime) nextAfter(t ScheduledTask, now time.Time) time.Time { return now.Add(d) } -// runOne executes one task on the private VM, guarded against a Lua-level -// error. A failure (or a hook that returns an error) is recorded but never -// propagates into request forwarding. -func (rt *PluginRuntime) runOne(t ScheduledTask) { - defer func() { +// runOne executes one task on the given generation. +// +// 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() - }() - _, err := rt.vm.Transform(rt.poolName, t.Fn, "{}") - rt.mu.Lock() - if err != nil { - rt.failures[t.Fn]++ - rt.mu.Unlock() - rt.logf("[scheduler] plugin %s task %s failed: %v", rt.plugin, t.Fn, err) return } - rt.failures[t.Fn] = 0 + 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() + defer func() { + rt.mu.Lock() + delete(gen.busy, t.Fn) + delete(rt.running, t.Fn) + delete(rt.startedAt, t.Fn) + rt.mu.Unlock() + }() + + p := gen.vm.pool(gen.poolName) + if p == nil { + rt.recordFailure(t.Fn, "no pool") + 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 { + rt.recordFailure(t.Fn, err.Error()) + return + } + 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. +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{}) { @@ -313,31 +545,81 @@ func (rt *PluginRuntime) logf(format string, args ...interface{}) { } } -// Stop halts the scheduler and closes the private VM. +// Stop halts the scheduler and abandons the current generation. func (rt *PluginRuntime) Stop() { if rt == nil { return } close(rt.stop) <-rt.done - if rt.vm != nil { - rt.vm.Stop() - } } // 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{})] -// randInt63n returns a pseudo-random value in [0,n). It avoids importing math/rand -// into this file's already-large import set for one call; the quality is -// irrelevant for jitter. +// 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. +// +// 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. +func callFnOn(L *golua.State, fn string) error { + L.SetTop(0) + defer L.SetTop(0) + L.GetGlobal(pluginGlobal) + if L.IsNil(-1) { + return fmt.Errorf("plugin table missing") + } + plug := L.GetTop() + L.GetGlobal(hookGuardName) + if L.IsNil(-1) { + L.Pop(1) + } + L.GetField(plug, fn) + if !L.IsFunction(-1) { + return fmt.Errorf("task %s missing", fn) + } + L.PushString("{}") + if err := L.Call(1, 2); err != nil { + return err + } + if L.GetTop() >= 2 && !L.IsNil(-1) { + msg := L.ToString(-1) + L.Pop(1) + if msg != "" { + return fmt.Errorf("%s", msg) + } + } + 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 { if n <= 0 { return 0 } - // xorshift seeded from wall clock — jitter only needs to differ across - // processes, not be cryptographically strong. seed := time.Now().UnixNano() ^ int64(n*2654435761) x := uint64(seed) x ^= x << 13 @@ -346,16 +628,6 @@ func randInt63n(n int64) int64 { return int64(x % uint64(n)) } -// sortedTaskNames returns the task function names of a spec (for diagnostics). -func (s scheduleSpec) sortedTaskNames() []string { - out := make([]string, 0, len(s.tasks)) - for _, t := range s.tasks { - out = append(out, t.Fn) - } - sort.Strings(out) - return out -} - // readNumberInt reads a numeric field as an int. Missing/zero yields 0, which // every caller treats as "not set" (a task with every<=0 and no at is disabled). func readNumberInt(L *golua.State, idx int, field string) int { @@ -366,3 +638,23 @@ func readNumberInt(L *golua.State, idx int, field string) int { } return int(L.ToNumber(-1)) } + +// sortedTaskNames returns the task function names of a spec (for diagnostics). +func (s scheduleSpec) sortedTaskNames() []string { + out := make([]string, 0, len(s.tasks)) + for _, t := range s.tasks { + out = append(out, t.Fn) + } + sort.Strings(out) + return out +} + +// PluginTasks lists the scheduled task names a plugin declared (for diagnostics +// and for the WebUI to show what periodic work exists). +func (p *Plugin) PluginTasks() []string { + out := make([]string, 0, len(p.Schedule)) + for _, t := range p.Schedule { + out = append(out, t.Fn) + } + return out +} diff --git a/internal/lua/plugin_runtime_test.go b/internal/lua/plugin_runtime_test.go index f608750..4e40d72 100644 --- a/internal/lua/plugin_runtime_test.go +++ b/internal/lua/plugin_runtime_test.go @@ -147,7 +147,13 @@ return plugin // Transform 只返回 Lua **字符串** 返回值；tick 返回的是 table，故 out 为空 // 不代表任务没跑。真正的观测点是 failures：任务体若抛错会被计入。 - if _, err := rt.vm.Transform(rt.poolName, "tick", "{}"); err != nil { + 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() @@ -157,7 +163,7 @@ return plugin t.Errorf("tick 失败 %d 次", fails) } // 用 keep 函数读回计数：任务把 n 累加在 state 里，读它证明任务体确实执行过。 - out, err := rt.vm.Transform(rt.poolName, "keep", "{}") + out, err := gen.vm.Transform(gen.poolName, "keep", "{}") if err != nil { t.Fatalf("keep: %v", err) } @@ -378,3 +384,181 @@ func luaQuote(s string) string { } var _ = os.TempDir + +// TestWedgedTaskAbandonsItsVM 是「抛弃式隔离」的核心测试。 +// +// 前提（实测三次）：本 LuaJIT 构建**无法中断**运行中的 Lua 调用。所以模型 +// 不是「中断它」，而是「不再等它」——把整个 generation（私有 VM + goroutine） +// 丢掉，重新起一个干净的。卡死的 goroutine 连同它自己的 VM 一起被丢弃， +// 永远不会碰到新一代的状态。 +func TestWedgedTaskAbandonsItsVM(t *testing.T) { + shortTick(t) + oldWedged := wedgedAfter + 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 +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("无运行时") + } + + // 手动触发一次任务（它会卡住），然后等 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) + for time.Now().Before(deadline) { + rt.mu.Lock() + abandoned := rt.gen != gen1 + rt.mu.Unlock() + if abandoned { + break + } + 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) +} + +// 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) { + shortTick(t) + _, ps, _ := runtimePlugins(t) + code := ` +local plugin = { name = "healthy", version = "1.0" } +plugin.schedule = { { fn = "q", every = 1 } } +function plugin.q() return nil 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.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) + } +}