From f3c85311f78263b2cbd9edd5d6099d6a73ac2984 Mon Sep 17 00:00:00 2001 From: TheDancingDeveloper Date: Tue, 28 Jul 2026 22:18:41 +0000 Subject: [PATCH] sync: reconcile rustnzb monorepo drift --- Cargo.lock | 67 +- Cargo.toml | 14 +- src/article_failure.rs | 11 +- src/bandwidth.rs | 20 + src/dispatch_engine.rs | 85 ++- src/download_engine.rs | 1115 +++++++++++++++++++++--------- src/lib.rs | 8 +- src/news_engine.rs | 765 -------------------- tests/news_engine_integration.rs | 201 ------ 9 files changed, 926 insertions(+), 1360 deletions(-) delete mode 100644 src/news_engine.rs delete mode 100644 tests/news_engine_integration.rs diff --git a/Cargo.lock b/Cargo.lock index e6eb76f..143b1e7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -67,7 +67,7 @@ checksum = "ae36dc4177970ef04fde5178d3e2429882def40e57a451f919c098f72baa6cec" dependencies = [ "proc-macro2", "quote", - "syn 3.0.2", + "syn 3.0.3", ] [[package]] @@ -105,9 +105,9 @@ checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" [[package]] name = "cc" -version = "1.3.0" +version = "1.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c89588d05638b5b4594a3348a2d6c20277e43a7f5c5202b05cc56888475a47b8" +checksum = "5add81bb678e6cb321aff7fa0dc7689ad82b112dbc032cea19f91d6b8e3582b9" dependencies = [ "find-msvc-tools", "shlex", @@ -199,9 +199,9 @@ dependencies = [ [[package]] name = "either" -version = "1.16.0" +version = "1.17.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "91622ff5e7162018101f2fea40d6ebf4a78bbe5a49736a2020649edf9693679e" +checksum = "9e5e8f6c15a24b9a3ee5efec809ccd006d3b30e8b3bb63c39af737c7f87daa1d" [[package]] name = "equivalent" @@ -470,9 +470,9 @@ checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" [[package]] name = "libc" -version = "0.2.188" +version = "0.2.189" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "22053b6a34f84abc97f9129e61334f40174659a1b9bd18c970b83db6a9a6348b" +checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" [[package]] name = "libsqlite3-sys" @@ -568,9 +568,9 @@ dependencies = [ [[package]] name = "nzb-core" -version = "0.2.16" +version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "aeb1c95d300cb97389207e26a9d49c4e81bd51f4a2119bff327ff4e5e201d6a5" +checksum = "b482d1318e4fa6a7c54be941b5521e50bc8f71eb82de0fe59213bffa5161386e" dependencies = [ "anyhow", "chrono", @@ -589,9 +589,9 @@ dependencies = [ [[package]] name = "nzb-decode" -version = "0.1.2" +version = "0.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4bbbe9a09f91ba963b43db5351d7344af4941a3b9819dc7a20aa98d1dfd2550d" +checksum = "4b396a1ec4d2893cccc42514a5750108fd6ebb74c0c92d879ce94fa8b90f2e57" dependencies = [ "anyhow", "bytes", @@ -605,7 +605,7 @@ dependencies = [ [[package]] name = "nzb-dispatch" -version = "0.2.4" +version = "0.2.7" dependencies = [ "anyhow", "arc-swap", @@ -614,8 +614,8 @@ dependencies = [ "governor", "nzb-core", "nzb-decode", - "nzb-news", "nzb-nntp", + "opentelemetry", "parking_lot", "serde", "serial_test", @@ -629,22 +629,11 @@ dependencies = [ "yenc-simd", ] -[[package]] -name = "nzb-news" -version = "0.1.12" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9edf66efd56281eadf4215e354b67a0b0eceea409e318698fb7912fce5220e70" -dependencies = [ - "nzb-nntp", - "tokio", - "tracing", -] - [[package]] name = "nzb-nntp" -version = "0.2.22" +version = "0.2.23" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a5b9243eb240e1e74ffdf90f9e85e6ca0612ca6ea0e7bbe3d9ad170f30f23e2b" +checksum = "62f41caecf94c0a12ec78ba5cfaad05581f1934505cae5ddd163b272add5346b" dependencies = [ "anyhow", "arc-swap", @@ -672,6 +661,20 @@ version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +[[package]] +name = "opentelemetry" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "236e667b670a5cdf90c258f5a55794ec5ac5027e960c224bff8367a59e1e6426" +dependencies = [ + "futures-core", + "futures-sink", + "js-sys", + "pin-project-lite", + "thiserror 2.0.19", + "tracing", +] + [[package]] name = "parking_lot" version = "0.12.5" @@ -898,9 +901,9 @@ dependencies = [ [[package]] name = "rustls-pki-types" -version = "1.15.0" +version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "764899a24af3980067ee14bc143654f297b22eaebfe3c7b6b211920a5a59b046" +checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96" dependencies = [ "zeroize", ] @@ -955,7 +958,7 @@ checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" dependencies = [ "proc-macro2", "quote", - "syn 3.0.2", + "syn 3.0.3", ] [[package]] @@ -1107,9 +1110,9 @@ dependencies = [ [[package]] name = "syn" -version = "3.0.2" +version = "3.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a207d6d6a2b7fc470b80443726053f18a2481b7e1eee970597051596567987a3" +checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3" dependencies = [ "proc-macro2", "quote", @@ -1166,7 +1169,7 @@ checksum = "43cbfe0cf76104d42a574802844187e84a305e531ed54455f11fbde0f10541cd" dependencies = [ "proc-macro2", "quote", - "syn 3.0.2", + "syn 3.0.3", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index b16a402..c234dcb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "nzb-dispatch" -version = "0.2.4" +version = "0.2.7" edition = "2024" description = "Article-level dispatcher: per-server worker pool, priority gating, retry + hopeless tracking. Part of the nzb-* layered usenet engine." license = "MIT" @@ -8,10 +8,9 @@ repository = "https://github.com/TheDancingDeveloper-org/nzb-dispatch" readme = "README.md" [dependencies] -nzb-nntp = { version = "0.2.17" } -nzb-decode = { version = "0.1.2" } -nzb-core = { version = "0.2.9" } -nzb-news = { version = "0.1.9" } +nzb-nntp = { version = "0.2.23" } +nzb-decode = { version = "0.1.3" } +nzb-core = { version = "0.2.17" } tokio = { version = "1", features = ["full"] } async-trait = "0.1" @@ -19,13 +18,14 @@ parking_lot = "0.12" arc-swap = "1" governor = "0.10" tracing = "0.1" +opentelemetry = "0.28" serde = { version = "1", features = ["derive"] } anyhow = "1" thiserror = "2" unicode-normalization = "0.1" [dev-dependencies] -nzb-nntp = { version = "0.2.17", features = ["test-support"] } +nzb-nntp = { version = "0.2.23", features = ["test-support"] } tempfile = "3" tokio = { version = "1", features = ["full", "test-util"] } tracing-subscriber = { version = "0.3", features = ["env-filter"] } @@ -39,5 +39,3 @@ all = { level = "warn", priority = -1 } [lints.rust] unused = "warn" - - diff --git a/src/article_failure.rs b/src/article_failure.rs index 0dc1ea4..230d136 100644 --- a/src/article_failure.rs +++ b/src/article_failure.rs @@ -102,6 +102,7 @@ impl ArticleFailure { NntpError::ArticleNotFound(_) => ArticleFailureKind::NotFound, NntpError::ServiceUnavailable(_) => ArticleFailureKind::ServerDown, NntpError::Auth(_) | NntpError::AuthRequired(_) => ArticleFailureKind::AuthFailed, + NntpError::PermissionDenied(_) => ArticleFailureKind::PermissionDenied, NntpError::Connection(_) => ArticleFailureKind::ConnectionClosed, NntpError::Io(_) => ArticleFailureKind::ConnectionClosed, NntpError::Timeout(_) => ArticleFailureKind::Timeout, @@ -132,11 +133,17 @@ impl ArticleFailure { /// Article is present nowhere — emitted when every enabled server has /// already been tried for this article and the last attempt failed. - pub fn not_found_anywhere(server_id: impl Into) -> Self { + pub fn not_found_anywhere( + server_id: impl Into, + provider_outcomes: impl Into, + ) -> Self { Self { kind: ArticleFailureKind::NotFound, server_id: server_id.into(), - message: "Article not found on any server".to_string(), + message: format!( + "Article explicitly not found on every eligible provider; outcomes: {}", + provider_outcomes.into() + ), } } diff --git a/src/bandwidth.rs b/src/bandwidth.rs index 834a748..91419d0 100644 --- a/src/bandwidth.rs +++ b/src/bandwidth.rs @@ -80,3 +80,23 @@ impl BandwidthLimiter { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn limiter_can_be_reconfigured_without_recreation() { + let limiter = BandwidthLimiter::new(BandwidthConfig::default()); + assert_eq!(limiter.get_download_bps(), None); + limiter + .acquire_download(NonZeroU32::new(1).unwrap()) + .await + .unwrap(); + + limiter.set_download_bps(NonZeroU32::new(1_000)); + assert_eq!(limiter.get_config().download_bps.unwrap().get(), 1_000); + limiter.set_download_bps(None); + assert_eq!(limiter.get_download_bps(), None); + } +} diff --git a/src/dispatch_engine.rs b/src/dispatch_engine.rs index 4f7f262..e7e8888 100644 --- a/src/dispatch_engine.rs +++ b/src/dispatch_engine.rs @@ -16,8 +16,11 @@ use std::time::Duration; use tokio::sync::mpsc; -use crate::download_engine::{ProgressUpdate, WorkerPool, build_job_submission}; +use crate::bandwidth::BandwidthLimiter; +use crate::download_engine::{ConnectionTracker, ProgressUpdate, WorkerPool, build_job_submission}; +use nzb_core::config::ServerConfig; use nzb_core::models::NzbJob; +use parking_lot::Mutex; /// Article-dispatch engine: accepts jobs, drives NNTP fetches, emits progress. /// @@ -49,11 +52,28 @@ pub trait DispatchEngine: Send + Sync { /// Abort `job_id` with a human-readable reason. Emits /// [`ProgressUpdate::JobAborted`] once outstanding articles drain. - fn abort_job(&self, job_id: &str, reason: String); + /// Returns `true` only for the caller that won terminal ownership. + fn abort_job(&self, job_id: &str, reason: String) -> bool; /// Is `job_id` currently known to the dispatcher? fn has_job(&self, job_id: &str) -> bool; + /// Release a terminal job's dispatcher and assembler resources before + /// post-processing opens the completed files. + fn release_completed_job(&self, job_id: &str); + + /// Replace server configuration and reconcile connection budgets/workers. + fn update_servers(&self, servers: Vec); + + /// Per-server allocated worker slots and configured limits. + fn connection_snapshot(&self) -> Vec<(String, usize, usize)>; + + /// Per-server connections actively transferring articles and limits. + fn active_connection_snapshot(&self) -> Vec<(String, usize, usize)>; + + /// Total allocated worker slots across current server pools. + fn connection_total(&self) -> usize; + /// Re-read the server list and adjust workers to match. Call after any /// mutation to the server config (add, remove, enable, disable, resize). fn reconcile_servers(&self); @@ -107,15 +127,21 @@ pub struct ServerAttemptStats { pub struct DispatchHandle(Arc); impl DispatchHandle { - pub fn new(pool: Arc) -> Self { - Self(pool) - } - - /// Escape hatch: access the underlying pool. Intended for callers that - /// still need pool-specific APIs not yet promoted to the trait (none - /// today, but keeps the migration incremental). - pub fn pool(&self) -> &Arc { - &self.0 + pub fn new( + servers: Arc>>, + bandwidth: Arc, + article_timeout_secs: u64, + ) -> Self { + let tracker = Arc::new(ConnectionTracker::new()); + for server in servers.lock().iter() { + tracker.set_limit(&server.id, &server.name, server.connections as usize); + } + Self(WorkerPool::new( + servers, + bandwidth, + tracker, + article_timeout_secs, + )) } } @@ -142,14 +168,47 @@ impl DispatchEngine for DispatchHandle { self.0.cancel_job(job_id); } - fn abort_job(&self, job_id: &str, reason: String) { - self.0.abort_job(job_id, reason); + fn abort_job(&self, job_id: &str, reason: String) -> bool { + self.0.abort_job(job_id, reason) } fn has_job(&self, job_id: &str) -> bool { self.0.has_job(job_id) } + fn release_completed_job(&self, job_id: &str) { + self.0.release_completed_job(job_id); + } + + fn update_servers(&self, servers: Vec) { + let new_ids: std::collections::HashSet<_> = + servers.iter().map(|server| server.id.clone()).collect(); + for (old_id, _, _) in self.0.conn_tracker().snapshot() { + if !new_ids.contains(&old_id) { + self.0.conn_tracker().remove_server(&old_id); + } + } + for server in &servers { + self.0 + .conn_tracker() + .set_limit(&server.id, &server.name, server.connections as usize); + } + *self.0.servers.lock() = servers; + self.0.reconcile_servers(); + } + + fn connection_snapshot(&self) -> Vec<(String, usize, usize)> { + self.0.conn_tracker().snapshot() + } + + fn active_connection_snapshot(&self) -> Vec<(String, usize, usize)> { + self.0.conn_tracker().connected_snapshot() + } + + fn connection_total(&self) -> usize { + self.0.conn_tracker().total() + } + fn reconcile_servers(&self) { self.0.reconcile_servers(); } diff --git a/src/download_engine.rs b/src/download_engine.rs index 1dfe2fb..02db34e 100644 --- a/src/download_engine.rs +++ b/src/download_engine.rs @@ -13,9 +13,8 @@ //! never torn down on job transitions. Pausing a job causes workers holding //! one of its items to return that item to the queue and pull something //! else. Cancelling a job drains its items and drops in-flight results. -//! - A supervisor task detects "all enabled servers circuit-broken for a -//! given job" and emits [`ProgressUpdate::NoServersAvailable`] so the user -//! can fix config and resume, matching the prior per-engine behaviour. +//! - Transient provider outages leave work queued until circuit-breaker +//! cooldown expires; only definitive provider outcomes resolve an article. //! //! Retry logic (per article): //! 1. Try the article on the current server up to [`MAX_TRIES_PER_SERVER`] @@ -25,8 +24,9 @@ //! 3. On connection loss — requeue and reconnect. //! 4. On decode error — treated like "not available on this server", try //! another. -//! 5. When every enabled server is in `tried_servers` (or circuit-broken), -//! the article is marked failed. +//! 5. An article is missing only when every enabled provider explicitly +//! returned `430`. Circuit-broken and transiently failing providers do +//! not contribute evidence of absence. //! 6. A job only fails if failed articles exceed the threshold and no par2 //! recovery is possible. @@ -51,6 +51,14 @@ use nzb_nntp::error::NntpError; use crate::bandwidth::BandwidthLimiter; +fn increment_counter(name: &'static str) { + opentelemetry::global::meter_provider() + .meter("nzb-dispatch") + .u64_counter(name) + .build() + .add(1, &[]); +} + // --------------------------------------------------------------------------- // Constants // --------------------------------------------------------------------------- @@ -142,6 +150,7 @@ struct ServerSlot { name: String, limit: usize, semaphore: Arc, + connected: Arc, } impl ConnectionTracker { @@ -190,6 +199,7 @@ impl ConnectionTracker { name: server_name.to_string(), limit, semaphore: Arc::new(tokio::sync::Semaphore::new(limit)), + connected: Arc::new(AtomicUsize::new(0)), }, ); if let Some(prev) = prev_limit { @@ -241,6 +251,8 @@ impl ConnectionTracker { server_id: server_id.to_string(), server_name: server_slot.name, semaphore_origin: server_slot.semaphore, + connected: server_slot.connected, + is_connected: false, _permit: permit, }) } @@ -285,6 +297,22 @@ impl ConnectionTracker { .collect() } + /// `(server_id, active, limit)` triples for sockets currently transferring + /// an article. Authenticated sockets waiting for work are idle/free. + pub fn connected_snapshot(&self) -> Vec<(String, usize, usize)> { + let pools = self.pools.lock(); + pools + .iter() + .map(|(id, slot)| { + ( + id.clone(), + slot.connected.load(Ordering::Relaxed).min(slot.limit), + slot.limit, + ) + }) + .collect() + } + /// Total currently-held permits across all servers in the live pool. /// Permits held against orphaned (replaced) semaphores are NOT counted — /// they'll go away as the holding workers exit. @@ -326,6 +354,8 @@ pub struct ConnectionSlot { /// Used by `ConnectionTracker::slot_is_current` to detect a stale slot /// after a `set_limit` shrink (which replaces the semaphore). semaphore_origin: Arc, + connected: Arc, + is_connected: bool, _permit: tokio::sync::OwnedSemaphorePermit, } @@ -336,6 +366,44 @@ impl ConnectionSlot { pub fn server_name(&self) -> &str { &self.server_name } + + fn mark_active(&mut self) { + if !self.is_connected { + self.connected.fetch_add(1, Ordering::Relaxed); + self.is_connected = true; + } + } + + fn mark_inactive(&mut self) { + if self.is_connected { + self.connected.fetch_sub(1, Ordering::Relaxed); + self.is_connected = false; + } + } + + fn activity(&mut self) -> ConnectionActivity<'_> { + self.mark_active(); + ConnectionActivity { slot: self } + } +} + +impl Drop for ConnectionSlot { + fn drop(&mut self) { + self.mark_inactive(); + } +} + +/// Marks a connection as actively transferring an article for the lifetime +/// of the guard. An authenticated but idle socket does not consume a "used" +/// connection in API/UI accounting. +struct ConnectionActivity<'a> { + slot: &'a mut ConnectionSlot, +} + +impl Drop for ConnectionActivity<'_> { + fn drop(&mut self) { + self.slot.mark_inactive(); + } } // --------------------------------------------------------------------------- @@ -401,6 +469,12 @@ pub type ServerHealthMap = Arc>>; #[derive(Debug, Clone)] pub enum ProgressUpdate { + /// Every enabled NNTP provider is temporarily unavailable. The job stays + /// active and will retry automatically when a provider recovers. + WaitingForProviders { job_id: String, message: String }, + /// At least one NNTP provider recovered after a + /// [`ProgressUpdate::WaitingForProviders`] notification. + ProvidersAvailable { job_id: String }, ArticleComplete { job_id: String, file_id: String, @@ -408,6 +482,9 @@ pub enum ProgressUpdate { decoded_bytes: u64, file_complete: bool, server_id: Option, + /// Filename declared by the yEnc header. This lets the queue layer + /// recognize obfuscated PAR2 files before terminal completion. + yenc_filename: Option, }, /// An article could not be retrieved. `failure` carries the typed /// classification of *why* (NotFound, ServerDown, AuthFailed, …). @@ -422,14 +499,13 @@ pub enum ProgressUpdate { job_id: String, success: bool, articles_failed: usize, - }, - NoServersAvailable { - job_id: String, - reason: String, + download_time_secs: f64, }, JobAborted { job_id: String, reason: String, + articles_failed: usize, + download_time_secs: f64, }, } @@ -446,6 +522,9 @@ pub(crate) struct WorkItem { pub(crate) segment_number: u32, /// Servers already tried for this article (by server ID). pub(crate) tried_servers: Vec, + /// Definitive per-provider outcomes. Transient failures are not recorded + /// because they provide no evidence about article availability. + pub(crate) provider_outcomes: HashMap, /// Number of attempts on the current server. pub(crate) tries_on_current: u32, } @@ -480,10 +559,15 @@ pub(crate) struct JobContext { pub total_assemble_us: Arc, pub total_articles_decoded: Arc, pub engine_start: Instant, + active_started: Mutex>, + active_elapsed: Mutex, /// Total bytes across all files (for perf summary throughput). pub total_bytes: u64, /// Ensures JobFinished/JobAborted is only emitted once. finished: AtomicBool, + /// Avoid repeatedly reporting the same all-provider outage while the + /// workers wait for circuit-breaker cooldown. + waiting_for_providers_reported: AtomicBool, } pub(crate) type JobContextMap = Arc>>>; @@ -516,20 +600,67 @@ impl JobContext { total_assemble_us: Arc::new(AtomicU64::new(0)), total_articles_decoded: Arc::new(AtomicU64::new(0)), engine_start: Instant::now(), + active_started: Mutex::new(Some(Instant::now())), + active_elapsed: Mutex::new(Duration::ZERO), total_bytes: job.total_bytes, finished: AtomicBool::new(false), + waiting_for_providers_reported: AtomicBool::new(false), } } - /// Crate-public accessor for [`resolve_one`](Self::resolve_one). Used - /// by alternative `DispatchEngine` impls in sibling modules. - pub(crate) fn resolve_one_public(&self) { - self.resolve_one(); + fn report_waiting_for_providers(&self, message: &str) { + if self + .waiting_for_providers_reported + .compare_exchange(false, true, Ordering::Relaxed, Ordering::Relaxed) + .is_ok() + { + try_send_progress( + &self.progress_tx, + &self.job_id, + ProgressUpdate::WaitingForProviders { + job_id: self.job_id.clone(), + message: message.to_string(), + }, + ); + } } - /// Crate-public accessor for [`emit_terminal`](Self::emit_terminal). - pub(crate) fn emit_terminal_public(&self) { - self.emit_terminal(); + fn report_providers_available(&self) { + if self + .waiting_for_providers_reported + .swap(false, Ordering::Relaxed) + { + try_send_progress( + &self.progress_tx, + &self.job_id, + ProgressUpdate::ProvidersAvailable { + job_id: self.job_id.clone(), + }, + ); + } + } + + fn pause_clock(&self) { + let mut started = self.active_started.lock(); + if let Some(at) = started.take() { + *self.active_elapsed.lock() += at.elapsed(); + } + } + + fn resume_clock(&self) { + let mut started = self.active_started.lock(); + if started.is_none() { + *started = Some(Instant::now()); + } + } + + fn active_elapsed(&self) -> Duration { + let elapsed = *self.active_elapsed.lock(); + elapsed + + self + .active_started + .lock() + .map_or(Duration::ZERO, |at| at.elapsed()) } /// Decrement articles_remaining. If it reaches zero, run deobfuscation @@ -554,6 +685,7 @@ impl JobContext { self.deobfuscate_files(); let download_elapsed = self.engine_start.elapsed(); + let active_elapsed = self.active_elapsed(); let decode_total_us = self.total_decode_us.load(Ordering::Relaxed); let assemble_total_us = self.total_assemble_us.load(Ordering::Relaxed); let articles_decoded = self.total_articles_decoded.load(Ordering::Relaxed); @@ -579,12 +711,15 @@ impl JobContext { let abort_reason = self.abort_reason.lock().clone(); if let Some(reason) = abort_reason { + let failed = self.articles_failed.load(Ordering::Relaxed); try_send_progress( &self.progress_tx, &self.job_id, ProgressUpdate::JobAborted { job_id: self.job_id.clone(), reason, + articles_failed: failed, + download_time_secs: active_elapsed.as_secs_f64(), }, ); return; @@ -598,6 +733,7 @@ impl JobContext { job_id: self.job_id.clone(), success: failed == 0, articles_failed: failed, + download_time_secs: active_elapsed.as_secs_f64(), }, ); } @@ -661,35 +797,30 @@ impl JobContext { // Shared work queue // --------------------------------------------------------------------------- -/// Multi-job FIFO work queue with PAR2-first priority within each submission. +/// Multi-job queue with per-server round-robin fairness and PAR2-first +/// priority within each submission. /// /// Items submitted via [`SharedWorkQueue::submit_items`] are inserted so that /// PAR2 index and volume files land ahead of data files (matching the prior /// per-job ordering), while data files land at the tail. Cross-job ordering -/// is FIFO by submission time, per the chosen FIFO priority model. +/// is FIFO by submission time until a server has served a job; it then prefers +/// another eligible job before returning to that job's backlog. pub(crate) struct SharedWorkQueue { - inner: Mutex, + inner: Mutex, notify: Notify, } -struct InnerState { - /// FIFO-ish queue of work items. Ordering is PAR2-first within each - /// `submit_items` batch; `push_front` (used for fast failover after a - /// per-server failure) prepends. +struct QueueState { items: VecDeque, - /// Per-server round-robin cursor: the `job_id` of the most recent item - /// this server popped. On the next pop the scan prefers items whose - /// `job_id` differs from this value so one active job with a large - /// backlog can't monopolise the server's workers while a sibling job - /// has workable items. Falls back to same-job items when no other - /// job has anything eligible. + /// Last job selected by each server. This is intentionally per server: + /// separate NNTP providers can make independent fair scheduling choices. last_served: HashMap, } impl SharedWorkQueue { pub fn new() -> Self { Self { - inner: Mutex::new(InnerState { + inner: Mutex::new(QueueState { items: VecDeque::new(), last_served: HashMap::new(), }), @@ -704,10 +835,10 @@ impl SharedWorkQueue { items.sort_by_key(|item| par2_sort_key(&item.filename)); let had_items = !items.is_empty(); { - let mut state = self.inner.lock(); - state.items.reserve(items.len()); + let mut q = self.inner.lock(); + q.items.reserve(items.len()); for item in items { - state.items.push_back(item); + q.items.push_back(item); } } if had_items { @@ -749,9 +880,9 @@ impl SharedWorkQueue { server_id: &str, higher_priority_servers: &[String], ) -> (usize, usize) { - let state = self.inner.lock(); - let total = state.items.len(); - let workable = state + let q = self.inner.lock(); + let total = q.items.len(); + let workable = q .items .iter() .filter(|i| !i.tried_servers.iter().any(|s| s == server_id)) @@ -764,83 +895,70 @@ impl SharedWorkQueue { (workable, total) } - /// Pop the next item that can be processed by a worker on `server_id`, - /// biased toward fair round-robin across active jobs. + /// Pop the next item that can be processed by a worker on `server_id`. /// - /// Two-pass scan: - /// 1. Prefer an item whose `job_id` differs from the last one served to - /// this server (fairness — sibling jobs don't starve behind a - /// backlog-heavy job). - /// 2. Fall back to any eligible item (the sibling-preferred pass found - /// nothing; same-job work is fine). - /// - /// Eligibility: - /// - `server_id` NOT in `item.tried_servers`, AND - /// - every server in `higher_priority_servers` IS in `item.tried_servers` - /// (priority gate — matches SABnzbd `get_article()`). + /// Prefers an eligible item from a job other than the one this server last + /// served, then falls back to any eligible item. It also enforces server + /// priority: items where any healthy higher-priority server has not yet + /// tried the article are skipped for the backup server. /// /// `higher_priority_servers` is a caller-prepared list of server IDs with - /// strictly higher priority than the caller, filtered to only enabled + - /// healthy servers. Empty slice disables the priority gate (priority-0 - /// servers, single-server setups, or all peers circuit-broken). + /// strictly higher priority (lower priority number) than the caller, filtered + /// to only enabled + healthy servers. See `run_worker_pipelined` and + /// `run_worker_serial` for the canonical computation. Empty slice disables + /// the priority gate (priority-0 servers, single-server setups, or all + /// higher-priority peers circuit-broken → backup can take over). + /// + /// Returns `None` if the queue is empty or if every item is either already + /// tried here or pending a higher-priority server. fn pop_workable( &self, server_id: &str, higher_priority_servers: &[String], ) -> Option { let mut state = self.inner.lock(); - - let eligible = |item: &WorkItem| -> bool { + let last_served = state.last_served.get(server_id).cloned(); + let is_eligible = |item: &WorkItem| { !item.tried_servers.iter().any(|s| s == server_id) && higher_priority_servers .iter() - .all(|hp| item.tried_servers.contains(hp)) + .all(|server| item.tried_servers.contains(server)) }; - let last_served = state.last_served.get(server_id).cloned(); - - // Pass 1: prefer different job than last served. - let mut chosen = None; - if let Some(ref last) = last_served { - chosen = state - .items - .iter() - .position(|item| eligible(item) && item.job_id != *last); - } - - // Pass 2: any eligible (fallback when no non-last-job eligible items exist). - if chosen.is_none() { - chosen = state.items.iter().position(eligible); + let selected = last_served + .as_deref() + .and_then(|last_job| { + state + .items + .iter() + .position(|item| is_eligible(item) && item.job_id != last_job) + }) + .or_else(|| state.items.iter().position(is_eligible)); + + if let Some(index) = selected { + let item = state.items.remove(index)?; + state + .last_served + .insert(server_id.to_string(), item.job_id.clone()); + Some(item) + } else { + None } - - let idx = chosen?; - // VecDeque::remove(i) is O(min(i, len - i)) — acceptable for typical - // queue lengths. Not using swap_remove_back because it would break - // PAR2-first ordering. - let item = state.items.remove(idx)?; - state - .last_served - .insert(server_id.to_string(), item.job_id.clone()); - Some(item) } /// Remove all items belonging to `job_id`. Used on cancel_job / remove_job. fn drain_job(&self, job_id: &str) -> Vec { - let mut state = self.inner.lock(); - let mut kept = VecDeque::with_capacity(state.items.len()); + let mut q = self.inner.lock(); + let mut kept = VecDeque::with_capacity(q.items.len()); let mut drained = Vec::new(); - while let Some(item) = state.items.pop_front() { + while let Some(item) = q.items.pop_front() { if item.job_id == job_id { drained.push(item); } else { kept.push_back(item); } } - state.items = kept; - // Drop any round-robin cursors pointing at the removed job so the - // next pop doesn't try to prefer items for a job that no longer - // exists. Purely a tidy-up; correctness isn't affected. - state.last_served.retain(|_, v| v != job_id); + q.items = kept; drained } @@ -873,7 +991,7 @@ struct ActiveWorker { pub struct WorkerPool { work_queue: Arc, job_contexts: JobContextMap, - servers: Arc>>, + pub(crate) servers: Arc>>, server_health: ServerHealthMap, bandwidth: Arc, conn_tracker: Arc, @@ -970,6 +1088,42 @@ impl WorkerPool { .collect() } + /// Return a queue-facing message when no enabled provider can currently + /// accept work. A circuit-broken provider is deliberately retried later, + /// so this is a waiting condition rather than an article failure. + fn provider_unavailable_message(&self) -> Option<&'static str> { + let servers = self.servers.lock(); + let enabled: Vec<_> = servers.iter().filter(|server| server.enabled).collect(); + if enabled.is_empty() { + return Some("Waiting for providers: no enabled NNTP servers are configured."); + } + + let health = self.server_health.lock(); + enabled + .iter() + .all(|server| health.get(&server.id).is_some_and(|entry| !entry.is_available())) + .then_some( + "Waiting for providers: every enabled server is temporarily unavailable. rustnzb will retry automatically.", + ) + } + + fn report_provider_outage(&self) { + let Some(message) = self.provider_unavailable_message() else { + return; + }; + let contexts: Vec<_> = self.job_contexts.lock().values().cloned().collect(); + for ctx in contexts { + ctx.report_waiting_for_providers(message); + } + } + + fn report_providers_available(&self) { + let contexts: Vec<_> = self.job_contexts.lock().values().cloned().collect(); + for ctx in contexts { + ctx.report_providers_available(); + } + } + /// Lifetime count of worker evictions performed by the heartbeat /// watchdog. Increases by 1 each time the supervisor reclaims a stalled /// worker. Test harnesses use this as a positive signal that the Phase @@ -1073,13 +1227,32 @@ impl WorkerPool { } self.job_contexts.lock().insert(job_id.clone(), ctx); self.work_queue.submit_items(items); + self.report_provider_outage(); debug!(job_id = %job_id, queue_len = self.work_queue.len(), "Job submitted to worker pool"); } + /// Unregister a normally completed job and close its assembler files. + /// + /// This must only be called after `JobFinished` is received, which means + /// every article has reached a definitive result and no worker can write + /// another segment for this job. Abort and cancellation paths unregister + /// their contexts separately because they may still have in-flight work. + pub(crate) fn release_completed_job(&self, job_id: &str) { + let ctx = self.job_contexts.lock().remove(job_id); + if let Some(ctx) = ctx { + // Workers can briefly retain an Arc after resolving + // the final article. Clear the assembler explicitly so those + // transient references do not keep every output file open during + // post-processing. + ctx.assembler.clear_job(job_id); + } + } + /// Pause a job: workers stop pulling its items, and any item currently /// being held while paused is returned to the queue. pub fn pause_job(&self, job_id: &str) { if let Some(ctx) = self.job_contexts.lock().get(job_id) { + ctx.pause_clock(); ctx.paused.store(true, Ordering::Relaxed); } } @@ -1087,6 +1260,7 @@ impl WorkerPool { /// Resume a paused job. pub fn resume_job(&self, job_id: &str) { if let Some(ctx) = self.job_contexts.lock().get(job_id) { + ctx.resume_clock(); ctx.paused.store(false, Ordering::Relaxed); // Wake any workers that were idle waiting for work. self.work_queue.notify.notify_waiters(); @@ -1095,12 +1269,18 @@ impl WorkerPool { /// Abort a job with a reason. Drains queued items, sets the abort flag, /// and emits JobAborted via the job's progress channel. - pub fn abort_job(&self, job_id: &str, reason: String) { + pub fn abort_job(&self, job_id: &str, reason: String) -> bool { let ctx = self.job_contexts.lock().get(job_id).cloned(); let Some(ctx) = ctx else { - return; + return false; }; - *ctx.abort_reason.lock() = Some(reason); + { + let mut abort_reason = ctx.abort_reason.lock(); + if abort_reason.is_some() { + return false; + } + *abort_reason = Some(reason); + } ctx.cancelled.store(true, Ordering::Relaxed); let drained = self.work_queue.drain_job(job_id); // Decrement the remaining counter for drained items so the terminal @@ -1108,8 +1288,13 @@ impl WorkerPool { for _ in drained { ctx.resolve_one(); } - ctx.emit_terminal(); - self.job_contexts.lock().remove(job_id); + // In-flight items still own the remaining count. They will resolve + // (without writing) as their responses arrive or after a connection + // failure requeues them. Only the final resolution emits JobAborted. + if ctx.articles_remaining.load(Ordering::Relaxed) == 0 { + ctx.emit_terminal(); + } + true } /// Cancel a job silently (no JobFinished / JobAborted emission). @@ -1124,30 +1309,12 @@ impl WorkerPool { let _ = self.work_queue.drain_job(job_id); } - /// Emit NoServersAvailable for a stuck job and unregister it. - fn mark_no_servers(&self, job_id: &str, reason: String) { - let ctx = self.job_contexts.lock().remove(job_id); - let Some(ctx) = ctx else { - return; - }; - ctx.paused.store(true, Ordering::Relaxed); - try_send_progress( - &ctx.progress_tx, - &ctx.job_id, - ProgressUpdate::NoServersAvailable { - job_id: ctx.job_id.clone(), - reason, - }, - ); - // Remove pending work for this job so other jobs aren't blocked. - let _ = self.work_queue.drain_job(job_id); - } - - /// Supervisor loop: periodically detects jobs whose remaining articles - /// cannot possibly be fetched (all enabled servers circuit-broken or /// Per-tick checks that maintain pool health: idle-worker eviction, - /// dead-worker reaping, reconcile (respawn missing workers), starvation - /// diagnostics, and the legacy "all servers broken" pause. + /// dead-worker reaping, worker reconciliation, and starvation diagnostics. + /// + /// Circuit-broken providers retain their queued articles. Workers retry + /// after the provider cooldown instead of converting a transient outage + /// into a persistent per-job pause. async fn supervisor_loop(self: Arc) { let mut ticker = tokio::time::interval(SUPERVISOR_INTERVAL); loop { @@ -1232,8 +1399,6 @@ impl WorkerPool { // can mean either (a) every item has already been tried here, or // (b) every item is still waiting on a higher-priority server // (backup server legitimately idle — not a bug). - let enabled_servers: Vec = - server_priorities.iter().map(|(id, _)| id.clone()).collect(); let now_instant = Instant::now(); for (sid, prio) in &server_priorities { let hp = self.higher_priority_servers(*prio, sid); @@ -1260,46 +1425,6 @@ impl WorkerPool { } } } - - // ---------- 4. Legacy "all servers broken" pause ---------- - if enabled_servers.is_empty() { - continue; - } - let healthy_servers: Vec = { - let health = self.server_health.lock(); - enabled_servers - .iter() - .filter(|sid| health.get(sid.as_str()).is_none_or(|h| h.is_available())) - .cloned() - .collect() - }; - let all_broken = healthy_servers.is_empty(); - - let ctxs: Vec> = self.job_contexts.lock().values().cloned().collect(); - for ctx in ctxs { - if ctx.articles_remaining.load(Ordering::Relaxed) == 0 { - continue; - } - if ctx.cancelled.load(Ordering::Relaxed) { - continue; - } - if all_broken { - let reason = { - let health = self.server_health.lock(); - health - .values() - .filter_map(|h| h.reason.clone()) - .next() - .unwrap_or_else(|| "All servers unavailable".into()) - }; - warn!( - job_id = %ctx.job_id, - remaining = ctx.articles_remaining.load(Ordering::Relaxed), - "All servers circuit-broken — pausing job for user intervention" - ); - self.mark_no_servers(&ctx.job_id, reason); - } - } } } @@ -1423,6 +1548,7 @@ async fn pool_worker( .is_some_and(|h| !h.is_available()) }; if circuit_broken { + pool.report_provider_outage(); tokio::time::sleep(WORKER_IDLE_POLL).await; continue 'reconnect; } @@ -1464,10 +1590,11 @@ async fn pool_worker( if should_exit(&worker_shutdown, &pool) { return; } + pool.report_provider_outage(); tokio::time::sleep(RECONNECT_DELAY).await; continue 'reconnect; } - + pool.report_providers_available(); let pipe_depth = primary_server.pipelining.max(1); let active_conns = pool.conn_tracker.total(); info!( @@ -1510,7 +1637,6 @@ async fn pool_worker( }; let _ = conn.quit().await; - match reconnect_needed { WorkerExit::Reconnect => { // Loop back to the top and reconnect — slot is preserved. @@ -1559,6 +1685,7 @@ async fn next_work_item( continue; }; if ctx.cancelled.load(Ordering::Relaxed) { + ctx.resolve_one(); continue; } // Respect per-job pause: return the item and wait. @@ -1586,7 +1713,7 @@ async fn run_worker_serial( worker_id: &str, worker_shutdown: &Arc, conn: &mut NntpConnection, - _conn_slot: &mut ConnectionSlot, + conn_slot: &mut ConnectionSlot, last_progress: &Arc, ) -> WorkerExit { let mut consecutive_errors: u32 = 0; @@ -1642,25 +1769,28 @@ async fn run_worker_serial( return WorkerExit::Exit; }; - let fetch_fut = - fetch_article_with_retry(conn, &item, &ctx.assembler, primary_server, worker_id); - let result = if let Some(timeout) = pool.stall_timeout { - match tokio::time::timeout(timeout, fetch_fut).await { - Ok(r) => r, - Err(_) => { - warn!( - worker = %worker_id, - server = %primary_server.name, - article = %item.message_id, - "Connection stalled — no response within {}s, reconnecting", - timeout.as_secs() - ); - pool.work_queue.push_front(item); - return WorkerExit::Reconnect; + let result = { + let _activity = conn_slot.activity(); + let fetch_fut = + fetch_article_with_retry(conn, &item, &ctx.assembler, primary_server, worker_id); + if let Some(timeout) = pool.stall_timeout { + match tokio::time::timeout(timeout, fetch_fut).await { + Ok(r) => r, + Err(_) => { + warn!( + worker = %worker_id, + server = %primary_server.name, + article = %item.message_id, + "Connection stalled — no response within {}s, reconnecting", + timeout.as_secs() + ); + pool.work_queue.push_front(item); + return WorkerExit::Reconnect; + } } + } else { + fetch_fut.await } - } else { - fetch_fut.await }; match result { @@ -1690,6 +1820,7 @@ async fn run_worker_serial( decoded_bytes: process_result.decoded_bytes, file_complete: process_result.file_complete, server_id: Some(primary_server.id.clone()), + yenc_filename: process_result.yenc_filename.clone(), }, ); ctx.resolve_one(); @@ -1704,6 +1835,7 @@ async fn run_worker_serial( &pool.server_health, &ctx, &pool.work_queue, + worker_id, crate::article_failure::ArticleFailureKind::NotFound, "Article not found on any server", ) { @@ -1711,6 +1843,7 @@ async fn run_worker_serial( } } Err(ArticleError::ConnectionLost(msg)) => { + increment_counter("nntp.transient_reconnects"); consecutive_errors += 1; warn!( worker = %worker_id, @@ -1734,6 +1867,32 @@ async fn run_worker_serial( } return WorkerExit::Reconnect; } + Err(ArticleError::ProviderUnavailable { kind, message }) => { + increment_counter("nntp.transient_reconnects"); + let is_auth = matches!( + kind, + crate::article_failure::ArticleFailureKind::AuthFailed + | crate::article_failure::ArticleFailureKind::PermissionDenied + ); + warn!( + job_id = %item.job_id, + file_id = %item.file_id, + segment_number = item.segment_number, + message_id = %item.message_id, + server_id = %primary_server.id, + worker_id = %worker_id, + original_failure_kind = kind.as_str(), + "Provider unavailable during article fetch: {message}" + ); + pool.server_health + .lock() + .entry(primary_server.id.clone()) + .or_default() + .record_failure(is_auth, &message); + pool.report_provider_outage(); + pool.work_queue.push_front(item); + return WorkerExit::Reconnect; + } Err(ArticleError::DecodeError(msg)) => { if handle_article_not_available( &mut item, @@ -1742,6 +1901,7 @@ async fn run_worker_serial( &pool.server_health, &ctx, &pool.work_queue, + worker_id, crate::article_failure::ArticleFailureKind::DecodeError, &format!("Decode error: {msg}"), ) { @@ -1749,7 +1909,7 @@ async fn run_worker_serial( } } Err(ArticleError::AssemblyError(msg)) => { - error!(article = %item.message_id, "Assembly error: {msg}"); + error!(job_id = %item.job_id, file_id = %item.file_id, segment_number = item.segment_number, message_id = %item.message_id, server_id = %primary_server.id, worker_id = %worker_id, original_failure_kind = "decode_error", terminal_failure_kind = "decode_error", "Assembly error: {msg}"); try_send_progress( &ctx.progress_tx, &item.job_id, @@ -1779,7 +1939,7 @@ async fn run_worker_pipelined( pipe_depth: u8, worker_shutdown: &Arc, conn: &mut NntpConnection, - _conn_slot: &mut ConnectionSlot, + conn_slot: &mut ConnectionSlot, last_progress: &Arc, ) -> WorkerExit { let mut pipeline = Pipeline::new(pipe_depth); @@ -1861,6 +2021,7 @@ async fn run_worker_pipelined( continue; }; if ctx.cancelled.load(Ordering::Relaxed) { + ctx.resolve_one(); continue; } if ctx.paused.load(Ordering::Relaxed) { @@ -1892,6 +2053,7 @@ async fn run_worker_pipelined( in_flight_items.insert(tag, first_item); } + let _activity = conn_slot.activity(); let flush_t = Instant::now(); if let Err(e) = pipeline.flush_sends(conn).await { warn!( @@ -1959,6 +2121,7 @@ async fn run_worker_pipelined( continue; }; if ctx.cancelled.load(Ordering::Relaxed) { + ctx.resolve_one(); continue; } @@ -1969,7 +2132,12 @@ async fn run_worker_pipelined( let yield_t = Instant::now(); tokio::task::yield_now().await; perf_yield_us += yield_t.elapsed().as_micros() as u64; - match decode_and_assemble(&item, &raw_data, &ctx.assembler) { + let decode_result = decode_and_assemble(&item, &raw_data, &ctx.assembler); + // Return the buffer to the connection's pool so the + // next article's fetch reuses it instead of + // allocating fresh. + conn.release_body_buffer(raw_data); + match decode_result { Ok(process_result) => { perf_decode_us += process_result.decode_us; perf_assemble_us += process_result.assemble_us; @@ -2003,6 +2171,7 @@ async fn run_worker_pipelined( decoded_bytes: process_result.decoded_bytes, file_complete: process_result.file_complete, server_id: Some(primary_server.id.clone()), + yenc_filename: process_result.yenc_filename.clone(), }, ); ctx.resolve_one(); @@ -2045,6 +2214,7 @@ async fn run_worker_pipelined( &pool.server_health, &ctx, &pool.work_queue, + worker_id, crate::article_failure::ArticleFailureKind::DecodeError, &format!("Decode error: {msg}"), ) { @@ -2052,7 +2222,7 @@ async fn run_worker_pipelined( } } Err(ArticleError::AssemblyError(msg)) => { - error!(article = %item.message_id, "Assembly error: {msg}"); + error!(job_id = %item.job_id, file_id = %item.file_id, segment_number = item.segment_number, message_id = %item.message_id, server_id = %primary_server.id, worker_id = %worker_id, original_failure_kind = "decode_error", terminal_failure_kind = "decode_error", "Assembly error: {msg}"); try_send_progress( &ctx.progress_tx, &item.job_id, @@ -2082,6 +2252,7 @@ async fn run_worker_pipelined( &pool.server_health, &ctx, &pool.work_queue, + worker_id, crate::article_failure::ArticleFailureKind::NotFound, "Article not found on any server", ) { @@ -2089,7 +2260,11 @@ async fn run_worker_pipelined( } } Err(NntpError::Connection(_) | NntpError::Io(_)) => { + increment_counter("nntp.transient_reconnects"); warn!( + job_id = %item.job_id, + file_id = %item.file_id, + segment_number = item.segment_number, worker = %worker_id, server = %primary_server.name, host = %primary_server.host, @@ -2108,24 +2283,36 @@ async fn run_worker_pipelined( return WorkerExit::Reconnect; } Err(e) => { - warn!(worker = %worker_id, article = %item.message_id, "Pipeline error: {e}"); - let kind = crate::article_failure::ArticleFailure::from_nntp( + increment_counter("nntp.transient_reconnects"); + let failure = crate::article_failure::ArticleFailure::from_nntp( &e, &primary_server.id, - ) - .kind; - if handle_article_not_available( - &mut item, - primary_server, - &pool.servers, - &pool.server_health, - &ctx, - &pool.work_queue, - kind, - &format!("Pipeline error: {e}"), - ) { - last_progress.store(pool.elapsed_ms(), Ordering::Relaxed); - } + ); + warn!( + job_id = %item.job_id, + file_id = %item.file_id, + segment_number = item.segment_number, + message_id = %item.message_id, + server_id = %primary_server.id, + worker_id = %worker_id, + original_failure_kind = failure.kind.as_str(), + attempt = item.tries_on_current + 1, + "Transient pipeline error — re-queuing in-flight work and reconnecting: {e}" + ); + let is_auth = matches!( + failure.kind, + crate::article_failure::ArticleFailureKind::AuthFailed + | crate::article_failure::ArticleFailureKind::PermissionDenied + ); + pool.server_health + .lock() + .entry(primary_server.id.clone()) + .or_default() + .record_failure(is_auth, &e.to_string()); + pool.report_provider_outage(); + pool.work_queue.push_front(item); + requeue_all(&mut in_flight_items, &pool.work_queue); + return WorkerExit::Reconnect; } } } @@ -2280,45 +2467,72 @@ fn handle_article_not_available( item: &mut WorkItem, primary_server: &ServerConfig, all_servers: &Arc>>, - server_health: &ServerHealthMap, + _server_health: &ServerHealthMap, ctx: &Arc, work_queue: &Arc, + worker_id: &str, kind: crate::article_failure::ArticleFailureKind, error_msg: &str, ) -> bool { - item.tried_servers.push(primary_server.id.clone()); + item.provider_outcomes + .insert(primary_server.id.clone(), kind); + if !item.tried_servers.contains(&primary_server.id) { + item.tried_servers.push(primary_server.id.clone()); + } item.tries_on_current = 0; - let all_tried = { - let servers = all_servers.lock(); - let health = server_health.lock(); - servers.iter().filter(|s| s.enabled).all(|s| { - item.tried_servers.contains(&s.id) - || health.get(&s.id).is_some_and(|h| !h.is_available()) - }) - }; + let all_definitive = + all_enabled_providers_definitive(&all_servers.lock(), &item.provider_outcomes); debug!( article = %item.message_id, server = %primary_server.id, kind = kind.as_str(), tried_count = item.tried_servers.len(), - all_tried, + all_definitive, "Article returned error on this server" ); // (debug log immediately below was added for observability) - if all_tried { - warn!(article = %item.message_id, kind = kind.as_str(), "{error_msg}"); - // Promote a per-server NotFound to a definitive NotFound now that - // every server has been exhausted. DecodeError keeps its kind. - let final_failure = if kind == crate::article_failure::ArticleFailureKind::DecodeError { + if all_definitive { + let mut provider_outcomes = item + .provider_outcomes + .iter() + .map(|(server, outcome)| format!("{server}={}", outcome.as_str())) + .collect::>(); + provider_outcomes.sort_unstable(); + let outcomes = provider_outcomes.join(","); + let all_not_found = item + .provider_outcomes + .values() + .all(|outcome| *outcome == crate::article_failure::ArticleFailureKind::NotFound); + let final_kind = if all_not_found { + crate::article_failure::ArticleFailureKind::NotFound + } else { + crate::article_failure::ArticleFailureKind::DecodeError + }; + warn!( + job_id = %item.job_id, + file_id = %item.file_id, + segment_number = item.segment_number, + message_id = %item.message_id, + server_id = %primary_server.id, + worker_id = %worker_id, + original_failure_kind = kind.as_str(), + terminal_failure_kind = final_kind.as_str(), + attempt = item.tried_servers.len(), + provider_outcomes = %outcomes, + "{error_msg}" + ); + let final_failure = if final_kind == crate::article_failure::ArticleFailureKind::DecodeError + { crate::article_failure::ArticleFailure::decode_error( &primary_server.id, - error_msg.to_string(), + format!("{error_msg}; provider outcomes: {outcomes}"), ) } else { - crate::article_failure::ArticleFailure::not_found_anywhere(&primary_server.id) + increment_counter("articles.explicit_global_absence"); + crate::article_failure::ArticleFailure::not_found_anywhere(&primary_server.id, outcomes) }; try_send_progress( &ctx.progress_tx, @@ -2346,6 +2560,16 @@ fn handle_article_not_available( } } +fn all_enabled_providers_definitive( + servers: &[ServerConfig], + outcomes: &HashMap, +) -> bool { + servers + .iter() + .filter(|server| server.enabled) + .all(|server| outcomes.contains_key(&server.id)) +} + /// Re-queue all in-flight items back to the work queue (on connection loss). fn requeue_all(in_flight: &mut HashMap, work_queue: &Arc) { let items: Vec = in_flight.drain().map(|(_, item)| item).collect(); @@ -2365,7 +2589,7 @@ fn par2_sort_key(filename: &str) -> u8 { } } -fn has_known_extension(name: &str) -> bool { +pub fn has_known_extension(name: &str) -> bool { let lower = name.to_lowercase(); if let Some(dot_pos) = lower.rfind('.') { let ext = &lower[dot_pos + 1..]; @@ -2449,15 +2673,15 @@ pub(crate) fn build_job_submission( .flat_map(|file| { file.articles .iter() - .enumerate() - .filter(|(_, a)| !a.downloaded) - .map(move |(idx, article)| WorkItem { + .filter(|article| !article.downloaded) + .map(move |article| WorkItem { job_id: job.id.clone(), file_id: file.id.clone(), filename: file.filename.clone(), message_id: article.message_id.clone(), - segment_number: (idx as u32) + 1, + segment_number: article.segment_number, tried_servers: Vec::new(), + provider_outcomes: HashMap::new(), tries_on_current: 0, }) }) @@ -2499,7 +2723,11 @@ async fn fetch_article_with_retry( fetch_us, "NNTP fetch complete" ); - return decode_and_assemble(item, &raw_data, assembler); + let result = decode_and_assemble(item, &raw_data, assembler); + // Return the buffer to the connection's pool so the next + // article's fetch reuses it instead of allocating fresh. + conn.release_body_buffer(raw_data); + return result; } Err(NntpError::ArticleNotFound(_)) => { debug!( @@ -2540,11 +2768,12 @@ async fn fetch_article_with_retry( error = %e, "Service unavailable (502) during article fetch — likely rate limited or blocked" ); - return Err(ArticleError::ConnectionLost(format!( - "Service unavailable: {e}" - ))); + return Err(ArticleError::ProviderUnavailable { + kind: crate::article_failure::ArticleFailureKind::ServerDown, + message: e.to_string(), + }); } - Err(e @ NntpError::AuthRequired(_)) => { + Err(e @ (NntpError::AuthRequired(_) | NntpError::Auth(_))) => { warn!( worker = %worker_id, article = %item.message_id, @@ -2552,9 +2781,16 @@ async fn fetch_article_with_retry( error = %e, "Auth required (480) during article fetch — session expired or rate limited" ); - return Err(ArticleError::ConnectionLost(format!( - "Auth required mid-session: {e}" - ))); + return Err(ArticleError::ProviderUnavailable { + kind: crate::article_failure::ArticleFailureKind::AuthFailed, + message: e.to_string(), + }); + } + Err(e @ NntpError::PermissionDenied(_)) => { + return Err(ArticleError::ProviderUnavailable { + kind: crate::article_failure::ArticleFailureKind::PermissionDenied, + message: e.to_string(), + }); } Err(e) => { last_error = Some(format!("{e}")); @@ -2581,7 +2817,7 @@ async fn fetch_article_with_retry( } } - Err(ArticleError::DecodeError( + Err(ArticleError::ConnectionLost( last_error.unwrap_or_else(|| "Unknown error after retries".into()), )) } @@ -2605,6 +2841,11 @@ enum ArticleError { ArticleNotFound, #[error("Connection lost: {0}")] ConnectionLost(String), + #[error("Provider unavailable ({kind:?}): {message}")] + ProviderUnavailable { + kind: crate::article_failure::ArticleFailureKind, + message: String, + }, #[error("Decode error: {0}")] DecodeError(String), #[error("Assembly error: {0}")] @@ -2673,6 +2914,98 @@ fn decode_and_assemble( mod tests { use super::*; + fn worker_pool_without_servers() -> Arc { + WorkerPool::new( + Arc::new(Mutex::new(Vec::new())), + Arc::new(BandwidthLimiter::new(Default::default())), + Arc::new(ConnectionTracker::new()), + 0, + ) + } + + fn test_job(job_id: &str, root: &std::path::Path) -> NzbJob { + NzbJob { + id: job_id.to_string(), + name: job_id.to_string(), + category: "Default".to_string(), + status: nzb_core::models::JobStatus::Downloading, + priority: nzb_core::models::Priority::Normal, + total_bytes: 1, + downloaded_bytes: 0, + file_count: 1, + files_completed: 0, + article_count: 1, + articles_downloaded: 0, + articles_failed: 0, + added_at: chrono::Utc::now(), + completed_at: None, + work_dir: root.to_path_buf(), + output_dir: root.join("complete"), + password: None, + error_message: None, + speed_bps: 0, + server_stats: Vec::new(), + files: Vec::new(), + } + } + + fn insert_test_context(pool: &WorkerPool, job: &NzbJob, assembler: Arc) { + let (progress_tx, _progress_rx) = mpsc::channel(1); + let ctx = Arc::new(JobContext::new(job, assembler, progress_tx, 1)); + pool.job_contexts.lock().insert(job.id.clone(), ctx); + } + + #[test] + fn submission_preserves_declared_segment_numbers() { + let temp = tempfile::tempdir().unwrap(); + let mut job = test_job("segment-identity", temp.path()); + let article = |message_id: &str, segment_number: u32, bytes: u64| nzb_nntp::Article { + message_id: message_id.to_string(), + segment_number, + bytes, + downloaded: false, + data_begin: None, + data_size: None, + crc32: None, + tried_servers: Vec::new(), + tries: 0, + }; + job.files.push(nzb_core::models::NzbFile { + id: "file-1".into(), + filename: "out-of-order.bin".into(), + bytes: 1_000, + bytes_downloaded: 0, + is_par2: false, + par2_setname: None, + par2_vol: None, + par2_blocks: None, + assembled: false, + groups: Vec::new(), + articles: vec![article("third", 3, 300), article("first", 1, 100)], + }); + let (tx, _rx) = mpsc::channel(4); + + let (_ctx, items) = build_job_submission(&job, tx); + + assert_eq!( + items + .iter() + .map(|item| (item.message_id.as_str(), item.segment_number)) + .collect::>(), + vec![("third", 3), ("first", 1)] + ); + } + + #[cfg(target_os = "linux")] + fn open_fd_count_under(root: &std::path::Path) -> usize { + std::fs::read_dir("/proc/self/fd") + .expect("read /proc/self/fd") + .filter_map(Result::ok) + .filter_map(|entry| std::fs::read_link(entry.path()).ok()) + .filter(|target| target.starts_with(root)) + .count() + } + #[test] fn has_known_extension_recognizes_archives() { assert!(has_known_extension("movie.rar")); @@ -2736,6 +3069,7 @@ mod tests { message_id: msg_id.to_string(), segment_number: 1, tried_servers: Vec::new(), + provider_outcomes: HashMap::new(), tries_on_current: 0, } } @@ -2767,6 +3101,141 @@ mod tests { assert_eq!(picked.message_id, "b"); } + #[test] + fn shared_queue_round_robins_eligible_jobs_per_server() { + let q = SharedWorkQueue::new(); + q.submit_items(vec![ + make_item("job-a", "a-1", "first.rar"), + make_item("job-a", "a-2", "second.rar"), + make_item("job-b", "b-1", "third.rar"), + ]); + + assert_eq!(q.pop_workable("srv1", &[]).unwrap().job_id, "job-a"); + assert_eq!(q.pop_workable("srv1", &[]).unwrap().job_id, "job-b"); + assert_eq!(q.pop_workable("srv1", &[]).unwrap().job_id, "job-a"); + } + + #[tokio::test] + async fn provider_outage_is_reported_once_then_cleared_on_recovery() { + let temp = tempfile::tempdir().unwrap(); + let server = ServerConfig::new("srv1", "one.invalid"); + let pool = WorkerPool::new( + Arc::new(Mutex::new(vec![server.clone()])), + Arc::new(BandwidthLimiter::new(Default::default())), + Arc::new(ConnectionTracker::new()), + 0, + ); + let job = test_job("provider-outage", temp.path()); + let (progress_tx, mut progress_rx) = mpsc::channel(4); + let ctx = Arc::new(JobContext::new( + &job, + Arc::new(FileAssembler::new()), + progress_tx, + 1, + )); + pool.job_contexts.lock().insert(job.id.clone(), ctx); + pool.server_health + .lock() + .entry(server.id.clone()) + .or_default() + .record_failure(false, "connection refused"); + pool.server_health + .lock() + .entry(server.id.clone()) + .or_default() + .record_failure(false, "connection refused"); + pool.server_health + .lock() + .entry(server.id.clone()) + .or_default() + .record_failure(false, "connection refused"); + + pool.report_provider_outage(); + pool.report_provider_outage(); + match progress_rx.recv().await.unwrap() { + ProgressUpdate::WaitingForProviders { message, .. } => { + assert!(message.contains("retry automatically")); + } + other => panic!("unexpected progress event: {other:?}"), + } + assert!( + progress_rx.try_recv().is_err(), + "outage event is deduplicated" + ); + + pool.server_health + .lock() + .get_mut(&server.id) + .unwrap() + .record_success(); + pool.report_providers_available(); + assert!(matches!( + progress_rx.recv().await, + Some(ProgressUpdate::ProvidersAvailable { .. }) + )); + } + + #[test] + fn unavailable_provider_does_not_prove_global_absence() { + let servers = vec![ + ServerConfig::new("srv1", "one.invalid"), + ServerConfig::new("srv2", "two.invalid"), + ]; + let mut outcomes = HashMap::new(); + outcomes.insert( + "srv1".to_string(), + crate::article_failure::ArticleFailureKind::NotFound, + ); + + assert!(!all_enabled_providers_definitive(&servers, &outcomes)); + outcomes.insert( + "srv2".to_string(), + crate::article_failure::ArticleFailureKind::NotFound, + ); + assert!(all_enabled_providers_definitive(&servers, &outcomes)); + } + + #[tokio::test] + async fn abort_has_one_owner_and_waits_for_in_flight_resolution() { + let temp = tempfile::tempdir().unwrap(); + let pool = worker_pool_without_servers(); + let job = test_job("abort-drain", temp.path()); + let assembler = Arc::new(FileAssembler::new()); + let (progress_tx, mut progress_rx) = mpsc::channel(8); + let ctx = Arc::new(JobContext::new(&job, assembler, progress_tx, 3)); + ctx.articles_failed.store(2, Ordering::Relaxed); + pool.job_contexts + .lock() + .insert(job.id.clone(), Arc::clone(&ctx)); + pool.work_queue.submit_items(vec![ + make_item(&job.id, "queued-1", "file.rar"), + make_item(&job.id, "queued-2", "file.rar"), + ]); + + assert!(pool.abort_job(&job.id, "original reason".into())); + assert!(!pool.abort_job(&job.id, "overwritten reason".into())); + assert_eq!(ctx.articles_remaining.load(Ordering::Relaxed), 1); + assert!(!ctx.finished.load(Ordering::Relaxed)); + assert!(progress_rx.try_recv().is_err()); + + // Simulate the sole in-flight worker settling after its write handle + // is no longer usable. Only now may the terminal event be published. + ctx.resolve_one(); + match progress_rx.recv().await.unwrap() { + ProgressUpdate::JobAborted { + reason, + articles_failed, + download_time_secs, + .. + } => { + assert_eq!(reason, "original reason"); + assert_eq!(articles_failed, 2); + assert!(download_time_secs >= 0.0); + } + other => panic!("unexpected terminal event: {other:?}"), + } + } + #[test] fn pop_workable_respects_priority() { // Fresh item (tried_servers empty). A backup-priority caller whose @@ -2845,115 +3314,66 @@ mod tests { assert_eq!(remaining.job_id, "j2"); } - // ----------------------------------------------------------------------- - // Per-job round-robin fairness - // ----------------------------------------------------------------------- - #[test] - fn pop_workable_alternates_between_jobs_on_single_server() { - // Prod scenario: two jobs both have many workable items; a single - // server must not drain one job entirely before touching the other. - let q = SharedWorkQueue::new(); - q.submit_items(vec![ - make_item("j1", "a1", "a1.rar"), - make_item("j1", "a2", "a2.rar"), - make_item("j1", "a3", "a3.rar"), - make_item("j2", "b1", "b1.rar"), - make_item("j2", "b2", "b2.rar"), - make_item("j2", "b3", "b3.rar"), - ]); - // Expect alternation: j1, j2, j1, j2, j1, j2. - let mut order: Vec = Vec::new(); - while let Some(item) = q.pop_workable("srv1", &[]) { - order.push(item.job_id); - } - assert_eq!( - order, - vec!["j1", "j2", "j1", "j2", "j1", "j2"], - "single-server pops must alternate across jobs, not drain one" - ); - } + fn release_completed_job_drops_context_and_closes_assembler_files() { + let tempdir = tempfile::tempdir().expect("tempdir"); + let assembler = Arc::new(FileAssembler::new()); + let job_id = "completed-job"; + assembler + .register_file(job_id, "file-1", tempdir.path().join("file.rar"), 1) + .expect("register file"); + assert_eq!(assembler.get_file_progress(job_id, "file-1"), (0, 1)); - #[test] - fn pop_workable_falls_back_when_only_same_job_is_available() { - // Round-robin PREFERS the other job but doesn't forbid same-job when - // that's all that's eligible. - let q = SharedWorkQueue::new(); - q.submit_items(vec![ - make_item("j1", "a1", "a.rar"), - make_item("j1", "a2", "b.rar"), - ]); - let first = q.pop_workable("srv1", &[]).unwrap(); - assert_eq!(first.job_id, "j1"); - let second = q.pop_workable("srv1", &[]).unwrap(); - assert_eq!( - second.job_id, "j1", - "falls back to same job when no sibling" - ); - } + let job = test_job(job_id, tempdir.path()); + let pool = worker_pool_without_servers(); + insert_test_context(&pool, &job, Arc::clone(&assembler)); + assert!(pool.has_job(job_id)); - #[test] - fn per_server_cursors_are_independent() { - // Two servers; cursor state is tracked per server so one server's - // round-robin choice doesn't bias the other. - let q = SharedWorkQueue::new(); - q.submit_items(vec![ - make_item("j1", "a1", "a1.rar"), - make_item("j2", "b1", "b1.rar"), - make_item("j1", "a2", "a2.rar"), - make_item("j2", "b2", "b2.rar"), - ]); - // srv_x has no cursor → picks first eligible = j1-a1. - let x1 = q.pop_workable("srv_x", &[]).unwrap(); - assert_eq!(x1.job_id, "j1"); - // srv_x's cursor is now j1 → next pop wants != j1 = j2-b1. - let x2 = q.pop_workable("srv_x", &[]).unwrap(); - assert_eq!(x2.job_id, "j2"); - // srv_y has never popped — independent from srv_x's j2 cursor. Picks - // first eligible in the remaining queue = j1-a2. - let y1 = q.pop_workable("srv_y", &[]).unwrap(); - assert_eq!( - y1.job_id, "j1", - "srv_y has its own cursor state; srv_x's j2 cursor must not leak" - ); - } + pool.release_completed_job(job_id); - #[test] - fn fairness_respects_tried_servers_and_priority() { - // The fairness preference must not override eligibility: a "preferred - // other-job" item that the server has already tried cannot be picked - // just because of round-robin. Same for priority-gated items. - let q = SharedWorkQueue::new(); - let mut j2_tried = make_item("j2", "b1", "b1.rar"); - j2_tried.tried_servers.push("srv1".to_string()); - q.submit_items(vec![make_item("j1", "a1", "a1.rar"), j2_tried]); - // First pop: j1 (no cursor, first eligible). - let first = q.pop_workable("srv1", &[]).unwrap(); - assert_eq!(first.job_id, "j1"); - // Cursor now points at j1. Fairness wants j2. But j2's item was - // tried by srv1 already → must fall through, returning None. - assert!( - q.pop_workable("srv1", &[]).is_none(), - "must not serve an ineligible item just to satisfy fairness" - ); + assert!(!pool.has_job(job_id)); + assert_eq!(assembler.get_file_progress(job_id, "file-1"), (0, 0)); } + #[cfg(target_os = "linux")] #[test] - fn drained_jobs_clear_last_served_cursor() { - // When a job is drained (cancelled), its entry in the last_served - // map should be cleared so future pops aren't biased toward an - // extinct job. - let q = SharedWorkQueue::new(); - q.submit_items(vec![ - make_item("j1", "a1", "a1.rar"), - make_item("j2", "b1", "b1.rar"), - ]); - let _ = q.pop_workable("srv1", &[]).unwrap(); // serves j1, cursor=j1 - q.drain_job("j1"); - // With j1 gone and last_served cleared, the next pop is unbiased - // and simply returns the first eligible item — j2. - let pick = q.pop_workable("srv1", &[]).unwrap(); - assert_eq!(pick.job_id, "j2"); + fn repeated_completed_jobs_do_not_accumulate_file_descriptors() { + const JOBS: usize = 64; + const FILES_PER_JOB: usize = 8; + + let tempdir = tempfile::tempdir().expect("tempdir"); + let pool = worker_pool_without_servers(); + assert_eq!(open_fd_count_under(tempdir.path()), 0); + + for job_index in 0..JOBS { + let job_id = format!("job-{job_index}"); + let job_dir = tempdir.path().join(&job_id); + let assembler = Arc::new(FileAssembler::new()); + for file_index in 0..FILES_PER_JOB { + assembler + .register_file( + &job_id, + &format!("file-{file_index}"), + job_dir.join(format!("file-{file_index}.rar")), + 1, + ) + .expect("register file"); + } + assert_eq!(open_fd_count_under(tempdir.path()), FILES_PER_JOB); + + let job = test_job(&job_id, &job_dir); + insert_test_context(&pool, &job, assembler); + pool.release_completed_job(&job_id); + + assert!(!pool.has_job(&job_id)); + assert_eq!( + open_fd_count_under(tempdir.path()), + 0, + "completed job {job_index} retained output file descriptors" + ); + } + + assert!(pool.job_contexts.lock().is_empty()); } // ----------------------------------------------------------------------- @@ -3063,4 +3483,35 @@ mod tests { assert_eq!(snap[0], ("srv1".into(), 2, 3)); assert_eq!(snap[1], ("srv2".into(), 1, 5)); } + + #[tokio::test] + async fn active_snapshot_counts_only_article_activity() { + let t = ConnectionTracker::new(); + t.set_limit("srv1", "Server 1", 3); + + let mut connected = t.acquire("srv1").await.unwrap(); + let _disconnected = t.acquire("srv1").await.unwrap(); + assert_eq!(t.connected_snapshot(), vec![("srv1".into(), 0, 3)]); + + connected.mark_active(); + assert_eq!(t.connected_snapshot(), vec![("srv1".into(), 1, 3)]); + + connected.mark_inactive(); + assert_eq!(t.connected_snapshot(), vec![("srv1".into(), 0, 3)]); + } + + #[tokio::test] + async fn activity_guard_clears_count_when_fetch_scope_ends() { + let t = ConnectionTracker::new(); + t.set_limit("srv1", "Server 1", 1); + let mut slot = t.acquire("srv1").await.unwrap(); + + { + let _activity = slot.activity(); + assert_eq!(t.connected_snapshot(), vec![("srv1".into(), 1, 1)]); + } + + assert_eq!(t.connected_snapshot(), vec![("srv1".into(), 0, 1)]); + assert_eq!(t.snapshot(), vec![("srv1".into(), 1, 1)]); + } } diff --git a/src/lib.rs b/src/lib.rs index a424fb8..35a90af 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -18,19 +18,13 @@ pub mod article_failure; pub mod bandwidth; pub mod dispatch_engine; pub mod download_engine; -pub mod news_engine; pub mod util; -pub use news_engine::{NewsDispatchEngine, NewsEngineConfig}; - -// Re-export the underlying probe-policy type so downstream crates (nzb-web, -// Arz, ...) can configure it without depending on nzb-news directly. -pub use nzb_news::ServerProbePolicy; - // Convenience re-exports — the types downstream crates reach for. pub use article_failure::{ArticleFailure, ArticleFailureKind}; pub use bandwidth::{BandwidthConfig, BandwidthLimiter}; pub use dispatch_engine::{DispatchEngine, DispatchHandle}; pub use download_engine::{ ConnectionSlot, ConnectionTracker, ProgressUpdate, ServerHealth, SlotStatus, WorkerPool, + has_known_extension, }; diff --git a/src/news_engine.rs b/src/news_engine.rs deleted file mode 100644 index 343ef8b..0000000 --- a/src/news_engine.rs +++ /dev/null @@ -1,765 +0,0 @@ -//! `NewsDispatchEngine` — `DispatchEngine` impl backed by the `nzb-news` crate. -//! -//! The layered news engine is a pure NNTP fetch layer: it takes per-article -//! work items and emits [`nzb_news::FetchOutcome`]s. This adapter bolts the -//! rest of the pipeline on top of it so it satisfies the contract the old -//! `WorkerPool` engine used to satisfy: -//! -//! 1. **Fetch** — delegated to `nzb_news::spawn_downloader`. -//! 2. **Decode** — `nzb_decode::decode_yenc` on each successful outcome. -//! 3. **Assemble** — `FileAssembler::assemble_article` writes the decoded -//! bytes at the yEnc-declared offset. -//! 4. **Progress** — translates per-article outcomes into -//! [`ProgressUpdate::ArticleComplete`] / [`ProgressUpdate::ArticleFailed`]; -//! drives job-level terminal via `JobContext::resolve_one`. -//! -//! Per-job lifecycle (pause/resume/cancel/abort) is tracked in this adapter -//! because the news engine is job-agnostic. We keep a `JobContext` per job -//! (same struct the old engine used — it owns the assembler, progress -//! channel, deobfuscation state, and terminal-emit logic). -//! -//! MVP limitations — marked with TODO comments: -//! - `pause_job` / `resume_job` are no-ops (work items are submitted -//! eagerly; pause-gating is a follow-up). -//! - `reconcile_servers` is a no-op (nzb-news doesn't expose mid-flight -//! server reconfiguration yet; requires a downloader rebuild). -//! - `set_max_worker_idle` / `eviction_count` are stubs (no idle-worker -//! pool concept in nzb-news). - -use std::collections::{HashMap, VecDeque}; -use std::sync::Arc; -use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; -use std::time::{Duration, Instant}; - -use parking_lot::{Mutex, RwLock}; -use tokio::sync::{Notify, mpsc}; -use tracing::{debug, info}; - -use nzb_core::models::NzbJob; -use nzb_nntp::config::ServerConfig; - -use crate::article_failure::{ArticleFailure, ArticleFailureKind}; -use crate::dispatch_engine::DispatchEngine; -use crate::download_engine::{JobContext, ProgressUpdate, build_job_submission}; - -// --------------------------------------------------------------------------- -// Tuning knobs -// --------------------------------------------------------------------------- - -/// How many articles the nzb-news downloader will hold in-flight across all -/// servers at once. Matches the old engine's rough ceiling. -const DEFAULT_MAX_CONCURRENT_FETCHES: usize = 40; - -/// Work channel depth inside nzb-news. Articles are buffered here between -/// `submit_job` enqueue and the per-server fan-out. -const DEFAULT_WORK_CHANNEL_CAPACITY: usize = 4096; - -/// Outcome channel depth. Must be large enough that a momentary backlog in -/// the decode path doesn't block the fetch loop. -const DEFAULT_OUTCOME_CHANNEL_CAPACITY: usize = 4096; - -// --------------------------------------------------------------------------- -// Config -// --------------------------------------------------------------------------- - -/// Configuration for [`NewsDispatchEngine`]. Mirrors the knobs exposed by -/// the old engine so the swap is drop-in from the caller's perspective. -/// -/// `servers` is held as an `Arc>` so the caller (queue manager) can -/// mutate it and [`DispatchEngine::reconcile_servers`] will pick up the new -/// list without requiring a new config instance or a full engine rebuild. -#[derive(Clone)] -pub struct NewsEngineConfig { - pub servers: Arc>>, - pub article_timeout: Duration, - pub max_concurrent_fetches: usize, - pub work_channel_capacity: usize, - pub outcome_channel_capacity: usize, - /// Optional backup-server probe policy. Forwarded verbatim to - /// `nzb_news::DownloaderConfig::probe_policy`. - /// - /// `None` disables probing (every cascade article tries every server). - /// `Some(_)` enables fast-fail on a backup server when the probed - /// hit-rate falls below the threshold for that job. - /// - /// Defaults to `Some(ServerProbePolicy::default())` which matches the - /// nzb-news default (probe 10 articles, require >=10% hits). - pub probe_policy: Option, -} - -impl NewsEngineConfig { - /// Construct a config from an owned server list. Wraps the list in an - /// `Arc>` internally; if the caller already owns a shared Arc - /// (e.g. queue manager's live server list), use - /// [`NewsEngineConfig::with_shared_servers`] instead so mutations are - /// visible to the engine. - pub fn new(servers: Vec, article_timeout: Duration) -> Self { - Self::with_shared_servers(Arc::new(Mutex::new(servers)), article_timeout) - } - - /// Construct a config sharing an existing `Arc>>` - /// with the caller. Mutating the Arc from outside and then calling - /// [`DispatchEngine::reconcile_servers`] rebuilds the downloader with - /// the latest server list — this is how live "add/remove server" - /// operations reach the fetch layer. - pub fn with_shared_servers( - servers: Arc>>, - article_timeout: Duration, - ) -> Self { - Self { - servers, - article_timeout, - max_concurrent_fetches: DEFAULT_MAX_CONCURRENT_FETCHES, - work_channel_capacity: DEFAULT_WORK_CHANNEL_CAPACITY, - outcome_channel_capacity: DEFAULT_OUTCOME_CHANNEL_CAPACITY, - probe_policy: Some(nzb_news::ServerProbePolicy::default()), - } - } -} - -// --------------------------------------------------------------------------- -// Adapter state -// --------------------------------------------------------------------------- - -/// Shared state held behind an `Arc` so the outcome-dispatcher task can -/// access jobs while the engine is owned by the caller. -struct Inner { - config: NewsEngineConfig, - /// Populated by `start()`. Holds the downloader's work-submission sender. - handle: RwLock>, - /// Job map: `job_id` → per-job state. Cloned into the outcome task. - jobs: RwLock>>, - /// Monotonic tag issued for every `WorkItem` submitted to the downloader. - /// Also serves as the routing key back to the originating article when - /// outcomes come out the other side. - next_tag: AtomicU64, - /// `tag` → in-flight article metadata. We remove on outcome; this is the - /// only place that holds the file_id / segment_number for an article - /// mid-flight. Cleared on job cancel to free memory fast. - in_flight: RwLock>, -} - -/// Per-job state owned by the adapter. -struct JobEntry { - /// Reused from the old engine — owns the assembler, progress channel, - /// deobfuscation state, and terminal emit logic. Everything the adapter - /// needs on the success/failure path is already a field here. - context: Arc, - /// When true, the pump task holds items in `pending` instead of - /// forwarding them to the downloader. In-flight articles still - /// complete — pause gates the *next* work only. - paused: AtomicBool, - /// Set by `cancel_job`. The pump exits and drains the queue. - cancelled: AtomicBool, - /// Work items waiting to be handed to the downloader. `submit_job` - /// pushes all items here; the pump task drains on start and on - /// resume. - pending: Mutex>, - /// Wake-up signal for the pump task. Notified when `submit_job` adds - /// items or when `resume_job` / `cancel_job` changes the gate. - pump_wake: Notify, -} - -/// Metadata recorded when a `WorkItem` is dispatched so we can route the -/// outcome back to the right job / file / segment. -#[derive(Clone)] -struct InFlight { - job_id: String, - file_id: String, - segment_number: u32, -} - -// --------------------------------------------------------------------------- -// NewsDispatchEngine -// --------------------------------------------------------------------------- - -/// `DispatchEngine` impl backed by the layered nzb-news fetch engine. -pub struct NewsDispatchEngine { - inner: Arc, -} - -impl NewsDispatchEngine { - /// Construct the engine. Does **not** spawn the downloader — - /// [`DispatchEngine::start`] does that. - pub fn new(config: NewsEngineConfig) -> Self { - Self { - inner: Arc::new(Inner { - config, - handle: RwLock::new(None), - jobs: RwLock::new(HashMap::new()), - next_tag: AtomicU64::new(1), - in_flight: RwLock::new(HashMap::new()), - }), - } - } -} - -#[async_trait::async_trait] -impl DispatchEngine for NewsDispatchEngine { - fn start(&self) { - let mut slot = self.inner.handle.write(); - if slot.is_some() { - return; // idempotent - } - - let servers_snapshot = self.inner.config.servers.lock().clone(); - if servers_snapshot.is_empty() { - // Deferred start: with zero servers, spawning the downloader - // would create an internal work queue with no per-server - // workers, and any items pushed would sit in limbo. - // `reconcile_servers` will start it once servers are added. - info!("NewsDispatchEngine start deferred — no servers configured"); - return; - } - spawn_and_install_downloader(&self.inner, &mut slot, servers_snapshot); - } - - fn submit_job(&self, job: &NzbJob, progress_tx: mpsc::Sender) { - // Reuse the old engine's job-submission builder: creates the - // FileAssembler, registers files, and filters out already-downloaded - // articles. We only use the returned `JobContext` — the WorkItem - // vec it produces is in the old engine's format; we build nzb-news - // work items fresh below. - let (ctx, legacy_items) = build_job_submission(job, progress_tx); - - // Build nzb-news wrapper types. One NzbObject for the job, one - // NzbFile per file, and one Article per work item. - let news_files: Vec> = job - .files - .iter() - .map(|f| { - Arc::new(nzb_news::NzbFile::new( - &f.id, - &job.id, - &f.filename, - f.articles.len() as u32, - )) - }) - .collect(); - let news_files_by_id: HashMap> = news_files - .iter() - .map(|nf| (nf.id.clone(), Arc::clone(nf))) - .collect(); - let total_articles = legacy_items.len() as u64; - let news_job = Arc::new(nzb_news::NzbObject::new( - &job.id, - &job.name, - total_articles, - job.total_bytes, - news_files.clone(), - )); - - // Convert each legacy WorkItem into an nzb-news WorkItem, recording - // the routing metadata into in_flight and the item itself into the - // job's pending queue. The pump task forwards from pending to the - // downloader, gated by the `paused` flag. - let mut pending = VecDeque::with_capacity(legacy_items.len()); - let tag_counter = &self.inner.next_tag; - for item in legacy_items { - let tag = tag_counter.fetch_add(1, Ordering::Relaxed); - let file = match news_files_by_id.get(&item.file_id) { - Some(f) => Arc::clone(f), - None => continue, // shouldn't happen — file_id came from the same job - }; - self.inner.in_flight.write().insert( - tag, - InFlight { - job_id: item.job_id.clone(), - file_id: item.file_id.clone(), - segment_number: item.segment_number, - }, - ); - let article = Arc::new(nzb_news::Article::new( - item.message_id.clone(), - item.file_id.clone(), - item.job_id.clone(), - 0, - item.segment_number, - tag, - )); - pending.push_back(nzb_news::WorkItem { - tag, - article, - file, - job: Arc::clone(&news_job), - }); - } - - let entry = Arc::new(JobEntry { - context: Arc::clone(&ctx), - paused: AtomicBool::new(false), - cancelled: AtomicBool::new(false), - pending: Mutex::new(pending), - pump_wake: Notify::new(), - }); - self.inner - .jobs - .write() - .insert(ctx.job_id.clone(), Arc::clone(&entry)); - - // Spawn the pump. It acquires a sender from the engine handle on - // each iteration and parks if the downloader is absent — so - // submitting a job before `start()` (or during a 0-server startup - // window) is safe: items wait in `pending` until `reconcile_servers` - // spawns the downloader. - let job_id = job.id.clone(); - tokio::spawn(pump_loop(entry, Arc::clone(&self.inner), job_id)); - } - - fn pause_job(&self, job_id: &str) { - if let Some(entry) = self.inner.jobs.read().get(job_id) { - // Local gate — stops the pump from handing new items to nzb-news. - entry.paused.store(true, Ordering::SeqCst); - } - // Scheduler-level gate — holds already-submitted articles in - // nzb-news's own pending queue. Without this, anything already - // accepted into `work_channel_capacity` (default 4096) would still - // route to servers despite the local gate. - if let Some(h) = self.inner.handle.read().as_ref() { - h.pause_job(job_id); - } - debug!(job_id, "paused"); - } - - fn resume_job(&self, job_id: &str) { - if let Some(entry) = self.inner.jobs.read().get(job_id) { - entry.paused.store(false, Ordering::SeqCst); - entry.pump_wake.notify_waiters(); - } - if let Some(h) = self.inner.handle.read().as_ref() { - h.resume_job(job_id); - } - debug!(job_id, "resumed"); - } - - fn cancel_job(&self, job_id: &str) { - let entry = self.inner.jobs.write().remove(job_id); - if let Some(entry) = entry { - // Signal pump to drain + exit. - entry.cancelled.store(true, Ordering::SeqCst); - entry.pump_wake.notify_waiters(); - // Drop any not-yet-dispatched items so the pump sees an empty - // queue and exits promptly. - entry.pending.lock().clear(); - // Clear in-flight entries for this job so stale outcomes are - // dropped silently by the dispatcher (unknown-tag path). - self.inner - .in_flight - .write() - .retain(|_, m| m.job_id != job_id); - // Purge nzb-news scheduler-level state: items already accepted - // into the downloader's work_channel or pending list get emitted - // as Cancelled outcomes and removed. Without this, a cancelled - // job would keep routing its buffered articles to servers. - if let Some(h) = self.inner.handle.read().as_ref() { - h.purge_job(job_id); - } - debug!(job_id, "cancelled"); - } - } - - fn abort_job(&self, job_id: &str, reason: String) { - // Emit terminal via the existing JobContext machinery — same path - // the old engine uses. `emit_terminal` is idempotent; cancel_job - // later is safe. - let entry = self.inner.jobs.read().get(job_id).cloned(); - if let Some(entry) = entry { - *entry.context.abort_reason.lock() = Some(reason); - entry.context.emit_terminal_public(); - } - self.cancel_job(job_id); - } - - fn has_job(&self, job_id: &str) -> bool { - self.inner.jobs.read().contains_key(job_id) - } - - fn reconcile_servers(&self) { - // Rebuild the downloader with the current server list. - // - // First-time (0 → N): when `start` was deferred for lack of - // servers, pump_loops parked waiting for a handle. Spawning the - // downloader here and notifying pumps resumes dispatch cleanly — - // no items are lost because `pump_loop` leaves unsent items in - // `pending` until a sender is available. - // - // Reconfigure (N → M, N > 0): the downloader is rebuilt and the - // old one shut down. Articles already in the old downloader's - // internal queue that had not completed may be lost; their job - // will stall until nzb-news grows a dynamic-server API. For the - // common "add/edit server" UI flows this is rare in practice and - // the user can retry a stalled job manually. Documented as a - // limitation rather than a silent partial failure. - let servers_snapshot = self.inner.config.servers.lock().clone(); - let server_count = servers_snapshot.len(); - - let old_handle = if servers_snapshot.is_empty() { - // Remove handle; pumps will park until a server is added. - self.inner.handle.write().take() - } else { - let mut slot = self.inner.handle.write(); - let old = slot.take(); - spawn_and_install_downloader(&self.inner, &mut slot, servers_snapshot); - old - }; - - if let Some(old) = old_handle { - old.shutdown(); - } - - // Wake all pump loops so they re-read the handle and either pick - // up the new sender or park on `pump_wake` until one arrives. - let entries: Vec> = self.inner.jobs.read().values().map(Arc::clone).collect(); - for entry in entries { - entry.pump_wake.notify_waiters(); - } - - info!( - servers = server_count, - "NewsDispatchEngine reconciled server list" - ); - } - - fn set_max_worker_idle(&self, _d: Duration) { - // No per-worker idle concept in nzb-news; workers are persistent - // until the downloader shuts down. - } - - fn eviction_count(&self) -> u64 { - 0 - } - - fn server_stats_snapshot(&self) -> Vec<(String, crate::dispatch_engine::ServerAttemptStats)> { - let guard = self.inner.handle.read(); - let Some(h) = guard.as_ref() else { - return Vec::new(); - }; - h.server_stats_snapshot() - .into_iter() - .map(|(id, s)| { - ( - id, - crate::dispatch_engine::ServerAttemptStats { - attempted: s.attempted, - succeeded: s.succeeded, - not_found: s.not_found, - transient_failed: s.transient_failed, - }, - ) - }) - .collect() - } - - async fn shutdown(&self) { - let handle = self.inner.handle.write().take(); - if let Some(h) = handle { - h.shutdown(); - h.join().await; - } - } -} - -// --------------------------------------------------------------------------- -// Outcome dispatcher -// --------------------------------------------------------------------------- - -/// Main loop: consume `FetchOutcome`s from nzb-news and translate each into -/// a `ProgressUpdate`, doing decode + assembly inline on success. Runs until -/// the outcome channel is closed (downloader shutdown). -async fn outcome_dispatcher( - inner: Arc, - mut outcomes: mpsc::Receiver, -) { - while let Some(outcome) = outcomes.recv().await { - match outcome { - nzb_news::FetchOutcome::Success { - tag, - server_id, - bytes, - article_bytes: _, - } => { - // Spawn each success so decode+assemble runs in parallel. - // The old engine got this for free because every worker did - // its own fetch+decode+assemble — centralising here would - // serialise all post-fetch work to a single task. - let inner2 = Arc::clone(&inner); - tokio::spawn(async move { - process_success(inner2, tag, server_id, bytes).await; - }); - } - nzb_news::FetchOutcome::Failed { tag, last_error } => { - process_failure(&inner, tag, last_error); - } - nzb_news::FetchOutcome::Cancelled { tag } => { - // Treat as benign discard — caller (queue manager) will - // observe JobAborted separately via abort_job. - inner.in_flight.write().remove(&tag); - } - } - } - debug!("outcome_dispatcher exiting: channel closed"); -} - -async fn process_success(inner: Arc, tag: u64, server_id: String, raw: Vec) { - let meta = inner.in_flight.write().remove(&tag); - let Some(meta) = meta else { - return; // stale / cancelled - }; - - let entry = inner.jobs.read().get(&meta.job_id).cloned(); - let Some(entry) = entry else { - return; // job cancelled after submit - }; - let ctx = &entry.context; - - // Decode (CPU-bound; SIMD is fast but not free). - let decode_start = Instant::now(); - let decoded = match nzb_decode::decode_yenc(&raw) { - Ok(d) => d, - Err(e) => { - let failure = ArticleFailure::decode_error(server_id, format!("yEnc decode: {e}")); - emit_failed(ctx, &meta, failure); - return; - } - }; - let decode_us = decode_start.elapsed().as_micros() as u64; - - // Record yEnc filename for deobfuscation. - if let Some(ref fname) = decoded.filename - && !fname.is_empty() - { - ctx.yenc_names - .lock() - .insert(meta.file_id.clone(), fname.clone()); - } - - let data_begin = decoded.part_begin.unwrap_or(0); - - // Assemble. - let assemble_start = Instant::now(); - let file_complete = match ctx.assembler.assemble_article( - &meta.job_id, - &meta.file_id, - meta.segment_number, - data_begin, - &decoded.data, - ) { - Ok(b) => b, - Err(e) => { - let failure = ArticleFailure::decode_error(server_id, format!("assembly: {e}")); - emit_failed(ctx, &meta, failure); - return; - } - }; - let assemble_us = assemble_start.elapsed().as_micros() as u64; - - // Timing stats. - ctx.total_decode_us.fetch_add(decode_us, Ordering::Relaxed); - ctx.total_assemble_us - .fetch_add(assemble_us, Ordering::Relaxed); - ctx.total_articles_decoded.fetch_add(1, Ordering::Relaxed); - - // Emit progress. - let decoded_bytes = decoded.data.len() as u64; - let _ = ctx.progress_tx.try_send(ProgressUpdate::ArticleComplete { - job_id: meta.job_id.clone(), - file_id: meta.file_id.clone(), - segment_number: meta.segment_number, - decoded_bytes, - file_complete, - server_id: Some(server_id), - }); - - ctx.resolve_one_public(); -} - -fn process_failure(inner: &Inner, tag: u64, last_error: Option) { - let meta = inner.in_flight.write().remove(&tag); - let Some(meta) = meta else { - return; - }; - let entry = inner.jobs.read().get(&meta.job_id).cloned(); - let Some(entry) = entry else { - return; - }; - let msg = last_error.unwrap_or_else(|| "all servers exhausted".into()); - // nzb-news doesn't carry structured error info at the outcome layer — - // only the last attempt's error string. Pattern-match common causes - // so the hopeless-tracker and queue_manager can distinguish "server - // is broken/quota-exhausted" (transient, don't count toward hopeless) - // from "article genuinely missing everywhere" (counts toward - // hopeless). Without this, an auth/quota failure trickles through as - // NotFound and aborts the job with "articles confirmed missing" — - // confusing diagnostics that blame the content instead of the server. - let kind = classify_error_message(&msg); - let failure = ArticleFailure { - kind, - server_id: String::new(), - message: msg, - }; - emit_failed(&entry.context, &meta, failure); -} - -/// Map an opaque nzb-news error string to a typed [`ArticleFailureKind`]. -/// -/// The strings come from `nzb_nntp::error::NntpError` (via nzb-news) and are -/// the only signal we have at this layer — nzb-news's `FetchOutcome` carries -/// `Option` rather than a structured kind. Order of checks matters: -/// more specific patterns are tested first. -fn classify_error_message(msg: &str) -> ArticleFailureKind { - let m = msg.to_ascii_lowercase(); - // NNTP response codes in the message body are the strongest signal. - if m.contains("(482)") || m.contains("(481)") || m.contains("auth") { - return ArticleFailureKind::AuthFailed; - } - if m.contains("(403)") || m.contains("permission") || m.contains("forbidden") { - return ArticleFailureKind::PermissionDenied; - } - if m.contains("(430)") || m.contains("article not found") || m.contains("no such article") { - return ArticleFailureKind::NotFound; - } - if m.contains("(502)") || m.contains("service unavailable") { - return ArticleFailureKind::ServerDown; - } - if m.contains("timeout") || m.contains("timed out") { - return ArticleFailureKind::Timeout; - } - if m.contains("connection") || m.contains("eof") || m.contains("reset") || m.contains("closed") - { - return ArticleFailureKind::ConnectionClosed; - } - // Default: treat unknown cascade exhaustion as NotFound — same as the - // old behaviour — so genuinely-missing articles still abort hopeless - // NZBs promptly. - ArticleFailureKind::NotFound -} - -#[cfg(test)] -#[allow(clippy::items_after_test_module)] -mod classify_tests { - use super::*; - - #[test] - fn classifies_auth_failures() { - let msg = "Authentication failed: PASS rejected (482): Your block account is fully used"; - assert_eq!(classify_error_message(msg), ArticleFailureKind::AuthFailed); - } - - #[test] - fn classifies_not_found() { - let msg = "NNTP (430) No such article"; - assert_eq!(classify_error_message(msg), ArticleFailureKind::NotFound); - } - - #[test] - fn classifies_service_unavailable() { - let msg = "Service unavailable (502)"; - assert_eq!(classify_error_message(msg), ArticleFailureKind::ServerDown); - } - - #[test] - fn classifies_timeout() { - let msg = "read timed out after 60s"; - assert_eq!(classify_error_message(msg), ArticleFailureKind::Timeout); - } - - #[test] - fn unknown_defaults_to_not_found() { - assert_eq!( - classify_error_message("all servers exhausted"), - ArticleFailureKind::NotFound - ); - } -} - -fn emit_failed(ctx: &JobContext, meta: &InFlight, failure: ArticleFailure) { - ctx.articles_failed.fetch_add(1, Ordering::Relaxed); - let _ = ctx.progress_tx.try_send(ProgressUpdate::ArticleFailed { - job_id: meta.job_id.clone(), - file_id: meta.file_id.clone(), - segment_number: meta.segment_number, - failure, - }); - ctx.resolve_one_public(); -} - -// --------------------------------------------------------------------------- -// Per-job pump task -// --------------------------------------------------------------------------- - -/// Drains a job's `pending` queue into the downloader's work channel, -/// respecting the `paused` gate and exiting on `cancelled`. -/// -/// The pump parks on `pump_wake` when `pending` is empty or when -/// `paused` is true. `submit_job` / `resume_job` notify to wake it. -async fn pump_loop(entry: Arc, inner: Arc, job_id: String) { - loop { - if entry.cancelled.load(Ordering::SeqCst) { - debug!(job_id, "pump exiting: cancelled"); - return; - } - if entry.paused.load(Ordering::SeqCst) { - entry.pump_wake.notified().await; - continue; - } - let next = entry.pending.lock().pop_front(); - let Some(item) = next else { - // Queue empty. submit_job enqueues every article up-front, so - // an empty queue means we're done. Park anyway so cancel can - // wake us. - entry.pump_wake.notified().await; - continue; - }; - - // Snapshot the current sender. If the downloader is absent (not - // started yet, or torn down during reconcile_servers with zero - // servers), stash the item back on the front of `pending` and - // park; reconcile_servers will notify us when a new handle exists. - let sender = inner.handle.read().as_ref().map(|h| h.sender()); - let Some(sender) = sender else { - entry.pending.lock().push_front(item); - entry.pump_wake.notified().await; - continue; - }; - - // Send. On SendError (sender closed mid-reconcile), return the - // item to the queue and park — the new handle is on its way. - if let Err(e) = sender.send(item).await { - entry.pending.lock().push_front(e.0); - entry.pump_wake.notified().await; - continue; - } - } -} - -/// Build a new `DownloaderConfig` from the engine's static knobs plus the -/// given server list, spawn the downloader, install its handle in `slot`, -/// and launch the outcome dispatcher task. Used by both `start()` and -/// `reconcile_servers` to avoid duplicating the construction. -/// -/// Precondition: `servers` is non-empty; caller decides the zero-server -/// policy. `slot` must already be held under a write lock. -fn spawn_and_install_downloader( - inner: &Arc, - slot: &mut Option, - servers: Vec, -) { - let cfg = &inner.config; - let server_count = servers.len(); - let dl_config = nzb_news::DownloaderConfig { - servers, - max_concurrent_fetches: cfg.max_concurrent_fetches, - article_timeout: cfg.article_timeout, - work_channel_capacity: cfg.work_channel_capacity, - outcome_channel_capacity: cfg.outcome_channel_capacity, - probe_policy: cfg.probe_policy.clone(), - }; - let (handle, outcomes) = nzb_news::spawn_downloader(dl_config); - let inner_for_task = Arc::clone(inner); - tokio::spawn(outcome_dispatcher(inner_for_task, outcomes)); - *slot = Some(handle); - info!( - servers = server_count, - "NewsDispatchEngine downloader spawned" - ); -} diff --git a/tests/news_engine_integration.rs b/tests/news_engine_integration.rs deleted file mode 100644 index bf8317c..0000000 --- a/tests/news_engine_integration.rs +++ /dev/null @@ -1,201 +0,0 @@ -//! End-to-end integration test for `NewsDispatchEngine`. -//! -//! Spawns a mock NNTP server that serves yEnc-encoded article bodies, -//! constructs a realistic `NzbJob`, drives it through the engine, and -//! asserts: -//! - `ArticleComplete` events arrive with decoded byte counts -//! - `JobFinished { success: true }` is emitted at terminal -//! - The assembled file on disk contains the original payload -//! -//! This exercises the full adapter surface: submit → pump → fetch → -//! decode → assemble → progress translation → terminal emit. - -use std::collections::HashMap; -use std::path::PathBuf; -use std::sync::Arc; -use std::time::Duration; - -use chrono::Utc; -use tempfile::TempDir; -use tokio::sync::mpsc; - -use nzb_core::models::{JobStatus, NzbFile, NzbJob, Priority}; -use nzb_dispatch::dispatch_engine::DispatchEngine; -use nzb_dispatch::download_engine::ProgressUpdate; -use nzb_dispatch::news_engine::{NewsDispatchEngine, NewsEngineConfig}; -use nzb_nntp::Article as NntpArticle; -use nzb_nntp::testutil::{MockConfig, MockNntpServer, test_config}; - -fn yenc_encode(filename: &str, payload: &[u8]) -> Vec { - // Single-part article; file_offset=0, total_file_size=payload.len(). - let (body, _crc) = yenc_simd::encode_article(payload, filename, 1, 1, 0, payload.len() as u64); - body -} - -fn make_job(work_dir: PathBuf, filename: &str, message_id: &str, bytes: u64) -> NzbJob { - let article = NntpArticle { - message_id: message_id.into(), - segment_number: 1, - bytes, - downloaded: false, - data_begin: None, - data_size: None, - crc32: None, - tried_servers: Vec::new(), - tries: 0, - }; - let file = NzbFile { - id: "f1".into(), - filename: filename.into(), - bytes, - bytes_downloaded: 0, - is_par2: false, - par2_setname: None, - par2_vol: None, - par2_blocks: None, - assembled: false, - groups: vec!["alt.binaries.test".into()], - articles: vec![article], - }; - NzbJob { - id: "j1".into(), - name: "integration-test".into(), - category: "test".into(), - status: JobStatus::Queued, - priority: Priority::Normal, - total_bytes: bytes, - downloaded_bytes: 0, - file_count: 1, - files_completed: 0, - article_count: 1, - articles_downloaded: 0, - articles_failed: 0, - added_at: Utc::now(), - completed_at: None, - work_dir: work_dir.clone(), - output_dir: work_dir, - password: None, - error_message: None, - speed_bps: 0, - server_stats: Vec::new(), - files: vec![file], - } -} - -#[tokio::test] -async fn submit_single_article_job_end_to_end() { - let _ = tracing_subscriber::fmt() - .with_env_filter("nzb_dispatch=debug,nzb_news=debug") - .with_test_writer() - .try_init(); - - // 1. Build a yEnc-encoded article body and serve it via mock NNTP. - let payload = b"hello world via nzb-news engine\n"; - let filename = "hello.txt"; - let msg_id = "msg-integration-1"; - let encoded = yenc_encode(filename, payload); - - let mut articles = HashMap::new(); - articles.insert(msg_id.to_string(), encoded); - let server = MockNntpServer::start(MockConfig { - articles, - ..Default::default() - }) - .await; - - let mut server_cfg = test_config(server.port()); - server_cfg.id = "s1".into(); - server_cfg.priority = 1; - server_cfg.connections = 2; - server_cfg.ramp_up_delay_ms = 0; - - // 2. Build the engine. - let news_cfg = NewsEngineConfig::new(vec![server_cfg], Duration::from_secs(10)); - let engine: Arc = Arc::new(NewsDispatchEngine::new(news_cfg)); - engine.start(); - - // 3. Build the job with a tmpdir as work_dir. - let tmp = TempDir::new().unwrap(); - let work_dir = tmp.path().to_path_buf(); - let job = make_job(work_dir.clone(), filename, msg_id, payload.len() as u64); - - // 4. Submit and collect progress. - let (tx, mut rx) = mpsc::channel::(64); - engine.submit_job(&job, tx); - - let mut article_complete_seen = false; - let mut job_finished_success: Option = None; - let deadline = tokio::time::Instant::now() + Duration::from_secs(10); - while tokio::time::Instant::now() < deadline { - match tokio::time::timeout(Duration::from_secs(5), rx.recv()).await { - Ok(Some(ProgressUpdate::ArticleComplete { - decoded_bytes, - file_complete, - server_id, - .. - })) => { - assert_eq!(decoded_bytes as usize, payload.len()); - assert!(file_complete, "single-article file should complete"); - assert_eq!(server_id.as_deref(), Some("s1")); - article_complete_seen = true; - } - Ok(Some(ProgressUpdate::JobFinished { success, .. })) => { - job_finished_success = Some(success); - break; - } - Ok(Some(ProgressUpdate::ArticleFailed { failure, .. })) => { - panic!("unexpected ArticleFailed: {failure}"); - } - Ok(Some(other)) => { - eprintln!("saw other event: {other:?}"); - } - Ok(None) => break, - Err(_) => panic!("timeout waiting for progress events"), - } - } - - assert!(article_complete_seen, "never saw ArticleComplete"); - assert_eq!( - job_finished_success, - Some(true), - "job should finish successfully" - ); - - // 5. Verify the file on disk matches the payload. - let out = std::fs::read(work_dir.join(filename)).expect("output file exists"); - assert_eq!(out, payload, "decoded bytes should equal input payload"); - - engine.shutdown().await; -} - -#[tokio::test] -async fn cancel_job_removes_from_engine() { - let server = MockNntpServer::start(MockConfig::default()).await; - let mut server_cfg = test_config(server.port()); - server_cfg.id = "s1".into(); - server_cfg.connections = 1; - server_cfg.ramp_up_delay_ms = 0; - - let news_cfg = NewsEngineConfig::new(vec![server_cfg], Duration::from_secs(5)); - let engine: Arc = Arc::new(NewsDispatchEngine::new(news_cfg)); - engine.start(); - - let tmp = TempDir::new().unwrap(); - let job = make_job( - tmp.path().to_path_buf(), - "unused.txt", - "no-such-message", - 10, - ); - let (tx, _rx) = mpsc::channel::(64); - engine.submit_job(&job, tx); - - assert!(engine.has_job("j1"), "job should be registered"); - engine.cancel_job("j1"); - assert!(!engine.has_job("j1"), "job should be gone after cancel"); - - // Skip engine.shutdown() — nzb-news's graceful drain waits on in-flight - // wrapper workers, and those keep retrying the (unfetchable) message - // until max_art_tries is exhausted. Runtime drop at test-fn exit - // terminates everything. -}