From 69da3c9b92b0975ffe98cc0bd8e6142c2485404e Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Thu, 15 Oct 2026 22:43:08 +0800 Subject: [PATCH] =?UTF-8?q?fix(kernel):=20Close()=20=E4=BB=8E=E4=B8=8D?= =?UTF-8?q?=E8=B0=83=20on=5Funload=20=E2=80=94=E2=80=94=20=E5=AD=A4?= =?UTF-8?q?=E5=84=BF=E8=BF=9B=E7=A8=8B=E4=B8=8E=E5=9B=9E=E8=B0=83=E6=96=87?= =?UTF-8?q?=E4=BB=B6=E6=B3=84=E6=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 症状 本机积累 **995 个 llmsproxy-loopback.py 进程**（828 个超 24 小时），全是 /tmp/Test*/ 与 .build-work/Test*/ 下的测试残留；生产 /etc/llmsproxy/plugins/ 另有 **18 个孤儿 callback-*.port**（对应 .json 早已消失），且永远清不掉。 ## 真因（两层） 1. **回环回收本是惰性的**：reapLoopback 只在**会话过期**时被 sweepSessions 触发。测试跑完直接退出 ⇒ 会话从未「过期」⇒ 进程常驻。生产会话真会过期， 所以生产不漏进程 —— 但见第 2 条。 2. **★ 根因在内核**：Plugins.Close() 只关 shard 与 saver，**从不调用 on_unload** —— 而 on_unload 只在**单插件 Unload** 时调。于是「进程关闭/ 测试结束」这条路径上，起了 OS 进程的插件永远收不了尾。插件的 plugin.on_unload（回收浏览器栈 + 回环）写得完全正确，只是从不被调用。 这是「两处只修一处」缺陷的又一例：Unload 有、Close 没有。 ## 修法 - 抽 invokeOnUnload(p, name) 助手，**Unload 与 Close 共用同一份实现** （两处各写一遍必然漂移，本次缺陷形状正是如此）；Close() 新增 callOnUnloadAll()。 - ★ 顺序：shards.Shutdown() → save.Close()（落盘）→ callOnUnloadAll()。 落盘必须在前 —— saver 的最终写入要读插件的 Lua 表，on_unload 之后 state 已关，读到的就是空，最后一个区间的累积静默丢失（TestCloseFlushesTail 正是 为守住这一点而存在）。我第一版把顺序写反，被该测试当场咬住。 - channels：启动清扫上一进程留下的孤儿 callback-*.port/.json。loopbacks 追踪表只活在内存，进程被强杀时 on_unload 没机会跑，重启后内存表为空 ⇒ reapLoopback 无从知晓 ⇒ 那 18 个文件永久残留。 - channels：补 reap_loopbacks 端点（生产运维也可用：不必等会话过期）。 - channels：__set_loopback_probe 从**覆盖**改为**合并**条目 —— 覆盖会丢掉 真实会话的 pid/port_file，让 reapLoopback 杀不掉接收器，**探针自己成为 泄漏源**（实测：其他测试已归零，只有它留 1 个孤儿）。 ## 判据（均做变异验证） - internal/lua/plugin_unload_on_close_test.go TestPluginsCloseCallsOnUnload 看真实外部效应（插件在 on_unload 写磁盘 marker），而非「钩子被调用过」这类内部信号 TestPluginsCloseSurvivesFailingOnUnload best-effort 契约 - internal/core/channels_orphan_callback_test.go TestPluginLoadSweepsOrphanCallbackFiles 看目录真实状态：孤儿必须消失， 非回调文件（.channels-store.json）不得误删 - 变异均咬住：注释掉 callOnUnloadAll → 报红；清扫改 no-op → 报红。 ## 实测 修复前跑一次起会话的测试批次留 7×N 个残留进程 → 修复后 **0 个**；全量回归 （go test -tags luajit ./... 10 包）后回环/VNC 残留均为 0。 ## 过程中自查出的两处疏漏（已修） 1. 判据顺序又写反一次（同上 TestCloseFlushesTail 咬住）。 2. **两枚变异残渣留在工作区**未还原：internal/gateway/discover.go 里 `add(raw, "/models", 0)` 被删、`classifyAuthStatus` 的 `|| len(body) >= 0` 恒真注入。已 git checkout 还原。教训：变异测试必须逐次收尾，别靠记忆。 --- .../core/channels_orphan_callback_test.go | 82 ++++++++++++++++++ internal/lua/plugin_unload_on_close_test.go | 77 +++++++++++++++++ internal/lua/plugins.go | 84 +++++++++++++++---- internal/lua/plugins/channels.lua | 49 ++++++++++- 4 files changed, 271 insertions(+), 21 deletions(-) create mode 100644 internal/core/channels_orphan_callback_test.go create mode 100644 internal/lua/plugin_unload_on_close_test.go diff --git a/internal/core/channels_orphan_callback_test.go b/internal/core/channels_orphan_callback_test.go new file mode 100644 index 0000000..f1f896b --- /dev/null +++ b/internal/core/channels_orphan_callback_test.go @@ -0,0 +1,82 @@ +package core + +import ( + "os" + "path/filepath" + "testing" + + "llmsproxy/internal/lua" +) + +// ★ 插件加载时必须清掉上一进程留下的孤儿回调文件。 +// +// 真实事故：生产 `/etc/llmsproxy/plugins/` 攒了 **18 个 callback-*.port** +// （对应的 .json 早已消失）。它们永远不会被清，因为： +// +// · loopbacks 追踪表只活在内存（plugin.state.loopbacks）； +// · 进程被强杀时 on_unload 没机会跑； +// · 重启后内存表是空的 ⇒ reapLoopback 根本不知道它们存在。 +// +// 判据看的是**目录的真实状态**（孤儿必须消失、活跃会话的文件不能被误删）， +// 而不是"某函数被调用过"。 +func TestPluginLoadSweepsOrphanCallbackFiles(t *testing.T) { + dir := t.TempDir() + pluginDir := filepath.Join(dir, "plugins") + if err := os.MkdirAll(pluginDir, 0755); err != nil { + t.Fatal(err) + } + t.Setenv("LLMSPROXY_PLUGIN_DIR", pluginDir) + + // 摆上一轮的残留（模拟上一进程被强杀后的现场） + orphans := []string{ + filepath.Join(pluginDir, "callback-deadbeef0001.port"), + filepath.Join(pluginDir, "callback-deadbeef0001.json"), + filepath.Join(pluginDir, "callback-deadbeef0002.port"), + } + for _, f := range orphans { + if err := os.WriteFile(f, []byte("12345"), 0600); err != nil { + t.Fatal(err) + } + } + // 一份**不属于**回调的文件：必须不被误删（否则清扫过头会毁掉账号池） + keep := filepath.Join(pluginDir, ".channels-store.json") + if err := os.WriteFile(keep, []byte(`{"accounts":{}}`), 0600); err != nil { + t.Fatal(err) + } + + // 加载插件 = 走真实启动路径 + c := kernelTestCore(t) + defer c.Close() + src, err := readBundledChannels(t) + if err != nil { + t.Fatal(err) + } + ps := c.Plugins() + if err := ps.LoadSource("channels", src); err != nil { + t.Fatal(err) + } + // 顶层清扫代码在加载期就跑了；给 popen 的 rm 一点时间落盘 + for i := 0; i < 50; i++ { + left := 0 + for _, f := range orphans { + if _, err := os.Stat(f); err == nil { + left++ + } + } + if left == 0 { + break + } + } + for _, f := range orphans { + if _, err := os.Stat(f); err == nil { + t.Fatalf("孤儿回调文件未被清扫: %s —— 它会永久留在磁盘上"+ + "（真实事故：生产攒了 18 个）", filepath.Base(f)) + } + } + if _, err := os.Stat(keep); err != nil { + t.Fatalf("清扫过头了：%s 被误删（账号池/索引绝不能碰）", filepath.Base(keep)) + } + t.Log("✓ 孤儿回调文件被清，非回调文件保留") +} + +var _ = lua.Plugins{} diff --git a/internal/lua/plugin_unload_on_close_test.go b/internal/lua/plugin_unload_on_close_test.go new file mode 100644 index 0000000..14fa4cb --- /dev/null +++ b/internal/lua/plugin_unload_on_close_test.go @@ -0,0 +1,77 @@ +package lua + +import ( + "os" + "path/filepath" + "testing" +) + +// ★ 关闭 Plugins 时必须给插件一次 on_unload 机会。 +// +// 真实缺陷：Close() 此前只关 shard 与 saver，**从不调用 on_unload** —— 而 +// on_unload 只在单插件 Unload 时调。于是「进程关闭/测试结束」这条路径上， +// 起了 OS 进程的插件（回环回调接收器、浏览器栈）永远收不了尾，留下孤儿进程。 +// 实测本机积累到 995 个 python 监听器（828 个超 24 小时），全是测试残留。 +// +// 这个判据看的是**真实外部效应**（插件在 on_unload 里改磁盘上的文件）， +// 而不是「钩子被调用过」这类内部信号 —— 后者在钩子写错名字时也会假绿。 +func TestPluginsCloseCallsOnUnload(t *testing.T) { + dir := t.TempDir() + marker := filepath.Join(dir, "unloaded.marker") + + pluginDir := filepath.Join(dir, "plugins") + if err := os.MkdirAll(pluginDir, 0755); err != nil { + t.Fatal(err) + } + src := ` +local plugin = { name = "leaky", version = "1.0.0", exports = {} } +plugin.ui = nil +function plugin.on_unload() + local f = io.open("` + marker + `", "w") + if f then f:write("reaped"); f:close() end +end +return plugin +` + if err := os.WriteFile(filepath.Join(pluginDir, "leaky.lua"), []byte(src), 0644); err != nil { + t.Fatal(err) + } + + ps := NewPlugins(nil, pluginDir) + if err := ps.LoadDir(); err != nil { + t.Fatalf("LoadDir: %v", err) + } + // 关之前不该有标记 + if _, err := os.Stat(marker); err == nil { + t.Fatal("on_unload 在 Close 之前就被调了？") + } + ps.Close() + + // ★ 真判据：文件必须被写出来（= on_unload 真的跑过） + if _, err := os.Stat(marker); err != nil { + t.Fatalf("Close() 没有调用 on_unload —— 起了 OS 进程的插件会留下孤儿进程"+ + "（真实缺陷：995 个 python 监听器残留）。marker 不存在: %v", err) + } + t.Log("✓ Close() 触发了 on_unload（孤儿进程不会再堆积）") +} + +// on_unload 抛错不得让 Close 崩掉（best-effort 契约）。 +func TestPluginsCloseSurvivesFailingOnUnload(t *testing.T) { + dir := t.TempDir() + pluginDir := filepath.Join(dir, "plugins") + if err := os.MkdirAll(pluginDir, 0755); err != nil { + t.Fatal(err) + } + src := ` +local plugin = { name = "boom", version = "1.0.0", exports = {} } +function plugin.on_unload() error("故意炸") end +return plugin +` + if err := os.WriteFile(filepath.Join(pluginDir, "boom.lua"), []byte(src), 0644); err != nil { + t.Fatal(err) + } + ps := NewPlugins(nil, pluginDir) + if err := ps.LoadDir(); err != nil { + t.Fatalf("LoadDir: %v", err) + } + ps.Close() // 不得 panic +} diff --git a/internal/lua/plugins.go b/internal/lua/plugins.go index a987f74..5c9d5d4 100644 --- a/internal/lua/plugins.go +++ b/internal/lua/plugins.go @@ -572,10 +572,72 @@ func (ps *Plugins) Close() { if ps == nil { return } + // ★ 顺序至关重要：**先落盘，再给插件收尾**。 + // + // saver 的最终写入要读每个插件的 Lua 表；on_unload 之后 state 已关， + // 读到的就是空 —— 最后一个区间的累积静默丢失（persist_test 的 + // TestCloseFlushesTail 正是为守住这一点而存在）。 + // + // 所以不能把 callOnUnloadAll 挪到前面，也不能把 save.Close 挪到后面。 // Shard VMs are separate golua states; leaving them open leaks a VM per // shard on every reload. ps.shards.Shutdown() ps.save.Close() + ps.callOnUnloadAll() +} + +// callOnUnloadAll 依次给每个插件调 on_unload（best-effort）。 +// +// 与 Unload 里的调用共用同一实现：两处若各写一遍，将来必然漂移成一个修了 +// 一个没修 —— 这正是本次缺陷的形状（Unload 有、Close 没有）。 +func (ps *Plugins) callOnUnloadAll() { + ps.mu.Lock() + plugins := make([]*Plugin, len(ps.plugins)) + copy(plugins, ps.plugins) + ps.mu.Unlock() + for _, p := range plugins { + ps.callOnUnload(p) + } +} + +// callOnUnload 调单个插件的 on_unload（存在才调），并关掉它的 Lua state。 +// +// 为什么关 state 也在这里：on_unload 之后 state 就没用了，而 state 一旦关掉， +// 钩子再也调不了 —— 调用与关闭必须紧邻，否则又会漏。Unload 与 Close 各自持有 +// 的 state 生命周期不同（Unload 还要 rebuild，Close 之后整体销毁），所以这里 +// 只做「调用 + 置空」，由调用方决定其余步骤。 +func (ps *Plugins) callOnUnload(p *Plugin) { + if p == nil { + return + } + if err := ps.invokeOnUnload(p, p.Info.Name); err != nil { + log.Printf("[plugins] %s: on_unload: %v", p.Info.Name, err) + } +} + +// invokeOnUnload 真正调钩子。抽出来是为了让 Unload 与 Close 共用一份实现 +// （含 state 判空与错误返回语义），避免两处契约不一致。 +func (ps *Plugins) invokeOnUnload(p *Plugin, name string) error { + p.mu.Lock() + defer p.mu.Unlock() + if p.state == nil || p.state.L == nil { + return nil + } + L := p.state.L + L.GetGlobal(pluginGlobal) + if L.Type(-1) != golua.LUA_TTABLE { + L.Pop(1) + return nil + } + L.GetField(-1, "on_unload") + var err error + if L.Type(-1) == golua.LUA_TFUNCTION { + err = callFnOn(L, "on_unload") + } + L.Pop(2) // plugin table + field remnant + p.state.L.Close() + p.state = nil + return err } // LoadDir loads every .lua file in dir as a plugin. Files are loaded in @@ -1941,24 +2003,10 @@ func (ps *Plugins) Unload(name string) error { // only guaranteed point where it still can. Best-effort by design — a // hook that errors must not block an unload the operator explicitly asked // for (a wedged plugin has its own path above). - p.mu.Lock() - if p.state != nil && p.state.L != nil { - p.state.L.GetGlobal(pluginGlobal) - if L := p.state.L; L.Type(-1) == golua.LUA_TTABLE { - L.GetField(-1, "on_unload") - if L.Type(-1) == golua.LUA_TFUNCTION { - if err := callFnOn(L, "on_unload"); err != nil { - log.Printf("[plugins] %s: on_unload: %v", name, err) - } - } - L.Pop(2) // adapter/plugin table + field remnant - } else { - L.Pop(1) - } - p.state.L.Close() - p.state = nil - } - p.mu.Unlock() + // + // 与 Close 共用 invokeOnUnload：两处各写一遍必然漂移 —— 本次缺陷正是 + // 「Unload 有、Close 没有」。 + _ = ps.invokeOnUnload(p, name) ps.rebuild() return nil } diff --git a/internal/lua/plugins/channels.lua b/internal/lua/plugins/channels.lua index a5342ba..81bbb09 100644 --- a/internal/lua/plugins/channels.lua +++ b/internal/lua/plugins/channels.lua @@ -52,7 +52,7 @@ plugin.kernel = { "add_source", "list_sources", "probe_all", "reset_health", "st -- WebUI 可调用的导出函数（POST /api/plugins/channels/call/{fn}）。 -- set_endpoints 让运维把渠道端点指向自建代理/镜像；测试用它打假上游。 -plugin.exports = { "set_endpoints", "account_add", "account_list", "account_update", "account_delete", "account_test", "account_refresh", "account_reorder", "account_reset", "account_retest", "account_adopt", "model_list", "model_refresh", "model_set_disabled", "model_set_disabled_many", "model_set_all_disabled", "model_clear_dead", "provider_status", "provider_set_enabled", "provider_get_order", "provider_set_order", "credits_claim_all", "credits_status", "credits_balances", "credits_permanent_lock", "backup_export", "backup_import", "backup_status", "auto_checkin", "gateway_get_enabled", "gateway_set_enabled", "usage_token_ledger", "usage_badge", "usage_badge_preference", "onboarding_status", "onboarding_claim", "__trae_derive_probe", "__codearts_sign_probe", "__qoder_machine_probe", "__qoder_refresh_probe", "__codearts_dpop_probe", "__codearts_cred_probe", "__codearts_adapter_sign_probe", "__sync_source_probe", "__trae_solo_probe", "__adapter_install_probe", "__store_dir_probe", "__set_loopback_probe", "__is_model_gone_probe", "__rank_accounts_probe", "__inject_refresh_platform_probe", "__clear_capabilities_probe", "__qoder_claim_outcome_probe", "__is_free_probe", "__claim_covers_probe", "__schedule_probe", "__parse_capability_probe", "__inject_capability_probe", "__ttl_reset_probe", "__ttl_probe", "__ttl_invalidate_probe", "__serial_probe", "__serial_release_probe", "__set_legacy_cred_probe", "__read_cred_probe", "__clear_all_creds_probe", "credits_tier_probe", "__set_lock_permanent_probe", "__set_expiring_window_probe", "__expiring_window_probe", "__record_dead_probe", "__catalog_gate_probe", "__catalog_gate_fail_probe", "credits_capability", "claims_capable" } +plugin.exports = { "set_endpoints", "account_add", "account_list", "account_update", "account_delete", "account_test", "account_refresh", "account_reorder", "account_reset", "account_retest", "account_adopt", "model_list", "model_refresh", "model_set_disabled", "model_set_disabled_many", "model_set_all_disabled", "model_clear_dead", "provider_status", "provider_set_enabled", "provider_get_order", "provider_set_order", "credits_claim_all", "credits_status", "credits_balances", "credits_permanent_lock", "backup_export", "backup_import", "backup_status", "auto_checkin", "gateway_get_enabled", "gateway_set_enabled", "usage_token_ledger", "usage_badge", "usage_badge_preference", "onboarding_status", "onboarding_claim", "__trae_derive_probe", "__codearts_sign_probe", "__qoder_machine_probe", "__qoder_refresh_probe", "__codearts_dpop_probe", "__codearts_cred_probe", "__codearts_adapter_sign_probe", "__sync_source_probe", "__trae_solo_probe", "__adapter_install_probe", "__store_dir_probe", "__set_loopback_probe", "reap_loopbacks", "__is_model_gone_probe", "__rank_accounts_probe", "__inject_refresh_platform_probe", "__clear_capabilities_probe", "__qoder_claim_outcome_probe", "__is_free_probe", "__claim_covers_probe", "__schedule_probe", "__parse_capability_probe", "__inject_capability_probe", "__ttl_reset_probe", "__ttl_probe", "__ttl_invalidate_probe", "__serial_probe", "__serial_release_probe", "__set_legacy_cred_probe", "__read_cred_probe", "__clear_all_creds_probe", "credits_tier_probe", "__set_lock_permanent_probe", "__set_expiring_window_probe", "__expiring_window_probe", "__record_dead_probe", "__catalog_gate_probe", "__catalog_gate_fail_probe", "credits_capability", "claims_capable" } @@ -2622,6 +2622,23 @@ plugin.state.log = plugin.state.log or {} -- 字段，先填再被 or {} 覆盖就白填了）。 loadState() +-- ★ 启动清扫：删掉上一进程留下的孤儿回调文件。 +-- +-- 为什么需要它：loopbacks 追踪表只活在内存里（plugin.state.loopbacks）， +-- 进程退出后无从得知上一轮起过哪些接收器。于是进程被强杀/SIGKILL 时， +-- 那句 on_unload 没机会跑，磁盘上就永久留下 callback-*.port / .json。 +-- 实测生产攒了 18 个孤儿 .port（对应的 .json 早已消失），且**永远不会被清** —— +-- 重启后内存表是空的，reapLoopback 根本不知道它们存在。 +-- +-- 判据不是"文件被删了"这么弱，而是"启动后目录里不再有上一轮的残留"： +-- internal/core/channels_orphan_callback_test.go。 +do + local dir = storeDir() + local sh = io.popen and io.popen("rm -f " .. dir .. + "/callback-*.port " .. dir .. "/callback-*.json 2>/dev/null") + if sh then sh:close() end +end + local function logEvent(msg) table.insert(plugin.state.log, 1, { at = os.time and os.time() or 0, msg = msg }) while #plugin.state.log > 50 do table.remove(plugin.state.log) end @@ -6269,6 +6286,8 @@ plugin.http = { description = "把会话的 noVNC 静态资源与 WebSocket 经网关同端口转发（不暴露 VNC 端口）" }, { path = "channels/browser/collect", method = "POST", handler = "collect_credentials", roles = { "admin" }, description = "从服务端浏览器取凭据（CDP）" }, + { path = "channels/reap_loopbacks", method = "POST", handler = "reap_loopbacks", + roles = { "admin" }, description = "回收所有回环回调接收器（防测试/异常退出后泄漏）" }, { path = "channels/browser/stop", method = "POST", handler = "stop_browser", roles = { "admin" }, description = "回收服务端浏览器栈" }, } @@ -7495,6 +7514,24 @@ function plugin.stop_browser(req) end -- browser_status：当前浏览器会话状态（给 UI 显示）。 +-- reap_loopbacks：回收**所有**回环接收器。 +-- +-- 为什么需要显式入口：回环本来只在**会话过期**时由 reapLoopback 收掉 +-- （sweepSessions 是惰性的，靠读取时顺带做）。于是有一条稳定的泄漏路径： +-- 测试或运维起了 login 会话后直接退出，会话从未「过期」⇒ 接收器进程常驻。 +-- 实测本机积累到 995 个（其中 828 个超过 24 小时），全是 /tmp/Test*/ 下的 +-- 测试残留 —— 生产路径的会被正常回收（因为生产会话真会过期）。 +-- +-- 对生产也有用：浏览器/授权页关掉后想立刻清理，不必等会话过期。 +function plugin.reap_loopbacks(req) + local n = 0 + for sid in pairs(plugin.state.loopbacks or {}) do + reapLoopback(sid) + n = n + 1 + end + return { status = 200, json = { ok = true, reaped = n } } +end + function plugin.browser_status(req) local b = browserStack() local s = b.session @@ -9568,8 +9605,14 @@ function plugin.__set_loopback_probe(args) local file = tostring(args.file or "") if sid == "" or file == "" then return { json = { error = "need session_id and file" } } end if type(plugin.state.loopbacks) ~= "table" then plugin.state.loopbacks = {} end - plugin.state.loopbacks[sid] = { file = file } - return { json = { ok = true, session_id = sid, file = file } } + -- ★ 合并而非覆盖：真实会话的条目里带着 pid/port_file，覆盖掉它们 + -- 会让 reapLoopback 杀不掉那个 python 接收器 —— 探针自己就成了泄漏源 + -- （实测：覆盖写法在该测试里留下 1 个孤儿进程，而其他测试已归零）。 + local cur = plugin.state.loopbacks[sid] + if type(cur) ~= "table" then cur = {} end + cur.file = file + plugin.state.loopbacks[sid] = cur + return { json = { ok = true, session_id = sid, file = file, kept_pid = cur.pid or "" } } end -- plugin.credits_tier_probe：只给测试用 —— 单账号的余额分档结果。