diff --git a/quest/m1/README.md b/quest/m1/README.md index a874a1f96c..d87527d3d5 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -135,7 +135,6 @@ above them. - [A/V clock](/quest/m1/av-clock.md) - the audio playhead drives Sync.reference while audio plays, through per-track sync handles - [Plan: watch worker](/quest/m1/plan-watch-worker.md) - prototype an invisible page worker against app-spawned workers, and land the jank harness that decides - [Watch worker](/quest/m1/watch-worker.md) - watch playback runs in a worker onto an OffscreenCanvas, so main-thread jank never stalls video or audio -- [Rust papercuts](/quest/m1/papercuts-rs.md) - HTTP refusals are counted, and a `u64::MAX` resume is unbounded - [Duration parser defaults](/quest/m1/cli-duration-defaults.md) - a moq-tokio duration set in code beats the parser default on a parsed or merged config, zero included, after #5350 - [Publish channel count](/quest/m1/publish-audio-channel-count.md) - forcing a channel count on an Audio.Capture stops costing the subscriber gaps of silence, and the media fixture's cold-start audio-tone flake is fixed at its cause - [JS request deadline](/quest/m1/js-request-deadline.md) - each JS request, PUBLISH_NAMESPACE included, has one fatal 10 s timer from create to answer and fails fast without stream credit, so nothing queues or retries diff --git a/quest/m1/papercuts-rs.md b/quest/m1/papercuts-rs.md deleted file mode 100644 index 38ac512ae7..0000000000 --- a/quest/m1/papercuts-rs.md +++ /dev/null @@ -1,31 +0,0 @@ -# [S] Rust papercuts from the 2026-10-07 audit - -## Goal - -Two small Rust fixes from the 2026-10-07 code audit, each with a regression -test that fails without it: - -- HTTP `/announced` and `/fetch` refusals count in the relay's refusal - metrics, as QUIC, WebSocket, and io_uring refusals already do. -- A replaced route whose newest group is `u64::MAX` keeps delivering that - group from the old copy instead of being capped at group 0. - -## Plan - -- `rs/moq-relay/src/web.rs`: both `admit_http` call sites (`serve_announced` - and `serve_fetch`) return the error with `?` and never call - `cluster.refusals.record`. The missing-subscriber 401 in each handler is - uncounted too. Record each refusal the way `websocket.rs` does. -- `Copy::wire` in `rs/moq-net/src/model/resume.rs`: `Position::after_group` - returns `None` for the last group, meaning "no end", and - `unwrap_or_default()` turns that into position 0. `until` is - `Option>`, and an inner `None` (replaced before any group) must - still end at 0; only `after_group`'s overflow is unbounded. Reuse `let_go`'s - `until.map_or(Some(0), |last| last.checked_add(1))` and test both cases. - lite-07 varints can carry `u64::MAX`, so this is reachable from the wire. - -Decided 2026-10-08: the `GOING_AWAY` fallback for future drafts and the -zero `UDP_GRO` stride were dropped. Neither is reachable today: no newer -draft exists, and Linux never emits a zero stride. - -Public API: none. Wire: none. diff --git a/quest/m2/http-fetch-stats.md b/quest/m2/http-fetch-stats.md index aa4b2865ee..d9edf49aca 100644 --- a/quest/m2/http-fetch-stats.md +++ b/quest/m2/http-fetch-stats.md @@ -42,8 +42,6 @@ older reader ignores and a newer one defaults to zero. ## Related -- [Rust papercuts](/quest/m1/papercuts-rs.md) - counts `/fetch` refusals in - `/metrics`; this counts what an admitted fetch serves - [Bitrate claim](/quest/m2/rate-claim.md) - HTTP refuses capped tokens, so stats are HTTP's only byte account - [Session outcomes](/quest/m1/session-outcomes.md) - the session counters diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 85e43c55d0..ec0e70f63e 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -267,13 +267,19 @@ impl Copy { start: max_some(reader.start, self.floor), ..reader.clone() }; - if let Some(until) = self.until { - let end = until.and_then(Position::after_group).unwrap_or_default(); + if let Some(end) = self.end().map(Position::group) { wire.end = Some(wire.end.map_or(end, |requested| requested.min(end))); } wire } + /// The first group past where this copy was replaced, if any: one replaced before + /// any group ends at 0, and one replaced at `u64::MAX` has no end. + fn end(&self) -> Option { + self.until + .and_then(|last| last.map_or(Some(0), |last| last.checked_add(1))) + } + fn update(&mut self, reader: &Subscription) { let wire = self.wire(reader); let _ = match &mut self.sub { @@ -513,7 +519,7 @@ impl Reader { if copy.done.is_some() { return false; } - let until = copy.until; + let end = copy.end(); let Poll::Ready(Ok(sub)) = copy.poll_ready(groups, &kio::Waiter::noop()) else { return false; }; @@ -522,7 +528,7 @@ impl Reader { // cap here, since unread groups become eligible again when it rises. sub.has_unread_group(&track::Unread { start: groups.0, - end: until.and_then(|until| until.map_or(Some(0), |last| last.checked_add(1))), + end, delivered, }) }); @@ -1546,6 +1552,27 @@ mod test { assert_eq!(recv(&mut sub).sequence, 6); } + #[test] + fn a_replaced_copy_at_the_last_sequence_has_no_overflow_cap() { + for (last, end) in [(Some(u64::MAX), None), (None, Some(Position::default()))] { + let routes = Producer::new(); + let a = copy(); + routes.serve(a.consume()); + let subscription = kio::Producer::new(Subscription::default()); + let sub = routes.consume().subscribe(subscription); + let mut reader = sub.reader.lock().unwrap(); + let old = &mut reader.copies[0]; + old.until = Some(last); + assert_eq!(old.wire(&Subscription::default()).end, end); + // A reader's narrower bound survives an unbounded replacement cap. + let requested = Subscription { + end: Some(Position::group(7)), + ..Default::default() + }; + assert_eq!(old.wire(&requested).end, Some(end.unwrap_or(Position::group(7)))); + } + } + #[test] #[cfg(feature = "async")] fn a_replaced_route_finishes_only_the_groups_it_had() { diff --git a/rs/moq-relay/src/web.rs b/rs/moq-relay/src/web.rs index 227d4af211..9e380bb655 100644 --- a/rs/moq-relay/src/web.rs +++ b/rs/moq-relay/src/web.rs @@ -935,7 +935,11 @@ async fn admit_http( request.server_name = request_host(uri, headers); request.remote = Some(remote.0); request.tls = mtls.and_then(|Extension(MtlsPeer(identity))| auth::peer(&identity)); - state.auth.admit(request).await + state + .auth + .admit(request) + .await + .inspect_err(|err| state.cluster.refusals.record(err.into())) } /// Serve the announced broadcasts for a given prefix. @@ -957,6 +961,7 @@ async fn serve_announced( let lease = admit_http(&state, prefix, query, &uri, &headers, remote, mtls).await?; let Some(origin) = state.cluster.subscriber(lease.token()) else { + state.cluster.refusals.record(crate::refusals::Refusal::Forbidden); return Err(StatusCode::UNAUTHORIZED.into()); }; @@ -1006,6 +1011,7 @@ async fn serve_fetch( let broadcast = lease.token().root.to_string(); let Some(origin) = state.cluster.subscriber(lease.token()) else { + state.cluster.refusals.record(crate::refusals::Refusal::Forbidden); return Err(StatusCode::UNAUTHORIZED.into()); }; @@ -1126,6 +1132,75 @@ mod tests { use std::io::Write; use tempfile::TempDir; + #[tokio::test(start_paused = true)] + async fn http_refusals_count_at_both_endpoints() { + use crate::refusals::Refusal; + let dir = TempDir::new().unwrap(); + let (_, cert, _) = make_certs(&dir); + let certificates = moq_tokio::tls::Certificates::from_pem(&std::fs::read(cert).unwrap()).unwrap(); + for fetch in [false, true] { + for publish_only in [false, true] { + let auth = if publish_only { + crate::auth::Config { + public_publish: vec![moq_auth::Pattern::all()], + ..Default::default() + } + .init("test", &moq_tokio::tls::Connect::default(), false) + .unwrap() + } else { + auth::Auth::refuse("test") + }; + let cluster = cluster::Cluster::new(crate::cluster::Options::default()).unwrap(); + let web = Web::new(auth, cluster, certificates.clone(), Config::default()); + let state = web.state; + let remote = ConnectInfo(crate::listener::Peer("127.0.0.1:1234".parse().unwrap())); + let status = if fetch { + serve_fetch( + Path("room/video".to_owned()), + Query(FetchParams { + auth: AuthQuery { jwt: None }, + group: FetchGroup::Latest, + }), + None, + remote, + "/fetch/room/video".parse().unwrap(), + http::HeaderMap::new(), + State(state.clone()), + ) + .await + .map(|_| ()) + .into_response() + .status() + } else { + serve_announced( + Some(Path("room".to_owned())), + Query(AuthQuery { jwt: None }), + None, + remote, + "/announced/room".parse().unwrap(), + http::HeaderMap::new(), + State(state.clone()), + ) + .await + .map(|_| ()) + .into_response() + .status() + }; + assert_eq!(status, StatusCode::UNAUTHORIZED); + let reason = if publish_only { + Refusal::Forbidden + } else { + Refusal::Refused + }; + assert_eq!( + state.cluster.refusals.count(reason), + 1, + "fetch={fetch}, publish_only={publish_only}" + ); + } + } + } + fn make_certs(dir: &TempDir) -> (PathBuf, PathBuf, PathBuf) { make_named_certs(dir, "server", "localhost") }