From 72928ba33be7ba0fb15db9beb866d86591ab2b01 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 15:29:26 -0700 Subject: [PATCH 1/7] quest: claim m0/broadcast-epoch/dynamic-epoch From ba4d9b3ac254408434c94f203c43955b622630ac Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 15:41:20 -0700 Subject: [PATCH 2/7] test(net): an epoch update re-keys a dynamic claim Dynamic::update (Rust) and Dynamic.update (JS) document what an epoch change does: a Restart downstream, held requests refused, the next request asks the handler again, served broadcasts keep their open subscriptions, and route selection still applies. Regressions cover A to B, A to none, and none to B through a relay on every version, same-epoch repricing keeping handles, and a second claim at the old epoch keeping re-requests. Co-Authored-By: Claude Opus 5.5 --- js/net/src/origin.test.ts | 146 ++++++++++++++++++ js/net/src/origin.ts | 10 +- quest/m0/broadcast-epoch/README.md | 1 - quest/m0/broadcast-epoch/dynamic-epoch.md | 50 ------- rs/moq-net/src/model/origin.rs | 13 +- rs/moq-net/tests/restart.rs | 171 +++++++++++++++++++++- 6 files changed, 332 insertions(+), 59 deletions(-) delete mode 100644 quest/m0/broadcast-epoch/dynamic-epoch.md diff --git a/js/net/src/origin.test.ts b/js/net/src/origin.test.ts index 6c6ede3fbd..4bcd648052 100644 --- a/js/net/src/origin.test.ts +++ b/js/net/src/origin.test.ts @@ -3,6 +3,7 @@ import { getter, Once, race, Signal } from "@moq/signals"; import { type Consumer as BroadcastConsumer, Producer as BroadcastProducer } from "./broadcast.ts"; import * as Epoch from "./epoch.ts"; import { StreamCode, StreamError } from "./error.ts"; +import { Producer as GroupProducer } from "./group.ts"; import { HopSchema, Route } from "./hop.ts"; import { spreadHash } from "./internal.ts"; import type { Consumer, Table } from "./origin.ts"; @@ -2258,6 +2259,151 @@ test("request never accepts an old handler after its epoch changes asynchronousl origin.close(); }); +const NEWER = Epoch.parse("01900000-0000-7000-8000-000000000002"); + +test.each([ + ["A to B", EPOCH, NEWER], + ["A to none", EPOCH, undefined], + ["none to B", undefined, NEWER], +])("an epoch update %s restarts the claim, while a re-price keeps every handle", async (_, from, to) => { + const origin = new Producer(); + const prefix = Path.from("live"); + const path = Path.from("live/cam"); + const dynamic = origin.dynamic(prefix, { epoch: from }); + const requests = dynamic.requested(); + const announced = origin.consume().announced(); + expect(await announced.next()).toMatchObject({ prefix, kind: "start" }); + + const held = origin.consume().request(path); + const old = new BroadcastProducer(); + const oldTrack = old.createTrack("video"); + (await requests.next()).value?.accept(old); + await settle(); + const resolved = held.active.peek(); + expect(resolved?.epoch).toBe(from); + + const sticky = resolved?.track("video").subscribe(); + oldTrack.appendGroup().close(); + expect((await sticky?.recvGroup())?.sequence).toBe(0); + + // A re-price at the same epoch keeps every handle, and the handler is not asked again. + dynamic.update({ ...dynamic.route, cost: 5n }); + expect(await announced.next()).toMatchObject({ prefix, kind: "update" }); + expect(held.active.peek()).toBe(resolved); + const same = origin.consume().request(path); + await settle(); + expect(same.active.peek()?.epoch).toBe(from); + + // Another epoch, or none, is another instance: the requests on the old one end. + dynamic.update({ ...dynamic.route, epoch: to }); + expect(await announced.next()).toMatchObject({ prefix, kind: "restart" }); + await settle(); + expect(held.active.peek()).toBeUndefined(); + expect(held.unroutable.peek()).toBe(true); + expect(same.active.peek()).toBeUndefined(); + + // The subscription already open stays on the old answer. + oldTrack.appendGroup().close(); + expect((await sticky?.recvGroup())?.sequence).toBe(1); + + // A re-request, local or from a session, asks the handler again and never reads the + // old answer. + const fresh = origin.consume().request(path); + const demanded = wireOf(origin.consume()).demand(path); + const next = new BroadcastProducer(); + const nextTrack = next.createTrack("video"); + (await requests.next()).value?.accept(next); + await settle(); + expect(fresh.active.peek()?.epoch).toBe(to); + const subscription = fresh.active.peek()?.track("video").subscribe(); + const forwarded = (await demanded)?.track("video").subscribe(); + nextTrack.writeGroup(new GroupProducer(7)); + expect((await subscription?.recvGroup())?.sequence).toBe(7); + expect((await forwarded?.recvGroup())?.sequence).toBe(7); + + await requests.return?.(); + sticky?.close(); + subscription?.close(); + forwarded?.close(); + for (const handle of [held, same, fresh]) handle.close(); + old.close(); + next.close(); + dynamic.close(); + announced.close(); + origin.close(); +}); + +test("a session's re-request after an epoch update asks the handler again", async () => { + const origin = new Producer(); + const path = Path.from("live/cam"); + const dynamic = origin.dynamic(Path.from("live"), { epoch: EPOCH }); + const requests = dynamic.requested(); + const wire = wireOf(origin.consume()); + + const before = wire.demand(path); + const old = new BroadcastProducer(); + (await requests.next()).value?.accept(old); + expect(await before).toBeDefined(); + + dynamic.update({ ...dynamic.route, epoch: NEWER }); + const after = wire.demand(path); + const next = new BroadcastProducer(); + const track = next.createTrack("video"); + (await requests.next()).value?.accept(next); + const subscription = (await after)?.track("video").subscribe(); + track.writeGroup(new GroupProducer(7)); + expect((await subscription?.recvGroup())?.sequence).toBe(7); + + subscription?.close(); + await requests.return?.(); + old.close(); + next.close(); + dynamic.close(); + origin.close(); +}); + +test("an update does not override route selection: another claim at the epoch keeps it", async () => { + const origin = new Producer(); + const prefix = Path.from("live"); + const path = Path.from("live/cam"); + const first = origin.dynamic(prefix, { epoch: EPOCH, cost: 1n }); + const second = origin.dynamic(prefix, { epoch: EPOCH, cost: 5n }); + const firstRequests = first.requested(); + const secondRequests = second.requested(); + + const held = origin.consume().request(path); + const answer = new BroadcastProducer(); + (await firstRequests.next()).value?.accept(answer); + await settle(); + expect(held.active.peek()?.epoch).toBe(EPOCH); + + // The first claim drops the epoch, so the second, still at it, outranks it: the + // request on the epoch moves to the second's answer rather than ending. + first.update({ ...first.route, epoch: undefined }); + const fresh = origin.consume().request(path); + const replacement = new BroadcastProducer(); + const track = replacement.createTrack("video"); + (await secondRequests.next()).value?.accept(replacement); + await settle(); + expect(held.active.peek()?.epoch).toBe(EPOCH); + expect(held.unroutable.peek()).toBe(false); + expect(fresh.active.peek()?.epoch).toBe(EPOCH); + const subscription = fresh.active.peek()?.track("video").subscribe(); + track.writeGroup(new GroupProducer(7)); + expect((await subscription?.recvGroup())?.sequence).toBe(7); + + subscription?.close(); + await firstRequests.return?.(); + await secondRequests.return?.(); + held.close(); + fresh.close(); + answer.close(); + replacement.close(); + first.close(); + second.close(); + origin.close(); +}); + test("a route without an epoch resolves a handle without one", () => { const origin = new Producer(); const path = Path.from("room"); diff --git a/js/net/src/origin.ts b/js/net/src/origin.ts index d1c837520d..b6dbc7368f 100644 --- a/js/net/src/origin.ts +++ b/js/net/src/origin.ts @@ -1807,8 +1807,14 @@ export class Dynamic { /** * Replace the route in place. The prefix is fixed at announce time. * - * The route is taken as given, epoch included: another epoch (or none) names another - * publisher instance, so re-price from the current one, `update({ ...dynamic.route, cost })`. + * The route is taken as given. At the same epoch this re-prices: consumers see an update + * and every handle survives. Another epoch, or none, names another publisher instance: + * consumers see a restart, the requests resolved through the old one end, and a + * re-request never joins it. Requests this handle still holds are refused as unroutable, + * and the next request for a path it served asks the handler again. The broadcasts it + * served keep running for the subscriptions already on them; close them to end those too. + * Route selection still applies: another route still at the old epoch outranks one + * without. To re-price, start from the current route, `update({ ...dynamic.route, cost })`. */ update(route: Route | { epoch?: Route["epoch"]; hops?: Route["hops"]; cost?: Route["cost"] }): void { if (this.#closed) throw new Error("dynamic is closed"); diff --git a/quest/m0/broadcast-epoch/README.md b/quest/m0/broadcast-epoch/README.md index 956eb80694..33fba69db9 100644 --- a/quest/m0/broadcast-epoch/README.md +++ b/quest/m0/broadcast-epoch/README.md @@ -100,4 +100,3 @@ then. MoQ Pro's VOD `storage.json` moves from the same seed to its own epoch. - [Bindings](/quest/m0/broadcast-epoch/bindings.md) - moq-ffi and every wrapper expose the epoch and let a publisher announce one - [moqsrc](/quest/m0/broadcast-epoch/moqsrc.md) - moqsrc switches to the new broadcast on a `Restart`, keeping its pads by rendition name - [Stats totals and prefix tracks](/quest/m0/broadcast-epoch/stats-split.md) - the same release retires the per-path stats maps for totals and on-demand prefix tracks (decided 2026-10-05) -- [Dynamic epoch update](/quest/m0/broadcast-epoch/dynamic-epoch.md) - an epoch change through `Dynamic::update` announces a `Restart` downstream, so subscribers drop the old copy and resubscribe, while the origin keeps serving diff --git a/quest/m0/broadcast-epoch/dynamic-epoch.md b/quest/m0/broadcast-epoch/dynamic-epoch.md deleted file mode 100644 index cc152bec14..0000000000 --- a/quest/m0/broadcast-epoch/dynamic-epoch.md +++ /dev/null @@ -1,50 +0,0 @@ -# [S] An epoch update re-keys a dynamic claim - -## Goal - -`Dynamic::update` (Rust) and `Dynamic.update` (JS) may change the route's -epoch, or add or clear it. The change is a source change downstream, under -the same rule as any other: announce consumers see a `Restart`, every -downstream subscriber (relays included) drops its copy of the old epoch and -resubscribes. Only an identical epoch splices, so no old-epoch group reaches -a new-epoch subscriber. The update invalidates this claim's old-epoch copies; -it does not override route selection. A re-request still resolves the -winning route, which ranks the newest epoch first and none last, so after A -to none, or A to an older B, another route still advertising A keeps -winning. The origin's own served broadcasts and pending requests carry over, -since two epochs may carry the same content; a handler that wants fresh -content under the new epoch closes its old broadcast itself. - -Non-goals: the epoch a claim-served answer carries -([Claim-served epochs](/quest/m1/claim-epochs.md)), and -same-epoch repricing, which keeps every handle and stays an `Update`. - -## Plan - -Found by the final-head audit of -[#4942](https://github.com/moq-dev/moq/pull/4942): an A to B update keeps -serving A's cached broadcast from the origin (`rs/moq-net/src/model/origin.rs` -`Dynamic::update` through `AnnounceProducer::update`), while its doc promises -the served broadcasts end. The JS `Dynamic.update` doc (`js/net/src/origin.ts`) -only says another epoch names another publisher instance. - -Decided 2026-10-07: carrying the served and pending state over is correct, -because an epoch is a cache identity, not a content guarantee. Rejected: -refusing an epoch change in `update` (forces drop and re-claim for no -benefit), and ending served broadcasts on an epoch change (more model code, -and the handler can already do it). - -Aligned 2026-10-08 with Restart, which -replaced the hard switch: downstream subscriptions are not ended by the -model; the `Restart` tells them to drop the old copy and resubscribe. - -- Rewrite both docs to the contract above. -- Regress in Rust and JS: an update A to B, A to none, and none to B delivers - `Restart` to an announce consumer and through a relay, the relay retires its - old-epoch copy for new requests, and a re-request is served under the new - epoch without any cached group from the old one. With a second route still - at A, A to none leaves re-requests on that route, by normal route - selection. If downstream already does this, the quest is the - docs and tests; otherwise fix the gap at its source. -- Keep the same-epoch repricing case in the same test, proving handles survive. - diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index dfc64587d2..0cdc75c84e 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -3724,11 +3724,14 @@ impl Dynamic { /// Replace the route in place, taken as given. /// - /// At the same [`Route::epoch`], consumers observe another active update for - /// the same prefix and sessions forward it as an update, so re-pricing never - /// looks like new content. Another epoch, or none, names another publisher - /// instance: consumers see a restart, the broadcasts served through the old one - /// end, and requests still waiting on it are refused with [`Error::Unroutable`]. + /// At the same [`Route::epoch`] this re-prices: consumers observe an update and + /// every handle survives. Another epoch, or none, names another publisher + /// instance: consumers see an [`AnnounceEvent::Restart`], and a re-request never + /// joins what was resolved under the old one. Requests this handle still holds + /// are refused with [`Error::Unroutable`], and the next request for a path it + /// served asks the handler again. The broadcasts it served keep running for the + /// subscriptions already on them; close them to end those too. Route selection + /// still applies: another route still at the old epoch outranks one without. /// To re-price, start from the current route: /// `dynamic.update(dynamic.route().with_cost(cost))`. /// diff --git a/rs/moq-net/tests/restart.rs b/rs/moq-net/tests/restart.rs index 489ca859b8..ad9d8085a8 100644 --- a/rs/moq-net/tests/restart.rs +++ b/rs/moq-net/tests/restart.rs @@ -10,7 +10,7 @@ mod support; use std::{cell::RefCell, rc::Rc, time::Duration}; -use moq_net::{Hop, Timestamp, Version, announce, broadcast, origin, track}; +use moq_net::{Epoch, Hop, Timestamp, Version, announce, broadcast, origin, track}; use support::harness::{MockConnectOptions, MockPair, connect_mock}; const TEST_TIMEOUT: Duration = Duration::from_secs(60); @@ -227,3 +227,172 @@ async fn prefix_pool_lite07() { .await .expect("timed out"); } + +/// A claim at `live` whose handler answers each request with a fresh output, kept so +/// the test can write into it: `outputs[n]` answered the `n`th request. +struct Claim { + dynamic: Rc, + outputs: Rc>>, +} + +impl Claim { + fn new(origin: &origin::Producer, epoch: Option) -> Self { + let mut route = origin::Route::default(); + route.epoch = epoch; + let dynamic = Rc::new(origin.dynamic("live", route).unwrap()); + let outputs = Rc::new(RefCell::new(Vec::new())); + let (handler, served) = (dynamic.clone(), outputs.clone()); + drop(moq_net_sim::spawn(async move { + while let Ok(request) = handler.requested_broadcast().await { + let output = broadcast::Info::new().produce(); + let track = output.create_track("video", None).unwrap(); + request.accept(&output); + served.borrow_mut().push((output, track)); + } + })); + Self { dynamic, outputs } + } + + /// Write `tag` into the output that answered the `n`th request. + fn write(&self, n: usize, tag: &str) { + write(&self.outputs.borrow()[n].1, tag); + } + + fn answered(&self) -> usize { + self.outputs.borrow().len() + } + + /// Move the claim to `epoch` in place, keeping the rest of its route. + fn update_epoch(&self, epoch: Option) { + let mut route = self.dynamic.route(); + route.epoch = epoch; + self.dynamic.update(route).unwrap(); + } +} + +/// `P -> R -> V`, where `P` claims `live` at `from` and updates the claim in place. +/// A same-epoch re-price keeps every handle. Moving to `to` restarts `V`, the +/// subscription already open stays on the old answer, and a re-request asks the +/// handler again and reads none of the old answer's cached groups. +async fn dynamic_epoch(version: &str, from: Option, to: Option) { + let version: Version = version.parse().unwrap(); + let publisher = produce_origin(1); + let relay = produce_origin(2); + let viewer = produce_origin(3); + let _upstream = link(version, &publisher, &relay).await; + let _downstream = link(version, &relay, &viewer).await; + + let claim = Claim::new(&publisher, from); + let mut announced = viewer.consume().announced(); + match next_event(&mut announced).await { + announce::Event::Start(announce) => assert_eq!(announce.prefix.as_str(), "live"), + other => panic!("{version}: expected the claim, got {other:?}"), + } + + let (stale, mut sticky) = subscribe(&viewer, "live/cam").await; + claim.write(0, "old:0"); + assert_eq!(read(&mut sticky).await, "old:0", "{version}"); + + // A re-price at the same epoch is an update at most: every handle survives. + claim.dynamic.update(claim.dynamic.route().with_cost(5)).unwrap(); + while let Ok(event) = moq_net_sim::timeout(Duration::from_secs(1), announced.next()).await { + assert!( + matches!(event, Some(announce::Event::Update(_))), + "{version}: a re-price delivered {event:?}" + ); + } + claim.write(0, "old:1"); + assert_eq!(read(&mut sticky).await, "old:1", "{version}"); + let (same, _) = subscribe(&viewer, "live/cam").await; + assert!(same.is_clone(&stale), "{version}: a re-price replaced the broadcast"); + assert_eq!(claim.answered(), 1, "{version}: a re-price asked the handler again"); + + // Another epoch, or none, is another instance downstream. + claim.update_epoch(to.clone()); + expect_restart(version, &mut announced, "live").await; + + // The subscription already open stays on the old answer. + claim.write(0, "old:2"); + assert_eq!(read(&mut sticky).await, "old:2", "{version}"); + + // A re-request asks the handler again and never reads the old answer. + let (fresh, mut subscription) = subscribe(&viewer, "live/cam").await; + assert!(!fresh.is_clone(&stale), "{version}: joined the old instance's copy"); + assert_eq!(claim.answered(), 2, "{version}: the old answer was served again"); + if version == "moq-lite-07-wip".parse().unwrap() { + assert_eq!(fresh.info().epoch, to, "{version}: resolved under the old epoch"); + } + claim.write(1, "new:0"); + assert_eq!(read(&mut subscription).await, "new:0", "{version}"); +} + +/// The epoch the claim starts at, and a newer one it moves to. +fn epochs() -> (Epoch, Epoch) { + ( + "01900000-0000-7000-8000-000000000001".parse().unwrap(), + "01900000-0000-7000-8000-000000000002".parse().unwrap(), + ) +} + +macro_rules! dynamic_epoch_tests { + ($($name:ident: $version:literal,)*) => { + $( + #[moq_net_sim::test] + async fn $name() { + let (a, b) = epochs(); + for (from, to) in [(Some(a.clone()), Some(b.clone())), (Some(a), None), (None, Some(b))] { + moq_net_sim::timeout(TEST_TIMEOUT, dynamic_epoch($version, from, to)) + .await + .expect("timed out"); + } + } + )* + }; +} + +dynamic_epoch_tests! { + dynamic_epoch_lite06: "moq-lite-06", + dynamic_epoch_lite07: "moq-lite-07-wip", + dynamic_epoch_ietf19: "moq-transport-19", + dynamic_epoch_ietf22: "moq-transport-22", +} + +/// An update does not override route selection: with a second claim still at the +/// epoch, a claim dropping it hands re-requests to the second, which outranks any +/// route without one. +#[moq_net_sim::test] +async fn dynamic_epoch_second_route_keeps_winning() { + let (a, _) = epochs(); + let origin = produce_origin(1); + let first = Claim::new(&origin, Some(a.clone())); + first.dynamic.update(first.dynamic.route().with_cost(1)).unwrap(); + let second = Claim::new(&origin, Some(a.clone())); + second.dynamic.update(second.dynamic.route().with_cost(5)).unwrap(); + let mut announced = origin.consume().announced(); + match next_event(&mut announced).await { + announce::Event::Start(announce) => assert_eq!(announce.route.epoch.as_ref(), Some(&a)), + other => panic!("expected the claim, got {other:?}"), + } + + let (resolved, mut sticky) = subscribe(&origin, "live/cam").await; + first.write(0, "first:0"); + assert_eq!(read(&mut sticky).await, "first:0"); + assert_eq!(second.answered(), 0, "the costlier claim answered"); + + // The epoch still stands through the second claim, so this is no restart, and the + // instance's subscriptions resume through it. + first.update_epoch(None); + while let Ok(event) = moq_net_sim::timeout(Duration::from_secs(1), announced.next()).await { + assert!( + matches!(event, Some(announce::Event::Update(_))), + "dropping a shared epoch delivered {event:?}" + ); + } + let fresh = origin.consume().request_broadcast("live/cam", None).await.unwrap(); + assert!(fresh.is_clone(&resolved), "the epoch's broadcast was replaced"); + assert_eq!((first.answered(), second.answered()), (1, 1)); + // The same epoch serves the same bytes: group 0 was delivered, so it resumes at 1. + second.write(0, "first:0"); + second.write(0, "second:1"); + assert_eq!(read(&mut sticky).await, "second:1"); +} From f3851ceff34aea9215caf55d2d147b34b8612350 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 18:10:40 -0700 Subject: [PATCH 3/7] fix(net): requests held across an epoch change ask again An epoch change through Dynamic::update released the requests the handler still held. In Rust that refusal ended the requester with Unroutable; in JS a session's demand came back empty. Both now ask the handler again under the new epoch, while a request pinned to the old epoch is still refused. The handler's late answer for the old instance is never served under the new epoch nor cached, and the served cache is still forgotten. Co-Authored-By: Claude Opus 5.5 --- js/net/src/origin.test.ts | 51 +++++++++++------ js/net/src/origin.ts | 21 +++++-- rs/moq-net/src/model/origin.rs | 100 ++++++++++++++++----------------- 3 files changed, 100 insertions(+), 72 deletions(-) diff --git a/js/net/src/origin.test.ts b/js/net/src/origin.test.ts index 4bcd648052..11fc01ea24 100644 --- a/js/net/src/origin.test.ts +++ b/js/net/src/origin.test.ts @@ -2150,19 +2150,33 @@ test("the newest epoch wins the path over a cheaper one and arrives as a new bro origin.close(); }); -test("demand naming an epoch is refused when the route changes instance while it waits", async () => { +const NEWER = Epoch.parse("01900000-0000-7000-8000-000000000002"); + +test("demand held across an epoch change asks again, unless it named the old epoch", async () => { const origin = new Producer(); const consumer = origin.consume(); + const path = Path.from("live/cam"); const handle = origin.dynamic(Path.from("live"), { epoch: EPOCH }); const it = handle.requested(); - const pending = wireOf(consumer).demand(Path.from("live/cam"), EPOCH); + const pinned = wireOf(consumer).demand(path, EPOCH); + const unpinned = wireOf(consumer).demand(path); const { value: req } = await it.next(); - // Another instance now serves the route; its answer is not the one the peer named. - handle.update({ ...handle.route, epoch: Epoch.mint() }); + // Another instance now serves the route: the old handler's answer is for the old one. + handle.update({ ...handle.route, epoch: NEWER }); req?.accept(new BroadcastProducer().consume()); - await expect(pending).rejects.toThrow("unroutable"); + await expect(pinned).rejects.toThrow("unroutable"); + const next = new BroadcastProducer(); + const track = next.createTrack("video"); + (await it.next()).value?.accept(next); + const subscription = (await unpinned)?.track("video").subscribe(); + track.writeGroup(new GroupProducer(7)); + expect((await subscription?.recvGroup())?.sequence).toBe(7); + + subscription?.close(); + await it.return?.(); + next.close(); handle.close(); origin.close(); }); @@ -2233,34 +2247,37 @@ test("a newer epoch ends pinned and unpinned requests on the replaced publisher origin.close(); }); -test("request never accepts an old handler after its epoch changes asynchronously", async () => { +test("a request held across an epoch change asks again and never takes the old handler's answer", async () => { const origin = new Producer(); const prefix = Path.from("room"); const path = Path.from("room/cam"); const dynamic = origin.dynamic(prefix, { epoch: EPOCH }); - const pending = origin.consume().request(path, { epoch: EPOCH }); + const unpinned = origin.consume().request(path); + const pinned = origin.consume().request(path, { epoch: EPOCH }); const requests = dynamic.requested(); const old = (await requests.next()).value; - const epoch = Epoch.parse("01900000-0000-7000-8000-000000000002"); - dynamic.update({ ...dynamic.route, epoch }); - const current = origin.consume().request(path, { epoch }); + dynamic.update({ ...dynamic.route, epoch: NEWER }); + const current = origin.consume().request(path, { epoch: NEWER }); old?.accept(new BroadcastProducer().consume()); - expect(pending.active.peek()).toBeUndefined(); - expect(pending.unroutable.peek()).toBe(true); + await settle(); + expect(pinned.active.peek()).toBeUndefined(); + expect(pinned.unroutable.peek()).toBe(true); + expect(unpinned.active.peek()).toBeUndefined(); expect(current.active.peek()).toBeUndefined(); const next = (await requests.next()).value; next?.accept(new BroadcastProducer().consume()); - expect(current.active.peek()?.epoch).toBe(epoch); - expect(pending.active.peek()).toBeUndefined(); + await settle(); + expect(unpinned.active.peek()?.epoch).toBe(NEWER); + expect(current.active.peek()?.epoch).toBe(NEWER); + expect(pinned.active.peek()).toBeUndefined(); await requests.return?.(); - pending.close(); + unpinned.close(); + pinned.close(); current.close(); dynamic.close(); origin.close(); }); -const NEWER = Epoch.parse("01900000-0000-7000-8000-000000000002"); - test.each([ ["A to B", EPOCH, NEWER], ["A to none", EPOCH, undefined], diff --git a/js/net/src/origin.ts b/js/net/src/origin.ts index b6dbc7368f..104ba51a42 100644 --- a/js/net/src/origin.ts +++ b/js/net/src/origin.ts @@ -367,7 +367,10 @@ class ServeState { this.reset(err); } - /** Release answers and invalidate pending handlers when the publisher instance changes. */ + /** + * Release answers and the requests handlers hold when the publisher instance changes: a + * handler's late answer is for the old one, and the requesters ask the new one again. + */ reset(err: Error = unroutable()): void { const queued = [...this.pending.values()]; this.pending.clear(); @@ -1740,6 +1743,7 @@ export class Consumer { const live = server.served.get(path); if (live && live.closed.peek() === undefined) return live; + const asked = entry.route.peek().epoch; server.enqueue(path); server.demanding.set(path, (server.demanding.get(path) ?? 0) + 1); try { @@ -1759,7 +1763,11 @@ export class Consumer { if (closed) throw closed; return undefined; } - if (!server.pending.has(path)) return undefined; + if (!server.pending.has(path)) { + // Released by an epoch change: its answer belongs to the old instance, so ask the new one. + if (entry.route.peek().epoch !== asked) return await this.#demand(path, epoch); + return undefined; + } await Signal.race(server.settled, server.closed); } } finally { @@ -1810,9 +1818,12 @@ export class Dynamic { * The route is taken as given. At the same epoch this re-prices: consumers see an update * and every handle survives. Another epoch, or none, names another publisher instance: * consumers see a restart, the requests resolved through the old one end, and a - * re-request never joins it. Requests this handle still holds are refused as unroutable, - * and the next request for a path it served asks the handler again. The broadcasts it - * served keep running for the subscriptions already on them; close them to end those too. + * re-request never joins it. Requests still waiting on this handle carry over: the handler + * is asked again under the new epoch, and its answer to a request asked before the change + * is dropped, never served under the new epoch. A request pinned to the old epoch is + * refused as unroutable. The answers already served are forgotten, so the next request + * for one of those paths asks the handler again too. The broadcasts it served keep + * running for the subscriptions already on them; close them to end those too. * Route selection still applies: another route still at the old epoch outranks one * without. To re-price, start from the current route, `update({ ...dynamic.route, cost })`. */ diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 0cdc75c84e..11db11285c 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -2921,11 +2921,13 @@ async fn serve_front(task: FrontTask) -> Vec { instances.insert(id, asked); Event::Resolved { route, result: Ok(id) } } + // A refusal for an instance the route has since left, such as a request + // released by an epoch change, is not the new one's answer either: ask again. Err(err) => Event::Resolved { route, result: Err(Refusal { err, - standing: standing(&front, route, &resolved), + standing: !stale(&front, route, &asked, &resolved) && standing(&front, route, &resolved), }), }, } @@ -3727,11 +3729,15 @@ impl Dynamic { /// At the same [`Route::epoch`] this re-prices: consumers observe an update and /// every handle survives. Another epoch, or none, names another publisher /// instance: consumers see an [`AnnounceEvent::Restart`], and a re-request never - /// joins what was resolved under the old one. Requests this handle still holds - /// are refused with [`Error::Unroutable`], and the next request for a path it - /// served asks the handler again. The broadcasts it served keep running for the - /// subscriptions already on them; close them to end those too. Route selection - /// still applies: another route still at the old epoch outranks one without. + /// joins what was resolved under the old one. Requests still waiting on this + /// handle carry over: the handler is asked again under the new epoch, and its + /// answer to a request asked before the change is dropped, never served under the + /// new epoch. A request pinned to the old epoch is refused with + /// [`Error::Unroutable`]. The answers already served are forgotten, so the next + /// request for one of those paths asks the handler again too. The broadcasts it + /// served keep running for the subscriptions already on them; close them to end + /// those too. Route selection still applies: another route still at the old epoch + /// outranks one without. /// To re-price, start from the current route: /// `dynamic.update(dynamic.route().with_cost(cost))`. /// @@ -3805,8 +3811,8 @@ impl ServeState { if state.closed { return; } - // Refused already (the route changed instance): that answer stands and nothing - // is cached for the next request. + // Released already (the route changed instance): the answer was for the old one, + // so nothing resolves or caches it. if state.requests.remove_if(path, |p| p.same_channel(producer)).is_none() { return; } @@ -3837,8 +3843,9 @@ impl ServeState { } } - /// Refuse every request, queued or handed to a handler, and forget the answers served, - /// because the route now names another publisher instance. + /// Release every request, queued or handed to a handler, and forget the answers served, + /// because the route now names another publisher instance. The fronts waiting on a + /// released request ask again under the new one; the handler's answer to it is dropped. fn renew(&mut self) { for producer in self.requests.drain_all() { if let Ok(mut request) = producer.write() { @@ -8200,67 +8207,60 @@ mod tests { deliver(&new_track, &mut subscription, b"new").await; } - /// A handler asked under one publisher instance does not answer for the next: the - /// route's epoch changing while the request waits refuses it, and its late answer - /// is neither handed out under the new epoch nor cached for the next request. - #[moq_net_sim::test] - async fn a_pending_answer_does_not_outlive_its_epoch() { + /// A request held when its route changes epoch carries over: the handler is asked + /// again under the new epoch rather than the requester failing. The old answer is + /// neither served under the new epoch nor cached for the next request, and a request + /// pinned to the old epoch is refused. `answer_first` lands the old answer just + /// before the change instead of after, before the front takes it. + async fn held_request_across_an_epoch_change(answer_first: bool) { let producer = origin(1).produce(); let consumer = producer.consume(); + let old = crate::Epoch::mint(); let dynamic = producer - .dynamic("room", Route::default().with_epoch(crate::Epoch::mint())) + .dynamic("room", Route::default().with_epoch(old.clone())) .unwrap(); let pending = consumer.request_broadcast("room/alice", None); + let pinned = consumer.request_broadcast("room/alice", old); let request = queued(&dynamic).await; - let epoch = crate::Epoch::mint(); - dynamic.update(dynamic.route().with_epoch(epoch.clone())).unwrap(); let stale = broadcast::Info::new().produce(); let _stale_track = stale.create_track("video", None).unwrap(); - request.accept(&stale); - assert!( - matches!(pending.await, Err(Error::Unroutable)), - "an answer for the old instance resolved" - ); + let epoch = crate::Epoch::mint(); + if answer_first { + request.accept(&stale); + dynamic.update(dynamic.route().with_epoch(epoch.clone())).unwrap(); + } else { + dynamic.update(dynamic.route().with_epoch(epoch.clone())).unwrap(); + request.accept(&stale); + } - let retry = consumer.request_broadcast("room/alice", None); let request = queued(&dynamic).await; let fresh = broadcast::Info::new().produce(); let fresh_track = fresh.create_track("video", None).unwrap(); request.accept(&fresh); - let resolved = retry.await.expect("the new instance resolves"); + let resolved = pending.await.expect("the request carries over to the new instance"); assert_eq!(resolved.info().epoch.as_ref(), Some(&epoch)); + assert!( + matches!(pinned.await, Err(Error::Unroutable)), + "a request pinned to the old epoch resolved" + ); + let cached = dynamic.state.lock().served.get(&Path::new("room/alice").to_owned()); + assert!( + cached.is_some_and(|cached| cached.consume().is_clone(&fresh.consume())), + "the old answer was cached" + ); let mut subscription = resolved.track("video").unwrap().subscribe(None).await.unwrap(); deliver(&fresh_track, &mut subscription, b"new").await; } - /// An answer that lands just before its route changes instance, but that the front - /// has not taken yet, is not stamped with the new epoch: the front asks again. #[moq_net_sim::test] - async fn an_answer_before_an_epoch_change_is_asked_again() { - let producer = origin(1).produce(); - let consumer = producer.consume(); - let dynamic = producer - .dynamic("room", Route::default().with_epoch(crate::Epoch::mint())) - .unwrap(); - let pending = consumer.request_broadcast("room/alice", None); - let request = queued(&dynamic).await; - - // The answer and the change land in the same tick. - let stale = broadcast::Info::new().produce(); - let _stale_track = stale.create_track("video", None).unwrap(); - request.accept(&stale); - let epoch = crate::Epoch::mint(); - dynamic.update(dynamic.route().with_epoch(epoch.clone())).unwrap(); + async fn a_request_held_across_an_epoch_change_is_asked_again() { + held_request_across_an_epoch_change(false).await; + } - let request = queued(&dynamic).await; - let fresh = broadcast::Info::new().produce(); - let fresh_track = fresh.create_track("video", None).unwrap(); - request.accept(&fresh); - let resolved = pending.await.expect("the new instance resolves"); - assert_eq!(resolved.info().epoch.as_ref(), Some(&epoch)); - let mut subscription = resolved.track("video").unwrap().subscribe(None).await.unwrap(); - deliver(&fresh_track, &mut subscription, b"new").await; + #[moq_net_sim::test] + async fn an_answer_before_an_epoch_change_is_asked_again() { + held_request_across_an_epoch_change(true).await; } /// A route that gains an epoch and then drops it serves a third instance, not the From 3c08a6a32048046192c37002f97824419c6be7c5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 19:56:55 -0700 Subject: [PATCH 4/7] fix(net): mark a request released by an epoch change explicitly Comparing epochs missed a change and back (A to B to A) before the waiter looked: the request was released, but the epoch read as unchanged, so Rust refused it and JS demand came back empty. Rust now flags the release on the pending request, and JS demand counts resets. JS demand also loops instead of recursing. Co-Authored-By: Claude Opus 5.5 --- js/net/src/origin.test.ts | 27 ++++++++++ js/net/src/origin.ts | 92 ++++++++++++++++++---------------- rs/moq-net/src/model/origin.rs | 75 ++++++++++++++++++++------- 3 files changed, 133 insertions(+), 61 deletions(-) diff --git a/js/net/src/origin.test.ts b/js/net/src/origin.test.ts index 11fc01ea24..f47f8de4bf 100644 --- a/js/net/src/origin.test.ts +++ b/js/net/src/origin.test.ts @@ -2181,6 +2181,33 @@ test("demand held across an epoch change asks again, unless it named the old epo origin.close(); }); +test("demand held across an epoch round trip still asks again", async () => { + const origin = new Producer(); + const path = Path.from("live/cam"); + const handle = origin.dynamic(Path.from("live"), { epoch: EPOCH }); + const it = handle.requested(); + const demanded = wireOf(origin.consume()).demand(path); + const { value: req } = await it.next(); + + // Released by the first change; the epoch coming back does not undo that. + handle.update({ ...handle.route, epoch: NEWER }); + handle.update({ ...handle.route, epoch: EPOCH }); + req?.accept(new BroadcastProducer().consume()); + + const next = new BroadcastProducer(); + const track = next.createTrack("video"); + (await it.next()).value?.accept(next); + const subscription = (await demanded)?.track("video").subscribe(); + track.writeGroup(new GroupProducer(7)); + expect((await subscription?.recvGroup())?.sequence).toBe(7); + + subscription?.close(); + await it.return?.(); + next.close(); + handle.close(); + origin.close(); +}); + test("re-pricing from the current route keeps the epoch", () => { const origin = new Producer(); const handle = origin.dynamic(Path.from("live"), { epoch: EPOCH, cost: 5n }); diff --git a/js/net/src/origin.ts b/js/net/src/origin.ts index 104ba51a42..685c2358c3 100644 --- a/js/net/src/origin.ts +++ b/js/net/src/origin.ts @@ -311,6 +311,8 @@ class ServeState { demanding = new Map(); closed = new Once(); settled = new Signal(0); + /** Counts {@link reset}s, so a waiter can tell its request was released rather than answered. */ + resets = 0; onChange: (path: Path.Valid) => void = () => {}; onReject: (path: Path.Valid, err: Error) => void = () => {}; @@ -372,6 +374,7 @@ class ServeState { * handler's late answer is for the old one, and the requesters ask the new one again. */ reset(err: Error = unroutable()): void { + this.resets++; const queued = [...this.pending.values()]; this.pending.clear(); this.queue.mutate((queue) => { @@ -1728,52 +1731,55 @@ export class Consumer { * @internal */ async #demand(path: Path.Valid, epoch?: Epoch.Valid): Promise { - const local = this.#local(path); - if (local) { - // A peer naming one publisher instance is never handed another. - if (epoch !== undefined && this.#state.advertisedLocal.peek()?.get(path)?.epoch !== epoch) - throw unroutable(); - return local; - } - const entry = this.#state.bestEntry(path, received); - if (!entry?.server) return undefined; - if (epoch !== undefined && entry.route.peek().epoch !== epoch) throw unroutable(); - - const server = entry.server; - const live = server.served.get(path); - if (live && live.closed.peek() === undefined) return live; - - const asked = entry.route.peek().epoch; - server.enqueue(path); - server.demanding.set(path, (server.demanding.get(path) ?? 0) + 1); - try { - for (;;) { - // The route may name another instance by the time its handler answers. - if (epoch !== undefined && this.#state.bestEntry(path, received)?.route.peek().epoch !== epoch) + // Each pass asks the instance that wins now; an epoch change releasing the request starts another. + for (;;) { + const local = this.#local(path); + if (local) { + // A peer naming one publisher instance is never handed another. + if (epoch !== undefined && this.#state.advertisedLocal.peek()?.get(path)?.epoch !== epoch) throw unroutable(); - const served = server.served.get(path); - if (served && served.closed.peek() === undefined) return served; - const rejected = server.rejected.get(path); - if (rejected) { - server.rejected.delete(path); - throw rejected; - } - const closed = server.closed.peek(); - if (closed !== undefined) { - if (closed) throw closed; - return undefined; - } - if (!server.pending.has(path)) { - // Released by an epoch change: its answer belongs to the old instance, so ask the new one. - if (entry.route.peek().epoch !== asked) return await this.#demand(path, epoch); - return undefined; + return local; + } + const entry = this.#state.bestEntry(path, received); + if (!entry?.server) return undefined; + if (epoch !== undefined && entry.route.peek().epoch !== epoch) throw unroutable(); + + const server = entry.server; + const live = server.served.get(path); + if (live && live.closed.peek() === undefined) return live; + + const resets = server.resets; + server.enqueue(path); + server.demanding.set(path, (server.demanding.get(path) ?? 0) + 1); + try { + for (;;) { + // The route may name another instance by the time its handler answers. + if (epoch !== undefined && this.#state.bestEntry(path, received)?.route.peek().epoch !== epoch) + throw unroutable(); + const served = server.served.get(path); + if (served && served.closed.peek() === undefined) return served; + const rejected = server.rejected.get(path); + if (rejected) { + server.rejected.delete(path); + throw rejected; + } + const closed = server.closed.peek(); + if (closed !== undefined) { + if (closed) throw closed; + return undefined; + } + if (!server.pending.has(path)) { + // Released by an epoch change: its answer belongs to the old instance. + if (server.resets !== resets) break; + return undefined; + } + await Signal.race(server.settled, server.closed); } - await Signal.race(server.settled, server.closed); + } finally { + const n = (server.demanding.get(path) ?? 1) - 1; + if (n <= 0) server.demanding.delete(path); + else server.demanding.set(path, n); } - } finally { - const n = (server.demanding.get(path) ?? 1) - 1; - if (n <= 0) server.demanding.delete(path); - else server.demanding.set(path, n); } } } diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 11db11285c..de5abc216f 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -2399,7 +2399,8 @@ async fn serve_front(task: FrontTask) -> Vec { /// What the wait below returns: one thing that happened. enum Step { Assigned(track::Request), - Resolved(u64, Result), + /// The route, its answer, and whether an epoch change released the request. + Resolved(u64, Result, bool), SourceClosed(u64), Info(Arc, u64, Result), Ended(Arc, u64, Result<(), Error>, bool), @@ -2802,18 +2803,15 @@ async fn serve_front(task: FrontTask) -> Vec { } if let Some((route, _, pending)) = &upstream && let Poll::Ready(result) = pending.poll(waiter, |p| match &p.resolved { - Some(result) => Poll::Ready(result.clone()), + Some(result) => Poll::Ready((result.clone(), p.released)), None => Poll::Pending, }) { - return Poll::Ready(Step::Resolved( - *route, - match result { - Ok(resolved) => resolved, - // The queue died unresolved (its handler dropped): the route - // could not serve. - Err(_closed) => Err(Error::Unroutable), - }, - )); + return Poll::Ready(match result { + Ok((resolved, released)) => Step::Resolved(*route, resolved, released), + // The queue died unresolved (its handler dropped): the route + // could not serve. + Err(_closed) => Step::Resolved(*route, Err(Error::Unroutable), false), + }); } if let Some(id) = front.serving() && let Some(source) = sources.get(&id) @@ -2901,12 +2899,21 @@ async fn serve_front(task: FrontTask) -> Vec { now: timers.now(), } } - Step::Resolved(route, result) => { + Step::Resolved(route, result, released) => { let Some((_, asked, _)) = upstream.take() else { continue }; match result { - // The route moved to another instance, or another won the path, while - // the request was in flight: ask again rather than resolve requesters - // onto a broadcast that is already replaced. + // An epoch change released the request, the route moved to another + // instance, or another won the path, while the request was in flight: + // ask again rather than resolve requesters onto a broadcast that is + // already replaced. A release is its own signal, since the epoch may + // have come back by now. + _ if released => Event::Resolved { + route, + result: Err(Refusal { + err: Error::Unroutable, + standing: false, + }), + }, Ok(_) if stale(&front, route, &asked, &resolved) => Event::Resolved { route, result: Err(Refusal { @@ -2921,13 +2928,11 @@ async fn serve_front(task: FrontTask) -> Vec { instances.insert(id, asked); Event::Resolved { route, result: Ok(id) } } - // A refusal for an instance the route has since left, such as a request - // released by an epoch change, is not the new one's answer either: ask again. Err(err) => Event::Resolved { route, result: Err(Refusal { err, - standing: !stale(&front, route, &asked, &resolved) && standing(&front, route, &resolved), + standing: standing(&front, route, &resolved), }), }, } @@ -3640,6 +3645,9 @@ impl OriginState { #[derive(Default)] struct PendingBroadcast { resolved: Option>, + /// Set when an epoch change released the request: whatever it resolved to speaks + /// for the old instance, so the front asks again. + released: bool, } /// A front's verdict, which its requesters wait on: the instance it resolved, or the @@ -3850,6 +3858,7 @@ impl ServeState { for producer in self.requests.drain_all() { if let Ok(mut request) = producer.write() { request.resolved.get_or_insert(Err(Error::Unroutable)); + request.released = true; } } self.served = WeakCache::default(); @@ -8263,6 +8272,36 @@ mod tests { held_request_across_an_epoch_change(true).await; } + /// A release is not undone by the epoch coming back before the front looks: a + /// request held across A to B to A is asked again, not refused. + #[moq_net_sim::test] + async fn a_request_held_across_an_epoch_round_trip_is_asked_again() { + let producer = origin(1).produce(); + let consumer = producer.consume(); + let old = crate::Epoch::mint(); + let dynamic = producer + .dynamic("room", Route::default().with_epoch(old.clone())) + .unwrap(); + let pending = consumer.request_broadcast("room/alice", None); + let request = queued(&dynamic).await; + + dynamic + .update(dynamic.route().with_epoch(crate::Epoch::mint())) + .unwrap(); + dynamic.update(dynamic.route().with_epoch(old.clone())).unwrap(); + let stale = broadcast::Info::new().produce(); + let _stale_track = stale.create_track("video", None).unwrap(); + request.accept(&stale); + + let fresh = broadcast::Info::new().produce(); + let fresh_track = fresh.create_track("video", None).unwrap(); + queued(&dynamic).await.accept(&fresh); + let resolved = pending.await.expect("the request carries over"); + assert_eq!(resolved.info().epoch.as_ref(), Some(&old)); + let mut subscription = resolved.track("video").unwrap().subscribe(None).await.unwrap(); + deliver(&fresh_track, &mut subscription, b"new").await; + } + /// A route that gains an epoch and then drops it serves a third instance, not the /// first: a request after the round trip is asked again rather than joining the /// front still serving the first instance's subscribers. From b68576a6165bac93f736e98b647598ebd98d0fa2 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 20:09:36 -0700 Subject: [PATCH 5/7] fix(net): count epoch-change releases instead of flagging the request An answer or refusal that landed just before the epoch change had already left the queue, so renew could not flag it: a refusal ended the held request, and an answer survived an A to B to A round trip. The front now snapshots the ServeState renewal count when it asks and compares it when the answer arrives. JS demand registers again after a reset and only uncounts its own registration, so a refused pinned demand no longer drops the count of a demand that carried over, which lost the replacement handler's refusal. Co-Authored-By: Claude Opus 5.5 --- js/net/src/origin.test.ts | 21 ++++++++ js/net/src/origin.ts | 17 ++++--- rs/moq-net/src/model/origin.rs | 93 ++++++++++++++++++++++++++-------- 3 files changed, 103 insertions(+), 28 deletions(-) diff --git a/js/net/src/origin.test.ts b/js/net/src/origin.test.ts index f47f8de4bf..821e1c6648 100644 --- a/js/net/src/origin.test.ts +++ b/js/net/src/origin.test.ts @@ -2181,6 +2181,27 @@ test("demand held across an epoch change asks again, unless it named the old epo origin.close(); }); +test("a demand refused across an epoch change does not lose the carried demand's answer", async () => { + const origin = new Producer(); + const consumer = origin.consume(); + const path = Path.from("live/cam"); + const handle = origin.dynamic(Path.from("live"), { epoch: EPOCH }); + const it = handle.requested(); + const unpinned = wireOf(consumer).demand(path); + const pinned = wireOf(consumer).demand(path, EPOCH); + await it.next(); + + handle.update({ ...handle.route, epoch: NEWER }); + await expect(pinned).rejects.toThrow("unroutable"); + // The replacement's refusal reaches the demand that carried over. + (await it.next()).value?.reject(new Error("gone")); + await expect(unpinned).rejects.toThrow("gone"); + + await it.return?.(); + handle.close(); + origin.close(); +}); + test("demand held across an epoch round trip still asks again", async () => { const origin = new Producer(); const path = Path.from("live/cam"); diff --git a/js/net/src/origin.ts b/js/net/src/origin.ts index 685c2358c3..a697275f7e 100644 --- a/js/net/src/origin.ts +++ b/js/net/src/origin.ts @@ -1768,17 +1768,18 @@ export class Consumer { if (closed) throw closed; return undefined; } - if (!server.pending.has(path)) { - // Released by an epoch change: its answer belongs to the old instance. - if (server.resets !== resets) break; - return undefined; - } + // Released by an epoch change, which also dropped this pass's count: ask again. + if (server.resets !== resets) break; + if (!server.pending.has(path)) return undefined; await Signal.race(server.settled, server.closed); } } finally { - const n = (server.demanding.get(path) ?? 1) - 1; - if (n <= 0) server.demanding.delete(path); - else server.demanding.set(path, n); + // A reset already cleared the count, and a later pass may have registered anew. + if (server.resets === resets) { + const n = (server.demanding.get(path) ?? 1) - 1; + if (n <= 0) server.demanding.delete(path); + else server.demanding.set(path, n); + } } } } diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index de5abc216f..929588debf 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -907,6 +907,10 @@ struct ServeState { // long-lived origin serving many distinct one-shot paths stays bounded by the live count. served: WeakCache, + // How many times the route changed instance: a front compares it to the count when + // it asked to tell a released request from an answered one. + renewals: u64, + // Set when the announcement is retracted or the origin tears down: new requests // fail immediately and the handler observes the end instead of parking forever. closed: bool, @@ -2399,8 +2403,7 @@ async fn serve_front(task: FrontTask) -> Vec { /// What the wait below returns: one thing that happened. enum Step { Assigned(track::Request), - /// The route, its answer, and whether an epoch change released the request. - Resolved(u64, Result, bool), + Resolved(Result), SourceClosed(u64), Info(Arc, u64, Result), Ended(Arc, u64, Result<(), Error>, bool), @@ -2420,8 +2423,18 @@ async fn serve_front(task: FrontTask) -> Vec { // The instance each source's route served when it was asked: what its content is from. let mut instances: HashMap = HashMap::new(); let mut next_source = 0u64; - // The in-flight upstream request: the route, its instance when asked, and its pending channel. - let mut upstream: Option<(u64, Instance, kio::Consumer)> = None; + /// The in-flight upstream request. + struct Upstream { + route: u64, + /// The route's instance when asked. + asked: Instance, + pending: kio::Consumer, + /// The queue asked, and its [`ServeState::renewals`] then: a renewal since + /// released the request, whatever it resolved to and whenever. + server: kio::Shared, + renewals: u64, + } + let mut upstream: Option = None; let mut tracks: HashMap, TrackIo> = HashMap::new(); let mut deadline = crate::time::Deadline::new(&timers); // The watch generation the last selection saw. @@ -2600,7 +2613,15 @@ async fn serve_front(task: FrontTask) -> Vec { } } }; - upstream = Some((route, instance, pending)); + let renewals = serve.renewals; + drop(serve); + upstream = Some(Upstream { + route, + asked: instance, + pending, + server, + renewals, + }); } Action::Detach { source } => { // Readers keep reading the source's copy until a replacement is @@ -2801,17 +2822,17 @@ async fn serve_front(task: FrontTask) -> Vec { if let Poll::Ready(Ok(request)) = dynamic.poll_requested_track(waiter) { return Poll::Ready(Step::Assigned(request)); } - if let Some((route, _, pending)) = &upstream - && let Poll::Ready(result) = pending.poll(waiter, |p| match &p.resolved { - Some(result) => Poll::Ready((result.clone(), p.released)), + if let Some(upstream) = &upstream + && let Poll::Ready(result) = upstream.pending.poll(waiter, |p| match &p.resolved { + Some(result) => Poll::Ready(result.clone()), None => Poll::Pending, }) { - return Poll::Ready(match result { - Ok((resolved, released)) => Step::Resolved(*route, resolved, released), + return Poll::Ready(Step::Resolved(match result { + Ok(resolved) => resolved, // The queue died unresolved (its handler dropped): the route // could not serve. - Err(_closed) => Step::Resolved(*route, Err(Error::Unroutable), false), - }); + Err(_closed) => Err(Error::Unroutable), + })); } if let Some(id) = front.serving() && let Some(source) = sources.get(&id) @@ -2899,14 +2920,26 @@ async fn serve_front(task: FrontTask) -> Vec { now: timers.now(), } } - Step::Resolved(route, result, released) => { - let Some((_, asked, _)) = upstream.take() else { continue }; + Step::Resolved(result) => { + let Some(Upstream { + route, + asked, + server, + renewals, + .. + }) = upstream.take() + else { + continue; + }; + // Counted rather than read off the request, which an answer just before + // the change already took out of the queue, and rather than compared by + // epoch, which may have come back since. + let released = server.lock().renewals != renewals; match result { // An epoch change released the request, the route moved to another // instance, or another won the path, while the request was in flight: // ask again rather than resolve requesters onto a broadcast that is - // already replaced. A release is its own signal, since the epoch may - // have come back by now. + // already replaced. _ if released => Event::Resolved { route, result: Err(Refusal { @@ -3645,9 +3678,6 @@ impl OriginState { #[derive(Default)] struct PendingBroadcast { resolved: Option>, - /// Set when an epoch change released the request: whatever it resolved to speaks - /// for the old instance, so the front asks again. - released: bool, } /// A front's verdict, which its requesters wait on: the instance it resolved, or the @@ -3858,10 +3888,10 @@ impl ServeState { for producer in self.requests.drain_all() { if let Ok(mut request) = producer.write() { request.resolved.get_or_insert(Err(Error::Unroutable)); - request.released = true; } } self.served = WeakCache::default(); + self.renewals += 1; } /// Drop the still-pending entry, if it is still ours. @@ -8272,6 +8302,29 @@ mod tests { held_request_across_an_epoch_change(true).await; } + /// A refusal that lands just before an epoch change, before the front takes it, + /// was the old instance's answer: the request is asked again, not ended. + #[moq_net_sim::test] + async fn a_refusal_before_an_epoch_change_is_asked_again() { + let producer = origin(1).produce(); + let consumer = producer.consume(); + let dynamic = producer + .dynamic("room", Route::default().with_epoch(crate::Epoch::mint())) + .unwrap(); + let pending = consumer.request_broadcast("room/alice", None); + queued(&dynamic).await.reject(Error::NotFound); + let epoch = crate::Epoch::mint(); + dynamic.update(dynamic.route().with_epoch(epoch.clone())).unwrap(); + + let fresh = broadcast::Info::new().produce(); + let fresh_track = fresh.create_track("video", None).unwrap(); + queued(&dynamic).await.accept(&fresh); + let resolved = pending.await.expect("the request carries over"); + assert_eq!(resolved.info().epoch.as_ref(), Some(&epoch)); + let mut subscription = resolved.track("video").unwrap().subscribe(None).await.unwrap(); + deliver(&fresh_track, &mut subscription, b"new").await; + } + /// A release is not undone by the epoch coming back before the front looks: a /// request held across A to B to A is asked again, not refused. #[moq_net_sim::test] From 35087326b5dfbdd55c9c1bf6e21eac7bad14ef47 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 20:40:29 -0700 Subject: [PATCH 6/7] fix(net): read the renewal count without the serve lock renew releases held requests while the updater holds the ServeState lock, and the release wakes the front. Reading the count through that lock from the woken path could deadlock under an inline executor, so the count is a shared atomic, bumped before any release. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/origin.rs | 26 +++++++++++++------------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 929588debf..d7ffd110ba 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -908,8 +908,9 @@ struct ServeState { served: WeakCache, // How many times the route changed instance: a front compares it to the count when - // it asked to tell a released request from an answered one. - renewals: u64, + // it asked to tell a released request from an answered one. Shared outside the lock, + // since the release wakes the front while the updater still holds it. + renewals: Arc, // Set when the announcement is retracted or the origin tears down: new requests // fail immediately and the handler observes the end instead of parking forever. @@ -2429,10 +2430,10 @@ async fn serve_front(task: FrontTask) -> Vec { /// The route's instance when asked. asked: Instance, pending: kio::Consumer, - /// The queue asked, and its [`ServeState::renewals`] then: a renewal since - /// released the request, whatever it resolved to and whenever. - server: kio::Shared, - renewals: u64, + /// The queue's [`ServeState::renewals`], and its count when asked: a renewal + /// since released the request, whatever it resolved to and whenever. + renewals: Arc, + asked_renewals: u64, } let mut upstream: Option = None; let mut tracks: HashMap, TrackIo> = HashMap::new(); @@ -2613,14 +2614,12 @@ async fn serve_front(task: FrontTask) -> Vec { } } }; - let renewals = serve.renewals; - drop(serve); upstream = Some(Upstream { route, asked: instance, pending, - server, - renewals, + renewals: serve.renewals.clone(), + asked_renewals: serve.renewals.load(Ordering::Acquire), }); } Action::Detach { source } => { @@ -2924,8 +2923,8 @@ async fn serve_front(task: FrontTask) -> Vec { let Some(Upstream { route, asked, - server, renewals, + asked_renewals, .. }) = upstream.take() else { @@ -2934,7 +2933,7 @@ async fn serve_front(task: FrontTask) -> Vec { // Counted rather than read off the request, which an answer just before // the change already took out of the queue, and rather than compared by // epoch, which may have come back since. - let released = server.lock().renewals != renewals; + let released = renewals.load(Ordering::Acquire) != asked_renewals; match result { // An epoch change released the request, the route moved to another // instance, or another won the path, while the request was in flight: @@ -3885,13 +3884,14 @@ impl ServeState { /// because the route now names another publisher instance. The fronts waiting on a /// released request ask again under the new one; the handler's answer to it is dropped. fn renew(&mut self) { + // Counted before any release wakes a front, which reads the count without the lock. + self.renewals.fetch_add(1, Ordering::Release); for producer in self.requests.drain_all() { if let Ok(mut request) = producer.write() { request.resolved.get_or_insert(Err(Error::Unroutable)); } } self.served = WeakCache::default(); - self.renewals += 1; } /// Drop the still-pending entry, if it is still ours. From 983758a4fff5be450f3e06e1fa54390c43e989ab Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 20:51:29 -0700 Subject: [PATCH 7/7] fix(net): refuse renewed requests after the table lock drops Dynamic::update renewed the serve queue under the origin table lock, and the refusal woke each released request's front, which reads the table. renew now returns the released requests, and update refuses them once it holds no lock, so an inline executor never re-enters the table from the wake. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/origin.rs | 26 +++++++++++++++++--------- 1 file changed, 17 insertions(+), 9 deletions(-) diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index d7ffd110ba..c9641a496c 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -1990,6 +1990,9 @@ impl AnnounceProducer { if shared.closed { return Err(Error::Closed); } + // The requests an epoch change released, refused once the table lock is gone: the + // refusal wakes their fronts, which read the table. + let mut released = Vec::new(); for (prefix, id) in &self.entries { shared.reannounced(prefix, &route.hops); let stale = shared.withdrawn_through(prefix, &route.hops); @@ -2007,7 +2010,7 @@ impl AnnounceProducer { if renewed { entry.generation = generation; if let Some(server) = &entry.server { - server.lock().renew(); + released.extend(server.lock().renew()); } } entry.hops = route.hops.clone(); @@ -2023,6 +2026,12 @@ impl AnnounceProducer { shared.sync_route(prefix, &claim); shared.prune_withdrawn(prefix); } + drop(shared); + for producer in released { + if let Ok(mut request) = producer.write() { + request.resolved.get_or_insert(Err(Error::Unroutable)); + } + } Ok(()) } @@ -3881,17 +3890,16 @@ impl ServeState { } /// Release every request, queued or handed to a handler, and forget the answers served, - /// because the route now names another publisher instance. The fronts waiting on a - /// released request ask again under the new one; the handler's answer to it is dropped. - fn renew(&mut self) { + /// because the route now names another publisher instance. Returns the released + /// requests for the caller to refuse once it holds no lock, since the refusal wakes + /// their fronts. Those ask again under the new instance; the handler's answer to a + /// released request is dropped. + #[must_use = "the released requests must be refused"] + fn renew(&mut self) -> Vec> { // Counted before any release wakes a front, which reads the count without the lock. self.renewals.fetch_add(1, Ordering::Release); - for producer in self.requests.drain_all() { - if let Ok(mut request) = producer.write() { - request.resolved.get_or_insert(Err(Error::Unroutable)); - } - } self.served = WeakCache::default(); + self.requests.drain_all() } /// Drop the still-pending entry, if it is still ours.