From 42808f7c3326438b676d0c42f097739675e5d69d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 28 Sep 2026 10:50:59 -0700 Subject: [PATCH 1/4] fix(net): age out an ended JS track's groups, match Rust abort semantics Producer.close stopped the prune timer and each sink detached its mirrors from the cache, so after a track ended nothing ever reclaimed the groups a stale subscriber had buffered. Keep pruning until the cache empties and keep the producer-closed sinks' mirrors tracked. Match moq-net: an abort drops the open groups nobody will finish rather than handing them out, and an abort after the declared end settled is a clean close. Co-Authored-By: Claude Opus 5.5 --- js/net/src/track.test.ts | 60 ++++++++++++++++++++++++++++++++++++++++ js/net/src/track.ts | 52 +++++++++++++++++++++++++--------- 2 files changed, 99 insertions(+), 13 deletions(-) diff --git a/js/net/src/track.test.ts b/js/net/src/track.test.ts index e3ca4517ff0..2c05e0cddd8 100644 --- a/js/net/src/track.test.ts +++ b/js/net/src/track.test.ts @@ -1687,3 +1687,63 @@ test("finished rejects when the track aborts or closes without an end", async () clean.close(); expect(await reader.finished()).toBe(1); }); + +test("an abort keeps finished groups for a slow reader, then reports it", async () => { + const producer = new TrackProducer("test").accept({ maxAge: Milli(10_000) }); + const arrival = producer.subscribe({ maxAge: Milli(10_000) }); + const ordered = producer.subscribe({ maxAge: Milli(10_000) }).ordered(); + for (let i = 0; i < 2; i++) { + const group = producer.appendGroup(); + group.writeString(`g${i}`); + group.close(); + } + producer.appendGroup().writeString("open"); + const boom = new Error("boom"); + producer.close(boom); + + for (const next of [() => arrival.recvGroup(), () => ordered.nextGroup()]) { + for (let i = 0; i < 2; i++) { + const group = await next(); + expect(group?.sequence).toBe(i); + expect(await group?.readString()).toBe(`g${i}`); + } + // The open group nobody will finish is not handed out. + await expect(next()).rejects.toBe(boom); + } +}); + +test("an abort after the declared end settles ends clean", async () => { + const producer = new TrackProducer("test").accept({ maxAge: Milli(10_000) }); + const arrival = producer.subscribe({ maxAge: Milli(10_000) }); + const ordered = producer.subscribe({ maxAge: Milli(10_000) }).ordered(); + for (let i = 0; i < 2; i++) { + const group = producer.appendGroup(); + group.writeString(`g${i}`); + group.close(); + } + producer.finishAt(2); + producer.close(new Error("boom")); + + for (const next of [() => arrival.recvGroup(), () => ordered.nextGroup()]) { + expect((await next())?.sequence).toBe(0); + expect((await next())?.sequence).toBe(1); + expect(await next()).toBeUndefined(); + } +}); + +test("an ended track's buffered groups age out for a stale subscriber", async () => { + // Real time: nothing writes after the close, so only the wakeup can reclaim them. + for (const abort of [undefined, new Error("boom")]) { + const producer = new TrackProducer("test").accept({ maxAge: Milli(30) }); + const stale = producer.subscribe({ maxAge: Milli(30) }); + const group = producer.appendGroup(); + group.writeString("x"); + group.close(); + producer.close(abort); + + await new Promise((resolve) => setTimeout(resolve, 200)); + // Nothing buffered is left: the subscriber sees only how the track ended. + if (abort) expect(() => stale.tryRecvGroup()).toThrow(abort); + else expect(stale.tryRecvGroup()).toBeUndefined(); + } +}); diff --git a/js/net/src/track.ts b/js/net/src/track.ts index dc6e8747ec7..d938169e7a9 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -565,6 +565,15 @@ export class Producer { forward(); this.#sinks.delete(sink); this.#updateSubscription(); + // Update demand: once the last subscriber leaves, the consumer wire (watching + // {@link unused}) tears the upstream down instead of downloading to nobody. + this.#used.set(this.#sinks.size > 0); + // The producer closing every sink keeps its mirrors tracked, so what the sink + // still buffers ages out with the cache instead of staying pinned. + if (this.#state.closed.peek() !== undefined) { + dispose(); + return; + } for (const entry of this.#cache) { const mirror = entry.mirrors.get(sink); if (mirror) { @@ -574,10 +583,6 @@ export class Producer { } for (const group of sink.groups.peek()) group.close(abort); dispose(); - - // Update demand: once the last subscriber leaves, the consumer wire (watching - // {@link unused}) tears the upstream down instead of downloading to nobody. - this.#used.set(this.#sinks.size > 0); }); } @@ -632,6 +637,13 @@ export class Producer { entry.mirrors.clear(); } + // Take a cached group out of the cache and every sink. + #drop(entry: CachedGroup): void { + this.#evict(entry); + this.#cache.splice(this.#cache.indexOf(entry), 1); + this.#cached.delete(entry.group.sequence); + } + // The one group retention never takes: the newest, while it is still open. That is // the live edge a publisher is appending to, and a track may legitimately keep it // open across a long quiet stretch (a catalog snapshot, a JSON stream). Every other @@ -685,9 +697,9 @@ export class Producer { if (oldest !== undefined) this.#wake(Math.max(oldest + maxAgeMs, now + slice)); } - // Arm the prune wakeup for `at`, unless one is already armed sooner. + // Arm the prune wakeup for `at`, unless one is already armed sooner. Kept after a close, + // until the cache empties, so what the track left behind still ages out. #wake(at: number): void { - if (this.#state.closed.peek() !== undefined) return; if (this.#pruneTimer !== undefined && this.#pruneTimerAt <= at) return; clearTimeout(this.#pruneTimer); @@ -752,9 +764,7 @@ export class Producer { if (!(existing.group.closed.peek() instanceof Error)) { throw new Error(`duplicate group: sequence=${group.sequence}`); } - this.#evict(existing); - this.#cache.splice(this.#cache.indexOf(existing), 1); - this.#cached.delete(group.sequence); + this.#drop(existing); } // Only advance the shared counter upward (for appendGroup auto-increment). @@ -848,21 +858,37 @@ export class Producer { * Close the track and every subscriber, mirroring the abort to their groups. Idempotent. * * A clean close keeps the end {@link finishAt} declared, or declares one past the highest - * sequence produced; an abort ends without one. + * sequence produced; an abort ends without one. Subscribers still draining get the + * finished groups first, then the end or the abort. An abort after the declared end + * settled (reached, with every group below it finished) is a clean close. The groups + * left behind still age out after the track's `maxAge`, so a stale subscriber can't pin + * them. */ close(abort?: Error) { - if (abort === undefined && this.#state.closed.peek() === undefined && this.#state.final.peek() === undefined) { + if (this.#state.closed.peek() !== undefined) return; + if (abort && this.#settled()) abort = undefined; + if (abort === undefined && this.#state.final.peek() === undefined) { this.#declareFinal(this.#sequence.next); } + // Nobody will finish these, so a subscriber that has not taken one yet never sees it. + const open = abort ? this.#cache.filter((entry) => entry.group.closed.peek() === undefined) : []; closeTrackState(this.#state, abort); - clearTimeout(this.#pruneTimer); - this.#pruneTimer = undefined; for (const { group } of this.#cache) group.close(abort); + for (const entry of open) this.#drop(entry); for (const sink of this.#sinks) { for (const group of sink.groups.peek()) group.close(abort); closeTrackState(sink, abort); } this.#sinks.clear(); + this.#prune(); + } + + // Whether the declared end was reached and every cached group below it finished, so + // the track already holds everything it promised. Mirrors the Rust `is_settled`. + #settled(): boolean { + const final = this.#state.final.peek(); + if (final === undefined || this.#sequence.next < final) return false; + return this.#cache.every(({ group }) => group.sequence >= final || group.closed.peek() === null); } /** Append a frame as its own single-frame group. */ From d55bb4eebd48210f89559899427742d1c3da811b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 28 Sep 2026 11:39:39 -0700 Subject: [PATCH 2/4] fix(net): keep a taken open group on abort, mock the prune clock Abort only unlinks open groups from the cache and every sink, so a reader already holding one keeps its frames and sees the abort instead of TooFarBehind. Eviction stays for maxAge pruning and the aborted-duplicate replace in writeGroup. The age-out test stubs the timers and clock instead of sleeping. Co-Authored-By: Claude Opus 5.5 --- js/net/src/track.test.ts | 68 ++++++++++++++++++++++++++++++++-------- js/net/src/track.ts | 25 ++++++++++----- 2 files changed, 73 insertions(+), 20 deletions(-) diff --git a/js/net/src/track.test.ts b/js/net/src/track.test.ts index 2c05e0cddd8..f4641ef5d1a 100644 --- a/js/net/src/track.test.ts +++ b/js/net/src/track.test.ts @@ -1731,19 +1731,61 @@ test("an abort after the declared end settles ends clean", async () => { } }); -test("an ended track's buffered groups age out for a stale subscriber", async () => { - // Real time: nothing writes after the close, so only the wakeup can reclaim them. - for (const abort of [undefined, new Error("boom")]) { - const producer = new TrackProducer("test").accept({ maxAge: Milli(30) }); - const stale = producer.subscribe({ maxAge: Milli(30) }); - const group = producer.appendGroup(); - group.writeString("x"); - group.close(); - producer.close(abort); +test("an ended track's buffered groups age out for a stale subscriber", () => { + // Nothing writes after the close, so only the prune wakeup can reclaim them. Stub the + // timers so the test fires exactly the wakeups still armed, at a mocked time. + const clock = mockMonotonicTime(10_000); + const realSet = globalThis.setTimeout; + const realClear = globalThis.clearTimeout; + const armed = new Map void>(); + // @ts-expect-error a stub, not a full setTimeout + globalThis.setTimeout = (fn: () => void) => { + const handle = { unref: () => {} }; + armed.set(handle, fn); + return handle; + }; + // @ts-expect-error a stub, not a full clearTimeout + globalThis.clearTimeout = (handle: object) => armed.delete(handle); + + try { + for (const abort of [undefined, new Error("boom")]) { + clock.set(10_000); + armed.clear(); + const producer = new TrackProducer("test").accept({ maxAge: Milli(30) }); + const stale = producer.subscribe({ maxAge: Milli(30) }); + const group = producer.appendGroup(); + group.writeString("x"); + group.close(); + producer.close(abort); + + clock.set(10_100); + for (const [handle, fire] of [...armed]) { + armed.delete(handle); + fire(); + } + // Nothing buffered is left: the subscriber sees only how the track ended. + if (abort) expect(() => stale.tryRecvGroup()).toThrow(abort); + else expect(stale.tryRecvGroup()).toBeUndefined(); + } + } finally { + globalThis.setTimeout = realSet; + globalThis.clearTimeout = realClear; + clock.restore(); + } +}); + +test("an abort leaves a group already taken readable, then reports the abort", async () => { + const producer = new TrackProducer("test").accept({ maxAge: Milli(10_000) }); + const arrival = producer.subscribe({ maxAge: Milli(10_000) }); + const ordered = producer.subscribe({ maxAge: Milli(10_000) }).ordered(); + producer.appendGroup().writeString("held"); + const held = [await arrival.recvGroup(), await ordered.nextGroup()]; + const boom = new Error("boom"); + producer.close(boom); - await new Promise((resolve) => setTimeout(resolve, 200)); - // Nothing buffered is left: the subscriber sees only how the track ended. - if (abort) expect(() => stale.tryRecvGroup()).toThrow(abort); - else expect(stale.tryRecvGroup()).toBeUndefined(); + for (const group of held) { + expect(group?.sequence).toBe(0); + expect(await group?.readString()).toBe("held"); + await expect(group?.readFrame()).rejects.toBe(boom); } }); diff --git a/js/net/src/track.ts b/js/net/src/track.ts index d938169e7a9..9b683f71edc 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -624,22 +624,28 @@ export class Producer { // drained it. The usual case, an already-closed group aging out, keeps its own // terminal state. if (!entry.group.isClosed) entry.group.close(new TooFarBehind()); + const mirrors = [...entry.mirrors.values()]; + for (const mirror of mirrors) hooks.evictGroup(mirror); + this.#unlink(entry); + for (const mirror of mirrors) mirror.close(); + } + + // Take a cached group's mirrors out of every sink, so a subscriber can no longer + // receive them. A reader already holding one keeps it as is. + #unlink(entry: CachedGroup): void { for (const [sink, mirror] of entry.mirrors) { - hooks.evictGroup(mirror); sink.groups.mutate((groups) => { const i = groups.indexOf(mirror); if (i >= 0) groups.splice(i, 1); }); const index = timelineIndex(sink.timeline, mirror.sequence); if (sink.timeline[index] === mirror) sink.timeline.splice(index, 1); - mirror.close(); } entry.mirrors.clear(); } - // Take a cached group out of the cache and every sink. - #drop(entry: CachedGroup): void { - this.#evict(entry); + // Take a cached group out of the cache lookups. + #uncache(entry: CachedGroup): void { this.#cache.splice(this.#cache.indexOf(entry), 1); this.#cached.delete(entry.group.sequence); } @@ -764,7 +770,8 @@ export class Producer { if (!(existing.group.closed.peek() instanceof Error)) { throw new Error(`duplicate group: sequence=${group.sequence}`); } - this.#drop(existing); + this.#evict(existing); + this.#uncache(existing); } // Only advance the shared counter upward (for appendGroup auto-increment). @@ -871,10 +878,14 @@ export class Producer { this.#declareFinal(this.#sequence.next); } // Nobody will finish these, so a subscriber that has not taken one yet never sees it. + // Not evicted: a reader already holding one keeps its frames and sees the abort. const open = abort ? this.#cache.filter((entry) => entry.group.closed.peek() === undefined) : []; closeTrackState(this.#state, abort); for (const { group } of this.#cache) group.close(abort); - for (const entry of open) this.#drop(entry); + for (const entry of open) { + this.#unlink(entry); + this.#uncache(entry); + } for (const sink of this.#sinks) { for (const group of sink.groups.peek()) group.close(abort); closeTrackState(sink, abort); From 7970978322661ba885ca48a84cf091c794fe238d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 28 Sep 2026 12:06:31 -0700 Subject: [PATCH 3/4] fix(net): settle an abort on the groups this producer received Sibling producers of a track and datagrams share the sequence counter, so it can reach the declared end without this producer holding those groups. Track this producer's own highest group, like Rust max_sequence. Co-Authored-By: Claude Opus 5.5 --- js/net/src/broadcast.test.ts | 26 ++++++++++++++++++++++++++ js/net/src/track.ts | 6 +++++- 2 files changed, 31 insertions(+), 1 deletion(-) diff --git a/js/net/src/broadcast.test.ts b/js/net/src/broadcast.test.ts index 1a0c019c428..28d202c48c6 100644 --- a/js/net/src/broadcast.test.ts +++ b/js/net/src/broadcast.test.ts @@ -109,6 +109,32 @@ test("concurrent dynamic producers share a sequence namespace", async () => { broadcast.close(); }); +test("a sibling producer's groups do not settle an aborted end", async () => { + const broadcast = new BroadcastProducer(); + const firstSubscriber = broadcast.track("media").subscribe(); + const secondSubscriber = broadcast.track("media").subscribe(); + const firstRequest = await wireOf(broadcast).requested(); + const secondRequest = await wireOf(broadcast).requested(); + if (!firstRequest || !secondRequest) throw new Error("expected requests"); + const firstProducer = firstRequest.accept(); + const secondProducer = secondRequest.accept(); + + const reader = firstProducer.subscribe(); + firstProducer.finishAt(3); + for (let i = 0; i < 3; i++) secondProducer.appendGroup().close(); + const boom = new Error("boom"); + firstProducer.close(boom); + + // The first producer never received the groups its end promised, so it was cut off. + expect(firstProducer.closed.peek()).toBe(boom); + await expect(reader.recvGroup()).rejects.toBe(boom); + + firstSubscriber.close(); + secondSubscriber.close(); + secondProducer.close(); + broadcast.close(); +}); + test("closing a broadcast rejects a dequeued request", async () => { const broadcast = new BroadcastProducer(); const subscriber = broadcast.track("media").subscribe(); diff --git a/js/net/src/track.ts b/js/net/src/track.ts index 9b683f71edc..4eb51ccbf86 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -426,6 +426,9 @@ export class Producer { // read mirrored sinks, never this state directly. #state = new TrackState(); #sequence: TrackSequence = { next: 0 }; + // One past the highest group this producer received. The shared counter above can run + // ahead of it: sibling producers of the same track and datagrams advance it too. + #received = 0; // Recently written source groups, retained for replay to late subscribers and // pruned once idle for longer than the cache window. Each entry tracks the mirror @@ -728,6 +731,7 @@ export class Producer { const entry: CachedGroup = { group, mirrors: new Map() }; this.#cache.push(entry); this.#cached.set(group.sequence, entry); + this.#received = Math.max(this.#received, group.sequence + 1); for (const sink of this.#sinks) this.#mirror(entry, sink); // Give held mirrors the new live edge before pruning their timeline entry, // so their latency guard can preserve a terminal expiry verdict. @@ -898,7 +902,7 @@ export class Producer { // the track already holds everything it promised. Mirrors the Rust `is_settled`. #settled(): boolean { const final = this.#state.final.peek(); - if (final === undefined || this.#sequence.next < final) return false; + if (final === undefined || this.#received < final) return false; return this.#cache.every(({ group }) => group.sequence >= final || group.closed.peek() === null); } From 1f8753bd33ce11678776e5b7a41d2d31926686b6 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 28 Sep 2026 12:57:44 -0700 Subject: [PATCH 4/4] fix(net): count this producer's datagrams toward a settled end Rust advances max_sequence for datagrams too, so a declared end reached by a datagram settles an abort into a clean close. Co-Authored-By: Claude Opus 5.5 --- js/net/src/track.test.ts | 8 ++++++++ js/net/src/track.ts | 6 ++++-- 2 files changed, 12 insertions(+), 2 deletions(-) diff --git a/js/net/src/track.test.ts b/js/net/src/track.test.ts index f4641ef5d1a..c8c516b67e6 100644 --- a/js/net/src/track.test.ts +++ b/js/net/src/track.test.ts @@ -1731,6 +1731,14 @@ test("an abort after the declared end settles ends clean", async () => { } }); +test("an abort after the declared end reached by a datagram ends clean", () => { + const producer = new TrackProducer("test").accept(); + producer.appendDatagram(Timestamp.fromMillis(0), enc.encode("d")); + producer.finishAt(1); + producer.close(new Error("boom")); + expect(producer.closed.peek()).toBeNull(); +}); + test("an ended track's buffered groups age out for a stale subscriber", () => { // Nothing writes after the close, so only the prune wakeup can reclaim them. Stub the // timers so the test fires exactly the wakeups still armed, at a mocked time. diff --git a/js/net/src/track.ts b/js/net/src/track.ts index 4eb51ccbf86..9f406cd55d5 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -426,8 +426,9 @@ export class Producer { // read mirrored sinks, never this state directly. #state = new TrackState(); #sequence: TrackSequence = { next: 0 }; - // One past the highest group this producer received. The shared counter above can run - // ahead of it: sibling producers of the same track and datagrams advance it too. + // One past the highest group or datagram this producer received, like the Rust + // `max_sequence`. The shared counter above can run ahead of it: sibling producers of + // the same track advance it too. #received = 0; // Recently written source groups, retained for replay to late subscribers and @@ -790,6 +791,7 @@ export class Producer { // Fan a datagram out to every live subscriber, dropping the oldest once the ring is full. // Late subscribers do NOT replay old datagrams (best-effort, unlike the group cache). #publishDatagram(datagram: Datagram): void { + this.#received = Math.max(this.#received, datagram.sequence + 1); for (const sink of this.#sinks) { sink.datagrams.mutate((list) => { if (list.length === MAX_DATAGRAMS) list.shift();