diff --git a/go/events/events.go b/go/events/events.go index 6e857a7e3..20bd454ec 100644 --- a/go/events/events.go +++ b/go/events/events.go @@ -22,14 +22,16 @@ import ( otelx "github.com/RigelBuild/compass/go/internal/otel" ) -// ringCapacity bounds replay memory; a subscriber that falls further behind -// than this recovers by re-snapshotting at sinceSeq = 0. -const ringCapacity = 1024 +// RingCapacity bounds replay memory; a subscriber that falls further behind +// than this recovers by re-snapshotting at sinceSeq = 0. Exported so tests in +// other packages can derive an overrun count from it instead of hardcoding a +// literal that silently stops overrunning if this grows. +const RingCapacity = 1024 // liveBufferCapacity is the per-subscriber live-tail buffer depth. Matched to // the ring so a subscriber lagging by less than the ring window can still // recover its gap via sinceSeq replay after re-subscribing. -const liveBufferCapacity = ringCapacity +const liveBufferCapacity = RingCapacity // Stamped is a published payload plus the ordering envelope the bus stamps onto // it: the monotonic Seq, the wall-clock publish time, and the per-boot @@ -161,7 +163,7 @@ func NewBus[P any]() *Bus[P] { return &Bus[P]{ instanceEpoch: epochNonce(), nextSeq: 1, - ring: make([]Stamped[P], 0, ringCapacity), + ring: make([]Stamped[P], 0, RingCapacity), } } @@ -197,7 +199,7 @@ func (b *Bus[P]) publish(traceparent string, payload P) uint64 { Traceparent: traceparent, } - if len(b.ring) == ringCapacity { + if len(b.ring) == RingCapacity { copy(b.ring, b.ring[1:]) b.ring[len(b.ring)-1] = event } else { diff --git a/go/events/events_test.go b/go/events/events_test.go index 3d186ea76..e24742126 100644 --- a/go/events/events_test.go +++ b/go/events/events_test.go @@ -6,7 +6,7 @@ package events // HeadSeq, and the replay->live handoff. // // White-box (package events) so the eviction and overrun contracts can be -// stated in terms of ringCapacity / liveBufferCapacity rather than a magic 1024 +// stated in terms of RingCapacity / liveBufferCapacity rather than a magic 1024 // that would silently drift if the constant changed. import ( @@ -118,7 +118,7 @@ func TestSubscribeAtZeroSnapshotsTheWholeRing(t *testing.T) { func TestSubscribeBelowEvictedSpanUnderflows(t *testing.T) { bus := NewBus[ev]() - for range ringCapacity + 10 { + for range RingCapacity + 10 { bus.Publish(ready()) } // seqs 1..=10 were evicted; oldest retained is 11. A cursor of 5 wants @@ -132,7 +132,7 @@ func TestSubscribeAtEvictionBoundarySucceedsAndReplaysFromNewOldest(t *testing.T bus := NewBus[ev]() // One eviction: fill the ring, then push one more so seq 1 drops and the // oldest retained becomes seq 2. - for range ringCapacity + 1 { + for range RingCapacity + 1 { bus.Publish(ready()) } // cursor=1 wants events from 2 onward, which are exactly what remains — the @@ -144,8 +144,8 @@ func TestSubscribeAtEvictionBoundarySucceedsAndReplaysFromNewOldest(t *testing.T if sub.Replay[0].Seq != 2 { t.Fatalf("first replay seq = %d, want 2 (new oldest)", sub.Replay[0].Seq) } - if last := sub.Replay[len(sub.Replay)-1].Seq; last != uint64(ringCapacity+1) { - t.Fatalf("last replay seq = %d, want %d", last, ringCapacity+1) + if last := sub.Replay[len(sub.Replay)-1].Seq; last != uint64(RingCapacity+1) { + t.Fatalf("last replay seq = %d, want %d", last, RingCapacity+1) } } @@ -388,17 +388,17 @@ func TestOverrunClosesLiveAndLatchesLagged(t *testing.T) { t.Fatal("Lagged() = false after overrun, want true") } - // The ring still holds the last ringCapacity events, so a re-subscribe + // The ring still holds the last RingCapacity events, so a re-subscribe // within the window recovers gap-free. recover, err := bus.Subscribe(0, 0) if err != nil { t.Fatalf("re-Subscribe(0,0) after lag: %v", err) } - if len(recover.Replay) != ringCapacity { - t.Fatalf("recovery replay len = %d, want %d (the retained ring)", len(recover.Replay), ringCapacity) + if len(recover.Replay) != RingCapacity { + t.Fatalf("recovery replay len = %d, want %d (the retained ring)", len(recover.Replay), RingCapacity) } - if last := recover.Replay[len(recover.Replay)-1].Seq; last != uint64(ringCapacity+1) { - t.Fatalf("recovery last seq = %d, want %d", last, ringCapacity+1) + if last := recover.Replay[len(recover.Replay)-1].Seq; last != uint64(RingCapacity+1) { + t.Fatalf("recovery last seq = %d, want %d", last, RingCapacity+1) } } @@ -704,7 +704,7 @@ func TestClosedBusSubscribeWithInvalidCursorReturnsTerminalStream(t *testing.T) // gate. name: "below evicted span", prepare: func(bus *Bus[ev]) (uint64, uint64) { - for range ringCapacity + 10 { + for range RingCapacity + 10 { bus.Publish(ready()) } return 5, bus.InstanceEpoch() diff --git a/go/internal/comms/subscribe_failclosed_test.go b/go/internal/comms/subscribe_failclosed_test.go index a8d85bf94..e4974e7ae 100644 --- a/go/internal/comms/subscribe_failclosed_test.go +++ b/go/internal/comms/subscribe_failclosed_test.go @@ -24,6 +24,11 @@ // the M5 path unreachable in a no-DB lane, since Comms holds the store // concretely — so the client-observable CodeInternal/opaque-message contract is // asserted against forwardComms's real return and the real errStreamVisibility. +// +// This file also hosts the untagged, no-database ring-lag resync and +// replay-boundary tests (RIG-3538): they share driveForwardComms + a real +// events.Bus, and the arm-2 underflow cases drive SubscribeComms with a nil +// store, whose terminal resync precedes any store access. package comms @@ -34,7 +39,9 @@ import ( "net/http" "net/http/httptest" "strings" + "sync" "testing" + "time" "connectrpc.com/connect" @@ -354,3 +361,314 @@ func TestForwardCommsSkipsNonVisibleEventAndContinues(t *testing.T) { t.Fatalf("delivered channel = %q, want the visible %q (hidden one must be skipped)", ch, "chan-visible") } } + +// visibleAlways is a no-DB eventVisibility that admits every event, so a +// bus-only test drives forwardComms's delivery + terminal-resync paths without a +// store: the ring-lag contract is a bus/handler property, not a visibility one. +type visibleAlways struct{} + +func (visibleAlways) IsTopicChannelMember(context.Context, store.AccountID, string) (bool, error) { + return true, nil +} +func (visibleAlways) IsChannelMember(context.Context, store.AccountID, store.ChannelID) (bool, error) { + return true, nil +} +func (visibleAlways) ChannelVisibleTo(context.Context, store.AccountID, store.ChannelID) (bool, error) { + return true, nil +} +func (visibleAlways) ChannelGroupVisibleTo(context.Context, store.AccountID, store.ChannelGroupID) (bool, error) { + return true, nil +} +func (visibleAlways) AccountVisibleTo(context.Context, store.AccountID, store.AccountID) (bool, error) { + return true, nil +} +func (visibleAlways) IsAgentWorkspaceVisible(context.Context, store.AccountID, store.AccountID) (bool, error) { + return true, nil +} +func (visibleAlways) SharesVisibleChannel(context.Context, store.AccountID, store.AccountID) (bool, error) { + return true, nil +} + +var _ eventVisibility = visibleAlways{} + +// overrunSubscription subscribes to a fresh bus, then publishes more than the +// live buffer holds without draining Live, so the bus latches the subscriber +// lagged and closes its channel (events_test.go TestOverrunClosesLiveAndLatchesLagged, +// a different package — the technique, not its symbols). The returned +// subscription drives forwardComms's sub.Lagged() terminal-resync branch. +func overrunSubscription(t *testing.T) events.Subscription[*compassv1.SubscribeCommsResponse] { + t.Helper() + bus := events.NewBus[*compassv1.SubscribeCommsResponse]() + sub, err := bus.Subscribe(0, 0) + if err != nil { + t.Fatalf("bus.Subscribe(0,0): %v", err) + } + // One past the live buffer (sized to events.RingCapacity): the overflowing + // publish is what the non-blocking fan-out cannot place, so it latches + // lagged. Derived from the constant, never a literal — a capacity bump must + // keep overrunning rather than silently stop. + for i := range events.RingCapacity + 1 { + bus.Publish(messagePostedOn(fmt.Sprintf("chan-%d", i))) + } + if !sub.Lagged() { + t.Fatalf("subscription not lagged after 1025 undrained publishes, want lagged") + } + return sub +} + +// TestForwardCommsLiveTailOverrunEmitsTerminalResync pins arm 1: a subscriber +// that overruns the ring (Lagged) makes forwardComms emit a CommsResyncRequired +// as its FINAL frame, then end the stream cleanly (nil, clean EOF) — never a +// fault. The last-frame assertion is the contract: a resync merely "present" +// somewhere would be a weaker claim than terminal. +// +// Teeth: dropping the sub.Lagged() send in forwardComms yields a final frame +// that is a MessagePosted, not a resync. +func TestForwardCommsLiveTailOverrunEmitsTerminalResync(t *testing.T) { + sub := overrunSubscription(t) + wantEpoch := sub.Epoch + + got := driveForwardComms(t, visibleAlways{}, sub) + + if got.fwdErr != nil { + t.Fatalf("forwardComms returned %v, want nil (an overrun ends cleanly, not a fault)", got.fwdErr) + } + if got.streamErr != nil { + t.Fatalf("client saw terminal error %v (code %v), want a clean EOF after the resync", got.streamErr, connect.CodeOf(got.streamErr)) + } + if len(got.received) == 0 { + t.Fatalf("received 0 frames, want the buffered live tail then a terminal resync") + } + // Every frame ahead of the resync must be real buffered content: a + // degradation that dropped the live tail but still sent the terminal frame + // would satisfy the last-frame assertion alone. + for i, f := range got.received[:len(got.received)-1] { + if f.GetMessagePosted() == nil { + t.Fatalf("frame %d = %T, want a buffered MessagePosted ahead of the terminal resync", i, f.GetPayload()) + } + } + final := got.received[len(got.received)-1] + if final.GetResyncRequired() == nil { + t.Fatalf("final frame = %T, want CommsResyncRequired as the last frame", final.GetPayload()) + } + if final.GetInstanceEpoch() != wantEpoch { + t.Fatalf("resync epoch = %d, want the subscription epoch %d", final.GetInstanceEpoch(), wantEpoch) + } +} + +// driveUnderflowResync opens a SubscribeComms stream over httptest against a +// nil-store Comms and returns the frames the client received plus its terminal +// error. The underflow branch returns its terminal resync before +// actorFromContext or any c.store access, so a nil store is sound here — not a +// weakened substitute. The bus supplies the seq space and instance epoch the +// cursor is validated against. +// +// The context is bounded because the failure mode of a dropped underflow guard +// is a registered live subscriber tailing forever: without a deadline that +// surfaces as a whole-suite timeout instead of this test failing. +func driveUnderflowResync(t *testing.T, bus *events.Bus[*compassv1.SubscribeCommsResponse], req *compassv1.SubscribeCommsRequest) forwardResult { + t.Helper() + svc := NewComms(nil, bus, testActor) + + path, handler := compassv1connect.NewCommsServiceHandler(svc) + mux := http.NewServeMux() + mux.Handle(path, handler) + srv := httptest.NewServer(mux) + t.Cleanup(srv.Close) + + client := compassv1connect.NewCommsServiceClient(srv.Client(), srv.URL) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + t.Cleanup(cancel) + stream, err := client.SubscribeComms(ctx, connect.NewRequest(req)) + if err != nil { + t.Fatalf("SubscribeComms: %v", err) + } + t.Cleanup(func() { _ = stream.Close() }) + + var received []*compassv1.SubscribeCommsResponse + for stream.Receive() { + received = append(received, stream.Msg()) + } + return forwardResult{received: received, streamErr: stream.Err()} +} + +// TestSubscribeCommsCursorAtOrBeyondHeadResyncs pins arm 2's at/beyond-head +// branch in Subscribe: a positioned cursor >= the next seq the bus would +// assign refers to an event never emitted, so SubscribeComms returns a single +// terminal CommsResyncRequired stamped with the CURRENT instance epoch, then a +// clean end. Distinct from the stale-epoch branch (same-epoch cursor here). +// +// Teeth: were the at/beyond-head guard removed, Subscribe would register a live +// subscriber and the stream would hang on the live tail instead of resyncing. +// +// The guard itself is covered at the bus layer too (events_test.go +// TestSubscribeAtOrBeyondNextSeqUnderflows). The redundancy is deliberate: the +// handler translates every underflow trigger into the same wire frame, so this +// is the black-box confirmation that THIS trigger reaches a client as a resync. +func TestSubscribeCommsCursorAtOrBeyondHeadResyncs(t *testing.T) { + bus := events.NewBus[*compassv1.SubscribeCommsResponse]() + bus.Publish(messagePostedOn("chan-a")) // seq 1, so nextSeq is 2 + + got := driveUnderflowResync(t, bus, &compassv1.SubscribeCommsRequest{ + SinceSeq: 5, InstanceEpoch: bus.InstanceEpoch(), + }) + + if got.streamErr != nil { + t.Fatalf("client saw terminal error %v, want a clean EOF after the resync", got.streamErr) + } + if len(got.received) != 1 { + t.Fatalf("received %d frames, want exactly 1 (the terminal resync)", len(got.received)) + } + if got.received[0].GetResyncRequired() == nil { + t.Fatalf("first frame = %T, want CommsResyncRequired", got.received[0].GetPayload()) + } + if epoch := got.received[0].GetInstanceEpoch(); epoch != bus.InstanceEpoch() { + t.Fatalf("resync epoch = %d, want the current instance epoch %d", epoch, bus.InstanceEpoch()) + } +} + +// TestSubscribeCommsCursorOlderThanRetainedResyncs pins arm 2's evicted-cursor +// branch in Subscribe, genuinely distinct from the at/beyond-head case: the +// ring is overrun so its oldest retained seq is far above 1, then a cursor at +// seq 1 (older than anything retained) can't be caught up by replay, so the +// stream is a single terminal resync at the current epoch. +// +// Teeth: without the eviction guard, replay would silently start mid-ring and +// the client would miss the gap between its cursor and the oldest retained seq. +func TestSubscribeCommsCursorOlderThanRetainedResyncs(t *testing.T) { + bus := events.NewBus[*compassv1.SubscribeCommsResponse]() + // Publish past ring capacity so seq 1 is evicted: after RingCapacity+N + // publishes the oldest retained seq is N+1, and the guard needs it above + // sinceSeq+1, so any N >= 2 works. Derived from the constant so a capacity + // bump still evicts rather than quietly registering a live tail. + for i := range events.RingCapacity + 10 { + bus.Publish(messagePostedOn(fmt.Sprintf("chan-%d", i))) + } + + got := driveUnderflowResync(t, bus, &compassv1.SubscribeCommsRequest{ + SinceSeq: 1, InstanceEpoch: bus.InstanceEpoch(), + }) + + if got.streamErr != nil { + t.Fatalf("client saw terminal error %v, want a clean EOF after the resync", got.streamErr) + } + if len(got.received) != 1 { + t.Fatalf("received %d frames, want exactly 1 (the terminal resync)", len(got.received)) + } + if got.received[0].GetResyncRequired() == nil { + t.Fatalf("first frame = %T, want CommsResyncRequired", got.received[0].GetPayload()) + } + if epoch := got.received[0].GetInstanceEpoch(); epoch != bus.InstanceEpoch() { + t.Fatalf("resync epoch = %d, want the current instance epoch %d", epoch, bus.InstanceEpoch()) + } +} + +// TestForwardCommsCursorAtHeadReplaysNothingBeforeLiveTail pins the +// replay-boundary exactness: a subscription positioned exactly at head has an +// empty Replay, so the FIRST frame forwardComms sends is a live-tail event +// published AFTER subscribing — never a replayed one. Proven by a sentinel +// (happens-before: publish is under the same lock Subscribe took), NOT a sleep: +// the sentinel is the first frame iff nothing replayed ahead of it. +// +// Teeth: an off-by-one that replayed the at-head event (Seq > sinceSeq flipped +// to >=) would put that event first, ahead of the sentinel. +func TestForwardCommsCursorAtHeadReplaysNothingBeforeLiveTail(t *testing.T) { + bus := events.NewBus[*compassv1.SubscribeCommsResponse]() + headSeq := bus.Publish(messagePostedOn("chan-before")) // the at-head event + + sub, err := bus.Subscribe(headSeq, bus.InstanceEpoch()) + if err != nil { + t.Fatalf("bus.Subscribe(%d, epoch): %v", headSeq, err) + } + if len(sub.Replay) != 0 { + t.Fatalf("replay len = %d for a cursor at head, want 0 (nothing after head)", len(sub.Replay)) + } + // Published after Subscribe registered the live subscriber, so it lands on + // the live tail: it must be the first thing forwardComms sends. + sentinelSeq := bus.Publish(messagePostedOn("chan-sentinel")) + bus.Close() + + got := driveForwardComms(t, visibleAlways{}, sub) + + if got.fwdErr != nil { + t.Fatalf("forwardComms returned %v, want nil on a clean drain", got.fwdErr) + } + if got.streamErr != nil { + t.Fatalf("client saw terminal error %v, want a clean EOF", got.streamErr) + } + if len(got.received) != 1 { + t.Fatalf("received %d frames, want exactly 1 (the sentinel, no replay ahead of it)", len(got.received)) + } + if seq := got.received[0].GetSeq(); seq != sentinelSeq { + t.Fatalf("first frame seq = %d, want the sentinel seq %d (an at-head cursor must replay nothing)", seq, sentinelSeq) + } + if ch := got.received[0].GetMessagePosted().GetMessage().GetTopicId(); ch != "chan-sentinel" { + t.Fatalf("first frame channel = %q, want %q", ch, "chan-sentinel") + } +} + +// TestForwardCommsConcurrentPublishesObserveOneTotalSeqOrder pins that two +// subscribers tailing one bus over concurrent publishes observe the SAME total +// seq order: the bus stamps seq under a single lock, so every subscriber sees +// one linearization. Both observed seq slices must be identical AND strictly +// increasing — "both non-empty" would not distinguish a shared order from two +// divergent ones. +// +// Teeth: were seq assignment not serialized, two concurrent publishers could +// interleave differently per subscriber and the two slices would diverge. +func TestForwardCommsConcurrentPublishesObserveOneTotalSeqOrder(t *testing.T) { + bus := events.NewBus[*compassv1.SubscribeCommsResponse]() + subA, err := bus.Subscribe(0, 0) + if err != nil { + t.Fatalf("bus.Subscribe A: %v", err) + } + subB, err := bus.Subscribe(0, 0) + if err != nil { + t.Fatalf("bus.Subscribe B: %v", err) + } + + // Publish concurrently: the bus's publish lock is what forces a single total + // order despite the racing goroutines. + const publishers, perPublisher = 4, 50 + var wg sync.WaitGroup + for p := range publishers { + wg.Go(func() { + for i := range perPublisher { + bus.Publish(messagePostedOn(fmt.Sprintf("chan-%d-%d", p, i))) + } + }) + } + wg.Wait() + bus.Close() + + gotA := driveForwardComms(t, visibleAlways{}, subA) + gotB := driveForwardComms(t, visibleAlways{}, subB) + if gotA.fwdErr != nil || gotB.fwdErr != nil { + t.Fatalf("forwardComms returned A=%v B=%v, want nil (clean drain both)", gotA.fwdErr, gotB.fwdErr) + } + + seqsA := seqsOf(gotA.received) + seqsB := seqsOf(gotB.received) + total := publishers * perPublisher + if len(seqsA) != total || len(seqsB) != total { + t.Fatalf("observed %d/%d events, want %d each", len(seqsA), len(seqsB), total) + } + for i := range seqsA { + if seqsA[i] != seqsB[i] { + t.Fatalf("seq order diverges at index %d: A=%d B=%d, want identical total order", i, seqsA[i], seqsB[i]) + } + if i > 0 && seqsA[i] <= seqsA[i-1] { + t.Fatalf("seq not strictly increasing at index %d: %d after %d", i, seqsA[i], seqsA[i-1]) + } + } +} + +// seqsOf projects the stream seq off each received frame, for the total-order +// comparison above. +func seqsOf(frames []*compassv1.SubscribeCommsResponse) []uint64 { + seqs := make([]uint64, len(frames)) + for i, f := range frames { + seqs[i] = f.GetSeq() + } + return seqs +} diff --git a/go/internal/delivery/helpers_test.go b/go/internal/delivery/helpers_test.go index 1af028242..ed3989874 100644 --- a/go/internal/delivery/helpers_test.go +++ b/go/internal/delivery/helpers_test.go @@ -33,14 +33,11 @@ const testTimeout = 10 * time.Second // busLagFloodCount is how many messages the bus-lag tests publish to force a // live-buffer overrun: it must exceed the events bus's per-subscriber live-tail -// buffer (events.liveBufferCapacity == events.ringCapacity == 1024) so the -// subscriber's channel latches lagged and closes — the exact condition the -// resync/sweep path under test triggers on. The events caps are unexported, so -// this constant restates the coupling explicitly with margin: if those caps ever -// rise, this must rise past them, or the overrun stops firing and the RIG-2514 -// regression guard silently degrades to a no-op (the tests would still pass -// while guarding nothing). -const busLagFloodCount = 1100 +// buffer (sized to events.RingCapacity) so the subscriber's channel latches +// lagged and closes — the exact condition the resync/sweep path under test +// triggers on. Derived from the exported cap, so raising it cannot silently +// degrade the RIG-2514 regression guard into a no-op that still passes. +const busLagFloodCount = events.RingCapacity + 76 func discardLogger() *slog.Logger { return slog.New(slog.DiscardHandler) }