fix(download): harden queue and lifecycle recovery

- Preserve selected queue priority during bulk starts and sparse moves
- Reconcile ambiguous Aria2 admission and bound metadata retries
- Snapshot scheduled stops and keep pause/completion state authoritative
- Bound shutdown waits and retain recoverable tray actions
- Cover staged pauses, queue races, and retry boundaries
This commit is contained in:
NimBold
2026-10-10 01:49:52 +03:30
parent 1c5895156a
commit 33b61448bc
16 changed files with 1444 additions and 187 deletions
+11
View File
@@ -689,6 +689,14 @@ pub struct SchedulerSettings {
pub post_queue_action: PostQueueAction,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct SchedulerRunSnapshot {
pub start_key: String,
pub stop_key: Option<String>,
pub stop_minute: Option<u32>,
}
#[derive(Clone, Debug, Serialize, Deserialize, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export, export_to = "../../src/bindings/")]
@@ -738,6 +746,9 @@ pub struct PersistedSettings {
#[ts(optional)]
pub scheduler_triggered_start_key: Option<String>,
pub scheduler_last_stop_key: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
#[ts(skip)]
pub scheduler_active_run_snapshot: Option<SchedulerRunSnapshot>,
pub last_custom_speed_limit_ki_b: u32,
#[serde(default = "default_speed_limit_unit")]
pub last_custom_speed_limit_unit: String,
+355 -59
View File
@@ -270,6 +270,113 @@ fn retry_metadata_with_cookies(
true
}
async fn wait_for_metadata_retry_with<S, Fut>(
retry_strike: &mut usize,
reason: &str,
retryable: bool,
mut sleep: S,
) -> bool
where
S: FnMut(Duration) -> Fut,
Fut: std::future::Future<Output = ()>,
{
if !retryable || *retry_strike >= crate::retry::MAX_RETRIES {
return false;
}
let delay = crate::retry::backoff_for_error(*retry_strike, reason);
log::debug!(
"metadata retry [retry={} error_class={} delay_ms={}]",
*retry_strike + 1,
crate::retry::network_error_class(reason),
delay.as_millis()
);
sleep(delay).await;
*retry_strike += 1;
true
}
/// Retry idempotent metadata reads after their fallback method has been
/// selected, sharing one bounded retry budget across ranged GETs, redirects,
/// and auth retry. HEAD failures still switch promptly to the GET fallback.
/// Transient HTTP statuses are retried as well as connection failures.
async fn retry_metadata_request_with<T, E, F, Fut, C, S, SleepFut>(
retry_strike: &mut usize,
mut send: F,
classify_response: C,
mut sleep: S,
) -> Result<T, E>
where
F: FnMut() -> Fut,
Fut: std::future::Future<Output = Result<T, E>>,
E: std::fmt::Display,
C: Fn(&T) -> String,
S: FnMut(Duration) -> SleepFut,
SleepFut: std::future::Future<Output = ()>,
{
loop {
match send().await {
Ok(response) => {
let reason = classify_response(&response);
if wait_for_metadata_retry_with(
retry_strike,
&reason,
crate::retry::is_transient_network_error(&reason),
&mut sleep,
)
.await
{
drop(response);
continue;
}
return Ok(response);
}
Err(error) => {
let reason = error.to_string();
if !wait_for_metadata_retry_with(
retry_strike,
&reason,
crate::retry::is_transient_network_error(&reason),
&mut sleep,
)
.await
{
return Err(error);
}
}
}
}
}
async fn send_metadata_request_with_retry<F, Fut>(
retry_strike: &mut usize,
send: F,
) -> Result<reqwest::Response, reqwest::Error>
where
F: FnMut() -> Fut,
Fut: std::future::Future<Output = Result<reqwest::Response, reqwest::Error>>,
{
retry_metadata_request_with(
retry_strike,
send,
|response: &reqwest::Response| format!("HTTP {}", response.status().as_u16()),
tokio::time::sleep,
)
.await
}
fn metadata_transport_error_message(operation: &str, error: &reqwest::Error) -> String {
let reason = match reqwest_error_code(error) {
"timeout" => "request timed out",
"dns" => "DNS lookup failed",
"connect" => "connection could not be established",
"body" => "response could not be read",
"request" => "request was invalid",
_ => "network error",
};
format!("{operation} metadata request failed: {reason}")
}
#[derive(Serialize, TS)]
#[ts(export, export_to = "../../src/bindings/")]
pub struct MetadataResponse {
@@ -1916,6 +2023,7 @@ async fn fetch_remote_torrent_bytes(
let cookies_available = metadata_cookie_header_present(headers, cookies, cookie_scopes);
let mut send_cookies = !cookies_available;
let mut cookie_retry_attempted = false;
let mut network_retry_strike = 0;
let mut redirect_count = 0;
loop {
@@ -1965,20 +2073,22 @@ async fn fetch_remote_torrent_bytes(
let client = builder
.build()
.map_err(|error| format!("could not create torrent metadata client: {error}"))?;
let response = client
.get(current.clone())
.header(
reqwest::header::ACCEPT,
"application/x-bittorrent, application/octet-stream",
)
.send()
.await
.map_err(|error| {
format!(
"remote torrent metadata request failed: {}",
crate::redact_sensitive_text(&error.to_string())
let response = send_metadata_request_with_retry(&mut network_retry_strike, || {
client
.get(current.clone())
.header(
reqwest::header::ACCEPT,
"application/x-bittorrent, application/octet-stream",
)
})?;
.send()
})
.await
.map_err(|error| {
format!(
"remote torrent metadata request failed: {}",
crate::redact_sensitive_text(&error.to_string())
)
})?;
if should_retry_metadata_with_cookies(
response.status(),
@@ -2027,13 +2137,34 @@ async fn fetch_remote_torrent_bytes(
let mut bytes = Vec::new();
let mut stream = response.bytes_stream();
let mut retry_body = false;
while let Some(chunk) = stream.next().await {
let chunk = chunk.map_err(|error| {
format!(
"remote torrent metadata response failed: {}",
crate::redact_sensitive_text(&error.to_string())
)
})?;
let chunk = match chunk {
Ok(chunk) => chunk,
Err(error) => {
let reason = error.to_string();
let retryable_body_error = error.is_timeout()
|| error.is_connect()
|| error.is_body()
|| crate::retry::is_transient_network_error(&reason);
if retryable_body_error
&& wait_for_metadata_retry_with(
&mut network_retry_strike,
&reason,
true,
tokio::time::sleep,
)
.await
{
retry_body = true;
break;
}
return Err(format!(
"remote torrent metadata response failed: {}",
crate::redact_sensitive_text(&reason)
));
}
};
if bytes
.len()
.checked_add(chunk.len())
@@ -2046,6 +2177,9 @@ async fn fetch_remote_torrent_bytes(
}
bytes.extend_from_slice(&chunk);
}
if retry_body {
continue;
}
if bytes.is_empty() {
return Err("remote torrent metadata response was empty".to_string());
}
@@ -2128,6 +2262,7 @@ async fn fetch_metadata(
let mut current_url = url.clone();
let original_origin = reqwest::Url::parse(&url).ok();
let mut redirects = 0;
let mut network_retry_strike = 0;
let cookies_available = metadata_cookie_header_present(
headers.as_deref(),
cookies.as_deref(),
@@ -2313,7 +2448,11 @@ async fn fetch_metadata(
Err(head_error) => {
let head_elapsed_ms = head_started.elapsed().as_millis();
let get_started = Instant::now();
match build_get_range().send().await {
match send_metadata_request_with_retry(&mut network_retry_strike, || {
build_get_range().send()
})
.await
{
Ok(response) => {
log::warn!(
"metadata [stage=metadata operation=head result=fallback host={} error_code={} head_elapsed_ms={} get_status={} get_elapsed_ms={} total_elapsed_ms={}]",
@@ -2378,7 +2517,11 @@ async fn fetch_metadata(
if needs_fallback {
let get_started = Instant::now();
current_res = match build_get_range().send().await {
current_res = match send_metadata_request_with_retry(&mut network_retry_strike, || {
build_get_range().send()
})
.await
{
Ok(response) => {
log::info!(
"metadata [stage=metadata operation=get_range host={} result=ok status={} elapsed_ms={} total_elapsed_ms={}]",
@@ -2397,7 +2540,7 @@ async fn fetch_metadata(
get_started.elapsed().as_millis(),
metadata_started.elapsed().as_millis()
);
return Err(error.to_string());
return Err(metadata_transport_error_message("Ranged GET", &error));
}
};
@@ -4252,15 +4395,15 @@ async fn shutdown_aria2_daemon(app_handle: tauri::AppHandle) {
}
};
(outcome, child_wait_started.elapsed())
})
.await;
match child_wait {
Ok((outcome, elapsed)) => log::info!(
});
match tokio::time::timeout(std::time::Duration::from_secs(10), child_wait).await {
Ok(Ok((outcome, elapsed))) => log::info!(
"app_shutdown_phase=aria2_child outcome={} elapsed_ms={}",
outcome,
elapsed.as_millis()
),
Err(error) => log::warn!("aria2 child wait task failed: {error}"),
Ok(Err(error)) => log::warn!("aria2 child wait task failed: {error}"),
Err(_) => log::warn!("app_shutdown_phase=aria2_child outcome=timeout"),
}
}
log::info!(
@@ -4269,6 +4412,27 @@ async fn shutdown_aria2_daemon(app_handle: tauri::AppHandle) {
);
}
fn reject_app_exit_request(app: &tauri::AppHandle, request: u64, reason: &str) {
app.state::<Aria2DaemonGuard>().cancel_shutdown();
let tray_action_pending = app
.state::<PendingTrayActionQueue>()
.reopen_after_exit_rejected();
if tray_action_pending {
if let Err(error) = app.emit("tray-action-available", ()) {
log::warn!("pending tray action wakeup failed after exit rejection: {error}");
}
}
let _ = app.emit_to(
"main",
"app-exit-cancelled",
serde_json::json!({
"requestId": request.to_string(),
"reason": reason
}),
);
restore_main_window(app);
}
impl Drop for Aria2DaemonGuard {
fn drop(&mut self) {
if let Ok(mut lock) = self.child.lock() {
@@ -6107,7 +6271,7 @@ async fn pause_download(
app_handle: tauri::AppHandle,
state: tauri::State<'_, AppState>,
id: String,
) -> Result<(), String> {
) -> Result<bool, String> {
properties_window::ensure_main_window(&caller)?;
log::info!("pause_download called for id: {}", id);
@@ -6201,7 +6365,7 @@ async fn pause_download(
progress.or(status_progress.clone()),
)
.await;
return Ok(());
return Ok(false);
}
Ok((status, progress)) if matches!(status.as_str(), "error" | "removed") => {
state.queue_manager.clear_aria2_allocation(&id).await;
@@ -6274,7 +6438,7 @@ async fn pause_download(
status_progress.clone(),
)
.await;
return Ok(());
return Ok(false);
}
terminal => {
state.queue_manager.clear_aria2_allocation(&id).await;
@@ -6300,7 +6464,7 @@ async fn pause_download(
event = event.with_progress(progress);
}
let _ = app_handle.emit("download-state", event);
return Ok(());
return Ok(true);
}
state.queue_manager.release_registered_id(&id).await;
return Err(format!(
@@ -6331,14 +6495,14 @@ async fn pause_download(
event = event.with_progress(progress);
}
let _ = app_handle.emit("download-state", event);
return Ok(());
return Ok(true);
}
// A terminal runner may have already released its lifecycle while this
// command was waiting for the per-download control lock. Treat pause as
// an idempotent no-op instead of emitting a stale Paused state.
if active_kind.is_none() && !removed_pending && registered_lifecycle_generation.is_none() {
return Ok(());
return Ok(false);
}
if matches!(active_kind, Some(crate::queue::TaskKind::Aria2)) {
@@ -6394,7 +6558,7 @@ async fn pause_download(
"download-state",
crate::ipc::DownloadStateEvent::paused_with_seed_remaining(id, seed_remaining),
);
Ok(())
Ok(true)
}
#[tauri::command]
@@ -9468,6 +9632,13 @@ fn ack_schedule_trigger(
}
"stop" => {
state.insert("schedulerLastStopKey".to_string(), serde_json::json!(key));
let snapshot_stop_key = state
.get("schedulerActiveRunSnapshot")
.and_then(|snapshot| snapshot.get("stopKey"))
.and_then(serde_json::Value::as_str);
if snapshot_stop_key == Some(key.as_str()) {
state.remove("schedulerActiveRunSnapshot");
}
}
_ => {}
})?;
@@ -9479,7 +9650,14 @@ fn ack_schedule_trigger(
settings.scheduler_last_start_key = key;
settings.scheduler_triggered_start_key = None;
} else {
settings.scheduler_last_stop_key = key;
settings.scheduler_last_stop_key = key.clone();
if settings
.scheduler_active_run_snapshot
.as_ref()
.is_some_and(|snapshot| snapshot.stop_key.as_deref() == Some(key.as_str()))
{
settings.scheduler_active_run_snapshot = None;
}
}
}
}
@@ -9499,6 +9677,48 @@ async fn get_pending_order(
Ok(state.queue_manager.pending_order(queue_id.as_deref()).await)
}
#[tauri::command]
async fn begin_queue_dispatch_hold(
caller: tauri::WebviewWindow,
state: tauri::State<'_, AppState>,
queue_id: String,
) -> Result<String, AppError> {
properties_window::ensure_main_window(&caller).map_err(AppError::Internal)?;
state
.queue_manager
.begin_queue_dispatch_hold(&queue_id)
.await
.map_err(AppError::Internal)
}
#[tauri::command]
async fn renew_queue_dispatch_hold(
caller: tauri::WebviewWindow,
state: tauri::State<'_, AppState>,
queue_id: String,
token: String,
) -> Result<bool, AppError> {
properties_window::ensure_main_window(&caller).map_err(AppError::Internal)?;
Ok(state
.queue_manager
.renew_queue_dispatch_hold(&queue_id, &token)
.await)
}
#[tauri::command]
async fn end_queue_dispatch_hold(
caller: tauri::WebviewWindow,
state: tauri::State<'_, AppState>,
queue_id: String,
token: String,
) -> Result<bool, AppError> {
properties_window::ensure_main_window(&caller).map_err(AppError::Internal)?;
Ok(state
.queue_manager
.end_queue_dispatch_hold(&queue_id, &token)
.await)
}
fn enqueue_lifecycle_generation(item: &queue::EnqueueItem) -> Result<u64, String> {
item.lifecycle_generation
.as_deref()
@@ -15869,7 +16089,7 @@ mod tests {
media_output_template, media_progress_args, media_progress_event_totals,
media_progress_speed,
cookie_scope_for_url, metadata_authentication_error, metadata_cookie_header_present,
metadata_headers, metadata_response_error,
metadata_headers, metadata_response_error, metadata_transport_error_message,
metadata_error_code,
reqwest_error_code,
is_remote_torrent_source,
@@ -15895,6 +16115,7 @@ mod tests {
browser_cookie_fallback, normalize_media_cookie_file_for_request,
should_fallback_from_browser_cookies, should_fallback_from_netscape_cookie_file,
retry_metadata_with_cookies, should_retry_metadata_with_cookies,
retry_metadata_request_with,
should_send_metadata_credentials, collect_log_files, FirelinkDeepLink,
ytdlp_youtube_extractor_args, YTDLP_YOUTUBE_PLAYER_CLIENTS,
percent_decode_metadata_value, MediaProgress,
@@ -15982,6 +16203,85 @@ mod tests {
assert!(error.is_dns());
assert_eq!(reqwest_error_code(&error), "dns");
assert_eq!(
metadata_transport_error_message("Ranged GET", &error),
"Ranged GET metadata request failed: DNS lookup failed"
);
}
#[tokio::test]
async fn metadata_request_retries_transient_statuses_and_stops_on_permanent_statuses() {
let mut retry_strike = 0;
let mut attempts = 0;
let status = retry_metadata_request_with(
&mut retry_strike,
|| {
attempts += 1;
std::future::ready(Ok::<u16, String>(if attempts == 1 { 503 } else { 200 }))
},
|status| format!("HTTP {status}"),
|_| std::future::ready(()),
)
.await
.expect("the retry should return the successful response");
assert_eq!(status, 200);
assert_eq!(attempts, 2);
assert_eq!(retry_strike, 1);
let mut retry_strike = 0;
let mut attempts = 0;
let status = retry_metadata_request_with(
&mut retry_strike,
|| {
attempts += 1;
std::future::ready(if attempts == 1 {
Err("operation timed out".to_string())
} else {
Ok(200_u16)
})
},
|status| format!("HTTP {status}"),
|_| std::future::ready(()),
)
.await
.expect("a transient transport failure should retry");
assert_eq!(status, 200);
assert_eq!(attempts, 2);
assert_eq!(retry_strike, 1);
let mut retry_strike = 0;
let mut attempts = 0;
let status = retry_metadata_request_with(
&mut retry_strike,
|| {
attempts += 1;
std::future::ready(Ok::<u16, String>(404))
},
|status| format!("HTTP {status}"),
|_| std::future::ready(()),
)
.await
.expect("permanent responses are returned to the caller");
assert_eq!(status, 404);
assert_eq!(attempts, 1);
assert_eq!(retry_strike, 0);
let mut retry_strike = crate::retry::MAX_RETRIES;
let mut attempts = 0;
let status = retry_metadata_request_with(
&mut retry_strike,
|| {
attempts += 1;
std::future::ready(Ok::<u16, String>(503))
},
|status| format!("HTTP {status}"),
|_| std::future::ready(()),
)
.await
.expect("exhausted retries leave the final response for normal error handling");
assert_eq!(status, 503);
assert_eq!(attempts, 1);
assert_eq!(retry_strike, crate::retry::MAX_RETRIES);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
@@ -23014,6 +23314,9 @@ pub fn run() {
cancel_enqueue_generation,
move_in_queue,
move_many_in_queue,
begin_queue_dispatch_hold,
renew_queue_dispatch_hold,
end_queue_dispatch_hold,
remove_from_queue,
get_pending_order,
]);
@@ -23158,6 +23461,9 @@ pub fn run() {
| "cancel_enqueue_generation"
| "move_in_queue"
| "move_many_in_queue"
| "begin_queue_dispatch_hold"
| "renew_queue_dispatch_hold"
| "end_queue_dispatch_hold"
| "remove_from_queue"
| "get_pending_order" => download_queue_handler(invoke),
"reveal_in_file_manager"
@@ -23242,9 +23548,18 @@ pub fn run() {
.clone();
let request = frontend_exit_flush.request();
let mut completed = frontend_exit_flush.completed.subscribe();
let _ = app.emit_to("main", "app-exit-requested", serde_json::json!({
if let Err(error) = app.emit_to("main", "app-exit-requested", serde_json::json!({
"requestId": request.to_string()
}));
})) {
log::warn!("app_shutdown_phase=frontend_flush signal failed: {error}");
frontend_exit_flush.finish(request, false);
reject_app_exit_request(&app, request, "signal-failed");
log::info!(
"app_shutdown_phase=total outcome=cancelled elapsed_ms={}",
exit_started.elapsed().as_millis()
);
return;
}
let flush_wait = async move {
loop {
let status = *completed.borrow_and_update();
@@ -23301,26 +23616,7 @@ pub fn run() {
} else {
"rejected"
};
app.state::<Aria2DaemonGuard>().cancel_shutdown();
let tray_action_pending = app
.state::<PendingTrayActionQueue>()
.reopen_after_exit_rejected();
if tray_action_pending {
if let Err(error) = app.emit("tray-action-available", ()) {
log::warn!(
"pending tray action wakeup failed after exit rejection: {error}"
);
}
}
let _ = app.emit_to(
"main",
"app-exit-cancelled",
serde_json::json!({
"requestId": request.to_string(),
"reason": cancel_reason
}),
);
restore_main_window(&app);
reject_app_exit_request(&app, request, cancel_reason);
log::info!(
"app_shutdown_phase=total outcome=cancelled elapsed_ms={}",
exit_started.elapsed().as_millis()
@@ -23358,7 +23654,7 @@ pub fn run() {
}
};
match tokio::time::timeout(
Duration::from_secs(60),
Duration::from_secs(5),
final_flush_wait,
)
.await
+371 -21
View File
@@ -28,6 +28,9 @@ const MAX_PENDING_DOWNLOAD_STARTS: usize = 1024;
pub const MEDIA_RUN_CANCELLED: &str = "__firelink_media_run_cancelled__";
pub const DOWNLOAD_CONNECTIONS_MIN: i32 = 1;
pub const DOWNLOAD_CONNECTIONS_MAX: i32 = 16;
// The frontend refreshes this lease every 20 seconds. A bounded expiry
// releases a queue if the renderer disappears before its finally cleanup.
const QUEUE_START_BATCH_HOLD_LEASE: Duration = Duration::from_secs(90);
fn encode_aria2_torrent_payload(bytes: &[u8]) -> String {
base64::engine::general_purpose::STANDARD.encode(bytes)
@@ -1277,6 +1280,10 @@ pub struct QueueManager<R: tauri::Runtime = tauri::Wry> {
/// Queue overrides are stored only for queues with an explicit limit.
/// Missing entries inherit the global target capacity.
queue_limits: Mutex<HashMap<String, usize>>,
/// Queue-scoped leases used while a multi-row start request re-enqueues
/// paused downloads. The lease prevents older pending rows from taking a
/// newly freed slot before the selected block has been restored to order.
queue_dispatch_holds: Mutex<HashMap<String, HashMap<String, Instant>>>,
/// One entry represents either a queued dispatch reservation or an active
/// transfer. Keeping both phases in one map makes queue-slot ownership
/// exactly-once across the async addUri handoff.
@@ -1408,6 +1415,7 @@ impl<R: tauri::Runtime> QueueManager<R> {
active_permit_generations: Mutex::new(HashMap::new()),
active_kinds: Mutex::new(HashMap::new()),
queue_limits: Mutex::new(HashMap::new()),
queue_dispatch_holds: Mutex::new(HashMap::new()),
queue_permit_ownership: Mutex::new(HashMap::new()),
admission_gate: Mutex::new(()),
system_action_pending: AtomicBool::new(false),
@@ -2387,6 +2395,101 @@ impl<R: tauri::Runtime> QueueManager<R> {
Ok(())
}
/// Begin a bounded queue-scoped dispatch hold for a multi-row start
/// operation. Other queues remain eligible while this queue is held.
pub async fn begin_queue_dispatch_hold(&self, queue_id: &str) -> Result<String, String> {
let queue_id = queue_id.trim();
if queue_id.is_empty() {
return Err("Queue id cannot be empty".to_string());
}
let _admission_gate = self.admission_gate.lock().await;
if self.system_action_pending.load(Ordering::Acquire) {
return Err("System action is already being performed".to_string());
}
let now = Instant::now();
let token = uuid::Uuid::new_v4().simple().to_string();
let mut holds = self.queue_dispatch_holds.lock().await;
let queue_holds = holds.entry(queue_id.to_string()).or_default();
queue_holds.retain(|_, expires_at| *expires_at > now);
queue_holds.insert(token.clone(), now + QUEUE_START_BATCH_HOLD_LEASE);
Ok(token)
}
/// Renew a live start-batch lease. An expired token cannot be resurrected;
/// the caller must stop relying on its queue-order guarantee in that case.
pub async fn renew_queue_dispatch_hold(&self, queue_id: &str, token: &str) -> bool {
let _admission_gate = self.admission_gate.lock().await;
let queue_id = queue_id.trim();
if queue_id.is_empty() {
return false;
}
let now = Instant::now();
let mut holds = self.queue_dispatch_holds.lock().await;
let Some(queue_holds) = holds.get_mut(queue_id) else {
return false;
};
let Some(expires_at) = queue_holds.get_mut(token) else {
return false;
};
if *expires_at <= now {
queue_holds.remove(token);
if queue_holds.is_empty() {
holds.remove(queue_id);
}
return false;
}
*expires_at = now + QUEUE_START_BATCH_HOLD_LEASE;
true
}
/// Release only the lease named by `token`, so late cleanup from an older
/// start request cannot release a newer request's hold.
pub async fn end_queue_dispatch_hold(&self, queue_id: &str, token: &str) -> bool {
let _admission_gate = self.admission_gate.lock().await;
let queue_id = queue_id.trim();
if queue_id.is_empty() {
return false;
}
let mut holds = self.queue_dispatch_holds.lock().await;
let removed = holds
.get_mut(queue_id)
.is_some_and(|queue_holds| queue_holds.remove(token).is_some());
if holds.get(queue_id).is_some_and(HashMap::is_empty) {
holds.remove(queue_id);
}
drop(holds);
drop(_admission_gate);
if removed {
self.notify.notify_one();
}
removed
}
async fn active_queue_dispatch_holds(&self) -> HashSet<String> {
let now = Instant::now();
let mut holds = self.queue_dispatch_holds.lock().await;
holds.retain(|_, queue_holds| {
queue_holds.retain(|_, expires_at| *expires_at > now);
!queue_holds.is_empty()
});
holds.keys().cloned().collect()
}
async fn next_queue_dispatch_hold_delay(&self) -> Option<Duration> {
let now = Instant::now();
let mut holds = self.queue_dispatch_holds.lock().await;
holds.retain(|_, queue_holds| {
queue_holds.retain(|_, expires_at| *expires_at > now);
!queue_holds.is_empty()
});
holds
.values()
.flat_map(HashMap::values)
.min()
.map(|expires_at| expires_at.saturating_duration_since(now))
}
pub async fn next_aria2_control_epoch(&self, id: &str) -> u64 {
let mut epochs = self.aria2_control_epochs.lock().await;
let epoch = epochs.get(id).copied().unwrap_or_default().wrapping_add(1);
@@ -3848,11 +3951,15 @@ impl<R: tauri::Runtime> QueueManager<R> {
return None;
}
let held_queues = self.active_queue_dispatch_holds().await;
let ownership = self.queue_permit_ownership.lock().await;
let mut queue_ids = Vec::new();
let mut seen = HashSet::new();
for task in pending.iter() {
if !ownership.contains_key(&task.id) && seen.insert(task.queue_id.clone()) {
if !ownership.contains_key(&task.id)
&& !held_queues.contains(&task.queue_id)
&& seen.insert(task.queue_id.clone())
{
queue_ids.push(task.queue_id.clone());
}
}
@@ -4441,7 +4548,14 @@ impl<R: tauri::Runtime> QueueManager<R> {
// all queue/global capacity is occupied. The notification
// future is created before inspection to close the lost-wake
// window without polling or sleeping.
notified.await;
if let Some(hold_delay) = self.next_queue_dispatch_hold_delay().await {
tokio::select! {
_ = notified => {}
_ = tokio::time::sleep(hold_delay) => {}
}
} else {
notified.await;
}
}
}
}
@@ -6478,11 +6592,6 @@ impl<R: tauri::Runtime> QueueManager<R> {
.map(|index| pending[*index].clone())
.collect::<Vec<_>>();
let selected_ids = ids.iter().collect::<HashSet<_>>();
let selected_tasks = queue_tasks
.iter()
.filter(|task| selected_ids.contains(&task.id))
.cloned()
.collect::<Vec<_>>();
let unselected_tasks = queue_tasks
.iter()
.filter(|task| !selected_ids.contains(&task.id))
@@ -6490,12 +6599,13 @@ impl<R: tauri::Runtime> QueueManager<R> {
.collect::<Vec<_>>();
let first_selected = *selected_positions.first().unwrap();
let last_selected = *selected_positions.last().unwrap();
let selected_count = selected_tasks.len();
let insert_index = match direction {
QueueDirection::Up => first_selected.saturating_sub(1),
QueueDirection::Down => (last_selected + 1)
.saturating_sub(selected_count)
// Move the selected block one slot after its current
// insertion point among unselected tasks. Using the last
// selected index and selected count collapses sparse
// selections too far toward the end.
QueueDirection::Down => first_selected
.saturating_add(1)
.min(unselected_tasks.len()),
};
@@ -7364,6 +7474,97 @@ fn aria2_add_uri_params(
serde_json::json!([transfer_uris, options])
}
fn new_aria2_transfer_gid() -> String {
// Aria2 GIDs are 16 hexadecimal characters. A random GID also acts as an
// idempotency key when an addUri/addTorrent response is lost: every retry
// carries the same value, so Aria2 cannot admit a second transfer.
uuid::Uuid::new_v4().simple().to_string()[..16].to_string()
}
fn apply_aria2_transfer_gid(
options: &mut serde_json::Map<String, serde_json::Value>,
gid: &str,
) {
options.insert("gid".to_string(), serde_json::json!(gid));
}
async fn add_transfer_rpc_with_reconciliation<AddCall, AddFuture, StatusCall, StatusFuture>(
method: &str,
expected_gid: &str,
deadline: Instant,
mut add_call: AddCall,
mut status_call: StatusCall,
) -> Result<serde_json::Value, String>
where
AddCall: FnMut() -> AddFuture,
AddFuture: Future<Output = Result<serde_json::Value, String>>,
StatusCall: FnMut() -> StatusFuture,
StatusFuture: Future<Output = Result<String, String>>,
{
let mut ambiguous_add = false;
let mut status_only = false;
loop {
if !status_only {
match add_call().await {
Ok(result) => return Ok(result),
Err(error) => {
let transient = is_aria2_rpc_unavailable(&error);
if transient {
ambiguous_add = true;
}
if !transient && !ambiguous_add {
return Err(error);
}
// A transport timeout/reset does not tell us whether
// Aria2 admitted the transfer before the response was
// lost. Query the stable GID before retrying the add.
// If that status query is itself unavailable, retries
// still use the same GID and remain idempotent.
if status_call().await.is_ok() {
return Ok(serde_json::json!(expected_gid));
}
// Once a retry returns an application error after an
// earlier ambiguous transport failure, keep checking the
// original GID instead of repeatedly issuing an add that
// Aria2 may already have accepted.
if !transient {
status_only = true;
}
if Instant::now() >= deadline {
if ambiguous_add {
log::warn!(
"aria2 {method} for gid {expected_gid} remained ambiguous after retry reconciliation; handing the GID to the queue poller"
);
return Ok(serde_json::json!(expected_gid));
}
return Err(error);
}
}
}
}
if status_only {
if status_call().await.is_ok() {
return Ok(serde_json::json!(expected_gid));
}
if Instant::now() >= deadline {
// The first add may have succeeded even though both its
// response and subsequent status checks were unavailable.
// Returning the stable GID lets normal queue reconciliation
// either retain the live transfer or recover a missing one.
log::warn!(
"aria2 {method} for gid {expected_gid} could not be verified; handing the GID to the queue poller"
);
return Ok(serde_json::json!(expected_gid));
}
}
tokio::time::sleep(Duration::from_millis(250)).await;
}
}
fn apply_aria2_normal_reliability_options(
options: &mut serde_json::Map<String, serde_json::Value>,
payload: &SpawnPayload,
@@ -8350,6 +8551,7 @@ impl ProductionSpawner {
state: &crate::AppState,
method: &str,
params: &serde_json::Value,
expected_gid: &str,
) -> Result<serde_json::Value, String> {
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(15);
loop {
@@ -8373,15 +8575,28 @@ impl ProductionSpawner {
continue;
}
match crate::rpc_call(port, &state.aria2_secret, method, params.clone()).await {
Ok(result) => return Ok(result),
Err(error) => {
if !is_aria2_rpc_unavailable(&error) || std::time::Instant::now() >= deadline {
return Err(error);
}
tokio::time::sleep(std::time::Duration::from_millis(250)).await;
}
}
let secret = state.aria2_secret.clone();
let secret_for_status = secret.clone();
let method_for_call = method.to_string();
let params_for_call = params.clone();
let gid_for_status = expected_gid.to_string();
return add_transfer_rpc_with_reconciliation(
method,
expected_gid,
deadline,
move || {
let secret = secret.clone();
let method = method_for_call.clone();
let params = params_for_call.clone();
async move { crate::rpc_call(port, &secret, &method, params).await }
},
move || {
let secret = secret_for_status.clone();
let gid = gid_for_status.clone();
async move { crate::aria2_download_status(port, &secret, &gid).await }
},
)
.await;
}
}
@@ -8447,6 +8662,8 @@ impl SidecarSpawner for ProductionSpawner {
let attempt_epoch = state.queue_manager.current_aria2_control_epoch(id).await;
let admission_started = Instant::now();
let mut options = serde_json::Map::new();
let expected_gid = new_aria2_transfer_gid();
apply_aria2_transfer_gid(&mut options, &expected_gid);
// Keep hostname resolution route-aware and non-blocking for every
// normal and Torrent transfer. When the Firelink route contract is
// available, transfers use NativeAsyncResolver on background worker
@@ -8674,7 +8891,10 @@ impl SidecarSpawner for ProductionSpawner {
return Err("aria2 admission canceled before RPC".to_string());
}
match self.add_transfer_rpc(&state, method, &params).await {
match self
.add_transfer_rpc(&state, method, &params, &expected_gid)
.await
{
Ok(result) => {
let gid = result.as_str().unwrap_or("").to_string();
if gid.is_empty() {
@@ -9451,6 +9671,35 @@ mod tests {
assert!(!manager.media_payload_completed("media-removal", 7).await);
}
#[tokio::test]
async fn expired_queue_dispatch_hold_cannot_be_renewed_and_stale_end_is_scoped() {
let app = tauri::test::mock_builder()
.build(tauri::test::mock_context(tauri::test::noop_assets()))
.unwrap();
let manager = QueueManager::test_new(app.handle().clone(), 1, Arc::new(TestSpawner));
let first = manager.begin_queue_dispatch_hold("queue").await.unwrap();
let second = manager.begin_queue_dispatch_hold("queue").await.unwrap();
assert!(manager.end_queue_dispatch_hold("queue", &first).await);
assert!(manager
.active_queue_dispatch_holds()
.await
.contains("queue"));
manager
.queue_dispatch_holds
.lock()
.await
.get_mut("queue")
.unwrap()
.insert(second.clone(), Instant::now() - Duration::from_secs(1));
assert!(!manager.renew_queue_dispatch_hold("queue", &second).await);
assert!(!manager
.active_queue_dispatch_holds()
.await
.contains("queue"));
}
#[tokio::test]
async fn explicit_allocation_telemetry_ends_without_payload_and_rejects_old_epoch() {
let app = tauri::test::mock_builder()
@@ -12352,6 +12601,107 @@ mod tests {
));
}
#[test]
fn aria2_transfer_gid_is_a_stable_16_digit_option() {
let gid = new_aria2_transfer_gid();
assert_eq!(gid.len(), 16);
assert!(gid
.chars()
.all(|character| character.is_ascii_hexdigit() && !character.is_ascii_uppercase()));
let mut options = serde_json::Map::new();
apply_aria2_transfer_gid(&mut options, &gid);
let params = aria2_add_uri_params(vec!["https://example.test/file".to_string()], options);
assert_eq!(params[1]["gid"], serde_json::json!(gid));
}
#[tokio::test]
async fn ambiguous_add_rpc_adopts_the_gid_after_a_lost_response() {
let gid = "0123456789abcdef";
let add_calls = Arc::new(AtomicUsize::new(0));
let status_calls = Arc::new(AtomicUsize::new(0));
let add_calls_for_rpc = Arc::clone(&add_calls);
let status_calls_for_rpc = Arc::clone(&status_calls);
let result = add_transfer_rpc_with_reconciliation(
"aria2.addUri",
gid,
Instant::now() + Duration::from_secs(2),
move || {
let attempt = add_calls_for_rpc.fetch_add(1, Ordering::SeqCst);
async move {
if attempt == 0 {
// Aria2 accepted the request, but the client lost the
// response and cannot yet query the daemon.
Err("error sending request: connection reset".to_string())
} else {
// The retry has the same GID, so Aria2 rejects a
// duplicate instead of creating a second transfer.
Err(format!("GID#{gid} already exists"))
}
}
},
move || {
let attempt = status_calls_for_rpc.fetch_add(1, Ordering::SeqCst);
async move {
if attempt == 0 {
Err("error trying to connect: connection refused".to_string())
} else {
Ok("active".to_string())
}
}
},
)
.await
.expect("the accepted GID should be adopted");
assert_eq!(result, serde_json::json!(gid));
assert_eq!(add_calls.load(Ordering::SeqCst), 2);
assert_eq!(status_calls.load(Ordering::SeqCst), 2);
}
#[tokio::test]
async fn nontransient_add_rejection_does_not_adopt_an_existing_gid() {
let gid = "0123456789abcdef";
let status_calls = Arc::new(AtomicUsize::new(0));
let status_calls_for_rpc = Arc::clone(&status_calls);
let result = add_transfer_rpc_with_reconciliation(
"aria2.addUri",
gid,
Instant::now() + Duration::from_secs(2),
|| async { Err("GID#0123456789abcdef is not unique".to_string()) },
move || {
status_calls_for_rpc.fetch_add(1, Ordering::SeqCst);
async { Ok("active".to_string()) }
},
)
.await;
assert!(result.is_err());
assert_eq!(status_calls.load(Ordering::SeqCst), 0);
}
#[tokio::test]
async fn unresolved_ambiguous_add_returns_gid_for_queue_reconciliation() {
let gid = "fedcba9876543210";
let add_calls = Arc::new(AtomicUsize::new(0));
let add_calls_for_rpc = Arc::clone(&add_calls);
let result = add_transfer_rpc_with_reconciliation(
"aria2.addTorrent",
gid,
Instant::now(),
move || {
add_calls_for_rpc.fetch_add(1, Ordering::SeqCst);
async { Err("request timed out".to_string()) }
},
|| async { Err("connection reset while checking status".to_string()) },
)
.await
.expect("ambiguous admission remains owned by normal queue reconciliation");
assert_eq!(result, serde_json::json!(gid));
assert_eq!(add_calls.load(Ordering::SeqCst), 1);
}
#[test]
fn known_aria2_startup_failure_is_returned_without_rpc_retry() {
assert_eq!(
+65 -20
View File
@@ -176,6 +176,25 @@ pub fn backoff_for(strike: usize) -> Duration {
.unwrap_or_else(|| *BACKOFF_SCHEDULE.last().expect("schedule is non-empty"))
}
/// Resolve the shared retry delay for an error without echoing the error into
/// diagnostics. Rate limiting gets its longer schedule only when the failure
/// actually contains an HTTP 429 status, not when an unrelated URL happens to
/// contain the digits `429`.
pub fn backoff_for_error(strike: usize, reason: &str) -> Duration {
let lower = reason.to_ascii_lowercase();
let is_rate_limited = contains_http_status(&lower, "429")
|| contains_labeled_http_status(&lower, "429")
|| lower.contains("429 too many requests");
if is_rate_limited {
BACKOFF_SCHEDULE_429
.get(strike)
.copied()
.unwrap_or_else(|| *BACKOFF_SCHEDULE_429.last().unwrap())
} else {
backoff_for(strike)
}
}
/// Classify an error string as a transient network condition worth retrying.
///
/// Returns `true` for socket drops, connect/read timeouts, connection resets,
@@ -232,6 +251,16 @@ fn contains_http_status(message: &str, status: &str) -> bool {
})
}
fn contains_labeled_http_status(message: &str, status: &str) -> bool {
let expected = format!("status={status}");
message.split_ascii_whitespace().any(|token| {
token
.trim_start_matches(|character| matches!(character, '(' | '['))
.trim_end_matches(|character: char| matches!(character, ')' | ']' | ',' | '.' | ';'))
== expected
})
}
pub fn is_transient_network_error(message: &str) -> bool {
if is_permanent_network_error(message) {
return false;
@@ -243,7 +272,7 @@ pub fn is_transient_network_error(message: &str) -> bool {
let m = message.to_ascii_lowercase();
const TRANSIENT: [&str; 36] = [
const TRANSIENT: [&str; 25] = [
// socket-layer / HTTP-client phrasing surfaced by aria2 and yt-dlp
"timed out",
"timeout",
@@ -269,17 +298,6 @@ pub fn is_transient_network_error(message: &str) -> bool {
"503 service unavailable",
"429 too many requests",
// aria2c HTTP error formats
"status=408",
"status=429",
"status=500",
"status=502",
"status=503",
"status=504",
"status=520",
"status=521",
"status=522",
"status=523",
"status=524",
// aria2c log phrasing
"connection was closed",
"timeout.",
@@ -292,6 +310,9 @@ pub fn is_transient_network_error(message: &str) -> bool {
.iter()
.any(|status| contains_http_status(&m, status))
|| TRANSIENT.iter().any(|t| m.contains(t))
|| TRANSIENT_HTTP_STATUS
.iter()
.any(|status| contains_labeled_http_status(&m, status))
}
/// Outcome of a cancel-safe backoff sleep wrapped around a transient retry.
@@ -314,14 +335,7 @@ pub async fn backoff_and_emit(
) -> BackoffOutcome {
let attempt = strike + 1;
emit(format!("Network drop — retry #{attempt}: {reason}"));
let delay = if reason.to_ascii_lowercase().contains("429") {
BACKOFF_SCHEDULE_429
.get(strike)
.copied()
.unwrap_or_else(|| *BACKOFF_SCHEDULE_429.last().unwrap())
} else {
backoff_for(strike)
};
let delay = backoff_for_error(strike, &reason);
tokio::select! {
_ = tokio::time::sleep(delay) => BackoffOutcome::Continue,
_ = interrupt => BackoffOutcome::Aborted,
@@ -414,6 +428,34 @@ mod tests {
assert_eq!(backoff_for(usize::MAX), Duration::from_secs(10));
}
#[test]
fn rate_limit_backoff_requires_an_http_429_status() {
assert_eq!(
backoff_for_error(0, "HTTP Error 429: Too Many Requests"),
Duration::from_secs(60)
);
assert_eq!(
backoff_for_error(1, "The response status is not successful. status=429"),
Duration::from_secs(120)
);
assert_eq!(
backoff_for_error(2, "HTTP 429 Too Many Requests"),
Duration::from_secs(300)
);
assert_eq!(
backoff_for_error(0, "https://example.test/file/429?token=x timed out"),
Duration::from_secs(2)
);
assert_eq!(
backoff_for_error(0, "https://example.test/file?status=429 connection failed"),
Duration::from_secs(2)
);
assert_eq!(
backoff_for_error(0, "request failed with status=429"),
Duration::from_secs(60)
);
}
// --- transient classification: positive cases -------------------------
#[test]
@@ -452,6 +494,9 @@ mod tests {
assert!(is_transient_network_error("HTTP Error 429: Too Many Requests"));
assert!(is_transient_network_error("429 too many requests"));
assert!(is_transient_network_error("The response status is not successful. status=429"));
assert!(!is_transient_network_error(
"https://example.test/file?status=429 connection failed"
));
}
#[test]
+146 -6
View File
@@ -1,4 +1,4 @@
use chrono::{Datelike, Local, Timelike};
use chrono::{Datelike, Local, NaiveDate, Timelike};
use std::collections::HashMap;
use std::sync::{Arc, RwLock};
use std::time::Duration;
@@ -71,16 +71,68 @@ fn overnight_stop_is_due(check: OvernightStopCheck<'_>) -> bool {
&& last_stop_key != stop_key
}
fn scheduler_run_snapshot(
start_key: &str,
scheduler: &crate::ipc::SchedulerSettings,
) -> crate::ipc::SchedulerRunSnapshot {
let stop = if scheduler.stop_time_enabled {
let start_minute = minute_of_day(&scheduler.start_time);
let stop_minute = minute_of_day(&scheduler.stop_time);
start_minute.zip(stop_minute).and_then(|(start, stop)| {
let date = start_key
.strip_suffix("-start")
.and_then(|date| NaiveDate::parse_from_str(date, "%Y-%m-%d").ok())?;
let stop_date = if stop < start { date.succ_opt()? } else { date };
Some((format!("{}-stop", stop_date.format("%Y-%m-%d")), stop))
})
} else {
None
};
crate::ipc::SchedulerRunSnapshot {
start_key: start_key.to_string(),
stop_key: stop.as_ref().map(|(key, _)| key.clone()),
stop_minute: stop.map(|(_, minute)| minute),
}
}
fn snapshot_stop_is_due(
snapshot: &crate::ipc::SchedulerRunSnapshot,
current_stop_key: &str,
current_minute: u32,
last_stop_key: &str,
) -> bool {
snapshot
.stop_key
.as_deref()
.zip(snapshot.stop_minute)
.is_some_and(|(stop_key, stop_minute)| {
stop_key == current_stop_key
&& current_minute >= stop_minute
&& last_stop_key != stop_key
})
}
fn persist_scheduler_start_trigger(
app_handle: &tauri::AppHandle,
settings_cache: &Arc<RwLock<Option<crate::ipc::PersistedSettings>>>,
key: &str,
scheduler: &crate::ipc::SchedulerSettings,
) {
let snapshot = scheduler_run_snapshot(key, scheduler);
if let Err(error) = crate::settings::update_settings_state(app_handle, |state| {
state.insert(
"schedulerTriggeredStartKey".to_string(),
serde_json::json!(key),
);
state.insert(
"schedulerActiveRunSnapshot".to_string(),
serde_json::json!({
"startKey": snapshot.start_key,
"stopKey": snapshot.stop_key,
"stopMinute": snapshot.stop_minute
}),
);
}) {
log::warn!("Failed to persist scheduler start trigger: {error}");
}
@@ -88,6 +140,7 @@ fn persist_scheduler_start_trigger(
if let Ok(mut settings) = settings_cache.write() {
if let Some(settings) = settings.as_mut() {
settings.scheduler_triggered_start_key = Some(key.to_string());
settings.scheduler_active_run_snapshot = Some(snapshot);
}
}
}
@@ -117,6 +170,7 @@ pub fn spawn_scheduler(
.clone()
.unwrap_or_default(),
settings.scheduler_last_stop_key.clone(),
settings.scheduler_active_run_snapshot.clone(),
)
})
});
@@ -125,6 +179,7 @@ pub fn spawn_scheduler(
scheduler_last_start_key,
persisted_triggered_start_key,
scheduler_last_stop_key,
scheduler_active_run_snapshot,
)) = settings
{
if !scheduler.enabled {
@@ -156,8 +211,11 @@ pub fn spawn_scheduler(
.get("start")
.is_none_or(|instant| instant.elapsed() >= Duration::from_secs(5))
{
if persisted_triggered_start_key != start_key
&& triggered_start_key != start_key
if (persisted_triggered_start_key != start_key
&& triggered_start_key != start_key)
|| scheduler_active_run_snapshot
.as_ref()
.is_none_or(|snapshot| snapshot.start_key != start_key)
{
// Record the dispatch intent before emitting so a
// crash between the native event and renderer ack
@@ -168,6 +226,7 @@ pub fn spawn_scheduler(
&app_handle,
&settings_cache,
&start_key,
&scheduler,
);
}
if app_handle.emit(
@@ -229,7 +288,25 @@ pub fn spawn_scheduler(
stop_key: &stop_key,
});
if (same_day_stop_due || overnight_stop_due)
let active_snapshot = scheduler_active_run_snapshot.as_ref().filter(|snapshot| {
snapshot.start_key == scheduler_last_start_key
|| snapshot.start_key == persisted_triggered_start_key
|| snapshot.start_key == triggered_start_key
});
let snapshot_stop_due = active_snapshot.is_some_and(|snapshot| {
snapshot_stop_is_due(
snapshot,
&stop_key,
current_minute,
&scheduler_last_stop_key,
)
});
let stop_due = if active_snapshot.is_some() {
snapshot_stop_due
} else {
same_day_stop_due || overnight_stop_due
};
if stop_due
&& last_emit
.get("stop")
.is_none_or(|instant| instant.elapsed() >= Duration::from_secs(5))
@@ -238,7 +315,9 @@ pub fn spawn_scheduler(
"schedule-trigger",
serde_json::json!({
"action": "stop",
"key": stop_key
"key": active_snapshot
.and_then(|snapshot| snapshot.stop_key.as_deref())
.unwrap_or(&stop_key)
}),
);
last_emit.insert("stop", std::time::Instant::now());
@@ -250,7 +329,28 @@ pub fn spawn_scheduler(
#[cfg(test)]
mod tests {
use super::{minute_of_day, overnight_stop_is_due, stop_is_due, OvernightStopCheck};
use super::{
minute_of_day, overnight_stop_is_due, scheduler_run_snapshot, snapshot_stop_is_due,
stop_is_due, OvernightStopCheck,
};
use crate::ipc::{PostQueueAction, SchedulerSettings};
fn scheduler_settings(
start_time: &str,
stop_enabled: bool,
stop_time: &str,
) -> SchedulerSettings {
SchedulerSettings {
enabled: true,
start_time: start_time.to_string(),
stop_time_enabled: stop_enabled,
stop_time: stop_time.to_string(),
everyday: false,
selected_days: vec![1],
selected_queue_ids: Vec::new(),
post_queue_action: PostQueueAction::None,
}
}
#[test]
fn parses_valid_scheduler_times() {
@@ -270,6 +370,46 @@ mod tests {
assert_eq!(minute_of_day("bad"), None);
}
#[test]
fn snapshots_overnight_stop_boundary_for_started_run() {
let snapshot = scheduler_run_snapshot(
"2026-06-22-start",
&scheduler_settings("22:00", true, "06:00"),
);
assert_eq!(snapshot.start_key, "2026-06-22-start");
assert_eq!(snapshot.stop_key.as_deref(), Some("2026-06-23-stop"));
assert_eq!(snapshot.stop_minute, Some(360));
// Later edits to weekdays or start/stop settings cannot change this
// already-started run's stop boundary.
assert!(snapshot_stop_is_due(&snapshot, "2026-06-23-stop", 420, ""));
assert!(!snapshot_stop_is_due(&snapshot, "2026-06-23-stop", 359, ""));
assert!(!snapshot_stop_is_due(
&snapshot,
"2026-06-23-stop",
420,
"2026-06-23-stop"
));
}
#[test]
fn snapshots_same_day_stop_and_disabled_stop() {
let same_day = scheduler_run_snapshot(
"2026-06-22-start",
&scheduler_settings("08:00", true, "18:30"),
);
assert_eq!(same_day.stop_key.as_deref(), Some("2026-06-22-stop"));
assert_eq!(same_day.stop_minute, Some(1110));
let no_stop = scheduler_run_snapshot(
"2026-06-22-start",
&scheduler_settings("08:00", false, "18:30"),
);
assert_eq!(no_stop.stop_key, None);
assert_eq!(no_stop.stop_minute, None);
assert!(!snapshot_stop_is_due(&no_stop, "2026-06-22-stop", 1200, ""));
}
#[test]
fn stop_requires_same_day_acknowledged_start() {
assert!(!stop_is_due(
+12 -1
View File
@@ -319,6 +319,7 @@ pub fn preserve_scheduler_runtime_keys(
"schedulerLastStartKey",
"schedulerTriggeredStartKey",
"schedulerLastStopKey",
"schedulerActiveRunSnapshot",
] {
if let Some(value) = existing_state.get(key) {
incoming_state.insert(key.to_string(), value.clone());
@@ -1051,6 +1052,7 @@ fn default_settings() -> PersistedSettings {
scheduler_last_start_key: String::new(),
scheduler_triggered_start_key: None,
scheduler_last_stop_key: String::new(),
scheduler_active_run_snapshot: None,
last_custom_speed_limit_ki_b: 1024,
last_custom_speed_limit_unit: "MB/s".to_string(),
per_server_connections: 16,
@@ -1118,7 +1120,12 @@ mod tests {
"state": {
"schedulerLastStartKey": "2026-06-22-start",
"schedulerTriggeredStartKey": "2026-06-22-start",
"schedulerLastStopKey": "2026-06-22-stop"
"schedulerLastStopKey": "2026-06-22-stop",
"schedulerActiveRunSnapshot": {
"startKey": "2026-06-23-start",
"stopKey": "2026-06-24-stop",
"stopMinute": 360
}
},
"version": 3
})
@@ -1142,6 +1149,10 @@ mod tests {
"2026-06-22-start"
);
assert_eq!(merged["state"]["schedulerLastStopKey"], "2026-06-22-stop");
assert_eq!(
merged["state"]["schedulerActiveRunSnapshot"]["stopKey"],
"2026-06-24-stop"
);
}
#[test]
+91
View File
@@ -1694,6 +1694,73 @@ async fn dispatcher_rotates_eligible_queues_without_starving_later_work() {
dispatcher.abort();
}
#[tokio::test]
async fn selected_start_hold_keeps_other_queues_eligible_and_preserves_priority() {
let (manager, _spawner) = make_manager(1);
let manager = Arc::new(manager);
let hold = manager
.begin_queue_dispatch_hold("selected")
.await
.expect("selected queue hold should start");
manager
.push(aria2_task_in_queue("older", "selected"))
.await
.unwrap();
manager
.push(aria2_task_in_queue("other-queue", "other"))
.await
.unwrap();
let dispatcher = {
let manager = Arc::clone(&manager);
tokio::spawn(async move { manager.run_dispatcher().await })
};
timeout(Duration::from_secs(1), async {
while manager.aria2_gid_for_download("other-queue").is_none() {
tokio::task::yield_now().await;
}
})
.await
.expect("a held queue must not block eligible work in another queue");
assert!(manager.aria2_gid_for_download("older").is_none());
// A becomes a newly selected start while older is still ahead in this
// queue. Keep the queue held through its atomic order update.
manager.release_permit("other-queue").await;
manager
.push(aria2_task_in_queue("selected", "selected"))
.await
.unwrap();
assert_eq!(
manager
.move_many_in_queue_to(&["selected".to_string()], "selected", 0)
.await,
vec!["selected", "older"]
);
assert!(manager
.end_queue_dispatch_hold("selected", &hold)
.await);
timeout(Duration::from_secs(1), async {
while manager.aria2_gid_for_download("selected").is_none() {
tokio::task::yield_now().await;
}
})
.await
.expect("the selected row should take the next available slot");
assert!(manager.aria2_gid_for_download("older").is_none());
manager.release_permit("selected").await;
timeout(Duration::from_secs(1), async {
while manager.aria2_gid_for_download("older").is_none() {
tokio::task::yield_now().await;
}
})
.await
.expect("older work should dispatch after the selected row");
dispatcher.abort();
}
#[tokio::test]
async fn queue_limit_increase_wakes_waiting_work_and_decrease_keeps_active_tasks() {
let (manager, spawner) = make_manager(3);
@@ -2883,6 +2950,30 @@ async fn multi_move_reorders_selected_items_as_one_atomic_block() {
);
}
#[tokio::test]
async fn multi_move_down_moves_sparse_selection_one_position_and_respects_boundary() {
use firelink_lib::ipc::QueueDirection;
let (mgr, _spawner) = make_manager(3);
for id in ["a", "b", "c", "d", "e"] {
mgr.push(sample_task(id)).await.unwrap();
}
let selected = vec!["b".to_string(), "d".to_string()];
assert_eq!(
mgr.move_many_in_queue(&selected, "main", QueueDirection::Down)
.await,
vec!["a", "c", "b", "d", "e"]
);
let bottom_selection = vec!["d".to_string(), "e".to_string()];
assert_eq!(
mgr.move_many_in_queue(&bottom_selection, "main", QueueDirection::Down)
.await,
vec!["a", "c", "b", "d", "e"]
);
}
#[tokio::test]
async fn direction_move_keeps_the_queue_order_for_an_unsorted_selection() {
use firelink_lib::ipc::QueueDirection;
+19
View File
@@ -1396,6 +1396,25 @@ function App() {
schedulerActiveDownloadIds
]);
useEffect(() => {
if (!coreReady || frontendExitPreparing || schedulerRunning) return;
if (schedulerActiveDownloadIds.length === 0) return;
const activeIds = new Set(downloads
.filter(download => isActiveDownloadStatus(download.status))
.map(download => download.id));
const remainingIds = schedulerActiveDownloadIds.filter(id => activeIds.has(id));
if (remainingIds.length === schedulerActiveDownloadIds.length) return;
void useSettingsStore.getState().persistSchedulerTracking(remainingIds, false).catch(error => {
console.error('Failed to clear inactive scheduler tracking:', error);
});
}, [
coreReady,
downloads,
frontendExitPreparing,
schedulerActiveDownloadIds,
schedulerRunning
]);
useEffect(() => {
const initNotifications = async () => {
if (!useSettingsStore.getState().showNotifications) return;
+28 -5
View File
@@ -237,14 +237,37 @@ export default function SchedulerView() {
const counts = await Promise.all(
targetQueueList.map(queueId => useDownloadStore.getState().pauseQueue(queueId))
);
const directPauseResults = await Promise.allSettled(
await Promise.allSettled(
trackedIdsOutsideQueues.map(id => useDownloadStore.getState().pauseDownload(id))
);
if (!isSchedulerControlCurrent(generation)) return;
const count = counts.reduce((total, queueCount) => total + queueCount, 0)
+ directPauseResults.filter(result => result.status === 'fulfilled').length;
useSettingsStore.getState().setSchedulerRunning(false);
useSettingsStore.getState().setSchedulerActiveDownloadIds([]);
const downloadsAfterPause = useDownloadStore.getState().downloads;
const directPausedCount = trackedIdsOutsideQueues.filter(id => {
const before = downloads.find(download => download.id === id);
const after = downloadsAfterPause.find(download => download.id === id);
return before && isActiveDownloadStatus(before.status) && after?.status === 'paused';
}).length;
const stillActiveTrackedIds = trackedDownloadIds.filter(id => {
const download = downloadsAfterPause.find(item => item.id === id);
return download && isActiveDownloadStatus(download.status);
});
const remainingTargetCount = downloadsAfterPause.filter(download => {
const queueId = download.queueId || MAIN_QUEUE_ID;
return (targetQueueSet.has(queueId) || trackedDownloadIds.includes(download.id))
&& isActiveDownloadStatus(download.status);
}).length;
await useSettingsStore.getState().persistSchedulerTracking(stillActiveTrackedIds, false);
const count = counts.reduce((total, queueCount) => total + queueCount, 0) + directPausedCount;
if (remainingTargetCount > 0) {
addToast({
message: remainingTargetCount === 1
? t($ => $.app.schedulerPauseOneFailed)
: t($ => $.app.schedulerPauseManyFailed, { count: remainingTargetCount }),
variant: 'error',
isActionable: true
});
return;
}
addToast({
message: count === 0
? t($ => $.scheduler.noActiveDownloads)
+5 -1
View File
@@ -84,7 +84,8 @@ type CommandMap = {
get_deno_engine_status: { args: undefined; result: EngineStatusItem };
reveal_in_file_manager: { args: { path: string }; result: void };
open_downloaded_file: { args: { path: string }; result: void };
pause_download: { args: { id: string }; result: void };
/** True when the pause won; false when a terminal/inactive lifecycle won. */
pause_download: { args: { id: string }; result: boolean };
resume_download: { args: { id: string; queueId: string }; result: boolean };
submit_download_removals: { args: { ids: string[]; deleteAssets: boolean }; result: void };
list_download_removals: { args: undefined; result: DownloadRemovalJob[] };
@@ -224,6 +225,9 @@ type CommandMap = {
enqueue_many: { args: { items: EnqueueItem[] }; result: import('./bindings/EnqueueResult').EnqueueResult[] };
move_in_queue: { args: { id: string; queueId: string; direction: 'up' | 'down' }; result: string[] };
move_many_in_queue: { args: { ids: string[]; queueId: string; direction: 'up' | 'down'; targetIndex?: number }; result: string[] };
begin_queue_dispatch_hold: { args: { queueId: string }; result: string };
renew_queue_dispatch_hold: { args: { queueId: string; token: string }; result: boolean };
end_queue_dispatch_hold: { args: { queueId: string; token: string }; result: boolean };
remove_from_queue: { args: { id: string }; result: boolean };
open_download_properties_window: { args: { id: string }; result: string };
get_properties_window_download_id: { args: undefined; result: string };
+163 -4
View File
@@ -1972,6 +1972,9 @@ describe('useDownloadStore', () => {
});
vi.mocked(ipc.invokeCommand).mockImplementation(async (command: string, args?: unknown) => {
if (command === 'begin_queue_dispatch_hold') return `hold-${(args as { queueId: string }).queueId}`;
if (command === 'renew_queue_dispatch_hold') return true;
if (command === 'end_queue_dispatch_hold') return true;
if (command === 'enqueue_download') {
const id = (args as { item: { id: string } }).item.id;
return { id, filename: id };
@@ -1987,6 +1990,12 @@ describe('useDownloadStore', () => {
.filter(([command]) => command === 'enqueue_download')
.map(([, args]) => (args as { item: { id: string } }).item.id);
expect(enqueueIds).toEqual(['selected-b', 'selected-a']);
const beginHoldCall = vi.mocked(ipc.invokeCommand).mock.calls
.findIndex(([command]) => command === 'begin_queue_dispatch_hold');
const enqueueCall = vi.mocked(ipc.invokeCommand).mock.calls
.findIndex(([command]) => command === 'enqueue_download');
expect(vi.mocked(ipc.invokeCommand).mock.invocationCallOrder[beginHoldCall])
.toBeLessThan(vi.mocked(ipc.invokeCommand).mock.invocationCallOrder[enqueueCall]);
expect(ipc.invokeCommand).toHaveBeenCalledWith('move_many_in_queue', {
ids: ['selected-b', 'selected-a'],
queueId: 'selection-queue',
@@ -2011,6 +2020,9 @@ describe('useDownloadStore', () => {
});
vi.mocked(ipc.invokeCommand).mockImplementation(async (command: string, args?: unknown) => {
if (command === 'begin_queue_dispatch_hold') return `hold-${(args as { queueId: string }).queueId}`;
if (command === 'renew_queue_dispatch_hold') return true;
if (command === 'end_queue_dispatch_hold') return true;
if (command === 'enqueue_download') {
const id = (args as { item: { id: string } }).item.id;
return { id, filename: id };
@@ -2062,6 +2074,9 @@ describe('useDownloadStore', () => {
});
vi.mocked(ipc.invokeCommand).mockImplementation(async (command: string, args?: unknown) => {
if (command === 'begin_queue_dispatch_hold') return `hold-${(args as { queueId: string }).queueId}`;
if (command === 'renew_queue_dispatch_hold') return true;
if (command === 'end_queue_dispatch_hold') return true;
if (command === 'enqueue_download') {
const item = (args as { item: { id: string; password: string | null } }).item;
return { id: item.id, filename: item.id };
@@ -2091,6 +2106,28 @@ describe('useDownloadStore', () => {
]);
});
it('releases the selected queue hold when re-enqueue fails', async () => {
useDownloadStore.setState({
downloads: [
{ id: 'selected-failure', url: 'http://failure', fileName: 'failure', destination: '/tmp', status: 'paused', category: 'Other', dateAdded: '', queueId: 'selected-failure-queue' },
] as any[],
});
vi.mocked(ipc.invokeCommand).mockImplementation(async (command: string) => {
if (command === 'begin_queue_dispatch_hold') return 'failure-hold';
if (command === 'renew_queue_dispatch_hold') return true;
if (command === 'end_queue_dispatch_hold') return true;
if (command === 'enqueue_download') throw new Error('queue persistence unavailable');
if (command === 'get_pending_order') return [];
return undefined;
});
await expect(useDownloadStore.getState().startSelected(['selected-failure'])).resolves.toBe(0);
expect(ipc.invokeCommand).toHaveBeenCalledWith('end_queue_dispatch_hold', {
queueId: 'selected-failure-queue',
token: 'failure-hold',
});
});
it('pauses queued items through the global pause action', async () => {
useDownloadStore.setState({
downloads: [
@@ -2098,8 +2135,9 @@ describe('useDownloadStore', () => {
{ id: 'queued-two', url: 'http://two', fileName: 'two', status: 'queued', category: 'Other', dateAdded: '', queueId: 'pause-queue' },
{ id: 'staged-one', url: 'http://staged', fileName: 'staged', status: 'staged', category: 'Other', dateAdded: '', queueId: 'pause-queue' },
] as any[],
backendRegisteredIds: new Set(['queued-one', 'queued-two', 'staged-one']),
});
vi.mocked(ipc.invokeCommand).mockResolvedValue(undefined as never);
vi.mocked(ipc.invokeCommand).mockResolvedValue(true as never);
await expect(useDownloadStore.getState().pauseAll()).resolves.toBe(3);
expect(
@@ -2108,6 +2146,30 @@ describe('useDownloadStore', () => {
expect(useDownloadStore.getState().downloads.every(item => item.status === 'paused')).toBe(true);
});
it('locally pauses an unregistered staged row when native owns no lifecycle', async () => {
useDownloadStore.setState({
downloads: [
{ id: 'unregistered-staged', status: 'staged', queueId: 'local-pause', speed: '1 MiB/s', eta: '2s' }
] as any[],
backendRegisteredIds: new Set(),
pendingOrder: [],
});
vi.mocked(ipc.invokeCommand).mockImplementation(async command => {
if (command === 'pause_download') return false as never;
return undefined as never;
});
await useDownloadStore.getState().pauseDownload('unregistered-staged');
expect(useDownloadStore.getState().downloads[0]).toMatchObject({
status: 'paused',
speed: '-',
eta: '-',
});
expect(useDownloadStore.getState().pendingOrder).not.toContain('unregistered-staged');
expect(ipc.invokeCommand).toHaveBeenCalledWith('pause_download', { id: 'unregistered-staged' });
});
it('moves a paused row behind the remaining runnable queue rows', async () => {
useDownloadStore.setState({
downloads: [
@@ -2116,9 +2178,10 @@ describe('useDownloadStore', () => {
{ id: 'pause-target', status: 'queued', queueId: 'ordered-pause', queuePosition: 2 },
{ id: 'queued-two', status: 'queued', queueId: 'ordered-pause', queuePosition: 3 }
] as any[],
backendRegisteredIds: new Set(['active', 'queued-one', 'pause-target', 'queued-two']),
pendingOrder: ['queued-one', 'pause-target', 'queued-two']
});
vi.mocked(ipc.invokeCommand).mockResolvedValue(undefined as never);
vi.mocked(ipc.invokeCommand).mockResolvedValue(true as never);
await useDownloadStore.getState().pauseDownload('pause-target');
@@ -2147,6 +2210,9 @@ describe('useDownloadStore', () => {
releaseEnqueue = resolve;
});
vi.mocked(ipc.invokeCommand).mockImplementation(async (command: string) => {
if (command === 'begin_queue_dispatch_hold') return 'hold-race-selected';
if (command === 'renew_queue_dispatch_hold') return true;
if (command === 'end_queue_dispatch_hold') return true;
if (command === 'enqueue_download') return enqueue;
if (command === 'get_pending_order') return [];
return undefined;
@@ -2177,6 +2243,10 @@ describe('useDownloadStore', () => {
expect(useDownloadStore.getState().downloads.map(item => item.status)).toEqual(['paused', 'paused']);
expect(vi.mocked(ipc.invokeCommand).mock.calls.filter(([command]) => command === 'pause_download'))
.toHaveLength(1);
expect(ipc.invokeCommand).toHaveBeenCalledWith('end_queue_dispatch_hold', {
queueId: 'race-selected',
token: 'hold-race-selected',
});
});
it('cleans an accepted backend enqueue when queue reconciliation fails', async () => {
@@ -4167,6 +4237,32 @@ describe('useDownloadStore', () => {
).toHaveLength(4);
});
it('does not turn a completed download into paused when completion wins a queue pause race', async () => {
useDownloadStore.setState({
downloads: [{
id: 'pause-completion-race',
url: 'https://example.com/file',
fileName: 'file.bin',
status: 'downloading',
category: 'Other',
dateAdded: '',
queueId: 'queue-race'
}] as any[]
});
vi.mocked(ipc.invokeCommand).mockImplementation(async command => {
if (command === 'pause_download') {
// Native completion reconciliation happened before pause_download
// returned its authoritative "pause did not win" result.
useDownloadStore.getState().updateDownload('pause-completion-race', { status: 'completed' });
return false as never;
}
return undefined as never;
});
await expect(useDownloadStore.getState().pauseQueue('queue-race')).resolves.toBe(0);
expect(useDownloadStore.getState().downloads[0].status).toBe('completed');
});
it('assigns selected unfinished downloads to a queue without moving completed items', async () => {
useDownloadStore.setState({
downloads: [
@@ -4265,7 +4361,7 @@ describe('useDownloadStore', () => {
{ id: 'two', status: 'queued', queueId: 'move-queue', queuePosition: 2 },
{ id: 'three', status: 'queued', queueId: 'move-queue', queuePosition: 3 }
] as any[],
backendRegisteredIds: new Set(['three']),
backendRegisteredIds: new Set(['one', 'two', 'three']),
pendingOrder: ['one', 'two', 'three']
});
vi.mocked(ipc.invokeCommand).mockImplementation(async (command: string) => {
@@ -4279,7 +4375,8 @@ describe('useDownloadStore', () => {
expect(vi.mocked(ipc.invokeCommand)).toHaveBeenCalledWith('move_many_in_queue', {
ids: ['three'],
queueId: 'move-queue',
direction: 'up'
direction: 'up',
targetIndex: 1,
});
expect(vi.mocked(ipc.invokeCommand)).not.toHaveBeenCalledWith('move_in_queue', expect.anything());
const positions = useDownloadStore.getState().downloads
@@ -4288,6 +4385,68 @@ describe('useDownloadStore', () => {
expect(new Set(positions).size).toBe(4);
});
it('moves a sparse selection down by one row as a block', async () => {
useDownloadStore.setState({
downloads: [
{ id: 'a', status: 'queued', queueId: 'sparse-down', queuePosition: 0 },
{ id: 'b', status: 'queued', queueId: 'sparse-down', queuePosition: 1 },
{ id: 'c', status: 'queued', queueId: 'sparse-down', queuePosition: 2 },
{ id: 'd', status: 'queued', queueId: 'sparse-down', queuePosition: 3 },
{ id: 'e', status: 'queued', queueId: 'sparse-down', queuePosition: 4 },
] as any[],
backendRegisteredIds: new Set(['a', 'b', 'c', 'd', 'e']),
pendingOrder: ['a', 'b', 'c', 'd', 'e'],
});
vi.mocked(ipc.invokeCommand).mockImplementation(async (command: string) => {
if (command === 'move_many_in_queue') return ['a', 'c', 'b', 'd', 'e'];
if (command === 'get_pending_order') return ['a', 'c', 'b', 'd', 'e'];
return undefined;
});
await useDownloadStore.getState().moveInQueue(['b', 'd'], 'down');
expect(useDownloadStore.getState().downloads
.filter(item => item.queueId === 'sparse-down')
.sort((left, right) => (left.queuePosition ?? 0) - (right.queuePosition ?? 0))
.map(item => item.id)).toEqual(['a', 'c', 'b', 'd', 'e']);
expect(ipc.invokeCommand).toHaveBeenCalledWith('move_many_in_queue', {
ids: ['b', 'd'],
queueId: 'sparse-down',
direction: 'up',
targetIndex: 2,
});
});
it('translates keyboard moves around staged rows before the atomic backend move', async () => {
useDownloadStore.setState({
downloads: [
{ id: 'a', status: 'queued', queueId: 'keyboard-staged', queuePosition: 0 },
{ id: 'staged', status: 'staged', queueId: 'keyboard-staged', queuePosition: 1 },
{ id: 'b', status: 'queued', queueId: 'keyboard-staged', queuePosition: 2 },
] as any[],
backendRegisteredIds: new Set(['a', 'b']),
pendingOrder: ['a', 'b'],
});
vi.mocked(ipc.invokeCommand).mockImplementation(async (command: string) => {
if (command === 'move_many_in_queue') return ['b', 'a'];
if (command === 'get_pending_order') return ['b', 'a'];
return undefined;
});
await useDownloadStore.getState().moveInQueue('b', 'up');
expect(ipc.invokeCommand).toHaveBeenCalledWith('move_many_in_queue', {
ids: ['b'],
queueId: 'keyboard-staged',
direction: 'up',
targetIndex: 1,
});
expect(useDownloadStore.getState().downloads
.filter(item => item.queueId === 'keyboard-staged')
.sort((left, right) => (left.queuePosition ?? 0) - (right.queuePosition ?? 0))
.map(item => item.id)).toEqual(['a', 'b', 'staged']);
});
it('rolls back and rejects a failed keyboard or header queue move', async () => {
useDownloadStore.setState({
downloads: [
+152 -65
View File
@@ -274,6 +274,8 @@ const queuePositionComparator = (left: DownloadItem, right: DownloadItem): numbe
comparableQueuePosition(left) - comparableQueuePosition(right) ||
left.id.localeCompare(right.id);
const QUEUE_START_BATCH_HOLD_REFRESH_MS = 20_000;
const queueItemsForReordering = (downloads: DownloadItem[], queueId: string): DownloadItem[] =>
downloads
.filter(download =>
@@ -1708,14 +1710,13 @@ export const useDownloadStore = create<DownloadState>((set, get) => {
].map(item => [item.id, item.queuePosition]));
const unselectedItems = queueItems.filter(item => !ids.includes(item.id));
const selectedIndices = selectedItems.map(item => queueItems.findIndex(d => d.id === item.id));
const firstSelectedIndex = Math.min(...selectedIndices);
let insertIndex = 0;
if (direction === 'up') {
const firstSelectedIndex = Math.min(...selectedIndices);
insertIndex = Math.max(0, firstSelectedIndex - 1);
} else {
const lastSelectedIndex = Math.max(...selectedIndices);
insertIndex = Math.min(unselectedItems.length, lastSelectedIndex - selectedItems.length + 2);
insertIndex = Math.min(unselectedItems.length, firstSelectedIndex + 1);
}
const reordered = [
@@ -1730,11 +1731,25 @@ export const useDownloadStore = create<DownloadState>((set, get) => {
.map(item => item.id);
if (registeredIdsToMove.length === 0) return;
// The local queue includes staged rows, but the native pending queue
// contains registered rows only. Translate the desired local order to
// its registered-only insertion point so keyboard and header moves
// cannot leave the UI and dispatcher with different orders.
const registeredItems = queueItems.filter(item => get().backendRegisteredIds.has(item.id));
const registeredSelectedIds = new Set(registeredIdsToMove);
const registeredDesiredOrder = reordered.filter(item => get().backendRegisteredIds.has(item.id));
const backendTargetIndex = targetIndexForDesiredOrder(
registeredItems,
registeredSelectedIds,
registeredDesiredOrder
);
try {
const order = await invoke('move_many_in_queue', {
ids: registeredIdsToMove,
queueId,
direction
direction: 'up',
targetIndex: backendTargetIndex
}) as string[];
if (Array.isArray(order)) {
const globalOrder = await invoke('get_pending_order', { queueId: null })
@@ -2489,9 +2504,30 @@ export const useDownloadStore = create<DownloadState>((set, get) => {
await pendingDispatch;
}
await invoke('pause_download', { id });
const pauseApplied = await invoke('pause_download', { id });
if (!isCurrentDownloadLifecycle(id, generation)) return;
// Completion can win after the UI snapshots an active row but before
// Aria2 processes forcePause. The native command reconciles that
// terminal state and reports false so a stale frontend snapshot cannot
// overwrite Completed with Paused in memory or durable storage. A row
// that is still queued/staged but has no native ownership is different:
// it never reached the backend queue, so preserve the user's pause
// locally instead of leaving an unstartable-looking row staged.
if (pauseApplied === false) {
const current = get().downloads.find(download => download.id === id);
const hasNativeOwnership = get().backendRegisteredIds.has(id) ||
get().pendingOrder.includes(id);
if (
current &&
(current.status === 'queued' || current.status === 'staged') &&
!hasNativeOwnership
) {
get().updateDownload(id, { status: 'paused', speed: '-', eta: '-' });
await commitDownloadState();
}
return;
}
const current = get().downloads.find(download => download.id === id);
if (current && current.status !== 'completed' && current.status !== 'failed') {
get().updateDownload(id, { status: 'paused', speed: '-', eta: '-' });
@@ -2584,71 +2620,118 @@ export const useDownloadStore = create<DownloadState>((set, get) => {
] as const)
);
// Make the selection's order explicit before any capacity becomes
// available. This keeps the frontend projection and backend pending
// queue aligned even when the selected rows were not adjacent.
for (const [queueId, queueIds] of selectedByQueue) {
const generation = selectedQueueGenerations.get(queueId);
if (generation === undefined || !isCurrentQueueControlGeneration(queueId, generation)) {
continue;
}
await get().moveManyInQueueToPosition(queueIds, queueId, 0, null, queueIds);
}
const dispatchHolds = new Map<string, string>();
const expiredDispatchHolds = new Set<string>();
let dispatchHoldRefresh: ReturnType<typeof setInterval> | undefined;
let startedCount = 0;
const startedByQueue = new Map<string, string[]>();
for (const id of orderedIds) {
const current = get().downloads.find(download => download.id === id);
if (!current || !canStartDownload(current.status)) continue;
const queueId = current.queueId || MAIN_QUEUE_ID;
const generation = selectedQueueGenerations.get(queueId);
if (generation === undefined || !isCurrentQueueControlGeneration(queueId, generation)) {
continue;
const ensureDispatchHold = async (queueId: string): Promise<void> => {
const token = dispatchHolds.get(queueId);
if (!token || expiredDispatchHolds.has(queueId)) {
throw new Error('Selected queue start lost its dispatch-order lease');
}
const resumed = await resumeDownloadInternal(id, {
preserveQueuePosition: true,
forceRequeue: true
});
if (resumed && !isCurrentQueueControlGeneration(queueId, generation)) {
// A queue pause can win while this item's requeue is in flight. The
// pause action is allowed to preempt this bulk-start operation so
// this item cannot remain running after the user's pause request.
await get().pauseDownload(id);
continue;
const renewed = await invoke('renew_queue_dispatch_hold', { queueId, token }) as boolean;
if (!renewed) {
expiredDispatchHolds.add(queueId);
throw new Error('Selected queue start lost its dispatch-order lease');
}
if (resumed) {
startedCount += 1;
const startedIds = startedByQueue.get(queueId) || [];
startedIds.push(id);
startedByQueue.set(queueId, startedIds);
};
// A requeued paused item is appended by the backend. Move the
// started prefix immediately so a newly available slot cannot let
// older pending work overtake the remaining selection while it is
// still being requeued.
if (isCurrentQueueControlGeneration(queueId, generation)) {
await get().moveManyInQueueToPosition(
startedIds,
queueId,
0,
null,
startedIds
);
try {
// Hold each affected queue before changing or re-enqueueing any
// selected row. The backend lease is queue-scoped, so unrelated
// queues remain eligible while this selection is assembled.
for (const queueId of selectedByQueue.keys()) {
const token = await invoke('begin_queue_dispatch_hold', { queueId }) as string;
if (!token) throw new Error('Could not reserve selected queue order');
dispatchHolds.set(queueId, token);
}
dispatchHoldRefresh = setInterval(() => {
for (const [queueId, token] of dispatchHolds) {
void invoke('renew_queue_dispatch_hold', { queueId, token })
.then(active => {
if (!active) expiredDispatchHolds.add(queueId);
})
.catch(error => {
console.warn('Could not refresh selected queue dispatch lease:', error);
});
}
}, QUEUE_START_BATCH_HOLD_REFRESH_MS);
// Make the selection's order explicit before any capacity becomes
// available. This keeps the frontend projection and backend pending
// queue aligned even when the selected rows were not adjacent.
for (const [queueId, queueIds] of selectedByQueue) {
const generation = selectedQueueGenerations.get(queueId);
if (generation === undefined || !isCurrentQueueControlGeneration(queueId, generation)) {
continue;
}
await get().moveManyInQueueToPosition(queueIds, queueId, 0, null, queueIds);
}
let startedCount = 0;
const startedByQueue = new Map<string, string[]>();
for (const id of orderedIds) {
const current = get().downloads.find(download => download.id === id);
if (!current || !canStartDownload(current.status)) continue;
const queueId = current.queueId || MAIN_QUEUE_ID;
const generation = selectedQueueGenerations.get(queueId);
if (generation === undefined || !isCurrentQueueControlGeneration(queueId, generation)) {
continue;
}
await ensureDispatchHold(queueId);
const resumed = await resumeDownloadInternal(id, {
preserveQueuePosition: true,
forceRequeue: true
});
if (resumed && !isCurrentQueueControlGeneration(queueId, generation)) {
// A queue pause can win while this item's requeue is in flight. The
// pause action is allowed to preempt this bulk-start operation so
// this item cannot remain running after the user's pause request.
await get().pauseDownload(id);
continue;
}
if (resumed) {
startedCount += 1;
const startedIds = startedByQueue.get(queueId) || [];
startedIds.push(id);
startedByQueue.set(queueId, startedIds);
// The queue hold prevents a newly enqueued item or older pending
// work from consuming capacity before this block is fully ordered.
if (isCurrentQueueControlGeneration(queueId, generation)) {
await get().moveManyInQueueToPosition(
startedIds,
queueId,
0,
null,
startedIds
);
}
}
}
}
// Newly dispatched rows are now visible to the backend pending list.
// Reapply the same ordered block so existing pending work cannot remain
// ahead of a user-selected start request.
for (const [queueId, queueIds] of selectedByQueue) {
const generation = selectedQueueGenerations.get(queueId);
if (generation === undefined || !isCurrentQueueControlGeneration(queueId, generation)) {
continue;
// Newly dispatched rows are now visible to the backend pending list.
// Reapply the same ordered block so existing pending work cannot remain
// ahead of a user-selected start request.
for (const [queueId, queueIds] of selectedByQueue) {
const generation = selectedQueueGenerations.get(queueId);
if (generation === undefined || !isCurrentQueueControlGeneration(queueId, generation)) {
continue;
}
await get().moveManyInQueueToPosition(queueIds, queueId, 0, null, queueIds);
}
await get().moveManyInQueueToPosition(queueIds, queueId, 0, null, queueIds);
return startedCount;
} finally {
if (dispatchHoldRefresh !== undefined) clearInterval(dispatchHoldRefresh);
await Promise.all(Array.from(dispatchHolds, async ([queueId, token]) => {
try {
await invoke('end_queue_dispatch_hold', { queueId, token });
} catch (error) {
console.error('Failed to release selected queue dispatch lease:', error);
}
}));
}
return startedCount;
});
},
startQueue: (queueId) => {
@@ -2802,8 +2885,10 @@ export const useDownloadStore = create<DownloadState>((set, get) => {
if (activeIds.length === 0) return 0;
const results = await Promise.allSettled(activeIds.map(id => get().pauseDownload(id)));
const pausedCount = results.filter(result => result.status === 'fulfilled').length;
const failedCount = activeIds.length - pausedCount;
const pausedCount = activeIds.filter(id =>
get().downloads.some(download => download.id === id && download.status === 'paused')
).length;
const failedCount = results.filter(result => result.status === 'rejected').length;
if (failedCount > 0) {
console.error(`Failed to pause ${failedCount} downloads in queue ${queueId}`);
}
@@ -2834,8 +2919,10 @@ export const useDownloadStore = create<DownloadState>((set, get) => {
.map(item => item.id);
if (activeIds.length === 0) return 0;
const results = await Promise.allSettled(activeIds.map(id => get().pauseDownload(id)));
const pausedCount = results.filter(result => result.status === 'fulfilled').length;
await Promise.allSettled(activeIds.map(id => get().pauseDownload(id)));
const pausedCount = activeIds.filter(id =>
get().downloads.some(download => download.id === id && download.status === 'paused')
).length;
syncSystemIntegrations();
return pausedCount;
},
+21
View File
@@ -1106,4 +1106,25 @@ describe('scheduled trigger durability', () => {
{ schedulerRunning: true, schedulerActiveDownloadIds: ['download-1'] }
]);
});
it('keeps pause-failed downloads tracked without treating a manual pause as a running schedule', async () => {
let savedState: { schedulerRunning: boolean; schedulerActiveDownloadIds: string[] } | undefined;
vi.mocked(ipc.invokeCommand).mockImplementation(async (command: string, args?: unknown) => {
if (command === 'db_save_settings') {
savedState = JSON.parse((args as { data: string }).data).state;
}
return undefined as never;
});
await useSettingsStore.getState().persistSchedulerTracking(['still-active'], false);
expect(useSettingsStore.getState()).toMatchObject({
schedulerRunning: false,
schedulerActiveDownloadIds: ['still-active']
});
expect(savedState).toMatchObject({
schedulerRunning: false,
schedulerActiveDownloadIds: ['still-active']
});
});
});
+3 -3
View File
@@ -414,7 +414,7 @@ export interface SettingsState {
setScheduler: (settings: SchedulerSettings) => void;
setSchedulerRunning: (running: boolean) => void;
setSchedulerActiveDownloadIds: (ids: string[]) => void;
persistSchedulerTracking: (ids: string[]) => Promise<void>;
persistSchedulerTracking: (ids: string[], running?: boolean) => Promise<void>;
setSchedulerLastStartKey: (key: string) => void;
setSchedulerLastStopKey: (key: string) => void;
setLastCustomSpeedLimitKiB: (limit: number) => void;
@@ -699,13 +699,13 @@ export const useSettingsStore = create<SettingsState>()(
setScheduler: (scheduler) => set({ scheduler }),
setSchedulerRunning: (schedulerRunning) => set({ schedulerRunning }),
setSchedulerActiveDownloadIds: (schedulerActiveDownloadIds) => set({ schedulerActiveDownloadIds }),
persistSchedulerTracking: (ids) => {
persistSchedulerTracking: (ids, running) => {
const schedulerActiveDownloadIds = [...new Set(
sanitizeSchedulerActiveDownloadIds(ids, [])
)];
set({
schedulerActiveDownloadIds,
schedulerRunning: schedulerActiveDownloadIds.length > 0
schedulerRunning: running ?? schedulerActiveDownloadIds.length > 0
});
// `waitForSettingsPersistence` intentionally observes only the queue's
// rejection-swallowing tail. Scheduler ACKs require confirmation from
+1 -1
View File
@@ -239,7 +239,7 @@ describe('credential-bearing extension header names', () => {
describe('allocation phase visibility', () => {
it('does not override paused or completed statuses', () => {
expect(isAllocationPhaseVisible(true, 'ready')).toBe(true);
expect(isAllocationPhaseVisible(true, 'failed')).toBe(false);
expect(isAllocationPhaseVisible(true, 'failed')).toBe(true);
expect(isAllocationPhaseVisible(true, 'verifying')).toBe(false);
expect(isAllocationPhaseVisible(true, 'seeding')).toBe(false);
expect(isAllocationPhaseVisible(true, 'waitingToSeed')).toBe(false);
+1 -1
View File
@@ -73,7 +73,7 @@ export const isTransferActiveStatus = (status: DownloadStatus): boolean =>
export const isAllocationPhaseVisible = (
allocationPending: boolean,
status: DownloadStatus,
): boolean => allocationPending && ['ready', 'staged', 'queued', 'downloading', 'retrying'].includes(status);
): boolean => allocationPending && ['ready', 'staged', 'queued', 'downloading', 'retrying', 'failed'].includes(status);
/**
* Allocation is a transient admission phase. Normal downloads retain the