From 3a26e64fc1bd0450594ff54b6e402ee042f75d15 Mon Sep 17 00:00:00 2001 From: thedancingdeveloper <306930456+thedancingdeveloper@users.noreply.github.com> Date: Fri, 9 Oct 2026 23:10:42 +0000 Subject: [PATCH 01/23] fix(queue): flush article checkpoint for active downloads on graceful shutdown Ported from xbmc4lyfe/rustnzb#165 (addresses #162). On graceful shutdown, jobs that were mid-download are marked Queued, but their in-memory article checkpoint was never persisted, so a restart re-downloaded completed articles. shutdown() now writes the checkpoint for every interrupted download before the process exits. Co-authored-by: xbmc4lyfe <168791846+xbmc4lyfe@users.noreply.github.com> --- crates/nzb-web/src/queue_manager.rs | 83 +++++++++++++++++++++++++++++ 1 file changed, 83 insertions(+) diff --git a/crates/nzb-web/src/queue_manager.rs b/crates/nzb-web/src/queue_manager.rs index cfd2279d..fb90d3d3 100644 --- a/crates/nzb-web/src/queue_manager.rs +++ b/crates/nzb-web/src/queue_manager.rs @@ -4948,12 +4948,14 @@ impl QueueManager { // progress-handler task handles so they can be aborted after the // pool drains. let mut handles = Vec::new(); + let mut interrupted: HashSet = HashSet::new(); { let mut jobs = self.jobs.lock(); for (id, state) in jobs.iter_mut() { if state.job.status == JobStatus::Downloading { info!(job_id = %id, "Marking download for shutdown"); state.job.status = JobStatus::Queued; // Will resume on restart + interrupted.insert(id.clone()); } if let Some(handle) = state.progress_handle.take() { handles.push(handle); @@ -4990,6 +4992,30 @@ impl QueueManager { } } + // 5. Flush per-file article checkpoints for downloads that were + // mid-flight, so a restart resumes from the last completed + // article instead of re-downloading whole files. + if !interrupted.is_empty() { + let jobs = self.jobs.lock(); + let db = self.db.lock(); + for id in &interrupted { + let Some(state) = jobs.get(id) else { + continue; + }; + let checkpoint = checkpoint_for_job(&state.job, state.output_dir_claimed); + match serde_json::to_vec(&checkpoint) { + Ok(data) => { + if let Err(e) = db.queue_store_job_data(id, &data) { + error!(job_id = %id, error = %e, "Failed to flush article checkpoint on shutdown"); + } + } + Err(e) => { + error!(job_id = %id, error = %e, "Failed to serialize article checkpoint on shutdown"); + } + } + } + } + info!("Queue manager shutdown complete"); } @@ -5258,6 +5284,11 @@ mod global_pause_tests { fn insert_job(manager: &QueueManager, job: NzbJob) { let id = job.id.clone(); + manager + .db + .lock() + .queue_insert(&job) + .expect("insert queue row"); manager.jobs.lock().insert( id.clone(), JobState { @@ -5576,6 +5607,58 @@ mod global_pause_tests { assert_eq!(manager.queue_size(), 1); } + #[tokio::test] + async fn shutdown_flushes_article_checkpoint_for_active_downloads() { + let (manager, tempdir) = manager(); + let mut jb = job("drain-checkpoint", JobStatus::Downloading, tempdir.path()); + jb.downloaded_bytes = 5; + jb.articles_downloaded = 1; + jb.file_count = 1; + jb.article_count = 1; + jb.files.push(NzbFile { + id: "f1".to_string(), + filename: "file.bin".to_string(), + bytes: 5, + bytes_downloaded: 5, + is_par2: false, + par2_setname: None, + par2_vol: None, + par2_blocks: None, + assembled: false, + groups: Vec::new(), + articles: vec![Article { + message_id: "".to_string(), + segment_number: 1, + bytes: 5, + downloaded: true, + data_begin: None, + data_size: None, + crc32: None, + tried_servers: Vec::new(), + tries: 0, + }], + }); + insert_job(&manager, jb); + + manager.shutdown().await; + + assert_eq!( + manager.get_job("drain-checkpoint").unwrap().status, + JobStatus::Queued + ); + let data = manager + .db + .lock() + .queue_load_job_data("drain-checkpoint") + .expect("checkpoint persisted"); + let checkpoint: JobCheckpoint = + serde_json::from_slice(data.as_deref().expect("checkpoint bytes")) + .expect("valid checkpoint"); + assert_eq!(checkpoint.articles_downloaded, 1); + assert_eq!(checkpoint.downloaded_bytes, 5); + assert_eq!(checkpoint.files.get("file.bin"), Some(&vec![1u32])); + } + #[tokio::test] async fn global_pause_blocks_job_resume_and_preserves_manual_pause() { let (manager, tempdir) = manager(); From 0a98b6e86ba33a677786451ccba1f54b7fde83fa Mon Sep 17 00:00:00 2001 From: thedancingdeveloper <306930456+thedancingdeveloper@users.noreply.github.com> Date: Fri, 9 Oct 2026 23:26:57 +0000 Subject: [PATCH 02/23] fix(startup): refuse second instance on the same data_dir Ported from xbmc4lyfe/rustnzb#166 Co-Authored-By: xbmc4lyfe --- crates/nzb-web/src/startup.rs | 131 +++++++++++++++++++++++++++++++++- 1 file changed, 129 insertions(+), 2 deletions(-) diff --git a/crates/nzb-web/src/startup.rs b/crates/nzb-web/src/startup.rs index b47fc7c1..c1c17105 100644 --- a/crates/nzb-web/src/startup.rs +++ b/crates/nzb-web/src/startup.rs @@ -98,6 +98,103 @@ pub struct StartupResult { pub state: Arc, pub queue_manager: Arc, pub log_buffer: LogBuffer, + /// Held for the lifetime of the process so a second instance on the + /// same data_dir refuses to start. + _instance_lock: InstanceLock, +} + +/// Exclusive lock on `/rustnzb.lock` ensuring only one rustnzb +/// instance uses a given data directory at a time. Released on drop. +pub struct InstanceLock { + #[cfg(unix)] + _file: std::fs::File, + #[cfg(windows)] + _file: std::fs::File, +} + +#[cfg(unix)] +impl InstanceLock { + fn acquire(data_dir: &Path) -> anyhow::Result { + use std::os::unix::io::AsRawFd; + + let file = open_lock_options() + .open(lock_path(data_dir)) + .with_context(|| format!("failed to open {}", lock_path(data_dir).display()))?; + let fd = file.as_raw_fd(); + let ret = unsafe { libc::flock(fd, libc::LOCK_EX | libc::LOCK_NB) }; + if ret != 0 { + let err = std::io::Error::last_os_error(); + if err.kind() == std::io::ErrorKind::WouldBlock { + anyhow::bail!( + "another rustnzb instance is using {}; stop it or choose a different data_dir", + data_dir.display() + ); + } + return Err(err).context(format!("failed to lock {}", lock_path(data_dir).display())); + } + Ok(Self { _file: file }) + } +} + +#[cfg(windows)] +impl InstanceLock { + fn acquire(data_dir: &Path) -> anyhow::Result { + use std::os::windows::fs::OpenOptionsExt; + + // share_mode(0) denies all sharing: a second open fails outright. + let file = std::fs::OpenOptions::new() + .read(true) + .write(true) + .create(true) + .share_mode(0) + .open(lock_path(data_dir)) + .map_err(|err| { + if err.raw_os_error() == Some(32) { + anyhow::anyhow!( + "another rustnzb instance is using {}; stop it or choose a different data_dir", + data_dir.display() + ) + } else { + anyhow::Error::new(err).context(format!( + "failed to lock {}", + lock_path(data_dir).display() + )) + } + })?; + Ok(Self { _file: file }) + } +} + +fn lock_path(data_dir: &Path) -> PathBuf { + data_dir.join("rustnzb.lock") +} + +fn open_lock_options() -> std::fs::OpenOptions { + let mut options = std::fs::OpenOptions::new(); + options.read(true).write(true).create(true); + // Deliberately no truncate: the lock file's contents are irrelevant and + // truncating would churn the file on every start. + options +} + +#[cfg(unix)] +impl Drop for InstanceLock { + fn drop(&mut self) { + use std::os::unix::io::AsRawFd; + unsafe { libc::flock(self._file.as_raw_fd(), libc::LOCK_UN) }; + } +} + +#[cfg(windows)] +impl Drop for InstanceLock { + fn drop(&mut self) { + // Closing the file handle releases the exclusive share. + } +} + +/// Acquire the single-instance lock for `data_dir`. +pub fn acquire_instance_lock(data_dir: &Path) -> anyhow::Result { + InstanceLock::acquire(data_dir) } /// Initialize the rustnzb engine: load config, open database, @@ -153,8 +250,12 @@ pub async fn initialize( warn_unsafe_categories(&config); - // Ensure directories exist + // Ensure directories exist. The instance lock is taken before any + // side-effect directories are created, so a second instance on the same + // data_dir fails without leaving anything behind. create_data_dir(&config.general.data_dir)?; + let instance_lock = acquire_instance_lock(&config.general.data_dir) + .context("failed to acquire single-instance lock")?; create_data_dir(&config.general.incomplete_dir)?; create_data_dir(&config.general.complete_dir)?; @@ -260,12 +361,13 @@ pub async fn initialize( state, queue_manager, log_buffer, + _instance_lock: instance_lock, }) } #[cfg(test)] mod tests { - use super::{create_data_dir, sanitize_loaded_config}; + use super::{acquire_instance_lock, create_data_dir, sanitize_loaded_config}; use crate::nzb_core::config::AppConfig; use crate::nzb_core::config::ServerConfig; @@ -313,6 +415,31 @@ mod tests { ); } + #[test] + fn second_instance_on_same_data_dir_fails_before_side_effects() { + let tmp = tempfile::tempdir().unwrap(); + let data_dir = tmp.path().join("data"); + // initialize always creates the data_dir before acquiring the lock; + // mirror that here so the lock file has a parent to live in. + std::fs::create_dir_all(&data_dir).unwrap(); + + let first = acquire_instance_lock(&data_dir).unwrap(); + + let second = match acquire_instance_lock(&data_dir) { + Err(e) => e, + Ok(_) => panic!("second acquire should fail while first lock is held"), + }; + let debug_text = format!("{second:?}"); + assert!( + debug_text.contains("another rustnzb instance"), + "error should mention the other instance, got: {debug_text}" + ); + + // Releasing the first lock lets a later instance acquire it again. + drop(first); + acquire_instance_lock(&data_dir).unwrap(); + } + #[test] fn sanitize_loaded_config_trims_server_fields() { let mut config = AppConfig::default(); From 8b4341a946babf519cc308f19eb6cfd003557f9f Mon Sep 17 00:00:00 2001 From: thedancingdeveloper <306930456+thedancingdeveloper@users.noreply.github.com> Date: Sat, 10 Oct 2026 00:35:54 +0000 Subject: [PATCH 03/23] fix: stop retry/restart double-counting downloaded bytes Downloaded bytes carried over from a previous attempt of the same job were being re-counted when computing average speed and history rows, inflating reported download speed. - record carried bytes at admission and subtract them from history statistics (nzb-core history_insert_with_carried_bytes) - persist a job's server stats and accumulated active download time in its checkpoint so a restart resumes accounting instead of resetting it (nzb-dispatch job_active_secs, queue manager checkpoint fields) - add integration tests covering retry and restart accounting Co-Authored-By: xbmc4lyfe --- Cargo.lock | 22 +-- Cargo.toml | 2 +- crates/nzb-core/src/db.rs | 34 +++- crates/nzb-dispatch/Cargo.toml | 2 +- crates/nzb-dispatch/src/dispatch_engine.rs | 11 ++ crates/nzb-dispatch/src/download_engine.rs | 26 +++ crates/nzb-web/src/queue_manager.rs | 133 +++++++++++++- crates/nzb-web/tests/harness_statistics.rs | 195 +++++++++++++++++++++ 8 files changed, 403 insertions(+), 22 deletions(-) create mode 100644 crates/nzb-web/tests/harness_statistics.rs diff --git a/Cargo.lock b/Cargo.lock index 200f5a35..a6707ae2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -102,7 +102,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -113,7 +113,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -681,7 +681,7 @@ dependencies = [ "libc", "option-ext", "redox_users", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -735,7 +735,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -1753,7 +1753,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -1811,7 +1811,7 @@ dependencies = [ [[package]] name = "nzb-dispatch" -version = "0.2.8" +version = "0.2.9" dependencies = [ "anyhow", "arc-swap", @@ -2460,7 +2460,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -2765,7 +2765,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -3130,7 +3130,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -3256,7 +3256,7 @@ dependencies = [ "getrandom 0.4.3", "once_cell", "rustix", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -4005,7 +4005,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 9df3672d..48b02936 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -93,7 +93,7 @@ nzb-nntp = { version = "0.2.24", path = "crates/nzb-nntp" } nzb-core = { version = "0.2.19", path = "crates/nzb-core", features = ["groups-db"] } nzb-decode = { version = "0.1.5", path = "crates/nzb-decode" } nzb-news = { version = "0.1.14", path = "crates/nzb-news" } -nzb-dispatch = { version = "0.2.8", path = "crates/nzb-dispatch" } +nzb-dispatch = { version = "0.2.9", path = "crates/nzb-dispatch" } nzb-postproc = { version = "0.2.8", path = "crates/nzb-postproc" } mock-nntp-server = { path = "crates/mock-nntp-server" } rust-par2 = { version = "0.1.3" } diff --git a/crates/nzb-core/src/db.rs b/crates/nzb-core/src/db.rs index b6aa8883..e6062fce 100644 --- a/crates/nzb-core/src/db.rs +++ b/crates/nzb-core/src/db.rs @@ -777,6 +777,18 @@ impl Database { /// Move a completed/failed job to history. pub fn history_insert(&self, entry: &HistoryEntry) -> Result<(), NzbError> { + self.history_insert_with_carried_bytes(entry, 0) + } + + /// Move a completed/failed job to history, excluding `carried_bytes` + /// from its statistics row. A history retry starts from the failed + /// attempt's checkpoint, whose bytes that attempt's own statistics row + /// already counts; the history row keeps the job's cumulative total. + pub fn history_insert_with_carried_bytes( + &self, + entry: &HistoryEntry, + carried_bytes: u64, + ) -> Result<(), NzbError> { // Invariant: a Failed row carries exactly one typed failure code, and a // non-Failed row carries none. This keeps the terminal history row a // trustworthy source of the failure reason. @@ -819,8 +831,9 @@ impl Database { .max(0) as f64 / 1000.0 }); + let downloaded_bytes = entry.downloaded_bytes.saturating_sub(carried_bytes); let average_speed_bps = if duration_secs > 0.0 { - (entry.downloaded_bytes as f64 / duration_secs) as u64 + (downloaded_bytes as f64 / duration_secs) as u64 } else { 0 }; @@ -834,7 +847,7 @@ impl Database { entry.completed_at.to_rfc3339(), entry.status.to_string(), entry.total_bytes as i64, - entry.downloaded_bytes as i64, + downloaded_bytes as i64, duration_secs, average_speed_bps as i64, server_stats_json, @@ -1984,6 +1997,23 @@ mod tests { assert_eq!(loaded[0].server_stats[0].articles_downloaded, 100); } + #[test] + fn test_download_statistics_exclude_carried_bytes() { + let db = Database::open_memory().unwrap(); + let mut entry = make_history("retry-1", "Retried Job"); + entry.downloaded_bytes = 10_000; + entry.download_time_secs = Some(2.0); + + db.history_insert_with_carried_bytes(&entry, 6_000).unwrap(); + + // History keeps the job's cumulative progress; the statistics row + // counts only what this attempt fetched. + assert_eq!(db.history_list(1).unwrap()[0].downloaded_bytes, 10_000); + let statistics = db.download_statistics_list().unwrap(); + assert_eq!(statistics[0].downloaded_bytes, 4_000); + assert_eq!(statistics[0].average_speed_bps, 2_000); + } + #[test] fn test_download_statistics_survive_history_deletion() { let db = Database::open_memory().unwrap(); diff --git a/crates/nzb-dispatch/Cargo.toml b/crates/nzb-dispatch/Cargo.toml index 1938de33..220f7146 100644 --- a/crates/nzb-dispatch/Cargo.toml +++ b/crates/nzb-dispatch/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "nzb-dispatch" -version = "0.2.8" +version = "0.2.9" edition = "2024" description = "Article-level dispatcher: per-server worker pool, priority gating, retry + hopeless tracking. Part of the nzb-* layered usenet engine." license = "MIT" diff --git a/crates/nzb-dispatch/src/dispatch_engine.rs b/crates/nzb-dispatch/src/dispatch_engine.rs index 266c254f..7b38952a 100644 --- a/crates/nzb-dispatch/src/dispatch_engine.rs +++ b/crates/nzb-dispatch/src/dispatch_engine.rs @@ -58,6 +58,13 @@ pub trait DispatchEngine: Send + Sync { /// Is `job_id` currently known to the dispatcher? fn has_job(&self, job_id: &str) -> bool; + /// Active download time `job_id` has accumulated in this dispatcher, + /// excluding pauses, or `None` when the job is not registered. Lets the + /// queue checkpoint elapsed time so a restart can carry it forward. + fn job_active_secs(&self, _job_id: &str) -> Option { + None + } + /// Release a terminal job's dispatcher and assembler resources before /// post-processing opens the completed files. fn release_completed_job(&self, job_id: &str); @@ -184,6 +191,10 @@ impl DispatchEngine for DispatchHandle { self.0.has_job(job_id) } + fn job_active_secs(&self, job_id: &str) -> Option { + self.0.job_active_secs(job_id) + } + fn release_completed_job(&self, job_id: &str) { self.0.release_completed_job(job_id); } diff --git a/crates/nzb-dispatch/src/download_engine.rs b/crates/nzb-dispatch/src/download_engine.rs index fb78a0e2..104522f3 100644 --- a/crates/nzb-dispatch/src/download_engine.rs +++ b/crates/nzb-dispatch/src/download_engine.rs @@ -1552,6 +1552,14 @@ impl WorkerPool { pub fn has_job(&self, job_id: &str) -> bool { self.job_contexts.lock().contains_key(job_id) } + + /// Active (unpaused) download time of a registered job, in seconds. + pub fn job_active_secs(&self, job_id: &str) -> Option { + self.job_contexts + .lock() + .get(job_id) + .map(|ctx| ctx.active_elapsed().as_secs_f64()) + } } async fn await_worker_shutdown(mut handles: Vec>, timeout: Duration) { @@ -4266,6 +4274,24 @@ mod tests { assert_eq!(assembler.get_file_progress(job_id, "file-1"), (0, 0)); } + #[test] + fn job_active_secs_reports_registered_jobs_only() { + let tempdir = tempfile::tempdir().expect("tempdir"); + let job = test_job("active-secs", tempdir.path()); + let pool = worker_pool_without_servers(); + assert_eq!(pool.job_active_secs(&job.id), None); + + insert_test_context(&pool, &job, Arc::new(FileAssembler::new())); + pool.pause_job(&job.id); + let paused = pool.job_active_secs(&job.id).expect("registered job"); + assert!(paused >= 0.0); + // A paused job's clock is stopped. + assert_eq!(pool.job_active_secs(&job.id), Some(paused)); + + pool.release_completed_job(&job.id); + assert_eq!(pool.job_active_secs(&job.id), None); + } + #[cfg(target_os = "linux")] #[test] fn repeated_completed_jobs_do_not_accumulate_file_descriptors() { diff --git a/crates/nzb-web/src/queue_manager.rs b/crates/nzb-web/src/queue_manager.rs index fb90d3d3..934fef0f 100644 --- a/crates/nzb-web/src/queue_manager.rs +++ b/crates/nzb-web/src/queue_manager.rs @@ -485,6 +485,17 @@ struct JobCheckpoint { /// attempt keeps writing to its own folder instead of reserving another. #[serde(default)] output_dir: Option, + /// Active download time accumulated across every engine run of this job, + /// so a resumed job's speed covers all of its bytes. + #[serde(default)] + download_time_secs: f64, + /// Bytes a history retry carried in from the failed attempt's + /// checkpoint. That attempt's statistics row already counts them. + #[serde(default)] + carried_bytes: u64, + /// Per-server article statistics accumulated so far. + #[serde(default)] + server_stats: Vec, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -545,9 +556,20 @@ fn checkpoint_for_job(job: &NzbJob, output_dir_claimed: bool) -> JobCheckpoint { .collect(), work_dir: Some(job.work_dir.clone()), output_dir: output_dir_claimed.then(|| job.output_dir.clone()), + download_time_secs: 0.0, + carried_bytes: 0, + server_stats: job.server_stats.clone(), } } +/// Restore the download accounting a checkpoint carries across a restart: +/// per-server statistics into the job, and the accumulated active time and +/// retry carry-over returned as `(prior_download_secs, carried_bytes)`. +fn restore_checkpoint_accounting(job: &mut NzbJob, checkpoint: &JobCheckpoint) -> (f64, u64) { + job.server_stats = checkpoint.server_stats.clone(); + (checkpoint.download_time_secs, checkpoint.carried_bytes) +} + fn apply_checkpoint(job: &mut NzbJob, checkpoint: &JobCheckpoint) { job.downloaded_bytes = checkpoint.downloaded_bytes; job.articles_downloaded = checkpoint.articles_downloaded; @@ -1070,8 +1092,15 @@ struct JobState { direct_unpacker: Option, /// Hopeless job tracker (None until download starts). hopeless_tracker: Option, - /// Active worker-pool duration captured at terminal download resolution. + /// Active download duration captured at terminal download resolution, + /// including `prior_download_secs`. download_time_secs: Option, + /// Active download time from engine runs before the current one (a + /// restart resumes a job in a fresh engine run). + prior_download_secs: f64, + /// Bytes this job carried in from a failed attempt's checkpoint when it + /// was retried; excluded from this job's statistics row. + carried_bytes: u64, /// Typed terminal failure code, tracked in memory until it is persisted to /// the terminal history row. failure_code: Option, @@ -1766,6 +1795,16 @@ impl QueueManager { if let Some(ref data) = nzb_data { let _ = db.queue_store_nzb_data(&job.id, data); } + // A history retry starts from the failed attempt's checkpoint. + // Persist it now so a restart before the first progress write + // keeps both the resumed articles and the carried-over bytes. + if job.downloaded_bytes > 0 { + let mut checkpoint = checkpoint_for_job(&job, false); + checkpoint.carried_bytes = job.downloaded_bytes; + if let Ok(data) = serde_json::to_vec(&checkpoint) { + let _ = db.queue_store_job_data(&job.id, &data); + } + } } self.activate_admitted_job(job, nzb_data); @@ -1847,6 +1886,9 @@ impl QueueManager { // A job admitted as Paused (e.g. SABnzbd priority -2) stays paused // individually: it must not start, and Resume All must not resume it. let requested_paused = job.status == JobStatus::Paused; + // Only a history retry is admitted with progress: the bytes it + // carries in from the failed attempt's checkpoint. + let carried_bytes = job.downloaded_bytes; // If globally paused, add as paused if requested_paused || self.globally_paused.load(Ordering::Relaxed) { @@ -1859,6 +1901,8 @@ impl QueueManager { direct_unpacker: None, hopeless_tracker: None, download_time_secs: None, + prior_download_secs: 0.0, + carried_bytes, failure_code: None, output_dir_claimed: false, }; @@ -1882,6 +1926,8 @@ impl QueueManager { direct_unpacker: None, hopeless_tracker: None, download_time_secs: None, + prior_download_secs: 0.0, + carried_bytes, failure_code: None, output_dir_claimed: false, }; @@ -2010,6 +2056,8 @@ impl QueueManager { serde_json::from_slice::(&cp_data) { apply_checkpoint(&mut state.job, &checkpoint); + (state.prior_download_secs, state.carried_bytes) = + restore_checkpoint_accounting(&mut state.job, &checkpoint); if !state.output_dir_claimed { state.output_dir_claimed = restore_claimed_output_dir(&mut state.job, &checkpoint); @@ -2575,7 +2623,8 @@ impl QueueManager { { let mut jobs = self.jobs.lock(); if let Some(state) = jobs.get_mut(&job_id) { - state.download_time_secs = Some(download_time_secs); + state.download_time_secs = + Some(state.prior_download_secs + download_time_secs); state.job.status = JobStatus::PostProcessing; state.job.completed_at = Some(chrono::Utc::now()); } @@ -2608,7 +2657,8 @@ impl QueueManager { { let mut jobs = self.jobs.lock(); if let Some(state) = jobs.get_mut(&job_id) { - state.download_time_secs = Some(download_time_secs); + state.download_time_secs = + Some(state.prior_download_secs + download_time_secs); state.job.status = JobStatus::Failed; state.job.error_message = Some(reason.clone()); state.failure_code = Some(JobFailureCode::ArticlesUnavailable); @@ -3179,7 +3229,9 @@ impl QueueManager { true } Ok(None) => { - if let Err(e) = db.history_insert(&history_entry) { + if let Err(e) = + db.history_insert_with_carried_bytes(&history_entry, state.carried_bytes) + { error!(job_id = %state.job.id, "Failed to insert history: {e}"); false } else { @@ -3263,6 +3315,11 @@ impl QueueManager { fn persist_job_progress(&self, job_id: &str) { let jobs = self.jobs.lock(); if let Some(state) = jobs.get(job_id) { + // Accumulated active time: the final total once the download + // resolved, otherwise earlier runs plus the live engine run. + let download_time_secs = state.download_time_secs.unwrap_or_else(|| { + state.prior_download_secs + self.dispatch.job_active_secs(job_id).unwrap_or(0.0) + }); let db = self.db.lock(); if let Err(e) = db.queue_update_progress( job_id, @@ -3276,7 +3333,9 @@ impl QueueManager { } // Build and store checkpoint of downloaded article segments - let checkpoint = checkpoint_for_job(&state.job, state.output_dir_claimed); + let mut checkpoint = checkpoint_for_job(&state.job, state.output_dir_claimed); + checkpoint.download_time_secs = download_time_secs; + checkpoint.carried_bytes = state.carried_bytes; if let Ok(data) = serde_json::to_vec(&checkpoint) && let Err(e) = db.queue_store_job_data(job_id, &data) @@ -3651,7 +3710,9 @@ impl QueueManager { nzb_data: state.nzb_data.clone(), retry_data: None, }; - if let Err(e) = db.history_insert(&history_entry) { + if let Err(e) = + db.history_insert_with_carried_bytes(&history_entry, state.carried_bytes) + { error!(job_id = %id, "Failed to insert history for removed failed job: {e}"); } else { self.history_changed(); @@ -4819,6 +4880,9 @@ impl QueueManager { // the queue. This keeps memory low with large queues (hundreds of jobs). let was_active = job.status == JobStatus::Downloading; let mut output_dir_claimed = false; + let mut prior_download_secs = 0.0; + let mut carried_bytes = 0; + let mut download_time_secs = None; let nzb_data = if was_active { let db = self.db.lock(); @@ -4851,6 +4915,8 @@ impl QueueManager { Ok(checkpoint) => { apply_checkpoint(&mut job, &checkpoint); output_dir_claimed = restore_claimed_output_dir(&mut job, &checkpoint); + (prior_download_secs, carried_bytes) = + restore_checkpoint_accounting(&mut job, &checkpoint); let remaining = job .article_count @@ -4884,6 +4950,11 @@ impl QueueManager { .and_then(|data| serde_json::from_slice::(data).ok()) { output_dir_claimed = restore_claimed_output_dir(&mut job, &checkpoint); + (prior_download_secs, carried_bytes) = + restore_checkpoint_accounting(&mut job, &checkpoint); + // The download already resolved; its checkpoint holds the + // final active time. Older checkpoints have none. + download_time_secs = (prior_download_secs > 0.0).then_some(prior_download_secs); } } @@ -4905,7 +4976,9 @@ impl QueueManager { nzb_data, direct_unpacker: None, hopeless_tracker: None, - download_time_secs: None, + download_time_secs, + prior_download_secs, + carried_bytes, failure_code: None, output_dir_claimed, }; @@ -5299,6 +5372,8 @@ mod global_pause_tests { direct_unpacker: None, hopeless_tracker: None, download_time_secs: None, + prior_download_secs: 0.0, + carried_bytes: 0, failure_code: None, output_dir_claimed: false, }, @@ -6497,6 +6572,50 @@ mod global_pause_tests { assert!(!foreign.with_file_name("SameName.2").exists()); } + #[tokio::test] + async fn checkpoint_carries_download_accounting_across_a_restart() { + let (manager, tempdir) = manager(); + let mut downloading = job("accounting", JobStatus::Downloading, tempdir.path()); + downloading.downloaded_bytes = 4_000; + downloading.server_stats = vec![ServerArticleStats { + server_id: "srv".into(), + server_name: "Provider".into(), + articles_downloaded: 4, + articles_failed: 1, + bytes_downloaded: 4_000, + }]; + manager.db.lock().queue_insert(&downloading).unwrap(); + insert_job(&manager, downloading); + { + let mut jobs = manager.jobs.lock(); + let state = jobs.get_mut("accounting").unwrap(); + state.prior_download_secs = 40.0; + state.carried_bytes = 1_000; + } + + manager.persist_job_progress("accounting"); + + let data = manager + .db + .lock() + .queue_load_job_data("accounting") + .unwrap() + .unwrap(); + let checkpoint: JobCheckpoint = serde_json::from_slice(&data).unwrap(); + // The job is not registered with the dispatcher, so no live run adds + // to the time accumulated by earlier runs. + assert_eq!(checkpoint.download_time_secs, 40.0); + assert_eq!(checkpoint.carried_bytes, 1_000); + let mut restored = job("accounting", JobStatus::Downloading, tempdir.path()); + assert_eq!( + restore_checkpoint_accounting(&mut restored, &checkpoint), + (40.0, 1_000) + ); + assert_eq!(restored.server_stats.len(), 1); + assert_eq!(restored.server_stats[0].articles_downloaded, 4); + assert_eq!(restored.server_stats[0].articles_failed, 1); + } + #[test] fn stored_output_claim_outside_the_job_target_is_ignored() { let tempdir = tempfile::tempdir().unwrap(); diff --git a/crates/nzb-web/tests/harness_statistics.rs b/crates/nzb-web/tests/harness_statistics.rs new file mode 100644 index 00000000..11174bb9 --- /dev/null +++ b/crates/nzb-web/tests/harness_statistics.rs @@ -0,0 +1,195 @@ +//! Statistics ledger accounting across retries and restarts. +//! +//! A job's statistics row must count only the bytes it actually fetched and +//! the active time it spent fetching them, so neither a history retry nor a +//! restart mid-download double-counts bytes or inflates the recorded speed. + +mod harness; + +use std::path::PathBuf; +use std::time::Duration; + +use harness::nzb_fixture::NzbFixture; +use harness::{HarnessBuilder, ServerProfile, yenc_articles}; +use nzb_nntp::testutil::MockConfig; +use nzb_web::nzb_core::db::Database; +use nzb_web::nzb_core::models::JobStatus; +use nzb_web::nzb_core::nzb_parser; + +#[tokio::test] +async fn history_retry_does_not_count_carried_over_bytes_again() { + let fixture = NzbFixture::new("stats-retry") + .add_file( + "present.bin", + &[("stats-retry-present", b"partial payload")], + ) + .add_file("missing.bin", &[("stats-retry-missing", b"never served")]) + .build(); + let triples: Vec<(&str, &[u8], &str)> = fixture + .articles + .iter() + .filter(|(m, _, _)| *m == "stats-retry-present") + .map(|(m, b, f)| (*m, *b, f.as_str())) + .collect(); + let server = ServerProfile::start( + "stats-retry", + MockConfig { + articles: yenc_articles(&triples), + ..Default::default() + }, + 2, + ) + .await; + let engine = HarnessBuilder::new() + .with_server(server) + .article_timeout(5) + .build(); + let qm = &engine.queue_manager; + + let id = engine + .submit_nzb_xml("stats-retry", fixture.xml.clone()) + .unwrap(); + assert!( + engine + .wait_for(Duration::from_secs(20), |_| engine.history_status(&id) + == Some(JobStatus::Failed)) + .await, + "first attempt did not fail" + ); + let first = qm.global_statistics(&[]).lifetime; + assert_eq!(first.downloads, 1); + assert!(first.bytes_downloaded > 0); + + // The retry reuses the partial download: the present article is already + // on disk, so the retry fetches nothing new before failing again. + let entry = qm.history_get(&id).unwrap().unwrap(); + let retry_data = qm.history_get_retry_data(&id).unwrap(); + let retry = qm + .prepare_retry_job(&entry, &fixture.xml, retry_data.as_deref()) + .unwrap(); + assert!(retry.downloaded_bytes > 0, "retry carries the checkpoint"); + let retry_id = retry.id.clone(); + qm.add_job(retry, Some(fixture.xml)).unwrap(); + assert!( + engine + .wait_for(Duration::from_secs(20), |_| engine + .history_status(&retry_id) + == Some(JobStatus::Failed)) + .await, + "retry did not fail" + ); + + let after = qm.global_statistics(&[]).lifetime; + assert_eq!(after.downloads, 2); + assert_eq!( + after.bytes_downloaded, first.bytes_downloaded, + "bytes carried over from the failed attempt are already in its statistics row" + ); + // The history row still reports the job's cumulative progress. + let retried = qm.history_get(&retry_id).unwrap().unwrap(); + assert_eq!(retried.downloaded_bytes, entry.downloaded_bytes); +} + +#[tokio::test] +async fn restart_keeps_server_stats_and_active_time_from_before_the_restart() { + let fixture = NzbFixture::new("stats-restart") + .add_file( + "restart.bin", + &[ + ("stats-restart-1", b"first"), + ("stats-restart-2", b"second"), + ], + ) + .build(); + let triples = fixture + .articles + .iter() + .map(|(id, bytes, name)| (*id, *bytes, name.as_str())) + .collect::>(); + let server = ServerProfile::start( + "stats-restart", + MockConfig { + articles: yenc_articles(&triples), + ..MockConfig::default() + }, + 1, + ) + .await; + let state = tempfile::tempdir().expect("restart state"); + let database_path = state.path().join("queue.sqlite"); + let incomplete_dir = state.path().join("incomplete"); + let complete_dir = state.path().join("complete"); + std::fs::create_dir_all(&incomplete_dir).unwrap(); + std::fs::create_dir_all(&complete_dir).unwrap(); + let mut job = nzb_parser::parse_nzb("stats-restart", &fixture.xml).unwrap(); + job.status = JobStatus::Downloading; + job.work_dir = incomplete_dir.join(&job.id); + job.output_dir = complete_dir.join(&job.name); + job.downloaded_bytes = 5; + let job_id = job.id.clone(); + let db = Database::open(&database_path).unwrap(); + db.queue_insert(&job).unwrap(); + db.queue_store_nzb_data(&job_id, &fixture.xml).unwrap(); + // The checkpoint the previous process left behind: one article fetched + // from this server over 100 seconds of active download time. + db.queue_store_job_data( + &job_id, + &serde_json::to_vec(&serde_json::json!({ + "files": {"restart.bin": [1]}, + "downloaded_bytes": 5, + "articles_downloaded": 1, + "articles_failed": 0, + "files_completed": 0, + "download_time_secs": 100.0, + "server_stats": [{ + "server_id": "stats-restart", + "server_name": "mock-stats-restart", + "articles_downloaded": 1, + "articles_failed": 0, + "bytes_downloaded": 5 + }] + })) + .unwrap(), + ) + .unwrap(); + drop(db); + + let engine = HarnessBuilder::restart_recovery(server) + .with_database_path(database_path) + .with_state_dir(PathBuf::from(state.path())) + .build(); + engine.queue_manager.restore_from_db().unwrap(); + assert!( + engine + .wait_for(Duration::from_secs(10), |_| engine.history_status(&job_id) + == Some(JobStatus::Completed)) + .await + ); + + let history = engine.queue_manager.history_get(&job_id).unwrap().unwrap(); + assert_eq!(history.downloaded_bytes, history.total_bytes); + let server_stats = history + .server_stats + .iter() + .find(|s| s.server_id == "stats-restart") + .expect("server stats"); + assert_eq!( + server_stats.articles_downloaded, 2, + "the article fetched before the restart is still attributed to its server" + ); + assert_eq!(server_stats.bytes_downloaded, history.downloaded_bytes); + let download_time = history.download_time_secs.expect("download time"); + assert!( + download_time >= 100.0, + "active time from before the restart is kept, got {download_time}" + ); + + let lifetime = engine.queue_manager.global_statistics(&[]).lifetime; + assert_eq!(lifetime.bytes_downloaded, history.downloaded_bytes); + assert!(lifetime.total_duration_secs >= 100.0); + assert!( + lifetime.fastest_download_bps <= history.downloaded_bytes / 100, + "speed must not be inflated by bytes fetched before the restart, got {} B/s", + lifetime.fastest_download_bps + ); +} From 265f808ce5d6ef54338bd7dc3066a25cde90cb5e Mon Sep 17 00:00:00 2001 From: thedancingdeveloper <306930456+thedancingdeveloper@users.noreply.github.com> Date: Sat, 10 Oct 2026 00:41:37 +0000 Subject: [PATCH 04/23] fix: count each job's server stats once; live progress as today Port of xbmc4lyfe/rustnzb#176. Server stats could double-count a job when it appeared both in the active queue and in the download ledger; recorded jobs are now skipped in the jobs pass and counted from the ledger. Live progress of a long-queued job is attributed to today instead of being dropped. Local adaptations: tests use the existing insert_job helper (which already inserts the queue row, so redundant queue_insert calls were removed) and set the extra HistoryEntry fields (delete_archives, post_processing) our schema carries. --- crates/nzb-web/src/queue_manager.rs | 90 ++++++++++++++++++++++++++--- 1 file changed, 82 insertions(+), 8 deletions(-) diff --git a/crates/nzb-web/src/queue_manager.rs b/crates/nzb-web/src/queue_manager.rs index 934fef0f..01a9d424 100644 --- a/crates/nzb-web/src/queue_manager.rs +++ b/crates/nzb-web/src/queue_manager.rs @@ -4384,18 +4384,25 @@ impl QueueManager { } }; + let ledger = { + let db = self.db.lock(); + db.download_statistics_list().unwrap_or_default() + }; { + // A terminal job stays in the map for a few seconds after + // move_to_history wrote its statistics row; count it once, from + // the ledger. Live progress is happening now, whenever the job + // was queued. + let recorded: HashSet<&str> = + ledger.iter().map(|entry| entry.job_id.as_str()).collect(); let jobs = self.jobs.lock(); for state in jobs.values() { - apply(state.job.added_at, &state.job.server_stats, true); + if !recorded.contains(state.job.id.as_str()) { + apply(now, &state.job.server_stats, true); + } } } - - let ledger = { - let db = self.db.lock(); - db.download_statistics_list().unwrap_or_default() - }; - for entry in ledger { + for entry in &ledger { apply(entry.completed_at, &entry.server_stats, false); } @@ -6584,7 +6591,6 @@ mod global_pause_tests { articles_failed: 1, bytes_downloaded: 4_000, }]; - manager.db.lock().queue_insert(&downloading).unwrap(); insert_job(&manager, downloading); { let mut jobs = manager.jobs.lock(); @@ -6616,6 +6622,74 @@ mod global_pause_tests { assert_eq!(restored.server_stats[0].articles_failed, 1); } + fn served(bytes: u64) -> Vec { + vec![ServerArticleStats { + server_id: "srv".into(), + server_name: "Provider".into(), + articles_downloaded: 2, + articles_failed: 1, + bytes_downloaded: bytes, + }] + } + + #[tokio::test] + async fn server_stats_count_a_job_kept_after_history_once() { + let (manager, tempdir) = manager(); + let mut finished = job("lingering", JobStatus::Completed, tempdir.path()); + finished.server_stats = served(1_000); + finished.completed_at = Some(Utc::now()); + // move_to_history wrote the history and statistics rows, but the job + // stays in the in-memory map for a few seconds afterwards. + manager + .db + .lock() + .history_insert(&HistoryEntry { + id: finished.id.clone(), + name: finished.name.clone(), + category: finished.category.clone(), + status: JobStatus::Completed, + total_bytes: 1_000, + downloaded_bytes: 1_000, + added_at: finished.added_at, + completed_at: Utc::now(), + download_time_secs: Some(1.0), + output_dir: finished.output_dir.clone(), + stages: Vec::new(), + error_message: None, + failure_code: None, + server_stats: finished.server_stats.clone(), + nzb_data: None, + retry_data: None, + delete_archives: None, + post_processing: None, + }) + .unwrap(); + insert_job(&manager, finished); + + let stats = manager.server_stats_get_all(&[]); + assert_eq!(stats.len(), 1); + assert_eq!(stats[0].total_bytes, 1_000); + assert_eq!(stats[0].total_ok, 2); + assert_eq!(stats[0].total_fail, 1); + assert_eq!(stats[0].today_bytes, 1_000); + } + + #[tokio::test] + async fn server_stats_count_live_progress_of_a_long_queued_job_today() { + let (manager, tempdir) = manager(); + let mut downloading = job("old-queued", JobStatus::Downloading, tempdir.path()); + // Queued two days ago, downloading now. + downloading.added_at = Utc::now() - chrono::Duration::days(2); + downloading.server_stats = served(500); + insert_job(&manager, downloading); + + let stats = manager.server_stats_get_all(&[]); + assert_eq!(stats[0].total_bytes, 500); + assert_eq!(stats[0].today_bytes, 500); + assert_eq!(stats[0].today_ok, 2); + assert_eq!(stats[0].week_bytes, 500); + } + #[test] fn stored_output_claim_outside_the_job_target_is_ignored() { let tempdir = tempfile::tempdir().unwrap(); From a8f632c5a910ec387e590b0e64de729974589511 Mon Sep 17 00:00:00 2001 From: thedancingdeveloper <306930456+thedancingdeveloper@users.noreply.github.com> Date: Sat, 10 Oct 2026 00:45:16 +0000 Subject: [PATCH 05/23] feat: paginate history server-side with filters and stats Move history filtering, paging and stats into the API: the history endpoint now accepts offset/status/category/search/days parameters, returns category totals and window stats (success rate, average duration, top fail reasons), and the web UI pages and filters server-side instead of loading the full history. Ported from xbmc4lyfe/rustnzb#192. Adaptation: the new integration test's HistoryEntry literals gained post_processing and delete_archives fields required by our models. --- .../history/history-view.component.spec.ts | 139 ++++++++-- .../history/history-view.component.ts | 259 +++++++++++++----- apps/rustnzb/src/handlers.rs | 143 +++++++++- apps/rustnzb/tests/history_pagination_api.rs | 183 +++++++++++++ e2e/tests/history.spec.ts | 2 +- 5 files changed, 622 insertions(+), 104 deletions(-) create mode 100644 apps/rustnzb/tests/history_pagination_api.rs diff --git a/apps/rustnzb/frontend/src/app/features/history/history-view.component.spec.ts b/apps/rustnzb/frontend/src/app/features/history/history-view.component.spec.ts index 6ac2e4ad..7fee5065 100644 --- a/apps/rustnzb/frontend/src/app/features/history/history-view.component.spec.ts +++ b/apps/rustnzb/frontend/src/app/features/history/history-view.component.spec.ts @@ -1,7 +1,7 @@ import '@angular/compiler'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; -import { of } from 'rxjs'; +import { firstValueFrom, of } from 'rxjs'; import { ApiService } from '../../core/services/api.service'; import { HistoryEntry } from '../../core/models/queue.model'; @@ -64,41 +64,130 @@ describe('HistoryViewComponent', () => { component.ngOnDestroy(); }); - it('combines name, status, category, and time filters', () => { - component.entries.set([ - entry(), - entry({ id: '2', name: 'Release.Failed', category: 'tv', status: 'failed' }), - entry({ id: '3', name: 'Ancient.Release', completed_at: '2025-01-01T00:00:00Z' }), - ]); - component.nameFilter = 'failed'; + function historyCalls() { + return api.get.mock.calls.filter(([path]) => path === '/history'); + } + + it('requests one page with the default 7-day window', () => { + component.load(); + expect(api.get).toHaveBeenCalledWith('/history', { offset: '0', limit: '50', days: '7' }); + }); + + it('sends name, status, category, and time filters to the server and resets to page one', () => { + component.offset.set(100); + component.nameFilter = ' failed '; component.filterStatus = 'failed'; component.filterCategory = 'tv'; - expect(component.filteredEntries().map((item) => item.id)).toEqual(['2']); + component.filterTime = 'all'; + component.onFiltersChanged(); + expect(component.offset()).toBe(0); + expect(historyCalls().at(-1)?.[1]).toEqual({ + offset: '0', + limit: '50', + status: 'failed', + category: 'tv', + search: 'failed', + }); + }); + + it('pages through the full history using the server total', () => { + api.get.mockImplementation((path: string) => + of(path === '/history' ? { entries: [entry()], total: 120, offset: 0, limit: 50 } : {}), + ); + component.load(); + expect(component.total()).toBe(120); + expect(component.rangeLabel()).toBe('1–50 of 120'); + expect(component.hasPrev()).toBe(false); + expect(component.hasNext()).toBe(true); + + component.nextPage(); + expect(component.offset()).toBe(50); + expect(historyCalls().at(-1)?.[1]).toMatchObject({ offset: '50', limit: '50' }); + component.nextPage(); + expect(component.offset()).toBe(100); + expect(component.rangeLabel()).toBe('101–120 of 120'); + expect(component.hasNext()).toBe(false); + component.nextPage(); + expect(component.offset()).toBe(100); + + component.prevPage(); + expect(component.offset()).toBe(50); + }); + + it('steps back to the last page when the current page disappears', () => { + api.get.mockImplementation((path: string, params?: Record) => + of(path === '/history' + ? { entries: params?.['offset'] === '50' ? [entry()] : [], total: 51 } + : {}), + ); + component.offset.set(100); + component.load(); + expect(component.offset()).toBe(50); + expect(component.entries()).toHaveLength(1); }); - it('derives sorted unique category options', () => { - component.entries.set([ - entry({ category: 'tv' }), - entry({ id: '2', category: 'movies' }), - entry({ id: '3', category: 'tv' }), - entry({ id: '4', category: '' }), - ]); + it('takes category options from the server rather than the visible page', () => { + api.get.mockImplementation((path: string) => + of(path === '/history' + ? { entries: [entry({ category: 'tv' })], total: 1, categories: ['movies', 'tv'] } + : {}), + ); + component.load(); expect(component.categoryOptions()).toEqual(['movies', 'tv']); }); - it('computes success statistics and failure summaries', () => { - component.entries.set([ - entry(), - entry({ id: '2', total_bytes: 3072 }), - entry({ id: '3', status: 'failed', error_message: 'CRC mismatch: bad block' }), - ]); + it('shows server-computed statistics for the whole window', () => { + api.get.mockImplementation((path: string) => + of(path === '/history' + ? { + entries: [entry()], + total: 300, + stats: { + completed: 200, + completed_bytes: 4096, + failed: 100, + success_pct: 67, + avg_duration_secs: 90, + fail_reasons: [{ reason: 'CRC mismatch', count: 60 }, { reason: 'Aborted', count: 40 }], + }, + } + : {}), + ); + component.load(); expect(component.statCards()).toMatchObject({ - completed: 2, + completed: 200, completedBytes: 4096, - failed: 1, - failReasons: '1 CRC mismatch', + failed: 100, + failReasons: '60 CRC mismatch · 40 Aborted', successPct: 67, + avgDurationLabel: '1m 30s', + }); + }); + + it('tolerates a server response without totals or stats', () => { + api.get.mockImplementation((path: string) => + of(path === '/history' ? { entries: [entry(), entry({ id: '2', category: 'tv' })] } : {}), + ); + component.load(); + expect(component.total()).toBe(2); + expect(component.categoryOptions()).toEqual(['movies', 'tv']); + expect(component.statCards()).toMatchObject({ completed: 0, failed: 0, failReasons: 'none' }); + }); + + it('exports every matching entry, not just the visible page', async () => { + const all = Array.from({ length: 450 }, (_, i) => entry({ id: `e${i}`, name: `Release.${i}` })); + api.get.mockImplementation((path: string, params?: Record) => { + const offset = Number(params?.['offset'] ?? 0); + const limit = Number(params?.['limit'] ?? 50); + return of(path === '/history' + ? { entries: all.slice(offset, offset + limit), total: all.length } + : {}); }); + component.filterStatus = 'completed'; + const exported = await firstValueFrom(component.fetchAllMatching()); + expect(exported).toHaveLength(450); + expect(historyCalls().every(([, params]) => params.status === 'completed')).toBe(true); + expect(historyCalls().length).toBeLessThanOrEqual(3); }); it('retries, removes, and queues media through the expected routes', () => { diff --git a/apps/rustnzb/frontend/src/app/features/history/history-view.component.ts b/apps/rustnzb/frontend/src/app/features/history/history-view.component.ts index ea8a9a53..967ae198 100644 --- a/apps/rustnzb/frontend/src/app/features/history/history-view.component.ts +++ b/apps/rustnzb/frontend/src/app/features/history/history-view.component.ts @@ -1,4 +1,5 @@ import { Component, OnInit, OnDestroy, signal, computed } from '@angular/core'; +import { EMPTY, Observable, expand, map, reduce } from 'rxjs'; import { CommonModule } from '@angular/common'; import { FormsModule } from '@angular/forms'; import { MatSnackBar, MatSnackBarModule } from '@angular/material/snack-bar'; @@ -11,6 +12,27 @@ import { showHttpError } from '../../core/http/http-error'; type StatusFilter = 'all' | 'completed' | 'failed'; type TimeFilter = '7d' | '30d' | 'all'; +/** Aggregates the server computes over the whole time window. */ +interface HistoryStats { + completed: number; + completed_bytes: number; + failed: number; + success_pct: number; + avg_duration_secs: number | null; + fail_reasons: { reason: string; count: number }[]; +} + +interface HistoryPage { + entries: HistoryEntry[]; + /** Entries matching the filters across all pages. */ + total?: number; + categories?: string[]; + stats?: HistoryStats; +} + +const PAGE_SIZE = 50; +const EXPORT_PAGE_SIZE = 200; + @Component({ selector: 'app-history-view', standalone: true, @@ -44,27 +66,29 @@ type TimeFilter = '7d' | '30d' | 'all';

