diff --git a/src/download.rs b/src/download.rs index 70553cba..946e9154 100644 --- a/src/download.rs +++ b/src/download.rs @@ -758,18 +758,6 @@ impl DownloadManager { id } - /// Fresh id that lands on no existing staging dir: the allocator must skip - /// leftover `grab--*` files in dest dirs and legacy tmp dirs, or a new row - /// would resume (or sweep) a stranger's scratch. - fn alloc_id_for(&self, dest_dir: &std::path::Path) -> u64 { - loop { - let id = self.alloc_id(); - if !crate::video::staging_occupied(dest_dir, id) { - return id; - } - } - } - /// In-memory allocator for callers that listed the destination off GTK. /// Callers must pass a complete listing; an unreadable directory must be /// reported rather than falling back to filesystem I/O on the main thread. @@ -782,20 +770,24 @@ impl DownloadManager { } } - /// Reuse a restored row's persisted id where possible, else allocate. The id is the staging key, so a fresh number would point at the wrong `grab--*` files and risk colliding with a later row. - fn claim_id(&self, stored: Option, dest_dir: &std::path::Path) -> u64 { - let Some(id) = stored else { - return self.alloc_id_for(dest_dir); - }; - // Already owned in this session: a hand-edited/duplicated queue must never share one staging directory. + /// Claim a stored id only if this session has not already assigned it. + fn claim_persisted_id(&self, stored: Option) -> Option { + let id = stored?; + // A hand-edited/duplicated queue must never share one staging identity. if self.find(id).is_some() || self.epoch.borrow().contains_key(&id) { - return self.alloc_id_for(dest_dir); + return None; } - // Keep the allocator ahead of every adopted id so a new row never reuses a restored row's number. + // Keep the allocator ahead of every adopted id so a new row never reuses its number. if id >= self.next_id.get() { self.next_id.set(id + 1); } - id + Some(id) + } + + /// Reuse a persisted id or allocate against a destination snapshot made off GTK. + fn claim_id_in(&self, stored: Option, existing: &[String]) -> u64 { + self.claim_persisted_id(stored) + .unwrap_or_else(|| self.alloc_id_for_in(existing)) } /// The underlying download list. @@ -1375,12 +1367,21 @@ impl DownloadManager { (added, pl.total, skipped_by_host_limit) } - /// Re-queue one persisted entry, preserving its intent (paused/failed stay). Names restore verbatim to keep resume identity. - /// The video-page source is staged *before* insert (see `enqueue_video`). + /// Synchronous restore helper for tests; production passes a prelisted destination. /// /// # Errors /// Returns a display-ready message when the stored entry is invalid. + #[cfg(test)] pub fn restore_existing(self: &Rc, stored: &StoredItem) -> Result { + let existing = crate::video_staging::dir_file_names(std::path::Path::new(&stored.dest_dir)); + self.restore_existing_with_listing(stored, &existing) + } + + fn restore_existing_with_listing( + self: &Rc, + stored: &StoredItem, + existing: &[String], + ) -> Result { let StoredItem { id: stored_id, url: raw_url, @@ -1406,7 +1407,7 @@ impl DownloadManager { } let filename = shorten_filename(raw_filename); let item = DownloadItem::new( - self.claim_id(*stored_id, std::path::Path::new(dest_dir)), + self.claim_id_in(*stored_id, existing), &url, &filename, dest_dir, @@ -1463,12 +1464,11 @@ impl DownloadManager { Ok(self.insert(item)) } - /// Restore a gallery row from a validated stored entry + parsed job. - /// Mirrors `enqueue_gallery` but reuses the stored id and status. - pub fn restore_existing_gallery( + fn restore_existing_gallery_with_listing( self: &Rc, stored: &StoredItem, job: Option, + existing: &[String], ) -> Result { let mut job = job.ok_or_else(|| "gallery entry missing job".to_string())?; let url = normalize_url(&stored.url)?; @@ -1485,7 +1485,7 @@ impl DownloadManager { // load; use it verbatim so the restored row matches the on-disk // subfolder instead of the payload's base title. let title = stored.filename.clone(); - let id = self.claim_id(stored.id, std::path::Path::new(&stored.dest_dir)); + let id = self.claim_id_in(stored.id, existing); job.item_id = id; let item = DownloadItem::new(id, &url, &title, &stored.dest_dir); item.set_public_only(public_only); @@ -1593,7 +1593,12 @@ impl DownloadManager { }); } - fn insert_history(self: &Rc, stored: StoredItem, output_dir: Option) { + fn insert_history( + self: &Rc, + stored: StoredItem, + output_dir: Option, + existing: &[String], + ) { let StoredItem { id: stored_id, url: raw_url, @@ -1615,12 +1620,7 @@ impl DownloadManager { tracing::warn!("skipping history entry with relative destination"); return; } - let item = DownloadItem::new( - self.claim_id(stored_id, std::path::Path::new(&dir)), - &url, - &name, - &dir, - ); + let item = DownloadItem::new(self.claim_id_in(stored_id, existing), &url, &name, &dir); item.set_progress(progress.clamp(0.0, 1.0)); item.set_status(DownloadStatus::Done); // Already trust-checked by the caller (must sit inside `dir`). @@ -4315,6 +4315,46 @@ impl DownloadManager { }) } + /// Find restore rows whose id allocation may need a staging-name scan. + /// Normal rows with unique persisted ids need no directory listing. + fn restore_dirs_requiring_id_listing(&self, pending: &[PendingRestore]) -> HashSet { + let mut used_ids: HashSet = self.items().map(|item| item.id()).collect(); + used_ids.extend(self.epoch.borrow().keys().copied()); + pending + .iter() + .filter_map(|entry| match entry.item.id { + Some(id) if used_ids.insert(id) => None, + _ => Some(entry.item.dest_dir.clone()), + }) + .collect() + } + + /// Read destination directories needed for fallback IDs. + /// + /// Missing destinations are empty; other listing failures are returned so + /// an ID is never selected from an incomplete directory snapshot. + /// Production callers run this on a blocking worker, not on GTK. + fn list_restore_destinations( + directories: HashSet, + ) -> std::io::Result>> { + directories + .into_iter() + .map(|dir| { + let path = std::path::Path::new(&dir); + let names = crate::video_staging::try_dir_file_names(path).map_err(|error| { + std::io::Error::new( + error.kind(), + format!( + "cannot list restore destination {}: {error}", + path.display() + ), + ) + })?; + Ok((dir, names)) + }) + .collect() + } + /// Read and validate the saved queue away from GTK's main context. /// Only ListStore/model mutation is applied on the main context; disk sweeps /// are sent to Tokio's blocking pool before restore becomes Ready. @@ -4328,7 +4368,18 @@ impl DownloadManager { // the same sweep implementation with production. let pending = Self::load_restore_queue(queue_file); if let Some(pending) = pending { - let plan = self.apply_restored_queue(pending); + let directories = self.restore_dirs_requiring_id_listing(&pending); + let existing_by_dir = match Self::list_restore_destinations(directories) { + Ok(existing) => existing, + Err(error) => { + tracing::error!( + "download queue destination listing failed: {error}; preserving the saved queue" + ); + self.finish_restore(RestoreOutcome::Failed); + return; + } + }; + let plan = self.apply_restored_queue(pending, &existing_by_dir); let keep = Self::run_restore_sweeps(plan); Self::schedule_session_orphan_sweep(keep); } @@ -4358,8 +4409,40 @@ impl DownloadManager { match loaded { Ok(Some(pending)) => { manager.restore_dirty.set(false); + let directories = manager.restore_dirs_requiring_id_listing(&pending); + let prepared = tokio_rt() + .spawn_blocking(move || { + Self::list_restore_destinations(directories) + .map(|existing_by_dir| (pending, existing_by_dir)) + }) + .await; + let (pending, existing_by_dir) = match prepared { + Ok(Ok(prepared)) => prepared, + Ok(Err(error)) => { + tracing::error!( + "download queue destination listing failed: {error}; preserving the saved queue" + ); + manager.finish_restore(RestoreOutcome::Failed); + completion.finished = true; + return; + } + Err(error) => { + tracing::error!( + "download queue destination listing task failed: {error}; preserving the saved queue" + ); + manager.finish_restore(RestoreOutcome::Failed); + completion.finished = true; + return; + } + }; + // The main context yielded while legacy/duplicate ids were checked. + if manager.restore_should_abort() { + manager.finish_restore(RestoreOutcome::Failed); + completion.finished = true; + return; + } let plan = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { - manager.apply_restored_queue(pending) + manager.apply_restored_queue(pending, &existing_by_dir) })); let plan = match plan { Ok(plan) => plan, @@ -4529,24 +4612,36 @@ impl DownloadManager { /// Apply validated rows to GTK's model and snapshot cleanup keep-sets. /// Filesystem traversal/deletion happens later on a blocking worker. - fn apply_restored_queue(self: &Rc, pending: Vec) -> RestoreSweepPlan { + fn apply_restored_queue( + self: &Rc, + pending: Vec, + existing_by_dir: &HashMap>, + ) -> RestoreSweepPlan { let mut selections = Vec::new(); { let _batch = self.batch_guard(); - // JSON parsing and per-row validation completed off-thread. + // JSON parsing, validation and any necessary listings completed off-thread. // GObject construction and model mutation stay on this context. for p in pending { + let existing = existing_by_dir + .get(&p.item.dest_dir) + .map(Vec::as_slice) + .unwrap_or(&[]); match p.item.status { DownloadStatus::Done => { if p.gallery_job.is_some() { let mut restored_item = p.item.clone(); restored_item.segments = p.segments; - match self.restore_existing_gallery(&restored_item, p.gallery_job) { + match self.restore_existing_gallery_with_listing( + &restored_item, + p.gallery_job, + existing, + ) { Ok(_) => {} Err(e) => tracing::warn!("skipping queue entry: {e}"), } } else { - self.insert_history(p.item, p.output_dir); + self.insert_history(p.item, p.output_dir, existing); } } status => { @@ -4561,9 +4656,13 @@ impl DownloadManager { restored_item.status = status; restored_item.segments = p.segments; let result = if let Some(job) = p.gallery_job { - self.restore_existing_gallery(&restored_item, Some(job)) + self.restore_existing_gallery_with_listing( + &restored_item, + Some(job), + existing, + ) } else { - self.restore_existing(&restored_item) + self.restore_existing_with_listing(&restored_item, existing) }; match result { Ok(restored) => { diff --git a/src/download_tests.rs b/src/download_tests.rs index b8edfbf9..4aecfc1a 100644 --- a/src/download_tests.rs +++ b/src/download_tests.rs @@ -220,6 +220,81 @@ fn test_port(offset: u16) -> u16 { % 40000) as u16 } +#[test] +fn legacy_restore_id_allocation_skips_prelisted_staging_names() { + let settings = test_settings(); + let manager = DownloadManager::new(gio::ListStore::new::(), settings); + manager.next_id.set(7); + let existing = vec!["Title.7.mp4.part".to_string()]; + + assert_eq!(manager.claim_id_in(None, &existing), 8); +} + +#[test] +fn restore_prepares_destination_listings_only_for_fallback_ids() { + let settings = test_settings(); + let manager = DownloadManager::new(gio::ListStore::new::(), settings); + let legacy = PendingRestore { + item: stored_row( + "https://example.test/file.mp4", + "/tmp/grab-restore-legacy", + "file.mp4", + DownloadStatus::Paused, + ), + output_dir: None, + segments: None, + gallery_job: None, + }; + assert_eq!( + manager.restore_dirs_requiring_id_listing(std::slice::from_ref(&legacy)), + std::collections::HashSet::from(["/tmp/grab-restore-legacy".to_string()]) + ); + + let mut current = legacy; + current.item.id = Some(7); + assert!( + manager + .restore_dirs_requiring_id_listing(std::slice::from_ref(¤t)) + .is_empty() + ); + + let mut duplicate = PendingRestore { + item: stored_row( + "https://example.test/duplicate.mp4", + "/tmp/grab-restore-duplicate", + "duplicate.mp4", + DownloadStatus::Paused, + ), + output_dir: None, + segments: None, + gallery_job: None, + }; + duplicate.item.id = Some(7); + assert_eq!( + manager.restore_dirs_requiring_id_listing(&[current, duplicate]), + std::collections::HashSet::from(["/tmp/grab-restore-duplicate".to_string()]) + ); +} + +#[test] +fn restore_destination_listing_reports_unreadable_directories() { + let path = std::env::temp_dir().join(format!( + "grab-restore-not-a-directory-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |duration| duration.as_nanos()), + )); + std::fs::write(&path, b"not a directory").expect("test fixture should be created"); + let directory = path.to_string_lossy().into_owned(); + let result = + DownloadManager::list_restore_destinations(std::collections::HashSet::from([directory])); + let _ = std::fs::remove_file(&path); + + let error = result.expect_err("a file cannot be listed as a destination directory"); + assert!(error.to_string().contains(&path.display().to_string())); +} + #[test] fn backoff_delay_stays_within_cap() { // Bound-only: deterministic, never flakes. Do NOT assert distinctness (jitter is random). @@ -2621,7 +2696,9 @@ fn unremove_restores_started_for_single_connection_partial() { // A row the engine started (single-connection: no segment bitmap). // Inserted Paused so nothing spawns before the snapshot; flipped to // Downloading to mirror a live row at remove time. - let id = manager.alloc_id_for(std::path::Path::new(&dest_s)); + let existing = crate::video_staging::try_dir_file_names(std::path::Path::new(&dest_s)) + .expect("test destination should be readable"); + let id = manager.alloc_id_for_in(&existing); let item = DownloadItem::new( id, "https://example.com/partial.bin", diff --git a/src/video.rs b/src/video.rs index 4897e85a..8a7ae6de 100644 --- a/src/video.rs +++ b/src/video.rs @@ -40,9 +40,7 @@ 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_occupied, staging_root, sweep_dest_staging, -}; +pub use crate::video_staging::{clean_dest_parts, staging_root, sweep_dest_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_staging.rs b/src/video_staging.rs index 01bf1a28..9b72ef25 100644 --- a/src/video_staging.rs +++ b/src/video_staging.rs @@ -109,19 +109,8 @@ pub(crate) fn staging_name_matches_id(file_name: &str, item_id: u64) -> bool { false } -/// Whether any staging file exists for this id: the id allocator must skip -/// it so a fresh row never lands on a leftover. -pub fn staging_occupied(dest_dir: &Path, item_id: u64) -> bool { - // Manifest exists = occupied (it tracks the exact staging name). The - // listing check also sees it when the directory is readable. - if manifest_path(dest_dir, item_id).exists() { - return true; - } - staging_occupied_in(&dir_file_names(dest_dir), item_id) -} - -/// In-memory form of [`staging_occupied`] for callers that already listed the -/// destination off the GTK thread. +/// In-memory check for staging files associated with an item id, using an +/// already-complete destination listing made off the GTK thread. pub(crate) fn staging_occupied_in(names: &[String], item_id: u64) -> bool { // Id-in-name scheme: the file itself carries the id, so a lost manifest // cannot orphan the id into reuse. @@ -475,8 +464,8 @@ pub(crate) fn is_grab_part(file_name: &str, stem: &str) -> bool { // Legacy ID-namespaced live: {id}.live. (e.g., "1.live.mp4.part"). // The bare {id}. pattern is NOT checked here: it false-positives on user // files like Movie.2024.mp4. Pattern-based deletion of id-in-name files - // is unsafe (no marker); only `staging_occupied` uses the pattern, and - // only conservatively. + // is unsafe (no marker); the ID allocator only checks these patterns + // conservatively. if let Some(dot_pos) = remainder.find('.') { let (id_part, rest) = remainder.split_at(dot_pos); if !id_part.is_empty() @@ -604,7 +593,7 @@ fn stem_has_subtitle_sidecar(names: &[String], stem: &str) -> bool { /// Only legacy marked patterns (`PART_KINDS`, `.live.`) are matched: the /// id-in-name scheme has no marker, so pattern-deleting it would risk user /// files like `Movie.2024.mp4.part`. ID reuse is prevented separately by -/// `staging_occupied`. +/// the allocator's `staging_occupied_in` check. pub fn clean_dest_parts(dest: &Path) { let (Some(dir), Some(stem)) = (dest.parent(), dest.file_stem().and_then(|s| s.to_str())) else { return; @@ -1338,11 +1327,11 @@ mod tests { // No manifest written. assert!( - staging_occupied(&dir, 7), + staging_occupied_in(&dir_file_names(&dir), 7), "id 7 must stay occupied via id-in-name even with no manifest" ); assert!( - !staging_occupied(&dir, 8), + !staging_occupied_in(&dir_file_names(&dir), 8), "unrelated id 8 must not be occupied" ); let _ = std::fs::remove_dir_all(&dir); @@ -1380,7 +1369,7 @@ mod tests { // Without a manifest, id-in-name sidecars are NOT swept by pattern: // without a marker, `Title.7.mp4.part` is indistinguishable from a // user's own file. Deletion requires the manifest's exact name. - // ID reuse is still prevented by `staging_occupied`. + // ID reuse is still prevented by the allocator's `staging_occupied_in` check. let dir = unique_dir("manifest-loss-sweep"); std::fs::create_dir_all(&dir).unwrap(); std::fs::write(dir.join("Title.7.mp4.part"), b"partial").unwrap(); @@ -1424,7 +1413,7 @@ mod tests { "corrupt manifest must read as None" ); assert!( - staging_occupied(&dir, 7), + staging_occupied_in(&dir_file_names(&dir), 7), "id 7 must stay occupied via id-in-name despite corrupt manifest" ); let _ = std::fs::remove_dir_all(&dir); @@ -1440,8 +1429,14 @@ mod tests { std::fs::write(dir.join("Title.4.mp4.part"), b"row4").unwrap(); std::fs::write(dir.join("Title.9.mp4.part"), b"row9").unwrap(); - assert!(staging_occupied(&dir, 4), "id 4 occupied"); - assert!(staging_occupied(&dir, 9), "id 9 occupied"); + assert!( + staging_occupied_in(&dir_file_names(&dir), 4), + "id 4 occupied" + ); + assert!( + staging_occupied_in(&dir_file_names(&dir), 9), + "id 9 occupied" + ); // Without a manifest, the sweep must NOT pattern-delete: Title.4.mp4.part // is indistinguishable from a user's own file. Deletion requires the