nodedb 0.2.0

// SPDX-License-Identifier: BUSL-1.1

//! Event Plane consumer: one Tokio task per Data Plane core ring buffer.
//!
//! Each consumer operates in one of two modes:
//!
//! ```text
//! Normal ──[sequence gap detected]──► WalCatchup
//!   ▲                                    │
//!   └──[caught up to WAL head]───────────┘
//! ```
//!
//! - **Normal**: polls ring buffer, processes events, persists watermark.
//! - **WalCatchup**: pauses ring buffer entirely, reads events exclusively
//!   from WAL on disk until caught up, then switches back. Ring buffer and
//!   WAL are NEVER read simultaneously (prevents "thundering WAL" spiral).

use std::sync::Arc;
use std::time::Duration;

use nodedb_bridge::backpressure::PressureState;
use tokio::sync::watch;
use tracing::{debug, info, trace, warn};

use super::bus::EventConsumerRx;
use super::metrics::CoreMetrics;
use super::trigger::dlq::TriggerDlq;
use super::trigger::retry::TriggerRetryQueue;
use super::watermark::WatermarkStore;
use crate::control::state::SharedState;
use crate::types::Lsn;
use crate::wal::WalManager;

use super::consumer_helpers::{
    accumulate_data_event, dispatch_event, drain_and_skip_stale, drain_ring_buffer,
    flush_watermark, maybe_flush_watermark, record_event,
};

/// Initial sleep when the ring buffer is empty. Adaptive backoff ramps
/// up to `EMPTY_POLL_MAX` after `EMPTY_POLL_RAMP` consecutive empty polls
/// so an idle Event Plane consumer does not wake every 1ms forever.
const EMPTY_POLL_MIN: Duration = Duration::from_millis(1);
/// Cap on the empty-poll sleep. 50ms keeps trigger / CDC dispatch latency
/// bounded for the first event after an idle period while limiting idle
/// CPU to ~20 wakes/sec per core.
const EMPTY_POLL_MAX: Duration = Duration::from_millis(50);
/// After this many consecutive empty polls (~32ms of idleness at 1ms),
/// switch to the long sleep.
const EMPTY_POLL_RAMP: u32 = 32;

/// How often to process the retry queue (check for due retries).
const RETRY_POLL_INTERVAL: Duration = Duration::from_millis(200);

/// Consumer mode state machine.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ConsumerMode {
    /// Reading from ring buffer.
    Normal,
    /// Ring buffer paused; reading from WAL on disk.
    WalCatchup,
}

/// Configuration for spawning a consumer.
pub struct ConsumerConfig {
    pub rx: EventConsumerRx,
    pub shutdown: watch::Receiver<bool>,
    pub wal: Arc<WalManager>,
    pub watermark_store: Arc<WatermarkStore>,
    pub shared_state: Arc<SharedState>,
    pub trigger_dlq: Arc<std::sync::Mutex<TriggerDlq>>,
    pub cdc_router: Arc<super::cdc::CdcRouter>,
    pub num_cores: usize,
    /// Per-consumer slab-pin accounting for WAL memory budget enforcement.
    pub slab_account: Arc<super::slab_budget::ConsumerSlabAccount>,
}

/// Handle to a running consumer task.
pub struct ConsumerHandle {
    pub core_id: usize,
    pub metrics: Arc<CoreMetrics>,
    join_handle: tokio::task::JoinHandle<()>,
}

impl ConsumerHandle {
    pub fn abort(&self) {
        self.join_handle.abort();
    }

    /// Abort the task and await its termination, consuming the handle so the
    /// task future (and every `Arc` it held) is definitely dropped by the
    /// time this returns. Used in shutdown paths that must observe `Drop`
    /// side effects before reopening resources (e.g. redb file locks).
    pub async fn abort_and_join(self) {
        self.join_handle.abort();
        let _ = self.join_handle.await;
    }

    pub fn events_processed(&self) -> u64 {
        use std::sync::atomic::Ordering;
        self.metrics.events_processed.load(Ordering::Relaxed)
    }
}

/// Spawn a consumer Tokio task for one Data Plane core's event ring buffer.
pub fn spawn_consumer(config: ConsumerConfig) -> ConsumerHandle {
    let core_id = config.rx.core_id();
    let metrics = Arc::new(CoreMetrics::new());
    let metrics_clone = Arc::clone(&metrics);

    let join_handle = tokio::spawn(async move {
        consumer_loop(config, metrics_clone).await;
    });

    ConsumerHandle {
        core_id,
        metrics,
        join_handle,
    }
}

