init impl for special behaviour for temporal prefixes. Default signal test needs to be fixed, need to add behaviour reserving prefixes from workflows, and waiting for default update to be merged to add behaviour preventing default update handler to be called with reserved names

Author: THardy98

packages/common/src/reserved.ts

@@ -0,0 +1,24 @@
+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 reservedPrefixes = [
+  TEMPORAL_RESERVED_PREFIX,
+  STACK_TRACE_RESERVED_PREFIX,
+  ENHANCED_STACK_TRACE_RESERVED_PREFIX,
+];
+
+export function throwIfReservedName(type: string, name: string): void {
+  const prefix = isReservedName(name);
+  if (prefix) {
+    throw Error(`Cannot register ${type} name: '${name}', with reserved prefix: '${prefix}'`);
+  }
+}
+
+export function isReservedName(name: string): string | undefined {
+  for (const prefix of reservedPrefixes) {
+    if (name.startsWith(prefix)) {
+      return prefix;
+    }
+  }
+}

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

@@ -14,7 +14,16 @@ import { searchAttributePayloadConverter } from '@temporalio/common/lib/converte
 import { msToNumber, tsToMs } from '@temporalio/common/lib/time';
 import { decode as payloadDecode, decodeFromPayloadsAtIndex } from '@temporalio/common/lib/internal-non-workflow';
 
-import { condition, defineQuery, defineSignal, setDefaultQueryHandler, setHandler, sleep } from '@temporalio/workflow';
+import {
+  condition,
+  defineQuery,
+  defineSignal,
+  defineUpdate,
+  setDefaultQueryHandler,
+  setHandler,
+  sleep,
+} from '@temporalio/workflow';
+import { reservedPrefixes } from '@temporalio/common/lib/reserved';
 import { configurableHelpers, createTestWorkflowBundle } from './helpers-integration';
 import * as activities from './activities';
 import * as workflows from './workflows';
@@ -763,3 +772,90 @@ test.serial('default query handler is not used if requested query exists', confi
     t.deepEqual(result, { name: definedQuery.name, args });
   });
 });
+
+test('Cannot register activities using reserved prefixes', configMacro, async (t, config) => {
+  const { createWorkerWithDefaults } = config;
+
+  for (const prefix of reservedPrefixes) {
+    const activityName = prefix + '_test';
+    await t.throwsAsync(
+      createWorkerWithDefaults(t, {
+        activities: { [activityName]: () => {} },
+      }),
+      {
+        instanceOf: Error,
+        message: `Cannot register activity name: '${activityName}', with reserved prefix: '${prefix}'`,
+      }
+    );
+  }
+});
+
+test('Cannot register task queues using reserved prefixes', configMacro, async (t, config) => {
+  const { createWorkerWithDefaults } = config;
+
+  for (const prefix of reservedPrefixes) {
+    const taskQueue = prefix + '_test';
+
+    await t.throwsAsync(
+      createWorkerWithDefaults(t, {
+        taskQueue,
+      }),
+      {
+        instanceOf: Error,
+        message: `Cannot register task queue name: '${taskQueue}', with reserved prefix: '${prefix}'`,
+      }
+    );
+  }
+});
+
+interface HandlerError {
+  name: string;
+  message: string;
+}
+
+export async function workflowBadPrefixHandler(prefix: 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'), () => {});
+  } catch (e) {
+    if (e instanceof Error) {
+      expectedErrors.push({ name: e.name, message: e.message });
+    }
+  }
+  try {
+    setHandler(defineUpdate(prefix + '_update'), () => {});
+  } catch (e) {
+    if (e instanceof Error) {
+      expectedErrors.push({ name: e.name, message: e.message });
+    }
+  }
+  try {
+    setHandler(defineQuery(prefix + '_query'), () => {});
+  } 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', configMacro, async (t, config) => {
+  const { env, createWorkerWithDefaults } = config;
+  const { executeWorkflow } = configurableHelpers(t, t.context.workflowBundle, env);
+  const worker = await createWorkerWithDefaults(t);
+  await worker.runUntil(async () => {
+    const prefix = reservedPrefixes[0];
+    // for (const prefix of reservedPrefixes) {
+    const result = await executeWorkflow(workflowBadPrefixHandler, {
+      args: [prefix],
+    });
+    console.log('result', result);
+    t.deepEqual(result, [
+      { name: 'Error', message: `Cannot register signal name: '${prefix}_signal', with reserved prefix: '${prefix}'` },
+      { name: 'Error', message: `Cannot register update name: '${prefix}_update', with reserved prefix: '${prefix}'` },
+      { name: 'Error', message: `Cannot register query name: '${prefix}_query', with reserved prefix: '${prefix}'` },
+    ]);
+    // }
+  });
+});

