From 7f62056adf71fb014e45a1d69ece9ded2437e034 Mon Sep 17 00:00:00 2001 From: phantomptr Date: Sat, 10 Oct 2026 10:15:53 -0700 Subject: [PATCH] Link downloads write .partial, resume, and refuse a duplicate A killed download left a short file under the real name, so every retry got 409 'Move or delete it first', and two downloads of one link wrote one file. Bytes now land in .partial and are renamed when complete; a retry resumes from the partial after re-checking its tail against the origin (a mismatch starts over); a file already complete is reported done; a second live download of the same file is refused with the first one's id. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Ub1mez9eaVtTF8pMEFXvA9 --- .../ps5upload-engine/src/remote_download.rs | 327 ++++++++++++++---- 1 file changed, 265 insertions(+), 62 deletions(-) diff --git a/engine/crates/ps5upload-engine/src/remote_download.rs b/engine/crates/ps5upload-engine/src/remote_download.rs index e14f1790..f2caa20b 100644 --- a/engine/crates/ps5upload-engine/src/remote_download.rs +++ b/engine/crates/ps5upload-engine/src/remote_download.rs @@ -33,8 +33,6 @@ // Only the copy loop uses these, and that is compiled out on Android. #[cfg(not(target_os = "android"))] use std::io::{Seek, SeekFrom, Write}; -#[cfg(not(target_os = "android"))] -use std::path::Path; use std::path::PathBuf; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; @@ -95,6 +93,9 @@ fn safe_file_name(raw: &str) -> String { pub struct Download { pub url_host: String, pub path: PathBuf, + /// Where the bytes land until the download is complete (`.partial`); renamed to + /// `path` at the end, so a killed download never leaves a short file under the real name. + pub partial: PathBuf, pub total: u64, pub written: AtomicU64, pub done: AtomicBool, @@ -167,6 +168,38 @@ fn local_name_for(probed: &str, keep_name: bool) -> String { name } +/// The in-progress name of a download: `.partial` beside it. +fn partial_path(path: &std::path::Path) -> PathBuf { + let mut name = path.file_name().unwrap_or_default().to_os_string(); + name.push(".partial"); + path.with_file_name(name) +} + +/// Whether a download is still writing (not finished, failed or cancelled). +fn is_live(dl: &Download) -> bool { + !dl.done.load(Ordering::Relaxed) + && !dl.cancelled.load(Ordering::Relaxed) + && dl.error.lock().unwrap_or_else(|e| e.into_inner()).is_none() +} + +/// Registers `dl` under `id` unless another live download writes the same file: two +/// downloads of one link would write one file. `Err` carries the live one's id. +fn register_unless_busy( + registry: &DownloadRegistry, + id: &str, + dl: Arc, +) -> Result<(), String> { + let mut items = registry.items.lock().unwrap_or_else(|e| e.into_inner()); + if let Some((busy, _)) = items + .iter() + .find(|(_, other)| other.path == dl.path && is_live(other)) + { + return Err(busy.clone()); + } + items.insert(id.to_string(), dl); + Ok(()) +} + fn json_err(code: StatusCode, msg: &str) -> Response { (code, Json(serde_json::json!({ "error": msg }))).into_response() } @@ -221,11 +254,17 @@ async fn start_handler( let name = local_name_for(&probe.filename, req.keep_name); let path = dir.join(name); - // Refuse to silently resume into, or clobber, an unrelated file of the - // same name: a half-written package that installs is far worse than an - // error, because the failure surfaces much later as a corrupt install. - if let Ok(meta) = std::fs::metadata(&path) { - if meta.len() != probe.total_size { + let partial = partial_path(&path); + let total = probe.total_size; + + // A file already under the final name is either this link's finished download (same + // size: done, nothing to fetch) or an unrelated file, never resumed into or clobbered: + // a half-written package that installs is far worse than an error, because the + // failure surfaces much later as a corrupt install. Unfinished bytes only ever live in + // `.partial`, so a killed download no longer leaves a short file here. + let complete = match std::fs::metadata(&path) { + Ok(meta) if meta.len() == total => true, + Ok(meta) => { return json_err( StatusCode::CONFLICT, &format!( @@ -233,11 +272,12 @@ async fn start_handler( Move or delete it first.", path.display(), meta.len(), - probe.total_size + total ), ); } - } + Err(_) => false, + }; let host = url .parse::() @@ -249,52 +289,62 @@ async fn start_handler( let dl = Arc::new(Download { url_host: host.clone(), path: path.clone(), - total: probe.total_size, - written: AtomicU64::new(0), - done: AtomicBool::new(false), + partial: partial.clone(), + total, + written: AtomicU64::new(if complete { total } else { 0 }), + done: AtomicBool::new(complete), cancelled: AtomicBool::new(false), error: Mutex::new(None), }); - state - .items - .lock() - .unwrap_or_else(|e| e.into_inner()) - .insert(id.clone(), dl.clone()); + if let Err(busy) = register_unless_busy(&state, &id, dl.clone()) { + return ( + StatusCode::CONFLICT, + Json(serde_json::json!({ + "error": format!("{} is already being downloaded.", path.display()), + "download_id": busy, + })), + ) + .into_response(); + } - // Never log the path or query of a link: signed download URLs carry - // credentials there. The host is enough to tell two origins apart. - crate::log_info!( - "link-download start: id={} host={} bytes={} dest={}", - id, - host, - probe.total_size, - path.display() - ); - - let total = probe.total_size; - let dest = path.clone(); - let worker = dl.clone(); - let existed = path.exists(); - tokio::task::spawn_blocking(move || { - let dl = worker; - let src = crate::remote_pkg::RemoteSource::new_with_options(url, total, insecure); - if let Err(e) = copy_to_file(&src, &dl, &dest) { - crate::log_warn!("link-download failed: host={} err={}", dl.url_host, e); - // A partial file this download made would refuse the retry as "already exists - // with a different size": remove it, as Cancel does. - if !existed { - let _ = std::fs::remove_file(&dest); + if complete { + crate::log_info!( + "link-download: {} is already complete ({} bytes)", + path.display(), + total + ); + } else { + // Never log the path or query of a link: signed download URLs carry + // credentials there. The host is enough to tell two origins apart. + crate::log_info!( + "link-download start: id={} host={} bytes={} dest={}", + id, + host, + total, + path.display() + ); + let worker = dl.clone(); + tokio::task::spawn_blocking(move || { + let dl = worker; + let src = crate::remote_pkg::RemoteSource::new_with_options(url, total, insecure); + match copy_to_file(&src, &dl) { + // The partial stays: the next try of the same link resumes from it. + Err(e) => { + crate::log_warn!("link-download failed: host={} err={}", dl.url_host, e); + *dl.error.lock().unwrap_or_else(|x| x.into_inner()) = Some(e.to_string()); + } + Ok(()) if !dl.cancelled.load(Ordering::Relaxed) => { + dl.done.store(true, Ordering::Relaxed); + crate::log_info!( + "link-download done: host={} bytes={}", + dl.url_host, + dl.written.load(Ordering::Relaxed) + ); + } + Ok(()) => {} } - *dl.error.lock().unwrap_or_else(|x| x.into_inner()) = Some(e.to_string()); - } else if !dl.cancelled.load(Ordering::Relaxed) { - dl.done.store(true, Ordering::Relaxed); - crate::log_info!( - "link-download done: host={} bytes={}", - dl.url_host, - dl.written.load(Ordering::Relaxed) - ); - } - }); + }); + } ( StatusCode::OK, @@ -318,26 +368,55 @@ async fn start_handler( ) } -/// Pull the package window by window and write it out sequentially. +/// Bytes re-fetched before a resume point and compared with the partial file's tail: a +/// `.partial` of the same name from a different link (or a different build of the package) +/// must not be spliced onto this one. +#[cfg(not(target_os = "android"))] +const RESUME_CHECK: u64 = 64 * 1024; + +/// Pull the package window by window into `dl.partial`, then rename it to `dl.path`. +/// +/// Resumes from the partial file's length: the probe already established that the origin +/// serves byte ranges (it refuses one that does not), so a resume is always possible. The +/// last `RESUME_CHECK` bytes are fetched again and compared first; a mismatch, or a partial +/// longer than the package, starts over. /// /// Sequential writes on purpose: the fetcher is already parallel *inside* a /// window, so this needs no concurrency of its own, and one append-only /// stream is what a spinning disk and a network share both handle best. #[cfg(not(target_os = "android"))] -fn copy_to_file( - src: &crate::remote_pkg::RemoteSource, - dl: &Arc, - path: &Path, -) -> std::io::Result<()> { +fn copy_to_file(src: &crate::remote_pkg::RemoteSource, dl: &Arc) -> std::io::Result<()> { + use std::io::Read; let mut file = std::fs::OpenOptions::new() .create(true) + .read(true) .write(true) .truncate(false) - .open(path)?; - file.seek(SeekFrom::Start(0))?; - + .open(&dl.partial)?; let total = dl.total; - let mut offset = 0u64; + let mut offset = file.metadata()?.len(); + if offset > total { + offset = 0; + } + if offset > 0 { + let from = offset.saturating_sub(RESUME_CHECK); + let mut have = vec![0u8; (offset - from) as usize]; + file.seek(SeekFrom::Start(from))?; + file.read_exact(&mut have)?; + let fresh = src.read_range(from, offset - 1)?; + if fresh != have { + crate::log_info!( + "link-download: the partial file is from another download; starting over" + ); + offset = 0; + } else { + crate::log_info!("link-download: resuming at {offset} of {total} bytes"); + } + } + file.set_len(offset)?; + file.seek(SeekFrom::Start(offset))?; + dl.written.store(offset, Ordering::Relaxed); + while offset < total { if dl.cancelled.load(Ordering::Relaxed) { return Ok(()); @@ -353,7 +432,11 @@ fn copy_to_file( // what it reads is what we fetched, not what is still in the page cache // of a machine that might be about to sleep. file.sync_all()?; - Ok(()) + drop(file); + if dl.cancelled.load(Ordering::Relaxed) { + return Ok(()); + } + std::fs::rename(&dl.partial, &dl.path) } async fn status_handler( @@ -401,7 +484,7 @@ async fn cancel_handler( // Only ever the incomplete one -- a download that already finished is a // file the user asked for, whatever a late cancel says. if !dl.done.load(Ordering::Relaxed) { - match std::fs::remove_file(&dl.path) { + match std::fs::remove_file(&dl.partial) { Ok(()) => crate::log_info!("link-download cancelled: removed the partial file"), Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} Err(e) => crate::log_warn!("link-download cancelled: could not remove partial: {e}"), @@ -457,6 +540,126 @@ mod tests { ); } + #[cfg(not(target_os = "android"))] + fn scratch(tag: &str) -> PathBuf { + let d = std::env::temp_dir().join(format!("ps5u-dl-{tag}-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&d).unwrap(); + d + } + + #[cfg(not(target_os = "android"))] + fn download_for(dir: &std::path::Path, total: u64) -> Arc { + let path = dir.join("game.pkg"); + Arc::new(Download { + partial: partial_path(&path), + path, + total, + ..Default::default() + }) + } + + /// The bytes land in `.partial` and only a complete file gets the real name. + #[cfg(not(target_os = "android"))] + #[test] + fn a_download_writes_a_partial_and_renames_it_when_complete() { + use crate::remote_pkg::origin_tests::{body, spawn_origin}; + let data = body(300_000); + let origin = spawn_origin(data.clone(), 0, false); + let dir = scratch("full"); + let dl = download_for(&dir, data.len() as u64); + assert_eq!(dl.partial, dir.join("game.pkg.partial")); + let src = crate::remote_pkg::RemoteSource::new( + format!("http://{}/game.pkg", origin.addr), + data.len() as u64, + ); + copy_to_file(&src, &dl).unwrap(); + assert_eq!(std::fs::read(&dl.path).unwrap(), data); + assert!(!dl.partial.exists()); + let _ = std::fs::remove_dir_all(&dir); + } + + /// A killed download resumes from its partial file instead of being refused. + #[cfg(not(target_os = "android"))] + #[test] + fn a_download_resumes_from_its_partial_file() { + use crate::remote_pkg::origin_tests::{body, spawn_origin}; + let data = body(300_000); + let origin = spawn_origin(data.clone(), 0, false); + let dir = scratch("resume"); + let dl = download_for(&dir, data.len() as u64); + std::fs::write(&dl.partial, &data[..200_000]).unwrap(); + let src = crate::remote_pkg::RemoteSource::new( + format!("http://{}/game.pkg", origin.addr), + data.len() as u64, + ); + copy_to_file(&src, &dl).unwrap(); + assert_eq!(std::fs::read(&dl.path).unwrap(), data); + let _ = std::fs::remove_dir_all(&dir); + } + + /// A partial of the same name from another file is not spliced on: it starts over. + #[cfg(not(target_os = "android"))] + #[test] + fn a_partial_from_another_file_starts_over() { + use crate::remote_pkg::origin_tests::{body, spawn_origin}; + let data = body(300_000); + let origin = spawn_origin(data.clone(), 0, false); + let dir = scratch("other"); + let dl = download_for(&dir, data.len() as u64); + std::fs::write(&dl.partial, vec![0xAAu8; 150_000]).unwrap(); + let src = crate::remote_pkg::RemoteSource::new( + format!("http://{}/game.pkg", origin.addr), + data.len() as u64, + ); + copy_to_file(&src, &dl).unwrap(); + assert_eq!(std::fs::read(&dl.path).unwrap(), data); + let _ = std::fs::remove_dir_all(&dir); + } + + /// The start route: an already complete file is done at once, and a second live + /// download of the same file is refused with the first one's id. + #[cfg(not(target_os = "android"))] + #[tokio::test(flavor = "multi_thread")] + async fn a_complete_file_is_done_and_a_second_live_download_is_refused() { + use crate::remote_pkg::origin_tests::{body, spawn_origin}; + let data = body(100_000); + let origin = spawn_origin(data.clone(), 0, false); + let dir = scratch("route"); + std::fs::write(dir.join("game.pkg"), &data).unwrap(); + let state: DownloadStateHandle = Arc::default(); + let req = || DownloadStartRequest { + url: format!("http://{}/game.pkg", origin.addr), + insecure_tls: false, + dest_dir: Some(dir.display().to_string()), + keep_name: false, + }; + let resp = start_handler(State(state.clone()), Json(req())).await; + assert_eq!(resp.status(), StatusCode::OK); + let item = state + .items + .lock() + .unwrap() + .values() + .next() + .cloned() + .unwrap(); + assert!(item.done.load(Ordering::Relaxed)); + assert_eq!(item.written.load(Ordering::Relaxed), data.len() as u64); + + // A live one for the same file blocks a second. + let live = download_for(&dir, 5); + state.items.lock().unwrap().clear(); + register_unless_busy(&state, "first", live.clone()).unwrap(); + assert_eq!( + register_unless_busy(&state, "second", download_for(&dir, 5)), + Err("first".to_string()) + ); + // Once it ended (failed here), the file is free again. + *live.error.lock().unwrap() = Some("x".into()); + assert!(register_unless_busy(&state, "third", download_for(&dir, 5)).is_ok()); + let _ = std::fs::remove_dir_all(&dir); + } + #[test] fn the_download_directory_is_overridable() { // The default must be under the user's home, not a temp dir that a