1use super::*;
2
3pub(super) const BTRFS_SUPER_MAGIC: i64 = 0x9123_683e;
4pub(super) const ZFS_SUPER_MAGIC: i64 = 0x2fc1_2fc1;
5pub(super) const FS_NOCOW_FL: u64 = 0x0080_0000;
6
7pub(super) const SUPERBLOCK_SLOT_SIZE: usize = reddb_file::PAGED_SUPERBLOCK_SLOT_SIZE;
9pub(super) const SUPERBLOCK_ZONE_SIZE: usize = reddb_file::PAGED_SUPERBLOCK_ZONE_SIZE;
10
11type NewestSuperblock = (
13 reddb_file::PagedSuperblockSelection,
14 [u8; SUPERBLOCK_SLOT_SIZE],
15);
16
17fn superblock_slot_image(page: &[u8; PAGE_SIZE]) -> [u8; SUPERBLOCK_SLOT_SIZE] {
20 let mut image = [0u8; SUPERBLOCK_SLOT_SIZE];
21 image.copy_from_slice(&page[..SUPERBLOCK_SLOT_SIZE]);
22 image
23}
24
25#[derive(Debug, Clone, Copy, PartialEq, Eq)]
26pub(super) enum CowFilesystemKind {
27 Zfs,
28 BtrfsDataCow,
29 TestOverride,
30}
31
32pub(super) fn classify_cow_filesystem(
33 fs_type: i64,
34 mount_options: Option<&str>,
35 inode_flags: Option<u64>,
36) -> Option<CowFilesystemKind> {
37 match fs_type {
38 ZFS_SUPER_MAGIC => Some(CowFilesystemKind::Zfs),
39 BTRFS_SUPER_MAGIC => {
40 let mount_options = mount_options?;
41 if mount_options.split(',').any(|option| option == "nodatacow") {
42 return None;
43 }
44
45 let inode_flags = inode_flags?;
46 if inode_flags & FS_NOCOW_FL != 0 {
47 return None;
48 }
49
50 Some(CowFilesystemKind::BtrfsDataCow)
51 }
52 _ => None,
53 }
54}
55
56#[cfg(target_os = "linux")]
57fn linux_fstatfs_type(file: &File) -> Option<i64> {
58 use std::mem::MaybeUninit;
59 use std::os::fd::AsRawFd;
60
61 let mut stat = MaybeUninit::<libc::statfs>::uninit();
62 let rc = unsafe { libc::fstatfs(file.as_raw_fd(), stat.as_mut_ptr()) };
63 if rc != 0 {
64 return None;
65 }
66 let stat = unsafe { stat.assume_init() };
67 #[allow(clippy::unnecessary_cast)]
70 Some(stat.f_type as i64)
71}
72
73#[cfg(target_os = "linux")]
74fn linux_inode_flags(file: &File) -> Option<u64> {
75 use std::os::fd::AsRawFd;
76
77 let mut flags: libc::c_long = 0;
78 let rc = unsafe { libc::ioctl(file.as_raw_fd(), libc::FS_IOC_GETFLAGS, &mut flags) };
79 if rc != 0 {
80 return None;
81 }
82 Some(flags as u64)
83}
84
85#[cfg(target_os = "linux")]
86fn linux_mount_options_for_path(path: &Path) -> Option<String> {
87 let path = path.canonicalize().ok()?;
88 let mountinfo = std::fs::read_to_string("/proc/self/mountinfo").ok()?;
89 parse_mountinfo_options_for_path(&mountinfo, &path)
90}
91
92#[cfg(target_os = "linux")]
93pub(super) fn parse_mountinfo_options_for_path(mountinfo: &str, path: &Path) -> Option<String> {
94 let mut best: Option<(usize, String)> = None;
95
96 for line in mountinfo.lines() {
97 let fields: Vec<&str> = line.split(' ').collect();
98 if fields.len() < 10 {
99 continue;
100 }
101
102 let Some(separator) = fields.iter().position(|field| *field == "-") else {
103 continue;
104 };
105 if separator + 3 >= fields.len() || separator < 6 {
106 continue;
107 }
108
109 let mount_point = mountinfo_unescape_path(fields[4]);
110 if !path.starts_with(&mount_point) {
111 continue;
112 }
113
114 let fs_type = fields[separator + 1];
115 if fs_type != "btrfs" && fs_type != "zfs" {
116 continue;
117 }
118
119 let mount_options = fields[5];
120 let super_options = fields[separator + 3];
121 let options = format!("{mount_options},{super_options}");
122 let depth = mount_point.components().count();
123 if best
124 .as_ref()
125 .map(|(best_depth, _)| depth > *best_depth)
126 .unwrap_or(true)
127 {
128 best = Some((depth, options));
129 }
130 }
131
132 best.map(|(_, options)| options)
133}
134
135#[cfg(target_os = "linux")]
136fn mountinfo_unescape_path(value: &str) -> PathBuf {
137 let bytes = value.as_bytes();
138 let mut out = Vec::with_capacity(bytes.len());
139 let mut i = 0;
140
141 while i < bytes.len() {
142 if bytes[i] == b'\\' && i + 3 < bytes.len() {
143 let octal = &value[i + 1..i + 4];
144 if let Ok(byte) = u8::from_str_radix(octal, 8) {
145 out.push(byte);
146 i += 4;
147 continue;
148 }
149 }
150 out.push(bytes[i]);
151 i += 1;
152 }
153
154 PathBuf::from(String::from_utf8_lossy(&out).into_owned())
155}
156
157impl Pager {
158 pub fn open<P: AsRef<Path>>(path: P, mut config: PagerConfig) -> Result<Self, PagerError> {
160 let path = path.as_ref().to_path_buf();
161 let exists = path.exists();
162
163 if !exists && !config.create {
164 return Err(PagerError::InvalidDatabase(
165 "Database does not exist".into(),
166 ));
167 }
168
169 if exists {
173 if let Some(sidecar) = reddb_file::layout::retired::first_present_phase1_sidecar(&path)
174 {
175 return Err(PagerError::LegacySidecarStore { sidecar });
176 }
177 }
178
179 if !exists && config.read_only {
180 return Err(PagerError::InvalidDatabase(
181 "Cannot create read-only database".into(),
182 ));
183 }
184
185 let file = OpenOptions::new()
188 .read(true)
189 .write(!config.read_only)
190 .create(config.create && !config.read_only)
191 .open(&path)?;
192
193 #[cfg(unix)]
199 {
200 use std::os::unix::fs::MetadataExt;
201 if let Ok(meta) = file.metadata() {
202 let fs_block_size = meta.blksize();
203 if Self::page_size_misaligned_with_block(PAGE_SIZE, fs_block_size) {
204 tracing::warn!(
205 page_size = PAGE_SIZE,
206 fs_block_size,
207 path = %path.display(),
208 "database page size is not a multiple of the filesystem \
209 block size; page writes will straddle FS blocks \
210 (read-modify-write amplification). Diagnostic only — \
211 the page size is unchanged."
212 );
213 }
214 }
215 }
216
217 let lock_file = if !config.read_only {
219 let lf = OpenOptions::new().read(true).write(true).open(&path)?;
220 lf.try_lock_exclusive().map_err(|_| PagerError::Locked)?;
221 Some(lf)
222 } else {
223 let lf = OpenOptions::new().read(true).open(&path)?;
224 match lf.try_lock_shared() {
225 Ok(_) => Some(lf),
226 Err(_) => None,
227 }
228 };
229
230 let fold_dwb = crate::physical::fold_dwb_into_wal_enabled();
242 if !config.double_write && !config.read_only && !fold_dwb {
243 let skip_dwb_on_cow =
244 Self::cow_filesystem_has_atomic_page_writes(&path, &file).is_some();
245 if !skip_dwb_on_cow {
246 tracing::warn!(
247 path = %path.display(),
248 "double_write=false requested, but the data file is not proven to be on \
249 ZFS or btrfs datacow; keeping the double-write buffer enabled"
250 );
251 config.double_write = true;
252 }
253 }
254
255 let dwb_file = if config.double_write && !config.read_only && !fold_dwb {
256 let f = Self::open_dwb_file(&path)?;
257 Some(Mutex::new(f))
258 } else {
259 if fold_dwb && !config.read_only {
260 let _ = std::fs::remove_file(Self::dwb_path(&path));
261 }
262 None
263 };
264
265 let mut pager = Self {
266 path,
267 file: Mutex::new(file),
268 _lock_file: lock_file,
269 dwb_file,
270 cache: PageCache::new(config.cache_size),
271 freelist: RwLock::new(FreeList::new()),
272 header: RwLock::new(DatabaseHeader::default()),
273 config,
274 header_dirty: Mutex::new(false),
275 wal: RwLock::new(None),
276 encryption: None,
277 };
278
279 if exists {
280 pager.recover_from_dwb()?;
282 if !pager.config.double_write && !pager.config.read_only {
283 let _ = std::fs::remove_file(Self::dwb_path(&pager.path));
284 }
285 pager.load_header()?;
287 pager.bind_encryption_for_existing()?;
288 } else {
289 pager.initialize()?;
291 pager.bind_encryption_for_new()?;
292 }
293
294 Ok(pager)
295 }
296
297 pub(crate) fn page_size_misaligned_with_block(page_size: usize, fs_block_size: u64) -> bool {
304 fs_block_size != 0 && !(page_size as u64).is_multiple_of(fs_block_size)
305 }
306
307 #[cfg(test)]
308 fn cow_filesystem_test_override() -> Option<bool> {
309 match COW_ATOMIC_WRITE_TEST_OVERRIDE.load(Ordering::Relaxed) {
310 1 => Some(true),
311 2 => Some(false),
312 _ => None,
313 }
314 }
315
316 #[cfg(not(test))]
317 fn cow_filesystem_test_override() -> Option<bool> {
318 None
319 }
320
321 fn cow_filesystem_has_atomic_page_writes(
322 path: &Path,
323 file: &File,
324 ) -> Option<CowFilesystemKind> {
325 if let Some(allowed) = Self::cow_filesystem_test_override() {
326 return allowed.then_some(CowFilesystemKind::TestOverride);
327 }
328 Self::probe_cow_filesystem(path, file)
329 }
330
331 #[cfg(target_os = "linux")]
332 fn probe_cow_filesystem(path: &Path, file: &File) -> Option<CowFilesystemKind> {
333 let fs_type = linux_fstatfs_type(file)?;
334 match fs_type {
335 ZFS_SUPER_MAGIC => Some(CowFilesystemKind::Zfs),
336 BTRFS_SUPER_MAGIC => {
337 let mount_options = linux_mount_options_for_path(path)?;
338 let inode_flags = linux_inode_flags(file)?;
339 classify_cow_filesystem(fs_type, Some(&mount_options), Some(inode_flags))
340 }
341 _ => None,
342 }
343 }
344
345 #[cfg(not(target_os = "linux"))]
346 fn probe_cow_filesystem(_path: &Path, _file: &File) -> Option<CowFilesystemKind> {
347 None
348 }
349
350 fn bind_encryption_for_existing(&mut self) -> Result<(), PagerError> {
360 if self.page_count().unwrap_or(0) == 0 {
361 return self.bind_encryption_for_new();
362 }
363 let (_, image) = self.newest_superblock()?;
364 let data = &image[..];
365 let has_marker = reddb_file::paged_encryption_marker_present(data);
366
367 let key = self.config.encryption.clone();
368 match (has_marker, key) {
369 (true, Some(key)) => {
370 let header_bytes =
371 reddb_file::paged_encryption_header_bytes(data).ok_or_else(|| {
372 PagerError::InvalidDatabase("encryption header parse failed".to_string())
373 })?;
374 let header = crate::storage::encryption::EncryptionHeader::from_bytes(header_bytes)
375 .map_err(|e| {
376 PagerError::InvalidDatabase(format!("encryption header parse failed: {e}"))
377 })?;
378 if !header.validate(&key) {
379 return Err(PagerError::InvalidKey);
380 }
381 let encryptor = crate::storage::encryption::PageEncryptor::new(key);
382 self.encryption = Some((encryptor, header));
383 Ok(())
384 }
385 (true, None) => Err(PagerError::EncryptionRequired),
386 (false, Some(_)) => Err(PagerError::PlainDatabaseRefusesKey),
387 (false, None) => Ok(()),
388 }
389 }
390
391 fn bind_encryption_for_new(&mut self) -> Result<(), PagerError> {
394 let Some(key) = self.config.encryption.clone() else {
395 return Ok(());
396 };
397 let header = crate::storage::encryption::EncryptionHeader::new(&key);
398 let encryptor = crate::storage::encryption::PageEncryptor::new(key);
399
400 if self.page_count().unwrap_or(0) > 0 {
403 let header_bytes = header.to_bytes();
404 self.commit_superblock(|image| {
405 reddb_file::write_paged_encryption_marker_and_header(image, &header_bytes)
406 .map_err(|err| PagerError::InvalidDatabase(err.to_string()))
407 })?;
408 }
409 self.encryption = Some((encryptor, header));
410 Ok(())
411 }
412
413 pub fn open_default<P: AsRef<Path>>(path: P) -> Result<Self, PagerError> {
415 Self::open(path, PagerConfig::default())
416 }
417
418 fn initialize(&self) -> Result<(), PagerError> {
420 if self.config.read_only {
421 return Err(PagerError::ReadOnly);
422 }
423
424 let initial_page_count = 3;
428 let header_page = Page::new_header_page(initial_page_count);
429 self.header_write()?.page_count = initial_page_count;
430
431 self.write_page_raw(0, &header_page)?;
436 let mut metadata_page = Page::new(PageType::Header, 1);
437 metadata_page.update_checksum();
438 self.write_page_raw(1, &metadata_page)?;
439 let mut vault_page = Page::new(PageType::Vault, 2);
440 vault_page.update_checksum();
441 self.write_page_raw(2, &vault_page)?;
442
443 let mut slot = superblock_slot_image(header_page.as_bytes());
447 self.write_superblock_slot(&mut slot, 0, 1)?;
448 self.write_superblock_slot(&mut slot, 1, 2)?;
449
450 self.sync()?;
452
453 Ok(())
454 }
455
456 fn header_write(&self) -> Result<std::sync::RwLockWriteGuard<'_, DatabaseHeader>, PagerError> {
458 self.header.write().map_err(|_| PagerError::LockPoisoned)
459 }
460
461 fn header_read(&self) -> Result<std::sync::RwLockReadGuard<'_, DatabaseHeader>, PagerError> {
463 self.header.read().map_err(|_| PagerError::LockPoisoned)
464 }
465
466 fn freelist_write(&self) -> Result<std::sync::RwLockWriteGuard<'_, FreeList>, PagerError> {
468 self.freelist.write().map_err(|_| PagerError::LockPoisoned)
469 }
470
471 fn file_lock(&self) -> Result<std::sync::MutexGuard<'_, File>, PagerError> {
473 self.file.lock().map_err(|_| PagerError::LockPoisoned)
474 }
475
476 fn header_dirty_lock(&self) -> Result<std::sync::MutexGuard<'_, bool>, PagerError> {
478 self.header_dirty
479 .lock()
480 .map_err(|_| PagerError::LockPoisoned)
481 }
482
483 fn read_superblock_zone(&self) -> Result<[u8; SUPERBLOCK_ZONE_SIZE], PagerError> {
487 let mut file = self.file_lock()?;
488 let len = file.metadata().map(|meta| meta.len()).unwrap_or(0);
489 let mut zone = [0u8; SUPERBLOCK_ZONE_SIZE];
490 let readable = len.min(SUPERBLOCK_ZONE_SIZE as u64) as usize;
491 if readable > 0 {
492 file.seek(SeekFrom::Start(0))?;
493 file.read_exact(&mut zone[..readable])?;
494 }
495 Ok(zone)
496 }
497
498 fn newest_superblock(&self) -> Result<NewestSuperblock, PagerError> {
504 let zone = self.read_superblock_zone()?;
505 let selection = reddb_file::select_paged_superblock(&zone)
506 .ok_or_else(|| PagerError::SuperblockZoneUnrecoverable(self.path.clone()))?;
507
508 let start = selection.copy_index * SUPERBLOCK_SLOT_SIZE;
509 let mut image = [0u8; SUPERBLOCK_SLOT_SIZE];
510 image.copy_from_slice(&zone[start..start + SUPERBLOCK_SLOT_SIZE]);
511 Ok((selection, image))
512 }
513
514 fn write_superblock_slot(
517 &self,
518 slot: &mut [u8; SUPERBLOCK_SLOT_SIZE],
519 copy_index: usize,
520 generation: u64,
521 ) -> Result<(), PagerError> {
522 if self.config.read_only {
523 return Err(PagerError::ReadOnly);
524 }
525 reddb_file::seal_paged_superblock_slot(slot, copy_index, generation)
526 .map_err(|err| PagerError::InvalidDatabase(err.to_string()))?;
527
528 let mut file = self.file_lock()?;
529 file.seek(SeekFrom::Start(reddb_file::paged_superblock_slot_offset(
530 copy_index,
531 )))?;
532 file.write_all(slot)?;
533 file.sync_all()?;
534 Ok(())
535 }
536
537 fn commit_superblock(
543 &self,
544 mutate: impl FnOnce(&mut [u8; SUPERBLOCK_SLOT_SIZE]) -> Result<(), PagerError>,
545 ) -> Result<(), PagerError> {
546 if self.config.read_only {
547 return Err(PagerError::ReadOnly);
548 }
549 let (newest, mut image) = self.newest_superblock()?;
550 mutate(&mut image)?;
551 let next_copy = reddb_file::paged_superblock_next_copy(Some(newest));
552 self.write_superblock_slot(&mut image, next_copy, newest.generation.saturating_add(1))
553 }
554
555 fn load_header(&self) -> Result<(), PagerError> {
557 let (_, image) = self.newest_superblock()?;
558
559 let decoded_header = reddb_file::decode_database_header(&image)
560 .map_err(|err| PagerError::InvalidDatabase(err.to_string()))?;
561 let freelist_head = decoded_header.freelist_head;
562
563 {
564 let mut header = self.header_write()?;
565 *header = decoded_header;
566 }
567
568 {
570 let mut freelist = self.freelist_write()?;
571 *freelist = FreeList::from_header(freelist_head, 0);
572 }
573
574 Ok(())
575 }
576
577 fn write_header(&self) -> Result<(), PagerError> {
579 if self.config.read_only {
580 return Err(PagerError::ReadOnly);
581 }
582
583 let header = self.header_read()?.clone();
584 self.commit_superblock(|image| {
585 reddb_file::encode_database_header(image, &header)
586 .map_err(|err| PagerError::InvalidDatabase(err.to_string()))
587 })?;
588 *self.header_dirty_lock()? = false;
589
590 Ok(())
591 }
592
593 fn read_page_raw(&self, page_id: u32) -> Result<Page, PagerError> {
595 let mut file = self.file_lock()?;
596 let offset = (page_id as u64) * (PAGE_SIZE as u64);
597
598 file.seek(SeekFrom::Start(offset))?;
599
600 let mut buf = [0u8; PAGE_SIZE];
601 file.read_exact(&mut buf)?;
602
603 let page = Page::from_bytes(buf);
604
605 if self.config.verify_checksums && page_id != 0 {
609 page.verify_checksum()?;
610 }
611
612 Ok(page)
613 }
614
615 fn write_page_raw(&self, page_id: u32, page: &Page) -> Result<(), PagerError> {
617 if self.config.read_only {
618 return Err(PagerError::ReadOnly);
619 }
620
621 let mut file = self.file_lock()?;
622 let offset = (page_id as u64) * (PAGE_SIZE as u64);
623
624 file.seek(SeekFrom::Start(offset))?;
625 file.write_all(page.as_bytes())?;
626
627 Ok(())
628 }
629
630 pub fn read_page(&self, page_id: u32) -> Result<Page, PagerError> {
632 if let Some(page) = self.cache.get(page_id) {
634 return Ok(page);
635 }
636
637 let page = self.read_page_raw(page_id)?;
639
640 if let Some(dirty_page) = self.cache.insert(page_id, page.clone()) {
642 let evicted_id = dirty_page.page_id();
644 self.write_page_raw(evicted_id, &dirty_page)?;
645 }
646
647 Ok(page)
648 }
649
650 pub fn read_page_no_checksum(&self, page_id: u32) -> Result<Page, PagerError> {
655 if let Some(page) = self.cache.get(page_id) {
657 return Ok(page);
658 }
659
660 let mut file = self.file_lock()?;
662 let offset = (page_id as u64) * (PAGE_SIZE as u64);
663
664 file.seek(SeekFrom::Start(offset))?;
665
666 let mut buf = [0u8; PAGE_SIZE];
667 file.read_exact(&mut buf)?;
668 drop(file);
669
670 let page = Page::from_bytes(buf);
671
672 if let Some(dirty_page) = self.cache.insert(page_id, page.clone()) {
674 let evicted_id = dirty_page.page_id();
676 self.write_page_raw(evicted_id, &dirty_page)?;
677 }
678
679 Ok(page)
680 }
681
682 pub fn write_page(&self, page_id: u32, mut page: Page) -> Result<(), PagerError> {
684 if self.config.read_only {
685 return Err(PagerError::ReadOnly);
686 }
687
688 page.update_checksum();
690
691 if let Some(dirty_page) = self.cache.insert(page_id, page) {
696 let evicted_id = dirty_page.page_id();
697 self.write_page_raw(evicted_id, &dirty_page)?;
698 }
699 self.cache.mark_dirty(page_id);
700
701 Ok(())
702 }
703
704 pub fn read_page_decrypted(&self, page_id: u32) -> Result<Page, PagerError> {
709 if page_id == 0 || self.encryption.is_none() {
710 return self.read_page(page_id);
711 }
712 let raw = self.read_page_no_checksum(page_id)?;
713 let (enc, _) = self
714 .encryption
715 .as_ref()
716 .expect("encryption presence checked above");
717 let plaintext = enc
718 .decrypt(page_id, raw.as_bytes())
719 .map_err(|e| PagerError::InvalidDatabase(format!("decrypt page {page_id}: {e}")))?;
720 let mut buf = [0u8; PAGE_SIZE];
721 let n = plaintext.len().min(PAGE_SIZE);
722 buf[..n].copy_from_slice(&plaintext[..n]);
723 Ok(Page::from_bytes(buf))
724 }
725
726 pub fn write_page_encrypted(&self, page_id: u32, page: Page) -> Result<(), PagerError> {
731 if page_id == 0 || self.encryption.is_none() {
732 return self.write_page(page_id, page);
733 }
734 const OVERHEAD: usize = crate::storage::encryption::page_encryptor::OVERHEAD;
737 let plaintext_len = PAGE_SIZE - OVERHEAD;
738 let plaintext = &page.as_bytes()[..plaintext_len];
739 let (enc, _) = self
740 .encryption
741 .as_ref()
742 .expect("encryption presence checked above");
743 let ciphertext = enc.encrypt(page_id, plaintext);
744 debug_assert_eq!(ciphertext.len(), PAGE_SIZE);
745 let mut buf = [0u8; PAGE_SIZE];
746 buf.copy_from_slice(&ciphertext);
747 let cipher_page = Page::from_bytes(buf);
748 self.write_page_no_checksum(page_id, cipher_page)
749 }
750
751 pub fn write_page_no_checksum(&self, page_id: u32, page: Page) -> Result<(), PagerError> {
756 if self.config.read_only {
757 return Err(PagerError::ReadOnly);
758 }
759
760 if let Some(dirty_page) = self.cache.insert(page_id, page) {
763 let evicted_id = dirty_page.page_id();
764 self.write_page_raw(evicted_id, &dirty_page)?;
765 }
766 self.cache.mark_dirty(page_id);
767
768 Ok(())
769 }
770
771 pub fn allocate_page(&self, page_type: PageType) -> Result<Page, PagerError> {
773 if self.config.read_only {
774 return Err(PagerError::ReadOnly);
775 }
776
777 let page_id = {
779 let mut freelist = self.freelist_write()?;
780 if let Some(id) = freelist.allocate() {
781 id
782 } else if freelist.trunk_head() != 0 {
783 let trunk_id = freelist.trunk_head();
784 drop(freelist);
785
786 let trunk = self.read_page(trunk_id).map_err(|e| match e {
787 PagerError::PageNotFound(_) => {
788 PagerError::InvalidDatabase("Freelist trunk missing".to_string())
789 }
790 other => other,
791 })?;
792
793 let mut freelist = self.freelist_write()?;
794 freelist
795 .load_from_trunk(&trunk)
796 .map_err(|e| PagerError::InvalidDatabase(format!("Freelist: {}", e)))?;
797 let id = freelist.allocate().ok_or_else(|| {
798 PagerError::InvalidDatabase("Freelist empty after trunk load".to_string())
799 })?;
800
801 let mut header = self.header_write()?;
802 header.freelist_head = freelist.trunk_head();
803 *self.header_dirty_lock()? = true;
804
805 id
806 } else {
807 let mut header = self.header_write()?;
809 let id = header.page_count;
810 header.page_count += 1;
811 *self.header_dirty_lock()? = true;
812 id
813 }
814 };
815
816 let page = Page::new(page_type, page_id);
817
818 if let Some(dirty_page) = self.cache.insert(page_id, page.clone()) {
830 let evicted_id = dirty_page.page_id();
831 self.write_page_raw(evicted_id, &dirty_page)?;
832 }
833 self.cache.mark_dirty(page_id);
834
835 Ok(page)
836 }
837
838 pub fn reserve_contig_extent(&self, n_pages: u32) -> Result<super::ExtentId, PagerError> {
840 if self.config.read_only {
841 return Err(PagerError::ReadOnly);
842 }
843 if n_pages == 0 {
844 return Err(PagerError::InvalidDatabase(
845 "contiguous extent must reserve at least one page".to_string(),
846 ));
847 }
848
849 let start_page = {
850 let mut header = self.header_write()?;
851 let start = header.page_count;
852 header.page_count = header.page_count.checked_add(n_pages).ok_or_else(|| {
853 PagerError::InvalidDatabase("contiguous extent page count overflow".to_string())
854 })?;
855 *self.header_dirty_lock()? = true;
856 start
857 };
858
859 for page_id in start_page..start_page + n_pages {
860 let mut page = Page::new(PageType::Vector, page_id);
861 page.update_checksum();
862 if let Some(dirty_page) = self.cache.insert(page_id, page) {
863 let evicted_id = dirty_page.page_id();
864 self.write_page_raw(evicted_id, &dirty_page)?;
865 }
866 self.cache.mark_dirty(page_id);
867 }
868
869 Ok(super::ExtentId {
870 start_page,
871 n_pages,
872 })
873 }
874
875 pub fn free_page(&self, page_id: u32) -> Result<(), PagerError> {
877 if self.config.read_only {
878 return Err(PagerError::ReadOnly);
879 }
880
881 self.cache.remove(page_id);
883
884 let mut freelist = self.freelist_write()?;
886 freelist.free(page_id);
887
888 *self.header_dirty_lock()? = true;
889
890 Ok(())
891 }
892
893 pub fn header(&self) -> Result<DatabaseHeader, PagerError> {
895 Ok(self.header_read()?.clone())
896 }
897
898 pub fn physical_header(&self) -> Result<PhysicalFileHeader, PagerError> {
899 Ok(self.header_read()?.physical)
900 }
901
902 pub fn update_physical_header(&self, physical: PhysicalFileHeader) -> Result<(), PagerError> {
903 if self.config.read_only {
904 return Err(PagerError::ReadOnly);
905 }
906
907 let mut header = self.header_write()?;
908 header.physical = physical;
909 *self.header_dirty_lock()? = true;
910 Ok(())
911 }
912
913 pub fn page_count(&self) -> Result<u32, PagerError> {
915 Ok(self.header_read()?.page_count)
916 }
917
918 pub fn set_wal_writer(&self, wal: Arc<Mutex<crate::storage::wal::writer::WalWriter>>) {
930 let mut slot = self.wal.write().unwrap_or_else(|p| p.into_inner());
931 *slot = Some(wal);
932 }
933
934 pub fn clear_wal_writer(&self) {
936 let mut slot = self.wal.write().unwrap_or_else(|p| p.into_inner());
937 *slot = None;
938 }
939
940 pub fn has_wal_writer(&self) -> bool {
942 self.wal.read().map(|s| s.is_some()).unwrap_or(false)
943 }
944
945 pub fn flush(&self) -> Result<(), PagerError> {
947 if self.config.read_only {
948 return Ok(());
949 }
950
951 let trunks = {
953 let mut freelist = self.freelist_write()?;
954 if freelist.is_dirty() {
955 let mut header = self.header_write()?;
956 let trunks = freelist.flush_to_trunks(0, || {
957 let id = header.page_count;
958 header.page_count += 1;
959 id
960 });
961 header.freelist_head = freelist.trunk_head();
962 *self.header_dirty_lock()? = true;
963 freelist.mark_clean();
964 trunks
965 } else {
966 Vec::new()
967 }
968 };
969
970 for trunk in trunks {
971 let page_id = trunk.page_id();
972 self.cache.insert(page_id, trunk);
973 self.cache.mark_dirty(page_id);
974 }
975
976 let dirty_pages = self.cache.flush_dirty();
978 if !dirty_pages.is_empty() {
979 let max_lsn = dirty_pages
984 .iter()
985 .filter_map(|(_, page)| page.header().ok().map(|h| h.lsn))
986 .max()
987 .unwrap_or(0);
988 if max_lsn > 0 {
989 if let Ok(slot) = self.wal.read() {
990 if let Some(wal) = slot.as_ref() {
991 let wal = Arc::clone(wal);
992 drop(slot);
996 let mut wal_guard = wal.lock().unwrap_or_else(|p| p.into_inner());
997 wal_guard.flush_until(max_lsn).map_err(PagerError::Io)?;
998 }
999 }
1000 }
1001 self.write_pages_through_dwb(&dirty_pages)?;
1002 }
1003
1004 if *self.header_dirty_lock()? {
1006 self.write_header()?;
1007 }
1008
1009 Ok(())
1010 }
1011
1012 pub fn sync(&self) -> Result<(), PagerError> {
1014 self.flush()?;
1015
1016 let file = self.file_lock()?;
1017 file.sync_all()?;
1018
1019 Ok(())
1020 }
1021
1022 pub fn cache_stats(&self) -> crate::storage::engine::page_cache::CacheStats {
1024 self.cache.stats()
1025 }
1026
1027 pub fn cache_len(&self) -> usize {
1030 self.cache.len()
1031 }
1032
1033 pub fn cache_capacity(&self) -> usize {
1036 self.cache.capacity()
1037 }
1038
1039 pub fn dirty_page_count(&self) -> usize {
1041 self.cache.dirty_count()
1042 }
1043
1044 pub fn dirty_fraction(&self) -> f64 {
1048 let capacity = self.cache.capacity().max(1) as f64;
1049 self.cache.dirty_count() as f64 / capacity
1050 }
1051
1052 pub fn flush_some_dirty(&self, max: usize) -> Result<usize, PagerError> {
1056 if self.config.read_only || max == 0 {
1057 return Ok(0);
1058 }
1059 let dirty_pages = self.cache.flush_some_dirty(max);
1060 if dirty_pages.is_empty() {
1061 return Ok(0);
1062 }
1063 let count = dirty_pages.len();
1064 for (page_id, page) in dirty_pages {
1071 self.write_page(page_id, page)?;
1072 }
1073 Ok(count)
1074 }
1075
1076 pub fn path(&self) -> &Path {
1078 &self.path
1079 }
1080
1081 pub fn is_read_only(&self) -> bool {
1083 self.config.read_only
1084 }
1085
1086 pub fn file_size(&self) -> Result<u64, PagerError> {
1088 let file = self.file_lock()?;
1089 Ok(file.metadata()?.len())
1090 }
1091
1092 pub fn prefetch_hint(&self, page_id: u32) {
1100 if let Ok(file) = self.file_lock() {
1101 let _ = crate::storage::btree::prefetch::prefetch_page(
1102 &file,
1103 page_id as u64,
1104 PAGE_SIZE as u32,
1105 );
1106 }
1107 }
1108
1109 fn dwb_path(db_path: &Path) -> PathBuf {
1113 reddb_file::layout::pager_dwb_shadow_path(db_path)
1114 }
1115
1116 fn open_dwb_file(db_path: &Path) -> Result<File, PagerError> {
1121 Ok(OpenOptions::new()
1122 .read(true)
1123 .write(true)
1124 .create(true)
1125 .truncate(false)
1126 .open(Self::dwb_path(db_path))?)
1127 }
1128
1129 fn clear_dwb_file(file: &mut File) -> Result<(), PagerError> {
1131 file.set_len(0)?;
1132 file.seek(SeekFrom::Start(0))?;
1133 file.sync_all()?;
1134 Ok(())
1135 }
1136
1137 pub fn read_manifest_page(&self) -> Result<Page, PagerError> {
1142 self.read_page(1)
1143 .map_err(|_| PagerError::ManifestZoneCorrupt(self.path.clone()))
1144 }
1145
1146 fn write_pages_through_dwb(&self, pages: &[(u32, Page)]) -> Result<(), PagerError> {
1153 if let Some(dwb_mutex) = &self.dwb_file {
1154 let mut dwb = dwb_mutex.lock().map_err(|_| PagerError::LockPoisoned)?;
1155
1156 let buf = reddb_file::encode_paged_dwb_frame(
1157 pages
1158 .iter()
1159 .map(|(page_id, page)| (*page_id, page.as_bytes())),
1160 );
1161
1162 dwb.seek(SeekFrom::Start(0))?;
1164 dwb.write_all(&buf)?;
1165 dwb.set_len(buf.len() as u64)?;
1166 dwb.sync_all()?;
1167
1168 for (page_id, page) in pages {
1170 self.write_page_raw(*page_id, page)?;
1171 }
1172
1173 Self::clear_dwb_file(&mut dwb)?;
1175
1176 Ok(())
1177 } else {
1178 for (page_id, page) in pages {
1180 self.write_page_raw(*page_id, page)?;
1181 }
1182 Ok(())
1183 }
1184 }
1185
1186 fn recover_from_dwb(&self) -> Result<(), PagerError> {
1191 let dwb_path = Self::dwb_path(&self.path);
1192 if !dwb_path.exists() {
1193 return Ok(());
1194 }
1195
1196 if let Some(dwb_mutex) = &self.dwb_file {
1197 let mut file = dwb_mutex.lock().map_err(|_| PagerError::LockPoisoned)?;
1198 return self.recover_from_dwb_file(&mut file);
1199 }
1200
1201 let mut file = OpenOptions::new().read(true).write(true).open(&dwb_path)?;
1202 self.recover_from_dwb_file(&mut file)
1203 }
1204
1205 fn recover_from_dwb_file(&self, file: &mut File) -> Result<(), PagerError> {
1206 file.seek(SeekFrom::Start(0))?;
1207 let len = file.metadata()?.len();
1208 let mut buf = vec![0u8; len as usize];
1209 file.read_exact(&mut buf)?;
1210
1211 let entries = match reddb_file::decode_paged_dwb_frame(&buf) {
1212 Ok(entries) => entries,
1213 Err(_) => return Self::clear_dwb_file(file),
1214 };
1215
1216 for entry in entries {
1218 let page = Page::from_bytes(entry.page);
1219 self.write_page_raw(entry.page_id, &page)?;
1220 }
1221
1222 {
1224 let file = self.file_lock()?;
1225 file.sync_all()?;
1226 }
1227
1228 Self::clear_dwb_file(file)
1229 }
1230
1231 pub fn write_header_and_sync(&self) -> Result<(), PagerError> {
1233 self.write_header()?;
1234 let file = self.file_lock()?;
1235 file.sync_all()?;
1236 Ok(())
1237 }
1238
1239 pub fn set_checkpoint_in_progress(
1241 &self,
1242 in_progress: bool,
1243 target_lsn: u64,
1244 ) -> Result<(), PagerError> {
1245 let mut header = self.header_write()?;
1246 header.checkpoint_in_progress = in_progress;
1247 header.checkpoint_target_lsn = target_lsn;
1248 *self.header_dirty_lock()? = true;
1249 drop(header);
1250 self.write_header_and_sync()
1251 }
1252
1253 pub fn complete_checkpoint(&self, lsn: u64) -> Result<(), PagerError> {
1255 let mut header = self.header_write()?;
1256 header.checkpoint_lsn = lsn;
1257 header.checkpoint_in_progress = false;
1258 header.checkpoint_target_lsn = 0;
1259 *self.header_dirty_lock()? = true;
1260 drop(header);
1261 self.write_header_and_sync()
1262 }
1263}
1264
1265#[cfg(test)]
1266mod tests {
1267 use super::*;
1268
1269 fn temp_db_path(name: &str) -> PathBuf {
1270 std::env::temp_dir().join(format!(
1271 "reddb-pager-{}-{}-{}.rdb",
1272 name,
1273 std::process::id(),
1274 crate::utils::now_unix_nanos()
1275 ))
1276 }
1277
1278 fn cleanup(path: &Path) {
1279 let _ = std::fs::remove_file(path);
1280 let _ = std::fs::remove_file(Pager::dwb_path(path));
1281 }
1282
1283 fn read_zone(path: &Path) -> Vec<u8> {
1284 let mut zone = vec![0u8; SUPERBLOCK_ZONE_SIZE];
1285 let mut file = File::open(path).expect("open() should succeed");
1286 file.read_exact(&mut zone)
1287 .expect("read_exact() should succeed");
1288 zone
1289 }
1290
1291 fn poke_slot(path: &Path, copy_index: usize, bytes: &[u8]) {
1293 let mut file = OpenOptions::new()
1294 .write(true)
1295 .open(path)
1296 .expect("open() should succeed");
1297 file.seek(SeekFrom::Start(reddb_file::paged_superblock_slot_offset(
1298 copy_index,
1299 )))
1300 .expect("operation should succeed");
1301 file.write_all(bytes).expect("write_all() should succeed");
1302 file.sync_all().expect("sync_all() should succeed");
1303 }
1304
1305 #[test]
1306 fn open_refuses_future_database_version() {
1307 let path = temp_db_path("future-version");
1308 let pager = Pager::open_default(&path).expect("open_default() should succeed");
1309 drop(pager);
1310
1311 let selection = reddb_file::select_paged_superblock(&read_zone(&path))
1315 .expect("operation should succeed");
1316 let mut slot = vec![0u8; SUPERBLOCK_SLOT_SIZE];
1317 reddb_file::init_database_header_page(&mut slot, 1)
1318 .expect("init_database_header_page() should succeed");
1319 reddb_file::set_database_header_version(&mut slot, reddb_file::PAGE_FILE_VERSION + 1)
1320 .expect("operation should succeed");
1321 reddb_file::seal_paged_superblock_slot(
1322 &mut slot,
1323 selection.copy_index,
1324 selection.generation + 1,
1325 )
1326 .expect("operation should succeed");
1327 poke_slot(&path, selection.copy_index, &slot);
1328
1329 let err = match Pager::open_default(&path) {
1330 Ok(_) => panic!("future database version should be rejected"),
1331 Err(err) => err,
1332 };
1333 match err {
1334 PagerError::InvalidDatabase(msg) => {
1335 assert!(msg.contains("newer than supported"), "{msg}");
1336 }
1337 other => panic!("expected InvalidDatabase, got {other:?}"),
1338 }
1339
1340 cleanup(&path);
1341 }
1342
1343 #[test]
1344 fn a_fresh_store_publishes_both_superblock_copies() {
1345 let path = temp_db_path("fresh-pair");
1348 let pager = Pager::open_default(&path).expect("open_default() should succeed");
1349 drop(pager);
1350
1351 let zone = read_zone(&path);
1352 for copy_index in 0..reddb_file::PAGED_SUPERBLOCK_SLOT_COUNT {
1353 let start = copy_index * SUPERBLOCK_SLOT_SIZE;
1354 assert!(
1355 reddb_file::paged_superblock_slot_generation(
1356 &zone[start..start + SUPERBLOCK_SLOT_SIZE],
1357 copy_index
1358 )
1359 .is_some(),
1360 "copy {copy_index} must be valid on a fresh store"
1361 );
1362 }
1363 cleanup(&path);
1364 }
1365
1366 #[test]
1367 fn an_update_writes_the_stale_copy_and_leaves_the_newest_one_intact() {
1368 let path = temp_db_path("ping-pong");
1369 let pager = Pager::open_default(&path).expect("open_default() should succeed");
1370 let before = reddb_file::select_paged_superblock(&read_zone(&path))
1371 .expect("operation should succeed");
1372
1373 pager
1374 .allocate_page(PageType::BTreeLeaf)
1375 .expect("allocate_page() should succeed");
1376 pager.sync().expect("sync() should succeed");
1377
1378 let zone = read_zone(&path);
1379 let after = reddb_file::select_paged_superblock(&zone)
1380 .expect("select_paged_superblock() should succeed");
1381 assert_ne!(
1382 after.copy_index, before.copy_index,
1383 "an update must target the stale copy"
1384 );
1385 assert!(after.generation > before.generation);
1386
1387 let start = before.copy_index * SUPERBLOCK_SLOT_SIZE;
1391 assert_eq!(
1392 reddb_file::paged_superblock_slot_generation(
1393 &zone[start..start + SUPERBLOCK_SLOT_SIZE],
1394 before.copy_index
1395 ),
1396 Some(before.generation)
1397 );
1398
1399 drop(pager);
1400 cleanup(&path);
1401 }
1402
1403 #[test]
1404 fn open_recovers_from_the_older_copy_when_the_newest_is_torn() {
1405 let path = temp_db_path("torn-newest");
1406 {
1407 let pager = Pager::open_default(&path).expect("open_default() should succeed");
1408 pager
1409 .allocate_page(PageType::BTreeLeaf)
1410 .expect("allocate_page() should succeed");
1411 pager.sync().expect("sync() should succeed");
1412 }
1413
1414 let newest = reddb_file::select_paged_superblock(&read_zone(&path))
1415 .expect("operation should succeed");
1416 let mut torn = vec![0u8; SUPERBLOCK_SLOT_SIZE];
1418 torn.copy_from_slice(
1419 &read_zone(&path)[newest.copy_index * SUPERBLOCK_SLOT_SIZE
1420 ..newest.copy_index * SUPERBLOCK_SLOT_SIZE + SUPERBLOCK_SLOT_SIZE],
1421 );
1422 torn[64] ^= 0xFF;
1423 poke_slot(&path, newest.copy_index, &torn);
1424
1425 let pager = Pager::open_default(&path).expect("older copy must root the store");
1426 assert!(pager.page_count().expect("page_count() should succeed") >= 3);
1427 drop(pager);
1428 cleanup(&path);
1429 }
1430
1431 #[test]
1432 fn open_refuses_didactically_when_both_superblock_copies_are_invalid() {
1433 let path = temp_db_path("both-invalid");
1434 {
1435 let _ = Pager::open_default(&path).expect("open_default() should succeed");
1436 }
1437
1438 let zone = read_zone(&path);
1439 for copy_index in 0..reddb_file::PAGED_SUPERBLOCK_SLOT_COUNT {
1440 let start = copy_index * SUPERBLOCK_SLOT_SIZE;
1441 let mut slot = zone[start..start + SUPERBLOCK_SLOT_SIZE].to_vec();
1442 slot[64] ^= 0xFF;
1443 poke_slot(&path, copy_index, &slot);
1444 }
1445
1446 let err = match Pager::open_default(&path) {
1447 Ok(_) => panic!("a rootless store must not open"),
1448 Err(err) => err,
1449 };
1450 let rendered = err.to_string();
1451 assert!(matches!(err, PagerError::SuperblockZoneUnrecoverable(_)));
1452 assert!(rendered.contains("superblock zone"), "{rendered}");
1453 assert!(rendered.contains("salvage"), "{rendered}");
1454
1455 cleanup(&path);
1456 }
1457
1458 #[test]
1459 fn open_refuses_a_legacy_sidecar_store_and_names_the_migration_tool() {
1460 let path = temp_db_path("legacy-sidecar");
1461 {
1462 let _ = Pager::open_default(&path).expect("open_default() should succeed");
1463 }
1464
1465 let sidecar = reddb_file::layout::retired::pager_header_path_v0(&path);
1466 std::fs::write(&sidecar, b"legacy header shadow").expect("write() should succeed");
1467
1468 let err = match Pager::open_default(&path) {
1469 Ok(_) => panic!("legacy stores need migration first"),
1470 Err(err) => err,
1471 };
1472 let rendered = err.to_string();
1473 assert!(matches!(err, PagerError::LegacySidecarStore { .. }));
1474 assert!(rendered.contains("migration tool"), "{rendered}");
1475
1476 let _ = std::fs::remove_file(&sidecar);
1477 cleanup(&path);
1478 }
1479
1480 #[test]
1481 fn no_phase_one_sidecar_survives_a_create_write_reopen_cycle() {
1482 let path = temp_db_path("census");
1483 {
1484 let pager = Pager::open_default(&path).expect("open_default() should succeed");
1485 let page = pager
1486 .allocate_page(PageType::BTreeLeaf)
1487 .expect("allocate_page() should succeed");
1488 pager
1489 .write_page(page.page_id(), page)
1490 .expect("operation should succeed");
1491 pager.sync().expect("sync() should succeed");
1492 }
1493 {
1494 let pager = Pager::open_default(&path).expect("open_default() should succeed");
1495 pager.sync().expect("sync() should succeed");
1496 }
1497
1498 assert_eq!(
1499 reddb_file::layout::retired::first_present_phase1_sidecar(&path),
1500 None,
1501 "the pager must never create a retired phase-1 sidecar"
1502 );
1503 cleanup(&path);
1504 }
1505}