Release v1.0.0-rc45

Cargo.lock

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

Cargo.toml

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

src/bin/dosh-client.rs

@@ -67,7 +67,7 @@ use tokio::net::windows::named_pipe::{ClientOptions, NamedPipeClient};
 use tokio::net::{TcpListener, TcpStream, UdpSocket};
 use tokio::sync::mpsc;
 
-const STREAM_INITIAL_WINDOW: usize = 1024 * 1024;
+const STREAM_INITIAL_WINDOW: usize = dosh::transport::DEFAULT_INITIAL_WINDOW;
 const STREAM_RETIRED_TOMBSTONES: usize = 16 * 1024;
 const STREAM_CONTROL_RETRANSMIT_MAX_ATTEMPTS: u32 = 8;
 const MAX_PENDING_USER_INPUT_BYTES: usize = 1024 * 1024;
@@ -486,7 +486,8 @@ async fn main() -> Result<()> {
                         detach_once(&socket, &cred, 2).await?;
                         return Ok(());
                     }
-                    return run_terminal(
+                    return trace_terminal_result(
+                        run_terminal(
                             socket,
                             cred,
                             Some(target_udp_addr),
@@ -501,7 +502,8 @@ async fn main() -> Result<()> {
                             false,
                             None,
                         )
-                    .await;
+                        .await,
+                    );
                 }
                 Err(err) => {
                     log_debug(
@@ -525,7 +527,8 @@ async fn main() -> Result<()> {
                         detach_once(&socket, &cred, 2).await?;
                         return Ok(());
                     }
-                    return run_terminal(
+                    return trace_terminal_result(
+                        run_terminal(
                             socket,
                             cred,
                             Some(target_udp_addr),
@@ -540,7 +543,8 @@ async fn main() -> Result<()> {
                             false,
                             None,
                         )
-                    .await;
+                        .await,
+                    );
                 }
                 Err(err) => {
                     log_debug(
@@ -601,7 +605,8 @@ async fn main() -> Result<()> {
                     detach_once(&socket, &cred, 2).await?;
                     return Ok(());
                 }
-                return run_terminal(
+                return trace_terminal_result(
+                    run_terminal(
                         socket,
                         cred,
                         Some(addr),
@@ -616,7 +621,8 @@ async fn main() -> Result<()> {
                         args.forward_only,
                         agent_sock,
                     )
-                .await;
+                    .await,
+                );
             }
             Err(err) if !forwarding_requested && auth_allows(&auth_preference, "ssh") => {
                 log_debug(
@@ -680,6 +686,7 @@ async fn main() -> Result<()> {
         detach_once(&socket, &cred, 2).await?;
         return Ok(());
     }
+    trace_terminal_result(
         run_terminal(
             socket,
             cred,
@@ -695,7 +702,8 @@ async fn main() -> Result<()> {
             false,
             None,
         )
-    .await
+        .await,
+    )
 }
 
 fn run_trust_command(config: &dosh::config::ClientConfig, args: &Args) -> Result<()> {
@@ -2380,8 +2388,21 @@ struct FileForwarder {
 
 impl Drop for FileForwarder {
     fn drop(&mut self) {
+        let exited = self.child.try_wait().ok().flatten();
+        if exited.is_none() {
             let _ = self.child.kill();
             let _ = self.child.wait();
+            return;
+        }
+        if exited.is_some_and(|status| !status.success()) {
+            let mut stderr = String::new();
+            if let Some(mut pipe) = self.child.stderr.take() {
+                let _ = pipe.read_to_string(&mut stderr);
+            }
+            if !stderr.trim().is_empty() {
+                eprintln!("Dosh service forwarder failed:\n{}", stderr.trim_end());
+            }
+        }
     }
 }
 
@@ -6482,6 +6503,13 @@ fn terminal_input_channel() -> (mpsc::Sender<Vec<u8>>, mpsc::Receiver<Vec<u8>>)
     mpsc::channel(STDIN_QUEUE_CAPACITY)
 }
 
+fn trace_terminal_result(result: Result<()>) -> Result<()> {
+    if let Err(err) = &result {
+        dosh::trace::event("client.terminal_error", &[("error", format!("{err:#}"))]);
+    }
+    result
+}
+
 #[allow(clippy::too_many_arguments)]
 async fn run_terminal(
     socket: UdpSocket,
@@ -7500,6 +7528,10 @@ async fn run_terminal(
                         let Ok(ok) = protocol::from_body::<StreamOpenOk>(&plain) else {
                             continue;
                         };
+                        dosh::trace::event(
+                            "client.stream_open_ok",
+                            &[("stream", ok.stream_id.to_string())],
+                        );
                         last_packet_at = Instant::now();
                         let Some(pending_open) = stream_pending_opens.remove(&ok.stream_id) else {
                             continue;
@@ -7549,6 +7581,13 @@ async fn run_terminal(
                         let Ok(reject) = protocol::from_body::<StreamOpenReject>(&plain) else {
                             continue;
                         };
+                        dosh::trace::event(
+                            "client.stream_open_reject",
+                            &[
+                                ("stream", reject.stream_id.to_string()),
+                                ("reason", reject.reason.clone()),
+                            ],
+                        );
                         last_packet_at = Instant::now();
                         stream_pending_opens.remove(&reject.stream_id);
                         if pending_socks_replies.remove(&reject.stream_id)
@@ -7658,6 +7697,10 @@ async fn run_terminal(
                         let Ok(eof) = protocol::from_body::<StreamEof>(&plain) else {
                             continue;
                         };
+                        dosh::trace::event(
+                            "client.stream_eof_received",
+                            &[("stream", eof.stream_id.to_string())],
+                        );
                         last_packet_at = Instant::now();
                         if !opened_streams.contains(&eof.stream_id) {
                             continue;
@@ -7703,6 +7746,10 @@ async fn run_terminal(
                         let Ok(close) = protocol::from_body::<StreamClose>(&plain) else {
                             continue;
                         };
+                        dosh::trace::event(
+                            "client.stream_close_received",
+                            &[("stream", close.stream_id.to_string())],
+                        );
                         last_packet_at = Instant::now();
                         pending_socks_replies.remove(&close.stream_id);
                         stream_close_retransmit.remove(&close.stream_id);
@@ -7732,6 +7779,13 @@ async fn run_terminal(
             forward_event = forward_rx.recv() => {
                 match forward_event {
                     Some(ForwardEvent::Open { stream_id, target_host, target_port, writer, socks5_reply }) => {
+                        dosh::trace::event(
+                            "client.forward_open",
+                            &[
+                                ("stream", stream_id.to_string()),
+                                ("target", format!("{target_host}:{target_port}")),
+                            ],
+                        );
                         stream_writers.insert(stream_id, writer);
                         stream_send_credit.insert(stream_id, STREAM_INITIAL_WINDOW);
                         stream_next_send_offset.entry(stream_id).or_insert(0);
@@ -7772,6 +7826,10 @@ async fn run_terminal(
                         ).await?;
                     }
                     Some(ForwardEvent::Eof { stream_id }) => {
+                        dosh::trace::event(
+                            "client.forward_eof",
+                            &[("stream", stream_id.to_string())],
+                        );
                         stream_local_eof_sent.insert(stream_id);
                         if opened_streams.contains(&stream_id) {
                             send_stream_eof(&socket, addr, &cred, &mut send_seq, stream_id).await?;
@@ -7811,6 +7869,10 @@ async fn run_terminal(
                         }
                     }
                     Some(ForwardEvent::Close { stream_id }) => {
+                        dosh::trace::event(
+                            "client.forward_close",
+                            &[("stream", stream_id.to_string())],
+                        );
                         pending_socks_replies.remove(&stream_id);
                         retire_stream_state(
                             stream_id,

src/bin/dosh-server/unix.rs

@@ -48,7 +48,7 @@ use tokio::net::{TcpListener, TcpStream, UdpSocket, UnixListener};
 use tokio::process::Command as TokioCommand;
 use tokio::sync::mpsc;
 
-const STREAM_INITIAL_WINDOW: usize = 1024 * 1024;
+const STREAM_INITIAL_WINDOW: usize = dosh::transport::DEFAULT_INITIAL_WINDOW;
 const STREAM_RETIRED_TOMBSTONES: usize = 16 * 1024;
 const STREAM_CONTROL_RETRANSMIT_MAX_ATTEMPTS: u32 = 8;
 
@@ -2089,6 +2089,14 @@ async fn handle_stream_open(
     }
 
     if open.target_host == FILE_STREAM_SENTINEL {
+        dosh::trace::event(
+            "server.file_stream_open",
+            &[
+                ("session", session_name.clone()),
+                ("stream", open.stream_id.to_string()),
+                ("peer", peer.to_string()),
+            ],
+        );
         let (writer_tx, writer_rx) = mpsc::channel::<Vec<u8>>(1024);
         register_opened_stream(
             state,
@@ -2113,6 +2121,7 @@ async fn handle_stream_open(
             )
             .await
             {
+                eprintln!("file service stream {stream_id} failed: {err:#}");
                 let _ = send_file_response_to_client(
                     &state,
                     &socket,
@@ -2464,6 +2473,14 @@ async fn handle_stream_eof(
     let (key, session_name) = find_client_decrypt_key(state, &packet.header)?;
     let body = protocol::decrypt_body(packet, &key, CLIENT_TO_SERVER)?;
     let eof: StreamEof = protocol::from_body(&body)?;
+    dosh::trace::event(
+        "server.stream_eof_received",
+        &[
+            ("session", session_name.clone()),
+            ("stream", eof.stream_id.to_string()),
+            ("peer", peer.to_string()),
+        ],
+    );
     let mut locked = state.lock().expect("server state poisoned");
     let Some(session) = locked.sessions.get_mut(&session_name) else {
         return Ok(());
@@ -2495,6 +2512,14 @@ async fn handle_stream_close(
     let (key, session_name) = find_client_decrypt_key(state, &packet.header)?;
     let body = protocol::decrypt_body(packet, &key, CLIENT_TO_SERVER)?;
     let close: StreamClose = protocol::from_body(&body)?;
+    dosh::trace::event(
+        "server.stream_close_received",
+        &[
+            ("session", session_name.clone()),
+            ("stream", close.stream_id.to_string()),
+            ("peer", peer.to_string()),
+        ],
+    );
     let mut locked = state.lock().expect("server state poisoned");
     let Some(session) = locked.sessions.get_mut(&session_name) else {
         return Ok(());
@@ -2887,6 +2912,7 @@ async fn run_file_stream_service(
             )
             .await
             {
+                eprintln!("file service request on stream {stream_id} failed: {err:#}");
                 send_file_response_to_client(
                     &state,
                     &socket,
@@ -2900,6 +2926,10 @@ async fn run_file_stream_service(
             }
         }
     }
+    dosh::trace::event(
+        "server.file_stream_input_closed",
+        &[("stream", stream_id.to_string())],
+    );
     Ok(())
 }
 
@@ -3817,6 +3847,13 @@ async fn send_stream_close_to_client(
     client_id: [u8; 16],
     stream_id: u64,
 ) -> Result<()> {
+    dosh::trace::event(
+        "server.stream_close_sent",
+        &[
+            ("stream", stream_id.to_string()),
+            ("client", hex_id(client_id)),
+        ],
+    );
     {
         let mut locked = state.lock().expect("server state poisoned");
         if let Some(client) = locked.client_mut(&client_id) {

src/transport.rs

@@ -31,7 +31,10 @@ use std::sync::Arc;
 use std::time::{Duration, Instant};
 use tokio::net::UdpSocket;
 
-pub const DEFAULT_INITIAL_WINDOW: usize = 1024 * 1024;
+/// Per-stream bytes allowed in flight before window credit returns. This bounds
+/// each sender burst to 64 MTU-safe packets so bulk streams cannot starve the
+/// terminal while retaining useful bandwidth on high-latency links.
+pub const DEFAULT_INITIAL_WINDOW: usize = 64 * 1024;
 pub const DEFAULT_RETRANSMIT_AFTER: Duration = Duration::from_millis(200);
 pub const ADAPTIVE_RETRANSMIT_PAD: Duration = Duration::from_millis(10);
 pub const ADAPTIVE_RETRANSMIT_MIN: Duration = Duration::from_millis(10);
@@ -39,7 +42,10 @@ pub const DEFAULT_KEEPALIVE_AFTER: Duration = Duration::from_secs(2);
 pub const DEFAULT_RETIRED_STREAM_TOMBSTONES: usize = 16 * 1024;
 pub const STREAM_CONTROL_RETRANSMIT_MAX_ATTEMPTS: u32 = 8;
 pub const SERVICE_TARGET_PREFIX: &str = "@dosh-";
-pub const MAX_STREAM_DATA_BYTES: usize = 60 * 1024;
+/// Maximum application bytes in one encrypted UDP stream packet. Keeping this
+/// aligned with terminal output framing avoids IP fragmentation and stays
+/// below macOS route MTUs after Dosh, AEAD, UDP, and IP overhead.
+pub const MAX_STREAM_DATA_BYTES: usize = 1024;
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct TransportConfig {
@@ -1675,6 +1681,31 @@ mod tests {
         assert_eq!(third.bytes.len(), 13);
     }
 
+    #[test]
+    fn maximum_stream_chunk_fits_a_safe_udp_datagram() {
+        let body = protocol::to_body(&StreamData {
+            stream_id: u64::MAX,
+            offset: u64::MAX,
+            bytes: vec![0xff; MAX_STREAM_DATA_BYTES],
+        })
+        .unwrap();
+        let packet = protocol::encode_encrypted(
+            PacketKind::StreamData,
+            [0xff; 16],
+            u64::MAX,
+            u64::MAX,
+            &[0xff; 32],
+            CLIENT_TO_SERVER,
+            &body,
+        )
+        .unwrap();
+        assert!(
+            packet.len() <= 1200,
+            "encrypted stream datagram is {} bytes",
+            packet.len()
+        );
+    }
+
     #[test]
     fn queued_large_write_flushes_in_window_sized_chunks_after_open() {
         let mut mux = StreamMux::new(TransportConfig {

tests/integration_smoke.rs

@@ -1027,6 +1027,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();
+    let large_payload: Vec<u8> = (0..128 * 1024).map(|index| (index % 251) as u8).collect();
+    fs::write(src.join("large.bin"), &large_payload).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();
 
@@ -1134,6 +1136,10 @@ fn native_file_copy_recursive_round_trip() {
         fs::read_to_string(downloaded.join("nested/child.txt")).unwrap(),
         "child file\n"
     );
+    assert_eq!(
+        fs::read(downloaded.join("large.bin")).unwrap(),
+        large_payload
+    );
     assert_eq!(
         fs::read_link(downloaded.join("root-link")).unwrap(),
         PathBuf::from("root.txt")