Skip to content

feat(extstore): run store and retrieve operations concurrently (default 3, configurable) #2294

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
VihaanAgarwal wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(extstore): run store and retrieve operations concurrently (default 3, configurable) #2294

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions packages/common/src/converter/extstore.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -120,17 +123,21 @@ export class ExternalStorage {
*/
readonly driverSelector: StorageDriverSelector;
readonly payloadSizeThreshold: number;
readonly maxConcurrentOperations: number;
private readonly driversByName: ReadonlyMap<string, StorageDriver>;

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');
Expand All @@ -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<string, StorageDriver>();
for (const driver of drivers) {
Expand All @@ -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;
}

Expand Down
19 changes: 15 additions & 4 deletions packages/common/src/internal-non-workflow/external-storage-visitor.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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';
Expand All @@ -25,7 +25,10 @@ export interface ExternalStorageStoreOptions {
initialTarget?: StorageDriverTargetInfo;
/** Derives new storage target from the current message. */
deriveContext?: ContextDeriver<StoreTarget>;
/** 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;
Expand All @@ -37,7 +40,12 @@ export interface ExternalStorageStoreOptions {
*/
export function extstoreStoreOptions(
externalStorage: ExternalStorage,
{ initialTarget, deriveContext, limit, abortSignal }: ExternalStorageStoreOptions = {}
{
initialTarget,
deriveContext,
limit = concurrencyLimit(externalStorage.maxConcurrentOperations),

@mjameswh mjameswh Aug 5, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think that's correct. The concurrency set on externalStorage should be shared across all callers, not independently per visitors.

For example, if I set a limit of four on my externalStorage that's because I don't want to ever have more than four pending requests to the store. As it is written now, if I have three tasks being decoded concurrently, each task visitor gets it's own concurrency limit, potentially resulting in a total of 12 concurrent requests.

@mjameswh mjameswh Aug 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also, I'd argue that if a concurrency limit is set both on the store and on the visitor, then should both apply; i.e. the latter doesn't simply override the former.

@mjameswh mjameswh Aug 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In any case, I'd recommend to wait for @jmaeagle99 to chime in before changing your PR. He's the lead on the external storage initiative, so I'd be interested to hear his opinion. He's out today, but should be back tomorrow.

abortSignal,
}: ExternalStorageStoreOptions = {}
): VisitOptions<StoreTarget> {
const runner = new ExternalStorageRunner(externalStorage);
return {
Expand All @@ -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<void> {
const runner = new ExternalStorageRunner(externalStorage);
return {
Expand Down
17 changes: 17 additions & 0 deletions packages/test/src/test-extstore-config.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
90 changes: 89 additions & 1 deletion packages/test/src/test-extstore-visit.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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<void>((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 = {
Expand Down