From 540925cd0bd5895a3322e87abfa55dd5febef5f4 Mon Sep 17 00:00:00 2001 From: xarmian Date: Thu, 8 Oct 2026 16:46:26 -0400 Subject: [PATCH] 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) --- .../bug2741_closed_receive_test.go | 133 ++++++++++++++++++ internal/watchevents/heartbeat.go | 4 +- internal/watchevents/redis_bus.go | 19 +++ 3 files changed, 155 insertions(+), 1 deletion(-) create mode 100644 internal/watchevents/bug2741_closed_receive_test.go diff --git a/internal/watchevents/bug2741_closed_receive_test.go b/internal/watchevents/bug2741_closed_receive_test.go new file mode 100644 index 000000000..c5a35d7b0 --- /dev/null +++ b/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) + } +} diff --git a/internal/watchevents/heartbeat.go b/internal/watchevents/heartbeat.go index 05713fe4b..5d9c711e1 100644 --- a/internal/watchevents/heartbeat.go +++ b/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() diff --git a/internal/watchevents/redis_bus.go b/internal/watchevents/redis_bus.go index 7317bad7c..f846ced4a 100644 --- a/internal/watchevents/redis_bus.go +++ b/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 }