History - {{ filteredEntries().length }} of {{ entries().length }} shown + {{ rangeLabel() }}

- - + - @for (cat of categoryOptions(); track cat) { } - - - @if (entries().length > 0) { + + @if (total() > 0) { }
@@ -203,10 +227,12 @@ type TimeFilter = '7d' | '30d' | 'all'; } @else if (filteredEntries().length === 0) { - @if (entries().length === 0) { + @if (narrowingFilters()) { + No entries match the current filter. + } @else if (filterTime === 'all') { No download history yet. Finished jobs will show up here. } @else { - No entries match the current filter. + No downloads finished in the last {{ statCards().windowLabel }}. } @@ -214,6 +240,13 @@ type TimeFilter = '7d' | '30d' | 'all';
+ @if (hasPrev() || hasNext()) { +
+ + {{ rangeLabel() }} + +
+ }
`, styles: [` @@ -242,6 +275,10 @@ type TimeFilter = '7d' | '30d' | 'all'; justify-content: flex-end; gap: 2px; } + .pager { + display: flex; align-items: center; justify-content: flex-end; gap: 10px; + padding: 10px 14px; border-top: 1px solid var(--line); + } .history-row { cursor: pointer; } .history-row:focus { outline: 1px solid var(--accent); outline-offset: -1px; } .history-row.selected td { background: var(--row-hover); border-bottom-color: transparent; } @@ -271,7 +308,17 @@ type TimeFilter = '7d' | '30d' | 'all'; }) export class HistoryViewComponent implements OnInit, OnDestroy { loading = signal(true); + /** The current page of entries, already filtered by the server. */ entries = signal([]); + /** Entries matching the filters across all pages. */ + total = signal(0); + offset = signal(0); + readonly pageSize = PAGE_SIZE; + exporting = signal(false); + private categories = signal(null); + private stats = signal(null); + private loadSeq = 0; + private nameDebounce: ReturnType | null = null; selectedId = signal(null); selectedEntry = signal(null); detailLoading = signal(false); @@ -300,15 +347,89 @@ export class HistoryViewComponent implements OnInit, OnDestroy { ngOnDestroy(): void { if (this.pollTimer) clearInterval(this.pollTimer); + if (this.nameDebounce) clearTimeout(this.nameDebounce); + } + + /** Query parameters for the active filters (shared by paging and export). */ + private filterParams(): Record { + const params: Record = {}; + if (this.filterStatus !== 'all') params['status'] = this.filterStatus; + if (this.filterCategory) params['category'] = this.filterCategory; + const name = this.nameFilter.trim(); + if (name) params['search'] = name; + if (this.filterTime !== 'all') params['days'] = this.filterTime === '7d' ? '7' : '30'; + return params; } load(): void { - this.api.get<{ entries: HistoryEntry[] }>('/history').subscribe({ - next: r => { this.entries.set(r.entries || []); this.loading.set(false); }, + const seq = ++this.loadSeq; + const params = { offset: String(this.offset()), limit: String(PAGE_SIZE), ...this.filterParams() }; + this.api.get('/history', params).subscribe({ + next: r => { + // A slower response for superseded filters/page must not win. + if (seq !== this.loadSeq) return; + const entries = r.entries || []; + const total = r.total ?? entries.length; + // The page we were on vanished (deletes, retention): go to the last one. + if (entries.length === 0 && this.offset() > 0 && this.offset() >= total) { + this.offset.set(Math.max(0, Math.ceil(total / PAGE_SIZE) - 1) * PAGE_SIZE); + this.load(); + return; + } + this.entries.set(entries); + this.total.set(total); + this.categories.set(r.categories ?? null); + this.stats.set(r.stats ?? null); + this.loading.set(false); + }, error: () => this.loading.set(false), }); } + onFiltersChanged(): void { + this.offset.set(0); + this.closeDetails(); + this.load(); + } + + onNameFilterChanged(): void { + if (this.nameDebounce) clearTimeout(this.nameDebounce); + this.nameDebounce = setTimeout(() => this.onFiltersChanged(), 250); + } + + hasPrev(): boolean { + return this.offset() > 0; + } + + hasNext(): boolean { + return this.offset() + PAGE_SIZE < this.total(); + } + + prevPage(): void { + if (!this.hasPrev()) return; + this.offset.set(Math.max(0, this.offset() - PAGE_SIZE)); + this.load(); + } + + nextPage(): void { + if (!this.hasNext()) return; + this.offset.set(this.offset() + PAGE_SIZE); + this.load(); + } + + rangeLabel(): string { + const total = this.total(); + if (total === 0) return '0 of 0'; + const start = Math.min(this.offset() + 1, total); + const end = Math.min(this.offset() + PAGE_SIZE, total); + return `${start}–${end} of ${total}`; + } + + /** Filters other than the time window are narrowing the list. */ + narrowingFilters(): boolean { + return this.filterStatus !== 'all' || !!this.filterCategory || !!this.nameFilter.trim(); + } + selectEntry(entry: HistoryEntry): void { if (this.selectedId() === entry.id) { this.closeDetails(); @@ -332,78 +453,39 @@ export class HistoryViewComponent implements OnInit, OnDestroy { this.detailLoading.set(false); } - categoryOptions = computed(() => - Array.from(new Set(this.entries().map(e => e.category).filter(c => !!c))).sort() + /** All categories in history (server-provided), not just the visible page. */ + categoryOptions = computed( + () => + this.categories() ?? + Array.from(new Set(this.entries().map(e => e.category).filter(c => !!c))).sort(), ); - /** - * Returns entries filtered by *all* active filters. Not memoized as a - * signal because it depends on plain fields (ngModel) that don't trigger - * signal recomputation — the template re-renders on change detection - * anyway. - */ + /** The current page; filtering happens server-side. */ filteredEntries(): HistoryEntry[] { - const cutoff = this.timeCutoffMs(); - const name = this.nameFilter.trim().toLowerCase(); - return this.entries().filter(e => { - if (this.filterStatus !== 'all' && e.status !== this.filterStatus) return false; - if (this.filterCategory && e.category !== this.filterCategory) return false; - if (cutoff > 0 && new Date(e.completed_at).getTime() < cutoff) return false; - if (name && !e.name.toLowerCase().includes(name)) return false; - return true; - }); - } - - private timeCutoffMs(): number { - if (this.filterTime === 'all') return 0; - const now = Date.now(); - const days = this.filterTime === '7d' ? 7 : 30; - return now - days * 86400_000; + return this.entries(); } /** - * Computed aggregate for the 4 stat cards at the top. Uses the time - * window filter (but ignores the status filter) so the success-rate - * card remains meaningful when the user filters to just failures. + * The 4 stat cards at the top. The server computes them over the whole + * time window (ignoring the status/category/name filters) so the + * success-rate card stays meaningful when the user filters to failures. */ statCards = computed(() => { - const cutoff = this.timeCutoffMs(); - const inWindow = this.entries().filter(e => - cutoff === 0 || new Date(e.completed_at).getTime() >= cutoff - ); - const completed = inWindow.filter(e => e.status === 'completed'); - const failed = inWindow.filter(e => e.status === 'failed'); - const completedBytes = completed.reduce((n, e) => n + e.total_bytes, 0); - const total = inWindow.length; - const successPct = total === 0 ? 0 : Math.round((completed.length / total) * 100); - - let avgDurationLabel = '—'; - if (completed.length > 0) { - const total = completed.reduce((n, e) => { - return n + (new Date(e.completed_at).getTime() - new Date(e.added_at).getTime()); - }, 0); - avgDurationLabel = this.formatShortDuration(total / completed.length / 1000); - } - - const reasonCounts = new Map(); - for (const f of failed) { - const reason = (f.error_message || 'unknown').split(/[.:]/)[0].slice(0, 32); - reasonCounts.set(reason, (reasonCounts.get(reason) || 0) + 1); - } - const topReasons = [...reasonCounts.entries()] - .sort((a, b) => b[1] - a[1]) + const st = this.stats(); + const failed = st?.failed ?? 0; + const topReasons = (st?.fail_reasons ?? []) .slice(0, 2) - .map(([r, n]) => `${n} ${r}`) - .join(' · ') || 'none'; - + .map(r => `${r.count} ${r.reason}`) + .join(' · '); return { windowLabel: this.filterTime === 'all' ? 'all time' : this.filterTime === '7d' ? '7 days' : '30 days', - completed: completed.length, - completedBytes, - failed: failed.length, - failReasons: failed.length === 0 ? 'none' : topReasons, - successPct, - avgDurationLabel, + completed: st?.completed ?? 0, + completedBytes: st?.completed_bytes ?? 0, + failed, + failReasons: failed === 0 || !topReasons ? 'none' : topReasons, + successPct: st?.success_pct ?? 0, + avgDurationLabel: + st?.avg_duration_secs != null ? this.formatShortDuration(st.avg_duration_secs) : '—', }; }); @@ -456,9 +538,40 @@ export class HistoryViewComponent implements OnInit, OnDestroy { this.snack.open(e.output_dir || '(no output path recorded)', 'Close', { duration: 5000 }); } + /** Every entry matching the current filters, fetched page by page. */ + fetchAllMatching(): Observable { + const filters = this.filterParams(); + const page = (offset: number) => + this.api + .get('/history', { offset: String(offset), limit: String(EXPORT_PAGE_SIZE), ...filters }) + .pipe(map(r => ({ offset, entries: r.entries || [], total: r.total ?? 0 }))); + return page(0).pipe( + expand(p => { + const next = p.offset + p.entries.length; + return p.entries.length > 0 && next < p.total ? page(next) : EMPTY; + }), + reduce((all, p) => all.concat(p.entries), [] as HistoryEntry[]), + ); + } + exportCsv(): void { + if (this.exporting()) return; + this.exporting.set(true); + this.fetchAllMatching().subscribe({ + next: entries => { + this.exporting.set(false); + this.downloadCsv(entries); + }, + error: () => { + this.exporting.set(false); + this.snack.open('Failed to export history', 'Close', { duration: 3000 }); + }, + }); + } + + private downloadCsv(entries: HistoryEntry[]): void { const rows = [['name', 'category', 'size_bytes', 'average_speed_bps', 'status', 'added_at', 'completed_at', 'error']]; - for (const e of this.filteredEntries()) { + for (const e of entries) { rows.push([ e.name, e.category || '', String(e.total_bytes), String(this.averageSpeed(e)), e.status, e.added_at, e.completed_at, e.error_message || '', diff --git a/apps/rustnzb/src/handlers.rs b/apps/rustnzb/src/handlers.rs index 8d483a14..fc311c66 100644 --- a/apps/rustnzb/src/handlers.rs +++ b/apps/rustnzb/src/handlers.rs @@ -70,7 +70,18 @@ pub struct QueueQuery { #[derive(Deserialize, Default)] pub struct HistoryQuery { + /// Page size (default 50). pub limit: Option, + /// Number of matching entries to skip (default 0). + pub offset: Option, + /// Only `completed` or `failed` entries. + pub status: Option, + /// Only entries in this category. + pub category: Option, + /// Case-insensitive substring of the job name. + pub search: Option, + /// Only entries completed within the last N days. Also bounds `stats`. + pub days: Option, } #[derive(Deserialize)] @@ -157,7 +168,80 @@ pub struct QueueJob { #[derive(Serialize)] pub struct HistoryResponse { pub entries: Vec, + /// Number of entries matching the filters (across all pages). pub total: usize, + pub offset: usize, + pub limit: usize, + /// Every category present in history, sorted. + pub categories: Vec, + /// Aggregates over the `days` window (ignores status/category/search). + pub stats: HistoryStats, +} + +#[derive(Serialize, Default)] +pub struct HistoryStats { + pub completed: usize, + pub completed_bytes: u64, + pub failed: usize, + /// Completed share of completed + failed, rounded to a whole percent. + pub success_pct: u32, + /// Mean wall-clock time from add to completion of completed jobs. + pub avg_duration_secs: Option, + /// The two most common failure reasons. + pub fail_reasons: Vec, +} + +#[derive(Serialize)] +pub struct HistoryFailReason { + pub reason: String, + pub count: usize, +} + +impl HistoryStats { + fn from_entries<'a>(entries: impl Iterator) -> Self { + let mut stats = Self::default(); + let mut duration_total = 0.0; + let mut reasons: Vec = Vec::new(); + for e in entries { + match e.status { + JobStatus::Completed => { + stats.completed += 1; + stats.completed_bytes += e.total_bytes; + duration_total += + (e.completed_at - e.added_at).num_milliseconds() as f64 / 1000.0; + } + JobStatus::Failed => { + stats.failed += 1; + // First clause of the message, as a short grouping key. + let msg = e.error_message.as_deref().unwrap_or("unknown"); + let reason: String = msg + .split(['.', ':']) + .next() + .unwrap_or_default() + .chars() + .take(32) + .collect(); + match reasons.iter_mut().find(|r| r.reason == reason) { + Some(r) => r.count += 1, + None => reasons.push(HistoryFailReason { reason, count: 1 }), + } + } + _ => {} + } + } + let finished = stats.completed + stats.failed; + if finished > 0 { + stats.success_pct = ((stats.completed as f64 / finished as f64) * 100.0).round() as u32; + } + if stats.completed > 0 { + stats.avg_duration_secs = Some(duration_total / stats.completed as f64); + } + // Stable sort keeps first-seen order among equal counts. + reasons.sort_by_key(|r| std::cmp::Reverse(r.count)); + reasons.truncate(2); + stats.fail_reasons = reasons; + stats + } } #[derive(Serialize)] @@ -790,13 +874,62 @@ pub async fn h_history_list( Query(q): Query, ) -> Result, ApiError> { let limit = q.limit.unwrap_or(50); - let entries = state + let offset = q.offset.unwrap_or(0); + // Filtering, totals and stats need the whole history (newest first), as + // the SABnzbd history endpoint already does. + let all = state .queue_manager - .history_list(limit) + .history_list(i64::MAX as usize) .map_err(ApiError::from)?; - let total = entries.len(); - let entries: Vec = entries.into_iter().map(Into::into).collect(); - Ok(Json(HistoryResponse { entries, total })) + + let mut categories: Vec = all + .iter() + .filter(|e| !e.category.is_empty()) + .map(|e| e.category.clone()) + .collect(); + categories.sort(); + categories.dedup(); + + let cutoff = q + .days + .map(|d| chrono::Utc::now() - chrono::Duration::days(i64::from(d))); + let in_window = |e: &&HistoryEntry| cutoff.is_none_or(|c| e.completed_at >= c); + let stats = HistoryStats::from_entries(all.iter().filter(in_window)); + + let status = q.status.as_deref().filter(|s| !s.is_empty() && *s != "all"); + let category = q.category.as_deref().filter(|c| !c.is_empty()); + let search = q + .search + .as_deref() + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(str::to_lowercase); + let matching: Vec = all + .into_iter() + .filter(|e| in_window(&e)) + .filter(|e| status.is_none_or(|s| e.status.to_string().eq_ignore_ascii_case(s))) + .filter(|e| category.is_none_or(|c| e.category == c)) + .filter(|e| { + search + .as_deref() + .is_none_or(|s| e.name.to_lowercase().contains(s)) + }) + .collect(); + let total = matching.len(); + let entries: Vec = matching + .into_iter() + .skip(offset) + .take(limit) + .map(Into::into) + .collect(); + Ok(Json(HistoryResponse { + entries, + total, + offset, + limit, + categories, + stats, + })) } /// GET /api/history/{id} -- Detailed information for one completed/failed job. diff --git a/apps/rustnzb/tests/history_pagination_api.rs b/apps/rustnzb/tests/history_pagination_api.rs new file mode 100644 index 00000000..dde897dc --- /dev/null +++ b/apps/rustnzb/tests/history_pagination_api.rs @@ -0,0 +1,183 @@ +//! GET /api/history pages through the full history: `total` counts every +//! matching entry (not just the page), `offset` pages, filters run +//! server-side, and the summary stats cover the whole time window. + +use std::sync::Arc; + +use arc_swap::ArcSwap; +use axum::extract::{Query, State}; +use chrono::{Duration, Utc}; +use nzb_web::auth::{CredentialStore, TokenStore}; +use nzb_web::nzb_core::config::AppConfig; +use nzb_web::nzb_core::db::Database; +use nzb_web::nzb_core::models::{HistoryEntry, JobFailureCode, JobStatus}; +use nzb_web::{AppState, LogBuffer, QueueManager}; +use rustnzb::handlers::{HistoryQuery, h_history_list}; +use serde_json::Value; + +fn history(i: usize, status: JobStatus, category: &str, age_days: i64) -> HistoryEntry { + let completed_at = Utc::now() - Duration::days(age_days) - Duration::minutes(i as i64); + HistoryEntry { + id: format!("h{i:03}"), + name: format!("Release.{i:03}"), + category: category.to_string(), + status, + total_bytes: 1000, + downloaded_bytes: 1000, + added_at: completed_at - Duration::seconds(60), + completed_at, + download_time_secs: None, + output_dir: "/tmp/out".into(), + stages: Vec::new(), + error_message: (status == JobStatus::Failed).then(|| "CRC mismatch: bad".to_string()), + failure_code: (status == JobStatus::Failed).then_some(JobFailureCode::ArchiveInvalid), + post_processing: None, + delete_archives: None, + server_stats: Vec::new(), + nzb_data: None, + retry_data: None, + } +} + +/// 120 entries: 100 completed `tv` (recent), 15 failed `movies` (recent), +/// 5 completed `old` entries completed 60 days ago. +fn state_with_history() -> (Arc, tempfile::TempDir) { + let tempdir = tempfile::tempdir().expect("tempdir"); + let db = Database::open_memory().expect("database"); + let mut i = 0; + for _ in 0..100 { + db.history_insert(&history(i, JobStatus::Completed, "tv", 0)) + .unwrap(); + i += 1; + } + for _ in 0..15 { + db.history_insert(&history(i, JobStatus::Failed, "movies", 0)) + .unwrap(); + i += 1; + } + for _ in 0..5 { + db.history_insert(&history(i, JobStatus::Completed, "old", 60)) + .unwrap(); + i += 1; + } + let log_buffer = LogBuffer::default(); + let manager = QueueManager::new( + Vec::new(), + db, + tempdir.path().join("incomplete"), + tempdir.path().join("complete"), + log_buffer.clone(), + 1, + Vec::new(), + 0, + 0, + false, + 5, + false, + false, + 100.0, + 30, + ); + let state = AppState::new( + Arc::new(ArcSwap::from_pointee(AppConfig::default())), + tempdir.path().join("config.toml"), + manager, + log_buffer, + Arc::new(TokenStore::new()), + Arc::new(CredentialStore::new(tempdir.path().to_path_buf())), + ); + (Arc::new(state), tempdir) +} + +async fn list(state: &Arc, query: &str) -> Value { + let uri: axum::http::Uri = format!("/api/history?{query}").parse().expect("uri"); + let q: Query = Query::try_from_uri(&uri).expect("query"); + let resp = h_history_list(State(state.clone()), q) + .await + .unwrap_or_else(|_| panic!("history list failed for {query:?}")); + serde_json::to_value(&resp.0).expect("json") +} + +fn ids(v: &Value) -> Vec { + v["entries"] + .as_array() + .unwrap() + .iter() + .map(|e| e["id"].as_str().unwrap().to_string()) + .collect() +} + +#[tokio::test] +async fn total_counts_all_entries_and_offset_pages() { + let (state, _dir) = state_with_history(); + + let first = list(&state, "").await; + assert_eq!(first["entries"].as_array().unwrap().len(), 50); + assert_eq!( + first["total"], 120, + "total must count all history, not the page" + ); + assert_eq!(first["offset"], 0); + assert_eq!(first["limit"], 50); + + let third = list(&state, "offset=100&limit=50").await; + assert_eq!(ids(&third).len(), 20); + assert_eq!(third["total"], 120); + assert_eq!(ids(&third)[0], "h100"); + + let past_end = list(&state, "offset=500").await; + assert!(ids(&past_end).is_empty()); + assert_eq!(past_end["total"], 120); +} + +#[tokio::test] +async fn filters_apply_before_paging() { + let (state, _dir) = state_with_history(); + + let failed = list(&state, "status=failed&limit=10").await; + assert_eq!(failed["total"], 15); + assert_eq!(ids(&failed).len(), 10); + + let movies = list(&state, "category=movies&offset=10").await; + assert_eq!(movies["total"], 15); + assert_eq!(ids(&movies).len(), 5); + + let search = list(&state, "search=release.11").await; + assert_eq!( + ids(&search), + vec![ + "h110", "h111", "h112", "h113", "h114", "h115", "h116", "h117", "h118", "h119" + ] + ); + + let recent = list(&state, "days=7").await; + assert_eq!(recent["total"], 115); +} + +#[tokio::test] +async fn categories_and_stats_cover_the_whole_window_not_the_page() { + let (state, _dir) = state_with_history(); + + let all = list(&state, "limit=5").await; + assert_eq!( + all["categories"], + serde_json::json!(["movies", "old", "tv"]) + ); + let stats = &all["stats"]; + assert_eq!(stats["completed"], 105); + assert_eq!(stats["failed"], 15); + assert_eq!(stats["completed_bytes"], 105_000); + assert_eq!(stats["success_pct"], 88); + assert_eq!(stats["avg_duration_secs"], 60.0); + assert_eq!( + stats["fail_reasons"], + serde_json::json!([{ "reason": "CRC mismatch", "count": 15 }]) + ); + + // The stats honour the time window but ignore status/category/name, so + // the success rate stays meaningful while the table is filtered. + let week = list(&state, "days=7&status=failed&limit=1").await; + assert_eq!(week["stats"]["completed"], 100); + assert_eq!(week["stats"]["failed"], 15); + assert_eq!(week["stats"]["success_pct"], 87); +} diff --git a/e2e/tests/history.spec.ts b/e2e/tests/history.spec.ts index 2915c2bb..d22d8f66 100644 --- a/e2e/tests/history.spec.ts +++ b/e2e/tests/history.spec.ts @@ -65,7 +65,7 @@ test.describe('5. Download History', () => { const error = 'Aborted: only 99.9% of content available (need 100.2%), 6 of 4408 content articles missing'; - await page.route('**/api/history', async (route) => { + await page.route(/\/api\/history(\?|$)/, async (route) => { await route.fulfill({ contentType: 'application/json', body: JSON.stringify({ From 3f28eed065ab206146db6703c49c1121a7743b05 Mon Sep 17 00:00:00 2001 From: thedancingdeveloper <306930456+thedancingdeveloper@users.noreply.github.com> Date: Sat, 10 Oct 2026 00:51:30 +0000 Subject: [PATCH 06/23] fix(frontend): groups view filter, paging and fetch-poll teardown Port of xbmc4lyfe/rustnzb#177. - Bind the group-name filter to a signal so filtering applies to headers and status alike, and reload on change. - Load headers with offset-0 windows so "load more" extends the same list instead of paging away from it. - Stop header-fetch polling on destroy, on fetch failure, and when a new fetch starts, so pollers never stack; give up after 120s. Co-Authored-By: xbmc4lyfe Co-Authored-By: Claude Opus 5.5 --- .../groups/groups-view.component.spec.ts | 86 ++++++++++++++++++- .../features/groups/groups-view.component.ts | 58 ++++++++----- 2 files changed, 121 insertions(+), 23 deletions(-) diff --git a/apps/rustnzb/frontend/src/app/features/groups/groups-view.component.spec.ts b/apps/rustnzb/frontend/src/app/features/groups/groups-view.component.spec.ts index c622bd75..1e8a5ac1 100644 --- a/apps/rustnzb/frontend/src/app/features/groups/groups-view.component.spec.ts +++ b/apps/rustnzb/frontend/src/app/features/groups/groups-view.component.spec.ts @@ -1,7 +1,7 @@ import '@angular/compiler'; -import { describe, expect, it, vi } from 'vitest'; -import { of, throwError } from 'rxjs'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { Subject, of, throwError } from 'rxjs'; import { GroupService } from '../../core/services/group.service'; import { GroupRow, HeaderRow } from '../../core/models/group.model'; @@ -68,4 +68,86 @@ describe('GroupsViewComponent', () => { expect(service.downloadSelected).toHaveBeenCalledWith(7, ['']); expect(snack.open).toHaveBeenCalledWith('Download failed', 'Close', { duration: 5000 }); }); + + it('recomputes the subscribed-group list when the name filter changes (BUG-116)', () => { + const other: GroupRow = { ...group, id: 8, name: 'alt.binaries.movies' }; + const { component } = makeComponent({ + list: vi.fn(() => of({ groups: [group, other], total: 2, limit: 500, offset: 0 })), + }); + component.ngOnInit(); + expect(component.filteredGroups().map((g) => g.id)).toEqual([7, 8]); + component.groupNameFilter.set('MOVIES'); + expect(component.filteredGroups().map((g) => g.id)).toEqual([8]); + component.groupNameFilter.set(''); + expect(component.filteredGroups().map((g) => g.id)).toEqual([7, 8]); + }); + + it('keeps earlier pages when headers are reloaded after "Load more" (BUG-117)', () => { + const page = (start: number, n: number): HeaderRow[] => + Array.from({ length: n }, (_, i) => ({ ...header, id: start + i, message_id: `<${start + i}>` })); + const listHeaders = vi.fn((_id: number, q: { limit: number; offset: number }) => + of({ headers: page(q.offset, Math.min(q.limit, 250 - q.offset)), total: 250, limit: q.limit, offset: q.offset }), + ); + const { component } = makeComponent({ listHeaders }); + component.selectGroup(group); + expect(component.headers().length).toBe(100); + component.loadMore(); + expect(component.headers().length).toBe(200); + // A later refresh (fetch poll, mark-all-read) must not collapse the list + // to just the most recent page. + component.loadHeaders(); + expect(component.headers().length).toBe(200); + expect(component.headers()[0].message_id).toBe('<0>'); + expect(new Set(component.headers().map((h) => h.message_id)).size).toBe(200); + component.markAllRead(); + expect(component.headers().length).toBe(200); + }); + + describe('header fetch polling (BUG-118)', () => { + afterEach(() => vi.useRealTimers()); + + it('stops polling and never shows the give-up toast after the view is destroyed', () => { + vi.useFakeTimers(); + const { component, service, snack } = makeComponent(); + component.selectGroup(group); + component.fetchHeaders(); + snack.open.mockClear(); + component.ngOnDestroy(); + const calls = service.listHeaders.mock.calls.length; + vi.advanceTimersByTime(130_000); + expect(service.listHeaders.mock.calls.length).toBe(calls); + expect(snack.open).not.toHaveBeenCalled(); + }); + + it('stops polling when the fetch request fails', () => { + vi.useFakeTimers(); + const fetch = new Subject(); + const { component, service, snack } = makeComponent({ + fetchHeaders: vi.fn(() => fetch.asObservable()), + }); + component.selectGroup(group); + component.fetchHeaders(); + fetch.error(new Error('no server')); + expect(component.fetching()).toBe(false); + const calls = service.listHeaders.mock.calls.length; + snack.open.mockClear(); + vi.advanceTimersByTime(130_000); + expect(service.listHeaders.mock.calls.length).toBe(calls); + expect(snack.open).not.toHaveBeenCalledWith( + expect.stringContaining('taking longer'), expect.anything(), expect.anything(), + ); + }); + + it('does not stack pollers when fetch is clicked twice', () => { + vi.useFakeTimers(); + const { component, service } = makeComponent(); + component.selectGroup(group); + component.fetchHeaders(); + component.fetchHeaders(); + const before = service.listHeaders.mock.calls.length; + vi.advanceTimersByTime(3000); + expect(service.listHeaders.mock.calls.length - before).toBe(1); + component.ngOnDestroy(); + }); + }); }); diff --git a/apps/rustnzb/frontend/src/app/features/groups/groups-view.component.ts b/apps/rustnzb/frontend/src/app/features/groups/groups-view.component.ts index 0b69b546..3550ad59 100644 --- a/apps/rustnzb/frontend/src/app/features/groups/groups-view.component.ts +++ b/apps/rustnzb/frontend/src/app/features/groups/groups-view.component.ts @@ -1,4 +1,4 @@ -import { Component, OnInit, signal, computed } from '@angular/core'; +import { Component, OnDestroy, OnInit, signal, computed } from '@angular/core'; import { CommonModule } from '@angular/common'; import { FormsModule } from '@angular/forms'; import { MatSnackBar, MatSnackBarModule } from '@angular/material/snack-bar'; @@ -44,7 +44,7 @@ import { showHttpError } from '../../core/http/http-error';