From 4d1407bbb482fd2386ca5f9aa090c260990cada0 Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Sun, 11 Oct 2026 10:26:39 +0800 Subject: [PATCH] =?UTF-8?q?feat(lua):=20=E6=8F=92=E4=BB=B6=E9=95=BF?= =?UTF-8?q?=E6=9C=9F=E8=BF=90=E8=A1=8C=E7=8E=AF=E5=A2=83=EF=BC=88R4-b?= =?UTF-8?q?=EF=BC=89=E2=80=94=E2=80=94=20=E5=86=85=E6=A0=B8=E9=A9=B1?= =?UTF-8?q?=E5=8A=A8=E8=B0=83=E5=BA=A6=20+=20=E7=8B=AC=E7=AB=8B=20VM?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 用户需求：插件需要长期运行环境（独立 LuaVM 线程化），以支持自动签到、 token 轮换这类"没有浏览器、没有人操作时也要发生"的能力。 旧模型做不到：插件的周期工作只能靠前端 pluginAPI.call 驱动，页面没开就 一次都不跑。这正是自动签到的死穴。 新增 internal/lua/plugin_runtime.go：插件声明，内核驱动。 PLUGIN.schedule = { { fn = "refresh_token", every = 3600 }, { fn = "daily_checkin", at = "03:30" }, { fn = "poll_quota", every = 300, jitter = 30 }, } - 每个声明了 schedule 的插件获得**私有 VM**（不是共享 plugin.state 的那个： 钩子要累加，任务要长跑且网络阻塞，两者对隔离的要求相反） - 内核单个 ticker 驱动所有任务；每个任务**独立 goroutine**，一个长任务 不推迟另一个的 due - `at="HH:MM"` 每日固定时刻；`jitter` 打散首触发，避免整点齐射 - 任务失败计数进入 hook_errors 面（静默坏任务必须可见） - Unload 先停调度再关状态（顺序反了会让在跑的任务写进已关闭的 Lua 状态） 明确边界：**适配器不参与调度**。适配器仍是无状态请求级池；token 轮换靠 插件（这里）按周期调用 Phase 3 的 UpdateAdapter。 修掉两个把进程直接搞崩的缺陷： 1. 私有 VM 的池我给 sync.Cond 传了新 mutex 而非池自己的 mu → "unlock of unlocked mutex"，fatal error 而非可恢复错误。 2. Transform 硬编码 adapterGlobal，而插件模式的池把表存在 pluginGlobal → 健康的状态也报 "has no table"。加 poolGlobal(p) 按模式选择。 测试 8 个：无任何触发自动执行（核心）/真正执行了（计数读回）/无声明则零开销/ at 解析与畸形退化/不完整条目被忽略/失败隔离/卸载停 goroutine/6 插件×3 任务压力。 --- internal/lua/plugin_runtime.go | 368 +++++++++++++++++++++++++++ internal/lua/plugin_runtime_test.go | 380 ++++++++++++++++++++++++++++ internal/lua/plugins.go | 41 +++ internal/lua/vm.go | 13 +- 4 files changed, 801 insertions(+), 1 deletion(-) create mode 100644 internal/lua/plugin_runtime.go create mode 100644 internal/lua/plugin_runtime_test.go diff --git a/internal/lua/plugin_runtime.go b/internal/lua/plugin_runtime.go new file mode 100644 index 0000000..682a152 --- /dev/null +++ b/internal/lua/plugin_runtime.go @@ -0,0 +1,368 @@ +// Package lua — 插件长期运行环境（R4-b / Phase 4） +// +// 用户需求：插件需要**长期运行环境（独立 LuaVM 线程化）**，以支持自动签到、 +// token 轮换等需要"没有浏览器、没有人操作时也会发生"的能力。 +// +// 为什么需要独立 VM：现有插件模型是「一个插件 = 一个 Lua 状态 + 一把锁」， +// 因为钩子会往 plugin.state 累加，多个状态会让总额被劈成几份（见 docs 的 +// 运行时约束）。这把锁的代价是**一个卡住的钩子会卡住所有插件**——而签到、 +// token 轮换这类任务天然是长耗时的网络等待。 +// +// 本文件给出调度层：内核按插件声明的周期，在**独立 VM + 独立 goroutine** 上 +// 执行它，从而： +// +// - 一个插件的长任务不阻塞其它插件的钩子（独立 VM，无共享锁） +// - 任务在没有浏览器、没有人操作时也会执行（内核 ticker 驱动，不是前端轮询） +// - 状态与主 VM 分离，轮换中崩溃不会带走已累计的钩子统计 +// +// 声明方式（插件侧）： +// +// PLUGIN.schedule = { +// { 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，不影响钩子与请求转发。 +package lua + +import ( + "fmt" + "sort" + "sync" + "sync/atomic" + "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"` +} + +// 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") + 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 long-lived execution environment. +// +// 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. +type PluginRuntime struct { + plugin string + vm *VM // a private VM running the SAME plugin source + poolName string + name string + + mu sync.Mutex + spec scheduleSpec + nextDue map[string]time.Time + running map[string]bool + stop chan struct{} + done chan struct{} + // failures counts consecutive task failures, for the hook-error surface. + failures map[string]int +} + +// startRuntime boots a private VM for the plugin and starts its scheduler. +// +// 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). +func (ps *Plugins) startRuntime(name, source string, spec scheduleSpec) *PluginRuntime { + if len(spec.tasks) == 0 || ps.vm == nil { + return nil + } + rt := &PluginRuntime{ + plugin: name, + spec: spec, + nextDue: make(map[string]time.Time, len(spec.tasks)), + running: make(map[string]bool, len(spec.tasks)), + failures: 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) + 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. + pool := &adapterPool{ + name: "__pluginrt_" + name, + script: source, + static: staticInfo{name: name, headers: map[string]string{}}, + pluginMode: true, + maxW: 1, + revision: 1, + } + 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 +} + +// 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 + } + j := time.Duration(t.Jitter) * time.Second + if 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 plugin's tasks. +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.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() + rt.mu.Lock() + 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.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 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() { + rt.mu.Lock() + delete(rt.running, 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.Unlock() +} + +func (rt *PluginRuntime) logf(format string, args ...interface{}) { + if f := activeLogger.Load(); f != nil { + (*f)(format, args...) + } +} + +// Stop halts the scheduler and closes the private VM. +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. +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 + x ^= x >> 7 + x ^= x << 17 + 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 { + L.GetField(idx, field) + defer L.Pop(1) + if L.Type(-1) != golua.LUA_TNUMBER { + return 0 + } + return int(L.ToNumber(-1)) +} diff --git a/internal/lua/plugin_runtime_test.go b/internal/lua/plugin_runtime_test.go new file mode 100644 index 0000000..f608750 --- /dev/null +++ b/internal/lua/plugin_runtime_test.go @@ -0,0 +1,380 @@ +package lua + +import ( + "fmt" + "os" + "path/filepath" + "sync/atomic" + "testing" + "time" +) + +// 本文件验证 R4-b 的核心主张：**插件声明的周期任务在没有任何浏览器、 +// 没有任何外部触发时会自己执行**。 +// +// 这正是「自动签到」这类能力的死穴：旧模型下插件的周期工作只能靠前端 +// pluginAPI.call 驱动，页面没开就一次都不跑。这里用「内核 ticker + 独立 VM」 +// 替换那个机制。 + +func shortTick(t *testing.T) { + t.Helper() + old := runtimeTick + runtimeTick = 10 * time.Millisecond + t.Cleanup(func() { runtimeTick = old }) +} + +// runtimePlugins 建一个带插件目录的 Plugins，并把 outbound allowlist 设好， +// 让插件的 HTTP 调用在白名单内可通。 +func runtimePlugins(t *testing.T) (*VM, *Plugins, string) { + t.Helper() + dir := filepath.Join(t.TempDir(), "adapters") + vm := NewVM(dir) + if err := vm.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(vm.Stop) + pdir := filepath.Join(t.TempDir(), "plugins") + ps := NewPlugins(vm, pdir) + ps.SetOutboundHosts([]string{"127.0.0.1"}) + return vm, ps, pdir +} + +// 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.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 +end +return plugin +` + if err := ps.LoadSource("checkin", code); err != nil { + t.Fatalf("load: %v", err) + } + // 等到运行时就位 + 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 + } + } + 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) + } +} + +// TestScheduledTaskActuallyRuns 用独立的可观测面（task 计数器经 state 暴露） +// 证明任务体真的被执行了，而不是仅仅"没有失败"。 +func TestScheduledTaskActuallyRuns(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 +end +-- keep 返回字符串，便于测试从 Transform 的字符串返回值里读回计数 +function plugin.keep() + return tostring((plugin.state and plugin.state.n) or 0) +end +return plugin +` + if err := ps.LoadSource("counter", code); err != nil { + t.Fatalf("load: %v", err) + } + time.Sleep(1400 * time.Millisecond) + + // 直接在运行时的私有 VM 上调用一次 tick 读回计数 + // （独立 VM 的状态不在共享 plugin.state 里，这是设计使然）。 + ps.mu.RLock() + var rt *PluginRuntime + for _, p := range ps.plugins { + if p.Info.Name == "counter" { + rt = p.Runtime + } + } + ps.mu.RUnlock() + if rt == nil { + t.Fatal("无运行时") + } + + // Transform 只返回 Lua **字符串** 返回值；tick 返回的是 table，故 out 为空 + // 不代表任务没跑。真正的观测点是 failures：任务体若抛错会被计入。 + if _, err := rt.vm.Transform(rt.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 := rt.vm.Transform(rt.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) +} + +// 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 { + t.Fatal(err) + } + 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 + } + } + 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) { + _, ps, _ := runtimePlugins(t) + code := ` +local plugin = { name = "partial", version = "1.0" } +plugin.schedule = { + { fn = "good", every = 60 }, + { every = 60 }, -- 缺 fn + { fn = "noPeriod" }, -- 缺 every 且无 at + { fn = "alsoGood", at = "03:00" }, +} +return plugin +` + if err := ps.LoadSource("partial", code); err != nil { + t.Fatal(err) + } + ps.mu.RLock() + defer ps.mu.RUnlock() + 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 + } + } + t.Fatal("插件未加载") +} + +// 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 +return plugin +` + if err := ps.LoadSource("mixed", 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 + } + } + 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"]) + } +} + +// TestUnloadStopsRuntime：卸载插件必须停掉它的运行时（不泄漏 goroutine/VM）。 +func TestUnloadStopsRuntime(t *testing.T) { + _, ps, _ := runtimePlugins(t) + code := ` +local plugin = { name = "temp", version = "1.0" } +plugin.schedule = { { fn = "t", every = 3600 } } +function plugin.t() return nil end +return plugin +` + if err := ps.LoadSource("temp", 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 + } + } + ps.mu.RUnlock() + if rt == nil { + t.Fatal("无运行时") + } + if err := ps.Unload("temp"); err != nil { + t.Fatalf("unload: %v", err) + } + // done 应已关闭 + select { + case <-rt.done: + case <-time.After(2 * time.Second): + t.Error("卸载后调度 goroutine 未退出（泄漏）") + } +} + +// 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 diff --git a/internal/lua/plugins.go b/internal/lua/plugins.go index 2758674..6d5e9c6 100644 --- a/internal/lua/plugins.go +++ b/internal/lua/plugins.go @@ -243,6 +243,16 @@ type Plugin struct { // is still the compiled-in default would immediately overwrite the file it // was supposed to inherit from. persisted bool + + // Schedule is the periodic work this plugin declared, and Runtime is the + // long-lived environment that runs it (private VM + kernel ticker). + // + // 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. + Schedule []ScheduledTask + Runtime *PluginRuntime } // Plugins is the loaded plugin set, owned by the VM. @@ -696,6 +706,30 @@ 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 + // 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 + // looking at the dashboard" case the feature exists for. + // + // The parse happens here (right after boot) so a malformed schedule is + // visible at load time rather than silently ignored. + 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}) + if p.Runtime == nil { + // The private VM failed to boot. 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", + p.Info.Name, len(p.Schedule)) + } + } + // ---- UI ---- if ui, ok := readPluginUI(w.L, p.Info.Name); ok { p.UI = ui @@ -1758,6 +1792,13 @@ 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. + if p.Runtime != nil { + p.Runtime.Stop() + p.Runtime = nil + } p.mu.Lock() if p.state != nil && p.state.L != nil { p.state.L.Close() diff --git a/internal/lua/vm.go b/internal/lua/vm.go index 5cdd8ab..f3b259b 100644 --- a/internal/lua/vm.go +++ b/internal/lua/vm.go @@ -889,6 +889,17 @@ func (v *VM) RemoveAdapter(name string) { } } +// poolGlobal returns the Lua global holding a pool's returned table: plugins +// store theirs under one global, adapters under another. Transform and friends +// must read the SAME one boot() wrote, or a plugin-mode pool answers +// "has no table" even though the state is perfectly healthy. +func poolGlobal(p *adapterPool) string { + if p != nil && p.pluginMode { + return pluginGlobal + } + return adapterGlobal +} + func (v *VM) pool(name string) *adapterPool { v.mu.RLock() defer v.mu.RUnlock() @@ -966,7 +977,7 @@ func (v *VM) Transform(name, fn, raw string) (string, error) { L.SetTop(0) defer L.SetTop(0) - L.GetGlobal(adapterGlobal) + L.GetGlobal(poolGlobal(p)) if L.IsNil(-1) { return "", fmt.Errorf("adapter %s has no table", name) }