From ce0ae233938a0f82e4e37777f3d6d5a214647ebe Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 22 Oct 2026 20:56:17 -0700 Subject: [PATCH 1/3] chore(quest): claim quest/m1/error-cleanups.md Quest-Claim: 14efbf5acb1dcab0 From ece5edf3ee00f7f71ffe3023612237f01b3ad5fa Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 22 Oct 2026 21:57:47 -0700 Subject: [PATCH 2/3] refactor(net)!: remove Error::SessionClosed track::Producer::close aborts open groups with StreamError::Session(SessionError::Cancel) directly, the value SessionClosed converted to, so the wire is unchanged. moq-ffi keeps reporting that local close as Closed. Co-Authored-By: Claude Opus 5.5 --- quest/m1/README.md | 1 - quest/m1/error-cleanups.md | 34 ------------------------------- rs/moq-ffi/src/error.rs | 19 ++++++++++++++++- rs/moq-gst/src/source/imp.rs | 2 +- rs/moq-hls/src/export/mod.rs | 4 +++- rs/moq-hls/src/server/routes.rs | 4 +++- rs/moq-net/MIGRATE.md | 8 +++++++- rs/moq-net/src/error.rs | 8 +------- rs/moq-net/src/model/resume.rs | 36 +++++++++++++++++++++++---------- rs/moq-net/src/model/track.rs | 18 +++++++++++------ rs/moq-relay/src/connection.rs | 7 +------ 11 files changed, 71 insertions(+), 70 deletions(-) delete mode 100644 quest/m1/error-cleanups.md diff --git a/quest/m1/README.md b/quest/m1/README.md index afb1c5d6779..bd934b0355a 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -150,7 +150,6 @@ tracks, so per-group stats stop costing cluster announces. - [Main is no slower than release](/quest/m1/release-perf/README.md) - every benchmark where main is slower than release is fixed or accepted with a reason, checked by a final paired run before the cut - [Subscribe update camera check](/quest/m1/subscribe-update-churn.md) - a person checks a physical-camera publish for SUBSCRIBE_UPDATE chatter; bbb and the fake camera measured quiet - [Unprefixed C++ free functions](/quest/m1/cpp-free-function-prefix.md) - generated free functions drop `moq_`, and `moq.hpp` deletes its five pass-through wrappers before C++ first publishes -- [Error cleanups](/quest/m1/error-cleanups.md) - delete the unused `Error::SessionClosed`; raw transport failures already map to `Unroutable` - [FromStr parsers](/quest/m1/from-str-trait.md) - hang, moq-auth, and moq-msf implement `FromStr` instead of inherent `from_str` - [Drop dead items](/quest/m1/drop-dead-items.md) - the audit's uncalled public items and test helpers are removed - [Warmup in the hang draft](/quest/m1/warmup-draft.md) - the draft lets a dual-decoder consumer present before `warmup`, matching the shipping `decode::Consumer` diff --git a/quest/m1/error-cleanups.md b/quest/m1/error-cleanups.md deleted file mode 100644 index 71e13ca42a0..00000000000 --- a/quest/m1/error-cleanups.md +++ /dev/null @@ -1,34 +0,0 @@ -# [XS] Error cleanups - -## Goal - -`moq_net::Error::SessionClosed`, which has no meaning of its own, is -deleted. - -## Plan - -Moved to m1 on 2026-10-22 (maintainer, from an external API audit): the -cut is breaking anyway, so this break ships with it instead of forcing -another breaking release later. - -Decided with the maintainer on 2026-10-17: the goal as drafted, from the -follow-ups of [#5729](https://github.com/moq-dev/moq/pull/5729), widened the -same day to Rust after review found Rust sends `Internal` too. - -#5729 landed as `50da59e67`. It maps `Transport`, `IdleTimeout` and a -codeless session failure to `Unroutable` in both languages, with unit tests -in `rs/moq-net/src/error.rs` and `js/net/src/error.test.ts`. Keep that -behavior and its tests; only the `SessionClosed` removal remains here. - -`SessionClosed` has match arms in `rs/moq-net` (`error.rs`, -`model/resume.rs`), `rs/moq-ffi/src/error.rs` and `rs/moq-relay`, and -test-only uses in `rs/moq-hls`, `rs/moq-gst` and `rs/moq-relay`. -`rs/moq-net/src/model/track.rs`'s `close` constructs it only to convert it; -decided 2026-10-22 (maintainer): keep the removal and construct -`StreamError::Session(SessionError::Cancel)` there directly. Removing a -public variant is a breaking Rust change: add the `rs/moq-net/MIGRATE.md` -entry. Wire: none. - -## Related - -- [One clean close](/quest/m1/api-audit/clean-close.md) - collapses the clean session-close spellings this removal touches diff --git a/rs/moq-ffi/src/error.rs b/rs/moq-ffi/src/error.rs index 5fc89ceff14..308b182363f 100644 --- a/rs/moq-ffi/src/error.rs +++ b/rs/moq-ffi/src/error.rs @@ -398,7 +398,11 @@ impl From for MoqError { reason: "transport idle timeout".into(), }, moq_net::Error::NotFound | moq_net::Error::NotFetchable => Self::NotFound, - moq_net::Error::Closed | moq_net::Error::GoingAway | moq_net::Error::SessionClosed => Self::Closed, + // A track whose own session closed aborts its open groups with this; a remote + // SESSION_CLOSED decodes as `Internal`, so only the local close lands here. + moq_net::Error::Closed + | moq_net::Error::GoingAway + | moq_net::Error::Stream(moq_net::StreamError::Session(moq_net::SessionError::Cancel)) => Self::Closed, moq_net::Error::Cancel => Self::Cancelled, moq_net::Error::Unauthorized => Self::Unauthorized, moq_net::Error::Expired => Self::Expired, @@ -674,6 +678,19 @@ mod tests { } } + /// A group cut by its own session closing is a shutdown; one reset by a peer is not. + #[test] + fn local_session_close_is_closed() { + let local = moq_net::StreamError::Session(moq_net::SessionError::Cancel); + assert!(matches!(MoqError::from(moq_net::Error::from(local)), MoqError::Closed)); + + let received = moq_net::StreamError::from_code(0x3); + let MoqError::Protocol { details } = MoqError::from(moq_net::Error::from(received)) else { + panic!("a peer's SESSION_CLOSED lost its code"); + }; + assert_eq!(details.kind, MoqProtocolKind::SessionClosed); + } + #[test] fn transport_stays_out_of_protocol() { let err = MoqError::from(moq_net::Error::Transport("dial failed".into())); diff --git a/rs/moq-gst/src/source/imp.rs b/rs/moq-gst/src/source/imp.rs index 9316007c684..7080b018d16 100644 --- a/rs/moq-gst/src/source/imp.rs +++ b/rs/moq-gst/src/source/imp.rs @@ -1724,7 +1724,7 @@ mod session_tests { } for err in [ moq_net::Error::Closed, - moq_net::Error::SessionClosed, + moq_net::Error::Stream(moq_net::StreamError::Session(moq_net::SessionError::Cancel)), moq_net::Error::Cancel, ] { assert!(super::source_lost(&err), "{err:?}"); diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index 2eed7f670e4..1fc65caa4a3 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -4272,7 +4272,9 @@ mod tests { /// The timeline's publisher loses its session: the track and its open group fail. fn cut(track: moq_net::track::Producer, timeline: moq_mux::timeline::Producer) { drop(timeline); - track.abort(moq_net::Error::SessionClosed).unwrap(); + track + .abort(moq_net::Error::Transport("connection lost".into())) + .unwrap(); } /// Wait until `rendition`'s playlist fails, returning the error it answers. diff --git a/rs/moq-hls/src/server/routes.rs b/rs/moq-hls/src/server/routes.rs index 08b28075ccd..29651effede 100644 --- a/rs/moq-hls/src/server/routes.rs +++ b/rs/moq-hls/src/server/routes.rs @@ -914,7 +914,9 @@ mod tests { assert_eq!(status(&app, "/live/manifest.mpd").await, StatusCode::OK); drop(timeline); - track.abort(moq_net::Error::SessionClosed).unwrap(); + track + .abort(moq_net::Error::Transport("connection lost".into())) + .unwrap(); for uri in [ "/live/video/video0/media.m3u8", "/live/video/video0/media.m3u8?start=0&end=2", diff --git a/rs/moq-net/MIGRATE.md b/rs/moq-net/MIGRATE.md index e16d361e179..bb76890fbb5 100644 --- a/rs/moq-net/MIGRATE.md +++ b/rs/moq-net/MIGRATE.md @@ -4,6 +4,12 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `Error::SessionClosed` is removed + +Nothing produces it any more. +`track::Producer::close` aborts its open groups with `Error::Stream(StreamError::Session(SessionError::Cancel))`, which encodes as the same `SESSION_CLOSED` stream code. +Match that, or any `StreamError::Session(_)`, where you matched `Error::SessionClosed`. + ### IETF whole-group FETCH answers name the requested group ([#6199](https://github.com/moq-dev/moq/issues/6199)) On moq-transport drafts 14 through 19, a whole-group `FETCH_OK` now names `{g, 0}` instead of `{g+1, 0}`. @@ -80,7 +86,7 @@ A group followed by a frameless one is judged against the next group that has a ### Ending a track aborts its open groups (#6111) Aborting a track (`abort` or `abort_unused`) now aborts its unfinished groups with the same error, so their readers end with it and their writers fail. -Dropping the last `track::Producer` without a finish aborts them with `Error::Dropped`, and `close` with `Error::SessionClosed`. +Dropping the last `track::Producer` without a finish aborts them with `Error::Dropped`, and `close` with `Error::Stream(StreamError::Session(SessionError::Cancel))`. `track::Producer::close`, `track::Producer::received`, and `group::Producer::info` are now public. ### Priorities are signed, with 0 unset (#6142) diff --git a/rs/moq-net/src/error.rs b/rs/moq-net/src/error.rs index 0f4f1e052dc..ab8d5c22204 100644 --- a/rs/moq-net/src/error.rs +++ b/rs/moq-net/src/error.rs @@ -517,11 +517,6 @@ pub enum Error { #[error("unsupported extension")] UnsupportedExtension, - /// The stream was torn down because the session closed. The specific reason travels on - /// the session close, not here. - #[error("session closed")] - SessionClosed, - /// A session-scoped protocol error, with its registry and verbatim code. /// /// Produced by [`from_transport`](Self::from_transport) for a session close, and by @@ -621,7 +616,7 @@ impl From<&Error> for SessionError { Error::Stream(StreamError::App(app)) => Self::App(*app), // A stream-scoped code has no meaning in this registry; don't forward the number. Error::Stream(_) => Self::Internal, - Error::Cancel | Error::Closed | Error::GoingAway | Error::SessionClosed => Self::Cancel, + Error::Cancel | Error::Closed | Error::GoingAway => Self::Cancel, Error::Unauthorized => Self::Unauthorized, Error::Expired => Self::Expired, Error::Unsupported => Self::Unsupported, @@ -682,7 +677,6 @@ impl From<&Error> for StreamError { // this session is fine, and the failed session's code means nothing here. Error::Session(_) | Error::Transport(_) | Error::IdleTimeout => Self::Unroutable, Error::Cancel | Error::Closed => Self::Cancel, - Error::SessionClosed => Self::Session(SessionError::Cancel), Error::Decode(coding::DecodeError::Session(err)) => Self::Session(err.clone()), Error::Old => Self::Old, Error::Evicted => Self::Evicted, diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 9279f581176..11d95c892be 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -340,7 +340,6 @@ fn route_failed(err: &Error) -> bool { Error::Transport(_) | Error::IdleTimeout | Error::Session(_) - | Error::SessionClosed | Error::GoingAway | Error::GoawayTimeout | Error::Dropped @@ -2012,8 +2011,7 @@ mod test { use futures::FutureExt; use super::*; - use crate::Timestamp; - use crate::broadcast; + use crate::{SessionError, Timestamp, broadcast}; /// A route's copy of the track. fn copy() -> track::Producer { @@ -2080,8 +2078,10 @@ mod test { newer.write_frame(ts(1), b"newer".as_ref()).unwrap(); newer.finish().unwrap(); routes.serve(b.consume()); - open.abort(Error::SessionClosed).unwrap(); - a.abort(Error::SessionClosed).unwrap(); + open.abort(Error::Stream(StreamError::Session(SessionError::Cancel))) + .unwrap(); + a.abort(Error::Stream(StreamError::Session(SessionError::Cancel))) + .unwrap(); assert!(held.poll_read_frame(&waiter).is_pending()); let Poll::Ready(request) = dynamic.poll_requested_group(&waiter) else { panic!("recovery fetch") @@ -3304,7 +3304,9 @@ mod test { whole.write_frame(ts(1), b"a".as_ref()).unwrap(); whole.write_frame(ts(2), b"b".as_ref()).unwrap(); whole.finish().unwrap(); - failing.abort(Error::SessionClosed).unwrap(); + failing + .abort(Error::Stream(StreamError::Session(SessionError::Cancel))) + .unwrap(); assert_eq!(read(&mut reading), Some(b"b".to_vec()), "ordered={ordered}"); assert_eq!(read(&mut reading), None); } @@ -3346,7 +3348,9 @@ mod test { whole.write_frame(ts(1), b"a".as_ref()).unwrap(); whole.write_frame(ts(2), b"b".as_ref()).unwrap(); whole.finish().unwrap(); - failing.abort(Error::SessionClosed).unwrap(); + failing + .abort(Error::Stream(StreamError::Session(SessionError::Cancel))) + .unwrap(); assert_eq!(read(&mut reading), Some(b"b".to_vec()), "ordered={ordered}"); assert_eq!(read(&mut reading), None); } @@ -3433,7 +3437,9 @@ mod test { assert_eq!(reading.sequence, 1); // a's route fails too, after it finished the group. a.clone().abort(Error::Dropped).unwrap(); - failing.abort(Error::SessionClosed).unwrap(); + failing + .abort(Error::Stream(StreamError::Session(SessionError::Cancel))) + .unwrap(); assert_eq!(read(&mut reading), Some(b"a".to_vec())); assert_eq!(read(&mut reading), Some(b"b".to_vec())); assert_eq!(read(&mut reading), None); @@ -3466,7 +3472,11 @@ mod test { failing.write_frame(ts(1), b"a".as_ref()).unwrap(); let mut failing = Some(failing); if early { - failing.take().unwrap().abort(Error::SessionClosed).unwrap(); + failing + .take() + .unwrap() + .abort(Error::Stream(StreamError::Session(SessionError::Cancel))) + .unwrap(); } let noop = kio::Waiter::noop(); let res = match ordered { @@ -3480,7 +3490,9 @@ mod test { // Reading on must not let go of the replaced copy while the group still needs it. assert_eq!(drain(&mut sub, ordered), Vec::::new()); if let Some(failing) = failing { - failing.abort(Error::SessionClosed).unwrap(); + failing + .abort(Error::Stream(StreamError::Session(SessionError::Cancel))) + .unwrap(); } assert_eq!(read(&mut reading), Some(b"a".to_vec())); assert_eq!(read(&mut reading), Some(b"b".to_vec())); @@ -3522,7 +3534,9 @@ mod test { // The old route goes away, and the reader polls on and sees its copy end. a.clone().abort(Error::Dropped).unwrap(); assert_eq!(drain(&mut sub, ordered), Vec::::new()); - failing.abort(Error::SessionClosed).unwrap(); + failing + .abort(Error::Stream(StreamError::Session(SessionError::Cancel))) + .unwrap(); assert_eq!(read(&mut reading), Some(b"b".to_vec())); assert_eq!(read(&mut reading), None); } diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index aa2ebf373a3..ce0c8a9ed3f 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -13,7 +13,7 @@ //! //! The track is closed with [Error] when all writers or readers are dropped. -use crate::{Error, Result, StreamError, Timescale, Timestamp, coding}; +use crate::{Error, Result, SessionError, StreamError, Timescale, Timestamp, coding}; use crate::{broadcast, cache, group, stats}; use super::delivered::Delivered; @@ -2912,14 +2912,14 @@ impl Producer { /// End the track without declaring where it ends, as a closed session does. /// /// Consumers drain the groups that finished and then end cleanly. The open groups are - /// aborted with [`Error::SessionClosed`], so a consumer that already pulled one ends + /// aborted with [`StreamError::Session`], so a consumer that already pulled one ends /// with it and their writers fail. A reader following the track's routes resumes from /// another one where this copy stopped. Use [`Self::finish`] or [`Self::finish_at`] /// when the end is known. pub fn close(self) -> Result<()> { let mut state = self.modify()?; state.sealed = true; - let open = state.tear_down(Error::SessionClosed); + let open = state.tear_down(Error::Stream(StreamError::Session(SessionError::Cancel))); state.close(); open.abort(); Ok(()) @@ -11894,7 +11894,7 @@ mod test { } /// Closing a track aborts its open groups like [`Producer::abort`], with - /// [`Error::SessionClosed`]. + /// [`StreamError::Session`]. #[test] #[cfg(feature = "async")] fn close_ends_a_pulled_open_group() { @@ -11909,9 +11909,15 @@ mod test { producer.close().unwrap(); let read = group.read_frame().now_or_never().expect("should not block"); - assert!(matches!(read, Err(Error::SessionClosed)), "{read:?}"); + assert!( + matches!(read, Err(Error::Stream(StreamError::Session(SessionError::Cancel)))), + "{read:?}" + ); let write = writer.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"y")); - assert!(matches!(write, Err(Error::SessionClosed)), "{write:?}"); + assert!( + matches!(write, Err(Error::Stream(StreamError::Session(SessionError::Cancel)))), + "{write:?}" + ); } /// Aborting an unused track aborts an open group whose reader outlived the track's diff --git a/rs/moq-relay/src/connection.rs b/rs/moq-relay/src/connection.rs index e9c44913af2..cd7ebfa99f0 100644 --- a/rs/moq-relay/src/connection.rs +++ b/rs/moq-relay/src/connection.rs @@ -427,11 +427,7 @@ pub(crate) fn failure(error: &moq_net::Error) -> Option use moq_net::stats::Failure; use moq_net::{Error, SessionError}; match error { - Error::Cancel - | Error::Closed - | Error::GoingAway - | Error::SessionClosed - | Error::Session(SessionError::Cancel) => None, + Error::Cancel | Error::Closed | Error::GoingAway | Error::Session(SessionError::Cancel) => None, Error::IdleTimeout => Some(Failure::IdleTimeout), Error::Transport(_) => Some(Failure::Transport), _ => Some(Failure::Application), @@ -468,7 +464,6 @@ mod tests { Error::Cancel, Error::Closed, Error::GoingAway, - Error::SessionClosed, Error::Session(SessionError::Cancel), ] { assert_eq!(failure(&error), None); From 49d8db1120442015a3cd57d288656973152e8862 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 23 Oct 2026 12:41:29 -0700 Subject: [PATCH 3/3] test(net): abort new resume test routes with the stream close error A test that landed on main after this branch last merged still aborted with Error::SessionClosed. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/resume.rs | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index f8dad2c2c2a..0810e7913f8 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -2866,7 +2866,8 @@ mod test { sub.poll_recv_group(&kio::Waiter::noop()).is_pending(), "B replays only group 0" ); - b.abort(Error::SessionClosed).unwrap(); + b.abort(Error::Stream(StreamError::Session(SessionError::Cancel))) + .unwrap(); // C lacks the group; reading a later one from it lets go of what nothing needs. let c = copy(); @@ -2877,8 +2878,10 @@ mod test { assert_eq!(recv(&mut sub).sequence, 1); assert!(sub.poll_recv_group(&kio::Waiter::noop()).is_pending()); - open.abort(Error::SessionClosed).unwrap(); - a.abort(Error::SessionClosed).unwrap(); + open.abort(Error::Stream(StreamError::Session(SessionError::Cancel))) + .unwrap(); + a.abort(Error::Stream(StreamError::Session(SessionError::Cancel))) + .unwrap(); assert_eq!(read(&mut reading), Some(b"b".to_vec())); assert_eq!(read(&mut reading), None); }