/// 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 {
                    empty_polls = 0;
                    dirty_watermark = true;

                    process_normal_batch(
                        &events,
                        &shared_state,
                        &mut retry_queue,
                        &cdc_router,
                        &slab_account,
                    )
                    .await;

                    let batch_payload_bytes: u64 = events
                        .iter()
                        .map(|e| {
                            e.new_value.as_ref().map_or(0, |v| v.len() as u64)
                                + e.old_value.as_ref().map_or(0, |v| v.len() as u64)
                        })
                        .sum();
                    slab_account.add_pinned(batch_payload_bytes);
                    drop(events);
                    slab_account.release_pinned(batch_payload_bytes);

                    trace!(core_id, batch_count, "event batch processed");

                    if slab_account.is_shed() {
                        info!(core_id, "slab budget shed — entering WAL catchup mode");
                        slab_account.reset();
                        slab_account.clear_shed();
                        mode = ConsumerMode::WalCatchup;
                        metrics.record_wal_catchup_enter();
                        continue;
                    }

                    if rx.pressure_state() == PressureState::Suspended {
                        info!(
                            core_id,
                            "backpressure SUSPENDED — entering WAL catchup mode"
                        );
                        mode = ConsumerMode::WalCatchup;
                        metrics.record_wal_catchup_enter();
                        continue;
                    }

                    tokio::task::yield_now().await;
                    continue;
                }

                // No new events — process retry queue if due.
                if !retry_queue.is_empty() && last_retry_poll.elapsed() >= RETRY_POLL_INTERVAL {
                    process_retry_queue(&mut retry_queue, &trigger_dlq, &shared_state).await;
                    last_retry_poll = tokio::time::Instant::now();
                }

                maybe_flush_watermark(
                    &watermark_store,
                    core_id,
                    last_lsn,
                    &mut dirty_watermark,
                    &mut last_watermark_flush,
                );

                empty_polls = empty_polls.saturating_add(1);
                let poll_sleep = if empty_polls < EMPTY_POLL_RAMP {
                    EMPTY_POLL_MIN
                } else {
                    EMPTY_POLL_MAX
                };

                tokio::select! {
                    _ = tokio::time::sleep(poll_sleep) => {}
                    _ = shutdown.changed() => {
                        if dirty_watermark {
                            flush_watermark(&watermark_store, core_id, last_lsn);
                        }
                        debug!(core_id, "event plane consumer received shutdown");
                        break;
                    }
                }
            }

            ConsumerMode::WalCatchup => {
                const MAX_WAL_RETRIES: u32 = 10;

                info!(
                    core_id,
                    from_lsn = last_lsn.as_u64(),
                    "WAL catchup: replaying from WAL"
                );

                match super::wal_replay::replay_wal_mmap(
                    &wal,
                    last_lsn.next(),
                    core_id,
                    num_cores,
                    last_sequence,
                )
                .or_else(|_| {
                    super::wal_replay::replay_wal_to_events(
                        &wal,
                        last_lsn.next(),
                        core_id,
                        num_cores,
                        last_sequence,
                    )
                }) {
                    Ok(events) => {
                        wal_retry_count = 0;
                        let count = events.len() as u64;
                        for event in &events {
                            record_event(core_id, event, &metrics);
                            dispatch_event(event, &shared_state, &mut retry_queue, &cdc_router)
                                .await;
                            last_sequence = event.sequence;
                            if event.lsn.is_ahead_of(last_lsn) {
                                last_lsn = event.lsn;
                            }
                        }
                        if count > 0 {
                            metrics.record_wal_replay(count);
                            info!(
                                core_id,
                                events_replayed = count,
                                new_lsn = last_lsn.as_u64(),
                                "WAL catchup complete"
                            );
                        } else {
                            debug!(core_id, "WAL catchup: no new events");
                        }
                    }
                    Err(e) => {
                        wal_retry_count += 1;
                        if wal_retry_count > MAX_WAL_RETRIES {
                            tracing::error!(
                                core_id,
                                error = %e,
                                retries = MAX_WAL_RETRIES,
                                "WAL catchup failed after max retries, returning to Normal mode"
                            );
                        } else {
                            warn!(
                                core_id,
                                error = %e,
                                retry = wal_retry_count,
                                max_retries = MAX_WAL_RETRIES,
                                "WAL catchup replay failed, retrying after delay"
                            );
                            tokio::time::sleep(Duration::from_millis(100)).await;
                            continue;
                        }
                    }
                }

                drain_and_skip_stale(&mut rx, last_sequence);

                flush_watermark(&watermark_store, core_id, last_lsn);
                dirty_watermark = false;
                last_watermark_flush = tokio::time::Instant::now();

                mode = ConsumerMode::Normal;
                info!(core_id, "returned to Normal mode");
            }
        }
    }

    let processed = {
        use std::sync::atomic::Ordering;
        metrics.events_processed.load(Ordering::Relaxed)
    };
    debug!(
        core_id,
        total_processed = processed,
        "event plane consumer stopped"
    );
}

