getClient() and fix versioning strategy

Author: bergundy

packages/common/src/converter/failure-converter.ts

@@ -58,7 +58,7 @@ const CUTOFF_STACK_PATTERNS = combineRegExp(
   /** Activity execution */
   /\s+at Activity\.execute \(.*[\\/]worker[\\/](?:src|lib)[\\/]activity\.[jt]s:\d+:\d+\)/,
   /** Nexus execution */
-  /\s+at NexusHandler\.invokeUserCode \(.*[\\/]worker[\\/](?:src|lib)[\\/]nexus\.[jt]s:\d+:\d+\)/,
+  /\s+at( async)? NexusHandler\.invokeUserCode \(.*[\\/]worker[\\/](?:src|lib)[\\/]nexus\/index\.[jt]s:\d+:\d+\)/,
   /** Workflow activation */
   /\s+at Activator\.\S+NextHandler \(.*[\\/]workflow[\\/](?:src|lib)[\\/]internals\.[jt]s:\d+:\d+\)/,
   /** Workflow run anything in context */

packages/core-bridge/src/worker.rs

@@ -465,7 +465,7 @@ mod config {
 
     use temporal_sdk_core::{
         api::worker::{
-            ActivitySlotKind, LocalActivitySlotKind, NexusSlotKind, PollerBehavior as CorePollerBehavior, SlotKind, WorkerConfig, WorkerConfigBuilder, WorkerConfigBuilderError, WorkflowSlotKind
+            ActivitySlotKind, LocalActivitySlotKind, NexusSlotKind, PollerBehavior as CorePollerBehavior, SlotKind, WorkerConfig, WorkerConfigBuilder, WorkerConfigBuilderError, WorkerVersioningStrategy, WorkflowSlotKind
         }, ResourceBasedSlotsOptions, ResourceBasedSlotsOptionsBuilder, ResourceSlotOptions, SlotSupplierOptions as CoreSlotSupplierOptions, TunerHolder, TunerHolderOptionsBuilder
     };
 
@@ -510,6 +510,7 @@ mod config {
             // Set all other options
             let mut builder = WorkerConfigBuilder::default();
             builder
+                .versioning_strategy(WorkerVersioningStrategy::None{build_id: "".to_owned()})
                 .client_identity_override(Some(self.identity))
                 .task_queue(self.task_queue)
                 .namespace(self.namespace)

packages/nexus/src/context.ts

@@ -1,16 +1,14 @@
 import { HandlerContext as BaseHandlerContext, getHandlerContext } from 'nexus-rpc/lib/handler';
 import { Logger, LogLevel, LogMetadata } from '@temporalio/common';
+import { Client } from '@temporalio/client';
 
 export interface HandlerContext extends BaseHandlerContext {
   log: Logger;
+  client: Client;
 }
 
 function getLogger() {
-  const ctx = getHandlerContext<HandlerContext>();
-  if (ctx == null) {
-    throw new ReferenceError('HandlerContext uninitialized');
-  }
-  return ctx.log;
+  return getHandlerContext<HandlerContext>().log;
 }
 
 /**
@@ -36,3 +34,13 @@ export const log: Logger = {
     return getLogger().error(message, meta);
   },
 };
+
+/**
+ * Returns a client to be used in a Nexus Operation's context, this Client is powered by the same Connection that the
+ * worker was created with.
+ */
+export function getClient(): Client {
+  return getHandlerContext<HandlerContext>().client;
+}
+
+// TODO: also support getting a metrics handler.

packages/nexus/src/index.ts

@@ -1 +1 @@
-export { log } from './context';
+export { log, getClient } from './context';

packages/test/src/test-nexus-handler.ts

@@ -1,3 +1,5 @@
+// TODO: Remove /Users/bergundy/... for stack traces, requires publishing nexus-rpc.
+
 import { randomUUID } from 'node:crypto';
 import anyTest, { TestFn } from 'ava';
 import getPort from 'get-port';
@@ -313,7 +315,8 @@ test('start operation handler errors', async (t) => {
         cleanStackTrace(err.stack!),
         `ApplicationFailure: deliberate failure
     at Function.create (common/src/failure.ts)
-    at op (test/src/test-nexus-handler.ts)`
+    at op (test/src/test-nexus-handler.ts)
+    at ServiceRegistry.start (/Users/bergundy/temporal/nexus-sdk-typescript/src/operation.ts)`
       );
       t.deepEqual((err as ApplicationFailure).details, ['details']);
       t.is((err as ApplicationFailure).failure?.source, 'TypeScriptSDK');
@@ -338,7 +341,8 @@ test('start operation handler errors', async (t) => {
       t.deepEqual(
         cleanStackTrace(err.stack!),
         `HandlerError: deliberate error
-    at op (test/src/test-nexus-handler.ts)`
+    at op (test/src/test-nexus-handler.ts)
+    at ServiceRegistry.start (/Users/bergundy/temporal/nexus-sdk-typescript/src/operation.ts)`
       );
     }
     {
@@ -361,7 +365,8 @@ test('start operation handler errors', async (t) => {
       t.deepEqual(
         cleanStackTrace(err.stack!),
         `OperationError: deliberate error
-    at op (test/src/test-nexus-handler.ts)`
+    at op (test/src/test-nexus-handler.ts)
+    at ServiceRegistry.start (/Users/bergundy/temporal/nexus-sdk-typescript/src/operation.ts)`
       );
     }
   });
@@ -447,7 +452,8 @@ test('cancel operation handler errors', async (t) => {
         cleanStackTrace(err.stack!),
         `ApplicationFailure: deliberate failure
     at Function.create (common/src/failure.ts)
-    at Object.cancel (test/src/test-nexus-handler.ts)`
+    at Object.cancel (test/src/test-nexus-handler.ts)
+    at ServiceRegistry.cancel (/Users/bergundy/temporal/nexus-sdk-typescript/src/operation.ts)`
       );
       t.deepEqual((err as ApplicationFailure).details, ['details']);
       t.is((err as ApplicationFailure).failure?.source, 'TypeScriptSDK');
@@ -475,7 +481,8 @@ test('cancel operation handler errors', async (t) => {
       t.deepEqual(
         cleanStackTrace(err.stack!),
         `HandlerError: deliberate error
-    at Object.cancel (test/src/test-nexus-handler.ts)`
+    at Object.cancel (test/src/test-nexus-handler.ts)
+    at ServiceRegistry.cancel (/Users/bergundy/temporal/nexus-sdk-typescript/src/operation.ts)`
       );
     }
   });
@@ -528,3 +535,38 @@ test('logger is available in handler context', async (t) => {
     input: 'hello',
   });
 });
+
+test('getClient is available in handler context', async (t) => {
+  const { env, taskQueue, httpPort, endpointId } = t.context;
+
+  const w = await Worker.create({
+    connection: env.nativeConnection,
+    namespace: env.namespace,
+    taskQueue,
+    nexusServices: [
+      nexus.serviceHandler(
+        nexus.service('testService', {
+          testSyncOp: nexus.operation<void, boolean>(),
+        }),
+        {
+          async testSyncOp() {
+            const systemInfo = await temporalnexus.getClient().connection.workflowService.getSystemInfo({ namespace: 'default' });
+            return systemInfo.capabilities?.nexus ?? false;
+          },
+        }
+      ),
+    ],
+  });
+
+  await w.runUntil(async () => {
+    const res = await fetch(
+      `http://localhost:${httpPort}/nexus/endpoints/${endpointId}/services/testService/testSyncOp`,
+      {
+        method: 'POST',
+      }
+    );
+    t.true(res.ok);
+    const output = await res.json();
+    t.is(output, true);
+  });
+});

packages/worker/src/nexus/index.ts

@@ -22,7 +22,7 @@ import {
 } from '@temporalio/common/lib/internal-non-workflow';
 import { fixBuffers } from '@temporalio/common/lib/proto-utils';
 import { isAbortError } from '@temporalio/common/lib/type-helpers';
-import { isGrpcServiceError, ServiceError } from '@temporalio/client';
+import { Client, isGrpcServiceError, ServiceError } from '@temporalio/client';
 import { Logger, withMetadata } from '../logger';
 
 const UNINITIALIZED = Symbol();
@@ -80,6 +80,7 @@ export class NexusHandler {
   constructor(
     public readonly taskToken: Uint8Array,
     public readonly info: nexus.HandlerInfo,
+    public readonly client: Client,
     public readonly abortController: AbortController,
     public readonly serviceRegistry: nexus.ServiceRegistry,
     public readonly dataConverter: LoadedDataConverter,
@@ -276,10 +277,17 @@ export class NexusHandler {
   ): Promise<coresdk.nexus.INexusTaskCompletion> {
     const context: HandlerContext = {
       info: this.info,
+      client: this.client,
       links: [],
       log: withMetadata(this.workerLogger, { sdkComponent: SdkComponent.nexus }),
     };
-    return await withContext(context, this.execute.bind(this, task));
+    let execute = this.execute.bind(this, task);
+    // Ensure that client calls made with the worker's client in this handler's context are tied to the abort signal.
+    // TODO: Actually support canceling requests backed by NativeConnection. Once it does, this functionality should be
+    // tested.
+    // TS can't infer this and typing out the types is redundant and hard to read.
+    execute = this.client.withAbortSignal.bind(this.client, this.info.abortSignal, execute) as any;
+    return await withContext(context, execute);
   }
 }
 

packages/worker/src/worker.ts

@@ -56,6 +56,7 @@ import {
 import { errorMessage, NonNullableObject, OmitFirstParam } from '@temporalio/common/lib/type-helpers';
 import { workflowLogAttributes } from '@temporalio/workflow/lib/logs';
 import { native } from '@temporalio/core-bridge';
+import { Client } from '@temporalio/client';
 import { coresdk, temporal } from '@temporalio/proto';
 import { type SinkCall, type WorkflowInfo } from '@temporalio/workflow';
 import { Activity, CancelReason, activityLogAttributes } from './activity';
@@ -455,6 +456,8 @@ export class Worker {
    */
   protected hasOutstandingNexusPoll = false;
 
+  private client?: Client;
+
   protected readonly numInFlightActivationsSubject = new BehaviorSubject<number>(0);
   protected readonly numInFlightActivitiesSubject = new BehaviorSubject<number>(0);
   protected readonly numInFlightNonLocalActivitiesSubject = new BehaviorSubject<number>(0);
@@ -492,11 +495,11 @@ export class Worker {
         ...compiledOptions,
         ...(compiledOptions.workflowBundle && isCodeBundleOption(compiledOptions.workflowBundle)
           ? {
-              // Avoid dumping workflow bundle code to the console
-              workflowBundle: <WorkflowBundle>{
-                code: `<string of length ${compiledOptions.workflowBundle.code.length}>`,
-              },
-            }
+            // Avoid dumping workflow bundle code to the console
+            workflowBundle: <WorkflowBundle>{
+              code: `<string of length ${compiledOptions.workflowBundle.code.length}>`,
+            },
+          }
           : {}),
       },
     });
@@ -710,7 +713,7 @@ export class Worker {
       ) {
         logger.warn(
           'Ignoring WorkerOptions.interceptors.workflowModules because WorkerOptions.workflowBundle is set.\n' +
-            'To use workflow interceptors with a workflowBundle, pass them in the call to bundleWorkflowCode.'
+          'To use workflow interceptors with a workflowBundle, pass them in the call to bundleWorkflowCode.'
         );
       }
 
@@ -756,6 +759,15 @@ export class Worker {
     protected readonly isReplayWorker: boolean = false
   ) {
     this.workflowCodecRunner = new WorkflowCodecRunner(options.loadedDataConverter.payloadCodecs);
+    if (connection !== null) {
+      this.client = new Client({
+        namespace: options.namespace,
+        connection,
+        identity: options.identity,
+        dataConverter: options.dataConverter,
+        // TODO: support interceptors.
+      });
+    }
   }
 
   /**
@@ -883,8 +895,8 @@ export class Worker {
    */
   protected pollLoop$<T>(pollFn: () => Promise<T>): Observable<T> {
     return from(
-      (async function* () {
-        for (;;) {
+      (async function*() {
+        for (; ;) {
           try {
             yield await pollFn();
           } catch (err) {
@@ -919,14 +931,14 @@ export class Worker {
               // so it can be cancelled if requested
               let output:
                 | {
-                    type: 'result';
-                    result: coresdk.activity_result.IActivityExecutionResult;
-                  }
+                  type: 'result';
+                  result: coresdk.activity_result.IActivityExecutionResult;
+                }
                 | {
-                    type: 'run';
-                    activity: Activity;
-                    input: ActivityExecuteInput;
-                  }
+                  type: 'run';
+                  activity: Activity;
+                  input: ActivityExecuteInput;
+                }
                 | { type: 'ignore' };
               switch (variant) {
                 case 'start': {
@@ -1176,6 +1188,7 @@ export class Worker {
       const nexusHandler = new NexusHandler(
         taskToken,
         info,
+        this.client!, // Must be defined if we are handling Nexus tasks.
         ctrl,
         this.options.nexusServiceRegistry!, // Must be defined if we are handling Nexus tasks.
         this.options.loadedDataConverter,
@@ -1284,9 +1297,9 @@ export class Worker {
         const completion = synthetic
           ? undefined
           : coresdk.workflow_completion.WorkflowActivationCompletion.encodeDelimited({
-              runId: activation.runId,
-              successful: {},
-            }).finish();
+            runId: activation.runId,
+            successful: {},
+          }).finish();
         return { state: undefined, output: { close, completion } };
       }