From 8a258981ceff129ef304d6d13fa89035aff8b4c3 Mon Sep 17 00:00:00 2001 From: NimBold Date: Thu, 25 Jun 2026 00:19:18 +0330 Subject: [PATCH] fix(security): resolve all medium severity vulnerabilities - Ensure yt-dlp partial files are cleaned up on error exits. - Handle Semaphore closure gracefully without panicking. - Replace busy-polling in wait_permit_released with Notify. - Fix extension server silent failure with retry loop and logging. - Ignore malformed existing scheduler settings instead of failing. - Add final symlink checks before revealing/opening downloads. - Delay listener registration in App.tsx until core is ready. - Sanitize Referer header to prevent injection. --- src-tauri/src/commands.rs | 8 + src-tauri/src/lib.rs | 305 +++++++++++++++------------ src-tauri/src/queue.rs | 33 ++- src-tauri/src/settings.rs | 13 +- src/App.tsx | 3 +- src/components/AddDownloadsModal.tsx | 2 +- 6 files changed, 221 insertions(+), 143 deletions(-) diff --git a/src-tauri/src/commands.rs b/src-tauri/src/commands.rs index 3f7ba97..3bfe312 100644 --- a/src-tauri/src/commands.rs +++ b/src-tauri/src/commands.rs @@ -15,6 +15,10 @@ pub async fn reveal_in_file_manager( ) })?; + if std::fs::symlink_metadata(&path).is_ok_and(|metadata| metadata.file_type().is_symlink()) { + return Err("Download path was replaced by a symlink before reveal".to_string()); + } + app_handle .opener() .reveal_item_in_dir(path.to_string_lossy().as_ref()) @@ -33,6 +37,10 @@ pub async fn open_downloaded_file( return Err(format!("Downloaded file is missing: {}", path.display())); } + if std::fs::symlink_metadata(&path).is_ok_and(|metadata| metadata.file_type().is_symlink()) { + return Err("Download path was replaced by a symlink before open".to_string()); + } + app_handle .opener() .open_path(path.to_string_lossy().as_ref(), None::) diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 7632e91..112efdd 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -745,6 +745,9 @@ async fn validate_url_ssrf(url: &str) -> Result, - pub aria2_port: u16, + pub aria2_port: std::sync::Arc, pub aria2_secret: String, pub media_semaphore: Arc, pub sleep_preventer: Arc>>, @@ -1551,7 +1554,7 @@ async fn test_aria2c(app_handle: tauri::AppHandle, state: tauri::State<'_, AppSt } let result = rpc_call( - state.aria2_port, + state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), &state.aria2_secret, "aria2.getVersion", serde_json::json!([]), @@ -1926,7 +1929,7 @@ async fn get_engine_status( app_handle: tauri::AppHandle, state: tauri::State<'_, AppState>, ) -> Result { - let port = state.aria2_port; + let port = state.aria2_port.load(std::sync::atomic::Ordering::Relaxed); let secret = state.aria2_secret.clone(); let (aria2, ytdlp, ffmpeg, deno) = tokio::join!( @@ -1946,7 +1949,7 @@ async fn get_aria2_engine_status( app_handle: tauri::AppHandle, state: tauri::State<'_, AppState>, ) -> Result { - Ok(check_aria2(&app_handle, state.aria2_port, &state.aria2_secret).await) + Ok(check_aria2(&app_handle, state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), &state.aria2_secret).await) } #[tauri::command] @@ -2072,7 +2075,7 @@ pub(crate) async fn start_media_download_internal( .arg("--progress-delta").arg("0.2") .arg("--progress-template") .arg(format!("download:{MEDIA_PROGRESS_PREFIX}%(progress)j")) - .arg("--no-check-formats") + .arg("--socket-timeout").arg("20") .arg("--retries").arg("3") .arg("--extractor-retries").arg("3") @@ -2242,6 +2245,7 @@ pub(crate) async fn start_media_download_internal( } Some(tauri_plugin_shell::process::CommandEvent::Error(err)) => { log::error!("yt-dlp shell error [{}]: {}", id, err); + cleanup_media_artifacts(&out_path, false).await; break err; } Some(tauri_plugin_shell::process::CommandEvent::Terminated(payload)) => { @@ -2262,6 +2266,7 @@ pub(crate) async fn start_media_download_internal( return Ok(()); } log::error!("yt-dlp exited with non-zero code {:?} for id: {}", payload.code, id); + cleanup_media_artifacts(&out_path, false).await; break if stderr_tail.is_empty() { format!("yt-dlp exited with code {:?}", payload.code) } else { @@ -2270,6 +2275,7 @@ pub(crate) async fn start_media_download_internal( } Some(_) => {} None => { + cleanup_media_artifacts(&out_path, false).await; break if stderr_tail.is_empty() { "yt-dlp process ended unexpectedly".to_string() } else { @@ -2325,7 +2331,7 @@ async fn pause_download( let gid = state.queue_manager.aria2_gid_for_download(&id); if let Some(gid) = gid.as_deref().filter(|gid| !gid.starts_with("native:")) { let status = aria2_download_status( - state.aria2_port, + state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), &state.aria2_secret, gid, ) @@ -2336,7 +2342,7 @@ async fn pause_download( } "active" | "waiting" => { let result = rpc_call( - state.aria2_port, + state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), &state.aria2_secret, "aria2.forcePause", serde_json::json!([gid]), @@ -2412,7 +2418,7 @@ async fn resume_download( return Ok(false); } - let status = aria2_download_status(state.aria2_port, &state.aria2_secret, &gid).await?; + let status = aria2_download_status(state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), &state.aria2_secret, &gid).await?; match status.as_str() { "paused" => { use tauri::Emitter; @@ -2422,11 +2428,12 @@ async fn resume_download( ); let queue_manager = state.queue_manager.clone(); - let aria2_port = state.aria2_port; + let aria2_port = state.aria2_port.load(std::sync::atomic::Ordering::Relaxed); let aria2_secret = state.aria2_secret.clone(); let id_clone = id.clone(); let gid_clone = gid.clone(); + let app_handle_clone = app_handle.clone(); tauri::async_runtime::spawn(async move { let acquired = queue_manager.ensure_aria2_permit(&id_clone).await; let result = match rpc_call( @@ -2443,6 +2450,13 @@ async fn resume_download( queue_manager.release_permit(&id_clone).await; } log::error!("failed to resume aria2 gid {}: {}", gid_clone, error); + let _ = app_handle_clone.emit( + "download-state", + crate::ipc::DownloadStateEvent::new( + &id_clone, + crate::ipc::DownloadStatus::Failed, + ), + ); return; } }; @@ -2451,6 +2465,13 @@ async fn resume_download( queue_manager.release_permit(&id_clone).await; } log::error!("failed to resume aria2 gid {}: {}", gid_clone, error); + let _ = app_handle_clone.emit( + "download-state", + crate::ipc::DownloadStateEvent::new( + &id_clone, + crate::ipc::DownloadStatus::Failed, + ), + ); return; } log::info!("aria2 resume [{}]: unpaused gid {}", id_clone, gid_clone); @@ -2510,8 +2531,8 @@ async fn remove_download( let gid = state.queue_manager.aria2_gid_for_download(&id); if let Some(gid) = gid.as_deref().filter(|gid| !gid.starts_with("native:")) { let removal_result = async { - force_remove_aria2_gid(state.aria2_port, &state.aria2_secret, gid).await?; - wait_for_aria2_stopped(state.aria2_port, &state.aria2_secret, gid).await + force_remove_aria2_gid(state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), &state.aria2_secret, gid).await?; + wait_for_aria2_stopped(state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), &state.aria2_secret, gid).await } .await; if let Err(error) = removal_result { @@ -2605,13 +2626,13 @@ async fn detach_download_for_reconfigure( if let Some(gid) = gid.as_deref().filter(|gid| !gid.starts_with("native:")) { let removal_result = async { rpc_call( - state.aria2_port, + state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), &state.aria2_secret, "aria2.forcePause", serde_json::json!([gid]), ) .await?; - wait_for_aria2_stopped(state.aria2_port, &state.aria2_secret, gid).await + wait_for_aria2_stopped(state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), &state.aria2_secret, gid).await } .await; if let Err(error) = removal_result { @@ -2956,7 +2977,7 @@ async fn set_global_speed_limit(state: tauri::State<'_, AppState>, limit: Option .and_then(normalize_speed_limit_for_aria2) .unwrap_or_else(|| "0".to_string()); rpc_call( - state.aria2_port, + state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), &state.aria2_secret, "aria2.changeGlobalOption", serde_json::json!([{"max-overall-download-limit": limit_str}]) @@ -3313,7 +3334,7 @@ async fn read_logs(app_handle: tauri::AppHandle, limit: usize) -> Result Result<(), String> { for file in log_files(&app_handle).await? { - let _ = tokio::fs::write(&file, "").await; + tokio::fs::write(&file, "").await.map_err(|e| format!("Failed to clear log file {:?}: {}", file, e))?; } Ok(()) } @@ -3332,7 +3353,7 @@ async fn export_logs( chrono::Utc::now().to_rfc3339(), ); let (aria2, ytdlp, ffmpeg, deno) = tokio::join!( - check_aria2(&app_handle, state.aria2_port, &state.aria2_secret), + check_aria2(&app_handle, state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), &state.aria2_secret), check_ytdlp(&app_handle), check_ffmpeg(&app_handle), check_deno(&app_handle), @@ -3883,10 +3904,9 @@ 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 aria2_port = std::net::TcpListener::bind("127.0.0.1:0") - .and_then(|listener| listener.local_addr()) - .map(|addr| addr.port()) - .unwrap_or(6800); + let initial_aria2_port = 6800; // Will be determined dynamically in background + let aria2_port = Arc::new(std::sync::atomic::AtomicU16::new(initial_aria2_port)); + let aria2_port_clone = Arc::clone(&aria2_port); let aria2_secret = uuid::Uuid::new_v4().to_string(); tauri::Builder::default() .plugin(tauri_plugin_single_instance::init(|app, _args, _cwd| { @@ -3956,7 +3976,7 @@ pub fn run() { extension_frontend_ready, extension_server_port, extension_server_shutdown: extension_server_shutdown_tx.clone(), - aria2_port, + aria2_port: aria2_port.clone(), aria2_secret: aria2_secret.clone(), media_semaphore: Arc::new(tokio::sync::Semaphore::new(3)), sleep_preventer: Arc::new(Mutex::new(None)), @@ -4016,123 +4036,141 @@ pub fn run() { .map(|settings| settings.global_speed_limit) .unwrap_or_default(); - match resolve_bundled_binary_path(app.handle(), "aria2c") { - Ok(binary_path) => { - let mut cmd = std::process::Command::new(&binary_path); - - let mut config_file = tempfile::Builder::new().prefix("aria2-").suffix(".conf").tempfile().expect("failed to create aria2 config file"); - use std::io::Write; - let config_content = format!("rpc-secret={}\n", aria2_secret); - config_file.write_all(config_content.as_bytes()).expect("failed to write aria2 config file"); - let config_path = config_file.into_temp_path(); + let aria2_secret_clone = aria2_secret.clone(); + let app_handle_bg = app.handle().clone(); + tauri::async_runtime::spawn(async move { - cmd.arg("--enable-rpc=true") - .arg(format!("--conf-path={}", config_path.display())) - .arg(format!("--rpc-listen-port={}", aria2_port)) - .arg("--rpc-listen-all=false") - .arg("--continue=true") - .arg("--retry-wait=2") - .arg("--allow-overwrite=false") - .arg("--summary-interval=1") - .arg("--console-log-level=warn") - .arg("--download-result=hide") - .arg("--max-concurrent-downloads=9999") - .arg("--check-certificate=true"); - - if let Some(global_speed_limit) = normalize_speed_limit_for_aria2(&global_speed_limit) { - cmd.arg(format!("--max-overall-download-limit={}", global_speed_limit)); - } - - cmd.stdout(std::process::Stdio::null()); - cmd.stderr(std::process::Stdio::piped()); - - match cmd.spawn() { - Ok(mut child) => { - log::info!("aria2c spawned successfully on port {}", aria2_port); - - let daemon_app = app.handle().clone(); - if let Some(stderr) = child.stderr.take() { - std::thread::spawn(move || { - use std::io::BufRead; - let reader = std::io::BufReader::new(stderr); - for line in reader.lines().map_while(Result::ok) { - let trimmed = line.trim().to_string(); - if let Ok(mut stderr_lock) = daemon_app.state::().last_stderr.lock() { - stderr_lock.push_str(&trimmed); - stderr_lock.push('\n'); - let excess = stderr_lock.len().saturating_sub(8192); - if excess > 0 { - let _ = stderr_lock.drain(..excess); - } - } - let lower = trimmed.to_lowercase(); - if lower.contains("error") || lower.contains("critical") { - log::error!("aria2c stderr: {}", trimmed); - } - } - }); + let mut ws_port = 6800; + match resolve_bundled_binary_path(&app_handle_bg, "aria2c") { + Ok(binary_path) => { + let mut success = false; + for attempt_port in 6800..6900 { + let mut cmd = std::process::Command::new(&binary_path); + + let mut config_file = tempfile::Builder::new().prefix("aria2-").suffix(".conf").tempfile().expect("failed to create aria2 config file"); + use std::io::Write; + let config_content = format!("rpc-secret={}\n", aria2_secret_clone); + config_file.write_all(config_content.as_bytes()).expect("failed to write aria2 config file"); + let config_path = config_file.into_temp_path(); + + cmd.arg("--enable-rpc=true") + .arg(format!("--conf-path={}", config_path.display())) + .arg(format!("--rpc-listen-port={}", attempt_port)) + .arg("--rpc-listen-all=false") + .arg("--continue=true") + .arg("--retry-wait=2") + .arg("--allow-overwrite=false") + .arg("--summary-interval=1") + .arg("--console-log-level=warn") + .arg("--download-result=hide") + .arg("--max-concurrent-downloads=9999") + .arg("--check-certificate=true"); + + if let Some(limit) = normalize_speed_limit_for_aria2(&global_speed_limit) { + cmd.arg(format!("--max-overall-download-limit={}", limit)); } + + cmd.stdout(std::process::Stdio::null()); + cmd.stderr(std::process::Stdio::piped()); + + match cmd.spawn() { + Ok(mut child) => { + // Give it a moment to fail if port is in use + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + if let Ok(Some(_)) = child.try_wait() { + // Process exited, likely port collision, try next + continue; + } - let guard = app.state::(); - *guard.child.lock().unwrap() = Some(child); - *guard.config_path.lock().unwrap() = Some(config_path); + log::info!("aria2c spawned successfully on port {}", attempt_port); - let port = aria2_port; - let secret = aria2_secret.clone(); - let start = std::time::Instant::now(); - let ready = tauri::async_runtime::block_on(async { - let mut last_err = String::new(); - loop { - if start.elapsed() > std::time::Duration::from_secs(5) { - return Err(if last_err.is_empty() { - "aria2 daemon did not become ready within 5 seconds".to_string() - } else { - format!("aria2 did not become ready: {last_err}") + aria2_port_clone.store(attempt_port, std::sync::atomic::Ordering::Relaxed); + ws_port = attempt_port; + success = true; + + let daemon_app = app_handle_bg.clone(); + if let Some(stderr) = child.stderr.take() { + std::thread::spawn(move || { + use std::io::BufRead; + let reader = std::io::BufReader::new(stderr); + for line in reader.lines().map_while(Result::ok) { + let trimmed = line.trim().to_string(); + if let Ok(mut stderr_lock) = daemon_app.state::().last_stderr.lock() { + stderr_lock.push_str(&trimmed); + stderr_lock.push('\n'); + let excess = stderr_lock.len().saturating_sub(8192); + if excess > 0 { + let _ = stderr_lock.drain(..excess); + } + } + let lower = trimmed.to_lowercase(); + if lower.contains("error") || lower.contains("critical") { + log::error!("aria2c stderr: {}", trimmed); + } + } }); } - match rpc_call(port, &secret, "aria2.getVersion", serde_json::json!([])).await { - Ok(ver) => return Ok(ver), - Err(e) => { - last_err = e; - tokio::time::sleep(std::time::Duration::from_millis(100)).await; + + let guard = app_handle_bg.state::(); + *guard.child.lock().unwrap() = Some(child); + *guard.config_path.lock().unwrap() = Some(config_path); + + let mut last_err = String::new(); + let start = std::time::Instant::now(); + let mut ready = false; + while start.elapsed() < std::time::Duration::from_secs(5) { + match rpc_call(attempt_port, &aria2_secret_clone, "aria2.getVersion", serde_json::json!([])).await { + Ok(ver) => { + let v = ver.get("version").and_then(|v| v.as_str()).unwrap_or("unknown"); + log::info!("aria2 daemon ready (version {}) on port {}", v, attempt_port); + ready = true; + break; + } + Err(e) => { + last_err = e; + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } } } - } - }); - - match ready { - Ok(ver) => { - let v = ver.get("version").and_then(|v| v.as_str()).unwrap_or("unknown"); - log::info!("aria2 daemon ready (version {}) on port {}", v, port); + if !ready { + let err = if last_err.is_empty() { "aria2 daemon did not become ready within 5 seconds".to_string() } else { format!("aria2 did not become ready: {last_err}") }; + log::error!("{}", err); + *guard.startup_error.lock().unwrap() = Some(err); + } + break; } Err(e) => { - log::error!("aria2 daemon readiness check failed: {}", e); - let guard = app.state::(); - *guard.startup_error.lock().unwrap() = Some(e); + log::error!("Failed to spawn aria2c: {}", e); + let guard = app_handle_bg.state::(); + *guard.startup_error.lock().unwrap() = Some(format!("Failed to spawn aria2c: {e}")); + break; } } } - Err(e) => { - log::error!("Failed to spawn aria2c: {}", e); - let guard = app.state::(); - *guard.startup_error.lock().unwrap() = Some(format!("Failed to spawn aria2c: {e}")); + if !success { + let guard = app_handle_bg.state::(); + *guard.startup_error.lock().unwrap() = Some("Failed to find open port for aria2c".to_string()); } } + Err(e) => { + log::error!("Failed to resolve aria2c binary: {}", e); + let guard = app_handle_bg.state::(); + *guard.startup_error.lock().unwrap() = Some(format!("Failed to resolve aria2c: {e}")); + } } - Err(e) => { - log::error!("Failed to resolve aria2c binary: {}", e); - let guard = app.state::(); - *guard.startup_error.lock().unwrap() = Some(format!("Failed to resolve aria2c: {e}")); - } - } - let app_handle_ws = app.handle().clone(); - let ws_port = aria2_port; - tauri::async_runtime::spawn(async move { + let mut ws_retries = 0; loop { - tokio::time::sleep(std::time::Duration::from_secs(1)).await; + if ws_retries > 10 { + log::error!("Max WebSocket reconnection attempts reached. aria2 integration is disabled."); + let guard = app_handle_bg.state::(); + *guard.startup_error.lock().unwrap() = Some("Max WebSocket reconnection attempts reached.".to_string()); + break; + } + tokio::time::sleep(std::time::Duration::from_secs(2)).await; let ws_url = format!("ws://127.0.0.1:{}/jsonrpc", ws_port); if let Ok((ws_stream, _)) = tokio_tungstenite::connect_async(&ws_url).await { + ws_retries = 0; // reset on success use futures_util::StreamExt; let (_, mut read) = ws_stream.split(); while let Some(msg) = read.next().await { @@ -4142,7 +4180,7 @@ 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 state = app_handle_ws.state::(); + let state = app_handle_bg.state::(); let outcome = match method { "aria2.onDownloadComplete" => Some(crate::queue::PendingOutcome::Complete), "aria2.onDownloadError" => { @@ -4164,14 +4202,14 @@ pub fn run() { } } } - // Connection lost, clear permits and reconnect - app_handle_ws.state::().queue_manager.clear_aria2_permits().await; + ws_retries += 1; + app_handle_bg.state::().queue_manager.clear_aria2_permits().await; tokio::time::sleep(std::time::Duration::from_secs(4)).await; } }); let app_handle_poll = app.handle().clone(); - let poll_port = aria2_port; + let poll_port = aria2_port.clone(); let poll_secret = aria2_secret.clone(); let poll_mgr = Arc::clone(&queue_manager_poll); tauri::async_runtime::spawn(async move { @@ -4179,7 +4217,7 @@ pub fn run() { loop { interval.tick().await; let params = serde_json::json!([["gid", "status", "totalLength", "completedLength", "downloadSpeed", "errorMessage"]]); - if let Ok(active_list) = rpc_call(poll_port, &poll_secret, "aria2.tellActive", params).await { + if let Ok(active_list) = rpc_call(poll_port.load(std::sync::atomic::Ordering::Relaxed), &poll_secret, "aria2.tellActive", params).await { if let Some(active_arr) = active_list.as_array() { for status_info in active_arr { let gid = status_info.get("gid").and_then(|s| s.as_str()).unwrap_or(""); @@ -4221,14 +4259,19 @@ pub fn run() { let ext_app_handle = app.handle().clone(); tauri::async_runtime::spawn(async move { - if let Err(error) = extension_server::start_server( - ext_app_handle, - server_pairing_token.clone(), - server_frontend_ready.clone(), - server_extension_port.clone(), - extension_server_shutdown_rx, - ).await { - eprintln!("Browser extension server unavailable: {error}"); + loop { + if let Err(error) = extension_server::start_server( + ext_app_handle.clone(), + server_pairing_token.clone(), + server_frontend_ready.clone(), + server_extension_port.clone(), + extension_server_shutdown_rx.clone(), + ).await { + log::error!("Browser extension server unavailable: {error}"); + tokio::time::sleep(std::time::Duration::from_secs(10)).await; + } else { + break; + } } }); Ok(()) diff --git a/src-tauri/src/queue.rs b/src-tauri/src/queue.rs index 94927f9..b0424c1 100644 --- a/src-tauri/src/queue.rs +++ b/src-tauri/src/queue.rs @@ -96,6 +96,7 @@ pub struct QueueManager { target_capacity: AtomicUsize, slots_to_retire: AtomicUsize, notify: Notify, + notify_permit_released: Notify, /// aria2 gid -> download id map (shared with the WS poller). pub aria2_gids: Arc>>, @@ -143,6 +144,7 @@ impl QueueManager { target_capacity: AtomicUsize::new(capacity), slots_to_retire: AtomicUsize::new(0), notify: Notify::new(), + notify_permit_released: Notify::new(), aria2_gids: Arc::new(std::sync::RwLock::new(HashMap::new())), pending_completion: Arc::new(Mutex::new(HashMap::new())), aria2_payloads: Mutex::new(HashMap::new()), @@ -191,12 +193,12 @@ impl QueueManager { } /// Acquire a permit from the semaphore (blocks until one is available). - pub async fn acquire_permit(&self) -> OwnedSemaphorePermit { + pub async fn acquire_permit(&self) -> Option { self.semaphore .clone() .acquire_owned() .await - .expect("semaphore closed") + .ok() } /// Park an already-acquired permit under `id`. @@ -219,7 +221,10 @@ impl QueueManager { return false; } - let permit = self.acquire_permit().await; + let permit = match self.acquire_permit().await { + Some(p) => p, + None => return false, + }; let mut permits = self.active_permits.lock().await; if permits.contains_key(id) { drop(permits); @@ -239,6 +244,7 @@ impl QueueManager { let removed = self.active_permits.lock().await.remove(id).is_some(); self.active_kinds.lock().await.remove(id); if removed { + self.notify_permit_released.notify_waiters(); self.notify.notify_one(); } } @@ -323,12 +329,16 @@ impl QueueManager { continue; } // (2) Acquire a slot. - let permit = self + let permit_opt = self .semaphore .clone() .acquire_owned() .await - .expect("semaphore closed"); + .ok(); + let permit = match permit_opt { + Some(p) => p, + None => break, // Semaphore closed, exit dispatcher + }; // (3) CAS retirement — never underflows to usize::MAX. let mut retired = false; let mut debt = self.slots_to_retire.load(Ordering::Relaxed); @@ -561,7 +571,14 @@ impl QueueManager { if !self.active_permits.lock().await.contains_key(id) { return; } - tokio::time::sleep(Duration::from_millis(100)).await; + let notified = self.notify_permit_released.notified(); + if !self.active_permits.lock().await.contains_key(id) { + return; + } + tokio::select! { + _ = notified => {} + _ = tokio::time::sleep(std::time::Duration::from_secs(5)) => {} + } } } @@ -888,7 +905,7 @@ impl SidecarSpawner for ProductionSpawner { 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 { + match crate::rpc_call(state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), &state.aria2_secret, "aria2.addUri", params).await { Ok(result) => { let gid = result.as_str().unwrap_or("").to_string(); if gid.is_empty() { @@ -931,7 +948,7 @@ impl SidecarSpawner for ProductionSpawner { async fn remove_uri(&self, gid: &str) -> Result<(), String> { let state = self.app_handle.state::(); let result = crate::rpc_call( - state.aria2_port, + state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), &state.aria2_secret, "aria2.forceRemove", serde_json::json!([gid]), diff --git a/src-tauri/src/settings.rs b/src-tauri/src/settings.rs index a80f3aa..d3351f2 100644 --- a/src-tauri/src/settings.rs +++ b/src-tauri/src/settings.rs @@ -50,8 +50,17 @@ pub fn preserve_scheduler_runtime_keys( let Some(existing) = existing else { return Ok(incoming.to_string()); }; - let existing_document = decode_document(&Value::String(existing.to_string()))?; - let existing_state = settings_state(&existing_document)?; + let existing_document = match decode_document(&Value::String(existing.to_string())) { + Ok(doc) => doc, + Err(e) => { + log::warn!("Failed to decode existing settings, dropping runtime keys: {}", e); + return Ok(incoming.to_string()); + } + }; + let existing_state = match settings_state(&existing_document) { + Ok(state) => state, + Err(_) => return Ok(incoming.to_string()), + }; let mut incoming_document = decode_document(&Value::String(incoming.to_string()))?; let incoming_state = settings_state_mut(&mut incoming_document)?; for key in ["schedulerLastStartKey", "schedulerLastStopKey"] { diff --git a/src/App.tsx b/src/App.tsx index 0847602..60eaa77 100644 --- a/src/App.tsx +++ b/src/App.tsx @@ -513,6 +513,7 @@ function App() { }, [theme]); useEffect(() => { + if (!coreReady) return; const unlistenDownload = initDownloadListener(); const unlistenTerminalState = listen('download-state', (event) => { @@ -571,7 +572,7 @@ function App() { unlistenDeepLink.then(f => f()); unlistenDownload.then(f => { if (f) f(); }); }; - }, []); + }, [coreReady]); return (
diff --git a/src/components/AddDownloadsModal.tsx b/src/components/AddDownloadsModal.tsx index e8b6b24..a15de75 100644 --- a/src/components/AddDownloadsModal.tsx +++ b/src/components/AddDownloadsModal.tsx @@ -107,7 +107,7 @@ export const AddDownloadsModal = () => { setChecksumAlgo('SHA-256'); setChecksumValue(''); setHeaders([ - pendingAddReferer ? `Referer: ${pendingAddReferer}` : '', + pendingAddReferer ? `Referer: ${pendingAddReferer.replace(/[\r\n]/g, '')}` : '', pendingAddHeaders ].filter(Boolean).join('\n')); setCookies(pendingAddCookies);