From fc758475800fed5837749ba347414e6535a6d215 Mon Sep 17 00:00:00 2001 From: weichen Date: Sun, 11 Oct 2026 09:30:29 +0000 Subject: [PATCH] Catch up event streams by cursor with real reconnect and periodic rescue (U6 slice 1) Why: plan section 2/KTD6 found the listener fetched only the last event of a batch (intermediate events invisible to live subscribers) and kept waiting on a dead connection after a drop (no actual reconnect). Evidence: pgbus suite green twice against real PostgreSQL - batch of 64 all visible with continuous seq, 1200 events across pages, connection-kill reconnect recovers the outage window, no-notify events delivered by the periodic rescue; full backend suite green on both tiers. Impact: Listener tracks per-conversation cursors (first-seen catch-up from 0 is harmless under SSE seq dedup), pages via EventsAfter to the notified watermark, rebuilds the connection with backoff and resumes known streams, and a rescue ticker covers lost notifications; Notifier contract unchanged. Refs: proposals/2026-10-11-collaboration-storage-and-distribution.md U6 slice 1, KTD6; R5. --- deps/agent-system-core | 2 +- internal/pgbus/pgbus.go | 147 +++++++++++++++++++++++++++++---- internal/pgbus/pgbus_test.go | 154 +++++++++++++++++++++++++++++++++++ 3 files changed, 285 insertions(+), 18 deletions(-) diff --git a/deps/agent-system-core b/deps/agent-system-core index 7a98a7e..452daf3 160000 --- a/deps/agent-system-core +++ b/deps/agent-system-core @@ -1 +1 @@ -Subproject commit 7a98a7e7fe88e66106916ddcd02e489ad01a8c71 +Subproject commit 452daf3ffa1d94d54ccd0d84a46070ce501bc14b diff --git a/internal/pgbus/pgbus.go b/internal/pgbus/pgbus.go index 32792ec..d476eaa 100644 --- a/internal/pgbus/pgbus.go +++ b/internal/pgbus/pgbus.go @@ -1,8 +1,14 @@ -// Package pgbus 是 PG 档的事件跨实例扇出（阶段 3）： +// Package pgbus 是 PG 档的事件跨实例扇出（阶段 3 / 协作存储 U6 切片一）： // 落盘事务提交后 NOTIFY agent_events '|'；每实例一个 LISTEN 连接， -// 收到通知按 seq 从库取回事件并转发本地 bus（SSE 以 seq 去重，双投无害）。 +// 收到通知按会话游标从库分页追赶转发本地 bus（SSE 以 seq 去重，重投无害）。 // 多实例部署契约：run 在其归属实例内存中执行，会话流量需按会话亲和路由 // （网关/客户端择点）；本包只解决“其他实例的订阅者”可见性。 +// +// U6（KTD6）修复的两个既有缺口： +// - 批内遗漏：通知只带末 seq，旧实现 EventAt 只取末条——批内中间事件对 +// 实时订阅者不可见。现按会话游标 EventsAfter 分页追赶，末 seq 即水位； +// - 断线假重连：旧实现对原连接继续 WaitForNotification。现真正重建 +// 连接→LISTEN→按全部已知会话追赶，另有周期补查覆盖通知丢失。 package pgbus import ( @@ -10,6 +16,7 @@ import ( "log" "strconv" "strings" + "sync" "time" "agent-system-backend/internal/bus" @@ -21,6 +28,12 @@ import ( const channel = "agent_events" +// catchUpPage 追赶分页大小（验收含 1000+ 事件分页可见）。 +const catchUpPage = 500 + +// rescueEvery 周期补查间隔：覆盖 NOTIFY 丢失与监听窗口前后竞争（KTD6）。 +const rescueEvery = 2 * time.Second + // Notifier 落盘后通知（任意连接池连接可用）。 type Notifier struct{ st *store.Store } @@ -39,53 +52,153 @@ func (n *Notifier) Notify(conversationID string, evs []core.StreamEvent) { } } -// Listener 每实例一个：LISTEN → 按 seq 取回 → 本地 bus 转发。 +// Listener 每实例一个：LISTEN → 按会话游标追赶 → 本地 bus 转发。 type Listener struct { conn *pgx.Conn + dsn string st *store.Store bus *bus.Bus done chan struct{} + + mu sync.Mutex + cursor map[string]int64 // conv → 已转发水位（重启清零：SSE 侧重放兜底） } // Start 建立监听连接并开始转发循环；返回后即可用，随 ctx 取消停机。 func Start(ctx context.Context, dsn string, st *store.Store, b *bus.Bus) (*Listener, error) { - conn, err := pgx.Connect(ctx, dsn) - if err != nil { + l := &Listener{dsn: dsn, st: st, bus: b, done: make(chan struct{}), cursor: map[string]int64{}} + if err := l.connect(ctx); err != nil { return nil, err } - if _, err := conn.Exec(ctx, "LISTEN "+channel); err != nil { - conn.Close(ctx) - return nil, err - } - l := &Listener{conn: conn, st: st, bus: b, done: make(chan struct{})} go l.loop(ctx) return l, nil } +func (l *Listener) connect(ctx context.Context) error { + conn, err := pgx.Connect(ctx, l.dsn) + if err != nil { + return err + } + if _, err := conn.Exec(ctx, "LISTEN "+channel); err != nil { + conn.Close(ctx) + return err + } + l.conn = conn + return nil +} + func (l *Listener) loop(ctx context.Context) { defer close(l.done) + // 周期补查独立 goroutine：通知丢失场景恰恰没有通知可触发——不能只在 + // 收到通知后顺带补查（KTD6）。 + go l.rescueLoop(ctx) for { n, err := l.conn.WaitForNotification(ctx) if err != nil { if ctx.Err() != nil { return // 停机 } - log.Printf("[pgbus] wait: %v（1s 后重连）", err) - time.Sleep(time.Second) + // 真重连（U6）：旧实现对原连接继续 Wait——断线后永不恢复。 + // 退避重建 → LISTEN → 按已知会话追赶（补断线窗口内落盘的事件）。 + log.Printf("[pgbus] wait: %v（重连中）", err) + l.reconnect(ctx) continue } conv, seq, ok := parsePayload(n.Payload) if !ok { continue } - ev, err := l.st.EventAt(conv, seq) - if err != nil { - continue // 事件不可见（极早期窗口）——SSE 重放兜底 - } - l.bus.Publish(ev) + l.catchUp(conv, seq) } } +// rescueLoop 周期补查心跳：随 ctx 或主循环退出（done）停止。 +func (l *Listener) rescueLoop(ctx context.Context) { + t := time.NewTicker(rescueEvery) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-l.done: + return + case <-t.C: + l.rescueKnown() + } + } +} + +// reconnect 重建监听连接并立即追赶全部已知会话（有界退避重试）。 +func (l *Listener) reconnect(ctx context.Context) { + _ = l.conn.Close(context.Background()) + for attempt := 0; attempt < 3; attempt++ { + cctx, cancel := context.WithTimeout(ctx, 5*time.Second) + err := l.connect(cctx) + cancel() + if err == nil { + l.rescueKnown() + return + } + select { + case <-ctx.Done(): + return + case <-time.After(time.Duration(attempt+1) * time.Second): + } + } +} + +// rescueKnown 周期补查：不经通知，按各会话游标继续分页追赶（覆盖 NOTIFY +// 丢失、监听生效窗口前后竞争——KTD6「只有落库事件才是重放依据」）。 +// 调用方：重连成功后与 loop 的 rescue 心跳（消息稀疏时由 ticker 兜底）。 +func (l *Listener) rescueKnown() { + l.mu.Lock() + convs := make([]string, 0, len(l.cursor)) + for c := range l.cursor { + convs = append(convs, c) + } + l.mu.Unlock() + for _, conv := range convs { + l.catchUp(conv, 1<<62) + } +} + +// catchUp 从会话游标之后分页取回并转发，直到取空或越过水位 toSeq。 +// 首见会话从 0 开始（SSE 端按 seq 去重，重投无害；这保证批内中间事件可见）。 +func (l *Listener) catchUp(conv string, toSeq int64) { + l.mu.Lock() + since := l.cursor[conv] + l.mu.Unlock() + for { + evs, err := l.st.EventsAfter(conv, since, catchUpPage) + if err != nil { + log.Printf("[pgbus] catchup %s@%d: %v", conv, since, err) + return + } + stop := false + for _, ev := range evs { + if ev.Seq > toSeq { + stop = true + break + } + l.bus.Publish(ev) + since = ev.Seq + } + l.setCursor(conv, since) + if stop || len(evs) < catchUpPage { + return + } + } +} + +func (l *Listener) setCursor(conv string, seq int64) { + l.mu.Lock() + if seq > l.cursor[conv] { + l.cursor[conv] = seq + } + l.mu.Unlock() +} + +// parsePayload 拆 '|'（会话 ID 不含 '|'）。 func parsePayload(p string) (string, int64, bool) { i := strings.LastIndex(p, "|") if i < 0 { diff --git a/internal/pgbus/pgbus_test.go b/internal/pgbus/pgbus_test.go index 920c954..f5400fa 100644 --- a/internal/pgbus/pgbus_test.go +++ b/internal/pgbus/pgbus_test.go @@ -86,3 +86,157 @@ func TestNotifyListenerFanout(t *testing.T) { t.Fatal("no fanout within 3s") } } + +// ---- U6 切片一（KTD6）：批内全可见、分页追赶、断线重连、通知丢失补查 ---- + +// drainUntil 从订阅通道收集事件直到谓词为真或超时；返回收到的全部事件。 +func drainUntil(ch <-chan core.StreamEvent, done func([]core.StreamEvent) bool, timeout time.Duration) []core.StreamEvent { + var got []core.StreamEvent + deadline := time.After(timeout) + for { + select { + case e := <-ch: + got = append(got, e) + if done(got) { + return got + } + case <-deadline: + return got + } + } +} + +// 一批 64 条事件（单条 NOTIFY 只带末 seq）对实时订阅者全部可见—— +// 旧实现 EventAt 只取末条，批内 1..63 全部遗漏。 +func TestBatchEventsAllVisible(t *testing.T) { + st, dsn := pgTestStore(t) + b2 := bus.New() + listener, err := Start(context.Background(), dsn, st, b2) + if err != nil { + t.Fatal(err) + } + defer listener.Close(context.Background()) + ch, stop := b2.Subscribe("conv_batch") + defer stop() + + notifier := NewNotifier(st) + evs := make([]core.StreamEvent, 64) + for i := range evs { + evs[i] = core.StreamEvent{ConversationID: "conv_batch", RunID: "r1", TS: time.Now().UTC().Format(time.RFC3339Nano), V: 1, Type: core.EvRunPhase, Payload: []byte(`{}`)} + } + if _, err := st.AppendEvents("conv_batch", evs); err != nil { + t.Fatal(err) + } + notifier.Notify("conv_batch", evs) + + got := drainUntil(ch, func(g []core.StreamEvent) bool { return len(g) >= 64 }, 5*time.Second) + if len(got) != 64 { + t.Fatalf("批 64 条应全部可见，收到 %d（旧实现只末条可见）", len(got)) + } + // 序号连续（首见会话从 0 追赶，SSE 去重语义下的完整性）。 + for i, e := range got { + if e.Seq != int64(i+1) { + t.Fatalf("序号不连续：got[%d].Seq=%d", i, e.Seq) + } + } +} + +// 1200 条（跨 500/页 的三页）分页追赶全部可见。 +func TestPagedCatchUpBeyondOnePage(t *testing.T) { + st, dsn := pgTestStore(t) + b2 := bus.New() + listener, err := Start(context.Background(), dsn, st, b2) + if err != nil { + t.Fatal(err) + } + defer listener.Close(context.Background()) + ch, stop := b2.Subscribe("conv_page") + defer stop() + + notifier := NewNotifier(st) + const total = 1200 + evs := make([]core.StreamEvent, 0, total) + for i := 0; i < total; i++ { + evs = append(evs, core.StreamEvent{ConversationID: "conv_page", RunID: "r1", TS: time.Now().UTC().Format(time.RFC3339Nano), V: 1, Type: core.EvRunPhase, Payload: []byte(`{}`)}) + } + if _, err := st.AppendEvents("conv_page", evs); err != nil { + t.Fatal(err) + } + notifier.Notify("conv_page", evs) + + got := drainUntil(ch, func(g []core.StreamEvent) bool { return len(g) >= total }, 15*time.Second) + if len(got) != total { + t.Fatalf("1200 条应分页全部可见，收到 %d", len(got)) + } +} + +// 断线重连：底层连接被杀后，新落盘+通知的事件仍可达（真重连路径）。 +func TestListenerReconnectsAfterConnLoss(t *testing.T) { + st, dsn := pgTestStore(t) + b2 := bus.New() + listener, err := Start(context.Background(), dsn, st, b2) + if err != nil { + t.Fatal(err) + } + defer listener.Close(context.Background()) + ch, stop := b2.Subscribe("conv_re") + defer stop() + + notifier := NewNotifier(st) + first := []core.StreamEvent{{ConversationID: "conv_re", RunID: "r1", TS: time.Now().UTC().Format(time.RFC3339Nano), V: 1, Type: core.EvRunPhase, Payload: []byte(`{}`)}} + if _, err := st.AppendEvents("conv_re", first); err != nil { + t.Fatal(err) + } + notifier.Notify("conv_re", first) + drainUntil(ch, func(g []core.StreamEvent) bool { return len(g) >= 1 }, 3*time.Second) + + // 杀掉监听连接（模拟网络断开；旧实现会对此连接永远 Wait 下去）。 + if err := listener.conn.Close(context.Background()); err != nil { + t.Logf("close conn: %v", err) + } + // 断线窗口内落盘一批（重连后应经追赶补上）。 + during := []core.StreamEvent{{ConversationID: "conv_re", RunID: "r1", TS: time.Now().UTC().Format(time.RFC3339Nano), V: 1, Type: core.EvRunPhase, Payload: []byte(`{}`)}} + if _, err := st.AppendEvents("conv_re", during); err != nil { + t.Fatal(err) + } + notifier.Notify("conv_re", during) + + // 第二段 drain 从零计数：断线窗口内的一条事件到达即证明重连+追赶生效。 + got := drainUntil(ch, func(g []core.StreamEvent) bool { return len(g) >= 1 }, 10*time.Second) + if len(got) < 1 { + t.Fatalf("重连后应补见断线窗口内事件：got=%d", len(got)) + } +} + +// 通知丢失补查：事件落盘但**不**发 NOTIFY，周期补查应在 rescueEvery 内送达。 +func TestRescueCoversLostNotifications(t *testing.T) { + st, dsn := pgTestStore(t) + b2 := bus.New() + listener, err := Start(context.Background(), dsn, st, b2) + if err != nil { + t.Fatal(err) + } + defer listener.Close(context.Background()) + ch, stop := b2.Subscribe("conv_lost") + defer stop() + + // 先用一次正常通知建立该会话的游标（补查只覆盖已知会话）。 + notifier := NewNotifier(st) + seed := []core.StreamEvent{{ConversationID: "conv_lost", RunID: "r1", TS: time.Now().UTC().Format(time.RFC3339Nano), V: 1, Type: core.EvRunPhase, Payload: []byte(`{}`)}} + if _, err := st.AppendEvents("conv_lost", seed); err != nil { + t.Fatal(err) + } + notifier.Notify("conv_lost", seed) + drainUntil(ch, func(g []core.StreamEvent) bool { return len(g) >= 1 }, 3*time.Second) + + // 只落盘、不通知（模拟 NOTIFY 丢失）。 + lost := []core.StreamEvent{{ConversationID: "conv_lost", RunID: "r1", TS: time.Now().UTC().Format(time.RFC3339Nano), V: 1, Type: core.EvRunPhase, Payload: []byte(`{}`)}} + if _, err := st.AppendEvents("conv_lost", lost); err != nil { + t.Fatal(err) + } + + got := drainUntil(ch, func(g []core.StreamEvent) bool { return len(g) >= 1 }, 3*rescueEvery+2*time.Second) + if len(got) < 1 { + t.Fatalf("周期补查应送达无通知事件：got=%d", len(got)) + } +}