Fix skipping events on resume handshake (#9336)

* Fix skipping events on resume handshake * Update test/Tester/StreamingTests/StreamingResumeTests.cs * Update test/Tester/StreamingTests/StreamingResumeTests.cs Co-authored-by: Günther Foidl <[email protected]> * fix test --------- Co-authored-by: Reuben Bond <[email protected]> Co-authored-by: Günther Foidl <[email protected]> Co-authored-by: Reuben Bond <[email protected]>
dotnet · Feb 10, 2025 · d2af146 · d2af146
1 parent b9827a4
commit d2af146
Show file tree

Hide file tree

Showing 5 changed files with 57 additions and 3 deletions.
diff --git a/src/Orleans.Streaming/Internal/StreamSubscriptionHandleImpl.cs b/src/Orleans.Streaming/Internal/StreamSubscriptionHandleImpl.cs
@@ -107,6 +107,14 @@ public async Task<StreamHandshakeToken> DeliverBatch(IBatchContainer batch, Stre
             {
                 if (!this.expectedToken.Equals(handshakeToken))
                     return this.expectedToken;
+
+                // Check if this even already has been delivered
+                if (IsRewindable)
+                {
+                    var currentToken = StreamHandshakeToken.CreateDeliveyToken(batch.SequenceToken);
+                    if (this.expectedToken.Equals(currentToken))
+                        return this.expectedToken;
+                }
             }
 
             if (batch is IBatchContainerBatch)
@@ -142,6 +150,13 @@ public async Task<StreamHandshakeToken> DeliverItem(object item, StreamSequenceT
             {
                 if (!this.expectedToken.Equals(handshakeToken))
                     return this.expectedToken;
+
+                // Check if this even already has been delivered
+                if (IsRewindable)
+                {
+                    if (this.expectedToken.Equals(currentToken))
+                        return this.expectedToken;
+                }
             }
 
             T typedItem;

diff --git a/src/Orleans.Streaming/PersistentStreams/PersistentStreamPullingAgent.cs b/src/Orleans.Streaming/PersistentStreams/PersistentStreamPullingAgent.cs
@@ -287,10 +287,7 @@ private async Task<bool> DoHandshakeWithConsumer(
                     if (requestedHandshakeToken != null)
                     {
                         consumerData.SafeDisposeCursor(logger);
-                        // The handshake token points to an already processed event, we need to advance the cursor to
-                        // the next event.
                         consumerData.Cursor = queueCache.GetCacheCursor(consumerData.StreamId, requestedHandshakeToken.Token);
-                        consumerData.Cursor.MoveNext(); //
                     }
                     else
                     {

diff --git a/test/Extensions/ServiceBus.Tests/Streaming/EHStreamingResumeTests.cs b/test/Extensions/ServiceBus.Tests/Streaming/EHStreamingResumeTests.cs
@@ -27,6 +27,12 @@ public void Configure(ISiloBuilder hostBuilder)
                         {
                             options.StreamInactivityPeriod = StreamInactivityPeriod;
                         }));
+                        b.ConfigureCacheEviction(ob => ob.Configure(options =>
+                        {
+                            options.MetadataMinTimeInCache = MetadataMinTimeInCache;
+                            options.DataMaxAgeInCache = DataMaxAgeInCache;
+                            options.DataMinTimeInCache = DataMinTimeInCache;
+                        }));
                         b.ConfigureEventHub(ob => ob.Configure(options =>
                         {
                             options.ConfigureTestDefaults(EHPath, EHConsumerGroup);

diff --git a/test/Tester/StreamingTests/MemoryStreamResumeTests.cs b/test/Tester/StreamingTests/MemoryStreamResumeTests.cs
@@ -27,6 +27,12 @@ public void Configure(ISiloBuilder hostBuilder)
                         {
                             options.StreamInactivityPeriod = StreamInactivityPeriod;
                         }));
+                        b.ConfigureCacheEviction(ob => ob.Configure(options =>
+                        {
+                            options.MetadataMinTimeInCache = MetadataMinTimeInCache;
+                            options.DataMaxAgeInCache = DataMaxAgeInCache;
+                            options.DataMinTimeInCache = DataMinTimeInCache;
+                        }));
                     });
             }
         }

diff --git a/test/Tester/StreamingTests/StreamingResumeTests.cs b/test/Tester/StreamingTests/StreamingResumeTests.cs
@@ -8,11 +8,25 @@ namespace Tester.StreamingTests
     public abstract class StreamingResumeTests : TestClusterPerTest
     {
         protected static readonly TimeSpan StreamInactivityPeriod = TimeSpan.FromSeconds(5);
+        protected static readonly TimeSpan MetadataMinTimeInCache = StreamInactivityPeriod * 100;
+        protected static readonly TimeSpan DataMaxAgeInCache = StreamInactivityPeriod * 5;
+        protected static readonly TimeSpan DataMinTimeInCache = StreamInactivityPeriod * 4;
 
         protected const string StreamProviderName = "StreamingCacheMissTests";
 
         [SkippableFact]
         public virtual async Task ResumeAfterInactivity()
+        {
+            await ResumeAfterInactivityImpl(false);
+        }
+
+        [SkippableFact]
+        public virtual async Task ResumeAfterInactivityNotInCache()
+        {
+            await ResumeAfterInactivityImpl(true);
+        }
+
+        protected virtual async Task ResumeAfterInactivityImpl(bool waitForCacheToFlush)
         {
             var streamProvider = this.Client.GetStreamProvider(StreamProviderName);
 
@@ -31,6 +45,22 @@ public virtual async Task ResumeAfterInactivity()
             // Wait for the stream to become inactive
             await Task.Delay(StreamInactivityPeriod.Multiply(3));
 
+            if (waitForCacheToFlush)
+            {
+                for (var i = 0; i < 5; i++)
+                {
+                    var otherStream = streamProvider.GetStream<byte[]>(nameof(IImplicitSubscriptionCounterGrain), Guid.NewGuid());
+                    await otherStream.OnNextAsync(interestingData);
+                }
+                // Wait a bit more for the cache to flush some events
+                await Task.Delay(StreamInactivityPeriod.Multiply(3));
+                for (var i = 0; i < 5; i++)
+                {
+                    var otherStream = streamProvider.GetStream<byte[]>(nameof(IImplicitSubscriptionCounterGrain), Guid.NewGuid());
+                    await otherStream.OnNextAsync(interestingData);
+                }
+            }
+
             await stream.OnNextAsync(interestingData);
 
             await Task.Delay(2_000);