nodedb_types/sync/wire/
frame.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Wire frame format and message-type discriminants.
4//!
5//! Additional opcodes beyond what the module-level doc in `mod.rs` lists:
6//! - `0xA2` VectorInsert (client → server)
7//! - `0xA3` VectorInsertAck (server → client)
8//! - `0xA4` VectorDelete (client → server)
9//! - `0xA5` VectorDeleteAck (server → client)
10//! - `0xA6` FtsIndex (client → server)
11//! - `0xA7` FtsIndexAck (server → client)
12//! - `0xA8` FtsDelete (client → server)
13//! - `0xA9` FtsDeleteAck (server → client)
14//! - `0xAA` SpatialInsert (client → server)
15//! - `0xAB` SpatialInsertAck (server → client)
16//! - `0xAC` SpatialDelete (client → server)
17//! - `0xAD` SpatialDeleteAck (server → client)
18
19/// Sync message type identifiers.
20#[derive(Debug, Clone, Copy, PartialEq, Eq)]
21#[repr(u8)]
22#[non_exhaustive]
23pub enum SyncMessageType {
24    Handshake = 0x01,
25    HandshakeAck = 0x02,
26    DeltaPush = 0x10,
27    DeltaAck = 0x11,
28    DeltaReject = 0x12,
29    /// Collection schema descriptor announcement (bidirectional, 0x13).
30    /// Carries a `CollectionDescriptor` so any peer can materialize the
31    /// collection in its catalog with the correct engine + config.
32    CollectionSchema = 0x13,
33    /// Collection purged notification (server → client, 0x14).
34    /// Sent when an Origin collection is hard-deleted (UNDROP window
35    /// expired or explicit `DROP COLLECTION ... PURGE`). The client
36    /// must drop local Loro state and remove the collection's redb
37    /// record; future deltas for the collection are not sync-eligible.
38    CollectionPurged = 0x14,
39    ShapeSubscribe = 0x20,
40    ShapeSnapshot = 0x21,
41    ShapeDelta = 0x22,
42    ShapeUnsubscribe = 0x23,
43    VectorClockSync = 0x30,
44    /// Timeseries metric batch push (client → server, 0x40).
45    TimeseriesPush = 0x40,
46    /// Timeseries push acknowledgment (server → client, 0x41).
47    TimeseriesAck = 0x41,
48    /// Re-sync request (bidirectional, 0x50).
49    /// Sent when sequence gaps or checksum failures are detected.
50    ResyncRequest = 0x50,
51    /// Downstream throttle (client → server, 0x52).
52    /// Sent when Lite's incoming queue is overwhelmed.
53    Throttle = 0x52,
54    /// Token refresh request (client → server, 0x60).
55    TokenRefresh = 0x60,
56    /// Token refresh acknowledgment (server → client, 0x61).
57    TokenRefreshAck = 0x61,
58    /// Definition sync (server → client, 0x70).
59    /// Carries function/trigger/procedure definitions from Origin to Lite.
60    DefinitionSync = 0x70,
61    /// Presence update (client → server, 0x80).
62    PresenceUpdate = 0x80,
63    /// Presence broadcast (server → all subscribers except sender, 0x81).
64    PresenceBroadcast = 0x81,
65    /// Presence leave (server → all subscribers, 0x82).
66    PresenceLeave = 0x82,
67    /// Array CRDT delta (single op, client → server, 0x90).
68    ArrayDelta = 0x90,
69    /// Array CRDT delta batch (multiple ops, client → server, 0x91).
70    ArrayDeltaBatch = 0x91,
71    /// Array snapshot header (server → client, 0x92).
72    ArraySnapshot = 0x92,
73    /// Array snapshot chunk (server → client, 0x93).
74    ArraySnapshotChunk = 0x93,
75    /// Array schema CRDT sync (bidirectional, 0x94).
76    ArraySchema = 0x94,
77    /// Array ack — advances GC frontier (client → server, 0x95).
78    ArrayAck = 0x95,
79    /// Array reject (server → client, 0x96). Compensation hint.
80    ArrayReject = 0x96,
81    /// Array catchup request (client → server, 0x97).
82    ArrayCatchupRequest = 0x97,
83    /// Columnar batch insert (client → server, 0xA0).
84    ColumnarInsert = 0xA0,
85    /// Columnar insert acknowledgment (server → client, 0xA1).
86    ColumnarInsertAck = 0xA1,
87    /// Vector insert (client → server, 0xA2).
88    VectorInsert = 0xA2,
89    /// Vector insert acknowledgment (server → client, 0xA3).
90    VectorInsertAck = 0xA3,
91    /// Vector delete (client → server, 0xA4).
92    VectorDelete = 0xA4,
93    /// Vector delete acknowledgment (server → client, 0xA5).
94    VectorDeleteAck = 0xA5,
95    /// FTS document index (client → server, 0xA6).
96    FtsIndex = 0xA6,
97    /// FTS index acknowledgment (server → client, 0xA7).
98    FtsIndexAck = 0xA7,
99    /// FTS document delete (client → server, 0xA8).
100    FtsDelete = 0xA8,
101    /// FTS delete acknowledgment (server → client, 0xA9).
102    FtsDeleteAck = 0xA9,
103    /// Spatial geometry insert (client → server, 0xAA).
104    SpatialInsert = 0xAA,
105    /// Spatial insert acknowledgment (server → client, 0xAB).
106    SpatialInsertAck = 0xAB,
107    /// Spatial geometry delete (client → server, 0xAC).
108    SpatialDelete = 0xAC,
109    /// Spatial delete acknowledgment (server → client, 0xAD).
110    SpatialDeleteAck = 0xAD,
111    PingPong = 0xFF,
112}
113
114impl SyncMessageType {
115    pub fn from_u8(v: u8) -> Option<Self> {
116        match v {
117            0x01 => Some(Self::Handshake),
118            0x02 => Some(Self::HandshakeAck),
119            0x10 => Some(Self::DeltaPush),
120            0x11 => Some(Self::DeltaAck),
121            0x12 => Some(Self::DeltaReject),
122            0x13 => Some(Self::CollectionSchema),
123            0x14 => Some(Self::CollectionPurged),
124            0x20 => Some(Self::ShapeSubscribe),
125            0x21 => Some(Self::ShapeSnapshot),
126            0x22 => Some(Self::ShapeDelta),
127            0x23 => Some(Self::ShapeUnsubscribe),
128            0x30 => Some(Self::VectorClockSync),
129            0x40 => Some(Self::TimeseriesPush),
130            0x41 => Some(Self::TimeseriesAck),
131            0x50 => Some(Self::ResyncRequest),
132            0x52 => Some(Self::Throttle),
133            0x60 => Some(Self::TokenRefresh),
134            0x61 => Some(Self::TokenRefreshAck),
135            0x70 => Some(Self::DefinitionSync),
136            0x80 => Some(Self::PresenceUpdate),
137            0x81 => Some(Self::PresenceBroadcast),
138            0x82 => Some(Self::PresenceLeave),
139            0x90 => Some(Self::ArrayDelta),
140            0x91 => Some(Self::ArrayDeltaBatch),
141            0x92 => Some(Self::ArraySnapshot),
142            0x93 => Some(Self::ArraySnapshotChunk),
143            0x94 => Some(Self::ArraySchema),
144            0x95 => Some(Self::ArrayAck),
145            0x96 => Some(Self::ArrayReject),
146            0x97 => Some(Self::ArrayCatchupRequest),
147            0xA0 => Some(Self::ColumnarInsert),
148            0xA1 => Some(Self::ColumnarInsertAck),
149            0xA2 => Some(Self::VectorInsert),
150            0xA3 => Some(Self::VectorInsertAck),
151            0xA4 => Some(Self::VectorDelete),
152            0xA5 => Some(Self::VectorDeleteAck),
153            0xA6 => Some(Self::FtsIndex),
154            0xA7 => Some(Self::FtsIndexAck),
155            0xA8 => Some(Self::FtsDelete),
156            0xA9 => Some(Self::FtsDeleteAck),
157            0xAA => Some(Self::SpatialInsert),
158            0xAB => Some(Self::SpatialInsertAck),
159            0xAC => Some(Self::SpatialDelete),
160            0xAD => Some(Self::SpatialDeleteAck),
161            0xFF => Some(Self::PingPong),
162            _ => None,
163        }
164    }
165}
166
167/// Wire frame: wraps a message type + serialized body with format versioning
168/// and CRC32C integrity protection.
169///
170/// Layout: `[version: 1B][msg_type: 1B][length: 4B LE][crc32c: 4B LE][body: N bytes]`
171/// Total header: 10 bytes.
172///
173/// Mirrors the WAL record header model: every frame carries a format version
174/// byte (enabling hard rejection of future incompatible formats) and a CRC32C
175/// checksum over the body for silent-corruption detection. There is no
176/// checksum-skip sentinel — CRC32C is always computed and always verified.
177///
178/// `#[non_exhaustive]` — additional header fields (e.g. compression flag,
179/// session token) may be added without breaking downstream consumers.
180#[non_exhaustive]
181#[derive(Clone)]
182pub struct SyncFrame {
183    pub msg_type: SyncMessageType,
184    pub body: Vec<u8>,
185}
186
187impl SyncFrame {
188    /// Current wire format version embedded in every frame header.
189    pub const FORMAT_VERSION: u8 = 1;
190
191    /// Total size of the frame header in bytes.
192    ///
193    /// Layout: `[version:1][msg_type:1][len:4 LE][crc32c:4 LE]` = 10 bytes.
194    pub const HEADER_SIZE: usize = 10;
195
196    /// Serialize a frame to bytes.
197    ///
198    /// Produces `[FORMAT_VERSION][msg_type][body_len as u32 LE][crc32c(body) as u32 LE][body]`.
199    pub fn to_bytes(&self) -> Vec<u8> {
200        let len = self.body.len() as u32;
201        let crc = crc32c::crc32c(&self.body);
202        let mut buf = Vec::with_capacity(Self::HEADER_SIZE + self.body.len());
203        buf.push(Self::FORMAT_VERSION);
204        buf.push(self.msg_type as u8);
205        buf.extend_from_slice(&len.to_le_bytes());
206        buf.extend_from_slice(&crc.to_le_bytes());
207        buf.extend_from_slice(&self.body);
208        buf
209    }
210
211    /// Deserialize a frame from bytes.
212    ///
213    /// Returns `None` if:
214    /// - the buffer is shorter than `HEADER_SIZE`,
215    /// - the version byte does not match `FORMAT_VERSION` (unknown future version),
216    /// - the message type discriminant is unrecognised,
217    /// - the buffer is too short to contain the declared body length, or
218    /// - the CRC32C of the body does not match the header checksum (corrupt frame).
219    pub fn from_bytes(data: &[u8]) -> Option<Self> {
220        if data.len() < Self::HEADER_SIZE {
221            return None;
222        }
223        let version = data[0];
224        if version != Self::FORMAT_VERSION {
225            return None;
226        }
227        let msg_type = SyncMessageType::from_u8(data[1])?;
228        let len = u32::from_le_bytes(data[2..6].try_into().ok()?) as usize;
229        let expected_crc = u32::from_le_bytes(data[6..10].try_into().ok()?);
230        if data.len() < Self::HEADER_SIZE + len {
231            return None;
232        }
233        let body = data[Self::HEADER_SIZE..Self::HEADER_SIZE + len].to_vec();
234        let actual_crc = crc32c::crc32c(&body);
235        if actual_crc != expected_crc {
236            tracing::warn!(
237                msg_type = data[1],
238                expected_crc,
239                actual_crc,
240                "sync frame CRC32C mismatch; dropping corrupt frame"
241            );
242            return None;
243        }
244        Some(Self { msg_type, body })
245    }
246
247    /// Create a frame with a MessagePack-serialized body.
248    pub fn new_msgpack<T: zerompk::ToMessagePack>(
249        msg_type: SyncMessageType,
250        value: &T,
251    ) -> Option<Self> {
252        let body = zerompk::to_msgpack_vec(value).ok()?;
253        Some(Self { msg_type, body })
254    }
255
256    /// Try to encode a value into a SyncFrame body.
257    ///
258    /// Returns `None` and logs an error on serialization failure — callers
259    /// should propagate via `?`. The protocol must never ship a frame
260    /// whose body did not serialize successfully.
261    pub fn try_encode<T: zerompk::ToMessagePack>(
262        msg_type: SyncMessageType,
263        value: &T,
264    ) -> Option<Self> {
265        match zerompk::to_msgpack_vec(value) {
266            Ok(body) => Some(Self { msg_type, body }),
267            Err(e) => {
268                tracing::error!(
269                    msg_type = msg_type as u8,
270                    error = %e,
271                    "failed to encode sync frame body; dropping response"
272                );
273                None
274            }
275        }
276    }
277
278    /// Deserialize the body from MessagePack.
279    pub fn decode_body<T: zerompk::FromMessagePackOwned>(&self) -> Option<T> {
280        zerompk::from_msgpack(&self.body).ok()
281    }
282}
283
284#[cfg(test)]
285mod tests {
286    use super::*;
287
288    fn make_frame(msg_type: SyncMessageType, body: Vec<u8>) -> SyncFrame {
289        SyncFrame { msg_type, body }
290    }
291
292    #[test]
293    fn roundtrip_preserves_msg_type_and_body() {
294        let body = b"hello sync world".to_vec();
295        let frame = make_frame(SyncMessageType::PingPong, body.clone());
296        let bytes = frame.to_bytes();
297        let decoded = SyncFrame::from_bytes(&bytes).unwrap();
298        assert_eq!(decoded.msg_type, SyncMessageType::PingPong);
299        assert_eq!(decoded.body, body);
300    }
301
302    #[test]
303    fn flipped_body_byte_returns_none() {
304        let body = b"integrity check".to_vec();
305        let frame = make_frame(SyncMessageType::DeltaPush, body);
306        let mut bytes = frame.to_bytes();
307        // Flip a bit in the first body byte (located at HEADER_SIZE).
308        bytes[SyncFrame::HEADER_SIZE] ^= 0xFF;
309        assert!(SyncFrame::from_bytes(&bytes).is_none());
310    }
311
312    #[test]
313    fn truncated_buffer_returns_none() {
314        // Buffer shorter than HEADER_SIZE.
315        assert!(SyncFrame::from_bytes(&[]).is_none());
316        assert!(SyncFrame::from_bytes(&[1u8; SyncFrame::HEADER_SIZE - 1]).is_none());
317
318        // Header intact but body truncated by one byte.
319        let frame = make_frame(SyncMessageType::PingPong, b"abcdef".to_vec());
320        let bytes = frame.to_bytes();
321        let truncated = &bytes[..bytes.len() - 1];
322        assert!(SyncFrame::from_bytes(truncated).is_none());
323    }
324
325    #[test]
326    fn wrong_version_returns_none() {
327        let frame = make_frame(SyncMessageType::PingPong, b"version test".to_vec());
328        let mut bytes = frame.to_bytes();
329        // Overwrite the version byte with something other than FORMAT_VERSION.
330        bytes[0] = SyncFrame::FORMAT_VERSION.wrapping_add(1);
331        assert!(SyncFrame::from_bytes(&bytes).is_none());
332    }
333
334    #[test]
335    fn header_size_is_ten_and_total_length_is_correct() {
336        assert_eq!(SyncFrame::HEADER_SIZE, 10);
337        let body = b"nodedb".to_vec();
338        let frame = make_frame(SyncMessageType::PingPong, body.clone());
339        let bytes = frame.to_bytes();
340        assert_eq!(bytes.len(), SyncFrame::HEADER_SIZE + body.len());
341    }
342
343    #[test]
344    fn crc32c_field_matches_crate_output() {
345        let body = b"crc check".to_vec();
346        let frame = make_frame(SyncMessageType::Handshake, body.clone());
347        let bytes = frame.to_bytes();
348        // CRC32C is stored at bytes[6..10] LE.
349        let stored = u32::from_le_bytes(bytes[6..10].try_into().unwrap());
350        let expected = crc32c::crc32c(&body);
351        assert_eq!(stored, expected);
352    }
353}
nodedb_types/sync/wire/frame.rs

nodedb_types/sync/wire/
frame.rs
