// Package lua — 插件长期运行环境(R4-b / Phase 4) // // 用户需求:插件需要**长期运行环境**,以支持自动签到、token 轮换等需要 // "没有浏览器、没有人操作时也会发生"的能力。 // // 模型(重要):**一个插件 = 一个 Lua VM = 一个调度器**。 // // 钩子(请求路径)和定时任务(内核 ticker 驱动)跑在**同一个** Lua 状态上, // 由同一把锁串行化。这不是实现细节,而是正确性要求: // // 定时任务 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 }, -- 带抖动 // } // // 崩溃隔离:任务函数出错只记 hookErr,不影响钩子与请求转发。但一个**永不返回** // 的任务会永久持有该插件的锁(Lua 调用无法中断,见 checkWedged 的能力说明), // 因此超时后该插件被**停用**而不是"换一个 VM 重来"—— 换 VM 会把累计状态 // (计费总额、已签到的天数)一起丢掉,那比停用更糟。 package lua import ( "errors" "fmt" "log" "sync" "time" golua "github.com/aarzilli/golua/lua" ) // ScheduledTask is one declared periodic job. type ScheduledTask struct { // Fn is the plugin function name to invoke. Fn string // Every is the interval in seconds. 0 disables the task. Every int `json:"every"` // At is an optional daily local time "HH:MM"; when set, Every is ignored // and the task fires once per day at that wall-clock time. At string `json:"at,omitempty"` // 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 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"` } // scheduleSpec is the parsed `PLUGIN.schedule` table. type scheduleSpec struct { tasks []ScheduledTask } // parseSchedule extracts the schedule from a loaded plugin table. It is read // from the already-compiled state's global, not by re-running the script, so a // plugin's schedule is available the moment it loads. func parseSchedule(L *golua.State) scheduleSpec { var spec scheduleSpec if L == nil { return spec } L.GetGlobal(pluginGlobal) if L.IsNil(-1) { L.Pop(1) return spec } if !getFieldBatch(L, "schedule") { L.Pop(1) return spec } // schedule is an array of tables if L.Type(-1) != golua.LUA_TTABLE { L.Pop(2) return spec } arr := L.GetTop() idx := 1 for { // golua exposes indexed table access through RawGeti; GetField only // takes a string key. L.RawGeti(arr, idx) if L.IsNil(-1) { L.Pop(1) break } if L.Type(-1) == golua.LUA_TTABLE { var t ScheduledTask t.Fn = readString(L, -1, "fn") 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) } } L.Pop(1) idx++ } L.Pop(2) // schedule table + plugin global return spec } // PluginRuntime owns one plugin's periodic work. // // 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 *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 mu sync.Mutex // nextDue is when each task should next fire. nextDue map[string]time.Time // 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 // 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 stop chan struct{} done chan struct{} } // 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. var wedgedAfter = 10 * time.Minute // 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 attaches a scheduler to a plugin that declared one. // // 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: 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{}), } for _, t := range spec.tasks { rt.nextDue[t.Fn] = rt.firstDue(t) } go rt.loop() return rt } // 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 { now := time.Now() if t.At != "" { return nextDailyOccurrence(t.At, now) } d := time.Duration(t.Every) * time.Second if d <= 0 { d = time.Hour } if j := time.Duration(t.Jitter) * time.Second; j > 0 { d += time.Duration(randInt63n(int64(j))) } return now.Add(d) } // nextDailyOccurrence returns the next time "HH:MM" passes after now (today if // it has not yet passed, else tomorrow). func nextDailyOccurrence(at string, now time.Time) time.Time { var hh, mm int if _, err := fmt.Sscanf(at, "%d:%d", &hh, &mm); err != nil || hh < 0 || hh > 23 || mm < 0 || mm > 59 { // A malformed time must not wedge the task forever; fall back to hourly. return now.Add(time.Hour) } t := time.Date(now.Year(), now.Month(), now.Day(), hh, mm, 0, 0, now.Location()) if !t.After(now) { t = t.Add(24 * time.Hour) } return t } // runtimeTick is the scheduler's resolution. A var so tests can shorten it // without waiting whole seconds; production always uses 1s. var runtimeTick = time.Second // 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) defer tick.Stop() for { select { case <-rt.stop: return case <-tick.C: rt.checkWedged() rt.runDue() } } } // checkWedged disarms a plugin whose task has overrun wedgedAfter. // // 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() if rt.disarmed { rt.mu.Unlock() return } var wedged []string for fn, at := range rt.startedAt { if time.Since(at) >= wedgedAfter { wedged = append(wedged, fn) } } if len(wedged) == 0 { rt.mu.Unlock() return } rt.disarmed = true rt.mu.Unlock() for _, fn := range wedged { ps := rt.ps if ps != nil { ps.hookErr.note(StageRequestEnd, fmt.Sprintf("%s: task %s overran %s; plugin disabled", rt.pluginInfoName(), fn, wedgedAfter)) } log.Printf("[scheduler] plugin %s task %s exceeded %s; disabling the plugin "+ "(reload it to retry)", rt.pluginInfoName(), fn, wedgedAfter) } } // 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.mu.Unlock() return } now := time.Now() var due []ScheduledTask for _, t := range rt.spec.tasks { if rt.running[t.Fn] { continue // still executing; skip this cycle } if now.Before(rt.nextDue[t.Fn]) { continue } rt.running[t.Fn] = true rt.active++ rt.nextDue[t.Fn] = rt.nextAfter(t, now) due = append(due, t) } rt.mu.Unlock() for _, t := range due { go rt.runOne(t) } } // nextAfter computes the following due time for a task that just fired. func (rt *PluginRuntime) nextAfter(t ScheduledTask, now time.Time) time.Time { if t.At != "" { return nextDailyOccurrence(t.At, now) } d := time.Duration(t.Every) * time.Second if d <= 0 { d = time.Hour } return now.Add(d) } // runOne executes one task on the plugin's own VM, under the plugin's own lock. // // 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(rt.running, t.Fn) delete(rt.startedAt, t.Fn) rt.active-- rt.mu.Unlock() }() if p == nil { rt.recordFailure(t.Fn, "plugin is gone") return } 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.mu.Unlock() } // 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() 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) } 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 } close(rt.stop) <-rt.done } // callFnOn invokes a plugin function on an already-locked state. // // 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) 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 } // 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 } seed := time.Now().UnixNano() ^ int64(n*2654435761) x := uint64(seed) x ^= x << 13 x ^= x >> 7 x ^= x << 17 return int64(x % uint64(n)) } // 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 { L.GetField(idx, field) defer L.Pop(1) if L.Type(-1) != golua.LUA_TNUMBER { return 0 } return int(L.ToNumber(-1)) }