diff --git a/src/application.rs b/src/application.rs index 824e30e91..41c7dfdde 100644 --- a/src/application.rs +++ b/src/application.rs @@ -157,7 +157,10 @@ async fn intake_remote_torrent( Routed::OpenedCard } Ok(_) => { - if let Err(e) = manager.enqueue_torrent_file(bytes, &stem, None, None) { + if let Err(e) = manager + .enqueue_torrent_file_async(bytes, &stem, None, None) + .await + { fail(e); return Routed::Failed; } @@ -560,8 +563,9 @@ fn handle_open_files(s: Rc, files: Vec) { add_card.open_torrent_picker(file_name, bytes, entries); } Ok(_) => { - if let Err(e) = - manager.enqueue_torrent_file(bytes, &stem, None, None) + if let Err(e) = manager + .enqueue_torrent_file_async(bytes, &stem, None, None) + .await { toasts.add_toast(adw::Toast::new(&crate::ui_util::esc_markup( &e, diff --git a/src/download.rs b/src/download.rs index 474401817..923c1e624 100644 --- a/src/download.rs +++ b/src/download.rs @@ -278,6 +278,10 @@ pub struct DownloadManager { store: gio::ListStore, settings: crate::settings::AppSettings, running: RefCell>>, + /// Rows whose filesystem preparation is in flight; these consume scheduler slots too. + preparing: RefCell>, + /// Filesystem-preparation task handles; abort them on manager drop so thread-bound GLib futures cannot leak into another test's context. + preparation_handles: RefCell>>, /// Pump task handles by download ID, so tests can wait for pump completion (a `spawn_future_local` task reaped on another thread trips glib's thread guard). pub(crate) pump_handles: RefCell>>, next_id: Cell, @@ -350,7 +354,7 @@ pub struct DownloadManager { /// sole writer, which makes writes ordered by construction. /// Tests write synchronously and never use the channel. #[cfg_attr(test, allow(dead_code))] - persist_tx: RefCell>>>, + persist_tx: RefCell>>>, /// Handle for the queue persist task; awaited on shutdown after the final /// state is sent. Tests never spawn it. #[cfg_attr(test, allow(dead_code))] @@ -443,10 +447,12 @@ impl Drop for RestoreTaskGuard { impl Drop for DownloadManager { fn drop(&mut self) { - // Abort any pump tasks: they're `spawn_future_local`, so a stale one - // reaped on another thread trips glib's thread guard. Tests wait for - // pump completion via `drain_engine` (which checks `pump_handles`), - // so in practice none are pending here; this is a backstop. + // Abort thread-bound local tasks on their owning thread. Tests and app + // shutdown normally drain them; this backstop prevents an unfinished + // preparation/pump from being reaped by another test's context thread. + for (_, handle) in self.preparation_handles.borrow_mut().drain() { + handle.abort(); + } for (_, handle) in self.pump_handles.borrow_mut().drain() { handle.abort(); } @@ -569,9 +575,11 @@ impl DownloadManager { // so the last persist completes before shutdown awaits the task. #[cfg(not(test))] let (persist_tx, persist_task) = { - let (tx, rx) = tokio::sync::watch::channel(None::); - let task = tokio_rt().spawn(Self::persist_task_loop(rx, |text: String| { - Self::write_queue_file(&text) + let (tx, rx) = tokio::sync::watch::channel(None::); + let task = tokio_rt().spawn(Self::persist_task_loop(rx, |queue: StoredQueue| { + if let Some(text) = Self::serialize_snapshot(queue) { + Self::write_queue_file(&text); + } })); (Some(tx), Some(task)) }; @@ -583,6 +591,8 @@ impl DownloadManager { store, settings, running: RefCell::new(HashMap::new()), + preparing: RefCell::new(HashSet::new()), + preparation_handles: RefCell::new(HashMap::new()), pump_handles: RefCell::new(HashMap::new()), next_id: Cell::new(1), on_change: RefCell::new(None), @@ -814,6 +824,34 @@ impl DownloadManager { } } + /// Delete a segmented partial away from GTK. Reserve its exact path until + /// unlink completes so an immediate retry cannot recreate it mid-cleanup. + fn schedule_partial_cleanup(self: &Rc, path: std::path::PathBuf) { + self.reserve_dest(&path); + let weak = Rc::downgrade(self); + glib::spawn_future_local(async move { + let worker_path = path.clone(); + match gio::spawn_blocking(move || std::fs::remove_file(worker_path)).await { + Ok(Ok(())) => {} + Ok(Err(e)) if e.kind() == std::io::ErrorKind::NotFound => {} + Ok(Err(e)) => tracing::warn!( + path = %path.display(), + error = %e, + "could not remove cancelled segmented partial" + ), + Err(e) => tracing::warn!( + path = %path.display(), + error = ?e, + "partial cleanup worker failed" + ), + } + if let Some(manager) = weak.upgrade() { + manager.release_dest(&path); + manager.start_next(); + } + }); + } + /// Whether the row is currently capturing a live stream (stop-and-keep applies). pub fn is_live_video(&self, id: u64) -> bool { self.live_rows.borrow().contains(&id) @@ -891,29 +929,65 @@ impl DownloadManager { /// /// # Errors /// Returns a display-ready message when the bytes are not a valid torrent. - pub fn enqueue_torrent_file( + /// Archive and enqueue a .torrent without performing disk I/O on GTK. + pub async fn enqueue_torrent_file_async( self: &Rc, bytes: Vec, file_name: &str, dest_dir: Option<&str>, only_files: Option>, ) -> Result { - let pseudo = crate::torrent::archive_torrent_file(file_name, &bytes)?; - if let Some(sel) = only_files { - crate::torrent::stage_selection(&pseudo, sel); + let (item, guard) = self + .enqueue_torrent_file_async_batched(bytes, file_name, dest_dir, only_files) + .await?; + drop(guard); + Ok(item) + } + + pub(crate) async fn enqueue_torrent_file_async_batched( + self: &Rc, + bytes: Vec, + file_name: &str, + dest_dir: Option<&str>, + only_files: Option>, + ) -> Result<(DownloadItem, BatchGuard), String> { + let file_name = file_name.to_string(); + let dir = self.resolve_dir(dest_dir); + let worker_file_name = file_name.clone(); + let worker_dir = dir.clone(); + let (pseudo, base, existing) = gio::spawn_blocking(move || { + // Parse metadata before archiving so invalid input never leaves an archive behind. + let base = crate::torrent::intake_plan(&bytes).map(|(base, _multi)| base); + let pseudo = crate::torrent::archive_torrent_file(&worker_file_name, &bytes)?; + let existing = crate::video_staging::dir_file_names(std::path::Path::new(&worker_dir)); + Ok::<_, String>((pseudo, base, existing)) + }) + .await + .map_err(|error| format!("Could not prepare torrent file: {error:?}"))??; + + // Hold the queue only during the synchronous commit, not during disk I/O. + let guard = self.batch_guard(); + if let Some(selection) = only_files { + crate::torrent::stage_selection(&pseudo, selection); } - let stub = crate::torrent::stub_name_for_file(file_name); + let stub = crate::torrent::stub_name_for_file(&file_name); let item = self.enqueue(&pseudo, dest_dir, Some(&stub))?; - // The engine writes with overwrite:true, so record a deduped subfolder the engine then uses as-is. - if let Some((base, _)) = crate::torrent::intake_plan(&bytes) { - let dir = self.resolve_dir(dest_dir); - if std::path::Path::new(&dir).join(&base).exists() { - let name = dedupe_filename(&base, |n| { - let p = std::path::Path::new(&dir).join(n); - p.exists() - // Only video-row finished names reserve stems (see plain `enqueue`). - || self.items() - .any(|it| it.output_dir() == p.to_string_lossy()) + + // The engine writes with overwrite:true, so reserve a collision-free + // output folder from the listing snapshot before it can be spawned. + if let Some(base) = base { + let candidate = std::path::Path::new(&dir).join(&base); + let candidate_s = candidate.to_string_lossy().into_owned(); + let disk_taken = existing.iter().any(|entry| entry == &base); + let row_taken = self.items().any(|row| row.output_dir() == candidate_s); + if disk_taken || row_taken || self.dest_reserved(&candidate) { + let name = dedupe_filename(&base, |name| { + let path = std::path::Path::new(&dir).join(name); + existing.iter().any(|entry| entry == name) + || self.dest_reserved(&path) + || self + .items() + .any(|row| row.output_dir() == path.to_string_lossy()) }); item.set_output_dir( std::path::Path::new(&dir) @@ -923,7 +997,7 @@ impl DownloadManager { ); } } - Ok(item) + Ok((item, guard)) } /// Final filename policy, shared by intake naming and late server/container suggestions: shorten, then the opt-in ASCII fold. Dedupe stays at the call sites. @@ -944,45 +1018,83 @@ impl DownloadManager { /// /// # Errors /// Returns a display-ready message when the URL or filename is invalid. - pub fn enqueue_video( + /// Enqueue a video page after preparing its directory listing off GTK. + #[cfg(test)] + pub async fn enqueue_video_async( self: &Rc, page_url: &str, dest_dir: Option<&str>, filename: Option<&str>, choices: crate::media_types::VideoChoices, ) -> Result { - // One readdir per intake. + let (item, guard) = self + .enqueue_video_async_batched(page_url, dest_dir, filename, choices) + .await?; + drop(guard); + Ok(item) + } + + pub(crate) async fn enqueue_video_async_batched( + self: &Rc, + page_url: &str, + dest_dir: Option<&str>, + filename: Option<&str>, + choices: crate::media_types::VideoChoices, + ) -> Result<(DownloadItem, BatchGuard), String> { let dir = self.resolve_dir(dest_dir); - let existing = crate::video_staging::dir_file_names(std::path::Path::new(&dir)); - self.enqueue_video_staged(page_url, &dir, filename, choices, &existing) + let listing_dir = dir.clone(); + let existing = gio::spawn_blocking(move || { + crate::video_staging::dir_file_names(std::path::Path::new(&listing_dir)) + }) + .await + .map_err(|error| format!("Could not inspect download folder: {error:?}"))?; + let guard = self.batch_guard(); + let item = self.enqueue_video_staged(page_url, &dir, filename, choices, &existing)?; + Ok((item, guard)) } /// Enqueue a gallery download (image gallery via gallery-dl). /// Creates a DownloadItem with gallery flags set; the actual download /// is driven by the gallery runner (phase 12). - pub fn enqueue_gallery( + /// Enqueue a gallery after listing its destination off GTK. + #[cfg(test)] + pub async fn enqueue_gallery_async( self: &Rc, job: &crate::gallery_argv::GalleryJob, ) -> Result { + let (item, guard) = self.enqueue_gallery_async_batched(job).await?; + drop(guard); + Ok(item) + } + + pub(crate) async fn enqueue_gallery_async_batched( + self: &Rc, + job: &crate::gallery_argv::GalleryJob, + ) -> Result<(DownloadItem, BatchGuard), String> { let url = normalize_url(&job.page_url)?; - // Resolve the destination: empty/"." fall back to the download dir; - // relative paths are rejected (would resolve against the launcher CWD). + // Resolve the destination before recording it: empty/relative paths must never + // become dependent on the launcher CWD. The sanitized title is used as a + // restore-safe folder component, not as an arbitrary path from metadata. let dir = self.resolve_dir(Some(&job.dest_dir.to_string_lossy())); - // Gallery title is the folder name; sanitize it (probe falls back to - // the URL, which contains `/` and would fail restore validation). let title = crate::file_names::sanitize_folder_name(&job.title); - // Ensure the gallery directory is unique and provably ours: if a - // directory with the sanitized title already exists on disk - // (pre-existing user folder), or another gallery row already claims - // that name (same title enqueued twice, neither downloaded yet), - // append -2, -3, etc. until a free name is found. This prevents the - // delete path from trashing unrelated files. + let listing_dir = dir.clone(); + let existing = gio::spawn_blocking(move || { + crate::video_staging::dir_file_names(std::path::Path::new(&listing_dir)) + }) + .await + .map_err(|error| format!("Could not inspect gallery destination: {error:?}"))?; + + // Only hold the queue during the synchronous commit, never across listing I/O. + let guard = self.batch_guard(); + // Only claim a folder that this row can prove is new. A pre-existing user + // directory or another queued gallery must never become a trash target. let mut unique_title = title.clone(); let mut counter = 1; - while std::path::Path::new(&dir).join(&unique_title).exists() + while existing.iter().any(|entry| entry == &unique_title) || self .items() .any(|it| it.dest_dir() == dir && it.filename() == unique_title) + || self.dest_reserved(&std::path::Path::new(&dir).join(&unique_title)) { counter += 1; unique_title = format!("{title}-{counter}"); @@ -998,15 +1110,12 @@ impl DownloadManager { item.set_gallery_files_done(0); item.set_gallery_url(url.clone()); item.set_detail(format!("Gallery • {} files", job.file_count.unwrap_or(0))); - // Gallery files go into a subfolder named after the (unique) title. let gallery_dir = std::path::Path::new(&dir).join(&unique_title); - // Store a resolved copy: absolute dest_dir (the gallery subfolder) - // and the real row ID. (The caller's job keeps its placeholder ID.) let mut stored_job = job.clone(); stored_job.dest_dir = gallery_dir; stored_job.item_id = item.id(); self.gallery_jobs.borrow_mut().insert(item.id(), stored_job); - Ok(self.insert(item)) + Ok((self.insert(item), guard)) } /// [`DownloadManager::enqueue_video`] with a precomputed directory listing, @@ -1065,13 +1174,18 @@ impl DownloadManager { } /// Queue one row per playlist entry for collection-shaped resolves with no picked entry. Returns (added, reported total). One persist for the whole import. - fn expand_playlist_rows( - self: &Rc, + /// Queue collection children after creating/listing their destination off GTK. + async fn expand_playlist_rows_async( + weak: std::rc::Weak, id: u64, + generation: u64, item: &DownloadItem, pl: &crate::media_types::PlaylistInfo, ) -> (usize, usize, usize) { - let (quality, audio_only) = match self.video_source(id) { + let Some(this) = weak.upgrade() else { + return (0, pl.total, 0); + }; + let (quality, audio_only) = match this.video_source(id) { Some(crate::media_types::VideoSource::Page { quality, audio_only, @@ -1079,31 +1193,67 @@ impl DownloadManager { }) => (quality, audio_only), _ => return (0, pl.total, 0), }; - let dest_dir = self.resolve_dir(Some(&item.dest_dir())); - let _batch = self.batch_guard(); - // Multi-item collections land in a titled subfolder, torrent-style; a - // lone item keeps the flat behavior. - let dest_dir = if pl.total > 1 { - crate::file_names::collection_subdir(&dest_dir, &pl.title) - } else { - dest_dir + let base_dir = this.resolve_dir(Some(&item.dest_dir())); + let multi = pl.total > 1; + let title = pl.title.clone(); + // The local task must not keep a strong manager Rc pinned while the + // directory worker is pending; upgrade again before touching queue state. + drop(this); + let prepared = gio::spawn_blocking(move || { + let dest_dir = if multi { + crate::file_names::collection_subdir(&base_dir, &title) + } else { + base_dir + }; + let existing = crate::video_staging::dir_file_names(std::path::Path::new(&dest_dir)); + (dest_dir, existing) + }) + .await; + let (dest_dir, existing) = match prepared { + Ok(prepared) => prepared, + Err(error) => { + tracing::warn!( + error = ?error, + parent_id = id, + "playlist import directory preparation failed" + ); + return (0, pl.total, 0); + } + }; + + let Some(this) = weak.upgrade() else { + return (0, pl.total, 0); }; - // The shared network-policy helper enforces a collection-local host - // cap for untrusted browser handoffs and remembers accepted hosts. + // The main context yielded while the folder was prepared. Do not + // enqueue stale children after pause/cancel/remove. + // Collection metadata is untrusted. Re-check the carrier after yielding so + // removed/cancelled rows cannot create children from stale resolver output. + if !this.is_current(id, generation) { + return (0, pl.total, 0); + } + let Some(current) = this.find(id) else { + return (0, pl.total, 0); + }; + if matches!( + current.status(), + DownloadStatus::Cancelled | DownloadStatus::Paused + ) { + return (0, pl.total, 0); + } + + let _batch = this.batch_guard(); + // Collection results are untrusted input. Keep policy inheritance and the + // distinct-host cap in the shared helper so automatic expansion and the picker + // enforce the same destination/network rules. let mut guarded_child_hosts = HashSet::new(); let mut skipped_by_host_limit = 0usize; - // One readdir for the whole expansion instead of one per entry. - let existing = crate::video_staging::dir_file_names(std::path::Path::new(&dest_dir)); - let mut added = 0; + let mut added = 0usize; for entry in &pl.items { let Some(url) = crate::video_probe::expand_child_target(&item.url(), &pl.page_url, entry) else { continue; }; - // Collection results are untrusted input. Keep the shared helper - // responsible for both policy inheritance and the distinct-host - // limit, so picker and automatic expansion behave identically. match crate::download_net::inherit_public_resolution_for_collection( &item.url(), &url, @@ -1116,13 +1266,13 @@ impl DownloadManager { } Err(crate::download_net::CollectionHostError::UnsafeDestination) => continue, } - // Match the interactive picker: carry each entry's title into its - // output filename, falling back to an entry URL slug when needed. - let remux = crate::video_prefs::remux_video_active(&self.settings.remux_video()); + // Match the interactive picker by carrying each entry title into its filename. + let remux = crate::video_prefs::remux_video_active(&this.settings.remux_video()); let name = crate::video::default_video_filename(&entry.title, audio_only, remux.as_deref()); - // Live streams queued from a playlist take the VOD path; each child re-resolves its own page. - if self + // Live streams discovered inside a collection use the VOD path; each child + // resolves its own page rather than inheriting the carrier's live-capture state. + if this .enqueue_video_staged( &url, &dest_dir, @@ -1141,8 +1291,8 @@ impl DownloadManager { added += 1; } } - // Reported total, not attempted ("500 of 600"). Include a separate - // count for the security cap so the carrier row can explain omissions. + // Report the collection's stated total, not merely attempted entries, so the + // carrier can accurately show "added of total" alongside host-cap omissions. (added, pl.total, skipped_by_host_limit) } @@ -1452,7 +1602,7 @@ impl DownloadManager { self.discards .borrow_mut() .retain(|_, pending| !pending.finalizer.is_finished()); - while self.running.borrow().len() < self.max_concurrent() { + while self.running.borrow().len() + self.preparing.borrow().len() < self.max_concurrent() { let next = self .items() // A discard still tearing down owns its destination: the row stays Queued until the reservation clears. @@ -1473,106 +1623,251 @@ impl DownloadManager { } fn spawn(self: &Rc, item: DownloadItem) { + let id = item.id(); + let dest = item.file_path(); + let dest_dir = item.dest_dir().to_string(); + let url = item.url().to_string(); + let is_torrent = crate::torrent::is_torrent(&url); + let is_video = matches!( + self.video_source(id), + Some(crate::media_types::VideoSource::Page { .. }) + ); + let segments = self.segment_state.borrow().get(&id).cloned(); + let started = self.started.borrow().contains(&id); + // Reserve the target and a scheduler slot while filesystem validation is in flight. + // This keeps another queued row or a rapid retry from claiming the same path, + // and prevents async preparation from bypassing max_concurrent. + self.reserve_dest(&dest); + self.preparing.borrow_mut().insert(id); + let weak = Rc::downgrade(self); + let task_weak = weak.clone(); + let handle = glib::spawn_future_local(async move { + async move { + let worker_dest = dest.clone(); + let worker_dir = dest_dir.clone(); + let worker_recorded = item.output_dir().to_string(); + let worker_segments = segments.clone(); + let preparation = gio::spawn_blocking(move || { + let parent = worker_dest.parent(); + if let Some(parent) = parent { + // Re-check containment on the worker immediately before touching the + // path: a symlink planted after intake must not redirect the write. + match ensure_contained_parent(parent, &worker_dir) { + Ok(true) => {} + Ok(false) => { + return Err(gettext("Download path escapes the download folder")); + } + Err(error) => { + return Err(gettext( + "Could not create the download folder: {error}", + ) + .replace("{error}", &error.to_string())); + } + } + } + + if is_torrent { + let dir = std::path::PathBuf::from(&worker_dir); + let recorded_path = std::path::PathBuf::from(&worker_recorded); + let (output, output_dir) = if worker_recorded.is_empty() { + (dir.clone(), None) + } else { + // Re-guard the recorded folder at spawn time; a symlink may have + // been planted after intake. Never follow it outside the destination. + let guarded = + crate::torrent::guard_output_folder(&dir, recorded_path.clone()); + let output_dir = (guarded != recorded_path) + .then(|| guarded.to_string_lossy().into_owned()); + if output_dir.is_none() && guarded.parent() != Some(dir.as_path()) { + std::fs::create_dir_all(&guarded).map_err(|error| { + gettext("Could not create the download folder: {error}") + .replace("{error}", &error.to_string()) + })?; + } + (guarded, output_dir) + }; + if worker_recorded.is_empty() { + std::fs::create_dir_all(&dir).map_err(|error| { + gettext("Could not create the download folder: {error}") + .replace("{error}", &error.to_string()) + })?; + } + return Ok(PreparedSpawn::Torrent { + dest: output, + output_dir, + }); + } + if is_video { + return Ok(PreparedSpawn::Video); + } + + let mut drop_segments = false; + let mode = match worker_segments { + Some(state) => { + // Drop an invalid bitmap and stale partial together. Do both on + // the worker so resume validation does not block GTK. + let len = std::fs::metadata(&worker_dest) + .map(|metadata| metadata.len()) + .unwrap_or(0); + if len < state.prefix_len() || len > state.total { + let _ = std::fs::remove_file(&worker_dest); + drop_segments = true; + StartMode::Fresh + } else { + StartMode::Resume(state) + } + } + // Existing destination is refused for a fresh row; never overwrite a + // file that may belong to the user or another application. + None => match std::fs::metadata(&worker_dest) { + Err(_) => StartMode::Fresh, + Ok(_) if !started => { + return Err(DEST_EXISTS.to_string()); + } + Ok(_) => StartMode::Single, + }, + }; + Ok(PreparedSpawn::Plain { + mode, + drop_segments, + }) + }) + .await; + + let Some(manager) = weak.upgrade() else { + return; + }; + manager.preparing.borrow_mut().remove(&id); + manager.release_dest(&dest); + let Some(current) = manager.find(id) else { + manager.start_next(); + return; + }; + if current.status() != DownloadStatus::Queued || current.file_path() != dest { + manager.start_next(); + return; + } + // A cancel/retry can happen while the blocking task runs. If any input + // changed, discard its answer and let the scheduler take a fresh snapshot. + let current_segments = manager.segment_state.borrow().get(&id).cloned(); + let same_segments = match (&segments, ¤t_segments) { + (None, None) => true, + (Some(before), Some(after)) => { + before.total == after.total && before.done == after.done + } + _ => false, + }; + if !same_segments || manager.started.borrow().contains(&id) != started { + manager.start_next(); + return; + } + + let prepared = match preparation { + Ok(Ok(prepared)) => prepared, + Ok(Err(detail)) => { + current.set_detail(detail.clone()); + current.set_status(DownloadStatus::Failed); + manager.changed(); + manager.notify_finished(¤t, Err(detail)); + manager.start_next(); + return; + } + Err(error) => { + let detail = gettext("Could not prepare download destination: {error}") + .replace("{error}", &format!("{error:?}")); + current.set_detail(detail.clone()); + current.set_status(DownloadStatus::Failed); + manager.changed(); + manager.notify_finished(¤t, Err(detail)); + manager.start_next(); + return; + } + }; + if matches!( + &prepared, + PreparedSpawn::Plain { + drop_segments: true, + .. + } + ) { + manager.segment_state.borrow_mut().remove(&id); + manager.persist_queue(); + } + manager.spawn_prepared(current, prepared); + } + .await; + if let Some(manager) = task_weak.upgrade() { + manager.preparation_handles.borrow_mut().remove(&id); + } + }); + // Insert after creating the task, before the main context can poll it. + if let Some(old) = self.preparation_handles.borrow_mut().insert(id, handle) { + old.abort(); + } + } + + fn spawn_prepared(self: &Rc, item: DownloadItem, prepared: PreparedSpawn) { let opts = DownloadOptions::from_settings(&self.settings); // Re-stash per attempt: a stale Last-Modified must never apply when the new run's server sends none. self.server_mtime.borrow_mut().remove(&item.id()); let dest = item.file_path(); - if let Some(parent) = dest.parent() { - // A planted symlink in any parent component would divert the - // download outside the destination: refuse instead of following - // it. (Filenames can't contain separators, so the parent is the - // destination itself today; this pins the invariant.) - // A mkdir failure is reported distinctly: it is not a containment - // refusal, so "escapes the download folder" would mislead. - let blocked = match ensure_contained_parent(parent, &item.dest_dir()) { - Ok(true) => None, - Ok(false) => Some(gettext("Download path escapes the download folder")), - Err(e) => Some( - gettext("Could not create the download folder: {error}") - .replace("{error}", &e.to_string()), - ), - }; - if let Some(detail) = blocked { - item.set_detail(detail.clone()); - item.set_status(DownloadStatus::Failed); - self.changed(); - self.notify_finished(&item, Err(detail)); - return; - } - } // Fail fast on junk; later edits apply without re-queueing. let limit = opts.limit_rate.trim(); if !limit.is_empty() && limit != "0" && parse_rate(limit).is_none() { item.set_detail(gettext("Invalid speed limit: {limit}").replace("{limit}", limit)); item.set_status(DownloadStatus::Failed); self.changed(); + self.start_next(); return; } publish_rate_limit(&self.settings); let url = item.url().to_string(); - if crate::torrent::is_torrent(&url) { - return self.spawn_torrent(item, url); - } - if let Some(crate::media_types::VideoSource::Page { - page_url, - quality, - audio_only, - is_live, - video_format_id, - playlist_item_id, - .. - }) = self.video_source(item.id()) - { - return self.spawn_video(SpawnVideoParams { - item, - page_url, - quality, - audio_only, - video_format_id, - is_live, - playlist_item_id, - }); - } - let connections = (opts.connections.max(1) as usize).min(16); - let timeout = Duration::from_secs(30); - // A saved bitmap means non-contiguous pieces: only a segmented resume is correct. - let mode = match self.segment_state.borrow().get(&item.id()).cloned() { - Some(st) => { - // Drop the bitmap if the file no longer holds the completed prefix (deleted/truncated/replaced while paused). - let len = std::fs::metadata(&dest).map(|m| m.len()).unwrap_or(0); - if len < st.prefix_len() || len > st.total { - self.segment_state.borrow_mut().remove(&item.id()); - let _ = std::fs::remove_file(&dest); - StartMode::Fresh - } else { - let frac = - (st.completed_bytes() as f64 / st.total.max(1) as f64).clamp(0.0, 1.0); - item.set_progress(frac); - StartMode::Resume(st) + let mode = match prepared { + PreparedSpawn::Torrent { dest, output_dir } => { + if let Some(output_dir) = output_dir { + item.set_output_dir(output_dir); } + return self.spawn_torrent(item, url, dest); } - None => match std::fs::metadata(&dest) { - // A missing destination is a fresh download. An existing - // destination — even zero bytes — is refused: it may not be - // ours, and silently deleting it could destroy foreign data. - Err(_) => StartMode::Fresh, - Ok(_) => { - // Parabolic-style dedup: refuse an existing destination - // instead of truncating or renaming it. Restored/retried - // rows carry `started` and resume their own partials - // below (progress alone can't tell: a crash before the - // first tick restores at 0.0 with bytes on disk). - if !self.started.borrow().contains(&item.id()) { - item.set_detail(DEST_EXISTS.to_string()); - item.set_status(DownloadStatus::Failed); - self.changed(); - self.notify_finished(&item, Err(DEST_EXISTS.to_string())); - return; - } - StartMode::Single - } - }, + PreparedSpawn::Video => { + let Some(crate::media_types::VideoSource::Page { + page_url, + quality, + audio_only, + is_live, + video_format_id, + playlist_item_id, + .. + }) = self.video_source(item.id()) + else { + let detail = gettext("Could not prepare video source"); + item.set_detail(detail); + item.set_status(DownloadStatus::Failed); + self.changed(); + self.start_next(); + return; + }; + return self.spawn_video(SpawnVideoParams { + item, + page_url, + quality, + audio_only, + video_format_id, + is_live, + playlist_item_id, + }); + } + PreparedSpawn::Plain { mode, .. } => mode, }; + let connections = (opts.connections.max(1) as usize).min(16); + let timeout = Duration::from_secs(30); + // Restore the visible fraction from the saved bitmap before starting the + // resumed engine; preparation runs off-thread but UI state remains main-thread-owned. + if let StartMode::Resume(state) = &mode { + let fraction = + (state.completed_bytes() as f64 / state.total.max(1) as f64).clamp(0.0, 1.0); + item.set_progress(fraction); + } let generation = self.epoch.borrow().get(&item.id()).cloned().unwrap_or(0) + 1; self.epoch.borrow_mut().insert(item.id(), generation); let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); @@ -1583,6 +1878,7 @@ impl DownloadManager { item.set_detail(e); item.set_status(DownloadStatus::Failed); self.changed(); + self.start_next(); return; } }; @@ -1592,6 +1888,7 @@ impl DownloadManager { item.set_detail(e); item.set_status(DownloadStatus::Failed); self.changed(); + self.start_next(); return; } }; @@ -1625,6 +1922,7 @@ impl DownloadManager { timeout, tx, }; + let resuming = matches!(&mode, StartMode::Resume(_)); let handle = tokio_rt().spawn(run_download(ctx, connections, mode)); self.running.borrow_mut().insert(item.id(), handle); // The engine owns the destination from here: later spawns resume it rather than refusing as foreign. @@ -1632,13 +1930,18 @@ impl DownloadManager { self.started.borrow_mut().insert(item.id()); self.persist_queue(); item.set_status(DownloadStatus::Downloading); - // Attempts start indeterminate: clear any stale fraction so the row pulses through the connecting phase. - item.set_progress(0.0); + // Fresh attempts pulse while connecting; a segmented resume keeps its + // restored completed-byte fraction visible instead of flashing back to zero. + if !resuming { + item.set_progress(0.0); + } let host = url::Url::parse(&item.url()) .ok() .and_then(|u| u.host_str().map(|h| h.to_string())) .unwrap_or_default(); - item.set_detail(if host.is_empty() { + item.set_detail(if resuming { + gettext("Resuming download…") + } else if host.is_empty() { gettext("Starting…") } else { gettext("Connecting to {host}…").replace("{host}", &host) @@ -1649,9 +1952,38 @@ impl DownloadManager { self.pump(item, item_id, generation, rx); } + /// Atomically adopt a finished filename. The directory listing can go stale between + /// deduplication and rename, so remember each candidate that lost EEXIST rather + /// than retrying the same name from the stale snapshot. + fn adopt_final_name( + old_path: &std::path::Path, + dir: &std::path::Path, + wanted: &str, + is_taken: impl Fn(&str) -> bool, + ) -> Option { + let lost_claims = RefCell::new(Vec::::new()); + let taken = + |name: &str| lost_claims.borrow().iter().any(|lost| lost == name) || is_taken(name); + for _ in 0..8 { + let candidate = dedupe_filename(wanted, taken); + let target = dir.join(&candidate); + if old_path == target { + return Some(candidate); + } + match rename_noreplace(old_path, &target) { + Ok(()) => return Some(candidate), + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => { + lost_claims.borrow_mut().push(candidate); + } + Err(_) => return None, + } + } + None + } + /// Whether a finished-name candidate is taken (file on disk, reserved stem, or live row holding it). Used by the Finished claim loop. fn is_name_taken(&self, dir: &str, existing: &[String], n: &str) -> bool { - std::path::Path::new(dir).join(n).exists() + existing.iter().any(|entry| entry == n) || crate::video_staging::stem_reserved_in(existing, name_stem(n)) // A discard still tearing down owns this destination and its // stem: the finalizer's sweep (and the orphan removal when the @@ -1777,38 +2109,71 @@ impl DownloadManager { { // A pause keeps its pending name: resumed single-stream attempts never re-suggest. let pending = this.pending_names.borrow_mut().remove(&id); - if let Some(name) = pending { - // Re-apply the final name policy: adoption is the last gate before dedupe/rename. Idempotent on finalized names. + // Drop the strong manager reference across the blocking directory + // listing. Re-upgrade into the arm's continuing binding afterward. + let (this, pending, existing) = if let Some(name) = pending { + // Re-apply the final name policy before yielding; idempotent on finalized names. let name = this.finalize_filename(&name); let current = item.filename().to_string(); let dir = item.dest_dir().to_string(); - let existing = crate::video_staging::dir_file_names( - std::path::Path::new(&dir), - ); - let taken = |n: &str| this.is_name_taken(&dir, &existing, n); + let listing_dir = dir.clone(); + drop(this); + let existing = match gio::spawn_blocking(move || { + crate::video_staging::dir_file_names(std::path::Path::new( + &listing_dir, + )) + }) + .await + { + Ok(names) => names, + Err(error) => { + tracing::warn!(error = ?error, dir = %dir, "finished-name directory listing failed"); + Vec::new() + } + }; + let Some(this) = weak.upgrade() else { + done = true; + break; + }; + if !this.is_current(id, generation) { + done = true; + break; + } + (this, Some((name, current, dir)), Some(existing)) + } else { + (this, None, None) + }; + if let Some((name, current, dir)) = pending { + let existing = existing.unwrap_or_default(); + // The main context was yielded while the directory was read. Do not finalize a row that was + // paused/cancelled, removed, or replaced while the worker ran. + if item.status() == DownloadStatus::Cancelled + || item.status() == DownloadStatus::Paused + { + done = true; + break; + } // Claim-then-move so a file appearing between dedupe and rename is never clobbered. let old_path = item.file_path(); - let mut final_name = dedupe_filename(&name, taken); - let mut moved = !old_path.exists(); - for _ in 0..8 { - if moved - || old_path == std::path::Path::new(&dir).join(&final_name) - { - break; - } - match rename_noreplace( - &old_path, - &std::path::Path::new(&dir).join(&final_name), - ) { - Ok(()) => { - moved = true; - } - Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => { - final_name = dedupe_filename(&name, taken); + let (final_name, moved) = + if !existing.iter().any(|entry| entry == ¤t) { + ( + dedupe_filename(&name, |n| { + this.is_name_taken(&dir, &existing, n) + }), + true, + ) + } else { + match Self::adopt_final_name( + &old_path, + std::path::Path::new(&dir), + &name, + |n| this.is_name_taken(&dir, &existing, n), + ) { + Some(name) => (name, true), + None => (current.clone(), false), } - Err(_) => break, - } - } + }; if moved && final_name != current { item.set_filename(final_name); } @@ -1818,32 +2183,63 @@ impl DownloadManager { // pre-rename one. Paints the engine's size now, // refines with the measured size when it lands. this.set_finished_detail(&item, size); - // Server file date, when asked: best-effort, never fails the row. + // Server-date stamping and torrent directory cleanup can perform filesystem + // metadata reads and recursive walks; snapshot plain inputs and defer them. let mtime = this.server_mtime.borrow_mut().remove(&id); - if this.settings.keep_server_date() - && let Some(t) = mtime - && let Ok(f) = std::fs::File::open(item.file_path()) - { - let _ = f.set_modified(t); - } + let keep_server_date = this.settings.keep_server_date(); + let cleanup_torrent = crate::torrent::is_torrent(&url) + && !this.settings.torrent_seed_finished(); + // Snapshot the filter before dropping its row-scoped state; the + // filesystem cleanup runs later on a blocking worker. + let cleanup_selection = cleanup_torrent + .then(|| crate::torrent::get_selection(&url)) + .flatten(); + let finished_path = item.file_path(); + let torrent_dest_dir = item.dest_dir().to_string(); + let torrent_output_dir = item.output_dir().to_string(); + let torrent_filename = item.filename().to_string(); item.set_progress(1.0); item.set_status(DownloadStatus::Done); // One finished record per URL (Parabolic parity): re-downloads replace instead of stacking. this.drop_finished_duplicates(&url, id); this.segment_state.borrow_mut().remove(&id); this.torrent_pieces.borrow_mut().remove(&id); - // Filtered torrents: drop untoggled files now that every selected byte is on disk. Skipped while seeding (serving still reads those spans). - if crate::torrent::is_torrent(&url) - && !this.settings.torrent_seed_finished() - { - let folder = Self::torrent_folder(&item); - if folder.is_dir() { - crate::torrent::cleanup_unselected(&folder, &url); - } + if cleanup_torrent { // The intake filter is spent: keeping it would pin the index list for a done row. crate::torrent::drop_selection(&url); } - this.notify_finished(&item, Ok(())); + if this.find(id).is_some() { + this.notify_finished(&item, Ok(())); + } + if (keep_server_date && mtime.is_some()) || cleanup_torrent { + let worker_url = url.clone(); + glib::spawn_future_local(async move { + let _ = gio::spawn_blocking(move || { + if keep_server_date + && let Some(time) = mtime + && let Ok(file) = std::fs::File::open(&finished_path) + { + let _ = file.set_modified(time); + } + if let Some(selected) = cleanup_selection { + let folder = DownloadManager::torrent_folder_from_parts( + &torrent_dest_dir, + &worker_url, + &torrent_output_dir, + &torrent_filename, + ); + if folder.is_dir() { + crate::torrent::cleanup_unselected( + &folder, + &worker_url, + &selected, + ); + } + } + }) + .await; + }); + } } done = true; break; @@ -1857,8 +2253,25 @@ impl DownloadManager { break; } // Playlist expansion runs on the main thread where queueing is legal; the new rows are the feedback. + let weak = Rc::downgrade(&this); + drop(this); let (added, total, skipped_by_host_limit) = - this.expand_playlist_rows(id, &item, &pl); + Self::expand_playlist_rows_async( + weak.clone(), + id, + generation, + &item, + &pl, + ) + .await; + let Some(this) = weak.upgrade() else { + done = true; + break; + }; + if !this.is_current(id, generation) { + done = true; + break; + } let skipped_detail = || { ngettext( "{skipped} item skipped because the collection exceeded the limit of {limit} distinct hosts", @@ -1941,11 +2354,27 @@ impl DownloadManager { // get 200 on every piece, and fail again permanently. // Explicitly delete the partial file: its bytes belong to the dead version and must not survive // into a retry (previously this was only emergent via 416/hole handling). - let _ = std::fs::remove_file(item.file_path()); + let invalid_path = item.file_path(); this.segment_state.borrow_mut().remove(&id); this.pending_names.borrow_mut().remove(&id); item.set_etag(String::new()); item.set_last_modified(String::new()); + this.persist_queue(); + this.reserve_dest(&invalid_path); + let worker_path = invalid_path.clone(); + drop(this); + let _ = + gio::spawn_blocking(move || std::fs::remove_file(worker_path)).await; + let Some(this) = weak.upgrade() else { + done = true; + break; + }; + this.release_dest(&invalid_path); + if !this.is_current(id, generation) { + done = true; + break; + } + this.start_next(); if item.status() != DownloadStatus::Cancelled && item.status() != DownloadStatus::Paused { @@ -1968,6 +2397,7 @@ impl DownloadManager { // no bitmap, and a 416 would pass it off as complete on // filesystems that zero-fill instead of leaving holes. this.persist_queue(); + drop(this); ack.send(()).await.ok(); } EngineMsg::PieceDone(idx) => { @@ -1985,19 +2415,79 @@ impl DownloadManager { } } EngineMsg::TruncatePrefix => { - if let Some(st) = this.segment_state.borrow_mut().get_mut(&id) { - truncate_to_prefix(&item.file_path(), st); - // The file lost everything past the prefix; the bitmap must forget it too or a later resume skips missing pieces. - st.forget_beyond_prefix(); + // Commit the matching bitmap first: Undo/retry during the worker must never see a + // prefix-truncated file paired with bits beyond the first gap. + let state = { + let mut states = this.segment_state.borrow_mut(); + states.get_mut(&id).map(|st| { + st.forget_beyond_prefix(); + st.clone() + }) + }; + if let Some(state) = state { + this.persist_queue(); + let path = item.file_path(); + this.reserve_dest(&path); + let worker_path = path.clone(); + drop(this); + let _ = gio::spawn_blocking(move || { + truncate_to_prefix(&worker_path, &state) + }) + .await; + let Some(this) = weak.upgrade() else { + done = true; + break; + }; + this.release_dest(&path); + if !this.is_current(id, generation) { + done = true; + break; + } + this.start_next(); } } EngineMsg::FallbackSingle { ack } => { - // Server throttled parallel connections: shrink to the completed prefix and forget the bitmap on the main thread. The engine waits for this ack before appending at EOF. - if let Some(st) = this.segment_state.borrow().get(&id) { - truncate_to_prefix(&item.file_path(), st); + // Keep the bitmap persisted until the sparse/segmented file is + // actually truncated. A crash during the worker must still restore + // with the bitmap rather than append to a file with holes. + let segment_state = { this.segment_state.borrow().get(&id).cloned() }; + if let Some(state) = segment_state { + let path = item.file_path(); + this.reserve_dest(&path); + let worker_path = path.clone(); + drop(this); + let truncated = gio::spawn_blocking(move || { + truncate_to_prefix(&worker_path, &state) + }) + .await + .is_ok(); + let Some(this) = weak.upgrade() else { + done = true; + break; + }; + this.release_dest(&path); + if !this.is_current(id, generation) { + done = true; + break; + } + if truncated { + this.segment_state.borrow_mut().remove(&id); + this.persist_queue(); + this.start_next(); + drop(this); + ack.send(()).await.ok(); + } else { + tracing::warn!(path = %path.display(), "could not truncate segmented file before single-stream fallback"); + // Dropping ack tells the engine that fallback cannot safely proceed. + done = true; + break; + } + } else { + this.segment_state.borrow_mut().remove(&id); + this.persist_queue(); + drop(this); + ack.send(()).await.ok(); } - this.segment_state.borrow_mut().remove(&id); - ack.send(()).await.ok(); } EngineMsg::SuggestName(name) => { // Stash the server-advertised name for adoption at Finished: @@ -2098,29 +2588,13 @@ impl DownloadManager { } /// Spawn the torrent engine for a magnet row. Mirrors `spawn`'s contract so pause/cancel/retry and the stale-pump guard keep working. - fn spawn_torrent(self: &Rc, item: DownloadItem, magnet: String) { - let dir = std::path::PathBuf::from(item.dest_dir().to_string()); - // Magnets recorded their subfolder at enqueue; archived torrents recompute theirs from metadata. + fn spawn_torrent( + self: &Rc, + item: DownloadItem, + magnet: String, + dest: std::path::PathBuf, + ) { let recorded = item.output_dir().to_string(); - let dest = if recorded.is_empty() { - let _ = std::fs::create_dir_all(&dir); - dir.clone() - } else { - // Re-guard the recorded subfolder at spawn: a symlink planted - // after intake (a wide window for queued torrents) would - // otherwise divert the engine's writes. A squatted name dedupes - // to a fresh folder, like collection subfolders. - let recorded_path = std::path::PathBuf::from(&recorded); - let guarded = crate::torrent::guard_output_folder(&dir, recorded_path.clone()); - if guarded != recorded_path { - item.set_output_dir(guarded.to_string_lossy().into_owned()); - } else if guarded.parent() != Some(dir.as_path()) { - // Not under the dest dir (shouldn't happen: every recorded - // path is dir-joined): keep the old ensure-exists behavior. - let _ = std::fs::create_dir_all(&guarded); - } - guarded - }; let settings = &self.settings; let seed_finished = settings.torrent_seed_finished(); let peer_limit: Option = None; // rqbit default @@ -2217,6 +2691,20 @@ impl DownloadManager { } } +/// Validated worker preparation. The enum makes impossible combinations (such as a +/// torrent without an output directory or a plain row without a start mode) unrepresentable. +enum PreparedSpawn { + Torrent { + dest: std::path::PathBuf, + output_dir: Option, + }, + Video, + Plain { + mode: StartMode, + drop_segments: bool, + }, +} + /// Video-page spawn inputs, bundled so the field list doesn't trip clippy's too-many-arguments lint. struct SpawnVideoParams { item: DownloadItem, @@ -2776,7 +3264,9 @@ impl DownloadManager { /// Cancel a download; retry with [`DownloadManager::retry`]. pub fn cancel(self: &Rc, id: u64) { - self.cancel_inner(id, false, Stop::Preserve); + if let Some(path) = self.cancel_inner(id, false, Stop::Preserve) { + self.schedule_partial_cleanup(path); + } self.persist_queue(); self.changed(); self.start_next(); @@ -2833,7 +3323,7 @@ impl DownloadManager { } } - fn cancel_inner(&self, id: u64, keep_partial: bool, stop: Stop) { + fn cancel_inner(&self, id: u64, keep_partial: bool, stop: Stop) -> Option { // Live rows keep what's recorded (Stop, not Cancel): signal the worker and skip the abort plus the Cancelled preset, so its Finished still lands. Removal is the one caller that discards. let live = self.live_rows.borrow().contains(&id); // Only a *video* worker can be told to discard: plain rows have no `video_abort` sender, so a Discard-only path would leak the engine and its slot. @@ -2852,10 +3342,12 @@ impl DownloadManager { self.pending_names.borrow_mut().remove(&id); let had_segments = self.segment_state.borrow_mut().remove(&id).is_some(); self.torrent_pieces.borrow_mut().remove(&id); + let mut partial_to_remove = None; if let Some(item) = self.find(id) { // Segmented partials have holes: without the bitmap cancel drops the file — unless the row may come back via Undo. + // Return the path so callers can reserve it and unlink off the GTK thread. if had_segments && !keep_partial { - let _ = std::fs::remove_file(item.file_path()); + partial_to_remove = Some(item.file_path()); } // Unfinished torrent rows drop their session entry but keep partial files, so cancel/retry and remove/Undo resume. if crate::torrent::is_torrent(&item.url()) && item.status() != DownloadStatus::Done { @@ -2869,6 +3361,7 @@ impl DownloadManager { item.set_detail(item.status().label()); } } + partial_to_remove } /// Re-queue a failed or cancelled download. @@ -2900,7 +3393,7 @@ impl DownloadManager { /// Cancel and drop a row; restore with [`DownloadManager::unremove`]. The partial file is kept so Undo can resume. pub fn remove(self: &Rc, id: u64) { - self.cancel_inner(id, true, Stop::Discard); + let _ = self.cancel_inner(id, true, Stop::Discard); self.epoch.borrow_mut().remove(&id); self.started.borrow_mut().remove(&id); // Video rows defer the sweep to `finish_discard`, which waits for the worker: a finalize path still inside its rename would otherwise deliver a file for a row that no longer exists. @@ -2998,6 +3491,7 @@ impl DownloadManager { /// recorded branch refuses a path that is itself a symlink. Either way, a /// symlink planted after the download finished can't divert the /// delete/trash onto its target. + #[cfg(test)] fn torrent_folder(item: &DownloadItem) -> std::path::PathBuf { let dest_dir = item.dest_dir().to_string(); let url = item.url().to_string(); @@ -3148,6 +3642,7 @@ impl DownloadManager { /// Cancel every queued, downloading or paused item. pub fn cancel_all(self: &Rc) { // One persist at the end and no per-row start_next: cancel() would briefly spawn the next queued row only to cancel it right after. + let mut partials_to_remove = Vec::new(); self.for_matching( |s| { matches!( @@ -3155,8 +3650,15 @@ impl DownloadManager { DownloadStatus::Queued | DownloadStatus::Downloading | DownloadStatus::Paused ) }, - |m, id| m.cancel_inner(id, false, Stop::Preserve), + |m, id| { + if let Some(path) = m.cancel_inner(id, false, Stop::Preserve) { + partials_to_remove.push(path); + } + }, ); + for path in partials_to_remove { + self.schedule_partial_cleanup(path); + } self.persist_queue(); self.changed(); self.start_next(); @@ -3395,8 +3897,10 @@ impl DownloadManager { Self::write_queue_file(&text); } #[cfg(not(test))] - if let (Some(text), Some(tx)) = (self.serialize_queue(), &*self.persist_tx.borrow()) { - tx.send_replace(Some(text)); + if let (Some(queue), Some(tx)) = + (self.snapshot_queue_inner(false), &*self.persist_tx.borrow()) + { + tx.send_replace(Some(queue)); } } @@ -3404,17 +3908,19 @@ impl DownloadManager { /// Extracted for testing: `watch::changed()` delivers the final value before /// the closed-channel Err, so dropping the sender after a final send guarantees /// the last write lands. - pub(crate) async fn persist_task_loop( - mut rx: tokio::sync::watch::Receiver>, - write: impl Fn(String) + Clone + Send + 'static, - ) { + pub(crate) async fn persist_task_loop( + mut rx: tokio::sync::watch::Receiver>, + write: impl Fn(T) + Clone + Send + 'static, + ) where + T: Clone + Send + 'static, + { while rx.changed().await.is_ok() { - let Some(text) = rx.borrow_and_update().clone() else { + let Some(value) = rx.borrow_and_update().clone() else { continue; }; - // File I/O must not block the async runtime. + // Queue serialization and disk I/O must not block GTK or the async runtime. let write = write.clone(); - let _ = tokio::task::spawn_blocking(move || write(text)).await; + let _ = tokio::task::spawn_blocking(move || write(value)).await; } } @@ -3435,29 +3941,24 @@ impl DownloadManager { tracing::warn!( "shutdown: queue restore did not complete; preserving the saved queue file" ); - } else if let (Some(text), Some(tx)) = - (self.serialize_queue_inner(true), &*self.persist_tx.borrow()) + } else if let (Some(queue), Some(tx)) = + (self.snapshot_queue_inner(true), &*self.persist_tx.borrow()) { - tx.send_replace(Some(text)); + tx.send_replace(Some(queue)); } drop(self.persist_tx.borrow_mut().take()); + // This fallback runs after the main loop has exited, when blocking + // cannot freeze GTK. Bound the wait so a stalled disk cannot hang exit. if let Some(task) = self.persist_task.borrow_mut().take() { - // Bound the shutdown wait: a stalled disk must not hang quit. - let _ = tokio_rt().block_on(async { - tokio::time::timeout(std::time::Duration::from_secs(5), task).await - }); + let _ = tokio_rt() + .block_on(async { tokio::time::timeout(Duration::from_secs(5), task).await }); } } } - /// Serialize the queue to JSON. Returns None if batching or on serialize error. - fn serialize_queue(&self) -> Option { - self.serialize_queue_inner(false) - } - - /// Serialize the queue to JSON, bypassing the batch gate when `force` is true. - /// Shutdown uses `force=true`: quitting mid-batch must persist the final state. - fn serialize_queue_inner(&self, force: bool) -> Option { + /// Build a thread-safe snapshot from GTK-owned rows. The expensive JSON + /// encoding is done by the background persistence worker. + fn snapshot_queue_inner(&self, force: bool) -> Option { if !force && self.batch.get() > 0 { return None; } @@ -3526,20 +4027,33 @@ impl DownloadManager { }); } } - let data = StoredQueue { + Some(StoredQueue { version: QUEUE_VERSION, items, - }; - // Compact output cuts queue size ~3x but does not enforce the 10 MB - // restore cap; a very large paused set can still be quarantined. - let text = match serde_json::to_string(&data) { - Ok(text) => text, - Err(e) => { - tracing::error!("could not serialize download queue: {e}"); - return None; + }) + } + + /// Test helper for assertions over the serialized queue representation. + #[cfg(test)] + fn serialize_queue(&self) -> Option { + self.serialize_queue_inner(false) + } + + /// Synchronous serialization is used only by deterministic test persistence. + #[cfg(test)] + fn serialize_queue_inner(&self, force: bool) -> Option { + let queue = self.snapshot_queue_inner(force)?; + Self::serialize_snapshot(queue) + } + + fn serialize_snapshot(queue: StoredQueue) -> Option { + match serde_json::to_string(&queue) { + Ok(text) => Some(text), + Err(error) => { + tracing::error!("could not serialize download queue: {error}"); + None } - }; - Some(text) + } } /// Write the serialized queue atomically: temp file + sync_all + rename + @@ -4230,10 +4744,10 @@ impl DownloadManager { #[cfg(not(test))] { if self.restore_state.get() == RestoreState::Ready { - if let (Some(text), Some(tx)) = - (self.serialize_queue_inner(true), &*self.persist_tx.borrow()) + if let (Some(queue), Some(tx)) = + (self.snapshot_queue_inner(true), &*self.persist_tx.borrow()) { - tx.send_replace(Some(text)); + tx.send_replace(Some(queue)); } } else { tracing::warn!( @@ -4262,50 +4776,22 @@ impl DownloadManager { } } - pub(crate) fn shutdown_with_timeout(&self, timeout: Duration) { + pub(crate) fn shutdown_with_timeout(&self, _timeout: Duration) { self.stop_scheduler(); self.draining.set(true); let handles: Vec<_> = self.running.borrow_mut().drain().map(|(_, h)| h).collect(); - // Discard workers first through their retained abort handles: aborting a finalizer would detach its worker instead of stopping it. Finalizers are awaited (not aborted) so cleanup still runs. let finals: Vec = self.discards.borrow_mut().drain().map(|(_, p)| p).collect(); - for handle in handles.iter() { + for handle in &handles { handle.abort(); } for pending in &finals { pending.worker_abort.abort(); } - // Bound the join: on timeout the remaining handles are dropped - // (detached) and partial files are left alone — the next launch - // reconciles them via the length-vs-prefix check in spawn(). - let joined = tokio_rt() - .block_on(async { - tokio::time::timeout(timeout, async { - for handle in handles { - let _ = handle.await; - } - for pending in finals { - let _ = pending.finalizer.await; - } - }) - .await - }) - .is_ok(); - if !joined { - tracing::warn!("shutdown: join timed out; persisting without truncating partial files"); - self.persist_for_shutdown(); - return; - } - let ids: Vec = self.segment_state.borrow().keys().cloned().collect(); - for id in ids { - if let Some(item) = self.find(id) - && let Some(st) = self.segment_state.borrow_mut().get_mut(&id) - { - truncate_to_prefix(&item.file_path(), st); - st.forget_beyond_prefix(); - } - } - // Persist BEFORE returning: pump tails exit silently once draining is set, so this is the only persist that matters. + // This synchronous fallback is only for shutdown paths that bypass the + // normal GLib async close flow. Do not block GTK joining workers or + // truncating files. Preserve the bitmap as-is; the next launch can + // reconcile stale partial lengths without risking a false bitmap. self.persist_for_shutdown(); } } diff --git a/src/download_store.rs b/src/download_store.rs index b92377b22..4190aead1 100644 --- a/src/download_store.rs +++ b/src/download_store.rs @@ -89,7 +89,7 @@ pub(crate) struct StoredItem { pub(crate) gallery: Option, } -#[derive(Debug, serde::Serialize, serde::Deserialize)] +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub(crate) struct StoredQueue { pub(crate) version: u32, pub(crate) items: Vec, diff --git a/src/download_tests.rs b/src/download_tests.rs index 2674d9825..ca709b1bb 100644 --- a/src/download_tests.rs +++ b/src/download_tests.rs @@ -30,15 +30,56 @@ static QUEUE_FILE_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); /// wrong thread. Take AFTER QUEUE_FILE_LOCK, always that order. static MAIN_LOOP_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); +fn lock_queue_file() -> std::sync::MutexGuard<'static, ()> { + QUEUE_FILE_LOCK + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} + +fn lock_main_loop() -> std::sync::MutexGuard<'static, ()> { + MAIN_LOOP_LOCK + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} + /// Both locks in the safe order (queue, then loop); reversed order deadlocks the suite. fn test_locks() -> ( std::sync::MutexGuard<'static, ()>, std::sync::MutexGuard<'static, ()>, ) { - // Recover from poison: a panicking test shouldn't cascade into 28 identical failures. - let q = QUEUE_FILE_LOCK.lock().unwrap(); - let l = MAIN_LOOP_LOCK.lock().unwrap(); - (q, l) + (lock_queue_file(), lock_main_loop()) +} + +/// Wait for the asynchronous spawn preparation to hand a queued row to its engine. +fn wait_for_started(manager: &DownloadManager, id: u64, timeout: Duration) -> bool { + let ctx = glib::MainContext::default(); + let deadline = std::time::Instant::now() + timeout; + while !manager.started.borrow().contains(&id) + && manager + .find(id) + .is_some_and(|item| item.status() == DownloadStatus::Queued) + && std::time::Instant::now() < deadline + { + ctx.iteration(false); + std::thread::sleep(std::time::Duration::from_millis(1)); + } + manager.started.borrow().contains(&id) +} + +/// Wait for async destination preparation to settle, including its local-task cleanup. +fn wait_for_preparation(manager: &DownloadManager, id: u64, timeout: Duration) -> bool { + let ctx = glib::MainContext::default(); + let deadline = std::time::Instant::now() + timeout; + while (manager.preparing.borrow().contains(&id) + || manager.preparation_handles.borrow().contains_key(&id)) + && std::time::Instant::now() < deadline + { + ctx.iteration(false); + std::thread::sleep(std::time::Duration::from_millis(1)); + } + quiesce(&ctx); + !manager.preparing.borrow().contains(&id) + && !manager.preparation_handles.borrow().contains_key(&id) } /// Minimal persisted row for `restore_existing` fixtures (no id => pre-v3 allocate path). @@ -83,7 +124,12 @@ fn drain_engine(manager: &DownloadManager, id: u64) { // Also wait for finished-row sizings: `set_finished_detail` spawns one // per completion, and it outlives the engine that triggered it. while (manager.running.borrow().contains_key(&id) + || manager.preparing.borrow().contains(&id) + || manager.preparation_handles.borrow().contains_key(&id) || manager.pump_handles.borrow().contains_key(&id) + || manager + .find(id) + .is_some_and(|item| manager.dest_reserved(&item.file_path())) || manager.sizing_inflight.get() > 0) && std::time::Instant::now() < deadline { @@ -93,6 +139,25 @@ fn drain_engine(manager: &DownloadManager, id: u64) { quiesce(&ctx); } +/// Wait for cancel_all's path reservations to clear, proving its blocking +/// cleanup workers and their main-context continuations have completed. +fn drain_cancel_cleanups(manager: &DownloadManager) { + let ctx = glib::MainContext::default(); + let deadline = std::time::Instant::now() + Duration::from_secs(15); + while manager + .items() + .any(|item| manager.dest_reserved(&item.file_path())) + { + assert!( + std::time::Instant::now() < deadline, + "timed out draining cancellation cleanup workers" + ); + ctx.iteration(false); + std::thread::sleep(Duration::from_millis(1)); + } + quiesce(&ctx); +} + /// Pump until quiet (50 idle rounds) so no woken tail trips another test's loop. fn quiesce(ctx: &glib::MainContext) { let mut idle_rounds = 0; @@ -433,7 +498,7 @@ fn dedupe_caps_iterations() { #[test] fn corrupt_queue_is_quarantined() { - let _lock = QUEUE_FILE_LOCK.lock().unwrap(); + let _lock = lock_queue_file(); let qf = test_queue_file("quarantine"); std::fs::write(&qf, b"{not json").unwrap(); let settings = test_settings(); @@ -451,7 +516,7 @@ fn invalid_utf8_queue_is_quarantined() { // read_to_string fails on invalid UTF-8: must quarantine like the other // corrupt-queue branches instead of silently starting empty (a later // persist would then overwrite the unreadable file). - let _lock = QUEUE_FILE_LOCK.lock().unwrap(); + let _lock = lock_queue_file(); let qf = test_queue_file("quarantine-utf8"); std::fs::write(&qf, b"{\xff\xfe not utf-8").unwrap(); let settings = test_settings(); @@ -468,7 +533,7 @@ fn invalid_utf8_queue_is_quarantined() { fn shutdown_with_timeout_returns_promptly_with_stuck_finalizer() { // A wedged finalizer must not hang the GTK thread on quit: the join is // bounded, partial files are left for the next launch to reconcile. - let _lock = QUEUE_FILE_LOCK.lock().unwrap(); + let _lock = lock_queue_file(); let _qf = test_queue_file("shutdown-timeout"); let settings = test_settings(); let m = DownloadManager::new(gio::ListStore::new::(), settings); @@ -828,7 +893,7 @@ fn parses_rates() { #[test] fn rate_limit_follows_settings_live() { - let _lock = QUEUE_FILE_LOCK.lock().unwrap(); + let _lock = lock_queue_file(); let _qf = test_queue_file("rate-live"); let settings = test_settings(); // Watch publishes; cap follows with no re-queue, junk reads as unlimited. @@ -860,7 +925,7 @@ fn notification_toggles() { #[test] fn transferring_ignores_paused() { - let _lock = QUEUE_FILE_LOCK.lock().unwrap(); + let _lock = lock_queue_file(); let _qf = test_queue_file("transferring"); let settings = test_settings(); let manager = DownloadManager::new(gio::ListStore::new::(), settings); @@ -1058,8 +1123,8 @@ fn enqueue_video_spawns_and_fails_without_tools() { .join("grab-video-enqueue") .to_string_lossy() .into_owned(); - let item = manager - .enqueue_video( + let item = glib::MainContext::default() + .block_on(manager.enqueue_video_async( "https://www.youtube.com/watch?v=gXtp6C-3JKo", Some(&dest), Some("My Video.mp4"), @@ -1070,7 +1135,7 @@ fn enqueue_video_spawns_and_fails_without_tools() { is_live: false, playlist_item_id: None, }, - ) + )) .expect("video enqueue"); let id = item.id(); // The worker runs and fails fast on the missing tools (no network). @@ -1105,8 +1170,8 @@ fn enqueue_video_restrict_filenames_folds_name() { let dest = std::env::temp_dir().join("grab-video-restrict"); std::fs::create_dir_all(&dest).unwrap(); let dest = dest.to_string_lossy().into_owned(); - let item = manager - .enqueue_video( + let item = glib::MainContext::default() + .block_on(manager.enqueue_video_async( "https://www.youtube.com/watch?v=gXtp6C-3JKo", Some(&dest), Some("Café & Croissants.mp4"), @@ -1117,7 +1182,7 @@ fn enqueue_video_restrict_filenames_folds_name() { is_live: false, playlist_item_id: None, }, - ) + )) .expect("video enqueue"); assert_eq!(item.filename(), "Cafe_Croissants.mp4"); let id = item.id(); @@ -1299,8 +1364,8 @@ fn video_source_survives_restore_and_retry() { let id = { let settings = test_settings(); let manager = DownloadManager::new(gio::ListStore::new::(), settings); - let item = manager - .enqueue_video( + let item = glib::MainContext::default() + .block_on(manager.enqueue_video_async( "https://vimeo.com/123456", Some(&dest), Some("Clip.mp4"), @@ -1311,7 +1376,7 @@ fn video_source_survives_restore_and_retry() { is_live: false, playlist_item_id: None, }, - ) + )) .expect("video enqueue"); // Persisted with the Page marker (not silently dropped). let text = std::fs::read_to_string(&qf).unwrap(); @@ -1339,7 +1404,7 @@ fn video_source_survives_restore_and_retry() { #[test] fn mismatched_video_source_dropped_on_restore() { - let _lock = QUEUE_FILE_LOCK.lock().unwrap(); + let _lock = lock_queue_file(); let qf = test_queue_file("video-mismatch"); // Hand-edited queue: the Page marker names another video. let queue = StoredQueue { @@ -1461,8 +1526,8 @@ fn unremove_restores_video_source() { .join("grab-video-unremove") .to_string_lossy() .into_owned(); - let item = manager - .enqueue_video( + let item = glib::MainContext::default() + .block_on(manager.enqueue_video_async( "https://vimeo.com/123456", Some(&dest), Some("Clip.mp4"), @@ -1473,7 +1538,7 @@ fn unremove_restores_video_source() { is_live: false, playlist_item_id: None, }, - ) + )) .expect("video enqueue"); let id = item.id(); drain_engine(&manager, id); @@ -1513,14 +1578,14 @@ fn unremove_restores_video_source() { #[test] fn queue_file_never_carries_cookies() { // Secrets must not reach the persisted queue, whatever the settings. - let _lock = QUEUE_FILE_LOCK.lock().unwrap(); + let (_lock, _loop) = test_locks(); let _notools = NoVideoTools::apply(); let qf = test_queue_file("video-cookies-persist"); let settings = test_settings(); settings.set_string("cookies-browser", "firefox").unwrap(); let manager = DownloadManager::new(gio::ListStore::new::(), settings); - let item = manager - .enqueue_video( + let item = glib::MainContext::default() + .block_on(manager.enqueue_video_async( "https://vimeo.com/123456", Some("/tmp/dl"), Some("Clip.mp4"), @@ -1531,7 +1596,7 @@ fn queue_file_never_carries_cookies() { is_live: false, playlist_item_id: None, }, - ) + )) .expect("video enqueue"); // Drain the worker here so no woken pump tail trips another test's thread guard. drain_engine(&manager, item.id()); @@ -1548,7 +1613,7 @@ fn queue_file_never_carries_cookies() { #[test] fn active_count_covers_cancel_all_scope() { - let _lock = QUEUE_FILE_LOCK.lock().unwrap(); + let _lock = lock_queue_file(); let _qf = test_queue_file("active-count"); let settings = test_settings(); let manager = DownloadManager::new(gio::ListStore::new::(), settings); @@ -1571,7 +1636,7 @@ fn active_count_covers_cancel_all_scope() { #[test] fn error_banner_ignores_cancelled() { - let _lock = QUEUE_FILE_LOCK.lock().unwrap(); + let _lock = lock_queue_file(); let _qf = test_queue_file("error-banner"); let settings = test_settings(); let manager = DownloadManager::new(gio::ListStore::new::(), settings); @@ -1632,7 +1697,7 @@ fn path_size_async_measures_off_thread() { // The async wrapper must actually measure, not just resolve: a stub that // returns None without scanning would leave finished rows unsized. // MAIN_LOOP_LOCK: glib futures abort if polled from the wrong thread. - let _loop = MAIN_LOOP_LOCK.lock().unwrap(); + let _loop = lock_main_loop(); let dir = std::env::temp_dir().join(format!("grab-path-size-async-{}", std::process::id())); let _ = std::fs::remove_dir_all(&dir); let sub = dir.join("sub"); @@ -1802,7 +1867,7 @@ fn staged_selection_survives_respawn() { #[test] fn torrent_file_enqueue_uses_stem_stub() { - let _lock = QUEUE_FILE_LOCK.lock().unwrap(); + let (_lock, _loop) = test_locks(); let _qf = test_queue_file("torrent-enqueue"); let settings = test_settings(); // Occupy the only slot so nothing spawns a real engine below. @@ -1813,13 +1878,13 @@ fn torrent_file_enqueue_uses_stem_stub() { }); manager.running.borrow_mut().insert(99, holder); let stem = format!("grab-enqueue-{}", std::process::id()); - let item = manager - .enqueue_torrent_file( + let item = glib::MainContext::default() + .block_on(manager.enqueue_torrent_file_async( single_torrent_bytes(), &format!("{stem}.torrent"), None, None, - ) + )) .unwrap(); assert_eq!(item.status(), DownloadStatus::Queued); assert!(crate::torrent::is_torrent_url(&item.url())); @@ -1834,7 +1899,7 @@ fn torrent_file_enqueue_uses_stem_stub() { #[test] fn staged_selection_reaches_spawn_take() { - let _lock = QUEUE_FILE_LOCK.lock().unwrap(); + let (_lock, _loop) = test_locks(); let _qf = test_queue_file("torrent-take"); let settings = test_settings(); // Occupy the only slot so nothing spawns a real engine below. @@ -1845,13 +1910,13 @@ fn staged_selection_reaches_spawn_take() { }); manager.running.borrow_mut().insert(99, holder); let stem = format!("grab-take-{}", std::process::id()); - let item = manager - .enqueue_torrent_file( + let item = glib::MainContext::default() + .block_on(manager.enqueue_torrent_file_async( single_torrent_bytes(), &format!("{stem}.torrent"), None, Some(vec![0]), - ) + )) .unwrap(); // Must survive the archive → normalize → store round-trip identically. assert_eq!( @@ -1893,6 +1958,7 @@ fn delete_download_trashes_torrent_files() { .block_on(manager.delete_download_async(7)) .is_ok() ); + quiesce(&glib::MainContext::default()); assert!(!file.exists()); assert_eq!(store.n_items(), 0); assert!(crate::torrent::archive_path_for_url(&pseudo).is_none()); @@ -1905,7 +1971,7 @@ fn delete_download_trashes_torrent_files() { #[test] fn remove_keeps_torrent_archive_for_undo() { - let _lock = QUEUE_FILE_LOCK.lock().unwrap(); + let (_lock, _loop) = test_locks(); let _qf = test_queue_file("torrent-remove"); let settings = test_settings(); // Occupy the only slot so nothing spawns a real engine below. @@ -1916,13 +1982,13 @@ fn remove_keeps_torrent_archive_for_undo() { }); manager.running.borrow_mut().insert(99, holder); let stem = format!("grab-undo-{}", std::process::id()); - let item = manager - .enqueue_torrent_file( + let item = glib::MainContext::default() + .block_on(manager.enqueue_torrent_file_async( single_torrent_bytes(), &format!("{stem}.torrent"), None, None, - ) + )) .unwrap(); let id = item.id(); let url = item.url().to_string(); @@ -1942,7 +2008,7 @@ fn remove_keeps_torrent_archive_for_undo() { #[test] fn rejects_relative_download_dir() { - let _lock = QUEUE_FILE_LOCK.lock().unwrap(); + let _lock = lock_queue_file(); let _qf = test_queue_file("relative-dir"); let settings = test_settings(); // Occupy the only slot so nothing spawns a real engine below. @@ -2034,6 +2100,7 @@ fn delete_download_removes_file_and_row() { .block_on(manager.delete_download_async(7)) .is_ok() ); + quiesce(&glib::MainContext::default()); assert!(!file.exists()); assert_eq!(store.n_items(), 0); // Undo the test's own Trash litter. @@ -2053,6 +2120,7 @@ fn delete_download_removes_file_and_row() { .block_on(manager.delete_download_async(8)) .is_ok() ); + quiesce(&glib::MainContext::default()); assert_eq!(store.n_items(), 0); let _ = std::fs::remove_dir_all(&dir); } @@ -2123,7 +2191,11 @@ fn finish_cleans_untoggled_placeholders() { std::fs::create_dir_all(folder.join("sub")).unwrap(); std::fs::write(folder.join("a.txt"), b"kept").unwrap(); std::fs::write(folder.join("sub").join("b.txt"), b"").unwrap(); - crate::torrent::cleanup_unselected(&folder, &pseudo); + let selected = crate::torrent::get_selection(&pseudo).unwrap(); + // Finished drops the URL-scoped selection before its deferred worker runs. + // Passing the snapshot must still remove untoggled placeholders. + crate::torrent::drop_selection(&pseudo); + crate::torrent::cleanup_unselected(&folder, &pseudo, &selected); assert!(folder.join("a.txt").exists()); assert!(!folder.join("sub").exists()); assert!(folder.exists()); @@ -2160,6 +2232,7 @@ fn delete_download_trashes_torrent_subfolder() { .block_on(manager.delete_download_async(9)) .is_ok() ); + quiesce(&glib::MainContext::default()); // Owned file trashed, unrelated file survives, folder left in place. assert!(!folder.join("a.txt").exists()); assert!(folder.join("unrelated.txt").exists()); @@ -2202,6 +2275,7 @@ fn delete_download_magnet_leaves_preexisting_folder() { .block_on(manager.delete_download_async(10)) .is_ok() ); + quiesce(&glib::MainContext::default()); // Folder and unrelated file must survive. assert!(folder.exists()); assert_eq!( @@ -2247,6 +2321,7 @@ fn torrent_delete_refuses_symlinked_subdirectory() { .block_on(manager.delete_download_async(11)) .is_ok() ); + quiesce(&glib::MainContext::default()); // Precious file and symlink must survive. assert_eq!(std::fs::read(outside.join("b.txt")).unwrap(), b"precious"); assert!(folder.join("sub").is_symlink()); @@ -2399,7 +2474,12 @@ fn enqueue_keeps_name_when_file_exists() { ) .expect("enqueue"); assert_eq!(item.filename(), "dup.bin"); - // insert -> start_next -> spawn run synchronously: the fresh row refuses the foreign file. + // Destination validation now runs asynchronously off the main context. + assert!(wait_for_preparation( + &manager, + item.id(), + Duration::from_secs(15) + )); assert_eq!(item.status(), DownloadStatus::Failed); assert_eq!(item.detail(), crate::engine_msg::DEST_EXISTS); assert_eq!( @@ -2423,8 +2503,8 @@ fn enqueue_video_keeps_name_when_file_exists() { std::fs::create_dir_all(&dest).unwrap(); std::fs::write(dest.join("clip.mp4"), b"someone else").unwrap(); let dest_s = dest.to_string_lossy().into_owned(); - let item = manager - .enqueue_video( + let item = glib::MainContext::default() + .block_on(manager.enqueue_video_async( "https://www.youtube.com/watch?v=gXtp6C-3JKo", Some(&dest_s), Some("clip.mp4"), @@ -2435,7 +2515,7 @@ fn enqueue_video_keeps_name_when_file_exists() { is_live: false, playlist_item_id: None, }, - ) + )) .expect("video enqueue"); assert_eq!(item.filename(), "clip.mp4"); assert_eq!( @@ -2469,6 +2549,11 @@ fn restored_unstarted_row_refuses_existing_dest() { // Explicit never-started (new queue file): the bytes are foreign. stored.started = Some(false); let item = manager.restore_existing(&stored).expect("restore"); + assert!(wait_for_preparation( + &manager, + item.id(), + Duration::from_secs(15) + )); assert_eq!(item.status(), DownloadStatus::Failed); assert_eq!(item.detail(), crate::engine_msg::DEST_EXISTS); assert_eq!( @@ -2589,8 +2674,8 @@ fn expand_playlist_rows_use_titled_subfolder() { let _ = std::fs::remove_dir_all(&dest); std::fs::create_dir_all(&dest).unwrap(); let dest_s = dest.to_string_lossy().into_owned(); - let parent = manager - .enqueue_video( + let parent = glib::MainContext::default() + .block_on(manager.enqueue_video_async( "https://www.youtube.com/playlist?list=PLx", Some(&dest_s), Some("parent.mp4"), @@ -2601,7 +2686,7 @@ fn expand_playlist_rows_use_titled_subfolder() { is_live: false, playlist_item_id: None, }, - ) + )) .expect("parent enqueue"); // Stage a Page source so the expansion reads quality/audio_only. manager.video_sources.borrow_mut().insert( @@ -2637,8 +2722,25 @@ fn expand_playlist_rows_use_titled_subfolder() { ] .into_boxed_slice(), }; - let (added, total, skipped_by_host_limit) = - manager.expand_playlist_rows(parent.id(), &parent, &pl); + assert!( + wait_for_preparation(&manager, parent.id(), Duration::from_secs(15)), + "playlist parent preparation did not settle" + ); + let generation = manager + .epoch + .borrow() + .get(&parent.id()) + .copied() + .unwrap_or(0); + let (added, total, skipped_by_host_limit) = glib::MainContext::default().block_on( + crate::download::DownloadManager::expand_playlist_rows_async( + Rc::downgrade(&manager), + parent.id(), + generation, + &parent, + &pl, + ), + ); assert_eq!((added, total, skipped_by_host_limit), (2, 2, 0)); let sub = dest.join("My Mix: Vol. 1"); assert!(sub.is_dir()); @@ -2715,7 +2817,7 @@ fn restrict_filename_ascii_folds_to_ascii() { #[test] fn restore_keeps_exact_filename() { - let _lock = QUEUE_FILE_LOCK.lock().unwrap(); + let _lock = lock_queue_file(); let _qf = test_queue_file("restore-exact"); test_settings(); let dir = std::env::temp_dir().join(format!("grab-restore-{}", std::process::id())); @@ -2847,7 +2949,7 @@ fn history_restore_roundtrip() { #[test] fn selection_survives_persist_restore() { - let _lock = QUEUE_FILE_LOCK.lock().unwrap(); + let _lock = lock_queue_file(); let qf = test_queue_file("selection-roundtrip"); let settings = test_settings(); settings.set_int("max-concurrent", 1).unwrap(); @@ -2908,7 +3010,7 @@ fn sane_filenames() { #[test] fn restore_rejects_bad_filenames() { - let _lock = QUEUE_FILE_LOCK.lock().unwrap(); + let _lock = lock_queue_file(); let _qf = test_queue_file("restore-bad"); let settings = test_settings(); let manager = DownloadManager::new(gio::ListStore::new::(), settings); @@ -2998,15 +3100,11 @@ fn cancel_frees_slot_immediately() { manager.cancel(41); assert_eq!(a.status(), DownloadStatus::Cancelled); assert!(!manager.running.borrow().contains_key(&41)); - assert_eq!(b.status(), DownloadStatus::Downloading); - assert!(manager.running.borrow().contains_key(&42)); - // Drain B's UI future here: already queued, or another test's loop aborts on thread-affinity. - let ctx = glib::MainContext::default(); - let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10); - while manager.running.borrow().contains_key(&42) && std::time::Instant::now() < deadline { - ctx.iteration(false); - std::thread::sleep(std::time::Duration::from_millis(1)); - } + assert!( + wait_for_started(&manager, 42, Duration::from_secs(10)), + "cancel must release the slot and start the next queued row" + ); + drain_engine(&manager, 42); assert!(!manager.running.borrow().contains_key(&42)); assert_eq!(b.status(), DownloadStatus::Failed); } @@ -3030,15 +3128,13 @@ fn pause_starts_next_queued() { manager.pause(61); assert_eq!(a.status(), DownloadStatus::Paused); assert!(!manager.running.borrow().contains_key(&61)); - assert_eq!(b.status(), DownloadStatus::Downloading); - assert!(manager.running.borrow().contains_key(&62)); - let ctx = glib::MainContext::default(); - let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10); - while manager.running.borrow().contains_key(&62) && std::time::Instant::now() < deadline { - ctx.iteration(false); - std::thread::sleep(std::time::Duration::from_millis(1)); - } + assert!( + wait_for_started(&manager, 62, Duration::from_secs(10)), + "pause must release the slot and start the next queued row" + ); + drain_engine(&manager, 62); assert!(!manager.running.borrow().contains_key(&62)); + assert_eq!(b.status(), DownloadStatus::Failed); } #[test] @@ -3060,22 +3156,20 @@ fn raising_max_concurrent_starts_queued() { assert_eq!(b.status(), DownloadStatus::Queued); // The preferences SpinRow writes this key; the queued row must start with no other event. settings.set_int("max-concurrent", 2).unwrap(); - assert_eq!(b.status(), DownloadStatus::Downloading); - assert!(manager.running.borrow().contains_key(&64)); - let ctx = glib::MainContext::default(); - let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10); - while manager.running.borrow().contains_key(&64) && std::time::Instant::now() < deadline { - ctx.iteration(false); - std::thread::sleep(std::time::Duration::from_millis(1)); - } + assert!( + wait_for_started(&manager, 64, Duration::from_secs(10)), + "raising the concurrency limit must start the newly eligible row" + ); + drain_engine(&manager, 64); assert!(!manager.running.borrow().contains_key(&64)); + assert_eq!(b.status(), DownloadStatus::Failed); // Restore: the memory backend is shared across tests. settings.set_int("max-concurrent", 3).unwrap(); } #[test] fn defer_yields_slot_and_goes_last() { - let _lock = QUEUE_FILE_LOCK.lock().unwrap(); + let (_lock, _loop) = test_locks(); let _qf = test_queue_file("defer"); let settings = test_settings(); settings.set_int("max-concurrent", 1).unwrap(); @@ -3106,12 +3200,15 @@ fn defer_yields_slot_and_goes_last() { // Cancel first so no queued row remains for the backend restore to spawn. manager.cancel_all(); assert!(!manager.running.borrow().contains_key(&72)); + // cancel_all schedules partial-file cleanup on the GLib context. Drain + // the workers and their continuations before another test owns the context. + drain_cancel_cleanups(&manager); settings.set_int("max-concurrent", 3).unwrap(); } #[test] fn lowering_max_concurrent_parks_newest() { - let _lock = QUEUE_FILE_LOCK.lock().unwrap(); + let _lock = lock_queue_file(); let _qf = test_queue_file("preempt"); let settings = test_settings(); settings.set_int("max-concurrent", 2).unwrap(); @@ -3201,7 +3298,7 @@ fn restore_preserves_intent() { #[test] fn restored_status_mapping() { - let _lock = QUEUE_FILE_LOCK.lock().unwrap(); + let _lock = lock_queue_file(); let _qf = test_queue_file("restored-mapping"); let settings = test_settings(); // Occupy the only slot so restored rows never spawn an engine. @@ -3235,7 +3332,7 @@ fn restored_status_mapping() { #[test] fn retry_persists_immediately() { - let _lock = QUEUE_FILE_LOCK.lock().unwrap(); + let _lock = lock_queue_file(); let qf = test_queue_file("retry-persist"); let settings = test_settings(); settings.set_int("max-concurrent", 1).unwrap(); @@ -3590,12 +3687,30 @@ fn stale_pump_future_ignores_respawned_row() { .enqueue(&url, Some(&dest), Some("h.bin")) .unwrap_or_else(|e| abort(&server, &e)); let id = item.id(); + assert!( + wait_for_started(&manager, id, Duration::from_secs(15)), + "download preparation did not start the engine" + ); assert_eq!(item.status(), DownloadStatus::Downloading); // Let the first engine get going, then defer: park aborts it and a // second engine takes over the same row. std::thread::sleep(std::time::Duration::from_millis(500)); manager.defer(id); - assert_eq!(item.status(), DownloadStatus::Downloading); + assert_eq!(item.status(), DownloadStatus::Queued); + // Preparation is now asynchronous; wait until the replacement engine owns + // the row before checking that the stale pump cannot overwrite its state. + let restart_deadline = std::time::Instant::now() + Duration::from_secs(15); + while (!manager.running.borrow().contains_key(&id) + || item.status() != DownloadStatus::Downloading) + && std::time::Instant::now() < restart_deadline + { + glib::MainContext::default().iteration(false); + std::thread::sleep(Duration::from_millis(1)); + } + assert!( + manager.running.borrow().contains_key(&id) && item.status() == DownloadStatus::Downloading, + "replacement engine did not start after defer" + ); // Let the aborted engine die and the hanging one settle, then pump: // the stale future's tail runs here and must stay silent. std::thread::sleep(std::time::Duration::from_millis(500)); @@ -3618,7 +3733,11 @@ fn stale_pump_future_ignores_respawned_row() { } manager.cancel(id); let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10); - while manager.running.borrow().contains_key(&id) && std::time::Instant::now() < deadline { + while (manager.running.borrow().contains_key(&id) + || manager.preparing.borrow().contains(&id) + || manager.preparation_handles.borrow().contains_key(&id)) + && std::time::Instant::now() < deadline + { ctx.iteration(false); std::thread::sleep(std::time::Duration::from_millis(1)); } @@ -3760,13 +3879,15 @@ fn segmented_pause_resume_keeps_bytes() { .enqueue(&url, Some(&dest), Some("big.bin")) .unwrap_or_else(|e| abort(&server, &e)); let id = item.id(); - // Wait for one 1 MB piece (5%), then pause. Status check is in the loop condition - // so it cannot slip to Done before pause() with no await in between. + // Wait for asynchronous preparation to start the engine, then reach one + // 1 MB piece (5%) before pausing. A queued row is not evidence of completion. let mut waited = 0; - while item.status() == DownloadStatus::Downloading - && item.progress() < 0.05 - && waited < 2000 + while item.status() == DownloadStatus::Queued + || (item.status() == DownloadStatus::Downloading && item.progress() < 0.05) { + if waited >= 2000 { + break; + } glib::timeout_future(std::time::Duration::from_millis(5)).await; waited += 1; } @@ -3811,7 +3932,7 @@ fn segmented_pause_resume_keeps_bytes() { #[test] fn bitmap_persists_across_managers() { - let _lock = QUEUE_FILE_LOCK.lock().unwrap(); + let _lock = lock_queue_file(); let qf = test_queue_file("segpersist"); let settings = test_settings(); let m1 = DownloadManager::new(gio::ListStore::new::(), settings.clone()); @@ -3970,10 +4091,12 @@ fn shutdown_restart_resumes_segmented() { .unwrap_or_else(|e| abort(&server, &e)); let id = item.id(); let mut waited = 0; - while item.status() == DownloadStatus::Downloading - && item.progress() < 0.05 - && waited < 2000 + while item.status() == DownloadStatus::Queued + || (item.status() == DownloadStatus::Downloading && item.progress() < 0.05) { + if waited >= 2000 { + break; + } glib::timeout_future(std::time::Duration::from_millis(5)).await; waited += 1; } @@ -4232,10 +4355,14 @@ fn restart_with_smaller_file_keeps_partial() { .restore_existing(&stored) .unwrap_or_else(|e| abort(&server, &e)); let id = item.id(); - // Drain the engine's pump future on this thread (see MAIN_LOOP_LOCK). + // Drain both asynchronous destination preparation and the engine/pump on this thread. let ctx = glib::MainContext::default(); let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30); - while manager.running.borrow().contains_key(&id) && std::time::Instant::now() < deadline { + while (manager.running.borrow().contains_key(&id) + || manager.preparing.borrow().contains(&id) + || manager.preparation_handles.borrow().contains_key(&id)) + && std::time::Instant::now() < deadline + { ctx.iteration(false); std::thread::sleep(std::time::Duration::from_millis(1)); } @@ -4280,7 +4407,7 @@ fn aggregate_downsamples_by_half() { #[test] fn piece_bitmap_prefers_segment_then_torrent_then_bytes() { - let _lock = QUEUE_FILE_LOCK.lock().unwrap(); + let _lock = lock_queue_file(); let _qf = test_queue_file("piece-bitmap"); let settings = test_settings(); let manager = DownloadManager::new(gio::ListStore::new::(), settings); @@ -5251,7 +5378,7 @@ exit 0 #[test] fn clear_finished_drops_only_done() { - let _lock = QUEUE_FILE_LOCK.lock().unwrap(); + let _lock = lock_queue_file(); let qf = test_queue_file("clear-finished"); let settings = test_settings(); let m = DownloadManager::new(gio::ListStore::new::(), settings); @@ -5353,7 +5480,7 @@ fn restore_dedups_finished_urls() { fn drop_finished_duplicates_keeps_active_and_newest() { // The finish hook's policy, directly: the just-finished row stays, // older Done rows for the URL go, everything else is user intent. - let _lock = QUEUE_FILE_LOCK.lock().unwrap(); + let _lock = lock_queue_file(); let _qf = test_queue_file("history-dedup-live"); let settings = test_settings(); let m = DownloadManager::new(gio::ListStore::new::(), settings); @@ -5398,8 +5525,8 @@ fn enqueue_video_reserves_part_namespaced_stems() { std::fs::create_dir_all(&dest).unwrap(); std::fs::write(dest.join("Clip.video.mp4"), b"foreign").unwrap(); let dest_s = dest.to_string_lossy().into_owned(); - let item = manager - .enqueue_video( + let item = glib::MainContext::default() + .block_on(manager.enqueue_video_async( "https://www.youtube.com/watch?v=gXtp6C-3JKo", Some(&dest_s), Some("Clip.mp4"), @@ -5410,7 +5537,7 @@ fn enqueue_video_reserves_part_namespaced_stems() { is_live: false, playlist_item_id: None, }, - ) + )) .expect("video enqueue"); assert_eq!(item.filename(), "Clip (1).mp4"); // The foreign file is untouched and the fresh stem is unreserved. @@ -5484,11 +5611,13 @@ fn delete_download_trashes_video_sidecars() { .block_on(manager.delete_download_async(11)) .is_ok() ); + quiesce(&glib::MainContext::default()); assert!( glib::MainContext::default() .block_on(manager.delete_download_async(12)) .is_ok() ); + quiesce(&glib::MainContext::default()); // Video + its sidecars trashed; plain row's same-named srt stays. for n in ["Clip.mp4", "Clip.en.srt", "Clip.fr.srt", "Other.mp4"] { assert!(!dir.join(n).exists(), "{n} must be trashed"); @@ -5517,8 +5646,8 @@ fn enqueue_video_reserves_subtitle_sidecar_stems() { std::fs::create_dir_all(&dest).unwrap(); std::fs::write(dest.join("Clip.en.srt"), b"foreign").unwrap(); let dest_s = dest.to_string_lossy().into_owned(); - let item = manager - .enqueue_video( + let item = glib::MainContext::default() + .block_on(manager.enqueue_video_async( "https://www.youtube.com/watch?v=gXtp6C-3JKo", Some(&dest_s), Some("Clip.mp4"), @@ -5529,7 +5658,7 @@ fn enqueue_video_reserves_subtitle_sidecar_stems() { is_live: false, playlist_item_id: None, }, - ) + )) .expect("video enqueue"); assert_eq!(item.filename(), "Clip (1).mp4"); assert_eq!(std::fs::read(dest.join("Clip.en.srt")).unwrap(), b"foreign"); @@ -5553,18 +5682,18 @@ fn enqueue_video_accepts_unlisted_url() { is_live: false, playlist_item_id: None, }; - let item = manager - .enqueue_video( + let item = glib::MainContext::default() + .block_on(manager.enqueue_video_async( "https://example.com/f.iso", Some("/tmp/dl"), None, choices.clone(), - ) + )) .expect("probed intake"); assert_eq!(item.filename(), "f.iso"); assert!( - manager - .enqueue_video("not a url", Some("/tmp/dl"), None, choices) + glib::MainContext::default() + .block_on(manager.enqueue_video_async("not a url", Some("/tmp/dl"), None, choices)) .is_err() ); // Drain the spawned worker (fails fast without tools) so no woken @@ -5618,11 +5747,23 @@ fn a_video_attempt_gets_a_gate_and_a_plain_row_does_not() { backend: None, }); let item = manager.restore_existing(&row).expect("video row"); + // Destination validation is asynchronous now. Wait until the attempt has + // passed preparation and installed its delivery gate before asserting the + // fail-fast proxy path; don't assume GTK setup runs in the enqueue call. + let context = glib::MainContext::default(); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + while manager.gate_for(item.id()).is_none() + && item.status() == DownloadStatus::Queued + && std::time::Instant::now() < deadline + { + context.iteration(false); + std::thread::sleep(std::time::Duration::from_millis(1)); + } assert!( manager.gate_for(item.id()).is_some(), - "a video row with no gate cannot arbitrate delivery, so a removal \ - could not stop it placing a file" + "a video attempt must install its delivery gate before engine validation" ); + quiesce(&context); manager .settings() @@ -5866,8 +6007,8 @@ fn a_reserved_destination_forces_video_intake_to_dedupe() { std::fs::create_dir_all(&dest).unwrap(); let dest_s = dest.to_string_lossy().into_owned(); manager.reserve_dest(&dest.join("Clip.mp4")); - let item = manager - .enqueue_video( + let item = glib::MainContext::default() + .block_on(manager.enqueue_video_async( "https://www.youtube.com/watch?v=gXtp6C-3JKo", Some(&dest_s), Some("Clip.mp4"), @@ -5878,7 +6019,7 @@ fn a_reserved_destination_forces_video_intake_to_dedupe() { is_live: false, playlist_item_id: None, }, - ) + )) .expect("video enqueue"); assert_eq!( item.filename(), @@ -5932,8 +6073,8 @@ fn a_reserved_stem_forces_video_intake_to_dedupe() { std::fs::create_dir_all(&dest).unwrap(); let dest_s = dest.to_string_lossy().into_owned(); manager.reserve_dest(&dest.join("Clip.mp4")); - let item = manager - .enqueue_video( + let item = glib::MainContext::default() + .block_on(manager.enqueue_video_async( "https://www.youtube.com/watch?v=gXtp6C-3JKo", Some(&dest_s), Some("Clip.m4a"), @@ -5944,7 +6085,7 @@ fn a_reserved_stem_forces_video_intake_to_dedupe() { is_live: false, playlist_item_id: None, }, - ) + )) .expect("video enqueue"); assert_eq!( item.filename(), @@ -7578,7 +7719,9 @@ fn enqueue_gallery_resolves_dest_and_id() { let mut job = gallery_job_fixture(); job.dest_dir = std::path::PathBuf::from(dest); job.item_id = 0; // placeholder - let item = m.enqueue_gallery(&job).expect("enqueue"); + let item = glib::MainContext::default() + .block_on(m.enqueue_gallery_async(&job)) + .expect("enqueue"); // Destination resolved to an absolute path. assert!(std::path::Path::new(&item.dest_dir()).is_absolute()); assert!(!item.dest_dir().is_empty()); @@ -7657,7 +7800,9 @@ fn gallery_remove_discards_handoff_cookies() { crate::extension_server::store_handoff_cookies(&job.page_url, path, None); assert!(crate::extension_server::live_handoff_cookie_path(&job.page_url, "none").is_some()); // Enqueue the gallery, then remove it. - let item = m.enqueue_gallery(&job).expect("enqueue"); + let item = glib::MainContext::default() + .block_on(m.enqueue_gallery_async(&job)) + .expect("enqueue"); let id = item.id(); // Drain the spawned task to avoid tripping the glib thread guard. drain_engine(&m, id); @@ -7780,7 +7925,9 @@ fn gallery_enqueue_uses_unique_dir_when_title_collides() { let mut job = gallery_job_fixture(); job.dest_dir = tmp.clone(); job.title = "Test Gallery".to_string(); - let item = m.enqueue_gallery(&job).expect("enqueue"); + let item = glib::MainContext::default() + .block_on(m.enqueue_gallery_async(&job)) + .expect("enqueue"); // Must NOT use the pre-existing directory. assert_ne!(item.filename(), "Test Gallery"); assert!(item.filename().starts_with("Test Gallery-")); @@ -7793,7 +7940,9 @@ fn gallery_enqueue_uses_unique_dir_when_title_collides() { assert!(preexisting.join("unrelated.txt").exists()); // A second gallery with the same title (neither downloaded yet, so no // dir on disk) must get a *different* unique name — in-memory collision. - let item2 = m.enqueue_gallery(&job).expect("enqueue second"); + let item2 = glib::MainContext::default() + .block_on(m.enqueue_gallery_async(&job)) + .expect("enqueue second"); assert_ne!(item2.filename(), item.filename()); assert_ne!(item2.filename(), "Test Gallery"); // Drain both spawned engines so no woken tail trips a later test's @@ -7988,6 +8137,7 @@ fn delete_download_legacy_gallery_row_keeps_unrelated_folder() { .block_on(manager.delete_download_async(7)) .is_ok() ); + quiesce(&glib::MainContext::default()); assert_eq!(store.n_items(), 0); // The unrelated folder and its content survive. assert!(decoy.join("keep.txt").exists()); @@ -8015,6 +8165,7 @@ fn delete_download_gallery_row_without_job_keeps_folder() { .block_on(manager.delete_download_async(7)) .is_ok() ); + quiesce(&glib::MainContext::default()); assert_eq!(store.n_items(), 0); assert!(folder.join("keep.txt").exists()); let _ = std::fs::remove_dir_all(&dir); @@ -8044,6 +8195,7 @@ fn delete_download_new_gallery_row_trashes_subfolder() { .block_on(manager.delete_download_async(7)) .is_ok() ); + quiesce(&glib::MainContext::default()); assert_eq!(store.n_items(), 0); assert!(!sub.exists()); // Undo the test's own Trash litter. @@ -8080,6 +8232,7 @@ fn delete_download_after_gallery_rename_trashes_renamed_folder() { .block_on(manager.delete_download_async(7)) .is_ok() ); + quiesce(&glib::MainContext::default()); assert_eq!(store.n_items(), 0); // The renamed folder — the one Grab owns — must be trashed. assert!(!dir.join("renamed").exists()); @@ -8124,3 +8277,58 @@ fn restoring_public_only_row_reinstates_its_host_policy() { let _ = std::fs::remove_file(queue_file); } + +#[test] +fn start_next_counts_preparing_rows_against_the_cap() { + let (_lock, _loop) = test_locks(); + let _qf = test_queue_file("preparing-cap"); + let settings = test_settings(); + settings.set_int("max-concurrent", 2).unwrap(); + let manager = DownloadManager::new(gio::ListStore::new::(), settings); + + for i in 0..5 { + manager + .enqueue(&format!("http://127.0.0.1:9/f{i}"), Some("/tmp"), None) + .unwrap(); + } + + assert!( + manager.preparing.borrow().len() + manager.running.borrow().len() <= 2, + "filesystem preparation must consume a scheduler slot" + ); + assert_eq!(manager.preparing.borrow().len(), 2); + manager.cancel_all(); + + let context = glib::MainContext::default(); + let deadline = std::time::Instant::now() + Duration::from_secs(15); + while !manager.preparing.borrow().is_empty() && std::time::Instant::now() < deadline { + context.iteration(false); + std::thread::sleep(Duration::from_millis(1)); + } + assert!( + manager.preparing.borrow().is_empty(), + "preparation workers must drain" + ); + quiesce(&context); +} + +#[test] +fn finished_name_adoption_retries_after_stale_listing_collision() { + let dir = std::env::temp_dir().join(format!("grab-finished-name-race-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + let source = dir.join("tmp.bin"); + let occupied = dir.join("a.mp4"); + std::fs::write(&source, b"downloaded").unwrap(); + std::fs::write(&occupied, b"foreign").unwrap(); + + // Simulate a stale listing that says every name is free. The atomic rename + // must reject a.mp4 and retry with a distinct name, never overwrite it. + let adopted = DownloadManager::adopt_final_name(&source, &dir, "a.mp4", |_| false) + .expect("a later collision-free candidate should be adopted"); + assert_ne!(adopted, "a.mp4"); + assert!(dir.join(&adopted).exists()); + assert_eq!(std::fs::read(&occupied).unwrap(), b"foreign"); + assert_eq!(std::fs::read(dir.join(adopted)).unwrap(), b"downloaded"); + let _ = std::fs::remove_dir_all(&dir); +} diff --git a/src/inline_add.rs b/src/inline_add.rs index b4674f347..036955f49 100644 --- a/src/inline_add.rs +++ b/src/inline_add.rs @@ -844,9 +844,9 @@ fn submit_probed_single( .is_empty() .then(|| default_name_for(settings, &v.title, audio_only)); let name = if typed.is_empty() { - auto.as_deref() + auto } else { - Some(typed.as_str()) + Some(typed.clone()) }; // Exact picks pin the format with its height as fallback, so a dropped pin still // degrades to the chosen height; audio-only rows drop the pin, and Automatic (no @@ -867,28 +867,35 @@ fn submit_probed_single( .unwrap_or_else(|| manager.settings().video_quality()), None => manager.settings().video_quality(), }; - enqueue_and_close( + let page_url = v.page_url.clone(); + let is_live = v.is_live; + let enqueue_manager = Rc::clone(manager); + let error_step = Rc::clone(step); + let error_lookup_add = Rc::clone(lookup_add); + enqueue_and_close_async( manager, dest, close_card, scheduled_at, - |d| { - manager.enqueue_video( - &v.page_url, - d, - name, - crate::media_types::VideoChoices { - quality, - audio_only, - video_format_id: format_id, - is_live: v.is_live, - playlist_item_id: None, - }, - ) + move |d| async move { + enqueue_manager + .enqueue_video_async_batched( + &page_url, + d.as_deref(), + name.as_deref(), + crate::media_types::VideoChoices { + quality, + audio_only, + video_format_id: format_id, + is_live, + playlist_item_id: None, + }, + ) + .await }, - |e| { - show_video_error(step, e); - set_lookup_add(lookup_add, true); + move |e| { + show_video_error(&error_step, e); + set_lookup_add(&error_lookup_add, true); }, ); } @@ -999,19 +1006,22 @@ fn submit_gallery_all( speed_limit: crate::gallery_runner::gallery_speed_limit(settings), range: None, }; - enqueue_and_close( + let enqueue_manager = Rc::clone(manager); + let error_step = Rc::clone(step); + let error_lookup_add = Rc::clone(lookup_add); + enqueue_and_close_async( manager, dest, close_card, scheduled_at, - |d| { + move |d| async move { let mut job = job; - job.dest_dir = std::path::PathBuf::from(d.unwrap_or(".")); - manager.enqueue_gallery(&job) + job.dest_dir = std::path::PathBuf::from(d.unwrap_or_else(|| ".".to_string())); + enqueue_manager.enqueue_gallery_async_batched(&job).await }, - |e| { - show_video_error(step, e); - set_lookup_add(lookup_add, true); + move |e| { + show_video_error(&error_step, e); + set_lookup_add(&error_lookup_add, true); }, ); } @@ -1062,6 +1072,37 @@ pub(crate) fn enqueue_and_close( } } +pub(crate) fn enqueue_and_close_async( + _manager: &Rc, + dest: &Rc>, + close_card: &Rc, + scheduled_at: Option, + enqueue: impl FnOnce(Option) -> F + 'static, + on_err: impl FnOnce(&str) + 'static, +) where + F: std::future::Future> + + 'static, +{ + let dest = Rc::clone(dest); + let close_card = Rc::clone(close_card); + glib::spawn_future_local(async move { + // Own the destination value before awaiting; close_card re-borrows the + // destination when it resets the form after a successful enqueue. + let destination = Some(dest.borrow().clone()); + // Preparation runs without a batch open. The intake API opens one only + // for its synchronous commit and returns the guard so scheduling is set + // before the queue is allowed to start the row. + match enqueue(destination).await { + Ok((item, guard)) => { + apply_scheduled_at(&item, scheduled_at); + close_card(); + drop(guard); + } + Err(error) => on_err(&error), + } + }); +} + /// Queue a probed link as a plain file and collapse the card: the fallback when /// extraction finds no playable media on an unlisted page. `Err` when plain intake rejects the URL. fn queue_plain( @@ -1196,7 +1237,7 @@ fn push_playlist_items_page( let picker_page_weak = picker_page_weak.clone(); let fallback_to_picker = fallback_to_picker.clone(); let add_btn_click = add_btn.clone(); - add_btn_click.connect_clicked(move |_| { + add_btn_click.clone().connect_clicked(move |_| { // Guard against re-entry: Enter emits clicked directly, bypassing // the sensitivity check, so a second press while the import is in // flight would enqueue duplicates. @@ -1493,7 +1534,8 @@ fn push_torrent_picker_page( let grid = grid.clone(); let fallback_to_picker = fallback_to_picker.clone(); let add_btn_click = add_btn.clone(); - add_btn_click.connect_clicked(move |_| { + let add_btn_signal = add_btn_click.clone(); + add_btn_signal.connect_clicked(move |_| { let selected: Vec = grid.selected_indices(); if selected.is_empty() { fallback_to_picker(); @@ -1506,13 +1548,28 @@ fn push_torrent_picker_page( let only = (selected.len() < grid.len()).then_some(selected); let show_picker_on_error = fallback_to_picker.clone(); let error_caption_on_error = error_caption.clone(); - enqueue_and_close( + add_btn_click.set_sensitive(false); + let add_btn_on_error = add_btn_click.clone(); + let enqueue_manager = Rc::clone(&manager); + let enqueue_bytes = bytes.clone(); + let enqueue_name = file_name.clone(); + enqueue_and_close_async( &manager, &dest_dir, &close_card, scheduled_at.get(), - |d| manager.enqueue_torrent_file(bytes.clone(), &file_name, d, only), + move |d| async move { + enqueue_manager + .enqueue_torrent_file_async_batched( + enqueue_bytes, + &enqueue_name, + d.as_deref(), + only, + ) + .await + }, move |e| { + add_btn_on_error.set_sensitive(true); show_picker_on_error(); error_caption_on_error.set_text(e); error_caption_on_error.set_visible(true); @@ -1536,8 +1593,10 @@ fn wire_torrent_picker( scheduled_at: Rc>>, ) { let nav = nav.clone(); + let torrent_btn = torrent_btn.clone(); let scheduled_at_c = Rc::clone(&scheduled_at); - torrent_btn.connect_clicked(move |_| { + torrent_btn.clone().connect_clicked(move |_| { + let torrent_btn = torrent_btn.clone(); let m = manager.clone(); let dd = dest_dir.clone(); let close_card = close_card.clone(); @@ -1583,13 +1642,21 @@ fn wire_torrent_picker( } }; if entries.len() <= 1 { - enqueue_and_close( + torrent_btn.set_sensitive(false); + let enqueue_manager = Rc::clone(&m); + let torrent_btn_on_error = torrent_btn.clone(); + enqueue_and_close_async( &m, &dd, &close_card, scheduled_at.get(), - |d| m.enqueue_torrent_file(bytes, &name, d, None), - |e| { + move |d| async move { + enqueue_manager + .enqueue_torrent_file_async_batched(bytes, &name, d.as_deref(), None) + .await + }, + move |e| { + torrent_btn_on_error.set_sensitive(true); set_field_error(&torrent_row, e); }, ); diff --git a/src/inline_gallery.rs b/src/inline_gallery.rs index 7d1cb9735..1121473c1 100644 --- a/src/inline_gallery.rs +++ b/src/inline_gallery.rs @@ -347,25 +347,29 @@ pub(crate) fn push_gallery_picker_page( speed_limit: crate::gallery_runner::gallery_speed_limit(settings), range: Some(range), }; - // Batch guard defers start_next until the Scheduled status is set. - let _guard = manager.batch_guard(); - let mut job = base_job.clone(); - job.dest_dir = std::path::PathBuf::from(dest_dir.borrow().clone()); - match manager.enqueue_gallery(&job) { - Ok(item) => { - apply_scheduled_at(&item, scheduled_at); - // Don't destroy the currently emitting button/page in its - // own clicked callback. close_card already pops back to form; - // defer that teardown until the signal has unwound. - let close = close_card.clone(); - gtk4::glib::idle_add_local_once(move || close()); + let manager = Rc::clone(&manager); + let dest_dir = Rc::clone(&dest_dir); + let close_card = Rc::clone(&close_card); + let error_caption = error_caption.clone(); + let add_btn_guard = add_btn_guard.clone(); + gtk4::glib::spawn_future_local(async move { + let mut job = base_job; + job.dest_dir = std::path::PathBuf::from(dest_dir.borrow().clone()); + match manager.enqueue_gallery_async_batched(&job).await { + Ok((item, guard)) => { + apply_scheduled_at(&item, scheduled_at); + // Defer picker teardown until the clicked signal unwinds. + let close = close_card.clone(); + gtk4::glib::idle_add_local_once(move || close()); + drop(guard); + } + Err(e) => { + error_caption.set_text(&e); + error_caption.set_visible(true); + add_btn_guard.set_sensitive(true); + } } - Err(e) => { - error_caption.set_text(&e); - error_caption.set_visible(true); - add_btn_guard.set_sensitive(true); - } - } + }); }); } diff --git a/src/torrent.rs b/src/torrent.rs index daa91af6d..ade46526d 100644 --- a/src/torrent.rs +++ b/src/torrent.rs @@ -396,10 +396,7 @@ pub(crate) fn single_file_path(dest: &std::path::Path, url: &str) -> Option = selected.into_iter().collect(); + let keep: std::collections::HashSet = selected.iter().copied().collect(); for (i, entry) in entries.iter().enumerate() { if let Some(target) = deletion_target(folder, i, &keep, entry) { remove_file_and_prune_parents(&target, folder);