From a9fd70de520f9d215477a5b90d59902f3a35ecbf Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 24 Sep 2026 10:31:30 -0700 Subject: [PATCH 1/2] quest: claim uring-handshake-flush Co-Authored-By: Claude Opus 5.5 From 7daa9b17ac4e8cea59eff8121aa1536910925064 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 24 Sep 2026 10:57:24 -0700 Subject: [PATCH 2/2] fix(uring): resolve a dial only once its final handshake flight is staged noq reports the handshake complete before the client's last flight has left: congestion control and pacing can hold part of it back, so a worker stopped the moment the dial resolved stranded the server. The driver now hands the connection over only once noq has no handshake data left to send, and a dial whose worker stopped fails instead of pending forever. Needs Connection::has_pending_handshake_data from moq-noq-proto, which is not released yet. Co-Authored-By: Claude Opus 5.5 --- rs/moq-uring/src/quic/client.rs | 3 + rs/moq-uring/src/quic/noq/connection.rs | 29 +++- rs/moq-uring/src/quic/noq/endpoint.rs | 4 + rs/moq-uring/tests/handshake.rs | 176 ++++++++++++++++++++++++ 4 files changed, 208 insertions(+), 4 deletions(-) create mode 100644 rs/moq-uring/tests/handshake.rs diff --git a/rs/moq-uring/src/quic/client.rs b/rs/moq-uring/src/quic/client.rs index 29ca017170..e870067494 100644 --- a/rs/moq-uring/src/quic/client.rs +++ b/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 diff --git a/rs/moq-uring/src/quic/noq/connection.rs b/rs/moq-uring/src/quic/noq/connection.rs index e5e0acf0c3..4159609b46 100644 --- a/rs/moq-uring/src/quic/noq/connection.rs +++ b/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 { 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 diff --git a/rs/moq-uring/src/quic/noq/endpoint.rs b/rs/moq-uring/src/quic/noq/endpoint.rs index 9d2edf818b..744f741046 100644 --- a/rs/moq-uring/src/quic/noq/endpoint.rs +++ b/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 { if let Some(err) = self.inner.closed() { return Err(err); diff --git a/rs/moq-uring/tests/handshake.rs b/rs/moq-uring/tests/handshake.rs new file mode 100644 index 0000000000..bfd1456b10 --- /dev/null +++ b/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 { + 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 = (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, + quiet: Duration, +) -> (SocketAddr, std::thread::JoinHandle) { + 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) { + 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:?}"), + } +}