gsadmin/Firelink
1
0
Fork 0
mirror of https://github.com/nimbold/Firelink.git synced 2026-07-26 12:08:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(downloads): close media lifecycle cleanup gaps

Browse Source 
Validate every Aria2 mirror URI, keep metadata cleanup errors truthful, and serialize media registration with lifecycle cancellation so abandoned tombstones cannot accumulate.
This commit is contained in:
NimBold
2026-07-17 01:03:52 +03:30
parent 6ef911919d
commit 50c3da2f5d
3 changed files with 128 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/download.rs
+34 -5
View File
@@ -77,7 +77,7 @@ impl DownloadCoordinator {
&self,
id: String,
lifecycle_generation: u64,
ack: tokio::sync::oneshot::Sender<()>,
ack: tokio::sync::oneshot::Sender<bool>,
) -> Result<(), String> {
self.media_tx
.send(MediaCmd::PauseWithAck {
@@ -128,7 +128,7 @@ enum MediaCmd {
PauseWithAck {
id: String,
lifecycle_generation: u64,
ack: tokio::sync::oneshot::Sender<()>,
ack: tokio::sync::oneshot::Sender<bool>,
},
Finished {
id: String,
@@ -144,7 +144,7 @@ async fn run_coordinator(
let mut active_media = HashMap::<String, (u64, watch::Sender<bool>)>::new();
let mut cancelled_media_generations = HashMap::<String, u64>::new();
let mut pending_media_acks =
HashMap::<(String, u64), tokio::sync::oneshot::Sender<()>>::new();
HashMap::<(String, u64), tokio::sync::oneshot::Sender<bool>>::new();
let mut pending_captured_urls = Vec::<String>::new();
let mut frontend_ready = false;
let mut command_open = true;
@@ -228,7 +228,10 @@ async fn run_coordinator(
.entry(id)
.and_modify(|generation| *generation = (*generation).max(lifecycle_generation))
.or_insert(lifecycle_generation);
let _ = ack.send(());
// No runner was registered for this generation.
// The caller may retire the cancellation tombstone
// while it still owns the per-download lock.
let _ = ack.send(false);
}
}
MediaCmd::Finished { id, lifecycle_generation } => {
@@ -238,7 +241,9 @@ async fn run_coordinator(
if let Some(ack) =
pending_media_acks.remove(&(id.clone(), lifecycle_generation))
{
let _ = ack.send(());
// A pending acknowledgement means the runner was
// registered and has now observed cancellation.
let _ = ack.send(true);
}
if cancelled_media_generations
.get(&id)
@@ -404,6 +409,30 @@ mod tests {
.expect("late media registration was not cancelled");
}
#[tokio::test]
async fn unregistered_media_pause_tombstone_can_be_reconciled() {
let (coordinator, _events) = DownloadCoordinator::spawn_headless();
let (ack_tx, ack_rx) = tokio::sync::oneshot::channel();
coordinator
.pause_media_with_ack("abandoned-media".to_string(), 9, ack_tx)
.await
.unwrap();
assert!(!ack_rx.await.unwrap());
// Once the queue lifecycle is invalidated under its control lock,
// it can retire the tombstone without allowing a late runner to
// start. A later registration must therefore remain uncancelled.
coordinator
.finish_media("abandoned-media".to_string(), 9)
.await;
let cancel_rx = coordinator
.register_media("abandoned-media".to_string(), 9)
.await
.unwrap();
assert!(!*cancel_rx.borrow());
}
#[tokio::test]
async fn stale_media_registration_cannot_replace_a_newer_lifecycle() {
let (coordinator, _events) = DownloadCoordinator::spawn_headless();
src-tauri/src/lib.rs
+80 -35
View File
@@ -1928,6 +1928,15 @@ struct ShellCommandOutput {
stderr: Vec<u8>,
}
fn terminate_shell_process_tree(child: tauri_plugin_shell::process::CommandChild) {
crate::process::kill_process_tree(child.pid());
// The process-tree helper may already have signalled the root. The
// direct kill is still needed when the root was not visible in the
// snapshot, but its result is cleanup telemetry, not the operation's
// primary error.
let _ = child.kill();
}
async fn shell_command_output_with_timeout(
command: tauri_plugin_shell::process::Command,
timeout: Duration,
@@ -1988,33 +1997,21 @@ async fn shell_command_output_with_timeout(
match tokio::time::timeout(timeout, collect_output).await {
Ok(Ok(output)) if output.status_code.is_some() => Ok(output),
Ok(Ok(_)) => match child.kill() {
Ok(()) => Err(format!(
"{operation} ended without a process exit status"
)),
Err(error) => Err(format!(
"{operation} ended without a process exit status and failed to terminate the child process: {error}"
)),
},
Ok(Err(error)) => {
let kill_error = child.kill().err();
match kill_error {
Some(kill_error) => Err(format!(
"{error}; failed to terminate the child process: {kill_error}"
)),
None => Err(error),
}
Ok(Ok(_)) => {
terminate_shell_process_tree(child);
Err(format!("{operation} ended without a process exit status"))
}
Err(_) => match child.kill() {
Ok(()) => Err(format!(
Ok(Err(error)) => {
terminate_shell_process_tree(child);
Err(error)
}
Err(_) => {
terminate_shell_process_tree(child);
Err(format!(
"{operation} timed out after {}s",
timeout.as_secs()
)),
Err(error) => Err(format!(
"{operation} timed out after {}s and failed to terminate the child process: {error}",
timeout.as_secs()
)),
},
))
}
}
}
@@ -3770,8 +3767,7 @@ pub(crate) async fn start_media_download_internal(
let failure_reason = loop {
tokio::select! {
_ = cancel_rx.changed() => {
crate::process::kill_process_tree(child.pid());
let _ = child.kill();
terminate_shell_process_tree(child);
if processing_started {
cleanup_media_processing_artifacts(&out_path).await;
}
@@ -4148,10 +4144,20 @@ async fn pause_download(
.pause_media_with_ack(id.clone(), media_lifecycle_generation, tx)
.await?;
} else {
let _ = tx.send(());
let _ = tx.send(false);
}
rx.await
let media_was_registered = rx
.await
.map_err(|_| "download worker stopped without acknowledging pause".to_string())?;
if matches!(active_kind, Some(crate::queue::TaskKind::Media)) && !media_was_registered {
// No media runner reached the coordinator. The queue control lock is
// still held, and run_media now checks the same lifecycle generation
// before registration, so this tombstone can be retired safely.
state
.download_coordinator
.finish_media(id.clone(), media_lifecycle_generation)
.await;
}
state.queue_manager.release_permit(&id).await;
if registered_lifecycle_generation.is_some()
@@ -4514,9 +4520,15 @@ async fn remove_download(
.pause_media_with_ack(id.clone(), media_lifecycle_generation, tx)
.await?;
} else {
let _ = tx.send(());
let _ = tx.send(false);
}
let media_was_registered = rx.await.unwrap_or(false);
if matches!(active_kind, Some(crate::queue::TaskKind::Media)) && !media_was_registered {
state
.download_coordinator
.finish_media(id.clone(), media_lifecycle_generation)
.await;
}
let _ = rx.await;
state.queue_manager.release_permit(&id).await;
state.queue_manager.clear_aria2_retry_state(&id).await;
@@ -4682,9 +4694,15 @@ async fn detach_download_for_reconfigure(
.pause_media_with_ack(id.clone(), media_lifecycle_generation, tx)
.await?;
} else {
let _ = tx.send(()); // Fallback if no task exists
let _ = tx.send(false); // Fallback if no task exists
}
let media_was_registered = rx.await.unwrap_or(false); // Wait for the writer to stop
if matches!(active_kind, Some(crate::queue::TaskKind::Media)) && !media_was_registered {
state
.download_coordinator
.finish_media(id.clone(), media_lifecycle_generation)
.await;
}
let _ = rx.await; // Wait for the writer to stop
state.queue_manager.release_permit(&id).await;
state.queue_manager.clear_aria2_retry_state(&id).await;
@@ -5101,13 +5119,20 @@ async fn validate_enqueue_url(url: &str) -> Result<(), String> {
}
}
async fn validate_enqueue_uris(url: &str, mirrors: Option<&str>) -> Result<(), String> {
for uri in collect_download_uris(url, mirrors) {
validate_enqueue_url(&uri).await?;
}
Ok(())
}
#[tauri::command]
async fn enqueue_download(
app_handle: tauri::AppHandle,
state: tauri::State<'_, AppState>,
mut item: queue::EnqueueItem,
) -> Result<crate::ipc::EnqueueAccepted, AppError> {
validate_enqueue_url(&item.url)
validate_enqueue_uris(&item.url, item.mirrors.as_deref())
.await
.map_err(AppError::Internal)?;
let id = item.id.clone();
@@ -5174,7 +5199,7 @@ async fn enqueue_many(
let mut results = Vec::with_capacity(items.len());
for mut item in items {
let id = item.id.clone();
if let Err(error) = validate_enqueue_url(&item.url).await {
if let Err(error) = validate_enqueue_uris(&item.url, item.mirrors.as_deref()).await {
results.push(crate::ipc::EnqueueResult {
id,
success: false,
@@ -6199,7 +6224,7 @@ mod tests {
observe_aria2_connections, observe_aria2_connections_with_epoch,
Aria2ConnectionObservation, Aria2ConnectionSample, Aria2RecoveryReason,
parse_media_playlist_metadata,
validate_enqueue_url,
validate_enqueue_url, validate_enqueue_uris,
};
use serde_json::json;
use std::time::{Duration, Instant};
@@ -6228,6 +6253,26 @@ mod tests {
);
}
#[tokio::test]
async fn enqueue_uri_validation_covers_mirrors_not_only_the_primary_url() {
assert_eq!(
validate_enqueue_uris(
"http://192.0.2.1/file.zip",
Some("http://127.0.0.1/internal\nfile:///etc/passwd"),
)
.await,
Err("SSRF blocked: Private/local IP not allowed".to_string())
);
assert_eq!(
validate_enqueue_uris(
"http://192.0.2.1/file.zip",
Some("file:///etc/passwd"),
)
.await,
Err("Unsupported URL scheme".to_string())
);
}
#[test]
fn slow_nonzero_aria2_throughput_recovers_after_a_sustained_degradation() {
let start = Instant::now();
src-tauri/src/queue.rs
+14
View File
@@ -2076,10 +2076,24 @@ impl SidecarSpawner for ProductionSpawner {
lifecycle_generation: u64,
) -> Result<(), String> {
let state = self.app_handle.state::<crate::AppState>();
// Serialize registration with pause/remove/detach. If the queue
// lifecycle was invalidated before this worker reached the
// coordinator, do not create a late media registration that could
// outlive the row's permit and ownership.
let control_guard = state.queue_manager.acquire_aria2_control(id).await;
if !state
.queue_manager
.is_registered_generation(id, lifecycle_generation)
.await
{
drop(control_guard);
return Err(crate::queue::MEDIA_RUN_CANCELLED.to_string());
}
let mut cancel_rx = state
.download_coordinator
.register_media(id.to_string(), lifecycle_generation)
.await?;
drop(control_guard);
let outcome = if *cancel_rx.borrow() {
Err(crate::queue::MEDIA_RUN_CANCELLED.to_string())
} else {