Skip to main content

reddb_server/storage/engine/pager/
impl.rs

1use super::*;
2
3pub(super) const BTRFS_SUPER_MAGIC: i64 = 0x9123_683e;
4pub(super) const ZFS_SUPER_MAGIC: i64 = 0x2fc1_2fc1;
5pub(super) const FS_NOCOW_FL: u64 = 0x0080_0000;
6
7/// The superblock zone occupies the head of page 0 (ADR 0038 §2 phase 1).
8pub(super) const SUPERBLOCK_SLOT_SIZE: usize = reddb_file::PAGED_SUPERBLOCK_SLOT_SIZE;
9pub(super) const SUPERBLOCK_ZONE_SIZE: usize = reddb_file::PAGED_SUPERBLOCK_ZONE_SIZE;
10pub(super) const PHASE3_DWB_ZONE_FILE_VERSION: u32 = reddb_file::PAGE_FILE_VERSION;
11
12/// The newest valid superblock copy together with its slot image.
13type NewestSuperblock = (
14    reddb_file::PagedSuperblockSelection,
15    [u8; SUPERBLOCK_SLOT_SIZE],
16);
17
18/// Take the first slot's worth of a freshly-built page-0 image. The database
19/// header and the encryption header both live well inside this prefix.
20fn superblock_slot_image(page: &[u8; PAGE_SIZE]) -> [u8; SUPERBLOCK_SLOT_SIZE] {
21    let mut image = [0u8; SUPERBLOCK_SLOT_SIZE];
22    image.copy_from_slice(&page[..SUPERBLOCK_SLOT_SIZE]);
23    image
24}
25
26#[derive(Debug, Clone, Copy, PartialEq, Eq)]
27pub(super) enum CowFilesystemKind {
28    Zfs,
29    BtrfsDataCow,
30    TestOverride,
31}
32
33pub(super) fn classify_cow_filesystem(
34    fs_type: i64,
35    mount_options: Option<&str>,
36    inode_flags: Option<u64>,
37) -> Option<CowFilesystemKind> {
38    match fs_type {
39        ZFS_SUPER_MAGIC => Some(CowFilesystemKind::Zfs),
40        BTRFS_SUPER_MAGIC => {
41            let mount_options = mount_options?;
42            if mount_options.split(',').any(|option| option == "nodatacow") {
43                return None;
44            }
45
46            let inode_flags = inode_flags?;
47            if inode_flags & FS_NOCOW_FL != 0 {
48                return None;
49            }
50
51            Some(CowFilesystemKind::BtrfsDataCow)
52        }
53        _ => None,
54    }
55}
56
57#[cfg(target_os = "linux")]
58fn linux_fstatfs_type(file: &File) -> Option<i64> {
59    use std::mem::MaybeUninit;
60    use std::os::fd::AsRawFd;
61
62    let mut stat = MaybeUninit::<libc::statfs>::uninit();
63    let rc = unsafe { libc::fstatfs(file.as_raw_fd(), stat.as_mut_ptr()) };
64    if rc != 0 {
65        return None;
66    }
67    let stat = unsafe { stat.assume_init() };
68    // `statfs.f_type` is i32 on 32-bit glibc targets (armv7) and i64 on
69    // 64-bit ones — widen explicitly so both compile.
70    #[allow(clippy::unnecessary_cast)]
71    Some(stat.f_type as i64)
72}
73
74#[cfg(target_os = "linux")]
75fn linux_inode_flags(file: &File) -> Option<u64> {
76    use std::os::fd::AsRawFd;
77
78    let mut flags: libc::c_long = 0;
79    let rc = unsafe { libc::ioctl(file.as_raw_fd(), libc::FS_IOC_GETFLAGS, &mut flags) };
80    if rc != 0 {
81        return None;
82    }
83    Some(flags as u64)
84}
85
86#[cfg(target_os = "linux")]
87fn linux_mount_options_for_path(path: &Path) -> Option<String> {
88    let path = path.canonicalize().ok()?;
89    let mountinfo = std::fs::read_to_string("/proc/self/mountinfo").ok()?;
90    parse_mountinfo_options_for_path(&mountinfo, &path)
91}
92
93#[cfg(target_os = "linux")]
94pub(super) fn parse_mountinfo_options_for_path(mountinfo: &str, path: &Path) -> Option<String> {
95    let mut best: Option<(usize, String)> = None;
96
97    for line in mountinfo.lines() {
98        let fields: Vec<&str> = line.split(' ').collect();
99        if fields.len() < 10 {
100            continue;
101        }
102
103        let Some(separator) = fields.iter().position(|field| *field == "-") else {
104            continue;
105        };
106        if separator + 3 >= fields.len() || separator < 6 {
107            continue;
108        }
109
110        let mount_point = mountinfo_unescape_path(fields[4]);
111        if !path.starts_with(&mount_point) {
112            continue;
113        }
114
115        let fs_type = fields[separator + 1];
116        if fs_type != "btrfs" && fs_type != "zfs" {
117            continue;
118        }
119
120        let mount_options = fields[5];
121        let super_options = fields[separator + 3];
122        let options = format!("{mount_options},{super_options}");
123        let depth = mount_point.components().count();
124        if best
125            .as_ref()
126            .map(|(best_depth, _)| depth > *best_depth)
127            .unwrap_or(true)
128        {
129            best = Some((depth, options));
130        }
131    }
132
133    best.map(|(_, options)| options)
134}
135
136#[cfg(target_os = "linux")]
137fn mountinfo_unescape_path(value: &str) -> PathBuf {
138    let bytes = value.as_bytes();
139    let mut out = Vec::with_capacity(bytes.len());
140    let mut i = 0;
141
142    while i < bytes.len() {
143        if bytes[i] == b'\\' && i + 3 < bytes.len() {
144            let octal = &value[i + 1..i + 4];
145            if let Ok(byte) = u8::from_str_radix(octal, 8) {
146                out.push(byte);
147                i += 4;
148                continue;
149            }
150        }
151        out.push(bytes[i]);
152        i += 1;
153    }
154
155    PathBuf::from(String::from_utf8_lossy(&out).into_owned())
156}
157
158impl Pager {
159    /// Open or create a database file
160    pub fn open<P: AsRef<Path>>(path: P, mut config: PagerConfig) -> Result<Self, PagerError> {
161        let path = path.as_ref().to_path_buf();
162        let exists = path.exists();
163
164        if !exists && !config.create {
165            return Err(PagerError::InvalidDatabase(
166                "Database does not exist".into(),
167            ));
168        }
169
170        // ADR 0038 §4 phase 1 is a clean break: a store still carrying the
171        // retired `rdb-hdr`/`rdb-meta` sidecars is never read by the engine.
172        // The offline migration tool is the only path forward.
173        if exists {
174            if let Some(sidecar) = reddb_file::layout::retired::first_present_phase1_sidecar(&path)
175            {
176                return Err(PagerError::LegacySidecarStore { sidecar });
177            }
178        }
179
180        if !exists && config.read_only {
181            return Err(PagerError::InvalidDatabase(
182                "Cannot create read-only database".into(),
183            ));
184        }
185
186        // Open file
187        // Note: create requires write access, so disable it for read-only mode
188        let file = OpenOptions::new()
189            .read(true)
190            .write(!config.read_only)
191            .create(config.create && !config.read_only)
192            .open(&path)?;
193
194        // gh-892: diagnostic probe of the filesystem block size. If the
195        // compile-time 16 KiB PAGE_SIZE is not a multiple of the FS block
196        // size, page writes straddle FS blocks and incur read-modify-write
197        // amplification. Pure diagnostic — emitted once per open, never
198        // mutates the page size. `blksize()` is the fstat `st_blksize`.
199        #[cfg(unix)]
200        {
201            use std::os::unix::fs::MetadataExt;
202            if let Ok(meta) = file.metadata() {
203                let fs_block_size = meta.blksize();
204                if Self::page_size_misaligned_with_block(PAGE_SIZE, fs_block_size) {
205                    tracing::warn!(
206                        page_size = PAGE_SIZE,
207                        fs_block_size,
208                        path = %path.display(),
209                        "database page size is not a multiple of the filesystem \
210                         block size; page writes will straddle FS blocks \
211                         (read-modify-write amplification). Diagnostic only — \
212                         the page size is unchanged."
213                    );
214                }
215            }
216        }
217
218        // Acquire file lock (exclusive for writes, shared for read-only)
219        let lock_file = if !config.read_only {
220            let lf = OpenOptions::new().read(true).write(true).open(&path)?;
221            lf.try_lock_exclusive().map_err(|_| PagerError::Locked)?;
222            Some(lf)
223        } else {
224            let lf = OpenOptions::new().read(true).open(&path)?;
225            match lf.try_lock_shared() {
226                Ok(_) => Some(lf),
227                Err(_) => None,
228            }
229        };
230
231        // ADR 0038 phase 3: the DWB sidecar is retired. When DWB is active,
232        // the recovery frame is written to a reserved zone inside the data
233        // file and fsynced before in-place page writes. A pre-existing sidecar
234        // marks a legacy store that must go through the offline migration tool.
235        //
236        // gh-895: an explicit `double_write = false` request is honored only
237        // when the already-open data file is proven to live on a filesystem
238        // with atomic CoW page writes. Unknown and non-CoW filesystems fail
239        // closed by keeping the in-file DWB zone enabled.
240        let fold_dwb = crate::physical::fold_dwb_into_wal_enabled();
241        if exists {
242            if let Some(sidecar) =
243                reddb_file::layout::retired::first_present_phase3_dwb_sidecar(&path)
244            {
245                return Err(PagerError::LegacySidecarStore { sidecar });
246            }
247        }
248        if !config.double_write && !config.read_only && !fold_dwb {
249            let skip_dwb_on_cow =
250                Self::cow_filesystem_has_atomic_page_writes(&path, &file).is_some();
251            if !skip_dwb_on_cow {
252                tracing::warn!(
253                    path = %path.display(),
254                    "double_write=false requested, but the data file is not proven to be on \
255                     ZFS or btrfs datacow; keeping the in-file double-write buffer enabled"
256                );
257                config.double_write = true;
258            }
259        }
260        let dwb_enabled = config.double_write && !config.read_only && !fold_dwb;
261
262        let mut pager = Self {
263            path,
264            file: Mutex::new(file),
265            _lock_file: lock_file,
266            dwb_enabled,
267            cache: PageCache::new(config.cache_size),
268            freelist: RwLock::new(FreeList::new()),
269            header: RwLock::new(DatabaseHeader::default()),
270            config,
271            header_dirty: Mutex::new(false),
272            wal: RwLock::new(None),
273            encryption: None,
274        };
275
276        if exists {
277            // Recover from the in-file double-write zone before loading the
278            // header or allowing WAL replay to read pages.
279            pager.ensure_phase3_dwb_zone_layout()?;
280            pager.recover_from_dwb()?;
281            // Load existing database (with header shadow fallback)
282            pager.load_header()?;
283            pager.bind_encryption_for_existing()?;
284        } else {
285            // Initialize new database
286            pager.initialize()?;
287            pager.bind_encryption_for_new()?;
288        }
289
290        Ok(pager)
291    }
292
293    /// gh-892 diagnostic predicate: returns `true` when the database
294    /// `page_size` is **not** an integer multiple of the filesystem's
295    /// reported block size (`st_blksize`), i.e. `page_size % fs_block_size
296    /// != 0`. A `fs_block_size` of `0` (probe unavailable / unknown) is
297    /// treated as aligned so a missing probe never produces a warning.
298    /// Pure function with no I/O so the warn decision is unit-testable.
299    pub(crate) fn page_size_misaligned_with_block(page_size: usize, fs_block_size: u64) -> bool {
300        fs_block_size != 0 && !(page_size as u64).is_multiple_of(fs_block_size)
301    }
302
303    #[cfg(test)]
304    fn cow_filesystem_test_override() -> Option<bool> {
305        match COW_ATOMIC_WRITE_TEST_OVERRIDE.load(Ordering::Relaxed) {
306            1 => Some(true),
307            2 => Some(false),
308            _ => None,
309        }
310    }
311
312    #[cfg(not(test))]
313    fn cow_filesystem_test_override() -> Option<bool> {
314        None
315    }
316
317    fn cow_filesystem_has_atomic_page_writes(
318        path: &Path,
319        file: &File,
320    ) -> Option<CowFilesystemKind> {
321        if let Some(allowed) = Self::cow_filesystem_test_override() {
322            return allowed.then_some(CowFilesystemKind::TestOverride);
323        }
324        Self::probe_cow_filesystem(path, file)
325    }
326
327    #[cfg(target_os = "linux")]
328    fn probe_cow_filesystem(path: &Path, file: &File) -> Option<CowFilesystemKind> {
329        let fs_type = linux_fstatfs_type(file)?;
330        match fs_type {
331            ZFS_SUPER_MAGIC => Some(CowFilesystemKind::Zfs),
332            BTRFS_SUPER_MAGIC => {
333                let mount_options = linux_mount_options_for_path(path)?;
334                let inode_flags = linux_inode_flags(file)?;
335                classify_cow_filesystem(fs_type, Some(&mount_options), Some(inode_flags))
336            }
337            _ => None,
338        }
339    }
340
341    #[cfg(not(target_os = "linux"))]
342    fn probe_cow_filesystem(_path: &Path, _file: &File) -> Option<CowFilesystemKind> {
343        None
344    }
345
346    /// Inspect page 0 for the `RDBE` encryption marker, then resolve
347    /// the (key, marker) matrix:
348    ///
349    /// | Marker | Key supplied | Result                              |
350    /// |--------|--------------|-------------------------------------|
351    /// | yes    | yes          | Bind encryptor; validate key        |
352    /// | yes    | no           | `EncryptionRequired` (fail closed)  |
353    /// | no     | yes          | `PlainDatabaseRefusesKey`           |
354    /// | no     | no           | Plain pager — no binding needed     |
355    fn bind_encryption_for_existing(&mut self) -> Result<(), PagerError> {
356        if self.page_count().unwrap_or(0) == 0 {
357            return self.bind_encryption_for_new();
358        }
359        let (_, image) = self.newest_superblock()?;
360        let data = &image[..];
361        let has_marker = reddb_file::paged_encryption_marker_present(data);
362
363        let key = self.config.encryption.clone();
364        match (has_marker, key) {
365            (true, Some(key)) => {
366                let header_bytes =
367                    reddb_file::paged_encryption_header_bytes(data).ok_or_else(|| {
368                        PagerError::InvalidDatabase("encryption header parse failed".to_string())
369                    })?;
370                let header = crate::storage::encryption::EncryptionHeader::from_bytes(header_bytes)
371                    .map_err(|e| {
372                        PagerError::InvalidDatabase(format!("encryption header parse failed: {e}"))
373                    })?;
374                if !header.validate(&key) {
375                    return Err(PagerError::InvalidKey);
376                }
377                let encryptor = crate::storage::encryption::PageEncryptor::new(key);
378                self.encryption = Some((encryptor, header));
379                Ok(())
380            }
381            (true, None) => Err(PagerError::EncryptionRequired),
382            (false, Some(_)) => Err(PagerError::PlainDatabaseRefusesKey),
383            (false, None) => Ok(()),
384        }
385    }
386
387    /// New DB: if a key is configured, write the marker + header to
388    /// page 0 so subsequent opens detect encryption.
389    fn bind_encryption_for_new(&mut self) -> Result<(), PagerError> {
390        let Some(key) = self.config.encryption.clone() else {
391            return Ok(());
392        };
393        let header = crate::storage::encryption::EncryptionHeader::new(&key);
394        let encryptor = crate::storage::encryption::PageEncryptor::new(key);
395
396        // Stamp the marker + header into the superblock zone if `initialize()`
397        // has already published it.
398        if self.page_count().unwrap_or(0) > 0 {
399            let header_bytes = header.to_bytes();
400            self.commit_superblock(|image| {
401                reddb_file::write_paged_encryption_marker_and_header(image, &header_bytes)
402                    .map_err(|err| PagerError::InvalidDatabase(err.to_string()))
403            })?;
404        }
405        self.encryption = Some((encryptor, header));
406        Ok(())
407    }
408
409    /// Open with default configuration
410    pub fn open_default<P: AsRef<Path>>(path: P) -> Result<Self, PagerError> {
411        Self::open(path, PagerConfig::default())
412    }
413
414    /// Initialize a new database
415    fn initialize(&self) -> Result<(), PagerError> {
416        if self.config.read_only {
417            return Err(PagerError::ReadOnly);
418        }
419
420        // Create header page. Page ids 1 and 2 are reserved for fixed
421        // metadata/vault pages. ADR 0038 phase 3 reserves the following DWB
422        // zone in-file so normal B-tree allocation starts after it.
423        let initial_page_count = FIRST_ALLOCATABLE_PAGE;
424        let header_page = Page::new_header_page(initial_page_count);
425        self.header_write()?.page_count = initial_page_count;
426
427        // Reserve page 0 (the superblock zone) plus the reserved pages so any
428        // scan over 0..page_count can read every allocated page in a brand-new
429        // database. The zone's own bytes are published below, one slot at a
430        // time; this first whole-page write only sizes the file.
431        self.write_page_raw(0, &header_page)?;
432        let mut metadata_page = Page::new(PageType::Header, 1);
433        metadata_page.update_checksum();
434        self.write_page_raw(1, &metadata_page)?;
435        let mut vault_page = Page::new(PageType::Vault, 2);
436        vault_page.update_checksum();
437        self.write_page_raw(2, &vault_page)?;
438        self.clear_dwb_zone()?;
439
440        // Publish both superblock copies so the ping-pong invariant — one
441        // valid copy always survives a crash — holds from the very first
442        // update onward, not just from the second.
443        let mut slot = superblock_slot_image(header_page.as_bytes());
444        self.write_superblock_slot(&mut slot, 0, 1)?;
445        self.write_superblock_slot(&mut slot, 1, 2)?;
446
447        // Sync to disk
448        self.sync()?;
449
450        Ok(())
451    }
452
453    /// Acquire a write lock on the header RwLock, mapping poison errors.
454    fn header_write(&self) -> Result<std::sync::RwLockWriteGuard<'_, DatabaseHeader>, PagerError> {
455        self.header.write().map_err(|_| PagerError::LockPoisoned)
456    }
457
458    /// Acquire a read lock on the header RwLock, mapping poison errors.
459    fn header_read(&self) -> Result<std::sync::RwLockReadGuard<'_, DatabaseHeader>, PagerError> {
460        self.header.read().map_err(|_| PagerError::LockPoisoned)
461    }
462
463    /// Acquire a write lock on the freelist RwLock, mapping poison errors.
464    fn freelist_write(&self) -> Result<std::sync::RwLockWriteGuard<'_, FreeList>, PagerError> {
465        self.freelist.write().map_err(|_| PagerError::LockPoisoned)
466    }
467
468    /// Acquire a lock on the file Mutex, mapping poison errors.
469    fn file_lock(&self) -> Result<std::sync::MutexGuard<'_, File>, PagerError> {
470        self.file.lock().map_err(|_| PagerError::LockPoisoned)
471    }
472
473    /// Acquire a lock on the header_dirty Mutex, mapping poison errors.
474    fn header_dirty_lock(&self) -> Result<std::sync::MutexGuard<'_, bool>, PagerError> {
475        self.header_dirty
476            .lock()
477            .map_err(|_| PagerError::LockPoisoned)
478    }
479
480    /// Read the whole superblock zone (both ping-pong copies) from the head of
481    /// the file. A file shorter than the zone yields zero-filled tail bytes,
482    /// which decode as "slot never written".
483    fn read_superblock_zone(&self) -> Result<[u8; SUPERBLOCK_ZONE_SIZE], PagerError> {
484        let mut file = self.file_lock()?;
485        let len = file.metadata().map(|meta| meta.len()).unwrap_or(0);
486        let mut zone = [0u8; SUPERBLOCK_ZONE_SIZE];
487        let readable = len.min(SUPERBLOCK_ZONE_SIZE as u64) as usize;
488        if readable > 0 {
489            file.seek(SeekFrom::Start(0))?;
490            file.read_exact(&mut zone[..readable])?;
491        }
492        Ok(zone)
493    }
494
495    /// The newest valid superblock copy, with its slot image.
496    ///
497    /// Both copies invalid is the one unrecoverable superblock state (ADR 0074
498    /// §2): the store does not open, and the error names the zone and points
499    /// at salvage rather than panicking or silently re-initialising.
500    fn newest_superblock(&self) -> Result<NewestSuperblock, PagerError> {
501        let zone = self.read_superblock_zone()?;
502        let selection = reddb_file::select_paged_superblock(&zone)
503            .ok_or_else(|| PagerError::SuperblockZoneUnrecoverable(self.path.clone()))?;
504
505        let start = selection.copy_index * SUPERBLOCK_SLOT_SIZE;
506        let mut image = [0u8; SUPERBLOCK_SLOT_SIZE];
507        image.copy_from_slice(&zone[start..start + SUPERBLOCK_SLOT_SIZE]);
508        Ok((selection, image))
509    }
510
511    fn ensure_phase3_dwb_zone_layout(&self) -> Result<(), PagerError> {
512        let (_, image) = self.newest_superblock()?;
513        let header = reddb_file::decode_database_header(&image)
514            .map_err(|err| PagerError::InvalidDatabase(err.to_string()))?;
515        if header.version < PHASE3_DWB_ZONE_FILE_VERSION {
516            return Err(PagerError::LegacyPagedStore {
517                path: self.path.clone(),
518            });
519        }
520        Ok(())
521    }
522
523    /// Seal `slot` for `copy_index`/`generation` and write exactly that slot,
524    /// then fsync. The caller guarantees `copy_index` is the stale copy.
525    fn write_superblock_slot(
526        &self,
527        slot: &mut [u8; SUPERBLOCK_SLOT_SIZE],
528        copy_index: usize,
529        generation: u64,
530    ) -> Result<(), PagerError> {
531        if self.config.read_only {
532            return Err(PagerError::ReadOnly);
533        }
534        reddb_file::seal_paged_superblock_slot(slot, copy_index, generation)
535            .map_err(|err| PagerError::InvalidDatabase(err.to_string()))?;
536
537        let mut file = self.file_lock()?;
538        file.seek(SeekFrom::Start(reddb_file::paged_superblock_slot_offset(
539            copy_index,
540        )))?;
541        file.write_all(slot)?;
542        file.sync_all()?;
543        Ok(())
544    }
545
546    /// Publish `mutate`'s edit of the newest slot image into the *stale* copy.
547    ///
548    /// Reading the newest image first means every byte the pager does not
549    /// explicitly rewrite — most importantly the encryption marker and header —
550    /// carries forward untouched.
551    fn commit_superblock(
552        &self,
553        mutate: impl FnOnce(&mut [u8; SUPERBLOCK_SLOT_SIZE]) -> Result<(), PagerError>,
554    ) -> Result<(), PagerError> {
555        if self.config.read_only {
556            return Err(PagerError::ReadOnly);
557        }
558        let (newest, mut image) = self.newest_superblock()?;
559        mutate(&mut image)?;
560        let next_copy = reddb_file::paged_superblock_next_copy(Some(newest));
561        self.write_superblock_slot(&mut image, next_copy, newest.generation.saturating_add(1))
562    }
563
564    /// Load the database header from the newest valid superblock copy.
565    fn load_header(&self) -> Result<(), PagerError> {
566        let (_, image) = self.newest_superblock()?;
567
568        let mut decoded_header = reddb_file::decode_database_header(&image)
569            .map_err(|err| PagerError::InvalidDatabase(err.to_string()))?;
570        // Cache eviction (or DWB recovery) can persist pages before the next
571        // superblock publication. After a crash the allocation counter may
572        // therefore lag behind live pages already in the file. Reserve the
573        // entire physical extent before allocating; never overwrite that tail.
574        // A partial last page is reserved too, without treating it as verified.
575        let physical_pages = self
576            .file_lock()?
577            .metadata()?
578            .len()
579            .div_ceil(PAGE_SIZE as u64);
580        let physical_pages = u32::try_from(physical_pages).map_err(|_| {
581            PagerError::InvalidDatabase("data file exceeds the page address space".into())
582        })?;
583        if decoded_header.page_count < physical_pages {
584            decoded_header.page_count = physical_pages;
585            if !self.config.read_only {
586                *self.header_dirty_lock()? = true;
587            }
588        }
589        let freelist_head = decoded_header.freelist_head;
590
591        {
592            let mut header = self.header_write()?;
593            *header = decoded_header;
594        }
595
596        // Initialize freelist
597        {
598            let mut freelist = self.freelist_write()?;
599            *freelist = FreeList::from_header(freelist_head, 0);
600        }
601
602        Ok(())
603    }
604
605    /// Publish the in-memory database header as a new superblock generation.
606    fn write_header(&self) -> Result<(), PagerError> {
607        if self.config.read_only {
608            return Err(PagerError::ReadOnly);
609        }
610
611        let header = self.header_read()?.clone();
612        self.commit_superblock(|image| {
613            reddb_file::encode_database_header(image, &header)
614                .map_err(|err| PagerError::InvalidDatabase(err.to_string()))
615        })?;
616        *self.header_dirty_lock()? = false;
617
618        Ok(())
619    }
620
621    /// Read a page from disk (bypassing cache)
622    fn read_page_raw(&self, page_id: u32) -> Result<Page, PagerError> {
623        let mut file = self.file_lock()?;
624        let offset = (page_id as u64) * (PAGE_SIZE as u64);
625
626        file.seek(SeekFrom::Start(offset))?;
627
628        let mut buf = [0u8; PAGE_SIZE];
629        file.read_exact(&mut buf)?;
630
631        let page = Page::from_bytes(buf);
632
633        // Page 0 is the superblock zone, not a checksummed page: its two slots
634        // are sealed independently, so no whole-page checksum can hold across
635        // a ping-pong write. Slot CRCs are its integrity contract.
636        if self.config.verify_checksums && page_id != 0 {
637            page.verify_checksum()?;
638        }
639
640        Ok(page)
641    }
642
643    /// Write a page to disk (bypassing cache)
644    fn write_page_raw(&self, page_id: u32, page: &Page) -> Result<(), PagerError> {
645        if self.config.read_only {
646            return Err(PagerError::ReadOnly);
647        }
648
649        let mut file = self.file_lock()?;
650        let offset = (page_id as u64) * (PAGE_SIZE as u64);
651
652        file.seek(SeekFrom::Start(offset))?;
653        file.write_all(page.as_bytes())?;
654
655        Ok(())
656    }
657
658    /// Read a page (cache-aware)
659    pub fn read_page(&self, page_id: u32) -> Result<Page, PagerError> {
660        // Check cache first
661        if let Some(page) = self.cache.get(page_id) {
662            return Ok(page);
663        }
664
665        // Cache miss - read from disk
666        let page = self.read_page_raw(page_id)?;
667
668        // Add to cache
669        if let Some(dirty_page) = self.cache.insert(page_id, page.clone()) {
670            // Evicted page was dirty, need to write it back
671            let evicted_id = dirty_page.page_id();
672            self.write_page_raw(evicted_id, &dirty_page)?;
673        }
674
675        Ok(page)
676    }
677
678    /// Read a page without verifying checksum (for encrypted pages)
679    ///
680    /// Use this when the page content has its own integrity protection
681    /// (e.g., AES-GCM authentication tag for encrypted pages).
682    pub fn read_page_no_checksum(&self, page_id: u32) -> Result<Page, PagerError> {
683        // Check cache first
684        if let Some(page) = self.cache.get(page_id) {
685            return Ok(page);
686        }
687
688        // Cache miss - read from disk (skip checksum verification)
689        let mut file = self.file_lock()?;
690        let offset = (page_id as u64) * (PAGE_SIZE as u64);
691
692        file.seek(SeekFrom::Start(offset))?;
693
694        let mut buf = [0u8; PAGE_SIZE];
695        file.read_exact(&mut buf)?;
696        drop(file);
697
698        let page = Page::from_bytes(buf);
699
700        // Add to cache (no checksum verification)
701        if let Some(dirty_page) = self.cache.insert(page_id, page.clone()) {
702            // Evicted page was dirty, need to write it back
703            let evicted_id = dirty_page.page_id();
704            self.write_page_raw(evicted_id, &dirty_page)?;
705        }
706
707        Ok(page)
708    }
709
710    /// Write a page (cache-aware)
711    pub fn write_page(&self, page_id: u32, mut page: Page) -> Result<(), PagerError> {
712        if self.config.read_only {
713            return Err(PagerError::ReadOnly);
714        }
715
716        // Update checksum
717        page.update_checksum();
718
719        // Add to cache and mark dirty. If the cache had to evict a
720        // dirty entry, write it through immediately — the evicted
721        // page will never be flushed otherwise (same bug fixed in
722        // `allocate_page`).
723        if let Some(dirty_page) = self.cache.insert(page_id, page) {
724            let evicted_id = dirty_page.page_id();
725            self.write_page_raw(evicted_id, &dirty_page)?;
726        }
727        self.cache.mark_dirty(page_id);
728
729        Ok(())
730    }
731
732    /// Read a page through the configured encryptor if any. Page 0
733    /// is always returned plaintext (it carries the encryption marker
734    /// + header). Callers that want raw cipher bytes can use
735    ///   `read_page_no_checksum` directly.
736    pub fn read_page_decrypted(&self, page_id: u32) -> Result<Page, PagerError> {
737        if page_id == 0 || self.encryption.is_none() {
738            return self.read_page(page_id);
739        }
740        let raw = self.read_page_no_checksum(page_id)?;
741        let (enc, _) = self
742            .encryption
743            .as_ref()
744            .expect("encryption presence checked above");
745        let plaintext = enc
746            .decrypt(page_id, raw.as_bytes())
747            .map_err(|e| PagerError::InvalidDatabase(format!("decrypt page {page_id}: {e}")))?;
748        let mut buf = [0u8; PAGE_SIZE];
749        let n = plaintext.len().min(PAGE_SIZE);
750        buf[..n].copy_from_slice(&plaintext[..n]);
751        Ok(Page::from_bytes(buf))
752    }
753
754    /// Write a page through the configured encryptor if any. Page 0
755    /// bypasses encryption and goes through the normal checksummed
756    /// path. Encrypted pages skip the checksum update because
757    /// AES-GCM's authentication tag is the integrity guarantee.
758    pub fn write_page_encrypted(&self, page_id: u32, page: Page) -> Result<(), PagerError> {
759        if page_id == 0 || self.encryption.is_none() {
760            return self.write_page(page_id, page);
761        }
762        // Canonical per-page envelope overhead (nonce + GCM tag),
763        // owned by reddb-io-crypto (#1053, ADR 0054).
764        const OVERHEAD: usize = crate::storage::encryption::page_encryptor::OVERHEAD;
765        let plaintext_len = PAGE_SIZE - OVERHEAD;
766        let plaintext = &page.as_bytes()[..plaintext_len];
767        let (enc, _) = self
768            .encryption
769            .as_ref()
770            .expect("encryption presence checked above");
771        let ciphertext = enc.encrypt(page_id, plaintext);
772        debug_assert_eq!(ciphertext.len(), PAGE_SIZE);
773        let mut buf = [0u8; PAGE_SIZE];
774        buf.copy_from_slice(&ciphertext);
775        let cipher_page = Page::from_bytes(buf);
776        self.write_page_no_checksum(page_id, cipher_page)
777    }
778
779    /// Write a page without updating checksum (for encrypted pages)
780    ///
781    /// Use this when the page content has its own integrity protection
782    /// (e.g., AES-GCM authentication tag for encrypted pages).
783    pub fn write_page_no_checksum(&self, page_id: u32, page: Page) -> Result<(), PagerError> {
784        if self.config.read_only {
785            return Err(PagerError::ReadOnly);
786        }
787
788        // Add to cache and mark dirty (no checksum update). Same
789        // eviction-write-through guard as `write_page`.
790        if let Some(dirty_page) = self.cache.insert(page_id, page) {
791            let evicted_id = dirty_page.page_id();
792            self.write_page_raw(evicted_id, &dirty_page)?;
793        }
794        self.cache.mark_dirty(page_id);
795
796        Ok(())
797    }
798
799    /// Allocate a new page
800    pub fn allocate_page(&self, page_type: PageType) -> Result<Page, PagerError> {
801        if self.config.read_only {
802            return Err(PagerError::ReadOnly);
803        }
804
805        // Try to get from freelist first
806        let page_id = {
807            let mut freelist = self.freelist_write()?;
808            if let Some(id) = freelist.allocate() {
809                id
810            } else if freelist.trunk_head() != 0 {
811                let trunk_id = freelist.trunk_head();
812                drop(freelist);
813
814                let trunk = self.read_page(trunk_id).map_err(|e| match e {
815                    PagerError::PageNotFound(_) => {
816                        PagerError::InvalidDatabase("Freelist trunk missing".to_string())
817                    }
818                    other => other,
819                })?;
820
821                let mut freelist = self.freelist_write()?;
822                freelist
823                    .load_from_trunk(&trunk)
824                    .map_err(|e| PagerError::InvalidDatabase(format!("Freelist: {}", e)))?;
825                let id = freelist.allocate().ok_or_else(|| {
826                    PagerError::InvalidDatabase("Freelist empty after trunk load".to_string())
827                })?;
828
829                let mut header = self.header_write()?;
830                header.freelist_head = freelist.trunk_head();
831                *self.header_dirty_lock()? = true;
832
833                id
834            } else {
835                // No free pages, extend file
836                let mut header = self.header_write()?;
837                let id = header.page_count;
838                header.page_count += 1;
839                *self.header_dirty_lock()? = true;
840                id
841            }
842        };
843
844        let page = Page::new(page_type, page_id);
845
846        // Write to cache. The evicted page (if any) is dirty by
847        // definition — `cache.insert` only returns `Some` when it
848        // had to evict a dirty entry to make room. The previous
849        // version dropped that return value, which silently lost
850        // writes whenever a freshly-allocated page caused an LRU
851        // eviction. This shows up under heavy ingest as
852        // "B-tree insert error: Pager error: I/O error: failed to
853        // fill whole buffer" later, when something tries to read
854        // back the never-flushed page. Mirror `read_page`'s
855        // handling: write the evicted page through immediately so
856        // the on-disk image stays consistent.
857        if let Some(dirty_page) = self.cache.insert(page_id, page.clone()) {
858            let evicted_id = dirty_page.page_id();
859            self.write_page_raw(evicted_id, &dirty_page)?;
860        }
861        self.cache.mark_dirty(page_id);
862
863        Ok(page)
864    }
865
866    /// Reserve a contiguous extent of vector pages.
867    pub fn reserve_contig_extent(&self, n_pages: u32) -> Result<super::ExtentId, PagerError> {
868        if self.config.read_only {
869            return Err(PagerError::ReadOnly);
870        }
871        if n_pages == 0 {
872            return Err(PagerError::InvalidDatabase(
873                "contiguous extent must reserve at least one page".to_string(),
874            ));
875        }
876
877        let start_page = {
878            let mut header = self.header_write()?;
879            let start = header.page_count;
880            header.page_count = header.page_count.checked_add(n_pages).ok_or_else(|| {
881                PagerError::InvalidDatabase("contiguous extent page count overflow".to_string())
882            })?;
883            *self.header_dirty_lock()? = true;
884            start
885        };
886
887        for page_id in start_page..start_page + n_pages {
888            let mut page = Page::new(PageType::Vector, page_id);
889            page.update_checksum();
890            if let Some(dirty_page) = self.cache.insert(page_id, page) {
891                let evicted_id = dirty_page.page_id();
892                self.write_page_raw(evicted_id, &dirty_page)?;
893            }
894            self.cache.mark_dirty(page_id);
895        }
896
897        Ok(super::ExtentId {
898            start_page,
899            n_pages,
900        })
901    }
902
903    /// Free a page (return to freelist)
904    pub fn free_page(&self, page_id: u32) -> Result<(), PagerError> {
905        if self.config.read_only {
906            return Err(PagerError::ReadOnly);
907        }
908
909        // Remove from cache
910        self.cache.remove(page_id);
911
912        // Add to freelist
913        let mut freelist = self.freelist_write()?;
914        freelist.free(page_id);
915
916        *self.header_dirty_lock()? = true;
917
918        Ok(())
919    }
920
921    /// Get database header
922    pub fn header(&self) -> Result<DatabaseHeader, PagerError> {
923        Ok(self.header_read()?.clone())
924    }
925
926    pub fn physical_header(&self) -> Result<PhysicalFileHeader, PagerError> {
927        Ok(self.header_read()?.physical)
928    }
929
930    pub fn update_physical_header(&self, physical: PhysicalFileHeader) -> Result<(), PagerError> {
931        if self.config.read_only {
932            return Err(PagerError::ReadOnly);
933        }
934
935        let mut header = self.header_write()?;
936        header.physical = physical;
937        *self.header_dirty_lock()? = true;
938        Ok(())
939    }
940
941    /// Get page count
942    pub fn page_count(&self) -> Result<u32, PagerError> {
943        Ok(self.header_read()?.page_count)
944    }
945
946    /// Attach a WAL writer to enforce WAL-first flush ordering.
947    ///
948    /// After this call, [`Pager::flush`] computes the maximum
949    /// `header.lsn` over all dirty pages and calls
950    /// `WalWriter::flush_until(max_lsn)` before any page is written
951    /// to the data file. This is the postgres rule: a page on disk
952    /// implies its WAL record is already durable on disk.
953    ///
954    /// Existing call sites that construct a Pager without a WAL
955    /// keep their previous behaviour (no LSN check) — wiring is
956    /// strictly opt-in.
957    pub fn set_wal_writer(&self, wal: Arc<Mutex<crate::storage::wal::writer::WalWriter>>) {
958        let mut slot = self.wal.write().unwrap_or_else(|p| p.into_inner());
959        *slot = Some(wal);
960    }
961
962    /// Detach the WAL writer (test / shutdown path).
963    pub fn clear_wal_writer(&self) {
964        let mut slot = self.wal.write().unwrap_or_else(|p| p.into_inner());
965        *slot = None;
966    }
967
968    /// Has a WAL writer been attached?
969    pub fn has_wal_writer(&self) -> bool {
970        self.wal.read().map(|s| s.is_some()).unwrap_or(false)
971    }
972
973    /// Flush all dirty pages to disk
974    pub fn flush(&self) -> Result<(), PagerError> {
975        if self.config.read_only {
976            return Ok(());
977        }
978
979        // Persist freelist to trunk pages when dirty
980        let trunks = {
981            let mut freelist = self.freelist_write()?;
982            if freelist.is_dirty() {
983                let mut header = self.header_write()?;
984                let trunks = freelist.flush_to_trunks(0, || {
985                    let id = header.page_count;
986                    header.page_count += 1;
987                    id
988                });
989                header.freelist_head = freelist.trunk_head();
990                *self.header_dirty_lock()? = true;
991                freelist.mark_clean();
992                trunks
993            } else {
994                Vec::new()
995            }
996        };
997
998        let mut dirty_pages = Vec::new();
999        for trunk in trunks {
1000            let page_id = trunk.page_id();
1001            if let Some(evicted) = self.cache.insert(page_id, trunk) {
1002                // Trunk publication can evict dirty data. Keep it in the same
1003                // WAL-first, double-write-protected batch as resident pages.
1004                dirty_pages.push((evicted.page_id(), evicted));
1005            }
1006            self.cache.mark_dirty(page_id);
1007        }
1008
1009        // Flush dirty pages from cache (through DWB if enabled)
1010        dirty_pages.extend(self.cache.flush_dirty());
1011        if !dirty_pages.is_empty() {
1012            // WAL-FIRST: ensure every WAL record describing a dirty
1013            // page is durable BEFORE the page itself reaches disk.
1014            // Pages with `lsn == 0` are exempt (freelist trunks, header
1015            // shadow pages, anything not produced by a WAL append).
1016            let max_lsn = dirty_pages
1017                .iter()
1018                .filter_map(|(_, page)| page.header().ok().map(|h| h.lsn))
1019                .max()
1020                .unwrap_or(0);
1021            if max_lsn > 0 {
1022                if let Ok(slot) = self.wal.read() {
1023                    if let Some(wal) = slot.as_ref() {
1024                        let wal = Arc::clone(wal);
1025                        // Drop the read lock before taking the WAL
1026                        // mutex so an unrelated reader cannot block
1027                        // the flush path.
1028                        drop(slot);
1029                        let mut wal_guard = wal.lock().unwrap_or_else(|p| p.into_inner());
1030                        wal_guard.flush_until(max_lsn).map_err(PagerError::Io)?;
1031                    }
1032                }
1033            }
1034            self.write_pages_through_dwb(&dirty_pages)?;
1035        }
1036
1037        // Write header if dirty
1038        if *self.header_dirty_lock()? {
1039            self.write_header()?;
1040        }
1041
1042        Ok(())
1043    }
1044
1045    /// Sync file to disk (fsync)
1046    pub fn sync(&self) -> Result<(), PagerError> {
1047        self.flush()?;
1048
1049        let file = self.file_lock()?;
1050        file.sync_all()?;
1051
1052        Ok(())
1053    }
1054
1055    /// Get cache statistics
1056    pub fn cache_stats(&self) -> crate::storage::engine::page_cache::CacheStats {
1057        self.cache.stats()
1058    }
1059
1060    /// Pages currently resident in the page cache. Slots are fixed size, so
1061    /// this times the page size is the cache's memory footprint (ADR 0073 §2).
1062    pub fn cache_len(&self) -> usize {
1063        self.cache.len()
1064    }
1065
1066    /// Preallocated page-cache slots — the sizing this process derived from
1067    /// its page-cache budget share.
1068    pub fn cache_capacity(&self) -> usize {
1069        self.cache.capacity()
1070    }
1071
1072    /// Count dirty pages currently in the page cache.
1073    pub fn dirty_page_count(&self) -> usize {
1074        self.cache.dirty_count()
1075    }
1076
1077    /// Estimated fraction of the page cache holding dirty pages.
1078    /// Returned in `[0, 1]`. Used by the background writer to
1079    /// decide when to kick in aggressive flushing.
1080    pub fn dirty_fraction(&self) -> f64 {
1081        let capacity = self.cache.capacity().max(1) as f64;
1082        self.cache.dirty_count() as f64 / capacity
1083    }
1084
1085    /// Flush up to `max` dirty pages from the cache. Returns the
1086    /// number actually written. Background-writer entry point —
1087    /// reuses the same persistence path as `flush()` but bounded.
1088    pub fn flush_some_dirty(&self, max: usize) -> Result<usize, PagerError> {
1089        if self.config.read_only || max == 0 {
1090            return Ok(0);
1091        }
1092        let dirty_pages = self.cache.flush_some_dirty(max);
1093        if dirty_pages.is_empty() {
1094            return Ok(0);
1095        }
1096        let count = dirty_pages.len();
1097        // WAL-first: every cached dirty page carries an LSN that the
1098        // WAL must have already persisted. The full `flush()` path
1099        // enforces this with `wal.flush(max_lsn)`; here we simply
1100        // write through the pager — safe because callers only reach
1101        // this path via the bgwriter, which runs asynchronously
1102        // alongside normal commits that already respect WAL-first.
1103        for (page_id, page) in dirty_pages {
1104            self.write_page(page_id, page)?;
1105        }
1106        Ok(count)
1107    }
1108
1109    /// Get database file path
1110    pub fn path(&self) -> &Path {
1111        &self.path
1112    }
1113
1114    /// Check if database is read-only
1115    pub fn is_read_only(&self) -> bool {
1116        self.config.read_only
1117    }
1118
1119    /// Get file size in bytes
1120    pub fn file_size(&self) -> Result<u64, PagerError> {
1121        let file = self.file_lock()?;
1122        Ok(file.metadata()?.len())
1123    }
1124
1125    /// Issue an OS-level read-ahead hint for `page_id`.
1126    ///
1127    /// A6 prefetch wire: called from `BTreeCursor::next` when the
1128    /// cursor passes 50% of the current leaf, so the kernel fetches
1129    /// the next leaf page while CPU processes the remaining half of
1130    /// the current one. Failures are silent — a missed prefetch is a
1131    /// performance miss, never a correctness bug.
1132    pub fn prefetch_hint(&self, page_id: u32) {
1133        if let Ok(file) = self.file_lock() {
1134            let _ = super::super::prefetch::prefetch_page(&file, page_id as u64, PAGE_SIZE as u32);
1135        }
1136    }
1137
1138    // ── Corruption defense helpers ──────────────────────────────────
1139
1140    /// Path for the double-write buffer file
1141    fn dwb_path(db_path: &Path) -> PathBuf {
1142        reddb_file::layout::pager_dwb_shadow_path(db_path)
1143    }
1144
1145    fn dwb_zone_offset() -> u64 {
1146        DWB_ZONE_START_PAGE as u64 * PAGE_SIZE as u64
1147    }
1148
1149    fn write_dwb_zone_frame(&self, buf: &[u8]) -> Result<(), PagerError> {
1150        if buf.len() > DWB_ZONE_BYTES {
1151            return Err(PagerError::InvalidDatabase(
1152                "DWB frame exceeds the reserved in-file zone".to_string(),
1153            ));
1154        }
1155        let mut file = self.file_lock()?;
1156        file.seek(SeekFrom::Start(Self::dwb_zone_offset()))?;
1157        file.write_all(buf)?;
1158        file.sync_all()?;
1159        Ok(())
1160    }
1161
1162    /// Clear the in-file DWB zone with valid reserved pages.
1163    fn clear_dwb_zone(&self) -> Result<(), PagerError> {
1164        let mut file = self.file_lock()?;
1165        for page_id in DWB_ZONE_START_PAGE..FIRST_ALLOCATABLE_PAGE {
1166            let mut page = Page::new(PageType::Header, page_id);
1167            page.update_checksum();
1168            file.seek(SeekFrom::Start(page_id as u64 * PAGE_SIZE as u64))?;
1169            file.write_all(page.as_bytes())?;
1170        }
1171        file.sync_all()?;
1172        Ok(())
1173    }
1174
1175    fn dwb_zone_is_clear(buf: &[u8]) -> bool {
1176        if buf.len() != DWB_ZONE_BYTES {
1177            return false;
1178        }
1179        for (idx, chunk) in buf.chunks_exact(PAGE_SIZE).enumerate() {
1180            let Ok(page) = Page::from_slice(chunk) else {
1181                return false;
1182            };
1183            if page.verify_checksum().is_err() {
1184                return false;
1185            }
1186            if page.page_id() != DWB_ZONE_START_PAGE + idx as u32 {
1187                return false;
1188            }
1189            if page.page_type().ok() != Some(PageType::Header) {
1190                return false;
1191            }
1192        }
1193        true
1194    }
1195
1196    /// Read the internal manifest page (page 1), the zone the superblock roots.
1197    ///
1198    /// ADR 0074 §2: manifest corruption fails the open didactically, naming the
1199    /// zone — it never falls back to a sidecar and never returns garbage.
1200    pub fn read_manifest_page(&self) -> Result<Page, PagerError> {
1201        self.read_page(1)
1202            .map_err(|_| PagerError::ManifestZoneCorrupt(self.path.clone()))
1203    }
1204
1205    /// Write pages through the double-write buffer for torn page protection.
1206    ///
1207    /// 1. Write pages to the in-file DWB zone with a header (magic + count + checksum)
1208    /// 2. fsync the DWB zone
1209    /// 3. Write all pages to their final locations in the .rdb file
1210    /// 4. Clear the DWB zone (marks as consumed)
1211    fn write_pages_through_dwb(&self, pages: &[(u32, Page)]) -> Result<(), PagerError> {
1212        if self.dwb_enabled {
1213            for chunk in pages.chunks(DWB_MAX_PAGES_PER_FRAME.max(1)) {
1214                let buf = reddb_file::encode_paged_dwb_frame(
1215                    chunk
1216                        .iter()
1217                        .map(|(page_id, page)| (*page_id, page.as_bytes())),
1218                );
1219
1220                // Write DWB zone and fsync
1221                self.write_dwb_zone_frame(&buf)?;
1222
1223                // Now write pages to their final locations
1224                for (page_id, page) in chunk {
1225                    self.write_page_raw(*page_id, page)?;
1226                }
1227                {
1228                    let file = self.file_lock()?;
1229                    file.sync_all()?;
1230                }
1231
1232                // Reset the zone to its inactive page images.
1233                self.clear_dwb_zone()?;
1234            }
1235
1236            Ok(())
1237        } else {
1238            // DWB disabled — write directly
1239            for (page_id, page) in pages {
1240                self.write_page_raw(*page_id, page)?;
1241            }
1242            Ok(())
1243        }
1244    }
1245
1246    /// Recover from double-write buffer after a crash.
1247    ///
1248    /// If the DWB zone contains valid pages, they were written before the crash
1249    /// interrupted writing to their home locations. Re-apply them.
1250    fn recover_from_dwb(&self) -> Result<(), PagerError> {
1251        let len = {
1252            let file = self.file_lock()?;
1253            file.metadata()?.len()
1254        };
1255        let zone_end = Self::dwb_zone_offset() + DWB_ZONE_BYTES as u64;
1256        if len < zone_end {
1257            return Ok(());
1258        }
1259
1260        let mut buf = vec![0u8; DWB_ZONE_BYTES];
1261        {
1262            let mut file = self.file_lock()?;
1263            file.seek(SeekFrom::Start(Self::dwb_zone_offset()))?;
1264            file.read_exact(&mut buf)?;
1265        }
1266        if buf.get(0..4) != Some(reddb_file::DWB_MAGIC.as_slice()) {
1267            if Self::dwb_zone_is_clear(&buf) {
1268                return Ok(());
1269            }
1270            self.clear_dwb_zone()?;
1271            return Ok(());
1272        }
1273
1274        let entries = match reddb_file::decode_paged_dwb_frame(&buf) {
1275            Ok(entries) => entries,
1276            Err(_) => {
1277                self.clear_dwb_zone()?;
1278                return Ok(());
1279            }
1280        };
1281
1282        // DWB is valid — re-apply pages to main file
1283        for entry in entries {
1284            let page = Page::from_bytes(entry.page);
1285            self.write_page_raw(entry.page_id, &page)?;
1286        }
1287
1288        // Sync and clean up
1289        {
1290            let file = self.file_lock()?;
1291            file.sync_all()?;
1292        }
1293
1294        self.clear_dwb_zone()
1295    }
1296
1297    /// Write header and sync to disk (public for checkpointer).
1298    pub fn write_header_and_sync(&self) -> Result<(), PagerError> {
1299        self.write_header()?;
1300        let file = self.file_lock()?;
1301        file.sync_all()?;
1302        Ok(())
1303    }
1304
1305    /// Set the checkpoint_in_progress flag in the header.
1306    pub fn set_checkpoint_in_progress(
1307        &self,
1308        in_progress: bool,
1309        target_lsn: u64,
1310    ) -> Result<(), PagerError> {
1311        let mut header = self.header_write()?;
1312        header.checkpoint_in_progress = in_progress;
1313        header.checkpoint_target_lsn = target_lsn;
1314        *self.header_dirty_lock()? = true;
1315        drop(header);
1316        self.write_header_and_sync()
1317    }
1318
1319    /// Update the checkpoint LSN and clear the in-progress flag.
1320    pub fn complete_checkpoint(&self, lsn: u64) -> Result<(), PagerError> {
1321        let mut header = self.header_write()?;
1322        header.checkpoint_lsn = lsn;
1323        header.checkpoint_in_progress = false;
1324        header.checkpoint_target_lsn = 0;
1325        *self.header_dirty_lock()? = true;
1326        drop(header);
1327        self.write_header_and_sync()
1328    }
1329}
1330
1331#[cfg(test)]
1332mod tests {
1333    use super::*;
1334
1335    fn temp_db_path(name: &str) -> PathBuf {
1336        std::env::temp_dir().join(format!(
1337            "reddb-pager-{}-{}-{}.rdb",
1338            name,
1339            std::process::id(),
1340            crate::utils::now_unix_nanos()
1341        ))
1342    }
1343
1344    fn cleanup(path: &Path) {
1345        let _ = std::fs::remove_file(path);
1346        let _ = std::fs::remove_file(Pager::dwb_path(path));
1347    }
1348
1349    fn read_zone(path: &Path) -> Vec<u8> {
1350        let mut zone = vec![0u8; SUPERBLOCK_ZONE_SIZE];
1351        let mut file = File::open(path).expect("open() should succeed");
1352        file.read_exact(&mut zone)
1353            .expect("read_exact() should succeed");
1354        zone
1355    }
1356
1357    /// Overwrite one superblock slot in place, leaving the sibling untouched.
1358    fn poke_slot(path: &Path, copy_index: usize, bytes: &[u8]) {
1359        let mut file = OpenOptions::new()
1360            .write(true)
1361            .open(path)
1362            .expect("open() should succeed");
1363        file.seek(SeekFrom::Start(reddb_file::paged_superblock_slot_offset(
1364            copy_index,
1365        )))
1366        .expect("operation should succeed");
1367        file.write_all(bytes).expect("write_all() should succeed");
1368        file.sync_all().expect("sync_all() should succeed");
1369    }
1370
1371    fn stamp_pre_phase3_layout(path: &Path, page_count: u32) {
1372        let zone = read_zone(path);
1373        for copy_index in 0..reddb_file::PAGED_SUPERBLOCK_SLOT_COUNT {
1374            let start = copy_index * SUPERBLOCK_SLOT_SIZE;
1375            let mut slot = zone[start..start + SUPERBLOCK_SLOT_SIZE].to_vec();
1376            let generation = reddb_file::paged_superblock_slot_generation(&slot, copy_index)
1377                .expect("fresh store should publish both slots");
1378            reddb_file::set_database_header_version(&mut slot, PHASE3_DWB_ZONE_FILE_VERSION - 1)
1379                .expect("set_database_header_version() should succeed");
1380            reddb_file::set_database_header_page_count(&mut slot, page_count)
1381                .expect("set_database_header_page_count() should succeed");
1382            reddb_file::seal_paged_superblock_slot(&mut slot, copy_index, generation)
1383                .expect("seal_paged_superblock_slot() should succeed");
1384            poke_slot(path, copy_index, &slot);
1385        }
1386    }
1387
1388    #[test]
1389    fn open_refuses_future_database_version() {
1390        let path = temp_db_path("future-version");
1391        let pager = Pager::open_default(&path).expect("open_default() should succeed");
1392        drop(pager);
1393
1394        // A *valid* superblock slot — correct magic, copy index and CRC — whose
1395        // database header names a version we cannot read. The slot is intact;
1396        // the header is from the future.
1397        let selection = reddb_file::select_paged_superblock(&read_zone(&path))
1398            .expect("operation should succeed");
1399        let mut slot = vec![0u8; SUPERBLOCK_SLOT_SIZE];
1400        reddb_file::init_database_header_page(&mut slot, 1)
1401            .expect("init_database_header_page() should succeed");
1402        reddb_file::set_database_header_version(&mut slot, reddb_file::PAGE_FILE_VERSION + 1)
1403            .expect("operation should succeed");
1404        reddb_file::seal_paged_superblock_slot(
1405            &mut slot,
1406            selection.copy_index,
1407            selection.generation + 1,
1408        )
1409        .expect("operation should succeed");
1410        poke_slot(&path, selection.copy_index, &slot);
1411
1412        let err = match Pager::open_default(&path) {
1413            Ok(_) => panic!("future database version should be rejected"),
1414            Err(err) => err,
1415        };
1416        match err {
1417            PagerError::InvalidDatabase(msg) => {
1418                assert!(msg.contains("newer than supported"), "{msg}");
1419            }
1420            other => panic!("expected InvalidDatabase, got {other:?}"),
1421        }
1422
1423        cleanup(&path);
1424    }
1425
1426    #[test]
1427    fn a_fresh_store_publishes_both_superblock_copies() {
1428        // The ping-pong invariant must hold from the very first update, which
1429        // means creation cannot leave one copy blank.
1430        let path = temp_db_path("fresh-pair");
1431        let pager = Pager::open_default(&path).expect("open_default() should succeed");
1432        drop(pager);
1433
1434        let zone = read_zone(&path);
1435        for copy_index in 0..reddb_file::PAGED_SUPERBLOCK_SLOT_COUNT {
1436            let start = copy_index * SUPERBLOCK_SLOT_SIZE;
1437            assert!(
1438                reddb_file::paged_superblock_slot_generation(
1439                    &zone[start..start + SUPERBLOCK_SLOT_SIZE],
1440                    copy_index
1441                )
1442                .is_some(),
1443                "copy {copy_index} must be valid on a fresh store"
1444            );
1445        }
1446        cleanup(&path);
1447    }
1448
1449    #[test]
1450    fn an_update_writes_the_stale_copy_and_leaves_the_newest_one_intact() {
1451        let path = temp_db_path("ping-pong");
1452        let pager = Pager::open_default(&path).expect("open_default() should succeed");
1453        let before = reddb_file::select_paged_superblock(&read_zone(&path))
1454            .expect("operation should succeed");
1455
1456        pager
1457            .allocate_page(PageType::BTreeLeaf)
1458            .expect("allocate_page() should succeed");
1459        pager.sync().expect("sync() should succeed");
1460
1461        let zone = read_zone(&path);
1462        let after = reddb_file::select_paged_superblock(&zone)
1463            .expect("select_paged_superblock() should succeed");
1464        assert_ne!(
1465            after.copy_index, before.copy_index,
1466            "an update must target the stale copy"
1467        );
1468        assert!(after.generation > before.generation);
1469
1470        // The copy that was newest before the update is still readable — this
1471        // is the byte-level statement of "a crash mid-write always leaves one
1472        // valid copy".
1473        let start = before.copy_index * SUPERBLOCK_SLOT_SIZE;
1474        assert_eq!(
1475            reddb_file::paged_superblock_slot_generation(
1476                &zone[start..start + SUPERBLOCK_SLOT_SIZE],
1477                before.copy_index
1478            ),
1479            Some(before.generation)
1480        );
1481
1482        drop(pager);
1483        cleanup(&path);
1484    }
1485
1486    #[test]
1487    fn open_recovers_from_the_older_copy_when_the_newest_is_torn() {
1488        let path = temp_db_path("torn-newest");
1489        {
1490            let pager = Pager::open_default(&path).expect("open_default() should succeed");
1491            pager
1492                .allocate_page(PageType::BTreeLeaf)
1493                .expect("allocate_page() should succeed");
1494            pager.sync().expect("sync() should succeed");
1495        }
1496
1497        let newest = reddb_file::select_paged_superblock(&read_zone(&path))
1498            .expect("operation should succeed");
1499        // Tear the newest copy: a torn write leaves its CRC broken.
1500        let mut torn = vec![0u8; SUPERBLOCK_SLOT_SIZE];
1501        torn.copy_from_slice(
1502            &read_zone(&path)[newest.copy_index * SUPERBLOCK_SLOT_SIZE
1503                ..newest.copy_index * SUPERBLOCK_SLOT_SIZE + SUPERBLOCK_SLOT_SIZE],
1504        );
1505        torn[64] ^= 0xFF;
1506        poke_slot(&path, newest.copy_index, &torn);
1507
1508        let pager = Pager::open_default(&path).expect("older copy must root the store");
1509        assert!(pager.page_count().expect("page_count() should succeed") >= 3);
1510        drop(pager);
1511        cleanup(&path);
1512    }
1513
1514    #[test]
1515    fn open_refuses_didactically_when_both_superblock_copies_are_invalid() {
1516        let path = temp_db_path("both-invalid");
1517        {
1518            let _ = Pager::open_default(&path).expect("open_default() should succeed");
1519        }
1520
1521        let zone = read_zone(&path);
1522        for copy_index in 0..reddb_file::PAGED_SUPERBLOCK_SLOT_COUNT {
1523            let start = copy_index * SUPERBLOCK_SLOT_SIZE;
1524            let mut slot = zone[start..start + SUPERBLOCK_SLOT_SIZE].to_vec();
1525            slot[64] ^= 0xFF;
1526            poke_slot(&path, copy_index, &slot);
1527        }
1528
1529        let err = match Pager::open_default(&path) {
1530            Ok(_) => panic!("a rootless store must not open"),
1531            Err(err) => err,
1532        };
1533        let rendered = err.to_string();
1534        assert!(matches!(err, PagerError::SuperblockZoneUnrecoverable(_)));
1535        assert!(rendered.contains("superblock zone"), "{rendered}");
1536        assert!(rendered.contains("salvage"), "{rendered}");
1537
1538        cleanup(&path);
1539    }
1540
1541    #[test]
1542    fn open_refuses_a_legacy_sidecar_store_and_names_the_migration_tool() {
1543        let path = temp_db_path("legacy-sidecar");
1544        {
1545            let _ = Pager::open_default(&path).expect("open_default() should succeed");
1546        }
1547
1548        let sidecar = reddb_file::layout::retired::pager_header_path_v0(&path);
1549        std::fs::write(&sidecar, b"legacy header shadow").expect("write() should succeed");
1550
1551        let err = match Pager::open_default(&path) {
1552            Ok(_) => panic!("legacy stores need migration first"),
1553            Err(err) => err,
1554        };
1555        let rendered = err.to_string();
1556        assert!(matches!(err, PagerError::LegacySidecarStore { .. }));
1557        assert!(rendered.contains("migration tool"), "{rendered}");
1558
1559        let _ = std::fs::remove_file(&sidecar);
1560        cleanup(&path);
1561    }
1562
1563    #[test]
1564    fn pre_phase3_large_store_without_sidecar_routes_to_migration() {
1565        let path = temp_db_path("pre-phase3-large");
1566        {
1567            let _ = Pager::open_default(&path).expect("open_default() should succeed");
1568        }
1569        stamp_pre_phase3_layout(&path, FIRST_ALLOCATABLE_PAGE);
1570
1571        let err = match Pager::open_default(&path) {
1572            Ok(_) => panic!("pre-phase-3 stores need migration first"),
1573            Err(err) => err,
1574        };
1575        let rendered = err.to_string();
1576        assert!(matches!(err, PagerError::LegacyPagedStore { .. }));
1577        assert!(rendered.contains("migration tool"), "{rendered}");
1578
1579        cleanup(&path);
1580    }
1581
1582    #[test]
1583    fn pre_phase3_small_store_without_sidecar_routes_to_migration() {
1584        let path = temp_db_path("pre-phase3-small");
1585        {
1586            let _ = Pager::open_default(&path).expect("open_default() should succeed");
1587        }
1588        stamp_pre_phase3_layout(&path, 3);
1589        OpenOptions::new()
1590            .write(true)
1591            .open(&path)
1592            .expect("open() should succeed")
1593            .set_len(3 * PAGE_SIZE as u64)
1594            .expect("set_len() should succeed");
1595
1596        let err = match Pager::open_default(&path) {
1597            Ok(_) => panic!("pre-phase-3 stores need migration first"),
1598            Err(err) => err,
1599        };
1600        assert!(matches!(err, PagerError::LegacyPagedStore { .. }));
1601
1602        cleanup(&path);
1603    }
1604
1605    #[test]
1606    fn no_phase_one_sidecar_survives_a_create_write_reopen_cycle() {
1607        let path = temp_db_path("census");
1608        {
1609            let pager = Pager::open_default(&path).expect("open_default() should succeed");
1610            let page = pager
1611                .allocate_page(PageType::BTreeLeaf)
1612                .expect("allocate_page() should succeed");
1613            pager
1614                .write_page(page.page_id(), page)
1615                .expect("operation should succeed");
1616            pager.sync().expect("sync() should succeed");
1617        }
1618        {
1619            let pager = Pager::open_default(&path).expect("open_default() should succeed");
1620            pager.sync().expect("sync() should succeed");
1621        }
1622
1623        assert_eq!(
1624            reddb_file::layout::retired::first_present_phase1_sidecar(&path),
1625            None,
1626            "the pager must never create a retired phase-1 sidecar"
1627        );
1628        cleanup(&path);
1629    }
1630}