test: the last host-clock waits run on a mocked clock or await their event

cpp/obs/test/moq-dock-stop-test.cpp

@@ -6,9 +6,9 @@
 
 #include "moq-dock-stop.h"
 
-#include <atomic>
 #include <chrono>
 #include <cstdio>
+#include <mutex>
 #include <thread>
 
 #define CHECK(cond)                                                                         \
@@ -36,20 +36,25 @@ int main()
 	{
 		MoQDockStopBridge bridge;
 		CHECK(bridge.begin());
-		std::atomic<bool> ended{false};
+		// Ends only once the close began, so the close returning with nothing active
+		// means it waited for this end. The worker's deadline only bounds a lost close
+		// notify, after which it ends anyway, so it can never fail the test. The close's
+		// deadline is one the test never reaches, so only the end releases it.
 		std::thread worker([&] {
-			std::this_thread::sleep_for(std::chrono::milliseconds(30));
+			{
+				std::unique_lock<std::mutex> lock(bridge.mutex);
+				bridge.cv.wait_for(lock, std::chrono::seconds(10), [&] { return bridge.closing; });
+			}
 			bridge.end();
-			ended.store(true);
 		});
-		const auto start = std::chrono::steady_clock::now();
-		bridge.closeAndWait(std::chrono::milliseconds(500));
-		const auto elapsed = std::chrono::steady_clock::now() - start;
+		bridge.closeAndWait(std::chrono::hours(1));
+		bool idle;
+		{
+			std::lock_guard<std::mutex> lock(bridge.mutex);
+			idle = bridge.active == 0;
+		}
 		worker.join();
-		CHECK(ended.load());
-		CHECK(bridge.active == 0);
-		CHECK(elapsed >= std::chrono::milliseconds(20));
-		CHECK(elapsed < std::chrono::milliseconds(400));
+		CHECK(idle);
 		CHECK(!bridge.begin());
 	}
 	printf("close waits for in-flight end: ok\n");

cpp/obs/test/moq-output-test.cpp

@@ -427,6 +427,9 @@ struct MoQOutputStatsTest {
 		stale.bytes_sent = 123;
 		return output.CommitConnectionStats(session, std::move(stale), accepted);
 	}
+	// Runs the continuations already queued, the last that can report once the attempt
+	// is retired.
+	static void Drain(MoQOutput &output) { ::Drain(output.worker); }
 	static void Disconnect(MoQOutput &output)
 	{
 		std::lock_guard<std::mutex> lock(output.mutex);
@@ -956,7 +959,8 @@ int main()
 		o.CopyLastFailure(&code, &reason);
 		CHECK(!reason.empty());
 		CHECK(!o.IsLiveSession());
-		std::this_thread::sleep_for(std::chrono::milliseconds(50));
+		// The failure retired the attempt, so nothing it queued can report again.
+		MoQOutputStatsTest::Drain(o);
 		CHECK(signalCount() == 1);
 	}
 	printf("fatal before connect: ok\n");
@@ -1036,7 +1040,8 @@ int main()
 		MoQOutput o(nullptr, OUTPUT);
 		CHECK(o.Start());
 		o.Stop();
-		std::this_thread::sleep_for(std::chrono::milliseconds(100));
+		// The stop dropped the connect, so only a result it already queued could still run.
+		MoQOutputStatsTest::Drain(o);
 		CHECK(signalCount() == 1);
 		CHECK(signalAt(0).code == OBS_OUTPUT_SUCCESS);
 	}
@@ -1085,7 +1090,7 @@ int main()
 		CHECK(WaitFor([] { return signalCount() == 2; }));
 		CHECK(signalAt(0).begin_capture_at == 1);
 		CHECK(signalAt(1).begin_capture_at == 2);
-		std::this_thread::sleep_for(std::chrono::milliseconds(50));
+		MoQOutputStatsTest::Drain(o);
 		CHECK(signalCount() == 2);
 	}
 	printf("OBS-driven restart: ok\n");
@@ -1128,7 +1133,8 @@ int main()
 			const size_t starts = round % 2 ? 2 : 1;
 			CHECK(signalCount() <= starts + 1);
 		}
-		std::this_thread::sleep_for(std::chrono::milliseconds(50));
+		// A late report could only come from a continuation, and the destructor joined the
+		// worker that runs them.
 		CHECK(g_signals_after_destroy == 0);
 	}
 	printf("failures outliving the output: ok\n");

cpp/obs/test/moq-source-test.cpp

@@ -1803,10 +1803,15 @@ int main()
 		const int blanks = g_blank_calls;
 		g_info.update(source, settingsData());
 		CHECK(g_blank_calls == blanks + 1);
+		auto *ctx = static_cast<moq_source *>(source);
+		{
+			// No connection is left to subscribe with, so later frames have nowhere to land.
+			std::lock_guard<std::mutex> lock(ctx->mutex);
+			CHECK(!ctx->connection);
+		}
+		// A frame the dropped connection already queued finds it gone and outputs nothing.
 		const int frames = g_output_frames;
