diff --git a/src/download.rs b/src/download.rs index 946e9154..e24b3540 100644 --- a/src/download.rs +++ b/src/download.rs @@ -433,6 +433,7 @@ struct RestoreSweepPlan { referenced: HashSet, live: HashSet, dest_dirs: HashSet, + state_keep: HashSet<(String, u64)>, selections: Vec<(String, Vec)>, session_keep: Option>, } @@ -4683,8 +4684,21 @@ impl DownloadManager { .map(|it| it.url().to_string()) .filter(|u| crate::torrent::is_torrent_url(u)) .collect(); - let live: HashSet = self.items().map(|it| it.id()).collect(); + // Only rows that can still resume own staging: a Done row has + // delivered, so anything it left in its dest dir is debris. + let live: HashSet = self + .items() + .filter(|it| it.status() != DownloadStatus::Done) + .map(|it| it.id()) + .collect(); let dest_dirs: HashSet = self.items().map(|it| it.dest_dir()).collect(); + // State manifests belong to rows that can still resume. Done rows have + // delivered; anything they leave behind is crash debris. + let state_keep: HashSet<(String, u64)> = self + .items() + .filter(|it| it.status() != DownloadStatus::Done) + .map(|it| (it.dest_dir(), it.id())) + .collect(); let session_keep = crate::torrent::session_handle().is_some().then(|| { self.items() .filter(|it| { @@ -4697,6 +4711,7 @@ impl DownloadManager { referenced, live, dest_dirs, + state_keep, selections, session_keep, } @@ -4711,6 +4726,7 @@ impl DownloadManager { for dest_dir in plan.dest_dirs { crate::video::sweep_dest_staging(std::path::Path::new(&dest_dir), &plan.live); } + crate::video::sweep_state_staging(&plan.state_keep); crate::cookies::sweep_cookie_staging(); plan.session_keep } diff --git a/src/download_tests.rs b/src/download_tests.rs index 4aecfc1a..bb8d0750 100644 --- a/src/download_tests.rs +++ b/src/download_tests.rs @@ -4549,7 +4549,9 @@ fn remove_cleans_video_staging() { let manifest_json = format!( r#"{{"page_url":"https://x.com/u/status/1","quality":"1080p","video_format_id":null,"video_ext":"mp4","audio_format_id":"","audio_ext":"","final_bytes":null,"staging_name":"v.{id}.mp4"}}"# ); - std::fs::write(destdir.join(format!(".{id}.manifest.json")), manifest_json).unwrap(); + let manifest_file = crate::video_staging::manifest_path(&destdir, id); + std::fs::create_dir_all(manifest_file.parent().unwrap()).unwrap(); + std::fs::write(&manifest_file, manifest_json).unwrap(); for n in ["v.mp4", "v.srt"] { std::fs::write(destdir.join(n), b"x").unwrap(); } @@ -4578,7 +4580,7 @@ fn remove_cleans_video_staging() { ); manager.remove(id); assert!( - !destdir.join(format!(".{id}.manifest.json")).exists(), + !manifest_file.exists(), "staged sidecar must go with the row" ); assert!( diff --git a/src/video.rs b/src/video.rs index 8a7ae6de..92622732 100644 --- a/src/video.rs +++ b/src/video.rs @@ -40,7 +40,9 @@ pub use crate::video_staging::clean_staging; pub use crate::video_staging::clean_staging_files; /// Facade: staging/parts/manifest/resume lives in /// [`video_staging`](crate::video_staging) now. -pub use crate::video_staging::{clean_dest_parts, staging_root, sweep_dest_staging}; +pub use crate::video_staging::{ + clean_dest_parts, staging_root, sweep_dest_staging, sweep_state_staging, +}; /// Facade: tool provisioning lives in [`video_tools`](crate::video_tools) now. pub use crate::video_tools::{ install_deno, install_ytdlp, latest_deno_tag, latest_ytdlp_tag, resolve_libraries, diff --git a/src/video_argv.rs b/src/video_argv.rs index d9a7c48f..cf7e8df3 100644 --- a/src/video_argv.rs +++ b/src/video_argv.rs @@ -3,7 +3,9 @@ use crate::video_prefs::subtitle_cli_args; use crate::video_quality::quality_height; -use crate::video_staging::{VideoManifest, dest_part_path, manifest_path, ytdlp_output_template}; +use crate::video_staging::{ + VideoManifest, dest_part_path, manifest_path, write_private, ytdlp_output_template, +}; use crate::video_tools::{VideoError, ffmpeg_location_dir}; use std::path::{Path, PathBuf}; @@ -12,8 +14,13 @@ pub(crate) async fn write_manifest( item_id: u64, manifest: &VideoManifest, ) -> Result<(), VideoError> { - let text = serde_json::to_string_pretty(manifest).map_err(VideoError::staging)?; - tokio::fs::write(manifest_path(dir, item_id), text) + // Record the destination at the single writer choke point: callers build + // the selection fields, never the path bookkeeping. + let mut manifest = manifest.clone(); + manifest.dest_dir = Some(dir.to_path_buf()); + let text = serde_json::to_string_pretty(&manifest).map_err(VideoError::staging)?; + // Private: the manifest carries the page URL and destination path. + write_private(&manifest_path(dir, item_id), text.as_bytes()) .await .map_err(VideoError::staging) } diff --git a/src/video_runner.rs b/src/video_runner.rs index d384fe0a..72d1e24d 100644 --- a/src/video_runner.rs +++ b/src/video_runner.rs @@ -23,10 +23,10 @@ use crate::video_spawn::{ }; use crate::video_staging::{ ResumePlan, ResumeQuery, VideoManifest, clean_dest_parts, clean_staging_files, collect_sidecar, - dest_part_path, discover_unified_output, ensure_staging_dir_in, file_len, read_manifest, - release_remux_lease, reserve_remux_temp, resume_plan, sidecar_path_for, - staging_location_for_dest, sweep_partial_remuxes, sweep_staging_preserving_recordings, - write_private, + dest_part_path, discover_unified_output, ensure_staging_dir_in, file_len, info_json_path, + live_info_json_path, read_manifest, release_remux_lease, reserve_remux_temp, resume_plan, + sidecar_path_for, staging_location_for_dest, sweep_partial_remuxes, + sweep_staging_preserving_recordings, write_private, }; use crate::video_tools::{VideoError, ensure_tool_versions, resolve_libraries}; use crate::video_types::{FetchedVideo, VideoOutcome, yt_dlp_reports_currently_live}; @@ -47,6 +47,26 @@ pub enum StopIntent { Discard, } +/// Remove a single-use probe info JSON on drop. Both attempt paths have early +/// returns (abort, failure, lost race) that would otherwise leak it. +struct InfoJsonGuard { + path: Option, +} + +impl InfoJsonGuard { + fn new(path: Option) -> Self { + Self { path } + } +} + +impl Drop for InfoJsonGuard { + fn drop(&mut self) { + if let Some(path) = self.path.take() { + let _ = std::fs::remove_file(path); + } + } +} + /// Run one attempt: resolve → download parts → merge → rename into place. /// Returns the final size, or `None` when aborted (the pauser/canceller already set the row status). /// @@ -325,6 +345,10 @@ async fn run_video_download_inner( let plan = resume_plan(&query); match plan { ResumePlan::Finished => { + // The finished file is in place; this row's scratch and manifest + // are spent. Sweep before returning, or the bookkeeping outlives + // the download. + sweep_staging_preserving_recordings(&staging, job.item_id); return Ok(match file_len(&job.dest) { Some(n) => VideoOutcome::Finished(n), None => VideoOutcome::Aborted, @@ -338,9 +362,6 @@ async fn run_video_download_inner( } // Clear this item's staging files (grab--*), not the dest dir itself: a previous attempt's detached writers may still hold old inodes, so unlink first. Only mismatches/oversize leftovers land here; same-selection resume never does. crate::video::clean_staging_files(&staging, job.item_id); - // Reclaim a previous attempt's info-JSON temp file (crash debris; - // the normal path deletes it after the attempt). - let _ = std::fs::remove_file(staging.join(format!(".grab-info-{}.json", job.item_id))); // Dest-dir parts are Grab-namespaced, so a mismatch restarts clean instead of resuming into a foreign lookalike. The finished file itself is never touched. clean_dest_parts(&job.dest); // Record this attempt's selection up front: a pause from here on leaves a matchable sidecar, so the next attempt resumes instead of wiping. @@ -363,6 +384,7 @@ async fn run_video_download_inner( // `{stem}.{id}.`; record the prefix so cleanup can reclaim // them. Manifest-anchored, not inferred. staging_prefix: Some(unified_staging_prefix(&job.dest, job.item_id)), + dest_dir: None, }, ) .await?; @@ -555,8 +577,8 @@ pub(crate) async fn run_unified_ytdlp( // resolve above), so every requested format ID is present. yt-dlp still // re-extracts automatically if a loaded format URL 403/404s (safety net // for stale signed URLs). - let info_json_path: Option = if let Some(json) = job.info_json.as_ref() { - let json_path = staging.join(format!(".grab-info-{}.json", job.item_id)); + let json_path: Option = if let Some(json) = job.info_json.as_ref() { + let json_path = info_json_path(job.item_id); // Private write: the JSON can hold signed media URLs and headers. let written = match serde_json::to_string(json) { Ok(text) => write_private(&json_path, text.as_bytes()).await.is_ok(), @@ -579,6 +601,8 @@ pub(crate) async fn run_unified_ytdlp( } else { None }; + // Single-use: removed on drop, including every early return below. + let info_guard = InfoJsonGuard::new(json_path); // Single-counter progress with the pump's granularity gate; `done` is capped against the metadata total so the bar never passes 100%. let sent = Arc::new(AtomicU64::new(0)); let report = { @@ -619,12 +643,9 @@ pub(crate) async fn run_unified_ytdlp( timeout, ) .await; - // The info JSON temp file is single-use: delete it now that yt-dlp has - // read it (or failed to). Staging's manifest cleanup never owns it. - if let Some(path) = info_json_path { - let _ = std::fs::remove_file(path); - } let (done, after_move, abort_intent) = attempt_result?; + // yt-dlp is done reading it; don't keep signed URLs on disk through merge. + drop(info_guard); let Some(()) = done else { // User abort: on Preserve, try merging partial DASH parts into a // single file instead of abandoning them. @@ -1641,6 +1662,7 @@ pub(crate) async fn run_live_ytdlp( final_bytes: None, staging_name: Some(staging_name.clone()), staging_prefix: None, + dest_dir: None, }, ) .await; @@ -1661,7 +1683,7 @@ pub(crate) async fn run_live_ytdlp( .and_then(|json| serde_json::to_string(json).ok()) { Some(text) => { - let path = staging.join(format!(".grab-live-{}.json", job.item_id)); + let path = live_info_json_path(job.item_id); // Private write: the JSON can hold signed manifest URLs. match write_private(&path, text.as_bytes()).await { Ok(()) => Some(path), @@ -1672,23 +1694,11 @@ pub(crate) async fn run_live_ytdlp( }; // Reclaim a previous run's crash debris (the normal path deletes it below). if live_info_path.is_none() { - let _ = std::fs::remove_file(staging.join(format!(".grab-live-{}.json", job.item_id))); + let _ = std::fs::remove_file(live_info_json_path(job.item_id)); } // Deleted on drop: the attempt loop has early returns (quiesce failure, // nothing recorded) that would otherwise leak it. - struct InfoJsonGuard { - path: Option, - } - impl Drop for InfoJsonGuard { - fn drop(&mut self) { - if let Some(p) = self.path.take() { - let _ = std::fs::remove_file(p); - } - } - } - let _info_guard = InfoJsonGuard { - path: live_info_path.clone(), - }; + let _info_guard = InfoJsonGuard::new(live_info_path.clone()); let mut downgraded: Option = None; let src = 'attempt: loop { let attempt: &VideoJob = downgraded.as_ref().unwrap_or(job); @@ -2075,6 +2085,9 @@ pub(crate) async fn run_live_ytdlp( Exit::Delivered, ) .await; + // Delivered: the manifest's job is done. Failure paths above keep it, + // because it is the ownership proof for a salvaged recording. + sweep_staging_preserving_recordings(staging, job.item_id); Ok(file_len(&job.dest)) } diff --git a/src/video_staging.rs b/src/video_staging.rs index 9b72ef25..293f7bf1 100644 --- a/src/video_staging.rs +++ b/src/video_staging.rs @@ -19,7 +19,7 @@ use crate::video_prefs::subtitle_content_languages; use crate::video_tools::VideoError; use gettextrs::gettext; -use std::path::{Path, PathBuf}; +use std::path::{Component, Path, PathBuf}; /// Shared root for tiny engine scratch (cookie dumps, probe output). Stays on /// tmpfs: only small files ever land here. Bulky download parts stage beside @@ -29,6 +29,91 @@ pub fn staging_root() -> PathBuf { std::env::temp_dir().join("grab-video") } +/// Root for persistent per-item bookkeeping (resume manifests). This is +/// application state, not user content: it survives restarts but is not a +/// download. Debug-only `GRAB_STATE_DIR` seam: tests can point it at a scratch +/// dir; release builds always use the real location. +fn state_root() -> PathBuf { + if let Some(dir) = debug_dir_override("GRAB_STATE_DIR") { + return dir; + } + #[cfg(test)] + { + test_root("state") + } + #[cfg(not(test))] + { + xdg_dir("XDG_STATE_HOME") + .unwrap_or_else(|| glib::home_dir().join(".local/state")) + .join("grab") + } +} + +/// Root for single-use private scratch (probe info JSON holding signed URLs). +/// Ephemeral under `$XDG_RUNTIME_DIR`; when the session provides none, the +/// state root keeps it private and the startup sweep reclaims it. +fn runtime_root() -> PathBuf { + if let Some(dir) = debug_dir_override("GRAB_RUNTIME_DIR") { + return dir; + } + #[cfg(test)] + { + test_root("runtime") + } + #[cfg(not(test))] + { + match xdg_dir("XDG_RUNTIME_DIR") { + Some(dir) => dir.join("grab"), + None => state_root().join("run"), + } + } +} + +/// Single-use probe JSON paths in the private scratch root. Private because +/// they can carry signed media URLs and request headers; single-use because a +/// resumed attempt resolves fresh. +pub(crate) fn info_json_path(item_id: u64) -> PathBuf { + runtime_root().join(format!("info-{item_id}.json")) +} + +pub(crate) fn live_info_json_path(item_id: u64) -> PathBuf { + runtime_root().join(format!("live-{item_id}.json")) +} + +/// An absolute XDG base dir, or `None` when unset or relative (the spec says +/// relative paths must be ignored). +#[cfg(not(test))] +fn xdg_dir(var: &str) -> Option { + std::env::var_os(var) + .map(PathBuf::from) + .filter(|p| p.is_absolute()) +} + +#[cfg(debug_assertions)] +fn debug_dir_override(var: &str) -> Option { + std::env::var_os(var) + .filter(|v| !v.is_empty()) + .map(PathBuf::from) +} + +#[cfg(not(debug_assertions))] +fn debug_dir_override(_var: &str) -> Option { + None +} + +/// One scratch root per test process. Tests run serially but their temp dirs +/// differ; the shared root keeps them off the developer's real state dirs. +#[cfg(test)] +fn test_root(kind: &str) -> PathBuf { + static ROOT: std::sync::OnceLock = std::sync::OnceLock::new(); + ROOT.get_or_init(|| { + let root = std::env::temp_dir().join(format!("grab-test-state-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&root); + root + }) + .join(kind) +} + /// Staging files live directly in the destination dir, visible like other download managers: /// A resolved per-item staging dir plus the root it is guarded under: the /// dest-side root for current staging, the tmp root for legacy dirs still @@ -162,12 +247,42 @@ fn create_private_dir(dir: &Path) -> std::io::Result<()> { std::fs::create_dir(dir) } -/// The staging leaf must be a real directory, never a symlink: build the -/// parent chain, then create the leaf atomically, so a pre-planted link is -/// refused up front and a link racing for the name can't be followed into -/// place. Staging names are item ids, so unlike shareable folders there is -/// nothing to dedupe to — refusing loudly is the safe behavior. -fn ensure_real_staging_leaf(dir: &Path) -> Result<(), VideoError> { +/// Create a private bookkeeping dir and any missing ancestors with owner-only +/// permissions (0700). A symlinked leaf is refused, never followed: these dirs +/// hold manifests and probe JSONs. +fn ensure_private_dir(dir: &Path) -> std::io::Result<()> { + let mut missing = Vec::new(); + let mut current = dir; + while std::fs::symlink_metadata(current).is_err() { + missing.push(current.to_path_buf()); + let Some(parent) = current.parent() else { + break; + }; + current = parent; + } + for path in missing.iter().rev() { + match create_private_dir(path) { + Ok(()) => {} + // Lost a race for the name: the check below validates the winner. + Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {} + Err(e) => return Err(e), + } + } + match std::fs::symlink_metadata(dir) { + Ok(meta) if meta.file_type().is_dir() => Ok(()), + _ => Err(std::io::Error::other( + "bookkeeping directory is not a real directory", + )), + } +} + +/// The leaf must be a real directory, never a symlink: build the parent +/// chain, then create the leaf atomically, so a pre-planted link is refused +/// up front and a link racing for the name can't be followed into place. +/// Used for staging dirs and for Grab's private bookkeeping dirs; names are +/// item ids, so unlike shareable folders there is nothing to dedupe to — +/// refusing loudly is the safe behavior. +pub(crate) fn ensure_real_staging_leaf(dir: &Path) -> Result<(), VideoError> { let refused = || VideoError::staging(gettext("staging directory is not a real directory")); match std::fs::symlink_metadata(dir) { // `symlink_metadata` doesn't follow the final component, so a link @@ -271,18 +386,18 @@ pub(crate) fn drop_empty_staging_root(root: &Path) { let _ = std::fs::remove_dir(canon); } -/// Reclaim an orphan staging dir: scratch goes, completed `final.*` recordings -/// stay. Only numeric dir names; runs at startup after queue restore, so -/// nothing live is touched. -/// Sweep one destination's staging files: files for item IDs with no live row -/// are reclaimed. A missing dest dir is a no-op. +/// Startup sweep for manifests that older versions left in the destination +/// dir (canonical + legacy names). New manifests live in the state dir and are +/// handled by [`sweep_state_staging`]. A missing dest dir is a no-op. /// -/// Startup sweep: reclaim staging files for items with no live row. -/// Scans for manifest files, extracts their ids, and deletes exactly the -/// files each manifest owns (`staging_name` / `staging_prefix`). A manifest -/// that records no ownership is left alone. A user's own files are never -/// touched: without a manifest, everything is preserved. +/// Also removes the previous versions' single-use probe JSONs +/// (`.grab-info-*` / `.grab-live-*`): they are never reused across restarts. /// +/// Legacy manifests are no longer parsed: the sweep deletes the bookkeeping +/// files themselves, plus anything a *state* manifest for the same row still +/// proves ownership of (`staging_name` / `staging_prefix`). A manifest that +/// records no ownership is left alone. A user's own files are never touched: +/// without a manifest, everything is preserved. pub fn sweep_dest_staging(dest_dir: &Path, keep: &std::collections::HashSet) { // Manifest-driven cleanup: the ONLY destructive authority for staging // files is the manifest's `staging_name` / `staging_prefix`. Scan for @@ -293,6 +408,11 @@ pub fn sweep_dest_staging(dest_dir: &Path, keep: &std::collections::HashSet for entry in entries.filter_map(|e| e.ok()) { let name = entry.file_name(); let name_str = name.to_str().unwrap_or(""); + // Old single-use probe JSONs: stale by definition after a restart. + if is_legacy_info_name(name_str) { + let _ = std::fs::remove_file(entry.path()); + continue; + } // Find manifest files: .{id}.manifest.json (canonical) or legacy. let id = if let Some(id_str) = name_str .strip_prefix('.') @@ -324,6 +444,105 @@ pub fn sweep_dest_staging(dest_dir: &Path, keep: &std::collections::HashSet } } +/// Startup sweep for state-dir manifests: reclaim every manifest whose +/// `(destination, id)` no longer has a queue row. `keep` carries the raw queue +/// dest strings; keys are normalized here so a trailing slash cannot hide a +/// live row. Also clears stale runtime scratch. +pub fn sweep_state_staging(keep: &std::collections::HashSet<(String, u64)>) { + let keep: std::collections::HashSet<(String, u64)> = keep + .iter() + .map(|(dest, id)| (dest_key(Path::new(dest)), *id)) + .collect(); + if let Ok(entries) = std::fs::read_dir(state_root().join("video")) { + for entry in entries.filter_map(|e| e.ok()) { + let Some(name) = entry.file_name().to_str().map(str::to_owned) else { + continue; + }; + // An interrupted atomic write leaves this temp, never a truncated + // manifest. No writer runs during startup, so it is always debris. + if name.ends_with(".tmp") { + let _ = std::fs::remove_file(entry.path()); + continue; + } + let Some(id) = state_manifest_id(&name) else { + continue; + }; + let Some(manifest) = read_manifest_file(&entry.path()) else { + // Unreadable manifest: keep it while any queued row still + // claims the id, or a crash mid-write would drop resume state. + if !keep.iter().any(|(_, kid)| *kid == id) { + let _ = std::fs::remove_file(entry.path()); + } + continue; + }; + let Some(dest) = manifest.dest_dir.as_deref() else { + // No destination recorded = no file it could prove ownership + // of. State manifests always record one; drop the stray unless + // a queued row still claims the id. + if !keep.iter().any(|(_, kid)| *kid == id) { + let _ = std::fs::remove_file(entry.path()); + } + continue; + }; + if keep.contains(&(dest_key(dest), id)) { + continue; + } + // Act on the manifest we opened: its recorded destination may no + // longer hash to the path in its own filename (moved/symlink). + let Some(ownership) = manifest_ownership_of(&manifest) else { + continue; + }; + remove_owned_files(dest, &ownership); + let _ = std::fs::remove_file(entry.path()); + remove_manifest_files(dest, id); + } + } + sweep_private_temps(); +} + +/// `{id}-{key}.manifest.json` -> id. +fn state_manifest_id(name: &str) -> Option { + let (id, key) = name.strip_suffix(".manifest.json")?.split_once('-')?; + if key.len() != 16 || !key.bytes().all(|b| b.is_ascii_hexdigit()) { + return None; + } + id.parse().ok() +} + +/// Startup: all scratch JSON is per-attempt and never reused across a process. +fn sweep_private_temps() { + let Ok(entries) = std::fs::read_dir(runtime_root()) else { + return; + }; + for entry in entries.filter_map(|e| e.ok()) { + let Some(name) = entry.file_name().to_str().map(str::to_owned) else { + continue; + }; + if is_private_temp_name(&name) { + let _ = std::fs::remove_file(entry.path()); + } + } +} + +/// `.grab-info-{id}.json` / `.grab-live-{id}.json` (pre-runtime-dir names). +fn is_legacy_info_name(name: &str) -> bool { + let rest = name + .strip_prefix(".grab-info-") + .or_else(|| name.strip_prefix(".grab-live-")); + rest.and_then(|s| s.strip_suffix(".json")) + .is_some_and(|id| !id.is_empty() && id.bytes().all(|b| b.is_ascii_digit())) +} + +/// `info-{id}.json` / `live-{id}.json` in the private scratch root. +fn is_private_temp_name(name: &str) -> bool { + let name = name.strip_suffix(".tmp").unwrap_or(name); + let rest = name + .strip_prefix("info-") + .or_else(|| name.strip_prefix("live-")); + rest.and_then(|s| s.strip_suffix(".json")) + .is_some_and(|id| !id.is_empty() && id.bytes().all(|b| b.is_ascii_digit())) +} + /// Sidecar recording completed parts, so a retry can skip straight to the merge. #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub(crate) struct VideoManifest { @@ -346,6 +565,12 @@ pub(crate) struct VideoManifest { /// pattern is manifest-anchored, not inferred. #[serde(default)] pub(crate) staging_prefix: Option, + /// Destination directory this attempt writes into. The state-dir filename + /// already carries a hash of it; recording it lets the startup sweep find + /// the files without reversing the hash. Absent in older manifests (which + /// still live inside the destination and need no lookup). + #[serde(default)] + pub(crate) dest_dir: Option, } impl VideoManifest { @@ -560,11 +785,8 @@ pub fn sweep_staging_preserving_recordings(staging: &Path, item_id: u64) { } let _ = std::fs::remove_file(path); } - // Remove the manifest itself (canonical + legacy names). - let _ = std::fs::remove_file(manifest_path(staging, item_id)); - for path in legacy_manifest_paths(staging, item_id) { - let _ = std::fs::remove_file(path); - } + // Remove the manifest itself plus the item's single-use probe temps. + remove_manifest_files(staging, item_id); // Never remove_dir: staging is the user's dest dir, not a dedicated subfolder. } @@ -643,15 +865,60 @@ pub(crate) fn collect_sidecar(src: &Path, dest: &Path, lang: &str) { } } +/// Stable identifier for a destination directory: FNV-1a 64 over the resolved +/// absolute path (canonical when it exists, lexical otherwise). Two +/// destinations that reuse an item id get distinct manifests, and a restart +/// that spells the same path differently — a symlinked Downloads dir, a +/// trailing slash, `.`/`..` segments — still finds the same one. +pub(crate) fn dest_key(dest: &Path) -> String { + let normalized = std::fs::canonicalize(dest).unwrap_or_else(|_| lexical_absolute(dest)); + let mut hash = 0xcbf2_9ce4_8422_2325u64; + for byte in normalized.to_string_lossy().as_bytes() { + hash ^= u64::from(*byte); + hash = hash.wrapping_mul(0x0000_0100_0000_01b3); + } + format!("{hash:016x}") +} + +/// Absolute path with `.` dropped and `..` resolved lexically. No filesystem +/// access: the key must stay stable for a destination that does not exist yet. +fn lexical_absolute(path: &Path) -> PathBuf { + let absolute = if path.is_absolute() { + path.to_path_buf() + } else { + std::env::current_dir().unwrap_or_default().join(path) + }; + let mut out = PathBuf::new(); + for component in absolute.components() { + match component { + Component::CurDir => {} + Component::ParentDir => { + if !out.pop() { + out.push(component); + } + } + other => out.push(other.as_os_str()), + } + } + out +} + +/// Persistent manifest location: `$XDG_STATE_HOME/grab/video/{id}-{key}.json`. +/// The key scopes it to one destination dir, so an id reused elsewhere never +/// adopts this row's parts. pub(crate) fn manifest_path(dest_dir: &Path, item_id: u64) -> PathBuf { - // Genuinely hidden: dot at the start of the basename. Internal bookkeeping, - // invisible in normal file views. Just the ID: simple and unique. - dest_dir.join(format!(".{item_id}.manifest.json")) + state_root() + .join("video") + .join(format!("{item_id}-{}.manifest.json", dest_key(dest_dir))) } -/// Previous manifest names, for backward-compatible reads. -fn legacy_manifest_paths(dest_dir: &Path, item_id: u64) -> [PathBuf; 3] { +/// Previous manifest names, all inside the destination dir: the canonical +/// pre-migration name first, then three older variants. They are never read; +/// the startup sweep deletes the files by name so the dest dir stops +/// collecting Grab bookkeeping. +fn legacy_manifest_paths(dest_dir: &Path, item_id: u64) -> [PathBuf; 4] { [ + dest_dir.join(format!(".{item_id}.manifest.json")), dest_dir.join(format!(".grab-{item_id}-manifest.json")), dest_dir.join(format!("grab-{item_id}-.manifest.json")), dest_dir.join(format!("grab-{item_id}-manifest.json")), @@ -669,7 +936,11 @@ struct ManifestOwnership { } fn manifest_ownership(dest_dir: &Path, item_id: u64) -> Option { - let m = read_manifest(dest_dir, item_id)?; + manifest_ownership_of(&read_manifest(dest_dir, item_id)?) +} + +/// Ownership fields of one already-parsed manifest. +fn manifest_ownership_of(m: &VideoManifest) -> Option { let mut exact = Vec::new(); if let Some(base) = &m.staging_name { exact.push(base.clone()); @@ -709,12 +980,23 @@ fn owned_file_names(dest_dir: &Path, ownership: &ManifestOwnership) -> Vec Option { - // Try the hidden path first, then legacy names. - std::fs::read_to_string(manifest_path(dest_dir, item_id)) - .ok() - .or_else(|| { - legacy_manifest_paths(dest_dir, item_id) - .iter() - .find_map(|p| std::fs::read_to_string(p).ok()) - }) - .and_then(|text| serde_json::from_str(&text).ok()) + read_manifest_file(&manifest_path(dest_dir, item_id)) +} + +/// Parse one manifest file. Corrupt or unreadable sidecars read as absent. +fn read_manifest_file(path: &Path) -> Option { + let text = std::fs::read_to_string(path).ok()?; + serde_json::from_str(&text).ok() } pub(crate) fn file_len(path: &Path) -> Option { @@ -765,26 +1053,40 @@ pub(crate) fn file_len(path: &Path) -> Option { /// Write `bytes` to `path` with owner-only permissions (0600), without /// blocking the async executor. /// -/// The probe info-JSON temp files can hold signed media URLs and request -/// headers, so they must not be world-readable. Any pre-existing file is -/// removed first: `mode(0o600)` only applies at creation, and a previous -/// run's crash debris could otherwise keep wider permissions. The open then -/// uses `create_new`: the staging dir sits inside the user's destination -/// folder, which may be shared, and anything planted in the gap after the -/// removal — a symlink, most importantly — must fail the open, not be -/// followed or truncated. Flushes before returning: `write_all` only queues -/// into tokio's buffer, and a backend spawn reading the file right after -/// must see the whole thing. +/// The manifests and probe info-JSON temp files can hold signed media URLs +/// and request headers, so they must not be world-readable. The bytes go to a +/// sibling temp first and are renamed into place: an interrupted write leaves +/// the previous file intact. `mode(0o600)` only applies at creation, so the +/// temp is removed first (previous crash debris) and opened with `create_new`: +/// the state dir can sit below a shared folder, and anything planted at the +/// temp path — a symlink, most importantly — must fail the open, not be +/// followed or truncated. Flushes before renaming: `write_all` only queues +/// into tokio's buffer, and a reader right after must see the whole thing. #[cfg(unix)] pub(crate) async fn write_private(path: &Path, bytes: &[u8]) -> std::io::Result<()> { use tokio::io::AsyncWriteExt as _; - // Remove first so the 0600 mode in `open_new_private` always applies - // (creation only). - let _ = tokio::fs::remove_file(path).await; - let mut file = open_new_private(path).await?; + if let Some(parent) = path.parent().filter(|p| !p.as_os_str().is_empty()) { + ensure_private_dir(parent)?; + } + // Write beside the target and rename: an interrupted write leaves the + // previous file intact instead of truncating it. + let temp = private_temp_path(path); + let _ = tokio::fs::remove_file(&temp).await; + let mut file = open_new_private(&temp).await?; file.write_all(bytes).await?; file.flush().await?; - Ok(()) + drop(file); + tokio::fs::rename(&temp, path).await +} + +/// Sibling path for an atomic write, e.g. `info-1.json.tmp`. +fn private_temp_path(path: &Path) -> PathBuf { + let mut name = path + .file_name() + .map(|n| n.to_os_string()) + .unwrap_or_default(); + name.push(".tmp"); + path.with_file_name(name) } /// Open `path` exclusively for writing with owner-only permissions: if @@ -806,6 +1108,9 @@ async fn open_new_private(path: &Path) -> std::io::Result { #[cfg(not(unix))] pub(crate) async fn write_private(path: &Path, bytes: &[u8]) -> std::io::Result<()> { use tokio::io::AsyncWriteExt as _; + if let Some(parent) = path.parent().filter(|p| !p.as_os_str().is_empty()) { + ensure_private_dir(parent)?; + } let mut file = tokio::fs::File::create(path).await?; file.write_all(bytes).await?; file.flush().await?; @@ -962,6 +1267,27 @@ mod tests { )) } + fn test_manifest(staging_name: &str, dest_dir: Option) -> VideoManifest { + VideoManifest { + page_url: "https://example.com".to_string(), + quality: "1080p".to_string(), + video_format_id: None, + video_ext: "mp4".to_string(), + audio_format_id: String::new(), + audio_ext: String::new(), + final_bytes: None, + staging_name: Some(staging_name.to_string()), + staging_prefix: None, + dest_dir, + } + } + + fn write_state_manifest(dir: &Path, id: u64, manifest: &VideoManifest) { + let path = manifest_path(dir, id); + ensure_private_dir(path.parent().unwrap()).unwrap(); + std::fs::write(path, serde_json::to_string(manifest).unwrap()).unwrap(); + } + /// Regression: info-JSON temp files must be owner-only (0600), even /// when a wider-mode file from a crashed run is already in place. #[tokio::test] @@ -1040,25 +1366,17 @@ mod tests { } #[test] - fn success_path_leaves_no_empty_staging_root() { - // The run_unified_ytdlp success tail: sweep the item's manifest-owned files. - // The dest dir itself is never removed. + fn success_path_leaves_no_bookkeeping_in_the_dest_dir() { + // The runner's success tail: sweep the item's manifest-owned files and + // its state-dir manifest. The dest dir itself is never removed. let staging = unique_dir("success-root"); std::fs::create_dir_all(&staging).unwrap(); - // Create manifest owning the staging file. - let manifest = VideoManifest { - page_url: "https://example.com".to_string(), - quality: "1080p".to_string(), - video_format_id: None, - video_ext: "mp4".to_string(), - audio_format_id: String::new(), - audio_ext: String::new(), - final_bytes: None, - staging_name: Some("Title.42.mp4".to_string()), - staging_prefix: None, - }; - let manifest_json = serde_json::to_string(&manifest).unwrap(); - std::fs::write(staging.join(".42.manifest.json"), manifest_json).unwrap(); + let manifest = test_manifest("Title.42.mp4", Some(staging.clone())); + std::fs::write( + manifest_path(&staging, 42), + serde_json::to_string(&manifest).unwrap(), + ) + .unwrap(); std::fs::write(staging.join("Title.42.mp4.part"), b"scratch").unwrap(); std::fs::write(staging.join("unrelated.txt"), b"keep").unwrap(); sweep_staging_preserving_recordings(&staging, 42); @@ -1066,11 +1384,20 @@ mod tests { !staging.join("Title.42.mp4.part").exists(), "item staging file is swept on success" ); + assert!( + read_manifest(&staging, 42).is_none(), + "state-dir manifest is swept on success" + ); assert!(staging.exists(), "dest dir is never removed"); assert!( staging.join("unrelated.txt").exists(), "unrelated files are never touched" ); + let json: Vec = dir_file_names(&staging) + .into_iter() + .filter(|n| n.ends_with(".json")) + .collect(); + assert!(json.is_empty(), "JSON left in the dest dir: {json:?}"); let _ = std::fs::remove_dir_all(&staging); } @@ -1092,9 +1419,9 @@ mod tests { final_bytes: None, staging_name: Some("Title.44.mp4".to_string()), staging_prefix: None, + dest_dir: None, }; - let manifest_json = serde_json::to_string(&manifest).unwrap(); - std::fs::write(staging.join(".44.manifest.json"), manifest_json).unwrap(); + write_state_manifest(&staging, 44, &manifest); std::fs::write(staging.join("Title.44.mp4.part"), b"scratch").unwrap(); sweep_staging_preserving_recordings(&staging, 44); assert!( @@ -1515,8 +1842,9 @@ mod tests { } #[test] - fn sweep_dest_staging_removes_manifest_owned_file() { - // With a manifest recording staging_name, the exact files are removed. + fn state_sweep_removes_manifest_owned_file() { + // With a state manifest recording staging_name, the exact files are + // removed. let dir = unique_dir("manifest-owned"); std::fs::create_dir_all(&dir).unwrap(); @@ -1530,19 +1858,19 @@ mod tests { final_bytes: None, staging_name: Some("Title.42.mp4".to_string()), staging_prefix: None, + dest_dir: Some(dir.clone()), }; - let manifest_json = serde_json::to_string(&manifest).unwrap(); - std::fs::write(dir.join(".42.manifest.json"), manifest_json).unwrap(); + write_state_manifest(&dir, 42, &manifest); std::fs::write(dir.join("Title.42.mp4.part"), b"staging").unwrap(); - sweep_dest_staging(&dir, &std::collections::HashSet::new()); + sweep_state_staging(&std::collections::HashSet::new()); assert!( !dir.join("Title.42.mp4.part").exists(), "manifest-owned file should be removed" ); assert!( - !dir.join(".42.manifest.json").exists(), + read_manifest(&dir, 42).is_none(), "manifest should be removed after cleanup" ); @@ -1566,9 +1894,9 @@ mod tests { final_bytes: None, staging_name: None, staging_prefix: Some("My Video.42.".to_string()), + dest_dir: Some(dir.clone()), }; - let manifest_json = serde_json::to_string(&manifest).unwrap(); - std::fs::write(dir.join(".42.manifest.json"), manifest_json).unwrap(); + write_state_manifest(&dir, 42, &manifest); std::fs::write(dir.join("My Video.42.f137.mp4.part"), b"part").unwrap(); std::fs::write(dir.join("My Video.42.f251.webm.part"), b"part").unwrap(); std::fs::write(dir.join("My Video.42.mp4.ytdl"), b"meta").unwrap(); @@ -1600,7 +1928,7 @@ mod tests { "user file outside the prefix must survive" ); assert!( - !dir.join(".42.manifest.json").exists(), + read_manifest(&dir, 42).is_none(), "manifest goes after its files" ); @@ -1625,15 +1953,15 @@ mod tests { final_bytes: None, staging_name: None, staging_prefix: None, + dest_dir: Some(dir.clone()), }; - let manifest_json = serde_json::to_string(&manifest).unwrap(); - std::fs::write(dir.join(".42.manifest.json"), manifest_json).unwrap(); + write_state_manifest(&dir, 42, &manifest); std::fs::write(dir.join("orphan.mp4.part"), b"scratch").unwrap(); remove_manifest_owned_files(&dir, 42); assert!( - dir.join(".42.manifest.json").exists(), + read_manifest(&dir, 42).is_some(), "manifest without ownership must survive the sweep" ); assert!( @@ -1670,4 +1998,162 @@ mod tests { let _ = std::fs::remove_dir_all(&staging); } + + /// Fresh profiles have no `grab/video` chain: `write_private` builds it + /// owner-only, and every dir it creates is 0700. + #[tokio::test] + #[cfg(unix)] + async fn write_private_creates_private_parent_chain() { + use std::os::unix::fs::PermissionsExt as _; + let root = unique_dir("write-parent"); + let path = root.join("grab/video/x.json"); + write_private(&path, b"{}").await.unwrap(); + assert_eq!(std::fs::read(&path).unwrap(), b"{}"); + for dir in [&root, &root.join("grab"), &root.join("grab/video")] { + let mode = std::fs::metadata(dir).unwrap().permissions().mode() & 0o777; + assert_eq!(mode, 0o700, "{} must be owner-only", dir.display()); + } + let _ = std::fs::remove_dir_all(&root); + } + + /// An item id reused in a second destination must not read the first + /// destination's manifest: the file name carries a destination key. + #[test] + fn manifest_is_scoped_to_its_destination() { + let dest_a = unique_dir("manifest-scope-a"); + let dest_b = unique_dir("manifest-scope-b"); + std::fs::create_dir_all(&dest_a).unwrap(); + std::fs::create_dir_all(&dest_b).unwrap(); + let id = 8_000_001u64; + let path = manifest_path(&dest_a, id); + let manifest = test_manifest("v.8000001.mp4", Some(dest_a.clone())); + std::fs::write(&path, serde_json::to_string(&manifest).unwrap()).unwrap(); + + assert_eq!(read_manifest(&dest_a, id).as_ref(), Some(&manifest)); + assert!( + read_manifest(&dest_b, id).is_none(), + "an id reused in another destination must not adopt this manifest" + ); + assert_ne!(manifest_path(&dest_a, id), manifest_path(&dest_b, id)); + + let _ = std::fs::remove_file(&path); + let _ = std::fs::remove_dir_all(&dest_a); + let _ = std::fs::remove_dir_all(&dest_b); + } + + /// Startup reclaims state manifests whose `(destination, id)` is gone, + /// deleting exactly the files that manifest owns, and clears stale probe + /// temps while it is there. + #[test] + fn state_sweep_reclaims_orphans_and_keeps_live_rows() { + let live = unique_dir("state-sweep-live"); + let orphan = unique_dir("state-sweep-orphan"); + std::fs::create_dir_all(&live).unwrap(); + std::fs::create_dir_all(&orphan).unwrap(); + let live_id = 8_000_002u64; + let orphan_id = 8_000_003u64; + for (dir, id) in [(&live, live_id), (&orphan, orphan_id)] { + let name = format!("v.{id}.mp4"); + std::fs::write(dir.join(format!("{name}.part")), b"scratch").unwrap(); + let manifest = test_manifest(&name, Some(dir.clone())); + std::fs::write( + manifest_path(dir, id), + serde_json::to_string(&manifest).unwrap(), + ) + .unwrap(); + } + let stale_temp = info_json_path(orphan_id); + ensure_private_dir(stale_temp.parent().unwrap()).unwrap(); + std::fs::write(&stale_temp, b"{}").unwrap(); + + let keep = + std::collections::HashSet::from([(live.to_string_lossy().into_owned(), live_id)]); + sweep_state_staging(&keep); + + assert!( + read_manifest(&live, live_id).is_some(), + "a queued row keeps its resume manifest" + ); + assert!(live.join(format!("v.{live_id}.mp4.part")).exists()); + assert!( + read_manifest(&orphan, orphan_id).is_none(), + "orphan state manifest is reclaimed" + ); + assert!( + !orphan.join(format!("v.{orphan_id}.mp4.part")).exists(), + "the orphan manifest's owned part is reclaimed" + ); + assert!(!stale_temp.exists(), "startup clears stale probe temps"); + + let _ = std::fs::remove_file(manifest_path(&live, live_id)); + let _ = std::fs::remove_dir_all(&live); + let _ = std::fs::remove_dir_all(&orphan); + } + + /// A crash mid-write can leave an unreadable manifest for a queued row. + /// The sweep must not delete it: the row still needs to resume (or at + /// least keep its parts reclaimable) despite the bad sidecar. + #[test] + fn state_sweep_keeps_a_queued_rows_unreadable_manifest() { + let dir = unique_dir("state-sweep-corrupt"); + std::fs::create_dir_all(&dir).unwrap(); + let id = 8_000_005u64; + let path = manifest_path(&dir, id); + ensure_private_dir(path.parent().unwrap()).unwrap(); + std::fs::write(&path, b"{truncated").unwrap(); + let keep = std::collections::HashSet::from([(dir.to_string_lossy().into_owned(), id)]); + + sweep_state_staging(&keep); + assert!( + path.exists(), + "a queued row keeps its manifest even when unreadable" + ); + + sweep_state_staging(&std::collections::HashSet::new()); + assert!( + !path.exists(), + "an orphaned unreadable manifest is reclaimed" + ); + + let _ = std::fs::remove_dir_all(&dir); + } + + /// Discarding a row removes its state manifest and both probe temps, + /// including the old dest-dir names from pre-migration versions. + #[test] + fn clean_staging_files_removes_state_and_probe_temps() { + let dir = unique_dir("clean-state"); + std::fs::create_dir_all(&dir).unwrap(); + let id = 8_000_004u64; + let name = format!("v.{id}.mp4"); + std::fs::write(dir.join(format!("{name}.part")), b"scratch").unwrap(); + std::fs::write(dir.join("keep.txt"), b"user").unwrap(); + let manifest = test_manifest(&name, Some(dir.clone())); + std::fs::write( + manifest_path(&dir, id), + serde_json::to_string(&manifest).unwrap(), + ) + .unwrap(); + ensure_private_dir(&runtime_root()).unwrap(); + let info = info_json_path(id); + let live = live_info_json_path(id); + std::fs::write(&info, b"{}").unwrap(); + std::fs::write(&live, b"{}").unwrap(); + let legacy_info = dir.join(format!(".grab-info-{id}.json")); + let legacy_live = dir.join(format!(".grab-live-{id}.json")); + std::fs::write(&legacy_info, b"{}").unwrap(); + std::fs::write(&legacy_live, b"{}").unwrap(); + + clean_staging_files(&dir, id); + + assert!(!dir.join(format!("{name}.part")).exists()); + assert!(read_manifest(&dir, id).is_none(), "state manifest must go"); + assert!(!info.exists(), "runtime probe temp must go with the row"); + assert!(!live.exists(), "runtime live temp must go with the row"); + assert!(!legacy_info.exists(), "old dest-dir info JSON must go"); + assert!(!legacy_live.exists(), "old dest-dir live JSON must go"); + assert!(dir.join("keep.txt").exists(), "user files stay"); + + let _ = std::fs::remove_dir_all(&dir); + } } diff --git a/src/video_tests.rs b/src/video_tests.rs index f250f569..1309a252 100644 --- a/src/video_tests.rs +++ b/src/video_tests.rs @@ -735,6 +735,7 @@ fn test_manifest() -> VideoManifest { final_bytes: None, staging_name: None, staging_prefix: None, + dest_dir: None, } } @@ -4093,6 +4094,9 @@ fn live_capture_adopts_part_and_remuxes() { let staging = dir.clone(); let mut job = live_test_job(); job.dest = dir.join("v.mp4"); + // Force the probe-info temp through the runner: it must be private scratch, + // never a file beside the finished download. + job.info_json = Some(serde_json::json!({"id": "grabtest"})); let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); let (_abort_tx, abort_rx) = tokio::sync::oneshot::channel::(); // Natural end with a `.part` shell: adopted, remuxed, delivered. @@ -4111,17 +4115,21 @@ fn live_capture_adopts_part_and_remuxes() { )); assert!(matches!(res, Ok(Some(_))), "got {res:?}"); assert_eq!(std::fs::read(&job.dest).unwrap(), b"tsbytes"); - let prefix = format!("grab-{}-", job.item_id); + // The download dir holds only the payload and the test's fake binaries: + // manifest and probe JSON are bookkeeping, kept in the state/runtime roots. + let leftovers: Vec = std::fs::read_dir(&dir) + .unwrap() + .filter_map(|e| e.ok()) + .map(|e| e.file_name().to_string_lossy().into_owned()) + .filter(|n| n.ends_with(".json")) + .collect(); assert!( - !std::fs::read_dir(&dir) - .unwrap() - .filter_map(|e| e.ok()) - .any(|e| e - .file_name() - .to_str() - .map(|n| n.starts_with(&prefix) && !n.ends_with(".manifest.json")) - .unwrap_or(false)), - "staging cleaned" + leftovers.is_empty(), + "JSON left in the download dir: {leftovers:?}" + ); + assert!( + read_manifest(&staging, job.item_id).is_none(), + "delivered live row kept its resume manifest" ); assert!( !dir.join("v.1.mp4.part").exists(), @@ -4919,9 +4927,11 @@ fn sweep_never_touches_bystander_with_id_like_name() { final_bytes: None, staging_name: Some("My Video.mp4".into()), staging_prefix: None, + dest_dir: Some(staging.clone()), }; - let manifest_path = staging.join(".4.manifest.json"); - std::fs::write(&manifest_path, serde_json::to_string(&manifest).unwrap()).unwrap(); + let manifest_file = manifest_path(&staging, 4); + std::fs::create_dir_all(manifest_file.parent().unwrap()).unwrap(); + std::fs::write(&manifest_file, serde_json::to_string(&manifest).unwrap()).unwrap(); std::fs::write(staging.join("My Video.mp4.part"), b"partial").unwrap(); sweep_staging_preserving_recordings(&staging, 4); @@ -4938,7 +4948,7 @@ fn sweep_never_touches_bystander_with_id_like_name() { assert_eq!(std::fs::read(staging.join("x.4.y")).unwrap(), b"bystander"); // Our staging is reclaimed. assert!(!staging.join("My Video.mp4.part").exists()); - assert!(!manifest_path.exists()); + assert!(!manifest_file.exists()); let _ = std::fs::remove_dir_all(&staging); } @@ -9506,6 +9516,9 @@ fn unified_runner_downloads_claims_and_collects() { let mut job = direct_test_job(); job.dest = dir.join("v.mp4"); job.subtitles = Some("en".into()); + // Exercise the probe-info temp path: it must live in private scratch, not + // beside the finished download. + job.info_json = Some(serde_json::json!({"id": "v123"})); let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); let (_abort_tx, mut abort_rx) = tokio::sync::oneshot::channel::(); let gate = AttemptGate::new(); @@ -9531,18 +9544,21 @@ fn unified_runner_downloads_claims_and_collects() { finalizer would never know to reclaim it" ); assert_eq!(std::fs::read(dir.join("v.en.srt")).unwrap(), b"subtitles"); - let prefix = format!("grab-{}-", job.item_id); - assert!( - !std::fs::read_dir(&dir) + // Nothing JSON may remain in the download dirs: the probe temp is runtime + // scratch and the manifest lives under the state root. + for dir in [&dir, &staging] { + let leftovers: Vec = std::fs::read_dir(dir) .unwrap() .filter_map(|e| e.ok()) - .any(|e| e - .file_name() - .to_str() - .map(|n| n.starts_with(&prefix)) - .unwrap_or(false)), - "staging cleaned" - ); + .map(|e| e.file_name().to_string_lossy().into_owned()) + .filter(|n| n.ends_with(".json")) + .collect(); + assert!( + leftovers.is_empty(), + "JSON left in {}: {leftovers:?}", + dir.display() + ); + } let mut progress = false; let mut merging = false; while let Ok(msg) = rx.try_recv() {