Function process_normal_batch

nodedb/src/event/consumer.rs:358–394

Process a batch of Normal-mode events: trigger batching, CDC, permission cache, streaming MVs, CRDT sync. Statement-level trigger dispatch is per-event (no batching).

(
    events: &[super::types::WriteEvent],
    shared_state: &Arc<SharedState>,
    retry_queue: &mut TriggerRetryQueue,
    cdc_router: &Arc<super::cdc::CdcRouter>,
    _slab_account: &Arc<super::sla

/// Process a batch of Normal-mode events: trigger batching, CDC, permission cache,
/// streaming MVs, CRDT sync. Statement-level trigger dispatch is per-event (no batching).
async fn process_normal_batch(
    events: &[super::types::WriteEvent],
    shared_state: &Arc<SharedState>,
    retry_queue: &mut TriggerRetryQueue,
    cdc_router: &Arc<super::cdc::CdcRouter>,
    _slab_account: &Arc<super::slab_budget::ConsumerSlabAccount>,
) {
    let mut trigger_collector =
        crate::control::trigger::batch::collector::TriggerBatchCollector::new(
            crate::control::trigger::batch::BatchConfig::default().batch_size,
        );

    for event in events {
        if !event.op.is_data_event() {
            shared_state
                .watermark_tracker
                .advance_lsn_only(event.vshard_id.as_u32(), event.lsn.as_u64());
            continue;
        }

        // DML audit: record to audit log before dispatching triggers.
        super::audit_dml::audit_dml_event(event, shared_state);

        if let Some(batch) =
            accumulate_data_event(event, shared_state, &mut trigger_collector, cdc_router)
        {
            super::trigger::dispatcher::dispatch_trigger_batch(&batch, shared_state, retry_queue)
                .await;
        }

        super::trigger::dispatcher::dispatch_triggers(event, shared_state, retry_queue).await;
    }

    if let Some(batch) = trigger_collector.flush() {
        super::trigger::dispatcher::dispatch_trigger_batch(&batch, shared_state, retry_queue).await;
    }
}

/// Process the retry queue: DLQ exhausted entries and retry ready ones.
async fn process_retry_queue(