fix(workflow): Correctly handle condition with timeout of 0 (#985)

Author: mjameswh

.github/workflows/ci.yml

@@ -292,11 +292,12 @@ jobs:
 
   # Runs the sdk features repo tests with this repo's current SDK code
   sdk-features-tests:
-    uses: temporalio/sdk-features/.github/workflows/typescript.yaml@main
+    uses: temporalio/sdk-features/.github/workflows/typescript.yaml@ts-next
     with:
       typescript-repo-path: ${{github.event.pull_request.head.repo.full_name}}
       version: ${{github.event.pull_request.head.ref}}
       version-is-repo-ref: true
+      sdk-features-repo-ref: ts-next
 
   stress-tests:
     uses: ./.github/workflows/stress.yml

packages/test/src/integration-tests.ts

@@ -52,7 +52,7 @@ import { cleanOptionalStackTrace, u8 } from './helpers';
 import * as workflows from './workflows';
 import { withZeroesHTTPServer } from './zeroes-http-server';
 
-const { EVENT_TYPE_TIMER_STARTED, EVENT_TYPE_TIMER_FIRED, EVENT_TYPE_TIMER_CANCELED } =
+const { EVENT_TYPE_TIMER_STARTED, EVENT_TYPE_TIMER_FIRED, EVENT_TYPE_TIMER_CANCELED, EVENT_TYPE_MARKER_RECORDED } =
   iface.temporal.api.enums.v1.EventType;
 
 const timerEventTypes = new Set([EVENT_TYPE_TIMER_STARTED, EVENT_TYPE_TIMER_FIRED, EVENT_TYPE_TIMER_CANCELED]);
@@ -1421,4 +1421,27 @@ export function runIntegrationTests(codec?: PayloadCodec): void {
     );
     t.pass();
   });
+
+  test('Condition with timeout 0 does not block indefinitely', async (t) => {
+    const { client } = t.context;
+    const workflowId = uuid4();
+    const handle = await client.start(workflows.conditionTimeout0Simple, {
+      taskQueue: 'test',
+      workflowId,
+    });
+
+    t.false(await handle.result());
+
+    const history = await handle.fetchHistory();
+    const timerStartedEvents = history.events!.filter(({ eventType }) => eventType! === EVENT_TYPE_TIMER_STARTED);
+    t.is(timerStartedEvents.length, 2);
+    t.is(timerStartedEvents[0].timerStartedEventAttributes!.timerId, '1');
+    t.is(tsToMs(timerStartedEvents[0].timerStartedEventAttributes!.startToFireTimeout), 1000);
+    t.is(timerStartedEvents[1].timerStartedEventAttributes!.timerId, '2');
+    t.is(tsToMs(timerStartedEvents[1].timerStartedEventAttributes!.startToFireTimeout), 1);
+
+    const markersEvents = history.events!.filter(({ eventType }) => eventType! === EVENT_TYPE_MARKER_RECORDED);
+    t.is(markersEvents.length, 1);
+    t.is(markersEvents[0].markerRecordedEventAttributes!.markerName, CHANGE_MARKER_NAME);
+  });
 }

packages/test/src/test-workflows.ts

@@ -1872,3 +1872,83 @@ test('query not found - successString', async (t) => {
     );
   }
 });
+
+test('condition with timeout 0 maintain pre 1.5.0 compatibility - conditionTimeout0', async (t) => {
+  const { workflowType } = t.context;
+  {
+    const act: coresdk.workflow_activation.IWorkflowActivation = {
+      runId: 'test-runId',
+      timestamp: msToTs(Date.now()),
+      isReplaying: true,
+      jobs: [makeStartWorkflowJob(workflowType)],
+    };
+    const completion = await activate(t, act);
+    compareCompletion(t, completion, makeSuccess([]));
+  }
+  {
+    const completion = await activate(t, await makeSignalWorkflow('a', []));
+    compareCompletion(
+      t,
+      completion,
+      makeSuccess([
+        makeSetPatchMarker('__sdk_internal_patch_number:1', false),
+        makeStartTimerCommand({
+          seq: 1,
+          startToFireTimeout: msToTs(1),
+        }),
+      ])
+    );
+  }
+  {
+    const completion = await activate(t, await makeFireTimer(1));
+    compareCompletion(
+      t,
+      completion,
+      makeSuccess([makeCompleteWorkflowExecution(defaultPayloadConverter.toPayload(1))])
+    );
+  }
+});
+
+test('condition with timeout 0 in >1.5.0 - conditionTimeout0', async (t) => {
+  const { workflowType } = t.context;
+  {
+    const act: coresdk.workflow_activation.IWorkflowActivation = {
+      runId: 'test-runId',
+      timestamp: msToTs(Date.now()),
+      jobs: [makeStartWorkflowJob(workflowType)],
+    };
+    const completion = await activate(t, act);
+    compareCompletion(
+      t,
+      completion,
+      makeSuccess([
+        makeSetPatchMarker('__sdk_internal_patch_number:1', false),
+        makeStartTimerCommand({
+          seq: 1,
+          startToFireTimeout: msToTs(1),
+        }),
+      ])
+    );
+  }
+  {
+    const completion = await activate(t, await makeFireTimer(1));
+    compareCompletion(
+      t,
+      completion,
+      makeSuccess([
+        makeStartTimerCommand({
+          seq: 2,
+          startToFireTimeout: msToTs(1),
+        }),
+      ])
+    );
+  }
+  {
+    const completion = await activate(t, await makeFireTimer(2));
+    compareCompletion(
+      t,
+      completion,
+      makeSuccess([makeCompleteWorkflowExecution(defaultPayloadConverter.toPayload(0))])
+    );
+  }
+});

