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
17/// Pager error types
18#[derive(Debug)]
19pub enum PagerError {
20    /// I/O error
21    Io(std::io::Error),
22    /// Page error
23    Page(PageError),
24    /// Invalid database file
25    InvalidDatabase(String),
26    /// Database is read-only
27    ReadOnly,
28    /// Page not found
29    PageNotFound(u32),
30    /// Database is locked
31    Locked,
32    /// A Mutex or RwLock was poisoned (another thread panicked while holding it)
33    LockPoisoned,
34    /// Database is encrypted but no key was supplied.
35    EncryptionRequired,
36    /// Plain (unencrypted) database opened with an encryption key.
37    PlainDatabaseRefusesKey,
38    /// Encryption key validation failed for an encrypted database.
39    InvalidKey,
40    /// Both superblock copies failed validation (ADR 0074 §2).
41    SuperblockZoneUnrecoverable(PathBuf),
42    /// The internal manifest zone failed its checksum (ADR 0074 §2).
43    ManifestZoneCorrupt(PathBuf),
44    /// The in-file double-write zone failed validation (ADR 0074 §2).
45    DwbZoneCorrupt(PathBuf),
46    /// The store still carries a retired phase-1 pager sidecar (ADR 0038 §4).
47    LegacySidecarStore { sidecar: PathBuf },
48    /// The store predates the phase-3 in-file DWB layout marker (ADR 0038 §4).
49    LegacyPagedStore { path: PathBuf },
50}
51
52/// A contiguous run of database pages reserved for vector-turbo payloads.
53#[derive(Debug, Clone, Copy, PartialEq, Eq)]
54pub struct ExtentId {
55    pub start_page: u32,
56    pub n_pages: u32,
57}
58
59impl std::fmt::Display for PagerError {
60    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
61        match self {
62            Self::Io(e) => write!(f, "I/O error: {}", e),
63            Self::Page(e) => write!(f, "Page error: {}", e),
64            Self::InvalidDatabase(msg) => write!(f, "Invalid database: {}", msg),
65            Self::ReadOnly => write!(f, "Database is read-only"),
66            Self::PageNotFound(id) => write!(f, "Page {} not found", id),
67            Self::Locked => write!(f, "Database is locked"),
68            Self::LockPoisoned => write!(f, "Internal lock poisoned (concurrent thread panicked)"),
69            Self::EncryptionRequired => write!(
70                f,
71                "Database is encrypted but no key was supplied (set PagerConfig::encryption)"
72            ),
73            Self::PlainDatabaseRefusesKey => write!(
74                f,
75                "Plain (unencrypted) database opened with an encryption key — refusing"
76            ),
77            Self::InvalidKey => write!(f, "Encryption key validation failed for this database"),
78            Self::SuperblockZoneUnrecoverable(path) => write!(
79                f,
80                "superblock zone of {} is unrecoverable: both ping-pong copies failed \
81                 validation, so no generation can be trusted to root the store. The store \
82                 will not be opened. Recover what survives with red salvage \
83                 (ADR 0074 §4); it reads the damaged file without writing to it and \
84                 reports what it could not verify.",
85                path.display()
86            ),
87            Self::ManifestZoneCorrupt(path) => write!(
88                f,
89                "internal manifest zone of {} failed its checksum: the zone that names \
90                 collections, indexes and the checkpoint boundary cannot be trusted, so no \
91                 rows are returned rather than garbage ones. Run the `SCRUB` statement to classify the fault \
92                 and `red salvage` to extract what survives (ADR 0074 §2/§4).",
93                path.display()
94            ),
95            Self::DwbZoneCorrupt(path) => write!(
96                f,
97                "double-write zone of {} failed validation: page recovery cannot prove \
98                 whether the protected in-place write completed. The store will not be opened \
99                 silently; run the `SCRUB` statement to classify the fault and `red salvage` to extract trusted \
100                 rows (ADR 0074 §2/§4).",
101                path.display()
102            ),
103            Self::LegacySidecarStore { sidecar } => write!(
104                f,
105                "refusing to open a legacy sidecar-backed store: found {}. The superblock \
106                 pair, internal manifest, and double-write recovery state now live inside the \
107                 .rdb file (ADR 0038 §2/§4), \
108                 and the engine never reads the retired sidecars silently. Convert the store \
109                 first with the offline migration tool \
110                 (`reddb_server::pager_zone_migration::migrate_to_zoned`); it is reversible.",
111                sidecar.display()
112            ),
113            Self::LegacyPagedStore { path } => write!(
114                f,
115                "refusing to open a pre-phase-3 paged store: {} does not carry the layout marker \
116                 for the in-file double-write zone. Convert the store first with the offline \
117                 migration tool (`reddb_server::pager_zone_migration::migrate_to_zoned`); it is \
118                 reversible.",
119                path.display()
120            ),
121        }
122    }
123}
124
125impl std::error::Error for PagerError {
126    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
127        match self {
128            Self::Io(e) => Some(e),
129            Self::Page(e) => Some(e),
130            _ => None,
131        }
132    }
133}
134
135impl From<std::io::Error> for PagerError {
136    fn from(e: std::io::Error) -> Self {
137        Self::Io(e)
138    }
139}
140
141impl From<PageError> for PagerError {
142    fn from(e: PageError) -> Self {
143        Self::Page(e)
144    }
145}
146
147impl From<pager_io::PagerIoError> for PagerError {
148    fn from(error: pager_io::PagerIoError) -> Self {
149        match error {
150            pager_io::PagerIoError::Io(error) => Self::Io(error),
151            pager_io::PagerIoError::Page(error) => Self::Page(error),
152            pager_io::PagerIoError::InvalidDatabase(message) => Self::InvalidDatabase(message),
153        }
154    }
155}
156
157/// Host-independent pager configuration. File creation and ownership are host policy.
158#[derive(Debug, Clone)]
159pub struct PagerConfig {
160    pub cache_size: usize,
161    pub read_only: bool,
162    pub verify_checksums: bool,
163    pub double_write: bool,
164}
165
166impl Default for PagerConfig {
167    fn default() -> Self {
168        Self {
169            cache_size: 10_000,
170            read_only: false,
171            verify_checksums: true,
172            double_write: true,
173        }
174    }
175}
176
177/// The host's WAL durability barrier, called before publishing dirty page batches.
178/// Implementations must make all records through `lsn` durable or return an error.
179pub trait WalBarrier {
180    fn flush_until(&self, lsn: u64) -> std::io::Result<()>;
181
182    /// A checkpoint host can preserve the first image of an existing page
183    /// before the cache accepts an overwrite. Newly reserved pages need no undo.
184    fn needs_page_image(&self, _page_id: u32) -> std::io::Result<bool> {
185        Ok(false)
186    }
187
188    /// Must make the image durable before returning. The pager may evict the
189    /// replacement immediately; recording only at flush time is too late.
190    fn capture_page_image(&self, _page: &Page) -> std::io::Result<()> {
191        Ok(())
192    }
193
194    /// Called before pages are removed from allocator ownership.
195    fn reserve_pages(&self, _start: u32, _count: u32) -> std::io::Result<()> {
196        Ok(())
197    }
198
199    /// Called before a persisted free-list trunk is detached from its head.
200    fn detach_free_pages(&self, _pages: &[u32]) -> std::io::Result<()> {
201        Ok(())
202    }
203
204    /// Called before retired pages enter the free list.
205    fn retire_pages(&self, _pages: &[u32]) -> std::io::Result<()> {
206        Ok(())
207    }
208}
209
210/// A pager without a WAL may only publish pages whose LSN is zero.
211pub struct NoWal;
212impl WalBarrier for NoWal {
213    fn flush_until(&self, _lsn: u64) -> std::io::Result<()> {
214        Err(std::io::Error::other(
215            "nonzero page LSN requires a WAL durability barrier",
216        ))
217    }
218}
219pub struct Pager<F: VfsFile, W: WalBarrier> {
220    /// Database file path
221    path: PathBuf,
222    /// File handle
223    file: Mutex<F>,
224    /// Whether mutable pages are protected by the in-file DWB zone.
225    dwb_enabled: bool,
226    /// A failed writeback may leave the only complete image in the DWB.
227    /// Refuse further writes until reopen has recovered that image.
228    writeback_failed: std::sync::atomic::AtomicBool,
229    /// Page cache
230    cache: PageCache,
231    /// Free page list
232    freelist: RwLock<FreeList>,
233    /// Store-managed frees stay unavailable until their removal from the
234    /// durable catalog has crossed a publication boundary.
235    deferred_reclamation: Mutex<DeferredPageReclamation>,
236    /// Excludes page ownership changes while a host publishes an epoch.
237    /// Reentrant so the publishing thread can use ordinary pager methods.
238    ownership_mutations: parking_lot::ReentrantMutex<()>,
239    /// Reused page ownership must reach both superblock copies before a
240    /// manifest can refer to those pages, including single-slot recovery.
241    allocation_reused: std::sync::atomic::AtomicBool,
242    /// A consumed trunk cannot be reused until both header fences succeed.
243    allocation_fence_pending: std::sync::atomic::AtomicBool,
244    /// Database header
245    header: RwLock<DatabaseHeader>,
246    /// Serialize read-select-write sequences for the superblock pair.
247    superblock_publication: Mutex<()>,
248    /// Configuration
249    config: PagerConfig,
250    /// Dirty flag for header
251    header_dirty: Mutex<bool>,
252    wal: W,
253}
254
255#[derive(Default)]
256struct DeferredPageReclamation {
257    enabled: bool,
258    pending: std::collections::BTreeSet<u32>,
259    published: std::collections::BTreeSet<u32>,
260}
261
262/// The superblock zone occupies the head of page 0 (ADR 0038 §2 phase 1).
263const SUPERBLOCK_SLOT_SIZE: usize = reddb_file::PAGED_SUPERBLOCK_SLOT_SIZE;
264const SUPERBLOCK_ZONE_SIZE: usize = reddb_file::PAGED_SUPERBLOCK_ZONE_SIZE;
265// The phase-3 layout was introduced in 1.0.1. Later file format versions
266// must not make existing phase-3 stores look like legacy sidecar stores.
267const PHASE3_DWB_ZONE_FILE_VERSION: u32 = 0x0001_0001;
268
269/// The newest valid superblock copy together with its slot image.
270pub type NewestSuperblock = (
271    reddb_file::PagedSuperblockSelection,
272    [u8; SUPERBLOCK_SLOT_SIZE],
273);
274
275/// Take the first slot's worth of a freshly-built page-0 image. The database
276/// header and the encryption header both live well inside this prefix.
277fn superblock_slot_image(page: &[u8; PAGE_SIZE]) -> [u8; SUPERBLOCK_SLOT_SIZE] {
278    let mut image = [0u8; SUPERBLOCK_SLOT_SIZE];
279    image.copy_from_slice(&page[..SUPERBLOCK_SLOT_SIZE]);
280    image
281}
282
283impl<F: VfsFile, W: WalBarrier> Pager<F, W> {
284    /// Use an already-owned file. The host retains exclusive ownership until
285    /// this pager is dropped, and supplies the native WAL/encryption policy.
286    /// `create` is explicit: a corrupt or empty existing file is never recreated.
287    pub fn open_owned(
288        file: F,
289        path: PathBuf,
290        config: PagerConfig,
291        wal: W,
292        create: bool,
293    ) -> Result<Self, PagerError> {
294        if create && file.file_len()? != 0 {
295            return Err(PagerError::InvalidDatabase(
296                "new pager file is not empty".into(),
297            ));
298        }
299        let pager = Self {
300            path,
301            file: Mutex::new(file),
302            dwb_enabled: config.double_write && !config.read_only,
303            writeback_failed: std::sync::atomic::AtomicBool::new(false),
304            cache: PageCache::new(config.cache_size),
305            freelist: RwLock::new(FreeList::new()),
306            deferred_reclamation: Mutex::new(DeferredPageReclamation::default()),
307            ownership_mutations: parking_lot::ReentrantMutex::new(()),
308            allocation_reused: std::sync::atomic::AtomicBool::new(false),
309            allocation_fence_pending: std::sync::atomic::AtomicBool::new(false),
310            header: RwLock::new(DatabaseHeader::default()),
311            superblock_publication: Mutex::new(()),
312            config,
313            header_dirty: Mutex::new(false),
314            wal,
315        };
316        if create {
317            pager.initialize()?;
318        } else {
319            pager.ensure_phase3_dwb_zone_layout()?;
320            pager.recover_from_dwb()?;
321            pager.load_header()?;
322        }
323        Ok(pager)
324    }
325
326    pub fn wal_barrier(&self) -> &W {
327        &self.wal
328    }
329
330    /// Hold across an entire ownership publication, not just individual
331    /// allocation calls. Readers may continue accessing already-published pages.
332    pub fn lock_ownership_mutations(&self) -> parking_lot::ReentrantMutexGuard<'_, ()> {
333        self.ownership_mutations.lock()
334    }
335
336    /// An interrupted publication must be recovered before any more writes.
337    pub fn fence_interrupted_ownership(&self) {
338        self.writeback_failed
339            .store(true, std::sync::atomic::Ordering::Release);
340    }
341
342    /// Run a host operation while holding the same lock as protected page I/O.
343    /// This is used only for native prefetch and fault-injection tests.
344    #[doc(hidden)]
345    pub fn with_file<T>(&self, operation: impl FnOnce(&mut F) -> T) -> Result<T, PagerError> {
346        Ok(operation(&mut *self.file_lock()?))
347    }
348
349    /// Initialize a new database
350    fn initialize(&self) -> Result<(), PagerError> {
351        if self.config.read_only {
352            return Err(PagerError::ReadOnly);
353        }
354
355        // Create header page. Page ids 1 and 2 are reserved for fixed
356        // metadata/vault pages. ADR 0038 phase 3 reserves the following DWB
357        // zone in-file so normal B-tree allocation starts after it.
358        let initial_page_count = FIRST_ALLOCATABLE_PAGE;
359        let header_page = Page::new_header_page(initial_page_count);
360        self.header_write()?.page_count = initial_page_count;
361
362        // Reserve page 0 (the superblock zone) plus the reserved pages so any
363        // scan over 0..page_count can read every allocated page in a brand-new
364        // database. The zone's own bytes are published below, one slot at a
365        // time; this first whole-page write only sizes the file.
366        self.write_page_raw(0, &header_page)?;
367        let mut metadata_page = Page::new(PageType::Header, 1);
368        metadata_page.update_checksum();
369        self.write_page_raw(1, &metadata_page)?;
370        let mut vault_page = Page::new(PageType::Vault, 2);
371        vault_page.update_checksum();
372        self.write_page_raw(2, &vault_page)?;
373        self.clear_dwb_zone()?;
374
375        // Publish both superblock copies so the ping-pong invariant — one
376        // valid copy always survives a crash — holds from the very first
377        // update onward, not just from the second.
378        let mut slot = superblock_slot_image(header_page.as_bytes());
379        self.write_superblock_slot(&mut slot, 0, 1)?;
380        self.write_superblock_slot(&mut slot, 1, 2)?;
381
382        // Sync to disk
383        self.sync()?;
384
385        Ok(())
386    }
387
388    /// Acquire a write lock on the header RwLock, mapping poison errors.
389    fn header_write(&self) -> Result<std::sync::RwLockWriteGuard<'_, DatabaseHeader>, PagerError> {
390        self.header.write().map_err(|_| PagerError::LockPoisoned)
391    }
392
393    /// Acquire a read lock on the header RwLock, mapping poison errors.
394    fn header_read(&self) -> Result<std::sync::RwLockReadGuard<'_, DatabaseHeader>, PagerError> {
395        self.header.read().map_err(|_| PagerError::LockPoisoned)
396    }
397
398    /// Acquire a write lock on the freelist RwLock, mapping poison errors.
399    fn freelist_write(&self) -> Result<std::sync::RwLockWriteGuard<'_, FreeList>, PagerError> {
400        self.freelist.write().map_err(|_| PagerError::LockPoisoned)
401    }
402
403    /// Acquire a lock on the file Mutex, mapping poison errors.
404    fn file_lock(&self) -> Result<std::sync::MutexGuard<'_, F>, PagerError> {
405        self.file.lock().map_err(|_| PagerError::LockPoisoned)
406    }
407
408    /// Acquire a lock on the header_dirty Mutex, mapping poison errors.
409    fn header_dirty_lock(&self) -> Result<std::sync::MutexGuard<'_, bool>, PagerError> {
410        self.header_dirty
411            .lock()
412            .map_err(|_| PagerError::LockPoisoned)
413    }
414
415    /// Read the whole superblock zone (both ping-pong copies) from the head of
416    /// the file. A file shorter than the zone yields zero-filled tail bytes,
417    /// which decode as "slot never written".
418    fn read_superblock_zone(&self) -> Result<[u8; SUPERBLOCK_ZONE_SIZE], PagerError> {
419        let mut file = self.file_lock()?;
420        Ok(pager_io::read_superblock_zone(&mut *file)?)
421    }
422
423    /// The newest valid superblock copy, with its slot image.
424    ///
425    /// Both copies invalid is the one unrecoverable superblock state (ADR 0074
426    /// §2): the store does not open, and the error names the zone and points
427    /// at salvage rather than panicking or silently re-initialising.
428    #[doc(hidden)]
429    pub fn newest_superblock(&self) -> Result<NewestSuperblock, PagerError> {
430        let zone = self.read_superblock_zone()?;
431        let selection = reddb_file::select_paged_superblock(&zone)
432            .ok_or_else(|| PagerError::SuperblockZoneUnrecoverable(self.path.clone()))?;
433
434        let start = selection.copy_index * SUPERBLOCK_SLOT_SIZE;
435        let mut image = [0u8; SUPERBLOCK_SLOT_SIZE];
436        image.copy_from_slice(&zone[start..start + SUPERBLOCK_SLOT_SIZE]);
437        Ok((selection, image))
438    }
439
440    fn ensure_phase3_dwb_zone_layout(&self) -> Result<(), PagerError> {
441        let (_, image) = self.newest_superblock()?;
442        let header = reddb_file::decode_database_header(&image)
443            .map_err(|err| PagerError::InvalidDatabase(err.to_string()))?;
444        if header.version < PHASE3_DWB_ZONE_FILE_VERSION {
445            return Err(PagerError::LegacyPagedStore {
446                path: self.path.clone(),
447            });
448        }
449        Ok(())
450    }
451
452    /// Seal `slot` for `copy_index`/`generation` and write exactly that slot,
453    /// then fsync. The caller guarantees `copy_index` is the stale copy.
454    fn write_superblock_slot(
455        &self,
456        slot: &mut [u8; SUPERBLOCK_SLOT_SIZE],
457        copy_index: usize,
458        generation: u64,
459    ) -> Result<(), PagerError> {
460        self.ensure_writeback_available()?;
461        if self.config.read_only {
462            return Err(PagerError::ReadOnly);
463        }
464        let mut file = self.file_lock()?;
465        self.ensure_writeback_available()?;
466        Ok(pager_io::write_superblock_slot(
467            &mut *file, slot, copy_index, generation,
468        )?)
469    }
470
471    /// Publish `mutate`'s edit of the newest slot image into the *stale* copy.
472    ///
473    /// Reading the newest image first means every byte the pager does not
474    /// explicitly rewrite — most importantly the encryption marker and header —
475    /// carries forward untouched.
476    #[doc(hidden)]
477    pub fn commit_superblock(
478        &self,
479        mutate: impl FnOnce(&mut [u8; SUPERBLOCK_SLOT_SIZE]) -> Result<(), PagerError>,
480    ) -> Result<(), PagerError> {
481        let _ownership = self.lock_ownership_mutations();
482        let _publication = self
483            .superblock_publication
484            .lock()
485            .map_err(|_| PagerError::LockPoisoned)?;
486        self.commit_superblock_locked(mutate)
487    }
488
489    fn commit_superblock_locked(
490        &self,
491        mutate: impl FnOnce(&mut [u8; SUPERBLOCK_SLOT_SIZE]) -> Result<(), PagerError>,
492    ) -> Result<(), PagerError> {
493        if self.config.read_only {
494            return Err(PagerError::ReadOnly);
495        }
496        let (newest, mut image) = self.newest_superblock()?;
497        mutate(&mut image)?;
498        let next_copy = reddb_file::paged_superblock_next_copy(Some(newest));
499        self.write_superblock_slot(&mut image, next_copy, newest.generation.saturating_add(1))
500    }
501
502    /// Fence a file-format upgrade in both recoverable superblock copies.
503    ///
504    /// The caller must exclude other mutators while the fence is in progress
505    /// and may publish a new mandatory format only after this succeeds. A
506    /// failed write fences subsequent mutation until reopen and recovery.
507    pub fn fence_file_format_version(&self, version: u32) -> Result<(), PagerError> {
508        let _ownership = self.lock_ownership_mutations();
509        self.ensure_writeback_available()?;
510        if self.config.read_only {
511            return Err(PagerError::ReadOnly);
512        }
513        if !(PHASE3_DWB_ZONE_FILE_VERSION..=reddb_file::PAGE_FILE_VERSION).contains(&version) {
514            return Err(PagerError::InvalidDatabase(
515                "unsupported file format fence".into(),
516            ));
517        }
518        let _publication = self
519            .superblock_publication
520            .lock()
521            .map_err(|_| PagerError::LockPoisoned)?;
522        if version < self.header_read()?.version {
523            return Err(PagerError::InvalidDatabase(
524                "file format downgrade is not supported".into(),
525            ));
526        }
527        let result = (|| {
528            for _ in 0..reddb_file::PAGED_SUPERBLOCK_SLOT_COUNT {
529                self.commit_superblock_locked(|image| {
530                    reddb_file::set_database_header_version(image, version)
531                        .map_err(|error| PagerError::InvalidDatabase(error.to_string()))
532                })?;
533            }
534            self.header_write()?.version = version;
535            Ok(())
536        })();
537        if result.is_err() {
538            self.writeback_failed
539                .store(true, std::sync::atomic::Ordering::Release);
540        }
541        result
542    }
543
544    /// Load the database header from the newest valid superblock copy.
545    fn load_header(&self) -> Result<(), PagerError> {
546        let (_, image) = self.newest_superblock()?;
547
548        let mut decoded_header = reddb_file::decode_database_header(&image)
549            .map_err(|err| PagerError::InvalidDatabase(err.to_string()))?;
550        // Cache eviction (or DWB recovery) can persist pages before the next
551        // superblock publication. After a crash the allocation counter may
552        // therefore lag behind live pages already in the file. Reserve the
553        // entire physical extent before allocating; never overwrite that tail.
554        // A partial last page is reserved too, without treating it as verified.
555        let physical_pages = self.file_lock()?.file_len()?.div_ceil(PAGE_SIZE as u64);
556        let physical_pages = u32::try_from(physical_pages).map_err(|_| {
557            PagerError::InvalidDatabase("data file exceeds the page address space".into())
558        })?;
559        if decoded_header.page_count < physical_pages {
560            decoded_header.page_count = physical_pages;
561            if !self.config.read_only {
562                *self.header_dirty_lock()? = true;
563            }
564        }
565        let freelist_head = decoded_header.freelist_head;
566
567        {
568            let mut header = self.header_write()?;
569            *header = decoded_header;
570        }
571
572        // Initialize freelist
573        {
574            let mut freelist = self.freelist_write()?;
575            *freelist = FreeList::from_header(freelist_head, 0);
576        }
577
578        Ok(())
579    }
580
581    /// Publish the in-memory database header as a new superblock generation.
582    fn write_header(&self) -> Result<(), PagerError> {
583        if self.config.read_only {
584            return Err(PagerError::ReadOnly);
585        }
586
587        let header = self.header_read()?.clone();
588        self.commit_superblock(|image| {
589            reddb_file::encode_database_header(image, &header)
590                .map_err(|err| PagerError::InvalidDatabase(err.to_string()))
591        })?;
592        *self.header_dirty_lock()? = false;
593
594        Ok(())
595    }
596
597    /// Read a page from disk (bypassing cache)
598    fn read_page_raw(&self, page_id: u32) -> Result<Page, PagerError> {
599        let mut file = self.file_lock()?;
600        Ok(pager_io::read_page(
601            &mut *file,
602            page_id,
603            self.config.verify_checksums,
604        )?)
605    }
606
607    fn ensure_writeback_available(&self) -> Result<(), PagerError> {
608        if self
609            .writeback_failed
610            .load(std::sync::atomic::Ordering::Acquire)
611        {
612            return Err(PagerError::Io(std::io::Error::other(
613                "page writeback failed; reopen the database to recover before writing",
614            )));
615        }
616        Ok(())
617    }
618
619    fn record_ownership(
620        &self,
621        operation: impl FnOnce(&W) -> std::io::Result<()>,
622    ) -> Result<(), PagerError> {
623        if let Err(error) = operation(&self.wal) {
624            self.writeback_failed
625                .store(true, std::sync::atomic::Ordering::Release);
626            return Err(PagerError::Io(error));
627        }
628        Ok(())
629    }
630
631    /// Amortize protected eviction over one bounded DWB frame. Keeping the
632    /// remaining cache dirty would otherwise pay three fsyncs per victim.
633    fn write_evicted_dirty_page(&self, page: Page) -> Result<(), PagerError> {
634        let mut ids = self
635            .cache
636            .dirty_page_ids(pager_io::DWB_MAX_PAGES_PER_FRAME.saturating_sub(1));
637        ids.sort_unstable();
638        let mut pages = self.cache.flush_dirty_pages(&ids);
639        pages.push((page.page_id(), page));
640        pages.sort_unstable_by_key(|(id, _)| *id);
641        self.write_dirty_page_batch(&pages)
642    }
643
644    /// Write a page to disk (bypassing cache)
645    fn write_page_raw(&self, page_id: u32, page: &Page) -> Result<(), PagerError> {
646        self.ensure_writeback_available()?;
647        if self.config.read_only {
648            return Err(PagerError::ReadOnly);
649        }
650        let mut file = self.file_lock()?;
651        self.ensure_writeback_available()?;
652        Ok(pager_io::write_page(&mut *file, page_id, page)?)
653    }
654
655    /// Read a page (cache-aware)
656    pub fn read_page(&self, page_id: u32) -> Result<Page, PagerError> {
657        // Check cache first
658        if let Some(page) = self.cache.get(page_id) {
659            return Ok(page);
660        }
661
662        // Cache miss - read from disk
663        let page = self.read_page_raw(page_id)?;
664
665        // Add to cache
666        if let Some(dirty_page) = self.cache.insert(page_id, page.clone()) {
667            // Evicted page was dirty, need to write it back
668            self.write_evicted_dirty_page(dirty_page)?;
669        }
670
671        Ok(page)
672    }
673
674    /// Read a page without verifying checksum (for encrypted pages)
675    ///
676    /// Use this when the page content has its own integrity protection
677    /// (e.g., AES-GCM authentication tag for encrypted pages).
678    pub fn read_page_no_checksum(&self, page_id: u32) -> Result<Page, PagerError> {
679        // Check cache first
680        if let Some(page) = self.cache.get(page_id) {
681            return Ok(page);
682        }
683
684        // Cache miss - use the same byte I/O without a page CRC check.
685        let mut file = self.file_lock()?;
686        let page = pager_io::read_page(&mut *file, page_id, false)?;
687        drop(file);
688
689        // Add to cache (no checksum verification)
690        if let Some(dirty_page) = self.cache.insert(page_id, page.clone()) {
691            // Evicted page was dirty, need to write it back
692            self.write_evicted_dirty_page(dirty_page)?;
693        }
694
695        Ok(page)
696    }
697
698    /// Write a page (cache-aware)
699    pub fn write_page(&self, page_id: u32, mut page: Page) -> Result<(), PagerError> {
700        let _ownership = self.lock_ownership_mutations();
701        self.ensure_writeback_available()?;
702        if self.config.read_only {
703            return Err(PagerError::ReadOnly);
704        }
705
706        self.capture_page_before_write(page_id)?;
707
708        // Update checksum
709        page.update_checksum();
710
711        // Add to cache and mark dirty. If the cache had to evict a
712        // dirty entry, write it through immediately — the evicted
713        // page will never be flushed otherwise (same bug fixed in
714        // `allocate_page`).
715        if let Some(dirty_page) = self.cache.insert(page_id, page) {
716            self.write_evicted_dirty_page(dirty_page)?;
717        }
718        self.cache.mark_dirty(page_id);
719
720        Ok(())
721    }
722
723    /// Write a page without updating checksum (for encrypted pages)
724    ///
725    /// Use this when the page content has its own integrity protection
726    /// (e.g., AES-GCM authentication tag for encrypted pages).
727    pub fn write_page_no_checksum(&self, page_id: u32, page: Page) -> Result<(), PagerError> {
728        let _ownership = self.lock_ownership_mutations();
729        self.ensure_writeback_available()?;
730        if self.config.read_only {
731            return Err(PagerError::ReadOnly);
732        }
733
734        self.capture_page_before_write(page_id)?;
735
736        // Add to cache and mark dirty (no checksum update). Same
737        // eviction-write-through guard as `write_page`.
738        if let Some(dirty_page) = self.cache.insert(page_id, page) {
739            self.write_evicted_dirty_page(dirty_page)?;
740        }
741        self.cache.mark_dirty(page_id);
742
743        Ok(())
744    }
745
746    fn capture_page_before_write(&self, page_id: u32) -> Result<(), PagerError> {
747        let mut required = false;
748        self.record_ownership(|wal| {
749            required = wal.needs_page_image(page_id)?;
750            Ok(())
751        })?;
752        if required {
753            let previous = self.read_page(page_id)?;
754            self.record_ownership(|wal| wal.capture_page_image(&previous))?;
755        }
756        Ok(())
757    }
758
759    /// Allocate a new page
760    pub fn allocate_page(&self, page_type: PageType) -> Result<Page, PagerError> {
761        let _ownership = self.lock_ownership_mutations();
762        self.ensure_writeback_available()?;
763        if self.config.read_only {
764            return Err(PagerError::ReadOnly);
765        }
766
767        // Try to get from freelist first
768        let page_id = {
769            let mut freelist = self.freelist_write()?;
770            // A previous fence can fail after loading the trunk into memory.
771            // Retry it before the fast path can consume any of those pages.
772            self.complete_allocation_fence()?;
773            if let Some(id) = freelist.next_in_memory_page() {
774                self.record_ownership(|wal| wal.reserve_pages(id, 1))?;
775                let id = freelist
776                    .allocate()
777                    .expect("free page checked while allocator is locked");
778                self.allocation_reused
779                    .store(true, std::sync::atomic::Ordering::Release);
780                id
781            } else if freelist.trunk_head() != 0 {
782                let trunk_id = freelist.trunk_head();
783
784                let trunk = self.read_page(trunk_id).map_err(|e| match e {
785                    PagerError::PageNotFound(_) => {
786                        PagerError::InvalidDatabase("Freelist trunk missing".to_string())
787                    }
788                    other => other,
789                })?;
790                if trunk.page_id() != trunk_id {
791                    return Err(PagerError::InvalidDatabase(
792                        "misdirected freelist trunk".into(),
793                    ));
794                }
795
796                // Persist ownership of every page detached from the trunk
797                // before either superblock stops naming that trunk.
798                let mut detached = FreeList::new();
799                detached
800                    .load_from_trunk(&trunk)
801                    .map_err(|e| PagerError::InvalidDatabase(format!("Freelist: {e}")))?;
802                let mut pages = Vec::new();
803                let mut seen = std::collections::HashSet::new();
804                let page_count = self.page_count()?;
805                while let Some(id) = detached.allocate() {
806                    if id < FIRST_ALLOCATABLE_PAGE || id >= page_count || !seen.insert(id) {
807                        return Err(PagerError::InvalidDatabase(
808                            "invalid freelist trunk page".into(),
809                        ));
810                    }
811                    pages.push(id);
812                }
813                self.record_ownership(|wal| wal.detach_free_pages(&pages))?;
814
815                freelist
816                    .load_from_trunk(&trunk)
817                    .map_err(|e| PagerError::InvalidDatabase(format!("Freelist: {}", e)))?;
818                let mut header = self.header_write()?;
819                header.freelist_head = freelist.trunk_head();
820                *self.header_dirty_lock()? = true;
821                drop(header);
822                // The consumed trunk itself becomes reusable. Forget its old
823                // allocator role durably before cache eviction can overwrite it.
824                // Keep the allocator locked so another allocator cannot race
825                // ahead of this fence. A failed fence leaves allocation stopped.
826                self.allocation_fence_pending
827                    .store(true, std::sync::atomic::Ordering::Release);
828                self.complete_allocation_fence()?;
829                let next = freelist.next_in_memory_page().ok_or_else(|| {
830                    PagerError::InvalidDatabase("Freelist empty after trunk load".to_string())
831                })?;
832                self.record_ownership(|wal| wal.reserve_pages(next, 1))?;
833                let id = freelist.allocate().ok_or_else(|| {
834                    PagerError::InvalidDatabase("Freelist empty after trunk load".to_string())
835                })?;
836                self.allocation_reused
837                    .store(true, std::sync::atomic::Ordering::Release);
838                id
839            } else {
840                // No free pages, extend file
841                let mut header = self.header_write()?;
842                let id = header.page_count;
843                let next = id
844                    .checked_add(1)
845                    .ok_or_else(|| PagerError::InvalidDatabase("page count overflow".into()))?;
846                self.record_ownership(|wal| wal.reserve_pages(id, 1))?;
847                header.page_count = next;
848                *self.header_dirty_lock()? = true;
849                id
850            }
851        };
852
853        let page = Page::new(page_type, page_id);
854
855        // Write to cache. The evicted page (if any) is dirty by
856        // definition — `cache.insert` only returns `Some` when it
857        // had to evict a dirty entry to make room. The previous
858        // version dropped that return value, which silently lost
859        // writes whenever a freshly-allocated page caused an LRU
860        // eviction. This shows up under heavy ingest as
861        // "B-tree insert error: Pager error: I/O error: failed to
862        // fill whole buffer" later, when something tries to read
863        // back the never-flushed page. Mirror `read_page`'s
864        // handling: write the evicted page through immediately so
865        // the on-disk image stays consistent.
866        if let Some(dirty_page) = self.cache.insert(page_id, page.clone()) {
867            self.write_evicted_dirty_page(dirty_page)?;
868        }
869        self.cache.mark_dirty(page_id);
870
871        Ok(page)
872    }
873
874    /// Reserve a contiguous extent of vector pages.
875    pub fn reserve_contig_extent(&self, n_pages: u32) -> Result<ExtentId, PagerError> {
876        let _ownership = self.lock_ownership_mutations();
877        self.ensure_writeback_available()?;
878        if self.config.read_only {
879            return Err(PagerError::ReadOnly);
880        }
881        if n_pages == 0 {
882            return Err(PagerError::InvalidDatabase(
883                "contiguous extent must reserve at least one page".to_string(),
884            ));
885        }
886
887        let start_page = {
888            let mut header = self.header_write()?;
889            let start = header.page_count;
890            let end = header.page_count.checked_add(n_pages).ok_or_else(|| {
891                PagerError::InvalidDatabase("contiguous extent page count overflow".to_string())
892            })?;
893            self.record_ownership(|wal| wal.reserve_pages(start, n_pages))?;
894            header.page_count = end;
895            *self.header_dirty_lock()? = true;
896            start
897        };
898
899        for page_id in start_page..start_page + n_pages {
900            let mut page = Page::new(PageType::Vector, page_id);
901            page.update_checksum();
902            if let Some(dirty_page) = self.cache.insert(page_id, page) {
903                self.write_evicted_dirty_page(dirty_page)?;
904            }
905            self.cache.mark_dirty(page_id);
906        }
907
908        Ok(ExtentId {
909            start_page,
910            n_pages,
911        })
912    }
913
914    /// Free a page (return to freelist)
915    pub fn free_page(&self, page_id: u32) -> Result<(), PagerError> {
916        let _ownership = self.lock_ownership_mutations();
917        self.ensure_writeback_available()?;
918        if self.config.read_only {
919            return Err(PagerError::ReadOnly);
920        }
921
922        let mut deferred = self
923            .deferred_reclamation
924            .lock()
925            .map_err(|_| PagerError::LockPoisoned)?;
926        if deferred.enabled {
927            if page_id < FIRST_ALLOCATABLE_PAGE || page_id >= self.page_count()? {
928                return Err(PagerError::InvalidDatabase(
929                    "invalid deferred free page".into(),
930                ));
931            }
932            self.record_ownership(|wal| wal.retire_pages(&[page_id]))?;
933            // Keep dirty forwarding/empty nodes until publication. An
934            // incidental flush must leave the previous root traversable.
935            deferred.pending.insert(page_id);
936            return Ok(());
937        }
938        drop(deferred);
939
940        self.record_ownership(|wal| wal.retire_pages(&[page_id]))?;
941
942        // Remove from cache
943        self.cache.remove(page_id);
944
945        // Add to freelist
946        let mut freelist = self.freelist_write()?;
947        freelist.free(page_id);
948
949        *self.header_dirty_lock()? = true;
950
951        Ok(())
952    }
953
954    pub fn enable_deferred_reclamation(&self) -> Result<(), PagerError> {
955        let _ownership = self.lock_ownership_mutations();
956        self.deferred_reclamation
957            .lock()
958            .map_err(|_| PagerError::LockPoisoned)?
959            .enabled = true;
960        Ok(())
961    }
962
963    pub fn deferred_reclamation_pages(&self) -> Result<(Vec<u32>, Vec<u32>), PagerError> {
964        let deferred = self
965            .deferred_reclamation
966            .lock()
967            .map_err(|_| PagerError::LockPoisoned)?;
968        Ok((
969            deferred.pending.iter().copied().collect(),
970            deferred.published.iter().copied().collect(),
971        ))
972    }
973
974    pub fn restore_deferred_reclamation(&self, pages: &[u32]) -> Result<(), PagerError> {
975        let _ownership = self.lock_ownership_mutations();
976        let count = self.page_count()?;
977        if pages
978            .iter()
979            .any(|id| *id < FIRST_ALLOCATABLE_PAGE || *id >= count)
980        {
981            return Err(PagerError::InvalidDatabase(
982                "invalid persisted deferred free page".into(),
983            ));
984        }
985        let mut deferred = self
986            .deferred_reclamation
987            .lock()
988            .map_err(|_| PagerError::LockPoisoned)?;
989        deferred.pending.extend(pages);
990        deferred.published.extend(pages);
991        Ok(())
992    }
993
994    /// The caller has durably published a catalog that retains `retained`
995    /// and no longer names `released`, after waiting for old tree readers.
996    pub fn complete_deferred_publication(
997        &self,
998        retained: &[u32],
999        released: &[u32],
1000    ) -> Result<(), PagerError> {
1001        let _ownership = self.lock_ownership_mutations();
1002        let mut deferred = self
1003            .deferred_reclamation
1004            .lock()
1005            .map_err(|_| PagerError::LockPoisoned)?;
1006        if !released.is_empty() {
1007            self.free_retired_pages(released)?;
1008        }
1009        for id in released {
1010            deferred.pending.remove(id);
1011        }
1012        deferred.published = retained.iter().copied().collect();
1013        Ok(())
1014    }
1015
1016    /// Record the pages whose catalog ownership is about to be retired.
1017    pub fn record_retirement_intent(&self, pages: &[u32]) -> Result<(), PagerError> {
1018        let _ownership = self.lock_ownership_mutations();
1019        self.ensure_writeback_available()?;
1020        if self.config.read_only {
1021            return Err(PagerError::ReadOnly);
1022        }
1023        let page_count = self.page_count()?;
1024        let mut seen = std::collections::HashSet::with_capacity(pages.len());
1025        if pages
1026            .iter()
1027            .any(|id| *id < FIRST_ALLOCATABLE_PAGE || *id >= page_count || !seen.insert(*id))
1028        {
1029            return Err(PagerError::InvalidDatabase(
1030                "invalid retirement intent".into(),
1031            ));
1032        }
1033        self.record_ownership(|wal| wal.retire_pages(pages))
1034    }
1035
1036    /// Return a fully validated, unpublished tree generation in one allocator
1037    /// mutation. Call only after its removal from the manifest is durable.
1038    pub fn free_retired_pages(&self, pages: &[u32]) -> Result<(), PagerError> {
1039        let _ownership = self.lock_ownership_mutations();
1040        self.ensure_writeback_available()?;
1041        if self.config.read_only {
1042            return Err(PagerError::ReadOnly);
1043        }
1044        let page_count = self.page_count()?;
1045        if pages
1046            .iter()
1047            .any(|id| *id < FIRST_ALLOCATABLE_PAGE || *id >= page_count)
1048        {
1049            return Err(PagerError::InvalidDatabase(
1050                "invalid retired page range".into(),
1051            ));
1052        }
1053        let mut seen = std::collections::HashSet::with_capacity(pages.len());
1054        if pages.iter().any(|id| !seen.insert(*id)) {
1055            return Err(PagerError::InvalidDatabase("duplicate retired page".into()));
1056        }
1057        let mut freelist = self.freelist_write()?;
1058        if freelist.contains_any(&seen) {
1059            return Err(PagerError::InvalidDatabase(
1060                "retired page is already free".into(),
1061            ));
1062        }
1063        // Some free pages remain in on-disk trunks until allocation reaches
1064        // them. Check those too before changing the cache or the allocator.
1065        // The allocator lock prevents concurrent consumption or publication.
1066        let mut trunk_id = freelist.trunk_head();
1067        let mut visited = std::collections::HashSet::new();
1068        while trunk_id != 0 && !seen.is_empty() {
1069            if trunk_id < FIRST_ALLOCATABLE_PAGE
1070                || trunk_id >= page_count
1071                || !visited.insert(trunk_id)
1072            {
1073                return Err(PagerError::InvalidDatabase(
1074                    "invalid freelist trunk chain".into(),
1075                ));
1076            }
1077            let trunk = self.read_page(trunk_id)?;
1078            if trunk.page_id() != trunk_id {
1079                return Err(PagerError::InvalidDatabase(
1080                    "misdirected freelist trunk".into(),
1081                ));
1082            }
1083            let mut loaded = FreeList::new();
1084            let next = loaded
1085                .load_from_trunk(&trunk)
1086                .map_err(|error| PagerError::InvalidDatabase(format!("Freelist: {error}")))?;
1087            while let Some(id) = loaded.allocate() {
1088                if id < FIRST_ALLOCATABLE_PAGE || id >= page_count {
1089                    return Err(PagerError::InvalidDatabase(
1090                        "invalid free page range".into(),
1091                    ));
1092                }
1093                if seen.contains(&id) {
1094                    return Err(PagerError::InvalidDatabase(
1095                        "retired page is already free".into(),
1096                    ));
1097                }
1098            }
1099            trunk_id = next.unwrap_or(0);
1100        }
1101        self.record_ownership(|wal| wal.retire_pages(pages))?;
1102        for id in pages {
1103            self.cache.remove(*id);
1104        }
1105        freelist.free_batch(pages);
1106        *self.header_dirty_lock()? = true;
1107        Ok(())
1108    }
1109
1110    fn pages_not_in_freelist(
1111        &self,
1112        pages: &[u32],
1113        freelist: &FreeList,
1114        page_count: u32,
1115    ) -> Result<std::collections::HashSet<u32>, PagerError> {
1116        let mut remaining = std::collections::HashSet::with_capacity(pages.len());
1117        if pages
1118            .iter()
1119            .any(|id| *id < FIRST_ALLOCATABLE_PAGE || *id >= page_count || !remaining.insert(*id))
1120        {
1121            return Err(PagerError::InvalidDatabase(
1122                "invalid recovery page set".into(),
1123            ));
1124        }
1125        let requested = remaining.clone();
1126        freelist
1127            .remove_existing_from(&requested, &mut remaining)
1128            .map_err(|error| PagerError::InvalidDatabase(format!("Freelist: {error}")))?;
1129        let mut trunk_id = freelist.trunk_head();
1130        let mut visited = std::collections::HashSet::new();
1131        while trunk_id != 0 {
1132            if trunk_id < FIRST_ALLOCATABLE_PAGE
1133                || trunk_id >= page_count
1134                || !visited.insert(trunk_id)
1135            {
1136                return Err(PagerError::InvalidDatabase(
1137                    "invalid freelist trunk chain".into(),
1138                ));
1139            }
1140            let trunk = self.read_page(trunk_id)?;
1141            if trunk.page_id() != trunk_id {
1142                return Err(PagerError::InvalidDatabase(
1143                    "misdirected freelist trunk".into(),
1144                ));
1145            }
1146            let mut loaded = FreeList::new();
1147            let next = loaded
1148                .load_from_trunk(&trunk)
1149                .map_err(|error| PagerError::InvalidDatabase(format!("Freelist: {error}")))?;
1150            while let Some(id) = loaded.allocate() {
1151                if id < FIRST_ALLOCATABLE_PAGE || id >= page_count {
1152                    return Err(PagerError::InvalidDatabase(
1153                        "invalid free page range".into(),
1154                    ));
1155                }
1156                if requested.contains(&id) && !remaining.remove(&id) {
1157                    return Err(PagerError::InvalidDatabase(
1158                        "duplicate recovery page".into(),
1159                    ));
1160                }
1161            }
1162            trunk_id = next.unwrap_or(0);
1163        }
1164        Ok(remaining)
1165    }
1166
1167    /// Add proven-unowned pages to the allocator exactly once during recovery.
1168    /// The caller must exclude all allocation and publication until this state
1169    /// has been synced to both superblocks and the journal is completed.
1170    pub fn recover_unowned_pages(&self, pages: &[u32]) -> Result<(), PagerError> {
1171        let _ownership = self.lock_ownership_mutations();
1172        self.ensure_writeback_available()?;
1173        if self.config.read_only {
1174            return Err(PagerError::ReadOnly);
1175        }
1176        let page_count = self.page_count()?;
1177        let mut freelist = self.freelist_write()?;
1178        let remaining = self.pages_not_in_freelist(pages, &freelist, page_count)?;
1179        if !remaining.is_empty() {
1180            let mut missing: Vec<_> = remaining.into_iter().collect();
1181            missing.sort_unstable();
1182            for id in &missing {
1183                self.cache.remove(*id);
1184            }
1185            freelist.free_batch(&missing);
1186            *self.header_dirty_lock()? = true;
1187        }
1188        Ok(())
1189    }
1190
1191    /// Reject a recovery plan if any page it must preserve is already free.
1192    /// The caller must exclude allocation and freelist mutation during recovery.
1193    pub fn validate_owned_pages(&self, pages: &[u32]) -> Result<(), PagerError> {
1194        let page_count = self.page_count()?;
1195        let freelist = self.freelist_write()?;
1196        let absent = self.pages_not_in_freelist(pages, &freelist, page_count)?;
1197        if absent.len() != pages.len() {
1198            return Err(PagerError::InvalidDatabase(
1199                "owned recovery page is free".into(),
1200            ));
1201        }
1202        Ok(())
1203    }
1204
1205    /// Persist the current allocator in both fallback copies during recovery.
1206    /// The caller must exclude allocation until journal completion is durable.
1207    pub fn sync_recovered_allocator(&self) -> Result<(), PagerError> {
1208        let _ownership = self.lock_ownership_mutations();
1209        let result = (|| {
1210            self.sync()?;
1211            let _publication = self
1212                .superblock_publication
1213                .lock()
1214                .map_err(|_| PagerError::LockPoisoned)?;
1215            let header = self.header_read()?.clone();
1216            for _ in 0..reddb_file::PAGED_SUPERBLOCK_SLOT_COUNT {
1217                self.commit_superblock_locked(|image| {
1218                    reddb_file::encode_database_header(image, &header)
1219                        .map_err(|error| PagerError::InvalidDatabase(error.to_string()))
1220                })?;
1221            }
1222            Ok(())
1223        })();
1224        if result.is_err() {
1225            self.writeback_failed
1226                .store(true, std::sync::atomic::Ordering::Release);
1227        }
1228        result
1229    }
1230
1231    /// Get database header
1232    /// Pages available to the allocator, including pages on unloaded trunks.
1233    pub fn free_page_count(&self) -> Result<u32, PagerError> {
1234        Ok(self.freelist_write()?.total_free())
1235    }
1236
1237    pub fn header(&self) -> Result<DatabaseHeader, PagerError> {
1238        Ok(self.header_read()?.clone())
1239    }
1240
1241    pub fn physical_header(&self) -> Result<PhysicalFileHeader, PagerError> {
1242        Ok(self.header_read()?.physical)
1243    }
1244
1245    pub fn update_physical_header(&self, physical: PhysicalFileHeader) -> Result<(), PagerError> {
1246        let _ownership = self.lock_ownership_mutations();
1247        self.ensure_writeback_available()?;
1248        if self.config.read_only {
1249            return Err(PagerError::ReadOnly);
1250        }
1251
1252        let mut header = self.header_write()?;
1253        header.physical = physical;
1254        *self.header_dirty_lock()? = true;
1255        Ok(())
1256    }
1257
1258    /// Get page count
1259    pub fn page_count(&self) -> Result<u32, PagerError> {
1260        Ok(self.header_read()?.page_count)
1261    }
1262
1263    /// Flush all dirty pages to disk
1264    pub fn flush(&self) -> Result<(), PagerError> {
1265        let _ownership = self.lock_ownership_mutations();
1266        self.ensure_writeback_available()?;
1267        if self.config.read_only {
1268            return Ok(());
1269        }
1270
1271        // Persist freelist to trunk pages when dirty
1272        let trunks = {
1273            let mut freelist = self.freelist_write()?;
1274            if freelist.is_dirty() {
1275                let mut header = self.header_write()?;
1276                let trunks = freelist.flush_to_trunks(0, || {
1277                    let id = header.page_count;
1278                    header.page_count += 1;
1279                    id
1280                });
1281                header.freelist_head = freelist.trunk_head();
1282                *self.header_dirty_lock()? = true;
1283                freelist.mark_clean();
1284                trunks
1285            } else {
1286                Vec::new()
1287            }
1288        };
1289
1290        let mut dirty_pages = Vec::new();
1291        for trunk in trunks {
1292            let page_id = trunk.page_id();
1293            if let Some(evicted) = self.cache.insert(page_id, trunk) {
1294                // Trunk publication can evict dirty data. Keep it in the same
1295                // WAL-first, double-write-protected batch as resident pages.
1296                dirty_pages.push((evicted.page_id(), evicted));
1297            }
1298            self.cache.mark_dirty(page_id);
1299        }
1300
1301        // IDs are compact; page images are not. At 16 KiB per page, staging
1302        // 256 images plus one DWB frame uses about 5 MiB, independently of
1303        // the dirty cache size. Publish the header only after every batch.
1304        const FLUSH_BATCH_PAGES: usize = 256;
1305        for batch in dirty_pages.chunks(FLUSH_BATCH_PAGES) {
1306            self.write_dirty_page_batch(batch)?;
1307        }
1308        drop(dirty_pages);
1309        // Batch dirty IDs, not every resident ID: otherwise a few edits in
1310        // a mostly clean cache pay a protected write per cache-sized slice.
1311        // Physical order also avoids arbitrary seeks among each batch's homes.
1312        let mut page_ids = self.cache.dirty_page_ids(usize::MAX);
1313        page_ids.sort_unstable();
1314        for ids in page_ids.chunks(FLUSH_BATCH_PAGES) {
1315            let pages = self.cache.flush_dirty_pages(ids);
1316            self.write_dirty_page_batch(&pages)?;
1317        }
1318
1319        // Write header if dirty
1320        if *self.header_dirty_lock()? {
1321            self.write_header()?;
1322        }
1323
1324        Ok(())
1325    }
1326
1327    fn write_dirty_page_batch(&self, pages: &[(u32, Page)]) -> Result<(), PagerError> {
1328        if pages.is_empty() {
1329            return Ok(());
1330        }
1331        // Every image's WAL must be durable before its home write, including
1332        // dirty data evicted while publishing freelist trunks. LSN zero is
1333        // reserved for pages not produced by a WAL append.
1334        let result = (|| {
1335            let max_lsn = pages
1336                .iter()
1337                .filter_map(|(_, page)| page.header().ok().map(|header| header.lsn))
1338                .max()
1339                .unwrap_or(0);
1340            if max_lsn > 0 {
1341                self.wal.flush_until(max_lsn)?;
1342            }
1343            self.write_pages_through_dwb(pages)
1344        })();
1345        if result.is_err() {
1346            self.writeback_failed
1347                .store(true, std::sync::atomic::Ordering::Release);
1348            for (page_id, _) in pages {
1349                self.cache.mark_dirty(*page_id);
1350            }
1351        }
1352        result
1353    }
1354
1355    /// Sync file to disk (fsync)
1356    pub fn sync(&self) -> Result<(), PagerError> {
1357        let _ownership = self.lock_ownership_mutations();
1358        self.flush()?;
1359
1360        if self
1361            .allocation_reused
1362            .swap(false, std::sync::atomic::Ordering::AcqRel)
1363        {
1364            // flush published the current allocator in one slot. Mirror that
1365            // ownership before callers publish roots that use reclaimed pages;
1366            // recovery through the sibling slot must never allocate them again.
1367            if let Err(error) = self.write_header_and_sync() {
1368                self.allocation_reused
1369                    .store(true, std::sync::atomic::Ordering::Release);
1370                return Err(error);
1371            }
1372        }
1373
1374        let file = self.file_lock()?;
1375        file.sync_all()?;
1376
1377        Ok(())
1378    }
1379
1380    /// Get cache statistics
1381    pub fn cache_stats(&self) -> super::page_cache::CacheStats {
1382        self.cache.stats()
1383    }
1384
1385    /// Pages currently resident in the page cache. Slots are fixed size, so
1386    /// this times the page size is the cache's memory footprint (ADR 0073 §2).
1387    pub fn cache_len(&self) -> usize {
1388        self.cache.len()
1389    }
1390
1391    /// Preallocated page-cache slots — the sizing this process derived from
1392    /// its page-cache budget share.
1393    pub fn cache_capacity(&self) -> usize {
1394        self.cache.capacity()
1395    }
1396
1397    /// Count dirty pages currently in the page cache.
1398    pub fn dirty_page_count(&self) -> usize {
1399        self.cache.dirty_count()
1400    }
1401
1402    /// Estimated fraction of the page cache holding dirty pages.
1403    /// Returned in `[0, 1]`. Used by the background writer to
1404    /// decide when to kick in aggressive flushing.
1405    pub fn dirty_fraction(&self) -> f64 {
1406        let capacity = self.cache.capacity().max(1) as f64;
1407        self.cache.dirty_count() as f64 / capacity
1408    }
1409
1410    /// Flush up to `max` dirty pages from the cache. Returns the
1411    /// number actually written. Background-writer entry point —
1412    /// reuses the same persistence path as `flush()` but bounded.
1413    pub fn flush_some_dirty(&self, max: usize) -> Result<usize, PagerError> {
1414        let _ownership = self.lock_ownership_mutations();
1415        self.ensure_writeback_available()?;
1416        if self.config.read_only || max == 0 {
1417            return Ok(0);
1418        }
1419        let dirty_pages = self.cache.flush_some_dirty(max);
1420        if dirty_pages.is_empty() {
1421            return Ok(0);
1422        }
1423        let count = dirty_pages.len();
1424        // WAL-first: every cached dirty page carries an LSN that the
1425        // WAL must have already persisted. The full `flush()` path
1426        // enforces this with `wal.flush(max_lsn)`; here we simply
1427        // write through the pager — safe because callers only reach
1428        // this path via the bgwriter, which runs asynchronously
1429        // alongside normal commits that already respect WAL-first.
1430        for (page_id, page) in dirty_pages {
1431            self.write_page(page_id, page)?;
1432        }
1433        Ok(count)
1434    }
1435
1436    /// Get database file path
1437    pub fn path(&self) -> &Path {
1438        &self.path
1439    }
1440
1441    /// Check if database is read-only
1442    pub fn is_read_only(&self) -> bool {
1443        self.config.read_only
1444    }
1445
1446    /// Get file size in bytes
1447    pub fn file_size(&self) -> Result<u64, PagerError> {
1448        let file = self.file_lock()?;
1449        Ok(file.file_len()?)
1450    }
1451
1452    /// Clear the in-file DWB zone with valid reserved pages.
1453    fn clear_dwb_zone(&self) -> Result<(), PagerError> {
1454        let mut file = self.file_lock()?;
1455        Ok(pager_io::clear_dwb_zone(&mut *file)?)
1456    }
1457
1458    /// Read the internal manifest page (page 1), the zone the superblock roots.
1459    ///
1460    /// ADR 0074 §2: manifest corruption fails the open didactically, naming the
1461    /// zone — it never falls back to a sidecar and never returns garbage.
1462    pub fn read_manifest_page(&self) -> Result<Page, PagerError> {
1463        self.read_page(1)
1464            .map_err(|_| PagerError::ManifestZoneCorrupt(self.path.clone()))
1465    }
1466
1467    /// Write pages through the double-write buffer for torn page protection.
1468    ///
1469    /// 1. Write pages to the in-file DWB zone with a header (magic + count + checksum)
1470    /// 2. fsync the DWB zone
1471    /// 3. Write all pages to their final locations in the .rdb file
1472    /// 4. Clear the DWB zone (marks as consumed)
1473    fn write_pages_through_dwb(&self, pages: &[(u32, Page)]) -> Result<(), PagerError> {
1474        if self.config.read_only {
1475            return Err(PagerError::ReadOnly);
1476        }
1477        // Keep the file locked through staging, home writes and clearing so
1478        // another page writer cannot interleave with a protected batch.
1479        let mut file = self.file_lock()?;
1480        self.ensure_writeback_available()?;
1481        let result = pager_io::write_pages(&mut *file, pages, self.dwb_enabled);
1482        if result.is_err() {
1483            // Set this while holding the file lock: a waiting writer must not
1484            // replace the recovery frame after an interrupted home write.
1485            self.writeback_failed
1486                .store(true, std::sync::atomic::Ordering::Release);
1487        }
1488        Ok(result?)
1489    }
1490
1491    /// Recover from double-write buffer after a crash.
1492    ///
1493    /// If the DWB zone contains valid pages, they were written before the crash
1494    /// interrupted writing to their home locations. Re-apply them.
1495    fn recover_from_dwb(&self) -> Result<(), PagerError> {
1496        let mut file = self.file_lock()?;
1497        Ok(pager_io::recover_from_dwb(&mut *file)?)
1498    }
1499
1500    /// Finish a consumed trunk's durable ownership change, retaining errors
1501    /// across allocation attempts until both copies have been synchronized.
1502    /// Called with the freelist locked when consuming or retrying a trunk.
1503    fn complete_allocation_fence(&self) -> Result<(), PagerError> {
1504        if self
1505            .allocation_fence_pending
1506            .load(std::sync::atomic::Ordering::Acquire)
1507        {
1508            self.write_header_and_sync()?;
1509            self.write_header_and_sync()?;
1510            self.allocation_fence_pending
1511                .store(false, std::sync::atomic::Ordering::Release);
1512        }
1513        Ok(())
1514    }
1515
1516    /// Write header and sync to disk (public for checkpointer).
1517    pub fn write_header_and_sync(&self) -> Result<(), PagerError> {
1518        let _ownership = self.lock_ownership_mutations();
1519        self.write_header()?;
1520        let file = self.file_lock()?;
1521        file.sync_all()?;
1522        Ok(())
1523    }
1524
1525    /// Set the checkpoint_in_progress flag in the header.
1526    pub fn set_checkpoint_in_progress(
1527        &self,
1528        in_progress: bool,
1529        target_lsn: u64,
1530    ) -> Result<(), PagerError> {
1531        let _ownership = self.lock_ownership_mutations();
1532        let mut header = self.header_write()?;
1533        header.checkpoint_in_progress = in_progress;
1534        header.checkpoint_target_lsn = target_lsn;
1535        *self.header_dirty_lock()? = true;
1536        drop(header);
1537        self.write_header_and_sync()
1538    }
1539
1540    /// Update the checkpoint LSN and clear the in-progress flag.
1541    pub fn complete_checkpoint(&self, lsn: u64) -> Result<(), PagerError> {
1542        let _ownership = self.lock_ownership_mutations();
1543        let mut header = self.header_write()?;
1544        header.checkpoint_lsn = lsn;
1545        header.checkpoint_in_progress = false;
1546        header.checkpoint_target_lsn = 0;
1547        *self.header_dirty_lock()? = true;
1548        drop(header);
1549        self.write_header_and_sync()
1550    }
1551}