From 1ef1fdeac94c0c6546317be67c721adf088bfb7f Mon Sep 17 00:00:00 2001 From: weichen Date: Sun, 11 Oct 2026 08:12:03 +0000 Subject: [PATCH] Make notice redelivery iterative instead of self-recursive Why: deliverNotices re-invoked itself synchronously from a defer after a failed startNoticeRun requeued the batch - a persistently failing delivery (busy conversation, missing row, slow DB) recursed without bound and overflowed the stack (real-PG CI job 2352, internal/runmgr). Evidence: regression test TestNoticeDeliveryFailureIsBounded (missing conversation row) hangs on the old code and passes with the fix; full backend suite green on SQLite and against real PostgreSQL 18.6 + Redis; clean-HEAD reproduction isolated an unrelated count=2 admin-login artifact. Impact: failed batches now return after requeue (retried on the next enqueue or restart); notices arriving during delivery are picked up by a bounded loop tail instead of defer recursion; no behavior change on the success path. Refs: Dev PR #333; plan R8 real-database gate. --- internal/runmgr/notice_recursion_test.go | 46 ++++++++++++++++++++++ internal/runmgr/runmgr.go | 50 ++++++++++++++---------- 2 files changed, 76 insertions(+), 20 deletions(-) create mode 100644 internal/runmgr/notice_recursion_test.go diff --git a/internal/runmgr/notice_recursion_test.go b/internal/runmgr/notice_recursion_test.go new file mode 100644 index 0000000..f251e70 --- /dev/null +++ b/internal/runmgr/notice_recursion_test.go @@ -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) + } +} diff --git a/internal/runmgr/runmgr.go b/internal/runmgr/runmgr.go index dfcaef2..a71eb1f 100644 --- a/internal/runmgr/runmgr.go +++ b/internal/runmgr/runmgr.go @@ -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