From b0e4fffa62f92957bc40c330c1c0a105c7c597c4 Mon Sep 17 00:00:00 2001 From: prateek Date: Sun, 31 May 2026 00:39:19 +0530 Subject: [PATCH] test(cdc): add full-stack E2E tests through the real store + snapshot source The cdc integration test covers the synchronous Drain/Poll happy path but (1) resyncs from a fake snapshot and (2) never runs the publisher/consumer as the concurrent goroutines the daemon actually uses. Add two E2E tests in the composition-root package that wire the real sqlite.Store, outboxAdapter, Publisher, JSONL log, Consumer, Broadcaster and the REAL snapshotSource (store.ListAll): - RealSnapshotResyncThroughRotation: forces a rotation and asserts the consumer rebuilds from the sessions table, delivering the persisted record payload, with the offset landing at the change_log head. - ConcurrentPublisherConsumer: runs both as goroutines on their tickers and asserts every write is delivered exactly once, in order, offset at head (also exercises the broadcaster hand-off under -race). Co-Authored-By: Claude Opus 4.8 (1M context) --- backend/cdc_e2e_test.go | 194 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 194 insertions(+) create mode 100644 backend/cdc_e2e_test.go diff --git a/backend/cdc_e2e_test.go b/backend/cdc_e2e_test.go new file mode 100644 index 000000000..29b04534c --- /dev/null +++ b/backend/cdc_e2e_test.go @@ -0,0 +1,194 @@ +package main + +import ( + "context" + "encoding/json" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/aoagents/agent-orchestrator/backend/internal/cdc" + "github.com/aoagents/agent-orchestrator/backend/internal/domain" + "github.com/aoagents/agent-orchestrator/backend/internal/ports" +) + +// These are full-stack end-to-end tests of the write+delivery path wired exactly +// as main.go wires it: real sqlite.Store -> real outboxAdapter -> real +// cdc.Publisher -> real JSONL log -> real cdc.Consumer -> real cdc.Broadcaster, +// using the REAL snapshotSource (store.ListAll) rather than a fake. The cdc +// package's own integration test covers the synchronous Drain/Poll happy path +// with a fake snapshot; these cover the two gaps it leaves: a rotation that +// resyncs from the actual sessions table, and the concurrent goroutine model +// the daemon actually runs. + +// TestE2E_RealSnapshotResyncThroughRotation forces a log rotation and asserts the +// consumer rebuilds state from the REAL sessions-table snapshot (not the +// rotated-away bytes), delivering the persisted record's payload. +func TestE2E_RealSnapshotResyncThroughRotation(t *testing.T) { + ctx := context.Background() + store := newWiringStore(t) + dir := t.TempDir() + log, err := cdc.OpenLog(dir, 80) // tiny cap: the second write forces a rotation + if err != nil { + t.Fatal(err) + } + defer log.Close() + + var mu sync.Mutex + var got []cdc.Event + bc := cdc.NewBroadcaster() + bc.Subscribe(func(e cdc.Event) { mu.Lock(); got = append(got, e); mu.Unlock() }) + + con := cdc.NewConsumer("fe", filepath.Join(dir, cdc.LogFileName), store, bc, + cdc.ConsumerConfig{Snapshot: snapshotSource{store: store}}) + if _, err := con.Start(ctx); err != nil { + t.Fatal(err) + } + pub := cdc.NewPublisher(outboxAdapter{store: store}, log, cdc.PublisherConfig{}) + + // First canonical write: drained and consumed live from the original file. + if err := store.Upsert(ctx, wiringRec("s1"), ports.EventSessionCreated); err != nil { + t.Fatal(err) + } + if err := pub.Drain(ctx); err != nil { + t.Fatal(err) + } + if err := con.Poll(ctx); err != nil { + t.Fatal(err) + } + mu.Lock() + before := len(got) + mu.Unlock() + + // Second write pushes the log past its cap -> rotation. The consumer sees a + // fresh file and must resync from the sessions table. + r := wiringRec("s1") + r.Lifecycle.Revision = 1 + if err := store.Upsert(ctx, r, ports.EventSessionStateChanged); err != nil { + t.Fatal(err) + } + if err := pub.Drain(ctx); err != nil { + t.Fatal(err) + } + if err := con.Poll(ctx); err != nil { + t.Fatal(err) + } + + mu.Lock() + defer mu.Unlock() + if len(got) <= before { + t.Fatalf("resync delivered nothing after rotation (got %d, before %d)", len(got), before) + } + // A real session_snapshot for s1 must have been delivered, carrying the full + // record persisted in the sessions table. + var snap *cdc.Event + for i := range got { + if got[i].EventType == "session_snapshot" && got[i].SessionID == "s1" { + snap = &got[i] + } + } + if snap == nil { + t.Fatalf("no real session_snapshot delivered after rotation; got %+v", got) + } + var rec domain.SessionRecord + if err := json.Unmarshal([]byte(snap.Payload), &rec); err != nil { + t.Fatalf("snapshot payload not a SessionRecord: %v", err) + } + if rec.ID != "s1" || rec.Lifecycle.Session.State != domain.SessionWorking { + t.Fatalf("snapshot payload mismatch: %+v", rec) + } + // The consumer's durable offset advanced to the change_log head. + off, err := store.GetOffset(ctx, "fe") + if err != nil { + t.Fatal(err) + } + maxSeq, err := store.MaxChangeLogSeq(ctx) + if err != nil { + t.Fatal(err) + } + if off != maxSeq { + t.Fatalf("offset = %d, want change_log head %d", off, maxSeq) + } +} + +// TestE2E_ConcurrentPublisherConsumer runs the publisher and consumer as the +// daemon runs them — independent goroutines on their own tickers — and asserts +// every canonical write is delivered exactly once, in order, with the offset +// landing at the head. Run under -race this also guards the broadcaster/consumer +// hand-off. +func TestE2E_ConcurrentPublisherConsumer(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + store := newWiringStore(t) + dir := t.TempDir() + log, err := cdc.OpenLog(dir, 0) + if err != nil { + t.Fatal(err) + } + defer log.Close() + + var mu sync.Mutex + var got []cdc.Event + bc := cdc.NewBroadcaster() + bc.Subscribe(func(e cdc.Event) { mu.Lock(); got = append(got, e); mu.Unlock() }) + + pub := cdc.NewPublisher(outboxAdapter{store: store}, log, cdc.PublisherConfig{}) + con := cdc.NewConsumer("fe", filepath.Join(dir, cdc.LogFileName), store, bc, cdc.ConsumerConfig{}) + + pubDone := pub.Start(ctx) + conDone, err := con.Start(ctx) + if err != nil { + t.Fatal(err) + } + + const n = 5 + for i := 0; i < n; i++ { + r := wiringRec("s1") + r.Lifecycle.Revision = i + evt := ports.EventSessionStateChanged + if i == 0 { + evt = ports.EventSessionCreated + } + if err := store.Upsert(ctx, r, evt); err != nil { + t.Fatalf("upsert %d: %v", i, err) + } + } + + // Bounded wait for the goroutine pipeline to deliver everything. + deadline := time.Now().Add(5 * time.Second) + for { + mu.Lock() + count := len(got) + mu.Unlock() + if count >= n { + break + } + if time.Now().After(deadline) { + t.Fatalf("timed out: delivered %d/%d events", count, n) + } + time.Sleep(20 * time.Millisecond) + } + + cancel() + <-pubDone + <-conDone + + mu.Lock() + defer mu.Unlock() + if len(got) != n { + t.Fatalf("delivered %d events, want %d", len(got), n) + } + for i, e := range got { + if e.Seq != int64(i+1) { + t.Fatalf("event %d has seq %d, want %d (out-of-order or duplicate)", i, e.Seq, i+1) + } + } + off, err := store.GetOffset(context.Background(), "fe") + if err != nil { + t.Fatal(err) + } + if off != n { + t.Fatalf("offset = %d, want %d", off, n) + } +}