graven/
envelope.rs

1//! The optional envelope that wraps a raw Graven message for disk or network.
2//!
3//! Layout (little-endian):
4//!
5//! ```text
6//! offset  size  field
7//! 0       4     magic = "GRVN"
8//! 4       1     format_version = 1
9//! 5       1     flags = 0
10//! 6       2     reserved = 0
11//! 8       8     schema_id
12//! 16      4     payload_len
13//! 20      N     payload
14//! 20+N    4     crc32c of bytes [0, 20+N)
15//! ```
16use crate::{Error, crc32c};
17
18/// Software-only CRC32C (for benchmarking against the accelerated version). Not a stable API.
19#[doc(hidden)]
20pub use crate::crc::crc32c_sw;
21use bytes::{BufMut, BytesMut};
22use core::iter::FusedIterator;
23
24/// The four magic bytes that start every envelope.
25pub const MAGIC: [u8; 4] = *b"GRVN";
26/// The only envelope format version this crate reads and writes.
27pub const FORMAT_VERSION: u8 = 1;
28/// Size of the fixed header preceding the payload.
29pub const HEADER_LEN: usize = 20;
30/// Size of the CRC32C trailer following the payload.
31pub const TRAILER_LEN: usize = 4;
32
33/// FNV-1a 64-bit hash of a fully qualified type name, e.g. `"bank.Transaction"`.
34pub const fn schema_id(qualified_name: &str) -> u64 {
35    let b = qualified_name.as_bytes();
36    let mut h: u64 = 0xcbf29ce484222325;
37    let mut i = 0;
38    while i < b.len() {
39        h ^= b[i] as u64;
40        h = h.wrapping_mul(0x100000001b3);
41        i += 1;
42    }
43    h
44}
45
46/// Appends an envelope around `payload` to `out`.
47///
48/// Fails with [`Error::TooLarge`] if `payload` is longer than `i32::MAX` bytes; `out` is
49/// left untouched in that case.
50pub fn encode(schema_id: u64, payload: &[u8], out: &mut BytesMut) -> Result<(), Error> {
51    if payload.len() as u64 > i32::MAX as u64 {
52        return Err(Error::TooLarge);
53    }
54    let start = out.len();
55    out.reserve(HEADER_LEN + payload.len() + TRAILER_LEN);
56    out.put_slice(&header(schema_id, payload.len() as u32));
57    out.put_slice(payload);
58    let crc = crc32c(&out[start..]);
59    out.put_u32_le(crc);
60    Ok(())
61}
62
63/// The 20-byte header for a payload of `payload_len` bytes. Shared by [`encode`] and the
64/// `Builder`, which writes it into the slot reserved before each message.
65#[inline]
66pub(crate) fn header(schema_id: u64, payload_len: u32) -> [u8; HEADER_LEN] {
67    let mut h = [0u8; HEADER_LEN];
68    h[..4].copy_from_slice(&MAGIC);
69    h[4] = FORMAT_VERSION;
70    // h[5] flags = 0, h[6..8] reserved = 0
71    h[8..16].copy_from_slice(&schema_id.to_le_bytes());
72    h[16..20].copy_from_slice(&payload_len.to_le_bytes());
73    h
74}
75
76/// A validated envelope borrowed from the input buffer.
77#[derive(Debug, Clone, Copy, PartialEq, Eq)]
78pub struct Envelope<'a> {
79    /// The root type's schema id.
80    pub schema_id: u64,
81    /// The raw Graven message.
82    pub payload: &'a [u8],
83    len: usize,
84}
85
86impl Envelope<'_> {
87    /// Total bytes this envelope occupied in the input (`24 + payload.len()`).
88    #[allow(clippy::len_without_is_empty)]
89    pub fn len(&self) -> usize {
90        self.len
91    }
92}
93
94/// Parses and validates the envelope at the start of `buf`.
95///
96/// Trailing bytes after the CRC are accepted; use [`Envelope::len`] to step to the next
97/// envelope. Checks run in this order: header length, magic, version, flags/reserved,
98/// payload length, CRC. [`Error::Incomplete`] means "more bytes needed" (e.g. a file cut
99/// by a crash); [`Error::ChecksumMismatch`] means a complete envelope whose CRC is wrong,
100/// i.e. real corruption. Never panics.
101pub fn decode(buf: &[u8]) -> Result<Envelope<'_>, Error> {
102    if buf.len() < HEADER_LEN {
103        return Err(Error::Incomplete { needed: HEADER_LEN });
104    }
105    if buf[..4] != MAGIC {
106        return Err(Error::BadMagic);
107    }
108    if buf[4] != FORMAT_VERSION {
109        return Err(Error::UnsupportedVersion(buf[4]));
110    }
111    if buf[5] != 0 || buf[6] != 0 || buf[7] != 0 {
112        return Err(Error::BadHeader);
113    }
114    let schema_id = u64::from_le_bytes(buf[8..16].try_into().unwrap());
115    let payload_len = u32::from_le_bytes(buf[16..20].try_into().unwrap());
116
117    // u64 cannot overflow here: 20 + u32::MAX + 4 < 2^33.
118    let needed64 = HEADER_LEN as u64 + payload_len as u64 + TRAILER_LEN as u64;
119    let total = match usize::try_from(needed64) {
120        Ok(t) if buf.len() >= t => t,
121        Ok(t) => return Err(Error::Incomplete { needed: t }),
122        Err(_) => return Err(Error::Incomplete { needed: usize::MAX }),
123    };
124
125    let crc_at = total - TRAILER_LEN;
126    let stored = u32::from_le_bytes(buf[crc_at..total].try_into().unwrap());
127    if crc32c(&buf[..crc_at]) != stored {
128        return Err(Error::ChecksumMismatch);
129    }
130    Ok(Envelope { schema_id, payload: &buf[HEADER_LEN..crc_at], len: total })
131}
132
133/// Iterates over envelopes concatenated in `buf` (e.g. a batch from
134/// `Builder::seal_envelope` … `take_frames`, or an append-only file).
135///
136/// Each item is [`decode`] of the remaining bytes. After the first `Err` (corruption or a
137/// truncated tail) the iterator yields `None`. An empty buffer yields nothing.
138#[inline]
139pub fn envelopes(buf: &[u8]) -> Envelopes<'_> {
140    Envelopes { buf, failed: false }
141}
142
143/// Iterator returned by [`envelopes`].
144#[derive(Debug, Clone)]
145pub struct Envelopes<'a> {
146    buf: &'a [u8],
147    failed: bool,
148}
149
150impl<'a> Iterator for Envelopes<'a> {
151    type Item = Result<Envelope<'a>, Error>;
152
153    #[inline]
154    fn next(&mut self) -> Option<Self::Item> {
155        if self.buf.is_empty() || self.failed {
156            return None;
157        }
158        match decode(self.buf) {
159            Ok(env) => {
160                self.buf = &self.buf[env.len..];
161                Some(Ok(env))
162            }
163            Err(e) => {
164                self.failed = true;
165                Some(Err(e))
166            }
167        }
168    }
169}
170
171impl FusedIterator for Envelopes<'_> {}
graven/envelope.rs

graven/
envelope.rs
