Skip to main content

reddb_engine/storage/
pager.rs

1//! Shared pager orchestration over an exclusively owned VFS file.
2//!
3//! Native and browser use the same cache, allocation fences, freelist, DWB,
4//! superblock publication and checkpoint algorithms. Opening/locking files,
5//! scheduling, WAL implementation and encryption remain host policy.
6//! The owner explicitly calls `sync` before teardown; this low-level pager
7//! does not hide I/O failures in a destructor. It is not a transaction API.
8use super::freelist::FreeList;
9use super::page::{Page, PageError, PageType, PAGE_SIZE};
10use super::page_cache::PageCache;
11use super::pager_io::{self, FIRST_ALLOCATABLE_PAGE};
12use reddb_file::VfsFile;
13pub use reddb_file::{DatabaseHeader, PhysicalFileHeader};
14use std::path::{Path, PathBuf};
15use std::sync::{Mutex, RwLock};
16
17mod batch;
18
19/// Pager error types
20#[derive(Debug)]
21pub enum PagerError {
22    /// I/O error
23    Io(std::io::Error),
24    /// Page error
25    Page(PageError),
26    /// Invalid database file
27    InvalidDatabase(String),
28    /// Database is read-only
29    ReadOnly,
30    /// Page not found
31    PageNotFound(u32),
32    /// Database is locked
33    Locked,
34    /// A Mutex or RwLock was poisoned (another thread panicked while holding it)
35    LockPoisoned,
36    /// Database is encrypted but no key was supplied.
37    EncryptionRequired,
38    /// Plain (unencrypted) database opened with an encryption key.
39    PlainDatabaseRefusesKey,
40    /// Encryption key validation failed for an encrypted database.
41    InvalidKey,
42    /// Both superblock copies failed validation (ADR 0074 §2).
43    SuperblockZoneUnrecoverable(PathBuf),
44    /// The internal manifest zone failed its checksum (ADR 0074 §2).
45    ManifestZoneCorrupt(PathBuf),
46    /// The in-file double-write zone failed validation (ADR 0074 §2).
47    DwbZoneCorrupt(PathBuf),
48    /// The store still carries a retired phase-1 pager sidecar (ADR 0038 §4).
49    LegacySidecarStore { sidecar: PathBuf },
50    /// The store predates the phase-3 in-file DWB layout marker (ADR 0038 §4).
51    LegacyPagedStore { path: PathBuf },
52}
53
54/// A contiguous run of database pages reserved for vector-turbo payloads.
55#[derive(Debug, Clone, Copy, PartialEq, Eq)]
56pub struct ExtentId {
57    pub start_page: u32,
58    pub n_pages: u32,
59}
60
61impl std::fmt::Display for PagerError {
62    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
63        match self {
64            Self::Io(e) => write!(f, "I/O error: {}", e),
65            Self::Page(e) => write!(f, "Page error: {}", e),
66            Self::InvalidDatabase(msg) => write!(f, "Invalid database: {}", msg),
67            Self::ReadOnly => write!(f, "Database is read-only"),
68            Self::PageNotFound(id) => write!(f, "Page {} not found", id),
69            Self::Locked => write!(f, "Database is locked"),
70            Self::LockPoisoned => write!(f, "Internal lock poisoned (concurrent thread panicked)"),
71            Self::EncryptionRequired => write!(
72                f,
73                "Database is encrypted but no key was supplied (set PagerConfig::encryption)"
74            ),
75            Self::PlainDatabaseRefusesKey => write!(
76                f,
77                "Plain (unencrypted) database opened with an encryption key — refusing"
78            ),
79            Self::InvalidKey => write!(f, "Encryption key validation failed for this database"),
80            Self::SuperblockZoneUnrecoverable(path) => write!(
81                f,
82                "superblock zone of {} is unrecoverable: both ping-pong copies failed \
83                 validation, so no generation can be trusted to root the store. The store \
84                 will not be opened. Preserve the database and its recovery files; \
85                 restore from a verified backup.",
86                path.display()
87            ),
88            Self::ManifestZoneCorrupt(path) => write!(
89                f,
90                "internal manifest zone of {} failed its checksum: the zone that names \
91                 collections, indexes and the checkpoint boundary cannot be trusted, so no \
92                 rows are returned. Preserve the database and its recovery files; \
93                 restore from a verified backup.",
94                path.display()
95            ),
96            Self::DwbZoneCorrupt(path) => write!(
97                f,
98                "double-write zone of {} failed validation: page recovery cannot prove \
99                 whether the protected in-place write completed. The store will not be opened \
100                 silently. Preserve the database and its recovery files; restore from \
101                 a verified backup.",
102                path.display()
103            ),
104            Self::LegacySidecarStore { sidecar } => write!(
105                f,
106                "refusing to open a legacy sidecar-backed store: found {}. The superblock \
107                 pair, internal manifest, and double-write recovery state now live inside the \
108                 .rdb file (ADR 0038 §2/§4), \
109                 and the engine never reads the retired sidecars silently. Convert the store \
110                 first with the offline migration tool \
111                 (`reddb_server::pager_zone_migration::migrate_to_zoned`); it is reversible.",
112                sidecar.display()
113            ),
114            Self::LegacyPagedStore { path } => write!(
115                f,
116                "refusing to open a pre-phase-3 paged store: {} does not carry the layout marker \
117                 for the in-file double-write zone. Convert the store first with the offline \
118                 migration tool (`reddb_server::pager_zone_migration::migrate_to_zoned`); it is \
119                 reversible.",
120                path.display()
121            ),
122        }
123    }
124}
125
126impl std::error::Error for PagerError {
127    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
128        match self {
129            Self::Io(e) => Some(e),
130            Self::Page(e) => Some(e),
131            _ => None,
132        }
133    }
134}
135
136impl From<std::io::Error> for PagerError {
137    fn from(e: std::io::Error) -> Self {
138        Self::Io(e)
139    }
140}
141
142impl From<PageError> for PagerError {
143    fn from(e: PageError) -> Self {
144        Self::Page(e)
145    }
146}
147
148impl From<pager_io::PagerIoError> for PagerError {
149    fn from(error: pager_io::PagerIoError) -> Self {
150        match error {
151            pager_io::PagerIoError::Io(error) => Self::Io(error),
152            pager_io::PagerIoError::Page(error) => Self::Page(error),
153            pager_io::PagerIoError::InvalidDatabase(message) => Self::InvalidDatabase(message),
154        }
155    }
156}
157
158/// Host-independent pager configuration. File creation and ownership are host policy.
159#[derive(Debug, Clone)]
160pub struct PagerConfig {
161    /// Host guarantees immutable published trees and generation-bound resume counters.
162    pub disk_residency: bool,
163    pub cache_size: usize,
164    pub read_only: bool,
165    pub verify_checksums: bool,
166    pub double_write: bool,
167}
168
169impl Default for PagerConfig {
170    fn default() -> Self {
171        Self {
172            disk_residency: false,
173            cache_size: 10_000,
174            read_only: false,
175            verify_checksums: true,
176            double_write: true,
177        }
178    }
179}
180
181/// The host's WAL durability barrier, called before publishing dirty page batches.
182/// Implementations must make all records through `lsn` durable or return an error.
183pub trait WalBarrier {
184    fn flush_until(&self, lsn: u64) -> std::io::Result<()>;
185
186    /// A cold page must not claim changes beyond the host's durable WAL.
187    /// LSN zero and the completed physical checkpoint need no live WAL handle.
188    fn validate_page_lsn(&self, lsn: u64, checkpoint_lsn: u64) -> std::io::Result<()> {
189        if lsn <= checkpoint_lsn {
190            return Ok(());
191        }
192        Err(std::io::Error::new(
193            std::io::ErrorKind::InvalidData,
194            "page LSN has no durable WAL authority",
195        ))
196    }
197
198    /// A checkpoint host can preserve the first image of an existing page
199    /// before the cache accepts an overwrite. Newly reserved pages need no undo.
200    fn needs_page_image(&self, _page_id: u32) -> std::io::Result<bool> {
201        Ok(false)
202    }
203
204    /// Must make the image durable before returning. The pager may evict the
205    /// replacement immediately; recording only at flush time is too late.
206    fn capture_page_image(&self, _page: &Page) -> std::io::Result<()> {
207        Ok(())
208    }
209
210    /// Called before pages are removed from allocator ownership.
211    fn reserve_pages(&self, _start: u32, _count: u32) -> std::io::Result<()> {
212        Ok(())
213    }
214
215    /// Called before a persisted free-list trunk is detached from its head.
216    fn detach_free_pages(&self, _pages: &[u32]) -> std::io::Result<()> {
217        Ok(())
218    }
219
220    /// True only while detached free IDs have durable recovery ownership.
221    /// No-op callbacks must not authorize moving a whole trunk into RAM.
222    fn has_page_ownership_journal(&self) -> std::io::Result<bool> {
223        Ok(false)
224    }
225
226    /// Called before retired pages enter the free list.
227    fn retire_pages(&self, _pages: &[u32]) -> std::io::Result<()> {
228        Ok(())
229    }
230}
231
232/// A pager without a WAL may only publish pages whose LSN is zero.
233pub struct NoWal;
234impl WalBarrier for NoWal {
235    fn flush_until(&self, _lsn: u64) -> std::io::Result<()> {
236        Err(std::io::Error::other(
237            "nonzero page LSN requires a WAL durability barrier",
238        ))
239    }
240}
241pub struct Pager<F: VfsFile, W: WalBarrier> {
242    /// Database file path
243    path: PathBuf,
244    /// File handle
245    file: Mutex<F>,
246    /// Whether mutable pages are protected by the in-file DWB zone.
247    dwb_enabled: bool,
248    staged_pages: Mutex<Option<pager_io::StagedPageBatch>>,
249    /// A failed writeback may leave the only complete image in the DWB.
250    /// Refuse further writes until reopen has recovered that image.
251    writeback_failed: std::sync::atomic::AtomicBool,
252    /// Page cache
253    cache: PageCache,
254    /// Recovery images must survive cache eviction without writing the source.
255    read_only_recovery: RwLock<std::collections::HashMap<u32, Page>>,
256    /// Free page list
257    freelist: RwLock<FreeList>,
258    /// Store-managed frees stay unavailable until their removal from the
259    /// durable catalog has crossed a publication boundary.
260    deferred_reclamation: Mutex<DeferredPageReclamation>,
261    /// Excludes page ownership changes while a host publishes an epoch.
262    /// Reentrant so the publishing thread can use ordinary pager methods.
263    ownership_mutations: parking_lot::ReentrantMutex<()>,
264    /// Reused page ownership must reach both superblock copies before a
265    /// manifest can refer to those pages, including single-slot recovery.
266    allocation_reused: std::sync::atomic::AtomicBool,
267    /// A consumed trunk cannot be reused until both header fences succeed.
268    allocation_fence_pending: std::sync::atomic::AtomicBool,
269    /// Database header
270    header: RwLock<DatabaseHeader>,
271    /// Serialize read-select-write sequences for the superblock pair.
272    superblock_publication: Mutex<()>,
273    /// Configuration
274    config: PagerConfig,
275    /// Dirty flag for header
276    header_dirty: Mutex<bool>,
277    wal: W,
278}
279
280#[derive(Default)]
281struct DeferredPageReclamation {
282    enabled: bool,
283    pending: std::collections::BTreeSet<u32>,
284    published: std::collections::BTreeSet<u32>,
285}
286
287/// The superblock zone occupies the head of page 0 (ADR 0038 §2 phase 1).
288const SUPERBLOCK_SLOT_SIZE: usize = reddb_file::PAGED_SUPERBLOCK_SLOT_SIZE;
289const SUPERBLOCK_ZONE_SIZE: usize = reddb_file::PAGED_SUPERBLOCK_ZONE_SIZE;
290// The phase-3 layout was introduced in 1.0.1. Later file format versions
291// must not make existing phase-3 stores look like legacy sidecar stores.
292const PHASE3_DWB_ZONE_FILE_VERSION: u32 = 0x0001_0001;
293
294/// The newest valid superblock copy together with its slot image.
295pub type NewestSuperblock = (
296    reddb_file::PagedSuperblockSelection,
297    [u8; SUPERBLOCK_SLOT_SIZE],
298);
299
300/// Take the first slot's worth of a freshly-built page-0 image. The database
301/// header and the encryption header both live well inside this prefix.
302fn superblock_slot_image(page: &[u8; PAGE_SIZE]) -> [u8; SUPERBLOCK_SLOT_SIZE] {
303    let mut image = [0u8; SUPERBLOCK_SLOT_SIZE];
304    image.copy_from_slice(&page[..SUPERBLOCK_SLOT_SIZE]);
305    image
306}
307
308impl<F: VfsFile, W: WalBarrier> Pager<F, W> {
309    /// Use an already-owned file. The host retains exclusive ownership until
310    /// this pager is dropped, and supplies the native WAL/encryption policy.
311    /// `create` is explicit: a corrupt or empty existing file is never recreated.
312    pub fn open_owned(
313        file: F,
314        path: PathBuf,
315        config: PagerConfig,
316        wal: W,
317        create: bool,
318    ) -> Result<Self, PagerError> {
319        if create && file.file_len()? != 0 {
320            return Err(PagerError::InvalidDatabase(
321                "new pager file is not empty".into(),
322            ));
323        }
324        let pager = Self {
325            path,
326            file: Mutex::new(file),
327            dwb_enabled: config.double_write && !config.read_only,
328            staged_pages: Mutex::new(None),
329            writeback_failed: std::sync::atomic::AtomicBool::new(false),
330            cache: PageCache::new(config.cache_size),
331            read_only_recovery: RwLock::new(std::collections::HashMap::new()),
332            freelist: RwLock::new(FreeList::new()),
333            deferred_reclamation: Mutex::new(DeferredPageReclamation::default()),
334            ownership_mutations: parking_lot::ReentrantMutex::new(()),
335            allocation_reused: std::sync::atomic::AtomicBool::new(false),
336            allocation_fence_pending: std::sync::atomic::AtomicBool::new(false),
337            header: RwLock::new(DatabaseHeader::default()),
338            superblock_publication: Mutex::new(()),
339            config,
340            header_dirty: Mutex::new(false),
341            wal,
342        };
343        if create {
344            pager.initialize()?;
345        } else {
346            pager.ensure_phase3_dwb_zone_layout()?;
347            pager.recover_from_dwb()?;
348            pager.load_header()?;
349        }
350        Ok(pager)
351    }
352
353    pub fn wal_barrier(&self) -> &W {
354        &self.wal
355    }
356
357    /// Hold across an entire ownership publication, not just individual
358    /// allocation calls. Readers may continue accessing already-published pages.
359    pub fn lock_ownership_mutations(&self) -> parking_lot::ReentrantMutexGuard<'_, ()> {
360        self.ownership_mutations.lock()
361    }
362
363    /// An interrupted publication must be recovered before any more writes.
364    pub fn fence_interrupted_ownership(&self) {
365        self.writeback_failed
366            .store(true, std::sync::atomic::Ordering::Release);
367    }
368
369    /// Run a host operation while holding the same lock as protected page I/O.
370    /// This is used only for native prefetch and fault-injection tests.
371    #[doc(hidden)]
372    pub fn with_file<T>(&self, operation: impl FnOnce(&mut F) -> T) -> Result<T, PagerError> {
373        Ok(operation(&mut *self.file_lock()?))
374    }
375
376    /// Initialize a new database
377    fn initialize(&self) -> Result<(), PagerError> {
378        if self.config.read_only {
379            return Err(PagerError::ReadOnly);
380        }
381
382        // Create header page. Page ids 1 and 2 are reserved for fixed
383        // metadata/vault pages. ADR 0038 phase 3 reserves the following DWB
384        // zone in-file so normal B-tree allocation starts after it.
385        let initial_page_count = FIRST_ALLOCATABLE_PAGE;
386        let header_page = Page::new_header_page(initial_page_count);
387        self.header_write()?.page_count = initial_page_count;
388
389        // Reserve page 0 (the superblock zone) plus the reserved pages so any
390        // scan over 0..page_count can read every allocated page in a brand-new
391        // database. The zone's own bytes are published below, one slot at a
392        // time; this first whole-page write only sizes the file.
393        self.write_page_raw(0, &header_page)?;
394        let mut metadata_page = Page::new(PageType::Header, 1);
395        metadata_page.update_checksum();
396        self.write_page_raw(1, &metadata_page)?;
397        let mut vault_page = Page::new(PageType::Vault, 2);
398        vault_page.update_checksum();
399        self.write_page_raw(2, &vault_page)?;
400        self.clear_dwb_zone()?;
401
402        // Publish both superblock copies so the ping-pong invariant — one
403        // valid copy always survives a crash — holds from the very first
404        // update onward, not just from the second.
405        let mut slot = superblock_slot_image(header_page.as_bytes());
406        self.write_superblock_slot(&mut slot, 0, 1)?;
407        self.write_superblock_slot(&mut slot, 1, 2)?;
408
409        // Sync to disk
410        self.sync()?;
411
412        Ok(())
413    }
414
415    /// Acquire a write lock on the header RwLock, mapping poison errors.
416    fn header_write(&self) -> Result<std::sync::RwLockWriteGuard<'_, DatabaseHeader>, PagerError> {
417        self.header.write().map_err(|_| PagerError::LockPoisoned)
418    }
419
420    /// Acquire a read lock on the header RwLock, mapping poison errors.
421    fn header_read(&self) -> Result<std::sync::RwLockReadGuard<'_, DatabaseHeader>, PagerError> {
422        self.header.read().map_err(|_| PagerError::LockPoisoned)
423    }
424
425    /// Acquire a write lock on the freelist RwLock, mapping poison errors.
426    fn freelist_write(&self) -> Result<std::sync::RwLockWriteGuard<'_, FreeList>, PagerError> {
427        self.freelist.write().map_err(|_| PagerError::LockPoisoned)
428    }
429
430    /// Acquire a lock on the file Mutex, mapping poison errors.
431    fn file_lock(&self) -> Result<std::sync::MutexGuard<'_, F>, PagerError> {
432        self.file.lock().map_err(|_| PagerError::LockPoisoned)
433    }
434
435    /// Acquire a lock on the header_dirty Mutex, mapping poison errors.
436    fn header_dirty_lock(&self) -> Result<std::sync::MutexGuard<'_, bool>, PagerError> {
437        self.header_dirty
438            .lock()
439            .map_err(|_| PagerError::LockPoisoned)
440    }
441
442    /// Read the whole superblock zone (both ping-pong copies) from the head of
443    /// the file. A file shorter than the zone yields zero-filled tail bytes,
444    /// which decode as "slot never written".
445    fn read_superblock_zone(&self) -> Result<[u8; SUPERBLOCK_ZONE_SIZE], PagerError> {
446        let mut file = self.file_lock()?;
447        Ok(pager_io::read_superblock_zone(&mut *file)?)
448    }
449
450    /// The newest valid superblock copy, with its slot image.
451    ///
452    /// Both copies invalid is the one unrecoverable superblock state (ADR 0074
453    /// §2): the store does not open, and the error names the zone rather
454    /// than panicking or silently re-initialising.
455    #[doc(hidden)]
456    pub fn newest_superblock(&self) -> Result<NewestSuperblock, PagerError> {
457        let zone = self.read_superblock_zone()?;
458        let selection = reddb_file::select_paged_superblock(&zone)
459            .ok_or_else(|| PagerError::SuperblockZoneUnrecoverable(self.path.clone()))?;
460
461        let start = selection.copy_index * SUPERBLOCK_SLOT_SIZE;
462        let mut image = [0u8; SUPERBLOCK_SLOT_SIZE];
463        image.copy_from_slice(&zone[start..start + SUPERBLOCK_SLOT_SIZE]);
464        Ok((selection, image))
465    }
466
467    fn ensure_phase3_dwb_zone_layout(&self) -> Result<(), PagerError> {
468        let (_, image) = self.newest_superblock()?;
469        let header = reddb_file::decode_published_database_header(&image)
470            .map_err(|err| PagerError::InvalidDatabase(err.to_string()))?;
471        if header.is_recovery_read_only() && !self.config.read_only {
472            return Err(PagerError::InvalidDatabase(
473                "recovered database is read-only; export verified data into a new store".into(),
474            ));
475        }
476        if header.requires_disk_residency() && !self.config.disk_residency && !self.config.read_only
477        {
478            return Err(PagerError::InvalidDatabase(
479                "prepared disk-resident database requires a disk-resident writer".into(),
480            ));
481        }
482        if header.version < PHASE3_DWB_ZONE_FILE_VERSION {
483            return Err(PagerError::LegacyPagedStore {
484                path: self.path.clone(),
485            });
486        }
487        Ok(())
488    }
489
490    /// Seal `slot` for `copy_index`/`generation` and write exactly that slot,
491    /// then fsync. The caller guarantees `copy_index` is the stale copy.
492    fn write_superblock_slot(
493        &self,
494        slot: &mut [u8; SUPERBLOCK_SLOT_SIZE],
495        copy_index: usize,
496        generation: u64,
497    ) -> Result<(), PagerError> {
498        self.ensure_writeback_available()?;
499        if self.config.read_only {
500            return Err(PagerError::ReadOnly);
501        }
502        let mut file = self.file_lock()?;
503        self.ensure_writeback_available()?;
504        Ok(pager_io::write_superblock_slot(
505            &mut *file, slot, copy_index, generation,
506        )?)
507    }
508
509    /// Publish `mutate`'s edit of the newest slot image into the *stale* copy.
510    ///
511    /// Reading the newest image first means every byte the pager does not
512    /// explicitly rewrite — most importantly the encryption marker and header —
513    /// carries forward untouched.
514    #[doc(hidden)]
515    pub fn commit_superblock(
516        &self,
517        mutate: impl FnOnce(&mut [u8; SUPERBLOCK_SLOT_SIZE]) -> Result<(), PagerError>,
518    ) -> Result<(), PagerError> {
519        let _ownership = self.lock_ownership_mutations();
520        let _publication = self
521            .superblock_publication
522            .lock()
523            .map_err(|_| PagerError::LockPoisoned)?;
524        self.commit_superblock_locked(mutate)
525    }
526
527    fn commit_superblock_locked(
528        &self,
529        mutate: impl FnOnce(&mut [u8; SUPERBLOCK_SLOT_SIZE]) -> Result<(), PagerError>,
530    ) -> Result<(), PagerError> {
531        if self.config.read_only {
532            return Err(PagerError::ReadOnly);
533        }
534        let (newest, mut image) = self.newest_superblock()?;
535        mutate(&mut image)?;
536        let next_copy = reddb_file::paged_superblock_next_copy(Some(newest));
537        self.write_superblock_slot(&mut image, next_copy, newest.generation.saturating_add(1))
538    }
539
540    /// Fence a file-format upgrade in both recoverable superblock copies.
541    ///
542    /// The caller must exclude other mutators while the fence is in progress
543    /// and may publish a new mandatory format only after this succeeds. A
544    /// failed write fences subsequent mutation until reopen and recovery.
545    pub fn fence_file_format_version(&self, version: u32) -> Result<(), PagerError> {
546        let _ownership = self.lock_ownership_mutations();
547        self.ensure_writeback_available()?;
548        if self.config.read_only {
549            return Err(PagerError::ReadOnly);
550        }
551        if version == reddb_file::file_format::PAGE_RECOVERY_READ_ONLY_FILE_VERSION {
552            return Err(PagerError::InvalidDatabase(
553                "recovery read-only format is reserved".into(),
554            ));
555        }
556        if !(PHASE3_DWB_ZONE_FILE_VERSION..=reddb_file::PAGE_FILE_VERSION).contains(&version) {
557            return Err(PagerError::InvalidDatabase(
558                "unsupported file format fence".into(),
559            ));
560        }
561        let current = self.header_read()?.clone();
562        if version < current.version {
563            return Err(PagerError::InvalidDatabase(
564                "file format downgrade is not supported".into(),
565            ));
566        }
567        let legacy_disk_residency =
568            reddb_file::file_format::page_format_is_disk_resident(current.version);
569        let required_capabilities = current.required_capabilities
570            | if version >= reddb_file::PAGE_REQUIRED_CAPABILITIES_FILE_VERSION
571                && legacy_disk_residency
572            {
573                reddb_file::FILE_CAPABILITY_DISK_RESIDENCY
574            } else {
575                0
576            };
577        let requires_disk_residency =
578            if version < reddb_file::PAGE_REQUIRED_CAPABILITIES_FILE_VERSION {
579                reddb_file::file_format::page_format_is_disk_resident(version)
580            } else {
581                required_capabilities & reddb_file::FILE_CAPABILITY_DISK_RESIDENCY != 0
582            };
583        if requires_disk_residency && !self.config.disk_residency {
584            return Err(PagerError::InvalidDatabase(
585                "disk-residency format requires a disk-resident writer".into(),
586            ));
587        }
588        let _publication = self
589            .superblock_publication
590            .lock()
591            .map_err(|_| PagerError::LockPoisoned)?;
592        let result = (|| {
593            for _ in 0..reddb_file::PAGED_SUPERBLOCK_SLOT_COUNT {
594                self.commit_superblock_locked(|image| {
595                    reddb_file::set_database_header_version(image, version)
596                        .map_err(|error| PagerError::InvalidDatabase(error.to_string()))?;
597                    if version >= reddb_file::PAGE_REQUIRED_CAPABILITIES_FILE_VERSION {
598                        reddb_file::set_database_header_required_capabilities(
599                            image,
600                            required_capabilities,
601                        )
602                        .map_err(|error| PagerError::InvalidDatabase(error.to_string()))?;
603                    }
604                    Ok(())
605                })?;
606            }
607            let mut header = self.header_write()?;
608            header.version = version;
609            if version >= reddb_file::PAGE_REQUIRED_CAPABILITIES_FILE_VERSION {
610                header.required_capabilities = required_capabilities;
611            }
612            Ok(())
613        })();
614        if result.is_err() {
615            self.writeback_failed
616                .store(true, std::sync::atomic::Ordering::Release);
617        }
618        result
619    }
620
621    /// Durably enable mandatory reader capabilities before dependent pages exist.
622    ///
623    /// The set only grows. Both superblock copies cross the fence before this
624    /// returns, so a crash can leave a conservatively declared unused feature,
625    /// never reachable bytes whose decoder requirement was not published.
626    pub fn fence_required_capabilities(&self, capabilities: u64) -> Result<(), PagerError> {
627        let _ownership = self.lock_ownership_mutations();
628        self.ensure_writeback_available()?;
629        if self.config.read_only {
630            return Err(PagerError::ReadOnly);
631        }
632        let unsupported = capabilities & !reddb_file::FILE_CAPABILITIES_SUPPORTED;
633        if unsupported != 0 {
634            return Err(PagerError::InvalidDatabase(format!(
635                "unsupported required database capabilities: {unsupported:#018x}"
636            )));
637        }
638
639        let current = self.header_read()?.clone();
640        if current.is_recovery_read_only() {
641            return Err(PagerError::InvalidDatabase(
642                "recovery read-only format cannot enable capabilities".into(),
643            ));
644        }
645        let legacy_disk_residency =
646            reddb_file::file_format::page_format_is_disk_resident(current.version);
647        let required = current.required_capabilities
648            | capabilities
649            | if legacy_disk_residency {
650                reddb_file::FILE_CAPABILITY_DISK_RESIDENCY
651            } else {
652                0
653            };
654        let version = current
655            .version
656            .max(reddb_file::PAGE_REQUIRED_CAPABILITIES_FILE_VERSION);
657        if required == current.required_capabilities && version == current.version {
658            return Ok(());
659        }
660        if required & reddb_file::FILE_CAPABILITY_DISK_RESIDENCY != 0 && !self.config.disk_residency
661        {
662            return Err(PagerError::InvalidDatabase(
663                "disk-residency capability requires a disk-resident writer".into(),
664            ));
665        }
666        if self.batch_active()? {
667            return Err(PagerError::InvalidDatabase(
668                "capability publication requires no active prepared batch".into(),
669            ));
670        }
671
672        let _publication = self
673            .superblock_publication
674            .lock()
675            .map_err(|_| PagerError::LockPoisoned)?;
676        let result = (|| {
677            for _ in 0..reddb_file::PAGED_SUPERBLOCK_SLOT_COUNT {
678                self.commit_superblock_locked(|image| {
679                    reddb_file::set_database_header_version(image, version)
680                        .and_then(|()| {
681                            reddb_file::set_database_header_required_capabilities(image, required)
682                        })
683                        .map_err(|error| PagerError::InvalidDatabase(error.to_string()))?;
684                    if required & reddb_file::FILE_CAPABILITY_CATALOG_ROOT != 0
685                        && current.required_capabilities & reddb_file::FILE_CAPABILITY_CATALOG_ROOT
686                            == 0
687                    {
688                        reddb_file::file_format::set_database_header_manifest_reference(
689                            image,
690                            reddb_file::page_proof::PageReference::default(),
691                        )
692                        .map_err(|error| PagerError::InvalidDatabase(error.to_string()))?;
693                    }
694                    if required & reddb_file::file_format::FILE_CAPABILITY_DWB_PUBLICATION != 0
695                        && current.required_capabilities
696                            & reddb_file::file_format::FILE_CAPABILITY_DWB_PUBLICATION
697                            == 0
698                    {
699                        if !self.dwb_enabled {
700                            return Err(PagerError::InvalidDatabase(
701                                "DWB generation authority requires protected page writes".into(),
702                            ));
703                        }
704                        reddb_file::file_format::set_database_header_dwb_publication(image, None)
705                            .map_err(|error| PagerError::InvalidDatabase(error.to_string()))?;
706                    }
707                    Ok(())
708                })?;
709            }
710            let mut header = self.header_write()?;
711            header.version = version;
712            header.required_capabilities = required;
713            Ok(())
714        })();
715        if result.is_err() {
716            self.writeback_failed
717                .store(true, std::sync::atomic::Ordering::Release);
718        }
719        result
720    }
721
722    /// Load the database header from the newest valid superblock copy.
723    fn load_header(&self) -> Result<(), PagerError> {
724        let (_, image) = self.newest_superblock()?;
725
726        let mut decoded_header = reddb_file::decode_published_database_header(&image)
727            .map_err(|err| PagerError::InvalidDatabase(err.to_string()))?;
728        // Cache eviction (or DWB recovery) can persist pages before the next
729        // superblock publication. After a crash the allocation counter may
730        // therefore lag behind live pages already in the file. Reserve the
731        // entire physical extent before allocating; never overwrite that tail.
732        // A partial last page is reserved too, without treating it as verified.
733        let physical_pages = self.file_lock()?.file_len()?.div_ceil(PAGE_SIZE as u64);
734        let mut physical_pages = u32::try_from(physical_pages).map_err(|_| {
735            PagerError::InvalidDatabase("data file exceeds the page address space".into())
736        })?;
737        if self.config.read_only {
738            if let Some(id) = self
739                .read_only_recovery
740                .read()
741                .map_err(|_| PagerError::LockPoisoned)?
742                .keys()
743                .max()
744            {
745                physical_pages = physical_pages.max(id.checked_add(1).ok_or_else(|| {
746                    PagerError::InvalidDatabase(
747                        "recovery page exceeds the page address space".into(),
748                    )
749                })?);
750            }
751        }
752        if decoded_header.page_count < physical_pages {
753            decoded_header.page_count = physical_pages;
754            if !self.config.read_only {
755                *self.header_dirty_lock()? = true;
756            }
757        }
758        let freelist_head = decoded_header.freelist_head;
759
760        {
761            let mut header = self.header_write()?;
762            *header = decoded_header;
763        }
764
765        // Initialize freelist
766        {
767            let mut freelist = self.freelist_write()?;
768            *freelist = FreeList::from_header(freelist_head, 0);
769        }
770
771        Ok(())
772    }
773
774    /// Publish the in-memory database header as a new superblock generation.
775    fn write_header(&self) -> Result<(), PagerError> {
776        if self.config.read_only {
777            return Err(PagerError::ReadOnly);
778        }
779
780        let header = self.header_read()?.clone();
781        self.commit_superblock(|image| {
782            reddb_file::encode_database_header(image, &header)
783                .map_err(|err| PagerError::InvalidDatabase(err.to_string()))
784        })?;
785        *self.header_dirty_lock()? = false;
786
787        Ok(())
788    }
789
790    /// Read a page from disk (bypassing cache)
791    fn read_page_raw(&self, page_id: u32) -> Result<Page, PagerError> {
792        let mut file = self.file_lock()?;
793        let page = match self
794            .staged_pages
795            .lock()
796            .map_err(|_| PagerError::LockPoisoned)?
797            .as_ref()
798        {
799            Some(batch) => match batch.read(&mut *file, page_id)? {
800                Some(page) => page,
801                None => pager_io::read_page(&mut *file, page_id, self.config.verify_checksums)?,
802            },
803            None => pager_io::read_page(&mut *file, page_id, self.config.verify_checksums)?,
804        };
805        drop(file);
806        if self.config.verify_checksums && page_id != 0 && page.lsn() != 0 {
807            self.wal
808                .validate_page_lsn(page.lsn(), self.header_read()?.checkpoint_lsn)
809                .map_err(|error| {
810                    PagerError::InvalidDatabase(format!(
811                        "page {page_id} has unverified LSN {}: {error}",
812                        page.lsn()
813                    ))
814                })?;
815        }
816        Ok(page)
817    }
818
819    fn ensure_writeback_available(&self) -> Result<(), PagerError> {
820        if self
821            .writeback_failed
822            .load(std::sync::atomic::Ordering::Acquire)
823        {
824            return Err(PagerError::Io(std::io::Error::other(
825                "page writeback failed; reopen the database to recover before writing",
826            )));
827        }
828        Ok(())
829    }
830
831    /// Reject new logical writes after a physical page writeback failed.
832    ///
833    /// The failed write may have reached disk partially. Reopening lets the
834    /// double-write buffer and WAL establish the durable generation before
835    /// the database accepts another mutation.
836    pub fn check_writeback_available(&self) -> Result<(), PagerError> {
837        self.ensure_writeback_available()
838    }
839
840    fn record_ownership(
841        &self,
842        operation: impl FnOnce(&W) -> std::io::Result<()>,
843    ) -> Result<(), PagerError> {
844        if let Err(error) = operation(&self.wal) {
845            self.writeback_failed
846                .store(true, std::sync::atomic::Ordering::Release);
847            return Err(PagerError::Io(error));
848        }
849        Ok(())
850    }
851
852    /// Amortize protected eviction over one bounded DWB frame. Keeping the
853    /// remaining cache dirty would otherwise pay three fsyncs per victim.
854    fn write_evicted_dirty_page(
855        &self,
856        page: Page,
857        pending_page: Option<u32>,
858    ) -> Result<(), PagerError> {
859        if self.bounded_writeback()? {
860            return self.write_dirty_page_batch(&[(page.page_id(), page)]);
861        }
862        let mut ids = self.cache.dirty_page_ids(pager_io::DWB_MAX_PAGES_PER_FRAME);
863        // The admitting write must not join its predecessor's eviction batch.
864        // Its image is dirty for readers, but publication still belongs to the
865        // caller after the earlier writeback has succeeded.
866        ids.retain(|id| Some(*id) != pending_page);
867        // Leave room for the already evicted image without reducing a full
868        // frame merely because the pending image occupied one candidate slot.
869        ids.truncate(pager_io::DWB_MAX_PAGES_PER_FRAME.saturating_sub(1));
870        ids.sort_unstable();
871        let mut pages = self.cache.flush_dirty_pages(&ids);
872        pages.push((page.page_id(), page));
873        pages.sort_unstable_by_key(|(id, _)| *id);
874        self.write_dirty_page_batch(&pages)
875    }
876
877    /// Write a page to disk (bypassing cache)
878    fn write_page_raw(&self, page_id: u32, page: &Page) -> Result<(), PagerError> {
879        self.ensure_writeback_available()?;
880        if self.config.read_only {
881            return Err(PagerError::ReadOnly);
882        }
883        let mut file = self.file_lock()?;
884        self.ensure_writeback_available()?;
885        if let Some(batch) = self
886            .staged_pages
887            .lock()
888            .map_err(|_| PagerError::LockPoisoned)?
889            .as_mut()
890        {
891            return Ok(batch.stage(&mut *file, page_id, page)?);
892        }
893        Ok(pager_io::write_page(&mut *file, page_id, page)?)
894    }
895
896    /// Read a page (cache-aware)
897    pub fn read_page(&self, page_id: u32) -> Result<Page, PagerError> {
898        if self.config.read_only {
899            if let Some(page) = self
900                .read_only_recovery
901                .read()
902                .map_err(|_| PagerError::LockPoisoned)?
903                .get(&page_id)
904            {
905                if self.config.verify_checksums {
906                    page.verify_checksum()?;
907                }
908                return Ok(page.clone());
909            }
910        }
911        // Check cache first
912        if let Some(page) = self.cache.get(page_id) {
913            return Ok(page);
914        }
915
916        // A miss can evict and write dirty pages. Keep disk loading and cache
917        // publication inside the same ownership epoch as writers/checkpoints.
918        // Hits retain their existing path without taking this guard.
919        let _ownership = self.lock_ownership_mutations();
920        if let Some(page) = self.cache.get(page_id) {
921            return Ok(page);
922        }
923
924        // Cache miss - read from disk
925        let page = self.read_page_raw(page_id)?;
926
927        // Add to cache
928        if let Some(dirty_page) = self.cache.insert_if_absent(page_id, page.clone()) {
929            // Evicted page was dirty, need to write it back
930            self.write_evicted_dirty_page(dirty_page, None)?;
931        }
932
933        Ok(page)
934    }
935
936    /// Read a page without verifying checksum (for encrypted pages)
937    ///
938    /// Use this when the page content has its own integrity protection
939    /// (e.g., AES-GCM authentication tag for encrypted pages).
940    pub fn read_page_no_checksum(&self, page_id: u32) -> Result<Page, PagerError> {
941        if self.config.read_only {
942            if let Some(page) = self
943                .read_only_recovery
944                .read()
945                .map_err(|_| PagerError::LockPoisoned)?
946                .get(&page_id)
947            {
948                return Ok(page.clone());
949            }
950        }
951        // Check cache first
952        if let Some(page) = self.cache.get(page_id) {
953            return Ok(page);
954        }
955
956        let _ownership = self.lock_ownership_mutations();
957        if let Some(page) = self.cache.get(page_id) {
958            return Ok(page);
959        }
960
961        // Cache miss - use the same byte I/O without a page CRC check.
962        let mut file = self.file_lock()?;
963        let page = pager_io::read_page(&mut *file, page_id, false)?;
964        drop(file);
965
966        // Add to cache (no checksum verification)
967        if let Some(dirty_page) = self.cache.insert_if_absent(page_id, page.clone()) {
968            // Evicted page was dirty, need to write it back
969            self.write_evicted_dirty_page(dirty_page, None)?;
970        }
971
972        Ok(page)
973    }
974
975    /// Write a page (cache-aware)
976    pub fn write_page(&self, page_id: u32, mut page: Page) -> Result<(), PagerError> {
977        let _ownership = self.lock_ownership_mutations();
978        self.ensure_writeback_available()?;
979        if self.config.read_only {
980            return Err(PagerError::ReadOnly);
981        }
982
983        pager_io::check_page_identity(&page, page_id)?;
984        self.capture_page_before_write(page_id)?;
985
986        // Update checksum
987        page.update_checksum();
988
989        // Add to cache and mark dirty. If the cache had to evict a
990        // dirty entry, write it through immediately — the evicted
991        // page will never be flushed otherwise (same bug fixed in
992        // `allocate_page`).
993        if let Some(dirty_page) = self.cache.insert_dirty(page_id, page) {
994            self.write_evicted_dirty_page(dirty_page, Some(page_id))?;
995        }
996
997        Ok(())
998    }
999
1000    /// Write a page without updating checksum (for encrypted pages)
1001    ///
1002    /// Use this when the page content has its own integrity protection
1003    /// (e.g., AES-GCM authentication tag for encrypted pages).
1004    pub fn write_page_no_checksum(&self, page_id: u32, page: Page) -> Result<(), PagerError> {
1005        let _ownership = self.lock_ownership_mutations();
1006        self.ensure_writeback_available()?;
1007        if self.config.read_only {
1008            return Err(PagerError::ReadOnly);
1009        }
1010
1011        self.capture_page_before_write(page_id)?;
1012
1013        // Add to cache and mark dirty (no checksum update). Same
1014        // eviction-write-through guard as `write_page`.
1015        if let Some(dirty_page) = self.cache.insert_dirty(page_id, page) {
1016            self.write_evicted_dirty_page(dirty_page, Some(page_id))?;
1017        }
1018
1019        Ok(())
1020    }
1021
1022    fn capture_page_before_write(&self, page_id: u32) -> Result<(), PagerError> {
1023        let mut required = false;
1024        self.record_ownership(|wal| {
1025            required = wal.needs_page_image(page_id)?;
1026            Ok(())
1027        })?;
1028        if required {
1029            let previous = self.read_page(page_id)?;
1030            self.record_ownership(|wal| wal.capture_page_image(&previous))?;
1031        }
1032        Ok(())
1033    }
1034
1035    /// Allocate a new page
1036    pub fn allocate_page(&self, page_type: PageType) -> Result<Page, PagerError> {
1037        let _ownership = self.lock_ownership_mutations();
1038        self.ensure_writeback_available()?;
1039        if self.config.read_only {
1040            return Err(PagerError::ReadOnly);
1041        }
1042
1043        // Try to get from freelist first
1044        let page_id = if self.batch_active()? {
1045            // Prepared frames may extend the file but cannot reuse allocator
1046            // state whose published roots still name the old ownership.
1047            let mut header = self.header_write()?;
1048            let id = header.page_count;
1049            header.page_count = id
1050                .checked_add(1)
1051                .ok_or_else(|| PagerError::InvalidDatabase("page count overflow".into()))?;
1052            *self.header_dirty_lock()? = true;
1053            id
1054        } else {
1055            let mut freelist = self.freelist_write()?;
1056            // A previous fence can fail after loading the trunk into memory.
1057            // Retry it before the fast path can consume any of those pages.
1058            self.complete_allocation_fence()?;
1059            if let Some(id) = freelist.next_in_memory_page() {
1060                self.record_ownership(|wal| wal.reserve_pages(id, 1))?;
1061                let id = freelist
1062                    .allocate()
1063                    .expect("free page checked while allocator is locked");
1064                self.allocation_reused
1065                    .store(true, std::sync::atomic::Ordering::Release);
1066                id
1067            } else if freelist.trunk_head() != 0 {
1068                let trunk_id = freelist.trunk_head();
1069
1070                let trunk = self.read_page(trunk_id).map_err(|e| match e {
1071                    PagerError::PageNotFound(_) => {
1072                        PagerError::InvalidDatabase("Freelist trunk missing".to_string())
1073                    }
1074                    other => other,
1075                })?;
1076                if trunk.page_id() != trunk_id {
1077                    return Err(PagerError::InvalidDatabase(
1078                        "misdirected freelist trunk".into(),
1079                    ));
1080                }
1081
1082                // Persist ownership of every page detached from the trunk
1083                // before either superblock stops naming that trunk.
1084                let mut detached = FreeList::new();
1085                detached
1086                    .load_from_trunk(&trunk)
1087                    .map_err(|e| PagerError::InvalidDatabase(format!("Freelist: {e}")))?;
1088                let mut pages = Vec::new();
1089                let mut seen = std::collections::HashSet::new();
1090                let page_count = self.page_count()?;
1091                while let Some(id) = detached.allocate() {
1092                    if id < FIRST_ALLOCATABLE_PAGE || id >= page_count || !seen.insert(id) {
1093                        return Err(PagerError::InvalidDatabase(
1094                            "invalid freelist trunk page".into(),
1095                        ));
1096                    }
1097                    pages.push(id);
1098                }
1099                let mut journaled = false;
1100                self.record_ownership(|wal| {
1101                    journaled = wal.has_page_ownership_journal()?;
1102                    Ok(())
1103                })?;
1104                if !journaled {
1105                    self.allocate_unjournaled_trunk(&mut freelist, trunk, &pages)?
1106                } else {
1107                    self.record_ownership(|wal| wal.detach_free_pages(&pages))?;
1108
1109                    freelist
1110                        .load_from_trunk(&trunk)
1111                        .map_err(|e| PagerError::InvalidDatabase(format!("Freelist: {}", e)))?;
1112                    let mut header = self.header_write()?;
1113                    header.freelist_head = freelist.trunk_head();
1114                    *self.header_dirty_lock()? = true;
1115                    drop(header);
1116                    // The consumed trunk itself becomes reusable. Forget its old
1117                    // allocator role durably before cache eviction can overwrite it.
1118                    // Keep the allocator locked so another allocator cannot race
1119                    // ahead of this fence. A failed fence leaves allocation stopped.
1120                    self.allocation_fence_pending
1121                        .store(true, std::sync::atomic::Ordering::Release);
1122                    self.complete_allocation_fence()?;
1123                    let next = freelist.next_in_memory_page().ok_or_else(|| {
1124                        PagerError::InvalidDatabase("Freelist empty after trunk load".to_string())
1125                    })?;
1126                    self.record_ownership(|wal| wal.reserve_pages(next, 1))?;
1127                    let id = freelist.allocate().ok_or_else(|| {
1128                        PagerError::InvalidDatabase("Freelist empty after trunk load".to_string())
1129                    })?;
1130                    self.allocation_reused
1131                        .store(true, std::sync::atomic::Ordering::Release);
1132                    id
1133                }
1134            } else {
1135                // No free pages, extend file
1136                let mut header = self.header_write()?;
1137                let id = header.page_count;
1138                let next = id
1139                    .checked_add(1)
1140                    .ok_or_else(|| PagerError::InvalidDatabase("page count overflow".into()))?;
1141                self.record_ownership(|wal| wal.reserve_pages(id, 1))?;
1142                header.page_count = next;
1143                *self.header_dirty_lock()? = true;
1144                id
1145            }
1146        };
1147
1148        let page = Page::new(page_type, page_id);
1149
1150        // Write to cache. The evicted page (if any) is dirty by
1151        // definition — `cache.insert_dirty` only returns `Some` when it
1152        // had to evict a dirty entry to make room. The previous
1153        // version dropped that return value, which silently lost
1154        // writes whenever a freshly-allocated page caused an LRU
1155        // eviction. This shows up under heavy ingest as
1156        // "B-tree insert error: Pager error: I/O error: failed to
1157        // fill whole buffer" later, when something tries to read
1158        // back the never-flushed page. Mirror `read_page`'s
1159        // handling: write the evicted page through immediately so
1160        // the on-disk image stays consistent.
1161        if let Some(dirty_page) = self.cache.insert_dirty(page_id, page.clone()) {
1162            self.write_evicted_dirty_page(dirty_page, Some(page_id))?;
1163        }
1164
1165        Ok(page)
1166    }
1167
1168    /// Keep untouched free IDs reachable on disk when no ownership epoch can
1169    /// recover a detached trunk. Removing one ID costs one protected page write;
1170    /// checkpoint/replay epochs retain their batched, journaled allocation path.
1171    fn allocate_unjournaled_trunk(
1172        &self,
1173        freelist: &mut FreeList,
1174        mut trunk: Page,
1175        pages: &[u32],
1176    ) -> Result<u32, PagerError> {
1177        let id = *pages
1178            .first()
1179            .expect("validated trunk includes its own page");
1180        self.record_ownership(|wal| wal.reserve_pages(id, 1))?;
1181        if pages.len() > 1 {
1182            let remaining =
1183                u32::try_from(pages.len() - 2).expect("a validated trunk fits in one page");
1184            trunk.content_mut()[4..8].copy_from_slice(&remaining.to_le_bytes());
1185            trunk.update_checksum();
1186            self.cache.remove(trunk.page_id());
1187            self.write_pages_through_dwb(&[(trunk.page_id(), trunk)])?;
1188            freelist.note_durable_trunk_allocation(remaining);
1189        } else {
1190            // The empty trunk itself is the last available page. Fence both
1191            // superblocks before it can be overwritten as user data.
1192            freelist
1193                .load_from_trunk(&trunk)
1194                .map_err(|error| PagerError::InvalidDatabase(error.to_string()))?;
1195            self.header_write()?.freelist_head = freelist.trunk_head();
1196            *self.header_dirty_lock()? = true;
1197            self.allocation_fence_pending
1198                .store(true, std::sync::atomic::Ordering::Release);
1199            self.complete_allocation_fence()?;
1200            assert_eq!(freelist.allocate(), Some(id));
1201        }
1202        self.allocation_reused
1203            .store(true, std::sync::atomic::Ordering::Release);
1204        Ok(id)
1205    }
1206
1207    /// Reserve a contiguous extent of vector pages.
1208    pub fn reserve_contig_extent(&self, n_pages: u32) -> Result<ExtentId, PagerError> {
1209        let _ownership = self.lock_ownership_mutations();
1210        self.ensure_writeback_available()?;
1211        if self.config.read_only {
1212            return Err(PagerError::ReadOnly);
1213        }
1214        if n_pages == 0 {
1215            return Err(PagerError::InvalidDatabase(
1216                "contiguous extent must reserve at least one page".to_string(),
1217            ));
1218        }
1219
1220        let start_page = {
1221            let mut header = self.header_write()?;
1222            let start = header.page_count;
1223            let end = header.page_count.checked_add(n_pages).ok_or_else(|| {
1224                PagerError::InvalidDatabase("contiguous extent page count overflow".to_string())
1225            })?;
1226            self.record_ownership(|wal| wal.reserve_pages(start, n_pages))?;
1227            header.page_count = end;
1228            *self.header_dirty_lock()? = true;
1229            start
1230        };
1231
1232        for page_id in start_page..start_page + n_pages {
1233            let mut page = Page::new(PageType::Vector, page_id);
1234            page.update_checksum();
1235            if let Some(dirty_page) = self.cache.insert_dirty(page_id, page) {
1236                self.write_evicted_dirty_page(dirty_page, Some(page_id))?;
1237            }
1238        }
1239
1240        Ok(ExtentId {
1241            start_page,
1242            n_pages,
1243        })
1244    }
1245
1246    /// Free a page (return to freelist)
1247    pub fn free_page(&self, page_id: u32) -> Result<(), PagerError> {
1248        if self.batch_active()? {
1249            return Err(PagerError::InvalidDatabase(
1250                "prepared DWB batches cannot retire published pages".into(),
1251            ));
1252        }
1253        let _ownership = self.lock_ownership_mutations();
1254        self.ensure_writeback_available()?;
1255        if self.config.read_only {
1256            return Err(PagerError::ReadOnly);
1257        }
1258
1259        let mut deferred = self
1260            .deferred_reclamation
1261            .lock()
1262            .map_err(|_| PagerError::LockPoisoned)?;
1263        if deferred.enabled {
1264            if page_id < FIRST_ALLOCATABLE_PAGE || page_id >= self.page_count()? {
1265                return Err(PagerError::InvalidDatabase(
1266                    "invalid deferred free page".into(),
1267                ));
1268            }
1269            self.record_ownership(|wal| wal.retire_pages(&[page_id]))?;
1270            // Keep dirty forwarding/empty nodes until publication. An
1271            // incidental flush must leave the previous root traversable.
1272            deferred.pending.insert(page_id);
1273            return Ok(());
1274        }
1275        drop(deferred);
1276
1277        self.record_ownership(|wal| wal.retire_pages(&[page_id]))?;
1278
1279        // Remove from cache
1280        self.cache.remove(page_id);
1281
1282        // Add to freelist
1283        let mut freelist = self.freelist_write()?;
1284        freelist.free(page_id);
1285
1286        *self.header_dirty_lock()? = true;
1287
1288        Ok(())
1289    }
1290
1291    pub fn enable_deferred_reclamation(&self) -> Result<(), PagerError> {
1292        let _ownership = self.lock_ownership_mutations();
1293        self.deferred_reclamation
1294            .lock()
1295            .map_err(|_| PagerError::LockPoisoned)?
1296            .enabled = true;
1297        Ok(())
1298    }
1299
1300    pub fn deferred_reclamation_pages(&self) -> Result<(Vec<u32>, Vec<u32>), PagerError> {
1301        let deferred = self
1302            .deferred_reclamation
1303            .lock()
1304            .map_err(|_| PagerError::LockPoisoned)?;
1305        Ok((
1306            deferred.pending.iter().copied().collect(),
1307            deferred.published.iter().copied().collect(),
1308        ))
1309    }
1310
1311    pub fn restore_deferred_reclamation(&self, pages: &[u32]) -> Result<(), PagerError> {
1312        let _ownership = self.lock_ownership_mutations();
1313        let count = self.page_count()?;
1314        if pages
1315            .iter()
1316            .any(|id| *id < FIRST_ALLOCATABLE_PAGE || *id >= count)
1317        {
1318            return Err(PagerError::InvalidDatabase(
1319                "invalid persisted deferred free page".into(),
1320            ));
1321        }
1322        let mut deferred = self
1323            .deferred_reclamation
1324            .lock()
1325            .map_err(|_| PagerError::LockPoisoned)?;
1326        deferred.pending.extend(pages);
1327        deferred.published.extend(pages);
1328        Ok(())
1329    }
1330
1331    /// The caller has durably published a catalog that retains `retained`
1332    /// and no longer names `released`, after waiting for old tree readers.
1333    pub fn complete_deferred_publication(
1334        &self,
1335        retained: &[u32],
1336        released: &[u32],
1337    ) -> Result<(), PagerError> {
1338        let _ownership = self.lock_ownership_mutations();
1339        let mut deferred = self
1340            .deferred_reclamation
1341            .lock()
1342            .map_err(|_| PagerError::LockPoisoned)?;
1343        if !released.is_empty() {
1344            self.free_retired_pages(released)?;
1345        }
1346        for id in released {
1347            deferred.pending.remove(id);
1348        }
1349        deferred.published = retained.iter().copied().collect();
1350        Ok(())
1351    }
1352
1353    /// Record the pages whose catalog ownership is about to be retired.
1354    pub fn record_retirement_intent(&self, pages: &[u32]) -> Result<(), PagerError> {
1355        let _ownership = self.lock_ownership_mutations();
1356        self.ensure_writeback_available()?;
1357        if self.config.read_only {
1358            return Err(PagerError::ReadOnly);
1359        }
1360        let page_count = self.page_count()?;
1361        let mut seen = std::collections::HashSet::with_capacity(pages.len());
1362        if pages
1363            .iter()
1364            .any(|id| *id < FIRST_ALLOCATABLE_PAGE || *id >= page_count || !seen.insert(*id))
1365        {
1366            return Err(PagerError::InvalidDatabase(
1367                "invalid retirement intent".into(),
1368            ));
1369        }
1370        self.record_ownership(|wal| wal.retire_pages(pages))
1371    }
1372
1373    /// Return a fully validated, unpublished tree generation in one allocator
1374    /// mutation. Call only after its removal from the manifest is durable.
1375    pub fn free_retired_pages(&self, pages: &[u32]) -> Result<(), PagerError> {
1376        let _ownership = self.lock_ownership_mutations();
1377        self.ensure_writeback_available()?;
1378        if self.config.read_only {
1379            return Err(PagerError::ReadOnly);
1380        }
1381        let page_count = self.page_count()?;
1382        if pages
1383            .iter()
1384            .any(|id| *id < FIRST_ALLOCATABLE_PAGE || *id >= page_count)
1385        {
1386            return Err(PagerError::InvalidDatabase(
1387                "invalid retired page range".into(),
1388            ));
1389        }
1390        let mut seen = std::collections::HashSet::with_capacity(pages.len());
1391        if pages.iter().any(|id| !seen.insert(*id)) {
1392            return Err(PagerError::InvalidDatabase("duplicate retired page".into()));
1393        }
1394        let mut freelist = self.freelist_write()?;
1395        if freelist.contains_any(&seen) {
1396            return Err(PagerError::InvalidDatabase(
1397                "retired page is already free".into(),
1398            ));
1399        }
1400        // Some free pages remain in on-disk trunks until allocation reaches
1401        // them. Check those too before changing the cache or the allocator.
1402        // The allocator lock prevents concurrent consumption or publication.
1403        let mut trunk_id = freelist.trunk_head();
1404        let mut visited = std::collections::HashSet::new();
1405        while trunk_id != 0 && !seen.is_empty() {
1406            if trunk_id < FIRST_ALLOCATABLE_PAGE
1407                || trunk_id >= page_count
1408                || !visited.insert(trunk_id)
1409            {
1410                return Err(PagerError::InvalidDatabase(
1411                    "invalid freelist trunk chain".into(),
1412                ));
1413            }
1414            let trunk = self.read_page(trunk_id)?;
1415            if trunk.page_id() != trunk_id {
1416                return Err(PagerError::InvalidDatabase(
1417                    "misdirected freelist trunk".into(),
1418                ));
1419            }
1420            let mut loaded = FreeList::new();
1421            let next = loaded
1422                .load_from_trunk(&trunk)
1423                .map_err(|error| PagerError::InvalidDatabase(format!("Freelist: {error}")))?;
1424            while let Some(id) = loaded.allocate() {
1425                if id < FIRST_ALLOCATABLE_PAGE || id >= page_count {
1426                    return Err(PagerError::InvalidDatabase(
1427                        "invalid free page range".into(),
1428                    ));
1429                }
1430                if seen.contains(&id) {
1431                    return Err(PagerError::InvalidDatabase(
1432                        "retired page is already free".into(),
1433                    ));
1434                }
1435            }
1436            trunk_id = next.unwrap_or(0);
1437        }
1438        self.record_ownership(|wal| wal.retire_pages(pages))?;
1439        for id in pages {
1440            self.cache.remove(*id);
1441        }
1442        freelist.free_batch(pages);
1443        *self.header_dirty_lock()? = true;
1444        Ok(())
1445    }
1446
1447    fn pages_not_in_freelist(
1448        &self,
1449        pages: &[u32],
1450        freelist: &FreeList,
1451        page_count: u32,
1452    ) -> Result<std::collections::HashSet<u32>, PagerError> {
1453        let mut remaining = std::collections::HashSet::with_capacity(pages.len());
1454        if pages
1455            .iter()
1456            .any(|id| *id < FIRST_ALLOCATABLE_PAGE || *id >= page_count || !remaining.insert(*id))
1457        {
1458            return Err(PagerError::InvalidDatabase(
1459                "invalid recovery page set".into(),
1460            ));
1461        }
1462        let requested = remaining.clone();
1463        freelist
1464            .remove_existing_from(&requested, &mut remaining)
1465            .map_err(|error| PagerError::InvalidDatabase(format!("Freelist: {error}")))?;
1466        let mut trunk_id = freelist.trunk_head();
1467        let mut visited = std::collections::HashSet::new();
1468        while trunk_id != 0 {
1469            if trunk_id < FIRST_ALLOCATABLE_PAGE
1470                || trunk_id >= page_count
1471                || !visited.insert(trunk_id)
1472            {
1473                return Err(PagerError::InvalidDatabase(
1474                    "invalid freelist trunk chain".into(),
1475                ));
1476            }
1477            let trunk = self.read_page(trunk_id)?;
1478            if trunk.page_id() != trunk_id {
1479                return Err(PagerError::InvalidDatabase(
1480                    "misdirected freelist trunk".into(),
1481                ));
1482            }
1483            let mut loaded = FreeList::new();
1484            let next = loaded
1485                .load_from_trunk(&trunk)
1486                .map_err(|error| PagerError::InvalidDatabase(format!("Freelist: {error}")))?;
1487            while let Some(id) = loaded.allocate() {
1488                if id < FIRST_ALLOCATABLE_PAGE || id >= page_count {
1489                    return Err(PagerError::InvalidDatabase(
1490                        "invalid free page range".into(),
1491                    ));
1492                }
1493                if requested.contains(&id) && !remaining.remove(&id) {
1494                    return Err(PagerError::InvalidDatabase(
1495                        "duplicate recovery page".into(),
1496                    ));
1497                }
1498            }
1499            trunk_id = next.unwrap_or(0);
1500        }
1501        Ok(remaining)
1502    }
1503
1504    /// Add proven-unowned pages to the allocator exactly once during recovery.
1505    /// The caller must exclude all allocation and publication until this state
1506    /// has been synced to both superblocks and the journal is completed.
1507    pub fn recover_unowned_pages(&self, pages: &[u32]) -> Result<(), PagerError> {
1508        let _ownership = self.lock_ownership_mutations();
1509        self.ensure_writeback_available()?;
1510        if self.config.read_only {
1511            return Err(PagerError::ReadOnly);
1512        }
1513        let page_count = self.page_count()?;
1514        let mut freelist = self.freelist_write()?;
1515        let remaining = self.pages_not_in_freelist(pages, &freelist, page_count)?;
1516        if !remaining.is_empty() {
1517            let mut missing: Vec<_> = remaining.into_iter().collect();
1518            missing.sort_unstable();
1519            for id in &missing {
1520                self.cache.remove(*id);
1521            }
1522            freelist.free_batch(&missing);
1523            *self.header_dirty_lock()? = true;
1524        }
1525        Ok(())
1526    }
1527
1528    /// Reject a recovery plan if any page it must preserve is already free.
1529    /// The caller must exclude allocation and freelist mutation during recovery.
1530    pub fn validate_owned_pages(&self, pages: &[u32]) -> Result<(), PagerError> {
1531        let page_count = self.page_count()?;
1532        let freelist = self.freelist_write()?;
1533        let absent = self.pages_not_in_freelist(pages, &freelist, page_count)?;
1534        if absent.len() != pages.len() {
1535            return Err(PagerError::InvalidDatabase(
1536                "owned recovery page is free".into(),
1537            ));
1538        }
1539        Ok(())
1540    }
1541
1542    /// Persist the current allocator in both fallback copies during recovery.
1543    /// The caller must exclude allocation until journal completion is durable.
1544    pub fn sync_recovered_allocator(&self) -> Result<(), PagerError> {
1545        let _ownership = self.lock_ownership_mutations();
1546        let result = (|| {
1547            self.sync()?;
1548            let _publication = self
1549                .superblock_publication
1550                .lock()
1551                .map_err(|_| PagerError::LockPoisoned)?;
1552            let header = self.header_read()?.clone();
1553            for _ in 0..reddb_file::PAGED_SUPERBLOCK_SLOT_COUNT {
1554                self.commit_superblock_locked(|image| {
1555                    reddb_file::encode_database_header(image, &header)
1556                        .map_err(|error| PagerError::InvalidDatabase(error.to_string()))
1557                })?;
1558            }
1559            Ok(())
1560        })();
1561        if result.is_err() {
1562            self.writeback_failed
1563                .store(true, std::sync::atomic::Ordering::Release);
1564        }
1565        result
1566    }
1567
1568    /// Get database header
1569    /// Pages available to the allocator, including pages on unloaded trunks.
1570    pub fn free_page_count(&self) -> Result<u32, PagerError> {
1571        Ok(self.freelist_write()?.total_free())
1572    }
1573
1574    pub fn header(&self) -> Result<DatabaseHeader, PagerError> {
1575        Ok(self.header_read()?.clone())
1576    }
1577
1578    pub fn physical_header(&self) -> Result<PhysicalFileHeader, PagerError> {
1579        Ok(self.header_read()?.physical)
1580    }
1581
1582    pub fn update_physical_header(&self, physical: PhysicalFileHeader) -> Result<(), PagerError> {
1583        let _ownership = self.lock_ownership_mutations();
1584        self.ensure_writeback_available()?;
1585        if self.config.read_only {
1586            return Err(PagerError::ReadOnly);
1587        }
1588
1589        let mut header = self.header_write()?;
1590        header.physical = physical;
1591        *self.header_dirty_lock()? = true;
1592        Ok(())
1593    }
1594
1595    /// Get page count
1596    pub fn page_count(&self) -> Result<u32, PagerError> {
1597        Ok(self.header_read()?.page_count)
1598    }
1599
1600    /// Flush all dirty pages to disk
1601    pub fn flush(&self) -> Result<(), PagerError> {
1602        let _ownership = self.lock_ownership_mutations();
1603        self.ensure_writeback_available()?;
1604        if self.config.read_only {
1605            return Ok(());
1606        }
1607
1608        // Persist freelist to trunk pages when dirty
1609        let trunks = {
1610            let mut freelist = self.freelist_write()?;
1611            if freelist.is_dirty() {
1612                let mut header = self.header_write()?;
1613                let trunks = freelist.flush_to_trunks(0, || {
1614                    let id = header.page_count;
1615                    header.page_count += 1;
1616                    id
1617                });
1618                header.freelist_head = freelist.trunk_head();
1619                *self.header_dirty_lock()? = true;
1620                freelist.mark_clean();
1621                trunks
1622            } else {
1623                Vec::new()
1624            }
1625        };
1626
1627        let mut dirty_pages = Vec::new();
1628        for trunk in trunks {
1629            let page_id = trunk.page_id();
1630            if let Some(evicted) = self.cache.insert_dirty(page_id, trunk) {
1631                // Trunk publication can evict dirty data. Keep it in the same
1632                // WAL-first, double-write-protected batch as resident pages.
1633                dirty_pages.push((evicted.page_id(), evicted));
1634            }
1635        }
1636
1637        // IDs are compact; page images are not. At 16 KiB per page, staging
1638        // 256 images plus one DWB frame uses about 5 MiB, independently of
1639        // the dirty cache size. Publish the header only after every batch.
1640        const FLUSH_BATCH_PAGES: usize = 256;
1641        for batch in dirty_pages.chunks(FLUSH_BATCH_PAGES) {
1642            self.write_dirty_page_batch(batch)?;
1643        }
1644        drop(dirty_pages);
1645        // Batch dirty IDs, not every resident ID: otherwise a few edits in
1646        // a mostly clean cache pay a protected write per cache-sized slice.
1647        // Physical order also avoids arbitrary seeks among each batch's homes.
1648        let mut page_ids = self.cache.dirty_page_ids(usize::MAX);
1649        page_ids.sort_unstable();
1650        if self.bounded_writeback()? {
1651            for id in page_ids {
1652                if let Some(page) = self.cache.flush_dirty_page(id) {
1653                    self.write_dirty_page_batch(&[(id, page)])?;
1654                }
1655            }
1656        } else {
1657            for ids in page_ids.chunks(FLUSH_BATCH_PAGES) {
1658                let pages = self.cache.flush_dirty_pages(ids);
1659                self.write_dirty_page_batch(&pages)?;
1660            }
1661        }
1662
1663        // Write header if dirty
1664        if !self.batch_active()? && *self.header_dirty_lock()? {
1665            self.write_header()?;
1666        }
1667
1668        Ok(())
1669    }
1670
1671    fn write_dirty_page_batch(&self, pages: &[(u32, Page)]) -> Result<(), PagerError> {
1672        if pages.is_empty() {
1673            return Ok(());
1674        }
1675        // Every image's WAL must be durable before its home write, including
1676        // dirty data evicted while publishing freelist trunks. LSN zero is
1677        // reserved for pages not produced by a WAL append.
1678        let result = (|| {
1679            let max_lsn = pages
1680                .iter()
1681                .filter_map(|(_, page)| page.header().ok().map(|header| header.lsn))
1682                .max()
1683                .unwrap_or(0);
1684            if max_lsn > 0 {
1685                self.wal.flush_until(max_lsn)?;
1686            }
1687            self.write_pages_through_dwb(pages)
1688        })();
1689        if result.is_err() {
1690            self.writeback_failed
1691                .store(true, std::sync::atomic::Ordering::Release);
1692            for (page_id, _) in pages {
1693                self.cache.mark_dirty(*page_id);
1694            }
1695        }
1696        result
1697    }
1698
1699    /// Sync file to disk (fsync)
1700    pub fn sync(&self) -> Result<(), PagerError> {
1701        let _ownership = self.lock_ownership_mutations();
1702        self.flush()?;
1703
1704        if self
1705            .allocation_reused
1706            .swap(false, std::sync::atomic::Ordering::AcqRel)
1707        {
1708            // flush published the current allocator in one slot. Mirror that
1709            // ownership before callers publish roots that use reclaimed pages;
1710            // recovery through the sibling slot must never allocate them again.
1711            if let Err(error) = self.write_header_and_sync() {
1712                self.allocation_reused
1713                    .store(true, std::sync::atomic::Ordering::Release);
1714                return Err(error);
1715            }
1716        }
1717
1718        let file = self.file_lock()?;
1719        file.sync_all()?;
1720
1721        Ok(())
1722    }
1723
1724    /// Get cache statistics
1725    pub fn cache_stats(&self) -> super::page_cache::CacheStats {
1726        self.cache.stats()
1727    }
1728
1729    /// Pages currently resident in the page cache. Slots are fixed size, so
1730    /// this times the page size is the cache's memory footprint (ADR 0073 §2).
1731    pub fn cache_len(&self) -> usize {
1732        self.cache.len()
1733            + if self.config.read_only {
1734                self.read_only_recovery
1735                    .read()
1736                    .unwrap_or_else(|poisoned| poisoned.into_inner())
1737                    .len()
1738            } else {
1739                0
1740            }
1741    }
1742
1743    /// Preallocated page-cache slots — the sizing this process derived from
1744    /// its page-cache budget share.
1745    pub fn cache_capacity(&self) -> usize {
1746        self.cache.capacity()
1747    }
1748
1749    /// Count dirty pages currently in the page cache.
1750    pub fn dirty_page_count(&self) -> usize {
1751        self.cache.dirty_count()
1752    }
1753
1754    /// Estimated fraction of the page cache holding dirty pages.
1755    /// Returned in `[0, 1]`. Used by the background writer to
1756    /// decide when to kick in aggressive flushing.
1757    pub fn dirty_fraction(&self) -> f64 {
1758        let capacity = self.cache.capacity().max(1) as f64;
1759        self.cache.dirty_count() as f64 / capacity
1760    }
1761
1762    /// Flush up to `max` dirty pages from the cache. Returns the
1763    /// number actually written. Background-writer entry point —
1764    /// reuses the same persistence path as `flush()` but bounded.
1765    pub fn flush_some_dirty(&self, max: usize) -> Result<usize, PagerError> {
1766        let _ownership = self.lock_ownership_mutations();
1767        self.ensure_writeback_available()?;
1768        if self.config.read_only || max == 0 {
1769            return Ok(0);
1770        }
1771        let dirty_pages = self.cache.flush_some_dirty(max);
1772        if dirty_pages.is_empty() {
1773            return Ok(0);
1774        }
1775        let count = dirty_pages.len();
1776        // WAL-first: every cached dirty page carries an LSN that the
1777        // WAL must have already persisted. The full `flush()` path
1778        // enforces this with `wal.flush(max_lsn)`; here we simply
1779        // write through the pager — safe because callers only reach
1780        // this path via the bgwriter, which runs asynchronously
1781        // alongside normal commits that already respect WAL-first.
1782        for (page_id, page) in dirty_pages {
1783            self.write_page(page_id, page)?;
1784        }
1785        Ok(count)
1786    }
1787
1788    /// Get database file path
1789    pub fn path(&self) -> &Path {
1790        &self.path
1791    }
1792
1793    /// Return the exclusively owned handle. Callers must complete their
1794    /// publication boundary before consuming the pager.
1795    pub fn into_inner(self) -> F {
1796        self.file
1797            .into_inner()
1798            .unwrap_or_else(|poisoned| poisoned.into_inner())
1799    }
1800
1801    /// Check if database is read-only
1802    pub fn is_read_only(&self) -> bool {
1803        self.config.read_only
1804    }
1805
1806    /// Get file size in bytes
1807    pub fn file_size(&self) -> Result<u64, PagerError> {
1808        let file = self.file_lock()?;
1809        Ok(file.file_len()?)
1810    }
1811
1812    /// Clear the in-file DWB zone with valid reserved pages.
1813    fn clear_dwb_zone(&self) -> Result<(), PagerError> {
1814        let mut file = self.file_lock()?;
1815        Ok(pager_io::clear_dwb_zone(&mut *file)?)
1816    }
1817
1818    /// Read catalog authority from the newest sealed slot, independently of the
1819    /// cached ordinary header fields.
1820    pub fn manifest_reference(&self) -> Result<reddb_file::page_proof::PageReference, PagerError> {
1821        let _ownership = self.lock_ownership_mutations();
1822        let (_, image) = self.newest_superblock()?;
1823        reddb_file::file_format::database_header_manifest_reference(&image)
1824            .map_err(|err| PagerError::InvalidDatabase(err.to_string()))
1825    }
1826
1827    pub fn manifest_page_id(&self) -> Result<u32, PagerError> {
1828        let reference = self.manifest_reference()?;
1829        Ok(if reference.page_id == 0 {
1830            1
1831        } else {
1832            reference.page_id
1833        })
1834    }
1835
1836    /// Read the sealed catalog root, or page 1 for a legacy bootstrap header.
1837    ///
1838    /// ADR 0074 §2: manifest corruption fails the open didactically, naming the
1839    /// zone — it never falls back to a sidecar and never returns garbage.
1840    pub fn read_manifest_page(&self) -> Result<Page, PagerError> {
1841        let _ownership = self.lock_ownership_mutations();
1842        let root = self.manifest_page_id()?;
1843        self.read_page(root)
1844            .map_err(|_| PagerError::ManifestZoneCorrupt(self.path.clone()))
1845    }
1846
1847    /// The new catalog and all its children must already be durable. Both
1848    /// fallback headers stop naming its predecessor before that root is freed.
1849    pub fn publish_manifest_reference(
1850        &self,
1851        reference: reddb_file::page_proof::PageReference,
1852    ) -> Result<(), PagerError> {
1853        let _ownership = self.lock_ownership_mutations();
1854        self.ensure_writeback_available()?;
1855        if self.config.read_only {
1856            return Err(PagerError::ReadOnly);
1857        }
1858        if reference.page_id < FIRST_ALLOCATABLE_PAGE
1859            || reference.page_id >= self.page_count()?
1860            || reference.digest == [0; 32]
1861            || self.header_read()?.required_capabilities & reddb_file::FILE_CAPABILITY_CATALOG_ROOT
1862                == 0
1863        {
1864            return Err(PagerError::InvalidDatabase(
1865                "invalid catalog generation publication".into(),
1866            ));
1867        }
1868        let header = self.header_read()?.clone();
1869        let _publication = self
1870            .superblock_publication
1871            .lock()
1872            .map_err(|_| PagerError::LockPoisoned)?;
1873        let result = (|| {
1874            for _ in 0..reddb_file::PAGED_SUPERBLOCK_SLOT_COUNT {
1875                self.commit_superblock_locked(|image| {
1876                    reddb_file::encode_database_header(image, &header)
1877                        .map_err(|err| PagerError::InvalidDatabase(err.to_string()))?;
1878                    reddb_file::file_format::set_database_header_manifest_reference(
1879                        image, reference,
1880                    )
1881                    .map_err(|err| PagerError::InvalidDatabase(err.to_string()))
1882                })?;
1883            }
1884            *self.header_dirty_lock()? = false;
1885            Ok(())
1886        })();
1887        if result.is_err() {
1888            self.writeback_failed
1889                .store(true, std::sync::atomic::Ordering::Release);
1890        }
1891        result
1892    }
1893
1894    /// Write pages through the double-write buffer for torn page protection.
1895    ///
1896    /// 1. Write pages to the in-file DWB zone with a header (magic + count + checksum)
1897    /// 2. fsync the DWB zone
1898    /// 3. Write all pages to their final locations in the .rdb file
1899    /// 4. Clear the DWB zone (marks as consumed)
1900    fn write_pages_through_dwb(&self, pages: &[(u32, Page)]) -> Result<(), PagerError> {
1901        if self.config.read_only {
1902            return Err(PagerError::ReadOnly);
1903        }
1904        // Keep the file locked through staging, home writes and clearing so
1905        // another page writer cannot interleave with a protected batch.
1906        let _publication = self
1907            .superblock_publication
1908            .lock()
1909            .map_err(|_| PagerError::LockPoisoned)?;
1910        let mut file = self.file_lock()?;
1911        self.ensure_writeback_available()?;
1912        let result: Result<(), PagerError> = (|| {
1913            let mut staged = self
1914                .staged_pages
1915                .lock()
1916                .map_err(|_| PagerError::LockPoisoned)?;
1917            if let Some(batch) = staged.as_mut() {
1918                for (id, page) in pages {
1919                    batch.stage(&mut *file, *id, page)?;
1920                }
1921                Ok(())
1922            } else {
1923                Ok(pager_io::write_pages(&mut *file, pages, self.dwb_enabled)?)
1924            }
1925        })();
1926        if result.is_err() {
1927            // Set this while holding the file lock: a waiting writer must not
1928            // replace the recovery frame after an interrupted home write.
1929            self.writeback_failed
1930                .store(true, std::sync::atomic::Ordering::Release);
1931        }
1932        result
1933    }
1934
1935    /// Recover from double-write buffer after a crash.
1936    ///
1937    /// If the DWB zone contains valid pages, they were written before the crash
1938    /// interrupted writing to their home locations. Re-apply them.
1939    fn recover_from_dwb(&self) -> Result<(), PagerError> {
1940        let mut file = self.file_lock()?;
1941        if !self.config.read_only
1942            && !self.dwb_enabled
1943            && pager_io::requires_dwb_publication(&mut *file)?
1944        {
1945            return Err(PagerError::InvalidDatabase(
1946                "DWB generation authority requires protected page writes".into(),
1947            ));
1948        }
1949        if self.config.read_only {
1950            for (id, page) in pager_io::read_dwb_pages(&mut *file)? {
1951                // The DWB checksum seals raw bytes, which may be ciphertext.
1952                // Normal reads check the page checksum; encryption hosts use
1953                // read_page_no_checksum and authenticate after decrypting.
1954                self.install_read_only_page(id, page)?;
1955            }
1956            return Ok(());
1957        }
1958        Ok(pager_io::recover_from_dwb(&mut *file)?)
1959    }
1960
1961    /// Install a validated recovery image only in this reader's private view.
1962    /// The source and ordinary mutation APIs remain read-only. Bound pinned
1963    /// images by the cache budget (with room for one complete DWB frame).
1964    pub fn restore_read_only_page(&self, page_id: u32, page: Page) -> Result<(), PagerError> {
1965        if !self.config.read_only || page_id == 0 || page.page_id() != page_id {
1966            return Err(PagerError::InvalidDatabase(
1967                "invalid read-only recovery image".into(),
1968            ));
1969        }
1970        page.verify_checksum()?;
1971        self.install_read_only_page(page_id, page)
1972    }
1973
1974    fn install_read_only_page(&self, page_id: u32, page: Page) -> Result<(), PagerError> {
1975        let mut images = self
1976            .read_only_recovery
1977            .write()
1978            .map_err(|_| PagerError::LockPoisoned)?;
1979        let limit = self
1980            .config
1981            .cache_size
1982            .max(pager_io::DWB_MAX_PAGES_PER_FRAME);
1983        if images.len() >= limit && !images.contains_key(&page_id) {
1984            return Err(PagerError::InvalidDatabase(
1985                "read-only recovery exceeds the page cache budget".into(),
1986            ));
1987        }
1988        images.insert(page_id, page);
1989        Ok(())
1990    }
1991
1992    /// Finish a consumed trunk's durable ownership change, retaining errors
1993    /// across allocation attempts until both copies have been synchronized.
1994    /// Called with the freelist locked when consuming or retrying a trunk.
1995    fn complete_allocation_fence(&self) -> Result<(), PagerError> {
1996        if self
1997            .allocation_fence_pending
1998            .load(std::sync::atomic::Ordering::Acquire)
1999        {
2000            self.write_header_and_sync()?;
2001            self.write_header_and_sync()?;
2002            self.allocation_fence_pending
2003                .store(false, std::sync::atomic::Ordering::Release);
2004        }
2005        Ok(())
2006    }
2007
2008    /// Write header and sync to disk (public for checkpointer).
2009    pub fn write_header_and_sync(&self) -> Result<(), PagerError> {
2010        let _ownership = self.lock_ownership_mutations();
2011        self.write_header()?;
2012        let file = self.file_lock()?;
2013        file.sync_all()?;
2014        Ok(())
2015    }
2016
2017    /// Set the checkpoint_in_progress flag in the header.
2018    pub fn set_checkpoint_in_progress(
2019        &self,
2020        in_progress: bool,
2021        target_lsn: u64,
2022    ) -> Result<(), PagerError> {
2023        let _ownership = self.lock_ownership_mutations();
2024        let mut header = self.header_write()?;
2025        header.checkpoint_in_progress = in_progress;
2026        header.checkpoint_target_lsn = target_lsn;
2027        *self.header_dirty_lock()? = true;
2028        drop(header);
2029        self.write_header_and_sync()
2030    }
2031
2032    /// Update the checkpoint LSN and clear the in-progress flag.
2033    pub fn complete_checkpoint(&self, lsn: u64) -> Result<(), PagerError> {
2034        let _ownership = self.lock_ownership_mutations();
2035        let mut header = self.header_write()?;
2036        header.checkpoint_lsn = lsn;
2037        header.checkpoint_in_progress = false;
2038        header.checkpoint_target_lsn = 0;
2039        *self.header_dirty_lock()? = true;
2040        drop(header);
2041        self.write_header_and_sync()
2042    }
2043}
2044
2045#[cfg(test)]
2046mod integrity_guidance_tests {
2047    use super::PagerError;
2048    use std::path::PathBuf;
2049
2050    #[test]
2051    fn damaged_paged_zones_recommend_preserving_files_and_a_verified_backup() {
2052        for error in [
2053            PagerError::SuperblockZoneUnrecoverable(PathBuf::from("data.rdb")),
2054            PagerError::ManifestZoneCorrupt(PathBuf::from("data.rdb")),
2055            PagerError::DwbZoneCorrupt(PathBuf::from("data.rdb")),
2056        ] {
2057            let message = error.to_string();
2058            assert!(message.contains("Preserve the database"), "{message}");
2059            assert!(message.contains("verified backup"), "{message}");
2060            assert!(!message.contains("salvage"), "{message}");
2061            assert!(!message.contains("SCRUB"), "{message}");
2062        }
2063    }
2064}