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            allocation_reused: std::sync::atomic::AtomicBool::new(false),
270            allocation_fence_pending: std::sync::atomic::AtomicBool::new(false),
271            header: RwLock::new(DatabaseHeader::default()),
272            config,
273            header_dirty: Mutex::new(false),
274            wal: RwLock::new(None),
275            encryption: None,
276        };
277
278        if exists {
279            // Recover from the in-file double-write zone before loading the
280            // header or allowing WAL replay to read pages.
281            pager.ensure_phase3_dwb_zone_layout()?;
282            pager.recover_from_dwb()?;
283            // Load existing database (with header shadow fallback)
284            pager.load_header()?;
285            pager.bind_encryption_for_existing()?;
286        } else {
287            // Initialize new database
288            pager.initialize()?;
289            pager.bind_encryption_for_new()?;
290        }
291
292        Ok(pager)
293    }
294
295    /// gh-892 diagnostic predicate: returns `true` when the database
296    /// `page_size` is **not** an integer multiple of the filesystem's
297    /// reported block size (`st_blksize`), i.e. `page_size % fs_block_size
298    /// != 0`. A `fs_block_size` of `0` (probe unavailable / unknown) is
299    /// treated as aligned so a missing probe never produces a warning.
300    /// Pure function with no I/O so the warn decision is unit-testable.
301    pub(crate) fn page_size_misaligned_with_block(page_size: usize, fs_block_size: u64) -> bool {
302        fs_block_size != 0 && !(page_size as u64).is_multiple_of(fs_block_size)
303    }
304
305    #[cfg(test)]
306    fn cow_filesystem_test_override() -> Option<bool> {
307        match COW_ATOMIC_WRITE_TEST_OVERRIDE.load(Ordering::Relaxed) {
308            1 => Some(true),
309            2 => Some(false),
310            _ => None,
311        }
312    }
313
314    #[cfg(not(test))]
315    fn cow_filesystem_test_override() -> Option<bool> {
316        None
317    }
318
319    fn cow_filesystem_has_atomic_page_writes(
320        path: &Path,
321        file: &File,
322    ) -> Option<CowFilesystemKind> {
323        if let Some(allowed) = Self::cow_filesystem_test_override() {
324            return allowed.then_some(CowFilesystemKind::TestOverride);
325        }
326        Self::probe_cow_filesystem(path, file)
327    }
328
329    #[cfg(target_os = "linux")]
330    fn probe_cow_filesystem(path: &Path, file: &File) -> Option<CowFilesystemKind> {
331        let fs_type = linux_fstatfs_type(file)?;
332        match fs_type {
333            ZFS_SUPER_MAGIC => Some(CowFilesystemKind::Zfs),
334            BTRFS_SUPER_MAGIC => {
335                let mount_options = linux_mount_options_for_path(path)?;
336                let inode_flags = linux_inode_flags(file)?;
337                classify_cow_filesystem(fs_type, Some(&mount_options), Some(inode_flags))
338            }
339            _ => None,
340        }
341    }
342
343    #[cfg(not(target_os = "linux"))]
344    fn probe_cow_filesystem(_path: &Path, _file: &File) -> Option<CowFilesystemKind> {
345        None
346    }
347
348    /// Inspect page 0 for the `RDBE` encryption marker, then resolve
349    /// the (key, marker) matrix:
350    ///
351    /// | Marker | Key supplied | Result                              |
352    /// |--------|--------------|-------------------------------------|
353    /// | yes    | yes          | Bind encryptor; validate key        |
354    /// | yes    | no           | `EncryptionRequired` (fail closed)  |
355    /// | no     | yes          | `PlainDatabaseRefusesKey`           |
356    /// | no     | no           | Plain pager — no binding needed     |
357    fn bind_encryption_for_existing(&mut self) -> Result<(), PagerError> {
358        if self.page_count().unwrap_or(0) == 0 {
359            return self.bind_encryption_for_new();
360        }
361        let (_, image) = self.newest_superblock()?;
362        let data = &image[..];
363        let has_marker = reddb_file::paged_encryption_marker_present(data);
364
365        let key = self.config.encryption.clone();
366        match (has_marker, key) {
367            (true, Some(key)) => {
368                let header_bytes =
369                    reddb_file::paged_encryption_header_bytes(data).ok_or_else(|| {
370                        PagerError::InvalidDatabase("encryption header parse failed".to_string())
371                    })?;
372                let header = crate::storage::encryption::EncryptionHeader::from_bytes(header_bytes)
373                    .map_err(|e| {
374                        PagerError::InvalidDatabase(format!("encryption header parse failed: {e}"))
375                    })?;
376                if !header.validate(&key) {
377                    return Err(PagerError::InvalidKey);
378                }
379                let encryptor = crate::storage::encryption::PageEncryptor::new(key);
380                self.encryption = Some((encryptor, header));
381                Ok(())
382            }
383            (true, None) => Err(PagerError::EncryptionRequired),
384            (false, Some(_)) => Err(PagerError::PlainDatabaseRefusesKey),
385            (false, None) => Ok(()),
386        }
387    }
388
389    /// New DB: if a key is configured, write the marker + header to
390    /// page 0 so subsequent opens detect encryption.
391    fn bind_encryption_for_new(&mut self) -> Result<(), PagerError> {
392        let Some(key) = self.config.encryption.clone() else {
393            return Ok(());
394        };
395        let header = crate::storage::encryption::EncryptionHeader::new(&key);
396        let encryptor = crate::storage::encryption::PageEncryptor::new(key);
397
398        // Stamp the marker + header into the superblock zone if `initialize()`
399        // has already published it.
400        if self.page_count().unwrap_or(0) > 0 {
401            let header_bytes = header.to_bytes();
402            self.commit_superblock(|image| {
403                reddb_file::write_paged_encryption_marker_and_header(image, &header_bytes)
404                    .map_err(|err| PagerError::InvalidDatabase(err.to_string()))
405            })?;
406        }
407        self.encryption = Some((encryptor, header));
408        Ok(())
409    }
410
411    /// Open with default configuration
412    pub fn open_default<P: AsRef<Path>>(path: P) -> Result<Self, PagerError> {
413        Self::open(path, PagerConfig::default())
414    }
415
416    /// Initialize a new database
417    fn initialize(&self) -> Result<(), PagerError> {
418        if self.config.read_only {
419            return Err(PagerError::ReadOnly);
420        }
421
422        // Create header page. Page ids 1 and 2 are reserved for fixed
423        // metadata/vault pages. ADR 0038 phase 3 reserves the following DWB
424        // zone in-file so normal B-tree allocation starts after it.
425        let initial_page_count = FIRST_ALLOCATABLE_PAGE;
426        let header_page = Page::new_header_page(initial_page_count);
427        self.header_write()?.page_count = initial_page_count;
428
429        // Reserve page 0 (the superblock zone) plus the reserved pages so any
430        // scan over 0..page_count can read every allocated page in a brand-new
431        // database. The zone's own bytes are published below, one slot at a
432        // time; this first whole-page write only sizes the file.
433        self.write_page_raw(0, &header_page)?;
434        let mut metadata_page = Page::new(PageType::Header, 1);
435        metadata_page.update_checksum();
436        self.write_page_raw(1, &metadata_page)?;
437        let mut vault_page = Page::new(PageType::Vault, 2);
438        vault_page.update_checksum();
439        self.write_page_raw(2, &vault_page)?;
440        self.clear_dwb_zone()?;
441
442        // Publish both superblock copies so the ping-pong invariant — one
443        // valid copy always survives a crash — holds from the very first
444        // update onward, not just from the second.
445        let mut slot = superblock_slot_image(header_page.as_bytes());
446        self.write_superblock_slot(&mut slot, 0, 1)?;
447        self.write_superblock_slot(&mut slot, 1, 2)?;
448
449        // Sync to disk
450        self.sync()?;
451
452        Ok(())
453    }
454
455    /// Acquire a write lock on the header RwLock, mapping poison errors.
456    fn header_write(&self) -> Result<std::sync::RwLockWriteGuard<'_, DatabaseHeader>, PagerError> {
457        self.header.write().map_err(|_| PagerError::LockPoisoned)
458    }
459
460    /// Acquire a read lock on the header RwLock, mapping poison errors.
461    fn header_read(&self) -> Result<std::sync::RwLockReadGuard<'_, DatabaseHeader>, PagerError> {
462        self.header.read().map_err(|_| PagerError::LockPoisoned)
463    }
464
465    /// Acquire a write lock on the freelist RwLock, mapping poison errors.
466    fn freelist_write(&self) -> Result<std::sync::RwLockWriteGuard<'_, FreeList>, PagerError> {
467        self.freelist.write().map_err(|_| PagerError::LockPoisoned)
468    }
469
470    /// Acquire a lock on the file Mutex, mapping poison errors.
471    fn file_lock(&self) -> Result<std::sync::MutexGuard<'_, File>, PagerError> {
472        self.file.lock().map_err(|_| PagerError::LockPoisoned)
473    }
474
475    /// Acquire a lock on the header_dirty Mutex, mapping poison errors.
476    fn header_dirty_lock(&self) -> Result<std::sync::MutexGuard<'_, bool>, PagerError> {
477        self.header_dirty
478            .lock()
479            .map_err(|_| PagerError::LockPoisoned)
480    }
481
482    /// Read the whole superblock zone (both ping-pong copies) from the head of
483    /// the file. A file shorter than the zone yields zero-filled tail bytes,
484    /// which decode as "slot never written".
485    fn read_superblock_zone(&self) -> Result<[u8; SUPERBLOCK_ZONE_SIZE], PagerError> {
486        let mut file = self.file_lock()?;
487        let len = file.metadata().map(|meta| meta.len()).unwrap_or(0);
488        let mut zone = [0u8; SUPERBLOCK_ZONE_SIZE];
489        let readable = len.min(SUPERBLOCK_ZONE_SIZE as u64) as usize;
490        if readable > 0 {
491            file.seek(SeekFrom::Start(0))?;
492            file.read_exact(&mut zone[..readable])?;
493        }
494        Ok(zone)
495    }
496
497    /// The newest valid superblock copy, with its slot image.
498    ///
499    /// Both copies invalid is the one unrecoverable superblock state (ADR 0074
500    /// §2): the store does not open, and the error names the zone and points
501    /// at salvage rather than panicking or silently re-initialising.
502    fn newest_superblock(&self) -> Result<NewestSuperblock, PagerError> {
503        let zone = self.read_superblock_zone()?;
504        let selection = reddb_file::select_paged_superblock(&zone)
505            .ok_or_else(|| PagerError::SuperblockZoneUnrecoverable(self.path.clone()))?;
506
507        let start = selection.copy_index * SUPERBLOCK_SLOT_SIZE;
508        let mut image = [0u8; SUPERBLOCK_SLOT_SIZE];
509        image.copy_from_slice(&zone[start..start + SUPERBLOCK_SLOT_SIZE]);
510        Ok((selection, image))
511    }
512
513    fn ensure_phase3_dwb_zone_layout(&self) -> Result<(), PagerError> {
514        let (_, image) = self.newest_superblock()?;
515        let header = reddb_file::decode_database_header(&image)
516            .map_err(|err| PagerError::InvalidDatabase(err.to_string()))?;
517        if header.version < PHASE3_DWB_ZONE_FILE_VERSION {
518            return Err(PagerError::LegacyPagedStore {
519                path: self.path.clone(),
520            });
521        }
522        Ok(())
523    }
524
525    /// Seal `slot` for `copy_index`/`generation` and write exactly that slot,
526    /// then fsync. The caller guarantees `copy_index` is the stale copy.
527    fn write_superblock_slot(
528        &self,
529        slot: &mut [u8; SUPERBLOCK_SLOT_SIZE],
530        copy_index: usize,
531        generation: u64,
532    ) -> Result<(), PagerError> {
533        if self.config.read_only {
534            return Err(PagerError::ReadOnly);
535        }
536        reddb_file::seal_paged_superblock_slot(slot, copy_index, generation)
537            .map_err(|err| PagerError::InvalidDatabase(err.to_string()))?;
538
539        let mut file = self.file_lock()?;
540        file.seek(SeekFrom::Start(reddb_file::paged_superblock_slot_offset(
541            copy_index,
542        )))?;
543        file.write_all(slot)?;
544        file.sync_all()?;
545        Ok(())
546    }
547
548    /// Publish `mutate`'s edit of the newest slot image into the *stale* copy.
549    ///
550    /// Reading the newest image first means every byte the pager does not
551    /// explicitly rewrite — most importantly the encryption marker and header —
552    /// carries forward untouched.
553    fn commit_superblock(
554        &self,
555        mutate: impl FnOnce(&mut [u8; SUPERBLOCK_SLOT_SIZE]) -> Result<(), PagerError>,
556    ) -> Result<(), PagerError> {
557        if self.config.read_only {
558            return Err(PagerError::ReadOnly);
559        }
560        let (newest, mut image) = self.newest_superblock()?;
561        mutate(&mut image)?;
562        let next_copy = reddb_file::paged_superblock_next_copy(Some(newest));
563        self.write_superblock_slot(&mut image, next_copy, newest.generation.saturating_add(1))
564    }
565
566    /// Load the database header from the newest valid superblock copy.
567    fn load_header(&self) -> Result<(), PagerError> {
568        let (_, image) = self.newest_superblock()?;
569
570        let mut decoded_header = reddb_file::decode_database_header(&image)
571            .map_err(|err| PagerError::InvalidDatabase(err.to_string()))?;
572        // Cache eviction (or DWB recovery) can persist pages before the next
573        // superblock publication. After a crash the allocation counter may
574        // therefore lag behind live pages already in the file. Reserve the
575        // entire physical extent before allocating; never overwrite that tail.
576        // A partial last page is reserved too, without treating it as verified.
577        let physical_pages = self
578            .file_lock()?
579            .metadata()?
580            .len()
581            .div_ceil(PAGE_SIZE as u64);
582        let physical_pages = u32::try_from(physical_pages).map_err(|_| {
583            PagerError::InvalidDatabase("data file exceeds the page address space".into())
584        })?;
585        if decoded_header.page_count < physical_pages {
586            decoded_header.page_count = physical_pages;
587            if !self.config.read_only {
588                *self.header_dirty_lock()? = true;
589            }
590        }
591        let freelist_head = decoded_header.freelist_head;
592
593        {
594            let mut header = self.header_write()?;
595            *header = decoded_header;
596        }
597
598        // Initialize freelist
599        {
600            let mut freelist = self.freelist_write()?;
601            *freelist = FreeList::from_header(freelist_head, 0);
602        }
603
604        Ok(())
605    }
606
607    /// Publish the in-memory database header as a new superblock generation.
608    fn write_header(&self) -> Result<(), PagerError> {
609        if self.config.read_only {
610            return Err(PagerError::ReadOnly);
611        }
612
613        let header = self.header_read()?.clone();
614        self.commit_superblock(|image| {
615            reddb_file::encode_database_header(image, &header)
616                .map_err(|err| PagerError::InvalidDatabase(err.to_string()))
617        })?;
618        *self.header_dirty_lock()? = false;
619
620        Ok(())
621    }
622
623    /// Read a page from disk (bypassing cache)
624    fn read_page_raw(&self, page_id: u32) -> Result<Page, PagerError> {
625        let mut file = self.file_lock()?;
626        let offset = (page_id as u64) * (PAGE_SIZE as u64);
627
628        file.seek(SeekFrom::Start(offset))?;
629
630        let mut buf = [0u8; PAGE_SIZE];
631        file.read_exact(&mut buf)?;
632
633        let page = Page::from_bytes(buf);
634
635        // Page 0 is the superblock zone, not a checksummed page: its two slots
636        // are sealed independently, so no whole-page checksum can hold across
637        // a ping-pong write. Slot CRCs are its integrity contract.
638        if self.config.verify_checksums && page_id != 0 {
639            page.verify_checksum()?;
640        }
641
642        Ok(page)
643    }
644
645    /// Write a page to disk (bypassing cache)
646    fn write_page_raw(&self, page_id: u32, page: &Page) -> Result<(), PagerError> {
647        if self.config.read_only {
648            return Err(PagerError::ReadOnly);
649        }
650
651        let mut file = self.file_lock()?;
652        let offset = (page_id as u64) * (PAGE_SIZE as u64);
653
654        file.seek(SeekFrom::Start(offset))?;
655        file.write_all(page.as_bytes())?;
656
657        Ok(())
658    }
659
660    /// Read a page (cache-aware)
661    pub fn read_page(&self, page_id: u32) -> Result<Page, PagerError> {
662        // Check cache first
663        if let Some(page) = self.cache.get(page_id) {
664            return Ok(page);
665        }
666
667        // Cache miss - read from disk
668        let page = self.read_page_raw(page_id)?;
669
670        // Add to cache
671        if let Some(dirty_page) = self.cache.insert(page_id, page.clone()) {
672            // Evicted page was dirty, need to write it back
673            let evicted_id = dirty_page.page_id();
674            self.write_page_raw(evicted_id, &dirty_page)?;
675        }
676
677        Ok(page)
678    }
679
680    /// Read a page without verifying checksum (for encrypted pages)
681    ///
682    /// Use this when the page content has its own integrity protection
683    /// (e.g., AES-GCM authentication tag for encrypted pages).
684    pub fn read_page_no_checksum(&self, page_id: u32) -> Result<Page, PagerError> {
685        // Check cache first
686        if let Some(page) = self.cache.get(page_id) {
687            return Ok(page);
688        }
689
690        // Cache miss - read from disk (skip checksum verification)
691        let mut file = self.file_lock()?;
692        let offset = (page_id as u64) * (PAGE_SIZE as u64);
693
694        file.seek(SeekFrom::Start(offset))?;
695
696        let mut buf = [0u8; PAGE_SIZE];
697        file.read_exact(&mut buf)?;
698        drop(file);
699
700        let page = Page::from_bytes(buf);
701
702        // Add to cache (no checksum verification)
703        if let Some(dirty_page) = self.cache.insert(page_id, page.clone()) {
704            // Evicted page was dirty, need to write it back
705            let evicted_id = dirty_page.page_id();
706            self.write_page_raw(evicted_id, &dirty_page)?;
707        }
708
709        Ok(page)
710    }
711
712    /// Write a page (cache-aware)
713    pub fn write_page(&self, page_id: u32, mut page: Page) -> Result<(), PagerError> {
714        if self.config.read_only {
715            return Err(PagerError::ReadOnly);
716        }
717
718        // Update checksum
719        page.update_checksum();
720
721        // Add to cache and mark dirty. If the cache had to evict a
722        // dirty entry, write it through immediately — the evicted
723        // page will never be flushed otherwise (same bug fixed in
724        // `allocate_page`).
725        if let Some(dirty_page) = self.cache.insert(page_id, page) {
726            let evicted_id = dirty_page.page_id();
727            self.write_page_raw(evicted_id, &dirty_page)?;
728        }
729        self.cache.mark_dirty(page_id);
730
731        Ok(())
732    }
733
734    /// Read a page through the configured encryptor if any. Page 0
735    /// is always returned plaintext (it carries the encryption marker
736    /// + header). Callers that want raw cipher bytes can use
737    ///   `read_page_no_checksum` directly.
738    pub fn read_page_decrypted(&self, page_id: u32) -> Result<Page, PagerError> {
739        if page_id == 0 || self.encryption.is_none() {
740            return self.read_page(page_id);
741        }
742        let raw = self.read_page_no_checksum(page_id)?;
743        let (enc, _) = self
744            .encryption
745            .as_ref()
746            .expect("encryption presence checked above");
747        let plaintext = enc
748            .decrypt(page_id, raw.as_bytes())
749            .map_err(|e| PagerError::InvalidDatabase(format!("decrypt page {page_id}: {e}")))?;
750        let mut buf = [0u8; PAGE_SIZE];
751        let n = plaintext.len().min(PAGE_SIZE);
752        buf[..n].copy_from_slice(&plaintext[..n]);
753        Ok(Page::from_bytes(buf))
754    }
755
756    /// Write a page through the configured encryptor if any. Page 0
757    /// bypasses encryption and goes through the normal checksummed
758    /// path. Encrypted pages skip the checksum update because
759    /// AES-GCM's authentication tag is the integrity guarantee.
760    pub fn write_page_encrypted(&self, page_id: u32, page: Page) -> Result<(), PagerError> {
761        if page_id == 0 || self.encryption.is_none() {
762            return self.write_page(page_id, page);
763        }
764        // Canonical per-page envelope overhead (nonce + GCM tag),
765        // owned by reddb-io-crypto (#1053, ADR 0054).
766        const OVERHEAD: usize = crate::storage::encryption::page_encryptor::OVERHEAD;
767        let plaintext_len = PAGE_SIZE - OVERHEAD;
768        let plaintext = &page.as_bytes()[..plaintext_len];
769        let (enc, _) = self
770            .encryption
771            .as_ref()
772            .expect("encryption presence checked above");
773        let ciphertext = enc.encrypt(page_id, plaintext);
774        debug_assert_eq!(ciphertext.len(), PAGE_SIZE);
775        let mut buf = [0u8; PAGE_SIZE];
776        buf.copy_from_slice(&ciphertext);
777        let cipher_page = Page::from_bytes(buf);
778        self.write_page_no_checksum(page_id, cipher_page)
779    }
780
781    /// Write a page without updating checksum (for encrypted pages)
782    ///
783    /// Use this when the page content has its own integrity protection
784    /// (e.g., AES-GCM authentication tag for encrypted pages).
785    pub fn write_page_no_checksum(&self, page_id: u32, page: Page) -> Result<(), PagerError> {
786        if self.config.read_only {
787            return Err(PagerError::ReadOnly);
788        }
789
790        // Add to cache and mark dirty (no checksum update). Same
791        // eviction-write-through guard as `write_page`.
792        if let Some(dirty_page) = self.cache.insert(page_id, page) {
793            let evicted_id = dirty_page.page_id();
794            self.write_page_raw(evicted_id, &dirty_page)?;
795        }
796        self.cache.mark_dirty(page_id);
797
798        Ok(())
799    }
800
801    /// Allocate a new page
802    pub fn allocate_page(&self, page_type: PageType) -> Result<Page, PagerError> {
803        if self.config.read_only {
804            return Err(PagerError::ReadOnly);
805        }
806
807        // Try to get from freelist first
808        let page_id = {
809            let mut freelist = self.freelist_write()?;
810            // A previous fence can fail after loading the trunk into memory.
811            // Retry it before the fast path can consume any of those pages.
812            self.complete_allocation_fence()?;
813            if let Some(id) = freelist.allocate() {
814                self.allocation_reused
815                    .store(true, std::sync::atomic::Ordering::Release);
816                id
817            } else if freelist.trunk_head() != 0 {
818                let trunk_id = freelist.trunk_head();
819
820                let trunk = self.read_page(trunk_id).map_err(|e| match e {
821                    PagerError::PageNotFound(_) => {
822                        PagerError::InvalidDatabase("Freelist trunk missing".to_string())
823                    }
824                    other => other,
825                })?;
826
827                freelist
828                    .load_from_trunk(&trunk)
829                    .map_err(|e| PagerError::InvalidDatabase(format!("Freelist: {}", e)))?;
830                let mut header = self.header_write()?;
831                header.freelist_head = freelist.trunk_head();
832                *self.header_dirty_lock()? = true;
833                drop(header);
834                // The consumed trunk itself becomes reusable. Forget its old
835                // allocator role durably before cache eviction can overwrite it.
836                // Keep the allocator locked so another allocator cannot race
837                // ahead of this fence. A failed fence leaves allocation stopped.
838                self.allocation_fence_pending
839                    .store(true, std::sync::atomic::Ordering::Release);
840                self.complete_allocation_fence()?;
841                let id = freelist.allocate().ok_or_else(|| {
842                    PagerError::InvalidDatabase("Freelist empty after trunk load".to_string())
843                })?;
844                self.allocation_reused
845                    .store(true, std::sync::atomic::Ordering::Release);
846                id
847            } else {
848                // No free pages, extend file
849                let mut header = self.header_write()?;
850                let id = header.page_count;
851                header.page_count += 1;
852                *self.header_dirty_lock()? = true;
853                id
854            }
855        };
856
857        let page = Page::new(page_type, page_id);
858
859        // Write to cache. The evicted page (if any) is dirty by
860        // definition — `cache.insert` only returns `Some` when it
861        // had to evict a dirty entry to make room. The previous
862        // version dropped that return value, which silently lost
863        // writes whenever a freshly-allocated page caused an LRU
864        // eviction. This shows up under heavy ingest as
865        // "B-tree insert error: Pager error: I/O error: failed to
866        // fill whole buffer" later, when something tries to read
867        // back the never-flushed page. Mirror `read_page`'s
868        // handling: write the evicted page through immediately so
869        // the on-disk image stays consistent.
870        if let Some(dirty_page) = self.cache.insert(page_id, page.clone()) {
871            let evicted_id = dirty_page.page_id();
872            self.write_page_raw(evicted_id, &dirty_page)?;
873        }
874        self.cache.mark_dirty(page_id);
875
876        Ok(page)
877    }
878
879    /// Reserve a contiguous extent of vector pages.
880    pub fn reserve_contig_extent(&self, n_pages: u32) -> Result<super::ExtentId, PagerError> {
881        if self.config.read_only {
882            return Err(PagerError::ReadOnly);
883        }
884        if n_pages == 0 {
885            return Err(PagerError::InvalidDatabase(
886                "contiguous extent must reserve at least one page".to_string(),
887            ));
888        }
889
890        let start_page = {
891            let mut header = self.header_write()?;
892            let start = header.page_count;
893            header.page_count = header.page_count.checked_add(n_pages).ok_or_else(|| {
894                PagerError::InvalidDatabase("contiguous extent page count overflow".to_string())
895            })?;
896            *self.header_dirty_lock()? = true;
897            start
898        };
899
900        for page_id in start_page..start_page + n_pages {
901            let mut page = Page::new(PageType::Vector, page_id);
902            page.update_checksum();
903            if let Some(dirty_page) = self.cache.insert(page_id, page) {
904                let evicted_id = dirty_page.page_id();
905                self.write_page_raw(evicted_id, &dirty_page)?;
906            }
907            self.cache.mark_dirty(page_id);
908        }
909
910        Ok(super::ExtentId {
911            start_page,
912            n_pages,
913        })
914    }
915
916    /// Free a page (return to freelist)
917    pub fn free_page(&self, page_id: u32) -> Result<(), PagerError> {
918        if self.config.read_only {
919            return Err(PagerError::ReadOnly);
920        }
921
922        // Remove from cache
923        self.cache.remove(page_id);
924
925        // Add to freelist
926        let mut freelist = self.freelist_write()?;
927        freelist.free(page_id);
928
929        *self.header_dirty_lock()? = true;
930
931        Ok(())
932    }
933
934    /// Return a fully validated, unpublished tree generation in one allocator
935    /// mutation. Call only after its removal from the manifest is durable.
936    pub(crate) fn free_retired_pages(&self, pages: &[u32]) -> Result<(), PagerError> {
937        if self.config.read_only {
938            return Err(PagerError::ReadOnly);
939        }
940        let page_count = self.page_count()?;
941        if pages
942            .iter()
943            .any(|id| *id < FIRST_ALLOCATABLE_PAGE || *id >= page_count)
944        {
945            return Err(PagerError::InvalidDatabase(
946                "invalid retired page range".into(),
947            ));
948        }
949        let mut freelist = self.freelist_write()?;
950        for id in pages {
951            self.cache.remove(*id);
952        }
953        freelist.free_batch(pages);
954        *self.header_dirty_lock()? = true;
955        Ok(())
956    }
957
958    /// Get database header
959    pub fn header(&self) -> Result<DatabaseHeader, PagerError> {
960        Ok(self.header_read()?.clone())
961    }
962
963    pub fn physical_header(&self) -> Result<PhysicalFileHeader, PagerError> {
964        Ok(self.header_read()?.physical)
965    }
966
967    pub fn update_physical_header(&self, physical: PhysicalFileHeader) -> Result<(), PagerError> {
968        if self.config.read_only {
969            return Err(PagerError::ReadOnly);
970        }
971
972        let mut header = self.header_write()?;
973        header.physical = physical;
974        *self.header_dirty_lock()? = true;
975        Ok(())
976    }
977
978    /// Get page count
979    pub fn page_count(&self) -> Result<u32, PagerError> {
980        Ok(self.header_read()?.page_count)
981    }
982
983    /// Attach a WAL writer to enforce WAL-first flush ordering.
984    ///
985    /// After this call, [`Pager::flush`] computes the maximum
986    /// `header.lsn` over all dirty pages and calls
987    /// `WalWriter::flush_until(max_lsn)` before any page is written
988    /// to the data file. This is the postgres rule: a page on disk
989    /// implies its WAL record is already durable on disk.
990    ///
991    /// Existing call sites that construct a Pager without a WAL
992    /// keep their previous behaviour (no LSN check) — wiring is
993    /// strictly opt-in.
994    pub fn set_wal_writer(&self, wal: Arc<Mutex<crate::storage::wal::writer::WalWriter>>) {
995        let mut slot = self.wal.write().unwrap_or_else(|p| p.into_inner());
996        *slot = Some(wal);
997    }
998
999    /// Detach the WAL writer (test / shutdown path).
1000    pub fn clear_wal_writer(&self) {
1001        let mut slot = self.wal.write().unwrap_or_else(|p| p.into_inner());
1002        *slot = None;
1003    }
1004
1005    /// Has a WAL writer been attached?
1006    pub fn has_wal_writer(&self) -> bool {
1007        self.wal.read().map(|s| s.is_some()).unwrap_or(false)
1008    }
1009
1010    /// Flush all dirty pages to disk
1011    pub fn flush(&self) -> Result<(), PagerError> {
1012        if self.config.read_only {
1013            return Ok(());
1014        }
1015
1016        // Persist freelist to trunk pages when dirty
1017        let trunks = {
1018            let mut freelist = self.freelist_write()?;
1019            if freelist.is_dirty() {
1020                let mut header = self.header_write()?;
1021                let trunks = freelist.flush_to_trunks(0, || {
1022                    let id = header.page_count;
1023                    header.page_count += 1;
1024                    id
1025                });
1026                header.freelist_head = freelist.trunk_head();
1027                *self.header_dirty_lock()? = true;
1028                freelist.mark_clean();
1029                trunks
1030            } else {
1031                Vec::new()
1032            }
1033        };
1034
1035        let mut dirty_pages = Vec::new();
1036        for trunk in trunks {
1037            let page_id = trunk.page_id();
1038            if let Some(evicted) = self.cache.insert(page_id, trunk) {
1039                // Trunk publication can evict dirty data. Keep it in the same
1040                // WAL-first, double-write-protected batch as resident pages.
1041                dirty_pages.push((evicted.page_id(), evicted));
1042            }
1043            self.cache.mark_dirty(page_id);
1044        }
1045
1046        // Flush dirty pages from cache (through DWB if enabled)
1047        dirty_pages.extend(self.cache.flush_dirty());
1048        if !dirty_pages.is_empty() {
1049            // WAL-FIRST: ensure every WAL record describing a dirty
1050            // page is durable BEFORE the page itself reaches disk.
1051            // Pages with `lsn == 0` are exempt (freelist trunks, header
1052            // shadow pages, anything not produced by a WAL append).
1053            let max_lsn = dirty_pages
1054                .iter()
1055                .filter_map(|(_, page)| page.header().ok().map(|h| h.lsn))
1056                .max()
1057                .unwrap_or(0);
1058            if max_lsn > 0 {
1059                if let Ok(slot) = self.wal.read() {
1060                    if let Some(wal) = slot.as_ref() {
1061                        let wal = Arc::clone(wal);
1062                        // Drop the read lock before taking the WAL
1063                        // mutex so an unrelated reader cannot block
1064                        // the flush path.
1065                        drop(slot);
1066                        let mut wal_guard = wal.lock().unwrap_or_else(|p| p.into_inner());
1067                        wal_guard.flush_until(max_lsn).map_err(PagerError::Io)?;
1068                    }
1069                }
1070            }
1071            self.write_pages_through_dwb(&dirty_pages)?;
1072        }
1073
1074        // Write header if dirty
1075        if *self.header_dirty_lock()? {
1076            self.write_header()?;
1077        }
1078
1079        Ok(())
1080    }
1081
1082    /// Sync file to disk (fsync)
1083    pub fn sync(&self) -> Result<(), PagerError> {
1084        self.flush()?;
1085
1086        if self
1087            .allocation_reused
1088            .swap(false, std::sync::atomic::Ordering::AcqRel)
1089        {
1090            // flush published the current allocator in one slot. Mirror that
1091            // ownership before callers publish roots that use reclaimed pages;
1092            // recovery through the sibling slot must never allocate them again.
1093            if let Err(error) = self.write_header_and_sync() {
1094                self.allocation_reused
1095                    .store(true, std::sync::atomic::Ordering::Release);
1096                return Err(error);
1097            }
1098        }
1099
1100        let file = self.file_lock()?;
1101        file.sync_all()?;
1102
1103        Ok(())
1104    }
1105
1106    /// Get cache statistics
1107    pub fn cache_stats(&self) -> crate::storage::engine::page_cache::CacheStats {
1108        self.cache.stats()
1109    }
1110
1111    /// Pages currently resident in the page cache. Slots are fixed size, so
1112    /// this times the page size is the cache's memory footprint (ADR 0073 §2).
1113    pub fn cache_len(&self) -> usize {
1114        self.cache.len()
1115    }
1116
1117    /// Preallocated page-cache slots — the sizing this process derived from
1118    /// its page-cache budget share.
1119    pub fn cache_capacity(&self) -> usize {
1120        self.cache.capacity()
1121    }
1122
1123    /// Count dirty pages currently in the page cache.
1124    pub fn dirty_page_count(&self) -> usize {
1125        self.cache.dirty_count()
1126    }
1127
1128    /// Estimated fraction of the page cache holding dirty pages.
1129    /// Returned in `[0, 1]`. Used by the background writer to
1130    /// decide when to kick in aggressive flushing.
1131    pub fn dirty_fraction(&self) -> f64 {
1132        let capacity = self.cache.capacity().max(1) as f64;
1133        self.cache.dirty_count() as f64 / capacity
1134    }
1135
1136    /// Flush up to `max` dirty pages from the cache. Returns the
1137    /// number actually written. Background-writer entry point —
1138    /// reuses the same persistence path as `flush()` but bounded.
1139    pub fn flush_some_dirty(&self, max: usize) -> Result<usize, PagerError> {
1140        if self.config.read_only || max == 0 {
1141            return Ok(0);
1142        }
1143        let dirty_pages = self.cache.flush_some_dirty(max);
1144        if dirty_pages.is_empty() {
1145            return Ok(0);
1146        }
1147        let count = dirty_pages.len();
1148        // WAL-first: every cached dirty page carries an LSN that the
1149        // WAL must have already persisted. The full `flush()` path
1150        // enforces this with `wal.flush(max_lsn)`; here we simply
1151        // write through the pager — safe because callers only reach
1152        // this path via the bgwriter, which runs asynchronously
1153        // alongside normal commits that already respect WAL-first.
1154        for (page_id, page) in dirty_pages {
1155            self.write_page(page_id, page)?;
1156        }
1157        Ok(count)
1158    }
1159
1160    /// Get database file path
1161    pub fn path(&self) -> &Path {
1162        &self.path
1163    }
1164
1165    /// Check if database is read-only
1166    pub fn is_read_only(&self) -> bool {
1167        self.config.read_only
1168    }
1169
1170    /// Get file size in bytes
1171    pub fn file_size(&self) -> Result<u64, PagerError> {
1172        let file = self.file_lock()?;
1173        Ok(file.metadata()?.len())
1174    }
1175
1176    /// Issue an OS-level read-ahead hint for `page_id`.
1177    ///
1178    /// A6 prefetch wire: called from `BTreeCursor::next` when the
1179    /// cursor passes 50% of the current leaf, so the kernel fetches
1180    /// the next leaf page while CPU processes the remaining half of
1181    /// the current one. Failures are silent — a missed prefetch is a
1182    /// performance miss, never a correctness bug.
1183    pub fn prefetch_hint(&self, page_id: u32) {
1184        if let Ok(file) = self.file_lock() {
1185            let _ = super::super::prefetch::prefetch_page(&file, page_id as u64, PAGE_SIZE as u32);
1186        }
1187    }
1188
1189    // ── Corruption defense helpers ──────────────────────────────────
1190
1191    /// Path for the double-write buffer file
1192    fn dwb_path(db_path: &Path) -> PathBuf {
1193        reddb_file::layout::pager_dwb_shadow_path(db_path)
1194    }
1195
1196    fn dwb_zone_offset() -> u64 {
1197        DWB_ZONE_START_PAGE as u64 * PAGE_SIZE as u64
1198    }
1199
1200    fn write_dwb_zone_frame(&self, buf: &[u8]) -> Result<(), PagerError> {
1201        if buf.len() > DWB_ZONE_BYTES {
1202            return Err(PagerError::InvalidDatabase(
1203                "DWB frame exceeds the reserved in-file zone".to_string(),
1204            ));
1205        }
1206        let mut file = self.file_lock()?;
1207        file.seek(SeekFrom::Start(Self::dwb_zone_offset()))?;
1208        file.write_all(buf)?;
1209        file.sync_all()?;
1210        Ok(())
1211    }
1212
1213    /// Clear the in-file DWB zone with valid reserved pages.
1214    fn clear_dwb_zone(&self) -> Result<(), PagerError> {
1215        let mut file = self.file_lock()?;
1216        for page_id in DWB_ZONE_START_PAGE..FIRST_ALLOCATABLE_PAGE {
1217            let mut page = Page::new(PageType::Header, page_id);
1218            page.update_checksum();
1219            file.seek(SeekFrom::Start(page_id as u64 * PAGE_SIZE as u64))?;
1220            file.write_all(page.as_bytes())?;
1221        }
1222        file.sync_all()?;
1223        Ok(())
1224    }
1225
1226    fn dwb_zone_is_clear(buf: &[u8]) -> bool {
1227        if buf.len() != DWB_ZONE_BYTES {
1228            return false;
1229        }
1230        for (idx, chunk) in buf.chunks_exact(PAGE_SIZE).enumerate() {
1231            let Ok(page) = Page::from_slice(chunk) else {
1232                return false;
1233            };
1234            if page.verify_checksum().is_err() {
1235                return false;
1236            }
1237            if page.page_id() != DWB_ZONE_START_PAGE + idx as u32 {
1238                return false;
1239            }
1240            if page.page_type().ok() != Some(PageType::Header) {
1241                return false;
1242            }
1243        }
1244        true
1245    }
1246
1247    /// Read the internal manifest page (page 1), the zone the superblock roots.
1248    ///
1249    /// ADR 0074 §2: manifest corruption fails the open didactically, naming the
1250    /// zone — it never falls back to a sidecar and never returns garbage.
1251    pub fn read_manifest_page(&self) -> Result<Page, PagerError> {
1252        self.read_page(1)
1253            .map_err(|_| PagerError::ManifestZoneCorrupt(self.path.clone()))
1254    }
1255
1256    /// Write pages through the double-write buffer for torn page protection.
1257    ///
1258    /// 1. Write pages to the in-file DWB zone with a header (magic + count + checksum)
1259    /// 2. fsync the DWB zone
1260    /// 3. Write all pages to their final locations in the .rdb file
1261    /// 4. Clear the DWB zone (marks as consumed)
1262    fn write_pages_through_dwb(&self, pages: &[(u32, Page)]) -> Result<(), PagerError> {
1263        if self.dwb_enabled {
1264            for chunk in pages.chunks(DWB_MAX_PAGES_PER_FRAME.max(1)) {
1265                let buf = reddb_file::encode_paged_dwb_frame(
1266                    chunk
1267                        .iter()
1268                        .map(|(page_id, page)| (*page_id, page.as_bytes())),
1269                );
1270
1271                // Write DWB zone and fsync
1272                self.write_dwb_zone_frame(&buf)?;
1273
1274                // Now write pages to their final locations
1275                for (page_id, page) in chunk {
1276                    self.write_page_raw(*page_id, page)?;
1277                }
1278                {
1279                    let file = self.file_lock()?;
1280                    file.sync_all()?;
1281                }
1282
1283                // Reset the zone to its inactive page images.
1284                self.clear_dwb_zone()?;
1285            }
1286
1287            Ok(())
1288        } else {
1289            // DWB disabled — write directly
1290            for (page_id, page) in pages {
1291                self.write_page_raw(*page_id, page)?;
1292            }
1293            Ok(())
1294        }
1295    }
1296
1297    /// Recover from double-write buffer after a crash.
1298    ///
1299    /// If the DWB zone contains valid pages, they were written before the crash
1300    /// interrupted writing to their home locations. Re-apply them.
1301    fn recover_from_dwb(&self) -> Result<(), PagerError> {
1302        let len = {
1303            let file = self.file_lock()?;
1304            file.metadata()?.len()
1305        };
1306        let zone_end = Self::dwb_zone_offset() + DWB_ZONE_BYTES as u64;
1307        if len < zone_end {
1308            return Ok(());
1309        }
1310
1311        let mut buf = vec![0u8; DWB_ZONE_BYTES];
1312        {
1313            let mut file = self.file_lock()?;
1314            file.seek(SeekFrom::Start(Self::dwb_zone_offset()))?;
1315            file.read_exact(&mut buf)?;
1316        }
1317        if buf.get(0..4) != Some(reddb_file::DWB_MAGIC.as_slice()) {
1318            if Self::dwb_zone_is_clear(&buf) {
1319                return Ok(());
1320            }
1321            self.clear_dwb_zone()?;
1322            return Ok(());
1323        }
1324
1325        let entries = match reddb_file::decode_paged_dwb_frame(&buf) {
1326            Ok(entries) => entries,
1327            Err(_) => {
1328                self.clear_dwb_zone()?;
1329                return Ok(());
1330            }
1331        };
1332
1333        // DWB is valid — re-apply pages to main file
1334        for entry in entries {
1335            let page = Page::from_bytes(entry.page);
1336            self.write_page_raw(entry.page_id, &page)?;
1337        }
1338
1339        // Sync and clean up
1340        {
1341            let file = self.file_lock()?;
1342            file.sync_all()?;
1343        }
1344
1345        self.clear_dwb_zone()
1346    }
1347
1348    /// Finish a consumed trunk's durable ownership change, retaining errors
1349    /// across allocation attempts until both copies have been synchronized.
1350    /// Called with the freelist locked when consuming or retrying a trunk.
1351    fn complete_allocation_fence(&self) -> Result<(), PagerError> {
1352        if self
1353            .allocation_fence_pending
1354            .load(std::sync::atomic::Ordering::Acquire)
1355        {
1356            self.write_header_and_sync()?;
1357            self.write_header_and_sync()?;
1358            self.allocation_fence_pending
1359                .store(false, std::sync::atomic::Ordering::Release);
1360        }
1361        Ok(())
1362    }
1363
1364    /// Write header and sync to disk (public for checkpointer).
1365    pub fn write_header_and_sync(&self) -> Result<(), PagerError> {
1366        self.write_header()?;
1367        let file = self.file_lock()?;
1368        file.sync_all()?;
1369        Ok(())
1370    }
1371
1372    /// Set the checkpoint_in_progress flag in the header.
1373    pub fn set_checkpoint_in_progress(
1374        &self,
1375        in_progress: bool,
1376        target_lsn: u64,
1377    ) -> Result<(), PagerError> {
1378        let mut header = self.header_write()?;
1379        header.checkpoint_in_progress = in_progress;
1380        header.checkpoint_target_lsn = target_lsn;
1381        *self.header_dirty_lock()? = true;
1382        drop(header);
1383        self.write_header_and_sync()
1384    }
1385
1386    /// Update the checkpoint LSN and clear the in-progress flag.
1387    pub fn complete_checkpoint(&self, lsn: u64) -> Result<(), PagerError> {
1388        let mut header = self.header_write()?;
1389        header.checkpoint_lsn = lsn;
1390        header.checkpoint_in_progress = false;
1391        header.checkpoint_target_lsn = 0;
1392        *self.header_dirty_lock()? = true;
1393        drop(header);
1394        self.write_header_and_sync()
1395    }
1396}
1397
1398#[cfg(test)]
1399mod tests {
1400    use super::*;
1401
1402    fn temp_db_path(name: &str) -> PathBuf {
1403        std::env::temp_dir().join(format!(
1404            "reddb-pager-{}-{}-{}.rdb",
1405            name,
1406            std::process::id(),
1407            crate::utils::now_unix_nanos()
1408        ))
1409    }
1410
1411    fn cleanup(path: &Path) {
1412        let _ = std::fs::remove_file(path);
1413        let _ = std::fs::remove_file(Pager::dwb_path(path));
1414    }
1415
1416    fn read_zone(path: &Path) -> Vec<u8> {
1417        let mut zone = vec![0u8; SUPERBLOCK_ZONE_SIZE];
1418        let mut file = File::open(path).expect("open() should succeed");
1419        file.read_exact(&mut zone)
1420            .expect("read_exact() should succeed");
1421        zone
1422    }
1423
1424    /// Overwrite one superblock slot in place, leaving the sibling untouched.
1425    fn poke_slot(path: &Path, copy_index: usize, bytes: &[u8]) {
1426        let mut file = OpenOptions::new()
1427            .write(true)
1428            .open(path)
1429            .expect("open() should succeed");
1430        file.seek(SeekFrom::Start(reddb_file::paged_superblock_slot_offset(
1431            copy_index,
1432        )))
1433        .expect("operation should succeed");
1434        file.write_all(bytes).expect("write_all() should succeed");
1435        file.sync_all().expect("sync_all() should succeed");
1436    }
1437
1438    fn stamp_pre_phase3_layout(path: &Path, page_count: u32) {
1439        let zone = read_zone(path);
1440        for copy_index in 0..reddb_file::PAGED_SUPERBLOCK_SLOT_COUNT {
1441            let start = copy_index * SUPERBLOCK_SLOT_SIZE;
1442            let mut slot = zone[start..start + SUPERBLOCK_SLOT_SIZE].to_vec();
1443            let generation = reddb_file::paged_superblock_slot_generation(&slot, copy_index)
1444                .expect("fresh store should publish both slots");
1445            reddb_file::set_database_header_version(&mut slot, PHASE3_DWB_ZONE_FILE_VERSION - 1)
1446                .expect("set_database_header_version() should succeed");
1447            reddb_file::set_database_header_page_count(&mut slot, page_count)
1448                .expect("set_database_header_page_count() should succeed");
1449            reddb_file::seal_paged_superblock_slot(&mut slot, copy_index, generation)
1450                .expect("seal_paged_superblock_slot() should succeed");
1451            poke_slot(path, copy_index, &slot);
1452        }
1453    }
1454
1455    #[test]
1456    fn open_refuses_future_database_version() {
1457        let path = temp_db_path("future-version");
1458        let pager = Pager::open_default(&path).expect("open_default() should succeed");
1459        drop(pager);
1460
1461        // A *valid* superblock slot — correct magic, copy index and CRC — whose
1462        // database header names a version we cannot read. The slot is intact;
1463        // the header is from the future.
1464        let selection = reddb_file::select_paged_superblock(&read_zone(&path))
1465            .expect("operation should succeed");
1466        let mut slot = vec![0u8; SUPERBLOCK_SLOT_SIZE];
1467        reddb_file::init_database_header_page(&mut slot, 1)
1468            .expect("init_database_header_page() should succeed");
1469        reddb_file::set_database_header_version(&mut slot, reddb_file::PAGE_FILE_VERSION + 1)
1470            .expect("operation should succeed");
1471        reddb_file::seal_paged_superblock_slot(
1472            &mut slot,
1473            selection.copy_index,
1474            selection.generation + 1,
1475        )
1476        .expect("operation should succeed");
1477        poke_slot(&path, selection.copy_index, &slot);
1478
1479        let err = match Pager::open_default(&path) {
1480            Ok(_) => panic!("future database version should be rejected"),
1481            Err(err) => err,
1482        };
1483        match err {
1484            PagerError::InvalidDatabase(msg) => {
1485                assert!(msg.contains("newer than supported"), "{msg}");
1486            }
1487            other => panic!("expected InvalidDatabase, got {other:?}"),
1488        }
1489
1490        cleanup(&path);
1491    }
1492
1493    #[test]
1494    fn a_fresh_store_publishes_both_superblock_copies() {
1495        // The ping-pong invariant must hold from the very first update, which
1496        // means creation cannot leave one copy blank.
1497        let path = temp_db_path("fresh-pair");
1498        let pager = Pager::open_default(&path).expect("open_default() should succeed");
1499        drop(pager);
1500
1501        let zone = read_zone(&path);
1502        for copy_index in 0..reddb_file::PAGED_SUPERBLOCK_SLOT_COUNT {
1503            let start = copy_index * SUPERBLOCK_SLOT_SIZE;
1504            assert!(
1505                reddb_file::paged_superblock_slot_generation(
1506                    &zone[start..start + SUPERBLOCK_SLOT_SIZE],
1507                    copy_index
1508                )
1509                .is_some(),
1510                "copy {copy_index} must be valid on a fresh store"
1511            );
1512        }
1513        cleanup(&path);
1514    }
1515
1516    #[test]
1517    fn an_update_writes_the_stale_copy_and_leaves_the_newest_one_intact() {
1518        let path = temp_db_path("ping-pong");
1519        let pager = Pager::open_default(&path).expect("open_default() should succeed");
1520        let before = reddb_file::select_paged_superblock(&read_zone(&path))
1521            .expect("operation should succeed");
1522
1523        pager
1524            .allocate_page(PageType::BTreeLeaf)
1525            .expect("allocate_page() should succeed");
1526        pager.sync().expect("sync() should succeed");
1527
1528        let zone = read_zone(&path);
1529        let after = reddb_file::select_paged_superblock(&zone)
1530            .expect("select_paged_superblock() should succeed");
1531        assert_ne!(
1532            after.copy_index, before.copy_index,
1533            "an update must target the stale copy"
1534        );
1535        assert!(after.generation > before.generation);
1536
1537        // The copy that was newest before the update is still readable — this
1538        // is the byte-level statement of "a crash mid-write always leaves one
1539        // valid copy".
1540        let start = before.copy_index * SUPERBLOCK_SLOT_SIZE;
1541        assert_eq!(
1542            reddb_file::paged_superblock_slot_generation(
1543                &zone[start..start + SUPERBLOCK_SLOT_SIZE],
1544                before.copy_index
1545            ),
1546            Some(before.generation)
1547        );
1548
1549        drop(pager);
1550        cleanup(&path);
1551    }
1552
1553    #[test]
1554    fn failed_trunk_fence_is_completed_before_retrying_allocation() {
1555        let path = temp_db_path("retry-trunk-fence");
1556        let pager = Pager::open_default(&path).expect("pager");
1557        let mut pages = Vec::new();
1558        for _ in 0..8 {
1559            pages.push(pager.allocate_page(PageType::BTreeLeaf).unwrap().page_id());
1560        }
1561        pager.sync().unwrap();
1562        for id in pages {
1563            pager.free_page(id).unwrap();
1564        }
1565        pager.sync().unwrap();
1566        let trunk = pager.header().unwrap().freelist_head;
1567        assert_ne!(trunk, 0);
1568
1569        // A real write error after the trunk has been read must not turn a
1570        // subsequent allocation into an unfenced fast-path reuse.
1571        let writable = {
1572            let mut file = pager.file_lock().unwrap();
1573            std::mem::replace(&mut *file, File::open(&path).unwrap())
1574        };
1575        assert!(pager.allocate_page(PageType::BTreeLeaf).is_err());
1576        *pager.file_lock().unwrap() = writable;
1577        pager.allocate_page(PageType::BTreeLeaf).unwrap();
1578
1579        let zone = read_zone(&path);
1580        for copy in 0..reddb_file::PAGED_SUPERBLOCK_SLOT_COUNT {
1581            let start = copy * SUPERBLOCK_SLOT_SIZE;
1582            let header =
1583                reddb_file::decode_database_header(&zone[start..start + SUPERBLOCK_SLOT_SIZE])
1584                    .unwrap();
1585            assert_ne!(
1586                header.freelist_head, trunk,
1587                "copy {copy} still owns consumed trunk"
1588            );
1589        }
1590        drop(pager);
1591        cleanup(&path);
1592    }
1593
1594    #[test]
1595    fn sibling_superblock_recovery_does_not_reallocate_live_reused_pages() {
1596        let path = temp_db_path("reused-allocator-pair");
1597        let pager = Pager::open_default(&path).expect("pager");
1598        let mut pages = Vec::new();
1599        for _ in 0..32 {
1600            pages.push(
1601                pager
1602                    .allocate_page(PageType::BTreeLeaf)
1603                    .expect("allocate")
1604                    .page_id(),
1605            );
1606        }
1607        pager.sync().expect("seed");
1608        for id in pages {
1609            pager.free_page(id).expect("free");
1610        }
1611        pager.sync().expect("publish free list");
1612        let mut live = std::collections::HashSet::new();
1613        for _ in 0..8 {
1614            let page = pager.allocate_page(PageType::BTreeLeaf).expect("reuse");
1615            live.insert(page.page_id());
1616            pager
1617                .write_page(page.page_id(), page)
1618                .expect("write valid page");
1619        }
1620        pager.sync().expect("publish reused ownership");
1621        drop(pager);
1622        let selection = reddb_file::select_paged_superblock(&read_zone(&path)).expect("superblock");
1623        poke_slot(&path, selection.copy_index, &[0; SUPERBLOCK_SLOT_SIZE]);
1624        let recovered = Pager::open_default(&path).expect("sibling recovery");
1625        for _ in 0..64 {
1626            let id = recovered
1627                .allocate_page(PageType::BTreeLeaf)
1628                .expect("allocate after recovery")
1629                .page_id();
1630            assert!(
1631                !live.contains(&id),
1632                "sibling allocator still considers live page {id} free"
1633            );
1634        }
1635        drop(recovered);
1636        cleanup(&path);
1637    }
1638
1639    #[test]
1640    fn open_recovers_from_the_older_copy_when_the_newest_is_torn() {
1641        let path = temp_db_path("torn-newest");
1642        {
1643            let pager = Pager::open_default(&path).expect("open_default() should succeed");
1644            pager
1645                .allocate_page(PageType::BTreeLeaf)
1646                .expect("allocate_page() should succeed");
1647            pager.sync().expect("sync() should succeed");
1648        }
1649
1650        let newest = reddb_file::select_paged_superblock(&read_zone(&path))
1651            .expect("operation should succeed");
1652        // Tear the newest copy: a torn write leaves its CRC broken.
1653        let mut torn = vec![0u8; SUPERBLOCK_SLOT_SIZE];
1654        torn.copy_from_slice(
1655            &read_zone(&path)[newest.copy_index * SUPERBLOCK_SLOT_SIZE
1656                ..newest.copy_index * SUPERBLOCK_SLOT_SIZE + SUPERBLOCK_SLOT_SIZE],
1657        );
1658        torn[64] ^= 0xFF;
1659        poke_slot(&path, newest.copy_index, &torn);
1660
1661        let pager = Pager::open_default(&path).expect("older copy must root the store");
1662        assert!(pager.page_count().expect("page_count() should succeed") >= 3);
1663        drop(pager);
1664        cleanup(&path);
1665    }
1666
1667    #[test]
1668    fn open_refuses_didactically_when_both_superblock_copies_are_invalid() {
1669        let path = temp_db_path("both-invalid");
1670        {
1671            let _ = Pager::open_default(&path).expect("open_default() should succeed");
1672        }
1673
1674        let zone = read_zone(&path);
1675        for copy_index in 0..reddb_file::PAGED_SUPERBLOCK_SLOT_COUNT {
1676            let start = copy_index * SUPERBLOCK_SLOT_SIZE;
1677            let mut slot = zone[start..start + SUPERBLOCK_SLOT_SIZE].to_vec();
1678            slot[64] ^= 0xFF;
1679            poke_slot(&path, copy_index, &slot);
1680        }
1681
1682        let err = match Pager::open_default(&path) {
1683            Ok(_) => panic!("a rootless store must not open"),
1684            Err(err) => err,
1685        };
1686        let rendered = err.to_string();
1687        assert!(matches!(err, PagerError::SuperblockZoneUnrecoverable(_)));
1688        assert!(rendered.contains("superblock zone"), "{rendered}");
1689        assert!(rendered.contains("salvage"), "{rendered}");
1690
1691        cleanup(&path);
1692    }
1693
1694    #[test]
1695    fn open_refuses_a_legacy_sidecar_store_and_names_the_migration_tool() {
1696        let path = temp_db_path("legacy-sidecar");
1697        {
1698            let _ = Pager::open_default(&path).expect("open_default() should succeed");
1699        }
1700
1701        let sidecar = reddb_file::layout::retired::pager_header_path_v0(&path);
1702        std::fs::write(&sidecar, b"legacy header shadow").expect("write() should succeed");
1703
1704        let err = match Pager::open_default(&path) {
1705            Ok(_) => panic!("legacy stores need migration first"),
1706            Err(err) => err,
1707        };
1708        let rendered = err.to_string();
1709        assert!(matches!(err, PagerError::LegacySidecarStore { .. }));
1710        assert!(rendered.contains("migration tool"), "{rendered}");
1711
1712        let _ = std::fs::remove_file(&sidecar);
1713        cleanup(&path);
1714    }
1715
1716    #[test]
1717    fn pre_phase3_large_store_without_sidecar_routes_to_migration() {
1718        let path = temp_db_path("pre-phase3-large");
1719        {
1720            let _ = Pager::open_default(&path).expect("open_default() should succeed");
1721        }
1722        stamp_pre_phase3_layout(&path, FIRST_ALLOCATABLE_PAGE);
1723
1724        let err = match Pager::open_default(&path) {
1725            Ok(_) => panic!("pre-phase-3 stores need migration first"),
1726            Err(err) => err,
1727        };
1728        let rendered = err.to_string();
1729        assert!(matches!(err, PagerError::LegacyPagedStore { .. }));
1730        assert!(rendered.contains("migration tool"), "{rendered}");
1731
1732        cleanup(&path);
1733    }
1734
1735    #[test]
1736    fn pre_phase3_small_store_without_sidecar_routes_to_migration() {
1737        let path = temp_db_path("pre-phase3-small");
1738        {
1739            let _ = Pager::open_default(&path).expect("open_default() should succeed");
1740        }
1741        stamp_pre_phase3_layout(&path, 3);
1742        OpenOptions::new()
1743            .write(true)
1744            .open(&path)
1745            .expect("open() should succeed")
1746            .set_len(3 * PAGE_SIZE as u64)
1747            .expect("set_len() should succeed");
1748
1749        let err = match Pager::open_default(&path) {
1750            Ok(_) => panic!("pre-phase-3 stores need migration first"),
1751            Err(err) => err,
1752        };
1753        assert!(matches!(err, PagerError::LegacyPagedStore { .. }));
1754
1755        cleanup(&path);
1756    }
1757
1758    #[test]
1759    fn no_phase_one_sidecar_survives_a_create_write_reopen_cycle() {
1760        let path = temp_db_path("census");
1761        {
1762            let pager = Pager::open_default(&path).expect("open_default() should succeed");
1763            let page = pager
1764                .allocate_page(PageType::BTreeLeaf)
1765                .expect("allocate_page() should succeed");
1766            pager
1767                .write_page(page.page_id(), page)
1768                .expect("operation should succeed");
1769            pager.sync().expect("sync() should succeed");
1770        }
1771        {
1772            let pager = Pager::open_default(&path).expect("open_default() should succeed");
1773            pager.sync().expect("sync() should succeed");
1774        }
1775
1776        assert_eq!(
1777            reddb_file::layout::retired::first_present_phase1_sidecar(&path),
1778            None,
1779            "the pager must never create a retired phase-1 sidecar"
1780        );
1781        cleanup(&path);
1782    }
1783}