diff --git a/packages/core-bridge/src/worker.rs b/packages/core-bridge/src/worker.rs index cefffa08d..d8f395d57 100644 --- a/packages/core-bridge/src/worker.rs +++ b/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, shutdown_grace_time: Option, plugins: Vec, + storage_drivers: Vec, workflow_failure_errors: HashSet, workflow_types_to_failure_errors: HashMap>, } @@ -638,6 +639,12 @@ mod config { }) .collect::>(), ) + .storage_drivers( + self.storage_drivers + .into_iter() + .map(|r#type| StorageDriverInfo { r#type }) + .collect::>(), + ) .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, diff --git a/packages/core-bridge/ts/native.ts b/packages/core-bridge/ts/native.ts index 20d4ac7a6..b6d0bfe53 100644 --- a/packages/core-bridge/ts/native.ts +++ b/packages/core-bridge/ts/native.ts @@ -256,6 +256,7 @@ export interface WorkerOptions { maxActivitiesPerSecond: Option; shutdownGraceTime: number; plugins: string[]; + storageDrivers: string[]; workflowFailureErrors: WorkflowErrorType[]; workflowTypesToFailureErrors: Record; } diff --git a/packages/test/src/test-bridge.ts b/packages/test/src/test-bridge.ts index ea391cd92..e139b6462 100644 --- a/packages/test/src/test-bridge.ts +++ b/packages/test/src/test-bridge.ts @@ -315,6 +315,7 @@ const GenericConfigs = { maxActivitiesPerSecond: null, shutdownGraceTime: 1000, plugins: [], + storageDrivers: [], workflowFailureErrors: [], workflowTypesToFailureErrors: {}, } satisfies native.WorkerOptions, diff --git a/packages/test/src/test-extstore-worker-options.ts b/packages/test/src/test-extstore-worker-options.ts new file mode 100644 index 000000000..fff32f51d --- /dev/null +++ b/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']); +}); diff --git a/packages/worker/src/worker-options.ts b/packages/worker/src/worker-options.ts index 84c604cd3..e28b5c1cf 100644 --- a/packages/worker/src/worker-options.ts +++ b/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, }; diff --git a/packages/worker/src/worker.ts b/packages/worker/src/worker.ts index 9d3113d54..7ae14f2e8 100644 --- a/packages/worker/src/worker.ts +++ b/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',