fix(streaming): reject active implicit rewinds - #10662
Merged
ReubenBond merged 1 commit intoAug 24, 2026
Merged
ReubenBond merged 1 commit into
ReubenBond merged 1 commit into
Conversation
Contributor
There was a problem hiding this comment.
Pull request overview
This PR fixes a long-standing gap in the implicit stream subscription path where rewinding/resuming from an earlier sequence token would not “wake” the persistent pulling agent until a new event was produced. It does so by retaining the active producer identity during the consumer handshake and re-notifying that producer when an observing subscription resumes, enabling immediate replay from retained cache.
Changes:
- Propagate/retain the pulling-agent
GrainIdduring the consumer handshake (viaRequestContext) so the consumer can remember which producer to refresh. - On
ResumeAsync, preserve filter data and proactively refresh the producer for already-observing subscriptions to trigger immediate replay. - Add a regression test covering “drain 5, rewind to first token, replay without new production” for implicit subscriptions.
Show a summary per file
| File | Description |
|---|---|
| test/Orleans.Streaming.Tests/StreamingTests/MemoryStreamResumeTests.cs | Adds a regression test validating immediate replay on implicit rewind without producing new events. |
| test/Grains/TestGrains/ImplicitSubscriptionCounterGrain.cs | Captures the first token and adds a rewind helper to resume from the first retained token. |
| test/Grains/TestGrainInterfaces/IImplicitSubscriptionCounterGrain.cs | Exposes the new rewind helper on the test grain interface. |
| src/Orleans.Streaming/PersistentStreams/PersistentStreamPullingAgent.cs | Wraps the sequence-token handshake call to flow producer identity via RequestContext. |
| src/Orleans.Streaming/Internal/StreamSubscriptionHandleImpl.cs | Exposes filter data and whether the handle currently has an observer attached (internal). |
| src/Orleans.Streaming/Internal/StreamConsumerExtension.cs | Remembers producer per subscription and adds a refresh hook to re-add the subscriber on resume. |
| src/Orleans.Streaming/Internal/StreamConsumer.cs | On resume, preserves filter data and refreshes the producer for already-observing subscriptions (with rollback on failure). |
| src/Orleans.Streaming/Internal/IStreamGrainExtensions.cs | Adds an internal RequestContext key for producer identity. |
Review details
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
- Files reviewed: 8/8 changed files
- Comments generated: 1
- Review effort level: Lite
This was referenced Aug 19, 2026
Contributor
There was a problem hiding this comment.
Review details
Suppressed comments (1)
Previously missed (1) — in code that hasn't changed since the last review.
test/Grains/TestGrains/ImplicitSubscriptionCounterGrain.cs:101
- RewindToFirstToken() resumes using the ResumeAsync(onNext, token) overload, which uses default no-op error/completed handlers. That means any delivery errors during the rewind/replay path will no longer increment ErrorCounter or persist state like the initial subscription path does (OnSubscribed wires an OnError handler which does that). Consider resuming with the same OnError/OnCompleted behavior for consistency.
throw new InvalidOperationException("The stream must deliver an event before it can rewind.");
}
this.streamHandle = await this.streamHandle.ResumeAsync(OnNext, this.State.FirstToken);
}
- Files reviewed: 8/8 changed files
- Comments generated: 0 new
- Review effort level: Lite
Contributor
There was a problem hiding this comment.
Review details
Suppressed comments (1)
Previously missed (1) — in code that hasn't changed since the last review.
src/Orleans.Streaming/Internal/StreamConsumerExtension.cs:224
- GetSequenceToken records the stream producer in streamProducers even when there is no local observer/handle for the subscriptionId. In a race where the observer was already removed, this can leave behind stale producer mappings for subscriptions which will never be refreshed/removed (until activation teardown). Only persist the producer mapping when an observer exists for that subscriptionId.
public Task<StreamHandshakeToken?> GetSequenceToken(GuidId subscriptionId)
{
if (RequestContext.Get(StreamRequestContextKeys.StreamProducer) is GrainId streamProducer)
{
streamProducers[subscriptionId] = streamProducer;
- Files reviewed: 9/9 changed files
- Comments generated: 0 new
- Review effort level: Lite
ReubenBond
force-pushed
the
rb-issue-901-delayed-implicit-stream-rewind
branch
from
August 21, 2026 06:06
5f892f5 to
d48bc43
Compare
ReubenBond
force-pushed
the
rb-issue-901-delayed-implicit-stream-rewind
branch
from
August 21, 2026 13:07
6b19ecf to
c031b2b
Compare
This was referenced Aug 31, 2026
This was referenced Sep 7, 2026
Open
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.
Problem
An active implicit stream subscription could request an earlier sequence token after it had already completed delivery. Implicit pub/sub is stateless, so it has no authoritative producer coordination for repositioning that live subscription. The request could appear to succeed and then wait for unrelated stream activity.
Solution
ResumeAsynctokens locally before replacing the observerResumeAsync(null)replaces an observer, including replacement from inside a delivery callbackRationale
This gives unsupported post-processing rewinds an immediate, deterministic result while preserving the stateless implicit pub/sub design. Applications which intentionally reposition an active subscription use explicit subscriptions, whose pub/sub protocol coordinates that operation.
Fixes #901
Microsoft Reviewers: Open in CodeFlow