Function coord_delete

nodedb-cluster/src/distributed_array/coordinator/write.rs:127–175

Partition a flat coord list by Hilbert tile and fan delete requests to owning shards. `coords` — each element is `(hilbert_prefix, zerompk-encoded single-coord bytes)`. `prefix_bits` — routing granularity (1–16). `wal_lsn` — WAL sequence number allocated by the Control Plane. Atomicity is per-shard only (same contract as `coord_put`).

(
    params: &ArrayWriteCoordParams,
    array_id_msgpack: Vec<u8>,
    prefix_bits: u8,
    wal_lsn: u64,
    coords: &[(u64, Vec<u8>)],
    dispatch: &Arc<dyn ShardRpcDispatch>,
    circuit_breaker

///
/// Atomicity is per-shard only (same contract as `coord_put`).
pub async fn coord_delete(
    params: &ArrayWriteCoordParams,
    array_id_msgpack: Vec<u8>,
    prefix_bits: u8,
    wal_lsn: u64,
    coords: &[(u64, Vec<u8>)],
    dispatch: &Arc<dyn ShardRpcDispatch>,
    circuit_breaker: &Arc<CircuitBreaker>,
) -> Result<Vec<ArrayShardDeleteResp>> {
    if coords.is_empty() {
        return Ok(Vec::new());
    }

    let buckets = partition_delete_coords(coords, prefix_bits)?;

    let fo_params = FanOutPartitionedParams {
        timeout_ms: params.timeout_ms,
        source_node: params.source_node,
    };

    let encoded: Result<Vec<(u32, Vec<u8>)>> = buckets
        .into_iter()
        .map(|b| {
            let req = ArrayShardDeleteReq {
                array_id_msgpack: array_id_msgpack.clone(),
                coords_msgpack: b.coords_msgpack,
                wal_lsn,
                representative_hilbert_prefix: b.representative_hilbert_prefix,
                prefix_bits,
            };
            zerompk::to_msgpack_vec(&req)
                .map(|bytes| (b.vshard_id, bytes))
                .map_err(|e| ClusterError::Codec {
                    detail: format!("ArrayShardDeleteReq serialise (shard {}): {e}", b.vshard_id),
                })
        })
        .collect();

    let raw = fan_out_partitioned(
        &fo_params,
        super::super::opcodes::ARRAY_SHARD_DELETE_REQ,
        &encoded?,
        dispatch,
        circuit_breaker,
    )
    .await?;

    decode_resps::<ArrayShardDeleteResp>(&raw)
}