feat(llc, persistence): add offline support for reactions - #2847
feat(llc, persistence): add offline support for reactions#2847VelikovPetar wants to merge 15 commits into
Conversation
Co-Authored-By: Claude <noreply@anthropic.com>
|
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:
📝 WalkthroughWalkthroughAdds durable pending-operation storage for reaction mutations, queues retriable offline reaction sends/deletes, and replays them in order after websocket recovery. Persistence includes Drift schema, DAO, mappings, client methods, mocks, and tests. ChangesOffline reaction operations
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant StreamChatClient
participant PendingOperationReplayer
participant ChatPersistenceClient
participant StreamChatAPI
StreamChatClient->>PendingOperationReplayer: replay on connectionRecovered
PendingOperationReplayer->>ChatPersistenceClient: getPendingOperations
PendingOperationReplayer->>StreamChatAPI: replay reaction operation
StreamChatAPI-->>PendingOperationReplayer: success or terminal response
PendingOperationReplayer->>ChatPersistenceClient: deletePendingOperation
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches 💡 1🛠️ Fix failing CI checks 💡
🧪 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 |
|
# Conflicts: # packages/stream_chat/CHANGELOG.md
| // optimistic `ownReactions` for messages that previously had none. | ||
| state?.replaceMessage(message); | ||
| } | ||
| rethrow; |
There was a problem hiding this comment.
Not entirely sure if we should always rethrow here, maybe just in the non-retry-able case?
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #2847 +/- ##
==========================================
+ Coverage 73.96% 74.06% +0.09%
==========================================
Files 435 440 +5
Lines 28149 28272 +123
==========================================
+ Hits 20821 20939 +118
- Misses 7328 7333 +5 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (3)
packages/stream_chat_persistence/lib/src/dao/pending_operation_dao.dart (1)
19-23: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valueConsider bounding the queue.
getPendingOperationsloads the entire table into memory and the queue has no cap, so a long offline session (or an operation that keeps failing retriably) can grow it without bound. Alimiton the read, or a retention cap/TTL on insert, would keep replay predictable.🤖 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_persistence/lib/src/dao/pending_operation_dao.dart` around lines 19 - 23, Bound the pending-operation queue in getPendingOperations by applying an explicit maximum read limit while preserving ascending id order, using the project’s established queue-size configuration or constant if available. Ensure replay remains predictable without loading the entire pendingOperations table into memory.packages/stream_chat_persistence/lib/src/db/drift_chat_database.dart (1)
61-82: 🗄️ Data Integrity & Integration | 🔵 TrivialDestructive migration will silently drop unreplayed pending operations on future schema bumps.
onUpgradedeletes and recreates every table (including the newpending_operationstable) wheneverschemaVersionchanges. For the other tables this is a harmless cache eviction, butpending_operationsnow holds durable, not-yet-acknowledged user mutations — a future schema bump before replay would silently drop a user's queued reaction/delete with no error surfaced.Consider special-casing
pending_operationsin the migration (e.g. preserve/re-insert its rows acrossonUpgrade), or at minimum documenting this data-loss risk for maintainers bumpingschemaVersiongoing forward.🤖 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_persistence/lib/src/db/drift_chat_database.dart` around lines 61 - 82, Update the MigrationStrategy.onUpgrade flow so schema bumps do not silently delete durable rows from the pending_operations table: preserve and restore its pending operations while recreating the other tables, or otherwise explicitly retain that table and migrate it safely. Keep the existing destructive cache migration for non-durable tables, and ensure queued mutations remain available for replay after upgrades.packages/stream_chat/test/src/client/pending_operation_replayer_test.dart (1)
38-62: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDuplicated reaction pending-operation test scaffolding across two files.
addOpandstubSendReactionOkare copy-pasted verbatim between the two suites; the shared root cause is a missing common test helper for building/stubbing reaction pending operations.
packages/stream_chat/test/src/client/pending_operation_replayer_test.dart#L38-L62: keep as the canonical definition (or move) and have the other site import it.packages/stream_chat/test/src/client/client_test.dart#L5470-L5489: replace this local copy with an import of the shared helper instead of redefiningaddOp/stubSendReactionOk.🤖 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/pending_operation_replayer_test.dart` around lines 38 - 62, The reaction pending-operation helpers addOp and stubSendReactionOk are duplicated across the test suites. Keep or extract the canonical shared definitions from packages/stream_chat/test/src/client/pending_operation_replayer_test.dart:38-62, then update packages/stream_chat/test/src/client/client_test.dart:5470-5489 to import and reuse them, removing its local copies.
🤖 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/lib/src/client/channel.dart`:
- Around line 1685-1695: Update _enqueuePendingOperation to return whether
insertPendingOperation succeeded instead of swallowing the outcome. At both call
sites, including sendReaction, only retain the optimistic state when enqueue
succeeds; when it fails, fall back to state?.replaceMessage(message) so the
local reaction is reconciled.
In `@packages/stream_chat/lib/src/client/pending_operation_replayer.dart`:
- Around line 89-110: Validate operation.targetMessageId before returning either
deferred replay closure in _replayCallFor, and treat a null value as malformed
so the existing caller drops it through its malformed-payload handling. Replace
the deferred non-null assertions used by sendReaction and deleteReaction with
the validated value while preserving the existing replay behavior for valid
operations.
In `@packages/stream_chat/lib/src/db/chat_persistence_client.dart`:
- Around line 359-367: Update the doc comments for insertPendingOperation,
getPendingOperations, and deletePendingOperation to explicitly warn that the
default no-op implementations silently discard pending operations and can
prevent optimistic reaction replay; instruct custom ChatPersistenceClient
implementations to override all three methods together when persistence is
enabled.
---
Nitpick comments:
In `@packages/stream_chat_persistence/lib/src/dao/pending_operation_dao.dart`:
- Around line 19-23: Bound the pending-operation queue in getPendingOperations
by applying an explicit maximum read limit while preserving ascending id order,
using the project’s established queue-size configuration or constant if
available. Ensure replay remains predictable without loading the entire
pendingOperations table into memory.
In `@packages/stream_chat_persistence/lib/src/db/drift_chat_database.dart`:
- Around line 61-82: Update the MigrationStrategy.onUpgrade flow so schema bumps
do not silently delete durable rows from the pending_operations table: preserve
and restore its pending operations while recreating the other tables, or
otherwise explicitly retain that table and migrate it safely. Keep the existing
destructive cache migration for non-durable tables, and ensure queued mutations
remain available for replay after upgrades.
In `@packages/stream_chat/test/src/client/pending_operation_replayer_test.dart`:
- Around line 38-62: The reaction pending-operation helpers addOp and
stubSendReactionOk are duplicated across the test suites. Keep or extract the
canonical shared definitions from
packages/stream_chat/test/src/client/pending_operation_replayer_test.dart:38-62,
then update packages/stream_chat/test/src/client/client_test.dart:5470-5489 to
import and reuse them, removing its local copies.
🪄 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: 18d78a56-10bc-41d9-8129-154bb6dd524d
📒 Files selected for processing (25)
packages/stream_chat/CHANGELOG.mdpackages/stream_chat/lib/src/client/channel.dartpackages/stream_chat/lib/src/client/client.dartpackages/stream_chat/lib/src/client/pending_operation_replayer.dartpackages/stream_chat/lib/src/client/reaction_pending_operation.dartpackages/stream_chat/lib/src/core/models/pending_operation.dartpackages/stream_chat/lib/src/db/chat_persistence_client.dartpackages/stream_chat/lib/stream_chat.dartpackages/stream_chat/test/src/client/channel_test.dartpackages/stream_chat/test/src/client/client_test.dartpackages/stream_chat/test/src/client/pending_operation_replayer_test.dartpackages/stream_chat/test/src/core/models/pending_operation_test.dartpackages/stream_chat/test/src/mocks.dartpackages/stream_chat_persistence/CHANGELOG.mdpackages/stream_chat_persistence/lib/src/dao/dao.dartpackages/stream_chat_persistence/lib/src/dao/pending_operation_dao.dartpackages/stream_chat_persistence/lib/src/dao/pending_operation_dao.g.dartpackages/stream_chat_persistence/lib/src/db/drift_chat_database.dartpackages/stream_chat_persistence/lib/src/db/drift_chat_database.g.dartpackages/stream_chat_persistence/lib/src/entity/entity.dartpackages/stream_chat_persistence/lib/src/entity/pending_operations.dartpackages/stream_chat_persistence/lib/src/mapper/mapper.dartpackages/stream_chat_persistence/lib/src/mapper/pending_operation_mapper.dartpackages/stream_chat_persistence/lib/src/stream_chat_persistence_client.dartpackages/stream_chat_persistence/test/src/dao/pending_operation_dao_test.dart
- Roll back the optimistic reaction when enqueueing the pending operation fails, so the local state can no longer diverge from the server. - Validate targetMessageId eagerly in the replayer so a malformed operation is dropped instead of being retried forever. - Document the default no-op pending-operation methods on ChatPersistenceClient and group them at the bottom of the public surface. - Add persistence-client delegation tests for the pending-operation methods. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
|
Thanks for the nitpicks — decisions on each: Bound the queue ( Destructive migration drops pending ops ( Duplicated test helpers: leaving the small duplication. The two suites test different layers (replayer semantics vs. client wiring) and the helpers are tiny. |
…fline_support' into feature/FLU-506_add_reactions_offline_support
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 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_persistence/lib/src/db/drift_chat_database.g.dart`:
- Around line 4275-4574: The generated Drift artifact was edited directly
instead of being regenerated. Update the source Drift table and database
definitions corresponding to PendingOperations, then run the project’s Drift
code-generation command to recreate drift_chat_database.g.dart and keep the
table, PendingOperationEntity, and PendingOperationsCompanion definitions
synchronized.
- Around line 4315-4358: Regenerate the Drift-generated table code with Drift
2.33.0 so PendingOperationsTable.validateIntegrity includes payload metadata
validation and the required-column missing check for payload. Ensure the
generated handling covers both provided values and inserts that omit the
required converted payload column.
🪄 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: 624dca50-9bf9-44ce-84bb-7c2169f8ad9a
📒 Files selected for processing (2)
packages/stream_chat_persistence/lib/src/dao/pending_operation_dao.g.dartpackages/stream_chat_persistence/lib/src/db/drift_chat_database.g.dart
🚧 Files skipped from review as they are similar to previous changes (1)
- packages/stream_chat_persistence/lib/src/dao/pending_operation_dao.g.dart
# Conflicts: # packages/stream_chat/test/src/client/client_test.dart
Introduce an always-on in-memory pending-operation queue so a reaction that fails on a transient/offline error is replayed on reconnect for every client, not only those with offline persistence. This gives same-session resilience to network jank and short outages. When persistence is enabled the queue is additionally mirrored to the database, so operations survive process death and are re-hydrated into memory on the next connect. - Expand `PendingOperationReplayer` into `PendingOperationsManager` (`@internal`): it owns the in-memory queue and its lifecycle — `enqueue` (write-through to persistence), `hydrate` (DB -> memory at connect), `clear` (on disconnect), and `replay` (single-attempt, per-op, FIFO, before `/sync` and channel re-query). - `Channel.sendReaction` / `deleteReaction` now enqueue on any retriable error and keep the optimistic change, rolling back only on terminal (server-rejected) errors — regardless of whether persistence is on. - Clear the in-memory queue on `disconnectUser` so a user's queued operations can never replay under the next connected user. - `ChatPersistenceClient.insertPendingOperation` now returns the stored row id, used to correlate memory and DB entries. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
# Conflicts: # packages/stream_chat/CHANGELOG.md # packages/stream_chat_persistence/CHANGELOG.md
renefloor
left a comment
renefloor
left a comment
There was a problem hiding this comment.
Review: offline reactions (2 independent passes + cross-examination)
Reviewed this twice with independent lenses — runtime correctness/concurrency and data-model/API/tests — then had both passes cross-examine each other's findings against the code. What follows is the converged set. dart analyze --fatal-infos is clean on both packages and all touched tests pass (604 in stream_chat, 75 in stream_chat_persistence), so everything below is logic or design, not lint.
The design is sound and several non-obvious things are right: replay correctly bypasses the optimistic path by calling _client.sendReaction rather than Channel.sendReaction (no double update, no recursive re-enqueue); the persisted mirror is genuinely user-scoped (db_$userId.sqlite); the stored reaction payload is byte-for-byte what MessageApi.sendReaction puts on the wire; flush() clears the new table via allTables; and the schema bump plus drop-and-recreate migration is correct. The layered memory/persistence split is the right shape.
Inline comments cover one blocker and six majors, plus a type-safety suggestion for PendingOperation. Minor findings are omitted deliberately.
Blocker
clear()does not cancel an in-flightreplay(), so queued operations keep firing afterdisconnectUser()— and if a reconnect has landed, they are sent under the next user's credentials, with a follow-ondeletePendingOperationagainst that user's database. ~3-line fix.
Major
clear()resets_memorySeq, so session ids are reused and a stale replay silently deletes a new operation. Nothing is logged.- The "replay before re-query" invariant only holds when replay succeeds. On a retriable failure the reaction is kept locally and then wiped by
sync()/queryChannelsOnline(), becauseMessage.updateWithtakesreactionGroupswholesale from the server. The reaction visibly disappears on reconnect and reappears later — worse than the pre-PR rollback. - The replay loop does not fail fast on connection-level errors, and it
awaits on the critical path of recovery: N queued operations × Dio timeout blockssync()and channel recovery. Non-StreamChatNetworkErrorfailures also pin an operation forever. - Replay has a single trigger (
connectionRecovered); a trigger dropped under_isReplayingis never rescheduled, andconnectWebSocket: falsenever replays at all. PendingOperationshas nocreatedAtand no attempt/terminal marker — no TTL, no cap, no way to give up, no cleanup on message/channel deletion.onUpgradealready drops and recreates every table in this PR, so adding the columns is free now and costs another cache-wiping bump later.- The enqueue wiring is untested: deleting the
enqueuecalls fromchannel.dartfails zero tests.
Questions I could not settle from the code
- Is
disconnectUserduring an in-flight reconnect a supported flow? It determines how urgent the blocker is — though since the failure mode is a cross-user write, I would not ship it either way. - Is
ChatPersistenceClientintended to beimplements-able? Adding the three new members is a source break forimplementsusers (concrete bodies only rescueextends), and CLAUDE.md names this class explicitly. Non-blocking, but it wants a🔄 Changedchangelog line — or abasemodifier behind a deprecation ifimplementswas never intended. - Is re-replaying an operation the server already terminally rejected acceptable after a failed DB delete? A terminal marker column answers this and the TTL gap at once.
One changelog note worth adding regardless: the schema bump means this release wipes every user's local cache on upgrade. Pre-existing mechanism, but this PR is what triggers it.
| try { | ||
| // Copy so removals during replay don't mutate the list being iterated. | ||
| final operations = List.of(_operations); | ||
| for (final operation in operations) { |
There was a problem hiding this comment.
Blocker — clear() does not cancel an in-flight replay, so operations can be sent under the next user's credentials.
replay() iterates this snapshot, awaiting one network call per operation. clear() (L90-93) only empties _operations — it sets no cancellation flag, the loop never re-checks membership, and _isReplaying self-clears in the finally, so nothing stops the iteration either.
Sequence (verified against this branch with a throwaway test):
- User A has
[m1, m2]queued; connection recovers,replay()starts,sendReaction(m1)is in flight on a slow network. disconnectUser()runs →pendingOperationsManager.clear()(client.dart:2561). Queue is empty.m1completes. The loop continues over its stale copy and calls_client.sendReaction(m2, ...).
verify(sendReaction('m2', ...)).called(1) passes after clear(). If connectUser(B) has landed by step 3, _client holds B's token, so A's reaction is posted as B.
Second-order effect on the same path: _remove(id) then calls deletePendingOperation(id) against B's database (L99-101 early-returns only on !persistenceEnabled || id < 0). Ids are per-DB autoincrement from 1, so A's id: 1 plausibly deletes B's row 1.
The disconnectUser regression test in client_test.dart only covers replay() invoked after clear(), not clear() during a replay — so this is not caught.
Suggested fix — an epoch bumped by clear() and checked per iteration and before _remove:
int _generation = 0;
void clear() {
_operations.clear();
_generation++;
}
// inside replay():
final generation = _generation;
for (final operation in operations) {
if (generation != _generation) return; // cleared mid-replay
...
}Checking _operations.any((it) => it.id == operation.id) before each call would also work, but an epoch is cheaper and additionally protects the _remove call.
| void clear() { | ||
| _operations.clear(); | ||
| _memorySeq = 0; | ||
| } |
There was a problem hiding this comment.
Major — resetting _memorySeq makes session ids collide, and a stale replay then silently deletes a new operation.
_nextMemoryId() returns --_memorySeq, so after clear() the next memory-only operation gets -1 again. _remove(id) matches purely on it.id == id (L98). Combined with the in-flight-replay issue above:
- Persistence off.
m1(id-1) andm2(id-2) queued.replay()starts,m1in flight. clear()(disconnect). New session enqueuesm99→ id-1.m1succeeds → the stale loop calls_remove(-1)→removeWheredeletesm99.
The user's reaction is never sent, local optimistic state keeps showing it, and nothing is logged. Ids need to stay unique for the process lifetime:
| void clear() { | |
| _operations.clear(); | |
| _memorySeq = 0; | |
| } | |
| void clear() { | |
| _operations.clear(); | |
| // Do NOT reset `_memorySeq`: session ids must stay unique for the whole | |
| // process lifetime, otherwise a replay that is still draining its snapshot | |
| // can `_remove` an id that now belongs to a new session's operation. | |
| } |
This is the demonstrated instance of a broader gap: nothing owns the id value domain. PendingOperation.id is documented as "the database autoincrement id" while this class also mints negative session ids; insertPendingOperation's contract never states that ids must be positive and unique; _remove overloads the sign as a storage-tier discriminator (L99); and hydrate() trusts id! unvalidated, so a custom ChatPersistenceClient returning a null id throws into the blanket catch and pins that operation in memory for the rest of the process. See the type-safety suggestion on pending_operation.dart for a fix that removes the whole class of problem.
| try { | ||
| await call(); | ||
| } on StreamChatNetworkError catch (error) { | ||
| // Keep transient failures for the next recovery. | ||
| if (error.isRetriable) continue; | ||
| } | ||
|
|
||
| // Accepted or terminally rejected by the server — drop it. | ||
| await _remove(operation.id!); |
There was a problem hiding this comment.
Major — no fail-fast on connection-level failures, and non-network errors pin an operation forever.
Two separate problems in this block:
1. The loop continues past connection errors instead of breaking. A replay that is doomed after operation #1 (still offline, DNS failure, captive portal) still attempts every remaining operation, each burning a full Dio timeout. With N queued operations that is N × timeout seconds — and because client.dart:621 awaits this whole thing, it blocks sync() and channel recovery for that entire time. Recommend breaking out on the first StreamChatNetworkErrorType.connectionError / *Timeout, since those say nothing about the next operation but everything about the connection.
2. Only StreamChatNetworkError is caught here. Anything else — a TypeError from an unexpected response shape, a raw DioException, a StateError — escapes to the outer catch at L154, gets logged, and the operation is neither removed nor marked. With no attempt counter on the row (see the PendingOperations table comment), such a poison operation is retried on every reconnect for the lifetime of the install. The observable symptom is "silently stuck", not a retry storm, because nothing re-triggers replay in between.
Worth noting the related isRetriable sharp edge, since this line is where it decides whether local state is kept: isRetriable => data == null classifies 429 and any enveloped 5xx as terminal, so those get rolled back rather than queued. True offline (connectionError, no body) does queue correctly, so the headline path works — but the canonical "retry later" responses currently don't.
| /// Replays each queued operation against the server in insertion order. | ||
| Future<void> replay() async { | ||
| if (_isReplaying) return; | ||
| _isReplaying = true; |
There was a problem hiding this comment.
Major — replay has exactly one trigger, a dropped trigger is never rescheduled, and there is one supported mode where it never fires at all.
git grep pendingOperationsManager on this branch returns a single replay() call site: the connectionRecovered branch at client.dart:621. Three consequences:
- Retriable failure while the WebSocket stays up strands the queue. A receive timeout leaves operations queued, but there is no timer, no backoff and no
RetryQueue/RetryPolicyhook — they wait for the next full disconnect → connect cycle. if (_isReplaying) return;drops the trigger silently. There is no "replay again when the current pass finishes" flag, so a recovery that lands while a replay is draining is simply lost. Combined with (1), operations can sit for the rest of the session while local state stays diverged — and, per theclient.dartcomment, visibly wrong after the re-query.connectUser(..., connectWebSocket: false)never produces a status transition, so queued operations never replay in connection-less mode. That is a documented public flag, and the feature is inert there.
(3) is the sharpest of the three — worth either handling explicitly or documenting as unsupported. The guard itself is also completely uncovered by tests; a test that calls replay() twice against an unresolved future would pin the intended behaviour.
| Future<void> Function()? _replayCallFor(PendingOperation operation) { | ||
| switch (operation.type) { | ||
| case ReactionPendingOperation.addType: | ||
| final targetMessageId = operation.targetMessageId; | ||
| if (targetMessageId == null) { | ||
| throw StateError('Missing targetMessageId for ${operation.type}'); | ||
| } | ||
| final reaction = Reaction.fromJson( | ||
| operation.payload[ReactionPendingOperation.reactionKey] as Map<String, dynamic>, | ||
| ); | ||
| final skipPush = operation.payload[ReactionPendingOperation.skipPushKey] as bool? ?? false; | ||
| final enforceUnique = operation.payload[ReactionPendingOperation.enforceUniqueKey] as bool? ?? false; | ||
| return () => _client.sendReaction( | ||
| targetMessageId, | ||
| reaction, | ||
| skipPush: skipPush, | ||
| enforceUnique: enforceUnique, | ||
| ); | ||
| case ReactionPendingOperation.deleteType: | ||
| final targetMessageId = operation.targetMessageId; | ||
| if (targetMessageId == null) { | ||
| throw StateError('Missing targetMessageId for ${operation.type}'); | ||
| } | ||
| final reactionType = operation.payload[ReactionPendingOperation.reactionTypeKey] as String; | ||
| return () => _client.deleteReaction(targetMessageId, reactionType); | ||
| default: | ||
| // Unknown operation type — cannot be replayed by this version. | ||
| return null; | ||
| } | ||
| } |
There was a problem hiding this comment.
Major — inconsistent strictness at the parse boundary; the silent ?? false is a real divergence bug, not just a style issue.
Within one function there are three different policies for missing data:
targetMessageIdmissing →throw StateError(L179-181, L195-197)reaction→ uncheckedas Map<String, dynamic>(L183)skip_push/enforce_unique→ silentas bool? ?? false(L185-186)
The third one is the problem. enforceUnique is the flag the optimistic update already applied (channel.dart:1613 calls addMyReaction(..., enforceUnique: enforceUnique)), so if the key is ever missing or renamed, replay sends enforceUnique: false: local state shows one reaction replacing the previous one, while the server keeps both. Then the post-recovery re-query overwrites reactionGroups from server truth (see the client.dart thread) and the user ends up with two chips they never asked for. Silent, permanent, user-visible.
Also note the StateErrors exist only because targetMessageId is nullable on the model and in the DB column while being required by every operation type that exists — two of the eight manager tests exist purely to prove an impossible state is detected.
Minimum fix here is consistency — treat a missing enforce_unique/skip_push exactly like a missing targetMessageId, so it lands in the "unreplayable, drop it" path rather than silently changing semantics:
final skipPush = operation.payload[ReactionPendingOperation.skipPushKey] as bool;
final enforceUnique = operation.payload[ReactionPendingOperation.enforceUniqueKey] as bool;The better fix is to stop parsing at replay time altogether — see the suggestion on pending_operation.dart, which makes this whole function exhaustive and cast-free.
| // Replay pending offline operations (e.g. reactions) BEFORE any | ||
| // server-state refresh, so the server has each mutation before a re-query | ||
| // returns state that would otherwise clobber the optimistic change. | ||
| await pendingOperationsManager.replay(); |
There was a problem hiding this comment.
Major — the stated invariant only holds when replay succeeds; when it fails retriably the re-query wipes exactly the state this feature is meant to protect.
The comment is right about ordering, but it is load-bearing in a way the code doesn't guarantee. If replay() fails retriably, the operation stays queued and the optimistic reaction stays in local state — and then sync() (L627) and queryChannelsOnline() (L632) run immediately and overwrite the message from server truth.
Concretely, Message.updateWith returns other.copyWith(...) where other is the incoming server payload, and its preserve list is localCreatedAt/localUpdatedAt/localDeletedAt, deletedForMe, poll, sharedLocation, ownReactions, quotedMessage. reactionGroups and latestReactions are not in it, so both come wholesale from the server. Reaction chips render from reactionGroups (stream_message_reactions.dart:83; ownReactions only styles a chip that reactionGroups already says exists).
Net user-visible behaviour: the reaction added offline disappears on reconnect and reappears at some later recovery when replay finally lands. That is worse than the pre-PR rollback, which was at least immediate and self-consistent.
Two options: skip the re-query for channels that still have queued operations, or re-apply the optimistic state after reconciliation. Either way the invariant should be stated as "holds only if replay succeeded", because right now the failure path is the interesting one.
Separately, this await puts an unbounded queue on the critical path of connection recovery — it runs ahead of sync() and channel re-query, so N queued operations × per-request latency directly delays offline-event sync. Combined with the missing fail-fast in the replay loop, a flaky reconnect can block recovery for N × Dio timeout. Worth either bounding/coalescing the queue (there is no dedup of (messageId, type), so an offline user tapping the same reaction N times produces N operations and N round-trips) or moving replay off the blocking path.
| class PendingOperations extends Table { | ||
| /// Autoincrement id. | ||
| IntColumn get id => integer().autoIncrement()(); | ||
|
|
||
| /// The operation-type discriminator (e.g. `reaction.add`). | ||
| TextColumn get type => text()(); | ||
|
|
||
| /// The id of the message the operation targets, if any. | ||
| TextColumn get targetMessageId => text().nullable()(); | ||
|
|
||
| /// The operation-specific value fields, stored as JSON. | ||
| TextColumn get payload => text().map(MapConverter())(); |
There was a problem hiding this comment.
Major — no createdAt and no attempt/terminal marker, and this is the one moment when adding them is free.
schemaVersion already moves 1000 + 35 → 1000 + 36 in this PR, and MigrationStrategy.onUpgrade (drift_chat_database.dart:74-81) drops and recreates every table on any version change. So these columns cost nothing right now, and cost another cache-wiping schema bump later.
What their absence causes today:
- No TTL and no cap. A user offline for a week accumulates unbounded rows that all replay serially on reconnect — and there is no data with which to implement expiry later.
- No attempt/failure count. A permanently-failing operation can never be given up on. This is what makes the non-
StreamChatNetworkErrorcase inPendingOperationsManager.replay()unrecoverable, and what makes a faileddeletePendingOperationunbounded:_removedrops the operation from memory first, so if the DB delete fails the row survives,hydrate()reloads it on the nextconnectUser, and it is replayed again — including operations the server already terminally rejected. - No
channel_cidand no FK toMessages. Contrastreactions.dart:11, which does.references(Messages, #id, onDelete: KeyAction.cascade). Nothing indeleteChannels/deleteMessageByIds/truncatetouches this table, so queued reactions for locally-deleted messages survive and replay.
Every other entity in this package (locations, polls, channels) carries timestamps; this one carries none.
| class PendingOperations extends Table { | |
| /// Autoincrement id. | |
| IntColumn get id => integer().autoIncrement()(); | |
| /// The operation-type discriminator (e.g. `reaction.add`). | |
| TextColumn get type => text()(); | |
| /// The id of the message the operation targets, if any. | |
| TextColumn get targetMessageId => text().nullable()(); | |
| /// The operation-specific value fields, stored as JSON. | |
| TextColumn get payload => text().map(MapConverter())(); | |
| class PendingOperations extends Table { | |
| /// Autoincrement id. | |
| IntColumn get id => integer().autoIncrement()(); | |
| /// The operation-type discriminator (e.g. `reaction.add`). | |
| TextColumn get type => text()(); | |
| /// The id of the message the operation targets, if any. | |
| TextColumn get targetMessageId => text().nullable()(); | |
| /// The operation-specific value fields, stored as JSON. | |
| TextColumn get payload => text().map(MapConverter())(); | |
| /// When the operation was queued, used to expire stale operations. | |
| DateTimeColumn get createdAt => dateTime()(); | |
| /// How many times replay has been attempted, used to give up on an | |
| /// operation that keeps failing rather than retrying it forever. | |
| IntColumn get attempts => integer().withDefault(const Constant(0))(); | |
| } |
A createdAt column also unlocks a safer policy for the forward-incompatible case in replay(): an operation whose type is unknown to this version (written by a newer app, read after a downgrade or during a staged rollout) is currently deleted, silently destroying recorded user intent. With a timestamp you can skip-and-expire instead of dropping on sight.
| test( | ||
| 'a retriable failure keeps the optimistic reaction for replay', | ||
| () async { | ||
| const type = 'like'; | ||
| final message = Message(id: 'offline-msg', state: MessageState.sent); | ||
| final reaction = Reaction( | ||
| type: type, | ||
| messageId: message.id, | ||
| user: client.state.currentUser, | ||
| ); | ||
|
|
||
| // data == null → retriable/offline error. | ||
| when( | ||
| () => client.sendReaction(message.id, reaction), | ||
| ).thenThrow(StreamChatNetworkError(ChatErrorCode.inputError)); | ||
|
|
||
| await expectLater( | ||
| channel.sendReaction(message, reaction), | ||
| throwsA(isA<StreamChatNetworkError>()), | ||
| ); | ||
|
|
||
| // The optimistic reaction is kept (queued for replay on reconnect) | ||
| final current = channel.state!.messages.firstWhere( | ||
| (m) => m.id == message.id, | ||
| ); | ||
| expect(current.ownReactions?.map((r) => r.type), contains(type)); | ||
| }, | ||
| ); |
There was a problem hiding this comment.
Major — the headline behaviour of this PR is untested: the enqueue wiring can be deleted and CI stays green.
This test (and its deleteReaction twin) asserts only that the optimistic reaction is kept. That outcome is produced entirely by the if (retriable) { ... } else { state?.replaceMessage(message) } branch — the enqueue call contributes nothing to it. And the group's setUpAll stubs when(() => client.chatPersistenceClient).thenReturn(null) (L244), so persistenceEnabled is false and the enqueue is memory-only, while PendingOperationsManager exposes no way to inspect its queue.
Net result: deleting await _client.pendingOperationsManager.enqueue(...) from channel.dart:1630 and :1670 fails zero tests. "Reactions get queued for replay" — the thing this PR is for — has no coverage at the point where it is triggered.
Related: the comment at L242-243 says "pending-operation tests install a MockPersistenceClient locally", but no test in this group does (grep MockPersistenceClient channel_test.dart only hits the unrelated "User messages deleted event" group), which also makes the tearDown re-stub at L260-261 dead code justified by tests that don't exist.
Fix: install a MockPersistenceClient in these two tests and assert the queued operation, including flag pass-through:
expect(persistence.storedPendingOperations, hasLength(1));
final op = persistence.storedPendingOperations.single;
expect(op.type, ReactionPendingOperation.addType);
expect(op.targetMessageId, message.id);
expect(op.payload[ReactionPendingOperation.enforceUniqueKey], isTrue);Alternatively expose an @visibleForTesting queue view on PendingOperationsManager so the memory-only path (the one that is always on) can be asserted directly. Either works — but without one of them the PR's central wiring is uncovered.
Other gaps worth closing while you are here: add→delete of the same reaction (the common offline toggle, and the only ordering where FIFO changes the outcome); a nested-Map disk round-trip (the DAO test only ever stores flat primitives, yet payload['reaction'] as Map<String, dynamic> is load-bearing); clear() during an in-flight replay; memory-id reuse after clear(); and the connectUser → hydrate() wiring end to end (the new client_test.dart group seeds the queue after connectUser, so hydrate-on-connect is never exercised).
| class PendingOperation extends Equatable { | ||
| /// {@macro pendingOperation} | ||
| const PendingOperation({ | ||
| required this.type, | ||
| required this.payload, | ||
| this.id, | ||
| this.targetMessageId, | ||
| }); | ||
|
|
||
| /// The database autoincrement id, assigned when the operation is stored; | ||
| /// `null` until then. | ||
| final int? id; | ||
|
|
||
| /// The discriminator persisted in the `type` column, e.g. `reaction.add`. | ||
| final String type; | ||
|
|
||
| /// The id of the message the operation targets, if any. | ||
| final String? targetMessageId; | ||
|
|
||
| /// The operation-specific value fields, stored as JSON. | ||
| final Map<String, dynamic> payload; | ||
|
|
||
| /// Returns a copy of this operation with the given fields replaced. | ||
| PendingOperation copyWith({ | ||
| int? id, | ||
| String? type, | ||
| String? targetMessageId, | ||
| Map<String, dynamic>? payload, | ||
| }) => PendingOperation( | ||
| id: id ?? this.id, | ||
| type: type ?? this.type, | ||
| targetMessageId: targetMessageId ?? this.targetMessageId, | ||
| payload: payload ?? this.payload, | ||
| ); | ||
|
|
||
| @override | ||
| List<Object?> get props => [ | ||
| type, | ||
| targetMessageId, | ||
| payload, | ||
| ]; | ||
| } |
There was a problem hiding this comment.
Suggestion — make PendingOperation type-safe with a sealed hierarchy, so illegal states stop being representable.
As written this is a stringly-typed bag: type is a String, payload is an untyped Map<String, dynamic>, and targetMessageId is nullable even though every operation type that exists requires it. All of the validation therefore happens at replay time in PendingOperationsManager._replayCallFor, which is why that function needs two StateError throws, two unchecked casts, and two silent ?? false defaults — and why two of the eight manager tests exist only to prove an impossible state is detected.
The DB row should stay generic (type, payload) — that is what makes the table reusable. The in-memory model doesn't have to be. Parse once at the persistence boundary, and replay becomes exhaustive and cast-free.
The repo is on Dart ^3.11 and already uses sealed hierarchies for exactly this shape (MessageState, MessageDeleteScope), so this is idiomatic here:
/// Identity of a queued operation: either a persisted row or session-only.
sealed class PendingOperationId extends Equatable {
const PendingOperationId();
}
/// An operation mirrored to persistence, identified by its row id.
final class PersistedOperationId extends PendingOperationId {
const PersistedOperationId(this.value) : assert(value > 0, 'row ids are positive');
final int value;
@override
List<Object?> get props => [value];
}
/// An operation that lives only in memory for this session.
final class SessionOperationId extends PendingOperationId {
const SessionOperationId(this.value);
final int value;
@override
List<Object?> get props => [value];
}
/// {@macro pendingOperation}
sealed class PendingOperation extends Equatable {
const PendingOperation({this.id});
/// Identity of this operation, `null` until it is queued.
final PendingOperationId? id;
/// The discriminator persisted in the `type` column.
String get type;
/// The message this operation targets.
String get targetMessageId;
/// The operation-specific value fields. Must be JSON-encodable.
Map<String, dynamic> toPayload();
/// Returns a copy of this operation with [id] assigned.
PendingOperation withId(PendingOperationId id);
/// Rebuilds a stored operation, or returns `null` when [type] is unknown to
/// this version. Throws [FormatException] on a malformed payload.
static PendingOperation? fromStored({
required PendingOperationId id,
required String type,
required String? targetMessageId,
required Map<String, dynamic> payload,
}) => switch (type) {
AddReactionOperation.opType => AddReactionOperation.fromPayload(id, payload),
DeleteReactionOperation.opType => DeleteReactionOperation.fromPayload(id, payload),
_ => null,
};
}
/// A reaction added while offline, awaiting replay.
final class AddReactionOperation extends PendingOperation {
const AddReactionOperation({
required this.reaction,
this.skipPush = false,
this.enforceUnique = false,
super.id,
});
/// The discriminator persisted for this operation type.
static const opType = 'reaction.add';
/// The reaction to send.
final Reaction reaction;
/// Whether to skip the push notification for the reaction.
final bool skipPush;
/// Whether the reaction replaces the user's existing one.
final bool enforceUnique;
@override
String get type => opType;
@override
String get targetMessageId => reaction.messageId!;
// ... toPayload / fromPayload / withId / props
}_replayCallFor then collapses to an exhaustive switch with no default, no null return, and no casts:
Future<void> Function() _replayCallFor(PendingOperation operation) =>
switch (operation) {
AddReactionOperation(:final targetMessageId, :final reaction, :final skipPush, :final enforceUnique) =>
() => _client.sendReaction(targetMessageId, reaction, skipPush: skipPush, enforceUnique: enforceUnique),
DeleteReactionOperation(:final targetMessageId, :final reactionType) =>
() => _client.deleteReaction(targetMessageId, reactionType),
};What this buys, concretely:
enforceUniquecan no longer silently default tofalse— the malformed-payload case becomes aFormatExceptionat the parse boundary, not a semantic change at send time.- No
StateErrorfor a missingtargetMessageId— it is non-nullable by construction. - No unchecked
as Map<String, dynamic>/as String. - The
idsign hack disappears._removebecomesswitch (id) { PersistedOperationId(:final value) => deletePendingOperation(value), SessionOperationId() => null }instead ofif (id < 0) return, which is what makes the_memorySeqcollision above possible in the first place. Theassert(value > 0)also documents and enforces theinsertPendingOperationcontract that is currently only implied. - Adding
message.sendor a channel operation becomes a compile error until handled, instead of a silentdefault: return nullthat drops the row.
Two smaller things on the current class while it is being touched: props (L46-50) excludes id, so op == op.copyWith(id: 7) — harmless today because removal matches on id, but a future List.remove(op) would delete the wrong entry. And the doc on L19-20 says "the database autoincrement id, null until then" while PendingOperationsManager also assigns negative session ids to this field, so the documented value domain and the real one disagree.
Fully understand if the sealed refactor is out of scope for this PR — in that case the minimum I would ask for is making the strictness in _replayCallFor consistent (throw on a missing enforce_unique exactly as for targetMessageId), plus documenting on insertPendingOperation that ids must be positive and unique and that payload must be JSON-encodable.
There was a problem hiding this comment.
Follow-up: React Native already ships exactly this shape, which I think settles the "is this over-engineering?" question.
RN's equivalent type is a discriminated union whose payload type is derived from the method being replayed (stream-chat/src/offline-support/types.ts):
export type PendingTask = {
channelId: string;
channelType: string;
messageId: string;
id?: number;
} & (
| { type: 'send-reaction'; payload: Parameters<Channel['sendReaction']> }
| { type: 'delete-reaction'; payload: Parameters<Channel['deleteReaction']> }
| { type: 'delete-message'; payload: Parameters<StreamChat['deleteMessage']> }
);and its executor narrows on the discriminant and spreads the payload, so there are no casts and no defaults to get wrong (offline_support_api.ts:1268-1320):
if (task.type === 'send-reaction') return await channel._sendReaction(...task.payload);
if (task.type === 'delete-reaction') return await channel._deleteReaction(...task.payload);Same idea as the sealed hierarchy above: the row stays generic (type, payload) on disk, the in-memory model is typed, and the payload can't drift from the signature it feeds. Dart's sealed classes + exhaustive switch expressions are the direct analogue of TS's discriminated union + narrowing, and this repo already uses that pattern (MessageState, MessageDeleteScope).
Two extra reasons this matters more here than it looks:
- RN's queue already carries four operation families (reactions,
send-message,delete-message/update-message, drafts). If Flutter follows,_replayCallFor'sdefault: return nullbecomes the place every future operation type can be silently forgotten — whereas an exhaustive switch makes each addition a compile error until handled. - It closes the
idhole from the other thread by construction. Asealed PendingOperationIdturns_remove'sif (id < 0) returninto a pattern match, so the_memorySeqcollision stops being expressible.
Worth noting RN also carries channelId/channelType/threadId on the task, because its executor resolves a Channel to replay against. This PR routes replay through _client.sendReaction by message id, so it genuinely doesn't need them — but that's also why nothing here can clean up queued operations per channel.
| final retriable = e is StreamChatNetworkError && e.isRetriable; | ||
| if (retriable) { |
There was a problem hiding this comment.
Cross-SDK divergence — this predicate makes Flutter roll back exactly the errors React Native queues.
Raising this as its own thread (it was an aside in my PendingOperationsManager.replay() comment) because comparing against the RN/JS implementation turns it from a theoretical sharp edge into a behavioural inconsistency between Stream SDKs.
RN classifies with an explicit terminal allowlist (stream-chat/src/offline-support/offline_support_api.ts:1154):
private shouldSkipQueueingTask = (error: AxiosError<APIErrorResponse>) =>
error?.response?.data?.code === 4 || // bad request data
error?.response?.data?.code === 17; // missing own_capabilitiesEverything not on that list is queued and replayed. Flutter's isRetriable => data == null is the inverse: any response carrying a body is terminal. So for the same user action:
| Failure | RN | This PR |
|---|---|---|
| Offline / connection error | queued | queued ✅ |
| 429 rate limited | queued | rolled back |
| 5xx with the Stream error envelope | queued | rolled back |
| Bad request data (code 4) | dropped | dropped ✅ |
| Missing capability (code 17) | dropped | rolled back ✅ |
429 and transient 5xx are the canonical "retry this later" responses, and they're the ones a real device hits under load — so the headline feature currently doesn't engage in the failure modes most likely to occur after plain offline. Worth deciding deliberately rather than inheriting it from data == null.
Two things RN does here that are worth borrowing:
- Enumerate the terminal cases instead of inferring them from body presence — an allowlist of codes/statuses fails safe (unknown error ⇒ keep the operation) where
data == nullfails destructive (unknown error with a body ⇒ discard the user's reaction). - Pre-check the connection before spending a request. RN's
queueTask(offline_support_api.ts:1127-1136) checkswsConnection?.isHealthyand throws anOfflineErrorwithout a network round-trip, so the common offline case never waits on a Dio timeout. Here every offline reaction pays a full timeout before being queued.
I'd leave isRetriable itself alone — it's pre-existing and shared with RetryQueue. The narrow fix is to not reuse it for this decision, since this is the first place it determines whether local state is kept or reverted rather than merely whether to retry a send.
Cross-check against the React Native SDKSince RN has shipped offline queuing for reactions (plus messages and drafts) for a while, I compared this PR against it — How RN is structured, for context: Where RN corroborates the findings above
Where RN is no better — these are shared gaps, not regressions in this PR
Where this PR is better than RN
NetNothing in the RN comparison changes the blocker or the six majors. It does raise my confidence on the schema columns and the error classification, supplies a concrete precedent for the type-safety suggestion, and shows that three of the weaknesses are shared across SDKs and probably worth an upstream issue rather than just a fix here. |
3 participants
Submit a pull request
Linear: FLU-506
Github Issue: #
CLA
Description of the pull request
Adds offline support for reactions with a layered replay model. When a reaction add/remove fails with a transient/offline error, the optimistic change is kept and the operation is queued for replay when the connection recovers, instead of being reverted. Terminal (server-rejected) failures still roll back; only transient/offline errors are queued.
Low-level client (
stream_chat):PendingOperationmodel and internalReactionPendingOperation(add/delete) describing a queued reaction op.PendingOperationsManager(@internal) owns the in-memory queue and its lifecycle:enqueue(write-through to persistence when enabled),hydrate(DB → memory at connect),clear(on disconnect — prevents a user's ops leaking into the next session), andreplay(single-attempt, per-op, FIFO, on connection recovery before/syncand channel re-query so the server sees mutations first).Channel.sendReaction/Channel.deleteReactionenqueue on any retriable error and keep the optimistic change, rolling back only on terminal errors.ChatPersistenceClient.insertPendingOperationreturns the stored row id (used to correlate memory and DB entries).Persistence (
stream_chat_persistence):PendingOperationsDrift table,PendingOperationDao, entity, and mapper, wired intoDriftChatDatabaseandStreamChatPersistenceClient;insertPendingOperationreturns the autoincrement id.Tests:
PendingOperationsManager(persistence on/off, hydrate, clear, insert-failure fallback, and FIFO / transient-keep / terminal-drop replay).disconnectUser.sample_appE2E reactions integration test.Screenshots / Videos
No UI changes.