Function dispatch_task

nodedb/src/control/server/native/dispatch/sql.rs:256–299

Dispatch a single PhysicalTask. Broadcast plans (scans, InsertSelect) are handled locally; all other tasks flow through `dispatch_task_via_gateway` which routes via the gateway when available, or falls back to the local SPSC path on single-node boot.

(ctx: &DispatchCtx<'_>, task: PhysicalTask)

/// flow through `dispatch_task_via_gateway` which routes via the gateway when
/// available, or falls back to the local SPSC path on single-node boot.
async fn dispatch_task(ctx: &DispatchCtx<'_>, task: PhysicalTask) -> crate::Result<Response> {
    if matches!(
        task.plan,
        crate::bridge::envelope::PhysicalPlan::Document(
            nodedb_physical::physical_plan::DocumentOp::InsertSelect { .. }
        )
    ) {
        return broadcast_count_to_all_cores(
            ctx.state,
            task.tenant_id,
            task.plan,
            TraceId::ZERO,
            "inserted",
        )
        .await;
    }

    // `DROP ARRAY` fans out to every core so per-core stores are released.
    if matches!(
        task.plan,
        crate::bridge::envelope::PhysicalPlan::Array(
            nodedb_physical::physical_plan::ArrayOp::DropArray { .. }
        )
    ) {
        return broadcast_count_to_all_cores(
            ctx.state,
            task.tenant_id,
            task.plan,
            TraceId::ZERO,
            "dropped",
        )
        .await;
    }

    // Broadcast scans must fan-out to all cores regardless of gateway state.
    if task.plan.is_broadcast_scan() {
        return broadcast_to_all_cores(ctx.state, task.tenant_id, task.plan, TraceId::ZERO).await;
    }

    // All other tasks — point ops, writes, Raft-replicated writes — route
    // through the gateway when available (cluster-aware routing + retry),
    // or via the local SPSC path when the gateway is not yet wired.
    dispatch_task_via_gateway(ctx, task).await
}

// ─── SET / SHOW / RESET (SQL form) ─────────────────────────────────