Add default update handler

packages/test/src/test-workflows.ts

@@ -363,6 +363,40 @@ function makeSetPatchMarker(myPatchId: string, deprecated: boolean): coresdk.wor
   };
 }
 
+function makeUpdateActivationJob(
+  id: string,
+  protocolInstanceId: string,
+  name: string,
+  input: unknown
+): coresdk.workflow_activation.IWorkflowActivationJob {
+  return {
+    doUpdate: {
+      id,
+      protocolInstanceId,
+      name,
+      input: toPayloads(defaultPayloadConverter, input),
+    },
+  };
+}
+
+function makeUpdateAcceptedResponse(id: string): coresdk.workflow_commands.IWorkflowCommand {
+  return {
+    updateResponse: {
+      protocolInstanceId: id,
+      accepted: {},
+    },
+  };
+}
+
+function makeUpdateCompleteResponse(id: string, result: unknown): coresdk.workflow_commands.IWorkflowCommand {
+  return {
+    updateResponse: {
+      protocolInstanceId: id,
+      completed: defaultPayloadConverter.toPayload(result),
+    },
+  };
+}
+
 test('random', async (t) => {
   const { logs, workflowType } = t.context;
   {
@@ -2402,23 +2436,9 @@ test('Signals/Updates/Activities/Timers - Trace promises completion order - pre-
       ...makeActivation(
         undefined,
         makeSignalWorkflowJob('aaSignal', ['signal1']),
-        {
-          doUpdate: {
-            id: 'first',
-            name: 'aaUpdate',
-            protocolInstanceId: '1',
-            input: toPayloads(defaultPayloadConverter, ['update1']),
-          },
-        },
+        makeUpdateActivationJob('first', '1', 'aaUpdate', ['update1']),
         makeSignalWorkflowJob('aaSignal', ['signal2']),
-        {
-          doUpdate: {
-            id: 'second',
-            name: 'aaUpdate',
-            protocolInstanceId: '2',
-            input: toPayloads(defaultPayloadConverter, ['update2']),
-          },
-        },
+        makeUpdateActivationJob('second', '2', 'aaUpdate', ['update2']),
         makeFireTimerJob(1),
         makeResolveActivityJob(1, { completed: {} })
       ),
@@ -2481,23 +2501,9 @@ test('Signals/Updates/Activities/Timers - Trace promises completion order - 1.11
       ...makeActivation(
         undefined,
         makeSignalWorkflowJob('aaSignal', ['signal1']),
-        {
-          doUpdate: {
-            id: 'first',
-            name: 'aaUpdate',
-            protocolInstanceId: '1',
-            input: toPayloads(defaultPayloadConverter, ['update1']),
-          },
-        },
+        makeUpdateActivationJob('first', '1', 'aaUpdate', ['update1']),
         makeSignalWorkflowJob('aaSignal', ['signal2']),
-        {
-          doUpdate: {
-            id: 'second',
-            name: 'aaUpdate',
-            protocolInstanceId: '2',
-            input: toPayloads(defaultPayloadConverter, ['update2']),
-          },
-        },
+        makeUpdateActivationJob('second', '2', 'aaUpdate', ['update2']),
         makeFireTimerJob(1),
         makeResolveActivityJob(1, { completed: {} })
       ),
@@ -2528,3 +2534,138 @@ test('Signals/Updates/Activities/Timers - Trace promises completion order - 1.11
     );
   }
 });
+
+test('Buffered updates are dispatched in the correct order - updatesOrdering', async (t) => {
+  const { workflowType } = t.context;
+  {
+    const completion = await activate(
+      t,
+      makeActivation(
+        undefined,
+        makeInitializeWorkflowJob(workflowType),
+        makeUpdateActivationJob('1', '1', 'non-existant', 1),
+        makeUpdateActivationJob('2', '2', 'updateA', 2),
+        makeUpdateActivationJob('3', '3', 'updateA', 3),
+        makeUpdateActivationJob('4', '4', 'updateC', 4),
+        makeUpdateActivationJob('5', '5', 'updateB', 5),
+        makeUpdateActivationJob('6', '6', 'non-existant', 6),
+        makeUpdateActivationJob('7', '7', 'updateB', 7)
+      )
+    );
+
+    // The activation above:
+    // - initializes the workflow
+    // - buffers all its updates (we attempt update jobs first, but since there are no handlers, they get buffered)
+    // - enters the workflow code
+    // - workflow code sets handler for updateA
+    //   - handler is registered for updateA
+    //   - we attempt to dispatch buffered updates
+    //     - buffered updates for handler A are *accepted* but not executed
+    //    (executing an update is a promise/async, so it instead goes on the node event queue)
+    // - we continue/re-enter the workflow code
+    // - ...and do the same pattern for updateB, the default update handler, the updateC
+    // - once updates have been accepted, node processes the waiting events in its queue (the waiting updates)
+    //   - these are processesed in FIFO order, so we get execution for updateA, then updateB, the default handler, then updateC
+
+    // As such, the expected order of these updates is the order that the handlers were registered.
+    // Note that because the default handler was registered *before* updateC, all remaining buffered updates were dispatched
+    // to it, including the update for updateC.
+
+    compareCompletion(
+      t,
+      completion,
+      makeSuccess(
+        [
+          // FIFO accepted order
+          makeUpdateAcceptedResponse('2'),
+          makeUpdateAcceptedResponse('3'),
+          makeUpdateAcceptedResponse('5'),
+          makeUpdateAcceptedResponse('7'),
+          makeUpdateAcceptedResponse('1'),
+          makeUpdateAcceptedResponse('4'),
+          makeUpdateAcceptedResponse('6'),
+          // FIFO executed order
+          makeUpdateCompleteResponse('2', { handler: 'updateA', args: [2] }),
+          makeUpdateCompleteResponse('3', { handler: 'updateA', args: [3] }),
+          makeUpdateCompleteResponse('5', { handler: 'updateB', args: [5] }),
+          makeUpdateCompleteResponse('7', { handler: 'updateB', args: [7] }),
+          makeUpdateCompleteResponse('1', { handler: 'default', updateName: 'non-existant', args: [1] }),
+          // updateC handled by default handler.
+          makeUpdateCompleteResponse('4', { handler: 'default', updateName: 'updateC', args: [4] }),
+          makeUpdateCompleteResponse('6', { handler: 'default', updateName: 'non-existant', args: [6] }),
+          // No expected update response from updateC handler
+          makeCompleteWorkflowExecution(),
+        ]
+        // [SdkFlags.ProcessWorkflowActivationJobsAsSingleBatch]
+      )
+    );
+  }
+});
+
+test('Buffered updates are reentrant - updatesAreReentrant', async (t) => {
+  const { workflowType } = t.context;
+  {
+    const completion = await activate(
+      t,
+      makeActivation(
+        undefined,
+        makeInitializeWorkflowJob(workflowType),
+        makeUpdateActivationJob('1', '1', 'non-existant', 1),
+        makeUpdateActivationJob('2', '2', 'updateA', 2),
+        makeUpdateActivationJob('3', '3', 'updateA', 3),
+        makeUpdateActivationJob('4', '4', 'updateC', 4),
+        makeUpdateActivationJob('5', '5', 'updateB', 5),
+        makeUpdateActivationJob('6', '6', 'non-existant', 6),
+        makeUpdateActivationJob('7', '7', 'updateB', 7),
+        makeUpdateActivationJob('8', '8', 'updateC', 8)
+      )
+    );
+
+    // The activation above:
+    // - initializes the workflow
+    // - buffers all its updates (we attempt update jobs first, but since there are no handlers, they get buffered)
+    // - enters the workflow code
+    // - workflow code sets handler for updateA
+    //   - handler is registered for updateA
+    //   - we attempt to dispatch buffered updates
+    //     - buffered updates for handler A are *accepted* but not executed
+    //    (executing an update is a promise/async, so it instead goes on the node event queue)
+    //  - however, there is no more workflow code, node dequues event queue and we immediately run the update handler
+    //    (we begin executing the update which...)
+    //    - deletes the current handler and registers the next one (updateB)
+    //  - this pattern repeats (updateA -> updateB -> updateC -> default) until there are no more updates to handle
+    //  - at this point, all updates have been accepted and are executing
+    //  - due to the call order in the workflow, the completion order of the updates follows the call stack, LIFO
+
+    // This workflow is interesting in that updates are accepted FIFO, but executed LIFO
+
+    compareCompletion(
+      t,
+      completion,
+      makeSuccess(
+        [
+          // FIFO accepted order
+          makeUpdateAcceptedResponse('2'),
+          makeUpdateAcceptedResponse('5'),
+          makeUpdateAcceptedResponse('4'),
+          makeUpdateAcceptedResponse('1'),
+          makeUpdateAcceptedResponse('3'),
+          makeUpdateAcceptedResponse('7'),
+          makeUpdateAcceptedResponse('8'),
+          makeUpdateAcceptedResponse('6'),
+          // LIFO executed order
+          makeUpdateCompleteResponse('6', { handler: 'default', updateName: 'non-existant', args: [6] }),
+          makeUpdateCompleteResponse('8', { handler: 'updateC', args: [8] }),
+          makeUpdateCompleteResponse('7', { handler: 'updateB', args: [7] }),
+          makeUpdateCompleteResponse('3', { handler: 'updateA', args: [3] }),
+          makeUpdateCompleteResponse('1', { handler: 'default', updateName: 'non-existant', args: [1] }),
+          makeUpdateCompleteResponse('4', { handler: 'updateC', args: [4] }),
+          makeUpdateCompleteResponse('5', { handler: 'updateB', args: [5] }),
+          makeUpdateCompleteResponse('2', { handler: 'updateA', args: [2] }),
+          makeCompleteWorkflowExecution(),
+        ]
+        // [SdkFlags.ProcessWorkflowActivationJobsAsSingleBatch]
+      )
+    );
+  }
+});

packages/test/src/workflows/index.ts

@@ -89,3 +89,4 @@ export * from './upsert-and-read-search-attributes';
 export * from './wait-on-user';
 export * from './workflow-cancellation-scenarios';
 export * from './upsert-and-read-memo';
+export * from './updates-ordering';

packages/test/src/workflows/updates-ordering.ts

@@ -0,0 +1,77 @@
+import { defineUpdate, setDefaultUpdateHandler, setHandler } from '@temporalio/workflow';
+
+const updateA = defineUpdate<ProcessedUpdate, [number]>('updateA');
+const updateB = defineUpdate<ProcessedUpdate, [number]>('updateB');
+const updateC = defineUpdate<ProcessedUpdate, [number]>('updateC');
+
+interface ProcessedUpdate {
+  handler: string;
+  updateName?: string;
+  args: unknown[];
+}
+
+/*
+    There's a surprising amount going on with the workflow below. Let's simplify it to just updateA and updateB 
+    (no updateC or the default) and walk through it.
+
+    1. setHandler for updateA
+    - this is all synchronous code until we run `UpdateScope.updateWithInfo(updateId, name, doUpdateImpl)`, 
+      which calls `doUpdateImpl` which is promise/async, so...
+    2. queue doUpdateImpl for A on node event queue: [doUpdateImplA]
+    3. continue running the workflow code (currently running code, we aren't awaiting the promise)
+    4. setHandler for updateB
+    - same deal as A
+    5. queue doUpdateImpl for B on node event queue: [doUpdateImplA, doUpdateImplB]
+    6. finished workflow code, go through the event queue
+    7. doUpdateImplA
+    - synchronous until we get to `execute`, which means we've accepted the update, command ordering [acceptA]
+    8. `execute` returns a promise, add it to the node event queue: [doUpdateImplB, executeA]
+    9. doUpdateImplB
+    - same deal as A, command ordering [acceptA, acceptB]
+    - `execute` returns promise, node event queue [executeA, executeB]
+    10. execute update A, node event queue [executeB], command ordering [acceptA, acceptB, executeA]
+    11. execute update B, node event queue [] (empty), command ordering [acceptA, acceptB, executeA, executeB]
+
+    The only additional complexity with the workflow below is that once the default handler is registered, buffered updates for C will be
+    dispatched to the default handler. So in this scenario: C1, C2 -> default registered -> C registered, both C1 and C2 will be dispatched
+    to the default handler.
+*/
+export async function updatesOrdering(): Promise<void> {
+  setHandler(updateA, (...args: any[]) => {
+    return { handler: 'updateA', args };
+  });
+  setHandler(updateB, (...args: any[]) => {
+    return { handler: 'updateB', args };
+  });
+  setDefaultUpdateHandler((updateName, ...args: any[]) => {
+    return { handler: 'default', updateName, args };
+  });
+  setHandler(updateC, (...args: any[]) => {
+    return { handler: 'updateC', args };
+  });
+}
+
+export async function updatesAreReentrant(): Promise<void> {
+  function handlerA(...args: any[]) {
+    setHandler(updateA, undefined);
+    setHandler(updateB, handlerB);
+    return { handler: 'updateA', args };
+  }
+  function handlerB(...args: any[]) {
+    setHandler(updateB, undefined);
+    setHandler(updateC, handlerC);
+    return { handler: 'updateB', args };
+  }
+  function handlerC(...args: any[]) {
+    setHandler(updateC, undefined);
+    setDefaultUpdateHandler(defaultHandler);
+    return { handler: 'updateC', args };
+  }
+  function defaultHandler(updateName: string, ...args: any[]) {
+    setDefaultUpdateHandler(undefined);
+    setHandler(updateA, handlerA);
+    return { handler: 'default', updateName, args };
+  }
+
+  setHandler(updateA, handlerA);
+}

packages/workflow/src/interfaces.ts

@@ -538,6 +538,11 @@ export type Handler<
  */
 export type DefaultSignalHandler = (signalName: string, ...args: unknown[]) => void | Promise<void>;
 
+/**
+ * A handler function accepting update calls for non-registered update names.
+ */
+export type DefaultUpdateHandler = (updateName: string, ...args: any[]) => Promise<any> | any;
+
 /**
  * A validation function capable of accepting the arguments for a given UpdateDefinition.
  */

packages/workflow/src/internals.ts

@@ -41,6 +41,7 @@ import {
   WorkflowInfo,
   WorkflowCreateOptionsInternal,
   ActivationCompletion,
+  DefaultUpdateHandler,
 } from './interfaces';
 import { type SinkCall } from './sinks';
 import { untrackPromise } from './stack-helpers';
@@ -189,6 +190,11 @@ export class Activator implements ActivationHandler {
    */
   defaultSignalHandler?: DefaultSignalHandler;
 
+  /**
+   * A update handler that catches calls for non-registered update names.
+   */
+  defaultUpdateHandler?: DefaultUpdateHandler;
+
   /**
    * Source map file for looking up the source files in response to __enhanced_stack_trace
    */
@@ -665,12 +671,21 @@ export class Activator implements ActivationHandler {
     if (!protocolInstanceId) {
       throw new TypeError('Missing activation update protocolInstanceId');
     }
-    const entry = this.updateHandlers.get(name);
-    if (!entry) {
+    if (!this.updateHandlers.get(name) && !this.defaultUpdateHandler) {
       this.bufferedUpdates.push(activation);
       return;
     }
 
+    const entry = this.updateHandlers.get(name) ?? {
+      // Logically, this must be defined as we got passed the conditional above
+      // pushing to the buffer. But Typescript doesn't know that so we use a
+      // non-null assertion (!).
+      handler: (...args: any[]) => this.defaultUpdateHandler!(name, ...args),
+      validator: undefined,
+      // Default to a warning policy.
+      unfinishedPolicy: HandlerUnfinishedPolicy.WARN_AND_ABANDON,
+    };
+
     const makeInput = (): UpdateInput => ({
       updateId,
       args: arrayFromPayloads(this.payloadConverter, activation.input),
@@ -758,13 +773,21 @@ export class Activator implements ActivationHandler {
   public dispatchBufferedUpdates(): void {
     const bufferedUpdates = this.bufferedUpdates;
     while (bufferedUpdates.length) {
-      const foundIndex = bufferedUpdates.findIndex((update) => this.updateHandlers.has(update.name as string));
-      if (foundIndex === -1) {
-        // No buffered Updates have a handler yet.
-        break;
+      // We have a default update handler, so all updates are dispatchable.
+      if (this.defaultUpdateHandler) {
+        const update = bufferedUpdates.shift();
+        // Logically, this must be defined as we're in the loop.
+        // But Typescript doesn't know that so we use a non-null assertion (!).
+        this.doUpdate(update!);
+      } else {
+        const foundIndex = bufferedUpdates.findIndex((update) => this.updateHandlers.has(update.name as string));
+        if (foundIndex === -1) {
+          // No buffered Updates have a handler yet.
+          break;
+        }
+        const [update] = bufferedUpdates.splice(foundIndex, 1);
+        this.doUpdate(update);
       }
-      const [update] = bufferedUpdates.splice(foundIndex, 1);
-      this.doUpdate(update);
     }
   }