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 in-process processing modes (the run-loop drivers live in the driver module, gated on the transport feature):

• 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 take the canonical Record slice (the same currency the WorkBatch carries), chunk large batches, and track stats atomically.

Inbound braking under memory pressure is no longer a blocking pause between chunks (the retired check_memory_pressure proto-actuator). It is now the self-regulation governor’s job: the inbound GATE pauses the source transport and the streaming byte-budget lever bounds peak in-flight memory. See run_governed. With the governor OFF, there is no active brake – that is the deliberate opt-out.

Implementations

impl BatchEngine

pub async fn run_workbatch<R, P, Sink, SinkFut, Ticker, TickerFut>( &self, receiver: &R, shutdown: CancellationToken, process: P, sink: Sink, commit: CommitMode, ticker: Option<(Duration, Ticker)>, ) -> Result<(), EngineError>

where R: TransportReceiver, P: Fn(WorkBatch<R::Token>) -> Result<WorkBatch<R::Token>, EngineError>, Sink: FnMut(&WorkBatch<R::Token>) -> SinkFut, SinkFut: Future<Output = Result<(), EngineError>>, Ticker: FnMut() -> TickerFut, TickerFut: Future<Output = Result<(), EngineError>>,

Available on crate feature transport only.

Unified on-demand WorkBatch driver – the default data-plane loop.

Drives one WorkBatch at a time through recv -> filter-DLQ policy -> ingress lease -> process -> sink -> commit. The driver does NOT pre-parse: process reads fields on demand via codec::parse. Pass-through apps pay zero parse cost.

• process runs on the loop task (cancellation-aware between awaits) and may fan records out or in; it MUST preserve commit_tokens (use WorkBatch::map_records, which does so automatically).
• sink is async and receives the WHOLE out-batch by reference.
• commit selects CommitMode::Auto (engine commits after sink Ok) or CommitMode::SinkManaged (sink owns commit).
• ticker is an optional (interval, fn) that fires on the interval inside the select loop (flush timers, periodic maintenance).

Stops cleanly when shutdown is cancelled.

Errors

Returns EngineError::Transport if recv fails fatally, EngineError::FilterDlqUnrouted if inline-DLQ entries appear under the default FilterDlqPolicy::Reject, or the error returned by process.

A sink error (and, under CommitMode::Auto, a commit error) is TERMINAL: it stops the run loop and propagates. This is the ack barrier for the ORDERED/cumulative source commit (Kafka “commit up to offset N”): the failed block’s tokens are NOT committed, and – crucially – no LATER block is fetched and committed past them, which would silently skip the never-sent records (data loss). On restart the source re-delivers from the last committed watermark, preserving at-least-once. The app owns restart/retry policy.

pub async fn run_workbatch_streaming<R, P, Sink, SinkFut, Ticker, TickerFut>( &self, receiver: &R, shutdown: CancellationToken, process: P, sink: Sink, commit: CommitMode, sub_block_bytes: u64, ticker: Option<(Duration, Ticker)>, ) -> Result<(), EngineError>

where R: TransportReceiver, P: Fn(WorkBatch<R::Token>) -> Result<WorkBatch<R::Token>, EngineError>, Sink: FnMut(&WorkBatch<R::Token>) -> SinkFut, SinkFut: Future<Output = Result<(), EngineError>>, Ticker: FnMut() -> TickerFut, TickerFut: Future<Output = Result<(), EngineError>>,

Available on crate feature transport only.

Streaming WorkBatch driver – the opt-in peak-memory-bounded path.

Identical loop shape to run_workbatch, but each received block is processed in consecutive byte-budget-sized SUB-BLOCKS rather than all at once. Peak in-flight ingress memory is bounded to ONE sub-block (~sub_block_bytes) instead of the whole block: the per-sub-block ingress lease is dropped (releasing those bytes) BEFORE the next sub-block is leased and processed.

The source acks for the WHOLE block are committed EXACTLY ONCE, after the FINAL sub-block’s sink returns Ok (under CommitMode::Auto) – so at-least-once is preserved: a sink error on any sub-block stops the block and skips the commit, so the WHOLE block is re-delivered. The sub-block views carry EMPTY commit_tokens; the batch’s tokens are committed once at the end.

sub_block_bytes is the target sum of payload.len() per sub-block (floor one record, so a record larger than the target is still its own sub-block and the loop never stalls). It is taken as a parameter so the path is testable; Phase 3 wires the byte budget from the governor.

Fan-out WITHIN a sub-block’s process is fine (records grow); the source acks are still the batch’s input tokens, committed once at the end.

Errors

