fix(net): refuse to serve a broadcast a session delivered

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

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"));

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<Request[]>([]);
@@ -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);

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<BroadcastProducer> {
+	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();

js/net/src/wire.ts

@@ -23,6 +23,11 @@ export interface Broadcast {
 	resolveTrackInfo(name: string): Promise<track.Info>;
 	fetchGroup(name: string, sequence: number, options?: track.FetchGroupOptions): Promise<GroupConsumer>;
 	requested(): Promise<track.Request | undefined>;
+	/**
+	 * 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<Pick<Broadcast, "resolveTrackInfo" | "fetchGroup">>,
 ): 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;

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

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

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::<std::task::Waker>));
 
 		let thread_flag = flag.clone();
-		let waker_slot = std::sync::Arc::new(std::sync::Mutex::new(None::<std::task::Waker>));
 		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:?}");
 	}
 }