vole_document/container/
record.rs

1//! Length-delimited record framing.
2//!
3//! ```text
4//! record := tag:u8 flags:u8 reserved:u16=0 length:u32 payload:[u8;length] crc32c:u32
5//! ```
6//!
7//! The CRC covers the 8-byte record header and the payload. `reserved` must be
8//! zero. The `flags` bit [`FLAG_OPTIONAL`] marks a record whose *tag* a decoder
9//! may skip if unknown; unknown non-optional tags fail closed.
10
11use crate::error::{Error, Result};
12use crate::integrity::crc32c;
13use crate::limits::Limits;
14
15/// Bytes of per-record framing before the payload.
16pub const RECORD_HEADER_LEN: usize = 8;
17/// Bytes of per-record framing after the payload (the CRC).
18pub const RECORD_TRAILER_LEN: usize = 4;
19/// Total framing overhead per record.
20pub const RECORD_OVERHEAD: usize = RECORD_HEADER_LEN + RECORD_TRAILER_LEN;
21
22/// Flag: this record's tag may be skipped by a decoder that does not know it.
23pub const FLAG_OPTIONAL: u8 = 0x01;
24
25/// Known record classes.
26#[derive(Debug, Clone, Copy, PartialEq, Eq)]
27#[repr(u8)]
28pub enum RecordTag {
29    /// Universe declaration (opcode/coder/limit semantics version).
30    Universe = 0x01,
31    /// Source-format descriptor.
32    Format = 0x02,
33    /// A raw byte object.
34    Object = 0x10,
35    /// The reconstruction graph (DRA program).
36    Graph = 0x20,
37    /// An entropy model (Phase 2+).
38    Model = 0x30,
39    /// An entropy channel capsule (Phase 2+).
40    EntropyChannel = 0x40,
41    /// A typed residual stream (later phases).
42    Residual = 0x50,
43    /// A random-access checkpoint (later phases).
44    Checkpoint = 0x60,
45    /// A physical/logical index (later phases).
46    Index = 0x70,
47    /// A reference to an external content-addressed object (Phase 9+).
48    ExternalRef = 0x80,
49    /// Whole-source integrity manifest.
50    Integrity = 0xF0,
51    /// Terminal record.
52    Trailer = 0xFF,
53}
54
55impl RecordTag {
56    /// Map a raw tag byte to a known tag, if any.
57    pub const fn from_u8(b: u8) -> Option<RecordTag> {
58        match b {
59            0x01 => Some(RecordTag::Universe),
60            0x02 => Some(RecordTag::Format),
61            0x10 => Some(RecordTag::Object),
62            0x20 => Some(RecordTag::Graph),
63            0x30 => Some(RecordTag::Model),
64            0x40 => Some(RecordTag::EntropyChannel),
65            0x50 => Some(RecordTag::Residual),
66            0x60 => Some(RecordTag::Checkpoint),
67            0x70 => Some(RecordTag::Index),
68            0x80 => Some(RecordTag::ExternalRef),
69            0xF0 => Some(RecordTag::Integrity),
70            0xFF => Some(RecordTag::Trailer),
71            _ => None,
72        }
73    }
74
75    /// Stable short name for diagnostics and receipts.
76    pub const fn name(self) -> &'static str {
77        match self {
78            RecordTag::Universe => "UNIVERSE",
79            RecordTag::Format => "FORMAT",
80            RecordTag::Object => "OBJECT",
81            RecordTag::Graph => "GRAPH",
82            RecordTag::Model => "MODEL",
83            RecordTag::EntropyChannel => "ENTROPY_CHANNEL",
84            RecordTag::Residual => "RESIDUAL",
85            RecordTag::Checkpoint => "CHECKPOINT",
86            RecordTag::Index => "INDEX",
87            RecordTag::ExternalRef => "EXTERNAL_REF",
88            RecordTag::Integrity => "INTEGRITY",
89            RecordTag::Trailer => "TRAILER",
90        }
91    }
92}
93
94/// A single decoded record.
95#[derive(Debug, Clone, PartialEq, Eq)]
96pub struct Record {
97    /// Raw tag byte (kept raw so unknown optional tags survive a reparse).
98    pub tag: u8,
99    /// Raw flags byte.
100    pub flags: u8,
101    /// Payload bytes.
102    pub payload: Vec<u8>,
103}
104
105impl Record {
106    /// Construct a record with the given tag byte and payload.
107    pub fn new(tag: u8, payload: Vec<u8>) -> Self {
108        Record {
109            tag,
110            flags: 0,
111            payload,
112        }
113    }
114
115    /// Construct a record for a known tag.
116    pub fn known(tag: RecordTag, payload: Vec<u8>) -> Self {
117        Record {
118            tag: tag as u8,
119            flags: 0,
120            payload,
121        }
122    }
123
124    /// Whether the optional-skip flag is set.
125    pub fn is_optional(&self) -> bool {
126        self.flags & FLAG_OPTIONAL != 0
127    }
128}
129
130/// Append an encoded record to `out`. `payload` must fit in `u32`.
131pub fn write_record(out: &mut Vec<u8>, tag: u8, flags: u8, payload: &[u8]) -> Result<()> {
132    let len = u32::try_from(payload.len())
133        .map_err(|_| Error::resource_limit("record payload exceeds 4 GiB"))?;
134    let start = out.len();
135    out.push(tag);
136    out.push(flags);
137    out.extend_from_slice(&0u16.to_le_bytes());
138    out.extend_from_slice(&len.to_le_bytes());
139    out.extend_from_slice(payload);
140    let crc = crc32c(&out[start..]);
141    out.extend_from_slice(&crc.to_le_bytes());
142    Ok(())
143}
144
145/// An iterator over the records of a byte slice.
146pub struct RecordReader<'a> {
147    data: &'a [u8],
148    pos: usize,
149    limits: Limits,
150    count: u32,
151}
152
153impl<'a> RecordReader<'a> {
154    /// Create a reader starting at `offset` within `data`.
155    pub fn new(data: &'a [u8], offset: usize, limits: Limits) -> Self {
156        RecordReader {
157            data,
158            pos: offset,
159            limits,
160            count: 0,
161        }
162    }
163
164    /// Current byte offset of the next record header.
165    pub fn position(&self) -> usize {
166        self.pos
167    }
168
169    /// Records successfully read so far.
170    pub fn records_read(&self) -> u32 {
171        self.count
172    }
173
174    /// Read the next record, or `None` at end of input.
175    pub fn next_record(&mut self) -> Result<Option<Record>> {
176        if self.pos == self.data.len() {
177            return Ok(None);
178        }
179        if self.count >= self.limits.max_record_count {
180            return Err(Error::resource_limit("record count limit exceeded"));
181        }
182        let remaining = self.data.len() - self.pos;
183        if remaining < RECORD_OVERHEAD {
184            return Err(Error::invalid_container(format!(
185                "truncated record header: {remaining} bytes remain"
186            )));
187        }
188        let hdr = &self.data[self.pos..self.pos + RECORD_HEADER_LEN];
189        let tag = hdr[0];
190        let flags = hdr[1];
191        let reserved = u16::from_le_bytes([hdr[2], hdr[3]]);
192        if reserved != 0 {
193            return Err(Error::invalid_container(
194                "record reserved field must be zero",
195            ));
196        }
197        let len = u32::from_le_bytes([hdr[4], hdr[5], hdr[6], hdr[7]]);
198        if len > self.limits.max_record_len {
199            return Err(Error::resource_limit(format!(
200                "record payload length {len} exceeds limit {}",
201                self.limits.max_record_len
202            )));
203        }
204        let total = RECORD_HEADER_LEN
205            .checked_add(len as usize)
206            .and_then(|n| n.checked_add(RECORD_TRAILER_LEN))
207            .ok_or_else(|| Error::invalid_container("record length overflow"))?;
208        if total > remaining {
209            return Err(Error::invalid_container(format!(
210                "truncated record payload: need {total}, have {remaining}"
211            )));
212        }
213        let body = &self.data[self.pos..self.pos + RECORD_HEADER_LEN + len as usize];
214        let want_crc = u32::from_le_bytes([
215            self.data[self.pos + RECORD_HEADER_LEN + len as usize],
216            self.data[self.pos + RECORD_HEADER_LEN + len as usize + 1],
217            self.data[self.pos + RECORD_HEADER_LEN + len as usize + 2],
218            self.data[self.pos + RECORD_HEADER_LEN + len as usize + 3],
219        ]);
220        let got_crc = crc32c(body);
221        if want_crc != got_crc {
222            return Err(Error::invalid_container(format!(
223                "record CRC32C mismatch at offset {}: declared {want_crc:#010x}, computed {got_crc:#010x}",
224                self.pos
225            )));
226        }
227        let payload = self.data
228            [self.pos + RECORD_HEADER_LEN..self.pos + RECORD_HEADER_LEN + len as usize]
229            .to_vec();
230        self.pos += total;
231        self.count += 1;
232        Ok(Some(Record {
233            tag,
234            flags,
235            payload,
236        }))
237    }
238}
239
240#[cfg(test)]
241mod tests {
242    use super::*;
243
244    #[test]
245    fn write_read_roundtrip() {
246        let mut buf = Vec::new();
247        write_record(&mut buf, RecordTag::Universe as u8, 0, b"hello").unwrap();
248        write_record(&mut buf, RecordTag::Object as u8, 0, &[1, 2, 3]).unwrap();
249        let mut r = RecordReader::new(&buf, 0, Limits::DEFAULT);
250        let a = r.next_record().unwrap().unwrap();
251        assert_eq!(a.tag, RecordTag::Universe as u8);
252        assert_eq!(a.payload, b"hello");
253        let b = r.next_record().unwrap().unwrap();
254        assert_eq!(b.tag, RecordTag::Object as u8);
255        assert_eq!(b.payload, vec![1, 2, 3]);
256        assert!(r.next_record().unwrap().is_none());
257    }
258
259    #[test]
260    fn detects_payload_corruption() {
261        let mut buf = Vec::new();
262        write_record(&mut buf, 0x10, 0, b"abcdef").unwrap();
263        buf[9] ^= 0xFF; // corrupt a payload byte
264        let mut r = RecordReader::new(&buf, 0, Limits::DEFAULT);
265        let e = r.next_record().unwrap_err();
266        assert_eq!(e.class(), crate::ErrorClass::InvalidContainer);
267    }
268
269    #[test]
270    fn detects_truncation() {
271        let mut buf = Vec::new();
272        write_record(&mut buf, 0x10, 0, b"abcdef").unwrap();
273        buf.truncate(buf.len() - 2);
274        let mut r = RecordReader::new(&buf, 0, Limits::DEFAULT);
275        let e = r.next_record().unwrap_err();
276        assert_eq!(e.class(), crate::ErrorClass::InvalidContainer);
277    }
278
279    #[test]
280    fn rejects_nonzero_reserved() {
281        let mut buf = Vec::new();
282        write_record(&mut buf, 0x10, 0, b"x").unwrap();
283        buf[2] = 1; // reserved
284        let mut r = RecordReader::new(&buf, 0, Limits::DEFAULT);
285        let e = r.next_record().unwrap_err();
286        assert_eq!(e.class(), crate::ErrorClass::InvalidContainer);
287    }
288
289    #[test]
290    fn enforces_record_length_limit() {
291        let mut buf = Vec::new();
292        write_record(&mut buf, 0x10, 0, &[0u8; 100]).unwrap();
293        let limits = Limits {
294            max_record_len: 10,
295            ..Limits::DEFAULT
296        };
297        let mut r = RecordReader::new(&buf, 0, limits);
298        let e = r.next_record().unwrap_err();
299        assert_eq!(e.class(), crate::ErrorClass::ResourceLimit);
300    }
301}
vole_document/container/record.rs

vole_document/container/
record.rs