Same as run_workbatch.

pub async fn run_governed<R, P, Sink, SinkFut, Ticker, TickerFut>( &self, receiver: &R, shutdown: CancellationToken, process: P, sink: Sink, commit: CommitMode, ticker: Option<(Duration, Ticker)>, ) -> Result<(), EngineError>

where R: TransportReceiver, P: Fn(WorkBatch<R::Token>) -> Result<WorkBatch<R::Token>, EngineError>, Sink: FnMut(&WorkBatch<R::Token>) -> SinkFut, SinkFut: Future<Output = Result<(), EngineError>>, Ticker: FnMut() -> TickerFut, TickerFut: Future<Output = Result<(), EngineError>>,

Available on crate features governor and transport only.

Governed WorkBatch driver – the default-ON self-regulation run path.

This is what a self-regulating app calls instead of choosing between run_workbatch and run_workbatch_streaming by hand. It dispatches on whether the byte-budget lever is wired (set_byte_budget, done by ServiceRuntime when self_regulation.enabled = true):

• Governor ON (budget wired): streams each received block in sub-blocks sized to the CURRENT byte budget (re-read per block), bounds peak in-flight memory to one sub-block, and folds each block’s (bytes, process_time, ingest_interval) into the AIMD loop via observe. The recv max is capped to the budget’s poll-safety record_cap. While pressure is LOW the budget sits at its big start value, so the block becomes a SINGLE sub-block – no per-record overhead, behaviour matches the whole-batch loop.
• Governor OFF (no budget): delegates verbatim to run_workbatch – byte-identical to pre-governor behaviour.

The inbound GATE (Kafka pause-partitions / HTTP-gRPC 503) is wired SEPARATELY into the receive transport, not here – this method is the driver-side lever (sub-block sizing + AIMD), the gate is the transport-side brake. The two share the same UnifiedPressure.

Errors

Same as run_workbatch.

pub async fn run_workbatch_parsed<R, P, Sink, SinkFut, Ticker, TickerFut>( &self, receiver: &R, shutdown: CancellationToken, process_parsed: P, sink: Sink, commit: CommitMode, ticker: Option<(Duration, Ticker)>, ) -> Result<(), EngineError>

where R: TransportReceiver, P: Fn(ParsedBatch<'_, R::Token>) -> Result<WorkBatch<R::Token>, EngineError>, Sink: FnMut(&WorkBatch<R::Token>) -> SinkFut, SinkFut: Future<Output = Result<(), EngineError>>, Ticker: FnMut() -> TickerFut, TickerFut: Future<Output = Result<(), EngineError>>,

Available on crate feature transport only.

Unified pre-parsed WorkBatch driver – the opt-in hot path.

Identical loop shape to run_workbatch, except the driver PRE-PARSES the whole block via codec::parse (SIMD JSON / native MsgPack) on the worker pool and hands process_parsed a ParsedBatch (records + aligned parsed payloads + shared FieldInterner). This keeps the batch-parse + interner throughput win for apps that opt in.

Records that fail to parse are handled per the configured ParseErrorAction (Dlq -> dlq_entries, Skip -> drop+counted, FailBatch -> terminal no-commit) – see ParsedBatch for the parse-failure contract. process_parsed returns the final WorkBatch and MUST preserve the input commit_tokens.

Errors

Same as run_workbatch.

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 set_byte_budget(&mut self, budget: Arc<ByteBudgetController>)

Available on crate feature governor only.

Wire the self-regulation byte-budget lever (governor feature).

Called by ServiceRuntime::build() when the governor is enabled. Once wired, run_governed streams the input in budget-sized sub-blocks and drives the AIMD loop per block. Without it (governor off), run_governed falls back to the whole-batch loop.

pub fn is_self_regulated(&self) -> bool

Available on crate feature governor only.

Whether the self-regulation byte-budget lever is wired (governor feature). When false, run_governed is the whole-batch loop.

pub fn with_filter_dlq_policy(self, policy: FilterDlqPolicy) -> Self

Available on crate feature transport only.

Set the policy for inbound-filter DLQ entries in the run loops.

Default is FilterDlqPolicy::Reject – the run loop errors if an inbound action: dlq filter produces entries and no routing is set, so dead-letters are never silently dropped.

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: &[Record], transform: F, ) -> Vec<Result<O, E>>

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

Available on crate feature transport only.

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: &[Record], transform: F, ) -> Vec<Result<O, E>>

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

Available on crate feature transport only.

Pre-route and transform a batch of records without parsing.

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

Trait Implementations

impl Debug for BatchEngine

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

Formats the value using the given formatter.