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

nodedb_types/sync/wire/
frame.rs
