diff --git a/internal/plugin/proc/arena.go b/internal/plugin/proc/arena.go index 8587db97..89d75faa 100644 --- a/internal/plugin/proc/arena.go +++ b/internal/plugin/proc/arena.go @@ -458,6 +458,40 @@ func (a *arenaRegion) coalesceLocked(off uint32) { } } +// ReclaimRefs 强制回收**指定的一批**块(不校验 owner),返回回收块数。 +// +// 用途:取消超时的强制回收。这些块的 owner 是**插件**,而回收决策来自内核—— +// 内核在此刻是全权的(见文件头的所有权模型),所以这里刻意**不**走 +// Free 的归属校验:那条校验是防"插件伪造 ref 动别人的块", +// 而这里要防的是完全相反的事——插件已经不打算/已经不能归还了。 +// +// 安全边界仍在:每个 ref 都必须真的是一个**已分配块的数据起点** +// (findBlockByData 走完整块链),伪造/越界的 offset 一律跳过。 +func (a *arenaRegion) ReclaimRefs(refs []SharedRef) int { + if len(refs) == 0 { + return 0 + } + a.mu.Lock() + defer a.mu.Unlock() + + n := 0 + for _, ref := range refs { + off, ok := a.findBlockByData(ref.Offset) + if !ok { + continue + } + if a.blockState(off) != blockUsed { + continue // 已被插件自己归还或已被前面的合并吸收 + } + a.setBlockState(off, blockFree) + a.setBlockOwner(off, OwnerHost) + a.accountUsed(-int32(align8(ref.Length))) + a.coalesceLocked(off) + n++ + } + return n +} + // ReclaimOwner 归还 owner 名下所有块,返回回收块数。 // // 用于插件进程退出:崩溃的插件无法归还自己申请的块,若不管会把 arena diff --git a/internal/plugin/proc/arena_leak_test.go b/internal/plugin/proc/arena_leak_test.go new file mode 100644 index 00000000..c9a045dc --- /dev/null +++ b/internal/plugin/proc/arena_leak_test.go @@ -0,0 +1,69 @@ +package proc + +import ( + "strings" + "testing" +) + +// 回归判据:内核代插件回收扩容块时必须用**插件自己的 ownerID**。 +// +// 病根(2026-10-25 线上):插件经 arena.alloc 申请扩容块时块头 owner 是 +// 它自己的 ownerID,而内核三处归还全写着 Free(OwnerHost, ref)。归属校验 +// 必然失败、错误又被 `_ =` 吞掉 ⇒ 每返回一次大结果就永久泄漏一块。 +// 4MB 的 arena 被大工具结果(单次 64KB+ 的返回)慢慢吃满后,连内核标定的 +// 64KB 调用帧都分不出来,所有插件工具一起报「arena: 空间不足」。 +// +// 判据两半: +// 1. Free(OwnerHost, 插件块) 必须失败 —— 确认这条路径本来就是错的 +// 2. Free(插件owner, 插件块) 必须成功且 Stats 归零 —— 确认修法真的回收 +// +// 变异:把 invokeTool / invokeCleaner / invokeOutput 里任一处改回 +// Free(OwnerHost, ...),E2E 判据立即变红。 +func TestArena_ExpandBlockReclaimedWithPluginOwner(t *testing.T) { + a := newTestArena(t, 1<<20) + const pluginOwner = 7 + + ref, err := a.Alloc(pluginOwner, 100000, 1) + if err != nil { + t.Fatalf("Alloc: %v", err) + } + + // ① 旧写法必须失败:这就是泄漏机制本身。 + err = a.Free(OwnerHost, ref) + if err == nil { + t.Fatal("Free(OwnerHost, 插件块) 竟然成功了——归属校验失效,本判据失去意义") + } + if !strings.Contains(err.Error(), "不属于调用者") { + t.Fatalf("期望归属校验错误,实际: %v", err) + } + if used, _ := a.Stats(); used == 0 { + t.Fatal("旧路径失败后块却被回收了?") + } + + // ② 正确写法:用插件自己的 ownerID 归还。 + if err := a.Free(pluginOwner, ref); err != nil { + t.Fatalf("按插件 owner 归还应成功: %v", err) + } + if used, _ := a.Stats(); used != 0 { + t.Fatalf("回收后 arena 应归零,实际 used=%d", used) + } +} + +// 扩容块归还后空间必须可复用:连做 30 轮「申请 200KB → 归还」, +// 总分配量远超 arena 容量,但 used 每轮都应回到 0。 +func TestArena_ExpandBlocksDoNotAccumulate(t *testing.T) { + a := newTestArena(t, 1<<20) + const owner = 3 + for i := 0; i < 30; i++ { + ref, err := a.Alloc(owner, 200*1024, 1) + if err != nil { + t.Fatalf("第 %d 轮分配失败(累计分配 %dKB 未回收?): %v", i, i*200, err) + } + if err := a.Free(owner, ref); err != nil { + t.Fatalf("第 %d 轮归还失败: %v", i, err) + } + if used, _ := a.Stats(); used != 0 { + t.Fatalf("第 %d 轮后 used=%d,块未回收", i, used) + } + } +} diff --git a/internal/plugin/proc/callstate.go b/internal/plugin/proc/callstate.go new file mode 100644 index 00000000..33e5f01f --- /dev/null +++ b/internal/plugin/proc/callstate.go @@ -0,0 +1,165 @@ +package proc + +import ( + "sync" + "sync/atomic" +) + +// 单次跨进程调用的**状态机**(SDK / 传输层内部逻辑,插件作者看不到)。 +// +// ## 为什么需要显式状态而不是一个 map + delete +// +// 旧实现把「调用在进行中」这件事只表达为 `pending[id] 存在`。取消时 +// 直接 `delete(pending, id)` —— 于是状态的**全部痕迹**都消失了,之后 +// 插件返回的结果无法与任何状态对上,只能走「未知 id,丢弃 + 打日志」。 +// +// 那条路径能跑,但它是**副作用式的正确**:正确性依赖"恰好没人再读这个 +// 通道",而不是依赖某个明确规则。真实后果有三个: +// +// 1. 取消后插件仍会**回一个应答**(它不知道已经没人要了),内核每次都要 +// 打一条"丢弃迟到应答"日志——在取消频繁的场景里这就是噪声,且很像是 +// 故障信号(第一版我确实把它误判成管道问题)。 +// 2. 无法区分「取消后插件认输」与「取消后插件其实跑完了」——两者对 +// 上层(是否该重试、是否该记日志)意义完全不同。 +// 3. 没有任何地方能回答「这次调用现在处于什么状态」,诊断只能靠猜。 +// +// ## 状态迁移 +// +// pending ──应答到达────────→ done +// │ +// ├──取消/超时──→ canceled ──迟到应答到达──→ done(结果被丢弃,但状态被记录) +// │ +// └──进程退出──→ failed +// +// 关键规则:**状态只前进一步,且不可回退**。cancelled 之后再收到的应答 +// 不会把状态改回 done("完成"),只是把"结果到了"这件事记下来。 +// +// 所有迁移都是 CAS,因此并发(应答与取消赛跑)时谁先到谁赢,且只有一方 +// 会得到迁移成功的返回值——这正是"该由谁负责收尾"的判据。 +type callState int32 + +const ( + // callPending 调用已发出,等待应答。 + callPending callState = iota + // callDone 已收到应答(或确定的失败),结果已交回调用方。 + callDone + // callCanceled 调用方已放弃(超时/取消/抢占)。 + // + // 与 callFailed 分开:取消是**主动**的(我方不再要结果), + // 进程退出是**被动**的(对端没了)。上层的处置不同——取消不该报警, + // 进程退出该触发插件健康计数。 + callCanceled + // callFailed 对端进程退出,调用无法完成。 + callFailed +) + +func (s callState) String() string { + switch s { + case callPending: + return "pending" + case callDone: + return "done" + case callCanceled: + return "canceled" + case callFailed: + return "failed" + } + return "unknown" +} + +// call 是一次跨进程调用的**全部状态**。 +// +// 所有与该调用相关的状态都放在这里,而不是散落到几张平行 map: +// +// state —— 生命周期(pending/done/canceled/failed) +// ch —— 应答投递通道 +// blocks —— 该调用申请过的扩容块(取消超时时要强制回收) +// +// 放在一起的理由:这几样东西的**生命周期完全相同**(都随调用开始、随调用收敛)。 +// 拆成多张表就必须各自维护增删与过期,且任何一处漏删都是泄漏—— +// 第一版我确实这么写过(独立的 callblocks 表 + 自己的 TTL), +// 结果是同一件事记在两处、两个过期策略要同时正确。现在只有一处。 +type call struct { + id uint64 + ch chan *Response + kind string // 调用类型(tool.invoke / stage.invoke …),仅用于日志与诊断 + state atomic.Int32 // callState + cancel atomic.Bool // 是否已向对端发过 call.cancel(避免重复通知) + + // blocksMu 保护 blocks(分配来自插件 RPC 的另一个 goroutine)。 + blocksMu sync.Mutex + blocks []SharedRef +} + +// trackBlock 记下一块属于本次调用的扩容块。 +// +// 为什么要记:插件为大结果申请扩容块后,若调用被取消,插件会静默退出 +// (正确行为),内核**从未见过**那个 ResultRef —— 块就永远还不回来了, +// 且全程没有任何日志。记在这里,取消超时时内核才能自己把账清了。 +func (c *call) trackBlock(ref SharedRef) { + if c == nil || ref.IsZero() { + return + } + c.blocksMu.Lock() + c.blocks = append(c.blocks, ref) + c.blocksMu.Unlock() +} + +// takeBlocks 取出并清空本次调用名下的块。 +func (c *call) takeBlocks() []SharedRef { + if c == nil { + return nil + } + c.blocksMu.Lock() + defer c.blocksMu.Unlock() + out := c.blocks + c.blocks = nil + return out +} + +func newCall(id uint64, kind string) *call { + c := &call{id: id, ch: make(chan *Response, 1), kind: kind} + c.state.Store(int32(callPending)) + return c +} + +// legalTransition 是**全局唯一**的合法迁移表。 +// +// 为什么要有它,而不是让每个调用方自己挑 from/to: +// transitionTo 是 CAS,它只校验"当前值是 from",**不校验 to 是否合适**。 +// 于是 `transitionTo(callDone, callFailed)` 这种明显非法的调用会静默成功, +// 把终态改回去——状态机的单向性就白写了。实测中这确实出现过一次 +// (retire 曾写成 `pending→to || canceled→to`,导致 done 可以被改成 failed)。 +// +// 现在把"能不能迁移"集中在一处判定:新增迁移必须在这里登记, +// 写错的 from/to 组合会直接被拒(并留下日志线索)。 +var legalTransition = map[callState][]callState{ + callPending: {callDone, callCanceled, callFailed}, + callCanceled: {callDone, callFailed}, // 迟到应答 / 进程随后退出 + callDone: nil, // 终态 + callFailed: nil, // 终态 +} + +func canTransition(from, to callState) bool { + for _, ok := range legalTransition[from] { + if ok == to { + return true + } + } + return false +} + +// transitionTo 尝试把状态从 from 迁移到 to(CAS)。 +// 返回 true 表示**本次调用**赢得了这次迁移,负责后续动作。 +// +// 非法组合(不在 legalTransition 里)一律返回 false 且状态不变—— +// 让错误尽早暴露成"迁移没发生",而不是悄悄把终态改回去。 +func (c *call) transitionTo(from, to callState) bool { + if !canTransition(from, to) { + return false + } + return c.state.CompareAndSwap(int32(from), int32(to)) +} + +// State 返回当前状态(诊断用)。 +func (c *call) State() callState { return callState(c.state.Load()) } diff --git a/internal/plugin/proc/callstate_test.go b/internal/plugin/proc/callstate_test.go new file mode 100644 index 00000000..91378ccb --- /dev/null +++ b/internal/plugin/proc/callstate_test.go @@ -0,0 +1,200 @@ +package proc + +import ( + "context" + "sync" + "testing" + "time" +) + +// 状态机的基本性质:只能前进,不可回退。 +func TestCallState_TransitionsAreMonotonic(t *testing.T) { + c := newCall(1, "tool.invoke") + if c.State() != callPending { + t.Fatalf("初始应为 pending,实际 %s", c.State()) + } + if !c.transitionTo(callPending, callCanceled) { + t.Fatal("pending→canceled 应成功") + } + if c.State() != callCanceled { + t.Fatalf("应为 canceled,实际 %s", c.State()) + } + // 回退必须失败:canceled 之后不能变回 pending,也不能被当成"又一次取消"。 + if c.transitionTo(callPending, callDone) { + t.Fatal("canceled 不得回退到 pending") + } + if c.transitionTo(callPending, callCanceled) { + t.Fatal("重复取消不应再次赢得迁移(否则会重复发 cancel 通知)") + } + // canceled → done:允许(迟到应答到达时记录结果已到)。 + if !c.transitionTo(callCanceled, callDone) { + t.Fatal("canceled→done 应成功(迟到应答仍要有归属)") + } + // 之后再无迁移可能。 + if c.transitionTo(callDone, callFailed) { + t.Fatal("done 之后不得再迁移") + } +} + +// 并发竞争:无论应答与取消谁先到,最终状态必须是二者之一,且迁移只能成功一次。 +// +// 这条守的是"取消与应答赛跑"这个真实场景(插件可能刚好在 cancel 生效前跑完)。 +func TestCallState_ConcurrentTransitionWinsOnce(t *testing.T) { + for i := 0; i < 200; i++ { + c := newCall(uint64(i), "tool.invoke") + var wins int32 + var mu sync.Mutex + var wg sync.WaitGroup + // 一个 goroutine 声称"完成",另一个声称"取消"。 + for _, to := range []callState{callDone, callCanceled} { + wg.Add(1) + go func(to callState) { + defer wg.Done() + if c.transitionTo(callPending, to) { + mu.Lock() + wins++ + mu.Unlock() + } + }(to) + } + wg.Wait() + if wins != 1 { + t.Fatalf("第 %d 轮:应恰好一方赢得 pending 迁移,实际 %d 方", i, wins) + } + } +} + +// 取消后到达的应答**必须能被认出来**,而不是被当成"未知 id"。 +// +// 这是本次改造的核心判据。旧实现取消时直接 delete(pending, id), +// 于是迟到的应答只能走"找不到 id ⇒ 丢弃 + 打日志",它无法说明这条应答 +// 属于哪次调用、那次调用为什么结束。现在状态保留,因此可以精确归类。 +// +// 变异:把 abandon 改回 delete(p.inflight, id)(不迁移状态) +// ⇒ 本判据变红(应答会被判为"未知 id")。 +func TestProcess_CanceledCallStillOwnsItsLateReply(t *testing.T) { + p := startKillablePlugin(t) + defer p.Kill() + + // ★ 必须走**真实的** abandon 路径,手工构造状态会绕过被检验的代码。 + // + // 第一版我就是手工 transitionTo + registerCallForTest,结果 + // "变异:abandon 摘表" 依然全绿——测试根本没覆盖 abandon。 + c := newCall(p.allocCallID(t), "tool.invoke") + p.registerCallForTest(c) + p.abandon(c, context.Canceled) // ← 被检验的就是它摘不摘表 + if c.State() != callCanceled { + t.Fatalf("abandon 后应为 canceled,实际 %s", c.State()) + } + + // 插件侧此时才回一个应答。按新逻辑:id 仍在表里、状态是 canceled, + // 应答应被识别(canceled→done)而不是报"未知 id"。 + delivered := p.deliverResponseForTest(&Response{ID: c.id, Result: []byte(`"late"`)}) + if !delivered { + t.Fatal(`取消后的迟到应答应被识别为属于该调用,而不是「未知 id」`) + } + if c.State() != callDone { + t.Fatalf("迟到应答应把状态推进到 done,实际 %s", c.State()) + } +} + +// 已经结束并摘表的调用再来应答,才是真正的未知 id。 +func TestProcess_UnknownIDIsReported(t *testing.T) { + p := startKillablePlugin(t) + defer p.Kill() + + if p.deliverResponseForTest(&Response{ID: 999999}) { + t.Fatal("表里没有的 id 不应被投递") + } +} + +// ---- 测试辅助 ---- + +func (p *Process) allocCallID(t *testing.T) uint64 { + t.Helper() + p.mu.Lock() + defer p.mu.Unlock() + if p.inflight == nil { + p.inflight = make(map[uint64]*call) + } + p.nextID++ + return p.nextID +} + +func (p *Process) registerCallForTest(c *call) { + p.mu.Lock() + if p.inflight == nil { + p.inflight = make(map[uint64]*call) + } + p.inflight[c.id] = c + p.mu.Unlock() +} + +// deliverResponseForTest 直接走**真实的**分发实现(deliverResponse)。 +// +// 第一版我在这里复制了一份分发逻辑,结果 onCancelHonored 只在真路径里 +// 被调用,测试却显示通过——假绿。现在只做一行转发,保证测的就是线上代码。 +func (p *Process) deliverResponseForTest(resp *Response) bool { + before := p.inflightCountForTest() + p.deliverResponse(resp) + return p.inflightCountForTest() < before +} + +func (p *Process) inflightCountForTest() int { + p.mu.Lock() + defer p.mu.Unlock() + return len(p.inflight) +} + +func startKillablePlugin(t *testing.T) *Process { + t.Helper() + bin := buildTestPlugin(t, "echoplugin.go") + p, err := Spawn("state", bin, Options{Handler: noopHandler}) + if err != nil { + t.Fatalf("Spawn: %v", err) + } + t.Cleanup(func() { _ = p.Kill() }) + // 等它就绪,避免后续测试与握手竞争。 + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if _, err := p.CallContext(ctx, "plugin.init", map[string]interface{}{"name": "state"}); err != nil { + // echoplugin 可能不支持 plugin.init,忽略——本测试只用表状态。 + t.Logf("plugin.init(可忽略): %v", err) + } + return p +} + +// 宽限期到期后,已取消且对端始终不回答的条目必须被回收。 +// +// 守的是"取消频繁 + 对端从不回答"这个组合:条目若永久保留, +// 内存会随取消次数单调增长。 +// +// 这里把宽限期临时缩短到 ms 级——否则测试要等一分钟。 +func TestProcess_CanceledCallReclaimedAfterGrace(t *testing.T) { + p := startKillablePlugin(t) + + c := newCall(p.allocCallID(t), "tool.invoke") + p.registerCallForTest(c) + p.abandon(c, context.Canceled) + + // 宽限期内应仍在表里(这样迟到的应答才认得出归属)。 + if !p.hasInflight(c) { + t.Fatal("取消后应立即保留条目(供迟到应答认领)") + } + + // 直接触发回收:等价于宽限期到期时 time.AfterFunc 做的事。 + p.retire(c, callFailed) + + if p.hasInflight(c) { + t.Fatal("宽限期后条目必须被回收,否则内存随取消次数单调增长") + } + if c.State() != callFailed { + t.Fatalf("回收后应为 failed,实际 %s", c.State()) + } +} + +func (p *Process) hasInflight(c *call) bool { + p.mu.Lock() + defer p.mu.Unlock() + return p.inflight[c.id] == c +} diff --git a/internal/plugin/proc/cancel_ignored_test.go b/internal/plugin/proc/cancel_ignored_test.go new file mode 100644 index 00000000..8dbfd5d4 --- /dev/null +++ b/internal/plugin/proc/cancel_ignored_test.go @@ -0,0 +1,156 @@ +package proc + +import ( + "bufio" + "context" + "io" + "sync/atomic" + "testing" + "time" +) + +// newHeadlessProcess 构造一个**不 spawn 真进程**的 Process,供状态机单测使用。 +// +// stdin 接到 io.Discard:abandon 会真的发 call.cancel 通知, +// 没有写端就会 nil 解引用。这样测试仍然走**真实代码路径** +// (包括发通知这一步),而不是把它 stub 掉——后者会让"取消到底有没有发出去" +// 变得无法被覆盖。 +func newHeadlessProcess(name string) *Process { + return &Process{ + name: name, + stdin: bufio.NewWriter(io.Discard), + inflight: make(map[uint64]*call), + exited: make(chan struct{}), + } +} + +// 取消超时必须:① 强制回收该调用名下的扩容块;② 上报插件失去响应。 +// +// 判据要区分两类超时(这是本设计的核心,2026-10-25 用户裁定): +// +// 业务超时 = 任务跑太久 → 正常,不作为不稳定依据 +// 取消超时 = 喊停之后还不停 → 异常,是插件卡死的直接证据 +// +// 变异:删掉 expireCanceled 里的 ReclaimRefs 调用 ⇒ 块泄漏判据变红。 +// 变异:不调 onCancelIgnored ⇒ 上报判据变红。 +func TestProcess_CancelIgnoredReclaimsBlocksAndReports(t *testing.T) { + host, err := NewHost() + if err != nil { + t.Fatalf("NewHost: %v", err) + } + defer host.Close() + + var ( + reportedCalls atomic.Uint64 + reclaimed atomic.Int64 + ) + p := newHeadlessProcess("probe") + p.SetReclaimBlocks(func(refs []SharedRef) int { + n := host.ReclaimRefs(refs) + reclaimed.Add(int64(n)) + return n + }) + p.OnCancelIgnored(func(callID uint64, grace time.Duration) { + reportedCalls.Add(1) + }) + + // 造一次"插件申请了扩容块、随后被取消"的调用。 + gen := host.Generation() + ref, err := host.Arena().Alloc(42, 4096, gen) // owner 42 = 插件 + if err != nil { + t.Fatalf("Alloc: %v", err) + } + c := newCall(7, "tool.invoke") + c.trackBlock(ref) + p.registerCallForTest(c) + + if used, _ := host.Arena().Stats(); used == 0 { + t.Fatal("前置:块应已占用") + } + + p.abandon(c, context.Canceled) + + // 宽限期未到:不应回收(插件可能还在收尾)。 + if reclaimed.Load() != 0 { + t.Fatal("宽限期内不应强制回收") + } + + // 模拟宽限期到期。 + p.expireCanceled(c) + + // ① 块必须被强制回收 —— 否则"取消了却瞒下内存",无日志只涨不落。 + if used, _ := host.Arena().Stats(); used != 0 { + t.Fatalf("取消超时后 arena 应归零,实际 used=%d(块泄漏)", used) + } + if reclaimed.Load() == 0 { + t.Fatal("应调用强制回收") + } + + // ② 必须上报(上层据此判定不稳定并重启)。 + if reportedCalls.Load() != 1 { + t.Fatalf("应上报 1 次取消无响应,实际 %d", reportedCalls.Load()) + } + + // ③ 状态收敛到 failed(这次调用确实没正常完成)。 + if c.State() != callFailed { + t.Fatalf("状态应为 failed,实际 %s", c.State()) + } +} + +// 插件在宽限期内停手(回了应答)→ 不上报,且连续计数应被清零。 +func TestProcess_CancelHonoredDoesNotReport(t *testing.T) { + var reported, honored atomic.Int64 + p := newHeadlessProcess("probe") + p.OnCancelIgnored(func(uint64, time.Duration) { reported.Add(1) }) + p.OnCancelHonored(func(uint64) { honored.Add(1) }) + + c := newCall(11, "tool.invoke") + p.registerCallForTest(c) + p.abandon(c, context.Canceled) + + // 插件随后回了应答(说明它收到 cancel 并停手了)。 + if !p.deliverResponseForTest(&Response{ID: c.id, Result: []byte(`"x"`)}) { + t.Fatal("取消后的迟到应答应被认领") + } + + if honored.Load() != 1 { + t.Fatalf("应记为已响应 1 次,实际 %d", honored.Load()) + } + + // 宽限期后再触发:已收敛,不应再上报。 + p.expireCanceled(c) + if reported.Load() != 0 { + t.Fatal("已响应的调用不应被上报为无响应") + } +} + +// 连续两次取消无响应才重启;中间成功响应一次即清零。 +// +// 这是用户明确要求的判据:"注意是 cancel 态超时,超过两次说明插件本身 +// 处于崩溃/不稳定态,自动重启对应插件"。 +func TestPlugin_CancelIgnoredThreshold(t *testing.T) { + p := &Plugin{name: "flaky"} + var restarts atomic.Int64 + p.onCrash = func(name string, err error) { restarts.Add(1) } + + // 第 1 次:不重启(可能是收尾慢)。 + p.noteCancelIgnored(1, cancelGrace) + if restarts.Load() != 0 { + t.Fatal("第 1 次取消无响应不应重启") + } + + // 第 2 次:达到阈值,重启。 + p.noteCancelIgnored(2, cancelGrace) + if restarts.Load() != 1 { + t.Fatalf("连续第 %d 次应触发重启,实际 %d", cancelIgnoredLimit, restarts.Load()) + } + + // 计数清零后重新开始("连续"而非"累计")。 + p.cancelIgnoredMu.Lock() + p.cancelIgnored = 0 + p.cancelIgnoredMu.Unlock() + p.noteCancelIgnored(3, cancelGrace) + if restarts.Load() != 1 { + t.Fatal("清零后第一次不应再次重启(判据是连续)") + } +} diff --git a/internal/plugin/proc/cancel_probe_test.go b/internal/plugin/proc/cancel_probe_test.go new file mode 100644 index 00000000..3fde7027 --- /dev/null +++ b/internal/plugin/proc/cancel_probe_test.go @@ -0,0 +1,84 @@ +package proc + +// cancelProbePluginSource 是「取消是否真的送达」的观察装置。 +// +// 设计要点: +// - probe_slow 用 **ToolHandlerCtx** 注册,观察 ctx 并记录结果; +// - 观察到取消后把状态写进包级变量(同进程,测试可直接读回); +// - probe_state 返回该变量,供测试轮询。 +// +// 之所以能这么写:模板真的会把 call.cancel 变成 handler 的 ctx cancel +// (见 hmapdev 模板的 trackInflight/cancelInflight)。若那一环没接上, +// probe_slow 永远观察不到取消,测试自然红。 +const cancelProbePluginSource = `package main + +import ( + "context" + "sync" + "time" + + sdk "github.com/JianFeeeee/hologramagentsdk/sdk" +) + +type probePlugin struct { + name string + mu sync.Mutex + last string +} + +func NewPluginFactory(name string, config map[string]interface{}) (sdk.Plugin, error) { + return &probePlugin{name: name}, nil +} + +func (p *probePlugin) Name() string { return p.name } + +func (p *probePlugin) Stop() error { return nil } + +func (p *probePlugin) Start(s *sdk.PluginSDK) error { + s.RegisterToolCtx("probe_slow", sdk.ToolDef{ + Description: "长任务,用于验证取消是否送达", + Parameters: map[string]interface{}{"type": "object", "properties": map[string]interface{}{}}, + }, func(ctx context.Context, args map[string]interface{}) (interface{}, error) { + select { + case <-ctx.Done(): + p.mu.Lock() + p.last = "canceled" + p.mu.Unlock() + return "canceled", ctx.Err() + case <-time.After(30 * time.Second): + p.mu.Lock() + p.last = "timeout" + p.mu.Unlock() + return "finished", nil + } + }) + + // probe_block:**用旧签名**注册,完全不看 ctx。 + // + // 这个工具是"取消后不回应答"那条路径的**唯一**观察点: + // Ctx 版 handler 会因为 ctx 取消而自己提前返回,从而绕过 + // "跑完后检查是否已被取消" 那段代码,测不到它。 + // 旧签名 handler 跑满 300ms 后才返回——此时内核早已取消, + // 运行时必须靠 inflightCall.canceled 拦住应答。 + s.RegisterTool("probe_block", sdk.ToolDef{ + Description: "旧签名长任务:不看 ctx,用于验证取消后不回应答", + Parameters: map[string]interface{}{"type": "object", "properties": map[string]interface{}{}}, + }, func(args map[string]interface{}) (interface{}, error) { + time.Sleep(300 * time.Millisecond) + return "finished", nil + }) + + s.RegisterTool("probe_state", sdk.ToolDef{ + Description: "读回上一次长任务的观察结果", + Parameters: map[string]interface{}{"type": "object", "properties": map[string]interface{}{}}, + }, func(args map[string]interface{}) (interface{}, error) { + p.mu.Lock() + defer p.mu.Unlock() + if p.last == "" { + return "idle", nil + } + return p.last, nil + }) + return nil +} +` diff --git a/internal/plugin/proc/cancel_test.go b/internal/plugin/proc/cancel_test.go new file mode 100644 index 00000000..e27b015d --- /dev/null +++ b/internal/plugin/proc/cancel_test.go @@ -0,0 +1,183 @@ +package proc + +import ( + "context" + "errors" + "strings" + "testing" + "time" +) + +// ---- 单元:旁路 ctx 的存取与剥除 ---- + +// 旁路键必须**不泄漏**给插件,同时不影响真实参数。 +func TestExecCtxArg_RoundtripAndStrip(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + args := map[string]interface{}{"path": "/tmp/a"} + bounded := map[string]interface{}{"path": "/tmp/a", execCtxArgKey: ctx} + + if got := execCtxFromArgs(bounded); got != ctx { + t.Fatal("应取回绑定的 ctx") + } + got := stripExecCtx(bounded) + if _, leaked := got[execCtxArgKey]; leaked { + t.Fatal("旁路键必须从下发给插件的参数里剥掉(否则严格校验参数的插件会报未知参数)") + } + if got["path"] != "/tmp/a" { + t.Fatal("剥除不应影响真实参数") + } + + // 没有绑定键:原样返回,不在热路径上多复制一次。 + plain := stripExecCtx(args) + if len(plain) != 1 || plain["path"] != "/tmp/a" { + t.Fatal("无旁路键时应原样返回") + } +} + +// 没有绑定时必须返回可用的 ctx(调用方无需判空)。 +func TestExecCtxArg_DefaultsToBackground(t *testing.T) { + for _, args := range []map[string]interface{}{nil, {}, {"a": 1}} { + ctx := execCtxFromArgs(args) + if ctx == nil { + t.Fatal("必须返回非 nil ctx") + } + if ctx.Err() != nil { + t.Fatal("默认 ctx 不应处于取消态") + } + } +} + +// ---- 端到端:取消真的送达插件 ---- + +// 内核超时后必须让**插件**停手,而不只是自己不再等。 +// +// 判据:用真实 hmapdev 模板编译一个注册了长任务工具的插件,插件侧用 +// ToolHandlerCtx 观察 ctx;内核取消后,插件必须很快观察到取消并回报。 +// +// 变异:①去掉 Process.CallContext 里的 cancelCall(...) ⇒ 插件永远收不到取消, +// 本判据超时变红;②模板不处理 call.cancel ⇒ 同上。 +func TestE2E_CancelReachesPlugin(t *testing.T) { + bin := buildPluginWithRealTemplate(t, cancelProbePluginSource) + + host, err := NewHost() + if err != nil { + t.Fatalf("NewHost: %v", err) + } + defer host.Close() + + core := newFakeCore() + plg := New("probe", bin, t.TempDir(), nil, host, nil) + if err := plg.Start(core); err != nil { + t.Fatalf("Start: %v", err) + } + defer plg.Close() + + core.mu.Lock() + _, ok := core.tools["probe_slow"] + core.mu.Unlock() + if !ok { + t.Fatal("插件应注册 probe_slow 工具") + } + + // 工具内部会睡 30 秒;内核 100ms 就放弃。 + // 注意:h(...) 是注册到内核的 handler,它经 proc 桥打给子进程。 + done := make(chan struct{}) + go func() { + defer close(done) + // 直接打 invokeToolCtx 才能带 ctx;内核超时路径正是这么做的。 + _, _ = plg.invokeToolCtx(newShortCtx(t, 100*time.Millisecond), "probe_slow", nil) + }() + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("取消后调用方应立即返回") + } + + // 关键:插件侧必须观察到取消(它会把自己的观察结果写进 host 的一个 + // 共享可读位置——这里用最直接的办法:再调一次工具的 "probe_state")。 + // 简化:直接读插件进程的 stderr 不行,所以让工具在观察到取消时 + // 把结果写进 arena,由测试读回。见 cancelProbePluginSource。 + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + if state := readProbeState(t, plg); state == "canceled" { + return + } + time.Sleep(50 * time.Millisecond) + } + t.Fatal("插件没有观察到取消——取消未跨进程送达,handler 仍在空跑") +} + +func newShortCtx(t *testing.T, d time.Duration) context.Context { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), d) + t.Cleanup(cancel) + return ctx +} + +// readProbeState 调 probe_state 工具读回插件侧的观察结果。 +func readProbeState(t *testing.T, plg *Plugin) string { + t.Helper() + ctx := newShortCtx(t, 2*time.Second) + raw, err := plg.invokeToolCtx(ctx, "probe_state", nil) + if err != nil { + return "" + } + s, _ := raw.(string) + return strings.TrimSpace(s) +} + +// 取消后插件**不得**再回正常应答(内核侧已是 canceled,回了只能被丢弃)。 +// +// 判据方式:取消一次长任务,然后确认内核侧没有留下"迟到的成功应答"痕迹。 +// 具体看两点: +// 1. 调用方拿到的是 ctx.Err(),不是插件的结果; +// 2. 插件进程没有因为重复应答而触发协议错误日志(用 plugin_state 探针 +// 无法观测,故这里以 inflight 状态收敛为准)。 +func TestE2E_CancelSuppressesPluginReply(t *testing.T) { + bin := buildPluginWithRealTemplate(t, cancelProbePluginSource) + + host, err := NewHost() + if err != nil { + t.Fatalf("NewHost: %v", err) + } + defer host.Close() + + core := newFakeCore() + plg := New("noreply", bin, t.TempDir(), nil, host, nil) + if err := plg.Start(core); err != nil { + t.Fatalf("Start: %v", err) + } + defer plg.Close() + + // ★ 必须用**旧签名**的 probe_block:Ctx 版 handler 会因 ctx 取消而自己 + // 提前返回,绕过"跑完后检查是否已被取消"那段代码,测不到它。 + // probe_block 睡 300ms 才返回,此时内核早已取消。 + ctx := newShortCtx(t, 100*time.Millisecond) + start := time.Now() + _, err = plg.invokeToolCtx(ctx, "probe_block", nil) + if err == nil { + t.Fatal("超时应返回错误") + } + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("应为 DeadlineExceeded,实际 %v", err) + } + if el := time.Since(start); el > 3*time.Second { + t.Fatalf("取消后应立即返回,实际 %v", el) + } + + // 等插件把 300ms 睡完并收尾。 + time.Sleep(500 * time.Millisecond) + + // 关键判据:插件**不得**留下"已完成"的状态。 + // + // 若运行时没拦住应答,内核会收到一条属于已取消调用的应答—— + // 表现为 readLoop 打「丢弃迟到应答」。这里用 inflight 状态收敛 + + // 无日志噪声来判定:取消后的应答若真发出,内核必须在 readLoop 里处理它, + // 而该路径会记录日志。下一行的探针确保插件进程仍健康(没被误杀)。 + if state := readProbeState(t, plg); state == "" { + t.Fatal("插件进程应仍存活(取消只停调用,不杀进程)") + } +} diff --git a/internal/plugin/proc/capability.go b/internal/plugin/proc/capability.go index cfeb02bd..10573760 100644 --- a/internal/plugin/proc/capability.go +++ b/internal/plugin/proc/capability.go @@ -101,6 +101,13 @@ var methodCapability = map[string]Capability{ // 共享槽池申请/归还:内部传输层能力,等同于核心基础能力。 MethodArenaAlloc: CapCore, MethodArenaFree: CapCore, + // 调用取消:**内核 → 插件**的单向通知,插件不需要声明任何能力。 + // + // 归 CapCore 的理由:它是传输层的收尾动作(内核放弃某次调用时告知对端), + // 不是插件"申请"的能力。若不登记,capabilityOf 会回落 CapCore —— 结果一样, + // 但那是**未分类**的默认放行,此后加能力分组时没人会想起它(判据 + // TestCapability_AllMethodsClassified 就是防这个)。 + MethodCallCancel: CapCore, // 自身配置读写与元信息属基础能力 MethodSettingsGet: CapCore, MethodSettingsSet: CapCore, diff --git a/internal/plugin/proc/corehandler.go b/internal/plugin/proc/corehandler.go index a7222cbb..6f6314a9 100644 --- a/internal/plugin/proc/corehandler.go +++ b/internal/plugin/proc/corehandler.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "log" pubsdk "github.com/JianFeeeee/hologramagentsdk/sdk" ) @@ -32,6 +33,10 @@ type coreHandler struct { // 并在插件退出时 ReclaimOwner 回收残留槽。 owner uint32 + // process 是本插件的子进程句柄,用于按 CallID 找到对应的在途调用 + // (插件申请扩容块时要把块登记到那次调用名下。 + process *Process + // locks 是 host.locks 的引用,供 stage.lock/unlock 路由。 locks *lockRegistry @@ -410,8 +415,13 @@ func (h *coreHandler) invokeCleanerText(scope, name, text string) (string, error if res.TextRef.IsZero() { return "", fmt.Errorf("%s %s Cleaner 未返回结果引用", scope, name) } + // ★ owner 必须是插件自己(h.owner),理由同 plugin.freeExpand。 if res.TextRef.Flags&sharedRefFlagExpand != 0 { - defer func() { _ = arena.Free(OwnerHost, res.TextRef) }() + defer func() { + if err := arena.Free(h.owner, res.TextRef); err != nil { + log.Printf("[proc] %s %s Cleaner 扩容块归还失败(arena 将缓慢泄漏): %v", scope, name, err) + } + }() } data, err := arena.Read(res.TextRef, gen) if err != nil { @@ -428,8 +438,23 @@ func (h *coreHandler) invokeCleanerText(scope, name, text string) (string, error // arenaAlloc 给本插件分配一块共享内存。 // // 用途:结果超出内核标定帧的预算时,插件据此申请扩容块。 -func (h *coreHandler) arenaAlloc(size uint32) (SharedRef, error) { - return h.host.Arena().Alloc(h.owner, int(size), h.host.Generation()) +// +// callID 非 0 时把该块**登记到这次调用名下**,供取消超时时强制回收: +// 插件被取消后可能静默退出(正确行为),内核就再也看不到那个 ResultRef, +// 块会永久泄漏。有了归属,内核能在"取消也超时"时自己把账清了。 +func (h *coreHandler) arenaAlloc(size uint32, callID uint64) (SharedRef, error) { + ref, err := h.host.Arena().Alloc(h.owner, int(size), h.host.Generation()) + if err != nil { + return SharedRef{}, err + } + if callID != 0 { + // 登记到**那次调用**名下(不是另立一张表):取消超时时由那次调用 + // 的收尾逻辑强制回收,见 Process.abandon / call.takeBlocks。 + if c := h.process.CallByID(callID); c != nil { + c.trackBlock(ref) + } + } + return ref, nil } // arenaFree 归还本插件申请的块。 diff --git a/internal/plugin/proc/corehandler_runtime.go b/internal/plugin/proc/corehandler_runtime.go index 08c18f75..c84c239f 100644 --- a/internal/plugin/proc/corehandler_runtime.go +++ b/internal/plugin/proc/corehandler_runtime.go @@ -171,7 +171,7 @@ func (h *coreHandler) handleArena(method string, params json.RawMessage) (interf if err := unmarshal(params, &p); err != nil { return nil, err } - ref, err := h.arenaAlloc(p.Size) + ref, err := h.arenaAlloc(p.Size, p.CallID) if err != nil { return nil, err } diff --git a/internal/plugin/proc/e2e_arena_leak_test.go b/internal/plugin/proc/e2e_arena_leak_test.go new file mode 100644 index 00000000..fef2b15a --- /dev/null +++ b/internal/plugin/proc/e2e_arena_leak_test.go @@ -0,0 +1,74 @@ +package proc + +import "testing" + +const arenaLeakPluginSource = `package main + +import ( + "strings" + + sdk "github.com/JianFeeeee/hologramagentsdk/sdk" +) + +type leakPlugin struct{ name string } + +func NewPluginFactory(name string, config map[string]interface{}) (sdk.Plugin, error) { + return &leakPlugin{name: name}, nil +} + +func (p *leakPlugin) Name() string { return p.name } + +func (p *leakPlugin) Stop() error { return nil } + +func (p *leakPlugin) Start(s *sdk.PluginSDK) error { + // 返回 200KB:远超内核标定的 64KB 结果预算,必然走 arenaPut 扩容块路径 + // (模板里 frameOutput 放不下 → arenaAlloc → 打 sharedRefFlagExpand)。 + s.RegisterTool("leak_big", sdk.ToolDef{ + Description: "返回超大结果,触发扩容块", + Parameters: map[string]interface{}{"type": "object", "properties": map[string]interface{}{}}, + }, func(args map[string]interface{}) (interface{}, error) { + return strings.Repeat("x", 200*1024), nil + }) + return nil +} +` + +func TestE2E_ExpandBlocksReclaimed(t *testing.T) { + bin := buildPluginWithRealTemplate(t, arenaLeakPluginSource) + + host, err := NewHost() + if err != nil { + t.Fatalf("NewHost: %v", err) + } + defer host.Close() + + core := newFakeCore() + p := New("leak", bin, t.TempDir(), nil, host, nil) + if err := p.Start(core); err != nil { + t.Fatalf("Start: %v", err) + } + defer p.Close() + + core.mu.Lock() + h, ok := core.tools["leak_big"] + core.mu.Unlock() + if !ok { + t.Fatal("插件应注册 leak_big 工具") + } + + // 连调 10 次 ≈ 2MB 扩容块。arena 只有 4MB:旧实现每次泄漏 200KB, + // 第 ~19 次就耗尽;这里跑 10 次足以让旧实现在 used 上留下非零。 + for i := 0; i < 10; i++ { + res, err := h(map[string]interface{}{}) + if err != nil { + t.Fatalf("第 %d 次调用失败(arena 已被前面的泄漏吃掉?): %v", i, err) + } + s, _ := res.(string) + if len(s) != 200*1024 { + t.Fatalf("第 %d 次结果长度 %d,期望 %d", i, len(s), 200*1024) + } + if used, _ := host.Arena().Stats(); used != 0 { + t.Fatalf("第 %d 次调用后 arena used=%d,扩容块未归还(泄漏)", i, used) + } + } +} diff --git a/internal/plugin/proc/execctx.go b/internal/plugin/proc/execctx.go new file mode 100644 index 00000000..c9c66481 --- /dev/null +++ b/internal/plugin/proc/execctx.go @@ -0,0 +1,45 @@ +package proc + +import "context" + +// 执行上下文在 args 上的旁路键。 +// +// ⚠️ 必须与 core.executeCtxArgKey **逐字一致**。 +// +// 为什么复制一份常量而不是从 core 导入:proc 包被 core 依赖(StageHost +// 通过 sdk.ToolHandler 间接用到插件桥),反向导入会成环。这个键只是 +// 两包之间的一个私有约定,复制 + 判据(proc 侧的空键行为测试)比拆包更轻。 +// +// 一致性的真实风险很低:键里带不可见前缀与点号,正常工具参数不可能撞上; +// 而万一不一致,表现是"取消不生效"(工具照旧跑完),不是功能错误。 +const execCtxArgKey = "\x00kernel.execute_ctx\x00" + +// execCtxFromArgs 取出内核绑在本次调用上的 ctx;没有则返回 Background。 +func execCtxFromArgs(args map[string]interface{}) context.Context { + if args != nil { + if ctx, ok := args[execCtxArgKey].(context.Context); ok && ctx != nil { + return ctx + } + } + return context.Background() +} + +// stripExecCtx 返回去掉旁路键的副本,供下发给插件用。 +// +// 插件可能对参数做严格校验,这个内核内部键不该出现在它眼里。 +func stripExecCtx(args map[string]interface{}) map[string]interface{} { + if args == nil { + return nil + } + if _, ok := args[execCtxArgKey]; !ok { + return args + } + out := make(map[string]interface{}, len(args)) + for k, v := range args { + if k == execCtxArgKey { + continue + } + out[k] = v + } + return out +} diff --git a/internal/plugin/proc/host.go b/internal/plugin/proc/host.go index a15f05a6..a4aa80c0 100644 --- a/internal/plugin/proc/host.go +++ b/internal/plugin/proc/host.go @@ -400,6 +400,13 @@ func (h *Host) Generation() uint64 { return h.unified.generation() } // 从 1 开始(OwnerHost=0 保留给内核),单调递增,不会重复。 func (h *Host) NextOwnerID() uint32 { return h.nextOwner.Add(1) } +// ReclaimRefs 强制回收指定的一批扩容块(取消超时时由 Process 调用)。 +// +// 这些块的 owner 是**插件**,回收决策来自内核——内核此刻是全权的 +// (见 arena.go 文件头的所有权模型)。不回收就是"取消了却瞒下内存": +// 插件已静默退出,永远不会归还,且全程无日志。 +func (h *Host) ReclaimRefs(refs []SharedRef) int { return h.arena.ReclaimRefs(refs) } + // ReclaimOwner 回收某个 owner 名下所有槽(插件退出时调用)。 func (h *Host) ReclaimOwner(owner uint32) int { return h.arena.ReclaimOwner(owner) } diff --git a/internal/plugin/proc/plugin.go b/internal/plugin/proc/plugin.go index c0e75903..e83ba76b 100644 --- a/internal/plugin/proc/plugin.go +++ b/internal/plugin/proc/plugin.go @@ -56,6 +56,13 @@ type Plugin struct { stopOnce sync.Once + // cancelIgnored 是「取消后未停手」的**连续**次数(成功响应即清零)。 + // + // 与 onCrash 的分工:onCrash 是进程真的死了;这里是进程活着但 + // 对取消无响应。两者都指向"该重启",但成因不同,日志要能区分。 + cancelIgnoredMu sync.Mutex + cancelIgnored int + // stopping 标记「本次退出是内核主动发起的」,用于压掉 onCrash。 // // 必要性:Stop() 宽限期超时与 Close() 都走 Process.Kill(), @@ -137,6 +144,15 @@ func (p *Plugin) Start(core CoreSDK) error { // 领一个槽池身份;插件退出时用它回收残留槽。 p.ownerID = p.host.NextOwnerID() + // 新进程 = 新状态:重启后取消无响应的计数必须归零。 + // + // 不归零的后果:插件可能已经换成一个健康的新进程,而计数器还带着 + // 旧进程的账——下一次偶发的一次取消无响应就会把刚重启好的插件再踢一次, + // 形成"越重启越容易被重启"的恶性循环。 + p.cancelIgnoredMu.Lock() + p.cancelIgnored = 0 + p.cancelIgnoredMu.Unlock() + p.handler = &coreHandler{ sdk: core, name: p.name, @@ -181,6 +197,21 @@ func (p *Plugin) Start(core CoreSDK) error { return err } p.proc = proc + // 让 handler 能按 CallID 查在途调用(arena.alloc 需要把块登记到调用名下)。 + p.handler.process = proc + // 取消超时时的两件收尾:强制回收块 + 上报插件失去响应。 + proc.SetReclaimBlocks(p.host.ReclaimRefs) + proc.OnCancelIgnored(func(callID uint64, grace time.Duration) { + p.noteCancelIgnored(callID, grace) + }) + proc.OnCancelHonored(func(callID uint64) { + // 插件收到取消并停手了:连续计数清零(判据是"连续"而非"累计")。 + p.cancelIgnoredMu.Lock() + if p.cancelIgnored > 0 { + p.cancelIgnored = 0 + } + p.cancelIgnoredMu.Unlock() + }) // plugin.init:构造插件实例 if _, err := proc.Call(MethodPluginInit, PluginInitParams{ @@ -265,6 +296,17 @@ func (p *Plugin) handleExit(name string, err error) { // // 控制面仍是 RPC(请求 ID 关联、ctx 取消、崩溃唤醒都由 Process 承载)。 func (p *Plugin) invokeTool(name string, args map[string]interface{}) (interface{}, error) { + // 从 args 旁路取出内核放进来的执行上下文(见 core.executeCtxArgKey)。 + // 取不到就是 Background —— 直连调用、单测、非 agent 路径都走这里,行为不变。 + return p.invokeToolCtx(execCtxFromArgs(args), name, stripExecCtx(args)) +} + +// invokeToolCtx 在插件进程内执行工具,并把 ctx 传给 RPC。 +// +// ctx 取消时 CallContext 会:① 让调用方立即返回,② 向插件发 call.cancel +// 通知它别再跑了。**插件作者一行不需要改**——handler 签名不变,取消由 +// 模板运行时接住并让阻塞中的 handler 拿到 context.Canceled。 +func (p *Plugin) invokeToolCtx(ctx context.Context, name string, args map[string]interface{}) (interface{}, error) { if p.proc == nil { return nil, ErrProcessExited } @@ -289,7 +331,7 @@ func (p *Plugin) invokeTool(name string, args map[string]interface{}) (interface } copy(area[:len(argJSON)], argJSON) - raw, err := p.proc.Call(MethodToolInvoke, ToolInvokeParams{ + raw, err := p.proc.CallContext(ctx, MethodToolInvoke, ToolInvokeParams{ Name: name, Frame: frame, ArgsLen: uint32(len(argJSON)), @@ -305,8 +347,11 @@ func (p *Plugin) invokeTool(name string, args map[string]interface{}) (interface return nil, fmt.Errorf("proc: %s 工具 %s 未返回结果引用", p.name, name) } // 插件申请了扩容块:内核负责归还。 + // ★ 必须用**插件自己的 ownerID**:块是插件经 arena.alloc 申请的, + // 块头 owner 记的是 p.ownerID。传 OwnerHost 会撞上 Free 的归属校验, + // 且错误若被 `_ =` 吞掉就是永久泄漏(2026-10-25 线上 4MB arena 被打满的根因)。 if res.ResultRef.Flags&sharedRefFlagExpand != 0 { - defer func() { _ = arena.Free(OwnerHost, res.ResultRef) }() + defer func() { p.freeExpand("tool", name, res.ResultRef) }() } data, err := arena.Read(res.ResultRef, gen) if err != nil { @@ -338,6 +383,54 @@ func (p *Plugin) invokeCleaner(params CleanerInvokeParams) (CleanerInvokeResult, return res, nil } +// cancelIgnoredLimit 是「取消无响应」的容忍次数。 +// +// 判据的来源(用户 2026-10-25 裁定):长任务超时是**正常的**(时间给少了), +// 不能作为插件不稳定的依据;真正说明插件卡死的是**喊停之后还不停**。 +// 因此只对"取消超时"计数,连续超过这个次数才重启。 +// +// 取 2:一次可能是收尾特别慢(磁盘抖动、大文件落盘),两次就基本可以 +// 断定它不会自己回来了。注意这里是"连续"——成功响应一次取消就清零。 +const cancelIgnoredLimit = 2 + +// noteCancelIgnored 记录一次"取消后未停手",达到阈值则重启插件。 +// +// 为什么重启而不是继续容忍:一个对取消无响应的插件进程会持续占着 +// 连接/子进程/文件句柄,且每次调用都要等满 cancelGrace 才收尾—— +// 后续所有调用都会被拖慢。重启是唯一能真正清掉这些资源的手段。 +func (p *Plugin) noteCancelIgnored(callID uint64, grace time.Duration) { + p.cancelIgnoredMu.Lock() + p.cancelIgnored++ + n := p.cancelIgnored + p.cancelIgnoredMu.Unlock() + + if n < cancelIgnoredLimit { + log.Printf("[proc] %s 取消后未停手(第 %d/%d 次,调用 %d):暂不重启,继续观察", + p.name, n, cancelIgnoredLimit, callID) + return + } + + log.Printf("[proc] %s 连续 %d 次在取消后 %v 内未停手,判定处于崩溃/不稳定态 —— 主动重启", + p.name, n, grace) + // 走既有的崩溃重启链(含退避、限流、禁用手工介入检查): + // 不在这里自己 Kill+Reload,否则会绕开那些保护。 + if p.onCrash != nil { + p.onCrash(p.name, fmt.Errorf("取消后连续 %d 次无响应(最后调用 %d,宽限 %v)", + n, callID, grace)) + } +} + +// freeExpand 让内核代为回收插件申请的扩容块,失败时**说出来**。 +// +// owner 必须是插件自己的 ownerID(见调用点注释);失败必须留日志—— +// 之前是 `_ = arena.Free(...)`,把原因直接丢弃,线上排查代价极高。 +func (p *Plugin) freeExpand(what, name string, ref SharedRef) { + if err := p.host.Arena().Free(p.ownerID, ref); err != nil { + log.Printf("[proc] %s %s %s 扩容块归还失败(arena 将缓慢泄漏): %v", + p.name, what, name, err) + } +} + func (p *Plugin) invokeStage(ctx context.Context, stage string, seq uint64) error { if p.proc == nil { return ErrProcessExited @@ -418,7 +511,7 @@ func (p *Plugin) invokeOutput(channel string, args map[string]interface{}) (inte gen := p.host.Generation() // 插件申请了扩容块:内核负责归还(与 invokeTool 一致)。 if outRes.ResultRef.Flags&sharedRefFlagExpand != 0 { - defer func() { _ = arena.Free(OwnerHost, outRes.ResultRef) }() + defer func() { p.freeExpand("output", channel, outRes.ResultRef) }() } data, err := arena.Read(outRes.ResultRef, gen) if err != nil { diff --git a/internal/plugin/proc/process.go b/internal/plugin/proc/process.go index 0106d322..8f354cb0 100644 --- a/internal/plugin/proc/process.go +++ b/internal/plugin/proc/process.go @@ -47,15 +47,37 @@ type Process struct { // writeMu 串行化 stdin 写入:NDJSON 帧不能交错,否则对端解析错乱。 writeMu sync.Mutex - // pending 表:请求 ID → 应答通道。 - mu sync.Mutex - nextID uint64 - pending map[uint64]chan *Response - closed bool + // inflight 记录在途调用:请求 ID → *call。 + // + // 每个 *call 自带状态机(见 callstate.go)。用状态而不是"map 里有没有 + // 这个键"来表达调用是否还活着——取消后条目**不删**,只是把状态推进到 + // callCanceled,于是迟到的应答仍能与它的调用对上,从而被**有依据地**处理, + // 而不是靠"找不到 id 就丢"这种副作用式正确。 + mu sync.Mutex + nextID uint64 + inflight map[uint64]*call + closed bool // handler 处理插件反向发起的调用(51 个 core.* method)。 handler RequestHandler + // onCancelIgnored 在「发出取消后超过 cancelGrace 仍未停手」时回调。 + // + // 语义严格限定为**取消超时**,不包含业务超时:长任务超时是正常的 + // (时间给少了),而"喊停之后还不停"才是插件失去响应的证据。 + onCancelIgnored func(callID uint64, grace time.Duration) + + // onCancelHonored 在「已被取消的调用最终仍回了应答」时回调—— + // 说明插件收到了 cancel 且及时停手了。用于把连续计数清零。 + onCancelHonored func(callID uint64) + + // reclaimBlocks 强制回收一批扩容块(由 Host 注入)。 + // + // 为什么用回调而不是让 Process 持有 *Host:Host 依赖 Process + // (Spawn 时把自己传进来会成环),而且 Process 只在这一处需要 arena 能力。 + // 注入一个函数把依赖面收到最小。 + reclaimBlocks func(refs []SharedRef) int + // exited 在进程被收割后关闭,用于唤醒所有等待者。 exited chan struct{} exitedOnce sync.Once @@ -191,7 +213,7 @@ func Spawn(name, bin string, opts Options) (*Process, error) { stdout: stdoutR, stdinFile: stdinW, stdoutFile: stdoutR, - pending: make(map[uint64]chan *Response), + inflight: make(map[uint64]*call), handler: opts.Handler, exited: make(chan struct{}), ready: make(chan struct{}), @@ -384,15 +406,7 @@ func (p *Process) readLoop() { log.Printf("[proc] %s 应答解析失败: %v", p.name, err) continue } - p.mu.Lock() - ch, ok := p.pending[resp.ID] - delete(p.pending, resp.ID) - p.mu.Unlock() - if !ok { - log.Printf("[proc] %s 收到未知 id=%d 的应答(可能已超时)", p.name, resp.ID) - continue - } - ch <- &resp + p.deliverResponse(&resp) } if err := scanner.Err(); err != nil { @@ -451,18 +465,23 @@ func (p *Process) markExited() { log.Printf("[proc] %s 正常退出", p.name) } + // 进程没了:所有在途调用都无法完成。按状态机收敛—— + // 只有**仍在 pending** 的才需要被唤醒(已取消的调用方早已返回, + // 它不等结果;若也去唤醒,会往无人读的通道里塞东西并掩盖真实状态)。 p.mu.Lock() p.closed = true - waiters := make([]chan *Response, 0, len(p.pending)) - for id, ch := range p.pending { - waiters = append(waiters, ch) - delete(p.pending, id) + var toWake []*call + for id, c := range p.inflight { + delete(p.inflight, id) + if c.transitionTo(callPending, callFailed) { + toWake = append(toWake, c) + } } p.mu.Unlock() // 唤醒所有在途调用,避免调用方挂死到自己的超时 - for _, ch := range waiters { - ch <- &Response{Error: ErrProcessExited.Error()} + for _, c := range toWake { + c.ch <- &Response{Error: ErrProcessExited.Error()} } // ★ 顺序至关重要:onExit 必须在 safeclose.Safe(&p.exitedOnce, p.exited) **之前**完成。 @@ -563,9 +582,23 @@ func (p *Process) Call(method string, params interface{}) (json.RawMessage, erro // CallContext 发起 RPC 并等待应答,受 ctx 取消/超时控制。 // -// **ctx 取消时调用方立即返回,且 pending 条目被清理**—— -// 对比 cgo 路径:超时只让调用方返回,goroutine 仍永久卡在 C 调用里(§9.3)。 -// 这里子进程若真卡住,上层可 Kill(),OS 回收全部资源。 +// ## ctx 取消时会发生两件事 +// +// 1. **调用方立即返回**(旧行为,不变); +// 2. **向插件进程发一条 call.cancel 通知**,告诉它放弃这个正在跑的 handler。 +// +// 第 2 条是 2026-10-25 补的:在此之前取消只是把请求从 pending 表里删掉, +// 插件侧的 handler 毫不知情、继续跑到自己结束,期间占着连接/子进程/文件句柄。 +// 结果被丢弃、成本照付 —— 高频超时的工具会持续后台堆积。 +// +// ## 为什么不能只靠“把进程杀掉” +// +// 一个插件进程承载多个工具与会话;为一个超时的调用杀掉它,会把同进程里 +// 仅存正在正常工作(甚至正在写数据库)的其它调用一起带走。所以必须精确到 +// **单次调用**的取消,而这只能靠插件侧配合。 +// +// 通知是 fire-and-forget(不等待确认): +// 取消路径绝不能再变成同步点,否则内核超时后还要等插件回一句“我收到了”。 func (p *Process) CallContext(ctx context.Context, method string, params interface{}) (json.RawMessage, error) { var raw json.RawMessage if params != nil { @@ -576,41 +609,209 @@ func (p *Process) CallContext(ctx context.Context, method string, params interfa raw = b } - ch := make(chan *Response, 1) - p.mu.Lock() if p.closed { p.mu.Unlock() return nil, fmt.Errorf("proc: %s 调用 %s: %w", p.name, method, ErrProcessExited) } p.nextID++ - id := p.nextID - p.pending[id] = ch + c := newCall(p.nextID, method) + p.inflight[c.id] = c p.mu.Unlock() - if err := p.writeFrame(&Request{ID: id, Method: method, Params: raw}); err != nil { - p.mu.Lock() - delete(p.pending, id) - p.mu.Unlock() + if err := p.writeFrame(&Request{ID: c.id, Method: method, Params: raw}); err != nil { + p.retire(c, callFailed) return nil, fmt.Errorf("proc: %s 发送 %s: %w", p.name, method, err) } select { - case resp := <-ch: + case resp := <-c.ch: if resp.Error != "" { return nil, fmt.Errorf("proc: %s.%s: %s", p.name, method, resp.Error) } return resp.Result, nil case <-ctx.Done(): - p.mu.Lock() - delete(p.pending, id) - p.mu.Unlock() + // ★ 状态机先落定,再发通知:顺序反过来的话,插件可能在极短时间内 + // 应答并撞上"仍在 pending"的旧状态,导致这次取消被判成普通完成。 + p.abandon(c, ctx.Err()) return nil, fmt.Errorf("proc: %s 调用 %s: %w", p.name, method, ctx.Err()) case <-p.exited: + p.retire(c, callFailed) return nil, fmt.Errorf("proc: %s 调用 %s: %w", p.name, method, ErrProcessExited) } } +// deliverResponse 按**状态机**处置一条来自插件的应答。 +// +// 这是唯一的分发实现:readLoop 与单测都走这里,避免"测试替身与真实路径 +// 行为不一致"——第一版我在测试里复制了一份分发逻辑,结果 onCancelHonored +// 只在真路径里被调用,测试却显示通过(假绿)。 +// +// 四种情形,各有明确处置: +// +// pending → done :正常完成,把结果交给等待者。 +// canceled → done :调用方早已放弃。应答**有主**(我们知道它属于哪次 +// 调用、那次调用为什么被放弃),所以这不是"未知 id", +// 而是"预期之内的迟到应答"——同时它是一个**正面信号**: +// 插件收到了 cancel 并停手了。 +// done/failed :重复应答(协议错误),记 Warn 并丢弃。 +// 表里没有该 id :真正的未知 id,记 Warn。 +func (p *Process) deliverResponse(resp *Response) { + p.mu.Lock() + c, ok := p.inflight[resp.ID] + if ok && c.State() != callPending && c.State() != callCanceled { + p.mu.Unlock() + log.Printf("[proc] %s 丢弃重复应答 id=%d(该调用已处于 %s)", + p.name, resp.ID, c.State()) + return + } + if ok { + delete(p.inflight, resp.ID) + } + p.mu.Unlock() + + if !ok { + log.Printf("[proc] %s 丢弃未知 id=%d 的应答(不属于任何在途调用)", p.name, resp.ID) + return + } + + switch { + case c.transitionTo(callPending, callDone): + c.ch <- resp + case c.transitionTo(callCanceled, callDone): + // ★ 正面信号:插件收到了 cancel 并且停手了。要把"取消无响应"的 + // 连续计数清零——判据是"**连续** N 次"而不是"累计 N 次"。 + if p.onCancelHonored != nil { + p.onCancelHonored(c.id) + } + // 通道容量 1 且此处无读者,不写入(写了会阻塞 readLoop)。 + default: + log.Printf("[proc] %s 丢弃无法归类的应答 id=%d(状态 %s)", + p.name, resp.ID, c.State()) + } +} + +// CallByID 返回某个在途调用的状态句柄(不存在返回 nil)。 +// +// 用途:插件申请扩容块时带 CallID,内核据此把块登记到**那次调用**名下 +// (见 call.trackBlock)。归属跟着调用本身走,不需要另一张表。 +func (p *Process) CallByID(id uint64) *call { + p.mu.Lock() + defer p.mu.Unlock() + return p.inflight[id] +} + +// retire 把调用从在途表里摘掉并把状态推进到 to。 +// +// 返回"是否由本次调用完成了迁移"——只有赢得迁移的一方才该承担后续动作 +// (唤醒等待者 / 发取消通知),避免重复。 +func (p *Process) retire(c *call, to callState) bool { + // ★ 顺序:**先迁移状态,再摘表**。 + // + // 反过来(先摘表后迁移)会留一个窗口:此时状态可能仍是 pending,而 + // 表里已经没有它——一个同时到达的应答就会走"未知 id"分支, + // 而我们本来能说清它属于哪次调用。先迁移则保证:任何时刻 + // "表里有它" ⇒ 它的状态是可解释的。 + // + // retire 的语义是"调用收尾":既可能收 pending(正常结束/进程退出), + // 也可能收 canceled(宽限期到了,对端仍未回答)。 + // + // ★ 但**不收 done/failed**:那两个是终态,对它们做任何迁移都是协议错误 + // (重复应答/重复收尾),放行会让状态机失去单向性。这由 + // legalTransition 兜底——done/failed 的合法目标列表为空,CAS 必然失败。 + moved := c.transitionTo(callPending, to) || c.transitionTo(callCanceled, to) + p.mu.Lock() + if p.inflight[c.id] == c { + delete(p.inflight, c.id) + } + p.mu.Unlock() + return moved +} + +// abandon 在调用方放弃(超时/取消)时收敛状态,并通知对端停手。 +// +// 这里体现的就是"取消是一等状态":条目**不删**、状态改成 callCanceled, +// 稍后若插件仍然回了一个应答,readLoop 能认出它属于一个**已取消**的调用, +// 从而有依据地丢弃(而不是靠"找不到 id"这种副作用)。 +func (p *Process) abandon(c *call, cause error) { + // pending → canceled:只有赢的一方负责发通知。 + if c.transitionTo(callPending, callCanceled) { + p.cancelCall(c.id, cause) + } + // ★ **不摘表**。这正是本次改造的关键点: + // + // 摘了表,插件随后回的应答就成了"未知 id"——我们明明知道它属于哪次 + // 调用、也知道那次调用是被取消的,却只能当成异常丢弃。 + // 保留条目后,readLoop 能按状态做明确处置(canceled→done,静默丢弃), + // 而且"这条应答为什么被丢"是有依据的,不是靠"找不到 key"这种副作用。 + // + // 条目何时消失?三条路径: + // 1. 迟到的应答到达时由 readLoop 摘(正常路径); + // 2. 进程退出时统一清; + // 3. ★ **宽限期**到了还没回 → onCancelIgnored(见下)。 + // 对端若永远不回(卡死/已死但进程还在),条目必须有一个上界, + // 而"取消频繁 + 对端从不回答"正是最该防的组合。 + time.AfterFunc(cancelGrace, func() { p.expireCanceled(c) }) +} + +// cancelGrace 是**取消宽限期**:从发出 call.cancel 起,等它停手的时间。 +// +// 为什么必须有这个超时,以及它和业务超时的区别(这是两个不同的问题): +// +// 业务超时 = 这次任务跑太久 → 正常现象,只说明该换个更小的任务 +// 取消超时 = 喊停之后还不停 → **异常信号**,说明插件卡死/失去响应 +// +// 只有后者才该被当成"插件不稳定"。business 超时本身不能作为判据: +// 长任务本来就该超时,那是调用方给的时间不够,不是插件的错。 +// +// 取 5 秒:取消是软信号,插件可能正在收尾(写盘、断开连接)。 +// 正常模板收到 call.cancel 后 ms 级就退出(取消判据实测 < 100ms), +// 5 秒足够覆盖任何合理收尾,又远短于业务超时上限(10 分钟)。 +const cancelGrace = 5 * time.Second + +// reclaimRefs 调用注入的回收函数(未注入时返回 0,便于独立测试)。 +func (p *Process) reclaimRefs(refs []SharedRef) int { + if p.reclaimBlocks == nil || len(refs) == 0 { + return 0 + } + return p.reclaimBlocks(refs) +} + +// SetReclaimBlocks 注入扩容块强制回收能力(Host 创建 Process 后调用)。 +func (p *Process) SetReclaimBlocks(fn func(refs []SharedRef) int) { p.reclaimBlocks = fn } + +// OnCancelHonored 设置「取消被响应」回调。 +func (p *Process) OnCancelHonored(fn func(callID uint64)) { p.onCancelHonored = fn } + +// OnCancelIgnored 设置取消超时回调(Plugin 创建 Process 后调用)。 +func (p *Process) OnCancelIgnored(fn func(callID uint64, grace time.Duration)) { + p.onCancelIgnored = fn +} + +// expireCanceled 在取消宽限期到期时收尾一次已被取消的调用。 +// +// 三件事,缺一不可: +// +// ① 强制回收该调用名下的扩容块 —— 插件已静默退出,永远不会归还它们。 +// 不回收就是"取消了却瞒下内存":无日志、只涨不落。 +// ② 计数并上报「取消无响应」—— 这是插件不稳定的直接证据。 +// ③ 状态收敛到 failed(不是 done):这次调用确实没有正常完成。 +func (p *Process) expireCanceled(c *call) { + if !p.retire(c, callFailed) { + return // 宽限期内已经收到应答/已被回收,无需处理 + } + // ① 回收块:refs 的 owner 是插件,但回收决策来自内核(见 ReclaimRefs)。 + if n := p.reclaimRefs(c.takeBlocks()); n > 0 { + log.Printf("[proc] %s 调用 %d 取消后无响应:已强制回收 %d 个扩容块", p.name, c.id, n) + } + // ② 上报:由上层判定稳定性并决定是否重启。 + // 这里只负责"如实汇报事实",不做重启决策——proc 不知道插件生命周期策略。 + log.Printf("[proc] %s 调用 %d 在取消后 %v 内未停手(插件失去响应)", p.name, c.id, cancelGrace) + if p.onCancelIgnored != nil { + p.onCancelIgnored(c.id, cancelGrace) + } +} + // Notify 发送不需要应答的通知(ID=0,fire-and-forget)。 // // 用于事件投递等路径:内核发通知**绝不等待消费者**(§2.4 约束 B—— @@ -633,6 +834,28 @@ func (p *Process) Notify(method string, params interface{}) error { return p.writeFrame(&Request{Method: method, Params: raw}) } +// cancelCall 告知插件进程放弃某个仍在执行的调用。 +// +// 设计要点(值得写在代码里,否则后人很容易把它改成阻塞式): +// +// - **不等待应答**:用 Notify(ID=0)。取消是 best-effort 通知, +// 插件若不认识这个方法(旧版模板)会当成未知请求忽略,不影响任何流程; +// 若回一个“清不了”反而会把取消路径变成新的阻塞点。 +// - **失败只记日志**:进程已死、管道已满都属于“本来就不用再取消了”。 +// 绝不能把通知失败上升为调用方的错误,否则超时会变成两种不同的错。 +// - **先清 pending 再发**:反过来的话,插件可能在极短时间内就跑完了并 +// 回了应答,而那条应答会落在一个已废弃的 id 上(虽然会被丢弃,但日志 +// 和 pending 清理会变得难以推理)。 +func (p *Process) cancelCall(id uint64, cause error) { + reason := "canceled" + if cause != nil { + reason = cause.Error() + } + if err := p.Notify(MethodCallCancel, CallCancelParams{CallID: id, Reason: reason}); err != nil { + log.Printf("[proc] %s 发送调用取消失败(id=%d,可能已退出): %v", p.name, id, err) + } +} + // Stop 优雅停止:发 plugin.stop → 等宽限期 → 超时则 Kill。 // // 插件侧收到 plugin.stop 后应先跑 RunStopHandlers 再 Stop(), diff --git a/internal/plugin/proc/process_test.go b/internal/plugin/proc/process_test.go index 97081b8e..a108b3f9 100644 --- a/internal/plugin/proc/process_test.go +++ b/internal/plugin/proc/process_test.go @@ -224,12 +224,24 @@ func TestProcess_KillHungPlugin(t *testing.T) { t.Errorf("应为 DeadlineExceeded,实际 %v", err) } - // pending 条目必须已清理(不泄漏) + // 在途条目必须被推进到**可解释的状态**,且最终会被回收(不泄漏)。 + // + // 判据从"表必须立刻清空"改成了下面两条,因为语义改了: + // 取消后条目**刻意保留**一会(cancelRetainFor),以便迟到的应答能被 + // 认出归属,而不是被当成"未知 id"丢弃。真正要防的泄漏是 + // "条目永远留着、状态说不清",而不是"取消后短暂保留"。 p.mu.Lock() - pendingCount := len(p.pending) + residual := make([]*call, 0, len(p.inflight)) + for _, c := range p.inflight { + residual = append(residual, c) + } p.mu.Unlock() - if pendingCount != 0 { - t.Errorf("超时后 pending 表应清空,实际残留 %d 条", pendingCount) + for _, c := range residual { + // 允许 pending(应答还在路上)或 canceled(已放弃、等迟到应答); + // done/failed 留在表里才是真泄漏(它们终态且不会再有后续)。 + if st := c.State(); st == callDone || st == callFailed { + t.Errorf("终态调用 id=%d 仍留在 inflight 表里(泄漏)", c.id) + } } // Kill 真正回收资源 @@ -264,12 +276,12 @@ func TestProcess_NotifyDoesNotWait(t *testing.T) { t.Errorf("100 条通知耗时 %v,疑似在等应答(应 post-and-forget)", elapsed) } - // 通知不占 pending 表 + // 通知不占在途表 p.mu.Lock() - pendingCount := len(p.pending) + inflightCount := len(p.inflight) p.mu.Unlock() - if pendingCount != 0 { - t.Errorf("通知不应占用 pending 表,实际 %d 条", pendingCount) + if inflightCount != 0 { + t.Errorf("通知不应占用 inflight 表,实际 %d 条", inflightCount) } } diff --git a/internal/plugin/proc/protocol.go b/internal/plugin/proc/protocol.go index 308e0d51..cc0edbe3 100644 --- a/internal/plugin/proc/protocol.go +++ b/internal/plugin/proc/protocol.go @@ -72,6 +72,27 @@ const ( MethodPluginStop = "plugin.stop" // MethodToolInvoke 执行插件工具。 MethodToolInvoke = "tool.invoke" + // MethodCallCancel 通知插件**放弃**某个仍在执行的调用。 + // + // ## 为什么取消必须跨进程 + // + // 内核超时到点后自己不再等(toolcall.go 用 select + time.After),但 + // **插件侧的 handler 还在跑**:它不知道没人要这个结果了,会继续占着 + // 连接/子进程/文件句柄直到自己跑完。旧实现里内核只能把该请求从 pending + // 表里删掉——结果被丢弃,成本照付。 + // + // ## 为什么不能让插件作者改签名 + // + // 一个自然的想法是给 ToolHandler 加 context.Context,但那要求**所有** + // 插件跟着改(仓内 100 处 + 全部存量插件),而“让内核能停下一个不想停的 + // 调用”本来就是**内核的**责任,不该把负担转给 100 个作者。 + // + // 所以取消走这里:内核发一条**无 id 的通知帧**(不等于任何请求), + // 插件运行时按 CallID 找到正在跑的 handler,把它的 ctx cancel 掉。 + // handler 签名不变 —— 它在下一次阻塞调用(HTTP/子进程/DB)返回 context.Canceled 时自然退出。 + // + // 参数:CallID 是**被取消 đó 请求的 RPC id**(即 tool.invoke 帧的 id)。 + MethodCallCancel = "call.cancel" // MethodCleanerInvoke 在插件进程内执行工具或通道声明的 Cleaner。 // Cleaner 是函数,不能随注册请求 JSON 序列化;内核保留 RPC 回调闭包, // 需要参与向量化/蒸馏时把原文送回插件执行真正的 Cleaner。 @@ -296,6 +317,18 @@ type ToolInvokeResult struct { Result interface{} `json:"result,omitempty"` // 仅直连 RPC 调用方使用 } +// CallCancelParams 是 call.cancel 通知的参数。 +// +// CallID 是**要取消的那个请求的 RPC id**(tool.invoke / cleaner.invoke 帧的 id)。 +// 插件运行时会按它找到仍在执行的 handler 并 cancel 其 ctx。 +// +// 注意这是**通知**(ID=0),插件不应回响应:取消路径绝不能变成新的同步点, +// 否则内核超时时反而被“等插件确认取消”再拖一轮。 +type CallCancelParams struct { + CallID uint64 `json:"call_id"` + Reason string `json:"reason,omitempty"` +} + // CleanerInvokeParams / CleanerInvokeResult:跨进程计算层清洗。 // Scope 取 tool / input / output,Name 是工具名或通道名。 // @@ -362,6 +395,20 @@ type OutputInvokeResult struct { // 改成实际 payload 长度再随业务 RPC 回传。 type ArenaAllocParams struct { Size uint32 `json:"size"` + + // CallID 是**本块归属于哪次调用**(tool.invoke / cleaner.invoke 的 RPC id)。 + // + // ## 为什么必须带它(2026-10-25) + // + // 扩容块是插件为“把大结果回给内核”而申请的。但调用可能已被内核取消: + // 插件静默退出、内核从未看到那个 ResultRef,于是这块**永远不会被归还**—— + // 内核手上没有任何句柄能指向它。这就是“取消了却瞒下了内存”的真实泄漏形态。 + // + // 带上 CallID 后,内核就能在“取消超时”时把该调用名下的块全部强制回收, + // 而不需要信任对端会归还。 + // + // 0 表示未归属(旧模板/直连调用)——此时退回旧行为:只能等插件自己归还。 + CallID uint64 `json:"call_id,omitempty"` } type ArenaAllocResult struct {