Skip to main content

loonfs_api/
sst_blocks.rs

1//! Block-granular encoding for metadata and derived-index segments.
2//!
3//! A segment contains data blocks followed by one filter block and one index
4//! block. The manifest descriptor stores the index and filter handles. Each
5//! handle identifies a byte range and its CRC32C.
6//!
7//! The block format supports any CBOR row type, including [`MetadataRow`] and
8//! grep index rows. The durable layout is:
9//!
10//! - A **data block** holds prefix-compressed entries: each entry stores
11//!   `(shared_prefix_len, key_suffix_len)` as LEB128 varints, the key
12//!   suffix, and a length-prefixed CBOR row. Every [`RESTART_INTERVAL`]th
13//!   entry stores its full key. Restart offsets and their count end the
14//!   zstd-compressed block.
15//! - The **index block** is a zstd-compressed CBOR list with one entry per
16//!   data block: the block's last row key and its [`BlockHandle`].
17//! - The **filter block** is a bloom filter over caller-chosen filter keys
18//!   and is stored without compression.
19//! - Every section's CRC32C is stored in its handle.
20//! - Bloom hashing uses two xxh64 hashes with fixed seeds.
21
22use crate::wire::manifest::MetadataRow;
23use serde::{Deserialize, Serialize};
24use std::io::Read;
25use std::num::NonZeroUsize;
26use thiserror::Error;
27use xxhash_rust::xxh64::xxh64;
28
29/// Target uncompressed size of one data block.
30pub const DEFAULT_TARGET_BLOCK_BYTES: usize = 64 * 1024;
31/// Number of level-zero runs that triggers reorganization.
32pub const DEFAULT_MAX_DELTA_RUNS: usize = 8;
33/// Target number of rows in one immutable segment.
34pub const DEFAULT_MAX_ROWS_PER_SEGMENT: usize = 65_536;
35/// Target decoded data bytes in one metadata segment. A row is never split.
36pub const DEFAULT_TARGET_SEGMENT_BYTES: usize = 8 * 1024 * 1024;
37/// Maximum number of runs read by one reorganization step.
38pub const DEFAULT_MAX_REORGANIZATION_INPUT_RUNS: usize = 8;
39/// Maximum number of decoded rows read by one reorganization step.
40pub const DEFAULT_MAX_REORGANIZATION_INPUT_ROWS: usize = 131_072;
41/// Maximum decoded input size for one build or reorganization step.
42pub const DEFAULT_MAX_REORGANIZATION_INPUT_BYTES: usize = 64 * 1024 * 1024;
43/// Maximum stored filter size embedded in a segment descriptor.
44pub const DEFAULT_INLINE_FILTER_MAX_BYTES: u32 = 1024;
45/// Entries between restart points inside a data block.
46pub const RESTART_INTERVAL: usize = 16;
47/// Bloom filter sizing: bits reserved per inserted filter key.
48pub const FILTER_BITS_PER_KEY: usize = 10;
49/// Bloom filter probe count, chosen for [`FILTER_BITS_PER_KEY`].
50pub const FILTER_HASH_COUNT: u32 = 7;
51
52const FILTER_HASH_SEED_ONE: u64 = 0;
53const FILTER_HASH_SEED_TWO: u64 = 0x9e37_79b9_7f4a_7c15;
54/// Compression level for every zstd-compressed durable artifact.
55pub(crate) const ZSTD_LEVEL: i32 = 3;
56
57/// Where one stored section lives inside a segment object, and how to
58/// verify it: the CRC32C of the stored bytes and their decoded length.
59#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
60#[serde(deny_unknown_fields)]
61pub struct BlockHandle {
62    /// Zero-based byte offset of the section within its immutable segment object.
63    pub offset: u64,
64    /// Number of bytes to range-read and checksum before decoding.
65    pub stored_len: u32,
66    /// Expected byte length after optional section decompression.
67    pub decoded_len: u32,
68    /// CRC32C over the exact `stored_len` bytes at `offset`.
69    pub crc32c: u32,
70}
71
72/// One index entry: the last row key of a data block plus its handle.
73#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
74#[serde(deny_unknown_fields)]
75pub struct SegmentIndexEntry {
76    /// Greatest row key in `block`, used to binary-search candidate blocks.
77    pub last_row_key: String,
78    /// Data-section location and integrity metadata.
79    pub block: BlockHandle,
80}
81
82/// A finished segment: the object bytes plus everything the manifest
83/// descriptor must carry to read them back.
84#[derive(Debug, Clone, PartialEq, Eq)]
85pub struct BuiltSegmentBlocks {
86    /// Complete immutable object body, with data sections followed by filter and index sections.
87    pub bytes: Vec<u8>,
88    /// Handle callers persist in the segment descriptor to bootstrap reads.
89    pub index: BlockHandle,
90    /// Handle callers persist for negative point-lookup filtering.
91    pub filter: BlockHandle,
92    /// Number of rows accepted by the builder, including adjacent duplicate keys.
93    pub row_count: u64,
94    /// Least row key in the non-empty segment.
95    pub min_row_key: String,
96    /// Greatest row key in the non-empty segment.
97    pub max_row_key: String,
98}
99
100impl BuiltSegmentBlocks {
101    /// Encodes a small stored filter for an inline segment descriptor.
102    pub fn inline_filter_hex(&self) -> Option<String> {
103        (self.filter.stored_len <= DEFAULT_INLINE_FILTER_MAX_BYTES).then(|| {
104            let start = self.filter.offset as usize;
105            crate::wire::hex::hex_encode_bytes(
106                &self.bytes[start..start + self.filter.stored_len as usize],
107            )
108        })
109    }
110}
111
112/// Returns whether a non-empty segment key range can intersect a query range.
113pub fn key_range_may_intersect(
114    min_row_key: &str,
115    max_row_key: &str,
116    row_count: u64,
117    lower_bound: &str,
118    upper_bound: Option<&str>,
119) -> bool {
120    row_count != 0
121        && max_row_key >= lower_bound
122        && upper_bound.is_none_or(|upper_bound| min_row_key < upper_bound)
123}
124
125/// One decoded data block: row keys and rows, parallel and in key order.
126/// The row type defaults to [`MetadataRow`]; index segments decode their
127/// own row payload through [`decode_data_block_rows`].
128#[derive(Debug, Clone, PartialEq, Eq)]
129pub struct DecodedDataBlock<R = MetadataRow> {
130    /// Reconstructed row keys in the same ascending order as `rows`.
131    pub row_keys: Vec<String>,
132    /// Decoded row payloads positionally paired with `row_keys`.
133    pub rows: Vec<R>,
134}
135
136/// A decoded bloom filter; answers "definitely absent" or "maybe present".
137#[derive(Debug, Clone, PartialEq, Eq)]
138pub struct SegmentFilter {
139    n_hashes: u32,
140    bit_len: u64,
141    bits: Vec<u8>,
142}
143
144/// Describes a violation encountered while building or validating an SST section.
145///
146/// See [metadata segments](../../../docs/specs/format.md#a7-block-segment-encoding).
147#[derive(Debug, Clone, PartialEq, Eq, Error)]
148#[non_exhaustive]
149pub enum SstBlockCodecError {
150    /// Reports a request to finish a segment before any row was supplied.
151    #[error("segment must contain at least one row")]
152    EmptySegment,
153    /// Reports a builder input that would violate durable ascending row-key order.
154    #[error("row key `{offered}` is not in ascending order after `{previous}`")]
155    RowKeysOutOfOrder {
156        /// Last key the builder accepted.
157        previous: String,
158        /// Descending key rejected by the builder.
159        offered: String,
160    },
161    /// Reports a range-read body whose byte count disagrees with its handle.
162    #[error("stored bytes length {actual} does not match handle length {expected}")]
163    StoredLengthMismatch {
164        /// Stored byte count recorded in the persisted `BlockHandle`.
165        expected: u32,
166        /// Byte count returned to the decoder.
167        actual: usize,
168    },
169    /// Reports stored section bytes that fail the CRC32C recorded in their handle.
170    #[error("block checksum mismatch: expected {expected:#010x}, actual {actual:#010x}")]
171    ChecksumMismatch {
172        /// CRC32C recorded in the persisted `BlockHandle`.
173        expected: u32,
174        /// CRC32C recomputed from the supplied stored bytes.
175        actual: u32,
176    },
177    /// Reports a section whose decompressed size disagrees with its handle.
178    #[error("decoded length {actual} does not match handle length {expected}")]
179    DecodedLengthMismatch {
180        /// Decoded byte count recorded in the persisted `BlockHandle`.
181        expected: u32,
182        /// Byte count produced by section decompression.
183        actual: usize,
184    },
185    /// Reports structurally invalid framing, ordering, UTF-8, or filter metadata.
186    #[error("malformed block: {0}")]
187    Malformed(String),
188    /// Reports a CBOR or zstd failure while encoding or decoding a section.
189    #[error("block codec error: {0}")]
190    Codec(String),
191}
192
193/// Builds one segment's blocks from rows fed in ascending row-key order.
194#[derive(Debug)]
195#[must_use]
196pub struct SegmentBlocksBuilder {
197    target_block_bytes: usize,
198    entries: Vec<u8>,
199    restarts: Vec<u32>,
200    entry_count: usize,
201    /// Last row key the builder accepted. It anchors prefix compression
202    /// inside a block, floors the ascending-order guard, and becomes the
203    /// segment's max row key. A block's first entry stores its key in full
204    /// regardless, because a restart point always begins a block.
205    previous_key: String,
206    bytes: Vec<u8>,
207    index: Vec<SegmentIndexEntry>,
208    decoded_data_bytes: usize,
209    filter_hashes: Vec<(u64, u64)>,
210    row_count: u64,
211    min_row_key: String,
212}
213
214impl Default for SegmentBlocksBuilder {
215    fn default() -> Self {
216        Self::new(const { NonZeroUsize::new(DEFAULT_TARGET_BLOCK_BYTES).unwrap() })
217    }
218}
219
220impl SegmentBlocksBuilder {
221    /// Creates a builder that closes a data block after reaching the target decoded byte size.
222    pub fn new(target_block_bytes: NonZeroUsize) -> Self {
223        Self {
224            target_block_bytes: target_block_bytes.get(),
225            entries: Vec::new(),
226            restarts: Vec::new(),
227            entry_count: 0,
228            previous_key: String::new(),
229            bytes: Vec::new(),
230            index: Vec::new(),
231            decoded_data_bytes: 0,
232            filter_hashes: Vec::new(),
233            row_count: 0,
234            min_row_key: String::new(),
235        }
236    }
237
238    /// Appends one row. `filter_key` is the lookup prefix point reads will
239    /// probe for this row; the caller derives it per family. The row is any
240    /// CBOR payload; a segment must hold one row type throughout, named by
241    /// the descriptor family that references it.
242    pub fn push<R: Serialize>(
243        &mut self,
244        row_key: &str,
245        filter_key: &str,
246        row: &R,
247    ) -> Result<(), SstBlockCodecError> {
248        self.push_with_encoded_row(row_key, filter_key, row)
249            .map(drop)
250    }
251
252    /// Appends one row and returns its CBOR encoding.
253    pub fn push_with_encoded_row<R: Serialize>(
254        &mut self,
255        row_key: &str,
256        filter_key: &str,
257        row: &R,
258    ) -> Result<Vec<u8>, SstBlockCodecError> {
259        if self.row_count > 0 && row_key < self.previous_key.as_str() {
260            return Err(SstBlockCodecError::RowKeysOutOfOrder {
261                previous: self.previous_key.clone(),
262                offered: row_key.to_owned(),
263            });
264        }
265        if self.row_count == 0 {
266            self.min_row_key = row_key.to_owned();
267        }
268        self.filter_hashes.push(filter_key_hashes(filter_key));
269
270        let restart = self.entry_count % RESTART_INTERVAL == 0;
271        if restart {
272            self.restarts.push(self.entries.len() as u32);
273        }
274        let shared_len = if restart {
275            0
276        } else {
277            shared_prefix_len(&self.previous_key, row_key)
278        };
279        let suffix = &row_key.as_bytes()[shared_len..];
280        let mut row_bytes = Vec::new();
281        ciborium::ser::into_writer(row, &mut row_bytes)
282            .map_err(|error| SstBlockCodecError::Codec(error.to_string()))?;
283        let previous_len = self.entries.len();
284        write_varint(&mut self.entries, shared_len as u64);
285        write_varint(&mut self.entries, suffix.len() as u64);
286        self.entries.extend_from_slice(suffix);
287        write_varint(&mut self.entries, row_bytes.len() as u64);
288        self.entries.extend_from_slice(&row_bytes);
289
290        self.decoded_data_bytes += self.entries.len() - previous_len;
291        self.entry_count += 1;
292        self.row_count += 1;
293        self.previous_key.clear();
294        self.previous_key.push_str(row_key);
295        if self.entries.len() >= self.target_block_bytes {
296            self.finish_data_block()?;
297        }
298        Ok(row_bytes)
299    }
300
301    fn finish_data_block(&mut self) -> Result<(), SstBlockCodecError> {
302        if self.entries.is_empty() {
303            return Ok(());
304        }
305        let mut payload = std::mem::take(&mut self.entries);
306        for restart in &self.restarts {
307            payload.extend_from_slice(&restart.to_le_bytes());
308        }
309        payload.extend_from_slice(&(self.restarts.len() as u32).to_le_bytes());
310        self.decoded_data_bytes += (self.restarts.len() + 1) * 4;
311        self.restarts.clear();
312        self.entry_count = 0;
313        let block = append_section(&mut self.bytes, &payload, true)?;
314        self.index.push(SegmentIndexEntry {
315            last_row_key: self.previous_key.clone(),
316            block,
317        });
318        Ok(())
319    }
320
321    /// Number of rows accepted by this builder.
322    pub fn row_count(&self) -> u64 {
323        self.row_count
324    }
325
326    /// Decoded data bytes accepted so far, including block restart tables.
327    /// The open block's pending restart table is included.
328    pub fn decoded_data_bytes(&self) -> usize {
329        self.decoded_data_bytes
330            + if self.entries.is_empty() {
331                0
332            } else {
333                (self.restarts.len() + 1) * 4
334            }
335    }
336
337    /// Encodes the remaining rows and assembles the object bytes.
338    pub fn finish(mut self) -> Result<BuiltSegmentBlocks, SstBlockCodecError> {
339        if self.row_count == 0 {
340            return Err(SstBlockCodecError::EmptySegment);
341        }
342        self.finish_data_block()?;
343        let mut bytes = self.bytes;
344        let index = self.index;
345
346        let filter_payload = build_filter_payload(&self.filter_hashes);
347        let filter = append_section(&mut bytes, &filter_payload, false)?;
348
349        let mut index_payload = Vec::new();
350        ciborium::ser::into_writer(&index, &mut index_payload)
351            .map_err(|error| SstBlockCodecError::Codec(error.to_string()))?;
352        let index = append_section(&mut bytes, &index_payload, true)?;
353
354        Ok(BuiltSegmentBlocks {
355            bytes,
356            index,
357            filter,
358            row_count: self.row_count,
359            min_row_key: self.min_row_key,
360            max_row_key: self.previous_key,
361        })
362    }
363}
364
365/// Decodes the index block from exactly the bytes its handle names.
366pub fn decode_index_block(
367    stored: &[u8],
368    handle: &BlockHandle,
369) -> Result<Vec<SegmentIndexEntry>, SstBlockCodecError> {
370    let payload = decode_section(stored, handle, true)?;
371    let entries: Vec<SegmentIndexEntry> = ciborium::de::from_reader(payload.as_slice())
372        .map_err(|error| SstBlockCodecError::Codec(error.to_string()))?;
373    // Index keys must be sorted, and block ranges must tile their region:
374    // every range fits in `u64`, and each block starts exactly where its
375    // predecessor ends — the builder writes blocks back to back. Range
376    // lookup relies on key order; span loading and its bulk-read budget
377    // rely on contiguous, overflow-free ranges.
378    let mut previous: Option<(&String, u64)> = None;
379    for entry in &entries {
380        let end = entry
381            .block
382            .offset
383            .checked_add(u64::from(entry.block.stored_len))
384            .ok_or_else(|| {
385                SstBlockCodecError::Malformed(format!(
386                    "index block `{}` byte range overflows",
387                    entry.last_row_key
388                ))
389            })?;
390        if let Some((previous_key, previous_end)) = previous {
391            if previous_key > &entry.last_row_key {
392                return Err(SstBlockCodecError::Malformed(format!(
393                    "index blocks out of key order: `{}` follows `{previous_key}`",
394                    entry.last_row_key
395                )));
396            }
397            if entry.block.offset != previous_end {
398                return Err(SstBlockCodecError::Malformed(format!(
399                    "index block `{}` does not start where `{previous_key}` ends",
400                    entry.last_row_key
401                )));
402            }
403        }
404        previous = Some((&entry.last_row_key, end));
405    }
406    Ok(entries)
407}
408
409/// Decodes one data block from exactly the bytes its handle names.
410pub fn decode_data_block(
411    stored: &[u8],
412    handle: &BlockHandle,
413) -> Result<DecodedDataBlock, SstBlockCodecError> {
414    decode_data_block_rows::<MetadataRow>(stored, handle)
415}
416
417/// Decodes one data block whose rows are `R`, for segment families whose
418/// row payload is not [`MetadataRow`] (gram index segments).
419pub fn decode_data_block_rows<R: serde::de::DeserializeOwned>(
420    stored: &[u8],
421    handle: &BlockHandle,
422) -> Result<DecodedDataBlock<R>, SstBlockCodecError> {
423    let payload = decode_section(stored, handle, true)?;
424    if payload.len() < 4 {
425        return Err(SstBlockCodecError::Malformed(
426            "data block shorter than its restart count".to_owned(),
427        ));
428    }
429    let (body, restart_count_bytes) = payload.split_at(payload.len() - 4);
430    let restart_count = u32::from_le_bytes(
431        restart_count_bytes
432            .try_into()
433            .expect("split_at should leave exactly four bytes"),
434    ) as usize;
435    let restarts_len = restart_count
436        .checked_mul(4)
437        .filter(|len| *len <= body.len())
438        .ok_or_else(|| SstBlockCodecError::Malformed("restart array exceeds block".to_owned()))?;
439    let (entries, restarts) = body.split_at(body.len() - restarts_len);
440    let mut restarts = restarts.chunks_exact(4);
441
442    let mut row_keys = Vec::new();
443    let mut rows = Vec::new();
444    let mut cursor = 0usize;
445    let mut previous_key = String::new();
446    while cursor < entries.len() {
447        let restart = rows.len() % RESTART_INTERVAL == 0;
448        if restart {
449            let offset = restarts.next().map(|bytes| {
450                u32::from_le_bytes(
451                    bytes
452                        .try_into()
453                        .expect("restart offsets should have four bytes"),
454                )
455            });
456            if offset.map(u64::from) != Some(cursor as u64) {
457                return Err(SstBlockCodecError::Malformed(
458                    "restart offset disagrees with its entry".to_owned(),
459                ));
460            }
461        }
462        let shared_len = read_entry_length(entries, &mut cursor)?;
463        let suffix_len = read_entry_length(entries, &mut cursor)?;
464        if restart && shared_len != 0 {
465            return Err(SstBlockCodecError::Malformed(
466                "restart entry has a shared prefix".to_owned(),
467            ));
468        }
469        if shared_len > previous_key.len() || !previous_key.is_char_boundary(shared_len) {
470            return Err(SstBlockCodecError::Malformed(
471                "shared prefix exceeds previous key".to_owned(),
472            ));
473        }
474        let suffix = take_slice(entries, &mut cursor, suffix_len)?;
475        let suffix = std::str::from_utf8(suffix)
476            .map_err(|_| SstBlockCodecError::Malformed("row key is not utf-8".to_owned()))?;
477        let key_len = shared_len.checked_add(suffix.len()).ok_or_else(|| {
478            SstBlockCodecError::Malformed("row key length exceeds address space".to_owned())
479        })?;
480        let mut key = String::with_capacity(key_len);
481        key.push_str(&previous_key[..shared_len]);
482        key.push_str(suffix);
483        let row_len = read_entry_length(entries, &mut cursor)?;
484        let row_bytes = take_slice(entries, &mut cursor, row_len)?;
485        let row: R = ciborium::de::from_reader(row_bytes)
486            .map_err(|error| SstBlockCodecError::Codec(error.to_string()))?;
487        // Ascending row-key order is a format requirement; readers
488        // binary-search on it, so an out-of-order block is malformed.
489        if key.as_str() < previous_key.as_str() {
490            return Err(SstBlockCodecError::Malformed(format!(
491                "rows out of row-key order: `{key}` follows `{previous_key}`"
492            )));
493        }
494        previous_key.clear();
495        previous_key.push_str(&key);
496        row_keys.push(key);
497        rows.push(row);
498    }
499    if restarts.next().is_some() {
500        return Err(SstBlockCodecError::Malformed(
501            "restart array contains unused offsets".to_owned(),
502        ));
503    }
504    Ok(DecodedDataBlock { row_keys, rows })
505}
506
507/// Decodes the filter block from exactly the bytes its handle names.
508pub fn decode_filter_block(
509    stored: &[u8],
510    handle: &BlockHandle,
511) -> Result<SegmentFilter, SstBlockCodecError> {
512    let payload = decode_section(stored, handle, false)?;
513    if payload.len() < 12 {
514        return Err(SstBlockCodecError::Malformed(
515            "filter block shorter than its header".to_owned(),
516        ));
517    }
518    let n_hashes = u32::from_le_bytes(
519        payload[0..4]
520            .try_into()
521            .expect("header length should be checked above"),
522    );
523    let bit_len = u64::from_le_bytes(
524        payload[4..12]
525            .try_into()
526            .expect("header length should be checked above"),
527    );
528    if n_hashes != FILTER_HASH_COUNT {
529        return Err(SstBlockCodecError::Malformed(
530            "filter hash count must be seven".to_owned(),
531        ));
532    }
533    if bit_len < 64 {
534        return Err(SstBlockCodecError::Malformed(
535            "filter bit length must be at least 64".to_owned(),
536        ));
537    }
538    let bits = &payload[12..];
539    if bit_len.div_ceil(8) != bits.len() as u64 {
540        return Err(SstBlockCodecError::Malformed(
541            "filter bit length disagrees with its bytes".to_owned(),
542        ));
543    }
544    Ok(SegmentFilter {
545        n_hashes,
546        bit_len,
547        bits: bits.to_vec(),
548    })
549}
550
551impl SegmentFilter {
552    /// False means no row with this filter key is in the segment; true
553    /// means one may be.
554    pub fn may_contain(&self, filter_key: &str) -> bool {
555        let (h1, h2) = filter_key_hashes(filter_key);
556        for probe in 0..u64::from(self.n_hashes) {
557            let bit = h1.wrapping_add(probe.wrapping_mul(h2)) % self.bit_len;
558            let byte = self.bits[(bit / 8) as usize];
559            if byte & (1 << (bit % 8)) == 0 {
560                return false;
561            }
562        }
563        true
564    }
565}
566
567/// The exclusive upper bound for every row key beginning with `prefix`.
568///
569/// Row keys are ordered as byte strings, so a prefix scan is the range
570/// `[prefix, string_prefix_upper_bound(prefix))`. `None` means the prefix is
571/// all `0xff` bytes and nothing sorts above it, so the scan runs to the end.
572pub fn string_prefix_upper_bound(prefix: &str) -> Option<String> {
573    let mut bytes = prefix.as_bytes().to_vec();
574    for index in (0..bytes.len()).rev() {
575        if bytes[index] != u8::MAX {
576            bytes[index] += 1;
577            bytes.truncate(index + 1);
578            return String::from_utf8(bytes).ok();
579        }
580    }
581    None
582}
583
584/// Index positions of the blocks that can hold keys in
585/// `[lower_bound, upper_bound)`; `None` bounds the range at the last block.
586pub fn index_blocks_for_key_range(
587    index: &[SegmentIndexEntry],
588    lower_bound: &str,
589    upper_bound: Option<&str>,
590) -> std::ops::Range<usize> {
591    let start = index.partition_point(|entry| entry.last_row_key.as_str() < lower_bound);
592    let end = upper_bound.map_or(index.len(), |upper_bound| {
593        // A block whose last row key equals the exclusive upper bound can still
594        // hold keys below it, so the bound block itself is included.
595        index
596            .partition_point(|entry| entry.last_row_key.as_str() < upper_bound)
597            .saturating_add(1)
598            .min(index.len())
599    });
600    start..end.max(start)
601}
602
603fn append_section(
604    bytes: &mut Vec<u8>,
605    payload: &[u8],
606    compress: bool,
607) -> Result<BlockHandle, SstBlockCodecError> {
608    let stored = if compress {
609        zstd::bulk::compress(payload, ZSTD_LEVEL)
610            .map_err(|error| SstBlockCodecError::Codec(error.to_string()))?
611    } else {
612        payload.to_vec()
613    };
614    let handle = BlockHandle {
615        offset: bytes.len() as u64,
616        stored_len: stored.len() as u32,
617        decoded_len: payload.len() as u32,
618        crc32c: crc32c::crc32c(&stored),
619    };
620    bytes.extend_from_slice(&stored);
621    Ok(handle)
622}
623
624fn decode_section(
625    stored: &[u8],
626    handle: &BlockHandle,
627    compressed: bool,
628) -> Result<Vec<u8>, SstBlockCodecError> {
629    if stored.len() != handle.stored_len as usize {
630        return Err(SstBlockCodecError::StoredLengthMismatch {
631            expected: handle.stored_len,
632            actual: stored.len(),
633        });
634    }
635    let actual = crc32c::crc32c(stored);
636    if actual != handle.crc32c {
637        return Err(SstBlockCodecError::ChecksumMismatch {
638            expected: handle.crc32c,
639            actual,
640        });
641    }
642    let payload = if compressed {
643        let mut payload =
644            Vec::with_capacity((handle.decoded_len as usize).min(DEFAULT_TARGET_BLOCK_BYTES));
645        zstd::Decoder::new(stored)
646            .and_then(|decoder| {
647                decoder
648                    .take(u64::from(handle.decoded_len) + 1)
649                    .read_to_end(&mut payload)
650            })
651            .map_err(|error| SstBlockCodecError::Codec(error.to_string()))?;
652        payload
653    } else {
654        if stored.len() != handle.decoded_len as usize {
655            return Err(SstBlockCodecError::DecodedLengthMismatch {
656                expected: handle.decoded_len,
657                actual: stored.len(),
658            });
659        }
660        stored.to_vec()
661    };
662    if payload.len() != handle.decoded_len as usize {
663        return Err(SstBlockCodecError::DecodedLengthMismatch {
664            expected: handle.decoded_len,
665            actual: payload.len(),
666        });
667    }
668    Ok(payload)
669}
670
671fn build_filter_payload(hashes: &[(u64, u64)]) -> Vec<u8> {
672    let bit_len = (hashes.len() * FILTER_BITS_PER_KEY).max(64) as u64;
673    let mut bits = vec![0u8; bit_len.div_ceil(8) as usize];
674    for (h1, h2) in hashes {
675        for probe in 0..u64::from(FILTER_HASH_COUNT) {
676            let bit = h1.wrapping_add(probe.wrapping_mul(*h2)) % bit_len;
677            bits[(bit / 8) as usize] |= 1 << (bit % 8);
678        }
679    }
680    let mut payload = Vec::with_capacity(12 + bits.len());
681    payload.extend_from_slice(&FILTER_HASH_COUNT.to_le_bytes());
682    payload.extend_from_slice(&bit_len.to_le_bytes());
683    payload.extend_from_slice(&bits);
684    payload
685}
686
687fn filter_key_hashes(filter_key: &str) -> (u64, u64) {
688    (
689        xxh64(filter_key.as_bytes(), FILTER_HASH_SEED_ONE),
690        xxh64(filter_key.as_bytes(), FILTER_HASH_SEED_TWO),
691    )
692}
693
694fn shared_prefix_len(previous: &str, current: &str) -> usize {
695    let mut len = previous
696        .as_bytes()
697        .iter()
698        .zip(current.as_bytes())
699        .take_while(|(a, b)| a == b)
700        .count();
701    // Both inputs are valid UTF-8 strings; back the byte-wise prefix off to
702    // a character boundary so key reconstruction can slice the previous key.
703    while !current.is_char_boundary(len) {
704        len -= 1;
705    }
706    len
707}
708
709/// Appends `value` as an unsigned LEB128 integer.
710pub fn write_varint(bytes: &mut Vec<u8>, mut value: u64) {
711    loop {
712        let byte = (value & 0x7f) as u8;
713        value >>= 7;
714        if value == 0 {
715            bytes.push(byte);
716            return;
717        }
718        bytes.push(byte | 0x80);
719    }
720}
721
722/// Reads the LEB128 varint at `cursor` and advances `cursor` past it.
723pub fn read_varint(bytes: &[u8], cursor: &mut usize) -> Result<u64, SstBlockCodecError> {
724    let mut value = 0u64;
725    let mut shift = 0u32;
726    loop {
727        let byte = *bytes.get(*cursor).ok_or_else(|| {
728            SstBlockCodecError::Malformed("varint runs past the block".to_owned())
729        })?;
730        *cursor += 1;
731        if shift == 63 && byte > 1 {
732            return Err(SstBlockCodecError::Malformed(
733                "varint exceeds 64 bits".to_owned(),
734            ));
735        }
736        value |= u64::from(byte & 0x7f) << shift;
737        if byte & 0x80 == 0 {
738            return Ok(value);
739        }
740        shift += 7;
741    }
742}
743
744fn read_entry_length(bytes: &[u8], cursor: &mut usize) -> Result<usize, SstBlockCodecError> {
745    usize::try_from(read_varint(bytes, cursor)?)
746        .map_err(|_| SstBlockCodecError::Malformed("entry length exceeds address space".to_owned()))
747}
748
749fn take_slice<'a>(
750    bytes: &'a [u8],
751    cursor: &mut usize,
752    len: usize,
753) -> Result<&'a [u8], SstBlockCodecError> {
754    let end = cursor.checked_add(len).filter(|end| *end <= bytes.len());
755    match end {
756        Some(end) => {
757            let slice = &bytes[*cursor..end];
758            *cursor = end;
759            Ok(slice)
760        }
761        None => Err(SstBlockCodecError::Malformed(
762            "entry runs past the block".to_owned(),
763        )),
764    }
765}
766
767#[cfg(test)]
768mod tests {
769    use super::*;
770    use crate::{ChangeSeq, InodeId, InodeKind};
771
772    #[test]
773    fn decoded_byte_accounting_includes_closed_and_open_blocks() {
774        let mut builder =
775            SegmentBlocksBuilder::new(NonZeroUsize::new(512).expect("nonzero block target"));
776        for index in 0..17 {
777            let (key, filter, row) = inode_row(index);
778            builder.push(&key, &filter, &row).expect("encode row");
779        }
780        let decoded_bytes = builder.decoded_data_bytes();
781        assert_eq!(builder.row_count(), 17);
782        assert!(
783            !builder.index.is_empty(),
784            "completed blocks are already compressed"
785        );
786        let built = builder.finish().expect("finish segment");
787        let start = built.index.offset as usize;
788        let end = start + built.index.stored_len as usize;
789        let index = decode_index_block(&built.bytes[start..end], &built.index).expect("index");
790        assert_eq!(
791            decoded_bytes,
792            index
793                .iter()
794                .map(|entry| entry.block.decoded_len as usize)
795                .sum::<usize>(),
796        );
797    }
798
799    fn inode_row(inode_id: u64) -> (String, String, MetadataRow) {
800        let row = MetadataRow::Inode(crate::manifest::InodeRecord {
801            inode_id: InodeId(inode_id),
802            inode_kind: InodeKind::File,
803            created_seq: ChangeSeq(inode_id),
804            commit_id: crate::CommitId::parse(format!("c_row_{inode_id}"))
805                .expect("valid commit id"),
806            created_by: crate::ActorId::loonfs(),
807            created_at_ms: inode_id,
808        });
809        let key = row.row_key();
810        (key.clone(), key, row)
811    }
812
813    fn build_segment(rows: usize) -> BuiltSegmentBlocks {
814        let mut builder = SegmentBlocksBuilder::default();
815        for index in 0..rows {
816            let (key, filter_key, row) = inode_row(index as u64);
817            builder.push(&key, &filter_key, &row).expect("push row");
818        }
819        builder.finish().expect("finish segment")
820    }
821
822    const SHARED_SEGMENT_ROWS: usize = 5_000;
823
824    fn shared_segment() -> &'static BuiltSegmentBlocks {
825        static SEGMENT: std::sync::OnceLock<BuiltSegmentBlocks> = std::sync::OnceLock::new();
826        SEGMENT.get_or_init(|| build_segment(SHARED_SEGMENT_ROWS))
827    }
828
829    fn section<'a>(bytes: &'a [u8], handle: &BlockHandle) -> &'a [u8] {
830        &bytes[handle.offset as usize..handle.offset as usize + handle.stored_len as usize]
831    }
832
833    fn encode_index(entries: &[SegmentIndexEntry]) -> (Vec<u8>, BlockHandle) {
834        let mut payload = Vec::new();
835        ciborium::ser::into_writer(entries, &mut payload).expect("encode index");
836        let mut bytes = Vec::new();
837        let handle = append_section(&mut bytes, &payload, true).expect("append section");
838        (bytes, handle)
839    }
840
841    fn index_entry(last_row_key: &str, offset: u64, stored_len: u32) -> SegmentIndexEntry {
842        SegmentIndexEntry {
843            last_row_key: last_row_key.to_owned(),
844            block: BlockHandle {
845                offset,
846                stored_len,
847                decoded_len: stored_len,
848                crc32c: 0,
849            },
850        }
851    }
852
853    #[test]
854    fn segment_round_trips_every_row_through_index_and_blocks() {
855        let rows = SHARED_SEGMENT_ROWS;
856        let built = shared_segment();
857        let index =
858            decode_index_block(section(&built.bytes, &built.index), &built.index).expect("index");
859        assert!(index.len() > 1, "5k inode rows should span several blocks");
860
861        let mut recovered = Vec::new();
862        for entry in &index {
863            let block = decode_data_block(section(&built.bytes, &entry.block), &entry.block)
864                .expect("data block");
865            assert_eq!(block.row_keys.len(), block.rows.len());
866            assert_eq!(
867                block.row_keys.last().expect("blocks are never empty"),
868                &entry.last_row_key
869            );
870            recovered.extend(block.row_keys.iter().cloned());
871        }
872        let expected: Vec<String> = (0..rows).map(|i| inode_row(i as u64).0).collect();
873        assert_eq!(recovered, expected);
874        assert_eq!(built.row_count, rows as u64);
875        assert_eq!(built.min_row_key, expected[0]);
876        assert_eq!(&built.max_row_key, expected.last().expect("rows"));
877    }
878
879    #[test]
880    fn index_narrows_point_lookups_to_one_block() {
881        let built = shared_segment();
882        let index =
883            decode_index_block(section(&built.bytes, &built.index), &built.index).expect("index");
884        let (key, _, row) = inode_row(3_217);
885        let upper = format!("{key}\0");
886        let range = index_blocks_for_key_range(&index, &key, Some(&upper));
887        assert_eq!(range.len(), 1, "a point lookup should touch one block");
888        let entry = &index[range.start];
889        let block =
890            decode_data_block(section(&built.bytes, &entry.block), &entry.block).expect("block");
891        let position = block
892            .row_keys
893            .binary_search_by(|candidate| candidate.as_str().cmp(key.as_str()))
894            .expect("row should be present");
895        assert_eq!(block.rows[position], row);
896    }
897
898    #[test]
899    fn key_range_scan_covers_exactly_the_matching_blocks() {
900        let built = shared_segment();
901        let index =
902            decode_index_block(section(&built.bytes, &built.index), &built.index).expect("index");
903        let lower = inode_row(1_000).0;
904        let upper = inode_row(1_500).0;
905        let range = index_blocks_for_key_range(&index, &lower, Some(&upper));
906        let mut keys = Vec::new();
907        for entry in &index[range] {
908            let block = decode_data_block(section(&built.bytes, &entry.block), &entry.block)
909                .expect("block");
910            keys.extend(block.row_keys);
911        }
912        let keys: Vec<&String> = keys
913            .iter()
914            .filter(|key| key.as_str() >= lower.as_str() && key.as_str() < upper.as_str())
915            .collect();
916        assert_eq!(keys.len(), 500);
917    }
918
919    #[test]
920    fn out_of_order_and_empty_segments_are_rejected() {
921        let mut builder = SegmentBlocksBuilder::default();
922        let (key_b, filter_b, row_b) = inode_row(2);
923        let (key_a, filter_a, row_a) = inode_row(1);
924        builder.push(&key_b, &filter_b, &row_b).expect("first row");
925        let error = builder
926            .push(&key_a, &filter_a, &row_a)
927            .expect_err("descending key should be rejected");
928        assert!(matches!(
929            error,
930            SstBlockCodecError::RowKeysOutOfOrder { .. }
931        ));
932
933        let error = SegmentBlocksBuilder::default()
934            .finish()
935            .expect_err("empty segment should be rejected");
936        assert!(matches!(error, SstBlockCodecError::EmptySegment));
937    }
938
939    #[test]
940    fn max_key_survives_a_last_row_that_closes_its_block() {
941        // Calibrate the target so the crossing lands on the final row.
942        // Building the same rows as one block reports how many entry bytes
943        // they occupy: a block payload is the entries, then one `u32` per
944        // restart point, then the restart count.
945        let rows = 100usize;
946        let single_block = build_segment(rows);
947        let calibration = decode_index_block(
948            section(&single_block.bytes, &single_block.index),
949            &single_block.index,
950        )
951        .expect("index");
952        assert_eq!(calibration.len(), 1, "the calibration segment is one block");
953        let restarts = rows.div_ceil(RESTART_INTERVAL);
954        let entry_bytes = calibration[0].block.decoded_len as usize - 4 * restarts - 4;
955
956        let mut builder =
957            SegmentBlocksBuilder::new(NonZeroUsize::new(entry_bytes).expect("positive target"));
958        for index in 0..rows {
959            let (key, filter_key, row) = inode_row(index as u64);
960            builder.push(&key, &filter_key, &row).expect("push row");
961        }
962        let built = builder.finish().expect("finish segment");
963
964        let expected_max = inode_row((rows - 1) as u64).0;
965        assert_eq!(built.min_row_key, inode_row(0).0);
966        assert_eq!(built.max_row_key, expected_max);
967        assert_eq!(built.row_count, rows as u64);
968        // The last push closed the block, so `finish` appended nothing. The
969        // object must still be the same bytes a larger target produces.
970        assert_eq!(built.bytes, single_block.bytes);
971        let index =
972            decode_index_block(section(&built.bytes, &built.index), &built.index).expect("index");
973        assert_eq!(index.len(), 1);
974        assert_eq!(index[0].last_row_key, expected_max);
975    }
976
977    #[test]
978    fn block_geometry_does_not_change_the_segment_key_range() {
979        let rows = 400usize;
980        let expected: Vec<String> = (0..rows).map(|index| inode_row(index as u64).0).collect();
981        for target in [1usize, 64, 257, 1_024, 4_096, 65_536] {
982            let mut builder =
983                SegmentBlocksBuilder::new(NonZeroUsize::new(target).expect("positive target"));
984            for index in 0..rows {
985                let (key, filter_key, row) = inode_row(index as u64);
986                builder.push(&key, &filter_key, &row).expect("push row");
987            }
988            let built = builder.finish().expect("finish segment");
989            assert_eq!(built.min_row_key, expected[0], "target {target}");
990            assert_eq!(
991                &built.max_row_key,
992                expected.last().expect("rows"),
993                "target {target}"
994            );
995            assert_eq!(built.row_count, rows as u64, "target {target}");
996
997            let index = decode_index_block(section(&built.bytes, &built.index), &built.index)
998                .expect("index");
999            let mut recovered = Vec::new();
1000            for entry in &index {
1001                let block = decode_data_block(section(&built.bytes, &entry.block), &entry.block)
1002                    .expect("data block");
1003                // Each block still names its own last row key, so the index the
1004                // reader binary-searches keeps its shape.
1005                assert_eq!(
1006                    block.row_keys.last().expect("blocks are never empty"),
1007                    &entry.last_row_key,
1008                    "target {target}"
1009                );
1010                recovered.extend(block.row_keys);
1011            }
1012            assert_eq!(recovered, expected, "target {target}");
1013            assert_eq!(
1014                index.last().expect("blocks").last_row_key,
1015                built.max_row_key,
1016                "target {target}"
1017            );
1018        }
1019    }
1020
1021    #[test]
1022    fn a_descending_row_after_a_block_boundary_is_rejected() {
1023        let mut builder = SegmentBlocksBuilder::new(NonZeroUsize::MIN);
1024        let (key_high, filter_high, row_high) = inode_row(9);
1025        builder
1026            .push(&key_high, &filter_high, &row_high)
1027            .expect("first row");
1028        let (key_low, filter_low, row_low) = inode_row(3);
1029        let error = builder
1030            .push(&key_low, &filter_low, &row_low)
1031            .expect_err("a descending key across a block boundary should be rejected");
1032        assert!(
1033            matches!(
1034                &error,
1035                SstBlockCodecError::RowKeysOutOfOrder { previous, offered }
1036                    if previous == &key_high && offered == &key_low
1037            ),
1038            "unexpected error: {error}"
1039        );
1040    }
1041
1042    #[test]
1043    fn adjacent_equal_keys_are_permitted() {
1044        let mut builder = SegmentBlocksBuilder::default();
1045        let (key, filter_key, row) = inode_row(7);
1046        builder.push(&key, &filter_key, &row).expect("first copy");
1047        builder.push(&key, &filter_key, &row).expect("second copy");
1048        let built = builder.finish().expect("finish");
1049        assert_eq!(built.row_count, 2);
1050    }
1051
1052    #[test]
1053    fn corrupted_sections_fail_their_checksums() {
1054        let built = build_segment(200);
1055        let index =
1056            decode_index_block(section(&built.bytes, &built.index), &built.index).expect("index");
1057
1058        let mut corrupted = built.bytes.clone();
1059        let target = index[0].block.offset as usize + 3;
1060        corrupted[target] ^= 0xff;
1061        let error = decode_data_block(section(&corrupted, &index[0].block), &index[0].block)
1062            .expect_err("corrupted data block should fail");
1063        assert!(matches!(error, SstBlockCodecError::ChecksumMismatch { .. }));
1064
1065        let mut corrupted = built.bytes.clone();
1066        let target = built.index.offset as usize + 3;
1067        corrupted[target] ^= 0xff;
1068        let error = decode_index_block(section(&corrupted, &built.index), &built.index)
1069            .expect_err("corrupted index should fail");
1070        assert!(matches!(error, SstBlockCodecError::ChecksumMismatch { .. }));
1071
1072        let mut corrupted = built.bytes.clone();
1073        let target = built.filter.offset as usize + 12;
1074        corrupted[target] ^= 0xff;
1075        let error = decode_filter_block(section(&corrupted, &built.filter), &built.filter)
1076            .expect_err("corrupted filter should fail");
1077        assert!(matches!(error, SstBlockCodecError::ChecksumMismatch { .. }));
1078    }
1079
1080    #[test]
1081    fn block_expansion_stops_after_the_first_byte_beyond_the_declared_length() {
1082        let mut bytes = Vec::new();
1083        let mut handle = append_section(&mut bytes, &[0; 1024], true).expect("compress block");
1084        handle.decoded_len = 8;
1085        assert_eq!(
1086            decode_data_block(&bytes, &handle).expect_err("expansion exceeds the handle"),
1087            SstBlockCodecError::DecodedLengthMismatch {
1088                expected: 8,
1089                actual: 9,
1090            }
1091        );
1092    }
1093
1094    #[test]
1095    fn a_large_declared_length_does_not_require_a_large_initial_allocation() {
1096        let mut bytes = Vec::new();
1097        let mut handle = append_section(&mut bytes, &[0; 4], true).expect("compress block");
1098        handle.decoded_len = u32::MAX;
1099        assert_eq!(
1100            decode_index_block(&bytes, &handle).expect_err("declared length exceeds expansion"),
1101            SstBlockCodecError::DecodedLengthMismatch {
1102                expected: u32::MAX,
1103                actual: 4,
1104            }
1105        );
1106    }
1107
1108    #[test]
1109    fn truncated_index_and_filter_sections_are_rejected_before_decoding() {
1110        let built = shared_segment();
1111        let index = section(&built.bytes, &built.index);
1112        let filter = section(&built.bytes, &built.filter);
1113        for (handle, error) in [
1114            (
1115                built.index,
1116                decode_index_block(&index[..index.len() - 1], &built.index)
1117                    .expect_err("truncated index"),
1118            ),
1119            (
1120                built.filter,
1121                decode_filter_block(&filter[..filter.len() - 1], &built.filter)
1122                    .expect_err("truncated filter"),
1123            ),
1124        ] {
1125            assert_eq!(
1126                error,
1127                SstBlockCodecError::StoredLengthMismatch {
1128                    expected: handle.stored_len,
1129                    actual: handle.stored_len as usize - 1,
1130                }
1131            );
1132        }
1133    }
1134
1135    #[test]
1136    fn filter_headers_reject_zero_bits_and_invalid_hash_counts() {
1137        for (n_hashes, bit_len) in [(FILTER_HASH_COUNT, 0u64), (u32::MAX, 64), (0, 64)] {
1138            let mut payload = Vec::new();
1139            payload.extend_from_slice(&n_hashes.to_le_bytes());
1140            payload.extend_from_slice(&bit_len.to_le_bytes());
1141            payload.resize(12 + bit_len.div_ceil(8) as usize, 0xff);
1142            let mut bytes = Vec::new();
1143            let handle = append_section(&mut bytes, &payload, false).expect("filter section");
1144            assert!(matches!(
1145                decode_filter_block(&bytes, &handle),
1146                Err(SstBlockCodecError::Malformed(_))
1147            ));
1148        }
1149    }
1150
1151    #[test]
1152    fn restart_entries_bound_shared_prefix_reconstruction() {
1153        let mut payload = Vec::new();
1154        let mut restart_offsets = Vec::new();
1155        for index in 0..=RESTART_INTERVAL {
1156            if index % RESTART_INTERVAL == 0 {
1157                restart_offsets.push(payload.len() as u32);
1158            }
1159            write_varint(&mut payload, u64::from(index != 0));
1160            write_varint(&mut payload, u64::from(index == 0));
1161            if index == 0 {
1162                payload.push(b'a');
1163            }
1164            write_varint(&mut payload, 1);
1165            payload.push(0);
1166        }
1167        for offset in restart_offsets {
1168            payload.extend_from_slice(&offset.to_le_bytes());
1169        }
1170        payload.extend_from_slice(&2u32.to_le_bytes());
1171        let mut bytes = Vec::new();
1172        let handle = append_section(&mut bytes, &payload, true).expect("data section");
1173        assert_eq!(
1174            decode_data_block_rows::<u8>(&bytes, &handle).expect_err("shared restart prefix"),
1175            SstBlockCodecError::Malformed("restart entry has a shared prefix".to_owned())
1176        );
1177    }
1178
1179    #[test]
1180    fn restart_offsets_must_name_their_entries() {
1181        let payload = [0, 1, b'a', 1, 0, 1, 0, 0, 0, 1, 0, 0, 0];
1182        let mut bytes = Vec::new();
1183        let handle = append_section(&mut bytes, &payload, true).expect("data section");
1184        assert_eq!(
1185            decode_data_block_rows::<u8>(&bytes, &handle).expect_err("wrong restart offset"),
1186            SstBlockCodecError::Malformed("restart offset disagrees with its entry".to_owned())
1187        );
1188    }
1189
1190    #[test]
1191    fn varints_reject_overflow_in_the_tenth_byte() {
1192        let mut bytes = [0xff; 10];
1193        for last in [2, 0x80] {
1194            bytes[9] = last;
1195            assert_eq!(
1196                read_varint(&bytes, &mut 0).expect_err("overflowing varint"),
1197                SstBlockCodecError::Malformed("varint exceeds 64 bits".to_owned())
1198            );
1199        }
1200        bytes[9] = 1;
1201        assert_eq!(
1202            read_varint(&bytes, &mut 0).expect("largest varint"),
1203            u64::MAX
1204        );
1205    }
1206
1207    #[test]
1208    fn filter_has_no_false_negatives_and_few_false_positives() {
1209        let rows = 2_000;
1210        let built = build_segment(rows);
1211        let filter = decode_filter_block(section(&built.bytes, &built.filter), &built.filter)
1212            .expect("filter");
1213        for index in 0..rows {
1214            let (key, _, _) = inode_row(index as u64);
1215            assert!(filter.may_contain(&key), "inserted key must stay positive");
1216        }
1217        let mut false_positives = 0usize;
1218        let probes = 10_000usize;
1219        for index in 0..probes {
1220            let (absent, _, _) = inode_row((rows + 10_000 + index) as u64);
1221            if filter.may_contain(&absent) {
1222                false_positives += 1;
1223            }
1224        }
1225        let rate = false_positives as f64 / probes as f64;
1226        assert!(rate < 0.02, "false positive rate {rate} exceeds 2%");
1227    }
1228
1229    #[test]
1230    fn decoding_rejects_out_of_order_rows_in_a_block() {
1231        // A hostile block with descending keys and a valid CRC: encode two
1232        // full-key entries in the wrong order through the private helpers.
1233        let mut entries = Vec::new();
1234        for inode in [9u64, 3u64] {
1235            let (key, _, row) = inode_row(inode);
1236            let mut row_bytes = Vec::new();
1237            ciborium::ser::into_writer(&row, &mut row_bytes).expect("encode row");
1238            write_varint(&mut entries, 0);
1239            write_varint(&mut entries, key.len() as u64);
1240            entries.extend_from_slice(key.as_bytes());
1241            write_varint(&mut entries, row_bytes.len() as u64);
1242            entries.extend_from_slice(&row_bytes);
1243        }
1244        let mut payload = entries;
1245        payload.extend_from_slice(&0u32.to_le_bytes());
1246        payload.extend_from_slice(&1u32.to_le_bytes());
1247        let mut bytes = Vec::new();
1248        let handle = append_section(&mut bytes, &payload, true).expect("append section");
1249
1250        let error =
1251            decode_data_block(&bytes, &handle).expect_err("descending rows should be rejected");
1252        assert!(
1253            matches!(&error, SstBlockCodecError::Malformed(message) if message.contains("row-key order")),
1254            "unexpected error: {error}"
1255        );
1256    }
1257
1258    #[test]
1259    fn decoding_rejects_a_shared_prefix_inside_a_utf8_code_point() {
1260        let (_, _, row) = inode_row(1);
1261        let mut row_bytes = Vec::new();
1262        ciborium::ser::into_writer(&row, &mut row_bytes).expect("encode row");
1263        let mut payload = Vec::new();
1264        write_varint(&mut payload, 0);
1265        write_varint(&mut payload, "é".len() as u64);
1266        payload.extend_from_slice("é".as_bytes());
1267        write_varint(&mut payload, row_bytes.len() as u64);
1268        payload.extend_from_slice(&row_bytes);
1269        write_varint(&mut payload, 1);
1270        write_varint(&mut payload, 0);
1271        write_varint(&mut payload, row_bytes.len() as u64);
1272        payload.extend_from_slice(&row_bytes);
1273        payload.extend_from_slice(&0u32.to_le_bytes());
1274        payload.extend_from_slice(&1u32.to_le_bytes());
1275
1276        let mut bytes = Vec::new();
1277        let handle = append_section(&mut bytes, &payload, true).expect("append section");
1278        let error = decode_data_block(&bytes, &handle)
1279            .expect_err("a partial utf-8 prefix should be rejected");
1280        assert!(matches!(
1281            &error,
1282            SstBlockCodecError::Malformed(message)
1283                if message == "shared prefix exceeds previous key"
1284        ));
1285    }
1286
1287    #[test]
1288    fn decoding_rejects_out_of_order_index_entries() {
1289        let entries = vec![
1290            index_entry("inode-00000000000000000009", 0, 1),
1291            index_entry("inode-00000000000000000003", 0, 1),
1292        ];
1293        let (bytes, handle) = encode_index(&entries);
1294
1295        let error =
1296            decode_index_block(&bytes, &handle).expect_err("descending index should be rejected");
1297        assert!(
1298            matches!(&error, SstBlockCodecError::Malformed(message) if message.contains("key order")),
1299            "unexpected error: {error}"
1300        );
1301    }
1302
1303    #[test]
1304    fn decoding_rejects_out_of_order_index_offsets() {
1305        let entries = [index_entry("a", 10, 1), index_entry("b", 5, 1)];
1306        let (bytes, handle) = encode_index(&entries);
1307
1308        let error = decode_index_block(&bytes, &handle)
1309            .expect_err("descending block offsets should be rejected");
1310        assert!(
1311            matches!(&error, SstBlockCodecError::Malformed(message) if message.contains("does not start where")),
1312            "unexpected error: {error}"
1313        );
1314    }
1315
1316    #[test]
1317    fn decoding_rejects_overlapping_index_ranges() {
1318        let entries = [index_entry("a", 10, 5), index_entry("b", 14, 1)];
1319        let (bytes, handle) = encode_index(&entries);
1320
1321        let error = decode_index_block(&bytes, &handle)
1322            .expect_err("overlapping block ranges should be rejected");
1323        assert!(
1324            matches!(&error, SstBlockCodecError::Malformed(message) if message.contains("does not start where")),
1325            "unexpected error: {error}"
1326        );
1327    }
1328
1329    #[test]
1330    fn decoding_rejects_a_gap_between_index_blocks() {
1331        let entries = [index_entry("a", 0, 10), index_entry("b", 20, 1)];
1332        let (bytes, handle) = encode_index(&entries);
1333
1334        let error = decode_index_block(&bytes, &handle)
1335            .expect_err("a gap between block ranges should be rejected");
1336        assert!(
1337            matches!(&error, SstBlockCodecError::Malformed(message) if message.contains("does not start where")),
1338            "unexpected error: {error}"
1339        );
1340    }
1341
1342    #[test]
1343    fn decoding_rejects_a_single_block_range_past_the_integer_edge() {
1344        let entries = [index_entry("a", u64::MAX - 10, 100)];
1345        let (bytes, handle) = encode_index(&entries);
1346
1347        let error = decode_index_block(&bytes, &handle)
1348            .expect_err("an overflowing single range should be rejected");
1349        assert!(
1350            matches!(&error, SstBlockCodecError::Malformed(message) if message.contains("byte range overflows")),
1351            "unexpected error: {error}"
1352        );
1353    }
1354
1355    #[test]
1356    fn decoding_rejects_a_final_block_range_past_the_integer_edge() {
1357        let entries = [
1358            index_entry("a", u64::MAX - 110, 100),
1359            index_entry("b", u64::MAX - 10, 100),
1360        ];
1361        let (bytes, handle) = encode_index(&entries);
1362
1363        let error = decode_index_block(&bytes, &handle)
1364            .expect_err("a trailing overflowing range should be rejected");
1365        assert!(
1366            matches!(&error, SstBlockCodecError::Malformed(message) if message.contains("byte range overflows")),
1367            "unexpected error: {error}"
1368        );
1369    }
1370}