diff --git a/Cargo.lock b/Cargo.lock index 3f2a36f..3f416fc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -436,7 +436,7 @@ dependencies = [ [[package]] name = "dosh" -version = "1.0.0-rc23" +version = "1.0.0-rc24" dependencies = [ "anyhow", "base64", diff --git a/Cargo.toml b/Cargo.toml index 7f82cc0..8665076 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "dosh" -version = "1.0.0-rc23" +version = "1.0.0-rc24" edition = "2024" license = "MIT" diff --git a/src/bin/dosh-client.rs b/src/bin/dosh-client.rs index 6852348..80e76b6 100644 --- a/src/bin/dosh-client.rs +++ b/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)); diff --git a/src/bin/dosh-server.rs b/src/bin/dosh-server.rs index 6e9feb5..6f76014 100644 --- a/src/bin/dosh-server.rs +++ b/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_else(|| &upload.as_ref().unwrap().final_path) + .unwrap_or(&active_upload.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); } diff --git a/src/transport.rs b/src/transport.rs index 93916dd..5674f82 100644 --- a/src/transport.rs +++ b/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) diff --git a/tests/hostile_network.rs b/tests/hostile_network.rs index 10a714b..3799dcc 100644 --- a/tests/hostile_network.rs +++ b/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 {