fix(streaming): advance checkpoints past drained subscriptions - #11249

Updates persistent-stream checkpointing so drained subscriptions no longer hold back progress while preserving replay safety.

Changes:

• Tracks subscription progress, read boundaries, and replay state.
• Defers checkpoint advancement during registration and handshakes.
• Adds checkpoint and replay regression tests.

Suppressed comments (3)

src/Orleans.Streaming/PersistentStreams/PersistentStreamPullingAgent.cs:1056

• This now seeds the value passed to IQueueCache.UpdateDeliveryProgress with the latest queue-read token. The interface contract defines this parameter as the earliest last-processed token across registered subscriptions and says that null means there are no active subscriptions (src/Orleans.Streaming/QueueAdapters/IQueueCache.cs:101-110); when all consumers are drained or none exist, custom cache implementations can therefore interpret this non-null value as an active-subscription watermark. Please either update that contract and its implementations to explicitly make this a queue checkpoint boundary, or keep the read boundary separate from this API.

            earliest = _lastReadToken;

src/Orleans.Streaming/PersistentStreams/PersistentStreamPullingAgent.cs:1397

• GetBatchForConsumer deliberately returns Success with no progress when it saw a batch whose position is missing (the sawBatch path below). This branch only makes IsCaughtUp false transiently; after a later refresh returns NoData, the !HasDeliveryProgressError check can mark the subscription caught up and let shutdown skip the unknown position. Record a delivery-progress error for any non-NoData result with no progress so this subscription remains a checkpoint cap.

                            consumerData.IsCaughtUp = nextBatch.CursorResult.Kind == QueueCacheCursorMoveResultKind.NoData
                                && !consumerData.HasDeliveryProgressError
                                && consumerData.UnconfirmedDeliveryToken is null;

src/Orleans.Streaming/PersistentStreams/PersistentStreamPullingAgent.cs:360

• A reattach handshake can acknowledge a position at or beyond UnconfirmedDeliveryToken, but this path only records LastProcessedToken; it never clears the replay marker. After a successful rewind followed by reattachment, the cursor can drain while UnconfirmedDeliveryToken remains set, so IsCaughtUp stays false and this subscription continues to pin the checkpoint indefinitely. Apply the same acknowledgement/clear rule used after delivery to successful handshake responses (including the implicit-subscription token case where it represents an acknowledgement).

                    RecordDeliveryProgress(data, GetInitialDeliveryProgress(data.LastToken, data.LastProcessedToken));

• Files reviewed: 3/3 changed files
• Comments generated: 2
• Review effort level: Lite

Suppressed comments (2)

src/Orleans.Streaming/PersistentStreams/PersistentStreamPullingAgent.cs:361

• A reattachment handshake can acknowledge a position at or beyond an in-flight delivery, but this path only updates LastProcessedToken; it never reconciles UnconfirmedDeliveryToken. If the old callback is then rejected by the cursor-version check and the replacement cursor is already drained, RunConsumerCursor leaves IsCaughtUp false solely because that stale unconfirmed token remains, so the checkpoint stays pinned. Clear the unconfirmed token when the handshake's acknowledged progress reaches it.

                    RecordDeliveryProgress(data, GetInitialDeliveryProgress(data.LastToken, data.LastProcessedToken));

src/Orleans.Streaming/PersistentStreams/PersistentStreamPullingAgent.cs:162

• Initialize resets _lastReadToken but leaves _hasUnknownReplayPosition latched. After a previous read with a missing/incompatible token or failed cache admission, that flag suppresses every later NotifyDeliveryProgress, so a receiver reinitialized on the same agent can never publish a checkpoint even after successful reads. Reset this per-receiver state here as well.

            _lastReadToken = null;
            pendingConsumerRecoveries.Clear();

• Files reviewed: 3/4 changed files
• Comments generated: 1
• Review effort level: Lite

• Files reviewed: 3/4 changed files
• Comments generated: 2
• Review effort level: Lite

• Files reviewed: 3/4 changed files
• Comments generated: 2
• Review effort level: Lite

• Files reviewed: 4/5 changed files
• Comments generated: 1
• Review effort level: Lite

Suppressed comments (2)

src/Orleans.Streaming/PersistentStreams/PersistentStreamPullingAgent.cs:1596

• When an attachment fails after producer registration succeeds, this path leaves the task completed but retries the whole RegisterAsStreamProducer call. PubSubRendezvousGrain.RegisterProducer writes state, increments producer metrics, and emits ProducerRegistered on every call even though the HashSet deduplicates the producer, so each bounded attachment retry causes duplicate persistence work, metrics, and registration events. Avoid re-registering an already registered producer on attachment-only retries (for example, retain the subscriber set or add a read/reattach operation) while still retrying the failed attachments.

                    var subscribers = await RegisterAsStreamProducer(streamId, cancellationToken);

                    if (IsShutdown || cancellationToken.IsCancellationRequested)

src/Orleans.Streaming/PersistentStreams/PersistentStreamPullingAgent.cs:931

• This guard pauses ReadFromQueue for the entire queue whenever any stream has a non-null RegistrationTask. Registration and handshake work can retry for a long time (producer registration is configured for infinite retries), so one stalled or unavailable subscriber blocks unrelated streams from being read and prevents their checkpoints from advancing; this also contradicts the per-stream background-registration rationale at the call site below. Keep the pending stream pinned and excluded from checkpoint calculation while allowing unrelated stream groups to continue, or otherwise isolate registration stalls.

            if (pubSubCache.Values.Any(static stream => stream.RegistrationTask is not null))
            {

• Files reviewed: 4/5 changed files
• Comments generated: 1
• Review effort level: Lite

• Files reviewed: 5/5 changed files
• Comments generated: 1
• Review effort level: Lite

• Files reviewed: 5/5 changed files
• Comments generated: 1
• Review effort level: Lite

Suppressed comments (2)

src/Orleans.Streaming/PersistentStreams/PersistentStreamPullingAgent.cs:785

• consumer.LastProcessedToken can legitimately be null after an initial handshake which returns a StartToken, before the first batch establishes delivery progress. If the subscription is removed while the queue boundary is known, this condition reaches TryCompareQueueProgress despite the null-forgiving operator, and Compare dereferences the null token instead of removing the subscription. Guard the token before comparing and fall back to legacy progress.

                    && (!HasCaughtUpDeliveryProgress(consumer, streamData.LastReadToken)
                        || _lastReadToken is null
                        || !TryCompareQueueProgress(consumer.LastProcessedToken!, _lastReadToken, out _))))

src/Orleans.Streaming/PersistentStreams/PersistentStreamPullingAgent.cs:1057

• Cleanup can encounter a subscription whose initial StartToken handshake left LastProcessedToken null because no batch has established progress yet. With a known queue read boundary, this new check then calls TryCompareQueueProgress with null and throws from the queue-read cleanup path, preventing subsequent reads; add the same null guard before comparing.

                            !HasCaughtUpDeliveryProgress(consumer, streamData.LastReadToken)
                            || _lastReadToken is null
                            || !TryCompareQueueProgress(consumer.LastProcessedToken!, _lastReadToken, out _))))

• Files reviewed: 5/5 changed files
• Comments generated: 0 new
• Review effort level: Lite

• Files reviewed: 5/5 changed files
• Comments generated: 0 new
• Review effort level: Lite