Method spawn

nodedb/src/event/plane.rs:51–230

Spawn the Event Plane: one consumer Tokio task per Data Plane core. On startup, each consumer loads its persisted watermark and replays WAL entries from that point forward. `consumers_rx` must have exactly one entry per core, in core-ID order. `shutdown` is the node-wide [`ShutdownWatch`] from `SharedState`. All Event Plane subsystems subscribe to this watch instead of a private channel, so the

(
        consumers_rx: Vec<EventConsumerRx>,
        wal: Arc<WalManager>,
        watermark_store: Arc<WatermarkStore>,
        shared_state: Arc<SharedState>,
        trigger_dlq: Arc<std::sync::Mu

    /// private channel, so the unified shutdown bus controls all drain
    /// signalling.
    pub fn spawn(
        consumers_rx: Vec<EventConsumerRx>,
        wal: Arc<WalManager>,
        watermark_store: Arc<WatermarkStore>,
        shared_state: Arc<SharedState>,
        trigger_dlq: Arc<std::sync::Mutex<TriggerDlq>>,
        cdc_router: Arc<CdcRouter>,
        shutdown: Arc<ShutdownWatch>,
    ) -> Self {
        let num_cores = consumers_rx.len();

        let slab_budget = Arc::new(super::slab_budget::SlabBudget::for_cores(num_cores));
        let mut slab_accounts: Vec<Arc<super::slab_budget::ConsumerSlabAccount>> = Vec::new();

        let consumers: Vec<ConsumerHandle> = consumers_rx
            .into_iter()
            .enumerate()
            .map(|(i, rx)| {
                let account = Arc::new(super::slab_budget::ConsumerSlabAccount::new(i));
                slab_accounts.push(Arc::clone(&account));
                spawn_consumer(ConsumerConfig {
                    rx,
                    shutdown: shutdown.raw_receiver(),
                    wal: Arc::clone(&wal),
                    watermark_store: Arc::clone(&watermark_store),
                    shared_state: Arc::clone(&shared_state),
                    trigger_dlq: Arc::clone(&trigger_dlq),
                    cdc_router: Arc::clone(&cdc_router),
                    num_cores,
                    slab_account: account,
                })
            })
            .collect();

        // Spawn periodic slab budget enforcement (every 5s).
        {
            let budget = Arc::clone(&slab_budget);
            let accounts = slab_accounts.clone();
            let mut shutdown_rx = shutdown.raw_receiver();
            let slab_budget_handle = tokio::spawn(async move {
                loop {
                    tokio::select! {
                        _ = tokio::time::sleep(std::time::Duration::from_secs(5)) => {
                            let refs: Vec<&super::slab_budget::ConsumerSlabAccount> =
                                accounts.iter().map(|a| a.as_ref()).collect();
                            budget.check_and_shed(&refs);
                        }
                        _ = shutdown_rx.changed() => {
                            if *shutdown_rx.borrow() { return; }
                        }
                    }
                }
            });
            let _ = shared_state.loop_registry.register(
                "event_plane::slab_budget",
                crate::control::shutdown::LoopHandle::Async(slab_budget_handle),
            );
        }