fix(uring): resolve a dial only once its final handshake flight is staged

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

@@ -50,6 +50,9 @@ impl Config {
 
 /// Dial [`Config::peer`] over `socket`, driving the handshake to completion.
 ///
+/// Resolves once the handshake's final flight is staged on the socket, so the
+/// worker may stop right after without stranding the peer.
+///
 /// Shorthand for a dial-only [`Endpoint`](super::Endpoint) and one
 /// [`connect`](super::Endpoint::connect) through it. The connection's driver
 /// runs as a task on the worker that adopted `socket`, so the returned

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

@@ -39,6 +39,12 @@ pub(crate) struct State {
 	/// fresh egress reaches the wire.
 	driver: kio::WaiterList,
 
+	/// noq reported the handshake complete, which it does before this side's
+	/// final flight is sent.
+	connected: bool,
+	/// The handshake is complete and its final flight staged: congestion
+	/// control or pacing can hold part of that flight back, and a peer that
+	/// never receives it never finishes its handshake.
 	established: bool,
 	establish_waiters: kio::WaiterList,
 
@@ -90,6 +96,7 @@ impl State {
 	fn new() -> Self {
 		Self {
 			driver: kio::WaiterList::new(),
+			connected: false,
 			established: false,
 			establish_waiters: kio::WaiterList::new(),
 			accept_bi_waiters: kio::WaiterList::new(),
@@ -383,6 +390,10 @@ pub(crate) async fn establish(shared: Shared) -> Result<Connection, Error> {
 		if let Some(err) = &state.closed {
 			return Poll::Ready(Err(err.clone()));
 		}
+		// Nothing will ever stage the rest of the handshake.
+		if shared.owner.handle().is_none() {
+			return Poll::Ready(Err(Error::Io(crate::shared::Shared::gone_error().to_string())));
+		}
 		waiter.register(&mut state.establish_waiters);
 		Poll::Pending
 	})
@@ -680,6 +691,7 @@ impl Driver {
 				self.shared.state.borrow_mut().fail(err);
 				return Poll::Ready(());
 			}
+			self.publish_established();
 			self.publish_close();
 
 			// Arm, *then* poll: the poll is what registers the waiter, so
@@ -725,10 +737,7 @@ impl Driver {
 
 			let mut state = self.shared.state.borrow_mut();
 			match event {
-				moq_noq_proto::Event::Connected => {
-					state.established = true;
-					state.establish_waiters.wake();
-				}
+				moq_noq_proto::Event::Connected => state.connected = true,
 				moq_noq_proto::Event::ConnectionLost { reason } => state.fail(reason.into()),
 				moq_noq_proto::Event::DatagramReceived => state.datagram_recv_waiters.wake(),
 				moq_noq_proto::Event::DatagramsUnblocked => state.datagram_send_waiters.wake(),
@@ -741,6 +750,18 @@ impl Driver {
 		}
 	}
 
+	/// Hand the connection over once the handshake is complete and the flush
+	/// above has staged all of its final flight, so a caller that stops the
+	/// worker the moment the handshake resolves strands no peer.
+	fn publish_established(&mut self) {
+		let mut state = self.shared.state.borrow_mut();
+		if !state.connected || state.established || self.shared.conn.borrow().has_pending_handshake_data() {
+			return;
+		}
+		state.established = true;
+		state.establish_waiters.wake();
+	}
+
 	/// Publish the terminal error for a close this side asked for.
 	///
 	/// noq raises no event for it, so the driver is what reports it, and

rs/moq-uring/src/quic/noq/endpoint.rs

@@ -152,6 +152,10 @@ impl Endpoint {
 
 	/// Dial [`Config::peer`](crate::quic::client::Config::peer) through this
 	/// endpoint's socket, driving the handshake to completion.
+	///
+	/// Resolves once the handshake's final flight is staged on the socket, so
+	/// the worker may stop right after without stranding the peer. A worker
+	/// that stops before then fails the dial.
 	pub async fn connect(&self, config: &crate::quic::client::Config) -> Result<Connection, Error> {
 		if let Some(err) = self.inner.closed() {
 			return Err(err);

rs/moq-uring/tests/handshake.rs

@@ -0,0 +1,176 @@
+//! A dial resolves only once its final handshake flight is staged, so a
+//! client that stops its worker the moment the dial resolves strands no peer.
+//!
+//! Kernel-gated: skips loudly below the Linux 6.12 floor (GitHub-hosted CI),
+//! and runs everywhere else.
+
+#![cfg(all(target_os = "linux", feature = "noq"))]
+
+#[path = "support.rs"]
+mod support;
+
+use std::net::{SocketAddr, UdpSocket};
+use std::task::{Context, Poll, Waker};
+use std::time::Duration;
+
+use moq_uring::{Config, Error, Worker, quic, udp};
+
+const ALPN: &str = "moq-uring-handshake";
+
+/// One-shot dials per test, each on a worker dropped as soon as it resolves.
+const DIALS: usize = 20;
+
+fn worker() -> Option<Worker> {
+	match Worker::new(Config::default()) {
+		Ok(worker) => Some(worker),
+		Err(Error::Unsupported(reason)) => {
+			eprintln!("skipping io_uring handshake test: {reason}");
+			None
+		}
+		Err(err) => panic!("worker setup failed: {err}"),
+	}
+}
+
+fn socket(handle: &moq_uring::Handle) -> udp::Socket {
+	handle
+		.udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default())
+		.expect("socket")
+}
+
+/// A client certificate, and the root file the server verifies it against.
+struct Client {
+	identity: quic::Identity,
+	root: std::path::PathBuf,
+	_dir: tempfile::TempDir,
+}
+
+/// A self-signed client certificate of about 20 KB: more than the initial
+/// congestion window, so congestion control splits the final flight across
+/// round trips, and pacing splits it again on a slow link.
+fn big_client() -> Client {
+	let names: Vec<String> = (0..600).map(|i| format!("client-{i:04}.padding.example.com")).collect();
+	let signed = rcgen::generate_simple_self_signed(names).expect("client certificate");
+	let dir = tempfile::tempdir().expect("tempdir");
+	let root = dir.path().join("client.pem");
+	std::fs::write(&root, signed.cert.pem()).expect("write root");
+	Client {
+		identity: quic::Identity::from_pem(signed.cert.pem(), signed.signing_key.serialize_pem()),
+		root,
+		_dir: dir,
+	}
+}
+
+/// Serve on a thread of its own, counting the handshakes that complete until
+/// `quiet` passes without one.
+fn serve(
+	congestion: quic::Congestion,
+	client_root: Option<std::path::PathBuf>,
+	quiet: Duration,
+) -> (SocketAddr, std::thread::JoinHandle<usize>) {
+	let certs = support::certs().expect("certificates");
+	let identity = quic::Identity::open(&certs.cert, &certs.key).expect("identity");
+	let sock = UdpSocket::bind("127.0.0.1:0").expect("bind server");
+	let addr = sock.local_addr().expect("server addr");
+	let thread = std::thread::spawn(move || {
+		let mut worker = Worker::new(Config::default()).expect("server worker");
+		let handle = worker.handle();
+		let sock = handle.udp(sock, udp::Config::default()).expect("server socket");
+		let mut config = quic::server::Config::new(identity);
+		config.alpn = vec![ALPN.to_string()];
+		config.transport.congestion = congestion;
+		if let Some(root) = client_root {
+			config.client_auth = quic::server::ClientAuth::Required(vec![root]);
+		}
+		let endpoint =
+			quic::Endpoint::new(sock, quic::endpoint::Config::default().with_server(config)).expect("endpoint");
+		worker
+			.block_on(async move {
+				let mut accepted = Vec::new();
+				loop {
+					let mut deadline = moq_uring::Timer::after(&handle, quiet);
+					let mut accept = std::pin::pin!(endpoint.accept());
+					let conn = kio::wait(|waiter| {
+						let mut cx = Context::from_waker(waiter.waker());
+						if let Poll::Ready(conn) = accept.as_mut().poll(&mut cx) {
+							return Poll::Ready(Some(conn.expect("accept")));
+						}
+						deadline.poll(waiter).map(|()| None)
+					})
+					.await;
+					match conn {
+						Some(conn) => accepted.push(conn),
+						None => return accepted.len(),
+					}
+				}
+			})
+			.expect("server worker")
+	});
+	(addr, thread)
+}
+
+/// The one-shot client shape: dial, then drop the worker the moment the dial
+/// resolves. The server must still finish every handshake, rather than time
+/// out on a client that believes it is connected.
+fn stop_after_dial(congestion: quic::Congestion, client: Option<Client>) {
+	if worker().is_none() {
+		return;
+	}
+	let root = client.as_ref().map(|client| client.root.clone());
+	let (addr, server) = serve(congestion, root, Duration::from_secs(2));
+
+	let mut config = quic::client::Config::new(addr, "localhost");
+	config.alpn = vec![ALPN.to_string()];
+	config.verify = false;
+	config.transport.congestion = congestion;
+	config.identity = client.as_ref().map(|client| client.identity.clone());
+
+	for _ in 0..DIALS {
+		let mut worker = worker().expect("client worker");
+		let sock = socket(&worker.handle());
+		worker
+			.block_on(quic::client::connect(sock, &config))
+			.expect("client worker")
+			.expect("dial");
+		drop(worker);
+	}
+
+	let accepted = server.join().expect("server thread");
+	assert_eq!(accepted, DIALS, "handshakes the server finished");
+}
+
+#[test]
+fn a_stopped_dial_finishes_the_server_handshake() {
+	stop_after_dial(quic::Congestion::Loss, None);
+	stop_after_dial(quic::Congestion::Delay, None);
+}
+
+/// A final flight too big to leave in one burst: the dial waits for all of
+/// it, not just the part congestion control and pacing let out first.
+#[test]
+fn a_stopped_dial_sends_all_of_a_split_final_flight() {
+	stop_after_dial(quic::Congestion::Loss, Some(big_client()));
+	stop_after_dial(quic::Congestion::Delay, Some(big_client()));
+}
+
+/// A worker dropped mid-handshake fails the dial rather than leaving it
+/// pending on a flight nothing will ever send.
+#[test]
+fn a_worker_stopped_mid_dial_fails_it() {
+	let Some(worker) = worker() else { return };
+	// A peer that never answers, so the handshake cannot finish first.
+	let silent = UdpSocket::bind("127.0.0.1:0").expect("bind peer");
+	let mut config = quic::client::Config::new(silent.local_addr().expect("peer addr"), "localhost");
+	config.alpn = vec![ALPN.to_string()];
+	config.verify = false;
+
+	let endpoint = quic::Endpoint::new(socket(&worker.handle()), quic::endpoint::Config::default()).expect("endpoint");
+	let mut dial = std::pin::pin!(endpoint.connect(&config));
+	let mut cx = Context::from_waker(Waker::noop());
+	assert!(dial.as_mut().poll(&mut cx).is_pending(), "the dial is under way");
+
+	drop(worker);
+	match dial.as_mut().poll(&mut cx) {
+		Poll::Ready(Err(quic::Error::Io(_))) => {}
+		other => panic!("a dial whose worker stopped should fail, got {other:?}"),
+	}
+}