weichen
fc75847580
ci / vet-and-test (pull_request) Successful in 43s
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.
243 lines
8.2 KiB
Go
243 lines
8.2 KiB
Go
package pgbus
|
||
|
||
// PG 档事件扇出集成测试（环境门控：AGENT_PG_TEST_DSN，原生 PostgreSQL 实例）。
|
||
// 验证：实例 A 落盘+NOTIFY → 实例 B 的 LISTEN 取回并转发到其本地 bus。
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"fmt"
|
||
"os"
|
||
"path/filepath"
|
||
"strconv"
|
||
"testing"
|
||
"time"
|
||
|
||
"agent-system-backend/internal/bus"
|
||
"agent-system-backend/internal/store"
|
||
core "agent-system-core"
|
||
)
|
||
|
||
func pgTestStore(t *testing.T) (*store.Store, string) {
|
||
t.Helper()
|
||
dsn := os.Getenv("AGENT_PG_TEST_DSN")
|
||
if dsn == "" {
|
||
t.Skip("AGENT_PG_TEST_DSN not set")
|
||
}
|
||
schema := "pgbus_" + strconv.Itoa(os.Getpid()) + "_" + fmt.Sprint(time.Now().UnixNano()%1e6)
|
||
admin, err := sql.Open("pgx", dsn)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
// pg_trgm 预装 public（幂等），理由见 store 包 openTestStore 注释：
|
||
// 避免扩展落进用例 schema 后随 CASCADE 删除造成的并行竞态。
|
||
if _, err := admin.Exec(`CREATE EXTENSION IF NOT EXISTS pg_trgm`); err != nil {
|
||
admin.Close()
|
||
t.Fatalf("preinstall pg_trgm: %v", err)
|
||
}
|
||
if _, err := admin.Exec("CREATE SCHEMA " + schema); err != nil {
|
||
admin.Close()
|
||
t.Fatal(err)
|
||
}
|
||
t.Cleanup(func() {
|
||
admin.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE")
|
||
admin.Close()
|
||
})
|
||
full := store.AppendPGRuntimeParam(dsn, "search_path", schema+",public")
|
||
st, err := store.OpenPostgres(full, filepath.Join(t.TempDir(), "artifacts"))
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
t.Cleanup(func() { st.Close() })
|
||
return st, full
|
||
}
|
||
|
||
func TestNotifyListenerFanout(t *testing.T) {
|
||
st, dsn := pgTestStore(t)
|
||
|
||
// “实例 B”：独立 bus + 监听连接。
|
||
b2 := bus.New()
|
||
listener, err := Start(context.Background(), dsn, st, b2)
|
||
if err != nil {
|
||
t.Fatalf("listen: %v", err)
|
||
}
|
||
defer listener.Close(context.Background())
|
||
|
||
ch, stop := b2.Subscribe("conv_x")
|
||
defer stop()
|
||
|
||
// “实例 A”：批落盘 + NOTIFY（Notifier 与 listener 共库不同连接）。
|
||
notifier := NewNotifier(st)
|
||
evs := []core.StreamEvent{
|
||
{ConversationID: "conv_x", RunID: "r1", TS: time.Now().UTC().Format(time.RFC3339Nano), V: 1, Type: core.EvRunPhase, Payload: []byte(`{"phase":"thinking"}`)},
|
||
{ConversationID: "conv_x", RunID: "r1", TS: time.Now().UTC().Format(time.RFC3339Nano), V: 1, Type: core.EvRunCompleted, Payload: []byte(`{}`)},
|
||
}
|
||
if _, err := st.AppendEvents("conv_x", evs); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
notifier.Notify("conv_x", evs)
|
||
|
||
select {
|
||
case e := <-ch:
|
||
if e.ConversationID != "conv_x" || e.Seq == 0 {
|
||
t.Fatalf("fanout event malformed: %+v", e)
|
||
}
|
||
case <-time.After(3 * time.Second):
|
||
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))
|
||
}
|
||
}
|