Function spawn_wal_catchup_task

nodedb/src/control/wal_catchup.rs:37–68

Spawn the WAL catch-up background task. Runs on the Tokio runtime (Control Plane). Periodically reads unflushed WAL TimeseriesBatch records and dispatches them to the Data Plane. `initial_lsn` should be `wal.next_lsn()` after startup WAL replay — everything before that has already been replayed.

(
    shared: Arc<SharedState>,
    initial_lsn: Lsn,
    mut shutdown: tokio::sync::watch::Receiver<bool>,
)

/// `initial_lsn` should be `wal.next_lsn()` after startup WAL replay —
/// everything before that has already been replayed.
pub fn spawn_wal_catchup_task(
    shared: Arc<SharedState>,
    initial_lsn: Lsn,
    mut shutdown: tokio::sync::watch::Receiver<bool>,
) {
    shared
        .wal_catchup_lsn
        .store(initial_lsn.as_u64(), Ordering::Release);

    tokio::spawn(async move {
        // Adaptive interval: 500ms default, tighten when catching up, relax when idle.
        let mut interval_ms: u64 = 500;

        loop {
            tokio::select! {
                _ = tokio::time::sleep(std::time::Duration::from_millis(interval_ms)) => {
                    let result = run_catchup_cycle(&shared).await;
                    interval_ms = match result {
                        CatchupResult::HasMore => 100,      // rapid drain
                        CatchupResult::Dispatched => 250,   // active, normal pace
                        CatchupResult::Backpressured => 200, // retry soon
                        CatchupResult::Idle => 2000,        // nothing to do
                    };
                }
                _ = shutdown.changed() => {
                    info!("WAL catch-up task shutting down");
                    break;
                }
            }
        }
    });
}

/// Result of a single catch-up cycle.
enum CatchupResult {