mirror of
https://github.com/nimbold/Firelink.git
synced 2026-08-31 13:08:17 +00:00
fix(download): honor retry and queue limits
Treat max_tries as the configured retry count instead of total attempts across native, media, aria2, and fallback paths. Route queue slot acquisition through retirement-aware permit handling so shrinking concurrency does not dispatch new work while active transfers still meet the smaller limit. Update retry-budget and queue-manager tests to cover the corrected attempt counts and paused aria2 resume behavior.
This commit is contained in:
@@ -469,10 +469,10 @@ async fn download_file(
|
|||||||
// until this future resolves). `download_attempt` re-issues a Range header
|
// until this future resolves). `download_attempt` re-issues a Range header
|
||||||
// from the existing partial file on every retry, so no bytes are discarded.
|
// from the existing partial file on every retry, so no bytes are discarded.
|
||||||
//
|
//
|
||||||
// The legacy `max_tries` payload field is still honored as a cap on
|
// `max_tries` is the user-facing retry count. Attempts include the first
|
||||||
// attempts, but transient backoff is additionally bounded by
|
// try plus those configured retries.
|
||||||
// `retry::MAX_RETRIES` so a single URL cannot spin forever.
|
let max_retries = payload.max_tries as usize;
|
||||||
let max_attempts = payload.max_tries.max(1) as usize;
|
let max_attempts = max_retries + 1;
|
||||||
'url: for url in &payload.urls {
|
'url: for url in &payload.urls {
|
||||||
let mut strike = 0_usize;
|
let mut strike = 0_usize;
|
||||||
let mut attempts = 0_usize;
|
let mut attempts = 0_usize;
|
||||||
@@ -513,7 +513,7 @@ async fn download_file(
|
|||||||
}
|
}
|
||||||
|
|
||||||
let transient = crate::retry::is_transient_network_error(&error);
|
let transient = crate::retry::is_transient_network_error(&error);
|
||||||
let strikes_left = strike < crate::retry::MAX_RETRIES;
|
let strikes_left = strike < max_retries;
|
||||||
|
|
||||||
if transient && strikes_left {
|
if transient && strikes_left {
|
||||||
// Transient: announce `Retrying`, back off, then retry.
|
// Transient: announce `Retrying`, back off, then retry.
|
||||||
|
|||||||
+5
-10
@@ -2472,9 +2472,7 @@ pub(crate) async fn start_media_download_internal(
|
|||||||
let config_path = config_file.into_temp_path();
|
let config_path = config_file.into_temp_path();
|
||||||
|
|
||||||
use crate::ipc::DownloadStateEvent;
|
use crate::ipc::DownloadStateEvent;
|
||||||
use crate::retry::{
|
use crate::retry::{backoff_and_emit_cancel, is_transient_network_error, BackoffOutcome};
|
||||||
backoff_and_emit_cancel, is_transient_network_error, BackoffOutcome, MAX_RETRIES,
|
|
||||||
};
|
|
||||||
|
|
||||||
const STDERR_TAIL: usize = 2048;
|
const STDERR_TAIL: usize = 2048;
|
||||||
|
|
||||||
@@ -2504,10 +2502,11 @@ pub(crate) async fn start_media_download_internal(
|
|||||||
// user-installed tools, or platform-specific executable aliases.
|
// user-installed tools, or platform-specific executable aliases.
|
||||||
let trusted_path = crate::platform::trusted_system_path()?;
|
let trusted_path = crate::platform::trusted_system_path()?;
|
||||||
|
|
||||||
|
let max_retries = max_tries.unwrap_or(0).max(0) as usize;
|
||||||
let mut strike = 0_usize;
|
let mut strike = 0_usize;
|
||||||
let mut processing_started = false;
|
let mut processing_started = false;
|
||||||
|
|
||||||
while strike <= MAX_RETRIES {
|
while strike <= max_retries {
|
||||||
let ytdlp_path = resolve_bundled_binary_path(&app_handle, "yt-dlp")?;
|
let ytdlp_path = resolve_bundled_binary_path(&app_handle, "yt-dlp")?;
|
||||||
let mut cmd = app_handle
|
let mut cmd = app_handle
|
||||||
.shell()
|
.shell()
|
||||||
@@ -2520,7 +2519,7 @@ pub(crate) async fn start_media_download_internal(
|
|||||||
.arg("--socket-timeout")
|
.arg("--socket-timeout")
|
||||||
.arg("20")
|
.arg("20")
|
||||||
.arg("--retries")
|
.arg("--retries")
|
||||||
.arg("3")
|
.arg(max_retries.to_string())
|
||||||
.arg("--extractor-retries")
|
.arg("--extractor-retries")
|
||||||
.arg("3")
|
.arg("3")
|
||||||
.arg("--ffmpeg-location")
|
.arg("--ffmpeg-location")
|
||||||
@@ -2569,10 +2568,6 @@ pub(crate) async fn start_media_download_internal(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if let Some(tries) = max_tries {
|
|
||||||
cmd = cmd.arg("--retries").arg(tries.to_string());
|
|
||||||
}
|
|
||||||
|
|
||||||
if let Some(loc) = config_location.as_ref() {
|
if let Some(loc) = config_location.as_ref() {
|
||||||
cmd = cmd.arg("--config-location").arg(loc);
|
cmd = cmd.arg("--config-location").arg(loc);
|
||||||
}
|
}
|
||||||
@@ -2781,7 +2776,7 @@ pub(crate) async fn start_media_download_internal(
|
|||||||
};
|
};
|
||||||
|
|
||||||
let transient = is_transient_network_error(&failure_reason);
|
let transient = is_transient_network_error(&failure_reason);
|
||||||
let strikes_left = strike < MAX_RETRIES;
|
let strikes_left = strike < max_retries;
|
||||||
if !(transient && strikes_left) {
|
if !(transient && strikes_left) {
|
||||||
return Err(failure_reason);
|
return Err(failure_reason);
|
||||||
}
|
}
|
||||||
|
|||||||
+55
-44
@@ -1,5 +1,5 @@
|
|||||||
use crate::ipc::{DownloadStateEvent, DownloadStatus, QueueDirection};
|
use crate::ipc::{DownloadStateEvent, DownloadStatus, QueueDirection};
|
||||||
use crate::retry::{backoff_and_emit, is_transient_network_error, BackoffOutcome, MAX_RETRIES};
|
use crate::retry::{backoff_and_emit, is_transient_network_error, BackoffOutcome};
|
||||||
use log;
|
use log;
|
||||||
use serde::Deserialize;
|
use serde::Deserialize;
|
||||||
use serde_json;
|
use serde_json;
|
||||||
@@ -194,6 +194,33 @@ impl<R: tauri::Runtime> QueueManager<R> {
|
|||||||
self.semaphore.clone().acquire_owned().await.ok()
|
self.semaphore.clone().acquire_owned().await.ok()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn acquire_permit_after_retirement(&self) -> Option<OwnedSemaphorePermit> {
|
||||||
|
loop {
|
||||||
|
let permit = self.acquire_permit().await?;
|
||||||
|
if self.retire_slot_if_needed() {
|
||||||
|
permit.forget();
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
return Some(permit);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn retire_slot_if_needed(&self) -> bool {
|
||||||
|
let mut debt = self.slots_to_retire.load(Ordering::Relaxed);
|
||||||
|
while debt > 0 {
|
||||||
|
match self.slots_to_retire.compare_exchange_weak(
|
||||||
|
debt,
|
||||||
|
debt - 1,
|
||||||
|
Ordering::Relaxed,
|
||||||
|
Ordering::Relaxed,
|
||||||
|
) {
|
||||||
|
Ok(_) => return true,
|
||||||
|
Err(actual) => debt = actual,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
false
|
||||||
|
}
|
||||||
|
|
||||||
/// Park an already-acquired permit under `id`.
|
/// Park an already-acquired permit under `id`.
|
||||||
pub async fn park_permit(&self, id: &str, permit: OwnedSemaphorePermit) {
|
pub async fn park_permit(&self, id: &str, permit: OwnedSemaphorePermit) {
|
||||||
self.active_permits
|
self.active_permits
|
||||||
@@ -214,7 +241,7 @@ impl<R: tauri::Runtime> QueueManager<R> {
|
|||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
|
|
||||||
let permit = match self.acquire_permit().await {
|
let permit = match self.acquire_permit_after_retirement().await {
|
||||||
Some(p) => p,
|
Some(p) => p,
|
||||||
None => return false,
|
None => return false,
|
||||||
};
|
};
|
||||||
@@ -325,34 +352,10 @@ impl<R: tauri::Runtime> QueueManager<R> {
|
|||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
// (2) Acquire a slot.
|
// (2) Acquire a slot.
|
||||||
let permit_opt = self.semaphore.clone().acquire_owned().await.ok();
|
let permit = match self.acquire_permit_after_retirement().await {
|
||||||
let permit = match permit_opt {
|
|
||||||
Some(p) => p,
|
Some(p) => p,
|
||||||
None => break, // Semaphore closed, exit dispatcher
|
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);
|
|
||||||
while debt > 0 {
|
|
||||||
match self.slots_to_retire.compare_exchange_weak(
|
|
||||||
debt,
|
|
||||||
debt - 1,
|
|
||||||
Ordering::Relaxed,
|
|
||||||
Ordering::Relaxed,
|
|
||||||
) {
|
|
||||||
Ok(_) => {
|
|
||||||
retired = true;
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
Err(actual) => {
|
|
||||||
debt = actual;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if retired {
|
|
||||||
drop(permit);
|
|
||||||
continue;
|
|
||||||
}
|
|
||||||
// (4) Re-pop under lock — guards against racing removals between
|
// (4) Re-pop under lock — guards against racing removals between
|
||||||
// waking from Notify and acquiring the permit.
|
// waking from Notify and acquiring the permit.
|
||||||
let task = match self.pending.lock().await.pop_front() {
|
let task = match self.pending.lock().await.pop_front() {
|
||||||
@@ -612,20 +615,6 @@ impl<R: tauri::Runtime> QueueManager<R> {
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
let strike = {
|
|
||||||
let mut strikes = self.aria2_retry_strikes.lock().await;
|
|
||||||
let entry = strikes.entry(id.clone()).or_insert(0);
|
|
||||||
*entry
|
|
||||||
};
|
|
||||||
|
|
||||||
let transient = is_retryable_aria2_error(&error);
|
|
||||||
let strikes_left = strike < MAX_RETRIES;
|
|
||||||
if !(transient && strikes_left) {
|
|
||||||
self.apply_completion(&id, PendingOutcome::Error(error))
|
|
||||||
.await;
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
let payload = self.aria2_payloads.lock().await.get(&id).cloned();
|
let payload = self.aria2_payloads.lock().await.get(&id).cloned();
|
||||||
if payload.is_none() {
|
if payload.is_none() {
|
||||||
self.apply_completion(&id, PendingOutcome::Error(error))
|
self.apply_completion(&id, PendingOutcome::Error(error))
|
||||||
@@ -634,6 +623,20 @@ impl<R: tauri::Runtime> QueueManager<R> {
|
|||||||
}
|
}
|
||||||
let mut payload = payload.unwrap();
|
let mut payload = payload.unwrap();
|
||||||
|
|
||||||
|
let strike = {
|
||||||
|
let mut strikes = self.aria2_retry_strikes.lock().await;
|
||||||
|
let entry = strikes.entry(id.clone()).or_insert(0);
|
||||||
|
*entry
|
||||||
|
};
|
||||||
|
|
||||||
|
let transient = is_retryable_aria2_error(&error);
|
||||||
|
let strikes_left = strike < automatic_retry_limit(payload.max_tries);
|
||||||
|
if !(transient && strikes_left) {
|
||||||
|
self.apply_completion(&id, PendingOutcome::Error(error))
|
||||||
|
.await;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
if is_aria2_range_mode_error(&error) {
|
if is_aria2_range_mode_error(&error) {
|
||||||
log::warn!(
|
log::warn!(
|
||||||
"aria2 range mode [{}]: server rejected bounded chunk ranges; restarting with a single connection",
|
"aria2 range mode [{}]: server rejected bounded chunk ranges; restarting with a single connection",
|
||||||
@@ -850,6 +853,14 @@ impl<R: tauri::Runtime> QueueManager<R> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn automatic_retry_limit(max_tries: Option<i32>) -> usize {
|
||||||
|
max_tries.unwrap_or(0).max(0) as usize
|
||||||
|
}
|
||||||
|
|
||||||
|
fn aria2_attempt_limit(max_tries: Option<i32>) -> u32 {
|
||||||
|
(automatic_retry_limit(max_tries) + 1) as u32
|
||||||
|
}
|
||||||
|
|
||||||
fn is_retryable_aria2_error(error: &str) -> bool {
|
fn is_retryable_aria2_error(error: &str) -> bool {
|
||||||
is_transient_network_error(error) || is_aria2_range_mode_error(error)
|
is_transient_network_error(error) || is_aria2_range_mode_error(error)
|
||||||
}
|
}
|
||||||
@@ -1080,7 +1091,7 @@ impl SidecarSpawner for ProductionSpawner {
|
|||||||
"max-connection-per-server".to_string(),
|
"max-connection-per-server".to_string(),
|
||||||
serde_json::json!(conn.to_string()),
|
serde_json::json!(conn.to_string()),
|
||||||
);
|
);
|
||||||
let mt = payload.max_tries.unwrap_or(1).max(1) as u32;
|
let mt = aria2_attempt_limit(payload.max_tries);
|
||||||
options.insert("max-tries".to_string(), serde_json::json!(mt.to_string()));
|
options.insert("max-tries".to_string(), serde_json::json!(mt.to_string()));
|
||||||
options.insert("retry-wait".to_string(), serde_json::json!("2"));
|
options.insert("retry-wait".to_string(), serde_json::json!("2"));
|
||||||
options.insert("continue".to_string(), serde_json::json!("true"));
|
options.insert("continue".to_string(), serde_json::json!("true"));
|
||||||
@@ -1149,7 +1160,7 @@ impl SidecarSpawner for ProductionSpawner {
|
|||||||
// aria2 unavailable — fall back to native coordinator.
|
// aria2 unavailable — fall back to native coordinator.
|
||||||
log::warn!("aria2 addUri failed, falling back to native: {}", e);
|
log::warn!("aria2 addUri failed, falling back to native: {}", e);
|
||||||
let download_id = uuid::Uuid::parse_str(id).map_err(|e| e.to_string())?;
|
let download_id = uuid::Uuid::parse_str(id).map_err(|e| e.to_string())?;
|
||||||
let mt = payload.max_tries.unwrap_or(1).max(1) as u32;
|
let mt = automatic_retry_limit(payload.max_tries) as u32;
|
||||||
let safe_filename =
|
let safe_filename =
|
||||||
crate::download_ownership::canonical_download_filename(&payload.filename);
|
crate::download_ownership::canonical_download_filename(&payload.filename);
|
||||||
state
|
state
|
||||||
@@ -1241,7 +1252,7 @@ impl SidecarSpawner for ProductionSpawner {
|
|||||||
async fn run_native(&self, id: &str, payload: &SpawnPayload) -> Result<(), String> {
|
async fn run_native(&self, id: &str, payload: &SpawnPayload) -> Result<(), String> {
|
||||||
let state = self.app_handle.state::<crate::AppState>();
|
let state = self.app_handle.state::<crate::AppState>();
|
||||||
let download_id = uuid::Uuid::parse_str(id).map_err(|e| e.to_string())?;
|
let download_id = uuid::Uuid::parse_str(id).map_err(|e| e.to_string())?;
|
||||||
let mt = payload.max_tries.unwrap_or(1).max(1) as u32;
|
let mt = automatic_retry_limit(payload.max_tries) as u32;
|
||||||
let resolved_dest = crate::resolve_path(&payload.destination, &self.app_handle);
|
let resolved_dest = crate::resolve_path(&payload.destination, &self.app_handle);
|
||||||
let safe_filename =
|
let safe_filename =
|
||||||
crate::download_ownership::canonical_download_filename(&payload.filename);
|
crate::download_ownership::canonical_download_filename(&payload.filename);
|
||||||
|
|||||||
@@ -427,7 +427,7 @@ async fn reports_terminal_http_errors_after_retry_budget() {
|
|||||||
};
|
};
|
||||||
|
|
||||||
assert!(error.contains("500 Internal Server Error"));
|
assert!(error.contains("500 Internal Server Error"));
|
||||||
assert_eq!(server.state.requests.load(Ordering::SeqCst), 2);
|
assert_eq!(server.state.requests.load(Ordering::SeqCst), 3);
|
||||||
assert!(!output_path.exists());
|
assert!(!output_path.exists());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -231,22 +231,62 @@ async fn shrink_converges_to_target_without_killing_active() {
|
|||||||
// Shrink to 2 while 4 are "active" (permits parked).
|
// Shrink to 2 while 4 are "active" (permits parked).
|
||||||
mgr_arc.set_capacity(2);
|
mgr_arc.set_capacity(2);
|
||||||
|
|
||||||
// Release active permits. Debt is 2; two releases retire both.
|
// Release active permits. Debt is 2; two releases retire both, but the
|
||||||
|
// remaining active count still equals the shrunken target.
|
||||||
mgr_arc.release_permit("t0").await;
|
mgr_arc.release_permit("t0").await;
|
||||||
mgr_arc.release_permit("t1").await;
|
mgr_arc.release_permit("t1").await;
|
||||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||||
|
|
||||||
// Debt was 2; two releases retired both. The 2 remaining pending tasks
|
assert_eq!(
|
||||||
// (t4, t5) can now dispatch since debt is 0 and slots freed.
|
spawner.native_calls.load(Ordering::SeqCst),
|
||||||
|
4,
|
||||||
|
"pending tasks must not dispatch while active count already meets the shrunken target"
|
||||||
|
);
|
||||||
|
|
||||||
|
mgr_arc.release_permit("t2").await;
|
||||||
|
mgr_arc.release_permit("t3").await;
|
||||||
|
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||||
|
|
||||||
assert_eq!(
|
assert_eq!(
|
||||||
spawner.native_calls.load(Ordering::SeqCst),
|
spawner.native_calls.load(Ordering::SeqCst),
|
||||||
6,
|
6,
|
||||||
"shrink converges: after debt exhausted, remaining pending dispatch"
|
"pending tasks dispatch after active count falls below the shrunken target"
|
||||||
);
|
);
|
||||||
|
|
||||||
handle.abort();
|
handle.abort();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn aria2_resume_waits_for_shrunk_capacity() {
|
||||||
|
let (mgr, _spawner) = make_manager(3);
|
||||||
|
let mgr_arc = Arc::new(mgr);
|
||||||
|
|
||||||
|
let active = mgr_arc.acquire_permit().await.unwrap();
|
||||||
|
mgr_arc.park_permit("active", active).await;
|
||||||
|
let paused = mgr_arc.acquire_permit().await.unwrap();
|
||||||
|
mgr_arc.park_permit("paused", paused).await;
|
||||||
|
mgr_arc.release_permit("paused").await;
|
||||||
|
|
||||||
|
mgr_arc.set_capacity(1);
|
||||||
|
|
||||||
|
let waiter = {
|
||||||
|
let mgr_clone = Arc::clone(&mgr_arc);
|
||||||
|
tokio::spawn(async move { mgr_clone.ensure_aria2_permit("paused").await })
|
||||||
|
};
|
||||||
|
|
||||||
|
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||||
|
assert!(
|
||||||
|
!waiter.is_finished(),
|
||||||
|
"paused resume must wait while current active count already meets shrunk limit"
|
||||||
|
);
|
||||||
|
|
||||||
|
mgr_arc.release_permit("active").await;
|
||||||
|
assert!(timeout(Duration::from_secs(1), waiter)
|
||||||
|
.await
|
||||||
|
.expect("resume permit should unblock after active transfer exits")
|
||||||
|
.expect("resume task should not panic"));
|
||||||
|
}
|
||||||
|
|
||||||
fn aria2_task(id: &str) -> QueuedTask {
|
fn aria2_task(id: &str) -> QueuedTask {
|
||||||
QueuedTask {
|
QueuedTask {
|
||||||
id: id.to_string(),
|
id: id.to_string(),
|
||||||
|
|||||||
Reference in New Issue
Block a user