feat(streaming)!: unify stream partition recovery

[ReubenBond]

Problem

Persistent stream recovery and checkpoint behavior is provider-specific, so delivery progress, cache replay, and durable resume semantics can diverge. Kinesis offsets use arbitrary precision beyond the numeric range represented by the existing cache path. The alpha ADO.NET provider acknowledges delivery by destructively removing queue records, coupling storage lifetime to one delivery attempt instead of preserving shared partition history for multicast replay and checkpoint-safe retention.

Solution

• Extract a backend-neutral, optimistic-concurrency checkpoint state machine with throttling, flush, cancellation, and non-regression guarantees; keep grain, Azure Table, and DynamoDB compatibility facades.
• Report the earliest fully delivered subscription position periodically using provider token ordering.
• Add a reusable pooled recoverable stream partition pipeline with lazy decoding, provider-defined cached-offset comparison, pressure control, delivery-watermark reconciliation, and per-queue receiver/cache identity.
• Migrate Kinesis to arbitrary-precision shard tokens, strict-after checkpoint resume, raw pooled records, and rewindable behavior while preserving its wire format and public configuration.
• Replace the alpha ADO.NET visibility/dead-letter queue with versioned immutable stream partitions, commit-ordered stream record appends, owner-epoch fencing, rewindable pooled delivery, checkpoint-aware retention, hard retention diagnostics, former queue schema guidance, and stream partition benchmarks for SQL Server, PostgreSQL, and MySQL.

Rationale

A single recovery model makes at-least-once behavior explicit: crashes can duplicate delivery, while durable checkpoints advance only through fully delivered records. Partition history remains immutable for multicast subscriptions, default retention stops at the checkpoint, and future partitioned stream providers can supply partition discovery, a source/data adapter, and checkpoint storage while reusing the stream partition pipeline.

This prerelease change requires coordinated drain and ADO.NET streaming schema replacement.

Microsoft Reviewers: Open in CodeFlow

[Copilot]

Pull request overview

Introduces a provider-neutral “recoverable streams” pipeline in Orleans.Streaming to unify durable resume/checkpoint semantics across streaming providers, and migrates Kinesis + the ADO.NET (alpha) provider to use retained-log + optimistic-concurrency checkpointing while preserving compatibility facades where required.

Changes:

• Add a reusable recoverable receiver/cache/checkpoint coordination layer (source + data adapter + pooled cache + checkpoint store).
• Update pooled cache token comparison to support provider-defined offset ordering, and add periodic delivery-progress updates in the pulling agent.
• Migrate Kinesis to arbitrary-precision shard sequence tokens and pooled recoverable receiver behavior; replace ADO.NET streaming with a versioned immutable partition log + retention/cleanup configuration.

Show a summary per file

