fix culling of waiters in instance state (#93)

microsoft · Dec 22, 2021 · 266c31d · 266c31d
1 parent 3989fb0
commit 266c31d
Show file tree

Hide file tree

Showing 3 changed files with 27 additions and 8 deletions.
diff --git a/src/DurableTask.Netherite/Events/PartitionEvents/External/FromClients/WaitRequestReceived.cs b/src/DurableTask.Netherite/Events/PartitionEvents/External/FromClients/WaitRequestReceived.cs
@@ -4,6 +4,7 @@
 namespace DurableTask.Netherite
 {
     using DurableTask.Core;
+    using System;
     using System.Runtime.Serialization;
     using System.Text;
 
@@ -22,6 +23,9 @@ class WaitRequestReceived : ClientRequestEventWithPrefetch
         [IgnoreDataMember]
         public override string TracedInstanceId => this.InstanceId;
 
+        [DataMember]
+        public DateTime Timestamp { get; set; }
+
         [IgnoreDataMember]
         public WaitResponseReceived ResponseToSend { get; set; } // used to communicate response to ClientState
 

diff --git a/src/DurableTask.Netherite/OrchestrationService/Client.cs b/src/DurableTask.Netherite/OrchestrationService/Client.cs
@@ -400,6 +400,7 @@ public async Task<OrchestrationState> WaitForOrchestrationAsync(
                 RequestId = Interlocked.Increment(ref this.SequenceNumber),
                 InstanceId = instanceId,
                 ExecutionId = executionId,
+                Timestamp = DateTime.UtcNow,
                 TimeoutUtc = this.GetTimeoutBucket(timeout),
             };
 

diff --git a/src/DurableTask.Netherite/PartitionState/InstanceState.cs b/src/DurableTask.Netherite/PartitionState/InstanceState.cs
@@ -80,19 +80,36 @@ public override void Process(CreationRequestReceived creationRequestReceived, Ef
             };
         }
 
+        void CullWaiters(DateTime threshold)
+        {
+            // remove all waiters whose timeout is before the threshold
+            if (this.Waiters.Any(request => request.TimeoutUtc <= threshold))
+            {
+                this.Waiters = this.Waiters
+                            .Where(request => request.TimeoutUtc > threshold)
+                            .ToList();
+            }
+        }
 
         public override void Process(BatchProcessed evt, EffectTracker effects)
         {
             // update the state of an orchestration
             this.OrchestrationState = evt.State;
 
             // if the orchestration is complete, notify clients that are waiting for it
-            if (this.Waiters != null && WaitRequestReceived.SatisfiesWaitCondition(this.OrchestrationState))
+            if (this.Waiters != null)
             {
-                // we do not need effects.Add(TrackedObjectKey.Outbox) because it has already been added by SessionsState
-                evt.ResponsesToSend = this.Waiters.Select(request => request.CreateResponse(this.OrchestrationState)).ToList();
+                if (WaitRequestReceived.SatisfiesWaitCondition(this.OrchestrationState))
+                {
+                    // we do not need effects.Add(TrackedObjectKey.Outbox) because it has already been added by SessionsState
+                    evt.ResponsesToSend = this.Waiters.Select(request => request.CreateResponse(this.OrchestrationState)).ToList();
 
-                this.Waiters = null;
+                    this.Waiters = null;
+                }
+                else
+                {
+                    this.CullWaiters(evt.Timestamp);
+                }
             }
         }
 
@@ -111,10 +128,7 @@ public override void Process(WaitRequestReceived evt, EffectTracker effects)
                 }
                 else
                 {
-                    // cull the list of waiters to remove requests that have already timed out
-                    this.Waiters = this.Waiters
-                        .Where(request => request.TimeoutUtc > DateTime.UtcNow)
-                        .ToList();
+                    this.CullWaiters(evt.Timestamp);
                 }
 
                 this.Waiters.Add(evt);