fix(workflow): Fix internal patches NDE on replay (#1106)

mjameswh · bergundy · web-flow · commit c553b61bf4b0 · 2023-04-26T09:43:22.000-07:00
Co-authored-by: Roey Berman &lt;roey.berman@gmail.com&gt;
diff --git a/packages/test/src/helpers.ts b/packages/test/src/helpers.ts
@@ -81,7 +81,6 @@ export const bundlerOptions = {
     '@temporalio/common/lib/internal-non-workflow',
     '@temporalio/activity',
     '@temporalio/client',
-    '@temporalio/proto',
     '@temporalio/testing',
     '@temporalio/worker',
     'ava',
diff --git a/packages/test/src/integration-tests.ts b/packages/test/src/integration-tests.ts
@@ -45,7 +45,7 @@ import { cleanOptionalStackTrace, registerDefaultCustomSearchAttributes, u8, Wor
 import * as workflows from './workflows';
 import { withZeroesHTTPServer } from './zeroes-http-server';
 
-const { EVENT_TYPE_TIMER_STARTED, EVENT_TYPE_TIMER_FIRED, EVENT_TYPE_TIMER_CANCELED, EVENT_TYPE_MARKER_RECORDED } =
+const { EVENT_TYPE_TIMER_STARTED, EVENT_TYPE_TIMER_FIRED, EVENT_TYPE_TIMER_CANCELED } =
   iface.temporal.api.enums.v1.EventType;
 
 const timerEventTypes = new Set([EVENT_TYPE_TIMER_STARTED, EVENT_TYPE_TIMER_FIRED, EVENT_TYPE_TIMER_CANCELED]);
@@ -1385,27 +1385,4 @@ 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);
-  });
 }
diff --git a/packages/test/src/test-integration-workflows.ts b/packages/test/src/test-integration-workflows.ts
@@ -204,3 +204,13 @@ test('Worker allows heartbeating activities after shutdown has been requested',
   await worker.run();
   t.is(cancelReason, 'CANCELLED');
 });
+
+export async function conditionTimeout0(): Promise<boolean | undefined> {
+  return await workflow.condition(() => false, 0);
+}
+
+test('Condition 0 patch sets a timer', async (t) => {
+  const { createWorker, executeWorkflow } = helpers(t);
+  const worker = await createWorker();
+  t.false(await worker.runUntil(executeWorkflow(conditionTimeout0)));
+});
diff --git a/packages/test/src/test-workflows.ts b/packages/test/src/test-workflows.ts
@@ -1889,86 +1889,6 @@ 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, 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, makeFireTimer(1));
-    compareCompletion(
-      t,
-      completion,
-      makeSuccess([
-        makeStartTimerCommand({
-          seq: 2,
-          startToFireTimeout: msToTs(1),
-        }),
-      ])
-    );
-  }
-  {
-    const completion = await activate(t, makeFireTimer(2));
-    compareCompletion(
-      t,
-      completion,
-      makeSuccess([makeCompleteWorkflowExecution(defaultPayloadConverter.toPayload(0))])
-    );
-  }
-});
-
 test('Buffered signals are dispatched to correct handler and in correct order - signalsOrdering', async (t) => {
   const { workflowType } = t.context;
   {
diff --git a/packages/test/src/workflows/condition-timeout-0.ts b/packages/test/src/workflows/condition-timeout-0.ts
@@ -3,7 +3,7 @@
  * 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 } from '@temporalio/workflow';
+import { condition, setHandler, defineSignal } from '@temporalio/workflow';
 
 export const aSignal = defineSignal('a');
 export const bSignal = defineSignal('b');
@@ -30,12 +30,3 @@ export async function conditionTimeout0(): Promise<number> {
 
   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);
-}
diff --git a/packages/workflow/src/internals.ts b/packages/workflow/src/internals.ts
@@ -86,16 +86,6 @@ 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.
  *
@@ -281,12 +271,6 @@ 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({
@@ -304,8 +288,8 @@ export class Activator implements ActivationHandler {
     this.random = alea(randomnessSeed);
 
     if (info.unsafe.isReplaying) {
-      for (const patch of patches) {
-        this.knownPresentPatches.add(patch);
+      for (const patchId of patches) {
+        this.notifyHasPatch({ patchId });
       }
     }
   }
@@ -611,28 +595,7 @@ export class Activator implements ActivationHandler {
     if (!activation.patchId) {
       throw new TypeError('Notify has patch missing patch name');
     }
-    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;
+    this.knownPresentPatches.add(activation.patchId);
   }
 
   public removeFromCache(): void {
diff --git a/packages/workflow/src/workflow.ts b/packages/workflow/src/workflow.ts
@@ -565,6 +565,9 @@ export function proxyLocalActivities<A = UntypedActivities>(options: LocalActivi
 
 // TODO: deprecate this patch after "enough" time has passed
 const EXTERNAL_WF_CANCEL_PATCH = '__temporal_internal_connect_external_handle_cancel_to_scope';
+// This generic name of this patch comes from an attempt to build a generic internal patching mechanism.
+// That effort has been abandoned in favor of a newer WorkflowTaskCompletedMetadata based mechanism.
+const CONDITION_0_PATCH = '__sdk_internal_patch_number:1';
 
 /**
  * Returns a client-side handle that can be used to signal and cancel an existing Workflow execution.
@@ -1072,7 +1075,7 @@ export function condition(fn: () => boolean): Promise<void>;
 
 export async function condition(fn: () => boolean, timeout?: number | string): Promise<void | boolean> {
   // Prior to 1.5.0, `condition(fn, 0)` was treated as equivalent to `condition(fn, undefined)`
-  if (timeout === 0 && !getActivator().checkInternalPatchAtLeast(1)) {
+  if (timeout === 0 && !patched(CONDITION_0_PATCH)) {
     return conditionInner(fn);
   }
   if (typeof timeout === 'number' || typeof timeout === 'string') {