packages/test/src/test-workflows.ts

@@ -21,6 +21,12 @@ import { VMWorkflow, VMWorkflowCreator } from '@temporalio/worker/lib/workflow/v
 import { SdkFlag, SdkFlags } from '@temporalio/workflow/lib/flags';
 import { ReusableVMWorkflow, ReusableVMWorkflowCreator } from '@temporalio/worker/lib/workflow/reusable-vm';
 import { parseWorkflowCode } from '@temporalio/worker/lib/worker';
+import {
+  ENHANCED_STACK_TRACE_RESERVED_PREFIX,
+  reservedPrefixes,
+  STACK_TRACE_RESERVED_PREFIX,
+  TEMPORAL_RESERVED_PREFIX,
+} from '@temporalio/common/lib/reserved';
 import * as activityFunctions from './activities';
 import { cleanStackTrace, REUSE_V8_CONTEXT, u8 } from './helpers';
 import { ProcessedSignal } from './workflows';

packages/test/src/workflows/index.ts

@@ -92,3 +92,4 @@ export * from './workflow-cancellation-scenarios';
 export * from './upsert-and-read-memo';
 export * from './updates-ordering';
 export * from './wait-on-signal-then-activity';
+export * from './workflow-with-default-handlers';

packages/test/src/workflows/workflow-with-default-handlers.ts

@@ -0,0 +1,16 @@
+import {
+  condition,
+  defineSignal,
+  setDefaultQueryHandler,
+  setDefaultSignalHandler,
+  setHandler,
+} from '@temporalio/workflow';
+
+export async function workflowWithDefaultHandlers(): Promise<void> {
+  const complete = true;
+  setDefaultQueryHandler(() => {});
+  setDefaultSignalHandler(() => {});
+  setHandler(defineSignal('completeSignal'), () => {});
+
+  await condition(() => complete);
+}

packages/worker/src/worker-options.ts

@@ -14,6 +14,9 @@ import { loadDataConverter } from '@temporalio/common/lib/internal-non-workflow'
 import { LoggerSinks } from '@temporalio/workflow';
 import { Context } from '@temporalio/activity';
 import { native } from '@temporalio/core-bridge';
+import { checkExtends } from '@temporalio/common/lib/type-helpers';
+import { WorkerOptions as NativeWorkerOptions, WorkerTuner as NativeWorkerTuner } from '@temporalio/core-bridge';
+import { throwIfReservedName } from '@temporalio/common/lib/reserved';
 import { ActivityInboundLogInterceptor } from './activity-log-interceptor';
 import { NativeConnection } from './connection';
 import { CompiledWorkerInterceptors, WorkerInterceptors } from './interceptors';
@@ -953,6 +956,9 @@ export function compileWorkerOptions(
   }
 
   const activities = new Map(Object.entries(opts.activities ?? {}).filter(([_, v]) => typeof v === 'function'));
