From ac2a1a447f298586886165db1efa6962ff12c9fc Mon Sep 17 00:00:00 2001 From: Andy Dixon Date: Fri, 10 Jul 2026 23:45:47 +0100 Subject: [PATCH] docs(events): correct sequence semantics; test cursors across sequence gaps ADR-0002 described the bigserial seq as gap-free. PostgreSQL sequences are not: values consumed by rolled-back inserts are never reused. The code only ever relied on monotonic order (seq > cursor), so this is a documentation correction plus a regression test that burns sequence values and checks paging, cursor resumption from inside a gap and stats. --- docs/adr/0002-event-immutability.md | 6 ++-- internal/events/events_test.go | 55 +++++++++++++++++++++++++++++ internal/events/store.go | 4 +++ 3 files changed, 63 insertions(+), 2 deletions(-) diff --git a/docs/adr/0002-event-immutability.md b/docs/adr/0002-event-immutability.md index 24abd32..6693f43 100644 --- a/docs/adr/0002-event-immutability.md +++ b/docs/adr/0002-event-immutability.md @@ -9,12 +9,14 @@ Provenance and replay require that history never changes. Application discipline ## Decision `cognitive_events` (and `belief_versions`, `personality_versions`, `audit_log`, `approval_decisions`) get triggers that raise on UPDATE and DELETE. Belief changes, corrections, consolidations and retractions are new events referencing old ones. Retention purges, if ever configured, run through a dedicated, audited maintenance path that disables the trigger inside one transaction and records what was purged. -Events use UUIDv7 ids (time-ordered) plus a `bigserial seq` for gap-free cursors. Durable consumers read by `seq`; the in-memory bus is only a wake-up. +Events use UUIDv7 ids (time-ordered) plus a `bigserial seq` that gives a strictly increasing, monotonic ordering key. Durable consumers read by `seq`; the in-memory bus is only a wake-up. + +The sequence is **not** gap-free and nothing may assume it is: PostgreSQL sequences are non-transactional, so a value consumed by a statement that later rolls back is never reused, and values may also be skipped across crashes or cache pre-allocation. Consumers therefore treat `seq` as an ordering cursor only: "everything after N" is `seq > N`, never `seq = N + 1`. A missing value is not evidence of a missing event; corruption is detected by the append-only triggers and integrity checks, not by counting. `internal/events.TestCursorsToleratePlaceholderGaps` guards this. ## Consequences - History is trustworthy by construction; provenance links never dangle. - Storage grows monotonically; retention policy is an explicit operator decision. -- Cursor-based consumption makes restarts and replay simple. +- Cursor-based consumption makes restarts and replay simple, and tolerates gaps in `seq` by construction. ## Alternatives considered - Event sourcing with projections rebuilt from events only: elegant but forces all reads through replay; we keep mutable projection tables (beliefs, memories) with immutable version tables beside them. diff --git a/internal/events/events_test.go b/internal/events/events_test.go index 1d18fbb..ef9c4e3 100644 --- a/internal/events/events_test.go +++ b/internal/events/events_test.go @@ -112,3 +112,58 @@ func TestStoreAppendQueryImmutableAndCursors(t *testing.T) { t.Fatalf("stats wrong: %+v", st) } } + +// TestCursorsToleratePlaceholderGaps burns sequence values the way a rolled +// back insert does and checks that cursors, forward paging and SSE-style +// resumption (seq > last) see every real event and nothing else. A gap in +// seq must never be mistaken for a missing event (ADR-0002). +func TestCursorsToleratePlaceholderGaps(t *testing.T) { + pool := storage.TestPool(t, migrations.FS) + ctx := context.Background() + m, err := minds.NewStore(pool).Create(ctx, minds.CreateParams{Slug: "gaps"}) + if err != nil { + t.Fatal(err) + } + s := NewStore(pool, NewBus(10), nil) + first, err := s.Append(ctx, events.New{MindID: m.ID, Type: events.Observation, Source: "test", Summary: "one"}) + if err != nil { + t.Fatal(err) + } + // A failed insert consumes a sequence value and rolls back; emulate several. + for i := 0; i < 3; i++ { + if _, err := pool.Exec(ctx, `SELECT nextval(pg_get_serial_sequence('cognitive_events','seq'))`); err != nil { + t.Fatal(err) + } + } + tx, _ := pool.Begin(ctx) + if _, err := s.AppendIn(ctx, tx, events.New{MindID: m.ID, Type: events.Observation, Source: "test", Summary: "rolled back"}); err != nil { + t.Fatal(err) + } + tx.Rollback(ctx) //nolint:errcheck + second, err := s.Append(ctx, events.New{MindID: m.ID, Type: events.Observation, Source: "test", Summary: "two"}) + if err != nil { + t.Fatal(err) + } + if second[0].Seq <= first[0].Seq+1 { + t.Fatalf("test did not create a gap: %d then %d", first[0].Seq, second[0].Seq) + } + // Forward paging from the cursor sees exactly the real events after it. + after, err := s.Query(ctx, Filter{MindID: m.ID, AfterSeq: first[0].Seq, Ascending: true, Limit: 10}) + if err != nil || len(after) != 1 || after[0].ID != second[0].ID { + t.Fatalf("paging across a gap: %v %+v", err, after) + } + // A cursor parked inside the gap still resumes at the next real event. + if err := s.SetCursor(ctx, pool, m.ID, "test", first[0].Seq+2); err != nil { + t.Fatal(err) + } + cur, _ := s.Cursor(ctx, pool, m.ID, "test") + resume, _ := s.Query(ctx, Filter{MindID: m.ID, AfterSeq: cur, Ascending: true, Limit: 10}) + if len(resume) != 1 || resume[0].ID != second[0].ID { + t.Fatalf("resume from inside a gap: %+v", resume) + } + // Stats count real rows, not sequence span. + st, _ := s.Stats(ctx, m.ID, time.Time{}) + if st.Total != 2 || st.LastSeq != second[0].Seq { + t.Fatalf("stats: %+v", st) + } +} diff --git a/internal/events/store.go b/internal/events/store.go index 8891eda..f08e7df 100644 --- a/internal/events/store.go +++ b/internal/events/store.go @@ -385,6 +385,10 @@ func scan(row pgx.Row) (events.Event, error) { } // Cursor returns the last sequence consumed by consumer for a mind. +// +// seq is an ordering key, not a counter: bigserial values are skipped when a +// transaction rolls back, so consumers always ask for "seq > cursor" and +// never for "cursor + 1" (ADR-0002). func (s *Store) Cursor(ctx context.Context, q storage.Querier, mind ids.ID, consumer string) (int64, error) { var seq int64 err := q.QueryRow(ctx, `SELECT last_seq FROM mind_cursors WHERE mind_id=$1 AND consumer=$2`, mind, consumer).Scan(&seq)