package lua import ( "fmt" "os" "testing" "time" ) // 本文件验证 R4-b 的核心主张:**插件声明的周期任务在没有任何浏览器、 // 没有任何外部触发时会自己执行**,并且跑在**插件自己的那个 VM** 上。 // // 第二点是这个文件最重要的断言。早期实现给定时任务开了一个私有 VM,结果 // 任务写 plugin.state 时写进了一个请求路径永远读不到的副本 —— token 轮换 // 看起来成功了,对实际流量完全无效。TestTaskWriteIsVisibleToHooks 锁死这一点。 func shortTick(t *testing.T) { t.Helper() old := runtimeTick runtimeTick = 10 * time.Millisecond t.Cleanup(func() { runtimeTick = old }) } // runtimePlugins 建一个带插件目录的 Plugins。 func runtimePlugins(t *testing.T) (*VM, *Plugins, string) { t.Helper() dir := fmt.Sprintf("%s/adapters", t.TempDir()) vm := NewVM(dir) if err := vm.Start(); err != nil { t.Fatal(err) } t.Cleanup(vm.Stop) pdir := fmt.Sprintf("%s/plugins", t.TempDir()) // ⚠️ **必须**设 LLMSPROXY_PLUGIN_DIR。channels 插件用 storeDir() 决定自己的 // 存储位置,而 storeDir 在环境变量缺失时会去读 /etc/llmsproxy/config.yaml —— // 那是**生产路径**。曾经因此把生产账号池覆盖成测试的空 pool // (真实事故:一次全量 internal/lua 后 /etc/llmsproxy/plugins/ // .channels-store.json 里 13 个账号变成 0 个,凭据却还留在保险库)。 if os.Getenv("LLMSPROXY_PLUGIN_DIR") == "" { t.Setenv("LLMSPROXY_PLUGIN_DIR", pdir) } ps := NewPlugins(vm, pdir) ps.SetOutboundHosts([]string{"127.0.0.1"}) // ⚠️ 关掉状态持久化。插件声明了 schedule 时内核会自动起调度器 // (plugins.go: p.Runtime = ps.startRuntime(...)),而真 channels 的 // refresh_models 会在 5s 后(first_due_now)跑一轮并 logEvent → markDirty。 // 那是**后台 goroutine**,比 t.Setenv 的生命周期长:测试结束后它仍可能落盘, // 此时环境变量已还原,storeDir() 就兜底到生产路径。 // 实测:只跑 internal/lua 全会话,生产 store 的 log 会多出调度器条目。 ps.DisableStatePersistence() 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 是主断言。 // // 插件声明一个极短周期的任务,测试**不做任何外部调用**,只等 —— 任务必须自己跑。 func TestScheduledTaskFiresWithoutAnyTrigger(t *testing.T) { shortTick(t) _, ps, _ := runtimePlugins(t) code := ` local plugin = { name = "checkin", version = "1.0" } plugin.state = { fired = 0 } plugin.schedule = { { fn = "do_checkin", every = 1 } } function plugin.do_checkin() 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) for time.Now().Before(deadline) { 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 // 通过:没有任何外部触发,任务自己跑了 } } time.Sleep(20 * time.Millisecond) } t.Fatal("任务在没有任何外部触发的情况下没有执行") } // TestTaskWriteIsVisibleToHooks 是这个模型的关键正确性断言。 // // 定时任务写 plugin.state.token,请求路径(同一个 VM 上的钩子)必须能读到 // 它。早期"每个任务一个私有 VM"的实现会在这里失败 —— 任务写进副本,钩子 // 看到的还是旧值,token 轮换静默失效。 func TestTaskWriteIsVisibleToHooks(t *testing.T) { shortTick(t) _, ps, _ := runtimePlugins(t) code := ` 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 plugin.hooks = { routed = "read_token" } function plugin.read_token(payload) payload.token = plugin.state.token -- 请求路径读到的值 return payload end return plugin ` if err := ps.LoadSource("rotator", code); err != nil { t.Fatalf("load: %v", err) } runtimeOf(t, ps, "rotator", 3*time.Second) // 等任务把 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) } out := ps.Fire(StageRouted, map[string]interface{}{}) t.Fatalf("定时任务写的 token 对请求钩子不可见(钩子读到 %v)—— 说明任务跑在了另一个 VM 上", out["token"]) } // TestNoScheduleMeansNoRuntime:没声明 schedule 就不该有额外 VM/goroutine。 func TestNoScheduleMeansNoRuntime(t *testing.T) { _, ps, _ := runtimePlugins(t) code := ` local plugin = { name = "plain", version = "1.0" } 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" && p.Runtime != nil { t.Error("未声明 schedule 的插件不应有运行时") } } } // TestScheduleParsing:every / at / jitter / timeout 解析。 func TestScheduleParsing(t *testing.T) { _, ps, _ := runtimePlugins(t) code := ` local plugin = { name = "sched", version = "1.0" } plugin.schedule = { { 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("sched", code); err != nil { t.Fatal(err) } ps.mu.RLock() var sched []ScheduledTask for _, p := range ps.plugins { if p.Info.Name == "sched" { sched = p.Schedule } } 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) } } // 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 = "flaky", version = "1.0" } plugin.schedule = { { fn = "boom", every = 1 } } function plugin.boom() error("任务故意失败") end return plugin ` if err := ps.LoadSource("flaky", code); err != nil { t.Fatal(err) } 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) } t.Fatal("任务错误未被记账") } // TestUnloadStopsRuntime:卸载后调度器 goroutine 必须停止。 func TestUnloadStopsRuntime(t *testing.T) { shortTick(t) _, ps, _ := runtimePlugins(t) code := ` 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("stoppable", code); err != nil { t.Fatal(err) } rt := runtimeOf(t, ps, "stoppable", 3*time.Second) if err := ps.Unload("stoppable"); err != nil { t.Fatal(err) } // Stop 已返回 ⇒ done 已关闭 ⇒ loop 已退出 select { case <-rt.done: default: t.Error("卸载后调度器 loop 未退出") } } // TestWedgedTaskDisarmsPlugin:任务超时后插件被停用(不重建 VM —— 重建会丢掉 // 累计状态,那正是这个插件存在的意义)。 func TestWedgedTaskDisarmsPlugin(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) } rt := runtimeOf(t, ps, "wedger", 3*time.Second) // 手动跑一次会卡死的任务(它会拿到 p.mu 并永不返回) go rt.runOne(ScheduledTask{Fn: "spin", Every: 3600}) // 等 checkWedged 判定并停用 deadline := time.Now().Add(3 * time.Second) for time.Now().Before(deadline) { 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) } t.Fatal("卡死的任务未导致插件停用") } // 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() plugin.state.n = plugin.state.n + 1 end return plugin ` if err := ps.LoadSource("healthy", code); err != nil { t.Fatal(err) } rt := runtimeOf(t, ps, "healthy", 3*time.Second) time.Sleep(600 * time.Millisecond) if rt.Disarmed() { t.Error("健康任务不应导致插件停用") } rt.mu.Lock() fails := rt.failures["q"] rt.mu.Unlock() if fails != 0 { t.Errorf("健康任务不应失败: %d", fails) } } // 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) }