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..2fe38a93 100644 --- a/packages/peer/src/octet-stream.ts +++ b/packages/peer/src/octet-stream.ts @@ -9,7 +9,17 @@ export function toOctetStream( return new ReadableStream({ async pull(controller) { try { - 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. + */ + let message: PeerOctetStreamMessage + do { + message = await queue.pull() + } while (!message.binary && !message.json.close) + + const { json, binary } = message if (binary) { controller.enqueue( diff --git a/packages/peer/src/server.test.ts b/packages/peer/src/server.test.ts index 10e7d701..72c4753a 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,33 @@ describe('serverPeer', () => { await promise }) + it('aborts a pending read on close after messages with neither binary nor close', async () => { + const reading = promiseWithResolvers() + const handler = vi.fn().mockImplementation(async (req) => { + const body = (await req.resolveBody()) as ReadableStream + const read = body.getReader().read() + reading.resolve() + await expect(read).rejects.toThrow(AbortError) + return jsonResponse() + }) + + 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() + + // settles only once the close reaches the handler's pending read + await promise + expect(send).not.toHaveBeenCalled() + }) + it('ignores octet-stream for non-existing request', async () => { await peer.message( makeOctetStreamMessage('nonexist', undefined, new Uint8Array([1, 2, 3])),