Struct AdaptiveWorkerPool 

pub struct AdaptiveWorkerPool { /* private fields */ }

Available on crate features worker-batch or worker-pool or worker only.

Adaptive worker pool with hybrid rayon (CPU) + tokio (async I/O) execution.

Provides two APIs:

• process_batch – CPU-bound work via rayon (JSON parsing, transforms, compression, CEL evaluation)
• fan_out_async – async I/O via tokio (enrichment, external APIs, storage writes)

The pool auto-scales active threads based on CPU/memory pressure using watermark bands. All thresholds are config-cascade overridable and emitted as gauge metrics.

Implementations

impl AdaptiveWorkerPool

pub fn try_new(config: WorkerPoolConfig) -> Result<Self, ConfigError>

Create a new worker pool with the given configuration, validating it.

Resolves max_threads = 0 to the detected CPU count, then validates the resolved config (rejects min_threads > max_threads, bad watermark ordering, zero async_concurrency, etc.) before building. This prevents invalid runtime state that would otherwise panic later in the scaler’s clamp(min, max) (Codex review 2026-06-03).

Errors

Returns ConfigError if the resolved config is invalid.

pub fn new(config: WorkerPoolConfig) -> Self

Create a new worker pool with the given configuration.

Resolves max_threads = 0 to the detected CPU count.

Panics

Panics immediately with a clear message if the config is invalid (e.g. min_threads > max_threads). Use try_new for fallible construction from untrusted config.

pub fn from_cascade(key: &str) -> Result<Self, ConfigError>

Create a new worker pool from the config cascade.

Errors

Returns an error if the config cascade is not initialised or validation fails.

pub fn process_batch<T, R, E, F>(&self, items: &[T], f: F) -> Vec<Result<R, E>>

where T: Sync, R: Send, E: Send, F: Fn(&T) -> Result<R, E> + Sync,

Process a batch of items in parallel using rayon (CPU-bound work).

Each item is processed by the provided closure on a rayon worker thread. A semaphore limits how many threads are active simultaneously (controlled by the scaling controller). Results are returned in input order.

Use this for: JSON parsing, transforms, compression, CEL evaluation, routing. Do NOT use for work that needs .await – use fan_out_async.

pub async fn fan_out_async<T, R, E, F, Fut>( &self, items: &[T], f: F, ) -> Vec<Option<Result<R, E>>>

where T: Sync + Send, R: Send + 'static, E: Send + 'static, F: Fn(&T) -> Fut + Send + Sync, Fut: Future<Output = Result<R, E>> + Send + 'static,

Fan out async work across tokio tasks with bounded concurrency.

Each item is processed by the provided async closure on a tokio task. Concurrency is limited by async_concurrency config.

Return contract

The returned Vec has the same length as items and entries correspond by index (input-order preserved):

• Some(Ok(r)) – task completed successfully with result r
• Some(Err(e)) – task returned Err(e)
• None – task panicked; the panic was logged at error level with the input index. The wrapping Option exists so the panic doesn’t silently shorten the result vector (which was the previous behaviour and violated the input-order contract).

Use this for: enrichment lookups, external API calls, storage writes.

pub async fn fan_out_async_with_policy<T, R, E, F, Fut>( &self, items: &[T], policy: &FanOutPolicy, f: F, ) -> Vec<FanOutResult<R, E>>

where T: Sync + Send, R: Send + 'static, E: Send + 'static, F: Fn(&T) -> Fut + Send + Sync, Fut: Future<Output = Result<R, E>> + Send + 'static,

Concurrency-bounded async fan-out with per-item timeout + cancellation.

Unlike fan_out_async, this streams completions (a JoinSet) so a slow item never blocks faster ones, applies an optional per-item timeout, and stops spawning new work when the policy’s cancellation token fires. Results are returned in input order; each is a FanOutResult. Emits dfe_fanout_* metrics (in-flight gauge, timeout/ panic counters, batch-duration histogram) when the metrics feature is on.

pub fn map_owned<T, R, F>(&self, items: Vec<T>, f: F) -> Vec<R>

where T: Send, R: Send, F: Fn(T) -> R + Sync,

Map owned items in parallel under the concurrency limiter.

Like process_batch but takes OWNED items and a closure that consumes each (so the transform may mutate its own copy), returning an arbitrary R per item. Crucially the semaphore IS applied per item, so this path obeys the scaler target – unlike install. Results are collected in input order.

Used by BatchEngine’s parsed mid-tier transform so the parsed path is throttled identically to the raw path.

pub fn install<R: Send>(&self, f: impl FnOnce() -> R + Send) -> R

Execute a closure on the rayon thread pool.

Provides direct access to the rayon pool for operations that need par_iter_mut or other rayon primitives not covered by the throttled helpers. The semaphore is NOT applied – callers manage their own concurrency. Prefer process_batch or map_owned, which respect the scaler target; reach for install only for genuine rayon-primitive escape hatches.

pub fn register_metrics(&self, manager: &MetricsManager)

Register worker pool metrics with the MetricsManager.

Registers operational metrics and emits threshold gauges with current values. Call this once during startup after creating the pool.

pub fn start_scaling_loop(self: &Arc<Self>, cancel: CancellationToken)

Start the background scaling controller.

The controller samples CPU/memory every scale_interval_secs and adjusts the semaphore permits based on watermark bands. Stops on cancellation.

pub fn set_memory_guard(&self, guard: Arc<MemoryGuard>)

Available on crate feature memory only.

Attach a MemoryGuard for dual-source memory pressure reading.

pub fn set_scaling_pressure(&self, pressure: Arc<ScalingPressure>)

Available on crate feature scaling only.

Attach a ScalingPressure for bidirectional pressure integration.

pub fn active_threads(&self) -> usize

Number of permits currently leased – true in-flight worker count.

An idle pool reports 0 regardless of the scaler target. This is the telemetry-grade “active” count (vs target_threads, the scaler’s desired ceiling).

pub fn target_threads(&self) -> usize

Current scaler target concurrency (the admission ceiling).

pub fn available_threads(&self) -> usize

Headroom: permits that could be leased right now (target - leased).

pub fn max_threads(&self) -> usize

Maximum thread count (pool size).