+  for (const activityName of activities.keys()) {
+    throwIfReservedName('activity', activityName);
+  }
   const tuner = asNativeTuner(opts.tuner, logger);
 
   return {

packages/worker/src/worker.ts

@@ -57,6 +57,7 @@ import { workflowLogAttributes } from '@temporalio/workflow/lib/logs';
 import { native } from '@temporalio/core-bridge';
 import { coresdk, temporal } from '@temporalio/proto';
 import { type SinkCall, type WorkflowInfo } from '@temporalio/workflow';
+import { throwIfReservedName } from '@temporalio/common/lib/reserved';
 import { Activity, CancelReason, activityLogAttributes } from './activity';
 import { extractNativeClient, extractReferenceHolders, InternalNativeConnection, NativeConnection } from './connection';
 import { ActivityExecuteInput } from './interceptors';
@@ -469,6 +470,8 @@ export class Worker {
   public static async create(options: WorkerOptions): Promise<Worker> {
     const runtime = Runtime.instance();
     const logger = LoggerWithComposedMetadata.compose(runtime.logger, {
+    throwIfReservedName('task queue', options.taskQueue);
+    const logger = withMetadata(Runtime.instance().logger, {
       sdkComponent: SdkComponent.worker,
       taskQueue: options.taskQueue ?? 'default',
     });

packages/workflow/src/internals.ts

@@ -31,6 +31,12 @@ import {
 import { composeInterceptors } from '@temporalio/common/lib/interceptors';
 import { makeProtoEnumConverters } from '@temporalio/common/lib/internal-workflow';
 import type { coresdk, temporal } from '@temporalio/proto';
+import {
+  ENHANCED_STACK_TRACE_RESERVED_PREFIX,
+  STACK_TRACE_RESERVED_PREFIX,
+  isReservedName,
+  throwIfReservedName,
+} from '@temporalio/common/lib/reserved';
 import { alea, RNG } from './alea';
 import { RootCancellationScope } from './cancellation-scope';
 import { UpdateScope } from './update-scope';
@@ -260,7 +266,7 @@ export class Activator implements ActivationHandler {
    */
   public readonly queryHandlers = new Map<string, WorkflowQueryAnnotatedType>([
     [
-      '__stack_trace',
+      STACK_TRACE_RESERVED_PREFIX,
       {
         handler: () => {
           return this.getStackTraces()
@@ -271,7 +277,7 @@ export class Activator implements ActivationHandler {
       },
     ],
     [
-      '__enhanced_stack_trace',
+      ENHANCED_STACK_TRACE_RESERVED_PREFIX,
       {
         handler: (): EnhancedStackTrace => {
           const { sourceMap } = this;
@@ -679,17 +685,28 @@ 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 execute = composeInterceptors(
       this.interceptors.inbound,
       'handleQuery',
       this.queryWorkflowNextHandler.bind(this)
     );
-    execute({
-      queryName: queryType,
-      args: arrayFromPayloads(this.payloadConverter, activation.arguments),
-      queryId,
-      headers: headers ?? {},
-    }).then(
+    execute(queryInput).then(
       (result) => this.completeQuery(queryId, result),
       (reason) => this.failQuery(queryId, reason)
     );
@@ -847,6 +864,8 @@ 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}`);

packages/workflow/src/workflow.ts

@@ -33,6 +33,7 @@ import { versioningIntentToProto } from '@temporalio/common/lib/versioning-inten
 import { Duration, msOptionalToTs, msToNumber, msToTs, requiredTsToMs } from '@temporalio/common/lib/time';
 import { composeInterceptors } from '@temporalio/common/lib/interceptors';
 import { temporal } from '@temporalio/proto';
+import { throwIfReservedName } from '@temporalio/common/lib/reserved';
 import { CancellationScope, registerSleepImplementation } from './cancellation-scope';
 import { UpdateScope } from './update-scope';
 import {
@@ -1272,6 +1273,8 @@ export function setHandler<
   options?: QueryHandlerOptions | SignalHandlerOptions | UpdateHandlerOptions<Args>
 ): void {
   const activator = assertInWorkflowContext('Workflow.setHandler(...) may only be used from a Workflow Execution.');
+  // Cannot register handler for reserved names
+  throwIfReservedName(def.type, def.name);
   const description = options?.description;
   if (def.type === 'update') {
     if (typeof handler === 'function') {