Files
ModelRouter/internal/lua/plugin_runtime_test.go
JianFeeeee 4d1407bbb4 feat(lua): 插件长期运行环境(R4-b)—— 内核驱动调度 + 独立 VM
用户需求:插件需要长期运行环境(独立 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 任务压力。
2026-10-11 10:26:39 +08:00

381 lines
11 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

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