1use 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#[derive(Debug)]
21pub enum PagerError {
22 Io(std::io::Error),
24 Page(PageError),
26 InvalidDatabase(String),
28 ReadOnly,
30 PageNotFound(u32),
32 Locked,
34 LockPoisoned,
36 EncryptionRequired,
38 PlainDatabaseRefusesKey,
40 InvalidKey,
42 SuperblockZoneUnrecoverable(PathBuf),
44 ManifestZoneCorrupt(PathBuf),
46 DwbZoneCorrupt(PathBuf),
48 LegacySidecarStore { sidecar: PathBuf },
50 LegacyPagedStore { path: PathBuf },
52}
53
54#[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#[derive(Debug, Clone)]
160pub struct PagerConfig {
161 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
181pub trait WalBarrier {
184 fn flush_until(&self, lsn: u64) -> std::io::Result<()>;
185
186 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 fn needs_page_image(&self, _page_id: u32) -> std::io::Result<bool> {
201 Ok(false)
202 }
203
204 fn capture_page_image(&self, _page: &Page) -> std::io::Result<()> {
207 Ok(())
208 }
209
210 fn reserve_pages(&self, _start: u32, _count: u32) -> std::io::Result<()> {
212 Ok(())
213 }
214
215 fn detach_free_pages(&self, _pages: &[u32]) -> std::io::Result<()> {
217 Ok(())
218 }
219
220 fn has_page_ownership_journal(&self) -> std::io::Result<bool> {
223 Ok(false)
224 }
225
226 fn retire_pages(&self, _pages: &[u32]) -> std::io::Result<()> {
228 Ok(())
229 }
230}
231
232pub 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 path: PathBuf,
244 file: Mutex<F>,
246 dwb_enabled: bool,
248 staged_pages: Mutex<Option<pager_io::StagedPageBatch>>,
249 writeback_failed: std::sync::atomic::AtomicBool,
252 cache: PageCache,
254 read_only_recovery: RwLock<std::collections::HashMap<u32, Page>>,
256 freelist: RwLock<FreeList>,
258 deferred_reclamation: Mutex<DeferredPageReclamation>,
261 ownership_mutations: parking_lot::ReentrantMutex<()>,
264 allocation_reused: std::sync::atomic::AtomicBool,
267 allocation_fence_pending: std::sync::atomic::AtomicBool,
269 header: RwLock<DatabaseHeader>,
271 superblock_publication: Mutex<()>,
273 config: PagerConfig,
275 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
287const SUPERBLOCK_SLOT_SIZE: usize = reddb_file::PAGED_SUPERBLOCK_SLOT_SIZE;
289const SUPERBLOCK_ZONE_SIZE: usize = reddb_file::PAGED_SUPERBLOCK_ZONE_SIZE;
290const PHASE3_DWB_ZONE_FILE_VERSION: u32 = 0x0001_0001;
293
294pub type NewestSuperblock = (
296 reddb_file::PagedSuperblockSelection,
297 [u8; SUPERBLOCK_SLOT_SIZE],
298);
299
300fn 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 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 pub fn lock_ownership_mutations(&self) -> parking_lot::ReentrantMutexGuard<'_, ()> {
360 self.ownership_mutations.lock()
361 }
362
363 pub fn fence_interrupted_ownership(&self) {
365 self.writeback_failed
366 .store(true, std::sync::atomic::Ordering::Release);
367 }
368
369 #[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 fn initialize(&self) -> Result<(), PagerError> {
378 if self.config.read_only {
379 return Err(PagerError::ReadOnly);
380 }
381
382 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 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 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 self.sync()?;
411
412 Ok(())
413 }
414
415 fn header_write(&self) -> Result<std::sync::RwLockWriteGuard<'_, DatabaseHeader>, PagerError> {
417 self.header.write().map_err(|_| PagerError::LockPoisoned)
418 }
419
420 fn header_read(&self) -> Result<std::sync::RwLockReadGuard<'_, DatabaseHeader>, PagerError> {
422 self.header.read().map_err(|_| PagerError::LockPoisoned)
423 }
424
425 fn freelist_write(&self) -> Result<std::sync::RwLockWriteGuard<'_, FreeList>, PagerError> {
427 self.freelist.write().map_err(|_| PagerError::LockPoisoned)
428 }
429
430 fn file_lock(&self) -> Result<std::sync::MutexGuard<'_, F>, PagerError> {
432 self.file.lock().map_err(|_| PagerError::LockPoisoned)
433 }
434
435 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 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 #[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 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 #[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 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 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 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 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 {
767 let mut freelist = self.freelist_write()?;
768 *freelist = FreeList::from_header(freelist_head, 0);
769 }
770
771 Ok(())
772 }
773
774 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 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 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 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 ids.retain(|id| Some(*id) != pending_page);
867 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 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 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 if let Some(page) = self.cache.get(page_id) {
913 return Ok(page);
914 }
915
916 let _ownership = self.lock_ownership_mutations();
920 if let Some(page) = self.cache.get(page_id) {
921 return Ok(page);
922 }
923
924 let page = self.read_page_raw(page_id)?;
926
927 if let Some(dirty_page) = self.cache.insert_if_absent(page_id, page.clone()) {
929 self.write_evicted_dirty_page(dirty_page, None)?;
931 }
932
933 Ok(page)
934 }
935
936 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 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 let mut file = self.file_lock()?;
963 let page = pager_io::read_page(&mut *file, page_id, false)?;
964 drop(file);
965
966 if let Some(dirty_page) = self.cache.insert_if_absent(page_id, page.clone()) {
968 self.write_evicted_dirty_page(dirty_page, None)?;
970 }
971
972 Ok(page)
973 }
974
975 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 page.update_checksum();
988
989 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 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 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 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 let page_id = if self.batch_active()? {
1045 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 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 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 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 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 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 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 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 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 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 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 self.cache.remove(page_id);
1281
1282 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 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 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 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 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 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 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 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 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 pub fn page_count(&self) -> Result<u32, PagerError> {
1597 Ok(self.header_read()?.page_count)
1598 }
1599
1600 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 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 dirty_pages.push((evicted.page_id(), evicted));
1634 }
1635 }
1636
1637 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 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 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 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 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 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 pub fn cache_stats(&self) -> super::page_cache::CacheStats {
1726 self.cache.stats()
1727 }
1728
1729 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 pub fn cache_capacity(&self) -> usize {
1746 self.cache.capacity()
1747 }
1748
1749 pub fn dirty_page_count(&self) -> usize {
1751 self.cache.dirty_count()
1752 }
1753
1754 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 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 for (page_id, page) in dirty_pages {
1783 self.write_page(page_id, page)?;
1784 }
1785 Ok(count)
1786 }
1787
1788 pub fn path(&self) -> &Path {
1790 &self.path
1791 }
1792
1793 pub fn into_inner(self) -> F {
1796 self.file
1797 .into_inner()
1798 .unwrap_or_else(|poisoned| poisoned.into_inner())
1799 }
1800
1801 pub fn is_read_only(&self) -> bool {
1803 self.config.read_only
1804 }
1805
1806 pub fn file_size(&self) -> Result<u64, PagerError> {
1808 let file = self.file_lock()?;
1809 Ok(file.file_len()?)
1810 }
1811
1812 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 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 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 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 fn write_pages_through_dwb(&self, pages: &[(u32, Page)]) -> Result<(), PagerError> {
1901 if self.config.read_only {
1902 return Err(PagerError::ReadOnly);
1903 }
1904 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 self.writeback_failed
1930 .store(true, std::sync::atomic::Ordering::Release);
1931 }
1932 result
1933 }
1934
1935 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 self.install_read_only_page(id, page)?;
1955 }
1956 return Ok(());
1957 }
1958 Ok(pager_io::recover_from_dwb(&mut *file)?)
1959 }
1960
1961 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 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 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 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 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}