Method execute_checkpoint

nodedb/src/data/executor/handlers/control/snapshot.rs:244–288

Execute a coordinated checkpoint: flush all engine state to disk and return this core's checkpoint LSN. 1. Checkpoint vector indexes (HNSW segments → disk files). 2. Export CRDT snapshots (Loro docs → disk files). 3. redb sparse engine is already ACID — no action needed. 4. CSR is rebuilt from redb edge store on startup — no action needed. 5. Return the core's watermark LSN as the checkpoint poin

(
        &mut self,
        task: &ExecutionTask,
    )

    /// 4. CSR is rebuilt from redb edge store on startup — no action needed.
    /// 5. Return the core's watermark LSN as the checkpoint point.
    pub(in crate::data::executor) fn execute_checkpoint(
        &mut self,
        task: &ExecutionTask,
    ) -> Response {
        let checkpoint_lsn = self.watermark.as_u64();

        // 1. Flush vector indexes to disk.
        let vectors_checkpointed = self.checkpoint_vector_indexes();

        // 2. Flush CRDT snapshots to disk.
        let crdts_checkpointed = self.checkpoint_crdt_engines();

        // 3. Flush spatial R-tree indexes to disk.
        let spatial_checkpointed = self.checkpoint_spatial_indexes();

        // 4. Compact CSR write buffers into dense arrays for clean state.
        if let Err(e) = self.csr.compact_all() {
            tracing::warn!(error = %e, "CSR compaction rejected by memory governor during snapshot; skipping");
        }

        // 5. Record completed flushes in the checkpoint coordinator
        //    and advance the checkpoint LSN for WAL truncation safety.
        self.checkpoint_coordinator
            .record_flush("vector", vectors_checkpointed);
        self.checkpoint_coordinator
            .record_flush("crdt", crdts_checkpointed);
        self.checkpoint_coordinator
            .record_flush("spatial", spatial_checkpointed);
        self.checkpoint_coordinator
            .complete_checkpoint(checkpoint_lsn);

        info!(
            core = self.core_id,
            checkpoint_lsn,
            vectors_checkpointed,
            crdts_checkpointed,
            spatial_checkpointed,
            dirty_pages = self.checkpoint_coordinator.total_dirty_pages(),
            "core checkpoint complete"
        );

        // Return the checkpoint LSN as the response payload.
        let payload = checkpoint_lsn.to_le_bytes().to_vec();
        self.response_with_payload(task, payload)
    }

    /// Checkpoint all CRDT tenant engines to disk.
    ///