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
17#[derive(Debug)]
19pub enum PagerError {
20 Io(std::io::Error),
22 Page(PageError),
24 InvalidDatabase(String),
26 ReadOnly,
28 PageNotFound(u32),
30 Locked,
32 LockPoisoned,
34 EncryptionRequired,
36 PlainDatabaseRefusesKey,
38 InvalidKey,
40 SuperblockZoneUnrecoverable(PathBuf),
42 ManifestZoneCorrupt(PathBuf),
44 DwbZoneCorrupt(PathBuf),
46 LegacySidecarStore { sidecar: PathBuf },
48 LegacyPagedStore { path: PathBuf },
50}
51
52#[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#[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
177pub trait WalBarrier {
180 fn flush_until(&self, lsn: u64) -> std::io::Result<()>;
181
182 fn needs_page_image(&self, _page_id: u32) -> std::io::Result<bool> {
185 Ok(false)
186 }
187
188 fn capture_page_image(&self, _page: &Page) -> std::io::Result<()> {
191 Ok(())
192 }
193
194 fn reserve_pages(&self, _start: u32, _count: u32) -> std::io::Result<()> {
196 Ok(())
197 }
198
199 fn detach_free_pages(&self, _pages: &[u32]) -> std::io::Result<()> {
201 Ok(())
202 }
203
204 fn retire_pages(&self, _pages: &[u32]) -> std::io::Result<()> {
206 Ok(())
207 }
208}
209
210pub 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 path: PathBuf,
222 file: Mutex<F>,
224 dwb_enabled: bool,
226 writeback_failed: std::sync::atomic::AtomicBool,
229 cache: PageCache,
231 freelist: RwLock<FreeList>,
233 deferred_reclamation: Mutex<DeferredPageReclamation>,
236 ownership_mutations: parking_lot::ReentrantMutex<()>,
239 allocation_reused: std::sync::atomic::AtomicBool,
242 allocation_fence_pending: std::sync::atomic::AtomicBool,
244 header: RwLock<DatabaseHeader>,
246 superblock_publication: Mutex<()>,
248 config: PagerConfig,
250 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
262const SUPERBLOCK_SLOT_SIZE: usize = reddb_file::PAGED_SUPERBLOCK_SLOT_SIZE;
264const SUPERBLOCK_ZONE_SIZE: usize = reddb_file::PAGED_SUPERBLOCK_ZONE_SIZE;
265const PHASE3_DWB_ZONE_FILE_VERSION: u32 = 0x0001_0001;
268
269pub type NewestSuperblock = (
271 reddb_file::PagedSuperblockSelection,
272 [u8; SUPERBLOCK_SLOT_SIZE],
273);
274
275fn 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 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 pub fn lock_ownership_mutations(&self) -> parking_lot::ReentrantMutexGuard<'_, ()> {
333 self.ownership_mutations.lock()
334 }
335
336 pub fn fence_interrupted_ownership(&self) {
338 self.writeback_failed
339 .store(true, std::sync::atomic::Ordering::Release);
340 }
341
342 #[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 fn initialize(&self) -> Result<(), PagerError> {
351 if self.config.read_only {
352 return Err(PagerError::ReadOnly);
353 }
354
355 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 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 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 self.sync()?;
384
385 Ok(())
386 }
387
388 fn header_write(&self) -> Result<std::sync::RwLockWriteGuard<'_, DatabaseHeader>, PagerError> {
390 self.header.write().map_err(|_| PagerError::LockPoisoned)
391 }
392
393 fn header_read(&self) -> Result<std::sync::RwLockReadGuard<'_, DatabaseHeader>, PagerError> {
395 self.header.read().map_err(|_| PagerError::LockPoisoned)
396 }
397
398 fn freelist_write(&self) -> Result<std::sync::RwLockWriteGuard<'_, FreeList>, PagerError> {
400 self.freelist.write().map_err(|_| PagerError::LockPoisoned)
401 }
402
403 fn file_lock(&self) -> Result<std::sync::MutexGuard<'_, F>, PagerError> {
405 self.file.lock().map_err(|_| PagerError::LockPoisoned)
406 }
407
408 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 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 #[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 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 #[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 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 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 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 {
574 let mut freelist = self.freelist_write()?;
575 *freelist = FreeList::from_header(freelist_head, 0);
576 }
577
578 Ok(())
579 }
580
581 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 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 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 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 pub fn read_page(&self, page_id: u32) -> Result<Page, PagerError> {
657 if let Some(page) = self.cache.get(page_id) {
659 return Ok(page);
660 }
661
662 let page = self.read_page_raw(page_id)?;
664
665 if let Some(dirty_page) = self.cache.insert(page_id, page.clone()) {
667 self.write_evicted_dirty_page(dirty_page)?;
669 }
670
671 Ok(page)
672 }
673
674 pub fn read_page_no_checksum(&self, page_id: u32) -> Result<Page, PagerError> {
679 if let Some(page) = self.cache.get(page_id) {
681 return Ok(page);
682 }
683
684 let mut file = self.file_lock()?;
686 let page = pager_io::read_page(&mut *file, page_id, false)?;
687 drop(file);
688
689 if let Some(dirty_page) = self.cache.insert(page_id, page.clone()) {
691 self.write_evicted_dirty_page(dirty_page)?;
693 }
694
695 Ok(page)
696 }
697
698 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 page.update_checksum();
710
711 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 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 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 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 let page_id = {
769 let mut freelist = self.freelist_write()?;
770 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 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 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 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 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 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 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 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 self.cache.remove(page_id);
944
945 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 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 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 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 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 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 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 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 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 pub fn page_count(&self) -> Result<u32, PagerError> {
1260 Ok(self.header_read()?.page_count)
1261 }
1262
1263 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 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 dirty_pages.push((evicted.page_id(), evicted));
1297 }
1298 self.cache.mark_dirty(page_id);
1299 }
1300
1301 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 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 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 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 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 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 pub fn cache_stats(&self) -> super::page_cache::CacheStats {
1382 self.cache.stats()
1383 }
1384
1385 pub fn cache_len(&self) -> usize {
1388 self.cache.len()
1389 }
1390
1391 pub fn cache_capacity(&self) -> usize {
1394 self.cache.capacity()
1395 }
1396
1397 pub fn dirty_page_count(&self) -> usize {
1399 self.cache.dirty_count()
1400 }
1401
1402 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 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 for (page_id, page) in dirty_pages {
1431 self.write_page(page_id, page)?;
1432 }
1433 Ok(count)
1434 }
1435
1436 pub fn path(&self) -> &Path {
1438 &self.path
1439 }
1440
1441 pub fn is_read_only(&self) -> bool {
1443 self.config.read_only
1444 }
1445
1446 pub fn file_size(&self) -> Result<u64, PagerError> {
1448 let file = self.file_lock()?;
1449 Ok(file.file_len()?)
1450 }
1451
1452 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 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 fn write_pages_through_dwb(&self, pages: &[(u32, Page)]) -> Result<(), PagerError> {
1474 if self.config.read_only {
1475 return Err(PagerError::ReadOnly);
1476 }
1477 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 self.writeback_failed
1486 .store(true, std::sync::atomic::Ordering::Release);
1487 }
1488 Ok(result?)
1489 }
1490
1491 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 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 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 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 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}