Harden file resume prefix validation

Cargo.lock

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

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "dosh"
-version = "1.0.0-rc11"
+version = "1.0.0-rc26"
 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 date = git(&["show", "-s", "--format=%cs", "HEAD"]).unwrap_or_else(|| "unknown".into());
-    let dirty = Command::new("git")
-        .args(["diff", "--quiet", "--ignore-submodules", "--"])
-        .status()
-        .map(|status| if status.success() { "" } else { "+dirty" })
-        .unwrap_or("");
+    let hash = env::var("DOSH_BUILD_GIT_HASH")
+        .ok()
+        .filter(|value| !value.is_empty())
+        .or_else(|| git(&["rev-parse", "--short=12", "HEAD"]))
+        .unwrap_or_else(|| "unknown".into());
+    let date = env::var("DOSH_BUILD_COMMIT_DATE")
+        .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 \
-  "$out_dir/dosh-linux-x86_64.tar.gz" \
-  "$out_dir/dosh-macos-aarch64.tar.gz" \
-  "$out_dir/dosh-windows-x86_64.zip"; do
+for artifact in $artifacts; do
   if [ ! -f "$artifact" ]; then
     echo "missing release artifact: $artifact" >&2
     failed=1

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;
@@ -47,11 +47,13 @@ use tokio::process::Command as TokioCommand;
 use tokio::sync::mpsc;
 
 const STREAM_INITIAL_WINDOW: usize = 1024 * 1024;
+const STREAM_RETIRED_TOMBSTONES: usize = 16 * 1024;
 
 /// Persist a persistent session's screen to disk after this many bytes of output
 /// have accumulated since the last mirror. Keeps the atomic write off the
 /// per-packet hot path while bounding how much fresh output a crash can lose.
 const SCREEN_PERSIST_BYTES: usize = 4096;
+const SCREEN_PERSIST_MAX_AGE: Duration = Duration::from_millis(250);
 const IMPLICIT_EMPTY_SESSION_GRACE_SECS: u64 = 300;
 
 /// Sentinel `target_host` sent to the client on a server-initiated `StreamOpen`
