Add trace command for reconnect debugging

Cargo.lock

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

Cargo.toml

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

README.md

@@ -68,6 +68,24 @@ dosh restart HOST
 Agent forwarding is opt-in with `-A` and must also be enabled on the server.
 File copy must be enabled by the server config.
 
+## Tracing
+
+For terminal/reconnect bugs, run a client with:
+
+```sh
+dosh trace HOST
+```
+
+The client log path is printed before the session starts. To choose it:
+
+```sh
+dosh trace --client-log /tmp/dosh-client.log HOST
+```
+
+Set `DOSH_TRACE=/tmp/dosh-server.log` on `dosh-server` for matching server
+events. Trace byte prefixes are enabled for `dosh trace`, so use it only for
+short reproductions.
+
 ## VS Code
 
 Dosh can carry VS Code Remote-SSH through its transport:
@@ -108,6 +126,10 @@ 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.

src/bin/dosh-server.rs

@@ -67,6 +67,15 @@ static AGENT_SOCK_COUNTER: AtomicU64 = AtomicU64::new(0);
 /// before it is swept by the cleanup task.
 const NATIVE_HANDSHAKE_TTL_SECS: u64 = 30;
 
+fn hex_id(id: [u8; 16]) -> String {
+    let mut out = String::with_capacity(32);
+    for byte in id {
+        use std::fmt::Write as _;
+        let _ = write!(&mut out, "{byte:02x}");
+    }
+    out
+}
+
 #[derive(Debug, Parser)]
 #[command(
     name = "dosh-server",
@@ -361,6 +370,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>>,
@@ -387,6 +397,14 @@ struct Session {
     last_screen_persist_at: Instant,
 }
 
+#[derive(Clone)]
+struct RestoredScreenSnapshot {
+    cols: u16,
+    rows: u16,
+    output_seq: u64,
+    snapshot: Vec<u8>,
+}
+
 #[derive(Clone)]
 struct ClientState {
     endpoint: SocketAddr,
@@ -512,6 +530,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),
@@ -641,15 +660,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),
@@ -762,6 +789,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"
 }
