refactor(net)!: remove Error::SessionClosed

quest/m1/README.md

@@ -147,7 +147,6 @@ audit's receive-memory, priority, and lease bugs rank ahead of that reshape.
 - [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`
 - [API-diff check](/quest/m1/api-diff.md) - a recipe diffs every published package's API against its last release, run on PRs into release and against main before the cut
 - [API Extractor checks](/quest/m1/api-extractor-checks.md) - API Extractor fails a PR on a forgotten export, an unresolved declaration, or an undocumented export, replacing the hand-rolled `.d.ts` checkers
 - [Generated JS API reference](/quest/m1/js-api-reference.md) - the doc site renders an API reference for every `@moq/*` package from API Extractor's doc model

rs/moq-ffi/src/error.rs

@@ -391,7 +391,11 @@ impl From<moq_net::Error> 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,
@@ -667,6 +671,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()));

rs/moq-gst/src/source/imp.rs

@@ -2021,7 +2021,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:?}");

rs/moq-hls/src/export/mod.rs

@@ -4337,7 +4337,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.

rs/moq-hls/src/server/routes.rs

@@ -923,7 +923,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",
@@ -1230,7 +1232,7 @@ mod tests {
 		// A timeline failure refuses listings, but its already-retained media
 		// still drains. Authorization checks must not broaden that failure.
 		let retained = broadcaster.rendition(Kind::Video, "video0").unwrap();
-		retained.fail(crate::Error::Moq(moq_net::Error::SessionClosed));
+		retained.fail(crate::Error::Moq(moq_net::Error::Transport("connection lost".into())));
 		assert!(retained.media_playlist(None).is_err());
 		assert!(retained.init().await.unwrap().is_some());
 		assert!(retained.segment("656e3d1b", 0).await.unwrap().is_some());

rs/moq-mux/src/container/source.rs

@@ -560,7 +560,7 @@ impl ExportSource {
 						moq_net::Error::Cancel
 						| moq_net::Error::Closed
 						| moq_net::Error::GoingAway
-						| moq_net::Error::SessionClosed
+						| moq_net::Error::Stream(moq_net::StreamError::Session(moq_net::SessionError::Cancel))
 						| moq_net::Error::Session(moq_net::SessionError::Cancel),
 					)))
 					| Poll::Pending

rs/moq-net/src/error.rs

@@ -521,11 +521,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
@@ -625,7 +620,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,
@@ -687,7 +682,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,

rs/moq-net/src/model/resume.rs

@@ -341,7 +341,6 @@ fn route_failed(err: &Error) -> bool {
 		Error::Transport(_)
 			| Error::IdleTimeout
 			| Error::Session(_)
-			| Error::SessionClosed
 			| Error::GoingAway
 			| Error::GoawayTimeout
 			| Error::Dropped
@@ -2010,8 +2009,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 {
@@ -2078,8 +2076,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")
@@ -2867,7 +2867,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();
@@ -2878,8 +2879,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);
 	}
@@ -3346,7 +3349,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);
 		}
@@ -3388,7 +3393,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);
 		}
@@ -3475,7 +3482,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);
@@ -3508,7 +3517,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 {
@@ -3522,7 +3535,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::<u64>::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()));
@@ -3564,7 +3579,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::<u64>::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);
 		}

rs/moq-net/src/model/track.rs

@@ -14,7 +14,7 @@
 //! The track is closed with [Error] when all writers or readers are dropped.
 
 use crate::sync::Mutex;
-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;
@@ -3038,14 +3038,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(())
@@ -12330,7 +12330,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() {
@@ -12345,9 +12345,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

rs/moq-relay/src/connection.rs

@@ -427,11 +427,7 @@ pub(crate) fn failure(error: &moq_net::Error) -> Option<moq_net::stats::Failure>
 	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);