Fix InProcessTestHost ContinueAsNew stuck-instance race condition

src/InProcessTestHost/InProcessTestHost.csproj

@@ -5,7 +5,7 @@
     <RootNamespace>Microsoft.DurableTask.Testing</RootNamespace>
     <AssemblyName>Microsoft.DurableTask.InProcessTestHost</AssemblyName>
     <PackageId>Microsoft.DurableTask.InProcessTestHost</PackageId>
-    <Version>0.2.2-preview.1</Version>
+    <Version>0.2.3-preview.1</Version>
 
     <!-- Suppress CA1848: Use LoggerMessage delegates for high-performance logging scenarios -->
     <NoWarn>$(NoWarn);CA1848</NoWarn>

src/InProcessTestHost/Sidecar/Dispatcher/TaskOrchestrationDispatcher.cs

@@ -111,12 +111,12 @@ protected override async Task ExecuteWorkItemAsync(TaskOrchestrationWorkItem wor
         {
             throw new ArgumentException($"Could not find an orchestration instance ID in the work item's runtime state.", nameof(workItem));
         }
-
-        var activityMessages = new List<TaskMessage>();
-        var timerMessages = new List<TaskMessage>();
-        var orchestratorMessages = new List<TaskMessage>();
-
-        // We loop for as long as the orchestrator does a ContinueAsNew
+
+        var activityMessages = new List<TaskMessage>();
+        var timerMessages = new List<TaskMessage>();
+        var orchestratorMessages = new List<TaskMessage>();
+
+        // We loop for as long as the orchestrator does a ContinueAsNew
         while (true)
         {
             if (this.log.IsEnabled(LogLevel.Debug))
@@ -149,6 +149,16 @@ protected override async Task ExecuteWorkItemAsync(TaskOrchestrationWorkItem wor
                 out bool continueAsNew);
             if (continueAsNew)
             {
+                // The previous execution is being replaced by a new one. Clear any
+                // accumulated messages from the old execution so they are not
+                // re-enqueued when the work item completes. Without this, stale
+                // activity/timer/orchestrator messages from prior iterations would
+                // be committed alongside the new execution, causing duplicate
+                // activities, stale timer fires, and ultimately stuck instances.
+                activityMessages.Clear();
+                timerMessages.Clear();
+                orchestratorMessages.Clear();
+
                 // Continue running the orchestration with a new history.
                 // Renew the lock if we're getting close to its expiration.
                 if (workItem.LockedUntilUtc != default && DateTime.UtcNow.AddMinutes(1) > workItem.LockedUntilUtc)
@@ -365,11 +375,11 @@ void ApplyOrchestratorActions(
                     Input = sendEventAction.EventData,
                 };
 
-                runtimeState.AddEvent(sendEvent);
-
-                EventRaisedEvent eventRaisedEvent = new(-1, sendEventAction.EventData)
-                {
-                    Name = sendEventAction.EventName
+                runtimeState.AddEvent(sendEvent);
+
+                EventRaisedEvent eventRaisedEvent = new(-1, sendEventAction.EventData)
+                {
+                    Name = sendEventAction.EventName
                 };
 
                 orchestratorMessages.Add(new TaskMessage

src/InProcessTestHost/Sidecar/Dispatcher/WorkItemDispatcher.cs

@@ -156,10 +156,10 @@ async Task WaitForAllClear(CancellationToken cancellationToken)
     {
         TimeSpan logInterval = TimeSpan.FromMinutes(1);
 
-        // IMPORTANT: This logic assumes only a single logical "thread" is executing the receive loop,
-        //            and that there's no possible race condition when comparing work-item counts.
+        // IMPORTANT: The receive loop is expected to have a single logical execution path, but the
+        //            work-item count can still change concurrently and must be read using thread-safe access.
         DateTime nextLogTime = DateTime.MinValue;
-        while (this.currentWorkItems >= this.MaxWorkItems)
+        while (Volatile.Read(ref this.currentWorkItems) >= this.MaxWorkItems)
         {
             // Periodically log that we're waiting for available concurrency.
             // No need to use UTC for this. Local time is a bit easier to debug.
@@ -169,7 +169,7 @@ async Task WaitForAllClear(CancellationToken cancellationToken)
                 this.log.FetchingThrottled(
                     dispatcher: this.name,
                     details: "The current active work-item count has reached the allowed maximum.",
-                    this.currentWorkItems,
+                    Volatile.Read(ref this.currentWorkItems),
                     this.MaxWorkItems);
                 nextLogTime = now.Add(logInterval);
             }
@@ -192,14 +192,14 @@ async Task WaitForAllClear(CancellationToken cancellationToken)
     async Task WaitForOutstandingWorkItems(CancellationToken cancellationToken)
     {
         DateTime nextLogTime = DateTime.MinValue;
-        while (this.currentWorkItems > 0)
+        while (Volatile.Read(ref this.currentWorkItems) > 0)
         {
             // Periodically log that we're waiting for outstanding work items to complete.
             // No need to use UTC for this. Local time is a bit easier to debug.
             DateTime now = DateTime.Now;
             if (now >= nextLogTime)
             {
-                this.log.DispatcherStopping(this.name, this.currentWorkItems);
+                this.log.DispatcherStopping(this.name, Volatile.Read(ref this.currentWorkItems));
                 nextLogTime = now.AddMinutes(1);
             }
 
@@ -265,7 +265,7 @@ async Task FetchAndExecuteLoop(CancellationToken cancellationToken)
 
                     if (workItem != null)
                     {
-                        this.currentWorkItems++;
+                        Interlocked.Increment(ref this.currentWorkItems);
                         this.log.FetchWorkItemCompleted(
                             this.name,
                             this.GetWorkItemId(workItem),
@@ -334,7 +334,7 @@ async Task ExecuteWorkItem(T workItem)
                     details: ex.ToString());
             }
 
-            this.currentWorkItems--;
+            Interlocked.Decrement(ref this.currentWorkItems);
         }
     }
 }

src/InProcessTestHost/Sidecar/InMemoryOrchestrationService.cs

@@ -653,7 +653,7 @@ public void AddMessage(TaskMessage message)
             SerializedInstanceState state = this.store.GetOrAdd(instanceId, id => new SerializedInstanceState(id, executionId));
             lock (state)
             {
-                bool isRestart = state.ExecutionId != null && state.ExecutionId != executionId;
+                bool isRestart = executionId != null && state.ExecutionId != null && state.ExecutionId != executionId;
 
                 if (message.Event is ExecutionStartedEvent startEvent)
                 {
@@ -684,11 +684,14 @@ public void AddMessage(TaskMessage message)
                 else if (state.IsCompleted)
                 {
                     // Drop the message since we're completed
-                    // GOOD: The user-provided the instanceId
-                    // logger.LogWarning(
-                    //     "Dropped {eventType} message for instance '{instanceId}' because the orchestration has already completed.",
-                    //     message.Event.EventType,
-                    //     instanceId);
+                    return;
+                }
+                else if (isRestart)
+                {
+                    // Drop messages that belong to a previous execution (stale activity
+                    // completions, timer fires, etc.). These can arrive when a
+                    // ContinueAsNew created a new execution but work items from the
+                    // old execution are still in flight.
                     return;
                 }