From b564e92532fcf1d91b2ce8746de0757517984148 Mon Sep 17 00:00:00 2001 From: NimBold Date: Sun, 30 Aug 2026 14:58:01 +0330 Subject: [PATCH] fix(network): harden route-aware torrent resolution - keep reqwest metadata on its supported proxy route and require Aria2 asynchronous DNS - fence magnet probe ownership and delayed RPC cleanup without cross-transfer removal - validate Torrent DHT nodes, web seeds, proxy inputs, and literal IPv4/IPv6 targets - extend regression and smoke coverage for issue #38 Refs: #38 --- scripts/smoke-aria2-resolver.js | 18 +- scripts/smoke-torrent.js | 9 +- src-tauri/src/lib.rs | 121 +++-- src-tauri/src/network.rs | 140 +++++- src-tauri/src/queue.rs | 332 ++---------- src-tauri/src/torrent.rs | 117 ++++- src-tauri/src/torrent_probe.rs | 838 ++++++++++++++++++++----------- src-tauri/tests/queue_manager.rs | 52 +- 8 files changed, 918 insertions(+), 709 deletions(-) diff --git a/scripts/smoke-aria2-resolver.js b/scripts/smoke-aria2-resolver.js index 9724dd4..2dae624 100644 --- a/scripts/smoke-aria2-resolver.js +++ b/scripts/smoke-aria2-resolver.js @@ -189,6 +189,7 @@ const child = spawn(binaryPath, [ `--dir=${tempRoot}`, '--file-allocation=none', '--enable-dht=false', + '--async-dns=true', '--console-log-level=error', '--quiet=true', ], { env: environment, stdio: ['ignore', 'ignore', 'pipe'] }); @@ -198,15 +199,17 @@ child.stderr.on('data', chunk => { stderr += chunk.toString(); }); try { const version = await waitForRpc(rpcPort, secret); const features = Array.isArray(version.enabledFeatures) ? version.enabledFeatures : []; - console.log(`[INFO] aria2 ${version.version || 'unknown'}; Async DNS: ${features.includes('Async DNS') ? 'supported' : 'not advertised'}`); + if (!features.includes('Async DNS')) { + throw new Error(`packaged aria2 must advertise Async DNS for route-safe transfers: ${JSON.stringify(version)}`); + } + console.log(`[INFO] aria2 ${version.version || 'unknown'}; Async DNS: supported`); const uriResult = await rpc(rpcPort, secret, 'aria2.addUri', [[`http://127.0.0.1:${contentPort}/file`], { - 'async-dns': 'false', out: 'resolver-normal.bin', }]); const uriOptions = await rpc(rpcPort, secret, 'aria2.getOption', [uriResult]); - if (uriOptions['async-dns'] !== 'false') { - throw new Error(`aria2.addUri did not retain async-dns=false: ${JSON.stringify(uriOptions)}`); + if (uriOptions['async-dns'] === 'false') { + throw new Error(`direct aria2.addUri unexpectedly disabled asynchronous DNS: ${JSON.stringify(uriOptions)}`); } const torrent = bencode({ @@ -218,12 +221,11 @@ try { }, }).toString('base64'); const torrentResult = await rpc(rpcPort, secret, 'aria2.addTorrent', [torrent, [], { - 'async-dns': 'false', dir: tempRoot, }]); const torrentOptions = await rpc(rpcPort, secret, 'aria2.getOption', [torrentResult]); - if (torrentOptions['async-dns'] !== 'false') { - throw new Error(`aria2.addTorrent did not retain async-dns=false: ${JSON.stringify(torrentOptions)}`); + if (torrentOptions['async-dns'] === 'false') { + throw new Error(`direct aria2.addTorrent unexpectedly disabled asynchronous DNS: ${JSON.stringify(torrentOptions)}`); } const proxyRoute = 'http://127.0.0.1:9'; @@ -256,7 +258,7 @@ try { await forceRemoveIfPresent(rpcPort, secret, proxiedUriResult); await forceRemoveIfPresent(rpcPort, secret, proxiedTorrentResult); - console.log('[PASS] Aria2 retained system-resolver mode for direct normal/Torrent transfers and configured proxy mode for proxied transfers'); + console.log('[PASS] Aria2 kept asynchronous DNS for fresh direct and proxied normal/Torrent transfers'); } catch (error) { const detail = stderr.trim(); throw new Error(`${error.message}${detail ? `\n${detail}` : ''}`); diff --git a/scripts/smoke-torrent.js b/scripts/smoke-torrent.js index 370040f..937a340 100644 --- a/scripts/smoke-torrent.js +++ b/scripts/smoke-torrent.js @@ -305,6 +305,7 @@ async function startDaemon({ name, rpcPort, listenPort, directory, extraArgs = [ '--enable-dht=false', '--enable-peer-exchange=false', '--bt-enable-lpd=false', + '--async-dns=true', '--console-log-level=error', '--quiet=true', ...(selectedListenPort ? [`--listen-port=${selectedListenPort}`] : []), @@ -720,7 +721,6 @@ async function main() { 'bt-metadata-only': 'false', 'bt-save-metadata': 'false', 'follow-torrent': 'false', - 'async-dns': 'false', 'max-tries': '3', 'retry-wait': '2', 'connect-timeout': '20', @@ -767,7 +767,7 @@ async function main() { assert(directHandoff.parent.status === 'complete', `normal magnet parent did not complete metadata: ${JSON.stringify(directHandoff.parent)}`); assert(directHandoff.parent.files?.some(file => String(file.path).startsWith('[METADATA]')), 'normal magnet parent did not expose a metadata file'); assert(directOptions['bt-metadata-only'] === 'false', 'normal magnet child did not retain payload mode'); - assert(directOptions['async-dns'] === 'false', 'direct Torrent did not retain system DNS resolution'); + assert(directOptions['async-dns'] !== 'false', 'fresh direct Torrent unexpectedly disabled asynchronous DNS'); await rpc(client.rpcPort, client.secret, 'aria2.removeDownloadResult', [directGid]); try { await rpc(client.rpcPort, client.secret, 'aria2.removeDownloadResult', [directHandoff.childGid]); @@ -790,7 +790,6 @@ async function main() { dir: probeDir, 'bt-metadata-only': 'true', 'bt-save-metadata': 'true', - 'async-dns': 'false', 'max-tries': '3', 'retry-wait': '1', 'connect-timeout': '5', @@ -798,7 +797,7 @@ async function main() { 'auto-file-renaming': 'false', }]); const probeOptions = await rpc(client.rpcPort, client.secret, 'aria2.getOption', [probeGid]); - assert(probeOptions['async-dns'] === 'false', 'direct magnet metadata probe did not retain system DNS resolution'); + assert(probeOptions['async-dns'] !== 'false', 'fresh direct magnet probe unexpectedly disabled asynchronous DNS'); const probeStatus = await waitForTerminal(client, probeGid, 30000); assert(probeStatus.status === 'complete', 'magnet metadata probe did not complete'); const savedTorrentPaths = fs.readdirSync(probeDir) @@ -859,7 +858,7 @@ async function main() { assert(proxiedProbeOptions['async-dns'] !== 'false', 'proxied magnet metadata probe unexpectedly forced system DNS resolution'); assert(await forceRemoveIfPresent(client, proxiedProbeGid), 'proxied magnet metadata probe was not removable'); await waitForRemoved(client, proxiedProbeGid); - console.log('[OK] metadata probe was removed after resolution; direct and proxied resolver modes were retained'); + console.log('[OK] metadata probe was removed after resolution; direct and proxied probes kept asynchronous DNS'); const finalGid = await rpc(client.rpcPort, client.secret, 'aria2.addTorrent', [ trackerlessTorrentBytes.toString('base64'), diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 55e80a6..5d01254 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -1739,11 +1739,11 @@ async fn fetch_remote_torrent_bytes( cookie_scopes: Option<&[extension_server::ExtensionCookieScope]>, ) -> Result, String> { ensure_reqwest_crypto_provider(); - let proxy = proxy - .map(crate::queue::aria2_all_proxy_value) - .transpose()? - .flatten(); - let route = crate::network::NetworkRoute::from_proxy(proxy.as_deref()); + // This fetch is reqwest-backed, so preserve the route exactly as supplied. + // In particular, SOCKS is a valid reqwest route even though Aria2 cannot + // use it for normal transfers. Sharing Aria2's proxy validator here would + // reject a supported remote-torrent metadata path before reqwest sees it. + let route = crate::network::NetworkRoute::from_proxy(proxy); let mut current = reqwest::Url::parse(source) .map_err(|_| "SSRF blocked: Invalid URL".to_string())?; let original_origin = Some(current.clone()); @@ -3466,6 +3466,17 @@ struct Aria2DaemonGuard { shutdown_state: AtomicU8, } +fn aria2_supports_async_dns(version: &serde_json::Value) -> bool { + version + .get("enabledFeatures") + .and_then(|features| features.as_array()) + .is_some_and(|features| { + features + .iter() + .any(|feature| feature.as_str() == Some("Async DNS")) + }) +} + impl Aria2DaemonGuard { fn new() -> Self { Self { @@ -9190,6 +9201,10 @@ async fn resolve_magnet_metadata( cache: bool, ) -> Result { let expected = crate::torrent::inspect_source(source)?; + // Validate and normalize the Magnet before creating any filesystem + // state. Otherwise a malformed secondary parameter can leave an orphaned + // probe directory even though Aria2 was never contacted. + let sanitized_source = crate::torrent::sanitize_magnet_uri_for_aria2(source)?; let managed_path = crate::torrent::managed_torrent_path(app_handle, id)?; let proxy_value = proxy .map(crate::queue::aria2_all_proxy_value) @@ -9244,15 +9259,6 @@ async fn resolve_magnet_metadata( 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 proxy_value - .as_deref() - .is_none_or(|proxy| proxy.trim().is_empty()) - { - // Keep optional magnet metadata refreshes on the same system DNS - // route as direct Torrent transfers when a TUN client owns resolver - // interception. Explicit proxy routes retain their own DNS policy. - options.insert("async-dns".to_string(), serde_json::json!("false")); - } if let Some(proxy) = proxy_value.as_deref() { options.insert("all-proxy".to_string(), serde_json::json!(proxy)); } @@ -9274,23 +9280,26 @@ async fn resolve_magnet_metadata( } let client = std::sync::Arc::new(Aria2RpcClient { port, secret }); - let sanitized_source = crate::torrent::sanitize_magnet_uri_for_aria2(source)?; - let direct_route = proxy_value + let resolver_mode = if proxy_value .as_deref() - .is_none_or(|proxy| proxy.trim().is_empty()); - let alternate_resolver_available = - direct_route && state.queue_manager.aria2_async_dns_supported(); - let first_resolver_mode = if direct_route { "system" } else { "configured" }; - let metadata_result = crate::torrent_probe::run_metadata_probe_with_resolver_fallback( + .is_none_or(|proxy| proxy.trim().is_empty()) + { + "automatic" + } else { + "configured" + }; + let metadata_result = crate::torrent_probe::run_bounded_metadata_probe( client, &sanitized_source, options, &metadata_path, - first_resolver_mode, - alternate_resolver_available, - Duration::from_secs(60), - Duration::from_secs(20), - Duration::from_millis(250), + resolver_mode, + crate::torrent_probe::MetadataProbeSchedule { + total_timeout: Duration::from_secs(60), + metadata_timeout: Duration::from_secs(55), + cleanup_reserve: Duration::from_secs(5), + poll_interval: Duration::from_millis(250), + }, ) .await; @@ -14025,6 +14034,7 @@ mod tests { normalize_media_cookie_source, validate_enqueue_url, validate_enqueue_uris, validate_keychain_grant_request_id, validate_torrent_metadata_network_policy, + aria2_supports_async_dns, aria2_gid_not_found, aria2_download_state_progress, preflight_download_destination_access, retained_torrent_id_from_persisted_record, @@ -14062,6 +14072,17 @@ mod tests { assert!(!aria2_gid_not_found("aria2 error code 3: Resource not found")); } + #[test] + fn aria2_async_dns_capability_requires_an_explicit_feature() { + assert!(aria2_supports_async_dns(&json!({ + "enabledFeatures": ["Async DNS", "BitTorrent"] + }))); + assert!(!aria2_supports_async_dns(&json!({ + "enabledFeatures": ["BitTorrent"] + }))); + assert!(!aria2_supports_async_dns(&json!({}))); + } + #[test] fn stale_lifecycle_cleanup_only_targets_the_expected_native_owner() { assert!(!stale_lifecycle_cleanup_is_noop(Some(7), Some(7))); @@ -15057,7 +15078,7 @@ mod tests { validate_torrent_metadata_network_policy( b"d8:announce25:http://127.0.0.1/announce4:infod6:lengthi5e4:name4:test12:piece lengthi2e6:pieces20:01234567890123456789ee" ), - Err("SSRF blocked: Private/local IP not allowed".to_string()) + Err("torrent metadata contains an invalid network destination".to_string()) ); assert_eq!( validate_torrent_metadata_network_policy( @@ -15103,16 +15124,6 @@ mod tests { .await, Err("Torrent metadata URLs must not contain credentials".to_string()) ); - let proxy_error = super::fetch_remote_torrent_bytes( - "https://example.com/sample.torrent", - Some("socks5://127.0.0.1:1080"), - None, - None, - None, - ) - .await - .expect_err("remote Torrent metadata must use the shared proxy policy"); - assert!(proxy_error.contains("SOCKS")); } #[tokio::test] @@ -18696,8 +18707,6 @@ pub fn run() { }); let queue_manager_poll = Arc::clone(&queue_manager); - let queue_manager_capability = Arc::clone(&queue_manager); - app.manage(AppState { download_coordinator: download::DownloadCoordinator::spawn(app.handle().clone()), storage_layout, @@ -18871,6 +18880,10 @@ pub fn run() { .arg("--download-result=hide") .arg("--max-concurrent-downloads=9999") .arg("--check-certificate=true") + // Firelink relies on Aria2's asynchronous + // resolver so a slow DNS answer cannot block + // the daemon's shared RPC/control loop. + .arg("--async-dns=true") .arg(format!("--stop-with-process={}", std::process::id())); apply_aria2_torrent_global_options( @@ -18999,23 +19012,31 @@ pub fn run() { match rpc_call(attempt_port, &aria2_secret_clone, "aria2.getVersion", serde_json::json!([])).await { Ok(ver) => { let v = ver.get("version").and_then(|v| v.as_str()).unwrap_or("unknown"); - let async_dns_supported = ver - .get("enabledFeatures") - .and_then(|features| features.as_array()) - .map(|features| { - features.iter().any(|feature| { - feature.as_str() == Some("Async DNS") - }) - }) - .unwrap_or(false); - queue_manager_capability - .set_aria2_async_dns_supported(async_dns_supported); + let async_dns_supported = + aria2_supports_async_dns(&ver); log::info!( "aria2 daemon ready (version {}) on port {} (async DNS: {})", v, attempt_port, async_dns_supported ); + if !async_dns_supported { + let error = "bundled aria2 does not support asynchronous DNS; network transfers are disabled".to_string(); + log::error!("{}", error); + *guard.startup_error.lock().unwrap() = + Some(error); + // The route-safe Aria2 contract depends on + // asynchronous DNS. Do not leave a daemon + // running in a mode that can block its RPC + // loop behind a system resolver. + shutdown_aria2_daemon(app_handle_bg.clone()) + .await; + aria2_port_clone.store( + 0, + std::sync::atomic::Ordering::Relaxed, + ); + return; + } ready = true; break; } diff --git a/src-tauri/src/network.rs b/src-tauri/src/network.rs index ccb3360..a7d665d 100644 --- a/src-tauri/src/network.rs +++ b/src-tauri/src/network.rs @@ -45,22 +45,31 @@ impl NetworkRoute { /// Translate the route into Aria2's all-proxy option without changing the /// target hostname. Aria2 accepts HTTP-family proxy endpoints for normal - /// transfers; SOCKS remains a yt-dlp-only route in Firelink. + /// transfers; reqwest and yt-dlp consumers may still retain SOCKS routes. pub(crate) fn aria2_proxy_value(&self) -> Result, String> { match self { Self::Inherited => Ok(None), Self::Direct => Ok(Some(String::new())), Self::Proxy(value) => { - let is_socks = value.split_once("://").is_some_and(|(scheme, _)| { - scheme.eq_ignore_ascii_case("socks") - || scheme.to_ascii_lowercase().starts_with("socks") - }); + let parsed = Url::parse(value).map_err(|error| { + crate::redact_sensitive_text(&format!("invalid Aria2 proxy URL: {error}")) + })?; + if parsed.host_str().is_none_or(str::is_empty) { + return Err("invalid Aria2 proxy URL: proxy must include a host".to_string()); + } + let is_socks = parsed.scheme().eq_ignore_ascii_case("socks") + || parsed.scheme().to_ascii_lowercase().starts_with("socks"); if is_socks { return Err( "SOCKS system proxies are not supported for normal file downloads because aria2 only accepts HTTP/HTTPS/FTP proxy URLs. Use an HTTP proxy endpoint for normal downloads, or use media downloads where yt-dlp supports SOCKS." .to_string(), ); } + if !matches!(parsed.scheme(), "http" | "https" | "ftp") { + return Err( + "Aria2 proxy must use an HTTP, HTTPS, or FTP proxy URL".to_string(), + ); + } Ok(Some(value.clone())) } } @@ -116,15 +125,7 @@ pub(crate) fn validate_url( } } - let normalized_host = host.trim_matches(['[', ']']); - if is_local_hostname(normalized_host) { - return Err("SSRF blocked: Private/local IP not allowed".to_string()); - } - if parse_literal_ip(normalized_host).is_some_and(is_blocked_network_address) { - return Err("SSRF blocked: Private/local IP not allowed".to_string()); - } - - Ok(()) + validate_host(host) } pub(crate) fn parse_and_validate_url( @@ -148,13 +149,71 @@ pub(crate) fn is_local_hostname(host: &str) -> bool { || normalized.ends_with(".local") } +/// Validate a network hostname without asking the application to resolve it. +/// +/// This is also used for hostname/port pairs embedded in Torrent metadata, +/// where no URL parser has normalized legacy numeric IPv4 spellings for us. +pub(crate) fn validate_host(host: &str) -> Result<(), String> { + if host.is_empty() + || host + .chars() + .any(|character| character.is_control() || character.is_whitespace()) + { + return Err("SSRF blocked: Invalid host".to_string()); + } + let normalized_host = host.trim_end_matches('.'); + let bracketed = normalized_host.starts_with('[') || normalized_host.ends_with(']'); + let normalized_host = match ( + normalized_host.starts_with('['), + normalized_host.ends_with(']'), + ) { + (true, true) => &normalized_host[1..normalized_host.len() - 1], + (false, false) => normalized_host, + _ => return Err("SSRF blocked: Invalid host".to_string()), + }; + if normalized_host.is_empty() { + return Err("SSRF blocked: Invalid host".to_string()); + } + if bracketed && (!normalized_host.contains(':') || parse_literal_ip(normalized_host).is_none()) + { + return Err("SSRF blocked: Invalid host".to_string()); + } + if is_local_hostname(normalized_host) + || parse_literal_ip(normalized_host).is_some_and(is_blocked_network_address) + { + return Err("SSRF blocked: Private/local IP not allowed".to_string()); + } + if normalized_host.contains(':') && parse_literal_ip(normalized_host).is_none() { + return Err("SSRF blocked: Invalid host".to_string()); + } + if !normalized_host.contains(':') && url::Host::parse(normalized_host).is_err() { + return Err("SSRF blocked: Invalid host".to_string()); + } + Ok(()) +} + fn parse_literal_ip(host: &str) -> Option { let host = host.trim_end_matches('.'); if let Ok(ip) = host.parse::() { return Some(ip); } - host.split_once("%25") - .and_then(|(address, _zone)| address.parse::().ok()) + if let Some((address, _zone)) = host.split_once("%25") { + if let Ok(ip) = address.parse::() { + return Some(ip); + } + } + + // URL parsers commonly canonicalize legacy IPv4 literals such as 127.1, + // decimal IPv4, hexadecimal IPv4, and octal IPv4. Reuse that canonical + // parser for raw Torrent node hosts so those spellings cannot bypass the + // literal-target policy without performing DNS. + let candidate = if host.contains(':') { + format!("http://[{host}]/") + } else { + format!("http://{host}/") + }; + let parsed = Url::parse(&candidate).ok()?; + parsed.host_str()?.parse::().ok() } pub(crate) fn is_blocked_network_address(ip: IpAddr) -> bool { @@ -164,7 +223,11 @@ pub(crate) fn is_blocked_network_address(ip: IpAddr) -> bool { match ip { IpAddr::V4(ipv4) => ipv4.is_private() || ipv4.is_link_local(), IpAddr::V6(ipv6) => { - ipv6.to_ipv4() + // Check both IPv4-mapped and deprecated IPv4-compatible forms; + // either can encode a local IPv4 destination behind an IPv6 + // literal. + ipv6.to_ipv4_mapped() + .or_else(|| ipv6.to_ipv4()) .is_some_and(|ipv4| is_blocked_network_address(ipv4.into())) || (ipv6.segments()[0] & 0xfe00) == 0xfc00 || (ipv6.segments()[0] & 0xffc0) == 0xfe80 @@ -188,6 +251,7 @@ mod tests { "http://169.254.10.2/file", "http://[::1]/file", "http://[::ffff:127.0.0.1]/file", + "http://[::ffff:169.254.169.254]/file", "http://[fc00::1]/file", "http://[fe80::1]/file", "http://127.0.0.1./file", @@ -242,10 +306,32 @@ mod tests { ), Ok(()) ); + assert_eq!( + validate("https://[2001:db8::1]/file", &["http", "https"]), + Ok(()) + ); + assert_eq!( + validate_host("2001:db8::1"), + Ok(()), + "public IPv6 literals must remain usable without DNS" + ); + } + + #[test] + fn raw_hosts_reject_legacy_literals_without_resolving_public_names() { + for host in ["127.1", "2130706433", "0x7f000001", "0177.0.0.1", "0"] { + assert_eq!( + validate_host(host), + Err("SSRF blocked: Private/local IP not allowed".to_string()), + "{host}" + ); + } + assert!(validate_host("node-does-not-resolve.invalid").is_ok()); } #[test] fn route_mapping_preserves_direct_and_proxy_choices() { + crate::ensure_reqwest_crypto_provider(); assert_eq!(NetworkRoute::from_proxy(None), NetworkRoute::Inherited); assert_eq!(NetworkRoute::from_proxy(Some("none")), NetworkRoute::Direct); assert_eq!(NetworkRoute::from_proxy(Some(" ")), NetworkRoute::Direct); @@ -274,6 +360,26 @@ mod tests { .aria2_proxy_value() .is_err() ); + assert!(NetworkRoute::from_proxy(Some("http://[invalid")) + .aria2_proxy_value() + .is_err()); + assert!(NetworkRoute::from_proxy(Some("file:///tmp/proxy")) + .aria2_proxy_value() + .is_err()); + assert!( + NetworkRoute::from_proxy(Some("socks5://proxy.example:1080")) + .configure_reqwest(reqwest::Client::builder()) + .and_then(|builder| builder.build().map_err(|error| error.to_string())) + .is_ok(), + "reqwest-backed metadata must retain supported SOCKS routes" + ); + } + + #[test] + fn raw_hosts_reject_unbalanced_ipv6_brackets() { + assert!(validate_host("[2001:db8::1").is_err()); + assert!(validate_host("2001:db8::1]").is_err()); + assert!(validate_host("[download.example]").is_err()); } #[test] diff --git a/src-tauri/src/queue.rs b/src-tauri/src/queue.rs index 69e22ab..c24626f 100644 --- a/src-tauri/src/queue.rs +++ b/src-tauri/src/queue.rs @@ -5,8 +5,8 @@ use crate::ipc::{ }; use crate::power::PowerManager; use crate::retry::{ - aria2_error_code, backoff_and_emit, is_aria2_name_resolution_error, - is_transient_network_error, network_error_class, BackoffOutcome, MAX_RETRIES, + aria2_error_code, backoff_and_emit, is_transient_network_error, network_error_class, + BackoffOutcome, MAX_RETRIES, }; use log; use serde::Deserialize; @@ -1002,28 +1002,13 @@ fn is_direct_magnet_payload(payload: &SpawnPayload) -> bool { .is_some_and(|scheme| scheme.eq_ignore_ascii_case("magnet:")) } -fn proxy_uses_direct_network(proxy: Option<&str>) -> bool { - proxy.is_none_or(|proxy| { - let proxy = proxy.trim(); - proxy.is_empty() || proxy.eq_ignore_ascii_case("none") - }) -} - -fn payload_uses_direct_network(payload: &SpawnPayload) -> bool { - proxy_uses_direct_network(payload.proxy.as_deref()) -} - -fn torrent_uses_direct_network(payload: &SpawnPayload) -> bool { - payload.is_torrent - && !payload.torrent_verify_only - && payload_uses_direct_network(payload) -} - -fn aria2_effective_resolver_mode(payload: &SpawnPayload) -> &'static str { - if payload.aria2_resolver_mode == Aria2ResolverMode::System - && payload_uses_direct_network(payload) +fn aria2_resolver_route_for_log(payload: &SpawnPayload) -> &'static str { + if payload + .proxy + .as_deref() + .is_some_and(|proxy| !proxy.trim().is_empty() && !proxy.eq_ignore_ascii_case("none")) { - "system" + "configured" } else { "automatic" } @@ -1123,23 +1108,6 @@ enum SeedAdmissionOutcome { /// Args mirroring start_download / start_media_download. Kept untyped-loose /// (String/Option) to match the existing command signatures exactly. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum Aria2ResolverMode { - /// Use Aria2's normal asynchronous resolver configuration. This is the - /// bounded alternate mode for a direct transfer after the system route - /// cannot resolve the target. - Automatic, - /// Use the host operating system resolver for this direct transfer. This - /// is the initial mode so TUN/VPN DNS interception remains authoritative. - System, -} - -impl Default for Aria2ResolverMode { - fn default() -> Self { - Self::Automatic - } -} - #[derive(Debug, Clone, Default)] pub struct SpawnPayload { pub url: String, @@ -1160,9 +1128,6 @@ pub struct SpawnPayload { pub retry_not_found_errors: bool, pub adaptive_mirror_selection: bool, pub proxy: Option, - /// Runtime-only resolver selection. This is never part of an enqueue or - /// persisted download payload. - pub aria2_resolver_mode: Aria2ResolverMode, pub format_selector: Option, pub cookie_source: Option, pub is_media: bool, @@ -1365,11 +1330,6 @@ pub struct QueueManager { /// cap as a degraded connection pool. aria2_global_speed_limit: Arc>>, - /// Capability reported by aria2.getVersion. Keep the fallback disabled - /// until the daemon explicitly advertises Async DNS; an unknown - /// capability must not be treated as permission to change resolver mode. - aria2_async_dns_supported: AtomicBool, - /// 0-based transient-error strike counter per aria2 download id. aria2_retry_strikes: Mutex>, @@ -1466,7 +1426,6 @@ impl QueueManager { aria2_dispatch_inflight: Mutex::new(HashMap::new()), aria2_dispatch_notify: Notify::new(), aria2_global_speed_limit: Arc::new(StdMutex::new(None)), - aria2_async_dns_supported: AtomicBool::new(false), aria2_retry_strikes: Mutex::new(HashMap::new()), aria2_retry_cancelled: Mutex::new(HashSet::new()), aria2_retry_inflight: Mutex::new(HashMap::new()), @@ -1488,19 +1447,6 @@ impl QueueManager { Arc::clone(&self.power_manager) } - pub fn set_aria2_async_dns_supported(&self, supported: bool) { - self.aria2_async_dns_supported - .store(supported, Ordering::Relaxed); - } - - pub(crate) fn aria2_async_dns_supported(&self) -> bool { - self.aria2_async_dns_supported.load(Ordering::Relaxed) - } - - fn aria2_alternate_resolver_available(&self) -> bool { - self.aria2_async_dns_supported() - } - /// Accept one lifecycle-fenced Aria2 status sample and return Firelink's /// monotonic lifetime counters. Poller callers must already have checked /// the mapping; the key and epoch checks here provide a second fence at @@ -6062,13 +6008,7 @@ impl QueueManager { *entry }; - let retry_action = aria2_retry_action( - &payload, - &error, - strike, - self.aria2_alternate_resolver_available(), - ); - let resolver_fallback = retry_action == Aria2RetryAction::AlternateResolverFallback; + let retry_action = aria2_retry_action(&payload, &error, strike); let requested_connections = self .aria2_requested_connections(&id) .await @@ -6078,7 +6018,6 @@ impl QueueManager { .await .unwrap_or(requested_connections); let action = match retry_action { - Aria2RetryAction::AlternateResolverFallback => "alternate_resolver_fallback", Aria2RetryAction::OrdinaryRetry => "ordinary_retry", Aria2RetryAction::Terminal => "terminal", }; @@ -6091,16 +6030,12 @@ impl QueueManager { mapping.epoch, strike, action, - aria2_effective_resolver_mode(&payload), + aria2_resolver_route_for_log(&payload), network_error_class(&error), error_code, requested_connections, effective_connections ); - // Switching resolver strategy is a bounded transfer repair, not an - // ordinary retry. It must still run when the user configured zero - // automatic retries, while every later failure follows the normal - // retry budget and never switches back to the first strategy. if retry_action == Aria2RetryAction::Terminal { self.apply_completion_locked_with_progress( &id, @@ -6145,14 +6080,6 @@ impl QueueManager { .insert(id.clone(), payload.clone()); } - if resolver_fallback { - payload.aria2_resolver_mode = Aria2ResolverMode::Automatic; - self.aria2_payloads - .lock() - .await - .insert(id.clone(), payload.clone()); - } - let this = Arc::clone(self); let id_for_task = id.clone(); let error_for_emit = error.clone(); @@ -6174,11 +6101,7 @@ impl QueueManager { }; let outcome = backoff_and_emit(strike, error_for_emit, retry_cancel, |reason| { use tauri::Emitter; - let event = if resolver_fallback { - DownloadStateEvent::retrying_with_resolver_fallback(&id_for_task, reason) - } else { - DownloadStateEvent::retrying(&id_for_task, reason) - }; + let event = DownloadStateEvent::retrying(&id_for_task, reason); let _ = this.app_handle.emit("download-state", event); }) .await; @@ -6262,12 +6185,10 @@ impl QueueManager { .await; return; } - if !resolver_fallback { - this.aria2_retry_strikes - .lock() - .await - .insert(id_for_task.clone(), strike + 1); - } + this.aria2_retry_strikes + .lock() + .await + .insert(id_for_task.clone(), strike + 1); let new_gid_for_event = new_gid.clone(); let buffered_outcome = this.remember_gid(id_for_task.clone(), new_gid).await; // Install the replacement GID before exposing the @@ -6634,33 +6555,13 @@ fn is_retryable_aria2_error_for_payload(payload: &SpawnPayload, error: &str) -> && is_aria2_low_speed_error(error)) } -fn should_use_aria2_alternate_resolver_fallback( - payload: &SpawnPayload, - error: &str, - async_dns_supported: bool, -) -> bool { - async_dns_supported - && payload.aria2_resolver_mode == Aria2ResolverMode::System - && payload_uses_direct_network(payload) - && is_aria2_name_resolution_error(error) -} - #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum Aria2RetryAction { - AlternateResolverFallback, OrdinaryRetry, Terminal, } -fn aria2_retry_action( - payload: &SpawnPayload, - error: &str, - strike: usize, - async_dns_supported: bool, -) -> Aria2RetryAction { - if should_use_aria2_alternate_resolver_fallback(payload, error, async_dns_supported) { - return Aria2RetryAction::AlternateResolverFallback; - } +fn aria2_retry_action(payload: &SpawnPayload, error: &str, strike: usize) -> Aria2RetryAction { if is_retryable_aria2_error_for_payload(payload, error) && strike < automatic_retry_limit(payload.max_tries) { @@ -7407,17 +7308,6 @@ fn apply_aria2_normal_reliability_options( Ok(()) } -fn apply_aria2_resolver_options( - options: &mut serde_json::Map, - payload: &SpawnPayload, -) { - if payload.aria2_resolver_mode == Aria2ResolverMode::System - && payload_uses_direct_network(payload) - { - options.insert("async-dns".to_string(), serde_json::json!("false")); - } -} - fn should_apply_aria2_connection_options(payload: &SpawnPayload) -> bool { !payload.is_torrent } @@ -8215,18 +8105,6 @@ fn apply_aria2_torrent_options( options.insert("bt-metadata-only".to_string(), serde_json::json!("false")); options.insert("bt-save-metadata".to_string(), serde_json::json!("false")); options.insert("follow-torrent".to_string(), serde_json::json!("false")); - if torrent_uses_direct_network(payload) - && payload.aria2_resolver_mode == Aria2ResolverMode::System - { - // Aria2's async resolver is independent of the system resolver. TUN - // clients commonly install DNS interception/routing at the system - // resolver boundary, so direct Torrent discovery must use the same - // resolver path as the working HTTP/media clients. Do not override an - // explicitly configured proxy route, where local DNS could leak or - // bypass the proxy's name-resolution policy. - options.insert("async-dns".to_string(), serde_json::json!("false")); - } - let seed_time = payload .torrent_seed_time .map(|value| format_aria2_torrent_number(value, "seed time")) @@ -8574,7 +8452,6 @@ impl SidecarSpawner for ProductionSpawner { if let Some(prox) = proxy_value { options.insert("all-proxy".to_string(), serde_json::json!(prox)); } - apply_aria2_resolver_options(&mut options, payload); let retry_strike = state.queue_manager.aria2_retry_strike(id).await; let transfer_host = transfer_uris .first() @@ -8590,7 +8467,7 @@ impl SidecarSpawner for ProductionSpawner { requested_connections, transfer_connections, transfer_uris.len(), - aria2_effective_resolver_mode(payload), + aria2_resolver_route_for_log(payload), proxy_route_for_log(payload.proxy.as_deref()), admission_started.elapsed().as_millis() ); @@ -9258,11 +9135,6 @@ impl EnqueueItem { let mut item = self; item.strip_torrent_credentials(); let media = item.is_media.unwrap_or(false); - let aria2_resolver_mode = if proxy_uses_direct_network(item.proxy.as_deref()) { - Aria2ResolverMode::System - } else { - Aria2ResolverMode::Automatic - }; let kind = if media { TaskKind::Media } else { @@ -9299,7 +9171,6 @@ impl EnqueueItem { retry_not_found_errors: item.retry_not_found_errors.unwrap_or(false), adaptive_mirror_selection: item.adaptive_mirror_selection.unwrap_or(true), proxy: item.proxy, - aria2_resolver_mode, format_selector: item.format_selector, cookie_source: item.cookie_source, is_media: media, @@ -9898,59 +9769,6 @@ mod tests { .is_err()); } - #[test] - fn aria2_system_resolver_mode_is_per_transfer_and_preserves_automatic_default() { - let mut automatic = serde_json::Map::new(); - apply_aria2_resolver_options( - &mut automatic, - &SpawnPayload { - aria2_resolver_mode: Aria2ResolverMode::Automatic, - ..Default::default() - }, - ); - assert!(!automatic.contains_key("async-dns")); - - let mut system = serde_json::Map::new(); - apply_aria2_resolver_options( - &mut system, - &SpawnPayload { - aria2_resolver_mode: Aria2ResolverMode::System, - ..Default::default() - }, - ); - assert_eq!(system.get("async-dns"), Some(&serde_json::json!("false"))); - } - - #[test] - fn enqueue_route_selects_system_resolver_for_direct_transfers() { - let direct: EnqueueItem = serde_json::from_value(serde_json::json!({ - "id": "direct", - "queue_id": "main", - "url": "https://example.com/file", - "destination": "/tmp", - "filename": "file" - })) - .unwrap(); - assert_eq!( - direct.into_task().payload.aria2_resolver_mode, - Aria2ResolverMode::System - ); - - let proxied: EnqueueItem = serde_json::from_value(serde_json::json!({ - "id": "proxied", - "queue_id": "main", - "url": "https://example.com/file", - "destination": "/tmp", - "filename": "file", - "proxy": "http://proxy.example:8080" - })) - .unwrap(); - assert_eq!( - proxied.into_task().payload.aria2_resolver_mode, - Aria2ResolverMode::Automatic - ); - } - #[test] fn resolver_state_events_are_typed_and_redacted() { let event = DownloadStateEvent::failed( @@ -10198,7 +10016,6 @@ mod tests { &mut options, &SpawnPayload { is_torrent: true, - aria2_resolver_mode: Aria2ResolverMode::System, ..Default::default() }, ) @@ -10216,10 +10033,7 @@ mod tests { options.get("follow-torrent"), Some(&serde_json::json!("false")) ); - assert_eq!( - options.get("async-dns"), - Some(&serde_json::json!("false")) - ); + assert!(!options.contains_key("async-dns")); let mut proxied_options = serde_json::Map::new(); apply_aria2_torrent_options( @@ -10232,54 +10046,6 @@ mod tests { ) .unwrap(); assert!(!proxied_options.contains_key("async-dns")); - assert_eq!( - aria2_effective_resolver_mode(&SpawnPayload { - is_torrent: true, - aria2_resolver_mode: Aria2ResolverMode::System, - ..Default::default() - }), - "system" - ); - assert_eq!( - aria2_effective_resolver_mode(&SpawnPayload { - is_torrent: true, - proxy: Some("http://127.0.0.1:8080".to_string()), - ..Default::default() - }), - "automatic" - ); - } - - #[test] - fn resolver_options_never_force_local_dns_for_configured_proxy_routes() { - let mut direct = serde_json::Map::new(); - apply_aria2_resolver_options( - &mut direct, - &SpawnPayload { - aria2_resolver_mode: Aria2ResolverMode::System, - ..Default::default() - }, - ); - assert_eq!(direct.get("async-dns"), Some(&serde_json::json!("false"))); - - let mut proxied = serde_json::Map::new(); - apply_aria2_resolver_options( - &mut proxied, - &SpawnPayload { - aria2_resolver_mode: Aria2ResolverMode::System, - proxy: Some("http://proxy.example:8080".to_string()), - ..Default::default() - }, - ); - assert!(!proxied.contains_key("async-dns")); - assert_eq!( - aria2_effective_resolver_mode(&SpawnPayload { - aria2_resolver_mode: Aria2ResolverMode::System, - proxy: Some("http://proxy.example:8080".to_string()), - ..Default::default() - }), - "automatic" - ); } #[test] @@ -12319,7 +12085,7 @@ mod tests { let low_speed = "aria2 error code 5: Download speed is too slow"; assert_eq!( - aria2_retry_action(&SpawnPayload::default(), not_found, 0, false), + aria2_retry_action(&SpawnPayload::default(), not_found, 0), Aria2RetryAction::Terminal ); assert_eq!( @@ -12331,7 +12097,6 @@ mod tests { }, not_found, 0, - false, ), Aria2RetryAction::OrdinaryRetry ); @@ -12344,12 +12109,11 @@ mod tests { }, not_found, 1, - false, ), Aria2RetryAction::Terminal ); assert_eq!( - aria2_retry_action(&SpawnPayload::default(), low_speed, 0, false), + aria2_retry_action(&SpawnPayload::default(), low_speed, 0), Aria2RetryAction::Terminal ); assert_eq!( @@ -12361,7 +12125,6 @@ mod tests { }, low_speed, 0, - false, ), Aria2RetryAction::OrdinaryRetry ); @@ -12376,68 +12139,37 @@ mod tests { }, low_speed, 0, - false, ), Aria2RetryAction::Terminal ); } #[test] - fn aria2_name_resolution_error_is_retryable_for_resolver_recovery() { + fn aria2_name_resolution_error_stays_on_bounded_retry_path() { let error = "aria2 error code 19: Name resolution for example.test failed: Could not contact DNS servers."; assert!(is_retryable_aria2_error(error)); - let direct_initial = SpawnPayload { - aria2_resolver_mode: Aria2ResolverMode::System, - ..Default::default() - }; - assert!(should_use_aria2_alternate_resolver_fallback( - &direct_initial, - error, - true, - )); - assert_eq!( - aria2_retry_action(&direct_initial, error, 0, true), - Aria2RetryAction::AlternateResolverFallback - ); - assert!(!should_use_aria2_alternate_resolver_fallback( - &SpawnPayload { - aria2_resolver_mode: Aria2ResolverMode::Automatic, - ..Default::default() - }, - error, - true, - )); - assert!(!should_use_aria2_alternate_resolver_fallback( - &SpawnPayload::default(), - error, - false, - )); assert_eq!( aria2_retry_action( &SpawnPayload { - aria2_resolver_mode: Aria2ResolverMode::System, - max_tries: Some(0), - ..Default::default() - }, - error, - 0, - true, - ), - Aria2RetryAction::AlternateResolverFallback - ); - assert_eq!( - aria2_retry_action( - &SpawnPayload { - aria2_resolver_mode: Aria2ResolverMode::Automatic, max_tries: Some(1), ..Default::default() }, error, 0, - true, ), Aria2RetryAction::OrdinaryRetry ); + assert_eq!( + aria2_retry_action( + &SpawnPayload { + max_tries: Some(0), + ..Default::default() + }, + error, + 0, + ), + Aria2RetryAction::Terminal + ); } #[test] diff --git a/src-tauri/src/torrent.rs b/src-tauri/src/torrent.rs index 51dd778..5f54a93 100644 --- a/src-tauri/src/torrent.rs +++ b/src-tauri/src/torrent.rs @@ -9,6 +9,7 @@ use tokio::io::AsyncReadExt; use crate::ipc::{TorrentFile, TorrentMetadata}; pub const MAX_TORRENT_BYTES: usize = 16 * 1024 * 1024; +const MAX_TORRENT_DHT_NODES: usize = 256; #[derive(Debug, Clone)] pub struct ParsedTorrent { @@ -417,6 +418,12 @@ fn bounded_uri(value: &str, schemes: &[&str]) -> Option { { return None; } + crate::network::validate_url( + &parsed, + schemes, + crate::network::CredentialPolicy::Allow, + ) + .ok()?; Some(parsed.to_string()) } @@ -487,6 +494,40 @@ fn torrent_tracker_metadata_is_safe(root: &BTreeMap, BencodeValue>) -> b true } +fn torrent_nodes_metadata_is_safe(root: &BTreeMap, BencodeValue>) -> bool { + let Some(nodes) = root.get(b"nodes".as_slice()) else { + return true; + }; + let BencodeValue::List(nodes) = nodes else { + return false; + }; + if nodes.len() > MAX_TORRENT_DHT_NODES { + return false; + } + + nodes.iter().all(|node| { + let BencodeValue::List(parts) = node else { + return false; + }; + if parts.len() != 2 { + return false; + } + let BencodeValue::Bytes(host) = &parts[0] else { + return false; + }; + if host.is_empty() || host.len() > crate::queue::MAX_TORRENT_NETWORK_VALUE_LENGTH { + return false; + } + let Ok(host) = std::str::from_utf8(host) else { + return false; + }; + if crate::network::validate_host(host).is_err() { + return false; + } + matches!(&parts[1], BencodeValue::Integer(port) if (1..=u16::MAX as i64).contains(port)) + }) +} + /// Validate the tracker fields before handing original metainfo to Aria2. /// `torrent_details_from_bytes` intentionally omits malformed tracker values /// from its display projection, but Aria2 consumes the original bencode and @@ -501,10 +542,10 @@ pub fn validate_torrent_tracker_metadata(bytes: &[u8]) -> Result<(), String> { BencodeValue::Dict(value) => value, _ => return Err("torrent root is not a dictionary".to_string()), }; - if torrent_tracker_metadata_is_safe(&root) { + if torrent_tracker_metadata_is_safe(&root) && torrent_nodes_metadata_is_safe(&root) { Ok(()) } else { - Err("torrent metadata contains an invalid tracker URI".to_string()) + Err("torrent metadata contains an invalid network destination".to_string()) } } @@ -528,8 +569,25 @@ fn parse_torrent_web_seeds(value: Option<&BencodeValue>) -> Result, let value = String::from_utf8(bytes.clone()) .map_err(|_| "torrent url-list contains invalid UTF-8".to_string())?; let value = value.trim(); - let uri = bounded_uri(value, &["http", "https"]) - .ok_or_else(|| "torrent url-list contains an invalid HTTP(S) web seed".to_string())?; + if value.len() > 2_048 || value.chars().any(char::is_control) { + return Err("torrent url-list contains an invalid HTTP(S) web seed".to_string()); + } + let parsed = url::Url::parse(value) + .map_err(|_| "torrent url-list contains an invalid HTTP(S) web seed".to_string())?; + if !matches!(parsed.scheme(), "http" | "https") + || parsed.host_str().is_none_or(str::is_empty) + || !parsed.username().is_empty() + || parsed.password().is_some() + || parsed.fragment().is_some() + { + return Err("torrent url-list contains an invalid HTTP(S) web seed".to_string()); + } + crate::network::validate_url( + &parsed, + &["http", "https"], + crate::network::CredentialPolicy::Allow, + )?; + let uri = parsed.to_string(); if !normalized.contains(&uri) { normalized.push(uri); } @@ -1367,6 +1425,21 @@ pub async fn remove_managed_torrent( #[cfg(test)] mod tests { use super::*; + use std::collections::BTreeMap; + + fn torrent_with_root_value(key: &[u8], value: BencodeValue) -> Vec { + let info = BencodeValue::Dict(BTreeMap::from([ + (b"length".to_vec(), BencodeValue::Integer(5)), + (b"name".to_vec(), BencodeValue::Bytes(b"test".to_vec())), + ])); + let root = BencodeValue::Dict(BTreeMap::from([ + (b"info".to_vec(), info), + (key.to_vec(), value), + ])); + let mut bytes = Vec::new(); + encode(&root, &mut bytes); + bytes + } #[test] fn parses_single_file_torrent_and_hashes_info_dictionary() { @@ -1560,7 +1633,7 @@ mod tests { assert!(validate_torrent_tracker_metadata( b"d8:announce25:http://127.0.0.1/announce4:infod6:lengthi5e4:name4:testee" ) - .is_ok()); + .is_err()); assert!(validate_torrent_tracker_metadata( b"d8:announce32:https://tracker.example/announce4:infod6:lengthi5e4:name4:testee" ) @@ -1571,6 +1644,40 @@ mod tests { .is_err()); } + #[test] + fn torrent_network_metadata_rejects_local_nodes_and_seeds_without_dns() { + for host in [ + b"127.1".as_slice(), + b"2130706433".as_slice(), + b"[::ffff:127.0.0.1]".as_slice(), + b"localhost".as_slice(), + ] { + let bytes = torrent_with_root_value( + b"nodes", + BencodeValue::List(vec![BencodeValue::List(vec![ + BencodeValue::Bytes(host.to_vec()), + BencodeValue::Integer(6881), + ])]), + ); + assert!( + validate_torrent_tracker_metadata(&bytes).is_err(), + "{host:?}" + ); + } + + let public_node = torrent_with_root_value( + b"nodes", + BencodeValue::List(vec![BencodeValue::List(vec![ + BencodeValue::Bytes(b"node-does-not-resolve.invalid".to_vec()), + BencodeValue::Integer(6881), + ])]), + ); + assert!(validate_torrent_tracker_metadata(&public_node).is_ok()); + + let local_seed = b"d4:infod6:lengthi5e4:name4:teste8:url-list17:http://127.0.0.1/ee"; + assert!(parse_torrent_bytes(local_seed).is_err()); + } + #[test] fn canonical_cache_temporary_names_are_strictly_recognized() { assert!(is_canonical_torrent_temp_file( diff --git a/src-tauri/src/torrent_probe.rs b/src-tauri/src/torrent_probe.rs index 802dd31..878713c 100644 --- a/src-tauri/src/torrent_probe.rs +++ b/src-tauri/src/torrent_probe.rs @@ -8,6 +8,7 @@ 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); +const CANCELLATION_CLEANUP_ATTEMPT_TIMEOUT: Duration = Duration::from_secs(4); #[async_trait] pub(crate) trait RpcClient: Send + Sync { @@ -20,6 +21,14 @@ pub(crate) enum ProbeFailure { Cleanup(String), } +#[derive(Debug, Clone, Copy)] +pub(crate) struct MetadataProbeSchedule { + pub total_timeout: Duration, + pub metadata_timeout: Duration, + pub cleanup_reserve: Duration, + pub poll_interval: Duration, +} + #[allow(dead_code)] pub(crate) async fn run_metadata_probe( client: Arc, @@ -44,9 +53,8 @@ pub(crate) async fn run_metadata_probe( /// Run one metadata probe with separate metadata and cleanup deadlines. /// -/// The split is important for resolver fallback: the first resolver gets its -/// own bounded attempt, while cleanup of its GID consumes only the shared -/// operation deadline before a second resolver may be started. +/// The split keeps metadata polling bounded while reserving a separate window +/// for cleanup of any GID that Aria2 accepted before the probe failed. pub(crate) async fn run_metadata_probe_with_deadlines( client: Arc, source: &str, @@ -58,42 +66,80 @@ pub(crate) async fn run_metadata_probe_with_deadlines( ) -> Result, ProbeFailure> { // This probe only resolves magnet metadata. It must never allow Aria2 to // interpret a downloaded metadata file as another child download because - // the probe cleanup guard owns exactly one GID. + // the probe cleanup guard owns the complete addUri lifecycle. options.insert("follow-torrent".to_string(), json!("false")); options.insert("follow-metalink".to_string(), json!("false")); + let planned_gid = new_probe_gid(); + options.insert("gid".to_string(), json!(&planned_gid)); let mut cleanup_guard = ProbeCleanupGuard::new(Arc::clone(&client), metadata_path); + cleanup_guard.set_planned_gid(planned_gid); + cleanup_guard.set_pending_add(tokio::spawn({ + let client = Arc::clone(&client); + let source = source.to_string(); + async move { + client + .call("aria2.addUri", json!([[source], options])) + .await + } + })); let add_result = tokio::time::timeout( metadata_deadline.saturating_duration_since(Instant::now()), - client.call("aria2.addUri", json!([[source], options])), + cleanup_guard + .pending_add_mut() + .expect("metadata probe add task should be armed"), ) .await; let result = match add_result { Err(_) => { - cleanup_guard.disarm(); - return Err(ProbeFailure::Metadata( + return finish_failed_probe( + &mut cleanup_guard, + cleanup_deadline, "Aria2 could not start magnet metadata resolution before the resolver attempt deadline" .to_string(), - )); + ) + .await; } Ok(Err(error)) => { - cleanup_guard.disarm(); - return Err(ProbeFailure::Metadata(format!( - "Aria2 could not start magnet metadata resolution: {}", - crate::redact_sensitive_text(&error) - ))); + cleanup_guard.take_pending_add(); + return finish_failed_probe( + &mut cleanup_guard, + cleanup_deadline, + format!( + "Aria2 could not start magnet metadata resolution: RPC task failed: {}", + crate::redact_sensitive_text(&error.to_string()) + ), + ) + .await; + } + Ok(Ok(Err(error))) => { + cleanup_guard.take_pending_add(); + return finish_failed_probe( + &mut cleanup_guard, + cleanup_deadline, + format!( + "Aria2 could not start magnet metadata resolution: {}", + crate::redact_sensitive_text(&error) + ), + ) + .await; + } + Ok(Ok(Ok(result))) => { + cleanup_guard.take_pending_add(); + result } - Ok(Ok(result)) => result, }; let gid = match result.as_str().filter(|value| !value.is_empty()) { Some(gid) => gid.to_string(), None => { - cleanup_guard.disarm(); - return Err(ProbeFailure::Metadata( + return finish_failed_probe( + &mut cleanup_guard, + cleanup_deadline, "Aria2 returned an empty metadata probe GID".to_string(), - )); + ) + .await; } }; - cleanup_guard.set_gid(gid.clone()); + cleanup_guard.confirm_gid(gid.clone()); let metadata_result = async { loop { @@ -214,106 +260,61 @@ pub(crate) async fn run_metadata_probe_with_deadlines( metadata_result } -/// Run direct magnet metadata discovery through the system resolver first, -/// then use Aria2's alternate resolver only after the first probe has failed -/// and its GID has been cleaned up. Both attempts share one absolute deadline. -pub(crate) async fn run_metadata_probe_with_resolver_fallback( - client: Arc, - source: &str, - initial_options: Map, - metadata_path: &Path, - initial_resolver_mode: &'static str, - fallback_to_automatic: bool, - total_timeout: Duration, - initial_timeout: Duration, - poll_interval: Duration, -) -> Result, ProbeFailure> { - let operation_started = Instant::now(); - let operation_deadline = operation_started + total_timeout; - let initial_deadline = (operation_started + initial_timeout).min(operation_deadline); - - log::debug!( - "magnet metadata probe [resolver_mode={initial_resolver_mode} outcome=started error_class=none elapsed_ms=0]" - ); - let first_started = Instant::now(); - let first_result = run_metadata_probe_with_deadlines( - Arc::clone(&client), - source, - initial_options.clone(), - metadata_path, - initial_deadline, - operation_deadline, - poll_interval, - ) - .await; - log_probe_result(initial_resolver_mode, first_started, &first_result); - - let should_try_alternate = initial_resolver_mode == "system" - && fallback_to_automatic - && probe_failure_allows_resolver_fallback(&first_result) - && Instant::now() < operation_deadline; - if !should_try_alternate { - return first_result; - } - - // The first probe has completed its lifecycle cleanup. Remove any - // partial metadata before reusing the path so a stale result can never be - // adopted by the alternate resolver. - let clear_result = match tokio::time::timeout( - operation_deadline.saturating_duration_since(Instant::now()), - tokio::fs::remove_file(metadata_path), - ) - .await - { - Ok(Ok(())) => Ok(()), - Ok(Err(error)) if error.kind() == std::io::ErrorKind::NotFound => Ok(()), - Ok(Err(error)) => Err(format!("filesystem error ({:?})", error.kind())), - Err(_) => Err("timed out before the metadata operation deadline".to_string()), - }; - if let Err(error) = clear_result { - let cleanup_failure = ProbeFailure::Cleanup(format!( - "could not clear stale magnet metadata before resolver fallback: {error}" - )); - log::debug!( - "magnet metadata probe [resolver_mode=automatic outcome=cleanup-failure error_class=cleanup elapsed_ms={}]", - Instant::now().duration_since(operation_started).as_millis() - ); - return Err(cleanup_failure); - } - - let mut alternate_options = initial_options; - alternate_options.remove("async-dns"); - log::debug!( - "magnet metadata probe [resolver_mode=automatic outcome=started error_class=none elapsed_ms={}]", - operation_started.elapsed().as_millis() - ); - let alternate_started = Instant::now(); - let alternate_result = run_metadata_probe_with_deadlines( - client, - source, - alternate_options, - metadata_path, - operation_deadline, - operation_deadline, - poll_interval, - ) - .await; - log_probe_result("automatic", alternate_started, &alternate_result); - alternate_result +fn new_probe_gid() -> String { + uuid::Uuid::new_v4().simple().to_string()[..16].to_string() } -fn probe_failure_allows_resolver_fallback(result: &Result, ProbeFailure>) -> bool { - let Err(ProbeFailure::Metadata(error)) = result else { - return false; - }; - // An alternate resolver can only repair a resolver-owned failure. Keep - // malformed metadata, missing output, daemon/status errors, cancellation, - // and every cleanup uncertainty on the original result. The generic - // operation timeout is emitted only by the bounded add/status polling - // path; file-read and addUri-start deadlines use distinct messages so an - // uncertain GID is never duplicated by a fallback attempt. - crate::retry::is_aria2_name_resolution_error(error) - || error == "Aria2 magnet metadata resolution timed out" +async fn finish_failed_probe( + cleanup_guard: &mut ProbeCleanupGuard, + cleanup_deadline: Instant, + message: String, +) -> Result, ProbeFailure> { + match cleanup_guard.cleanup_until(cleanup_deadline).await { + Ok(()) => Err(ProbeFailure::Metadata(message)), + Err(error) => Err(ProbeFailure::Cleanup(format!( + "{message}; failed to clean up magnet metadata probe: {error}" + ))), + } +} + +/// Run magnet metadata discovery through Aria2's asynchronous resolver while +/// reserving part of the absolute deadline for deterministic GID cleanup. A +/// synchronous system-resolver retry is intentionally excluded because +/// Aria2 executes getaddrinfo on its shared event loop and can otherwise +/// stall every transfer and the local RPC control channel. +pub(crate) async fn run_bounded_metadata_probe( + client: Arc, + source: &str, + options: Map, + metadata_path: &Path, + resolver_mode: &'static str, + schedule: MetadataProbeSchedule, +) -> Result, ProbeFailure> { + let operation_started = Instant::now(); + let operation_deadline = operation_started + schedule.total_timeout; + let metadata_operation_deadline = operation_deadline + .checked_sub(schedule.cleanup_reserve) + .filter(|deadline| *deadline >= operation_started) + .unwrap_or(operation_started); + let metadata_deadline = + (operation_started + schedule.metadata_timeout).min(metadata_operation_deadline); + + log::debug!( + "magnet metadata probe [resolver_mode={resolver_mode} outcome=started error_class=none elapsed_ms=0]" + ); + let probe_started = Instant::now(); + let result = run_metadata_probe_with_deadlines( + client, + source, + options, + metadata_path, + metadata_deadline, + operation_deadline, + schedule.poll_interval, + ) + .await; + log_probe_result(resolver_mode, probe_started, &result); + result } fn log_probe_result( @@ -349,7 +350,9 @@ fn probe_error_class(error: &str) -> &'static str { struct ProbeCleanupGuard { client: Arc, - gid: Option, + gids: Vec, + planned_gid: Option, + pending_add: Option>>, probe_dir: Option, } @@ -357,7 +360,9 @@ impl ProbeCleanupGuard { fn new(client: Arc, metadata_path: &Path) -> Self { Self { client, - gid: None, + gids: Vec::new(), + planned_gid: None, + pending_add: None, probe_dir: metadata_path .parent() .filter(|path| !path.as_os_str().is_empty()) @@ -365,21 +370,142 @@ impl ProbeCleanupGuard { } } - fn set_gid(&mut self, gid: String) { - self.gid = Some(gid); + fn set_planned_gid(&mut self, gid: String) { + self.planned_gid = Some(gid); + } + + fn confirm_gid(&mut self, gid: String) { + self.planned_gid = None; + self.gids.clear(); + self.gids.push(gid); + } + + fn set_pending_add(&mut self, task: tokio::task::JoinHandle>) { + self.pending_add = Some(task); + } + + fn pending_add_mut(&mut self) -> Option<&mut tokio::task::JoinHandle>> { + self.pending_add.as_mut() + } + + fn take_pending_add(&mut self) -> Option>> { + self.pending_add.take() + } + + fn adopt_add_result(&mut self, result: &Value) { + if let Some(gid) = result.as_str().filter(|value| !value.is_empty()) { + self.confirm_gid(gid.to_string()); + } + } + + async fn cleanup_until(&mut self, deadline: Instant) -> Result<(), String> { + if self.pending_add.is_some() { + let add_result = tokio::time::timeout( + deadline.saturating_duration_since(Instant::now()), + self.pending_add + .as_mut() + .expect("metadata probe add task should remain armed"), + ) + .await; + match add_result { + Ok(Ok(Ok(result))) => { + self.take_pending_add(); + self.adopt_add_result(&result); + } + Ok(Ok(Err(_))) | Ok(Err(_)) => { + self.take_pending_add(); + } + Err(_) => { + return Err( + "Aria2 addUri did not settle before the metadata cleanup deadline" + .to_string(), + ); + } + } + } + + if let Some(planned_gid) = self.planned_gid.take() { + let Some(probe_dir) = self.probe_dir.as_ref() else { + self.planned_gid = Some(planned_gid); + return Err("cannot verify magnet metadata probe ownership".to_string()); + }; + let ownership = tokio::time::timeout( + deadline.saturating_duration_since(Instant::now()), + query_probe_gid_directory(self.client.as_ref(), &planned_gid), + ) + .await; + match ownership { + Ok(Ok(Some(directory))) if directory == *probe_dir => { + self.gids.push(planned_gid); + } + Ok(Ok(Some(_))) | Ok(Ok(None)) => { + // A planned GID that belongs to another directory, or + // no longer exists, is not ours to remove. This closes the + // collision/ambiguous-add path without force-removing an + // unrelated Aria2 transfer. + } + Ok(Err(error)) => { + self.planned_gid = Some(planned_gid); + return Err(error); + } + Err(_) => { + self.planned_gid = Some(planned_gid); + return Err( + "could not verify magnet metadata probe ownership before cleanup deadline" + .to_string(), + ); + } + } + } + + let mut first_error = None; + for gid in self.gids.clone() { + match tokio::time::timeout( + deadline.saturating_duration_since(Instant::now()), + cleanup_metadata_probe(self.client.as_ref(), &gid), + ) + .await + { + Ok(Ok(())) => {} + Ok(Err(error)) => { + first_error.get_or_insert(error); + } + Err(_) => { + first_error.get_or_insert(format!( + "failed to remove aria2 gid {gid} before the metadata operation deadline" + )); + } + } + if first_error.is_some() { + break; + } + } + if let Some(error) = first_error { + return Err(error); + } + self.disarm(); + Ok(()) } fn disarm(&mut self) { - self.gid = None; + self.gids.clear(); + self.planned_gid = None; + self.pending_add = None; self.probe_dir = None; } } impl Drop for ProbeCleanupGuard { fn drop(&mut self) { - let gid = self.gid.take(); + let mut gids = std::mem::take(&mut self.gids); + let mut planned_gid = self.planned_gid.take(); + let pending_add = self.pending_add.take(); let probe_dir = self.probe_dir.take(); - if gid.is_none() && probe_dir.is_none() { + if gids.is_empty() + && planned_gid.is_none() + && pending_add.is_none() + && probe_dir.is_none() + { return; } @@ -389,26 +515,107 @@ impl Drop for ProbeCleanupGuard { }; let client = Arc::clone(&self.client); runtime.spawn(async move { - let cleanup_result = if let Some(gid) = gid { + if let Some(mut pending_add) = pending_add { + match tokio::time::timeout( + CANCELLATION_CLEANUP_ATTEMPT_TIMEOUT, + &mut pending_add, + ) + .await + { + Ok(Ok(Ok(result))) => { + if let Some(gid) = result.as_str().filter(|value| !value.is_empty()) { + gids.clear(); + gids.push(gid.to_string()); + planned_gid = None; + } + } + Ok(Ok(Err(error))) => log::debug!( + "canceled magnet metadata probe addUri failed before ownership was confirmed: {}", + crate::redact_sensitive_text(&error) + ), + Ok(Err(error)) => log::debug!( + "canceled magnet metadata probe addUri task ended before ownership was confirmed: {error}" + ), + Err(_) => { + log::warn!( + "canceled magnet metadata probe addUri did not settle before cleanup deadline" + ); + // The request may still be accepted by Aria2. Keep the + // probe directory because no safe GID cleanup fence is + // available yet. + return; + } + } + } + + if let Some(candidate) = planned_gid.take() { + let Some(probe_dir) = probe_dir.as_ref() else { + log::warn!( + "canceled magnet metadata probe could not verify planned GID ownership" + ); + return; + }; + match tokio::time::timeout( + CANCELLATION_CLEANUP_ATTEMPT_TIMEOUT, + query_probe_gid_directory(client.as_ref(), &candidate), + ) + .await + { + Ok(Ok(Some(directory))) if directory == *probe_dir => { + gids.push(candidate); + } + Ok(Ok(Some(_))) | Ok(Ok(None)) => { + // The candidate is either another transfer's GID or + // already gone; neither case is safe to force-remove. + } + Ok(Err(error)) => { + log::warn!( + "canceled magnet metadata probe ownership check failed: {}", + crate::redact_sensitive_text(&error) + ); + return; + } + Err(_) => { + log::warn!( + "canceled magnet metadata probe ownership check timed out" + ); + return; + } + } + } + + let mut cleanup_result = None; + for gid in gids { let mut last_error = None; for attempt in 0..CANCELLATION_CLEANUP_ATTEMPTS { - match cleanup_metadata_probe(client.as_ref(), &gid).await { - Ok(()) => { + match tokio::time::timeout( + CANCELLATION_CLEANUP_ATTEMPT_TIMEOUT, + cleanup_metadata_probe(client.as_ref(), &gid), + ) + .await + { + Ok(Ok(())) => { last_error = None; break; } - Err(error) => { + Ok(Err(error)) => { last_error = Some(error); - if attempt + 1 < CANCELLATION_CLEANUP_ATTEMPTS { - tokio::time::sleep(CANCELLATION_CLEANUP_INTERVAL).await; - } + } + Err(_) => { + last_error = Some( + "cleanup attempt exceeded its bounded RPC deadline".to_string(), + ); } } + if last_error.is_some() && attempt + 1 < CANCELLATION_CLEANUP_ATTEMPTS { + tokio::time::sleep(CANCELLATION_CLEANUP_INTERVAL).await; + } } - last_error - } else { - None - }; + if last_error.is_some() { + cleanup_result = last_error; + break; + } + } if let Some(error) = cleanup_result { log::warn!( "canceled magnet metadata probe cleanup failed: {}", @@ -451,6 +658,31 @@ async fn cleanup_metadata_probe(client: &C, gid: &str) -> Result<( } } +async fn query_probe_gid_directory( + client: &C, + gid: &str, +) -> Result, String> { + let result = match client + .call("aria2.tellStatus", json!([gid, ["status", "dir"]])) + .await + { + Ok(result) => result, + Err(error) if crate::aria2_gid_not_found(&error) => return Ok(None), + Err(error) => { + return Err(format!( + "failed to verify aria2 gid {gid} ownership: {}", + crate::redact_sensitive_text(&error) + )); + } + }; + let directory = result + .get("dir") + .and_then(Value::as_str) + .filter(|value| !value.is_empty()) + .ok_or_else(|| format!("aria2 gid {gid} ownership response has no directory"))?; + Ok(Some(PathBuf::from(directory))) +} + async fn aria2_status(client: &C, gid: &str) -> Result { let result = client .call("aria2.tellStatus", json!([gid, ["status"]])) @@ -527,6 +759,7 @@ mod tests { RpcError(String), HttpError(StatusCode, String), Malformed(String), + EchoLastString, Delay(Duration, Box), Hang, } @@ -721,11 +954,12 @@ mod tests { .unwrap_or_else(|| { ScriptedReply::RpcError(format!("unexpected scripted RPC method {method}")) }); - scripted_reply_response(id, reply, state.termination.subscribe()).await + scripted_reply_response(id, params, reply, state.termination.subscribe()).await } async fn scripted_reply_response( id: Value, + params: Value, mut reply: ScriptedReply, mut termination: watch::Receiver, ) -> Response { @@ -761,6 +995,19 @@ mod tests { ScriptedReply::Malformed(body) => { return (StatusCode::OK, Body::from(body)).into_response(); } + ScriptedReply::EchoLastString => { + let value = params + .as_array() + .and_then(|params| params.last()) + .and_then(Value::as_str) + .unwrap_or_default(); + return Json(json!({ + "jsonrpc": "2.0", + "id": id, + "result": value, + })) + .into_response(); + } } } } @@ -1161,125 +1408,15 @@ mod tests { ScriptedReply::Result(json!({ "status": status })) } - #[tokio::test(flavor = "current_thread")] - async fn resolver_fallback_cleans_the_first_gid_and_rejects_stale_metadata() { - let server = ScriptedRpcServer::start(scripts([ - ( - "aria2.addUri", - vec![ - ScriptedReply::Result(json!("gid-1")), - ScriptedReply::Delay( - Duration::from_millis(50), - Box::new(ScriptedReply::Result(json!("gid-2"))), - ), - ], - ), - ( - "aria2.tellStatus", - vec![ - ScriptedReply::Result(json!({ - "status": "error", - "errorCode": "19", - "errorMessage": "Name resolution failed", - })), - status_reply("removed"), - status_reply("complete"), - status_reply("removed"), - ], - ), - ( - "aria2.forceRemove", - vec![ - ScriptedReply::Result(json!("gid-1")), - ScriptedReply::Result(json!("gid-2")), - ], - ), - ])) - .await; - let (_temporary, probe_dir, metadata_path) = probe_fixture().await; - let probe_path = metadata_path.clone(); - let client = server.client(); - let task = tokio::spawn(async move { - let mut options = Map::new(); - options.insert("async-dns".to_string(), json!("false")); - run_metadata_probe_with_resolver_fallback( - client, - "magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567", - options, - &probe_path, - "system", - true, - HTTP_PROBE_TIMEOUT, - HTTP_INITIAL_PROBE_TIMEOUT, - Duration::ZERO, - ) - .await - }); - - server.wait_for_method("aria2.addUri", 2).await; - assert!( - !metadata_path.exists(), - "the first probe's stale metadata must be removed before fallback" - ); - tokio::fs::write(&metadata_path, b"metadata from alternate resolver") - .await - .expect("alternate metadata fixture should be writable"); - let bytes = tokio::time::timeout(HTTP_TEST_TIMEOUT, task) - .await - .expect("resolver fallback should remain bounded") - .expect("resolver fallback task should not panic") - .expect("alternate resolver should complete the probe"); - assert_eq!(bytes, b"metadata from alternate resolver"); - - let calls = server.calls(); - let add_calls = calls - .iter() - .filter(|(method, _)| method == "aria2.addUri") - .collect::>(); - assert_eq!(add_calls.len(), 2); - let first_options = add_calls[0] - .1 - .as_array() - .and_then(|params| params.get(2)) - .and_then(Value::as_object) - .expect("first addUri options should be recorded"); - assert_eq!(first_options.get("async-dns"), Some(&json!("false"))); - let alternate_options = add_calls[1] - .1 - .as_array() - .and_then(|params| params.get(2)) - .and_then(Value::as_object) - .expect("alternate addUri options should be recorded"); - assert!(!alternate_options.contains_key("async-dns")); - - let first_force_remove = calls - .iter() - .position(|(method, _)| method == "aria2.forceRemove") - .expect("first probe must be cleaned up"); - let second_add = calls - .iter() - .rposition(|(method, _)| method == "aria2.addUri") - .expect("alternate probe must be started"); - assert!( - first_force_remove < second_add, - "alternate probing must wait for first-GID cleanup" - ); - assert_eq!( - calls - .iter() - .filter(|(method, _)| method == "aria2.forceRemove") - .count(), - 2 - ); - - tokio::fs::remove_dir_all(&probe_dir) - .await - .expect("successful fallback fixture should be removable"); - server.shutdown().await; + fn status_reply_with_directory(status: &str, directory: &Path) -> ScriptedReply { + ScriptedReply::Result(json!({ + "status": status, + "dir": directory.to_string_lossy().to_string(), + })) } #[tokio::test(flavor = "current_thread")] - async fn resolver_fallback_does_not_repeat_invalid_metadata_failures() { + async fn bounded_probe_returns_metadata_failure_without_a_second_add() { let server = ScriptedRpcServer::start(scripts([ ("aria2.addUri", vec![ScriptedReply::Result(json!("gid-1"))]), ( @@ -1300,16 +1437,18 @@ mod tests { ])) .await; let (_temporary, probe_dir, metadata_path) = probe_fixture().await; - let error = run_metadata_probe_with_resolver_fallback( + let error = run_bounded_metadata_probe( server.client(), "magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567", Map::new(), &metadata_path, - "system", - true, - HTTP_PROBE_TIMEOUT, - HTTP_INITIAL_PROBE_TIMEOUT, - Duration::ZERO, + "automatic", + MetadataProbeSchedule { + total_timeout: HTTP_PROBE_TIMEOUT, + metadata_timeout: HTTP_INITIAL_PROBE_TIMEOUT, + cleanup_reserve: Duration::from_secs(1), + poll_interval: Duration::ZERO, + }, ) .await .expect_err("invalid metadata failure should remain on the initial route"); @@ -1329,7 +1468,119 @@ mod tests { } #[tokio::test(flavor = "current_thread")] - async fn resolver_fallback_does_not_run_after_cleanup_uncertainty() { + async fn generic_metadata_timeout_does_not_enter_blocking_system_resolver() { + let server = ScriptedRpcServer::start(scripts([ + ("aria2.addUri", vec![ScriptedReply::Result(json!("gid-1"))]), + ( + "aria2.tellStatus", + vec![ScriptedReply::Hang, status_reply("removed")], + ), + ( + "aria2.forceRemove", + vec![ScriptedReply::Result(json!("gid-1"))], + ), + ])) + .await; + let (_temporary, probe_dir, metadata_path) = probe_fixture().await; + let error = run_bounded_metadata_probe( + server.client(), + "magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567", + Map::new(), + &metadata_path, + "automatic", + MetadataProbeSchedule { + total_timeout: Duration::from_millis(500), + metadata_timeout: Duration::from_millis(100), + cleanup_reserve: Duration::from_millis(200), + poll_interval: Duration::ZERO, + }, + ) + .await + .expect_err("a metadata timeout should remain on the non-blocking route"); + assert!(matches!(error, ProbeFailure::Metadata(message) if message.contains("timed out"))); + assert_eq!( + server + .calls() + .iter() + .filter(|(method, _)| method == "aria2.addUri") + .count(), + 1, + "peer or tracker timeouts must not trigger the synchronous resolver" + ); + tokio::fs::remove_dir_all(&probe_dir) + .await + .expect("timed-out probe fixture should be removable"); + server.shutdown().await; + } + + #[tokio::test(flavor = "current_thread")] + async fn delayed_add_response_is_fenced_before_probe_cleanup() { + let (_temporary, probe_dir, metadata_path) = probe_fixture().await; + let server = ScriptedRpcServer::start(scripts([ + ( + "aria2.addUri", + vec![ScriptedReply::Delay( + Duration::from_millis(250), + Box::new(ScriptedReply::Malformed("{".to_string())), + )], + ), + ("aria2.forceRemove", vec![ScriptedReply::EchoLastString]), + ( + "aria2.tellStatus", + vec![ + status_reply_with_directory("active", &probe_dir), + status_reply("removed"), + ], + ), + ])) + .await; + let error = run_bounded_metadata_probe( + server.client(), + "magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567", + Map::new(), + &metadata_path, + "automatic", + MetadataProbeSchedule { + total_timeout: Duration::from_secs(1), + metadata_timeout: Duration::from_millis(50), + cleanup_reserve: Duration::from_millis(700), + poll_interval: Duration::ZERO, + }, + ) + .await + .expect_err("a delayed add response should remain a metadata failure"); + assert!(matches!(error, ProbeFailure::Metadata(message) if message.contains("could not start"))); + + let calls = server.calls(); + let add_params = calls + .iter() + .find(|(method, _)| method == "aria2.addUri") + .expect("addUri should be recorded") + .1 + .as_array() + .expect("addUri params should be an array"); + let planned_gid = add_params[2] + .get("gid") + .and_then(Value::as_str) + .expect("probe should reserve a GID before addUri"); + assert_eq!(planned_gid.len(), 16); + assert!(planned_gid.bytes().all(|byte| byte.is_ascii_hexdigit())); + let removed_gid = calls + .iter() + .find(|(method, _)| method == "aria2.forceRemove") + .and_then(|(_, params)| params.as_array()) + .and_then(|params| params.last()) + .and_then(Value::as_str) + .expect("cleanup should use the reserved GID"); + assert_eq!(removed_gid, planned_gid); + tokio::fs::remove_dir_all(&probe_dir) + .await + .expect("the caller should remove the probe directory after cleanup succeeds"); + server.shutdown().await; + } + + #[tokio::test(flavor = "current_thread")] + async fn cleanup_uncertainty_stops_the_bounded_probe() { let server = ScriptedRpcServer::start(scripts([ ("aria2.addUri", vec![ScriptedReply::Result(json!("gid-1"))]), ( @@ -1350,19 +1601,21 @@ mod tests { ])) .await; let (_temporary, probe_dir, metadata_path) = probe_fixture().await; - let error = run_metadata_probe_with_resolver_fallback( + let error = run_bounded_metadata_probe( server.client(), "magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567", Map::new(), &metadata_path, - "system", - true, - HTTP_PROBE_TIMEOUT, - HTTP_INITIAL_PROBE_TIMEOUT, - Duration::ZERO, + "automatic", + MetadataProbeSchedule { + total_timeout: HTTP_PROBE_TIMEOUT, + metadata_timeout: HTTP_INITIAL_PROBE_TIMEOUT, + cleanup_reserve: Duration::from_secs(1), + poll_interval: Duration::ZERO, + }, ) .await - .expect_err("cleanup uncertainty must stop resolver fallback"); + .expect_err("cleanup uncertainty must fail the probe"); assert!(matches!(error, ProbeFailure::Cleanup(_))); assert_eq!( server @@ -1379,7 +1632,7 @@ mod tests { } #[tokio::test(flavor = "current_thread")] - async fn resolver_fallback_is_not_started_after_cancellation() { + async fn cancellation_cleans_the_only_non_blocking_probe() { let server = ScriptedRpcServer::start(scripts([ ("aria2.addUri", vec![ScriptedReply::Result(json!("gid-1"))]), ( @@ -1396,16 +1649,18 @@ mod tests { let probe_path = metadata_path.clone(); let client = server.client(); let task = tokio::spawn(async move { - run_metadata_probe_with_resolver_fallback( + run_bounded_metadata_probe( client, "magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567", Map::new(), &probe_path, - "system", - true, - Duration::from_secs(60), - Duration::from_secs(20), - Duration::from_secs(60), + "automatic", + MetadataProbeSchedule { + total_timeout: Duration::from_secs(60), + metadata_timeout: Duration::from_secs(20), + cleanup_reserve: Duration::from_secs(5), + poll_interval: Duration::from_secs(60), + }, ) .await }); @@ -1420,7 +1675,7 @@ mod tests { .filter(|(method, _)| method == "aria2.addUri") .count(), 1, - "cancellation must not start the alternate resolver" + "cancellation must not start a second probe" ); wait_for_path(&probe_dir, false).await; server.terminate().await; @@ -1523,16 +1778,23 @@ mod tests { } #[tokio::test(flavor = "current_thread")] - async fn http_rpc_harness_reports_add_uri_failure_without_fabricating_cleanup() { - let server = ScriptedRpcServer::start(scripts([( - "aria2.addUri", - vec![ScriptedReply::HttpError( - StatusCode::SERVICE_UNAVAILABLE, - "daemon is starting".to_string(), - )], - )])) + async fn http_rpc_harness_does_not_remove_an_unrelated_planned_gid() { + let (temporary, probe_dir, metadata_path) = probe_fixture().await; + let unrelated_directory = temporary.path().join("other-transfer"); + let server = ScriptedRpcServer::start(scripts([ + ( + "aria2.addUri", + vec![ScriptedReply::HttpError( + StatusCode::SERVICE_UNAVAILABLE, + "daemon is starting".to_string(), + )], + ), + ( + "aria2.tellStatus", + vec![status_reply_with_directory("active", &unrelated_directory)], + ), + ])) .await; - let (_temporary, probe_dir, metadata_path) = probe_fixture().await; let error = run_metadata_probe( server.client(), "magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567", @@ -1553,9 +1815,12 @@ mod tests { .iter() .map(|(method, _)| method.as_str()) .collect::>(), - vec!["aria2.addUri"] + vec!["aria2.addUri", "aria2.tellStatus"] ); assert!(probe_dir.exists()); + tokio::fs::remove_dir_all(&probe_dir) + .await + .expect("unrelated planned GID fixture should be removable"); server.shutdown().await; } @@ -1789,6 +2054,12 @@ mod tests { Duration::from_millis(250), Box::new(ScriptedReply::Result(json!("gid-1"))), )], + ), ( + "aria2.forceRemove", + vec![ScriptedReply::EchoLastString], + ), ( + "aria2.tellStatus", + vec![status_reply("removed")], )])) .await; let (_temporary, probe_dir, metadata_path) = probe_fixture().await; @@ -1808,6 +2079,7 @@ mod tests { server.wait_for_method("aria2.addUri", 1).await; task.abort(); let _ = task.await; + server.wait_for_method("aria2.forceRemove", 1).await; wait_for_path(&probe_dir, false).await; server.shutdown().await; diff --git a/src-tauri/tests/queue_manager.rs b/src-tauri/tests/queue_manager.rs index 3548272..ebc890d 100644 --- a/src-tauri/tests/queue_manager.rs +++ b/src-tauri/tests/queue_manager.rs @@ -1,6 +1,6 @@ use firelink_lib::queue::{ - Aria2RecreateOutcome, Aria2RefreshOutcome, Aria2ResolverMode, QueueManager, QueuedTask, - SidecarSpawner, SpawnPayload, TaskKind, MEDIA_RUN_CANCELLED, + Aria2RecreateOutcome, Aria2RefreshOutcome, QueueManager, QueuedTask, SidecarSpawner, + SpawnPayload, TaskKind, MEDIA_RUN_CANCELLED, }; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; @@ -27,15 +27,7 @@ struct CountingSpawner { torrent_peer_options_release: tokio::sync::Notify, add_speed_limits: std::sync::Mutex>>, add_peer_options: std::sync::Mutex, Option)>>, - add_resolver_modes: std::sync::Mutex>, - add_transfer_context: std::sync::Mutex< - Vec<( - Aria2ResolverMode, - Option, - Option, - Option, - )>, - >, + add_transfer_context: std::sync::Mutex, Option, Option)>>, block_speed_limit: std::sync::atomic::AtomicBool, speed_limit_started: tokio::sync::Notify, speed_limit_release: tokio::sync::Notify, @@ -220,7 +212,6 @@ impl CountingSpawner { torrent_peer_options_release: tokio::sync::Notify::new(), add_speed_limits: std::sync::Mutex::new(Vec::new()), add_peer_options: std::sync::Mutex::new(Vec::new()), - add_resolver_modes: std::sync::Mutex::new(Vec::new()), add_transfer_context: std::sync::Mutex::new(Vec::new()), block_speed_limit: std::sync::atomic::AtomicBool::new(false), speed_limit_started: tokio::sync::Notify::new(), @@ -305,12 +296,7 @@ impl firelink_lib::queue::SidecarSpawner for CountingSpawner { payload.torrent_max_peers, payload.torrent_peer_speed_limit.clone(), )); - self.add_resolver_modes - .lock() - .unwrap() - .push(payload.aria2_resolver_mode); self.add_transfer_context.lock().unwrap().push(( - payload.aria2_resolver_mode, payload.headers.clone(), payload.proxy.clone(), payload.connections, @@ -2297,15 +2283,13 @@ async fn transient_aria2_error_reissues_after_backoff() { } #[tokio::test] -async fn resolver_failure_uses_one_system_fallback_without_retry_budget() { +async fn resolver_failure_retries_without_entering_blocking_system_dns() { use firelink_lib::queue::PendingOutcome; let (mgr, spawner) = make_manager(1); let manager = Arc::new(mgr); - manager.set_aria2_async_dns_supported(true); let mut task = aria2_task("resolver-fallback"); - task.payload.aria2_resolver_mode = Aria2ResolverMode::System; - task.payload.max_tries = Some(0); + task.payload.max_tries = Some(1); task.payload.headers = Some("X-Test: retained".to_string()); manager.push(task).await.unwrap(); @@ -2341,33 +2325,19 @@ async fn resolver_failure_uses_one_system_fallback_without_retry_budget() { } }) .await - .expect("resolver failure should re-add once with Aria2's alternate resolver"); - assert_eq!( - *spawner.add_resolver_modes.lock().unwrap(), - vec![Aria2ResolverMode::System, Aria2ResolverMode::Automatic] - ); + .expect("resolver failure should re-add once on the non-blocking resolver"); assert_eq!( *spawner.add_transfer_context.lock().unwrap(), vec![ - ( - Aria2ResolverMode::System, - Some("X-Test: retained".to_string()), - None, - None, - ), - ( - Aria2ResolverMode::Automatic, - Some("X-Test: retained".to_string()), - None, - None, - ), + (Some("X-Test: retained".to_string()), None, None), + (Some("X-Test: retained".to_string()), None, None), ] ); assert_eq!(spawner.add_uri_calls.load(Ordering::SeqCst), 2); - // A second resolver failure is now on the alternate mode. With - // max_tries=0 it must terminate instead of switching back or consuming - // another add. + // The configured retry budget is exhausted after one non-blocking retry. + // A repeated DNS error must terminate without entering system DNS or + // scheduling another add. manager .handle_aria2_event( "gid-2",