Function dispatch_route

nodedb/src/control/gateway/dispatcher.rs:39–84

Dispatch a single route and return the raw payload bytes. `tenant_id` — the authenticated tenant for this query. `trace_id` — distributed trace ID propagated from the client request. `deadline_ms` — remaining deadline in milliseconds. `version_set` — descriptor versions for the collections touched by the plan.

(
    route: TaskRoute,
    shared: &Arc<SharedState>,
    tenant_id: TenantId,
    database_id: DatabaseId,
    trace_id: TraceId,
    deadline_ms: u64,
    version_set: &GatewayVersionSet,
)

/// `deadline_ms` — remaining deadline in milliseconds.
/// `version_set` — descriptor versions for the collections touched by the plan.
pub async fn dispatch_route(
    route: TaskRoute,
    shared: &Arc<SharedState>,
    tenant_id: TenantId,
    database_id: DatabaseId,
    trace_id: TraceId,
    deadline_ms: u64,
    version_set: &GatewayVersionSet,
) -> Result<Vec<Vec<u8>>, Error> {
    match route.decision {
        RouteDecision::Local => dispatch_local(route, shared, tenant_id, trace_id).await,
        RouteDecision::Remote { node_id, vshard_id } => {
            dispatch_remote(RemoteDispatchArgs {
                plan: route.plan,
                shared,
                node_id,
                vshard_id,
                tenant_id,
                database_id,
                trace_id,
                deadline_ms,
                version_set,
            })
            .await
        }
        RouteDecision::Broadcast { .. } => {
            // Broadcast routes are split into individual Local/Remote routes
            // by the router before dispatch. This arm should not be reached.
            Err(Error::Internal {
                detail: "dispatcher: Broadcast route reached dispatch — should have been split"
                    .into(),
            })
        }
        RouteDecision::LeaderUnknown { vshard_id } => {
            // Cluster mode with no leader currently known for this vShard.
            // Surface as NotLeader so the gateway retry loop sleeps and
            // re-resolves the routing table on the next attempt — never
            // silently serve from a possibly-stale local replica.
            Err(Error::NotLeader {
                vshard_id: VShardId::new(vshard_id as u32),
                leader_node: 0,
                leader_addr: String::new(),
            })
        }
    }
}

/// Local dispatch via SPSC bridge.
async fn dispatch_local(