diff --git a/packages/peer/src/client.test.ts b/packages/peer/src/client.test.ts index b0edc650..1c710e95 100644 --- a/packages/peer/src/client.test.ts +++ b/packages/peer/src/client.test.ts @@ -208,6 +208,32 @@ describe('clientPeer', () => { }) describe('signal / abort', () => { + async function abortWhileSendingRequest() { + const requestSending = promiseWithResolvers() + const releaseRequest = promiseWithResolvers() + const bodyCancelled = promiseWithResolvers() + send.mockImplementation(async (message) => { + if (message.kind === 'request') { + requestSending.resolve(message) + await releaseRequest.promise + } + }) + + const controller = new AbortController() + const body = new ReadableStream({ cancel: () => bodyCancelled.resolve() }) + const promise = peer.request(makeRequest({ body, signal: controller.signal })) + const { id } = await requestSending.promise + const error = new Error('aborted during send') + controller.abort(error) + + // the request settles right away, but the cancel must not overtake the request message + await expect(promise).rejects.toBe(error) + expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request']) + + // the unsent body is released last, after any cancel message + return { id, releaseRequest, bodyCancelled: bodyCancelled.promise } + } + it('throws immediately if signal already aborted', async () => { const controller = new AbortController() controller.abort(new Error('pre-aborted')) @@ -239,8 +265,8 @@ describe('clientPeer', () => { controller.abort(new Error('aborted during encode')) await expect(promise).rejects.toThrow('aborted during encode') - // the request message is never sent, only the cancel - expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['cancel']) + // the request message is never sent, so neither is a cancel + expect(send).not.toHaveBeenCalled() }) it('throws when signal aborted during send', async () => { @@ -253,7 +279,7 @@ describe('clientPeer', () => { peer.request(makeRequest({ signal: controller.signal })), ).rejects.toThrow('aborted during send') - expect(send).toHaveBeenCalledTimes(2) + await vi.waitFor(() => expect(send).toHaveBeenCalledTimes(2)) const id = (send.mock.calls[0]![0] as PeerRequestMessage).id expect(send).toHaveBeenNthCalledWith(2, { id, kind: 'cancel' }) }) @@ -275,7 +301,7 @@ describe('clientPeer', () => { await expect(promise).rejects.toThrow(error) await vi.waitFor(() => expect(cancel).toHaveBeenCalledOnce()) expect(cancel).toHaveBeenCalledWith(error) - expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['cancel']) + expect(send).not.toHaveBeenCalled() }) it('returns an event-stream request body when the peer is closed while encoding', async () => { @@ -331,6 +357,24 @@ describe('clientPeer', () => { expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request', 'cancel']) }) + it('sends the cancel message once the in-flight request message is sent', async () => { + const { id, releaseRequest, bodyCancelled } = await abortWhileSendingRequest() + + releaseRequest.resolve() + await bodyCancelled + expect(send).toHaveBeenCalledTimes(2) + expect(send).toHaveBeenNthCalledWith(2, { id, kind: 'cancel' }) + }) + + it('does not send a cancel message when the in-flight request message fails to send', async () => { + const { releaseRequest, bodyCancelled } = await abortWhileSendingRequest() + + // the server never received the request, so there is nothing to cancel there + releaseRequest.reject(new Error('transport down')) + await bodyCancelled + expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request']) + }) + it('rejects pending request on server abort', async () => { const { id, promise } = await requestAndGetId() await peer.message(makeCancelMessage(id)) diff --git a/packages/peer/src/client.ts b/packages/peer/src/client.ts index e6a4d12d..00e2c357 100644 --- a/packages/peer/src/client.ts +++ b/packages/peer/src/client.ts @@ -15,6 +15,11 @@ interface ClientPeerRequestStateInternal { eventStreamTransmitter?: EventStreamTransmitter | undefined octetStreamTransmitter?: OctetStreamTransmitter | undefined removeAbortListener?: (() => void) | undefined + /** + * A cancel must not overtake the request message (the server ignores cancels for unknown ids), + * so until the request message is sent, transmitRequest sends the cancel instead of abortById. + */ + requestSent?: boolean | undefined } export class ClientPeer { @@ -88,9 +93,14 @@ export class ClientPeer { }, binary: encodedAtomicBody.binary, }) + state.requestSent = true // The request can already be settled/cancelled while was in flight if (this.requests.get(id) !== state) { + if (request.signal?.aborted) { + await this.send({ id, kind: 'cancel' }) + } + return } @@ -266,7 +276,7 @@ export class ClientPeer { state.octetStreamMessageQueue = undefined const promises = [ - this.send({ id, kind: 'cancel' }), + state.requestSent ? this.send({ id, kind: 'cancel' }) : undefined, state.eventStreamTransmitter?.cancel(), state.octetStreamTransmitter?.cancel(), ] diff --git a/packages/peer/tests/peer.test.ts b/packages/peer/tests/peer.test.ts index 9b3ec843..e6823fc2 100644 --- a/packages/peer/tests/peer.test.ts +++ b/packages/peer/tests/peer.test.ts @@ -2,6 +2,7 @@ import type { StandardLazyRequest, StandardResponse } from '@standard-server/cor import type { ClientPeer, ServerPeer } from '../src' import type { ClientPeerSendMessage, ServerPeerSendMessage } from '../src/types' import { ErrorEvent, getEventMeta, withEventMeta } from '@standard-server/core' +import { promiseWithResolvers } from '@standard-server/shared' import { ClientPeer as ClientPeerClass, decodePeerMessage, encodePeerMessage, ServerPeer as ServerPeerClass } from '../src' /** @@ -289,6 +290,36 @@ describe('peer integration (client <-> server over encoded wire)', () => { await vi.waitFor(() => expect(serverSignal!.aborted).toBe(true)) }) + it('propagates a client abort fired while the request message is still being sent', async () => { + const encodeStarted = promiseWithResolvers() + const releaseEncode = promiseWithResolvers() + let serverSignal: AbortSignal | undefined + + const { client } = connect(async (request) => { + serverSignal = request.signal + return new Promise(() => {}) // handler never resolves + }) + + // encoding the request message awaits `file.arrayBuffer()`, so the cancel message could overtake it + const file = new File(['hello'], 'hello.txt', { type: 'text/plain' }) + const arrayBuffer = file.arrayBuffer.bind(file) + vi.spyOn(file, 'arrayBuffer').mockImplementation(async () => { + encodeStarted.resolve() + await releaseEncode.promise + return arrayBuffer() + }) + + const controller = new AbortController() + const promise = client.request({ url: '/upload', method: 'POST', headers: {}, body: file, signal: controller.signal }) + + await encodeStarted.promise + controller.abort(new Error('user navigated away')) + await expect(promise).rejects.toThrow('user navigated away') + + releaseEncode.resolve() + await vi.waitFor(() => expect(serverSignal?.aborted).toBe(true)) + }) + it('carries malicious __proto__ payloads as inert data without polluting prototypes', async () => { let serverBody: Record | undefined