Skip to main content

loonfs_api/
sst_blocks.rs

1//! Block-granular encoding for metadata SST and derived-index segments.
2//!
3//! A segment object is a sequence of independently readable sections:
4//! data blocks, then one filter block, then one index block. There is no
5//! footer — the manifest's segment descriptor carries the index and filter
6//! handles, so the descriptor is the only entry point into the object.
7//! Readers fetch the byte range a handle names, verify its CRC32C, and
8//! decode just that section; nothing here performs IO.
9//!
10//! The block grammar is row-payload-agnostic: the builder and decoders
11//! carry any CBOR row type, and the segment's descriptor family says which
12//! one to expect — [`MetadataRow`] for metadata tables, `IndexRow` for gram
13//! index segments. The section framing, key compression, filter hashing,
14//! and checksums are identical either way.
15//!
16//! Durable layout, frozen by this module:
17//!
18//! - A **data block** holds prefix-compressed entries: each entry stores
19//!   `(shared_prefix_len, key_suffix_len)` as LEB128 varints, the key
20//!   suffix bytes, then the row as a CBOR-encoded payload length-
21//!   prefixed with a varint. Every [`RESTART_INTERVAL`]th entry is a
22//!   restart point storing its full key (shared prefix length zero). The
23//!   block ends with the restart offsets as little-endian `u32`s and their
24//!   count. The block payload is zstd-compressed.
25//! - The **index block** is a zstd-compressed CBOR list with one entry per
26//!   data block: the block's last row key and its [`BlockHandle`].
27//! - The **filter block** is a bloom filter over caller-chosen filter keys
28//!   (per-family lookup prefixes): `n_hashes` as a little-endian `u32`,
29//!   the bit length as a little-endian `u64`, then the bit bytes. Filter
30//!   bits do not compress, so the payload is stored raw.
31//! - Every section's CRC32C is computed over its stored bytes and lives in
32//!   the handle that names it (index entries for data blocks; the segment
33//!   descriptor for the index and filter), never inside the section.
34//! - Bloom hashing is two xxh64 passes with fixed seeds combined by double
35//!   hashing. The seeds, like the CRC and hash algorithm choices, are
36//!   frozen durable-format constants.
37
38use crate::wire::manifest::MetadataRow;
39use serde::{Deserialize, Serialize};
40use std::io::Read;
41use std::num::NonZeroUsize;
42use thiserror::Error;
43use xxhash_rust::xxh64::xxh64;
44
45/// Target uncompressed size of one data block, in bytes. Sized for direct
46/// object-store reads: request round-trips dominate transfer time at this
47/// scale, so bulk read paths (directory listings read most rows of several
48/// families) want few large ranged GETs, and a lookup fetching one block
49/// still moves trivial bytes. Benchmarked over 8 KiB, which priced a full
50/// listing at one GET per tiny block.
51pub const DEFAULT_TARGET_BLOCK_BYTES: usize = 64 * 1024;
52/// Entries between restart points inside a data block.
53pub const RESTART_INTERVAL: usize = 16;
54/// Bloom filter sizing: bits reserved per inserted filter key.
55pub const FILTER_BITS_PER_KEY: usize = 10;
56/// Bloom filter probe count, chosen for [`FILTER_BITS_PER_KEY`].
57pub const FILTER_HASH_COUNT: u32 = 7;
58
59const FILTER_HASH_SEED_ONE: u64 = 0;
60const FILTER_HASH_SEED_TWO: u64 = 0x9e37_79b9_7f4a_7c15;
61/// One compression level for every zstd-compressed durable artifact (SST
62/// blocks and WAL segment envelopes). 3 is also the library default, so
63/// this pins in a name what an implicit `0` would choose silently.
64pub(crate) const ZSTD_LEVEL: i32 = 3;
65
66/// Where one stored section lives inside a segment object, and how to
67/// verify it: the CRC32C of the stored bytes and their decoded length.
68#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
69pub struct BlockHandle {
70    /// Zero-based byte offset of the section within its immutable segment object.
71    pub offset: u64,
72    /// Number of bytes to range-read and checksum before decoding.
73    pub stored_len: u32,
74    /// Expected byte length after optional section decompression.
75    pub decoded_len: u32,
76    /// CRC32C over the exact `stored_len` bytes at `offset`.
77    pub crc32c: u32,
78}
79
80/// One index entry: the last row key of a data block plus its handle.
81#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
82pub struct SegmentIndexEntry {
83    /// Greatest row key in `block`, used to binary-search candidate blocks.
84    pub last_key: String,
85    /// Data-section location and integrity metadata.
86    pub block: BlockHandle,
87}
88
89/// A finished segment: the object bytes plus everything the manifest
90/// descriptor must carry to read them back.
91#[derive(Debug, Clone, PartialEq, Eq)]
92pub struct BuiltSegmentBlocks {
93    /// Complete immutable object body, with data sections followed by filter and index sections.
94    pub bytes: Vec<u8>,
95    /// Handle callers persist in the segment descriptor to bootstrap reads.
96    pub index: BlockHandle,
97    /// Handle callers persist for negative point-lookup filtering.
98    pub filter: BlockHandle,
99    /// Number of rows accepted by the builder, including adjacent duplicate keys.
100    pub row_count: u64,
101    /// Least row key in the non-empty segment.
102    pub min_key: String,
103    /// Greatest row key in the non-empty segment.
104    pub max_key: String,
105}
106
107/// One decoded data block: row keys and rows, parallel and in key order.
108/// The row type defaults to [`MetadataRow`]; index segments decode their
109/// own row payload through [`decode_data_block_rows`].
110#[derive(Debug, Clone, PartialEq, Eq)]
111pub struct DecodedDataBlock<R = MetadataRow> {
112    /// Reconstructed row keys in the same ascending order as `rows`.
113    pub row_keys: Vec<String>,
114    /// Decoded row payloads positionally paired with `row_keys`.
115    pub rows: Vec<R>,
116}
117
118/// A decoded bloom filter; answers "definitely absent" or "maybe present".
119#[derive(Debug, Clone, PartialEq, Eq)]
120pub struct SegmentFilter {
121    n_hashes: u32,
122    bit_len: u64,
123    bits: Vec<u8>,
124}
125
126/// Describes a violation encountered while building or validating an SST section.
127///
128/// See [metadata segments](../../../docs/specs/format.md#421-metadata-segments).
129#[derive(Debug, Clone, PartialEq, Eq, Error)]
130pub enum SstBlockCodecError {
131    /// Reports a request to finish a segment before any row was supplied.
132    #[error("segment must contain at least one row")]
133    EmptySegment,
134    /// Reports a builder input that would violate durable ascending row-key order.
135    #[error("row key `{offered}` is not in ascending order after `{previous}`")]
136    RowKeysOutOfOrder {
137        /// Last key the builder accepted.
138        previous: String,
139        /// Descending key rejected by the builder.
140        offered: String,
141    },
142    /// Reports a range-read body whose byte count disagrees with its handle.
143    #[error("stored bytes length {actual} does not match handle length {expected}")]
144    StoredLengthMismatch {
145        /// Stored byte count recorded in the persisted `BlockHandle`.
146        expected: u32,
147        /// Byte count returned to the decoder.
148        actual: usize,
149    },
150    /// Reports stored section bytes that fail the CRC32C recorded in their handle.
151    #[error("block checksum mismatch: expected {expected:#010x}, actual {actual:#010x}")]
152    ChecksumMismatch {
153        /// CRC32C recorded in the persisted `BlockHandle`.
154        expected: u32,
155        /// CRC32C recomputed from the supplied stored bytes.
156        actual: u32,
157    },
158    /// Reports a section whose decompressed size disagrees with its handle.
159    #[error("decoded length {actual} does not match handle length {expected}")]
160    DecodedLengthMismatch {
161        /// Decoded byte count recorded in the persisted `BlockHandle`.
162        expected: u32,
163        /// Byte count produced by section decompression.
164        actual: usize,
165    },
166    /// Reports structurally invalid framing, ordering, UTF-8, or filter metadata.
167    #[error("malformed block: {0}")]
168    Malformed(String),
169    /// Reports a CBOR or zstd failure while encoding or decoding a section.
170    #[error("block codec error: {0}")]
171    Codec(String),
172}
173
174/// Builds one segment's blocks from rows fed in ascending row-key order.
175#[derive(Debug)]
176pub struct SegmentBlocksBuilder {
177    target_block_bytes: usize,
178    entries: Vec<u8>,
179    restarts: Vec<u32>,
180    entry_count: usize,
181    previous_key: String,
182    block_first_key: String,
183    finished_blocks: Vec<(String, Vec<u8>)>,
184    filter_hashes: Vec<(u64, u64)>,
185    row_count: u64,
186    min_key: String,
187}
188
189impl Default for SegmentBlocksBuilder {
190    fn default() -> Self {
191        Self::new(const { NonZeroUsize::new(DEFAULT_TARGET_BLOCK_BYTES).unwrap() })
192    }
193}
194
195impl SegmentBlocksBuilder {
196    /// Creates a builder that closes a data block after reaching the target decoded byte size.
197    pub fn new(target_block_bytes: NonZeroUsize) -> Self {
198        Self {
199            target_block_bytes: target_block_bytes.get(),
200            entries: Vec::new(),
201            restarts: Vec::new(),
202            entry_count: 0,
203            previous_key: String::new(),
204            block_first_key: String::new(),
205            finished_blocks: Vec::new(),
206            filter_hashes: Vec::new(),
207            row_count: 0,
208            min_key: String::new(),
209        }
210    }
211
212    /// Appends one row. `filter_key` is the lookup prefix point reads will
213    /// probe for this row; the caller derives it per family. The row is any
214    /// CBOR payload; a segment must hold one row type throughout, named by
215    /// the descriptor family that references it.
216    pub fn push<R: Serialize>(
217        &mut self,
218        row_key: &str,
219        filter_key: &str,
220        row: &R,
221    ) -> Result<(), SstBlockCodecError> {
222        if self.row_count > 0 && row_key < self.previous_key.as_str() {
223            return Err(SstBlockCodecError::RowKeysOutOfOrder {
224                previous: self.previous_key.clone(),
225                offered: row_key.to_owned(),
226            });
227        }
228        if self.row_count == 0 {
229            self.min_key = row_key.to_owned();
230        }
231        self.filter_hashes.push(filter_key_hashes(filter_key));
232
233        let restart = self.entry_count % RESTART_INTERVAL == 0;
234        if restart {
235            self.restarts.push(self.entries.len() as u32);
236        }
237        if self.entries.is_empty() {
238            self.block_first_key = row_key.to_owned();
239        }
240        let shared_len = if restart {
241            0
242        } else {
243            shared_prefix_len(&self.previous_key, row_key)
244        };
245        let suffix = &row_key.as_bytes()[shared_len..];
246        let mut row_bytes = Vec::new();
247        ciborium::ser::into_writer(row, &mut row_bytes)
248            .map_err(|error| SstBlockCodecError::Codec(error.to_string()))?;
249        write_varint(&mut self.entries, shared_len as u64);
250        write_varint(&mut self.entries, suffix.len() as u64);
251        self.entries.extend_from_slice(suffix);
252        write_varint(&mut self.entries, row_bytes.len() as u64);
253        self.entries.extend_from_slice(&row_bytes);
254
255        self.entry_count += 1;
256        self.row_count += 1;
257        self.previous_key.clear();
258        self.previous_key.push_str(row_key);
259        if self.entries.len() >= self.target_block_bytes {
260            self.finish_data_block();
261        }
262        Ok(())
263    }
264
265    fn finish_data_block(&mut self) {
266        if self.entries.is_empty() {
267            return;
268        }
269        let mut payload = std::mem::take(&mut self.entries);
270        for restart in &self.restarts {
271            payload.extend_from_slice(&restart.to_le_bytes());
272        }
273        payload.extend_from_slice(&(self.restarts.len() as u32).to_le_bytes());
274        self.restarts.clear();
275        self.entry_count = 0;
276        self.finished_blocks
277            .push((std::mem::take(&mut self.previous_key), payload));
278    }
279
280    /// Encodes the remaining rows and assembles the object bytes.
281    pub fn finish(mut self) -> Result<BuiltSegmentBlocks, SstBlockCodecError> {
282        if self.row_count == 0 {
283            return Err(SstBlockCodecError::EmptySegment);
284        }
285        let max_key = self.previous_key.clone();
286        self.finish_data_block();
287
288        let mut bytes = Vec::new();
289        let mut index = Vec::with_capacity(self.finished_blocks.len());
290        for (last_key, payload) in std::mem::take(&mut self.finished_blocks) {
291            let block = append_section(&mut bytes, &payload, true)?;
292            index.push(SegmentIndexEntry { last_key, block });
293        }
294
295        let filter_payload = build_filter_payload(&self.filter_hashes);
296        let filter = append_section(&mut bytes, &filter_payload, false)?;
297
298        let mut index_payload = Vec::new();
299        ciborium::ser::into_writer(&index, &mut index_payload)
300            .map_err(|error| SstBlockCodecError::Codec(error.to_string()))?;
301        let index = append_section(&mut bytes, &index_payload, true)?;
302
303        Ok(BuiltSegmentBlocks {
304            bytes,
305            index,
306            filter,
307            row_count: self.row_count,
308            min_key: self.min_key,
309            max_key,
310        })
311    }
312}
313
314/// Decodes the index block from exactly the bytes its handle names.
315pub fn decode_index_block(
316    stored: &[u8],
317    handle: &BlockHandle,
318) -> Result<Vec<SegmentIndexEntry>, SstBlockCodecError> {
319    let payload = decode_section(stored, handle, true)?;
320    let entries: Vec<SegmentIndexEntry> = ciborium::de::from_reader(payload.as_slice())
321        .map_err(|error| SstBlockCodecError::Codec(error.to_string()))?;
322    // Ascending block order is a format requirement; key-range narrowing
323    // binary-searches the last keys, so an out-of-order index is malformed.
324    if let Some(pair) = entries
325        .windows(2)
326        .find(|pair| pair[0].last_key > pair[1].last_key)
327    {
328        return Err(SstBlockCodecError::Malformed(format!(
329            "index blocks out of key order: `{}` follows `{}`",
330            pair[1].last_key, pair[0].last_key
331        )));
332    }
333    Ok(entries)
334}
335
336/// Decodes one data block from exactly the bytes its handle names.
337pub fn decode_data_block(
338    stored: &[u8],
339    handle: &BlockHandle,
340) -> Result<DecodedDataBlock, SstBlockCodecError> {
341    decode_data_block_rows::<MetadataRow>(stored, handle)
342}
343
344/// Decodes one data block whose rows are `R`, for segment families whose
345/// row payload is not [`MetadataRow`] (gram index segments).
346pub fn decode_data_block_rows<R: serde::de::DeserializeOwned>(
347    stored: &[u8],
348    handle: &BlockHandle,
349) -> Result<DecodedDataBlock<R>, SstBlockCodecError> {
350    let payload = decode_section(stored, handle, true)?;
351    if payload.len() < 4 {
352        return Err(SstBlockCodecError::Malformed(
353            "data block shorter than its restart count".to_owned(),
354        ));
355    }
356    let (body, restart_count_bytes) = payload.split_at(payload.len() - 4);
357    let restart_count = u32::from_le_bytes(
358        restart_count_bytes
359            .try_into()
360            .expect("split_at should leave exactly four bytes"),
361    ) as usize;
362    let restarts_len = restart_count
363        .checked_mul(4)
364        .filter(|len| *len <= body.len())
365        .ok_or_else(|| SstBlockCodecError::Malformed("restart array exceeds block".to_owned()))?;
366    let entries = &body[..body.len() - restarts_len];
367
368    let mut row_keys = Vec::new();
369    let mut rows = Vec::new();
370    let mut cursor = 0usize;
371    let mut previous_key = String::new();
372    while cursor < entries.len() {
373        let shared_len = read_varint(entries, &mut cursor)? as usize;
374        let suffix_len = read_varint(entries, &mut cursor)? as usize;
375        if shared_len > previous_key.len() {
376            return Err(SstBlockCodecError::Malformed(
377                "shared prefix exceeds previous key".to_owned(),
378            ));
379        }
380        let suffix = take_slice(entries, &mut cursor, suffix_len)?;
381        let suffix = std::str::from_utf8(suffix)
382            .map_err(|_| SstBlockCodecError::Malformed("row key is not utf-8".to_owned()))?;
383        let mut key = String::with_capacity(shared_len + suffix.len());
384        key.push_str(&previous_key[..shared_len]);
385        key.push_str(suffix);
386        let row_len = read_varint(entries, &mut cursor)? as usize;
387        let row_bytes = take_slice(entries, &mut cursor, row_len)?;
388        let row: R = ciborium::de::from_reader(row_bytes)
389            .map_err(|error| SstBlockCodecError::Codec(error.to_string()))?;
390        // Ascending row-key order is a format requirement; readers
391        // binary-search on it, so an out-of-order block is malformed.
392        if key.as_str() < previous_key.as_str() {
393            return Err(SstBlockCodecError::Malformed(format!(
394                "rows out of row-key order: `{key}` follows `{previous_key}`"
395            )));
396        }
397        previous_key.clear();
398        previous_key.push_str(&key);
399        row_keys.push(key);
400        rows.push(row);
401    }
402    Ok(DecodedDataBlock { row_keys, rows })
403}
404
405/// Decodes the filter block from exactly the bytes its handle names.
406pub fn decode_filter_block(
407    stored: &[u8],
408    handle: &BlockHandle,
409) -> Result<SegmentFilter, SstBlockCodecError> {
410    let payload = decode_section(stored, handle, false)?;
411    if payload.len() < 12 {
412        return Err(SstBlockCodecError::Malformed(
413            "filter block shorter than its header".to_owned(),
414        ));
415    }
416    let n_hashes = u32::from_le_bytes(
417        payload[0..4]
418            .try_into()
419            .expect("header length should be checked above"),
420    );
421    let bit_len = u64::from_le_bytes(
422        payload[4..12]
423            .try_into()
424            .expect("header length should be checked above"),
425    );
426    let bits = payload[12..].to_vec();
427    if bit_len.div_ceil(8) != bits.len() as u64 {
428        return Err(SstBlockCodecError::Malformed(
429            "filter bit length disagrees with its bytes".to_owned(),
430        ));
431    }
432    Ok(SegmentFilter {
433        n_hashes,
434        bit_len,
435        bits,
436    })
437}
438
439impl SegmentFilter {
440    /// False means no row with this filter key is in the segment; true
441    /// means one may be.
442    pub fn may_contain(&self, filter_key: &str) -> bool {
443        if self.bit_len == 0 {
444            return false;
445        }
446        let (h1, h2) = filter_key_hashes(filter_key);
447        for probe in 0..u64::from(self.n_hashes) {
448            let bit = h1.wrapping_add(probe.wrapping_mul(h2)) % self.bit_len;
449            let byte = self.bits[(bit / 8) as usize];
450            if byte & (1 << (bit % 8)) == 0 {
451                return false;
452            }
453        }
454        true
455    }
456}
457
458/// The exclusive upper bound for every row key beginning with `prefix`.
459///
460/// Row keys are ordered as byte strings, so a prefix scan is the range
461/// `[prefix, string_prefix_upper_bound(prefix))`. `None` means the prefix is
462/// all `0xff` bytes and nothing sorts above it, so the scan runs to the end.
463pub fn string_prefix_upper_bound(prefix: &str) -> Option<String> {
464    let mut bytes = prefix.as_bytes().to_vec();
465    for index in (0..bytes.len()).rev() {
466        if bytes[index] != u8::MAX {
467            bytes[index] += 1;
468            bytes.truncate(index + 1);
469            return String::from_utf8(bytes).ok();
470        }
471    }
472    None
473}
474
475/// Index positions of the blocks that can hold keys in
476/// `[lower_bound, upper_bound)`; `None` bounds the range at the last block.
477pub fn index_blocks_for_key_range(
478    index: &[SegmentIndexEntry],
479    lower_bound: &str,
480    upper_bound: Option<&str>,
481) -> std::ops::Range<usize> {
482    let start = index.partition_point(|entry| entry.last_key.as_str() < lower_bound);
483    let end = upper_bound.map_or(index.len(), |upper_bound| {
484        // A block whose last key equals the exclusive upper bound can still
485        // hold keys below it, so the bound block itself is included.
486        index
487            .partition_point(|entry| entry.last_key.as_str() < upper_bound)
488            .saturating_add(1)
489            .min(index.len())
490    });
491    start..end.max(start)
492}
493
494fn append_section(
495    bytes: &mut Vec<u8>,
496    payload: &[u8],
497    compress: bool,
498) -> Result<BlockHandle, SstBlockCodecError> {
499    let stored = if compress {
500        zstd::bulk::compress(payload, ZSTD_LEVEL)
501            .map_err(|error| SstBlockCodecError::Codec(error.to_string()))?
502    } else {
503        payload.to_vec()
504    };
505    let handle = BlockHandle {
506        offset: bytes.len() as u64,
507        stored_len: stored.len() as u32,
508        decoded_len: payload.len() as u32,
509        crc32c: crc32c::crc32c(&stored),
510    };
511    bytes.extend_from_slice(&stored);
512    Ok(handle)
513}
514
515fn decode_section(
516    stored: &[u8],
517    handle: &BlockHandle,
518    compressed: bool,
519) -> Result<Vec<u8>, SstBlockCodecError> {
520    if stored.len() != handle.stored_len as usize {
521        return Err(SstBlockCodecError::StoredLengthMismatch {
522            expected: handle.stored_len,
523            actual: stored.len(),
524        });
525    }
526    let actual = crc32c::crc32c(stored);
527    if actual != handle.crc32c {
528        return Err(SstBlockCodecError::ChecksumMismatch {
529            expected: handle.crc32c,
530            actual,
531        });
532    }
533    let payload = if compressed {
534        let mut payload = Vec::with_capacity(handle.decoded_len as usize);
535        zstd::Decoder::new(stored)
536            .and_then(|mut decoder| decoder.read_to_end(&mut payload))
537            .map_err(|error| SstBlockCodecError::Codec(error.to_string()))?;
538        payload
539    } else {
540        stored.to_vec()
541    };
542    if payload.len() != handle.decoded_len as usize {
543        return Err(SstBlockCodecError::DecodedLengthMismatch {
544            expected: handle.decoded_len,
545            actual: payload.len(),
546        });
547    }
548    Ok(payload)
549}
550
551fn build_filter_payload(hashes: &[(u64, u64)]) -> Vec<u8> {
552    let bit_len = (hashes.len() * FILTER_BITS_PER_KEY).max(64) as u64;
553    let mut bits = vec![0u8; bit_len.div_ceil(8) as usize];
554    for (h1, h2) in hashes {
555        for probe in 0..u64::from(FILTER_HASH_COUNT) {
556            let bit = h1.wrapping_add(probe.wrapping_mul(*h2)) % bit_len;
557            bits[(bit / 8) as usize] |= 1 << (bit % 8);
558        }
559    }
560    let mut payload = Vec::with_capacity(12 + bits.len());
561    payload.extend_from_slice(&FILTER_HASH_COUNT.to_le_bytes());
562    payload.extend_from_slice(&bit_len.to_le_bytes());
563    payload.extend_from_slice(&bits);
564    payload
565}
566
567fn filter_key_hashes(filter_key: &str) -> (u64, u64) {
568    (
569        xxh64(filter_key.as_bytes(), FILTER_HASH_SEED_ONE),
570        xxh64(filter_key.as_bytes(), FILTER_HASH_SEED_TWO),
571    )
572}
573
574fn shared_prefix_len(previous: &str, current: &str) -> usize {
575    let mut len = previous
576        .as_bytes()
577        .iter()
578        .zip(current.as_bytes())
579        .take_while(|(a, b)| a == b)
580        .count();
581    // Both inputs are valid UTF-8 strings; back the byte-wise prefix off to
582    // a character boundary so key reconstruction can slice the previous key.
583    while !current.is_char_boundary(len) {
584        len -= 1;
585    }
586    len
587}
588
589pub(crate) fn write_varint(bytes: &mut Vec<u8>, mut value: u64) {
590    loop {
591        let byte = (value & 0x7f) as u8;
592        value >>= 7;
593        if value == 0 {
594            bytes.push(byte);
595            return;
596        }
597        bytes.push(byte | 0x80);
598    }
599}
600
601pub(crate) fn read_varint(bytes: &[u8], cursor: &mut usize) -> Result<u64, SstBlockCodecError> {
602    let mut value = 0u64;
603    let mut shift = 0u32;
604    loop {
605        let byte = *bytes.get(*cursor).ok_or_else(|| {
606            SstBlockCodecError::Malformed("varint runs past the block".to_owned())
607        })?;
608        *cursor += 1;
609        if shift >= 64 {
610            return Err(SstBlockCodecError::Malformed(
611                "varint exceeds 64 bits".to_owned(),
612            ));
613        }
614        value |= u64::from(byte & 0x7f) << shift;
615        if byte & 0x80 == 0 {
616            return Ok(value);
617        }
618        shift += 7;
619    }
620}
621
622fn take_slice<'a>(
623    bytes: &'a [u8],
624    cursor: &mut usize,
625    len: usize,
626) -> Result<&'a [u8], SstBlockCodecError> {
627    let end = cursor.checked_add(len).filter(|end| *end <= bytes.len());
628    match end {
629        Some(end) => {
630            let slice = &bytes[*cursor..end];
631            *cursor = end;
632            Ok(slice)
633        }
634        None => Err(SstBlockCodecError::Malformed(
635            "entry runs past the block".to_owned(),
636        )),
637    }
638}
639
640#[cfg(test)]
641mod tests {
642    use super::*;
643    use crate::{ChangeSeq, InodeId, InodeKind};
644
645    fn inode_row(inode_id: u64) -> (String, String, MetadataRow) {
646        let row = MetadataRow::Inode {
647            inode_id: InodeId(inode_id),
648            inode_kind: InodeKind::File,
649            created_seq: ChangeSeq(inode_id),
650        };
651        let key = row.row_key();
652        (key.clone(), key, row)
653    }
654
655    fn build_segment(rows: usize) -> BuiltSegmentBlocks {
656        let mut builder = SegmentBlocksBuilder::default();
657        for index in 0..rows {
658            let (key, filter_key, row) = inode_row(index as u64);
659            builder.push(&key, &filter_key, &row).expect("push row");
660        }
661        builder.finish().expect("finish segment")
662    }
663
664    fn section<'a>(bytes: &'a [u8], handle: &BlockHandle) -> &'a [u8] {
665        &bytes[handle.offset as usize..handle.offset as usize + handle.stored_len as usize]
666    }
667
668    #[test]
669    fn segment_round_trips_every_row_through_index_and_blocks() {
670        let rows = 5_000;
671        let built = build_segment(rows);
672        let index =
673            decode_index_block(section(&built.bytes, &built.index), &built.index).expect("index");
674        assert!(index.len() > 1, "5k inode rows should span several blocks");
675
676        let mut recovered = Vec::new();
677        for entry in &index {
678            let block = decode_data_block(section(&built.bytes, &entry.block), &entry.block)
679                .expect("data block");
680            assert_eq!(block.row_keys.len(), block.rows.len());
681            assert_eq!(
682                block.row_keys.last().expect("blocks are never empty"),
683                &entry.last_key
684            );
685            recovered.extend(block.row_keys.iter().cloned());
686        }
687        let expected: Vec<String> = (0..rows).map(|i| inode_row(i as u64).0).collect();
688        assert_eq!(recovered, expected);
689        assert_eq!(built.row_count, rows as u64);
690        assert_eq!(built.min_key, expected[0]);
691        assert_eq!(&built.max_key, expected.last().expect("rows"));
692    }
693
694    #[test]
695    fn index_narrows_point_lookups_to_one_block() {
696        let built = build_segment(5_000);
697        let index =
698            decode_index_block(section(&built.bytes, &built.index), &built.index).expect("index");
699        let (key, _, row) = inode_row(3_217);
700        let upper = format!("{key}\0");
701        let range = index_blocks_for_key_range(&index, &key, Some(&upper));
702        assert_eq!(range.len(), 1, "a point lookup should touch one block");
703        let entry = &index[range.start];
704        let block =
705            decode_data_block(section(&built.bytes, &entry.block), &entry.block).expect("block");
706        let position = block
707            .row_keys
708            .binary_search_by(|candidate| candidate.as_str().cmp(key.as_str()))
709            .expect("row should be present");
710        assert_eq!(block.rows[position], row);
711    }
712
713    #[test]
714    fn key_range_scan_covers_exactly_the_matching_blocks() {
715        let built = build_segment(5_000);
716        let index =
717            decode_index_block(section(&built.bytes, &built.index), &built.index).expect("index");
718        let lower = inode_row(1_000).0;
719        let upper = inode_row(1_500).0;
720        let range = index_blocks_for_key_range(&index, &lower, Some(&upper));
721        let mut keys = Vec::new();
722        for entry in &index[range] {
723            let block = decode_data_block(section(&built.bytes, &entry.block), &entry.block)
724                .expect("block");
725            keys.extend(block.row_keys);
726        }
727        let keys: Vec<&String> = keys
728            .iter()
729            .filter(|key| key.as_str() >= lower.as_str() && key.as_str() < upper.as_str())
730            .collect();
731        assert_eq!(keys.len(), 500);
732    }
733
734    #[test]
735    fn out_of_order_and_empty_segments_are_rejected() {
736        let mut builder = SegmentBlocksBuilder::default();
737        let (key_b, filter_b, row_b) = inode_row(2);
738        let (key_a, filter_a, row_a) = inode_row(1);
739        builder.push(&key_b, &filter_b, &row_b).expect("first row");
740        let error = builder
741            .push(&key_a, &filter_a, &row_a)
742            .expect_err("descending key should be rejected");
743        assert!(matches!(
744            error,
745            SstBlockCodecError::RowKeysOutOfOrder { .. }
746        ));
747
748        let error = SegmentBlocksBuilder::default()
749            .finish()
750            .expect_err("empty segment should be rejected");
751        assert!(matches!(error, SstBlockCodecError::EmptySegment));
752    }
753
754    #[test]
755    fn adjacent_equal_keys_are_permitted() {
756        let mut builder = SegmentBlocksBuilder::default();
757        let (key, filter_key, row) = inode_row(7);
758        builder.push(&key, &filter_key, &row).expect("first copy");
759        builder.push(&key, &filter_key, &row).expect("second copy");
760        let built = builder.finish().expect("finish");
761        assert_eq!(built.row_count, 2);
762    }
763
764    #[test]
765    fn corrupted_sections_fail_their_checksums() {
766        let built = build_segment(200);
767        let index =
768            decode_index_block(section(&built.bytes, &built.index), &built.index).expect("index");
769
770        let mut corrupted = built.bytes.clone();
771        let target = index[0].block.offset as usize + 3;
772        corrupted[target] ^= 0xff;
773        let error = decode_data_block(section(&corrupted, &index[0].block), &index[0].block)
774            .expect_err("corrupted data block should fail");
775        assert!(matches!(error, SstBlockCodecError::ChecksumMismatch { .. }));
776
777        let mut corrupted = built.bytes.clone();
778        let target = built.index.offset as usize + 3;
779        corrupted[target] ^= 0xff;
780        let error = decode_index_block(section(&corrupted, &built.index), &built.index)
781            .expect_err("corrupted index should fail");
782        assert!(matches!(error, SstBlockCodecError::ChecksumMismatch { .. }));
783
784        let mut corrupted = built.bytes.clone();
785        let target = built.filter.offset as usize + 12;
786        corrupted[target] ^= 0xff;
787        let error = decode_filter_block(section(&corrupted, &built.filter), &built.filter)
788            .expect_err("corrupted filter should fail");
789        assert!(matches!(error, SstBlockCodecError::ChecksumMismatch { .. }));
790    }
791
792    #[test]
793    fn filter_has_no_false_negatives_and_few_false_positives() {
794        let rows = 2_000;
795        let built = build_segment(rows);
796        let filter = decode_filter_block(section(&built.bytes, &built.filter), &built.filter)
797            .expect("filter");
798        for index in 0..rows {
799            let (key, _, _) = inode_row(index as u64);
800            assert!(filter.may_contain(&key), "inserted key must stay positive");
801        }
802        let mut false_positives = 0usize;
803        let probes = 10_000usize;
804        for index in 0..probes {
805            let (absent, _, _) = inode_row((rows + 10_000 + index) as u64);
806            if filter.may_contain(&absent) {
807                false_positives += 1;
808            }
809        }
810        let rate = false_positives as f64 / probes as f64;
811        assert!(rate < 0.02, "false positive rate {rate} exceeds 2%");
812    }
813
814    #[test]
815    fn durable_encoding_is_deterministic() {
816        let first = build_segment(300);
817        let second = build_segment(300);
818        assert_eq!(first.bytes, second.bytes);
819        assert_eq!(first.index, second.index);
820        assert_eq!(first.filter, second.filter);
821    }
822
823    #[test]
824    fn decoding_rejects_out_of_order_rows_in_a_block() {
825        // A hostile block with descending keys and a valid CRC: encode two
826        // full-key entries in the wrong order through the private helpers.
827        let mut entries = Vec::new();
828        for inode in [9u64, 3u64] {
829            let (key, _, row) = inode_row(inode);
830            let mut row_bytes = Vec::new();
831            ciborium::ser::into_writer(&row, &mut row_bytes).expect("encode row");
832            write_varint(&mut entries, 0);
833            write_varint(&mut entries, key.len() as u64);
834            entries.extend_from_slice(key.as_bytes());
835            write_varint(&mut entries, row_bytes.len() as u64);
836            entries.extend_from_slice(&row_bytes);
837        }
838        let mut payload = entries;
839        payload.extend_from_slice(&0u32.to_le_bytes());
840        payload.extend_from_slice(&0u32.to_le_bytes());
841        let mut bytes = Vec::new();
842        let handle = append_section(&mut bytes, &payload, true).expect("append section");
843
844        let error =
845            decode_data_block(&bytes, &handle).expect_err("descending rows should be rejected");
846        assert!(
847            matches!(&error, SstBlockCodecError::Malformed(message) if message.contains("row-key order")),
848            "unexpected error: {error}"
849        );
850    }
851
852    #[test]
853    fn decoding_rejects_out_of_order_index_entries() {
854        let block = BlockHandle {
855            offset: 0,
856            stored_len: 1,
857            decoded_len: 1,
858            crc32c: 0,
859        };
860        let entries = vec![
861            SegmentIndexEntry {
862                last_key: "inode-00000000000000000009".to_owned(),
863                block,
864            },
865            SegmentIndexEntry {
866                last_key: "inode-00000000000000000003".to_owned(),
867                block,
868            },
869        ];
870        let mut payload = Vec::new();
871        ciborium::ser::into_writer(&entries, &mut payload).expect("encode index");
872        let mut bytes = Vec::new();
873        let handle = append_section(&mut bytes, &payload, true).expect("append section");
874
875        let error =
876            decode_index_block(&bytes, &handle).expect_err("descending index should be rejected");
877        assert!(
878            matches!(&error, SstBlockCodecError::Malformed(message) if message.contains("key order")),
879            "unexpected error: {error}"
880        );
881    }
882}