Make notice redelivery iterative instead of self-recursive #142

Merged
weichen merged 1 commits from feature/notice-delivery-recursion-fix into main 2026-10-11 08:12:14 +00:00
2 changed files with 76 additions and 20 deletions
+46
View File
@@ -0,0 +1,46 @@
package runmgr
import (
"testing"
"time"
"agent-system-backend/internal/store"
core "agent-system-core"
)
// U3 真库门禁回归(job 2352 栈溢出):投递失败(会话行不存在)时旧实现的
// defer 同步再入 deliverNotices 无限自递归。修复后:单次有界返回、通知
// 保留在队列,由下一次触发或重启重投驱动。
func TestNoticeDeliveryFailureIsBounded(t *testing.T) {
runner := &gateRunner{parts: []core.Part{{Type: core.PartText, Text: "结果"}}}
c, _ := newSpawnEnv(t, runner)
c.mgr.SetWorkspaceRoot(t.TempDir())
convID := core.ID(store.NewID("cnv"))
c.mgr.enqueueNotice(convID, bgNotice{owner: c.run.Owner, name: "检索", payload: core.AgentResultPayload{Status: "completed"}})
done := make(chan struct{})
go func() {
c.mgr.deliverNotices(convID)
close(done)
}()
select {
case <-done:
case <-time.After(5 * time.Second):
t.Fatal("deliverNotices 未返回:失败重投仍在递归/自旋")
}
c.mgr.noticeMu.Lock()
queued := len(c.mgr.notices[convID])
c.mgr.noticeMu.Unlock()
if queued != 1 {
t.Fatalf("失败应保留待投通知:queued=%d", queued)
}
// 第二次触发同样有界(不因重试堆积)。
c.mgr.deliverNotices(convID)
c.mgr.noticeMu.Lock()
queued = len(c.mgr.notices[convID])
c.mgr.noticeMu.Unlock()
if queued != 1 {
t.Fatalf("重复失败不应堆积通知:queued=%d", queued)
}
}
+30 -20
View File
@@ -1821,34 +1821,44 @@ func (m *Manager) deliverNotices(convID string) {
defer func() {
m.noticeMu.Lock()
delete(m.delivering, convID)
more := len(m.notices[convID]) > 0
m.noticeMu.Unlock()
if more {
m.deliverNotices(convID)
}
}()
if err := m.startNoticeRun(convID, items); err != nil {
// 迭代而非递归(U3 真库门禁实测):原实现的 defer 在投递失败重新入队后
// 同步再入 deliverNotices——持续失败(如会话忙/查库错误)时无限自递归,
// test-pg 慢速时序下栈溢出(job 2352)。失败本轮即返回:队列保留,
// 由下一次 enqueueNotice 或重启重投驱动;投递期间新到的通知由循环
// 尾部再取,不再经 defer 递归。
for {
if err := m.startNoticeRun(convID, items); err != nil {
m.noticeMu.Lock()
if m.notices == nil {
m.notices = map[string][]bgNotice{}
}
m.notices[convID] = append(items, m.notices[convID]...)
m.noticeMu.Unlock()
return
}
// 投递成功才删持久化行(失败保留下次重投)。
ids := make([]string, 0, len(items))
for _, n := range items {
if n.id != "" {
ids = append(ids, n.id)
}
}
if err := m.st.DeleteNotices(ids); err != nil {
log.Printf("[runmgr] notice delete: %v", err)
}
m.noticeMu.Lock()
if m.notices == nil {
m.notices = map[string][]bgNotice{}
items = append([]bgNotice{}, m.notices[convID]...)
if len(items) == 0 {
m.noticeMu.Unlock()
return
}
m.notices[convID] = append(items, m.notices[convID]...)
delete(m.notices, convID)
m.noticeMu.Unlock()
return
}
// 投递成功才删持久化行(失败保留下次重投)。
ids := make([]string, 0, len(items))
for _, n := range items {
if n.id != "" {
ids = append(ids, n.id)
}
}
if err := m.st.DeleteNotices(ids); err != nil {
log.Printf("[runmgr] notice delete: %v", err)
}
}
func (m *Manager) startNoticeRun(convID string, items []bgNotice) error {
if len(items) == 0 {
return nil