Merge pull request #301 from microsoft/stevosyan/fix-terminating-suspended-orchestrations

Add logic to terminate suspended instances

Author: sophiatev

src/DurableTask.SqlServer/Scripts/logic.sql

@@ -487,7 +487,7 @@ BEGIN
 
     DECLARE @now datetime2(7) = SYSUTCDATETIME()
 
-    IF @existingStatus IN ('Running', 'Pending')
+    IF @existingStatus IN ('Running', 'Pending', 'Suspended')
     BEGIN
         -- Create a payload to store the reason, if any
         DECLARE @PayloadID uniqueidentifier = NULL

src/DurableTask.SqlServer/SqlUtils.cs

@@ -75,24 +75,12 @@ public static HistoryEvent GetHistoryEvent(this DbDataReader reader, bool isOrch
                         InstanceId = GetInstanceId(reader),
                     };
                     break;
-                case EventType.ExecutionCompleted:
+                case EventType.ExecutionCompleted:
+                    TryGetFailureDetails(reader, out FailureDetails? executionFailedDetails);
                     historyEvent = new ExecutionCompletedEvent(
                         eventId,
                         result: GetPayloadText(reader),
-                        orchestrationStatus: OrchestrationStatus.Completed);
-                    break;
-                case EventType.ExecutionFailed:
-                    string? executionFailedResult = null;
-                    if (!TryGetFailureDetails(reader, out FailureDetails? executionFailedDetails))
-                    {
-                        // Fall back to the old behavior
-                        executionFailedResult = GetPayloadText(reader);
-                    }
-
-                    historyEvent = new ExecutionCompletedEvent(
-                        eventId,
-                        result: executionFailedResult,
-                        orchestrationStatus: OrchestrationStatus.Failed,
+                        orchestrationStatus: GetRuntimeStatus(reader),
                         failureDetails: executionFailedDetails);
                     break;
                 case EventType.ExecutionStarted:

src/common.props

@@ -17,7 +17,7 @@
   <PropertyGroup>
     <MajorVersion>1</MajorVersion>
     <MinorVersion>5</MinorVersion>
-    <PatchVersion>2</PatchVersion>
+    <PatchVersion>3</PatchVersion>
     <VersionPrefix>$(MajorVersion).$(MinorVersion).$(PatchVersion)</VersionPrefix>
     <VersionSuffix></VersionSuffix>
     <AssemblyVersion>$(MajorVersion).$(MinorVersion).0.0</AssemblyVersion>

test/DurableTask.SqlServer.Tests/Integration/DatabaseManagement.cs

@@ -505,7 +505,7 @@ async Task ValidateDatabaseSchemaAsync(TestDatabase database, string schemaName
                 schemaName);
             Assert.Equal(1, currentSchemaVersion.Major);
             Assert.Equal(5, currentSchemaVersion.Minor);
-            Assert.Equal(2, currentSchemaVersion.Patch);
+            Assert.Equal(3, currentSchemaVersion.Patch);
         }
 
         sealed class TestDatabase : IDisposable

test/DurableTask.SqlServer.Tests/Integration/Orchestrations.cs

@@ -891,6 +891,57 @@ await instance.WaitForCompletion(
                 expectedOutput: "Bye!");
 
             LogAssert.NoWarningsOrErrors(this.testService.LogProvider);
+        }
+
+        [Fact]
+        public async Task TerminateSuspendedOrchestration()
+        {
+            string input = $"Hello {DateTime.UtcNow:o}";
+            string orchestrationName = "OrchestrationWithTimer";
+            TimeSpan delay = TimeSpan.FromSeconds(30);
+
+            // Performs a delay and then returns the input
+            TestInstance<string> instance = await this.testService.RunOrchestration(
+                input,
+                orchestrationName,
+                implementation: (ctx, input) => ctx.CreateTimer(ctx.CurrentUtcDateTime.Add(delay), input));
+
+            // Wait for the orchestration to finish starting
+            await instance.WaitForStart();
+
+            // Suspend the orchestration so that it won't process any new events
+            await instance.SuspendAsync();
+
+            // Wait for the orchestration to become suspended
+            OrchestrationState state = await instance.GetStateAsync();
+            TimeSpan waitForSuspendTimeout = TimeSpan.FromSeconds(5);
+            using CancellationTokenSource cts = new(waitForSuspendTimeout);
+            while (!cts.IsCancellationRequested && state.OrchestrationStatus != OrchestrationStatus.Suspended)
+            {
+                state = await instance.GetStateAsync();
+            }
+            Assert.Equal(OrchestrationStatus.Suspended, state.OrchestrationStatus);
+
+            // Now terminate the orchestration
+            await instance.TerminateAsync("Bye!");
+
+            TimeSpan waitForTerminationTimeout = TimeSpan.FromSeconds(5);
+            state = await instance.WaitForCompletion(
+                waitForTerminationTimeout,
+                expectedStatus: OrchestrationStatus.Terminated,
+                expectedOutput: "Bye!");
+
+            // Validate logs
+            LogAssert.NoWarningsOrErrors(this.testService.LogProvider);
+            LogAssert.Sequence(
+                this.testService.LogProvider,
+                LogAssert.AcquiredAppLock(),
+                LogAssert.CheckpointStarting(orchestrationName),
+                LogAssert.CheckpointCompleted(orchestrationName),
+                LogAssert.CheckpointStarting(orchestrationName),
+                LogAssert.CheckpointCompleted(orchestrationName),
+                LogAssert.CheckpointStarting(orchestrationName),
+                LogAssert.CheckpointCompleted(orchestrationName));
         }
     }
 }