Fix persistent stream subscriber registration race (#9927)

* Fix persistent stream registration race

Track the earliest stream token observed while subscriber registration is still in progress and use it to initialize the cursor when handshake has no explicit token.

This removes timing dependence from dropped-client producer scenarios without adding sleeps.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Keep earliest pending stream start token

Address PR feedback by updating pending token selection to retain the oldest token observed while registration is incomplete.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: ReubenBond

src/Orleans.Streaming/PersistentStreams/PersistentStreamPullingAgent.cs

@@ -243,6 +243,7 @@ private async Task AddSubscriber_Impl(
 
             if (await DoHandshakeWithConsumer(data, cacheToken))
             {
+                data.PendingStartToken = null;
                 data.IsRegistered = true;
                 if (data.State == StreamConsumerDataState.Inactive)
                     RunConsumerCursor(data).Ignore(); // Start delivering events if not actively doing so
@@ -268,15 +269,17 @@ private async Task<bool> DoHandshakeWithConsumer(
                          this.options.MaxEventDeliveryTime,
                          deliveryBackoffProvider);
 
-                    if (requestedHandshakeToken != null)
+                    var requestedToken = requestedHandshakeToken?.Token;
+                    if (requestedToken != null)
                     {
                         consumerData.SafeDisposeCursor(logger);
-                        consumerData.Cursor = queueCache.GetCacheCursor(consumerData.StreamId, requestedHandshakeToken.Token);
+                        consumerData.Cursor = queueCache.GetCacheCursor(consumerData.StreamId, requestedToken);
                     }
                     else
                     {
+                        var registrationToken = cacheToken ?? consumerData.PendingStartToken;
                         if (consumerData.Cursor == null) // if the consumer did not ask for a specific token and we already have a cursor, just keep using it.
-                            consumerData.Cursor = queueCache.GetCacheCursor(consumerData.StreamId, cacheToken);
+                            consumerData.Cursor = queueCache.GetCacheCursor(consumerData.StreamId, registrationToken);
                     }
                 }
                 catch (Exception exception)
@@ -298,7 +301,8 @@ private async Task<bool> DoHandshakeWithConsumer(
             {
                 try
                 {
-                    consumerData.Cursor = queueCache.GetCacheCursor(consumerData.StreamId, cacheToken);
+                    var registrationToken = cacheToken ?? consumerData.PendingStartToken;
+                    consumerData.Cursor = queueCache.GetCacheCursor(consumerData.StreamId, registrationToken);
                 }
                 catch (Exception)
                 {
@@ -513,6 +517,10 @@ private void StartInactiveCursors(StreamConsumerCollection streamData, StreamSeq
                 }
                 else
                 {
+                    if (consumerData.PendingStartToken is null || startToken.Older(consumerData.PendingStartToken))
+                    {
+                        consumerData.PendingStartToken = startToken;
+                    }
                     LogDebugPulledNewMessages(consumerData.StreamId);
                 }
             }

src/Orleans.Streaming/PersistentStreams/QueueStreamDataStructures.cs

@@ -33,6 +33,8 @@ internal sealed class StreamConsumerData
 
         [NonSerialized]
         public bool IsRegistered = false;
+        [NonSerialized]
+        public StreamSequenceToken? PendingStartToken;
 
         public StreamConsumerData(GuidId subscriptionId, QualifiedStreamId streamId, IStreamConsumerExtension streamConsumer, string filterData)
         {