Harden stream queues and hostile network tests

Cargo.lock

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

Cargo.toml

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

src/bin/dosh-client.rs

@@ -6253,7 +6253,9 @@ async fn flush_stream_pending_data(
         if credit < bytes.len() {
             break;
         }
-        let bytes = pending.pop_front().expect("pending front exists");
+        let Some(bytes) = pending.pop_front() else {
+            break;
+        };
         *stream_send_credit.entry(stream_id).or_default() -= bytes.len();
         let offset = *stream_next_send_offset.entry(stream_id).or_default();
         stream_next_send_offset.insert(stream_id, offset.saturating_add(bytes.len() as u64));

src/bin/dosh-server.rs

@@ -2752,10 +2752,13 @@ async fn handle_file_request(
                 atomic,
             });
             if let Some(mode) = mode {
-                let target = upload
+                let Some(active_upload) = upload.as_ref() else {
+                    return Err(anyhow!("upload state missing after start"));
+                };
+                let target = active_upload
+                    .temp_path
                     .as_ref()
-                    .and_then(|state| state.temp_path.as_ref())
+                    .unwrap_or(&active_upload.final_path)
-                    .unwrap_or_else(|| &upload.as_ref().unwrap().final_path)
                     .clone();
                 set_mode(&target, mode)?;
             }
@@ -3243,7 +3246,10 @@ async fn flush_stream_pending_data_to_client(
                 if credit < bytes.len() {
                     return Ok(());
                 }
-                let bytes = pending.pop_front().expect("pending front exists");
+                let Some(bytes) = pending.pop_front() else {
+                    client.stream_pending_data.remove(&stream_id);
+                    return Ok(());
+                };
                 if pending.is_empty() {
                     client.stream_pending_data.remove(&stream_id);
                 }

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)

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 {