feat(streaming)!: unify stream partition recovery

docs/site/src/content/docs/host/monitoring/signals.md

@@ -73,4 +73,4 @@ Page on user impact or imminent data/availability risk. Route isolated warnings
 
 ## Dashboard and telemetry backends
 
-The Orleans Dashboard provides a current operational view and method profiling. It isn't a replacement for retained logs, metrics, traces, or alerts. Secure it as an administrative endpoint and use OTLP for durable telemetry. See [Orleans Dashboard](../../dashboard/index.md).
+The Orleans Dashboard provides a current operational view and method profiling. Use logs, metrics, traces, and alerts for durable telemetry. Secure the dashboard as an administrative endpoint and export telemetry through OTLP. See [Orleans Dashboard](../../dashboard/index.md).

docs/site/src/content/docs/implementation/streams-implementation/index.md

@@ -81,7 +81,7 @@ The default maximum adapter batch-container batch size is 1 and the empty-poll p
 
 An <xref:Orleans.Streams.IQueueCache> decouples queue reads from consumer delivery. Each subscription has an <xref:Orleans.Streams.IQueueCacheCursor>, so a slow consumer does not directly block a fast consumer at a later cursor.
 
-The cache tracks the earliest delivery progress across active subscriptions. Purging must not remove an item still needed by any cursor. <xref:Orleans.Providers.Streams.Common.SimpleQueueCache> uses pressure buckets to stop or slow reads as lag grows instead of discarding undelivered events. Its default capacity is 4,096 batch containers.
+The cache tracks the earliest contiguous partition position which is safe across active subscriptions. A matching record becomes safe after delivery or intentional filtering. A cursor also advances safely across records for other streams when no earlier matching delivery is pending, so a quiet stream does not pin an otherwise busy partition. Purging must not remove an item still needed by any cursor. <xref:Orleans.Providers.Streams.Common.SimpleQueueCache> uses pressure buckets to stop or slow reads as lag grows instead of discarding undelivered events. Its default capacity is 4,096 batch containers.
 
 ```mermaid
 flowchart TB
@@ -97,11 +97,13 @@ flowchart TB
 
 Cache capacity is not durability. The queue remains the durable boundary, subject to the adapter's acknowledgement contract.
 
+Recoverable partitioned stream providers can compose a stream partition pipeline from <xref:Orleans.Providers.Streams.Common.RecoverableStreamReceiver%601>, a partition source, and a data adapter. The pipeline admits immutable stream records into pooled storage, reconstructs batches lazily, reconciles the earliest safe subscription scan/delivery watermark, and persists a checkpoint which resumes strictly after that position.
+
 ## Pub-sub handshake
 
 The agent registers as a producer for each stream and obtains subscription records from stream pub-sub. It holds a pin cursor while subscription handshakes complete so cache cleanup cannot pass the requested start token. New subscription notifications update the agent's local pub-sub cache.
 
-Sequence tokens allow a rewindable adapter to start from a supported historical position. An adapter whose <xref:Orleans.Streams.IQueueAdapter.IsRewindable?displayProperty=nameWithType> property is `false` must reject unsupported tokens rather than pretending to honor them.
+Sequence tokens allow a rewindable adapter to start from a supported historical position. A start token is inclusive and remains unsafe until its record is delivered or intentionally filtered. A delivery handshake token confirms that its position was already processed. Exact `EventSequenceToken` and `EventSequenceTokenV2` values interoperate for legacy compatibility. Derived tokens compare only with the same concrete type unless the provider overrides equality, ordering, and hashing together. An adapter whose <xref:Orleans.Streams.IQueueAdapter.IsRewindable?displayProperty=nameWithType> property is `false` must reject unsupported tokens rather than pretending to honor them.
 
 ## Delivery and failure semantics
 

docs/site/src/content/docs/streaming/custom-queue-adapter.md

@@ -51,7 +51,7 @@ The factory composes the adapter with queue mapping, caching, and failure handli
 
 `SimpleQueueAdapterCache` is suitable for a non-rewindable adapter whose queue remains the durability boundary. A rewindable adapter usually needs a cache and sequence-token implementation which can position cursors at retained historical messages.
 
-`AddPersistentStreams` leaves checkpointing to the adapter. The non-rewindable example acknowledges completed messages through its receiver and therefore has no independent checkpoint. For a retained-log transport, implement an <xref:Orleans.Streams.IStreamQueueCheckpointerFactory>, have the receiver or cache load and update the per-partition position, and register it as a named component with `ConfigureComponent`. Persist a checkpoint only after all consumers have advanced beyond the corresponding cached messages. A no-op checkpointer is suitable only when replay position is deliberately disposable.
+`AddPersistentStreams` leaves checkpointing to the adapter. The non-rewindable example acknowledges completed messages through its receiver and therefore has no independent checkpoint. For a partitioned stream transport, implement an <xref:Orleans.Streams.IStreamQueueCheckpointerFactory>, have the receiver or cache load and update the stream partition position, and register it as a named component with `ConfigureComponent`. Treat a requested cursor start as inclusive: selecting that position does not confirm its record. Persist only the earliest contiguous partition position which every subscription has delivered, intentionally filtered, or safely scanned as belonging to another stream. A no-op checkpointer is suitable only when replay position is deliberately disposable.
 
 ## Register the provider
 
@@ -75,6 +75,9 @@ Test the adapter against the real queue service, including:
 1. queue ownership moving between silos during membership changes;
 1. duplicate delivery and consumer idempotency;
 1. stable stream-to-partition mapping across restarts and upgrades; and
-1. sustained load beyond cache capacity to verify backpressure and queue retention.
+1. sustained load beyond cache capacity to verify backpressure and queue retention;
+1. quiet and busy streams sharing a partition, including restart after the quiet cursor scans unrelated records;
+1. cancellation while partition ownership acquisition is blocked, followed by reassignment and late command completion; and
+1. sequence-token equality, ordering, and hashing in both comparison directions.
 
 Monitor queue depth and oldest-message age by partition, receive and acknowledgement latency, redelivery count, throttling, pulling-agent errors, and consumer delivery failures. Alert before retention or visibility limits can cause data loss or a redelivery storm.

docs/site/src/content/docs/streaming/data-adapters.md

@@ -62,7 +62,7 @@ For Event Hubs, derive from <xref:Orleans.Streaming.EventHubs.EventHubDataAdapte
 - <xref:Orleans.Streaming.EventHubs.EventHubDataAdapter.ToQueueMessage*> to encode events published through Orleans; and
 - <xref:Orleans.Streaming.EventHubs.EventHubDataAdapter.GetPartitionKey*> to select the physical Event Hubs partition key.
 
-The adapter also participates in cache conversion and sequence positioning through <xref:Orleans.Streaming.EventHubs.IEventHubDataAdapter>. Preserve the Event Hubs offset and sequence number when constructing batch tokens so checkpoint and rewind behavior remains aligned with the partition log.
+The adapter also participates in cache conversion and sequence positioning through <xref:Orleans.Streaming.EventHubs.IEventHubDataAdapter>. Preserve the Event Hubs offset and sequence number when constructing batch tokens so checkpoint and rewind behavior remains aligned with the stream partition.
 
 Register the adapter and Event Hubs connection under the same provider name on silos and publishing clients. The silo registration also configures durable Azure Table checkpoints:
 

docs/site/src/content/docs/streaming/stream-providers.md

@@ -18,7 +18,7 @@ A stream provider connects the Orleans streaming API to a transport and defines
 | Azure Event Hubs | [`Microsoft.Orleans.Streaming.EventHubs`](https://www.nuget.org/packages/Microsoft.Orleans.Streaming.EventHubs) | Stable | Yes, within Event Hubs retention | Yes | Event Hubs namespace, hub, consumer group, and checkpoint storage |
 | Amazon Kinesis | [`Microsoft.Orleans.Streaming.Kinesis`](https://www.nuget.org/packages/Microsoft.Orleans.Streaming.Kinesis) | Stable | Yes, within Kinesis retention | Yes | Kinesis data stream, AWS credentials, region, and durable checkpoint storage |
 | Amazon SQS | [`Microsoft.Orleans.Streaming.SQS`](https://www.nuget.org/packages/Microsoft.Orleans.Streaming.SQS) | Stable | Yes, within SQS retention | No | AWS account, queue permissions, region/endpoint configuration |
-| ADO.NET | [`Microsoft.Orleans.Streaming.AdoNet`](https://www.nuget.org/packages/Microsoft.Orleans.Streaming.AdoNet) | **Alpha** | Yes, in relational tables until expiry/dead-letter eviction | No | Supported database, ADO.NET driver, and Orleans streaming SQL schema |
+| ADO.NET partitioned stream | [`Microsoft.Orleans.Streaming.AdoNet`](https://www.nuget.org/packages/Microsoft.Orleans.Streaming.AdoNet) | **Alpha** | Yes, in relational stream partitions | No; resumes after its durable safe checkpoint | Supported database, ADO.NET driver, and matching Orleans streaming SQL schema |
 | NATS JetStream | [`Microsoft.Orleans.Streaming.NATS`](https://www.nuget.org/packages/Microsoft.Orleans.Streaming.NATS) | **Alpha** | Configurable; file storage is the default | No | NATS server with JetStream and sufficient storage; subject/stream administration |
 | Redis Streams | [`Microsoft.Orleans.Streaming.Redis`](https://www.nuget.org/packages/Microsoft.Orleans.Streaming.Redis) | **Alpha** | Configurable through Redis persistence and stream retention | Yes, while entries remain | Redis deployment, persistence/HA policy, and retention sizing |
 
@@ -58,15 +58,17 @@ The Event Hubs provider supports a custom data adapter for provider-specific wir
 
 ## Amazon Kinesis
 
-Register [Amazon Kinesis Data Streams](https://docs.aws.amazon.com/streams/latest/dev/introduction.html) with <xref:Orleans.Hosting.SiloBuilderExtensions.AddKinesisStreams*>. Kinesis retains events independently of Orleans, and the provider persists each shard's last delivered sequence number so that delivery can resume after shutdown or queue reassignment. See [Stream with Amazon Kinesis](kinesis-streaming.md) for configuration, checkpoint choices, and operational constraints.
+Register [Amazon Kinesis Data Streams](https://docs.aws.amazon.com/streams/latest/dev/introduction.html) with <xref:Orleans.Hosting.SiloBuilderExtensions.AddKinesisStreams*>. Kinesis retains events independently of Orleans, and the provider persists each shard's earliest safe scan/delivery position so that delivery can resume after shutdown or queue reassignment. Kinesis sequence tokens compare only with other Kinesis tokens using the numeric shard sequence and event index. See [Stream with Amazon Kinesis](kinesis-streaming.md) for configuration, checkpoint choices, and operational constraints.
 
 ## Amazon SQS
 
 Register [Amazon SQS](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/welcome.html) with <xref:Orleans.Hosting.SiloBuilderExtensions.AddSqsStreams*>. Standard queues provide at-least-once delivery, while FIFO queues preserve ordering within each Orleans stream. SQS redelivers after the [visibility timeout](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-visibility-timeout.html) when processing isn't acknowledged. See [Stream with Amazon SQS](sqs-streaming.md) for standard and FIFO configuration, custom data adapters, permissions, and operational guidance.
 
-## ADO.NET streaming (alpha)
+## ADO.NET partitioned stream provider (alpha)
 
-Register [ADO.NET](https://learn.microsoft.com/dotnet/framework/data/adonet/ado-net-overview) streaming with `AddAdoNetStreams`. Install the matching database driver and apply the SQL Server, PostgreSQL, or MySQL streaming schema shipped in the package source. Messages are durable in relational tables but expire and can move to dead letters according to `AdoNetStreamOptions`. The provider isn't rewindable.
+Register the [ADO.NET](https://learn.microsoft.com/dotnet/framework/data/adonet/ado-net-overview) partitioned stream provider with <xref:Orleans.Hosting.SiloBuilderAdoNetStreamExtensions.AddAdoNetStreams*>. Install the matching database driver and apply the SQL Server, PostgreSQL, or MySQL streaming schema shipped in the package source. Each queue maps to an immutable, ordered stream partition with a durable, ownership-fenced checkpoint. The stream partition pipeline resumes strictly after the earliest safe scan/delivery checkpoint, redelivering uncheckpointed records after recovery. Ownership acquisition propagates cancellation, and a receiver retains its queue reservation until an in-flight acquisition command settles. <xref:Orleans.Configuration.AdoNetStreamOptions.RetentionPeriod*> retains checkpointed records for one day by default before cleanup. <xref:Orleans.Configuration.AdoNetStreamOptions.MaximumRetentionPeriod*> can impose a hard storage ceiling; deleting unprocessed records produces retention gap diagnostics. Retention and cleanup intervals round fractional seconds upward. Cleanup is bounded by <xref:Orleans.Configuration.AdoNetStreamOptions.CleanupBatchSize*> and deletes a contiguous eligible prefix. <xref:Orleans.Configuration.AdoNetStreamOptions.FaultOnDeliveryFailure*> can fault one failing subscription while preserving the multicast partition record.
+
+The current alpha schema is versioned and isn't compatible with the former queue, visibility-timeout, confirmation, or dead-letter schema. There is no in-place migration for alpha data. Before upgrading, stop producers and consumers, drop the former `OrleansStreamMessage`, `OrleansStreamDeadLetter`, `OrleansStreamControl`, and `OrleansStreamMessageSequence` objects, remove their streaming routines and `OrleansQuery` rows, and then apply the current streaming script. Drop `OrleansStreamPartition` too if a partial installation exists. Existing alpha rows aren't read or silently converted; preserve them externally first if the payloads are required.
 
 ## NATS JetStream streaming (alpha)
 

src/AWS/Orleans.Streaming.Kinesis/Streams/DynamoDBStreamCheckpointStore.cs

@@ -13,15 +13,7 @@
 
 namespace Orleans.Streaming.Kinesis
 {
-    internal interface IDynamoDBStreamCheckpointStore
-    {
-        ValueTask<string> Load(CancellationToken cancellationToken);
-
-        ValueTask<string> Update(
-            string checkpoint,
-            string expectedCheckpoint,
-            CancellationToken cancellationToken);
-    }
+    internal interface IDynamoDBStreamCheckpointStore : IStreamCheckpointStore;
 
     internal sealed partial class DynamoDBStreamCheckpointStore : IDynamoDBStreamCheckpointStore
     {
@@ -57,27 +49,27 @@ public DynamoDBStreamCheckpointStore(
             };
         }
 
-        public async ValueTask<string> Load(CancellationToken cancellationToken)
+        public async ValueTask<StreamCheckpointStoreState> Load(CancellationToken cancellationToken)
         {
             await _mutex.WaitAsync(cancellationToken);
             try
             {
                 await LoadCore(cancellationToken);
-                return _checkpoint;
+                return GetState();
             }
             finally
             {
                 _mutex.Release();
             }
         }
 
-        public async ValueTask<string> Update(
+        public async ValueTask<StreamCheckpointStoreState> Update(
             string checkpoint,
-            string expectedCheckpoint,
+            string expectedVersion,
             CancellationToken cancellationToken)
         {
             ArgumentNullException.ThrowIfNull(checkpoint);
-            ArgumentNullException.ThrowIfNull(expectedCheckpoint);
+            ArgumentNullException.ThrowIfNull(expectedVersion);
 
             await _mutex.WaitAsync(cancellationToken);
             try
@@ -87,9 +79,12 @@ public async ValueTask<string> Update(
                     await LoadCore(cancellationToken);
                 }
 
-                if (!string.Equals(_checkpoint, expectedCheckpoint, StringComparison.Ordinal))
+                var currentVersion = _version == 0
+                    ? string.Empty
+                    : _version.ToString(CultureInfo.InvariantCulture);
+                if (!string.Equals(currentVersion, expectedVersion, StringComparison.Ordinal))
                 {
-                    return _checkpoint;
+                    return GetState();
                 }
 
                 try
@@ -137,14 +132,20 @@ public async ValueTask<string> Update(
                     await LoadCore(cancellationToken);
                 }
 
-                return _checkpoint;
+                return GetState();
             }
+
             finally
             {
                 _mutex.Release();
             }
         }
 
+        private StreamCheckpointStoreState GetState()
+            => new(
+                _checkpoint,
+                _version == 0 ? string.Empty : _version.ToString(CultureInfo.InvariantCulture));
+
         internal static async Task InitializeTable(
             IAmazonDynamoDB client,
             DynamoDBStreamQueueCheckpointerOptions options,

src/AWS/Orleans.Streaming.Kinesis/Streams/DynamoDBStreamQueueCheckpointer.cs

@@ -11,15 +11,15 @@ namespace Orleans.Streaming.Kinesis
     /// </summary>
     internal sealed class DynamoDBStreamQueueCheckpointer : IStreamQueueCheckpointer<string>
     {
-        private readonly GrainStreamQueueCheckpointer _inner;
+        private readonly StreamQueueCheckpointer _inner;
 
         internal DynamoDBStreamQueueCheckpointer(
             IDynamoDBStreamCheckpointStore store,
             DynamoDBStreamQueueCheckpointerOptions options)
         {
-            _inner = new GrainStreamQueueCheckpointer(
-                new StreamCheckpointStoreAdapter(store),
-                new GrainStreamQueueCheckpointerOptions
+            _inner = new StreamQueueCheckpointer(
+                store,
+                new StreamQueueCheckpointerOptions
                 {
                     CheckpointComparer = StreamCheckpointComparers.Numeric,
                     PersistInterval = options.PersistInterval,
@@ -63,15 +63,5 @@ public void Update(string offset, DateTime utcNow, CancellationToken cancellatio
         /// <inheritdoc />
         public Task FlushAsync(CancellationToken cancellationToken) => _inner.FlushAsync(cancellationToken);
 
-        private sealed class StreamCheckpointStoreAdapter(IDynamoDBStreamCheckpointStore store) : IStreamCheckpointerGrain
-        {
-            public ValueTask<string> Load(CancellationToken cancellationToken) => store.Load(cancellationToken);
-
-            public ValueTask<string> Update(
-                string checkpoint,
-                string expectedCheckpoint,
-                CancellationToken cancellationToken)
-                => store.Update(checkpoint, expectedCheckpoint, cancellationToken);
-        }
     }
 }