From 057de7d9b8d16d97bdf5978db6581780dc0e6676 Mon Sep 17 00:00:00 2001 From: Dinh Le Date: Mon, 6 Jul 2026 10:07:13 +0700 Subject: [PATCH 1/2] feat: onAsyncIteratorObjectError, onReadableStreamError Fixes #604 --- .../client/src/async-iterator-object.test.ts | 6 +- packages/client/src/index.ts | 2 + packages/server/src/index.ts | 2 + packages/server/src/procedure-client.ts | 4 +- packages/shared/src/interceptor.test-d.ts | 20 ++- packages/shared/src/interceptor.test.ts | 132 +++++++++++++++++- packages/shared/src/interceptor.ts | 90 ++++++++++++ packages/shared/src/iterator.ts | 10 +- packages/shared/src/stream.ts | 11 +- 9 files changed, 259 insertions(+), 18 deletions(-) diff --git a/packages/client/src/async-iterator-object.test.ts b/packages/client/src/async-iterator-object.test.ts index c25b2f194..5a4c85b6c 100644 --- a/packages/client/src/async-iterator-object.test.ts +++ b/packages/client/src/async-iterator-object.test.ts @@ -69,7 +69,7 @@ describe('wrapAsyncIteratorPreservingEventMeta', () => { it('preserves metadata when mapping errors', async () => { const error = withEventMeta(new Error('TEST'), { id: 'error-1' }) const onError = vi.fn() - const mapError = vi.fn(async cause => ({ mapped: cause })) + const mapError = vi.fn(async cause => ({ mapped: cause } as any)) const iterator = (async function* () { throw error @@ -107,9 +107,9 @@ describe('wrapAsyncIteratorPreservingEventMeta', () => { const primitiveError = wrapAsyncIteratorPreservingEventMeta((async function* () { throw error })(), { - mapError: async () => 'mapped-error', + mapError: async () => 'mapped error' as any, }) - await expect(primitiveError.next()).rejects.toBe('mapped-error') + await expect(primitiveError.next()).rejects.toBe('mapped error') }) }) diff --git a/packages/client/src/index.ts b/packages/client/src/index.ts index 3b4631e6c..cd2c34454 100644 --- a/packages/client/src/index.ts +++ b/packages/client/src/index.ts @@ -42,8 +42,10 @@ export { * @deprecated Use `asyncIteratorToUnproxiedDataStream` instead. */ asyncIteratorToUnproxiedDataStream as eventIteratorToUnproxiedDataStream, + onAsyncIteratorObjectError, onError, onFinish, + onReadableStreamError, onStart, onSuccess, streamToAsyncIteratorObject, diff --git a/packages/server/src/index.ts b/packages/server/src/index.ts index ff2d3b734..dd7ca892c 100644 --- a/packages/server/src/index.ts +++ b/packages/server/src/index.ts @@ -98,8 +98,10 @@ export { * @deprecated Use `asyncIteratorToUnproxiedDataStream` instead. */ asyncIteratorToUnproxiedDataStream as eventIteratorToUnproxiedDataStream, + onAsyncIteratorObjectError, onError, onFinish, + onReadableStreamError, onStart, onSuccess, streamToAsyncIteratorObject, diff --git a/packages/server/src/procedure-client.ts b/packages/server/src/procedure-client.ts index 28cc177b9..47c5bf615 100644 --- a/packages/server/src/procedure-client.ts +++ b/packages/server/src/procedure-client.ts @@ -80,7 +80,7 @@ export function createProcedureClient< const context = await value(options.context, clientContext) as TInitialContext | undefined ?? {} as TInitialContext const errors = createORPCErrorConstructorMap(procedure['~orpc'].errorMap) - const reconcileError = async (e: unknown) => { + const reconcileError = async (e: ThrowableError) => { if (e instanceof ORPCError) { return await reconcileORPCError(procedure['~orpc'].errorMap, e) } @@ -142,7 +142,7 @@ export function createProcedureClient< * Defined errors take priority over inferable errors. * `reconcileError` attempts to mark the error as defined, or keeps it inferable if that's not possible. */ - throw await reconcileError(e) + throw await reconcileError(e as ThrowableError) } } } diff --git a/packages/shared/src/interceptor.test-d.ts b/packages/shared/src/interceptor.test-d.ts index 5c32d7071..ff50d6f27 100644 --- a/packages/shared/src/interceptor.test-d.ts +++ b/packages/shared/src/interceptor.test-d.ts @@ -1,8 +1,8 @@ import type { Interceptor } from './interceptor' import type { PromiseWithError } from './types' -import { onError, onFinish, onStart, onSuccess } from './interceptor' +import { onAsyncIteratorObjectError, onError, onFinish, onReadableStreamError, onStart, onSuccess } from './interceptor' -it('onStart, onSuccess, onError, onFinish can be used as an interceptor', () => { +it('onStart, onSuccess, onError, onFinish, onAsyncIteratorObjectError, onReadableStreamError can be used as an interceptor', () => { const interceptors: Interceptor<{ foo: string }, PromiseWithError<'success', 'error'>>[] = [] interceptors.push(onStart((options) => { @@ -41,4 +41,20 @@ it('onStart, onSuccess, onError, onFinish can be used as an interceptor', () => expectTypeOf(options.next).toBeCallableWith<[options?: { foo: string }]>() expectTypeOf(options.next()).toEqualTypeOf>() })) + + interceptors.push(onAsyncIteratorObjectError((error, options) => { + expectTypeOf(error).toEqualTypeOf<'error' | Error>() + + expectTypeOf(options.foo).toEqualTypeOf() + expectTypeOf(options.next).toBeCallableWith<[options?: { foo: string }]>() + expectTypeOf(options.next()).toEqualTypeOf>() + })) + + interceptors.push(onReadableStreamError((error, options) => { + expectTypeOf(error).toEqualTypeOf<'error' | Error>() + + expectTypeOf(options.foo).toEqualTypeOf() + expectTypeOf(options.next).toBeCallableWith<[options?: { foo: string }]>() + expectTypeOf(options.next()).toEqualTypeOf>() + })) }) diff --git a/packages/shared/src/interceptor.test.ts b/packages/shared/src/interceptor.test.ts index 461dbec27..cc4c75993 100644 --- a/packages/shared/src/interceptor.test.ts +++ b/packages/shared/src/interceptor.test.ts @@ -1,4 +1,4 @@ -import { intercept, onError, onFinish, onStart, onSuccess } from './interceptor' +import { intercept, onAsyncIteratorObjectError, onError, onFinish, onReadableStreamError, onStart, onSuccess } from './interceptor' describe('intercept', () => { const interceptor1 = vi.fn(({ next }) => next()) @@ -381,3 +381,133 @@ describe('lifecycle interceptors', () => { expect(onSuccessFn).toHaveBeenCalledTimes(0) }) }) + +describe('stream interceptors', () => { + beforeEach(() => { + vi.clearAllMocks() + }) + + describe('onAsyncIteratorError', () => { + it('does not invoke callback for non-iterator output', async () => { + const callback = vi.fn() + + const result = await intercept( + [onAsyncIteratorObjectError(callback)], + { foo: 'bar' }, + () => '__OUTPUT__', + ) + + expect(result).toBe('__OUTPUT__') + expect(callback).not.toHaveBeenCalled() + }) + + it('invokes callback when the returned async iterator errors while being consumed', async () => { + const error = new Error('__ITERATOR_ERROR__') + const iterator = (async function* () { + yield 'event' + throw error + })() + + const callback = vi.fn() + + const wrapped = await intercept( + [onAsyncIteratorObjectError(callback)], + { foo: 'bar' }, + () => iterator, + ) as AsyncIterableIterator + + await expect(wrapped.next()).resolves.toEqual({ value: 'event', done: false }) + await expect(wrapped.next()).rejects.toThrow(error) + + expect(callback).toHaveBeenCalledTimes(1) + expect(callback).toHaveBeenCalledWith( + error, + expect.objectContaining({ foo: 'bar', next: expect.any(Function) }), + ) + }) + + it('does not invoke callback when the async iterator completes successfully', async () => { + const iterator = (async function* () { + yield 'event' + return 'done' + })() + + const callback = vi.fn() + + const wrapped = await intercept( + [onAsyncIteratorObjectError(callback)], + { foo: 'bar' }, + () => iterator, + ) as AsyncIterableIterator + + await expect(wrapped.next()).resolves.toEqual({ value: 'event', done: false }) + await expect(wrapped.next()).resolves.toEqual({ value: 'done', done: true }) + + expect(callback).not.toHaveBeenCalled() + }) + }) + + describe('onReadableStreamError', () => { + it('does not invoke callback for non-stream output', async () => { + const callback = vi.fn() + + const result = await intercept( + [onReadableStreamError(callback)], + { foo: 'bar' }, + () => '__OUTPUT__', + ) + + expect(result).toBe('__OUTPUT__') + expect(callback).not.toHaveBeenCalled() + }) + + it('invokes callback when the returned readable stream errors while being consumed', async () => { + const error = new Error('__STREAM_ERROR__') + const stream = new ReadableStream({ + pull(controller) { + controller.error(error) + }, + }) + + const callback = vi.fn() + + const wrapped = await intercept( + [onReadableStreamError(callback)], + { foo: 'bar' }, + () => stream, + ) as ReadableStream + + const reader = wrapped.getReader() + await expect(reader.read()).rejects.toThrow(error) + + expect(callback).toHaveBeenCalledTimes(1) + expect(callback).toHaveBeenCalledWith( + error, + expect.objectContaining({ foo: 'bar', next: expect.any(Function) }), + ) + }) + + it('does not invoke callback when the readable stream completes successfully', async () => { + const stream = new ReadableStream({ + start(controller) { + controller.enqueue('event') + controller.close() + }, + }) + + const callback = vi.fn() + + const wrapped = await intercept( + [onReadableStreamError(callback)], + { foo: 'bar' }, + () => stream, + ) as ReadableStream + + const reader = wrapped.getReader() + await expect(reader.read()).resolves.toEqual({ value: 'event', done: false }) + await expect(reader.read()).resolves.toEqual({ value: undefined, done: true }) + + expect(callback).not.toHaveBeenCalled() + }) + }) +}) diff --git a/packages/shared/src/interceptor.ts b/packages/shared/src/interceptor.ts index 9841fafea..8777b3cbe 100644 --- a/packages/shared/src/interceptor.ts +++ b/packages/shared/src/interceptor.ts @@ -1,5 +1,9 @@ import type { Promisable } from 'type-fest' import type { PromiseWithError, ThrowableError } from './types' +import { isAsyncIteratorObject } from '@standardserver/shared' +import { wrapAsyncIterator } from './iterator' +import { override } from './proxy' +import { wrapReadableStream } from './stream' export type InterceptableOptions = Record @@ -96,6 +100,92 @@ export function onFinish any }, TRest extends } } +/** + * Creates an middleware or interceptor that invokes a callback when the returned async + * iterator object throws an error while being consumed. + * + * This does not replace the `onError`. `onError` only fires on the + * initial call (before the interceptor returns the iterator), whereas this + * callback only fires while consuming the iterator. Use both together to + * catch all possible errors. + */ +export function onAsyncIteratorObjectError< + T, + TOptions extends { next: () => any }, + TRest extends any[], +>( + callback: NoInfer<( + error: ThrowableError | (ReturnType extends PromiseWithError ? E : ThrowableError), + options: TOptions, + ...rest: TRest + ) => Promisable>, +): (options: TOptions, ...rest: TRest) => T | Promise>> { + // The typed error should always be combined with `ThrowableError` + // because an AsyncIterator can throw any error during iteration, + // while TError only represents errors from the initial promise. + // In oRPC client/server usage, initial and iteration errors are usually the same, + // but this utility is shared, so it needs to support the general case. + + return async (options, ...rest) => { + const output = await options.next() + + if (!isAsyncIteratorObject(output)) { + return output + } + + /** + * @warning + * Remember use `override` for AsyncIteratorObject to remain other special properties + */ + return override(output, wrapAsyncIterator(output, { + onError: error => callback(error, options, ...rest), + })) + } +} + +/** + * Creates an interceptor that invokes a callback when the returned readable + * stream errors while being consumed. + * + * This does not replace the `onError`. `onError` only fires on the + * initial call (before the interceptor returns the stream), whereas this + * callback only fires while consuming the stream. Use both together to catch + * all possible errors. + */ +export function onReadableStreamError< + T, + TOptions extends { next: () => any }, + TRest extends any[], +>( + callback: NoInfer<( + error: ThrowableError | (ReturnType extends PromiseWithError ? E : ThrowableError), + options: TOptions, + ...rest: TRest + ) => Promisable>, +): (options: TOptions, ...rest: TRest) => T | Promise>> { + // The typed error should always be combined with `ThrowableError` + // because an AsyncIterator can throw any error during iteration, + // while TError only represents errors from the initial promise. + // In oRPC client/server usage, initial and iteration errors are usually the same, + // but this utility is shared, so it needs to support the general case. + + return async (options, ...rest) => { + const output = await options.next() + + if (!(output instanceof ReadableStream)) { + return output + } + + /** + * @warning + * Remember use `override` for ReadableStream to remain other special properties + */ + return override(output, wrapReadableStream(output, { + onError: error => callback(error, options, ...rest), + })) + } +} + export function intercept( interceptors: undefined | Interceptor[], options: NoInfer, diff --git a/packages/shared/src/iterator.ts b/packages/shared/src/iterator.ts index fc2d1c22a..c8ea8c2dd 100644 --- a/packages/shared/src/iterator.ts +++ b/packages/shared/src/iterator.ts @@ -14,8 +14,8 @@ export interface WrapAsyncIteratorOptions(run: () => Promise) => Promise mapResult?: (result: IteratorResult) => Promisable> - mapError?: (error: unknown) => Promisable - onError?: (error: unknown) => Promisable + mapError?: (error: ThrowableError) => Promisable + onError?: (error: ThrowableError) => Promisable /** * Execute after the stream finishes or is cancelled. @@ -46,8 +46,8 @@ export function wrapAsyncIterator { try { @@ -58,7 +58,7 @@ export function wrapAsyncIterator iterator.return?.()) } catch (error) { - await onError?.(error) + await onError?.(error as ThrowableError) throw error } } diff --git a/packages/shared/src/stream.ts b/packages/shared/src/stream.ts index 2ae3b5ee1..7aae1303e 100644 --- a/packages/shared/src/stream.ts +++ b/packages/shared/src/stream.ts @@ -1,5 +1,6 @@ import type { Promisable } from 'type-fest' import type { StartSpanOptions } from './opentelemetry' +import type { ThrowableError } from './types' import { AsyncIteratorClass } from '@standardserver/shared' import { once } from './function' import { isPlainObject } from './object' @@ -38,8 +39,8 @@ export interface WrapReadableStreamOptions { */ runWith?: (run: () => Promise) => Promise mapResult?: (result: ReadableStreamReadResult) => Promisable> - mapError?: (error: unknown) => Promisable - onError?: (error: unknown) => Promisable + mapError?: (error: ThrowableError) => Promisable + onError?: (error: ThrowableError) => Promisable /** * Guaranteed to execute exactly once after the stream finishes or is cancelled. @@ -67,8 +68,8 @@ export function wrapReadableStream( } catch (error) { try { - await onError?.(error) - controller.error(mapError ? await mapError(error) : error) + await onError?.(error as ThrowableError) + controller.error(mapError ? await mapError(error as ThrowableError) : error) } finally { await finish() @@ -91,7 +92,7 @@ export function wrapReadableStream( await runWith(() => reader().cancel(reason)) } catch (error) { - await onError?.(error) + await onError?.(error as ThrowableError) throw error } } From 3e7b1c96facd2f9c0873cd39c31f6e65c32d56ec Mon Sep 17 00:00:00 2001 From: Dinh Le Date: Mon, 6 Jul 2026 11:12:08 +0700 Subject: [PATCH 2/2] typo --- packages/shared/src/interceptor.test.ts | 2 +- packages/shared/src/interceptor.ts | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/packages/shared/src/interceptor.test.ts b/packages/shared/src/interceptor.test.ts index cc4c75993..385137f06 100644 --- a/packages/shared/src/interceptor.test.ts +++ b/packages/shared/src/interceptor.test.ts @@ -387,7 +387,7 @@ describe('stream interceptors', () => { vi.clearAllMocks() }) - describe('onAsyncIteratorError', () => { + describe('onAsyncIteratorObjectError', () => { it('does not invoke callback for non-iterator output', async () => { const callback = vi.fn() diff --git a/packages/shared/src/interceptor.ts b/packages/shared/src/interceptor.ts index 8777b3cbe..4a2483caa 100644 --- a/packages/shared/src/interceptor.ts +++ b/packages/shared/src/interceptor.ts @@ -101,7 +101,7 @@ export function onFinish any }, TRest extends } /** - * Creates an middleware or interceptor that invokes a callback when the returned async + * Creates a middleware or interceptor that invokes a callback when the returned async * iterator object throws an error while being consumed. * * This does not replace the `onError`. `onError` only fires on the @@ -164,7 +164,7 @@ export function onReadableStreamError< ) => Promisable>, ): (options: TOptions, ...rest: TRest) => T | Promise>> { // The typed error should always be combined with `ThrowableError` - // because an AsyncIterator can throw any error during iteration, + // because a ReadableStream can throw any error during iteration, // while TError only represents errors from the initial promise. // In oRPC client/server usage, initial and iteration errors are usually the same, // but this utility is shared, so it needs to support the general case.