Add requestEagerStart

- add startWithDetails to WorkflowClientInterceptor to support return type of WorkflowStartOutput, inject if not defined at _start call

Author: THardy98

packages/client/src/interceptors.ts

@@ -35,6 +35,12 @@ export interface WorkflowStartUpdateInput {
   readonly options: WorkflowUpdateOptions;
 }
 
+/** Output for WorkflowClientInterceptor.startWithDetails */
+export interface WorkflowStartOutput {
+  readonly runId: string;
+  readonly eagerlyStarted: boolean;
+}
+
 /** Output for WorkflowClientInterceptor.startUpdate */
 export interface WorkflowStartUpdateOutput {
   readonly updateId: string;
@@ -120,6 +126,18 @@ export interface WorkflowClientInterceptor {
    * {@link signalWithStart} most likely needs to be implemented too
    */
   start?: (input: WorkflowStartInput, next: Next<this, 'start'>) => Promise<string /* runId */>;
+
+  /**
+   * Intercept a service call to startWorkflowExecution
+   *
+   * Successor to {@link start}. Unlike {@link start}, this method returns
+   * start details via {@link WorkflowStartOutput}.
+   *
+   * If you implement this method,
+   * {@link signalWithStart} most likely needs to be implemented too
+   */
+  startWithDetails?: (input: WorkflowStartInput, next: Next<this, 'startWithDetails'>) => Promise<WorkflowStartOutput>;
+
   /**
    * Intercept a service call to updateWorkflowExecution
    */

packages/client/src/types.ts

@@ -162,6 +162,12 @@ export interface ConnectionLike {
    * @see https://developer.mozilla.org/en-US/docs/Web/API/AbortSignal
    */
   withAbortSignal<R>(abortSignal: AbortSignal, fn: () => Promise<R>): Promise<R>;
+
+  /**
+   * Capability flag that determines whether the connection supports eager workflow start.
+   * This will only be true if the underlying connection is a {@link NativeConnection}.
+   */
+  readonly supportsEagerStart?: boolean;
 }
 
 export const QueryRejectCondition = {

packages/client/src/workflow-client.ts

@@ -61,6 +61,7 @@ import {
   WorkflowStartUpdateOutput,
   WorkflowStartUpdateWithStartInput,
   WorkflowStartUpdateWithStartOutput,
+  WorkflowStartOutput,
 } from './interceptors';
 import {
   CountWorkflowExecution,
@@ -263,6 +264,15 @@ export interface WorkflowHandleWithFirstExecutionRunId<T extends Workflow = Work
   readonly firstExecutionRunId: string;
 }
 
+/**
+ * This interface is exactly the same as {@link WorkflowHandleWithFirstExecutionRunId} except it
+ * includes the `eagerlyStarted` returned from {@link WorkflowClient.start}.
+ */
+export interface WorkflowHandleWithStartDetails<T extends Workflow = Workflow>
+  extends WorkflowHandleWithFirstExecutionRunId<T> {
+  readonly eagerlyStarted: boolean;
+}
+
 /**
  * This interface is exactly the same as {@link WorkflowHandle} except it
  * includes the `signaledRunId` returned from `signalWithStart`.
@@ -514,14 +524,22 @@ export class WorkflowClient extends BaseClient {
     workflowTypeOrFunc: string | T,
     options: WorkflowStartOptions<T>,
     interceptors: WorkflowClientInterceptor[]
-  ): Promise<string> {
+  ): Promise<WorkflowStartOutput> {
     const workflowType = extractWorkflowType(workflowTypeOrFunc);
     assertRequiredWorkflowOptions(options);
     const compiledOptions = compileWorkflowOptions(ensureArgs(options));
 
-    const start = composeInterceptors(interceptors, 'start', this._startWorkflowHandler.bind(this));
+    const adaptedInterceptors: WorkflowClientInterceptor[] = interceptors.map((i) =>
+      i.startWithDetails ? i : { ...i, startWithDetails: (input, next) => next(input) }
+    );
+
+    const startWithDetails = composeInterceptors(
+      adaptedInterceptors,
+      'startWithDetails',
+      this._startWorkflowHandler.bind(this)
+    );
 
-    return start({
+    return startWithDetails({
       options: compiledOptions,
       headers: {},
       workflowType,
@@ -537,7 +555,6 @@ export class WorkflowClient extends BaseClient {
     const { signal, signalArgs, ...rest } = options;
     assertRequiredWorkflowOptions(rest);
     const compiledOptions = compileWorkflowOptions(ensureArgs(rest));
-
     const signalWithStart = composeInterceptors(
       interceptors,
       'signalWithStart',
@@ -561,22 +578,25 @@ export class WorkflowClient extends BaseClient {
   public async start<T extends Workflow>(
     workflowTypeOrFunc: string | T,
     options: WorkflowStartOptions<T>
-  ): Promise<WorkflowHandleWithFirstExecutionRunId<T>> {
+  ): Promise<WorkflowHandleWithStartDetails<T>> {
     const { workflowId } = options;
     const interceptors = this.getOrMakeInterceptors(workflowId);
-    const runId = await this._start(workflowTypeOrFunc, { ...options, workflowId }, interceptors);
+    const wfStartOutput = await this._start(workflowTypeOrFunc, { ...options, workflowId }, interceptors);
     // runId is not used in handles created with `start*` calls because these
     // handles should allow interacting with the workflow if it continues as new.
-    const handle = this._createWorkflowHandle({
+    const baseHandle = this._createWorkflowHandle({
       workflowId,
       runId: undefined,
-      firstExecutionRunId: runId,
-      runIdForResult: runId,
+      firstExecutionRunId: wfStartOutput.runId,
+      runIdForResult: wfStartOutput.runId,
       interceptors,
       followRuns: options.followRuns ?? true,
-    }) as WorkflowHandleWithFirstExecutionRunId<T>; // Cast is safe because we know we add the firstExecutionRunId below
-    (handle as any) /* readonly */.firstExecutionRunId = runId;
-    return handle;
+    });
+    return {
+      ...baseHandle,
+      firstExecutionRunId: wfStartOutput.runId,
+      eagerlyStarted: wfStartOutput.eagerlyStarted,
+    };
   }
 
   /**
@@ -1246,7 +1266,7 @@ export class WorkflowClient extends BaseClient {
    *
    * Used as the final function of the start interceptor chain
    */
