fix(watchevents): a frame that wins the receive select after Close records and reports nothing (BUG-2741) (#1905)

* fix(watchevents): a frame that wins the receive select after Close records and reports nothing (BUG-2741)

* fix(watchevents): a frame after Close does not stamp the liveness clock either (BUG-2741 codex r1)

internal/watchevents/bug2741_closed_receive_test.go

@@ -0,0 +1,133 @@
+package watchevents
+
+import (
+	"testing"
+	"time"
+)
+
+// BUG-2741: Close cancels the receive loop's context AND closes its pubsub,
+// so both select cases can be ready at once and Go picks one at random. When
+// it picks the message, the receive path used to run to completion on a
+// closed bus: rebuilding the replay buffer, rewriting the coverage fields, and
+// REPORTING sequence resets and gaps for a bus with no subscribers left.
+//
+// The race itself is random, so each leg drives the receive-path function
+// directly on a bus Close has finished with: the state a frame that won the
+// select meets. Each leg first shows the SAME call reports on an open bus, so
+// "nothing reported" means the closed check refused it and not that the input
+// was inert. One leg per locked site, because each site needs its own check:
+// fanOutFromRedis and fanOutLocally lock separately, and dropCoverage is also
+// reached from outside the receive loop.
+
+func closedReceiveBus(t *testing.T) (*RedisBus, *recordingObserver) {
+	t.Helper()
+	b := newLocalOnlyBus(16)
+	obs := newRecordingObserver()
+	b.SetObserver(obs)
+	return b, obs
+}
+
+func TestBUG2741_FanOutFromRedisAfterCloseRecordsNothing(t *testing.T) {
+	// PREMISE: on an open bus, a new epoch is a reported reset.
+	open, openObs := closedReceiveBus(t)
+	open.fanOutFromRedis("1", Notification{ID: 1, Kind: "test"}, open.currentGen())
+	open.fanOutFromRedis("2", Notification{ID: 1, Kind: "test"}, open.currentGen())
+	if got := openObs.snapshot(); got.resets[ResetReasonEpochChange] != 1 {
+		t.Fatalf("premise: an epoch change on an open bus reported %+v, want one epoch_change reset", got)
+	}
+	open.Close()
+
+	b, obs := closedReceiveBus(t)
+	b.fanOutFromRedis("1", Notification{ID: 1, Kind: "test"}, b.currentGen())
+	b.Close()
+	before := obs.snapshot()
+
+	b.fanOutFromRedis("2", Notification{ID: 1, Kind: "test"}, b.currentGen())
+
+	if got := obs.snapshot(); got.totalEvents != before.totalEvents {
+		t.Fatalf("a frame after Close was reported: before %+v, after %+v", before, got)
+	}
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if b.epoch != "1" || b.lastAppendedID != 1 {
+		t.Fatalf("a frame after Close rewrote the bus: epoch %q lastAppendedID %d, want 1 / 1", b.epoch, b.lastAppendedID)
+	}
+}
+
+func TestBUG2741_FanOutLocallyAfterCloseRecordsNothing(t *testing.T) {
+	// PREMISE: on an open bus, id 5 after id 1 is a reported gap.
+	open, openObs := closedReceiveBus(t)
+	open.fanOutLocally(Notification{ID: 1, Kind: "test"}, open.currentGen())
+	open.fanOutLocally(Notification{ID: 5, Kind: "test"}, open.currentGen())
+	if got := openObs.snapshot(); got.gaps != 1 {
+		t.Fatalf("premise: a hole on an open bus reported %+v, want one gap", got)
+	}
+	open.Close()
+
+	b, obs := closedReceiveBus(t)
+	b.fanOutLocally(Notification{ID: 1, Kind: "test"}, b.currentGen())
+	b.Close()
+	before := obs.snapshot()
+
+	b.fanOutLocally(Notification{ID: 5, Kind: "test"}, b.currentGen())
+
+	if got := obs.snapshot(); got.totalEvents != before.totalEvents {
+		t.Fatalf("a frame after Close was reported: before %+v, after %+v", before, got)
+	}
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if b.lastAppendedID != 1 {
+		t.Fatalf("a frame after Close was appended: lastAppendedID %d, want 1", b.lastAppendedID)
+	}
+}
+
+func TestBUG2741_DropCoverageAfterCloseRecordsNothing(t *testing.T) {
+	// PREMISE: on an open bus, dropping coverage is a reported reset.
+	open, openObs := closedReceiveBus(t)
+	open.fanOutLocally(Notification{ID: 1, Kind: "test"}, open.currentGen())
+	open.dropCoverage(ResetReasonSubscriptionResumed)
+	if got := openObs.snapshot(); got.resets[ResetReasonSubscriptionResumed] != 1 {
+		t.Fatalf("premise: a coverage drop on an open bus reported %+v, want one subscription_resumed reset", got)
+	}
+	open.Close()
+
+	b, obs := closedReceiveBus(t)
+	b.fanOutLocally(Notification{ID: 1, Kind: "test"}, b.currentGen())
+	b.Close()
+	before := obs.snapshot()
+
+	b.dropCoverage(ResetReasonSubscriptionResumed)
+
+	if got := obs.snapshot(); got.totalEvents != before.totalEvents {
+		t.Fatalf("a coverage drop after Close was reported: before %+v, after %+v", before, got)
+	}
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if b.lastAppendedID != 1 {
+		t.Fatalf("a coverage drop after Close rewrote the bus: lastAppendedID %d, want 1", b.lastAppendedID)
+	}
+}
+
+func TestBUG2741_StampLastSeenAfterCloseRecordsNothing(t *testing.T) {
+	// PREMISE: on an open bus, a received frame stamps the liveness clock.
+	stamp := time.Unix(1_000, 0)
+	open, _ := closedReceiveBus(t)
+	open.nowFunc = func() time.Time { return stamp }
+	open.stampLastSeen(open.currentGen())
+	if !open.lastSeen.Equal(stamp) {
+		t.Fatalf("premise: an open bus did not stamp lastSeen (got %v)", open.lastSeen)
+	}
+	open.Close()
+
+	b, _ := closedReceiveBus(t)
+	b.nowFunc = func() time.Time { return stamp }
+	b.Close()
+
+	b.stampLastSeen(b.currentGen())
+
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if !b.lastSeen.IsZero() {
+		t.Fatalf("a frame after Close stamped lastSeen = %v; a closed bus has no subscription to vouch for", b.lastSeen)
+	}
+}

internal/watchevents/heartbeat.go

@@ -137,7 +137,9 @@ func (b *RedisBus) now() time.Time {
 // answer to an unreadable message; cycling the connection is not.
 func (b *RedisBus) stampLastSeen(gen int64) {
 	b.mu.Lock()
-	if b.subGen == gen {
+	// Not after Close (BUG-2741): a frame that wins the receive loop's select
+	// after shutdown is not evidence that a live subscription is healthy.
+	if !b.closed && b.subGen == gen {
 		b.lastSeen = b.now()
 	}
 	b.mu.Unlock()

internal/watchevents/redis_bus.go

@@ -1502,6 +1502,14 @@ func (b *RedisBus) fanOutFromRedis(epoch string, n Notification, gen int64) {
 	defer func() { b.flush(&pending) }()
 
 	b.mu.Lock()
+	if b.closed {
+		// A frame that won the receive loop's select after Close (BUG-2741):
+		// Close cancels the context AND closes the pubsub, so both cases can be
+		// ready and Go picks one at random. A closed bus has no subscribers to
+		// be honest to, so it records nothing and reports nothing.
+		b.mu.Unlock()
+		return
+	}
 	if b.subGen != gen {
 		// A straggler from a subscription that has already been replaced. It
 		// arrived on a socket this instance has stopped believing, so it is not
@@ -1562,6 +1570,12 @@ func (b *RedisBus) fanOutLocally(n Notification, gen int64) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
+	if b.closed {
+		// Checked here too, under the lock that appends, for the reason the
+		// subGen check below is: fanOutFromRedis checks in an earlier
+		// acquisition, and Close can land between the two (BUG-2741).
+		return
+	}
 	if b.subGen != gen {
 		// A straggler from a replaced subscription. Checked HERE, under the
 		// lock that appends, because fanOutFromRedis's check happens in a
@@ -1815,6 +1829,11 @@ func (b *RedisBus) dropCoverageChecked(reason string, gen *int64) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
+	if b.closed {
+		// A closed bus covers nothing and has no one to signal; a report from
+		// here would be a reset counted after shutdown (BUG-2741).
+		return
+	}
 	if gen != nil && b.subGen != *gen {
 		return
 	}