packages/test/src/workflows/condition-timeout-0.ts

@@ -0,0 +1,41 @@
+/**
+ * Prior to 1.5.0, `condition(fn, 0)` was treated the same as `condition(..., undefined)`,
+ * which means that the condition would block indefinitely and would return undefined once
+ * fn evaluates to true, rather than returning true or false.
+ */
+import { condition, setHandler, defineSignal, sleep, ApplicationFailure } from '@temporalio/workflow';
+
+export const aSignal = defineSignal('a');
+export const bSignal = defineSignal('b');
+
+export async function conditionTimeout0(): Promise<number> {
+  let counter = 0;
+
+  let aSignalReceived = false;
+  setHandler(aSignal, () => {
+    aSignalReceived = true;
+  });
+
+  let bSignalReceived = false;
+  setHandler(bSignal, () => {
+    bSignalReceived = true;
+  });
+
+  const aResult = await condition(() => aSignalReceived, 0);
+  if (aResult === true || aResult === undefined) counter += 1;
+
+  // Do it a second time, so that we can validate that patching logic works
+  const bResult = await condition(() => bSignalReceived, 0);
+  if (bResult === true || bResult === undefined) counter += 10;
+
+  return counter;
+}
+
+export async function conditionTimeout0Simple(): Promise<boolean | undefined> {
+  let validationTimerFired = false;
+  sleep(1000)
+    .then(() => (validationTimerFired = true))
+    .catch((e) => console.log(e));
+
+  return await condition(() => validationTimerFired, 0);
+}

packages/test/src/workflows/index.ts

@@ -26,6 +26,7 @@ export * from './child-workflow-termination';
 export * from './child-workflow-timeout';
 export * from './condition';
 export * from './condition-completion-race';
+export * from './condition-timeout-0';
 export * from './continue-as-new-same-workflow';
 export * from './continue-as-new-to-different-workflow';
 export * from './date';

packages/workflow/src/internals.ts

@@ -84,6 +84,16 @@ export type ActivationHandler = {
   [P in keyof coresdk.workflow_activation.IWorkflowActivationJob]: ActivationHandlerFunction<P>;
 };
 
+/**
+ * SDK Internal Patches are created by the SDK to avoid breaking history when behaviour
+ * of existing API need to be modified. This is the patch number supported by the current
+ * version of the SDK.
+ *
+ * History:
+ * 1: Fix `condition(..., 0)` is not the same as `condition(..., undefined)`
+ */
+export const LATEST_INTERNAL_PATCH_NUMBER = 1;
+
 /**
  * Keeps all of the Workflow runtime state like pending completions for activities and timers.
  *
@@ -260,6 +270,12 @@ export class Activator implements ActivationHandler {
    */
   public readonly sentPatches = new Set<string>();
 
+  /**
+   * SDK Internal Patches are created by the SDK to avoid breaking history when behaviour
+   * of existing API need to be modified.
+   */
+  public internalPatchNumber = 0;
+
   sinkCalls = Array<SinkCall>();
 
   constructor({
@@ -571,7 +587,28 @@ export class Activator implements ActivationHandler {
     if (!activation.patchId) {
       throw new TypeError('Notify has patch missing patch name');
     }
-    this.knownPresentPatches.add(activation.patchId);
+    if (activation.patchId.startsWith('__sdk_internal_patch_number:')) {
+      const internalPatchNumber = parseInt(activation.patchId.substring('__sdk_internal_patch_number:'.length));
+      if (internalPatchNumber > LATEST_INTERNAL_PATCH_NUMBER)
+        throw new IllegalStateError(
+          `Unsupported internal patch number: ${internalPatchNumber} > ${LATEST_INTERNAL_PATCH_NUMBER}`
+        );
+      if (this.internalPatchNumber < internalPatchNumber) this.internalPatchNumber = internalPatchNumber;
+    } else {
+      this.knownPresentPatches.add(activation.patchId);
+    }
+  }
+
+  public checkInternalPatchAtLeast(minimumPatchNumber: number): boolean {
+    if (this.internalPatchNumber >= minimumPatchNumber) return true;
+    if (!this.info.unsafe.isReplaying) {
+      this.internalPatchNumber = minimumPatchNumber;
+      this.pushCommand({
+        setPatchMarker: { patchId: `__sdk_internal_patch_number:${LATEST_INTERNAL_PATCH_NUMBER}`, deprecated: false },
+      });
+      return true;
+    }
+    return false;
   }
 
   public removeFromCache(): void {

packages/workflow/src/workflow.ts

@@ -1080,7 +1080,11 @@ export function condition(fn: () => boolean, timeout: number | string): Promise<
 export function condition(fn: () => boolean): Promise<void>;
 
 export async function condition(fn: () => boolean, timeout?: number | string): Promise<void | boolean> {
-  if (timeout) {
+  // Prior to 1.5.0, `condition(fn, 0)` was treated as equivalent to `condition(fn, undefined)`
+  if (timeout === 0 && !getActivator().checkInternalPatchAtLeast(1)) {
+    return conditionInner(fn);
+  }
+  if (typeof timeout === 'number' || typeof timeout === 'string') {
     return CancellationScope.cancellable(async () => {
       try {
         return await Promise.race([sleep(timeout).then(() => false), conditionInner(fn).then(() => true)]);