From 8ceb1352dd105fdf7b2572040780f94f4a23c9d0 Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Sun, 7 Jun 2026 20:19:15 +0200 Subject: [PATCH] fix(manager-core): F-P-007 concurrent queue dispatch per tick MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit tick_queue_arm called list_active_queue_consumers() and then iterated serially, awaiting one queue.claim(app, queue) per consumer. With N consumers and a 100ms tick the worst-case throughput was N × (claim + executor) / tick — one slow handler blocked every other queue's progress on the dispatcher's task. Replace the for-loop with `futures::stream::iter(consumers) .for_each_concurrent(QUEUE_DISPATCH_PARALLELISM, …)`. The execution gate already caps real script-concurrency to its permits, so this just removes the dispatcher-side serialization. QUEUE_DISPATCH_PARALLELISM = 32 (matches the default gate). Workspace gains `futures = 0.3` so `for_each_concurrent` is available. AUDIT.md anchor: F-P-007. Co-Authored-By: Claude Opus 4.7 (1M context) --- Cargo.lock | 29 +++++++++++++++++++++++++++ Cargo.toml | 4 +++- crates/manager-core/Cargo.toml | 1 + crates/manager-core/src/dispatcher.rs | 28 +++++++++++++++++++++----- 4 files changed, 56 insertions(+), 6 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 6db26bb..c9c897b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -830,6 +830,21 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "futures" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b147ee9d1f6d097cef9ce628cd2ee62288d963e16fb287bd9286455b241382d" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + [[package]] name = "futures-channel" version = "0.3.32" @@ -874,6 +889,17 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718" +[[package]] +name = "futures-macro" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "futures-sink" version = "0.3.32" @@ -892,8 +918,10 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" dependencies = [ + "futures-channel", "futures-core", "futures-io", + "futures-macro", "futures-sink", "futures-task", "memchr", @@ -1862,6 +1890,7 @@ dependencies = [ "chrono-tz", "cron", "data-encoding", + "futures", "hex", "hmac", "lettre", diff --git a/Cargo.toml b/Cargo.toml index 59900cf..7153515 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -53,8 +53,10 @@ chrono = { version = "0.4", features = ["serde"] } chrono-tz = "0.9" cron = "0.12" -# Async traits +# Async traits + bounded-concurrency stream utilities (queue dispatcher +# uses `for_each_concurrent`). async-trait = "0.1" +futures = "0.3" # Rhai scripting. Pinned exactly (`=1.24`) because the `internals` # feature surface is not semver-stable — future bumps must be deliberate. diff --git a/crates/manager-core/Cargo.toml b/crates/manager-core/Cargo.toml index 8de9825..eabe31f 100644 --- a/crates/manager-core/Cargo.toml +++ b/crates/manager-core/Cargo.toml @@ -14,6 +14,7 @@ picloud-executor-core.workspace = true picloud-orchestrator-core.workspace = true async-trait.workspace = true +futures.workspace = true axum.workspace = true rand.workspace = true serde.workspace = true diff --git a/crates/manager-core/src/dispatcher.rs b/crates/manager-core/src/dispatcher.rs index bf5431c..d7d9f4c 100644 --- a/crates/manager-core/src/dispatcher.rs +++ b/crates/manager-core/src/dispatcher.rs @@ -79,6 +79,11 @@ const TICK_INTERVAL: Duration = Duration::from_millis(100); /// 5-minute cap. const ASYNC_EXEC_TIMEOUT: Duration = Duration::from_secs(300); +/// F-P-007: per-tick concurrency cap for queue dispatches. Matches the +/// default execution-gate concurrency so the gate is what bounds real +/// parallelism, not the dispatcher's serial-await. +const QUEUE_DISPATCH_PARALLELISM: usize = 32; + impl Dispatcher { /// Spawn the dispatcher loop as a detached `tokio::task`. Also /// spawns the v1.1.9 queue visibility-timeout reclaim task. Both @@ -149,17 +154,30 @@ impl Dispatcher { /// attempt one claim per `(app_id, queue_name)`. Each successful /// claim runs through the same executor path as outbox-backed /// trigger dispatches, so retry / dead-letter wiring is shared. + /// + /// F-P-007: bound concurrency with `for_each_concurrent` instead of + /// serial-await. With N consumers and a 100ms tick the worst-case + /// throughput was N × (claim + executor) / tick; concurrent dispatch + /// lets one slow handler not block the others. The execution gate + /// still caps real parallelism — this just removes the + /// dispatcher-side serialization. async fn tick_queue_arm(&self) -> Result<(), DispatcherError> { + use futures::stream::{self, StreamExt}; let consumers = self .triggers .list_active_queue_consumers() .await .map_err(|e| DispatcherError::ResolveTrigger(e.to_string()))?; - for consumer in consumers { - if let Err(err) = self.dispatch_one_queue(&consumer).await { - tracing::warn!(?err, queue = %consumer.queue_name, "queue dispatch errored"); - } - } + // Concurrency cap matches the execution gate's typical default; + // overflow attempts return Err(Overloaded) and back off so a + // queue burst doesn't hammer the executor. + stream::iter(consumers) + .for_each_concurrent(QUEUE_DISPATCH_PARALLELISM, |consumer| async move { + if let Err(err) = self.dispatch_one_queue(&consumer).await { + tracing::warn!(?err, queue = %consumer.queue_name, "queue dispatch errored"); + } + }) + .await; Ok(()) }