Function fan_out_partitioned

nodedb-cluster/src/distributed_array/scatter.rs:168–218

Send a distinct payload to each shard and collect responses. `per_shard` — `(vshard_id, payload_bytes)` pairs, one per target shard. Returns `(shard_id, response_payload_bytes)` in arrival order.

(
    params: &FanOutPartitionedParams,
    opcode: u32,
    per_shard: &[(u32, Vec<u8>)],
    dispatch: &Arc<dyn ShardRpcDispatch>,
    circuit_breaker: &CircuitBreaker,
)

/// `per_shard` — `(vshard_id, payload_bytes)` pairs, one per target shard.
/// Returns `(shard_id, response_payload_bytes)` in arrival order.
pub async fn fan_out_partitioned(
    params: &FanOutPartitionedParams,
    opcode: u32,
    per_shard: &[(u32, Vec<u8>)],
    dispatch: &Arc<dyn ShardRpcDispatch>,
    circuit_breaker: &CircuitBreaker,
) -> Result<Vec<(u32, Vec<u8>)>> {
    if per_shard.is_empty() {
        return Ok(Vec::new());
    }

    let mut futs = futures::stream::FuturesUnordered::new();

    for (shard_id, payload) in per_shard {
        circuit_breaker.check(*shard_id as u64)?;

        let env = VShardEnvelope::new(
            msg_type_from_opcode(opcode)?,
            params.source_node,
            0,
            *shard_id,
            payload.clone(),
        );
        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(per_shard.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)
}

/// Map an opcode constant to a `VShardMessageType`.
///