diff --git a/src-tauri/src/ipc.rs b/src-tauri/src/ipc.rs index a1f6d4c..756f832 100644 --- a/src-tauri/src/ipc.rs +++ b/src-tauri/src/ipc.rs @@ -796,6 +796,31 @@ pub enum QueueDirection { Down, } +#[derive(Clone, Debug, Serialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export, export_to = "../../src/bindings/")] +pub struct DownloadStateProgress { + pub fraction: f64, + #[serde(skip_serializing_if = "Option::is_none")] + #[ts(optional)] + pub downloaded_bytes: Option, + #[serde(skip_serializing_if = "Option::is_none")] + #[ts(optional)] + pub total_bytes: Option, + #[serde(skip_serializing_if = "Option::is_none")] + #[ts(optional)] + pub total_is_estimate: Option, +} + +#[derive(Clone, Debug, Serialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export, export_to = "../../src/bindings/")] +pub struct DownloadAllocationEvent { + pub id: String, + pub pending: bool, + pub lifecycle_generation: String, +} + #[derive(Clone, Debug, Serialize, TS)] #[serde(rename_all = "camelCase")] #[ts(export, export_to = "../../src/bindings/")] @@ -816,6 +841,9 @@ pub struct DownloadStateEvent { pub destination: Option, #[ts(optional)] pub torrent_seed_remaining: Option, + #[serde(skip_serializing_if = "Option::is_none")] + #[ts(optional)] + pub progress: Option, } impl DownloadStateEvent { @@ -829,6 +857,7 @@ impl DownloadStateEvent { file_name: None, destination: None, torrent_seed_remaining: None, + progress: None, } } @@ -843,6 +872,7 @@ impl DownloadStateEvent { file_name: None, destination: None, torrent_seed_remaining: None, + progress: None, } } @@ -857,6 +887,7 @@ impl DownloadStateEvent { file_name: None, destination: None, torrent_seed_remaining: None, + progress: None, } } @@ -870,6 +901,7 @@ impl DownloadStateEvent { file_name: None, destination: None, torrent_seed_remaining: remaining, + progress: None, } } @@ -883,6 +915,7 @@ impl DownloadStateEvent { file_name: Some(file_name.into()), destination: None, torrent_seed_remaining: None, + progress: None, } } @@ -899,6 +932,7 @@ impl DownloadStateEvent { file_name: None, destination: None, torrent_seed_remaining: None, + progress: None, } } @@ -912,6 +946,7 @@ impl DownloadStateEvent { file_name: None, destination: None, torrent_seed_remaining: remaining, + progress: None, } } @@ -929,6 +964,11 @@ impl DownloadStateEvent { self } + pub fn with_progress(mut self, progress: DownloadStateProgress) -> Self { + self.progress = Some(progress); + self + } + fn safe_error(error: impl Into) -> (String, Option) { let error = crate::redact_sensitive_text(&error.into()); let error_kind = crate::retry::is_aria2_name_resolution_error(&error) diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 2979b4d..fb6ca3c 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -5144,8 +5144,14 @@ async fn pause_download( let removed_pending = state.queue_manager.remove_from_pending(&id).await; let gid = state.queue_manager.aria2_gid_for_download(&id); + if gid.is_none() { + // A queued or not-yet-registered transfer cannot be verified through + // Aria2. Removing it from pending is the terminal pause boundary for + // this lifecycle, so its native allocation marker can be cleared now. + state.queue_manager.clear_aria2_allocation(&id).await; + } if let Some(gid) = gid.as_deref() { - let status = aria2_download_status( + let (status, mut status_progress) = aria2_download_status_snapshot( state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), &state.aria2_secret, gid, @@ -5153,6 +5159,7 @@ async fn pause_download( .await?; match status.as_str() { "paused" => { + state.queue_manager.clear_aria2_allocation(&id).await; state.queue_manager.next_aria2_control_epoch(&id).await; state.queue_manager.cancel_aria2_retries(&id).await; log::info!("aria2 pause [{}]: gid {} was already paused", id, gid); @@ -5171,14 +5178,16 @@ async fn pause_download( Err(error) => Err(format!("failed to pause aria2 gid {gid}: {error}")), }; if let Err(pause_error) = pause_result { - match aria2_download_status( + match aria2_download_status_snapshot( state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), &state.aria2_secret, gid, ) .await { - Ok(status) if status == "paused" => { + Ok((status, progress)) if status == "paused" => { + state.queue_manager.clear_aria2_allocation(&id).await; + status_progress = progress.or(status_progress); // forcePause may have returned an RPC error after // the daemon actually paused the GID. Invalidate // terminal events already in flight before @@ -5191,27 +5200,34 @@ async fn pause_download( gid ); } - Ok(status) if status == "complete" => { + Ok((status, progress)) if status == "complete" => { + state.queue_manager.clear_aria2_allocation(&id).await; state .queue_manager - .apply_completion_locked(&id, crate::queue::PendingOutcome::Complete) + .apply_completion_locked_with_progress( + &id, + crate::queue::PendingOutcome::Complete, + progress.or(status_progress.clone()), + ) .await; return Ok(()); } - Ok(status) if matches!(status.as_str(), "error" | "removed") => { + Ok((status, progress)) if matches!(status.as_str(), "error" | "removed") => { + state.queue_manager.clear_aria2_allocation(&id).await; let terminal_error = format!( "cannot pause aria2 gid {gid}: {pause_error}; daemon reports terminal state {status}" ); state .queue_manager - .apply_completion_locked( + .apply_completion_locked_with_progress( &id, crate::queue::PendingOutcome::Error(terminal_error.clone()), + progress.or(status_progress.clone()), ) .await; return Err(terminal_error); } - Ok(status) => { + Ok((status, _)) => { state.queue_manager.allow_aria2_retries(&id).await; return Err(format!( "{pause_error}; aria2 gid {gid} is still {status}" @@ -5225,10 +5241,12 @@ async fn pause_download( } } } + state.queue_manager.clear_aria2_allocation(&id).await; state.queue_manager.next_aria2_control_epoch(&id).await; log::info!("aria2 pause [{}]: gid {} paused", id, gid); } "complete" => { + state.queue_manager.clear_aria2_allocation(&id).await; // Aria2 can reach complete before its terminal event updates // Firelink's row. Treat a pause request in that narrow window // as an idempotent completion reconciliation, not as an @@ -5240,11 +5258,16 @@ async fn pause_download( ); state .queue_manager - .apply_completion_locked(&id, crate::queue::PendingOutcome::Complete) + .apply_completion_locked_with_progress( + &id, + crate::queue::PendingOutcome::Complete, + status_progress.clone(), + ) .await; return Ok(()); } terminal => { + state.queue_manager.clear_aria2_allocation(&id).await; let retrying = state.queue_manager.has_aria2_retry_state(&id).await; state.queue_manager.clear_aria2_retry_state(&id).await; state.queue_manager.forget_aria2_gid(&id).await; @@ -5259,13 +5282,14 @@ async fn pause_download( // to repair it. state.queue_manager.release_registered_id(&id).await; use tauri::Emitter; - let _ = app_handle.emit( - "download-state", - crate::ipc::DownloadStateEvent::new( - id, - crate::ipc::DownloadStatus::Paused, - ), + let mut event = crate::ipc::DownloadStateEvent::new( + id, + crate::ipc::DownloadStatus::Paused, ); + if let Some(progress) = status_progress { + event = event.with_progress(progress); + } + let _ = app_handle.emit("download-state", event); return Ok(()); } state.queue_manager.release_registered_id(&id).await; @@ -5282,11 +5306,21 @@ async fn pause_download( }; state.queue_manager.release_seed_tracking(&id); state.queue_manager.release_permit(&id).await; + let paused_progress = aria2_download_status_snapshot( + state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), + &state.aria2_secret, + gid, + ) + .await + .ok() + .and_then(|(_, progress)| progress) + .or(status_progress); use tauri::Emitter; - let _ = app_handle.emit( - "download-state", - crate::ipc::DownloadStateEvent::paused_with_seed_remaining(id, seed_remaining), - ); + let mut event = crate::ipc::DownloadStateEvent::paused_with_seed_remaining(id, seed_remaining); + if let Some(progress) = paused_progress { + event = event.with_progress(progress); + } + let _ = app_handle.emit("download-state", event); return Ok(()); } @@ -5544,8 +5578,14 @@ async fn resume_download( Err(error) => Some(format!("failed to resume aria2 gid {gid_clone}: {error}")), }; if let Some(unpause_error) = unpause_error { - match verify_aria2_resume_status(aria2_port, &aria2_secret, &gid_clone).await { - Ok(status) if matches!(status.as_str(), "active" | "waiting") => { + match verify_aria2_resume_status_snapshot( + aria2_port, + &aria2_secret, + &gid_clone, + ) + .await + { + Ok((status, _)) if matches!(status.as_str(), "active" | "waiting") => { let still_current = queue_manager .is_aria2_control_epoch_current(&id_clone, control_epoch) .await @@ -5578,7 +5618,7 @@ async fn resume_download( ); return; } - Ok(status) if status == "complete" => { + Ok((status, progress)) if status == "complete" => { log::info!( "aria2 resume [{}]: {} but daemon reports gid {} complete; reconciling completion", id_clone, @@ -5586,14 +5626,15 @@ async fn resume_download( gid_clone ); queue_manager - .apply_completion_locked( + .apply_completion_locked_with_progress( &id_clone, crate::queue::PendingOutcome::Complete, + progress, ) .await; return; } - Ok(status) if status == "paused" => { + Ok((status, progress)) if status == "paused" => { queue_manager.next_aria2_control_epoch(&id_clone).await; queue_manager.cancel_aria2_retries(&id_clone).await; queue_manager.release_permit(&id_clone).await; @@ -5603,28 +5644,30 @@ async fn resume_download( unpause_error, gid_clone ); - let _ = app_handle_clone.emit( - "download-state", - crate::ipc::DownloadStateEvent::paused_with_error( - &id_clone, - unpause_error, - ), + let mut event = crate::ipc::DownloadStateEvent::paused_with_error( + &id_clone, + unpause_error, ); + if let Some(progress) = progress { + event = event.with_progress(progress); + } + let _ = app_handle_clone.emit("download-state", event); return; } - Ok(status) if matches!(status.as_str(), "error" | "removed") => { + Ok((status, progress)) if matches!(status.as_str(), "error" | "removed") => { let terminal_error = format!( "{unpause_error}; daemon reports gid {gid_clone} as {status}" ); queue_manager - .apply_completion_locked( + .apply_completion_locked_with_progress( &id_clone, crate::queue::PendingOutcome::Error(terminal_error), + progress, ) .await; return; } - Ok(status) => { + Ok((status, _)) => { // An unrecognized daemon state is not proof that // the transfer stopped. Keep its permit and // mapping so a later reconciliation can observe @@ -5654,7 +5697,7 @@ async fn resume_download( // aria2 may still report the GID as paused, complete, or // otherwise unavailable. Verify the daemon state before // publishing Downloading to the renderer. - let status_after_unpause = match verify_aria2_resume_status( + let (status_after_unpause, status_after_unpause_progress) = match verify_aria2_resume_status_snapshot( aria2_port, &aria2_secret, &gid_clone, @@ -5676,9 +5719,10 @@ async fn resume_download( "active" | "waiting" => {} "complete" => { queue_manager - .apply_completion_locked( + .apply_completion_locked_with_progress( &id_clone, crate::queue::PendingOutcome::Complete, + status_after_unpause_progress, ) .await; return; @@ -5688,9 +5732,10 @@ async fn resume_download( "aria2 resume left gid {gid_clone} in terminal state {status_after_unpause}" ); queue_manager - .apply_completion_locked( + .apply_completion_locked_with_progress( &id_clone, crate::queue::PendingOutcome::Error(terminal_error), + status_after_unpause_progress, ) .await; return; @@ -5706,10 +5751,14 @@ async fn resume_download( error, gid_clone ); - let _ = app_handle_clone.emit( - "download-state", - crate::ipc::DownloadStateEvent::paused_with_error(&id_clone, error), + let mut event = crate::ipc::DownloadStateEvent::paused_with_error( + &id_clone, + error, ); + if let Some(progress) = status_after_unpause_progress { + event = event.with_progress(progress); + } + let _ = app_handle_clone.emit("download-state", event); return; } other => { @@ -5973,6 +6022,10 @@ async fn remove_download( .queue_manager .release_permit_for_generation(&id, media_lifecycle_generation) .await; + state + .queue_manager + .clear_aria2_allocation_for_epoch(&id, removal_epoch) + .await; return Err("download lifecycle changed while waiting for aria2 dispatch".to_string()); } if let Some(late_gid) = state.queue_manager.aria2_gid_for_download(&id) { @@ -6395,17 +6448,78 @@ async fn aria2_download_status(port: u16, secret: &str, gid: &str) -> Result Result<(String, Option), String> { + let result = rpc_call( + port, + secret, + "aria2.tellStatus", + serde_json::json!([gid, ["status", "completedLength", "totalLength"]]), + ) + .await + .map_err(|error| format!("failed to query aria2 gid {gid}: {error}"))?; + let status = result + .get("status") + .and_then(serde_json::Value::as_str) + .map(str::to_string) + .ok_or_else(|| format!("aria2.tellStatus returned no status for gid {gid}"))?; + let progress = aria2_download_state_progress(Some(&result)); + Ok((status, progress)) +} + +fn aria2_u64(value: Option<&serde_json::Value>) -> Option { + value.and_then(|value| { + value + .as_str() + .and_then(|value| value.parse::().ok()) + .or_else(|| value.as_u64()) + }) +} + +fn aria2_download_state_progress( + status: Option<&serde_json::Value>, +) -> Option { + let status = status?; + let downloaded = aria2_u64(status.get("completedLength")); + let total = aria2_u64(status.get("totalLength")); + if downloaded.is_none() && total.is_none() { + return None; + } + let is_complete = status.get("status").and_then(|value| value.as_str()) == Some("complete"); + let fraction = if is_complete { + 1.0 + } else if let Some(total) = total.filter(|total| *total > 0) { + (downloaded.unwrap_or_default().min(total) as f64 / total as f64).clamp(0.0, 1.0) + } else { + 0.0 + }; + let total_bytes = total.filter(|value| *value > 0).map(|value| value as f64); + Some(crate::ipc::DownloadStateProgress { + fraction, + downloaded_bytes: downloaded.map(|value| value as f64), + total_bytes, + total_is_estimate: total_bytes.map(|_| false), + }) +} + /// Verify a retained-GID resume against Aria2's actual state. `unpause` can /// return before a paused GID becomes observable as active, and a transient /// tellStatus failure must not turn a successful resume into a false failure. /// Retry only that ambiguous observation; terminal and active states return /// immediately and remain authoritative. -async fn verify_aria2_resume_status(port: u16, secret: &str, gid: &str) -> Result { +async fn verify_aria2_resume_status_snapshot( + port: u16, + secret: &str, + gid: &str, +) -> Result<(String, Option), String> { let mut last_observation = Err(format!("aria2 resume status for gid {gid} was not observed")); for attempt in 0..4u32 { - last_observation = match aria2_download_status(port, secret, gid).await { - Ok(status) if status != "paused" => return Ok(status), - Ok(status) => Ok(status), + last_observation = match aria2_download_status_snapshot(port, secret, gid).await { + Ok((status, progress)) if status != "paused" => return Ok((status, progress)), + Ok((status, progress)) => Ok((status, progress)), Err(error) => Err(error), }; if attempt < 3 { @@ -6426,7 +6540,7 @@ async fn reconcile_aria2_downloads(app_handle: &tauri::AppHandle) { port, &secret, "aria2.tellStatus", - serde_json::json!([gid, ["status", "seeder", "errorCode", "errorMessage"]]), + serde_json::json!([gid, ["status", "seeder", "errorCode", "errorMessage", "completedLength", "totalLength"]]), ) .await { @@ -6502,7 +6616,11 @@ async fn reconcile_aria2_downloads(app_handle: &tauri::AppHandle) { }; if let Some(outcome) = outcome { - state.queue_manager.handle_aria2_event(&gid, outcome).await; + let progress = aria2_download_state_progress(Some(&status)); + state + .queue_manager + .handle_aria2_event_with_progress(&gid, outcome, progress) + .await; } } } @@ -7605,6 +7723,10 @@ async fn cancel_enqueue_generation( .queue_manager .cancel_enqueue_generation(&id, generation) .await; + state + .queue_manager + .clear_aria2_allocation_for_lifecycle_generation(&id, generation) + .await; Ok(()) } @@ -7643,6 +7765,19 @@ async fn enqueue_many( }); continue; } + if let Err(error) = preflight_download_destination_access( + &app_handle, + &id, + &item.destination, + ) { + results.push(crate::ipc::EnqueueResult { + id, + success: false, + filename: None, + error: Some(error), + }); + continue; + } item.filename = crate::download_ownership::canonical_download_filename(&item.filename); let filename = item.filename.clone(); let lifecycle_generation = match enqueue_lifecycle_generation(&item) { @@ -11607,6 +11742,7 @@ mod tests { normalize_media_connections, validate_enqueue_url, validate_enqueue_uris, validate_keychain_grant_request_id, aria2_gid_not_found, + aria2_download_state_progress, preflight_download_destination_access, retained_torrent_id_from_persisted_record, retained_torrent_info_hash_from_persisted_record, merge_durable_torrent_telemetry, torrent_identity_magnet, torrent_move_path_pair, @@ -11635,6 +11771,36 @@ mod tests { assert!(!aria2_gid_not_found("aria2 error code 3: Resource not found")); } + #[test] + fn terminal_aria2_status_preserves_exact_progress_snapshot() { + let snapshot = aria2_download_state_progress(Some(&json!({ + "status": "error", + "completedLength": "420", + "totalLength": "1000" + }))) + .expect("length fields should produce a progress snapshot"); + + assert!((snapshot.fraction - 0.42).abs() < f64::EPSILON); + assert_eq!(snapshot.downloaded_bytes, Some(420.0)); + assert_eq!(snapshot.total_bytes, Some(1000.0)); + assert_eq!(snapshot.total_is_estimate, Some(false)); + } + + #[test] + fn destination_preflight_returns_the_retryable_marker_before_admission() { + let app = tauri::test::mock_builder() + .build(tauri::test::mock_context(tauri::test::noop_assets())) + .expect("mock app"); + let error = preflight_download_destination_access( + app.handle(), + "preflight-only", + "relative/not-approved", + ) + .expect_err("an unapproved destination must not enter admission"); + + assert!(error.starts_with("destination access retryable:")); + } + #[test] fn renderer_download_snapshots_cannot_lower_native_torrent_totals() { let existing = vec![ @@ -15699,42 +15865,95 @@ pub fn run() { if let Some(event) = params.first().and_then(|p| p.as_object()) { if let Some(gid) = event.get("gid").and_then(|g| g.as_str()) { let state = app_handle_bg.state::(); + let mut progress = None; let outcome = match method { - "aria2.onDownloadComplete" => Some(crate::queue::PendingOutcome::Complete), - "aria2.onBtDownloadComplete" => Some(crate::queue::PendingOutcome::Seeding), + "aria2.onDownloadStart" => { + let downloaded_bytes = aria2_download_status_snapshot( + state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), + &state.aria2_secret, + gid, + ) + .await + .ok() + .and_then(|(_, progress)| progress) + .and_then(|progress| progress.downloaded_bytes) + .filter(|value| value.is_finite() && *value > 0.0) + .map(|value| value as u64) + .unwrap_or_default(); + state + .queue_manager + .handle_aria2_download_start(gid, downloaded_bytes) + .await; + None + } + "aria2.onDownloadComplete" => { + progress = aria2_download_state_progress( + rpc_call( + state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), + &state.aria2_secret, + "aria2.tellStatus", + serde_json::json!([gid, ["status", "completedLength", "totalLength"]]), + ) + .await + .ok() + .as_ref(), + ); + Some(crate::queue::PendingOutcome::Complete) + } + "aria2.onBtDownloadComplete" => { + progress = aria2_download_state_progress( + rpc_call( + state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), + &state.aria2_secret, + "aria2.tellStatus", + serde_json::json!([gid, ["status", "completedLength", "totalLength"]]), + ) + .await + .ok() + .as_ref(), + ); + Some(crate::queue::PendingOutcome::Seeding) + } "aria2.onDownloadError" => { let mut msg = event.get("error_message").and_then(|m| m.as_str()).unwrap_or("aria2 download error").to_string(); - let aria2_port = state.aria2_port.load(std::sync::atomic::Ordering::Relaxed); - let aria2_secret = state.aria2_secret.clone(); - if let Ok(status) = rpc_call(aria2_port, &aria2_secret, "aria2.tellStatus", serde_json::json!([gid, ["errorCode", "errorMessage"]])).await { - let err_msg = status - .get("errorMessage") - .and_then(|m| m.as_str()) - .filter(|m| !m.is_empty()); - let err_code = status - .get("errorCode") - .and_then(|m| m.as_str()) - .filter(|m| !m.is_empty()); - match (err_code, err_msg) { - (Some(code), Some(message)) => { - msg = format!("aria2 error code {code}: {message}"); + let status = rpc_call( + state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), + &state.aria2_secret, + "aria2.tellStatus", + serde_json::json!([gid, ["errorCode", "errorMessage", "completedLength", "totalLength", "status"]]), + ) + .await + .ok(); + if let Some(status) = status.as_ref() { + progress = aria2_download_state_progress(Some(status)); + let err_msg = status + .get("errorMessage") + .and_then(|m| m.as_str()) + .filter(|m| !m.is_empty()); + let err_code = status + .get("errorCode") + .and_then(|m| m.as_str()) + .filter(|m| !m.is_empty()); + match (err_code, err_msg) { + (Some(code), Some(message)) => { + msg = format!("aria2 error code {code}: {message}"); + } + (Some(code), None) => { + msg = format!("aria2 error code {code}: {msg}"); + } + (None, Some(message)) => { + msg = message.to_string(); + } + (None, None) => {} + } } - (Some(code), None) => { - msg = format!("aria2 error code {code}: {msg}"); - } - (None, Some(message)) => { - msg = message.to_string(); - } - (None, None) => {} - } - } Some(crate::queue::PendingOutcome::Error(msg)) } _ => None, }; if let Some(outcome) = outcome { Arc::clone(&state.queue_manager) - .handle_aria2_event(gid, outcome) + .handle_aria2_event_with_progress(gid, outcome, progress) .await; } } @@ -16129,6 +16348,22 @@ pub fn run() { None }; + // tellActive is the poller's confirmed native + // progress path. If the WebSocket start + // notification was lost, this is sufficient + // to end the allocation phase for the same + // mapped lifecycle. + poll_mgr + .complete_aria2_allocation_for_gid(gid, completed) + .await; + if !poll_mgr.is_current_aria2_gid_mapping(gid, &mapping) + || !poll_mgr + .is_aria2_control_epoch_current(&id, control_epoch) + .await + { + continue; + } + use tauri::Emitter; let _ = app_handle_poll.emit("download-progress", DownloadProgressEvent { id: id.clone(), @@ -16206,7 +16441,7 @@ pub fn run() { poll_port.load(std::sync::atomic::Ordering::Relaxed), &poll_secret, "aria2.tellStatus", - serde_json::json!([gid, ["status", "seeder", "errorCode", "errorMessage"]]), + serde_json::json!([gid, ["status", "seeder", "errorCode", "errorMessage", "completedLength", "totalLength"]]), ) .await { @@ -16389,6 +16624,7 @@ pub fn run() { _ => None, }; if let Some(outcome) = outcome { + let progress = aria2_download_state_progress(Some(&status)); let terminal_error = match &outcome { crate::queue::PendingOutcome::Error(error) => Some(error.as_str()), _ => None, @@ -16409,7 +16645,9 @@ pub fn run() { .and_then(crate::retry::aria2_error_code) .unwrap_or_else(|| "none".to_string()) ); - poll_mgr.handle_aria2_event(&gid, outcome).await; + poll_mgr + .handle_aria2_event_with_progress(&gid, outcome, progress) + .await; } } observations.retain(|id, _| seen_ids.contains(id)); diff --git a/src-tauri/src/queue.rs b/src-tauri/src/queue.rs index 5158f97..82616ac 100644 --- a/src-tauri/src/queue.rs +++ b/src-tauri/src/queue.rs @@ -1,5 +1,8 @@ use base64::Engine as _; -use crate::ipc::{DownloadStateEvent, DownloadStatus, QueueDirection}; +use crate::ipc::{ + DownloadAllocationEvent, DownloadStateEvent, DownloadStateProgress, DownloadStatus, + QueueDirection, +}; use crate::power::PowerManager; use crate::retry::{ aria2_error_code, backoff_and_emit, is_aria2_name_resolution_error, @@ -21,6 +24,7 @@ use ts_rs::TS; /// Default capacity when no setting is read yet. pub const DEFAULT_MAX_CONCURRENT: usize = 3; pub const MAX_QUEUE_CONCURRENT: usize = 12; +const MAX_PENDING_DOWNLOAD_STARTS: usize = 1024; pub const MEDIA_RUN_CANCELLED: &str = "__firelink_media_run_cancelled__"; pub const DOWNLOAD_CONNECTIONS_MIN: i32 = 1; pub const DOWNLOAD_CONNECTIONS_MAX: i32 = 16; @@ -868,6 +872,21 @@ pub enum PendingOutcome { Error(String), } +#[derive(Debug, Clone)] +pub struct PendingAria2Outcome { + pub outcome: PendingOutcome, + pub progress: Option, +} + +impl PendingAria2Outcome { + pub fn new(outcome: PendingOutcome) -> Self { + Self { + outcome, + progress: None, + } + } +} + /// Result of recycling an aria2 transfer's connections. A refresh can race /// with daemon completion or leave the transfer paused after an ambiguous /// unpause failure, so callers must handle the verified daemon outcome. @@ -1176,7 +1195,14 @@ pub struct QueueManager { /// gid -> buffered (id_placeholder, outcome) for completions that arrived /// before the gid was stored. Drained by `remember_gid`. - pub pending_completion: Arc>>, + pub pending_completion: Arc>>, + /// Aria2 can emit onDownloadStart before addUri's response has been + /// mapped to the Firelink download. Buffer that start marker until the + /// current GID mapping is installed. + pending_download_starts: Arc>>, + /// Current Aria2 lifecycles whose files are expected to be preallocated. + /// The generation fences late start/clear events from an older GID. + aria2_allocation_pending: Mutex>, /// download id -> spawn payload for aria2 transient-error re-addUri retries. aria2_payloads: Mutex>, @@ -1281,6 +1307,8 @@ impl QueueManager { torrent_move_cancellations: StdMutex::new(HashSet::new()), aria2_gids: Arc::new(std::sync::RwLock::new(HashMap::new())), pending_completion: Arc::new(Mutex::new(HashMap::new())), + pending_download_starts: Arc::new(Mutex::new(HashSet::new())), + aria2_allocation_pending: Mutex::new(HashMap::new()), aria2_payloads: Mutex::new(HashMap::new()), aria2_connection_options: Mutex::new(HashMap::new()), aria2_dispatch_inflight: Mutex::new(HashMap::new()), @@ -3808,6 +3836,11 @@ impl QueueManager { .await; } } + // Unknown start notifications belong to the current Aria2 daemon + // session. A reconnect/restart invalidates that association; the + // poller will provide a fresh positive-progress fallback after the + // next GID mapping instead of letting an old GID clear a new phase. + self.pending_download_starts.lock().await.clear(); } /// Number of un-acquired permits currently in the semaphore pool. @@ -3822,6 +3855,159 @@ impl QueueManager { .emit("download-state", DownloadStateEvent::new(id, status)); } + fn emit_allocation_event(&self, id: &str, pending: bool, lifecycle_generation: u64) { + use tauri::Emitter; + let _ = self.app_handle.emit( + "download-allocation", + DownloadAllocationEvent { + id: id.to_string(), + pending, + lifecycle_generation: lifecycle_generation.to_string(), + }, + ); + } + + pub fn aria2_allocation_phase_eligible(payload: &SpawnPayload) -> bool { + if payload.is_media || payload.torrent_verify_only { + return false; + } + if !payload.is_torrent { + return true; + } + normalize_torrent_file_allocation(payload.torrent_file_allocation.as_deref()) + .is_ok_and(|allocation| allocation != "none") + } + + async fn begin_aria2_allocation( + &self, + id: &str, + control_epoch: u64, + lifecycle_generation: u64, + payload: &SpawnPayload, + ) { + if !Self::aria2_allocation_phase_eligible(payload) { + return; + } + self.aria2_allocation_pending + .lock() + .await + .insert(id.to_string(), (control_epoch, lifecycle_generation)); + self.emit_allocation_event(id, true, lifecycle_generation); + } + + pub async fn clear_aria2_allocation_for_lifecycle_generation( + &self, + id: &str, + lifecycle_generation: u64, + ) { + let cleared_generation = { + let mut pending = self.aria2_allocation_pending.lock().await; + let Some((_, pending_generation)) = pending.get(id).copied() else { + return; + }; + if pending_generation != lifecycle_generation { + return; + } + pending.remove(id).map(|(_, generation)| generation) + }; + if let Some(cleared_generation) = cleared_generation { + self.emit_allocation_event(id, false, cleared_generation); + } + } + + pub async fn clear_aria2_allocation_for_epoch(&self, id: &str, control_epoch: u64) { + let lifecycle_generation = { + let mut pending = self.aria2_allocation_pending.lock().await; + let Some((pending_epoch, lifecycle_generation)) = pending.get(id).copied() else { + return; + }; + if pending_epoch != control_epoch { + return; + } + pending.remove(id); + lifecycle_generation + }; + self.emit_allocation_event(id, false, lifecycle_generation); + } + + pub async fn clear_aria2_allocation(&self, id: &str) { + let lifecycle_generation = self + .aria2_allocation_pending + .lock() + .await + .remove(id) + .map(|(_, lifecycle_generation)| lifecycle_generation); + if let Some(lifecycle_generation) = lifecycle_generation { + self.emit_allocation_event(id, false, lifecycle_generation); + } + } + + pub async fn complete_aria2_allocation_for_gid(&self, gid: &str, downloaded_bytes: u64) { + // Aria2 reports an active GID with completedLength=0 while it is still + // creating preallocated files. That observation is not native + // transfer progress and must not hide the allocation phase. + if downloaded_bytes == 0 { + return; + } + let mapping = { + let _gid_state = self.aria2_gid_state.lock().await; + if self.is_aria2_gid_ignored_locked(gid).await { + return; + } + let Some(mapping) = self.aria2_gid_mapping(gid) else { + let mut starts = self.pending_download_starts.lock().await; + if starts.len() < MAX_PENDING_DOWNLOAD_STARTS { + starts.insert(gid.to_string()); + } + return; + }; + mapping + }; + if !self + .is_current_aria2_gid_mapping(gid, &mapping) + || !self + .is_aria2_control_epoch_current(&mapping.id, mapping.epoch) + .await + { + return; + } + self.clear_aria2_allocation_for_epoch(&mapping.id, mapping.epoch) + .await; + } + + pub async fn handle_aria2_download_start(&self, gid: &str, downloaded_bytes: u64) { + // Aria2 can publish onDownloadStart before it finishes creating + // preallocated files. A zero-byte start is therefore not native + // transfer progress and must leave the allocation phase visible. + if downloaded_bytes == 0 { + return; + } + let mapping = { + let _gid_state = self.aria2_gid_state.lock().await; + if self.is_aria2_gid_ignored_locked(gid).await { + return; + } + let Some(mapping) = self.aria2_gid_mapping(gid) else { + let mut starts = self.pending_download_starts.lock().await; + if starts.len() < MAX_PENDING_DOWNLOAD_STARTS { + starts.insert(gid.to_string()); + } + return; + }; + mapping + }; + if !self + .is_current_aria2_gid_mapping(gid, &mapping) + || !self + .is_aria2_control_epoch_current(&mapping.id, mapping.epoch) + .await + { + return; + } + self.clear_aria2_allocation_for_epoch(&mapping.id, mapping.epoch) + .await; + } + /// Resize the global concurrency limit. Grow adds permits immediately; /// shrink records a retirement debt honored lazily by the dispatcher. pub fn set_capacity(&self, new_target: usize) { @@ -3947,6 +4133,17 @@ impl QueueManager { }; if let Some(epoch) = aria2_lifecycle_epoch { self.begin_aria2_dispatch(&id, epoch).await; + // Register the native allocation phase before releasing the + // lifecycle lock. A pause/remove can otherwise invalidate this + // dispatch in the gap and leave a stale pending marker behind + // when the asynchronous addUri call starts. + self.begin_aria2_allocation( + &id, + epoch, + lifecycle_generation, + &task.payload, + ) + .await; } drop(control_guard); @@ -4008,6 +4205,8 @@ impl QueueManager { self.clear_aria2_retry_state(&id).await; self.release_permit(&id).await; } + self.clear_aria2_allocation_for_epoch(&id, lifecycle_epoch) + .await; return; } // A queued task is not a live transfer until aria2 has @@ -4074,10 +4273,12 @@ impl QueueManager { } } if let Some(outcome) = buffered_outcome { - self.handle_aria2_event(&gid, outcome).await; + self.handle_aria2_pending_event(&gid, outcome).await; } } Err(error) => { + self.clear_aria2_allocation_for_epoch(&id, lifecycle_epoch) + .await; let _control_guard = self.acquire_aria2_control(&id).await; let current_lifecycle = self .is_aria2_control_epoch_current(&id, lifecycle_epoch) @@ -4161,24 +4362,46 @@ impl QueueManager { } fn emit_failed(&self, id: &str, error: String) { - use tauri::Emitter; - let _ = self - .app_handle - .emit("download-state", DownloadStateEvent::failed(id, error)); + self.emit_failed_with_progress(id, error, None); } - fn emit_paused_with_error(&self, id: &str, error: String) { + fn emit_failed_with_progress( + &self, + id: &str, + error: String, + progress: Option, + ) { use tauri::Emitter; + let mut event = DownloadStateEvent::failed(id, error); + if let Some(progress) = progress { + event = event.with_progress(progress); + } + let _ = self + .app_handle + .emit("download-state", event); + } + + fn emit_paused_with_error_and_progress( + &self, + id: &str, + error: String, + progress: Option, + ) { + use tauri::Emitter; + let mut event = DownloadStateEvent::paused_with_error(id, error); + if let Some(progress) = progress { + event = event.with_progress(progress); + } let _ = self.app_handle.emit( "download-state", - DownloadStateEvent::paused_with_error(id, error), + event, ); } /// Store gid -> id and return any buffered terminal event for the caller /// to reconcile against the correct event path. In particular, buffered /// errors must still pass through transient retry classification. - pub async fn remember_gid(&self, id: String, gid: String) -> Option { + pub async fn remember_gid(&self, id: String, gid: String) -> Option { let epoch = self.current_aria2_control_epoch(&id).await; let buffered_outcome = { let _gid_state = self.aria2_gid_state.lock().await; @@ -4217,6 +4440,14 @@ impl QueueManager { } buffered.remove(&gid).map(|(_buf_id, outcome)| outcome) }; + let start_buffered = self + .pending_download_starts + .lock() + .await + .remove(&gid); + if start_buffered { + self.clear_aria2_allocation_for_epoch(&id, epoch).await; + } let retry_strike = self.aria2_retry_strike(&id).await; log::info!( "aria2 gid transition [stage=gid_transition id={} gid={} epoch={} retry_strike={} action=mapped]", @@ -4261,6 +4492,17 @@ impl QueueManager { /// and lets commands reconcile an Aria2 terminal status without releasing /// the lock first. pub(crate) async fn apply_completion_locked(&self, id: &str, outcome: PendingOutcome) { + self.apply_completion_locked_with_progress(id, outcome, None) + .await; + } + + pub(crate) async fn apply_completion_locked_with_progress( + &self, + id: &str, + outcome: PendingOutcome, + progress: Option, + ) { + self.clear_aria2_allocation(id).await; if matches!(&outcome, PendingOutcome::Complete) { self.capture_torrent_verification_evidence(id).await; } @@ -4385,7 +4627,12 @@ impl QueueManager { _ => DownloadStatus::Completed, } }; - self.emit_state(id, restored_status); + let mut event = DownloadStateEvent::new(id, restored_status); + if let Some(progress) = progress { + event = event.with_progress(progress); + } + use tauri::Emitter; + let _ = self.app_handle.emit("download-state", event); } PendingOutcome::Error(error) => { self.forget_torrent_telemetry(id).await; @@ -4452,7 +4699,7 @@ impl QueueManager { self.release_registered_id(id).await; self.release_permit(id).await; if verification_only { - self.emit_paused_with_error(id, error); + self.emit_paused_with_error_and_progress(id, error, progress); } else { let aria2_code = aria2_error_code(&error).unwrap_or_else(|| "none".to_string()); log::error!( @@ -4466,7 +4713,7 @@ impl QueueManager { network_error_class(&error), aria2_code ); - self.emit_failed(id, error); + self.emit_failed_with_progress(id, error, progress); } } PendingOutcome::Seeding => unreachable!("seeding outcomes are normalized before terminal cleanup"), @@ -4556,6 +4803,7 @@ impl QueueManager { async fn ignore_aria2_gid_locked(&self, gid: &str) { const MAX_IGNORED_GIDS: usize = 1024; + self.pending_download_starts.lock().await.remove(gid); let mut ignored = self.aria2_ignored_gids.lock().await; if !ignored.iter().any(|known| known == gid) { ignored.push_back(gid.to_string()); @@ -4788,7 +5036,20 @@ impl QueueManager { let payload = self.aria2_payloads.lock().await.get(id).cloned(); let recreation = if let Some(payload) = payload.as_ref() { - self.spawner.recreate_uri(id, gid, payload).await? + let lifecycle_generation = self + .registered_lifecycle_generation(id) + .await + .unwrap_or_default(); + self.begin_aria2_allocation(id, observed_epoch, lifecycle_generation, payload) + .await; + match self.spawner.recreate_uri(id, gid, payload).await { + Ok(outcome) => outcome, + Err(error) => { + self.clear_aria2_allocation_for_epoch(id, observed_epoch) + .await; + return Err(error); + } + } } else { // Older persisted rows may briefly reach recovery before their // payload has been rebuilt. Keep the current lifecycle intact and @@ -4798,6 +5059,8 @@ impl QueueManager { if let Aria2RecreateOutcome::NewGid(new_gid) = recreation { if new_gid.trim().is_empty() || new_gid == gid { + self.clear_aria2_allocation_for_epoch(id, observed_epoch) + .await; return Err(format!( "aria2 connection recovery returned an invalid replacement gid for {gid}" )); @@ -4808,6 +5071,8 @@ impl QueueManager { && self.is_aria2_control_epoch_current(id, observed_epoch).await && self.aria2_gid_for_download(id).as_deref() == Some(gid); if !still_current { + self.clear_aria2_allocation_for_epoch(id, observed_epoch) + .await; self.ignore_aria2_gid(&new_gid).await; drop(_control_guard); self.remove_stale_aria2_gid(id, &new_gid).await; @@ -4828,15 +5093,24 @@ impl QueueManager { ); drop(_control_guard); if let Some(outcome) = buffered_outcome { - self.handle_aria2_event(&new_gid, outcome).await; + self.handle_aria2_pending_event(&new_gid, outcome).await; } return Ok(()); } let outcome = match recreation { Aria2RecreateOutcome::Complete => Aria2RefreshOutcome::Complete, - Aria2RecreateOutcome::Refresh => self.spawner.refresh_uri(gid).await?, + Aria2RecreateOutcome::Refresh => match self.spawner.refresh_uri(gid).await { + Ok(outcome) => outcome, + Err(error) => { + self.clear_aria2_allocation_for_epoch(id, observed_epoch) + .await; + return Err(error); + } + }, Aria2RecreateOutcome::Unavailable(error) => { + self.clear_aria2_allocation_for_epoch(id, observed_epoch) + .await; let still_current = self.is_registered(id).await && self.has_active_permit(id).await && !self.is_aria2_retry_cancelled(id).await @@ -4907,6 +5181,7 @@ impl QueueManager { /// Remove every gid mapping for a download and discard buffered terminal /// events for those gids. Returns the most recently encountered gid. pub async fn forget_aria2_gid(&self, id: &str) -> Option { + self.clear_aria2_allocation(id).await; let _gid_state = self.aria2_gid_state.lock().await; let removed = { let mut gids = self.aria2_gids.write().unwrap(); @@ -4944,10 +5219,12 @@ impl QueueManager { self: &Arc, gid: String, error: String, + progress: Option, ) -> Pin + Send + 'static>> { let this = Arc::clone(self); Box::pin(async move { - this.handle_aria2_download_error_inner(&gid, error).await; + this.handle_aria2_download_error_inner(&gid, error, progress) + .await; }) } @@ -4957,8 +5234,8 @@ impl QueueManager { async fn map_or_buffer_aria2_event( &self, gid: &str, - outcome: PendingOutcome, - ) -> Option<(Aria2GidMapping, PendingOutcome)> { + outcome: PendingAria2Outcome, + ) -> Option<(Aria2GidMapping, PendingAria2Outcome)> { let _gid_state = self.aria2_gid_state.lock().await; if self.is_aria2_gid_ignored_locked(gid).await { return None; @@ -4977,13 +5254,31 @@ impl QueueManager { None } - async fn handle_aria2_download_error_inner(self: &Arc, gid: &str, error: String) { - let Some((mapping, PendingOutcome::Error(error))) = self - .map_or_buffer_aria2_event(gid, PendingOutcome::Error(error)) + async fn handle_aria2_download_error_inner( + self: &Arc, + gid: &str, + error: String, + progress: Option, + ) { + let Some((mapping, pending)) = self + .map_or_buffer_aria2_event( + gid, + PendingAria2Outcome { + outcome: PendingOutcome::Error(error), + progress, + }, + ) .await else { return; }; + let PendingAria2Outcome { + outcome: PendingOutcome::Error(error), + progress, + } = pending + else { + return; + }; let _control_guard = self.acquire_aria2_control(&mapping.id).await; let current_mapping = { @@ -5000,6 +5295,11 @@ impl QueueManager { return; } let id = mapping.id; + // The failed GID is no longer allocating. Keep the native phase + // marker scoped to an actual replacement addUri rather than showing + // "Allocating files" throughout retry backoff or a failed pause. + self.clear_aria2_allocation_for_epoch(&id, mapping.epoch) + .await; if self.aria2_retry_cancelled.lock().await.contains(&id) { log::info!( "aria2 retry cancellation [{}]: ignoring error for gid {} during removal", @@ -5028,7 +5328,11 @@ impl QueueManager { let payload = self.aria2_payloads.lock().await.get(&id).cloned(); if payload.is_none() { - self.apply_completion_locked(&id, PendingOutcome::Error(error)) + self.apply_completion_locked_with_progress( + &id, + PendingOutcome::Error(error), + progress, + ) .await; return; } @@ -5083,7 +5387,11 @@ impl QueueManager { // 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(&id, PendingOutcome::Error(error)) + self.apply_completion_locked_with_progress( + &id, + PendingOutcome::Error(error), + progress, + ) .await; return; } @@ -5133,6 +5441,7 @@ impl QueueManager { let this = Arc::clone(self); let id_for_task = id.clone(); let error_for_emit = error.clone(); + let progress_for_retry = progress.clone(); tauri::async_runtime::spawn(async move { let retry_cancel = async { loop { @@ -5191,6 +5500,17 @@ impl QueueManager { return; } + let lifecycle_generation = this + .registered_lifecycle_generation(&id_for_task) + .await + .unwrap_or_default(); + this.begin_aria2_allocation( + &id_for_task, + retry_epoch, + lifecycle_generation, + ¤t_payload, + ) + .await; match this .spawner .add_uri(&id_for_task, ¤t_payload) @@ -5205,6 +5525,8 @@ impl QueueManager { || this.aria2_gid_for_download(&id_for_task).as_deref() != Some(retry_gid.as_str()); if stale { + this.clear_aria2_allocation_for_epoch(&id_for_task, retry_epoch) + .await; drop(control_guard); if let Err(error) = this.spawner.remove_uri(&new_gid).await { log::error!( @@ -5249,18 +5571,22 @@ impl QueueManager { this.aria2_retrying_gids.lock().await.remove(&retry_gid); drop(control_guard); if let Some(outcome) = buffered_outcome { - this.handle_aria2_event(&new_gid_for_event, outcome).await; + this.handle_aria2_pending_event(&new_gid_for_event, outcome) + .await; } } Err(retry_error) => { + this.clear_aria2_allocation_for_epoch(&id_for_task, retry_epoch) + .await; let stale = this.is_aria2_retry_cancelled(&id_for_task).await || !this .is_aria2_control_epoch_current(&id_for_task, retry_epoch) .await; if !stale { - this.apply_completion_locked( + this.apply_completion_locked_with_progress( &id_for_task, PendingOutcome::Error(retry_error), + progress_for_retry, ) .await; } @@ -5275,14 +5601,44 @@ impl QueueManager { /// Entry point for the aria2 WS poller. Resolves gid -> id; if not yet /// stored, buffers the outcome for reconciliation by remember_gid. pub async fn handle_aria2_event(self: &Arc, gid: &str, outcome: PendingOutcome) { + self.handle_aria2_pending_event(gid, PendingAria2Outcome::new(outcome)) + .await; + } + + pub async fn handle_aria2_event_with_progress( + self: &Arc, + gid: &str, + outcome: PendingOutcome, + progress: Option, + ) { + self.handle_aria2_pending_event( + gid, + PendingAria2Outcome { outcome, progress }, + ) + .await; + } + + async fn handle_aria2_pending_event( + self: &Arc, + gid: &str, + pending: PendingAria2Outcome, + ) { + let PendingAria2Outcome { outcome, progress } = pending; if let PendingOutcome::Error(error) = outcome { - self.handle_aria2_download_error(gid.to_string(), error) + self.handle_aria2_download_error(gid.to_string(), error, progress) .await; return; } - let Some((mapping, outcome)) = self.map_or_buffer_aria2_event(gid, outcome).await else { + let Some((mapping, pending)) = self + .map_or_buffer_aria2_event( + gid, + PendingAria2Outcome { outcome, progress }, + ) + .await + else { return; }; + let PendingAria2Outcome { outcome, progress } = pending; let _control_guard = self.acquire_aria2_control(&mapping.id).await; if self.aria2_retrying_gids.lock().await.contains(gid) { @@ -5301,7 +5657,8 @@ impl QueueManager { { return; } - self.apply_completion_locked(&mapping.id, outcome).await; + self.apply_completion_locked_with_progress(&mapping.id, outcome, progress) + .await; } /// Reorder a pending task up or down. Returns the new pending order. @@ -8206,6 +8563,315 @@ mod tests { } } + struct BlockingSpawner { + started: Arc, + release: Arc, + } + + #[async_trait::async_trait] + impl SidecarSpawner for BlockingSpawner { + async fn add_uri(&self, _id: &str, _payload: &SpawnPayload) -> Result { + self.started.notify_one(); + self.release.notified().await; + Ok("blocking-gid".to_string()) + } + + async fn remove_uri(&self, _gid: &str) -> Result<(), String> { + Ok(()) + } + + async fn run_media( + &self, + _id: &str, + _payload: &SpawnPayload, + _lifecycle_generation: u64, + ) -> Result<(), String> { + Ok(()) + } + } + + fn allocation_pending_epoch( + manager: &QueueManager, + id: &str, + ) -> Option<(u64, u64)> { + manager + .aria2_allocation_pending + .try_lock() + .ok() + .and_then(|pending| pending.get(id).copied()) + } + + #[test] + fn aria2_allocation_eligibility_matches_download_type_and_torrent_policy() { + assert!(QueueManager::::aria2_allocation_phase_eligible( + &SpawnPayload::default() + )); + assert!(!QueueManager::::aria2_allocation_phase_eligible( + &SpawnPayload { + is_media: true, + ..SpawnPayload::default() + } + )); + assert!(QueueManager::::aria2_allocation_phase_eligible( + &SpawnPayload { + is_torrent: true, + torrent_file_allocation: Some("prealloc".to_string()), + ..SpawnPayload::default() + } + )); + assert!(!QueueManager::::aria2_allocation_phase_eligible( + &SpawnPayload { + is_torrent: true, + torrent_file_allocation: Some("none".to_string()), + ..SpawnPayload::default() + } + )); + assert!(!QueueManager::::aria2_allocation_phase_eligible( + &SpawnPayload { + is_torrent: true, + torrent_verify_only: true, + ..SpawnPayload::default() + } + )); + } + + #[tokio::test] + async fn allocation_stays_pending_while_async_add_uri_is_in_flight() { + let app = tauri::test::mock_builder() + .build(tauri::test::mock_context(tauri::test::noop_assets())) + .expect("mock app"); + let started = Arc::new(tokio::sync::Notify::new()); + let release = Arc::new(tokio::sync::Notify::new()); + let manager = Arc::new(QueueManager::test_new( + app.handle().clone(), + 1, + Arc::new(BlockingSpawner { + started: Arc::clone(&started), + release: Arc::clone(&release), + }), + )); + manager + .reserve_enqueue_generation("allocation", 7) + .await + .expect("lifecycle reservation"); + manager + .commit_reserved_enqueue( + QueuedTask { + id: "allocation".to_string(), + queue_id: "main".to_string(), + kind: TaskKind::Aria2, + payload: SpawnPayload::default(), + lifecycle_generation: 7, + }, + 7, + ) + .await + .expect("queued task"); + + let dispatcher = tokio::spawn(Arc::clone(&manager).run_dispatcher()); + tokio::time::timeout(Duration::from_secs(1), started.notified()) + .await + .expect("addUri should begin"); + assert_eq!(allocation_pending_epoch(&manager, "allocation"), Some((1, 7))); + + release.notify_one(); + tokio::time::timeout(Duration::from_secs(1), async { + loop { + if manager.aria2_gid_for_download("allocation").as_deref() == Some("blocking-gid") { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("async addUri should install its GID"); + manager.handle_aria2_download_start("blocking-gid", 1).await; + assert_eq!(allocation_pending_epoch(&manager, "allocation"), None); + dispatcher.abort(); + } + + #[tokio::test] + async fn download_start_before_gid_registration_is_buffered_and_consumed() { + let app = tauri::test::mock_builder() + .build(tauri::test::mock_context(tauri::test::noop_assets())) + .expect("mock app"); + let manager = QueueManager::test_new(app.handle().clone(), 1, Arc::new(TestSpawner)); + let epoch = manager.next_aria2_control_epoch("buffered-start").await; + manager + .begin_aria2_allocation( + "buffered-start", + epoch, + 9, + &SpawnPayload::default(), + ) + .await; + manager.handle_aria2_download_start("early-gid", 1).await; + assert!(manager + .pending_download_starts + .lock() + .await + .contains("early-gid")); + + manager + .remember_gid("buffered-start".to_string(), "early-gid".to_string()) + .await; + assert_eq!(allocation_pending_epoch(&manager, "buffered-start"), None); + assert!(!manager + .pending_download_starts + .lock() + .await + .contains("early-gid")); + } + + #[tokio::test] + async fn allocation_fallback_and_stale_epochs_cannot_clear_a_new_lifecycle() { + let app = tauri::test::mock_builder() + .build(tauri::test::mock_context(tauri::test::noop_assets())) + .expect("mock app"); + let manager = QueueManager::test_new(app.handle().clone(), 1, Arc::new(TestSpawner)); + let first_epoch = manager.next_aria2_control_epoch("stale-allocation").await; + manager + .begin_aria2_allocation( + "stale-allocation", + first_epoch, + 10, + &SpawnPayload::default(), + ) + .await; + let second_epoch = manager.next_aria2_control_epoch("stale-allocation").await; + manager + .begin_aria2_allocation( + "stale-allocation", + second_epoch, + 11, + &SpawnPayload::default(), + ) + .await; + manager + .clear_aria2_allocation_for_epoch("stale-allocation", first_epoch) + .await; + manager + .clear_aria2_allocation_for_lifecycle_generation("stale-allocation", 10) + .await; + assert_eq!( + allocation_pending_epoch(&manager, "stale-allocation"), + Some((second_epoch, 11)) + ); + + manager + .remember_gid("stale-allocation".to_string(), "fallback-gid".to_string()) + .await; + manager + .handle_aria2_download_start("fallback-gid", 0) + .await; + assert_eq!( + allocation_pending_epoch(&manager, "stale-allocation"), + Some((second_epoch, 11)) + ); + manager + .complete_aria2_allocation_for_gid("fallback-gid", 0) + .await; + assert_eq!( + allocation_pending_epoch(&manager, "stale-allocation"), + Some((second_epoch, 11)) + ); + manager + .complete_aria2_allocation_for_gid("fallback-gid", 1) + .await; + assert_eq!(allocation_pending_epoch(&manager, "stale-allocation"), None); + + manager + .begin_aria2_allocation( + "stale-allocation", + second_epoch, + 11, + &SpawnPayload::default(), + ) + .await; + manager.handle_aria2_download_start("fallback-gid", 1).await; + assert_eq!(allocation_pending_epoch(&manager, "stale-allocation"), None); + } + + #[tokio::test] + async fn ignored_gid_start_markers_cannot_clear_a_new_lifecycle() { + let app = tauri::test::mock_builder() + .build(tauri::test::mock_context(tauri::test::noop_assets())) + .expect("mock app"); + let manager = QueueManager::test_new(app.handle().clone(), 1, Arc::new(TestSpawner)); + let old_epoch = manager.next_aria2_control_epoch("ignored-start").await; + manager + .begin_aria2_allocation( + "ignored-start", + old_epoch, + 20, + &SpawnPayload::default(), + ) + .await; + manager + .remember_gid("ignored-start".to_string(), "reused-gid".to_string()) + .await; + manager.forget_aria2_gid("ignored-start").await; + + manager.handle_aria2_download_start("reused-gid", 1).await; + assert!(!manager + .pending_download_starts + .lock() + .await + .contains("reused-gid")); + + let new_epoch = manager.next_aria2_control_epoch("ignored-start").await; + manager + .begin_aria2_allocation( + "ignored-start", + new_epoch, + 21, + &SpawnPayload::default(), + ) + .await; + manager + .remember_gid("ignored-start".to_string(), "new-gid".to_string()) + .await; + assert_eq!( + allocation_pending_epoch(&manager, "ignored-start"), + Some((new_epoch, 21)) + ); + manager.handle_aria2_download_start("new-gid", 1).await; + assert_eq!(allocation_pending_epoch(&manager, "ignored-start"), None); + } + + #[tokio::test] + async fn allocation_is_cleared_by_terminal_reconciliation_and_cancellation() { + let app = tauri::test::mock_builder() + .build(tauri::test::mock_context(tauri::test::noop_assets())) + .expect("mock app"); + let manager = QueueManager::test_new(app.handle().clone(), 1, Arc::new(TestSpawner)); + let epoch = manager.next_aria2_control_epoch("terminal-allocation").await; + manager + .begin_aria2_allocation( + "terminal-allocation", + epoch, + 12, + &SpawnPayload::default(), + ) + .await; + manager + .apply_completion("terminal-allocation", PendingOutcome::Error("disk full".to_string())) + .await; + assert_eq!(allocation_pending_epoch(&manager, "terminal-allocation"), None); + + let next_epoch = manager.next_aria2_control_epoch("terminal-allocation").await; + manager + .begin_aria2_allocation( + "terminal-allocation", + next_epoch, + 13, + &SpawnPayload::default(), + ) + .await; + manager.clear_aria2_allocation("terminal-allocation").await; + assert_eq!(allocation_pending_epoch(&manager, "terminal-allocation"), None); + } + struct SeedSpawner; #[async_trait::async_trait] diff --git a/src/bindings/DownloadAllocationEvent.ts b/src/bindings/DownloadAllocationEvent.ts new file mode 100644 index 0000000..ab6471c --- /dev/null +++ b/src/bindings/DownloadAllocationEvent.ts @@ -0,0 +1,3 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +export type DownloadAllocationEvent = { id: string, pending: boolean, lifecycleGeneration: string, }; diff --git a/src/bindings/DownloadStateEvent.ts b/src/bindings/DownloadStateEvent.ts index 4d2ae67..027425f 100644 --- a/src/bindings/DownloadStateEvent.ts +++ b/src/bindings/DownloadStateEvent.ts @@ -1,4 +1,5 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. import type { DownloadErrorKind } from "./DownloadErrorKind"; +import type { DownloadStateProgress } from "./DownloadStateProgress"; -export type DownloadStateEvent = { id: string, status: string, error: string | null, errorKind?: DownloadErrorKind, resolverFallback?: boolean, fileName?: string, destination?: string, torrentSeedRemaining?: number, }; +export type DownloadStateEvent = { id: string, status: string, error: string | null, errorKind?: DownloadErrorKind, resolverFallback?: boolean, fileName?: string, destination?: string, torrentSeedRemaining?: number, progress?: DownloadStateProgress, }; diff --git a/src/bindings/DownloadStateProgress.ts b/src/bindings/DownloadStateProgress.ts new file mode 100644 index 0000000..0b4a7c2 --- /dev/null +++ b/src/bindings/DownloadStateProgress.ts @@ -0,0 +1,3 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +export type DownloadStateProgress = { fraction: number, downloadedBytes?: number, totalBytes?: number, totalIsEstimate?: boolean, }; diff --git a/src/ipc.ts b/src/ipc.ts index 8e31625..760742f 100644 --- a/src/ipc.ts +++ b/src/ipc.ts @@ -4,6 +4,7 @@ import { listen as tauriListen, type Event, type EventCallback, type UnlistenFn import type { DownloadCategory } from './bindings/DownloadCategory'; import type { DownloadProgressEvent } from './bindings/DownloadProgressEvent'; import type { DownloadStateEvent } from './bindings/DownloadStateEvent'; +import type { DownloadAllocationEvent } from './bindings/DownloadAllocationEvent'; import type { ExtensionDownload } from './bindings/ExtensionDownload'; import type { ExtensionCookieScope } from './bindings/ExtensionCookieScope'; import type { MediaMetadata } from './bindings/MediaMetadata'; @@ -200,6 +201,7 @@ export function invokeCommand( type EventMap = { 'schedule-trigger': { action: 'start' | 'stop'; key: string }; 'download-progress': DownloadProgressEvent; + 'download-allocation': DownloadAllocationEvent; 'download-state': DownloadStateEvent; 'torrent-move-progress': import('./bindings/TorrentMoveProgressEvent').TorrentMoveProgressEvent; 'download-complete': string; diff --git a/src/store/downloadProgressStore.ts b/src/store/downloadProgressStore.ts index 020a0f0..a9c5339 100644 --- a/src/store/downloadProgressStore.ts +++ b/src/store/downloadProgressStore.ts @@ -3,15 +3,67 @@ import type { DownloadProgressEvent } from '../bindings/DownloadProgressEvent'; interface DownloadProgressState { progressMap: Record; + retainedProgressMap: Record; moveProgressMap: Record; updateDownloadProgress: (id: string, payload: DownloadProgressEvent) => void; clearDownloadProgress: (id: string) => void; + resetDownloadProgress: (id: string) => void; setMoveProgress: (id: string, fraction: number) => void; clearMoveProgress: (id: string) => void; } +const finiteNonNegative = (value: unknown): value is number => + typeof value === 'number' && Number.isFinite(value) && value >= 0; + +const retainProgressSnapshot = ( + previous: DownloadProgressEvent | undefined, + next: DownloadProgressEvent, +): DownloadProgressEvent => { + const previousDownloaded = finiteNonNegative(previous?.downloaded_bytes) + ? previous.downloaded_bytes + : undefined; + const nextDownloaded = finiteNonNegative(next.downloaded_bytes) + ? next.downloaded_bytes + : undefined; + const downloadedBytes = previousDownloaded === undefined + ? nextDownloaded + : nextDownloaded === undefined + ? previousDownloaded + : Math.max(previousDownloaded, nextDownloaded); + + const exactTotal = [next, previous] + .find(snapshot => snapshot?.total_is_estimate === false + && finiteNonNegative(snapshot.total_bytes)) + ?.total_bytes; + const totalBytes = exactTotal + ?? (finiteNonNegative(next.total_bytes) + ? next.total_bytes + : finiteNonNegative(previous?.total_bytes) + ? previous.total_bytes + : undefined); + const totalIsEstimate = exactTotal !== undefined + ? false + : next.total_is_estimate ?? previous?.total_is_estimate; + const fractions = [previous?.fraction, next.fraction] + .filter(finiteNonNegative); + if (downloadedBytes !== undefined && totalBytes !== undefined && totalBytes > 0) { + fractions.push(Math.min(downloadedBytes, totalBytes) / totalBytes); + } + + return { + ...next, + fraction: fractions.length > 0 + ? Math.min(1, Math.max(0, Math.max(...fractions))) + : next.fraction, + ...(downloadedBytes !== undefined ? { downloaded_bytes: downloadedBytes } : {}), + ...(totalBytes !== undefined ? { total_bytes: totalBytes } : {}), + ...(totalIsEstimate !== undefined ? { total_is_estimate: totalIsEstimate } : {}) + }; +}; + export const useDownloadProgressStore = create((set) => ({ progressMap: {}, + retainedProgressMap: {}, moveProgressMap: {}, updateDownloadProgress: (id, payload) => set((state) => ({ @@ -19,6 +71,10 @@ export const useDownloadProgressStore = create((set) => ( ...state.progressMap, [id]: payload, }, + retainedProgressMap: { + ...state.retainedProgressMap, + [id]: retainProgressSnapshot(state.retainedProgressMap[id], payload), + }, })), clearDownloadProgress: (id) => set((state) => { @@ -29,6 +85,23 @@ export const useDownloadProgressStore = create((set) => ( delete nextMove[id]; return { progressMap: next, moveProgressMap: nextMove }; }), + resetDownloadProgress: (id) => + set((state) => { + if (!(id in state.progressMap) + && !(id in state.retainedProgressMap) + && !(id in state.moveProgressMap)) return state; + const next = { ...state.progressMap }; + delete next[id]; + const nextRetained = { ...state.retainedProgressMap }; + delete nextRetained[id]; + const nextMove = { ...state.moveProgressMap }; + delete nextMove[id]; + return { + progressMap: next, + retainedProgressMap: nextRetained, + moveProgressMap: nextMove + }; + }), setMoveProgress: (id, fraction) => set((state) => ({ moveProgressMap: { ...state.moveProgressMap, [id]: fraction } diff --git a/src/store/downloadStore.test.ts b/src/store/downloadStore.test.ts index 35b8dad..c19b3ae 100644 --- a/src/store/downloadStore.test.ts +++ b/src/store/downloadStore.test.ts @@ -18,7 +18,7 @@ describe('useDownloadProgressStore', () => { beforeEach(() => { vi.clearAllMocks(); vi.mocked(ipc.invokeCommand).mockResolvedValue(undefined); - useDownloadProgressStore.setState({ progressMap: {}, moveProgressMap: {} }); + useDownloadProgressStore.setState({ progressMap: {}, retainedProgressMap: {}, moveProgressMap: {} }); clearDownloadControlIntents(); }); @@ -44,7 +44,7 @@ describe('useDownloadProgressStore', () => { const first = initDownloadListener(); const second = initDownloadListener(); - expect(ipc.listenEvent).toHaveBeenCalledTimes(4); + expect(ipc.listenEvent).toHaveBeenCalledTimes(5); const releaseFirst = await first; const releaseSecond = await second; @@ -52,7 +52,7 @@ describe('useDownloadProgressStore', () => { expect(unlisten).not.toHaveBeenCalled(); releaseSecond(); - expect(unlisten).toHaveBeenCalledTimes(4); + expect(unlisten).toHaveBeenCalledTimes(5); }); it('ignores late progress and opposite terminal events from an older lifecycle', async () => { @@ -92,6 +92,88 @@ describe('useDownloadProgressStore', () => { release(); }); + it('projects native allocation events after admission and ignores stale generations', async () => { + const handlers: Record void> = {}; + vi.mocked(ipc.listenEvent).mockImplementation((event, handler) => { + handlers[event] = handler as (event: any) => void; + return Promise.resolve(vi.fn()); + }); + useDownloadStore.setState({ + downloads: [{ + id: 'native-allocation', + url: 'https://example.com/file.bin', + fileName: 'file.bin', + status: 'queued', + category: 'Other', + dateAdded: '' + }] + }); + + const release = await initDownloadListener(); + handlers['download-allocation']({ payload: { + id: 'native-allocation', + pending: true, + lifecycleGeneration: '0' + } }); + expect(useDownloadStore.getState().allocationPendingIds.has('native-allocation')).toBe(true); + + handlers['download-allocation']({ payload: { + id: 'native-allocation', + pending: false, + lifecycleGeneration: '1' + } }); + expect(useDownloadStore.getState().allocationPendingIds.has('native-allocation')).toBe(true); + + handlers['download-allocation']({ payload: { + id: 'native-allocation', + pending: false, + lifecycleGeneration: '0' + } }); + expect(useDownloadStore.getState().allocationPendingIds.has('native-allocation')).toBe(false); + release(); + }); + + it('retains a native allocation marker received before row hydration', async () => { + const handlers: Record void> = {}; + vi.mocked(ipc.listenEvent).mockImplementation((event, handler) => { + handlers[event] = handler as (event: any) => void; + return Promise.resolve(vi.fn()); + }); + useDownloadStore.setState({ + downloads: [], + allocationPendingIds: new Set() + }); + + const release = await initDownloadListener(); + handlers['download-allocation']({ payload: { + id: 'hydrating-allocation', + pending: true, + lifecycleGeneration: '0' + } }); + + expect(useDownloadStore.getState().allocationPendingIds.has('hydrating-allocation')).toBe(true); + + useDownloadStore.setState({ + downloads: [{ + id: 'hydrating-allocation', + url: 'https://example.com/file.bin', + fileName: 'file.bin', + status: 'downloading', + category: 'Other', + dateAdded: '' + }] + }); + expect(useDownloadStore.getState().allocationPendingIds.has('hydrating-allocation')).toBe(true); + + handlers['download-allocation']({ payload: { + id: 'hydrating-allocation', + pending: false, + lifecycleGeneration: '0' + } }); + expect(useDownloadStore.getState().allocationPendingIds.has('hydrating-allocation')).toBe(false); + release(); + }); + it('applies the authoritative destination carried by Torrent move completion', async () => { const handlers: Record void> = {}; vi.mocked(ipc.listenEvent).mockImplementation((event, handler) => { @@ -470,6 +552,182 @@ describe('useDownloadProgressStore', () => { expect(row.totalBytes).toBe(10240); expect(row.totalIsEstimate).toBe(true); expect(useDownloadProgressStore.getState().progressMap).toEqual({}); + expect(useDownloadProgressStore.getState().retainedProgressMap.snapshot).toMatchObject({ + fraction: 0.8, + downloaded_bytes: 8192, + total_bytes: 10240 + }); + release(); + }); + + it('retains progress for failed and paused rows when the live entry is absent', async () => { + const handlers: Record void> = {}; + vi.mocked(ipc.listenEvent).mockImplementation((event, handler) => { + handlers[event] = handler as (event: any) => void; + return Promise.resolve(vi.fn()); + }); + useDownloadStore.setState({ + downloads: [{ + id: 'terminal-progress', + url: 'https://example.com/file.bin', + fileName: 'file.bin', + status: 'downloading', + category: 'Other', + dateAdded: '' + }] + }); + + const release = await initDownloadListener(); + handlers['download-progress']({ payload: { + id: 'terminal-progress', + fraction: 0.7, + speed: '1 MB/s', + eta: '2s', + size: '10 MB', + size_is_final: false, + downloaded_bytes: 7000, + total_bytes: 10000, + total_is_estimate: false + } }); + handlers['download-state']({ payload: { + id: 'terminal-progress', + status: 'paused', + } }); + expect(useDownloadStore.getState().downloads[0]).toMatchObject({ + status: 'paused', + fraction: 0.7, + downloadedBytes: 7000 + }); + + useDownloadStore.setState(state => ({ + downloads: state.downloads.map(download => ({ ...download, status: 'downloading' as const })) + })); + handlers['download-state']({ payload: { + id: 'terminal-progress', + status: 'failed', + error: 'network stopped', + progress: { + fraction: 0.8, + downloadedBytes: 8000, + totalBytes: 10000, + totalIsEstimate: false + } + } }); + expect(useDownloadProgressStore.getState().progressMap['terminal-progress']).toBeUndefined(); + expect(useDownloadStore.getState().downloads[0]).toMatchObject({ + status: 'failed', + fraction: 0.8, + downloadedBytes: 8000, + totalBytes: 10000, + totalIsEstimate: false + }); + release(); + }); + + it('keeps retained bytes when a paused GID resumes through a queued state', async () => { + const handlers: Record void> = {}; + vi.mocked(ipc.listenEvent).mockImplementation((event, handler) => { + handlers[event] = handler as (event: any) => void; + return Promise.resolve(vi.fn()); + }); + useDownloadStore.setState({ + downloads: [{ + id: 'same-gid-resume', + url: 'https://example.com/file.bin', + fileName: 'file.bin', + status: 'downloading', + category: 'Other', + dateAdded: '' + }] + }); + + const release = await initDownloadListener(); + handlers['download-progress']({ payload: { + id: 'same-gid-resume', + fraction: 0.6, + speed: '1 MB/s', + eta: '4s', + size: '10 KB', + size_is_final: false, + downloaded_bytes: 6000, + total_bytes: 10000, + total_is_estimate: false + } }); + useDownloadStore.setState(state => ({ + downloads: state.downloads.map(download => ({ + ...download, + status: 'queued' as const + })) + })); + handlers['download-state']({ payload: { + id: 'same-gid-resume', + status: 'queued' + } }); + + expect(useDownloadProgressStore.getState().progressMap['same-gid-resume']).toBeUndefined(); + expect(useDownloadProgressStore.getState().retainedProgressMap['same-gid-resume']).toMatchObject({ + downloaded_bytes: 6000, + total_bytes: 10000 + }); + release(); + }); + + it('keeps the greatest retained byte count across retry frames', async () => { + const handlers: Record void> = {}; + vi.mocked(ipc.listenEvent).mockImplementation((event, handler) => { + handlers[event] = handler as (event: any) => void; + return Promise.resolve(vi.fn()); + }); + useDownloadStore.setState({ + downloads: [{ + id: 'retry-progress', + url: 'https://example.com/file.bin', + fileName: 'file.bin', + status: 'downloading', + category: 'Other', + dateAdded: '' + }] + }); + + const release = await initDownloadListener(); + const progress = (fraction: number, downloadedBytes: number) => handlers['download-progress']({ payload: { + id: 'retry-progress', + fraction, + speed: '1 MB/s', + eta: '2s', + size: '10 KB', + size_is_final: false, + downloaded_bytes: downloadedBytes, + total_bytes: 10000, + total_is_estimate: false + } }); + progress(0.8, 8000); + handlers['download-state']({ payload: { + id: 'retry-progress', + status: 'retrying', + error: 'network dropped' + } }); + expect(useDownloadStore.getState().downloads[0]).toMatchObject({ + status: 'retrying', + fraction: 0.8, + downloadedBytes: 8000, + totalBytes: 10000, + totalIsEstimate: false + }); + useDownloadStore.getState().updateDownload('retry-progress', { status: 'downloading' }); + progress(0.1, 1000); + handlers['download-state']({ payload: { + id: 'retry-progress', + status: 'failed', + error: 'retry exhausted' + } }); + + expect(useDownloadStore.getState().downloads[0]).toMatchObject({ + fraction: 0.8, + downloadedBytes: 8000, + totalBytes: 10000, + totalIsEstimate: false + }); release(); }); diff --git a/src/store/downloadStore.ts b/src/store/downloadStore.ts index 2d9b787..f12c4b0 100644 --- a/src/store/downloadStore.ts +++ b/src/store/downloadStore.ts @@ -8,6 +8,7 @@ import { useDownloadProgressStore } from './downloadProgressStore'; import { clearDownloadControlIntent, commitDownloadState, + currentDownloadLifecycleGeneration, downloadControlIntentFor, hasStaleTemporaryMediaEstimate, useDownloadStore @@ -16,15 +17,94 @@ import { export { useDownloadProgressStore } from './downloadProgressStore'; let unlistenProgress: UnlistenFn | null = null; +let unlistenAllocation: UnlistenFn | null = null; let unlistenState: UnlistenFn | null = null; let unlistenMoveProgress: UnlistenFn | null = null; let unlistenTray: UnlistenFn | null = null; let listenerSetup: Promise | null = null; let listenerConsumers = 0; +type ProgressFields = { + fraction?: number; + downloadedBytes?: number; + totalBytes?: number; + totalIsEstimate?: boolean; +}; + +const finiteNonNegative = (value: unknown): value is number => + typeof value === 'number' && Number.isFinite(value) && value >= 0; + +const progressFields = (source: unknown): ProgressFields => { + if (!source || typeof source !== 'object') return {}; + const value = source as Record; + const downloadedBytes = value.downloadedBytes ?? value.downloaded_bytes; + const totalBytes = value.totalBytes ?? value.total_bytes; + const totalIsEstimate = value.totalIsEstimate ?? value.total_is_estimate; + return { + ...(finiteNonNegative(value.fraction) ? { fraction: value.fraction } : {}), + ...(finiteNonNegative(downloadedBytes) ? { downloadedBytes } : {}), + ...(finiteNonNegative(totalBytes) ? { totalBytes } : {}), + ...(typeof totalIsEstimate === 'boolean' ? { totalIsEstimate } : {}) + }; +}; + +const mergeTerminalProgress = ( + current: DownloadItem, + status: DownloadStatus, + nativeSnapshot: unknown, + retainedSnapshot: unknown, + liveSnapshot: unknown +): ProgressFields => { + const ordered = [nativeSnapshot, retainedSnapshot, liveSnapshot] + .map(progressFields); + const row = progressFields({ + fraction: current.fraction, + downloadedBytes: current.downloadedBytes, + totalBytes: current.totalBytes, + totalIsEstimate: current.totalIsEstimate + }); + const all = [...ordered, row]; + const downloadedCandidates = all + .map(snapshot => snapshot.downloadedBytes) + .filter((value): value is number => finiteNonNegative(value)); + const downloadedBytes = downloadedCandidates.length > 0 + ? Math.max(...downloadedCandidates) + : undefined; + + const exactTotals = all + .filter(snapshot => snapshot.totalIsEstimate === false && finiteNonNegative(snapshot.totalBytes)) + .map(snapshot => snapshot.totalBytes!); + const anyTotals = all + .map(snapshot => snapshot.totalBytes) + .filter((value): value is number => finiteNonNegative(value)); + const totalBytes = exactTotals[0] ?? anyTotals[0]; + const fractions = all + .map(snapshot => snapshot.fraction) + .filter((value): value is number => finiteNonNegative(value)); + if (downloadedBytes !== undefined && totalBytes !== undefined && totalBytes > 0) { + fractions.push(Math.min(downloadedBytes, totalBytes) / totalBytes); + } + if (status === 'completed') fractions.push(1); + const fraction = fractions.length > 0 + ? Math.min(1, Math.max(0, Math.max(...fractions))) + : undefined; + return { + ...(fraction !== undefined ? { fraction } : {}), + ...(downloadedBytes !== undefined ? { downloadedBytes } : {}), + ...(totalBytes !== undefined ? { totalBytes } : {}), + ...(exactTotals.length > 0 + ? { totalIsEstimate: false } + : ordered.find(snapshot => snapshot.totalIsEstimate !== undefined)?.totalIsEstimate !== undefined + ? { totalIsEstimate: ordered.find(snapshot => snapshot.totalIsEstimate !== undefined)!.totalIsEstimate } + : {}) + }; +}; + const disposeDownloadListeners = () => { unlistenProgress?.(); unlistenProgress = null; + unlistenAllocation?.(); + unlistenAllocation = null; unlistenState?.(); unlistenState = null; unlistenMoveProgress?.(); @@ -43,7 +123,7 @@ const startDownloadListeners = async () => { if (!current) { // A removed row can still have one queued sidecar event in flight. // Do not let that event recreate an orphaned progress entry. - useDownloadProgressStore.getState().clearDownloadProgress(payload.id); + useDownloadProgressStore.getState().resetDownloadProgress(payload.id); return; } // A sidecar can flush one last progress chunk after a pause, failure, @@ -112,12 +192,39 @@ const startDownloadListeners = async () => { mainStore.updateDownload(payload.id, updates); } }), + listen('download-allocation', (event) => { + const payload = event.payload; + const mainStore = useDownloadStore.getState(); + const current = mainStore.downloads.find(download => download.id === payload.id); + if (!current) { + // Keep a validated native marker until persisted startup state or a + // just-admitted row is projected. Dropping it here makes allocation + // invisible when the event wins the hydration race. + mainStore.setAllocationPending( + payload.id, + payload.pending, + payload.lifecycleGeneration + ); + return; + } + // Allocation events are native lifecycle markers. A late marker from an + // older GID/queue lifecycle must never hide the current lifecycle's + // phase or clear its pending state. + if (payload.lifecycleGeneration !== currentDownloadLifecycleGeneration(payload.id)) { + return; + } + mainStore.setAllocationPending( + payload.id, + payload.pending, + payload.lifecycleGeneration + ); + }), listen('download-state', async (event) => { const payload = event.payload; const mainStore = useDownloadStore.getState(); const current = mainStore.downloads.find(d => d.id === payload.id); if (!current) { - useDownloadProgressStore.getState().clearDownloadProgress(payload.id); + useDownloadProgressStore.getState().resetDownloadProgress(payload.id); return; } const status = payload.status as DownloadStatus; @@ -184,8 +291,26 @@ const startDownloadListeners = async () => { return; } - const progress = useDownloadProgressStore.getState().progressMap[payload.id]; - if (['queued', 'retrying', 'completed', 'failed', 'paused', 'waitingToSeed'].includes(status)) { + const progressState = useDownloadProgressStore.getState(); + const liveProgress = progressState.progressMap[payload.id]; + const retainedProgress = progressState.retainedProgressMap[payload.id]; + const isTerminalOrPaused = ['completed', 'failed', 'paused', 'retrying', 'waitingToSeed'].includes(status); + const terminalProgress = isTerminalOrPaused + ? mergeTerminalProgress( + current, + status, + payload.progress, + retainedProgress, + liveProgress + ) + : undefined; + if (status === 'queued') { + // A queued event can represent either a genuinely new admission or a + // same-GID resume of a paused Aria2 transfer. Lifecycle-changing + // callers reset the retained snapshot before admission; this event + // only ends the old live frame so a same-GID resume keeps its bytes. + useDownloadProgressStore.getState().clearDownloadProgress(payload.id); + } else if (['retrying', 'completed', 'failed', 'paused', 'waitingToSeed'].includes(status)) { useDownloadProgressStore.getState().clearDownloadProgress(payload.id); } const moveRestoreStatus = status === 'moving' @@ -196,16 +321,18 @@ const startDownloadListeners = async () => { const updates: Partial = { status, torrentMoveRestoreStatus: moveRestoreStatus, - ...(progress ? { - fraction: progress.fraction, - ...(progress.downloaded_bytes != null - ? { downloadedBytes: progress.downloaded_bytes } + ...(terminalProgress ? { + ...(terminalProgress.fraction !== undefined + ? { fraction: terminalProgress.fraction } : {}), - ...(progress.total_bytes != null - ? { totalBytes: progress.total_bytes } + ...(terminalProgress.downloadedBytes !== undefined + ? { downloadedBytes: terminalProgress.downloadedBytes } : {}), - ...(progress.total_is_estimate != null - ? { totalIsEstimate: progress.total_is_estimate } + ...(terminalProgress.totalBytes !== undefined + ? { totalBytes: terminalProgress.totalBytes } + : {}), + ...(terminalProgress.totalIsEstimate !== undefined + ? { totalIsEstimate: terminalProgress.totalIsEstimate } : {}) } : {}), ...(payload.error ? { @@ -326,13 +453,15 @@ const startDownloadListeners = async () => { throw failedRegistration.reason; } - const [progress, state, moveProgress, tray] = registrations as [ + const [progress, allocation, state, moveProgress, tray] = registrations as [ + PromiseFulfilledResult, PromiseFulfilledResult, PromiseFulfilledResult, PromiseFulfilledResult, PromiseFulfilledResult, ]; unlistenProgress = progress.value; + unlistenAllocation = allocation.value; unlistenState = state.value; unlistenMoveProgress = moveProgress.value; unlistenTray = tray.value; diff --git a/src/store/useDownloadStore.test.ts b/src/store/useDownloadStore.test.ts index 5cc5c49..13f01b7 100644 --- a/src/store/useDownloadStore.test.ts +++ b/src/store/useDownloadStore.test.ts @@ -108,7 +108,7 @@ describe('useDownloadStore', () => { pendingAddRequestContexts: {}, pendingAddRequestVersion: 0, }); - useDownloadProgressStore.setState({ progressMap: {} }); + useDownloadProgressStore.setState({ progressMap: {}, retainedProgressMap: {}, moveProgressMap: {} }); }); it('invalidates in-flight Add-modal handoffs when the modal is toggled', () => { @@ -1337,7 +1337,7 @@ describe('useDownloadStore', () => { ).toHaveLength(2); }); - it('exposes an indeterminate allocation phase while normal enqueue is blocked', async () => { + it('does not expose allocation while admission is merely blocked', async () => { useDownloadStore.setState({ downloads: [{ id: 'allocation-phase', @@ -1365,15 +1365,19 @@ describe('useDownloadStore', () => { const dispatch = dispatchItem('allocation-phase'); await vi.waitFor(() => { - expect(useDownloadStore.getState().allocationPendingIds.has('allocation-phase')).toBe(true); + expect(ipc.invokeCommand).toHaveBeenCalledWith( + 'enqueue_download', + expect.objectContaining({ item: expect.objectContaining({ id: 'allocation-phase' }) }) + ); }); + expect(useDownloadStore.getState().allocationPendingIds.has('allocation-phase')).toBe(false); resolveEnqueue({ id: 'allocation-phase', filename: 'file.bin' }); await expect(dispatch).resolves.toBe(true); expect(useDownloadStore.getState().allocationPendingIds.has('allocation-phase')).toBe(false); }); - it('exposes allocation phase for a preallocated Torrent and strips metadata credentials', async () => { + it('does not expose Torrent allocation while admission is merely blocked and strips metadata credentials', async () => { useDownloadStore.setState({ downloads: [{ id: 'torrent-allocation-phase', @@ -1411,8 +1415,12 @@ describe('useDownloadStore', () => { const dispatch = dispatchItem('torrent-allocation-phase'); await vi.waitFor(() => { - expect(useDownloadStore.getState().allocationPendingIds.has('torrent-allocation-phase')).toBe(true); + expect(ipc.invokeCommand).toHaveBeenCalledWith( + 'enqueue_download', + expect.objectContaining({ item: expect.objectContaining({ id: 'torrent-allocation-phase' }) }) + ); }); + expect(useDownloadStore.getState().allocationPendingIds.has('torrent-allocation-phase')).toBe(false); resolveEnqueue({ id: 'torrent-allocation-phase', filename: 'payload' }); await expect(dispatch).resolves.toBe(true); @@ -2555,7 +2563,46 @@ describe('useDownloadStore', () => { }); }); - it('shows and clears allocation phase for a blocked startup Torrent batch', async () => { + it('keeps startup destination permission failures retryable without backend registration', async () => { + vi.mocked(ipc.invokeCommand).mockImplementation(async (cmd: string) => { + if (cmd === 'db_get_all_queues') return []; + if (cmd === 'db_get_all_downloads') { + return [JSON.stringify({ + id: 'startup-destination-access', + url: 'https://example.com/file.bin', + fileName: 'file.bin', + destination: '/protected', + status: 'queued', + category: 'Other', + dateAdded: '', + queueId: '00000000-0000-0000-0000-000000000001', + hasBeenDispatched: true + })]; + } + if (cmd === 'enqueue_many') { + return [{ + id: 'startup-destination-access', + success: false, + error: 'destination access retryable: grant Firelink access to the selected folder and retry' + }]; + } + if (cmd === 'get_pending_order') return []; + return undefined; + }); + + await useDownloadStore.getState().initDB(); + await useDownloadStore.getState().resumePendingDownloads(); + + expect(useDownloadStore.getState().downloads[0]).toMatchObject({ + status: 'ready', + hasBeenDispatched: false, + lastErrorKind: 'destinationAccess', + lastError: 'grant Firelink access to the selected folder and retry' + }); + expect(useDownloadStore.getState().backendRegisteredIds.has('startup-destination-access')).toBe(false); + }); + + it('does not show allocation for a startup Torrent batch while it is merely queued', async () => { let releaseEnqueue!: (value: Array<{ id: string; success: boolean; filename: string }>) => void; const enqueue = new Promise>(resolve => { releaseEnqueue = resolve; @@ -2590,7 +2637,7 @@ describe('useDownloadStore', () => { const resume = useDownloadStore.getState().resumePendingDownloads(); await vi.waitFor(() => { - expect(useDownloadStore.getState().allocationPendingIds.has('startup-torrent-allocation')).toBe(true); + expect(useDownloadStore.getState().allocationPendingIds.has('startup-torrent-allocation')).toBe(false); expect(ipc.invokeCommand).toHaveBeenCalledWith( 'enqueue_many', expect.objectContaining({ diff --git a/src/store/useDownloadStore.ts b/src/store/useDownloadStore.ts index 10e0e7b..9dd1cde 100644 --- a/src/store/useDownloadStore.ts +++ b/src/store/useDownloadStore.ts @@ -10,7 +10,7 @@ import type { ExtensionCookieScope } from '../bindings/ExtensionCookieScope'; import type { Queue } from '../bindings/Queue'; import { useSettingsStore } from './useSettingsStore'; import { useDownloadProgressStore } from './downloadProgressStore'; -import { canonicalizeDownloadFileName, categoryForDownload, categoryForFileName, hasCredentialBearingHeaders, isActiveDownloadStatus, isAllocationPhaseEligible, isTransferActiveStatus, isValidTorrentExcludeTrackerList, isValidTorrentTrackerList, MAX_TORRENT_STOP_TIMEOUT, normalizeSpeedLimitForBackend, normalizeTorrentEncryptionPolicy, normalizeTorrentFileAllocation, normalizeTorrentPrioritizePiece, normalizeTorrentTrackerInterval, normalizeTorrentTrackerTimeout, redactDownloadForPersistence, resolveDownloadConnections } from '../utils/downloads'; +import { canonicalizeDownloadFileName, categoryForDownload, categoryForFileName, hasCredentialBearingHeaders, isActiveDownloadStatus, isTransferActiveStatus, isValidTorrentExcludeTrackerList, isValidTorrentTrackerList, MAX_TORRENT_STOP_TIMEOUT, normalizeSpeedLimitForBackend, normalizeTorrentEncryptionPolicy, normalizeTorrentFileAllocation, normalizeTorrentPrioritizePiece, normalizeTorrentTrackerInterval, normalizeTorrentTrackerTimeout, redactDownloadForPersistence, resolveDownloadConnections } from '../utils/downloads'; import { resolveCategoryDestination } from '../utils/downloadLocations'; @@ -214,14 +214,26 @@ const advanceDownloadLifecycle = (id: string): bigint => { const currentDownloadLifecycle = (id: string): bigint => downloadLifecycleGenerations.get(id) ?? 0n; +export const currentDownloadLifecycleGeneration = (id: string): string => + currentDownloadLifecycle(id).toString(); + type DispatchInvalidation = { generation: bigint; pendingDispatch?: Promise; }; -const invalidateDispatch = async (id: string): Promise => { +const invalidateDispatch = async ( + id: string, + resetRetainedProgress = false, +): Promise => { const generation = currentDownloadLifecycle(id); const nextGeneration = advanceDownloadLifecycle(id); + // A new lifecycle cannot inherit the previous native allocation phase. The + // backend will emit a fresh marker for the new generation after admission. + useDownloadStore.getState().clearAllocationPending(id); + if (resetRetainedProgress) { + useDownloadProgressStore.getState().resetDownloadProgress(id); + } try { await invoke('cancel_enqueue_generation', { id, generation: generation.toString() }); } catch (error) { @@ -230,8 +242,11 @@ const invalidateDispatch = async (id: string): Promise => return { generation: nextGeneration, pendingDispatch: backendDispatchPromises.get(id) }; }; -const invalidateAndWaitForDispatch = async (id: string): Promise => { - const { pendingDispatch } = await invalidateDispatch(id); +const invalidateAndWaitForDispatch = async ( + id: string, + resetRetainedProgress = false, +): Promise => { + const { pendingDispatch } = await invalidateDispatch(id, resetRetainedProgress); if (!pendingDispatch) return false; await pendingDispatch; return true; @@ -434,18 +449,7 @@ async function dispatchItemInternal(id: string, proxyOverride?: string | null): ) { return false; } - const showsAllocationPhase = isAllocationPhaseEligible(admittedItem); - if (showsAllocationPhase) { - useDownloadStore.getState().setAllocationPending(id, true); - } - let accepted; - try { - accepted = await invoke('enqueue_download', { item: enqueueItem }); - } finally { - if (showsAllocationPhase) { - useDownloadStore.getState().setAllocationPending(id, false); - } - } + const accepted = await invoke('enqueue_download', { item: enqueueItem }); backendAccepted = true; if (!isCurrentDownloadLifecycle(id, lifecycleGeneration)) { await removeStaleBackendDispatch(id); @@ -485,6 +489,12 @@ async function dispatchItemInternal(id: string, proxyOverride?: string | null): const proxyBlocked = isSystemProxyConfigurationError(e); const destinationAccessBlocked = isRetryableDestinationAccessError(e); const message = errorMessage(e); + if (destinationAccessBlocked) { + useDownloadStore.getState().clearAllocationPending(id); + useDownloadStore.setState(state => ({ + pendingOrder: state.pendingOrder.filter(value => value !== id) + })); + } useDownloadStore.getState().updateDownload(id, { status: proxyBlocked ? 'queued' : destinationAccessBlocked ? 'ready' : 'failed', hasBeenDispatched: false, @@ -1045,7 +1055,8 @@ interface DownloadState { allocationPendingIds: Set; registerBackendIds: (ids: string[]) => void; unregisterBackendIds: (ids: string[]) => void; - setAllocationPending: (id: string, pending: boolean) => void; + setAllocationPending: (id: string, pending: boolean, lifecycleGeneration: string) => void; + clearAllocationPending: (id: string) => void; applyProperties: (id: string, updates: Partial) => Promise; moveInQueue: (ids: string | string[], direction: 'up' | 'down') => Promise; moveManyInQueueToPosition: ( @@ -1326,7 +1337,7 @@ export const useDownloadStore = create((set, get) => { // Fence any older enqueue before replacing a paused backend lifecycle. // Otherwise a late addUri result can win the race and make this // selection start outside the requested order. - const { pendingDispatch } = await invalidateDispatch(id); + const { pendingDispatch } = await invalidateDispatch(id, true); if (pendingDispatch) await pendingDispatch; targetItem = get().downloads.find(download => download.id === id); if (!targetItem || !canStartDownload(targetItem.status)) { @@ -1425,7 +1436,7 @@ export const useDownloadStore = create((set, get) => { // A terminal aria2 gid is intentionally re-enqueued as a new // lifecycle. Advance and cancel the old generation before dispatching // so QueueManager does not reject the legitimate user retry as stale. - await invalidateAndWaitForDispatch(id); + await invalidateAndWaitForDispatch(id, true); dispatchSucceeded = await dispatchItemInternal(id); } @@ -1662,12 +1673,22 @@ export const useDownloadStore = create((set, get) => { for (const id of ids) nextSet.delete(id); return { backendRegisteredIds: nextSet }; }), - setAllocationPending: (id, pending) => set((state) => { + setAllocationPending: (id, pending, lifecycleGeneration) => set((state) => { + // Native allocation events can arrive while persisted rows are still + // hydrating. Validate against the frontend lifecycle counter even when no + // row exists yet, then retain the marker until that row is projected. + if (lifecycleGeneration !== currentDownloadLifecycleGeneration(id)) return state; const nextSet = new Set(state.allocationPendingIds); if (pending) nextSet.add(id); else nextSet.delete(id); return { allocationPendingIds: nextSet }; }), + clearAllocationPending: (id) => set((state) => { + if (!state.allocationPendingIds.has(id)) return state; + const nextSet = new Set(state.allocationPendingIds); + nextSet.delete(id); + return { allocationPendingIds: nextSet }; + }), isAddModalOpen: false, pendingAddUrls: '', pendingAddReferer: '', @@ -1852,6 +1873,8 @@ export const useDownloadStore = create((set, get) => { hasBeenDispatched: false }; advanceDownloadLifecycle(item.id); + get().clearAllocationPending(item.id); + useDownloadProgressStore.getState().resetDownloadProgress(item.id); set((state) => ({ downloads: reorderQueueWithPausedAtEnd([...state.downloads, ownedItem], queueId) })); @@ -1997,7 +2020,7 @@ export const useDownloadStore = create((set, get) => { } throw error; } - useDownloadProgressStore.getState().clearDownloadProgress(id); + useDownloadProgressStore.getState().resetDownloadProgress(id); info(`Download ${id} removed`); syncSystemIntegrations(); }, @@ -2066,6 +2089,7 @@ export const useDownloadStore = create((set, get) => { hasBeenDispatched: false, dateAdded: new Date().toISOString() }); + useDownloadProgressStore.getState().resetDownloadProgress(id); await commitDownloadState(); @@ -2765,25 +2789,24 @@ export const useDownloadStore = create((set, get) => { currentDownloadLifecycle(item.id).toString() === item.lifecycle_generation; }); if (dispatchableItems.length === 0) return; - const allocationPendingIds = dispatchableItems - .filter(item => { - const current = latestItems.get(item.id); - return current !== undefined && isAllocationPhaseEligible(current); - }) - .map(item => item.id); - allocationPendingIds.forEach(id => get().setAllocationPending(id, true)); - - let results; - try { - results = await invoke('enqueue_many', { items: dispatchableItems }); - } finally { - allocationPendingIds.forEach(id => get().setAllocationPending(id, false)); - } + const results = await invoke('enqueue_many', { items: dispatchableItems }); const registeredIds = results.filter(result => result.success).map(result => result.id); - const failedErrors = new Map( + const failedResults = new Map( results .filter(result => !result.success) - .map(result => [result.id, result.error || 'Backend rejected the queued download.']) + .map(result => { + const message = result.error || 'Backend rejected the queued download.'; + const destinationAccess = isRetryableDestinationAccessError(message); + return [result.id, { + message: destinationAccess + ? destinationAccessErrorMessage(message) + : message, + status: destinationAccess ? 'ready' as const : 'failed' as const, + errorKind: destinationAccess + ? ('destinationAccess' as DownloadErrorKind) + : undefined + }]; + }) ); const acceptedFilenames = new Map( results @@ -2816,12 +2839,15 @@ export const useDownloadStore = create((set, get) => { ...state.backendRegisteredIds, ...liveAcceptedIds ]), + pendingOrder: state.pendingOrder.filter(id => !failedResults.has(id)), downloads: state.downloads.map(download => - failedErrors.has(download.id) + failedResults.has(download.id) ? { ...download, - status: 'failed' as const, - lastError: failedErrors.get(download.id) + status: failedResults.get(download.id)!.status, + hasBeenDispatched: false, + lastError: failedResults.get(download.id)!.message, + lastErrorKind: failedResults.get(download.id)!.errorKind } : liveAcceptedIds.has(download.id) ? {