fix(kernel): Close() 从不调 on_unload —— 孤儿进程与回调文件泄漏

## 症状
本机积累 **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 还原。教训:变异测试必须逐次收尾,别靠记忆。
This commit is contained in:
JianFeeeee
2026-10-15 22:43:08 +08:00
parent 8d23005f7d
commit 89d759ece7
4 changed files with 271 additions and 21 deletions

View File

@ -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{}

View File

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

View File

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

View File

@ -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:只给测试用 —— 单账号的余额分档结果。