diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 8169743..7e3cddb 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -6435,6 +6435,30 @@ fn ensure_aria2_gid_result( } } +fn ensure_aria2_ok_result(method: &str, result: &serde_json::Value) -> Result<(), String> { + match result.as_str() { + Some("OK") => Ok(()), + Some(value) => Err(format!( + "aria2.{method} returned unexpected result {value}" + )), + None => Err(format!("aria2.{method} returned a non-string result")), + } +} + +const ARIA2_RECONCILIATION_STATUS_FIELDS: &[&str] = &[ + "status", + "seeder", + "errorCode", + "errorMessage", + "completedLength", + "totalLength", + "uploadLength", +]; + +fn aria2_reconciliation_status_params(gid: &str) -> serde_json::Value { + serde_json::json!([gid, ARIA2_RECONCILIATION_STATUS_FIELDS]) +} + async fn aria2_download_status(port: u16, secret: &str, gid: &str) -> Result { let result = rpc_call( port, @@ -6543,7 +6567,7 @@ async fn reconcile_aria2_downloads(app_handle: &tauri::AppHandle) { port, &secret, "aria2.tellStatus", - serde_json::json!([gid, ["status", "seeder", "errorCode", "errorMessage", "completedLength", "totalLength"]]), + aria2_reconciliation_status_params(&gid), ) .await { @@ -7005,7 +7029,7 @@ async fn validate_torrent_enqueue( item.mirrors = (!legacy_web_seeds.is_empty()).then_some(legacy_web_seeds.join("\n")); if let Some(path) = item.torrent_path.as_deref() { let path = crate::torrent::validate_managed_torrent_path(app_handle, &item.id, path)?; - let bytes = std::fs::read(path) + let bytes = crate::torrent::read_bounded_torrent_bytes_sync(&path) .map_err(|error| format!("could not read cached torrent metadata: {error}"))?; let metadata = crate::torrent::parse_torrent_bytes(&bytes)?; let normalized_user_web_seeds = queue::normalize_torrent_web_seeds( @@ -7120,7 +7144,7 @@ fn expected_torrent_output_paths( return Ok(None); }; let torrent_path = crate::torrent::validate_managed_torrent_path(app_handle, id, torrent_path)?; - let bytes = std::fs::read(torrent_path) + let bytes = crate::torrent::read_bounded_torrent_bytes_sync(&torrent_path) .map_err(|error| format!("could not read cached torrent metadata: {error}"))?; let metadata = crate::torrent::parse_torrent_bytes(&bytes)?; let selected = crate::torrent::validate_selected_indices( @@ -7585,7 +7609,7 @@ async fn rekey_torrent_metadata( &source.to_string_lossy(), ) .map_err(AppError::Internal)?; - let bytes = tokio::fs::read(&source) + let bytes = crate::torrent::read_bounded_torrent_bytes(&source) .await .map_err(|error| { AppError::Internal(format!("could not read cached torrent metadata: {error}")) @@ -8237,7 +8261,7 @@ async fn get_torrent_file_selection( .as_deref() .ok_or_else(|| "Torrent metadata is not resolved yet".to_string())?; let path = crate::torrent::validate_managed_torrent_path(&app_handle, &id, path)?; - let bytes = tokio::fs::read(path) + let bytes = crate::torrent::read_bounded_torrent_bytes(&path) .await .map_err(|error| format!("could not read cached torrent metadata: {error}"))?; let metadata = crate::torrent::parse_torrent_bytes(&bytes)?; @@ -8279,7 +8303,7 @@ async fn set_torrent_file_selection( .as_deref() .ok_or_else(|| "Torrent metadata is not resolved yet".to_string())?; let path = crate::torrent::validate_managed_torrent_path(&app_handle, &id, path)?; - let bytes = tokio::fs::read(path) + let bytes = crate::torrent::read_bounded_torrent_bytes(&path) .await .map_err(|error| format!("could not read cached torrent metadata: {error}"))?; let metadata = crate::torrent::parse_torrent_bytes(&bytes)?; @@ -8419,7 +8443,7 @@ async fn get_torrent_details( .as_deref() .ok_or_else(|| "Torrent metadata is not resolved yet".to_string())?; let path = crate::torrent::validate_managed_torrent_path(&app_handle, &id, path)?; - let bytes = tokio::fs::read(path) + let bytes = crate::torrent::read_bounded_torrent_bytes(&path) .await .map_err(|error| format!("could not read cached torrent metadata: {error}"))?; crate::torrent::torrent_details_from_bytes(&bytes) @@ -8457,7 +8481,7 @@ async fn get_torrent_magnet_link( .as_deref() .ok_or_else(|| "Torrent metadata is not resolved yet".to_string())?; let path = crate::torrent::validate_managed_torrent_path(&app_handle, &id, path)?; - let bytes = tokio::fs::read(path) + let bytes = crate::torrent::read_bounded_torrent_bytes(&path) .await .map_err(|_| "Torrent metadata is unavailable".to_string())?; let details = crate::torrent::torrent_details_from_bytes(&bytes)?; @@ -8510,7 +8534,7 @@ async fn export_torrent_metadata( .as_deref() .ok_or_else(|| "Torrent metadata is not resolved yet".to_string())?; let path = crate::torrent::validate_managed_torrent_path(&app_handle, &id, path)?; - let bytes = tokio::fs::read(path) + let bytes = crate::torrent::read_bounded_torrent_bytes(&path) .await .map_err(|_| "Torrent metadata is unavailable".to_string())?; let details = crate::torrent::torrent_details_from_bytes(&bytes)?; @@ -9131,7 +9155,7 @@ async fn move_torrent_data( .as_deref() .ok_or_else(|| "Torrent metadata is not resolved yet".to_string())?; let torrent_path = crate::torrent::validate_managed_torrent_path(&app_handle, &id, torrent_path)?; - let bytes = tokio::fs::read(&torrent_path) + let bytes = crate::torrent::read_bounded_torrent_bytes(&torrent_path) .await .map_err(|_| "Torrent metadata is unavailable".to_string())?; let metadata = crate::torrent::parse_torrent_bytes(&bytes)?; @@ -9635,7 +9659,7 @@ async fn verify_torrent_data( .as_deref() .ok_or_else(|| "Torrent metadata is not resolved yet".to_string())?; let path = crate::torrent::validate_managed_torrent_path(&app_handle, &id, path)?; - let bytes = tokio::fs::read(path) + let bytes = crate::torrent::read_bounded_torrent_bytes(&path) .await .map_err(|error| format!("could not read cached torrent metadata: {error}"))?; let metadata = crate::torrent::parse_torrent_bytes(&bytes)?; @@ -9899,7 +9923,7 @@ async fn normalize_persisted_torrent_web_seeds( .as_deref() .ok_or_else(|| "Torrent metadata is unavailable for web-seed management".to_string())?; let path = crate::torrent::validate_managed_torrent_path(app_handle, id, path)?; - let bytes = tokio::fs::read(path) + let bytes = crate::torrent::read_bounded_torrent_bytes(&path) .await .map_err(|error| format!("could not read cached Torrent metadata: {error}"))?; let metadata = crate::torrent::parse_torrent_bytes(&bytes)?; @@ -10228,7 +10252,7 @@ async fn set_torrent_max_open_files( serde_json::json!([{"bt-max-open-files": max_open_files.to_string()}]), ) .await - .map(|_| ()) + .and_then(|result| ensure_aria2_ok_result("changeGlobalOption", &result)) .map_err(|error| format!("Failed to set Torrent maximum open files: {error}")) } @@ -10248,7 +10272,7 @@ async fn set_torrent_overall_upload_limit( serde_json::json!([{"max-overall-upload-limit": limit_str}]), ) .await - .map(|_| ()) + .and_then(|result| ensure_aria2_ok_result("changeGlobalOption", &result)) .map_err(|error| format!("Failed to set Torrent overall upload limit: {error}")) } @@ -10274,6 +10298,7 @@ async fn set_global_speed_limit( serde_json::json!([{"max-overall-download-limit": limit_str}]), ) .await + .and_then(|result| ensure_aria2_ok_result("changeGlobalOption", &result)) .map(|_| { state .queue_manager @@ -11965,6 +11990,29 @@ mod tests { assert!(!magnet.contains("passkey")); } + #[test] + fn aria2_global_option_results_require_the_documented_ok_value() { + assert!(super::ensure_aria2_ok_result("changeGlobalOption", &json!("OK")).is_ok()); + let unexpected = super::ensure_aria2_ok_result("changeGlobalOption", &json!("accepted")) + .expect_err("unexpected global-option results must not claim success"); + assert!(unexpected.contains("unexpected result accepted")); + let malformed = super::ensure_aria2_ok_result("changeGlobalOption", &json!({})) + .expect_err("non-string global-option results must not claim success"); + assert!(malformed.contains("non-string result")); + } + + #[test] + fn aria2_reconciliation_requests_upload_length_for_torrent_telemetry() { + let params = super::aria2_reconciliation_status_params("gid-1"); + let fields = params + .get(1) + .and_then(serde_json::Value::as_array) + .expect("reconciliation params should contain a status field list"); + assert!(fields + .iter() + .any(|field| field.as_str() == Some("uploadLength"))); + } + #[test] fn torrent_move_root_mapping_handles_single_file_output() { let root = tempfile::tempdir_in(std::env::current_dir().unwrap()).unwrap(); @@ -16460,7 +16508,7 @@ pub fn run() { poll_port.load(std::sync::atomic::Ordering::Relaxed), &poll_secret, "aria2.tellStatus", - serde_json::json!([gid, ["status", "seeder", "errorCode", "errorMessage", "completedLength", "totalLength"]]), + aria2_reconciliation_status_params(&gid), ) .await { diff --git a/src-tauri/src/queue.rs b/src-tauri/src/queue.rs index f407348..99c61ac 100644 --- a/src-tauri/src/queue.rs +++ b/src-tauri/src/queue.rs @@ -1517,6 +1517,12 @@ impl QueueManager { }) } + fn seed_starting(&self, id: &str) -> bool { + self.seed_capacity + .lock() + .map_or(false, |state| state.starting.contains(id)) + } + fn add_seed_waiter(&self, waiter: SeedWaiter) { if let Ok(mut state) = self.seed_capacity.lock() { if !state.owners.contains(&waiter.id) @@ -1825,14 +1831,113 @@ impl QueueManager { // An unverified unpause must not release either the seed // owner or its download permit: the daemon may already be // active even though the RPC/status check was unavailable. - // Keep the item fenced in the starting state until a - // terminal event or an explicit user pause resolves it. + // Keep the item fenced in the starting state and retry the + // status-verified control until the daemon or a newer + // lifecycle resolves it. log::warn!( "Torrent seed resume [{}] could not be verified; retaining seed ownership and permit: {}", id, error ); self.emit_waiting_to_seed(&id, None); + let retry_waiter = SeedWaiter { + id: id.clone(), + queue_id: waiter.queue_id.clone(), + lifecycle_generation: waiter.lifecycle_generation, + }; + let manager = Arc::clone(&self); + tauri::async_runtime::spawn(async move { + manager + .retry_ambiguous_seed_resume(retry_waiter, gid, epoch) + .await; + }); + } + } + } + + async fn retry_ambiguous_seed_resume( + self: Arc, + waiter: SeedWaiter, + gid: String, + epoch: u64, + ) { + let mut delay = Duration::from_millis(250); + loop { + tokio::select! { + _ = tokio::time::sleep(delay) => {} + _ = self.notify.notified() => {} + } + + let _control_guard = self.acquire_aria2_control(&waiter.id).await; + let lifecycle_current = self + .is_registered_generation_or_legacy(&waiter.id, waiter.lifecycle_generation) + .await + && self.seed_starting(&waiter.id) + && matches!(self.active_kind(&waiter.id).await, Some(TaskKind::Aria2)) + && self.has_active_permit(&waiter.id).await + && self.aria2_gid_for_download(&waiter.id).as_deref() == Some(gid.as_str()) + && self.is_aria2_control_epoch_current(&waiter.id, epoch).await + && self.is_current_aria2_gid_mapping( + &gid, + &Aria2GidMapping { + id: waiter.id.clone(), + epoch, + }, + ); + if !lifecycle_current { + return; + } + + match self.spawner.resume_for_seed(&gid).await { + Ok(Aria2SeedControlOutcome::Resumed) => { + if !self.is_aria2_control_epoch_current(&waiter.id, epoch).await + || !self.is_current_aria2_gid_mapping( + &gid, + &Aria2GidMapping { + id: waiter.id.clone(), + epoch, + }, + ) + { + let _ = self.spawner.pause_for_seed(&gid).await; + let _ = self.spawner.pause_for_seed(&gid).await; + self.release_download_permit_for_seed(&waiter.id).await; + self.abandon_seed_start(&waiter.id); + return; + } + self.record_seed_started(&waiter.id).await; + self.release_download_permit_for_seed(&waiter.id).await; + self.finish_seed_start(&waiter.id); + self.emit_state(&waiter.id, DownloadStatus::Seeding); + return; + } + Ok(Aria2SeedControlOutcome::Complete) => { + self.finish_seed_start(&waiter.id); + self.apply_completion_locked(&waiter.id, PendingOutcome::Complete) + .await; + return; + } + Ok(Aria2SeedControlOutcome::Paused) => { + let remaining = self.capture_seed_remaining(&waiter.id).await; + self.release_download_permit_for_seed(&waiter.id).await; + self.finish_seed_start(&waiter.id); + self.abandon_seed_start(&waiter.id); + self.add_seed_waiter(SeedWaiter { + id: waiter.id.clone(), + queue_id: waiter.queue_id.clone(), + lifecycle_generation: waiter.lifecycle_generation, + }); + self.emit_waiting_to_seed(&waiter.id, remaining); + return; + } + Err(error) => { + log::warn!( + "Torrent seed resume [{}] remains unverified; retaining seed ownership and permit: {}", + waiter.id, + error + ); + delay = (delay * 2).min(Duration::from_secs(5)); + } } } } @@ -2333,7 +2438,7 @@ impl QueueManager { .as_deref() .ok_or_else(|| "Torrent metadata is unavailable for web-seed management".to_string())?; let path = crate::torrent::validate_managed_torrent_path(&self.app_handle, id, path)?; - let bytes = tokio::fs::read(path) + let bytes = crate::torrent::read_bounded_torrent_bytes(&path) .await .map_err(|error| format!("could not read cached Torrent metadata: {error}"))?; crate::torrent::parse_torrent_bytes(&bytes) @@ -3178,7 +3283,7 @@ impl QueueManager { id, torrent_path, )?; - let bytes = tokio::fs::read(&torrent_path) + let bytes = crate::torrent::read_bounded_torrent_bytes(&torrent_path) .await .map_err(|_| "live Torrent file progress is unavailable".to_string())?; let metadata = crate::torrent::parse_torrent_bytes(&bytes)?; @@ -7815,7 +7920,7 @@ impl SidecarSpawner for ProductionSpawner { id, path, )?; - let bytes = tokio::fs::read(&path) + let bytes = crate::torrent::read_bounded_torrent_bytes(&path) .await .map_err(|error| format!("could not read cached torrent metadata: {error}"))?; let (sanitized_bytes, embedded_web_seeds) = @@ -8902,6 +9007,42 @@ mod tests { } } + struct AmbiguousResumeSeedSpawner { + resume_calls: Arc, + } + + #[async_trait::async_trait] + impl SidecarSpawner for AmbiguousResumeSeedSpawner { + async fn add_uri(&self, id: &str, _payload: &SpawnPayload) -> Result { + Ok(format!("gid-{id}")) + } + + async fn remove_uri(&self, _gid: &str) -> Result<(), String> { + Ok(()) + } + + async fn pause_for_seed(&self, _gid: &str) -> Result { + Ok(Aria2SeedControlOutcome::Paused) + } + + async fn resume_for_seed(&self, _gid: &str) -> Result { + if self.resume_calls.fetch_add(1, Ordering::SeqCst) == 0 { + Err("seed resume status verification unavailable".to_string()) + } else { + Ok(Aria2SeedControlOutcome::Resumed) + } + } + + async fn run_media( + &self, + _id: &str, + _payload: &SpawnPayload, + _lifecycle_generation: u64, + ) -> Result<(), String> { + Ok(()) + } + } + #[test] fn aria2_connection_options_enable_requested_ranges_for_small_release_assets() { let mut options = serde_json::Map::new(); @@ -10871,6 +11012,82 @@ mod tests { assert_eq!(manager.current_aria2_control_epoch("second").await, 1); } + #[tokio::test] + async fn ambiguous_seed_resume_retries_without_stranding_the_seed_permit() { + let app = tauri::test::mock_builder() + .build(tauri::test::mock_context(tauri::test::noop_assets())) + .expect("mock app"); + let spawner = Arc::new(AmbiguousResumeSeedSpawner { + resume_calls: Arc::new(AtomicUsize::new(0)), + }); + let manager = Arc::new(QueueManager::test_new( + app.handle().clone(), + 1, + spawner.clone(), + )); + manager.configure_seed_capacity(true, 1); + for id in ["first", "second"] { + manager.reserve_enqueue_generation(id, 0).await.unwrap(); + assert!(manager.ensure_aria2_permit(id).await); + manager + .aria2_payloads + .lock() + .await + .insert( + id.to_string(), + SpawnPayload { + is_torrent: true, + torrent_seed_time: Some(5.0), + ..Default::default() + }, + ); + manager + .remember_gid(id.to_string(), format!("gid-{id}")) + .await; + manager + .apply_completion(id, PendingOutcome::Seeding) + .await; + } + + assert!(manager.is_waiting_to_seed("second")); + manager.release_registered_id("first").await; + assert!(manager.try_start_waiting_seed().await); + + let first_attempt = tokio::time::timeout(Duration::from_secs(1), async { + loop { + if spawner.resume_calls.load(Ordering::SeqCst) >= 1 + && manager.has_active_permit("second").await + && manager.is_waiting_to_seed("second") + { + break; + } + tokio::task::yield_now().await; + } + }) + .await; + assert!(first_attempt.is_ok(), "ambiguous seed resume was not retained safely"); + + manager.wake_seed_waiters(); + let recovered = tokio::time::timeout(Duration::from_secs(1), async { + loop { + if spawner.resume_calls.load(Ordering::SeqCst) >= 2 + && manager.is_seed_owner("second") + && !manager.is_waiting_to_seed("second") + && !manager.has_active_permit("second").await + { + break; + } + tokio::task::yield_now().await; + } + }) + .await; + assert!( + recovered.is_ok(), + "ambiguous seed resume did not reconcile after the daemon became reachable" + ); + assert_eq!(manager.current_aria2_control_epoch("second").await, 1); + } + #[tokio::test] async fn unexpected_seeding_outcome_without_policy_is_reconciled_as_complete() { let app = tauri::test::mock_builder() diff --git a/src-tauri/src/torrent.rs b/src-tauri/src/torrent.rs index d711bd5..e6f3c4b 100644 --- a/src-tauri/src/torrent.rs +++ b/src-tauri/src/torrent.rs @@ -1,5 +1,6 @@ use sha1::{Digest, Sha1}; use std::collections::{BTreeMap, HashSet}; +use std::io::Read; use std::path::{Path, PathBuf}; use tauri::Manager; @@ -788,7 +789,8 @@ pub fn inspect_source(source: &str) -> Result { return magnet_metadata(source.trim()); } let path = local_torrent_path(source)?; - let bytes = std::fs::read(&path).map_err(|error| format!("could not read torrent file: {error}"))?; + let bytes = read_bounded_torrent_bytes_sync(&path) + .map_err(|error| format!("could not read torrent file: {error}"))?; parse_torrent_bytes(&bytes) } @@ -1148,7 +1150,21 @@ fn is_canonical_torrent_temp_file(name: &str) -> bool { && temporary_id.bytes().all(|byte| byte.is_ascii_hexdigit()) } -async fn read_bounded_torrent_bytes(path: &Path) -> std::io::Result> { +pub(crate) fn read_bounded_torrent_bytes_sync(path: &Path) -> std::io::Result> { + let file = std::fs::File::open(path)?; + let mut bytes = Vec::with_capacity(std::cmp::min(MAX_TORRENT_BYTES, 64 * 1024)); + file.take((MAX_TORRENT_BYTES + 1) as u64) + .read_to_end(&mut bytes)?; + if bytes.len() > MAX_TORRENT_BYTES { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("torrent metadata exceeds {MAX_TORRENT_BYTES} bytes"), + )); + } + Ok(bytes) +} + +pub(crate) async fn read_bounded_torrent_bytes(path: &Path) -> std::io::Result> { let file = tokio::fs::File::open(path).await?; let mut bytes = Vec::with_capacity(std::cmp::min(MAX_TORRENT_BYTES, 64 * 1024)); file.take((MAX_TORRENT_BYTES + 1) as u64) @@ -1687,6 +1703,19 @@ mod tests { )); } + #[test] + fn bounded_local_torrent_reads_reject_oversized_files() { + let temporary = tempfile::tempdir().expect("temporary torrent directory should exist"); + let path = temporary.path().join("oversized.torrent"); + let file = std::fs::File::create(&path).expect("oversized torrent fixture should exist"); + file.set_len((MAX_TORRENT_BYTES + 1) as u64) + .expect("oversized torrent fixture should be sparse-writable"); + + let error = read_bounded_torrent_bytes_sync(&path) + .expect_err("oversized torrent metadata must be rejected before parsing"); + assert_eq!(error.kind(), std::io::ErrorKind::InvalidData); + } + #[test] fn rejects_noncanonical_lengths_and_invalid_files_field() { assert!(parse_torrent_bytes(b"d4:infod6:lengthi5e4:name04:testee").is_err());