Function run_apply_loop

nodedb/src/control/distributed_applier/apply_loop.rs:98–274

Run the background loop that applies committed Raft entries to the local Data Plane. This task reads from the apply channel, deserializes each entry, dispatches the write to the Data Plane via SPSC, and notifies proposers.

(
    mut apply_rx: mpsc::Receiver<ApplyBatch>,
    state: Arc<SharedState>,
    tracker: Arc<ProposeTracker>,
    calvin_read_result_senders: Arc<
        std::sync::Mutex<std::collections::BTreeMap<

/// This task reads from the apply channel, deserializes each entry, dispatches
/// the write to the Data Plane via SPSC, and notifies proposers.
pub async fn run_apply_loop(
    mut apply_rx: mpsc::Receiver<ApplyBatch>,
    state: Arc<SharedState>,
    tracker: Arc<ProposeTracker>,
    calvin_read_result_senders: Arc<
        std::sync::Mutex<std::collections::BTreeMap<u32, mpsc::Sender<ReadResultEvent>>>,
    >,
) {
    while let Some(batch) = apply_rx.recv().await {
        for entry in &batch.entries {
            // Extract idempotency key once at the top so every
            // tracker.complete on this entry can pass it. Returns 0
            // for unparseable / pre-key entries; the tracker treats
            // 0 as "no key" (no mismatch detection).
            let applied_key = ReplicatedEntry::from_bytes(&entry.data)
                .map(|e| e.idempotency_key)
                .unwrap_or(0);

            // ── Array CRDT variants — handled on the Control Plane, bypass Data Plane ──
            if let Some(replicated) = ReplicatedEntry::from_bytes(&entry.data) {
                let target_vshard = replicated.vshard_id;
                match replicated.write {
                    ReplicatedWrite::ArrayOp {
                        ref array,
                        ref op_bytes,
                        ..
                    } => {
                        apply_array_op(
                            &state,
                            &tracker,
                            batch.group_id,
                            entry.index,
                            applied_key,
                            array,
                            op_bytes,
                        )
                        .await;
                        continue;
                    }
                    ReplicatedWrite::ArraySchema {
                        ref array,
                        ref snapshot_payload,
                        schema_hlc_bytes,
                    } => {
                        apply_array_schema(
                            &state,
                            &tracker,
                            batch.group_id,
                            entry.index,
                            applied_key,
                            crate::control::array_sync::raft_apply::ArraySchemaPayload {
                                array,
                                snapshot_payload,
                                schema_hlc_bytes,
                            },
                        );
                        continue;
                    }