@@ -1189,7 +1229,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,
@@ -1337,7 +1377,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(
@@ -1473,7 +1513,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(
@@ -1571,9 +1611,22 @@ async fn handle_resume(
     peer: SocketAddr,
     packet: &protocol::Packet,
 ) -> Result<()> {
+    dosh::trace::event(
+        "server.resume_start",
+        &[
+            ("peer", peer.to_string()),
+            ("client", hex_id(packet.header.conn_id)),
+            ("seq", packet.header.seq.to_string()),
+            ("ack", packet.header.ack.to_string()),
+        ],
+    );
     let (key, session_name) = match find_client_decrypt_key(state, &packet.header) {
         Ok(found) => found,
         Err(_) => {
+            dosh::trace::event(
+                "server.resume_unknown_client",
+                &[("client", hex_id(packet.header.conn_id))],
+            );
             return send_reject_to_client(socket, peer, packet.header.conn_id, "unknown client")
                 .await;
         }
@@ -1586,20 +1639,40 @@ async fn handle_resume(
             .sessions
             .get_mut(&req.session)
             .ok_or_else(|| anyhow!("unknown session"))?;
-        let client = session
+        let (send_seq, allow_terminal_updates) = {
-            .clients
+            let client = session
-            .get_mut(&packet.header.conn_id)
+                .clients
-            .ok_or_else(|| anyhow!("unknown client"))?;
+                .get_mut(&packet.header.conn_id)
-        if !client.replay.accept(packet.header.seq) {
+                .ok_or_else(|| anyhow!("unknown client"))?;
-            return Ok(());
+            if !client.replay.accept(packet.header.seq) {
-        }
+                dosh::trace::event(
-        client.endpoint = peer;
+                    "server.resume_replay_drop",
-        client.last_acked = req.last_rendered_seq;
+                    &[
-        client.cols = req.cols;
+                        ("session", req.session.clone()),
-        client.rows = req.rows;
+                        ("seq", packet.header.seq.to_string()),
-        client.last_seen = Instant::now();
+                    ],
-        client.send_seq += 1;
+                );
-        if mode_allows_terminal_updates(&client.mode) {
+                return Ok(());
+            }
+            if client.endpoint != peer {
+                dosh::trace::event(
+                    "server.resume_roam",
+                    &[
+                        ("session", req.session.clone()),
+                        ("from", client.endpoint.to_string()),
+                        ("to", peer.to_string()),
+                    ],
+                );
+            }
+            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,
@@ -1607,8 +1680,8 @@ async fn handle_resume(
                 req.rows,
             )?;
         }
-        let snapshot = screen_snapshot(session.parser.screen());
+        let snapshot = attach_snapshot(session, req.cols, req.rows);
-        (client.send_seq, session.output_seq, snapshot)
+        (send_seq, session.output_seq, snapshot)
     };
     let frame = Frame {
         session: session_name,
@@ -1628,6 +1701,14 @@ async fn handle_resume(
         &body,
     )?;
     let _ = send_udp(socket, &out, peer).await?;
+    dosh::trace::event(
+        "server.resume_ok",
+        &[
+            ("session", frame.session),
+            ("output_seq", frame.output_seq.to_string()),
+            ("bytes", frame.bytes.len().to_string()),
+        ],
+    );
     Ok(())
 }
 
@@ -1639,6 +1720,7 @@ async fn handle_input(
     let (key, session_name) = find_client_decrypt_key(state, &packet.header)?;
     let body = protocol::decrypt_body(packet, &key, CLIENT_TO_SERVER)?;
     let input: Input = protocol::from_body(&body)?;
+    let input_summary = dosh::trace::bytes_summary(&input.bytes);
     let mut locked = state.lock().expect("server state poisoned");
     let session = locked
         .sessions
@@ -1649,13 +1731,37 @@ async fn handle_input(
         .get_mut(&packet.header.conn_id)
         .ok_or_else(|| anyhow!("unknown client"))?;
     if !client.replay.accept(packet.header.seq) {
+        dosh::trace::event(
+            "server.input_replay_drop",
+            &[
+                ("session", session_name.clone()),
+                ("seq", packet.header.seq.to_string()),
+                ("summary", input_summary),
+            ],
+        );
         return Ok(());
     }
     if client.endpoint != peer {
+        dosh::trace::event(
+            "server.input_roam",
+            &[
+                ("session", session_name.clone()),
+                ("from", client.endpoint.to_string()),
+                ("to", peer.to_string()),
+            ],
+        );
         client.endpoint = peer;
     }
     client.last_seen = Instant::now();
     if !mode_allows_terminal_updates(&client.mode) {
+        dosh::trace::event(
+            "server.input_rejected_mode",
+            &[
+                ("session", session_name),
+                ("mode", client.mode.clone()),
+                ("summary", input_summary),
+            ],
+        );
         return Ok(());
     }
     session
@@ -1663,6 +1769,14 @@ async fn handle_input(
         .as_ref()
         .context("terminal session has no pty")?
         .write_all(&input.bytes)?;
+    dosh::trace::event(
+        "server.pty_write",
+        &[
+            ("session", session_name),
+            ("seq", packet.header.seq.to_string()),
+            ("summary", input_summary),
+        ],
+    );
     Ok(())
 }
 
@@ -2532,8 +2646,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(
@@ -2645,6 +2761,46 @@ async fn handle_file_request(
             send_file_response_to_client(state, socket, client_id, stream_id, FileResponse::Ok)
                 .await
         }
+        FileRequest::Readlink { path } => {
+            let path = clean_remote_path(&path, home)?;
+            let target =
+                fs::read_link(&path).with_context(|| format!("readlink {}", path.display()))?;
+            let target = target
+                .to_str()
+                .ok_or_else(|| anyhow!("symlink target is not valid UTF-8"))?
+                .to_string();
+            send_file_response_to_client(
+                state,
+                socket,
+                client_id,
+                stream_id,
+                FileResponse::LinkTarget { target },
+            )
+            .await
+        }
+        FileRequest::Symlink {
+            path,
+            target,
+            overwrite,
+        } => {
+            let path = clean_remote_path(&path, home)?;
+            if let Some(parent) = path.parent() {
+                fs::create_dir_all(parent)
+                    .with_context(|| format!("create {}", parent.display()))?;
+            }
+            if let Ok(metadata) = fs::symlink_metadata(&path) {
+                anyhow::ensure!(overwrite, "destination exists: {}", path.display());
+                anyhow::ensure!(
+                    !metadata.is_dir(),
+                    "destination is a directory: {}",
+                    path.display()
+                );
+                fs::remove_file(&path).with_context(|| format!("remove {}", path.display()))?;
+            }
+            create_symlink(&target, &path)?;
+            send_file_response_to_client(state, socket, client_id, stream_id, FileResponse::Ok)
+                .await
+        }
         FileRequest::Remove { path, recursive } => {
             let path = clean_remote_path(&path, home)?;
             let metadata =
@@ -2682,31 +2838,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 (file, temp_path, written, atomic, original_len, prefix_sha256) =
-                let mut file = fs::OpenOptions::new()
+                if resume && final_path.exists() {
-                    .read(true)
+                    let original_len = fs::metadata(&final_path)
-                    .append(true)
+                        .with_context(|| format!("stat {}", final_path.display()))?
-                    .open(&final_path)
+                        .len();
-                    .with_context(|| format!("open {}", final_path.display()))?;
+                    let written = original_len.min(size);
-                let written = file.metadata()?.len().min(size);
+                    if written > 0 {
-                file.set_len(written)
+                        let mut prefix = fs::File::open(&final_path)?;
-                    .with_context(|| format!("truncate {}", final_path.display()))?;
+                        hash_exact_prefix(&mut prefix, written, &mut hasher)?;
-                if written > 0 {
+                    }
-                    let mut prefix = fs::File::open(&final_path)?;
+                    let prefix_sha256 = (written > 0).then(|| hasher.clone().finalize().into());
-                    hash_exact_prefix(&mut prefix, written, &mut hasher)?;
+                    let mut file = fs::OpenOptions::new()
-                }
+                        .read(true)
-                file.seek(SeekFrom::Start(written))?;
+                        .append(true)
-                (file, None, written, false)
+                        .open(&final_path)
-            } else {
+                        .with_context(|| format!("open {}", final_path.display()))?;
-                let temp_path = upload_temp_path(&final_path, stream_id);
+                    file.seek(SeekFrom::Start(written))?;
-                let file = fs::OpenOptions::new()
+                    (
-                    .create(true)
+                        file,
-                    .truncate(true)
+                        None,
-                    .write(true)
+                        written,
-                    .open(&temp_path)
+                        false,
-                    .with_context(|| format!("create {}", temp_path.display()))?;
+                        Some(original_len),
-                (file, Some(temp_path), 0, true)
+                        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,
@@ -2714,23 +2880,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
         }
@@ -2752,8 +2915,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 {}",
@@ -2764,9 +2925,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()
@@ -2779,6 +2948,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,
@@ -2806,9 +2978,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,
@@ -2817,14 +3000,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 {
@@ -3006,6 +3185,17 @@ fn set_mode(path: &Path, mode: u32) -> Result<()> {
     fs::set_permissions(path, permissions).with_context(|| format!("chmod {}", path.display()))
 }
 
+#[cfg(unix)]
+fn create_symlink(target: &str, path: &Path) -> Result<()> {
+    std::os::unix::fs::symlink(target, path)
+        .with_context(|| format!("symlink {} -> {}", path.display(), target))
+}
+
+#[cfg(not(unix))]
+fn create_symlink(_target: &str, _path: &Path) -> Result<()> {
+    bail!("symlink creation is not supported on this server platform")
+}
+
 fn upload_temp_path(final_path: &Path, stream_id: u64) -> PathBuf {
     let name = final_path
         .file_name()
@@ -3209,7 +3399,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);
                 }
@@ -3454,6 +3647,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());
@@ -4049,6 +4243,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();
@@ -4399,6 +4621,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(),
@@ -4463,6 +4686,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(),
@@ -4539,6 +4763,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(),
@@ -4679,6 +4904,7 @@ mod tests {
             Session {
                 pty: None,
                 parser: vt100::Parser::new(24, 80, 100),
+                restored_screen: None,
                 clients: HashMap::from([(
                     client_id,
                     ClientState {
@@ -4776,6 +5002,7 @@ mod tests {
             Session {
                 pty: None,
                 parser: vt100::Parser::new(24, 80, 100),
+                restored_screen: None,
                 clients: HashMap::from([(
                     client_id,
                     ClientState {

src/client.rs

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

src/file_transfer.rs

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

src/lib.rs

@@ -22,5 +22,6 @@ pub mod protocol;
 pub mod pty;
 pub mod server;
 pub mod ssh_agent;
+pub mod trace;
 pub mod transport;
 pub mod udp;

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/trace.rs

@@ -0,0 +1,161 @@
+use std::fs::{File, OpenOptions};
+use std::io::Write;
+use std::path::PathBuf;
+use std::sync::{Mutex, OnceLock};
+use std::time::{SystemTime, UNIX_EPOCH};
+
+static TRACE_FILE: OnceLock<Option<Mutex<File>>> = OnceLock::new();
+static TRACE_BYTES: OnceLock<bool> = OnceLock::new();
+
+pub fn enabled() -> bool {
+    TRACE_FILE.get_or_init(open_trace_file).is_some()
+}
+
+pub fn bytes_enabled() -> bool {
+    *TRACE_BYTES.get_or_init(|| truthy_env("DOSH_TRACE_BYTES"))
+}
+
+pub fn event(name: &str, fields: &[(&str, String)]) {
+    let Some(file) = TRACE_FILE.get_or_init(open_trace_file) else {
+        return;
+    };
+    let now = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|duration| duration.as_millis())
+        .unwrap_or(0);
+    let mut line = format!(
+        "ts_ms={} pid={} event={}",
+        now,
+        std::process::id(),
+        shell_escape(name)
+    );
+    for (key, value) in fields {
+        line.push(' ');
+        line.push_str(key);
+        line.push('=');
+        line.push_str(&shell_escape(value));
+    }
+    line.push('\n');
+    if let Ok(mut file) = file.lock() {
+        let _ = file.write_all(line.as_bytes());
+        let _ = file.flush();
+    }
+}
+
+pub fn bytes_summary(bytes: &[u8]) -> String {
+    let mut parts = vec![
+        format!("len={}", bytes.len()),
+        format!(
+            "esc={}",
+            contains_byte(bytes, 0x1b) || contains_byte(bytes, 0x9b)
+        ),
+        format!("focus={}", has_focus_report(bytes)),
+        format!("mouseish={}", looks_mouseish(bytes)),
+        format!("printable={}", printable_count(bytes)),
+    ];
+    if bytes_enabled() {
+        parts.push(format!("hex={}", hex_prefix(bytes, 160)));
+    }
+    parts.join(",")
+}
+
+fn open_trace_file() -> Option<Mutex<File>> {
+    let raw = std::env::var_os("DOSH_TRACE")?;
+    let normalized = raw.to_string_lossy().to_ascii_lowercase();
+    if normalized.is_empty() || matches!(normalized.as_str(), "0" | "false" | "off") {
+        return None;
+    }
+    let path = if matches!(normalized.as_str(), "1" | "true" | "on") {
+        default_trace_path()
+    } else {
+        PathBuf::from(raw)
+    };
+    if let Some(parent) = path.parent() {
+        let _ = std::fs::create_dir_all(parent);
+    }
+    OpenOptions::new()
+        .create(true)
+        .append(true)
+        .open(&path)
+        .ok()
+        .map(Mutex::new)
+}
+
+fn default_trace_path() -> PathBuf {
+    let root = dirs::cache_dir()
+        .unwrap_or_else(std::env::temp_dir)
+        .join("dosh")
+        .join("trace");
+    let exe = std::env::current_exe()
+        .ok()
+        .and_then(|path| {
+            path.file_stem()
+                .map(|stem| stem.to_string_lossy().to_string())
+        })
+        .unwrap_or_else(|| "dosh".to_string());
+    root.join(format!("{}-{}.log", exe, std::process::id()))
+}
+
+fn truthy_env(name: &str) -> bool {
+    std::env::var_os(name).is_some_and(|value| {
+        let normalized = value.to_string_lossy().to_ascii_lowercase();
+        !normalized.is_empty() && !matches!(normalized.as_str(), "0" | "false" | "off")
+    })
+}
+
+fn shell_escape(value: &str) -> String {
+    if value.bytes().all(|byte| {
+        byte.is_ascii_alphanumeric()
+            || matches!(byte, b'.' | b'-' | b'_' | b':' | b'/' | b',' | b'=')
+    }) {
+        return value.to_string();
+    }
+    let mut out = String::from("'");
+    for ch in value.chars() {
+        if ch == '\'' {
+            out.push_str("'\\''");
+        } else {
+            out.push(ch);
+        }
+    }
+    out.push('\'');
+    out
+}
+
+fn contains_byte(bytes: &[u8], needle: u8) -> bool {
+    bytes.contains(&needle)
+}
+
+fn has_focus_report(bytes: &[u8]) -> bool {
+    bytes
+        .windows(3)
+        .any(|window| matches!(window, b"\x1b[I" | b"\x1b[O"))
+        || bytes
+            .windows(2)
+            .any(|window| matches!(window, b"\x9bI" | b"\x9bO"))
+}
+
+fn looks_mouseish(bytes: &[u8]) -> bool {
+    bytes.windows(3).any(|window| window == b"\x1b[<")
+        || bytes.windows(2).any(|window| window == b"\x9b<")
+        || bytes.iter().filter(|byte| **byte == b';').take(3).count() >= 2
+}
+
+fn printable_count(bytes: &[u8]) -> usize {
+    bytes
+        .iter()
+        .filter(|byte| byte.is_ascii_graphic() || **byte == b' ')
+        .count()
+}
+
+fn hex_prefix(bytes: &[u8], max: usize) -> String {
+    let mut out = String::with_capacity(bytes.len().min(max) * 2);
+    for byte in bytes.iter().take(max) {
+        use std::fmt::Write as _;
+        let _ = write!(&mut out, "{byte:02x}");
+    }
+    if bytes.len() > max {
+        out.push_str("...");
+    }
+    out
+}

src/transport.rs

@@ -596,11 +596,13 @@ impl StreamMux {
             if self.send_credit.get(&stream_id).copied().unwrap_or(0) < front_len {
                 break;
             }
-            let bytes = self
+            let Some(bytes) = self
                 .pending_data
                 .get_mut(&stream_id)
                 .and_then(VecDeque::pop_front)
-                .expect("pending front exists");
+            else {
+                break;
+            };
             if self
                 .pending_data
                 .get(&stream_id)

src/udp.rs

@@ -1,5 +1,9 @@
 //! UDP error classification shared by Dosh clients, servers, and embedders.
 
+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(),
@@ -13,6 +17,54 @@ 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!(
@@ -52,7 +104,12 @@ fn is_transient_udp_os_error(_code: i32) -> bool {
 
 #[cfg(test)]
 mod tests {
-    use super::{is_transient_udp_error, 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_udp_errors() {
@@ -70,6 +127,35 @@ mod tests {
         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)]
     #[test]
     fn classifies_unix_network_churn_as_transient() {

tests/hostile_network.rs

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

tests/integration_smoke.rs

@@ -3,9 +3,11 @@
 use std::fs;
 use std::io::{Read, Write};
 use std::net::{TcpListener, TcpStream, UdpSocket};
+use std::os::unix::fs::PermissionsExt;
 use std::os::unix::net::{UnixListener, UnixStream};
 use std::path::{Path, PathBuf};
 use std::process::{Child, Command, Stdio};
+use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::mpsc;
 use std::thread;
 use std::time::Duration;
@@ -23,14 +25,35 @@ use dosh::protocol::{
 };
 use ed25519_dalek::{Signer, SigningKey, VerifyingKey};
 
+const TEST_PORT_START: u16 = 30_000;
+const TEST_PORT_END: u16 = 60_999;
+static NEXT_UDP_PORT: AtomicU32 = AtomicU32::new(TEST_PORT_START as u32);
+static NEXT_TCP_PORT: AtomicU32 = AtomicU32::new(TEST_PORT_START as u32);
+
+fn next_test_port(counter: &AtomicU32) -> u16 {
+    let span = u32::from(TEST_PORT_END - TEST_PORT_START) + 1;
+    let offset = counter.fetch_add(1, Ordering::SeqCst) % span;
+    TEST_PORT_START + offset as u16
+}
+
 fn free_udp_port() -> u16 {
-    let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
+    for _ in TEST_PORT_START..=TEST_PORT_END {
-    socket.local_addr().unwrap().port()
+        let port = next_test_port(&NEXT_UDP_PORT);
+        if UdpSocket::bind(("127.0.0.1", port)).is_ok() {
+            return port;
+        }
+    }
+    panic!("no free UDP test port in {TEST_PORT_START}-{TEST_PORT_END}");
 }
 
 fn free_tcp_port() -> u16 {
-    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+    for _ in TEST_PORT_START..=TEST_PORT_END {
-    listener.local_addr().unwrap().port()
+        let port = next_test_port(&NEXT_TCP_PORT);
+        if TcpListener::bind(("127.0.0.1", port)).is_ok() {
+            return port;
+        }
+    }
+    panic!("no free TCP test port in {TEST_PORT_START}-{TEST_PORT_END}");
 }
 
 fn write_server_config(dir: &tempfile::TempDir, port: u16) -> std::path::PathBuf {
@@ -980,6 +1003,8 @@ fn native_file_copy_recursive_round_trip() {
     fs::create_dir_all(src.join("nested")).unwrap();
     fs::write(src.join("root.txt"), b"root file\n").unwrap();
     fs::write(src.join("nested/child.txt"), b"child file\n").unwrap();
+    std::os::unix::fs::symlink("root.txt", src.join("root-link")).unwrap();
+    std::os::unix::fs::symlink("nested/child.txt", src.join("child-link")).unwrap();
 
     let client_bin = env!("CARGO_BIN_EXE_dosh-client");
     let upload = Command::new(client_bin)
@@ -1085,6 +1110,14 @@ fn native_file_copy_recursive_round_trip() {
         fs::read_to_string(downloaded.join("nested/child.txt")).unwrap(),
         "child file\n"
     );
+    assert_eq!(
+        fs::read_link(downloaded.join("root-link")).unwrap(),
+        PathBuf::from("root.txt")
+    );
+    assert_eq!(
+        fs::read_link(downloaded.join("child-link")).unwrap(),
+        PathBuf::from("nested/child.txt")
+    );
 
     let remove = Command::new(client_bin)
         .arg("--dosh-host")
@@ -1219,6 +1252,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();