fix(indexer): terminate abandoned GetSubscription streams on reconnect - #1143
Conversation
A stream serving a subscription id now holds an exclusive attachment on the listener: reconnecting with the same id displaces the previous stream instead of leaving it competing for events, and a stream whose listener is removed (unsubscribe or reconnect-timeout expiry) now terminates instead of idling forever. Cleanup on stream exit runs only for the stream that still owns the listener, so a displaced stream can no longer arm the reap timer or remove the listener out from under its successor.
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
WalkthroughThe broker now tracks exclusive listener attachments, ChangesAttachment lifecycle and subscription reconnect handling
Estimated code review effort: 4 (Complex) | ~60 minutes Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
internal/interface/grpc/handlers/indexer.go (1)
528-540: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winDisplaced stream can still consume a buffered event
Atinternal/interface/grpc/handlers/indexer.go:528-540,listener.done/attachment.displacedrace withlistener.chin the sameselect, so a stale stream can win the receive and drop an event that should go to the successor. Move the exit check ahead of the receive path, or otherwise make the receive conditional on the current attachment.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@internal/interface/grpc/handlers/indexer.go` around lines 528 - 540, The event loop in indexer stream handling can still read from listener.ch after listener.done or attachment.displaced is signaled, letting a stale stream consume a buffered event. Update the select/loop in the stream handler so the exit conditions are checked before any receive from listener.ch, or gate the receive on the current attachment state, ensuring only the active stream can consume events.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@internal/interface/grpc/handlers/indexer.go`:
- Around line 528-540: The event loop in indexer stream handling can still read
from listener.ch after listener.done or attachment.displaced is signaled,
letting a stale stream consume a buffered event. Update the select/loop in the
stream handler so the exit conditions are checked before any receive from
listener.ch, or gate the receive on the current attachment state, ensuring only
the active stream can consume events.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 6ba7c4e1-fe8f-424e-b90a-048881bc1ff9
📒 Files selected for processing (4)
internal/interface/grpc/handlers/broker.gointernal/interface/grpc/handlers/broker_test.gointernal/interface/grpc/handlers/indexer.gointernal/interface/grpc/handlers/indexer_test.go
The GetSubscription select loop treats the exit signals (context done, listener removed, displaced by reconnect) and the event channel as peers, so a displaced stream could still win the random select and drain an event that belongs to its successor. Check the exit signals in a non-blocking select before the blocking one, so a displaced stream stops consuming and leaves buffered events on the channel for the stream that took over.
|
Addressed in 2d7307d. The select now checks the exit signals (ctx done / listener removed / displaced) in a non-blocking select before the blocking one, so a displaced stream stops draining Since |
…on selects Remove the freshly-pushed listener when attach fails on the new flow so a failed attach cannot leak it. Add a concurrent attach/detach broker test and document why GetSubscription uses two selects (priority gate plus blocking wait).
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@internal/interface/grpc/handlers/broker_test.go`:
- Around line 561-562: The comment near the release assertion is stale and
contradicts the current test behavior in broker_test.go. Update the inline
comment around the `releaseTrue` assertion in the `Test...` block to describe
the intended idempotency/no-second-release-wins behavior, and keep the
`require.Zero(t, releaseTrue.Load())` assertion unchanged since the second
`release(att)` should always be false. Use the surrounding `release(att)` calls
and `releaseTrue.Load()` as the anchor when editing.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: aaa219aa-22c9-4045-a9e1-c2a3c96021c4
📒 Files selected for processing (5)
internal/interface/grpc/handlers/broker.gointernal/interface/grpc/handlers/broker_test.gointernal/interface/grpc/handlers/indexer.gointernal/interface/grpc/handlers/indexer_test.gointernal/interface/grpc/service.go
🚧 Files skipped from review as they are similar to previous changes (1)
- internal/interface/grpc/handlers/indexer_test.go
c04238e to
3f37a32
Compare
3 participants
GetSubscriptionstreams currently have no tie to their listener's lifecycle. Reconnecting with the same subscription id attaches a second stream to the same listener channel while the previous one stays parked in its select loop: if the server never observes the old client's disconnect (typical behind a load balancer that keeps the connection alive), every reconnect leaks the previous stream, and the leaked streams keep competing for events on the shared channel, so the reconnected client randomly loses events. A stream also survives its own subscription's removal (unsubscribe or reconnect-timeout expiry), since the loop never watcheslistener.done— unlikeGetEventStreamandGetTransactionsStream.This gives the broker an explicit notion of a listener's active consumer:
attachregisters the calling stream as the sole consumer (cancelling any pending reap timeout) and displaces a previously attached stream by closing itsdisplacedchannel;detachreleases it and reports whether the caller still owned the listener, so a displaced stream no longer arms the reap timer or removes the listener out from under its successor.GetSubscription's select now also exits onlistener.doneand on displacement, and the new and old flows share the same attach/detach path. Events buffered on the listener channel survive a reconnect and are delivered to the new stream.One behavioral note: a subscription id now serves exactly one stream at a time, newest wins. Two clients deliberately sharing an id previously got a random split of events (each event went to exactly one of them), which doesn't seem like behavior anyone could rely on; with this change the latest stream takes over and earlier ones end cleanly. Happy to adjust if concurrent consumers per subscription are meant to be supported.
Complements #1142: a max stream lifetime bounds streams whose clients never come back; this change removes the leak on the reconnect path and ties stream lifetime to the subscription itself.
Covered by new handler-level tests (reconnect displaces the previous stream and events flow to the successor only, streams end on listener removal, displaced streams do not reap or remove the listener under the successor) plus broker-level tests for the attach/detach contract.
go test -race ./internal/interface/grpc/handlers/andgolangci-lint run --tests=falseare green.Summary by CodeRabbit