fix(uring): close connections when the last session handle drops

quest/m2/README.md

@@ -139,7 +139,6 @@ ranked just below deterministic TS legs, which it requires.
 - [Archive S3 wire proof](/quest/m2/archive-s3.md) - the archive proof also runs through the S3 client against an in-process S3-compatible server
 - [Archive recovery listing](/quest/m2/archive-recovery-listing.md) - a resumed DVR lists what changed since its checkpoint, not every stored group
 - [IETF on the ring](/quest/m2/uring-ietf.md) - the io_uring workers serve moq-transport sessions too, so a uring relay drops no client protocol
-- [Dropped uring session closes](/quest/m2/uring-drop-close.md) - a moq-uring session dropped without close() closes its connection
 - [One connection lifecycle](/quest/m2/uring-connection-log.md) - every accept path, io_uring included, logs a connection's span and close from one owner, so Interop's accounting works on the ring
 - [Relay io_uring packages](/quest/m2/relay-io-uring-package.md) - Linux relay packages ship io_uring once the ring is on par with tokio
 - [Opus implementation](/quest/m2/audio-opus-backend.md) - compare Opus codec quality, CPU, build cost, and the loss recovery each backend offers

quest/m2/relay-io-uring-package.md

@@ -36,5 +36,4 @@ Public API: none. Wire: none.
 
 - [moq-transport on io_uring](/quest/m2/uring-ietf.md) - a packaged relay
   must not drop protocols when the ring is on
-- [Dropped session close](/quest/m2/uring-drop-close.md) - closes are delivered when a session is dropped on the ring
 - [One port on the io_uring workers](/quest/m2/uring-demux.md) - a one-port config is served on the ring too

rs/moq-uring/src/quic/driver/connection.rs

