Allow logging queries validators during replay (#1927)

Author: THardy98

packages/test/src/run-activation-perf-tests.ts

@@ -85,7 +85,7 @@ if (!wf.inWorkflowContext()) {
         historyLength: 3,
         historySize: 300,
         continueAsNewSuggested: false,
-        unsafe: { isReplaying: false, now: Date.now },
+        unsafe: { isReplaying: false, isReplayingHistoryEvents: false, now: Date.now },
         startTime: new Date(),
         runStartTime: new Date(),
       };

packages/test/src/test-integration-split-one.ts

@@ -752,7 +752,7 @@ test.serial('Workflow can read WorkflowInfo', configMacro, async (t, config) =>
     currentBuildId: res.currentBuildId, // eslint-disable-line deprecation/deprecation
     currentDeploymentVersion: res.currentDeploymentVersion,
     // unsafe.now is a function, so doesn't make it through serialization, but .now is required, so we need to cast
-    unsafe: { isReplaying: false } as UnsafeWorkflowInfo,
+    unsafe: { isReplaying: false, isReplayingHistoryEvents: false } as UnsafeWorkflowInfo,
     priority: {},
   });
 });

packages/test/src/test-sinks.ts

@@ -146,6 +146,7 @@ if (RUN_INTEGRATION_TESTS) {
       // unsafe.now() doesn't make it through serialization, but .now is required, so we need to cast
       unsafe: {
         isReplaying: false,
+        isReplayingHistoryEvents: false,
       } as UnsafeWorkflowInfo,
       priority: {
         fairnessKey: undefined,
@@ -554,4 +555,72 @@ if (RUN_INTEGRATION_TESTS) {
       },
     ]);
   });
+
+  test('Logging is allowed in query handlers and update validators', async (t) => {
+    const taskQueue = `${__filename}-${t.title}`;
+
+    let recordedMessages: { message: string; isReplaying: boolean }[] = [];
+    const sinks: InjectedSinks<workflows.CustomLoggerSinks> = {
+      customLogger: {
+        info: {
+          fn: async (info, message) => {
+            recordedMessages.push({
+              message,
+              isReplaying: info.unsafe.isReplaying,
+            });
+          },
+        },
+      },
+    };
+
+    const client = new WorkflowClient();
+    const handle = await client.start(workflows.queryAndValidatorLogging, { taskQueue, workflowId: uuid4() });
+
+    const worker = await Worker.create({
+      ...defaultOptions,
+      taskQueue,
+      sinks,
+      // Avoid waiting for sticky execution timeout on worker transition
+      stickyQueueScheduleToStartTimeout: '1s',
+    });
+
+    await worker.runUntil(async () => {
+      await handle.query(workflows.loggingQuery);
+      await handle.executeUpdate(workflows.loggingUpdate, { args: ['good'] });
+    });
+
+    let messages = recordedMessages.map((m) => m.message);
+    t.true(messages.includes('Query handler called'), 'Query handler log should be emitted');
+    t.true(messages.includes('Update validator called'), 'Update validator log should be emitted');
+    t.true(messages.includes('Update handler called'), 'Update handler log should be emitted');
+
+    const worker2 = await Worker.create({
+      ...defaultOptions,
+      taskQueue,
+      sinks,
+      // Avoid waiting for sticky execution timeout on worker transition
+      stickyQueueScheduleToStartTimeout: '1s',
+    });
+
+    // Empty recorded messages
+    recordedMessages = [];
+
+    // Run the entire workflow through worker 2 (will replay).
+    await worker2.runUntil(async () => {
+      await handle.query(workflows.loggingQuery);
+      // No update - it will be replayed
+      await handle.signal(workflows.unblockSignal);
+      const res = await handle.result();
+      // Assert that the update replayed (expect initial update arg as result)
+      t.is(res, 'good');
+    });
+
+    messages = recordedMessages.map((m) => m.message);
+    // Query is a live operation even during replay — log should be emitted
+    t.true(messages.includes('Query handler called'), 'Query handler log should be emitted on replay');
+    // Validator does not re-run during replay (Core sends runValidator: false for accepted updates)
+    t.false(messages.includes('Update validator called'), 'Update validator log should not be emitted on replay');
+    // Update handler re-runs during replay to rebuild state, but its logs should be suppressed
+    t.false(messages.includes('Update handler called'), 'Update handler log should not be emitted on replay');
+  });
 }