File	Description
test/Orleans.Streaming.Tests/OrleansRuntime/Streams/EncodedOffsetPooledQueueCacheTests.cs	New unit tests validating adapter-aware cached offset comparisons and cursor behavior.
test/Orleans.Streaming.Tests/Checkpointers/StreamQueueCheckpointerTests.cs	Adds coverage for numeric-boundary ordering when persisting checkpoints.
test/Orleans.Streaming.Tests/Checkpointers/ReusableStreamQueueCheckpointerTests.cs	New tests for retry semantics on conditional-update conflicts using store versions.
test/Extensions/Orleans.Streaming.Kinesis.Tests/KinesisSequenceTokenTests.cs	New tests for BigInteger shard-sequence ordering, equality, and serialization.
test/Extensions/Orleans.Streaming.Kinesis.Tests/KinesisRuntimeTests.cs	Expands runtime tests for rewindable factory, pooled receiver init/read, and checkpoint updates.
test/Extensions/Orleans.Streaming.Kinesis.Tests/KinesisBatchContainerTests.cs	Adds tests for event index assignment, legacy payload shape, and recoverable adapter behavior.
test/Extensions/Orleans.Streaming.Kinesis.Tests/KinesisAdapterTests.cs	Updates adapter tests to validate cache cursor replay using first observed tokens.
test/Extensions/Orleans.Streaming.Kinesis.Tests/DynamoDBStreamQueueCheckpointerTests.cs	Updates tests for versioned checkpoint store state and expected-version semantics.
test/Extensions/Orleans.Streaming.EventHubs.Tests/CheckpointerTests/EventHubCheckpointerTests.cs	Adjusts Event Hubs tests for new inner checkpointer/state/version model.
test/Extensions/Orleans.Streaming.EventHubs.Tests/CheckpointerTests/AzureTableStreamQueueCheckpointerContractTests.cs	New contract tests reusing common checkpointer test suite for Azure Table.
test/Extensions/Orleans.AdoNet.Tests/Streaming/AdoNetStreamSchemaTests.cs	New tests validating v2 immutable partition-log schema scripts.
test/Extensions/Orleans.AdoNet.Tests/Streaming/AdoNetStreamMessageStreamIdTests.cs	New tests ensuring StreamId bytes/namespace boundary round-trip from persisted columns.
test/Extensions/Orleans.AdoNet.Tests/Streaming/AdoNetStreamFailureHandlerTests.cs	Removes dead-letter/visibility-timeout failure handler tests (schema replaced).
test/Extensions/Orleans.AdoNet.Tests/Streaming/AdoNetQueueAdapterTests.cs	Updates tests for rewindable adapter and new persisted message shape.
test/Extensions/Orleans.AdoNet.Tests/Streaming/AdoNetQueueAdapterReceiverTests.cs	Removes legacy receiver tests tied to old dequeue/confirm schema.
test/Extensions/Orleans.AdoNet.Tests/Streaming/AdoNetQueueAdapterFactoryTests.cs	Updates factory tests for shared receiver/cache registry and fault-on-failure option.
test/Extensions/Orleans.AdoNet.Tests/Streaming/AdoNetBatchContainerTests.cs	Updates message model + adds recoverable adapter lazy decode tests.
test/Extensions/Orleans.AdoNet.Tests/AdoNetOptionsValidatorTests.cs	Adds validation tests for new retained-log/retention/cleanup streaming options.
test/Benchmarks.AdoNet/Streaming/RetainedLogReadCheckpointBenchmark.cs	New benchmarks for ordered reads, epoch-fenced checkpointing, and cleanup.
test/Benchmarks.AdoNet/Streaming/RetainedLogAppendBenchmark.cs	New benchmarks for immutable log append throughput/lock contention.
test/Benchmarks.AdoNet/Streaming/MessageQueueingBenchmark.cs	Removes legacy queueing benchmark (schema replaced).
test/Benchmarks.AdoNet/Streaming/MessageDequeueingBenchmark.cs	Removes legacy dequeueing benchmark (schema replaced).
src/Orleans.Streaming/PersistentStreams/PersistentStreamPullingAgent.cs	Adds periodic delivery-progress updates and adapter-compare exception handling.
src/Orleans.Streaming/PersistentStreams/Options/PersistentStreamProviderOptions.cs	Adds configurable delivery-progress update interval option.
src/Orleans.Streaming/Common/RecoverableStreams/RecoverableStreamStartPosition.cs	New value type describing checkpoint/start-from-now policy.
src/Orleans.Streaming/Common/RecoverableStreams/RecoverableStreamReceiver.cs	New coordinator implementing receiver+cache with durable checkpoint integration.
src/Orleans.Streaming/Common/RecoverableStreams/RecoverableStreamQueueCache.cs	New pooled cache wrapper supporting offset-aware purge/progress mechanics.
src/Orleans.Streaming/Common/RecoverableStreams/QueueAdapterReceiverRegistry.cs	New per-queue registry ensuring receiver/cache instance identity sharing.
src/Orleans.Streaming/Common/RecoverableStreams/IRecoverableStreamSource.cs	New abstraction for ordered partition reads and lifecycle hooks.
src/Orleans.Streaming/Common/RecoverableStreams/IRecoverableStreamQueueCache.cs	New cache contract used by the recoverable coordinator.
src/Orleans.Streaming/Common/RecoverableStreams/IRecoverableStreamDataAdapter.cs	New adapter contract bridging provider records and pooled cache storage.
src/Orleans.Streaming/Common/PooledCache/PooledQueueCache.cs	Routes comparisons through ICacheDataAdapter.Compare + uses adapter-aware block search.
src/Orleans.Streaming/Common/PooledCache/IEvictionStrategy.cs	Adds OnPurgeCompleted hook for owners which purge messages directly.
src/Orleans.Streaming/Common/PooledCache/ICacheDataAdapter.cs	Adds default Compare(ref CachedMessage, StreamSequenceToken) extensibility point.
src/Orleans.Streaming/Common/PooledCache/ChronologicalEvictionStrategy.cs	Refactors purge accounting and fixes buffer release logic for empty-cache cases.
src/Orleans.Streaming/Common/PooledCache/CachedMessageBlock.cs	Adds adapter-aware index search to support provider offset ordering.
src/Orleans.Streaming/Common/EventSequenceTokenV2.cs	Improves equality/compare interoperability with EventSequenceToken.
src/Orleans.Streaming/Common/EventSequenceToken.cs	Improves equality/compare interoperability with EventSequenceTokenV2.
src/Orleans.Streaming/Checkpointers/StreamQueueCheckpointerOptions.cs	New options type for backend-neutral checkpointer.
src/Orleans.Streaming/Checkpointers/StreamQueueCheckpointer.cs	New optimistic-concurrency, throttled checkpoint persistence state machine.
src/Orleans.Streaming/Checkpointers/StreamCheckpointStoreState.cs	New checkpoint+version state type for conditional updates.
src/Orleans.Streaming/Checkpointers/IStreamCheckpointStore.cs	New conditional checkpoint store abstraction (load + update w/ expected version).
src/Orleans.Streaming/Checkpointers/GrainStreamQueueCheckpointer.cs	Refactors grain-based checkpointer to wrap the new backend-neutral checkpointer.
src/Azure/Shared/Storage/AzureTableDataManager.cs	Adds cancellation support and conditional update helper for table entities.
src/Azure/Orleans.Streaming.EventHubs/Providers/Streams/EventHub/EventHubCheckpointer.cs	Adds CancellationToken overloads and threads cancellation through creation.
src/Azure/Orleans.Streaming.EventHubs/Providers/Streams/EventHub/EventHubAdapterFactory.cs	Switches receiver tracking to shared registry for receiver/cache identity.
src/Azure/Orleans.Streaming.EventHubs/Providers/Streams/EventHub/AzureTableStreamQueueCheckpointerFactory.cs	Adds CancellationToken overload and forwards to new Create signature.
src/AWS/Orleans.Streaming.Kinesis/Streams/KinesisSequenceToken.cs	Uses BigInteger-based ordering/equality for shard sequence numbers.
src/AWS/Orleans.Streaming.Kinesis/Streams/KinesisRecoverableStream.cs	Adds recoverable source+adapter types with raw payload caching and offset compare.
src/AWS/Orleans.Streaming.Kinesis/Streams/KinesisPooledAdapterReceiver.cs	Adds pooled recoverable receiver combining receiver+cache and lifecycle control.
src/AWS/Orleans.Streaming.Kinesis/Streams/KinesisBatchContainer.cs	Adds cached-record constructor and durable-order CompareTo behavior.
src/AWS/Orleans.Streaming.Kinesis/Streams/KinesisAdapterReceiver.cs	Clarifies receiver-local ordinal role vs durable shard sequence ordering.
src/AWS/Orleans.Streaming.Kinesis/Streams/KinesisAdapterFactory.cs	Makes Kinesis adapter rewindable and exposes receiver/cache via shared registry.
src/AWS/Orleans.Streaming.Kinesis/Streams/DynamoDBStreamQueueCheckpointer.cs	Moves DynamoDB checkpointer to new StreamQueueCheckpointer model.
src/AWS/Orleans.Streaming.Kinesis/Streams/DynamoDBStreamCheckpointStore.cs	Converts store to IStreamCheckpointStore returning checkpoint+version state.
src/api/Azure/Orleans.Streaming.EventHubs/Orleans.Streaming.EventHubs.cs	Updates public API surface for new overloads/signatures.
src/api/AdoNet/Orleans.Streaming.AdoNet/Orleans.Streaming.AdoNet.cs	Updates public API surface for new ADO.NET streaming options surface.
src/AdoNet/Shared/Storage/DbStoredQueries.cs	Adds schema-version validation and replaces legacy query keys with v2 retained-log keys.
src/AdoNet/Shared/Storage/DbExtensions.cs	Fixes AddParameter to forward DbType to CreateParameter.
src/AdoNet/Orleans.Streaming.AdoNet/README.md	Documents retained-log behavior and alpha schema upgrade guidance.
src/AdoNet/Orleans.Streaming.AdoNet/Extensions.cs	Removes legacy time/ceiling helpers used by old schema paths.
src/AdoNet/Orleans.Streaming.AdoNet/AdoNetStreamPartitionState.cs	Adds model for partition checkpoint + retention bounds.
src/AdoNet/Orleans.Streaming.AdoNet/AdoNetStreamOptionsValidator.cs	Validates new retained-log options (max read, retention, cleanup, init timeout).
src/AdoNet/Orleans.Streaming.AdoNet/AdoNetStreamOptions.cs	Replaces visibility/dead-letter config with retained-log/retention/cleanup options.
src/AdoNet/Orleans.Streaming.AdoNet/AdoNetStreamMessage.cs	Updates message schema to include StreamId bytes + namespace boundary and reconstruct StreamId.
src/AdoNet/Orleans.Streaming.AdoNet/AdoNetStreamFailureHandler.cs	Replaces dead-letter mutation with logging-only failure handling and optional faulting.
src/AdoNet/Orleans.Streaming.AdoNet/AdoNetStreamDeadLetter.cs	Removes legacy dead-letter entity model.
src/AdoNet/Orleans.Streaming.AdoNet/AdoNetStreamConfirmationAck.cs	Removes legacy confirmation ack entity model.
src/AdoNet/Orleans.Streaming.AdoNet/AdoNetStreamConfirmation.cs	Removes legacy confirmation entity model.
src/AdoNet/Orleans.Streaming.AdoNet/AdoNetStreamCleanupResult.cs	Adds bounded cleanup result model for retained-log cleanup.
src/AdoNet/Orleans.Streaming.AdoNet/AdoNetStreamCheckpointUpdate.cs	Adds epoch-fenced checkpoint update result model.
src/AdoNet/Orleans.Streaming.AdoNet/AdoNetRecoverableStream.cs	Implements recoverable partition source + checkpoint store backed by relational queries.
src/AdoNet/Orleans.Streaming.AdoNet/AdoNetQueueAdapterReceiver.cs	Refactors receiver into recoverable receiver+cache coordinator.
src/AdoNet/Orleans.Streaming.AdoNet/AdoNetQueueAdapterFactory.cs	Exposes queue cache via factory/adaptor shared registry and configures fault behavior.
src/AdoNet/Orleans.Streaming.AdoNet/AdoNetQueueAdapter.cs	Makes adapter rewindable; appends immutable records and shares receiver/cache instances.
src/AdoNet/Orleans.Streaming.AdoNet/AdoNetBatchContainer.cs	Aligns batch container dequeue count semantics with retained-log model.
docs/site/src/content/docs/streaming/stream-providers.md	Updates provider comparison + ADO.NET alpha documentation for retained-log semantics.
docs/site/src/content/docs/implementation/streams-implementation/index.md	Adds implementation doc note for recoverable partition composition model.

