From b0cee259240e28e8ca14865b3711d38678435267 Mon Sep 17 00:00:00 2001 From: robinhundt <24554122+robinhundt@users.noreply.github.com> Date: Mon, 20 Apr 2026 16:10:44 +0200 Subject: [PATCH 1/3] fix(net): potential deadlock --- .github/workflows/pull_request.yml | 6 +- cryprot-net/src/lib.rs | 106 ++++++++++++++++++----------- 2 files changed, 70 insertions(+), 42 deletions(-) diff --git a/.github/workflows/pull_request.yml b/.github/workflows/pull_request.yml index 9f3decc..2fb7bf3 100644 --- a/.github/workflows/pull_request.yml +++ b/.github/workflows/pull_request.yml @@ -24,6 +24,8 @@ jobs: rustflags: "" env: RUSTFLAGS: ${{ matrix.rustflags }} + RUST_BACKTRACE: 1 + RUST_LOG: debug steps: - &checkout name: Check out repository @@ -45,10 +47,10 @@ jobs: uses: ilammy/setup-nasm@72793074d3c8cdda771dba85f6deafe00623038b # v1.5.2 - name: Run tests (all features) - run: cargo test --workspace --verbose --all-features --no-fail-fast ${{ runner.os == 'macOS' && '-- --test-threads=1' || '' }} + run: cargo test --workspace --verbose --all-features --no-fail-fast - name: Run tests (no features) - run: cargo test --workspace --verbose --no-fail-fast ${{ runner.os == 'macOS' && '-- --test-threads=1' || '' }} + run: cargo test --workspace --verbose --no-fail-fast miri: name: Miri diff --git a/cryprot-net/src/lib.rs b/cryprot-net/src/lib.rs index 21455c5..9677d3e 100644 --- a/cryprot-net/src/lib.rs +++ b/cryprot-net/src/lib.rs @@ -64,6 +64,11 @@ pub struct StreamManager { acceptor: QuicStreamAcceptor, cmd_send: mpsc::UnboundedSender, cmd_recv: mpsc::UnboundedReceiver, + maps: StreamMaps, +} + +#[derive(Default)] +struct StreamMaps { pending: HashMap, accepted: HashMap, } @@ -146,8 +151,7 @@ impl StreamManager { acceptor, cmd_send, cmd_recv, - pending: Default::default(), - accepted: Default::default(), + maps: Default::default(), } } @@ -167,47 +171,32 @@ impl StreamManager { Self::accepted(stream, self.cmd_send.clone()); } Ok(None) => { - debug!("remote closed"); - return; + debug!("remote closed, draining pending commands"); + break; } Err(err) => { - error!(%err, "unable to accept stream"); - return; + error!(%err, "unable to accept stream, draining pending commands"); + break; } } } Some(cmd) = self.cmd_recv.recv() => { // recv() is cancel safe debug!(?cmd, "received cmd"); - match cmd { - Cmd::NewStream {uid, stream_return} => { - if let Some(accepted) = self.accepted.remove(&uid) { - if stream_return.send(accepted).is_err() { - debug!("accepted remote stream but local receiver is closed"); - } - debug!("sending new stream to receiver"); - continue; - } - match self.pending.entry(uid) { - Entry::Occupied(occupied_entry) => { - panic!("Duplicate unique id: {:?}", occupied_entry.key()) - }, - Entry::Vacant(vacant_entry) => {vacant_entry.insert(stream_return);}, - } - } - Cmd::AcceptedStream {uid, stream, bytes_read} => { - if let Some(stream_ret) = self.pending.remove(&uid) { - if stream_ret.send((stream, bytes_read)).is_err() { - debug!("accepted remote stream but local receiver is closed"); - } - } else { - debug!("accepted stream but no pending"); - self.accepted.insert(uid, (stream, bytes_read)); - } - } - } + self.maps.handle_cmd(cmd); } } } + // The QUIC acceptor is done (remote closed or error), but there may be + // in-flight `Self::accepted` tasks that already received a stream and + // are reading the UniqueId. Drain remaining commands so those streams + // are matched with pending requests. + while let Ok(cmd) = self.cmd_recv.try_recv() { + debug!(?cmd, "received cmd (draining)"); + self.maps.handle_cmd(cmd); + if self.maps.pending.is_empty() { + break; + } + } } // not taking &self to work around borrow issue @@ -220,17 +209,54 @@ impl StreamManager { return; } }; - cmd_send - .send(Cmd::AcceptedStream { - uid, - stream, - bytes_read, - }) - .expect("cmd_rcv is owned by StreamManager") + // StreamManager may have already exited if the connection closed + let _ = cmd_send.send(Cmd::AcceptedStream { + uid, + stream, + bytes_read, + }); }); } } +impl StreamMaps { + fn handle_cmd(&mut self, cmd: Cmd) { + match cmd { + Cmd::NewStream { uid, stream_return } => { + if let Some(accepted) = self.accepted.remove(&uid) { + if stream_return.send(accepted).is_err() { + debug!("accepted remote stream but local receiver is closed"); + } + debug!("sending new stream to receiver"); + return; + } + match self.pending.entry(uid) { + Entry::Occupied(occupied_entry) => { + panic!("Duplicate unique id: {:?}", occupied_entry.key()) + } + Entry::Vacant(vacant_entry) => { + vacant_entry.insert(stream_return); + } + } + } + Cmd::AcceptedStream { + uid, + stream, + bytes_read, + } => { + if let Some(stream_ret) = self.pending.remove(&uid) { + if stream_ret.send((stream, bytes_read)).is_err() { + debug!("accepted remote stream but local receiver is closed"); + } + } else { + debug!("accepted stream but no pending"); + self.accepted.insert(uid, (stream, bytes_read)); + } + } + } + } +} + /// Possible connection errors. #[derive(thiserror::Error, Debug)] pub enum ConnectionError { From a5c5782e1142bec51aaa31cf3692ea6f50a6180c Mon Sep 17 00:00:00 2001 From: robinhundt <24554122+robinhundt@users.noreply.github.com> Date: Sun, 21 Jun 2026 20:51:07 +0200 Subject: [PATCH 2/3] fix(net): finally fix deadlock? --- cryprot-net/src/lib.rs | 21 +++++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/cryprot-net/src/lib.rs b/cryprot-net/src/lib.rs index 9677d3e..6501c92 100644 --- a/cryprot-net/src/lib.rs +++ b/cryprot-net/src/lib.rs @@ -188,14 +188,23 @@ impl StreamManager { } // The QUIC acceptor is done (remote closed or error), but there may be // in-flight `Self::accepted` tasks that already received a stream and - // are reading the UniqueId. Drain remaining commands so those streams - // are matched with pending requests. - while let Ok(cmd) = self.cmd_recv.try_recv() { + // are still `await`ing on reading the UniqueId before they send their + // `Cmd::AcceptedStream`. We must keep handling commands until those + // tasks complete, otherwise we'd drop the paired `pending` request and + // the peer's stream would error out. + // + // We no longer accept new streams, so drop our own command sender. The + // remaining senders are held by live `Connection`s and the in-flight + // `accepted` tasks; once all of those are gone, `recv()` returns `None` + // and we stop. Using `recv().await` rather than `try_recv()` is crucial: + // an `accepted` task whose UniqueId read has not finished yet would be + // missed by `try_recv` (the channel is transiently empty), dropping the + // matching `pending` oneshot. This race is timing dependent and was + // observed on macOS/NixOS but not on the Linux CI runner. + drop(self.cmd_send); + while let Some(cmd) = self.cmd_recv.recv().await { debug!(?cmd, "received cmd (draining)"); self.maps.handle_cmd(cmd); - if self.maps.pending.is_empty() { - break; - } } } From a529b2bbb551811c13045b0e1b4b14f188ce1ac6 Mon Sep 17 00:00:00 2001 From: robinhundt <24554122+robinhundt@users.noreply.github.com> Date: Wed, 24 Jun 2026 21:35:58 +0000 Subject: [PATCH 3/3] net: fix timeout? --- cryprot-net/src/lib.rs | 13 ++++++++++++- cryprot-net/src/testing.rs | 17 ++++++++++++++++- 2 files changed, 28 insertions(+), 2 deletions(-) diff --git a/cryprot-net/src/lib.rs b/cryprot-net/src/lib.rs index 6501c92..d19f187 100644 --- a/cryprot-net/src/lib.rs +++ b/cryprot-net/src/lib.rs @@ -285,7 +285,18 @@ pub enum ConnectionError { impl Connection { pub fn new(quic_conn: s2n_quic::Connection) -> (Self, StreamManager) { - let (handle, acceptor) = quic_conn.split(); + let (mut handle, acceptor) = quic_conn.split(); + // Keep the connection alive even when no application data is exchanged. + // MPC protocols routinely interleave network communication with long, + // purely-local compute phases (e.g. AES tree expansion offloaded to + // rayon). During such a phase no QUIC packets flow and, under CPU + // contention, the gap can exceed the negotiated idle timeout, causing + // the connection to be torn down with `IdleTimerExpired`. Enabling + // keep-alive makes s2n-quic emit periodic PINGs (see + // `with_max_keep_alive_period`) so the idle timer is reset while we + // compute. `keep_alive` only fails if the connection is already closed, + // in which case the manager/handle will surface the error anyway. + let _ = handle.keep_alive(true); let stream_manager = StreamManager::new(acceptor); let conn = Self { cids: vec![], diff --git a/cryprot-net/src/testing.rs b/cryprot-net/src/testing.rs index eeb2a11..9f5de48 100644 --- a/cryprot-net/src/testing.rs +++ b/cryprot-net/src/testing.rs @@ -23,7 +23,22 @@ pub async fn local_conn() -> anyhow::Result<(Connection, Connection)> { let limits = Limits::new() .with_max_send_buffer_size(12 * MiB as u32)? .with_max_open_local_unidirectional_streams(max_streams)? - .with_max_open_remote_unidirectional_streams(max_streams)?; + .with_max_open_remote_unidirectional_streams(max_streams)? + // Both endpoints of a test connection live on the same current-thread + // tokio runtime as the protocol future. When the test suite runs in + // parallel, the protocols offload heavy AES/PPRF work to a shared global + // rayon pool which saturates all cores. The runtime thread driving the + // QUIC IO can then be descheduled for seconds at a time, during which no + // packets flow. With the default 30s idle timeout this intermittently + // tripped `IdleTimerExpired`, tearing down a perfectly healthy loopback + // connection mid-protocol (only observed under parallel execution, hence + // the previous macOS `--test-threads=1` CI workaround). A generous idle + // timeout tolerates these scheduling gaps, and keep-alive (enabled in + // `Connection::new`) refreshes the timer whenever the runtime is idle + // between compute phases. The keep-alive period must stay below the idle + // timeout to have any effect. + .with_max_idle_timeout(std::time::Duration::from_secs(300))? + .with_max_keep_alive_period(std::time::Duration::from_secs(15))?; let addr = "127.0.0.1:0".parse()?; let io = || {