Function consumer_loop

nodedb/src/event/consumer.rs:122–354

The main consumer loop.

(config: ConsumerConfig, metrics: Arc<CoreMetrics>)

/// The main consumer loop.
async fn consumer_loop(config: ConsumerConfig, metrics: Arc<CoreMetrics>) {
    let ConsumerConfig {
        mut rx,
        mut shutdown,
        wal,
        watermark_store,
        shared_state,
        trigger_dlq,
        cdc_router,
        num_cores,
        slab_account,
    } = config;

    let core_id = rx.core_id();
    let mut mode = ConsumerMode::Normal;
    let mut last_sequence: u64 = 0;
    let mut last_lsn = Lsn::ZERO;
    let mut dirty_watermark = false;
    let mut last_watermark_flush = tokio::time::Instant::now();
    let mut retry_queue = TriggerRetryQueue::new();
    let mut last_retry_poll = tokio::time::Instant::now();

    // Load persisted watermark.
    match watermark_store.load(core_id) {
        Ok(lsn) => {
            last_lsn = lsn;
            debug!(core_id, lsn = lsn.as_u64(), "loaded watermark");
        }
        Err(e) => {
            warn!(core_id, error = %e, "failed to load watermark, starting from ZERO");
        }
    }

    debug!(core_id, "event plane consumer started");
    let mut wal_retry_count: u32 = 0;
    let mut empty_polls: u32 = 0;

    loop {
        if *shutdown.borrow() {
            if dirty_watermark {
                flush_watermark(&watermark_store, core_id, last_lsn);
            }
            debug!(core_id, "event plane consumer shutting down");
            break;
        }

        match mode {
            ConsumerMode::Normal => {
                let events = drain_ring_buffer(
                    &mut rx,
                    &metrics,
                    core_id,
                    &mut last_sequence,
                    &mut last_lsn,
                );
                let batch_count = events.len();

                if batch_count > 0 {