test: halve the slowest tests' CPU

Cargo.toml

@@ -305,6 +305,12 @@ opt-level = 3
 [profile.dev.package.sonora-simd]
 opt-level = 3
 
+# Same for the rubato resampler, whose SIMD intrinsics are all of
+# `resample::tests::duration_and_samples_are_invariant_across_partitions_and_epochs`:
+# 0.9 CPU-seconds unoptimized, 0.14 optimized.
+[profile.dev.package.rubato]
+opt-level = 3
+
 # The Go mirror stores native staticlibs in Git, whose per-file limit is 100 MB.
 # Keep all release entry points on the same compact native profile.
 [profile.release]

quest/m0/ci/README.md

@@ -160,7 +160,6 @@ The historical samples locate candidates; they are not controlled benchmarks.
 - [Verify the R2 Rust cache](/quest/m0/ci/mbx-r2-verify.md) - confirm the hosted Nightly jobs reuse the previous Nightly, then measure restore time and expiry
 - [Verify the Nix cache](/quest/m0/ci/nix-cache-verify.md) - confirm releases use `nix.moq.dev`, check its bucket sizes, and retire Cachix
 - [Unify dependency features](/quest/m0/ci/hakari.md) - a cargo-hakari workspace-hack so every job compiles each dependency alike and mbx hits across jobs
