Function accumulate_data_event

nodedb/src/event/consumer_helpers.rs:158–197

Dispatch a data write event (op.is_data_event() == true): advances wall-time watermark, batches triggers, routes CDC, updates permission cache, feeds streaming MVs and CRDT sync. Returns a completed trigger batch if the collector filled on this event.

(
    event: &WriteEvent,
    shared_state: &Arc<SharedState>,
    trigger_collector: &mut crate::control::trigger::batch::collector::TriggerBatchCollector,
    cdc_router: &Arc<super::cdc::CdcRouter>

///
/// Returns a completed trigger batch if the collector filled on this event.
pub fn accumulate_data_event(
    event: &WriteEvent,
    shared_state: &Arc<SharedState>,
    trigger_collector: &mut crate::control::trigger::batch::collector::TriggerBatchCollector,
    cdc_router: &Arc<super::cdc::CdcRouter>,
) -> Option<crate::control::trigger::batch::collector::TriggerBatch> {
    let event_time_ms = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis() as u64;
    shared_state.watermark_tracker.advance(
        event.vshard_id.as_u32(),
        event.lsn.as_u64(),
        event_time_ms,
    );

    let batch =
        crate::control::trigger::batch::collector::push_write_event(trigger_collector, event);

    cdc_router.route_event(event, &shared_state.watermark_tracker);
    crate::control::security::permission_tree::event_handler::handle_permission_event(
        event,
        &shared_state.permission_cache,
    );
    let matching_streams = shared_state
        .stream_registry
        .find_matching(event.tenant_id.as_u64(), &event.collection);
    for stream_def in &matching_streams {
        super::streaming_mv::processor::process_write_event_for_mvs(
            event,
            &shared_state.mv_registry,
            &stream_def.name,
        );
    }
    shared_state
        .delta_packager
        .package_and_enqueue(event, &shared_state.crdt_sync_delivery);

    batch
}