fix(workflow): Fix isReplaying being incorrectly true due to query (#1211)

Author: mjameswh

packages/core-bridge/sdk-core

@@ -8,6 +8,7 @@ import {
   Runtime,
   LoggerSinks as DefaultLoggerSinks,
   InjectedSinkFunction,
+  WorkerOptions,
 } from '@temporalio/worker';
 import { SearchAttributes, WorkflowInfo } from '@temporalio/workflow';
 import { UnsafeWorkflowInfo } from '@temporalio/workflow/src/interfaces';
@@ -406,4 +407,56 @@ if (RUN_INTEGRATION_TESTS) {
       },
     ]);
   });
+
+  test('Core issue 589', async (t) => {
+    const taskQueue = `${__filename}-${t.title}`;
+
+    const recordedMessages = Array<{ message: string; historyLength: number; isReplaying: boolean }>();
+    const sinks: InjectedSinks<workflows.CustomLoggerSinks> = {
+      customLogger: {
+        info: {
+          fn: async (info, message) => {
+            recordedMessages.push({
+              message,
+              historyLength: info.historyLength,
+              isReplaying: info.unsafe.isReplaying,
+            });
+          },
+          callDuringReplay: true,
+        },
+      },
+    };
+
+    const client = new WorkflowClient();
+    const handle = await client.start(workflows.coreIssue589, { taskQueue, workflowId: uuid4() });
+
+    const workerOptions: WorkerOptions = {
+      ...defaultOptions,
+      taskQueue,
+      sinks,
+      maxCachedWorkflows: 2,
+      maxConcurrentWorkflowTaskExecutions: 2,
+
+      // Cut down on execution time
+      stickyQueueScheduleToStartTimeout: 1,
+    };
+
+    await (await Worker.create(workerOptions)).runUntil(new Promise((resolve) => setTimeout(resolve, 1000)));
+    await (
+      await Worker.create(workerOptions)
+    ).runUntil(async () => {
+      await handle.query('q').catch(() => undefined);
+      await handle.signal(workflows.unblockSignal);
+      await handle.result();
+    });
+
+    const checkpointEntries = recordedMessages.filter((m) => m.message.startsWith('Checkpoint'));
+    t.deepEqual(checkpointEntries, [
+      {
+        message: 'Checkpoint, replaying: false, hl: 8',
+        historyLength: 8,
+        isReplaying: false,
+      },
+    ]);
+  });
 }

packages/test/src/test-sinks.ts

@@ -0,0 +1,22 @@
+import * as wf from '@temporalio/workflow';
+import { CustomLoggerSinks } from './log-sink-tester';
+import { unblockSignal } from './definitions';
+
+const { customLogger } = wf.proxySinks<CustomLoggerSinks>();
+
+// Demo for https://github.com/temporalio/sdk-core/issues/589
+export async function coreIssue589(): Promise<void> {
+  wf.setHandler(wf.defineQuery('q'), () => {
+    return 'not important';
+  });
+
+  let unblocked = false;
+  wf.setHandler(unblockSignal, () => {
+    unblocked = true;
+  });
+  await wf.condition(() => unblocked, 10000);
+
+  customLogger.info(
+    `Checkpoint, replaying: ${wf.workflowInfo().unsafe.isReplaying}, hl: ${wf.workflowInfo().historyLength}`
+  );
+}

packages/test/src/workflows/core-issue-589.ts

@@ -29,6 +29,7 @@ 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 './core-issue-589';
 export * from './date';
 export * from './deferred-resolve';
 export * from './definitions';