From 33b61448bc8784d5e6f4d74ee152a2576875dfe9 Mon Sep 17 00:00:00 2001 From: NimBold Date: Sat, 10 Oct 2026 01:49:52 +0330 Subject: [PATCH] fix(download): harden queue and lifecycle recovery - Preserve selected queue priority during bulk starts and sparse moves - Reconcile ambiguous Aria2 admission and bound metadata retries - Snapshot scheduled stops and keep pause/completion state authoritative - Bound shutdown waits and retain recoverable tray actions - Cover staged pauses, queue races, and retry boundaries --- src-tauri/src/ipc.rs | 11 + src-tauri/src/lib.rs | 414 +++++++++++++++++++++++++---- src-tauri/src/queue.rs | 392 +++++++++++++++++++++++++-- src-tauri/src/retry.rs | 85 ++++-- src-tauri/src/scheduler.rs | 152 ++++++++++- src-tauri/src/settings.rs | 13 +- src-tauri/tests/queue_manager.rs | 91 +++++++ src/App.tsx | 19 ++ src/components/SchedulerView.tsx | 33 ++- src/ipc.ts | 6 +- src/store/useDownloadStore.test.ts | 167 +++++++++++- src/store/useDownloadStore.ts | 217 ++++++++++----- src/store/useSettingsStore.test.ts | 21 ++ src/store/useSettingsStore.ts | 6 +- src/utils/downloads.test.ts | 2 +- src/utils/downloads.ts | 2 +- 16 files changed, 1444 insertions(+), 187 deletions(-) diff --git a/src-tauri/src/ipc.rs b/src-tauri/src/ipc.rs index 0766dce..29f5a63 100644 --- a/src-tauri/src/ipc.rs +++ b/src-tauri/src/ipc.rs @@ -689,6 +689,14 @@ pub struct SchedulerSettings { pub post_queue_action: PostQueueAction, } +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct SchedulerRunSnapshot { + pub start_key: String, + pub stop_key: Option, + pub stop_minute: Option, +} + #[derive(Clone, Debug, Serialize, Deserialize, TS)] #[serde(rename_all = "camelCase")] #[ts(export, export_to = "../../src/bindings/")] @@ -738,6 +746,9 @@ pub struct PersistedSettings { #[ts(optional)] pub scheduler_triggered_start_key: Option, pub scheduler_last_stop_key: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[ts(skip)] + pub scheduler_active_run_snapshot: Option, pub last_custom_speed_limit_ki_b: u32, #[serde(default = "default_speed_limit_unit")] pub last_custom_speed_limit_unit: String, diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 0346994..ab00aff 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -270,6 +270,113 @@ fn retry_metadata_with_cookies( true } +async fn wait_for_metadata_retry_with( + retry_strike: &mut usize, + reason: &str, + retryable: bool, + mut sleep: S, +) -> bool +where + S: FnMut(Duration) -> Fut, + Fut: std::future::Future, +{ + if !retryable || *retry_strike >= crate::retry::MAX_RETRIES { + return false; + } + + let delay = crate::retry::backoff_for_error(*retry_strike, reason); + log::debug!( + "metadata retry [retry={} error_class={} delay_ms={}]", + *retry_strike + 1, + crate::retry::network_error_class(reason), + delay.as_millis() + ); + sleep(delay).await; + *retry_strike += 1; + true +} + +/// Retry idempotent metadata reads after their fallback method has been +/// selected, sharing one bounded retry budget across ranged GETs, redirects, +/// and auth retry. HEAD failures still switch promptly to the GET fallback. +/// Transient HTTP statuses are retried as well as connection failures. +async fn retry_metadata_request_with( + retry_strike: &mut usize, + mut send: F, + classify_response: C, + mut sleep: S, +) -> Result +where + F: FnMut() -> Fut, + Fut: std::future::Future>, + E: std::fmt::Display, + C: Fn(&T) -> String, + S: FnMut(Duration) -> SleepFut, + SleepFut: std::future::Future, +{ + loop { + match send().await { + Ok(response) => { + let reason = classify_response(&response); + if wait_for_metadata_retry_with( + retry_strike, + &reason, + crate::retry::is_transient_network_error(&reason), + &mut sleep, + ) + .await + { + drop(response); + continue; + } + return Ok(response); + } + Err(error) => { + let reason = error.to_string(); + if !wait_for_metadata_retry_with( + retry_strike, + &reason, + crate::retry::is_transient_network_error(&reason), + &mut sleep, + ) + .await + { + return Err(error); + } + } + } + } +} + +async fn send_metadata_request_with_retry( + retry_strike: &mut usize, + send: F, +) -> Result +where + F: FnMut() -> Fut, + Fut: std::future::Future>, +{ + retry_metadata_request_with( + retry_strike, + send, + |response: &reqwest::Response| format!("HTTP {}", response.status().as_u16()), + tokio::time::sleep, + ) + .await +} + +fn metadata_transport_error_message(operation: &str, error: &reqwest::Error) -> String { + let reason = match reqwest_error_code(error) { + "timeout" => "request timed out", + "dns" => "DNS lookup failed", + "connect" => "connection could not be established", + "body" => "response could not be read", + "request" => "request was invalid", + _ => "network error", + }; + format!("{operation} metadata request failed: {reason}") +} + #[derive(Serialize, TS)] #[ts(export, export_to = "../../src/bindings/")] pub struct MetadataResponse { @@ -1916,6 +2023,7 @@ async fn fetch_remote_torrent_bytes( let cookies_available = metadata_cookie_header_present(headers, cookies, cookie_scopes); let mut send_cookies = !cookies_available; let mut cookie_retry_attempted = false; + let mut network_retry_strike = 0; let mut redirect_count = 0; loop { @@ -1965,20 +2073,22 @@ async fn fetch_remote_torrent_bytes( let client = builder .build() .map_err(|error| format!("could not create torrent metadata client: {error}"))?; - let response = client - .get(current.clone()) - .header( - reqwest::header::ACCEPT, - "application/x-bittorrent, application/octet-stream", - ) - .send() - .await - .map_err(|error| { - format!( - "remote torrent metadata request failed: {}", - crate::redact_sensitive_text(&error.to_string()) + let response = send_metadata_request_with_retry(&mut network_retry_strike, || { + client + .get(current.clone()) + .header( + reqwest::header::ACCEPT, + "application/x-bittorrent, application/octet-stream", ) - })?; + .send() + }) + .await + .map_err(|error| { + format!( + "remote torrent metadata request failed: {}", + crate::redact_sensitive_text(&error.to_string()) + ) + })?; if should_retry_metadata_with_cookies( response.status(), @@ -2027,13 +2137,34 @@ async fn fetch_remote_torrent_bytes( let mut bytes = Vec::new(); let mut stream = response.bytes_stream(); + let mut retry_body = false; while let Some(chunk) = stream.next().await { - let chunk = chunk.map_err(|error| { - format!( - "remote torrent metadata response failed: {}", - crate::redact_sensitive_text(&error.to_string()) - ) - })?; + let chunk = match chunk { + Ok(chunk) => chunk, + Err(error) => { + let reason = error.to_string(); + let retryable_body_error = error.is_timeout() + || error.is_connect() + || error.is_body() + || crate::retry::is_transient_network_error(&reason); + if retryable_body_error + && wait_for_metadata_retry_with( + &mut network_retry_strike, + &reason, + true, + tokio::time::sleep, + ) + .await + { + retry_body = true; + break; + } + return Err(format!( + "remote torrent metadata response failed: {}", + crate::redact_sensitive_text(&reason) + )); + } + }; if bytes .len() .checked_add(chunk.len()) @@ -2046,6 +2177,9 @@ async fn fetch_remote_torrent_bytes( } bytes.extend_from_slice(&chunk); } + if retry_body { + continue; + } if bytes.is_empty() { return Err("remote torrent metadata response was empty".to_string()); } @@ -2128,6 +2262,7 @@ async fn fetch_metadata( let mut current_url = url.clone(); let original_origin = reqwest::Url::parse(&url).ok(); let mut redirects = 0; + let mut network_retry_strike = 0; let cookies_available = metadata_cookie_header_present( headers.as_deref(), cookies.as_deref(), @@ -2313,7 +2448,11 @@ async fn fetch_metadata( Err(head_error) => { let head_elapsed_ms = head_started.elapsed().as_millis(); let get_started = Instant::now(); - match build_get_range().send().await { + match send_metadata_request_with_retry(&mut network_retry_strike, || { + build_get_range().send() + }) + .await + { Ok(response) => { log::warn!( "metadata [stage=metadata operation=head result=fallback host={} error_code={} head_elapsed_ms={} get_status={} get_elapsed_ms={} total_elapsed_ms={}]", @@ -2378,7 +2517,11 @@ async fn fetch_metadata( if needs_fallback { let get_started = Instant::now(); - current_res = match build_get_range().send().await { + current_res = match send_metadata_request_with_retry(&mut network_retry_strike, || { + build_get_range().send() + }) + .await + { Ok(response) => { log::info!( "metadata [stage=metadata operation=get_range host={} result=ok status={} elapsed_ms={} total_elapsed_ms={}]", @@ -2397,7 +2540,7 @@ async fn fetch_metadata( get_started.elapsed().as_millis(), metadata_started.elapsed().as_millis() ); - return Err(error.to_string()); + return Err(metadata_transport_error_message("Ranged GET", &error)); } }; @@ -4252,15 +4395,15 @@ async fn shutdown_aria2_daemon(app_handle: tauri::AppHandle) { } }; (outcome, child_wait_started.elapsed()) - }) - .await; - match child_wait { - Ok((outcome, elapsed)) => log::info!( + }); + match tokio::time::timeout(std::time::Duration::from_secs(10), child_wait).await { + Ok(Ok((outcome, elapsed))) => log::info!( "app_shutdown_phase=aria2_child outcome={} elapsed_ms={}", outcome, elapsed.as_millis() ), - Err(error) => log::warn!("aria2 child wait task failed: {error}"), + Ok(Err(error)) => log::warn!("aria2 child wait task failed: {error}"), + Err(_) => log::warn!("app_shutdown_phase=aria2_child outcome=timeout"), } } log::info!( @@ -4269,6 +4412,27 @@ async fn shutdown_aria2_daemon(app_handle: tauri::AppHandle) { ); } +fn reject_app_exit_request(app: &tauri::AppHandle, request: u64, reason: &str) { + app.state::().cancel_shutdown(); + let tray_action_pending = app + .state::() + .reopen_after_exit_rejected(); + if tray_action_pending { + if let Err(error) = app.emit("tray-action-available", ()) { + log::warn!("pending tray action wakeup failed after exit rejection: {error}"); + } + } + let _ = app.emit_to( + "main", + "app-exit-cancelled", + serde_json::json!({ + "requestId": request.to_string(), + "reason": reason + }), + ); + restore_main_window(app); +} + impl Drop for Aria2DaemonGuard { fn drop(&mut self) { if let Ok(mut lock) = self.child.lock() { @@ -6107,7 +6271,7 @@ async fn pause_download( app_handle: tauri::AppHandle, state: tauri::State<'_, AppState>, id: String, -) -> Result<(), String> { +) -> Result { properties_window::ensure_main_window(&caller)?; log::info!("pause_download called for id: {}", id); @@ -6201,7 +6365,7 @@ async fn pause_download( progress.or(status_progress.clone()), ) .await; - return Ok(()); + return Ok(false); } Ok((status, progress)) if matches!(status.as_str(), "error" | "removed") => { state.queue_manager.clear_aria2_allocation(&id).await; @@ -6274,7 +6438,7 @@ async fn pause_download( status_progress.clone(), ) .await; - return Ok(()); + return Ok(false); } terminal => { state.queue_manager.clear_aria2_allocation(&id).await; @@ -6300,7 +6464,7 @@ async fn pause_download( event = event.with_progress(progress); } let _ = app_handle.emit("download-state", event); - return Ok(()); + return Ok(true); } state.queue_manager.release_registered_id(&id).await; return Err(format!( @@ -6331,14 +6495,14 @@ async fn pause_download( event = event.with_progress(progress); } let _ = app_handle.emit("download-state", event); - return Ok(()); + return Ok(true); } // A terminal runner may have already released its lifecycle while this // command was waiting for the per-download control lock. Treat pause as // an idempotent no-op instead of emitting a stale Paused state. if active_kind.is_none() && !removed_pending && registered_lifecycle_generation.is_none() { - return Ok(()); + return Ok(false); } if matches!(active_kind, Some(crate::queue::TaskKind::Aria2)) { @@ -6394,7 +6558,7 @@ async fn pause_download( "download-state", crate::ipc::DownloadStateEvent::paused_with_seed_remaining(id, seed_remaining), ); - Ok(()) + Ok(true) } #[tauri::command] @@ -9468,6 +9632,13 @@ fn ack_schedule_trigger( } "stop" => { state.insert("schedulerLastStopKey".to_string(), serde_json::json!(key)); + let snapshot_stop_key = state + .get("schedulerActiveRunSnapshot") + .and_then(|snapshot| snapshot.get("stopKey")) + .and_then(serde_json::Value::as_str); + if snapshot_stop_key == Some(key.as_str()) { + state.remove("schedulerActiveRunSnapshot"); + } } _ => {} })?; @@ -9479,7 +9650,14 @@ fn ack_schedule_trigger( settings.scheduler_last_start_key = key; settings.scheduler_triggered_start_key = None; } else { - settings.scheduler_last_stop_key = key; + settings.scheduler_last_stop_key = key.clone(); + if settings + .scheduler_active_run_snapshot + .as_ref() + .is_some_and(|snapshot| snapshot.stop_key.as_deref() == Some(key.as_str())) + { + settings.scheduler_active_run_snapshot = None; + } } } } @@ -9499,6 +9677,48 @@ async fn get_pending_order( Ok(state.queue_manager.pending_order(queue_id.as_deref()).await) } +#[tauri::command] +async fn begin_queue_dispatch_hold( + caller: tauri::WebviewWindow, + state: tauri::State<'_, AppState>, + queue_id: String, +) -> Result { + properties_window::ensure_main_window(&caller).map_err(AppError::Internal)?; + state + .queue_manager + .begin_queue_dispatch_hold(&queue_id) + .await + .map_err(AppError::Internal) +} + +#[tauri::command] +async fn renew_queue_dispatch_hold( + caller: tauri::WebviewWindow, + state: tauri::State<'_, AppState>, + queue_id: String, + token: String, +) -> Result { + properties_window::ensure_main_window(&caller).map_err(AppError::Internal)?; + Ok(state + .queue_manager + .renew_queue_dispatch_hold(&queue_id, &token) + .await) +} + +#[tauri::command] +async fn end_queue_dispatch_hold( + caller: tauri::WebviewWindow, + state: tauri::State<'_, AppState>, + queue_id: String, + token: String, +) -> Result { + properties_window::ensure_main_window(&caller).map_err(AppError::Internal)?; + Ok(state + .queue_manager + .end_queue_dispatch_hold(&queue_id, &token) + .await) +} + fn enqueue_lifecycle_generation(item: &queue::EnqueueItem) -> Result { item.lifecycle_generation .as_deref() @@ -15869,7 +16089,7 @@ mod tests { media_output_template, media_progress_args, media_progress_event_totals, media_progress_speed, cookie_scope_for_url, metadata_authentication_error, metadata_cookie_header_present, - metadata_headers, metadata_response_error, + metadata_headers, metadata_response_error, metadata_transport_error_message, metadata_error_code, reqwest_error_code, is_remote_torrent_source, @@ -15895,6 +16115,7 @@ mod tests { browser_cookie_fallback, normalize_media_cookie_file_for_request, should_fallback_from_browser_cookies, should_fallback_from_netscape_cookie_file, retry_metadata_with_cookies, should_retry_metadata_with_cookies, + retry_metadata_request_with, should_send_metadata_credentials, collect_log_files, FirelinkDeepLink, ytdlp_youtube_extractor_args, YTDLP_YOUTUBE_PLAYER_CLIENTS, percent_decode_metadata_value, MediaProgress, @@ -15982,6 +16203,85 @@ mod tests { assert!(error.is_dns()); assert_eq!(reqwest_error_code(&error), "dns"); + assert_eq!( + metadata_transport_error_message("Ranged GET", &error), + "Ranged GET metadata request failed: DNS lookup failed" + ); + } + + #[tokio::test] + async fn metadata_request_retries_transient_statuses_and_stops_on_permanent_statuses() { + let mut retry_strike = 0; + let mut attempts = 0; + let status = retry_metadata_request_with( + &mut retry_strike, + || { + attempts += 1; + std::future::ready(Ok::(if attempts == 1 { 503 } else { 200 })) + }, + |status| format!("HTTP {status}"), + |_| std::future::ready(()), + ) + .await + .expect("the retry should return the successful response"); + assert_eq!(status, 200); + assert_eq!(attempts, 2); + assert_eq!(retry_strike, 1); + + let mut retry_strike = 0; + let mut attempts = 0; + let status = retry_metadata_request_with( + &mut retry_strike, + || { + attempts += 1; + std::future::ready(if attempts == 1 { + Err("operation timed out".to_string()) + } else { + Ok(200_u16) + }) + }, + |status| format!("HTTP {status}"), + |_| std::future::ready(()), + ) + .await + .expect("a transient transport failure should retry"); + assert_eq!(status, 200); + assert_eq!(attempts, 2); + assert_eq!(retry_strike, 1); + + let mut retry_strike = 0; + let mut attempts = 0; + let status = retry_metadata_request_with( + &mut retry_strike, + || { + attempts += 1; + std::future::ready(Ok::(404)) + }, + |status| format!("HTTP {status}"), + |_| std::future::ready(()), + ) + .await + .expect("permanent responses are returned to the caller"); + assert_eq!(status, 404); + assert_eq!(attempts, 1); + assert_eq!(retry_strike, 0); + + let mut retry_strike = crate::retry::MAX_RETRIES; + let mut attempts = 0; + let status = retry_metadata_request_with( + &mut retry_strike, + || { + attempts += 1; + std::future::ready(Ok::(503)) + }, + |status| format!("HTTP {status}"), + |_| std::future::ready(()), + ) + .await + .expect("exhausted retries leave the final response for normal error handling"); + assert_eq!(status, 503); + assert_eq!(attempts, 1); + assert_eq!(retry_strike, crate::retry::MAX_RETRIES); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -23014,6 +23314,9 @@ pub fn run() { cancel_enqueue_generation, move_in_queue, move_many_in_queue, + begin_queue_dispatch_hold, + renew_queue_dispatch_hold, + end_queue_dispatch_hold, remove_from_queue, get_pending_order, ]); @@ -23158,6 +23461,9 @@ pub fn run() { | "cancel_enqueue_generation" | "move_in_queue" | "move_many_in_queue" + | "begin_queue_dispatch_hold" + | "renew_queue_dispatch_hold" + | "end_queue_dispatch_hold" | "remove_from_queue" | "get_pending_order" => download_queue_handler(invoke), "reveal_in_file_manager" @@ -23242,9 +23548,18 @@ pub fn run() { .clone(); let request = frontend_exit_flush.request(); let mut completed = frontend_exit_flush.completed.subscribe(); - let _ = app.emit_to("main", "app-exit-requested", serde_json::json!({ + if let Err(error) = app.emit_to("main", "app-exit-requested", serde_json::json!({ "requestId": request.to_string() - })); + })) { + log::warn!("app_shutdown_phase=frontend_flush signal failed: {error}"); + frontend_exit_flush.finish(request, false); + reject_app_exit_request(&app, request, "signal-failed"); + log::info!( + "app_shutdown_phase=total outcome=cancelled elapsed_ms={}", + exit_started.elapsed().as_millis() + ); + return; + } let flush_wait = async move { loop { let status = *completed.borrow_and_update(); @@ -23301,26 +23616,7 @@ pub fn run() { } else { "rejected" }; - app.state::().cancel_shutdown(); - let tray_action_pending = app - .state::() - .reopen_after_exit_rejected(); - if tray_action_pending { - if let Err(error) = app.emit("tray-action-available", ()) { - log::warn!( - "pending tray action wakeup failed after exit rejection: {error}" - ); - } - } - let _ = app.emit_to( - "main", - "app-exit-cancelled", - serde_json::json!({ - "requestId": request.to_string(), - "reason": cancel_reason - }), - ); - restore_main_window(&app); + reject_app_exit_request(&app, request, cancel_reason); log::info!( "app_shutdown_phase=total outcome=cancelled elapsed_ms={}", exit_started.elapsed().as_millis() @@ -23358,7 +23654,7 @@ pub fn run() { } }; match tokio::time::timeout( - Duration::from_secs(60), + Duration::from_secs(5), final_flush_wait, ) .await diff --git a/src-tauri/src/queue.rs b/src-tauri/src/queue.rs index d0b2f52..3282d38 100644 --- a/src-tauri/src/queue.rs +++ b/src-tauri/src/queue.rs @@ -28,6 +28,9 @@ 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; +// The frontend refreshes this lease every 20 seconds. A bounded expiry +// releases a queue if the renderer disappears before its finally cleanup. +const QUEUE_START_BATCH_HOLD_LEASE: Duration = Duration::from_secs(90); fn encode_aria2_torrent_payload(bytes: &[u8]) -> String { base64::engine::general_purpose::STANDARD.encode(bytes) @@ -1277,6 +1280,10 @@ pub struct QueueManager { /// Queue overrides are stored only for queues with an explicit limit. /// Missing entries inherit the global target capacity. queue_limits: Mutex>, + /// Queue-scoped leases used while a multi-row start request re-enqueues + /// paused downloads. The lease prevents older pending rows from taking a + /// newly freed slot before the selected block has been restored to order. + queue_dispatch_holds: Mutex>>, /// One entry represents either a queued dispatch reservation or an active /// transfer. Keeping both phases in one map makes queue-slot ownership /// exactly-once across the async addUri handoff. @@ -1408,6 +1415,7 @@ impl QueueManager { active_permit_generations: Mutex::new(HashMap::new()), active_kinds: Mutex::new(HashMap::new()), queue_limits: Mutex::new(HashMap::new()), + queue_dispatch_holds: Mutex::new(HashMap::new()), queue_permit_ownership: Mutex::new(HashMap::new()), admission_gate: Mutex::new(()), system_action_pending: AtomicBool::new(false), @@ -2387,6 +2395,101 @@ impl QueueManager { Ok(()) } + /// Begin a bounded queue-scoped dispatch hold for a multi-row start + /// operation. Other queues remain eligible while this queue is held. + pub async fn begin_queue_dispatch_hold(&self, queue_id: &str) -> Result { + let queue_id = queue_id.trim(); + if queue_id.is_empty() { + return Err("Queue id cannot be empty".to_string()); + } + + let _admission_gate = self.admission_gate.lock().await; + if self.system_action_pending.load(Ordering::Acquire) { + return Err("System action is already being performed".to_string()); + } + let now = Instant::now(); + let token = uuid::Uuid::new_v4().simple().to_string(); + let mut holds = self.queue_dispatch_holds.lock().await; + let queue_holds = holds.entry(queue_id.to_string()).or_default(); + queue_holds.retain(|_, expires_at| *expires_at > now); + queue_holds.insert(token.clone(), now + QUEUE_START_BATCH_HOLD_LEASE); + Ok(token) + } + + /// Renew a live start-batch lease. An expired token cannot be resurrected; + /// the caller must stop relying on its queue-order guarantee in that case. + pub async fn renew_queue_dispatch_hold(&self, queue_id: &str, token: &str) -> bool { + let _admission_gate = self.admission_gate.lock().await; + let queue_id = queue_id.trim(); + if queue_id.is_empty() { + return false; + } + let now = Instant::now(); + let mut holds = self.queue_dispatch_holds.lock().await; + let Some(queue_holds) = holds.get_mut(queue_id) else { + return false; + }; + let Some(expires_at) = queue_holds.get_mut(token) else { + return false; + }; + if *expires_at <= now { + queue_holds.remove(token); + if queue_holds.is_empty() { + holds.remove(queue_id); + } + return false; + } + *expires_at = now + QUEUE_START_BATCH_HOLD_LEASE; + true + } + + /// Release only the lease named by `token`, so late cleanup from an older + /// start request cannot release a newer request's hold. + pub async fn end_queue_dispatch_hold(&self, queue_id: &str, token: &str) -> bool { + let _admission_gate = self.admission_gate.lock().await; + let queue_id = queue_id.trim(); + if queue_id.is_empty() { + return false; + } + let mut holds = self.queue_dispatch_holds.lock().await; + let removed = holds + .get_mut(queue_id) + .is_some_and(|queue_holds| queue_holds.remove(token).is_some()); + if holds.get(queue_id).is_some_and(HashMap::is_empty) { + holds.remove(queue_id); + } + drop(holds); + drop(_admission_gate); + if removed { + self.notify.notify_one(); + } + removed + } + + async fn active_queue_dispatch_holds(&self) -> HashSet { + let now = Instant::now(); + let mut holds = self.queue_dispatch_holds.lock().await; + holds.retain(|_, queue_holds| { + queue_holds.retain(|_, expires_at| *expires_at > now); + !queue_holds.is_empty() + }); + holds.keys().cloned().collect() + } + + async fn next_queue_dispatch_hold_delay(&self) -> Option { + let now = Instant::now(); + let mut holds = self.queue_dispatch_holds.lock().await; + holds.retain(|_, queue_holds| { + queue_holds.retain(|_, expires_at| *expires_at > now); + !queue_holds.is_empty() + }); + holds + .values() + .flat_map(HashMap::values) + .min() + .map(|expires_at| expires_at.saturating_duration_since(now)) + } + pub async fn next_aria2_control_epoch(&self, id: &str) -> u64 { let mut epochs = self.aria2_control_epochs.lock().await; let epoch = epochs.get(id).copied().unwrap_or_default().wrapping_add(1); @@ -3848,11 +3951,15 @@ impl QueueManager { return None; } + let held_queues = self.active_queue_dispatch_holds().await; let ownership = self.queue_permit_ownership.lock().await; let mut queue_ids = Vec::new(); let mut seen = HashSet::new(); for task in pending.iter() { - if !ownership.contains_key(&task.id) && seen.insert(task.queue_id.clone()) { + if !ownership.contains_key(&task.id) + && !held_queues.contains(&task.queue_id) + && seen.insert(task.queue_id.clone()) + { queue_ids.push(task.queue_id.clone()); } } @@ -4441,7 +4548,14 @@ impl QueueManager { // all queue/global capacity is occupied. The notification // future is created before inspection to close the lost-wake // window without polling or sleeping. - notified.await; + if let Some(hold_delay) = self.next_queue_dispatch_hold_delay().await { + tokio::select! { + _ = notified => {} + _ = tokio::time::sleep(hold_delay) => {} + } + } else { + notified.await; + } } } } @@ -6478,11 +6592,6 @@ impl QueueManager { .map(|index| pending[*index].clone()) .collect::>(); let selected_ids = ids.iter().collect::>(); - let selected_tasks = queue_tasks - .iter() - .filter(|task| selected_ids.contains(&task.id)) - .cloned() - .collect::>(); let unselected_tasks = queue_tasks .iter() .filter(|task| !selected_ids.contains(&task.id)) @@ -6490,12 +6599,13 @@ impl QueueManager { .collect::>(); let first_selected = *selected_positions.first().unwrap(); - let last_selected = *selected_positions.last().unwrap(); - let selected_count = selected_tasks.len(); let insert_index = match direction { QueueDirection::Up => first_selected.saturating_sub(1), - QueueDirection::Down => (last_selected + 1) - .saturating_sub(selected_count) + // Move the selected block one slot after its current + // insertion point among unselected tasks. Using the last + // selected index and selected count collapses sparse + // selections too far toward the end. + QueueDirection::Down => first_selected .saturating_add(1) .min(unselected_tasks.len()), }; @@ -7364,6 +7474,97 @@ fn aria2_add_uri_params( serde_json::json!([transfer_uris, options]) } +fn new_aria2_transfer_gid() -> String { + // Aria2 GIDs are 16 hexadecimal characters. A random GID also acts as an + // idempotency key when an addUri/addTorrent response is lost: every retry + // carries the same value, so Aria2 cannot admit a second transfer. + uuid::Uuid::new_v4().simple().to_string()[..16].to_string() +} + +fn apply_aria2_transfer_gid( + options: &mut serde_json::Map, + gid: &str, +) { + options.insert("gid".to_string(), serde_json::json!(gid)); +} + +async fn add_transfer_rpc_with_reconciliation( + method: &str, + expected_gid: &str, + deadline: Instant, + mut add_call: AddCall, + mut status_call: StatusCall, +) -> Result +where + AddCall: FnMut() -> AddFuture, + AddFuture: Future>, + StatusCall: FnMut() -> StatusFuture, + StatusFuture: Future>, +{ + let mut ambiguous_add = false; + let mut status_only = false; + loop { + if !status_only { + match add_call().await { + Ok(result) => return Ok(result), + Err(error) => { + let transient = is_aria2_rpc_unavailable(&error); + if transient { + ambiguous_add = true; + } + + if !transient && !ambiguous_add { + return Err(error); + } + + // A transport timeout/reset does not tell us whether + // Aria2 admitted the transfer before the response was + // lost. Query the stable GID before retrying the add. + // If that status query is itself unavailable, retries + // still use the same GID and remain idempotent. + if status_call().await.is_ok() { + return Ok(serde_json::json!(expected_gid)); + } + // Once a retry returns an application error after an + // earlier ambiguous transport failure, keep checking the + // original GID instead of repeatedly issuing an add that + // Aria2 may already have accepted. + if !transient { + status_only = true; + } + if Instant::now() >= deadline { + if ambiguous_add { + log::warn!( + "aria2 {method} for gid {expected_gid} remained ambiguous after retry reconciliation; handing the GID to the queue poller" + ); + return Ok(serde_json::json!(expected_gid)); + } + return Err(error); + } + } + } + } + + if status_only { + if status_call().await.is_ok() { + return Ok(serde_json::json!(expected_gid)); + } + if Instant::now() >= deadline { + // The first add may have succeeded even though both its + // response and subsequent status checks were unavailable. + // Returning the stable GID lets normal queue reconciliation + // either retain the live transfer or recover a missing one. + log::warn!( + "aria2 {method} for gid {expected_gid} could not be verified; handing the GID to the queue poller" + ); + return Ok(serde_json::json!(expected_gid)); + } + } + + tokio::time::sleep(Duration::from_millis(250)).await; + } +} + fn apply_aria2_normal_reliability_options( options: &mut serde_json::Map, payload: &SpawnPayload, @@ -8350,6 +8551,7 @@ impl ProductionSpawner { state: &crate::AppState, method: &str, params: &serde_json::Value, + expected_gid: &str, ) -> Result { let deadline = std::time::Instant::now() + std::time::Duration::from_secs(15); loop { @@ -8373,15 +8575,28 @@ impl ProductionSpawner { continue; } - match crate::rpc_call(port, &state.aria2_secret, method, params.clone()).await { - Ok(result) => return Ok(result), - Err(error) => { - if !is_aria2_rpc_unavailable(&error) || std::time::Instant::now() >= deadline { - return Err(error); - } - tokio::time::sleep(std::time::Duration::from_millis(250)).await; - } - } + let secret = state.aria2_secret.clone(); + let secret_for_status = secret.clone(); + let method_for_call = method.to_string(); + let params_for_call = params.clone(); + let gid_for_status = expected_gid.to_string(); + return add_transfer_rpc_with_reconciliation( + method, + expected_gid, + deadline, + move || { + let secret = secret.clone(); + let method = method_for_call.clone(); + let params = params_for_call.clone(); + async move { crate::rpc_call(port, &secret, &method, params).await } + }, + move || { + let secret = secret_for_status.clone(); + let gid = gid_for_status.clone(); + async move { crate::aria2_download_status(port, &secret, &gid).await } + }, + ) + .await; } } @@ -8447,6 +8662,8 @@ impl SidecarSpawner for ProductionSpawner { let attempt_epoch = state.queue_manager.current_aria2_control_epoch(id).await; let admission_started = Instant::now(); let mut options = serde_json::Map::new(); + let expected_gid = new_aria2_transfer_gid(); + apply_aria2_transfer_gid(&mut options, &expected_gid); // Keep hostname resolution route-aware and non-blocking for every // normal and Torrent transfer. When the Firelink route contract is // available, transfers use NativeAsyncResolver on background worker @@ -8674,7 +8891,10 @@ impl SidecarSpawner for ProductionSpawner { return Err("aria2 admission canceled before RPC".to_string()); } - match self.add_transfer_rpc(&state, method, ¶ms).await { + match self + .add_transfer_rpc(&state, method, ¶ms, &expected_gid) + .await + { Ok(result) => { let gid = result.as_str().unwrap_or("").to_string(); if gid.is_empty() { @@ -9451,6 +9671,35 @@ mod tests { assert!(!manager.media_payload_completed("media-removal", 7).await); } + #[tokio::test] + async fn expired_queue_dispatch_hold_cannot_be_renewed_and_stale_end_is_scoped() { + let app = tauri::test::mock_builder() + .build(tauri::test::mock_context(tauri::test::noop_assets())) + .unwrap(); + let manager = QueueManager::test_new(app.handle().clone(), 1, Arc::new(TestSpawner)); + + let first = manager.begin_queue_dispatch_hold("queue").await.unwrap(); + let second = manager.begin_queue_dispatch_hold("queue").await.unwrap(); + assert!(manager.end_queue_dispatch_hold("queue", &first).await); + assert!(manager + .active_queue_dispatch_holds() + .await + .contains("queue")); + + manager + .queue_dispatch_holds + .lock() + .await + .get_mut("queue") + .unwrap() + .insert(second.clone(), Instant::now() - Duration::from_secs(1)); + assert!(!manager.renew_queue_dispatch_hold("queue", &second).await); + assert!(!manager + .active_queue_dispatch_holds() + .await + .contains("queue")); + } + #[tokio::test] async fn explicit_allocation_telemetry_ends_without_payload_and_rejects_old_epoch() { let app = tauri::test::mock_builder() @@ -12352,6 +12601,107 @@ mod tests { )); } + #[test] + fn aria2_transfer_gid_is_a_stable_16_digit_option() { + let gid = new_aria2_transfer_gid(); + assert_eq!(gid.len(), 16); + assert!(gid + .chars() + .all(|character| character.is_ascii_hexdigit() && !character.is_ascii_uppercase())); + + let mut options = serde_json::Map::new(); + apply_aria2_transfer_gid(&mut options, &gid); + let params = aria2_add_uri_params(vec!["https://example.test/file".to_string()], options); + assert_eq!(params[1]["gid"], serde_json::json!(gid)); + } + + #[tokio::test] + async fn ambiguous_add_rpc_adopts_the_gid_after_a_lost_response() { + let gid = "0123456789abcdef"; + let add_calls = Arc::new(AtomicUsize::new(0)); + let status_calls = Arc::new(AtomicUsize::new(0)); + let add_calls_for_rpc = Arc::clone(&add_calls); + let status_calls_for_rpc = Arc::clone(&status_calls); + let result = add_transfer_rpc_with_reconciliation( + "aria2.addUri", + gid, + Instant::now() + Duration::from_secs(2), + move || { + let attempt = add_calls_for_rpc.fetch_add(1, Ordering::SeqCst); + async move { + if attempt == 0 { + // Aria2 accepted the request, but the client lost the + // response and cannot yet query the daemon. + Err("error sending request: connection reset".to_string()) + } else { + // The retry has the same GID, so Aria2 rejects a + // duplicate instead of creating a second transfer. + Err(format!("GID#{gid} already exists")) + } + } + }, + move || { + let attempt = status_calls_for_rpc.fetch_add(1, Ordering::SeqCst); + async move { + if attempt == 0 { + Err("error trying to connect: connection refused".to_string()) + } else { + Ok("active".to_string()) + } + } + }, + ) + .await + .expect("the accepted GID should be adopted"); + + assert_eq!(result, serde_json::json!(gid)); + assert_eq!(add_calls.load(Ordering::SeqCst), 2); + assert_eq!(status_calls.load(Ordering::SeqCst), 2); + } + + #[tokio::test] + async fn nontransient_add_rejection_does_not_adopt_an_existing_gid() { + let gid = "0123456789abcdef"; + let status_calls = Arc::new(AtomicUsize::new(0)); + let status_calls_for_rpc = Arc::clone(&status_calls); + let result = add_transfer_rpc_with_reconciliation( + "aria2.addUri", + gid, + Instant::now() + Duration::from_secs(2), + || async { Err("GID#0123456789abcdef is not unique".to_string()) }, + move || { + status_calls_for_rpc.fetch_add(1, Ordering::SeqCst); + async { Ok("active".to_string()) } + }, + ) + .await; + + assert!(result.is_err()); + assert_eq!(status_calls.load(Ordering::SeqCst), 0); + } + + #[tokio::test] + async fn unresolved_ambiguous_add_returns_gid_for_queue_reconciliation() { + let gid = "fedcba9876543210"; + let add_calls = Arc::new(AtomicUsize::new(0)); + let add_calls_for_rpc = Arc::clone(&add_calls); + let result = add_transfer_rpc_with_reconciliation( + "aria2.addTorrent", + gid, + Instant::now(), + move || { + add_calls_for_rpc.fetch_add(1, Ordering::SeqCst); + async { Err("request timed out".to_string()) } + }, + || async { Err("connection reset while checking status".to_string()) }, + ) + .await + .expect("ambiguous admission remains owned by normal queue reconciliation"); + + assert_eq!(result, serde_json::json!(gid)); + assert_eq!(add_calls.load(Ordering::SeqCst), 1); + } + #[test] fn known_aria2_startup_failure_is_returned_without_rpc_retry() { assert_eq!( diff --git a/src-tauri/src/retry.rs b/src-tauri/src/retry.rs index 007850d..7fe9bbc 100644 --- a/src-tauri/src/retry.rs +++ b/src-tauri/src/retry.rs @@ -176,6 +176,25 @@ pub fn backoff_for(strike: usize) -> Duration { .unwrap_or_else(|| *BACKOFF_SCHEDULE.last().expect("schedule is non-empty")) } +/// Resolve the shared retry delay for an error without echoing the error into +/// diagnostics. Rate limiting gets its longer schedule only when the failure +/// actually contains an HTTP 429 status, not when an unrelated URL happens to +/// contain the digits `429`. +pub fn backoff_for_error(strike: usize, reason: &str) -> Duration { + let lower = reason.to_ascii_lowercase(); + let is_rate_limited = contains_http_status(&lower, "429") + || contains_labeled_http_status(&lower, "429") + || lower.contains("429 too many requests"); + if is_rate_limited { + BACKOFF_SCHEDULE_429 + .get(strike) + .copied() + .unwrap_or_else(|| *BACKOFF_SCHEDULE_429.last().unwrap()) + } else { + backoff_for(strike) + } +} + /// Classify an error string as a transient network condition worth retrying. /// /// Returns `true` for socket drops, connect/read timeouts, connection resets, @@ -232,6 +251,16 @@ fn contains_http_status(message: &str, status: &str) -> bool { }) } +fn contains_labeled_http_status(message: &str, status: &str) -> bool { + let expected = format!("status={status}"); + message.split_ascii_whitespace().any(|token| { + token + .trim_start_matches(|character| matches!(character, '(' | '[')) + .trim_end_matches(|character: char| matches!(character, ')' | ']' | ',' | '.' | ';')) + == expected + }) +} + pub fn is_transient_network_error(message: &str) -> bool { if is_permanent_network_error(message) { return false; @@ -243,7 +272,7 @@ pub fn is_transient_network_error(message: &str) -> bool { let m = message.to_ascii_lowercase(); - const TRANSIENT: [&str; 36] = [ + const TRANSIENT: [&str; 25] = [ // socket-layer / HTTP-client phrasing surfaced by aria2 and yt-dlp "timed out", "timeout", @@ -269,17 +298,6 @@ pub fn is_transient_network_error(message: &str) -> bool { "503 service unavailable", "429 too many requests", // aria2c HTTP error formats - "status=408", - "status=429", - "status=500", - "status=502", - "status=503", - "status=504", - "status=520", - "status=521", - "status=522", - "status=523", - "status=524", // aria2c log phrasing "connection was closed", "timeout.", @@ -292,6 +310,9 @@ pub fn is_transient_network_error(message: &str) -> bool { .iter() .any(|status| contains_http_status(&m, status)) || TRANSIENT.iter().any(|t| m.contains(t)) + || TRANSIENT_HTTP_STATUS + .iter() + .any(|status| contains_labeled_http_status(&m, status)) } /// Outcome of a cancel-safe backoff sleep wrapped around a transient retry. @@ -314,14 +335,7 @@ pub async fn backoff_and_emit( ) -> BackoffOutcome { let attempt = strike + 1; emit(format!("Network drop — retry #{attempt}: {reason}")); - let delay = if reason.to_ascii_lowercase().contains("429") { - BACKOFF_SCHEDULE_429 - .get(strike) - .copied() - .unwrap_or_else(|| *BACKOFF_SCHEDULE_429.last().unwrap()) - } else { - backoff_for(strike) - }; + let delay = backoff_for_error(strike, &reason); tokio::select! { _ = tokio::time::sleep(delay) => BackoffOutcome::Continue, _ = interrupt => BackoffOutcome::Aborted, @@ -414,6 +428,34 @@ mod tests { assert_eq!(backoff_for(usize::MAX), Duration::from_secs(10)); } + #[test] + fn rate_limit_backoff_requires_an_http_429_status() { + assert_eq!( + backoff_for_error(0, "HTTP Error 429: Too Many Requests"), + Duration::from_secs(60) + ); + assert_eq!( + backoff_for_error(1, "The response status is not successful. status=429"), + Duration::from_secs(120) + ); + assert_eq!( + backoff_for_error(2, "HTTP 429 Too Many Requests"), + Duration::from_secs(300) + ); + assert_eq!( + backoff_for_error(0, "https://example.test/file/429?token=x timed out"), + Duration::from_secs(2) + ); + assert_eq!( + backoff_for_error(0, "https://example.test/file?status=429 connection failed"), + Duration::from_secs(2) + ); + assert_eq!( + backoff_for_error(0, "request failed with status=429"), + Duration::from_secs(60) + ); + } + // --- transient classification: positive cases ------------------------- #[test] @@ -452,6 +494,9 @@ mod tests { assert!(is_transient_network_error("HTTP Error 429: Too Many Requests")); assert!(is_transient_network_error("429 too many requests")); assert!(is_transient_network_error("The response status is not successful. status=429")); + assert!(!is_transient_network_error( + "https://example.test/file?status=429 connection failed" + )); } #[test] diff --git a/src-tauri/src/scheduler.rs b/src-tauri/src/scheduler.rs index 07be3dd..1cb5b19 100644 --- a/src-tauri/src/scheduler.rs +++ b/src-tauri/src/scheduler.rs @@ -1,4 +1,4 @@ -use chrono::{Datelike, Local, Timelike}; +use chrono::{Datelike, Local, NaiveDate, Timelike}; use std::collections::HashMap; use std::sync::{Arc, RwLock}; use std::time::Duration; @@ -71,16 +71,68 @@ fn overnight_stop_is_due(check: OvernightStopCheck<'_>) -> bool { && last_stop_key != stop_key } +fn scheduler_run_snapshot( + start_key: &str, + scheduler: &crate::ipc::SchedulerSettings, +) -> crate::ipc::SchedulerRunSnapshot { + let stop = if scheduler.stop_time_enabled { + let start_minute = minute_of_day(&scheduler.start_time); + let stop_minute = minute_of_day(&scheduler.stop_time); + start_minute.zip(stop_minute).and_then(|(start, stop)| { + let date = start_key + .strip_suffix("-start") + .and_then(|date| NaiveDate::parse_from_str(date, "%Y-%m-%d").ok())?; + let stop_date = if stop < start { date.succ_opt()? } else { date }; + Some((format!("{}-stop", stop_date.format("%Y-%m-%d")), stop)) + }) + } else { + None + }; + + crate::ipc::SchedulerRunSnapshot { + start_key: start_key.to_string(), + stop_key: stop.as_ref().map(|(key, _)| key.clone()), + stop_minute: stop.map(|(_, minute)| minute), + } +} + +fn snapshot_stop_is_due( + snapshot: &crate::ipc::SchedulerRunSnapshot, + current_stop_key: &str, + current_minute: u32, + last_stop_key: &str, +) -> bool { + snapshot + .stop_key + .as_deref() + .zip(snapshot.stop_minute) + .is_some_and(|(stop_key, stop_minute)| { + stop_key == current_stop_key + && current_minute >= stop_minute + && last_stop_key != stop_key + }) +} + fn persist_scheduler_start_trigger( app_handle: &tauri::AppHandle, settings_cache: &Arc>>, key: &str, + scheduler: &crate::ipc::SchedulerSettings, ) { + let snapshot = scheduler_run_snapshot(key, scheduler); if let Err(error) = crate::settings::update_settings_state(app_handle, |state| { state.insert( "schedulerTriggeredStartKey".to_string(), serde_json::json!(key), ); + state.insert( + "schedulerActiveRunSnapshot".to_string(), + serde_json::json!({ + "startKey": snapshot.start_key, + "stopKey": snapshot.stop_key, + "stopMinute": snapshot.stop_minute + }), + ); }) { log::warn!("Failed to persist scheduler start trigger: {error}"); } @@ -88,6 +140,7 @@ fn persist_scheduler_start_trigger( if let Ok(mut settings) = settings_cache.write() { if let Some(settings) = settings.as_mut() { settings.scheduler_triggered_start_key = Some(key.to_string()); + settings.scheduler_active_run_snapshot = Some(snapshot); } } } @@ -117,6 +170,7 @@ pub fn spawn_scheduler( .clone() .unwrap_or_default(), settings.scheduler_last_stop_key.clone(), + settings.scheduler_active_run_snapshot.clone(), ) }) }); @@ -125,6 +179,7 @@ pub fn spawn_scheduler( scheduler_last_start_key, persisted_triggered_start_key, scheduler_last_stop_key, + scheduler_active_run_snapshot, )) = settings { if !scheduler.enabled { @@ -156,8 +211,11 @@ pub fn spawn_scheduler( .get("start") .is_none_or(|instant| instant.elapsed() >= Duration::from_secs(5)) { - if persisted_triggered_start_key != start_key - && triggered_start_key != start_key + if (persisted_triggered_start_key != start_key + && triggered_start_key != start_key) + || scheduler_active_run_snapshot + .as_ref() + .is_none_or(|snapshot| snapshot.start_key != start_key) { // Record the dispatch intent before emitting so a // crash between the native event and renderer ack @@ -168,6 +226,7 @@ pub fn spawn_scheduler( &app_handle, &settings_cache, &start_key, + &scheduler, ); } if app_handle.emit( @@ -229,7 +288,25 @@ pub fn spawn_scheduler( stop_key: &stop_key, }); - if (same_day_stop_due || overnight_stop_due) + let active_snapshot = scheduler_active_run_snapshot.as_ref().filter(|snapshot| { + snapshot.start_key == scheduler_last_start_key + || snapshot.start_key == persisted_triggered_start_key + || snapshot.start_key == triggered_start_key + }); + let snapshot_stop_due = active_snapshot.is_some_and(|snapshot| { + snapshot_stop_is_due( + snapshot, + &stop_key, + current_minute, + &scheduler_last_stop_key, + ) + }); + let stop_due = if active_snapshot.is_some() { + snapshot_stop_due + } else { + same_day_stop_due || overnight_stop_due + }; + if stop_due && last_emit .get("stop") .is_none_or(|instant| instant.elapsed() >= Duration::from_secs(5)) @@ -238,7 +315,9 @@ pub fn spawn_scheduler( "schedule-trigger", serde_json::json!({ "action": "stop", - "key": stop_key + "key": active_snapshot + .and_then(|snapshot| snapshot.stop_key.as_deref()) + .unwrap_or(&stop_key) }), ); last_emit.insert("stop", std::time::Instant::now()); @@ -250,7 +329,28 @@ pub fn spawn_scheduler( #[cfg(test)] mod tests { - use super::{minute_of_day, overnight_stop_is_due, stop_is_due, OvernightStopCheck}; + use super::{ + minute_of_day, overnight_stop_is_due, scheduler_run_snapshot, snapshot_stop_is_due, + stop_is_due, OvernightStopCheck, + }; + use crate::ipc::{PostQueueAction, SchedulerSettings}; + + fn scheduler_settings( + start_time: &str, + stop_enabled: bool, + stop_time: &str, + ) -> SchedulerSettings { + SchedulerSettings { + enabled: true, + start_time: start_time.to_string(), + stop_time_enabled: stop_enabled, + stop_time: stop_time.to_string(), + everyday: false, + selected_days: vec![1], + selected_queue_ids: Vec::new(), + post_queue_action: PostQueueAction::None, + } + } #[test] fn parses_valid_scheduler_times() { @@ -270,6 +370,46 @@ mod tests { assert_eq!(minute_of_day("bad"), None); } + #[test] + fn snapshots_overnight_stop_boundary_for_started_run() { + let snapshot = scheduler_run_snapshot( + "2026-06-22-start", + &scheduler_settings("22:00", true, "06:00"), + ); + + assert_eq!(snapshot.start_key, "2026-06-22-start"); + assert_eq!(snapshot.stop_key.as_deref(), Some("2026-06-23-stop")); + assert_eq!(snapshot.stop_minute, Some(360)); + // Later edits to weekdays or start/stop settings cannot change this + // already-started run's stop boundary. + assert!(snapshot_stop_is_due(&snapshot, "2026-06-23-stop", 420, "")); + assert!(!snapshot_stop_is_due(&snapshot, "2026-06-23-stop", 359, "")); + assert!(!snapshot_stop_is_due( + &snapshot, + "2026-06-23-stop", + 420, + "2026-06-23-stop" + )); + } + + #[test] + fn snapshots_same_day_stop_and_disabled_stop() { + let same_day = scheduler_run_snapshot( + "2026-06-22-start", + &scheduler_settings("08:00", true, "18:30"), + ); + assert_eq!(same_day.stop_key.as_deref(), Some("2026-06-22-stop")); + assert_eq!(same_day.stop_minute, Some(1110)); + + let no_stop = scheduler_run_snapshot( + "2026-06-22-start", + &scheduler_settings("08:00", false, "18:30"), + ); + assert_eq!(no_stop.stop_key, None); + assert_eq!(no_stop.stop_minute, None); + assert!(!snapshot_stop_is_due(&no_stop, "2026-06-22-stop", 1200, "")); + } + #[test] fn stop_requires_same_day_acknowledged_start() { assert!(!stop_is_due( diff --git a/src-tauri/src/settings.rs b/src-tauri/src/settings.rs index 505a9fe..2c75d58 100644 --- a/src-tauri/src/settings.rs +++ b/src-tauri/src/settings.rs @@ -319,6 +319,7 @@ pub fn preserve_scheduler_runtime_keys( "schedulerLastStartKey", "schedulerTriggeredStartKey", "schedulerLastStopKey", + "schedulerActiveRunSnapshot", ] { if let Some(value) = existing_state.get(key) { incoming_state.insert(key.to_string(), value.clone()); @@ -1051,6 +1052,7 @@ fn default_settings() -> PersistedSettings { scheduler_last_start_key: String::new(), scheduler_triggered_start_key: None, scheduler_last_stop_key: String::new(), + scheduler_active_run_snapshot: None, last_custom_speed_limit_ki_b: 1024, last_custom_speed_limit_unit: "MB/s".to_string(), per_server_connections: 16, @@ -1118,7 +1120,12 @@ mod tests { "state": { "schedulerLastStartKey": "2026-06-22-start", "schedulerTriggeredStartKey": "2026-06-22-start", - "schedulerLastStopKey": "2026-06-22-stop" + "schedulerLastStopKey": "2026-06-22-stop", + "schedulerActiveRunSnapshot": { + "startKey": "2026-06-23-start", + "stopKey": "2026-06-24-stop", + "stopMinute": 360 + } }, "version": 3 }) @@ -1142,6 +1149,10 @@ mod tests { "2026-06-22-start" ); assert_eq!(merged["state"]["schedulerLastStopKey"], "2026-06-22-stop"); + assert_eq!( + merged["state"]["schedulerActiveRunSnapshot"]["stopKey"], + "2026-06-24-stop" + ); } #[test] diff --git a/src-tauri/tests/queue_manager.rs b/src-tauri/tests/queue_manager.rs index 159d44a..ebc0fe2 100644 --- a/src-tauri/tests/queue_manager.rs +++ b/src-tauri/tests/queue_manager.rs @@ -1694,6 +1694,73 @@ async fn dispatcher_rotates_eligible_queues_without_starving_later_work() { dispatcher.abort(); } +#[tokio::test] +async fn selected_start_hold_keeps_other_queues_eligible_and_preserves_priority() { + let (manager, _spawner) = make_manager(1); + let manager = Arc::new(manager); + let hold = manager + .begin_queue_dispatch_hold("selected") + .await + .expect("selected queue hold should start"); + manager + .push(aria2_task_in_queue("older", "selected")) + .await + .unwrap(); + manager + .push(aria2_task_in_queue("other-queue", "other")) + .await + .unwrap(); + + let dispatcher = { + let manager = Arc::clone(&manager); + tokio::spawn(async move { manager.run_dispatcher().await }) + }; + timeout(Duration::from_secs(1), async { + while manager.aria2_gid_for_download("other-queue").is_none() { + tokio::task::yield_now().await; + } + }) + .await + .expect("a held queue must not block eligible work in another queue"); + assert!(manager.aria2_gid_for_download("older").is_none()); + + // A becomes a newly selected start while older is still ahead in this + // queue. Keep the queue held through its atomic order update. + manager.release_permit("other-queue").await; + manager + .push(aria2_task_in_queue("selected", "selected")) + .await + .unwrap(); + assert_eq!( + manager + .move_many_in_queue_to(&["selected".to_string()], "selected", 0) + .await, + vec!["selected", "older"] + ); + assert!(manager + .end_queue_dispatch_hold("selected", &hold) + .await); + + timeout(Duration::from_secs(1), async { + while manager.aria2_gid_for_download("selected").is_none() { + tokio::task::yield_now().await; + } + }) + .await + .expect("the selected row should take the next available slot"); + assert!(manager.aria2_gid_for_download("older").is_none()); + + manager.release_permit("selected").await; + timeout(Duration::from_secs(1), async { + while manager.aria2_gid_for_download("older").is_none() { + tokio::task::yield_now().await; + } + }) + .await + .expect("older work should dispatch after the selected row"); + dispatcher.abort(); +} + #[tokio::test] async fn queue_limit_increase_wakes_waiting_work_and_decrease_keeps_active_tasks() { let (manager, spawner) = make_manager(3); @@ -2883,6 +2950,30 @@ async fn multi_move_reorders_selected_items_as_one_atomic_block() { ); } +#[tokio::test] +async fn multi_move_down_moves_sparse_selection_one_position_and_respects_boundary() { + use firelink_lib::ipc::QueueDirection; + + let (mgr, _spawner) = make_manager(3); + for id in ["a", "b", "c", "d", "e"] { + mgr.push(sample_task(id)).await.unwrap(); + } + + let selected = vec!["b".to_string(), "d".to_string()]; + assert_eq!( + mgr.move_many_in_queue(&selected, "main", QueueDirection::Down) + .await, + vec!["a", "c", "b", "d", "e"] + ); + + let bottom_selection = vec!["d".to_string(), "e".to_string()]; + assert_eq!( + mgr.move_many_in_queue(&bottom_selection, "main", QueueDirection::Down) + .await, + vec!["a", "c", "b", "d", "e"] + ); +} + #[tokio::test] async fn direction_move_keeps_the_queue_order_for_an_unsorted_selection() { use firelink_lib::ipc::QueueDirection; diff --git a/src/App.tsx b/src/App.tsx index 1e149f3..ad4f422 100644 --- a/src/App.tsx +++ b/src/App.tsx @@ -1396,6 +1396,25 @@ function App() { schedulerActiveDownloadIds ]); + useEffect(() => { + if (!coreReady || frontendExitPreparing || schedulerRunning) return; + if (schedulerActiveDownloadIds.length === 0) return; + const activeIds = new Set(downloads + .filter(download => isActiveDownloadStatus(download.status)) + .map(download => download.id)); + const remainingIds = schedulerActiveDownloadIds.filter(id => activeIds.has(id)); + if (remainingIds.length === schedulerActiveDownloadIds.length) return; + void useSettingsStore.getState().persistSchedulerTracking(remainingIds, false).catch(error => { + console.error('Failed to clear inactive scheduler tracking:', error); + }); + }, [ + coreReady, + downloads, + frontendExitPreparing, + schedulerActiveDownloadIds, + schedulerRunning + ]); + useEffect(() => { const initNotifications = async () => { if (!useSettingsStore.getState().showNotifications) return; diff --git a/src/components/SchedulerView.tsx b/src/components/SchedulerView.tsx index f5da86e..fc3f07f 100644 --- a/src/components/SchedulerView.tsx +++ b/src/components/SchedulerView.tsx @@ -237,14 +237,37 @@ export default function SchedulerView() { const counts = await Promise.all( targetQueueList.map(queueId => useDownloadStore.getState().pauseQueue(queueId)) ); - const directPauseResults = await Promise.allSettled( + await Promise.allSettled( trackedIdsOutsideQueues.map(id => useDownloadStore.getState().pauseDownload(id)) ); if (!isSchedulerControlCurrent(generation)) return; - const count = counts.reduce((total, queueCount) => total + queueCount, 0) - + directPauseResults.filter(result => result.status === 'fulfilled').length; - useSettingsStore.getState().setSchedulerRunning(false); - useSettingsStore.getState().setSchedulerActiveDownloadIds([]); + const downloadsAfterPause = useDownloadStore.getState().downloads; + const directPausedCount = trackedIdsOutsideQueues.filter(id => { + const before = downloads.find(download => download.id === id); + const after = downloadsAfterPause.find(download => download.id === id); + return before && isActiveDownloadStatus(before.status) && after?.status === 'paused'; + }).length; + const stillActiveTrackedIds = trackedDownloadIds.filter(id => { + const download = downloadsAfterPause.find(item => item.id === id); + return download && isActiveDownloadStatus(download.status); + }); + const remainingTargetCount = downloadsAfterPause.filter(download => { + const queueId = download.queueId || MAIN_QUEUE_ID; + return (targetQueueSet.has(queueId) || trackedDownloadIds.includes(download.id)) + && isActiveDownloadStatus(download.status); + }).length; + await useSettingsStore.getState().persistSchedulerTracking(stillActiveTrackedIds, false); + const count = counts.reduce((total, queueCount) => total + queueCount, 0) + directPausedCount; + if (remainingTargetCount > 0) { + addToast({ + message: remainingTargetCount === 1 + ? t($ => $.app.schedulerPauseOneFailed) + : t($ => $.app.schedulerPauseManyFailed, { count: remainingTargetCount }), + variant: 'error', + isActionable: true + }); + return; + } addToast({ message: count === 0 ? t($ => $.scheduler.noActiveDownloads) diff --git a/src/ipc.ts b/src/ipc.ts index f959c11..de8dc75 100644 --- a/src/ipc.ts +++ b/src/ipc.ts @@ -84,7 +84,8 @@ type CommandMap = { get_deno_engine_status: { args: undefined; result: EngineStatusItem }; reveal_in_file_manager: { args: { path: string }; result: void }; open_downloaded_file: { args: { path: string }; result: void }; - pause_download: { args: { id: string }; result: void }; + /** True when the pause won; false when a terminal/inactive lifecycle won. */ + pause_download: { args: { id: string }; result: boolean }; resume_download: { args: { id: string; queueId: string }; result: boolean }; submit_download_removals: { args: { ids: string[]; deleteAssets: boolean }; result: void }; list_download_removals: { args: undefined; result: DownloadRemovalJob[] }; @@ -224,6 +225,9 @@ type CommandMap = { enqueue_many: { args: { items: EnqueueItem[] }; result: import('./bindings/EnqueueResult').EnqueueResult[] }; move_in_queue: { args: { id: string; queueId: string; direction: 'up' | 'down' }; result: string[] }; move_many_in_queue: { args: { ids: string[]; queueId: string; direction: 'up' | 'down'; targetIndex?: number }; result: string[] }; + begin_queue_dispatch_hold: { args: { queueId: string }; result: string }; + renew_queue_dispatch_hold: { args: { queueId: string; token: string }; result: boolean }; + end_queue_dispatch_hold: { args: { queueId: string; token: string }; result: boolean }; remove_from_queue: { args: { id: string }; result: boolean }; open_download_properties_window: { args: { id: string }; result: string }; get_properties_window_download_id: { args: undefined; result: string }; diff --git a/src/store/useDownloadStore.test.ts b/src/store/useDownloadStore.test.ts index 4c55056..df20e10 100644 --- a/src/store/useDownloadStore.test.ts +++ b/src/store/useDownloadStore.test.ts @@ -1972,6 +1972,9 @@ describe('useDownloadStore', () => { }); vi.mocked(ipc.invokeCommand).mockImplementation(async (command: string, args?: unknown) => { + if (command === 'begin_queue_dispatch_hold') return `hold-${(args as { queueId: string }).queueId}`; + if (command === 'renew_queue_dispatch_hold') return true; + if (command === 'end_queue_dispatch_hold') return true; if (command === 'enqueue_download') { const id = (args as { item: { id: string } }).item.id; return { id, filename: id }; @@ -1987,6 +1990,12 @@ describe('useDownloadStore', () => { .filter(([command]) => command === 'enqueue_download') .map(([, args]) => (args as { item: { id: string } }).item.id); expect(enqueueIds).toEqual(['selected-b', 'selected-a']); + const beginHoldCall = vi.mocked(ipc.invokeCommand).mock.calls + .findIndex(([command]) => command === 'begin_queue_dispatch_hold'); + const enqueueCall = vi.mocked(ipc.invokeCommand).mock.calls + .findIndex(([command]) => command === 'enqueue_download'); + expect(vi.mocked(ipc.invokeCommand).mock.invocationCallOrder[beginHoldCall]) + .toBeLessThan(vi.mocked(ipc.invokeCommand).mock.invocationCallOrder[enqueueCall]); expect(ipc.invokeCommand).toHaveBeenCalledWith('move_many_in_queue', { ids: ['selected-b', 'selected-a'], queueId: 'selection-queue', @@ -2011,6 +2020,9 @@ describe('useDownloadStore', () => { }); vi.mocked(ipc.invokeCommand).mockImplementation(async (command: string, args?: unknown) => { + if (command === 'begin_queue_dispatch_hold') return `hold-${(args as { queueId: string }).queueId}`; + if (command === 'renew_queue_dispatch_hold') return true; + if (command === 'end_queue_dispatch_hold') return true; if (command === 'enqueue_download') { const id = (args as { item: { id: string } }).item.id; return { id, filename: id }; @@ -2062,6 +2074,9 @@ describe('useDownloadStore', () => { }); vi.mocked(ipc.invokeCommand).mockImplementation(async (command: string, args?: unknown) => { + if (command === 'begin_queue_dispatch_hold') return `hold-${(args as { queueId: string }).queueId}`; + if (command === 'renew_queue_dispatch_hold') return true; + if (command === 'end_queue_dispatch_hold') return true; if (command === 'enqueue_download') { const item = (args as { item: { id: string; password: string | null } }).item; return { id: item.id, filename: item.id }; @@ -2091,6 +2106,28 @@ describe('useDownloadStore', () => { ]); }); + it('releases the selected queue hold when re-enqueue fails', async () => { + useDownloadStore.setState({ + downloads: [ + { id: 'selected-failure', url: 'http://failure', fileName: 'failure', destination: '/tmp', status: 'paused', category: 'Other', dateAdded: '', queueId: 'selected-failure-queue' }, + ] as any[], + }); + vi.mocked(ipc.invokeCommand).mockImplementation(async (command: string) => { + if (command === 'begin_queue_dispatch_hold') return 'failure-hold'; + if (command === 'renew_queue_dispatch_hold') return true; + if (command === 'end_queue_dispatch_hold') return true; + if (command === 'enqueue_download') throw new Error('queue persistence unavailable'); + if (command === 'get_pending_order') return []; + return undefined; + }); + + await expect(useDownloadStore.getState().startSelected(['selected-failure'])).resolves.toBe(0); + expect(ipc.invokeCommand).toHaveBeenCalledWith('end_queue_dispatch_hold', { + queueId: 'selected-failure-queue', + token: 'failure-hold', + }); + }); + it('pauses queued items through the global pause action', async () => { useDownloadStore.setState({ downloads: [ @@ -2098,8 +2135,9 @@ describe('useDownloadStore', () => { { id: 'queued-two', url: 'http://two', fileName: 'two', status: 'queued', category: 'Other', dateAdded: '', queueId: 'pause-queue' }, { id: 'staged-one', url: 'http://staged', fileName: 'staged', status: 'staged', category: 'Other', dateAdded: '', queueId: 'pause-queue' }, ] as any[], + backendRegisteredIds: new Set(['queued-one', 'queued-two', 'staged-one']), }); - vi.mocked(ipc.invokeCommand).mockResolvedValue(undefined as never); + vi.mocked(ipc.invokeCommand).mockResolvedValue(true as never); await expect(useDownloadStore.getState().pauseAll()).resolves.toBe(3); expect( @@ -2108,6 +2146,30 @@ describe('useDownloadStore', () => { expect(useDownloadStore.getState().downloads.every(item => item.status === 'paused')).toBe(true); }); + it('locally pauses an unregistered staged row when native owns no lifecycle', async () => { + useDownloadStore.setState({ + downloads: [ + { id: 'unregistered-staged', status: 'staged', queueId: 'local-pause', speed: '1 MiB/s', eta: '2s' } + ] as any[], + backendRegisteredIds: new Set(), + pendingOrder: [], + }); + vi.mocked(ipc.invokeCommand).mockImplementation(async command => { + if (command === 'pause_download') return false as never; + return undefined as never; + }); + + await useDownloadStore.getState().pauseDownload('unregistered-staged'); + + expect(useDownloadStore.getState().downloads[0]).toMatchObject({ + status: 'paused', + speed: '-', + eta: '-', + }); + expect(useDownloadStore.getState().pendingOrder).not.toContain('unregistered-staged'); + expect(ipc.invokeCommand).toHaveBeenCalledWith('pause_download', { id: 'unregistered-staged' }); + }); + it('moves a paused row behind the remaining runnable queue rows', async () => { useDownloadStore.setState({ downloads: [ @@ -2116,9 +2178,10 @@ describe('useDownloadStore', () => { { id: 'pause-target', status: 'queued', queueId: 'ordered-pause', queuePosition: 2 }, { id: 'queued-two', status: 'queued', queueId: 'ordered-pause', queuePosition: 3 } ] as any[], + backendRegisteredIds: new Set(['active', 'queued-one', 'pause-target', 'queued-two']), pendingOrder: ['queued-one', 'pause-target', 'queued-two'] }); - vi.mocked(ipc.invokeCommand).mockResolvedValue(undefined as never); + vi.mocked(ipc.invokeCommand).mockResolvedValue(true as never); await useDownloadStore.getState().pauseDownload('pause-target'); @@ -2147,6 +2210,9 @@ describe('useDownloadStore', () => { releaseEnqueue = resolve; }); vi.mocked(ipc.invokeCommand).mockImplementation(async (command: string) => { + if (command === 'begin_queue_dispatch_hold') return 'hold-race-selected'; + if (command === 'renew_queue_dispatch_hold') return true; + if (command === 'end_queue_dispatch_hold') return true; if (command === 'enqueue_download') return enqueue; if (command === 'get_pending_order') return []; return undefined; @@ -2177,6 +2243,10 @@ describe('useDownloadStore', () => { expect(useDownloadStore.getState().downloads.map(item => item.status)).toEqual(['paused', 'paused']); expect(vi.mocked(ipc.invokeCommand).mock.calls.filter(([command]) => command === 'pause_download')) .toHaveLength(1); + expect(ipc.invokeCommand).toHaveBeenCalledWith('end_queue_dispatch_hold', { + queueId: 'race-selected', + token: 'hold-race-selected', + }); }); it('cleans an accepted backend enqueue when queue reconciliation fails', async () => { @@ -4167,6 +4237,32 @@ describe('useDownloadStore', () => { ).toHaveLength(4); }); + it('does not turn a completed download into paused when completion wins a queue pause race', async () => { + useDownloadStore.setState({ + downloads: [{ + id: 'pause-completion-race', + url: 'https://example.com/file', + fileName: 'file.bin', + status: 'downloading', + category: 'Other', + dateAdded: '', + queueId: 'queue-race' + }] as any[] + }); + vi.mocked(ipc.invokeCommand).mockImplementation(async command => { + if (command === 'pause_download') { + // Native completion reconciliation happened before pause_download + // returned its authoritative "pause did not win" result. + useDownloadStore.getState().updateDownload('pause-completion-race', { status: 'completed' }); + return false as never; + } + return undefined as never; + }); + + await expect(useDownloadStore.getState().pauseQueue('queue-race')).resolves.toBe(0); + expect(useDownloadStore.getState().downloads[0].status).toBe('completed'); + }); + it('assigns selected unfinished downloads to a queue without moving completed items', async () => { useDownloadStore.setState({ downloads: [ @@ -4265,7 +4361,7 @@ describe('useDownloadStore', () => { { id: 'two', status: 'queued', queueId: 'move-queue', queuePosition: 2 }, { id: 'three', status: 'queued', queueId: 'move-queue', queuePosition: 3 } ] as any[], - backendRegisteredIds: new Set(['three']), + backendRegisteredIds: new Set(['one', 'two', 'three']), pendingOrder: ['one', 'two', 'three'] }); vi.mocked(ipc.invokeCommand).mockImplementation(async (command: string) => { @@ -4279,7 +4375,8 @@ describe('useDownloadStore', () => { expect(vi.mocked(ipc.invokeCommand)).toHaveBeenCalledWith('move_many_in_queue', { ids: ['three'], queueId: 'move-queue', - direction: 'up' + direction: 'up', + targetIndex: 1, }); expect(vi.mocked(ipc.invokeCommand)).not.toHaveBeenCalledWith('move_in_queue', expect.anything()); const positions = useDownloadStore.getState().downloads @@ -4288,6 +4385,68 @@ describe('useDownloadStore', () => { expect(new Set(positions).size).toBe(4); }); + it('moves a sparse selection down by one row as a block', async () => { + useDownloadStore.setState({ + downloads: [ + { id: 'a', status: 'queued', queueId: 'sparse-down', queuePosition: 0 }, + { id: 'b', status: 'queued', queueId: 'sparse-down', queuePosition: 1 }, + { id: 'c', status: 'queued', queueId: 'sparse-down', queuePosition: 2 }, + { id: 'd', status: 'queued', queueId: 'sparse-down', queuePosition: 3 }, + { id: 'e', status: 'queued', queueId: 'sparse-down', queuePosition: 4 }, + ] as any[], + backendRegisteredIds: new Set(['a', 'b', 'c', 'd', 'e']), + pendingOrder: ['a', 'b', 'c', 'd', 'e'], + }); + vi.mocked(ipc.invokeCommand).mockImplementation(async (command: string) => { + if (command === 'move_many_in_queue') return ['a', 'c', 'b', 'd', 'e']; + if (command === 'get_pending_order') return ['a', 'c', 'b', 'd', 'e']; + return undefined; + }); + + await useDownloadStore.getState().moveInQueue(['b', 'd'], 'down'); + + expect(useDownloadStore.getState().downloads + .filter(item => item.queueId === 'sparse-down') + .sort((left, right) => (left.queuePosition ?? 0) - (right.queuePosition ?? 0)) + .map(item => item.id)).toEqual(['a', 'c', 'b', 'd', 'e']); + expect(ipc.invokeCommand).toHaveBeenCalledWith('move_many_in_queue', { + ids: ['b', 'd'], + queueId: 'sparse-down', + direction: 'up', + targetIndex: 2, + }); + }); + + it('translates keyboard moves around staged rows before the atomic backend move', async () => { + useDownloadStore.setState({ + downloads: [ + { id: 'a', status: 'queued', queueId: 'keyboard-staged', queuePosition: 0 }, + { id: 'staged', status: 'staged', queueId: 'keyboard-staged', queuePosition: 1 }, + { id: 'b', status: 'queued', queueId: 'keyboard-staged', queuePosition: 2 }, + ] as any[], + backendRegisteredIds: new Set(['a', 'b']), + pendingOrder: ['a', 'b'], + }); + vi.mocked(ipc.invokeCommand).mockImplementation(async (command: string) => { + if (command === 'move_many_in_queue') return ['b', 'a']; + if (command === 'get_pending_order') return ['b', 'a']; + return undefined; + }); + + await useDownloadStore.getState().moveInQueue('b', 'up'); + + expect(ipc.invokeCommand).toHaveBeenCalledWith('move_many_in_queue', { + ids: ['b'], + queueId: 'keyboard-staged', + direction: 'up', + targetIndex: 1, + }); + expect(useDownloadStore.getState().downloads + .filter(item => item.queueId === 'keyboard-staged') + .sort((left, right) => (left.queuePosition ?? 0) - (right.queuePosition ?? 0)) + .map(item => item.id)).toEqual(['a', 'b', 'staged']); + }); + it('rolls back and rejects a failed keyboard or header queue move', async () => { useDownloadStore.setState({ downloads: [ diff --git a/src/store/useDownloadStore.ts b/src/store/useDownloadStore.ts index 46a13b4..b275aee 100644 --- a/src/store/useDownloadStore.ts +++ b/src/store/useDownloadStore.ts @@ -274,6 +274,8 @@ const queuePositionComparator = (left: DownloadItem, right: DownloadItem): numbe comparableQueuePosition(left) - comparableQueuePosition(right) || left.id.localeCompare(right.id); +const QUEUE_START_BATCH_HOLD_REFRESH_MS = 20_000; + const queueItemsForReordering = (downloads: DownloadItem[], queueId: string): DownloadItem[] => downloads .filter(download => @@ -1708,14 +1710,13 @@ export const useDownloadStore = create((set, get) => { ].map(item => [item.id, item.queuePosition])); const unselectedItems = queueItems.filter(item => !ids.includes(item.id)); const selectedIndices = selectedItems.map(item => queueItems.findIndex(d => d.id === item.id)); + const firstSelectedIndex = Math.min(...selectedIndices); let insertIndex = 0; if (direction === 'up') { - const firstSelectedIndex = Math.min(...selectedIndices); insertIndex = Math.max(0, firstSelectedIndex - 1); } else { - const lastSelectedIndex = Math.max(...selectedIndices); - insertIndex = Math.min(unselectedItems.length, lastSelectedIndex - selectedItems.length + 2); + insertIndex = Math.min(unselectedItems.length, firstSelectedIndex + 1); } const reordered = [ @@ -1730,11 +1731,25 @@ export const useDownloadStore = create((set, get) => { .map(item => item.id); if (registeredIdsToMove.length === 0) return; + // The local queue includes staged rows, but the native pending queue + // contains registered rows only. Translate the desired local order to + // its registered-only insertion point so keyboard and header moves + // cannot leave the UI and dispatcher with different orders. + const registeredItems = queueItems.filter(item => get().backendRegisteredIds.has(item.id)); + const registeredSelectedIds = new Set(registeredIdsToMove); + const registeredDesiredOrder = reordered.filter(item => get().backendRegisteredIds.has(item.id)); + const backendTargetIndex = targetIndexForDesiredOrder( + registeredItems, + registeredSelectedIds, + registeredDesiredOrder + ); + try { const order = await invoke('move_many_in_queue', { ids: registeredIdsToMove, queueId, - direction + direction: 'up', + targetIndex: backendTargetIndex }) as string[]; if (Array.isArray(order)) { const globalOrder = await invoke('get_pending_order', { queueId: null }) @@ -2489,9 +2504,30 @@ export const useDownloadStore = create((set, get) => { await pendingDispatch; } - await invoke('pause_download', { id }); + const pauseApplied = await invoke('pause_download', { id }); if (!isCurrentDownloadLifecycle(id, generation)) return; + // Completion can win after the UI snapshots an active row but before + // Aria2 processes forcePause. The native command reconciles that + // terminal state and reports false so a stale frontend snapshot cannot + // overwrite Completed with Paused in memory or durable storage. A row + // that is still queued/staged but has no native ownership is different: + // it never reached the backend queue, so preserve the user's pause + // locally instead of leaving an unstartable-looking row staged. + if (pauseApplied === false) { + const current = get().downloads.find(download => download.id === id); + const hasNativeOwnership = get().backendRegisteredIds.has(id) || + get().pendingOrder.includes(id); + if ( + current && + (current.status === 'queued' || current.status === 'staged') && + !hasNativeOwnership + ) { + get().updateDownload(id, { status: 'paused', speed: '-', eta: '-' }); + await commitDownloadState(); + } + return; + } const current = get().downloads.find(download => download.id === id); if (current && current.status !== 'completed' && current.status !== 'failed') { get().updateDownload(id, { status: 'paused', speed: '-', eta: '-' }); @@ -2584,71 +2620,118 @@ export const useDownloadStore = create((set, get) => { ] as const) ); - // Make the selection's order explicit before any capacity becomes - // available. This keeps the frontend projection and backend pending - // queue aligned even when the selected rows were not adjacent. - for (const [queueId, queueIds] of selectedByQueue) { - const generation = selectedQueueGenerations.get(queueId); - if (generation === undefined || !isCurrentQueueControlGeneration(queueId, generation)) { - continue; - } - await get().moveManyInQueueToPosition(queueIds, queueId, 0, null, queueIds); - } + const dispatchHolds = new Map(); + const expiredDispatchHolds = new Set(); + let dispatchHoldRefresh: ReturnType | undefined; - let startedCount = 0; - const startedByQueue = new Map(); - for (const id of orderedIds) { - const current = get().downloads.find(download => download.id === id); - if (!current || !canStartDownload(current.status)) continue; - const queueId = current.queueId || MAIN_QUEUE_ID; - const generation = selectedQueueGenerations.get(queueId); - if (generation === undefined || !isCurrentQueueControlGeneration(queueId, generation)) { - continue; + const ensureDispatchHold = async (queueId: string): Promise => { + const token = dispatchHolds.get(queueId); + if (!token || expiredDispatchHolds.has(queueId)) { + throw new Error('Selected queue start lost its dispatch-order lease'); } - const resumed = await resumeDownloadInternal(id, { - preserveQueuePosition: true, - forceRequeue: true - }); - if (resumed && !isCurrentQueueControlGeneration(queueId, generation)) { - // A queue pause can win while this item's requeue is in flight. The - // pause action is allowed to preempt this bulk-start operation so - // this item cannot remain running after the user's pause request. - await get().pauseDownload(id); - continue; + const renewed = await invoke('renew_queue_dispatch_hold', { queueId, token }) as boolean; + if (!renewed) { + expiredDispatchHolds.add(queueId); + throw new Error('Selected queue start lost its dispatch-order lease'); } - if (resumed) { - startedCount += 1; - const startedIds = startedByQueue.get(queueId) || []; - startedIds.push(id); - startedByQueue.set(queueId, startedIds); + }; - // A requeued paused item is appended by the backend. Move the - // started prefix immediately so a newly available slot cannot let - // older pending work overtake the remaining selection while it is - // still being requeued. - if (isCurrentQueueControlGeneration(queueId, generation)) { - await get().moveManyInQueueToPosition( - startedIds, - queueId, - 0, - null, - startedIds - ); + try { + // Hold each affected queue before changing or re-enqueueing any + // selected row. The backend lease is queue-scoped, so unrelated + // queues remain eligible while this selection is assembled. + for (const queueId of selectedByQueue.keys()) { + const token = await invoke('begin_queue_dispatch_hold', { queueId }) as string; + if (!token) throw new Error('Could not reserve selected queue order'); + dispatchHolds.set(queueId, token); + } + + dispatchHoldRefresh = setInterval(() => { + for (const [queueId, token] of dispatchHolds) { + void invoke('renew_queue_dispatch_hold', { queueId, token }) + .then(active => { + if (!active) expiredDispatchHolds.add(queueId); + }) + .catch(error => { + console.warn('Could not refresh selected queue dispatch lease:', error); + }); + } + }, QUEUE_START_BATCH_HOLD_REFRESH_MS); + + // Make the selection's order explicit before any capacity becomes + // available. This keeps the frontend projection and backend pending + // queue aligned even when the selected rows were not adjacent. + for (const [queueId, queueIds] of selectedByQueue) { + const generation = selectedQueueGenerations.get(queueId); + if (generation === undefined || !isCurrentQueueControlGeneration(queueId, generation)) { + continue; + } + await get().moveManyInQueueToPosition(queueIds, queueId, 0, null, queueIds); + } + + let startedCount = 0; + const startedByQueue = new Map(); + for (const id of orderedIds) { + const current = get().downloads.find(download => download.id === id); + if (!current || !canStartDownload(current.status)) continue; + const queueId = current.queueId || MAIN_QUEUE_ID; + const generation = selectedQueueGenerations.get(queueId); + if (generation === undefined || !isCurrentQueueControlGeneration(queueId, generation)) { + continue; + } + await ensureDispatchHold(queueId); + const resumed = await resumeDownloadInternal(id, { + preserveQueuePosition: true, + forceRequeue: true + }); + if (resumed && !isCurrentQueueControlGeneration(queueId, generation)) { + // A queue pause can win while this item's requeue is in flight. The + // pause action is allowed to preempt this bulk-start operation so + // this item cannot remain running after the user's pause request. + await get().pauseDownload(id); + continue; + } + if (resumed) { + startedCount += 1; + const startedIds = startedByQueue.get(queueId) || []; + startedIds.push(id); + startedByQueue.set(queueId, startedIds); + + // The queue hold prevents a newly enqueued item or older pending + // work from consuming capacity before this block is fully ordered. + if (isCurrentQueueControlGeneration(queueId, generation)) { + await get().moveManyInQueueToPosition( + startedIds, + queueId, + 0, + null, + startedIds + ); + } } } - } - // Newly dispatched rows are now visible to the backend pending list. - // Reapply the same ordered block so existing pending work cannot remain - // ahead of a user-selected start request. - for (const [queueId, queueIds] of selectedByQueue) { - const generation = selectedQueueGenerations.get(queueId); - if (generation === undefined || !isCurrentQueueControlGeneration(queueId, generation)) { - continue; + // Newly dispatched rows are now visible to the backend pending list. + // Reapply the same ordered block so existing pending work cannot remain + // ahead of a user-selected start request. + for (const [queueId, queueIds] of selectedByQueue) { + const generation = selectedQueueGenerations.get(queueId); + if (generation === undefined || !isCurrentQueueControlGeneration(queueId, generation)) { + continue; + } + await get().moveManyInQueueToPosition(queueIds, queueId, 0, null, queueIds); } - await get().moveManyInQueueToPosition(queueIds, queueId, 0, null, queueIds); + return startedCount; + } finally { + if (dispatchHoldRefresh !== undefined) clearInterval(dispatchHoldRefresh); + await Promise.all(Array.from(dispatchHolds, async ([queueId, token]) => { + try { + await invoke('end_queue_dispatch_hold', { queueId, token }); + } catch (error) { + console.error('Failed to release selected queue dispatch lease:', error); + } + })); } - return startedCount; }); }, startQueue: (queueId) => { @@ -2802,8 +2885,10 @@ export const useDownloadStore = create((set, get) => { if (activeIds.length === 0) return 0; const results = await Promise.allSettled(activeIds.map(id => get().pauseDownload(id))); - const pausedCount = results.filter(result => result.status === 'fulfilled').length; - const failedCount = activeIds.length - pausedCount; + const pausedCount = activeIds.filter(id => + get().downloads.some(download => download.id === id && download.status === 'paused') + ).length; + const failedCount = results.filter(result => result.status === 'rejected').length; if (failedCount > 0) { console.error(`Failed to pause ${failedCount} downloads in queue ${queueId}`); } @@ -2834,8 +2919,10 @@ export const useDownloadStore = create((set, get) => { .map(item => item.id); if (activeIds.length === 0) return 0; - const results = await Promise.allSettled(activeIds.map(id => get().pauseDownload(id))); - const pausedCount = results.filter(result => result.status === 'fulfilled').length; + await Promise.allSettled(activeIds.map(id => get().pauseDownload(id))); + const pausedCount = activeIds.filter(id => + get().downloads.some(download => download.id === id && download.status === 'paused') + ).length; syncSystemIntegrations(); return pausedCount; }, diff --git a/src/store/useSettingsStore.test.ts b/src/store/useSettingsStore.test.ts index bed374c..206f8bc 100644 --- a/src/store/useSettingsStore.test.ts +++ b/src/store/useSettingsStore.test.ts @@ -1106,4 +1106,25 @@ describe('scheduled trigger durability', () => { { schedulerRunning: true, schedulerActiveDownloadIds: ['download-1'] } ]); }); + + it('keeps pause-failed downloads tracked without treating a manual pause as a running schedule', async () => { + let savedState: { schedulerRunning: boolean; schedulerActiveDownloadIds: string[] } | undefined; + vi.mocked(ipc.invokeCommand).mockImplementation(async (command: string, args?: unknown) => { + if (command === 'db_save_settings') { + savedState = JSON.parse((args as { data: string }).data).state; + } + return undefined as never; + }); + + await useSettingsStore.getState().persistSchedulerTracking(['still-active'], false); + + expect(useSettingsStore.getState()).toMatchObject({ + schedulerRunning: false, + schedulerActiveDownloadIds: ['still-active'] + }); + expect(savedState).toMatchObject({ + schedulerRunning: false, + schedulerActiveDownloadIds: ['still-active'] + }); + }); }); diff --git a/src/store/useSettingsStore.ts b/src/store/useSettingsStore.ts index 1a33fc9..33aee0c 100644 --- a/src/store/useSettingsStore.ts +++ b/src/store/useSettingsStore.ts @@ -414,7 +414,7 @@ export interface SettingsState { setScheduler: (settings: SchedulerSettings) => void; setSchedulerRunning: (running: boolean) => void; setSchedulerActiveDownloadIds: (ids: string[]) => void; - persistSchedulerTracking: (ids: string[]) => Promise; + persistSchedulerTracking: (ids: string[], running?: boolean) => Promise; setSchedulerLastStartKey: (key: string) => void; setSchedulerLastStopKey: (key: string) => void; setLastCustomSpeedLimitKiB: (limit: number) => void; @@ -699,13 +699,13 @@ export const useSettingsStore = create()( setScheduler: (scheduler) => set({ scheduler }), setSchedulerRunning: (schedulerRunning) => set({ schedulerRunning }), setSchedulerActiveDownloadIds: (schedulerActiveDownloadIds) => set({ schedulerActiveDownloadIds }), - persistSchedulerTracking: (ids) => { + persistSchedulerTracking: (ids, running) => { const schedulerActiveDownloadIds = [...new Set( sanitizeSchedulerActiveDownloadIds(ids, []) )]; set({ schedulerActiveDownloadIds, - schedulerRunning: schedulerActiveDownloadIds.length > 0 + schedulerRunning: running ?? schedulerActiveDownloadIds.length > 0 }); // `waitForSettingsPersistence` intentionally observes only the queue's // rejection-swallowing tail. Scheduler ACKs require confirmation from diff --git a/src/utils/downloads.test.ts b/src/utils/downloads.test.ts index 7c8d905..5397ff1 100644 --- a/src/utils/downloads.test.ts +++ b/src/utils/downloads.test.ts @@ -239,7 +239,7 @@ describe('credential-bearing extension header names', () => { describe('allocation phase visibility', () => { it('does not override paused or completed statuses', () => { expect(isAllocationPhaseVisible(true, 'ready')).toBe(true); - expect(isAllocationPhaseVisible(true, 'failed')).toBe(false); + expect(isAllocationPhaseVisible(true, 'failed')).toBe(true); expect(isAllocationPhaseVisible(true, 'verifying')).toBe(false); expect(isAllocationPhaseVisible(true, 'seeding')).toBe(false); expect(isAllocationPhaseVisible(true, 'waitingToSeed')).toBe(false); diff --git a/src/utils/downloads.ts b/src/utils/downloads.ts index af557be..64a63fc 100644 --- a/src/utils/downloads.ts +++ b/src/utils/downloads.ts @@ -73,7 +73,7 @@ export const isTransferActiveStatus = (status: DownloadStatus): boolean => export const isAllocationPhaseVisible = ( allocationPending: boolean, status: DownloadStatus, -): boolean => allocationPending && ['ready', 'staged', 'queued', 'downloading', 'retrying'].includes(status); +): boolean => allocationPending && ['ready', 'staged', 'queued', 'downloading', 'retrying', 'failed'].includes(status); /** * Allocation is a transient admission phase. Normal downloads retain the