ModelRouter/internal/lua/plugin_runtime_test.go

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)
}