gsadmin/Firelink
1
0
Fork 0
mirror of https://github.com/nimbold/Firelink.git synced 2026-08-22 17:06:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(downloads): surface allocation and retain terminal progress

Browse Source 
- emit native allocation state around Aria2 preallocation
- preflight batch destinations before backend admission
- preserve exact progress across retries and terminal states
- fence allocation and progress across pauses, retries, and stale GIDs
This commit is contained in:
NimBold
2026-08-21 01:33:13 +03:30
parent c355e99913
commit 2d265ce7c8
12 changed files with 1652 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/ipc.rs
+40
View File
@@ -796,6 +796,31 @@ pub enum QueueDirection {
Down,
}
#[derive(Clone, Debug, Serialize, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export, export_to = "../../src/bindings/")]
pub struct DownloadStateProgress {
pub fraction: f64,
#[serde(skip_serializing_if = "Option::is_none")]
#[ts(optional)]
pub downloaded_bytes: Option<f64>,
#[serde(skip_serializing_if = "Option::is_none")]
#[ts(optional)]
pub total_bytes: Option<f64>,
#[serde(skip_serializing_if = "Option::is_none")]
#[ts(optional)]
pub total_is_estimate: Option<bool>,
}
#[derive(Clone, Debug, Serialize, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export, export_to = "../../src/bindings/")]
pub struct DownloadAllocationEvent {
pub id: String,
pub pending: bool,
pub lifecycle_generation: String,
}
#[derive(Clone, Debug, Serialize, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export, export_to = "../../src/bindings/")]
@@ -816,6 +841,9 @@ pub struct DownloadStateEvent {
pub destination: Option<String>,
#[ts(optional)]
pub torrent_seed_remaining: Option<f64>,
#[serde(skip_serializing_if = "Option::is_none")]
#[ts(optional)]
pub progress: Option<DownloadStateProgress>,
}
impl DownloadStateEvent {
@@ -829,6 +857,7 @@ impl DownloadStateEvent {
file_name: None,
destination: None,
torrent_seed_remaining: None,
progress: None,
}
}
@@ -843,6 +872,7 @@ impl DownloadStateEvent {
file_name: None,
destination: None,
torrent_seed_remaining: None,
progress: None,
}
}
@@ -857,6 +887,7 @@ impl DownloadStateEvent {
file_name: None,
destination: None,
torrent_seed_remaining: None,
progress: None,
}
}
@@ -870,6 +901,7 @@ impl DownloadStateEvent {
file_name: None,
destination: None,
torrent_seed_remaining: remaining,
progress: None,
}
}
@@ -883,6 +915,7 @@ impl DownloadStateEvent {
file_name: Some(file_name.into()),
destination: None,
torrent_seed_remaining: None,
progress: None,
}
}
@@ -899,6 +932,7 @@ impl DownloadStateEvent {
file_name: None,
destination: None,
torrent_seed_remaining: None,
progress: None,
}
}
@@ -912,6 +946,7 @@ impl DownloadStateEvent {
file_name: None,
destination: None,
torrent_seed_remaining: remaining,
progress: None,
}
}
@@ -929,6 +964,11 @@ impl DownloadStateEvent {
self
}
pub fn with_progress(mut self, progress: DownloadStateProgress) -> Self {
self.progress = Some(progress);
self
}
fn safe_error(error: impl Into<String>) -> (String, Option<DownloadErrorKind>) {
let error = crate::redact_sensitive_text(&error.into());
let error_kind = crate::retry::is_aria2_name_resolution_error(&error)
src-tauri/src/lib.rs
+311 -73
View File
@@ -5144,8 +5144,14 @@ async fn pause_download(
let removed_pending = state.queue_manager.remove_from_pending(&id).await;
let gid = state.queue_manager.aria2_gid_for_download(&id);
if gid.is_none() {
// A queued or not-yet-registered transfer cannot be verified through
// Aria2. Removing it from pending is the terminal pause boundary for
// this lifecycle, so its native allocation marker can be cleared now.
state.queue_manager.clear_aria2_allocation(&id).await;
}
if let Some(gid) = gid.as_deref() {
let status = aria2_download_status(
let (status, mut status_progress) = aria2_download_status_snapshot(
state.aria2_port.load(std::sync::atomic::Ordering::Relaxed),
&state.aria2_secret,
gid,
@@ -5153,6 +5159,7 @@ async fn pause_download(
.await?;
match status.as_str() {
"paused" => {
state.queue_manager.clear_aria2_allocation(&id).await;
state.queue_manager.next_aria2_control_epoch(&id).await;
state.queue_manager.cancel_aria2_retries(&id).await;
log::info!("aria2 pause [{}]: gid {} was already paused", id, gid);
@@ -5171,14 +5178,16 @@ async fn pause_download(
Err(error) => Err(format!("failed to pause aria2 gid {gid}: {error}")),
};
if let Err(pause_error) = pause_result {
match aria2_download_status(
match aria2_download_status_snapshot(
state.aria2_port.load(std::sync::atomic::Ordering::Relaxed),
&state.aria2_secret,
gid,
)
.await
{
Ok(status) if status == "paused" => {
Ok((status, progress)) if status == "paused" => {
state.queue_manager.clear_aria2_allocation(&id).await;
status_progress = progress.or(status_progress);
// forcePause may have returned an RPC error after
// the daemon actually paused the GID. Invalidate
// terminal events already in flight before
@@ -5191,27 +5200,34 @@ async fn pause_download(
gid
);
}
Ok(status) if status == "complete" => {
Ok((status, progress)) if status == "complete" => {
state.queue_manager.clear_aria2_allocation(&id).await;
state
.queue_manager
.apply_completion_locked(&id, crate::queue::PendingOutcome::Complete)
.apply_completion_locked_with_progress(
&id,
crate::queue::PendingOutcome::Complete,
progress.or(status_progress.clone()),
)
.await;
return Ok(());
}
Ok(status) if matches!(status.as_str(), "error" | "removed") => {
Ok((status, progress)) if matches!(status.as_str(), "error" | "removed") => {
state.queue_manager.clear_aria2_allocation(&id).await;
let terminal_error = format!(
"cannot pause aria2 gid {gid}: {pause_error}; daemon reports terminal state {status}"
);
state
.queue_manager
.apply_completion_locked(
.apply_completion_locked_with_progress(
&id,
crate::queue::PendingOutcome::Error(terminal_error.clone()),
progress.or(status_progress.clone()),
)
.await;
return Err(terminal_error);
}
Ok(status) => {
Ok((status, _)) => {
state.queue_manager.allow_aria2_retries(&id).await;
return Err(format!(
"{pause_error}; aria2 gid {gid} is still {status}"
@@ -5225,10 +5241,12 @@ async fn pause_download(
}
}
}
state.queue_manager.clear_aria2_allocation(&id).await;
state.queue_manager.next_aria2_control_epoch(&id).await;
log::info!("aria2 pause [{}]: gid {} paused", id, gid);
}
"complete" => {
state.queue_manager.clear_aria2_allocation(&id).await;
// Aria2 can reach complete before its terminal event updates
// Firelink's row. Treat a pause request in that narrow window
// as an idempotent completion reconciliation, not as an
@@ -5240,11 +5258,16 @@ async fn pause_download(
);
state
.queue_manager
.apply_completion_locked(&id, crate::queue::PendingOutcome::Complete)
.apply_completion_locked_with_progress(
&id,
crate::queue::PendingOutcome::Complete,
status_progress.clone(),
)
.await;
return Ok(());
}
terminal => {
state.queue_manager.clear_aria2_allocation(&id).await;
let retrying = state.queue_manager.has_aria2_retry_state(&id).await;
state.queue_manager.clear_aria2_retry_state(&id).await;
state.queue_manager.forget_aria2_gid(&id).await;
@@ -5259,13 +5282,14 @@ async fn pause_download(
// to repair it.
state.queue_manager.release_registered_id(&id).await;
use tauri::Emitter;
let _ = app_handle.emit(
"download-state",
crate::ipc::DownloadStateEvent::new(
id,
crate::ipc::DownloadStatus::Paused,
),
let mut event = crate::ipc::DownloadStateEvent::new(
id,
crate::ipc::DownloadStatus::Paused,
);
if let Some(progress) = status_progress {
event = event.with_progress(progress);
}
let _ = app_handle.emit("download-state", event);
return Ok(());
}
state.queue_manager.release_registered_id(&id).await;
@@ -5282,11 +5306,21 @@ async fn pause_download(
};
state.queue_manager.release_seed_tracking(&id);
state.queue_manager.release_permit(&id).await;
let paused_progress = aria2_download_status_snapshot(
state.aria2_port.load(std::sync::atomic::Ordering::Relaxed),
&state.aria2_secret,
gid,
)
.await
.ok()
.and_then(|(_, progress)| progress)
.or(status_progress);
use tauri::Emitter;
let _ = app_handle.emit(
"download-state",
crate::ipc::DownloadStateEvent::paused_with_seed_remaining(id, seed_remaining),
);
let mut event = crate::ipc::DownloadStateEvent::paused_with_seed_remaining(id, seed_remaining);
if let Some(progress) = paused_progress {
event = event.with_progress(progress);
}
let _ = app_handle.emit("download-state", event);
return Ok(());
}
@@ -5544,8 +5578,14 @@ async fn resume_download(
Err(error) => Some(format!("failed to resume aria2 gid {gid_clone}: {error}")),
};
if let Some(unpause_error) = unpause_error {
match verify_aria2_resume_status(aria2_port, &aria2_secret, &gid_clone).await {
Ok(status) if matches!(status.as_str(), "active" | "waiting") => {
match verify_aria2_resume_status_snapshot(
aria2_port,
&aria2_secret,
&gid_clone,
)
.await
{
Ok((status, _)) if matches!(status.as_str(), "active" | "waiting") => {
let still_current = queue_manager
.is_aria2_control_epoch_current(&id_clone, control_epoch)
.await
@@ -5578,7 +5618,7 @@ async fn resume_download(
);
return;
}
Ok(status) if status == "complete" => {
Ok((status, progress)) if status == "complete" => {
log::info!(
"aria2 resume [{}]: {} but daemon reports gid {} complete; reconciling completion",
id_clone,
@@ -5586,14 +5626,15 @@ async fn resume_download(
gid_clone
);
queue_manager
.apply_completion_locked(
.apply_completion_locked_with_progress(
&id_clone,
crate::queue::PendingOutcome::Complete,
progress,
)
.await;
return;
}
Ok(status) if status == "paused" => {
Ok((status, progress)) if status == "paused" => {
queue_manager.next_aria2_control_epoch(&id_clone).await;
queue_manager.cancel_aria2_retries(&id_clone).await;
queue_manager.release_permit(&id_clone).await;
@@ -5603,28 +5644,30 @@ async fn resume_download(
unpause_error,
gid_clone
);
let _ = app_handle_clone.emit(
"download-state",
crate::ipc::DownloadStateEvent::paused_with_error(
&id_clone,
unpause_error,
),
let mut event = crate::ipc::DownloadStateEvent::paused_with_error(
&id_clone,
unpause_error,
);
if let Some(progress) = progress {
event = event.with_progress(progress);
}
let _ = app_handle_clone.emit("download-state", event);
return;
}
Ok(status) if matches!(status.as_str(), "error" | "removed") => {
Ok((status, progress)) if matches!(status.as_str(), "error" | "removed") => {
let terminal_error = format!(
"{unpause_error}; daemon reports gid {gid_clone} as {status}"
);
queue_manager
.apply_completion_locked(
.apply_completion_locked_with_progress(
&id_clone,
crate::queue::PendingOutcome::Error(terminal_error),
progress,
)
.await;
return;
}
Ok(status) => {
Ok((status, _)) => {
// An unrecognized daemon state is not proof that
// the transfer stopped. Keep its permit and
// mapping so a later reconciliation can observe
@@ -5654,7 +5697,7 @@ async fn resume_download(
// aria2 may still report the GID as paused, complete, or
// otherwise unavailable. Verify the daemon state before
// publishing Downloading to the renderer.
let status_after_unpause = match verify_aria2_resume_status(
let (status_after_unpause, status_after_unpause_progress) = match verify_aria2_resume_status_snapshot(
aria2_port,
&aria2_secret,
&gid_clone,
@@ -5676,9 +5719,10 @@ async fn resume_download(
"active" | "waiting" => {}
"complete" => {
queue_manager
.apply_completion_locked(
.apply_completion_locked_with_progress(
&id_clone,
crate::queue::PendingOutcome::Complete,
status_after_unpause_progress,
)
.await;
return;
@@ -5688,9 +5732,10 @@ async fn resume_download(
"aria2 resume left gid {gid_clone} in terminal state {status_after_unpause}"
);
queue_manager
.apply_completion_locked(
.apply_completion_locked_with_progress(
&id_clone,
crate::queue::PendingOutcome::Error(terminal_error),
status_after_unpause_progress,
)
.await;
return;
@@ -5706,10 +5751,14 @@ async fn resume_download(
error,
gid_clone
);
let _ = app_handle_clone.emit(
"download-state",
crate::ipc::DownloadStateEvent::paused_with_error(&id_clone, error),
let mut event = crate::ipc::DownloadStateEvent::paused_with_error(
&id_clone,
error,
);
if let Some(progress) = status_after_unpause_progress {
event = event.with_progress(progress);
}
let _ = app_handle_clone.emit("download-state", event);
return;
}
other => {
@@ -5973,6 +6022,10 @@ async fn remove_download(
.queue_manager
.release_permit_for_generation(&id, media_lifecycle_generation)
.await;
state
.queue_manager
.clear_aria2_allocation_for_epoch(&id, removal_epoch)
.await;
return Err("download lifecycle changed while waiting for aria2 dispatch".to_string());
}
if let Some(late_gid) = state.queue_manager.aria2_gid_for_download(&id) {
@@ -6395,17 +6448,78 @@ 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 aria2_download_status_snapshot(
port: u16,
secret: &str,
gid: &str,
) -> Result<(String, Option<crate::ipc::DownloadStateProgress>), String> {
let result = rpc_call(
port,
secret,
"aria2.tellStatus",
serde_json::json!([gid, ["status", "completedLength", "totalLength"]]),
)
.await
.map_err(|error| format!("failed to query aria2 gid {gid}: {error}"))?;
let status = result
.get("status")
.and_then(serde_json::Value::as_str)
.map(str::to_string)
.ok_or_else(|| format!("aria2.tellStatus returned no status for gid {gid}"))?;
let progress = aria2_download_state_progress(Some(&result));
Ok((status, progress))
}
fn aria2_u64(value: Option<&serde_json::Value>) -> Option<u64> {
value.and_then(|value| {
value
.as_str()
.and_then(|value| value.parse::<u64>().ok())
.or_else(|| value.as_u64())
})
}
fn aria2_download_state_progress(
status: Option<&serde_json::Value>,
) -> Option<crate::ipc::DownloadStateProgress> {
let status = status?;
let downloaded = aria2_u64(status.get("completedLength"));
let total = aria2_u64(status.get("totalLength"));
if downloaded.is_none() && total.is_none() {
return None;
}
let is_complete = status.get("status").and_then(|value| value.as_str()) == Some("complete");
let fraction = if is_complete {
1.0
} else if let Some(total) = total.filter(|total| *total > 0) {
(downloaded.unwrap_or_default().min(total) as f64 / total as f64).clamp(0.0, 1.0)
} else {
0.0
};
let total_bytes = total.filter(|value| *value > 0).map(|value| value as f64);
Some(crate::ipc::DownloadStateProgress {
fraction,
downloaded_bytes: downloaded.map(|value| value as f64),
total_bytes,
total_is_estimate: total_bytes.map(|_| false),
})
}
/// Verify a retained-GID resume against Aria2's actual state. `unpause` can
/// return before a paused GID becomes observable as active, and a transient
/// tellStatus failure must not turn a successful resume into a false failure.
/// Retry only that ambiguous observation; terminal and active states return
/// immediately and remain authoritative.
async fn verify_aria2_resume_status(port: u16, secret: &str, gid: &str) -> Result<String, String> {
async fn verify_aria2_resume_status_snapshot(
port: u16,
secret: &str,
gid: &str,
) -> Result<(String, Option<crate::ipc::DownloadStateProgress>), String> {
let mut last_observation = Err(format!("aria2 resume status for gid {gid} was not observed"));
for attempt in 0..4u32 {
last_observation = match aria2_download_status(port, secret, gid).await {
Ok(status) if status != "paused" => return Ok(status),
Ok(status) => Ok(status),
last_observation = match aria2_download_status_snapshot(port, secret, gid).await {
Ok((status, progress)) if status != "paused" => return Ok((status, progress)),
Ok((status, progress)) => Ok((status, progress)),
Err(error) => Err(error),
};
if attempt < 3 {
@@ -6426,7 +6540,7 @@ async fn reconcile_aria2_downloads(app_handle: &tauri::AppHandle) {
port,
&secret,
"aria2.tellStatus",
serde_json::json!([gid, ["status", "seeder", "errorCode", "errorMessage"]]),
serde_json::json!([gid, ["status", "seeder", "errorCode", "errorMessage", "completedLength", "totalLength"]]),
)
.await
{
@@ -6502,7 +6616,11 @@ async fn reconcile_aria2_downloads(app_handle: &tauri::AppHandle) {
};
if let Some(outcome) = outcome {
state.queue_manager.handle_aria2_event(&gid, outcome).await;
let progress = aria2_download_state_progress(Some(&status));
state
.queue_manager
.handle_aria2_event_with_progress(&gid, outcome, progress)
.await;
}
}
}
@@ -7605,6 +7723,10 @@ async fn cancel_enqueue_generation(
.queue_manager
.cancel_enqueue_generation(&id, generation)
.await;
state
.queue_manager
.clear_aria2_allocation_for_lifecycle_generation(&id, generation)
.await;
Ok(())
}
@@ -7643,6 +7765,19 @@ async fn enqueue_many(
});
continue;
}
if let Err(error) = preflight_download_destination_access(
&app_handle,
&id,
&item.destination,
) {
results.push(crate::ipc::EnqueueResult {
id,
success: false,
filename: None,
error: Some(error),
});
continue;
}
item.filename = crate::download_ownership::canonical_download_filename(&item.filename);
let filename = item.filename.clone();
let lifecycle_generation = match enqueue_lifecycle_generation(&item) {
@@ -11607,6 +11742,7 @@ mod tests {
normalize_media_connections,
validate_enqueue_url, validate_enqueue_uris, validate_keychain_grant_request_id,
aria2_gid_not_found,
aria2_download_state_progress, preflight_download_destination_access,
retained_torrent_id_from_persisted_record,
retained_torrent_info_hash_from_persisted_record,
merge_durable_torrent_telemetry, torrent_identity_magnet, torrent_move_path_pair,
@@ -11635,6 +11771,36 @@ mod tests {
assert!(!aria2_gid_not_found("aria2 error code 3: Resource not found"));
}
#[test]
fn terminal_aria2_status_preserves_exact_progress_snapshot() {
let snapshot = aria2_download_state_progress(Some(&json!({
"status": "error",
"completedLength": "420",
"totalLength": "1000"
})))
.expect("length fields should produce a progress snapshot");
assert!((snapshot.fraction - 0.42).abs() < f64::EPSILON);
assert_eq!(snapshot.downloaded_bytes, Some(420.0));
assert_eq!(snapshot.total_bytes, Some(1000.0));
assert_eq!(snapshot.total_is_estimate, Some(false));
}
#[test]
fn destination_preflight_returns_the_retryable_marker_before_admission() {
let app = tauri::test::mock_builder()
.build(tauri::test::mock_context(tauri::test::noop_assets()))
.expect("mock app");
let error = preflight_download_destination_access(
app.handle(),
"preflight-only",
"relative/not-approved",
)
.expect_err("an unapproved destination must not enter admission");
assert!(error.starts_with("destination access retryable:"));
}
#[test]
fn renderer_download_snapshots_cannot_lower_native_torrent_totals() {
let existing = vec![
@@ -15699,42 +15865,95 @@ pub fn run() {
if let Some(event) = params.first().and_then(|p| p.as_object()) {
if let Some(gid) = event.get("gid").and_then(|g| g.as_str()) {
let state = app_handle_bg.state::<AppState>();
let mut progress = None;
let outcome = match method {
"aria2.onDownloadComplete" => Some(crate::queue::PendingOutcome::Complete),
"aria2.onBtDownloadComplete" => Some(crate::queue::PendingOutcome::Seeding),
"aria2.onDownloadStart" => {
let downloaded_bytes = aria2_download_status_snapshot(
state.aria2_port.load(std::sync::atomic::Ordering::Relaxed),
&state.aria2_secret,
gid,
)
.await
.ok()
.and_then(|(_, progress)| progress)
.and_then(|progress| progress.downloaded_bytes)
.filter(|value| value.is_finite() && *value > 0.0)
.map(|value| value as u64)
.unwrap_or_default();
state
.queue_manager
.handle_aria2_download_start(gid, downloaded_bytes)
.await;
None
}
"aria2.onDownloadComplete" => {
progress = aria2_download_state_progress(
rpc_call(
state.aria2_port.load(std::sync::atomic::Ordering::Relaxed),
&state.aria2_secret,
"aria2.tellStatus",
serde_json::json!([gid, ["status", "completedLength", "totalLength"]]),
)
.await
.ok()
.as_ref(),
);
Some(crate::queue::PendingOutcome::Complete)
}
"aria2.onBtDownloadComplete" => {
progress = aria2_download_state_progress(
rpc_call(
state.aria2_port.load(std::sync::atomic::Ordering::Relaxed),
&state.aria2_secret,
"aria2.tellStatus",
serde_json::json!([gid, ["status", "completedLength", "totalLength"]]),
)
.await
.ok()
.as_ref(),
);
Some(crate::queue::PendingOutcome::Seeding)
}
"aria2.onDownloadError" => {
let mut msg = event.get("error_message").and_then(|m| m.as_str()).unwrap_or("aria2 download error").to_string();
let aria2_port = state.aria2_port.load(std::sync::atomic::Ordering::Relaxed);
let aria2_secret = state.aria2_secret.clone();
if let Ok(status) = rpc_call(aria2_port, &aria2_secret, "aria2.tellStatus", serde_json::json!([gid, ["errorCode", "errorMessage"]])).await {
let err_msg = status
.get("errorMessage")
.and_then(|m| m.as_str())
.filter(|m| !m.is_empty());
let err_code = status
.get("errorCode")
.and_then(|m| m.as_str())
.filter(|m| !m.is_empty());
match (err_code, err_msg) {
(Some(code), Some(message)) => {
msg = format!("aria2 error code {code}: {message}");
let status = rpc_call(
state.aria2_port.load(std::sync::atomic::Ordering::Relaxed),
&state.aria2_secret,
"aria2.tellStatus",
serde_json::json!([gid, ["errorCode", "errorMessage", "completedLength", "totalLength", "status"]]),
)
.await
.ok();
if let Some(status) = status.as_ref() {
progress = aria2_download_state_progress(Some(status));
let err_msg = status
.get("errorMessage")
.and_then(|m| m.as_str())
.filter(|m| !m.is_empty());
let err_code = status
.get("errorCode")
.and_then(|m| m.as_str())
.filter(|m| !m.is_empty());
match (err_code, err_msg) {
(Some(code), Some(message)) => {
msg = format!("aria2 error code {code}: {message}");
}
(Some(code), None) => {
msg = format!("aria2 error code {code}: {msg}");
}
(None, Some(message)) => {
msg = message.to_string();
}
(None, None) => {}
}
}
(Some(code), None) => {
msg = format!("aria2 error code {code}: {msg}");
}
(None, Some(message)) => {
msg = message.to_string();
}
(None, None) => {}
}
}
Some(crate::queue::PendingOutcome::Error(msg))
}
_ => None,
};
if let Some(outcome) = outcome {
Arc::clone(&state.queue_manager)
.handle_aria2_event(gid, outcome)
.handle_aria2_event_with_progress(gid, outcome, progress)
.await;
}
}
@@ -16129,6 +16348,22 @@ pub fn run() {
None
};
// tellActive is the poller's confirmed native
// progress path. If the WebSocket start
// notification was lost, this is sufficient
// to end the allocation phase for the same
// mapped lifecycle.
poll_mgr
.complete_aria2_allocation_for_gid(gid, completed)
.await;
if !poll_mgr.is_current_aria2_gid_mapping(gid, &mapping)
|| !poll_mgr
.is_aria2_control_epoch_current(&id, control_epoch)
.await
{
continue;
}
use tauri::Emitter;
let _ = app_handle_poll.emit("download-progress", DownloadProgressEvent {
id: id.clone(),
@@ -16206,7 +16441,7 @@ pub fn run() {
poll_port.load(std::sync::atomic::Ordering::Relaxed),
&poll_secret,
"aria2.tellStatus",
serde_json::json!([gid, ["status", "seeder", "errorCode", "errorMessage"]]),
serde_json::json!([gid, ["status", "seeder", "errorCode", "errorMessage", "completedLength", "totalLength"]]),
)
.await
{
@@ -16389,6 +16624,7 @@ pub fn run() {
_ => None,
};
if let Some(outcome) = outcome {
let progress = aria2_download_state_progress(Some(&status));
let terminal_error = match &outcome {
crate::queue::PendingOutcome::Error(error) => Some(error.as_str()),
_ => None,
@@ -16409,7 +16645,9 @@ pub fn run() {
.and_then(crate::retry::aria2_error_code)
.unwrap_or_else(|| "none".to_string())
);
poll_mgr.handle_aria2_event(&gid, outcome).await;
poll_mgr
.handle_aria2_event_with_progress(&gid, outcome, progress)
.await;
}
}
observations.retain(|id, _| seen_ids.contains(id));
src-tauri/src/queue.rs
+695 -29
View File
@@ -1,5 +1,8 @@
use base64::Engine as _;
use crate::ipc::{DownloadStateEvent, DownloadStatus, QueueDirection};
use crate::ipc::{
DownloadAllocationEvent, DownloadStateEvent, DownloadStateProgress, DownloadStatus,
QueueDirection,
};
use crate::power::PowerManager;
use crate::retry::{
aria2_error_code, backoff_and_emit, is_aria2_name_resolution_error,
@@ -21,6 +24,7 @@ use ts_rs::TS;
/// Default capacity when no setting is read yet.
pub const DEFAULT_MAX_CONCURRENT: usize = 3;
pub const MAX_QUEUE_CONCURRENT: usize = 12;
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;
@@ -868,6 +872,21 @@ pub enum PendingOutcome {
Error(String),
}
#[derive(Debug, Clone)]
pub struct PendingAria2Outcome {
pub outcome: PendingOutcome,
pub progress: Option<DownloadStateProgress>,
}
impl PendingAria2Outcome {
pub fn new(outcome: PendingOutcome) -> Self {
Self {
outcome,
progress: None,
}
}
}
/// Result of recycling an aria2 transfer's connections. A refresh can race
/// with daemon completion or leave the transfer paused after an ambiguous
/// unpause failure, so callers must handle the verified daemon outcome.
@@ -1176,7 +1195,14 @@ pub struct QueueManager<R: tauri::Runtime = tauri::Wry> {
/// gid -> buffered (id_placeholder, outcome) for completions that arrived
/// before the gid was stored. Drained by `remember_gid`.
pub pending_completion: Arc<Mutex<HashMap<String, (String, PendingOutcome)>>>,
pub pending_completion: Arc<Mutex<HashMap<String, (String, PendingAria2Outcome)>>>,
/// Aria2 can emit onDownloadStart before addUri's response has been
/// mapped to the Firelink download. Buffer that start marker until the
/// current GID mapping is installed.
pending_download_starts: Arc<Mutex<HashSet<String>>>,
/// Current Aria2 lifecycles whose files are expected to be preallocated.
/// The generation fences late start/clear events from an older GID.
aria2_allocation_pending: Mutex<HashMap<String, (u64, u64)>>,
/// download id -> spawn payload for aria2 transient-error re-addUri retries.
aria2_payloads: Mutex<HashMap<String, SpawnPayload>>,
@@ -1281,6 +1307,8 @@ impl<R: tauri::Runtime> QueueManager<R> {
torrent_move_cancellations: StdMutex::new(HashSet::new()),
aria2_gids: Arc::new(std::sync::RwLock::new(HashMap::new())),
pending_completion: Arc::new(Mutex::new(HashMap::new())),
pending_download_starts: Arc::new(Mutex::new(HashSet::new())),
aria2_allocation_pending: Mutex::new(HashMap::new()),
aria2_payloads: Mutex::new(HashMap::new()),
aria2_connection_options: Mutex::new(HashMap::new()),
aria2_dispatch_inflight: Mutex::new(HashMap::new()),
@@ -3808,6 +3836,11 @@ impl<R: tauri::Runtime> QueueManager<R> {
.await;
}
}
// Unknown start notifications belong to the current Aria2 daemon
// session. A reconnect/restart invalidates that association; the
// poller will provide a fresh positive-progress fallback after the
// next GID mapping instead of letting an old GID clear a new phase.
self.pending_download_starts.lock().await.clear();
}
/// Number of un-acquired permits currently in the semaphore pool.
@@ -3822,6 +3855,159 @@ impl<R: tauri::Runtime> QueueManager<R> {
.emit("download-state", DownloadStateEvent::new(id, status));
}
fn emit_allocation_event(&self, id: &str, pending: bool, lifecycle_generation: u64) {
use tauri::Emitter;
let _ = self.app_handle.emit(
"download-allocation",
DownloadAllocationEvent {
id: id.to_string(),
pending,
lifecycle_generation: lifecycle_generation.to_string(),
},
);
}
pub fn aria2_allocation_phase_eligible(payload: &SpawnPayload) -> bool {
if payload.is_media || payload.torrent_verify_only {
return false;
}
if !payload.is_torrent {
return true;
}
normalize_torrent_file_allocation(payload.torrent_file_allocation.as_deref())
.is_ok_and(|allocation| allocation != "none")
}
async fn begin_aria2_allocation(
&self,
id: &str,
control_epoch: u64,
lifecycle_generation: u64,
payload: &SpawnPayload,
) {
if !Self::aria2_allocation_phase_eligible(payload) {
return;
}
self.aria2_allocation_pending
.lock()
.await
.insert(id.to_string(), (control_epoch, lifecycle_generation));
self.emit_allocation_event(id, true, lifecycle_generation);
}
pub async fn clear_aria2_allocation_for_lifecycle_generation(
&self,
id: &str,
lifecycle_generation: u64,
) {
let cleared_generation = {
let mut pending = self.aria2_allocation_pending.lock().await;
let Some((_, pending_generation)) = pending.get(id).copied() else {
return;
};
if pending_generation != lifecycle_generation {
return;
}
pending.remove(id).map(|(_, generation)| generation)
};
if let Some(cleared_generation) = cleared_generation {
self.emit_allocation_event(id, false, cleared_generation);
}
}
pub async fn clear_aria2_allocation_for_epoch(&self, id: &str, control_epoch: u64) {
let lifecycle_generation = {
let mut pending = self.aria2_allocation_pending.lock().await;
let Some((pending_epoch, lifecycle_generation)) = pending.get(id).copied() else {
return;
};
if pending_epoch != control_epoch {
return;
}
pending.remove(id);
lifecycle_generation
};
self.emit_allocation_event(id, false, lifecycle_generation);
}
pub async fn clear_aria2_allocation(&self, id: &str) {
let lifecycle_generation = self
.aria2_allocation_pending
.lock()
.await
.remove(id)
.map(|(_, lifecycle_generation)| lifecycle_generation);
if let Some(lifecycle_generation) = lifecycle_generation {
self.emit_allocation_event(id, false, lifecycle_generation);
}
}
pub async fn complete_aria2_allocation_for_gid(&self, gid: &str, downloaded_bytes: u64) {
// Aria2 reports an active GID with completedLength=0 while it is still
// creating preallocated files. That observation is not native
// transfer progress and must not hide the allocation phase.
if downloaded_bytes == 0 {
return;
}
let mapping = {
let _gid_state = self.aria2_gid_state.lock().await;
if self.is_aria2_gid_ignored_locked(gid).await {
return;
}
let Some(mapping) = self.aria2_gid_mapping(gid) else {
let mut starts = self.pending_download_starts.lock().await;
if starts.len() < MAX_PENDING_DOWNLOAD_STARTS {
starts.insert(gid.to_string());
}
return;
};
mapping
};
if !self
.is_current_aria2_gid_mapping(gid, &mapping)
|| !self
.is_aria2_control_epoch_current(&mapping.id, mapping.epoch)
.await
{
return;
}
self.clear_aria2_allocation_for_epoch(&mapping.id, mapping.epoch)
.await;
}
pub async fn handle_aria2_download_start(&self, gid: &str, downloaded_bytes: u64) {
// Aria2 can publish onDownloadStart before it finishes creating
// preallocated files. A zero-byte start is therefore not native
// transfer progress and must leave the allocation phase visible.
if downloaded_bytes == 0 {
return;
}
let mapping = {
let _gid_state = self.aria2_gid_state.lock().await;
if self.is_aria2_gid_ignored_locked(gid).await {
return;
}
let Some(mapping) = self.aria2_gid_mapping(gid) else {
let mut starts = self.pending_download_starts.lock().await;
if starts.len() < MAX_PENDING_DOWNLOAD_STARTS {
starts.insert(gid.to_string());
}
return;
};
mapping
};
if !self
.is_current_aria2_gid_mapping(gid, &mapping)
|| !self
.is_aria2_control_epoch_current(&mapping.id, mapping.epoch)
.await
{
return;
}
self.clear_aria2_allocation_for_epoch(&mapping.id, mapping.epoch)
.await;
}
/// Resize the global concurrency limit. Grow adds permits immediately;
/// shrink records a retirement debt honored lazily by the dispatcher.
pub fn set_capacity(&self, new_target: usize) {
@@ -3947,6 +4133,17 @@ impl<R: tauri::Runtime> QueueManager<R> {
};
if let Some(epoch) = aria2_lifecycle_epoch {
self.begin_aria2_dispatch(&id, epoch).await;
// Register the native allocation phase before releasing the
// lifecycle lock. A pause/remove can otherwise invalidate this
// dispatch in the gap and leave a stale pending marker behind
// when the asynchronous addUri call starts.
self.begin_aria2_allocation(
&id,
epoch,
lifecycle_generation,
&task.payload,
)
.await;
}
drop(control_guard);
@@ -4008,6 +4205,8 @@ impl<R: tauri::Runtime> QueueManager<R> {
self.clear_aria2_retry_state(&id).await;
self.release_permit(&id).await;
}
self.clear_aria2_allocation_for_epoch(&id, lifecycle_epoch)
.await;
return;
}
// A queued task is not a live transfer until aria2 has
@@ -4074,10 +4273,12 @@ impl<R: tauri::Runtime> QueueManager<R> {
}
}
if let Some(outcome) = buffered_outcome {
self.handle_aria2_event(&gid, outcome).await;
self.handle_aria2_pending_event(&gid, outcome).await;
}
}
Err(error) => {
self.clear_aria2_allocation_for_epoch(&id, lifecycle_epoch)
.await;
let _control_guard = self.acquire_aria2_control(&id).await;
let current_lifecycle = self
.is_aria2_control_epoch_current(&id, lifecycle_epoch)
@@ -4161,24 +4362,46 @@ impl<R: tauri::Runtime> QueueManager<R> {
}
fn emit_failed(&self, id: &str, error: String) {
use tauri::Emitter;
let _ = self
.app_handle
.emit("download-state", DownloadStateEvent::failed(id, error));
self.emit_failed_with_progress(id, error, None);
}
fn emit_paused_with_error(&self, id: &str, error: String) {
fn emit_failed_with_progress(
&self,
id: &str,
error: String,
progress: Option<DownloadStateProgress>,
) {
use tauri::Emitter;
let mut event = DownloadStateEvent::failed(id, error);
if let Some(progress) = progress {
event = event.with_progress(progress);
}
let _ = self
.app_handle
.emit("download-state", event);
}
fn emit_paused_with_error_and_progress(
&self,
id: &str,
error: String,
progress: Option<DownloadStateProgress>,
) {
use tauri::Emitter;
let mut event = DownloadStateEvent::paused_with_error(id, error);
if let Some(progress) = progress {
event = event.with_progress(progress);
}
let _ = self.app_handle.emit(
"download-state",
DownloadStateEvent::paused_with_error(id, error),
event,
);
}
/// Store gid -> id and return any buffered terminal event for the caller
/// to reconcile against the correct event path. In particular, buffered
/// errors must still pass through transient retry classification.
pub async fn remember_gid(&self, id: String, gid: String) -> Option<PendingOutcome> {
pub async fn remember_gid(&self, id: String, gid: String) -> Option<PendingAria2Outcome> {
let epoch = self.current_aria2_control_epoch(&id).await;
let buffered_outcome = {
let _gid_state = self.aria2_gid_state.lock().await;
@@ -4217,6 +4440,14 @@ impl<R: tauri::Runtime> QueueManager<R> {
}
buffered.remove(&gid).map(|(_buf_id, outcome)| outcome)
};
let start_buffered = self
.pending_download_starts
.lock()
.await
.remove(&gid);
if start_buffered {
self.clear_aria2_allocation_for_epoch(&id, epoch).await;
}
let retry_strike = self.aria2_retry_strike(&id).await;
log::info!(
"aria2 gid transition [stage=gid_transition id={} gid={} epoch={} retry_strike={} action=mapped]",
@@ -4261,6 +4492,17 @@ impl<R: tauri::Runtime> QueueManager<R> {
/// and lets commands reconcile an Aria2 terminal status without releasing
/// the lock first.
pub(crate) async fn apply_completion_locked(&self, id: &str, outcome: PendingOutcome) {
self.apply_completion_locked_with_progress(id, outcome, None)
.await;
}
pub(crate) async fn apply_completion_locked_with_progress(
&self,
id: &str,
outcome: PendingOutcome,
progress: Option<DownloadStateProgress>,
) {
self.clear_aria2_allocation(id).await;
if matches!(&outcome, PendingOutcome::Complete) {
self.capture_torrent_verification_evidence(id).await;
}
@@ -4385,7 +4627,12 @@ impl<R: tauri::Runtime> QueueManager<R> {
_ => DownloadStatus::Completed,
}
};
self.emit_state(id, restored_status);
let mut event = DownloadStateEvent::new(id, restored_status);
if let Some(progress) = progress {
event = event.with_progress(progress);
}
use tauri::Emitter;
let _ = self.app_handle.emit("download-state", event);
}
PendingOutcome::Error(error) => {
self.forget_torrent_telemetry(id).await;
@@ -4452,7 +4699,7 @@ impl<R: tauri::Runtime> QueueManager<R> {
self.release_registered_id(id).await;
self.release_permit(id).await;
if verification_only {
self.emit_paused_with_error(id, error);
self.emit_paused_with_error_and_progress(id, error, progress);
} else {
let aria2_code = aria2_error_code(&error).unwrap_or_else(|| "none".to_string());
log::error!(
@@ -4466,7 +4713,7 @@ impl<R: tauri::Runtime> QueueManager<R> {
network_error_class(&error),
aria2_code
);
self.emit_failed(id, error);
self.emit_failed_with_progress(id, error, progress);
}
}
PendingOutcome::Seeding => unreachable!("seeding outcomes are normalized before terminal cleanup"),
@@ -4556,6 +4803,7 @@ impl<R: tauri::Runtime> QueueManager<R> {
async fn ignore_aria2_gid_locked(&self, gid: &str) {
const MAX_IGNORED_GIDS: usize = 1024;
self.pending_download_starts.lock().await.remove(gid);
let mut ignored = self.aria2_ignored_gids.lock().await;
if !ignored.iter().any(|known| known == gid) {
ignored.push_back(gid.to_string());
@@ -4788,7 +5036,20 @@ impl<R: tauri::Runtime> QueueManager<R> {
let payload = self.aria2_payloads.lock().await.get(id).cloned();
let recreation = if let Some(payload) = payload.as_ref() {
self.spawner.recreate_uri(id, gid, payload).await?
let lifecycle_generation = self
.registered_lifecycle_generation(id)
.await
.unwrap_or_default();
self.begin_aria2_allocation(id, observed_epoch, lifecycle_generation, payload)
.await;
match self.spawner.recreate_uri(id, gid, payload).await {
Ok(outcome) => outcome,
Err(error) => {
self.clear_aria2_allocation_for_epoch(id, observed_epoch)
.await;
return Err(error);
}
}
} else {
// Older persisted rows may briefly reach recovery before their
// payload has been rebuilt. Keep the current lifecycle intact and
@@ -4798,6 +5059,8 @@ impl<R: tauri::Runtime> QueueManager<R> {
if let Aria2RecreateOutcome::NewGid(new_gid) = recreation {
if new_gid.trim().is_empty() || new_gid == gid {
self.clear_aria2_allocation_for_epoch(id, observed_epoch)
.await;
return Err(format!(
"aria2 connection recovery returned an invalid replacement gid for {gid}"
));
@@ -4808,6 +5071,8 @@ impl<R: tauri::Runtime> QueueManager<R> {
&& self.is_aria2_control_epoch_current(id, observed_epoch).await
&& self.aria2_gid_for_download(id).as_deref() == Some(gid);
if !still_current {
self.clear_aria2_allocation_for_epoch(id, observed_epoch)
.await;
self.ignore_aria2_gid(&new_gid).await;
drop(_control_guard);
self.remove_stale_aria2_gid(id, &new_gid).await;
@@ -4828,15 +5093,24 @@ impl<R: tauri::Runtime> QueueManager<R> {
);
drop(_control_guard);
if let Some(outcome) = buffered_outcome {
self.handle_aria2_event(&new_gid, outcome).await;
self.handle_aria2_pending_event(&new_gid, outcome).await;
}
return Ok(());
}
let outcome = match recreation {
Aria2RecreateOutcome::Complete => Aria2RefreshOutcome::Complete,
Aria2RecreateOutcome::Refresh => self.spawner.refresh_uri(gid).await?,
Aria2RecreateOutcome::Refresh => match self.spawner.refresh_uri(gid).await {
Ok(outcome) => outcome,
Err(error) => {
self.clear_aria2_allocation_for_epoch(id, observed_epoch)
.await;
return Err(error);
}
},
Aria2RecreateOutcome::Unavailable(error) => {
self.clear_aria2_allocation_for_epoch(id, observed_epoch)
.await;
let still_current = self.is_registered(id).await
&& self.has_active_permit(id).await
&& !self.is_aria2_retry_cancelled(id).await
@@ -4907,6 +5181,7 @@ impl<R: tauri::Runtime> QueueManager<R> {
/// 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> {
self.clear_aria2_allocation(id).await;
let _gid_state = self.aria2_gid_state.lock().await;
let removed = {
let mut gids = self.aria2_gids.write().unwrap();
@@ -4944,10 +5219,12 @@ impl<R: tauri::Runtime> QueueManager<R> {
self: &Arc<Self>,
gid: String,
error: String,
progress: Option<DownloadStateProgress>,
) -> Pin<Box<dyn Future<Output = ()> + Send + 'static>> {
let this = Arc::clone(self);
Box::pin(async move {
this.handle_aria2_download_error_inner(&gid, error).await;
this.handle_aria2_download_error_inner(&gid, error, progress)
.await;
})
}
@@ -4957,8 +5234,8 @@ impl<R: tauri::Runtime> QueueManager<R> {
async fn map_or_buffer_aria2_event(
&self,
gid: &str,
outcome: PendingOutcome,
) -> Option<(Aria2GidMapping, PendingOutcome)> {
outcome: PendingAria2Outcome,
) -> Option<(Aria2GidMapping, PendingAria2Outcome)> {
let _gid_state = self.aria2_gid_state.lock().await;
if self.is_aria2_gid_ignored_locked(gid).await {
return None;
@@ -4977,13 +5254,31 @@ impl<R: tauri::Runtime> QueueManager<R> {
None
}
async fn handle_aria2_download_error_inner(self: &Arc<Self>, gid: &str, error: String) {
let Some((mapping, PendingOutcome::Error(error))) = self
.map_or_buffer_aria2_event(gid, PendingOutcome::Error(error))
async fn handle_aria2_download_error_inner(
self: &Arc<Self>,
gid: &str,
error: String,
progress: Option<DownloadStateProgress>,
) {
let Some((mapping, pending)) = self
.map_or_buffer_aria2_event(
gid,
PendingAria2Outcome {
outcome: PendingOutcome::Error(error),
progress,
},
)
.await
else {
return;
};
let PendingAria2Outcome {
outcome: PendingOutcome::Error(error),
progress,
} = pending
else {
return;
};
let _control_guard = self.acquire_aria2_control(&mapping.id).await;
let current_mapping = {
@@ -5000,6 +5295,11 @@ impl<R: tauri::Runtime> QueueManager<R> {
return;
}
let id = mapping.id;
// The failed GID is no longer allocating. Keep the native phase
// marker scoped to an actual replacement addUri rather than showing
// "Allocating files" throughout retry backoff or a failed pause.
self.clear_aria2_allocation_for_epoch(&id, mapping.epoch)
.await;
if self.aria2_retry_cancelled.lock().await.contains(&id) {
log::info!(
"aria2 retry cancellation [{}]: ignoring error for gid {} during removal",
@@ -5028,7 +5328,11 @@ impl<R: tauri::Runtime> QueueManager<R> {
let payload = self.aria2_payloads.lock().await.get(&id).cloned();
if payload.is_none() {
self.apply_completion_locked(&id, PendingOutcome::Error(error))
self.apply_completion_locked_with_progress(
&id,
PendingOutcome::Error(error),
progress,
)
.await;
return;
}
@@ -5083,7 +5387,11 @@ impl<R: tauri::Runtime> QueueManager<R> {
// automatic retries, while every later failure follows the normal
// retry budget and never switches back to the first strategy.
if retry_action == Aria2RetryAction::Terminal {
self.apply_completion_locked(&id, PendingOutcome::Error(error))
self.apply_completion_locked_with_progress(
&id,
PendingOutcome::Error(error),
progress,
)
.await;
return;
}
@@ -5133,6 +5441,7 @@ impl<R: tauri::Runtime> QueueManager<R> {
let this = Arc::clone(self);
let id_for_task = id.clone();
let error_for_emit = error.clone();
let progress_for_retry = progress.clone();
tauri::async_runtime::spawn(async move {
let retry_cancel = async {
loop {
@@ -5191,6 +5500,17 @@ impl<R: tauri::Runtime> QueueManager<R> {
return;
}
let lifecycle_generation = this
.registered_lifecycle_generation(&id_for_task)
.await
.unwrap_or_default();
this.begin_aria2_allocation(
&id_for_task,
retry_epoch,
lifecycle_generation,
&current_payload,
)
.await;
match this
.spawner
.add_uri(&id_for_task, &current_payload)
@@ -5205,6 +5525,8 @@ impl<R: tauri::Runtime> QueueManager<R> {
|| this.aria2_gid_for_download(&id_for_task).as_deref()
!= Some(retry_gid.as_str());
if stale {
this.clear_aria2_allocation_for_epoch(&id_for_task, retry_epoch)
.await;
drop(control_guard);
if let Err(error) = this.spawner.remove_uri(&new_gid).await {
log::error!(
@@ -5249,18 +5571,22 @@ impl<R: tauri::Runtime> QueueManager<R> {
this.aria2_retrying_gids.lock().await.remove(&retry_gid);
drop(control_guard);
if let Some(outcome) = buffered_outcome {
this.handle_aria2_event(&new_gid_for_event, outcome).await;
this.handle_aria2_pending_event(&new_gid_for_event, outcome)
.await;
}
}
Err(retry_error) => {
this.clear_aria2_allocation_for_epoch(&id_for_task, retry_epoch)
.await;
let stale = this.is_aria2_retry_cancelled(&id_for_task).await
|| !this
.is_aria2_control_epoch_current(&id_for_task, retry_epoch)
.await;
if !stale {
this.apply_completion_locked(
this.apply_completion_locked_with_progress(
&id_for_task,
PendingOutcome::Error(retry_error),
progress_for_retry,
)
.await;
}
@@ -5275,14 +5601,44 @@ impl<R: tauri::Runtime> QueueManager<R> {
/// Entry point for the aria2 WS poller. Resolves gid -> id; if not yet
/// stored, buffers the outcome for reconciliation by remember_gid.
pub async fn handle_aria2_event(self: &Arc<Self>, gid: &str, outcome: PendingOutcome) {
self.handle_aria2_pending_event(gid, PendingAria2Outcome::new(outcome))
.await;
}
pub async fn handle_aria2_event_with_progress(
self: &Arc<Self>,
gid: &str,
outcome: PendingOutcome,
progress: Option<DownloadStateProgress>,
) {
self.handle_aria2_pending_event(
gid,
PendingAria2Outcome { outcome, progress },
)
.await;
}
async fn handle_aria2_pending_event(
self: &Arc<Self>,
gid: &str,
pending: PendingAria2Outcome,
) {
let PendingAria2Outcome { outcome, progress } = pending;
if let PendingOutcome::Error(error) = outcome {
self.handle_aria2_download_error(gid.to_string(), error)
self.handle_aria2_download_error(gid.to_string(), error, progress)
.await;
return;
}
let Some((mapping, outcome)) = self.map_or_buffer_aria2_event(gid, outcome).await else {
let Some((mapping, pending)) = self
.map_or_buffer_aria2_event(
gid,
PendingAria2Outcome { outcome, progress },
)
.await
else {
return;
};
let PendingAria2Outcome { outcome, progress } = pending;
let _control_guard = self.acquire_aria2_control(&mapping.id).await;
if self.aria2_retrying_gids.lock().await.contains(gid) {
@@ -5301,7 +5657,8 @@ impl<R: tauri::Runtime> QueueManager<R> {
{
return;
}
self.apply_completion_locked(&mapping.id, outcome).await;
self.apply_completion_locked_with_progress(&mapping.id, outcome, progress)
.await;
}
/// Reorder a pending task up or down. Returns the new pending order.
@@ -8206,6 +8563,315 @@ mod tests {
}
}
struct BlockingSpawner {
started: Arc<tokio::sync::Notify>,
release: Arc<tokio::sync::Notify>,
}
#[async_trait::async_trait]
impl SidecarSpawner for BlockingSpawner {
async fn add_uri(&self, _id: &str, _payload: &SpawnPayload) -> Result<String, String> {
self.started.notify_one();
self.release.notified().await;
Ok("blocking-gid".to_string())
}
async fn remove_uri(&self, _gid: &str) -> Result<(), String> {
Ok(())
}
async fn run_media(
&self,
_id: &str,
_payload: &SpawnPayload,
_lifecycle_generation: u64,
) -> Result<(), String> {
Ok(())
}
}
fn allocation_pending_epoch<R: tauri::Runtime>(
manager: &QueueManager<R>,
id: &str,
) -> Option<(u64, u64)> {
manager
.aria2_allocation_pending
.try_lock()
.ok()
.and_then(|pending| pending.get(id).copied())
}
#[test]
fn aria2_allocation_eligibility_matches_download_type_and_torrent_policy() {
assert!(QueueManager::<tauri::Wry>::aria2_allocation_phase_eligible(
&SpawnPayload::default()
));
assert!(!QueueManager::<tauri::Wry>::aria2_allocation_phase_eligible(
&SpawnPayload {
is_media: true,
..SpawnPayload::default()
}
));
assert!(QueueManager::<tauri::Wry>::aria2_allocation_phase_eligible(
&SpawnPayload {
is_torrent: true,
torrent_file_allocation: Some("prealloc".to_string()),
..SpawnPayload::default()
}
));
assert!(!QueueManager::<tauri::Wry>::aria2_allocation_phase_eligible(
&SpawnPayload {
is_torrent: true,
torrent_file_allocation: Some("none".to_string()),
..SpawnPayload::default()
}
));
assert!(!QueueManager::<tauri::Wry>::aria2_allocation_phase_eligible(
&SpawnPayload {
is_torrent: true,
torrent_verify_only: true,
..SpawnPayload::default()
}
));
}
#[tokio::test]
async fn allocation_stays_pending_while_async_add_uri_is_in_flight() {
let app = tauri::test::mock_builder()
.build(tauri::test::mock_context(tauri::test::noop_assets()))
.expect("mock app");
let started = Arc::new(tokio::sync::Notify::new());
let release = Arc::new(tokio::sync::Notify::new());
let manager = Arc::new(QueueManager::test_new(
app.handle().clone(),
1,
Arc::new(BlockingSpawner {
started: Arc::clone(&started),
release: Arc::clone(&release),
}),
));
manager
.reserve_enqueue_generation("allocation", 7)
.await
.expect("lifecycle reservation");
manager
.commit_reserved_enqueue(
QueuedTask {
id: "allocation".to_string(),
queue_id: "main".to_string(),
kind: TaskKind::Aria2,
payload: SpawnPayload::default(),
lifecycle_generation: 7,
},
7,
)
.await
.expect("queued task");
let dispatcher = tokio::spawn(Arc::clone(&manager).run_dispatcher());
tokio::time::timeout(Duration::from_secs(1), started.notified())
.await
.expect("addUri should begin");
assert_eq!(allocation_pending_epoch(&manager, "allocation"), Some((1, 7)));
release.notify_one();
tokio::time::timeout(Duration::from_secs(1), async {
loop {
if manager.aria2_gid_for_download("allocation").as_deref() == Some("blocking-gid") {
break;
}
tokio::task::yield_now().await;
}
})
.await
.expect("async addUri should install its GID");
manager.handle_aria2_download_start("blocking-gid", 1).await;
assert_eq!(allocation_pending_epoch(&manager, "allocation"), None);
dispatcher.abort();
}
#[tokio::test]
async fn download_start_before_gid_registration_is_buffered_and_consumed() {
let app = tauri::test::mock_builder()
.build(tauri::test::mock_context(tauri::test::noop_assets()))
.expect("mock app");
let manager = QueueManager::test_new(app.handle().clone(), 1, Arc::new(TestSpawner));
let epoch = manager.next_aria2_control_epoch("buffered-start").await;
manager
.begin_aria2_allocation(
"buffered-start",
epoch,
9,
&SpawnPayload::default(),
)
.await;
manager.handle_aria2_download_start("early-gid", 1).await;
assert!(manager
.pending_download_starts
.lock()
.await
.contains("early-gid"));
manager
.remember_gid("buffered-start".to_string(), "early-gid".to_string())
.await;
assert_eq!(allocation_pending_epoch(&manager, "buffered-start"), None);
assert!(!manager
.pending_download_starts
.lock()
.await
.contains("early-gid"));
}
#[tokio::test]
async fn allocation_fallback_and_stale_epochs_cannot_clear_a_new_lifecycle() {
let app = tauri::test::mock_builder()
.build(tauri::test::mock_context(tauri::test::noop_assets()))
.expect("mock app");
let manager = QueueManager::test_new(app.handle().clone(), 1, Arc::new(TestSpawner));
let first_epoch = manager.next_aria2_control_epoch("stale-allocation").await;
manager
.begin_aria2_allocation(
"stale-allocation",
first_epoch,
10,
&SpawnPayload::default(),
)
.await;
let second_epoch = manager.next_aria2_control_epoch("stale-allocation").await;
manager
.begin_aria2_allocation(
"stale-allocation",
second_epoch,
11,
&SpawnPayload::default(),
)
.await;
manager
.clear_aria2_allocation_for_epoch("stale-allocation", first_epoch)
.await;
manager
.clear_aria2_allocation_for_lifecycle_generation("stale-allocation", 10)
.await;
assert_eq!(
allocation_pending_epoch(&manager, "stale-allocation"),
Some((second_epoch, 11))
);
manager
.remember_gid("stale-allocation".to_string(), "fallback-gid".to_string())
.await;
manager
.handle_aria2_download_start("fallback-gid", 0)
.await;
assert_eq!(
allocation_pending_epoch(&manager, "stale-allocation"),
Some((second_epoch, 11))
);
manager
.complete_aria2_allocation_for_gid("fallback-gid", 0)
.await;
assert_eq!(
allocation_pending_epoch(&manager, "stale-allocation"),
Some((second_epoch, 11))
);
manager
.complete_aria2_allocation_for_gid("fallback-gid", 1)
.await;
assert_eq!(allocation_pending_epoch(&manager, "stale-allocation"), None);
manager
.begin_aria2_allocation(
"stale-allocation",
second_epoch,
11,
&SpawnPayload::default(),
)
.await;
manager.handle_aria2_download_start("fallback-gid", 1).await;
assert_eq!(allocation_pending_epoch(&manager, "stale-allocation"), None);
}
#[tokio::test]
async fn ignored_gid_start_markers_cannot_clear_a_new_lifecycle() {
let app = tauri::test::mock_builder()
.build(tauri::test::mock_context(tauri::test::noop_assets()))
.expect("mock app");
let manager = QueueManager::test_new(app.handle().clone(), 1, Arc::new(TestSpawner));
let old_epoch = manager.next_aria2_control_epoch("ignored-start").await;
manager
.begin_aria2_allocation(
"ignored-start",
old_epoch,
20,
&SpawnPayload::default(),
)
.await;
manager
.remember_gid("ignored-start".to_string(), "reused-gid".to_string())
.await;
manager.forget_aria2_gid("ignored-start").await;
manager.handle_aria2_download_start("reused-gid", 1).await;
assert!(!manager
.pending_download_starts
.lock()
.await
.contains("reused-gid"));
let new_epoch = manager.next_aria2_control_epoch("ignored-start").await;
manager
.begin_aria2_allocation(
"ignored-start",
new_epoch,
21,
&SpawnPayload::default(),
)
.await;
manager
.remember_gid("ignored-start".to_string(), "new-gid".to_string())
.await;
assert_eq!(
allocation_pending_epoch(&manager, "ignored-start"),
Some((new_epoch, 21))
);
manager.handle_aria2_download_start("new-gid", 1).await;
assert_eq!(allocation_pending_epoch(&manager, "ignored-start"), None);
}
#[tokio::test]
async fn allocation_is_cleared_by_terminal_reconciliation_and_cancellation() {
let app = tauri::test::mock_builder()
.build(tauri::test::mock_context(tauri::test::noop_assets()))
.expect("mock app");
let manager = QueueManager::test_new(app.handle().clone(), 1, Arc::new(TestSpawner));
let epoch = manager.next_aria2_control_epoch("terminal-allocation").await;
manager
.begin_aria2_allocation(
"terminal-allocation",
epoch,
12,
&SpawnPayload::default(),
)
.await;
manager
.apply_completion("terminal-allocation", PendingOutcome::Error("disk full".to_string()))
.await;
assert_eq!(allocation_pending_epoch(&manager, "terminal-allocation"), None);
let next_epoch = manager.next_aria2_control_epoch("terminal-allocation").await;
manager
.begin_aria2_allocation(
"terminal-allocation",
next_epoch,
13,
&SpawnPayload::default(),
)
.await;
manager.clear_aria2_allocation("terminal-allocation").await;
assert_eq!(allocation_pending_epoch(&manager, "terminal-allocation"), None);
}
struct SeedSpawner;
#[async_trait::async_trait]