Special behaviour for temporal prefixes (#1644)

Co-authored-by: James Watkins-Harvey <mjameswh@users.noreply.github.com>
Co-authored-by: James Watkins-Harvey <james.watkinsharvey@temporal.io>

Author: THardy98

packages/common/src/reserved.ts

@@ -0,0 +1,26 @@
+export const TEMPORAL_RESERVED_PREFIX = '__temporal_';
+export const STACK_TRACE_QUERY_NAME = '__stack_trace';
+export const ENHANCED_STACK_TRACE_QUERY_NAME = '__enhanced_stack_trace';
+
+/**
+ * Valid entity types that can be checked for reserved name violations
+ */
+export type ReservedNameEntityType = 'query' | 'signal' | 'update' | 'activity' | 'task queue' | 'sink' | 'workflow';
+
+/**
+ * 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}'`);
+  }
+
+  if (name === STACK_TRACE_QUERY_NAME || name === ENHANCED_STACK_TRACE_QUERY_NAME) {
+    throw new TypeError(`Cannot use ${type} name: '${name}', which is a reserved name`);
+  }
+}

packages/test/src/helpers.ts

@@ -293,21 +293,6 @@ export async function getRandomPort(fn = (_port: number) => Promise.resolve()):
   });
 }
 
-export function asSdkLoggerSink(
-  fn: (info: WorkflowInfo, message: string, attrs?: Record<string, unknown>) => Promise<void>,
-  opts?: Omit<worker.InjectedSinkFunction<any>, 'fn'>
-): worker.InjectedSinks<DefaultLoggerSinks> {
-  return {
-    __temporal_logger: {
-      trace: { fn, ...opts },
-      debug: { fn, ...opts },
-      info: { fn, ...opts },
-      warn: { fn, ...opts },
-      error: { fn, ...opts },
-    },
-  };
-}
-
 export async function loadHistory(fname: string): Promise<iface.temporal.api.history.v1.History> {
   const isJson = fname.endsWith('json');
   const fpath = path.resolve(__dirname, `../history_files/${fname}`);

packages/test/src/test-integration-workflows.ts

@@ -1,14 +1,24 @@
 import { setTimeout as setTimeoutPromise } from 'timers/promises';
 import { randomUUID } from 'crypto';
+import asyncRetry from 'async-retry';
 import { ExecutionContext } from 'ava';
 import { firstValueFrom, Subject } from 'rxjs';
-import { WorkflowFailedError } from '@temporalio/client';
+import { WorkflowFailedError, WorkflowHandle } from '@temporalio/client';
 import * as activity from '@temporalio/activity';
 import { msToNumber, tsToMs } from '@temporalio/common/lib/time';
 import { TestWorkflowEnvironment } from '@temporalio/testing';
 import { CancelReason } from '@temporalio/worker/lib/activity';
 import * as workflow from '@temporalio/workflow';
-import { defineQuery, defineSignal } from '@temporalio/workflow';
+import {
+  condition,
+  defineQuery,
+  defineSignal,
+  defineUpdate,
+  setDefaultQueryHandler,
+  setDefaultSignalHandler,
+  setDefaultUpdateHandler,
+  setHandler,
+} from '@temporalio/workflow';
 import { SdkFlags } from '@temporalio/workflow/lib/flags';
 import {
   ActivityCancellationType,
@@ -20,12 +30,17 @@ import {
   TypedSearchAttributes,
   WorkflowExecutionAlreadyStartedError,
 } from '@temporalio/common';
+import {
+  TEMPORAL_RESERVED_PREFIX,
+  STACK_TRACE_QUERY_NAME,
+  ENHANCED_STACK_TRACE_QUERY_NAME,
+} from '@temporalio/common/lib/reserved';
 import { signalSchedulingWorkflow } from './activities/helpers';
 import { activityStartedSignal } from './workflows/definitions';
 import * as workflows from './workflows';
 import { Context, createLocalTestEnvironment, helpers, makeTestFunction } from './helpers-integration';
 import { overrideSdkInternalFlag } from './mock-internal-flags';
-import { asSdkLoggerSink, loadHistory, RUN_TIME_SKIPPING_TESTS, waitUntil } from './helpers';
+import { loadHistory, RUN_TIME_SKIPPING_TESTS, waitUntil } from './helpers';
 
 const test = makeTestFunction({
   workflowsPath: __filename,
@@ -1126,53 +1141,6 @@ test('Workflow can upsert memo', async (t) => {
   });
 });
 
-test('Sink functions contains upserted memo', async (t) => {
-  const { createWorker, executeWorkflow } = helpers(t);
-  const recordedMessages = Array<{ message: string; memo: Record<string, unknown> | undefined }>();
-  const sinks = asSdkLoggerSink(async (info, message, _attrs) => {
-    recordedMessages.push({
-      message,
-      memo: info.memo,
-    });
-  });
-  const worker = await createWorker({ sinks });
-  await worker.runUntil(async () => {
-    await executeWorkflow(upsertAndReadMemo, {
-      memo: {
-        note1: 'aaa',
-        note2: 'bbb',
-        note4: 'eee',
-      },
-      args: [
-        {
-          note2: 'ccc',
-          note3: 'ddd',
-          note4: null,
-        },
-      ],
-    });
-  });
-
-  t.deepEqual(recordedMessages, [
-    {
-      message: 'Workflow started',
-      memo: {
-        note1: 'aaa',
-        note2: 'bbb',
-        note4: 'eee',
-      },
-    },
-    {
-      message: 'Workflow completed',
-      memo: {
-        note1: 'aaa',
-        note2: 'ccc',
-        note3: 'ddd',
-      },
-    },
-  ]);
-});
-
 export async function langFlagsReplayCorrectly(): Promise<void> {
   const { noopActivity } = workflow.proxyActivities({ scheduleToCloseTimeout: '10s' });
   await workflow.CancellationScope.withTimeout('10s', async () => {
@@ -1440,3 +1408,232 @@ test('Workflow can return root workflow', async (t) => {
     t.deepEqual(result, 'empty test-root-workflow-length');
   });
 });
+
+const reservedNames = [TEMPORAL_RESERVED_PREFIX, STACK_TRACE_QUERY_NAME, ENHANCED_STACK_TRACE_QUERY_NAME];
+
+test('Cannot register activities using reserved prefixes', async (t) => {
+  const { createWorker } = helpers(t);
+
+  for (const name of reservedNames) {
+    const activityName = name === TEMPORAL_RESERVED_PREFIX ? name + '_test' : name;
+    await t.throwsAsync(
+      createWorker({
+        activities: { [activityName]: () => {} },
+      }),
+      {
+        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`,
+      }
+    );
+  }
+});
+
+test('Cannot register task queues using reserved prefixes', async (t) => {
+  const { createWorker } = helpers(t);
+
+  for (const name of reservedNames) {
+    const taskQueue = name === TEMPORAL_RESERVED_PREFIX ? name + '_test' : name;
+
+    await t.throwsAsync(
+      createWorker({
+        taskQueue,
+      }),
+      {
+        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`,
+      }
+    );
+  }
+});
+
+interface HandlerError {
+  name: string;
+  message: string;
+}
+
+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(name === TEMPORAL_RESERVED_PREFIX ? name + '_signal' : name), () => {});
+  } catch (e) {
+    if (e instanceof Error) {
+      expectedErrors.push({ name: e.name, message: e.message });
+    }
+  }
+  try {
+    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(name === TEMPORAL_RESERVED_PREFIX ? name + '_query' : name), () => {});
+  } catch (e) {
+    if (e instanceof Error) {
+      expectedErrors.push({ name: e.name, message: e.message });
+    }
+  }
+  return expectedErrors;
+}
+
+test('Workflow failure if define signals/updates/queries with reserved prefixes', async (t) => {
+  const { createWorker, executeWorkflow } = helpers(t);
+  const worker = await createWorker();
+  await worker.runUntil(async () => {
+    for (const name of reservedNames) {
+      const result = await executeWorkflow(workflowReservedNameHandler, {
+        args: [name],
+      });
+      t.deepEqual(result, [
+        {
+          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: '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: '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`,
+        },
+      ]);
+    }
+  });
+});
+
+export const wfReadyQuery = defineQuery<boolean>('wf-ready');
+export async function workflowWithDefaultHandlers(): Promise<void> {
+  let unblocked = false;
+  setHandler(defineSignal('unblock'), () => {
+    unblocked = true;
+  });
+
+  setDefaultQueryHandler(() => {});
+  setDefaultSignalHandler(() => {});
+  setDefaultUpdateHandler(() => {});
+  setHandler(wfReadyQuery, () => true);
+
+  await condition(() => unblocked);
+}
+
+test('Default handlers fail given reserved prefix', async (t) => {
+  const { createWorker, startWorkflow } = helpers(t);
+  const worker = await createWorker();
+
+  const assertWftFailure = async (handle: WorkflowHandle, errMsg: string) => {
+    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, errMsg);
+      },
+      { minTimeout: 300, factor: 1, retries: 10 }
+    );
+  };
+
+  await worker.runUntil(async () => {
+    // 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();
+  });
+});