Support for workflow ID conflict policy (temporalio#304)

Fixes temporalio#217
cretz · Jul 12, 2024 · fa3867e · fa3867e
1 parent 1fceb0b
commit fa3867e
Show file tree

Hide file tree

Showing 4 changed files with 74 additions and 0 deletions.
diff --git a/src/Temporalio/Client/Schedules/ScheduleActionStartWorkflow.cs b/src/Temporalio/Client/Schedules/ScheduleActionStartWorkflow.cs
@@ -122,6 +122,10 @@ internal static ScheduleActionStartWorkflow FromProto(
             {
                 throw new ArgumentException("ID reuse policy cannot change from default for scheduled workflow");
             }
+            if (Options.IdConflictPolicy != Api.Enums.V1.WorkflowIdConflictPolicy.Unspecified)
+            {
+                throw new ArgumentException("ID conflict policy cannot change from default for scheduled workflow");
+            }
             if (Options.CronSchedule != null)
             {
                 throw new ArgumentException("Cron schedule cannot be set on scheduled workflow");

diff --git a/src/Temporalio/Client/TemporalClient.Workflow.cs b/src/Temporalio/Client/TemporalClient.Workflow.cs
@@ -504,6 +504,7 @@ private async Task<WorkflowHandle<TWorkflow, TResult>> StartWorkflowInternalAsyn
                     Identity = Client.Connection.Options.Identity,
                     RequestId = Guid.NewGuid().ToString(),
                     WorkflowIdReusePolicy = input.Options.IdReusePolicy,
+                    WorkflowIdConflictPolicy = input.Options.IdConflictPolicy,
                     RetryPolicy = input.Options.RetryPolicy?.ToProto(),
                     RequestEagerExecution = input.Options.RequestEagerStart,
                 };

diff --git a/src/Temporalio/Client/WorkflowOptions.cs b/src/Temporalio/Client/WorkflowOptions.cs
@@ -62,6 +62,15 @@ public WorkflowOptions(string id, string taskQueue)
         /// </summary>
         public WorkflowIdReusePolicy IdReusePolicy { get; set; } = WorkflowIdReusePolicy.AllowDuplicate;
 
+        /// <summary>
+        /// Gets or sets how already-existing workflows of the same ID are treated. Default is
+        /// <see cref="WorkflowIdConflictPolicy.Unspecified" /> which effectively means
+        /// <see cref="WorkflowIdConflictPolicy.Fail"/> on the server. If this value is set, then
+        /// <see cref="IdReusePolicy"/> cannot be set to
+        /// <see cref="WorkflowIdReusePolicy.TerminateIfRunning"/>.
+        /// </summary>
+        public WorkflowIdConflictPolicy IdConflictPolicy { get; set; } = WorkflowIdConflictPolicy.Unspecified;
+
         /// <summary>
         /// Gets or sets the retry policy for the workflow. If unset, workflow never retries.
         /// </summary>

diff --git a/tests/Temporalio.Tests/Worker/WorkflowWorkerTests.cs b/tests/Temporalio.Tests/Worker/WorkflowWorkerTests.cs
@@ -5580,6 +5580,66 @@ await AssertWarnings(
             shouldWarn: false);
     }
 
+    [Workflow]
+    public class IdConflictWorkflow
+    {
+        // Just wait forever
+        [WorkflowRun]
+        public Task RunAsync() => Workflow.WaitConditionAsync(() => false);
+    }
+
+    [Fact]
+    public async Task ExecuteWorkflowAsync_IdConflictPolicy_ProperlyApplies()
+    {
+        await ExecuteWorkerAsync<IdConflictWorkflow>(async worker =>
+        {
+            // Start a workflow
+            var handle = await Env.Client.StartWorkflowAsync(
+                (IdConflictWorkflow wf) => wf.RunAsync(),
+                new(id: $"wf-{Guid.NewGuid()}", taskQueue: worker.Options.TaskQueue!));
+            handle = handle with { RunId = handle.ResultRunId };
+
+            // Confirm another fails by default
+            await Assert.ThrowsAsync<WorkflowAlreadyStartedException>(() =>
+                Env.Client.StartWorkflowAsync(
+                    (IdConflictWorkflow wf) => wf.RunAsync(),
+                    new(id: handle.Id, taskQueue: worker.Options.TaskQueue!)));
+
+            // Confirm fails if explicitly given that option
+            await Assert.ThrowsAsync<WorkflowAlreadyStartedException>(() =>
+                Env.Client.StartWorkflowAsync(
+                    (IdConflictWorkflow wf) => wf.RunAsync(),
+                    new(id: handle.Id, taskQueue: worker.Options.TaskQueue!)
+                    {
+                        IdConflictPolicy = WorkflowIdConflictPolicy.Fail,
+                    }));
+
+            // Confirm gives back same handle if requested
+            var newHandle = await Env.Client.StartWorkflowAsync(
+                (IdConflictWorkflow wf) => wf.RunAsync(),
+                new(id: handle.Id, taskQueue: worker.Options.TaskQueue!)
+                {
+                    IdConflictPolicy = WorkflowIdConflictPolicy.UseExisting,
+                });
+            newHandle = newHandle with { RunId = newHandle.ResultRunId };
+            Assert.Equal(handle.RunId, newHandle.RunId);
+            Assert.Equal(WorkflowExecutionStatus.Running, (await handle.DescribeAsync()).Status);
+            Assert.Equal(WorkflowExecutionStatus.Running, (await newHandle.DescribeAsync()).Status);
+
+            // Confirm terminates and starts new if requested
+            newHandle = await Env.Client.StartWorkflowAsync(
+                (IdConflictWorkflow wf) => wf.RunAsync(),
+                new(id: handle.Id, taskQueue: worker.Options.TaskQueue!)
+                {
+                    IdConflictPolicy = WorkflowIdConflictPolicy.TerminateExisting,
+                });
+            newHandle = newHandle with { RunId = newHandle.ResultRunId };
+            Assert.NotEqual(handle.RunId, newHandle.RunId);
+            Assert.Equal(WorkflowExecutionStatus.Terminated, (await handle.DescribeAsync()).Status);
+            Assert.Equal(WorkflowExecutionStatus.Running, (await newHandle.DescribeAsync()).Status);
+        });
+    }
+
     internal static Task AssertTaskFailureContainsEventuallyAsync(
         WorkflowHandle handle, string messageContains)
     {