Merge pull request 'Catch up event streams by cursor with real reconnect and periodic rescue (U6 slice 1)' (#147) from feature/collab-storage-u6-events into main

internal/pgbus/pgbus.go

@@ -1,8 +1,14 @@
-// Package pgbus 是 PG 档的事件跨实例扇出（阶段 3）：
+// Package pgbus 是 PG 档的事件跨实例扇出（阶段 3 / 协作存储 U6 切片一）：
 // 落盘事务提交后 NOTIFY agent_events '<conv>|<lastSeq>'；每实例一个 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)
+	l := &Listener{dsn: dsn, st: st, bus: b, done: make(chan struct{}), cursor: map[string]int64{}}
-	if err != nil {
+	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)
+			// 真重连（U6）：旧实现对原连接继续 Wait——断线后永不恢复。
-			time.Sleep(time.Second)
+			// 退避重建 → 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)
+		l.catchUp(conv, seq)
-		if err != nil {
-			continue // 事件不可见（极早期窗口）——SSE 重放兜底
-		}
-		l.bus.Publish(ev)
 	}
 }
 
+// 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 拆 '<conv>|<seq>'（会话 ID 不含 '|'）。
 func parsePayload(p string) (string, int64, bool) {
 	i := strings.LastIndex(p, "|")
 	if i < 0 {

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))
+	}
+}