reddb_file/
file_format.rs

1//! Core persisted file-format constants.
2//!
3//! Runtime crates can own page management and columnar execution, but these
4//! durable magic/version values are file compatibility contracts.
5
6use std::fmt;
7
8/// Magic bytes for database file identification: `RDDB`.
9pub const PAGE_FILE_MAGIC: [u8; 4] = [0x52, 0x44, 0x44, 0x42];
10
11/// The first version that reserves the in-file double-write buffer zone.
12pub const PAGE_DWB_ZONE_FILE_VERSION: u32 = 0x0001_0001;
13/// The first version with durable page-ownership recovery.
14pub const PAGE_OWNERSHIP_FILE_VERSION: u32 = 0x0001_0002;
15/// Checkpoint before-images must be restored before following published roots.
16pub const PAGE_CHECKPOINT_UNDO_FILE_VERSION: u32 = 0x0001_0003;
17/// Current database file version. The phase-3 DWB layout itself starts at
18/// 1.0.1; older readers reject this version before using the new contract.
19pub const PAGE_DISK_RESIDENCY_FILE_VERSION: u32 = 0x0001_0004;
20/// Generation-bound secondary trees participate in catalog ownership/reclamation.
21/// Older readers must reject the file before recovering or recycling pages.
22pub const PAGE_SECONDARY_INDEX_FILE_VERSION: u32 = 0x0001_0005;
23/// Legacy recovery output: identity high-water marks are unknown and every
24/// writer must refuse this file. Older readers reject the version outright.
25pub const PAGE_RECOVERY_READ_ONLY_FILE_VERSION: u32 = 0x0001_0006;
26/// Result-cache epoch fencing for resident paged stores.
27pub const PAGE_RESULT_CACHE_FILE_VERSION: u32 = 0x0001_0007;
28/// Result-cache epoch fencing for disk-resident paged stores.
29pub const PAGE_RESULT_CACHE_DISK_FILE_VERSION: u32 = 0x0001_0008;
30/// The first version whose superblock declares mandatory reader capabilities.
31pub const PAGE_REQUIRED_CAPABILITIES_FILE_VERSION: u32 = 0x0001_0009;
32pub const PAGE_FILE_VERSION: u32 = PAGE_REQUIRED_CAPABILITIES_FILE_VERSION;
33
34/// Native measured-temperature values may appear in entity payloads.
35pub const FILE_CAPABILITY_TEMPERATURE: u64 = 1 << 0;
36/// The file's published trees require a disk-resident writer.
37pub const FILE_CAPABILITY_DISK_RESIDENCY: u64 = 1 << 1;
38/// Capabilities understood by this reader. Unknown mandatory bits are fatal.
39pub const FILE_CAPABILITIES_SUPPORTED: u64 =
40    FILE_CAPABILITY_TEMPERATURE | FILE_CAPABILITY_DISK_RESIDENCY;
41
42/// Whether a pre-capability format version implies disk residency.
43/// Capability-aware callers must use [`DatabaseHeader::requires_disk_residency`].
44pub fn page_format_is_disk_resident(version: u32) -> bool {
45    (PAGE_DISK_RESIDENCY_FILE_VERSION..PAGE_REQUIRED_CAPABILITIES_FILE_VERSION).contains(&version)
46        && version != PAGE_RESULT_CACHE_FILE_VERSION
47}
48
49/// Paged database page size.
50pub const PAGED_PAGE_SIZE: usize = 16_384;
51
52/// Paged database page-header size before the page-0 database header.
53pub const PAGED_PAGE_HEADER_SIZE: usize = 32;
54
55pub const PAGED_CELL_POINTER_SIZE: usize = 2;
56pub const PAGED_CELL_HEADER_SIZE: usize = 6;
57
58/// Raw persisted page header encoded at the start of every paged-store page.
59#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
60pub struct PagedPageHeader {
61    pub page_type: u8,
62    pub flags: u8,
63    pub cell_count: u16,
64    pub free_start: u16,
65    pub free_end: u16,
66    pub page_id: u32,
67    pub parent_id: u32,
68    pub right_child: u32,
69    pub lsn: u64,
70    pub checksum: u32,
71}
72
73pub fn encode_paged_page_header(header: &PagedPageHeader) -> [u8; PAGED_PAGE_HEADER_SIZE] {
74    let mut buf = [0u8; PAGED_PAGE_HEADER_SIZE];
75    buf[0] = header.page_type;
76    buf[1] = header.flags;
77    buf[2..4].copy_from_slice(&header.cell_count.to_le_bytes());
78    buf[4..6].copy_from_slice(&header.free_start.to_le_bytes());
79    buf[6..8].copy_from_slice(&header.free_end.to_le_bytes());
80    buf[8..12].copy_from_slice(&header.page_id.to_le_bytes());
81    buf[12..16].copy_from_slice(&header.parent_id.to_le_bytes());
82    buf[16..20].copy_from_slice(&header.right_child.to_le_bytes());
83    buf[20..28].copy_from_slice(&header.lsn.to_le_bytes());
84    buf[28..32].copy_from_slice(&header.checksum.to_le_bytes());
85    buf
86}
87
88pub fn decode_paged_page_header(buf: &[u8; PAGED_PAGE_HEADER_SIZE]) -> PagedPageHeader {
89    PagedPageHeader {
90        page_type: buf[0],
91        flags: buf[1],
92        cell_count: u16::from_le_bytes([buf[2], buf[3]]),
93        free_start: u16::from_le_bytes([buf[4], buf[5]]),
94        free_end: u16::from_le_bytes([buf[6], buf[7]]),
95        page_id: u32::from_le_bytes([buf[8], buf[9], buf[10], buf[11]]),
96        parent_id: u32::from_le_bytes([buf[12], buf[13], buf[14], buf[15]]),
97        right_child: u32::from_le_bytes([buf[16], buf[17], buf[18], buf[19]]),
98        lsn: u64::from_le_bytes([
99            buf[20], buf[21], buf[22], buf[23], buf[24], buf[25], buf[26], buf[27],
100        ]),
101        checksum: u32::from_le_bytes([buf[28], buf[29], buf[30], buf[31]]),
102    }
103}
104
105pub fn paged_page_type(page: &[u8; PAGED_PAGE_SIZE]) -> u8 {
106    page[0]
107}
108
109pub fn paged_page_id(page: &[u8; PAGED_PAGE_SIZE]) -> u32 {
110    read_u32(page, 8).expect("paged page has header")
111}
112
113pub fn paged_page_lsn(page: &[u8; PAGED_PAGE_SIZE]) -> u64 {
114    read_u64(page, 20).expect("paged page has header")
115}
116
117pub fn set_paged_page_lsn(page: &mut [u8; PAGED_PAGE_SIZE], lsn: u64) {
118    write_u64(page, 20, lsn).expect("paged page has header");
119}
120
121pub fn paged_page_cell_count(page: &[u8; PAGED_PAGE_SIZE]) -> u16 {
122    read_u16(page, 2).expect("paged page has header")
123}
124
125pub fn set_paged_page_cell_count(page: &mut [u8; PAGED_PAGE_SIZE], count: u16) {
126    write_u16(page, 2, count).expect("paged page has header");
127}
128
129pub fn paged_page_parent_id(page: &[u8; PAGED_PAGE_SIZE]) -> u32 {
130    read_u32(page, 12).expect("paged page has header")
131}
132
133pub fn set_paged_page_parent_id(page: &mut [u8; PAGED_PAGE_SIZE], parent_id: u32) {
134    write_u32(page, 12, parent_id).expect("paged page has header");
135}
136
137pub fn paged_page_right_child(page: &[u8; PAGED_PAGE_SIZE]) -> u32 {
138    read_u32(page, 16).expect("paged page has header")
139}
140
141pub fn set_paged_page_right_child(page: &mut [u8; PAGED_PAGE_SIZE], child_id: u32) {
142    write_u32(page, 16, child_id).expect("paged page has header");
143}
144
145pub fn paged_page_free_start(page: &[u8; PAGED_PAGE_SIZE]) -> u16 {
146    read_u16(page, 4).expect("paged page has header")
147}
148
149pub fn set_paged_page_free_start(page: &mut [u8; PAGED_PAGE_SIZE], offset: u16) {
150    write_u16(page, 4, offset).expect("paged page has header");
151}
152
153pub fn paged_page_free_end(page: &[u8; PAGED_PAGE_SIZE]) -> u16 {
154    read_u16(page, 6).expect("paged page has header")
155}
156
157pub fn set_paged_page_free_end(page: &mut [u8; PAGED_PAGE_SIZE], offset: u16) {
158    write_u16(page, 6, offset).expect("paged page has header");
159}
160
161pub fn paged_page_checksum(page: &[u8; PAGED_PAGE_SIZE]) -> u32 {
162    read_u32(page, 28).expect("paged page has header")
163}
164
165pub fn clear_paged_page_checksum(page: &mut [u8; PAGED_PAGE_SIZE]) {
166    write_u32(page, 28, 0).expect("paged page has header");
167}
168
169pub fn set_paged_page_checksum(page: &mut [u8; PAGED_PAGE_SIZE], checksum: u32) {
170    write_u32(page, 28, checksum).expect("paged page has header");
171}
172
173pub fn paged_cell_pointer_offset(index: usize) -> Option<usize> {
174    let offset = PAGED_PAGE_HEADER_SIZE.checked_add(index.checked_mul(PAGED_CELL_POINTER_SIZE)?)?;
175    (offset + PAGED_CELL_POINTER_SIZE <= PAGED_PAGE_SIZE).then_some(offset)
176}
177
178pub fn paged_cell_pointer(page: &[u8; PAGED_PAGE_SIZE], index: usize) -> Option<u16> {
179    read_u16(page, paged_cell_pointer_offset(index)?).ok()
180}
181
182pub fn set_paged_cell_pointer(
183    page: &mut [u8; PAGED_PAGE_SIZE],
184    index: usize,
185    pointer: u16,
186) -> bool {
187    let Some(offset) = paged_cell_pointer_offset(index) else {
188        return false;
189    };
190    if !paged_cell_pointer_is_valid(pointer) {
191        return false;
192    }
193    write_u16(page, offset, pointer).is_ok()
194}
195
196pub fn paged_cell_pointer_is_valid(pointer: u16) -> bool {
197    let pointer = pointer as usize;
198    (PAGED_PAGE_HEADER_SIZE..PAGED_PAGE_SIZE).contains(&pointer)
199}
200
201pub fn paged_cell_len(key_len: usize, value_len: usize) -> Option<usize> {
202    PAGED_CELL_HEADER_SIZE
203        .checked_add(key_len)?
204        .checked_add(value_len)
205}
206
207pub fn paged_cell_total_len(page: &[u8; PAGED_PAGE_SIZE], pointer: u16) -> Option<usize> {
208    let pointer = pointer as usize;
209    if pointer + PAGED_CELL_HEADER_SIZE > PAGED_PAGE_SIZE {
210        return None;
211    }
212    let key_len = read_u16(page, pointer).ok()? as usize;
213    let value_len = read_u32(page, pointer + 2).ok()? as usize;
214    let total_len = paged_cell_len(key_len, value_len)?;
215    (pointer + total_len <= PAGED_PAGE_SIZE).then_some(total_len)
216}
217
218pub fn paged_cell_bytes(page: &[u8; PAGED_PAGE_SIZE], pointer: u16) -> Option<&[u8]> {
219    let total_len = paged_cell_total_len(page, pointer)?;
220    let pointer = pointer as usize;
221    Some(&page[pointer..pointer + total_len])
222}
223
224pub fn paged_cell_key_value(cell: &[u8]) -> Option<(&[u8], &[u8])> {
225    if cell.len() < PAGED_CELL_HEADER_SIZE {
226        return None;
227    }
228    let key_len = u16::from_le_bytes(cell[0..2].try_into().ok()?) as usize;
229    let value_len = u32::from_le_bytes(cell[2..6].try_into().ok()?) as usize;
230    let total_len = paged_cell_len(key_len, value_len)?;
231    if total_len > cell.len() {
232        return None;
233    }
234    let key_start = PAGED_CELL_HEADER_SIZE;
235    let value_start = key_start + key_len;
236    Some((
237        &cell[key_start..value_start],
238        &cell[value_start..value_start + value_len],
239    ))
240}
241
242pub fn write_paged_cell(
243    page: &mut [u8; PAGED_PAGE_SIZE],
244    offset: u16,
245    key: &[u8],
246    value: &[u8],
247) -> bool {
248    let Ok(key_len) = u16::try_from(key.len()) else {
249        return false;
250    };
251    let Ok(value_len) = u32::try_from(value.len()) else {
252        return false;
253    };
254    let Some(total_len) = paged_cell_len(key.len(), value.len()) else {
255        return false;
256    };
257    let offset = offset as usize;
258    if offset + total_len > PAGED_PAGE_SIZE {
259        return false;
260    }
261
262    page[offset..offset + 2].copy_from_slice(&key_len.to_le_bytes());
263    page[offset + 2..offset + 6].copy_from_slice(&value_len.to_le_bytes());
264    page[offset + PAGED_CELL_HEADER_SIZE..offset + PAGED_CELL_HEADER_SIZE + key.len()]
265        .copy_from_slice(key);
266    page[offset + PAGED_CELL_HEADER_SIZE + key.len()..offset + total_len].copy_from_slice(value);
267    true
268}
269
270/// Encryption marker embedded in page 0 when page encryption is enabled.
271pub const PAGED_ENCRYPTION_MARKER: [u8; 4] = *b"RDBE";
272
273/// Current page-0 encryption marker offset.
274///
275/// This preserves the existing on-disk placement. It overlaps the historical
276/// physical-header region, so callers must preserve page-0 bytes around normal
277/// header writes.
278pub const PAGED_ENCRYPTION_MARKER_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 32;
279
280pub const PAGED_ENCRYPTION_SALT_SIZE: usize = 32;
281pub const PAGED_ENCRYPTION_KEY_CHECK_PLAINTEXT_SIZE: usize = 32;
282pub const PAGED_ENCRYPTION_KEY_CHECK_BLOB_SIZE: usize = 60;
283pub const PAGED_ENCRYPTION_HEADER_SIZE: usize =
284    PAGED_ENCRYPTION_SALT_SIZE + PAGED_ENCRYPTION_KEY_CHECK_BLOB_SIZE;
285
286#[derive(Debug, Clone, PartialEq, Eq)]
287pub struct PagedEncryptionHeader {
288    pub salt: [u8; PAGED_ENCRYPTION_SALT_SIZE],
289    pub key_check: Vec<u8>,
290}
291
292pub fn encode_paged_encryption_header(header: &PagedEncryptionHeader) -> Vec<u8> {
293    let mut out = Vec::with_capacity(PAGED_ENCRYPTION_HEADER_SIZE);
294    out.extend_from_slice(&header.salt);
295    out.extend_from_slice(&header.key_check);
296    out
297}
298
299pub fn decode_paged_encryption_header(
300    data: &[u8],
301) -> Result<PagedEncryptionHeader, DatabaseHeaderError> {
302    ensure_len(data, PAGED_ENCRYPTION_HEADER_SIZE)?;
303    let mut salt = [0u8; PAGED_ENCRYPTION_SALT_SIZE];
304    salt.copy_from_slice(&data[..PAGED_ENCRYPTION_SALT_SIZE]);
305    let key_check = data[PAGED_ENCRYPTION_SALT_SIZE..PAGED_ENCRYPTION_HEADER_SIZE].to_vec();
306    Ok(PagedEncryptionHeader { salt, key_check })
307}
308
309pub fn paged_encryption_marker_present(page: &[u8]) -> bool {
310    page.get(PAGED_ENCRYPTION_MARKER_OFFSET..PAGED_ENCRYPTION_MARKER_OFFSET + 4)
311        == Some(&PAGED_ENCRYPTION_MARKER)
312}
313
314pub fn paged_encryption_header_bytes(page: &[u8]) -> Option<&[u8]> {
315    let start = PAGED_ENCRYPTION_MARKER_OFFSET + PAGED_ENCRYPTION_MARKER.len();
316    page.get(start..start + PAGED_ENCRYPTION_HEADER_SIZE)
317}
318
319pub fn write_paged_encryption_marker_and_header(
320    page: &mut [u8],
321    header_bytes: &[u8],
322) -> Result<(), DatabaseHeaderError> {
323    let marker_end = PAGED_ENCRYPTION_MARKER_OFFSET + PAGED_ENCRYPTION_MARKER.len();
324    let header_end = marker_end + header_bytes.len();
325    ensure_len(page, header_end)?;
326    page[PAGED_ENCRYPTION_MARKER_OFFSET..marker_end].copy_from_slice(&PAGED_ENCRYPTION_MARKER);
327    page[marker_end..header_end].copy_from_slice(header_bytes);
328    Ok(())
329}
330
331const DB_MAGIC_OFFSET: usize = PAGED_PAGE_HEADER_SIZE;
332const DB_VERSION_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 4;
333const DB_PAGE_SIZE_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 8;
334const DB_PAGE_COUNT_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 12;
335const DB_FREELIST_HEAD_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 16;
336const DB_SCHEMA_VERSION_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 20;
337const DB_CHECKPOINT_LSN_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 24;
338const DB_PHYSICAL_FORMAT_VERSION_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 32;
339const DB_PHYSICAL_SEQUENCE_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 36;
340const DB_MANIFEST_ROOT_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 44;
341const DB_MANIFEST_OLDEST_ROOT_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 52;
342const DB_FREE_SET_ROOT_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 60;
343const DB_MANIFEST_PAGE_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 68;
344const DB_MANIFEST_CHECKSUM_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 72;
345const DB_COLLECTION_ROOTS_PAGE_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 80;
346const DB_COLLECTION_ROOTS_CHECKSUM_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 84;
347const DB_COLLECTION_ROOT_COUNT_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 92;
348const DB_SNAPSHOT_COUNT_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 96;
349const DB_INDEX_COUNT_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 100;
350const DB_CATALOG_COLLECTION_COUNT_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 104;
351const DB_CATALOG_TOTAL_ENTITIES_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 108;
352const DB_EXPORT_COUNT_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 116;
353const DB_GRAPH_PROJECTION_COUNT_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 120;
354const DB_ANALYTICS_JOB_COUNT_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 124;
355const DB_MANIFEST_EVENT_COUNT_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 128;
356const DB_REGISTRY_PAGE_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 132;
357const DB_REGISTRY_CHECKSUM_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 136;
358const DB_RECOVERY_PAGE_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 144;
359const DB_RECOVERY_CHECKSUM_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 148;
360const DB_CATALOG_PAGE_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 156;
361const DB_CATALOG_CHECKSUM_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 160;
362const DB_METADATA_STATE_PAGE_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 168;
363const DB_METADATA_STATE_CHECKSUM_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 172;
364const DB_VECTOR_ARTIFACT_PAGE_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 180;
365const DB_VECTOR_ARTIFACT_CHECKSUM_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 184;
366const DB_CHECKPOINT_IN_PROGRESS_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 192;
367const DB_CHECKPOINT_TARGET_LSN_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 193;
368const DB_REQUIRED_CAPABILITIES_OFFSET: usize = PAGED_PAGE_HEADER_SIZE + 208;
369const DB_HEADER_MIN_LEN: usize = DB_REQUIRED_CAPABILITIES_OFFSET + 8;
370
371/// Double-write-buffer file magic: `RDDW`.
372pub const DWB_MAGIC: [u8; 4] = [0x52, 0x44, 0x44, 0x57];
373pub const PAGED_DWB_HEADER_SIZE: usize = 12;
374pub const PAGED_DWB_ENTRY_HEADER_SIZE: usize = 4;
375pub const PAGED_DWB_ENTRY_SIZE: usize = PAGED_DWB_ENTRY_HEADER_SIZE + PAGED_PAGE_SIZE;
376
377#[derive(Debug, Clone, PartialEq, Eq)]
378pub struct PagedDwbEntry {
379    pub page_id: u32,
380    pub page: [u8; PAGED_PAGE_SIZE],
381}
382
383#[derive(Debug, Clone, PartialEq, Eq)]
384pub enum PagedDwbFrameError {
385    ShortHeader { got: usize },
386    InvalidMagic,
387    IncompleteFrame { expected: usize, got: usize },
388    ChecksumMismatch { expected: u32, actual: u32 },
389}
390
391impl fmt::Display for PagedDwbFrameError {
392    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
393        match self {
394            Self::ShortHeader { got } => write!(f, "DWB frame too short: got {got} bytes"),
395            Self::InvalidMagic => write!(f, "invalid DWB frame magic"),
396            Self::IncompleteFrame { expected, got } => {
397                write!(
398                    f,
399                    "incomplete DWB frame: expected {expected} bytes, got {got}"
400                )
401            }
402            Self::ChecksumMismatch { expected, actual } => write!(
403                f,
404                "DWB frame checksum mismatch: expected {expected}, actual {actual}"
405            ),
406        }
407    }
408}
409
410impl std::error::Error for PagedDwbFrameError {}
411
412pub fn encode_paged_dwb_header(count: u32, checksum: u32) -> [u8; PAGED_DWB_HEADER_SIZE] {
413    let mut header = [0; PAGED_DWB_HEADER_SIZE];
414    header[..4].copy_from_slice(&DWB_MAGIC);
415    header[4..8].copy_from_slice(&count.to_le_bytes());
416    header[8..12].copy_from_slice(&checksum.to_le_bytes());
417    header
418}
419
420pub fn decode_paged_dwb_header(bytes: &[u8]) -> Result<(u32, u32), PagedDwbFrameError> {
421    if bytes.len() < PAGED_DWB_HEADER_SIZE {
422        return Err(PagedDwbFrameError::ShortHeader { got: bytes.len() });
423    }
424    if bytes[..4] != DWB_MAGIC {
425        return Err(PagedDwbFrameError::InvalidMagic);
426    }
427    Ok((
428        u32::from_le_bytes(bytes[4..8].try_into().expect("header length checked")),
429        u32::from_le_bytes(bytes[8..12].try_into().expect("header length checked")),
430    ))
431}
432
433pub fn encode_paged_dwb_frame<'a>(
434    pages: impl IntoIterator<Item = (u32, &'a [u8; PAGED_PAGE_SIZE])>,
435) -> Vec<u8> {
436    let pages: Vec<_> = pages.into_iter().collect();
437    let total = PAGED_DWB_HEADER_SIZE + pages.len() * PAGED_DWB_ENTRY_SIZE;
438    let mut out = Vec::with_capacity(total);
439
440    out.extend_from_slice(&encode_paged_dwb_header(
441        u32::try_from(pages.len()).expect("DWB count fits u32"),
442        0,
443    ));
444
445    for (page_id, page) in pages {
446        out.extend_from_slice(&page_id.to_le_bytes());
447        out.extend_from_slice(page);
448    }
449
450    let checksum = crc32(&out[PAGED_DWB_HEADER_SIZE..]);
451    out[8..12].copy_from_slice(&checksum.to_le_bytes());
452    out
453}
454
455pub fn decode_paged_dwb_frame(bytes: &[u8]) -> Result<Vec<PagedDwbEntry>, PagedDwbFrameError> {
456    let (count, stored_checksum) = decode_paged_dwb_header(bytes)?;
457    let count = count as usize;
458    let expected_len = PAGED_DWB_HEADER_SIZE + count * PAGED_DWB_ENTRY_SIZE;
459    if bytes.len() < expected_len {
460        return Err(PagedDwbFrameError::IncompleteFrame {
461            expected: expected_len,
462            got: bytes.len(),
463        });
464    }
465
466    let actual_checksum = crc32(&bytes[PAGED_DWB_HEADER_SIZE..expected_len]);
467    if actual_checksum != stored_checksum {
468        return Err(PagedDwbFrameError::ChecksumMismatch {
469            expected: stored_checksum,
470            actual: actual_checksum,
471        });
472    }
473
474    let mut offset = PAGED_DWB_HEADER_SIZE;
475    let mut entries = Vec::with_capacity(count);
476    for _ in 0..count {
477        let page_id =
478            u32::from_le_bytes(bytes[offset..offset + 4].try_into().expect("len checked"));
479        offset += PAGED_DWB_ENTRY_HEADER_SIZE;
480        let mut page = [0u8; PAGED_PAGE_SIZE];
481        page.copy_from_slice(&bytes[offset..offset + PAGED_PAGE_SIZE]);
482        offset += PAGED_PAGE_SIZE;
483        entries.push(PagedDwbEntry { page_id, page });
484    }
485    Ok(entries)
486}
487
488fn crc32(data: &[u8]) -> u32 {
489    crc32fast::hash(data)
490}
491
492/// `b"RDCC"` — RedDB Columnar Chunk. Opens and closes every column block.
493pub const COLUMN_BLOCK_MAGIC: [u8; 4] = *b"RDCC";
494
495/// Column block on-disk format version.
496pub const COLUMN_BLOCK_VERSION_V1: u16 = 1;
497
498/// Reusable segment-level bloom frame v2 magic.
499pub const BLOOM_SEGMENT_V2_MAGIC: u8 = 0xC0;
500
501/// Legacy vector B-tree on-disk page format.
502pub const VECTOR_BTREE_FORMAT_VERSION_V1: u16 = 1;
503
504/// Current vector B-tree on-disk page format.
505pub const VECTOR_BTREE_FORMAT_VERSION_V2: u16 = 2;
506
507/// Vector B-tree format stamped into freshly-written page headers.
508pub const VECTOR_BTREE_FORMAT_VERSION: u16 = VECTOR_BTREE_FORMAT_VERSION_V2;
509
510/// Database file header information persisted in page 0 after the page header.
511#[derive(Debug, Clone, PartialEq, Eq)]
512pub struct DatabaseHeader {
513    pub version: u32,
514    pub page_size: u32,
515    pub page_count: u32,
516    pub freelist_head: u32,
517    pub schema_version: u32,
518    pub checkpoint_lsn: u64,
519    pub checkpoint_in_progress: bool,
520    pub checkpoint_target_lsn: u64,
521    /// Monotonic set of byte-level features required to decode this file.
522    pub required_capabilities: u64,
523    pub physical: PhysicalFileHeader,
524}
525
526impl DatabaseHeader {
527    /// Whether the file requires a writer that participates in the bounded
528    /// disk-residency protocol.
529    ///
530    /// Versions before the capability bitmap encoded that requirement in the
531    /// monotonically increasing version number. Newer files declare it as an
532    /// independent capability so unrelated codecs can evolve the format.
533    pub fn requires_disk_residency(&self) -> bool {
534        if self.version < PAGE_REQUIRED_CAPABILITIES_FILE_VERSION {
535            page_format_is_disk_resident(self.version)
536        } else {
537            self.required_capabilities & FILE_CAPABILITY_DISK_RESIDENCY != 0
538        }
539    }
540}
541
542impl Default for DatabaseHeader {
543    fn default() -> Self {
544        Self {
545            // Prepared disk residency is an explicit host opt-in.
546            version: PAGE_CHECKPOINT_UNDO_FILE_VERSION,
547            page_size: PAGED_PAGE_SIZE as u32,
548            page_count: 1,
549            freelist_head: 0,
550            schema_version: 0,
551            checkpoint_lsn: 0,
552            checkpoint_in_progress: false,
553            checkpoint_target_lsn: 0,
554            required_capabilities: 0,
555            physical: PhysicalFileHeader::default(),
556        }
557    }
558}
559
560/// Minimal physical state mirrored into page 0 for paged databases.
561///
562/// The pager owns when this is read and written, but the field layout is a
563/// durable file compatibility contract.
564#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
565pub struct PhysicalFileHeader {
566    pub format_version: u32,
567    pub sequence: u64,
568    pub manifest_oldest_root: u64,
569    pub manifest_root: u64,
570    pub free_set_root: u64,
571    pub manifest_page: u32,
572    pub manifest_checksum: u64,
573    pub collection_roots_page: u32,
574    pub collection_roots_checksum: u64,
575    pub collection_root_count: u32,
576    pub snapshot_count: u32,
577    pub index_count: u32,
578    pub catalog_collection_count: u32,
579    pub catalog_total_entities: u64,
580    pub export_count: u32,
581    pub graph_projection_count: u32,
582    pub analytics_job_count: u32,
583    pub manifest_event_count: u32,
584    pub registry_page: u32,
585    pub registry_checksum: u64,
586    pub recovery_page: u32,
587    pub recovery_checksum: u64,
588    pub catalog_page: u32,
589    pub catalog_checksum: u64,
590    pub metadata_state_page: u32,
591    pub metadata_state_checksum: u64,
592    pub vector_artifact_page: u32,
593    pub vector_artifact_checksum: u64,
594}
595
596#[derive(Debug, Clone, PartialEq, Eq)]
597pub enum DatabaseHeaderError {
598    ShortPage { need: usize, got: usize },
599    InvalidMagic,
600    UnsupportedPageSize(u32),
601    UnsupportedDatabaseVersion { file_version: u32, supported: u32 },
602    UnsupportedRequiredCapabilities(u64),
603}
604
605impl fmt::Display for DatabaseHeaderError {
606    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
607        match self {
608            Self::ShortPage { need, got } => {
609                write!(f, "database header page too short: need {need} bytes, got {got}")
610            }
611            Self::InvalidMagic => write!(f, "invalid database header magic"),
612            Self::UnsupportedPageSize(size) => write!(f, "Unsupported page size: {size}"),
613            Self::UnsupportedDatabaseVersion {
614                file_version,
615                supported,
616            } => write!(
617                f,
618                "Unsupported database version: file version {file_version} is newer than supported {supported}"
619            ),
620            Self::UnsupportedRequiredCapabilities(bits) => write!(
621                f,
622                "Unsupported required database capabilities: {bits:#018x}"
623            ),
624        }
625    }
626}
627
628impl std::error::Error for DatabaseHeaderError {}
629
630pub fn database_header_magic_matches(page: &[u8]) -> bool {
631    page.get(DB_MAGIC_OFFSET..DB_MAGIC_OFFSET + PAGE_FILE_MAGIC.len()) == Some(&PAGE_FILE_MAGIC)
632}
633
634pub fn init_database_header_page(
635    page: &mut [u8],
636    page_count: u32,
637) -> Result<(), DatabaseHeaderError> {
638    encode_database_header(
639        page,
640        &DatabaseHeader {
641            page_count,
642            ..DatabaseHeader::default()
643        },
644    )
645}
646
647pub fn database_header_page_count(page: &[u8]) -> Result<u32, DatabaseHeaderError> {
648    read_u32(page, DB_PAGE_COUNT_OFFSET)
649}
650
651pub fn set_database_header_version(
652    page: &mut [u8],
653    version: u32,
654) -> Result<(), DatabaseHeaderError> {
655    write_u32(page, DB_VERSION_OFFSET, version)
656}
657
658pub fn set_database_header_page_count(
659    page: &mut [u8],
660    page_count: u32,
661) -> Result<(), DatabaseHeaderError> {
662    write_u32(page, DB_PAGE_COUNT_OFFSET, page_count)
663}
664
665pub fn database_header_freelist_head(page: &[u8]) -> Result<u32, DatabaseHeaderError> {
666    read_u32(page, DB_FREELIST_HEAD_OFFSET)
667}
668
669pub fn set_database_header_freelist_head(
670    page: &mut [u8],
671    page_id: u32,
672) -> Result<(), DatabaseHeaderError> {
673    write_u32(page, DB_FREELIST_HEAD_OFFSET, page_id)
674}
675
676pub fn set_database_header_required_capabilities(
677    page: &mut [u8],
678    capabilities: u64,
679) -> Result<(), DatabaseHeaderError> {
680    write_u64(page, DB_REQUIRED_CAPABILITIES_OFFSET, capabilities)
681}
682
683pub fn database_header_page_size(page: &[u8]) -> Result<u32, DatabaseHeaderError> {
684    read_u32(page, DB_PAGE_SIZE_OFFSET)
685}
686
687pub fn decode_database_header(page: &[u8]) -> Result<DatabaseHeader, DatabaseHeaderError> {
688    ensure_len(page, DB_HEADER_MIN_LEN)?;
689    if !database_header_magic_matches(page) {
690        return Err(DatabaseHeaderError::InvalidMagic);
691    }
692
693    let version = read_u32(page, DB_VERSION_OFFSET)?;
694    let page_size = read_u32(page, DB_PAGE_SIZE_OFFSET)?;
695    if page_size != PAGED_PAGE_SIZE as u32 {
696        return Err(DatabaseHeaderError::UnsupportedPageSize(page_size));
697    }
698    if version > PAGE_FILE_VERSION {
699        return Err(DatabaseHeaderError::UnsupportedDatabaseVersion {
700            file_version: version,
701            supported: PAGE_FILE_VERSION,
702        });
703    }
704    let required_capabilities = if version >= PAGE_REQUIRED_CAPABILITIES_FILE_VERSION {
705        read_u64(page, DB_REQUIRED_CAPABILITIES_OFFSET)?
706    } else {
707        0
708    };
709    let unsupported_capabilities = required_capabilities & !FILE_CAPABILITIES_SUPPORTED;
710    if unsupported_capabilities != 0 {
711        return Err(DatabaseHeaderError::UnsupportedRequiredCapabilities(
712            unsupported_capabilities,
713        ));
714    }
715
716    Ok(DatabaseHeader {
717        version,
718        page_size,
719        page_count: read_u32(page, DB_PAGE_COUNT_OFFSET)?,
720        freelist_head: read_u32(page, DB_FREELIST_HEAD_OFFSET)?,
721        schema_version: read_u32(page, DB_SCHEMA_VERSION_OFFSET)?,
722        checkpoint_lsn: read_u64(page, DB_CHECKPOINT_LSN_OFFSET)?,
723        checkpoint_in_progress: page[DB_CHECKPOINT_IN_PROGRESS_OFFSET] != 0,
724        checkpoint_target_lsn: read_u64(page, DB_CHECKPOINT_TARGET_LSN_OFFSET)?,
725        required_capabilities,
726        physical: PhysicalFileHeader {
727            format_version: read_u32(page, DB_PHYSICAL_FORMAT_VERSION_OFFSET)?,
728            sequence: read_u64(page, DB_PHYSICAL_SEQUENCE_OFFSET)?,
729            manifest_oldest_root: read_u64(page, DB_MANIFEST_OLDEST_ROOT_OFFSET)?,
730            manifest_root: read_u64(page, DB_MANIFEST_ROOT_OFFSET)?,
731            free_set_root: read_u64(page, DB_FREE_SET_ROOT_OFFSET)?,
732            manifest_page: read_u32(page, DB_MANIFEST_PAGE_OFFSET)?,
733            manifest_checksum: read_u64(page, DB_MANIFEST_CHECKSUM_OFFSET)?,
734            collection_roots_page: read_u32(page, DB_COLLECTION_ROOTS_PAGE_OFFSET)?,
735            collection_roots_checksum: read_u64(page, DB_COLLECTION_ROOTS_CHECKSUM_OFFSET)?,
736            collection_root_count: read_u32(page, DB_COLLECTION_ROOT_COUNT_OFFSET)?,
737            snapshot_count: read_u32(page, DB_SNAPSHOT_COUNT_OFFSET)?,
738            index_count: read_u32(page, DB_INDEX_COUNT_OFFSET)?,
739            catalog_collection_count: read_u32(page, DB_CATALOG_COLLECTION_COUNT_OFFSET)?,
740            catalog_total_entities: read_u64(page, DB_CATALOG_TOTAL_ENTITIES_OFFSET)?,
741            export_count: read_u32(page, DB_EXPORT_COUNT_OFFSET)?,
742            graph_projection_count: read_u32(page, DB_GRAPH_PROJECTION_COUNT_OFFSET)?,
743            analytics_job_count: read_u32(page, DB_ANALYTICS_JOB_COUNT_OFFSET)?,
744            manifest_event_count: read_u32(page, DB_MANIFEST_EVENT_COUNT_OFFSET)?,
745            registry_page: read_u32(page, DB_REGISTRY_PAGE_OFFSET)?,
746            registry_checksum: read_u64(page, DB_REGISTRY_CHECKSUM_OFFSET)?,
747            recovery_page: read_u32(page, DB_RECOVERY_PAGE_OFFSET)?,
748            recovery_checksum: read_u64(page, DB_RECOVERY_CHECKSUM_OFFSET)?,
749            catalog_page: read_u32(page, DB_CATALOG_PAGE_OFFSET)?,
750            catalog_checksum: read_u64(page, DB_CATALOG_CHECKSUM_OFFSET)?,
751            metadata_state_page: read_u32(page, DB_METADATA_STATE_PAGE_OFFSET)?,
752            metadata_state_checksum: read_u64(page, DB_METADATA_STATE_CHECKSUM_OFFSET)?,
753            vector_artifact_page: read_u32(page, DB_VECTOR_ARTIFACT_PAGE_OFFSET)?,
754            vector_artifact_checksum: read_u64(page, DB_VECTOR_ARTIFACT_CHECKSUM_OFFSET)?,
755        },
756    })
757}
758
759pub fn encode_database_header(
760    page: &mut [u8],
761    header: &DatabaseHeader,
762) -> Result<(), DatabaseHeaderError> {
763    ensure_len(page, DB_HEADER_MIN_LEN)?;
764    page[DB_MAGIC_OFFSET..DB_MAGIC_OFFSET + PAGE_FILE_MAGIC.len()]
765        .copy_from_slice(&PAGE_FILE_MAGIC);
766    write_u32(page, DB_VERSION_OFFSET, header.version)?;
767    write_u32(page, DB_PAGE_SIZE_OFFSET, header.page_size)?;
768    write_u32(page, DB_PAGE_COUNT_OFFSET, header.page_count)?;
769    write_u32(page, DB_FREELIST_HEAD_OFFSET, header.freelist_head)?;
770    write_u32(page, DB_SCHEMA_VERSION_OFFSET, header.schema_version)?;
771    write_u64(page, DB_CHECKPOINT_LSN_OFFSET, header.checkpoint_lsn)?;
772    write_u32(
773        page,
774        DB_PHYSICAL_FORMAT_VERSION_OFFSET,
775        header.physical.format_version,
776    )?;
777    write_u64(page, DB_PHYSICAL_SEQUENCE_OFFSET, header.physical.sequence)?;
778    write_u64(page, DB_MANIFEST_ROOT_OFFSET, header.physical.manifest_root)?;
779    write_u64(
780        page,
781        DB_MANIFEST_OLDEST_ROOT_OFFSET,
782        header.physical.manifest_oldest_root,
783    )?;
784    write_u64(page, DB_FREE_SET_ROOT_OFFSET, header.physical.free_set_root)?;
785    write_u32(page, DB_MANIFEST_PAGE_OFFSET, header.physical.manifest_page)?;
786    write_u64(
787        page,
788        DB_MANIFEST_CHECKSUM_OFFSET,
789        header.physical.manifest_checksum,
790    )?;
791    write_u32(
792        page,
793        DB_COLLECTION_ROOTS_PAGE_OFFSET,
794        header.physical.collection_roots_page,
795    )?;
796    write_u64(
797        page,
798        DB_COLLECTION_ROOTS_CHECKSUM_OFFSET,
799        header.physical.collection_roots_checksum,
800    )?;
801    write_u32(
802        page,
803        DB_COLLECTION_ROOT_COUNT_OFFSET,
804        header.physical.collection_root_count,
805    )?;
806    write_u32(
807        page,
808        DB_SNAPSHOT_COUNT_OFFSET,
809        header.physical.snapshot_count,
810    )?;
811    write_u32(page, DB_INDEX_COUNT_OFFSET, header.physical.index_count)?;
812    write_u32(
813        page,
814        DB_CATALOG_COLLECTION_COUNT_OFFSET,
815        header.physical.catalog_collection_count,
816    )?;
817    write_u64(
818        page,
819        DB_CATALOG_TOTAL_ENTITIES_OFFSET,
820        header.physical.catalog_total_entities,
821    )?;
822    write_u32(page, DB_EXPORT_COUNT_OFFSET, header.physical.export_count)?;
823    write_u32(
824        page,
825        DB_GRAPH_PROJECTION_COUNT_OFFSET,
826        header.physical.graph_projection_count,
827    )?;
828    write_u32(
829        page,
830        DB_ANALYTICS_JOB_COUNT_OFFSET,
831        header.physical.analytics_job_count,
832    )?;
833    write_u32(
834        page,
835        DB_MANIFEST_EVENT_COUNT_OFFSET,
836        header.physical.manifest_event_count,
837    )?;
838    write_u32(page, DB_REGISTRY_PAGE_OFFSET, header.physical.registry_page)?;
839    write_u64(
840        page,
841        DB_REGISTRY_CHECKSUM_OFFSET,
842        header.physical.registry_checksum,
843    )?;
844    write_u32(page, DB_RECOVERY_PAGE_OFFSET, header.physical.recovery_page)?;
845    write_u64(
846        page,
847        DB_RECOVERY_CHECKSUM_OFFSET,
848        header.physical.recovery_checksum,
849    )?;
850    write_u32(page, DB_CATALOG_PAGE_OFFSET, header.physical.catalog_page)?;
851    write_u64(
852        page,
853        DB_CATALOG_CHECKSUM_OFFSET,
854        header.physical.catalog_checksum,
855    )?;
856    write_u32(
857        page,
858        DB_METADATA_STATE_PAGE_OFFSET,
859        header.physical.metadata_state_page,
860    )?;
861    write_u64(
862        page,
863        DB_METADATA_STATE_CHECKSUM_OFFSET,
864        header.physical.metadata_state_checksum,
865    )?;
866    write_u32(
867        page,
868        DB_VECTOR_ARTIFACT_PAGE_OFFSET,
869        header.physical.vector_artifact_page,
870    )?;
871    write_u64(
872        page,
873        DB_VECTOR_ARTIFACT_CHECKSUM_OFFSET,
874        header.physical.vector_artifact_checksum,
875    )?;
876    page[DB_CHECKPOINT_IN_PROGRESS_OFFSET] = if header.checkpoint_in_progress { 1 } else { 0 };
877    write_u64(
878        page,
879        DB_CHECKPOINT_TARGET_LSN_OFFSET,
880        header.checkpoint_target_lsn,
881    )?;
882    write_u64(
883        page,
884        DB_REQUIRED_CAPABILITIES_OFFSET,
885        header.required_capabilities,
886    )
887}
888
889// ── Paged superblock zone (ADR 0038 §2 phase 1) ─────────────────────────
890//
891// The first two sectors of a paged `.rdb` are the superblock zone: a
892// ping-pong pair of fixed-size slots. Each slot holds the page-0 image (page
893// header + database header + encryption header) followed by a trailer that
894// carries the slot's copy index, generation counter and CRC.
895//
896// An update always writes the *older* slot and fsyncs before it becomes the
897// newest valid copy, so a crash mid-write can only ever damage the copy that
898// was already stale. `select_paged_superblock` picks the highest generation
899// whose CRC verifies; both invalid means the store does not open.
900//
901// The zone lives inside page 0's 16 KiB, whose tail was unused. Page 0 is
902// therefore no longer a checksummed `Page` — the slot CRCs are its integrity
903// contract.
904
905/// Bytes per superblock slot. Sector-aligned so one slot write never touches
906/// its sibling.
907pub const PAGED_SUPERBLOCK_SLOT_SIZE: usize = 4096;
908
909/// The ping-pong pair.
910pub const PAGED_SUPERBLOCK_SLOT_COUNT: usize = 2;
911
912/// Total bytes reserved for the superblock zone at the head of the file.
913pub const PAGED_SUPERBLOCK_ZONE_SIZE: usize =
914    PAGED_SUPERBLOCK_SLOT_SIZE * PAGED_SUPERBLOCK_SLOT_COUNT;
915
916const PAGED_SUPERBLOCK_MAGIC: [u8; 8] = *b"RDBPSBK1";
917const PAGED_SUPERBLOCK_SLOT_VERSION: u32 = 1;
918
919/// Offset of the trailer within a slot: magic(8) ‖ version(4) ‖ copy(4) ‖
920/// generation(8) ‖ crc(4) ‖ reserved(4). Everything before it is the page-0
921/// image, which is why the offline migration tool can strip the trailer to
922/// recover the sidecar-era page 0 byte for byte.
923pub const PAGED_SUPERBLOCK_TRAILER_OFFSET: usize = PAGED_SUPERBLOCK_SLOT_SIZE - 32;
924const PAGED_SUPERBLOCK_VERSION_OFFSET: usize = PAGED_SUPERBLOCK_TRAILER_OFFSET + 8;
925const PAGED_SUPERBLOCK_COPY_OFFSET: usize = PAGED_SUPERBLOCK_TRAILER_OFFSET + 12;
926const PAGED_SUPERBLOCK_GENERATION_OFFSET: usize = PAGED_SUPERBLOCK_TRAILER_OFFSET + 16;
927const PAGED_SUPERBLOCK_CRC_OFFSET: usize = PAGED_SUPERBLOCK_TRAILER_OFFSET + 24;
928
929/// The newest valid copy found in a superblock zone.
930#[derive(Debug, Clone, Copy, PartialEq, Eq)]
931pub struct PagedSuperblockSelection {
932    pub copy_index: usize,
933    pub generation: u64,
934}
935
936/// Byte offset of `copy_index`'s slot from the start of the file.
937pub fn paged_superblock_slot_offset(copy_index: usize) -> u64 {
938    (copy_index % PAGED_SUPERBLOCK_SLOT_COUNT) as u64 * PAGED_SUPERBLOCK_SLOT_SIZE as u64
939}
940
941/// The copy a fresh update must target: always the one that is *not* newest,
942/// so the newest durable copy survives a crash during the write.
943pub fn paged_superblock_next_copy(newest: Option<PagedSuperblockSelection>) -> usize {
944    match newest {
945        Some(selection) => (selection.copy_index + 1) % PAGED_SUPERBLOCK_SLOT_COUNT,
946        None => 0,
947    }
948}
949
950/// Stamp `slot`'s trailer (copy index, generation) and seal it with a CRC over
951/// everything that precedes the checksum field.
952pub fn seal_paged_superblock_slot(
953    slot: &mut [u8],
954    copy_index: usize,
955    generation: u64,
956) -> Result<(), DatabaseHeaderError> {
957    ensure_len(slot, PAGED_SUPERBLOCK_SLOT_SIZE)?;
958    slot[PAGED_SUPERBLOCK_TRAILER_OFFSET..PAGED_SUPERBLOCK_TRAILER_OFFSET + 8]
959        .copy_from_slice(&PAGED_SUPERBLOCK_MAGIC);
960    write_u32(
961        slot,
962        PAGED_SUPERBLOCK_VERSION_OFFSET,
963        PAGED_SUPERBLOCK_SLOT_VERSION,
964    )?;
965    write_u32(
966        slot,
967        PAGED_SUPERBLOCK_COPY_OFFSET,
968        u32::try_from(copy_index).unwrap_or(u32::MAX),
969    )?;
970    write_u64(slot, PAGED_SUPERBLOCK_GENERATION_OFFSET, generation)?;
971    let crc = paged_superblock_crc(&slot[..PAGED_SUPERBLOCK_CRC_OFFSET]);
972    write_u32(slot, PAGED_SUPERBLOCK_CRC_OFFSET, crc)?;
973    // The 4 trailing reserved bytes stay zero.
974    write_u32(slot, PAGED_SUPERBLOCK_CRC_OFFSET + 4, 0)
975}
976
977/// Decode `slot`'s generation, or `None` when the slot was never written, is
978/// torn, carries a foreign copy index, or fails its CRC.
979pub fn paged_superblock_slot_generation(slot: &[u8], copy_index: usize) -> Option<u64> {
980    if slot.len() < PAGED_SUPERBLOCK_SLOT_SIZE {
981        return None;
982    }
983    if slot[PAGED_SUPERBLOCK_TRAILER_OFFSET..PAGED_SUPERBLOCK_TRAILER_OFFSET + 8]
984        != PAGED_SUPERBLOCK_MAGIC
985    {
986        return None;
987    }
988    if read_u32(slot, PAGED_SUPERBLOCK_VERSION_OFFSET).ok()? != PAGED_SUPERBLOCK_SLOT_VERSION {
989        return None;
990    }
991    // A slot that names a different copy is a misdirected write, not a
992    // legitimately older generation.
993    if read_u32(slot, PAGED_SUPERBLOCK_COPY_OFFSET).ok()? != u32::try_from(copy_index).ok()? {
994        return None;
995    }
996    let stored = read_u32(slot, PAGED_SUPERBLOCK_CRC_OFFSET).ok()?;
997    if paged_superblock_crc(&slot[..PAGED_SUPERBLOCK_CRC_OFFSET]) != stored {
998        return None;
999    }
1000    read_u64(slot, PAGED_SUPERBLOCK_GENERATION_OFFSET).ok()
1001}
1002
1003/// Pick the newest valid copy in a full superblock zone. `None` means both
1004/// copies are invalid: the caller must refuse to open the store.
1005pub fn select_paged_superblock(zone: &[u8]) -> Option<PagedSuperblockSelection> {
1006    (0..PAGED_SUPERBLOCK_SLOT_COUNT)
1007        .filter_map(|copy_index| {
1008            let start = copy_index * PAGED_SUPERBLOCK_SLOT_SIZE;
1009            let slot = zone.get(start..start + PAGED_SUPERBLOCK_SLOT_SIZE)?;
1010            Some(PagedSuperblockSelection {
1011                copy_index,
1012                generation: paged_superblock_slot_generation(slot, copy_index)?,
1013            })
1014        })
1015        .max_by_key(|selection| selection.generation)
1016}
1017
1018fn paged_superblock_crc(bytes: &[u8]) -> u32 {
1019    let mut hasher = crc32fast::Hasher::new();
1020    hasher.update(bytes);
1021    hasher.finalize()
1022}
1023
1024fn ensure_len(bytes: &[u8], need: usize) -> Result<(), DatabaseHeaderError> {
1025    if bytes.len() < need {
1026        return Err(DatabaseHeaderError::ShortPage {
1027            need,
1028            got: bytes.len(),
1029        });
1030    }
1031    Ok(())
1032}
1033
1034fn read_u32(bytes: &[u8], offset: usize) -> Result<u32, DatabaseHeaderError> {
1035    ensure_len(bytes, offset + 4)?;
1036    Ok(u32::from_le_bytes(
1037        bytes[offset..offset + 4].try_into().expect("len checked"),
1038    ))
1039}
1040
1041fn read_u16(bytes: &[u8], offset: usize) -> Result<u16, DatabaseHeaderError> {
1042    ensure_len(bytes, offset + 2)?;
1043    Ok(u16::from_le_bytes(
1044        bytes[offset..offset + 2].try_into().expect("len checked"),
1045    ))
1046}
1047
1048fn read_u64(bytes: &[u8], offset: usize) -> Result<u64, DatabaseHeaderError> {
1049    ensure_len(bytes, offset + 8)?;
1050    Ok(u64::from_le_bytes(
1051        bytes[offset..offset + 8].try_into().expect("len checked"),
1052    ))
1053}
1054
1055fn write_u32(bytes: &mut [u8], offset: usize, value: u32) -> Result<(), DatabaseHeaderError> {
1056    ensure_len(bytes, offset + 4)?;
1057    bytes[offset..offset + 4].copy_from_slice(&value.to_le_bytes());
1058    Ok(())
1059}
1060
1061fn write_u16(bytes: &mut [u8], offset: usize, value: u16) -> Result<(), DatabaseHeaderError> {
1062    ensure_len(bytes, offset + 2)?;
1063    bytes[offset..offset + 2].copy_from_slice(&value.to_le_bytes());
1064    Ok(())
1065}
1066
1067fn write_u64(bytes: &mut [u8], offset: usize, value: u64) -> Result<(), DatabaseHeaderError> {
1068    ensure_len(bytes, offset + 8)?;
1069    bytes[offset..offset + 8].copy_from_slice(&value.to_le_bytes());
1070    Ok(())
1071}
1072
1073#[cfg(test)]
1074mod tests {
1075    use super::*;
1076
1077    fn sealed_zone(gen_a: Option<u64>, gen_b: Option<u64>) -> Vec<u8> {
1078        let mut zone = vec![0u8; PAGED_SUPERBLOCK_ZONE_SIZE];
1079        for (copy_index, generation) in [gen_a, gen_b].into_iter().enumerate() {
1080            let Some(generation) = generation else {
1081                continue;
1082            };
1083            let start = copy_index * PAGED_SUPERBLOCK_SLOT_SIZE;
1084            let slot = &mut zone[start..start + PAGED_SUPERBLOCK_SLOT_SIZE];
1085            init_database_header_page(slot, 3).expect("init header");
1086            seal_paged_superblock_slot(slot, copy_index, generation).expect("seal slot");
1087        }
1088        zone
1089    }
1090
1091    #[test]
1092    fn superblock_zone_fits_two_sector_aligned_slots_before_page_one() {
1093        assert_eq!(PAGED_SUPERBLOCK_SLOT_SIZE, 4096);
1094        assert_eq!(PAGED_SUPERBLOCK_ZONE_SIZE, 8192);
1095        // Both slots live inside page 0, so page 1 is never touched by a
1096        // superblock update.
1097        assert!(PAGED_SUPERBLOCK_ZONE_SIZE <= PAGED_PAGE_SIZE);
1098        // The database header must not overrun into the slot trailer.
1099        assert!(DB_HEADER_MIN_LEN < PAGED_SUPERBLOCK_TRAILER_OFFSET);
1100        assert_eq!(paged_superblock_slot_offset(0), 0);
1101        assert_eq!(paged_superblock_slot_offset(1), 4096);
1102    }
1103
1104    #[test]
1105    fn sealed_slot_round_trips_generation_and_keeps_the_header_readable() {
1106        let mut slot = vec![0u8; PAGED_SUPERBLOCK_SLOT_SIZE];
1107        init_database_header_page(&mut slot, 7).expect("init header");
1108        seal_paged_superblock_slot(&mut slot, 1, 42).expect("seal slot");
1109
1110        assert_eq!(paged_superblock_slot_generation(&slot, 1), Some(42));
1111        assert_eq!(
1112            database_header_page_count(&slot).expect("database_header_page_count() should succeed"),
1113            7
1114        );
1115    }
1116
1117    #[test]
1118    fn selection_prefers_the_newest_valid_copy_and_survives_one_torn_slot() {
1119        let zone = sealed_zone(Some(4), Some(5));
1120        assert_eq!(
1121            select_paged_superblock(&zone),
1122            Some(PagedSuperblockSelection {
1123                copy_index: 1,
1124                generation: 5
1125            })
1126        );
1127
1128        // Tear the newer copy: open falls back to the older intact one.
1129        let mut torn = zone.clone();
1130        torn[PAGED_SUPERBLOCK_SLOT_SIZE + 64] ^= 0xFF;
1131        assert_eq!(
1132            select_paged_superblock(&torn),
1133            Some(PagedSuperblockSelection {
1134                copy_index: 0,
1135                generation: 4
1136            })
1137        );
1138    }
1139
1140    #[test]
1141    fn a_misdirected_slot_copy_is_rejected_not_read_as_an_older_generation() {
1142        // Slot 0's bytes written at slot 1's offset: right data, wrong place.
1143        let zone = sealed_zone(Some(9), None);
1144        let (slot_zero, _) = zone.split_at(PAGED_SUPERBLOCK_SLOT_SIZE);
1145        assert_eq!(paged_superblock_slot_generation(slot_zero, 1), None);
1146    }
1147
1148    #[test]
1149    fn both_copies_invalid_yields_no_selection() {
1150        assert_eq!(
1151            select_paged_superblock(&vec![0u8; PAGED_SUPERBLOCK_ZONE_SIZE]),
1152            None
1153        );
1154        assert_eq!(select_paged_superblock(&[]), None);
1155
1156        let mut zone = sealed_zone(Some(1), Some(2));
1157        zone[64] ^= 0xFF;
1158        zone[PAGED_SUPERBLOCK_SLOT_SIZE + 64] ^= 0xFF;
1159        assert_eq!(select_paged_superblock(&zone), None);
1160    }
1161
1162    #[test]
1163    fn the_next_copy_to_write_is_always_the_stale_one() {
1164        assert_eq!(paged_superblock_next_copy(None), 0);
1165        assert_eq!(
1166            paged_superblock_next_copy(Some(PagedSuperblockSelection {
1167                copy_index: 0,
1168                generation: 3
1169            })),
1170            1
1171        );
1172        assert_eq!(
1173            paged_superblock_next_copy(Some(PagedSuperblockSelection {
1174                copy_index: 1,
1175                generation: 3
1176            })),
1177            0
1178        );
1179    }
1180
1181    #[test]
1182    fn bit_rot_anywhere_in_the_slot_payload_fails_the_crc() {
1183        let mut slot = vec![0u8; PAGED_SUPERBLOCK_SLOT_SIZE];
1184        init_database_header_page(&mut slot, 3).expect("init header");
1185        seal_paged_superblock_slot(&mut slot, 0, 1).expect("seal slot");
1186
1187        for offset in [
1188            0,
1189            40,
1190            DB_HEADER_MIN_LEN - 1,
1191            PAGED_SUPERBLOCK_SLOT_SIZE - 33,
1192        ] {
1193            let mut rotted = slot.clone();
1194            rotted[offset] ^= 0x01;
1195            assert_eq!(
1196                paged_superblock_slot_generation(&rotted, 0),
1197                None,
1198                "bit rot at offset {offset} must fail the slot CRC"
1199            );
1200        }
1201    }
1202
1203    #[test]
1204    fn paged_page_header_round_trips_raw_layout() {
1205        let header = PagedPageHeader {
1206            page_type: 13,
1207            flags: 0b1010_0001,
1208            cell_count: 42,
1209            free_start: 44,
1210            free_end: 16_000,
1211            page_id: 99,
1212            parent_id: 88,
1213            right_child: 77,
1214            lsn: 66,
1215            checksum: 55,
1216        };
1217
1218        let encoded = encode_paged_page_header(&header);
1219
1220        assert_eq!(encoded.len(), PAGED_PAGE_HEADER_SIZE);
1221        assert_eq!(decode_paged_page_header(&encoded), header);
1222    }
1223
1224    #[test]
1225    fn paged_page_header_accessors_update_expected_offsets() {
1226        let mut page = [0u8; PAGED_PAGE_SIZE];
1227        page[0] = 13;
1228
1229        set_paged_page_cell_count(&mut page, 2);
1230        set_paged_page_free_start(&mut page, 34);
1231        set_paged_page_free_end(&mut page, 4096);
1232        set_paged_page_parent_id(&mut page, 7);
1233        set_paged_page_right_child(&mut page, 8);
1234        set_paged_page_lsn(&mut page, 9);
1235        set_paged_page_checksum(&mut page, 10);
1236
1237        assert_eq!(paged_page_type(&page), 13);
1238        assert_eq!(paged_page_cell_count(&page), 2);
1239        assert_eq!(paged_page_free_start(&page), 34);
1240        assert_eq!(paged_page_free_end(&page), 4096);
1241        assert_eq!(paged_page_parent_id(&page), 7);
1242        assert_eq!(paged_page_right_child(&page), 8);
1243        assert_eq!(paged_page_lsn(&page), 9);
1244        assert_eq!(paged_page_checksum(&page), 10);
1245        clear_paged_page_checksum(&mut page);
1246        assert_eq!(paged_page_checksum(&page), 0);
1247    }
1248
1249    #[test]
1250    fn paged_cell_helpers_round_trip_pointer_and_payload() {
1251        let mut page = [0u8; PAGED_PAGE_SIZE];
1252        let pointer = (PAGED_PAGE_SIZE - 14) as u16;
1253
1254        assert!(write_paged_cell(&mut page, pointer, b"key", b"value"));
1255        assert!(set_paged_cell_pointer(&mut page, 0, pointer));
1256
1257        assert_eq!(paged_cell_pointer(&page, 0), Some(pointer));
1258        let cell = paged_cell_bytes(&page, pointer).expect("cell bytes");
1259        let (key, value) = paged_cell_key_value(cell).expect("key value");
1260
1261        assert_eq!(key, b"key");
1262        assert_eq!(value, b"value");
1263        assert_eq!(paged_cell_total_len(&page, pointer), Some(14));
1264    }
1265
1266    #[test]
1267    fn database_header_field_helpers_preserve_page_zero_offsets() {
1268        let mut page = vec![0u8; PAGED_PAGE_SIZE];
1269
1270        init_database_header_page(&mut page, 7).expect("init database header page");
1271        set_database_header_version(&mut page, PAGE_FILE_VERSION + 1).expect("set version");
1272        set_database_header_page_count(&mut page, 9).expect("set page count");
1273        set_database_header_freelist_head(&mut page, 4).expect("set freelist head");
1274
1275        assert!(database_header_magic_matches(&page));
1276        assert!(matches!(
1277            decode_database_header(&page),
1278            Err(DatabaseHeaderError::UnsupportedDatabaseVersion { .. })
1279        ));
1280        set_database_header_version(&mut page, PAGE_FILE_VERSION).expect("restore version");
1281        assert_eq!(
1282            database_header_page_size(&page).unwrap(),
1283            PAGED_PAGE_SIZE as u32
1284        );
1285        assert_eq!(database_header_page_count(&page).unwrap(), 9);
1286        assert_eq!(database_header_freelist_head(&page).unwrap(), 4);
1287    }
1288
1289    #[test]
1290    fn required_capabilities_round_trip_and_unknown_bits_fail_before_page_reads() {
1291        let mut page = vec![0u8; PAGED_SUPERBLOCK_SLOT_SIZE];
1292        let header = DatabaseHeader {
1293            version: PAGE_REQUIRED_CAPABILITIES_FILE_VERSION,
1294            required_capabilities: FILE_CAPABILITY_TEMPERATURE,
1295            ..DatabaseHeader::default()
1296        };
1297        encode_database_header(&mut page, &header).expect("encode capabilities");
1298        assert_eq!(
1299            decode_database_header(&page)
1300                .expect("supported capabilities")
1301                .required_capabilities,
1302            FILE_CAPABILITY_TEMPERATURE
1303        );
1304
1305        set_database_header_required_capabilities(&mut page, 1 << 63)
1306            .expect("set unknown capability");
1307        assert_eq!(
1308            decode_database_header(&page),
1309            Err(DatabaseHeaderError::UnsupportedRequiredCapabilities(
1310                1 << 63
1311            ))
1312        );
1313    }
1314
1315    #[test]
1316    fn disk_residency_moves_from_legacy_version_to_explicit_capability() {
1317        let legacy = DatabaseHeader {
1318            version: PAGE_DISK_RESIDENCY_FILE_VERSION,
1319            ..DatabaseHeader::default()
1320        };
1321        assert!(legacy.requires_disk_residency());
1322
1323        let resident_result_cache = DatabaseHeader {
1324            version: PAGE_RESULT_CACHE_FILE_VERSION,
1325            ..DatabaseHeader::default()
1326        };
1327        assert!(!resident_result_cache.requires_disk_residency());
1328
1329        let disk_result_cache = DatabaseHeader {
1330            version: PAGE_RESULT_CACHE_DISK_FILE_VERSION,
1331            ..DatabaseHeader::default()
1332        };
1333        assert!(disk_result_cache.requires_disk_residency());
1334
1335        let temperature_only = DatabaseHeader {
1336            version: PAGE_REQUIRED_CAPABILITIES_FILE_VERSION,
1337            required_capabilities: FILE_CAPABILITY_TEMPERATURE,
1338            ..DatabaseHeader::default()
1339        };
1340        assert!(!temperature_only.requires_disk_residency());
1341
1342        let disk_resident = DatabaseHeader {
1343            required_capabilities: FILE_CAPABILITY_DISK_RESIDENCY,
1344            ..temperature_only
1345        };
1346        assert!(disk_resident.requires_disk_residency());
1347    }
1348
1349    #[test]
1350    fn legacy_headers_ignore_the_reserved_capability_bytes() {
1351        let mut page = vec![0u8; PAGED_SUPERBLOCK_SLOT_SIZE];
1352        init_database_header_page(&mut page, 7).expect("init header");
1353        set_database_header_required_capabilities(&mut page, u64::MAX)
1354            .expect("fill reserved bytes");
1355
1356        assert_eq!(
1357            decode_database_header(&page)
1358                .expect("legacy header")
1359                .required_capabilities,
1360            0
1361        );
1362    }
1363
1364    #[test]
1365    fn paged_dwb_frame_round_trips_entries_and_validates_checksum() {
1366        let mut page = [0u8; PAGED_PAGE_SIZE];
1367        page[0] = 7;
1368        let frame = encode_paged_dwb_frame([(42, &page)]);
1369
1370        let entries = decode_paged_dwb_frame(&frame).expect("decode DWB frame");
1371        assert_eq!(entries.len(), 1);
1372        assert_eq!(entries[0].page_id, 42);
1373        assert_eq!(entries[0].page, page);
1374
1375        let mut corrupted = frame;
1376        let last = corrupted.len() - 1;
1377        corrupted[last] ^= 0xFF;
1378        assert!(matches!(
1379            decode_paged_dwb_frame(&corrupted),
1380            Err(PagedDwbFrameError::ChecksumMismatch { .. })
1381        ));
1382    }
1383
1384    #[test]
1385    fn paged_encryption_header_round_trips_marker_and_raw_bytes() {
1386        let header = PagedEncryptionHeader {
1387            salt: [7u8; PAGED_ENCRYPTION_SALT_SIZE],
1388            key_check: vec![9u8; PAGED_ENCRYPTION_KEY_CHECK_BLOB_SIZE],
1389        };
1390        let mut page = vec![0u8; PAGED_PAGE_SIZE];
1391        let bytes = encode_paged_encryption_header(&header);
1392
1393        write_paged_encryption_marker_and_header(&mut page, &bytes)
1394            .expect("write encryption marker");
1395
1396        assert!(paged_encryption_marker_present(&page));
1397        let raw = paged_encryption_header_bytes(&page).expect("header bytes");
1398        assert_eq!(decode_paged_encryption_header(raw).unwrap(), header);
1399    }
1400
1401    #[test]
1402    fn database_header_round_trips_page_zero_contract() {
1403        let header = DatabaseHeader {
1404            version: PAGE_FILE_VERSION,
1405            page_size: PAGED_PAGE_SIZE as u32,
1406            page_count: 42,
1407            freelist_head: 7,
1408            schema_version: 3,
1409            checkpoint_lsn: 99,
1410            checkpoint_in_progress: true,
1411            checkpoint_target_lsn: 123,
1412            required_capabilities: FILE_CAPABILITY_TEMPERATURE,
1413            physical: PhysicalFileHeader {
1414                format_version: 11,
1415                sequence: 12,
1416                manifest_oldest_root: 13,
1417                manifest_root: 14,
1418                free_set_root: 15,
1419                manifest_page: 16,
1420                manifest_checksum: 17,
1421                collection_roots_page: 18,
1422                collection_roots_checksum: 19,
1423                collection_root_count: 20,
1424                snapshot_count: 21,
1425                index_count: 22,
1426                catalog_collection_count: 23,
1427                catalog_total_entities: 24,
1428                export_count: 25,
1429                graph_projection_count: 26,
1430                analytics_job_count: 27,
1431                manifest_event_count: 28,
1432                registry_page: 29,
1433                registry_checksum: 30,
1434                recovery_page: 31,
1435                recovery_checksum: 32,
1436                catalog_page: 33,
1437                catalog_checksum: 34,
1438                metadata_state_page: 35,
1439                metadata_state_checksum: 36,
1440                vector_artifact_page: 37,
1441                vector_artifact_checksum: 38,
1442            },
1443        };
1444        let mut page = vec![0u8; PAGED_PAGE_SIZE];
1445
1446        encode_database_header(&mut page, &header).expect("encode header");
1447
1448        assert!(database_header_magic_matches(&page));
1449        let decoded = decode_database_header(&page).expect("decode header");
1450        assert_eq!(decoded, header);
1451    }
1452
1453    #[test]
1454    fn database_header_rejects_newer_versions() {
1455        let mut page = vec![0u8; PAGED_PAGE_SIZE];
1456        let header = DatabaseHeader {
1457            version: PAGE_FILE_VERSION + 1,
1458            ..DatabaseHeader::default()
1459        };
1460        encode_database_header(&mut page, &header).expect("encode header");
1461
1462        assert!(matches!(
1463            decode_database_header(&page),
1464            Err(DatabaseHeaderError::UnsupportedDatabaseVersion { .. })
1465        ));
1466    }
1467}
reddb_file/file_format.rs

reddb_file/
file_format.rs
