Body lifecycle: empty chunks, multipart boundary quoting, logging tap after close (#77)

packages/core/src/body/multipart-body.test.ts

@@ -1,8 +1,9 @@
 // SPDX-License-Identifier: MIT
 // packages/core/src/body/multipart-body.test.ts
 // Exercises: BODY-2 (composite replayability, unknown-length collapse), HTTP-51 (shared framing routine,
-// boundary generation/validation, header quoting, and a part media type that cannot break the framing),
-// HTTP-26 (a media type is header-safe), RECOV-12 (a close failure never masks the primary failure)
+// boundary generation/validation, header quoting, a boundary parameter rendered so an RFC 9110 parser
+// can read it, and a part media type that cannot break the framing), HTTP-26 (a media type is
+// header-safe), RECOV-12 (a close failure never masks the primary failure)
 import {describe, expect, test} from 'bun:test';
 import fc from 'fast-check';
 import {MediaTypeParseError} from '../http/errors.js';
@@ -66,9 +67,11 @@ function collectingSink(): {
   };
 }
 
-async function drain(body: {
+// `Uint8Array<ArrayBuffer>`, not the bare alias: `BodyInit` excludes a view over a `SharedArrayBuffer`,
+// so the default `ArrayBufferLike` parameter is not assignable to the platform `Response` below.
+async function drainBytes(body: {
   writeTo: (sink: WritableStream<Uint8Array>) => Promise<void>;
-}): Promise<string> {
+}): Promise<Uint8Array<ArrayBuffer>> {
   const chunks: Uint8Array[] = [];
   await body.writeTo(new WritableStream({write: c => void chunks.push(c)}));
   const total = chunks.reduce((s, c) => s + c.length, 0);
@@ -78,7 +81,13 @@ async function drain(body: {
     out.set(c, offset);
     offset += c.length;
   }
-  return new TextDecoder().decode(out);
+  return out;
+}
+
+async function drain(body: {
+  writeTo: (sink: WritableStream<Uint8Array>) => Promise<void>;
+}): Promise<string> {
+  return new TextDecoder().decode(await drainBytes(body));
 }
 
 describe('MultipartBody replayability and length (BODY-2)', () => {
@@ -186,6 +195,47 @@ describe('MultipartBody boundary generation and validation (HTTP-51)', () => {
   });
 });
 
+describe('the rendered boundary parameter is a parseable one (HTTP-51)', () => {
+  // RFC 2046 `bchars` and RFC 9110 `tchar` are different sets: ' ', ',', ':', '=', '?', '/', '(' and
+  // ')' are legal in a boundary and illegal bare in a header parameter value. `validateBoundary` admits
+  // the first grammar and the renderer owes the second, so a boundary the constructor accepts must come
+  // back out quoted rather than bare.
+  test('a boundary that is not a bare token is quoted', () => {
+    expect(
+      multipartBody([{name: 'a', body: stringBody('x')}], 'a,b').mediaType,
+    ).toBe('multipart/form-data; boundary="a,b"');
+  });
+
+  test('a boundary that IS a bare token is left unquoted', () => {
+    expect(
+      multipartBody([{name: 'a', body: stringBody('x')}], 'plain-1').mediaType,
+    ).toBe('multipart/form-data; boundary=plain-1');
+    // The generated default stays byte-identical: it is drawn from ALPHA/DIGIT only.
+    expect(
+      multipartBody([{name: 'a', body: stringBody('x')}]).mediaType,
+    ).toMatch(/^multipart\/form-data; boundary=dexpace-[A-Za-z0-9]{32}$/);
+  });
+
+  test.each([['a,b'], ['bound ary'], ['a:b'], ['a=b'], ['a?b'], ['(a)/b']])(
+    'the header a peer receives round-trips through a real parameter parser: %p',
+    async boundary => {
+      // The runtime's own multipart parser, standing in for the peer. Bun's happens to tolerate the
+      // unquoted form, so this is the regression guard and NOT the reproducer: Node's (undici's)
+      // rejects the whole body with `TypeError: Failed to parse body as FormData`, which is why the
+      // same case is also in `tests/node-conformance/body-lifecycle.test.mjs`. Two independent
+      // parsers disagreeing about our own Content-Type is exactly what that tree is for.
+      const body = multipartBody(
+        [{name: 'field', body: stringBody('value')}],
+        boundary,
+      );
+      const response = new globalThis.Response(await drainBytes(body), {
+        headers: {'content-type': body.mediaType},
+      });
+      expect((await response.formData()).get('field')).toBe('value');
+    },
+  );
+});
+
 describe('MultipartBodyBuilder (HTTP-2, HTTP-3)', () => {
   test('static newBuilder and instance newBuilder pre-populates parts and boundary', async () => {
     const original = MultipartBody.newBuilder()

packages/core/src/body/multipart-body.ts

@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: MIT
 // packages/core/src/body/multipart-body.ts
 import type {Builder} from '../http/builder.js';
+import {MediaType} from '../http/media-type.js';
 import {EndOfStreamError} from '../io/errors.js';
 import {invariant} from '../invariant.js';
 import type {Body} from './body.js';
@@ -77,6 +78,31 @@ function renderPartHeader(part: MultipartPart, boundary: string): Uint8Array {
   return new TextEncoder().encode(header);
 }
 
+/**
+ * HTTP-51: the `Content-Type` a peer actually parses.
+ *
+ * RFC 2046 `bchars` and RFC 9110 `tchar` are different sets. `BOUNDARY_PATTERN` above admits ' ', ',',
+ * ':', '=', '?', '/', '(' and ')', none of which is a `tchar`, so interpolating the boundary bare
+ * produces a parameter value that stops at the first offending byte -- `boundary=a,b` reads as
+ * `boundary=a` plus a junk parameter, and the peer then never finds a delimiter. Node's own FormData
+ * parser rejects such a body outright with `TypeError: Failed to parse body as FormData`.
+ *
+ * Rendered through {@link MediaType} rather than a second quoting routine here: it is the module that
+ * owns HTTP-25's token-or-quoted-string decision, and `parse(render(x)) === x` is its guarantee. A
+ * boundary that IS a bare token still renders bare, so the generated default is byte-identical to what
+ * this class emitted before.
+ *
+ * Narrowing `validateBoundary` to `tchar` instead was rejected: HTTP-51 asks that a boundary VIOLATING
+ * the RFC 2046 grammar be refused, not that a conforming one be. The defect is in the rendering.
+ */
+function renderMediaType(boundary: string): string {
+  return MediaType.of(
+    'multipart',
+    'form-data',
+    new Map([['boundary', boundary]]),
+  ).render();
+}
+
 function trailerBytes(boundary: string): Uint8Array {
   return new TextEncoder().encode(`--${boundary}--\r\n`);
 }
@@ -144,7 +170,10 @@ function nonClosingSink(
 export class MultipartBody implements Body {
   /** Discriminates this variant within the {@link Body} union. */
   readonly kind = 'multipart' as const;
-  /** `multipart/form-data` carrying the boundary this instance frames its parts with. */
+  /**
+   * `multipart/form-data` carrying the boundary this instance frames its parts with, with the
+   * `boundary` parameter quoted whenever it is not a bare RFC 9110 token (HTTP-51).
+   */
   readonly mediaType: string;
   /** The total framed byte count, or -1 when any part's own length is unknown (BODY-2). */
   readonly contentLength: number;
@@ -157,7 +186,7 @@ export class MultipartBody implements Body {
     if (boundary !== undefined) validateBoundary(boundary);
     this.#boundary = boundary ?? generateBoundary();
     this.#parts = [...parts];
-    this.mediaType = `multipart/form-data; boundary=${this.#boundary}`;
+    this.mediaType = renderMediaType(this.#boundary);
     this.replayable = this.#parts.every(part => part.body.replayable);
     this.contentLength = computeContentLength(this.#parts, this.#boundary);
     invariant(

packages/core/src/body/response-body-logging.test.ts

@@ -4,11 +4,16 @@
 // reads), BODY-24 (exceeds-cap: prefix+tail once, second read fails), BODY-26 (drain failure cached,
 // partial bytes retained, error() does not drain), BODY-27 (close-once shared guard), BODY-28 (captured
 // buffer survives close), BODY-29 (reported length), BODY-32 (negative cap rejected), BODY-25 (a
-// zero-length delegate chunk is a stream-contract violation, never end-of-stream)
+// zero-length delegate chunk is a stream-contract violation, never end-of-stream), BODY-27/BODY-28 again
+// (close() ends the drain rather than poisoning the wrapper: snapshot still serves the captured prefix,
+// read() reports IO-42's state error, error() reports only a genuine drain failure)
 import {describe, expect, test} from 'bun:test';
 import fc from 'fast-check';
 import {InvariantViolation} from '../invariant.js';
-import {SourceContractViolationError} from '../io/errors.js';
+import {
+  ClosedResourceError,
+  SourceContractViolationError,
+} from '../io/errors.js';
 import {withResponseLogging} from './response-body-logging.js';
 
 function readableOf(...chunks: number[][]): ReadableStream<Uint8Array> {
@@ -282,6 +287,87 @@ describe('the tail path enforces the same chunk contract (BODY-25)', () => {
   });
 });
 
+describe('the tap is inert after close(), not poisoned by it (BODY-27, BODY-28)', () => {
+  // closeDelegate releases the reader. Every entry point that used to start a drain unconditionally then
+  // read from a detached reader, so `snapshot()` cached a raw `TypeError: Invalid state: The reader is
+  // not attached to a stream` as the wrapper's failure -- and `error()` reported that forever, over a
+  // capture that never failed. BODY-28 says the captured bytes survive close; they cannot survive it
+  // behind a fabricated error.
+
+  test('close-then-snapshot returns the captured prefix and starts no drain', async () => {
+    const logged = withResponseLogging(readableOf([1, 2], [3, 4]), 2);
+    await logged.read(); // exceeds-cap: the prefix is captured, the delegate stays live
+    await logged.close();
+
+    expect([...logged.snapshot()]).toEqual([1, 2]);
+    await new Promise(resolve => setTimeout(resolve, 0)); // a drain started here would settle by now
+    expect(logged.error()).toBeNull();
+  });
+
+  test('close-before-any-read leaves snapshot empty and error() null', async () => {
+    const logged = withResponseLogging(readableOf([1, 2, 3]), 100);
+    await logged.close();
+
+    expect([...logged.snapshot()]).toEqual([]);
+    await new Promise(resolve => setTimeout(resolve, 0));
+    expect(logged.error()).toBeNull();
+  });
+
+  test('close-then-read rejects with ClosedResourceError, not a detached-reader TypeError', async () => {
+    const logged = withResponseLogging(readableOf([1, 2, 3]), 100);
+    await logged.close();
+
+    const error = await rejection(logged.read());
+    expect(error).toBeInstanceOf(ClosedResourceError);
+    expect(error.message).toBe('LoggedResponseBody is closed');
+  });
+
+  test('close-then-read in the exceeds-cap regime rejects too -- there is no live tail left', async () => {
+    // The drain stopped at the cap and nobody took the tail, so the captured prefix is NOT the whole
+    // body. Serving it would hand the consumer a silently truncated response; the delegate that held
+    // the rest is gone.
+    const logged = withResponseLogging(readableOf([1, 2], [3, 4]), 2);
+    logged.snapshot(); // starts the drain without taking the tail
+    await new Promise(resolve => setTimeout(resolve, 0));
+    expect([...logged.snapshot()]).toEqual([1, 2]);
+    await logged.close();
+
+    expect(await rejection(logged.read())).toBeInstanceOf(ClosedResourceError);
+  });
+
+  test('close-then-error reports only a genuine drain failure (BODY-26)', async () => {
+    const boom = new Error('upstream reset');
+    const failing = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(Uint8Array.from([1, 2]));
+      },
+      pull(controller) {
+        controller.error(boom);
+      },
+    });
+    const logged = withResponseLogging(failing, 100);
+    expect(await rejection(logged.read())).toBe(boom);
+    // `cancel()` on an errored stream rejects with that stream's own stored error, which the
+    // 'a non-TypeError from cancel() propagates' case above already pins. Not what this test is about.
+    await logged.close().catch(() => undefined);
+
+    // The real failure is not displaced by a close-induced one, and snapshot still shows the partial
+    // capture BODY-26 asked to be retained.
+    expect(logged.error()).toBe(boom);
+    expect([...logged.snapshot()]).toEqual([1, 2]);
+  });
+
+  test('the fits-cap regime still serves repeatable reads after its own close (BODY-23, BODY-28)', async () => {
+    // The drain itself closes the delegate on this path, so "closed" must NOT mean "unreadable".
+    const logged = withResponseLogging(readableOf([1, 2, 3]), 100);
+    expect([...(await readAll(await logged.read()))]).toEqual([1, 2, 3]);
+    await logged.close();
+    expect([...(await readAll(await logged.read()))]).toEqual([1, 2, 3]);
+    expect([...logged.snapshot()]).toEqual([1, 2, 3]);
+    expect(logged.error()).toBeNull();
+  });
+});
+
 describe('snapshot is a drain trigger (BODY-22)', () => {
   test('calling snapshot starts the drain, without a read()', async () => {
     const logged = withResponseLogging(readableOf([1, 2, 3]), 100);

packages/core/src/body/response-body-logging.ts

@@ -2,7 +2,10 @@
 // packages/core/src/body/response-body-logging.ts
 import {invariant} from '../invariant.js';
 import {ByteQueue} from '../io/byte-queue.js';
-import {SourceContractViolationError} from '../io/errors.js';
+import {
+  ClosedResourceError,
+  SourceContractViolationError,
+} from '../io/errors.js';
 import {MAX_BYTE_ARRAY_LENGTH} from '../io/limits.js';
 import {ConsumedBodyError} from './errors.js';
 
@@ -20,11 +23,19 @@ export interface LoggedResponseBody {
    * (BODY-22). Fits-cap regime: every call, including calls after the first, returns a fresh
    * non-consuming view over the captured bytes (BODY-23). Exceeds-cap regime: exactly one call is
    * allowed; a second throws (BODY-24). If the drain failed, every call re-throws the cached error.
+   * After `close()` in any regime but fits-cap, throws `ClosedResourceError`: the delegate is gone and
+   * the captured prefix is not the whole body (BODY-27, BODY-28).
    */
   read(): Promise<ReadableStream<Uint8Array>>;
-  /** Non-consuming; reflects whatever has been captured so far, even after a failed drain (BODY-26). */
+  /**
+   * Non-consuming; reflects whatever has been captured so far, even after a failed drain (BODY-26) and
+   * after `close()` (BODY-28), which it never restarts a drain past.
+   */
   snapshot(): Uint8Array;
-  /** The cached drain failure, or null. MUST NOT trigger a drain (BODY-26). */
+  /**
+   * The cached drain failure, or null. MUST NOT trigger a drain (BODY-26), and reports only a genuine
+   * upstream failure -- never one manufactured by reading past `close()`.
+   */
   error(): Error | null;
   /** Captured size iff fully captured within the cap, else the delegate's declared length (BODY-29). */
   readonly contentLength: number;
@@ -129,8 +140,16 @@ async function drainOnce(state: DrainState): Promise<void> {
  * The detached `.catch` matters: a snapshot-triggered drain has no awaiter, so without it a drain failure
  * becomes an unhandled rejection. Attaching a handler to a *copy* leaves the stored promise rejected, so
  * `read()` still re-throws the cached failure on every call (BODY-26).
+ *
+ * BODY-28: after `close()` there is nothing left to drain -- `closeDelegate` released the reader, so
+ * starting one here reads from a detached reader, raises a raw `TypeError: Invalid state`, and
+ * `drainOnce`'s catch caches it as this wrapper's `failure`. `error()` would then report a fabricated
+ * upstream failure forever, over a capture that never failed, and the captured bytes BODY-28 promises
+ * survive close would be reachable only past that lie. A drain already in flight is left alone: on the
+ * fits-cap path the drain closes the delegate itself, and its own promise is what `read()` awaits.
  */
 function startDrain(state: DrainState): Promise<void> {
+  if (state.closed && state.started === undefined) return Promise.resolve();
   state.started ??= drainOnce(state);
   void state.started.catch(() => undefined);
   return state.started;
@@ -226,10 +245,17 @@ export function withResponseLogging(
   return {
     async read(): Promise<ReadableStream<Uint8Array>> {
       await startDrain(state); // a cached failure re-throws here on every call (BODY-26)
+      // Ordered deliberately. `fits` first: on that path the drain closed the delegate itself, and
+      // BODY-23 still requires every later read to be a fresh non-consuming view -- "closed" there does
+      // not mean "unreadable" (BODY-28).
       if (state.regime === 'fits') return capturedStream(state);
       if (state.tailConsumed) {
         throw new ConsumedBodyError('logged-response');
       }
+      // Anything else with the delegate gone: there is no live tail to continue from, and the captured
+      // prefix is not the whole body, so serving it would hand the consumer a silently truncated
+      // response. IO-42's state error, not the raw `TypeError` a detached reader throws.
+      if (state.closed) throw new ClosedResourceError('LoggedResponseBody');
       state.tailConsumed = true;
       return tailStream(state);
     },
@@ -238,6 +264,7 @@ export function withResponseLogging(
       // so it starts the drain and returns what has been captured so far rather than awaiting it; a
       // later read() awaits the very same in-flight promise, so the delegate is still read exactly once.
       // (BODY-26's "snapshot returns the partial bytes without throwing" is why it cannot await here.)
+      // After close() `startDrain` is a no-op, so this is the post-mortem accessor BODY-28 asks for.
       void startDrain(state);
       return state.captured.snapshot();
     },