diff --git a/packages/peer/src/client.test.ts b/packages/peer/src/client.test.ts index c64e44c..d9b1732 100644 --- a/packages/peer/src/client.test.ts +++ b/packages/peer/src/client.test.ts @@ -911,6 +911,41 @@ describe('clientPeer', () => { await promise }) + it('sends cancel message and rejects request when the body cannot be read after the request is sent', async () => { + const stream = new ReadableStream() + stream.getReader() + + await expect( + peer.request(makeRequest({ method: 'POST', headers: {}, body: stream })), + ).rejects.toThrow(TypeError) + + // the server already received the request, so it must be told to drop it + const id = (send.mock.calls[0]![0] as PeerRequestMessage).id + expect(send.mock.calls.map(([m]) => m)).toEqual([ + expect.objectContaining({ id, kind: 'request' }), + { id, kind: 'cancel' }, + ]) + }) + + it('silently ignores transport failures when cancelling after the body cannot be read', async () => { + send.mockImplementation(async (message) => { + if (message.kind === 'cancel') { + throw new Error('transport down') + } + }) + + const stream = new ReadableStream() + stream.getReader() + + await expect( + peer.request(makeRequest({ method: 'POST', headers: {}, body: stream })), + ).rejects.toThrow(TypeError) + + // let the failed cancel delivery settle; it must not surface anywhere + await sleep(1) + expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request', 'cancel']) + }) + it('stops transmitting the octet-stream request body when a full response arrives', async () => { const cancel = vi.fn() const stream = new ReadableStream({ diff --git a/packages/peer/src/client.ts b/packages/peer/src/client.ts index 96bafde..5a42a33 100644 --- a/packages/peer/src/client.ts +++ b/packages/peer/src/client.ts @@ -66,7 +66,7 @@ export class ClientPeer { state.removeAbortListener = () => signal.removeEventListener('abort', abortListener) } - void this.transmitRequest(id, state, request) + void this.transmitRequest(id, state, request).catch(() => {}) }) } @@ -125,23 +125,20 @@ export class ClientPeer { if (isAsyncIteratorObject(request.body)) { const transmitter = new EventStreamTransmitter(request.body, id, this.send) state.eventStreamTransmitter = transmitter - await transmitter.transmit().catch((error) => { - if (state.eventStreamTransmitter) { - return this.abortById(id, error) - } - }) + await transmitter.transmit() } else if (request.body instanceof ReadableStream) { const transmitter = new OctetStreamTransmitter(request.body, id, this.send) state.octetStreamTransmitter = transmitter - await transmitter.transmit().catch((error) => { - if (state.octetStreamTransmitter) { - return this.abortById(id, error) - } - }) + await transmitter.transmit() } } catch (reason) { failure = reason - await this.closeById(id, reason) + + if (!state.requestSent) { + await this.closeById(id, reason) + } else if (!state.streamCancelled) { + await this.abortById(id, reason) + } } finally { if (untransmittedBody !== undefined) { await cancelStandardBody(untransmittedBody, failure ?? request.signal?.reason).catch( diff --git a/packages/peer/tests/peer.test.ts b/packages/peer/tests/peer.test.ts index d4f1ef7..c7d4f78 100644 --- a/packages/peer/tests/peer.test.ts +++ b/packages/peer/tests/peer.test.ts @@ -380,6 +380,28 @@ describe('peer integration (client <-> server over encoded wire)', () => { await vi.waitFor(() => expect(serverSignal!.aborted).toBe(true)) }) + it('releases the server request when the client fails to stream a body after sending the request', async () => { + let serverSignal: AbortSignal | undefined + + const { client, server } = connect(async (request) => { + serverSignal = request.signal + // blocks until the upload ends or the request is cancelled + await ((await request.resolveBody()) as ReadableStream).getReader().read() + return { status: 200, headers: {} } + }) + + // a body the caller already locked cannot be streamed once the request message is out + const body = new ReadableStream() + body.getReader() + + await expect( + client.request({ url: '/upload', method: 'POST', headers: {}, body }), + ).rejects.toThrow(TypeError) + + await vi.waitFor(() => expect(serverSignal?.aborted).toBe(true)) + expect((server as any).requests.size).toBe(0) + }) + it('propagates a client abort fired while the request message is still being sent', async () => { const encodeStarted = promiseWithResolvers() const releaseEncode = promiseWithResolvers()