diff --git a/apps/desktop/src-tauri/Cargo.toml b/apps/desktop/src-tauri/Cargo.toml index 37f1898..af055b7 100644 --- a/apps/desktop/src-tauri/Cargo.toml +++ b/apps/desktop/src-tauri/Cargo.toml @@ -23,7 +23,7 @@ tauri-plugin-opener = "2" tauri-plugin-dialog = "2" serde = { version = "1", features = ["derive"] } serde_json = "1" -tokio = { version = "1", features = ["fs", "process", "io-util", "rt", "macros", "sync", "time"] } +tokio = { version = "1", features = ["fs", "process", "io-util", "rt", "rt-multi-thread", "macros", "sync", "time"] } regex = "1.10" reqwest = { version = "0.12", features = ["json", "stream"] } uuid = "1" diff --git a/apps/desktop/src-tauri/src/download.rs b/apps/desktop/src-tauri/src/download.rs index a77108d..331c468 100644 --- a/apps/desktop/src-tauri/src/download.rs +++ b/apps/desktop/src-tauri/src/download.rs @@ -1,5 +1,5 @@ -use futures_util::StreamExt; use crate::DownloadProgressEvent; +use futures_util::StreamExt; use reqwest::{ header::{self, HeaderMap, HeaderName, HeaderValue}, Client, StatusCode, @@ -30,6 +30,21 @@ pub enum DownloadCmd { FrontendReady(bool), } +#[derive(Clone, Debug, PartialEq)] +pub enum DownloadEvent { + Progress { + id: Uuid, + fraction: f64, + completed: u64, + total: Option, + }, + Completed(Uuid), + Failed { + id: Uuid, + error: String, + }, +} + #[derive(Debug)] pub struct DownloadPayload { pub id: Uuid, @@ -53,9 +68,21 @@ pub struct DownloadCoordinator { impl DownloadCoordinator { pub fn spawn(app_handle: AppHandle) -> Self { + Self::spawn_with_events(CoordinatorEventSink::Tauri(app_handle)) + } + + pub fn spawn_headless() -> (Self, mpsc::UnboundedReceiver) { + let (event_tx, event_rx) = mpsc::unbounded_channel(); + ( + Self::spawn_with_events(CoordinatorEventSink::Headless(event_tx)), + event_rx, + ) + } + + fn spawn_with_events(events: CoordinatorEventSink) -> Self { let (tx, rx) = mpsc::channel(128); let (media_tx, media_rx) = mpsc::channel(32); - tauri::async_runtime::spawn(run_coordinator(app_handle, rx, media_rx)); + tauri::async_runtime::spawn(run_coordinator(events, rx, media_rx)); Self { tx, media_tx } } @@ -87,6 +114,90 @@ impl DownloadCoordinator { } } +#[derive(Clone)] +enum CoordinatorEventSink { + Tauri(AppHandle), + Headless(mpsc::UnboundedSender), +} + +impl CoordinatorEventSink { + fn emit_progress( + &self, + id: Uuid, + completed: u64, + total: Option, + interval_bytes: u64, + interval: Duration, + ) { + let speed_bytes = if interval.is_zero() { + 0.0 + } else { + interval_bytes as f64 / interval.as_secs_f64() + }; + let fraction = total + .filter(|total| *total > 0) + .map(|total| completed as f64 / total as f64) + .unwrap_or(0.0) + .clamp(0.0, 1.0); + + match self { + Self::Tauri(app_handle) => { + let eta = total + .filter(|total| speed_bytes > 0.0 && *total > completed) + .map(|total| format_duration((total - completed) as f64 / speed_bytes)) + .unwrap_or_else(|| "-".to_string()); + let _ = app_handle.emit( + "download-progress", + DownloadProgressEvent { + id: id.to_string(), + fraction, + speed: format_speed(speed_bytes), + eta, + }, + ); + } + Self::Headless(event_tx) => { + let _ = event_tx.send(DownloadEvent::Progress { + id, + fraction, + completed, + total, + }); + } + } + } + + fn emit_completed(&self, id: Uuid) { + match self { + Self::Tauri(app_handle) => { + let _ = app_handle.emit("download-complete", id.to_string()); + } + Self::Headless(event_tx) => { + let _ = event_tx.send(DownloadEvent::Completed(id)); + } + } + } + + fn emit_failed(&self, id: Uuid, error: String) { + match self { + Self::Tauri(app_handle) => { + eprintln!("download {id} failed: {error}"); + let _ = app_handle.emit("download-failed", id.to_string()); + } + Self::Headless(event_tx) => { + let _ = event_tx.send(DownloadEvent::Failed { id, error }); + } + } + } + + fn emit_captured_urls(&self, payload: String) -> bool { + match self { + Self::Tauri(app_handle) => app_handle.emit("deep-link-add-download", payload).is_ok(), + Self::Headless(_) => true, + } + } +} + enum MediaCmd { Register { id: String, @@ -124,7 +235,7 @@ enum DownloadOutcome { } async fn run_coordinator( - app_handle: AppHandle, + events: CoordinatorEventSink, mut command_rx: mpsc::Receiver, mut media_rx: mpsc::Receiver, ) { @@ -154,10 +265,10 @@ async fn run_coordinator( let (control_tx, control_rx) = mpsc::channel(1); active.insert(id, ActiveDownload { generation, control_tx }); - let app_handle = app_handle.clone(); + let events = events.clone(); let worker_tx = worker_tx.clone(); tauri::async_runtime::spawn(async move { - let outcome = download_file(app_handle, payload, control_rx).await; + let outcome = download_file(events, payload, control_rx).await; let _ = worker_tx .send(WorkerEvent::Finished { id, generation, outcome }) .await; @@ -177,7 +288,7 @@ async fn run_coordinator( append_unique_urls(&mut pending_captured_urls, urls); if frontend_ready && !pending_captured_urls.is_empty() { let payload = pending_captured_urls.join("\n"); - if app_handle.emit("deep-link-add-download", payload).is_ok() { + if events.emit_captured_urls(payload) { pending_captured_urls.clear(); } } @@ -186,7 +297,7 @@ async fn run_coordinator( frontend_ready = ready; if ready && !pending_captured_urls.is_empty() { let payload = pending_captured_urls.join("\n"); - if app_handle.emit("deep-link-add-download", payload).is_ok() { + if events.emit_captured_urls(payload) { pending_captured_urls.clear(); } } @@ -207,11 +318,10 @@ async fn run_coordinator( match (is_current, outcome) { (true, DownloadOutcome::Completed) => { - let _ = app_handle.emit("download-complete", id.to_string()); + events.emit_completed(id); } (true, DownloadOutcome::Failed(error)) => { - eprintln!("download {id} failed: {error}"); - let _ = app_handle.emit("download-failed", id.to_string()); + events.emit_failed(id, error); } _ => {} } @@ -253,7 +363,7 @@ fn append_unique_urls(target: &mut Vec, urls: Vec) { } async fn download_file( - app_handle: AppHandle, + events: CoordinatorEventSink, payload: DownloadPayload, mut control_rx: mpsc::Receiver, ) -> DownloadOutcome { @@ -272,7 +382,7 @@ async fn download_file( for url in &payload.urls { for _ in 0..attempts_per_url { - match download_attempt(&app_handle, &client, &payload, url, &mut control_rx).await { + match download_attempt(&events, &client, &payload, url, &mut control_rx).await { Ok(()) => return DownloadOutcome::Completed, Err(AttemptError::Controlled(DownloadControl::Pause)) => { return DownloadOutcome::Paused; @@ -298,7 +408,7 @@ enum AttemptError { } async fn download_attempt( - app_handle: &AppHandle, + events: &CoordinatorEventSink, client: &Client, payload: &DownloadPayload, url: &str, @@ -392,8 +502,7 @@ async fn download_attempt( let now = Instant::now(); let interval = now.duration_since(last_emitted_at); if interval >= PROGRESS_INTERVAL { - emit_progress( - app_handle, + events.emit_progress( payload.id, completed, total_len, @@ -418,8 +527,7 @@ async fn download_attempt( .flush() .await .map_err(|error| AttemptError::Failed(error.to_string()))?; - emit_progress( - app_handle, + events.emit_progress( payload.id, completed, total_len, @@ -468,40 +576,6 @@ fn build_client(payload: &DownloadPayload) -> Result { builder.build().map_err(|error| error.to_string()) } -fn emit_progress( - app_handle: &AppHandle, - id: Uuid, - completed: u64, - total: Option, - interval_bytes: u64, - interval: Duration, -) { - let speed_bytes = if interval.is_zero() { - 0.0 - } else { - interval_bytes as f64 / interval.as_secs_f64() - }; - let fraction = total - .filter(|total| *total > 0) - .map(|total| completed as f64 / total as f64) - .unwrap_or(0.0) - .clamp(0.0, 1.0); - let eta = total - .filter(|total| speed_bytes > 0.0 && *total > completed) - .map(|total| format_duration((total - completed) as f64 / speed_bytes)) - .unwrap_or_else(|| "-".to_string()); - - let _ = app_handle.emit( - "download-progress", - DownloadProgressEvent { - id: id.to_string(), - fraction, - speed: format_speed(speed_bytes), - eta, - }, - ); -} - fn format_speed(bytes_per_second: f64) -> String { if bytes_per_second >= 1024.0 * 1024.0 { format!("{:.1} MB/s", bytes_per_second / (1024.0 * 1024.0)) diff --git a/apps/desktop/src-tauri/src/lib.rs b/apps/desktop/src-tauri/src/lib.rs index 8f12262..95d82fe 100644 --- a/apps/desktop/src-tauri/src/lib.rs +++ b/apps/desktop/src-tauri/src/lib.rs @@ -341,7 +341,7 @@ fn resign_aria2_debug_bundle(aria2c_path: &std::path::Path) -> Result<(), String Ok(()) } -mod download; +pub mod download; #[allow(dead_code)] mod ipc; mod parity; diff --git a/apps/desktop/src-tauri/tests/README.md b/apps/desktop/src-tauri/tests/README.md new file mode 100644 index 0000000..4e65f55 --- /dev/null +++ b/apps/desktop/src-tauri/tests/README.md @@ -0,0 +1,27 @@ +# Headless Download Engine Tests + +Run the full Rust suite: + +```sh +cd apps/desktop/src-tauri +cargo test --all-targets +``` + +Run only the async download integration harness with deterministic serial +performance measurements and visible test output: + +```sh +cd apps/desktop/src-tauri +RUST_BACKTRACE=1 cargo test --test download_engine -- --test-threads=1 --nocapture +``` + +The harness binds an ephemeral loopback port and requires no GUI, external +network access, or bundled media binaries. It validates: + +- aggregation of many streamed HTTP body chunks; +- pause and ranged resume through `DownloadCoordinator`; +- cancellation and partial-file cleanup; +- SHA-256 integrity after resume; +- retry recovery from transient HTTP failures; +- terminal error reporting after the retry budget is exhausted; +- a five-second local transfer performance budget for a 3 MiB fixture. diff --git a/apps/desktop/src-tauri/tests/download_engine.rs b/apps/desktop/src-tauri/tests/download_engine.rs new file mode 100644 index 0000000..9501af6 --- /dev/null +++ b/apps/desktop/src-tauri/tests/download_engine.rs @@ -0,0 +1,469 @@ +use axum::{ + body::{Body, Bytes}, + extract::State, + http::{ + header::{ACCEPT_RANGES, CONTENT_LENGTH, CONTENT_RANGE, RANGE}, + HeaderMap, HeaderValue, StatusCode, + }, + response::Response, + routing::get, + Router, +}; +use futures_util::stream; +use sha2::{Digest, Sha256}; +use std::{ + convert::Infallible, + sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, Mutex, + }, + time::{Duration, Instant}, +}; +use tauri_app_lib::download::{DownloadCmd, DownloadCoordinator, DownloadEvent, DownloadPayload}; +use tempfile::TempDir; +use tokio::{net::TcpListener, sync::mpsc, task::JoinHandle}; +use uuid::Uuid; + +const TEST_TIMEOUT: Duration = Duration::from_secs(10); + +#[derive(Clone)] +struct ServerState { + data: Arc>, + chunk_size: usize, + chunk_delay: Duration, + failures_remaining: Arc, + requests: Arc, + chunks_served: Arc, + range_starts: Arc>>>, +} + +struct TestServer { + base_url: String, + state: ServerState, + task: JoinHandle<()>, +} + +impl TestServer { + async fn spawn( + data: Vec, + chunk_size: usize, + chunk_delay: Duration, + failures_before_success: usize, + ) -> Self { + let state = ServerState { + data: Arc::new(data), + chunk_size, + chunk_delay, + failures_remaining: Arc::new(AtomicUsize::new(failures_before_success)), + requests: Arc::new(AtomicUsize::new(0)), + chunks_served: Arc::new(AtomicUsize::new(0)), + range_starts: Arc::new(Mutex::new(Vec::new())), + }; + let app = Router::new() + .route("/file", get(serve_file)) + .route("/always-fail", get(always_fail)) + .with_state(state.clone()); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let task = tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + + Self { + base_url: format!("http://{address}"), + state, + task, + } + } + + fn file_url(&self) -> String { + format!("{}/file", self.base_url) + } + + fn failure_url(&self) -> String { + format!("{}/always-fail", self.base_url) + } +} + +impl Drop for TestServer { + fn drop(&mut self) { + self.task.abort(); + } +} + +async fn serve_file(State(state): State, headers: HeaderMap) -> Response { + state.requests.fetch_add(1, Ordering::SeqCst); + if state + .failures_remaining + .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |remaining| { + remaining.checked_sub(1) + }) + .is_ok() + { + return response(StatusCode::SERVICE_UNAVAILABLE, Body::empty(), 0, None); + } + + let requested_start = headers + .get(RANGE) + .and_then(|value| value.to_str().ok()) + .and_then(parse_range_start); + state.range_starts.lock().unwrap().push(requested_start); + + let start = requested_start.unwrap_or(0); + if start >= state.data.len() { + return response( + StatusCode::RANGE_NOT_SATISFIABLE, + Body::empty(), + 0, + Some(format!("bytes */{}", state.data.len())), + ); + } + + let status = if requested_start.is_some() { + StatusCode::PARTIAL_CONTENT + } else { + StatusCode::OK + }; + let content_range = requested_start.map(|_| { + format!( + "bytes {start}-{}/{}", + state.data.len() - 1, + state.data.len() + ) + }); + let content_length = state.data.len() - start; + let data = state.data.clone(); + let chunk_size = state.chunk_size; + let chunk_delay = state.chunk_delay; + let chunks_served = state.chunks_served.clone(); + let body_stream = stream::unfold(start, move |offset| { + let data = data.clone(); + let chunks_served = chunks_served.clone(); + async move { + if offset >= data.len() { + return None; + } + if !chunk_delay.is_zero() { + tokio::time::sleep(chunk_delay).await; + } + let end = (offset + chunk_size).min(data.len()); + chunks_served.fetch_add(1, Ordering::SeqCst); + let chunk = Bytes::copy_from_slice(&data[offset..end]); + Some((Ok::<_, Infallible>(chunk), end)) + } + }); + + response( + status, + Body::from_stream(body_stream), + content_length, + content_range, + ) +} + +async fn always_fail(State(state): State) -> Response { + state.requests.fetch_add(1, Ordering::SeqCst); + response(StatusCode::INTERNAL_SERVER_ERROR, Body::empty(), 0, None) +} + +fn response( + status: StatusCode, + body: Body, + content_length: usize, + content_range: Option, +) -> Response { + let mut response = Response::new(body); + *response.status_mut() = status; + response + .headers_mut() + .insert(ACCEPT_RANGES, HeaderValue::from_static("bytes")); + response.headers_mut().insert( + CONTENT_LENGTH, + HeaderValue::from_str(&content_length.to_string()).unwrap(), + ); + if let Some(content_range) = content_range { + response.headers_mut().insert( + CONTENT_RANGE, + HeaderValue::from_str(&content_range).unwrap(), + ); + } + response +} + +fn parse_range_start(value: &str) -> Option { + value + .strip_prefix("bytes=")? + .strip_suffix('-')? + .parse() + .ok() +} + +fn fixture_data(size: usize) -> Vec { + (0..size) + .map(|index| ((index.wrapping_mul(31) + index / 7) % 251) as u8) + .collect() +} + +fn payload(id: Uuid, url: String, output_path: std::path::PathBuf) -> DownloadPayload { + DownloadPayload { + id, + urls: vec![url], + output_path, + speed_limit: None, + username: None, + password: None, + headers: None, + cookies: None, + user_agent: Some("Firelink integration test".to_string()), + max_tries: 1, + proxy: None, + } +} + +async fn next_event(events: &mut mpsc::UnboundedReceiver) -> DownloadEvent { + tokio::time::timeout(TEST_TIMEOUT, events.recv()) + .await + .expect("timed out waiting for download event") + .expect("download event channel closed") +} + +async fn wait_for_progress( + events: &mut mpsc::UnboundedReceiver, + id: Uuid, + minimum_completed: u64, +) { + loop { + match next_event(events).await { + DownloadEvent::Progress { + id: event_id, + completed, + .. + } if event_id == id && completed >= minimum_completed => return, + DownloadEvent::Failed { error, .. } => panic!("download failed: {error}"), + DownloadEvent::Completed(event_id) if event_id == id => { + panic!("download completed before it could be paused") + } + _ => {} + } + } +} + +async fn wait_for_completion( + events: &mut mpsc::UnboundedReceiver, + id: Uuid, +) -> Vec { + let mut observed = Vec::new(); + loop { + let event = next_event(events).await; + match &event { + DownloadEvent::Completed(event_id) if *event_id == id => { + observed.push(event); + return observed; + } + DownloadEvent::Failed { + id: event_id, + error, + } if *event_id == id => panic!("download failed: {error}"), + _ => observed.push(event), + } + } +} + +fn sha256(data: &[u8]) -> Vec { + Sha256::digest(data).to_vec() +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn pause_then_resume_uses_range_and_preserves_integrity() { + let expected = fixture_data(4 * 1024 * 1024); + let server = TestServer::spawn(expected.clone(), 16 * 1024, Duration::from_millis(4), 0).await; + let temp = TempDir::new().unwrap(); + let output_path = temp.path().join("paused.bin"); + let id = Uuid::from_u128(1); + let download = payload(id, server.file_url(), output_path.clone()); + let (coordinator, mut events) = DownloadCoordinator::spawn_headless(); + + coordinator + .send(DownloadCmd::Start(download)) + .await + .unwrap(); + wait_for_progress(&mut events, id, 256 * 1024).await; + coordinator.send(DownloadCmd::Pause(id)).await.unwrap(); + tokio::time::sleep(Duration::from_millis(100)).await; + + let paused_len = tokio::fs::metadata(&output_path).await.unwrap().len(); + assert!(paused_len >= 256 * 1024); + assert!(paused_len < expected.len() as u64); + + coordinator + .send(DownloadCmd::Start(payload( + id, + server.file_url(), + output_path.clone(), + ))) + .await + .unwrap(); + wait_for_completion(&mut events, id).await; + + let downloaded = tokio::fs::read(&output_path).await.unwrap(); + assert_eq!(downloaded.len(), expected.len()); + assert_eq!(sha256(&downloaded), sha256(&expected)); + let range_starts = server.state.range_starts.lock().unwrap().clone(); + assert_eq!(range_starts.first(), Some(&None)); + assert!( + range_starts + .iter() + .skip(1) + .flatten() + .any(|start| *start == paused_len as usize), + "resume request did not start at the paused file length: {range_starts:?}" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn aggregates_many_http_chunks_with_complete_progress() { + let expected = fixture_data(3 * 1024 * 1024); + let server = TestServer::spawn(expected.clone(), 8 * 1024, Duration::ZERO, 0).await; + let temp = TempDir::new().unwrap(); + let output_path = temp.path().join("chunked.bin"); + let id = Uuid::from_u128(2); + let (coordinator, mut events) = DownloadCoordinator::spawn_headless(); + let started = Instant::now(); + + coordinator + .send(DownloadCmd::Start(payload( + id, + server.file_url(), + output_path.clone(), + ))) + .await + .unwrap(); + let observed = wait_for_completion(&mut events, id).await; + + assert!( + started.elapsed() < Duration::from_secs(5), + "local 3 MiB transfer exceeded the performance budget" + ); + assert!(server.state.chunks_served.load(Ordering::SeqCst) > 100); + let final_progress = observed.iter().rev().find_map(|event| match event { + DownloadEvent::Progress { + id: event_id, + fraction, + completed, + total, + } if *event_id == id => Some((*fraction, *completed, *total)), + _ => None, + }); + assert_eq!( + final_progress, + Some((1.0, expected.len() as u64, Some(expected.len() as u64))) + ); + assert_eq!( + sha256(&tokio::fs::read(output_path).await.unwrap()), + sha256(&expected) + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn retries_transient_http_failures_then_completes() { + let expected = fixture_data(512 * 1024); + let server = TestServer::spawn(expected.clone(), 32 * 1024, Duration::ZERO, 2).await; + let temp = TempDir::new().unwrap(); + let output_path = temp.path().join("retry.bin"); + let id = Uuid::from_u128(3); + let (coordinator, mut events) = DownloadCoordinator::spawn_headless(); + let mut download = payload(id, server.file_url(), output_path.clone()); + download.max_tries = 3; + + coordinator + .send(DownloadCmd::Start(download)) + .await + .unwrap(); + wait_for_completion(&mut events, id).await; + + assert_eq!(server.state.requests.load(Ordering::SeqCst), 3); + assert_eq!( + sha256(&tokio::fs::read(output_path).await.unwrap()), + sha256(&expected) + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn reports_terminal_http_errors_after_retry_budget() { + let server = TestServer::spawn(Vec::new(), 8 * 1024, Duration::ZERO, 0).await; + let temp = TempDir::new().unwrap(); + let output_path = temp.path().join("failed.bin"); + let id = Uuid::from_u128(4); + let (coordinator, mut events) = DownloadCoordinator::spawn_headless(); + let mut download = payload(id, server.failure_url(), output_path.clone()); + download.max_tries = 2; + + coordinator + .send(DownloadCmd::Start(download)) + .await + .unwrap(); + let error = loop { + match next_event(&mut events).await { + DownloadEvent::Failed { + id: event_id, + error, + } if event_id == id => break error, + DownloadEvent::Completed(event_id) if event_id == id => { + panic!("failed download was reported as complete") + } + _ => {} + } + }; + + assert!(error.contains("500 Internal Server Error")); + assert_eq!(server.state.requests.load(Ordering::SeqCst), 2); + assert!(!output_path.exists()); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn cancel_removes_partial_file_without_terminal_success_event() { + let expected = fixture_data(4 * 1024 * 1024); + let server = TestServer::spawn(expected, 16 * 1024, Duration::from_millis(4), 0).await; + let temp = TempDir::new().unwrap(); + let output_path = temp.path().join("cancelled.bin"); + let id = Uuid::from_u128(5); + let (coordinator, mut events) = DownloadCoordinator::spawn_headless(); + + coordinator + .send(DownloadCmd::Start(payload( + id, + server.file_url(), + output_path.clone(), + ))) + .await + .unwrap(); + wait_for_progress(&mut events, id, 256 * 1024).await; + coordinator.send(DownloadCmd::Cancel(id)).await.unwrap(); + + tokio::time::timeout(TEST_TIMEOUT, async { + while output_path.exists() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("cancelled partial file was not removed"); + + while let Ok(Some(event)) = + tokio::time::timeout(Duration::from_millis(100), events.recv()).await + { + assert!( + !matches!( + event, + DownloadEvent::Completed(event_id) if event_id == id + ), + "cancelled download emitted a completion event" + ); + assert!( + !matches!( + event, + DownloadEvent::Failed { id: event_id, .. } if event_id == id + ), + "cancelled download emitted a failure event" + ); + } +}