nodedb_cluster/
wire.rs

1// SPDX-License-Identifier: BUSL-1.1
2
3//! Transport-agnostic vShard envelope.
4//!
5//! "The on-wire format for vShard migration (segment files, WAL tail,
6//! routing metadata) MUST be identical regardless of whether the transport
7//! is RDMA or QUIC/TCP. The transport layer is a dumb pipe; serialization
8//! logic MUST NOT branch on transport type."
9//!
10//! This module defines the canonical wire format for all vShard-related
11//! messages. Both RDMA and QUIC paths serialize/deserialize using this
12//! format — no transport-specific branches.
13
14/// Envelope wrapping all vShard wire messages.
15#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
16pub struct VShardEnvelope {
17    /// Protocol version for forward compatibility.
18    pub version: u16,
19    /// Message type discriminant.
20    pub msg_type: VShardMessageType,
21    /// Source node ID.
22    pub source_node: u64,
23    /// Target node ID.
24    pub target_node: u64,
25    /// vShard being referenced.
26    pub vshard_id: u32,
27    /// Opaque payload (type-dependent).
28    pub payload: Vec<u8>,
29}
30
31/// Message types for vShard wire protocol.
32#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
33#[repr(u16)]
34pub enum VShardMessageType {
35    /// Phase 1: Segment file chunk during base copy.
36    SegmentChunk = 1,
37    /// Phase 1: Segment transfer complete marker.
38    SegmentComplete = 2,
39    /// Phase 2: WAL tail entries for catch-up.
40    WalTail = 3,
41    /// Phase 3: Routing table update (atomic cut-over).
42    RoutingUpdate = 4,
43    /// Routing table acknowledgement.
44    RoutingAck = 5,
45    /// Ghost stub creation notification.
46    GhostCreate = 10,
47    /// Ghost stub deletion notification.
48    GhostDelete = 11,
49    /// Anti-entropy sweep query.
50    GhostVerifyRequest = 12,
51    /// Anti-entropy sweep response.
52    GhostVerifyResponse = 13,
53    /// Migration base-copy segment data.
54    MigrationBaseCopy = 20,
55    /// GSI forward entry.
56    GsiForward = 22,
57    /// Edge validation request.
58    EdgeValidation = 23,
59
60    // ── Timeseries Distributed Aggregation ──
61    /// Scatter: coordinator sends aggregation query to a shard.
62    TsScatterRequest = 40,
63    /// Gather: shard responds with partial aggregates.
64    TsScatterResponse = 41,
65    /// Coordinator broadcasts retention command to all shards.
66    TsRetentionCommand = 42,
67    /// Shard acknowledges retention execution.
68    TsRetentionAck = 43,
69    /// S3 archival command: coordinator tells shard to archive partitions.
70    TsArchiveCommand = 44,
71    /// S3 archival acknowledgement.
72    TsArchiveAck = 45,
73
74    // ── Distributed Vector Search ──
75    /// Scatter: coordinator sends k-NN query to a shard.
76    VectorScatterRequest = 50,
77    /// Gather: shard responds with local top-K hits.
78    VectorScatterResponse = 51,
79
80    // ── Compass: coarse-code routing ──
81    /// Phase 1 request: coordinator asks a shard for its coarse routing
82    /// descriptor (learned coarse codes, centroid summary, or equivalent).
83    /// The shard responds with `VectorCoarseRouteResponse` before the
84    /// coordinator selects the shard subset for the fine search phase.
85    VectorCoarseRouteRequest = 52,
86    /// Phase 1 response: shard returns its coarse routing descriptor so
87    /// the coordinator can decide whether to include it in phase 2.
88    VectorCoarseRouteResponse = 53,
89
90    // ── SPIRE: build-time centroid exchange ──
91    /// Build-time request: a shard sends its IVF centroid table to a peer
92    /// so the peer can build cross-shard centroid knowledge.
93    /// Sent shard-to-shard without coordinator involvement.
94    VectorBuildExchangeRequest = 54,
95    /// Build-time response: receiving shard acknowledges and optionally
96    /// echoes its own centroid summary back to the sender.
97    VectorBuildExchangeResponse = 55,
98
99    // ── CoTra-RDMA: one-sided read support ──
100    /// Registration request: a shard asks a peer to expose a named memory
101    /// region (e.g. a pinned HNSW graph segment) for one-sided reads.
102    /// The peer responds with `VectorMemRegionInfo` containing the address
103    /// and rkey, or indicates the region is unavailable.
104    VectorMemRegionRequest = 56,
105    /// Registration response: peer returns address/rkey for the requested
106    /// memory region, or `available = false` when not supported.
107    VectorMemRegionResponse = 57,
108
109    // ── Distributed Spatial Queries ──
110    /// Scatter: coordinator sends spatial predicate query to a shard.
111    SpatialScatterRequest = 60,
112    /// Gather: shard responds with matching document IDs.
113    SpatialScatterResponse = 61,
114
115    // ── Event Plane Cross-Shard Delivery ──
116    /// Cross-shard event write request (trigger DML, CDC, etc.).
117    CrossShardEvent = 70,
118    /// Acknowledgement for a cross-shard event write.
119    CrossShardEventAck = 71,
120    /// Broadcast NOTIFY message to all peers (LISTEN/NOTIFY cluster-wide).
121    NotifyBroadcast = 72,
122    /// Acknowledgement for a NOTIFY broadcast.
123    NotifyBroadcastAck = 73,
124
125    // ── Distributed Array (Hilbert-sharded sparse arrays) ──
126    /// Scatter: coordinator sends a coord-range slice query to a shard.
127    ArrayShardSliceReq = 80,
128    /// Gather: shard responds with matching row bytes.
129    ArrayShardSliceResp = 81,
130    /// Scatter: coordinator sends an aggregate query to a shard.
131    ArrayShardAggReq = 82,
132    /// Gather: shard responds with partial aggregate(s).
133    ArrayShardAggResp = 83,
134    /// Coordinator forwards a cell write batch to the owning shard.
135    ArrayShardPutReq = 84,
136    /// Shard acknowledges a cell write batch.
137    ArrayShardPutResp = 85,
138    /// Coordinator forwards a coord-based delete to the owning shard.
139    ArrayShardDeleteReq = 86,
140    /// Shard acknowledges a coord-based delete.
141    ArrayShardDeleteResp = 87,
142    /// Scatter: coordinator requests a surrogate bitmap scan from a shard.
143    ArrayShardSurrogateBitmapReq = 88,
144    /// Gather: shard returns the surrogate bitmap for matching cells.
145    ArrayShardSurrogateBitmapResp = 89,
146}
147
148/// Current wire protocol version.
149///
150/// v2 widens `vshard_id` u16→u32 in the binary frame, increasing min header
151/// size from 26 to 28 bytes.
152pub const WIRE_VERSION: u16 = 2;
153
154impl VShardEnvelope {
155    pub fn new(
156        msg_type: VShardMessageType,
157        source_node: u64,
158        target_node: u64,
159        vshard_id: u32,
160        payload: Vec<u8>,
161    ) -> Self {
162        Self {
163            version: WIRE_VERSION,
164            msg_type,
165            source_node,
166            target_node,
167            vshard_id,
168            payload,
169        }
170    }
171
172    /// Serialize to bytes (transport-agnostic).
173    ///
174    /// Binary format (v2, all little-endian):
175    ///   version(2) + msg_type(2) + source(8) + target(8) + vshard(4) + payload_len(4) + payload
176    pub fn to_bytes(&self) -> Vec<u8> {
177        let mut buf = Vec::with_capacity(28 + self.payload.len());
178        buf.extend_from_slice(&self.version.to_le_bytes());
179        buf.extend_from_slice(&(self.msg_type as u16).to_le_bytes());
180        buf.extend_from_slice(&self.source_node.to_le_bytes());
181        buf.extend_from_slice(&self.target_node.to_le_bytes());
182        buf.extend_from_slice(&self.vshard_id.to_le_bytes());
183        buf.extend_from_slice(&(self.payload.len() as u32).to_le_bytes());
184        buf.extend_from_slice(&self.payload);
185        buf
186    }
187
188    /// Deserialize from bytes.
189    pub fn from_bytes(buf: &[u8]) -> Option<Self> {
190        if buf.len() < 28 {
191            return None;
192        }
193        let version = u16::from_le_bytes(buf[0..2].try_into().ok()?);
194        let msg_type_raw = u16::from_le_bytes(buf[2..4].try_into().ok()?);
195        let source_node = u64::from_le_bytes(buf[4..12].try_into().ok()?);
196        let target_node = u64::from_le_bytes(buf[12..20].try_into().ok()?);
197        let vshard_id = u32::from_le_bytes(buf[20..24].try_into().ok()?);
198        let payload_len = u32::from_le_bytes(buf[24..28].try_into().ok()?) as usize;
199
200        if buf.len() < 28 + payload_len {
201            return None;
202        }
203        let payload = buf[28..28 + payload_len].to_vec();
204
205        let msg_type = match msg_type_raw {
206            1 => VShardMessageType::SegmentChunk,
207            2 => VShardMessageType::SegmentComplete,
208            3 => VShardMessageType::WalTail,
209            4 => VShardMessageType::RoutingUpdate,
210            5 => VShardMessageType::RoutingAck,
211            10 => VShardMessageType::GhostCreate,
212            11 => VShardMessageType::GhostDelete,
213            12 => VShardMessageType::GhostVerifyRequest,
214            13 => VShardMessageType::GhostVerifyResponse,
215            20 => VShardMessageType::MigrationBaseCopy,
216            22 => VShardMessageType::GsiForward,
217            23 => VShardMessageType::EdgeValidation,
218            // 30-33 retired (GraphAlgoSuperstep/Contributions/SuperstepAck/Complete)
219            40 => VShardMessageType::TsScatterRequest,
220            41 => VShardMessageType::TsScatterResponse,
221            42 => VShardMessageType::TsRetentionCommand,
222            43 => VShardMessageType::TsRetentionAck,
223            44 => VShardMessageType::TsArchiveCommand,
224            45 => VShardMessageType::TsArchiveAck,
225            50 => VShardMessageType::VectorScatterRequest,
226            51 => VShardMessageType::VectorScatterResponse,
227            52 => VShardMessageType::VectorCoarseRouteRequest,
228            53 => VShardMessageType::VectorCoarseRouteResponse,
229            54 => VShardMessageType::VectorBuildExchangeRequest,
230            55 => VShardMessageType::VectorBuildExchangeResponse,
231            56 => VShardMessageType::VectorMemRegionRequest,
232            57 => VShardMessageType::VectorMemRegionResponse,
233            60 => VShardMessageType::SpatialScatterRequest,
234            61 => VShardMessageType::SpatialScatterResponse,
235            70 => VShardMessageType::CrossShardEvent,
236            71 => VShardMessageType::CrossShardEventAck,
237            72 => VShardMessageType::NotifyBroadcast,
238            73 => VShardMessageType::NotifyBroadcastAck,
239            80 => VShardMessageType::ArrayShardSliceReq,
240            81 => VShardMessageType::ArrayShardSliceResp,
241            82 => VShardMessageType::ArrayShardAggReq,
242            83 => VShardMessageType::ArrayShardAggResp,
243            84 => VShardMessageType::ArrayShardPutReq,
244            85 => VShardMessageType::ArrayShardPutResp,
245            86 => VShardMessageType::ArrayShardDeleteReq,
246            87 => VShardMessageType::ArrayShardDeleteResp,
247            88 => VShardMessageType::ArrayShardSurrogateBitmapReq,
248            89 => VShardMessageType::ArrayShardSurrogateBitmapResp,
249            _ => return None,
250        };
251
252        Some(Self {
253            version,
254            msg_type,
255            source_node,
256            target_node,
257            vshard_id,
258            payload,
259        })
260    }
261}
262
263#[cfg(test)]
264mod tests {
265    use super::*;
266
267    #[test]
268    fn wire_version_is_2() {
269        assert_eq!(WIRE_VERSION, 2, "v2 widened vshard_id to u32");
270    }
271
272    #[test]
273    fn envelope_roundtrip() {
274        let env = VShardEnvelope::new(
275            VShardMessageType::SegmentChunk,
276            1,
277            2,
278            42,
279            b"segment data".to_vec(),
280        );
281        let bytes = env.to_bytes();
282        let decoded = VShardEnvelope::from_bytes(&bytes).unwrap();
283        assert_eq!(env, decoded);
284    }
285
286    #[test]
287    fn all_message_types_roundtrip() {
288        let types = [
289            VShardMessageType::SegmentChunk,
290            VShardMessageType::SegmentComplete,
291            VShardMessageType::WalTail,
292            VShardMessageType::RoutingUpdate,
293            VShardMessageType::RoutingAck,
294            VShardMessageType::GhostCreate,
295            VShardMessageType::GhostDelete,
296            VShardMessageType::GhostVerifyRequest,
297            VShardMessageType::GhostVerifyResponse,
298            VShardMessageType::TsScatterRequest,
299            VShardMessageType::TsScatterResponse,
300            VShardMessageType::TsRetentionCommand,
301            VShardMessageType::TsRetentionAck,
302            VShardMessageType::TsArchiveCommand,
303            VShardMessageType::TsArchiveAck,
304            VShardMessageType::VectorScatterRequest,
305            VShardMessageType::VectorScatterResponse,
306            VShardMessageType::VectorCoarseRouteRequest,
307            VShardMessageType::VectorCoarseRouteResponse,
308            VShardMessageType::VectorBuildExchangeRequest,
309            VShardMessageType::VectorBuildExchangeResponse,
310            VShardMessageType::VectorMemRegionRequest,
311            VShardMessageType::VectorMemRegionResponse,
312            VShardMessageType::SpatialScatterRequest,
313            VShardMessageType::SpatialScatterResponse,
314            VShardMessageType::CrossShardEvent,
315            VShardMessageType::CrossShardEventAck,
316            VShardMessageType::NotifyBroadcast,
317            VShardMessageType::NotifyBroadcastAck,
318            VShardMessageType::ArrayShardSliceReq,
319            VShardMessageType::ArrayShardSliceResp,
320            VShardMessageType::ArrayShardAggReq,
321            VShardMessageType::ArrayShardAggResp,
322            VShardMessageType::ArrayShardPutReq,
323            VShardMessageType::ArrayShardPutResp,
324            VShardMessageType::ArrayShardDeleteReq,
325            VShardMessageType::ArrayShardDeleteResp,
326            VShardMessageType::ArrayShardSurrogateBitmapReq,
327            VShardMessageType::ArrayShardSurrogateBitmapResp,
328        ];
329
330        for msg_type in types {
331            let env = VShardEnvelope::new(msg_type, 10, 20, 100, vec![1, 2, 3]);
332            let bytes = env.to_bytes();
333            let decoded = VShardEnvelope::from_bytes(&bytes).unwrap();
334            assert_eq!(decoded.msg_type, msg_type);
335        }
336    }
337
338    #[test]
339    fn truncated_buffer_returns_none() {
340        let env = VShardEnvelope::new(VShardMessageType::WalTail, 1, 2, 0, vec![0; 100]);
341        let bytes = env.to_bytes();
342        // Truncate payload.
343        assert!(VShardEnvelope::from_bytes(&bytes[..50]).is_none());
344        // Truncate header.
345        assert!(VShardEnvelope::from_bytes(&bytes[..10]).is_none());
346    }
347
348    #[test]
349    fn empty_payload() {
350        let env = VShardEnvelope::new(VShardMessageType::RoutingAck, 5, 6, 999, vec![]);
351        let bytes = env.to_bytes();
352        assert_eq!(bytes.len(), 28); // header only (v2: vshard_id widened u16→u32)
353        let decoded = VShardEnvelope::from_bytes(&bytes).unwrap();
354        assert!(decoded.payload.is_empty());
355    }
356
357    #[test]
358    fn unknown_message_type_returns_none() {
359        let mut env =
360            VShardEnvelope::new(VShardMessageType::SegmentChunk, 1, 2, 0, vec![]).to_bytes();
361        // Corrupt msg_type to unknown value.
362        env[2] = 0xFF;
363        env[3] = 0xFF;
364        assert!(VShardEnvelope::from_bytes(&env).is_none());
365    }
366
367    #[test]
368    fn wire_format_is_transport_agnostic() {
369        // The same bytes work whether sent over RDMA or QUIC.
370        // This test documents the invariant: no transport-specific branching.
371        let env = VShardEnvelope::new(VShardMessageType::SegmentChunk, 1, 2, 42, b"data".to_vec());
372
373        let rdma_bytes = env.to_bytes();
374        let quic_bytes = env.to_bytes();
375        assert_eq!(
376            rdma_bytes, quic_bytes,
377            "wire format must be transport-agnostic"
378        );
379    }
380
381    #[test]
382    fn large_vshard_id_roundtrip() {
383        // vshard_id is now u32; ensure values above old u16::MAX round-trip
384        // without truncation.
385        let env = VShardEnvelope::new(
386            VShardMessageType::WalTail,
387            10,
388            20,
389            0x0001_FFFF,
390            b"payload".to_vec(),
391        );
392        let bytes = env.to_bytes();
393        let decoded = VShardEnvelope::from_bytes(&bytes).unwrap();
394        assert_eq!(decoded.vshard_id, 0x0001_FFFFu32);
395    }
396
397    #[test]
398    fn truncated_below_28_returns_none() {
399        // With v2 header = 28 bytes, a 27-byte buffer must be rejected.
400        let env = VShardEnvelope::new(VShardMessageType::RoutingAck, 1, 2, 0, vec![]);
401        let bytes = env.to_bytes();
402        assert_eq!(bytes.len(), 28);
403        assert!(VShardEnvelope::from_bytes(&bytes[..27]).is_none());
404    }
405}
nodedb_cluster/wire.rs

nodedb_cluster/
wire.rs
