add checks for sinks, simplify reserved name checking, improve names

THardy98 · THardy98 · commit 73823fa2fa4c · 2025-07-23T10:17:16.000-04:00
diff --git a/packages/common/src/reserved.ts b/packages/common/src/reserved.ts
@@ -1,31 +1,26 @@
 export const TEMPORAL_RESERVED_PREFIX = '__temporal_';
-export const STACK_TRACE_RESERVED_PREFIX = '__stack_trace';
-export const ENHANCED_STACK_TRACE_RESERVED_PREFIX = '__enhanced_stack_trace';
+export const STACK_TRACE_RESERVED_NAME = '__stack_trace';
+export const ENHANCED_STACK_TRACE_RESERVED_NAME = '__enhanced_stack_trace';
 
-export const reservedPrefixes = [
-  TEMPORAL_RESERVED_PREFIX,
-  STACK_TRACE_RESERVED_PREFIX,
-  ENHANCED_STACK_TRACE_RESERVED_PREFIX,
-];
+/**
+ * Valid entity types that can be checked for reserved name violations
+ */
+export type ReservedNameEntityType = 'query' | 'signal' | 'update' | 'activity' | 'task queue' | 'sink' | 'workflow';
 
-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';
+/**
+ * Validates if the provided name contains any reserved prefixes or matches any reserved names.
+ * Throws a TypeError if validation fails, with a specific message indicating whether the issue
+ * is with a reserved prefix or an exact match to a reserved name.
+ *
+ * @param type The entity type being checked
+ * @param name The name to check against reserved prefixes/names
+ */
+export function throwIfReservedName(type: ReservedNameEntityType, name: string): void {
+  if (name.startsWith(TEMPORAL_RESERVED_PREFIX)) {
+    throw new TypeError(`Cannot use ${type} name: '${name}', with reserved prefix: '${TEMPORAL_RESERVED_PREFIX}'`);
   }
-}
-
-export function throwIfReservedName(type: string, name: string): void {
-  const prefix = maybeGetReservedPrefix(name);
-  if (prefix) {
-    throw new ReservedPrefixError(type, name, prefix);
-  }
-}
 
-export function maybeGetReservedPrefix(name: string): string | undefined {
-  for (const prefix of reservedPrefixes) {
-    if (name.startsWith(prefix)) {
-      return prefix;
-    }
+  if (name === STACK_TRACE_RESERVED_NAME || name === ENHANCED_STACK_TRACE_RESERVED_NAME) {
+    throw new TypeError(`Cannot use ${type} name: '${name}', which is a reserved name`);
   }
 }
diff --git a/packages/test/src/test-integration-workflows.ts b/packages/test/src/test-integration-workflows.ts
@@ -29,7 +29,11 @@ import {
   TypedSearchAttributes,
   WorkflowExecutionAlreadyStartedError,
 } from '@temporalio/common';
-import { reservedPrefixes } from '@temporalio/common/lib/reserved';
+import {
+  TEMPORAL_RESERVED_PREFIX,
+  STACK_TRACE_RESERVED_NAME,
+  ENHANCED_STACK_TRACE_RESERVED_NAME,
+} from '@temporalio/common/lib/reserved';
 import { signalSchedulingWorkflow } from './activities/helpers';
 import { activityStartedSignal } from './workflows/definitions';
 import * as workflows from './workflows';
@@ -1426,18 +1430,23 @@ test('Workflow can return root workflow', async (t) => {
   });
 });
 
