Sync dispatch engine with rustnzb monorepo

Cargo.toml

@@ -1,31 +1,31 @@
 [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"
 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"
 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"
-
-

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<String>) -> Self {
+    pub fn not_found_anywhere(
+        server_id: impl Into<String>,
+        provider_outcomes: impl Into<String>,
+    ) -> 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()
+            ),
         }
     }
 

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);
+    }
+}

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<ServerConfig>);
+
+    /// 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<WorkerPool>);
 
 impl DispatchHandle {
-    pub fn new(pool: Arc<WorkerPool>) -> 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<WorkerPool> {
-        &self.0
+    pub fn new(
+        servers: Arc<Mutex<Vec<ServerConfig>>>,
+        bandwidth: Arc<BandwidthLimiter>,
+        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<ServerConfig>) {
+        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();
     }