diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 7eed208..68aacde 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -2918,6 +2918,7 @@ mod platform; pub mod queue; pub mod process; pub mod retry; +mod torrent_probe; pub mod torrent; mod settings; mod storage; @@ -3275,6 +3276,22 @@ pub(crate) async fn rpc_call( .ok_or_else(|| "aria2 returned no result".to_string()) } +struct Aria2RpcClient { + port: u16, + secret: String, +} + +#[async_trait::async_trait] +impl crate::torrent_probe::RpcClient for Aria2RpcClient { + async fn call( + &self, + method: &str, + params: serde_json::Value, + ) -> Result { + rpc_call(self.port, &self.secret, method, params).await + } +} + #[tauri::command] async fn test_aria2c( app_handle: tauri::AppHandle, @@ -5688,6 +5705,17 @@ fn expected_torrent_output_paths( Ok(Some(paths)) } +async fn remove_magnet_metadata_probe_dir(path: &std::path::Path) -> Result<(), String> { + match tokio::fs::remove_dir_all(path).await { + Ok(()) => Ok(()), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(format!( + "could not remove magnet metadata probe ({:?})", + error.kind() + )), + } +} + async fn resolve_magnet_metadata( app_handle: &tauri::AppHandle, state: &AppState, @@ -5698,6 +5726,10 @@ async fn resolve_magnet_metadata( ) -> Result { let expected = crate::torrent::inspect_source(source)?; let managed_path = crate::torrent::managed_torrent_path(app_handle, id)?; + let proxy_value = proxy + .map(crate::queue::aria2_all_proxy_value) + .transpose()? + .flatten(); let storage_root = managed_path .parent() .ok_or_else(|| "torrent storage has no parent directory".to_string())?; @@ -5711,140 +5743,61 @@ async fn resolve_magnet_metadata( .load(std::sync::atomic::Ordering::Relaxed); let secret = state.aria2_secret.clone(); let metadata_path = probe_dir.join(format!("{}.torrent", expected.info_hash)); - let mut gid = None; - let metadata_result = async { - let mut options = serde_json::Map::new(); - options.insert( - "dir".to_string(), - serde_json::json!(probe_dir.to_string_lossy().to_string()), - ); - options.insert("bt-metadata-only".to_string(), serde_json::json!("true")); - options.insert("bt-save-metadata".to_string(), serde_json::json!("true")); - options.insert("max-tries".to_string(), serde_json::json!("3")); - options.insert("retry-wait".to_string(), serde_json::json!("2")); - options.insert("connect-timeout".to_string(), serde_json::json!("20")); - options.insert("timeout".to_string(), serde_json::json!("60")); - options.insert("auto-file-renaming".to_string(), serde_json::json!("false")); - if let Some(proxy) = proxy { - if let Some(proxy) = crate::queue::aria2_all_proxy_value(proxy)? { - options.insert("all-proxy".to_string(), serde_json::json!(proxy)); - } - } - - let result = crate::rpc_call( - port, - &secret, - "aria2.addUri", - serde_json::json!([[source], options]), - ) - .await - .map_err(|error| { - format!( - "Aria2 could not start magnet metadata resolution: {}", - crate::redact_sensitive_text(&error) - ) - })?; - let added_gid = result - .as_str() - .filter(|value| !value.is_empty()) - .ok_or_else(|| "Aria2 returned an empty metadata probe GID".to_string())? - .to_string(); - gid = Some(added_gid.clone()); - - let deadline = Instant::now() + Duration::from_secs(60); - loop { - let status = match crate::rpc_call( - port, - &secret, - "aria2.tellStatus", - serde_json::json!([&added_gid, ["status", "errorCode", "errorMessage"]]), - ) - .await - { - Ok(status) => status, - Err(error) if aria2_gid_not_found(&error) => { - return Err( - "Aria2 removed the magnet metadata probe before metadata was saved" - .to_string(), - ); - } - Err(error) => { - return Err(format!( - "Aria2 metadata resolution status failed: {}", - crate::redact_sensitive_text(&error) - )); - } - }; - match status.get("status").and_then(serde_json::Value::as_str) { - Some("complete") => break, - Some("error") | Some("removed") => { - let error_code = status - .get("errorCode") - .and_then(serde_json::Value::as_str) - .filter(|value| !value.is_empty()); - let error_message = status - .get("errorMessage") - .and_then(serde_json::Value::as_str) - .filter(|value| !value.is_empty()) - .unwrap_or("metadata probe ended without a torrent file"); - let detail = match error_code { - Some(code) => format!("aria2 error code {code}: {error_message}"), - None => error_message.to_string(), - }; - return Err(format!( - "Aria2 could not resolve magnet metadata: {}", - crate::redact_sensitive_text(&detail) - )); - } - Some("active") | Some("waiting") | Some("paused") | None => {} - Some(status) => { - return Err(format!( - "Aria2 returned an unsupported metadata probe status: {status}" - )); - } - } - if Instant::now() >= deadline { - return Err("Aria2 magnet metadata resolution timed out".to_string()); - } - tokio::time::sleep(Duration::from_millis(250)).await; - } - - let bytes = tokio::fs::read(&metadata_path) - .await - .map_err(|error| format!("Aria2 did not save magnet metadata: {error}"))?; - let parsed = crate::torrent::parse_torrent_bytes(&bytes)?; - crate::torrent::validate_info_hash(Some(&expected.info_hash), &parsed.info_hash)?; - Ok((parsed, bytes)) + let mut options = serde_json::Map::new(); + options.insert( + "dir".to_string(), + serde_json::json!(probe_dir.to_string_lossy().to_string()), + ); + options.insert("bt-metadata-only".to_string(), serde_json::json!("true")); + options.insert("bt-save-metadata".to_string(), serde_json::json!("true")); + options.insert("max-tries".to_string(), serde_json::json!("3")); + options.insert("retry-wait".to_string(), serde_json::json!("2")); + options.insert("connect-timeout".to_string(), serde_json::json!("20")); + options.insert("timeout".to_string(), serde_json::json!("60")); + options.insert("auto-file-renaming".to_string(), serde_json::json!("false")); + if let Some(proxy) = proxy_value { + options.insert("all-proxy".to_string(), serde_json::json!(proxy)); } + + let client = std::sync::Arc::new(Aria2RpcClient { port, secret }); + let metadata_result = crate::torrent_probe::run_metadata_probe( + client, + source, + options, + &metadata_path, + Duration::from_secs(60), + Duration::from_millis(250), + ) .await; - let cleanup_result = if let Some(gid) = gid.as_deref() { - let removal = force_remove_aria2_gid(port, &secret, gid).await; - match removal { - Ok(()) => wait_for_aria2_stopped(port, &secret, gid).await, - Err(error) => Err(error), + let bytes = match metadata_result { + Ok(bytes) => bytes, + Err(crate::torrent_probe::ProbeFailure::Metadata(error)) => { + if let Err(remove_error) = remove_magnet_metadata_probe_dir(&probe_dir).await { + return Err(remove_error); + } + return Err(error); } - } else { - Ok(()) - }; - if let Err(error) = cleanup_result { - if aria2_daemon_process_exited(app_handle) { - if let Err(remove_error) = tokio::fs::remove_dir_all(&probe_dir).await { + Err(crate::torrent_probe::ProbeFailure::Cleanup(error)) => { + if aria2_daemon_process_exited(app_handle) { + if let Err(remove_error) = remove_magnet_metadata_probe_dir(&probe_dir).await { + return Err(format!( + "could not clean up magnet metadata probe: {error}; could not remove probe after aria2 exit: {remove_error}" + )); + } return Err(format!( - "could not clean up magnet metadata probe: {error}; could not remove probe after aria2 exit: {remove_error}" + "could not clean up magnet metadata probe after aria2 exit: {error}" )); } - return Err(format!( - "could not clean up magnet metadata probe after aria2 exit: {error}" - )); + return Err(format!("could not clean up magnet metadata probe: {error}")); } - return Err(format!("could not clean up magnet metadata probe: {error}")); - } - if let Err(error) = tokio::fs::remove_dir_all(&probe_dir).await { - return Err(format!("could not remove magnet metadata probe: {error}")); - } + }; - let (parsed, bytes) = metadata_result?; + if let Err(error) = remove_magnet_metadata_probe_dir(&probe_dir).await { + return Err(error); + } + let parsed = crate::torrent::parse_torrent_bytes(&bytes)?; + crate::torrent::validate_info_hash(Some(&expected.info_hash), &parsed.info_hash)?; let torrent_path = if cache { Some(crate::torrent::cache_torrent_bytes(app_handle, id, &bytes).await?) } else { diff --git a/src-tauri/src/torrent_probe.rs b/src-tauri/src/torrent_probe.rs new file mode 100644 index 0000000..6360ac0 --- /dev/null +++ b/src-tauri/src/torrent_probe.rs @@ -0,0 +1,641 @@ +use async_trait::async_trait; +use serde_json::{json, Map, Value}; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +const STOP_POLL_ATTEMPTS: usize = 30; +const STOP_POLL_INTERVAL: Duration = Duration::from_millis(100); +const CANCELLATION_CLEANUP_ATTEMPTS: usize = 3; +const CANCELLATION_CLEANUP_INTERVAL: Duration = Duration::from_millis(250); + +#[async_trait] +pub(crate) trait RpcClient: Send + Sync { + async fn call(&self, method: &str, params: Value) -> Result; +} + +#[derive(Debug, PartialEq, Eq)] +pub(crate) enum ProbeFailure { + Metadata(String), + Cleanup(String), +} + +pub(crate) async fn run_metadata_probe( + client: Arc, + source: &str, + options: Map, + metadata_path: &Path, + timeout: Duration, + poll_interval: Duration, +) -> Result, ProbeFailure> { + let mut cleanup_guard = ProbeCleanupGuard::new(Arc::clone(&client), metadata_path); + let result = match client + .call("aria2.addUri", json!([[source], options])) + .await + { + Ok(result) => result, + Err(error) => { + cleanup_guard.disarm(); + return Err(ProbeFailure::Metadata(format!( + "Aria2 could not start magnet metadata resolution: {}", + crate::redact_sensitive_text(&error) + ))); + } + }; + let gid = match result.as_str().filter(|value| !value.is_empty()) { + Some(gid) => gid.to_string(), + None => { + cleanup_guard.disarm(); + return Err(ProbeFailure::Metadata( + "Aria2 returned an empty metadata probe GID".to_string(), + )); + } + }; + cleanup_guard.set_gid(gid.clone()); + + let metadata_result = async { + let deadline = Instant::now() + timeout; + loop { + let status = match client + .call( + "aria2.tellStatus", + json!([&gid, ["status", "errorCode", "errorMessage"]]), + ) + .await + { + Ok(status) => status, + Err(error) if crate::aria2_gid_not_found(&error) => { + return Err(ProbeFailure::Metadata( + "Aria2 removed the magnet metadata probe before metadata was saved" + .to_string(), + )); + } + Err(error) if crate::retry::is_transient_network_error(&error) => { + if Instant::now() >= deadline { + return Err(ProbeFailure::Metadata(format!( + "Aria2 metadata resolution status failed: {}", + crate::redact_sensitive_text(&error) + ))); + } + tokio::time::sleep(poll_interval).await; + continue; + } + Err(error) => { + return Err(ProbeFailure::Metadata(format!( + "Aria2 metadata resolution status failed: {}", + crate::redact_sensitive_text(&error) + ))); + } + }; + match status.get("status").and_then(Value::as_str) { + Some("complete") => break, + Some("error") | Some("removed") => { + let error_code = status + .get("errorCode") + .and_then(Value::as_str) + .filter(|value| !value.is_empty()); + let error_message = status + .get("errorMessage") + .and_then(Value::as_str) + .filter(|value| !value.is_empty()) + .unwrap_or("metadata probe ended without a torrent file"); + let detail = match error_code { + Some(code) => format!("aria2 error code {code}: {error_message}"), + None => error_message.to_string(), + }; + return Err(ProbeFailure::Metadata(format!( + "Aria2 could not resolve magnet metadata: {}", + crate::redact_sensitive_text(&detail) + ))); + } + Some("active") | Some("waiting") | Some("paused") => {} + None => { + return Err(ProbeFailure::Metadata( + "Aria2 returned an invalid metadata probe status".to_string(), + )); + } + Some(status) => { + return Err(ProbeFailure::Metadata(format!( + "Aria2 returned an unsupported metadata probe status: {status}" + ))); + } + } + if Instant::now() >= deadline { + return Err(ProbeFailure::Metadata( + "Aria2 magnet metadata resolution timed out".to_string(), + )); + } + tokio::time::sleep(poll_interval).await; + } + + tokio::fs::read(metadata_path).await.map_err(|error| { + ProbeFailure::Metadata(format!( + "Aria2 did not save magnet metadata ({:?})", + error.kind() + )) + }) + } + .await; + + let cleanup_result = cleanup_metadata_probe(client.as_ref(), &gid).await; + if let Err(error) = cleanup_result { + return Err(ProbeFailure::Cleanup(error)); + } + cleanup_guard.disarm(); + metadata_result +} + +struct ProbeCleanupGuard { + client: Arc, + gid: Option, + probe_dir: Option, +} + +impl ProbeCleanupGuard { + fn new(client: Arc, metadata_path: &Path) -> Self { + Self { + client, + gid: None, + probe_dir: metadata_path + .parent() + .filter(|path| !path.as_os_str().is_empty()) + .map(Path::to_path_buf), + } + } + + fn set_gid(&mut self, gid: String) { + self.gid = Some(gid); + } + + fn disarm(&mut self) { + self.gid = None; + self.probe_dir = None; + } +} + +impl Drop for ProbeCleanupGuard { + fn drop(&mut self) { + let gid = self.gid.take(); + let probe_dir = self.probe_dir.take(); + if gid.is_none() && probe_dir.is_none() { + return; + } + + let Some(runtime) = tokio::runtime::Handle::try_current().ok() else { + log::warn!("magnet metadata probe was canceled without an active Tokio runtime"); + return; + }; + let client = Arc::clone(&self.client); + runtime.spawn(async move { + let cleanup_result = if let Some(gid) = gid { + let mut last_error = None; + for attempt in 0..CANCELLATION_CLEANUP_ATTEMPTS { + match cleanup_metadata_probe(client.as_ref(), &gid).await { + Ok(()) => { + last_error = None; + break; + } + Err(error) => { + last_error = Some(error); + if attempt + 1 < CANCELLATION_CLEANUP_ATTEMPTS { + tokio::time::sleep(CANCELLATION_CLEANUP_INTERVAL).await; + } + } + } + } + last_error + } else { + None + }; + if let Some(error) = cleanup_result { + log::warn!( + "canceled magnet metadata probe cleanup failed: {}", + crate::redact_sensitive_text(&error) + ); + // Do not delete a directory while Aria2 may still own the + // GID. The startup reaper removes this orphan before the + // next daemon launch. + return; + } + if let Some(probe_dir) = probe_dir { + if let Err(error) = tokio::fs::remove_dir_all(&probe_dir).await { + if error.kind() != std::io::ErrorKind::NotFound { + log::warn!( + "canceled magnet metadata probe directory cleanup failed ({:?})", + error.kind() + ); + } + } + } + }); + } +} + +async fn cleanup_metadata_probe(client: &C, gid: &str) -> Result<(), String> { + match client.call("aria2.forceRemove", json!([gid])).await { + Ok(result) => { + crate::ensure_aria2_gid_result("forceRemove", gid, &result)?; + wait_for_stopped(client, gid).await + } + Err(error) if crate::aria2_gid_not_found(&error) => Ok(()), + Err(error) => match aria2_status(client, gid).await { + Ok(status) if matches!(status.as_str(), "complete" | "error" | "removed") => Ok(()), + Err(status_error) if crate::aria2_gid_not_found(&status_error) => Ok(()), + _ => Err(format!( + "failed to remove aria2 gid {gid}: {}", + crate::redact_sensitive_text(&error) + )), + }, + } +} + +async fn aria2_status(client: &C, gid: &str) -> Result { + let result = client + .call("aria2.tellStatus", json!([gid, ["status"]])) + .await + .map_err(|error| { + format!( + "failed to query aria2 gid {gid}: {}", + crate::redact_sensitive_text(&error) + ) + })?; + result + .get("status") + .and_then(Value::as_str) + .map(str::to_string) + .ok_or_else(|| format!("aria2.tellStatus returned no status for gid {gid}")) +} + +async fn wait_for_stopped(client: &C, gid: &str) -> Result<(), String> { + let mut last_transient_error = None; + for _ in 0..STOP_POLL_ATTEMPTS { + match aria2_status(client, gid).await { + Ok(status) + if matches!(status.as_str(), "paused" | "complete" | "error" | "removed") => + { + return Ok(()) + } + Ok(_) => {} + Err(error) if crate::aria2_gid_not_found(&error) => return Ok(()), + Err(error) if crate::retry::is_transient_network_error(&error) => { + last_transient_error = Some(error); + } + Err(error) => return Err(error), + } + tokio::time::sleep(STOP_POLL_INTERVAL).await; + } + match last_transient_error { + Some(error) => Err(format!( + "aria2 gid {gid} did not stop within 3 seconds after forceRemove: {}", + crate::redact_sensitive_text(&error) + )), + None => Err(format!( + "aria2 gid {gid} did not stop within 3 seconds after forceRemove" + )), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::VecDeque; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::Mutex; + + struct FakeRpc { + statuses: Mutex>>, + force_remove: Mutex>, + removed: AtomicBool, + calls: Mutex>, + call_notification: tokio::sync::Notify, + } + + impl FakeRpc { + fn new( + statuses: impl IntoIterator>, + force_remove: Result, + ) -> Self { + Self { + statuses: Mutex::new(statuses.into_iter().collect()), + force_remove: Mutex::new(force_remove), + removed: AtomicBool::new(false), + calls: Mutex::new(Vec::new()), + call_notification: tokio::sync::Notify::new(), + } + } + + fn status(name: &str) -> Result { + Ok(json!({ "status": name })) + } + + fn call_names(&self) -> Vec { + self.calls + .lock() + .expect("call log lock should work") + .clone() + } + + async fn wait_for_call(&self, method: &str) { + loop { + if self.call_names().iter().any(|call| call == method) { + return; + } + self.call_notification.notified().await; + } + } + } + + #[async_trait] + impl RpcClient for FakeRpc { + async fn call(&self, method: &str, _params: Value) -> Result { + self.calls + .lock() + .expect("call log lock should work") + .push(method.to_string()); + self.call_notification.notify_one(); + match method { + "aria2.addUri" => Ok(json!("gid-1")), + "aria2.tellStatus" => { + if let Some(status) = self + .statuses + .lock() + .expect("status queue lock should work") + .pop_front() + { + status + } else if self.removed.load(Ordering::Acquire) { + Self::status("removed") + } else { + Self::status("active") + } + } + "aria2.forceRemove" => { + let result = self + .force_remove + .lock() + .expect("forceRemove lock should work"); + if result.is_ok() { + self.removed.store(true, Ordering::Release); + } + match &*result { + Ok(value) => Ok(value.clone()), + Err(error) => Err(error.clone()), + } + } + _ => Err(format!("unexpected RPC method {method}")), + } + } + } + + async fn run_fake_probe( + rpc: Arc, + timeout: Duration, + poll_interval: Duration, + ) -> Result, ProbeFailure> { + let temporary = tempfile::tempdir().expect("temporary probe storage should exist"); + let metadata_path = temporary.path().join("metadata.torrent"); + tokio::fs::write(&metadata_path, b"torrent metadata") + .await + .expect("metadata fixture should be writable"); + run_metadata_probe( + rpc, + "magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567", + Map::new(), + &metadata_path, + timeout, + poll_interval, + ) + .await + } + + #[tokio::test(flavor = "current_thread")] + async fn resolves_active_waiting_complete_and_cleans_the_probe() { + let rpc = Arc::new(FakeRpc::new( + [ + FakeRpc::status("active"), + FakeRpc::status("waiting"), + FakeRpc::status("complete"), + FakeRpc::status("removed"), + ], + Ok(json!("gid-1")), + )); + let bytes = run_fake_probe(Arc::clone(&rpc), Duration::from_secs(1), Duration::ZERO) + .await + .expect("complete probe should return metadata"); + assert_eq!(bytes, b"torrent metadata"); + assert_eq!( + rpc.call_names(), + vec![ + "aria2.addUri", + "aria2.tellStatus", + "aria2.tellStatus", + "aria2.tellStatus", + "aria2.forceRemove", + "aria2.tellStatus", + ] + ); + } + + #[tokio::test(flavor = "current_thread")] + async fn returns_metadata_error_for_failed_status_and_cleans_the_probe() { + let rpc = Arc::new(FakeRpc::new( + [FakeRpc::status("error"), FakeRpc::status("removed")], + Ok(json!("gid-1")), + )); + let error = run_fake_probe(Arc::clone(&rpc), Duration::from_secs(1), Duration::ZERO) + .await + .expect_err("error status should fail the probe"); + assert!( + matches!(error, ProbeFailure::Metadata(message) if message.contains("could not resolve")) + ); + } + + #[tokio::test(flavor = "current_thread")] + async fn retries_transient_status_errors_during_metadata_polling() { + let rpc = Arc::new(FakeRpc::new( + [ + Err("connection reset by peer".to_string()), + FakeRpc::status("complete"), + FakeRpc::status("removed"), + ], + Ok(json!("gid-1")), + )); + let bytes = run_fake_probe(Arc::clone(&rpc), Duration::from_secs(1), Duration::ZERO) + .await + .expect("a transient status error should be retried"); + assert_eq!(bytes, b"torrent metadata"); + } + + #[tokio::test(flavor = "current_thread")] + async fn rejects_a_malformed_status_response_instead_of_waiting_for_timeout() { + let rpc = Arc::new(FakeRpc::new([Ok(json!({}))], Ok(json!("gid-1")))); + let error = run_fake_probe(Arc::clone(&rpc), Duration::from_secs(1), Duration::ZERO) + .await + .expect_err("a missing status field should fail immediately"); + assert!(matches!( + error, + ProbeFailure::Metadata(message) + if message.contains("invalid metadata probe status") + )); + } + + #[tokio::test(flavor = "current_thread")] + async fn does_not_expose_metadata_path_when_reading_fails() { + let temporary = tempfile::tempdir().expect("temporary probe storage should exist"); + let metadata_path = temporary.path().join("missing-metadata.torrent"); + let rpc = Arc::new(FakeRpc::new( + [FakeRpc::status("complete"), FakeRpc::status("removed")], + Ok(json!("gid-1")), + )); + let error = run_metadata_probe( + Arc::clone(&rpc), + "magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567", + Map::new(), + &metadata_path, + Duration::from_secs(1), + Duration::ZERO, + ) + .await + .expect_err("missing metadata should fail"); + let ProbeFailure::Metadata(message) = error else { + panic!("metadata read failure should remain a metadata error"); + }; + assert!(!message.contains(&*temporary.path().to_string_lossy())); + assert!(message.contains("NotFound")); + } + + #[tokio::test(flavor = "current_thread")] + async fn redacts_status_rpc_errors_during_cleanup() { + let rpc = Arc::new(FakeRpc::new( + [ + FakeRpc::status("complete"), + Err("token=super-secret-value".to_string()), + ], + Ok(json!("gid-1")), + )); + let error = run_fake_probe(Arc::clone(&rpc), Duration::from_secs(1), Duration::ZERO) + .await + .expect_err("cleanup status failure should be reported"); + let ProbeFailure::Cleanup(message) = error else { + panic!("cleanup status failure should remain a cleanup error"); + }; + assert!(!message.contains("super-secret-value")); + assert!(message.contains("[redacted]")); + } + + #[tokio::test(flavor = "current_thread")] + async fn retries_transient_status_errors_after_force_remove() { + let rpc = Arc::new(FakeRpc::new( + [ + FakeRpc::status("complete"), + Err("connection reset by peer".to_string()), + FakeRpc::status("removed"), + ], + Ok(json!("gid-1")), + )); + run_fake_probe(Arc::clone(&rpc), Duration::from_secs(1), Duration::ZERO) + .await + .expect("a transient cleanup status error should be retried"); + } + + #[tokio::test(flavor = "current_thread")] + async fn handles_missing_gid_and_timeout_without_leaking_cleanup() { + let missing = Arc::new(FakeRpc::new( + [ + Err("aria2 gid gid-1 not found".to_string()), + Err("aria2 gid gid-1 not found".to_string()), + ], + Err("temporary forceRemove transport failure".to_string()), + )); + let missing_error = + run_fake_probe(Arc::clone(&missing), Duration::from_secs(1), Duration::ZERO) + .await + .expect_err("missing gid should fail the probe"); + assert!( + matches!(missing_error, ProbeFailure::Metadata(message) if message.contains("removed")) + ); + + let timeout = Arc::new(FakeRpc::new(Vec::new(), Ok(json!("gid-1")))); + let timeout_error = run_fake_probe( + Arc::clone(&timeout), + Duration::from_millis(5), + Duration::from_millis(1), + ) + .await + .expect_err("active probe should time out"); + assert!( + matches!(timeout_error, ProbeFailure::Metadata(message) if message.contains("timed out")) + ); + } + + #[tokio::test(flavor = "current_thread")] + async fn accepts_force_remove_race_only_after_terminal_status() { + let terminal_race = Arc::new(FakeRpc::new( + [FakeRpc::status("error"), FakeRpc::status("complete")], + Err("temporary forceRemove transport failure".to_string()), + )); + let error = run_fake_probe( + Arc::clone(&terminal_race), + Duration::from_secs(1), + Duration::ZERO, + ) + .await + .expect_err("metadata error should still be returned"); + assert!( + matches!(error, ProbeFailure::Metadata(message) if message.contains("could not resolve")) + ); + + let active_race = Arc::new(FakeRpc::new( + [FakeRpc::status("error"), FakeRpc::status("active")], + Err("temporary forceRemove transport failure".to_string()), + )); + let cleanup_error = run_fake_probe( + Arc::clone(&active_race), + Duration::from_secs(1), + Duration::ZERO, + ) + .await + .expect_err("active race must report cleanup failure"); + assert!( + matches!(cleanup_error, ProbeFailure::Cleanup(message) if message.contains("failed to remove")) + ); + } + + #[tokio::test(flavor = "current_thread")] + async fn cancellation_still_removes_the_remote_gid_and_probe_directory() { + let temporary = tempfile::tempdir().expect("temporary probe storage should exist"); + let metadata_path = temporary.path().join("metadata.torrent"); + tokio::fs::write(&metadata_path, b"torrent metadata") + .await + .expect("metadata fixture should be writable"); + let rpc = Arc::new(FakeRpc::new(Vec::new(), Ok(json!("gid-1")))); + let probe = run_metadata_probe( + Arc::clone(&rpc), + "magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567", + Map::new(), + &metadata_path, + Duration::from_secs(60), + Duration::from_secs(60), + ); + assert!( + tokio::time::timeout(Duration::from_millis(20), probe) + .await + .is_err(), + "the active probe should be canceled while polling" + ); + + tokio::time::timeout( + Duration::from_secs(1), + rpc.wait_for_call("aria2.forceRemove"), + ) + .await + .expect("cancellation cleanup should force-remove the GID"); + tokio::time::timeout(Duration::from_secs(1), async { + while temporary.path().exists() { + tokio::task::yield_now().await; + } + }) + .await + .expect("cancellation cleanup should remove the probe directory"); + } +}