From 5163fadb90d5106c239201bb402f34c6d306ff31 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 12 Oct 2026 04:59:12 +0000 Subject: [PATCH 1/5] fix(peer): release client request state once a streamed response finishes ClientPeer kept a streamed response's request entry, its abort listener, and every buffered stream message until the caller consumed or cancelled the body. A caller that only read the status leaked all of it, even after the server had finished the stream and released its own state. Release the request as soon as the terminal stream message arrives (an event-stream `close`/`error` event, or an octet-stream `close` chunk). The queue is closed, not aborted, so a later resolveBody() still yields every buffered item, and cancelling that body sends no cancel for a stream the server already finished. A stream that never finishes still needs the caller to cancel the body or abort the signal; document that in the README and on ClientPeer.request. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AEr5Gq656PLdpCLEAw2zht --- packages/peer/README.md | 13 ++++ packages/peer/src/client.test.ts | 117 +++++++++++++++++++++++++++++++ packages/peer/src/client.ts | 27 ++++++- packages/peer/tests/peer.test.ts | 82 +++++++++++++++++++++- 4 files changed, 235 insertions(+), 4 deletions(-) diff --git a/packages/peer/README.md b/packages/peer/README.md index 1b9ad995..1c52ce9f 100644 --- a/packages/peer/README.md +++ b/packages/peer/README.md @@ -128,6 +128,19 @@ const payload = await response.resolveBody() Unlike the HTTP adapters, `resolveBody(hint?)` ignores the `hint` argument in this adapter. HTTP adapters receive the body as a raw byte stream and must decide how to parse it, so a hint can steer that decision. The peer protocol instead encodes the body in structured form at send time: JSON values travel as JSON, binary payloads travel as binary, event and octet streams flow as dedicated stream messages, and markers in the message distinguish the ambiguous cases such as `form-data` vs. `file`. By the time a message arrives, there are no raw bytes left to reinterpret — the body always resolves to exactly the representation the sender had, so a hint has nothing to override. +A streamed response body (an event stream or an octet stream) keeps its request open on the `ClientPeer` until the server finishes the stream. When the last stream message arrives, the client releases the request on its own, and the buffered messages stay readable through `resolveBody()` for as long as you keep the response. Before that, the request is only released when you cancel the body, abort the request signal, or close the peer, so a stream that never finishes keeps its request state, its abort listener, and every message the server streams in memory. If you may not read a streamed body, cancel it so the server stops streaming and both peers release the request: + +```ts +import { cancelStandardBody } from '@standard-server/core' + +const response = await clientPeer.request({ method: 'GET', url: '/events', headers: {} }) + +if (response.status !== 200) { + // does nothing for a body that is not a stream + await cancelStandardBody(await response.resolveBody()) +} +``` + ## Codec helpers Use `encodePeerMessage()` and `decodePeerMessage()` to bridge between the peer protocol and your underlying transport. diff --git a/packages/peer/src/client.test.ts b/packages/peer/src/client.test.ts index 1baa5b9c..319eed68 100644 --- a/packages/peer/src/client.test.ts +++ b/packages/peer/src/client.test.ts @@ -1,4 +1,5 @@ import type { StandardLazyResponse, StandardRequest } from '@standard-server/core' +import { ErrorEvent } from '@standard-server/core' import { AbortError, AsyncIteratorClass, @@ -819,6 +820,74 @@ describe('clientPeer', () => { new TypeError('Failed to read body: body stream already read'), ) }) + + it('releases the request once the stream closes, even if the body is never resolved', async () => { + const controller = new AbortController() + const addSpy = vi.spyOn(controller.signal, 'addEventListener') + const removeSpy = vi.spyOn(controller.signal, 'removeEventListener') + + const { id, promise } = await requestAndGetId(makeRequest({ signal: controller.signal })) + await peer.message(makeStreamingResponse(id, 'event-stream')) + const response = await promise + + await peer.message(makeEventStreamMessage(id, 'evt1')) + expect(getPeerSize(peer)).toBe(1) + expect(removeSpy).not.toHaveBeenCalled() + + // the server already released the request, so nothing waits for the body to be read + await peer.message(makeEventStreamMessage(id, 'bye', 'close')) + expect(getPeerSize(peer)).toBe(0) + expect(removeSpy).toHaveBeenCalledOnce() + expect(removeSpy).toHaveBeenCalledWith('abort', addSpy.mock.calls[0]![1]) + + // the buffered events stay readable + const iter = (await response.resolveBody()) as AsyncIterator + await expect(iter.next()).resolves.toEqual({ done: false, value: 'evt1' }) + await expect(iter.next()).resolves.toEqual({ done: true, value: 'bye' }) + await expect(iter.next()).resolves.toEqual({ done: true, value: undefined }) + + // aborting after completion is a no-op: no cancel message is sent + controller.abort(new Error('late abort')) + await sleep(1) + expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request']) + }) + + it('releases the request once the stream errors, even if the body is never resolved', async () => { + const { id, promise } = await requestAndGetId() + await peer.message(makeStreamingResponse(id, 'event-stream')) + const response = await promise + + await peer.message(makeEventStreamMessage(id, 'evt1')) + await peer.message(makeEventStreamMessage(id, { code: 'BOOM' }, 'error')) + expect(getPeerSize(peer)).toBe(0) + + const iter = (await response.resolveBody()) as AsyncIterator + await expect(iter.next()).resolves.toEqual({ done: false, value: 'evt1' }) + await expect(iter.next()).rejects.toSatisfy((error: ErrorEvent) => { + expect(error).toBeInstanceOf(ErrorEvent) + expect(error.data).toEqual({ code: 'BOOM' }) + return true + }) + await expect(iter.next()).resolves.toEqual({ done: true, value: undefined }) + expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request']) + }) + + it('does not send cancel message when canceling the iterator of a finished stream', async () => { + const { id, promise } = await requestAndGetId() + await peer.message(makeStreamingResponse(id, 'event-stream')) + const response = await promise + + await peer.message(makeEventStreamMessage(id, 'evt1')) + await peer.message(makeEventStreamMessage(id, 'evt2')) + await peer.message(makeEventStreamMessage(id, 'bye', 'close')) + + const iter = (await response.resolveBody()) as AsyncIterator + await expect(iter.next()).resolves.toEqual({ done: false, value: 'evt1' }) + await iter.return?.() + + // the server already released the request, so there is nothing to cancel + expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request']) + }) }) }) @@ -1177,6 +1246,54 @@ describe('clientPeer', () => { new TypeError('Failed to read body: body stream already read'), ) }) + + it('releases the request once the stream closes, even if the body is never resolved', async () => { + const controller = new AbortController() + const addSpy = vi.spyOn(controller.signal, 'addEventListener') + const removeSpy = vi.spyOn(controller.signal, 'removeEventListener') + + const { id, promise } = await requestAndGetId(makeRequest({ signal: controller.signal })) + await peer.message(makeStreamingResponse(id, 'octet-stream')) + const response = await promise + + await peer.message(makeOctetStreamMessage(id, false, new Uint8Array([1, 2]))) + expect(getPeerSize(peer)).toBe(1) + expect(removeSpy).not.toHaveBeenCalled() + + // the server already released the request, so nothing waits for the body to be read + await peer.message(makeOctetStreamMessage(id, true, new Uint8Array([3]))) + expect(getPeerSize(peer)).toBe(0) + expect(removeSpy).toHaveBeenCalledOnce() + expect(removeSpy).toHaveBeenCalledWith('abort', addSpy.mock.calls[0]![1]) + + // the buffered chunks stay readable + const reader = ((await response.resolveBody()) as ReadableStream).getReader() + expect(await reader.read()).toEqual({ done: false, value: new Uint8Array([1, 2]) }) + expect(await reader.read()).toEqual({ done: false, value: new Uint8Array([3]) }) + expect(await reader.read()).toEqual({ done: true, value: undefined }) + + // aborting after completion is a no-op: no cancel message is sent + controller.abort(new Error('late abort')) + await sleep(1) + expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request']) + }) + + it('does not send cancel message when canceling the ReadableStream of a finished stream', async () => { + const { id, promise } = await requestAndGetId() + await peer.message(makeStreamingResponse(id, 'octet-stream')) + const response = await promise + + await peer.message(makeOctetStreamMessage(id, false, new Uint8Array([1]))) + await peer.message(makeOctetStreamMessage(id, false, new Uint8Array([2]))) + await peer.message(makeOctetStreamMessage(id, true)) + + const reader = ((await response.resolveBody()) as ReadableStream).getReader() + expect(await reader.read()).toEqual({ done: false, value: new Uint8Array([1]) }) + await reader.cancel() + + // the server already released the request, so there is nothing to cancel + expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request']) + }) }) }) diff --git a/packages/peer/src/client.ts b/packages/peer/src/client.ts index 5a42a331..129d940c 100644 --- a/packages/peer/src/client.ts +++ b/packages/peer/src/client.ts @@ -43,6 +43,12 @@ export class ClientPeer { /** * Send a request to the server peer + * + * A streamed response body (async iterator or `ReadableStream`) keeps the request open until the + * server finishes the stream. If you may not read it to the end, cancel it, e.g. + * `cancelStandardBody(await response.resolveBody())` from `@standard-server/core`, or abort + * `request.signal`. Otherwise a stream that never finishes keeps the request state, its abort + * listener, and every streamed message in memory. */ request(request: StandardRequest): Promise { return new Promise((resolve, reject) => { @@ -178,13 +184,30 @@ export class ClientPeer { return } + /** + * The server releases the request once it sends the last stream message, so release it here too + * instead of waiting for a body the caller may never read. Closing keeps the buffered messages + * readable through `resolveBody()`, and the body cleanup then finds no state to cancel. + */ if (message.kind === 'event-stream') { - state.eventStreamMessageQueue?.push(message) + const queue = state.eventStreamMessageQueue + queue?.push(message) + + if (queue && (message.json.event === 'close' || message.json.event === 'error')) { + await this.closeById(id) + } + return } if (message.kind === 'octet-stream') { - state.octetStreamMessageQueue?.push(message) + const queue = state.octetStreamMessageQueue + queue?.push(message) + + if (queue && message.json.close) { + await this.closeById(id) + } + return } diff --git a/packages/peer/tests/peer.test.ts b/packages/peer/tests/peer.test.ts index c7d4f780..4c3ce99a 100644 --- a/packages/peer/tests/peer.test.ts +++ b/packages/peer/tests/peer.test.ts @@ -1,6 +1,8 @@ +import { getEventListeners } from 'node:events' + import type { StandardLazyRequest, StandardResponse } from '@standard-server/core' -import { ErrorEvent, getEventMeta, withEventMeta } from '@standard-server/core' -import { promiseWithResolvers } from '@standard-server/shared' +import { cancelStandardBody, ErrorEvent, getEventMeta, withEventMeta } from '@standard-server/core' +import { promiseWithResolvers, sleep } from '@standard-server/shared' import type { ClientPeer, ServerPeer } from '../src' import { @@ -357,6 +359,82 @@ describe('peer integration (client <-> server over encoded wire)', () => { await vi.waitFor(() => expect(cancel).toHaveBeenCalled()) }) + it('releases both peers when finished streamed response bodies are never resolved', async () => { + const { client, server } = connect(async (request) => ({ + status: 200, + headers: {}, + body: + request.url === '/events' + ? (async function* () { + yield 1 + yield 2 + })() + : new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array([1, 2])) + controller.close() + }, + }), + })) + + // a long-lived signal shared by every request, e.g. one per page or app + const controller = new AbortController() + + for (let i = 0; i < 100; i++) { + const response = await client.request({ + url: i % 2 === 0 ? '/events' : '/download', + method: 'GET', + headers: {}, + signal: controller.signal, + }) + expect(response.status).toBe(200) // body ignored + } + + await vi.waitFor(() => { + expect((client as any).requests.size).toBe(0) + expect((server as any).requests.size).toBe(0) + }) + expect(getEventListeners(controller.signal, 'abort')).toHaveLength(0) + }) + + it('releases both peers when endless streamed response bodies are cancelled without reading', async () => { + let finalized = 0 + + const { client, server } = connect(async () => ({ + status: 200, + headers: {}, + body: (async function* () { + try { + while (true) { + yield 'tick' + await sleep(1) + } + } finally { + finalized++ + } + })(), + })) + + const controller = new AbortController() + + for (let i = 0; i < 10; i++) { + const response = await client.request({ + url: '/events', + method: 'GET', + headers: {}, + signal: controller.signal, + }) + await cancelStandardBody(await response.resolveBody()) + } + + await vi.waitFor(() => { + expect((client as any).requests.size).toBe(0) + expect((server as any).requests.size).toBe(0) + expect(finalized).toBe(10) + }) + expect(getEventListeners(controller.signal, 'abort')).toHaveLength(0) + }) + it('propagates client aborts to the server handler signal', async () => { let serverSignal: AbortSignal | undefined From fa29b2364c8d7297b571cce62bf7b25b79214b7f Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 12 Oct 2026 07:33:17 +0000 Subject: [PATCH 2/5] refactor(peer): trim tests and docs for the streamed response release - drop the late-abort and exact-listener checks from the new unit tests; the existing abort-listener tests already cover them - cancel a single endless stream in the integration test instead of ten - shorten the ClientPeer.request JSDoc so it no longer repeats the README, and stop claiming the server always stops streaming on cancel (a hibernated stream is no longer tracked by ServerPeer) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AEr5Gq656PLdpCLEAw2zht --- packages/peer/README.md | 2 +- packages/peer/src/client.test.ts | 12 ------------ packages/peer/src/client.ts | 10 ++++------ packages/peer/tests/peer.test.ts | 25 +++++++++++-------------- 4 files changed, 16 insertions(+), 33 deletions(-) diff --git a/packages/peer/README.md b/packages/peer/README.md index 1c52ce9f..15e740d9 100644 --- a/packages/peer/README.md +++ b/packages/peer/README.md @@ -128,7 +128,7 @@ const payload = await response.resolveBody() Unlike the HTTP adapters, `resolveBody(hint?)` ignores the `hint` argument in this adapter. HTTP adapters receive the body as a raw byte stream and must decide how to parse it, so a hint can steer that decision. The peer protocol instead encodes the body in structured form at send time: JSON values travel as JSON, binary payloads travel as binary, event and octet streams flow as dedicated stream messages, and markers in the message distinguish the ambiguous cases such as `form-data` vs. `file`. By the time a message arrives, there are no raw bytes left to reinterpret — the body always resolves to exactly the representation the sender had, so a hint has nothing to override. -A streamed response body (an event stream or an octet stream) keeps its request open on the `ClientPeer` until the server finishes the stream. When the last stream message arrives, the client releases the request on its own, and the buffered messages stay readable through `resolveBody()` for as long as you keep the response. Before that, the request is only released when you cancel the body, abort the request signal, or close the peer, so a stream that never finishes keeps its request state, its abort listener, and every message the server streams in memory. If you may not read a streamed body, cancel it so the server stops streaming and both peers release the request: +A streamed response body (an event stream or an octet stream) keeps its request open on the `ClientPeer` until the server finishes the stream. When the last stream message arrives, the client releases the request on its own, and the buffered messages stay readable through `resolveBody()` for as long as you keep the response. A stream that never finishes stays open, keeping every message the server streams in memory, until you cancel the body, abort the request signal, or close the peer. So if you may not read a streamed body to the end, cancel it; the client then tells the server to stop and releases the request: ```ts import { cancelStandardBody } from '@standard-server/core' diff --git a/packages/peer/src/client.test.ts b/packages/peer/src/client.test.ts index 319eed68..1e630f4f 100644 --- a/packages/peer/src/client.test.ts +++ b/packages/peer/src/client.test.ts @@ -823,7 +823,6 @@ describe('clientPeer', () => { it('releases the request once the stream closes, even if the body is never resolved', async () => { const controller = new AbortController() - const addSpy = vi.spyOn(controller.signal, 'addEventListener') const removeSpy = vi.spyOn(controller.signal, 'removeEventListener') const { id, promise } = await requestAndGetId(makeRequest({ signal: controller.signal })) @@ -838,17 +837,12 @@ describe('clientPeer', () => { await peer.message(makeEventStreamMessage(id, 'bye', 'close')) expect(getPeerSize(peer)).toBe(0) expect(removeSpy).toHaveBeenCalledOnce() - expect(removeSpy).toHaveBeenCalledWith('abort', addSpy.mock.calls[0]![1]) // the buffered events stay readable const iter = (await response.resolveBody()) as AsyncIterator await expect(iter.next()).resolves.toEqual({ done: false, value: 'evt1' }) await expect(iter.next()).resolves.toEqual({ done: true, value: 'bye' }) await expect(iter.next()).resolves.toEqual({ done: true, value: undefined }) - - // aborting after completion is a no-op: no cancel message is sent - controller.abort(new Error('late abort')) - await sleep(1) expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request']) }) @@ -1249,7 +1243,6 @@ describe('clientPeer', () => { it('releases the request once the stream closes, even if the body is never resolved', async () => { const controller = new AbortController() - const addSpy = vi.spyOn(controller.signal, 'addEventListener') const removeSpy = vi.spyOn(controller.signal, 'removeEventListener') const { id, promise } = await requestAndGetId(makeRequest({ signal: controller.signal })) @@ -1264,17 +1257,12 @@ describe('clientPeer', () => { await peer.message(makeOctetStreamMessage(id, true, new Uint8Array([3]))) expect(getPeerSize(peer)).toBe(0) expect(removeSpy).toHaveBeenCalledOnce() - expect(removeSpy).toHaveBeenCalledWith('abort', addSpy.mock.calls[0]![1]) // the buffered chunks stay readable const reader = ((await response.resolveBody()) as ReadableStream).getReader() expect(await reader.read()).toEqual({ done: false, value: new Uint8Array([1, 2]) }) expect(await reader.read()).toEqual({ done: false, value: new Uint8Array([3]) }) expect(await reader.read()).toEqual({ done: true, value: undefined }) - - // aborting after completion is a no-op: no cancel message is sent - controller.abort(new Error('late abort')) - await sleep(1) expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request']) }) diff --git a/packages/peer/src/client.ts b/packages/peer/src/client.ts index 129d940c..c003a552 100644 --- a/packages/peer/src/client.ts +++ b/packages/peer/src/client.ts @@ -45,10 +45,8 @@ export class ClientPeer { * Send a request to the server peer * * A streamed response body (async iterator or `ReadableStream`) keeps the request open until the - * server finishes the stream. If you may not read it to the end, cancel it, e.g. - * `cancelStandardBody(await response.resolveBody())` from `@standard-server/core`, or abort - * `request.signal`. Otherwise a stream that never finishes keeps the request state, its abort - * listener, and every streamed message in memory. + * server finishes it. If you may not read it to the end, cancel it, e.g. + * `cancelStandardBody(await response.resolveBody())`, or abort `request.signal`. */ request(request: StandardRequest): Promise { return new Promise((resolve, reject) => { @@ -185,8 +183,8 @@ export class ClientPeer { } /** - * The server releases the request once it sends the last stream message, so release it here too - * instead of waiting for a body the caller may never read. Closing keeps the buffered messages + * By the last stream message the server has already released the request, so release it here + * too instead of waiting for a body the caller may never read. Closing keeps the buffered messages * readable through `resolveBody()`, and the body cleanup then finds no state to cancel. */ if (message.kind === 'event-stream') { diff --git a/packages/peer/tests/peer.test.ts b/packages/peer/tests/peer.test.ts index 4c3ce99a..55b7e313 100644 --- a/packages/peer/tests/peer.test.ts +++ b/packages/peer/tests/peer.test.ts @@ -397,8 +397,8 @@ describe('peer integration (client <-> server over encoded wire)', () => { expect(getEventListeners(controller.signal, 'abort')).toHaveLength(0) }) - it('releases both peers when endless streamed response bodies are cancelled without reading', async () => { - let finalized = 0 + it('releases both peers when an endless streamed response body is cancelled without reading', async () => { + const finalized = vi.fn() const { client, server } = connect(async () => ({ status: 200, @@ -410,27 +410,24 @@ describe('peer integration (client <-> server over encoded wire)', () => { await sleep(1) } } finally { - finalized++ + finalized() } })(), })) const controller = new AbortController() - - for (let i = 0; i < 10; i++) { - const response = await client.request({ - url: '/events', - method: 'GET', - headers: {}, - signal: controller.signal, - }) - await cancelStandardBody(await response.resolveBody()) - } + const response = await client.request({ + url: '/events', + method: 'GET', + headers: {}, + signal: controller.signal, + }) + await cancelStandardBody(await response.resolveBody()) await vi.waitFor(() => { expect((client as any).requests.size).toBe(0) expect((server as any).requests.size).toBe(0) - expect(finalized).toBe(10) + expect(finalized).toHaveBeenCalledOnce() }) expect(getEventListeners(controller.signal, 'abort')).toHaveLength(0) }) From 4d0338bbc45bc3db5ee21c37ad8b09e90d786469 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 12 Oct 2026 07:41:07 +0000 Subject: [PATCH 3/5] chore(peer): drop comments the new tests already cover Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AEr5Gq656PLdpCLEAw2zht --- packages/peer/src/client.test.ts | 6 ------ packages/peer/src/client.ts | 6 +----- packages/peer/tests/peer.test.ts | 3 +-- 3 files changed, 2 insertions(+), 13 deletions(-) diff --git a/packages/peer/src/client.test.ts b/packages/peer/src/client.test.ts index 1e630f4f..4ab058da 100644 --- a/packages/peer/src/client.test.ts +++ b/packages/peer/src/client.test.ts @@ -833,12 +833,10 @@ describe('clientPeer', () => { expect(getPeerSize(peer)).toBe(1) expect(removeSpy).not.toHaveBeenCalled() - // the server already released the request, so nothing waits for the body to be read await peer.message(makeEventStreamMessage(id, 'bye', 'close')) expect(getPeerSize(peer)).toBe(0) expect(removeSpy).toHaveBeenCalledOnce() - // the buffered events stay readable const iter = (await response.resolveBody()) as AsyncIterator await expect(iter.next()).resolves.toEqual({ done: false, value: 'evt1' }) await expect(iter.next()).resolves.toEqual({ done: true, value: 'bye' }) @@ -879,7 +877,6 @@ describe('clientPeer', () => { await expect(iter.next()).resolves.toEqual({ done: false, value: 'evt1' }) await iter.return?.() - // the server already released the request, so there is nothing to cancel expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request']) }) }) @@ -1253,12 +1250,10 @@ describe('clientPeer', () => { expect(getPeerSize(peer)).toBe(1) expect(removeSpy).not.toHaveBeenCalled() - // the server already released the request, so nothing waits for the body to be read await peer.message(makeOctetStreamMessage(id, true, new Uint8Array([3]))) expect(getPeerSize(peer)).toBe(0) expect(removeSpy).toHaveBeenCalledOnce() - // the buffered chunks stay readable const reader = ((await response.resolveBody()) as ReadableStream).getReader() expect(await reader.read()).toEqual({ done: false, value: new Uint8Array([1, 2]) }) expect(await reader.read()).toEqual({ done: false, value: new Uint8Array([3]) }) @@ -1279,7 +1274,6 @@ describe('clientPeer', () => { expect(await reader.read()).toEqual({ done: false, value: new Uint8Array([1]) }) await reader.cancel() - // the server already released the request, so there is nothing to cancel expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request']) }) }) diff --git a/packages/peer/src/client.ts b/packages/peer/src/client.ts index c003a552..2608d52a 100644 --- a/packages/peer/src/client.ts +++ b/packages/peer/src/client.ts @@ -182,11 +182,7 @@ export class ClientPeer { return } - /** - * By the last stream message the server has already released the request, so release it here - * too instead of waiting for a body the caller may never read. Closing keeps the buffered messages - * readable through `resolveBody()`, and the body cleanup then finds no state to cancel. - */ + // by the last stream message the server has already released the request, so release it too if (message.kind === 'event-stream') { const queue = state.eventStreamMessageQueue queue?.push(message) diff --git a/packages/peer/tests/peer.test.ts b/packages/peer/tests/peer.test.ts index 55b7e313..aaee8ebd 100644 --- a/packages/peer/tests/peer.test.ts +++ b/packages/peer/tests/peer.test.ts @@ -377,7 +377,6 @@ describe('peer integration (client <-> server over encoded wire)', () => { }), })) - // a long-lived signal shared by every request, e.g. one per page or app const controller = new AbortController() for (let i = 0; i < 100; i++) { @@ -387,7 +386,7 @@ describe('peer integration (client <-> server over encoded wire)', () => { headers: {}, signal: controller.signal, }) - expect(response.status).toBe(200) // body ignored + expect(response.status).toBe(200) } await vi.waitFor(() => { From 36aa9a15622defb2bee87bea0fdf36d98663dbc2 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 12 Oct 2026 07:44:48 +0000 Subject: [PATCH 4/5] docs(peer): drop the streamed body note from ClientPeer.request JSDoc Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AEr5Gq656PLdpCLEAw2zht --- packages/peer/src/client.ts | 4 ---- 1 file changed, 4 deletions(-) diff --git a/packages/peer/src/client.ts b/packages/peer/src/client.ts index 2608d52a..79eb6d7c 100644 --- a/packages/peer/src/client.ts +++ b/packages/peer/src/client.ts @@ -43,10 +43,6 @@ export class ClientPeer { /** * Send a request to the server peer - * - * A streamed response body (async iterator or `ReadableStream`) keeps the request open until the - * server finishes it. If you may not read it to the end, cancel it, e.g. - * `cancelStandardBody(await response.resolveBody())`, or abort `request.signal`. */ request(request: StandardRequest): Promise { return new Promise((resolve, reject) => { From 81661f494f76f6848e901db9b7262f81aafc2872 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 12 Oct 2026 07:45:11 +0000 Subject: [PATCH 5/5] docs(peer): drop the streamed body note from the README Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AEr5Gq656PLdpCLEAw2zht --- packages/peer/README.md | 13 ------------- 1 file changed, 13 deletions(-) diff --git a/packages/peer/README.md b/packages/peer/README.md index 15e740d9..1b9ad995 100644 --- a/packages/peer/README.md +++ b/packages/peer/README.md @@ -128,19 +128,6 @@ const payload = await response.resolveBody() Unlike the HTTP adapters, `resolveBody(hint?)` ignores the `hint` argument in this adapter. HTTP adapters receive the body as a raw byte stream and must decide how to parse it, so a hint can steer that decision. The peer protocol instead encodes the body in structured form at send time: JSON values travel as JSON, binary payloads travel as binary, event and octet streams flow as dedicated stream messages, and markers in the message distinguish the ambiguous cases such as `form-data` vs. `file`. By the time a message arrives, there are no raw bytes left to reinterpret — the body always resolves to exactly the representation the sender had, so a hint has nothing to override. -A streamed response body (an event stream or an octet stream) keeps its request open on the `ClientPeer` until the server finishes the stream. When the last stream message arrives, the client releases the request on its own, and the buffered messages stay readable through `resolveBody()` for as long as you keep the response. A stream that never finishes stays open, keeping every message the server streams in memory, until you cancel the body, abort the request signal, or close the peer. So if you may not read a streamed body to the end, cancel it; the client then tells the server to stop and releases the request: - -```ts -import { cancelStandardBody } from '@standard-server/core' - -const response = await clientPeer.request({ method: 'GET', url: '/events', headers: {} }) - -if (response.status !== 200) { - // does nothing for a body that is not a stream - await cancelStandardBody(await response.resolveBody()) -} -``` - ## Codec helpers Use `encodePeerMessage()` and `decodePeerMessage()` to bridge between the peer protocol and your underlying transport.