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

reddb_file/
file_format.rs