Review details

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

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

[Copilot]

Review details

Suppressed comments (2)

src/Orleans.Streaming/Common/RecoverableStreams/RecoverableStreamQueueCache.cs:191

• RecoverableStreamQueueCache.Dispose() only nulls the eviction callback and does not drain the cache or dispose/return any FixedSizeBuffer instances tracked by ChronologicalEvictionStrategy. Because buffers are only returned to the pool when purged, shutting down and recreating receivers can retain large buffers indefinitely and increase memory usage over time. Drain the cache and notify the eviction strategy so it can release all in-use buffers during disposal.
  src/AWS/Orleans.Streaming.Kinesis/Streams/KinesisRecoverableStream.cs:235
• KinesisRecoverableStreamDataAdapter.Compare parses both the cached shard sequence and the token shard sequence into BigInteger on every comparison. This compare method is used in cache cursor positioning and purge logic, so repeated BigInteger.Parse calls can become a CPU hot path even though KinesisSequenceToken already caches its parsed value. Consider exposing the cached numeric value from KinesisSequenceToken (eg, an internal BigInteger property/method) and using it here so the token side is not reparsed each time; optionally also cache the parsed value for the cached message if feasible.

    public int Compare(ref CachedMessage cachedMessage, StreamSequenceToken token)
    {
        if (token is not KinesisSequenceToken kinesisToken)
        {
            throw new ArgumentOutOfRangeException(nameof(token));
        }

        var offset = 0;
        var shardSequence = SegmentBuilder.ReadNextString(cachedMessage.Segment, ref offset)!;
        var difference = BigInteger.Parse(shardSequence, NumberStyles.None, CultureInfo.InvariantCulture)
            .CompareTo(BigInteger.Parse(kinesisToken.ShardSequence, NumberStyles.None, CultureInfo.InvariantCulture));
        return difference != 0 ? difference : cachedMessage.EventIndex.CompareTo(kinesisToken.EventIndex);

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

[ReubenBond]

CI failure analysis: the PostgreSQL, MySQL/MariaDB, and SQL Server jobs fail consistently on both net8.0 and net10.0 because the new ADO.NET retained-log tests execute commands whose CommandText is empty (System.InvalidOperationException: The CommandText property has not been properly initialized). The Event Hubs jobs also fail in the same run. This appears branch-related rather than transient infrastructure; I am rebasing the branch and correcting the query initialization before retriggering CI.

[ReubenBond]

Updated the branch in commit b20757c after rebasing onto current main. The fix removes whitespace-only MySQL batches, reads the retained message shape explicitly, aligns the SQL Server lock test with the procedure error, preserves/reclaims pooled buffers correctly, and fixes empty-cache cursors so they deliver the first subsequently added record. Fresh CI has been triggered.

[Copilot]

Review details

Suppressed comments (1)

src/AWS/Orleans.Streaming.Kinesis/Streams/KinesisRecoverableStream.cs:235

• Compare parses both shard sequence strings using BigInteger.Parse for every comparison. This is on the cache/cursor hot path (cursor positioning, delivery-progress purging, etc) and can become a significant CPU cost. Since shard sequences are decimal digits, you can compare numerically by trimming leading zeros, comparing length, then ordinal-comparing spans, avoiding BigInteger parsing entirely.

        var offset = 0;
        var shardSequence = SegmentBuilder.ReadNextString(cachedMessage.Segment, ref offset)!;
        var difference = BigInteger.Parse(shardSequence, NumberStyles.None, CultureInfo.InvariantCulture)
            .CompareTo(BigInteger.Parse(kinesisToken.ShardSequence, NumberStyles.None, CultureInfo.InvariantCulture));
        return difference != 0 ? difference : cachedMessage.EventIndex.CompareTo(kinesisToken.EventIndex);

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

[Copilot]

Review details

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

[Copilot]

Review details

Suppressed comments (1)

Previously missed (1) — in code that hasn't changed since the last review.

src/Orleans.Streaming/Checkpointers/StreamQueueCheckpointer.cs:35

• The ArgumentOutOfRangeException uses paramName: nameof(options), which makes the exception metadata point at the wrong parameter. Since the invalid value is options.PersistInterval, the exception should name that member (this improves diagnostics and matches typical .NET patterns).

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

[Copilot]

Review details

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

[Copilot]

Review details

Suppressed comments (2)

Previously missed (1) — in code that hasn't changed since the last review.

src/AWS/Orleans.Streaming.Kinesis/Streams/KinesisRecoverableStream.cs:176

• GetStreamPosition calls queueMessage.Record.Data.ToArray() to deserialize the body, which allocates/copies the record payload. Since this method can run for every record admitted to the cache, this defeats the allocation-minimizing goal of the pooled pipeline. You can usually avoid this allocation by deserializing directly from the underlying MemoryStream buffer when available (falling back to ToArray() only when needed).

This issue also appears on line 184 of the same file.

    public StreamPosition GetStreamPosition(KinesisCacheRecord queueMessage)
    {
        queueMessage.Body ??= serializer.Deserialize(queueMessage.Record.Data.ToArray())!;
        return new(

src/AWS/Orleans.Streaming.Kinesis/Streams/KinesisRecoverableStream.cs:190

• FromQueueMessage materializes the Kinesis record payload via queueMessage.Record.Data.ToArray() and then copies it into the pooled segment, causing an extra allocation/copy per record. Since SegmentBuilder.Append and CalculateAppendSize accept spans, this can be allocation-free in the common case by using MemoryStream.TryGetBuffer() (with a fallback to ToArray() only when required).

        var payload = queueMessage.Record.Data.ToArray();
        var size = SegmentBuilder.CalculateAppendSize(queueMessage.Record.SequenceNumber)
            + SegmentBuilder.CalculateAppendSize(payload);
        var segment = getSegment(size);
        var offset = 0;
        SegmentBuilder.Append(segment, ref offset, queueMessage.Record.SequenceNumber);
        SegmentBuilder.Append(segment, ref offset, payload);

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

[Copilot]

Copilot review overview

🔵 Needs a closer look

Review tier: Lite
Findings: 1

New issues introduced by this change (1)

Severity	Finding
	src/​AWS/​Orleans.Streaming.Kinesis/​Streams/​KinesisPooledAdapterReceiver.cs — KinesisPooledAdapterReceiver creates a long-lived CancellationTokenSource (_lifecycleCancellation)…

[Copilot]

Copilot review overview

🔵 Needs a closer look

Review tier: Lite
Findings: None

Issues resolved since last review (1)

Severity	Finding
	src/​AWS/​Orleans.Streaming.Kinesis/​Streams/​KinesisPooledAdapterReceiver.cs — KinesisPooledAdapterReceiver creates a long-lived CancellationTokenSource (_lifecycleCancellation)… View resolved comment