Track last ids for DataStorm filter subscriptions#5604
Merged
Conversation
getLastIds returned an empty dict for filter (any-key/filtered) subscriptions because only key subscriptions are recorded in subscriber.keys. As a result, after a session reconnect the peer writer was asked for samples from lastId 0 and re-sent its entire retained queue, delivering duplicate samples to a filtered/any-key reader (and tripping a debug assert). Report the lastId for each filter subscription directly from the element subscribers, keyed by the remote writer's element id. Iterating all filter subscriptions (rather than a single id) handles multiple any-key writers, which share a filter id but have distinct element ids. Adds an any-key reader reconnect test to the DataStorm/reliability test. Fixes #5471
pepone
force-pushed
the
fix-datastorm-filter-lastid
branch
2 times, most recently
from
7f978ea to
b6c163f
Compare
Contributor
There was a problem hiding this comment.
Pull request overview
Note
Copilot was unable to run its full agentic suite in this review.
Fixes incorrect last-id reporting for DataStorm filter (any-key/filtered) subscriptions to prevent duplicate sample delivery after reconnects.
Changes:
- Report lastIds for filter subscriptions by iterating element subscribers and keying results by the remote writer element id.
- Add a reliability test that forces an any-key reader reconnect and asserts the reader continues from the correct lastId (no duplicate retained samples).
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
| cpp/src/DataStorm/SessionI.cpp | Fixes getLastIds for filter subscriptions to return correct per-writer lastId values after reconnect. |
| cpp/test/DataStorm/reliability/Writer.cpp | Adds writer-side harness for an any-key reconnect scenario with retained history. |
| cpp/test/DataStorm/reliability/Reader.cpp | Adds reader-side reconnect test asserting no duplicate/rewound samples are delivered. |
…ter is the trigger
An any-key writer marshals its samples with keyId 0 and the key value inline (marshalKey = _keys.empty()). On reconnect, these resumed samples flow through subscriberInitialized for a single-key reader, whose subscription key is non-null. The assert there required either a null key or that keyId map to the subscription key, so it rejected the inline-key case -- even though the line just below it (key ? key : decode(sample.keyValue)) and the live data path in s() both handle it correctly. Relax the assert to accept an inline-key sample, matching s(). This is a debug-only assert; release behavior was already correct, which is why the failure showed up only on debug CI under the timing that delivers resumed samples through this path.
bernardnormier
approved these changes
Jun 24, 2026
bernardnormier
left a comment
bernardnormier
left a comment
Member
There was a problem hiding this comment.
LGTM. Verified the id round-trip: attachFilter stores the subscription under -elementId (the writer's element id), and the new branch emits it back as the positive id the writer matches via data.lastIds.find(_id). The scan over negative-id entries is the right call since keyOrFilterId is the peer subscriber id, not the _elements storage key. The assert relaxation is also correct (and incidentally avoids the spurious keys[0] insertion in debug). Nice test.
pepone
added a commit
that referenced
this pull request
Jul 1, 2026
(cherry picked from commit a1f3963)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
getLastIdsreturned an empty dict for a subscription to a filter (any-key/filtered) writer, because onlysubscriptions to key writers are recorded in
subscriber.keys. As a result, after a session reconnect the writerwas asked for samples from
lastId0 and re-sent its entire retained queue, delivering duplicate samples to theconnected reader (and tripping a debug assert).
The trigger is the writer being any-key/filtered: any reader connected to such a writer is affected, regardless of
whether the reader itself is a key or filtered reader (the reader's subscription to a filter writer element is not
indexed in
subscriber.keys).The fix reports the
lastIdfor each subscription to a filter writer element directly from the elementsubscribers, keyed by the writer's element id (the id the writer matches in
DataElementI::attach). It iteratesall such subscriptions rather than looking one up because filters have no per-id index (unlike keys), and a single
filter can match multiple writer elements with distinct element ids.
Adds a reconnect regression test to the DataStorm/reliability test.
Fixes #5471