brain_db_sdk/wire/
frame.rs

1//! Brain wire-protocol frame codec.
2//!
3//! Re-implements Brain's 32-byte `BRN0` frame header. The conformance
4//! corpus is the drift guard against the server.
5//!
6//! Layout — all multi-byte integers big-endian:
7//! ```text
8//! byte  0..4   magic "BRN0"
9//! byte  4      version (= 1)
10//! byte  5..7   opcode (u16 BE; high byte = namespace 0x00 / 0x01)
11//! byte  7      flags (EOS=0x80, MPL=0x40, CMP=0x20)
12//! byte  8..12  header_crc32c (u32 BE) — CRC32C over header bytes 0..8 ++ 12..32
13//! byte 12..16  stream_id (u32 BE)
14//! byte 16..19  payload_len (u24 BE)
15//! byte 19      reserved_a (= 0)
16//! byte 20..24  payload_crc32c (u32 BE; 0 when payload empty)
17//! byte 24..32  reserved_b (8 bytes, = 0)
18//! ```
19//! Payload (not part of this header) = CBOR map + optional trailing raw
20//! little-endian f32 vector section.
21
22// The types below mirror the server's wire structs one-for-one: same names,
23// same field names, same order. Their documentation is the protocol itself —
24// `conformance/protocol.json` carries every field's type and serde attributes,
25// and the corpus pins the bytes. A doc comment on each of ~2000 fields would
26// restate the field name and nothing more, and the ones that DO carry meaning
27// beyond their name (written below) would be lost in the noise.
28#![allow(missing_docs)]
29
30/// Magic bytes at offset 0: ASCII `"BRN0"`.
31pub const MAGIC: [u8; 4] = *b"BRN0";
32
33/// Wire-protocol version (header byte 4).
34pub const WIRE_VERSION: u8 = 1;
35
36/// Total header size, in bytes.
37pub const HEADER_SIZE: usize = 32;
38
39// Frame flag bits (header byte 7).
40pub const FLAG_EOS: u8 = 0x80;
41pub const FLAG_MPL: u8 = 0x40;
42pub const FLAG_CMP: u8 = 0x20;
43
44/// Flag bits that are defined; everything else MUST be zero on the wire.
45/// A corrupt or hostile peer that sets a reserved bit is rejected rather
46/// than silently reinterpreted.
47pub const FLAGS_DEFINED_MASK: u8 = FLAG_EOS | FLAG_MPL | FLAG_CMP;
48
49// Opcode namespaces (high byte of the u16 opcode).
50pub const OPCODE_NS_SUBSTRATE: u8 = 0x00;
51pub const OPCODE_NS_TYPED_GRAPH: u8 = 0x01;
52
53// Field byte offsets within the 32-byte header.
54pub const OFF_MAGIC: usize = 0;
55pub const OFF_VERSION: usize = 4;
56pub const OFF_OPCODE: usize = 5;
57pub const OFF_FLAGS: usize = 7;
58pub const OFF_HEADER_CRC: usize = 8;
59pub const OFF_STREAM_ID: usize = 12;
60pub const OFF_PAYLOAD_LEN: usize = 16;
61pub const OFF_RESERVED_A: usize = 19;
62pub const OFF_PAYLOAD_CRC: usize = 20;
63pub const OFF_RESERVED_B: usize = 24;
64
65/// Max payload size: `2^24 - 1` (the u24 `payload_len` field).
66pub const MAX_PAYLOAD_BYTES: usize = 0xff_ff_ff;
67
68/// Errors raised while decoding a frame off the wire. Each maps to a
69/// wire protocol-error condition; a mismatch is fail-stop for the
70/// connection (a corrupt frame desynchronizes the byte stream).
71#[derive(Debug, thiserror::Error, PartialEq, Eq)]
72pub enum FrameError {
73    /// First four bytes were not `b"BRN0"`.
74    #[error("bad frame magic")]
75    BadMagic,
76    /// Version byte did not match [`WIRE_VERSION`].
77    #[error("unsupported wire version: got {got}, expected {expected}")]
78    BadVersion { got: u8, expected: u8 },
79    /// `header_crc32c` did not match the recomputed CRC.
80    #[error("header CRC mismatch")]
81    BadHeaderCrc,
82    /// `payload_crc32c` did not match the CRC of the payload bytes.
83    #[error("payload CRC mismatch")]
84    BadPayloadCrc,
85    /// `payload_len` exceeded the negotiated maximum — rejected before
86    /// any payload buffer is reserved, so a hostile length can't force a
87    /// large allocation.
88    #[error("payload length {len} exceeds maximum {max}")]
89    OversizePayload { len: u32, max: u32 },
90    /// A reserved header byte or reserved flag bit was non-zero.
91    #[error("reserved field or flag bit was non-zero")]
92    ReservedNonZero,
93    /// Not enough bytes buffered yet to decode the frame; caller should
94    /// read more from the socket and retry.
95    #[error("truncated frame: have {have} bytes, need {need}")]
96    Truncated { have: usize, need: usize },
97}
98
99/// A decoded frame: header fields + the raw payload bytes (CBOR section
100/// plus any trailing raw-vector section, undivided at this layer).
101#[derive(Clone, Debug, PartialEq, Eq)]
102pub struct Frame {
103    pub opcode: u16,
104    pub flags: u8,
105    pub stream_id: u32,
106    pub payload: Vec<u8>,
107}
108
109/// CRC32C over the header bytes that exclude the `header_crc32c` field
110/// itself: bytes `0..8` concatenated with `12..32`. The 4-byte CRC slot
111/// is held at zero so encode and decode agree on the covered bytes.
112fn compute_header_crc(header: &[u8; HEADER_SIZE]) -> u32 {
113    let mut buf = [0u8; HEADER_SIZE - 4];
114    buf[..OFF_HEADER_CRC].copy_from_slice(&header[..OFF_HEADER_CRC]);
115    buf[OFF_HEADER_CRC..].copy_from_slice(&header[OFF_STREAM_ID..]);
116    crc32c::crc32c(&buf)
117}
118
119impl Frame {
120    /// Build a frame from its parts.
121    #[must_use]
122    pub fn new(opcode: u16, flags: u8, stream_id: u32, payload: Vec<u8>) -> Self {
123        Self {
124            opcode,
125            flags,
126            stream_id,
127            payload,
128        }
129    }
130
131    /// Serialize the frame to a wire-ready byte vector: the 32-byte
132    /// header with both CRC32C fields computed, followed by the payload.
133    ///
134    /// # Errors
135    ///
136    /// [`FrameError::OversizePayload`] if `payload.len()` exceeds
137    /// [`MAX_PAYLOAD_BYTES`] — the u24 length field physically cannot
138    /// represent more.
139    ///
140    /// This returns rather than panics because the length is *caller data*: a
141    /// caller encoding a large document would otherwise take down the process
142    /// (or abort the task) instead of getting an error they can handle. Python
143    /// raises `OversizePayload` and TypeScript throws `FrameError` for the same
144    /// input; this SDK used to be the only one that panicked.
145    ///
146    /// # Panics
147    ///
148    /// Never. The one `expect` below slices the first `HEADER_SIZE` bytes of a
149    /// buffer this function just allocated as `HEADER_SIZE + payload.len()`, so
150    /// the conversion cannot fail; it documents an invariant rather than
151    /// guarding a reachable case.
152    pub fn encode(&self) -> Result<Vec<u8>, FrameError> {
153        if self.payload.len() > MAX_PAYLOAD_BYTES {
154            return Err(FrameError::OversizePayload {
155                #[allow(clippy::cast_possible_truncation)]
156                len: self.payload.len() as u32,
157                #[allow(clippy::cast_possible_truncation)]
158                max: MAX_PAYLOAD_BYTES as u32,
159            });
160        }
161
162        let mut out = vec![0u8; HEADER_SIZE + self.payload.len()];
163        {
164            let header = &mut out[..HEADER_SIZE];
165            header[OFF_MAGIC..OFF_MAGIC + 4].copy_from_slice(&MAGIC);
166            header[OFF_VERSION] = WIRE_VERSION;
167            header[OFF_OPCODE..OFF_OPCODE + 2].copy_from_slice(&self.opcode.to_be_bytes());
168            header[OFF_FLAGS] = self.flags;
169            // header_crc32c (8..12) left zero until computed below.
170            header[OFF_STREAM_ID..OFF_STREAM_ID + 4].copy_from_slice(&self.stream_id.to_be_bytes());
171            #[allow(clippy::cast_possible_truncation)]
172            let len_be = (self.payload.len() as u32).to_be_bytes();
173            header[OFF_PAYLOAD_LEN..OFF_PAYLOAD_LEN + 3].copy_from_slice(&len_be[1..]);
174            // reserved_a (19) and reserved_b (24..32) left zero.
175            // An empty payload carries a zero payload CRC by convention,
176            // which crc32c of an empty slice already produces.
177            let payload_crc = crc32c::crc32c(&self.payload);
178            header[OFF_PAYLOAD_CRC..OFF_PAYLOAD_CRC + 4]
179                .copy_from_slice(&payload_crc.to_be_bytes());
180        }
181
182        let header_arr: [u8; HEADER_SIZE] = out[..HEADER_SIZE]
183            .try_into()
184            .expect("invariant: out starts with exactly HEADER_SIZE header bytes");
185        let header_crc = compute_header_crc(&header_arr);
186        out[OFF_HEADER_CRC..OFF_HEADER_CRC + 4].copy_from_slice(&header_crc.to_be_bytes());
187
188        out[HEADER_SIZE..].copy_from_slice(&self.payload);
189        Ok(out)
190    }
191
192    /// Decode a single frame off the front of `bytes`, returning the
193    /// frame and the unconsumed tail so a caller can loop over a stream.
194    ///
195    /// Validates magic / version / reserved bytes / reserved flag bits,
196    /// verifies the header CRC, then checks `payload_len` against
197    /// [`MAX_PAYLOAD_BYTES`] **before** slicing the payload, and finally
198    /// verifies the payload CRC.
199    ///
200    /// # Panics
201    ///
202    /// Never. The slice indexing below is all bounds-checked against
203    /// `HEADER_SIZE` and `payload_len` before it runs.
204    pub fn decode(bytes: &[u8]) -> Result<(Frame, &[u8]), FrameError> {
205        if bytes.len() < HEADER_SIZE {
206            return Err(FrameError::Truncated {
207                have: bytes.len(),
208                need: HEADER_SIZE,
209            });
210        }
211
212        let header: [u8; HEADER_SIZE] = bytes[..HEADER_SIZE]
213            .try_into()
214            .expect("invariant: slice is exactly HEADER_SIZE bytes after the length check");
215
216        if header[OFF_MAGIC..OFF_MAGIC + 4] != MAGIC {
217            return Err(FrameError::BadMagic);
218        }
219        if header[OFF_VERSION] != WIRE_VERSION {
220            return Err(FrameError::BadVersion {
221                got: header[OFF_VERSION],
222                expected: WIRE_VERSION,
223            });
224        }
225        if header[OFF_RESERVED_A] != 0 || header[OFF_RESERVED_B..HEADER_SIZE] != [0u8; 8] {
226            return Err(FrameError::ReservedNonZero);
227        }
228        let flags = header[OFF_FLAGS];
229        if flags & !FLAGS_DEFINED_MASK != 0 {
230            return Err(FrameError::ReservedNonZero);
231        }
232
233        let stored_header_crc = u32::from_be_bytes([
234            header[OFF_HEADER_CRC],
235            header[OFF_HEADER_CRC + 1],
236            header[OFF_HEADER_CRC + 2],
237            header[OFF_HEADER_CRC + 3],
238        ]);
239        if stored_header_crc != compute_header_crc(&header) {
240            return Err(FrameError::BadHeaderCrc);
241        }
242
243        let payload_len = u32::from_be_bytes([
244            0,
245            header[OFF_PAYLOAD_LEN],
246            header[OFF_PAYLOAD_LEN + 1],
247            header[OFF_PAYLOAD_LEN + 2],
248        ]);
249        // Reject an over-long claim before reserving any payload buffer —
250        // a corrupt or hostile length must not drive a large allocation.
251        #[allow(clippy::cast_possible_truncation)]
252        if payload_len as usize > MAX_PAYLOAD_BYTES {
253            return Err(FrameError::OversizePayload {
254                len: payload_len,
255                max: MAX_PAYLOAD_BYTES as u32,
256            });
257        }
258
259        let need = HEADER_SIZE + payload_len as usize;
260        if bytes.len() < need {
261            return Err(FrameError::Truncated {
262                have: bytes.len(),
263                need,
264            });
265        }
266        let payload_slice = &bytes[HEADER_SIZE..need];
267
268        let stored_payload_crc = u32::from_be_bytes([
269            header[OFF_PAYLOAD_CRC],
270            header[OFF_PAYLOAD_CRC + 1],
271            header[OFF_PAYLOAD_CRC + 2],
272            header[OFF_PAYLOAD_CRC + 3],
273        ]);
274        if stored_payload_crc != crc32c::crc32c(payload_slice) {
275            return Err(FrameError::BadPayloadCrc);
276        }
277
278        let frame = Frame {
279            opcode: u16::from_be_bytes([header[OFF_OPCODE], header[OFF_OPCODE + 1]]),
280            flags,
281            stream_id: u32::from_be_bytes([
282                header[OFF_STREAM_ID],
283                header[OFF_STREAM_ID + 1],
284                header[OFF_STREAM_ID + 2],
285                header[OFF_STREAM_ID + 3],
286            ]),
287            payload: payload_slice.to_vec(),
288        };
289        Ok((frame, &bytes[need..]))
290    }
291}
292
293#[cfg(test)]
294mod tests {
295    use super::*;
296
297    fn sample() -> Frame {
298        Frame::new(0x0021, FLAG_EOS, 7, b"hello brain".to_vec())
299    }
300
301    #[test]
302    fn encode_decode_round_trip() {
303        let original = sample();
304        let bytes = original.encode().expect("frame fits");
305        assert_eq!(bytes.len(), HEADER_SIZE + original.payload.len());
306        let (decoded, rest) = Frame::decode(&bytes).expect("decode");
307        assert_eq!(rest, &[] as &[u8]);
308        assert_eq!(decoded, original);
309    }
310
311    #[test]
312    fn empty_payload_has_zero_payload_crc() {
313        let frame = Frame::new(0x0010, 0, 0, Vec::new());
314        let bytes = frame.encode().expect("frame fits");
315        assert_eq!(bytes.len(), HEADER_SIZE);
316        assert_eq!(&bytes[OFF_PAYLOAD_CRC..OFF_PAYLOAD_CRC + 4], &[0u8; 4]);
317        let (decoded, rest) = Frame::decode(&bytes).expect("decode empty");
318        assert_eq!(rest, &[] as &[u8]);
319        assert_eq!(decoded.payload, Vec::<u8>::new());
320    }
321
322    #[test]
323    fn decode_returns_unconsumed_tail() {
324        let mut bytes = sample().encode().expect("frame fits");
325        bytes.extend_from_slice(b"TAIL");
326        let (_frame, rest) = Frame::decode(&bytes).expect("decode");
327        assert_eq!(rest, b"TAIL");
328    }
329
330    #[test]
331    fn rejects_bad_magic() {
332        let mut bytes = sample().encode().expect("frame fits");
333        bytes[0] = b'X';
334        assert_eq!(Frame::decode(&bytes), Err(FrameError::BadMagic));
335    }
336
337    #[test]
338    fn rejects_bad_version() {
339        let mut bytes = sample().encode().expect("frame fits");
340        bytes[OFF_VERSION] = 99;
341        assert_eq!(
342            Frame::decode(&bytes),
343            Err(FrameError::BadVersion {
344                got: 99,
345                expected: 1
346            })
347        );
348    }
349
350    #[test]
351    fn rejects_bad_header_crc() {
352        let mut bytes = sample().encode().expect("frame fits");
353        bytes[OFF_HEADER_CRC] ^= 0xFF;
354        assert_eq!(Frame::decode(&bytes), Err(FrameError::BadHeaderCrc));
355    }
356
357    #[test]
358    fn rejects_bad_payload_crc() {
359        let mut bytes = sample().encode().expect("frame fits");
360        bytes[HEADER_SIZE] ^= 0xFF;
361        assert_eq!(Frame::decode(&bytes), Err(FrameError::BadPayloadCrc));
362    }
363
364    #[test]
365    fn rejects_reserved_a_nonzero() {
366        let mut bytes = sample().encode().expect("frame fits");
367        bytes[OFF_RESERVED_A] = 1;
368        // Re-seal the header CRC so the reserved check (not the CRC
369        // check) is the one that fires.
370        let mut header: [u8; HEADER_SIZE] = bytes[..HEADER_SIZE].try_into().unwrap();
371        let crc = compute_header_crc(&{
372            let mut h = header;
373            h[OFF_HEADER_CRC..OFF_HEADER_CRC + 4].copy_from_slice(&[0; 4]);
374            h
375        });
376        header[OFF_HEADER_CRC..OFF_HEADER_CRC + 4].copy_from_slice(&crc.to_be_bytes());
377        bytes[..HEADER_SIZE].copy_from_slice(&header);
378        assert_eq!(Frame::decode(&bytes), Err(FrameError::ReservedNonZero));
379    }
380
381    #[test]
382    fn rejects_reserved_flag_bit() {
383        // 0x01 is a reserved flag bit. Build the frame raw so the header
384        // CRC covers the bad flag, isolating the reserved-flag check.
385        let frame = Frame::new(0x0010, 0x01, 0, Vec::new());
386        let bytes = frame.encode().expect("frame fits");
387        assert_eq!(Frame::decode(&bytes), Err(FrameError::ReservedNonZero));
388    }
389
390    #[test]
391    fn rejects_truncated_header() {
392        let bytes = sample().encode().expect("frame fits");
393        let truncated = &bytes[..10];
394        assert_eq!(
395            Frame::decode(truncated),
396            Err(FrameError::Truncated {
397                have: 10,
398                need: HEADER_SIZE
399            })
400        );
401    }
402
403    #[test]
404    fn rejects_truncated_payload() {
405        let bytes = sample().encode().expect("frame fits");
406        let truncated = &bytes[..bytes.len() - 3];
407        assert!(matches!(
408            Frame::decode(truncated),
409            Err(FrameError::Truncated { .. })
410        ));
411    }
412
413    /// Standard CRC32C check vector pins the polynomial choice.
414    #[test]
415    fn crc32c_check_vector() {
416        assert_eq!(crc32c::crc32c(b"123456789"), 0xE306_9283);
417    }
418}
brain_db_sdk/wire/frame.rs

brain_db_sdk/wire/
frame.rs
