From cc0bfd69609bdc05962c0acc52099c898c2a9422 Mon Sep 17 00:00:00 2001 From: mintaka Date: Sat, 22 Aug 2026 18:38:03 -0400 Subject: [PATCH 1/2] test(delivery): fix bus-lag test deadlock via non-blocking observation send (RIG-2514) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The bus-lag tests (TestBusLagResubscribesAndKeepsDelivering, TestBusLagResubscribeDeliversWindowMessageLive, TestBusLagTriggersSweepNotLoss) flake under cross-test load, hanging the internal/delivery package to the -timeout and reddening CI + the jj-hp pre-push gate ~50% of runs. Reproduced 7/12 on pristine main; each passes in isolation. The test fakes signal each observed dispatch/wake on a buffered recorded channel (cap 1024) with a BLOCKING send. The three tests deliberately flood 1100 > 1024 to force a live-buffer overrun. The recorded fact lands in a mutex-guarded slice before the token send and every waiter re-checks that slice each loop, so once a waiter finds its target it stops draining — the consumer's Run goroutine then wedges on the next blocking send, hanging the package. In isolation the waiter drains fast enough to never saturate; under load the buffer fills and deadlocks. Make the observation send non-blocking at all four identical sites via one documented helper, signalObserved(ch). The token is only a wakeup hint (the fact already lives in the calls slice), so a drop is safe and starvation-free: a drop happens only when the buffer is full, so a blocked waiter always has a token to drain and loop back; when empty the send always lands. Per the no-retries rule this is a determinism fix, not a retry: the observation channel stops exerting backpressure on the code under test. Verified: 20/20 green post-fix (was 7/12 failing); -race suite green; gofmt/vet/golangci-lint clean. Closes RIG-2514. Co-authored-by: Matt Wilkinson --- go/internal/delivery/consumer_test.go | 2 +- go/internal/delivery/helpers_test.go | 25 +++++++++++++++++++++++-- 2 files changed, 24 insertions(+), 3 deletions(-) diff --git a/go/internal/delivery/consumer_test.go b/go/internal/delivery/consumer_test.go index e7c38693..e89a07be 100644 --- a/go/internal/delivery/consumer_test.go +++ b/go/internal/delivery/consumer_test.go @@ -572,7 +572,7 @@ func (d *blockCapturingDispatcher) DispatchControl(_ context.Context, sessionID d.mu.Lock() d.calls = append(d.calls, blockRecord{sessionID: sessionID, messageID: msg.GetId(), firstText: first}) d.mu.Unlock() - d.recorded <- struct{}{} + signalObserved(d.recorded) return nil } diff --git a/go/internal/delivery/helpers_test.go b/go/internal/delivery/helpers_test.go index 59e9a60a..ebf0d26d 100644 --- a/go/internal/delivery/helpers_test.go +++ b/go/internal/delivery/helpers_test.go @@ -33,6 +33,27 @@ const testTimeout = 10 * time.Second func discardLogger() *slog.Logger { return slog.New(slog.DiscardHandler) } +// signalObserved does a NON-BLOCKING send of a per-call token on a test +// observation channel (the recorded/wake signals below). The token only wakes a +// waiter (waitForMessage / waitForDispatches / waitForWakes / waitFor) to +// re-check the fake's recorded set; the recorded FACT already lives in the fake's +// mutex-guarded calls slice BEFORE this send, so the token is a wakeup hint, +// never the source of truth. The send must not block: a blocking send turns the +// observation channel into backpressure on the code under test, so a test that +// produces more dispatches than the buffer holds (the bus-lag floods publish +// 1100 past the 1024-token buffer) wedges the consumer's Run goroutine on a full +// channel the moment a waiter stops draining — a deadlock that passes in +// isolation but hangs the whole package to the -timeout under cross-test load +// (RIG-2514). Dropping a token is safe: a drop happens only when the buffer is +// full (hence non-empty), so a blocked waiter still has a token to drain and loop +// back to re-check the snapshot; when the buffer is empty the send always lands. +func signalObserved(ch chan struct{}) { + select { + case ch <- struct{}{}: + default: + } +} + // opKind distinguishes a deliver op from a steer op in a recorded dispatch, so a // mention-routing test can assert the mentioned agent got a STEER and a plain // subscriber got a DELIVER. @@ -115,7 +136,7 @@ func (d *fakeDispatcher) DispatchControl(_ context.Context, sessionID string, op kind, messageID := classifyOp(op) d.calls = append(d.calls, dispatchRecord{sessionID: sessionID, messageID: messageID, kind: kind}) d.mu.Unlock() - d.recorded <- struct{}{} + signalObserved(d.recorded) return nil } @@ -211,7 +232,7 @@ func (w *fakeWaker) WakeAgent(_ context.Context, agent store.AccountID) { if hook != nil { hook(agent) } - w.recorded <- struct{}{} + signalObserved(w.recorded) } func (w *fakeWaker) count(agent store.AccountID) int { From 4d4fbc0f8f8dbc315fb6efc5d7de6f8f7f93609e Mon Sep 17 00:00:00 2001 From: mintaka Date: Sat, 22 Aug 2026 18:49:23 -0400 Subject: [PATCH 2/2] test(delivery): name the bus-lag flood count and its live-buffer coupling (RIG-2514 review low) The three bus-lag tests publish a hardcoded 1100 to overrun the events bus's per-subscriber live-tail buffer (liveBufferCapacity == ringCapacity == 1024), with the two magic numbers sitting in different files and no explicit tie. If the events caps ever rose past 1100 without the flood count following, the overrun would stop firing and the RIG-2514 regression guard would silently degrade to a no-op while the tests kept passing. Extract the flood count to a documented busLagFloodCount constant that states the must-exceed-liveBufferCapacity invariant, and derive all three flood loops from it. No behavior change: still 1100 > 1024, guard still triggers (15/15 green). Addresses the review low on PR #501. Co-authored-by: Matt Wilkinson --- go/internal/delivery/consumer_test.go | 16 +++++++++------- go/internal/delivery/helpers_test.go | 11 +++++++++++ 2 files changed, 20 insertions(+), 7 deletions(-) diff --git a/go/internal/delivery/consumer_test.go b/go/internal/delivery/consumer_test.go index e89a07be..81a1337d 100644 --- a/go/internal/delivery/consumer_test.go +++ b/go/internal/delivery/consumer_test.go @@ -350,9 +350,9 @@ func TestBusLagTriggersSweepNotLoss(t *testing.T) { c.bus.Publish(postedResponse(wireText("m0", author, "first"))) <-disp.enteredFirst - // Overrun the consumer's live buffer (capacity 1024) so its channel closes - // lagged. Publish comfortably past it. - for range 1100 { + // Overrun the consumer's live buffer (busLagFloodCount > liveBufferCapacity) + // so its channel closes lagged. + for range busLagFloodCount { c.bus.Publish(postedResponse(wireText("flood", author, "x"))) } @@ -436,8 +436,9 @@ func TestBusLagResubscribesAndKeepsDelivering(t *testing.T) { c.bus.Publish(postedResponse(wireText("m0", author, "first"))) <-disp.enteredFirst - // Overrun the live buffer (capacity 1024) so the channel closes lagged. - for range 1100 { + // Overrun the live buffer (busLagFloodCount > liveBufferCapacity) so the + // channel closes lagged. + for range busLagFloodCount { c.bus.Publish(postedResponse(wireText("flood", author, "x"))) } // Release the stall: the consumer drains its buffer, reads the lagged-closed @@ -510,8 +511,9 @@ func TestBusLagResubscribeDeliversWindowMessageLive(t *testing.T) { c.bus.Publish(postedResponse(wireText("m0", author, "first"))) <-disp.enteredFirst - // Overrun the live buffer (capacity 1024) so the channel closes lagged. - for range 1100 { + // Overrun the live buffer (busLagFloodCount > liveBufferCapacity) so the + // channel closes lagged. + for range busLagFloodCount { c.bus.Publish(postedResponse(wireText("flood", author, "x"))) } // The window publish must happen once the consumer is inside the resync sweep, diff --git a/go/internal/delivery/helpers_test.go b/go/internal/delivery/helpers_test.go index ebf0d26d..405b6604 100644 --- a/go/internal/delivery/helpers_test.go +++ b/go/internal/delivery/helpers_test.go @@ -31,6 +31,17 @@ import ( // device: tests gate on the recorder's observed dispatch count, not elapsed time. 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 + func discardLogger() *slog.Logger { return slog.New(slog.DiscardHandler) } // signalObserved does a NON-BLOCKING send of a per-call token on a test