Struct BatchEngine 

pub struct BatchEngine { /* private fields */ }

Available on crate feature worker-batch and (crate features worker-batch or worker-pool or worker) only.

Core batch processing engine for DFE pipelines.

Provides two processing modes:

• process_mid_tier – parse JSON via SIMD, extract known fields, apply pre-route filters, then parallel transform via rayon. The standard path for most DFE apps (loader, archiver, transforms).

• process_raw – skip parsing, apply pre-route on raw bytes, then parallel transform via rayon. For apps that handle raw bytes (receiver, binary protocols).

Both modes chunk large batches, track stats atomically, and pause between chunks when memory pressure is detected.

Implementations

impl BatchEngine

pub fn new(config: BatchProcessingConfig) -> Self

Create a standalone engine with its own worker pool.

Uses WorkerPoolConfig::default() for the pool. Prefer with_pool when a ServiceRuntime pool exists.

pub fn with_pool( pool: Arc<AdaptiveWorkerPool>, config: BatchProcessingConfig, ) -> Self

Create an engine that reuses an existing worker pool.

This is the preferred constructor when ServiceRuntime is available, as it avoids creating a second rayon thread pool.

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

Load configuration from the cascade and create a standalone engine.

Errors

Returns ConfigError if the cascade contains invalid data.

pub fn stats(&self) -> &Arc<PipelineStats>

Pipeline statistics (atomic, lock-free).

pub fn pool(&self) -> &Arc<AdaptiveWorkerPool>

Underlying worker pool.

pub fn config(&self) -> &BatchProcessingConfig

Engine configuration.

pub fn auto_wire( &mut self, metrics_manager: &MetricsManager, memory_guard: Option<&Arc<MemoryGuard>>, )

Auto-wire engine with infrastructure components.

Called by ServiceRuntime::build(). Apps never call this directly.

pub fn process_mid_tier<O, E, F>( &self, messages: &[RawMessage], transform: F, ) -> Vec<Result<O, E>>

where O: Send, E: Send + From<String>, F: Fn(&mut ParsedMessage) -> Result<O, E> + Sync,

Parse, filter, and transform a batch of raw messages.

Pipeline phases per chunk:

1. Pre-route – SIMD field extraction + filter evaluation (sequential, ~100 ns/msg)
2. Parse – sonic_rs::from_slice + known-field extraction (sequential, ~1-5 µs/msg)
3. Transform – user closure via rayon par_iter_mut (parallel)

Results contain one entry per non-filtered message. Filtered messages are silently removed (their commit tokens remain accessible via the original slice). DLQ’d and parse-error messages produce Err entries.

pub fn process_raw<O, E, F>( &self, messages: &[RawMessage], transform: F, ) -> Vec<Result<O, E>>

where O: Send, E: Send + From<String>, F: Fn(&RawMessage) -> Result<O, E> + Sync,

Pre-route and transform a batch of raw messages without parsing.

The transform closure receives immutable &RawMessage references. Use this for apps that handle raw bytes directly (e.g. receiver forwarding).

pub async fn run<R, O, E, Transform, Sink>( &self, receiver: &R, shutdown: CancellationToken, transform: Transform, sink: Sink, ) -> Result<(), EngineError>

where R: TransportReceiver, O: Send + 'static, E: Send + From<String> + Display + 'static, Transform: Fn(&mut ParsedMessage) -> Result<O, E> + Sync, Sink: FnMut(Vec<Result<O, E>>) -> Result<(), EngineError>,

Available on crate feature transport only.

Transport-wired async run loop for mid-tier (parsed JSON) processing.

Receives up to config.max_chunk_size messages per iteration, converts them to RawMessage, runs process_mid_tier, calls the sink, then commits transport tokens only on sink success.

Stops cleanly when shutdown is cancelled.

Errors

Returns EngineError::Transport if recv or commit fails fatally. Returns EngineError::Sink if the sink callback errors (commit skipped).

pub async fn run_raw<R, O, E, Transform, Sink>( &self, receiver: &R, shutdown: CancellationToken, transform: Transform, sink: Sink, ) -> Result<(), EngineError>

where R: TransportReceiver, O: Send + 'static, E: Send + From<String> + Display + 'static, Transform: Fn(&RawMessage) -> Result<O, E> + Sync, Sink: FnMut(Vec<Result<O, E>>) -> Result<(), EngineError>,

Available on crate feature transport only.

Transport-wired async run loop for raw byte processing.

Like run but uses process_raw: the transform closure receives &RawMessage bytes without JSON parsing.

Errors

Returns EngineError::Transport if recv or commit fails fatally. Returns EngineError::Sink if the sink callback errors (commit skipped).

pub async fn run_async<R, O, E, Transform, Sink, SinkFut, Ticker, TickerFut>( &self, receiver: &R, shutdown: CancellationToken, transform: Transform, sink: Sink, ticker: Option<(Duration, Ticker)>, ) -> Result<(), EngineError>

where R: TransportReceiver, O: Send + 'static, E: Send + From<String> + Display + 'static, Transform: Fn(&mut ParsedMessage) -> Result<O, E> + Sync, Sink: FnMut(Vec<Result<O, E>>, Vec<R::Token>) -> SinkFut, SinkFut: Future<Output = Result<(), EngineError>>, Ticker: FnMut() -> TickerFut, TickerFut: Future<Output = Result<(), EngineError>>,

Available on crate feature transport only.

Transport-wired async run loop with full control over commit semantics.

Like run but with two key differences:

1. Async sink – the sink is an async closure, enabling async I/O (ClickHouse inserts, Kafka produce, storage writes) inside the sink.

2. Sink-managed commits – the sink receives commit tokens and decides when to commit. The engine does NOT auto-commit. This enables deferred commit patterns (e.g., commit after ClickHouse flush, not after buffer push).

An optional ticker fires on a fixed interval within the select! loop, enabling flush timers and periodic maintenance without breaking out of the engine loop.

Errors

Returns EngineError::Transport if recv fails fatally. Returns EngineError::Sink if the sink callback errors.

pub async fn run_raw_async<R, O, E, Transform, Sink, SinkFut, Ticker, TickerFut>( &self, receiver: &R, shutdown: CancellationToken, transform: Transform, sink: Sink, ticker: Option<(Duration, Ticker)>, ) -> Result<(), EngineError>

where R: TransportReceiver, O: Send + 'static, E: Send + From<String> + Display + 'static, Transform: Fn(&RawMessage) -> Result<O, E> + Sync, Sink: FnMut(Vec<Result<O, E>>, Vec<R::Token>) -> SinkFut, SinkFut: Future<Output = Result<(), EngineError>>, Ticker: FnMut() -> TickerFut, TickerFut: Future<Output = Result<(), EngineError>>,

Available on crate feature transport only.

Transport-wired async run loop for raw byte processing with full control.

Like run_async but uses process_raw: the transform closure receives &RawMessage bytes without JSON parsing.

Errors

Returns EngineError::Transport if recv fails fatally. Returns EngineError::Sink if the sink callback errors.

Trait Implementations

impl Debug for BatchEngine

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.