Harden stream queues and hostile network tests

Cargo.lock

@@ -436,7 +436,7 @@ dependencies = [
 
 [[package]]
 name = "dosh"
-version = "1.0.0-rc20"
+version = "1.0.0-rc24"
 dependencies = [
  "anyhow",
  "base64",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "dosh"
-version = "1.0.0-rc20"
+version = "1.0.0-rc24"
 edition = "2024"
 license = "MIT"
 

src/bin/dosh-client.rs

@@ -36,7 +36,10 @@ use dosh::transport::{
     DoshTransport, SessionEvent, SessionRole, SessionTransportConfig, TransportConfig,
     TransportEvent,
 };
-use dosh::udp::is_transient_udp_send_error;
+use dosh::udp::{
+    is_transient_udp_error, is_transient_udp_send_error, recv_udp_retrying_transient,
+    send_udp_retrying_transient,
+};
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
 use std::collections::BTreeMap;
@@ -3789,14 +3792,11 @@ async fn try_native_auth(
             hello: hello.clone(),
         })?,
     )?;
-    socket.send_to(&packet, addr).await?;
+    let auth_timeout = Duration::from_millis(config.native_auth_timeout_ms.max(1));
+    send_udp_retrying_transient(socket, &packet, addr, auth_timeout).await?;
 
     let mut buf = vec![0u8; 65535];
