nodedb_cluster/distributed_array/coordinator/
read.rs

1// SPDX-License-Identifier: BUSL-1.1
2
3//! Read-path coordinator for distributed array operations.
4//!
5//! [`ArrayCoordinator`] drives fan-out reads (`coord_slice`, `coord_agg`,
6//! `coord_surrogate_bitmap_scan`) to the set of vShards whose Hilbert range
7//! overlaps the slice predicate.
8
9use std::sync::Arc;
10
11use crate::circuit_breaker::CircuitBreaker;
12use crate::error::{ClusterError, Result};
13
14use super::super::merge::{
15    ArrayAggPartial, any_truncated_before_horizon_agg, merge_slice_rows, merge_slice_rows_sorted,
16    reduce_agg_partials,
17};
18use super::super::rpc::ShardRpcDispatch;
19use super::super::scatter::{FanOutParams, FanOutPartitionedParams, fan_out, fan_out_partitioned};
20use super::super::wire::{
21    ArrayShardAggReq, ArrayShardAggResp, ArrayShardDeleteReq, ArrayShardDeleteResp,
22    ArrayShardSliceReq, ArrayShardSliceResp, ArrayShardSurrogateBitmapReq,
23    ArrayShardSurrogateBitmapResp,
24};
25
26/// Parameters common to read-path coordinator entry points (broadcast fan-out).
27pub struct ArrayCoordParams {
28    pub source_node: u64,
29    /// Pre-computed target shard IDs (overlapping shards for reads).
30    pub shard_ids: Vec<u32>,
31    /// Per-shard RPC timeout in milliseconds.
32    pub timeout_ms: u64,
33    /// Hilbert routing granularity (1–16). 0 means no shard-side routing
34    /// validation (e.g. when the coordinator was constructed without slice
35    /// range information, as in tests or unbounded scans).
36    pub prefix_bits: u8,
37    /// Inclusive Hilbert-prefix ranges `(lo, hi)` that this read covers.
38    /// Forwarded to each shard so it can verify it still owns the range.
39    /// Empty means unbounded — the shard skips routing validation.
40    pub slice_hilbert_ranges: Vec<(u64, u64)>,
41}
42
43/// Result of a coordinated slice fan-out.
44///
45/// Carries the merged shard rows together with the OR-reduced
46/// `truncated_before_horizon` flag so the upstream caller can surface a
47/// below-horizon warning to the client. Mirrors the single-node
48/// `ArraySliceResponse` shape so downstream encoding is symmetric.
49#[derive(Debug, Clone, Default)]
50pub struct CoordSliceResult {
51    pub rows: Vec<Vec<u8>>,
52    pub truncated_before_horizon: bool,
53}
54
55/// Result of a coordinated aggregate fan-out.
56///
57/// Carries the merged per-group partials together with the OR-reduced
58/// `truncated_before_horizon` flag, mirroring [`CoordSliceResult`]. Dropping
59/// the flag here would let a below-horizon bitemporal aggregate report
60/// complete results — the same silent-partial-success bug the slice path
61/// avoids.
62#[derive(Debug, Clone, Default)]
63pub struct CoordAggResult {
64    pub partials: Vec<ArrayAggPartial>,
65    pub truncated_before_horizon: bool,
66}
67
68/// Compute the inclusive Hilbert-prefix range `[lo, hi]` that vShard `shard_id`
69/// owns given the array's routing granularity `prefix_bits`.
70///
71/// Each bucket `b = shard_id / stride` covers the Hilbert range
72/// `[b << (64 - prefix_bits), ((b + 1) << (64 - prefix_bits)) - 1]`.
73/// The stride is `VSHARD_COUNT >> prefix_bits` (floored at 1).
74pub(super) fn shard_hilbert_range_for_vshard(shard_id: u32, prefix_bits: u8) -> (u64, u64) {
75    use crate::routing::VSHARD_COUNT;
76    let stride = (VSHARD_COUNT >> (prefix_bits as u32)).max(1);
77    let bucket = shard_id / stride;
78    let shift = 64u8.saturating_sub(prefix_bits);
79    let lo = (bucket as u64) << shift;
80    let hi = if shift == 0 {
81        u64::MAX
82    } else {
83        lo.saturating_add((1u64 << shift).saturating_sub(1))
84    };
85    (lo, hi)
86}
87
88/// Coordinator for distributed array read operations.
89pub struct ArrayCoordinator {
90    pub(super) params: ArrayCoordParams,
91    pub(super) dispatch: Arc<dyn ShardRpcDispatch>,
92    pub(super) circuit_breaker: Arc<CircuitBreaker>,
93}
94
95impl ArrayCoordinator {
96    pub fn new(
97        params: ArrayCoordParams,
98        dispatch: Arc<dyn ShardRpcDispatch>,
99        circuit_breaker: Arc<CircuitBreaker>,
100    ) -> Self {
101        Self {
102            params,
103            dispatch,
104            circuit_breaker,
105        }
106    }
107
108    /// Construct an `ArrayCoordinator` whose target shards are computed from
109    /// the Hilbert-prefix ranges that overlap a slice predicate.
110    ///
111    /// `slice_hilbert_ranges` — `(lo, hi)` pairs computed by the planner from
112    /// the `Slice` predicate. Pass an empty slice for an unbounded scan.
113    /// `prefix_bits` — the array's routing granularity from the catalog entry.
114    /// `total_shards` — the number of active vShards in the cluster.
115    pub fn for_slice(
116        source_node: u64,
117        timeout_ms: u64,
118        slice_hilbert_ranges: &[(u64, u64)],
119        prefix_bits: u8,
120        total_shards: u32,
121        dispatch: Arc<dyn ShardRpcDispatch>,
122        circuit_breaker: Arc<CircuitBreaker>,
123    ) -> crate::error::Result<Self> {
124        let shard_ids = super::super::routing::array_vshards_for_slice(
125            slice_hilbert_ranges,
126            prefix_bits,
127            total_shards,
128        )?;
129        Ok(Self {
130            params: ArrayCoordParams {
131                source_node,
132                shard_ids,
133                timeout_ms,
134                prefix_bits,
135                slice_hilbert_ranges: slice_hilbert_ranges.to_vec(),
136            },
137            dispatch,
138            circuit_breaker,
139        })
140    }
141
142    /// Fan out a coord-range slice to all target shards and merge the rows.
143    ///
144    /// Each shard receives the full slice request with the caller-supplied
145    /// `limit` pushed down so shards can stop scanning early. The coordinator
146    /// stamps a per-shard `shard_hilbert_range` so each shard only returns
147    /// cells whose Hilbert prefix falls within its owned range, preventing
148    /// duplicate rows in single-node harnesses where all vShards share one
149    /// Data Plane. The coordinator applies the same `limit` as a final
150    /// cut-off on the merged result.
151    ///
152    /// When `system_time` is `AllVersions`, the coordinator merge-sorts rows
153    /// from all shards by `ArrayCell::system_time` ascending (k-way merge via
154    /// `merge_slice_rows_sorted`) before applying the limit, to preserve global
155    /// audit ordering. For `Current`/`AsOf`, rows are concatenated in arrival
156    /// order (existing behavior).
157    ///
158    /// Returns merged rows plus the OR-reduced `truncated_before_horizon`
159    /// flag across all shards. If any shard fails the entire operation
160    /// returns `Err` — partial results are not silently dropped.
161    pub async fn coord_slice(
162        &self,
163        req: ArrayShardSliceReq,
164        coordinator_limit: u32,
165        system_time: nodedb_types::SystemTimeScope,
166    ) -> Result<CoordSliceResult> {
167        let prefix_bits = self.params.prefix_bits;
168        let per_shard: Vec<(u32, Vec<u8>)> = self
169            .params
170            .shard_ids
171            .iter()
172            .map(|&shard_id| {
173                let shard_hilbert_range = if prefix_bits > 0 {
174                    Some(shard_hilbert_range_for_vshard(shard_id, prefix_bits))
175                } else {
176                    None
177                };
178                let per_shard_req = ArrayShardSliceReq {
179                    prefix_bits,
180                    slice_hilbert_ranges: self.params.slice_hilbert_ranges.clone(),
181                    shard_hilbert_range,
182                    ..req.clone()
183                };
184                let bytes =
185                    zerompk::to_msgpack_vec(&per_shard_req).map_err(|e| ClusterError::Codec {
186                        detail: format!("ArrayShardSliceReq serialise: {e}"),
187                    })?;
188                Ok((shard_id, bytes))
189            })
190            .collect::<Result<Vec<_>>>()?;
191
192        let fo_params = FanOutPartitionedParams {
193            source_node: self.params.source_node,
194            timeout_ms: self.params.timeout_ms,
195        };
196        let raw = fan_out_partitioned(
197            &fo_params,
198            super::super::opcodes::ARRAY_SHARD_SLICE_REQ,
199            &per_shard,
200            &self.dispatch,
201            &self.circuit_breaker,
202        )
203        .await?;
204        let resps = decode_resps::<ArrayShardSliceResp>(&raw)?;
205        let truncated_before_horizon =
206            super::super::merge::any_truncated_before_horizon_slice(&resps);
207        let rows = if system_time.is_all_versions() {
208            // Audit-log: merge-sort by ArrayCell::system_time ascending across
209            // shards to preserve global ordering, then truncate to limit.
210            merge_slice_rows_sorted(&resps, coordinator_limit)?
211        } else {
212            merge_slice_rows(&resps, coordinator_limit)
213        };
214        Ok(CoordSliceResult {
215            rows,
216            truncated_before_horizon,
217        })
218    }
219
220    /// Fan out an aggregate request and reduce partial aggregates from all shards.
221    ///
222    /// Each shard receives its own `shard_hilbert_range` so it can apply a
223    /// Hilbert-prefix pre-filter and only count cells in its partition. This
224    /// prevents double-counting in configurations where multiple vShards share
225    /// a single Data Plane executor (e.g. single-node harnesses).
226    pub async fn coord_agg(&self, req: ArrayShardAggReq) -> Result<CoordAggResult> {
227        let prefix_bits = self.params.prefix_bits;
228        let per_shard: Vec<(u32, Vec<u8>)> = self
229            .params
230            .shard_ids
231            .iter()
232            .map(|&shard_id| {
233                let hilbert_range = if prefix_bits > 0 {
234                    Some(shard_hilbert_range_for_vshard(shard_id, prefix_bits))
235                } else {
236                    None
237                };
238                let per_shard_req = ArrayShardAggReq {
239                    shard_hilbert_range: hilbert_range,
240                    ..req.clone()
241                };
242                let bytes =
243                    zerompk::to_msgpack_vec(&per_shard_req).map_err(|e| ClusterError::Codec {
244                        detail: format!("ArrayShardAggReq serialise: {e}"),
245                    })?;
246                Ok((shard_id, bytes))
247            })
248            .collect::<Result<Vec<_>>>()?;
249
250        let fo_params = FanOutPartitionedParams {
251            source_node: self.params.source_node,
252            timeout_ms: self.params.timeout_ms,
253        };
254        let raw = fan_out_partitioned(
255            &fo_params,
256            super::super::opcodes::ARRAY_SHARD_AGG_REQ,
257            &per_shard,
258            &self.dispatch,
259            &self.circuit_breaker,
260        )
261        .await?;
262        let resps = decode_resps::<ArrayShardAggResp>(&raw)?;
263        Ok(CoordAggResult {
264            partials: reduce_agg_partials(&resps),
265            truncated_before_horizon: any_truncated_before_horizon_agg(&resps),
266        })
267    }
268
269    /// Forward a coord-based delete to the shard(s) that own the cells.
270    pub async fn coord_delete(
271        &self,
272        req: ArrayShardDeleteReq,
273    ) -> Result<Vec<ArrayShardDeleteResp>> {
274        let req_bytes = zerompk::to_msgpack_vec(&req).map_err(|e| ClusterError::Codec {
275            detail: format!("ArrayShardDeleteReq serialise: {e}"),
276        })?;
277        let raw = fan_out(
278            &self.fan_out_params(),
279            super::super::opcodes::ARRAY_SHARD_DELETE_REQ,
280            &req_bytes,
281            &self.dispatch,
282            &self.circuit_breaker,
283        )
284        .await?;
285        decode_resps::<ArrayShardDeleteResp>(&raw)
286    }
287
288    /// Fan out a surrogate bitmap scan, collect per-shard bitmap bytes, and
289    /// union all bitmaps on the coordinator.
290    ///
291    /// Returns the zerompk-encoded union `SurrogateBitmap` covering all shards.
292    pub async fn coord_surrogate_bitmap_scan(
293        &self,
294        req: ArrayShardSurrogateBitmapReq,
295    ) -> Result<Vec<ArrayShardSurrogateBitmapResp>> {
296        let req_bytes = zerompk::to_msgpack_vec(&req).map_err(|e| ClusterError::Codec {
297            detail: format!("ArrayShardSurrogateBitmapReq serialise: {e}"),
298        })?;
299        let raw = fan_out(
300            &self.fan_out_params(),
301            super::super::opcodes::ARRAY_SHARD_SURROGATE_BITMAP_REQ,
302            &req_bytes,
303            &self.dispatch,
304            &self.circuit_breaker,
305        )
306        .await?;
307        decode_resps::<ArrayShardSurrogateBitmapResp>(&raw)
308    }
309
310    pub(super) fn fan_out_params(&self) -> FanOutParams {
311        FanOutParams {
312            shard_ids: self.params.shard_ids.clone(),
313            timeout_ms: self.params.timeout_ms,
314            source_node: self.params.source_node,
315        }
316    }
317}
318
319/// Deserialise a slice of raw `(shard_id, bytes)` pairs into typed responses.
320pub(super) fn decode_resps<T>(raw: &[(u32, Vec<u8>)]) -> Result<Vec<T>>
321where
322    T: for<'a> zerompk::FromMessagePack<'a>,
323{
324    raw.iter()
325        .map(|(_, bytes)| {
326            zerompk::from_msgpack(bytes).map_err(|e| ClusterError::Codec {
327                detail: format!("array response deserialise: {e}"),
328            })
329        })
330        .collect()
331}
nodedb_cluster/distributed_array/coordinator/read.rs

nodedb_cluster/distributed_array/coordinator/
read.rs
