test(js): assert retention on reactions and listeners, not a GC heap count

js/binary/src/stream/stream.test.ts

@@ -1,5 +1,4 @@
-import { heapStats } from "bun:jsc";
-import { expect, test } from "bun:test";
+import { expect, spyOn, test } from "bun:test";
 import { DEFAULT_MAX_FRAME_SIZE } from "@moq/flate";
 import { Time, Track } from "@moq/net";
 import { Consumer, Producer, Rolled } from "./index.ts";
@@ -148,30 +147,40 @@ test("an undecodable payload ends the log for a reader already inside the group"
 	await expect(consumer.next()).rejects.toThrow("limit");
 });
 
+// Counts the reactions `run` attaches to promises still pending once it returns. A promise holds each
+// reaction until it settles, so one left per iteration on a promise that outlives the loop is a leak.
+// Recorded by hand: Bun's `mock.contexts` misses the engine's own calls from `Promise.race`.
+async function pendingReactions(run: () => Promise<void>): Promise<number> {
+	const reacted: Promise<unknown>[] = [];
+	const then = Promise.prototype.then;
+	const spy = spyOn(Promise.prototype, "then").mockImplementation(function (this: Promise<unknown>, ...args) {
+		reacted.push(this);
+		return then.apply(this, args);
+	} as typeof then);
+	try {
+		await run();
+	} finally {
+		spy.mockRestore();
+	}
+	return reacted.filter((promise) => Bun.peek.status(promise) === "pending").length;
+}
+
 // A blocked read races the frame against the track's next group, which stays pending for the whole
 // log. Racing it per payload must not leave a reaction behind on it each time.
 test("blocked reads leave nothing behind on the pending group read", async () => {
 	const track = new Track.Producer("test");
 	const producer = new Producer({ track });
 	const subscriber = track.subscribe();
 	const consumer = new Consumer({ track: subscriber });
-	const promises = () => {
-		Bun.gc(true);
-		return heapStats().objectTypeCounts.Promise ?? 0;
-	};
 
-	const read = async (from: number, count: number) => {
-		for (let n = from; n < from + count; n++) {
+	const reactions = await pendingReactions(async () => {
+		for (let n = 0; n < 1000; n++) {
 			const next = consumer.next();
 			producer.append(new Uint8Array([n & 0xff]));
 			expect((await next)?.[0]).toBe(n & 0xff);
 		}
-	};
-
-	await read(0, 50);
-	const before = promises();
-	await read(50, 1000);
-	expect(promises() - before).toBeLessThan(100);
+	});
+	expect(reactions).toBeLessThan(10);
 
 	subscriber.close();
 	producer.finish();

js/json/src/stream/stream.test.ts

@@ -1,5 +1,4 @@
-import { heapStats } from "bun:jsc";
-import { expect, test } from "bun:test";
+import { expect, spyOn, test } from "bun:test";
 import { Time, Track } from "@moq/net";
 import { Consumer, Producer, Rolled } from "./index.ts";
 
@@ -112,30 +111,40 @@ test("a second concurrent read is refused rather than served the first one's gro
 	expect(await first).toEqual({ n: 0 });
 });
 
+// Counts the reactions `run` attaches to promises still pending once it returns. A promise holds each
+// reaction until it settles, so one left per iteration on a promise that outlives the loop is a leak.
+// Recorded by hand: Bun's `mock.contexts` misses the engine's own calls from `Promise.race`.
+async function pendingReactions(run: () => Promise<void>): Promise<number> {
+	const reacted: Promise<unknown>[] = [];
+	const then = Promise.prototype.then;
+	const spy = spyOn(Promise.prototype, "then").mockImplementation(function (this: Promise<unknown>, ...args) {
+		reacted.push(this);
+		return then.apply(this, args);
+	} as typeof then);
+	try {
+		await run();
+	} finally {
+		spy.mockRestore();
+	}
+	return reacted.filter((promise) => Bun.peek.status(promise) === "pending").length;
+}
+
 // A blocked read races the frame against the track's next group, which stays pending for the whole
 // log. Racing it per record must not leave a reaction behind on it each time.
 test("blocked reads leave nothing behind on the pending group read", async () => {
 	const track = new Track.Producer("test");
 	const producer = new Producer<Rec>({ track });
 	const subscriber = track.subscribe();
 	const consumer = new Consumer<Rec>({ track: subscriber });
-	const promises = () => {
-		Bun.gc(true);
-		return heapStats().objectTypeCounts.Promise ?? 0;
-	};
 
-	const read = async (from: number, count: number) => {
-		for (let n = from; n < from + count; n++) {
+	const reactions = await pendingReactions(async () => {
+		for (let n = 0; n < 1000; n++) {
 			const next = consumer.next();
 			producer.append({ n });
 			expect((await next)?.n).toBe(n);
 		}
-	};
-
-	await read(0, 50);
-	const before = promises();
-	await read(50, 1000);
-	expect(promises() - before).toBeLessThan(100);
+	});
+	expect(reactions).toBeLessThan(10);
 
 	subscriber.close();
 	producer.finish();

js/watch/src/retention.test.ts

@@ -1,16 +1,83 @@
-import { heapStats } from "bun:jsc";
-import { expect, test } from "bun:test";
+import { expect, spyOn, test } from "bun:test";
 import { Container } from "@moq/hang";
 import * as Moq from "@moq/net";
 import { Time } from "@moq/net";
-import { Effect } from "@moq/signals";
+import { Effect, Signal } from "@moq/signals";
 import { nextMedia, subscribeMedia } from "./media";
 import { Sync } from "./sync";
 
+// These count what is still attached rather than the heap: `Bun.gc` scans the stack conservatively,
+// so a stale pointer can pin thousands of dead cells and fail a heap count under load.
+
+// Reactions `run` attaches to promises still pending once it returns, which each hold until they
+// settle. Recorded by hand: Bun's `mock.contexts` misses the engine's own calls from `Promise.race`.
+async function pendingReactions(run: () => Promise<void>): Promise<number> {
+	const reacted: Promise<unknown>[] = [];
+	const then = Promise.prototype.then;
+	const spy = spyOn(Promise.prototype, "then").mockImplementation(function (this: Promise<unknown>, ...args) {
+		reacted.push(this);
+		return then.apply(this, args);
+	} as typeof then);
+	try {
+		await run();
+	} finally {
+		spy.mockRestore();
+	}
+	return reacted.filter((promise) => Bun.peek.status(promise) === "pending").length;
+}
+
+// Signal listeners `run` registers that neither fired nor were disposed by the time it returns.
+async function pendingListeners(run: () => Promise<void>): Promise<number> {
+	const listening = new Set<object>();
+	const changed = Signal.prototype.changed;
+	const spy = spyOn(Signal.prototype, "changed").mockImplementation(function (
+		this: Signal<unknown>,
+		fn?: (value: unknown) => void,
+	) {
+		if (!fn) return (changed as () => Promise<unknown>).call(this);
+		const token = {};
+		listening.add(token);
+		const dispose = changed.call(this, (value) => {
+			listening.delete(token);
+			fn(value);
+		});
+		return () => {
+			listening.delete(token);
+			dispose();
+		};
+	} as typeof changed);
+	try {
+		await run();
+	} finally {
+		spy.mockRestore();
+	}
+	return listening.size;
+}
+
+// Frames sleep on the clock once each, so a sleep that keeps anything on a clock that never changes
+// piles it up for the life of the player.
+test("waits on a stable clock leave nothing behind", async () => {
+	const sync = new Sync({ delay: Time.Milli(10) });
+	// Let the delay effect run before anchoring.
+	await new Promise((resolve) => setTimeout(resolve, 0));
+	sync.received(Time.Milli.now());
+
+	const wait = async () => {
+		for (let round = 0; round < 10; round++) {
+			const now = Time.Milli.now();
+			await Promise.all(Array.from({ length: 100 }, () => sync.wait(now)));
+		}
+	};
+	expect(await pendingReactions(wait)).toBe(0);
+	expect(await pendingListeners(wait)).toBe(0);
+
+	sync.close();
+});
+
 // The player path a decoder drives, one frame per group like AAC audio: the container consumer
 // reads each frame, the shared clock anchors on it, and presentation waits on the clock against
-// the effect's teardown. Retention anywhere along it grows the heap with the frame count.
-test("a long subscription through the player path keeps a flat heap", async () => {
+// the effect's teardown. Retention anywhere along it grows with the frame count.
+test("a long subscription through the player path leaves nothing behind", async () => {
 	const broadcast = new Moq.Broadcast.Producer();
 	// A tiny publisher window, so the track's own replay cache stays flat too.
 	const track = broadcast.createTrack("audio", { maxAge: Time.Milli(1) });
@@ -47,15 +114,13 @@ test("a long subscription through the player path keeps a flat heap", async () =
 		await Promise.all(presenting);
 	};
 
-	const heap = () => {
-		Bun.gc(true);
-		return heapStats().objectCount;
-	};
-
-	await play(200);
-	const before = heap();
-	await play(2000);
-	expect(heap() - before).toBeLessThan(1000);
+	// What stays is bounded by the track's window of open groups, not the frame count.
+	let reactions = 0;
+	const listeners = await pendingListeners(async () => {
+		reactions = await pendingReactions(() => play(2000));
+	});
+	expect(reactions).toBeLessThan(100);
+	expect(listeners).toBeLessThan(100);
 
 	consumer.close();
 	effect.close();

js/watch/src/sync.test.ts

@@ -1,4 +1,3 @@
-import { heapStats } from "bun:jsc";
 import { describe, expect, it } from "bun:test";
 import { Time } from "@moq/net";
 import { Signal } from "@moq/signals";
@@ -99,25 +98,6 @@ describe("delay and buffer", () => {
 });
 
 describe("wait", () => {
-	const promises = () => {
-		Bun.gc(true);
-		return heapStats().objectTypeCounts.Promise ?? 0;
-	};
-
-	it("leaves nothing behind on a stable clock", async () => {
-		const sync = new Sync({ delay: 10 as Time.Milli });
-		await flush();
-		sync.received(Time.Milli.now());
-
-		const before = promises();
-		for (let round = 0; round < 10; round++) {
-			const now = Time.Milli.now();
-			await Promise.all(Array.from({ length: 100 }, () => sync.wait(now)));
-		}
-		expect(promises() - before).toBeLessThan(100);
-		sync.close();
-	});
-
 	it("wakes a sleeping wait when the delay switches to instant", async () => {
 		const delay = new Signal<Delay>(10_000 as Time.Milli);
 		const sync = new Sync({ delay });