fix(llc): Ignore events dispatched after the client is disposed - #2855
Conversation
📝 WalkthroughWalkthroughThe client ignores events after disposal and safely enqueues events during reconnect recovery. A regression test covers delayed recovery after disposal, and the changelog records the fix. ChangesClient disposal handling
Estimated code review effort: 2 (Simple) | ~10 minutes 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 |
There was a problem hiding this comment.
We should also use .safeAdd here and in all the cases where we are using .add in client.dart.
| return _eventController.safeAdd(event); |
There was a problem hiding this comment.
Fixing it! Do you think it makes sense to still keep the isClosed guard at the beginning of the handleEvent method? If we keep it - we still guard agains writes to state.updateUser and handling health.check.
There was a problem hiding this comment.
Yes, makes sense to block the other two functions
There was a problem hiding this comment.
we should also add the .safeAdd to remaining subjects in the client.dart file
There was a problem hiding this comment.
This seems to be the only remaining unsafe usage - every similar controller already uses .safeAdd.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
packages/stream_chat/lib/src/client/client.dart (1)
578-587: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winGuard the whole recovery branch against disposed clients.
handleEventonly protects status/health-check events, but_queryChannelsOnlineImpl()still appends users, submits delivery receipts, saves queries, and adds channels afterconnectUser → disconnectUser → connectUser. That whole tail can still run against a torn-downstate,channelDeliveryReporter, andchatPersistenceClientafter an in-flight query resolves. Check disposal state, e.g._eventController.isClosed, right after eachawaitin_onConnectionStatusChangedbefore continuing the recovery path.🤖 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 `@packages/stream_chat/lib/src/client/client.dart` around lines 578 - 587, Guard the recovery flow in _onConnectionStatusChanged by checking _eventController.isClosed immediately after each await before continuing. Return early when disposed so _queryChannelsOnlineImpl and subsequent user updates, delivery receipts, persistence, and channel additions do not access torn-down state; leave the existing handleEvent guard unchanged.
🤖 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 `@packages/stream_chat/test/src/client/client_test.dart`:
- Around line 5457-5502: Add a companion test near “disposing mid-recovery does
not surface a late recovery event” that completes the pending channel query
after client.dispose() with at least one channel. Ensure the non-empty recovery
path executes state.addChannels, channelDeliveryReporter.submitForDelivery, and
chatPersistenceClient?.saveChannelQueries, then verify the post-disposal
completion is discarded silently without an unhandled async error.
---
Outside diff comments:
In `@packages/stream_chat/lib/src/client/client.dart`:
- Around line 578-587: Guard the recovery flow in _onConnectionStatusChanged by
checking _eventController.isClosed immediately after each await before
continuing. Return early when disposed so _queryChannelsOnlineImpl and
subsequent user updates, delivery receipts, persistence, and channel additions
do not access torn-down state; leave the existing handleEvent guard unchanged.
🪄 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: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 8567bd6e-3bf7-434e-9a06-2cdd3237d98c
📒 Files selected for processing (3)
packages/stream_chat/CHANGELOG.mdpackages/stream_chat/lib/src/client/client.dartpackages/stream_chat/test/src/client/client_test.dart
| test('disposing mid-recovery does not surface a late recovery event', () async { | ||
| // Keep the recovery's channel query pending so the client is still | ||
| // mid-recovery at the moment it is disposed. | ||
| final pendingQuery = Completer<QueryChannelsResponse>(); | ||
| when( | ||
| () => api.channel.queryChannels( | ||
| filter: any(named: 'filter'), | ||
| sort: any(named: 'sort'), | ||
| state: any(named: 'state'), | ||
| watch: any(named: 'watch'), | ||
| presence: any(named: 'presence'), | ||
| memberLimit: any(named: 'memberLimit'), | ||
| messageLimit: any(named: 'messageLimit'), | ||
| paginationParams: any(named: 'paginationParams'), | ||
| ), | ||
| ).thenAnswer((_) => pendingQuery.future); | ||
|
|
||
| client = StreamChatClient(apiKey, chatApi: api, ws: ws); | ||
| await client.connectUser(user, token); | ||
| await delay(300); | ||
|
|
||
| // Track a channel so reconnecting triggers channel recovery, which then | ||
| // blocks on the pending query above. | ||
| final channel = Channel.fromState( | ||
| client, | ||
| ChannelState(channel: ChannelModel(cid: 'messaging:c1')), | ||
| ); | ||
| client.state.addChannels({'messaging:c1': channel}); | ||
|
|
||
| // Drop then restore the connection to start a reconnect recovery. | ||
| ws.connectionStatus = ConnectionStatus.disconnected; | ||
| await delay(100); | ||
| ws.connectionStatus = ConnectionStatus.connected; | ||
| await delay(100); | ||
|
|
||
| // Dispose while the recovery is still in flight. | ||
| await client.dispose(); | ||
| disposed = true; | ||
|
|
||
| // Let the now-orphaned recovery finish. Its trailing work must be | ||
| // discarded silently instead of thrown as an unhandled async error. | ||
| pendingQuery.complete(QueryChannelsResponse()..channels = []); | ||
| await delay(300); | ||
|
|
||
| expect(client.wsConnectionStatus, ConnectionStatus.disconnected); | ||
| }); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "Locate relevant files"
fd -a 'client_test.dart|client.dart' . | sed 's#^\./##' | head -100
echo
echo "Find references to _queryChannelsOnlineImpl and stubbed test"
rg -n "_queryChannelsOnlineImpl|disposing mid-recovery|queryChannels\(|addChannels|submitForDelivery|saveChannelQueries" packages -S || true
echo
echo "Show test lines around comment"
sed -n '5430,5515p' packages/stream_chat/test/src/client/client_test.dart 2>/dev/null || trueRepository: GetStream/stream-chat-flutter
Length of output: 20656
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "Locate relevant files"
fd -a 'client_test.dart|client.dart' . | sed 's#^\./##' | head -100
echo
echo "Find references to _queryChannelsOnlineImpl and stubbed test"
rg -n "_queryChannelsOnlineImpl|disposing mid-recovery|queryChannels\(Repository: GetStream/stream-chat-flutter
Length of output: 981
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "Inspect _queryChannelsOnlineImpl"
sed -n '860,970p' packages/stream_chat/lib/src/client/client.dart
echo
echo "Inspect fake persistence submitForDelivery state"
sed -n '170,215p' packages/stream_chat/test/src/fakes.dart
echo
echo "Inspect fake queryChannels implementation"
sed -n '340,430p' packages/stream_chat/test/src/fakes.dart
echo
echo "Behavioural/static verifier: check whether empty query returns before side effects"
python3 - <<'PY'
from pathlib import Path
text = Path('packages/stream_chat/lib/src/client/client.dart').read_text()
start = text.index('Future<QueryChannelsResult> _queryChannelsOnlineImpl')
end = text.index('\n `@Deprecated`(', start)
method = text[start:end]
print('_queryChannelsOnlineImpl length:', len(method))
for label in [
'final result = await _chatApi.channel.queryChannels',
'if (result.channels.isEmpty)',
'await channelDeliveryReporter.submitForDelivery',
'await chatPersistenceClient?.saveChannelQueries',
'this.state.addChannels',
]:
idx = method.find(label)
print(f'{label}: found={idx != -1!s}')
print('non-empty result path: addChannels inside result.channels.isNotEmpty block:', 'result.channels.isNotEmpty' in method)
PYRepository: GetStream/stream-chat-flutter
Length of output: 5300
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "Search fake query channel implementation"
rg -n "class FakeChatApi|fake.*queryChannels|queryChannels.*Completer|QueryChannelsResponse" packages/stream_chat/test/src/fakes.dart packages/stream_chat/test/src -S || true
echo
echo "Show _queryChannelsOnlineImpl early exit region with line numbers"
nl -ba packages/stream_chat/lib/src/client/client.dart | sed -n '866,978p'
echo
echo "Show related recovery stub setup lines"
nl -ba packages/stream_chat/lib/src/client/client.dart | sed -n '2898,2918p'
nl -ba packages/stream_chat/test/src/client/client_test.dart | sed -n '5240,5305p'Repository: GetStream/stream-chat-flutter
Length of output: 3023
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "FakeChatApi body"
sed -n '104,145p' packages/stream_chat/test/src/fakes.dart
echo
echo "Relevant recovery stub and disposal test"
sed -n '5240,5265p' packages/stream_chat/lib/src/client/client.dart
sed -n '5457,5505p' packages/stream_chat/test/src/client/client_test.dartRepository: GetStream/stream-chat-flutter
Length of output: 2963
Cover non-empty channel recovery after disposal
Recovering with channels = [] triggers the early return in _queryChannelsOnlineImpl, so it does not exercise the post-disposal paths for state.addChannels, channelDeliveryReporter.submitForDelivery, or chatPersistenceClient?.saveChannelQueries. Add a companion case where the pending query resolves with one or more channels after client.dispose().
🤖 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 `@packages/stream_chat/test/src/client/client_test.dart` around lines 5457 -
5502, Add a companion test near “disposing mid-recovery does not surface a late
recovery event” that completes the pending channel query after client.dispose()
with at least one channel. Ensure the non-empty recovery path executes
state.addChannels, channelDeliveryReporter.submitForDelivery, and
chatPersistenceClient?.saveChannelQueries, then verify the post-disposal
completion is discarded silently without an unhandled async error.
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #2855 +/- ##
=======================================
Coverage 72.83% 72.83%
=======================================
Files 428 428
Lines 27657 27658 +1
=======================================
+ Hits 20144 20145 +1
Misses 7513 7513 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
05a66ed to
8cd24c7
Compare
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
8cd24c7 to
b754761
Compare
There was a problem hiding this comment.
♻️ Duplicate comments (1)
packages/stream_chat/test/src/client/client_test.dart (1)
5428-5503: 🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy liftTest still uses
channels: [], so it doesn't exercise the risky post-disposal write path.This was already flagged on a prior commit of this test: recovering with an empty channel list makes
_queryChannelsOnlineImplreturn early (res.channels.isEmpty && paginationParams.offset == 0), before it ever reachesstate.addChannels,channelDeliveryReporter.submitForDelivery, orchatPersistenceClient?.saveChannelQueries. The current diff still completespendingQuerywithchannels = [](Line 5498), so the gap remains.This matters beyond "no unhandled error":
disconnectUser()(called bydispose()) reassignsstate = ClientState(this)synchronously but never cancels the in-flight_onConnectionStatusChanged/_queryChannelsOnlineImplcontinuation. If that stale continuation later resolves with non-empty channels, it will silently write into the newstateinstance (and call intochannelDeliveryReporter, which was alreadycancel()-ed). That's not limited todispose()— the same race exists if a caller callsdisconnectUser()and immediately logs in a new user while a recovery is still in flight, which would contradict the PR's claim thatdisconnectUser()-based reuse is unaffected.Please add the companion non-empty-channel case the prior review asked for, and verify whether
state/channelDeliveryReporterwrites from a stale recovery need to be guarded (e.g. capturing a generation token, or cancelling in-flight recovery indisconnectUser()/dispose()) rather than relying on the empty-channels shortcut to keep the test green.🤖 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 `@packages/stream_chat/test/src/client/client_test.dart` around lines 5428 - 5503, Extend the “dispose during reconnect recovery” test to complete pendingQuery with a non-empty channel response, exercising post-query writes such as state.addChannels and channelDeliveryReporter submission. Update the reconnect recovery flow around _onConnectionStatusChanged and _queryChannelsOnlineImpl to invalidate or cancel stale work during disconnectUser()/dispose(), guarding state and channelDeliveryReporter writes so a recovery from the previous client generation cannot affect newly initialized state or canceled reporters.
🤖 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.
Duplicate comments:
In `@packages/stream_chat/test/src/client/client_test.dart`:
- Around line 5428-5503: Extend the “dispose during reconnect recovery” test to
complete pendingQuery with a non-empty channel response, exercising post-query
writes such as state.addChannels and channelDeliveryReporter submission. Update
the reconnect recovery flow around _onConnectionStatusChanged and
_queryChannelsOnlineImpl to invalidate or cancel stale work during
disconnectUser()/dispose(), guarding state and channelDeliveryReporter writes so
a recovery from the previous client generation cannot affect newly initialized
state or canceled reporters.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: b93a6c83-5c99-4791-8cd1-265b5559b23d
📒 Files selected for processing (3)
packages/stream_chat/CHANGELOG.mdpackages/stream_chat/lib/src/client/client.dartpackages/stream_chat/test/src/client/client_test.dart
2 participants
Submit a pull request
Linear: FLU-658
Github Issue: #
CLA
Description of the pull request
StreamChatClient.handleEventcould throwStateError: Cannot add new events after calling closewhen the client is disposed while a reconnect recovery was still in flight.Root cause
_onConnectionStatusChangedisasync. On an offline→online transition it awaits a recovery pipeline (sync/queryChannelsOnline) and only then emits the finalconnectionRecoveredevent. The driving subscription is fire-and-forget (wsConnectionStatusStream.pairwise().listen(...)). Ifdispose()closes_eventControllerwhile that continuation is suspended on anawait, the trailinghandleEvent(connectionRecovered)adds to a closed controller — surfacing as an unhandled async error (zone /PlatformDispatcher.onError/ Crashlytics).This is pre-existing on
master; it was surfaced by the offline-reactions E2E test (user adds a reaction while offline) on #2847, where the Android emulator's slower timing loses the race deterministically.Fix
Guard
handleEventagainst a closed controller:_eventController.isClosedis terminal and set only indispose();disconnectUser()(logout→login reuse) leaves it open, so live/reused clients are unaffected. Post-close()the broadcast stream is already complete, so no consumer ever received these events — the guard removes only the internal throw, not any delivery.Test instructions
A new regression test in
client_test.dart(dispose during reconnect recovery) disposes the client while a reconnect recovery is suspended on a held-openqueryChannels, then resumes it — reproducing the exact CI stack. Red without the guard, green with it.Summary by CodeRabbit
StateErrorwhen disposing the client while reconnect recovery is still in progress.