http-transport: fix fixed-length response backpressure and abort handling

packages/transports/src/http-server/node.ts

@@ -304,28 +304,61 @@ async function handleResponseBody(
   await handleChunkedStream(res, response.body)
 }
 
-async function handleFixedLengthStream(
+// uWS honors only the first onWritable registration per response, so a single
+// handler dispatches drain (and abort) events to the pending waiter
+function createWritableWaiter(res: UwsResponse): () => Promise<void> {
+  let writableHandlerRegistered = false
+  return () =>
+    new Promise<void>((resolve, reject) => {
+      if (!writableHandlerRegistered) {
+        writableHandlerRegistered = true
+        res.onWritable(() => {
+          res.wakeWritable?.()
+          return true
+        })
+      }
+      res.wakeWritable = () => {
+        res.wakeWritable = undefined
+        if (res.aborted) reject(new Error('Response aborted'))
+        else resolve()
+      }
+    })
+}
+
+// exported for tests: the waiter dispatch is timing-sensitive and needs
+// deterministic coverage against a controlled response double
+export async function handleFixedLengthStream(
   res: UwsResponse,
   body: ReadableStream<Uint8Array>,
   totalSize: number,
 ): Promise<void> {
   const reader = body.getReader()
+  // abort must also cancel a pending read(): a stalled source would otherwise
+  // keep the pump and the reader lock alive forever
+  res.cancelBody = () => {
+    reader.cancel().catch(() => {})
+  }
+  const waitWritable = createWritableWaiter(res)
   try {
     if (totalSize === 0) {
       if (!res.aborted) res.cork(() => res.endWithoutBody(0))
       return
     }
 
-    let responded = false
-    while (!res.aborted && !responded) {
+    while (!res.aborted) {
       const { done, value } = await reader.read()
       if (done) break
       if (value.byteLength === 0) continue
-      responded = await handleFixedChunk(res, value, totalSize)
+      if (await handleFixedChunk(res, value, totalSize, waitWritable)) return
     }
 
-    if (!responded && !res.aborted) res.cork(() => res.close())
+    if (!res.aborted) res.cork(() => res.close())
   } finally {
+    res.cancelBody = undefined
+    // non-abort early exits (zero-length response, tryEnd done while the
+    // source is still open) must also cancel the source so its resources are
+    // released; no-op when the stream already closed or was cancelled on abort
+    reader.cancel().catch(() => {})
     reader.releaseLock()
   }
 }
@@ -342,24 +375,7 @@ export async function handleChunkedStream(
   res.cancelBody = () => {
     reader.cancel().catch(() => {})
   }
-  // uWS honors only the first onWritable registration per response, so a
-  // single handler dispatches drain (and abort) events to the pending waiter
-  let writableHandlerRegistered = false
-  const waitWritable = () =>
-    new Promise<void>((resolve, reject) => {
-      if (!writableHandlerRegistered) {
-        writableHandlerRegistered = true
-        res.onWritable(() => {
-          res.wakeWritable?.()
-          return true
-        })
-      }
-      res.wakeWritable = () => {
-        res.wakeWritable = undefined
-        if (res.aborted) reject(new Error('Response aborted'))
-        else resolve()
-      }
-    })
+  const waitWritable = createWritableWaiter(res)
   try {
     while (!res.aborted) {
       const { done, value } = await reader.read()
@@ -381,41 +397,33 @@ export async function handleChunkedStream(
   }
 }
 
-function handleFixedChunk(
+async function handleFixedChunk(
   res: UwsResponse,
   chunk: Uint8Array,
   totalSize: number,
+  waitWritable: () => Promise<void>,
 ): Promise<boolean> {
-  return new Promise((resolve, reject) => {
-    const chunkOffset = res.getWriteOffset()
-
-    const write = (offset: number) => {
-      if (res.aborted) {
-        reject(new Error('Response aborted'))
-        return false
-      }
-
-      const relativeOffset = offset - chunkOffset
-      const remaining =
-        relativeOffset > 0 ? chunk.subarray(relativeOffset) : chunk
-
-      let ok = false
-      let done = false
-      res.cork(() => {
-        ;[ok, done] = res.tryEnd(remaining, totalSize)
-      })
-
-      if (done || ok) {
-        resolve(done)
-        return ok
-      }
-
-      res.onWritable(write)
-      return ok
-    }
+  const chunkOffset = res.getWriteOffset()
+
+  while (true) {
+    if (res.aborted) throw new Error('Response aborted')
+
+    // on retry after a drain the write offset tells how much of this chunk
+    // uWS already buffered
+    const relativeOffset = res.getWriteOffset() - chunkOffset
+    const remaining =
+      relativeOffset > 0 ? chunk.subarray(relativeOffset) : chunk
+
+    // cork() returns the response object, not tryEnd()'s flags
+    let ok = false
+    let done = false
+    res.cork(() => {
+      ;[ok, done] = res.tryEnd(remaining, totalSize)
+    })
 
-    write(chunkOffset)
-  })
+    if (done || ok) return done
+    await waitWritable()
+  }
 }
 
 function getContentLength(headers: Headers): number | undefined {

packages/transports/tests/http-server/fixed-length-stream.spec.ts

@@ -0,0 +1,166 @@
+import { describe, expect, it } from 'vitest'
+
+import { handleFixedLengthStream } from '../../src/http-server/node.ts'
+
+describe('node host', () => {
+  describe('handleFixedLengthStream writable dispatcher', () => {
+    // Scripted stand-in for uWS HttpResponse: only the first onWritable
+    // registration takes effect, and partial writes advance the global offset.
+    function createResDouble(
+      script: Array<{ accept: number; ok: boolean; done: boolean }>,
+    ) {
+      const steps = [...script]
+      let offset = 0
+      let registrations = 0
+      let closes = 0
+      let handler: ((offset: number) => boolean) | undefined
+      const written: number[] = []
+      const res: any = {
+        aborted: false,
+        wakeWritable: undefined,
+        cork(cb: () => void) {
+          cb()
+          return res
+        },
+        getWriteOffset: () => offset,
+        tryEnd(data: Uint8Array, _totalSize: number) {
+          const step = steps.shift() ?? {
+            accept: data.byteLength,
+            ok: true,
+            done: false,
+          }
+          written.push(...data.subarray(0, step.accept))
+          offset += step.accept
+          return [step.ok, step.done]
+        },
+        close() {
+          closes++
+          return res
+        },
+        endWithoutBody() {
+          return res
+        },
+        onWritable(h: (offset: number) => boolean) {
+          registrations++
+          handler ??= h
+          return res
+        },
+      }
+      return {
+        res,
+        fireWritable: () => handler!(offset),
+        counts: () => ({ registrations, closes }),
+        written: () => written,
+        // Mirrors the route's onAborted handler so pending waits and reads
+        // observe the same cancellation order as production.
+        abort: () => {
+          res.aborted = true
+          res.wakeWritable?.()
+          res.cancelBody?.()
+        },
+      }
+    }
+
+    const streamOf = (...parts: Uint8Array[]) =>
+      new ReadableStream<Uint8Array>({
+        start(controller) {
+          for (const part of parts) controller.enqueue(part)
+          controller.close()
+        },
+      })
+
+    const tick = () => new Promise((resolve) => setImmediate(resolve))
+
+    it('drains two partially written chunks with one registration and current offsets', async () => {
+      const double = createResDouble([
+        { accept: 3, ok: false, done: false },
+        { accept: 5, ok: true, done: false },
+        { accept: 4, ok: false, done: false },
+        { accept: 4, ok: true, done: true },
+      ])
+      const chunk1 = Uint8Array.from({ length: 8 }, (_, i) => i)
+      const chunk2 = Uint8Array.from({ length: 8 }, (_, i) => i + 8)
+      const pump = handleFixedLengthStream(
+        double.res,
+        streamOf(chunk1, chunk2),
+        16,
+      )
+
+      await tick()
+      expect(double.counts().registrations).toBe(1)
+
+      double.fireWritable()
+      await tick()
+      // Chunk 1 finished and chunk 2 parked without replacing uWS's handler.
+      expect(double.counts().registrations).toBe(1)
+
+      double.fireWritable()
+      await pump
+      expect(double.counts()).toEqual({ registrations: 1, closes: 0 })
+      expect(double.written()).toEqual(Array.from({ length: 16 }, (_, i) => i))
+    })
+
+    it('settles a pending writable waiter on abort', async () => {
+      const double = createResDouble([{ accept: 0, ok: false, done: false }])
+      const body = streamOf(new Uint8Array(8), new Uint8Array(8))
+      const pump = handleFixedLengthStream(double.res, body, 16)
+
+      await tick()
+      expect(double.counts().registrations).toBe(1)
+
+      double.abort()
+      await expect(pump).rejects.toThrow('Response aborted')
+      expect(() => body.getReader()).not.toThrow()
+    })
+
+    it('cancels a stalled read on abort', async () => {
+      const double = createResDouble([])
+      let cancelled = false
+      const body = new ReadableStream<Uint8Array>({
+        // The source intentionally never produces so the pump parks in read().
+        pull: () => new Promise<never>(() => {}),
+        cancel() {
+          cancelled = true
+        },
+      })
+      const pump = handleFixedLengthStream(double.res, body, 16)
+
+      await tick()
+      double.abort()
+      await pump
+      expect(cancelled).toBe(true)
+      expect(double.counts()).toEqual({ registrations: 0, closes: 0 })
+    })
+
+    it('cancels the source for a zero-length response', async () => {
+      const double = createResDouble([])
+      let cancelled = false
+      const body = new ReadableStream<Uint8Array>({
+        cancel() {
+          cancelled = true
+        },
+      })
+
+      await handleFixedLengthStream(double.res, body, 0)
+      expect(cancelled).toBe(true)
+      expect(() => body.getReader()).not.toThrow()
+    })
+
+    it('cancels the source when the response finishes before the stream', async () => {
+      const double = createResDouble([{ accept: 8, ok: true, done: true }])
+      let cancelled = false
+      const body = new ReadableStream<Uint8Array>({
+        start(controller) {
+          controller.enqueue(new Uint8Array(8))
+        },
+        cancel() {
+          cancelled = true
+        },
+      })
+
+      await handleFixedLengthStream(double.res, body, 8)
+      expect(cancelled).toBe(true)
+      expect(() => body.getReader()).not.toThrow()
+    })
+  })
+})

packages/transports/tests/neemata/http/node-runtime.spec.ts

@@ -189,4 +189,45 @@ describe('node runtime adapter', () => {
       }
     })
   })
+
+  describe('fixed-length stream backpressure', () => {
+    it('pauses reading the source stream when the client does not consume', async () => {
+      const chunkSize = 256 * 1024
+      const totalChunks = 100
+      const chunk = new Uint8Array(chunkSize).fill(97)
+      let pulled = 0
+
+      const { url, stop } = await startServer(async () => {
+        // Explicit content-length exercises uWS's fixed-length tryEnd path.
+        const stream = new ReadableStream<Uint8Array>({
+          pull(controller) {
+            pulled++
+            controller.enqueue(chunk)
+            if (pulled >= totalChunks) controller.close()
+          },
+        })
+        return new Response(stream, {
+          headers: { 'content-length': String(chunkSize * totalChunks) },
+        })
+      })
+
+      try {
+        const response = await fetch(`${url}/test`)
+        expect(response.status).toBe(200)
+        expect(response.headers.get('content-length')).toBe(
+          String(chunkSize * totalChunks),
+        )
+
+        // Leave the body unread long enough for the server buffer to fill.
+        await delay(500)
+        expect(pulled).toBeLessThan(totalChunks)
+
+        const body = new Uint8Array(await response.arrayBuffer())
+        expect(pulled).toBe(totalChunks)
+        expect(body.byteLength).toBe(chunkSize * totalChunks)
+      } finally {
+        await stop()
+      }
+    })
+  })
 })