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..1b0cecc3d06 100644
--- a/quest/m1/mock-clock-tests/rs-relay.md
+++ b/quest/m1/mock-clock-tests/rs-relay.md
@@ -1,21 +1,55 @@
-# [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.
+
+Decided 2026-11-06 (maintainer, on
+[moq-dev/moq#8087](https://github.com/moq-dev/moq/pull/8087)):
+
+- ✅ **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).
+
+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.
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..b729f37ed18 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,23 @@ 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"
+		// 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"
 		);
-		drop(connection);
 
 		trigger.start();
 		kio::clock::timeout(TIMEOUT, running)
@@ -2034,9 +2065,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/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);
 }
diff --git a/rs/moq-relay/tests/integration/session_revalidate.rs b/rs/moq-relay/tests/integration/session_revalidate.rs
index cb58e328e4d..8595738694c 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,34 @@ 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 +145,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 +172,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 +201,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 +221,7 @@ impl Fixture {
 			url,
 			internal: format!("http://{internal_addr}").parse().unwrap(),
 			sessions,
+			stats,
 			_relay: handle,
 		}
 	}
@@ -306,14 +338,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 +351,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 +380,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 +394,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 +435,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 +448,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 +456,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"
 	);
 }
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;