Function broadcast_raw

nodedb/src/control/server/broadcast.rs:264–330

Broadcast a plan to all cores and return raw binary payloads concatenated. Unlike `broadcast_to_all_cores` (which merges as JSON), this returns the raw response bytes. Each core's payload is appended as-is. Used by the two-phase join: phase 1 scans the right collection across all cores and collects raw msgpack, phase 2 passes it as `broadcast_data` to the join.

(
    shared: &SharedState,
    tenant_id: TenantId,
    plan: PhysicalPlan,
    trace_id: TraceId,
)

/// two-phase join: phase 1 scans the right collection across all cores and
/// collects raw msgpack, phase 2 passes it as `broadcast_data` to the join.
pub async fn broadcast_raw(
    shared: &SharedState,
    tenant_id: TenantId,
    plan: PhysicalPlan,
    trace_id: TraceId,
) -> crate::Result<Vec<u8>> {
    BROADCAST_CALLS.fetch_add(1, Ordering::Relaxed);
    let num_cores = shared
        .dispatcher
        .lock()
        .unwrap_or_else(|p| p.into_inner())
        .num_cores();

    let mut receivers = Vec::with_capacity(num_cores);
    for core_id in 0..num_cores {
        let request_id = shared.next_request_id();
        let vshard_id = VShardId::new(core_id as u32);
        let request = Request {
            request_id,
            tenant_id,
            database_id: DatabaseId::DEFAULT,
            vshard_id,
            plan: plan.clone(),
            deadline: std::time::Instant::now()
                + std::time::Duration::from_secs(shared.tuning.network.default_deadline_secs),
            priority: crate::bridge::envelope::Priority::Normal,
            trace_id,
            consistency: ReadConsistency::Strong,
            idempotency_key: None,
            event_source: crate::event::EventSource::User,
            user_roles: Vec::new(),
            user_id: None,
            statement_digest: None,
        };

        let rx = shared.tracker.register(request_id);
        shared
            .dispatcher
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .dispatch_to_core(core_id, request)?;
        receivers.push(rx);
    }

    let mut merged = Vec::new();
    for mut rx in receivers {
        let resp = tokio::time::timeout(
            std::time::Duration::from_secs(shared.tuning.network.default_deadline_secs),
            async { rx.recv().await.ok_or(()) },
        )
        .await
        .map_err(|_| crate::Error::Dispatch {
            detail: "broadcast_raw timeout".into(),
        })?
        .map_err(|_| crate::Error::Dispatch {
            detail: "broadcast_raw channel closed".into(),
        })?;