test(delivery): fix bus-lag test deadlock via non-blocking observation send (RIG-2514)

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,
@@ -572,7 +574,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
 }
 

go/internal/delivery/helpers_test.go

@@ -31,8 +31,40 @@ 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
+// 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 +147,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 +243,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 {