From d2cc21cd14d54fe5daf7e94923580de44c756655 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 7 Nov 2026 23:15:52 +0000 Subject: [PATCH 1/5] refactor(shared, client, server): dedupe compression encoding helpers into shared Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VVaw7PANP3R1NnJpBLkx5b --- .../client/src/plugins/request-compression.ts | 3 ++- .../src/plugins/response-compression.ts | 24 ++---------------- .../server/src/plugins/request-compression.ts | 23 ++--------------- .../src/plugins/response-compression.ts | 3 ++- packages/shared/src/http.test.ts | 25 +++++++++++++++++++ packages/shared/src/http.ts | 24 ++++++++++++++++++ 6 files changed, 57 insertions(+), 45 deletions(-) diff --git a/packages/client/src/plugins/request-compression.ts b/packages/client/src/plugins/request-compression.ts index 49943adee..dbd0055db 100644 --- a/packages/client/src/plugins/request-compression.ts +++ b/packages/client/src/plugins/request-compression.ts @@ -1,3 +1,4 @@ +import type { CompressionEncoding } from '@orpc/shared' import { isAsyncIteratorObject, isCompressibleContentType, toArray } from '@orpc/shared' import type { StandardBodyHint, StandardHeaders } from '@standard-server/core' import { flattenStandardHeader, generateContentDisposition } from '@standard-server/core' @@ -21,7 +22,7 @@ export interface RequestCompressionLinkPluginOptions { * * @default 'gzip' */ - encoding?: 'gzip' | 'deflate' | 'deflate-raw' | undefined + encoding?: CompressionEncoding | undefined /** * The minimum request size in bytes required to trigger compression. diff --git a/packages/client/src/plugins/response-compression.ts b/packages/client/src/plugins/response-compression.ts index 2caf1869b..2c0d30860 100644 --- a/packages/client/src/plugins/response-compression.ts +++ b/packages/client/src/plugins/response-compression.ts @@ -1,4 +1,4 @@ -import { toArray } from '@orpc/shared' +import { isCompressionEncoding, parseContentEncodings, toArray } from '@orpc/shared' import { flattenStandardHeader } from '@standard-server/core' import { toFetchHeaders, toStandardBody } from '@standard-server/fetch' @@ -73,7 +73,7 @@ export class ResponseCompressionLinkPlugin< flattenStandardHeader(response.headers['content-encoding']), ) - if (encodings.length === 0 || !encodings.every(isSupportedEncoding)) { + if (encodings.length === 0 || !encodings.every(isCompressionEncoding)) { return response } @@ -120,23 +120,3 @@ export class ResponseCompressionLinkPlugin< } } } - -const SUPPORTED_ENCODINGS = ['gzip', 'deflate', 'deflate-raw'] as const -type SupportedEncoding = (typeof SUPPORTED_ENCODINGS)[number] - -function isSupportedEncoding(encoding: string): encoding is SupportedEncoding { - return (SUPPORTED_ENCODINGS as readonly string[]).includes(encoding) -} - -/** - * Parse Content-Encoding into ordered codings (order applied). - * - * @see https://www.rfc-editor.org/rfc/rfc9110.html#name-content-encoding - */ -function parseContentEncodings(header: string | undefined): string[] { - if (header === undefined) { - return [] - } - - return header.split(',').map((part) => part.trim().toLowerCase()) -} diff --git a/packages/server/src/plugins/request-compression.ts b/packages/server/src/plugins/request-compression.ts index d95e97f20..21943ff21 100644 --- a/packages/server/src/plugins/request-compression.ts +++ b/packages/server/src/plugins/request-compression.ts @@ -1,5 +1,5 @@ import { ORPCError } from '@orpc/client' -import { toArray } from '@orpc/shared' +import { isCompressionEncoding, parseContentEncodings, toArray } from '@orpc/shared' import { flattenStandardHeader } from '@standard-server/core' import { toFetchHeaders, toStandardBody } from '@standard-server/fetch' @@ -42,7 +42,7 @@ export class RequestCompressionHandlerPlugin< flattenStandardHeader(interceptorOptions.request.headers['content-encoding']), ) - if (encodings.length === 0 || !encodings.every(isSupportedEncoding)) { + if (encodings.length === 0 || !encodings.every(isCompressionEncoding)) { return next() } @@ -94,22 +94,3 @@ export class RequestCompressionHandlerPlugin< } } } - -const SUPPORTED_ENCODINGS = ['gzip', 'deflate', 'deflate-raw'] as const -type SupportedEncoding = (typeof SUPPORTED_ENCODINGS)[number] -function isSupportedEncoding(encoding: string): encoding is SupportedEncoding { - return (SUPPORTED_ENCODINGS as readonly string[]).includes(encoding) -} - -/** - * Parse Content-Encoding into ordered codings (order applied). - * - * @see https://www.rfc-editor.org/rfc/rfc9110.html#name-content-encoding - */ -function parseContentEncodings(header: string | undefined): string[] { - if (header === undefined) { - return [] - } - - return header.split(',').map((part) => part.trim().toLowerCase()) -} diff --git a/packages/server/src/plugins/response-compression.ts b/packages/server/src/plugins/response-compression.ts index c61953320..e42d08f1b 100644 --- a/packages/server/src/plugins/response-compression.ts +++ b/packages/server/src/plugins/response-compression.ts @@ -1,3 +1,4 @@ +import type { CompressionEncoding } from '@orpc/shared' import { isAcceptableEncoding, isAsyncIteratorObject, @@ -31,7 +32,7 @@ export interface ResponseCompressionHandlerPluginOptions { * * @default ['gzip', 'deflate'] */ - encodings?: readonly ('gzip' | 'deflate' | 'deflate-raw')[] | undefined + encodings?: readonly CompressionEncoding[] | undefined /** * The minimum response size in bytes required to trigger compression. diff --git a/packages/shared/src/http.test.ts b/packages/shared/src/http.test.ts index 497d4be90..324a1ff6e 100644 --- a/packages/shared/src/http.test.ts +++ b/packages/shared/src/http.test.ts @@ -1,12 +1,14 @@ import { isAcceptableEncoding, isCompressibleContentType, + isCompressionEncoding, isNoTransformCacheControl, matchesHttpPath, matchesHttpPathPrefix, mergeHttpPath, normalizeHttpPath, parseAcceptEncodingQualities, + parseContentEncodings, pathToHttpPath, varyByAcceptEncoding, } from './http.ts' @@ -240,6 +242,29 @@ describe('parseAcceptEncodingQualities', () => { }) }) +describe('isCompressionEncoding', () => { + it.each([ + ['gzip', true], + ['deflate', true], + ['deflate-raw', true], + ['br', false], + ['identity', false], + ['GZIP', false], + ])('%s: %s', (encoding, expected) => { + expect(isCompressionEncoding(encoding)).toBe(expected) + }) +}) + +describe('parseContentEncodings', () => { + it('returns no codings when the header is missing', () => { + expect(parseContentEncodings(undefined)).toEqual([]) + }) + + it('keeps the order codings were applied in, trimmed and lowercased', () => { + expect(parseContentEncodings(' GZIP , deflate')).toEqual(['gzip', 'deflate']) + }) +}) + describe('isAcceptableEncoding', () => { it.each([ ['gzip', 'gzip', true], diff --git a/packages/shared/src/http.ts b/packages/shared/src/http.ts index 48f21ab54..6154fdcd5 100644 --- a/packages/shared/src/http.ts +++ b/packages/shared/src/http.ts @@ -52,6 +52,30 @@ export function matchesHttpPath(url: `/${string}`, path: `/${string}`): boolean return charAfterPrefix === undefined || charAfterPrefix === '?' || charAfterPrefix === '#' } +/** + * Content codings that oRPC compression plugins can encode and decode. + */ +export const COMPRESSION_ENCODINGS = ['gzip', 'deflate', 'deflate-raw'] as const + +export type CompressionEncoding = (typeof COMPRESSION_ENCODINGS)[number] + +export function isCompressionEncoding(encoding: string): encoding is CompressionEncoding { + return (COMPRESSION_ENCODINGS as readonly string[]).includes(encoding) +} + +/** + * Parse Content-Encoding into ordered codings (order applied). + * + * @see https://www.rfc-editor.org/rfc/rfc9110.html#name-content-encoding + */ +export function parseContentEncodings(header: string | undefined): string[] { + if (header === undefined) { + return [] + } + + return header.split(',').map((part) => part.trim().toLowerCase()) +} + const ACCEPT_ENCODING_QUALITY_REGEX = /^\s*q=([\d.]+)\s*$/i /** From 7ed8e94238a06b088f49b2da970883057d9e74e4 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 7 Nov 2026 23:25:14 +0000 Subject: [PATCH 2/5] refactor(shared): dedupe type checks, regexes and small helpers into shared Move helpers copied across packages into @orpc/shared: - INTEGER_REGEX / NON_NEGATIVE_INTEGER_REGEX (shared bigint, json-schema coercer, client rpc serializer, openapi form-data) - isPlainObjectOrArray (client, openapi, mcp, json-schema) - listenAbort (server signal-sync, client dedupe, nest, msw) - stripHttpPathPrefix (server rpc matcher, openapi matcher) - takeLast (pinia-colada, tanstack-query stream queries) - estimateCompressibleFormDataSize / AVG_UTF8_BYTES_PER_CHAR (client and server compression plugins) - trailing-slash regex merges in link codecs now use mergeHttpPath Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VVaw7PANP3R1NnJpBLkx5b --- .../src/adapters/standard/rpc-link-codec.ts | 13 ++-- packages/client/src/plugins/dedupe.ts | 28 +++----- .../client/src/plugins/request-compression.ts | 45 ++++-------- packages/client/src/rpc-json-serializer.ts | 9 +-- packages/client/src/rpc-serializer.ts | 11 +-- packages/json-schema/src/coercer.ts | 9 +-- packages/mcp/src/registrar.ts | 4 +- packages/msw/src/http-procedure-utils.ts | 9 +-- packages/nest/src/implement.ts | 7 +- .../adapters/standard/openapi-link-codec.ts | 5 +- .../src/adapters/standard/openapi-matcher.ts | 18 ++--- packages/openapi/src/helpers/form-data.ts | 4 +- .../src/openapi-generator-components.ts | 8 +-- packages/pinia-colada/src/stream-query.ts | 16 ++--- .../src/adapters/standard/rpc-matcher.ts | 18 ++--- .../src/plugins/response-compression.ts | 39 ++-------- packages/server/src/plugins/signal-sync.ts | 11 +-- packages/shared/src/array.test.ts | 11 ++- packages/shared/src/array.ts | 11 +++ packages/shared/src/bigint.ts | 9 +-- packages/shared/src/http.test.ts | 52 ++++++++++++++ packages/shared/src/http.ts | 72 +++++++++++++++++++ packages/shared/src/index.ts | 1 + packages/shared/src/object.test.ts | 11 +++ packages/shared/src/object.ts | 9 +++ packages/shared/src/regex.test.ts | 26 +++++++ packages/shared/src/regex.ts | 10 +++ packages/shared/src/signal.test.ts | 39 +++++++++- packages/shared/src/signal.ts | 14 ++++ packages/tanstack-query/src/stream-query.ts | 16 ++--- 30 files changed, 342 insertions(+), 193 deletions(-) create mode 100644 packages/shared/src/regex.test.ts create mode 100644 packages/shared/src/regex.ts diff --git a/packages/client/src/adapters/standard/rpc-link-codec.ts b/packages/client/src/adapters/standard/rpc-link-codec.ts index 508cf792a..9b9ef66b3 100644 --- a/packages/client/src/adapters/standard/rpc-link-codec.ts +++ b/packages/client/src/adapters/standard/rpc-link-codec.ts @@ -1,5 +1,11 @@ import type { Promisable, Public, Value } from '@orpc/shared' -import { isAsyncIteratorObject, pathToHttpPath, stringifyJSON, value } from '@orpc/shared' +import { + isAsyncIteratorObject, + mergeHttpPath, + pathToHttpPath, + stringifyJSON, + value, +} from '@orpc/shared' import type { StandardHeaders, StandardLazyResponse, @@ -74,8 +80,6 @@ export interface RPCLinkCodecOptions { serializer?: Public | undefined } -const END_SLASH_REGEX = /\/$/ - export class RPCLinkCodec implements StandardLinkCodec { private readonly baseUrl: Exclude['url'], undefined> private readonly maxUrlLength: Exclude['maxUrlLength'], undefined> @@ -107,8 +111,7 @@ export class RPCLinkCodec implements StandardLinkCodec< const baseUrl = await value(this.baseUrl, options, path, input) const [pathname, search, hash] = parseStandardUrl(baseUrl) - const newPathname = - `${pathname.replace(END_SLASH_REGEX, '')}${pathToHttpPath(path)}` as StandardUrl + const newPathname: StandardUrl = mergeHttpPath(pathname, pathToHttpPath(path)) const serialized = this.serializer.serialize(input) if ( diff --git a/packages/client/src/plugins/dedupe.ts b/packages/client/src/plugins/dedupe.ts index 777da10c5..a37c84d68 100644 --- a/packages/client/src/plugins/dedupe.ts +++ b/packages/client/src/plugins/dedupe.ts @@ -9,6 +9,7 @@ import { defer, disposableAllAbortSignal, isAsyncIteratorObject, + listenAbort, replicateAsyncIterator, replicateReadableStream, runWithSignal, @@ -318,13 +319,7 @@ function closeAsyncIteratorOnAbort( return replica } - const abort = () => replica.return().catch(() => {}) - - if (signal.aborted) { - void abort() - } else { - signal.addEventListener('abort', abort, { once: true }) - } + const removeAbortListener = listenAbort(signal, () => void replica.return().catch(() => {})) return wrapAsyncIterator(replica, { mapResult(result) { @@ -332,7 +327,7 @@ function closeAsyncIteratorOnAbort( throwIfAborted(signal) return result }, - onFinish: () => signal.removeEventListener('abort', abort), + onFinish: removeAbortListener, }) } @@ -349,13 +344,10 @@ function closeReadableStreamOnAbort( } const reader = replica.getReader() - const abort = () => reader.cancel(signal.reason).catch(() => {}) - - if (signal.aborted) { - void abort() - } else { - signal.addEventListener('abort', abort, { once: true }) - } + const removeAbortListener = listenAbort( + signal, + () => void reader.cancel(signal.reason).catch(() => {}), + ) return new ReadableStream( { @@ -366,18 +358,18 @@ function closeReadableStreamOnAbort( throwIfAborted(signal) if (result.done) { - signal.removeEventListener('abort', abort) + removeAbortListener() controller.close() } else { controller.enqueue(result.value) } } catch (error) { - signal.removeEventListener('abort', abort) + removeAbortListener() throw error } }, async cancel(reason) { - signal.removeEventListener('abort', abort) + removeAbortListener() await reader.cancel(reason) }, }, diff --git a/packages/client/src/plugins/request-compression.ts b/packages/client/src/plugins/request-compression.ts index dbd0055db..e28179eec 100644 --- a/packages/client/src/plugins/request-compression.ts +++ b/packages/client/src/plugins/request-compression.ts @@ -1,5 +1,11 @@ import type { CompressionEncoding } from '@orpc/shared' -import { isAsyncIteratorObject, isCompressibleContentType, toArray } from '@orpc/shared' +import { + AVG_UTF8_BYTES_PER_CHAR, + estimateCompressibleFormDataSize, + isAsyncIteratorObject, + isCompressibleContentType, + toArray, +} from '@orpc/shared' import type { StandardBodyHint, StandardHeaders } from '@standard-server/core' import { flattenStandardHeader, generateContentDisposition } from '@standard-server/core' import { toFetchBody } from '@standard-server/fetch' @@ -12,10 +18,6 @@ import type { } from '../adapters/standard/index.ts' import type { ClientContext } from '../types.ts' -// Rough UTF-8 estimate. Mostly ASCII text stays close to 1 byte/char; -// occasional multi-byte characters increase the average. -const AVG_BYTES_PER_CHAR = 1.2 - export interface RequestCompressionLinkPluginOptions { /** * The compression scheme to use for request compression. @@ -152,32 +154,9 @@ export class RequestCompressionLinkPlugin< }) } } else if (request.body instanceof FormData) { - const PART_OVERHEAD = 64 // approx bytes for boundary + Content-Disposition/Content-Type headers per part - - let contentLength = 0 - for (const [key, value] of request.body) { - contentLength += PART_OVERHEAD + key.length - - if (value instanceof Blob) { - if (!Number.isFinite(value.size)) { - // Bun-s3 can use NaN for size - if (!this.isCompressibleContentType(value.type, interceptorOptions)) { - // Unknown non-compressible part size makes the estimate unreliable - contentLength = -Infinity - break - } - - // Unknown size for compressible content - still apply compression - contentLength = Infinity - } else { - contentLength += this.isCompressibleContentType(value.type, interceptorOptions) - ? value.size - : -value.size - } - } else { - contentLength += value.length * AVG_BYTES_PER_CHAR - } - } + const contentLength = estimateCompressibleFormDataSize(request.body, (contentType) => + this.isCompressibleContentType(contentType, interceptorOptions), + ) if (contentLength >= this.threshold) { const res = new Response(request.body) @@ -200,7 +179,7 @@ export class RequestCompressionLinkPlugin< } } else if (request.body instanceof URLSearchParams) { const string = request.body.toString() - if (string.length * AVG_BYTES_PER_CHAR >= this.threshold) { + if (string.length * AVG_UTF8_BYTES_PER_CHAR >= this.threshold) { const compressedStream = new Blob([string]) .stream() .pipeThrough(new CompressionStream(this.encoding)) @@ -226,7 +205,7 @@ export class RequestCompressionLinkPlugin< string, StandardHeaders, ] - if (string.length * AVG_BYTES_PER_CHAR >= this.threshold) { + if (string.length * AVG_UTF8_BYTES_PER_CHAR >= this.threshold) { const compressedStream = new Blob([string]) .stream() .pipeThrough(new CompressionStream(this.encoding)) diff --git a/packages/client/src/rpc-json-serializer.ts b/packages/client/src/rpc-json-serializer.ts index e6f4ef5e4..47010775c 100644 --- a/packages/client/src/rpc-json-serializer.ts +++ b/packages/client/src/rpc-json-serializer.ts @@ -3,6 +3,7 @@ import { copyOnWrite, DEFAULT_MAX_BIGINT_DIGITS, isPlainObject, + isPlainObjectOrArray, NullProtoObj, parseBigIntString, } from '@orpc/shared' @@ -50,14 +51,10 @@ function assertSerializedType(ok: boolean, type: string, expected: string): void } } -function isWalkable(value: unknown): boolean { - return Array.isArray(value) || isPlainObject(value) -} - function walkInto(parent: any, key: Segment, original: unknown, restored?: Set): any { const value = parent[key] - if (!isWalkable(value) || restored?.has(value)) { + if (!isPlainObjectOrArray(value) || restored?.has(value)) { throw invalidSerializedData(`segment "${key}" cannot be walked into.`) } @@ -562,7 +559,7 @@ export class RPCJsonSerializer { const value = handler.deserialize(currentRef[preSegment]) currentRef[preSegment] = value - if (isWalkable(value)) { + if (isPlainObjectOrArray(value)) { ;(restored ??= new Set()).add(value) } } diff --git a/packages/client/src/rpc-serializer.ts b/packages/client/src/rpc-serializer.ts index 54bcb00e0..55d16d6e2 100644 --- a/packages/client/src/rpc-serializer.ts +++ b/packages/client/src/rpc-serializer.ts @@ -1,4 +1,9 @@ -import { isAsyncIteratorObject, isPlainObject, stringifyJSON } from '@orpc/shared' +import { + isAsyncIteratorObject, + isPlainObject, + NON_NEGATIVE_INTEGER_REGEX, + stringifyJSON, +} from '@orpc/shared' import type { StandardBody } from '@standard-server/core' import { EventStreamErrorEvent } from '@standard-server/core' @@ -137,7 +142,7 @@ export class RPCSerializer extends RPCJsonSerializer { for (const [key, value] of data) { if (key === 'data' && raw === undefined && typeof value === 'string') { raw = value - } else if (value instanceof Blob && BLOB_KEY_REGEX.test(key)) { + } else if (value instanceof Blob && NON_NEGATIVE_INTEGER_REGEX.test(key)) { blobs[Number(key)] = value hasBlobs = true } else { @@ -156,8 +161,6 @@ export class RPCSerializer extends RPCJsonSerializer { } } -const BLOB_KEY_REGEX = /^(?:0|[1-9]\d*)$/ - function invalidFormat(): TypeError { return new TypeError('Invalid RPC serialized data format.') } diff --git a/packages/json-schema/src/coercer.ts b/packages/json-schema/src/coercer.ts index 632b7b70f..48f1c84dc 100644 --- a/packages/json-schema/src/coercer.ts +++ b/packages/json-schema/src/coercer.ts @@ -1,8 +1,10 @@ import { DEFAULT_MAX_BIGINT_DIGITS, getOwn, - isPlainObject, + INTEGER_REGEX, isNonPrimitive, + isPlainObject, + isPlainObjectOrArray, parseBigIntString, toArray, tryOrUndefined, @@ -553,7 +555,7 @@ function countValues(input: unknown, limit: number): number { while (stack.length) { const value = stack.pop() - if ((Array.isArray(value) || isPlainObject(value)) && !walked.has(value)) { + if (isPlainObjectOrArray(value) && !walked.has(value)) { walked.add(value) const items = Array.isArray(value) ? value : Object.values(value) @@ -591,9 +593,8 @@ function stringToNumber(value: string): number | string { return num } -const INTEGER_PATTERN = /^-?(?:0|[1-9]\d*)$/ function stringToInteger(value: string): number | string { - if (!INTEGER_PATTERN.test(value)) { + if (!INTEGER_REGEX.test(value)) { return value } diff --git a/packages/mcp/src/registrar.ts b/packages/mcp/src/registrar.ts index e7db6d202..c61a57342 100644 --- a/packages/mcp/src/registrar.ts +++ b/packages/mcp/src/registrar.ts @@ -52,7 +52,7 @@ import { encodeBase64, getOwn, isAsyncIteratorObject, - isPlainObject, + isPlainObjectOrArray, loadBytes, pathToHttpPath, resolveMaybeOptionalOptions, @@ -538,7 +538,7 @@ function collectNestedBlobs( ): void { if (json instanceof Blob) { blobs.push([toResourceUri(segments, json), json]) - } else if (Array.isArray(json) || isPlainObject(json)) { + } else if (isPlainObjectOrArray(json)) { for (const key of Object.keys(json)) { segments.push(key) collectNestedBlobs((json as Record)[key], segments, blobs) diff --git a/packages/msw/src/http-procedure-utils.ts b/packages/msw/src/http-procedure-utils.ts index 681620819..ae331d12b 100644 --- a/packages/msw/src/http-procedure-utils.ts +++ b/packages/msw/src/http-procedure-utils.ts @@ -11,7 +11,7 @@ import type { Context, ProcedureConfig, ProcedureHandler, Router } from '@orpc/s import { implement } from '@orpc/server' import type { FetchHandler, FetchHandlerHandleResult } from '@orpc/server/fetch' import type { Promisable, Value } from '@orpc/shared' -import { value } from '@orpc/shared' +import { listenAbort, value } from '@orpc/shared' import type { HttpHandler, HttpRequestResolverExtras, PathParams, ResponseResolverInfo } from 'msw' import { http, passthrough } from 'msw' @@ -163,12 +163,9 @@ export class HTTPProcedureUtils< return this.handler( ({ signal }) => new Promise((_, reject) => { - if (signal?.aborted) { - reject(signal.reason) - return + if (signal) { + listenAbort(signal, () => reject(signal.reason)) } - - signal?.addEventListener('abort', () => reject(signal.reason), { once: true }) }), ) } diff --git a/packages/nest/src/implement.ts b/packages/nest/src/implement.ts index f2df2cce6..6adea3204 100644 --- a/packages/nest/src/implement.ts +++ b/packages/nest/src/implement.ts @@ -36,6 +36,7 @@ import { StandardHandler } from '@orpc/server/standard' import type { Promisable } from '@orpc/shared' import { isAsyncIteratorObject, + listenAbort, mergeHttpPath, NullProtoObj, stringifyJSON, @@ -431,10 +432,8 @@ function toStreamableFile( const type = flattenStandardHeader(headers['content-type']) headers['content-type'] = undefined - if (signal?.aborted) { - stream.destroy() - } else { - signal?.addEventListener('abort', () => stream.destroy(), { once: true }) + if (signal) { + listenAbort(signal, () => stream.destroy()) } return new StreamableFile(stream, { type }) diff --git a/packages/openapi/src/adapters/standard/openapi-link-codec.ts b/packages/openapi/src/adapters/standard/openapi-link-codec.ts index 99851bc80..0c40a0461 100644 --- a/packages/openapi/src/adapters/standard/openapi-link-codec.ts +++ b/packages/openapi/src/adapters/standard/openapi-link-codec.ts @@ -93,7 +93,6 @@ export interface OpenAPILinkCodecOptions { errorResponseContentType?: string | undefined } -const END_SLASH_REGEX = /\/$/ const DOT_SEGMENT_REGEX = /(?:^|\/)\.{1,2}(?:\/|$)/ export class OpenAPILinkCodec implements StandardLinkCodec { @@ -170,7 +169,7 @@ export class OpenAPILinkCodec implements StandardLinkCo data = Object.keys(remaining).length > 0 ? remaining : undefined } - pathname = `${basePathname.replace(END_SLASH_REGEX, '')}${pathname}` as `/${string}` + pathname = mergeHttpPath(basePathname, pathname) if (isBodylessMethod(method)) { const queryString = this.serializeQueryString(data, meta?.queryStyles) @@ -236,7 +235,7 @@ export class OpenAPILinkCodec implements StandardLinkCo headers = mergeStandardHeaders(headers, serializeHeaders(input.headers, this.serializer)) } - pathname = `${basePathname.replace(END_SLASH_REGEX, '')}${pathname}` as `/${string}` + pathname = mergeHttpPath(basePathname, pathname) const queryString = this.serializeQueryString(input?.query, meta?.queryStyles) const search = combineSearch(baseSearch, queryString) const url = `${pathname}${search ?? ''}${baseHash ?? ''}` as StandardUrl diff --git a/packages/openapi/src/adapters/standard/openapi-matcher.ts b/packages/openapi/src/adapters/standard/openapi-matcher.ts index aa1da7b08..76b24d001 100644 --- a/packages/openapi/src/adapters/standard/openapi-matcher.ts +++ b/packages/openapi/src/adapters/standard/openapi-matcher.ts @@ -8,7 +8,7 @@ import { walkProcedureContractsSync, } from '@orpc/server' import type { Value } from '@orpc/shared' -import { mergeHttpPath, pathToHttpPath, value } from '@orpc/shared' +import { mergeHttpPath, pathToHttpPath, stripHttpPathPrefix, value } from '@orpc/shared' import { DEFAULT_OPENAPI_METHOD } from '../../constants.ts' import { getOpenAPIMeta } from '../../meta.ts' @@ -90,21 +90,13 @@ export class OpenAPIMatcher { | undefined > { if (prefix) { - if (!pathname.startsWith(prefix)) { - return undefined - } + const stripped = stripHttpPathPrefix(pathname, prefix) - const charAfterPrefix = pathname[prefix.length] - - if (charAfterPrefix === '/') { - pathname = pathname.slice(prefix.length) as `/${string}` - } else if (charAfterPrefix === undefined) { - pathname = '/' - } else if (prefix[prefix.length - 1] === '/') { - pathname = pathname.slice(prefix.length - 1) as `/${string}` - } else { + if (stripped === undefined) { return undefined } + + pathname = stripped } // most requests `await undefined` so conditionally await it to save a microtask turn diff --git a/packages/openapi/src/helpers/form-data.ts b/packages/openapi/src/helpers/form-data.ts index 12c5565c4..8312ec6b9 100644 --- a/packages/openapi/src/helpers/form-data.ts +++ b/packages/openapi/src/helpers/form-data.ts @@ -1,5 +1,5 @@ import { isSchemaIssue } from '@orpc/contract' -import { isNonPrimitive } from '@orpc/shared' +import { isNonPrimitive, NON_NEGATIVE_INTEGER_REGEX } from '@orpc/shared' import { BracketNotationSerializer } from '../bracket-notation.ts' @@ -97,7 +97,7 @@ export function getIssueMessage(error: unknown, path: string): string | undefine return issue.message } - if (path === '' && /^(?:0|[1-9]\d*)$/.test(issuePath)) { + if (path === '' && NON_NEGATIVE_INTEGER_REGEX.test(issuePath)) { return issue.message } } diff --git a/packages/openapi/src/openapi-generator-components.ts b/packages/openapi/src/openapi-generator-components.ts index f7dd8a9dc..b2b162f28 100644 --- a/packages/openapi/src/openapi-generator-components.ts +++ b/packages/openapi/src/openapi-generator-components.ts @@ -6,7 +6,7 @@ import { getJsonSchemaKeywordKind, mapJsonSchemaRefs, } from '@orpc/json-schema' -import { getOwn, isDeepEqual, isPlainObject, setOwn } from '@orpc/shared' +import { getOwn, isDeepEqual, isPlainObject, isPlainObjectOrArray, setOwn } from '@orpc/shared' import type { OpenAPIV3_2 } from './types.ts' @@ -234,10 +234,6 @@ interface ReuseComparisonContext { visited: WeakMap> } -function isSchemaContainer(value: unknown): value is unknown[] | Record { - return Array.isArray(value) || isPlainObject(value) -} - /** * Where a value sits, mirroring how hoisting rewrites refs: only refs in subschemas are * rewritten, so literal data (`const`, `default`, `examples`, ...) is compared as is. @@ -255,7 +251,7 @@ function areSchemasEquivalentForReuse( } // Only arrays and plain objects carry schema structure; primitives and values like Date are compared as a whole - if (!isSchemaContainer(candidate) || !isSchemaContainer(existing)) { + if (!isPlainObjectOrArray(candidate) || !isPlainObjectOrArray(existing)) { return isDeepEqual(candidate, existing) } diff --git a/packages/pinia-colada/src/stream-query.ts b/packages/pinia-colada/src/stream-query.ts index 824286f4e..b65108136 100644 --- a/packages/pinia-colada/src/stream-query.ts +++ b/packages/pinia-colada/src/stream-query.ts @@ -1,5 +1,5 @@ import type { Promisable } from '@orpc/shared' -import { throwIfAborted } from '@orpc/shared' +import { takeLast, throwIfAborted } from '@orpc/shared' import type { UseQueryFnContext } from './types.ts' @@ -78,7 +78,7 @@ export function serializableStreamedQuery( } const current = entry.state.value.data as T[] | undefined - let data = write(refetchMode === 'append' && current ? limitArraySize(current, maxChunks) : []) + let data = write(refetchMode === 'append' && current ? takeLast(current, maxChunks) : []) if (settleOnFirstData) { await stream[Symbol.asyncIterator]().return?.() @@ -90,7 +90,7 @@ export function serializableStreamedQuery( // read the entry, so writes made during the stream are kept const current = (entry.state.value.data as T[] | undefined) ?? [] - data = write(limitArraySize([...current, chunk], maxChunks)) + data = write(takeLast([...current, chunk], maxChunks)) } // a stream that ends on abort must not be stored as fresh data @@ -103,14 +103,6 @@ export function serializableStreamedQuery( } // only an array is trimmed; another value written after the stream (e.g. `null`) is kept as is - return Array.isArray(cached) ? limitArraySize(cached, maxChunks) : cached + return Array.isArray(cached) ? takeLast(cached, maxChunks) : cached } } - -function limitArraySize(items: T[], maxSize: number): T[] { - if (items.length <= maxSize) { - return items - } - - return items.slice(items.length - maxSize) -} diff --git a/packages/server/src/adapters/standard/rpc-matcher.ts b/packages/server/src/adapters/standard/rpc-matcher.ts index 2b6776cd0..89a02aa00 100644 --- a/packages/server/src/adapters/standard/rpc-matcher.ts +++ b/packages/server/src/adapters/standard/rpc-matcher.ts @@ -1,6 +1,6 @@ import type { AnyProcedureContract } from '@orpc/contract' import type { Value } from '@orpc/shared' -import { normalizeHttpPath, pathToHttpPath, value } from '@orpc/shared' +import { normalizeHttpPath, pathToHttpPath, stripHttpPathPrefix, value } from '@orpc/shared' import type { StandardMethod } from '@standard-server/core' import { unlazy } from '../../lazy.ts' @@ -126,21 +126,13 @@ export class RPCMatcher { } if (prefix) { - if (!pathname.startsWith(prefix)) { - return undefined - } + const stripped = stripHttpPathPrefix(pathname, prefix) - const charAfterPrefix = pathname[prefix.length] - - if (charAfterPrefix === '/') { - pathname = pathname.slice(prefix.length) as `/${string}` - } else if (charAfterPrefix === undefined) { - pathname = '/' - } else if (prefix[prefix.length - 1] === '/') { - pathname = pathname.slice(prefix.length - 1) as `/${string}` - } else { + if (stripped === undefined) { return undefined } + + pathname = stripped } if (pathname.length > 1 && pathname.endsWith('/')) { diff --git a/packages/server/src/plugins/response-compression.ts b/packages/server/src/plugins/response-compression.ts index e42d08f1b..873d5e3f5 100644 --- a/packages/server/src/plugins/response-compression.ts +++ b/packages/server/src/plugins/response-compression.ts @@ -1,5 +1,7 @@ import type { CompressionEncoding } from '@orpc/shared' import { + AVG_UTF8_BYTES_PER_CHAR, + estimateCompressibleFormDataSize, isAcceptableEncoding, isAsyncIteratorObject, isCompressibleContentType, @@ -20,10 +22,6 @@ import type { } from '../adapters/standard/index.ts' import type { Context } from '../context.ts' -// Rough UTF-8 estimate. Mostly ASCII text stays close to 1 byte/char; -// occasional multi-byte characters increase the average. -const AVG_BYTES_PER_CHAR = 1.2 - export interface ResponseCompressionHandlerPluginOptions { /** * The compression schemes to use for response compression. @@ -166,32 +164,9 @@ export class ResponseCompressionHandlerPlugin< } } } else if (body instanceof FormData) { - const PART_OVERHEAD = 64 // approx bytes for boundary + Content-Disposition/Content-Type headers per part - - let contentLength = 0 - for (const [key, value] of body) { - contentLength += PART_OVERHEAD + key.length - - if (value instanceof Blob) { - if (!Number.isFinite(value.size)) { - // Bun-s3 can use NaN for size - if (!this.isCompressibleContentType(value.type, interceptorOptions)) { - // Unknown non-compressible part size makes the estimate unreliable - contentLength = -Infinity - break - } - - // Unknown size for compressible content - still apply compression - contentLength = Infinity - } else { - contentLength += this.isCompressibleContentType(value.type, interceptorOptions) - ? value.size - : -value.size - } - } else { - contentLength += value.length * AVG_BYTES_PER_CHAR - } - } + const contentLength = estimateCompressibleFormDataSize(body, (contentType) => + this.isCompressibleContentType(contentType, interceptorOptions), + ) if (contentLength >= this.threshold) { const res = new Response(body) @@ -212,7 +187,7 @@ export class ResponseCompressionHandlerPlugin< } } else if (body instanceof URLSearchParams) { const string = body.toString() - if (string.length * AVG_BYTES_PER_CHAR >= this.threshold) { + if (string.length * AVG_UTF8_BYTES_PER_CHAR >= this.threshold) { const [, bodyHeaders] = toFetchBody(body, headers) return { @@ -230,7 +205,7 @@ export class ResponseCompressionHandlerPlugin< } } else if (body !== undefined && !isAsyncIteratorObject(body)) { const [string, bodyHeaders] = toFetchBody(body, headers) as [string, StandardHeaders] - if (string.length * AVG_BYTES_PER_CHAR >= this.threshold) { + if (string.length * AVG_UTF8_BYTES_PER_CHAR >= this.threshold) { return { ...result, response: { diff --git a/packages/server/src/plugins/signal-sync.ts b/packages/server/src/plugins/signal-sync.ts index 1214d4b6e..905b3ad4a 100644 --- a/packages/server/src/plugins/signal-sync.ts +++ b/packages/server/src/plugins/signal-sync.ts @@ -4,6 +4,7 @@ import { anyAbortSignal, AsyncIteratorClass, isAsyncIteratorObject, + listenAbort, override, runWithSignal, toArray, @@ -52,16 +53,6 @@ export class SignalSyncHandlerPlugin implements StandardHandl } } -function listenAbort(signal: AbortSignal, onAbort: () => void): () => void { - if (signal.aborted) { - onAbort() - } else { - signal.addEventListener('abort', onAbort, { once: true }) - } - - return () => signal.removeEventListener('abort', onAbort) -} - function syncAsyncIterator( iterator: AsyncIterator, signal: AbortSignal, diff --git a/packages/shared/src/array.test.ts b/packages/shared/src/array.test.ts index 3cc467525..ebfd716b7 100644 --- a/packages/shared/src/array.test.ts +++ b/packages/shared/src/array.test.ts @@ -1,4 +1,4 @@ -import { splitInHalf, toArray } from './array.ts' +import { splitInHalf, takeLast, toArray } from './array.ts' it('toArray', () => { expect(toArray(undefined)).toEqual([]) @@ -21,3 +21,12 @@ it('splitInHalf', () => { expect(splitInHalf([1])).toEqual([[1], []]) expect(splitInHalf([])).toEqual([[], []]) }) + +it('takeLast', () => { + const items = [1, 2, 3] + + expect(takeLast(items, 3)).toBe(items) + expect(takeLast(items, 5)).toBe(items) + expect(takeLast(items, 2)).toEqual([2, 3]) + expect(takeLast(items, 0)).toEqual([]) +}) diff --git a/packages/shared/src/array.ts b/packages/shared/src/array.ts index ba37808f0..5663f9a75 100644 --- a/packages/shared/src/array.ts +++ b/packages/shared/src/array.ts @@ -10,3 +10,14 @@ export function splitInHalf(arr: readonly T[]): [T[], T[]] { const half = Math.ceil(arr.length / 2) return [arr.slice(0, half), arr.slice(half)] } + +/** + * Keeps only the last `maxSize` items, returning the same array when it already fits. + */ +export function takeLast(items: T[], maxSize: number): T[] { + if (items.length <= maxSize) { + return items + } + + return items.slice(items.length - maxSize) +} diff --git a/packages/shared/src/bigint.ts b/packages/shared/src/bigint.ts index 9b90ec76d..b8548a6d9 100644 --- a/packages/shared/src/bigint.ts +++ b/packages/shared/src/bigint.ts @@ -1,13 +1,10 @@ +import { INTEGER_REGEX } from './regex.ts' + /** * Default maximum digits, excluding a leading `-`, when parsing a `bigint` from a string. */ export const DEFAULT_MAX_BIGINT_DIGITS = 4300 -/** - * Matches `bigint.toString()` output only. - */ -const BIGINT_PATTERN = /^-?(?:0|[1-9]\d*)$/ - /** * Parses a `bigint.toString()` string, or returns `undefined` if it is invalid or has more than `maxDigits` digits. * The length is checked first, since `BigInt()` is super-linear in the digit count. @@ -15,7 +12,7 @@ const BIGINT_PATTERN = /^-?(?:0|[1-9]\d*)$/ export function parseBigIntString(value: string, maxDigits: number): bigint | undefined { const digits = value.startsWith('-') ? value.length - 1 : value.length - if (digits > maxDigits || !BIGINT_PATTERN.test(value)) { + if (digits > maxDigits || !INTEGER_REGEX.test(value)) { return undefined } diff --git a/packages/shared/src/http.test.ts b/packages/shared/src/http.test.ts index 324a1ff6e..0cd1f3168 100644 --- a/packages/shared/src/http.test.ts +++ b/packages/shared/src/http.test.ts @@ -1,4 +1,6 @@ import { + AVG_UTF8_BYTES_PER_CHAR, + estimateCompressibleFormDataSize, isAcceptableEncoding, isCompressibleContentType, isCompressionEncoding, @@ -10,6 +12,7 @@ import { parseAcceptEncodingQualities, parseContentEncodings, pathToHttpPath, + stripHttpPathPrefix, varyByAcceptEncoding, } from './http.ts' @@ -120,6 +123,20 @@ describe('matchesHttpPathPrefix', () => { }) }) +describe('stripHttpPathPrefix', () => { + it.each([ + ['/api/ping', '/api', '/ping'], + ['/api', '/api', '/'], + ['/api/', '/api', '/'], + ['/api/ping', '/api/', '/ping'], + ['/apis/ping', '/api', undefined], + ['/other', '/api', undefined], + ['/api?x', '/api', undefined], + ] as const)('%s without %s is %s', (pathname, prefix, expected) => { + expect(stripHttpPathPrefix(pathname, prefix)).toBe(expected) + }) +}) + describe('matchesHttpPath', () => { it.each([ ['/api', '/api', true, 'matches exact path'], @@ -314,3 +331,38 @@ describe('isNoTransformCacheControl', () => { expect(isNoTransformCacheControl('no-transform-extension')).toBe(false) }) }) + +describe('estimateCompressibleFormDataSize', () => { + const isCompressible = (type: string) => type.startsWith('text/') + + it('counts strings, compressible blobs, and per-part overhead', () => { + const form = new FormData() + form.append('a', 'x'.repeat(10)) + form.append('b', new Blob(['y'.repeat(100)], { type: 'text/plain' })) + + expect(estimateCompressibleFormDataSize(form, isCompressible)).toBe( + 64 + 1 + 10 * AVG_UTF8_BYTES_PER_CHAR + 64 + 1 + 100, + ) + }) + + it('subtracts non-compressible blobs', () => { + const form = new FormData() + form.append('a', new Blob(['y'.repeat(100)], { type: 'image/png' })) + + expect(estimateCompressibleFormDataSize(form, isCompressible)).toBe(64 + 1 - 100) + }) + + it('handles blobs with an unknown size', () => { + const compressible = new Blob(['a'], { type: 'text/plain' }) + Object.defineProperty(compressible, 'size', { value: Number.NaN }) + const nonCompressible = new Blob(['a'], { type: 'image/png' }) + Object.defineProperty(nonCompressible, 'size', { value: Number.NaN }) + + const form = new FormData() + form.append('a', compressible) + expect(estimateCompressibleFormDataSize(form, isCompressible)).toBe(Infinity) + + form.append('b', nonCompressible) + expect(estimateCompressibleFormDataSize(form, isCompressible)).toBe(-Infinity) + }) +}) diff --git a/packages/shared/src/http.ts b/packages/shared/src/http.ts index 6154fdcd5..47fbfd98e 100644 --- a/packages/shared/src/http.ts +++ b/packages/shared/src/http.ts @@ -35,6 +35,35 @@ export function matchesHttpPathPrefix(url: `/${string}`, prefix: `/${string}`): ) } +/** + * Removes `prefix` from `pathname`, or returns `undefined` when `pathname` is not under `prefix`. + * The rest of the path must start at a segment boundary, so `/apis` is not under `/api`. + */ +export function stripHttpPathPrefix( + pathname: `/${string}`, + prefix: `/${string}`, +): `/${string}` | undefined { + if (!pathname.startsWith(prefix)) { + return undefined + } + + const charAfterPrefix = pathname[prefix.length] + + if (charAfterPrefix === '/') { + return pathname.slice(prefix.length) as `/${string}` + } + + if (charAfterPrefix === undefined) { + return '/' + } + + if (prefix[prefix.length - 1] === '/') { + return pathname.slice(prefix.length - 1) as `/${string}` + } + + return undefined +} + export function matchesHttpPath(url: `/${string}`, path: `/${string}`): boolean { const pathWithoutEndSlash = path.endsWith('/') ? path.slice(0, path.length - 1) : path @@ -76,6 +105,49 @@ export function parseContentEncodings(header: string | undefined): string[] { return header.split(',').map((part) => part.trim().toLowerCase()) } +/** + * Rough UTF-8 bytes per string character. Mostly ASCII text stays close to 1 byte/char; + * occasional multi-byte characters increase the average. + */ +export const AVG_UTF8_BYTES_PER_CHAR = 1.2 + +// approx bytes for boundary + Content-Disposition/Content-Type headers per part +const FORM_DATA_PART_OVERHEAD = 64 + +/** + * Estimates how many bytes of a FormData body are worth compressing. Non-compressible blobs count + * against the total. Returns `Infinity` when a compressible blob has an unknown size, and + * `-Infinity` when a non-compressible one does, since the estimate is then unreliable. + */ +export function estimateCompressibleFormDataSize( + formData: FormData, + isCompressible: (contentType: string) => boolean, +): number { + let size = 0 + + for (const [key, value] of formData) { + size += FORM_DATA_PART_OVERHEAD + key.length + + if (value instanceof Blob) { + if (!Number.isFinite(value.size)) { + // Bun-s3 can use NaN for size + if (!isCompressible(value.type)) { + return -Infinity + } + + // Unknown size for compressible content - still apply compression + size = Infinity + } else { + size += isCompressible(value.type) ? value.size : -value.size + } + } else { + size += value.length * AVG_UTF8_BYTES_PER_CHAR + } + } + + return size +} + const ACCEPT_ENCODING_QUALITY_REGEX = /^\s*q=([\d.]+)\s*$/i /** diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index 91bed8782..98ba6ea02 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -15,6 +15,7 @@ export * from './plugin.ts' export * from './promise.ts' export * from './proxy.ts' export * from './queue.ts' +export * from './regex.ts' export * from './signal.ts' export * from './stream.ts' export * from './tracing.ts' diff --git a/packages/shared/src/object.test.ts b/packages/shared/src/object.test.ts index c43e90946..7a07e7d42 100644 --- a/packages/shared/src/object.test.ts +++ b/packages/shared/src/object.test.ts @@ -12,6 +12,7 @@ import { getConstructors, getOwn, isPlainObject, + isPlainObjectOrArray, isPropertyKey, mergeTwoLevels, NullProtoObj, @@ -807,3 +808,13 @@ describe('bindMethods', () => { expect(methods.double()).toBe(246) }) }) + +it('isPlainObjectOrArray', () => { + expect(isPlainObjectOrArray([])).toBe(true) + expect(isPlainObjectOrArray({})).toBe(true) + expect(isPlainObjectOrArray(Object.create(null))).toBe(true) + expect(isPlainObjectOrArray(new Date())).toBe(false) + expect(isPlainObjectOrArray(new Map())).toBe(false) + expect(isPlainObjectOrArray(null)).toBe(false) + expect(isPlainObjectOrArray('a')).toBe(false) +}) diff --git a/packages/shared/src/object.ts b/packages/shared/src/object.ts index 50ba1ee05..acaf1d92a 100644 --- a/packages/shared/src/object.ts +++ b/packages/shared/src/object.ts @@ -68,6 +68,15 @@ export function isPlainObject(value: unknown): value is Record { + return Array.isArray(value) || isPlainObject(value) +} + /** * Returns `object[key]` only when it is an own property, so keys like * `toString` do not resolve through the prototype chain. diff --git a/packages/shared/src/regex.test.ts b/packages/shared/src/regex.test.ts new file mode 100644 index 000000000..709f9a3ac --- /dev/null +++ b/packages/shared/src/regex.test.ts @@ -0,0 +1,26 @@ +import { INTEGER_REGEX, NON_NEGATIVE_INTEGER_REGEX } from './regex.ts' + +it.each([ + ['0', true], + ['42', true], + ['-42', true], + ['-0', true], + ['007', false], + ['1.5', false], + ['+1', false], + ['', false], + ['1e3', false], +])('INTEGER_REGEX %s: %s', (value, expected) => { + expect(INTEGER_REGEX.test(value)).toBe(expected) +}) + +it.each([ + ['0', true], + ['42', true], + ['-1', false], + ['01', false], + ['1.0', false], + ['', false], +])('NON_NEGATIVE_INTEGER_REGEX %s: %s', (value, expected) => { + expect(NON_NEGATIVE_INTEGER_REGEX.test(value)).toBe(expected) +}) diff --git a/packages/shared/src/regex.ts b/packages/shared/src/regex.ts new file mode 100644 index 000000000..f261c4f4f --- /dev/null +++ b/packages/shared/src/regex.ts @@ -0,0 +1,10 @@ +/** + * Matches an integer string: an optional `-`, then digits without leading zeros (`007` is rejected). + * Note that `-0` is accepted. + */ +export const INTEGER_REGEX = /^-?(?:0|[1-9]\d*)$/ + +/** + * Matches a non-negative integer string without leading zeros, such as an array index. + */ +export const NON_NEGATIVE_INTEGER_REGEX = /^(?:0|[1-9]\d*)$/ diff --git a/packages/shared/src/signal.test.ts b/packages/shared/src/signal.test.ts index ffa8a2259..6051bddf7 100644 --- a/packages/shared/src/signal.test.ts +++ b/packages/shared/src/signal.test.ts @@ -1,6 +1,6 @@ import { getEventListeners } from 'node:events' -import { anyAbortSignal, disposableAllAbortSignal, runWithSignal } from './signal.ts' +import { anyAbortSignal, disposableAllAbortSignal, listenAbort, runWithSignal } from './signal.ts' describe('disposableAllAbortSignal', () => { it('returns an undefined signal and a no-op dispose if contains undefined or empty array', () => { @@ -250,3 +250,40 @@ describe('runWithSignal', () => { expect(removeSpy).toHaveBeenCalled() }) }) + +describe('listenAbort', () => { + it('calls the listener right away when already aborted', () => { + const controller = new AbortController() + controller.abort() + const onAbort = vi.fn() + + listenAbort(controller.signal, onAbort) + + expect(onAbort).toHaveBeenCalledTimes(1) + expect(getEventListeners(controller.signal, 'abort')).toHaveLength(0) + }) + + it('calls the listener once on abort', () => { + const controller = new AbortController() + const onAbort = vi.fn() + + listenAbort(controller.signal, onAbort) + expect(onAbort).not.toHaveBeenCalled() + + controller.abort() + controller.abort() + expect(onAbort).toHaveBeenCalledTimes(1) + }) + + it('returns a function that removes the listener', () => { + const controller = new AbortController() + const onAbort = vi.fn() + + const remove = listenAbort(controller.signal, onAbort) + remove() + controller.abort() + + expect(onAbort).not.toHaveBeenCalled() + expect(getEventListeners(controller.signal, 'abort')).toHaveLength(0) + }) +}) diff --git a/packages/shared/src/signal.ts b/packages/shared/src/signal.ts index d211a1a7f..c44e93a52 100644 --- a/packages/shared/src/signal.ts +++ b/packages/shared/src/signal.ts @@ -120,3 +120,17 @@ export async function runWithSignal( } } } + +/** + * Calls `onAbort` right away if the signal is already aborted, otherwise once when it aborts. + * Returns a function that removes the listener. + */ +export function listenAbort(signal: AbortSignal, onAbort: () => void): () => void { + if (signal.aborted) { + onAbort() + } else { + signal.addEventListener('abort', onAbort, { once: true }) + } + + return () => signal.removeEventListener('abort', onAbort) +} diff --git a/packages/tanstack-query/src/stream-query.ts b/packages/tanstack-query/src/stream-query.ts index 4cd79e8be..4713bc1eb 100644 --- a/packages/tanstack-query/src/stream-query.ts +++ b/packages/tanstack-query/src/stream-query.ts @@ -1,5 +1,5 @@ import type { Promisable } from '@orpc/shared' -import { throwIfAborted } from '@orpc/shared' +import { takeLast, throwIfAborted } from '@orpc/shared' import type { QueryFunction, QueryFunctionContext, QueryKey } from '@tanstack/query-core' export interface SerializableStreamedQueryOptions { @@ -72,7 +72,7 @@ export function serializableStreamedQuery< } let data = client.setQueryData(queryKey, (current) => - refetchMode === 'append' && current ? limitArraySize(current, maxChunks) : [], + refetchMode === 'append' && current ? takeLast(current, maxChunks) : [], )! if (settleOnFirstData) { @@ -85,7 +85,7 @@ export function serializableStreamedQuery< // read the cache, so writes made during the stream are kept data = client.setQueryData(queryKey, (current = []) => - limitArraySize([...current, chunk], maxChunks), + takeLast([...current, chunk], maxChunks), )! } @@ -96,17 +96,9 @@ export function serializableStreamedQuery< } // only an array is trimmed; another value written after the stream (e.g. `null`) is kept as is - return Array.isArray(cached) ? limitArraySize(cached, maxChunks) : cached + return Array.isArray(cached) ? takeLast(cached, maxChunks) : cached } finally { signal.removeEventListener('abort', markStale) } } } - -function limitArraySize(items: T[], maxSize: number): T[] { - if (items.length <= maxSize) { - return items - } - - return items.slice(items.length - maxSize) -} From 7bd5737b9234cb138119fa4227bc7cdad51b69e7 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 8 Nov 2026 01:40:03 +0000 Subject: [PATCH 3/5] test: cover every branch in files touched by the shared dedupe listenAbort now accepts an undefined signal and passes the abort reason, so the msw and nest call sites no longer need their own branch. Add tests for the remaining uncovered branches in the touched files. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VVaw7PANP3R1NnJpBLkx5b --- packages/client/src/plugins/dedupe.ts | 7 +- .../client/src/rpc-json-serializer.test.ts | 13 ++++ packages/json-schema/src/coercer.test.ts | 3 + packages/msw/src/http-procedure-utils.ts | 4 +- packages/nest/src/implement.ts | 4 +- .../pinia-colada/src/stream-query.test.tsx | 16 +++++ .../server/src/plugins/signal-sync.test.ts | 64 +++++++++++++++++++ packages/shared/src/signal.test.ts | 13 +++- packages/shared/src/signal.ts | 21 ++++-- 9 files changed, 128 insertions(+), 17 deletions(-) diff --git a/packages/client/src/plugins/dedupe.ts b/packages/client/src/plugins/dedupe.ts index a37c84d68..fba88965d 100644 --- a/packages/client/src/plugins/dedupe.ts +++ b/packages/client/src/plugins/dedupe.ts @@ -344,10 +344,9 @@ function closeReadableStreamOnAbort( } const reader = replica.getReader() - const removeAbortListener = listenAbort( - signal, - () => void reader.cancel(signal.reason).catch(() => {}), - ) + const removeAbortListener = listenAbort(signal, (reason) => { + reader.cancel(reason).catch(() => {}) + }) return new ReadableStream( { diff --git a/packages/client/src/rpc-json-serializer.test.ts b/packages/client/src/rpc-json-serializer.test.ts index 9b306035a..17b2f7df0 100644 --- a/packages/client/src/rpc-json-serializer.test.ts +++ b/packages/client/src/rpc-json-serializer.test.ts @@ -291,6 +291,19 @@ describe('rpcJsonSerializer: custom handlers', () => { expect(roundTripThroughWire(serializer, { value: date })).toEqual({ value: date }) }) + it('serializes like the defaults when no custom handler is set', () => { + const serializer = new RPCJsonSerializer({ + handlers: { + unused: undefined, + }, + }) + + const date = new Date('2023-01-01') + const data = { value: date, list: [undefined], text: 'a' } + expect(serializer.serializeJson(data)).toEqual(new RPCJsonSerializer().serializeJson(data)) + expect(roundTripThroughWire(serializer, data)).toEqual(data) + }) + it('supports disabling default handlers', () => { const serializer = new RPCJsonSerializer({ handlers: { diff --git a/packages/json-schema/src/coercer.test.ts b/packages/json-schema/src/coercer.test.ts index 92fe2672a..8020176b7 100644 --- a/packages/json-schema/src/coercer.test.ts +++ b/packages/json-schema/src/coercer.test.ts @@ -191,6 +191,9 @@ describe('jsonSchemaCoercer', () => { expect(coerce(DATE_SCHEMA, '2020-02-30T06:15:00Z')).toBe('2020-02-30T06:15:00Z') expect(coerce(DATE_SCHEMA, '2020-02-30 06:15')).toBe('2020-02-30 06:15') expect(coerce(DATE_SCHEMA, '0050-02-29')).toBe('0050-02-29') + // century years are leap years only when divisible by 400 + expect(coerce(DATE_SCHEMA, '1900-02-29')).toBe('1900-02-29') + expect(coerce(DATE_SCHEMA, '2000-02-29')).toEqual(new Date('2000-02-29')) // real days stay coerced, including a leap day and years below 100 expect(coerce(DATE_SCHEMA, '2020-02-29')).toEqual(new Date('2020-02-29')) diff --git a/packages/msw/src/http-procedure-utils.ts b/packages/msw/src/http-procedure-utils.ts index ae331d12b..99fc61bfe 100644 --- a/packages/msw/src/http-procedure-utils.ts +++ b/packages/msw/src/http-procedure-utils.ts @@ -163,9 +163,7 @@ export class HTTPProcedureUtils< return this.handler( ({ signal }) => new Promise((_, reject) => { - if (signal) { - listenAbort(signal, () => reject(signal.reason)) - } + listenAbort(signal, reject) }), ) } diff --git a/packages/nest/src/implement.ts b/packages/nest/src/implement.ts index 6adea3204..ca156752d 100644 --- a/packages/nest/src/implement.ts +++ b/packages/nest/src/implement.ts @@ -432,9 +432,7 @@ function toStreamableFile( const type = flattenStandardHeader(headers['content-type']) headers['content-type'] = undefined - if (signal) { - listenAbort(signal, () => stream.destroy()) - } + listenAbort(signal, () => stream.destroy()) return new StreamableFile(stream, { type }) } diff --git a/packages/pinia-colada/src/stream-query.test.tsx b/packages/pinia-colada/src/stream-query.test.tsx index feb49612d..fab261701 100644 --- a/packages/pinia-colada/src/stream-query.test.tsx +++ b/packages/pinia-colada/src/stream-query.test.tsx @@ -488,6 +488,22 @@ describe('serializableStreamedQuery', () => { await expect(queryCache.fetch(entry)).resolves.toMatchObject({ data: null }) }) + it('starts a new array when the entry data is cleared mid-stream', async () => { + const queryCache = createQueryCache() + const entry = queryCache.ensure({ + key: ['stream'], + query: serializableStreamedQuery(async () => + (async function* () { + yield 1 + entry.state.value = { status: 'pending', data: undefined, error: null } as any + yield 2 + })(), + ), + }) + + await expect(queryCache.fetch(entry)).resolves.toMatchObject({ data: [2] }) + }) + it('resolves with the streamed data when the entry data is cleared after the last chunk', async () => { const queryCache = createQueryCache() const entry = queryCache.ensure({ diff --git a/packages/server/src/plugins/signal-sync.test.ts b/packages/server/src/plugins/signal-sync.test.ts index 9f06474bd..71591ef69 100644 --- a/packages/server/src/plugins/signal-sync.test.ts +++ b/packages/server/src/plugins/signal-sync.test.ts @@ -276,4 +276,68 @@ describe('signalSyncHandlerPlugin', () => { expect((await promise).status).toBe(500) expect(signal?.reason).toBe(reason) }) + + describe('interceptor', () => { + function intercept(output: unknown, signal: AbortSignal): Promise { + const interceptor = new SignalSyncHandlerPlugin().init({}).clientInterceptors![0]! + return interceptor({ signal, next: async () => output } as any) + } + + it('ignores a rejecting iterator return when the signal aborts', async () => { + const iterator = { + next: () => new Promise(() => {}), + return: vi.fn(async () => { + throw new Error('return failed') + }), + [Symbol.asyncIterator]() { + return this + }, + } + + const controller = new AbortController() + await intercept(iterator, controller.signal) + controller.abort() + + expect(iterator.return).toHaveBeenCalledTimes(1) + await new Promise((resolve) => setTimeout(resolve, 0)) // any unhandled rejection would fail the test + }) + + it('ignores a rejecting stream cancel when the signal aborts', async () => { + const cancel = vi.fn(() => { + throw new Error('cancel failed') + }) + + const controller = new AbortController() + const synced: ReadableStream = await intercept( + new ReadableStream({ cancel }), + controller.signal, + ) + const reason = new Error('aborted') + controller.abort(reason) + + await expect(synced.getReader().read()).rejects.toBe(reason) + expect(cancel).toHaveBeenCalledExactlyOnceWith(reason) + await new Promise((resolve) => setTimeout(resolve, 0)) // any unhandled rejection would fail the test + }) + + it('drops a pending read that settles after the signal aborts', async () => { + const pulled = promiseWithResolvers() + + const controller = new AbortController() + const synced: ReadableStream = await intercept( + new ReadableStream({ pull: () => pulled.resolve() }, { highWaterMark: 0 }), + controller.signal, + ) + + const reader = synced.getReader() + const read = reader.read() + await pulled.promise + + const reason = new Error('aborted') + controller.abort(reason) + + await expect(read).rejects.toBe(reason) + await new Promise((resolve) => setTimeout(resolve, 0)) // let the pending inner read settle + }) + }) }) diff --git a/packages/shared/src/signal.test.ts b/packages/shared/src/signal.test.ts index 6051bddf7..453362420 100644 --- a/packages/shared/src/signal.test.ts +++ b/packages/shared/src/signal.test.ts @@ -260,9 +260,19 @@ describe('listenAbort', () => { listenAbort(controller.signal, onAbort) expect(onAbort).toHaveBeenCalledTimes(1) + expect(onAbort).toHaveBeenCalledWith(controller.signal.reason) expect(getEventListeners(controller.signal, 'abort')).toHaveLength(0) }) + it('does nothing without a signal', () => { + const onAbort = vi.fn() + + const remove = listenAbort(undefined, onAbort) + remove() + + expect(onAbort).not.toHaveBeenCalled() + }) + it('calls the listener once on abort', () => { const controller = new AbortController() const onAbort = vi.fn() @@ -270,9 +280,10 @@ describe('listenAbort', () => { listenAbort(controller.signal, onAbort) expect(onAbort).not.toHaveBeenCalled() - controller.abort() + controller.abort('reason') controller.abort() expect(onAbort).toHaveBeenCalledTimes(1) + expect(onAbort).toHaveBeenCalledWith('reason') }) it('returns a function that removes the listener', () => { diff --git a/packages/shared/src/signal.ts b/packages/shared/src/signal.ts index c44e93a52..5862cec14 100644 --- a/packages/shared/src/signal.ts +++ b/packages/shared/src/signal.ts @@ -122,15 +122,24 @@ export async function runWithSignal( } /** - * Calls `onAbort` right away if the signal is already aborted, otherwise once when it aborts. - * Returns a function that removes the listener. + * Calls `onAbort` with the abort reason right away if the signal is already aborted, otherwise once + * when it aborts. Does nothing without a signal. Returns a function that removes the listener. */ -export function listenAbort(signal: AbortSignal, onAbort: () => void): () => void { +export function listenAbort( + signal: AbortSignal | undefined, + onAbort: (reason: unknown) => void, +): () => void { + if (!signal) { + return () => {} + } + + const listener = () => onAbort(signal.reason) + if (signal.aborted) { - onAbort() + listener() } else { - signal.addEventListener('abort', onAbort, { once: true }) + signal.addEventListener('abort', listener, { once: true }) } - return () => signal.removeEventListener('abort', onAbort) + return () => signal.removeEventListener('abort', listener) } From ae810ed36a3974931a3bcded060fd5d2bcf63735 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 8 Nov 2026 02:01:08 +0000 Subject: [PATCH 4/5] refactor(shared): apply review fixes to the shared helpers - parseContentEncodings ignores empty list elements (RFC 9110 lists), so `Content-Encoding: gzip,` is still decoded - keep COMPRESSION_ENCODINGS private and replace the exported byte factor with estimateUtf8ByteLength - use CompressionEncoding in the client response and node batch compression options - use listenAbort for the remaining inline abort listeners (tanstack-query and pinia-colada stream/live queries, server batch) and takeLast in swr - tanstack-query streamed query no longer throws when the cache is set to null mid-stream - listenAbort: document the synchronous call and reuse one noop - tighten tests: swr error-after-unsubscribe tests now reach the abort path, the pinia mid-stream test asserts before the stream closes, the FormData size test appends Files, and a signal-sync test name matches what it checks Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VVaw7PANP3R1NnJpBLkx5b --- .../client/src/plugins/request-compression.ts | 6 ++-- .../src/plugins/response-compression.test.ts | 16 ++++++++++ .../src/plugins/response-compression.ts | 3 +- ...tch-response-compression-handler-plugin.ts | 7 ++--- packages/pinia-colada/src/live-query.ts | 10 ++----- .../pinia-colada/src/stream-query.test.tsx | 10 ++++++- packages/pinia-colada/src/stream-query.ts | 10 ++----- packages/server/src/plugins/batch.ts | 22 +++++++------- .../src/plugins/response-compression.ts | 6 ++-- .../server/src/plugins/signal-sync.test.ts | 2 +- packages/server/src/plugins/signal-sync.ts | 6 ++-- packages/shared/src/http.test.ts | 23 +++++++++++--- packages/shared/src/http.ts | 30 +++++++++++++------ packages/shared/src/signal.ts | 8 +++-- packages/swr/src/procedure-utils.test.ts | 16 ++++++++-- packages/swr/src/procedure-utils.ts | 10 +++---- packages/tanstack-query/src/live-query.ts | 12 ++++---- .../tanstack-query/src/stream-query.test.ts | 16 ++++++++++ packages/tanstack-query/src/stream-query.ts | 18 +++++------ 19 files changed, 150 insertions(+), 81 deletions(-) diff --git a/packages/client/src/plugins/request-compression.ts b/packages/client/src/plugins/request-compression.ts index e28179eec..10f97001f 100644 --- a/packages/client/src/plugins/request-compression.ts +++ b/packages/client/src/plugins/request-compression.ts @@ -1,6 +1,6 @@ import type { CompressionEncoding } from '@orpc/shared' import { - AVG_UTF8_BYTES_PER_CHAR, + estimateUtf8ByteLength, estimateCompressibleFormDataSize, isAsyncIteratorObject, isCompressibleContentType, @@ -179,7 +179,7 @@ export class RequestCompressionLinkPlugin< } } else if (request.body instanceof URLSearchParams) { const string = request.body.toString() - if (string.length * AVG_UTF8_BYTES_PER_CHAR >= this.threshold) { + if (estimateUtf8ByteLength(string) >= this.threshold) { const compressedStream = new Blob([string]) .stream() .pipeThrough(new CompressionStream(this.encoding)) @@ -205,7 +205,7 @@ export class RequestCompressionLinkPlugin< string, StandardHeaders, ] - if (string.length * AVG_UTF8_BYTES_PER_CHAR >= this.threshold) { + if (estimateUtf8ByteLength(string) >= this.threshold) { const compressedStream = new Blob([string]) .stream() .pipeThrough(new CompressionStream(this.encoding)) diff --git a/packages/client/src/plugins/response-compression.test.ts b/packages/client/src/plugins/response-compression.test.ts index 533396577..da0a65846 100644 --- a/packages/client/src/plugins/response-compression.test.ts +++ b/packages/client/src/plugins/response-compression.test.ts @@ -115,6 +115,22 @@ describe('responseCompressionLinkPlugin', () => { await expect(link.call(['test'], undefined, { context: {} })).resolves.toEqual('OK') }) + it('ignores empty content-encoding list elements', async () => { + const compressed = await compressAsync(JSON.stringify({ json: 'OK' }), 'gzip') + + const { link } = createLink({ + fetchImpl: async () => + new Response(compressed, { + headers: { + 'content-type': 'application/json', + 'content-encoding': 'gzip, ', + }, + }), + }) + + await expect(link.call(['test'], undefined, { context: {} })).resolves.toEqual('OK') + }) + it('decompresses response body when 5 content-encodings are applied', async () => { let body: Uint8Array = await compressAsync(JSON.stringify({ json: 'OK' }), 'gzip') for (let i = 1; i < 5; i++) { diff --git a/packages/client/src/plugins/response-compression.ts b/packages/client/src/plugins/response-compression.ts index 2c0d30860..fbd19fbed 100644 --- a/packages/client/src/plugins/response-compression.ts +++ b/packages/client/src/plugins/response-compression.ts @@ -1,3 +1,4 @@ +import type { CompressionEncoding } from '@orpc/shared' import { isCompressionEncoding, parseContentEncodings, toArray } from '@orpc/shared' import { flattenStandardHeader } from '@standard-server/core' import { toFetchHeaders, toStandardBody } from '@standard-server/fetch' @@ -22,7 +23,7 @@ export interface ResponseCompressionLinkPluginOptions<_T extends ClientContext> * * @default ['gzip', 'deflate'] */ - encodings?: readonly ('gzip' | 'deflate' | 'deflate-raw')[] | undefined + encodings?: readonly CompressionEncoding[] | undefined } /** diff --git a/packages/node/src/batch-response-compression-handler-plugin.ts b/packages/node/src/batch-response-compression-handler-plugin.ts index 29b50ca45..314503058 100644 --- a/packages/node/src/batch-response-compression-handler-plugin.ts +++ b/packages/node/src/batch-response-compression-handler-plugin.ts @@ -12,6 +12,7 @@ import type { StandardHandlerPlugin, StandardHandlerRoutingInterceptor, } from '@orpc/server/standard' +import type { CompressionEncoding } from '@orpc/shared' import { toArray } from '@orpc/shared' import type { StandardBodyHint, StandardHeaders } from '@standard-server/core' import { flattenStandardHeader } from '@standard-server/core' @@ -25,7 +26,7 @@ export interface BatchResponseCompressionHandlerPluginOptions { * * @default ['gzip', 'deflate'] */ - encodings?: readonly ('gzip' | 'deflate' | 'deflate-raw')[] | undefined + encodings?: readonly CompressionEncoding[] | undefined /** * The minimum response size in bytes required to trigger compression. @@ -188,9 +189,7 @@ interface CompressionStreamPair { * as soon as the batch produces it instead of waiting in the compressor for the batch to end. * The few bytes each flush costs are what keep a streaming batch streaming. */ -function createFlushingCompressionStream( - encoding: Exclude[number], -): CompressionStreamPair { +function createFlushingCompressionStream(encoding: CompressionEncoding): CompressionStreamPair { const zlibOptions = { flush: constants.Z_SYNC_FLUSH } const compressor = diff --git a/packages/pinia-colada/src/live-query.ts b/packages/pinia-colada/src/live-query.ts index cdc53a9e7..6729d1d7b 100644 --- a/packages/pinia-colada/src/live-query.ts +++ b/packages/pinia-colada/src/live-query.ts @@ -1,5 +1,5 @@ import type { Promisable } from '@orpc/shared' -import { stringifyJSON, throwIfAborted } from '@orpc/shared' +import { listenAbort, stringifyJSON, throwIfAborted } from '@orpc/shared' import type { SettleOnFirstDataOptions } from './stream-query.ts' import type { UseQueryFnContext } from './types.ts' @@ -16,13 +16,9 @@ export function liveQuery( const { entry, signal } = context // a fetch cut short (cancelled, superseded, or its last component left) reopens when used again - const markStale = () => { + listenAbort(signal, () => { entry.when = 0 - } - if (signal.aborted) { - markStale() - } - signal.addEventListener('abort', markStale, { once: true }) + }) const stream = await queryFn(context) diff --git a/packages/pinia-colada/src/stream-query.test.tsx b/packages/pinia-colada/src/stream-query.test.tsx index fab261701..eb2b53aa1 100644 --- a/packages/pinia-colada/src/stream-query.test.tsx +++ b/packages/pinia-colada/src/stream-query.test.tsx @@ -490,6 +490,7 @@ describe('serializableStreamedQuery', () => { it('starts a new array when the entry data is cleared mid-stream', async () => { const queryCache = createQueryCache() + const release = promiseWithResolvers() const entry = queryCache.ensure({ key: ['stream'], query: serializableStreamedQuery(async () => @@ -497,11 +498,18 @@ describe('serializableStreamedQuery', () => { yield 1 entry.state.value = { status: 'pending', data: undefined, error: null } as any yield 2 + await release.promise })(), ), }) - await expect(queryCache.fetch(entry)).resolves.toMatchObject({ data: [2] }) + const fetching = queryCache.fetch(entry) + + // checked while the stream is still open, so the final write cannot be what makes it pass + await vi.waitFor(() => expect(entry.state.value.data).toEqual([2]), { interval: 1 }) + + release.resolve() + await expect(fetching).resolves.toMatchObject({ data: [2] }) }) it('resolves with the streamed data when the entry data is cleared after the last chunk', async () => { diff --git a/packages/pinia-colada/src/stream-query.ts b/packages/pinia-colada/src/stream-query.ts index b65108136..e96c504a6 100644 --- a/packages/pinia-colada/src/stream-query.ts +++ b/packages/pinia-colada/src/stream-query.ts @@ -1,5 +1,5 @@ import type { Promisable } from '@orpc/shared' -import { takeLast, throwIfAborted } from '@orpc/shared' +import { listenAbort, takeLast, throwIfAborted } from '@orpc/shared' import type { UseQueryFnContext } from './types.ts' @@ -61,13 +61,9 @@ export function serializableStreamedQuery( } // a fetch cut short (cancelled, superseded, or its last component left) reopens when used again - const markStale = () => { + listenAbort(signal, () => { entry.when = 0 - } - if (signal.aborted) { - markStale() - } - signal.addEventListener('abort', markStale, { once: true }) + }) const stream = await queryFn(context) diff --git a/packages/server/src/plugins/batch.ts b/packages/server/src/plugins/batch.ts index 41f69a64d..7513ccf73 100644 --- a/packages/server/src/plugins/batch.ts +++ b/packages/server/src/plugins/batch.ts @@ -1,7 +1,13 @@ import { ORPCError } from '@orpc/client' import { BATCH_HEADER, DEFAULT_BATCH_MAX_SIZE } from '@orpc/client/plugins' import type { Promisable, Value } from '@orpc/shared' -import { isAsyncIteratorObject, promiseWithResolvers, toArray, value } from '@orpc/shared' +import { + isAsyncIteratorObject, + listenAbort, + promiseWithResolvers, + toArray, + value, +} from '@orpc/shared' import type { StandardHeaders, StandardLazyRequest, StandardResponse } from '@standard-server/core' import { flattenStandardHeader, parseStandardUrl } from '@standard-server/core' import type { ClientPeerSendMessage, ServerPeerSendMessage } from '@standard-server/peer' @@ -521,19 +527,15 @@ export class BatchHandlerPlugin implements StandardHandlerPlu .finally(() => resolve(undefined)) }), ) - const signal = interceptorOptions.request.signal - const closePeer = () => peer.close(signal?.reason) - - if (signal?.aborted) { - void closePeer() - } - - signal?.addEventListener('abort', closePeer) + const unlisten = listenAbort( + interceptorOptions.request.signal, + (reason) => void peer.close(reason), + ) try { await promise } finally { - signal?.removeEventListener('abort', closePeer) + unlisten() } } diff --git a/packages/server/src/plugins/response-compression.ts b/packages/server/src/plugins/response-compression.ts index 873d5e3f5..03a0b2f5d 100644 --- a/packages/server/src/plugins/response-compression.ts +++ b/packages/server/src/plugins/response-compression.ts @@ -1,6 +1,6 @@ import type { CompressionEncoding } from '@orpc/shared' import { - AVG_UTF8_BYTES_PER_CHAR, + estimateUtf8ByteLength, estimateCompressibleFormDataSize, isAcceptableEncoding, isAsyncIteratorObject, @@ -187,7 +187,7 @@ export class ResponseCompressionHandlerPlugin< } } else if (body instanceof URLSearchParams) { const string = body.toString() - if (string.length * AVG_UTF8_BYTES_PER_CHAR >= this.threshold) { + if (estimateUtf8ByteLength(string) >= this.threshold) { const [, bodyHeaders] = toFetchBody(body, headers) return { @@ -205,7 +205,7 @@ export class ResponseCompressionHandlerPlugin< } } else if (body !== undefined && !isAsyncIteratorObject(body)) { const [string, bodyHeaders] = toFetchBody(body, headers) as [string, StandardHeaders] - if (string.length * AVG_UTF8_BYTES_PER_CHAR >= this.threshold) { + if (estimateUtf8ByteLength(string) >= this.threshold) { return { ...result, response: { diff --git a/packages/server/src/plugins/signal-sync.test.ts b/packages/server/src/plugins/signal-sync.test.ts index 71591ef69..4a90661df 100644 --- a/packages/server/src/plugins/signal-sync.test.ts +++ b/packages/server/src/plugins/signal-sync.test.ts @@ -320,7 +320,7 @@ describe('signalSyncHandlerPlugin', () => { await new Promise((resolve) => setTimeout(resolve, 0)) // any unhandled rejection would fail the test }) - it('drops a pending read that settles after the signal aborts', async () => { + it('rejects a pending read with the abort reason when the signal aborts', async () => { const pulled = promiseWithResolvers() const controller = new AbortController() diff --git a/packages/server/src/plugins/signal-sync.ts b/packages/server/src/plugins/signal-sync.ts index 905b3ad4a..b77aca877 100644 --- a/packages/server/src/plugins/signal-sync.ts +++ b/packages/server/src/plugins/signal-sync.ts @@ -90,9 +90,9 @@ function syncReadableStream( return new ReadableStream( { start(controller) { - unlisten = listenAbort(signal, () => { - controller.error(signal.reason) - cancelling = reader.cancel(signal.reason) + unlisten = listenAbort(signal, (reason) => { + controller.error(reason) + cancelling = reader.cancel(reason) cancelling.catch(() => {}) // only awaited when the consumer cancels }) }, diff --git a/packages/shared/src/http.test.ts b/packages/shared/src/http.test.ts index 0cd1f3168..0fe3fd28a 100644 --- a/packages/shared/src/http.test.ts +++ b/packages/shared/src/http.test.ts @@ -1,6 +1,6 @@ import { - AVG_UTF8_BYTES_PER_CHAR, estimateCompressibleFormDataSize, + estimateUtf8ByteLength, isAcceptableEncoding, isCompressibleContentType, isCompressionEncoding, @@ -280,6 +280,20 @@ describe('parseContentEncodings', () => { it('keeps the order codings were applied in, trimmed and lowercased', () => { expect(parseContentEncodings(' GZIP , deflate')).toEqual(['gzip', 'deflate']) }) + + it('ignores empty list elements', () => { + expect(parseContentEncodings('gzip,')).toEqual(['gzip']) + expect(parseContentEncodings(', gzip, , deflate')).toEqual(['gzip', 'deflate']) + expect(parseContentEncodings('')).toEqual([]) + }) +}) + +describe('estimateUtf8ByteLength', () => { + it('scales with the string length', () => { + expect(estimateUtf8ByteLength('')).toBe(0) + expect(estimateUtf8ByteLength('a'.repeat(10))).toBeGreaterThanOrEqual(10) + expect(estimateUtf8ByteLength('a'.repeat(20))).toBe(2 * estimateUtf8ByteLength('a'.repeat(10))) + }) }) describe('isAcceptableEncoding', () => { @@ -341,7 +355,7 @@ describe('estimateCompressibleFormDataSize', () => { form.append('b', new Blob(['y'.repeat(100)], { type: 'text/plain' })) expect(estimateCompressibleFormDataSize(form, isCompressible)).toBe( - 64 + 1 + 10 * AVG_UTF8_BYTES_PER_CHAR + 64 + 1 + 100, + 64 + 1 + estimateUtf8ByteLength('x'.repeat(10)) + 64 + 1 + 100, ) }) @@ -353,9 +367,10 @@ describe('estimateCompressibleFormDataSize', () => { }) it('handles blobs with an unknown size', () => { - const compressible = new Blob(['a'], { type: 'text/plain' }) + // a File appended without a filename is stored as is, so the patched size is what gets read + const compressible = new File(['a'], 'a.txt', { type: 'text/plain' }) Object.defineProperty(compressible, 'size', { value: Number.NaN }) - const nonCompressible = new Blob(['a'], { type: 'image/png' }) + const nonCompressible = new File(['a'], 'a.png', { type: 'image/png' }) Object.defineProperty(nonCompressible, 'size', { value: Number.NaN }) const form = new FormData() diff --git a/packages/shared/src/http.ts b/packages/shared/src/http.ts index 47fbfd98e..9be2c7ab4 100644 --- a/packages/shared/src/http.ts +++ b/packages/shared/src/http.ts @@ -84,7 +84,7 @@ export function matchesHttpPath(url: `/${string}`, path: `/${string}`): boolean /** * Content codings that oRPC compression plugins can encode and decode. */ -export const COMPRESSION_ENCODINGS = ['gzip', 'deflate', 'deflate-raw'] as const +const COMPRESSION_ENCODINGS = ['gzip', 'deflate', 'deflate-raw'] as const export type CompressionEncoding = (typeof COMPRESSION_ENCODINGS)[number] @@ -93,23 +93,35 @@ export function isCompressionEncoding(encoding: string): encoding is Compression } /** - * Parse Content-Encoding into ordered codings (order applied). + * Parse Content-Encoding into ordered codings (order applied), ignoring empty list elements. * * @see https://www.rfc-editor.org/rfc/rfc9110.html#name-content-encoding + * @see https://www.rfc-editor.org/rfc/rfc9110.html#name-lists-rule-abnf-extension */ export function parseContentEncodings(header: string | undefined): string[] { - if (header === undefined) { - return [] + const codings: string[] = [] + + for (const part of header?.split(',') ?? []) { + const coding = part.trim().toLowerCase() + + if (coding !== '') { + codings.push(coding) + } } - return header.split(',').map((part) => part.trim().toLowerCase()) + return codings } +// Rough UTF-8 estimate. Mostly ASCII text stays close to 1 byte/char; +// occasional multi-byte characters increase the average. +const AVG_UTF8_BYTES_PER_CHAR = 1.2 + /** - * Rough UTF-8 bytes per string character. Mostly ASCII text stays close to 1 byte/char; - * occasional multi-byte characters increase the average. + * Roughly estimates the UTF-8 byte length of a string without encoding it. */ -export const AVG_UTF8_BYTES_PER_CHAR = 1.2 +export function estimateUtf8ByteLength(value: string): number { + return value.length * AVG_UTF8_BYTES_PER_CHAR +} // approx bytes for boundary + Content-Disposition/Content-Type headers per part const FORM_DATA_PART_OVERHEAD = 64 @@ -141,7 +153,7 @@ export function estimateCompressibleFormDataSize( size += isCompressible(value.type) ? value.size : -value.size } } else { - size += value.length * AVG_UTF8_BYTES_PER_CHAR + size += estimateUtf8ByteLength(value) } } diff --git a/packages/shared/src/signal.ts b/packages/shared/src/signal.ts index 5862cec14..a26e5c256 100644 --- a/packages/shared/src/signal.ts +++ b/packages/shared/src/signal.ts @@ -121,16 +121,18 @@ export async function runWithSignal( } } +const noop = () => {} + /** - * Calls `onAbort` with the abort reason right away if the signal is already aborted, otherwise once - * when it aborts. Does nothing without a signal. Returns a function that removes the listener. + * Calls `onAbort` with the abort reason once the signal aborts, or synchronously, before returning, + * if it is already aborted. Does nothing without a signal. Returns a function that removes the listener. */ export function listenAbort( signal: AbortSignal | undefined, onAbort: (reason: unknown) => void, ): () => void { if (!signal) { - return () => {} + return noop } const listener = () => onAbort(signal.reason) diff --git a/packages/swr/src/procedure-utils.test.ts b/packages/swr/src/procedure-utils.test.ts index 8f604b37e..8bdcacfa7 100644 --- a/packages/swr/src/procedure-utils.test.ts +++ b/packages/swr/src/procedure-utils.test.ts @@ -241,7 +241,11 @@ describe('procedureUtils', () => { }) it('on error after unsubscribe', async () => { + const started = promiseWithResolvers() client.mockImplementationOnce(async function* () { + yield 'started' + // resumed for the next value, so the first one has been handled + started.resolve() await new Promise((resolve) => setTimeout(resolve, 10)) throw new Error('__error__') }) @@ -249,9 +253,11 @@ describe('procedureUtils', () => { const next = vi.fn() const unsubscribe = subscriber(key, { next }) + // unsubscribe mid-stream, so the error reaches the subscriber after it was aborted + await started.promise unsubscribe() await new Promise((resolve) => setTimeout(resolve, 20)) - expect(next).toHaveBeenCalledTimes(0) + expect(next).not.toHaveBeenCalledWith(new Error('__error__')) }) it('on non-AsyncIteratorObject output', async () => { @@ -388,7 +394,11 @@ describe('procedureUtils', () => { }) it('on error after unsubscribe', async () => { + const started = promiseWithResolvers() client.mockImplementationOnce(async function* () { + yield 'started' + // resumed for the next value, so the first one has been handled + started.resolve() await new Promise((resolve) => setTimeout(resolve, 10)) throw new Error('__error__') }) @@ -396,9 +406,11 @@ describe('procedureUtils', () => { const next = vi.fn() const unsubscribe = subscriber(key, { next }) + // unsubscribe mid-stream, so the error reaches the subscriber after it was aborted + await started.promise unsubscribe() await new Promise((resolve) => setTimeout(resolve, 20)) - expect(next).toHaveBeenCalledTimes(0) + expect(next).not.toHaveBeenCalledWith(new Error('__error__')) }) it('on non-AsyncIteratorObject output', async () => { diff --git a/packages/swr/src/procedure-utils.ts b/packages/swr/src/procedure-utils.ts index a39c5918a..5a1571991 100644 --- a/packages/swr/src/procedure-utils.ts +++ b/packages/swr/src/procedure-utils.ts @@ -1,6 +1,6 @@ import type { Client, ClientContext } from '@orpc/client' import type { MaybeOptionalOptions } from '@orpc/shared' -import { isAsyncIteratorObject, resolveMaybeOptionalOptions } from '@orpc/shared' +import { isAsyncIteratorObject, resolveMaybeOptionalOptions, takeLast } from '@orpc/shared' import type { OperationKey, OperationKeyOptions, OperationKeyPrefixOptions } from './key.ts' import { generateOperationKey } from './key.ts' @@ -131,11 +131,9 @@ export class ProcedureUtils< next(undefined, (old) => { const newData = Array.isArray(old) ? [...old, event] : [event] - if (typeof maxChunks === 'number' && newData.length > maxChunks) { - return newData.slice(newData.length - maxChunks) as InferSubscriberOutput - } - - return newData as InferSubscriberOutput + return ( + typeof maxChunks === 'number' ? takeLast(newData, maxChunks) : newData + ) as InferSubscriberOutput }) } } catch (error) { diff --git a/packages/tanstack-query/src/live-query.ts b/packages/tanstack-query/src/live-query.ts index 3100ef686..f38883199 100644 --- a/packages/tanstack-query/src/live-query.ts +++ b/packages/tanstack-query/src/live-query.ts @@ -1,5 +1,5 @@ import type { Promisable } from '@orpc/shared' -import { stringifyJSON, throwIfAborted } from '@orpc/shared' +import { listenAbort, stringifyJSON, throwIfAborted } from '@orpc/shared' import type { QueryFunction, QueryFunctionContext, QueryKey } from '@tanstack/query-core' import type { SettleOnFirstDataOptions } from './stream-query.ts' @@ -13,11 +13,9 @@ export function liveQuery client.getQueryCache().find({ queryKey, exact: true })?.invalidate() - if (signal.aborted) { - markStale() - } - signal.addEventListener('abort', markStale, { once: true }) + const unlisten = listenAbort(signal, () => + client.getQueryCache().find({ queryKey, exact: true })?.invalidate(), + ) try { const stream = await queryFn(context) @@ -48,7 +46,7 @@ export function liveQuery { await promise1 }) + it('starts a new array when the cache is set to null mid-stream', async () => { + const key = ['null-mid-stream'] + + await expect( + queryClient.fetchQuery({ + queryKey: key, + queryFn: serializableStreamedQuery(async function* () { + yield 1 + queryClient.setQueryData(key, null) + yield 2 + }), + }), + ).resolves.toEqual([2]) + expect(queryClient.getQueryData(key)).toEqual([2]) + }) + it('keeps a null written after the last chunk', async () => { const key = ['null-after-stream'] diff --git a/packages/tanstack-query/src/stream-query.ts b/packages/tanstack-query/src/stream-query.ts index 4713bc1eb..e0a8057d3 100644 --- a/packages/tanstack-query/src/stream-query.ts +++ b/packages/tanstack-query/src/stream-query.ts @@ -1,5 +1,5 @@ import type { Promisable } from '@orpc/shared' -import { takeLast, throwIfAborted } from '@orpc/shared' +import { listenAbort, takeLast, throwIfAborted } from '@orpc/shared' import type { QueryFunction, QueryFunctionContext, QueryKey } from '@tanstack/query-core' export interface SerializableStreamedQueryOptions { @@ -56,11 +56,9 @@ export function serializableStreamedQuery< // a stream cut short (cancelled, or its last observer left) reopens when observed again; // marked at abort, so the writes of a fetch that replaces this one clear it - const markStale = () => client.getQueryCache().find({ queryKey, exact: true })?.invalidate() - if (signal.aborted) { - markStale() - } - signal.addEventListener('abort', markStale, { once: true }) + const unlisten = listenAbort(signal, () => + client.getQueryCache().find({ queryKey, exact: true })?.invalidate(), + ) try { const stream = await queryFn(context) @@ -83,9 +81,9 @@ export function serializableStreamedQuery< for await (const chunk of stream) { throwIfAborted(signal) - // read the cache, so writes made during the stream are kept - data = client.setQueryData(queryKey, (current = []) => - takeLast([...current, chunk], maxChunks), + // read the cache, so writes made during the stream are kept; a cleared entry (`null`) starts over + data = client.setQueryData(queryKey, (current) => + takeLast([...(current ?? []), chunk], maxChunks), )! } @@ -98,7 +96,7 @@ export function serializableStreamedQuery< // only an array is trimmed; another value written after the stream (e.g. `null`) is kept as is return Array.isArray(cached) ? takeLast(cached, maxChunks) : cached } finally { - signal.removeEventListener('abort', markStale) + unlisten() } } } From dcc436359c7599f6c357755e49479c4b01089c4b Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 8 Nov 2026 02:16:22 +0000 Subject: [PATCH 5/5] test(swr, tanstack-query): restore the early-unsubscribe test and tighten stream tests - swr: restore a subscriber test that unsubscribes before the call resolves and checks that next is never called, not even for the reset - swr: the error-after-unsubscribe tests now check that no callback at all arrives after unsubscribing - tanstack-query: assert the cache holds the new array while the stream is still open Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VVaw7PANP3R1NnJpBLkx5b --- packages/swr/src/procedure-utils.test.ts | 30 +++++++++++++++++-- .../tanstack-query/src/stream-query.test.ts | 24 +++++++++------ 2 files changed, 43 insertions(+), 11 deletions(-) diff --git a/packages/swr/src/procedure-utils.test.ts b/packages/swr/src/procedure-utils.test.ts index 8bdcacfa7..8517adf66 100644 --- a/packages/swr/src/procedure-utils.test.ts +++ b/packages/swr/src/procedure-utils.test.ts @@ -240,6 +240,30 @@ describe('procedureUtils', () => { expect(next).toHaveBeenCalledWith(new Error('__error__')) }) + it('does not start the stream when unsubscribed before the call resolves', async () => { + const started = vi.fn() + const connected = promiseWithResolvers() + client.mockImplementationOnce(async () => { + await connected.promise + return (async function* () { + started() + yield '__event__1' + })() + }) + const subscriber = utils.subscriber({ refetchMode: 'reset' }) + + const next = vi.fn() + const unsubscribe = subscriber(key, { next }) + + unsubscribe() + connected.resolve() + await new Promise((resolve) => setTimeout(resolve, 10)) + + // no reset either, so a late resolution cannot clear the cache + expect(next).not.toHaveBeenCalled() + expect(started).not.toHaveBeenCalled() + }) + it('on error after unsubscribe', async () => { const started = promiseWithResolvers() client.mockImplementationOnce(async function* () { @@ -255,9 +279,10 @@ describe('procedureUtils', () => { const unsubscribe = subscriber(key, { next }) // unsubscribe mid-stream, so the error reaches the subscriber after it was aborted await started.promise + const callsBeforeUnsubscribe = next.mock.calls.length unsubscribe() await new Promise((resolve) => setTimeout(resolve, 20)) - expect(next).not.toHaveBeenCalledWith(new Error('__error__')) + expect(next).toHaveBeenCalledTimes(callsBeforeUnsubscribe) }) it('on non-AsyncIteratorObject output', async () => { @@ -408,9 +433,10 @@ describe('procedureUtils', () => { const unsubscribe = subscriber(key, { next }) // unsubscribe mid-stream, so the error reaches the subscriber after it was aborted await started.promise + const callsBeforeUnsubscribe = next.mock.calls.length unsubscribe() await new Promise((resolve) => setTimeout(resolve, 20)) - expect(next).not.toHaveBeenCalledWith(new Error('__error__')) + expect(next).toHaveBeenCalledTimes(callsBeforeUnsubscribe) }) it('on non-AsyncIteratorObject output', async () => { diff --git a/packages/tanstack-query/src/stream-query.test.ts b/packages/tanstack-query/src/stream-query.test.ts index 7f5581e95..cac41da28 100644 --- a/packages/tanstack-query/src/stream-query.test.ts +++ b/packages/tanstack-query/src/stream-query.test.ts @@ -364,17 +364,23 @@ describe('serializableStreamedQuery', () => { it('starts a new array when the cache is set to null mid-stream', async () => { const key = ['null-mid-stream'] + const release = promiseWithResolvers() - await expect( - queryClient.fetchQuery({ - queryKey: key, - queryFn: serializableStreamedQuery(async function* () { - yield 1 - queryClient.setQueryData(key, null) - yield 2 - }), + const fetching = queryClient.fetchQuery({ + queryKey: key, + queryFn: serializableStreamedQuery(async function* () { + yield 1 + queryClient.setQueryData(key, null) + yield 2 + await release.promise }), - ).resolves.toEqual([2]) + }) + + // checked while the stream is still open, so the final write cannot be what makes it pass + await vi.waitFor(() => expect(queryClient.getQueryData(key)).toEqual([2]), { interval: 1 }) + + release.resolve() + await expect(fetching).resolves.toEqual([2]) expect(queryClient.getQueryData(key)).toEqual([2]) })