From c7522710dcf33a73d485a6251d73dbc21bfcddf4 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 00:15:06 +0000 Subject: [PATCH 1/3] fix(peer): keep pulling past octet-stream messages with neither binary nor close `toOctetStream` resolved `pull()` without enqueuing or closing when a message carried neither `binary` nor `close`. The stream only calls `pull` again if a read arrived meanwhile, so two such messages in a row left a pending read with nothing waiting on the queue: queue close/abort, later chunks, `ServerPeer.close()`, `ClientPeer.close()` and the request `AbortSignal` never reached the reader. A malicious peer could hang a handler's body read (its `finally` never ran) or a client's response read. `pull()` now loops until it enqueues a chunk or closes the stream, as `toAsyncIteratorObject` already does. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_016R6UEWNWXpmbcpXs2d1sTq --- packages/peer/src/client.test.ts | 37 ++++++++++++++++++++++++ packages/peer/src/octet-stream.test.ts | 40 +++++++++++++++++++++++++- packages/peer/src/octet-stream.ts | 30 +++++++++++++------ packages/peer/src/server.test.ts | 34 ++++++++++++++++++++++ 4 files changed, 131 insertions(+), 10 deletions(-) diff --git a/packages/peer/src/client.test.ts b/packages/peer/src/client.test.ts index 4ab058da..ffa8f8e3 100644 --- a/packages/peer/src/client.test.ts +++ b/packages/peer/src/client.test.ts @@ -1215,6 +1215,43 @@ describe('clientPeer', () => { await expect(reader.read()).rejects.toThrow('Server canceled the request') }) + it('rejects a pending read on signal abort after messages with neither binary nor close', async () => { + const controller = new AbortController() + const { id, promise } = await requestAndGetId(makeRequest({ signal: controller.signal })) + await peer.message(makeStreamingResponse(id, 'octet-stream')) + const response = await promise + + const reader = ((await response.resolveBody()) as ReadableStream).getReader() + const readPromise = reader.read() + + await peer.message(makeOctetStreamMessage(id)) + await peer.message(makeOctetStreamMessage(id)) + await sleep(1) // let the stream process both messages + + const error = new Error('timeout') + controller.abort(error) + + await expect(readPromise).rejects.toBe(error) + expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request', 'cancel']) + }) + + it('rejects a pending read on close after messages with neither binary nor close', async () => { + const { id, promise } = await requestAndGetId() + await peer.message(makeStreamingResponse(id, 'octet-stream')) + const response = await promise + + const reader = ((await response.resolveBody()) as ReadableStream).getReader() + const readPromise = reader.read() + + await peer.message(makeOctetStreamMessage(id)) + await peer.message(makeOctetStreamMessage(id)) + await sleep(1) // let the stream process both messages + + await peer.close() + + await expect(readPromise).rejects.toThrow(AbortError) + }) + it('throws on resolving the octet-stream response body multiple times', async () => { const { id, promise } = await requestAndGetId() await peer.message(makeStreamingResponse(id, 'octet-stream')) diff --git a/packages/peer/src/octet-stream.test.ts b/packages/peer/src/octet-stream.test.ts index c6a72eeb..f6f60a3c 100644 --- a/packages/peer/src/octet-stream.test.ts +++ b/packages/peer/src/octet-stream.test.ts @@ -1,6 +1,6 @@ import { Buffer } from 'node:buffer' -import { Queue } from '@standard-server/shared' +import { Queue, sleep } from '@standard-server/shared' import { OctetStreamTransmitter, toOctetStream } from './octet-stream' import type { PeerOctetStreamMessage } from './types' @@ -83,6 +83,44 @@ describe('toOctetStream', () => { expect(cleanup).toHaveBeenCalledWith({ kind: 'success' }) }) + it('skips messages with neither binary nor close', async () => { + const queue = new Queue() + const cleanup = vi.fn() + const stream = toOctetStream(queue, cleanup) + + queue.push({ id: '1', kind: 'octet-stream', json: {} }) + queue.push({ id: '1', kind: 'octet-stream', json: {} }) + queue.push({ id: '1', kind: 'octet-stream', json: {}, binary: new Uint8Array([1, 2, 3]) }) + queue.push({ id: '1', kind: 'octet-stream', json: { close: true } }) + + const reader = stream.getReader() + expect(await reader.read()).toEqual({ value: new Uint8Array([1, 2, 3]), done: false }) + expect(await reader.read()).toEqual({ value: undefined, done: true }) + expect(cleanup).toHaveBeenCalledWith({ kind: 'success' }) + }) + + it.each(['close', 'abort'] as const)( + 'rejects a pending read on queue %s after messages with neither binary nor close', + async (method) => { + const queue = new Queue() + const cleanup = vi.fn() + const stream = toOctetStream(queue, cleanup) + + const reader = stream.getReader() + const pending = reader.read() + + queue.push({ id: '1', kind: 'octet-stream', json: {} }) + queue.push({ id: '1', kind: 'octet-stream', json: {} }) + await sleep(1) // let the stream process both messages + + const error = new Error('closed') + queue[method](error) + + await expect(pending).rejects.toBe(error) + expect(cleanup).toHaveBeenCalledWith({ kind: 'error', error }) + }, + ) + it('calls cleanup(cancelled) on cancel', async () => { const queue = new Queue() const cleanup = vi.fn() diff --git a/packages/peer/src/octet-stream.ts b/packages/peer/src/octet-stream.ts index 88fe8184..5fe9b7c0 100644 --- a/packages/peer/src/octet-stream.ts +++ b/packages/peer/src/octet-stream.ts @@ -9,17 +9,29 @@ export function toOctetStream( return new ReadableStream({ async pull(controller) { try { - const { json, binary } = await queue.pull() + /** + * Keep pulling until a chunk is enqueued or the stream is closed: the stream calls `pull` + * again only if a read arrived meanwhile, so returning after a message with neither + * `binary` nor `close` can leave a pending read with nothing waiting on the queue. + */ + while (true) { + const { json, binary } = await queue.pull() - if (binary) { - controller.enqueue( - binary instanceof Uint8Array ? binary : new Uint8Array(await binary.arrayBuffer()), - ) - } + if (binary) { + controller.enqueue( + binary instanceof Uint8Array ? binary : new Uint8Array(await binary.arrayBuffer()), + ) + } + + if (json.close) { + await cleanup({ kind: 'success' }) + controller.close() + return + } - if (json.close) { - await cleanup({ kind: 'success' }) - controller.close() + if (binary) { + return + } } } catch (error) { await cleanup({ kind: 'error', error }) diff --git a/packages/peer/src/server.test.ts b/packages/peer/src/server.test.ts index 10e7d701..899a776b 100644 --- a/packages/peer/src/server.test.ts +++ b/packages/peer/src/server.test.ts @@ -4,6 +4,7 @@ import { AbortError, AsyncIteratorClass, isAsyncIteratorObject, + promiseWithResolvers, sleep, } from '@standard-server/shared' @@ -694,6 +695,39 @@ describe('serverPeer', () => { await promise }) + it('aborts a pending read on close after messages with neither binary nor close', async () => { + const reading = promiseWithResolvers() + const finished = vi.fn() + const handler = vi.fn().mockImplementation(async (req) => { + const body = (await req.resolveBody()) as ReadableStream + const reader = body.getReader() + try { + const read = reader.read() + reading.resolve() + await expect(read).rejects.toThrow(AbortError) + return jsonResponse() + } finally { + finished() + } + }) + + const msg = makeRequestMessage({ + headers: { 'content-type': 'application/octet-stream' }, + }) + const promise = peer.message(msg, handler) + await reading.promise + + await peer.message(makeOctetStreamMessage('1'), vi.fn()) + await peer.message(makeOctetStreamMessage('1'), vi.fn()) + await sleep(1) // let the stream process both messages + + await peer.close() + + await promise + expect(finished).toHaveBeenCalledOnce() + expect(send).not.toHaveBeenCalled() + }) + it('ignores octet-stream for non-existing request', async () => { await peer.message( makeOctetStreamMessage('nonexist', undefined, new Uint8Array([1, 2, 3])), From 5aad5ba9921f99326fb9e3da6de43da59fefd251 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 01:52:59 +0000 Subject: [PATCH 2/3] refactor(peer): simplify octet-stream skip loop and its server regression test Skip messages with neither `binary` nor `close` up front with one `continue` instead of re-checking `binary` to decide when to return, and drop the redundant `finally` spy from the server test: `peer.message()` settling already proves the handler ran to completion. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_016R6UEWNWXpmbcpXs2d1sTq --- packages/peer/src/octet-stream.ts | 19 ++++++++++--------- packages/peer/src/server.test.ts | 16 +++++----------- 2 files changed, 15 insertions(+), 20 deletions(-) diff --git a/packages/peer/src/octet-stream.ts b/packages/peer/src/octet-stream.ts index 5fe9b7c0..f7a81ff6 100644 --- a/packages/peer/src/octet-stream.ts +++ b/packages/peer/src/octet-stream.ts @@ -9,14 +9,18 @@ export function toOctetStream( return new ReadableStream({ async pull(controller) { try { - /** - * Keep pulling until a chunk is enqueued or the stream is closed: the stream calls `pull` - * again only if a read arrived meanwhile, so returning after a message with neither - * `binary` nor `close` can leave a pending read with nothing waiting on the queue. - */ while (true) { const { json, binary } = await queue.pull() + /** + * Skip messages with neither `binary` nor `close`: the stream calls `pull` again only if + * a read arrived meanwhile, so returning without enqueuing or closing can leave a pending + * read with nothing waiting on the queue. + */ + if (!binary && !json.close) { + continue + } + if (binary) { controller.enqueue( binary instanceof Uint8Array ? binary : new Uint8Array(await binary.arrayBuffer()), @@ -26,12 +30,9 @@ export function toOctetStream( if (json.close) { await cleanup({ kind: 'success' }) controller.close() - return } - if (binary) { - return - } + return } } catch (error) { await cleanup({ kind: 'error', error }) diff --git a/packages/peer/src/server.test.ts b/packages/peer/src/server.test.ts index 899a776b..72c4753a 100644 --- a/packages/peer/src/server.test.ts +++ b/packages/peer/src/server.test.ts @@ -697,18 +697,12 @@ describe('serverPeer', () => { it('aborts a pending read on close after messages with neither binary nor close', async () => { const reading = promiseWithResolvers() - const finished = vi.fn() const handler = vi.fn().mockImplementation(async (req) => { const body = (await req.resolveBody()) as ReadableStream - const reader = body.getReader() - try { - const read = reader.read() - reading.resolve() - await expect(read).rejects.toThrow(AbortError) - return jsonResponse() - } finally { - finished() - } + const read = body.getReader().read() + reading.resolve() + await expect(read).rejects.toThrow(AbortError) + return jsonResponse() }) const msg = makeRequestMessage({ @@ -723,8 +717,8 @@ describe('serverPeer', () => { await peer.close() + // settles only once the close reaches the handler's pending read await promise - expect(finished).toHaveBeenCalledOnce() expect(send).not.toHaveBeenCalled() }) From 36fe3d493e9baa8f870ef67c2e47eb42cfba02ab Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 04:15:13 +0000 Subject: [PATCH 3/3] refactor(peer): loop only around the queue pull when skipping empty octet-stream messages Wrap just the `queue.pull()` in the skip loop, so the enqueue/close body stays as it was before the fix and needs no `continue` or trailing `return`. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_016R6UEWNWXpmbcpXs2d1sTq --- packages/peer/src/octet-stream.ts | 43 ++++++++++++++----------------- 1 file changed, 20 insertions(+), 23 deletions(-) diff --git a/packages/peer/src/octet-stream.ts b/packages/peer/src/octet-stream.ts index f7a81ff6..2fe38a93 100644 --- a/packages/peer/src/octet-stream.ts +++ b/packages/peer/src/octet-stream.ts @@ -9,30 +9,27 @@ export function toOctetStream( return new ReadableStream({ async pull(controller) { try { - while (true) { - const { json, binary } = await queue.pull() - - /** - * Skip messages with neither `binary` nor `close`: the stream calls `pull` again only if - * a read arrived meanwhile, so returning without enqueuing or closing can leave a pending - * read with nothing waiting on the queue. - */ - if (!binary && !json.close) { - continue - } - - if (binary) { - controller.enqueue( - binary instanceof Uint8Array ? binary : new Uint8Array(await binary.arrayBuffer()), - ) - } - - if (json.close) { - await cleanup({ kind: 'success' }) - controller.close() - } + /** + * Skip messages with neither `binary` nor `close`: the stream calls `pull` again only if + * a read arrived meanwhile, so returning without enqueuing or closing can leave a pending + * read with nothing waiting on the queue. + */ + let message: PeerOctetStreamMessage + do { + message = await queue.pull() + } while (!message.binary && !message.json.close) + + const { json, binary } = message + + if (binary) { + controller.enqueue( + binary instanceof Uint8Array ? binary : new Uint8Array(await binary.arrayBuffer()), + ) + } - return + if (json.close) { + await cleanup({ kind: 'success' }) + controller.close() } } catch (error) { await cleanup({ kind: 'error', error })