feat(workflow): Support workflow ID conflict policy (#1490)

Co-authored-by: James Watkins-Harvey <[email protected]>

Author: antlai-temporal

packages/client/src/workflow-client.ts

@@ -939,6 +939,7 @@ export class WorkflowClient extends BaseClient {
       requestId: uuid4(),
       workflowId: options.workflowId,
       workflowIdReusePolicy: options.workflowIdReusePolicy,
+      workflowIdConflictPolicy: options.workflowIdConflictPolicy,
       workflowType: { name: workflowType },
       input: { payloads: await encodeToPayloads(this.dataConverter, ...options.args) },
       signalName,
@@ -989,6 +990,7 @@ export class WorkflowClient extends BaseClient {
       requestId: uuid4(),
       workflowId: opts.workflowId,
       workflowIdReusePolicy: opts.workflowIdReusePolicy,
+      workflowIdConflictPolicy: opts.workflowIdConflictPolicy,
       workflowType: { name: workflowType },
       input: { payloads: await encodeToPayloads(this.dataConverter, ...opts.args) },
       taskQueue: {

packages/client/src/workflow-options.ts

@@ -15,7 +15,8 @@ export interface WorkflowOptions extends CommonWorkflowOptions {
    *
    * Assign a meaningful business id.
    * This ID can be used to ensure starting Workflows is idempotent.
-   * Workflow IDs are unique, see also {@link WorkflowOptions.workflowIdReusePolicy}
+   * Workflow IDs are unique: see {@link WorkflowOptions.workflowIdReusePolicy}
+   * and {@link WorkflowOptions.workflowIdConflictPolicy}.
    */
   workflowId: string;
 

packages/common/src/failure.ts

@@ -279,7 +279,7 @@ export class ChildWorkflowFailure extends TemporalFailure {
 
 /**
  * This exception is thrown in the following cases:
- *  - Workflow with the same Workflow Id is currently running
+ *  - Workflow with the same Workflow ID is currently running and the {@link WorkflowOptions.workflowIdConflictPolicy} is `WORKFLOW_ID_CONFLICT_POLICY_FAIL`
  *  - There is a closed Workflow with the same Workflow Id and the {@link WorkflowOptions.workflowIdReusePolicy}
  *    is `WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE`
  *  - There is closed Workflow in the `Completed` state with the same Workflow Id and the {@link WorkflowOptions.workflowIdReusePolicy}

packages/common/src/workflow-options.ts

@@ -7,11 +7,15 @@ import { checkExtends } from './type-helpers';
 // Avoid importing the proto implementation to reduce workflow bundle size
 // Copied from temporal.api.enums.v1.WorkflowIdReusePolicy
 /**
+ * Defines what happens when trying to start a Workflow with the same ID as a *Closed* Workflow.
+ *
+ * See {@link WorkflowOptions.workflowIdConflictPolicy} for what happens when trying to start a
+ * Workflow with the same ID as a *Running* Workflow.
+ *
  * Concept: {@link https://docs.temporal.io/concepts/what-is-a-workflow-id-reuse-policy/ | Workflow Id Reuse Policy}
  *
- * Whether a Workflow can be started with a Workflow Id of a Closed Workflow.
+ * *Note: It is not possible to have two actively running Workflows with the same ID.*
  *
- * *Note: A Workflow can never be started with a Workflow Id of a Running Workflow.*
  */
 export enum WorkflowIdReusePolicy {
   /**
@@ -38,24 +42,78 @@ export enum WorkflowIdReusePolicy {
   WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE = 3,
 
   /**
-   * Terminate the current workflow if one is already running.
+   * Terminate the current Workflow if one is already running; otherwise allow reusing the Workflow ID.
+   *
+   * @deprecated Use {@link WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE} instead, and
+   *             set `WorkflowOptions.workflowIdConflictPolicy` to
+   *             {@link WorkflowIdConflictPolicy.WORKFLOW_ID_CONFLICT_POLICY_TERMINATE_EXISTING}.
+   *             When using this option, `WorkflowOptions.workflowIdConflictPolicy` must be left unspecified.
    */
   WORKFLOW_ID_REUSE_POLICY_TERMINATE_IF_RUNNING = 4,
 }
 
 checkExtends<temporal.api.enums.v1.WorkflowIdReusePolicy, WorkflowIdReusePolicy>();
 checkExtends<WorkflowIdReusePolicy, temporal.api.enums.v1.WorkflowIdReusePolicy>();
 
+// Avoid importing the proto implementation to reduce workflow bundle size
+// Copied from temporal.api.enums.v1.WorkflowIdConflictPolicy
+/**
+ * Defines what happens when trying to start a Workflow with the same ID as a *Running* Workflow.
+ *
+ * See {@link WorkflowOptions.workflowIdReusePolicy} for what happens when trying to start a Workflow
+ *  with the same ID as a *Closed* Workflow.
+ *
+ * *Note: It is not possible to have two actively running Workflows with the same ID.*
+ */
+export enum WorkflowIdConflictPolicy {
+  /**
+   *  This is the default so that we can set the policy
+   * `WORKFLOW_ID_REUSE_POLICY_TERMINATE_IF_RUNNING` in `WorkflowIdReusePolicy`, which is incompatible
+   * with setting any other `WorkflowIdConflictPolicy` values.
+   *
+   * The actual default behavior is `WORKFLOW_ID_CONFLICT_POLICY_FAIL` for a start Workflow request,
+   *  and `WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING` for a signal with start Workflow request.
+   */
+  WORKFLOW_ID_CONFLICT_POLICY_UNSPECIFIED = 0,
+
+  /**
+   * Do not start a new Workflow. Instead raise a `WorkflowExecutionAlreadyStartedError`.
+   */
+  WORKFLOW_ID_CONFLICT_POLICY_FAIL = 1,
+
+  /**
+   * Do not start a new Workflow. Instead return a Workflow Handle for the currently Running Workflow.
+   */
+  WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING = 2,
+
+  /**
+   * Start a new Workflow, terminating the current workflow if one is already running.
+   */
+  WORKFLOW_ID_CONFLICT_POLICY_TERMINATE_EXISTING = 3,
+}
+
+checkExtends<temporal.api.enums.v1.WorkflowIdConflictPolicy, WorkflowIdConflictPolicy>();
+checkExtends<WorkflowIdConflictPolicy, temporal.api.enums.v1.WorkflowIdConflictPolicy>();
+
 export interface BaseWorkflowOptions {
   /**
-   * Whether a Workflow can be started with a Workflow Id of a Closed Workflow.
+   * Defines what happens when trying to start a Workflow with the same ID as a *Closed* Workflow.
    *
-   * *Note: A Workflow can never be started with a Workflow Id of a Running Workflow.*
+   * *Note: It is not possible to have two actively running Workflows with the same ID.*
    *
    * @default {@link WorkflowIdReusePolicy.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE}
    */
   workflowIdReusePolicy?: WorkflowIdReusePolicy;
 
+  /**
+   * Defines what happens when trying to start a Workflow with the same ID as a *Running* Workflow.
+   *
+   * *Note: It is not possible to have two actively running Workflows with the same ID.*
+   *
+   * @default {@link WorkflowIdConflictPolicy.WORKFLOW_ID_CONFLICT_POLICY_UNSPECIFIED}
+   */
+  workflowIdConflictPolicy?: WorkflowIdConflictPolicy;
+
   /**
    * Controls how a Workflow Execution is retried.
    *

packages/test/src/test-integration-workflows.ts

@@ -7,9 +7,9 @@ import { msToNumber, tsToMs } from '@temporalio/common/lib/time';
 import { TestWorkflowEnvironment } from '@temporalio/testing';
 import { CancelReason } from '@temporalio/worker/lib/activity';
 import * as workflow from '@temporalio/workflow';
-import { defineQuery, defineSignal } from '@temporalio/workflow';
+import { defineQuery, defineSignal, WorkflowIdConflictPolicy } from '@temporalio/workflow';
 import { SdkFlags } from '@temporalio/workflow/lib/flags';
-import { ActivityCancellationType, ApplicationFailure } from '@temporalio/common';
+import { ActivityCancellationType, ApplicationFailure, WorkflowExecutionAlreadyStartedError } from '@temporalio/common';
 import { signalSchedulingWorkflow } from './activities/helpers';
 import { activityStartedSignal } from './workflows/definitions';
 import * as workflows from './workflows';
@@ -216,6 +216,122 @@ test('Start of workflow is delayed', async (t) => {
   t.is(tsToMs(startDelay), 5678000);
 });
 
+export async function conflictId(): Promise<void> {
+  await workflow.condition(() => false);
+}
+
+test('Start of workflow respects workflow id conflict policy', async (t) => {
+  const { createWorker, taskQueue } = helpers(t);
+  const wfid = `${taskQueue}-` + randomUUID();
+  const client = t.context.env.client;
+
+  const worker = await createWorker();
+  await worker.runUntil(async () => {
+    const handle = await client.workflow.start(conflictId, {
+      taskQueue,
+      workflowId: wfid,
+    });
+    const handleWithRunId = client.workflow.getHandle(handle.workflowId, handle.firstExecutionRunId);
+
+    // Confirm another fails by default
+    const err = await t.throwsAsync(
+      client.workflow.start(conflictId, {
+        taskQueue,
+        workflowId: wfid,
+      }),
+      {
+        instanceOf: WorkflowExecutionAlreadyStartedError,
+      }
+    );
+
+    t.true(err instanceof WorkflowExecutionAlreadyStartedError);
+
+    // Confirm fails with explicit option
+    const err1 = await t.throwsAsync(
+      client.workflow.start(conflictId, {
+        taskQueue,
+        workflowId: wfid,
+        workflowIdConflictPolicy: WorkflowIdConflictPolicy.WORKFLOW_ID_CONFLICT_POLICY_FAIL,
+      }),
+      {
+        instanceOf: WorkflowExecutionAlreadyStartedError,
+      }
+    );
+
+    t.true(err1 instanceof WorkflowExecutionAlreadyStartedError);
+
+    // Confirm gives back same handle
+    const handle2 = await client.workflow.start(conflictId, {
+      taskQueue,
+      workflowId: wfid,
+      workflowIdConflictPolicy: WorkflowIdConflictPolicy.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING,
+    });
+
+    const desc = await handleWithRunId.describe();
+    const desc2 = await handle2.describe();
+
+    t.is(desc.runId, desc2.runId);
+    t.is(desc.status.name, 'RUNNING');
+    t.is(desc2.status.name, 'RUNNING');
+
+    // Confirm terminates and starts new
+    const handle3 = await client.workflow.start(conflictId, {
+      taskQueue,
+      workflowId: wfid,
+      workflowIdConflictPolicy: WorkflowIdConflictPolicy.WORKFLOW_ID_CONFLICT_POLICY_TERMINATE_EXISTING,
+    });
+
+    const descWithRunId = await handleWithRunId.describe();
+    const desc3 = await handle3.describe();
+    t.not(descWithRunId.runId, desc3.runId);
+    t.is(descWithRunId.status.name, 'TERMINATED');
+    t.is(desc3.status.name, 'RUNNING');
+  });
+});
+
+test('Start of workflow with signal respects conflict id policy', async (t) => {
+  const { createWorker, taskQueue } = helpers(t);
+  const wfid = `${taskQueue}-` + randomUUID();
+  const client = t.context.env.client;
+  const worker = await createWorker();
+  await worker.runUntil(async () => {
+    const handle = await client.workflow.start(workflows.signalTarget, {
+      taskQueue,
+      workflowId: wfid,
+    });
+    const handleWithRunId = client.workflow.getHandle(handle.workflowId, handle.firstExecutionRunId);
+
+    // Confirm gives back same handle is the default policy
+    const handle2 = await t.context.env.client.workflow.signalWithStart(workflows.signalTarget, {
+      taskQueue,
+      workflowId: wfid,
+      signal: workflows.argsTestSignal,
+      signalArgs: [123, 'kid'],
+    });
+    const desc = await handleWithRunId.describe();
+    const desc2 = await handle2.describe();
+
+    t.deepEqual(desc.runId, desc2.runId);
+    t.deepEqual(desc.status.name, 'RUNNING');
+    t.deepEqual(desc2.status.name, 'RUNNING');
+
+    // Confirm terminates and starts new
+    const handle3 = await t.context.env.client.workflow.signalWithStart(workflows.signalTarget, {
+      taskQueue,
+      workflowId: wfid,
+      signal: workflows.argsTestSignal,
+      signalArgs: [123, 'kid'],
+      workflowIdConflictPolicy: WorkflowIdConflictPolicy.WORKFLOW_ID_CONFLICT_POLICY_TERMINATE_EXISTING,
+    });
+
+    const descWithRunId = await handleWithRunId.describe();
+    const desc3 = await handle3.describe();
+    t.true(descWithRunId.runId !== desc3.runId);
+    t.deepEqual(descWithRunId.status.name, 'TERMINATED');
+    t.deepEqual(desc3.status.name, 'RUNNING');
+  });
+});
+
 test('Start of workflow with signal is delayed', async (t) => {
   const { taskQueue } = helpers(t);
   // This workflow never runs