Merge remote-tracking branch 'origin/main' into feature/generative-ui-u2
ci / vet-and-test (pull_request) Successful in 40s
ci / vet-and-test (pull_request) Successful in 40s
This commit is contained in:
@@ -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
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user