From ce371f4a5bc958d6d5c3e373665ad3df222de069 Mon Sep 17 00:00:00 2001 From: Jeremy Date: Wed, 15 Jul 2026 10:56:51 -0700 Subject: [PATCH 1/4] feat(client): add WorkflowClient interceptors for list and fetchHistory Wire list and fetchHistory through composeInterceptors so client interceptors can observe or modify those calls, matching describe and Nexus list. Closes #945. --- packages/client/src/interceptors.ts | 21 +++++++++ packages/client/src/workflow-client.ts | 62 ++++++++++++++++++-------- packages/test/src/test-interceptors.ts | 41 +++++++++++++++++ 3 files changed, 105 insertions(+), 19 deletions(-) diff --git a/packages/client/src/interceptors.ts b/packages/client/src/interceptors.ts index eae5e03667..bcc9909e14 100644 --- a/packages/client/src/interceptors.ts +++ b/packages/client/src/interceptors.ts @@ -6,6 +6,7 @@ import type { Duration, SearchAttributePair, TypedSearchAttributes } from '@temporalio/common'; import { Headers, Next } from '@temporalio/common'; +import type { History } from '@temporalio/common/lib/proto-utils'; import type { temporal } from '@temporalio/proto'; import type { NexusOperationHandle } from './nexus-client'; import type { @@ -24,6 +25,7 @@ import type { RequestCancelWorkflowExecutionResponse, TerminateWorkflowExecutionResponse, WorkflowExecution, + WorkflowExecutionInfo, } from './types'; import type { CompiledWorkflowOptions, WorkflowUpdateOptions } from './workflow-options'; import type { ActivityHandle, ActivityOptions } from './activity-client'; @@ -128,6 +130,17 @@ export interface WorkflowDescribeInput { readonly workflowExecution: WorkflowExecution; } +/** Input for WorkflowClientInterceptor.fetchHistory */ +export interface WorkflowFetchHistoryInput { + readonly workflowExecution: WorkflowExecution; +} + +/** Input for WorkflowClientInterceptor.list */ +export interface WorkflowListInput { + readonly query?: string; + readonly pageSize?: number; +} + /** * Implement any of these methods to intercept {@link WorkflowClient} outbound calls * @@ -198,6 +211,14 @@ export interface WorkflowClientInterceptor { * Intercept a service call to describeWorkflowExecution */ describe?: (input: WorkflowDescribeInput, next: Next) => Promise; + /** + * Intercept a service call to getWorkflowExecutionHistory + */ + fetchHistory?: (input: WorkflowFetchHistoryInput, next: Next) => Promise; + /** + * Intercept a service call to listWorkflowExecutions + */ + list?: (input: WorkflowListInput, next: Next) => AsyncIterable; } /** @deprecated: Use {@link WorkflowClientInterceptor} instead */ diff --git a/packages/client/src/workflow-client.ts b/packages/client/src/workflow-client.ts index 6146bcb5b8..37dc79345d 100644 --- a/packages/client/src/workflow-client.ts +++ b/packages/client/src/workflow-client.ts @@ -55,6 +55,8 @@ import type { WorkflowClientInterceptor, WorkflowClientInterceptors, WorkflowDescribeInput, + WorkflowFetchHistoryInput, + WorkflowListInput, WorkflowQueryInput, WorkflowSignalInput, WorkflowSignalWithStartInput, @@ -1442,6 +1444,28 @@ export class WorkflowClient extends BaseClient { } } + /** + * Uses given input to make getWorkflowExecutionHistory call(s) to the service + * + * Used as the final function of the fetchHistory interceptor chain + */ + protected async _fetchHistoryHandler(input: WorkflowFetchHistoryInput): Promise { + let nextPageToken: Uint8Array | undefined = undefined; + const events = Array(); + for (;;) { + const response: temporal.api.workflowservice.v1.GetWorkflowExecutionHistoryResponse = + await this.workflowService.getWorkflowExecutionHistory({ + nextPageToken, + namespace: this.options.namespace, + execution: input.workflowExecution, + }); + events.push(...(response.history?.events ?? [])); + nextPageToken = response.nextPageToken; + if (nextPageToken == null || nextPageToken.length === 0) break; + } + return temporal.api.history.v1.History.create({ events }); + } + /** * Create a new workflow handle for new or existing Workflow execution */ @@ -1530,20 +1554,11 @@ export class WorkflowClient extends BaseClient { }; }, async fetchHistory() { - let nextPageToken: Uint8Array | undefined = undefined; - const events = Array(); - for (;;) { - const response: temporal.api.workflowservice.v1.GetWorkflowExecutionHistoryResponse = - await this.client.workflowService.getWorkflowExecutionHistory({ - nextPageToken, - namespace: this.client.options.namespace, - execution: { workflowId, runId }, - }); - events.push(...(response.history?.events ?? [])); - nextPageToken = response.nextPageToken; - if (nextPageToken == null || nextPageToken.length === 0) break; - } - return temporal.api.history.v1.History.create({ events }); + const next = this.client._fetchHistoryHandler.bind(this.client); + const fn = composeInterceptors(interceptors, 'fetchHistory', next); + return await fn({ + workflowExecution: { workflowId, runId }, + }); }, async startUpdate( def: UpdateDefinition | string, @@ -1627,16 +1642,16 @@ export class WorkflowClient extends BaseClient { }); } - protected async *_list(options?: ListOptions): AsyncIterable { + protected async *_list(input: WorkflowListInput): AsyncIterable { let nextPageToken: Uint8Array = Buffer.alloc(0); for (;;) { let response: temporal.api.workflowservice.v1.ListWorkflowExecutionsResponse; try { response = await this.workflowService.listWorkflowExecutions({ namespace: this.options.namespace, - query: options?.query, + query: input.query, nextPageToken, - pageSize: options?.pageSize, + pageSize: input.pageSize, }); } catch (e) { this.rethrowGrpcError(e, 'Failed to list workflows', undefined); @@ -1661,11 +1676,20 @@ export class WorkflowClient extends BaseClient { * https://docs.temporal.io/visibility */ public list(options?: ListOptions): AsyncWorkflowListIterable { + const input: WorkflowListInput = { + query: options?.query, + pageSize: options?.pageSize, + }; + // Deprecated interceptor factories require a workflowId and are skipped for client-level list. + const interceptors = Array.isArray(this.options.interceptors) + ? (this.options.interceptors as WorkflowClientInterceptor[]) + : []; + const list = composeInterceptors(interceptors, 'list', this._list.bind(this)); return { - [Symbol.asyncIterator]: () => this._list(options)[Symbol.asyncIterator](), + [Symbol.asyncIterator]: () => list(input)[Symbol.asyncIterator](), intoHistories: (intoHistoriesOptions?: IntoHistoriesOptions) => { return mapAsyncIterable( - this._list(options), + list(input), async ({ workflowId, runId }) => ({ workflowId, history: await this.getHandle(workflowId, runId).fetchHistory(), diff --git a/packages/test/src/test-interceptors.ts b/packages/test/src/test-interceptors.ts index 79ce01934a..9c03a1b83d 100644 --- a/packages/test/src/test-interceptors.ts +++ b/packages/test/src/test-interceptors.ts @@ -217,6 +217,47 @@ if (RUN_INTEGRATION_TESTS) { }); }); + test.serial('WorkflowClientInterceptor intercepts list and fetchHistory', async (t) => { + const taskQueue = 'test-interceptor-list-and-fetch-history'; + const workflowId = randomUUID(); + const worker = await Worker.create({ + ...defaultOptions, + taskQueue, + }); + let listCalls = 0; + let fetchHistoryCalls = 0; + const client = new WorkflowClient({ + interceptors: [ + { + list(input, next) { + listCalls += 1; + return next(input); + }, + async fetchHistory(input, next) { + fetchHistoryCalls += 1; + return next(input); + }, + }, + ], + }); + + await worker.runUntil(async () => { + await client.execute(successString, { + taskQueue, + workflowId, + }); + + const history = await client.getHandle(workflowId).fetchHistory(); + t.true((history.events?.length ?? 0) > 0); + t.is(fetchHistoryCalls, 1); + + for await (const _ of client.list({ query: `WorkflowId = "${workflowId}"` })) { + // consume iterator + } + t.is(listCalls, 1); + }); + }); + test.serial('Workflow continueAsNew can be intercepted', async (t) => { const taskQueue = 'test-continue-as-new-interceptor'; const worker = await Worker.create({ From 497d4111d84e9f6a64e0aa5a52b3d1735672b57a Mon Sep 17 00:00:00 2001 From: Jeremy Date: Wed, 15 Jul 2026 11:44:22 -0700 Subject: [PATCH 2/4] feat(client): instrument list and fetchHistory in OpenTelemetry interceptors Keep Required completeness after adding the new client interceptor methods. --- .../src/client/index.ts | 30 +++++++++++++++++++ .../src/workflow/definitions.ts | 10 +++++++ .../src/client/index.ts | 30 +++++++++++++++++++ .../src/workflow/definitions.ts | 10 +++++++ 4 files changed, 80 insertions(+) diff --git a/contrib/interceptors-opentelemetry-v2/src/client/index.ts b/contrib/interceptors-opentelemetry-v2/src/client/index.ts index cd8090b1fa..e2dc316f95 100644 --- a/contrib/interceptors-opentelemetry-v2/src/client/index.ts +++ b/contrib/interceptors-opentelemetry-v2/src/client/index.ts @@ -13,13 +13,18 @@ import type { WorkflowTerminateInput, WorkflowCancelInput, WorkflowDescribeInput, + WorkflowFetchHistoryInput, + WorkflowListInput, WorkflowClientInterceptor, TerminateWorkflowExecutionResponse, RequestCancelWorkflowExecutionResponse, DescribeWorkflowExecutionResponse, + WorkflowExecutionInfo, } from '@temporalio/client'; +import type { History } from '@temporalio/common/lib/proto-utils'; import { instrument, + instrumentSync, headersWithContext, RUN_ID_ATTR_KEY, WORKFLOW_ID_ATTR_KEY, @@ -217,4 +222,29 @@ export class OpenTelemetryWorkflowClientInterceptor implements WorkflowClientInt }, }); } + + async fetchHistory( + input: WorkflowFetchHistoryInput, + next: Next + ): Promise { + return await instrument({ + tracer: this.tracer, + spanName: SpanName.WORKFLOW_FETCH_HISTORY, + fn: async (span) => { + span.setAttribute(WORKFLOW_ID_ATTR_KEY, input.workflowExecution.workflowId); + if (input.workflowExecution.runId) { + span.setAttribute(RUN_ID_ATTR_KEY, input.workflowExecution.runId); + } + return await next(input); + }, + }); + } + + list(input: WorkflowListInput, next: Next): AsyncIterable { + return instrumentSync({ + tracer: this.tracer, + spanName: SpanName.WORKFLOW_LIST, + fn: () => next(input), + }); + } } diff --git a/contrib/interceptors-opentelemetry-v2/src/workflow/definitions.ts b/contrib/interceptors-opentelemetry-v2/src/workflow/definitions.ts index e56fadfc6b..f82b28c0a7 100644 --- a/contrib/interceptors-opentelemetry-v2/src/workflow/definitions.ts +++ b/contrib/interceptors-opentelemetry-v2/src/workflow/definitions.ts @@ -108,6 +108,16 @@ export enum SpanName { */ WORKFLOW_DESCRIBE = 'DescribeWorkflow', + /** + * Workflow history is fetched + */ + WORKFLOW_FETCH_HISTORY = 'FetchWorkflowHistory', + + /** + * Workflows are listed + */ + WORKFLOW_LIST = 'ListWorkflows', + /** * Workflow run is executing */ diff --git a/contrib/interceptors-opentelemetry/src/client/index.ts b/contrib/interceptors-opentelemetry/src/client/index.ts index 3e78488d27..9d1a9be9f3 100644 --- a/contrib/interceptors-opentelemetry/src/client/index.ts +++ b/contrib/interceptors-opentelemetry/src/client/index.ts @@ -13,13 +13,18 @@ import type { WorkflowTerminateInput, WorkflowCancelInput, WorkflowDescribeInput, + WorkflowFetchHistoryInput, + WorkflowListInput, WorkflowClientInterceptor, TerminateWorkflowExecutionResponse, RequestCancelWorkflowExecutionResponse, DescribeWorkflowExecutionResponse, + WorkflowExecutionInfo, } from '@temporalio/client'; +import type { History } from '@temporalio/common/lib/proto-utils'; import { instrument, + instrumentSync, headersWithContext, RUN_ID_ATTR_KEY, WORKFLOW_ID_ATTR_KEY, @@ -215,4 +220,29 @@ export class OpenTelemetryWorkflowClientInterceptor implements WorkflowClientInt }, }); } + + async fetchHistory( + input: WorkflowFetchHistoryInput, + next: Next + ): Promise { + return await instrument({ + tracer: this.tracer, + spanName: SpanName.WORKFLOW_FETCH_HISTORY, + fn: async (span) => { + span.setAttribute(WORKFLOW_ID_ATTR_KEY, input.workflowExecution.workflowId); + if (input.workflowExecution.runId) { + span.setAttribute(RUN_ID_ATTR_KEY, input.workflowExecution.runId); + } + return await next(input); + }, + }); + } + + list(input: WorkflowListInput, next: Next): AsyncIterable { + return instrumentSync({ + tracer: this.tracer, + spanName: SpanName.WORKFLOW_LIST, + fn: () => next(input), + }); + } } diff --git a/contrib/interceptors-opentelemetry/src/workflow/definitions.ts b/contrib/interceptors-opentelemetry/src/workflow/definitions.ts index 73c3a23b8c..4dd77ade58 100644 --- a/contrib/interceptors-opentelemetry/src/workflow/definitions.ts +++ b/contrib/interceptors-opentelemetry/src/workflow/definitions.ts @@ -109,6 +109,16 @@ export enum SpanName { */ WORKFLOW_DESCRIBE = 'DescribeWorkflow', + /** + * Workflow history is fetched + */ + WORKFLOW_FETCH_HISTORY = 'FetchWorkflowHistory', + + /** + * Workflows are listed + */ + WORKFLOW_LIST = 'ListWorkflows', + /** * Workflow run is executing */ From a0086572dda8515b47d6ed3d23fbc3cfaad94876 Mon Sep 17 00:00:00 2001 From: Jeremy Date: Fri, 14 Aug 2026 13:50:02 -0700 Subject: [PATCH 3/4] feat(client): pin list interceptor lifecycle across pagination Treat one list() consumption as a single intercepted operation, including early break and intoHistories composition. Add experimental TypeDoc, replace sync OTel list instrumentation with instrumentAsyncIterable, and close mapAsyncIterable sources on consumer termination so interceptor/OTel finally blocks run. --- .../test-instrument-async-iterable.ts | 184 ++++++++++++++++++ .../src/client/index.ts | 4 +- .../src/instrumentation.ts | 100 ++++++++++ .../test-instrument-async-iterable.ts | 184 ++++++++++++++++++ .../src/client/index.ts | 4 +- .../src/instrumentation.ts | 100 ++++++++++ packages/client/src/interceptors.ts | 19 +- packages/client/src/iterators-utils.ts | 15 +- packages/client/src/workflow-client.ts | 5 +- .../src/test-interceptors.cloud-pending.ts | 145 +++++++++++++- packages/test/src/test-iterators-utils.ts | 29 +++ 11 files changed, 778 insertions(+), 11 deletions(-) create mode 100644 contrib/interceptors-opentelemetry-v2/src/__tests__/test-instrument-async-iterable.ts create mode 100644 contrib/interceptors-opentelemetry/src/__tests__/test-instrument-async-iterable.ts diff --git a/contrib/interceptors-opentelemetry-v2/src/__tests__/test-instrument-async-iterable.ts b/contrib/interceptors-opentelemetry-v2/src/__tests__/test-instrument-async-iterable.ts new file mode 100644 index 0000000000..3a42691e16 --- /dev/null +++ b/contrib/interceptors-opentelemetry-v2/src/__tests__/test-instrument-async-iterable.ts @@ -0,0 +1,184 @@ +import test from 'ava'; +import * as otel from '@opentelemetry/api'; +import { SpanStatusCode } from '@opentelemetry/api'; +import { BasicTracerProvider, InMemorySpanExporter, SimpleSpanProcessor } from '@opentelemetry/sdk-trace-base'; +import { instrumentAsyncIterable } from '../instrumentation'; + +/** + * Minimal sync ContextManager so unit tests can assert active-context propagation + * without adding @opentelemetry/context-async-hooks as a dependency. + */ +class TestContextManager implements otel.ContextManager { + private current: otel.Context = otel.ROOT_CONTEXT; + + active(): otel.Context { + return this.current; + } + + with ReturnType>( + context: otel.Context, + fn: F, + thisArg?: ThisParameterType, + ...args: A + ): ReturnType { + const previous = this.current; + this.current = context; + try { + return Reflect.apply(fn, thisArg, args); + } finally { + this.current = previous; + } + } + + bind(context: otel.Context, target: T): T { + if (typeof target !== 'function') { + return target; + } + // eslint-disable-next-line @typescript-eslint/no-this-alias -- bind() needs the manager instance in the wrapper + const manager = this; + const bound = function (this: unknown, ...args: unknown[]) { + return manager.with(context, () => (target as (...args: unknown[]) => unknown).apply(this, args)); + }; + return bound as T; + } + + enable(): this { + return this; + } + + disable(): this { + this.current = otel.ROOT_CONTEXT; + return this; + } +} + +function setupTracer(name: string) { + const memoryExporter = new InMemorySpanExporter(); + const provider = new BasicTracerProvider({ + spanProcessors: [new SimpleSpanProcessor(memoryExporter)], + }); + otel.context.setGlobalContextManager(new TestContextManager().enable()); + return { + memoryExporter, + tracer: provider.getTracer(name), + }; +} + +async function* values(items: number[]): AsyncIterable { + for (const item of items) { + yield item; + } +} + +test('instrumentAsyncIterable does not open a span until iteration begins', async (t) => { + const { memoryExporter, tracer } = setupTracer('lazy-start'); + const iterable = instrumentAsyncIterable({ + tracer, + spanName: 'ListWorkflows', + fn: () => values([1, 2, 3]), + }); + + t.is(memoryExporter.getFinishedSpans().length, 0); + const iterator = iterable[Symbol.asyncIterator](); + t.is(memoryExporter.getFinishedSpans().length, 0); + + t.is((await iterator.next()).value, 1); + t.is(memoryExporter.getFinishedSpans().length, 0); + + t.is((await iterator.next()).value, 2); + t.is((await iterator.next()).value, 3); + t.true((await iterator.next()).done); + + const spans = memoryExporter.getFinishedSpans(); + t.is(spans.length, 1); + t.is(spans[0]!.name, 'ListWorkflows'); + t.is(spans[0]!.status.code, SpanStatusCode.OK); +}); + +test('instrumentAsyncIterable ends the span exactly once on early break', async (t) => { + const { memoryExporter, tracer } = setupTracer('early-break'); + const iterable = instrumentAsyncIterable({ + tracer, + spanName: 'ListWorkflows', + fn: () => values([1, 2, 3, 4]), + }); + + const seen: number[] = []; + for await (const item of iterable) { + seen.push(item); + if (item === 2) { + t.is(memoryExporter.getFinishedSpans().length, 0); + break; + } + } + + t.deepEqual(seen, [1, 2]); + const spans = memoryExporter.getFinishedSpans(); + t.is(spans.length, 1); + t.is(spans[0]!.status.code, SpanStatusCode.OK); +}); + +test('instrumentAsyncIterable records errors and ends once', async (t) => { + const { memoryExporter, tracer } = setupTracer('error'); + const error = new Error('downstream failed'); + async function* boom(): AsyncIterable { + yield 1; + throw error; + } + + const iterable = instrumentAsyncIterable({ + tracer, + spanName: 'ListWorkflows', + fn: () => boom(), + }); + + await t.throwsAsync( + async () => { + for await (const _ of iterable) { + // consume until error + } + }, + { is: error } + ); + + const spans = memoryExporter.getFinishedSpans(); + t.is(spans.length, 1); + t.is(spans[0]!.status.code, SpanStatusCode.ERROR); + t.is(spans[0]!.status.message, error.message); + t.is(spans[0]!.events.filter((event) => event.name === 'exception').length, 1); +}); + +test('instrumentAsyncIterable propagates active context to downstream next()', async (t) => { + const { memoryExporter, tracer } = setupTracer('context'); + const iterable = instrumentAsyncIterable({ + tracer, + spanName: 'ListWorkflows', + fn: () => ({ + [Symbol.asyncIterator]() { + let done = false; + return { + async next(): Promise> { + if (done) { + return { done: true, value: undefined }; + } + done = true; + const child = tracer.startSpan('child-during-next'); + child.end(); + return { done: false, value: 1 }; + }, + }; + }, + }), + }); + + for await (const _ of iterable) { + // consume + } + + const spans = memoryExporter.getFinishedSpans(); + const parent = spans.find((span) => span.name === 'ListWorkflows'); + const child = spans.find((span) => span.name === 'child-during-next'); + t.truthy(parent); + t.truthy(child); + t.is(child!.parentSpanContext?.spanId, parent!.spanContext().spanId); +}); diff --git a/contrib/interceptors-opentelemetry-v2/src/client/index.ts b/contrib/interceptors-opentelemetry-v2/src/client/index.ts index e2dc316f95..f6807b53ed 100644 --- a/contrib/interceptors-opentelemetry-v2/src/client/index.ts +++ b/contrib/interceptors-opentelemetry-v2/src/client/index.ts @@ -24,7 +24,7 @@ import type { import type { History } from '@temporalio/common/lib/proto-utils'; import { instrument, - instrumentSync, + instrumentAsyncIterable, headersWithContext, RUN_ID_ATTR_KEY, WORKFLOW_ID_ATTR_KEY, @@ -241,7 +241,7 @@ export class OpenTelemetryWorkflowClientInterceptor implements WorkflowClientInt } list(input: WorkflowListInput, next: Next): AsyncIterable { - return instrumentSync({ + return instrumentAsyncIterable({ tracer: this.tracer, spanName: SpanName.WORKFLOW_LIST, fn: () => next(input), diff --git a/contrib/interceptors-opentelemetry-v2/src/instrumentation.ts b/contrib/interceptors-opentelemetry-v2/src/instrumentation.ts index af6a3bd4dc..3e7aa5ef93 100644 --- a/contrib/interceptors-opentelemetry-v2/src/instrumentation.ts +++ b/contrib/interceptors-opentelemetry-v2/src/instrumentation.ts @@ -130,6 +130,10 @@ export interface InstrumentOptions { export type InstrumentOptionsSync = Omit, 'fn'> & { fn: (span: otel.Span) => T }; +export type InstrumentOptionsAsyncIterable = Omit, 'fn'> & { + fn: (span: otel.Span) => AsyncIterable; +}; + /** * Wraps `fn` in a span which ends when function returns or throws */ @@ -157,3 +161,99 @@ export function instrumentSync({ tracer, spanName, fn, context, acceptableErr } return tracer.startActiveSpan(spanName, (span) => wrapWithSpanSync(span, fn, acceptableErrors)); } + +/** + * Wraps an async iterable in a span whose lifetime matches iteration. + * + * The returned iterable is lazy: creating it does not open a span. The span starts on the first + * `next()` call, remains open while the iterator is in use, and ends on completion, early + * termination (`return`), or error. + */ +export function instrumentAsyncIterable({ + tracer, + spanName, + fn, + context, + acceptableErrors, +}: InstrumentOptionsAsyncIterable): AsyncIterable { + return { + [Symbol.asyncIterator](): AsyncIterator { + let span: otel.Span | undefined; + let spanContext: otel.Context | undefined; + let iterator: AsyncIterator | undefined; + let finished = false; + + const finish = (err?: unknown): void => { + if (finished || span === undefined) { + return; + } + finished = true; + if (err !== undefined) { + maybeAddErrorToSpan(err, span, acceptableErrors); + } else { + span.setStatus({ code: otel.SpanStatusCode.OK }); + } + span.end(); + }; + + const ensureStarted = (): void => { + if (iterator !== undefined) { + return; + } + const parentContext = context ?? otel.context.active(); + span = tracer.startSpan(spanName, undefined, parentContext); + spanContext = otel.trace.setSpan(parentContext, span); + const iterable = otel.context.with(spanContext, () => fn(span!)); + iterator = iterable[Symbol.asyncIterator](); + }; + + return { + async next(...args: [] | [undefined]): Promise> { + try { + ensureStarted(); + const result = await otel.context.with(spanContext!, async () => iterator!.next(...args)); + if (result.done) { + finish(); + } + return result; + } catch (err) { + finish(err); + throw err; + } + }, + async return(value?: unknown): Promise> { + if (iterator === undefined) { + return { done: true, value: undefined as any }; + } + try { + const result = iterator.return + ? await otel.context.with(spanContext!, async () => iterator!.return!(value)) + : ({ done: true, value: undefined } as IteratorResult); + finish(); + return result; + } catch (err) { + finish(err); + throw err; + } + }, + async throw(err?: unknown): Promise> { + try { + ensureStarted(); + if (iterator?.throw) { + const result = await otel.context.with(spanContext!, async () => iterator!.throw!(err)); + if (result.done) { + finish(); + } + return result; + } + finish(err); + throw err; + } catch (e) { + finish(e); + throw e; + } + }, + }; + }, + }; +} diff --git a/contrib/interceptors-opentelemetry/src/__tests__/test-instrument-async-iterable.ts b/contrib/interceptors-opentelemetry/src/__tests__/test-instrument-async-iterable.ts new file mode 100644 index 0000000000..b8aa27988c --- /dev/null +++ b/contrib/interceptors-opentelemetry/src/__tests__/test-instrument-async-iterable.ts @@ -0,0 +1,184 @@ +import test from 'ava'; +import * as otel from '@opentelemetry/api'; +import { SpanStatusCode } from '@opentelemetry/api'; +import { BasicTracerProvider, InMemorySpanExporter, SimpleSpanProcessor } from '@opentelemetry/sdk-trace-base'; +import { instrumentAsyncIterable } from '../instrumentation'; + +/** + * Minimal sync ContextManager so unit tests can assert active-context propagation + * without adding @opentelemetry/context-async-hooks as a dependency. + */ +class TestContextManager implements otel.ContextManager { + private current: otel.Context = otel.ROOT_CONTEXT; + + active(): otel.Context { + return this.current; + } + + with ReturnType>( + context: otel.Context, + fn: F, + thisArg?: ThisParameterType, + ...args: A + ): ReturnType { + const previous = this.current; + this.current = context; + try { + return Reflect.apply(fn, thisArg, args); + } finally { + this.current = previous; + } + } + + bind(context: otel.Context, target: T): T { + if (typeof target !== 'function') { + return target; + } + // eslint-disable-next-line @typescript-eslint/no-this-alias -- bind() needs the manager instance in the wrapper + const manager = this; + const bound = function (this: unknown, ...args: unknown[]) { + return manager.with(context, () => (target as (...args: unknown[]) => unknown).apply(this, args)); + }; + return bound as T; + } + + enable(): this { + return this; + } + + disable(): this { + this.current = otel.ROOT_CONTEXT; + return this; + } +} + +function setupTracer(name: string) { + const memoryExporter = new InMemorySpanExporter(); + const provider = new BasicTracerProvider(); + provider.addSpanProcessor(new SimpleSpanProcessor(memoryExporter)); + otel.context.setGlobalContextManager(new TestContextManager().enable()); + provider.register(); + return { + memoryExporter, + tracer: provider.getTracer(name), + }; +} + +async function* values(items: number[]): AsyncIterable { + for (const item of items) { + yield item; + } +} + +test('instrumentAsyncIterable does not open a span until iteration begins', async (t) => { + const { memoryExporter, tracer } = setupTracer('lazy-start'); + const iterable = instrumentAsyncIterable({ + tracer, + spanName: 'ListWorkflows', + fn: () => values([1, 2, 3]), + }); + + t.is(memoryExporter.getFinishedSpans().length, 0); + const iterator = iterable[Symbol.asyncIterator](); + t.is(memoryExporter.getFinishedSpans().length, 0); + + t.is((await iterator.next()).value, 1); + t.is(memoryExporter.getFinishedSpans().length, 0); + + t.is((await iterator.next()).value, 2); + t.is((await iterator.next()).value, 3); + t.true((await iterator.next()).done); + + const spans = memoryExporter.getFinishedSpans(); + t.is(spans.length, 1); + t.is(spans[0]!.name, 'ListWorkflows'); + t.is(spans[0]!.status.code, SpanStatusCode.OK); +}); + +test('instrumentAsyncIterable ends the span exactly once on early break', async (t) => { + const { memoryExporter, tracer } = setupTracer('early-break'); + const iterable = instrumentAsyncIterable({ + tracer, + spanName: 'ListWorkflows', + fn: () => values([1, 2, 3, 4]), + }); + + const seen: number[] = []; + for await (const item of iterable) { + seen.push(item); + if (item === 2) { + t.is(memoryExporter.getFinishedSpans().length, 0); + break; + } + } + + t.deepEqual(seen, [1, 2]); + const spans = memoryExporter.getFinishedSpans(); + t.is(spans.length, 1); + t.is(spans[0]!.status.code, SpanStatusCode.OK); +}); + +test('instrumentAsyncIterable records errors and ends once', async (t) => { + const { memoryExporter, tracer } = setupTracer('error'); + const error = new Error('downstream failed'); + async function* boom(): AsyncIterable { + yield 1; + throw error; + } + + const iterable = instrumentAsyncIterable({ + tracer, + spanName: 'ListWorkflows', + fn: () => boom(), + }); + + await t.throwsAsync( + async () => { + for await (const _ of iterable) { + // consume until error + } + }, + { is: error } + ); + + const spans = memoryExporter.getFinishedSpans(); + t.is(spans.length, 1); + t.is(spans[0]!.status.code, SpanStatusCode.ERROR); + t.is(spans[0]!.status.message, error.message); + t.is(spans[0]!.events.filter((event) => event.name === 'exception').length, 1); +}); + +test('instrumentAsyncIterable propagates active context to downstream next()', async (t) => { + const { memoryExporter, tracer } = setupTracer('context'); + const iterable = instrumentAsyncIterable({ + tracer, + spanName: 'ListWorkflows', + fn: () => ({ + [Symbol.asyncIterator]() { + let done = false; + return { + async next(): Promise> { + if (done) { + return { done: true, value: undefined }; + } + done = true; + const child = tracer.startSpan('child-during-next'); + child.end(); + return { done: false, value: 1 }; + }, + }; + }, + }), + }); + + for await (const _ of iterable) { + // consume + } + + const spans = memoryExporter.getFinishedSpans(); + const parent = spans.find((span) => span.name === 'ListWorkflows'); + const child = spans.find((span) => span.name === 'child-during-next'); + t.truthy(parent); + t.truthy(child); + t.is(child!.parentSpanId, parent!.spanContext().spanId); +}); diff --git a/contrib/interceptors-opentelemetry/src/client/index.ts b/contrib/interceptors-opentelemetry/src/client/index.ts index 9d1a9be9f3..3c7e8d7531 100644 --- a/contrib/interceptors-opentelemetry/src/client/index.ts +++ b/contrib/interceptors-opentelemetry/src/client/index.ts @@ -24,7 +24,7 @@ import type { import type { History } from '@temporalio/common/lib/proto-utils'; import { instrument, - instrumentSync, + instrumentAsyncIterable, headersWithContext, RUN_ID_ATTR_KEY, WORKFLOW_ID_ATTR_KEY, @@ -239,7 +239,7 @@ export class OpenTelemetryWorkflowClientInterceptor implements WorkflowClientInt } list(input: WorkflowListInput, next: Next): AsyncIterable { - return instrumentSync({ + return instrumentAsyncIterable({ tracer: this.tracer, spanName: SpanName.WORKFLOW_LIST, fn: () => next(input), diff --git a/contrib/interceptors-opentelemetry/src/instrumentation.ts b/contrib/interceptors-opentelemetry/src/instrumentation.ts index 21a3d3cecc..c5430988da 100644 --- a/contrib/interceptors-opentelemetry/src/instrumentation.ts +++ b/contrib/interceptors-opentelemetry/src/instrumentation.ts @@ -130,6 +130,10 @@ export interface InstrumentOptions { export type InstrumentOptionsSync = Omit, 'fn'> & { fn: (span: otel.Span) => T }; +export type InstrumentOptionsAsyncIterable = Omit, 'fn'> & { + fn: (span: otel.Span) => AsyncIterable; +}; + /** * Wraps `fn` in a span which ends when function returns or throws */ @@ -156,3 +160,99 @@ export function instrumentSync({ tracer, spanName, fn, context, acceptableErr } return tracer.startActiveSpan(spanName, (span) => wrapWithSpanSync(span, fn, acceptableErrors)); } + +/** + * Wraps an async iterable in a span whose lifetime matches iteration. + * + * The returned iterable is lazy: creating it does not open a span. The span starts on the first + * `next()` call, remains open while the iterator is in use, and ends on completion, early + * termination (`return`), or error. + */ +export function instrumentAsyncIterable({ + tracer, + spanName, + fn, + context, + acceptableErrors, +}: InstrumentOptionsAsyncIterable): AsyncIterable { + return { + [Symbol.asyncIterator](): AsyncIterator { + let span: otel.Span | undefined; + let spanContext: otel.Context | undefined; + let iterator: AsyncIterator | undefined; + let finished = false; + + const finish = (err?: unknown): void => { + if (finished || span === undefined) { + return; + } + finished = true; + if (err !== undefined) { + maybeAddErrorToSpan(err, span, acceptableErrors); + } else { + span.setStatus({ code: otel.SpanStatusCode.OK }); + } + span.end(); + }; + + const ensureStarted = (): void => { + if (iterator !== undefined) { + return; + } + const parentContext = context ?? otel.context.active(); + span = tracer.startSpan(spanName, undefined, parentContext); + spanContext = otel.trace.setSpan(parentContext, span); + const iterable = otel.context.with(spanContext, () => fn(span!)); + iterator = iterable[Symbol.asyncIterator](); + }; + + return { + async next(...args: [] | [undefined]): Promise> { + try { + ensureStarted(); + const result = await otel.context.with(spanContext!, async () => iterator!.next(...args)); + if (result.done) { + finish(); + } + return result; + } catch (err) { + finish(err); + throw err; + } + }, + async return(value?: unknown): Promise> { + if (iterator === undefined) { + return { done: true, value: undefined as any }; + } + try { + const result = iterator.return + ? await otel.context.with(spanContext!, async () => iterator!.return!(value)) + : ({ done: true, value: undefined } as IteratorResult); + finish(); + return result; + } catch (err) { + finish(err); + throw err; + } + }, + async throw(err?: unknown): Promise> { + try { + ensureStarted(); + if (iterator?.throw) { + const result = await otel.context.with(spanContext!, async () => iterator!.throw!(err)); + if (result.done) { + finish(); + } + return result; + } + finish(err); + throw err; + } catch (e) { + finish(e); + throw e; + } + }, + }; + }, + }; +} diff --git a/packages/client/src/interceptors.ts b/packages/client/src/interceptors.ts index bcc9909e14..b7e83f24ca 100644 --- a/packages/client/src/interceptors.ts +++ b/packages/client/src/interceptors.ts @@ -130,12 +130,20 @@ export interface WorkflowDescribeInput { readonly workflowExecution: WorkflowExecution; } -/** Input for WorkflowClientInterceptor.fetchHistory */ +/** + * Input for WorkflowClientInterceptor.fetchHistory + * + * @experimental This interceptor input type is experimental - API changes are still possible. + */ export interface WorkflowFetchHistoryInput { readonly workflowExecution: WorkflowExecution; } -/** Input for WorkflowClientInterceptor.list */ +/** + * Input for WorkflowClientInterceptor.list + * + * @experimental This interceptor input type is experimental - API changes are still possible. + */ export interface WorkflowListInput { readonly query?: string; readonly pageSize?: number; @@ -213,10 +221,17 @@ export interface WorkflowClientInterceptor { describe?: (input: WorkflowDescribeInput, next: Next) => Promise; /** * Intercept a service call to getWorkflowExecutionHistory + * + * @experimental This interceptor method is experimental - API changes are still possible. */ fetchHistory?: (input: WorkflowFetchHistoryInput, next: Next) => Promise; /** * Intercept a service call to listWorkflowExecutions + * + * One interception spans a single consumption of the returned async iterable, including + * pagination across multiple RPCs and early consumer termination. + * + * @experimental This interceptor method is experimental - API changes are still possible. */ list?: (input: WorkflowListInput, next: Next) => AsyncIterable; } diff --git a/packages/client/src/iterators-utils.ts b/packages/client/src/iterators-utils.ts index 13fbcf6d07..575d8de916 100644 --- a/packages/client/src/iterators-utils.ts +++ b/packages/client/src/iterators-utils.ts @@ -108,9 +108,18 @@ export async function* mapAsyncIterable( yield res; } } catch (err: unknown) { - if (isAbortError(err)) { - return; + if (!isAbortError(err)) { + throw err; } - throw err; + } finally { + // Stop producers and close the source iterable so interceptor/OTel lifecycles end on + // early consumer termination (break/return), not only after the source is exhausted. + controller.abort(); + try { + await sourceIterator.return?.(); + } catch { + // Ignore cleanup errors from an already-failed or exhausted source. + } + await Promise.allSettled(mappers); } } diff --git a/packages/client/src/workflow-client.ts b/packages/client/src/workflow-client.ts index 7cf139abe9..705eea31f3 100644 --- a/packages/client/src/workflow-client.ts +++ b/packages/client/src/workflow-client.ts @@ -1881,7 +1881,10 @@ export class WorkflowClient extends BaseClient { workflowId, history: await this.getHandle(workflowId, runId).fetchHistory(), }), - { concurrency: intoHistoriesOptions?.concurrency ?? 5 } + { + concurrency: intoHistoriesOptions?.concurrency ?? 5, + bufferLimit: intoHistoriesOptions?.bufferLimit, + } ); }, }; diff --git a/packages/test/src/test-interceptors.cloud-pending.ts b/packages/test/src/test-interceptors.cloud-pending.ts index 9c03a1b83d..45e976ba97 100644 --- a/packages/test/src/test-interceptors.cloud-pending.ts +++ b/packages/test/src/test-interceptors.cloud-pending.ts @@ -13,7 +13,7 @@ import { ApplicationFailure, TerminatedFailure } from '@temporalio/common'; import { DefaultLogger, Runtime } from '@temporalio/worker'; import type { WorkflowInfo } from '@temporalio/workflow'; import { defaultPayloadConverter } from '@temporalio/workflow'; -import { isBun, cleanOptionalStackTrace, compareStackTrace, RUN_INTEGRATION_TESTS, Worker } from './helpers'; +import { isBun, cleanOptionalStackTrace, compareStackTrace, RUN_INTEGRATION_TESTS, Worker, waitUntil } from './helpers'; import { defaultOptions } from './mock-native-worker'; import { checkDisposeRan, @@ -258,6 +258,149 @@ if (RUN_INTEGRATION_TESTS) { }); }); + test.serial('WorkflowClientInterceptor list spans multi-page consumption until early break', async (t) => { + const taskQueue = 'test-interceptor-list-pagination'; + const idPrefix = `list-pag-${randomUUID()}`; + const query = `WorkflowId STARTS_WITH "${idPrefix}"`; + const seeded = 10; + const worker = await Worker.create({ + ...defaultOptions, + taskQueue, + }); + + let listEntered = 0; + let listExited = 0; + let interceptedItems = 0; + const client = new WorkflowClient({ + interceptors: [ + { + async *list(input, next) { + listEntered += 1; + try { + for await (const execution of next(input)) { + interceptedItems += 1; + yield execution; + } + } finally { + listExited += 1; + } + }, + }, + ], + }); + const visibilityClient = new WorkflowClient(); + + await worker.runUntil(async () => { + await Promise.all( + Array.from({ length: seeded }, (_, i) => + client.execute(successString, { + taskQueue, + workflowId: `${idPrefix}-${i}`, + }) + ) + ); + + await waitUntil(async () => { + let count = 0; + for await (const _ of visibilityClient.list({ query })) { + count += 1; + } + return count >= seeded; + }, 30_000); + + t.is(listEntered, 0); + + const consumed = []; + for await (const execution of client.list({ query, pageSize: 2 })) { + consumed.push(execution); + if (consumed.length === 6) { + t.is(listEntered, 1); + t.is(listExited, 0); + break; + } + } + + t.is(consumed.length, 6); + t.is(interceptedItems, 6); + t.is(listEntered, 1); + t.is(listExited, 1); + t.true(seeded > consumed.length); + }); + }); + + test.serial('WorkflowClientInterceptor list().intoHistories() composes list and fetchHistory', async (t) => { + const taskQueue = 'test-interceptor-list-into-histories'; + const idPrefix = `list-hist-${randomUUID()}`; + const query = `WorkflowId STARTS_WITH "${idPrefix}"`; + const seeded = 10; + const worker = await Worker.create({ + ...defaultOptions, + taskQueue, + }); + + let listCalls = 0; + let listExited = 0; + let fetchHistoryCalls = 0; + const client = new WorkflowClient({ + interceptors: [ + { + async *list(input, next) { + listCalls += 1; + try { + for await (const execution of next(input)) { + yield execution; + } + } finally { + listExited += 1; + } + }, + async fetchHistory(input, next) { + fetchHistoryCalls += 1; + return next(input); + }, + }, + ], + }); + const visibilityClient = new WorkflowClient(); + + await worker.runUntil(async () => { + await Promise.all( + Array.from({ length: seeded }, (_, i) => + client.execute(successString, { + taskQueue, + workflowId: `${idPrefix}-${i}`, + }) + ) + ); + + await waitUntil(async () => { + let count = 0; + for await (const _ of visibilityClient.list({ query })) { + count += 1; + } + return count >= seeded; + }, 30_000); + + let histories = 0; + for await (const { history } of client.list({ query, pageSize: 2 }).intoHistories({ + concurrency: 2, + })) { + const events = (history as { events?: unknown[] } | undefined)?.events; + t.true((events?.length ?? 0) > 0); + histories += 1; + if (histories >= 7) { + break; + } + } + + t.is(histories, 7); + t.is(listCalls, 1); + t.is(listExited, 1); + t.true(fetchHistoryCalls > 6); + t.true(seeded > histories); + }); + }); + test.serial('Workflow continueAsNew can be intercepted', async (t) => { const taskQueue = 'test-continue-as-new-interceptor'; const worker = await Worker.create({ diff --git a/packages/test/src/test-iterators-utils.ts b/packages/test/src/test-iterators-utils.ts index eb16f6f4b0..f6ce00a9c0 100644 --- a/packages/test/src/test-iterators-utils.ts +++ b/packages/test/src/test-iterators-utils.ts @@ -123,6 +123,35 @@ test(`mapAsyncIterable (with concurrency) doesn't consume more input than requir t.is(counter, 15); }); +test(`mapAsyncIterable (with concurrency) closes the source iterable on early termination`, async (t) => { + let sourceExited = 0; + let produced = 0; + + async function* source(): AsyncIterable { + try { + for (;;) { + yield ++produced; + } + } finally { + sourceExited += 1; + } + } + + const iterable = mapAsyncIterable(source(), sleepThatTime, { concurrency: 3 }); + const seen: number[] = []; + for await (const value of iterable) { + seen.push(value); + if (seen.length === 4) { + break; + } + } + + t.is(seen.length, 4); + t.is(sourceExited, 1); + // Producers may have pulled a few extra source items for in-flight work, but must stop after close. + t.true(produced < 20); +}); + test(`mapAsyncIterable (with concurrency) doesn't hang on source exceptions`, async (t) => { async function* name(): AsyncIterable { for (;;) { From 3bf152fab7383f7738be7a02c5baee9b7cce742a Mon Sep 17 00:00:00 2001 From: Jeremy Date: Fri, 14 Aug 2026 15:07:09 -0700 Subject: [PATCH 4/4] fix(client): make intoHistories list interception lazy Defer composed list interceptors until iterable consumption, tighten concurrent mapAsyncIterable early-close/error wakeups, and add the required Unreleased changelog entry for the experimental API. --- CHANGELOG.md | 2 + .../test-instrument-async-iterable.ts | 5 ++ .../test-instrument-async-iterable.ts | 5 ++ packages/client/src/iterators-utils.ts | 50 ++++++++--- packages/client/src/workflow-client.ts | 7 +- .../src/test-interceptors.cloud-pending.ts | 36 +++++--- packages/test/src/test-iterators-utils.ts | 86 ++++++++++++++++++- 7 files changed, 162 insertions(+), 29 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index cf9cd90cb8..85563007c1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,8 @@ to docs, or any other relevant information. ### Added +- **Experimental**: Added `list` and `fetchHistory` interception to `WorkflowClientInterceptor`. + `list` interception spans one lazy iterable consumption across pagination and early termination. - **Experimental**: Workflow Clients can now use `TypeInfo` to encode Workflow inputs and decode Workflow results. - **Experimental**: `@temporalio/google-adk-agents` package for running Google ADK agents as durable Temporal Workflows. ADK's OpenTelemetry agent-loop spans can be exported replay-safely from the Workflow sandbox by composing with diff --git a/contrib/interceptors-opentelemetry-v2/src/__tests__/test-instrument-async-iterable.ts b/contrib/interceptors-opentelemetry-v2/src/__tests__/test-instrument-async-iterable.ts index 3a42691e16..088755f6e0 100644 --- a/contrib/interceptors-opentelemetry-v2/src/__tests__/test-instrument-async-iterable.ts +++ b/contrib/interceptors-opentelemetry-v2/src/__tests__/test-instrument-async-iterable.ts @@ -64,6 +64,11 @@ function setupTracer(name: string) { }; } +test.afterEach.always(() => { + otel.trace.disable(); + otel.context.disable(); +}); + async function* values(items: number[]): AsyncIterable { for (const item of items) { yield item; diff --git a/contrib/interceptors-opentelemetry/src/__tests__/test-instrument-async-iterable.ts b/contrib/interceptors-opentelemetry/src/__tests__/test-instrument-async-iterable.ts index b8aa27988c..24aa706a79 100644 --- a/contrib/interceptors-opentelemetry/src/__tests__/test-instrument-async-iterable.ts +++ b/contrib/interceptors-opentelemetry/src/__tests__/test-instrument-async-iterable.ts @@ -64,6 +64,11 @@ function setupTracer(name: string) { }; } +test.afterEach.always(() => { + otel.trace.disable(); + otel.context.disable(); +}); + async function* values(items: number[]): AsyncIterable { for (const item of items) { yield item; diff --git a/packages/client/src/iterators-utils.ts b/packages/client/src/iterators-utils.ts index 575d8de916..f93ea7dad3 100644 --- a/packages/client/src/iterators-utils.ts +++ b/packages/client/src/iterators-utils.ts @@ -57,6 +57,8 @@ export async function* mapAsyncIterable( const controller = new AbortController(); const emitterEventsIterable = on(emitter, 'result', { signal: controller.signal }) as AsyncIterable<[B]>; const emitterError: Promise = once(emitter, 'error'); + let sourceExhausted = false; + let stopped = false; const bufferLimitSemaphore = typeof bufferLimit === 'number' @@ -67,11 +69,13 @@ export async function* mapAsyncIterable( let value = bufferLimit + concurrency; return { - acquire: async () => { + acquire: async (): Promise => { while (value <= 0) { - await Promise.race([releaseEvents.next(), emitterError]); + const result = await Promise.race([releaseEvents.next(), emitterError]); + if (Array.isArray(result) || result.done) return false; } value--; + return true; }, release: () => { value++; @@ -83,11 +87,16 @@ export async function* mapAsyncIterable( const mapper = async () => { for (;;) { - await bufferLimitSemaphore?.acquire(); + if (stopped) return; + if (bufferLimitSemaphore && !(await bufferLimitSemaphore.acquire())) return; const val = await Promise.race([sourceIterator.next(), emitterError]); if (Array.isArray(val)) return; - if ((val as IteratorResult<[B]>)?.done) return; + if (stopped) return; + if ((val as IteratorResult<[B]>)?.done) { + sourceExhausted = true; + return; + } emitter.emit('result', await mapFn(val.value)); } @@ -102,24 +111,37 @@ export async function* mapAsyncIterable( (err) => emitter.emit('error', err) ); + const closeSource = async (suppressReturnError: boolean): Promise => { + stopped = true; + controller.abort(); + let hasReturnError = false; + let returnError: unknown; + if (!sourceExhausted) { + try { + await sourceIterator.return?.(); + } catch (err) { + hasReturnError = true; + returnError = err; + } + } + await Promise.allSettled(mappers); + if (hasReturnError && !suppressReturnError) { + throw returnError; + } + }; + + let hasPrimaryError = false; try { for await (const [res] of emitterEventsIterable) { bufferLimitSemaphore?.release(); yield res; } } catch (err: unknown) { - if (!isAbortError(err)) { + if (!isAbortError(err) || !sourceExhausted) { + hasPrimaryError = true; throw err; } } finally { - // Stop producers and close the source iterable so interceptor/OTel lifecycles end on - // early consumer termination (break/return), not only after the source is exhausted. - controller.abort(); - try { - await sourceIterator.return?.(); - } catch { - // Ignore cleanup errors from an already-failed or exhausted source. - } - await Promise.allSettled(mappers); + await closeSource(hasPrimaryError); } } diff --git a/packages/client/src/workflow-client.ts b/packages/client/src/workflow-client.ts index 705eea31f3..d59301226e 100644 --- a/packages/client/src/workflow-client.ts +++ b/packages/client/src/workflow-client.ts @@ -1872,11 +1872,14 @@ export class WorkflowClient extends BaseClient { ? (this.options.interceptors as WorkflowClientInterceptor[]) : []; const list = composeInterceptors(interceptors, 'list', this._list.bind(this)); - return { + const interceptedList: AsyncIterable = { [Symbol.asyncIterator]: () => list(input)[Symbol.asyncIterator](), + }; + return { + [Symbol.asyncIterator]: () => interceptedList[Symbol.asyncIterator](), intoHistories: (intoHistoriesOptions?: IntoHistoriesOptions) => { return mapAsyncIterable( - list(input), + interceptedList, async ({ workflowId, runId }) => ({ workflowId, history: await this.getHandle(workflowId, runId).fetchHistory(), diff --git a/packages/test/src/test-interceptors.cloud-pending.ts b/packages/test/src/test-interceptors.cloud-pending.ts index 45e976ba97..0988959c54 100644 --- a/packages/test/src/test-interceptors.cloud-pending.ts +++ b/packages/test/src/test-interceptors.cloud-pending.ts @@ -344,15 +344,18 @@ if (RUN_INTEGRATION_TESTS) { const client = new WorkflowClient({ interceptors: [ { - async *list(input, next) { + list(input, next) { listCalls += 1; - try { - for await (const execution of next(input)) { - yield execution; + const source = next(input); + return (async function* () { + try { + for await (const execution of source) { + yield execution; + } + } finally { + listExited += 1; } - } finally { - listExited += 1; - } + })(); }, async fetchHistory(input, next) { fetchHistoryCalls += 1; @@ -381,17 +384,26 @@ if (RUN_INTEGRATION_TESTS) { return count >= seeded; }, 30_000); - let histories = 0; - for await (const { history } of client.list({ query, pageSize: 2 }).intoHistories({ + const historiesIterable = client.list({ query, pageSize: 2 }).intoHistories({ concurrency: 2, - })) { + }); + t.is(listCalls, 0); + const historiesIterator = historiesIterable[Symbol.asyncIterator](); + t.is(listCalls, 0); + + let histories = 0; + while (histories < 7) { + const result = await historiesIterator.next(); + t.false(result.done); + const { history } = result.value!; const events = (history as { events?: unknown[] } | undefined)?.events; t.true((events?.length ?? 0) > 0); histories += 1; - if (histories >= 7) { - break; + if (histories === 1) { + t.is(listCalls, 1); } } + await historiesIterator.return?.(); t.is(histories, 7); t.is(listCalls, 1); diff --git a/packages/test/src/test-iterators-utils.ts b/packages/test/src/test-iterators-utils.ts index f6ce00a9c0..7fcdb6d7e8 100644 --- a/packages/test/src/test-iterators-utils.ts +++ b/packages/test/src/test-iterators-utils.ts @@ -137,7 +137,7 @@ test(`mapAsyncIterable (with concurrency) closes the source iterable on early te } } - const iterable = mapAsyncIterable(source(), sleepThatTime, { concurrency: 3 }); + const iterable = mapAsyncIterable(source(), sleepThatTime, { concurrency: 3, bufferLimit: 0 }); const seen: number[] = []; for await (const value of iterable) { seen.push(value); @@ -152,6 +152,63 @@ test(`mapAsyncIterable (with concurrency) closes the source iterable on early te t.true(produced < 20); }); +test(`mapAsyncIterable (with concurrency) propagates source return errors on early termination`, async (t) => { + const cleanupError = new Error('cleanup failed'); + let returnCalls = 0; + const source: AsyncIterable = { + [Symbol.asyncIterator]() { + return { + async next() { + return { done: false as const, value: 1 }; + }, + async return() { + returnCalls += 1; + throw cleanupError; + }, + }; + }, + }; + + await t.throwsAsync( + async () => { + for await (const _ of mapAsyncIterable(source, sleepThatTime, { concurrency: 2 })) { + break; + } + }, + { is: cleanupError } + ); + t.is(returnCalls, 1); +}); + +test(`mapAsyncIterable (with concurrency) does not close a naturally exhausted source`, async (t) => { + let nextCalls = 0; + let returnCalls = 0; + const source: AsyncIterable = { + [Symbol.asyncIterator]() { + return { + async next() { + nextCalls += 1; + return nextCalls <= 2 + ? { done: false as const, value: nextCalls } + : { done: true as const, value: undefined }; + }, + async return() { + returnCalls += 1; + return { done: true as const, value: undefined }; + }, + }; + }, + }; + + const values: number[] = []; + for await (const value of mapAsyncIterable(source, multBy10, { concurrency: 2 })) { + values.push(value); + } + + t.deepEqual(values.sort(), [10, 20]); + t.is(returnCalls, 0); +}); + test(`mapAsyncIterable (with concurrency) doesn't hang on source exceptions`, async (t) => { async function* name(): AsyncIterable { for (;;) { @@ -181,6 +238,33 @@ test(`mapAsyncIterable (with concurrency) doesn't hang on source exceptions`, as }); }); +test(`mapAsyncIterable (with concurrency) doesn't hang when a backpressured mapper fails`, async (t) => { + const mapError = new Error('Map Exception'); + async function* source(): AsyncIterable { + for (let value = 1; ; value++) { + yield value; + } + } + + const iterable = mapAsyncIterable( + source(), + async (value) => { + if (value === 2) { + await new Promise((resolve) => setTimeout(resolve, 25)); + throw mapError; + } + return value; + }, + { concurrency: 2, bufferLimit: 0 } + ); + const iterator = iterable[Symbol.asyncIterator](); + + t.is((await iterator.next()).value, 1); + await new Promise((resolve) => setTimeout(resolve, 50)); + t.is((await iterator.next()).value, 3); + await t.throwsAsync(iterator.next(), { is: mapError }); +}); + // FIXME: This test is producing rare flakes test(`mapAsyncIterable (with concurrency) doesn't hang mapFn exceptions`, async (t) => { async function* name(): AsyncIterable {