/// 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(
    retry_queue: &mut TriggerRetryQueue,
    trigger_dlq: &Arc<std::sync::Mutex<TriggerDlq>>,
    shared_state: &Arc<SharedState>,
) {
    let (ready, exhausted) = retry_queue.drain_due();
    if !exhausted.is_empty() {
        let mut dlq = trigger_dlq.lock().unwrap_or_else(|p| p.into_inner());
        for entry in &exhausted {
            let _ = dlq.enqueue(super::trigger::dlq::DlqEnqueueParams {
                tenant_id: entry.tenant_id,
                source_collection: entry.collection.clone(),
                row_id: entry.row_id.clone(),
                operation: entry.operation.clone(),
                trigger_name: entry.trigger_name.clone(),
                error: entry.last_error.clone(),
                retry_count: entry.attempts,
                source_lsn: entry.source_lsn,
                source_sequence: entry.source_sequence,
            });
        }
        // dlq MutexGuard dropped before any await.
    }

    for entry in ready {
        super::trigger::dispatcher::retry_single(&entry, shared_state, retry_queue).await;
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::event::bus::create_event_bus_with_capacity;
    use crate::event::consumer_helpers::detect_sequence_gap;
    use crate::event::types::{EventSource, RowId, WriteOp};
    use crate::types::{TenantId, VShardId};

    fn make_event(seq: u64) -> super::super::types::WriteEvent {
        super::super::types::WriteEvent {
            sequence: seq,
            collection: Arc::from("test"),
            op: WriteOp::Insert,
            row_id: RowId::new("row-1"),
            lsn: Lsn::new(seq * 10),
            tenant_id: TenantId::new(1),
            vshard_id: VShardId::new(0),
            source: EventSource::User,
            new_value: Some(Arc::from(b"data".as_slice())),
            old_value: None,
            system_time_ms: None,
            valid_time_ms: None,
            user_id: None,
            statement_digest: None,
        }
    }

    #[test]
    fn gap_detection() {
        let metrics = CoreMetrics::new();
        let e1 = make_event(1);
        let e5 = make_event(5);

        record_event(0, &e1, &metrics);
        detect_sequence_gap(0, &e5, 1, &metrics);
        record_event(0, &e5, &metrics);

        use std::sync::atomic::Ordering;
        assert_eq!(metrics.events_processed.load(Ordering::Relaxed), 2);
        assert_eq!(metrics.events_dropped.load(Ordering::Relaxed), 3);
    }

    #[tokio::test]
    async fn consumer_processes_and_persists_watermark() {
        let (mut producers, consumers) = create_event_bus_with_capacity(1, 64);
        let dir = tempfile::tempdir().unwrap();
        let (wal, watermark_store, shared_state, trigger_dlq, cdc_router) =
            crate::event::test_utils::event_test_deps(&dir);

        let (shutdown_tx, shutdown_rx) = watch::channel(false);

        // Emit events.
        for i in 1..=5 {
            producers[0].emit(make_event(i));
        }

        let handle = spawn_consumer(ConsumerConfig {
            rx: consumers.into_iter().next().unwrap(),
            shutdown: shutdown_rx,
            wal,
            watermark_store: Arc::clone(&watermark_store),
            shared_state,
            trigger_dlq,
            cdc_router,
            num_cores: 1,
            slab_account: Arc::new(crate::event::slab_budget::ConsumerSlabAccount::new(0)),
        });

        // Let consumer process.
        tokio::time::sleep(Duration::from_millis(50)).await;
        assert_eq!(handle.events_processed(), 5);

        // Shutdown (triggers final watermark flush).
        shutdown_tx.send(true).ok();
        tokio::time::sleep(Duration::from_millis(50)).await;

        // Verify watermark was persisted.
        let wm = watermark_store.load(0).unwrap();
        assert_eq!(wm, Lsn::new(50)); // seq 5 → lsn = 5*10 = 50
    }
}