Function run_catchup_cycle

nodedb/src/control/wal_catchup.rs:86–182

Run one catch-up cycle: read new WAL records, dispatch timeseries batches. Uses paginated mmap replay to bound memory. Passes WAL LSNs to the Data Plane for deduplication (records already ingested are skipped).

(shared: &SharedState)

/// Uses paginated mmap replay to bound memory. Passes WAL LSNs to the
/// Data Plane for deduplication (records already ingested are skipped).
async fn run_catchup_cycle(shared: &SharedState) -> CatchupResult {
    // Backpressure gate: don't compete with live ingest for SPSC slots.
    if shared.max_spsc_utilization() > 50 {
        return CatchupResult::Backpressured;
    }

    let catchup_lsn = shared.wal_catchup_lsn.load(Ordering::Acquire);

    // Read at most PAGE_SIZE WAL records via sequential I/O (bounded memory).
    // Uses sequential I/O (not mmap) to safely read the active segment
    // even when written via O_DIRECT which bypasses the page cache.
    let (records, has_more) = match shared
        .wal
        .replay_from_limit(Lsn::new(catchup_lsn + 1), PAGE_SIZE)
    {
        Ok(r) => r,
        Err(e) => {
            debug!(error = %e, lsn = catchup_lsn, "WAL catch-up replay failed");
            return CatchupResult::Idle;
        }
    };

    if records.is_empty() {
        return CatchupResult::Idle;
    }

    let mut dispatched = 0usize;
    let mut max_lsn = catchup_lsn;

    for record in &records {
        // Only process TimeseriesBatch records.
        let record_type = nodedb_wal::record::RecordType::from_raw(record.logical_record_type());
        if record_type != Some(nodedb_wal::record::RecordType::TimeseriesBatch) {
            max_lsn = max_lsn.max(record.header.lsn);
            continue;
        }

        // Deserialize WAL payload: (collection, raw_ilp_bytes).
        let Ok((collection, payload)): Result<(String, Vec<u8>), _> =
            zerompk::from_msgpack(&record.payload)
        else {
            max_lsn = max_lsn.max(record.header.lsn);
            continue;
        };

        let tenant_id = TenantId::new(record.header.tenant_id);
        let vshard_id = VShardId::new(record.header.vshard_id);

        let plan = PhysicalPlan::Timeseries(TimeseriesOp::Ingest {
            collection,
            payload,
            format: "ilp".to_string(),
            wal_lsn: Some(record.header.lsn),
            // Re-derived on the engine side during apply (record carries
            // raw ILP — row identities are reconstructed from the wire).
            surrogates: Vec::new(),
        });