extstore/replay and heartbeat metadata

packages/core-bridge/src/worker.rs

@@ -496,7 +496,7 @@ mod config {
     use std::collections::{HashMap, HashSet};
     use std::{sync::Arc, time::Duration};
     use temporalio_common::protos::temporal::api::enums::v1::VersioningBehavior as CoreVersioningBehavior;
-    use temporalio_common::protos::temporal::api::worker::v1::PluginInfo;
+    use temporalio_common::protos::temporal::api::worker::v1::{PluginInfo, StorageDriverInfo};
     use temporalio_common::worker::{
         WorkerDeploymentOptions as CoreWorkerDeploymentOptions,
         WorkerDeploymentVersion as CoreWorkerDeploymentVersion,
@@ -539,6 +539,7 @@ mod config {
         max_task_queue_activities_per_second: Option<f64>,
         shutdown_grace_time: Option<Duration>,
         plugins: Vec<String>,
+        storage_drivers: Vec<String>,
         workflow_failure_errors: HashSet<WorkflowErrorType>,
         workflow_types_to_failure_errors: HashMap<String, HashSet<WorkflowErrorType>>,
     }
@@ -638,6 +639,12 @@ mod config {
                         })
                         .collect::<HashSet<_>>(),
                 )
+                .storage_drivers(
+                    self.storage_drivers
+                        .into_iter()
+                        .map(|r#type| StorageDriverInfo { r#type })
+                        .collect::<HashSet<_>>(),
+                )
                 .workflow_failure_errors(into_core_workflow_error_set(self.workflow_failure_errors))
                 .workflow_types_to_failure_errors(into_core_workflow_error_map_of_sets(
                     self.workflow_types_to_failure_errors,

packages/core-bridge/ts/native.ts

@@ -256,6 +256,7 @@ export interface WorkerOptions {
   maxActivitiesPerSecond: Option<number>;
   shutdownGraceTime: number;
   plugins: string[];
+  storageDrivers: string[];
   workflowFailureErrors: WorkflowErrorType[];
   workflowTypesToFailureErrors: Record<string, WorkflowErrorType[]>;
 }

packages/test/src/test-bridge.ts

@@ -315,6 +315,7 @@ const GenericConfigs = {
       maxActivitiesPerSecond: null,
       shutdownGraceTime: 1000,
       plugins: [],
+      storageDrivers: [],
       workflowFailureErrors: [],
       workflowTypesToFailureErrors: {},
     } satisfies native.WorkerOptions,

packages/test/src/test-extstore-worker-options.ts

@@ -0,0 +1,42 @@
+import test from 'ava';
+import { Runtime } from '@temporalio/worker';
+import { compileWorkerOptions, toNativeWorkerOptions } from '@temporalio/worker/lib/worker-options';
+import { ExternalStorage, type StorageDriver } from '@temporalio/common/lib/converter/extstore';
+import { defaultOptions } from './mock-native-worker';
+
+/** Minimal driver whose reported `type` is independent of its (unique) `name`. */
+function driver(name: string, type: string): StorageDriver {
+  return { name, type, store: async () => [], retrieve: async () => [] };
+}
+
+/** Distinct storage driver types the native worker config would report via the worker heartbeat. */
+function reportedDriverTypes(externalStorage?: ExternalStorage): string[] {
+  const runtime = Runtime.instance();
+  const compiled = compileWorkerOptions(defaultOptions, runtime.logger, runtime.metricMeter);
+  const native = toNativeWorkerOptions({
+    ...compiled,
+    buildId: 'test-build-id',
+    loadedDataConverter: { ...compiled.loadedDataConverter, externalStorage },
+  });
+  return native.storageDrivers;
+}
+
+test('reports no storage drivers when external storage is unset', (t) => {
+  t.deepEqual(reportedDriverTypes(undefined), []);
+});
+
+test('reports each configured driver type', (t) => {
+  const externalStorage = new ExternalStorage({
+    drivers: [driver('primary', 'aws.s3driver'), driver('backup', 'gcp.gcsdriver')],
+    driverSelector: () => null,
+  });
+  t.deepEqual(reportedDriverTypes(externalStorage).sort(), ['aws.s3driver', 'gcp.gcsdriver']);
+});
+
+test('dedups repeated driver types across distinct drivers', (t) => {
+  const externalStorage = new ExternalStorage({
+    drivers: [driver('east', 'aws.s3driver'), driver('west', 'aws.s3driver')],
+    driverSelector: () => null,
+  });
+  t.deepEqual(reportedDriverTypes(externalStorage), ['aws.s3driver']);
+});

packages/worker/src/worker-options.ts

@@ -1179,6 +1179,7 @@ export function toNativeWorkerOptions(opts: CompiledWorkerOptionsWithBuildId): n
     maxActivitiesPerSecond: opts.maxActivitiesPerSecond ?? null,
     shutdownGraceTime: msToNumber(opts.shutdownGraceTime),
     plugins: opts.plugins?.map((p) => p.name) ?? [],
+    storageDrivers: [...new Set((opts.loadedDataConverter.externalStorage?.drivers ?? []).map((d) => d.type))],
     workflowFailureErrors,
     workflowTypesToFailureErrors,
   };

packages/worker/src/worker.ts

@@ -1463,7 +1463,8 @@ export class Worker {
       try {
         const unencodedCompletion = await workflow.workflow.activate(decodedActivation);
         const encodedCompletion = await workflowCodecRunner.encodeCompletion(unencodedCompletion);
-        if (externalStorage) {
+        // skip extstore.store for replay
+        if (externalStorage && !this.isReplayWorker) {
           await storeWorkflowActivationCompletion(externalStorage, encodedCompletion, {
             target: {
               kind: 'workflow',