-- [Halve the slowest tests' CPU](/quest/m0/ci/slow-tests.md) - the slowest 1% of tests take under half their 289 CPU-seconds, with coverage kept
 - [Release credentials off main](/quest/m0/ci/publish-release-only.md) - `publish` admits only `release` and release tags once its `workflow_run` publishers run on their tag
 - [The queue host keeps the queue's disk](/quest/m0/ci/queue-host-disk.md) - agent sessions can't starve the queue runner of disk, and a crashed runner restarts
 - [The next test-CPU cuts](/quest/m0/ci/test-cost-followups.md) - the slow-test leftovers from #6987, cut at their source

quest/m0/ci/five-minutes.md

@@ -67,8 +67,8 @@ What remains for a moq-net edit, largest first:
   lever was measured at about 30 ms of CPU per build and dropped), or running the rs lint beside the
   test build in a second target directory, which trades CPU for wall time
   against the epic's CPU measure. Decided 2026-10-30 (user): defer that
-  parallel lint; measure it after the slow-tests and hakari quests land.
-- The test run is the [slow tests](/quest/m0/ci/slow-tests.md) quest's.
+  parallel lint; measure it after the hakari quest lands (slow-tests landed in
+  [moq-dev/moq#6987](https://github.com/moq-dev/moq/pull/6987)).
 - The lint's remaining 230 s: the dev loom compile and `cargo doc`. Decided
   2026-10-31 (maintainer, see the [epic](/quest/m0/ci/README.md)): the gate
   drops the loom compile, the Send-depth check (now its own Nightly job), the
@@ -80,14 +80,13 @@ What remains for a moq-net edit, largest first:
 - `just test` has no root recipe: bare `just test` lists the cross-language
   suites. Name which command the `just test` baseline above counts before
   measuring it.
-- Prefer the epic's levers, such as [slow tests](/quest/m0/ci/slow-tests.md),
-  [hakari](/quest/m0/ci/hakari.md), and cache hits, and tighter diff selection,
+- Prefer the epic's levers, such as [hakari](/quest/m0/ci/hakari.md) and cache
+  hits, and tighter diff selection,
   over moving coverage. Move a check to the queue or nightly only under the
   epic's rule: evidence and an explicit coverage tradeoff for each.
 - Split any large single lever into its own quest.
 
 ## Related
 
-- [Halve the slowest tests' CPU](/quest/m0/ci/slow-tests.md) - the largest test-time lever
 - [Unify dependency features](/quest/m0/ci/hakari.md) - cache hits across feature shapes
 - [Land main through quest land](/quest/m0/ci/land.md) - the queue gate that keeps checks moved out of the local gate

rs/moq-audio/src/decode/consumer.rs

@@ -2056,11 +2056,12 @@ mod tests {
 	///
 	/// PCM rather than Opus only for speed: the frame term comes from the decoded
 	/// sample count either way, and an unoptimized Opus codec takes minutes over
-	/// the whole corpus. [`opus_states_its_own_frame_duration`] covers Opus.
+	/// the whole corpus. [`opus_states_its_own_frame_duration`] covers Opus. The
+	/// lowest common rate keeps each packet small, since only its duration counts.
 	async fn replay(case: &crate::jitter::tests::Case) -> Vec<f64> {
 		use crate::jitter::tests::duration;
 
-		let rate = 48_000;
+		let rate = 8_000;
 		let samples = (case.frame * rate as f64 / 1000.0) as usize;
 		let mut catalog = hang::catalog::AudioConfig::new(hang::catalog::AudioCodec::Pcm, rate, 1);
 		catalog.jitter = case.advertised.map(duration);

rs/moq-cli/src/publish.rs

@@ -1739,10 +1739,11 @@ pub(crate) mod tests {
 				.into_iter()
 				.map(|(sequence, frames)| (sequence, frames.expect("the recording reads every group whole")))
 				.collect();
-			let output: Vec<u8> = recording
+			let output = recording
 				.iter()
-				.flat_map(|(_, frames)| frames.iter().flat_map(|frame| frame.payload.iter().copied()))
-				.collect();
+				.flat_map(|(_, frames)| frames.iter().map(|frame| &frame.payload[..]))
+				.collect::<Vec<_>>()
+				.concat();
 			assert!(output.len() > KYRION.len(), "{version}: both copies crossed");
 			assert!(
 				input.ends_with(&output),

rs/moq-hls/src/import.rs

@@ -186,7 +186,9 @@ fn same_origin(entry: &Url, url: &Url) -> bool {
 /// Fetches playlists and segments over HTTP or from the filesystem, applying HLS byte
 /// ranges (and validating that the server honored them).
 struct Fetcher {
-	client: Client,
+	/// `None` for a `file` playlist: every fetch stays on its scheme, so it never needs one, nor
+	/// the system trust roots building one loads.
+	client: Option<Client>,
 	/// The caller's playlist URL. A playlist is untrusted, so every fetch stays on its origin.
 	entry: Url,
 }
@@ -198,6 +200,10 @@ impl Fetcher {
 			return Err(Error::ResourceRefused);
 		}
 
+		if entry.scheme() == "file" {
+			return Ok(Self { client: None, entry });
+		}
+
 		// An intermediate hop is already a request to its origin, so check each one before
 		// following it. A stopped hop reaches `fetch` as a redirect response, refused there.
 		let origin = entry.clone();
@@ -222,7 +228,10 @@ impl Fetcher {
 			.timeout(REQUEST_TIMEOUT)
 			.redirect(redirect)
 			.build()?;
-		Ok(Self { client, entry })
+		Ok(Self {
+			client: Some(client),
+			entry,
+		})
 	}
 
 	/// Fetch and parse a media playlist, with the URL its relative URIs resolve against.
@@ -251,7 +260,8 @@ impl Fetcher {
 		if !same_origin(&self.entry, url) {
 			return Err(Error::ResourceRefused);
 		}
-		if url.scheme() == "file" {
+		// Same origin, so a file URL exactly when the entry is one.
+		let Some(client) = &self.client else {
 			let path = url.to_file_path().map_err(|_| Error::InvalidFileUrl)?;
 			let Some(range) = resource.range else {
 				let bytes = tokio::fs::read(&path).await.map(Bytes::from).map_err(Error::from)?;
@@ -265,50 +275,49 @@ impl Fetcher {
 				.read_to_end(&mut bytes)
 				.await
 				.map_err(Error::from)?;
-			Ok((
+			return Ok((
 				Self::validate_range_length(url, range, Bytes::from(bytes))?,
 				url.clone(),
-			))
-		} else {
-			let response = self.request(resource).send().await?;
-			if response.status().is_redirection() {
-				return Err(Error::ResourceRefused);
-			}
-			let mut response = response.error_for_status()?;
-			let served = response.url().clone();
-			let Some(range) = resource.range else {
-				return Ok((response.bytes().await?, served));
-			};
+			));
+		};
+		let response = Self::request(client, resource).send().await?;
+		if response.status().is_redirection() {
+			return Err(Error::ResourceRefused);
+		}
+		let mut response = response.error_for_status()?;
+		let served = response.url().clone();
+		let Some(range) = resource.range else {
+			return Ok((response.bytes().await?, served));
+		};
 
-			// An origin that ignores `Range` sends the whole resource, which would cost every
-			// earlier byte of the file for each segment.
-			if response.status() != reqwest::StatusCode::PARTIAL_CONTENT {
-				return Err(Error::ByteRangeIgnored {
+		// An origin that ignores `Range` sends the whole resource, which would cost every
+		// earlier byte of the file for each segment.
+		if response.status() != reqwest::StatusCode::PARTIAL_CONTENT {
+			return Err(Error::ByteRangeIgnored {
+				url: redact(url),
+				status: response.status().as_u16(),
+			});
+		}
+		Self::validate_content_range(url, range, response.headers())?;
+
+		// Stream the body so an overrun fails without buffering the excess.
+		let mut body = BytesMut::new();
+		while let Some(chunk) = response.chunk().await? {
+			let read = body.len() + chunk.len();
+			if !u64::try_from(read).is_ok_and(|read| read <= range.length) {
+				return Err(Error::ByteRangeLengthMismatch {
 					url: redact(url),
-					status: response.status().as_u16(),
+					expected: range.length,
+					actual: read,
 				});
 			}
-			Self::validate_content_range(url, range, response.headers())?;
-
-			// Stream the body so an overrun fails without buffering the excess.
-			let mut body = BytesMut::new();
-			while let Some(chunk) = response.chunk().await? {
-				let read = body.len() + chunk.len();
-				if !u64::try_from(read).is_ok_and(|read| read <= range.length) {
-					return Err(Error::ByteRangeLengthMismatch {
-						url: redact(url),
-						expected: range.length,
-						actual: read,
-					});
-				}
-				body.extend_from_slice(&chunk);
-			}
-			Ok((Self::validate_range_length(url, range, body.freeze())?, served))
+			body.extend_from_slice(&chunk);
 		}
+		Ok((Self::validate_range_length(url, range, body.freeze())?, served))
 	}
 
-	fn request(&self, resource: &Resource) -> reqwest::RequestBuilder {
-		let mut request = self.client.get(resource.url.clone());
+	fn request(client: &Client, resource: &Resource) -> reqwest::RequestBuilder {
+		let mut request = client.get(resource.url.clone());
 		if let Some(range) = resource.range {
 			let end = range.start + range.length - 1;
 			request = request.header(reqwest::header::RANGE, format!("bytes={}-{}", range.start, end));
@@ -701,7 +710,8 @@ pub struct Import {
 impl Import {
 	/// Create a new HLS import that will write into the given broadcast.
 	///
-	/// Installs rind's rustls provider as the process default unless one already is.
+	/// For an HTTP playlist, installs rind's rustls provider as the process default unless one
+	/// already is. A file playlist builds no HTTP client, so it installs none.
 	pub fn new(broadcast: moq_net::broadcast::Producer, catalog: CatalogProducer, cfg: Config) -> Result<Self> {
 		let fetcher = Fetcher::new(&cfg)?;
 		Ok(Self {

rs/moq-json/src/snapshot/encoder.rs

@@ -584,6 +584,7 @@ mod test {
 		let mut runner = TestRunner::new(PropertyConfig {
 			cases,
 			rng_seed: RngSeed::Fixed(u64::from(seed)),
+			rng_algorithm: proptest::test_runner::RngAlgorithm::XorShift,
 			failure_persistence: None,
 			..Default::default()
 		});

rs/moq-json/src/stream/encoder.rs

@@ -48,13 +48,20 @@ impl<T> Pending<'_, T> {
 ///
 /// Payloads must reach the wire in order. A group has no size cap, so the encoder keeps no budget.
 pub struct Encoder<T> {
+	/// The most bytes one record may hold. A field rather than the constant so a test can shrink it
+	/// to a size it can reach.
+	pub(super) max_frame_bytes: u64,
+
 	_marker: PhantomData<fn(T)>,
 }
 
 impl<T> Encoder<T> {
 	/// Create an encoder.
 	pub fn new(_config: Config) -> Self {
-		Self { _marker: PhantomData }
+		Self {
+			max_frame_bytes: moq_net::group::MAX_CACHE_BYTES,
+			_marker: PhantomData,
+		}
 	}
 }
 
@@ -68,7 +75,7 @@ impl<T: Serialize> Encoder<T> {
 	/// ([`moq_net::group::MAX_CACHE_BYTES`]). The refused record leaves the encoder untouched.
 	pub fn encode(&mut self, value: &T) -> Result<Pending<'_, T>> {
 		let payload = Bytes::from(serde_json::to_vec(value)?);
-		if payload.len() as u64 > moq_net::group::MAX_CACHE_BYTES {
+		if payload.len() as u64 > self.max_frame_bytes {
 			return Err(moq_net::Error::FrameTooLarge.into());
 		}
 

rs/moq-json/src/stream/mod.rs

@@ -380,15 +380,19 @@ mod test {
 		matches!(result, Err(crate::Error::Net(moq_net::Error::FrameTooLarge)))
 	}
 
+	/// A frame limit small enough to pass cheaply, standing in for moq-net's 32 MiB cache.
+	const BUDGET: u64 = 32 * 1024;
+
 	/// A record larger than a frame is refused before anything is written, so the log carries on:
 	/// the next record lands and decodes.
 	#[test]
 	fn an_oversized_record_is_refused_and_the_log_continues() {
 		for compression in [false, true] {
 			let (mut producer, track) = producer_on(Config::default(), info(compression));
+			producer.set_max_frame_bytes(BUDGET);
 			producer.append(&json!({ "n": 0 })).unwrap();
 
-			let oversized = Value::String("x".repeat(moq_net::group::MAX_CACHE_BYTES as usize));
+			let oversized = Value::String("x".repeat(BUDGET as usize));
 			assert!(refused(producer.append(&oversized)), "compression={compression}");
 
 			producer.append(&json!({ "n": 1 })).unwrap();

rs/moq-json/src/stream/producer.rs

@@ -101,6 +101,14 @@ impl<T: Serialize> Producer<T> {
 	}
 }
 
+#[cfg(test)]
+impl<T> Producer<T> {
+	/// Shrink the frame limit to a size a test can reach.
+	pub(super) fn set_max_frame_bytes(&self, bytes: u64) {
+		self.inner.lock().unwrap().encoder.max_frame_bytes = bytes;
+	}
+}
+
 /// Shared publishing state behind [`Producer`]'s `Arc<Mutex>`.
 ///
 /// The track and the encoder are separate fields so a [`Pending`](super::Pending) record (which