-		for (int i = 0; i < 10; i++)
-			publisher.Write();
-		std::this_thread::sleep_for(std::chrono::milliseconds(50));
+		Drain(ctx->worker);
 		CHECK(g_output_frames == frames);
 		destroySource(source);
 	}
@@ -1903,11 +1908,25 @@ int main()
 		TestRelay relay;
 		reset(relay.Url());
 		void *source = createSource();
+		auto *ctx = static_cast<moq_source *>(source);
+		std::promise<void> offline;
 		{
 			Publisher publisher(relay, "obs/test");
 			CHECK(publisher.WriteUntil([] { return g_output_frames > 0; }));
+			// The source waits for the next instance, and re-arms that wait once it sees
+			// this one go offline, replacing the call it holds now. Observe the replacement,
+			// keeping the real cancellation.
+			std::lock_guard<std::mutex> lock(ctx->mutex);
+			auto &call = ctx->connection->available_call;
+			CHECK(call.has_value());
+			auto pending = std::make_shared<moq::Continuation>(std::move(*call));
+			call.emplace([pending, &offline] {
+				pending->cancel();
+				offline.set_value();
+			});
 		}
-		std::this_thread::sleep_for(std::chrono::milliseconds(50));
+		CHECK(offline.get_future().wait_for(std::chrono::seconds(10)) == std::future_status::ready);
+		// On a lost re-arm, the destroy runs the replacement, so `offline` outlives it.
 		destroySource(source);
 	}
 	report("broadcast ends");

cpp/obs/test/moq-test-relay.h

@@ -7,6 +7,8 @@
 
 #include <moq/moq.hpp>
 
+#include "moq-worker.h"
+
 #include <arpa/inet.h>
 #include <netinet/in.h>
 #include <sys/socket.h>
@@ -17,6 +19,7 @@
 #include <cstdio>
 #include <cstdlib>
 #include <functional>
+#include <future>
 #include <memory>
 #include <mutex>
 #include <string>
@@ -54,6 +57,22 @@ inline bool WaitFor(const std::function<bool()> &done, std::chrono::milliseconds
 	return true;
 }
 
+// Runs whatever `worker` already queued, so a test checks what an object's dropped calls
+// did without waiting out a window. Dropping a moq::Continuation does not recall a result
+// whose delivery already began, which can still queue after this returns: a check after
+// it can miss a stale continuation, never fail falsely.
+inline void Drain(const MoQWorker &worker)
+{
+	auto drained = std::make_shared<std::promise<void>>();
+	// A stopped worker runs nothing more, which is just as settled.
+	if (!worker.Executor()([drained] { drained->set_value(); }))
+		return;
+	if (drained->get_future().wait_for(std::chrono::seconds(10)) != std::future_status::ready) {
+		std::fprintf(stderr, "FAIL: the worker never reached the drain\n");
+		std::abort();
+	}
+}
+
 // A moq-ffi server that publishes every broadcast it consumes, so whatever one
 // session announces, another can subscribe to.
 //

cpp/obs/test/obs-moq-test.cpp

@@ -52,7 +52,8 @@ int main()
 	auto server = TestOk(moq::Server::init(config), "server init");
 	TestOk(server->listen().get(), "listen");
 	auto pending = server->accept();
