Function fan_out

nodedb-cluster/src/distributed_array/scatter.rs:109–162

Send `req_bytes` to every shard listed in `params` and collect responses. Each shard RPC runs concurrently via `FuturesUnordered`. Per-shard timeouts are enforced by `tokio::time::timeout`. The circuit breaker is checked before each call and updated on success/failure. Any shard error causes the whole fan-out to return `Err` — the coordinator decides whether to retry. Returns `Vec<(shard_id, res

(
    params: &FanOutParams,
    opcode: u32,
    req_bytes: &[u8],
    dispatch: &Arc<dyn ShardRpcDispatch>,
    circuit_breaker: &CircuitBreaker,
)

///
/// Returns `Vec<(shard_id, response_payload_bytes)>` in arrival order.
pub async fn fan_out(
    params: &FanOutParams,
    opcode: u32,
    req_bytes: &[u8],
    dispatch: &Arc<dyn ShardRpcDispatch>,
    circuit_breaker: &CircuitBreaker,
) -> Result<Vec<(u32, Vec<u8>)>> {
    if params.shard_ids.is_empty() {
        return Ok(Vec::new());
    }

    // Build one future per shard and collect via FuturesUnordered for
    // true concurrency (no sequential .await loop).
    let mut futs = futures::stream::FuturesUnordered::new();

    for &shard_id in &params.shard_ids {
        // Circuit-breaker gate: treat shard_id as the peer identifier.
        circuit_breaker.check(shard_id as u64)?;

        let env = VShardEnvelope::new(
            msg_type_from_opcode(opcode)?,
            params.source_node,
            0, // target_node resolved by the dispatch impl
            shard_id,
            req_bytes.to_vec(),
        );
        let timeout_ms = params.timeout_ms;
        let dispatch = Arc::clone(dispatch);
        let cb_shard = shard_id;

        futs.push(async move {
            match call_with_wrong_owner_retry(&dispatch, env, timeout_ms).await {
                Ok(resp) => Ok((cb_shard, resp.payload)),
                Err(e) => Err((cb_shard, e)),
            }
        });
    }

    let mut results = Vec::with_capacity(params.shard_ids.len());
    while let Some(outcome) = futs.next().await {
        match outcome {
            Ok((shard_id, payload)) => {
                circuit_breaker.record_success(shard_id as u64);
                results.push((shard_id, payload));
            }
            Err((shard_id, e)) => {
                circuit_breaker.record_failure(shard_id as u64);
                return Err(e);
            }
        }
    }

    Ok(results)
}

/// Send a distinct payload to each shard and collect responses.
///