Expand SSH identity path tokens

Cargo.lock

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

Cargo.toml

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

Makefile

@@ -36,7 +36,9 @@ persistence-check:
 package-check:
 	$(MAKE) package-release-linux
 	$(MAKE) package-release-windows
-	sh scripts/verify-release-artifacts.sh
+	sh scripts/verify-release-artifacts.sh \
+		target/dosh-release/dosh-linux-x86_64.tar.gz \
+		target/dosh-release/dosh-windows-x86_64.zip
 
 install:
 	sh install.sh --from-current

README.md

@@ -44,6 +44,7 @@ dosh setup HOST
 dosh HOST
 dosh HOST COMMAND
 dosh update
+dosh update --client|--server|--both
 ```
 
 Useful commands:
@@ -97,6 +98,20 @@ cache_attach_tickets = true
 disconnect_status = true
 ```
 
+Example host entry:
+
+```toml
+[prod]
+ssh = "prod"
+ssh_config = "prod" # OpenSSH alias for IdentityFile, ProxyJump, SendEnv, etc.
+dosh_host = "prod.example.com"
+port = 50000
+```
+
+`ProxyJump` and `ProxyCommand` can be used for SSH setup, but the Dosh UDP
+endpoint still has to be reachable directly from the client. Set `dosh_host` to
+the reachable UDP hostname or IP when it differs from the SSH alias target.
+
 ## Rust Library
 
 Dosh exposes a Rust transport for encrypted, reconnecting application streams.

build.rs

