diff --git a/doc/lib/js/net.md b/doc/lib/js/net.md index ef22aa606a..8a1daa2673 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.test.ts b/js/net/src/origin.test.ts index 4560f7f521..5b14aa00dd 100644 --- a/js/net/src/origin.test.ts +++ b/js/net/src/origin.test.ts @@ -9,7 +9,7 @@ import type { Consumer, Table } from "./origin.ts"; import { Producer } from "./origin.ts"; import * as Path from "./path.ts"; import { Timescale } from "./time.ts"; -import { wireOf } from "./wire.ts"; +import { overrideBroadcastWire, wireOf } from "./wire.ts"; function publish(origin: Producer, path: Path.Valid) { const broadcast = origin.createBroadcast(path); @@ -1160,6 +1160,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 e6ff63e54c..69c1421706 100644 --- a/js/net/src/origin.ts +++ b/js/net/src/origin.ts @@ -287,9 +287,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([]); @@ -876,7 +879,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: {}, @@ -1711,9 +1714,18 @@ export class Request { * * The caller keeps producing into `broadcast`; repeat requests for the path share * 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; + 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/untimed.test.ts b/js/net/src/untimed.test.ts index 8c8c2e3f3c..ca918dcb09 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,31 @@ 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 () => { + 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(); + } + // 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))); + } 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 +127,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 +149,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(); 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 afe5acaece..8138cfdcbc 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -138,7 +138,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 - [Rust papercuts](/quest/m1/papercuts-rs.md) - HTTP refusals are counted, a `u64::MAX` resume is unbounded, GOING_AWAY falls forward, and GRO stride 0 cannot panic - [JS papercuts](/quest/m1/papercuts-js.md) - IETF status 0 keeps the subgroup open, a muted rendition change settles, and bad element attributes warn - [Front parking](/quest/m1/origin-front-parks.md) - an unroutable request waits on a front instead of re-asking on every route-table move 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:?}"); } }