test(events): hold the SUBSCRIBE ack at the wire until the timer arm is taken (BUG-2786)

internal/events/redis_bus.go

@@ -571,16 +571,28 @@ type RedisBus struct {
 	afterSubscriptionConfirmed func()
 
 	// beforeUnconfirmedMark is a TEST SEAM, nil in production. It runs in
-	// markUnconfirmedAdmission BEFORE it takes b.mu, which is the only way to
-	// make the timer-versus-acknowledgement race deterministic: a hook that
-	// waits for the acknowledgement here reproduces, every time, the interleave
-	// where the select chose the timer but confirmSubscription has already run.
+	// markUnconfirmedAdmission BEFORE it takes b.mu, which is half of what
+	// makes the timer-versus-acknowledgement race deterministic: a hook that
+	// waits for the acknowledgement here reproduces the interleave where the
+	// select chose the timer but confirmSubscription has already run.
+	//
+	// THE OTHER HALF IS THAT THE HOOK ONLY RUNS ON THE TIMER ARM (BUG-2786).
+	// If the acknowledgement has already landed when the select runs, both
+	// arms are ready and the runtime picks one at random — and on the
+	// confirmed arm this seam is never reached, so a test hooked here sees
+	// nothing. A test that depends on it must also keep the acknowledgement
+	// from arriving until the hook has run (the gated SUBSCRIBE proxy in
+	// redis_subscribe_confirm_test.go does that at the wire, bounded by a
+	// failsafe the test checks before trusting the ordering); "every time"
+	// is only true under that condition.
 	//
 	// REPETITION DOES NOT SUBSTITUTE FOR IT, measured rather than assumed.
 	// Against the mutation that removes the confirmClosed re-check, 500
 	// establishments per run caught it in 0 of 10 runs — a near-zero bound
 	// makes the timer win OUTRIGHT far more often than it ties, and winning
-	// outright is the ordinary timeout path. With this seam: 10 of 10.
+	// outright is the ordinary timeout path. With this seam AND the
+	// acknowledgement held until it has run: 10 of 10, unloaded and under
+	// load alike (re-measured for BUG-2786).
 	beforeUnconfirmedMark func()
 
 	// afterRegisterBeforeEstablish is a TEST SEAM, nil in production. It runs

internal/events/redis_heartbeat_test.go

@@ -708,9 +708,11 @@ func TestAFreshlySubscribedWorkspaceIsNotCycled(t *testing.T) {
 // exactly the workspaces already having a bad time.
 func TestAnUnconfirmedAdmissionIsNotCycledAsIdle(t *testing.T) {
 	mr := miniredis.RunT(t)
-	// Park the SUBSCRIBE in flight for longer than the bus is willing to wait,
-	// so the caller is admitted with no acknowledgement behind it.
-	proxy := newSubscribeDelayProxy(t, mr.Addr(), 2*time.Second)
+	// Park the SUBSCRIBE in flight until this test is over (the gate proxy
+	// releases at cleanup), so the caller is admitted with no acknowledgement
+	// behind it — held at the wire rather than for a fixed interval, so the
+	// timer arm is the only one that can win (BUG-2786).
+	proxy := newSubscribeGateProxy(t, mr.Addr())
 	client := redis.NewClient(&redis.Options{Addr: proxy.addr()})
 	t.Cleanup(func() { _ = client.Close() })
 
@@ -733,8 +735,9 @@ func TestAnUnconfirmedAdmissionIsNotCycledAsIdle(t *testing.T) {
 	}
 	defer b.Unsubscribe(ch)
 
-	if proxy.held.Load() == 0 {
-		t.Fatal("the SUBSCRIBE was never parked; this test could not have discriminated")
+	proxy.waitParked(t)
+	if proxy.forcedOpen.Load() {
+		t.Fatal("the failsafe opened the gate before the test released it; the acknowledgement was not held by construction, so this test could not have discriminated")
 	}
 	if obs.unconfirmedCount() == 0 {
 		t.Fatal("the admission was confirmed after all; this test is not exercising the unconfirmed case")

internal/events/redis_subscribe_cancel_test.go

@@ -177,16 +177,21 @@ func TestCancelDuringConfirmationTearsDownWhenNobodyElseIsWaiting(t *testing.T)
 // everything settled would pass this test while proving nothing.
 func TestCancelledEstablisherStillFinishesTheWaitForItsJoiners(t *testing.T) {
 	mr := miniredis.RunT(t)
-	// Parked for longer than the confirmation bound below, so the hand-off
-	// runs into the bound and must mark the admission unconfirmed — the state
-	// that later becomes the joiner's sync_required.
-	proxy := newSubscribeDelayProxy(t, mr.Addr(), 3*time.Second)
+	// Parked until the hand-off has run into the confirmation bound and marked
+	// the admission unconfirmed — the state that later becomes the joiner's
+	// sync_required. Held at the wire and released only once the observer
+	// has reported that mark (below), rather than for a fixed interval, so
+	// the timer arm is the only one that can win and the acknowledgement is
+	// late by construction (BUG-2786).
+	proxy := newSubscribeGateProxy(t, mr.Addr())
 	client := redis.NewClient(&redis.Options{Addr: proxy.addr()})
 	t.Cleanup(func() { _ = client.Close() })
 
 	b := NewRedisBus(client)
 	t.Cleanup(b.Close)
 	b.confirmTimeout = 200 * time.Millisecond
+	obs := &recordingObserver{}
+	b.SetObserver(obs)
 
 	unconfirmed := make(chan struct{}, 1)
 	b.beforeUnconfirmedMark = func() {
@@ -254,6 +259,21 @@ func TestCancelledEstablisherStillFinishesTheWaitForItsJoiners(t *testing.T) {
 	case <-time.After(5 * time.Second):
 		t.Fatal("the confirmation wait was abandoned with the establisher: the joiner was admitted into an unacknowledged subscription and told nothing (BUG-2747's defect, re-created at the seam)")
 	}
+	// The hook fires BEFORE the mark takes b.mu; the observer reports AFTER
+	// it is set. Release the acknowledgement only on the second signal, or it
+	// can beat the mark and turn it into a no-op (codex round 3 P1).
+	deadline := time.Now().Add(5 * time.Second)
+	for obs.unconfirmedCount() == 0 {
+		if time.Now().After(deadline) {
+			t.Fatal("the hook ran but the unconfirmed mark was never reported; this test could not have discriminated")
+		}
+		time.Sleep(time.Millisecond)
+	}
+	proxy.waitParked(t)
+	if proxy.forcedOpen.Load() {
+		t.Fatal("the failsafe opened the gate before the test released it; the acknowledgement was not held by construction, so this test could not have discriminated")
+	}
+	proxy.release()
 
 	// And it reaches the joiner as a reconcile signal once the acknowledgement
 	// finally lands.

internal/events/redis_subscribe_confirm_test.go

@@ -36,15 +36,34 @@ type subscribeDelayProxy struct {
 	// held counts SUBSCRIBE commands actually parked, so a test can assert its
 	// own premise rather than passing because the instrument never armed.
 	held atomic.Int64
+
+	// gate, when non-nil, replaces delay: a parked SUBSCRIBE is released when
+	// the channel is closed rather than after a fixed sleep (see
+	// newSubscribeGateProxy). forcedOpen records that the failsafe opened it
+	// instead of the test, so the test can name that as its own distinct
+	// failure rather than reading it as a pass.
+	gate        chan struct{}
+	releaseOnce sync.Once
+	forcedOpen  atomic.Bool
 }
 
 func newSubscribeDelayProxy(t *testing.T, backend string, delay time.Duration) *subscribeDelayProxy {
+	t.Helper()
+	return newSubscribeProxy(t, backend, delay, nil)
+}
+
+// newSubscribeProxy is the constructor behind both proxies. Every field the
+// forward goroutine reads is set BEFORE the accept loop starts (codex round
+// 5): a gate assigned after the goroutine is running is published to it
+// without synchronisation, and a stale nil there takes the delay path with
+// delay=0 — the gate silently not a gate.
+func newSubscribeProxy(t *testing.T, backend string, delay time.Duration, gate chan struct{}) *subscribeDelayProxy {
 	t.Helper()
 	ln, err := net.Listen("tcp", "127.0.0.1:0")
 	if err != nil {
 		t.Fatalf("listen: %v", err)
 	}
-	p := &subscribeDelayProxy{ln: ln, delay: delay}
+	p := &subscribeDelayProxy{ln: ln, delay: delay, gate: gate}
 	t.Cleanup(func() { _ = ln.Close() })
 
 	go func() {
@@ -67,6 +86,76 @@ func newSubscribeDelayProxy(t *testing.T, backend string, delay time.Duration) *
 
 func (p *subscribeDelayProxy) addr() string { return p.ln.Addr().String() }
 
+// subscribeGateFailsafe bounds how long a gated SUBSCRIBE stays parked when
+// nothing releases it. It exists so a test whose release never fires ends
+// bounded instead of at the package timeout; it must NOT be short enough to
+// open the gate on a slow but correct run, or it becomes a second way for
+// the test to fail that looks like the first. 20s is four orders of
+// magnitude past the loaded ack latency measured for BUG-2786 (sub-ms) and
+// past every wait the gated tests contain. It is NOT self-reporting: a test
+// that relies on the gate must check forcedOpen before it trusts the
+// ordering the gate was holding (codex round 4), or a failsafe release
+// shows up as an unrelated assertion failing. When it fires it opens the
+// gate for every connection, so the test records ONE forced open rather
+// than one per connection.
+const subscribeGateFailsafe = 20 * time.Second
+
+// newSubscribeGateProxy is the delay proxy with the release under the TEST'S
+// control: the first SUBSCRIBE per connection is parked until release() is
+// called — or the failsafe fires, which forcedOpen records — not for a fixed
+// interval.
+//
+// WHY A SIGNAL AND NOT A DURATION (BUG-2786). A test that needs the
+// establisher to take the TIMER arm of its confirmation select cannot get
+// there by making the bound tiny: a 1ns bound only guarantees the timer is
+// ready, not that the acknowledgement is NOT. Under load the receive goroutine
+// lands the ack before the establisher reaches its select, both arms are
+// ready, Go picks uniformly at random, and half the time the timer arm — and
+// everything the test hooked onto it — is skipped. The measured shape was 10
+// of 1000 loaded runs, every one of them the hook never running. Holding the
+// ack at the wire until the timer arm has been ENTERED makes the interleave a
+// construction instead of a coin toss.
+func newSubscribeGateProxy(t *testing.T, backend string) *subscribeDelayProxy {
+	t.Helper()
+	p := newSubscribeProxy(t, backend, 0, make(chan struct{}))
+	// A test that ends without releasing leaves forward parked on the gate,
+	// holding both sides of the connection until the failsafe (codex round
+	// 3). Release at cleanup so teardown is prompt whichever way the test
+	// went.
+	t.Cleanup(p.release)
+	return p
+}
+
+// release lets every parked and future SUBSCRIBE through. Idempotent and
+// safe to call concurrently — the hook and the failsafe, or two connections'
+// failsafes, can race to be first (codex round 2).
+func (p *subscribeDelayProxy) release() {
+	p.releaseOnce.Do(func() { close(p.gate) })
+}
+
+// waitParked blocks until the proxy has parked a SUBSCRIBE, and fails the
+// test if none arrives. A one-shot read of held is NOT a barrier (codex
+// round 4): the client's write returns once the bytes are in the kernel,
+// and under load the proxy's forward goroutine may not have read them yet
+// when the test looks — so a plain `held == 0` premise check fails
+// spuriously, which is the family of flake this instrument exists to end.
+//
+// What it proves is exactly that the proxy recognised and parked one
+// command; it does not say WHY none was, and the causes it cannot tell
+// apart (the marker split across two reads, a dial that never reached the
+// proxy) all end the same way for the test — no held acknowledgement, so no
+// discrimination — which is what its message says.
+func (p *subscribeDelayProxy) waitParked(t *testing.T) {
+	t.Helper()
+	deadline := time.Now().Add(3 * time.Second)
+	for p.held.Load() == 0 {
+		if time.Now().After(deadline) {
+			t.Fatal("the SUBSCRIBE was never parked; this test could not have discriminated")
+		}
+		time.Sleep(time.Millisecond)
+	}
+}
+
 // forward copies client → server, parking the first chunk that carries a
 // SUBSCRIBE.
 func (p *subscribeDelayProxy) forward(client, server net.Conn) {
@@ -80,7 +169,22 @@ func (p *subscribeDelayProxy) forward(client, server net.Conn) {
 			if !parked && bytes.Contains(bytes.ToLower(chunk), []byte("\r\n$9\r\nsubscribe\r\n")) {
 				parked = true
 				p.held.Add(1)
-				time.Sleep(p.delay)
+				if p.gate != nil {
+					select {
+					case <-p.gate:
+					case <-time.After(subscribeGateFailsafe):
+						// OPENS THE GATE FOR EVERYONE, not just this chunk
+						// (codex round 1 P2): a later connection — a
+						// reconnect, a second workspace — would otherwise
+						// park for another failsafe interval, and forcedOpen
+						// would then be reporting a cause that belongs to
+						// the first connection alone.
+						p.forcedOpen.Store(true)
+						p.release()
+					}
+				} else {
+					time.Sleep(p.delay)
+				}
 			}
 			if _, werr := server.Write(chunk); werr != nil {
 				return
@@ -289,15 +393,17 @@ func TestSubscribeDoesNotHoldTheBusLockAcrossTheRedisDial(t *testing.T) {
 // current, so the acknowledgement carries the mid-stream signal BUG-2730 built.
 func TestAnUnconfirmedAdmissionTellsItsSubscribersWhenTheAckLands(t *testing.T) {
 	mr := miniredis.RunT(t)
-	proxy := newSubscribeDelayProxy(t, mr.Addr(), 400*time.Millisecond)
+	// GATED, NOT DELAYED (BUG-2786): the acknowledgement is held until the
+	// wait has given up, so the timer arm is the only one that can win —
+	// a fixed delay only made that likely, by a margin CI load can eat.
+	proxy := newSubscribeGateProxy(t, mr.Addr())
 	client := redis.NewClient(&redis.Options{Addr: proxy.addr()})
 	t.Cleanup(func() { _ = client.Close() })
 
 	b := NewRedisBus(client)
 	t.Cleanup(b.Close)
 	obs := &recordingObserver{}
 	b.SetObserver(obs)
-	// Shorter than the delay, so the wait gives up first.
 	b.confirmTimeout = 50 * time.Millisecond
 
 	ch, gaps, outcome := b.SubscribeIfAllowed(context.Background(), "ws-1", 0)
@@ -306,10 +412,21 @@ func TestAnUnconfirmedAdmissionTellsItsSubscribersWhenTheAckLands(t *testing.T)
 	}
 	defer b.Unsubscribe(ch)
 
+	proxy.waitParked(t)
+	if proxy.forcedOpen.Load() {
+		t.Fatal("the failsafe opened the gate before the test released it; the acknowledgement was not held by construction, so this test could not have discriminated")
+	}
 	if got := obs.unconfirmedCount(); got != 1 {
 		t.Fatalf("SubscriptionUnconfirmed reported %d times, want 1", got)
 	}
 
+	// RELEASED ONLY NOW, after the mark has been observed above — not from
+	// the beforeUnconfirmedMark hook, which runs before the mark takes b.mu
+	// and would let the acknowledgement beat it (codex round 3 P1). With
+	// Subscribe returned and the count at 1, the acknowledgement is late by
+	// construction rather than by margin.
+	proxy.release()
+
 	// The acknowledgement lands late; the subscriber is told to reconcile.
 	select {
 	case <-gaps:
@@ -318,7 +435,10 @@ func TestAnUnconfirmedAdmissionTellsItsSubscribersWhenTheAckLands(t *testing.T)
 	}
 
 	// Control: the same signal does not fire for a subscription that was
-	// confirmed before admission.
+	// confirmed before admission. The gate is open now, so ws-2 is an
+	// ordinary immediate confirmation — which is the property the control
+	// is about; the fixed-delay proxy this test used to run on made it a
+	// slow one as a side effect, and nothing here depended on that.
 	b.confirmTimeout = defaultSubscribeConfirmTimeout
 	before := obs.unconfirmedCount()
 	ch2, gaps2, _ := b.SubscribeIfAllowed(context.Background(), "ws-2", 0)
@@ -584,22 +704,39 @@ func TestASubscriberArrivingMidEstablishmentWaitsForTheAcknowledgement(t *testin
 // path and proves nothing. Measured against the mutation: 500 establishments per
 // run caught it in 0 of 10 runs. A one-in-ten detector reads as coverage and
 // is not.
+//
+// AND THE ACKNOWLEDGEMENT IS HELD AT THE WIRE UNTIL THE TIMER ARM IS TAKEN
+// (BUG-2786), because the seam alone was not enough. The hook runs only on the
+// timer arm, and a 1ns bound guarantees that arm is ready — not that the
+// acknowledgement is not. Under CI load the receive goroutine consumed the
+// SUBSCRIBE reply before the establisher reached its select, both arms were
+// ready, Go chose at random, and on the confirmed arm the hook never ran: the
+// guard below then reported the acknowledgement as never landing when it had
+// landed too EARLY for the interleave to exist. Measured before the gate: 10
+// of 1000 loaded runs (16 busy loops, GOMAXPROCS=2, -race), every one the hook
+// never having run; zero were a stalled acknowledgement. With the reply parked
+// until the hook releases it, only the timer arm can be ready at the select,
+// and the guard has exactly one cause left.
 func TestAConfirmedSubscriptionIsNeverLeftMarkedUnconfirmed(t *testing.T) {
 	mr := miniredis.RunT(t)
-	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+	proxy := newSubscribeGateProxy(t, mr.Addr())
+	client := redis.NewClient(&redis.Options{Addr: proxy.addr()})
 	t.Cleanup(func() { _ = client.Close() })
 
 	b := NewRedisBus(client)
 	t.Cleanup(b.Close)
 	obs := &recordingObserver{}
 	b.SetObserver(obs)
-	// Expires immediately, so the timer branch is the one taken.
+	// Expires immediately, so the timer branch is the one taken — and, with
+	// the acknowledgement parked at the proxy, the ONLY one that can be.
 	b.confirmTimeout = time.Nanosecond
 
-	// ...and then hold the mark until the acknowledgement has definitively
-	// landed, which is the interleave under test.
-	var raced atomic.Bool
+	// Now let the acknowledgement through, and hold the mark until it has
+	// definitively landed: that is the interleave under test.
+	var raced, hookRan atomic.Bool
 	b.beforeUnconfirmedMark = func() {
+		hookRan.Store(true)
+		proxy.release()
 		deadline := time.Now().Add(3 * time.Second)
 		for time.Now().Before(deadline) {
 			b.mu.Lock()
@@ -617,10 +754,22 @@ func TestAConfirmedSubscriptionIsNeverLeftMarkedUnconfirmed(t *testing.T) {
 	ch, gaps, _ := b.Subscribe(context.Background(), "ws-1")
 	defer b.Unsubscribe(ch)
 
-	// PREMISE: the acknowledgement really did beat the mark, or the interleave
-	// under test never happened.
+	// PREMISES, IN AN ORDER WHERE EACH GUARD'S CAUSE IS WHAT REMAINS (codex
+	// round 7): the proxy parked the SUBSCRIBE at all; the failsafe did not
+	// open it before the test released it; given both, the only way the hook
+	// did not run is that the timer arm was not taken (this is the cause the
+	// old guard misreported); and given all three, the only way raced is
+	// false is that the released acknowledgement never landed within the
+	// wait.
+	proxy.waitParked(t)
+	if proxy.forcedOpen.Load() {
+		t.Fatal("the failsafe opened the gate before the test released it; the acknowledgement was not held by construction, so this test could not have discriminated")
+	}
+	if !hookRan.Load() {
+		t.Fatal("the timer arm was not taken, so the hook never ran and the interleave under test was never constructed")
+	}
 	if !raced.Load() {
-		t.Fatal("the acknowledgement never landed before the mark; this test could not have discriminated")
+		t.Fatal("the acknowledgement was released before the mark but never landed within the wait; this test could not have discriminated")
 	}
 
 	b.mu.Lock()