packages/test/src/test-workflows.ts

@@ -122,7 +122,7 @@ async function createWorkflow(
       historyLength: 3,
       historySize: 300,
       continueAsNewSuggested: false,
-      unsafe: { isReplaying: false, now: Date.now },
+      unsafe: { isReplaying: false, isReplayingHistoryEvents: false, now: Date.now },
       startTime: new Date(),
       runStartTime: new Date(),
     },

packages/test/src/workflows/index.ts

@@ -54,6 +54,7 @@ export * from './patched';
 export * from './patched-top-level';
 export * from './priority';
 export * from './promise-all';
+export * from './query-and-validator-logging';
 export * from './promise-race';
 export * from './promise-then-promise';
 export * from './race';

packages/test/src/workflows/query-and-validator-logging.ts

@@ -0,0 +1,42 @@
+import * as wf from '@temporalio/workflow';
+import { CustomLoggerSinks } from './log-sink-tester';
+import { unblockSignal } from './definitions';
+
+const { customLogger } = wf.proxySinks<CustomLoggerSinks>();
+
+export const loggingQuery = wf.defineQuery<string>('loggingQuery');
+export const loggingUpdate = wf.defineUpdate<string, [string]>('loggingUpdate');
+
+export async function queryAndValidatorLogging(): Promise<string> {
+  let lastSignal = '';
+  let updateArg = '';
+
+  wf.setHandler(loggingQuery, () => {
+    customLogger.info('Query handler called');
+    return lastSignal;
+  });
+
+  wf.setHandler(
+    loggingUpdate,
+    (arg: string) => {
+      customLogger.info('Update handler called');
+      updateArg = arg;
+      return `update-result: ${arg}`;
+    },
+    {
+      validator: (arg: string) => {
+        customLogger.info('Update validator called');
+        if (arg === 'bad') {
+          throw new Error('Validation failed');
+        }
+      },
+    }
+  );
+
+  wf.setHandler(unblockSignal, () => {
+    lastSignal = 'unblocked';
+  });
+
+  await wf.condition(() => lastSignal === 'unblocked');
+  return updateArg;
+}

packages/worker/src/worker.ts

@@ -1459,12 +1459,10 @@ export class Worker {
           // When processing workflows through runReplayHistories, Core may still send non-replay
           // activations on the very last Workflow Task in some cases. Though Core is technically exact
           // here, the fact that sinks marked with callDuringReplay = false may get called on a replay
-          // worker is definitely a surprising behavior. For that reason, we extend the isReplaying flag in
-          // this case to also include anything running under in a replay worker.
-          const isReplaying = activation.isReplaying || this.isReplayWorker;
-
+          // worker is definitely a surprising behavior. For that reason, processSinkCalls checks
+          // this.isReplayWorker to suppress all non-callDuringReplay sinks on replay workers.
           const calls = await workflow.workflow.getAndResetSinkCalls();
-          await this.processSinkCalls(calls, isReplaying, workflow.logAttributes);
+          await this.processSinkCalls(calls, workflow.logAttributes);
         }
         this.logger.trace('Completed activation', workflow.logAttributes);
       }
@@ -1572,6 +1570,7 @@ export class Worker {
       unsafe: {
         now: () => Date.now(), // re-set in initRuntime
         isReplaying: activation.isReplaying,
+        isReplayingHistoryEvents: activation.isReplaying,
       },
       priority: decodePriority(priority),
     };
@@ -1597,11 +1596,7 @@ export class Worker {
    * This function does not throw, it will log in case of missing sinks
    * or failed sink function invocations.
    */
-  protected async processSinkCalls(
-    externalCalls: SinkCall[],
-    isReplaying: boolean,
-    logAttributes: Record<string, unknown>
-  ): Promise<void> {
+  protected async processSinkCalls(externalCalls: SinkCall[], logAttributes: Record<string, unknown>): Promise<void> {
     const { sinks } = this.options;
 
     const filteredCalls = externalCalls
@@ -1619,8 +1614,15 @@ export class Worker {
         });
         return false;
       })