@@ -250,7 +252,7 @@ async fn serve(config_path: Option<std::path::PathBuf>) -> Result<()> {
     if config.persist_sessions {
         let flush_state = Arc::clone(&state);
         tokio::spawn(async move {
-            let mut interval = tokio::time::interval(Duration::from_secs(2));
+            let mut interval = tokio::time::interval(SCREEN_PERSIST_MAX_AGE);
             loop {
                 interval.tick().await;
                 flush_persistent_screens(&flush_state);
@@ -260,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:#}");
         }
@@ -355,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>>,
@@ -375,6 +382,18 @@ struct Session {
     /// `output_seq` at the last successful screen mirror, so the periodic flush
     /// skips sessions whose screen has not changed.
     last_persisted_seq: u64,
+    /// Last time this session's screen was considered for a mirror. Combined
+    /// with the byte threshold so low-throughput prompt/TUI updates persist
+    /// quickly without writing on every packet in high-throughput bursts.
+    last_screen_persist_at: Instant,
+}
+
+#[derive(Clone)]
+struct RestoredScreenSnapshot {
+    cols: u16,
+    rows: u16,
+    output_seq: u64,
+    snapshot: Vec<u8>,
 }
 
 #[derive(Clone)]
@@ -399,6 +418,8 @@ struct ClientState {
     stream_sent_data: HashMap<u64, BTreeMap<u64, PendingStreamChunk>>,
     stream_next_recv_offset: HashMap<u64, u64>,
     stream_recv_pending: HashMap<u64, BTreeMap<u64, Vec<u8>>>,
+    stream_retired: HashSet<u64>,
+    stream_retired_order: VecDeque<u64>,
     /// Current transport key epoch (0 = original handshake key). Bumped on rekey.
     epoch: u64,
     /// When the current epoch began, for the wall-clock rekey trigger.
@@ -500,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),
@@ -508,6 +530,7 @@ impl ServerState {
                 persistent,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
+                last_screen_persist_at: Instant::now() - SCREEN_PERSIST_MAX_AGE,
             },
         );
         Ok(())
@@ -628,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),
@@ -645,6 +676,7 @@ impl ServerState {
                     persistent: true,
                     bytes_since_persist: 0,
                     last_persisted_seq: output_seq,
+                    last_screen_persist_at: Instant::now(),
                 },
             );
             eprintln!(
@@ -748,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"
 }
@@ -1175,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,
@@ -1201,6 +1246,8 @@ async fn handle_native_user_auth(
                     stream_sent_data: HashMap::new(),
                     stream_next_recv_offset: HashMap::new(),
                     stream_recv_pending: HashMap::new(),
+                    stream_retired: HashSet::new(),
+                    stream_retired_order: VecDeque::new(),
                     epoch: 0,
                     epoch_started: Instant::now(),
                     epoch_packets: 0,
@@ -1321,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(
@@ -1348,6 +1395,8 @@ async fn handle_bootstrap_attach(
                     stream_sent_data: HashMap::new(),
                     stream_next_recv_offset: HashMap::new(),
                     stream_recv_pending: HashMap::new(),
+                    stream_retired: HashSet::new(),
+                    stream_retired_order: VecDeque::new(),
                     epoch: 0,
                     epoch_started: Instant::now(),
                     epoch_packets: 0,
@@ -1455,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(
@@ -1482,6 +1531,8 @@ async fn handle_ticket_attach(
                     stream_sent_data: HashMap::new(),
                     stream_next_recv_offset: HashMap::new(),
                     stream_recv_pending: HashMap::new(),
+                    stream_retired: HashSet::new(),
+                    stream_retired_order: VecDeque::new(),
                     epoch: 0,
                     epoch_started: Instant::now(),
                     epoch_packets: 0,
@@ -1566,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,
@@ -1587,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,
@@ -1727,10 +1781,10 @@ async fn handle_detach(state: &Arc<Mutex<ServerState>>, packet: &protocol::Packe
     let (key, _) = find_client_decrypt_key(state, &packet.header)?;
     let _ = protocol::decrypt_body(packet, &key, CLIENT_TO_SERVER)?;
     let mut locked = state.lock().expect("server state poisoned");
-    if let Some(client) = locked.client_mut(&packet.header.conn_id) {
-        if !client.replay.accept(packet.header.seq) {
-            return Ok(());
-        }
+    if let Some(client) = locked.client_mut(&packet.header.conn_id)
+        && !client.replay.accept(packet.header.seq)
+    {
+        return Ok(());
     }
     locked.remove_client_everywhere(&packet.header.conn_id);
     Ok(())
@@ -1831,7 +1885,9 @@ async fn handle_stream_open(
         }
         client.endpoint = peer;
         client.last_seen = Instant::now();
-        if client.opened_streams.contains(&open.stream_id) {
+        if client.stream_retired.contains(&open.stream_id) {
+            StreamOpenCheck::New(Err(anyhow!("stream {} is closed", open.stream_id)))
+        } else if client.opened_streams.contains(&open.stream_id) {
             StreamOpenCheck::Duplicate
         } else {
             StreamOpenCheck::New(stream_open_allowed(&config, client, &open))
@@ -2053,6 +2109,9 @@ async fn handle_stream_data(
         }
         client.endpoint = peer;
         client.last_seen = Instant::now();
+        if !client.opened_streams.contains(&data.stream_id) {
+            return Ok(());
+        }
         let writer = client.stream_writers.get(&data.stream_id).cloned();
         let (writes, consumed, received_offset) = accept_stream_data(client, data);
         (writer, writes, consumed, received_offset)
@@ -2062,9 +2121,9 @@ async fn handle_stream_data(
             let _ = writer.send(bytes).await;
         }
     }
-    // Always return flow-control credit, even when the stream's writer is already
-    // gone (closed/unknown). The peer debited its send window for these bytes, so
-    // skipping the adjust would wedge the stream at a permanent credit deficit.
+    // Always return flow-control credit for live streams, even when the stream's
+    // writer is already gone. The peer debited its send window for these bytes,
+    // so skipping the adjust would wedge the stream at a permanent credit deficit.
     send_stream_window_adjust_to_client(
         state,
         socket,
@@ -2101,7 +2160,9 @@ async fn handle_stream_open_ok(
         }
         client.endpoint = peer;
         client.last_seen = Instant::now();
-        client.stream_pending_opens.remove(&ok.stream_id);
+        if client.stream_pending_opens.remove(&ok.stream_id).is_none() {
+            return Ok(());
+        }
         client.opened_streams.insert(ok.stream_id);
         client
             .stream_send_credit
@@ -2133,11 +2194,7 @@ async fn handle_stream_open_reject(
     }
     client.endpoint = peer;
     client.last_seen = Instant::now();
-    client.stream_writers.remove(&reject.stream_id);
-    client.stream_pending_opens.remove(&reject.stream_id);
-    client.opened_streams.remove(&reject.stream_id);
-    client.stream_send_credit.remove(&reject.stream_id);
-    client.stream_pending_data.remove(&reject.stream_id);
+    retire_client_stream(client, reject.stream_id);
     Ok(())
 }
 
@@ -2165,6 +2222,9 @@ async fn handle_stream_window_adjust(
         }
         client.endpoint = peer;
         client.last_seen = Instant::now();
+        if !client.opened_streams.contains(&adjust.stream_id) {
+            return Ok(());
+        }
         ack_stream_data(client, adjust.stream_id, adjust.received_offset);
     }
     flush_stream_pending_data_to_client(state, socket, packet.header.conn_id, adjust.stream_id)
@@ -2192,15 +2252,7 @@ async fn handle_stream_close(
     // Connection migration on any authenticated, fresh packet (spec §11).
     client.endpoint = peer;
     client.last_seen = Instant::now();
-    client.stream_writers.remove(&close.stream_id);
-    client.stream_pending_opens.remove(&close.stream_id);
-    client.opened_streams.remove(&close.stream_id);
-    client.stream_send_credit.remove(&close.stream_id);
-    client.stream_pending_data.remove(&close.stream_id);
-    client.stream_next_send_offset.remove(&close.stream_id);
-    client.stream_sent_data.remove(&close.stream_id);
-    client.stream_next_recv_offset.remove(&close.stream_id);
-    client.stream_recv_pending.remove(&close.stream_id);
+    retire_client_stream(client, close.stream_id);
     Ok(())
 }
 
@@ -2514,8 +2566,10 @@ struct UploadState {
     hasher: Sha256,
     written: u64,
     expected_size: u64,
+    mode: Option<u32>,
     modified_secs: Option<u64>,
     atomic: bool,
+    original_len: Option<u64>,
 }
 
 async fn run_file_stream_service(
@@ -2664,32 +2718,41 @@ async fn handle_file_request(
                 bail!("destination exists: {}", final_path.display());
             }
             let mut hasher = Sha256::new();
-            let (file, temp_path, written, atomic) = if resume && final_path.exists() {
-                let mut file = fs::OpenOptions::new()
-                    .read(true)
-                    .write(true)
-                    .append(true)
-                    .open(&final_path)
-                    .with_context(|| format!("open {}", final_path.display()))?;
-                let written = file.metadata()?.len().min(size);
-                file.set_len(written)
-                    .with_context(|| format!("truncate {}", final_path.display()))?;
-                if written > 0 {
-                    let mut prefix = fs::File::open(&final_path)?;
-                    hash_exact_prefix(&mut prefix, written, &mut hasher)?;
-                }
-                file.seek(SeekFrom::Start(written))?;
-                (file, None, written, false)
-            } else {
-                let temp_path = upload_temp_path(&final_path, stream_id);
-                let file = fs::OpenOptions::new()
-                    .create(true)
-                    .truncate(true)
-                    .write(true)
-                    .open(&temp_path)
-                    .with_context(|| format!("create {}", temp_path.display()))?;
-                (file, Some(temp_path), 0, true)
-            };
+            let (file, temp_path, written, atomic, original_len, prefix_sha256) =
+                if resume && final_path.exists() {
+                    let original_len = fs::metadata(&final_path)
+                        .with_context(|| format!("stat {}", final_path.display()))?
+                        .len();
+                    let written = original_len.min(size);
+                    if written > 0 {
+                        let mut prefix = fs::File::open(&final_path)?;
+                        hash_exact_prefix(&mut prefix, written, &mut hasher)?;
+                    }
+                    let prefix_sha256 = (written > 0).then(|| hasher.clone().finalize().into());
+                    let mut file = fs::OpenOptions::new()
+                        .read(true)
+                        .append(true)
+                        .open(&final_path)
+                        .with_context(|| format!("open {}", final_path.display()))?;
+                    file.seek(SeekFrom::Start(written))?;
+                    (
+                        file,
+                        None,
+                        written,
+                        false,
+                        Some(original_len),
+                        prefix_sha256,
+                    )
+                } else {
+                    let temp_path = upload_temp_path(&final_path, stream_id);
+                    let file = fs::OpenOptions::new()
+                        .create(true)
+                        .truncate(true)
+                        .write(true)
+                        .open(&temp_path)
+                        .with_context(|| format!("create {}", temp_path.display()))?;
+                    (file, Some(temp_path), 0, true, None, None)
+                };
             *upload = Some(UploadState {
                 final_path,
                 temp_path,
@@ -2697,23 +2760,20 @@ async fn handle_file_request(
                 hasher,
                 written,
                 expected_size: size,
+                mode,
                 modified_secs,
                 atomic,
+                original_len,
             });
-            if let Some(mode) = mode {
-                let target = upload
-                    .as_ref()
-                    .and_then(|state| state.temp_path.as_ref())
-                    .unwrap_or_else(|| &upload.as_ref().unwrap().final_path)
-                    .clone();
-                set_mode(&target, mode)?;
-            }
             send_file_response_to_client(
                 state,
                 socket,
                 client_id,
                 stream_id,
-                FileResponse::Resume { offset: written },
+                FileResponse::Resume {
+                    offset: written,
+                    prefix_sha256,
+                },
             )
             .await
         }
@@ -2735,8 +2795,6 @@ async fn handle_file_request(
             let Some(mut active) = upload.take() else {
                 bail!("no upload in progress");
             };
-            active.file.flush()?;
-            active.file.sync_all()?;
             anyhow::ensure!(
                 active.written == active.expected_size,
                 "upload size mismatch: got {}, expected {}",
@@ -2747,9 +2805,17 @@ async fn handle_file_request(
             if digest != sha256 {
                 if let Some(temp_path) = &active.temp_path {
                     let _ = fs::remove_file(temp_path);
+                } else if let Some(original_len) = active.original_len {
+                    let _ = active.file.set_len(original_len);
                 }
                 bail!("upload checksum mismatch");
             }
+            active
+                .file
+                .set_len(active.expected_size)
+                .with_context(|| format!("truncate {}", active.final_path.display()))?;
+            active.file.flush()?;
+            active.file.sync_all()?;
             drop(active.file);
             if active.atomic
                 && let Some(temp_path) = active.temp_path.take()
@@ -2762,6 +2828,9 @@ async fn handle_file_request(
                     )
                 })?;
             }
+            if let Some(mode) = active.mode {
+                set_mode(&active.final_path, mode)?;
+            }
             if let Some(modified_secs) = active.modified_secs {
                 filetime::set_file_mtime(
                     &active.final_path,
@@ -2789,9 +2858,20 @@ async fn handle_file_request(
                 "remote path is not a file: {}",
                 meta.path
             );
+            anyhow::ensure!(
+                offset <= meta.len,
+                "download offset {offset} is beyond file size {}",
+                meta.len
+            );
             let mut file =
                 fs::File::open(&path).with_context(|| format!("open {}", path.display()))?;
             file.seek(SeekFrom::Start(offset))?;
+            let mut hasher = Sha256::new();
+            if offset > 0 {
+                let mut prefix = fs::File::open(&path)?;
+                hash_exact_prefix(&mut prefix, offset, &mut hasher)?;
+            }
+            let prefix_sha256 = (offset > 0).then(|| hasher.clone().finalize().into());
             send_file_response_to_client(
                 state,
                 socket,
@@ -2800,14 +2880,10 @@ async fn handle_file_request(
                 FileResponse::Start {
                     meta: meta.clone(),
                     offset,
+                    prefix_sha256,
                 },
             )
             .await?;
-            let mut hasher = Sha256::new();
-            if offset > 0 {
-                let mut prefix = fs::File::open(&path)?;
-                hash_exact_prefix(&mut prefix, offset, &mut hasher)?;
-            }
             let mut sent = offset;
             let mut buf = vec![0u8; CHUNK_SIZE];
             loop {
@@ -3192,7 +3268,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);
                 }
@@ -3280,6 +3359,46 @@ fn accept_stream_data(client: &mut ClientState, data: StreamData) -> (Vec<Vec<u8
     (writes, consumed, *expected)
 }
 
+fn cleanup_client_stream(client: &mut ClientState, stream_id: u64) {
+    client.stream_writers.remove(&stream_id);
+    client.stream_pending_opens.remove(&stream_id);
+    client.opened_streams.remove(&stream_id);
+    client.stream_send_credit.remove(&stream_id);
+    client.stream_pending_data.remove(&stream_id);
+    client.stream_next_send_offset.remove(&stream_id);
+    client.stream_sent_data.remove(&stream_id);
+    client.stream_next_recv_offset.remove(&stream_id);
+    client.stream_recv_pending.remove(&stream_id);
+}
+
+fn client_has_stream_state(client: &ClientState, stream_id: u64) -> bool {
+    client.stream_writers.contains_key(&stream_id)
+        || client.stream_pending_opens.contains_key(&stream_id)
+        || client.opened_streams.contains(&stream_id)
+        || client.stream_send_credit.contains_key(&stream_id)
+        || client.stream_pending_data.contains_key(&stream_id)
+        || client.stream_next_send_offset.contains_key(&stream_id)
+        || client.stream_sent_data.contains_key(&stream_id)
+        || client.stream_next_recv_offset.contains_key(&stream_id)
+        || client.stream_recv_pending.contains_key(&stream_id)
+}
+
+fn retire_client_stream(client: &mut ClientState, stream_id: u64) {
+    let had_state = client_has_stream_state(client, stream_id);
+    cleanup_client_stream(client, stream_id);
+    if !had_state || STREAM_RETIRED_TOMBSTONES == 0 {
+        return;
+    }
+    if client.stream_retired.insert(stream_id) {
+        client.stream_retired_order.push_back(stream_id);
+    }
+    while client.stream_retired_order.len() > STREAM_RETIRED_TOMBSTONES {
+        if let Some(expired) = client.stream_retired_order.pop_front() {
+            client.stream_retired.remove(&expired);
+        }
+    }
+}
+
 fn ack_stream_data(client: &mut ClientState, stream_id: u64, received_offset: u64) {
     let Some(sent) = client.stream_sent_data.get_mut(&stream_id) else {
         return;
@@ -3340,14 +3459,7 @@ async fn send_stream_close_to_client(
     {
         let mut locked = state.lock().expect("server state poisoned");
         if let Some(client) = locked.client_mut(&client_id) {
-            client.stream_writers.remove(&stream_id);
-            client.opened_streams.remove(&stream_id);
-            client.stream_send_credit.remove(&stream_id);
-            client.stream_pending_data.remove(&stream_id);
-            client.stream_next_send_offset.remove(&stream_id);
-            client.stream_sent_data.remove(&stream_id);
-            client.stream_next_recv_offset.remove(&stream_id);
-            client.stream_recv_pending.remove(&stream_id);
+            retire_client_stream(client, stream_id);
         }
     }
     let body = protocol::to_body(&StreamClose { stream_id })?;
@@ -3404,6 +3516,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());
@@ -3411,16 +3524,20 @@ async fn broadcast_output(
             session.recent.pop_front();
         }
         // For a persistent session, mirror the screen to disk periodically so a
-        // post-restart reattach repaints. Throttled by accumulated bytes to keep
-        // the (atomic) write off the per-packet hot path. The actual file write
-        // happens after the lock is dropped.
+        // post-restart reattach repaints. Throttled by accumulated bytes or a
+        // short wall-clock age so low-throughput prompt/TUI changes persist
+        // quickly while large bursts still avoid per-packet disk writes. The
+        // actual file write happens after the lock is dropped.
         if session.persistent {
             session.bytes_since_persist = session
                 .bytes_since_persist
                 .saturating_add(output.bytes.len());
-            if session.bytes_since_persist >= SCREEN_PERSIST_BYTES {
+            let now = Instant::now();
+            if session.bytes_since_persist >= SCREEN_PERSIST_BYTES
+                || now.duration_since(session.last_screen_persist_at) >= SCREEN_PERSIST_MAX_AGE
+            {
                 session.bytes_since_persist = 0;
-                session.last_persisted_seq = output_seq;
+                session.last_screen_persist_at = now;
                 let screen = session.parser.screen();
                 persist_screen = Some((
                     output.session.clone(),
@@ -3479,6 +3596,8 @@ async fn broadcast_output(
             persist::save_screen(&sessions_dir, &name, cols, rows, output_seq, &snapshot)
         {
             eprintln!("screen persist for {name} failed: {err:#}");
+        } else {
+            mark_screen_persisted(state, &name, output_seq);
         }
     }
     for (endpoint, packet) in sends {
@@ -3760,8 +3879,8 @@ fn flush_persistent_screens(state: &Arc<Mutex<ServerState>>) {
             if !session.persistent || session.output_seq == session.last_persisted_seq {
                 continue;
             }
-            session.last_persisted_seq = session.output_seq;
             session.bytes_since_persist = 0;
+            session.last_screen_persist_at = Instant::now();
             let screen = session.parser.screen();
             to_write.push((
                 name.clone(),
@@ -3778,10 +3897,19 @@ fn flush_persistent_screens(state: &Arc<Mutex<ServerState>>) {
             persist::save_screen(&sessions_dir, &name, cols, rows, output_seq, &snapshot)
         {
             eprintln!("screen flush for {name} failed: {err:#}");
+        } else {
+            mark_screen_persisted(state, &name, output_seq);
         }
     }
 }
 
+fn mark_screen_persisted(state: &Arc<Mutex<ServerState>>, name: &str, output_seq: u64) {
+    let mut locked = state.lock().expect("server state poisoned");
+    if let Some(session) = locked.sessions.get_mut(name) {
+        session.last_persisted_seq = session.last_persisted_seq.max(output_seq);
+    }
+}
+
 fn cleanup_disconnected_clients(state: &Arc<Mutex<ServerState>>) {
     // Sessions removed here are dropped after the lock is released so their
     // shells are killed (PtyHandle::drop) without holding up the event loop.
@@ -3984,6 +4112,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();
@@ -4046,6 +4202,8 @@ mod tests {
             stream_sent_data: HashMap::new(),
             stream_next_recv_offset: HashMap::new(),
             stream_recv_pending: HashMap::new(),
+            stream_retired: HashSet::new(),
+            stream_retired_order: VecDeque::new(),
             epoch: 0,
             epoch_started: Instant::now(),
             epoch_packets: 0,
@@ -4053,6 +4211,125 @@ mod tests {
         }
     }
 
+    #[test]
+    fn cleanup_client_stream_removes_all_per_stream_state_only_for_target() {
+        let stream_id = 42;
+        let other_stream_id = 43;
+        let mut client = test_client_state([7u8; 32]);
+        let (writer_tx, _writer_rx) = mpsc::channel(1);
+        client.stream_writers.insert(stream_id, writer_tx);
+        client.opened_streams.insert(stream_id);
+        client.opened_streams.insert(other_stream_id);
+        client.stream_send_credit.insert(stream_id, 12);
+        client.stream_send_credit.insert(other_stream_id, 34);
+        client
+            .stream_pending_data
+            .insert(stream_id, VecDeque::from([vec![1]]));
+        client
+            .stream_pending_data
+            .insert(other_stream_id, VecDeque::from([vec![2]]));
+        client.stream_pending_opens.insert(
+            stream_id,
+            PendingStreamOpen {
+                target_host: "127.0.0.1".to_string(),
+                target_port: 1234,
+                last_sent: Instant::now(),
+                attempts: 1,
+            },
+        );
+        client.stream_pending_opens.insert(
+            other_stream_id,
+            PendingStreamOpen {
+                target_host: "127.0.0.1".to_string(),
+                target_port: 1235,
+                last_sent: Instant::now(),
+                attempts: 1,
+            },
+        );
+        client.stream_next_send_offset.insert(stream_id, 55);
+        client.stream_next_send_offset.insert(other_stream_id, 66);
+        client.stream_sent_data.insert(
+            stream_id,
+            BTreeMap::from([(
+                0,
+                PendingStreamChunk {
+                    offset: 0,
+                    bytes: vec![3],
+                    last_sent: Instant::now(),
+                    attempts: 1,
+                },
+            )]),
+        );
+        client.stream_sent_data.insert(
+            other_stream_id,
+            BTreeMap::from([(
+                0,
+                PendingStreamChunk {
+                    offset: 0,
+                    bytes: vec![4],
+                    last_sent: Instant::now(),
+                    attempts: 1,
+                },
+            )]),
+        );
+        client.stream_next_recv_offset.insert(stream_id, 77);
+        client.stream_next_recv_offset.insert(other_stream_id, 88);
+        client
+            .stream_recv_pending
+            .insert(stream_id, BTreeMap::from([(77, vec![5])]));
+        client
+            .stream_recv_pending
+            .insert(other_stream_id, BTreeMap::from([(88, vec![6])]));
+
+        cleanup_client_stream(&mut client, stream_id);
+
+        assert!(!client.stream_writers.contains_key(&stream_id));
+        assert!(!client.opened_streams.contains(&stream_id));
+        assert!(!client.stream_send_credit.contains_key(&stream_id));
+        assert!(!client.stream_pending_data.contains_key(&stream_id));
+        assert!(!client.stream_pending_opens.contains_key(&stream_id));
+        assert!(!client.stream_next_send_offset.contains_key(&stream_id));
+        assert!(!client.stream_sent_data.contains_key(&stream_id));
+        assert!(!client.stream_next_recv_offset.contains_key(&stream_id));
+        assert!(!client.stream_recv_pending.contains_key(&stream_id));
+
+        assert!(client.opened_streams.contains(&other_stream_id));
+        assert!(client.stream_send_credit.contains_key(&other_stream_id));
+        assert!(client.stream_pending_data.contains_key(&other_stream_id));
+        assert!(client.stream_pending_opens.contains_key(&other_stream_id));
+        assert!(
+            client
+                .stream_next_send_offset
+                .contains_key(&other_stream_id)
+        );
+        assert!(client.stream_sent_data.contains_key(&other_stream_id));
+        assert!(
+            client
+                .stream_next_recv_offset
+                .contains_key(&other_stream_id)
+        );
+        assert!(client.stream_recv_pending.contains_key(&other_stream_id));
+    }
+
+    #[test]
+    fn retire_client_stream_tombstones_only_real_stream_state() {
+        let stream_id = 42;
+        let mut client = test_client_state([8u8; 32]);
+        client.opened_streams.insert(stream_id);
+        client
+            .stream_send_credit
+            .insert(stream_id, STREAM_INITIAL_WINDOW);
+
+        retire_client_stream(&mut client, stream_id);
+        retire_client_stream(&mut client, 99);
+
+        assert!(!client.opened_streams.contains(&stream_id));
+        assert!(!client.stream_send_credit.contains_key(&stream_id));
+        assert!(client.stream_retired.contains(&stream_id));
+        assert!(!client.stream_retired.contains(&99));
+        assert_eq!(client.stream_retired_order, VecDeque::from([stream_id]));
+    }
+
     #[test]
     fn client_index_stays_in_sync_with_session_clients() {
         let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
@@ -4213,6 +4490,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(),
@@ -4221,6 +4499,7 @@ mod tests {
                 persistent: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
+                last_screen_persist_at: Instant::now(),
             },
         );
         state.client_index.insert(client_id, "test".to_string());
@@ -4258,6 +4537,76 @@ mod tests {
         assert_eq!(ok.stream_id, stream_id);
     }
 
+    #[tokio::test]
+    async fn retired_stream_open_is_rejected_not_reopened() {
+        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
+        let client_id = [15u8; 16];
+        let session_key = [16u8; 32];
+        let stream_id = 100;
+        let receiver = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let sender = Arc::new(UdpSocket::bind("127.0.0.1:0").await.unwrap());
+        let mut client = test_client_state(session_key);
+        client.endpoint = receiver.local_addr().unwrap();
+        client.stream_retired.insert(stream_id);
+        client.stream_retired_order.push_back(stream_id);
+        state.sessions.insert(
+            "test".to_string(),
+            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(),
+                empty_since: None,
+                holder_control: None,
+                persistent: false,
+                bytes_since_persist: 0,
+                last_persisted_seq: 0,
+                last_screen_persist_at: Instant::now(),
+            },
+        );
+        state.client_index.insert(client_id, "test".to_string());
+        let state = Arc::new(Mutex::new(state));
+        let raw = protocol::encode_encrypted(
+            PacketKind::StreamOpen,
+            client_id,
+            2,
+            0,
+            &session_key,
+            CLIENT_TO_SERVER,
+            &protocol::to_body(&StreamOpen {
+                stream_id,
+                target_host: "127.0.0.1".to_string(),
+                target_port: 9,
+            })
+            .unwrap(),
+        )
+        .unwrap();
+        let packet = protocol::decode(&raw).unwrap();
+
+        handle_stream_open(&state, &sender, receiver.local_addr().unwrap(), &packet)
+            .await
+            .unwrap();
+
+        let mut buf = [0u8; 2048];
+        let (n, _) = tokio::time::timeout(Duration::from_secs(1), receiver.recv_from(&mut buf))
+            .await
+            .unwrap()
+            .unwrap();
+        let packet = protocol::decode(&buf[..n]).unwrap();
+        assert_eq!(packet.header.kind, PacketKind::StreamOpenReject);
+        let plain = protocol::decrypt_body(&packet, &session_key, SERVER_TO_CLIENT).unwrap();
+        let reject: StreamOpenReject = protocol::from_body(&plain).unwrap();
+        assert_eq!(reject.stream_id, stream_id);
+        assert!(reject.reason.contains("closed"));
+
+        let locked = state.lock().unwrap();
+        let client = locked.sessions["test"].clients.get(&client_id).unwrap();
+        assert!(!client.opened_streams.contains(&stream_id));
+    }
+
     #[tokio::test]
     async fn pending_server_stream_open_is_retransmitted() {
         let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
@@ -4283,6 +4632,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(),
@@ -4291,6 +4641,7 @@ mod tests {
                 persistent: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
+                last_screen_persist_at: Instant::now(),
             },
         );
         state.client_index.insert(client_id, "test".to_string());
@@ -4422,6 +4773,7 @@ mod tests {
             Session {
                 pty: None,
                 parser: vt100::Parser::new(24, 80, 100),
+                restored_screen: None,
                 clients: HashMap::from([(
                     client_id,
                     ClientState {
@@ -4445,6 +4797,8 @@ mod tests {
                         stream_sent_data: HashMap::new(),
                         stream_next_recv_offset: HashMap::new(),
                         stream_recv_pending: HashMap::new(),
+                        stream_retired: HashSet::new(),
+                        stream_retired_order: VecDeque::new(),
                         epoch: 0,
                         epoch_started: Instant::now(),
                         epoch_packets: 0,
@@ -4458,6 +4812,7 @@ mod tests {
                 persistent: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
+                last_screen_persist_at: Instant::now(),
             },
         );
         // Keep the O(1) conn_id index in sync, matching `insert_client`.
@@ -4516,6 +4871,7 @@ mod tests {
             Session {
                 pty: None,
                 parser: vt100::Parser::new(24, 80, 100),
+                restored_screen: None,
                 clients: HashMap::from([(
                     client_id,
                     ClientState {
@@ -4539,6 +4895,8 @@ mod tests {
                         stream_sent_data: HashMap::new(),
                         stream_next_recv_offset: HashMap::new(),
                         stream_recv_pending: HashMap::new(),
+                        stream_retired: HashSet::new(),
+                        stream_retired_order: VecDeque::new(),
                         epoch: 0,
                         epoch_started: Instant::now(),
                         epoch_packets: 0,
@@ -4552,6 +4910,7 @@ mod tests {
                 persistent: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
+                last_screen_persist_at: Instant::now(),
             },
         );
         state.client_index.insert(client_id, "test".to_string());

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/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

@@ -46,13 +46,32 @@ pub enum FileRequest {
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 pub enum FileResponse {
     Ok,
-    Resume { offset: u64 },
-    Stat { meta: FileMeta },
-    List { entries: Vec<FileEntry> },
-    Start { meta: FileMeta, offset: u64 },
-    Chunk { offset: u64, bytes: Vec<u8> },
-    Done { sha256: [u8; 32], bytes: u64 },
-    Error { message: String },
+    Resume {
+        offset: u64,
+        prefix_sha256: Option<[u8; 32]>,
+    },
+    Stat {
+        meta: FileMeta,
+    },
+    List {
+        entries: Vec<FileEntry>,
+    },
+    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,
         }
     }
 }
@@ -141,6 +144,9 @@ pub enum TransportEvent {
     Close {
         stream_id: u64,
     },
+    Ignored {
+        stream_id: u64,
+    },
 }
 
 #[derive(Debug, Clone)]
@@ -170,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)]
@@ -225,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(),
         }
     }
 
@@ -235,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
@@ -262,6 +275,9 @@ 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",
@@ -295,19 +311,39 @@ impl StreamMux {
         )
     }
 
-    pub fn handle_open_ok(&mut self, ok: StreamOpenOk) -> Result<Vec<OutgoingStreamPacket>> {
-        self.pending_opens.remove(&ok.stream_id);
+    pub fn reject_open(
+        &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) {
-        self.cleanup_stream(reject.stream_id);
+    pub fn handle_open_reject(&mut self, reject: StreamOpenReject) -> bool {
+        if !self.has_stream_state(reject.stream_id) {
+            return false;
+        }
+        self.retire_stream(reject.stream_id);
+        true
     }
 
     pub fn send_data(
@@ -344,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,
@@ -355,37 +394,52 @@ 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) {
-        self.cleanup_stream(close.stream_id);
+    pub fn handle_close(&mut self, close: StreamClose) -> bool {
+        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,
@@ -397,32 +451,47 @@ impl StreamMux {
             PacketKind::StreamOpenOk => {
                 let ok: StreamOpenOk = protocol::from_body(body)?;
                 let stream_id = ok.stream_id;
-                let flushed = self.handle_open_ok(ok)?;
-                Ok(TransportEvent::OpenOk { stream_id, flushed })
+                match self.handle_open_ok(ok)? {
+                    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());
-                Ok(TransportEvent::OpenReject {
-                    stream_id: reject.stream_id,
-                    reason: reject.reason,
-                })
+                let stream_id = reject.stream_id;
+                if self.handle_open_reject(reject.clone()) {
+                    Ok(TransportEvent::OpenReject {
+                        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)?;
-                Ok(TransportEvent::WindowAdjust { stream_id, flushed })
+                match self.handle_window_adjust(adjust)? {
+                    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);
-                Ok(TransportEvent::Close { stream_id })
+                if self.handle_close(close) {
+                    Ok(TransportEvent::Close { stream_id })
+                } else {
+                    Ok(TransportEvent::Ignored { stream_id })
+                }
             }
             _ => bail!("packet kind {kind:?} is not a stream transport packet"),
         }
@@ -527,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)
@@ -610,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);
@@ -727,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)?)
-            .await
+        let packet = self.mux.reject_open(stream_id, reason)?;
+        self.send_outgoing(packet).await
     }
 
     pub async fn send(&mut self, stream_id: u64, bytes: impl Into<Vec<u8>>) -> Result<()> {
@@ -757,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,
@@ -853,7 +954,8 @@ impl DoshTransport {
             }
             TransportEvent::Open(_)
             | TransportEvent::OpenReject { .. }
-            | TransportEvent::Close { .. } => {}
+            | TransportEvent::Close { .. }
+            | TransportEvent::Ignored { .. } => {}
         }
         Ok(())
     }
@@ -888,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,
@@ -937,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();
@@ -976,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]);
@@ -999,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);
@@ -1009,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);
@@ -1023,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());
@@ -1034,6 +1176,7 @@ mod tests {
                 received_offset: 5,
                 bytes: 5,
             })
+            .unwrap()
             .unwrap();
         assert_eq!(flushed.len(), 1);
         let data: StreamData = decode(&flushed[0]);
@@ -1049,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);
 
@@ -1059,6 +1204,7 @@ mod tests {
                 received_offset: 5,
                 bytes: 0,
             })
+            .unwrap()
             .unwrap();
         assert!(flushed.is_empty());
         assert_eq!(mux.send_credit[&1], 5);
@@ -1119,13 +1265,142 @@ mod tests {
         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();
+        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();
@@ -1291,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(
-            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

@@ -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();
-    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 {
@@ -1219,6 +1242,80 @@ fn native_file_copy_resume_truncates_oversized_destinations() {
     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");