From 755f2e708d16244ba274d1255c08f38a1b4d99d3 Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Fri, 6 Nov 2026 13:31:08 -0800
Subject: [PATCH 1/7] chore(quest): claim quest/m1/mock-clock-tests/rs-relay.md

Quest-Claim: 900e454fcb0275cb

From 8a165d33305124acb08aca727b53c0336da8a992 Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Fri, 6 Nov 2026 13:52:14 -0800
Subject: [PATCH 2/7] test(relay): auth and revalidate tests prove a session
 stayed open by its end reason

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 rs/moq-relay/src/auth.rs                      |   8 +-
 rs/moq-relay/src/session.rs                   |   9 +-
 .../tests/integration/auth_lifetime.rs        | 213 ++++++++++--------
 .../tests/integration/session_revalidate.rs   | 114 ++++++----
 4 files changed, 198 insertions(+), 146 deletions(-)

diff --git a/rs/moq-relay/src/auth.rs b/rs/moq-relay/src/auth.rs
index 720f3cbff8b..5e0bb770b67 100644
--- a/rs/moq-relay/src/auth.rs
+++ b/rs/moq-relay/src/auth.rs
@@ -1856,14 +1856,12 @@ mod tests {
 
 	#[tokio::test]
 	async fn dropping_the_last_auth_ends_admissions() {
+		use futures::FutureExt;
+
 		let (auth, mut admissions) = Auth::embedded("relay-1");
 		let clone = auth.clone();
 		drop(auth);
-		assert!(
-			kio::clock::timeout(std::time::Duration::ZERO, admissions.next())
-				.await
-				.is_err()
-		);
+		assert!(admissions.next().now_or_never().is_none(), "a clone still admits");
 		drop(clone);
 		assert!(admissions.next().await.is_none());
 	}
diff --git a/rs/moq-relay/src/session.rs b/rs/moq-relay/src/session.rs
index f3f4878d4e0..e76ea793502 100644
--- a/rs/moq-relay/src/session.rs
+++ b/rs/moq-relay/src/session.rs
@@ -590,6 +590,8 @@ mod tests {
 
 	#[tokio::test]
 	async fn a_push_wakes_the_registration_once() {
+		use futures::FutureExt;
+
 		let registry = Registry::new();
 		let request = request("abc", "/room", "127.0.0.1:1");
 		let reg = registry.register(request);
@@ -598,11 +600,6 @@ mod tests {
 		kio::clock::timeout(Duration::from_secs(1), reg.nudged())
 			.await
 			.expect("the registration woke");
-		assert!(
-			kio::clock::timeout(std::time::Duration::from_millis(20), reg.nudged())
-				.await
-				.is_err(),
-			"a single push wakes once"
-		);
+		assert!(reg.nudged().now_or_never().is_none(), "a single push wakes once");
 	}
 }
diff --git a/rs/moq-relay/tests/integration/auth_lifetime.rs b/rs/moq-relay/tests/integration/auth_lifetime.rs
index 6feab4c60f3..c3e6153c04f 100644
--- a/rs/moq-relay/tests/integration/auth_lifetime.rs
+++ b/rs/moq-relay/tests/integration/auth_lifetime.rs
@@ -54,8 +54,8 @@ struct Script {
 	seen: Arc<Mutex<Vec<Request>>>,
 	/// Answers each `end` only once a permit is added, when [`Self::hold_ends`] set it.
 	end_gate: Arc<Mutex<Option<Arc<tokio::sync::Semaphore>>>>,
-	/// Notified as each `end` arrives.
-	end_arrived: Arc<tokio::sync::Notify>,
+	/// Bumped as each request arrives, so a test awaits one instead of sleeping.
+	arrived: Arc<tokio::sync::watch::Sender<()>>,
 }
 
 impl Script {
@@ -66,7 +66,7 @@ impl Script {
 			revalidate_http: Arc::new(Mutex::new(None)),
 			seen: Arc::new(Mutex::new(Vec::new())),
 			end_gate: Arc::default(),
-			end_arrived: Arc::default(),
+			arrived: Arc::new(tokio::sync::watch::Sender::new(())),
 		}
 	}
 
@@ -94,13 +94,37 @@ impl Script {
 			.lock()
 			.unwrap()
 			.iter()
-			.filter(|r| matches!(r.event, Event::End { .. }))
+			.filter(|r| is_end(r))
 			.cloned()
 			.collect()
 	}
 
+	/// Wait until the requests seen so far satisfy `done`, and return them.
+	async fn until(&self, what: &str, done: impl Fn(&[Request]) -> bool) -> Vec<Request> {
+		let mut arrived = self.arrived.subscribe();
+		let waited = kio::clock::timeout(TIMEOUT, async {
+			loop {
+				let seen = self.seen.lock().unwrap().clone();
+				if done(&seen) {
+					return seen;
+				}
+				arrived.changed().await.expect("the script outlives its waiters");
+			}
+		})
+		.await;
+		waited.unwrap_or_else(|_| panic!("{what}: got {:?}", self.seen.lock().unwrap()))
+	}
+
+	/// Wait for at least `n` `end`s, and return every one.
+	async fn ends_at_least(&self, n: usize, what: &str) -> Vec<Request> {
+		self.until(what, |seen| seen.iter().filter(|r| is_end(r)).count() >= n)
+			.await;
+		self.ends()
+	}
+
 	async fn handle(State(script): State<Script>, Json(request): Json<Request>) -> Response {
 		script.seen.lock().unwrap().push(request.clone());
+		script.arrived.send_replace(());
 		let answer = match request.event {
 			Event::Connect | Event::Attach => script.connect.lock().unwrap().clone(),
 			Event::Revalidate if request.transport == moq_auth::Transport::Http => script
@@ -111,7 +135,6 @@ impl Script {
 				.unwrap_or_else(|| script.revalidate.lock().unwrap().clone()),
 			Event::Revalidate => script.revalidate.lock().unwrap().clone(),
 			Event::End { .. } => {
-				script.end_arrived.notify_one();
 				let gate = script.end_gate.lock().unwrap().clone();
 				if let Some(gate) = gate {
 					drop(gate.acquire().await);
@@ -408,6 +431,37 @@ async fn assert_closed(session: moq_tokio::Connection, within: Duration, what: &
 		.unwrap_or_else(|_| panic!("relay should close the {what} session"));
 }
 
+fn is_end(request: &Request) -> bool {
+	matches!(request.event, Event::End { .. })
+}
+
+/// The reason an `end` carries for a session its client closed.
+fn client_closed() -> moq_auth::lease::Reason {
+	"closed".into()
+}
+
+/// Close `sessions` from the client and assert each `end` reports that close, so the
+/// relay had ended none of them first. Earlier ends must have arrived already.
+async fn assert_open_until_closed(
+	script: &Script,
+	sessions: impl IntoIterator<Item = moq_tokio::Connection>,
+	what: &str,
+) {
+	let before = script.ends().len();
+	let mut closing = 0;
+	for session in sessions {
+		session.abort(moq_net::Error::Cancel);
+		closing += 1;
+	}
+	let ends = script.ends_at_least(before + closing, what).await;
+	for end in &ends[before..] {
+		let Event::End { reason, .. } = &end.event else {
+			unreachable!()
+		};
+		assert_eq!(*reason, client_closed(), "{what}: the relay ended the session first");
+	}
+}
+
 /// The scripted server admits, and every request carries what the relay knows.
 #[tokio::test]
 async fn admits_and_reports_the_session() {
@@ -562,18 +616,15 @@ async fn a_configured_token_is_the_connection_credential() {
 			})
 			.await
 			.expect("grant timeout");
-		} else {
-			// Lite-06 has no AUTH answer and a client is live before admission, so give a
-			// refusal time to land.
-			assert!(
-				kio::clock::timeout(Duration::from_millis(500), session.closed())
-					.await
-					.is_err(),
-				"{versions:?}: the relay refused the configured token"
-			);
 		}
 
-		let seen = script.seen.lock().unwrap().clone();
+		// Lite-06 has no AUTH answer and a client is live before admission, so wait for
+		// what the server was sent.
+		let seen = script
+			.until(&format!("{versions:?}: a connect"), |seen| {
+				seen.iter().any(|r| r.event == Event::Connect)
+			})
+			.await;
 		let connect = seen.iter().find(|r| r.event == Event::Connect).expect("a connect");
 		assert_eq!(connect.token, setup, "{versions:?}");
 		assert_eq!(connect.query.as_deref(), query, "{versions:?}");
@@ -760,19 +811,7 @@ async fn a_moved_tier_retags_the_live_session() {
 		"later presence and bytes move to the new label"
 	);
 
-	assert!(
-		kio::clock::timeout(Duration::from_millis(200), pub_session.closed())
-			.await
-			.is_err(),
-		"a tier change must not close the publisher"
-	);
-	assert!(
-		kio::clock::timeout(Duration::from_millis(200), sub_session.closed())
-			.await
-			.is_err(),
-		"a tier change must not close the subscriber"
-	);
-
+	assert_open_until_closed(&script, [pub_session, sub_session], "a tier change").await;
 	relay.abort();
 }
 
@@ -920,14 +959,7 @@ async fn a_rechecked_grant_resizes_live_sessions() {
 			}
 		}
 
-		for (session, what) in [(pub_session, "publisher"), (sub_session, "subscriber")] {
-			assert!(
-				kio::clock::timeout(Duration::from_millis(200), session.closed())
-					.await
-					.is_err(),
-				"{scheme}: a resize must not close the {what}"
-			);
-		}
+		assert_open_until_closed(&script, [pub_session, sub_session], &format!("{scheme}: a resize")).await;
 		relay.abort();
 	}))
 	.await;
@@ -970,9 +1002,7 @@ async fn a_refusal_closes_live_sessions() {
 		assert_closed(pub_session, Duration::from_secs(5), &format!("{label} publisher")).await;
 		assert_closed(sub_session, Duration::from_secs(5), &format!("{label} subscriber")).await;
 
-		kio::clock::sleep(Duration::from_millis(200)).await;
-		let ends = script.ends();
-		assert!(ends.len() >= 2, "{label}: an end per session, got {}", ends.len());
+		let ends = script.ends_at_least(2, &format!("{label}: an end per session")).await;
 		for end in &ends {
 			let Event::End { reason, .. } = &end.event else {
 				unreachable!()
@@ -1073,7 +1103,14 @@ async fn http_routes_hold_a_lease() {
 		Event::End { reason, .. } => reason.clone(),
 		_ => unreachable!(),
 	};
-	kio::clock::sleep(Duration::from_millis(200)).await;
+	let http_ends = |n: usize| {
+		move |seen: &[Request]| {
+			seen.iter()
+				.filter(|r| is_end(r) && r.transport == moq_auth::Transport::Http)
+				.count() >= n
+		}
+	};
+	script.until("the announced session ended", http_ends(1)).await;
 	let done = ends(&script);
 	assert_eq!(done.len(), 1, "the announced session ended");
 	assert_eq!(end_reason(&done[0]), "done".into());
@@ -1091,10 +1128,6 @@ async fn http_routes_hold_a_lease() {
 		.expect("body ended early");
 	assert_eq!(&first[..], b"hello");
 
-	// The body is still streaming, so the fetch's lease is still held.
-	kio::clock::sleep(Duration::from_millis(200)).await;
-	assert_eq!(ends(&script).len(), 1, "the fetch must not end while its body streams");
-
 	// A refusal on the fetch's re-check cuts the transfer; the publisher is untouched.
 	script.on_revalidate_http(Answer::Status(403));
 	let cut = kio::clock::timeout(Duration::from_secs(5), async {
@@ -1109,16 +1142,12 @@ async fn http_routes_hold_a_lease() {
 	.expect("the refused fetch kept streaming");
 	assert!(cut.is_err(), "a refused fetch must not end cleanly: {cut:?}");
 
-	kio::clock::sleep(Duration::from_millis(200)).await;
+	script.until("the fetch session ended", http_ends(2)).await;
 	let ended = ends(&script);
 	assert_eq!(ended.len(), 2, "the fetch session ended");
+	// Refused, not `done`: the lease was still held while the body streamed.
 	assert_eq!(end_reason(&ended[1]), moq_auth::lease::Reason::Refused);
-	assert!(
-		kio::clock::timeout(Duration::from_millis(200), pub_session.closed())
-			.await
-			.is_err(),
-		"refusing the fetch must not close the publisher"
-	);
+	assert_open_until_closed(&script, [pub_session], "refusing the fetch").await;
 
 	drop(group);
 	drop(track);
@@ -1135,26 +1164,22 @@ async fn the_end_carries_duration_and_bytes() {
 	let url: url::Url = format!("moql://127.0.0.1:{}/room?jwt=token", addr.port())
 		.parse()
 		.unwrap();
+	// Each lease starts after `dialing` and is reported ended before `reported`.
+	let dialing = kio::clock::now();
 	let (pub_session, sub_session) = connect_and_round_trip(&url).await;
-
-	kio::clock::sleep(Duration::from_millis(300)).await;
 	drop(pub_session);
 	drop(sub_session);
 
-	let deadline = kio::clock::now() + Duration::from_secs(5);
-	let ends = loop {
-		let ends = script.ends();
-		if ends.len() >= 2 {
-			break ends;
-		}
-		assert!(kio::clock::now() < deadline, "ends never arrived: {}", ends.len());
-		kio::clock::sleep(Duration::from_millis(50)).await;
-	};
+	let ends = script.ends_at_least(2, "an end per session").await;
+	let reported = kio::clock::now();
 	for end in &ends {
 		let Event::End { duration, bytes, .. } = &end.event else {
 			unreachable!()
 		};
-		assert!(*duration >= Duration::from_millis(200), "duration {duration:?}");
+		assert!(
+			!duration.is_zero() && *duration <= reported - dialing,
+			"duration {duration:?}"
+		);
 		assert!(bytes.sent > 0 && bytes.received > 0, "bytes {bytes:?}");
 	}
 
@@ -1207,9 +1232,7 @@ async fn the_drain_waits_for_a_closed_session_to_report_its_end() {
 	clock.advance(Duration::from_secs(1));
 	let closed = kio::clock::wall();
 	clients[0].abort(moq_net::Error::Cancel);
-	kio::clock::timeout(TIMEOUT, script.end_arrived.notified())
-		.await
-		.expect("first end");
+	script.ends_at_least(1, "first end").await;
 	assert_eq!(
 		observer.changed().await,
 		moq_relay::shutdown::Tally { open: 1, ending: 1 }
@@ -1226,9 +1249,7 @@ async fn the_drain_waits_for_a_closed_session_to_report_its_end() {
 		async move { observer.drained().await }
 	});
 	clock.advance(drain_timeout - Duration::from_secs(1));
-	kio::clock::timeout(TIMEOUT, script.end_arrived.notified())
-		.await
-		.expect("deadline end");
+	script.ends_at_least(2, "deadline end").await;
 	assert_eq!(
 		observer.changed().await,
 		moq_relay::shutdown::Tally { open: 0, ending: 2 }
@@ -1326,9 +1347,14 @@ async fn a_certificate_admits_only_what_the_server_grants() {
 		Patterns::new(),
 	)))
 	.await;
-	let (addr, relay) = spawn_quic_relay(build_auth(narrow), Some(root.clone())).await;
+	let cluster = crate::spawn_cluster(cluster::Options::default());
+	let relayed = cluster.origin.consume();
+	let (addr, relay) = spawn_quic_relay_with(build_auth(narrow), Some(root.clone()), cluster).await;
 	let url: url::Url = format!("moql://127.0.0.1:{}/mine", addr.port()).parse().unwrap();
 	let origin = moq_tokio::origin::spawn();
+	let _broadcast = origin
+		.route("cam", moq_net::origin::RouteOptions::default())
+		.expect("create broadcast");
 	let session = kio::clock::timeout(
 		TIMEOUT,
 		mtls_client()
@@ -1340,12 +1366,12 @@ async fn a_certificate_admits_only_what_the_server_grants() {
 	.await
 	.expect("connect timeout")
 	.expect("a scoped certificate is admitted");
-	assert!(
-		kio::clock::timeout(Duration::from_secs(2), session.closed())
-			.await
-			.is_err(),
-		"the scoped publisher stays admitted"
-	);
+	// The session can establish before the verdict; its announce reaching the relay
+	// is what proves the verdict admitted it.
+	kio::clock::timeout(TIMEOUT, relayed.follow("mine/cam", None).unwrap().available())
+		.await
+		.expect("the scoped publisher's announce never reached the relay")
+		.expect("announced broadcast resolves");
 	// A subscribe-only client has nothing granted, so it is refused at the handshake.
 	assert_refused_with(mtls_client(), &url).await;
 	// The rules are rooted at `/`, so a certificate dialed outside `mine` gets nothing.
@@ -1511,12 +1537,14 @@ async fn an_embedded_decider_admits_and_revokes() {
 	let publisher = decider.producers.lock().unwrap().remove(0);
 	publisher.revoke(moq_auth::lease::Reason::Refused);
 	assert_closed(pub_session, Duration::from_secs(5), "publisher").await;
-	assert!(
-		kio::clock::timeout(Duration::from_millis(200), sub_session.closed())
-			.await
-			.is_err(),
-		"revoking one lease must not close the other session"
-	);
+	// The subscriber's lease reports the client's own close, so the revocation had
+	// not ended it.
+	let subscriber = decider.producers.lock().unwrap().remove(0);
+	sub_session.abort(moq_net::Error::Cancel);
+	let (reason, _) = kio::clock::timeout(TIMEOUT, subscriber.closed())
+		.await
+		.expect("the subscriber's lease never ended");
+	assert_eq!(reason, client_closed(), "revoking one lease ended the other session");
 
 	// A refusal reaches the session's transport before it establishes.
 	let (refusing, mut refusals) = moq_relay::auth::Auth::embedded("test-relay");
@@ -1787,20 +1815,19 @@ async fn a_late_decision_within_the_deadline_admits() {
 		tokio::time::pause();
 		tokio::time::advance(wait).await;
 		tokio::time::resume();
-		admission.grant(moq_auth::lease::Consumer::fixed(grant(Duration::from_secs(3600))));
+		let (lease, consumer) = moq_auth::lease::Producer::new(grant(Duration::from_secs(3600)));
+		admission.grant(consumer);
 		let connection = kio::clock::timeout(TIMEOUT, connecting)
 			.await
 			.expect("the late decision never reached the client")
 			.expect("client task")
 			.unwrap_or_else(|err| panic!("{timeout:?}: refused after {wait:?}: {err}"));
-		// Admitted, so the session stays up; a refusal would have closed it by now.
-		assert!(
-			kio::clock::timeout(Duration::from_secs(1), connection.closed())
-				.await
-				.is_err(),
-			"{timeout:?}: closed after the decision {wait:?} in"
-		);
-		drop(connection);
+		// Admitted, so the lease lives until the client's own close ends it.
+		connection.abort(moq_net::Error::Cancel);
+		let (reason, _) = kio::clock::timeout(TIMEOUT, lease.closed())
+			.await
+			.expect("the lease never ended");
+		assert_eq!(reason, client_closed(), "{timeout:?}: ended after the decision {wait:?} in");
 
 		trigger.start();
 		kio::clock::timeout(TIMEOUT, running)
@@ -2034,9 +2061,7 @@ async fn request_token_lease_ends_before_the_connection_and_reopens_later() {
 	.unwrap();
 	assert!(script.ends().iter().all(|request| request.auth.is_none()));
 	drop(second);
-	kio::clock::timeout(TIMEOUT, script.end_arrived.notified())
-		.await
-		.unwrap();
+	script.ends_at_least(1, "the attach's end").await;
 	let ended = script.ends();
 	assert_eq!(ended.len(), 1);
 	assert_eq!(ended[0].id, attached[0].id);
diff --git a/rs/moq-relay/tests/integration/session_revalidate.rs b/rs/moq-relay/tests/integration/session_revalidate.rs
index cb58e328e4d..e6e70d89d00 100644
--- a/rs/moq-relay/tests/integration/session_revalidate.rs
+++ b/rs/moq-relay/tests/integration/session_revalidate.rs
@@ -31,6 +31,8 @@ struct Script {
 	connect: Arc<Mutex<Answer>>,
 	revalidate: Arc<Mutex<Answer>>,
 	seen: Arc<Mutex<Vec<Request>>>,
+	/// Bumped as each request arrives, so a test awaits one instead of sleeping.
+	arrived: Arc<tokio::sync::watch::Sender<()>>,
 }
 
 impl Script {
@@ -39,6 +41,7 @@ impl Script {
 			connect: Arc::new(Mutex::new(Answer::Grant(grant.clone()))),
 			revalidate: Arc::new(Mutex::new(Answer::Grant(grant))),
 			seen: Arc::new(Mutex::new(Vec::new())),
+			arrived: Arc::new(tokio::sync::watch::Sender::new(())),
 		}
 	}
 
@@ -70,8 +73,32 @@ impl Script {
 			.collect()
 	}
 
+	/// Wait until the requests seen so far satisfy `done`.
+	async fn until(&self, what: &str, done: impl Fn(&Self) -> bool) {
+		let mut arrived = self.arrived.subscribe();
+		kio::clock::timeout(TIMEOUT, async {
+			while !done(self) {
+				arrived.changed().await.expect("the script outlives its waiters");
+			}
+		})
+		.await
+		.unwrap_or_else(|_| panic!("{what}: got {:?}", self.events()));
+	}
+
+	/// The `end` of session `id`, once it arrives.
+	async fn end_of(&self, id: &str) -> moq_auth::lease::Reason {
+		self.until(&format!("{id} ended"), |script| script.ends().iter().any(|r| r.id == id))
+			.await;
+		let end = self.ends().into_iter().find(|r| r.id == id).unwrap();
+		match end.event {
+			Event::End { reason, .. } => reason,
+			_ => unreachable!(),
+		}
+	}
+
 	async fn handle(State(script): State<Script>, Json(request): Json<Request>) -> Response {
 		script.seen.lock().unwrap().push(request.clone());
+		script.arrived.send_replace(());
 		let answer = match request.event {
 			Event::Connect | Event::Attach => script.connect.lock().unwrap().clone(),
 			Event::Revalidate => script.revalidate.lock().unwrap().clone(),
@@ -116,6 +143,7 @@ struct Fixture {
 	url: url::Url,
 	internal: url::Url,
 	sessions: moq_relay::session::Registry,
+	stats: moq_net::stats::Registry,
 	_relay: tokio::task::JoinHandle<()>,
 }
 
@@ -142,8 +170,10 @@ impl Fixture {
 			let _ = internal.run().await;
 		});
 
+		let mut cluster = crate::spawn_cluster(cluster::Options::default());
+		cluster.stats = moq_net::stats::Registry::new(moq_net::stats::Config::new());
+		let stats = cluster.stats.clone();
 		let (url, handle) = if websocket {
-			let cluster = crate::spawn_cluster(cluster::Options::default());
 			let mut server_config = moq_tokio::listen::Config::default();
 			server_config.bind = Some("[::]:0".parse().unwrap());
 			server_config.tls.generate = vec!["localhost".into()];
@@ -169,7 +199,6 @@ impl Fixture {
 			let server = config.init(Default::default()).expect("server init");
 			let mut server = server.listen().await.expect("listen");
 			let port = server.tcp_local_addr().expect("TCP listener is configured").port();
-			let cluster = crate::spawn_cluster(cluster::Options::default());
 			let sessions = sessions.clone();
 			let handle = tokio::spawn(async move {
 				let mut id = 0;
@@ -190,6 +219,7 @@ impl Fixture {
 			url,
 			internal: format!("http://{internal_addr}").parse().unwrap(),
 			sessions,
+			stats,
 			_relay: handle,
 		}
 	}
@@ -306,14 +336,8 @@ async fn push_by_id_refuses_inside_the_cadence() {
 		.expect("a push must close the session well inside the cadence")
 		.expect_err("refused as Unauthorized");
 
-	kio::clock::sleep(Duration::from_millis(200)).await;
-	let ends = script.ends();
-	assert_eq!(ends.len(), 1);
-	assert_eq!(ends[0].id, id);
-	match &ends[0].event {
-		Event::End { reason, .. } => assert_eq!(*reason, moq_auth::lease::Reason::Refused),
-		other => panic!("expected end, got {other:?}"),
-	}
+	assert_eq!(script.end_of(&id).await, moq_auth::lease::Reason::Refused);
+	assert_eq!(script.ends().len(), 1);
 }
 
 /// A path pattern and a CIDR each match the right subset of three sessions.
@@ -325,7 +349,11 @@ async fn path_and_cidr_select_a_subset() {
 	let a = connect(fixture.path("/demo/one"), "127.0.0.1:0").await;
 	let b = connect(fixture.path("/demo/two"), "127.0.0.1:0").await;
 	let c = connect(fixture.path("/other"), "127.0.0.1:0").await;
-	wait_listed(&fixture, 3).await;
+	let listed = wait_listed(&fixture, 3).await;
+	let id_of = |path: &str| {
+		let session = listed.sessions.iter().find(|session| session.path == path);
+		session.unwrap_or_else(|| panic!("{path} is listed")).id.clone()
+	};
 
 	let (status, path) = fixture.list("?path=demo/**").await;
 	assert_eq!(status, reqwest::StatusCode::OK);
@@ -350,7 +378,11 @@ async fn path_and_cidr_select_a_subset() {
 	script.on_revalidate(Answer::Status(403));
 	let (status, nudged) = fixture.revalidate("?path=demo/**").await;
 	assert_eq!(status, reqwest::StatusCode::ACCEPTED);
-	assert_eq!(nudged.ids.len(), 2);
+	let mut ids = nudged.ids.clone();
+	ids.sort();
+	let mut matched = vec![id_of("/demo/one"), id_of("/demo/two")];
+	matched.sort();
+	assert_eq!(ids, matched, "only the matched sessions are re-checked");
 
 	kio::clock::timeout(Duration::from_secs(3), a.closed())
 		.await
@@ -360,10 +392,12 @@ async fn path_and_cidr_select_a_subset() {
 		.await
 		.expect("path match b")
 		.expect_err("refused");
-	assert!(
-		kio::clock::timeout(Duration::from_millis(200), c.closed())
-			.await
-			.is_err(),
+	// The unmatched session's end reports its client's own close, so the push had not
+	// ended it.
+	c.abort(moq_net::Error::Cancel);
+	assert_eq!(
+		script.end_of(&id_of("/other")).await,
+		"closed".into(),
 		"the unmatched session must stay"
 	);
 }
@@ -399,16 +433,11 @@ async fn empty_filter_reaches_all() {
 	assert_eq!(status, reqwest::StatusCode::ACCEPTED);
 	assert_eq!(nudged.ids.len(), 2);
 
-	kio::clock::timeout(Duration::from_secs(3), async {
-		loop {
-			if script.revalidate_ids().len() >= 2 {
-				break;
-			}
-			kio::clock::sleep(Duration::from_millis(20)).await;
-		}
-	})
-	.await
-	.expect("empty filter re-POSTs once per session");
+	script
+		.until("empty filter re-POSTs once per session", |script| {
+			script.revalidate_ids().len() >= 2
+		})
+		.await;
 }
 
 /// A retier arrives on the next reply after a push, and the session stays.
@@ -417,7 +446,7 @@ async fn a_pushed_retier_arrives_on_the_next_reply() {
 	let script = Script::new(grant());
 	let fixture = Fixture::tcp(build_auth(script.spawn().await)).await;
 	let session = connect(fixture.path("/room"), "127.0.0.1:0").await;
-	wait_listed(&fixture, 1).await;
+	let id = wait_listed(&fixture, 1).await.sessions[0].id.clone();
 
 	let mut moved = grant();
 	moved.tier = Some("gold".into());
@@ -425,20 +454,23 @@ async fn a_pushed_retier_arrives_on_the_next_reply() {
 	let (status, _) = fixture.revalidate("").await;
 	assert_eq!(status, reqwest::StatusCode::ACCEPTED);
 
-	kio::clock::timeout(Duration::from_secs(3), async {
-		loop {
-			if script.events().contains(&Event::Revalidate) {
-				break;
-			}
-			kio::clock::sleep(Duration::from_millis(20)).await;
-		}
-	})
-	.await
-	.expect("the push re-checked");
-	assert!(
-		kio::clock::timeout(Duration::from_millis(200), session.closed())
-			.await
-			.is_err(),
+	// A re-check applies a new tier only once it has decided to keep the session.
+	let gold = moq_net::stats::Tier::new(moq_net::Path::from_static("gold"));
+	let deadline = kio::clock::now() + TIMEOUT;
+	while !fixture
+		.stats
+		.snapshot()
+		.sessions()
+		.iter()
+		.any(|(tier, presence)| *tier == gold && presence.active() == 1)
+	{
+		assert!(kio::clock::now() < deadline, "the pushed retier never applied");
+		kio::clock::sleep(Duration::from_millis(20)).await;
+	}
+	session.abort(moq_net::Error::Cancel);
+	assert_eq!(
+		script.end_of(&id).await,
+		"closed".into(),
 		"a retier must not close the session"
 	);
 }

From 9fc89a6f0e642ae672b6262a31059742416d7942 Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Fri, 6 Nov 2026 14:01:16 -0800
Subject: [PATCH 3/7] test(relay): tenant and socket listings prove absence
 once synced

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 rs/moq-relay/tests/integration/socket_auth.rs | 10 +-
 .../tests/integration/tenant_matrix.rs        | 93 +++++++++++++++----
 2 files changed, 78 insertions(+), 25 deletions(-)

diff --git a/rs/moq-relay/tests/integration/socket_auth.rs b/rs/moq-relay/tests/integration/socket_auth.rs
index 6b394d201c8..1803c651b86 100644
--- a/rs/moq-relay/tests/integration/socket_auth.rs
+++ b/rs/moq-relay/tests/integration/socket_auth.rs
@@ -15,10 +15,6 @@ use moq_relay::{Accepted, Connection};
 
 const TIMEOUT: Duration = Duration::from_secs(10);
 
-/// Long enough for a listing to have reported a track had it listed one. The relay links
-/// are in memory, so this is many round trips.
-const SETTLE: Duration = Duration::from_millis(500);
-
 /// The socket prefix, and every token's root.
 const ROOT: &str = "room";
 
@@ -280,10 +276,8 @@ async fn a_reader_of_everything_lists_no_session() {
 			.await
 			.expect("listing timeout")
 			.expect("listing");
+		// The session tracks existed before this listing began, so a `Synced` first
+		// means it listed none of them.
 		assert!(matches!(event, Some(TracksEvent::Synced)), "{id} listed {event:?}");
-		assert!(
-			kio::clock::timeout(SETTLE, tracks.next()).await.is_err(),
-			"{id} listed a track"
-		);
 	}
 }
diff --git a/rs/moq-relay/tests/integration/tenant_matrix.rs b/rs/moq-relay/tests/integration/tenant_matrix.rs
index 45414539d36..a8fa9d0e8a4 100644
--- a/rs/moq-relay/tests/integration/tenant_matrix.rs
+++ b/rs/moq-relay/tests/integration/tenant_matrix.rs
@@ -8,9 +8,10 @@
 
 #![cfg(feature = "quic")]
 
-use std::collections::{BTreeMap, HashMap};
+use std::collections::{BTreeMap, BTreeSet, HashMap};
 use std::net::SocketAddr;
 use std::sync::{Arc, Mutex};
+use std::task::Poll;
 use std::time::Duration;
 
 use axum::extract::State;
@@ -27,10 +28,6 @@ use crate::refusals::{client, config};
 
 const TIMEOUT: Duration = Duration::from_secs(10);
 
-/// How long a broadcast that should never arrive gets to show up anyway, once
-/// everything published before it has.
-const SETTLE: Duration = Duration::from_millis(300);
-
 /// A millisecond track: the frames written here carry timestamps.
 fn timed() -> moq_net::track::Info {
 	moq_net::track::Info::default().with_timescale(moq_net::Timescale::MILLI)
@@ -69,6 +66,10 @@ impl Tenants {
 		let tenants = Self::default();
 		tenants.grant("a", tenant_a());
 		tenants.grant("b", tenant_b());
+		// Rooted where it dials, so it can publish at another tenant's root.
+		let mut open = rooted("");
+		open.root = None;
+		tenants.grant("open", open);
 		tenants
 	}
 
@@ -212,6 +213,11 @@ impl Publisher {
 	}
 
 	fn publish(&mut self, name: &str) {
+		self.publish_frame(name, b"hello");
+	}
+
+	/// Publish `name` holding one group with `payload` in `video`.
+	fn publish_frame(&mut self, name: &str, payload: &'static [u8]) {
 		let broadcast = self
 			.origin
 			.route(name, moq_net::origin::RouteOptions::default())
@@ -222,7 +228,7 @@ impl Publisher {
 			.expect("create track");
 		let mut group = track.append_group().expect("append group");
 		group
-			.write_frame(moq_net::Timestamp::ZERO, b"hello".as_ref())
+			.write_frame(moq_net::Timestamp::ZERO, payload)
 			.expect("write frame");
 		self.held.push(Box::new((broadcast, track, group)));
 	}
@@ -319,6 +325,49 @@ impl Watcher {
 	}
 }
 
+/// Every broadcast `url`'s scope announces, as a fresh subscriber reads it once synced
+/// with the relay: a name missing here was not announced when the relay answered.
+async fn synced_listing(url: url::Url) -> BTreeSet<String> {
+	let origin = moq_tokio::origin::spawn();
+	let mut routes = origin.consume().routes();
+	let _session = kio::clock::timeout(
+		TIMEOUT,
+		client()
+			.with_subscriber(origin)
+			.with_reconnect(false)
+			.connect(url.clone())
+			.established(),
+	)
+	.await
+	.expect("listing connect timeout")
+	.expect("listing connect failed");
+	let mut listed = BTreeSet::new();
+	kio::clock::timeout(TIMEOUT, async {
+		loop {
+			// Synced counts only once every event queued before it was read.
+			let next = moq_net::kio::wait(|waiter| match routes.poll_next(waiter) {
+				Poll::Ready(event) => Poll::Ready(event),
+				Poll::Pending => routes.poll_synced(waiter).map(|()| None),
+			})
+			.await;
+			use moq_net::origin::RouteEvent;
+			match next {
+				Some(RouteEvent::End(route)) => listed.remove(route.prefix.as_str()),
+				Some(
+					RouteEvent::Start(route)
+					| RouteEvent::Update(route)
+					| RouteEvent::Restart(route)
+					| RouteEvent::Takeover { announce: route, .. },
+				) => listed.insert(route.prefix.to_string()),
+				None => return,
+			};
+		}
+	})
+	.await
+	.unwrap_or_else(|_| panic!("{url} never synced"));
+	listed
+}
+
 /// Dial raw QUIC as moq-lite-06 and send a SETUP whose Path is exactly `target`,
 /// bytes no URL type touched. The relay may close it once admission decides;
 /// the caller watches the auth server for what it was admitted as.
@@ -413,9 +462,10 @@ async fn matrix(io_uring: bool) {
 	conn.close(driver::VarInt::from_u32(0), "done");
 	endpoint.close(driver::VarInt::from_u32(0), "done").await;
 
-	// Give anything that should not arrive time to.
-	kio::clock::sleep(SETTLE).await;
-	let a_seen = a_view.seen();
+	// B's announces were all decided at step 4, so a viewer synced since lists any
+	// that landed in A's scope.
+	let a_seen = synced_listing(relay.url("/", "a")).await;
+	assert!(a_seen.contains("secret/cam"), "A's own listing: {a_seen:?}");
 	assert!(
 		!a_seen
 			.iter()
@@ -427,7 +477,8 @@ async fn matrix(io_uring: bool) {
 		"B wrote into A through its read-only mount: {a_seen:?}"
 	);
 	// 2. B never sees A's secret.
-	let b_seen = b_view.seen();
+	let b_seen = synced_listing(relay.url("/tenant-a", "b")).await;
+	assert!(b_seen.contains("marker"), "B's own listing: {b_seen:?}");
 	assert!(
 		!b_seen.iter().any(|name| name.contains("secret")),
 		"B saw A's secret: {b_seen:?}"
@@ -494,14 +545,22 @@ async fn http_routes_keep_tenants_apart() {
 		assert!(body.contains("probe"), "B's listing is not its own root: {body}");
 	}
 
-	// A fetch waits up to 30 seconds for its broadcast to appear. A's secret is
-	// already here, so a fetch that would serve it answers at once; one still
-	// waiting has served nothing.
-	if let Ok(fetch) = kio::clock::timeout(Duration::from_secs(2), get("/fetch/tenant-a/secret/cam/video?jwt=b")).await
-	{
-		assert_ne!(fetch.expect("fetch request").status(), 200, "B fetched A's secret");
-	}
+	// A fetch serves the broadcast at its token's root, so B's lands on B's root,
+	// where a decoy waits, rather than on the dialed path, where A's secret is.
+	let mut decoy = Publisher::connect(relay.url("/", "open")).await;
+	decoy.publish_frame("tenant-b", b"decoy");
+	let mut fetch = get("/fetch/tenant-a/secret/cam/video?jwt=b")
+		.await
+		.expect("fetch request");
+	assert_eq!(fetch.status(), 200);
+	let body = kio::clock::timeout(TIMEOUT, fetch.chunk())
+		.await
+		.expect("fetch timeout")
+		.expect("fetch body")
+		.expect("fetch body ended early");
+	assert_eq!(&body[..], b"decoy", "B's fetch escaped its root");
 
+	drop(decoy);
 	drop(a);
 	drop(b);
 	relay.stop().await;

From 746df88fc984ee1a314a90aee37e0f21f1a1609c Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Fri, 6 Nov 2026 14:36:14 -0800
Subject: [PATCH 4/7] test(relay): the GOAWAY handover deadline runs on a
 manual clock

The healthy bar defaults to one second, so the old 120ms sleep never cleared it
and the test took the backoff path instead of the replacement dial it targets.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../tests/integration/goaway_cluster.rs       | 42 +++++++++++++------
 1 file changed, 29 insertions(+), 13 deletions(-)

diff --git a/rs/moq-relay/tests/integration/goaway_cluster.rs b/rs/moq-relay/tests/integration/goaway_cluster.rs
index fd70d041624..219449bcb60 100644
--- a/rs/moq-relay/tests/integration/goaway_cluster.rs
+++ b/rs/moq-relay/tests/integration/goaway_cluster.rs
@@ -803,8 +803,22 @@ fn goaway_handover_is_enforced_while_the_replacement_dial_hangs() {
 /// old one still draining. Redirect that dial at a listener that accepts TCP and
 /// then says nothing, so the handshake hangs, and the predecessor must still be
 /// closed on schedule rather than held for as long as the dial takes.
+///
+/// Time passes only when the test moves the clock, by no more than the handover,
+/// so no dial timeout can end the hang first.
 async fn goaway_handover_is_enforced_while_the_replacement_dial_hangs_inner() {
 	let _ = rind_rustls::default_provider().unwrap().install_default();
+	let clock = kio::clock::Manual::new().install(&tokio::runtime::Handle::current());
+	// The manual clock moves only when told, so the watchdog runs on tokio's.
+	#[expect(
+		clippy::disallowed_methods,
+		reason = "real sockets settle on tokio's clock, beside the installed manual clock"
+	)]
+	async fn within<T>(step: &str, fut: impl std::future::Future<Output = T>) -> T {
+		tokio::time::timeout(TEST_TIMEOUT, fut)
+			.await
+			.unwrap_or_else(|_| panic!("timed out: {step}"))
+	}
 
 	let upstream_origin = moq_tokio::origin::spawn();
 	let (port, mut accepted, _handle) = spawn_upstream(upstream_origin).await;
@@ -812,11 +826,13 @@ async fn goaway_handover_is_enforced_while_the_replacement_dial_hangs_inner() {
 	// Accepts the connection, then never speaks MoQ: the dial hangs in the handshake.
 	let black_hole = std::net::TcpListener::bind("127.0.0.1:0").expect("bind black hole");
 	let black_hole_port = black_hole.local_addr().expect("local addr").port();
+	let (dialed_tx, mut dialed) = tokio::sync::mpsc::unbounded_channel();
 	let _black_hole = std::thread::spawn(move || {
 		// Hold every accepted socket open and idle for the life of the test.
 		let mut held = Vec::new();
 		while let Ok((socket, _)) = black_hole.accept() {
 			held.push(socket);
+			let _ = dialed_tx.send(());
 		}
 	});
 
@@ -827,7 +843,8 @@ async fn goaway_handover_is_enforced_while_the_replacement_dial_hangs_inner() {
 	// The GOAWAY has to land on a *healthy* session, which is the path that goes
 	// straight into the replacement dial. Below this bar it takes the immediate
 	// redirect path instead, whose sleep polls the drain either way.
-	client_config.backoff.initial = Some((Duration::from_millis(50)).into());
+	let healthy = Duration::from_millis(100);
+	client_config.backoff.healthy = Some((healthy).into());
 	let client = client_config.init(Default::default()).expect("client init");
 
 	let url: Url = format!("tcp://127.0.0.1:{port}/").parse().expect("parse url");
@@ -842,7 +859,7 @@ async fn goaway_handover_is_enforced_while_the_replacement_dial_hangs_inner() {
 
 	// Clear the healthy bar, so the GOAWAY continues straight into the replacement
 	// dial rather than sleeping first.
-	kio::clock::sleep(Duration::from_millis(120)).await;
+	clock.advance(healthy);
 
 	// Send the client somewhere that will never finish handshaking. No wire timeout:
 	// our own `--connect-goaway-handover` is then the only thing bounding the old session.
@@ -853,17 +870,16 @@ async fn goaway_handover_is_enforced_while_the_replacement_dial_hangs_inner() {
 		)))
 		.expect("send goaway");
 
-	// The predecessor must close on its own deadline, not whenever that dial gives
-	// up. The bound is generous against a loaded runner but far below the seconds a
-	// hanging handshake takes, which is the gap being measured.
-	let sent = kio::clock::now();
-	let closed = kio::clock::timeout(handover * 10, server_session.closed()).await;
-	let elapsed = sent.elapsed();
-	assert!(
-		closed.is_ok() && elapsed < handover * 5,
-		"the old session outlived its {handover:?} handover by {elapsed:?}: \
-		 the replacement dial was never interrupted to enforce it"
-	);
+	// The handover deadline is set before the replacement dial starts, so once the
+	// dial reaches the black hole, passing the handover must close the predecessor
+	// while the dial still hangs.
+	within("the replacement dial reaches the black hole", dialed.recv()).await;
+	clock.advance(handover);
+	within(
+		"the old session outlived its handover: the replacement dial was never interrupted to enforce it",
+		server_session.closed(),
+	)
+	.await;
 
 	drop(connection);
 }

From d5a480a2c242f37c36646abd56297089c9d481f9 Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Fri, 6 Nov 2026 14:36:56 -0800
Subject: [PATCH 5/7] quest: narrow rs-relay to the mesh tests that remain

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 quest/m1/mock-clock-tests/README.md   |  2 +-
 quest/m1/mock-clock-tests/rs-relay.md | 53 ++++++++++++++++++++-------
 2 files changed, 40 insertions(+), 15 deletions(-)

diff --git a/quest/m1/mock-clock-tests/README.md b/quest/m1/mock-clock-tests/README.md
index f1df1479e8a..f85d34e2cbb 100644
--- a/quest/m1/mock-clock-tests/README.md
+++ b/quest/m1/mock-clock-tests/README.md
@@ -40,7 +40,7 @@ before starting, since tests land and move.
 
 - [js/net tests run on a mocked clock](/quest/m1/mock-clock-tests/js-net.md) - about 120 js/net tests stop waiting on real time, through their shared helpers
 - [Other JS tests run on a mocked clock](/quest/m1/mock-clock-tests/js-rest.md) - hang's `settle(20)`, publish's fanout helpers, and the rest of JS
-- [moq-relay tests run on a mocked clock](/quest/m1/mock-clock-tests/rs-relay.md) - the relay's `timeout(..).is_err()` session proofs and settle sleeps
+- [moq-relay's mesh tests run on a mocked clock](/quest/m1/mock-clock-tests/rs-relay.md) - the relay's multi-relay quiet windows and settle sleeps
 - [Transport crate tests run on a mocked clock](/quest/m1/mock-clock-tests/rs-transport.md) - moq-tokio, moq-quic, moq-sock, and moq-uring
 - [Media crate tests run on a mocked clock](/quest/m1/mock-clock-tests/rs-media.md) - moq-audio, moq-video, moq-hls, moq-mux, and the bridges
 - [Binding and CLI tests run on a mocked clock](/quest/m1/mock-clock-tests/rs-bindings.md) - moq-ffi, moq-c, moq-cli, and kio
diff --git a/quest/m1/mock-clock-tests/rs-relay.md b/quest/m1/mock-clock-tests/rs-relay.md
index 2488dd3ebf8..ded51d29ecf 100644
--- a/quest/m1/mock-clock-tests/rs-relay.md
+++ b/quest/m1/mock-clock-tests/rs-relay.md
@@ -1,21 +1,46 @@
-# [M] moq-relay tests run on a mocked clock
+# [M] moq-relay's mesh tests run on a mocked clock
 
 ## Goal
 
-moq-relay's tests stop proving a session stayed open with
-`timeout(200ms, session.closed()).is_err()` on the real clock, and stop
-sleeping to settle. See the [epic](/quest/m1/mock-clock-tests/README.md)
-for the rules.
+moq-relay's multi-relay and process-signal tests stop proving that nothing
+happened by waiting out a quiet window, and stop sleeping to settle. See the
+[epic](/quest/m1/mock-clock-tests/README.md) for the rules.
 
 ## Plan
 
-As of 2026-11-06, 28 tests: `tests/integration/auth_lifetime.rs` 9,
-`session_revalidate.rs` 4, `public_rules.rs` 3, `goaway_cluster.rs` 2,
-`runtime_uring.rs` 2, `tenant_matrix.rs` 2, `web.rs` 2, and one each in
-`cluster_unknown.rs`, `socket_auth.rs`, `src/auth.rs`, and `src/session.rs`.
-Many run over real sockets: assert on the event the relay emits (a
-revalidation, a close reason), or drive the deadline with the
-[simulator or a `Manual` clock](/quest/m0/test-flakes-2/README.md), rather
-than waiting out a window.
+The single-relay tests are done: a session that stays open is proven by the
+`end` its lease reports for the client's own close, an absent broadcast by a
+listing read to `Synced`, and the GOAWAY handover by a manual clock. What is
+left, as of 2026-11-06:
 
-Public API: none. Wire: none.
+- `cluster_unknown.rs`: `assert_unknown_publisher_stays_announced` watches for
+  a flap for `SETTLE` (1s) after the frame reaches D.
+- `goaway_cluster.rs`: `cluster_migrates_on_upstream_goaway` and
+  `cluster_continues_a_group_split_by_goaway` end on a 500ms window with no
+  retraction, and the second also on a 500ms window with no downstream GOAWAY.
+- `cluster_drain.rs`: `pass` moves a manual clock in steps and sleeps `SETTLE`
+  on tokio's clock after each, so real sockets catch up.
+- `shutdown_signal.rs`: `an_embedder_owns_the_signals` waits `DRAIN_TIMEOUT / 2`
+  for a GOAWAY that must not come.
+
+Each crosses a mesh of relays over loopback, where no event marks convergence:
+`Synced` is per hop, origin cursors coalesce across prefixes, and retractions
+are held for `update_hold`. A quiet window is the only proof that fits today.
+Only the simulator's `idle` proves a mesh has settled, and the cluster dials
+peers only through `moq_tokio` sockets.
+
+Tests that only wait for progress, or bound real-socket timing from below, stay
+on the real clock: the `runtime_uring.rs` and listing poll loops, `web.rs`'s
+`start.elapsed() >= timeout`, `bitrate_caps.rs`'s `LEAST`, and
+`shutdown_signal.rs`'s drain-window lower bounds.
+
+Public API: none, unless the maintainer picks the dial hook. Wire: none.
+
+## Waiting
+
+- Maintainer: how mesh tests prove convergence. (1) Recommended: give
+  `cluster::Options` a dial hook that an in-memory link can fill, so these tests
+  run on `kio::clock::sim` and end on `sim::idle`; this adds public API whose only
+  consumer is tests. (2) Keep these windows as documented real-clock exceptions
+  and opt them out of the [Rust guard](/quest/m0/clock-guard/rust.md). Check: a
+  recorded pick in this file's Plan.

From d12cf09d4afde4ef93b488126334fc85de0f3f0a Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Fri, 6 Nov 2026 14:50:42 -0800
Subject: [PATCH 6/7] style: rustfmt

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 rs/moq-relay/tests/integration/auth_lifetime.rs      | 6 +++++-
 rs/moq-relay/tests/integration/session_revalidate.rs | 6 ++++--
 2 files changed, 9 insertions(+), 3 deletions(-)

diff --git a/rs/moq-relay/tests/integration/auth_lifetime.rs b/rs/moq-relay/tests/integration/auth_lifetime.rs
index c3e6153c04f..b729f37ed18 100644
--- a/rs/moq-relay/tests/integration/auth_lifetime.rs
+++ b/rs/moq-relay/tests/integration/auth_lifetime.rs
@@ -1827,7 +1827,11 @@ async fn a_late_decision_within_the_deadline_admits() {
 		let (reason, _) = kio::clock::timeout(TIMEOUT, lease.closed())
 			.await
 			.expect("the lease never ended");
-		assert_eq!(reason, client_closed(), "{timeout:?}: ended after the decision {wait:?} in");
+		assert_eq!(
+			reason,
+			client_closed(),
+			"{timeout:?}: ended after the decision {wait:?} in"
+		);
 
 		trigger.start();
 		kio::clock::timeout(TIMEOUT, running)
diff --git a/rs/moq-relay/tests/integration/session_revalidate.rs b/rs/moq-relay/tests/integration/session_revalidate.rs
index e6e70d89d00..8595738694c 100644
--- a/rs/moq-relay/tests/integration/session_revalidate.rs
+++ b/rs/moq-relay/tests/integration/session_revalidate.rs
@@ -87,8 +87,10 @@ impl Script {
 
 	/// The `end` of session `id`, once it arrives.
 	async fn end_of(&self, id: &str) -> moq_auth::lease::Reason {
-		self.until(&format!("{id} ended"), |script| script.ends().iter().any(|r| r.id == id))
-			.await;
+		self.until(&format!("{id} ended"), |script| {
+			script.ends().iter().any(|r| r.id == id)
+		})
+		.await;
 		let end = self.ends().into_iter().find(|r| r.id == id).unwrap();
 		match end.event {
 			Event::End { reason, .. } => reason,

From bd2f12fa168425a584ad9ba2cda7921a9175decd Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Fri, 6 Nov 2026 15:13:53 -0800
Subject: [PATCH 7/7] quest: record the dial-hook decision for the relay's mesh
 tests

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 quest/m1/mock-clock-tests/rs-relay.md | 25 +++++++++++++++++--------
 1 file changed, 17 insertions(+), 8 deletions(-)

diff --git a/quest/m1/mock-clock-tests/rs-relay.md b/quest/m1/mock-clock-tests/rs-relay.md
index ded51d29ecf..1b0cecc3d06 100644
--- a/quest/m1/mock-clock-tests/rs-relay.md
+++ b/quest/m1/mock-clock-tests/rs-relay.md
@@ -34,13 +34,22 @@ on the real clock: the `runtime_uring.rs` and listing poll loops, `web.rs`'s
 `start.elapsed() >= timeout`, `bitrate_caps.rs`'s `LEAST`, and
 `shutdown_signal.rs`'s drain-window lower bounds.
 
-Public API: none, unless the maintainer picks the dial hook. Wire: none.
+Decided 2026-11-06 (maintainer, on
+[moq-dev/moq#8087](https://github.com/moq-dev/moq/pull/8087)):
 
-## Waiting
+- ✅ **A dial hook on `cluster::Options` that an in-memory link can fill**, so
+  the four tests above run on `kio::clock::sim` and prove convergence with
+  `sim::idle` instead of a quiet window. Not chosen: keeping them as documented
+  real-clock exceptions opted out of the [Rust guard](/quest/m0/clock-guard/rust.md).
 
-- Maintainer: how mesh tests prove convergence. (1) Recommended: give
-  `cluster::Options` a dial hook that an in-memory link can fill, so these tests
-  run on `kio::clock::sim` and end on `sim::idle`; this adds public API whose only
-  consumer is tests. (2) Keep these windows as documented real-clock exceptions
-  and opt them out of the [Rust guard](/quest/m0/clock-guard/rust.md). Check: a
-  recorded pick in this file's Plan.
+Guidance:
+
+- The hook is public API whose only consumer today is these tests; name it by
+  role (how a peer is reached), not by the test transport.
+- The mesh tests build whole relays with `Relay::load` over loopback, so the
+  simulator needs a relay that accepts in-memory links too, as
+  `tests/integration/memory.rs` already does for one relay.
+- `shutdown_signal`'s window waits on a process signal, not a mesh: its relay
+  must also take that signal from a source the simulator can drive.
+
+Public API: a new `cluster::Options` dial hook. Wire: none.