-    let (n, _) = tokio::time::timeout(
-        Duration::from_millis(config.native_auth_timeout_ms.max(1)),
-        socket.recv_from(&mut buf),
-    )
-    .await??;
+    let (n, _) = recv_udp_retrying_transient(socket, &mut buf, auth_timeout).await?;
     let packet = protocol::decode(&buf[..n])?;
     if packet.header.kind != PacketKind::NativeServerHello {
         if packet.header.kind == PacketKind::AttachReject {
@@ -3859,13 +3859,9 @@ async fn try_native_auth(
         CLIENT_TO_SERVER,
         &auth_body,
     )?;
-    socket.send_to(&packet, addr).await?;
+    send_udp_retrying_transient(socket, &packet, addr, auth_timeout).await?;
 
-    let (n, _) = tokio::time::timeout(
-        Duration::from_millis(config.native_auth_timeout_ms.max(1)),
-        socket.recv_from(&mut buf),
-    )
-    .await??;
+    let (n, _) = recv_udp_retrying_transient(socket, &mut buf, auth_timeout).await?;
     let packet = protocol::decode(&buf[..n])?;
     if packet.header.kind != PacketKind::NativeAuthOk {
         if packet.header.kind == PacketKind::AttachReject {
@@ -3947,14 +3943,11 @@ async fn try_native_auth_check(
             hello: hello.clone(),
         })?,
     )?;
-    socket.send_to(&packet, addr).await?;
+    let auth_timeout = Duration::from_millis(config.native_auth_timeout_ms.max(1));
+    send_udp_retrying_transient(socket, &packet, addr, auth_timeout).await?;
 
     let mut buf = vec![0u8; 65535];
-    let (n, _) = tokio::time::timeout(
-        Duration::from_millis(config.native_auth_timeout_ms.max(1)),
-        socket.recv_from(&mut buf),
-    )
-    .await??;
+    let (n, _) = recv_udp_retrying_transient(socket, &mut buf, auth_timeout).await?;
     let packet = protocol::decode(&buf[..n])?;
     if packet.header.kind != PacketKind::NativeServerHello {
         if packet.header.kind == PacketKind::AttachReject {
@@ -4016,13 +4009,9 @@ async fn try_native_auth_check(
         CLIENT_TO_SERVER,
         &protocol::to_body(&NativeUserAuthBody { auth })?,
     )?;
-    socket.send_to(&packet, addr).await?;
+    send_udp_retrying_transient(socket, &packet, addr, auth_timeout).await?;
 
-    let (n, _) = tokio::time::timeout(
-        Duration::from_millis(config.native_auth_timeout_ms.max(1)),
-        socket.recv_from(&mut buf),
-    )
-    .await??;
+    let (n, _) = recv_udp_retrying_transient(socket, &mut buf, auth_timeout).await?;
     let packet = protocol::decode(&buf[..n])?;
     if packet.header.kind != PacketKind::NativeAuthCheckOk {
         if packet.header.kind == PacketKind::AttachReject {
@@ -4182,7 +4171,7 @@ async fn bootstrap_attach(
     let body = protocol::to_body(&req)?;
     let packet =
         protocol::encode_plain(PacketKind::BootstrapAttachRequest, [0u8; 16], 1, 0, &body)?;
-    socket.send_to(&packet, addr).await?;
+    send_udp_retrying_transient(socket, &packet, addr, Duration::from_secs(5)).await?;
     let expected_key_id = protocol::session_key_id(&bootstrap.session_key);
     let ok = recv_response_until(socket, Duration::from_secs(5), |packet| {
         if packet.header.kind == PacketKind::AttachReject && packet.header.conn_id == [0u8; 16] {
@@ -4262,7 +4251,7 @@ async fn try_ticket_attach(
         0,
         &protocol::to_body(&envelope)?,
     )?;
-    socket.send_to(&packet, addr).await?;
+    send_udp_retrying_transient(socket, &packet, addr, Duration::from_millis(700)).await?;
 
     let ok = recv_response_until(socket, Duration::from_millis(700), |packet| {
         if packet.header.kind == PacketKind::AttachReject && packet.header.conn_id == [0u8; 16] {
@@ -4338,7 +4327,7 @@ async fn try_resume_fresh_process(
         CLIENT_TO_SERVER,
         &body,
     )?;
-    socket.send_to(&packet, addr).await?;
+    send_udp_retrying_transient(socket, &packet, addr, Duration::from_millis(700)).await?;
     let frame = recv_response_until(socket, Duration::from_millis(700), |packet| {
         if !is_resume_response_for_client(&packet, cached.client_id) {
             return Ok(None);
@@ -4393,7 +4382,7 @@ async fn try_live_resume(
         &body,
     )?;
     *send_seq += 1;
-    socket.send_to(&packet, addr).await?;
+    send_udp_retrying_transient(socket, &packet, addr, Duration::from_millis(700)).await?;
     let frame = recv_response_until(socket, Duration::from_millis(700), |packet| {
         if !is_resume_response_for_client(&packet, cached.client_id) {
             return Ok(None);
@@ -4440,9 +4429,9 @@ where
             return Err(anyhow!("timed out waiting for server response"));
         }
         let remaining = wait - elapsed;
-        let (n, _) = tokio::time::timeout(remaining, socket.recv_from(&mut buf))
+        let (n, _) = recv_udp_retrying_transient(socket, &mut buf, remaining)
             .await
-            .context("timed out waiting for server response")??;
+            .context("timed out waiting for server response")?;
         let Ok(packet) = protocol::decode(&buf[..n]) else {
             continue;
         };
@@ -4832,7 +4821,11 @@ async fn run_terminal(
                 }
             }
             recv = socket.recv_from(&mut recv_buf) => {
-                let (n, _) = recv?;
+                let (n, _) = match recv {
+                    Ok(value) => value,
+                    Err(err) if is_transient_udp_error(&err) => continue,
+                    Err(err) => return Err(err.into()),
+                };
                 let Ok(packet) = protocol::decode(&recv_buf[..n]) else {
                     continue;
                 };
@@ -6260,7 +6253,9 @@ async fn flush_stream_pending_data(
         if credit < bytes.len() {
             break;
         }
-        let bytes = pending.pop_front().expect("pending front exists");
+        let Some(bytes) = pending.pop_front() else {
+            break;
+        };
         *stream_send_credit.entry(stream_id).or_default() -= bytes.len();
         let offset = *stream_next_send_offset.entry(stream_id).or_default();
         stream_next_send_offset.insert(stream_id, offset.saturating_add(bytes.len() as u64));
@@ -7770,8 +7765,8 @@ mod tests {
         retransmit_stream_opens, rewrite_forward_command, selected_predict_mode, selected_udp_host,
         server_version_mismatch, should_flush_terminal_input_after_contact,
         should_hold_during_startup_gate, should_hold_post_submit_input,
-        should_repaint_idle_alternate_screen, split_after_command_submit, ssh_destination_host,
-        ssh_username, ssh_with_user, ssh_command_target, startup_command, status_ssh_target,
+        should_repaint_idle_alternate_screen, split_after_command_submit, ssh_command_target,
+        ssh_destination_host, ssh_username, ssh_with_user, startup_command, status_ssh_target,
         strip_stale_mouse_reports, strip_terminal_focus_reports, terminal_private_mode_transition,
         toml_bare_key_or_quoted, unix_update_script, update_installer_url, update_version_status,
         upsert_managed_block, valid_forward_host, vscode_safe_alias, windows_update_script,

src/bin/dosh-server.rs

@@ -28,7 +28,7 @@ use dosh::protocol::{
     TicketAttachBody, TicketAttachEnvelope, TicketAttachOkEnvelope,
 };
 use dosh::pty::{PtyHandle, PtyOutput, adopt_pty_from_fd, spawn_pty_session};
-use dosh::udp::is_transient_udp_send_error;
+use dosh::udp::{is_transient_udp_error, is_transient_udp_send_error};
 use sha2::{Digest, Sha256};
 use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
 use std::fs;
@@ -262,7 +262,11 @@ async fn serve(config_path: Option<std::path::PathBuf>) -> Result<()> {
 
     let mut buf = vec![0u8; 65535];
     loop {
-        let (n, peer) = socket.recv_from(&mut buf).await?;
+        let (n, peer) = match socket.recv_from(&mut buf).await {
+            Ok(value) => value,
+            Err(err) if is_transient_udp_error(&err) => continue,
+            Err(err) => return Err(err.into()),
+        };
         if let Err(err) = handle_packet(&state, &socket, peer, &buf[..n]).await {
             eprintln!("packet from {peer}: {err:#}");
         }
@@ -357,6 +361,7 @@ impl NativeAuthRateLimiter {
 struct Session {
     pty: Option<PtyHandle>,
     parser: vt100::Parser,
+    restored_screen: Option<RestoredScreenSnapshot>,
     clients: HashMap<[u8; 16], ClientState>,
     output_seq: u64,
     recent: VecDeque<Vec<u8>>,
@@ -383,6 +388,14 @@ struct Session {
     last_screen_persist_at: Instant,
 }
 
+#[derive(Clone)]
+struct RestoredScreenSnapshot {
+    cols: u16,
+    rows: u16,
+    output_seq: u64,
+    snapshot: Vec<u8>,
+}
+
 #[derive(Clone)]
 struct ClientState {
     endpoint: SocketAddr,
@@ -508,6 +521,7 @@ impl ServerState {
             Session {
                 pty,
                 parser: vt100::Parser::new(rows.max(1), cols.max(1), self.config.scrollback),
+                restored_screen: None,
                 clients: HashMap::new(),
                 output_seq: 0,
                 recent: VecDeque::with_capacity(self.config.scrollback),
@@ -637,15 +651,23 @@ impl ServerState {
                 .unwrap_or((80, 24));
             let mut parser = vt100::Parser::new(rows, cols, self.config.scrollback);
             let mut output_seq = 0;
+            let mut restored_screen = None;
             if let Some(saved) = saved {
                 parser.process(&saved.snapshot);
                 output_seq = saved.output_seq;
+                restored_screen = Some(RestoredScreenSnapshot {
+                    cols,
+                    rows,
+                    output_seq,
+                    snapshot: saved.snapshot,
+                });
             }
             self.sessions.insert(
                 name.clone(),
                 Session {
                     pty: Some(pty),
                     parser,
+                    restored_screen,
                     clients: HashMap::new(),
                     output_seq,
                     recent: VecDeque::with_capacity(self.config.scrollback),
@@ -758,6 +780,19 @@ fn apply_terminal_size(
     Ok(())
 }
 
+fn attach_snapshot(session: &Session, cols: u16, rows: u16) -> Vec<u8> {
+    let cols = cols.max(1);
+    let rows = rows.max(1);
+    if let Some(restored) = &session.restored_screen
+        && restored.output_seq == session.output_seq
+        && restored.cols == cols
+        && restored.rows == rows
+    {
+        return restored.snapshot.clone();
+    }
+    screen_snapshot(session.parser.screen())
+}
+
 fn mode_allows_terminal_updates(mode: &str) -> bool {
     mode != "view-only" && mode != "forward-only"
 }
@@ -1185,7 +1220,7 @@ async fn handle_native_user_auth(
                 apply_terminal_size(session.pty.as_ref(), &mut session.parser, cols, rows)?;
             }
             let client_id = crypto::random_16();
-            let snapshot = screen_snapshot(session.parser.screen());
+            let snapshot = attach_snapshot(session, cols, rows);
             let output_seq = session.output_seq;
             locked.insert_client(
                 &session_name,
@@ -1333,7 +1368,7 @@ async fn handle_bootstrap_attach(
                 )?;
             }
             let client_id = crypto::random_16();
-            let snapshot = screen_snapshot(session.parser.screen());
+            let snapshot = attach_snapshot(session, req.cols, req.rows);
             let output_seq = session.output_seq;
             let bootstrap_session = req.bootstrap.session.clone();
             locked.insert_client(
@@ -1469,7 +1504,7 @@ async fn handle_ticket_attach(
                 )?;
             }
             let client_id = crypto::random_16();
-            let snapshot = screen_snapshot(session.parser.screen());
+            let snapshot = attach_snapshot(session, req.cols, req.rows);
             let output_seq = session.output_seq;
             let ticket_session = req.session.clone();
             locked.insert_client(
@@ -1582,20 +1617,23 @@ async fn handle_resume(
             .sessions
             .get_mut(&req.session)
             .ok_or_else(|| anyhow!("unknown session"))?;
-        let client = session
-            .clients
-            .get_mut(&packet.header.conn_id)
-            .ok_or_else(|| anyhow!("unknown client"))?;
-        if !client.replay.accept(packet.header.seq) {
-            return Ok(());
-        }
-        client.endpoint = peer;
-        client.last_acked = req.last_rendered_seq;
-        client.cols = req.cols;
-        client.rows = req.rows;
-        client.last_seen = Instant::now();
-        client.send_seq += 1;
-        if mode_allows_terminal_updates(&client.mode) {
+        let (send_seq, allow_terminal_updates) = {
+            let client = session
+                .clients
+                .get_mut(&packet.header.conn_id)
+                .ok_or_else(|| anyhow!("unknown client"))?;
+            if !client.replay.accept(packet.header.seq) {
+                return Ok(());
+            }
+            client.endpoint = peer;
+            client.last_acked = req.last_rendered_seq;
+            client.cols = req.cols;
+            client.rows = req.rows;
+            client.last_seen = Instant::now();
+            client.send_seq += 1;
+            (client.send_seq, mode_allows_terminal_updates(&client.mode))
+        };
+        if allow_terminal_updates {
             apply_terminal_size(
                 session.pty.as_ref(),
                 &mut session.parser,
@@ -1603,8 +1641,8 @@ async fn handle_resume(
                 req.rows,
             )?;
         }
-        let snapshot = screen_snapshot(session.parser.screen());
-        (client.send_seq, session.output_seq, snapshot)
+        let snapshot = attach_snapshot(session, req.cols, req.rows);
+        (send_seq, session.output_seq, snapshot)
     };
     let frame = Frame {
         session: session_name,
@@ -2714,10 +2752,13 @@ async fn handle_file_request(
                 atomic,
             });
             if let Some(mode) = mode {
-                let target = upload
+                let Some(active_upload) = upload.as_ref() else {
+                    return Err(anyhow!("upload state missing after start"));
+                };
+                let target = active_upload
+                    .temp_path
                     .as_ref()
-                    .and_then(|state| state.temp_path.as_ref())
-                    .unwrap_or_else(|| &upload.as_ref().unwrap().final_path)
+                    .unwrap_or(&active_upload.final_path)
                     .clone();
                 set_mode(&target, mode)?;
             }
@@ -3205,7 +3246,10 @@ async fn flush_stream_pending_data_to_client(
                 if credit < bytes.len() {
                     return Ok(());
                 }
-                let bytes = pending.pop_front().expect("pending front exists");
+                let Some(bytes) = pending.pop_front() else {
+                    client.stream_pending_data.remove(&stream_id);
+                    return Ok(());
+                };
                 if pending.is_empty() {
                     client.stream_pending_data.remove(&stream_id);
                 }
@@ -3450,6 +3494,7 @@ async fn broadcast_output(
             .get_mut(&output.session)
             .ok_or_else(|| anyhow!("unknown session"))?;
         session.parser.process(&output.bytes);
+        session.restored_screen = None;
         session.output_seq += 1;
         let output_seq = session.output_seq;
         session.recent.push_back(output.bytes.clone());
@@ -4045,6 +4090,34 @@ mod tests {
         assert_eq!(ServerConfig::default().client_timeout_secs, 2_592_000);
     }
 
+    #[test]
+    fn attach_snapshot_prefers_restored_screen_for_same_size_reattach() {
+        let restored = b"\x1b[?1049lRESTORED_SCREEN_MARKER".to_vec();
+        let mut session = Session {
+            pty: None,
+            parser: vt100::Parser::new(24, 80, 100),
+            restored_screen: Some(RestoredScreenSnapshot {
+                cols: 80,
+                rows: 24,
+                output_seq: 7,
+                snapshot: restored.clone(),
+            }),
+            clients: HashMap::new(),
+            output_seq: 7,
+            recent: VecDeque::new(),
+            empty_since: Some(Instant::now()),
+            holder_control: None,
+            persistent: true,
+            bytes_since_persist: 0,
+            last_persisted_seq: 7,
+            last_screen_persist_at: Instant::now(),
+        };
+
+        session.parser.set_size(24, 80);
+        assert_eq!(attach_snapshot(&session, 80, 24), restored);
+        assert_ne!(attach_snapshot(&session, 100, 30), restored);
+    }
+
     #[tokio::test]
     async fn unknown_resume_reject_keeps_client_id() {
         let (pty_tx, _rx) = mpsc::unbounded_channel();
@@ -4395,6 +4468,7 @@ mod tests {
             Session {
                 pty: None,
                 parser: vt100::Parser::new(24, 80, 100),
+                restored_screen: None,
                 clients: HashMap::from([(client_id, client)]),
                 output_seq: 0,
                 recent: VecDeque::new(),
@@ -4459,6 +4533,7 @@ mod tests {
             Session {
                 pty: None,
                 parser: vt100::Parser::new(24, 80, 100),
+                restored_screen: None,
                 clients: HashMap::from([(client_id, client)]),
                 output_seq: 0,
                 recent: VecDeque::new(),
@@ -4535,6 +4610,7 @@ mod tests {
             Session {
                 pty: None,
                 parser: vt100::Parser::new(24, 80, 100),
+                restored_screen: None,
                 clients: HashMap::from([(client_id, client)]),
                 output_seq: 0,
                 recent: VecDeque::new(),
@@ -4675,6 +4751,7 @@ mod tests {
             Session {
                 pty: None,
                 parser: vt100::Parser::new(24, 80, 100),
+                restored_screen: None,
                 clients: HashMap::from([(
                     client_id,
                     ClientState {
@@ -4772,6 +4849,7 @@ mod tests {
             Session {
                 pty: None,
                 parser: vt100::Parser::new(24, 80, 100),
+                restored_screen: None,
                 clients: HashMap::from([(
                     client_id,
                     ClientState {

src/client.rs

@@ -13,6 +13,7 @@ use crate::protocol::{
 };
 use crate::ssh_agent;
 use crate::transport::{DoshTransport, SessionRole, SessionTransportConfig, TransportConfig};
+use crate::udp::{recv_udp_retrying_transient, send_udp_retrying_transient};
 use anyhow::{Context, Result, anyhow, bail};
 use std::net::{SocketAddr, ToSocketAddrs};
 use std::path::PathBuf;
@@ -214,10 +215,10 @@ impl DoshClientBuilder {
                 hello: hello.clone(),
             })?,
         )?;
-        socket.send_to(&packet, peer_addr).await?;
+        send_udp_retrying_transient(&socket, &packet, peer_addr, timeout).await?;
 
         let mut buf = vec![0u8; 65535];
-        let (n, _) = tokio::time::timeout(timeout, socket.recv_from(&mut buf)).await??;
+        let (n, _) = recv_udp_retrying_transient(&socket, &mut buf, timeout).await?;
         let packet = protocol::decode(&buf[..n])?;
         if packet.header.kind != PacketKind::NativeServerHello {
             if packet.header.kind == PacketKind::AttachReject {
@@ -261,8 +262,8 @@ impl DoshClientBuilder {
             CLIENT_TO_SERVER,
             &protocol::to_body(&NativeUserAuthBody { auth })?,
         )?;
-        socket.send_to(&auth_packet, peer_addr).await?;
-        let (n, _) = tokio::time::timeout(timeout, socket.recv_from(&mut buf)).await??;
+        send_udp_retrying_transient(&socket, &auth_packet, peer_addr, timeout).await?;
+        let (n, _) = recv_udp_retrying_transient(&socket, &mut buf, timeout).await?;
         let packet = protocol::decode(&buf[..n])?;
         if packet.header.kind != PacketKind::NativeAuthOk {
             if packet.header.kind == PacketKind::AttachReject {

src/persist.rs

@@ -23,12 +23,14 @@ use std::io::{Read, Write};
 use std::os::fd::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
 use std::os::unix::net::{UnixListener, UnixStream};
 use std::path::{Path, PathBuf};
+use std::sync::Mutex;
 use std::time::Duration;
 
 /// One-byte commands a server sends to a holder over its control socket after
 /// the holder has handed back the master fd.
 const HOLDER_CMD_SHUTDOWN: u8 = b'X';
 const HOLDER_STARTUP_TIMEOUT: Duration = Duration::from_millis(750);
+static SCREEN_WRITE_LOCK: Mutex<()> = Mutex::new(());
 
 /// Magic written into a holder's `meta` file, bumped if the on-disk layout
 /// changes so a stale holder from an incompatible build is ignored.
@@ -140,13 +142,28 @@ pub fn save_screen(
         // No holder runtime for this session (non-persistent): nothing to do.
         return Ok(());
     }
+    let _guard = SCREEN_WRITE_LOCK
+        .lock()
+        .map_err(|_| anyhow!("screen persistence lock poisoned"))?;
+    let path = screen_path(&dir);
+    if existing_screen_output_seq(&path).is_some_and(|existing| existing >= output_seq) {
+        return Ok(());
+    }
     let mut buf = Vec::with_capacity(snapshot.len() + 32);
     buf.extend_from_slice(&cols.to_be_bytes());
     buf.extend_from_slice(&rows.to_be_bytes());
     buf.extend_from_slice(&output_seq.to_be_bytes());
     buf.extend_from_slice(&(snapshot.len() as u32).to_be_bytes());
     buf.extend_from_slice(snapshot);
-    atomic_write(&screen_path(&dir), &buf)
+    atomic_write(&path, &buf)
+}
+
+fn existing_screen_output_seq(path: &Path) -> Option<u64> {
+    let data = std::fs::read(path).ok()?;
+    if data.len() < 12 {
+        return None;
+    }
+    Some(u64::from_be_bytes(data[4..12].try_into().ok()?))
 }
 
 /// A restored screen for a re-adopted session.
@@ -619,6 +636,20 @@ mod tests {
         assert_eq!(loaded.snapshot, snap);
     }
 
+    #[test]
+    fn save_screen_never_overwrites_newer_snapshot_with_older_seq() {
+        let tmp = tempfile::tempdir().unwrap();
+        let sessions_dir = tmp.path();
+        ensure_runtime_dir(sessions_dir, "work").unwrap();
+
+        save_screen(sessions_dir, "work", 80, 24, 10, b"new-screen").unwrap();
+        save_screen(sessions_dir, "work", 80, 24, 9, b"old-screen").unwrap();
+
+        let loaded = load_screen(sessions_dir, "work").expect("screen restored");
+        assert_eq!(loaded.output_seq, 10);
+        assert_eq!(loaded.snapshot, b"new-screen");
+    }
+
     #[test]
     fn load_screen_absent_is_none() {
         let tmp = tempfile::tempdir().unwrap();

src/server.rs

@@ -13,7 +13,7 @@ use crate::transport::{
     DoshTransport, SessionEvent, SessionRole, SessionTransportConfig, TransportConfig,
     service_name_from_target,
 };
-use crate::udp::is_transient_udp_send_error;
+use crate::udp::{is_transient_udp_error, is_transient_udp_send_error};
 use anyhow::{Context, Result, anyhow, bail};
 use ed25519_dalek::SigningKey;
 use std::collections::{HashMap, HashSet};
@@ -169,7 +169,11 @@ impl DoshServer {
         let mut buf = vec![0u8; 65535];
         loop {
             self.expire_pending();
-            let (n, peer) = self.socket.recv_from(&mut buf).await?;
+            let (n, peer) = match self.socket.recv_from(&mut buf).await {
+                Ok(value) => value,
+                Err(err) if is_transient_udp_error(&err) => continue,
+                Err(err) => return Err(err.into()),
+            };
             let packet = match protocol::decode(&buf[..n]) {
                 Ok(packet) => packet,
                 Err(_) => continue,

src/transport.rs

@@ -19,7 +19,7 @@ use crate::protocol::{
     self, CLIENT_TO_SERVER, PacketKind, ReplayWindow, SERVER_TO_CLIENT, StreamClose, StreamData,
     StreamOpen, StreamOpenOk, StreamOpenReject, StreamWindowAdjust,
 };
-use crate::udp::is_transient_udp_send_error;
+use crate::udp::{is_transient_udp_error, is_transient_udp_send_error};
 use anyhow::{Result, anyhow, bail};
 use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
 use std::net::SocketAddr;
@@ -596,11 +596,13 @@ impl StreamMux {
             if self.send_credit.get(&stream_id).copied().unwrap_or(0) < front_len {
                 break;
             }
-            let bytes = self
+            let Some(bytes) = self
                 .pending_data
                 .get_mut(&stream_id)
                 .and_then(VecDeque::pop_front)
-                .expect("pending front exists");
+            else {
+                break;
+            };
             if self
                 .pending_data
                 .get(&stream_id)
@@ -852,7 +854,11 @@ impl DoshTransport {
     pub async fn recv(&mut self) -> Result<SessionEvent> {
         let mut buf = vec![0u8; 65535];
         loop {
-            let (n, peer) = self.socket.recv_from(&mut buf).await?;
+            let (n, peer) = match self.socket.recv_from(&mut buf).await {
+                Ok(value) => value,
+                Err(err) if transport_recv_error_is_retryable(&err) => continue,
+                Err(err) => return Err(err.into()),
+            };
             let packet = match protocol::decode(&buf[..n]) {
                 Ok(packet) => packet,
                 Err(_) => continue,
@@ -984,6 +990,10 @@ impl DoshTransport {
     }
 }
 
+fn transport_recv_error_is_retryable(err: &std::io::Error) -> bool {
+    is_transient_udp_error(err)
+}
+
 fn initial_stream_id(role: SessionRole) -> u64 {
     match role {
         SessionRole::Client => 1,
@@ -1033,6 +1043,35 @@ mod tests {
         assert!(service_target("../bad").is_err());
     }
 
+    #[test]
+    fn transport_recv_retries_network_churn_errors() {
+        for kind in [
+            std::io::ErrorKind::Interrupted,
+            std::io::ErrorKind::TimedOut,
+            std::io::ErrorKind::WouldBlock,
+        ] {
+            let err = std::io::Error::from(kind);
+            assert!(transport_recv_error_is_retryable(&err));
+        }
+        assert!(!transport_recv_error_is_retryable(&std::io::Error::from(
+            std::io::ErrorKind::PermissionDenied,
+        )));
+
+        #[cfg(unix)]
+        for code in [libc::ENETUNREACH, libc::EHOSTUNREACH, libc::ECONNRESET] {
+            assert!(transport_recv_error_is_retryable(
+                &std::io::Error::from_raw_os_error(code)
+            ));
+        }
+
+        #[cfg(windows)]
+        for code in [10051, 10054, 10065] {
+            assert!(transport_recv_error_is_retryable(
+                &std::io::Error::from_raw_os_error(code)
+            ));
+        }
+    }
+
     #[test]
     fn service_registry_matches_registered_service_opens() {
         let mut registry = ServiceRegistry::new();

src/udp.rs

@@ -1,6 +1,10 @@
 //! UDP error classification shared by Dosh clients, servers, and embedders.
 
-pub fn is_transient_udp_send_error(err: &std::io::Error) -> bool {
+use std::net::SocketAddr;
+use std::time::{Duration, Instant};
+use tokio::net::UdpSocket;
+
+pub fn is_transient_udp_error(err: &std::io::Error) -> bool {
     matches!(
         err.kind(),
         std::io::ErrorKind::Interrupted
@@ -9,6 +13,58 @@ pub fn is_transient_udp_send_error(err: &std::io::Error) -> bool {
     ) || err.raw_os_error().is_some_and(is_transient_udp_os_error)
 }
 
+pub fn is_transient_udp_send_error(err: &std::io::Error) -> bool {
+    is_transient_udp_error(err)
+}
+
+pub async fn send_udp_retrying_transient(
+    socket: &UdpSocket,
+    packet: &[u8],
+    peer: SocketAddr,
+    wait: Duration,
+) -> std::io::Result<()> {
+    let deadline = Instant::now() + wait.max(Duration::from_millis(1));
+    loop {
+        match socket.send_to(packet, peer).await {
+            Ok(_) => return Ok(()),
+            Err(err) if is_transient_udp_send_error(&err) && Instant::now() < deadline => {
+                sleep_until_retry(deadline).await;
+            }
+            Err(err) => return Err(err),
+        }
+    }
+}
+
+pub async fn recv_udp_retrying_transient(
+    socket: &UdpSocket,
+    buf: &mut [u8],
+    wait: Duration,
+) -> std::io::Result<(usize, SocketAddr)> {
+    let deadline = Instant::now() + wait.max(Duration::from_millis(1));
+    loop {
+        let now = Instant::now();
+        if now >= deadline {
+            return Err(std::io::Error::from(std::io::ErrorKind::TimedOut));
+        }
+        match tokio::time::timeout(deadline - now, socket.recv_from(buf)).await {
+            Ok(Ok(value)) => return Ok(value),
+            Ok(Err(err)) if is_transient_udp_error(&err) && Instant::now() < deadline => {
+                sleep_until_retry(deadline).await;
+            }
+            Ok(Err(err)) => return Err(err),
+            Err(_) => return Err(std::io::Error::from(std::io::ErrorKind::TimedOut)),
+        }
+    }
+}
+
+async fn sleep_until_retry(deadline: Instant) {
+    let now = Instant::now();
+    if now >= deadline {
+        return;
+    }
+    tokio::time::sleep((deadline - now).min(Duration::from_millis(20))).await;
+}
+
 #[cfg(unix)]
 fn is_transient_udp_os_error(code: i32) -> bool {
     matches!(
@@ -48,20 +104,56 @@ fn is_transient_udp_os_error(_code: i32) -> bool {
 
 #[cfg(test)]
 mod tests {
-    use super::is_transient_udp_send_error;
+    use super::{
+        is_transient_udp_error, is_transient_udp_send_error, recv_udp_retrying_transient,
+        send_udp_retrying_transient,
+    };
+    use std::time::Duration;
+    use tokio::net::UdpSocket;
 
     #[test]
-    fn classifies_portable_transient_send_errors() {
+    fn classifies_portable_transient_udp_errors() {
         for kind in [
             std::io::ErrorKind::Interrupted,
             std::io::ErrorKind::TimedOut,
             std::io::ErrorKind::WouldBlock,
         ] {
-            assert!(is_transient_udp_send_error(&std::io::Error::from(kind)));
+            let err = std::io::Error::from(kind);
+            assert!(is_transient_udp_error(&err));
+            assert!(is_transient_udp_send_error(&err));
         }
-        assert!(!is_transient_udp_send_error(&std::io::Error::from(
-            std::io::ErrorKind::PermissionDenied,
-        )));
+        let fatal = std::io::Error::from(std::io::ErrorKind::PermissionDenied);
+        assert!(!is_transient_udp_error(&fatal));
+        assert!(!is_transient_udp_send_error(&fatal));
+    }
+
+    #[tokio::test]
+    async fn retrying_udp_send_and_receive_round_trip() {
+        let receiver = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let sender = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let addr = receiver.local_addr().unwrap();
+
+        send_udp_retrying_transient(&sender, b"hello", addr, Duration::from_millis(50))
+            .await
+            .unwrap();
+        let mut buf = [0u8; 16];
+        let (n, peer) = recv_udp_retrying_transient(&receiver, &mut buf, Duration::from_millis(50))
+            .await
+            .unwrap();
+
+        assert_eq!(&buf[..n], b"hello");
+        assert_eq!(peer, sender.local_addr().unwrap());
+    }
+
+    #[tokio::test]
+    async fn retrying_udp_receive_times_out_cleanly() {
+        let receiver = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let mut buf = [0u8; 16];
+        let err = recv_udp_retrying_transient(&receiver, &mut buf, Duration::from_millis(1))
+            .await
+            .unwrap_err();
+
+        assert_eq!(err.kind(), std::io::ErrorKind::TimedOut);
     }
 
     #[cfg(unix)]
@@ -78,9 +170,9 @@ mod tests {
             libc::ENETUNREACH,
             libc::ETIMEDOUT,
         ] {
-            assert!(is_transient_udp_send_error(
-                &std::io::Error::from_raw_os_error(code)
-            ));
+            let err = std::io::Error::from_raw_os_error(code);
+            assert!(is_transient_udp_error(&err));
+            assert!(is_transient_udp_send_error(&err));
         }
     }
 
@@ -90,9 +182,9 @@ mod tests {
         for code in [
             10049, 10050, 10051, 10052, 10054, 10060, 10061, 10064, 10065,
         ] {
-            assert!(is_transient_udp_send_error(
-                &std::io::Error::from_raw_os_error(code)
-            ));
+            let err = std::io::Error::from_raw_os_error(code);
+            assert!(is_transient_udp_error(&err));
+            assert!(is_transient_udp_send_error(&err));
         }
     }
 }

tests/hostile_network.rs

@@ -29,7 +29,7 @@ use std::io::{Read, Write};
 use std::net::{SocketAddr, TcpListener, UdpSocket};
 use std::process::{Child, Command, Stdio};
 use std::sync::Arc;
-use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering};
 use std::sync::mpsc::{Receiver, Sender, channel};
 use std::thread;
 use std::time::{Duration, Instant};
@@ -49,9 +49,20 @@ use ed25519_dalek::{SigningKey, VerifyingKey};
 use rand::rngs::StdRng;
 use rand::{Rng, SeedableRng};
 
+const TEST_UDP_PORT_START: u16 = 31_000;
+const TEST_UDP_PORT_END: u16 = 60_999;
+static NEXT_UDP_PORT: AtomicU32 = AtomicU32::new(TEST_UDP_PORT_START as u32);
+
 fn free_udp_port() -> u16 {
-    let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
-    socket.local_addr().unwrap().port()
+    let span = u32::from(TEST_UDP_PORT_END - TEST_UDP_PORT_START) + 1;
+    for _ in TEST_UDP_PORT_START..=TEST_UDP_PORT_END {
+        let offset = NEXT_UDP_PORT.fetch_add(1, Ordering::SeqCst) % span;
+        let port = TEST_UDP_PORT_START + offset as u16;
+        if UdpSocket::bind(("127.0.0.1", port)).is_ok() {
+            return port;
+        }
+    }
+    panic!("no free UDP test port in {TEST_UDP_PORT_START}-{TEST_UDP_PORT_END}");
 }
 
 fn write_server_config(dir: &tempfile::TempDir, port: u16) -> std::path::PathBuf {

tests/integration_smoke.rs

@@ -6,6 +6,7 @@ use std::net::{TcpListener, TcpStream, UdpSocket};
 use std::os::unix::net::{UnixListener, UnixStream};
 use std::path::{Path, PathBuf};
 use std::process::{Child, Command, Stdio};
+use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::mpsc;
 use std::thread;
 use std::time::Duration;
@@ -23,14 +24,35 @@ use dosh::protocol::{
 };
 use ed25519_dalek::{Signer, SigningKey, VerifyingKey};
 
+const TEST_PORT_START: u16 = 30_000;
+const TEST_PORT_END: u16 = 60_999;
+static NEXT_UDP_PORT: AtomicU32 = AtomicU32::new(TEST_PORT_START as u32);
+static NEXT_TCP_PORT: AtomicU32 = AtomicU32::new(TEST_PORT_START as u32);
+
+fn next_test_port(counter: &AtomicU32) -> u16 {
+    let span = u32::from(TEST_PORT_END - TEST_PORT_START) + 1;
+    let offset = counter.fetch_add(1, Ordering::SeqCst) % span;
+    TEST_PORT_START + offset as u16
+}
+
 fn free_udp_port() -> u16 {
-    let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
-    socket.local_addr().unwrap().port()
+    for _ in TEST_PORT_START..=TEST_PORT_END {
+        let port = next_test_port(&NEXT_UDP_PORT);
+        if UdpSocket::bind(("127.0.0.1", port)).is_ok() {
+            return port;
+        }
+    }
+    panic!("no free UDP test port in {TEST_PORT_START}-{TEST_PORT_END}");
 }
 
 fn free_tcp_port() -> u16 {
-    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
-    listener.local_addr().unwrap().port()
+    for _ in TEST_PORT_START..=TEST_PORT_END {
+        let port = next_test_port(&NEXT_TCP_PORT);
+        if TcpListener::bind(("127.0.0.1", port)).is_ok() {
+            return port;
+        }
+    }
+    panic!("no free TCP test port in {TEST_PORT_START}-{TEST_PORT_END}");
 }
 
 fn write_server_config(dir: &tempfile::TempDir, port: u16) -> std::path::PathBuf {