Function apply_array_schema

nodedb/src/control/array_sync/raft_apply.rs:320–409

Apply a committed `ArraySchema` entry on the local node. 1. Imports the Loro snapshot into the local `OriginSchemaRegistry`. 2. Decodes the `ArraySchema` and registers an `ArrayCatalogEntry` so the Data Plane can open the array when a subsequent `ArrayOp` arrives. This is the canonical DDL propagation path for followers: the Raft `ArraySchema` entry is the single source of truth — no out-of-band

(
    state: &Arc<SharedState>,
    tracker: &Arc<ProposeTracker>,
    group_id: u64,
    log_index: u64,
    applied_key: u64,
    payload: ArraySchemaPayload<'_>,
)

///    `ArraySchema` entry is the single source of truth — no out-of-band
///    catalog registration is needed.
pub(crate) fn apply_array_schema(
    state: &Arc<SharedState>,
    tracker: &Arc<ProposeTracker>,
    group_id: u64,
    log_index: u64,
    applied_key: u64,
    payload: ArraySchemaPayload<'_>,
) {
    use nodedb_array::sync::hlc::Hlc;
    use nodedb_array::types::ArrayId;
    use nodedb_types::TenantId as NdTenantId;

    use crate::control::array_catalog::entry::ArrayCatalogEntry;

    let ArraySchemaPayload {
        array,
        snapshot_payload,
        schema_hlc_bytes,
    } = payload;
    let remote_hlc = Hlc::from_bytes(&schema_hlc_bytes);

    // Use the replicated import path so every replica converges to the same
    // schema_hlc (the one committed in the Raft log entry) rather than each
    // bumping independently via their local HLC generator.
    if let Err(e) =
        state
            .array_sync_schemas
            .import_snapshot_replicated(array, snapshot_payload, remote_hlc)
    {
        warn!(
            group_id, index = log_index, array = %array, error = %e,
            "apply_array_schema: import_snapshot_replicated failed"
        );
        tracker.complete(
            group_id,
            log_index,
            applied_key,
            Err(crate::Error::Internal {
                detail: format!("schema import: {e}"),
            }),
        );
        return;
    }

    // Decode the ArraySchema from the just-imported Loro document and register
    // it in the array catalog so the Data Plane can open the array on this node.
    match state.array_sync_schemas.to_array_schema(array) {
        Some(schema) => match zerompk::to_msgpack_vec(&schema) {
            Ok(schema_msgpack) => {
                let array_id = ArrayId::new(NdTenantId::new(0), array);
                let entry = ArrayCatalogEntry {
                    array_id,
                    name: array.to_string(),
                    schema_msgpack,
                    schema_hash: 0,
                    created_at_ms: 0,
                    prefix_bits: 8,
                    audit_retain_ms: None,