From e9ccc1b559e9797ed6f168dcc49be887e9829720 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 30 Sep 2026 12:34:14 -0700 Subject: [PATCH 1/3] chore: claim IETF close drain quest Co-Authored-By: GPT-6 From 9d126ec978496c223f992b2dce961f11b7b5ace0 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 30 Sep 2026 13:08:22 -0700 Subject: [PATCH 2/3] fix(net): drain IETF serves before session close Co-Authored-By: GPT-6 --- doc/lib/rs/moq-net.md | 4 +- quest/m1/README.md | 1 - quest/m1/close-tail.md | 3 +- quest/m1/ietf-drain-before-close.md | 14 ------ rs/moq-net/src/driver.rs | 8 ++-- rs/moq-net/src/ietf/publisher.rs | 68 ++++++++++++++++++++++++++++- rs/moq-net/src/ietf/session.rs | 43 ++++++++++++++++-- rs/moq-net/src/session.rs | 2 +- rs/moq-net/tests/session_close.rs | 37 +++++++++++++--- 9 files changed, 146 insertions(+), 34 deletions(-) delete mode 100644 quest/m1/ietf-drain-before-close.md diff --git a/doc/lib/rs/moq-net.md b/doc/lib/rs/moq-net.md index 322894a4c3..e3ed0a54b6 100644 --- a/doc/lib/rs/moq-net.md +++ b/doc/lib/rs/moq-net.md @@ -62,8 +62,8 @@ Dropping the last session handle requests closure on the next poll, and peer has not acknowledged yet. `session.close().await` first waits, up to one second, for finished tracks to deliver their last groups and FIN, returning `Error::Timeout` if it gave up. Finish or abort live tracks before calling it. -moq-transport (IETF) sessions close without waiting. Dropping the driver -cancels the session. `moq-tokio` and `moq-wasm` drive sessions for their +This drain applies to both moq-lite and moq-transport (IETF). Dropping the +driver cancels the session. `moq-tokio` and `moq-wasm` drive sessions for their callers. `origin::Producer::new` returns a driver with the same `time::Driver` diff --git a/quest/m1/README.md b/quest/m1/README.md index 12f1f067c0..01b8c7aabe 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -30,7 +30,6 @@ QUIC studies there on that rule. - [FFI shape](/quest/m1/ffi-shape/README.md) - the bindings mirror Rust's layers: net at the root, then media, json, flate, audio, and video namespaces built from the handle below - [Android logcat](/quest/m1/android-logcat.md) - on dev, Android builds always log to logcat and the `android-logcat` feature is gone - [Raw stream codes](/quest/m1/raw-stream-codes.md) - raw QUIC stream resets and stops carry the application's code, not an HTTP/3-mapped one -- [IETF drain before close](/quest/m1/ietf-drain-before-close.md) - moq-transport sessions deliver finished tracks before a graceful close, as moq-lite does - [Close waits for the tail](/quest/m1/close-tail.md) - on lite-07, `close()` returns `Ok` only after each subscriber FINs its Subscribe Stream, having read the track to its end - [Demo serve-hls renditions](/quest/m1/serve-hls-renditions.md) - `just pub serve-hls` serves 720p and 144p instead of two 256-wide copies - [moq play drain tail](/quest/m1/play-drain-tail.md) - retired renditions and finite tracks play their last 10 ms of audio diff --git a/quest/m1/close-tail.md b/quest/m1/close-tail.md index 1c1e75dfb9..7dc35b91a5 100644 --- a/quest/m1/close-tail.md +++ b/quest/m1/close-tail.md @@ -38,7 +38,7 @@ Decided (maintainer, 2026-09-30): so close keeps today's ack-based drain rather than time out every close. #4508's repro is on lite-05, and that path stays as it is: this quest closes the issue by fixing the close on the version that can carry the FIN rule. - IETF sessions stay with [IETF drain before close](/quest/m1/ietf-drain-before-close.md). + IETF sessions keep their ack-based drain. - Both subscribers change: moq-net and `@moq/net`. Update `drafts/draft-lcurley-moq-lite.md` (the Subscribe section gains the @@ -65,5 +65,4 @@ after reading the track's end, and a publisher closing gracefully waits for it. ## Related -- [IETF drain before close](/quest/m1/ietf-drain-before-close.md) - the same drain for moq-transport sessions - [SUBSCRIBE_DROP](/quest/m1/subscribe-drop.md) - replaces the lite-07 tail accounting the FIN rule waits on diff --git a/quest/m1/ietf-drain-before-close.md b/quest/m1/ietf-drain-before-close.md deleted file mode 100644 index 732dd31a91..0000000000 --- a/quest/m1/ietf-drain-before-close.md +++ /dev/null @@ -1,14 +0,0 @@ -# [S] IETF sessions drain before close - -## Goal - -`Session::close` on a moq-transport (IETF) session waits, like moq-lite, for -finished tracks to deliver their last groups and FIN before closing, within -the shared one second `CLOSE_TIMEOUT`. - -## Plan - -`Protocol::drained` in `rs/moq-net/src/driver.rs` returns `true` for IETF, so -its sessions close at once. Count the IETF publisher's in-flight subscribe and -fetch serves the way `lite::Publisher::drained` does, and report them there. -Extend `rs/moq-net/tests/session_close.rs` to cover an IETF version. diff --git a/rs/moq-net/src/driver.rs b/rs/moq-net/src/driver.rs index 22c677e4bb..29c2e5b317 100644 --- a/rs/moq-net/src/driver.rs +++ b/rs/moq-net/src/driver.rs @@ -34,7 +34,7 @@ pub struct Driver { pub(crate) enum Protocol { /// Boxed for size only: a concrete box, so `Send` stays inferred. Lite(Box>), - Ietf(crate::util::MaybeSendBox<'static, Result<(), Error>>), + Ietf(crate::ietf::Driver), } /// Protocol and lifecycle work owned by the driver. @@ -74,16 +74,16 @@ impl Protocol { fn poll(&mut self, waiter: &kio::Waiter) -> Poll> { match self { Self::Lite(driver) => driver.poll(waiter), - Self::Ietf(driver) => waiter.poll_future(driver.as_mut()), + Self::Ietf(driver) => waiter.poll_future(std::pin::Pin::new(driver)), } } /// Whether the protocol owes the peer no queued data, so a draining close can - /// proceed. The IETF driver does not track this and closes at once. + /// proceed. fn drained(&self) -> bool { match self { Self::Lite(driver) => driver.drained(), - Self::Ietf(_) => true, + Self::Ietf(driver) => driver.drained(), } } } diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 5de4a889e8..ac5a7b7742 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -5,7 +5,7 @@ use std::{ ops::Bound, sync::{ Arc, - atomic::{AtomicU64, Ordering}, + atomic::{AtomicU64, AtomicUsize, Ordering}, }, task::{Poll, ready}, time::Duration, @@ -323,6 +323,16 @@ pub(super) struct Publisher { // Shared across request handlers; None marks a dispatched subscription still resolving. joins: kio::Shared>>, version: Version, + // Dispatched finite serves, including those not yet polled. + pub(super) owed: Arc, +} + +struct Serve(Arc); + +impl Drop for Serve { + fn drop(&mut self) { + self.0.fetch_sub(1, Ordering::Relaxed); + } } /// The snapshot a joining FETCH inherits from its subscription. @@ -378,6 +388,7 @@ where peer_setup, joins: Default::default(), version, + owed: Default::default(), } } @@ -462,6 +473,11 @@ where stream: Stream, ) -> Result, Error> { let this = self.clone(); + // Count at dispatch so close cannot pass a request waiting for its first poll. + let serve = matches!(id, ietf::Subscribe::ID | ietf::Fetch::ID).then(|| { + self.owed.fetch_add(1, Ordering::Relaxed); + Serve(self.owed.clone()) + }); let task = match id { ietf::Subscribe::ID => { let msg = ietf::Subscribe::decode_msg(&mut data, this.version)?; @@ -471,6 +487,7 @@ where tracing::debug!(message = ?msg, "received subscribe"); let task = this.run_subscribe_stream(stream, msg); async move { + let _serve = serve; if let Err(err) = task.await { tracing::debug!(%err, "subscribe stream error"); } @@ -484,6 +501,7 @@ where } tracing::debug!(message = ?msg, "received fetch"); async move { + let _serve = serve; if let Err(err) = this.run_fetch_stream(stream, msg).await { tracing::debug!(%err, "fetch stream error"); } @@ -3006,6 +3024,54 @@ mod serve_tests { } } + /// A close must count requests before their first poll and release them even + /// when a dispatched task is cancelled or its message is refused. + #[tokio::test(start_paused = true)] + async fn drain_counts_dispatched_subscribe_and_fetch() { + for version in [Version::Draft14, Version::Draft19, Version::Draft20] { + let h = serve(version); + let mut subscribe_body = bytes::BytesMut::new(); + subscribe(Filter::NextObject, None) + .encode_msg(&mut subscribe_body, version) + .unwrap(); + let mut fetch_body = bytes::BytesMut::new(); + ietf::Fetch { + request_id: FETCH_ID, + subscriber_priority: 128, + group_order: GroupOrder::Ascending, + fetch_type: FetchType::Standalone { + namespace: crate::Path::new("missing"), + track: "video".into(), + start: Location { group: 0, object: 0 }, + end: Location { group: 0, object: 1 }, + }, + } + .encode_msg(&mut fetch_body, version) + .unwrap(); + + for (id, body) in [ + (ietf::Subscribe::ID, subscribe_body.freeze()), + (ietf::Fetch::ID, fetch_body.freeze()), + ] { + let stream = Stream::open(&mut h.session.clone(), version).await.unwrap(); + let task = h.publisher.handle_stream(id, body.clone(), stream).unwrap(); + assert_eq!(h.publisher.owed.load(Ordering::Relaxed), 1, "count before polling"); + drop(task); + assert_eq!(h.publisher.owed.load(Ordering::Relaxed), 0, "release a cancelled task"); + + let stream = Stream::open(&mut h.session.clone(), version).await.unwrap(); + assert!(h.publisher.handle_stream(id, bytes::Bytes::new(), stream).is_err()); + assert_eq!(h.publisher.owed.load(Ordering::Relaxed), 0, "release malformed input"); + + if id == ietf::Fetch::ID { + let stream = Stream::open(&mut h.session.clone(), version).await.unwrap(); + h.publisher.handle_stream(id, body, stream).unwrap().await; + assert_eq!(h.publisher.owed.load(Ordering::Relaxed), 0, "release a completed fetch"); + } + } + } + } + /// A distinctive request id, so `[FetchHeader::TYPE, REQUEST_ID]` is a usable needle. const REQUEST_ID: u64 = 0x2B; diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index 439c259b4e..16e23ae492 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -1,3 +1,13 @@ +use std::{ + future::Future, + pin::Pin, + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, + task::{Context, Poll}, +}; + use crate::origin; use crate::{ Error, Hop, SessionError, StreamError, @@ -67,7 +77,28 @@ pub struct Config { pub peer_declared: Option, } -pub fn start(config: Config) -> Result<(MaybeSendBox<'static, Result<(), Error>>, crate::goaway::Handle), Error> +/// Runs the IETF protocol and tracks the data owed to its peer. +pub struct Driver { + run: MaybeSendBox<'static, Result<(), Error>>, + owed: Arc, +} + +impl Driver { + /// Whether no dispatched subscription or fetch still owes the peer data. + pub fn drained(&self) -> bool { + self.owed.load(Ordering::Relaxed) == 0 + } +} + +impl Future for Driver { + type Output = Result<(), Error>; + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + self.run.as_mut().poll(cx) + } +} + +pub fn start(config: Config) -> Result<(Driver, crate::goaway::Handle), Error> where S: crate::transport::poll::Boxable, { @@ -94,6 +125,8 @@ where // server to open connections (draft-19 sect 10.4). let (goaway_handle, goaway) = crate::goaway::Handle::new(!client); + let owed = Arc::new(AtomicUsize::new(0)); + let serving = owed.clone(); let driver = async move { // Our own Hop ID, taken from whichever origin the caller actually supplied so // every session out of this process stamps the same one and cross-session loop @@ -132,7 +165,7 @@ where let control = Control::new(request_id_max, client); let adapter = ControlStreamAdapter::new(session.clone(), control.clone(), version); - let publisher = Publisher::new( + let mut publisher = Publisher::new( runtime.clone(), adapter.clone(), publish, @@ -141,6 +174,7 @@ where peer_setup.clone(), version, ); + publisher.owed = serving.clone(); let (tasks, mut task_set) = TaskSet::new(); let subscriber = Subscriber::new( runtime.clone(), @@ -293,7 +327,7 @@ where }; let control = Control::new(None, client); - let publisher = Publisher::new( + let mut publisher = Publisher::new( runtime.clone(), session.clone(), publish, @@ -302,6 +336,7 @@ where peer_setup.clone(), version, ); + publisher.owed = serving.clone(); let (tasks, mut task_set) = TaskSet::new(); let subscriber = Subscriber::new( runtime.clone(), @@ -438,7 +473,7 @@ where } .maybe_boxed(); - Ok((driver, goaway_handle)) + Ok((Driver { run: driver, owed }, goaway_handle)) } /// What a peer's SETUP told us, beyond the stream it arrived on. diff --git a/rs/moq-net/src/session.rs b/rs/moq-net/src/session.rs index ae54e504d2..aed7ffb7e9 100644 --- a/rs/moq-net/src/session.rs +++ b/rs/moq-net/src/session.rs @@ -169,7 +169,7 @@ impl Session { /// the session's terminal error if it ended some other way first. A track that /// is still live never finishes, so finish or abort tracks before closing. /// - /// moq-transport (IETF) sessions close without waiting. + /// Both moq-lite and moq-transport (IETF) drain before closing. pub async fn close(self) -> Result<(), Error> { if let Ok(mut close) = self.close.write() && close.is_none() diff --git a/rs/moq-net/tests/session_close.rs b/rs/moq-net/tests/session_close.rs index 9c7308f6ac..408bc78d54 100644 --- a/rs/moq-net/tests/session_close.rs +++ b/rs/moq-net/tests/session_close.rs @@ -29,14 +29,14 @@ struct Setup { _broadcast: moq_net::broadcast::Producer, } -async fn setup() -> Setup { +async fn setup(version: &str) -> Setup { let publisher = produce_origin(1); let broadcast = publisher.create_broadcast("bcast").unwrap(); let track = broadcast.create_track("video", None).unwrap(); broadcast.announce(Default::default()).unwrap(); let subscriber = produce_origin(2); - let mut options = MockConnectOptions::new("moq-lite-05".parse::().unwrap()); + let mut options = MockConnectOptions::new(version.parse::().unwrap()); options.client_publish = Some(publisher.consume()); options.server_subscribe = Some(subscriber.clone()); let pair = connect_mock(options).await; @@ -88,9 +88,18 @@ async fn setup() -> Setup { /// close is requested before the session wrote them. #[tokio::test(start_paused = true)] async fn close_delivers_a_finished_track() { + close_delivers_a_finished_track_for("moq-lite-05").await; +} + +#[tokio::test(start_paused = true)] +async fn ietf_close_delivers_a_finished_track() { + close_delivers_a_finished_track_for("moq-transport-20").await; +} + +async fn close_delivers_a_finished_track_for(version: &str) { let Setup { pair, track, reader, .. - } = setup().await; + } = setup(version).await; let mut group = track.append_group().unwrap(); group.write_frame(Timestamp::ZERO, b"last".as_slice()).unwrap(); @@ -119,9 +128,18 @@ async fn close_delivers_a_finished_track() { /// session closes anyway. #[tokio::test(start_paused = true)] async fn close_gives_up_on_a_live_track() { + close_gives_up_on_a_live_track_for("moq-lite-05").await; +} + +#[tokio::test(start_paused = true)] +async fn ietf_close_gives_up_on_a_live_track() { + close_gives_up_on_a_live_track_for("moq-transport-20").await; +} + +async fn close_gives_up_on_a_live_track_for(version: &str) { let Setup { pair, track: _track, .. - } = setup().await; + } = setup(version).await; let started = tokio::time::Instant::now(); let res = tokio::time::timeout(TIMEOUT, pair.client.close()) @@ -134,9 +152,18 @@ async fn close_gives_up_on_a_live_track() { /// An abort from another handle cuts a drain short. #[tokio::test(start_paused = true)] async fn abort_cuts_a_drain_short() { + abort_cuts_a_drain_short_for("moq-lite-05").await; +} + +#[tokio::test(start_paused = true)] +async fn ietf_abort_cuts_a_drain_short() { + abort_cuts_a_drain_short_for("moq-transport-20").await; +} + +async fn abort_cuts_a_drain_short_for(version: &str) { let Setup { pair, track: _track, .. - } = setup().await; + } = setup(version).await; let other = pair.client.clone(); let started = tokio::time::Instant::now(); From 461ae1b04de9a8b7396f497d046503304bd24c75 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 3 Oct 2026 09:51:37 -0700 Subject: [PATCH 3/3] docs(net): rewrap the session close paragraph Co-Authored-By: Claude Opus 5.5 --- doc/lib/rs/moq-net.md | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/doc/lib/rs/moq-net.md b/doc/lib/rs/moq-net.md index 1a3a83ee0a..44adfc8168 100644 --- a/doc/lib/rs/moq-net.md +++ b/doc/lib/rs/moq-net.md @@ -65,9 +65,10 @@ second, for finished tracks to deliver their last groups and FIN, returning Both protocols withdraw the session's announcements and wait for delivery under that same deadline. IETF drafts 14 through 16 send withdrawals without waiting. Dropping the driver cancels the session. `moq-tokio` and `moq-wasm` -drive sessions for their callers. A deliberate local session close ends received tracks cleanly after -their delivered groups. A peer close ends received tracks and open group -readers with the session error, preserving its close code. +drive sessions for their callers. A deliberate local session close ends +received tracks cleanly after their delivered groups. A peer close ends +received tracks and open group readers with the session error, preserving its +close code. `origin::Producer::new` returns a driver with the same `time::Driver` interface. It calls `cache::Pool::gc(now)` after each poll and folds the next