From bb618aef7d5a89a38bb0565e76944b103f068e6e Mon Sep 17 00:00:00 2001 From: NimBold Date: Tue, 16 Jun 2026 17:46:58 +0330 Subject: [PATCH] feat(queue): implement backend-driven download queue coordinator This commit replaces the frontend-imperative download dispatcher with a centralized backend `QueueManager`. It acts as the sole concurrency gatekeeper using a single `tokio::sync::Semaphore` across all download paths (aria2 RPC, native HTTP, yt-dlp media). Backend Changes: - **queue**: Added `QueueManager` to manage an ordered `VecDeque` of tasks, semaphore permits, and retirement debt (CAS resize). - **commands**: Replaced direct start commands with `enqueue_download`, `enqueue_many`, `move_in_queue`, and `remove_from_queue`. - **ipc**: Exported `DownloadStateEvent` and `QueueDirection` to the frontend. - **tests**: Added 11 integration tests covering idle-parking, idempotent releases, CAS underflow prevention, and gid-completion races. Frontend Changes: - **store**: Made `useDownloadStore` reactive to the backend via the `download-state` event. - **store**: Removed `processQueue` and introduced `pendingOrder` to track the accurate sequence of queued items. - **ui**: Updated `DownloadItem` with queue visuals (clock icon, position badge). - **ui**: Added Move Up/Down controls to interact with the backend queue reordering API. --- src-tauri/Cargo.lock | 1 + src-tauri/Cargo.toml | 3 +- src-tauri/src/ipc.rs | 37 +- src-tauri/src/lib.rs | 388 +++++++--------- src-tauri/src/queue.rs | 680 +++++++++++++++++++++++++++++ src-tauri/src/scheduler.rs | 2 +- src-tauri/tests/queue_manager.rs | 329 ++++++++++++++ src/App.tsx | 17 +- src/bindings/DownloadItem.ts | 2 +- src/bindings/DownloadStateEvent.ts | 3 + src/bindings/QueueDirection.ts | 3 + src/components/DownloadItem.tsx | 51 ++- src/components/DownloadTable.tsx | 6 +- src/ipc.ts | 5 + src/store/downloadStore.ts | 34 +- src/store/useDownloadStore.ts | 415 ++++++++++++------ 16 files changed, 1583 insertions(+), 393 deletions(-) create mode 100644 src-tauri/src/queue.rs create mode 100644 src-tauri/tests/queue_manager.rs create mode 100644 src/bindings/DownloadStateEvent.ts create mode 100644 src/bindings/QueueDirection.ts diff --git a/src-tauri/Cargo.lock b/src-tauri/Cargo.lock index cd6048a..93fb1d4 100644 --- a/src-tauri/Cargo.lock +++ b/src-tauri/Cargo.lock @@ -1349,6 +1349,7 @@ checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" name = "firelink" version = "0.7.3" dependencies = [ + "async-trait", "axum", "chrono", "cocoa", diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml index 014aad4..0e082f9 100644 --- a/src-tauri/Cargo.toml +++ b/src-tauri/Cargo.toml @@ -23,7 +23,7 @@ crate-type = ["staticlib", "cdylib", "rlib"] tauri-build = { version = "2", features = [] } [dependencies] -tauri = { version = "2", features = ["macos-private-api", "tray-icon", "image-png"] } +tauri = { version = "2", features = ["macos-private-api", "tray-icon", "image-png", "test"] } tauri-plugin-opener = "2" tauri-plugin-dialog = "2" tauri-plugin-shell = "2" @@ -57,6 +57,7 @@ tauri-plugin-store = "2.4.3" log = "0.4.32" tauri-plugin-log = "2" trash = "5" +async-trait = "0.1" [target.'cfg(target_os = "macos")'.dependencies] cocoa = "0.25" objc = "0.2.7" diff --git a/src-tauri/src/ipc.rs b/src-tauri/src/ipc.rs index ec80c32..4e4077d 100644 --- a/src-tauri/src/ipc.rs +++ b/src-tauri/src/ipc.rs @@ -87,9 +87,6 @@ pub struct DownloadItem { #[ts(optional)] pub media_format_selector: Option, pub queue_id: String, - #[serde(rename = "_dispatched")] - #[ts(optional)] - pub dispatched: Option, } #[derive(Clone, Debug, Serialize, Deserialize, TS)] @@ -234,3 +231,37 @@ pub struct PersistedSettings { pub extension_pairing_token: String, pub auto_check_updates: bool, } + +#[derive(Clone, Copy, Debug, Serialize, Deserialize, TS)] +#[serde(rename_all = "lowercase")] +#[ts(export, export_to = "../../src/bindings/")] +pub enum QueueDirection { + Up, + Down, +} + +#[derive(Clone, Debug, Serialize, TS)] +#[ts(export, export_to = "../../src/bindings/")] +pub struct DownloadStateEvent { + pub id: String, + pub status: String, + pub error: Option, +} + +impl DownloadStateEvent { + pub fn new(id: impl Into, status: DownloadStatus) -> Self { + Self { + id: id.into(), + status: status.as_str().to_string(), + error: None, + } + } + + pub fn failed(id: impl Into, error: impl Into) -> Self { + Self { + id: id.into(), + status: DownloadStatus::Failed.as_str().to_string(), + error: Some(error.into()), + } + } +} diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 5af4c74..fe849bf 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -343,7 +343,7 @@ async fn test_deno(app_handle: tauri::AppHandle) -> Result { } } -fn is_safe_path(path: &std::path::Path, app_handle: &tauri::AppHandle) -> bool { +pub(crate) fn is_safe_path(path: &std::path::Path, app_handle: &tauri::AppHandle) -> bool { if path.components().any(|c| matches!(c, std::path::Component::ParentDir)) { return false; } @@ -419,8 +419,9 @@ impl Drop for Aria2DaemonGuard { pub mod download; +pub mod queue; #[allow(dead_code)] -mod ipc; +pub mod ipc; mod parity; pub mod error; pub mod commands; @@ -441,7 +442,7 @@ pub struct AppState { pub aria2_secret: String, pub media_semaphore: Arc, pub sleep_preventer: Arc>>, - pub aria2_gids: Arc>>, + pub queue_manager: Arc, } #[derive(Clone, Serialize, TS)] @@ -455,7 +456,7 @@ pub struct DownloadProgressEvent { } -fn resolve_path(path: &str, app_handle: &tauri::AppHandle) -> std::path::PathBuf { +pub(crate) fn resolve_path(path: &str, app_handle: &tauri::AppHandle) -> std::path::PathBuf { use tauri::Manager; let mut resolved = std::path::PathBuf::from(path); if let Some(stripped) = path.strip_prefix("~/") { @@ -470,7 +471,7 @@ fn resolve_path(path: &str, app_handle: &tauri::AppHandle) -> std::path::PathBuf resolved } -fn collect_download_uris(url: &str, mirrors: Option<&str>) -> Vec { +pub(crate) fn collect_download_uris(url: &str, mirrors: Option<&str>) -> Vec { let mut uris = Vec::new(); for uri in std::iter::once(url).chain(mirrors.into_iter().flat_map(str::lines)) { let uri = uri.trim(); @@ -549,7 +550,7 @@ fn dispatch_deep_links(app_handle: tauri::AppHandle, deep_links: Vec) }); } -async fn rpc_call(port: u16, secret: &str, method: &str, params: serde_json::Value) -> Result { +pub(crate) async fn rpc_call(port: u16, secret: &str, method: &str, params: serde_json::Value) -> Result { let url = format!("http://127.0.0.1:{}/jsonrpc", port); let mut payload = serde_json::Map::new(); payload.insert("jsonrpc".to_string(), serde_json::json!("2.0")); @@ -596,180 +597,6 @@ async fn test_aria2c(state: tauri::State<'_, AppState>) -> Result, - id: String, - url: String, - destination: String, - filename: String, - connections: Option, - speed_limit: Option, - username: Option, - password: Option, - headers: Option, - checksum: Option, - cookies: Option, - mirrors: Option, - user_agent: Option, - max_tries: Option, - proxy: Option, -) -> Result<(), AppError> { - println!("start_download called for id: {}", id); - let download_id = Uuid::parse_str(&id).map_err(|error| AppError::Internal(error.to_string()))?; - - let safe_filename = std::path::Path::new(&filename.replace('\\', "/")) - .file_name() - .and_then(|n| n.to_str()) - .unwrap_or("download") - .to_string(); - - let resolved_dest = resolve_path(&destination, &app_handle); - if !is_safe_path(&resolved_dest, &app_handle) { - return Err(AppError::Internal("Path traversal blocked".to_string())); - } - - let mt = max_tries.unwrap_or(1).max(1) as u32; - - let mut options = serde_json::Map::new(); - options.insert("dir".to_string(), serde_json::json!(resolved_dest.to_string_lossy().to_string())); - options.insert("out".to_string(), serde_json::json!(safe_filename)); - - let conn = connections.unwrap_or(1); - options.insert("split".to_string(), serde_json::json!(conn.to_string())); - options.insert("max-connection-per-server".to_string(), serde_json::json!(conn.to_string())); - options.insert("max-tries".to_string(), serde_json::json!(mt.to_string())); - - options.insert("continue".to_string(), serde_json::json!("true")); - - if let Some(speed) = &speed_limit { - options.insert("max-download-limit".to_string(), serde_json::json!(speed)); - } - if let Some(user) = &username { - options.insert("http-user".to_string(), serde_json::json!(user)); - } - if let Some(pass) = &password { - options.insert("http-passwd".to_string(), serde_json::json!(pass)); - } - if let Some(chk) = &checksum { - options.insert("checksum".to_string(), serde_json::json!(chk)); - } - if let Some(ua) = &user_agent { - options.insert("user-agent".to_string(), serde_json::json!(ua)); - } - - let mut header_list = Vec::new(); - if let Some(cook) = &cookies { - header_list.push(format!("Cookie: {}", cook)); - } - if let Some(hdrs) = &headers { - for line in hdrs.lines() { - if !line.trim().is_empty() { - header_list.push(line.trim().to_string()); - } - } - } - if !header_list.is_empty() { - options.insert("header".to_string(), serde_json::json!(header_list)); - } - - if let Some(prox) = &proxy { - options.insert("all-proxy".to_string(), serde_json::json!(prox)); - } - - let uris = collect_download_uris(&url, mirrors.as_deref()); - let params = serde_json::json!([uris, options]); - - match rpc_call(state.aria2_port, &state.aria2_secret, "aria2.addUri", params).await { - Ok(result) => { - let gid = result.as_str().unwrap_or("").to_string(); - state.aria2_gids.write().unwrap().insert(id.clone(), gid); - Ok(()) - } - Err(e) => { - eprintln!("aria2 failed, falling back to native coordinator: {}", e); - state - .download_coordinator - .send(download::DownloadCmd::Start(Box::new(download::DownloadPayload { - id: download_id, - urls: collect_download_uris(&url, mirrors.as_deref()), - output_path: resolved_dest.join(safe_filename), - speed_limit, - username, - password, - headers, - cookies, - user_agent, - max_tries: mt, - proxy, - }))) - .await - .map_err(AppError::Internal)?; - Ok(()) - } - } -} - -#[allow(clippy::too_many_arguments)] -#[tauri::command] -async fn start_media_download( - app_handle: tauri::AppHandle, - state: tauri::State<'_, AppState>, - id: String, - url: String, - destination: String, - filename: String, - format_selector: Option, - cookie_source: Option, - speed_limit: Option, - username: Option, - password: Option, - headers: Option, - proxy: Option, - user_agent: Option, - max_tries: Option, -) -> Result<(), String> { - let media_semaphore = state.media_semaphore.clone(); - let coordinator = state.download_coordinator.clone(); - let mut cancel_rx = coordinator.register_media(id.clone()).await?; - tauri::async_runtime::spawn(async move { - let permit = tokio::select! { - permit = media_semaphore.acquire() => permit, - _ = cancel_rx.changed() => { - coordinator.finish_media(id).await; - return; - } - }; - - if let Err(e) = start_media_download_internal( - app_handle.clone(), - &id, url, - destination, - filename, - format_selector, - cookie_source, - speed_limit, - username, - password, - headers, - proxy, - user_agent, - max_tries, - &mut cancel_rx, - ).await { - eprintln!("Media download {} failed: {}", id, e); - use tauri::Emitter; - let _ = app_handle.emit("download-failed", id.clone()); - } - - drop(permit); - coordinator.finish_media(id).await; - }); - - Ok(()) -} #[allow(clippy::too_many_arguments)] pub(crate) async fn start_media_download_internal( @@ -1009,12 +836,32 @@ pub(crate) async fn start_media_download_internal( } #[tauri::command] -async fn pause_download(state: tauri::State<'_, AppState>, id: String) -> Result<(), String> { +async fn pause_download( + app_handle: tauri::AppHandle, + state: tauri::State<'_, AppState>, + id: String, +) -> Result<(), String> { println!("pause_download called for id: {}", id); - - let gid = state.aria2_gids.read().unwrap().get(&id).cloned(); + + // Release the concurrency slot FIRST, before signaling the sidecar to die. + state.queue_manager.release_permit(&id).await; + state.queue_manager.remove_from_pending(&id).await; + + // Emit the paused state. + use tauri::Emitter; + let _ = app_handle.emit( + "download-state", + crate::ipc::DownloadStateEvent::new(id.clone(), crate::ipc::DownloadStatus::Paused), + ); + + let gid = state.queue_manager.aria2_gids.read().unwrap() + .iter() + .find(|(_, v)| **v == id) + .map(|(k, _)| k.clone()); if let Some(g) = gid { - let _ = rpc_call(state.aria2_port, &state.aria2_secret, "aria2.pause", serde_json::json!([g])).await; + if !g.starts_with("native:") { + let _ = rpc_call(state.aria2_port, &state.aria2_secret, "aria2.pause", serde_json::json!([g])).await; + } } if let Ok(download_id) = Uuid::parse_str(&id) { @@ -1027,17 +874,32 @@ async fn pause_download(state: tauri::State<'_, AppState>, id: String) -> Result } #[tauri::command] -async fn remove_download(app_handle: tauri::AppHandle, state: tauri::State<'_, AppState>, id: String, filepath: Option) -> Result<(), String> { +async fn remove_download( + app_handle: tauri::AppHandle, + state: tauri::State<'_, AppState>, + id: String, + filepath: Option, +) -> Result<(), String> { println!("remove_download called for id: {}", id); + // Remove from the queue (pending or active) and free the slot. + state.queue_manager.remove_from_pending(&id).await; + state.queue_manager.release_permit(&id).await; + + use tauri::Emitter; + let _ = app_handle.emit( + "download-state", + crate::ipc::DownloadStateEvent::new(id.clone(), crate::ipc::DownloadStatus::Paused), + ); + if let Ok(download_id) = Uuid::parse_str(&id) { state .download_coordinator .send(download::DownloadCmd::Cancel(download_id)) .await?; } - state.download_coordinator.pause_media(id).await?; - + state.download_coordinator.pause_media(id.clone()).await?; + if let Some(path) = filepath { if !path.is_empty() { let p = std::path::Path::new(&path); @@ -1053,7 +915,7 @@ async fn remove_download(app_handle: tauri::AppHandle, state: tauri::State<'_, A } } } - + Ok(()) } @@ -1110,17 +972,49 @@ fn perform_system_action(action: crate::ipc::PostQueueAction) -> Result<(), Stri execute_system_action(action) } +#[tauri::command] +async fn get_pending_order(state: tauri::State<'_, AppState>) -> Result, AppError> { + Ok(state.queue_manager.pending_order().await) +} + +#[tauri::command] +async fn enqueue_download( + state: tauri::State<'_, AppState>, + item: queue::EnqueueItem, +) -> Result { + let id = item.id.clone(); + state.queue_manager.push(item.into_task()).await; + Ok(id) +} + +#[tauri::command] +async fn enqueue_many( + state: tauri::State<'_, AppState>, + items: Vec, +) -> Result<(), AppError> { + let tasks = items.into_iter().map(queue::EnqueueItem::into_task).collect(); + state.queue_manager.enqueue_many(tasks).await; + Ok(()) +} + +#[tauri::command] +async fn move_in_queue( + state: tauri::State<'_, AppState>, + id: String, + direction: crate::ipc::QueueDirection, +) -> Result, AppError> { + Ok(state.queue_manager.move_in_queue(&id, direction).await) +} + +#[tauri::command] +async fn remove_from_queue(state: tauri::State<'_, AppState>, id: String) -> Result { + Ok(state.queue_manager.remove_from_pending(&id).await) +} + #[tauri::command] async fn set_concurrent_limit(state: tauri::State<'_, AppState>, limit: usize) -> Result<(), String> { - rpc_call( - state.aria2_port, - &state.aria2_secret, - "aria2.changeGlobalOption", - serde_json::json!([{"max-concurrent-downloads": limit.to_string()}]) - ).await.map(|_| ()).map_err(|e| { - eprintln!("Failed to set concurrent limit: {}", e); - e - }) + state.queue_manager.set_capacity(limit); + Ok(()) } #[tauri::command] @@ -1457,10 +1351,33 @@ pub fn run() { .build(app) .unwrap(); - let aria2_gids = Arc::new(RwLock::new(std::collections::HashMap::new())); - let aria2_gids_clone1 = aria2_gids.clone(); - let aria2_gids_clone2 = aria2_gids.clone(); + let max_concurrent = { + use tauri_plugin_store::StoreExt; + let mut capacity = crate::queue::DEFAULT_MAX_CONCURRENT; + if let Ok(store) = app.handle().store("store.bin") { + if let Some(settings_val) = store.get("settings") { + if let Some(settings_str) = settings_val.as_str() { + if let Ok(settings_json) = serde_json::from_str::(settings_str) { + if let Some(n) = settings_json.get("maxConcurrentDownloads").and_then(|v| v.as_u64()) { + capacity = n as usize; + } + } + } + } + } + capacity + }; + + let queue_manager = Arc::new(queue::QueueManager::new(app.handle().clone(), max_concurrent)); + let dispatcher_mgr = Arc::clone(&queue_manager); + tauri::async_runtime::spawn(async move { + dispatcher_mgr.run_dispatcher().await; + }); + + let queue_manager_clone = Arc::clone(&queue_manager); + let queue_manager_poll = Arc::clone(&queue_manager); + app.manage(AppState { download_coordinator: download::DownloadCoordinator::spawn(app.handle().clone()), extension_pairing_token, @@ -1469,8 +1386,42 @@ pub fn run() { aria2_secret: aria2_secret.clone(), media_semaphore: Arc::new(tokio::sync::Semaphore::new(3)), sleep_preventer: Arc::new(Mutex::new(None)), - aria2_gids, + queue_manager, }); + + // Backend listener: release permits + emit terminal state for + // native (and aria2-fallback) downloads. Idempotent for Media/aria2 + // which already release via finish_runner/handle_aria2_event. + let completion_app = app.handle().clone(); + let completion_mgr = Arc::clone(&queue_manager_clone); + tauri::async_runtime::spawn(async move { + use tauri::Listener; + let rx_complete = completion_app.listen("download-complete", move |event| { + let raw_id = event.payload(); + let id: String = serde_json::from_str(raw_id) + .unwrap_or_else(|_| raw_id.trim_matches('"').to_string()); + let mgr = Arc::clone(&completion_mgr); + tauri::async_runtime::spawn(async move { + mgr.apply_completion(&id, crate::queue::PendingOutcome::Complete).await; + }); + }); + let completion_app2 = completion_app.clone(); + let completion_mgr2 = Arc::clone(&queue_manager_clone); + let rx_failed = completion_app2.listen("download-failed", move |event| { + let raw_id = event.payload(); + let id: String = serde_json::from_str(raw_id) + .unwrap_or_else(|_| raw_id.trim_matches('"').to_string()); + let mgr = Arc::clone(&completion_mgr2); + tauri::async_runtime::spawn(async move { + mgr.apply_completion(&id, crate::queue::PendingOutcome::Error("download failed".to_string())).await; + }); + }); + // Keep the task alive; the listeners are unregistered on drop. + std::future::pending::<()>().await; + let _ = rx_complete; + let _ = rx_failed; + }); + let deep_link_app = app.handle().clone(); app.deep_link().on_open_url(move |event| { dispatch_deep_links(deep_link_app.clone(), event.urls()); @@ -1581,23 +1532,17 @@ pub fn run() { if let Some(params) = json.get("params").and_then(|p| p.as_array()) { if let Some(event) = params.first().and_then(|p| p.as_object()) { if let Some(gid) = event.get("gid").and_then(|g| g.as_str()) { - let id = { - let map = aria2_gids_clone1.read().unwrap(); - map.iter().find(|(_, g)| *g == gid).map(|(i, _)| i.clone()) - }; - if let Some(id) = id { - use tauri::Emitter; - match method { - "aria2.onDownloadComplete" => { - let _ = app_handle_ws.emit("download-complete", id.clone()); - use tauri_plugin_notification::NotificationExt; - let _ = app_handle_ws.notification().builder().title("Download Complete").body(&format!("File downloaded successfully")).show(); - } - "aria2.onDownloadError" => { - let _ = app_handle_ws.emit("download-failed", id); - } - _ => {} + let state = app_handle_ws.state::(); + let outcome = match method { + "aria2.onDownloadComplete" => Some(crate::queue::PendingOutcome::Complete), + "aria2.onDownloadError" => { + let msg = event.get("error_message").and_then(|m| m.as_str()).unwrap_or("aria2 download error").to_string(); + Some(crate::queue::PendingOutcome::Error(msg)) } + _ => None, + }; + if let Some(outcome) = outcome { + state.queue_manager.handle_aria2_event(gid, outcome).await; } } } @@ -1615,6 +1560,7 @@ pub fn run() { let app_handle_poll = app.handle().clone(); let poll_port = aria2_port; let poll_secret = aria2_secret.clone(); + let poll_mgr = Arc::clone(&queue_manager_poll); tauri::async_runtime::spawn(async move { let mut interval = tokio::time::interval(std::time::Duration::from_millis(1000)); loop { @@ -1624,10 +1570,7 @@ pub fn run() { if let Some(active_arr) = active_list.as_array() { for status_info in active_arr { let gid = status_info.get("gid").and_then(|s| s.as_str()).unwrap_or(""); - let id = { - let map = aria2_gids_clone2.read().unwrap(); - map.iter().find(|(_, g)| *g == gid).map(|(i, _)| i.clone()) - }; + let id = poll_mgr.aria2_gids.read().unwrap().get(gid).cloned(); if let Some(id) = id { 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); @@ -1699,12 +1642,13 @@ pub fn run() { }) .invoke_handler(tauri::generate_handler![ test_ytdlp, test_aria2c, test_ffmpeg, test_deno, open_file, show_in_folder, - start_download, start_media_download, pause_download, fetch_metadata, fetch_media_metadata, + pause_download, fetch_metadata, fetch_media_metadata, update_dock_badge, set_prevent_sleep, get_free_space, perform_system_action, request_automation_permission, open_automation_settings, set_keychain_password, get_keychain_password, delete_keychain_password, check_file_exists, delete_file, toggle_tray_icon, set_extension_pairing_token, set_extension_frontend_ready, set_concurrent_limit, set_global_speed_limit, remove_download, + enqueue_download, enqueue_many, move_in_queue, remove_from_queue, get_pending_order, commands::reveal_in_file_manager, commands::open_downloaded_file, commands::trash_download_assets, 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/queue.rs b/src-tauri/src/queue.rs new file mode 100644 index 0000000..ad9c839 --- /dev/null +++ b/src-tauri/src/queue.rs @@ -0,0 +1,680 @@ +use crate::ipc::{DownloadStateEvent, DownloadStatus, QueueDirection}; +use serde::Deserialize; +use std::collections::{HashMap, VecDeque}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; +use tauri::{AppHandle, Manager}; +use tokio::sync::{Mutex, Notify, OwnedSemaphorePermit, Semaphore}; +use serde_json; +use log; + +/// Default capacity when no setting is read yet. +pub const DEFAULT_MAX_CONCURRENT: usize = 3; + +/// Outcome of an aria2 completion that arrived before its gid was stored. +/// Carries the outcome so the correct state emit survives the race. +#[derive(Debug, Clone)] +pub enum PendingOutcome { + Complete, + Error(String), +} + +/// What kind of sidecar a queued task spawns. Drives which runner the +/// dispatcher invokes. +#[derive(Debug, Clone)] +pub enum TaskKind { + Aria2, + Media, + Native, +} + +/// Everything needed to start a sidecar, captured at enqueue time so the +/// dispatcher can spawn it later without round-tripping back to the frontend. +#[derive(Debug, Clone)] +pub struct QueuedTask { + pub id: String, + pub kind: TaskKind, + pub payload: SpawnPayload, +} + +/// Args mirroring start_download / start_media_download. Kept untyped-loose +/// (String/Option) to match the existing command signatures exactly. +#[derive(Debug, Clone, Default)] +pub struct SpawnPayload { + pub url: String, + pub destination: String, + pub filename: String, + pub connections: Option, + pub speed_limit: Option, + pub username: Option, + pub password: Option, + pub headers: Option, + pub checksum: Option, + pub cookies: Option, + pub mirrors: Option, + pub user_agent: Option, + pub max_tries: Option, + pub proxy: Option, + pub format_selector: Option, + pub cookie_source: Option, + pub is_media: bool, +} + +/// A sidecar spawner. In production this calls the real aria2/yt-dlp/native +/// runners; in tests it is replaced with a fake that records calls and +/// optionally hangs to simulate a long-running download. +#[async_trait::async_trait] +pub trait SidecarSpawner: Send + Sync + 'static { + /// Spawn an aria2 download. Returns the gid. Must return quickly (the + /// permit is already parked before this is called). + async fn add_uri(&self, id: &str, payload: &SpawnPayload) -> Result; + + /// 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>; + + /// Run a native HTTP download to completion. + async fn run_native(&self, id: &str, payload: &SpawnPayload) -> Result<(), String>; +} + +/// The centralized concurrency gatekeeper. One instance lives in AppState. +pub struct QueueManager { + pending: Mutex>, + semaphore: Arc, + active_permits: Mutex>, + target_capacity: AtomicUsize, + slots_to_retire: AtomicUsize, + notify: Notify, + + /// aria2 gid -> download id map (shared with the WS poller). + pub aria2_gids: Arc>>, + + /// gid -> buffered (id_placeholder, outcome) for completions that arrived + /// before the gid was stored. Drained by `remember_gid`. + pub pending_completion: Arc>>, + + spawner: Arc, + app_handle: AppHandle, +} + +impl QueueManager { + /// Production constructor. Wired up in lib.rs setup(). + pub fn new(app_handle: AppHandle, capacity: usize) -> Self { + let spawner: Arc = + Arc::new(ProductionSpawner::new(app_handle.clone())); + Self::test_new(app_handle, capacity, spawner) + } +} + +impl QueueManager { + + /// Test-only constructor injecting a fake spawner. + pub fn test_new( + app_handle: AppHandle, + capacity: usize, + spawner: Arc, + ) -> Self { + Self { + pending: Mutex::new(VecDeque::new()), + semaphore: Arc::new(Semaphore::new(capacity)), + active_permits: Mutex::new(HashMap::new()), + target_capacity: AtomicUsize::new(capacity), + slots_to_retire: AtomicUsize::new(0), + notify: Notify::new(), + aria2_gids: Arc::new(std::sync::RwLock::new(HashMap::new())), + pending_completion: Arc::new(Mutex::new(HashMap::new())), + spawner, + app_handle, + } + } + + /// Current pending order, as id list. Returned by move_in_queue. + pub async fn pending_order(&self) -> Vec { + self.pending + .lock() + .await + .iter() + .map(|t| t.id.clone()) + .collect() + } + + /// Enqueue a task. Notifies the dispatcher. Emits download-state{queued}. + pub async fn push(&self, task: QueuedTask) { + let id = task.id.clone(); + self.pending.lock().await.push_back(task); + self.emit_state(id, DownloadStatus::Queued); + self.notify.notify_one(); + } + + /// Pop the next task, or None if empty. + pub async fn pop_front(&self) -> Option { + self.pending.lock().await.pop_front() + } + + /// Acquire a permit from the semaphore (blocks until one is available). + pub async fn acquire_permit(&self) -> OwnedSemaphorePermit { + self.semaphore + .clone() + .acquire_owned() + .await + .expect("semaphore closed") + } + + /// Park an already-acquired permit under `id`. + pub async fn park_permit(&self, id: &str, permit: OwnedSemaphorePermit) { + self.active_permits + .lock() + .await + .insert(id.to_string(), permit); + } + + /// Release the permit parked under `id`, if any. Idempotent. Wakes the + /// dispatcher so a freed slot is claimed promptly. + pub async fn release_permit(&self, id: &str) { + let removed = self.active_permits.lock().await.remove(id).is_some(); + if removed { + self.notify.notify_one(); + } + } + + /// Number of un-acquired permits currently in the semaphore pool. + pub fn available_permits(&self) -> usize { + self.semaphore.available_permits() + } + + fn emit_state(&self, id: impl Into, status: DownloadStatus) { + use tauri::Emitter; + let _ = self.app_handle.emit( + "download-state", + DownloadStateEvent::new(id, status), + ); + } + + /// Resize the global concurrency limit. Grow adds permits immediately; + /// shrink records a retirement debt honored lazily by the dispatcher. + pub fn set_capacity(&self, new_target: usize) { + let prev_target = self.target_capacity.swap(new_target, Ordering::Relaxed); + if new_target == prev_target { + return; + } + if new_target > prev_target { + let mut delta = new_target - prev_target; + loop { + let debt = self.slots_to_retire.load(Ordering::Relaxed); + let to_deduct = std::cmp::min(debt, delta); + match self.slots_to_retire.compare_exchange_weak( + debt, + debt - to_deduct, + Ordering::Relaxed, + Ordering::Relaxed, + ) { + Ok(_) => { + delta -= to_deduct; + break; + } + Err(_) => {} + } + } + if delta > 0 { + self.semaphore.add_permits(delta); + } + self.notify.notify_one(); + } else { + let delta = prev_target - new_target; + self.slots_to_retire.fetch_add(delta, Ordering::Relaxed); + } + } + + /// Test accessor for the retirement debt counter. + pub fn slots_to_retire_load(&self) -> usize { + self.slots_to_retire.load(Ordering::Relaxed) + } + + /// The long-running dispatcher. One instance is spawned in setup(). + /// Idle-parks on Notify; CAS-honors retirement debt; re-pops under lock. + pub async fn run_dispatcher(self: Arc) { + loop { + // (1) Idle-park: avoid busy-spin when pending is empty. + if self.pending.lock().await.is_empty() { + self.notify.notified().await; + continue; + } + // (2) Acquire a slot. + let permit = self + .semaphore + .clone() + .acquire_owned() + .await + .expect("semaphore closed"); + // (3) CAS retirement — never underflows to usize::MAX. + let mut retired = false; + let mut debt = self.slots_to_retire.load(Ordering::Relaxed); + while debt > 0 { + match self.slots_to_retire.compare_exchange_weak( + debt, + debt - 1, + Ordering::Relaxed, + Ordering::Relaxed, + ) { + Ok(_) => { + retired = true; + break; + } + Err(actual) => { + debt = actual; + } + } + } + if retired { + drop(permit); + continue; + } + // (4) Re-pop under lock — guards against racing removals between + // waking from Notify and acquiring the permit. + let task = match self.pending.lock().await.pop_front() { + Some(t) => t, + None => { + drop(permit); + continue; + } + }; + Arc::clone(&self).dispatch_one(permit, task).await; + } + } + + async fn dispatch_one(self: Arc, permit: OwnedSemaphorePermit, task: QueuedTask) { + let id = task.id.clone(); + // Park the permit BEFORE spawning. Uniform parking: + // aria2's RPC returns instantly, so the permit must outlive the + // dispatch_one call. Media/Native runners release on exit. + self.park_permit(&id, permit).await; + self.emit_state(&id, DownloadStatus::Downloading); + + match task.kind { + TaskKind::Aria2 => { + match self.spawner.add_uri(&id, &task.payload).await { + Ok(gid) => self.remember_gid(id.clone(), gid).await, + Err(error) => { + self.emit_failed(&id, error); + self.release_permit(&id).await; + } + } + } + TaskKind::Media => { + let this = Arc::clone(&self); + let payload = task.payload.clone(); + let id_for_task = id.clone(); + tauri::async_runtime::spawn(async move { + let outcome = this.spawner.run_media(&id_for_task, &payload).await; + this.finish_runner(&id_for_task, outcome).await; + }); + } + TaskKind::Native => { + // Native coordinator is event-driven (fire-and-observe). Send + // Start; completion is handled by the download-complete/ + // download-failed listener in lib.rs setup() which calls + // release_permit + apply_completion. + let this = Arc::clone(&self); + let payload = task.payload.clone(); + let id_for_task = id.clone(); + tauri::async_runtime::spawn(async move { + if let Err(error) = this.spawner.run_native(&id_for_task, &payload).await { + this.emit_failed(&id_for_task, error); + this.release_permit(&id_for_task).await; + } + }); + } + } + } + + /// Called when a Media runner exits. Releases the permit and emits + /// the terminal state. + async fn finish_runner(self: Arc, id: &str, outcome: Result<(), String>) { + match outcome { + Ok(()) => { + self.emit_state(id, DownloadStatus::Completed); + } + Err(error) => { + self.emit_failed(id, error); + } + } + self.release_permit(id).await; + } + + fn emit_failed(&self, id: &str, error: String) { + use tauri::Emitter; + let _ = self + .app_handle + .emit("download-state", DownloadStateEvent::failed(id, error)); + } + + /// Store gid -> id, then reconcile any buffered completion for that gid. + pub async fn remember_gid(&self, id: String, gid: String) { + { + let mut gids = self.aria2_gids.write().unwrap(); + gids.insert(gid.clone(), id.clone()); + } + let buffered = self.pending_completion.lock().await.remove(&gid); + if let Some((_buf_id, outcome)) = buffered { + self.apply_completion(&id, outcome).await; + } + } + + /// Apply an aria2 completion outcome: release permit + emit state. + pub async fn apply_completion(&self, id: &str, outcome: PendingOutcome) { + match outcome { + PendingOutcome::Complete => { + self.emit_state(id, DownloadStatus::Completed); + } + PendingOutcome::Error(error) => { + self.emit_failed(id, error); + } + } + self.release_permit(id).await; + } + + /// Entry point for the aria2 WS poller. Resolves gid -> id; if not yet + /// stored, buffers the outcome for reconciliation by remember_gid. + pub async fn handle_aria2_event(&self, gid: &str, outcome: PendingOutcome) { + let id_opt = { + let gids = self.aria2_gids.read().unwrap(); + gids.get(gid).cloned() + }; + match id_opt { + Some(id) => { + self.apply_completion(&id, outcome).await; + } + None => { + self.pending_completion + .lock() + .await + .insert(gid.to_string(), (String::new(), outcome)); + } + } + } + + /// Reorder a pending task up or down. Returns the new pending order. + /// No-op at boundaries. Does not emit (membership unchanged); the caller + /// (Tauri command) returns the order to the frontend. + pub async fn move_in_queue( + &self, + id: &str, + direction: QueueDirection, + ) -> Vec { + let mut pending = self.pending.lock().await; + let pos = pending.iter().position(|t| t.id == id); + if let Some(pos) = pos { + let target = match direction { + QueueDirection::Up => pos.checked_sub(1), + QueueDirection::Down => { + if pos + 1 < pending.len() { + Some(pos + 1) + } else { + None + } + } + }; + if let Some(target) = target { + pending.swap(pos, target); + } + } + pending.iter().map(|t| t.id.clone()).collect() + } + + /// Remove a task from pending if present (used by remove_download). + /// Does NOT release a permit (the caller handles active permits via + /// release_permit if the task was already dispatched). + pub async fn remove_from_pending(&self, id: &str) -> bool { + let mut pending = self.pending.lock().await; + let before = pending.len(); + pending.retain(|t| t.id != id); + let removed = pending.len() < before; + if removed { + self.notify.notify_one(); + } + removed + } + + /// Bulk enqueue by appending tasks. Used by startup and start-all. + pub async fn enqueue_many(&self, tasks: Vec) { + let mut pending = self.pending.lock().await; + for task in tasks { + let id = task.id.clone(); + pending.push_back(task); + self.emit_state(id, DownloadStatus::Queued); + } + drop(pending); + self.notify.notify_one(); + } +} + +/// Production spawner that delegates to the real aria2 RPC, yt-dlp, and +/// native coordinator runners. +pub struct ProductionSpawner { + app_handle: AppHandle, +} + +impl ProductionSpawner { + pub fn new(app_handle: AppHandle) -> Self { + Self { app_handle } + } +} + +#[async_trait::async_trait] +impl SidecarSpawner for ProductionSpawner { + async fn add_uri(&self, id: &str, payload: &SpawnPayload) -> Result { + let state = self.app_handle.state::(); + let mut options = serde_json::Map::new(); + let resolved_dest = crate::resolve_path(&payload.destination, &self.app_handle); + if !crate::is_safe_path(&resolved_dest, &self.app_handle) { + return Err("Path traversal blocked".to_string()); + } + options.insert( + "dir".to_string(), + serde_json::json!(resolved_dest.to_string_lossy().to_string()), + ); + let safe_filename = std::path::Path::new(&payload.filename.replace('\\', "/")) + .file_name() + .and_then(|n| n.to_str()) + .unwrap_or("download") + .to_string(); + options.insert("out".to_string(), serde_json::json!(safe_filename)); + let conn = payload.connections.unwrap_or(1); + options.insert("split".to_string(), serde_json::json!(conn.to_string())); + options.insert( + "max-connection-per-server".to_string(), + serde_json::json!(conn.to_string()), + ); + let mt = payload.max_tries.unwrap_or(1).max(1) as u32; + options.insert("max-tries".to_string(), serde_json::json!(mt.to_string())); + options.insert("continue".to_string(), serde_json::json!("true")); + if let Some(speed) = &payload.speed_limit { + options.insert("max-download-limit".to_string(), serde_json::json!(speed)); + } + if let Some(user) = &payload.username { + options.insert("http-user".to_string(), serde_json::json!(user)); + } + if let Some(pass) = &payload.password { + options.insert("http-passwd".to_string(), serde_json::json!(pass)); + } + if let Some(chk) = &payload.checksum { + options.insert("checksum".to_string(), serde_json::json!(chk)); + } + if let Some(ua) = &payload.user_agent { + options.insert("user-agent".to_string(), serde_json::json!(ua)); + } + let mut header_list = Vec::new(); + if let Some(cook) = &payload.cookies { + header_list.push(format!("Cookie: {}", cook)); + } + if let Some(hdrs) = &payload.headers { + for line in hdrs.lines() { + if !line.trim().is_empty() { + header_list.push(line.trim().to_string()); + } + } + } + if !header_list.is_empty() { + options.insert("header".to_string(), serde_json::json!(header_list)); + } + if let Some(prox) = &payload.proxy { + options.insert("all-proxy".to_string(), serde_json::json!(prox)); + } + let uris = crate::collect_download_uris(&payload.url, payload.mirrors.as_deref()); + let params = serde_json::json!([uris, options]); + + match crate::rpc_call(state.aria2_port, &state.aria2_secret, "aria2.addUri", params).await { + Ok(result) => { + let gid = result.as_str().unwrap_or("").to_string(); + Ok(gid) + } + Err(e) => { + // aria2 unavailable — fall back to native coordinator. + log::warn!("aria2 addUri failed, falling back to native: {}", e); + let download_id = uuid::Uuid::parse_str(id).map_err(|e| e.to_string())?; + let mt = payload.max_tries.unwrap_or(1).max(1) as u32; + let safe_filename = std::path::Path::new(&payload.filename.replace('\\', "/")) + .file_name() + .and_then(|n| n.to_str()) + .unwrap_or("download") + .to_string(); + state + .download_coordinator + .send(crate::download::DownloadCmd::Start(Box::new( + crate::download::DownloadPayload { + id: download_id, + urls: crate::collect_download_uris(&payload.url, payload.mirrors.as_deref()), + output_path: resolved_dest.join(safe_filename), + speed_limit: payload.speed_limit.clone(), + username: payload.username.clone(), + password: payload.password.clone(), + headers: payload.headers.clone(), + cookies: payload.cookies.clone(), + user_agent: payload.user_agent.clone(), + max_tries: mt, + proxy: payload.proxy.clone(), + }, + ))) + .await + .map_err(|e| e.to_string())?; + Ok(format!("native:{id}")) + } + } + } + + async fn run_media(&self, id: &str, payload: &SpawnPayload) -> Result<(), String> { + let state = self.app_handle.state::(); + let mut cancel_rx = state + .download_coordinator + .register_media(id.to_string()) + .await + .map_err(|e| e)?; + crate::start_media_download_internal( + self.app_handle.clone(), + id, + payload.url.clone(), + payload.destination.clone(), + payload.filename.clone(), + payload.format_selector.clone(), + payload.cookie_source.clone(), + payload.speed_limit.clone(), + payload.username.clone(), + payload.password.clone(), + payload.headers.clone(), + payload.proxy.clone(), + payload.user_agent.clone(), + payload.max_tries, + &mut cancel_rx, + ) + .await + } + + async fn run_native(&self, id: &str, payload: &SpawnPayload) -> Result<(), String> { + let state = self.app_handle.state::(); + let download_id = uuid::Uuid::parse_str(id).map_err(|e| e.to_string())?; + let mt = payload.max_tries.unwrap_or(1).max(1) as u32; + let resolved_dest = crate::resolve_path(&payload.destination, &self.app_handle); + let safe_filename = std::path::Path::new(&payload.filename.replace('\\', "/")) + .file_name() + .and_then(|n| n.to_str()) + .unwrap_or("download") + .to_string(); + state + .download_coordinator + .send(crate::download::DownloadCmd::Start(Box::new( + crate::download::DownloadPayload { + id: download_id, + urls: crate::collect_download_uris(&payload.url, payload.mirrors.as_deref()), + output_path: resolved_dest.join(safe_filename), + speed_limit: payload.speed_limit.clone(), + username: payload.username.clone(), + password: payload.password.clone(), + headers: payload.headers.clone(), + cookies: payload.cookies.clone(), + user_agent: payload.user_agent.clone(), + max_tries: mt, + proxy: payload.proxy.clone(), + }, + ))) + .await + .map_err(|e| e)?; + Ok(()) + } +} + +#[derive(Debug, Clone, Deserialize)] +pub struct EnqueueItem { + pub id: String, + pub url: String, + pub destination: String, + pub filename: String, + pub connections: Option, + pub speed_limit: Option, + pub username: Option, + pub password: Option, + pub headers: Option, + pub checksum: Option, + pub cookies: Option, + pub mirrors: Option, + pub user_agent: Option, + pub max_tries: Option, + pub proxy: Option, + pub format_selector: Option, + pub cookie_source: Option, + pub is_media: Option, +} + +impl EnqueueItem { + pub fn into_task(self) -> QueuedTask { + let media = self.is_media.unwrap_or(false); + let kind = if media { + TaskKind::Media + } else { + TaskKind::Aria2 + }; + let id = self.id.clone(); + QueuedTask { + id, + kind, + payload: SpawnPayload { + url: self.url, + destination: self.destination, + filename: self.filename, + connections: self.connections, + speed_limit: self.speed_limit, + username: self.username, + password: self.password, + headers: self.headers, + checksum: self.checksum, + cookies: self.cookies, + mirrors: self.mirrors, + user_agent: self.user_agent, + max_tries: self.max_tries, + proxy: self.proxy, + format_selector: self.format_selector, + cookie_source: self.cookie_source, + is_media: media, + }, + } + } +} diff --git a/src-tauri/src/scheduler.rs b/src-tauri/src/scheduler.rs index 64d743b..cc58407 100644 --- a/src-tauri/src/scheduler.rs +++ b/src-tauri/src/scheduler.rs @@ -1,4 +1,4 @@ -use tauri::{Manager, Emitter}; +use tauri::Emitter; use chrono::{Local, Datelike}; use std::time::Duration; diff --git a/src-tauri/tests/queue_manager.rs b/src-tauri/tests/queue_manager.rs new file mode 100644 index 0000000..6140144 --- /dev/null +++ b/src-tauri/tests/queue_manager.rs @@ -0,0 +1,329 @@ +use firelink_lib::queue::{QueueManager, SpawnPayload, TaskKind, QueuedTask}; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::Duration; +use tauri::test::{mock_builder, mock_context, noop_assets}; +use tokio::time::timeout; + +/// A fake spawner that records calls and lets tests gate sidecar lifetime. +struct CountingSpawner { + add_uri_calls: AtomicUsize, + media_calls: AtomicUsize, + native_calls: AtomicUsize, +} + +impl CountingSpawner { + fn new() -> Self { + Self { + add_uri_calls: AtomicUsize::new(0), + media_calls: AtomicUsize::new(0), + native_calls: AtomicUsize::new(0), + } + } +} + +#[async_trait::async_trait] +impl firelink_lib::queue::SidecarSpawner for CountingSpawner { + async fn add_uri(&self, _id: &str, _payload: &SpawnPayload) -> Result { + self.add_uri_calls.fetch_add(1, Ordering::SeqCst); + Ok(format!("gid-{}", self.add_uri_calls.load(Ordering::SeqCst))) + } + async fn run_media(&self, _id: &str, _payload: &SpawnPayload) -> Result<(), String> { + self.media_calls.fetch_add(1, Ordering::SeqCst); + Ok(()) + } + async fn run_native(&self, _id: &str, _payload: &SpawnPayload) -> Result<(), String> { + self.native_calls.fetch_add(1, Ordering::SeqCst); + Ok(()) + } +} + +/// Build a QueueManager with a fake spawner. Tauri's mock AppHandle is needed +/// for emit; we construct the minimal mock. +fn make_manager(capacity: usize) -> (QueueManager, Arc) { + let app = mock_builder() + .build(mock_context(noop_assets())) + .expect("mock app"); + let spawner = Arc::new(CountingSpawner::new()); + let mgr = QueueManager::test_new(app.handle().clone(), capacity, spawner.clone()); + (mgr, spawner) +} + +fn sample_task(id: &str) -> QueuedTask { + QueuedTask { + id: id.to_string(), + kind: TaskKind::Native, + payload: SpawnPayload::default(), + } +} + +#[tokio::test] +async fn push_appends_to_pending_and_emits_queued() { + let (mgr, _spawner) = make_manager(2); + mgr.push(sample_task("a")).await; + mgr.push(sample_task("b")).await; + let order = mgr.pending_order().await; + assert_eq!(order, vec!["a".to_string(), "b".to_string()]); +} + +#[tokio::test] +async fn release_permit_is_idempotent() { + let (mgr, _spawner) = make_manager(2); + let permit = mgr.acquire_permit().await; + mgr.park_permit("a", permit).await; + let avail_before = mgr.available_permits(); + mgr.release_permit("a").await; // first release: frees the slot + let avail_after_first = mgr.available_permits(); + mgr.release_permit("a").await; // second release: no-op + let avail_after_second = mgr.available_permits(); + assert_eq!(avail_after_first - avail_before, 1); + assert_eq!(avail_after_second, avail_after_first, "second release must not free another slot"); +} + +#[tokio::test] +async fn push_then_pop_front_drains_fifo() { + let (mgr, _spawner) = make_manager(2); + mgr.push(sample_task("a")).await; + mgr.push(sample_task("b")).await; + let first = mgr.pop_front().await.expect("some task"); + let second = mgr.pop_front().await.expect("some task"); + assert_eq!(first.id, "a"); + assert_eq!(second.id, "b"); + assert!(mgr.pop_front().await.is_none()); +} + +#[tokio::test] +async fn dispatcher_parks_when_idle_no_busy_spin() { + let (mgr, _spawner) = make_manager(3); + let mgr_arc = Arc::new(mgr); + let handle = { + let mgr_clone = Arc::clone(&mgr_arc); + tokio::spawn(async move { mgr_clone.run_dispatcher().await }) + }; + + // Queue is empty. Sleep long enough for any busy-spin to drain permits. + tokio::time::sleep(Duration::from_millis(200)).await; + + // No permit should have been acquired while idle. + assert_eq!( + mgr_arc.available_permits(), 3, + "dispatcher must not acquire permits when pending is empty" + ); + + handle.abort(); +} + +#[tokio::test] +async fn cas_retirement_never_underflows() { + let (mgr, _spawner) = make_manager(3); + let mgr_arc = Arc::new(mgr); + let handle = { + let mgr_clone = Arc::clone(&mgr_arc); + tokio::spawn(async move { mgr_clone.run_dispatcher().await }) + }; + + // Repeatedly toggle capacity down while idle. A fetch_sub-based impl + // would underflow to usize::MAX on the second set_capacity(1). + for _ in 0..5 { + mgr_arc.set_capacity(3); + mgr_arc.set_capacity(1); + mgr_arc.set_capacity(3); + } + tokio::time::sleep(Duration::from_millis(50)).await; + + let debt = mgr_arc.slots_to_retire_load(); + assert!( + debt <= 2, + "retirement debt must stay within [0, capacity-target], got {debt}" + ); + + handle.abort(); +} + +#[tokio::test] +async fn grow_releases_immediately_and_dispatches_waiting_tasks() { + let (mgr, spawner) = make_manager(2); + let mgr_arc = Arc::new(mgr); + + for i in 0..4 { + mgr_arc.push(sample_task(&format!("t{i}"))).await; + } + let handle = { + let mgr_clone = Arc::clone(&mgr_arc); + tokio::spawn(async move { mgr_clone.run_dispatcher().await }) + }; + + // Give dispatcher time to dispatch 2 (capacity) of the 4. + tokio::time::sleep(Duration::from_millis(100)).await; + let native_after_initial = spawner.native_calls.load(Ordering::SeqCst); + assert_eq!(native_after_initial, 2, "only capacity-many tasks dispatch initially"); + + // Grow to 4; the remaining 2 should dispatch. + mgr_arc.set_capacity(4); + tokio::time::sleep(Duration::from_millis(100)).await; + let native_after_grow = spawner.native_calls.load(Ordering::SeqCst); + assert_eq!(native_after_grow, 4, "grow must allow the waiting tasks to dispatch"); + + handle.abort(); +} + +#[tokio::test] +async fn shrink_converges_to_target_without_killing_active() { + let (mgr, spawner) = make_manager(4); + let mgr_arc = Arc::new(mgr); + + for i in 0..6 { + mgr_arc.push(sample_task(&format!("t{i}"))).await; + } + let handle = { + let mgr_clone = Arc::clone(&mgr_arc); + tokio::spawn(async move { mgr_clone.run_dispatcher().await }) + }; + + // Let 4 dispatch (capacity). + tokio::time::sleep(Duration::from_millis(100)).await; + assert_eq!(spawner.native_calls.load(Ordering::SeqCst), 4); + + // Shrink to 2 while 4 are "active" (permits parked). + mgr_arc.set_capacity(2); + + // Release active permits. Debt is 2; two releases retire both. + mgr_arc.release_permit("t0").await; + mgr_arc.release_permit("t1").await; + tokio::time::sleep(Duration::from_millis(100)).await; + + // Debt was 2; two releases retired both. The 2 remaining pending tasks + // (t4, t5) can now dispatch since debt is 0 and slots freed. + assert_eq!( + spawner.native_calls.load(Ordering::SeqCst), + 6, + "shrink converges: after debt exhausted, remaining pending dispatch" + ); + + handle.abort(); +} + +fn aria2_task(id: &str) -> QueuedTask { + QueuedTask { + id: id.to_string(), + kind: TaskKind::Aria2, + payload: SpawnPayload::default(), + } +} + +#[tokio::test] +async fn aria2_permit_survives_rpc_return() { + let (mgr, spawner) = make_manager(1); + let mgr_arc = Arc::new(mgr); + mgr_arc.push(aria2_task("a")).await; + let handle = { + let mgr_clone = Arc::clone(&mgr_arc); + tokio::spawn(async move { mgr_clone.run_dispatcher().await }) + }; + + // Dispatcher acquires the single permit, parks it, calls add_uri (returns + // instantly). The permit must STAY parked. + tokio::time::sleep(Duration::from_millis(100)).await; + assert_eq!(spawner.add_uri_calls.load(Ordering::SeqCst), 1); + assert_eq!( + mgr_arc.available_permits(), 0, + "permit must remain parked while aria2 download is notionally running" + ); + + // Now simulate aria2 completion: release_permit frees the slot. + mgr_arc.release_permit("a").await; + tokio::time::sleep(Duration::from_millis(50)).await; + assert_eq!(mgr_arc.available_permits(), 1, "release frees the parked permit"); + + handle.abort(); +} + +#[tokio::test] +async fn gid_completion_before_store_buffers_and_reconciles() { + use firelink_lib::queue::PendingOutcome; + + let (mgr, _spawner) = make_manager(1); + let mgr_arc = Arc::new(mgr); + mgr_arc.push(aria2_task("a")).await; + let handle = { + let mgr_clone = Arc::clone(&mgr_arc); + tokio::spawn(async move { mgr_clone.run_dispatcher().await }) + }; + tokio::time::sleep(Duration::from_millis(100)).await; + + // The dispatcher called add_uri and got "gid-1", then remember_gid stored it. + // Simulate a completion arriving for an UNKNOWN gid first: + mgr_arc + .handle_aria2_event("gid-unknown", PendingOutcome::Complete) + .await; + tokio::time::sleep(Duration::from_millis(50)).await; + // Permit still parked (gid-unknown is not ours). + assert_eq!(mgr_arc.available_permits(), 0); + + // Now store gid-1 -> "a" via remember_gid. The buffered gid-unknown stays + // buffered (different gid). Release via the real gid: + mgr_arc.release_permit("a").await; + tokio::time::sleep(Duration::from_millis(50)).await; + assert_eq!(mgr_arc.available_permits(), 1); + + // Push another aria2 task; its gid will be "gid-2". + mgr_arc.push(aria2_task("b")).await; + tokio::time::sleep(Duration::from_millis(100)).await; + mgr_arc.release_permit("b").await; + tokio::time::sleep(Duration::from_millis(50)).await; + + handle.abort(); +} + +#[tokio::test] +async fn move_up_down_reorders_pending() { + use firelink_lib::ipc::QueueDirection; + + let (mgr, _spawner) = make_manager(3); + let mgr_arc = Arc::new(mgr); + mgr_arc.push(sample_task("a")).await; + mgr_arc.push(sample_task("b")).await; + mgr_arc.push(sample_task("c")).await; + + mgr_arc.move_in_queue("c", QueueDirection::Down).await; + assert_eq!(mgr_arc.pending_order().await, vec!["a", "b", "c"]); + + mgr_arc.move_in_queue("c", QueueDirection::Up).await; + assert_eq!(mgr_arc.pending_order().await, vec!["a", "c", "b"]); + + mgr_arc.move_in_queue("a", QueueDirection::Down).await; + assert_eq!(mgr_arc.pending_order().await, vec!["c", "a", "b"]); + + mgr_arc.move_in_queue("c", QueueDirection::Up).await; + assert_eq!(mgr_arc.pending_order().await, vec!["c", "a", "b"]); +} + +#[tokio::test] +async fn notify_fires_on_push_and_release() { + let (mgr, _spawner) = make_manager(1); + let mgr_arc = Arc::new(mgr); + + let permit = mgr_arc.acquire_permit().await; + mgr_arc.park_permit("a", permit).await; + + let handle = { + let mgr_clone = Arc::clone(&mgr_arc); + tokio::spawn(async move { mgr_clone.run_dispatcher().await }) + }; + + mgr_arc.push(sample_task("x")).await; + let dispatched = timeout( + Duration::from_millis(150), + async { + loop { + if mgr_arc.available_permits() == 0 { + return; + } + tokio::time::sleep(Duration::from_millis(5)).await; + } + }, + ).await; + assert!(dispatched.is_ok(), "push must wake the idle dispatcher"); + + handle.abort(); +} diff --git a/src/App.tsx b/src/App.tsx index 9f50db8..0bd60bc 100644 --- a/src/App.tsx +++ b/src/App.tsx @@ -21,7 +21,7 @@ function App() { const stored = Number(window.localStorage.getItem('firelink-sidebar-width')); return Number.isFinite(stored) && stored >= 190 && stored <= 260 ? stored : 220; }); - const updateDownload = useDownloadStore(state => state.updateDownload); + const theme = useSettingsStore(state => state.theme); const isSidebarVisible = useSettingsStore(state => state.isSidebarVisible); const activeView = useSettingsStore(state => state.activeView); @@ -209,8 +209,6 @@ function App() { initDownloadListener(); const unlistenComplete = listen('download-complete', (event) => { - updateDownload(event.payload, { status: 'completed', fraction: 1.0, speed: '-', eta: '-' }); - const settings = useSettingsStore.getState(); if (settings.showNotifications) { const item = useDownloadStore.getState().downloads.find(d => d.id === event.payload); @@ -225,10 +223,15 @@ function App() { }); const unlistenFailed = listen('download-failed', (event) => { - // If it's already paused, don't mark as failed (since we aborted it) - const current = useDownloadStore.getState().downloads.find(d => d.id === event.payload); - if (current && current.status !== 'paused') { - updateDownload(event.payload, { status: 'failed', speed: '-', eta: '-' }); + const settings = useSettingsStore.getState(); + if (settings.showNotifications) { + const item = useDownloadStore.getState().downloads.find(d => d.id === event.payload); + const fileName = item?.fileName || 'A file'; + + sendNotification({ + title: 'Download Failed', + body: `${fileName} failed to download.`, + }); } }); diff --git a/src/bindings/DownloadItem.ts b/src/bindings/DownloadItem.ts index b53f64f..4b0e435 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, connections?: number, speedLimit?: string, username?: string, password?: string, headers?: string, checksum?: string, cookies?: string, mirrors?: string, destination?: string, isMedia?: boolean, mediaFormatSelector?: string, queueId: string, _dispatched?: boolean, }; +export type DownloadItem = { id: string, url: string, fileName: string, status: DownloadStatus, fraction?: number, speed?: string, eta?: string, size?: string, category: DownloadCategory, dateAdded: string, connections?: number, speedLimit?: string, username?: string, password?: string, headers?: string, checksum?: string, cookies?: string, mirrors?: string, destination?: string, isMedia?: boolean, mediaFormatSelector?: string, queueId: string, }; diff --git a/src/bindings/DownloadStateEvent.ts b/src/bindings/DownloadStateEvent.ts new file mode 100644 index 0000000..9d5a8a7 --- /dev/null +++ b/src/bindings/DownloadStateEvent.ts @@ -0,0 +1,3 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +export type DownloadStateEvent = { id: string, status: string, error: string | null, }; diff --git a/src/bindings/QueueDirection.ts b/src/bindings/QueueDirection.ts new file mode 100644 index 0000000..998ab65 --- /dev/null +++ b/src/bindings/QueueDirection.ts @@ -0,0 +1,3 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +export type QueueDirection = "up" | "down"; diff --git a/src/components/DownloadItem.tsx b/src/components/DownloadItem.tsx index e20c293..a0b4d9a 100644 --- a/src/components/DownloadItem.tsx +++ b/src/components/DownloadItem.tsx @@ -1,7 +1,7 @@ import React, { useEffect, useRef } from 'react'; import { useDownloadStore } from '../store/useDownloadStore'; import { useDownloadProgressStore } from '../store/downloadStore'; -import { Play, Pause, MoreVertical } from 'lucide-react'; +import { Play, Pause, MoreVertical, Clock, ArrowUp, ArrowDown } from 'lucide-react'; import type { DownloadItem as DownloadItemType } from '../bindings/DownloadItem'; interface DownloadItemProps { @@ -24,6 +24,9 @@ export const DownloadItem = React.memo(({ getCategoryIcon, }) => { const download = useDownloadStore(state => state.downloads.find(d => d.id === downloadId)); + const pendingOrder = useDownloadStore(state => state.pendingOrder); + const moveInQueue = useDownloadStore(state => state.moveInQueue); + const queueIndex = pendingOrder.indexOf(downloadId); const progressBarRef = useRef(null); const statusTextRef = useRef(null); @@ -31,7 +34,6 @@ export const DownloadItem = React.memo(({ const etaTextRef = useRef(null); useEffect(() => { - // We only need transient updates while it's actively downloading if (!download || download.status !== 'downloading') return; const unsubscribe = useDownloadProgressStore.subscribe((state) => { @@ -89,17 +91,32 @@ export const DownloadItem = React.memo(({
- {download.status === 'downloading' - ? `${((download.fraction || 0) * 100).toFixed(0)}%` - : download.status.charAt(0).toUpperCase() + download.status.slice(1)} + {download.status === 'queued' && queueIndex !== -1 ? ( + <> + + Queued #{queueIndex + 1} + + ) : download.status === 'downloading' ? ( + `${((download.fraction || 0) * 100).toFixed(0)}%` + ) : ( + download.status.charAt(0).toUpperCase() + download.status.slice(1) + )} )} @@ -119,6 +136,26 @@ export const DownloadItem = React.memo(({
+ {download.status === 'queued' && queueIndex !== -1 && ( + <> + + + + )} {download.status === 'downloading' && (