feat(ctxmgr,server): T20-B5 chat turn compensation observability + orphan sweeper
This commit is contained in:
weichen
@@ -0,0 +1,69 @@
|
||||
package ctxmgr
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// TestSweepExpiredProcessingTurns_Memory 验证 T20-B5 孤儿批量扫描：
|
||||
// 过租约的 processing 回合被标 failed；未过期/已终态不受影响；幂等可重复。
|
||||
func TestSweepExpiredProcessingTurns_Memory(t *testing.T) {
|
||||
mgr := NewContextManager(nil, nil, DefaultTimeDecayConfig())
|
||||
ctx := context.Background()
|
||||
|
||||
// 直接在内存 map 构造三个状态：过期 processing / 新鲜 processing / 已完成
|
||||
past := time.Now().Add(-1 * time.Minute)
|
||||
future := time.Now().Add(5 * time.Minute)
|
||||
mgr.mu.Lock()
|
||||
mgr.turns[turnMemKey("sess-sweep", "turn-expired")] = &ChatTurnRecord{
|
||||
SessionID: "sess-sweep",
|
||||
ClientTurnID: "turn-expired",
|
||||
Status: TurnStatusProcessing,
|
||||
LeaseExpiresAt: &past,
|
||||
}
|
||||
mgr.turns[turnMemKey("sess-sweep", "turn-fresh")] = &ChatTurnRecord{
|
||||
SessionID: "sess-sweep",
|
||||
ClientTurnID: "turn-fresh",
|
||||
Status: TurnStatusProcessing,
|
||||
LeaseExpiresAt: &future,
|
||||
}
|
||||
done := time.Now()
|
||||
mgr.turns[turnMemKey("sess-sweep", "turn-done")] = &ChatTurnRecord{
|
||||
SessionID: "sess-sweep",
|
||||
ClientTurnID: "turn-done",
|
||||
Status: TurnStatusCompleted,
|
||||
CompletedAt: &done,
|
||||
}
|
||||
mgr.mu.Unlock()
|
||||
|
||||
n, err := mgr.SweepExpiredProcessingTurns(ctx, 100)
|
||||
if err != nil {
|
||||
t.Fatalf("sweep: %v", err)
|
||||
}
|
||||
if n != 1 {
|
||||
t.Fatalf("expected exactly 1 swept, got %d", n)
|
||||
}
|
||||
|
||||
if rec, err := mgr.GetChatTurn(ctx, "sess-sweep", "turn-expired"); err != nil {
|
||||
t.Fatalf("get after sweep: %v", err)
|
||||
} else if rec.Status != TurnStatusFailed {
|
||||
t.Fatalf("expired turn must be failed, got %s", rec.Status)
|
||||
}
|
||||
if rec, err := mgr.GetChatTurn(ctx, "sess-sweep", "turn-fresh"); err != nil {
|
||||
t.Fatalf("get fresh: %v", err)
|
||||
} else if rec.Status != TurnStatusProcessing {
|
||||
t.Fatalf("fresh turn must stay processing, got %s", rec.Status)
|
||||
}
|
||||
if rec, err := mgr.GetChatTurn(ctx, "sess-sweep", "turn-done"); err != nil {
|
||||
t.Fatalf("get done: %v", err)
|
||||
} else if rec.Status != TurnStatusCompleted {
|
||||
t.Fatalf("done turn must stay completed, got %s", rec.Status)
|
||||
}
|
||||
|
||||
// 幂等：再扫一次无新变化
|
||||
n2, _ := mgr.SweepExpiredProcessingTurns(ctx, 100)
|
||||
if n2 != 0 {
|
||||
t.Fatalf("second sweep expect 0, got %d", n2)
|
||||
}
|
||||
}
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"time"
|
||||
)
|
||||
|
||||
@@ -281,6 +282,12 @@ func (m *ContextManager) CompleteChatTurn(ctx context.Context, sessionID, client
|
||||
SET status = $1, assistant_message_id = $2, completed_at = $3, updated_at = $3
|
||||
WHERE session_id = $4 AND client_turn_id = $5
|
||||
`, TurnStatusCompleted, assistantMessageID, now, sessionID, clientTurnID)
|
||||
if err != nil {
|
||||
// T20-B5：补偿路径不再静默；结构化日志便于告警与计数。
|
||||
slog.ErrorContext(ctx, "chat_turn_complete_failed",
|
||||
"session_id", sessionID, "client_turn_id", clientTurnID,
|
||||
"assistant_message_id", assistantMessageID, "error", err)
|
||||
}
|
||||
return err
|
||||
}
|
||||
m.mu.Lock()
|
||||
@@ -301,6 +308,11 @@ func (m *ContextManager) FailChatTurn(ctx context.Context, sessionID, clientTurn
|
||||
UPDATE chat_turns SET status = $1, updated_at = now()
|
||||
WHERE session_id = $2 AND client_turn_id = $3
|
||||
`, TurnStatusFailed, sessionID, clientTurnID)
|
||||
if err != nil {
|
||||
// T20-B5：失败补偿自身失败必须可见（否则孤儿 processing 无声累积）。
|
||||
slog.ErrorContext(ctx, "chat_turn_fail_compensation_failed",
|
||||
"session_id", sessionID, "client_turn_id", clientTurnID, "error", err)
|
||||
}
|
||||
return err
|
||||
}
|
||||
m.mu.Lock()
|
||||
@@ -397,3 +409,52 @@ func (m *ContextManager) GetChatTurnByAssistantMessageID(ctx context.Context, se
|
||||
}
|
||||
return nil, ErrChatTurnNotFound
|
||||
}
|
||||
|
||||
// SweepExpiredProcessingTurns 批量清理孤儿回合（T20-B5）：
|
||||
// status=processing 且 lease_expires_at < now 的回合标记为 failed。
|
||||
// 返回清理条数；错误不打断扫描（单条失败仅记日志）。
|
||||
// 幂等：重复扫描对已终态回合无影响。
|
||||
func (m *ContextManager) SweepExpiredProcessingTurns(ctx context.Context, limit int) (int, error) {
|
||||
if limit <= 0 {
|
||||
limit = 100
|
||||
}
|
||||
if m.pool != nil {
|
||||
tag, err := m.pool.Exec(ctx, `
|
||||
UPDATE chat_turns SET status = $1, updated_at = now()
|
||||
WHERE id IN (
|
||||
SELECT id FROM chat_turns
|
||||
WHERE status = $2 AND lease_expires_at < now()
|
||||
ORDER BY lease_expires_at ASC
|
||||
LIMIT $3
|
||||
)
|
||||
`, TurnStatusFailed, TurnStatusProcessing, limit)
|
||||
if err != nil {
|
||||
slog.ErrorContext(ctx, "chat_turn_sweep_failed", "error", err)
|
||||
return 0, err
|
||||
}
|
||||
n := tag.RowsAffected()
|
||||
if n > 0 {
|
||||
slog.WarnContext(ctx, "chat_turn_sweep_expired_processing",
|
||||
"marked_failed", n, "limit", limit)
|
||||
}
|
||||
return int(n), nil
|
||||
}
|
||||
// 内存模式：扫描过期 processing
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
now := time.Now()
|
||||
swept := 0
|
||||
for key, rec := range m.turns {
|
||||
if rec.Status == TurnStatusProcessing &&
|
||||
rec.LeaseExpiresAt != nil && rec.LeaseExpiresAt.Before(now) {
|
||||
rec.Status = TurnStatusFailed
|
||||
rec.UpdatedAt = now
|
||||
swept++
|
||||
slog.WarnContext(ctx, "chat_turn_sweep_expired_processing", "key", key)
|
||||
}
|
||||
if swept >= limit {
|
||||
break
|
||||
}
|
||||
}
|
||||
return swept, nil
|
||||
}
|
||||
|
||||
@@ -0,0 +1,31 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
)
|
||||
|
||||
// startChatTurnSweeper 周期清理孤儿 chat 回合（T20-B5）：
|
||||
// status=processing 且租约过期的回合批量标 failed，防止崩溃/超时后
|
||||
// 幂等键被永久占位。间隔保守（2 分钟），批上限 200。
|
||||
func (s *Server) startChatTurnSweeper() {
|
||||
if s == nil || s.ctxMgr == nil {
|
||||
return
|
||||
}
|
||||
s.chatTurnSweepStop = make(chan struct{})
|
||||
go func() {
|
||||
t := time.NewTicker(2 * time.Minute)
|
||||
defer t.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-s.chatTurnSweepStop:
|
||||
return
|
||||
case <-t.C:
|
||||
// 独立超时，避免扫描被慢查询拖住 drain 窗口。
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
_, _ = s.ctxMgr.SweepExpiredProcessingTurns(ctx, 200)
|
||||
cancel()
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
@@ -78,6 +78,7 @@ type Server struct {
|
||||
lastMaintToken string
|
||||
wsReg *workspace.FileRegistry
|
||||
wsGCStop chan struct{}
|
||||
chatTurnSweepStop chan struct{}
|
||||
httpServer *http.Server
|
||||
intentPlans intentflow.Service
|
||||
planAccessFn func(ctx context.Context, projectID, userID int64) error
|
||||
@@ -154,6 +155,7 @@ func NewFromRuntime(rt *app.Runtime) *Server {
|
||||
s.initArtifactVault()
|
||||
s.wireFinalArtifactHooks()
|
||||
s.startVaultPublisher()
|
||||
s.startChatTurnSweeper()
|
||||
// Durable notify is live via events/docnotify consumer; do not assign OnOutbox here
|
||||
// (see internal/app.bindDocumentNotify and unhook_test).
|
||||
return s
|
||||
@@ -464,6 +466,10 @@ func (s *Server) Shutdown(ctx context.Context) error {
|
||||
close(s.vaultStop)
|
||||
s.vaultStop = nil
|
||||
}
|
||||
if s.chatTurnSweepStop != nil {
|
||||
close(s.chatTurnSweepStop)
|
||||
s.chatTurnSweepStop = nil
|
||||
}
|
||||
if s.taskRuns != nil {
|
||||
_ = s.taskRuns.WaitIdle(ctx)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user