diff --git a/src-tauri/src/ipc.rs b/src-tauri/src/ipc.rs index c3c9d3a..7ad52af 100644 --- a/src-tauri/src/ipc.rs +++ b/src-tauri/src/ipc.rs @@ -110,6 +110,8 @@ pub struct DownloadItem { pub has_been_dispatched: Option, #[ts(optional)] pub last_error: Option, + #[ts(optional)] + pub last_try: Option, } #[derive(Clone, Debug, Serialize, Deserialize, TS)] diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index b96b7f2..24256cf 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -3746,6 +3746,80 @@ async fn aria2_download_status(port: u16, secret: &str, gid: &str) -> Result(); + let port = state.aria2_port.load(std::sync::atomic::Ordering::Relaxed); + let secret = state.aria2_secret.clone(); + let mappings = state.queue_manager.aria2_gid_mappings(); + + for (gid, id) in mappings { + let status = match rpc_call( + port, + &secret, + "aria2.tellStatus", + serde_json::json!([gid, ["status", "errorCode", "errorMessage"]]), + ) + .await + { + Ok(status) => status, + Err(error) => { + log::debug!( + "aria2 reconnect reconciliation [{}]: could not query gid {}: {}", + id, + gid, + error + ); + continue; + } + }; + + let status_name = status.get("status").and_then(|value| value.as_str()); + let outcome = match status_name { + Some("complete") => Some(crate::queue::PendingOutcome::Complete), + Some("error") | Some("removed") => { + let error_code = status + .get("errorCode") + .and_then(|value| value.as_str()) + .filter(|value| !value.is_empty()); + let error_message = status + .get("errorMessage") + .and_then(|value| value.as_str()) + .filter(|value| !value.is_empty()) + .unwrap_or("aria2 download ended while the event channel was disconnected"); + Some(crate::queue::PendingOutcome::Error(match error_code { + Some(code) => format!("aria2 error code {code}: {error_message}"), + None => error_message.to_string(), + })) + } + _ => None, + }; + + if let Some(outcome) = outcome { + state.queue_manager.handle_aria2_event(&gid, outcome).await; + } + } +} + +async fn aria2_daemon_is_reachable(port: u16, secret: &str) -> bool { + for attempt in 0..2 { + if rpc_call( + port, + secret, + "aria2.getVersion", + serde_json::json!([]), + ) + .await + .is_ok() + { + return true; + } + if attempt == 0 { + tokio::time::sleep(Duration::from_millis(250)).await; + } + } + false +} + fn aria2_gid_not_found(error: &str) -> bool { let lower = error.to_ascii_lowercase(); lower.contains("gid") && lower.contains("not found") @@ -4713,6 +4787,7 @@ async fn clear_logs(app_handle: tauri::AppHandle) -> Result<(), String> { async fn export_logs( app_handle: tauri::AppHandle, state: tauri::State<'_, AppState>, + destination: Option, ) -> Result { let mut output = format!( "Firelink support logs\nVersion: {}\nOS: {} {}\nArchitecture: {}\nGenerated: {}\n\n", @@ -4773,6 +4848,16 @@ async fn export_logs( } output.push('\n'); } + + if let Some(destination) = destination { + let destination = std::path::PathBuf::from(destination.trim()); + if destination.file_name().is_none() { + return Err("log export destination must be a file path".to_string()); + } + tokio::fs::write(&destination, &output) + .await + .map_err(|error| format!("failed to write exported logs to '{}': {error}", destination.display()))?; + } Ok(output) } @@ -5959,16 +6044,31 @@ pub fn run() { let mut ws_retries = 0; loop { - if ws_retries > 10 { - log::error!("Max WebSocket reconnection attempts reached. aria2 integration is disabled."); + if ws_retries == 10 { + log::error!("Aria2 WebSocket is still unavailable after repeated reconnection attempts; continuing to retry."); let guard = app_handle_bg.state::(); *guard.startup_error.lock().unwrap() = Some("Max WebSocket reconnection attempts reached.".to_string()); - break; + // Keep retrying. The daemon or local WebSocket can + // recover after a prolonged outage; permanently + // stopping this task strands active transfers. + ws_retries = 0; } tokio::time::sleep(std::time::Duration::from_secs(2)).await; let ws_url = format!("ws://127.0.0.1:{}/jsonrpc", ws_port); if let Ok((ws_stream, _)) = tokio_tungstenite::connect_async(&ws_url).await { ws_retries = 0; // reset on success + if let Ok(mut startup_error) = app_handle_bg + .state::() + .startup_error + .lock() + { + if startup_error.as_deref() + == Some("Max WebSocket reconnection attempts reached.") + { + *startup_error = None; + } + } + reconcile_aria2_downloads(&app_handle_bg).await; use futures_util::StreamExt; let (_, mut read) = ws_stream.split(); while let Some(msg) = read.next().await { @@ -6025,7 +6125,11 @@ pub fn run() { } } ws_retries += 1; - app_handle_bg.state::().queue_manager.clear_aria2_permits().await; + let state = app_handle_bg.state::(); + let port = state.aria2_port.load(std::sync::atomic::Ordering::Relaxed); + if !aria2_daemon_is_reachable(port, &state.aria2_secret).await { + state.queue_manager.clear_aria2_permits().await; + } tokio::time::sleep(std::time::Duration::from_secs(4)).await; } }); @@ -6035,19 +6139,99 @@ pub fn run() { let poll_secret = aria2_secret.clone(); let poll_mgr = Arc::clone(&queue_manager_poll); tauri::async_runtime::spawn(async move { + #[derive(Default)] + struct Aria2ConnectionObservation { + gid: String, + saw_multiple_connections: bool, + degraded_since: Option, + no_progress_since: Option, + refreshed: bool, + last_completed: u64, + } + + const RECOVERY_DELAY: Duration = Duration::from_secs(30); + const MIN_REMAINING_FOR_CONNECTION_RECOVERY: u64 = 1024 * 1024; let mut interval = tokio::time::interval(std::time::Duration::from_millis(1000)); + let mut observations: HashMap = HashMap::new(); loop { interval.tick().await; - let params = serde_json::json!([["gid", "status", "totalLength", "completedLength", "downloadSpeed", "errorMessage"]]); + let params = serde_json::json!([["gid", "status", "totalLength", "completedLength", "downloadSpeed", "connections", "errorMessage"]]); if let Ok(active_list) = rpc_call(poll_port.load(std::sync::atomic::Ordering::Relaxed), &poll_secret, "aria2.tellActive", params).await { if let Some(active_arr) = active_list.as_array() { + let mut seen_ids = std::collections::HashSet::new(); for status_info in active_arr { let gid = status_info.get("gid").and_then(|s| s.as_str()).unwrap_or(""); let id = poll_mgr.aria2_gids.read().unwrap().get(gid).cloned(); if let Some(id) = id { + seen_ids.insert(id.clone()); + let status = status_info.get("status").and_then(|value| value.as_str()).unwrap_or(""); let total = status_info.get("totalLength").and_then(|s| s.as_str()).unwrap_or("0").parse::().unwrap_or(0); let completed = status_info.get("completedLength").and_then(|s| s.as_str()).unwrap_or("0").parse::().unwrap_or(0); let speed_bytes = status_info.get("downloadSpeed").and_then(|s| s.as_str()).unwrap_or("0").parse::().unwrap_or(0.0); + let active_connections = status_info.get("connections").and_then(|s| s.as_str()).unwrap_or("0").parse::().unwrap_or(0); + let requested_connections = poll_mgr + .aria2_requested_connections(&id) + .await + .unwrap_or(1) + .max(1); + let now = Instant::now(); + let observation = observations.entry(id.clone()).or_default(); + if observation.gid != gid { + *observation = Aria2ConnectionObservation { + gid: gid.to_string(), + last_completed: completed, + ..Default::default() + }; + } + + let remaining = total.saturating_sub(completed); + let mut should_refresh = false; + if status == "active" && total > completed { + if completed > observation.last_completed { + observation.no_progress_since = None; + } else if speed_bytes <= 0.0 { + let stalled_since = observation.no_progress_since.get_or_insert(now); + if now.duration_since(*stalled_since) >= RECOVERY_DELAY + && !observation.refreshed + { + observation.refreshed = true; + should_refresh = true; + } + } else { + observation.no_progress_since = None; + } + + if requested_connections > 1 && active_connections > 1 { + observation.saw_multiple_connections = true; + observation.degraded_since = None; + if !should_refresh { + observation.refreshed = false; + } + } else if requested_connections > 1 + && observation.saw_multiple_connections + && active_connections <= 1 + && remaining >= MIN_REMAINING_FOR_CONNECTION_RECOVERY + { + let degraded_since = observation.degraded_since.get_or_insert(now); + if now.duration_since(*degraded_since) >= RECOVERY_DELAY + && !observation.refreshed + { + observation.refreshed = true; + should_refresh = true; + log::warn!( + "aria2 connection pool degraded [{}]: gid {} has {} connection(s), requested {}; refreshing", + id, + gid, + active_connections, + requested_connections + ); + } + } + } else { + observation.degraded_since = None; + observation.no_progress_since = None; + } + observation.last_completed = completed; let fraction = if total > 0 { completed as f64 / total as f64 } else { 0.0 }; let speed = crate::download::format_speed(speed_bytes); @@ -6064,15 +6248,27 @@ pub fn run() { use tauri::Emitter; let _ = app_handle_poll.emit("download-progress", DownloadProgressEvent { - id, + id: id.clone(), fraction, speed, eta, size, - size_is_final: false, - }); + size_is_final: false, + }); + + if should_refresh { + if let Err(error) = poll_mgr.refresh_aria2_connections(&id, gid).await { + log::warn!( + "aria2 connection refresh [{}] for gid {} failed: {}", + id, + gid, + error + ); + } + } } } + observations.retain(|id, _| seen_ids.contains(id)); } } } diff --git a/src-tauri/src/queue.rs b/src-tauri/src/queue.rs index 0a0c1ec..08a87e6 100644 --- a/src-tauri/src/queue.rs +++ b/src-tauri/src/queue.rs @@ -76,6 +76,18 @@ pub trait SidecarSpawner: Send + Sync + 'static { /// cancellation. async fn remove_uri(&self, gid: &str) -> Result<(), String>; + /// Recycle the connections for an active aria2 transfer without changing + /// its gid or releasing its queue permit. Production uses forcePause / + /// unpause; test spawners can leave this unsupported. + async fn refresh_uri(&self, _gid: &str) -> Result<(), String> { + Err("aria2 connection refresh is unavailable".to_string()) + } + + /// Leave an aria2 transfer paused after a refresh races with a user pause. + async fn pause_uri(&self, _gid: &str) -> Result<(), String> { + Err("aria2 pause is unavailable".to_string()) + } + /// Run a media download to completion. The permit is parked for the full /// duration; release is handled by QueueManager on the runner's exit. async fn run_media(&self, id: &str, payload: &SpawnPayload) -> Result<(), String>; @@ -297,6 +309,15 @@ impl QueueManager { == epoch } + pub async fn current_aria2_control_epoch(&self, id: &str) -> u64 { + self.aria2_control_epochs + .lock() + .await + .get(id) + .copied() + .unwrap_or_default() + } + pub async fn is_aria2_retry_cancelled(&self, id: &str) -> bool { self.aria2_retry_cancelled.lock().await.contains(id) } @@ -305,6 +326,14 @@ impl QueueManager { self.aria2_retry_strikes.lock().await.contains_key(id) } + pub async fn aria2_requested_connections(&self, id: &str) -> Option { + self.aria2_payloads + .lock() + .await + .get(id) + .and_then(|payload| payload.connections) + } + /// Pop the next task, or None if empty. pub async fn pop_front(&self) -> Option { self.pending.lock().await.pop_front() @@ -659,6 +688,39 @@ impl QueueManager { .find_map(|(gid, download_id)| (download_id == id).then(|| gid.clone())) } + pub fn aria2_gid_mappings(&self) -> Vec<(String, String)> { + self.aria2_gids + .read() + .unwrap() + .iter() + .map(|(gid, id)| (gid.clone(), id.clone())) + .collect() + } + + /// Recycle an active transfer's connections after the poller observes a + /// persistent connection-pool collapse or a true zero-progress stall. + /// The transfer keeps its gid, partial file, and queue permit. + pub async fn refresh_aria2_connections(&self, id: &str, gid: &str) -> Result<(), String> { + if self.aria2_gid_for_download(id).as_deref() != Some(gid) + || !self.is_registered(id).await + || self.is_aria2_retry_cancelled(id).await + { + return Ok(()); + } + + let epoch = self.current_aria2_control_epoch(id).await; + self.spawner.refresh_uri(gid).await?; + + let still_current = self.is_registered(id).await + && !self.is_aria2_retry_cancelled(id).await + && self.is_aria2_control_epoch_current(id, epoch).await + && self.aria2_gid_for_download(id).as_deref() == Some(gid); + if !still_current { + let _ = self.spawner.pause_uri(gid).await; + } + Ok(()) + } + /// 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 { @@ -1250,6 +1312,8 @@ impl SidecarSpawner for ProductionSpawner { let mt = aria2_attempt_limit(payload.max_tries); options.insert("max-tries".to_string(), serde_json::json!(mt.to_string())); options.insert("retry-wait".to_string(), serde_json::json!("2")); + options.insert("connect-timeout".to_string(), serde_json::json!("20")); + options.insert("timeout".to_string(), serde_json::json!("60")); options.insert("continue".to_string(), serde_json::json!("true")); options.insert("always-resume".to_string(), serde_json::json!("true")); options.insert("auto-file-renaming".to_string(), serde_json::json!("false")); @@ -1332,6 +1396,44 @@ impl SidecarSpawner for ProductionSpawner { } } + async fn refresh_uri(&self, gid: &str) -> Result<(), String> { + let state = self.app_handle.state::(); + let port = state.aria2_port.load(std::sync::atomic::Ordering::Relaxed); + let secret = &state.aria2_secret; + let paused = crate::rpc_call( + port, + secret, + "aria2.forcePause", + serde_json::json!([gid]), + ) + .await + .map_err(|error| format!("failed to refresh aria2 gid {gid}: {error}"))?; + crate::ensure_aria2_gid_result("forcePause", gid, &paused)?; + + let resumed = crate::rpc_call( + port, + secret, + "aria2.unpause", + serde_json::json!([gid]), + ) + .await + .map_err(|error| format!("failed to refresh aria2 gid {gid}: {error}"))?; + crate::ensure_aria2_gid_result("unpause", gid, &resumed) + } + + async fn pause_uri(&self, gid: &str) -> Result<(), String> { + let state = self.app_handle.state::(); + let result = crate::rpc_call( + state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), + &state.aria2_secret, + "aria2.forcePause", + serde_json::json!([gid]), + ) + .await + .map_err(|error| format!("failed to pause aria2 gid {gid}: {error}"))?; + crate::ensure_aria2_gid_result("forcePause", gid, &result) + } + async fn run_media(&self, id: &str, payload: &SpawnPayload) -> Result<(), String> { let state = self.app_handle.state::(); let mut cancel_rx = state diff --git a/src-tauri/src/retry.rs b/src-tauri/src/retry.rs index 76c4a15..fa4ef0a 100644 --- a/src-tauri/src/retry.rs +++ b/src-tauri/src/retry.rs @@ -97,7 +97,7 @@ pub fn is_transient_network_error(message: &str) -> bool { let m = message.to_ascii_lowercase(); - const TRANSIENT: [&str; 35] = [ + const TRANSIENT: [&str; 38] = [ // socket-layer / HTTP-client phrasing surfaced by aria2 and yt-dlp "timed out", "timeout", @@ -113,6 +113,9 @@ pub fn is_transient_network_error(message: &str) -> bool { "connection aborted", "error sending request", // reqwest wrapper for connect/send failures "dns error", // transient resolver failures + "protocol error", // aria2 read/protocol failures after a link drop + "tls handshake failure", + "ssl/tls handshake failure", // HTTP-level transient "http 408", "request timeout", @@ -275,6 +278,12 @@ mod tests { assert!(is_transient_network_error("The response status is not successful. status=503")); assert!(is_transient_network_error("The response status is not successful. status=502")); assert!(is_transient_network_error("Invalid range header. Request: 106954752-361758719/383882118, Response: 106954752-383882117/383882118")); + assert!(is_transient_network_error( + "aria2 error code 1: Failed to receive data, cause: protocol error" + )); + assert!(is_transient_network_error( + "SSL/TLS handshake failure: protocol error" + )); } // --- transient classification: negative cases ------------------------- diff --git a/src-tauri/tests/queue_manager.rs b/src-tauri/tests/queue_manager.rs index 2ce0049..b5e5057 100644 --- a/src-tauri/tests/queue_manager.rs +++ b/src-tauri/tests/queue_manager.rs @@ -534,7 +534,9 @@ async fn transient_aria2_error_reissues_after_backoff() { manager .handle_aria2_event( "gid-1", - PendingOutcome::Error("Timeout.".to_string()), + PendingOutcome::Error( + "aria2 error code 1: Failed to receive data, cause: protocol error".to_string(), + ), ) .await; @@ -552,7 +554,7 @@ async fn transient_aria2_error_reissues_after_backoff() { manager .handle_aria2_event( "gid-2", - PendingOutcome::Error("Timeout.".to_string()), + PendingOutcome::Error("SSL/TLS handshake failure: protocol error".to_string()), ) .await; assert_eq!(spawner.add_uri_calls.load(Ordering::SeqCst), 2); diff --git a/src/bindings/DownloadItem.ts b/src/bindings/DownloadItem.ts index b0d187a..f7e539a 100644 --- a/src/bindings/DownloadItem.ts +++ b/src/bindings/DownloadItem.ts @@ -2,4 +2,4 @@ import type { DownloadCategory } from "./DownloadCategory"; import type { DownloadStatus } from "./DownloadStatus"; -export type DownloadItem = { id: string, url: string, fileName: string, status: DownloadStatus, fraction?: number, speed?: string, eta?: string, size?: string, category: DownloadCategory, dateAdded: string, resumable?: boolean, connections?: number, speedLimit?: string, username?: string, password?: string, headers?: string, checksum?: string, cookies?: string, mirrors?: string, destination?: string, isMedia?: boolean, mediaFormatSelector?: string, queueId?: string, queuePosition?: number, hasBeenDispatched?: boolean, lastError?: string, }; +export type DownloadItem = { id: string, url: string, fileName: string, status: DownloadStatus, fraction?: number, speed?: string, eta?: string, size?: string, category: DownloadCategory, dateAdded: string, resumable?: boolean, connections?: number, speedLimit?: string, username?: string, password?: string, headers?: string, checksum?: string, cookies?: string, mirrors?: string, destination?: string, isMedia?: boolean, mediaFormatSelector?: string, queueId?: string, queuePosition?: number, hasBeenDispatched?: boolean, lastError?: string, lastTry?: string, }; diff --git a/src/components/LogsView.tsx b/src/components/LogsView.tsx index 18f9b3c..4ff9c20 100644 --- a/src/components/LogsView.tsx +++ b/src/components/LogsView.tsx @@ -1,7 +1,6 @@ import { useEffect, useRef, useState } from 'react'; import { invokeCommand as invoke } from '../ipc'; import { save } from '@tauri-apps/plugin-dialog'; -import { writeTextFile } from '@tauri-apps/plugin-fs'; import { attachLogger, setLogPaused, initLogger } from '../utils/logger'; import { FileDown, Trash2, Terminal, Filter, Play, Pause, Info, Copy } from 'lucide-react'; import { WindowDragRegion } from './WindowDragRegion'; @@ -162,8 +161,7 @@ export default function LogsView() { filters: [{ name: 'Log Files', extensions: ['log'] }], }); if (!path) return; - const logsContent = await invoke('export_logs', {}); - await writeTextFile(path, logsContent); + await invoke('export_logs', { destination: path }); addToast({ message: 'Support logs exported', variant: 'success' }); } catch (e) { console.error('Export failed:', e); diff --git a/src/components/PropertiesModal.tsx b/src/components/PropertiesModal.tsx index 7581059..6da6bf2 100644 --- a/src/components/PropertiesModal.tsx +++ b/src/components/PropertiesModal.tsx @@ -13,6 +13,14 @@ import { type LoginMode = 'matching' | 'custom' | 'none'; +const formatLastTry = (value?: string): string => { + if (!value) return '-'; + const date = new Date(value); + return Number.isNaN(date.getTime()) + ? '-' + : date.toLocaleString(undefined, { dateStyle: 'medium', timeStyle: 'short' }); +}; + export const PropertiesModal = () => { const selectedPropertiesDownloadId = useDownloadStore(state => state.selectedPropertiesDownloadId); const setSelectedPropertiesDownloadId = useDownloadStore(state => state.setSelectedPropertiesDownloadId); @@ -204,7 +212,7 @@ export const PropertiesModal = () => {
Connections{item.connections || perServerConnections || '-'}
Speed cap{item.speedLimit || '-'}
Category{item.category}
-
Last try-
+
Last try{formatLastTry(item.lastTry)}
Date added{new Date(item.dateAdded).toLocaleString(undefined, { dateStyle: 'medium', timeStyle: 'short' })}
Destination{saveLocation || baseDownloadFolder}
diff --git a/src/ipc.ts b/src/ipc.ts index 85eb8bc..7498b8a 100644 --- a/src/ipc.ts +++ b/src/ipc.ts @@ -71,7 +71,7 @@ type CommandMap = { args: { baseFolder: string; subfolders: Record }; result: void; }; - export_logs: { args: Record; result: string }; + export_logs: { args: { destination?: string }; result: string }; read_logs: { args: { limit: number }; result: string[] }; clear_logs: { args: undefined; result: void }; toggle_log_pause: { args: { pause: boolean }; result: void }; diff --git a/src/store/downloadStore.ts b/src/store/downloadStore.ts index f81e8b5..1167cba 100644 --- a/src/store/downloadStore.ts +++ b/src/store/downloadStore.ts @@ -89,7 +89,10 @@ const startDownloadListeners = async () => { const updates: Partial = { status, ...(progress ? { fraction: progress.fraction } : {}), - ...(payload.error ? { lastError: payload.error } : {}) + ...(payload.error ? { lastError: payload.error } : {}), + ...((status === 'downloading' || status === 'retrying') + ? { lastTry: new Date().toISOString() } + : {}) }; if (!payload.error && status !== 'failed' && status !== 'retrying') { updates.lastError = undefined; diff --git a/src/store/useDownloadStore.test.ts b/src/store/useDownloadStore.test.ts index 39f3231..c13c5c6 100644 --- a/src/store/useDownloadStore.test.ts +++ b/src/store/useDownloadStore.test.ts @@ -347,26 +347,79 @@ describe('useDownloadStore', () => { }); it('resumeDownload unregisters ID and re-dispatches if un-resumable', async () => { + let enqueueGeneration: string | undefined; useDownloadStore.setState({ downloads: [ - { id: '1', url: 'http://test1', fileName: 'f1', destination: '/tmp', status: 'paused', category: 'Other', dateAdded: '', queueId: 'MAIN', hasBeenDispatched: true }, + { id: 'resume-generation', url: 'http://test1', fileName: 'f1', destination: '/tmp', status: 'paused', category: 'Other', dateAdded: '', queueId: 'MAIN', hasBeenDispatched: true }, ] as any[], - backendRegisteredIds: new Set(['1']), + backendRegisteredIds: new Set(['resume-generation']), }); - vi.mocked(ipc.invokeCommand).mockImplementation(async (cmd: string) => { + vi.mocked(ipc.invokeCommand).mockImplementation(async (cmd: string, args?: unknown) => { if (cmd === 'resume_download') return false; // Not resumable - if (cmd === 'get_pending_order') return ['1']; + if (cmd === 'enqueue_download') { + enqueueGeneration = (args as { item: { lifecycle_generation: string } }).item.lifecycle_generation; + return { id: 'resume-generation', filename: 'f1' }; + } + if (cmd === 'get_pending_order') return ['resume-generation']; return undefined; }); - await useDownloadStore.getState().resumeDownload('1'); + await useDownloadStore.getState().resumeDownload('resume-generation'); // It should have called resume_download, then unregistered, then enqueue_download const calls = vi.mocked(ipc.invokeCommand).mock.calls; expect(calls.some(c => c[0] === 'resume_download')).toBe(true); expect(calls.some(c => c[0] === 'enqueue_download')).toBe(true); - expect(useDownloadStore.getState().backendRegisteredIds.has('1')).toBe(true); // Re-registered by dispatchItem + expect(enqueueGeneration).toBe('1'); + expect(useDownloadStore.getState().downloads[0].lastTry).toEqual(expect.any(String)); + expect(useDownloadStore.getState().backendRegisteredIds.has('resume-generation')).toBe(true); // Re-registered by dispatchItem + }); + + it('does not re-enqueue when the existing resume RPC fails', async () => { + useDownloadStore.setState({ + downloads: [ + { id: 'resume-rpc-error', url: 'http://test1', fileName: 'f1', destination: '/tmp', status: 'paused', category: 'Other', dateAdded: '', queueId: 'MAIN', hasBeenDispatched: true }, + ] as any[], + backendRegisteredIds: new Set(['resume-rpc-error']), + }); + vi.mocked(ipc.invokeCommand).mockImplementation(async (cmd: string) => { + if (cmd === 'resume_download') throw new Error('aria2 RPC unavailable'); + return undefined; + }); + + await expect(useDownloadStore.getState().resumeDownload('resume-rpc-error')).resolves.toBe(false); + + expect( + vi.mocked(ipc.invokeCommand).mock.calls.some(([command]) => command === 'enqueue_download') + ).toBe(false); + expect(useDownloadStore.getState().downloads[0]).toMatchObject({ + status: 'paused', + lastTry: expect.any(String), + }); + }); + + it('cleans an accepted backend enqueue when queue reconciliation fails', async () => { + useDownloadStore.setState({ + downloads: [ + { id: 'enqueue-reconcile-error', url: 'http://test1', fileName: 'f1', destination: '/tmp', status: 'queued', category: 'Other', dateAdded: '', queueId: 'MAIN', hasBeenDispatched: false }, + ] as any[], + }); + vi.mocked(ipc.invokeCommand).mockImplementation(async (cmd: string) => { + if (cmd === 'enqueue_download') return { id: 'enqueue-reconcile-error', filename: 'f1' }; + if (cmd === 'get_pending_order') throw new Error('queue state unavailable'); + return undefined; + }); + + await expect(useDownloadStore.getState().startQueue('MAIN')).resolves.toEqual([]); + + expect( + vi.mocked(ipc.invokeCommand).mock.calls.some(([command]) => command === 'remove_download') + ).toBe(true); + expect(useDownloadStore.getState().downloads[0]).toMatchObject({ + status: 'failed', + lastError: 'queue state unavailable', + }); }); diff --git a/src/store/useDownloadStore.ts b/src/store/useDownloadStore.ts index 094b610..e91794f 100644 --- a/src/store/useDownloadStore.ts +++ b/src/store/useDownloadStore.ts @@ -90,6 +90,7 @@ export async function dispatchItem(id: string): Promise { const promise = (async () => { let lifecycleGeneration: bigint | null = null; + let backendAccepted = false; try { const state = useDownloadStore.getState(); const item = state.downloads.find(d => d.id === id); @@ -139,7 +140,11 @@ export async function dispatchItem(id: string): Promise { lifecycle_generation: lifecycleGeneration.toString(), }; + useDownloadStore.getState().updateDownload(id, { + lastTry: new Date().toISOString() + }); const accepted = await invoke('enqueue_download', { item: enqueueItem }); + backendAccepted = true; if (!isCurrentDownloadLifecycle(id, lifecycleGeneration)) { await removeStaleBackendDispatch(id); return false; @@ -164,6 +169,9 @@ export async function dispatchItem(id: string): Promise { return true; } catch (e) { console.error(`Failed to dispatch ${id}:`, e); + if (backendAccepted && lifecycleGeneration !== null) { + await removeStaleBackendDispatch(id); + } if (lifecycleGeneration !== null && isCurrentDownloadLifecycle(id, lifecycleGeneration)) { useDownloadStore.getState().updateDownload(id, { status: 'failed', @@ -783,14 +791,19 @@ export const useDownloadStore = create((set, get) => ({ status: 'queued', speed: '-', eta: '-', - queuePosition: maxPos + 1 + queuePosition: maxPos + 1, + lastTry: new Date().toISOString() }); - const resumedExisting = await invoke('resume_download', { id }).catch(() => false); + const resumedExisting = await invoke('resume_download', { id }); let dispatchSucceeded = resumedExisting; if (!dispatchSucceeded) { get().unregisterBackendIds([id]); + // 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); dispatchSucceeded = await dispatchItem(id); }