fix(peer): send the cancel after an in-flight request message

packages/peer/src/client.test.ts

@@ -208,6 +208,32 @@ describe('clientPeer', () => {
   })
 
   describe('signal / abort', () => {
+    async function abortWhileSendingRequest() {
+      const requestSending = promiseWithResolvers<PeerRequestMessage>()
+      const releaseRequest = promiseWithResolvers<void>()
+      const bodyCancelled = promiseWithResolvers<void>()
+      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))

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(),
     ]

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<void>()
+    const releaseEncode = promiseWithResolvers<void>()
+    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<string, unknown> | undefined