mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-11-01 06:47:29 +00:00
fix(proc): arena 扩容块泄漏 + 取消成为一等状态
## 1. arena 泄漏（线上 arena 被打满的根因）
插件经 arena.alloc 申请的扩容块，块头 owner 记的是**插件的** ownerID，
而内核三处归还全写着 Free(OwnerHost, ref)。归属校验必然失败，错误又被
`_ =` 吞掉 ⇒ 每返回一次大结果就永久泄漏一块。
线上表现（2026-10-25）：重任务跑几轮后 4MB arena 满，之后连 64KB 调用帧
都分不出来，所有插件工具一起报「arena: 空间不足（请求 65779 字节）」。
修法：三处改用插件自己的 ownerID，且失败必须留日志（吞掉错误等于把
下一次「空间不足」的原因提前销毁）。
## 2. 取消成为一等状态
旧实现里「调用在进行中」只表达为 pending 表里有这个键：取消时直接
delete，状态的痕迹全部消失，插件随后回的应答只能走「未知 id，丢弃」。
那是副作用的正确，不是规则的正确。
改成每次调用带显式状态机（callstate.go）：
pending ──应答──→ done
├──取消/超时──→ canceled ──迟到应答──→ done
└──进程退出──→ failed
迁移表集中在 legalTransition，非法组合一律拒绝（CAS 只校验 from，
不校验 to——实测曾把 done 改回 failed）。
## 3. 取消跨进程送达
内核超时到点后只停止「等」，插件侧的 handler 还在跑，占着连接/子进程/
文件句柄，结果却已被丢弃。现在 CallContext 取消时向插件发 call.cancel
通知（fire-and-forget），插件运行时按 **RPC id** 定位并 cancel 对应调用的 ctx。
不按工具名定位：同一工具可并发多次，按名字会误伤其它实例。
插件作者零改动：handler 签名不变，取消由模板运行时接住。
## 4. 取消超时判定插件不稳定
严格区分两类超时（用户裁定）：
业务超时 = 任务跑太久 → 正常，不作为不稳定依据
取消超时 = 喊停之后还不停 → 异常，是插件失去响应的直接证据
取消宽限（cancelGrace=5s）到期仍无应答 ⇒ ① 强制回收该调用名下的扩容块
（插件已静默退出，永远不会归还）② 计数 ③ 连续两次触发既有崩溃重启链。
插件一旦响应取消即清零；重启后也清零（判据是"连续"不是"累计"）。
判据：8 条新增单测 + 全部变异验证（不回收/不上报/不清零/取消不跨进程
任一处改坏，对应用例立刻变红）。
Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
JianFeeeee
@ -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
|
||||
|
||||
69
internal/plugin/proc/arena_leak_test.go
Normal file
69
internal/plugin/proc/arena_leak_test.go
Normal file
@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
165
internal/plugin/proc/callstate.go
Normal file
165
internal/plugin/proc/callstate.go
Normal file
@ -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()) }
|
||||
200
internal/plugin/proc/callstate_test.go
Normal file
200
internal/plugin/proc/callstate_test.go
Normal file
@ -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
|
||||
}
|
||||
156
internal/plugin/proc/cancel_ignored_test.go
Normal file
156
internal/plugin/proc/cancel_ignored_test.go
Normal file
@ -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("清零后第一次不应再次重启（判据是连续）")
|
||||
}
|
||||
}
|
||||
84
internal/plugin/proc/cancel_probe_test.go
Normal file
84
internal/plugin/proc/cancel_probe_test.go
Normal file
@ -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
|
||||
}
|
||||
`
|
||||
183
internal/plugin/proc/cancel_test.go
Normal file
183
internal/plugin/proc/cancel_test.go
Normal file
@ -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("插件进程应仍存活（取消只停调用，不杀进程）")
|
||||
}
|
||||
}
|
||||
@ -101,6 +101,13 @@ var methodCapability = map[string]Capability{
|
||||
// 共享槽池申请/归还：内部传输层能力，等同于核心基础能力。
|
||||
MethodArenaAlloc: CapCore,
|
||||
MethodArenaFree: CapCore,
|
||||
// 调用取消：**内核 → 插件**的单向通知，插件不需要声明任何能力。
|
||||
//
|
||||
// 归 CapCore 的理由：它是传输层的收尾动作（内核放弃某次调用时告知对端），
|
||||
// 不是插件"申请"的能力。若不登记，capabilityOf 会回落 CapCore —— 结果一样，
|
||||
// 但那是**未分类**的默认放行，此后加能力分组时没人会想起它（判据
|
||||
// TestCapability_AllMethodsClassified 就是防这个）。
|
||||
MethodCallCancel: CapCore,
|
||||
// 自身配置读写与元信息属基础能力
|
||||
MethodSettingsGet: CapCore,
|
||||
MethodSettingsSet: CapCore,
|
||||
|
||||
@ -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 归还本插件申请的块。
|
||||
|
||||
@ -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
|
||||
}
|
||||
|
||||
74
internal/plugin/proc/e2e_arena_leak_test.go
Normal file
74
internal/plugin/proc/e2e_arena_leak_test.go
Normal file
@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
45
internal/plugin/proc/execctx.go
Normal file
45
internal/plugin/proc/execctx.go
Normal file
@ -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
|
||||
}
|
||||
@ -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) }
|
||||
|
||||
|
||||
@ -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 {
|
||||
|
||||
@ -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()，
|
||||
|
||||
@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user