@@ -278,8 +278,8 @@ impl Inner {
 /// concurrent pending operations don't trample each other's wakeups. Dropping
 /// every handle (and every stream) drops the driver's `Rc` peers, but the
 /// driver itself keeps the connection alive until it ends; close explicitly
-/// with [`close`](web_transport_trait::poll::Session::close) (which moq's
-/// session machine does).
+/// with [`close`](web_transport_trait::poll::Session::close), or wrap it in
+/// [`crate::quic::web::Session`], whose last handle closes it.
 ///
 /// That close only records the code: the driver task is what frames the
 /// CONNECTION_CLOSE and hands it to the socket. Drive the worker until

rs/moq-uring/src/quic/web.rs

@@ -429,8 +429,10 @@ struct State {
 ///
 /// [`Request::respond`] builds the web
 /// flavor; [`Session::raw`] wraps a raw-QUIC [`Connection`] with the layering
-/// disabled. Clones share the session.
+/// disabled. Clones share the session; dropping the last closes the connection.
 pub struct Session {
+	// Capsule tasks retain Web/Connection, but only public session handles own this.
+	handles: Rc<()>,
 	conn: Connection,
 	web: Option<Rc<Web>>,
 	/// What [`protocol`](web_transport_trait::poll::Session::protocol)
@@ -455,6 +457,7 @@ impl Session {
 	pub fn raw(conn: Connection) -> Self {
 		let protocol = web_transport_trait::poll::Session::protocol(&conn).map(str::to_owned);
 		Self {
+			handles: Rc::new(()),
 			conn,
 			web: None,
 			protocol,
@@ -522,6 +525,7 @@ impl Session {
 			.spawn(async move { read_capsules(capsules, capsule_conn, recv).await });
 
 		Self {
+			handles: Rc::new(()),
 			conn,
 			web: Some(web),
 			protocol,
@@ -542,6 +546,7 @@ impl Session {
 impl Clone for Session {
 	fn clone(&self) -> Self {
 		Self {
+			handles: self.handles.clone(),
 			conn: self.conn.clone(),
 			web: self.web.clone(),
 			protocol: self.protocol.clone(),
@@ -551,6 +556,14 @@ impl Clone for Session {
 	}
 }
 
+impl Drop for Session {
+	fn drop(&mut self) {
+		if Rc::strong_count(&self.handles) == 1 {
+			web_transport_trait::poll::Session::close(self, 0, "dropped");
+		}
+	}
+}
+
 impl std::fmt::Debug for Session {
 	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
 		f.debug_struct("Session")

rs/moq-uring/tests/integration/endpoint.rs

@@ -99,6 +99,72 @@ fn dial_and_accept_share_one_socket() {
 		.expect("worker");
 }
 
+/// A session owns the connection independently of the driver and stream handles.
+#[test]
+fn dropping_the_last_raw_session_closes_the_connection() {
+	dropped_raw_session_closes(false);
+}
+
+#[test]
+fn an_explicit_raw_close_survives_the_last_session_drop() {
+	dropped_raw_session_closes(true);
+}
+
+fn dropped_raw_session_closes(explicit: bool) {
+	let Some(mut worker) = worker() else { return };
+	let handle = worker.handle();
+	let certs = support::certs().expect("certificates");
+	let socket = || {
+		handle
+			.udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default())
+			.expect("socket")
+	};
+	let server = quic::Endpoint::new(
+		socket(),
+		quic::endpoint::Config::default().with_server(server_config(&certs)),
+	)
+	.expect("server endpoint");
+	let client = quic::Endpoint::new(socket(), quic::endpoint::Config::default()).expect("client endpoint");
+	worker
+		.block_on(async move {
+			let conn = client.connect(&dial_config(server.local_addr())).await.expect("dial");
+			let mut peer = server.accept().await.expect("accept");
+			let session = quic::web::Session::raw(conn);
+			let mut remaining = session.clone();
+			drop(session);
+			let mut send = std::future::poll_fn(|cx| remaining.poll_open_uni(cx))
+				.await
+				.expect("surviving clone opens a stream");
+			std::future::poll_fn(|cx| send.poll_write(cx, b"alive"))
+				.await
+				.expect("write");
+			let mut recv = std::future::poll_fn(|cx| peer.poll_accept_uni(cx))
+				.await
+				.expect("accept stream");
+			let mut payload = [0; 5];
+			let read = std::future::poll_fn(|cx| recv.poll_read(cx, &mut payload))
+				.await
+				.expect("read");
+			assert_eq!(read, Some(5));
+			assert_eq!(&payload, b"alive");
+			let code = if explicit {
+				remaining.close(CLOSE_CODE, CLOSE_REASON);
+				u64::from(CLOSE_CODE)
+			} else {
+				0
+			};
+			drop(remaining);
+			// Stream handles intentionally survive the final session handle.
+			let error = std::future::poll_fn(|cx| peer.poll_closed(cx)).await;
+			assert!(
+				matches!(error, quic::Error::App { code: actual, .. } if actual == code),
+				"peer saw {error:?}"
+			);
+			drop((send, recv));
+		})
+		.expect("worker");
+}
+
 /// A close reaches both ends carrying the code and reason it was given. The
 /// closer reads its own close back out of `poll_closed`, which is what moq's
 /// session machine waits on, and the peer reads the same one off the wire.

rs/moq-uring/tests/integration/web.rs

@@ -744,6 +744,15 @@ fn a_reset_stream_keeps_its_header_without_credit() {
 /// it as fatal, so this one is hand-rolled and checks the stream directly.
 #[test]
 fn a_dropped_session_delivers_its_close() {
+	dropped_session_delivers_close(true);
+}
+
+#[test]
+fn the_last_web_session_drop_delivers_a_close_capsule() {
+	dropped_session_delivers_close(false);
+}
+
+fn dropped_session_delivers_close(explicit: bool) {
 	use bytes::Buf as _;
 
 	let Some(mut worker) = worker() else { return };
@@ -787,7 +796,9 @@ fn a_dropped_session_delivers_its_close() {
 			let request = within(&handle, "the CONNECT", quic::web::Request::accept(conn))
 				.await
 				.expect("handshake");
-			let mut session = request.ok().await.expect("respond");
+			let session = request.ok().await.expect("respond");
+			let mut remaining = session.clone();
+			drop(session);
 
 			let mut response = Vec::new();
 			let consumed = loop {
@@ -817,8 +828,13 @@ fn a_dropped_session_delivers_its_close() {
 				}
 			}
 
-			session.close(CLOSE_CODE, CLOSE_REASON);
-			drop(session);
+			let (code, reason) = if explicit {
+				remaining.close(CLOSE_CODE, CLOSE_REASON);
+				(CLOSE_CODE, CLOSE_REASON)
+			} else {
+				(0, "dropped")
+			};
+			drop(remaining);
 
 			let capsule = within(&handle, "the close capsule", async {
 				loop {
@@ -836,8 +852,8 @@ fn a_dropped_session_delivers_its_close() {
 			assert_eq!(
 				capsule,
 				web_transport_proto::Capsule::CloseWebTransportSession {
-					code: CLOSE_CODE,
-					reason: CLOSE_REASON.to_string(),
+					code,
+					reason: reason.to_string(),
 				}
 			);