fix(workflow): Fix non-determinism on replay when using a patched statement in a condition (#859)

Author: bergundy

packages/test/src/load/worker.ts

@@ -155,7 +155,6 @@ async function main() {
         activityInbound: [() => new ConnectionInjectorInterceptor(clientConnection)],
       },
     });
-    console.log('Created worker with options', worker.options);
 
     await withOptionalStatusServer(worker, statusPort, async () => {
       await worker.run();

packages/test/src/test-patch-and-condition.ts

@@ -0,0 +1,58 @@
+import crypto from 'node:crypto';
+import test from 'ava';
+import { Worker } from '@temporalio/worker';
+import { WorkflowClient } from '@temporalio/client';
+import { RUN_INTEGRATION_TESTS } from './helpers';
+import * as workflows from './workflows/patch-and-condition-pre-patch';
+
+if (RUN_INTEGRATION_TESTS) {
+  test('Patch in condition does not cause non-determinism error on replay', async (t) => {
+    const client = new WorkflowClient();
+    const workflowId = crypto.randomUUID();
+
+    // Create the first worker with pre-patched version of the workflow
+    const worker1 = await Worker.create({
+      taskQueue: 'patch-in-condition',
+      workflowsPath: require.resolve('./workflows/patch-and-condition-pre-patch'),
+    });
+
+    // Start the workflow and wait for the first task to be processed
+    const handle = await worker1.runUntil(async () => {
+      const handle = await client.start(workflows.patchInCondition, {
+        taskQueue: 'patch-in-condition',
+        workflowId,
+        workflowTaskTimeout: '1m', // Give our local activities enough time to run in CI
+      });
+      await handle.query('__stack_trace');
+      return handle;
+    });
+
+    // Create the second worker with post-patched version of the workflow
+    const worker2 = await Worker.create({
+      taskQueue: 'patch-in-condition',
+      workflowsPath: require.resolve('./workflows/patch-and-condition-post-patch'),
+    });
+
+    // Trigger a signal and wait for it to be processed
+    await worker2.runUntil(async () => {
+      await handle.signal(workflows.generateCommandSignal);
+      await handle.query('__stack_trace');
+    });
+
+    // Create the third worker that is identical to the second one
+    const worker3 = await Worker.create({
+      taskQueue: 'patch-in-condition',
+      workflowsPath: require.resolve('./workflows/patch-and-condition-post-patch'),
+    });
+
+    // Trigger a workflow task that will cause replay.
+    await worker3.runUntil(async () => {
+      await handle.signal(workflows.generateCommandSignal);
+      await handle.result();
+    });
+
+    // If the workflow completes, commands are generated in the right order and it is safe to use a patched statement
+    // inside a condition.
+    t.pass();
+  });
+}

packages/test/src/workflows/patch-and-condition-post-patch.ts

@@ -0,0 +1,27 @@
+/**
+ * Post-patched version of a workflow used to reproduce an issue with patch inside a condition.
+ *
+ * Uses a patched statement inside a condition to replicate a bug where the runtime could generate out of order
+ * commands on replay.
+ *
+ * @module
+ */
+import * as wf from '@temporalio/workflow';
+import { generateCommandSignal } from './patch-and-condition-pre-patch';
+
+/**
+ * Patches the workflow from ./patch-and-condition-pre-patch, adds a patched statement inside a condition.
+ *
+ */
+export async function patchInCondition(): Promise<void> {
+  // The signal handler here is important for the repro.
+  // We use it so the workflow generates a command that will conflict with the patch.
+  wf.setHandler(generateCommandSignal, async () => {
+    // Ignore completion, it's irrelevant, just generate a command
+    await wf.sleep('1s');
+  });
+
+  // Note that the condition always returns false, we don't want it to resolve the promise, just to using it to test the
+  // edge case.
+  await Promise.race([wf.sleep('5s'), wf.condition(() => wf.patched('irrelevant') && false)]);
+}

packages/test/src/workflows/patch-and-condition-pre-patch.ts

@@ -0,0 +1,22 @@
+/**
+ * Pre-patched version of a workflow used to reproduce an issue with patch inside a condition.
+ *
+ * @module
+ */
+import * as wf from '@temporalio/workflow';
+
+export const generateCommandSignal = wf.defineSignal('generate-command');
+
+/**
+ * Unpatched version of the workflow - just sleep and set up our signal handler
+ */
+export async function patchInCondition(): Promise<void> {
+  // The signal handler here is important for the repro.
+  // We use it so the workflow generates a command that will conflict with the patch.
+  wf.setHandler(generateCommandSignal, async () => {
+    // Ignore completion, it's irrelevant, just generate a command
+    await wf.sleep('1s');
+  });
+
+  await wf.sleep('5s');
+}

packages/worker/src/workflow/bundler.ts

@@ -1,4 +1,3 @@
-import crypto from 'crypto';
 import * as realFS from 'fs';
 import * as memfs from 'memfs';
 import { builtinModules } from 'module';
@@ -121,11 +120,11 @@ export class WorkflowCodeBundler {
     if (stat.isFile()) {
       // workflowsPath is a file; make the entrypoint a sibling of that file
       const { root, dir, name } = path.parse(workflowsPath);
-      return path.format({ root, dir, base: `${name}-entrypoint-${crypto.randomBytes(8).toString('hex')}.js` });
+      return path.format({ root, dir, base: `${name}-autogenerated-entrypoint.js` });
     } else {
       // workflowsPath is a directory; make the entrypoint a sibling of that directory
       const { root, dir, base } = path.parse(workflowsPath);
-      return path.format({ root, dir, base: `${base}-entrypoint-${crypto.randomBytes(8).toString('hex')}.js` });
+      return path.format({ root, dir, base: `${base}-autogenerated-entrypoint.js` });
     }
   }
 
@@ -232,7 +231,7 @@ export { api };
       devtool: 'inline-source-map',
       output: {
         path: distDir,
-        filename: 'workflow-isolate-[fullhash].js',
+        filename: 'workflow-bundle-[fullhash].js',
         devtoolModuleFilenameTemplate: '[absolute-resource-path]',
         library: '__TEMPORAL__',
       },

packages/worker/src/workflow/vm.ts

@@ -2,7 +2,7 @@ import { cutoffStackTrace, IllegalStateError } from '@temporalio/common';
 import { coresdk } from '@temporalio/proto';
 import { WorkflowInfo, FileLocation } from '@temporalio/workflow';
 import { SinkCall } from '@temporalio/workflow/lib/sinks';
-import type * as internals from '@temporalio/workflow/lib/worker-interface';
+import * as internals from '@temporalio/workflow/lib/worker-interface';
 import assert from 'assert';
 import { AsyncLocalStorage } from 'async_hooks';
 import semver from 'semver';
@@ -397,8 +397,7 @@ export class VMWorkflow implements Workflow {
         coresdk.workflow_activation.WorkflowActivation.fromObject({ ...activation, jobs }),
         batchIndex++
       );
-      // Only trigger conditions for non-query jobs
-      if (!jobs[0].queryWorkflow) {
+      if (internals.showUnblockConditions(jobs[0])) {
         await this.tryUnblockConditions();
       }
     }

packages/workflow/src/worker-interface.ts

@@ -222,8 +222,7 @@ export function activate(activation: coresdk.workflow_activation.WorkflowActivat
         return;
       }
       state.activator[job.variant](variant as any /* TS can't infer this type */);
-      // Only trigger conditions for non-query jobs
-      if (!job.queryWorkflow) {
+      if (showUnblockConditions(job)) {
         tryUnblockConditions();
       }
     }
@@ -279,6 +278,13 @@ export function tryUnblockConditions(): number {
   return numUnblocked;
 }
 
+/**
+ * Predicate used to prevent triggering conditions for non-query and non-patch jobs.
+ */
+export function showUnblockConditions(job: coresdk.workflow_activation.IWorkflowActivationJob): boolean {
+  return !job.queryWorkflow && !job.notifyHasPatch;
+}
+
 export async function dispose(): Promise<void> {
   const dispose = composeInterceptors(state.interceptors.internals, 'dispose', async () => {
     storage.disable();