fix(peer): skip empty octet-stream messages so a pending read can't stall

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<Uint8Array>).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<Uint8Array>).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'))

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<PeerOctetStreamMessage>()
+    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<PeerOctetStreamMessage>()
+      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<PeerOctetStreamMessage>()
     const cleanup = vi.fn()

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(

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<void>()
+        const handler = vi.fn<HandlerFn>().mockImplementation(async (req) => {
+          const body = (await req.resolveBody()) as ReadableStream<Uint8Array>
+          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])),