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