From d1452ce0c1a762d817e218a4fcf3efcfc0322e32 Mon Sep 17 00:00:00 2001 From: NimBold Date: Wed, 22 Jul 2026 14:04:59 +0330 Subject: [PATCH] feat(downloads): expose active Aria2 connections --- src-tauri/src/lib.rs | 82 +++++++++++++++++++++++---- src-tauri/src/queue.rs | 78 +++++++++++++++++++++++-- src-tauri/tests/queue_manager.rs | 26 +++++++++ src/bindings/DownloadProgressEvent.ts | 2 +- src/components/PropertiesModal.tsx | 29 +++++++++- src/i18n/catalogs/en.ts | 3 + src/i18n/catalogs/fa.ts | 3 + src/i18n/catalogs/he.ts | 3 + src/i18n/catalogs/ru.ts | 3 + src/i18n/catalogs/uk.ts | 3 + src/i18n/catalogs/zh-CN.ts | 3 + src/store/downloadStore.test.ts | 34 +++++++++++ 12 files changed, 249 insertions(+), 20 deletions(-) diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 31c2b7d..6375690 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -1464,6 +1464,8 @@ fn emit_media_progress( downloaded_bytes, total_bytes, total_is_estimate, + active_connections: None, + requested_connections: None, }, ); state.last_progress_at = now; @@ -2988,6 +2990,10 @@ pub struct DownloadProgressEvent { total_bytes: Option, #[ts(optional)] total_is_estimate: Option, + #[ts(optional)] + active_connections: Option, + #[ts(optional)] + requested_connections: Option, } #[derive(Debug, Clone, Serialize, TS)] @@ -4100,6 +4106,8 @@ pub(crate) async fn start_media_download_internal( downloaded_bytes: None, total_bytes: None, total_is_estimate: Some(false), + active_connections: None, + requested_connections: None, }); } let lower = line.to_lowercase(); @@ -4171,6 +4179,8 @@ pub(crate) async fn start_media_download_internal( downloaded_bytes: Some(metadata.len() as f64), total_bytes: Some(metadata.len() as f64), total_is_estimate: Some(false), + active_connections: None, + requested_connections: None, }); } } @@ -6598,12 +6608,34 @@ mod tests { MediaProgressEmitterState, MediaSpeedSampler, MEDIA_PROGRESS_PREFIX, observe_aria2_connections, observe_aria2_connections_with_epoch, Aria2ConnectionObservation, Aria2ConnectionSample, Aria2RecoveryReason, + aria2_active_connection_count, parse_media_playlist_metadata, validate_enqueue_url, validate_enqueue_uris, }; use serde_json::json; use std::time::{Duration, Instant}; + #[test] + fn aria2_active_connection_count_uses_only_nonnegative_daemon_values() { + assert_eq!( + aria2_active_connection_count(&json!({"connections": "16"})), + 16 + ); + assert_eq!( + aria2_active_connection_count(&json!({"connections": 16})), + 16 + ); + assert_eq!( + aria2_active_connection_count(&json!({"connections": "-1"})), + 0 + ); + assert_eq!( + aria2_active_connection_count(&json!({"connections": -1})), + 0 + ); + assert_eq!(aria2_active_connection_count(&json!({})), 0); + } + #[tokio::test] async fn enqueue_url_validation_blocks_local_http_but_preserves_ftp() { assert_eq!( @@ -8312,6 +8344,19 @@ struct Aria2ConnectionSample<'a> { now: Instant, } +fn aria2_active_connection_count(status_info: &serde_json::Value) -> i32 { + status_info + .get("connections") + .and_then(|value| { + value + .as_str() + .and_then(|value| value.parse::().ok()) + .or_else(|| value.as_i64().and_then(|value| i32::try_from(value).ok())) + }) + .filter(|value| *value >= 0) + .unwrap_or(0) +} + const ARIA2_CONNECTION_RECOVERY_DELAY: Duration = Duration::from_secs(30); const ARIA2_CONNECTION_RECOVERY_COOLDOWN: Duration = Duration::from_secs(45); const ARIA2_MIN_REMAINING_FOR_CONNECTION_RECOVERY: u64 = 1024 * 1024; @@ -8928,28 +8973,39 @@ pub fn run() { let mut seen_gids = 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) - .map(|mapping| mapping.id.clone()); - if let Some(id) = id { - seen_ids.insert(id.clone()); - seen_gids.insert(gid.to_string()); + let Some(mapping) = poll_mgr.aria2_gid_mapping(gid) else { + continue; + }; + let id = mapping.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::().unwrap_or(0); let completed = status_info.get("completedLength").and_then(|s| s.as_str()).unwrap_or("0").parse::().unwrap_or(0); let speed_bytes = status_info.get("downloadSpeed").and_then(|s| s.as_str()).unwrap_or("0").parse::().unwrap_or(0.0); - let active_connections = status_info.get("connections").and_then(|s| s.as_str()).unwrap_or("0").parse::().unwrap_or(0); + let active_connections = + aria2_active_connection_count(status_info); let requested_connections = poll_mgr .aria2_requested_connections(&id) .await .unwrap_or(1) .max(1); let speed_limited = poll_mgr.aria2_speed_limited(&id).await; - let control_epoch = - poll_mgr.current_aria2_control_epoch(&id).await; + let control_epoch = mapping.epoch; + // The status snapshot and the requested + // connection lookup both await. A pause, + // retry, or same-GID resume may replace the + // mapping while those awaits are in + // flight. Only emit telemetry if this + // snapshot still owns the same GID epoch. + if !poll_mgr.is_current_aria2_gid_mapping(gid, &mapping) + || !poll_mgr + .is_aria2_control_epoch_current(&id, control_epoch) + .await + { + continue; + } + seen_ids.insert(id.clone()); + seen_gids.insert(gid.to_string()); let now = Instant::now(); let observation = observations.entry(id.clone()).or_default(); let recovery_reason = observe_aria2_connections_with_epoch( @@ -8992,6 +9048,8 @@ pub fn run() { downloaded_bytes: Some(completed as f64), total_bytes: (total > 0).then_some(total as f64), total_is_estimate: Some(false), + active_connections: Some(active_connections), + requested_connections: Some(requested_connections), }); if let Some(reason) = recovery_reason { diff --git a/src-tauri/src/queue.rs b/src-tauri/src/queue.rs index 8a0d720..2e9645e 100644 --- a/src-tauri/src/queue.rs +++ b/src-tauri/src/queue.rs @@ -18,7 +18,7 @@ pub const MEDIA_RUN_CANCELLED: &str = "__firelink_media_run_cancelled__"; type Aria2ControlLocks = Arc>>>>; -#[derive(Debug, Clone)] +#[derive(Debug, Clone, PartialEq, Eq)] pub struct Aria2GidMapping { pub id: String, pub epoch: u64, @@ -1111,6 +1111,22 @@ impl QueueManager { .find_map(|(gid, mapping)| (mapping.id == id).then(|| gid.clone())) } + /// Capture the GID ownership token used by the poller. The mapping's + /// epoch must still be current when an asynchronous status snapshot is + /// emitted; otherwise a late response from an older GID can be attributed + /// to a newer lifecycle for the same download id. + pub fn aria2_gid_mapping(&self, gid: &str) -> Option { + self.aria2_gids.read().unwrap().get(gid).cloned() + } + + pub fn is_current_aria2_gid_mapping( + &self, + gid: &str, + expected: &Aria2GidMapping, + ) -> bool { + self.aria2_gid_mapping(gid).as_ref() == Some(expected) + } + pub fn aria2_gid_mappings(&self) -> Vec<(String, String)> { self.aria2_gids .read() @@ -1854,6 +1870,30 @@ pub struct ProductionSpawner { app_handle: AppHandle, } +const ARIA2_MIN_SPLIT_SIZE: &str = "1M"; + +fn apply_aria2_connection_options( + options: &mut serde_json::Map, + connections: i32, +) { + let connections = connections.max(1); + options.insert( + "split".to_string(), + serde_json::json!(connections.to_string()), + ); + options.insert( + "max-connection-per-server".to_string(), + serde_json::json!(connections.to_string()), + ); + // aria2's 20M default suppresses segmentation for files smaller than + // 40M. Keep the requested connection count useful for ordinary release + // assets while retaining a 1M lower bound to avoid tiny range requests. + options.insert( + "min-split-size".to_string(), + serde_json::json!(ARIA2_MIN_SPLIT_SIZE), + ); +} + impl ProductionSpawner { pub fn new(app_handle: AppHandle) -> Self { Self { app_handle } @@ -1909,11 +1949,7 @@ impl SidecarSpawner for ProductionSpawner { crate::download_ownership::canonical_download_filename(&payload.filename); options.insert("out".to_string(), serde_json::json!(safe_filename)); let conn = effective_aria2_connections(id, payload).await; - options.insert("split".to_string(), serde_json::json!(conn.to_string())); - options.insert( - "max-connection-per-server".to_string(), - serde_json::json!(conn.to_string()), - ); + apply_aria2_connection_options(&mut options, conn); 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")); @@ -2208,6 +2244,36 @@ impl EnqueueItem { mod tests { use super::*; + #[test] + fn aria2_connection_options_enable_requested_ranges_for_small_release_assets() { + let mut options = serde_json::Map::new(); + + apply_aria2_connection_options(&mut options, 16); + + assert_eq!(options.get("split"), Some(&serde_json::json!("16"))); + assert_eq!( + options.get("max-connection-per-server"), + Some(&serde_json::json!("16")) + ); + assert_eq!( + options.get("min-split-size"), + Some(&serde_json::json!("1M")) + ); + } + + #[test] + fn aria2_connection_options_never_emit_zero_connections() { + let mut options = serde_json::Map::new(); + + apply_aria2_connection_options(&mut options, 0); + + assert_eq!(options.get("split"), Some(&serde_json::json!("1"))); + assert_eq!( + options.get("max-connection-per-server"), + Some(&serde_json::json!("1")) + ); + } + #[test] fn bounded_range_probe_accepts_exact_requested_byte() { assert_eq!( diff --git a/src-tauri/tests/queue_manager.rs b/src-tauri/tests/queue_manager.rs index 1089532..93b5790 100644 --- a/src-tauri/tests/queue_manager.rs +++ b/src-tauri/tests/queue_manager.rs @@ -357,6 +357,32 @@ async fn resumed_gid_rebinds_to_the_new_control_epoch() { assert!(manager.aria2_gid_for_download("resumed-gid").is_none()); } +#[tokio::test] +async fn gid_mapping_snapshot_is_invalid_after_epoch_or_gid_replacement() { + let (mgr, _spawner) = make_manager(1); + + mgr.remember_gid("snapshot".to_string(), "gid-old".to_string()) + .await; + let old_mapping = mgr + .aria2_gid_mapping("gid-old") + .expect("the first gid should be mapped"); + assert!(mgr.is_current_aria2_gid_mapping("gid-old", &old_mapping)); + + let resume_epoch = mgr.next_aria2_control_epoch("snapshot").await; + assert!(mgr + .rebind_aria2_gid_epoch("snapshot", "gid-old", resume_epoch) + .await); + assert!(!mgr.is_current_aria2_gid_mapping("gid-old", &old_mapping)); + + mgr.remember_gid("snapshot".to_string(), "gid-new".to_string()) + .await; + assert!(mgr.aria2_gid_mapping("gid-old").is_none()); + let new_mapping = mgr + .aria2_gid_mapping("gid-new") + .expect("the replacement gid should be mapped"); + assert!(mgr.is_current_aria2_gid_mapping("gid-new", &new_mapping)); +} + #[tokio::test] async fn forgetting_aria2_gid_clears_mapping_without_releasing_twice() { let (mgr, _spawner) = make_manager(1); diff --git a/src/bindings/DownloadProgressEvent.ts b/src/bindings/DownloadProgressEvent.ts index c909bda..a497cec 100644 --- a/src/bindings/DownloadProgressEvent.ts +++ b/src/bindings/DownloadProgressEvent.ts @@ -1,3 +1,3 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. -export type DownloadProgressEvent = { id: string, fraction: number, speed: string, eta: string, size: string | null, size_is_final: boolean, downloaded_bytes?: number, total_bytes?: number, total_is_estimate?: boolean, }; +export type DownloadProgressEvent = { id: string, fraction: number, speed: string, eta: string, size: string | null, size_is_final: boolean, downloaded_bytes?: number, total_bytes?: number, total_is_estimate?: boolean, active_connections?: number, requested_connections?: number, }; diff --git a/src/components/PropertiesModal.tsx b/src/components/PropertiesModal.tsx index 8240fa6..ef3e496 100644 --- a/src/components/PropertiesModal.tsx +++ b/src/components/PropertiesModal.tsx @@ -207,6 +207,32 @@ export const PropertiesModal = () => { const identityLocked = getIdentityLocked(item.status); const transferLocked = getTransferLocked(item.status); + const configuredConnections = resolveDownloadConnections(item.connections, perServerConnections); + const observedConnectionTotal = Math.max( + 1, + liveProgress?.requested_connections ?? configuredConnections + ); + const observedActiveConnections = liveProgress?.active_connections; + const connectionTelemetryActive = item.status === 'downloading' || + item.status === 'processing' || + item.status === 'retrying'; + const connectionStatus = (() => { + if (item.isMedia) return t($ => $.properties.connectionsUnavailable); + if (!connectionTelemetryActive) return String(configuredConnections); + if (typeof observedActiveConnections === 'number') { + return t($ => $.properties.connectionCount, { + active: observedActiveConnections, + total: observedConnectionTotal, + }); + } + if (item.status === 'downloading') { + return t($ => $.properties.connectionCountUnknown, { total: observedConnectionTotal }); + } + return t($ => $.properties.connectionCount, { + active: 0, + total: observedConnectionTotal, + }); + })(); const displayedFraction = item.status === 'completed' ? 1 : liveProgress?.fraction ?? item.fraction ?? 0; @@ -298,7 +324,7 @@ export const PropertiesModal = () => {
{t($ => $.properties.speed)}{displayedSpeed}
{t($ => $.properties.eta)}{displayedEta}
-
{t($ => $.properties.connections)} $.properties.savedTooltip) : t($ => $.properties.defaultTooltip)}>{resolveDownloadConnections(item.connections, perServerConnections)}{item.connections !== undefined ? t($ => $.properties.saved) : t($ => $.properties.defaultValue)}
+
{t($ => $.properties.connections)} $.properties.savedTooltip) : t($ => $.properties.defaultTooltip)}>{connectionStatus}{item.connections !== undefined ? t($ => $.properties.saved) : t($ => $.properties.defaultValue)}
{t($ => $.properties.speedCap)}{item.speedLimit || '-'}
{t($ => $.properties.category)}{categoryLabel(item.category)}
{t($ => $.properties.lastTry)}{formatLastTry(item.lastTry)}
@@ -352,6 +378,7 @@ export const PropertiesModal = () => {
{ setConnections(Number(e.target.value)); setConnectionsDirty(true); }} disabled={transferLocked} className="w-16 bg-bg-input border border-border-modal rounded-lg px-2.5 py-1.5 text-xs text-text-primary focus:outline-none focus:border-accent disabled:opacity-50" /> {t($ => $.properties.perFile)} + {connectionStatus} {!transferLocked && item.connections !== undefined && item.connections !== perServerConnections && (