From 9c6237716e99fb8295b6df47375952ab053b77a8 Mon Sep 17 00:00:00 2001 From: NimBold Date: Mon, 10 Aug 2026 20:35:53 +0330 Subject: [PATCH] fix(persistence): make download admission durable --- src-tauri/src/db.rs | 134 ++++++++++--- src-tauri/src/lib.rs | 102 +++++++++- src/App.tsx | 24 ++- src/ipc.ts | 6 + src/store/downloadStore.test.ts | 72 +++++++ src/store/downloadStore.ts | 49 +++-- src/store/useDownloadStore.test.ts | 277 ++++++++++++++++++++++++++- src/store/useDownloadStore.ts | 293 ++++++++++++++++++++++++----- 8 files changed, 868 insertions(+), 89 deletions(-) diff --git a/src-tauri/src/db.rs b/src-tauri/src/db.rs index fc55dbd..b8807d2 100644 --- a/src-tauri/src/db.rs +++ b/src-tauri/src/db.rs @@ -917,19 +917,7 @@ pub fn replace_downloads( data: &str, portable: bool, ) -> Result<(), String> { - let values: Vec = serde_json::from_str(data) - .map_err(|error| format!("failed to decode downloads: {error}"))?; - let strings = values - .into_iter() - .map(|mut value| { - remove_live_download_metadata(&mut value); - if portable { - remove_persisted_transfer_secrets(&mut value); - } - serde_json::to_string(&value) - .map_err(|error| format!("failed to encode download: {error}")) - }) - .collect::, _>>()?; + let strings = prepare_download_strings(data, portable)?; let transaction = connection .transaction() .map_err(|error| format!("failed to begin download save: {error}"))?; @@ -939,6 +927,24 @@ pub fn replace_downloads( .map_err(|error| format!("failed to commit download save: {error}")) } +pub fn replace_downloads_and_queues( + connection: &mut Connection, + downloads_data: &str, + queues_data: &str, + portable: bool, +) -> Result<(), String> { + let downloads = prepare_download_strings(downloads_data, portable)?; + let queues = prepare_queue_strings(queues_data)?; + let transaction = connection + .transaction() + .map_err(|error| format!("failed to begin download state save: {error}"))?; + replace_downloads_tx(&transaction, &downloads)?; + replace_queues_tx(&transaction, &queues)?; + transaction + .commit() + .map_err(|error| format!("failed to commit download state save: {error}")) +} + /// Mutate exactly one persisted download inside a database transaction. /// /// Native lifecycle code must not rebuild the renderer-owned download array: @@ -1304,14 +1310,7 @@ pub fn load_queues(connection: &Connection) -> Result, String> { } pub fn replace_queues(connection: &mut Connection, data: &str) -> Result<(), String> { - let values: Vec = - serde_json::from_str(data).map_err(|error| format!("failed to decode queues: {error}"))?; - let strings = values - .iter() - .map(|value| { - serde_json::to_string(value).map_err(|error| format!("failed to encode queue: {error}")) - }) - .collect::, _>>()?; + let strings = prepare_queue_strings(data)?; let transaction = connection .transaction() .map_err(|error| format!("failed to begin queue save: {error}"))?; @@ -1321,6 +1320,33 @@ pub fn replace_queues(connection: &mut Connection, data: &str) -> Result<(), Str .map_err(|error| format!("failed to commit queue save: {error}")) } +fn prepare_download_strings(data: &str, portable: bool) -> Result, String> { + let values: Vec = serde_json::from_str(data) + .map_err(|error| format!("failed to decode downloads: {error}"))?; + values + .into_iter() + .map(|mut value| { + remove_live_download_metadata(&mut value); + if portable { + remove_persisted_transfer_secrets(&mut value); + } + serde_json::to_string(&value) + .map_err(|error| format!("failed to encode download: {error}")) + }) + .collect() +} + +fn prepare_queue_strings(data: &str) -> Result, String> { + let values: Vec = + serde_json::from_str(data).map_err(|error| format!("failed to decode queues: {error}"))?; + values + .iter() + .map(|value| { + serde_json::to_string(value).map_err(|error| format!("failed to encode queue: {error}")) + }) + .collect() +} + fn replace_queues_tx(transaction: &Transaction<'_>, queues: &[String]) -> Result<(), String> { transaction .execute("DELETE FROM queues", []) @@ -2357,6 +2383,72 @@ mod tests { assert_eq!(saved["torrentExcludeTrackers"], "https://tracker.example/exclude"); } + #[test] + fn download_state_commit_is_atomic_across_downloads_and_queues() { + let temp = TempDir::new().unwrap(); + let state = init_at_path(temp.path()).unwrap(); + let mut connection = state.lock().unwrap(); + replace_downloads( + &mut connection, + &json!([{ + "id": "old-download", + "status": "paused", + "queueId": "old-queue" + }]) + .to_string(), + false, + ) + .unwrap(); + replace_queues( + &mut connection, + &json!([{ + "id": "old-queue", + "name": "Old Queue", + "isMain": true + }]) + .to_string(), + ) + .unwrap(); + + let result = replace_downloads_and_queues( + &mut connection, + &json!([{ + "id": "new-download", + "status": "queued", + "queueId": "new-queue" + }]) + .to_string(), + &json!([{ + "name": "missing-id" + }]) + .to_string(), + false, + ); + assert!(result.is_err()); + assert!(load_downloads(&connection).unwrap()[0].contains("old-download")); + assert!(load_queues(&connection).unwrap()[0].contains("old-queue")); + + replace_downloads_and_queues( + &mut connection, + &json!([{ + "id": "new-download", + "status": "queued", + "queueId": "new-queue" + }]) + .to_string(), + &json!([{ + "id": "new-queue", + "name": "New Queue", + "isMain": true + }]) + .to_string(), + false, + ) + .unwrap(); + assert!(load_downloads(&connection).unwrap()[0].contains("new-download")); + assert!(load_queues(&connection).unwrap()[0].contains("new-queue")); + } + #[test] fn native_download_mutation_keeps_object_records_and_unrelated_rows_unchanged() { let temp = TempDir::new().unwrap(); diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 7d4debf..33af2cc 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -3085,9 +3085,33 @@ fn push_unique_path(paths: &mut Vec, path: std::path::PathBu } } -use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, AtomicU8, Ordering}; use std::sync::{Arc, Mutex, RwLock}; +struct FrontendExitFlush { + next_request: AtomicU64, + completed: tokio::sync::watch::Sender, +} + +impl FrontendExitFlush { + fn new() -> Self { + let (completed, _) = tokio::sync::watch::channel(0); + Self { + next_request: AtomicU64::new(0), + completed, + } + } + + fn request(&self) -> u64 { + self.next_request.fetch_add(1, Ordering::AcqRel) + 1 + } + + fn acknowledge(&self) { + let request = self.next_request.load(Ordering::Acquire); + let _ = self.completed.send(request); + } +} + struct Aria2DaemonGuard { child: Mutex>, startup_error: Mutex>, @@ -3218,6 +3242,7 @@ pub struct AppState { pub extension_acks: extension_server::SharedExtensionAcks, pub extension_server_port: extension_server::SharedServerPort, pub extension_server_shutdown: tokio::sync::watch::Sender, + frontend_exit_flush: Arc, pub aria2_port: std::sync::Arc, pub aria2_secret: String, pub media_semaphore: Arc, @@ -10316,6 +10341,26 @@ fn db_replace_downloads( crate::db::replace_downloads(&mut connection, &data, portable) } +#[tauri::command] +fn db_commit_download_state( + caller: tauri::WebviewWindow, + state: tauri::State<'_, crate::db::DbState>, + downloads_data: String, + queues_data: String, +) -> Result<(), String> { + properties_window::ensure_main_window(&caller)?; + let portable = state.is_portable(); + let mut connection = state.lock()?; + let existing = crate::db::load_downloads(&connection)?; + let downloads_data = merge_durable_torrent_telemetry(&existing, &downloads_data)?; + crate::db::replace_downloads_and_queues( + &mut connection, + &downloads_data, + &queues_data, + portable, + ) +} + fn persisted_destinations_equal(left: &str, right: &str) -> bool { let left = std::path::Path::new(left.trim()); let right = std::path::Path::new(right.trim()); @@ -10917,6 +10962,16 @@ fn set_extension_frontend_ready( Ok(()) } +#[tauri::command] +fn ack_frontend_exit( + caller: tauri::WebviewWindow, + state: tauri::State<'_, AppState>, +) -> Result<(), String> { + properties_window::ensure_main_window(&caller)?; + state.frontend_exit_flush.acknowledge(); + Ok(()) +} + #[tauri::command] fn ack_extension_download( caller: tauri::WebviewWindow, @@ -10971,6 +11026,7 @@ mod tests { MediaProgressEmitterState, MediaSpeedSampler, MEDIA_PROGRESS_PREFIX, observe_aria2_connections, observe_aria2_connections_with_epoch, Aria2ConnectionObservation, Aria2ConnectionSample, Aria2RecoveryReason, + FrontendExitFlush, aria2_active_connection_count, parse_media_playlist_metadata, normalize_media_connections, @@ -11266,6 +11322,18 @@ mod tests { assert!(!guard.begin_shutdown()); } + #[test] + fn frontend_exit_flush_acknowledges_the_current_request_generation() { + let flush = FrontendExitFlush::new(); + let completed = flush.completed.subscribe(); + let request = flush.request(); + assert_eq!(*completed.borrow(), 0); + + flush.acknowledge(); + + assert_eq!(*completed.borrow(), request); + } + #[test] fn aria2_torrent_global_options_are_bounded_and_explicit() { let mut command = std::process::Command::new("aria2c"); @@ -13947,6 +14015,7 @@ pub fn run() { let server_extension_port = extension_server_port.clone(); let (extension_server_shutdown_tx, extension_server_shutdown_rx) = tokio::sync::watch::channel(false); + let frontend_exit_flush = Arc::new(FrontendExitFlush::new()); let initial_aria2_port = 6800; // Will be determined dynamically in background let aria2_port = Arc::new(std::sync::atomic::AtomicU16::new(initial_aria2_port)); @@ -14200,6 +14269,7 @@ pub fn run() { extension_acks, extension_server_port, extension_server_shutdown: extension_server_shutdown_tx.clone(), + frontend_exit_flush, aria2_port: aria2_port.clone(), aria2_secret: aria2_secret.clone(), media_semaphore: Arc::new(tokio::sync::Semaphore::new(3)), @@ -15274,7 +15344,7 @@ pub fn run() { authorize_keychain_access, acknowledge_pairing_token_change, check_file_exists, toggle_tray_icon, set_extension_pairing_token, - get_extension_server_port, set_extension_frontend_ready, ack_extension_download, set_concurrent_limit, set_queue_concurrency_limits, set_download_speed_limit, set_torrent_upload_limit, set_torrent_peer_options, get_torrent_peers, get_torrent_availability, get_torrent_file_progress, get_torrent_piece_progress, get_torrent_file_selection, set_torrent_file_selection, get_torrent_details, get_torrent_magnet_link, export_torrent_metadata, move_torrent_data, cancel_torrent_move_data, verify_torrent_data, get_torrent_web_seeds, set_torrent_web_seeds, set_torrent_max_open_files, set_torrent_overall_upload_limit, set_global_speed_limit, remove_download, get_download_primary_path, + get_extension_server_port, set_extension_frontend_ready, ack_frontend_exit, ack_extension_download, set_concurrent_limit, set_queue_concurrency_limits, set_download_speed_limit, set_torrent_upload_limit, set_torrent_peer_options, get_torrent_peers, get_torrent_availability, get_torrent_file_progress, get_torrent_piece_progress, get_torrent_file_selection, set_torrent_file_selection, get_torrent_details, get_torrent_magnet_link, export_torrent_metadata, move_torrent_data, cancel_torrent_move_data, verify_torrent_data, get_torrent_web_seeds, set_torrent_web_seeds, set_torrent_max_open_files, set_torrent_overall_upload_limit, set_global_speed_limit, remove_download, get_download_primary_path, detach_download_for_reconfigure, enqueue_download, enqueue_many, cancel_enqueue_generation, move_in_queue, move_many_in_queue, remove_from_queue, get_pending_order, commands::reveal_in_file_manager, commands::open_downloaded_file, @@ -15289,6 +15359,7 @@ pub fn run() { 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, db_save_settings, db_load_settings, db_get_all_downloads, db_replace_downloads, + db_commit_download_state, clear_torrent_removal_paths, reconcile_torrent_removal_reservations, db_get_all_queues, db_replace_queues, read_logs, export_logs, toggle_log_pause, is_log_paused, clear_logs, @@ -15314,14 +15385,37 @@ pub fn run() { } } tauri::RunEvent::ExitRequested { code, api, .. } => { - let state = app_handle.state::(); - let _ = state.extension_server_shutdown.send(true); let guard = app_handle.state::(); if !guard.exit_allowed() { api.prevent_exit(); if guard.begin_shutdown() { let app = app_handle.clone(); tauri::async_runtime::spawn(async move { + let frontend_exit_flush = app.state::().frontend_exit_flush.clone(); + let extension_server_shutdown = app + .state::() + .extension_server_shutdown + .clone(); + let request = frontend_exit_flush.request(); + let mut completed = frontend_exit_flush.completed.subscribe(); + let _ = app.emit_to("main", "app-exit-requested", ()); + let flush_wait = async move { + loop { + if *completed.borrow() >= request { + break; + } + if completed.changed().await.is_err() { + break; + } + } + }; + if tokio::time::timeout(Duration::from_secs(2), flush_wait) + .await + .is_err() + { + log::warn!("frontend persistence flush timed out during exit"); + } + let _ = extension_server_shutdown.send(true); shutdown_aria2_daemon(app.clone()).await; app.state::().allow_exit(); app.exit(code.unwrap_or(0)); diff --git a/src/App.tsx b/src/App.tsx index 5891766..bdb3981 100644 --- a/src/App.tsx +++ b/src/App.tsx @@ -10,7 +10,7 @@ import { KeychainPermissionModal } from './components/KeychainPermissionModal'; import { extractValidDownloadUrls } from './utils/url'; import { readClipboardDownloadUrls } from './utils/clipboard'; import { listenEvent as listen, invokeCommand as invoke } from "./ipc"; -import { initializeDownloadPersistence, useDownloadStore, MAIN_QUEUE_ID, type ExtensionDownloadRequest } from './store/useDownloadStore'; +import { flushDownloadPersistence, initializeDownloadPersistence, useDownloadStore, MAIN_QUEUE_ID, type ExtensionDownloadRequest } from './store/useDownloadStore'; import { getCurrentWindow } from '@tauri-apps/api/window'; import { initDownloadListener } from './store/downloadStore'; import { subscribeToSettingsPersistenceErrors, useSettingsStore } from "./store/useSettingsStore"; @@ -411,6 +411,24 @@ function App() { const disposePersistence = initializeDownloadPersistence(getCurrentWindow().label); let active = true; let cleanupListeners: (() => void) | null = null; + let unlistenExit: (() => void) | null = null; + const exitListener = listen('app-exit-requested', async () => { + try { + await flushDownloadPersistence(); + } catch (error) { + console.error('Failed to flush download state before exit:', error); + } finally { + await invoke('ack_frontend_exit').catch(error => { + console.error('Failed to acknowledge frontend exit flush:', error); + }); + } + }); + void exitListener.then(unlisten => { + if (active) unlistenExit = unlisten; + else unlisten(); + }).catch(error => { + console.error('Failed to listen for frontend exit flush:', error); + }); const initialize = async () => { let unlistenDownload: (() => void) | null = null; let unlistenTerminalState: (() => void) | null = null; @@ -418,6 +436,8 @@ function App() { let unlistenDeepLink: (() => void) | null = null; const disposeListeners = () => { void queueFrontendReadyUpdate(false).catch(() => {}); + unlistenExit?.(); + unlistenExit = null; unlistenTerminalState?.(); unlistenTerminalState = null; unlistenExtension?.(); @@ -624,6 +644,8 @@ function App() { pendingStartupInputs.current = []; cleanupListeners?.(); cleanupListeners = null; + unlistenExit?.(); + unlistenExit = null; disposePersistence(); }; }, [addToast, queueFrontendReadyUpdate]); diff --git a/src/ipc.ts b/src/ipc.ts index 4cd0fe8..335d417 100644 --- a/src/ipc.ts +++ b/src/ipc.ts @@ -136,6 +136,7 @@ type CommandMap = { abandon_keychain_grant: { args: { requestId: string }; result: PairingTokenHydration | null }; acknowledge_pairing_token_change: { args: undefined; result: void }; set_extension_frontend_ready: { args: { ready: boolean }; result: void }; + ack_frontend_exit: { args: undefined; result: void }; ack_extension_download: { args: { requestId: string }; result: void }; get_system_proxy: { args: undefined; result: string | null }; get_file_category: { args: { filename: string }; result: DownloadCategory }; @@ -145,6 +146,10 @@ type CommandMap = { db_load_settings: { args: undefined; result: string | null }; db_get_all_downloads: { args: undefined; result: string[] }; db_replace_downloads: { args: { data: string }; result: void }; + db_commit_download_state: { + args: { downloadsData: string; queuesData: string }; + result: void; + }; db_get_all_queues: { args: undefined; result: string[] }; db_replace_queues: { args: { data: string }; result: void }; create_category_directories: { @@ -198,6 +203,7 @@ type EventMap = { 'extension-add-download': ExtensionDownload; 'deep-link-add-download': string; 'tray-action': 'pause-all' | 'resume-all'; + 'app-exit-requested': null; }; export function listenEvent( diff --git a/src/store/downloadStore.test.ts b/src/store/downloadStore.test.ts index 58db6ae..91de03f 100644 --- a/src/store/downloadStore.test.ts +++ b/src/store/downloadStore.test.ts @@ -2,6 +2,7 @@ import { beforeEach, describe, expect, it, vi } from 'vitest'; import { initDownloadListener, useDownloadProgressStore } from './downloadStore'; import { clearDownloadControlIntents, + initializeDownloadPersistence, downloadControlIntentFor, setDownloadControlIntent, useDownloadStore @@ -241,6 +242,77 @@ describe('useDownloadProgressStore', () => { release(); }); + it('accepts Torrent verification while a row is seeding', async () => { + const handlers: Record void> = {}; + vi.mocked(ipc.listenEvent).mockImplementation((event, handler) => { + handlers[event] = handler as (event: any) => void; + return Promise.resolve(vi.fn()); + }); + useDownloadStore.setState({ + downloads: [{ + id: 'torrent-seeding-verification', + url: 'magnet:?xt=urn:btih:test', + fileName: 'ubuntu.iso', + status: 'seeding', + category: 'Other', + dateAdded: '' + }] + }); + + const release = await initDownloadListener(); + handlers['download-state']({ payload: { + id: 'torrent-seeding-verification', + status: 'verifying' + } }); + + expect(useDownloadStore.getState().downloads[0].status).toBe('verifying'); + release(); + }); + + it('durably acknowledges a completed Torrent verification before clearing its marker', async () => { + const handlers: Record unknown> = {}; + vi.mocked(ipc.listenEvent).mockImplementation((event, handler) => { + handlers[event] = handler as (event: any) => unknown; + return Promise.resolve(vi.fn()); + }); + const persistedMarkers: Array = []; + vi.mocked(ipc.invokeCommand).mockImplementation(async (command: string, args?: any) => { + if (command === 'db_commit_download_state') { + const records = JSON.parse(args.downloadsData) as Array<{ torrentVerifyOnly?: boolean }>; + persistedMarkers.push(records[0]?.torrentVerifyOnly); + } + return undefined; + }); + useDownloadStore.setState({ + downloads: [{ + id: 'torrent-verification-ack', + url: 'magnet:?xt=urn:btih:test', + fileName: 'ubuntu.iso', + status: 'paused', + category: 'Other', + dateAdded: '', + isTorrent: true, + torrentVerifyOnly: true, + torrentVerifyRestoreStatus: 'paused' + }] as any[] + }); + const disposePersistence = initializeDownloadPersistence('main'); + + try { + const release = await initDownloadListener(); + await handlers['download-state']({ payload: { + id: 'torrent-verification-ack', + status: 'paused' + } }); + + expect(persistedMarkers).toEqual([true, undefined]); + expect(useDownloadStore.getState().downloads[0].torrentVerifyOnly).toBeUndefined(); + release(); + } finally { + disposePersistence(); + } + }); + it('clears progress when events arrive after a download row was removed', async () => { const handlers: Record void> = {}; vi.mocked(ipc.listenEvent).mockImplementation((event, handler) => { diff --git a/src/store/downloadStore.ts b/src/store/downloadStore.ts index 1dce88e..509e61a 100644 --- a/src/store/downloadStore.ts +++ b/src/store/downloadStore.ts @@ -7,6 +7,7 @@ import { useDownloadProgressStore } from './downloadProgressStore'; import { clearDownloadControlIntent, + commitDownloadState, downloadControlIntentFor, hasStaleTemporaryMediaEstimate, useDownloadStore @@ -111,7 +112,7 @@ const startDownloadListeners = async () => { mainStore.updateDownload(payload.id, updates); } }), - listen('download-state', (event) => { + listen('download-state', async (event) => { const payload = event.payload; const mainStore = useDownloadStore.getState(); const current = mainStore.downloads.find(d => d.id === payload.id); @@ -169,6 +170,7 @@ const startDownloadListeners = async () => { if (current.status === 'seeding' && status !== 'seeding' && status !== 'waitingToSeed' && + status !== 'verifying' && status !== 'paused' && status !== 'completed' && status !== 'failed' && @@ -231,16 +233,10 @@ const startDownloadListeners = async () => { updates.speed = '-'; updates.eta = '-'; } - if ( - current.torrentVerifyOnly === true && - ['ready', 'staged', 'paused', 'completed', 'failed'].includes(status) - ) { - // Verification is a maintenance lifecycle layered over the existing - // row. Clear its markers once Aria2 has reached the restored terminal - // state so restart cannot replay verification indefinitely. - updates.torrentVerifyOnly = undefined; - updates.torrentVerifyRestoreStatus = undefined; - } + const verificationRestoreStatus = current.torrentVerifyRestoreStatus; + const verificationNeedsAcknowledgement = current.torrentVerifyOnly === true && + typeof verificationRestoreStatus === 'string' && + ['ready', 'staged', 'paused', 'completed', 'failed'].includes(status); mainStore.updateDownload(payload.id, updates); if (status === 'completed' || status === 'failed' || status === 'paused' || status === 'seeding' || status === 'waitingToSeed') { @@ -258,6 +254,37 @@ const startDownloadListeners = async () => { } else if (status === 'completed' || status === 'failed') { mainStore.unregisterBackendIds([payload.id]); } + + if (verificationNeedsAcknowledgement) { + try { + // The native persistence marker is intentionally acknowledged in a + // separate durable snapshot before the renderer clears its copy. + // Coalescing both updates into one snapshot would let the native + // marker protect an already-finished verification forever. + await commitDownloadState(); + const acknowledged = useDownloadStore.getState().downloads.find( + download => download.id === payload.id + ); + if ( + !acknowledged || + acknowledged.status !== status || + acknowledged.torrentVerifyOnly !== true || + acknowledged.torrentVerifyRestoreStatus !== verificationRestoreStatus + ) { + return; + } + mainStore.updateDownload(payload.id, { + torrentVerifyOnly: undefined, + torrentVerifyRestoreStatus: undefined + }); + await commitDownloadState(); + } catch (error) { + // Keep the marker in the durable/native path when the acknowledgement + // cannot be committed. Restarting verification is safer than losing + // the integrity-maintenance lifecycle. + console.error('Failed to acknowledge Torrent verification:', error); + } + } }), listen('torrent-move-progress', (event) => { const payload = event.payload; diff --git a/src/store/useDownloadStore.test.ts b/src/store/useDownloadStore.test.ts index 4ae2df3..5d13b8e 100644 --- a/src/store/useDownloadStore.test.ts +++ b/src/store/useDownloadStore.test.ts @@ -1,5 +1,5 @@ import { describe, it, expect, vi, beforeEach } from 'vitest'; -import { dispatchItem, getProxyArgs, getSiteLogin, hasStaleTemporaryMediaEstimate, normalizeCustomProxy, normalizePersistedDownloadProgress, normalizePersistedQueueState, normalizePersistedQueues, useDownloadStore } from './useDownloadStore'; +import { commitDownloadState, dispatchItem, flushDownloadPersistence, getProxyArgs, getSiteLogin, hasStaleTemporaryMediaEstimate, initializeDownloadPersistence, normalizeCustomProxy, normalizePersistedDownloadProgress, normalizePersistedQueueState, normalizePersistedQueues, useDownloadStore } from './useDownloadStore'; import { useDownloadProgressStore } from './downloadProgressStore'; import { useSettingsStore } from './useSettingsStore'; import * as ipc from '../ipc'; @@ -1607,6 +1607,281 @@ describe('useDownloadStore', () => { expect(ipc.invokeCommand).not.toHaveBeenCalledWith('enqueue_download', expect.anything()); }); + it('waits for durable admission before dispatching a start-now download', async () => { + const disposePersistence = initializeDownloadPersistence('main'); + const events: string[] = []; + let releaseCommit!: () => void; + let signalCommitStarted!: () => void; + const commitStarted = new Promise(resolve => { + signalCommitStarted = resolve; + }); + const commitGate = new Promise(resolve => { + releaseCommit = resolve; + }); + vi.mocked(ipc.invokeCommand).mockImplementation(async (command: string) => { + if (command === 'db_commit_download_state') { + events.push('commit-start'); + signalCommitStarted(); + await commitGate; + events.push('commit-complete'); + return undefined; + } + if (command === 'enqueue_download') { + events.push('enqueue'); + return { id: 'durable-admission', filename: 'file.bin' }; + } + if (command === 'get_pending_order') return []; + return undefined; + }); + + try { + const adding = useDownloadStore.getState().addDownload({ + id: 'durable-admission', + url: 'https://example.com/file.bin', + fileName: 'file.bin', + category: 'Other', + dateAdded: '' + }, { type: 'start-now' }); + + await commitStarted; + expect(events).toEqual(['commit-start']); + expect(ipc.invokeCommand).not.toHaveBeenCalledWith('enqueue_download', expect.anything()); + + releaseCommit(); + await expect(adding).resolves.toBe(true); + const enqueueIndex = events.indexOf('enqueue'); + expect(enqueueIndex).toBeGreaterThan(0); + expect(events.slice(0, enqueueIndex).filter(event => event === 'commit-start').length) + .toBe(events.slice(0, enqueueIndex).filter(event => event === 'commit-complete').length); + } finally { + disposePersistence(); + } + }); + + it('does not dispatch when durable admission fails', async () => { + const disposePersistence = initializeDownloadPersistence('main'); + vi.mocked(ipc.invokeCommand).mockImplementation(async (command: string) => { + if (command === 'db_commit_download_state') { + throw new Error('database unavailable'); + } + if (command === 'enqueue_download') { + throw new Error('enqueue must not run'); + } + return undefined; + }); + + try { + await expect(useDownloadStore.getState().addDownload({ + id: 'durable-admission-failure', + url: 'https://example.com/file.bin', + fileName: 'file.bin', + category: 'Other', + dateAdded: '' + }, { type: 'start-now' })).resolves.toBe(false); + expect(useDownloadStore.getState().downloads[0]).toMatchObject({ + id: 'durable-admission-failure', + status: 'failed', + lastError: 'database unavailable' + }); + expect(ipc.invokeCommand).not.toHaveBeenCalledWith('enqueue_download', expect.anything()); + } finally { + disposePersistence(); + } + }); + + it('does not enqueue after a lifecycle is invalidated during durable admission', async () => { + const disposePersistence = initializeDownloadPersistence('main'); + let releaseCommit!: () => void; + let signalCommitStarted!: () => void; + const commitStarted = new Promise(resolve => { + signalCommitStarted = resolve; + }); + const commitGate = new Promise(resolve => { + releaseCommit = resolve; + }); + vi.mocked(ipc.invokeCommand).mockImplementation(async (command: string) => { + if (command === 'db_commit_download_state') { + signalCommitStarted(); + await commitGate; + return undefined; + } + if (command === 'enqueue_download') { + throw new Error('stale dispatch must not enqueue'); + } + return undefined; + }); + useDownloadStore.setState({ + downloads: [{ + id: 'admission-lifecycle-race', + url: 'https://example.com/file.bin', + fileName: 'file.bin', + destination: '/tmp', + status: 'queued', + category: 'Other', + dateAdded: '' + }] as any[] + }); + + try { + const dispatching = dispatchItem('admission-lifecycle-race'); + await commitStarted; + const pausing = useDownloadStore.getState().pauseDownload('admission-lifecycle-race'); + releaseCommit(); + + await expect(dispatching).resolves.toBe(false); + await expect(pausing).resolves.toBeUndefined(); + expect(ipc.invokeCommand).not.toHaveBeenCalledWith('enqueue_download', expect.anything()); + expect(useDownloadStore.getState().downloads[0].status).toBe('paused'); + } finally { + disposePersistence(); + } + }); + + it('waits for the latest full snapshot when state changes during a durable commit', async () => { + const disposePersistence = initializeDownloadPersistence('main'); + const persistedIds: string[] = []; + let releaseFirstCommit!: () => void; + let signalFirstCommit!: () => void; + const firstCommitStarted = new Promise(resolve => { + signalFirstCommit = resolve; + }); + const firstCommitGate = new Promise(resolve => { + releaseFirstCommit = resolve; + }); + let commitCount = 0; + vi.mocked(ipc.invokeCommand).mockImplementation(async (command: string, args?: any) => { + if (command === 'db_commit_download_state') { + persistedIds.push((JSON.parse(args.downloadsData) as Array<{ id: string }>)[0]?.id || 'empty'); + commitCount += 1; + if (commitCount === 1) { + signalFirstCommit(); + await firstCommitGate; + } + return undefined; + } + return undefined; + }); + const first = { + id: 'commit-first', + url: 'https://example.com/first', + fileName: 'first.bin', + status: 'ready' as const, + category: 'Other' as const, + dateAdded: '' + }; + const second = { ...first, id: 'commit-second', fileName: 'second.bin' }; + + try { + useDownloadStore.setState({ downloads: [first] as any[] }); + await firstCommitStarted; + const committing = commitDownloadState(); + useDownloadStore.setState({ downloads: [second] as any[] }); + releaseFirstCommit(); + + await committing; + expect(persistedIds).toEqual(['commit-first', 'commit-second']); + } finally { + disposePersistence(); + } + }); + + it('does not leave an older in-flight snapshot after state returns to the committed value', async () => { + const disposePersistence = initializeDownloadPersistence('main'); + const persistedIds: string[] = []; + let releaseFirstCommit!: () => void; + let signalFirstCommit!: () => void; + const firstCommitStarted = new Promise(resolve => { + signalFirstCommit = resolve; + }); + const firstCommitGate = new Promise(resolve => { + releaseFirstCommit = resolve; + }); + let commitCount = 0; + vi.mocked(ipc.invokeCommand).mockImplementation(async (command: string, args?: any) => { + if (command === 'db_commit_download_state') { + const records = JSON.parse(args.downloadsData) as Array<{ id: string }>; + persistedIds.push(records[0]?.id || 'empty'); + commitCount += 1; + if (commitCount === 1) { + signalFirstCommit(); + await firstCommitGate; + } + return undefined; + } + return undefined; + }); + + try { + const first = { + id: 'snapshot-a', + url: 'https://example.com/a', + fileName: 'a.bin', + status: 'ready' as const, + category: 'Other' as const, + dateAdded: '' + }; + const second = { ...first, id: 'snapshot-b', fileName: 'b.bin' }; + useDownloadStore.setState({ downloads: [first] as any[] }); + await firstCommitStarted; + useDownloadStore.setState({ downloads: [second] as any[] }); + useDownloadStore.setState({ downloads: [first] as any[] }); + + releaseFirstCommit(); + await flushDownloadPersistence(); + + expect(persistedIds).toEqual(['snapshot-a', 'snapshot-a']); + } finally { + disposePersistence(); + } + }); + + it('waits for durable queued state before resuming an existing lifecycle', async () => { + useDownloadStore.setState({ + downloads: [{ + id: 'durable-resume', + url: 'https://example.com/resume.bin', + fileName: 'resume.bin', + status: 'paused', + category: 'Other', + dateAdded: '', + queueId: 'main' + }] as any[] + }); + const disposePersistence = initializeDownloadPersistence('main'); + let releaseCommit!: () => void; + let signalCommitStarted!: () => void; + const commitStarted = new Promise(resolve => { + signalCommitStarted = resolve; + }); + const commitGate = new Promise(resolve => { + releaseCommit = resolve; + }); + vi.mocked(ipc.invokeCommand).mockImplementation(async (command: string) => { + if (command === 'db_commit_download_state') { + signalCommitStarted(); + await commitGate; + return undefined; + } + if (command === 'resume_download') return true; + return undefined; + }); + + try { + const resuming = useDownloadStore.getState().resumeDownload('durable-resume'); + await commitStarted; + expect(ipc.invokeCommand).not.toHaveBeenCalledWith('resume_download', expect.anything()); + + releaseCommit(); + await expect(resuming).resolves.toBe(true); + expect(ipc.invokeCommand).toHaveBeenCalledWith('resume_download', { + id: 'durable-resume', + queueId: 'main' + }); + } finally { + disposePersistence(); + } + }); + it('normalizes new Torrent rows before resolving their default destination', async () => { await useDownloadStore.getState().addDownload({ id: 'torrent-default', diff --git a/src/store/useDownloadStore.ts b/src/store/useDownloadStore.ts index d7c38a3..09f470c 100644 --- a/src/store/useDownloadStore.ts +++ b/src/store/useDownloadStore.ts @@ -407,6 +407,15 @@ async function dispatchItemInternal(id: string, proxyOverride?: string | null): useDownloadStore.getState().updateDownload(id, { lastTry: new Date().toISOString() }); + await commitDownloadState(); + const admittedItem = useDownloadStore.getState().downloads.find(download => download.id === id); + if ( + !admittedItem || + !isCurrentDownloadLifecycle(id, lifecycleGeneration) || + !['ready', 'staged', 'failed', 'queued'].includes(admittedItem.status) + ) { + return false; + } const accepted = await invoke('enqueue_download', { item: enqueueItem }); backendAccepted = true; if (!isCurrentDownloadLifecycle(id, lifecycleGeneration)) { @@ -1055,6 +1064,22 @@ export const useDownloadStore = create((set, get) => { const state = get(); const item = state.downloads.find(d => d.id === id); if (!item) return; + const previousItem = item; + const commitProperties = async (): Promise => { + try { + await commitDownloadState(); + } catch (error) { + // Do not leave a renderer-only Properties edit that will disappear on + // restart. Restore the prior row while retaining the native lifecycle + // fencing already performed for this operation. + set(current => ({ + downloads: current.downloads.map(download => + download.id === id ? previousItem : download + ) + })); + throw error; + } + }; const credentialsUpdated = (['password', 'cookies', 'headers'] as const) .some(field => Object.prototype.hasOwnProperty.call(updates, field)); const nextCredentialMaterial = (['password', 'cookies', 'headers'] as const) @@ -1084,6 +1109,7 @@ export const useDownloadStore = create((set, get) => { await invoke('clear_torrent_removal_paths', { id }); } state.updateDownload(id, normalizedUpdates); + await commitProperties(); return; } @@ -1100,6 +1126,7 @@ export const useDownloadStore = create((set, get) => { await invoke('clear_torrent_removal_paths', { id }); } state.updateDownload(id, normalizedUpdates); + await commitProperties(); if (isRegistered || wasDispatching) { const dispatched = await dispatchItemInternal(id); if (dispatched) { @@ -1125,6 +1152,7 @@ export const useDownloadStore = create((set, get) => { await invoke('clear_torrent_removal_paths', { id }); } state.updateDownload(id, normalizedUpdates); + await commitProperties(); } }; @@ -1196,6 +1224,21 @@ export const useDownloadStore = create((set, get) => { if (currentTargetItem.status === 'ready' || currentTargetItem.status === 'staged') { get().updateDownload(id, { status: 'queued', hasBeenDispatched: true }); + try { + await commitDownloadState(); + } catch (error) { + get().updateDownload(id, { + status: currentTargetItem.status, + lastError: errorMessage(error) + }); + clearDownloadControlIntent(id, 'resume'); + return false; + } + const queuedItem = get().downloads.find(download => download.id === id); + if (!queuedItem || queuedItem.status !== 'queued') { + clearDownloadControlIntent(id, 'resume'); + return false; + } if (await dispatchItemInternal(id)) { return true; } @@ -1221,6 +1264,23 @@ export const useDownloadStore = create((set, get) => { lastTry: new Date().toISOString() }); + try { + await commitDownloadState(); + } catch (error) { + get().updateDownload(id, { + status: prevStatus, + lastError: errorMessage(error) + }); + clearDownloadControlIntent(id, 'resume'); + return false; + } + + const queuedItem = get().downloads.find(download => download.id === id); + if (!queuedItem || queuedItem.status !== 'queued') { + clearDownloadControlIntent(id, 'resume'); + return false; + } + const resumedExisting = options.forceRequeue ? false : await invoke('resume_download', { @@ -1649,6 +1709,18 @@ export const useDownloadStore = create((set, get) => { downloads: reorderQueueWithPausedAtEnd([...state.downloads, ownedItem], queueId) })); + try { + // Admission must not reach Aria2 or yt-dlp before the row and its queue + // position are committed. If the process dies after this point, startup + // recovery still has an authoritative row to resume. + await commitDownloadState(); + } catch (error) { + const message = errorMessage(error); + console.error(`Failed to persist download ${item.id} before admission:`, error); + get().updateDownload(item.id, { status: 'failed', lastError: message }); + return false; + } + if (action.type === 'add-to-queue') { info(`Download ${item.id} added to queue ${action.queueId}`); return true; @@ -1749,6 +1821,25 @@ export const useDownloadStore = create((set, get) => { Array.from(state.backendRegisteredIds).filter(registeredId => registeredId !== id) ) })); + try { + await commitDownloadState(); + } catch (error) { + const message = errorMessage(error); + console.error(`Failed to persist removal of ${id}:`, error); + if (item) { + set(state => state.downloads.some(download => download.id === id) + ? {} + : { + downloads: [...state.downloads, { + ...item, + status: 'failed' as const, + lastError: message, + hasBeenDispatched: false + }] + }); + } + throw error; + } useDownloadProgressStore.getState().clearDownloadProgress(id); info(`Download ${id} removed`); syncSystemIntegrations(); @@ -1772,6 +1863,7 @@ export const useDownloadStore = create((set, get) => { const current = get().downloads.find(download => download.id === id); if (current && current.status !== 'completed' && current.status !== 'failed') { get().updateDownload(id, { status: 'paused', speed: '-', eta: '-' }); + await commitDownloadState(); } } finally { clearDownloadControlIntent(id, 'pause'); @@ -1818,6 +1910,8 @@ export const useDownloadStore = create((set, get) => { dateAdded: new Date().toISOString() }); + await commitDownloadState(); + if (!await dispatchItemInternal(id)) { console.error("Failed to enqueue redownload"); get().updateDownload(id, { status: 'failed' }); @@ -2154,6 +2248,7 @@ export const useDownloadStore = create((set, get) => { ); return { downloads: reorderQueueWithPausedAtEnd(downloads, queueId) }; }); + await commitDownloadState(); }); }, setDownloadSpeedLimit: (id, limit) => runDownloadLifecycleOperation( @@ -2363,6 +2458,7 @@ export const useDownloadStore = create((set, get) => { selectedQueueIds }); } + await commitDownloadState(); }, resumePendingDownloads: () => { if (pendingStartupResume) return pendingStartupResume; @@ -2382,6 +2478,10 @@ export const useDownloadStore = create((set, get) => { : download) })); } + // Startup converts interrupted active lifecycles into queued rows before + // rebuilding backend ownership. Commit that recovery state first so a + // crash during enqueue_many cannot lose the restartable row. + await commitDownloadState(); const active = get().downloads .filter(d => d.status === 'queued') .sort((a, b) => (a.queuePosition ?? 0) - (b.queuePosition ?? 0)); @@ -2485,7 +2585,7 @@ export const useDownloadStore = create((set, get) => { } const currentItems = new Map(get().downloads.map(item => [item.id, item])); - const dispatchableItems = itemsToEnqueue.filter(item => { + let dispatchableItems = itemsToEnqueue.filter(item => { const current = currentItems.get(item.id); return current && current.status === 'queued' && @@ -2495,6 +2595,17 @@ export const useDownloadStore = create((set, get) => { }); if (dispatchableItems.length === 0) return; + await commitDownloadState(); + const latestItems = new Map(get().downloads.map(item => [item.id, item])); + dispatchableItems = dispatchableItems.filter(item => { + const current = latestItems.get(item.id); + return current && + current.status === 'queued' && + !get().backendRegisteredIds.has(item.id) && + !backendDispatchPromises.has(item.id) && + currentDownloadLifecycle(item.id).toString() === item.lifecycle_generation; + }); + if (dispatchableItems.length === 0) return; const results = await invoke('enqueue_many', { items: dispatchableItems }); const registeredIds = results.filter(result => result.success).map(result => result.id); const failedErrors = new Map( @@ -2665,43 +2776,137 @@ export const useDownloadStore = create((set, get) => { }; }); -let lastSavedDownloads = ''; -let isSavingDownloads = false; -let nextDownloadsData: string | null = null; +type PersistenceSnapshot = { + key: string; + downloadsData: string; + queuesData: string; + revision: number; +}; -async function processDownloadsSave() { - if (isSavingDownloads || !nextDownloadsData) return; - isSavingDownloads = true; - while (nextDownloadsData) { - const data = nextDownloadsData; - nextDownloadsData = null; - try { - await invoke('db_replace_downloads', { data }); - } catch (error) { - console.error('Failed to persist downloads:', error); +type PersistenceWaiter = { + revision: number; + resolve: () => void; + reject: (error: unknown) => void; +}; + +let persistenceRevision = 0; +let committedPersistenceRevision = 0; +let lastRequestedPersistenceKey: string | null = null; +let lastCommittedPersistenceKey: string | null = null; +let nextPersistenceSnapshot: PersistenceSnapshot | null = null; +let persistenceSaveInFlight = false; +let persistenceWaiters: PersistenceWaiter[] = []; +let downloadPersistenceReady = false; + +const persistenceSnapshotForState = (state: Pick): Omit => { + // Strip secret fields (password/cookies/headers) and volatile progress + // before writing to disk. Secrets remain on the in-memory item for the + // active session only. + const downloadsData = JSON.stringify(state.downloads.map(redactDownloadForPersistence)); + const queuesData = JSON.stringify(state.queues); + return { + key: JSON.stringify([downloadsData, queuesData]), + downloadsData, + queuesData + }; +}; + +const waitForPersistenceRevision = (revision: number): Promise => { + if (revision <= committedPersistenceRevision) return Promise.resolve(); + return new Promise((resolve, reject) => { + persistenceWaiters.push({ revision, resolve, reject }); + }); +}; + +const settlePersistenceWaiters = (revision: number, error?: unknown): void => { + const remaining: PersistenceWaiter[] = []; + for (const waiter of persistenceWaiters) { + if (waiter.revision > revision) { + remaining.push(waiter); + continue; } + if (error === undefined) waiter.resolve(); + else waiter.reject(error); + } + persistenceWaiters = remaining; +}; + +const queuePersistenceSnapshot = (snapshot: Omit): Promise => { + const hasUncommittedPersistence = persistenceSaveInFlight || nextPersistenceSnapshot !== null; + if (snapshot.key === lastCommittedPersistenceKey && !hasUncommittedPersistence) { + return Promise.resolve(); + } + + const existingRevision = snapshot.key === lastRequestedPersistenceKey + ? persistenceRevision + : null; + if (existingRevision !== null) return waitForPersistenceRevision(existingRevision); + + const revision = ++persistenceRevision; + lastRequestedPersistenceKey = snapshot.key; + nextPersistenceSnapshot = { ...snapshot, revision }; + const completion = waitForPersistenceRevision(revision); + void processPersistenceSave(); + return completion; +}; + +async function processPersistenceSave(): Promise { + if (persistenceSaveInFlight) return; + persistenceSaveInFlight = true; + try { + while (nextPersistenceSnapshot) { + const snapshot = nextPersistenceSnapshot; + nextPersistenceSnapshot = null; + try { + await invoke('db_commit_download_state', { + downloadsData: snapshot.downloadsData, + queuesData: snapshot.queuesData + }); + lastCommittedPersistenceKey = snapshot.key; + committedPersistenceRevision = snapshot.revision; + settlePersistenceWaiters(snapshot.revision); + } catch (error) { + if (lastRequestedPersistenceKey === snapshot.key) { + lastRequestedPersistenceKey = null; + } + settlePersistenceWaiters(snapshot.revision, error); + console.error('Failed to persist download state:', error); + } + } + } finally { + persistenceSaveInFlight = false; + if (nextPersistenceSnapshot) void processPersistenceSave(); } - isSavingDownloads = false; } -let lastSavedQueues = ''; -let isSavingQueues = false; -let nextQueuesData: string | null = null; - -async function processQueuesSave() { - if (isSavingQueues || !nextQueuesData) return; - isSavingQueues = true; - while (nextQueuesData) { - const data = nextQueuesData; - nextQueuesData = null; - try { - await invoke('db_replace_queues', { data }); - } catch (error) { - console.error('Failed to persist queues:', error); +export const commitDownloadState = async (): Promise => { + if (!downloadPersistenceReady) return; + while (true) { + const snapshot = persistenceSnapshotForState(useDownloadStore.getState()); + await queuePersistenceSnapshot(snapshot); + const current = persistenceSnapshotForState(useDownloadStore.getState()); + if ( + current.key === snapshot.key && + current.key === lastCommittedPersistenceKey && + !persistenceSaveInFlight && + nextPersistenceSnapshot === null + ) { + return; } } - isSavingQueues = false; -} +}; + +export const flushDownloadPersistence = async (): Promise => { + if (!downloadPersistenceReady) return; + while (true) { + const snapshot = persistenceSnapshotForState(useDownloadStore.getState()); + if (snapshot.key === lastCommittedPersistenceKey) return; + await queuePersistenceSnapshot(snapshot); + if (persistenceSnapshotForState(useDownloadStore.getState()).key === lastCommittedPersistenceKey) { + return; + } + } +}; let downloadPersistenceUnsubscribe: (() => void) | null = null; @@ -2714,31 +2919,17 @@ export const initializeDownloadPersistence = (windowLabel: string): (() => void) if (windowLabel !== 'main' || downloadPersistenceUnsubscribe) return () => undefined; downloadPersistenceUnsubscribe = useDownloadStore.subscribe((state, prevState) => { - if (state.queues !== prevState.queues) { - const data = JSON.stringify(state.queues); - if (data !== lastSavedQueues) { - lastSavedQueues = data; - nextQueuesData = data; - void processQueuesSave(); - } - } - - if (state.downloads !== prevState.downloads) { - // Strip secret fields (password/cookies/headers) and volatile progress - // before writing to disk. Secrets remain on the in-memory item for the - // active session only. - const staticDownloads = state.downloads.map(redactDownloadForPersistence); - const currentSerialized = JSON.stringify(staticDownloads); - if (currentSerialized !== lastSavedDownloads) { - lastSavedDownloads = currentSerialized; - nextDownloadsData = currentSerialized; - void processDownloadsSave(); - } + if (state.queues !== prevState.queues || state.downloads !== prevState.downloads) { + void queuePersistenceSnapshot(persistenceSnapshotForState(state)).catch(error => { + console.error('Failed to persist download state:', error); + }); } }); + downloadPersistenceReady = true; return () => { downloadPersistenceUnsubscribe?.(); downloadPersistenceUnsubscribe = null; + downloadPersistenceReady = false; }; };