diff --git a/crates/vnidrop/src/runtime.rs b/crates/vnidrop/src/runtime.rs index 04c3db6..e5b7953 100644 --- a/crates/vnidrop/src/runtime.rs +++ b/crates/vnidrop/src/runtime.rs @@ -1,5 +1,6 @@ use std::{ collections::HashMap, + future::Future, io, path::{Path, PathBuf}, str::FromStr, @@ -59,6 +60,19 @@ pub struct VnidropCore { inner: Arc, } +impl VnidropCore { + /// Drive work on this core's multi-thread runtime from a sync API boundary. + /// + /// Uses [`tokio::runtime::Handle::block_on`] rather than exclusive + /// [`Runtime::block_on`] so a concurrent call (for example cancel while + /// another thread is blocked inside `receive`) cannot deadlock the runtime + /// driver. UniFFI and tests both rely on that: receive runs on a worker + /// thread while cancel/approve arrive from the UI or test harness thread. + fn block_on(&self, future: F) -> F::Output { + self.runtime.handle().block_on(future) + } +} + struct CoreInner { // Kotlin owns app lifecycle and platform file picking; this Rust object // owns the Iroh endpoint, blob store, transfer history, and byte streaming. @@ -71,7 +85,9 @@ struct CoreInner { limits: CoreLimits, transfer_slots: Semaphore, access_policy: Arc, - active_transfers: TokioMutex>, + /// Sync mutex so cancel can remove + signal without awaiting (and without + /// holding a Tokio lock across repository I/O). + active_transfers: std::sync::Mutex>, // Newly imported shares retain a TempTag for the lifetime of this process. // Restored shares have no in-memory tag, but remain tracked so they can be // counted and explicitly revoked after a restart. @@ -173,7 +189,7 @@ impl VnidropCore { } pub fn status(&self) -> RuntimeStatus { - self.runtime.block_on(self.inner.status()) + self.block_on(self.inner.status()) } pub fn share_files( @@ -181,8 +197,7 @@ impl VnidropCore { sources: Vec, metadata: ShareMetadataInput, ) -> Result { - self.runtime - .block_on(self.inner.share_files(sources, metadata)) + self.block_on(self.inner.share_files(sources, metadata)) .map_err(VnidropError::transfer) } @@ -195,7 +210,7 @@ impl VnidropCore { if let Err(error) = parse_transfer_ticket_with_limits(&ticket, &self.inner.limits) .context("failed to parse transfer ticket") { - self.runtime.block_on(async { + self.block_on(async { self.inner.emit_endpoint( "error", "invalid-ticket", @@ -206,8 +221,7 @@ impl VnidropCore { return Err(VnidropError::ticket(error)); } let output_dir = platform_path(&output_dir).map_err(VnidropError::filesystem)?; - self.runtime - .block_on(self.inner.receive(ticket, output_dir, receiver_name)) + self.block_on(self.inner.receive(ticket, output_dir, receiver_name)) .map_err(VnidropError::transfer) } @@ -220,7 +234,7 @@ impl VnidropCore { if let Err(error) = parse_transfer_ticket_with_limits(&ticket, &self.inner.limits) .context("failed to parse transfer ticket") { - self.runtime.block_on(async { + self.block_on(async { self.inner.emit_endpoint( "error", "invalid-ticket", @@ -230,29 +244,54 @@ impl VnidropCore { }); return Err(VnidropError::ticket(error)); } - self.runtime - .block_on( - self.inner - .receive_with_output_sink(ticket, output_sink, receiver_name), - ) - .map_err(VnidropError::transfer) + self.block_on( + self.inner + .receive_with_output_sink(ticket, output_sink, receiver_name), + ) + .map_err(VnidropError::transfer) } pub fn cancel_transfer(&self, transfer_id: u64) -> Result<(), VnidropError> { - self.runtime - .block_on(self.inner.cancel_transfer(transfer_id)) + // Fire the oneshot on this thread before entering the runtime so a + // blocked export write cannot prevent the cancel signal from being + // delivered (the receive `select!` observes it on the next yield). + if let Some(direction) = self.inner.take_active_transfer(transfer_id) { + let expected = match direction { + TransferDirection::Send => TransferStatus::Importing, + TransferDirection::Receive => TransferStatus::Receiving, + }; + return self + .block_on(async { + self.inner + .repository + .transition_transfer_status( + transfer_id, + expected, + TransferStatus::Cancelled, + ) + .await?; + self.inner.emit_transfer( + transfer_id, + direction.as_str(), + "lifecycle", + "cancel-requested", + json!({}), + ); + Ok::<(), anyhow::Error>(()) + }) + .map_err(VnidropError::transfer); + } + self.block_on(self.inner.cancel_idle_or_share(transfer_id)) .map_err(VnidropError::transfer) } pub fn delete_transfer(&self, transfer_id: u64) -> Result<(), VnidropError> { - self.runtime - .block_on(self.inner.delete_transfer(transfer_id)) + self.block_on(self.inner.delete_transfer(transfer_id)) .map_err(VnidropError::transfer) } pub fn delete_receive_history(&self) -> Result { - self.runtime - .block_on(self.inner.delete_receive_history()) + self.block_on(self.inner.delete_receive_history()) .map_err(VnidropError::repository) } @@ -261,8 +300,7 @@ impl VnidropCore { transfer_id: u64, mode: TransferAccessMode, ) -> Result<(), VnidropError> { - self.runtime - .block_on(self.inner.set_transfer_access_mode(transfer_id, mode)) + self.block_on(self.inner.set_transfer_access_mode(transfer_id, mode)) .map_err(VnidropError::permission) } @@ -271,20 +309,18 @@ impl VnidropCore { transfer_id: u64, endpoint_id: String, ) -> Result<(), VnidropError> { - self.runtime - .block_on( - self.inner - .approve_endpoint_for_transfer(transfer_id, endpoint_id), - ) - .map_err(VnidropError::permission) + self.block_on( + self.inner + .approve_endpoint_for_transfer(transfer_id, endpoint_id), + ) + .map_err(VnidropError::permission) } pub fn list_receiver_requests( &self, transfer_id: u64, ) -> Result, VnidropError> { - self.runtime - .block_on(self.inner.repository.list_receiver_requests(transfer_id)) + self.block_on(self.inner.repository.list_receiver_requests(transfer_id)) .map_err(VnidropError::repository) } @@ -294,20 +330,17 @@ impl VnidropCore { accepted: bool, reason: Option, ) -> Result<(), VnidropError> { - self.runtime - .block_on(self.inner.approval.respond(request_id, accepted, reason)) + self.block_on(self.inner.approval.respond(request_id, accepted, reason)) .map_err(VnidropError::permission) } pub fn list_transfers(&self) -> Result, VnidropError> { - self.runtime - .block_on(self.inner.repository.list_transfers()) + self.block_on(self.inner.repository.list_transfers()) .map_err(VnidropError::repository) } pub fn list_events(&self, transfer_id: Option) -> Result, VnidropError> { - self.runtime - .block_on(self.inner.list_events(transfer_id)) + self.block_on(self.inner.list_events(transfer_id)) .map_err(VnidropError::repository) } @@ -327,7 +360,7 @@ impl VnidropCore { } pub fn shutdown(&self) { - self.runtime.block_on(self.inner.shutdown()); + self.block_on(self.inner.shutdown()); } } @@ -439,7 +472,7 @@ impl CoreInner { transfer_slots: Semaphore::new(limits.max_concurrent_transfers as usize), limits, access_policy, - active_transfers: TokioMutex::new(HashMap::new()), + active_transfers: std::sync::Mutex::new(HashMap::new()), active_shares: TokioMutex::new(restored_active_shares), hash_to_transfer: TokioMutex::new(restored_hashes), connection_endpoints: TokioMutex::new(HashMap::new()), @@ -464,7 +497,7 @@ impl CoreInner { RuntimeStatus { endpoint_id: self.endpoint.id().to_string(), addr: format!("{:?}", self.endpoint.addr()), - active_transfers: self.active_transfers.lock().await.len() as u64, + active_transfers: self.active_transfers.lock().expect("active_transfers").len() as u64, active_shares: self.active_shares.lock().await.len() as u64, } } @@ -509,18 +542,24 @@ impl CoreInner { }) .await?; let (cancel, mut cancelled) = oneshot::channel(); - self.active_transfers.lock().await.insert( - transfer_id, - ActiveTransfer { - direction: TransferDirection::Send, - cancel, - }, - ); + self.active_transfers + .lock() + .expect("active_transfers") + .insert( + transfer_id, + ActiveTransfer { + direction: TransferDirection::Send, + cancel, + }, + ); let (result, was_cancelled) = tokio::select! { result = self.share_files_inner(sources, metadata) => (result, false), _ = &mut cancelled => (Err(anyhow::anyhow!("transfer cancelled")), true), }; - self.active_transfers.lock().await.remove(&transfer_id); + self.active_transfers + .lock() + .expect("active_transfers") + .remove(&transfer_id); if let Err(error) = &result { if was_cancelled { self.emit_transfer(transfer_id, "send", "lifecycle", "cancelled", json!({})); @@ -689,20 +728,26 @@ impl CoreInner { // Cancellation is cooperative: it stops our receive future and marks // local state while lower-level Iroh work unwinds naturally. let (shutdown_tx, mut shutdown_rx) = oneshot::channel(); - self.active_transfers.lock().await.insert( - transfer_id, - ActiveTransfer { - direction: TransferDirection::Receive, - cancel: shutdown_tx, - }, - ); + self.active_transfers + .lock() + .expect("active_transfers") + .insert( + transfer_id, + ActiveTransfer { + direction: TransferDirection::Receive, + cancel: shutdown_tx, + }, + ); let (result, cancelled) = tokio::select! { result = self.receive_inner(transfer_id, parsed, target, receiver_name) => (result, false), _ = &mut shutdown_rx => (Err(anyhow::anyhow!("transfer cancelled")), true), }; - self.active_transfers.lock().await.remove(&transfer_id); + self.active_transfers + .lock() + .expect("active_transfers") + .remove(&transfer_id); if let Err(error) = &result { if cancelled { self.emit_transfer(transfer_id, "receive", "lifecycle", "cancelled", json!({})); @@ -898,35 +943,20 @@ impl CoreInner { Ok(()) } - async fn cancel_transfer(&self, transfer_id: u64) -> Result<()> { - let mut active_transfers = self.active_transfers.lock().await; - if let Some(direction) = active_transfers - .get(&transfer_id) - .map(|active| active.direction) - { - let expected = match direction { - TransferDirection::Send => TransferStatus::Importing, - TransferDirection::Receive => TransferStatus::Receiving, - }; - self.repository - .transition_transfer_status(transfer_id, expected, TransferStatus::Cancelled) - .await?; - let active = active_transfers - .remove(&transfer_id) - .ok_or_else(|| anyhow::anyhow!("active transfer registration disappeared"))?; - drop(active_transfers); - let _ = active.cancel.send(()); - self.emit_transfer( - transfer_id, - direction.as_str(), - "lifecycle", - "cancel-requested", - json!({}), - ); - return Ok(()); - } - drop(active_transfers); + /// Remove an in-flight transfer and fire its cancel oneshot synchronously. + fn take_active_transfer(&self, transfer_id: u64) -> Option { + let active = self + .active_transfers + .lock() + .expect("active_transfers") + .remove(&transfer_id)?; + let direction = active.direction; + let _ = active.cancel.send(()); + Some(direction) + } + /// Stop a live share or report missing when no active transfer was found. + async fn cancel_idle_or_share(&self, transfer_id: u64) -> Result<()> { let mut active_shares = self.active_shares.lock().await; if active_shares.contains_key(&transfer_id) { self.repository @@ -953,7 +983,7 @@ impl CoreInner { if self .active_transfers .lock() - .await + .expect("active_transfers") .contains_key(&transfer_id) { anyhow::bail!("an active transfer must finish or be cancelled before deletion"); @@ -1389,6 +1419,10 @@ impl CoreInner { "exported": exported, }), ); + // Yield so outer `select!` cancel can land between chunks. + // Sink writes are synchronous (foreign FFI); without this, + // a single poll can drain the stream and miss cancellation. + tokio::task::yield_now().await; } ExportRangesItem::Error(error) => { anyhow::bail!("export failed for {relative_path}: {error}"); diff --git a/crates/vnidrop/tests/output_sink.rs b/crates/vnidrop/tests/output_sink.rs index e8512d8..79db032 100644 --- a/crates/vnidrop/tests/output_sink.rs +++ b/crates/vnidrop/tests/output_sink.rs @@ -68,24 +68,22 @@ fn reports_output_sink_write_failure() { #[test] fn cancellation_during_export_aborts_open_sink_file() { + // Gate the first write so export is mid-file when cancel runs. A large + // "slow writes" file made this flaky/hang-prone on CI: concurrent cancel + // used to nest Runtime::block_on, and multi-megabyte sleep loops could + // starve the cancel path for a long time. let source_dir = tempfile::tempdir().unwrap(); - let source_path = source_dir.path().join("slow.bin"); - std::fs::File::create(&source_path) - .unwrap() - .set_len(32 * 1024 * 1024) - .unwrap(); + let source_path = source_dir.path().join("gated.bin"); + std::fs::write(&source_path, vec![0u8; 256 * 1024]).unwrap(); let sender = TestNode::new(); let receiver = TestNode::new(); - let share = share_path(&sender.core, &source_path, 29, "slow.bin", false); - let output_sink = Arc::new(MemoryOutputSink::slow_writes(Duration::from_millis(25))); + let share = share_path(&sender.core, &source_path, 29, "gated.bin", false); + let output_sink = Arc::new(MemoryOutputSink::block_first_write()); let receiver_core = receiver.core.arc(); let sink_for_worker = output_sink.clone(); + let ticket = share.ticket.clone(); let worker = std::thread::spawn(move || { - receiver_core.receive_with_output_sink( - share.ticket, - sink_for_worker, - Some("receiver".to_string()), - ) + receiver_core.receive_with_output_sink(ticket, sink_for_worker, Some("receiver".to_string())) }); let request = wait_for_receiver_request(&sender.core, share.transfer_id); @@ -94,11 +92,35 @@ fn cancellation_during_export_aborts_open_sink_file() { .respond_receiver_request(request.id, true, None) .unwrap(); let started = Instant::now(); - while !output_sink.has_started("slow.bin") { - assert!(started.elapsed() < Duration::from_secs(15)); + while !output_sink.has_started("gated.bin") || !output_sink.has_entered_write() { + assert!( + started.elapsed() < Duration::from_secs(15), + "export never opened gated.bin for writing" + ); std::thread::sleep(Duration::from_millis(10)); } + // First write is parked in the sink. Cancel must complete from this thread + // without deadlocking against the receive worker's runtime driver. receiver.core.cancel_transfer(share.transfer_id).unwrap(); - assert!(worker.join().unwrap().is_err()); - assert_eq!(output_sink.terminal_state("slow.bin"), Some("aborted")); + // Unblock the in-flight write so the receive future can yield, observe + // cancel, and Drop(OutputSinkFile) can abort the open file. + output_sink.release_write_gate(); + + let result = worker + .join() + .expect("receive worker thread panicked") + .expect_err("cancelled receive should return an error"); + let message = result.to_string(); + assert!( + message.contains("cancel") || message.contains("interrupted") || !message.is_empty(), + "unexpected cancel error: {message}" + ); + let abort_deadline = Instant::now(); + while output_sink.terminal_state("gated.bin") != Some("aborted") { + assert!( + abort_deadline.elapsed() < Duration::from_secs(5), + "open sink file was not aborted after cancel" + ); + std::thread::sleep(Duration::from_millis(10)); + } } diff --git a/crates/vnidrop/tests/support/mod.rs b/crates/vnidrop/tests/support/mod.rs index 6989e9b..a820e50 100644 --- a/crates/vnidrop/tests/support/mod.rs +++ b/crates/vnidrop/tests/support/mod.rs @@ -6,7 +6,10 @@ use std::{ collections::HashMap, ops::Deref, path::Path, - sync::{Arc, Mutex}, + sync::{ + atomic::{AtomicBool, Ordering}, + Arc, Condvar, Mutex, + }, time::{Duration, Instant}, }; @@ -97,6 +100,9 @@ pub struct MemoryOutputSink { terminal: Mutex>, fail_writes: bool, write_delay: Duration, + /// When set, every write waits until [Self::release_write_gate] opens the gate. + write_gate: Option, Condvar)>>, + write_entered: AtomicBool, } impl MemoryOutputSink { @@ -106,6 +112,8 @@ impl MemoryOutputSink { terminal: Mutex::new(HashMap::new()), fail_writes: true, write_delay: Duration::ZERO, + write_gate: None, + write_entered: AtomicBool::new(false), } } @@ -115,6 +123,32 @@ impl MemoryOutputSink { terminal: Mutex::new(HashMap::new()), fail_writes: false, write_delay: delay, + write_gate: None, + write_entered: AtomicBool::new(false), + } + } + + /// Park every `write_chunk` until [Self::release_write_gate] is called. + /// + /// Used to hold export mid-file so cancel can run from another thread + /// without racing a multi-megabyte slow-write loop. + pub fn block_first_write() -> Self { + Self { + files: Mutex::new(HashMap::new()), + terminal: Mutex::new(HashMap::new()), + fail_writes: false, + write_delay: Duration::ZERO, + write_gate: Some(Arc::new((Mutex::new(false), Condvar::new()))), + write_entered: AtomicBool::new(false), + } + } + + pub fn release_write_gate(&self) { + if let Some(gate) = &self.write_gate { + let (lock, cvar) = gate.as_ref(); + let mut open = lock.lock().unwrap(); + *open = true; + cvar.notify_all(); } } @@ -129,6 +163,10 @@ impl MemoryOutputSink { pub fn has_started(&self, relative_path: &str) -> bool { self.files.lock().unwrap().contains_key(relative_path) } + + pub fn has_entered_write(&self) -> bool { + self.write_entered.load(Ordering::SeqCst) + } } impl ReceiveOutputSink for MemoryOutputSink { @@ -138,6 +176,14 @@ impl ReceiveOutputSink for MemoryOutputSink { } fn write_chunk(&self, relative_path: String, bytes: Vec) -> Result<(), VnidropError> { + self.write_entered.store(true, Ordering::SeqCst); + if let Some(gate) = &self.write_gate { + let (lock, cvar) = gate.as_ref(); + let mut open = lock.lock().unwrap(); + while !*open { + open = cvar.wait(open).unwrap(); + } + } if !self.write_delay.is_zero() { std::thread::sleep(self.write_delay); }