From 245b26a21edfa0ba314bee54e9b95b834e5ce816 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 19 Oct 2026 03:03:23 +0000 Subject: [PATCH 1/3] feat!: carry batched cookies and shared headers on the batch response Browsers only read the headers of the batch response, so cookies set in a batched call were never stored, and they ended up in the batch body where JavaScript could read them, even HttpOnly ones. BatchHandlerPlugin now starts the batch response once every call has responded, and streams it only when a call returns an event iterator or a ReadableStream: - set-cookie never stays on a call. The batch response carries every call's cookies, each distinct value once. - A header every call sets to the same value moves onto the batch response, and the link merges it back into each call. Body headers (content-*) stay on their call, and the headers option wins. - A new mapSubresponse option decides what stays on each call. Its default removes set-cookie and the headers the batch response already has with the same value. BatchLinkPlugin loses its mode option and BatchLinkPluginMode type, and sends `orpc-batch: 1`. Links from before this change still work with the new server. BREAKING CHANGE: BatchLinkPlugin no longer accepts `mode`, and `BatchLinkPluginMode` is removed. A batch no longer sends each response as soon as it is ready. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UtE8iZthQGzPQQoMXCSgzc --- .../content/docs/integrations/better-auth.mdx | 2 +- apps/content/docs/migrations/from-v1.mdx | 2 + .../plugins/batch-response-compression.mdx | 2 +- apps/content/docs/plugins/batch.mdx | 46 +- packages/client/src/plugins/batch.test.ts | 64 ++- packages/client/src/plugins/batch.ts | 15 +- ...esponse-compression-handler-plugin.test.ts | 82 ++-- ...tch-response-compression-handler-plugin.ts | 2 +- packages/server/src/plugins/batch.test.ts | 420 +++++++++++------- packages/server/src/plugins/batch.ts | 209 ++++++--- .../client-server.compression-hono-fetch.ts | 3 +- .../client-server.compression-node-http.ts | 7 +- .../__shared__/client-server.hono-fetch.ts | 3 +- .../__shared__/client-server.node-http.ts | 3 +- tests/batch/__shared__/client-server.ts | 3 - tests/batch/batch-plugin.test.ts | 36 +- tests/plugins/all-plugins.test.ts | 29 +- 17 files changed, 542 insertions(+), 386 deletions(-) diff --git a/apps/content/docs/integrations/better-auth.mdx b/apps/content/docs/integrations/better-auth.mdx index 350ac92c74..c2c11c76ed 100644 --- a/apps/content/docs/integrations/better-auth.mdx +++ b/apps/content/docs/integrations/better-auth.mdx @@ -8,7 +8,7 @@ sidebar: Use your [Better Auth](https://better-auth.com/) instance with oRPC's [context](/docs/context) and [middleware](/docs/middleware). No extra package is needed. :::tip -You may need to forward Better Auth's [response headers](https://better-auth.com/docs/concepts/api#getting-headers), such as a refreshed session cookie. The [Response Headers Plugin](/docs/plugins/response-headers) can help unless you use the [Batch Plugin](/docs/plugins/batch). +You may need to forward Better Auth's [response headers](https://better-auth.com/docs/concepts/api#getting-headers), such as a refreshed session cookie. The [Response Headers Plugin](/docs/plugins/response-headers) can help. ::: ## Resolve the Session in Middleware diff --git a/apps/content/docs/migrations/from-v1.mdx b/apps/content/docs/migrations/from-v1.mdx index dd38563d9d..28389152c3 100644 --- a/apps/content/docs/migrations/from-v1.mdx +++ b/apps/content/docs/migrations/from-v1.mdx @@ -939,6 +939,8 @@ const batchPlugin = new BatchLinkPlugin({ +The `mode` option was removed. The server now starts the batch response once every procedure has responded, so it can carry their [headers](/docs/plugins/batch#response-headers), and [streams](/docs/plugins/batch#streaming-responses) only when a procedure returns a stream. + :::warning Some cases still need to skip batching, for example procedures that rely on [Hibernation](/docs/integrations/hibernation), which cannot work through batched responses. For those, use `filter`. Its meaning is inverted compared to `exclude`: `exclude` returned `true` to skip batching, while `filter` returns `false` to skip batching. Negate your predicate when migrating. diff --git a/apps/content/docs/plugins/batch-response-compression.mdx b/apps/content/docs/plugins/batch-response-compression.mdx index d710b8f483..27f8110587 100644 --- a/apps/content/docs/plugins/batch-response-compression.mdx +++ b/apps/content/docs/plugins/batch-response-compression.mdx @@ -56,7 +56,7 @@ The whole envelope is compressed, binary subresponses included. Leave this plugi ## Why a Node.js Plugin -A streaming batch sends each response as soon as its procedure resolves. A compressor that cannot flush would hold every early response in its buffer until the slowest one finished, trading streaming for compression. The web [CompressionStream](https://developer.mozilla.org/en-US/docs/Web/API/CompressionStream) has no flush, so this plugin uses [zlib](https://nodejs.org/api/zlib.html) and ends each write with a sync flush instead. The few bytes each flush costs are what keep the batch streaming, [keep-alive frames](/docs/plugins/batch#keep-alive-timer) included. +A streaming batch sends each message of its streams as soon as it is ready. A compressor that cannot flush would hold those messages in its buffer until the stream ended, trading streaming for compression. The web [CompressionStream](https://developer.mozilla.org/en-US/docs/Web/API/CompressionStream) has no flush, so this plugin uses [zlib](https://nodejs.org/api/zlib.html) and ends each write with a sync flush instead. The few bytes each flush costs are what keep the batch streaming, [keep-alive frames](/docs/plugins/batch#keep-alive-timer) included. ## Client diff --git a/apps/content/docs/plugins/batch.mdx b/apps/content/docs/plugins/batch.mdx index 47952c9428..df721edfae 100644 --- a/apps/content/docs/plugins/batch.mdx +++ b/apps/content/docs/plugins/batch.mdx @@ -56,32 +56,13 @@ const cors = new CORSHandlerPlugin({ ::: -## Response Modes +## Streaming Responses -By default, the plugin uses `streaming` mode. Responses are sent as soon as they are ready, so one slow request does not block the rest of the batch. - -If your environment does not support streaming responses, such as some serverless platforms or older browsers, switch to `buffered` mode instead. In this mode, all responses are collected and sent together, unless a procedure returns an [AsyncIteratorObject](/docs/async-iterator-object) or a [`ReadableStream`](/docs/binary-data). Such a stream can stay open indefinitely, so the batch switches to streaming as soon as one is returned: the server writes out the responses collected so far, then sends the rest as they are ready. If your client cannot read streamed responses, keep these procedures out of the batch with [`filter`](#filtering-requests). - -```ts -const link = new RPCLink({ - url: '/rpc', - plugins: [ - new BatchLinkPlugin({ - mode: 'buffered', - groups: [ - { - condition: () => true, - context: {}, - }, - ], - }), - ], -}) -``` +The server starts the batch response once every procedure has responded, so it can carry their [headers](#response-headers). The response streams if one of them returned an [AsyncIteratorObject](/docs/async-iterator-object) or a [`ReadableStream`](/docs/binary-data). If your client cannot read streamed responses, keep these procedures out of the batch with [`filter`](#filtering-requests). ### Keep-Alive Timer -In **streaming** mode, or once a buffered batch switches to streaming, long-running batch responses may remain idle for extended periods. The server plugin can send keep-alive frames to keep the connection alive. +A streaming batch response may stay idle for long periods. The server plugin can send keep-alive frames to keep the connection alive. ```ts const handler = new RPCHandler(router, { @@ -238,26 +219,9 @@ const link = new RPCLink({ Per-request data passed through the link's [`headers` option](/docs/rpc/link#url-and-header-options) is resolved for each request, so it is safe without `scope`. ::: -## Limitations +## Response Headers -Cookies set in a batched call are never stored, because the browser only reads headers from the batch response itself, not from each subresponse. Send procedures that set cookies outside the batch: - -```ts -const link = new RPCLink({ - url: '/rpc', - plugins: [ - new BatchLinkPlugin({ - filter: ({ path }) => path[0] !== 'auth', // auth procedures set cookies - groups: [ - { - condition: () => true, - context: {}, - }, - ], - }), - ], -}) -``` +The batch response carries the cookies of its subresponses, and the headers they all share. Other headers stay on their subresponse. To change what stays on each subresponse, set the server plugin's `mapSubresponse` option. ## Learn More diff --git a/packages/client/src/plugins/batch.test.ts b/packages/client/src/plugins/batch.test.ts index ae09422d3e..778f655813 100644 --- a/packages/client/src/plugins/batch.test.ts +++ b/packages/client/src/plugins/batch.test.ts @@ -114,7 +114,7 @@ async function startStreamingBatch() { return { status: 207, headers: {}, resolveBody: async () => stream } }, }, { - plugins: [new BatchLinkPlugin({ groups: [{ condition: () => true, context: {} }], mode: 'streaming' })], + plugins: [new BatchLinkPlugin({ groups: [{ condition: () => true, context: {} }] })], }) const outputA = link.call(['a'], {}, { context: {} }) @@ -351,7 +351,6 @@ describe('batchLinkPlugin', () => { const link = new StandardLink(codec, transport, { plugins: [new BatchLinkPlugin({ groups: [defaultGroup], - mode: 'buffered', })], }) @@ -362,7 +361,7 @@ describe('batchLinkPlugin', () => { expect(transport.send).toHaveBeenCalledTimes(1) const sentRequest = vi.mocked(transport.send).mock.calls[0]![0] - expect(sentRequest.headers['orpc-batch']).toBe('buffered') + expect(sentRequest.headers['orpc-batch']).toBe('1') }) it('splits batches when exceeding maxSize', async () => { @@ -372,7 +371,6 @@ describe('batchLinkPlugin', () => { const link = new StandardLink(codec, transport, { plugins: [new BatchLinkPlugin({ groups: [defaultGroup], - mode: 'buffered', maxSize: 2, })], }) @@ -535,7 +533,6 @@ describe('batchLinkPlugin', () => { const link = new StandardLink(codec, transport, { plugins: [new BatchLinkPlugin({ groups: [defaultGroup], - mode: 'buffered', })], }) @@ -552,14 +549,14 @@ describe('batchLinkPlugin', () => { const sentGetRequest = vi.mocked(transport.send).mock.calls.find(([request]) => request.method === 'GET')![0] expect(extractBatchMessagesFromRequest(sentGetRequest).map(message => message.json.method)).toEqual(['GET', 'GET']) - expect(sentGetRequest.headers['orpc-batch']).toBe('buffered') + expect(sentGetRequest.headers['orpc-batch']).toBe('1') const sentQueryRequest = vi.mocked(transport.send).mock.calls.find(([request]) => request.method === 'QUERY')![0] expect(extractBatchMessagesFromRequest(sentQueryRequest).map(message => message.json.method)).toEqual(['QUERY', 'QUERY']) const sentPostRequest = vi.mocked(transport.send).mock.calls.find(([request]) => request.method === 'POST')![0] expect(extractBatchMessagesFromRequest(sentPostRequest).map(message => message.json.method)).toEqual(['PUT', 'PUT']) - expect(sentPostRequest.headers['orpc-batch']).toBe('buffered') + expect(sentPostRequest.headers['orpc-batch']).toBe('1') }) it('aborts grouped batch request when all sub-requests are aborted', async () => { @@ -578,7 +575,7 @@ describe('batchLinkPlugin', () => { }) const link = new StandardLink(codec, transport, { - plugins: [new BatchLinkPlugin({ groups: [defaultGroup], mode: 'streaming' })], + plugins: [new BatchLinkPlugin({ groups: [defaultGroup] })], }) const controller1 = new AbortController() @@ -610,7 +607,7 @@ describe('batchLinkPlugin', () => { const transport = makeTransport() const link = new StandardLink(codec, transport, { - plugins: [new BatchLinkPlugin({ groups: [defaultGroup], mode: 'buffered', wait: 100 })], + plugins: [new BatchLinkPlugin({ groups: [defaultGroup], wait: 100 })], }) const promise1 = link.call(['a'], {}, { context: {} }) @@ -645,7 +642,7 @@ describe('batchLinkPlugin', () => { const transport = makeTransport() const link = new StandardLink(codec, transport, { - plugins: [new BatchLinkPlugin({ groups: [defaultGroup], mode: 'buffered', wait: 100 })], + plugins: [new BatchLinkPlugin({ groups: [defaultGroup], wait: 100 })], }) const controller = new AbortController() @@ -671,7 +668,6 @@ describe('batchLinkPlugin', () => { const link = new StandardLink(codec, transport, { plugins: [new BatchLinkPlugin({ groups: [defaultGroup], - mode: 'buffered', url: async () => { await sleep(50) return '/__batch__' as const @@ -702,7 +698,6 @@ describe('batchLinkPlugin', () => { const link = new StandardLink(codec, transport, { plugins: [new BatchLinkPlugin({ groups: [defaultGroup], - mode: 'buffered', mapSubrequest: ({ request }) => { if (request.url === '/a') { // Runs after the peer starts sending this subrequest, before its request message @@ -735,7 +730,6 @@ describe('batchLinkPlugin', () => { const link = new StandardLink(codec, transport, { plugins: [new BatchLinkPlugin({ groups: [defaultGroup], - mode: 'buffered', mapSubrequest: ({ request }) => { queueMicrotask(() => controller.abort(new Error('TEST_ABORT'))) return request @@ -767,7 +761,6 @@ describe('batchLinkPlugin', () => { const link = new StandardLink(codec, transport, { plugins: [new BatchLinkPlugin({ groups: [defaultGroup], - mode: 'buffered', mapSubrequest: ({ request }) => { if (request.url !== '/b') { return request @@ -808,7 +801,7 @@ describe('batchLinkPlugin', () => { }) const link = new StandardLink(codec, transport, { - plugins: [new BatchLinkPlugin({ groups: [defaultGroup], mode: 'buffered' })], + plugins: [new BatchLinkPlugin({ groups: [defaultGroup] })], }) await Promise.all([ @@ -835,7 +828,7 @@ describe('batchLinkPlugin', () => { }) const link = new StandardLink(codec, transport, { - plugins: [new BatchLinkPlugin({ groups: [defaultGroup], mode: 'streaming', wait: 20 })], + plugins: [new BatchLinkPlugin({ groups: [defaultGroup], wait: 20 })], }) const controller1 = new AbortController() @@ -976,7 +969,7 @@ describe('batchLinkPlugin', () => { }) const link = new StandardLink(codec, transport, { - plugins: [new BatchLinkPlugin({ groups: [defaultGroup], mode: 'buffered' })], + plugins: [new BatchLinkPlugin({ groups: [defaultGroup] })], }) await Promise.all([ @@ -1020,7 +1013,7 @@ describe('batchLinkPlugin', () => { }) const link = new StandardLink(codec, transport, { - plugins: [new BatchLinkPlugin({ groups: [defaultGroup], mode: 'streaming' })], + plugins: [new BatchLinkPlugin({ groups: [defaultGroup] })], }) await Promise.all([ @@ -1067,7 +1060,7 @@ describe('batchLinkPlugin', () => { }) const link = new StandardLink(codec, transport, { - plugins: [new BatchLinkPlugin({ groups: [defaultGroup], mode: 'streaming' })], + plugins: [new BatchLinkPlugin({ groups: [defaultGroup] })], }) await Promise.all([ @@ -1111,7 +1104,7 @@ describe('batchLinkPlugin', () => { }) const link = new StandardLink(codec, transport, { - plugins: [new BatchLinkPlugin({ groups: [defaultGroup], mode: 'streaming' })], + plugins: [new BatchLinkPlugin({ groups: [defaultGroup] })], }) await Promise.all([ @@ -1133,7 +1126,7 @@ describe('batchLinkPlugin', () => { }) const link = new StandardLink(codec, transport, { - plugins: [new BatchLinkPlugin({ groups: [defaultGroup], mode: 'buffered' })], + plugins: [new BatchLinkPlugin({ groups: [defaultGroup] })], }) await Promise.all([ @@ -1155,7 +1148,7 @@ describe('batchLinkPlugin', () => { }) const link = new StandardLink(codec, transport, { - plugins: [new BatchLinkPlugin({ groups: [defaultGroup], mode: 'buffered' })], + plugins: [new BatchLinkPlugin({ groups: [defaultGroup] })], }) await Promise.all([ @@ -1177,7 +1170,7 @@ describe('batchLinkPlugin', () => { }) const link = new StandardLink(codec, transport, { - plugins: [new BatchLinkPlugin({ groups: [defaultGroup], mode: 'buffered' })], + plugins: [new BatchLinkPlugin({ groups: [defaultGroup] })], }) await Promise.all([ @@ -1201,7 +1194,7 @@ describe('batchLinkPlugin', () => { }) const link = new StandardLink(codec, transport, { - plugins: [new BatchLinkPlugin({ groups: [defaultGroup], mode: 'buffered' })], + plugins: [new BatchLinkPlugin({ groups: [defaultGroup] })], }) await Promise.all([ @@ -1228,7 +1221,7 @@ describe('batchLinkPlugin', () => { }) const link = new StandardLink(codec, transport, { - plugins: [new BatchLinkPlugin({ groups: [defaultGroup], mode: 'buffered' })], + plugins: [new BatchLinkPlugin({ groups: [defaultGroup] })], }) await Promise.all([ @@ -1255,7 +1248,7 @@ describe('batchLinkPlugin', () => { }) const link = new StandardLink(codec, transport, { - plugins: [new BatchLinkPlugin({ groups: [defaultGroup], mode: 'buffered' })], + plugins: [new BatchLinkPlugin({ groups: [defaultGroup] })], }) await Promise.all([ @@ -1284,7 +1277,7 @@ describe('batchLinkPlugin', () => { }) const link = new StandardLink(codec, transport, { - plugins: [new BatchLinkPlugin({ groups: [defaultGroup], mode: 'buffered' })], + plugins: [new BatchLinkPlugin({ groups: [defaultGroup] })], }) await Promise.all([ @@ -1295,7 +1288,7 @@ describe('batchLinkPlugin', () => { }) describe('error batch responses', () => { - it.each(['buffered', 'streaming'] as const)('forwards %s batch responses with status >= 400 to every subrequest', async (mode) => { + it('forwards batch responses with status >= 400 to every subrequest', async () => { const codec = makeCodec() const transport = makeTransport() @@ -1308,7 +1301,7 @@ describe('batchLinkPlugin', () => { }) const link = new StandardLink(codec, transport, { - plugins: [new BatchLinkPlugin({ groups: [defaultGroup], mode })], + plugins: [new BatchLinkPlugin({ groups: [defaultGroup] })], }) await Promise.all([ @@ -1337,7 +1330,7 @@ describe('batchLinkPlugin', () => { }) const link = new StandardLink(codec, transport, { - plugins: [new BatchLinkPlugin({ groups: [defaultGroup], mode: 'buffered' })], + plugins: [new BatchLinkPlugin({ groups: [defaultGroup] })], }) await Promise.all([ @@ -1354,7 +1347,7 @@ describe('batchLinkPlugin', () => { const transport = makeTransport() const link = new StandardLink(codec, transport, { - plugins: [new BatchLinkPlugin({ groups: [defaultGroup], mode: 'buffered' })], + plugins: [new BatchLinkPlugin({ groups: [defaultGroup] })], }) await Promise.all([ @@ -1365,7 +1358,7 @@ describe('batchLinkPlugin', () => { }) describe('batch option failures', () => { - it.each(['maxSize', 'url', 'headers', 'mode', 'mapSubrequest'] as const)('rejects every subrequest when %s throws', async (option) => { + it.each(['maxSize', 'url', 'headers', 'mapSubrequest'] as const)('rejects every subrequest when %s throws', async (option) => { const error = new Error('option failed') const codec = makeCodec() const transport = makeTransport() @@ -1448,7 +1441,6 @@ describe('batchLinkPlugin', () => { const link = new StandardLink(codec, transport, { plugins: [new BatchLinkPlugin({ groups: [defaultGroup], - mode: 'buffered', maxUrlLength: 1, })], }) @@ -1477,7 +1469,6 @@ describe('batchLinkPlugin', () => { const link = new StandardLink(codec, transport, { plugins: [new BatchLinkPlugin({ groups: [defaultGroup], - mode: 'buffered', url: () => '/custom/__batch__?existing=1#anchor', })], }) @@ -1511,7 +1502,6 @@ describe('batchLinkPlugin', () => { const link = new StandardLink(codec, transport, { plugins: [new BatchLinkPlugin({ groups: [defaultGroup], - mode: 'buffered', url: () => '/custom-no-hash/__batch__?existing=1', })], }) @@ -1603,8 +1593,8 @@ describe('batchLinkPlugin', () => { expect(transport.send).toHaveBeenCalledTimes(2) expect(vi.mocked(transport.send).mock.calls.map(([request]) => request.headers)).toEqual([ - { 'x-user': 'alice', 'orpc-batch': 'streaming' }, - { 'x-user': 'bob', 'orpc-batch': 'streaming' }, + { 'x-user': 'alice', 'orpc-batch': '1' }, + { 'x-user': 'bob', 'orpc-batch': '1' }, ]) }) diff --git a/packages/client/src/plugins/batch.ts b/packages/client/src/plugins/batch.ts index 514bba585f..3053082b4e 100644 --- a/packages/client/src/plugins/batch.ts +++ b/packages/client/src/plugins/batch.ts @@ -7,8 +7,6 @@ import { captureAsyncContext, defer, isAsyncIteratorObject, loadBytes, once, pro import { parseStandardUrl } from '@standard-server/core' import { ClientPeer, decodePeerMessage, isServerPeerSendMessage } from '@standard-server/peer' -export type BatchLinkPluginMode = 'streaming' | 'buffered' - export interface BatchLinkPluginGroup { /** * Determines whether a request should be included in this batch group. @@ -65,14 +63,6 @@ export interface BatchLinkPluginOptions { */ wait?: number - /** - * The batch response mode. - * A `buffered` batch still streams once a procedure in it returns a stream. - * - * @default 'streaming' - */ - mode?: Value, ...StandardLinkTransportInterceptorOptions[]]]> - /** * URL for the batch request. * @@ -129,7 +119,6 @@ export class BatchLinkPlugin implements StandardLinkPlu private readonly scope: Exclude['scope'], undefined> private readonly maxSize: Exclude['maxSize'], undefined> private readonly wait: Exclude['wait'], undefined> - private readonly mode: Exclude['mode'], undefined> private readonly batchUrl: Exclude['url'], undefined> private readonly maxUrlLength: Exclude['maxUrlLength'], undefined> private readonly batchHeaders: Exclude['headers'], undefined> @@ -144,7 +133,6 @@ export class BatchLinkPlugin implements StandardLinkPlu this.scope = options.scope ?? (() => undefined) this.maxSize = options.maxSize ?? 10 this.wait = options.wait ?? 0 - this.mode = options.mode ?? 'streaming' this.batchUrl = options.url ?? ((options) => { const [pathname] = parseStandardUrl(options[0].request.url) return `${pathname}/__batch__` @@ -301,7 +289,6 @@ export class BatchLinkPlugin implements StandardLinkPlu const url = await value(this.batchUrl, subOptionsList) const headers = await value(this.batchHeaders, subOptionsList) - const mode = value(this.mode, subOptionsList) let suppressErrorFromCurrentBatch = false const controller = new AbortController() @@ -370,7 +357,7 @@ export class BatchLinkPlugin implements StandardLinkPlu const request: StandardRequest = { url, method, - headers: { ...headers, 'orpc-batch': mode }, + headers: { ...headers, 'orpc-batch': '1' }, signal: controller.signal, } diff --git a/packages/node/src/batch-response-compression-handler-plugin.test.ts b/packages/node/src/batch-response-compression-handler-plugin.test.ts index da7efc2683..18a9b42d6b 100644 --- a/packages/node/src/batch-response-compression-handler-plugin.test.ts +++ b/packages/node/src/batch-response-compression-handler-plugin.test.ts @@ -16,13 +16,12 @@ type Encoding = 'gzip' | 'deflate' | 'deflate-raw' * A batch request for one subrequest per url, which is all any of these tests needs to name. */ function createBatchRequest( - mode: 'buffered' | 'streaming', urls: string[], headers: Record = { 'accept-encoding': 'gzip' }, ) { return new Request('https://example.com/__batch__', { method: 'POST', - headers: { ...headers, 'content-type': 'application/json', 'orpc-batch': mode }, + headers: { ...headers, 'content-type': 'application/json', 'orpc-batch': '1' }, body: JSON.stringify(urls.map((url, id) => ({ kind: 'request', id, @@ -40,6 +39,9 @@ describe('batchResponseCompressionHandlerPlugin', () => { const router = { ping: os.handler(() => largeValue), blob: os.handler(() => new Blob([largeValue], { type: 'application/octet-stream' })), + events: os.handler(async function* () { + yield largeValue + }), } function createHandler( @@ -54,8 +56,8 @@ describe('batchResponseCompressionHandlerPlugin', () => { }) } - it('compresses streaming batch responses', async () => { - const { matched, response } = await createHandler().handle(createBatchRequest('streaming', ['/ping', '/ping'])) + it('compresses batch responses that stream', async () => { + const { matched, response } = await createHandler().handle(createBatchRequest(['/events', '/ping'])) expect(matched).toBe(true) expect(response!.status).toBe(207) @@ -72,17 +74,19 @@ describe('batchResponseCompressionHandlerPlugin', () => { /** * The reason this plugin exists: `CompressionStream` cannot flush, so every message of a - * streaming batch would sit in the compressor until the slowest subrequest resolved. + * batch that streams would sit in the compressor until the stream ended. */ it('delivers each message as soon as it is produced, without waiting for the whole batch', async () => { const slow = promiseWithResolvers() const handler = createHandler({}, { - fast: os.handler(() => largeValue), - slow: os.handler(() => slow.promise), + stream: os.handler(async function* () { + yield largeValue + yield await slow.promise + }), }) - const { response } = await handler.handle(createBatchRequest('streaming', ['/fast', '/slow'])) + const { response } = await handler.handle(createBatchRequest(['/stream'])) const reader = response!.body!.pipeThrough(new DecompressionStream('gzip')).getReader() const decoder = new TextDecoder() let received = '' @@ -102,20 +106,26 @@ describe('batchResponseCompressionHandlerPlugin', () => { }) /** - * Keep-alive frames hold the connection open while the remaining subrequests are pending, so a - * compressor that buffered them would let the very connection they protect time out. + * Keep-alive frames hold the connection open while a stream is idle, so a compressor + * that buffered them would let the very connection they protect time out. */ it('keeps sending keep-alive frames while the batch is idle', async () => { const pending = promiseWithResolvers() - const handler = new RPCHandler({ ping: os.handler(() => 'pong'), pending: os.handler(() => pending.promise) }, { + const router = { + pending: os.handler(async function* () { + yield await pending.promise + }), + } + + const handler = new RPCHandler(router, { plugins: [ new BatchHandlerPlugin({ keepAlive: { enabled: true, interval: 20 } }), new BatchResponseCompressionHandlerPlugin(), ], }) - const { response } = await handler.handle(createBatchRequest('streaming', ['/ping', '/pending'])) + const { response } = await handler.handle(createBatchRequest(['/pending'])) const reader = response!.body!.pipeThrough(new DecompressionStream('gzip')).getReader() let received = Buffer.alloc(0) @@ -127,7 +137,7 @@ describe('batchResponseCompressionHandlerPlugin', () => { } } - // The ping response frame, then two zero-length length-prefixed frames, four bytes each + // The response message frame, then two zero-length length-prefixed frames, four bytes each await readAtLeast(4) const messageEnd = 4 + received.readUInt32BE(0) await readAtLeast(messageEnd + 8) @@ -137,8 +147,8 @@ describe('batchResponseCompressionHandlerPlugin', () => { await reader.cancel() }) - it('compresses buffered batch responses that carry binary subresponses', async () => { - const { response } = await createHandler().handle(createBatchRequest('buffered', ['/blob', '/ping'])) + it('compresses batch responses that carry binary subresponses', async () => { + const { response } = await createHandler().handle(createBatchRequest(['/blob', '/ping'])) expect(response!.status).toBe(207) expect(response!.headers.get('content-encoding')).toBe('gzip') @@ -152,11 +162,11 @@ describe('batchResponseCompressionHandlerPlugin', () => { }) /** - * A json-only buffered batch is a plain array of messages rather than a framed body, and it is + * A json-only batch is a plain array of messages rather than a framed body, and it is * still a batch response, so this plugin compresses it as json instead of leaving it behind. */ - it('compresses json-only buffered batch responses', async () => { - const { response } = await createHandler().handle(createBatchRequest('buffered', ['/ping', '/ping'])) + it('compresses json-only batch responses', async () => { + const { response } = await createHandler().handle(createBatchRequest(['/ping', '/ping'])) expect(response!.status).toBe(207) expect(response!.headers.get('content-encoding')).toBe('gzip') @@ -175,13 +185,13 @@ describe('batchResponseCompressionHandlerPlugin', () => { it.each([ ['with a binary subresponse', '/blob'], ['of json only', '/tiny'], - ] as const)('leaves a buffered batch %s below the threshold alone', async (_case, url) => { + ] as const)('leaves a batch %s below the threshold alone', async (_case, url) => { const handler = createHandler({ threshold: 10 * 1024 }, { ...router, tiny: os.handler(() => 'ok'), }) - const { response } = await handler.handle(createBatchRequest('buffered', [url])) + const { response } = await handler.handle(createBatchRequest([url])) expect(response!.headers.get('content-encoding')).toBe(null) }) @@ -192,7 +202,7 @@ describe('batchResponseCompressionHandlerPlugin', () => { it('leaves a failed batch response alone', async () => { const { response } = await createHandler().handle(new Request('https://example.com/__batch__', { method: 'POST', - headers: { 'accept-encoding': 'gzip', 'content-type': 'application/json', 'orpc-batch': 'streaming' }, + headers: { 'accept-encoding': 'gzip', 'content-type': 'application/json', 'orpc-batch': '1' }, body: JSON.stringify({ not: 'an array of messages' }), })) @@ -230,7 +240,7 @@ describe('batchResponseCompressionHandlerPlugin', () => { })], }) - const { response } = await handler.handle(createBatchRequest('streaming', ['/ping'])) + const { response } = await handler.handle(createBatchRequest(['/ping'])) expect(response!.headers.get('content-encoding')).toBe(null) await expect(response!.text()).resolves.toContain('invalid') @@ -267,7 +277,7 @@ describe('batchResponseCompressionHandlerPlugin', () => { })], }) - const { response } = await handler.handle(createBatchRequest('streaming', ['/ping'])) + const { response } = await handler.handle(createBatchRequest(['/ping'])) expect(response!.status).toBe(206) expect(response!.headers.get('content-encoding')).toBe(null) @@ -284,7 +294,7 @@ describe('batchResponseCompressionHandlerPlugin', () => { ] as const)('given accept-encoding %s, compresses with %s', async (acceptEncoding, encoding) => { const handler = createHandler({ encodings: ['gzip', 'deflate', 'deflate-raw'] }) - const { response } = await handler.handle(createBatchRequest('streaming', ['/ping'], { 'accept-encoding': acceptEncoding })) + const { response } = await handler.handle(createBatchRequest(['/ping'], { 'accept-encoding': acceptEncoding })) expect(response!.headers.get('content-encoding')).toBe(encoding) await expect(decompress(response!, encoding)).resolves.toContain(largeValue) @@ -293,7 +303,7 @@ describe('batchResponseCompressionHandlerPlugin', () => { it('prefers the first configured encoding the client accepts', async () => { const handler = createHandler({ encodings: ['deflate', 'gzip'] }) - const { response } = await handler.handle(createBatchRequest('streaming', ['/ping'], { 'accept-encoding': 'gzip, deflate' })) + const { response } = await handler.handle(createBatchRequest(['/ping'], { 'accept-encoding': 'gzip, deflate' })) expect(response!.headers.get('content-encoding')).toBe('deflate') }) @@ -303,7 +313,7 @@ describe('batchResponseCompressionHandlerPlugin', () => { ['the client rejects every configured coding', { 'accept-encoding': 'gzip;q=0, deflate;q=0' }], ['the client sends no accept-encoding', {}], ])('does not compress when %s', async (_case, headers) => { - const { response } = await createHandler().handle(createBatchRequest('streaming', ['/ping'], headers)) + const { response } = await createHandler().handle(createBatchRequest(['/ping'], headers)) expect(response!.headers.get('content-encoding')).toBe(null) await expect(response!.text()).resolves.toContain(largeValue) @@ -318,7 +328,7 @@ describe('batchResponseCompressionHandlerPlugin', () => { ], }) - const { response } = await handler.handle(createBatchRequest('streaming', ['/ping'])) + const { response } = await handler.handle(createBatchRequest(['/ping'])) expect(response!.headers.get('vary')).toBe('origin, accept-encoding') }) @@ -331,7 +341,7 @@ describe('batchResponseCompressionHandlerPlugin', () => { ], }) - const { response } = await handler.handle(createBatchRequest('streaming', ['/ping'])) + const { response } = await handler.handle(createBatchRequest(['/ping'])) expect(response!.headers.get('content-encoding')).toBe('gzip') // A strong tag shared with the identity bytes would let If-Range splice them into compressed ones @@ -349,7 +359,7 @@ describe('batchResponseCompressionHandlerPlugin', () => { ], }) - const { response } = await handler.handle(createBatchRequest('streaming', ['/ping'])) + const { response } = await handler.handle(createBatchRequest(['/ping'])) expect(response!.headers.get('content-encoding')).toBe('gzip') await expect(decompress(response!, 'gzip')).resolves.toContain(largeValue) @@ -363,7 +373,7 @@ describe('batchResponseCompressionHandlerPlugin', () => { ], }) - const { response } = await handler.handle(createBatchRequest('streaming', ['/ping'])) + const { response } = await handler.handle(createBatchRequest(['/ping'])) expect(response!.headers.get('content-encoding')).toBe(null) await expect(response!.text()).resolves.toContain(largeValue) @@ -377,7 +387,7 @@ describe('batchResponseCompressionHandlerPlugin', () => { plugins: [new BatchHandlerPlugin(), ...createPlugins()], }) - const { response } = await handler.handle(createBatchRequest('streaming', ['/ping'])) + const { response } = await handler.handle(createBatchRequest(['/ping'])) expect(response!.headers.get('content-encoding')).toBe('gzip') expect(response!.headers.get('vary')).toBe('accept-encoding') @@ -386,13 +396,13 @@ describe('batchResponseCompressionHandlerPlugin', () => { it('stops a batch cleanly when the client cancels mid-stream', async () => { const handler = createHandler({}, { - fast: os.handler(() => largeValue), - never: os.handler(({ signal }) => new Promise((_, reject) => { - signal?.addEventListener('abort', () => reject(signal.reason)) - })), + stream: os.handler(async function* ({ signal }) { + yield largeValue + await new Promise((_, reject) => signal?.addEventListener('abort', () => reject(signal.reason))) + }), }) - const { response } = await handler.handle(createBatchRequest('streaming', ['/fast', '/never'])) + const { response } = await handler.handle(createBatchRequest(['/stream'])) const reader = response!.body!.getReader() const { value } = await reader.read() diff --git a/packages/node/src/batch-response-compression-handler-plugin.ts b/packages/node/src/batch-response-compression-handler-plugin.ts index 23da2d3afb..5879f1b31f 100644 --- a/packages/node/src/batch-response-compression-handler-plugin.ts +++ b/packages/node/src/batch-response-compression-handler-plugin.ts @@ -101,7 +101,7 @@ export class BatchResponseCompressionHandlerPlugin implements let bodyHeaders: StandardHeaders /** - * A buffered batch with no binary subresponse stays a plain array of messages, serialized + * A batch with no binary or stream subresponse stays a plain array of messages, serialized * as json by the adapter. The header and status above already say this is a batch envelope, * so the array itself is the only thing left to recognise. */ diff --git a/packages/server/src/plugins/batch.test.ts b/packages/server/src/plugins/batch.test.ts index 8768e0e064..eb64a414eb 100644 --- a/packages/server/src/plugins/batch.test.ts +++ b/packages/server/src/plugins/batch.test.ts @@ -1,4 +1,6 @@ +import type { StandardResponse } from '@standard-server/core' import type { AnyRouter } from '../router' +import type { ResponseHeadersHandlerPluginContext } from './response-headers' import { ORPCError } from '@orpc/client' import { promiseWithResolvers, sleep } from '@orpc/shared' import { decodePeerMessage } from '@standard-server/peer' @@ -7,6 +9,7 @@ import { os } from '../builder' import { BatchHandlerPlugin } from './batch' import { RequestCompressionHandlerPlugin } from './request-compression' import { RequestLimitHandlerPlugin } from './request-limit' +import { ResponseHeadersHandlerPlugin } from './response-headers' beforeEach(() => { vi.clearAllMocks() @@ -22,7 +25,6 @@ function makePeerRequestMessage(id: number | string, url: string, method = 'POST } function createBatchRequest(options: { - mode: 'buffered' | 'streaming' messages?: unknown method?: 'POST' | 'GET' | 'QUERY' data?: string @@ -33,14 +35,14 @@ function createBatchRequest(options: { return new Request(`https://example.com/__batch__${search}`, { method: 'GET', - headers: { 'orpc-batch': options.mode }, + headers: { 'orpc-batch': '1' }, signal: options.signal, }) } return new Request('https://example.com/__batch__', { method: options.method ?? 'POST', - headers: { 'orpc-batch': options.mode, 'content-type': 'application/json' }, + headers: { 'orpc-batch': '1', 'content-type': 'application/json' }, body: JSON.stringify(options.messages), signal: options.signal, }) @@ -109,7 +111,7 @@ describe('batchHandlerPlugin', () => { plugin = new BatchHandlerPlugin(), handlerRouter: AnyRouter = router, ) => new RPCHandler(handlerRouter, { - plugins: [plugin], + plugins: [plugin, new ResponseHeadersHandlerPlugin()], }) it('passes through non-batch requests', async () => { @@ -124,8 +126,8 @@ describe('batchHandlerPlugin', () => { expect(handlerFn).toHaveBeenCalledTimes(1) }) - describe('buffered mode', () => { - it('handles buffered batch POST requests', async () => { + describe('responses', () => { + it('handles batch POST requests', async () => { const handler = createHandler() const peerMessages = [ makePeerRequestMessage(0, '/ping'), @@ -133,7 +135,6 @@ describe('batchHandlerPlugin', () => { ] const { matched, response } = await handler.handle(createBatchRequest({ - mode: 'buffered', messages: peerMessages, })) @@ -146,7 +147,7 @@ describe('batchHandlerPlugin', () => { expect(handlerFn).toHaveBeenCalledTimes(2) }) - it('returns binary payload in buffered mode when sub-response contains binary', async () => { + it('returns binary payload when sub-response contains binary', async () => { const binaryRouter = { file: os.handler(() => new Blob([new Uint8Array([1, 2, 3])], { type: 'application/octet-stream', @@ -162,7 +163,6 @@ describe('batchHandlerPlugin', () => { ] const { response } = await handler.handle(createBatchRequest({ - mode: 'buffered', messages: peerMessages, })) @@ -181,7 +181,6 @@ describe('batchHandlerPlugin', () => { const handler = createHandler() const { matched, response } = await handler.handle(createBatchRequest({ - mode: 'buffered', messages: [makePeerRequestMessage(0, '/nonexistent')], })) @@ -198,7 +197,6 @@ describe('batchHandlerPlugin', () => { const handler = createHandler() const { response } = await handler.handle(createBatchRequest({ - mode: 'buffered', messages: 'not-an-array', })) @@ -215,7 +213,7 @@ describe('batchHandlerPlugin', () => { const { response } = await handler.handle(new Request('https://example.com/__batch__', { method: 'POST', headers: { - 'orpc-batch': 'buffered', + 'orpc-batch': '1', 'content-type': 'application/json', 'content-encoding': 'gzip, gzip, gzip, gzip, gzip, gzip', }, @@ -233,7 +231,6 @@ describe('batchHandlerPlugin', () => { }) const { response } = await handler.handle(createBatchRequest({ - mode: 'buffered', messages: [makePeerRequestMessage(0, '/ping')], })) @@ -250,7 +247,6 @@ describe('batchHandlerPlugin', () => { ] const { response } = await handler.handle(createBatchRequest({ - mode: 'buffered', messages: peerMessages, })) @@ -273,7 +269,32 @@ describe('batchHandlerPlugin', () => { })) const { response } = await handler.handle(createBatchRequest({ - mode: 'buffered', + messages: [makePeerRequestMessage(0, '/ping')], + })) + + expect(response!.status).toBe(207) + + const body = await response!.json() as any + expect(body[0].json.status).toBe(500) + expect(body[0].json.body).toBe('Internal server error') + expect(rejectSpy).toHaveBeenCalledTimes(1) + }) + + it('returns 500 sub-response when mapSubresponse throws', async ({ onTestFinished }) => { + const rejectSpy = vi.spyOn(Promise, 'reject') + .mockImplementation(() => new Promise(() => {}) as Promise) + + onTestFinished(() => { + rejectSpy.mockRestore() + }) + + const handler = createHandler(new BatchHandlerPlugin({ + mapSubresponse: () => { + throw new Error('boom') + }, + })) + + const { response } = await handler.handle(createBatchRequest({ messages: [makePeerRequestMessage(0, '/ping')], })) @@ -305,9 +326,8 @@ describe('batchHandlerPlugin', () => { stream: os.handler(() => createStream(gate.promise)), }) - // The stream stays open until the gate is released, so a buffered batch would never answer + // The stream stays open until the gate is released, so the batch must stream to answer const { response } = await handler.handle(createBatchRequest({ - mode: 'buffered', messages: [makePeerRequestMessage('0', '/ping'), makePeerRequestMessage('1', '/stream')], })) @@ -328,6 +348,33 @@ describe('batchHandlerPlugin', () => { ]) }) + it('waits for every sub-response before it streams', async () => { + const gate = promiseWithResolvers() + const handler = createHandler(new BatchHandlerPlugin(), { + ...createSlowRouter(gate.promise), + events: os.handler(async function* () { + yield 'event' + }), + }) + + let settled = false + const result = handler.handle(createBatchRequest({ + messages: [makePeerRequestMessage('0', '/events'), makePeerRequestMessage('1', '/slow')], + })).finally(() => { + settled = true + }) + + await sleep(10) + expect(settled).toBe(false) + + gate.resolve() + const { response } = await result + expect(response!.headers.get('content-type')).toEqual('application/vnd.orpc.batch') + + const messages = await readAllMessages(response!) + expect(messages.filter(m => m.kind === 'response').map(m => m.id).sort()).toEqual(['0', '1']) + }) + it('writes out the sub-responses buffered before it, binary included', async () => { const gate = promiseWithResolvers() const ping = vi.fn(() => 'pong') @@ -346,7 +393,6 @@ describe('batchHandlerPlugin', () => { }) const result = handler.handle(createBatchRequest({ - mode: 'buffered', messages: [ makePeerRequestMessage('0', '/ping'), makePeerRequestMessage('1', '/file'), @@ -384,97 +430,6 @@ describe('batchHandlerPlugin', () => { }) }) - describe('streaming mode', () => { - it('handles streaming batch POST requests returning readable stream', async () => { - const handler = createHandler() - const peerMessages = [makePeerRequestMessage(0, '/ping')] - - const { matched, response } = await handler.handle(createBatchRequest({ - mode: 'streaming', - messages: peerMessages, - })) - - expect(matched).toBe(true) - expect(response!.status).toBe(207) - expect(response!.headers.get('standard-server')).toEqual('octet-stream') - expect(response!.headers.get('content-type')).toEqual('application/vnd.orpc.batch') - expect(handlerFn).toHaveBeenCalledTimes(1) - - // Verify the binary format: 4-byte length prefix + encoded peer message. - const buffer = new Uint8Array(await response!.arrayBuffer()) - expect(buffer.length).toBeGreaterThan(4) - - const { messageLength } = readLengthPrefixedChunk(buffer) - expect(buffer.length).toBe(4 + messageLength) - }) - - it('encodes streaming batch response as binary when sub-response is a blob', async () => { - const binaryRouter = { - file: os.handler(() => new Blob(['__TEST__'], { - type: 'application/octet-stream', - })), - } - - const handler = createHandler(new BatchHandlerPlugin(), binaryRouter) - - const { response } = await handler.handle(createBatchRequest({ - mode: 'streaming', - messages: [makePeerRequestMessage(0, '/file')], - })) - - expect(response!.status).toBe(207) - - const buffer = new Uint8Array(await response!.arrayBuffer()) - expect(buffer.length).toBeGreaterThan(4) - - const { messageLength, payload } = readLengthPrefixedChunk(buffer) - expect(messageLength).toBe(payload.length) - expect(new TextDecoder().decode(payload)).toContain('__TEST__') - }) - - describe('holding the response', () => { - function createGatedHandler() { - const gate = promiseWithResolvers() - - const handler = createHandler(new BatchHandlerPlugin(), createSlowRouter(gate.promise)) - - return { handler, gate } - } - - it('holds the response until a sub-request sends its first message', async () => { - const { handler, gate } = createGatedHandler() - let settled = false - const result = handler.handle(createBatchRequest({ - mode: 'streaming', - messages: [makePeerRequestMessage('0', '/slow')], - })).finally(() => { - settled = true - }) - - await sleep(10) - expect(settled).toBe(false) - - gate.resolve() - await expect(result).resolves.toMatchObject({ matched: true }) - }) - - it('starts streaming at the first message, without waiting for the slow sub-request', async () => { - const { handler, gate } = createGatedHandler() - - // Awaited before the gate opens, so holding the response for the slow sub-request would time out - const { response } = await handler.handle(createBatchRequest({ - mode: 'streaming', - messages: [makePeerRequestMessage('0', '/slow'), makePeerRequestMessage('1', '/ping')], - })) - - gate.resolve() - const received = await readAllMessages(response!) - - expect(received.map(m => m.id)).toEqual(['1', '0']) - }) - }) - }) - describe('batch request abort', () => { const started = vi.fn() const stopped = vi.fn() @@ -510,27 +465,27 @@ describe('batchHandlerPlugin', () => { const messages = [makePeerRequestMessage(0, '/wait'), makePeerRequestMessage(1, '/subscribe')] it.each([ - ['buffered', 'buffered', messages], - ['streaming', 'streaming', messages], - ['buffered (no stream sub-response)', 'buffered', [makePeerRequestMessage(0, '/wait'), makePeerRequestMessage(1, '/wait')]], - ] as const)('aborts %s sub-requests when the batch request is aborted', async (_, mode, batchMessages) => { + // The stream sub-response waits for `wait`, so its generator never starts + ['with', messages, 1], + ['without', [makePeerRequestMessage(0, '/wait'), makePeerRequestMessage(1, '/wait')], 2], + ] as const)('aborts sub-requests %s a stream sub-response when the batch request is aborted', async (_, batchMessages, running) => { const controller = new AbortController() - const result = handler.handle(createBatchRequest({ mode, messages: batchMessages, signal: controller.signal })) + const result = handler.handle(createBatchRequest({ messages: batchMessages, signal: controller.signal })) - await vi.waitFor(() => expect(started).toHaveBeenCalledTimes(2)) + await vi.waitFor(() => expect(started).toHaveBeenCalledTimes(running)) expect(stopped).not.toHaveBeenCalled() controller.abort() - await vi.waitFor(() => expect(stopped.mock.calls).toEqual([[true], [true]])) + await vi.waitFor(() => expect(stopped.mock.calls).toEqual(Array.from({ length: running }, () => [true]))) await expect(result).resolves.toMatchObject({ matched: true }) + expect(started).toHaveBeenCalledTimes(running) }) - /** - * In buffered mode, the stream sub-response makes the batch a stream too. - */ - it.each(['buffered', 'streaming'] as const)('aborts %s sub-requests when the response stream is cancelled', async (mode) => { - const { response } = await handler.handle(createBatchRequest({ mode, messages })) + it('aborts sub-requests when the response stream is cancelled', async () => { + const { response } = await handler.handle(createBatchRequest({ + messages: [makePeerRequestMessage(0, '/subscribe'), makePeerRequestMessage(1, '/subscribe')], + })) await vi.waitFor(() => expect(started).toHaveBeenCalledTimes(2)) await response!.body!.cancel() @@ -538,9 +493,8 @@ describe('batchHandlerPlugin', () => { expect(stopped.mock.calls).toEqual([[true], [true]]) }) - it.each(['buffered', 'streaming'] as const)('aborts %s sub-requests when the batch request is already aborted', async (mode) => { + it('aborts sub-requests when the batch request is already aborted', async () => { const { response } = await handler.handle(createBatchRequest({ - mode, messages: [makePeerRequestMessage(0, '/wait')], signal: AbortSignal.abort(), })) @@ -558,7 +512,6 @@ describe('batchHandlerPlugin', () => { ])) const { matched, response } = await handler.handle(createBatchRequest({ - mode: 'buffered', method: 'GET', data, })) @@ -571,7 +524,6 @@ describe('batchHandlerPlugin', () => { const handler = createHandler() const { response } = await handler.handle(createBatchRequest({ - mode: 'buffered', method: 'GET', data: 'invalid-json', })) @@ -583,7 +535,6 @@ describe('batchHandlerPlugin', () => { const handler = createHandler() const { response } = await handler.handle(createBatchRequest({ - mode: 'buffered', method: 'GET', })) @@ -596,7 +547,6 @@ describe('batchHandlerPlugin', () => { const data = encodeURIComponent(JSON.stringify({ not: 'an-array' })) const { response } = await handler.handle(createBatchRequest({ - mode: 'buffered', method: 'GET', data, })) @@ -613,7 +563,6 @@ describe('batchHandlerPlugin', () => { ])) const { response } = await handler.handle(createBatchRequest({ - mode: 'buffered', method: 'GET', data, })) @@ -630,7 +579,6 @@ describe('batchHandlerPlugin', () => { ])) const { response } = await handler.handle(createBatchRequest({ - mode: 'buffered', method: 'GET', data, })) @@ -646,7 +594,6 @@ describe('batchHandlerPlugin', () => { const handler = createHandler() const { response } = await handler.handle(createBatchRequest({ - mode: 'buffered', method: 'QUERY', messages: [ { kind: 'response', id: 0, json: { method: 'POST', url: '/ping', headers: {} } }, @@ -661,7 +608,6 @@ describe('batchHandlerPlugin', () => { const handler = createHandler() const { response } = await handler.handle(createBatchRequest({ - mode: 'buffered', method: 'QUERY', messages: [ makePeerRequestMessage(0, '/ping', 'QUERY'), @@ -693,7 +639,7 @@ describe('batchHandlerPlugin', () => { function createBatchRequestWithHeaders(headers: Record, subRequestHeaders: Record) { return new Request('https://example.com/__batch__', { method: 'POST', - headers: { 'orpc-batch': 'buffered', 'content-type': 'application/json', ...headers }, + headers: { 'orpc-batch': '1', 'content-type': 'application/json', ...headers }, body: JSON.stringify([ { kind: 'request', id: 0, json: { method: 'POST', url: '/ping', headers: subRequestHeaders }, binary: undefined }, ]), @@ -728,13 +674,177 @@ describe('batchHandlerPlugin', () => { }) }) + describe('response headers', () => { + const base = os.$context() + + const headersRouter = { + ping: base.handler(() => 'pong'), + login: base.handler(({ context }) => { + context.resHeaders?.append('set-cookie', 'session=abc; HttpOnly') + context.resHeaders?.append('set-cookie', 'theme=dark') + context.resHeaders?.set('cache-control', 'max-age=60') + context.resHeaders?.set('content-language', 'en') + context.resHeaders?.set('x-only', 'login') + return 'ok' + }), + refresh: base.handler(({ context }) => { + context.resHeaders?.append('set-cookie', 'token=xyz; HttpOnly') + context.resHeaders?.set('cache-control', 'max-age=60') + context.resHeaders?.set('content-language', 'en') + return 'ok' + }), + uncached: base.handler(({ context }) => { + context.resHeaders?.set('cache-control', 'no-store') + return 'ok' + }), + image: base.handler(({ context }) => { + context.resHeaders?.append('set-cookie', 'download=1') + context.resHeaders?.set('cache-control', 'max-age=60') + return new Blob(['__TEST__'], { type: 'image/png' }) + }), + events: base.handler(({ context }) => { + context.resHeaders?.set('cache-control', 'max-age=60') + return (async function* () { + yield 'event' + })() + }), + } + + it('moves cookies and shared headers onto the batch response', async () => { + const { response } = await createHandler(undefined, headersRouter).handle(createBatchRequest({ + messages: [makePeerRequestMessage(0, '/login'), makePeerRequestMessage(1, '/refresh')], + })) + + expect(response!.status).toBe(207) + expect(response!.headers.getSetCookie()).toEqual(['session=abc; HttpOnly', 'theme=dark', 'token=xyz; HttpOnly']) + expect(response!.headers.get('cache-control')).toBe('max-age=60') + // describes each sub-response's body, not the batch response's + expect(response!.headers.get('content-language')).toBe(null) + + const body = await response!.json() as any[] + expect(body.find(m => m.id === 0).json.headers).toEqual({ 'content-language': 'en', 'x-only': 'login' }) + expect(body.find(m => m.id === 1).json.headers).toEqual({ 'content-language': 'en' }) + }) + + it.each([ + ['has another value', '/uncached'], + ['lacks it', '/ping'], + ])('keeps a header on the sub-responses when one of them %s', async (_, url) => { + const { response } = await createHandler(undefined, headersRouter).handle(createBatchRequest({ + messages: [makePeerRequestMessage(0, '/login'), makePeerRequestMessage(1, url)], + })) + + expect(response!.headers.getSetCookie()).toEqual(['session=abc; HttpOnly', 'theme=dark']) + expect(response!.headers.get('cache-control')).toBe(null) + + const body = await response!.json() as any[] + expect(body.find(m => m.id === 0).json.headers['cache-control']).toBe('max-age=60') + }) + + it('merges cookies with, and keeps shared headers off, the ones the headers option sets', async () => { + const handler = createHandler(new BatchHandlerPlugin({ + headers: { 'set-cookie': 'batch=1', 'cache-control': 'no-cache' }, + }), headersRouter) + + const { response } = await handler.handle(createBatchRequest({ + messages: [makePeerRequestMessage(0, '/login'), makePeerRequestMessage(1, '/refresh')], + })) + + expect(response!.headers.getSetCookie()).toEqual(['batch=1', 'session=abc; HttpOnly', 'theme=dark', 'token=xyz; HttpOnly']) + expect(response!.headers.get('cache-control')).toBe('no-cache') + + const body = await response!.json() as any[] + expect(body.map(m => m.json.headers['cache-control'])).toEqual(['max-age=60', 'max-age=60']) + }) + + it('drops a sub-response header the batch response already has with the same value', async () => { + const handler = createHandler(new BatchHandlerPlugin({ headers: { 'x-only': 'login' } }), headersRouter) + + const { response } = await handler.handle(createBatchRequest({ + messages: [makePeerRequestMessage(0, '/login'), makePeerRequestMessage(1, '/refresh')], + })) + + expect(response!.headers.get('x-only')).toBe('login') + + const body = await response!.json() as any[] + expect(body.find(m => m.id === 0).json.headers).toEqual({ 'content-language': 'en' }) + }) + + it('lets mapSubresponse keep headers on each sub-response, cookies included', async () => { + const mapSubresponse = vi.fn((subresponse: StandardResponse) => subresponse) + const handler = createHandler(new BatchHandlerPlugin({ mapSubresponse }), headersRouter) + + const { response } = await handler.handle(createBatchRequest({ + messages: [makePeerRequestMessage(0, '/login'), makePeerRequestMessage(1, '/refresh')], + })) + + // The batch response still carries them + expect(response!.headers.getSetCookie()).toEqual(['session=abc; HttpOnly', 'theme=dark', 'token=xyz; HttpOnly']) + expect(response!.headers.get('cache-control')).toBe('max-age=60') + + const body = await response!.json() as any[] + expect(body.find(m => m.id === 1).json.headers).toEqual({ + 'set-cookie': 'token=xyz; HttpOnly', + 'cache-control': 'max-age=60', + 'content-language': 'en', + }) + + expect(mapSubresponse).toHaveBeenCalledTimes(2) + expect(mapSubresponse).toHaveBeenCalledWith( + expect.objectContaining({ status: 200 }), + { status: 207, headers: expect.objectContaining({ 'cache-control': 'max-age=60' }) }, + expect.objectContaining({ request: expect.objectContaining({ url: '/__batch__' }) }), + ) + }) + + it('sets each distinct cookie once, including the ones the headers option sets', async () => { + const handler = createHandler(new BatchHandlerPlugin({ headers: { 'set-cookie': 'theme=dark' } }), headersRouter) + + const { response } = await handler.handle(createBatchRequest({ + messages: [makePeerRequestMessage(0, '/login'), makePeerRequestMessage(1, '/login')], + })) + + expect(response!.headers.getSetCookie()).toEqual(['theme=dark', 'session=abc; HttpOnly']) + }) + + it('moves cookies and shared headers onto a batch response that carries binary', async () => { + const { response } = await createHandler(undefined, headersRouter).handle(createBatchRequest({ + messages: [makePeerRequestMessage('0', '/image'), makePeerRequestMessage('1', '/refresh')], + })) + + expect(response!.headers.get('content-type')).toBe('application/vnd.orpc.batch') + expect(response!.headers.getSetCookie()).toEqual(['download=1', 'token=xyz; HttpOnly']) + expect(response!.headers.get('cache-control')).toBe('max-age=60') + + const received = await readAllMessages(response!) + const image = received.find(m => m.id === '0') + expect(JSON.stringify(received)).not.toContain('set-cookie') + expect(image.json.headers['cache-control']).toBe(undefined) + expect(image.json.headers['content-type']).toBe('image/png') + expect(new TextDecoder().decode(image.binary)).toBe('__TEST__') + }) + + it('moves cookies and shared headers onto a batch response that streams', async () => { + const { response } = await createHandler(undefined, headersRouter).handle(createBatchRequest({ + messages: [makePeerRequestMessage('0', '/events'), makePeerRequestMessage('1', '/login')], + })) + + expect(response!.headers.get('content-type')).toBe('application/vnd.orpc.batch') + expect(response!.headers.getSetCookie()).toEqual(['session=abc; HttpOnly', 'theme=dark']) + expect(response!.headers.get('cache-control')).toBe('max-age=60') + + const received = await readAllMessages(response!) + expect(JSON.stringify(received)).not.toContain('set-cookie') + expect(received.filter(m => m.kind === 'response').map(m => m.json.headers?.['cache-control'])).toEqual([undefined, undefined]) + }) + }) + describe('configuration options', () => { it('supports custom successStatus', async () => { const handler = createHandler(new BatchHandlerPlugin({ successStatus: 200 })) const peerMessages = [makePeerRequestMessage(0, '/ping')] const { response } = await handler.handle(createBatchRequest({ - mode: 'buffered', messages: peerMessages, })) @@ -751,7 +861,22 @@ describe('batchHandlerPlugin', () => { vi.useRealTimers() }) - it.each(['buffered', 'streaming'] as const)('sends zero-length keep-alive frames while a %s batch stream is idle', async (mode) => { + /** + * A stream sub-response that stays idle until the batch is cancelled. + */ + const idleRouter = { + idle: os.handler(async function* ({ signal }) { + await new Promise(resolve => signal?.addEventListener('abort', resolve)) + }), + } + + const eventsRouter = { + events: os.handler(async function* () { + yield 'event' + }), + } + + it('sends zero-length keep-alive frames while the batch stream is idle', async () => { const gate = promiseWithResolvers() const handler = createHandler(new BatchHandlerPlugin({ @@ -763,7 +888,6 @@ describe('batchHandlerPlugin', () => { }) const { response } = await handler.handle(createBatchRequest({ - mode, messages: [makePeerRequestMessage(0, '/events')], })) @@ -788,11 +912,10 @@ describe('batchHandlerPlugin', () => { it('does not schedule a keep-alive timer when disabled', async () => { const handler = createHandler(new BatchHandlerPlugin({ keepAlive: { enabled: false }, - })) + }), eventsRouter) const { response } = await handler.handle(createBatchRequest({ - mode: 'streaming', - messages: [makePeerRequestMessage(0, '/ping')], + messages: [makePeerRequestMessage(0, '/events')], })) expect(response!.body).toBeInstanceOf(ReadableStream) @@ -804,21 +927,18 @@ describe('batchHandlerPlugin', () => { }) it('stops keep-alive after the stream is cancelled', async () => { - const slowRouter = createSlowRouter(new Promise(() => {})) - const handler = createHandler(new BatchHandlerPlugin({ keepAlive: { enabled: true, interval: 50 }, - }), slowRouter) + }), idleRouter) const { response } = await handler.handle(createBatchRequest({ - mode: 'streaming', - messages: [makePeerRequestMessage(0, '/ping'), makePeerRequestMessage(1, '/slow')], + messages: [makePeerRequestMessage(0, '/idle')], })) expect(vi.getTimerCount()).toBeGreaterThan(0) const reader = (response!.body as ReadableStream).getReader() - await skipMessage(reader) // the ping response + await skipMessage(reader) // the response message const first = reader.read() await vi.advanceTimersByTimeAsync(50) @@ -830,8 +950,6 @@ describe('batchHandlerPlugin', () => { }) it('clears keep-alive timer when keep-alive enqueue fails', async ({ onTestFinished }) => { - const slowRouter = createSlowRouter(new Promise(() => {})) - const originalEnqueue = ReadableStreamDefaultController.prototype.enqueue const enqueueSpy = vi.spyOn(ReadableStreamDefaultController.prototype, 'enqueue') .mockImplementation(function (this: ReadableStreamDefaultController, chunk: Uint8Array) { @@ -856,11 +974,10 @@ describe('batchHandlerPlugin', () => { const handler = createHandler(new BatchHandlerPlugin({ keepAlive: { enabled: true, interval: 50 }, - }), slowRouter) + }), idleRouter) await handler.handle(createBatchRequest({ - mode: 'streaming', - messages: [makePeerRequestMessage(0, '/ping'), makePeerRequestMessage(1, '/slow')], + messages: [makePeerRequestMessage(0, '/idle')], })) // One keep-alive setInterval is scheduled while the stream is idle. @@ -882,11 +999,10 @@ describe('batchHandlerPlugin', () => { const handler = createHandler(new BatchHandlerPlugin({ keepAlive: { enabled: false }, - })) + }), eventsRouter) const { response } = await handler.handle(createBatchRequest({ - mode: 'streaming', - messages: [makePeerRequestMessage(0, '/ping')], + messages: [makePeerRequestMessage(0, '/events')], })) const reader = (response!.body as ReadableStream).getReader() diff --git a/packages/server/src/plugins/batch.ts b/packages/server/src/plugins/batch.ts index e68d94098f..0c6a5b01b5 100644 --- a/packages/server/src/plugins/batch.ts +++ b/packages/server/src/plugins/batch.ts @@ -1,4 +1,3 @@ -import type { BatchLinkPluginMode } from '@orpc/client/plugins' import type { Promisable, Value } from '@orpc/shared' import type { StandardHeaders, StandardLazyRequest, StandardResponse } from '@standard-server/core' import type { ClientPeerSendMessage, ServerPeerSendMessage } from '@standard-server/peer' @@ -10,8 +9,8 @@ import { flattenStandardHeader, parseStandardUrl } from '@standard-server/core' import { encodePeerMessage, isClientPeerSendMessage, ServerPeer } from '@standard-server/peer' /** - * Content type for batch responses that use the length-prefixed binary framing - * (streaming mode, and buffered mode when any sub-response contains binary or is a stream). + * Content type for batch responses that use the length-prefixed binary framing, + * which they do when any sub-response contains binary or is a stream. * * Decoding is driven by the `standard-server` body hint, not this header, * so it only serves to describe the payload to logs, proxies, and dev tools. @@ -36,6 +35,46 @@ async function encodeFrame(message: ServerPeerSendMessage): Promise flattenStandardHeader(other[name]) === flattened)) { + shared[name] = value + } + } + + const headers: StandardHeaders = { ...batchHeaders, ...shared } + const cookies = new Set([batchHeaders, ...subresponseHeaders].flatMap(other => toArray(other['set-cookie']))) + + if (cookies.size > 0) { + headers['set-cookie'] = [...cookies] + } + + return headers +} + export interface BatchHandlerPluginOptions { /** * The max size of the batch allowed. @@ -52,6 +91,18 @@ export interface BatchHandlerPluginOptions { */ mapSubrequest?: (subrequest: StandardLazyRequest, batchOptions: StandardHandlerRoutingInterceptorOptions) => StandardLazyRequest + /** + * Map each sub-response in the batch before it is sent. + * + * @default removes the headers the batch response already has with the same value, which the client + * merges back, and `set-cookie`, which the browser ignores there and JavaScript could read + */ + mapSubresponse?: ( + subresponse: StandardResponse, + partialBatchResponse: Pick, + batchOptions: StandardHandlerRoutingInterceptorOptions, + ) => StandardResponse + /** * Success batch response status code. * @@ -62,16 +113,18 @@ export interface BatchHandlerPluginOptions { /** * Success batch response headers. * + * The batch response also gets the cookies its sub-responses set, and the headers they all + * share unless set here. + * * @default {} */ headers?: Value, [batchOptions: StandardHandlerRoutingInterceptorOptions]> /** - * Keep-alive settings for streaming batch responses. + * Keep-alive settings for batch responses that stream, which they do when a sub-response is a stream. * * When enabled, a zero-length length-prefixed frame is sent periodically while the * stream is idle (no message sent for `interval` ms). Clients ignore these frames. - * Only applies to streaming responses: streaming mode, or a buffered batch with a stream sub-response. * * @default { enabled: true, interval: 15000 } */ @@ -96,8 +149,8 @@ export interface BatchHandlerPluginOptions { * batch into sub-requests and streaming their responses back together. * * @remarks - * In buffered mode, the batch still streams once a sub-response is a stream - * (an event iterator or a `ReadableStream`), since it may never end. + * The batch starts the response once every sub-response is ready, so it carries the cookies sub-requests + * set and the headers every sub-response shares. It streams when a sub-response is a stream. * * **Note**: HTTP/2 and later already multiplex requests over a single connection, which often makes this plugin unnecessary. * @@ -114,6 +167,7 @@ export class BatchHandlerPlugin implements StandardHandlerPlu private readonly maxSize: Exclude['maxSize'], undefined> private readonly mapSubrequest: Exclude['mapSubrequest'], undefined> + private readonly mapSubresponse: Exclude['mapSubresponse'], undefined> private readonly successStatus: Exclude['successStatus'], undefined> private readonly headers: Exclude['headers'], undefined> private readonly keepAliveEnabled: boolean @@ -136,6 +190,18 @@ export class BatchHandlerPlugin implements StandardHandlerPlu }, })) + this.mapSubresponse = options.mapSubresponse ?? ((subresponse, { headers: batchHeaders }) => { + const headers: StandardHeaders = {} + + for (const [name, value] of Object.entries(subresponse.headers)) { + if (name !== 'set-cookie' && flattenStandardHeader(value) !== flattenStandardHeader(batchHeaders[name])) { + headers[name] = value + } + } + + return { ...subresponse, headers } + }) + this.successStatus = options.successStatus ?? 207 this.headers = options.headers ?? {} this.keepAliveEnabled = options.keepAlive?.enabled ?? true @@ -144,14 +210,10 @@ export class BatchHandlerPlugin implements StandardHandlerPlu init(options: StandardHandlerOptions): StandardHandlerOptions { const routingInterceptor: StandardHandlerRoutingInterceptor = async (interceptorOptions) => { - const batchHeader = flattenStandardHeader(interceptorOptions.request.headers['orpc-batch']) - - if (batchHeader === undefined) { + if (flattenStandardHeader(interceptorOptions.request.headers['orpc-batch']) === undefined) { return interceptorOptions.next() } - const mode: BatchLinkPluginMode = batchHeader === 'buffered' ? 'buffered' : 'streaming' - let messages: ClientPeerSendMessage[] try { @@ -230,6 +292,17 @@ export class BatchHandlerPlugin implements StandardHandlerPlu let streamController: ReadableStreamDefaultController | undefined let keepAliveTimer: ReturnType | undefined + // Every subresponse waits here until all of them are ready, so the batch response carries their headers + const ready = messages.map(() => promiseWithResolvers()) + const lifting = Promise.all(ready.map(({ promise }) => promise)).then((all) => { + const responses = all.filter(response => response !== undefined) + + return { + batchHeaders: liftSubresponseHeaders(headers, responses.map(response => response.headers)), + streams: responses.some(({ body }) => body instanceof ReadableStream || isAsyncIteratorObject(body)), + } + }) + const clearKeepAlive = () => { if (keepAliveTimer !== undefined) { clearInterval(keepAliveTimer) @@ -263,10 +336,6 @@ export class BatchHandlerPlugin implements StandardHandlerPlu } const peer = new ServerPeer(async (message) => { - if (mode === 'streaming') { - startStreaming() - } - if (streamController === undefined) { bufferedMessages.push(message) } @@ -275,67 +344,50 @@ export class BatchHandlerPlugin implements StandardHandlerPlu } }) - const streaming = promiseWithResolvers<{ body: ReadableStream>, flushed: Promise }>() - /** - * Opens the binary length-prefixed stream that every later message is written to, - * starting with the messages buffered so far, which `flushed` waits for. - * - * A function declaration, since it and the peer above refer to each other. + * Errors should not occur at the routing interceptor level. + * Reject the promise so it can be handled by the unhandledRejection handler + * for global logging or error handling. */ - function startStreaming() { - if (streamController !== undefined) { - return - } + const internalError = (err: unknown): StandardResponse => { + Promise.reject(err) - const body = new ReadableStream>({ - start(controller) { - streamController = controller - scheduleKeepAlive() - }, - async cancel(reason) { - clearKeepAlive() - await peer.close(reason) - }, - }) - - streaming.resolve({ body, flushed: Promise.all(bufferedMessages.splice(0).map(enqueueMessage)) }) + return { status: 500, headers: {}, body: 'Internal server error' } } - const handleIndividualRequest = async (request: StandardLazyRequest): Promise => { + const handleIndividualRequest = async ( + request: StandardLazyRequest, + markReady: (response: StandardResponse) => void, + ): Promise => { + let response: StandardResponse + try { request = this.mapSubrequest(request, interceptorOptions) - const { matched, response } = await interceptorOptions.next({ ...interceptorOptions, request }) - - if (!matched) { - return { status: 404, headers: {}, body: 'No procedure matched' } - } + const result = await interceptorOptions.next({ ...interceptorOptions, request }) + response = result.matched ? result.response : { status: 404, headers: {}, body: 'No procedure matched' } + } + catch (err) { + response = internalError(err) + } - /** - * A stream is meant to be read as it arrives and might never end, so buffering it would - * hold back the whole batch. Start streaming before the peer sends its first message, - * so none of that subresponse is buffered. - */ - if (response.body instanceof ReadableStream || isAsyncIteratorObject(response.body)) { - startStreaming() - } + markReady(response) + const { batchHeaders } = await lifting - return response + try { + return this.mapSubresponse(response, { status, headers: batchHeaders }, interceptorOptions) } catch (err) { - /** - * Errors should not occur at the routing interceptor level. - * Reject the promise so it can be handled by the unhandledRejection handler - * for global logging or error handling. - */ - Promise.reject(err) - - return { status: 500, headers: {}, body: 'Internal server error' } + return internalError(err) } } const runSubrequests = async (): Promise => { - const promise = Promise.all(messages.map(msg => peer.message(msg, handleIndividualRequest))) + const promise = Promise.all(messages.map((msg, i) => { + const { resolve } = ready[i]! + + // Not every message reaches the handler, such as a cancel or a repeated id + return peer.message(msg, request => handleIndividualRequest(request, resolve)).finally(() => resolve(undefined)) + })) const signal = interceptorOptions.request.signal const closePeer = () => peer.close(signal?.reason) @@ -354,11 +406,10 @@ export class BatchHandlerPlugin implements StandardHandlerPlu } const subrequests = runSubrequests() + const { batchHeaders, streams } = await lifting - // Undefined when every subresponse finished before any turned out to be a stream - const started = await Promise.race([streaming.promise, subrequests.then(() => undefined)]) - - if (started === undefined) { + if (!streams) { + await subrequests await peer.close() if (bufferedMessages.some(msg => msg.binary !== undefined)) { @@ -368,7 +419,7 @@ export class BatchHandlerPlugin implements StandardHandlerPlu matched: true, response: { status, - headers, + headers: batchHeaders, body: new Blob(chunks, { type: BATCH_CONTENT_TYPE }), }, } @@ -376,12 +427,28 @@ export class BatchHandlerPlugin implements StandardHandlerPlu return { matched: true, - response: { status, headers, body: bufferedMessages }, + response: { status, headers: batchHeaders, body: bufferedMessages }, } } + /** + * A stream is meant to be read as it arrives and might never end, so the batch answers with + * the length-prefixed stream that every message is written to, starting with those buffered so far. + */ + const body = new ReadableStream>({ + start(controller) { + streamController = controller + scheduleKeepAlive() + }, + async cancel(reason) { + clearKeepAlive() + await peer.close(reason) + }, + }) + const flushed = Promise.all(bufferedMessages.splice(0).map(enqueueMessage)) + // DO NOT await here to block streaming response - Promise.all([started.flushed, subrequests]) + Promise.all([flushed, subrequests]) .then(async () => { clearKeepAlive() streamController!.close() @@ -397,8 +464,8 @@ export class BatchHandlerPlugin implements StandardHandlerPlu matched: true, response: { status, - headers: { ...headers, 'content-type': BATCH_CONTENT_TYPE }, - body: started.body, + headers: { ...batchHeaders, 'content-type': BATCH_CONTENT_TYPE }, + body, }, } } diff --git a/tests/batch/__shared__/client-server.compression-hono-fetch.ts b/tests/batch/__shared__/client-server.compression-hono-fetch.ts index 2b2ce7dca5..57d26ae7a6 100644 --- a/tests/batch/__shared__/client-server.compression-hono-fetch.ts +++ b/tests/batch/__shared__/client-server.compression-hono-fetch.ts @@ -18,7 +18,6 @@ export const createCompressionHonoFetchBatchClientServerTest: CreateBatchClientS { context = defaultBatchClientServerOptions.context, method = 'GET', - mode = defaultBatchClientServerOptions.mode, serializer = defaultBatchClientServerOptions.serializer, } = {}, ) => { @@ -69,7 +68,7 @@ export const createCompressionHonoFetchBatchClientServerTest: CreateBatchClientS serializer, fetch: fetchSpy, plugins: [ - new BatchLinkPlugin({ groups: [defaultBatchGroup], mode }), + new BatchLinkPlugin({ groups: [defaultBatchGroup] }), // fetch already automatically decompresses response // new ResponseCompressionLinkPlugin(), ], diff --git a/tests/batch/__shared__/client-server.compression-node-http.ts b/tests/batch/__shared__/client-server.compression-node-http.ts index f9bc25ef46..e33f2f504c 100644 --- a/tests/batch/__shared__/client-server.compression-node-http.ts +++ b/tests/batch/__shared__/client-server.compression-node-http.ts @@ -11,15 +11,14 @@ import { defaultBatchClientServerOptions, defaultBatchGroup } from './client-ser /** * The same suite as the plain node-http one, with batch responses compressed on the wire. - * Its streaming test is what proves the compressor flushes: a buffering one would hold every - * subresponse until the slowest of them resolved, which that test only allows after reading. + * Its stream test is what proves the compressor flushes: a buffering one would hold every + * stream message until the stream ended, which that test only allows after reading. */ export const createCompressionNodeHttpBatchClientServerTest: CreateBatchClientServerTest = ( router, { context = defaultBatchClientServerOptions.context, method = 'POST', - mode = defaultBatchClientServerOptions.mode, serializer = defaultBatchClientServerOptions.serializer, } = {}, ) => { @@ -65,7 +64,7 @@ export const createCompressionNodeHttpBatchClientServerTest: CreateBatchClientSe serializer, fetch: fetchSpy, // fetch already automatically decompresses the response - plugins: [new BatchLinkPlugin({ groups: [defaultBatchGroup], mode })], + plugins: [new BatchLinkPlugin({ groups: [defaultBatchGroup] })], }) return { diff --git a/tests/batch/__shared__/client-server.hono-fetch.ts b/tests/batch/__shared__/client-server.hono-fetch.ts index c311c858a4..c657df3007 100644 --- a/tests/batch/__shared__/client-server.hono-fetch.ts +++ b/tests/batch/__shared__/client-server.hono-fetch.ts @@ -13,7 +13,6 @@ export const createHonoFetchBatchClientServerTest: CreateBatchClientServerTest = { context = defaultBatchClientServerOptions.context, method = 'GET', - mode = defaultBatchClientServerOptions.mode, serializer = defaultBatchClientServerOptions.serializer, } = {}, ) => { @@ -48,7 +47,7 @@ export const createHonoFetchBatchClientServerTest: CreateBatchClientServerTest = origin: `http://localhost:${addressInfo.port}`, serializer, fetch: fetchSpy, - plugins: [new BatchLinkPlugin({ groups: [defaultBatchGroup], mode })], + plugins: [new BatchLinkPlugin({ groups: [defaultBatchGroup] })], }) return { diff --git a/tests/batch/__shared__/client-server.node-http.ts b/tests/batch/__shared__/client-server.node-http.ts index 429a1eb08c..316966b91f 100644 --- a/tests/batch/__shared__/client-server.node-http.ts +++ b/tests/batch/__shared__/client-server.node-http.ts @@ -13,7 +13,6 @@ export const createNodeHttpBatchClientServerTest: CreateBatchClientServerTest = { context = defaultBatchClientServerOptions.context, method = 'POST', - mode = defaultBatchClientServerOptions.mode, serializer = defaultBatchClientServerOptions.serializer, } = {}, ) => { @@ -45,7 +44,7 @@ export const createNodeHttpBatchClientServerTest: CreateBatchClientServerTest = method, serializer, fetch: fetchSpy, - plugins: [new BatchLinkPlugin({ groups: [defaultBatchGroup], mode })], + plugins: [new BatchLinkPlugin({ groups: [defaultBatchGroup] })], }) return { diff --git a/tests/batch/__shared__/client-server.ts b/tests/batch/__shared__/client-server.ts index 7410cad621..ef717ea691 100644 --- a/tests/batch/__shared__/client-server.ts +++ b/tests/batch/__shared__/client-server.ts @@ -1,5 +1,4 @@ import type { RPCSerializer } from '@orpc/client' -import type { BatchLinkPluginMode } from '@orpc/client/plugins' import type { AnyRouter, Context, RouterClient } from '@orpc/server' import type { Public } from '@orpc/shared' import { defaultSerializer } from '../../rpc/__shared__/client-server' @@ -7,7 +6,6 @@ import { defaultSerializer } from '../../rpc/__shared__/client-server' export interface BatchClientServerTestOptions { context?: Context method?: 'GET' | 'POST' | 'QUERY' - mode?: BatchLinkPluginMode serializer?: Public } @@ -22,7 +20,6 @@ export interface CreateBatchClientServerTest { export const defaultBatchClientServerOptions = { context: {}, - mode: 'streaming' as BatchLinkPluginMode, serializer: defaultSerializer, } diff --git a/tests/batch/batch-plugin.test.ts b/tests/batch/batch-plugin.test.ts index 6f1d64d3c1..c111871f03 100644 --- a/tests/batch/batch-plugin.test.ts +++ b/tests/batch/batch-plugin.test.ts @@ -47,15 +47,15 @@ describe.each([ }) /** - * A buffered batch with no binary subresponse is a plain json array rather than the binary + * A batch with no binary subresponse is a plain json array rather than the binary * framing every other case here produces, so it travels a different encoding path. */ - it('round-trips a buffered batch of json-only responses', async () => { + it('round-trips a batch of json-only responses', async () => { const router = { echo: os.input(z.string()).handler(({ input }) => `echo:${input}`), } - const { client, fetchSpy } = createClientServer(router, { mode: 'buffered' }) + const { client, fetchSpy } = createClientServer(router) await Promise.all([ expect(client.echo('alpha')).resolves.toBe('echo:alpha'), @@ -66,16 +66,16 @@ describe.each([ }) /** - * Without a stream subresponse, a buffered batch is the binary framing collected into a blob + * Without a stream subresponse, a batch is the binary framing collected into a blob * once every subresponse is ready, rather than a stream. */ - it('round-trips a buffered batch of blob and json responses', async () => { + it('round-trips a batch of blob and json responses', async () => { const router = { - blob: os.handler(() => new Blob(['buffered payload'], { type: 'application/octet-stream' })), - json: os.handler(() => ({ ok: 'buffered-mixed' })), + blob: os.handler(() => new Blob(['blob payload'], { type: 'application/octet-stream' })), + json: os.handler(() => ({ ok: 'mixed' })), } - const { client, fetchSpy } = createClientServer(router, { mode: 'buffered' }) + const { client, fetchSpy } = createClientServer(router) const [file, info] = await Promise.all([ client.blob(), @@ -83,15 +83,15 @@ describe.each([ ]) expect(file).toBeInstanceOf(Blob) - await expect((file as Blob).text()).resolves.toBe('buffered payload') - expect(info).toEqual({ ok: 'buffered-mixed' }) + await expect((file as Blob).text()).resolves.toBe('blob payload') + expect(info).toEqual({ ok: 'mixed' }) expect(fetchSpy).toHaveBeenCalledTimes(1) // ensure batch was used }) /** - * A stream may never end, so a buffered batch starts streaming once a subresponse is one. + * A stream may never end, so a batch streams once every subresponse is ready and one of them is a stream. */ - it.each(['streaming', 'buffered'] as const)('supports readable stream, blob, json, and AsyncIteratorObject responses in %s mode', async (mode) => { + it('supports readable stream, blob, json, and AsyncIteratorObject responses', async () => { const part2 = promiseWithResolvers() const event2 = promiseWithResolvers() @@ -104,9 +104,9 @@ describe.each([ }, })) - const blobProcedure = vi.fn(async () => new Blob([`${mode} mixed payload`], { type: 'application/octet-stream' })) + const blobProcedure = vi.fn(async () => new Blob(['blob payload'], { type: 'application/octet-stream' })) - const jsonProcedure = vi.fn(async () => ({ ok: `${mode}-mixed` })) + const jsonProcedure = vi.fn(async () => ({ ok: 'mixed' })) const eventProcedure = vi.fn(async function* () { yield 'event 1' @@ -122,7 +122,7 @@ describe.each([ events: os.handler(eventProcedure), } - const { client, fetchSpy } = createClientServer(router, { mode }) + const { client, fetchSpy } = createClientServer(router) // Both streams hold everything after their first item until the test releases it, so a batch // that waited for every subresponse to finish would hang here instead of resolving. @@ -134,8 +134,8 @@ describe.each([ ]) expect(file).toBeInstanceOf(Blob) - await expect((file as Blob).text()).resolves.toBe(`${mode} mixed payload`) - expect(info).toEqual({ ok: `${mode}-mixed` }) + await expect((file as Blob).text()).resolves.toBe('blob payload') + expect(info).toEqual({ ok: 'mixed' }) const reader = stream.getReader() @@ -211,7 +211,7 @@ describe.each([ }), } - const { client, fetchSpy } = createClientServer(router, { mode: 'streaming' }) + const { client, fetchSpy } = createClientServer(router) const [info, file, stream, events, endless] = await Promise.all([ client.json(), diff --git a/tests/plugins/all-plugins.test.ts b/tests/plugins/all-plugins.test.ts index 132dfbb0d2..ac31bbe793 100644 --- a/tests/plugins/all-plugins.test.ts +++ b/tests/plugins/all-plugins.test.ts @@ -56,6 +56,9 @@ const contract = { ping: oc .input(z.object({ value: z.number() })) .output(z.object({ value: z.number() })), + login: oc + .input(z.object({ user: z.string() })) + .output(z.object({ user: z.string() })), } const spec: OpenAPIDocument<'3.2.0'> = { @@ -67,6 +70,11 @@ const implementer = implement(contract).$context() const router = implementer.router({ ping: implementer.ping.handler(({ input }) => input), + login: implementer.login.handler(({ input, context }) => { + context.resHeaders?.append('set-cookie', `session=${input.user}; HttpOnly`) + context.resHeaders?.set('cache-control', 'no-store') + return input + }), }) function createHandlerPlugins() { @@ -162,13 +170,32 @@ describe('all plugins on one handler and one link', () => { ]) expect(exchanges).toHaveLength(1) // both calls travelled in a single batch request - expect(exchanges[0]!.request.headers.get('orpc-batch')).toEqual('streaming') + expect(exchanges[0]!.request.headers.get('orpc-batch')).toEqual('1') // the batch request is compressed as a whole, and decompressed before it is split expect(exchanges[0]!.request.headers.get('content-encoding')).toEqual('gzip') // the batch response is compressed as a whole, and decompressed before it is decoded expect(exchanges[0]!.response.headers.get('content-encoding')).toEqual('gzip') }) + it('sets the cookies and shared headers of batched calls on the batch response', async () => { + const { client, exchanges } = createClientServer() + + await expect(Promise.all([ + client.login({ user: 'alice' }), + client.login({ user: 'bob' }), + ])).resolves.toEqual([ + { user: 'alice' }, + { user: 'bob' }, + ]) + + expect(exchanges).toHaveLength(1) + expect(exchanges[0]!.response.headers.getSetCookie().sort()).toEqual([ + 'session=alice; HttpOnly', + 'session=bob; HttpOnly', + ]) + expect(exchanges[0]!.response.headers.get('cache-control')).toEqual('no-store') + }) + it('compresses a standalone call in both directions through every plugin', async () => { const { client, exchanges } = createClientServer() From ec3261628722be51f0721cd2e789fe7fc654c472 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 19 Oct 2026 03:14:22 +0000 Subject: [PATCH 2/3] refactor(server): free batched bodies once sent and let the headers option opt out Each call now hands the batch only its headers and whether it streams, so a large result is freed once it is sent instead of living until the longest stream in the batch ends. The headers option wins over the headers the calls share by spread order, so setting one to undefined keeps it off the batch response. internalError moves to module scope, and tests and comments left over from streaming mode are removed. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UtE8iZthQGzPQQoMXCSgzc --- packages/client/src/plugins/batch.test.ts | 2 - ...esponse-compression-handler-plugin.test.ts | 4 +- packages/server/src/plugins/batch.test.ts | 116 +++++------------- packages/server/src/plugins/batch.ts | 64 ++++++---- .../get-method-csrf-protection.test.ts | 2 +- tests/batch/batch-plugin.test.ts | 19 --- 6 files changed, 75 insertions(+), 132 deletions(-) diff --git a/packages/client/src/plugins/batch.test.ts b/packages/client/src/plugins/batch.test.ts index 778f655813..4a3f6e5568 100644 --- a/packages/client/src/plugins/batch.test.ts +++ b/packages/client/src/plugins/batch.test.ts @@ -549,14 +549,12 @@ describe('batchLinkPlugin', () => { const sentGetRequest = vi.mocked(transport.send).mock.calls.find(([request]) => request.method === 'GET')![0] expect(extractBatchMessagesFromRequest(sentGetRequest).map(message => message.json.method)).toEqual(['GET', 'GET']) - expect(sentGetRequest.headers['orpc-batch']).toBe('1') const sentQueryRequest = vi.mocked(transport.send).mock.calls.find(([request]) => request.method === 'QUERY')![0] expect(extractBatchMessagesFromRequest(sentQueryRequest).map(message => message.json.method)).toEqual(['QUERY', 'QUERY']) const sentPostRequest = vi.mocked(transport.send).mock.calls.find(([request]) => request.method === 'POST')![0] expect(extractBatchMessagesFromRequest(sentPostRequest).map(message => message.json.method)).toEqual(['PUT', 'PUT']) - expect(sentPostRequest.headers['orpc-batch']).toBe('1') }) it('aborts grouped batch request when all sub-requests are aborted', async () => { diff --git a/packages/node/src/batch-response-compression-handler-plugin.test.ts b/packages/node/src/batch-response-compression-handler-plugin.test.ts index 18a9b42d6b..20cdd82a05 100644 --- a/packages/node/src/batch-response-compression-handler-plugin.test.ts +++ b/packages/node/src/batch-response-compression-handler-plugin.test.ts @@ -76,7 +76,7 @@ describe('batchResponseCompressionHandlerPlugin', () => { * The reason this plugin exists: `CompressionStream` cannot flush, so every message of a * batch that streams would sit in the compressor until the stream ended. */ - it('delivers each message as soon as it is produced, without waiting for the whole batch', async () => { + it('delivers each message as soon as it is produced, without waiting for the stream to end', async () => { const slow = promiseWithResolvers() const handler = createHandler({}, { @@ -94,7 +94,7 @@ describe('batchResponseCompressionHandlerPlugin', () => { const readUntil = async (text: string) => { while (!received.includes(text)) { const { done, value } = await reader.read() - // A buffering compressor ends the stream (or stalls) before the fast message arrives + // A buffering compressor ends the stream (or stalls) before the first message arrives expect(done).toBe(false) received += decoder.decode(value, { stream: true }) } diff --git a/packages/server/src/plugins/batch.test.ts b/packages/server/src/plugins/batch.test.ts index eb64a414eb..36b4a5625f 100644 --- a/packages/server/src/plugins/batch.test.ts +++ b/packages/server/src/plugins/batch.test.ts @@ -88,17 +88,22 @@ async function skipMessage(reader: ReadableStreamDefaultReader) { await reader.read() } +function waitForAbort(signal: AbortSignal) { + return new Promise((resolve) => { + if (signal.aborted) { + resolve(undefined) + } + signal.addEventListener('abort', resolve) + }) +} + /** - * A router with a `ping` procedure that answers right away and a `slow` one that waits for `gate`. + * A router with an `events` procedure that returns a stream of one event. */ -function createSlowRouter(gate: Promise) { - return { - ping: os.handler(() => 'pong'), - slow: os.handler(async () => { - await gate - return 'slow' - }), - } +const eventsRouter = { + events: os.handler(async function* () { + yield 'event' + }), } describe('batchHandlerPlugin', () => { @@ -351,15 +356,16 @@ describe('batchHandlerPlugin', () => { it('waits for every sub-response before it streams', async () => { const gate = promiseWithResolvers() const handler = createHandler(new BatchHandlerPlugin(), { - ...createSlowRouter(gate.promise), - events: os.handler(async function* () { - yield 'event' + ...eventsRouter, + file: os.handler(async () => { + await gate.promise + return new Blob(['__FILE__'], { type: 'application/octet-stream' }) }), }) let settled = false const result = handler.handle(createBatchRequest({ - messages: [makePeerRequestMessage('0', '/events'), makePeerRequestMessage('1', '/slow')], + messages: [makePeerRequestMessage('0', '/events'), makePeerRequestMessage('1', '/file')], })).finally(() => { settled = true }) @@ -373,59 +379,7 @@ describe('batchHandlerPlugin', () => { const messages = await readAllMessages(response!) expect(messages.filter(m => m.kind === 'response').map(m => m.id).sort()).toEqual(['0', '1']) - }) - - it('writes out the sub-responses buffered before it, binary included', async () => { - const gate = promiseWithResolvers() - const ping = vi.fn(() => 'pong') - const file = vi.fn(() => new Blob(['__FILE__'], { type: 'application/octet-stream' })) - - const handler = createHandler(new BatchHandlerPlugin(), { - ping: os.handler(ping), - file: os.handler(file), - events: os.handler(async () => { - await gate.promise - - return (async function* () { - yield 'event' - })() - }), - }) - - const result = handler.handle(createBatchRequest({ - messages: [ - makePeerRequestMessage('0', '/ping'), - makePeerRequestMessage('1', '/file'), - makePeerRequestMessage('2', '/events'), - ], - })) - - await vi.waitFor(() => { - expect(ping).toHaveBeenCalledTimes(1) - expect(file).toHaveBeenCalledTimes(1) - }) - await sleep(10) // let both responses reach the buffer - - gate.resolve() - const { response } = await result - expect(response!.headers.get('standard-server')).toEqual('octet-stream') - - // Only messages of the same sub-request keep their order - const messages = await readAllMessages(response!) - const fileMessages = messages.filter(m => m.id === '1') - - expect(messages.filter(m => m.id === '0')).toEqual([ - expect.objectContaining({ kind: 'response' }), - ]) - expect(fileMessages).toEqual([ - expect.objectContaining({ kind: 'response' }), - ]) - expect(new TextDecoder().decode(fileMessages[0].binary)).toBe('__FILE__') - expect(messages.filter(m => m.id === '2')).toEqual([ - expect.objectContaining({ kind: 'response' }), - expect.objectContaining({ kind: 'event-stream' }), - expect.objectContaining({ kind: 'event-stream', json: expect.objectContaining({ event: 'close' }) }), - ]) + expect(new TextDecoder().decode(messages.find(m => m.id === '1').binary)).toBe('__FILE__') }) }) }) @@ -434,15 +388,6 @@ describe('batchHandlerPlugin', () => { const started = vi.fn() const stopped = vi.fn() - function waitForAbort(signal: AbortSignal) { - return new Promise((resolve) => { - if (signal.aborted) { - resolve(undefined) - } - signal.addEventListener('abort', resolve) - }) - } - const handler = createHandler(new BatchHandlerPlugin(), { wait: os.handler(async ({ signal }) => { started() @@ -797,6 +742,19 @@ describe('batchHandlerPlugin', () => { ) }) + it('keeps a shared header off the batch response when the headers option sets it to undefined', async () => { + const handler = createHandler(new BatchHandlerPlugin({ headers: { 'cache-control': undefined } }), headersRouter) + + const { response } = await handler.handle(createBatchRequest({ + messages: [makePeerRequestMessage(0, '/login'), makePeerRequestMessage(1, '/refresh')], + })) + + expect(response!.headers.get('cache-control')).toBe(null) + + const body = await response!.json() as any[] + expect(body.map(m => m.json.headers['cache-control'])).toEqual(['max-age=60', 'max-age=60']) + }) + it('sets each distinct cookie once, including the ones the headers option sets', async () => { const handler = createHandler(new BatchHandlerPlugin({ headers: { 'set-cookie': 'theme=dark' } }), headersRouter) @@ -866,13 +824,7 @@ describe('batchHandlerPlugin', () => { */ const idleRouter = { idle: os.handler(async function* ({ signal }) { - await new Promise(resolve => signal?.addEventListener('abort', resolve)) - }), - } - - const eventsRouter = { - events: os.handler(async function* () { - yield 'event' + await waitForAbort(signal!) }), } diff --git a/packages/server/src/plugins/batch.ts b/packages/server/src/plugins/batch.ts index 0c6a5b01b5..3ff32968a9 100644 --- a/packages/server/src/plugins/batch.ts +++ b/packages/server/src/plugins/batch.ts @@ -43,8 +43,28 @@ function isBodyHeader(name: string): boolean { } /** - * Builds the batch response headers: the subresponses' distinct cookies, and the headers they all - * have with the same value, as the client does with the headers every subrequest shares. + * Errors should not occur at the routing interceptor level. + * Reject the promise so it can be handled by the unhandledRejection handler + * for global logging or error handling. + */ +function internalError(err: unknown): StandardResponse { + Promise.reject(err) + + return { status: 500, headers: {}, body: 'Internal server error' } +} + +/** + * What the batch needs from a subresponse before it answers, without keeping its body alive. + */ +interface SubresponseHead { + headers: StandardHeaders + stream: boolean +} + +/** + * Builds the batch response headers: the headers the subresponses all have with the same value, as + * the client does with the headers every subrequest shares, then the batch headers, which win, and + * the distinct cookies of both. */ function liftSubresponseHeaders( batchHeaders: StandardHeaders, @@ -54,7 +74,7 @@ function liftSubresponseHeaders( const shared: StandardHeaders = {} for (const [name, value] of Object.entries(first)) { - if (value === undefined || name === 'set-cookie' || isBodyHeader(name) || batchHeaders[name] !== undefined) { + if (value === undefined || isBodyHeader(name)) { continue } @@ -65,7 +85,7 @@ function liftSubresponseHeaders( } } - const headers: StandardHeaders = { ...batchHeaders, ...shared } + const headers: StandardHeaders = { ...shared, ...batchHeaders } const cookies = new Set([batchHeaders, ...subresponseHeaders].flatMap(other => toArray(other['set-cookie']))) if (cookies.size > 0) { @@ -113,15 +133,15 @@ export interface BatchHandlerPluginOptions { /** * Success batch response headers. * - * The batch response also gets the cookies its sub-responses set, and the headers they all - * share unless set here. + * The batch response also gets the cookies its sub-responses set, and the headers they all share. + * Headers set here win, and setting one to `undefined` keeps it off. * * @default {} */ headers?: Value, [batchOptions: StandardHandlerRoutingInterceptorOptions]> /** - * Keep-alive settings for batch responses that stream, which they do when a sub-response is a stream. + * Keep-alive settings for batch responses that stream. * * When enabled, a zero-length length-prefixed frame is sent periodically while the * stream is idle (no message sent for `interval` ms). Clients ignore these frames. @@ -146,7 +166,7 @@ export interface BatchHandlerPluginOptions { /** * Handles batch requests sent by the client Batch Link Plugin, splitting each - * batch into sub-requests and streaming their responses back together. + * batch into sub-requests and sending their responses back together. * * @remarks * The batch starts the response once every sub-response is ready, so it carries the cookies sub-requests @@ -287,19 +307,19 @@ export class BatchHandlerPlugin implements StandardHandlerPlu const status = await value(this.successStatus, interceptorOptions) const headers = await value(this.headers, interceptorOptions) - // Response messages are held here until the batch response starts streaming + // Messages the peer sends before the response streams, or the whole body when it does not const bufferedMessages: ServerPeerSendMessage[] = [] let streamController: ReadableStreamDefaultController | undefined let keepAliveTimer: ReturnType | undefined // Every subresponse waits here until all of them are ready, so the batch response carries their headers - const ready = messages.map(() => promiseWithResolvers()) + const ready = messages.map(() => promiseWithResolvers()) const lifting = Promise.all(ready.map(({ promise }) => promise)).then((all) => { - const responses = all.filter(response => response !== undefined) + const heads = all.filter(head => head !== undefined) return { - batchHeaders: liftSubresponseHeaders(headers, responses.map(response => response.headers)), - streams: responses.some(({ body }) => body instanceof ReadableStream || isAsyncIteratorObject(body)), + batchHeaders: liftSubresponseHeaders(headers, heads.map(head => head.headers)), + streams: heads.some(head => head.stream), } }) @@ -344,20 +364,9 @@ export class BatchHandlerPlugin implements StandardHandlerPlu } }) - /** - * Errors should not occur at the routing interceptor level. - * Reject the promise so it can be handled by the unhandledRejection handler - * for global logging or error handling. - */ - const internalError = (err: unknown): StandardResponse => { - Promise.reject(err) - - return { status: 500, headers: {}, body: 'Internal server error' } - } - const handleIndividualRequest = async ( request: StandardLazyRequest, - markReady: (response: StandardResponse) => void, + markReady: (head: SubresponseHead) => void, ): Promise => { let response: StandardResponse @@ -370,7 +379,10 @@ export class BatchHandlerPlugin implements StandardHandlerPlu response = internalError(err) } - markReady(response) + markReady({ + headers: response.headers, + stream: response.body instanceof ReadableStream || isAsyncIteratorObject(response.body), + }) const { batchHeaders } = await lifting try { diff --git a/packages/server/src/plugins/get-method-csrf-protection.test.ts b/packages/server/src/plugins/get-method-csrf-protection.test.ts index 5d7412a505..d9ccde8745 100644 --- a/packages/server/src/plugins/get-method-csrf-protection.test.ts +++ b/packages/server/src/plugins/get-method-csrf-protection.test.ts @@ -222,7 +222,7 @@ describe('getMethodCsrfProtectionHandlerPlugin', () => { }])) return handler.handle(new Request(`https://api.example.com/__batch__?data=${data}`, { - headers: { 'orpc-batch': 'buffered', ...transportHeaders }, + headers: { 'orpc-batch': '1', ...transportHeaders }, })) } diff --git a/tests/batch/batch-plugin.test.ts b/tests/batch/batch-plugin.test.ts index c111871f03..87920e82f4 100644 --- a/tests/batch/batch-plugin.test.ts +++ b/tests/batch/batch-plugin.test.ts @@ -46,25 +46,6 @@ describe.each([ expect(fetchSpy).toHaveBeenCalledTimes(1) // ensure batch was used }) - /** - * A batch with no binary subresponse is a plain json array rather than the binary - * framing every other case here produces, so it travels a different encoding path. - */ - it('round-trips a batch of json-only responses', async () => { - const router = { - echo: os.input(z.string()).handler(({ input }) => `echo:${input}`), - } - - const { client, fetchSpy } = createClientServer(router) - - await Promise.all([ - expect(client.echo('alpha')).resolves.toBe('echo:alpha'), - expect(client.echo('beta')).resolves.toBe('echo:beta'), - ]) - - expect(fetchSpy).toHaveBeenCalledTimes(1) // ensure batch was used - }) - /** * Without a stream subresponse, a batch is the binary framing collected into a blob * once every subresponse is ready, rather than a stream. From 98a07928f3aa7ac80fbab2a157d82183e0829f91 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 19 Oct 2026 03:50:45 +0000 Subject: [PATCH 3/3] fix(server): merge every sub-response's vary fields onto the batch response A cache-control every call shares moves onto the batch response, but a Vary only some calls set stayed inside the body, so a shared cache could reuse the batch across languages or, with Vary: Cookie, across sessions. The batch response now varies on every field any call or the headers option varies on, and on everything when one varies on `*`. The headers option docs now say set-cookie and vary merge instead of being replaced. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UtE8iZthQGzPQQoMXCSgzc --- apps/content/docs/plugins/batch.mdx | 2 +- packages/server/src/plugins/batch.test.ts | 49 +++++++++++++++++++++++ packages/server/src/plugins/batch.ts | 21 +++++++++- 3 files changed, 69 insertions(+), 3 deletions(-) diff --git a/apps/content/docs/plugins/batch.mdx b/apps/content/docs/plugins/batch.mdx index df721edfae..2ae02a61af 100644 --- a/apps/content/docs/plugins/batch.mdx +++ b/apps/content/docs/plugins/batch.mdx @@ -221,7 +221,7 @@ Per-request data passed through the link's [`headers` option](/docs/rpc/link#url ## Response Headers -The batch response carries the cookies of its subresponses, and the headers they all share. Other headers stay on their subresponse. To change what stays on each subresponse, set the server plugin's `mapSubresponse` option. +The batch response carries the cookies and `Vary` fields of its subresponses, and the headers they all share. Other headers stay on their subresponse. To change what stays on each subresponse, set the server plugin's `mapSubresponse` option. ## Learn More diff --git a/packages/server/src/plugins/batch.test.ts b/packages/server/src/plugins/batch.test.ts index 36b4a5625f..49ea27ebfe 100644 --- a/packages/server/src/plugins/batch.test.ts +++ b/packages/server/src/plugins/batch.test.ts @@ -755,6 +755,55 @@ describe('batchHandlerPlugin', () => { expect(body.map(m => m.json.headers['cache-control'])).toEqual(['max-age=60', 'max-age=60']) }) + it('merges every sub-response\'s vary fields onto the batch response', async () => { + const handler = createHandler(new BatchHandlerPlugin({ headers: { vary: 'Origin' } }), { + localized: os.$context().handler(({ context }) => { + context.resHeaders?.set('cache-control', 'public, max-age=60') + context.resHeaders?.set('vary', 'Accept-Language, origin') + return 'ok' + }), + personal: os.$context().handler(({ context }) => { + context.resHeaders?.set('cache-control', 'public, max-age=60') + context.resHeaders?.set('vary', 'Cookie') + return 'ok' + }), + shared: os.$context().handler(({ context }) => { + context.resHeaders?.set('cache-control', 'public, max-age=60') + return 'ok' + }), + }) + + const { response } = await handler.handle(createBatchRequest({ + messages: [ + makePeerRequestMessage(0, '/localized'), + makePeerRequestMessage(1, '/personal'), + makePeerRequestMessage(2, '/shared'), + ], + })) + + expect(response!.headers.get('cache-control')).toBe('public, max-age=60') + expect(response!.headers.get('vary')).toBe('Origin, Accept-Language, Cookie') + }) + + it('varies the batch response on everything when a sub-response varies on everything', async () => { + const handler = createHandler(new BatchHandlerPlugin(), { + any: os.$context().handler(({ context }) => { + context.resHeaders?.set('vary', '*') + return 'ok' + }), + localized: os.$context().handler(({ context }) => { + context.resHeaders?.set('vary', 'Accept-Language') + return 'ok' + }), + }) + + const { response } = await handler.handle(createBatchRequest({ + messages: [makePeerRequestMessage(0, '/any'), makePeerRequestMessage(1, '/localized')], + })) + + expect(response!.headers.get('vary')).toBe('*') + }) + it('sets each distinct cookie once, including the ones the headers option sets', async () => { const handler = createHandler(new BatchHandlerPlugin({ headers: { 'set-cookie': 'theme=dark' } }), headersRouter) diff --git a/packages/server/src/plugins/batch.ts b/packages/server/src/plugins/batch.ts index 3ff32968a9..ef3b40d598 100644 --- a/packages/server/src/plugins/batch.ts +++ b/packages/server/src/plugins/batch.ts @@ -64,7 +64,7 @@ interface SubresponseHead { /** * Builds the batch response headers: the headers the subresponses all have with the same value, as * the client does with the headers every subrequest shares, then the batch headers, which win, and - * the distinct cookies of both. + * the distinct cookies and `vary` fields of both. A cache must vary on whatever any subresponse varies on. */ function liftSubresponseHeaders( batchHeaders: StandardHeaders, @@ -92,6 +92,23 @@ function liftSubresponseHeaders( headers['set-cookie'] = [...cookies] } + const varies = new Map() + + for (const value of [batchHeaders, ...subresponseHeaders].flatMap(other => toArray(other.vary))) { + for (const field of value.split(',')) { + const name = field.trim() + const key = name.toLowerCase() + + if (name && !varies.has(key)) { + varies.set(key, name) + } + } + } + + if (varies.size > 0) { + headers.vary = varies.has('*') ? '*' : [...varies.values()].join(', ') + } + return headers } @@ -134,7 +151,7 @@ export interface BatchHandlerPluginOptions { * Success batch response headers. * * The batch response also gets the cookies its sub-responses set, and the headers they all share. - * Headers set here win, and setting one to `undefined` keeps it off. + * Headers set here win, and setting one to `undefined` keeps it off, except `set-cookie` and `vary`, which merge. * * @default {} */