-  protected async _startWorkflowHandler(input: WorkflowStartInput): Promise<string> {
+  protected async _startWorkflowHandler(input: WorkflowStartInput): Promise<WorkflowStartOutput> {
     const req = await this.createStartWorkflowRequest(input);
     const { options: opts, workflowType } = input;
     const internalOptions = (opts as InternalWorkflowStartOptions)[InternalWorkflowStartOptionsSymbol];
@@ -1255,7 +1275,10 @@ export class WorkflowClient extends BaseClient {
       if (internalOptions != null) {
         internalOptions.backLink = response.link ?? undefined;
       }
-      return response.runId;
+      return {
+        runId: response.runId,
+        eagerlyStarted: response.eagerWorkflowTask != null,
+      };
     } catch (err: any) {
       if (err.code === grpcStatus.ALREADY_EXISTS) {
         throw new WorkflowExecutionAlreadyStartedError(
@@ -1273,6 +1296,13 @@ export class WorkflowClient extends BaseClient {
     const { identity, namespace } = this.options;
     const internalOptions = (opts as InternalWorkflowStartOptions)[InternalWorkflowStartOptionsSymbol];
 
+    if (opts.requestEagerStart && !('supportsEagerStart' in this.connection && this.connection.supportsEagerStart)) {
+      throw new Error(
+        'Eager workflow start requires a NativeConnection shared between client and worker. ' +
+          'Pass a NativeConnection via ClientOptions.connection, or disable requestEagerStart.'
+      );
+    }
+
     return {
       namespace,
       identity,
@@ -1303,6 +1333,7 @@ export class WorkflowClient extends BaseClient {
       userMetadata: await encodeUserMetadata(this.dataConverter, opts.staticSummary, opts.staticDetails),
       priority: opts.priority ? compilePriority(opts.priority) : undefined,
       versioningOverride: opts.versioningOverride ?? undefined,
+      requestEagerExecution: opts.requestEagerStart,
       ...filterNullAndUndefined(internalOptions ?? {}),
     };
   }

packages/client/src/workflow-options.ts

@@ -54,6 +54,12 @@ export interface WorkflowOptions extends CommonWorkflowOptions {
    * @experimental Deployment based versioning is experimental and may change in the future.
    */
   versioningOverride?: VersioningOverride;
+
+  /**
+   * Potentially reduce the latency to start this workflow by encouraging the server to
+   * start it on a local worker running with this same client.
+   */
+  requestEagerStart?: boolean;
 }
 
 export type WithCompiledWorkflowOptions<T extends WorkflowOptions> = Replace<
@@ -97,7 +103,8 @@ export type WorkflowSignalWithStartOptions<SignalArgs extends any[] = []> = Sign
   ? WorkflowSignalWithStartOptionsWithArgs<SignalArgs>
   : WorkflowSignalWithStartOptionsWithoutArgs<SignalArgs>;
 
-export interface WorkflowSignalWithStartOptionsWithoutArgs<SignalArgs extends any[]> extends WorkflowOptions {
+export interface WorkflowSignalWithStartOptionsWithoutArgs<SignalArgs extends any[]>
+  extends Omit<WorkflowOptions, 'requestEagerStart'> {
   /**
    * SignalDefinition or name of signal
    */
@@ -109,7 +116,8 @@ export interface WorkflowSignalWithStartOptionsWithoutArgs<SignalArgs extends an
   signalArgs?: SignalArgs;
 }
 
-export interface WorkflowSignalWithStartOptionsWithArgs<SignalArgs extends any[]> extends WorkflowOptions {
+export interface WorkflowSignalWithStartOptionsWithArgs<SignalArgs extends any[]>
+  extends Omit<WorkflowOptions, 'requestEagerStart'> {
   /**
    * SignalDefinition or name of signal
    */

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

@@ -3,7 +3,7 @@ import { randomUUID } from 'crypto';
 import asyncRetry from 'async-retry';
 import { ExecutionContext } from 'ava';
 import { firstValueFrom, Subject } from 'rxjs';
-import { WorkflowFailedError, WorkflowHandle } from '@temporalio/client';
+import { Client, Connection, WorkflowClient, WorkflowFailedError, WorkflowHandle } from '@temporalio/client';
 import * as activity from '@temporalio/activity';
 import { msToNumber, tsToMs } from '@temporalio/common/lib/time';
 import { TestWorkflowEnvironment } from '@temporalio/testing';
@@ -1731,3 +1731,55 @@ test('Default handlers fail given reserved prefix', async (t) => {
     await handle.terminate();
   });
 });
+
+export async function helloWorkflow(name: string): Promise<string> {
+  return `Hello, ${name}!`;
+}
+
+test('Workflow can be started eagerly with shared NativeConnection', async (t) => {
+  const { createWorker, taskQueue } = helpers(t);
+  const client = new Client({
+    connection: t.context.env.nativeConnection,
+    namespace: t.context.env.client.options.namespace,
+  });
+
+  const worker = await createWorker();
+  await worker.runUntil(async () => {
+    const handle = await client.workflow.start(helloWorkflow, {
+      args: ['Temporal'],
+      workflowId: `eager-workflow-${randomUUID()}`,
+      taskQueue,
+      requestEagerStart: true,
+      workflowTaskTimeout: '1h', // hang if retry needed
+    });
+
+    t.true(handle.eagerlyStarted);
+
+    const result = await handle.result();
+    t.is(result, 'Hello, Temporal!');
+  });
+});
+
+test('Error thrown when requestEagerStart is used with regular Connection', async (t) => {
+  const { taskQueue } = helpers(t);
+
+  // Create a regular connection instead of native
+  const regularConnection = await Connection.connect();
+  const client = new WorkflowClient({ connection: regularConnection });
+
+  try {
+    await t.throwsAsync(
+      client.start(helloWorkflow, {
+        args: ['Temporal'],
+        workflowId: `eager-workflow-error-${randomUUID()}`,
+        taskQueue,
+        requestEagerStart: true,
+      }),
+      {
+        message: /Eager workflow start requires a NativeConnection/,
+      }
+    );
+  } finally {
+    await regularConnection.close();
+  }
+});

packages/worker/src/connection.ts

@@ -25,6 +25,7 @@ import { Runtime } from './runtime';
  * This class can be used to power `@temporalio/client`'s Client objects.
  */
 export class NativeConnection implements ConnectionLike {
+  public readonly supportsEagerStart = true;
   /**
    * referenceHolders is used internally by the framework, it can be accessed with `extractReferenceHolders` (below)
    */