+const reservedNames = [TEMPORAL_RESERVED_PREFIX, STACK_TRACE_RESERVED_NAME, ENHANCED_STACK_TRACE_RESERVED_NAME];
+
 test('Cannot register activities using reserved prefixes', async (t) => {
   const { createWorker } = helpers(t);
 
-  for (const prefix of reservedPrefixes) {
-    const activityName = prefix + '_test';
+  for (const name of reservedNames) {
+    const activityName = name === TEMPORAL_RESERVED_PREFIX ? name + '_test' : name;
     await t.throwsAsync(
       createWorker({
         activities: { [activityName]: () => {} },
       }),
       {
-        name: 'ReservedPrefixError',
-        message: `Cannot use activity name: '${activityName}', with reserved prefix: '${prefix}'`,
+        name: 'TypeError',
+        message:
+          name === TEMPORAL_RESERVED_PREFIX
+            ? `Cannot use activity name: '${activityName}', with reserved prefix: '${name}'`
+            : `Cannot use activity name: '${activityName}', which is a reserved name`,
       }
     );
   }
@@ -1446,16 +1455,45 @@ test('Cannot register activities using reserved prefixes', async (t) => {
 test('Cannot register task queues using reserved prefixes', async (t) => {
   const { createWorker } = helpers(t);
 
-  for (const prefix of reservedPrefixes) {
-    const taskQueue = prefix + '_test';
+  for (const name of reservedNames) {
+    const taskQueue = name === TEMPORAL_RESERVED_PREFIX ? name + '_test' : name;
 
     await t.throwsAsync(
       createWorker({
         taskQueue,
       }),
       {
-        name: 'ReservedPrefixError',
-        message: `Cannot use task queue name: '${taskQueue}', with reserved prefix: '${prefix}'`,
+        name: 'TypeError',
+        message:
+          name === TEMPORAL_RESERVED_PREFIX
+            ? `Cannot use task queue name: '${taskQueue}', with reserved prefix: '${name}'`
+            : `Cannot use task queue name: '${taskQueue}', which is a reserved name`,
+      }
+    );
+  }
+});
+
+test('Cannot register sinks using reserved prefixes', async (t) => {
+  const { createWorker } = helpers(t);
+
+  for (const name of reservedNames) {
+    const sinkName = name === TEMPORAL_RESERVED_PREFIX ? name + '_test' : name;
+    await t.throwsAsync(
+      createWorker({
+        sinks: {
+          [sinkName]: {
+            test: {
+              fn: () => {},
+            },
+          },
+        },
+      }),
+      {
+        name: 'TypeError',
+        message:
+          name === TEMPORAL_RESERVED_PREFIX
+            ? `Cannot use sink name: '${sinkName}', with reserved prefix: '${name}'`
+            : `Cannot use sink name: '${sinkName}', which is a reserved name`,
       }
     );
   }
@@ -1466,25 +1504,25 @@ interface HandlerError {
   message: string;
 }
 
-export async function workflowBadPrefixHandler(prefix: string): Promise<HandlerError[]> {
+export async function workflowReservedNameHandler(name: string): Promise<HandlerError[]> {
   // Re-package errors, default payload converter has trouble converting native errors (no 'data' field).
   const expectedErrors: HandlerError[] = [];
   try {
-    setHandler(defineSignal(prefix + '_signal'), () => {});
+    setHandler(defineSignal(name === TEMPORAL_RESERVED_PREFIX ? name + '_signal' : name), () => {});
   } catch (e) {
     if (e instanceof Error) {
       expectedErrors.push({ name: e.name, message: e.message });
     }
   }
   try {
-    setHandler(defineUpdate(prefix + '_update'), () => {});
+    setHandler(defineUpdate(name === TEMPORAL_RESERVED_PREFIX ? name + '_update' : name), () => {});
   } catch (e) {
     if (e instanceof Error) {
       expectedErrors.push({ name: e.name, message: e.message });
     }
   }
   try {
-    setHandler(defineQuery(prefix + '_query'), () => {});
+    setHandler(defineQuery(name === TEMPORAL_RESERVED_PREFIX ? name + '_query' : name), () => {});
   } catch (e) {
     if (e instanceof Error) {
       expectedErrors.push({ name: e.name, message: e.message });
@@ -1497,22 +1535,31 @@ test('Workflow failure if define signals/updates/queries with reserved prefixes'
   const { createWorker, executeWorkflow } = helpers(t);
   const worker = await createWorker();
   await worker.runUntil(async () => {
-    for (const prefix of reservedPrefixes) {
-      const result = await executeWorkflow(workflowBadPrefixHandler, {
-        args: [prefix],
+    for (const name of reservedNames) {
+      const result = await executeWorkflow(workflowReservedNameHandler, {
+        args: [name],
       });
       t.deepEqual(result, [
         {
-          name: 'ReservedPrefixError',
-          message: `Cannot use signal name: '${prefix}_signal', with reserved prefix: '${prefix}'`,
+          name: 'TypeError',
+          message:
+            name === TEMPORAL_RESERVED_PREFIX
+              ? `Cannot use signal name: '${name}_signal', with reserved prefix: '${name}'`
+              : `Cannot use signal name: '${name}', which is a reserved name`,
         },
         {
-          name: 'ReservedPrefixError',
-          message: `Cannot use update name: '${prefix}_update', with reserved prefix: '${prefix}'`,
+          name: 'TypeError',
+          message:
+            name === TEMPORAL_RESERVED_PREFIX
+              ? `Cannot use update name: '${name}_update', with reserved prefix: '${name}'`
+              : `Cannot use update name: '${name}', which is a reserved name`,
         },
         {
-          name: 'ReservedPrefixError',
-          message: `Cannot use query name: '${prefix}_query', with reserved prefix: '${prefix}'`,
+          name: 'TypeError',
+          message:
+            name === TEMPORAL_RESERVED_PREFIX
+              ? `Cannot use query name: '${name}_query', with reserved prefix: '${name}'`
+              : `Cannot use query name: '${name}', which is a reserved name`,
         },
       ]);
     }
@@ -1557,53 +1604,57 @@ test('Default handlers fail given reserved prefix', async (t) => {
   };
 
   await worker.runUntil(async () => {
-    for (const prefix of reservedPrefixes) {
-      // Reserved query
-      let handle = await startWorkflow(workflowWithDefaultHandlers);
-      await asyncRetry(async () => {
-        if (!(await handle.query(wfReadyQuery))) {
-          throw new Error('Workflow not ready yet');
-        }
-      });
-      const queryName = `${prefix}_query`;
-      await t.throwsAsync(
-        handle.query(queryName),
-        {
-          // ReservedPrefixError transforms to a QueryNotRegisteredError on the way back from server
-          name: 'QueryNotRegisteredError',
-          message: `Cannot use query name: '${queryName}', with reserved prefix: '${prefix}'`,
-        },
-        `Query ${queryName} should fail`
-      );
-      await handle.terminate();
-
-      // Reserved signal
-      handle = await startWorkflow(workflowWithDefaultHandlers);
-      await asyncRetry(async () => {
-        if (!(await handle.query(wfReadyQuery))) {
-          throw new Error('Workflow not ready yet');
-        }
-      });
-      const signalName = `${prefix}_signal`;
-      await handle.signal(signalName);
-      await assertWftFailure(handle, `Cannot use signal name: '${signalName}', with reserved prefix: '${prefix}'`);
-      await handle.terminate();
-
-      // Reserved update
-      handle = await startWorkflow(workflowWithDefaultHandlers);
-      await asyncRetry(async () => {
-        if (!(await handle.query(wfReadyQuery))) {
-          throw new Error('Workflow not ready yet');
-        }
-      });
-      const updateName = `${prefix}_update`;
-      handle.executeUpdate(updateName).catch(() => {
-        // Expect failure. The error caught here is a WorkflowNotFound because
-        // the workflow will have already failed, so the update cannot go through.
-        // We assert on the expected failure below.
-      });
-      await assertWftFailure(handle, `Cannot use update name: '${updateName}', with reserved prefix: '${prefix}'`);
-      await handle.terminate();
-    }
+    // Reserved query
+    let handle = await startWorkflow(workflowWithDefaultHandlers);
+    await asyncRetry(async () => {
+      if (!(await handle.query(wfReadyQuery))) {
+        throw new Error('Workflow not ready yet');
+      }
+    });
+    const queryName = `${TEMPORAL_RESERVED_PREFIX}_query`;
+    await t.throwsAsync(
+      handle.query(queryName),
+      {
+        // TypeError transforms to a QueryNotRegisteredError on the way back from server
+        name: 'QueryNotRegisteredError',
+        message: `Cannot use query name: '${queryName}', with reserved prefix: '${TEMPORAL_RESERVED_PREFIX}'`,
+      },
+      `Query ${queryName} should fail`
+    );
+    await handle.terminate();
+
+    // Reserved signal
+    handle = await startWorkflow(workflowWithDefaultHandlers);
+    await asyncRetry(async () => {
+      if (!(await handle.query(wfReadyQuery))) {
+        throw new Error('Workflow not ready yet');
+      }
+    });
+    const signalName = `${TEMPORAL_RESERVED_PREFIX}_signal`;
+    await handle.signal(signalName);
+    await assertWftFailure(
+      handle,
+      `Cannot use signal name: '${signalName}', with reserved prefix: '${TEMPORAL_RESERVED_PREFIX}'`
+    );
+    await handle.terminate();
+
+    // Reserved update
+    handle = await startWorkflow(workflowWithDefaultHandlers);
+    await asyncRetry(async () => {
+      if (!(await handle.query(wfReadyQuery))) {
+        throw new Error('Workflow not ready yet');
+      }
+    });
+    const updateName = `${TEMPORAL_RESERVED_PREFIX}_update`;
+    handle.executeUpdate(updateName).catch(() => {
+      // Expect failure. The error caught here is a WorkflowNotFound because
+      // the workflow will have already failed, so the update cannot go through.
+      // We assert on the expected failure below.
+    });
+    await assertWftFailure(
+      handle,
+      `Cannot use update name: '${updateName}', with reserved prefix: '${TEMPORAL_RESERVED_PREFIX}'`
+    );
+    await handle.terminate();
   });
 });
diff --git a/packages/worker/src/worker-options.ts b/packages/worker/src/worker-options.ts
@@ -957,6 +957,17 @@ export function compileWorkerOptions(
   for (const activityName of activities.keys()) {
     throwIfReservedName('activity', activityName);
   }
+
+  // Validate sink names to ensure they don't use reserved prefixes/names
+  if (opts.sinks) {
+    for (const sinkName of Object.keys(opts.sinks)) {
+      // Allow internal sinks used by the SDK
+      if (sinkName !== '__temporal_logger' && sinkName !== '__temporal_metrics') {
+        throwIfReservedName('sink', sinkName);
+      }
+    }
+  }
+
   const tuner = asNativeTuner(opts.tuner, logger);
 
   return {
diff --git a/packages/worker/src/worker.ts b/packages/worker/src/worker.ts
@@ -468,6 +468,9 @@ export class Worker {
    * This method initiates a connection to the server and will throw (asynchronously) on connection failure.
    */
   public static async create(options: WorkerOptions): Promise<Worker> {
+    if (!options.taskQueue) {
+      throw new TypeError('Task queue name is required');
+    }
     throwIfReservedName('task queue', options.taskQueue);
     const runtime = Runtime.instance();
     const logger = LoggerWithComposedMetadata.compose(runtime.logger, {
diff --git a/packages/workflow/src/internals.ts b/packages/workflow/src/internals.ts
