From e828f695db77f740d7c99f6305761ac6c3c28f11 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 00:11:00 -0700 Subject: [PATCH 1/5] chore: claim quest/m1/papercuts Co-authored-by: Grok 4.7 From 3cc1e52308030a4d9c0a3d8aa1fe884e1a603ad2 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 00:24:37 -0700 Subject: [PATCH 2/5] fix(net): refuse to serve a broadcast a session delivered An originated route throws when accept is handed a session consumer, so upstream content is not labeled with this origin's hop. remote_wake_unparks waits until the worker parks instead of sleeping. Co-Authored-By: Grok 4.7 --- js/net/src/origin.test.ts | 45 +++++++++++++++++++++++++++++- js/net/src/origin.ts | 13 +++++++-- js/net/src/wire.ts | 14 ++++++++-- quest/m1/README.md | 1 - quest/m1/papercuts.md | 27 ------------------ quest/m1/uring-tests-under-load.md | 5 ++-- rs/moq-uring/src/worker.rs | 38 +++++++++++++++++++------ 7 files changed, 98 insertions(+), 45 deletions(-) delete mode 100644 quest/m1/papercuts.md diff --git a/js/net/src/origin.test.ts b/js/net/src/origin.test.ts index f1254022dd..1f26b15f8b 100644 --- a/js/net/src/origin.test.ts +++ b/js/net/src/origin.test.ts @@ -8,7 +8,7 @@ import { spreadHash } from "./internal.ts"; import type { Consumer, Table } from "./origin.ts"; import { Producer } from "./origin.ts"; import * as Path from "./path.ts"; -import { wireOf } from "./wire.ts"; +import { overrideBroadcastWire, wireOf } from "./wire.ts"; function publish(origin: Producer, path: Path.Valid) { const broadcast = origin.createBroadcast(path); @@ -1176,6 +1176,49 @@ test("a handle serves a request under live", async () => { origin.close(); }); +test("an originated route refuses a broadcast a session delivered", async () => { + const origin = new Producer(); + const handle = origin.dynamic(Path.from("live")); + const it = handle.requested(); + const pending = wireOf(origin.consume()).demand(Path.from("live/cam")); + const { value: req } = await it.next(); + if (!req) throw new Error("expected request"); + + // What lite and IETF do when a session hands back a consumed broadcast. + const foreign = new BroadcastProducer().consume(); + overrideBroadcastWire(foreign, {}); + expect(() => req.accept(foreign)).toThrow("origin cannot serve a broadcast it did not produce"); + + const produced = new BroadcastProducer(); + produced.createTrack("video"); + req.accept(produced.consume()); + expect(await pending).toBeDefined(); + + handle.close(); + produced.close(); + foreign.close(); + origin.close(); +}); + +test("a received route still serves a broadcast a session delivered", async () => { + const origin = new Producer(); + const path = Path.from("room"); + const upstream = new BroadcastProducer(); + upstream.createTrack("video"); + const foreign = upstream.consume(); + overrideBroadcastWire(foreign, {}); + const stop = serve(origin, path, () => foreign); + + const request = origin.request(path); + await settle(); + expect(request.active.peek()?.track("video").subscribe()).toBeDefined(); + + request.close(); + stop(); + upstream.close(); + origin.close(); +}); + test("a re-priced route is delivered as an update", async () => { const origin = new Producer(); const handle = origin.dynamic(Path.from("live")); diff --git a/js/net/src/origin.ts b/js/net/src/origin.ts index 2f63a1557c..edb136953c 100644 --- a/js/net/src/origin.ts +++ b/js/net/src/origin.ts @@ -281,9 +281,12 @@ function unroutable(): StreamError { /** A served route from {@link Producer.dynamic}: the queue a handler drains. */ class ServeState { readonly root: Path.Valid; + /** False when a session announced the route: it may serve that session's broadcast. */ + readonly originated: boolean; - constructor(root: Path.Valid) { + constructor(root: Path.Valid, originated: boolean) { this.root = root; + this.originated = originated; } queue = new Signal([]); @@ -860,7 +863,7 @@ export class Producer implements Table { #insertRoute(prefix: Path.Valid, route: Route, originated: boolean): Dynamic { prefix = this.#scope.prefix(prefix); - const server = new ServeState(this.#scope.root); + const server = new ServeState(this.#scope.root, originated); server.onChange = (path) => this.#state.refresh(path); const entry: RouteEntry = { identity: {}, @@ -1683,10 +1686,14 @@ export class Request { * Accept the request, resolving every awaiting requester with `broadcast`. * * The caller keeps producing into `broadcast`; repeat requests for the path share - * it for as long as it stays live. + * it for as long as it stays live. An originated route refuses a broadcast a session + * delivered: serving it would label upstream content with this origin's hop. */ accept(source: broadcast.Producer | broadcast.Consumer): void { if (this.#done) return; + if (this.#server.originated && !(source instanceof broadcast.Producer) && wireOf(source).fromSession) { + throw new Error("origin cannot serve a broadcast it did not produce"); + } this.#done = true; const front = source instanceof broadcast.Producer ? source.consume() : source; this.#server.accept(this, front); diff --git a/js/net/src/wire.ts b/js/net/src/wire.ts index a87b135cd1..214218e043 100644 --- a/js/net/src/wire.ts +++ b/js/net/src/wire.ts @@ -23,6 +23,11 @@ export interface Broadcast { resolveTrackInfo(name: string): Promise; fetchGroup(name: string, sequence: number, options?: track.FetchGroupOptions): Promise; requested(): Promise; + /** + * A session delivered this broadcast. An originated route refuses to serve it: + * publishing it would name this origin's hop for upstream content. + */ + fromSession?: boolean; } /** The protocol-facing operations behind an origin producer. */ @@ -86,14 +91,19 @@ export function registerWire(handle: object, view: View): void { views.set(handle, view); } -/** Replace selected operations on a broadcast's package-private view. */ +/** + * Replace selected operations on a broadcast's package-private view. + * + * Only a session-delivered broadcast does this, so the replacement is marked. + * An origin refuses to serve that broadcast on a route it originated. + */ export function overrideBroadcastWire( handle: broadcast.Consumer, overrides: Partial>, ): void { const view = views.get(handle); if (!view) throw new Error("broadcast has no wire view"); - views.set(handle, { ...(view as Broadcast), ...overrides }); + views.set(handle, { ...(view as Broadcast), ...overrides, fromSession: true }); } export function wireOf(handle: broadcast.Producer | broadcast.Consumer): Broadcast; diff --git a/quest/m1/README.md b/quest/m1/README.md index 21d2307fea..b44c0ffabd 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -155,7 +155,6 @@ blocks. The quests that gated m0 lines moved under them. - [Frame slot charge](/quest/m1/frame-slot-charge.md) - a group's frame slots past the first four count against the cache pool, including capacity a released group keeps - [Front deadlines](/quest/m1/front-deadline-index.md) - a front's per-event cost stops growing with its track count: an expiry index and per-track wakes, proven by a churn benchmark - [Listener deadlines](/quest/m1/listener-deadlines.md) - io_uring, HTTP/2, and the internal listener bound slow handshakes and headers, and iroh honors `quic.keep_alive` -- [Papercuts](/quest/m1/papercuts.md) - JS refuses to serve a broadcast it did not produce, and a uring test stops sleeping - [Front parking](/quest/m1/origin-front-parks.md) - an unroutable request waits on a front instead of re-asking on every route-table move - [Route wakes](/quest/m1/route-wakes.md) - a route change wakes only the fronts it can move, so pool churn stops scaling with served paths - [Copy-walk bench](/quest/m1/admission-bench.md) - a front's walk over its copies on a route change is benchmarked over tracks and copies diff --git a/quest/m1/papercuts.md b/quest/m1/papercuts.md deleted file mode 100644 index 80822b735d..0000000000 --- a/quest/m1/papercuts.md +++ /dev/null @@ -1,27 +0,0 @@ -# [S] Papercuts from the 2026-09-30 spawn - -## Goal - -Two small fixes found while landing the m0 quests: `@moq/net` refuses to -serve a broadcast it did not produce, and `remote_wake_unparks` passes under -load. - -## Plan - -- JS republish: `@moq/net` publishes only what it produces (decided with - moq-dev/moq#4599), naming its own origin. Requests already skip received - routes (`#demand` resolves through `bestEntry(path, received)` in - `js/net/src/origin.ts`), so the gap is an app handing a consumed - `broadcast.Consumer` (one a session delivered) back to an origin for - serving, e.g. `Request.accept(consumer)`, which labels upstream content - with the local origin's hop. Refuse that loudly at the point it enters the - origin, not in the lite publisher's `local(..) ?? demand(..)` resolve. - -- `rs/moq-uring/src/worker.rs` `remote_wake_unparks`: observe that the worker - parked instead of sleeping 50 ms and asserting elapsed wall time, per the - rule that unit tests mock time. -- Dropped in the 2026-10-05 audit: the `.scratch/` formatter excludes. #4598 - retired the agents' scratch-clone convention, and the only writer left is - the X11 capture benchmark's log, which no formatter touches. - -Public API: none. Wire: none. diff --git a/quest/m1/uring-tests-under-load.md b/quest/m1/uring-tests-under-load.md index b03aefd7a4..f78534d09d 100644 --- a/quest/m1/uring-tests-under-load.md +++ b/quest/m1/uring-tests-under-load.md @@ -17,13 +17,12 @@ Fix at the cause, never with a retry or longer timeout: `Error::ring` already reports `RLIMIT_MEMLOCK` on ENOMEM; check that this reaches the failing tests. A clearer failure alone does not meet the goal. - The worker tests: check the two named tests for wall-clock dependencies and - move them to mocked time or event assertions. `remote_wake_unparks` belongs - to [Papercuts](/quest/m1/papercuts.md), not this quest. + move them to mocked time or event assertions. Leave `remote_wake_unparks` + alone; it is not this quest. - Keep this resource-sharing follow-up standalone: it has no shared fixture with the current children of the load-flake questline. Run the affected tests during parallel checks and wire any new coverage into CI. ## Related -- [Papercuts](/quest/m1/papercuts.md) - owns the remote worker wake assertion - [More tests under load](/quest/m1/test-flakes-2/README.md) - the same cause-first rules and final loaded check apply here diff --git a/rs/moq-uring/src/worker.rs b/rs/moq-uring/src/worker.rs index 129f736a6d..8573b50a4f 100644 --- a/rs/moq-uring/src/worker.rs +++ b/rs/moq-uring/src/worker.rs @@ -1371,6 +1371,8 @@ mod tests { assert_eq!(metrics.snapshot().timers_active(), 0); } + /// A remote wake has to find the worker parked, then kick the futex. The park + /// word is that observation: a sleep would pass or fail with the load. #[test] fn remote_wake_unparks() { let metrics = Metrics::default(); @@ -1379,21 +1381,40 @@ mod tests { ..Default::default() }; let Some(mut worker) = worker_with(config) else { return }; + let unpark = worker.shared.unpark.clone(); let flag = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let waker_slot = std::sync::Arc::new(std::sync::Mutex::new(None::)); let thread_flag = flag.clone(); - let waker_slot = std::sync::Arc::new(std::sync::Mutex::new(None::)); let thread_slot = waker_slot.clone(); + let thread_metrics = metrics.clone(); let thread = std::thread::spawn(move || { - // Wait until the worker has parked on the future below. - std::thread::sleep(Duration::from_millis(50)); - thread_flag.store(true, Ordering::Release); - if let Some(waker) = thread_slot.lock().unwrap().take() { + loop { + while unpark.word.load(Ordering::Acquire) != PARKED { + std::hint::spin_loop(); + } + // Park stores the waker first. Cloning leaves it in place so a + // lost race can try again with whatever the next poll stored. + let Some(waker) = thread_slot.lock().unwrap().clone() else { + continue; + }; + let wakes = thread_metrics.snapshot().wakes; waker.wake(); + // EINTR can end the park between the load and this wake. Only a + // wake that found the word parked counts. + if thread_metrics.snapshot().wakes == wakes { + continue; + } + thread_flag.store(true, Ordering::Release); + // The poll after the futex wake may have run before the flag and + // parked again. Waking whatever it stored lets it observe the flag. + if let Some(waker) = thread_slot.lock().unwrap().take() { + waker.wake(); + } + break; } }); - let start = Instant::now(); worker .block_on(std::future::poll_fn(move |cx| { if flag.load(Ordering::Acquire) { @@ -1403,10 +1424,11 @@ mod tests { Poll::Pending })) .unwrap(); - assert!(start.elapsed() >= Duration::from_millis(50)); thread.join().unwrap(); + let snap = metrics.snapshot(); + assert!(snap.parks > 0, "the worker never parked: {snap:?}"); // The futex syscall the wake had to make is the expensive half, and the // only counter written from off the worker's thread. - assert!(metrics.snapshot().wakes > 0, "the remote wake went unreported"); + assert!(snap.wakes > 0, "the remote wake went unreported: {snap:?}"); } } From 4f53d4b7e4e355ccdcca090156230fe04e523c43 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 23:25:55 -0700 Subject: [PATCH 3/5] test(js/net): forward the untimed track by republishing a copy A JS origin now refuses to serve a broadcast a session delivered, so the forwarding node in the untimed test copies the track into a broadcast it produced. Document that JS apps do not proxy on Request.accept and in the net guide. Co-Authored-By: Claude Opus 5.5 --- doc/lib/js/net.md | 5 ++++- js/net/src/origin.ts | 9 +++++++-- js/net/src/untimed.test.ts | 28 ++++++++++++++++++++++++++-- 3 files changed, 37 insertions(+), 5 deletions(-) diff --git a/doc/lib/js/net.md b/doc/lib/js/net.md index 7adff50256..50a812af2c 100644 --- a/doc/lib/js/net.md +++ b/doc/lib/js/net.md @@ -104,7 +104,10 @@ Three operations, on an origin: claim should stay advertised; `close()` retracts it. A request beneath it that no announced local broadcast wins is an `Origin.Request` to `accept` or `reject`; reject what you will not serve rather than narrowing the claim, since a - route is always a prefix on every wire. + route is always a prefix on every wire. JS apps do not proxy: `accept` + throws on a broadcast a session delivered, since it would go out labeled + with this origin's hop. Copy its tracks into a broadcast you produce and + accept that instead. A route is a capability, not an inventory. `origin.announced(scope)` yields `Announce.Event` values. `kind` is `"start"`, `"update"` (a reprice in diff --git a/js/net/src/origin.ts b/js/net/src/origin.ts index ae678d3234..69c1421706 100644 --- a/js/net/src/origin.ts +++ b/js/net/src/origin.ts @@ -1713,8 +1713,13 @@ export class Request { * Accept the request, resolving every awaiting requester with `broadcast`. * * The caller keeps producing into `broadcast`; repeat requests for the path share - * it for as long as it stays live. An originated route refuses a broadcast a session - * delivered: serving it would label upstream content with this origin's hop. + * it for as long as it stays live. + * + * A JS app does not proxy: a broadcast a session delivered would go out labeled with + * this origin's hop. To serve upstream content, copy its tracks into a broadcast you + * produce and accept that. + * + * @throws Error when handed a broadcast a session delivered. The request stays open. */ accept(source: broadcast.Producer | broadcast.Consumer): void { if (this.#done) return; diff --git a/js/net/src/untimed.test.ts b/js/net/src/untimed.test.ts index 8c8c2e3f3c..508da31cb5 100644 --- a/js/net/src/untimed.test.ts +++ b/js/net/src/untimed.test.ts @@ -1,6 +1,7 @@ // A track keeps its timedness end to end: no receive path fills in arrival time, and a // wire that can't declare a timeline delivers frames untimed. Mirrors rs/moq-net/tests/untimed.rs. import { expect, test } from "bun:test"; +import { type Consumer as BroadcastConsumer, Producer as BroadcastProducer } from "./broadcast.ts"; import { accept, connect } from "./connection/index.ts"; import * as Ietf from "./ietf/index.ts"; import * as Lite from "./lite/index.ts"; @@ -75,6 +76,22 @@ async function roundTrip(wire: Wire, timescale: Timescale | undefined, timestamp return { info, frames }; } +/** Copy `input`'s video track into a broadcast this node produced, until `input` closes. */ +async function republish(input: BroadcastConsumer): Promise { + const subscription = input.track("video").subscribe(); + const copy = new BroadcastProducer(); + const video = copy.createTrack("video", await subscription.info()); + void (async () => { + const groups = subscription.ordered(); + for (let group = await groups.nextGroup(); group; group = await groups.nextGroup()) { + const out = video.appendGroup(); + for (let frame = await group.readFrame(); frame; frame = await group.readFrame()) out.writeFrame(frame); + out.close(); + } + })().finally(() => copy.close()); + return copy; +} + test("a forwarded untimed track stays untimed", async () => { // The publisher's hop (lite-04) can't declare a timeline, and the forwarding node must // not claim one on its moq-transport hop either. @@ -101,10 +118,16 @@ test("a forwarded untimed track stays untimed", async () => { } })(); - // Serve the downstream session from the upstream one. + // Serve the downstream session from the upstream one. A JS origin refuses to serve a + // broadcast a session delivered, so the forwarding node republishes a copy it produced. const dynamic = relay.dynamic(Path.from("test")); + const inputs: BroadcastConsumer[] = []; const forwarding = (async () => { - for await (const request of dynamic.requested()) request.accept(wireOf(ingest).consume(request.path)); + for await (const request of dynamic.requested()) { + const input = wireOf(ingest).consume(request.path); + inputs.push(input); + request.accept(await republish(input)); + } })(); const remote = wireOf(subscriber).consume(Path.from("test")); @@ -117,6 +140,7 @@ test("a forwarded untimed track stays untimed", async () => { broadcast.close(); await serving; remote.close(); + for (const input of inputs) input.close(); dynamic.close(); await forwarding; for (const session of [publisher, ingest, serve, subscriber]) session.abort(); From 518a5cb2ed8e532873270e60222784c9f08794f2 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 23:38:24 -0700 Subject: [PATCH 4/5] test(js/net): abort the republished track when the upstream copy fails The copier in the untimed forwarding test left a rejected task unhandled and closed a partial copy as a clean end. Close the track with the error instead, and still close the broadcast. Co-Authored-By: Claude Opus 5.5 --- js/net/src/untimed.test.ts | 19 +++++++++++++------ 1 file changed, 13 insertions(+), 6 deletions(-) diff --git a/js/net/src/untimed.test.ts b/js/net/src/untimed.test.ts index 508da31cb5..51deec76db 100644 --- a/js/net/src/untimed.test.ts +++ b/js/net/src/untimed.test.ts @@ -82,13 +82,20 @@ async function republish(input: BroadcastConsumer): Promise { const copy = new BroadcastProducer(); const video = copy.createTrack("video", await subscription.info()); void (async () => { - const groups = subscription.ordered(); - for (let group = await groups.nextGroup(); group; group = await groups.nextGroup()) { - const out = video.appendGroup(); - for (let frame = await group.readFrame(); frame; frame = await group.readFrame()) out.writeFrame(frame); - out.close(); + try { + const groups = subscription.ordered(); + for (let group = await groups.nextGroup(); group; group = await groups.nextGroup()) { + const out = video.appendGroup(); + for (let frame = await group.readFrame(); frame; frame = await group.readFrame()) out.writeFrame(frame); + out.close(); + } + } catch (err) { + // An upstream abort goes downstream as an abort, never as a clean end. + video.close(err instanceof Error ? err : new Error(String(err))); + } finally { + copy.close(); } - })().finally(() => copy.close()); + })(); return copy; } From f484c6eccf1fba6a46aa69856222f43d7d86e4ac Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 23:42:35 -0700 Subject: [PATCH 5/5] test(js/net): finish the republished track on a clean upstream end Closing the copy broadcast does not end a track inserted into it, so a reader waiting on the next group never saw the end. Co-Authored-By: Claude Opus 5.5 --- js/net/src/untimed.test.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/js/net/src/untimed.test.ts b/js/net/src/untimed.test.ts index 51deec76db..ca918dcb09 100644 --- a/js/net/src/untimed.test.ts +++ b/js/net/src/untimed.test.ts @@ -89,6 +89,8 @@ async function republish(input: BroadcastConsumer): Promise { for (let frame = await group.readFrame(); frame; frame = await group.readFrame()) out.writeFrame(frame); out.close(); } + // Closing the broadcast does not end a track inserted into it. + video.close(); } catch (err) { // An upstream abort goes downstream as an abort, never as a clean end. video.close(err instanceof Error ? err : new Error(String(err)));