Worker heartbeating

Author: yuandrew

packages/core-bridge/src/runtime.rs

@@ -59,11 +59,13 @@ pub struct Runtime {
 pub fn runtime_new(
     bridge_options: config::RuntimeOptions,
 ) -> BridgeResult<OpaqueOutboundHandle<Runtime>> {
-    let (telemetry_options, metrics_options, logging_options) = bridge_options.try_into()?;
+    let (telemetry_options, metrics_options, logging_options, worker_heartbeat_interval_millis) =
+        bridge_options.try_into()?;
 
     // Create core runtime which starts tokio multi-thread runtime
     let runtime_options = RuntimeOptionsBuilder::default()
         .telemetry_options(telemetry_options)
+        .heartbeat_interval(worker_heartbeat_interval_millis.map(Duration::from_millis))
         .build()
         .context("Failed to build runtime options")?;
     let mut core_runtime = CoreRuntime::new(runtime_options, TokioRuntimeBuilder::default())
@@ -265,6 +267,7 @@ mod config {
         log_exporter: LogExporterOptions,
         telemetry: TelemetryOptions,
         metrics_exporter: Option<MetricsExporterOptions>,
+        worker_heartbeat_interval_millis: Option<u64>,
     }
 
     #[derive(Debug, Clone, TryFromJs)]
@@ -321,6 +324,7 @@ mod config {
             CoreTelemetryOptions,
             Option<super::BridgeMetricsExporter>,
             super::BridgeLogExporter,
+            Option<u64>,
         )> for RuntimeOptions
     {
         type Error = BridgeError;
@@ -330,8 +334,16 @@ mod config {
             CoreTelemetryOptions,
             Option<super::BridgeMetricsExporter>,
             super::BridgeLogExporter,
+            Option<u64>,
         )> {
-            let (telemetry_logger, log_exporter) = match self.log_exporter {
+            let Self {
+                log_exporter,
+                telemetry,
+                metrics_exporter,
+                worker_heartbeat_interval_millis,
+            } = self;
+
+            let (telemetry_logger, log_exporter) = match log_exporter {
                 LogExporterOptions::Console { filter } => (
                     CoreTelemetryLogger::Console { filter },
                     BridgeLogExporter::Console,
@@ -351,17 +363,21 @@ mod config {
             let mut telemetry_options = TelemetryOptionsBuilder::default();
             let telemetry_options = telemetry_options
                 .logging(telemetry_logger)
-                .metric_prefix(self.telemetry.metric_prefix)
-                .attach_service_name(self.telemetry.attach_service_name)
+                .metric_prefix(telemetry.metric_prefix)
+                .attach_service_name(telemetry.attach_service_name)
                 .build()
                 .context("Failed to build telemetry options")?;
 
-            let metrics_exporter = self
-                .metrics_exporter
+            let metrics_exporter = metrics_exporter
                 .map(std::convert::TryInto::try_into)
                 .transpose()?;
 
-            Ok((telemetry_options, metrics_exporter, log_exporter))
+            Ok((
+                telemetry_options,
+                metrics_exporter,
+                log_exporter,
+                worker_heartbeat_interval_millis,
+            ))
         }
     }
 

packages/core-bridge/src/worker.rs

@@ -295,7 +295,7 @@ pub fn worker_complete_nexus_task(
             .complete_nexus_task(nexus_completion)
             .await
             .map_err(|err| match err {
-                CompleteNexusError::NexusNotEnabled {} => {
+                CompleteNexusError::NexusNotEnabled => {
                     BridgeError::UnexpectedError(format!("{err}"))
                 }
                 CompleteNexusError::MalformedNexusCompletion { reason } => BridgeError::TypeError {

packages/core-bridge/ts/native.ts

@@ -40,7 +40,7 @@ export type JsonString<_T> = string;
 // Runtime
 ////////////////////////////////////////////////////////////////////////////////////////////////////
 
-export declare function newRuntime(telemOptions: RuntimeOptions): Runtime;
+export declare function newRuntime(runtimeOptions: RuntimeOptions): Runtime;
 
 export declare function runtimeShutdown(runtime: Runtime): void;
 
@@ -52,6 +52,7 @@ export type RuntimeOptions = {
   logExporter: LogExporterOptions;
   telemetry: TelemetryOptions;
   metricsExporter: MetricExporterOptions;
+  workerHeartbeatIntervalMillis: Option<number>;
 };
 
 export type TelemetryOptions = {

packages/test/src/test-bridge.ts

@@ -241,6 +241,7 @@ const GenericConfigs = {
         attachServiceName: false,
       },
       metricsExporter: null,
+      workerHeartbeatIntervalMillis: null,
     } satisfies native.RuntimeOptions,
   },
   client: {

packages/test/src/test-runtime.ts

@@ -156,6 +156,29 @@ if (RUN_INTEGRATION_TESTS) {
     t.is(logEntries.filter((x) => x.message.startsWith('workflow log ')).length, 5);
     t.is(logEntries.filter((x) => x.message.startsWith('final log')).length, 1);
   });
+
+  test.serial('Runtime handle heartbeat duration default', async (t) => {
+    const runtime = Runtime.install({});
+    const SIXTY_SECONDS = 60 * 1000;
+    t.true(runtime.options.runtimeOptions.workerHeartbeatIntervalMillis === SIXTY_SECONDS);
+    await runtime.shutdown();
+  });
+
+  test.serial('Runtime handle heartbeat duration null', async (t) => {
+    const runtime = Runtime.install({
+      workerHeartbeatInterval: null,
+    });
+    t.true(runtime.options.runtimeOptions.workerHeartbeatIntervalMillis === null);
+    await runtime.shutdown();
+  });
+
+  test.serial('Runtime handle heartbeat duration undefined', async (t) => {
+    const runtime = Runtime.install({
+      workerHeartbeatInterval: 13 * 1000,
+    });
+    t.true(runtime.options.runtimeOptions.workerHeartbeatIntervalMillis === 13 * 1000);
+    await runtime.shutdown();
+  });
 }
 
 export async function log5Times(): Promise<void> {

packages/worker/src/runtime-options.ts

@@ -32,6 +32,14 @@ export interface RuntimeOptions {
    */
   telemetryOptions?: TelemetryOptions;
 
+  /**
+   * Interval for worker heartbeats. `null` disables heartbeating.
+   *
+   * @format number of milliseconds or {@link https://www.npmjs.com/package/ms | ms-formatted string}
+   * @default 60000 (60 seconds)
+   */
+  workerHeartbeatInterval?: Duration | null;
+
   /**
    * Automatically shutdown workers on any of these signals.
    *
@@ -359,18 +367,25 @@ export interface PrometheusMetricsExporter {
  */
 export interface CompiledRuntimeOptions {
   shutdownSignals: NodeJS.Signals[];
-  telemetryOptions: native.RuntimeOptions;
+  runtimeOptions: native.RuntimeOptions;
   logger: Logger;
 }
 
 export function compileOptions(options: RuntimeOptions): CompiledRuntimeOptions {
   const { metrics, noTemporalPrefixForMetrics } = options.telemetryOptions ?? {}; // eslint-disable-line deprecation/deprecation
   const [logger, logExporter] = compileLoggerOptions(options);
 
+  let workerHeartbeatIntervalMillis: number | null;
+  if (options.workerHeartbeatInterval === null) {
+    workerHeartbeatIntervalMillis = null;
+  } else {
+    workerHeartbeatIntervalMillis = msToNumber(options.workerHeartbeatInterval ?? '60s');
+  }
+
   return {
     logger,
     shutdownSignals: options.shutdownSignals ?? ['SIGINT', 'SIGTERM', 'SIGQUIT', 'SIGUSR2'],
-    telemetryOptions: {
+    runtimeOptions: {
       logExporter,
       telemetry: {
         metricPrefix: metrics?.metricPrefix ?? (noTemporalPrefixForMetrics ? '' : 'temporal_'),
@@ -400,6 +415,7 @@ export function compileOptions(options: RuntimeOptions): CompiledRuntimeOptions
                 globalTags: metrics.globalTags ?? {},
               } satisfies native.MetricExporterOptions)
             : null,
+      workerHeartbeatIntervalMillis,
     },
   };
 }

packages/worker/src/runtime.ts

@@ -50,7 +50,7 @@ export class Runtime {
     public readonly options: CompiledRuntimeOptions
   ) {
     this.logger = options.logger;
-    this.metricMeter = options.telemetryOptions.metricsExporter
+    this.metricMeter = options.runtimeOptions.metricsExporter
       ? MetricMeterWithComposedTags.compose(new RuntimeMetricMeter(this.native), {}, true)
       : noopMetricMeter;
 
@@ -97,7 +97,7 @@ export class Runtime {
    */
   protected static create(options: RuntimeOptions, instantiator: 'install' | 'instance'): Runtime {
     const compiledOptions = compileOptions(options);
-    const runtime = native.newRuntime(compiledOptions.telemetryOptions);
+    const runtime = native.newRuntime(compiledOptions.runtimeOptions);
 
     // Remember the provided options in case Core is reinstantiated after being shut down
     this.defaultOptions = options;