From 50c3da2f5d4b507ae388bcd3cb5ed9ade1a5061b Mon Sep 17 00:00:00 2001 From: NimBold Date: Fri, 17 Jul 2026 01:03:52 +0330 Subject: [PATCH] fix(downloads): close media lifecycle cleanup gaps Validate every Aria2 mirror URI, keep metadata cleanup errors truthful, and serialize media registration with lifecycle cancellation so abandoned tombstones cannot accumulate. --- src-tauri/src/download.rs | 39 +++++++++++-- src-tauri/src/lib.rs | 115 ++++++++++++++++++++++++++------------ src-tauri/src/queue.rs | 14 +++++ 3 files changed, 128 insertions(+), 40 deletions(-) diff --git a/src-tauri/src/download.rs b/src-tauri/src/download.rs index 9a2eb44..a3eb0ac 100644 --- a/src-tauri/src/download.rs +++ b/src-tauri/src/download.rs @@ -77,7 +77,7 @@ impl DownloadCoordinator { &self, id: String, lifecycle_generation: u64, - ack: tokio::sync::oneshot::Sender<()>, + ack: tokio::sync::oneshot::Sender, ) -> Result<(), String> { self.media_tx .send(MediaCmd::PauseWithAck { @@ -128,7 +128,7 @@ enum MediaCmd { PauseWithAck { id: String, lifecycle_generation: u64, - ack: tokio::sync::oneshot::Sender<()>, + ack: tokio::sync::oneshot::Sender, }, Finished { id: String, @@ -144,7 +144,7 @@ async fn run_coordinator( let mut active_media = HashMap::)>::new(); let mut cancelled_media_generations = HashMap::::new(); let mut pending_media_acks = - HashMap::<(String, u64), tokio::sync::oneshot::Sender<()>>::new(); + HashMap::<(String, u64), tokio::sync::oneshot::Sender>::new(); let mut pending_captured_urls = Vec::::new(); let mut frontend_ready = false; let mut command_open = true; @@ -228,7 +228,10 @@ async fn run_coordinator( .entry(id) .and_modify(|generation| *generation = (*generation).max(lifecycle_generation)) .or_insert(lifecycle_generation); - let _ = ack.send(()); + // No runner was registered for this generation. + // The caller may retire the cancellation tombstone + // while it still owns the per-download lock. + let _ = ack.send(false); } } MediaCmd::Finished { id, lifecycle_generation } => { @@ -238,7 +241,9 @@ async fn run_coordinator( if let Some(ack) = pending_media_acks.remove(&(id.clone(), lifecycle_generation)) { - let _ = ack.send(()); + // A pending acknowledgement means the runner was + // registered and has now observed cancellation. + let _ = ack.send(true); } if cancelled_media_generations .get(&id) @@ -404,6 +409,30 @@ mod tests { .expect("late media registration was not cancelled"); } + #[tokio::test] + async fn unregistered_media_pause_tombstone_can_be_reconciled() { + let (coordinator, _events) = DownloadCoordinator::spawn_headless(); + let (ack_tx, ack_rx) = tokio::sync::oneshot::channel(); + + coordinator + .pause_media_with_ack("abandoned-media".to_string(), 9, ack_tx) + .await + .unwrap(); + assert!(!ack_rx.await.unwrap()); + + // Once the queue lifecycle is invalidated under its control lock, + // it can retire the tombstone without allowing a late runner to + // start. A later registration must therefore remain uncancelled. + coordinator + .finish_media("abandoned-media".to_string(), 9) + .await; + let cancel_rx = coordinator + .register_media("abandoned-media".to_string(), 9) + .await + .unwrap(); + assert!(!*cancel_rx.borrow()); + } + #[tokio::test] async fn stale_media_registration_cannot_replace_a_newer_lifecycle() { let (coordinator, _events) = DownloadCoordinator::spawn_headless(); diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index c2c104c..ca42ad7 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -1928,6 +1928,15 @@ struct ShellCommandOutput { stderr: Vec, } +fn terminate_shell_process_tree(child: tauri_plugin_shell::process::CommandChild) { + crate::process::kill_process_tree(child.pid()); + // The process-tree helper may already have signalled the root. The + // direct kill is still needed when the root was not visible in the + // snapshot, but its result is cleanup telemetry, not the operation's + // primary error. + let _ = child.kill(); +} + async fn shell_command_output_with_timeout( command: tauri_plugin_shell::process::Command, timeout: Duration, @@ -1988,33 +1997,21 @@ async fn shell_command_output_with_timeout( match tokio::time::timeout(timeout, collect_output).await { Ok(Ok(output)) if output.status_code.is_some() => Ok(output), - Ok(Ok(_)) => match child.kill() { - Ok(()) => Err(format!( - "{operation} ended without a process exit status" - )), - Err(error) => Err(format!( - "{operation} ended without a process exit status and failed to terminate the child process: {error}" - )), - }, - Ok(Err(error)) => { - let kill_error = child.kill().err(); - match kill_error { - Some(kill_error) => Err(format!( - "{error}; failed to terminate the child process: {kill_error}" - )), - None => Err(error), - } + Ok(Ok(_)) => { + terminate_shell_process_tree(child); + Err(format!("{operation} ended without a process exit status")) } - Err(_) => match child.kill() { - Ok(()) => Err(format!( + Ok(Err(error)) => { + terminate_shell_process_tree(child); + Err(error) + } + Err(_) => { + terminate_shell_process_tree(child); + Err(format!( "{operation} timed out after {}s", timeout.as_secs() - )), - Err(error) => Err(format!( - "{operation} timed out after {}s and failed to terminate the child process: {error}", - timeout.as_secs() - )), - }, + )) + } } } @@ -3770,8 +3767,7 @@ pub(crate) async fn start_media_download_internal( let failure_reason = loop { tokio::select! { _ = cancel_rx.changed() => { - crate::process::kill_process_tree(child.pid()); - let _ = child.kill(); + terminate_shell_process_tree(child); if processing_started { cleanup_media_processing_artifacts(&out_path).await; } @@ -4148,10 +4144,20 @@ async fn pause_download( .pause_media_with_ack(id.clone(), media_lifecycle_generation, tx) .await?; } else { - let _ = tx.send(()); + let _ = tx.send(false); } - rx.await + let media_was_registered = rx + .await .map_err(|_| "download worker stopped without acknowledging pause".to_string())?; + if matches!(active_kind, Some(crate::queue::TaskKind::Media)) && !media_was_registered { + // No media runner reached the coordinator. The queue control lock is + // still held, and run_media now checks the same lifecycle generation + // before registration, so this tombstone can be retired safely. + state + .download_coordinator + .finish_media(id.clone(), media_lifecycle_generation) + .await; + } state.queue_manager.release_permit(&id).await; if registered_lifecycle_generation.is_some() @@ -4514,9 +4520,15 @@ async fn remove_download( .pause_media_with_ack(id.clone(), media_lifecycle_generation, tx) .await?; } else { - let _ = tx.send(()); + let _ = tx.send(false); + } + let media_was_registered = rx.await.unwrap_or(false); + if matches!(active_kind, Some(crate::queue::TaskKind::Media)) && !media_was_registered { + state + .download_coordinator + .finish_media(id.clone(), media_lifecycle_generation) + .await; } - let _ = rx.await; state.queue_manager.release_permit(&id).await; state.queue_manager.clear_aria2_retry_state(&id).await; @@ -4682,9 +4694,15 @@ async fn detach_download_for_reconfigure( .pause_media_with_ack(id.clone(), media_lifecycle_generation, tx) .await?; } else { - let _ = tx.send(()); // Fallback if no task exists + let _ = tx.send(false); // Fallback if no task exists + } + let media_was_registered = rx.await.unwrap_or(false); // Wait for the writer to stop + if matches!(active_kind, Some(crate::queue::TaskKind::Media)) && !media_was_registered { + state + .download_coordinator + .finish_media(id.clone(), media_lifecycle_generation) + .await; } - let _ = rx.await; // Wait for the writer to stop state.queue_manager.release_permit(&id).await; state.queue_manager.clear_aria2_retry_state(&id).await; @@ -5101,13 +5119,20 @@ async fn validate_enqueue_url(url: &str) -> Result<(), String> { } } +async fn validate_enqueue_uris(url: &str, mirrors: Option<&str>) -> Result<(), String> { + for uri in collect_download_uris(url, mirrors) { + validate_enqueue_url(&uri).await?; + } + Ok(()) +} + #[tauri::command] async fn enqueue_download( app_handle: tauri::AppHandle, state: tauri::State<'_, AppState>, mut item: queue::EnqueueItem, ) -> Result { - validate_enqueue_url(&item.url) + validate_enqueue_uris(&item.url, item.mirrors.as_deref()) .await .map_err(AppError::Internal)?; let id = item.id.clone(); @@ -5174,7 +5199,7 @@ async fn enqueue_many( let mut results = Vec::with_capacity(items.len()); for mut item in items { let id = item.id.clone(); - if let Err(error) = validate_enqueue_url(&item.url).await { + if let Err(error) = validate_enqueue_uris(&item.url, item.mirrors.as_deref()).await { results.push(crate::ipc::EnqueueResult { id, success: false, @@ -6199,7 +6224,7 @@ mod tests { observe_aria2_connections, observe_aria2_connections_with_epoch, Aria2ConnectionObservation, Aria2ConnectionSample, Aria2RecoveryReason, parse_media_playlist_metadata, - validate_enqueue_url, + validate_enqueue_url, validate_enqueue_uris, }; use serde_json::json; use std::time::{Duration, Instant}; @@ -6228,6 +6253,26 @@ mod tests { ); } + #[tokio::test] + async fn enqueue_uri_validation_covers_mirrors_not_only_the_primary_url() { + assert_eq!( + validate_enqueue_uris( + "http://192.0.2.1/file.zip", + Some("http://127.0.0.1/internal\nfile:///etc/passwd"), + ) + .await, + Err("SSRF blocked: Private/local IP not allowed".to_string()) + ); + assert_eq!( + validate_enqueue_uris( + "http://192.0.2.1/file.zip", + Some("file:///etc/passwd"), + ) + .await, + Err("Unsupported URL scheme".to_string()) + ); + } + #[test] fn slow_nonzero_aria2_throughput_recovers_after_a_sustained_degradation() { let start = Instant::now(); diff --git a/src-tauri/src/queue.rs b/src-tauri/src/queue.rs index 18ea852..d442f46 100644 --- a/src-tauri/src/queue.rs +++ b/src-tauri/src/queue.rs @@ -2076,10 +2076,24 @@ impl SidecarSpawner for ProductionSpawner { lifecycle_generation: u64, ) -> Result<(), String> { let state = self.app_handle.state::(); + // Serialize registration with pause/remove/detach. If the queue + // lifecycle was invalidated before this worker reached the + // coordinator, do not create a late media registration that could + // outlive the row's permit and ownership. + let control_guard = state.queue_manager.acquire_aria2_control(id).await; + if !state + .queue_manager + .is_registered_generation(id, lifecycle_generation) + .await + { + drop(control_guard); + return Err(crate::queue::MEDIA_RUN_CANCELLED.to_string()); + } let mut cancel_rx = state .download_coordinator .register_media(id.to_string(), lifecycle_generation) .await?; + drop(control_guard); let outcome = if *cancel_rx.borrow() { Err(crate::queue::MEDIA_RUN_CANCELLED.to_string()) } else {