Worker heartbeating

Author: yuandrew

package.json

@@ -100,5 +100,6 @@
     "node": ">= 18.0.0",
     "npm": ">= 7.0.0",
     "rustc": ">= 1.53.0"
-  }
+  },
+  "packageManager": "pnpm@10.7.1+sha512.2d92c86b7928dc8284f53494fb4201f983da65f0fb4f0d40baafa5cf628fa31dae3e5968f12466f17df7e97310e30f343a648baea1b9b350685dafafffdf5808"
 }

packages/core-bridge/src/runtime.rs

@@ -36,6 +36,10 @@ macro_rules! enter_sync {
 pub fn init(cx: &mut neon::prelude::ModuleContext) -> neon::prelude::NeonResult<()> {
     cx.export_function("newRuntime", runtime_new)?;
     cx.export_function("runtimeShutdown", runtime_shutdown)?;
+    // cx.export_function(
+    //     "runtimeGetWorkerHeartbeatIntervalMillis",
+    //     runtime_get_worker_heartbeat_interval_millis,
+    // )?;
 
     Ok(())
 }
@@ -51,6 +55,7 @@ pub struct Runtime {
     // For some unknown reason, the otel metrics exporter will go crazy on shutdown in some
     // scenarios if we don't hold on to the `CoreOtelMeter` till the `Runtime` finally gets dropped.
     _otel_metrics_exporter: Option<Arc<dyn CoreMeter + 'static>>,
+    // worker_heartbeat_interval_millis: Option<u64>,
 }
 
 /// Initialize Core global telemetry and create the tokio runtime required to run Core.
@@ -59,11 +64,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())
@@ -125,6 +132,7 @@ pub fn runtime_new(
         log_exporter_task,
         metrics_exporter_task: prom_metrics_exporter_task.map(Arc::new),
         _otel_metrics_exporter: otel_metrics_exporter,
+        // worker_heartbeat_interval_millis: runtime_options.worker_heartbeat_interval.map(|d| d.as_millis() as u64),
     }))
 }
 
@@ -138,6 +146,21 @@ pub fn runtime_shutdown(runtime: OpaqueInboundHandle<Runtime>) -> BridgeResult<(
     Ok(())
 }
 
+// #[js_function]
+// pub fn runtime_get_worker_heartbeat_interval_millis(
+//     runtime: OpaqueInboundHandle<Runtime>,
+// ) -> BridgeResult<Option<u32>> {
+//     runtime
+//         .borrow()?
+//         .worker_heartbeat_interval_millis
+//         .map(u32::try_from)
+//         .transpose()
+//         .map_err(|_| BridgeError::TypeError {
+//             field: None,
+//             message: "workerHeartbeatIntervalMillis is too large to represent in JavaScript".into(),
+//         })
+// }
+
 /// Drop will handle the cleanup
 impl MutableFinalize for Runtime {}
 
@@ -265,6 +288,7 @@ mod config {
         log_exporter: LogExporterOptions,
         telemetry: TelemetryOptions,
         metrics_exporter: Option<MetricsExporterOptions>,
+        worker_heartbeat_interval_millis: Option<u64>,
     }
 
     #[derive(Debug, Clone, TryFromJs)]
@@ -321,6 +345,7 @@ mod config {
             CoreTelemetryOptions,
             Option<super::BridgeMetricsExporter>,
             super::BridgeLogExporter,
+            Option<u64>,
         )> for RuntimeOptions
     {
         type Error = BridgeError;
@@ -330,8 +355,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 +384,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,10 +40,12 @@ 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;
 
+export declare function runtimeGetWorkerHeartbeatIntervalMillis(runtime: Runtime): number | null;
+
 export interface Runtime {
   type: 'runtime';
 }
@@ -52,6 +54,7 @@ export type RuntimeOptions = {
   logExporter: LogExporterOptions;
   telemetry: TelemetryOptions;
   metricsExporter: MetricExporterOptions;
+  workerHeartbeatIntervalMillis: Option<number>;
 };
 
 export type TelemetryOptions = {

packages/test/src/test-bridge.ts

@@ -227,6 +227,20 @@ test("Stopping Worker after creating another runtime doesn't fail", async (t) =>
   t.pass();
 });
 
+test('Creating runtime with heartbeat enabled plumbs heartbeat duration', (t) => {
+  const runtime = native.newRuntime({
+    ...GenericConfigs.runtime.basic,
+    workerHeartbeatIntervalMillis: 100,
+  });
+  t.is(native.runtimeGetWorkerHeartbeatIntervalMillis(runtime), 100);
+
+  const runtime1 = native.newRuntime({
+    ...GenericConfigs.runtime.basic,
+    workerHeartbeatIntervalMillis: null,
+  });
+  t.is(native.runtimeGetWorkerHeartbeatIntervalMillis(runtime1), null);
+});
+
 // Sample configs ///////////////////////////////////////////////////////////////////////////////////
 
 const GenericConfigs = {
@@ -241,6 +255,7 @@ const GenericConfigs = {
         attachServiceName: false,
       },
       metricsExporter: null,
+      workerHeartbeatIntervalMillis: null,
     } satisfies native.RuntimeOptions,
   },
   client: {

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.
    *
@@ -367,6 +375,14 @@ export function compileOptions(options: RuntimeOptions): CompiledRuntimeOptions
   const { metrics, noTemporalPrefixForMetrics } = options.telemetryOptions ?? {}; // eslint-disable-line deprecation/deprecation
   const [logger, logExporter] = compileLoggerOptions(options);
 
+  // Handle worker heartbeat interval - default to 60s, allow null to disable
+  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'],
@@ -376,6 +392,7 @@ export function compileOptions(options: RuntimeOptions): CompiledRuntimeOptions
         metricPrefix: metrics?.metricPrefix ?? (noTemporalPrefixForMetrics ? '' : 'temporal_'),
         attachServiceName: metrics?.attachServiceName ?? true,
       },
+      workerHeartbeatIntervalMillis,
       metricsExporter:
         metrics && isPrometheusMetricsExporter(metrics)
           ? ({