limnifs_core/
drop_record.rs

1//! Drop record (spec §3.3, `bit-level/31-drop-record.md`).
2//!
3//! One 48-byte descriptor per drop in a slab, locating the drop's
4//! bytes inside one of the slab's solid windows.
5
6use crate::cursor::ManifestCursor;
7use crate::error::CoreError;
8use crate::slab::SlabHeader;
9use limnifs_format::{DropId, Representation};
10
11/// Width of a single drop record on the wire.
12///
13/// v0.2: extended from 48 to 49 bytes by adding `dict_id` (1 byte)
14/// at the end. `dict_id` = 0xFF means "no dictionary"; 0..254
15/// references an entry in the `dictionary_section` manifest section.
16pub const DROP_RECORD_LEN: usize = 50;
17
18/// Sentinel `dict_id` meaning "no dictionary used for this drop".
19pub const NO_DICT: u8 = 0xFF;
20
21/// Default per-drop plaintext-size ceiling. The spec's writer pipeline
22/// typically produces drops in the 4–64 MiB range via `FastCDC`; larger
23/// values are rejected unless the manifest overrides.
24pub const DEFAULT_DROP_MAX_PLAINTEXT_BYTES: u64 = 64 * 1024 * 1024;
25
26/// Parsed drop record.
27#[derive(Copy, Clone, Debug, Eq, PartialEq)]
28pub struct DropRecord {
29    pub drop_id: DropId,
30    pub plaintext_len: u32,
31    pub representation: Representation,
32    pub solid_window_index: u8,
33    pub offset_in_window: u32,
34    pub len_in_window: u32,
35    /// Dictionary id for dict-aided decompression.
36    /// 0xFF = no dictionary; 0..254 = index into `dictionary_section`.
37    pub dict_id: u8,
38    /// Record flags (trailing byte). Bit meanings:
39    /// [`crate::seekable::DROP_FLAG_SEEKABLE`].
40    pub flags: u8,
41}
42
43/// Parse a single drop record from the cursor's current position.
44///
45/// Performs the self-contained checks: buffer length, plaintext-size
46/// ceiling, slab-vs-record cross-field consistency (representation's
47/// aead must be 0 in a plaintext slab; ec must be 0 in a no-EC slab),
48/// and the `offset_in_window + len_in_window` u32 overflow check.
49///
50/// Does NOT check `solid_window_index` against the slab's actual
51/// solid-window count (the count is only known after parsing every
52/// drop record in the slab). That cross-record check runs at the
53/// slab-walker layer.
54///
55/// # Errors
56///
57/// - [`CoreError::TooShort`] if the cursor has fewer than 48 bytes.
58/// - [`CoreError::Corrupt`] if `plaintext_len` exceeds `max_plaintext`,
59///   if the slab-vs-record AEAD/EC consistency rules are violated, or
60///   if `offset_in_window + len_in_window` overflows u32.
61pub fn parse_drop_record(
62    cursor: &mut ManifestCursor<'_>,
63    slab: &SlabHeader,
64) -> Result<DropRecord, CoreError> {
65    parse_drop_record_with_ceiling(cursor, slab, DEFAULT_DROP_MAX_PLAINTEXT_BYTES)
66}
67
68/// Same as [`parse_drop_record`] but with a caller-supplied ceiling
69/// on `plaintext_len`. Used by readers that have parsed a manifest
70/// with a non-default drop-size parameter.
71///
72/// # Errors
73///
74/// Inherits all errors from [`parse_drop_record`].
75pub fn parse_drop_record_with_ceiling(
76    cursor: &mut ManifestCursor<'_>,
77    slab: &SlabHeader,
78    max_plaintext: u64,
79) -> Result<DropRecord, CoreError> {
80    let drop_id_bytes = cursor.read_n(32)?;
81    let mut drop_id_array = [0u8; 32];
82    drop_id_array.copy_from_slice(drop_id_bytes);
83    let drop_id = DropId::from_bytes(drop_id_array);
84
85    let plaintext_len = cursor.read_u32_le()?;
86    if u64::from(plaintext_len) > max_plaintext {
87        return Err(CoreError::Corrupt {
88            reason: format!("drop plaintext_len {plaintext_len} exceeds ceiling {max_plaintext}"),
89        });
90    }
91    let repr_bytes = cursor.read_n(3)?;
92    let representation = Representation::from_bytes([repr_bytes[0], repr_bytes[1], repr_bytes[2]]);
93    if !slab.is_sealed() && representation.aead != 0x00 {
94        return Err(CoreError::Corrupt {
95            reason: format!(
96                "drop record declares aead=0x{:02X} but slab is plaintext (crypto_hint=0)",
97                representation.aead
98            ),
99        });
100    }
101    if !slab.has_erasure_coding() && representation.ec != 0x00 {
102        return Err(CoreError::Corrupt {
103            reason: format!(
104                "drop record declares ec=0x{:02X} but slab has no EC (ec_descriptor=0)",
105                representation.ec
106            ),
107        });
108    }
109    let solid_window_index = cursor.read_u8()?;
110    let offset_in_window = cursor.read_u32_le()?;
111    let len_in_window = cursor.read_u32_le()?;
112    if offset_in_window.checked_add(len_in_window).is_none() {
113        return Err(CoreError::Corrupt {
114            reason: format!(
115                "drop record offset_in_window {offset_in_window} + len_in_window {len_in_window} overflows u32"
116            ),
117        });
118    }
119    let dict_id = cursor.read_u8()?;
120    let flags = cursor.read_u8()?;
121    Ok(DropRecord {
122        drop_id,
123        plaintext_len,
124        representation,
125        solid_window_index,
126        offset_in_window,
127        len_in_window,
128        dict_id,
129        flags,
130    })
131}
132
133#[cfg(test)]
134mod tests {
135    use super::*;
136    use crate::slab::parse_slab_header;
137    use limnifs_format::{SlabId, SLAB_MAGIC};
138
139    fn make_plaintext_slab_header() -> SlabHeader {
140        SlabHeader {
141            format_version: 1,
142            slab_id: SlabId::new(0, [0; 32]),
143            total_length: 4096,
144            ec_descriptor: 0x00,
145            crypto_hint: 0x00,
146        }
147    }
148
149    fn make_sealed_ec_slab_header() -> SlabHeader {
150        SlabHeader {
151            format_version: 1,
152            slab_id: SlabId::new(0, [0; 32]),
153            total_length: 16384,
154            ec_descriptor: 0x01,
155            crypto_hint: 0x01,
156        }
157    }
158
159    fn make_drop_record_bytes(record: &DropRecord) -> [u8; DROP_RECORD_LEN] {
160        let mut bytes = [0u8; DROP_RECORD_LEN];
161        bytes[..32].copy_from_slice(record.drop_id.as_bytes());
162        bytes[32..36].copy_from_slice(&record.plaintext_len.to_le_bytes());
163        bytes[36..39].copy_from_slice(&record.representation.to_bytes());
164        bytes[39] = record.solid_window_index;
165        bytes[40..44].copy_from_slice(&record.offset_in_window.to_le_bytes());
166        bytes[44..48].copy_from_slice(&record.len_in_window.to_le_bytes());
167        bytes[48] = record.dict_id;
168        bytes[49] = record.flags;
169        bytes
170    }
171
172    fn sample_record() -> DropRecord {
173        DropRecord {
174            drop_id: DropId::from_bytes([0x55; 32]),
175            plaintext_len: 1024,
176            representation: Representation::STORE_PLAINTEXT,
177            solid_window_index: 0,
178            offset_in_window: 0,
179            len_in_window: 1024,
180            dict_id: NO_DICT,
181            flags: 0,
182        }
183    }
184
185    #[test]
186    fn parses_store_plaintext_drop() {
187        let slab = make_plaintext_slab_header();
188        let record = sample_record();
189        let bytes = make_drop_record_bytes(&record);
190        let mut cursor = ManifestCursor::new(&bytes);
191        let parsed = parse_drop_record(&mut cursor, &slab).expect("plaintext drop parses");
192        assert_eq!(parsed, record);
193        assert_eq!(cursor.position(), DROP_RECORD_LEN);
194    }
195
196    #[test]
197    fn parses_lz4_sealed_drop() {
198        let slab = make_sealed_ec_slab_header();
199        let record = DropRecord {
200            drop_id: DropId::from_bytes([0x77; 32]),
201            plaintext_len: 4096,
202            representation: Representation::new(0x01, 0x01, 0x01),
203            solid_window_index: 1,
204            offset_in_window: 2048,
205            len_in_window: 1024,
206            dict_id: NO_DICT,
207            flags: 0,
208        };
209        let bytes = make_drop_record_bytes(&record);
210        let mut cursor = ManifestCursor::new(&bytes);
211        let parsed = parse_drop_record(&mut cursor, &slab).expect("sealed drop parses");
212        assert_eq!(parsed, record);
213    }
214
215    #[test]
216    fn rejects_short_buffer() {
217        let slab = make_plaintext_slab_header();
218        let bytes = [0u8; 40];
219        let mut cursor = ManifestCursor::new(&bytes);
220        match parse_drop_record(&mut cursor, &slab) {
221            Err(CoreError::TooShort { .. }) => {}
222            other => panic!("expected TooShort, got {other:?}"),
223        }
224    }
225
226    #[test]
227    fn rejects_plaintext_len_above_ceiling() {
228        let slab = make_plaintext_slab_header();
229        let mut record = sample_record();
230        record.plaintext_len = u32::MAX;
231        let bytes = make_drop_record_bytes(&record);
232        let mut cursor = ManifestCursor::new(&bytes);
233        match parse_drop_record(&mut cursor, &slab) {
234            Err(CoreError::Corrupt { reason }) => {
235                assert!(reason.contains("ceiling"), "got: {reason}");
236            }
237            other => panic!("expected Corrupt, got {other:?}"),
238        }
239    }
240
241    #[test]
242    fn rejects_aead_in_plaintext_slab() {
243        let slab = make_plaintext_slab_header();
244        let mut record = sample_record();
245        record.representation = Representation::new(0x00, 0x01, 0x00);
246        let bytes = make_drop_record_bytes(&record);
247        let mut cursor = ManifestCursor::new(&bytes);
248        match parse_drop_record(&mut cursor, &slab) {
249            Err(CoreError::Corrupt { reason }) => {
250                assert!(reason.contains("aead"), "got: {reason}");
251                assert!(reason.contains("plaintext"));
252            }
253            other => panic!("expected Corrupt, got {other:?}"),
254        }
255    }
256
257    #[test]
258    fn rejects_ec_in_no_ec_slab() {
259        let slab = make_plaintext_slab_header();
260        let mut record = sample_record();
261        record.representation = Representation::new(0x00, 0x00, 0x01);
262        let bytes = make_drop_record_bytes(&record);
263        let mut cursor = ManifestCursor::new(&bytes);
264        match parse_drop_record(&mut cursor, &slab) {
265            Err(CoreError::Corrupt { reason }) => {
266                assert!(reason.contains("ec"), "got: {reason}");
267                assert!(reason.contains("no EC"));
268            }
269            other => panic!("expected Corrupt, got {other:?}"),
270        }
271    }
272
273    #[test]
274    fn rejects_offset_len_overflow() {
275        let slab = make_plaintext_slab_header();
276        let mut record = sample_record();
277        record.offset_in_window = u32::MAX;
278        record.len_in_window = 1;
279        let bytes = make_drop_record_bytes(&record);
280        let mut cursor = ManifestCursor::new(&bytes);
281        match parse_drop_record(&mut cursor, &slab) {
282            Err(CoreError::Corrupt { reason }) => {
283                assert!(reason.contains("overflows u32"), "got: {reason}");
284            }
285            other => panic!("expected Corrupt, got {other:?}"),
286        }
287    }
288
289    #[test]
290    fn parses_after_a_real_slab_header() {
291        let mut bytes = Vec::new();
292        let mut header_bytes = [0u8; 56];
293        header_bytes[..4].copy_from_slice(&SLAB_MAGIC);
294        header_bytes[4..6].copy_from_slice(&1u16.to_le_bytes());
295        // slab_id ordinal 7, hash 0xAA
296        header_bytes[6..14].copy_from_slice(&7u64.to_le_bytes());
297        for byte in &mut header_bytes[14..46] {
298            *byte = 0xAA;
299        }
300        header_bytes[46..54].copy_from_slice(&8192u64.to_le_bytes());
301        // ec_descriptor=0, crypto_hint=0
302        bytes.extend_from_slice(&header_bytes);
303
304        let record = sample_record();
305        bytes.extend_from_slice(&make_drop_record_bytes(&record));
306
307        let mut cursor = ManifestCursor::new(&bytes);
308        let slab_parsed = parse_slab_header(&mut cursor).expect("slab header parses");
309        let record_parsed = parse_drop_record(&mut cursor, &slab_parsed).expect("drop parses");
310        assert_eq!(record_parsed, record);
311        assert_eq!(cursor.position(), 56 + DROP_RECORD_LEN);
312    }
313
314    #[test]
315    fn parses_flags_byte() {
316        let slab = make_plaintext_slab_header();
317        let record = sample_record();
318        let mut bytes = Vec::with_capacity(DROP_RECORD_LEN);
319        bytes.extend_from_slice(&make_drop_record_bytes(&record));
320        let mut cursor = ManifestCursor::new(&bytes);
321        let parsed = parse_drop_record(&mut cursor, &slab).expect("drop parses");
322        assert_eq!(parsed, record);
323        assert_eq!(cursor.position(), DROP_RECORD_LEN);
324    }
325
326    #[test]
327    fn parses_seekable_flag() {
328        let slab = make_plaintext_slab_header();
329        let mut record = sample_record();
330        record.flags = crate::seekable::DROP_FLAG_SEEKABLE;
331        let bytes = make_drop_record_bytes(&record);
332        let mut cursor = ManifestCursor::new(&bytes);
333        let parsed = parse_drop_record(&mut cursor, &slab).expect("drop parses");
334        assert_eq!(
335            parsed.flags,
336            crate::seekable::DROP_FLAG_SEEKABLE,
337            "trailing flags byte is parsed"
338        );
339    }
340}
limnifs_core/drop_record.rs

limnifs_core/
drop_record.rs
