test: make the two retention repairs observable instead of merely correct

The audit's second round showed both resource fixes were mutation-surviving:
restoring the per-chunk accumulator, or the shared Promise.race companion, left
every test green. Neither property is visible through behavior — both shapes
relay and reassemble identically — so each needed its own observable.

bounded-body now counts buffer reallocations for tests. A geometric buffer grows
a handful of times regardless of how the peer fragments the body; an exact-fit
accumulator grows once per chunk, which is the retention shape the repair
removed. The new test compares 20k one-byte chunks against the same body in one
chunk and pins growth to a small constant.

The eager relay's property is structural, so it is pinned structurally, the way
this repository already pins the star-consent guard: neither relay may race a
read against a shared abort promise (comments stripped first — both files
describe the banned shape in prose), and the eager producer must keep the
reader-cancel wake-up that replaced it. Both driven red by restoring the old
implementations.

Author: lidge-jun

src/lib/bounded-body.ts

@@ -39,6 +39,19 @@ export interface BoundedBodyResult {
 const TOTAL_TIMEOUT = Symbol("bounded body total timeout");
 const INACTIVITY_TIMEOUT = Symbol("bounded body inactivity timeout");
 
+/**
+ * Test-only instrumentation: how many times the retained buffer was reallocated
+ * during the most recent read. The accumulator grows geometrically, so this is
+ * logarithmic in the body size and independent of how many chunks the peer sends.
+ * The per-chunk array it replaced retained one object per chunk instead, which a
+ * fragmenting peer can inflate far past the payload ceiling — a property no
+ * correctness assertion can see, which is why it is observable here.
+ */
+let bufferGrowthsForTests = 0;
+export function boundedBodyBufferGrowthsForTests(): number {
+	return bufferGrowthsForTests;
+}
+
 function timeoutPromise(ms: number, value: symbol): { promise: Promise<symbol>; clear: () => void } {
 	let timer: ReturnType<typeof setTimeout> | undefined;
 	const promise = new Promise<symbol>((resolve) => {
@@ -105,6 +118,7 @@ export async function readBoundedResponseBody(
 	// beyond the payload ceiling on large budgets.
 	let retained = new Uint8Array(Math.min(maxBytes, 64 * 1024));
 	let retainedBytes = 0;
+	bufferGrowthsForTests = 0;
 	let mustCancel = false;
 	let cancelReason: unknown;
 	const total = timeoutPromise(options.totalTimeoutMs ?? BOUNDED_BODY_TIMEOUT_MS, TOTAL_TIMEOUT);
@@ -197,6 +211,7 @@ export async function readBoundedResponseBody(
 				);
 				grown.set(retained.subarray(0, retainedBytes));
 				retained = grown;
+				bufferGrowthsForTests += 1;
 			}
 			retained.set(value, retainedBytes);
 			retainedBytes += value.byteLength;

tests/bounded-body.test.ts

@@ -1,6 +1,7 @@
 import { describe, expect, test } from "bun:test";
 import {
 	BOUNDED_BODY_MAX_BYTES,
+	boundedBodyBufferGrowthsForTests,
 	readBoundedResponseBody,
 } from "../src/lib/bounded-body";
 
@@ -162,6 +163,32 @@ describe("readBoundedResponseBody", () => {
 			expect(result.text).toBe("a".repeat(chunkCount));
 			expect(result.oversized).toBe(false);
 		});
+
+		test("retention is logarithmic in the body, not linear in the chunk count", async () => {
+			// The accumulator is the security property, and correctness cannot see it:
+			// a per-chunk array reassembles identically while retaining one object per
+			// transport chunk, which a fragmenting peer inflates far past the payload
+			// ceiling. Growth count is the observable that separates the two — a single
+			// geometric buffer doubles a handful of times regardless of fragmentation.
+			const fine = Array.from({ length: 20_000 }, () => new Uint8Array([0x61]));
+			await readBoundedResponseBody(responseFromChunks(...fine), { maxBytes: CUSTOM_CAP });
+			const fineGrowths = boundedBodyBufferGrowthsForTests();
+
+			const coarse = [new Uint8Array(20_000).fill(0x61)];
+			await readBoundedResponseBody(responseFromChunks(...coarse), { maxBytes: CUSTOM_CAP });
+			const coarseGrowths = boundedBodyBufferGrowthsForTests();
+
+			// 20k one-byte chunks fit inside the 64 KiB seed: no growth at all, and the
+			// same body delivered as one chunk behaves identically.
+			expect(fineGrowths).toBe(coarseGrowths);
+			expect(fineGrowths).toBeLessThanOrEqual(2);
+
+			// Past the seed, growth stays logarithmic: doubling from 64 KiB to 256 KiB is
+			// two reallocations no matter how the peer fragments it.
+			const big = Array.from({ length: 256 }, () => new Uint8Array(1024).fill(0x61));
+			await readBoundedResponseBody(responseFromChunks(...big), { maxBytes: CUSTOM_CAP });
+			expect(boundedBodyBufferGrowthsForTests()).toBeLessThanOrEqual(4);
+		});
 	});
 
 	test("parent abort rejects with the exact reason object", async () => {

tests/relay-eager.test.ts

@@ -111,6 +111,29 @@ async function readAll(stream: ReadableStream<Uint8Array>): Promise<string> {
 }
 
 describe("relaySseEagerBounded — inline payload rewrite (#864)", () => {
+  test("neither relay races reads against a shared abort promise", async () => {
+    // Retention shape, not behavior: racing every read against ONE never-settled
+    // promise attaches a reaction per completed read and holds it until abort, so a
+    // long stream retains O(chunk-count) callbacks. Both relays relay identically
+    // either way, which is exactly why no behavioral assertion catches a regression
+    // here — relay.ts already states the rule in prose at its own drain, and this
+    // pins it for both files. The sanctioned shape is: cancel the reader on abort.
+    const eager = await Bun.file(new URL("../src/server/relay-eager.ts", import.meta.url)).text();
+    const relay = await Bun.file(new URL("../src/server/relay.ts", import.meta.url)).text();
+
+    // Strip comments first: both files DESCRIBE the banned shape in prose, and the
+    // rule is about the code, not the explanation of why the code avoids it.
+    const stripComments = (source: string): string =>
+      source.replace(/\/\*[\s\S]*?\*\//g, "").replace(/(^|[^:])\/\/.*$/gm, "$1");
+
+    for (const [name, source] of [["relay-eager.ts", eager], ["relay.ts", relay]] as const) {
+      const racesReads = /Promise\.race\(\s*\[\s*reader\.read\(\)/.test(stripComments(source));
+      expect(`${name} races reads: ${racesReads}`).toBe(`${name} races reads: false`);
+    }
+    // And the eager producer must keep the reader-cancel wake-up that replaced it.
+    expect(eager).toMatch(/reader\.cancel\(upstream\.signal\.reason\)/);
+  });
+
   test("a terminal frame settling in the same tick as abort is still recorded", async () => {
     // Post-cancel drain: the terminal arrives, and the drain deadline aborts
     // upstream in the same tick. Honoring the signal before examining the settled