feat(worker): Don't forcefully shutdown worker (#1072)

BREAKING CHANGE: `WorkerOptions.shutdownGraceTime` no longer forcefully
shuts the worker down. Set `WorkerOptions.shutdownForceTime` to force
shutdown.

As before, in-flight activities are automatically cancelled on worker
shutdown. Setting `shutdownGraceTime` will add a delay before
cancelling the activities. Any activities that respond to this synthetic
cancellation will fail their current attempt as opposted to ending up
cancelled.

Author: bergundy

packages/core-bridge/Cargo.lock

@@ -323,6 +323,10 @@ impl ObjectHandleConversionsExt for Handle<'_, JsObject> {
             js_optional_getter!(cx, self, "maxTaskQueueActivitiesPerSecond", JsNumber)
                 .map(|num| num.value(cx) as f64);
 
+        let graceful_shutdown_period =
+            js_optional_getter!(cx, self, "shutdownGraceTimeMs", JsNumber)
+                .map(|num| Duration::from_millis(num.value(cx) as u64));
+
         match WorkerConfigBuilder::default()
             .worker_build_id(js_value_getter!(cx, self, "buildId", JsString))
             .client_identity_override(Some(js_value_getter!(cx, self, "identity", JsString)))
@@ -332,6 +336,7 @@ impl ObjectHandleConversionsExt for Handle<'_, JsObject> {
             .max_outstanding_local_activities(max_outstanding_local_activities)
             .max_cached_workflows(max_cached_workflows)
             .sticky_queue_schedule_to_start_timeout(sticky_queue_schedule_to_start_timeout)
+            .graceful_shutdown_period(graceful_shutdown_period)
             .namespace(namespace)
             .task_queue(task_queue)
             .max_heartbeat_throttle_interval(max_heartbeat_throttle_interval)
@@ -374,7 +379,7 @@ impl ObjectHandleConversionsExt for Handle<'_, JsObject> {
 
                 let exec_version = match version.as_str() {
                     "default" => {
-                        temporal_sdk_core::ephemeral_server::EphemeralExeVersion::Default {
+                        temporal_sdk_core::ephemeral_server::EphemeralExeVersion::SDKDefault {
                             sdk_name: "sdk-typescript".to_owned(),
                             sdk_version,
                         }

packages/core-bridge/sdk-core

@@ -144,13 +144,16 @@ pub fn start_bridge_loop(
                     headers,
                     callback,
                 } => {
-                    // `metrics_meter` (second arg) can be None here since we don't use the
-                    // returned client directly at the moment, when we repurpose the client to be
-                    // used by a Worker, `init_worker` will attach the correct metrics meter for
-                    // us.
+                    // `metrics_meter` can be None here since we don't use the returned client
+                    // directly at the moment, when we repurpose the client to be used by a Worker,
+                    // `init_worker` will attach the correct metrics meter for us.
                     core_runtime.tokio_handle().spawn(async move {
+                        let metrics_meter = None;
                         match options
-                            .connect_no_namespace(None, headers.map(|h| Arc::new(RwLock::new(h))))
+                            .connect_no_namespace(
+                                metrics_meter,
+                                headers.map(|h| Arc::new(RwLock::new(h))),
+                            )
                             .await
                         {
                             Err(err) => {

packages/core-bridge/src/conversions.rs

@@ -0,0 +1,20 @@
+import { Client, WorkflowHandle } from '@temporalio/client';
+import { QueryDefinition } from '@temporalio/common';
+import { getContext } from './interceptors';
+
+function getSchedulingWorkflowHandle(): WorkflowHandle {
+  const { info, connection, dataConverter } = getContext();
+  const { workflowExecution } = info;
+  const client = new Client({ connection, namespace: info.workflowNamespace, dataConverter });
+  return client.workflow.getHandle(workflowExecution.workflowId, workflowExecution.runId);
+}
+
+export async function signalSchedulingWorkflow(signalName: string): Promise<void> {
+  const handle = getSchedulingWorkflowHandle();
+  await handle.signal(signalName);
+}
+
+export async function queryOwnWf<R, A extends any[]>(queryDef: QueryDefinition<R, A>, ...args: A): Promise<R> {
+  const handle = getSchedulingWorkflowHandle();
+  return await handle.query(queryDef, ...args);
+}

packages/core-bridge/src/runtime.rs

@@ -1,13 +1,13 @@
 /* eslint-disable @typescript-eslint/no-empty-function */
 /* eslint-disable @typescript-eslint/explicit-module-boundary-types */
 import { Context } from '@temporalio/activity';
-import { Client, WorkflowHandle } from '@temporalio/client';
-import { ApplicationFailure, QueryDefinition } from '@temporalio/common';
+import { ApplicationFailure } from '@temporalio/common';
 import { ProtoActivityInput, ProtoActivityResult } from '../../protos/root';
 import { cancellableFetch as cancellableFetchInner } from './cancellable-fetch';
 import { fakeProgress as fakeProgressInner } from './fake-progress';
-import { getContext } from './interceptors';
+import { signalSchedulingWorkflow } from './helpers';
 
+export { queryOwnWf, signalSchedulingWorkflow } from './helpers';
 export { throwSpecificError } from './failure-tester';
 
 /**
@@ -63,23 +63,6 @@ export async function waitForCancellation(): Promise<void> {
   await Context.current().cancelled;
 }
 
-function getSchedulingWorkflowHandle(): WorkflowHandle {
-  const { info, connection, dataConverter } = getContext();
-  const { workflowExecution } = info;
-  const client = new Client({ connection, namespace: info.workflowNamespace, dataConverter });
-  return client.workflow.getHandle(workflowExecution.workflowId, workflowExecution.runId);
-}
-
-export async function signalSchedulingWorkflow(signalName: string) {
-  const handle = getSchedulingWorkflowHandle();
-  await handle.signal(signalName);
-}
-
-export async function queryOwnWf<R, A extends any[]>(queryDef: QueryDefinition<R, A>, ...args: A): Promise<R> {
-  const handle = getSchedulingWorkflowHandle();
-  return await handle.query(queryDef, ...args);
-}
-
 export async function fakeProgress(sleepIntervalMs = 1000, numIters = 1000): Promise<void> {
   await signalSchedulingWorkflow('activityStarted');
   await fakeProgressInner(sleepIntervalMs, numIters);

packages/test/src/activities/helpers.ts

@@ -119,6 +119,7 @@ export function runIntegrationTests(codec?: PayloadCodec): void {
     // NOTE: at the time this was added temporalite did not expose the grpc OperatorService.
     try {
       await connection.operatorService.addSearchAttributes({
+        namespace: 'default',
         searchAttributes: {
           CustomIntField: iface.temporal.api.enums.v1.IndexedValueType.INDEXED_VALUE_TYPE_INT,
           CustomBoolField: iface.temporal.api.enums.v1.IndexedValueType.INDEXED_VALUE_TYPE_BOOL,