Function ensure_array_open

nodedb/src/control/array_sync/raft_apply.rs:233–303

Ensure the Data Plane has the array open before dispatching Put/Delete. Looks up the catalog entry for `array_id.name`, then dispatches `OpenArray` to the Data Plane. This is idempotent on the Data Plane side: if the array is already open with the same schema hash, the handler returns `Ok`. Returns an error if the catalog entry is missing (the array was never registered on this node) or if the `

(
    state: &Arc<SharedState>,
    array_id: &nodedb_array::types::ArrayId,
    vshard: crate::types::VShardId,
    tenant_id: crate::types::TenantId,
)

/// Returns an error if the catalog entry is missing (the array was never
/// registered on this node) or if the `OpenArray` dispatch fails.
async fn ensure_array_open(
    state: &Arc<SharedState>,
    array_id: &nodedb_array::types::ArrayId,
    vshard: crate::types::VShardId,
    tenant_id: crate::types::TenantId,
) -> crate::Result<()> {
    let (schema_msgpack, schema_hash, prefix_bits) = {
        let cat = state
            .array_catalog
            .read()
            .unwrap_or_else(|p| p.into_inner());
        match cat.lookup_by_name(&array_id.name) {
            Some(entry) => (
                entry.schema_msgpack.clone(),
                entry.schema_hash,
                entry.prefix_bits,
            ),
            None => {
                return Err(crate::Error::Internal {
                    detail: format!(
                        "ensure_array_open: array '{}' not in catalog — register it before applying ops",
                        array_id.name
                    ),
                });
            }
        }
    };

    let open_request_id = state.next_request_id();
    let open_plan = crate::bridge::envelope::PhysicalPlan::Array(
        nodedb_physical::physical_plan::ArrayOp::OpenArray {
            array_id: array_id.clone(),
            schema_msgpack,
            schema_hash,
            prefix_bits,
        },
    );
    let open_request = Request {
        request_id: open_request_id,
        tenant_id,
        database_id: DatabaseId::DEFAULT,
        vshard_id: vshard,
        plan: open_plan,
        deadline: std::time::Instant::now() + Duration::from_secs(30),
        priority: Priority::Normal,
        trace_id: TraceId::generate(),
        consistency: ReadConsistency::Strong,
        idempotency_key: None,
        event_source: crate::event::EventSource::CrdtSync,
        user_roles: Vec::new(),
        user_id: None,
        statement_digest: None,
    };

    let mut open_rx = state.tracker.register(open_request_id);

    let dispatch_result = match state.dispatcher.lock() {
        Ok(mut d) => d.dispatch(open_request),