-      // If appropriate, reject calls to sink functions not configured with `callDuringReplay = true`
-      .filter(({ sink }) => sink?.callDuringReplay || !isReplaying);
+      // If appropriate, reject calls to sink functions not configured with `callDuringReplay = true`.
+      // Use per-call isReplayingHistoryEvents (which is false during queries and update validators)
+      // rather than per-activation isReplaying, so that logging is permitted during live read-only operations.
+      // Replay workers still suppress all non-callDuringReplay sinks regardless.
+      .filter(({ call, sink }) => {
+        if (sink?.callDuringReplay) return true;
+        if (this.isReplayWorker) return false;
+        return !call.workflowInfo.unsafe.isReplayingHistoryEvents;
+      });
 
     // Make a wrapper function, to make things easier afterward
     await Promise.all(

packages/worker/src/workflow/vm-shared.ts

@@ -116,7 +116,7 @@ export function injectGlobals(context: vm.Context): void {
         console[level](`[not in workflow context]`, ...args);
       } else {
         const { info } = sandboxGlobalThis.__TEMPORAL_ACTIVATOR__!;
-        if (info.unsafe.isReplaying) return;
+        if (info.unsafe.isReplayingHistoryEvents) return;
         console[level](`[${info.workflowType}(${info.workflowId})]`, ...args);
       }
     };
@@ -394,6 +394,7 @@ export abstract class BaseVMWorkflow implements Workflow {
         unsafe: {
           ...info.unsafe,
           isReplaying: activation.isReplaying ?? false,
+          isReplayingHistoryEvents: activation.isReplaying ?? false,
         },
       }));
       this.activator.addKnownFlags(activation.availableInternalFlags ?? []);
@@ -470,6 +471,8 @@ export abstract class BaseVMWorkflow implements Workflow {
       unsafe: {
         ...info.unsafe,
         isReplaying: true,
+        // Queries are live read-only operations, not replay of history events
+        isReplayingHistoryEvents: false,
       },
     }));
     this.workflowModule.activate(activation);

packages/workflow/src/interfaces.ts

@@ -233,7 +233,22 @@ export interface UnsafeWorkflowInfo {
    */
   readonly now: () => number;
 
+  /**
+   * Whether the workflow is currently replaying.
+   */
   readonly isReplaying: boolean;
+
+  /**
+   * Whether the workflow is currently replaying history events.
+   *
+   * This is similar to {@link isReplaying}, but returns `false` during query handlers and update
+   * validators, which are live read-only operations that should not be considered as replaying
+   * history events.
+   *
+   * When this property is true, workflow log messages are suppressed and sinks defined with
+   * callDuringReplay=false won't get processed.
+   */
+  readonly isReplayingHistoryEvents: boolean;
 }
 
 /**

packages/workflow/src/internals.ts

@@ -858,12 +858,26 @@ export class Activator implements ActivationHandler {
       let input: UpdateInput;
       try {
         if (runValidator && entry.validator) {
-          const validate = composeInterceptors(
-            interceptors,
-            'validateUpdate',
-            this.validateUpdateNextHandler.bind(this, entry.validator)
-          );
-          validate(makeInput());
+          // Temporarily mark as not replaying history events during validator execution
+          // so that logging is permitted. Validators are live read-only operations.
+          const wasReplayingHistoryEvents = this.info.unsafe.isReplayingHistoryEvents;
+          this.mutateWorkflowInfo((info) => ({
+            ...info,
+            unsafe: { ...info.unsafe, isReplayingHistoryEvents: false },
+          }));
+          try {
+            const validate = composeInterceptors(
+              interceptors,
+              'validateUpdate',
+              this.validateUpdateNextHandler.bind(this, entry.validator)
+            );
+            validate(makeInput());
+          } finally {
+            this.mutateWorkflowInfo((info) => ({
+              ...info,
+              unsafe: { ...info.unsafe, isReplayingHistoryEvents: wasReplayingHistoryEvents },
+            }));
+          }
         }
         input = makeInput();
       } catch (error) {