mirror of
https://github.com/nimbold/Firelink.git
synced 2026-08-12 20:47:22 +00:00
fix(downloads): harden aria2 recovery and resume lifecycle
This commit is contained in:
NimBold
@@ -110,6 +110,8 @@ pub struct DownloadItem {
|
||||
pub has_been_dispatched: Option<bool>,
|
||||
#[ts(optional)]
|
||||
pub last_error: Option<String>,
|
||||
#[ts(optional)]
|
||||
pub last_try: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize, TS)]
|
||||
|
||||
+204
-8
@@ -3746,6 +3746,80 @@ async fn aria2_download_status(port: u16, secret: &str, gid: &str) -> Result<Str
|
||||
.ok_or_else(|| format!("aria2.tellStatus returned no status for gid {gid}"))
|
||||
}
|
||||
|
||||
async fn reconcile_aria2_downloads(app_handle: &tauri::AppHandle) {
|
||||
let state = app_handle.state::<AppState>();
|
||||
let port = state.aria2_port.load(std::sync::atomic::Ordering::Relaxed);
|
||||
let secret = state.aria2_secret.clone();
|
||||
let mappings = state.queue_manager.aria2_gid_mappings();
|
||||
|
||||
for (gid, id) in mappings {
|
||||
let status = match rpc_call(
|
||||
port,
|
||||
&secret,
|
||||
"aria2.tellStatus",
|
||||
serde_json::json!([gid, ["status", "errorCode", "errorMessage"]]),
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(status) => status,
|
||||
Err(error) => {
|
||||
log::debug!(
|
||||
"aria2 reconnect reconciliation [{}]: could not query gid {}: {}",
|
||||
id,
|
||||
gid,
|
||||
error
|
||||
);
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
let status_name = status.get("status").and_then(|value| value.as_str());
|
||||
let outcome = match status_name {
|
||||
Some("complete") => Some(crate::queue::PendingOutcome::Complete),
|
||||
Some("error") | Some("removed") => {
|
||||
let error_code = status
|
||||
.get("errorCode")
|
||||
.and_then(|value| value.as_str())
|
||||
.filter(|value| !value.is_empty());
|
||||
let error_message = status
|
||||
.get("errorMessage")
|
||||
.and_then(|value| value.as_str())
|
||||
.filter(|value| !value.is_empty())
|
||||
.unwrap_or("aria2 download ended while the event channel was disconnected");
|
||||
Some(crate::queue::PendingOutcome::Error(match error_code {
|
||||
Some(code) => format!("aria2 error code {code}: {error_message}"),
|
||||
None => error_message.to_string(),
|
||||
}))
|
||||
}
|
||||
_ => None,
|
||||
};
|
||||
|
||||
if let Some(outcome) = outcome {
|
||||
state.queue_manager.handle_aria2_event(&gid, outcome).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn aria2_daemon_is_reachable(port: u16, secret: &str) -> bool {
|
||||
for attempt in 0..2 {
|
||||
if rpc_call(
|
||||
port,
|
||||
secret,
|
||||
"aria2.getVersion",
|
||||
serde_json::json!([]),
|
||||
)
|
||||
.await
|
||||
.is_ok()
|
||||
{
|
||||
return true;
|
||||
}
|
||||
if attempt == 0 {
|
||||
tokio::time::sleep(Duration::from_millis(250)).await;
|
||||
}
|
||||
}
|
||||
false
|
||||
}
|
||||
|
||||
fn aria2_gid_not_found(error: &str) -> bool {
|
||||
let lower = error.to_ascii_lowercase();
|
||||
lower.contains("gid") && lower.contains("not found")
|
||||
@@ -4713,6 +4787,7 @@ async fn clear_logs(app_handle: tauri::AppHandle) -> Result<(), String> {
|
||||
async fn export_logs(
|
||||
app_handle: tauri::AppHandle,
|
||||
state: tauri::State<'_, AppState>,
|
||||
destination: Option<String>,
|
||||
) -> Result<String, String> {
|
||||
let mut output = format!(
|
||||
"Firelink support logs\nVersion: {}\nOS: {} {}\nArchitecture: {}\nGenerated: {}\n\n",
|
||||
@@ -4773,6 +4848,16 @@ async fn export_logs(
|
||||
}
|
||||
output.push('\n');
|
||||
}
|
||||
|
||||
if let Some(destination) = destination {
|
||||
let destination = std::path::PathBuf::from(destination.trim());
|
||||
if destination.file_name().is_none() {
|
||||
return Err("log export destination must be a file path".to_string());
|
||||
}
|
||||
tokio::fs::write(&destination, &output)
|
||||
.await
|
||||
.map_err(|error| format!("failed to write exported logs to '{}': {error}", destination.display()))?;
|
||||
}
|
||||
Ok(output)
|
||||
}
|
||||
|
||||
@@ -5959,16 +6044,31 @@ pub fn run() {
|
||||
|
||||
let mut ws_retries = 0;
|
||||
loop {
|
||||
if ws_retries > 10 {
|
||||
log::error!("Max WebSocket reconnection attempts reached. aria2 integration is disabled.");
|
||||
if ws_retries == 10 {
|
||||
log::error!("Aria2 WebSocket is still unavailable after repeated reconnection attempts; continuing to retry.");
|
||||
let guard = app_handle_bg.state::<Aria2DaemonGuard>();
|
||||
*guard.startup_error.lock().unwrap() = Some("Max WebSocket reconnection attempts reached.".to_string());
|
||||
break;
|
||||
// Keep retrying. The daemon or local WebSocket can
|
||||
// recover after a prolonged outage; permanently
|
||||
// stopping this task strands active transfers.
|
||||
ws_retries = 0;
|
||||
}
|
||||
tokio::time::sleep(std::time::Duration::from_secs(2)).await;
|
||||
let ws_url = format!("ws://127.0.0.1:{}/jsonrpc", ws_port);
|
||||
if let Ok((ws_stream, _)) = tokio_tungstenite::connect_async(&ws_url).await {
|
||||
ws_retries = 0; // reset on success
|
||||
if let Ok(mut startup_error) = app_handle_bg
|
||||
.state::<Aria2DaemonGuard>()
|
||||
.startup_error
|
||||
.lock()
|
||||
{
|
||||
if startup_error.as_deref()
|
||||
== Some("Max WebSocket reconnection attempts reached.")
|
||||
{
|
||||
*startup_error = None;
|
||||
}
|
||||
}
|
||||
reconcile_aria2_downloads(&app_handle_bg).await;
|
||||
use futures_util::StreamExt;
|
||||
let (_, mut read) = ws_stream.split();
|
||||
while let Some(msg) = read.next().await {
|
||||
@@ -6025,7 +6125,11 @@ pub fn run() {
|
||||
}
|
||||
}
|
||||
ws_retries += 1;
|
||||
app_handle_bg.state::<AppState>().queue_manager.clear_aria2_permits().await;
|
||||
let state = app_handle_bg.state::<AppState>();
|
||||
let port = state.aria2_port.load(std::sync::atomic::Ordering::Relaxed);
|
||||
if !aria2_daemon_is_reachable(port, &state.aria2_secret).await {
|
||||
state.queue_manager.clear_aria2_permits().await;
|
||||
}
|
||||
tokio::time::sleep(std::time::Duration::from_secs(4)).await;
|
||||
}
|
||||
});
|
||||
@@ -6035,19 +6139,99 @@ pub fn run() {
|
||||
let poll_secret = aria2_secret.clone();
|
||||
let poll_mgr = Arc::clone(&queue_manager_poll);
|
||||
tauri::async_runtime::spawn(async move {
|
||||
#[derive(Default)]
|
||||
struct Aria2ConnectionObservation {
|
||||
gid: String,
|
||||
saw_multiple_connections: bool,
|
||||
degraded_since: Option<Instant>,
|
||||
no_progress_since: Option<Instant>,
|
||||
refreshed: bool,
|
||||
last_completed: u64,
|
||||
}
|
||||
|
||||
const RECOVERY_DELAY: Duration = Duration::from_secs(30);
|
||||
const MIN_REMAINING_FOR_CONNECTION_RECOVERY: u64 = 1024 * 1024;
|
||||
let mut interval = tokio::time::interval(std::time::Duration::from_millis(1000));
|
||||
let mut observations: HashMap<String, Aria2ConnectionObservation> = HashMap::new();
|
||||
loop {
|
||||
interval.tick().await;
|
||||
let params = serde_json::json!([["gid", "status", "totalLength", "completedLength", "downloadSpeed", "errorMessage"]]);
|
||||
let params = serde_json::json!([["gid", "status", "totalLength", "completedLength", "downloadSpeed", "connections", "errorMessage"]]);
|
||||
if let Ok(active_list) = rpc_call(poll_port.load(std::sync::atomic::Ordering::Relaxed), &poll_secret, "aria2.tellActive", params).await {
|
||||
if let Some(active_arr) = active_list.as_array() {
|
||||
let mut seen_ids = std::collections::HashSet::new();
|
||||
for status_info in active_arr {
|
||||
let gid = status_info.get("gid").and_then(|s| s.as_str()).unwrap_or("");
|
||||
let id = poll_mgr.aria2_gids.read().unwrap().get(gid).cloned();
|
||||
if let Some(id) = id {
|
||||
seen_ids.insert(id.clone());
|
||||
let status = status_info.get("status").and_then(|value| value.as_str()).unwrap_or("");
|
||||
let total = status_info.get("totalLength").and_then(|s| s.as_str()).unwrap_or("0").parse::<u64>().unwrap_or(0);
|
||||
let completed = status_info.get("completedLength").and_then(|s| s.as_str()).unwrap_or("0").parse::<u64>().unwrap_or(0);
|
||||
let speed_bytes = status_info.get("downloadSpeed").and_then(|s| s.as_str()).unwrap_or("0").parse::<f64>().unwrap_or(0.0);
|
||||
let active_connections = status_info.get("connections").and_then(|s| s.as_str()).unwrap_or("0").parse::<i32>().unwrap_or(0);
|
||||
let requested_connections = poll_mgr
|
||||
.aria2_requested_connections(&id)
|
||||
.await
|
||||
.unwrap_or(1)
|
||||
.max(1);
|
||||
let now = Instant::now();
|
||||
let observation = observations.entry(id.clone()).or_default();
|
||||
if observation.gid != gid {
|
||||
*observation = Aria2ConnectionObservation {
|
||||
gid: gid.to_string(),
|
||||
last_completed: completed,
|
||||
..Default::default()
|
||||
};
|
||||
}
|
||||
|
||||
let remaining = total.saturating_sub(completed);
|
||||
let mut should_refresh = false;
|
||||
if status == "active" && total > completed {
|
||||
if completed > observation.last_completed {
|
||||
observation.no_progress_since = None;
|
||||
} else if speed_bytes <= 0.0 {
|
||||
let stalled_since = observation.no_progress_since.get_or_insert(now);
|
||||
if now.duration_since(*stalled_since) >= RECOVERY_DELAY
|
||||
&& !observation.refreshed
|
||||
{
|
||||
observation.refreshed = true;
|
||||
should_refresh = true;
|
||||
}
|
||||
} else {
|
||||
observation.no_progress_since = None;
|
||||
}
|
||||
|
||||
if requested_connections > 1 && active_connections > 1 {
|
||||
observation.saw_multiple_connections = true;
|
||||
observation.degraded_since = None;
|
||||
if !should_refresh {
|
||||
observation.refreshed = false;
|
||||
}
|
||||
} else if requested_connections > 1
|
||||
&& observation.saw_multiple_connections
|
||||
&& active_connections <= 1
|
||||
&& remaining >= MIN_REMAINING_FOR_CONNECTION_RECOVERY
|
||||
{
|
||||
let degraded_since = observation.degraded_since.get_or_insert(now);
|
||||
if now.duration_since(*degraded_since) >= RECOVERY_DELAY
|
||||
&& !observation.refreshed
|
||||
{
|
||||
observation.refreshed = true;
|
||||
should_refresh = true;
|
||||
log::warn!(
|
||||
"aria2 connection pool degraded [{}]: gid {} has {} connection(s), requested {}; refreshing",
|
||||
id,
|
||||
gid,
|
||||
active_connections,
|
||||
requested_connections
|
||||
);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
observation.degraded_since = None;
|
||||
observation.no_progress_since = None;
|
||||
}
|
||||
observation.last_completed = completed;
|
||||
|
||||
let fraction = if total > 0 { completed as f64 / total as f64 } else { 0.0 };
|
||||
let speed = crate::download::format_speed(speed_bytes);
|
||||
@@ -6064,15 +6248,27 @@ pub fn run() {
|
||||
|
||||
use tauri::Emitter;
|
||||
let _ = app_handle_poll.emit("download-progress", DownloadProgressEvent {
|
||||
id,
|
||||
id: id.clone(),
|
||||
fraction,
|
||||
speed,
|
||||
eta,
|
||||
size,
|
||||
size_is_final: false,
|
||||
});
|
||||
size_is_final: false,
|
||||
});
|
||||
|
||||
if should_refresh {
|
||||
if let Err(error) = poll_mgr.refresh_aria2_connections(&id, gid).await {
|
||||
log::warn!(
|
||||
"aria2 connection refresh [{}] for gid {} failed: {}",
|
||||
id,
|
||||
gid,
|
||||
error
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
observations.retain(|id, _| seen_ids.contains(id));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -76,6 +76,18 @@ pub trait SidecarSpawner: Send + Sync + 'static {
|
||||
/// cancellation.
|
||||
async fn remove_uri(&self, gid: &str) -> Result<(), String>;
|
||||
|
||||
/// Recycle the connections for an active aria2 transfer without changing
|
||||
/// its gid or releasing its queue permit. Production uses forcePause /
|
||||
/// unpause; test spawners can leave this unsupported.
|
||||
async fn refresh_uri(&self, _gid: &str) -> Result<(), String> {
|
||||
Err("aria2 connection refresh is unavailable".to_string())
|
||||
}
|
||||
|
||||
/// Leave an aria2 transfer paused after a refresh races with a user pause.
|
||||
async fn pause_uri(&self, _gid: &str) -> Result<(), String> {
|
||||
Err("aria2 pause is unavailable".to_string())
|
||||
}
|
||||
|
||||
/// Run a media download to completion. The permit is parked for the full
|
||||
/// duration; release is handled by QueueManager on the runner's exit.
|
||||
async fn run_media(&self, id: &str, payload: &SpawnPayload) -> Result<(), String>;
|
||||
@@ -297,6 +309,15 @@ impl<R: tauri::Runtime> QueueManager<R> {
|
||||
== epoch
|
||||
}
|
||||
|
||||
pub async fn current_aria2_control_epoch(&self, id: &str) -> u64 {
|
||||
self.aria2_control_epochs
|
||||
.lock()
|
||||
.await
|
||||
.get(id)
|
||||
.copied()
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
pub async fn is_aria2_retry_cancelled(&self, id: &str) -> bool {
|
||||
self.aria2_retry_cancelled.lock().await.contains(id)
|
||||
}
|
||||
@@ -305,6 +326,14 @@ impl<R: tauri::Runtime> QueueManager<R> {
|
||||
self.aria2_retry_strikes.lock().await.contains_key(id)
|
||||
}
|
||||
|
||||
pub async fn aria2_requested_connections(&self, id: &str) -> Option<i32> {
|
||||
self.aria2_payloads
|
||||
.lock()
|
||||
.await
|
||||
.get(id)
|
||||
.and_then(|payload| payload.connections)
|
||||
}
|
||||
|
||||
/// Pop the next task, or None if empty.
|
||||
pub async fn pop_front(&self) -> Option<QueuedTask> {
|
||||
self.pending.lock().await.pop_front()
|
||||
@@ -659,6 +688,39 @@ impl<R: tauri::Runtime> QueueManager<R> {
|
||||
.find_map(|(gid, download_id)| (download_id == id).then(|| gid.clone()))
|
||||
}
|
||||
|
||||
pub fn aria2_gid_mappings(&self) -> Vec<(String, String)> {
|
||||
self.aria2_gids
|
||||
.read()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.map(|(gid, id)| (gid.clone(), id.clone()))
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Recycle an active transfer's connections after the poller observes a
|
||||
/// persistent connection-pool collapse or a true zero-progress stall.
|
||||
/// The transfer keeps its gid, partial file, and queue permit.
|
||||
pub async fn refresh_aria2_connections(&self, id: &str, gid: &str) -> Result<(), String> {
|
||||
if self.aria2_gid_for_download(id).as_deref() != Some(gid)
|
||||
|| !self.is_registered(id).await
|
||||
|| self.is_aria2_retry_cancelled(id).await
|
||||
{
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let epoch = self.current_aria2_control_epoch(id).await;
|
||||
self.spawner.refresh_uri(gid).await?;
|
||||
|
||||
let still_current = self.is_registered(id).await
|
||||
&& !self.is_aria2_retry_cancelled(id).await
|
||||
&& self.is_aria2_control_epoch_current(id, epoch).await
|
||||
&& self.aria2_gid_for_download(id).as_deref() == Some(gid);
|
||||
if !still_current {
|
||||
let _ = self.spawner.pause_uri(gid).await;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Remove every gid mapping for a download and discard buffered terminal
|
||||
/// events for those gids. Returns the most recently encountered gid.
|
||||
pub async fn forget_aria2_gid(&self, id: &str) -> Option<String> {
|
||||
@@ -1250,6 +1312,8 @@ impl SidecarSpawner for ProductionSpawner {
|
||||
let mt = aria2_attempt_limit(payload.max_tries);
|
||||
options.insert("max-tries".to_string(), serde_json::json!(mt.to_string()));
|
||||
options.insert("retry-wait".to_string(), serde_json::json!("2"));
|
||||
options.insert("connect-timeout".to_string(), serde_json::json!("20"));
|
||||
options.insert("timeout".to_string(), serde_json::json!("60"));
|
||||
options.insert("continue".to_string(), serde_json::json!("true"));
|
||||
options.insert("always-resume".to_string(), serde_json::json!("true"));
|
||||
options.insert("auto-file-renaming".to_string(), serde_json::json!("false"));
|
||||
@@ -1332,6 +1396,44 @@ impl SidecarSpawner for ProductionSpawner {
|
||||
}
|
||||
}
|
||||
|
||||
async fn refresh_uri(&self, gid: &str) -> Result<(), String> {
|
||||
let state = self.app_handle.state::<crate::AppState>();
|
||||
let port = state.aria2_port.load(std::sync::atomic::Ordering::Relaxed);
|
||||
let secret = &state.aria2_secret;
|
||||
let paused = crate::rpc_call(
|
||||
port,
|
||||
secret,
|
||||
"aria2.forcePause",
|
||||
serde_json::json!([gid]),
|
||||
)
|
||||
.await
|
||||
.map_err(|error| format!("failed to refresh aria2 gid {gid}: {error}"))?;
|
||||
crate::ensure_aria2_gid_result("forcePause", gid, &paused)?;
|
||||
|
||||
let resumed = crate::rpc_call(
|
||||
port,
|
||||
secret,
|
||||
"aria2.unpause",
|
||||
serde_json::json!([gid]),
|
||||
)
|
||||
.await
|
||||
.map_err(|error| format!("failed to refresh aria2 gid {gid}: {error}"))?;
|
||||
crate::ensure_aria2_gid_result("unpause", gid, &resumed)
|
||||
}
|
||||
|
||||
async fn pause_uri(&self, gid: &str) -> Result<(), String> {
|
||||
let state = self.app_handle.state::<crate::AppState>();
|
||||
let result = crate::rpc_call(
|
||||
state.aria2_port.load(std::sync::atomic::Ordering::Relaxed),
|
||||
&state.aria2_secret,
|
||||
"aria2.forcePause",
|
||||
serde_json::json!([gid]),
|
||||
)
|
||||
.await
|
||||
.map_err(|error| format!("failed to pause aria2 gid {gid}: {error}"))?;
|
||||
crate::ensure_aria2_gid_result("forcePause", gid, &result)
|
||||
}
|
||||
|
||||
async fn run_media(&self, id: &str, payload: &SpawnPayload) -> Result<(), String> {
|
||||
let state = self.app_handle.state::<crate::AppState>();
|
||||
let mut cancel_rx = state
|
||||
|
||||
+10
-1
@@ -97,7 +97,7 @@ pub fn is_transient_network_error(message: &str) -> bool {
|
||||
|
||||
let m = message.to_ascii_lowercase();
|
||||
|
||||
const TRANSIENT: [&str; 35] = [
|
||||
const TRANSIENT: [&str; 38] = [
|
||||
// socket-layer / HTTP-client phrasing surfaced by aria2 and yt-dlp
|
||||
"timed out",
|
||||
"timeout",
|
||||
@@ -113,6 +113,9 @@ pub fn is_transient_network_error(message: &str) -> bool {
|
||||
"connection aborted",
|
||||
"error sending request", // reqwest wrapper for connect/send failures
|
||||
"dns error", // transient resolver failures
|
||||
"protocol error", // aria2 read/protocol failures after a link drop
|
||||
"tls handshake failure",
|
||||
"ssl/tls handshake failure",
|
||||
// HTTP-level transient
|
||||
"http 408",
|
||||
"request timeout",
|
||||
@@ -275,6 +278,12 @@ mod tests {
|
||||
assert!(is_transient_network_error("The response status is not successful. status=503"));
|
||||
assert!(is_transient_network_error("The response status is not successful. status=502"));
|
||||
assert!(is_transient_network_error("Invalid range header. Request: 106954752-361758719/383882118, Response: 106954752-383882117/383882118"));
|
||||
assert!(is_transient_network_error(
|
||||
"aria2 error code 1: Failed to receive data, cause: protocol error"
|
||||
));
|
||||
assert!(is_transient_network_error(
|
||||
"SSL/TLS handshake failure: protocol error"
|
||||
));
|
||||
}
|
||||
|
||||
// --- transient classification: negative cases -------------------------
|
||||
|
||||
Reference in New Issue
Block a user