diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 10b96f5..418f3ec 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -4884,16 +4884,22 @@ async fn remove_download( ) -> Result<(), String> { log::info!("remove_download called for id: {}", id); let preserve_resumable = preserve_resumable.unwrap_or(false); - let mut owned_paths = crate::download_ownership::owned_paths_for_id(&app_handle, &id)?; - let _control_guard = state.queue_manager.acquire_aria2_control(&id).await; + let control_guard = state.queue_manager.acquire_aria2_control(&id).await; let active_kind = state.queue_manager.active_kind(&id).await; - let media_lifecycle_generation = state + let registered_lifecycle_generation = state .queue_manager .registered_lifecycle_generation(&id) - .await - .unwrap_or_default(); - state.queue_manager.remove_from_pending(&id).await; + .await; + let media_lifecycle_generation = registered_lifecycle_generation.unwrap_or_default(); + if let Some(generation) = registered_lifecycle_generation { + state + .queue_manager + .remove_from_pending_for_generation(&id, generation) + .await; + } else { + state.queue_manager.remove_from_pending(&id).await; + } let gid = state.queue_manager.aria2_gid_for_download(&id); if let Some(gid) = gid.as_deref() { @@ -4909,7 +4915,6 @@ async fn remove_download( error ); } - owned_paths = crate::download_ownership::owned_paths_for_id(&app_handle, &id)?; // Keep the current epoch and mapping alive until daemon removal is // confirmed. If removal fails, terminal events from this lifecycle // must still be accepted so the permit and mapping can be cleaned up. @@ -4942,7 +4947,7 @@ async fn remove_download( // There may be an addUri or retry handoff in flight with no mapped // GID yet. Invalidate that pending lifecycle before acknowledging the // removal so its late result cannot resurrect the download. - state.queue_manager.next_aria2_control_epoch(&id).await; + let removal_epoch = state.queue_manager.next_aria2_control_epoch(&id).await; state.queue_manager.cancel_aria2_retries(&id).await; let (tx, rx) = tokio::sync::oneshot::channel(); if matches!(active_kind, Some(crate::queue::TaskKind::Media)) { @@ -4961,11 +4966,79 @@ async fn remove_download( .await; } + // The initial addUri call intentionally runs outside the control lock. + // Do not delete the guessed magnet path and ownership record until a + // late GID has been removed; otherwise the daemon can finish creating + // an output after cleanup and leave an untracked file behind. + drop(control_guard); + state.queue_manager.wait_for_aria2_dispatch(&id).await; + let _control_guard = state.queue_manager.acquire_aria2_control(&id).await; + let current_generation = state + .queue_manager + .registered_lifecycle_generation(&id) + .await; + let lifecycle_changed = registered_lifecycle_generation.is_some() + && current_generation != registered_lifecycle_generation + || registered_lifecycle_generation.is_none() && current_generation.is_some(); + if lifecycle_changed { + state + .queue_manager + .release_permit_for_generation(&id, media_lifecycle_generation) + .await; + return Err("download lifecycle changed while waiting for aria2 dispatch".to_string()); + } + if let Some(late_gid) = state.queue_manager.aria2_gid_for_download(&id) { + if let Err(error) = state + .queue_manager + .reconcile_aria2_torrent_ownership(&id, &late_gid) + .await + { + log::debug!( + "aria2 remove [{}]: could not resolve files for late gid {}: {}", + id, + late_gid, + error + ); + } + let removal_result = async { + force_remove_aria2_gid( + state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), + &state.aria2_secret, + &late_gid, + ) + .await?; + wait_for_aria2_stopped( + state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), + &state.aria2_secret, + &late_gid, + ) + .await + } + .await; + if let Err(error) = removal_result { + state.queue_manager.allow_aria2_retries(&id).await; + return Err(error); + } + state.queue_manager.next_aria2_control_epoch(&id).await; + state.queue_manager.clear_aria2_retry_state(&id).await; + state.queue_manager.forget_aria2_gid(&id).await; + } else if !state + .queue_manager + .is_aria2_control_epoch_current(&id, removal_epoch) + .await + { + // A same-generation pause/resume control action may advance the + // epoch while the late add is being retired. Removal still owns + // the registered row, so fence that action before cleaning it. + state.queue_manager.next_aria2_control_epoch(&id).await; + } state.queue_manager.release_permit(&id).await; state.queue_manager.clear_aria2_retry_state(&id).await; state.queue_manager.forget_aria2_gid(&id).await; } + let owned_paths = crate::download_ownership::owned_paths_for_id(&app_handle, &id)?; + use tauri::Emitter; let _ = app_handle.emit( "download-state", @@ -5545,6 +5618,10 @@ async fn validate_torrent_enqueue( let bytes = std::fs::read(path) .map_err(|error| format!("could not read cached torrent metadata: {error}"))?; let metadata = crate::torrent::parse_torrent_bytes(&bytes)?; + crate::torrent::validate_info_hash( + item.torrent_info_hash.as_deref(), + &metadata.info_hash, + )?; crate::torrent::validate_selected_indices( item.torrent_file_indices.as_deref(), metadata.files.len(), @@ -5559,7 +5636,8 @@ async fn validate_torrent_enqueue( if item.torrent_file_indices.is_some() { return Err("magnet file selection requires resolved torrent metadata".to_string()); } - crate::torrent::inspect_source(&item.url).map(|_| ()) + let metadata = crate::torrent::inspect_source(&item.url)?; + crate::torrent::validate_info_hash(item.torrent_info_hash.as_deref(), &metadata.info_hash) } fn expected_torrent_output_paths( diff --git a/src-tauri/src/queue.rs b/src-tauri/src/queue.rs index ff4fa64..a80df57 100644 --- a/src-tauri/src/queue.rs +++ b/src-tauri/src/queue.rs @@ -298,6 +298,11 @@ pub struct QueueManager { /// download id -> spawn payload for aria2 transient-error re-addUri retries. aria2_payloads: Mutex>, + /// Initial aria2 addUri handoffs that have not yet either published a GID + /// or removed a stale late GID. Removal waits for these handoffs before + /// deleting owned assets so a magnet cannot leave an orphaned output. + aria2_dispatch_inflight: Mutex>>, + aria2_dispatch_notify: Notify, /// The daemon-wide download cap currently applied to aria2. This mirrors /// successful RPC changes so the poller can avoid treating an intentional @@ -376,6 +381,8 @@ impl QueueManager { aria2_gids: Arc::new(std::sync::RwLock::new(HashMap::new())), pending_completion: Arc::new(Mutex::new(HashMap::new())), aria2_payloads: Mutex::new(HashMap::new()), + aria2_dispatch_inflight: Mutex::new(HashMap::new()), + aria2_dispatch_notify: Notify::new(), aria2_global_speed_limit: Arc::new(StdMutex::new(None)), aria2_retry_strikes: Mutex::new(HashMap::new()), aria2_retry_cancelled: Mutex::new(HashSet::new()), @@ -1203,7 +1210,7 @@ impl QueueManager { } } - async fn release_permit_for_generation(&self, id: &str, generation: u64) { + pub(crate) async fn release_permit_for_generation(&self, id: &str, generation: u64) { let _admission_gate = self.admission_gate.lock().await; let removed = { let mut permits = self.active_permits.lock().await; @@ -1392,6 +1399,9 @@ impl QueueManager { } else { None }; + if let Some(epoch) = aria2_lifecycle_epoch { + self.begin_aria2_dispatch(&id, epoch).await; + } self.emit_state(&id, DownloadStatus::Downloading); drop(control_guard); @@ -1399,7 +1409,8 @@ impl QueueManager { TaskKind::Aria2 => { let lifecycle_epoch = aria2_lifecycle_epoch .expect("aria2 dispatch must initialize a control epoch"); - match self.spawner.add_uri(&id, &task.payload).await { + let add_result = self.spawner.add_uri(&id, &task.payload).await; + match add_result { Ok(gid) => { let control_guard = self.acquire_aria2_control(&id).await; let cancelled = self.aria2_retry_cancelled.lock().await.contains(&id); @@ -1414,6 +1425,23 @@ impl QueueManager { id, gid ); + if task.payload.is_torrent { + if let Err(error) = self + .reconcile_aria2_torrent_ownership_for_payload( + &id, + &gid, + &task.payload, + ) + .await + { + log::debug!( + "aria2 dispatch cancellation [{}]: could not resolve torrent output paths for gid {}: {}", + id, + gid, + error + ); + } + } if let Err(error) = self.spawner.remove_uri(&gid).await { log::warn!( "aria2 dispatch cancellation [{}]: failed to remove late gid {}: {}", @@ -1423,6 +1451,7 @@ impl QueueManager { ); } self.ignore_aria2_gid(&gid).await; + self.finish_aria2_dispatch(&id, lifecycle_epoch).await; if current_lifecycle { self.clear_aria2_retry_state(&id).await; self.release_permit(&id).await; @@ -1430,6 +1459,7 @@ impl QueueManager { return; } let buffered_outcome = self.remember_gid(id.clone(), gid.clone()).await; + self.finish_aria2_dispatch(&id, lifecycle_epoch).await; drop(control_guard); if let Some(outcome) = buffered_outcome { self.handle_aria2_event(&gid, outcome).await; @@ -1454,6 +1484,7 @@ impl QueueManager { id ); } + self.finish_aria2_dispatch(&id, lifecycle_epoch).await; } } } @@ -1755,23 +1786,55 @@ impl QueueManager { .find_map(|(gid, mapping)| (mapping.id == id).then(|| gid.clone())) } - /// Refresh ownership from Aria2's resolved file list before a torrent - /// lifecycle is forgotten. Magnet metadata is not available when the row - /// is enqueued, so the user-provided display name is not a safe ownership - /// path until Aria2 reports the actual files. - pub async fn reconcile_aria2_torrent_ownership( + async fn begin_aria2_dispatch(&self, id: &str, epoch: u64) { + self.aria2_dispatch_inflight + .lock() + .await + .entry(id.to_string()) + .or_default() + .insert(epoch); + } + + async fn finish_aria2_dispatch(&self, id: &str, epoch: u64) { + let mut inflight = self.aria2_dispatch_inflight.lock().await; + let removed = if let Some(epochs) = inflight.get_mut(id) { + let removed = epochs.remove(&epoch); + if epochs.is_empty() { + inflight.remove(id); + } + removed + } else { + false + }; + drop(inflight); + if removed { + self.aria2_dispatch_notify.notify_waiters(); + } + } + + pub async fn wait_for_aria2_dispatch(&self, id: &str) { + loop { + let notified = self.aria2_dispatch_notify.notified(); + tokio::pin!(notified); + notified.as_mut().enable(); + if !self + .aria2_dispatch_inflight + .lock() + .await + .contains_key(id) + { + return; + } + notified.await; + } + } + + async fn reconcile_aria2_torrent_ownership_for_payload( &self, id: &str, gid: &str, + payload: &SpawnPayload, ) -> Result<(), String> { - let payload = self.aria2_payloads.lock().await.get(id).cloned(); - let Some(payload) = payload.filter(|payload| payload.is_torrent) else { - return Ok(()); - }; - let mapping = self - .aria2_gid_mapping(gid) - .filter(|mapping| mapping.id == id) - .ok_or_else(|| "aria2 torrent ownership has no current gid mapping".to_string())?; let state = self.app_handle.state::(); let result = crate::rpc_call( state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), @@ -1780,14 +1843,15 @@ impl QueueManager { serde_json::json!([gid]), ) .await?; - if !self.is_current_aria2_gid_mapping(gid, &mapping) - || !self - .is_aria2_control_epoch_current(id, mapping.epoch) - .await - { - return Err("aria2 torrent lifecycle changed while resolving output paths".to_string()); - } + self.persist_aria2_torrent_ownership(id, payload, result) + } + fn persist_aria2_torrent_ownership( + &self, + id: &str, + payload: &SpawnPayload, + result: serde_json::Value, + ) -> Result<(), String> { let paths = result .as_array() .into_iter() @@ -1841,6 +1905,41 @@ impl QueueManager { ) } + /// Refresh ownership from Aria2's resolved file list before a torrent + /// lifecycle is forgotten. Magnet metadata is not available when the row + /// is enqueued, so the user-provided display name is not a safe ownership + /// path until Aria2 reports the actual files. + pub async fn reconcile_aria2_torrent_ownership( + &self, + id: &str, + gid: &str, + ) -> Result<(), String> { + let payload = self.aria2_payloads.lock().await.get(id).cloned(); + let Some(payload) = payload.filter(|payload| payload.is_torrent) else { + return Ok(()); + }; + let mapping = self + .aria2_gid_mapping(gid) + .filter(|mapping| mapping.id == id) + .ok_or_else(|| "aria2 torrent ownership has no current gid mapping".to_string())?; + let state = self.app_handle.state::(); + let result = crate::rpc_call( + state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), + &state.aria2_secret, + "aria2.getFiles", + serde_json::json!([gid]), + ) + .await?; + if !self.is_current_aria2_gid_mapping(gid, &mapping) + || !self + .is_aria2_control_epoch_current(id, mapping.epoch) + .await + { + return Err("aria2 torrent lifecycle changed while resolving output paths".to_string()); + } + self.persist_aria2_torrent_ownership(id, &payload, result) + } + /// Capture the GID ownership token used by the poller. The mapping's /// epoch must still be current when an asynchronous status snapshot is /// emitted; otherwise a late response from an older GID can be attributed @@ -2467,6 +2566,17 @@ impl QueueManager { } removed } + + pub async fn remove_from_pending_for_generation(&self, id: &str, generation: u64) -> bool { + let mut pending = self.pending.lock().await; + let before = pending.len(); + pending.retain(|task| !(task.id == id && task.lifecycle_generation == generation)); + let removed = pending.len() < before; + if removed { + self.notify.notify_one(); + } + removed + } } fn automatic_retry_limit(max_tries: Option) -> usize { @@ -2950,7 +3060,14 @@ impl SidecarSpawner for ProductionSpawner { ) .await?; match result.as_str() { - Some(returned_gid) if returned_gid == gid => Ok(()), + Some(returned_gid) if returned_gid == gid => { + crate::wait_for_aria2_stopped( + state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), + &state.aria2_secret, + gid, + ) + .await + } Some(returned_gid) => Err(format!( "aria2.forceRemove returned unexpected gid {returned_gid}, expected {gid}" )), diff --git a/src-tauri/src/torrent.rs b/src-tauri/src/torrent.rs index 6833658..f261b23 100644 --- a/src-tauri/src/torrent.rs +++ b/src-tauri/src/torrent.rs @@ -292,6 +292,55 @@ fn parse_info(info: &BencodeValue) -> Result { Ok(ParsedTorrent { name, total_bytes, files, info_hash }) } +fn canonical_btih(value: &str) -> Option { + let normalized = value.trim().to_ascii_lowercase(); + if normalized.len() == 40 && normalized.bytes().all(|byte| byte.is_ascii_hexdigit()) { + return Some(normalized); + } + if normalized.len() != 32 { + return None; + } + + let mut decoded = Vec::with_capacity(20); + let mut accumulator = 0u64; + let mut bits = 0u8; + for byte in normalized.bytes() { + let value = match byte { + b'a'..=b'z' => byte - b'a', + b'2'..=b'7' => byte - b'2' + 26, + _ => return None, + } as u64; + accumulator = (accumulator << 5) | value; + bits += 5; + while bits >= 8 { + bits -= 8; + decoded.push(((accumulator >> bits) & 0xff) as u8); + } + if bits == 0 { + accumulator = 0; + } else { + accumulator &= (1u64 << bits) - 1; + } + } + if bits != 0 || decoded.len() != 20 { + return None; + } + + Some(decoded.iter().map(|byte| format!("{byte:02x}")).collect()) +} + +pub fn validate_info_hash(expected: Option<&str>, actual: &str) -> Result<(), String> { + let Some(expected) = expected else { + return Ok(()); + }; + let expected = canonical_btih(expected) + .ok_or_else(|| "torrent metadata has an invalid expected info hash".to_string())?; + if expected != actual { + return Err("torrent metadata changed before it was queued".to_string()); + } + Ok(()) +} + pub fn parse_torrent_bytes(bytes: &[u8]) -> Result { if bytes.is_empty() || bytes.len() > MAX_TORRENT_BYTES { return Err(format!("torrent metadata must be between 1 byte and {MAX_TORRENT_BYTES} bytes")); @@ -319,17 +368,7 @@ fn magnet_metadata(source: &str) -> Result { return None; } let value = value.strip_prefix("urn:btih:")?; - let normalized = value.trim().to_ascii_lowercase(); - if (normalized.len() == 40 && normalized.bytes().all(|byte| byte.is_ascii_hexdigit())) - || (normalized.len() == 32 - && normalized - .bytes() - .all(|byte| byte.is_ascii_lowercase() || matches!(byte, b'2'..=b'7'))) - { - Some(normalized) - } else { - None - } + canonical_btih(value) }) .ok_or_else(|| "magnet URI has no valid BitTorrent info hash".to_string())?; let name = parsed @@ -542,6 +581,30 @@ mod tests { assert!(parsed.files.is_empty()); } + #[test] + fn canonicalizes_base32_magnet_hashes_to_hex() { + let parsed = inspect_source( + "magnet:?xt=urn:btih:AERUKZ4JVPG66AJDIVTYTK6N54ASGRLH&dn=Base32", + ) + .expect("a valid Base32 hash should parse"); + assert_eq!(parsed.info_hash, "0123456789abcdef0123456789abcdef01234567"); + } + + #[test] + fn validates_expected_hashes_across_hex_and_base32_encodings() { + validate_info_hash( + Some("AERUKZ4JVPG66AJDIVTYTK6N54ASGRLH"), + "0123456789abcdef0123456789abcdef01234567", + ) + .expect("equivalent Base32 and hexadecimal hashes should match"); + assert!(validate_info_hash( + Some("0123456789abcdef0123456789abcdef01234567"), + "fedcba9876543210fedcba9876543210fedcba98", + ) + .is_err()); + validate_info_hash(None, "not-used").expect("missing legacy identity should remain compatible"); + } + #[test] fn rejects_malformed_base32_magnet_hashes() { let error = inspect_source( @@ -549,6 +612,12 @@ mod tests { ) .expect_err("a 32-character hash with non-base32 characters must be rejected"); assert!(error.contains("valid BitTorrent info hash")); + + let error = inspect_source( + "magnet:?xt=urn:btih:00000000000000000000000000000000&dn=Invalid", + ) + .expect_err("characters outside RFC 4648 Base32 must be rejected"); + assert!(error.contains("valid BitTorrent info hash")); } #[test] diff --git a/src-tauri/tests/queue_manager.rs b/src-tauri/tests/queue_manager.rs index 4c0570a..a4ef29b 100644 --- a/src-tauri/tests/queue_manager.rs +++ b/src-tauri/tests/queue_manager.rs @@ -2043,7 +2043,9 @@ async fn late_aria2_gid_after_cancellation_is_removed_without_leaking_permit() { manager.release_registered_id("late").await; manager.release_permit("late").await; - tokio::time::sleep(Duration::from_millis(100)).await; + timeout(Duration::from_secs(1), manager.wait_for_aria2_dispatch("late")) + .await + .expect("late dispatch must be fully removed before it is considered finished"); assert!(manager.aria2_gid_for_download("late").is_none()); assert_eq!(manager.available_permits(), 1);