diff --git a/js/binary/src/stream/stream.test.ts b/js/binary/src/stream/stream.test.ts index b8745147fb..4b21ace239 100644 --- a/js/binary/src/stream/stream.test.ts +++ b/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,6 +147,24 @@ 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): Promise { + const reacted: Promise[] = []; + const then = Promise.prototype.then; + const spy = spyOn(Promise.prototype, "then").mockImplementation(function (this: Promise, ...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 () => { @@ -155,23 +172,15 @@ test("blocked reads leave nothing behind on the pending group read", async () => 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(); diff --git a/js/json/src/stream/stream.test.ts b/js/json/src/stream/stream.test.ts index 35512bac88..1e342998c2 100644 --- a/js/json/src/stream/stream.test.ts +++ b/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,6 +111,24 @@ 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): Promise { + const reacted: Promise[] = []; + const then = Promise.prototype.then; + const spy = spyOn(Promise.prototype, "then").mockImplementation(function (this: Promise, ...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 () => { @@ -119,23 +136,15 @@ test("blocked reads leave nothing behind on the pending group read", async () => 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({ 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(); diff --git a/js/watch/src/retention.test.ts b/js/watch/src/retention.test.ts index af0af0860e..3afeee39ea 100644 --- a/js/watch/src/retention.test.ts +++ b/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): Promise { + const reacted: Promise[] = []; + const then = Promise.prototype.then; + const spy = spyOn(Promise.prototype, "then").mockImplementation(function (this: Promise, ...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): Promise { + const listening = new Set(); + const changed = Signal.prototype.changed; + const spy = spyOn(Signal.prototype, "changed").mockImplementation(function ( + this: Signal, + fn?: (value: unknown) => void, + ) { + if (!fn) return (changed as () => Promise).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(); diff --git a/js/watch/src/sync.test.ts b/js/watch/src/sync.test.ts index 2376f625e9..d8d18764b8 100644 --- a/js/watch/src/sync.test.ts +++ b/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(10_000 as Time.Milli); const sync = new Sync({ delay });