improvements

Author: THardy98

packages/common/src/reserved.ts

@@ -8,14 +8,21 @@ export const reservedPrefixes = [
   ENHANCED_STACK_TRACE_RESERVED_PREFIX,
 ];
 
+export class ReservedPrefixError extends Error {
+  constructor(type: string, name: string, prefix: string) {
+    super(`Cannot use ${type} name: '${name}', with reserved prefix: '${prefix}'`);
+    this.name = 'ReservedPrefixError';
+  }
+}
+
 export function throwIfReservedName(type: string, name: string): void {
-  const prefix = isReservedName(name);
+  const prefix = maybeGetReservedPrefix(name);
   if (prefix) {
-    throw Error(`Cannot register ${type} name: '${name}', with reserved prefix: '${prefix}'`);
+    throw new ReservedPrefixError(type, name, prefix);
   }
 }
 
-export function isReservedName(name: string): string | undefined {
+export function maybeGetReservedPrefix(name: string): string | undefined {
   for (const prefix of reservedPrefixes) {
     if (name.startsWith(prefix)) {
       return prefix;

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

@@ -2,7 +2,7 @@
 import asyncRetry from 'async-retry';
 import { v4 as uuid4 } from 'uuid';
 import * as iface from '@temporalio/proto';
-import { WorkflowContinuedAsNewError, WorkflowFailedError } from '@temporalio/client';
+import { WorkflowContinuedAsNewError, WorkflowFailedError, WorkflowHandle } from '@temporalio/client';
 import {
   ApplicationFailure,
   defaultPayloadConverter,
@@ -20,6 +20,7 @@ import {
   defineSignal,
   defineUpdate,
   setDefaultQueryHandler,
+  setDefaultSignalHandler,
   setHandler,
   sleep,
 } from '@temporalio/workflow';
@@ -859,3 +860,71 @@ test('Workflow failure if define signals/updates/queries with reserved prefixes'
     // }
   });
 });
+
+export async function workflowWithDefaultHandlers(): Promise<void> {
+  let unblocked = false;
+  setHandler(defineSignal('unblock'), () => {
+    unblocked = true;
+  });
+
+  setDefaultQueryHandler(() => {});
+  setDefaultSignalHandler(() => {});
+  setDefaultUpdateHandler({
+    handler: () => {},
+  });
+
+  await condition(() => unblocked);
+}
+
+test('Default handlers fail WFT given reserved prefix', configMacro, async (t, config) => {
+  const { env, createWorkerWithDefaults } = config;
+  const { startWorkflow } = configurableHelpers(t, t.context.workflowBundle, env);
+  const worker = await createWorkerWithDefaults(t);
+
+  const assertWftFailure = async (
+    handle: WorkflowHandle,
+    name: string,
+    prefix: string,
+    handlerType: 'query' | 'signal' | 'update'
+  ) => {
+    await asyncRetry(
+      async () => {
+        const history = await handle.fetchHistory();
+        const wftFailedEvent = history.events?.findLast((ev) => ev.workflowTaskFailedEventAttributes);
+        if (wftFailedEvent === undefined) {
+          throw new Error('No WFT failed event found');
+        }
+        const { failure } = wftFailedEvent.workflowTaskFailedEventAttributes ?? {};
+        if (!failure) {
+          return t.fail('Expected failure in workflowTaskFailedEventAttributes');
+        }
+        t.is(failure.message, `Cannot use ${handlerType} name: '${name}', with reserved prefix: '${prefix}'`);
+      },
+      { minTimeout: 300, factor: 1, retries: 10 }
+    );
+  };
+
+  await worker.runUntil(async () => {
+    for (const prefix of reservedPrefixes) {
+      // Test Query
+      let handle = await startWorkflow(workflowWithDefaultHandlers);
+      const queryName = `${prefix}_query`;
+      await t.throwsAsync(handle.query(queryName), undefined, `Query ${queryName} should fail`);
+      await assertWftFailure(handle, queryName, prefix, 'query');
+      await handle.terminate();
+       // Test Signal
+       handle = await startWorkflow(workflowWithDefaultHandlers);
+       const signalName = `${prefix}_signal`;
+       await handle.signal(signalName);
+       await assertWftFailure(handle, signalName, prefix, 'signal');
+       await handle.terminate();
+ 
+       // Test Update
+       handle = await startWorkflow(workflowWithDefaultHandlers);
+       const updateName = `${prefix}_update`;
+       await t.throwsAsync(handle.executeUpdate(updateName), undefined, `Update ${updateName} should fail`);
+       await assertWftFailure(handle, updateName, prefix, 'update');
+       await handle.terminate();
+    }
+  });
+});

packages/workflow/src/internals.ts

@@ -33,8 +33,9 @@ import { makeProtoEnumConverters } from '@temporalio/common/lib/internal-workflo
 import type { coresdk, temporal } from '@temporalio/proto';
 import {
   ENHANCED_STACK_TRACE_RESERVED_PREFIX,
+  ReservedPrefixError,
   STACK_TRACE_RESERVED_PREFIX,
-  isReservedName,
+  maybeGetReservedPrefix,
   throwIfReservedName,
 } from '@temporalio/common/lib/reserved';
 import { alea, RNG } from './alea';
@@ -685,28 +686,27 @@ export class Activator implements ActivationHandler {
       throw new TypeError('Missing query activation attributes');
     }
 
-    const queryInput = {
-      queryName: queryType,
-      args: arrayFromPayloads(this.payloadConverter, activation.arguments),
-      queryId,
-      headers: headers ?? {},
-    };
-
-    // Skip interceptors if this is an internal query.
-    if (isReservedName(queryType)) {
-      this.queryWorkflowNextHandler(queryInput).then(
-        (result) => this.completeQuery(queryId, result),
-        (reason) => this.failQuery(queryId, reason)
-      );
-      return;
+    const reservedPrefix = maybeGetReservedPrefix(queryType)
+    if (reservedPrefix) {
+      // Must have (internal) query handler for reserved query.
+      if (!this.queryHandlers.has(queryType)) {
+        throw new ReservedPrefixError('query', queryType, reservedPrefix);
+      }
     }
 
+    // Skip interceptors if it is an internal query
+    let interceptors = reservedPrefix ? [] : this.interceptors.inbound
     const execute = composeInterceptors(
-      this.interceptors.inbound,
+      interceptors,
       'handleQuery',
       this.queryWorkflowNextHandler.bind(this)
     );
-    execute(queryInput).then(
+    execute({
+      queryName: queryType,
+      args: arrayFromPayloads(this.payloadConverter, activation.arguments),
+      queryId,
+      headers: headers ?? {},
+    }).then(
       (result) => this.completeQuery(queryId, result),
       (reason) => this.failQuery(queryId, reason)
     );
@@ -737,6 +737,11 @@ export class Activator implements ActivationHandler {
 
     // If we don't have an entry from either source, buffer and return
     if (entry === null) {
+      const reservedPrefix = maybeGetReservedPrefix(name);
+      if (reservedPrefix) {
+        // Must have (internal) update handler for reserved update.
+        throw new ReservedPrefixError('update', name, reservedPrefix);
+      }
       this.bufferedUpdates.push(activation);
       return;
     }
@@ -864,8 +869,6 @@ export class Activator implements ActivationHandler {
     if (fn) {
       return await fn(...args);
     } else if (this.defaultSignalHandler) {
-      // Do not call default signal handler with reserved signal name.
-      throwIfReservedName('signal', signalName);
       return await this.defaultSignalHandler(signalName, ...args);
     } else {
       throw new IllegalStateError(`No registered signal handler for signal: ${signalName}`);
@@ -878,6 +881,14 @@ export class Activator implements ActivationHandler {
       throw new TypeError('Missing activation signalName');
     }
 
+    const reservedPrefix = maybeGetReservedPrefix(signalName);
+    if (reservedPrefix) {
+      if (!this.signalHandlers.has(signalName)) {
+        // Must have (internal) signal handler for reserved signal.
+        throw new ReservedPrefixError('signal', signalName, reservedPrefix);
+      }
+    }
+
     if (!this.signalHandlers.has(signalName) && !this.defaultSignalHandler) {
       this.bufferedSignals.push(activation);
       return;