Method coord_agg

nodedb-cluster/src/distributed_array/coordinator/read.rs:197–235

Fan out an aggregate request and reduce partial aggregates from all shards. Each shard receives its own `shard_hilbert_range` so it can apply a Hilbert-prefix pre-filter and only count cells in its partition. This prevents double-counting in configurations where multiple vShards share a single Data Plane executor (e.g. single-node harnesses).

(&self, req: ArrayShardAggReq)

    /// prevents double-counting in configurations where multiple vShards share
    /// a single Data Plane executor (e.g. single-node harnesses).
    pub async fn coord_agg(&self, req: ArrayShardAggReq) -> Result<Vec<ArrayAggPartial>> {
        let prefix_bits = self.params.prefix_bits;
        let per_shard: Vec<(u32, Vec<u8>)> = self
            .params
            .shard_ids
            .iter()
            .map(|&shard_id| {
                let hilbert_range = if prefix_bits > 0 {
                    Some(shard_hilbert_range_for_vshard(shard_id, prefix_bits))
                } else {
                    None
                };
                let per_shard_req = ArrayShardAggReq {
                    shard_hilbert_range: hilbert_range,
                    ..req.clone()
                };
                let bytes =
                    zerompk::to_msgpack_vec(&per_shard_req).map_err(|e| ClusterError::Codec {
                        detail: format!("ArrayShardAggReq serialise: {e}"),
                    })?;
                Ok((shard_id, bytes))
            })
            .collect::<Result<Vec<_>>>()?;

        let fo_params = FanOutPartitionedParams {
            source_node: self.params.source_node,
            timeout_ms: self.params.timeout_ms,
        };
        let raw = fan_out_partitioned(
            &fo_params,
            super::super::opcodes::ARRAY_SHARD_AGG_REQ,
            &per_shard,
            &self.dispatch,
            &self.circuit_breaker,
        )
        .await?;
        let resps = decode_resps::<ArrayShardAggResp>(&raw)?;
        Ok(reduce_agg_partials(&resps))
    }

    /// Forward a coord-based delete to the shard(s) that own the cells.
    pub async fn coord_delete(