diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 7274a91..2abd0f5 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -977,30 +977,47 @@ fn aggregate_media_fraction( ((*current_track + *last_fraction) / total_tracks).clamp(0.0, 1.0) } +struct MediaProgressEmitterState { + current_track: f64, + last_fraction: f64, + speed_sampler: MediaSpeedSampler, + last_progress_at: Instant, +} + +impl MediaProgressEmitterState { + fn new() -> Self { + Self { + current_track: 0.0, + last_fraction: 0.0, + speed_sampler: MediaSpeedSampler::default(), + last_progress_at: Instant::now() + .checked_sub(MEDIA_PROGRESS_EMIT_INTERVAL) + .unwrap_or_else(Instant::now), + } + } +} + fn emit_media_progress( app_handle: &tauri::AppHandle, id: &str, progress: MediaProgress, total_tracks: f64, - current_track: &mut f64, - last_fraction: &mut f64, - speed_sampler: &mut MediaSpeedSampler, - last_progress_at: &mut Instant, + state: &mut MediaProgressEmitterState, ) { - let previous_track = *current_track; + let previous_track = state.current_track; let overall_fraction = aggregate_media_fraction( total_tracks, - current_track, - last_fraction, + &mut state.current_track, + &mut state.last_fraction, progress.fraction, ); - if *current_track != previous_track { - speed_sampler.reset(); + if state.current_track != previous_track { + state.speed_sampler.reset(); } - let (speed, eta) = media_progress_speed(&progress, Instant::now(), speed_sampler); + let (speed, eta) = media_progress_speed(&progress, Instant::now(), &mut state.speed_sampler); let now = Instant::now(); - if now.duration_since(*last_progress_at) >= MEDIA_PROGRESS_EMIT_INTERVAL { + if now.duration_since(state.last_progress_at) >= MEDIA_PROGRESS_EMIT_INTERVAL { let _ = app_handle.emit( "download-progress", DownloadProgressEvent { @@ -1012,7 +1029,7 @@ fn emit_media_progress( size_is_final: false, }, ); - *last_progress_at = now; + state.last_progress_at = now; } } @@ -1408,6 +1425,7 @@ static MEDIA_METADATA_LOCKS: OnceLock< tokio::sync::Mutex>>>, > = OnceLock::new(); +#[allow(clippy::too_many_arguments)] // Hash every user-controlled yt-dlp input explicitly. fn media_metadata_cache_key( url: &str, cookie_browser: &Option, @@ -1454,6 +1472,7 @@ fn resolve_metadata_ytdlp_path( } #[tauri::command] +#[allow(clippy::too_many_arguments)] // Keep the generated TypeScript IPC contract flat and stable. async fn fetch_media_metadata( app_handle: tauri::AppHandle, url: String, @@ -1564,6 +1583,7 @@ async fn fetch_media_metadata( result } +#[allow(clippy::too_many_arguments)] // Mirrors the stable command contract for the fallback call. async fn fetch_media_metadata_uncached( app_handle: tauri::AppHandle, url: String, @@ -2840,12 +2860,7 @@ pub(crate) async fn start_media_download_internal( None }; let _keep_alive = config_path; - let mut current_track: f64 = 0.0; - let mut last_fraction: f64 = 0.0; - let mut speed_sampler = MediaSpeedSampler::default(); - let mut last_progress_at = std::time::Instant::now() - .checked_sub(MEDIA_PROGRESS_EMIT_INTERVAL) - .unwrap_or_else(std::time::Instant::now); + let mut progress_state = MediaProgressEmitterState::new(); // Resolve absolute paths to bundled binaries let aria2c_path = resolve_bundled_binary_path(&app_handle, "aria2c")?; @@ -2993,10 +3008,7 @@ pub(crate) async fn start_media_download_internal( id, progress, total_tracks, - &mut current_track, - &mut last_fraction, - &mut speed_sampler, - &mut last_progress_at, + &mut progress_state, ); } else { let candidate = line.trim(); @@ -3022,10 +3034,7 @@ pub(crate) async fn start_media_download_internal( id, progress, total_tracks, - &mut current_track, - &mut last_fraction, - &mut speed_sampler, - &mut last_progress_at, + &mut progress_state, ); } if !processing_started && is_media_processing_line(&line) { @@ -3064,10 +3073,7 @@ pub(crate) async fn start_media_download_internal( id, progress, total_tracks, - &mut current_track, - &mut last_fraction, - &mut speed_sampler, - &mut last_progress_at, + &mut progress_state, ); } else { let candidate = line.trim(); @@ -3086,10 +3092,7 @@ pub(crate) async fn start_media_download_internal( id, progress, total_tracks, - &mut current_track, - &mut last_fraction, - &mut speed_sampler, - &mut last_progress_at, + &mut progress_state, ); } if !processing_started && is_media_processing_line(&line) { @@ -3953,7 +3956,21 @@ async fn enqueue_download( &item.destination, &item.filename, )?; - if let Err(e) = state.queue_manager.push(item.into_task()).await { + let lifecycle_generation = item + .lifecycle_generation + .as_deref() + .map(|generation| { + generation + .parse::() + .map_err(|_| AppError::Internal("Invalid enqueue lifecycle generation".to_string())) + }) + .transpose()? + .unwrap_or_default(); + if let Err(e) = state + .queue_manager + .push_with_generation(item.into_task(), lifecycle_generation) + .await + { let _ = crate::download_ownership::remove(&app_handle, &id); state.queue_manager.release_registered_id(&id).await; return Err(AppError::Internal(e)); @@ -3964,6 +3981,22 @@ async fn enqueue_download( }) } +#[tauri::command] +async fn cancel_enqueue_generation( + state: tauri::State<'_, AppState>, + id: String, + generation: String, +) -> Result<(), AppError> { + let generation = generation + .parse::() + .map_err(|_| AppError::Internal("Invalid enqueue lifecycle generation".to_string()))?; + state + .queue_manager + .cancel_enqueue_generation(&id, generation) + .await; + Ok(()) +} + #[tauri::command] async fn enqueue_many( app_handle: tauri::AppHandle, @@ -5793,7 +5826,7 @@ pub fn run() { check_file_exists, toggle_tray_icon, set_extension_pairing_token, get_extension_server_port, set_extension_frontend_ready, set_concurrent_limit, set_global_speed_limit, remove_download, detach_download_for_reconfigure, - enqueue_download, enqueue_many, move_in_queue, remove_from_queue, get_pending_order, + enqueue_download, enqueue_many, cancel_enqueue_generation, move_in_queue, remove_from_queue, get_pending_order, commands::reveal_in_file_manager, commands::open_downloaded_file, parity::get_system_proxy, parity::get_file_category, parity::check_for_updates, parity::is_supported_media, parity::get_supported_media_domains, parity::create_category_directories, diff --git a/src-tauri/src/parity.rs b/src-tauri/src/parity.rs index d671400..7afecbb 100644 --- a/src-tauri/src/parity.rs +++ b/src-tauri/src/parity.rs @@ -86,10 +86,10 @@ fn proxy_from_environment() -> Option { fn command_stdout(command: &mut Command) -> std::io::Result { let output = command.output()?; if !output.status.success() { - return Err(std::io::Error::new( - std::io::ErrorKind::Other, - format!("command exited with {}", output.status), - )); + return Err(std::io::Error::other(format!( + "command exited with {}", + output.status + ))); } Ok(String::from_utf8_lossy(&output.stdout).to_string()) } diff --git a/src-tauri/src/queue.rs b/src-tauri/src/queue.rs index d44f4f6..7f149bf 100644 --- a/src-tauri/src/queue.rs +++ b/src-tauri/src/queue.rs @@ -84,6 +84,7 @@ pub trait SidecarSpawner: Send + Sync + 'static { /// The centralized concurrency gatekeeper. One instance lives in AppState. pub struct QueueManager { registered_ids: Mutex>, + enqueue_cancellations: Mutex>, pending: Mutex>, semaphore: Arc, active_permits: Mutex>, @@ -134,6 +135,7 @@ impl QueueManager { ) -> Self { Self { registered_ids: Mutex::new(HashSet::new()), + enqueue_cancellations: Mutex::new(HashMap::new()), pending: Mutex::new(VecDeque::new()), semaphore: Arc::new(Semaphore::new(capacity)), active_permits: Mutex::new(HashMap::new()), @@ -173,6 +175,41 @@ impl QueueManager { self.registered_ids.lock().await.contains(id) } + /// Reject an in-flight enqueue generation if a newer UI action supersedes it. + pub async fn cancel_enqueue_generation(&self, id: &str, generation: u64) { + let mut cancellations = self.enqueue_cancellations.lock().await; + cancellations + .entry(id.to_string()) + .and_modify(|current| *current = (*current).max(generation)) + .or_insert(generation); + } + + /// Atomically checks the cancellation watermark before registering a task. + pub async fn push_with_generation( + &self, + task: QueuedTask, + generation: u64, + ) -> Result<(), String> { + let id = task.id.clone(); + let cancellations = self.enqueue_cancellations.lock().await; + if cancellations.get(&id).is_some_and(|cancelled| *cancelled >= generation) { + return Err("Download enqueue was superseded by a newer user action".to_string()); + } + + let mut registered = self.registered_ids.lock().await; + if registered.contains(&id) { + return Err("Duplicate task".to_string()); + } + registered.insert(id.clone()); + drop(registered); + drop(cancellations); + + self.pending.lock().await.push_back(task); + self.emit_state(id, DownloadStatus::Queued); + self.notify.notify_one(); + Ok(()) + } + 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); @@ -1336,6 +1373,9 @@ pub struct EnqueueItem { pub format_selector: Option, pub cookie_source: Option, pub is_media: Option, + #[serde(default)] + #[ts(optional)] + pub lifecycle_generation: Option, } impl EnqueueItem { diff --git a/src-tauri/tests/queue_manager.rs b/src-tauri/tests/queue_manager.rs index af4744b..b574365 100644 --- a/src-tauri/tests/queue_manager.rs +++ b/src-tauri/tests/queue_manager.rs @@ -101,6 +101,21 @@ async fn push_appends_to_pending_and_emits_queued() { assert_eq!(order, vec!["a".to_string(), "b".to_string()]); } +#[tokio::test] +async fn cancelled_enqueue_generation_cannot_register_after_a_newer_user_action() { + let (mgr, _spawner) = make_manager(2); + mgr.cancel_enqueue_generation("a", 4).await; + + let stale = mgr.push_with_generation(sample_task("a"), 4).await; + assert!(stale.is_err(), "cancelled generation must not enter the queue"); + assert!(!mgr.is_registered("a").await); + + mgr.push_with_generation(sample_task("a"), 5) + .await + .expect("newer generation should be accepted"); + assert_eq!(mgr.pending_order(None).await, vec!["a".to_string()]); +} + #[tokio::test] async fn release_permit_is_idempotent() { let (mgr, _spawner) = make_manager(2); diff --git a/src/App.tsx b/src/App.tsx index c4f2d01..e657236 100644 --- a/src/App.tsx +++ b/src/App.tsx @@ -442,7 +442,7 @@ function App() { const trackedIds = state.schedulerActiveDownloadIds; if (trackedIds.length > 0) { const pauseResults = await Promise.allSettled( - trackedIds.map(id => invoke('pause_download', { id })) + trackedIds.map(id => useDownloadStore.getState().pauseDownload(id)) ); const failedPauses = pauseResults.filter(result => result.status === 'rejected').length; if (failedPauses > 0) { @@ -586,6 +586,7 @@ function App() { useEffect(() => { if (!coreReady) return; + let disposed = false; const unlistenDownload = initDownloadListener(); const unlistenTerminalState = listen('download-state', (event) => { @@ -630,10 +631,14 @@ function App() { useDownloadStore.getState().openAddModalWithUrls(event.payload); }); Promise.all([unlistenExtension, unlistenDeepLink]) - .then(() => invoke('set_extension_frontend_ready', { ready: true })) + .then(() => { + if (disposed) return; + return invoke('set_extension_frontend_ready', { ready: true }); + }) .catch(error => console.error('Failed to activate browser extension integration:', error)); return () => { + disposed = true; invoke('set_extension_frontend_ready', { ready: false }).catch(() => {}); unlistenTerminalState.then(f => f()); unlistenExtension.then(f => f()); diff --git a/src/ErrorBoundary.tsx b/src/ErrorBoundary.tsx deleted file mode 100644 index 8453338..0000000 --- a/src/ErrorBoundary.tsx +++ /dev/null @@ -1,52 +0,0 @@ -import { Component, ErrorInfo, ReactNode } from "react"; - -interface Props { - children?: ReactNode; -} - -interface State { - hasError: boolean; - error: Error | null; - errorInfo: ErrorInfo | null; -} - -export class ErrorBoundary extends Component { - public state: State = { - hasError: false, - error: null, - errorInfo: null - }; - - public static getDerivedStateFromError(error: Error): State { - return { hasError: true, error, errorInfo: null }; - } - - public componentDidCatch(error: Error, errorInfo: ErrorInfo) { - console.error("Uncaught error:", error, errorInfo); - this.setState({ errorInfo }); - } - - public render() { - if (this.state.hasError) { - return ( -
-
-

Firelink could not display this window.

-

- The error was written to Logs. Reload the interface to reconnect to the running download service. -

- -
-
- ); - } - - return this.props.children; - } -} diff --git a/src/bindings/EnqueueItem.ts b/src/bindings/EnqueueItem.ts index a2dcbfc..77a1aff 100644 --- a/src/bindings/EnqueueItem.ts +++ b/src/bindings/EnqueueItem.ts @@ -1,3 +1,3 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. -export type EnqueueItem = { id: string, queue_id: string, url: string, destination: string, filename: string, connections: number | null, speed_limit: string | null, username: string | null, password: string | null, headers: string | null, checksum: string | null, cookies: string | null, mirrors: string | null, user_agent: string | null, max_tries: number | null, proxy: string | null, format_selector: string | null, cookie_source: string | null, is_media: boolean | null, }; +export type EnqueueItem = { id: string, queue_id: string, url: string, destination: string, filename: string, connections: number | null, speed_limit: string | null, username: string | null, password: string | null, headers: string | null, checksum: string | null, cookies: string | null, mirrors: string | null, user_agent: string | null, max_tries: number | null, proxy: string | null, format_selector: string | null, cookie_source: string | null, is_media: boolean | null, lifecycle_generation?: string, }; diff --git a/src/components/DownloadTable.tsx b/src/components/DownloadTable.tsx index 7aa4f5f..08efbff 100644 --- a/src/components/DownloadTable.tsx +++ b/src/components/DownloadTable.tsx @@ -338,7 +338,7 @@ export const DownloadTable: React.FC = ({ filter }) => { } try { - await invoke('pause_download', { id }); + await useDownloadStore.getState().pauseDownload(id); } catch (e) { console.error("Failed to pause:", e); showInteractionError('Could not pause download', e); diff --git a/src/ipc.ts b/src/ipc.ts index 0b70b69..4db9b12 100644 --- a/src/ipc.ts +++ b/src/ipc.ts @@ -78,6 +78,7 @@ type CommandMap = { is_log_paused: { args: undefined; result: boolean }; get_pending_order: { args: { queueId: string | null }; result: string[] }; enqueue_download: { args: { item: EnqueueItem }; result: EnqueueAccepted }; + cancel_enqueue_generation: { args: { id: string; generation: string }; result: void }; enqueue_many: { args: { items: EnqueueItem[] }; result: import('./bindings/EnqueueResult').EnqueueResult[] }; move_in_queue: { args: { id: string; queueId: string; direction: 'up' | 'down' }; result: string[] }; remove_from_queue: { args: { id: string }; result: boolean }; diff --git a/src/store/downloadStore.test.ts b/src/store/downloadStore.test.ts index 88b6bf8..aeebb5a 100644 --- a/src/store/downloadStore.test.ts +++ b/src/store/downloadStore.test.ts @@ -1,8 +1,15 @@ -import { beforeEach, describe, expect, it } from 'vitest'; -import { useDownloadProgressStore } from './downloadStore'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import { initDownloadListener, useDownloadProgressStore } from './downloadStore'; +import * as ipc from '../ipc'; + +vi.mock('../ipc', () => ({ + invokeCommand: vi.fn(), + listenEvent: vi.fn(), +})); describe('useDownloadProgressStore', () => { beforeEach(() => { + vi.clearAllMocks(); useDownloadProgressStore.setState({ progressMap: {} }); }); @@ -20,4 +27,22 @@ describe('useDownloadProgressStore', () => { expect(useDownloadProgressStore.getState().progressMap).toEqual({}); }); + + it('shares listener setup across overlapping consumers and tears down after the last release', async () => { + const unlisten = vi.fn(); + vi.mocked(ipc.listenEvent).mockResolvedValue(unlisten); + + const first = initDownloadListener(); + const second = initDownloadListener(); + + expect(ipc.listenEvent).toHaveBeenCalledTimes(3); + + const releaseFirst = await first; + const releaseSecond = await second; + releaseFirst(); + expect(unlisten).not.toHaveBeenCalled(); + + releaseSecond(); + expect(unlisten).toHaveBeenCalledTimes(3); + }); }); diff --git a/src/store/downloadStore.ts b/src/store/downloadStore.ts index f9ba395..f81e8b5 100644 --- a/src/store/downloadStore.ts +++ b/src/store/downloadStore.ts @@ -35,40 +35,50 @@ export const useDownloadProgressStore = create((set) => ( let unlistenProgress: UnlistenFn | null = null; let unlistenState: UnlistenFn | null = null; let unlistenTray: UnlistenFn | null = null; +let listenerSetup: Promise | null = null; +let listenerConsumers = 0; -export async function initDownloadListener() { - if (unlistenProgress) return; - unlistenProgress = await listen('download-progress', (event) => { - const payload = event.payload; - useDownloadProgressStore.getState().updateDownloadProgress(payload.id, payload); +const disposeDownloadListeners = () => { + unlistenProgress?.(); + unlistenProgress = null; + unlistenState?.(); + unlistenState = null; + unlistenTray?.(); + unlistenTray = null; + listenerSetup = null; +}; - const mainStore = useDownloadStore.getState(); - const current = mainStore.downloads.find(d => d.id === payload.id); - if (current) { - const shouldUpdateSize = Boolean(payload.size && (!current.isMedia || payload.size_is_final)); - const updates: Partial = {}; - if (current.status === 'downloading' || current.status === 'processing') { - updates.fraction = payload.fraction; - updates.speed = payload.speed; - updates.eta = payload.eta; - } - if (shouldUpdateSize && current.size !== payload.size) { - updates.size = payload.size!; - } - if (Object.keys(updates).length > 0) { - mainStore.updateDownload(payload.id, updates); - } - } - }); +const startDownloadListeners = async () => { + const registrations = await Promise.allSettled([ + listen('download-progress', (event) => { + const payload = event.payload; + useDownloadProgressStore.getState().updateDownloadProgress(payload.id, payload); - if (!unlistenState) { - unlistenState = await listen('download-state', (event) => { + const mainStore = useDownloadStore.getState(); + const current = mainStore.downloads.find(d => d.id === payload.id); + if (current) { + const shouldUpdateSize = Boolean(payload.size && (!current.isMedia || payload.size_is_final)); + const updates: Partial = {}; + if (current.status === 'downloading' || current.status === 'processing') { + updates.fraction = payload.fraction; + updates.speed = payload.speed; + updates.eta = payload.eta; + } + if (shouldUpdateSize && current.size !== payload.size) { + updates.size = payload.size!; + } + if (Object.keys(updates).length > 0) { + mainStore.updateDownload(payload.id, updates); + } + } + }), + listen('download-state', (event) => { const payload = event.payload; const mainStore = useDownloadStore.getState(); const current = mainStore.downloads.find(d => d.id === payload.id); if (current) { const status = payload.status as DownloadStatus; - + // Prevent race condition: don't transition backwards from terminal state if ((current.status === 'completed' || current.status === 'failed') && (status !== 'completed' && status !== 'failed')) { @@ -111,32 +121,58 @@ export async function initDownloadListener() { useDownloadProgressStore.getState().clearDownloadProgress(payload.id); } } - }); - } - - if (!unlistenTray) { - unlistenTray = await listen('tray-action', (event) => { + }), + listen('tray-action', (event) => { const mainStore = useDownloadStore.getState(); if (event.payload === 'pause-all') { void mainStore.pauseAll(); } else if (event.payload === 'resume-all') { void mainStore.startAll(); } + }), + ]); + + const failedRegistration = registrations.find( + (registration): registration is PromiseRejectedResult => registration.status === 'rejected' + ); + if (failedRegistration) { + for (const registration of registrations) { + if (registration.status === 'fulfilled') registration.value(); + } + throw failedRegistration.reason; + } + + const [progress, state, tray] = registrations as [ + PromiseFulfilledResult, + PromiseFulfilledResult, + PromiseFulfilledResult, + ]; + unlistenProgress = progress.value; + unlistenState = state.value; + unlistenTray = tray.value; +}; + +export async function initDownloadListener(): Promise<() => void> { + listenerConsumers += 1; + if (!listenerSetup) { + listenerSetup = startDownloadListeners().catch(error => { + disposeDownloadListeners(); + throw error; }); } + try { + await listenerSetup; + } catch (error) { + listenerConsumers -= 1; + throw error; + } + + let released = false; return () => { - if (unlistenProgress) { - unlistenProgress(); - unlistenProgress = null; - } - if (unlistenState) { - unlistenState(); - unlistenState = null; - } - if (unlistenTray) { - unlistenTray(); - unlistenTray = null; - } + if (released) return; + released = true; + listenerConsumers -= 1; + if (listenerConsumers === 0) disposeDownloadListeners(); }; } diff --git a/src/store/useDownloadStore.test.ts b/src/store/useDownloadStore.test.ts index 0da13b7..c874d0f 100644 --- a/src/store/useDownloadStore.test.ts +++ b/src/store/useDownloadStore.test.ts @@ -214,6 +214,120 @@ describe('useDownloadStore', () => { }); }); + it('does not resurrect a row removed while its backend enqueue is in flight', async () => { + useDownloadStore.setState({ + downloads: [ + { id: 'late', url: 'http://test', fileName: 'late.bin', destination: '/tmp', status: 'queued', category: 'Other', dateAdded: '', queueId: 'MAIN', hasBeenDispatched: false }, + ] as any[], + }); + + let resolveEnqueue!: (value: { id: string; filename: string }) => void; + const enqueue = new Promise<{ id: string; filename: string }>(resolve => { + resolveEnqueue = resolve; + }); + vi.mocked(ipc.invokeCommand).mockImplementation((command: string) => { + if (command === 'enqueue_download') return enqueue as never; + if (command === 'get_pending_order') return Promise.resolve(['late']) as never; + return Promise.resolve(undefined) as never; + }); + + const start = useDownloadStore.getState().startQueue('MAIN'); + await vi.waitFor(() => { + expect(ipc.invokeCommand).toHaveBeenCalledWith( + 'enqueue_download', + expect.objectContaining({ item: expect.objectContaining({ id: 'late' }) }) + ); + }); + + await useDownloadStore.getState().removeDownload('late'); + resolveEnqueue({ id: 'late', filename: 'late.bin' }); + + await expect(start).resolves.toEqual([]); + expect(useDownloadStore.getState().downloads).toEqual([]); + expect(useDownloadStore.getState().backendRegisteredIds.has('late')).toBe(false); + expect( + vi.mocked(ipc.invokeCommand).mock.calls.filter(([command]) => command === 'remove_download') + ).toHaveLength(2); + }); + + it('re-enqueues the edited values only after an obsolete queued dispatch is removed', async () => { + useDownloadStore.setState({ + downloads: [ + { id: 'edited', url: 'http://test', fileName: 'old.bin', destination: '/tmp', status: 'queued', category: 'Other', dateAdded: '', queueId: 'MAIN', hasBeenDispatched: false }, + ] as any[], + }); + + let resolveFirstEnqueue!: (value: { id: string; filename: string }) => void; + const firstEnqueue = new Promise<{ id: string; filename: string }>(resolve => { + resolveFirstEnqueue = resolve; + }); + let enqueueCount = 0; + vi.mocked(ipc.invokeCommand).mockImplementation((command: string) => { + if (command === 'enqueue_download') { + enqueueCount += 1; + return (enqueueCount === 1 + ? firstEnqueue + : Promise.resolve({ id: 'edited', filename: 'new.bin' })) as never; + } + if (command === 'get_pending_order') return Promise.resolve(['edited']) as never; + return Promise.resolve(undefined) as never; + }); + + const start = useDownloadStore.getState().startQueue('MAIN'); + await vi.waitFor(() => expect(enqueueCount).toBe(1)); + const update = useDownloadStore.getState().applyProperties('edited', { fileName: 'new.bin' }); + resolveFirstEnqueue({ id: 'edited', filename: 'old.bin' }); + + await expect(update).resolves.toBeUndefined(); + await expect(start).resolves.toEqual([]); + expect(enqueueCount).toBe(2); + expect(vi.mocked(ipc.invokeCommand)).toHaveBeenCalledWith( + 'get_pending_order', + { queueId: 'MAIN' } + ); + expect(useDownloadStore.getState().downloads[0]).toMatchObject({ + fileName: 'new.bin', + hasBeenDispatched: true, + }); + }); + + it('settles an in-flight dispatch before pausing so it cannot start after pause succeeds', async () => { + useDownloadStore.setState({ + downloads: [ + { id: 'paused', url: 'http://test', fileName: 'paused.bin', destination: '/tmp', status: 'queued', category: 'Other', dateAdded: '', queueId: 'MAIN', hasBeenDispatched: false }, + ] as any[], + }); + + let resolveEnqueue!: (value: { id: string; filename: string }) => void; + const enqueue = new Promise<{ id: string; filename: string }>(resolve => { + resolveEnqueue = resolve; + }); + vi.mocked(ipc.invokeCommand).mockImplementation((command: string) => { + if (command === 'enqueue_download') return enqueue as never; + return Promise.resolve(undefined) as never; + }); + + const start = useDownloadStore.getState().startQueue('MAIN'); + await vi.waitFor(() => { + expect(ipc.invokeCommand).toHaveBeenCalledWith( + 'enqueue_download', + expect.objectContaining({ item: expect.objectContaining({ id: 'paused' }) }) + ); + }); + const pause = useDownloadStore.getState().pauseDownload('paused'); + resolveEnqueue({ id: 'paused', filename: 'paused.bin' }); + + await expect(pause).resolves.toBeUndefined(); + await expect(start).resolves.toEqual([]); + expect(useDownloadStore.getState().downloads[0]).toMatchObject({ status: 'paused' }); + expect( + vi.mocked(ipc.invokeCommand).mock.calls.filter(([command]) => command === 'remove_download') + ).toHaveLength(1); + expect( + vi.mocked(ipc.invokeCommand).mock.calls.filter(([command]) => command === 'pause_download') + ).toHaveLength(1); + }); + it('resumeDownload unregisters ID and re-dispatches if un-resumable', async () => { useDownloadStore.setState({ downloads: [ @@ -536,7 +650,12 @@ describe('useDownloadStore', () => { { id: 'active', url: 'https://example.com/file', fileName: 'file', status: 'downloading', category: 'Other', dateAdded: '', queueId: 'main' } ] as any[] }); - vi.mocked(ipc.invokeCommand).mockRejectedValueOnce(new Error('writer did not stop')); + vi.mocked(ipc.invokeCommand).mockImplementation((command: string) => { + if (command === 'remove_download') { + return Promise.reject(new Error('writer did not stop')) as never; + } + return Promise.resolve(undefined) as never; + }); await expect(useDownloadStore.getState().removeDownload('active', true)) .rejects.toThrow('writer did not stop'); diff --git a/src/store/useDownloadStore.ts b/src/store/useDownloadStore.ts index ea3cf3a..8bea402 100644 --- a/src/store/useDownloadStore.ts +++ b/src/store/useDownloadStore.ts @@ -16,6 +16,53 @@ import { canPauseDownload, canStartDownload } from '../utils/downloadActions'; export type { DownloadCategory } from '../utils/downloads'; const backendDispatchPromises = new Map>(); +const downloadLifecycleGenerations = new Map(); + +const advanceDownloadLifecycle = (id: string): bigint => { + const nextGeneration = (downloadLifecycleGenerations.get(id) ?? 0n) + 1n; + downloadLifecycleGenerations.set(id, nextGeneration); + return nextGeneration; +}; + +const currentDownloadLifecycle = (id: string): bigint => + downloadLifecycleGenerations.get(id) ?? 0n; + +type DispatchInvalidation = { + generation: bigint; + pendingDispatch?: Promise; +}; + +const invalidateDispatch = async (id: string): Promise => { + const generation = currentDownloadLifecycle(id); + const nextGeneration = advanceDownloadLifecycle(id); + try { + await invoke('cancel_enqueue_generation', { id, generation: generation.toString() }); + } catch (error) { + console.warn(`Failed to cancel stale backend enqueue for ${id}:`, error); + } + return { generation: nextGeneration, pendingDispatch: backendDispatchPromises.get(id) }; +}; + +const invalidateAndWaitForDispatch = async (id: string): Promise => { + const { pendingDispatch } = await invalidateDispatch(id); + if (!pendingDispatch) return false; + await pendingDispatch; + return true; +}; + +const isCurrentDownloadLifecycle = (id: string, generation: bigint): boolean => + currentDownloadLifecycle(id) === generation && + useDownloadStore.getState().downloads.some(download => download.id === id); + +const removeStaleBackendDispatch = async (id: string): Promise => { + try { + await invoke('remove_download', { id, deleteAssets: false }); + } catch (error) { + // The original remove request may already have won this race. Either way, + // never allow a stale enqueue to make the deleted row live again. + console.warn(`Failed to remove stale backend dispatch for ${id}:`, error); + } +}; const errorMessage = (error: unknown): string => error instanceof Error ? error.message : String(error); @@ -32,15 +79,19 @@ export async function dispatchItem(id: string): Promise { if (backendDispatchPromises.has(id)) return backendDispatchPromises.get(id)!; const promise = (async () => { + let lifecycleGeneration: bigint | null = null; try { const state = useDownloadStore.getState(); const item = state.downloads.find(d => d.id === id); if (!item) return false; if (state.backendRegisteredIds.has(id)) return true; + lifecycleGeneration = currentDownloadLifecycle(id); const settings = useSettingsStore.getState(); const destination = item.destination || await resolveCategoryDestination(settings, item.category); + if (!isCurrentDownloadLifecycle(id, lifecycleGeneration)) return false; + const login = getSiteLogin(item.url, settings); let keychainPassword = null; if (login) { @@ -50,6 +101,10 @@ export async function dispatchItem(id: string): Promise { console.warn("Failed to retrieve keychain password for dispatch:", e); } } + if (!isCurrentDownloadLifecycle(id, lifecycleGeneration)) return false; + + const proxy = await getProxyArgs(settings); + if (!isCurrentDownloadLifecycle(id, lifecycleGeneration)) return false; const enqueueItem = { id: item.id, @@ -67,13 +122,19 @@ export async function dispatchItem(id: string): Promise { mirrors: item.mirrors || null, user_agent: settings.customUserAgent.trim() || null, max_tries: settings.maxAutomaticRetries, - proxy: await getProxyArgs(settings), + proxy, format_selector: item.mediaFormatSelector || null, cookie_source: settings.mediaCookieSource !== 'none' ? settings.mediaCookieSource : null, - is_media: item.isMedia || false + is_media: item.isMedia || false, + lifecycle_generation: lifecycleGeneration.toString(), }; const accepted = await invoke('enqueue_download', { item: enqueueItem }); + if (!isCurrentDownloadLifecycle(id, lifecycleGeneration)) { + await removeStaleBackendDispatch(id); + return false; + } + const acceptedFilename = accepted?.filename || item.fileName; if (acceptedFilename !== item.fileName) { useDownloadStore.getState().updateDownload(id, { @@ -82,16 +143,23 @@ export async function dispatchItem(id: string): Promise { }); } const order = await invoke('get_pending_order', { queueId: item.queueId || MAIN_QUEUE_ID }); + if (!isCurrentDownloadLifecycle(id, lifecycleGeneration)) { + await removeStaleBackendDispatch(id); + return false; + } + useDownloadStore.getState().setPendingOrder(order); useDownloadStore.getState().registerBackendIds([id]); useDownloadStore.getState().updateDownload(id, { lastError: undefined }); return true; } catch (e) { console.error(`Failed to dispatch ${id}:`, e); - useDownloadStore.getState().updateDownload(id, { - status: 'failed', - lastError: errorMessage(e) - }); + if (lifecycleGeneration !== null && isCurrentDownloadLifecycle(id, lifecycleGeneration)) { + useDownloadStore.getState().updateDownload(id, { + status: 'failed', + lastError: errorMessage(e) + }); + } return false; } finally { backendDispatchPromises.delete(id); @@ -289,6 +357,7 @@ interface DownloadState { addDownload: (item: DownloadDraft, action: AddDownloadAction) => Promise; updateDownload: (id: string, updates: Partial) => void; removeDownload: (id: string, deleteFile?: boolean) => Promise; + pauseDownload: (id: string) => Promise; redownload: (id: string) => Promise; resumeDownload: (id: string) => Promise; startQueue: (queueId: string) => Promise; @@ -510,6 +579,7 @@ export const useDownloadStore = create((set, get) => ({ queuePosition, hasBeenDispatched: false }; + advanceDownloadLifecycle(item.id); set((state) => ({ downloads: [...state.downloads, ownedItem] })); if (action.type === 'add-to-queue') { @@ -526,6 +596,7 @@ export const useDownloadStore = create((set, get) => ({ return false; }, applyProperties: async (id, updates) => { + const wasDispatching = await invalidateAndWaitForDispatch(id); const state = get(); const item = state.downloads.find(d => d.id === id); if (!item) return; @@ -548,8 +619,11 @@ export const useDownloadStore = create((set, get) => ({ state.unregisterBackendIds([id]); } state.updateDownload(id, updates); - if (isRegistered) { - if (!await dispatchItem(id)) { + if (isRegistered || wasDispatching) { + const dispatched = await dispatchItem(id); + if (dispatched) { + state.updateDownload(id, { hasBeenDispatched: true }); + } else { state.removeFromQueue(id); } } @@ -590,6 +664,7 @@ export const useDownloadStore = create((set, get) => ({ } }, removeDownload: async (id, deleteFile = false) => { + await invalidateDispatch(id); const item = get().downloads.find(d => d.id === id); if (item) { @@ -606,6 +681,17 @@ export const useDownloadStore = create((set, get) => ({ info(`Download ${id} removed`); syncSystemIntegrations(); }, + pauseDownload: async (id) => { + const { generation } = await invalidateDispatch(id); + + await invoke('pause_download', { id }); + + if (!isCurrentDownloadLifecycle(id, generation)) 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: '-' }); + } + }, redownload: async (id) => { const targetItem = get().downloads.find(d => d.id === id); if (!targetItem) { @@ -619,6 +705,8 @@ export const useDownloadStore = create((set, get) => ({ const url = targetItem.url?.trim(); if (!url) throw new Error('Cannot redownload: original URL is missing.'); + await invalidateAndWaitForDispatch(id); + // Remove from backend to clear its state and delete the existing file so we can overwrite try { await invoke('remove_download', { id, deleteAssets: true }); @@ -746,9 +834,7 @@ export const useDownloadStore = create((set, get) => ({ if (activeIds.length === 0) return 0; - const results = await Promise.allSettled( - activeIds.map(id => invoke('pause_download', { id })) - ); + 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; if (failedCount > 0) { @@ -778,9 +864,7 @@ export const useDownloadStore = create((set, get) => ({ .map(item => item.id); if (activeIds.length === 0) return 0; - const results = await Promise.allSettled( - activeIds.map(id => invoke('pause_download', { id })) - ); + const results = await Promise.allSettled(activeIds.map(id => get().pauseDownload(id))); const pausedCount = results.filter(result => result.status === 'fulfilled').length; syncSystemIntegrations(); return pausedCount; @@ -793,7 +877,13 @@ export const useDownloadStore = create((set, get) => ({ throw new Error(`Pause ${locked.fileName} before moving it to another queue.`); } - for (const item of selected) { + await Promise.all( + selected + .filter(item => item.status !== 'completed') + .map(item => invalidateAndWaitForDispatch(item.id)) + ); + + for (const item of get().downloads.filter(item => selectedIds.has(item.id))) { if (!get().backendRegisteredIds.has(item.id)) continue; if (item.status === 'queued') { await invoke('remove_from_queue', { id: item.id }); @@ -927,7 +1017,8 @@ export const useDownloadStore = create((set, get) => ({ proxy: await getProxyArgs(settings), format_selector: item.mediaFormatSelector || null, cookie_source: settings.mediaCookieSource !== 'none' ? settings.mediaCookieSource : null, - is_media: item.isMedia || false + is_media: item.isMedia || false, + lifecycle_generation: currentDownloadLifecycle(item.id).toString(), }); } const results = await invoke('enqueue_many', { items: itemsToEnqueue });