-	if (pending.wait_for(std::chrono::milliseconds(50)) != std::future_status::timeout) {
+	// Nothing dials, so only the unload below can end it; read without waiting.
+	if (pending.wait_for(std::chrono::seconds(0)) != std::future_status::timeout) {
 		fprintf(stderr, "FAIL: accept resolved with no peer\n");
 		return 1;
 	}

quest/m1/README.md

@@ -41,7 +41,6 @@ moves what remains of them once their claims merge.
 - [Audio jitter target](/quest/m1/audio-jitter-target/README.md) - the playout target is default in both languages; time-stretched browser playback without steady-state underruns and a native trace replay remain
 - [Buffered playback subscribes in ascending order](/quest/m1/buffered-ascending.md) - a buffered `<moq-watch>` asks for the oldest group first, so a voice reply's opening arrives first, and audio's head hold shrinks to the floor
 - [Push-to-talk's press-to-audible latency fits a budget on the defaults](/quest/m1/push-to-talk-latency.md) - press to audible on the browser defaults, excluding device output, stays under a kixtop-measured budget that `just test push-to-talk` enforces
-- [Tests run on a mocked clock](/quest/m1/mock-clock-tests/README.md) - about 300 tests stop proving nothing happened by waiting, sleeping to settle, or asserting on elapsed time
 - [Lifecycles are explicit state](/quest/m1/explicit-state/README.md) - every lifecycle the 2026-11-07 audit found implicit is one enum or union, fixing the bugs in the 11 lifecycles it redesigns, with review and agent rules against new ones
 - [Track drop runs are forgotten below the floor](/quest/m1/drop-runs-prune.md) - a track's dropped-group record stays proportional to its window in Rust and JS
 - [A macOS camera delivers frames the renderer imports](/quest/m1/camera-nv12-macos.md) - AVFoundation capture asks for NV12, so a camera whose native format is `2vuy` reaches the renderer

rs/moq-ffi/src/test.rs

@@ -7517,7 +7517,6 @@ async fn cancelled_state_wait_does_not_swallow_the_next_transition() {
 #[cfg(feature = "video")]
 #[tokio::test]
 async fn video_encoder_follows_a_shrinking_grant() {
-	kio::clock::host_time();
 	use crate::bandwidth::MoqAllocator;
 	use crate::video::*;
 
@@ -7568,18 +7567,8 @@ async fn video_encoder_follows_a_shrinking_grant() {
 		.unwrap();
 	assert_eq!(reservation.rate(), Some(1_000_000));
 
-	let deadline = kio::clock::Instant::now() + TIMEOUT;
-	loop {
-		if video.applied_bitrate() == 1_000_000 {
-			break;
-		}
-		assert!(
-			kio::clock::Instant::now() < deadline,
-			"encoder did not follow the shrinking grant, last applied {}",
-			video.applied_bitrate()
-		);
-		kio::clock::sleep(Duration::from_millis(10)).await;
-	}
+	// A drop applies at once, so this waits on the follow thread alone, never a timer.
+	video.wait_applied(1_000_000);
 
 	video.finish().unwrap();
 	broadcast.close().unwrap();

rs/moq-ffi/src/video.rs

@@ -360,6 +360,10 @@ struct VideoProducer {
 	/// an allocator. Beside the encoder, so a move to a new track, a finish, and an abort
 	/// change all three together.
 	follow: Option<Following>,
+	/// Notified under this rendition's lock each time the follower retunes, so a test
+	/// awaits the follower instead of polling `target`.
+	#[cfg(test)]
+	retuned: Arc<std::sync::Condvar>,
 }
 
 /// A rendition's bandwidth claim on its current track, and the thread retuning the encoder
@@ -610,13 +614,29 @@ impl MoqVideoProducer {
 	/// The bitrate the encoder produces at, or 0 while disabled.
 	#[cfg(test)]
 	pub(crate) fn applied_bitrate(&self) -> u64 {
-		match self.inner.lock().unwrap().as_ref().map(|producer| producer.target) {
-			Some(moq_mux::rate::Target::Enabled(bitrate)) => bitrate.as_bps(),
-			_ => 0,
+		applied(self.inner.lock().unwrap().as_ref())
+	}
+
+	/// Block until the follower has retuned the encoder to `bitrate`.
+	#[cfg(test)]
+	pub(crate) fn wait_applied(&self, bitrate: u64) {
+		let mut guard = self.inner.lock().unwrap();
+		while applied(guard.as_ref()) != bitrate {
+			let retuned = guard.as_ref().expect("the rendition is open").retuned.clone();
+			guard = retuned.wait(guard).unwrap();
 		}
 	}
 }
 
+/// The bitrate `producer` encodes at, or 0 while disabled or gone.
+#[cfg(test)]
+fn applied(producer: Option<&VideoProducer>) -> u64 {
+	match producer.map(|producer| producer.target) {
+		Some(moq_mux::rate::Target::Enabled(bitrate)) => bitrate.as_bps(),
+		_ => 0,
+	}
+}
+
 // Dropped or closed before its finish, the track ends the way a publisher leaving does.
 impl crate::producer::Child for MoqVideoProducer {
 	fn abort(&self, err: moq_net::Error) {
@@ -855,6 +875,8 @@ impl MoqVideoProducer {
 				priority,
 				target: moq_mux::rate::Target::Enabled(ceiling),
 				follow: None,
+				#[cfg(test)]
+				retuned: Arc::default(),
 			}))),
 			children: children.clone(),
 		};
@@ -1000,6 +1022,8 @@ async fn follow_reservation(followed: Followed) {
 		let policy = video_policy(*producer.ceiling.read(), below);
 		let before = producer.producer.demand();
 		let retuned = retune(&mut follower, policy, target, |target| producer.apply(target));
+		#[cfg(test)]
+		producer.retuned.notify_all();
 		// A disable drains the encoder into the track, and output with a new decoder identity
 		// moves the rendition, so the claim moves with it. The replaced follower is this
 		// thread, which ends here, so it is detached rather than joined.
@@ -1662,6 +1686,7 @@ mod finish_tests {
 			ceiling: kio::Producer::new(moq_net::bandwidth::Rate::from_mbps(1)),
 			priority: hang::catalog::PRIORITY.video,
 			follow: None,
+			retuned: Arc::default(),
 		};
 		let (video, disabled) = tokio::task::spawn_blocking(move || {
 			video