diff --git a/apps/content/docs/plugins/body-compression.md b/apps/content/docs/plugins/body-compression.md index 487fd5fd0..9ae768768 100644 --- a/apps/content/docs/plugins/body-compression.md +++ b/apps/content/docs/plugins/body-compression.md @@ -25,6 +25,12 @@ const handler = new RPCHandler(router, { +## Batch and Event Stream Responses + +Event stream responses are never compressed: the web `CompressionStream` API cannot flush between chunks, so early events would sit in the compressor buffer until the stream ends, defeating streaming semantics. + +[Batch requests](/docs/plugins/batch-requests) (identified by the `orpc-batch` header) with an `application/octet-stream` body are compressed by the fetch adapter when the runtime exposes the Node.js `zlib` and `stream` builtins via `process.getBuiltinModule` (Node.js ≥ 22.3, Bun, and compatible runtimes): a flush-per-message `zlib` stream is used instead of `CompressionStream`, so each response in a streaming batch is delivered to the client as soon as it is ready — compressed. In runtimes without these builtins, such batch responses stay uncompressed. + ## Learn More For implementation details, see the [fetch adapter source code](https://github.com/middleapi/orpc/blob/main/packages/server/src/adapters/fetch/body-compression-plugin.ts) and the [node adapter source code](https://github.com/middleapi/orpc/blob/main/packages/server/src/adapters/node/body-compression-plugin.ts). diff --git a/packages/server/src/adapters/fetch/body-compression-plugin.test.ts b/packages/server/src/adapters/fetch/body-compression-plugin.test.ts index dfbf74c5b..2b2d7761a 100644 --- a/packages/server/src/adapters/fetch/body-compression-plugin.test.ts +++ b/packages/server/src/adapters/fetch/body-compression-plugin.test.ts @@ -1,4 +1,6 @@ import type { FetchHandlerFetchInterceptor } from './handler' +import { os } from '../../builder' +import { BatchHandlerPlugin } from '../../plugins/batch' import { BodyCompressionHandlerPlugin } from './body-compression-plugin' import { RPCHandler } from './rpc-handler' @@ -176,4 +178,240 @@ describe('bodyCompressionHandlerPlugin', () => { await expect(response!.text()).resolves.toBe(largeText) expect(response!.headers.has('content-encoding')).toBe(false) }) + + it('does not compress octet-stream responses for non-batch requests', async () => { + const body = new TextEncoder().encode(largeText) + + const handler = createHandler(new Response(body, { + headers: { + 'content-length': body.byteLength.toString(), + 'content-type': 'application/octet-stream', + }, + })) + + const { matched, response } = await handler.handle(new Request('https://example.com/ping', { + headers: { + 'accept-encoding': 'gzip', + }, + })) + + expect(matched).toBe(true) + await expect(response!.text()).resolves.toBe(largeText) + expect(response!.headers.has('content-encoding')).toBe(false) + }) + + describe('batch responses', () => { + const largeValue = 'a'.repeat(500) + + function makePeerRequestMessage(id: number, url: string) { + return { + kind: 'request', + id, + json: { method: 'POST', url, headers: {}, body: undefined }, + binary: undefined, + } + } + + function createBatchRequest(mode: 'buffered' | 'streaming', messages: unknown) { + return new Request('https://example.com/__batch__', { + method: 'POST', + headers: { + 'accept-encoding': 'gzip', + 'content-type': 'application/json', + 'orpc-batch': mode, + }, + body: JSON.stringify(messages), + }) + } + + function createBatchHandler( + router: Parameters[0], + options: ConstructorParameters[0] = {}, + ): RPCHandler { + return new RPCHandler(router, { + plugins: [new BatchHandlerPlugin(), new BodyCompressionHandlerPlugin(options)], + }) + } + + it('compresses streaming batch responses', async () => { + const handler = createBatchHandler({ + ping: os.handler(() => largeValue), + }) + + const { matched, response } = await handler.handle(createBatchRequest('streaming', [ + makePeerRequestMessage(0, '/ping'), + makePeerRequestMessage(1, '/ping'), + ])) + + expect(matched).toBe(true) + expect(response!.status).toBe(207) + expect(response!.headers.get('content-encoding')).toBe('gzip') + + const decompressed = response!.body!.pipeThrough(new DecompressionStream('gzip')) + const text = await new Response(decompressed).text() + expect(text).toContain(largeValue) + }) + + it('flushes each compressed batch message without waiting for the whole batch', async () => { + let resolveSlow!: (value: string) => void + const slow = new Promise((resolve) => { + resolveSlow = resolve + }) + + const handler = createBatchHandler({ + fast: os.handler(() => largeValue), + slow: os.handler(() => slow), + }) + + const { response } = await handler.handle(createBatchRequest('streaming', [ + makePeerRequestMessage(0, '/fast'), + makePeerRequestMessage(1, '/slow'), + ])) + + expect(response!.headers.get('content-encoding')).toBe('gzip') + + const reader = response!.body!.pipeThrough(new DecompressionStream('gzip')).getReader() + const decoder = new TextDecoder() + + /** + * A flush-capable compressor delivers the fast message while the slow one + * is still pending. A buffering compressor (e.g. `CompressionStream`) + * would hold it back until the stream ends and these reads would never + * resolve (test times out) because the slow handler is not resolved yet. + */ + let received = '' + while (!received.includes(largeValue)) { + const { done, value } = await reader.read() + if (done) { + expect.unreachable('stream ended before the fast message was delivered') + } + received += decoder.decode(value, { stream: true }) + } + + resolveSlow('slow-result') + + while (!received.includes('slow-result')) { + const { done, value } = await reader.read() + if (done) { + break + } + received += decoder.decode(value, { stream: true }) + } + + expect(received).toContain('slow-result') + }) + + it('compresses streaming batch responses with deflate when preferred', async () => { + const handler = createBatchHandler({ + ping: os.handler(() => largeValue), + }) + + const { response } = await handler.handle(new Request('https://example.com/__batch__', { + method: 'POST', + headers: { + 'accept-encoding': 'deflate', + 'content-type': 'application/json', + 'orpc-batch': 'streaming', + }, + body: JSON.stringify([makePeerRequestMessage(0, '/ping')]), + })) + + expect(response!.headers.get('content-encoding')).toBe('deflate') + + const decompressed = response!.body!.pipeThrough(new DecompressionStream('deflate')) + const text = await new Response(decompressed).text() + expect(text).toContain(largeValue) + }) + + it('compresses buffered batch responses as json (unchanged behavior)', async () => { + const handler = createBatchHandler({ + ping: os.handler(() => largeValue), + }) + + const { response } = await handler.handle(createBatchRequest('buffered', [ + makePeerRequestMessage(0, '/ping'), + makePeerRequestMessage(1, '/ping'), + ])) + + expect(response!.status).toBe(207) + expect(response!.headers.get('content-type')).toContain('application/json') + expect(response!.headers.get('content-encoding')).toBe('gzip') + + const decompressed = response!.body!.pipeThrough(new DecompressionStream('gzip')) + const text = await new Response(decompressed).text() + expect(text).toContain(largeValue) + }) + + it('respects a custom filter that rejects batch responses', async () => { + const filter = vi.fn(() => false) + + const handler = createBatchHandler({ + ping: os.handler(() => largeValue), + }, { filter }) + + const { response } = await handler.handle(createBatchRequest('streaming', [ + makePeerRequestMessage(0, '/ping'), + ])) + + expect(filter).toHaveBeenCalled() + expect(response!.headers.has('content-encoding')).toBe(false) + + const text = await new Response(response!.body).text() + expect(text).toContain(largeValue) + }) + + it('propagates cancellation through the compressed stream to the source', async () => { + let sourceCancelled = false + const source = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(largeValue)) + }, + cancel() { + sourceCancelled = true + }, + }) + + const handler = createHandler(new Response(source, { + headers: { + 'content-type': 'application/octet-stream', + }, + })) + + const { response } = await handler.handle(new Request('https://example.com/__batch__', { + headers: { + 'accept-encoding': 'gzip', + 'orpc-batch': 'streaming', + }, + })) + + expect(response!.headers.get('content-encoding')).toBe('gzip') + + const reader = response!.body!.getReader() + const { value } = await reader.read() + expect(value!.byteLength).toBeGreaterThan(0) + + await reader.cancel() + await vi.waitFor(() => expect(sourceCancelled).toBe(true)) + }) + + it('supports cancelling a compressed streaming batch response mid-batch', async () => { + const handler = createBatchHandler({ + fast: os.handler(() => largeValue), + never: os.handler(() => new Promise(() => {})), + }) + + const { response } = await handler.handle(createBatchRequest('streaming', [ + makePeerRequestMessage(0, '/fast'), + makePeerRequestMessage(1, '/never'), + ])) + + expect(response!.headers.get('content-encoding')).toBe('gzip') + + const reader = response!.body!.getReader() + const { value } = await reader.read() + expect(value!.byteLength).toBeGreaterThan(0) + + await expect(reader.cancel()).resolves.toBeUndefined() + }) + }) }) diff --git a/packages/server/src/adapters/fetch/body-compression-plugin.ts b/packages/server/src/adapters/fetch/body-compression-plugin.ts index 1645ff5e0..d2a7c33bf 100644 --- a/packages/server/src/adapters/fetch/body-compression-plugin.ts +++ b/packages/server/src/adapters/fetch/body-compression-plugin.ts @@ -28,7 +28,10 @@ export interface BodyCompressionHandlerPluginOptions { * Override the default content-type filter used to determine which responses should be compressed. * * @warning Event stream responses are never compressed, regardless of this filter's return value. - * @default only responses with compressible content types are compressed. + * @default only responses with compressible content types are compressed. Batch responses + * (`orpc-batch` request header) with an `application/octet-stream` body are also compressed + * when the runtime exposes Node.js `zlib`/`stream` builtins — a flush-per-message compressor + * keeps streaming batch semantics intact. */ filter?: (request: Request, response: Response) => boolean } @@ -51,6 +54,10 @@ export class BodyCompressionHandlerPlugin implements FetchHan return false } + if (isFlushCompressibleBatchResponse(request, response)) { + return options.filter ? options.filter(request, response) : true + } + return options.filter ? options.filter(request, response) : isCompressibleContentType(contentType) @@ -98,7 +105,22 @@ export class BodyCompressionHandlerPlugin implements FetchHan return result } - const compressedBody = response.body.pipeThrough(new CompressionStream(encoding)) + /** + * `CompressionStream` cannot flush between chunks, so early messages of a + * streaming batch response would sit in the compressor buffer until the + * stream ends, defeating streaming semantics. Batch responses therefore + * use a flush-per-message zlib stream instead (the filter guarantees it is + * available whenever a batch response passes through). + */ + const compression = isFlushCompressibleBatchResponse(interceptorOptions.request, response) + ? createFlushableCompressionStream(encoding) + : new CompressionStream(encoding) + + if (compression === undefined) { + return result + } + + const compressedBody = response.body.pipeThrough(compression) const compressedHeaders = new Headers(response.headers) compressedHeaders.delete('content-length') @@ -136,3 +158,71 @@ function isNoTransformCacheControl(cacheControl: string | null): boolean { return CACHE_CONTROL_NO_TRANSFORM_REGEX.test(cacheControl) } + +/** + * Batch responses carry a `application/octet-stream` body (length-prefixed peer messages), + * which the default content-type filter would never compress even though the payload is + * mostly JSON. They are only compressible when a flush-capable compressor is available, + * otherwise streaming batch responses would lose their streaming semantics. + */ +function isFlushCompressibleBatchResponse(request: Request, response: Response): boolean { + return request.headers.has('orpc-batch') + && !response.headers.has('content-disposition') + && (response.headers.get('content-type')?.startsWith('application/octet-stream') ?? false) + && getNodeCompressionCompat() !== undefined +} + +interface NodeCompressionCompat { + zlib: typeof import('node:zlib') + stream: typeof import('node:stream') +} + +/** + * `false` = probed and unavailable. Runtime builtins cannot appear later, + * so the probe runs at most once per process. + */ +let cachedNodeCompressionCompat: NodeCompressionCompat | false | undefined + +function getNodeCompressionCompat(): NodeCompressionCompat | undefined { + if (cachedNodeCompressionCompat === undefined) { + cachedNodeCompressionCompat = false + + try { + /** + * Deliberately accessed via `globalThis` (not imported) so this cross-runtime + * file never hard-depends on node builtins — runtimes without them fall back + * to the uncompressed behavior. + */ + // eslint-disable-next-line node/prefer-global/process + const zlib = globalThis.process?.getBuiltinModule?.('node:zlib') + // eslint-disable-next-line node/prefer-global/process + const stream = globalThis.process?.getBuiltinModule?.('node:stream') + + if (zlib !== undefined && typeof stream?.Duplex?.toWeb === 'function') { + cachedNodeCompressionCompat = { zlib, stream } + } + } + catch { + // runtimes without node builtins keep the `false` sentinel + } + } + + return cachedNodeCompressionCompat === false ? undefined : cachedNodeCompressionCompat +} + +function createFlushableCompressionStream( + encoding: (typeof ORDERED_SUPPORTED_ENCODINGS)[number], +): ReadableWritablePair | undefined { + const compat = getNodeCompressionCompat() + + if (compat === undefined) { + return undefined + } + + const zlibOptions = { flush: compat.zlib.constants.Z_SYNC_FLUSH } + const compressor = encoding === 'gzip' + ? compat.zlib.createGzip(zlibOptions) + : compat.zlib.createDeflate(zlibOptions) + + return compat.stream.Duplex.toWeb(compressor) as ReadableWritablePair +}