@@ -1,3 +1,4 @@
+use std::env;
 use std::process::Command;
 
 fn git(args: &[&str]) -> Option<String> {
@@ -12,14 +13,29 @@ fn git(args: &[&str]) -> Option<String> {
 fn main() {
     println!("cargo:rerun-if-changed=.git/HEAD");
     println!("cargo:rerun-if-changed=.git/index");
+    println!("cargo:rerun-if-env-changed=DOSH_BUILD_GIT_HASH");
+    println!("cargo:rerun-if-env-changed=DOSH_BUILD_COMMIT_DATE");
+    println!("cargo:rerun-if-env-changed=DOSH_BUILD_GIT_DIRTY");
 
-    let hash = git(&["rev-parse", "--short=12", "HEAD"]).unwrap_or_else(|| "unknown".into());
+    let hash = env::var("DOSH_BUILD_GIT_HASH")
-    let date = git(&["show", "-s", "--format=%cs", "HEAD"]).unwrap_or_else(|| "unknown".into());
+        .ok()
-    let dirty = Command::new("git")
+        .filter(|value| !value.is_empty())
-        .args(["diff", "--quiet", "--ignore-submodules", "--"])
+        .or_else(|| git(&["rev-parse", "--short=12", "HEAD"]))
-        .status()
+        .unwrap_or_else(|| "unknown".into());
-        .map(|status| if status.success() { "" } else { "+dirty" })
+    let date = env::var("DOSH_BUILD_COMMIT_DATE")
-        .unwrap_or("");
+        .ok()
+        .filter(|value| !value.is_empty())
+        .or_else(|| git(&["show", "-s", "--format=%cs", "HEAD"]))
+        .unwrap_or_else(|| "unknown".into());
+    let dirty = match env::var("DOSH_BUILD_GIT_DIRTY").ok().as_deref() {
+        Some("1" | "true" | "yes" | "+dirty") => "+dirty",
+        Some(_) => "",
+        None => Command::new("git")
+            .args(["diff", "--quiet", "--ignore-submodules", "--"])
+            .status()
+            .map(|status| if status.success() { "" } else { "+dirty" })
+            .unwrap_or(""),
+    };
 
     println!("cargo:rustc-env=DOSH_GIT_HASH={hash}{dirty}");
     println!("cargo:rustc-env=DOSH_COMMIT_DATE={date}");

scripts/package-release.sh

@@ -52,6 +52,22 @@ else
   release_dir="target/release"
 fi
 
+if command -v git >/dev/null 2>&1 && git rev-parse --is-inside-work-tree >/dev/null 2>&1; then
+  if [ -z "${DOSH_BUILD_GIT_HASH:-}" ]; then
+    export DOSH_BUILD_GIT_HASH="$(git rev-parse --short=12 HEAD)"
+  fi
+  if [ -z "${DOSH_BUILD_COMMIT_DATE:-}" ]; then
+    export DOSH_BUILD_COMMIT_DATE="$(git show -s --format=%cs HEAD)"
+  fi
+  if [ -z "${DOSH_BUILD_GIT_DIRTY+x}" ]; then
+    if git diff --quiet --ignore-submodules --; then
+      export DOSH_BUILD_GIT_DIRTY=0
+    else
+      export DOSH_BUILD_GIT_DIRTY=1
+    fi
+  fi
+fi
+
 if [ "$os" = "windows" ]; then
   if [ -n "$target" ]; then
     cargo build --release --target "$target" --bin dosh-client --bin dosh-bench

scripts/verify-release-artifacts.sh

@@ -22,11 +22,14 @@ artifact_version() {
   esac
 }
 
+if [ "$#" -gt 0 ]; then
+  artifacts="$*"
+else
+  artifacts="$out_dir/dosh-linux-x86_64.tar.gz $out_dir/dosh-macos-aarch64.tar.gz $out_dir/dosh-windows-x86_64.zip"
+fi
+
 failed=0
-for artifact in \
+for artifact in $artifacts; do
-  "$out_dir/dosh-linux-x86_64.tar.gz" \
-  "$out_dir/dosh-macos-aarch64.tar.gz" \
-  "$out_dir/dosh-windows-x86_64.zip"; do
   if [ ! -f "$artifact" ]; then
     echo "missing release artifact: $artifact" >&2
     failed=1

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?;
+        send_udp_retrying_transient(&socket, &auth_packet, peer_addr, timeout).await?;
-        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::NativeAuthOk {
             if packet.header.kind == PacketKind::AttachReject {

src/config.rs

@@ -150,6 +150,12 @@ pub struct ClientConfig {
 #[derive(Debug, Clone, Default, Serialize, Deserialize)]
 pub struct HostConfig {
     pub ssh: Option<String>,
+    /// Optional OpenSSH config host name to use for SSH-backed bootstrap,
+    /// trust/setup/doctor checks, and native-auth identity selection. This lets
+    /// a Dosh host keep a separate UDP endpoint while still inheriting
+    /// alias-scoped OpenSSH settings like IdentityFile, ProxyJump, and SendEnv.
+    #[serde(default)]
+    pub ssh_config: Option<String>,
     pub user: Option<String>,
     pub dosh_host: Option<String>,
     pub port: Option<u16>,

src/file_transfer.rs

@@ -18,6 +18,14 @@ pub enum FileRequest {
         path: String,
         mode: Option<u32>,
     },
+    Readlink {
+        path: String,
+    },
+    Symlink {
+        path: String,
+        target: String,
+        overwrite: bool,
+    },
     Remove {
         path: String,
         recursive: bool,
@@ -46,13 +54,35 @@ pub enum FileRequest {
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 pub enum FileResponse {
     Ok,
-    Resume { offset: u64 },
+    Resume {
-    Stat { meta: FileMeta },
+        offset: u64,
-    List { entries: Vec<FileEntry> },
+        prefix_sha256: Option<[u8; 32]>,
-    Start { meta: FileMeta, offset: u64 },
+    },
-    Chunk { offset: u64, bytes: Vec<u8> },
+    Stat {
-    Done { sha256: [u8; 32], bytes: u64 },
+        meta: FileMeta,
-    Error { message: String },
+    },
+    List {
+        entries: Vec<FileEntry>,
+    },
+    LinkTarget {
+        target: String,
+    },
+    Start {
+        meta: FileMeta,
+        offset: u64,
+        prefix_sha256: Option<[u8; 32]>,
+    },
+    Chunk {
+        offset: u64,
+        bytes: Vec<u8>,
+    },
+    Done {
+        sha256: [u8; 32],
+        bytes: u64,
+    },
+    Error {
+        message: String,
+    },
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]

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;
@@ -30,6 +30,7 @@ use tokio::net::UdpSocket;
 pub const DEFAULT_INITIAL_WINDOW: usize = 1024 * 1024;
 pub const DEFAULT_RETRANSMIT_AFTER: Duration = Duration::from_millis(200);
 pub const DEFAULT_KEEPALIVE_AFTER: Duration = Duration::from_secs(2);
+pub const DEFAULT_RETIRED_STREAM_TOMBSTONES: usize = 16 * 1024;
 pub const SERVICE_TARGET_PREFIX: &str = "@dosh-";
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -37,6 +38,7 @@ pub struct TransportConfig {
     pub initial_window: usize,
     pub retransmit_after: Duration,
     pub keepalive_after: Duration,
+    pub retired_stream_tombstones: usize,
 }
 
 impl Default for TransportConfig {
@@ -45,6 +47,7 @@ impl Default for TransportConfig {
             initial_window: DEFAULT_INITIAL_WINDOW,
             retransmit_after: DEFAULT_RETRANSMIT_AFTER,
             keepalive_after: DEFAULT_KEEPALIVE_AFTER,
+            retired_stream_tombstones: DEFAULT_RETIRED_STREAM_TOMBSTONES,
         }
     }
 }
@@ -121,6 +124,10 @@ pub struct AcceptedService {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum TransportEvent {
     Open(IncomingStreamOpen),
+    DuplicateOpen {
+        stream_id: u64,
+        response: OutgoingStreamPacket,
+    },
     OpenOk {
         stream_id: u64,
         flushed: Vec<OutgoingStreamPacket>,
@@ -137,6 +144,9 @@ pub enum TransportEvent {
     Close {
         stream_id: u64,
     },
+    Ignored {
+        stream_id: u64,
+    },
 }
 
 #[derive(Debug, Clone)]
@@ -166,6 +176,8 @@ pub struct StreamMux {
     next_send_offset: HashMap<u64, u64>,
     next_recv_offset: HashMap<u64, u64>,
     recv_pending: HashMap<u64, BTreeMap<u64, Vec<u8>>>,
+    retired_streams: HashSet<u64>,
+    retired_stream_order: VecDeque<u64>,
 }
 
 #[derive(Debug, Clone, Default)]
@@ -221,6 +233,8 @@ impl StreamMux {
             next_send_offset: HashMap::new(),
             next_recv_offset: HashMap::new(),
             recv_pending: HashMap::new(),
+            retired_streams: HashSet::new(),
+            retired_stream_order: VecDeque::new(),
         }
     }
 
@@ -231,7 +245,10 @@ impl StreamMux {
         target_port: u16,
     ) -> Result<OutgoingStreamPacket> {
         let target_host = target_host.into();
-        if self.opened_streams.contains(&stream_id) || self.pending_opens.contains_key(&stream_id) {
+        if self.opened_streams.contains(&stream_id)
+            || self.pending_opens.contains_key(&stream_id)
+            || self.retired_streams.contains(&stream_id)
+        {
             bail!("stream {stream_id} already exists");
         }
         self.send_credit
@@ -258,6 +275,15 @@ impl StreamMux {
     }
 
     pub fn accept_open(&mut self, open: StreamOpen) -> Result<IncomingStreamOpen> {
+        if self.retired_streams.contains(&open.stream_id) {
+            bail!("stream {} was already closed", open.stream_id);
+        }
+        if self.pending_opens.contains_key(&open.stream_id) {
+            bail!(
+                "stream {} collides with a local pending open",
+                open.stream_id
+            );
+        }
         self.opened_streams.insert(open.stream_id);
         self.send_credit
             .entry(open.stream_id)
@@ -285,19 +311,39 @@ impl StreamMux {
         )
     }
 
-    pub fn handle_open_ok(&mut self, ok: StreamOpenOk) -> Result<Vec<OutgoingStreamPacket>> {
+    pub fn reject_open(
-        self.pending_opens.remove(&ok.stream_id);
+        &mut self,
+        stream_id: u64,
+        reason: impl Into<String>,
+    ) -> Result<OutgoingStreamPacket> {
+        if self.has_stream_state(stream_id) {
+            self.retire_stream(stream_id);
+        }
+        Self::open_reject(stream_id, reason)
+    }
+
+    pub fn handle_open_ok(
+        &mut self,
+        ok: StreamOpenOk,
+    ) -> Result<Option<Vec<OutgoingStreamPacket>>> {
+        if self.pending_opens.remove(&ok.stream_id).is_none() {
+            return Ok(None);
+        }
         self.opened_streams.insert(ok.stream_id);
         self.send_credit
             .entry(ok.stream_id)
             .or_insert(self.config.initial_window);
         self.next_send_offset.entry(ok.stream_id).or_insert(0);
         self.next_recv_offset.entry(ok.stream_id).or_insert(0);
-        self.flush_pending_data(ok.stream_id)
+        self.flush_pending_data(ok.stream_id).map(Some)
     }
 
-    pub fn handle_open_reject(&mut self, reject: StreamOpenReject) {
+    pub fn handle_open_reject(&mut self, reject: StreamOpenReject) -> bool {
-        self.cleanup_stream(reject.stream_id);
+        if !self.has_stream_state(reject.stream_id) {
+            return false;
+        }
+        self.retire_stream(reject.stream_id);
+        true
     }
 
     pub fn send_data(
@@ -309,8 +355,17 @@ impl StreamMux {
         if bytes.is_empty() {
             return Ok(Vec::new());
         }
-        if !self.opened_streams.contains(&stream_id)
+        if !self.opened_streams.contains(&stream_id) {
-            || self.send_credit.get(&stream_id).copied().unwrap_or(0) < bytes.len()
+            if self.pending_opens.contains_key(&stream_id) {
+                self.pending_data
+                    .entry(stream_id)
+                    .or_default()
+                    .push_back(bytes);
+                return Ok(Vec::new());
+            }
+            bail!("stream {stream_id} is not open");
+        }
+        if self.send_credit.get(&stream_id).copied().unwrap_or(0) < bytes.len()
             || self
                 .pending_data
                 .get(&stream_id)
@@ -325,8 +380,11 @@ impl StreamMux {
         Ok(vec![self.send_data_now(stream_id, bytes)?])
     }
 
-    pub fn handle_data(&mut self, data: StreamData) -> Result<IncomingStreamData> {
+    pub fn handle_data(&mut self, data: StreamData) -> Result<Option<IncomingStreamData>> {
         let stream_id = data.stream_id;
+        if !self.opened_streams.contains(&stream_id) {
+            return Ok(None);
+        }
         let (chunks, consumed, received_offset) = self.accept_data(data);
         let window_adjust = encode_packet(
             PacketKind::StreamWindowAdjust,
@@ -336,68 +394,104 @@ impl StreamMux {
                 bytes: consumed.min(u32::MAX as usize) as u32,
             },
         )?;
-        Ok(IncomingStreamData {
+        Ok(Some(IncomingStreamData {
             stream_id,
             chunks,
             consumed,
             received_offset,
             window_adjust,
-        })
+        }))
     }
 
     pub fn handle_window_adjust(
         &mut self,
         adjust: StreamWindowAdjust,
-    ) -> Result<Vec<OutgoingStreamPacket>> {
+    ) -> Result<Option<Vec<OutgoingStreamPacket>>> {
+        if !self.opened_streams.contains(&adjust.stream_id) {
+            return Ok(None);
+        }
         let acked_bytes = self.ack_data(adjust.stream_id, adjust.received_offset);
         self.add_credit(adjust.stream_id, acked_bytes);
-        self.flush_pending_data(adjust.stream_id)
+        self.flush_pending_data(adjust.stream_id).map(Some)
     }
 
     pub fn close_stream(&mut self, stream_id: u64) -> Result<OutgoingStreamPacket> {
-        self.cleanup_stream(stream_id);
+        if !self.has_stream_state(stream_id) {
+            bail!("stream {stream_id} is not open");
+        }
+        self.retire_stream(stream_id);
         encode_packet(PacketKind::StreamClose, &StreamClose { stream_id })
     }
 
-    pub fn handle_close(&mut self, close: StreamClose) {
+    pub fn handle_close(&mut self, close: StreamClose) -> bool {
-        self.cleanup_stream(close.stream_id);
+        if !self.has_stream_state(close.stream_id) {
+            return false;
+        }
+        self.retire_stream(close.stream_id);
+        true
     }
 
     pub fn handle_packet(&mut self, kind: PacketKind, body: &[u8]) -> Result<TransportEvent> {
         match kind {
             PacketKind::StreamOpen => {
                 let open: StreamOpen = protocol::from_body(body)?;
+                if self.retired_streams.contains(&open.stream_id) {
+                    return Ok(TransportEvent::Ignored {
+                        stream_id: open.stream_id,
+                    });
+                }
+                if self.opened_streams.contains(&open.stream_id) {
+                    return Ok(TransportEvent::DuplicateOpen {
+                        stream_id: open.stream_id,
+                        response: Self::open_ok(open.stream_id)?,
+                    });
+                }
                 self.accept_open(open).map(TransportEvent::Open)
             }
             PacketKind::StreamOpenOk => {
                 let ok: StreamOpenOk = protocol::from_body(body)?;
                 let stream_id = ok.stream_id;
-                let flushed = self.handle_open_ok(ok)?;
+                match self.handle_open_ok(ok)? {
-                Ok(TransportEvent::OpenOk { stream_id, flushed })
+                    Some(flushed) => Ok(TransportEvent::OpenOk { stream_id, flushed }),
+                    None => Ok(TransportEvent::Ignored { stream_id }),
+                }
             }
             PacketKind::StreamOpenReject => {
                 let reject: StreamOpenReject = protocol::from_body(body)?;
-                self.handle_open_reject(reject.clone());
+                let stream_id = reject.stream_id;
-                Ok(TransportEvent::OpenReject {
+                if self.handle_open_reject(reject.clone()) {
-                    stream_id: reject.stream_id,
+                    Ok(TransportEvent::OpenReject {
-                    reason: reject.reason,
+                        stream_id,
-                })
+                        reason: reject.reason,
+                    })
+                } else {
+                    Ok(TransportEvent::Ignored { stream_id })
+                }
             }
             PacketKind::StreamData => {
                 let data: StreamData = protocol::from_body(body)?;
-                self.handle_data(data).map(TransportEvent::Data)
+                let stream_id = data.stream_id;
+                match self.handle_data(data)? {
+                    Some(data) => Ok(TransportEvent::Data(data)),
+                    None => Ok(TransportEvent::Ignored { stream_id }),
+                }
             }
             PacketKind::StreamWindowAdjust => {
                 let adjust: StreamWindowAdjust = protocol::from_body(body)?;
                 let stream_id = adjust.stream_id;
-                let flushed = self.handle_window_adjust(adjust)?;
+                match self.handle_window_adjust(adjust)? {
-                Ok(TransportEvent::WindowAdjust { stream_id, flushed })
+                    Some(flushed) => Ok(TransportEvent::WindowAdjust { stream_id, flushed }),
+                    None => Ok(TransportEvent::Ignored { stream_id }),
+                }
             }
             PacketKind::StreamClose => {
                 let close: StreamClose = protocol::from_body(body)?;
                 let stream_id = close.stream_id;
-                self.handle_close(close);
+                if self.handle_close(close) {
-                Ok(TransportEvent::Close { stream_id })
+                    Ok(TransportEvent::Close { stream_id })
+                } else {
+                    Ok(TransportEvent::Ignored { stream_id })
+                }
             }
             _ => bail!("packet kind {kind:?} is not a stream transport packet"),
         }
@@ -502,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)
@@ -585,6 +681,32 @@ impl StreamMux {
         *credit = credit.saturating_add(bytes).min(self.config.initial_window);
     }
 
+    fn has_stream_state(&self, stream_id: u64) -> bool {
+        self.opened_streams.contains(&stream_id)
+            || self.pending_opens.contains_key(&stream_id)
+            || self.send_credit.contains_key(&stream_id)
+            || self.pending_data.contains_key(&stream_id)
+            || self.sent_data.contains_key(&stream_id)
+            || self.next_send_offset.contains_key(&stream_id)
+            || self.next_recv_offset.contains_key(&stream_id)
+            || self.recv_pending.contains_key(&stream_id)
+    }
+
+    fn retire_stream(&mut self, stream_id: u64) {
+        self.cleanup_stream(stream_id);
+        if self.config.retired_stream_tombstones == 0 {
+            return;
+        }
+        if self.retired_streams.insert(stream_id) {
+            self.retired_stream_order.push_back(stream_id);
+        }
+        while self.retired_stream_order.len() > self.config.retired_stream_tombstones {
+            if let Some(expired) = self.retired_stream_order.pop_front() {
+                self.retired_streams.remove(&expired);
+            }
+        }
+    }
+
     fn cleanup_stream(&mut self, stream_id: u64) {
         self.opened_streams.remove(&stream_id);
         self.pending_opens.remove(&stream_id);
@@ -625,7 +747,7 @@ impl DoshTransport {
             ack_seq: config.initial_ack,
             replay: ReplayWindow::default(),
             mux: StreamMux::new(config.stream),
-            next_stream_id: 1,
+            next_stream_id: initial_stream_id(config.role),
             last_contact: now,
             last_keepalive: now,
         }
@@ -669,7 +791,10 @@ impl DoshTransport {
 
     pub fn allocate_stream_id(&mut self) -> u64 {
         let id = self.next_stream_id;
-        self.next_stream_id = self.next_stream_id.saturating_add(1).max(1);
+        self.next_stream_id = self
+            .next_stream_id
+            .wrapping_add(1)
+            .max(initial_stream_id(self.role));
         id
     }
 
@@ -699,8 +824,8 @@ impl DoshTransport {
     }
 
     pub async fn reject_stream(&mut self, stream_id: u64, reason: impl Into<String>) -> Result<()> {
-        self.send_outgoing(StreamMux::open_reject(stream_id, reason)?)
+        let packet = self.mux.reject_open(stream_id, reason)?;
-            .await
+        self.send_outgoing(packet).await
     }
 
     pub async fn send(&mut self, stream_id: u64, bytes: impl Into<Vec<u8>>) -> Result<()> {
@@ -729,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,
@@ -811,6 +940,9 @@ impl DoshTransport {
 
     async fn send_followups(&mut self, event: &TransportEvent) -> Result<()> {
         match event {
+            TransportEvent::DuplicateOpen { response, .. } => {
+                self.send_outgoing(response.clone()).await?;
+            }
             TransportEvent::OpenOk { flushed, .. }
             | TransportEvent::WindowAdjust { flushed, .. } => {
                 for packet in flushed.clone() {
@@ -822,7 +954,8 @@ impl DoshTransport {
             }
             TransportEvent::Open(_)
             | TransportEvent::OpenReject { .. }
-            | TransportEvent::Close { .. } => {}
+            | TransportEvent::Close { .. }
+            | TransportEvent::Ignored { .. } => {}
         }
         Ok(())
     }
@@ -857,6 +990,17 @@ 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,
+        SessionRole::Server => 1u64 << 63,
+    }
+}
+
 pub fn service_target(name: &str) -> Result<String> {
     if name.is_empty()
         || !name
@@ -899,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();
@@ -938,7 +1111,10 @@ mod tests {
         assert!(mux.send_data(7, b"hello".to_vec()).unwrap().is_empty());
         assert_eq!(mux.pending_bytes(7), 5);
 
-        let flushed = mux.handle_open_ok(StreamOpenOk { stream_id: 7 }).unwrap();
+        let flushed = mux
+            .handle_open_ok(StreamOpenOk { stream_id: 7 })
+            .unwrap()
+            .unwrap();
         assert_eq!(flushed.len(), 1);
         assert_eq!(flushed[0].kind, PacketKind::StreamData);
         let data: StreamData = decode(&flushed[0]);
@@ -961,6 +1137,7 @@ mod tests {
                 offset: 5,
                 bytes: b"world".to_vec(),
             })
+            .unwrap()
             .unwrap();
         assert!(first.chunks.is_empty());
         assert_eq!(first.received_offset, 0);
@@ -971,6 +1148,7 @@ mod tests {
                 offset: 0,
                 bytes: b"hello".to_vec(),
             })
+            .unwrap()
             .unwrap();
         assert_eq!(second.chunks, vec![b"hello".to_vec(), b"world".to_vec()]);
         assert_eq!(second.consumed, 10);
@@ -985,7 +1163,9 @@ mod tests {
             ..TransportConfig::default()
         });
         mux.open_stream(1, "@dosh-test", 0).unwrap();
-        mux.handle_open_ok(StreamOpenOk { stream_id: 1 }).unwrap();
+        mux.handle_open_ok(StreamOpenOk { stream_id: 1 })
+            .unwrap()
+            .unwrap();
         let sent = mux.send_data(1, b"hello".to_vec()).unwrap();
         assert_eq!(sent.len(), 1);
         assert!(mux.send_data(1, b"!".to_vec()).unwrap().is_empty());
@@ -996,6 +1176,7 @@ mod tests {
                 received_offset: 5,
                 bytes: 5,
             })
+            .unwrap()
             .unwrap();
         assert_eq!(flushed.len(), 1);
         let data: StreamData = decode(&flushed[0]);
@@ -1011,7 +1192,9 @@ mod tests {
             ..TransportConfig::default()
         });
         mux.open_stream(1, "@dosh-test", 0).unwrap();
-        mux.handle_open_ok(StreamOpenOk { stream_id: 1 }).unwrap();
+        mux.handle_open_ok(StreamOpenOk { stream_id: 1 })
+            .unwrap()
+            .unwrap();
         assert_eq!(mux.send_data(1, b"hello".to_vec()).unwrap().len(), 1);
         assert_eq!(mux.send_credit[&1], 0);
 
@@ -1021,6 +1204,7 @@ mod tests {
                 received_offset: 5,
                 bytes: 0,
             })
+            .unwrap()
             .unwrap();
         assert!(flushed.is_empty());
         assert_eq!(mux.send_credit[&1], 5);
@@ -1047,6 +1231,281 @@ mod tests {
         );
     }
 
+    #[test]
+    fn duplicate_stream_open_reacks_without_second_open_event() {
+        let mut mux = StreamMux::new(TransportConfig::default());
+        let open = StreamOpen {
+            stream_id: 9,
+            target_host: "@dosh-app".to_string(),
+            target_port: 0,
+        };
+        let body = protocol::to_body(&open).unwrap();
+        assert!(matches!(
+            mux.handle_packet(PacketKind::StreamOpen, &body).unwrap(),
+            TransportEvent::Open(_)
+        ));
+
+        let duplicate = mux.handle_packet(PacketKind::StreamOpen, &body).unwrap();
+        match duplicate {
+            TransportEvent::DuplicateOpen {
+                stream_id,
+                response,
+            } => {
+                assert_eq!(stream_id, 9);
+                assert_eq!(response.kind, PacketKind::StreamOpenOk);
+                let ok: StreamOpenOk = decode(&response);
+                assert_eq!(ok.stream_id, 9);
+            }
+            other => panic!("unexpected duplicate open event {other:?}"),
+        }
+    }
+
+    #[test]
+    fn send_data_to_closed_or_unknown_stream_fails() {
+        let mut mux = StreamMux::new(TransportConfig::default());
+        mux.open_stream(1, "@dosh-test", 0).unwrap();
+        assert!(mux.send_data(1, b"queued".to_vec()).unwrap().is_empty());
+        mux.handle_open_ok(StreamOpenOk { stream_id: 1 })
+            .unwrap()
+            .unwrap();
+        mux.close_stream(1).unwrap();
+
+        assert!(mux.send_data(1, b"lost".to_vec()).is_err());
+        assert!(mux.send_data(99, b"lost".to_vec()).is_err());
+    }
+
+    #[test]
+    fn rejecting_incoming_stream_retires_local_state() {
+        let mut mux = StreamMux::new(TransportConfig::default());
+        mux.accept_open(StreamOpen {
+            stream_id: 7,
+            target_host: "@dosh-test".to_string(),
+            target_port: 0,
+        })
+        .unwrap();
+        assert!(mux.is_open(7));
+
+        let reject = mux.reject_open(7, "nope").unwrap();
+
+        assert_eq!(reject.kind, PacketKind::StreamOpenReject);
+        assert!(!mux.is_open(7));
+        assert!(mux.retired_streams.contains(&7));
+        assert!(mux.send_data(7, b"must-not-send".to_vec()).is_err());
+        assert!(
+            mux.accept_open(StreamOpen {
+                stream_id: 7,
+                target_host: "@dosh-test".to_string(),
+                target_port: 0,
+            })
+            .is_err()
+        );
+    }
+
+    #[test]
+    fn stale_packets_after_close_do_not_recreate_stream_state() {
+        let mut mux = StreamMux::new(TransportConfig::default());
+        mux.open_stream(1, "@dosh-test", 0).unwrap();
+        mux.handle_open_ok(StreamOpenOk { stream_id: 1 })
+            .unwrap()
+            .unwrap();
+        assert_eq!(mux.send_data(1, b"hello".to_vec()).unwrap().len(), 1);
+        mux.close_stream(1).unwrap();
+
+        let stale_data = protocol::to_body(&StreamData {
+            stream_id: 1,
+            offset: 0,
+            bytes: b"stale".to_vec(),
+        })
+        .unwrap();
+        assert_eq!(
+            mux.handle_packet(PacketKind::StreamData, &stale_data)
+                .unwrap(),
+            TransportEvent::Ignored { stream_id: 1 }
+        );
+
+        let stale_open_ok = protocol::to_body(&StreamOpenOk { stream_id: 1 }).unwrap();
+        assert_eq!(
+            mux.handle_packet(PacketKind::StreamOpenOk, &stale_open_ok)
+                .unwrap(),
+            TransportEvent::Ignored { stream_id: 1 }
+        );
+
+        let stale_adjust = protocol::to_body(&StreamWindowAdjust {
+            stream_id: 1,
+            received_offset: 5,
+            bytes: 5,
+        })
+        .unwrap();
+        assert_eq!(
+            mux.handle_packet(PacketKind::StreamWindowAdjust, &stale_adjust)
+                .unwrap(),
+            TransportEvent::Ignored { stream_id: 1 }
+        );
+
+        let stale_open = protocol::to_body(&StreamOpen {
+            stream_id: 1,
+            target_host: "@dosh-test".to_string(),
+            target_port: 0,
+        })
+        .unwrap();
+        assert_eq!(
+            mux.handle_packet(PacketKind::StreamOpen, &stale_open)
+                .unwrap(),
+            TransportEvent::Ignored { stream_id: 1 }
+        );
+
+        let stale_close = protocol::to_body(&StreamClose { stream_id: 1 }).unwrap();
+        assert_eq!(
+            mux.handle_packet(PacketKind::StreamClose, &stale_close)
+                .unwrap(),
+            TransportEvent::Ignored { stream_id: 1 }
+        );
+
+        assert!(!mux.is_open(1));
+        assert!(!mux.pending_opens.contains_key(&1));
+        assert!(!mux.send_credit.contains_key(&1));
+        assert!(!mux.pending_data.contains_key(&1));
+        assert!(!mux.sent_data.contains_key(&1));
+        assert!(!mux.next_send_offset.contains_key(&1));
+        assert!(!mux.next_recv_offset.contains_key(&1));
+        assert!(!mux.recv_pending.contains_key(&1));
+        assert!(mux.retired_streams.contains(&1));
+    }
+
+    #[test]
+    fn retired_stream_tombstones_are_bounded() {
+        let mut mux = StreamMux::new(TransportConfig {
+            retired_stream_tombstones: 2,
+            ..TransportConfig::default()
+        });
+        for stream_id in 1..=3 {
+            mux.open_stream(stream_id, "@dosh-test", 0).unwrap();
+            mux.close_stream(stream_id).unwrap();
+        }
+
+        assert_eq!(mux.retired_streams.len(), 2);
+        assert!(!mux.retired_streams.contains(&1));
+        assert!(mux.retired_streams.contains(&2));
+        assert!(mux.retired_streams.contains(&3));
+
+        let stale_open = protocol::to_body(&StreamOpen {
+            stream_id: 3,
+            target_host: "@dosh-test".to_string(),
+            target_port: 0,
+        })
+        .unwrap();
+        assert_eq!(
+            mux.handle_packet(PacketKind::StreamOpen, &stale_open)
+                .unwrap(),
+            TransportEvent::Ignored { stream_id: 3 }
+        );
+    }
+
+    #[tokio::test]
+    async fn session_transport_partitions_client_and_server_stream_ids() {
+        let client_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let server_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let client_addr = client_socket.local_addr().unwrap();
+        let server_addr = server_socket.local_addr().unwrap();
+        let key = [44u8; 32];
+        let conn_id = [8u8; 16];
+        let mut client = DoshTransport::new_owned(
+            client_socket,
+            SessionTransportConfig {
+                role: SessionRole::Client,
+                conn_id,
+                session_key: key,
+                peer_addr: server_addr,
+                initial_send_seq: 1,
+                initial_ack: 0,
+                stream: TransportConfig::default(),
+            },
+        );
+        let mut server = DoshTransport::new_owned(
+            server_socket,
+            SessionTransportConfig {
+                role: SessionRole::Server,
+                conn_id,
+                session_key: key,
+                peer_addr: client_addr,
+                initial_send_seq: 1,
+                initial_ack: 0,
+                stream: TransportConfig::default(),
+            },
+        );
+
+        assert_eq!(client.allocate_stream_id(), 1);
+        assert_eq!(client.allocate_stream_id(), 2);
+        assert_eq!(server.allocate_stream_id(), 1u64 << 63);
+        assert_eq!(server.allocate_stream_id(), (1u64 << 63) + 1);
+    }
+
+    #[tokio::test]
+    async fn session_transport_handles_simultaneous_opens_without_id_collision() {
+        let client_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let server_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let client_addr = client_socket.local_addr().unwrap();
+        let server_addr = server_socket.local_addr().unwrap();
+        let key = [45u8; 32];
+        let conn_id = [9u8; 16];
+        let mut client = DoshTransport::new_owned(
+            client_socket,
+            SessionTransportConfig {
+                role: SessionRole::Client,
+                conn_id,
+                session_key: key,
+                peer_addr: server_addr,
+                initial_send_seq: 1,
+                initial_ack: 0,
+                stream: TransportConfig::default(),
+            },
+        );
+        let mut server = DoshTransport::new_owned(
+            server_socket,
+            SessionTransportConfig {
+                role: SessionRole::Server,
+                conn_id,
+                session_key: key,
+                peer_addr: client_addr,
+                initial_send_seq: 1,
+                initial_ack: 0,
+                stream: TransportConfig::default(),
+            },
+        );
+
+        let client_stream = client.open_service("client_app").await.unwrap();
+        let server_stream = server.open_service("server_app").await.unwrap();
+        assert_ne!(client_stream, server_stream);
+
+        let server_seen = server.recv().await.unwrap();
+        let client_seen = client.recv().await.unwrap();
+        match server_seen {
+            SessionEvent::Stream(TransportEvent::Open(open)) => {
+                assert_eq!(open.stream_id, client_stream);
+                assert_eq!(open.target_host, "@dosh-client_app");
+                server.accept_stream(open.stream_id).await.unwrap();
+            }
+            other => panic!("unexpected server event {other:?}"),
+        }
+        match client_seen {
+            SessionEvent::Stream(TransportEvent::Open(open)) => {
+                assert_eq!(open.stream_id, server_stream);
+                assert_eq!(open.target_host, "@dosh-server_app");
+                client.accept_stream(open.stream_id).await.unwrap();
+            }
+            other => panic!("unexpected client event {other:?}"),
+        }
+
+        assert!(matches!(
+            client.recv().await.unwrap(),
+            SessionEvent::Stream(TransportEvent::OpenOk { stream_id, .. }) if stream_id == client_stream
+        ));
+        assert!(matches!(
+            server.recv().await.unwrap(),
+            SessionEvent::Stream(TransportEvent::OpenOk { stream_id, .. }) if stream_id == server_stream
+        ));
+    }
+
     #[tokio::test]
     async fn session_transport_opens_service_and_moves_data_over_encrypted_udp() {
         let client_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
@@ -1107,6 +1566,63 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn session_transport_reject_stream_retires_server_state() {
+        let client_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let server_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let client_addr = client_socket.local_addr().unwrap();
+        let server_addr = server_socket.local_addr().unwrap();
+        let key = [46u8; 32];
+        let conn_id = [10u8; 16];
+        let mut client = DoshTransport::new_owned(
+            client_socket,
+            SessionTransportConfig {
+                role: SessionRole::Client,
+                conn_id,
+                session_key: key,
+                peer_addr: server_addr,
+                initial_send_seq: 1,
+                initial_ack: 0,
+                stream: TransportConfig::default(),
+            },
+        );
+        let mut server = DoshTransport::new_owned(
+            server_socket,
+            SessionTransportConfig {
+                role: SessionRole::Server,
+                conn_id,
+                session_key: key,
+                peer_addr: client_addr,
+                initial_send_seq: 1,
+                initial_ack: 0,
+                stream: TransportConfig::default(),
+            },
+        );
+
+        let stream_id = client.open_service("app").await.unwrap();
+        match server.recv().await.unwrap() {
+            SessionEvent::Stream(TransportEvent::Open(open)) => {
+                assert_eq!(open.stream_id, stream_id);
+                assert!(server.mux().is_open(stream_id));
+                server.reject_stream(stream_id, "disabled").await.unwrap();
+            }
+            other => panic!("unexpected event {other:?}"),
+        }
+
+        assert!(!server.mux().is_open(stream_id));
+        assert!(
+            server
+                .send(stream_id, b"must-not-send".to_vec())
+                .await
+                .is_err()
+        );
+        assert!(matches!(
+            client.recv().await.unwrap(),
+            SessionEvent::Stream(TransportEvent::OpenReject { stream_id: rejected, reason })
+                if rejected == stream_id && reason == "disabled"
+        ));
+    }
+
     #[tokio::test]
     async fn session_transport_roams_to_authenticated_new_peer_addr() {
         let server_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();

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(
+        let fatal = std::io::Error::from(std::io::ErrorKind::PermissionDenied);
-            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(
+            let err = std::io::Error::from_raw_os_error(code);
-                &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(
+            let err = std::io::Error::from_raw_os_error(code);
-                &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();
+    let span = u32::from(TEST_UDP_PORT_END - TEST_UDP_PORT_START) + 1;
-    socket.local_addr().unwrap().port()
+    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

@@ -3,9 +3,11 @@
 use std::fs;
 use std::io::{Read, Write};
 use std::net::{TcpListener, TcpStream, UdpSocket};
+use std::os::unix::fs::PermissionsExt;
 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 +25,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();
+    for _ in TEST_PORT_START..=TEST_PORT_END {
-    socket.local_addr().unwrap().port()
+        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();
+    for _ in TEST_PORT_START..=TEST_PORT_END {
-    listener.local_addr().unwrap().port()
+        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 {
@@ -980,6 +1003,8 @@ fn native_file_copy_recursive_round_trip() {
     fs::create_dir_all(src.join("nested")).unwrap();
     fs::write(src.join("root.txt"), b"root file\n").unwrap();
     fs::write(src.join("nested/child.txt"), b"child file\n").unwrap();
+    std::os::unix::fs::symlink("root.txt", src.join("root-link")).unwrap();
+    std::os::unix::fs::symlink("nested/child.txt", src.join("child-link")).unwrap();
 
     let client_bin = env!("CARGO_BIN_EXE_dosh-client");
     let upload = Command::new(client_bin)
@@ -1085,6 +1110,14 @@ fn native_file_copy_recursive_round_trip() {
         fs::read_to_string(downloaded.join("nested/child.txt")).unwrap(),
         "child file\n"
     );
+    assert_eq!(
+        fs::read_link(downloaded.join("root-link")).unwrap(),
+        PathBuf::from("root.txt")
+    );
+    assert_eq!(
+        fs::read_link(downloaded.join("child-link")).unwrap(),
+        PathBuf::from("nested/child.txt")
+    );
 
     let remove = Command::new(client_bin)
         .arg("--dosh-host")
@@ -1108,6 +1141,191 @@ fn native_file_copy_recursive_round_trip() {
     let _ = server.wait();
 }
 
+#[test]
+fn native_file_copy_resume_truncates_oversized_destinations() {
+    let dir = tempfile::tempdir().unwrap();
+    let port = free_udp_port();
+    let config = write_server_config(&dir, port);
+    write_native_client_auth(&dir, &config);
+    let mut server = start_server(&dir, &config);
+    let client_bin = env!("CARGO_BIN_EXE_dosh-client");
+
+    let local_short = dir.path().join("short.txt");
+    fs::write(&local_short, b"short\n").unwrap();
+
+    let seed_remote = Command::new(client_bin)
+        .arg("--dosh-host")
+        .arg("127.0.0.1")
+        .arg("--dosh-port")
+        .arg(port.to_string())
+        .arg("exec")
+        .arg("local")
+        .arg("printf 'short\\nSTALE' > \"$HOME/resume-upload.txt\"")
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    assert!(
+        seed_remote.status.success(),
+        "seed remote failed: stdout={} stderr={}",
+        String::from_utf8_lossy(&seed_remote.stdout),
+        String::from_utf8_lossy(&seed_remote.stderr)
+    );
+
+    let resume_upload = Command::new(client_bin)
+        .arg("--dosh-host")
+        .arg("127.0.0.1")
+        .arg("--dosh-port")
+        .arg(port.to_string())
+        .arg("cp")
+        .arg("--resume")
+        .arg(&local_short)
+        .arg("local:resume-upload.txt")
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    assert!(
+        resume_upload.status.success(),
+        "resume upload failed: stdout={} stderr={}",
+        String::from_utf8_lossy(&resume_upload.stdout),
+        String::from_utf8_lossy(&resume_upload.stderr)
+    );
+
+    let cat_upload = Command::new(client_bin)
+        .arg("--dosh-host")
+        .arg("127.0.0.1")
+        .arg("--dosh-port")
+        .arg(port.to_string())
+        .arg("cat")
+        .arg("local:resume-upload.txt")
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    assert!(
+        cat_upload.status.success(),
+        "cat upload failed: stdout={} stderr={}",
+        String::from_utf8_lossy(&cat_upload.stdout),
+        String::from_utf8_lossy(&cat_upload.stderr)
+    );
+    assert_eq!(String::from_utf8_lossy(&cat_upload.stdout), "short\n");
+
+    let seed_download = Command::new(client_bin)
+        .arg("--dosh-host")
+        .arg("127.0.0.1")
+        .arg("--dosh-port")
+        .arg(port.to_string())
+        .arg("exec")
+        .arg("local")
+        .arg("printf 'tiny\\n' > \"$HOME/resume-download.txt\"")
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    assert!(
+        seed_download.status.success(),
+        "seed download failed: stdout={} stderr={}",
+        String::from_utf8_lossy(&seed_download.stdout),
+        String::from_utf8_lossy(&seed_download.stderr)
+    );
+
+    let local_download = dir.path().join("resume-download-local.txt");
+    fs::write(&local_download, b"tiny\nSTALE").unwrap();
+    let resume_download = Command::new(client_bin)
+        .arg("--dosh-host")
+        .arg("127.0.0.1")
+        .arg("--dosh-port")
+        .arg(port.to_string())
+        .arg("cp")
+        .arg("--resume")
+        .arg("local:resume-download.txt")
+        .arg(&local_download)
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    assert!(
+        resume_download.status.success(),
+        "resume download failed: stdout={} stderr={}",
+        String::from_utf8_lossy(&resume_download.stdout),
+        String::from_utf8_lossy(&resume_download.stderr)
+    );
+    assert_eq!(fs::read_to_string(&local_download).unwrap(), "tiny\n");
+
+    let _ = server.kill();
+    let _ = server.wait();
+}
+
+#[test]
+fn native_file_copy_resume_prefix_mismatch_preserves_destinations() {
+    let dir = tempfile::tempdir().unwrap();
+    let port = free_udp_port();
+    let config = write_server_config(&dir, port);
+    write_native_client_auth(&dir, &config);
+    let mut server = start_server(&dir, &config);
+    let client_bin = env!("CARGO_BIN_EXE_dosh-client");
+
+    let local_upload = dir.path().join("upload-prefix.txt");
+    let remote_upload = dir.path().join("remote-upload.txt");
+    fs::write(&local_upload, b"local-prefix\n").unwrap();
+    fs::write(&remote_upload, b"remote-prefix-with-tail\n").unwrap();
+    fs::set_permissions(&local_upload, fs::Permissions::from_mode(0o755)).unwrap();
+    fs::set_permissions(&remote_upload, fs::Permissions::from_mode(0o600)).unwrap();
+
+    let bad_upload_resume = Command::new(client_bin)
+        .arg("--dosh-host")
+        .arg("127.0.0.1")
+        .arg("--dosh-port")
+        .arg(port.to_string())
+        .arg("cp")
+        .arg("--resume")
+        .arg(&local_upload)
+        .arg("local:remote-upload.txt")
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    assert!(
+        !bad_upload_resume.status.success(),
+        "bad upload resume unexpectedly succeeded: stdout={} stderr={}",
+        String::from_utf8_lossy(&bad_upload_resume.stdout),
+        String::from_utf8_lossy(&bad_upload_resume.stderr)
+    );
+    assert_eq!(
+        fs::read_to_string(&remote_upload).unwrap(),
+        "remote-prefix-with-tail\n"
+    );
+    assert_eq!(
+        fs::metadata(&remote_upload).unwrap().permissions().mode() & 0o777,
+        0o600
+    );
+
+    fs::write(dir.path().join("remote-download.txt"), b"remote-download\n").unwrap();
+    let local_download = dir.path().join("download-prefix.txt");
+    fs::write(&local_download, b"local-download-with-tail\n").unwrap();
+
+    let bad_download_resume = Command::new(client_bin)
+        .arg("--dosh-host")
+        .arg("127.0.0.1")
+        .arg("--dosh-port")
+        .arg(port.to_string())
+        .arg("cp")
+        .arg("--resume")
+        .arg("local:remote-download.txt")
+        .arg(&local_download)
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    assert!(
+        !bad_download_resume.status.success(),
+        "bad download resume unexpectedly succeeded: stdout={} stderr={}",
+        String::from_utf8_lossy(&bad_download_resume.stdout),
+        String::from_utf8_lossy(&bad_download_resume.stderr)
+    );
+    assert_eq!(
+        fs::read_to_string(&local_download).unwrap(),
+        "local-download-with-tail\n"
+    );
+
+    let _ = server.kill();
+    let _ = server.wait();
+}
+
 #[test]
 fn native_exec_command_smoke() {
     let dir = tempfile::tempdir().unwrap();

tests/release_scripts.rs

@@ -19,6 +19,46 @@ fn release_scripts_skip_stale_artifacts_when_auto_discovering() {
     assert!(publish.contains("actual=\"$(artifact_version \"$artifact\" || true)\""));
 }
 
+#[test]
+fn package_check_verifies_only_artifacts_it_builds() {
+    let makefile = include_str!("../Makefile");
+    let verify = include_str!("../scripts/verify-release-artifacts.sh");
+    assert!(
+        verify.contains("if [ \"$#\" -gt 0 ]; then"),
+        "release verifier must accept an explicit artifact list"
+    );
+    let package_check = makefile
+        .split("package-check:")
+        .nth(1)
+        .and_then(|tail| tail.split("\ninstall:").next())
+        .expect("package-check target");
+    assert!(package_check.contains("package-release-linux"));
+    assert!(package_check.contains("package-release-windows"));
+    assert!(package_check.contains("dosh-linux-x86_64.tar.gz"));
+    assert!(package_check.contains("dosh-windows-x86_64.zip"));
+    assert!(
+        !package_check.contains("dosh-macos-aarch64.tar.gz"),
+        "Linux package-check does not build macOS artifacts, so it must not depend on one"
+    );
+}
+
+#[test]
+fn package_release_stamps_git_build_info() {
+    let package = include_str!("../scripts/package-release.sh");
+    let build = include_str!("../build.rs");
+    for name in [
+        "DOSH_BUILD_GIT_HASH",
+        "DOSH_BUILD_COMMIT_DATE",
+        "DOSH_BUILD_GIT_DIRTY",
+    ] {
+        assert!(package.contains(name), "package script must export {name}");
+        assert!(build.contains(name), "build.rs must consume {name}");
+    }
+    assert!(package.contains("git rev-parse --short=12 HEAD"));
+    assert!(package.contains("git show -s --format=%cs HEAD"));
+    assert!(build.contains("cargo:rerun-if-env-changed=DOSH_BUILD_GIT_HASH"));
+}
+
 #[test]
 fn systemd_service_does_not_sandbox_remote_shells() {
     let service = include_str!("../packaging/systemd/dosh-server.service");