diff --git a/packages/common/src/converter/extstore.ts b/packages/common/src/converter/extstore.ts index 90caf0f4e..8e1721aa9 100644 --- a/packages/common/src/converter/extstore.ts +++ b/packages/common/src/converter/extstore.ts @@ -104,6 +104,9 @@ export type StorageDriverSelector = (context: StorageDriverStoreContext, payload /** Default {@link ExternalStorage.payloadSizeThreshold}: 256 KiB. */ const DEFAULT_PAYLOAD_SIZE_THRESHOLD = 256 * 1024; +/** Default {@link ExternalStorage.maxConcurrentOperations}: 3. */ +const DEFAULT_MAX_CONCURRENT_OPERATIONS = 3; + /** * Configuration for external storage. Holds the registered drivers, an * optional selector, and the size threshold above which payloads are @@ -120,17 +123,21 @@ export class ExternalStorage { */ readonly driverSelector: StorageDriverSelector; readonly payloadSizeThreshold: number; + readonly maxConcurrentOperations: number; private readonly driversByName: ReadonlyMap; constructor({ drivers, driverSelector, payloadSizeThreshold = DEFAULT_PAYLOAD_SIZE_THRESHOLD, + maxConcurrentOperations = DEFAULT_MAX_CONCURRENT_OPERATIONS, }: { drivers: StorageDriver[]; driverSelector?: StorageDriverSelector; /** Omit for default (256 KiB). Set `0` to consider all payloads regardless of size. */ payloadSizeThreshold?: number; + /** Maximum concurrent store/retrieve driver calls per payload walk. Omit for default (3). */ + maxConcurrentOperations?: number; }) { if (!Array.isArray(drivers) || drivers.length === 0) { throw new ValueError('ExternalStorage requires at least one driver'); @@ -144,6 +151,11 @@ export class ExternalStorage { `ExternalStorage.payloadSizeThreshold must be a non-negative finite number, got ${String(payloadSizeThreshold)}` ); } + if (!Number.isInteger(maxConcurrentOperations) || maxConcurrentOperations < 1) { + throw new ValueError( + `ExternalStorage.maxConcurrentOperations must be a positive integer, got ${String(maxConcurrentOperations)}` + ); + } const driversByName = new Map(); for (const driver of drivers) { @@ -163,6 +175,7 @@ export class ExternalStorage { this.drivers = [...drivers]; this.driverSelector = driverSelector ?? (() => drivers[0] as StorageDriver); this.payloadSizeThreshold = payloadSizeThreshold; + this.maxConcurrentOperations = maxConcurrentOperations; this.driversByName = driversByName; } diff --git a/packages/common/src/internal-non-workflow/external-storage-visitor.ts b/packages/common/src/internal-non-workflow/external-storage-visitor.ts index 90934c6e0..ae3700714 100644 --- a/packages/common/src/internal-non-workflow/external-storage-visitor.ts +++ b/packages/common/src/internal-non-workflow/external-storage-visitor.ts @@ -4,7 +4,7 @@ * * @module */ -import type { ConcurrencyLimit } from '../concurrency/limit'; +import { limit as concurrencyLimit, type ConcurrencyLimit } from '../concurrency/limit'; import type { ExternalStorage, StorageDriverTargetInfo } from '../converter/extstore'; import { ExternalStorageNotConfiguredError } from '../errors'; import type { Payload } from '../interfaces'; @@ -25,7 +25,10 @@ export interface ExternalStorageStoreOptions { initialTarget?: StorageDriverTargetInfo; /** Derives new storage target from the current message. */ deriveContext?: ContextDeriver; - /** Bounds concurrent transform calls across payload sites. Omit for sequential. */ + /** + * Bounds concurrent transform calls across payload sites. Omit to derive one from + * {@link ExternalStorage.maxConcurrentOperations}. + */ limit?: ConcurrencyLimit; /** Aborts the walk and every in-flight driver call. */ abortSignal?: AbortSignal; @@ -37,7 +40,12 @@ export interface ExternalStorageStoreOptions { */ export function extstoreStoreOptions( externalStorage: ExternalStorage, - { initialTarget, deriveContext, limit, abortSignal }: ExternalStorageStoreOptions = {} + { + initialTarget, + deriveContext, + limit = concurrencyLimit(externalStorage.maxConcurrentOperations), + abortSignal, + }: ExternalStorageStoreOptions = {} ): VisitOptions { const runner = new ExternalStorageRunner(externalStorage); return { @@ -55,7 +63,10 @@ export function extstoreStoreOptions( function extstoreRetrieveOptions( externalStorage: ExternalStorage, - { limit, abortSignal }: { limit?: ConcurrencyLimit; abortSignal?: AbortSignal } = {} + { + limit = concurrencyLimit(externalStorage.maxConcurrentOperations), + abortSignal, + }: { limit?: ConcurrencyLimit; abortSignal?: AbortSignal } = {} ): VisitOptions { const runner = new ExternalStorageRunner(externalStorage); return { diff --git a/packages/test/src/test-extstore-config.ts b/packages/test/src/test-extstore-config.ts index e47248b0b..7c2dffc62 100644 --- a/packages/test/src/test-extstore-config.ts +++ b/packages/test/src/test-extstore-config.ts @@ -67,6 +67,23 @@ test('ExternalStorage accepts payloadSizeThreshold = 0', (t) => { t.is(config.payloadSizeThreshold, 0); }); +test('ExternalStorage defaults maxConcurrentOperations to 3', (t) => { + const config = new ExternalStorage({ drivers: [stubDriver('only')] }); + t.is(config.maxConcurrentOperations, 3); +}); + +test('ExternalStorage rejects a non-positive maxConcurrentOperations', (t) => { + t.throws(() => new ExternalStorage({ drivers: [stubDriver('only')], maxConcurrentOperations: 0 }), { + instanceOf: ValueError, + }); +}); + +test('ExternalStorage rejects a non-integer maxConcurrentOperations', (t) => { + t.throws(() => new ExternalStorage({ drivers: [stubDriver('only')], maxConcurrentOperations: 1.5 }), { + instanceOf: ValueError, + }); +}); + test('ExternalStorage rejects negative payloadSizeThreshold', (t) => { t.throws(() => new ExternalStorage({ drivers: [stubDriver('only')], payloadSizeThreshold: -1 }), { instanceOf: ValueError, diff --git a/packages/test/src/test-extstore-visit.ts b/packages/test/src/test-extstore-visit.ts index 84e61e8e0..45d597d13 100644 --- a/packages/test/src/test-extstore-visit.ts +++ b/packages/test/src/test-extstore-visit.ts @@ -2,7 +2,7 @@ import test from 'ava'; import type { Payload } from '@temporalio/common'; import { ExternalStorageNotConfiguredError } from '@temporalio/common'; -import { ExternalStorage } from '@temporalio/common/lib/converter/extstore'; +import { ExternalStorage, StorageDriverClaim } from '@temporalio/common/lib/converter/extstore'; import { ExternalStorageRunner, extstoreInboundOptions, @@ -246,6 +246,94 @@ test('client response retrieve resolves the query result (via generic visit)', a t.deepEqual(response.queryResult!.payloads![0], queryResult); }); +/** + * A driver whose store/retrieve calls park until `expected` of them are in flight at once, and + * records the peak number of concurrent calls. Parked calls release after a short timeout so a + * sequential walk fails the peak assertion instead of deadlocking. + */ +function makeConcurrencyProbe(expected: number) { + let inflight = 0; + let peak = 0; + let arrived: () => void; + const gate = new Promise((resolve) => (arrived = resolve)); + const enter = async () => { + inflight++; + peak = Math.max(peak, inflight); + if (inflight >= expected) arrived(); + await Promise.race([gate, new Promise((resolve) => setTimeout(resolve, 100))]); + inflight--; + }; + const driver = makeFakeDriver({ + onStore: async (payloads) => { + await enter(); + return payloads.map(() => new StorageDriverClaim({ id: 'x' })); + }, + onRetrieve: async (claims) => { + await enter(); + return claims.map(() => makePayload(0)); + }, + }); + return { driver, peak: () => peak }; +} + +test('store operations across payload sites run concurrently, capped at 3 by default', async (t) => { + const { driver, peak } = makeConcurrencyProbe(3); + const externalStorage = new ExternalStorage({ drivers: [driver], payloadSizeThreshold: 96 }); + const completion: coresdk.workflow_completion.IWorkflowActivationCompletion = { + successful: { + commands: Array.from({ length: 5 }, (_, i) => ({ scheduleActivity: { arguments: [makePayload(256, i)] } })), + }, + }; + + await visit( + completion, + walkWorkflowActivationCompletion, + extstoreStoreOptions(externalStorage, { initialTarget: WORKFLOW_TARGET }) + ); + + t.is(peak(), 3); +}); + +test('retrieve operations across payload sites run concurrently, capped at 3 by default', async (t) => { + // Same driver name as the probe below so the probe's storage can resolve these references. + const { externalStorage: sourceStorage } = externalStorageWith(makeFakeDriver()); + const jobs = await Promise.all( + Array.from({ length: 5 }, async (_, i) => ({ + resolveActivity: { result: { completed: { result: await toReference(sourceStorage, makePayload(256, i)) } } }, + })) + ); + + const { driver, peak } = makeConcurrencyProbe(3); + const externalStorage = new ExternalStorage({ drivers: [driver], payloadSizeThreshold: 96 }); + const activation: coresdk.workflow_activation.IWorkflowActivation = { jobs }; + + await visit(activation, walkWorkflowActivation, extstoreInboundOptions(externalStorage)); + + t.is(peak(), 3); +}); + +test('store concurrency is configurable via maxConcurrentOperations', async (t) => { + const { driver, peak } = makeConcurrencyProbe(2); + const externalStorage = new ExternalStorage({ + drivers: [driver], + payloadSizeThreshold: 96, + maxConcurrentOperations: 2, + }); + const completion: coresdk.workflow_completion.IWorkflowActivationCompletion = { + successful: { + commands: Array.from({ length: 5 }, (_, i) => ({ scheduleActivity: { arguments: [makePayload(256, i)] } })), + }, + }; + + await visit( + completion, + walkWorkflowActivationCompletion, + extstoreStoreOptions(externalStorage, { initialTarget: WORKFLOW_TARGET }) + ); + + t.is(peak(), 2); +}); + test('inbound options raise TMPRL1105 on a reference when storage is not configured', async (t) => { const { externalStorage } = externalStorageWith(); const task: coresdk.activity_task.IActivityTask = {