cu29_logstream/
record.rs

1//! Record encapsulation without content versions. Payloads require the producing
2//! application's matching decoder; this envelope does not select a content schema.
3
4use crate::{Error, Result};
5use alloc::vec::Vec;
6
7const RECORD_MAGIC: [u8; 4] = *b"CUSR";
8pub const RECORD_HEADER_LEN: usize = 45;
9const RECORD_DIGEST_OFFSET: usize = 13;
10
11/// Semantic record families carried by the log stream.
12#[derive(Clone, Copy, Debug, PartialEq, Eq)]
13#[repr(u8)]
14pub enum RecordKind {
15    Manifest = 0,
16    CopperList = 1,
17    KeyFrame = 2,
18    StructuredLog = 3,
19    Lifecycle = 4,
20    Gap = 5,
21    RecoveryPoint = 6,
22}
23
24impl TryFrom<u8> for RecordKind {
25    type Error = Error;
26
27    fn try_from(value: u8) -> Result<Self> {
28        match value {
29            0 => Ok(Self::Manifest),
30            1 => Ok(Self::CopperList),
31            2 => Ok(Self::KeyFrame),
32            3 => Ok(Self::StructuredLog),
33            4 => Ok(Self::Lifecycle),
34            5 => Ok(Self::Gap),
35            6 => Ok(Self::RecoveryPoint),
36            _ => Err(Error::UnknownRecordKind(value)),
37        }
38    }
39}
40
41/// A verified semantic record borrowing its recovered payload.
42#[derive(Clone, Copy, Debug, PartialEq, Eq)]
43pub struct DecodedRecord<'a> {
44    pub kind: RecordKind,
45    pub object_id: u64,
46    /// Digest binding the record kind, identity, length, and payload.
47    pub digest: [u8; 32],
48    pub payload: &'a [u8],
49}
50
51/// Frame and digest one semantic payload before FEC is applied.
52pub fn encode_record(kind: RecordKind, object_id: u64, payload: &[u8]) -> Result<Vec<u8>> {
53    let payload_len = u64::try_from(payload.len())
54        .map_err(|_| Error::InvalidConfig("record payload length exceeds u64"))?;
55    let capacity = RECORD_HEADER_LEN
56        .checked_add(payload.len())
57        .ok_or(Error::InvalidConfig("record length overflow"))?;
58    let mut record = Vec::with_capacity(capacity);
59    record.extend_from_slice(&RECORD_MAGIC);
60    record.push(kind as u8);
61    record.extend_from_slice(&object_id.to_be_bytes());
62    record.extend_from_slice(&[0_u8; 32]);
63    record.extend_from_slice(payload);
64
65    let digest = record_digest(kind, object_id, payload_len, payload);
66    record[RECORD_DIGEST_OFFSET..RECORD_HEADER_LEN].copy_from_slice(digest.as_bytes());
67    Ok(record)
68}
69
70/// Writes a semantic-record header for a payload already stored immediately
71/// after `header` in the caller's record buffer.
72pub(crate) fn encode_record_header(
73    kind: RecordKind,
74    object_id: u64,
75    payload: &[u8],
76    header: &mut [u8],
77) -> Result<()> {
78    if header.len() < RECORD_HEADER_LEN {
79        return Err(Error::BufferTooSmall {
80            needed: RECORD_HEADER_LEN,
81            available: header.len(),
82        });
83    }
84    let payload_len = u64::try_from(payload.len())
85        .map_err(|_| Error::InvalidConfig("record payload length exceeds u64"))?;
86    let header = &mut header[..RECORD_HEADER_LEN];
87    header.fill(0);
88    header[..4].copy_from_slice(&RECORD_MAGIC);
89    header[4] = kind as u8;
90    header[5..13].copy_from_slice(&object_id.to_be_bytes());
91    let digest = record_digest(kind, object_id, payload_len, payload);
92    header[RECORD_DIGEST_OFFSET..RECORD_HEADER_LEN].copy_from_slice(digest.as_bytes());
93    Ok(())
94}
95
96/// Validates and decodes one complete semantic record envelope.
97///
98/// The payload occupies the remainder of the reassembled record. Its derived
99/// length remains part of the digest input, binding the exact payload extent.
100pub fn decode_record(record: &[u8]) -> Result<DecodedRecord<'_>> {
101    if record.len() < RECORD_HEADER_LEN {
102        return Err(Error::TruncatedRecord);
103    }
104    if record[..4] != RECORD_MAGIC {
105        return Err(Error::InvalidMagic);
106    }
107    let kind = RecordKind::try_from(record[4])?;
108    let object_id = u64::from_be_bytes(record[5..13].try_into().unwrap());
109    let payload = &record[RECORD_HEADER_LEN..];
110    let payload_len = u64::try_from(payload.len())
111        .map_err(|_| Error::InvalidConfig("record payload length exceeds u64"))?;
112    let expected_digest = record_digest(kind, object_id, payload_len, payload);
113    if record[RECORD_DIGEST_OFFSET..RECORD_HEADER_LEN] != expected_digest.as_bytes()[..] {
114        return Err(Error::RecordDigestMismatch);
115    }
116    Ok(DecodedRecord {
117        kind,
118        object_id,
119        digest: *expected_digest.as_bytes(),
120        payload,
121    })
122}
123
124fn record_digest(
125    kind: RecordKind,
126    object_id: u64,
127    payload_len: u64,
128    payload: &[u8],
129) -> blake3::Hash {
130    let mut hasher = blake3::Hasher::new();
131    hasher.update(&[kind as u8]);
132    hasher.update(&object_id.to_be_bytes());
133    hasher.update(&payload_len.to_be_bytes());
134    hasher.update(payload);
135    hasher.finalize()
136}
137
138#[cfg(test)]
139mod tests {
140    use super::*;
141
142    #[test]
143    fn packed_record_header_contains_only_framing_fields() {
144        let record = encode_record(RecordKind::CopperList, 42, &[42]).unwrap();
145        assert_eq!(record.len(), 46);
146        assert_eq!(&record[..5], b"CUSR\x01");
147        assert_eq!(&record[5..13], &42_u64.to_be_bytes());
148        assert_eq!(record[45], 42);
149        assert_eq!(decode_record(&record).unwrap().payload, &[42]);
150        let mut header = [0xaa; RECORD_HEADER_LEN];
151        encode_record_header(RecordKind::CopperList, 42, &[42], &mut header).unwrap();
152        assert_eq!(header, record[..RECORD_HEADER_LEN]);
153    }
154
155    #[test]
156    fn record_extent_defines_payload_length_without_changing_digest_binding() {
157        for kind in [
158            RecordKind::Manifest,
159            RecordKind::CopperList,
160            RecordKind::KeyFrame,
161            RecordKind::StructuredLog,
162            RecordKind::Lifecycle,
163            RecordKind::Gap,
164            RecordKind::RecoveryPoint,
165        ] {
166            for object_id in [0, 42, u64::MAX] {
167                for payload_len in [0, 1, 95, 96, 4096] {
168                    let payload = alloc::vec![0xa5; payload_len];
169                    let encoded = encode_record(kind, object_id, &payload).unwrap();
170                    assert_eq!(encoded.len(), 45 + payload_len);
171                    let decoded = decode_record(&encoded).unwrap();
172                    assert_eq!(decoded.kind, kind);
173                    assert_eq!(decoded.object_id, object_id);
174                    assert_eq!(decoded.payload, payload);
175                    assert_eq!(decoded.payload.as_ptr(), encoded[45..].as_ptr());
176
177                    // Keep the original digest preimage, including the omitted length.
178                    let mut preimage = alloc::vec![kind as u8];
179                    preimage.extend_from_slice(&object_id.to_be_bytes());
180                    preimage.extend_from_slice(&(payload_len as u64).to_be_bytes());
181                    preimage.extend_from_slice(&payload);
182                    let digest = blake3::hash(&preimage);
183                    assert_eq!(decoded.digest, *digest.as_bytes());
184                    assert_eq!(&encoded[13..45], digest.as_bytes());
185
186                    let mut header = [0xaa; RECORD_HEADER_LEN + 1];
187                    encode_record_header(kind, object_id, &payload, &mut header).unwrap();
188                    assert_eq!(&header[..RECORD_HEADER_LEN], &encoded[..45]);
189                    assert_eq!(header[RECORD_HEADER_LEN], 0xaa);
190                }
191            }
192        }
193    }
194
195    #[test]
196    fn record_rejects_truncation_appended_bytes_and_corruption() {
197        for payload in [b"".as_slice(), b"state"] {
198            let encoded = encode_record(RecordKind::KeyFrame, 42, payload).unwrap();
199            for end in 0..encoded.len() {
200                let expected = if end < RECORD_HEADER_LEN {
201                    Error::TruncatedRecord
202                } else {
203                    Error::RecordDigestMismatch
204                };
205                assert_eq!(decode_record(&encoded[..end]), Err(expected));
206            }
207            for extra in [b"\0".as_slice(), b"state", encoded.as_slice()] {
208                let mut extended = encoded.clone();
209                extended.extend_from_slice(extra);
210                assert_eq!(decode_record(&extended), Err(Error::RecordDigestMismatch));
211            }
212            for offset in 0..encoded.len() {
213                let mut corrupted = encoded.clone();
214                corrupted[offset] ^= 1; // Also changes KeyFrame to a valid StructuredLog kind.
215                let expected = if offset < 4 {
216                    Error::InvalidMagic
217                } else {
218                    Error::RecordDigestMismatch
219                };
220                assert_eq!(decode_record(&corrupted), Err(expected));
221            }
222            let mut invalid_kind = encoded;
223            invalid_kind[4] = u8::MAX;
224            assert_eq!(
225                decode_record(&invalid_kind),
226                Err(Error::UnknownRecordKind(u8::MAX))
227            );
228        }
229    }
230
231    #[test]
232    fn record_header_rejects_short_buffers_without_writing() {
233        for len in 0..RECORD_HEADER_LEN {
234            let mut header = [0xaa; RECORD_HEADER_LEN];
235            assert_eq!(
236                encode_record_header(RecordKind::CopperList, 42, b"state", &mut header[..len]),
237                Err(Error::BufferTooSmall {
238                    needed: RECORD_HEADER_LEN,
239                    available: len,
240                })
241            );
242            assert_eq!(header, [0xaa; RECORD_HEADER_LEN]);
243        }
244    }
245}
cu29_logstream/record.rs

cu29_logstream/
record.rs
