1use std::collections::HashMap;
8use std::fs::{self, File, OpenOptions};
9use std::io::{Read, Seek, SeekFrom, Write};
10use std::path::{Path, PathBuf};
11use std::sync::{Arc, Mutex, OnceLock};
12use std::time::{SystemTime, UNIX_EPOCH};
13
14use fs2::FileExt;
15
16pub type RdbFileResult<T> = Result<T, RdbFileError>;
17
18pub const DEFAULT_FORMAT_VERSION: u32 = 1;
19
20#[derive(Debug)]
21pub enum RdbFileError {
22 InvalidOperation(String),
23 Io(std::io::Error),
24 ZoneUnrecoverable {
28 zone: &'static str,
29 path: PathBuf,
30 },
31}
32
33impl std::fmt::Display for RdbFileError {
34 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
35 match self {
36 Self::InvalidOperation(msg) => write!(f, "INVALID_OPERATION: {msg}"),
37 Self::Io(err) => write!(f, "io error: {err}"),
38 Self::ZoneUnrecoverable { zone, path } => write!(
39 f,
40 "{zone} zone of {} failed validation, so the store will not be opened \
41 (opening it could only return data the zone can no longer vouch for). \
42 Run the `SCRUB` statement to classify the fault and `red salvage` to \
43 extract every entity the damage did not touch; `red salvage` never writes \
44 into the damaged file (ADR 0074 §2/§4).",
45 path.display()
46 ),
47 }
48 }
49}
50
51impl std::error::Error for RdbFileError {}
52
53impl From<std::io::Error> for RdbFileError {
54 fn from(err: std::io::Error) -> Self {
55 Self::Io(err)
56 }
57}
58
59fn crc32(data: &[u8]) -> u32 {
60 let mut hasher = crc32fast::Hasher::new();
61 hasher.update(data);
62 hasher.finalize()
63}
64
65pub const EMBEDDED_RDB_SUPERBLOCK_SIZE: u64 = 4096;
66pub const EMBEDDED_RDB_SUPERBLOCK_0_OFFSET: u64 = 0;
67pub const EMBEDDED_RDB_SUPERBLOCK_1_OFFSET: u64 = EMBEDDED_RDB_SUPERBLOCK_SIZE;
68
69pub const EMBEDDED_RDB_MANIFEST_SLOT_SIZE: u64 = 4096;
78pub const EMBEDDED_RDB_MANIFEST_0_OFFSET: u64 = EMBEDDED_RDB_SUPERBLOCK_SIZE * 2;
79pub const EMBEDDED_RDB_MANIFEST_1_OFFSET: u64 =
80 EMBEDDED_RDB_MANIFEST_0_OFFSET + EMBEDDED_RDB_MANIFEST_SLOT_SIZE;
81pub const EMBEDDED_RDB_MANIFEST_ZONE_END: u64 =
82 EMBEDDED_RDB_MANIFEST_1_OFFSET + EMBEDDED_RDB_MANIFEST_SLOT_SIZE;
83
84const SUPERBLOCK_MAGIC: &[u8; 8] = b"RDBSBLK1";
85const MANIFEST_MAGIC: &[u8; 8] = b"RDBMNFS1";
86const SUPERBLOCK_VERSION: u32 = 2;
87const MANIFEST_VERSION: u32 = 2;
88const LEGACY_SUPERBLOCK_VERSION: u32 = 1;
89const LEGACY_MANIFEST_VERSION: u32 = 1;
90const CHECKSUM_LEN: usize = 4;
91const MANIFEST_REGION_BYTES: u64 = EMBEDDED_RDB_MANIFEST_SLOT_SIZE;
92const WAL_REGION_BYTES: u64 = 64 * 1024;
93const SNAPSHOT_ALIGNMENT: u64 = 4096;
94const SNAPSHOT_MAGIC: &[u8; 4] = b"RDST";
95const WAL_FRAME_MAGIC: &[u8; 8] = b"RDBEWAL1";
96const WAL_FRAME_VERSION: u16 = 2;
97const WAL_FRAME_HEADER_BYTES: usize = 8 + 2 + 2 + 8 + 4 + 4 + 4 + 4;
98const LEGACY_WAL_FRAME_HEADER_BYTES: usize = 8 + 4 + 4;
99const CRASH_INJECT_ENV: &str = "REDDB_EMBEDDED_RDB_CRASH_AT";
100
101#[derive(Debug, Clone, Copy, PartialEq, Eq)]
102pub struct EmbeddedRdbManifest {
103 pub version: u32,
104 pub wal_region_offset: u64,
105 pub wal_region_bytes: u64,
106 pub wal_recovery_boundary: u64,
107 pub wal_live_bytes: u64,
108 pub snapshot_offset: u64,
109 pub snapshot_bytes: u64,
110 pub snapshot_checksum: u32,
111 pub created_at_unix_ms: u128,
112 pub checksum: u32,
113}
114
115#[derive(Debug, Clone, Copy, PartialEq, Eq)]
116pub struct EmbeddedRdbSuperblock {
117 pub copy_index: u8,
118 pub generation: u64,
119 pub format_version: u32,
120 pub manifest_offset: u64,
121 pub manifest_len: u64,
122 pub manifest_checksum: u32,
123 pub wal_region_offset: u64,
124 pub wal_region_bytes: u64,
125 pub wal_recovery_boundary: u64,
126 pub wal_live_bytes: u64,
127 pub snapshot_offset: u64,
128 pub snapshot_bytes: u64,
129 pub snapshot_checksum: u32,
130 pub checksum: u32,
131}
132
133#[derive(Debug, Clone, PartialEq, Eq)]
134pub struct EmbeddedRdbOpen {
135 pub path: PathBuf,
136 pub selected_superblock: EmbeddedRdbSuperblock,
137 pub manifest: EmbeddedRdbManifest,
138}
139
140#[derive(Debug, Default)]
141struct WalScan {
142 payloads: Vec<Vec<u8>>,
143 next_sequence: u64,
144 previous_frame_crc: u32,
145 valid_bytes: u64,
146}
147
148#[derive(Default)]
149struct EmbeddedPathState {
150 tail: Option<CachedWalTail>,
151}
152
153struct CachedWalTail {
154 stamp: FileMutationStamp,
157 superblock: EmbeddedRdbSuperblock,
158 next_sequence: u64,
159 previous_frame_crc: u32,
160 valid_bytes: u64,
161}
162
163#[derive(PartialEq, Eq)]
167struct FileMutationStamp {
168 device: u64,
169 inode: u64,
170 len: u64,
171 modified: (i64, i64),
172 changed: (i64, i64),
173}
174
175fn file_mutation_stamp(file: &File) -> std::io::Result<Option<FileMutationStamp>> {
176 #[cfg(unix)]
177 {
178 use std::os::unix::fs::MetadataExt;
179 let metadata = file.metadata()?;
180 Ok(Some(FileMutationStamp {
181 device: metadata.dev(),
182 inode: metadata.ino(),
183 len: metadata.len(),
184 modified: (metadata.mtime(), metadata.mtime_nsec()),
185 changed: (metadata.ctime(), metadata.ctime_nsec()),
186 }))
187 }
188 #[cfg(not(unix))]
189 {
190 let _ = file;
191 Ok(None)
192 }
193}
194
195pub struct EmbeddedRdbWriterLock {
209 #[cfg_attr(not(unix), allow(dead_code))]
210 path: PathBuf,
211 _file: File,
212}
213
214impl EmbeddedRdbWriterLock {
215 pub fn acquire(data_path: impl AsRef<Path>) -> RdbFileResult<Self> {
216 let data_path = data_path.as_ref();
217 let path = crate::layout::embedded_writer_lock_path(data_path);
218 if let Some(parent) = path.parent() {
219 if !parent.as_os_str().is_empty() {
220 fs::create_dir_all(parent)?;
221 }
222 }
223 loop {
224 let file = OpenOptions::new()
225 .create(true)
226 .truncate(false)
227 .read(true)
228 .write(true)
229 .open(&path)?;
230 if let Err(err) = file.try_lock_exclusive() {
231 if err.raw_os_error() == fs2::lock_contended_error().raw_os_error() {
232 return Err(RdbFileError::InvalidOperation(format!(
233 "{} is already open for writing by another process; an embedded \
234 single-file store has exactly one writer process",
235 data_path.display()
236 )));
237 }
238 return Err(err.into());
239 }
240 if lock_file_is_live(&file, &path)? {
244 return Ok(Self { path, _file: file });
245 }
246 }
247 }
248}
249
250impl Drop for EmbeddedRdbWriterLock {
251 fn drop(&mut self) {
252 #[cfg(unix)]
257 {
258 let _ = fs::remove_file(&self.path);
259 }
260 }
261}
262
263fn lock_file_is_live(file: &File, path: &Path) -> RdbFileResult<bool> {
264 #[cfg(unix)]
265 {
266 use std::os::unix::fs::MetadataExt;
267 let held = file.metadata()?;
268 match fs::metadata(path) {
269 Ok(current) => Ok(held.dev() == current.dev() && held.ino() == current.ino()),
270 Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(false),
271 Err(err) => Err(err.into()),
272 }
273 }
274 #[cfg(not(unix))]
275 {
276 let _ = (file, path);
277 Ok(true)
278 }
279}
280
281pub struct EmbeddedRdbArtifact;
282
283impl EmbeddedRdbArtifact {
284 pub fn create(path: impl AsRef<Path>) -> RdbFileResult<EmbeddedRdbOpen> {
285 Self::create_with_snapshot(path, &[])
286 }
287
288 pub fn create_with_snapshot(
289 path: impl AsRef<Path>,
290 snapshot: &[u8],
291 ) -> RdbFileResult<EmbeddedRdbOpen> {
292 let path = path.as_ref();
293 if let Some(parent) = path.parent() {
294 if !parent.as_os_str().is_empty() {
295 fs::create_dir_all(parent)?;
296 }
297 }
298
299 let created_at_unix_ms = now_unix_ms();
300 let wal_region_offset = EMBEDDED_RDB_MANIFEST_ZONE_END;
301 let snapshot_offset = wal_region_offset + WAL_REGION_BYTES;
302 let manifest = EmbeddedRdbManifest {
303 version: MANIFEST_VERSION,
304 wal_region_offset,
305 wal_region_bytes: WAL_REGION_BYTES,
306 wal_recovery_boundary: wal_region_offset,
307 wal_live_bytes: 0,
308 snapshot_offset,
309 snapshot_bytes: snapshot.len() as u64,
310 snapshot_checksum: crc32(snapshot),
311 created_at_unix_ms,
312 checksum: 0,
313 };
314 let manifest_bytes = encode_manifest(manifest);
315 let manifest_checksum = trailer_checksum(&manifest_bytes);
316
317 let mut file = OpenOptions::new()
318 .create(true)
319 .truncate(true)
320 .read(true)
321 .write(true)
322 .open(path)?;
323 file.set_len(snapshot_offset + snapshot.len() as u64)?;
324 write_at(&mut file, EMBEDDED_RDB_MANIFEST_0_OFFSET, &manifest_bytes)?;
325 if !snapshot.is_empty() {
326 write_at(&mut file, snapshot_offset, snapshot)?;
327 }
328
329 let base = EmbeddedRdbSuperblock {
330 copy_index: 0,
331 generation: 1,
332 format_version: DEFAULT_FORMAT_VERSION,
333 manifest_offset: EMBEDDED_RDB_MANIFEST_0_OFFSET,
334 manifest_len: manifest_bytes.len() as u64,
335 manifest_checksum,
336 wal_region_offset,
337 wal_region_bytes: WAL_REGION_BYTES,
338 wal_recovery_boundary: wal_region_offset,
339 wal_live_bytes: 0,
340 snapshot_offset,
341 snapshot_bytes: snapshot.len() as u64,
342 snapshot_checksum: crc32(snapshot),
343 checksum: 0,
344 };
345 Self::write_superblock_copy(&mut file, &base)?;
346 Self::write_superblock_copy(
347 &mut file,
348 &EmbeddedRdbSuperblock {
349 copy_index: 1,
350 generation: 2,
351 ..base
352 },
353 )?;
354 file.sync_all()?;
355
356 Self::open(path)
357 }
358
359 pub fn open(path: impl AsRef<Path>) -> RdbFileResult<EmbeddedRdbOpen> {
360 Self::open_inner(path, true)
361 }
362
363 fn open_inner(
364 path: impl AsRef<Path>,
365 validate_snapshot_refs: bool,
366 ) -> RdbFileResult<EmbeddedRdbOpen> {
367 let path = path.as_ref();
368 let mut file = File::open(path)?;
369 Self::open_inner_with_file(&mut file, path, validate_snapshot_refs)
370 }
371
372 fn open_inner_with_file(
386 file: &mut File,
387 path: &Path,
388 validate_snapshot_refs: bool,
389 ) -> RdbFileResult<EmbeddedRdbOpen> {
390 let mut superblocks: Vec<EmbeddedRdbSuperblock> = [
393 read_superblock_copy(file, 0)?,
394 read_superblock_copy(file, 1)?,
395 ]
396 .into_iter()
397 .flatten()
398 .collect();
399 superblocks.sort_by_key(|superblock| std::cmp::Reverse(superblock.generation));
400
401 if superblocks.is_empty() {
402 return Err(RdbFileError::ZoneUnrecoverable {
403 zone: "superblock",
404 path: path.to_path_buf(),
405 });
406 }
407
408 for selected_superblock in superblocks {
409 let mut manifest = read_manifest(file, selected_superblock).map_err(|_| {
415 RdbFileError::ZoneUnrecoverable {
416 zone: "manifest",
417 path: path.to_path_buf(),
418 }
419 })?;
420 manifest.wal_recovery_boundary = selected_superblock.wal_recovery_boundary;
421 manifest.wal_live_bytes = selected_superblock.wal_live_bytes;
422 if validate_snapshot_refs && !snapshot_reference_valid(file, &manifest)? {
423 continue;
424 }
425 return Ok(EmbeddedRdbOpen {
426 path: path.to_path_buf(),
427 selected_superblock,
428 manifest,
429 });
430 }
431
432 Err(RdbFileError::ZoneUnrecoverable {
433 zone: "snapshot",
434 path: path.to_path_buf(),
435 })
436 }
437
438 pub fn wal_payloads_encoded_len(payloads: &[Vec<u8>]) -> RdbFileResult<u64> {
439 let mut len = 0u64;
440 for payload in payloads {
441 let payload_len = u32::try_from(payload.len()).map_err(|_| {
442 RdbFileError::InvalidOperation("embedded wal payload too large".into())
443 })?;
444 let frame_len = WAL_FRAME_HEADER_BYTES as u64 + payload_len as u64;
445 len = len.checked_add(frame_len).ok_or_else(|| {
446 RdbFileError::InvalidOperation("embedded wal encoded length overflow".into())
447 })?;
448 }
449 Ok(len)
450 }
451
452 pub fn write_snapshot_with_wal_capacity(
453 path: impl AsRef<Path>,
454 snapshot: &[u8],
455 min_wal_bytes: u64,
456 ) -> RdbFileResult<EmbeddedRdbOpen> {
457 let path = path.as_ref();
458 let path_lock = embedded_path_lock(path);
459 let mut path_guard = path_lock
460 .lock()
461 .unwrap_or_else(|poisoned| poisoned.into_inner());
462 path_guard.tail = None;
463 let mut file = OpenOptions::new().read(true).write(true).open(path)?;
467 file.lock_exclusive()?;
468
469 let open = Self::open_inner_with_file(&mut file, path, true)?;
470 let wal_scan = match wal_region_start_checked(&open)? {
471 Some(wal_start) => scan_wal_with_file(&mut file, &open, wal_start)?,
472 None => WalScan {
473 next_sequence: 1,
474 ..WalScan::default()
475 },
476 };
477 let checkpoint_boundary = wal_boundary_after_live_bytes(&open, wal_scan.valid_bytes)?;
478 let wal_region_bytes =
479 grow_wal_region_bytes(open.manifest.wal_region_bytes, min_wal_bytes)?;
480 let snapshot_checksum = crc32(snapshot);
481 if open.manifest.wal_live_bytes == 0
486 && wal_scan.valid_bytes == 0
487 && wal_region_bytes == open.manifest.wal_region_bytes
488 && open.manifest.snapshot_bytes == snapshot.len() as u64
489 && open.manifest.snapshot_checksum == snapshot_checksum
490 && published_snapshot_matches(&mut file, &open, snapshot)?
491 {
492 FileExt::unlock(&file)?;
493 return Ok(open);
494 }
495 let snapshot_offset = next_snapshot_offset(&mut file, &open, wal_region_bytes, snapshot)?;
496 let manifest = EmbeddedRdbManifest {
497 wal_region_bytes,
498 wal_recovery_boundary: checkpoint_boundary,
499 wal_live_bytes: 0,
500 snapshot_offset,
501 snapshot_bytes: snapshot.len() as u64,
502 snapshot_checksum,
503 checksum: 0,
504 ..open.manifest
505 };
506 let manifest_bytes = encode_manifest(manifest);
507 let manifest_checksum = trailer_checksum(&manifest_bytes);
508
509 let snapshot_end = snapshot_offset + snapshot.len() as u64;
512 if file.metadata()?.len() < snapshot_end {
513 file.set_len(snapshot_end)?;
514 }
515 if !snapshot.is_empty() {
516 write_at(&mut file, snapshot_offset, snapshot)?;
517 }
518 crash_inject("snapshot_after_image_write");
519 file.sync_data()?;
520 crash_inject("snapshot_after_image_sync");
521
522 let next_manifest_offset = inactive_manifest_offset(open.selected_superblock)?;
527 write_at(&mut file, next_manifest_offset, &manifest_bytes)?;
528 crash_inject("snapshot_after_manifest_write");
529 file.sync_data()?;
530 crash_inject("snapshot_after_manifest_sync");
531
532 let next_copy_index = if open.selected_superblock.copy_index == 0 {
533 1
534 } else {
535 0
536 };
537 let next_superblock = EmbeddedRdbSuperblock {
538 copy_index: next_copy_index,
539 generation: open.selected_superblock.generation.saturating_add(1),
540 manifest_offset: next_manifest_offset,
541 manifest_len: manifest_bytes.len() as u64,
542 manifest_checksum,
543 wal_region_bytes,
544 wal_recovery_boundary: checkpoint_boundary,
545 wal_live_bytes: 0,
546 snapshot_offset,
547 snapshot_bytes: snapshot.len() as u64,
548 snapshot_checksum,
549 checksum: 0,
550 ..open.selected_superblock
551 };
552 Self::write_superblock_copy(&mut file, &next_superblock)?;
553 crash_inject("snapshot_after_superblock_write");
554 file.sync_all()?;
555
556 let retained_end = snapshot_end
560 .max(open.manifest.snapshot_offset + open.manifest.snapshot_bytes)
561 .max(open.manifest.wal_region_offset + wal_region_bytes);
562 if file.metadata()?.len() > retained_end {
563 file.set_len(retained_end)?;
564 file.sync_all()?;
565 }
566 FileExt::unlock(&file)?;
569 Self::open(path)
570 }
571
572 pub fn open_strict_manifest(path: impl AsRef<Path>) -> RdbFileResult<EmbeddedRdbOpen> {
573 let path = path.as_ref();
574 let mut file = File::open(path)?;
575 let selected_superblock = [
576 read_superblock_copy(&mut file, 0)?,
577 read_superblock_copy(&mut file, 1)?,
578 ]
579 .into_iter()
580 .flatten()
581 .max_by_key(|superblock| superblock.generation)
582 .ok_or_else(|| RdbFileError::InvalidOperation("no valid embedded superblock".into()))?;
583
584 let mut manifest = read_manifest(&mut file, selected_superblock)?;
585 manifest.wal_recovery_boundary = selected_superblock.wal_recovery_boundary;
586 manifest.wal_live_bytes = selected_superblock.wal_live_bytes;
590 Ok(EmbeddedRdbOpen {
591 path: path.to_path_buf(),
592 selected_superblock,
593 manifest,
594 })
595 }
596
597 pub fn read_snapshot(open: &EmbeddedRdbOpen) -> RdbFileResult<Option<Vec<u8>>> {
598 if open.manifest.snapshot_bytes == 0 {
599 return Ok(None);
600 }
601 let mut file = File::open(&open.path)?;
602 let mut bytes = vec![0u8; open.manifest.snapshot_bytes as usize];
603 file.seek(SeekFrom::Start(open.manifest.snapshot_offset))?;
604 file.read_exact(&mut bytes)?;
605 let checksum = crc32(&bytes);
606 if checksum != open.manifest.snapshot_checksum {
607 return Err(RdbFileError::InvalidOperation(format!(
608 "embedded snapshot checksum mismatch: stored {:#010x}, computed {:#010x}",
609 open.manifest.snapshot_checksum, checksum
610 )));
611 }
612 if bytes.len() >= SNAPSHOT_MAGIC.len() && &bytes[..SNAPSHOT_MAGIC.len()] != SNAPSHOT_MAGIC {
613 return Err(RdbFileError::InvalidOperation(
614 "invalid embedded snapshot magic".into(),
615 ));
616 }
617 Ok(Some(bytes))
618 }
619
620 pub fn write_snapshot(
621 path: impl AsRef<Path>,
622 snapshot: &[u8],
623 ) -> RdbFileResult<EmbeddedRdbOpen> {
624 Self::write_snapshot_with_wal_capacity(path, snapshot, 0)
625 }
626
627 pub fn read_wal_payloads(open: &EmbeddedRdbOpen) -> RdbFileResult<Vec<Vec<u8>>> {
628 Ok(scan_wal(open)?.payloads)
629 }
630
631 pub fn append_wal_payloads(
632 path: impl AsRef<Path>,
633 payloads: &[Vec<u8>],
634 ) -> RdbFileResult<EmbeddedRdbOpen> {
635 let path = path.as_ref();
636 if payloads.is_empty() {
637 return Self::open(path);
638 }
639
640 let path_lock = embedded_path_lock(path);
641 let mut path_guard = path_lock
642 .lock()
643 .unwrap_or_else(|poisoned| poisoned.into_inner());
644 let cached_tail = path_guard.tail.take();
647 let mut file = OpenOptions::new().read(true).write(true).open(path)?;
650 file.lock_exclusive()?;
651
652 let open = Self::open_inner_with_file(&mut file, path, false)?;
653 let stamp = file_mutation_stamp(&file)?;
654 let cached_tail = cached_tail.filter(|tail| {
655 stamp.as_ref() == Some(&tail.stamp) && tail.superblock == open.selected_superblock
656 });
657 let snapshot_validated = cached_tail.is_some();
658 let wal_scan = match cached_tail {
659 Some(tail) => WalScan {
660 next_sequence: tail.next_sequence,
661 previous_frame_crc: tail.previous_frame_crc,
662 valid_bytes: tail.valid_bytes,
663 payloads: Vec::new(),
664 },
665 None => match wal_region_start_checked(&open)? {
666 Some(wal_start) => scan_wal_with_file(&mut file, &open, wal_start)?,
667 None => WalScan {
668 next_sequence: 1,
669 ..WalScan::default()
670 },
671 },
672 };
673 if wal_scan.valid_bytes != open.manifest.wal_live_bytes {
676 return Err(RdbFileError::ZoneUnrecoverable {
677 zone: "wal",
678 path: path.to_path_buf(),
679 });
680 }
681 let mut sequence = wal_scan.next_sequence;
682 let mut previous_frame_crc = wal_scan.previous_frame_crc;
683 let mut encoded = Vec::new();
684 for payload in payloads {
685 let (frame, frame_crc) = encode_wal_frame(sequence, previous_frame_crc, payload)?;
686 encoded.extend_from_slice(&frame);
687 previous_frame_crc = frame_crc;
688 sequence = sequence.saturating_add(1);
689 }
690
691 let encoded_len = encoded.len() as u64;
692 let free_bytes = open
693 .manifest
694 .wal_region_bytes
695 .checked_sub(wal_scan.valid_bytes)
696 .ok_or_else(|| {
697 RdbFileError::InvalidOperation("embedded wal live bytes overflow".into())
698 })?;
699 if encoded_len > free_bytes {
700 return Err(RdbFileError::InvalidOperation(
701 "embedded wal region full".into(),
702 ));
703 }
704 let next_boundary =
705 wal_boundary_after_live_bytes(&open, wal_scan.valid_bytes + encoded_len)?;
706
707 write_circular_wal_bytes(&mut file, &open, &encoded)?;
708 crash_inject("wal_after_frame_write");
709 file.sync_data()?;
710 crash_inject("wal_after_frame_sync");
711
712 let next_copy_index = if open.selected_superblock.copy_index == 0 {
713 1
714 } else {
715 0
716 };
717 let next_superblock = EmbeddedRdbSuperblock {
718 copy_index: next_copy_index,
719 generation: open.selected_superblock.generation.saturating_add(1),
720 wal_recovery_boundary: next_boundary,
721 wal_live_bytes: wal_scan.valid_bytes + encoded_len,
722 checksum: 0,
723 ..open.selected_superblock
724 };
725 Self::write_superblock_copy(&mut file, &next_superblock)?;
726 crash_inject("wal_after_superblock_write");
727 file.sync_all()?;
728 let next_open = Self::open_inner_with_file(&mut file, path, !snapshot_validated)?;
735 let next_stamp = file_mutation_stamp(&file)?;
736 FileExt::unlock(&file)?;
737 if let Some(stamp) = next_stamp
738 .filter(|_| next_open.selected_superblock.generation == next_superblock.generation)
739 {
740 path_guard.tail = Some(CachedWalTail {
741 stamp,
742 superblock: next_open.selected_superblock,
743 next_sequence: sequence,
744 previous_frame_crc,
745 valid_bytes: wal_scan.valid_bytes + encoded_len,
746 });
747 }
748 Ok(next_open)
749 }
750
751 pub fn write_superblock_copy(
752 file: &mut File,
753 superblock: &EmbeddedRdbSuperblock,
754 ) -> RdbFileResult<()> {
755 let offset = superblock_offset(superblock.copy_index)?;
756 write_at(file, offset, &encode_superblock(*superblock)?)?;
757 Ok(())
758 }
759}
760
761fn read_superblock_copy(
762 file: &mut File,
763 copy_index: u8,
764) -> RdbFileResult<Option<EmbeddedRdbSuperblock>> {
765 let offset = superblock_offset(copy_index)?;
766 let mut bytes = vec![0u8; EMBEDDED_RDB_SUPERBLOCK_SIZE as usize];
767 file.seek(SeekFrom::Start(offset))?;
768 file.read_exact(&mut bytes)?;
769 Ok(decode_superblock(copy_index, &bytes).ok())
770}
771
772fn inactive_manifest_offset(superblock: EmbeddedRdbSuperblock) -> RdbFileResult<u64> {
779 match superblock.manifest_offset {
780 EMBEDDED_RDB_MANIFEST_0_OFFSET => Ok(EMBEDDED_RDB_MANIFEST_1_OFFSET),
781 EMBEDDED_RDB_MANIFEST_1_OFFSET => Ok(EMBEDDED_RDB_MANIFEST_0_OFFSET),
782 other => Err(RdbFileError::InvalidOperation(format!(
783 "superblock names a manifest offset outside the zone: {other}"
784 ))),
785 }
786}
787
788fn read_manifest(
789 file: &mut File,
790 superblock: EmbeddedRdbSuperblock,
791) -> RdbFileResult<EmbeddedRdbManifest> {
792 if superblock.manifest_len < CHECKSUM_LEN as u64
793 || superblock.manifest_len > MANIFEST_REGION_BYTES
794 {
795 return Err(RdbFileError::InvalidOperation(format!(
796 "invalid embedded manifest length {}",
797 superblock.manifest_len
798 )));
799 }
800 if !matches!(
803 superblock.manifest_offset,
804 EMBEDDED_RDB_MANIFEST_0_OFFSET | EMBEDDED_RDB_MANIFEST_1_OFFSET
805 ) {
806 return Err(RdbFileError::InvalidOperation(format!(
807 "superblock names a manifest offset outside the zone: {}",
808 superblock.manifest_offset
809 )));
810 }
811
812 let mut bytes = vec![0u8; superblock.manifest_len as usize];
813 file.seek(SeekFrom::Start(superblock.manifest_offset))?;
814 file.read_exact(&mut bytes)?;
815 let checksum = trailer_checksum(&bytes);
816 if checksum != superblock.manifest_checksum {
817 return Err(RdbFileError::InvalidOperation(format!(
818 "embedded manifest checksum mismatch: stored {:#010x}, computed {:#010x}",
819 superblock.manifest_checksum, checksum
820 )));
821 }
822 decode_manifest(&bytes)
823}
824
825fn snapshot_reference_valid(
826 file: &mut File,
827 manifest: &EmbeddedRdbManifest,
828) -> RdbFileResult<bool> {
829 if manifest.snapshot_bytes == 0 {
830 return Ok(true);
831 }
832 let snapshot_end = manifest
833 .snapshot_offset
834 .checked_add(manifest.snapshot_bytes)
835 .ok_or_else(|| RdbFileError::InvalidOperation("embedded snapshot end overflow".into()))?;
836 if snapshot_end > file.metadata()?.len() {
837 return Ok(false);
838 }
839
840 let mut bytes = vec![0u8; manifest.snapshot_bytes as usize];
841 file.seek(SeekFrom::Start(manifest.snapshot_offset))?;
842 if file.read_exact(&mut bytes).is_err() {
843 return Ok(false);
844 }
845 if crc32(&bytes) != manifest.snapshot_checksum {
846 return Ok(false);
847 }
848 if bytes.len() >= SNAPSHOT_MAGIC.len() && &bytes[..SNAPSHOT_MAGIC.len()] != SNAPSHOT_MAGIC {
849 return Ok(false);
850 }
851 Ok(true)
852}
853
854fn published_snapshot_matches(
857 file: &mut File,
858 open: &EmbeddedRdbOpen,
859 snapshot: &[u8],
860) -> RdbFileResult<bool> {
861 if snapshot.is_empty() {
862 return Ok(true);
863 }
864 let mut published = vec![0u8; snapshot.len()];
865 file.seek(SeekFrom::Start(open.manifest.snapshot_offset))?;
866 file.read_exact(&mut published)?;
867 Ok(published == snapshot)
868}
869
870fn grow_wal_region_bytes(current: u64, min_required: u64) -> RdbFileResult<u64> {
871 let mut next = current.max(WAL_REGION_BYTES);
872 while next < min_required {
873 next = next.checked_mul(2).ok_or_else(|| {
874 RdbFileError::InvalidOperation("embedded wal region size overflow".into())
875 })?;
876 }
877 Ok(next)
878}
879
880fn embedded_path_lock(path: &Path) -> Arc<Mutex<EmbeddedPathState>> {
881 static LOCKS: OnceLock<Mutex<HashMap<PathBuf, Arc<Mutex<EmbeddedPathState>>>>> =
882 OnceLock::new();
883 let key = path.canonicalize().unwrap_or_else(|_| path.to_path_buf());
884 let mut locks = LOCKS
885 .get_or_init(|| Mutex::new(HashMap::new()))
886 .lock()
887 .unwrap_or_else(|poisoned| poisoned.into_inner());
888 locks
889 .entry(key)
890 .or_insert_with(|| Arc::new(Mutex::new(EmbeddedPathState::default())))
891 .clone()
892}
893
894fn next_snapshot_offset(
902 file: &mut File,
903 open: &EmbeddedRdbOpen,
904 wal_region_bytes: u64,
905 snapshot: &[u8],
906) -> RdbFileResult<u64> {
907 let base = open
908 .manifest
909 .wal_region_offset
910 .checked_add(wal_region_bytes)
911 .ok_or_else(|| {
912 RdbFileError::InvalidOperation("embedded snapshot offset overflow".into())
913 })?;
914 if open.manifest.snapshot_bytes == 0 && snapshot.is_empty() {
915 return Ok(base);
916 }
917
918 let other_copy = if open.selected_superblock.copy_index == 0 {
919 1
920 } else {
921 0
922 };
923 let mut live = vec![(open.manifest.snapshot_offset, open.manifest.snapshot_bytes)];
924 if let Some(other) = read_superblock_copy(file, other_copy)? {
925 live.push((other.snapshot_offset, other.snapshot_bytes));
926 }
927 let mut live_ranges = Vec::with_capacity(live.len());
928 for (offset, bytes) in live {
929 if bytes == 0 {
930 continue;
931 }
932 let end = offset.checked_add(bytes).ok_or_else(|| {
933 RdbFileError::InvalidOperation("embedded snapshot end overflow".into())
934 })?;
935 live_ranges.push((offset, end));
936 }
937
938 let len = snapshot.len() as u64;
939 let mut candidates = vec![align_up(base, SNAPSHOT_ALIGNMENT)?];
940 for &(_, end) in &live_ranges {
941 candidates.push(align_up(end.max(base), SNAPSHOT_ALIGNMENT)?);
942 }
943 candidates.sort_unstable();
944 for &candidate in &candidates {
945 let end = candidate.checked_add(len).ok_or_else(|| {
946 RdbFileError::InvalidOperation("embedded snapshot end overflow".into())
947 })?;
948 if live_ranges
949 .iter()
950 .all(|&(start, live_end)| end <= start || candidate >= live_end)
951 {
952 return Ok(candidate);
953 }
954 }
955 Ok(candidates[candidates.len() - 1])
957}
958
959fn align_up(value: u64, alignment: u64) -> RdbFileResult<u64> {
960 if alignment == 0 {
961 return Ok(value);
962 }
963 let remainder = value % alignment;
964 if remainder == 0 {
965 return Ok(value);
966 }
967 value
968 .checked_add(alignment - remainder)
969 .ok_or_else(|| RdbFileError::InvalidOperation("embedded alignment overflow".into()))
970}
971
972fn wal_region_start_checked(open: &EmbeddedRdbOpen) -> RdbFileResult<Option<u64>> {
976 let wal_start = open.manifest.wal_region_offset;
977 let wal_end = wal_start
978 .checked_add(open.manifest.wal_region_bytes)
979 .ok_or_else(|| RdbFileError::InvalidOperation("embedded wal end overflow".into()))?;
980 let append_boundary = open.manifest.wal_recovery_boundary;
981 if append_boundary < wal_start || append_boundary > wal_end {
982 return Err(RdbFileError::InvalidOperation(format!(
983 "invalid embedded wal boundary {append_boundary}"
984 )));
985 }
986 if open.manifest.wal_live_bytes > open.manifest.wal_region_bytes {
987 return Err(RdbFileError::InvalidOperation(format!(
988 "invalid embedded wal live bytes {}",
989 open.manifest.wal_live_bytes
990 )));
991 }
992 if open.manifest.wal_live_bytes == 0 {
993 return Ok(None);
994 }
995 Ok(Some(wal_start))
996}
997
998fn scan_wal(open: &EmbeddedRdbOpen) -> RdbFileResult<WalScan> {
1001 let wal_start = open.manifest.wal_region_offset;
1002 let wal_end = wal_start
1003 .checked_add(open.manifest.wal_region_bytes)
1004 .ok_or_else(|| RdbFileError::InvalidOperation("embedded wal end overflow".into()))?;
1005 let append_boundary = open.manifest.wal_recovery_boundary;
1006 if append_boundary < wal_start || append_boundary > wal_end {
1007 return Err(RdbFileError::InvalidOperation(format!(
1008 "invalid embedded wal boundary {append_boundary}"
1009 )));
1010 }
1011 if open.manifest.wal_live_bytes > open.manifest.wal_region_bytes {
1012 return Err(RdbFileError::InvalidOperation(format!(
1013 "invalid embedded wal live bytes {}",
1014 open.manifest.wal_live_bytes
1015 )));
1016 }
1017 if open.manifest.wal_live_bytes == 0 {
1018 return Ok(WalScan {
1019 next_sequence: 1,
1020 ..WalScan::default()
1021 });
1022 }
1023
1024 let mut file = File::open(&open.path)?;
1025 scan_wal_with_file(&mut file, open, wal_start)
1026}
1027
1028fn scan_wal_with_file(
1032 file: &mut File,
1033 open: &EmbeddedRdbOpen,
1034 wal_start: u64,
1035) -> RdbFileResult<WalScan> {
1036 let file_len = file.metadata()?.len();
1037 if file_len <= wal_start {
1038 return Ok(WalScan {
1039 next_sequence: 1,
1040 ..WalScan::default()
1041 });
1042 }
1043 let bytes = read_circular_wal_bytes(file, open, file_len)?;
1044 Ok(scan_wal_bytes(&bytes))
1045}
1046
1047fn read_circular_wal_bytes(
1048 file: &mut File,
1049 open: &EmbeddedRdbOpen,
1050 file_len: u64,
1051) -> RdbFileResult<Vec<u8>> {
1052 let live_start = wal_live_start_relative(open)?;
1053 let region_bytes = open.manifest.wal_region_bytes;
1054 let live_bytes = open.manifest.wal_live_bytes;
1055 let mut remaining = live_bytes.min(file_len.saturating_sub(open.manifest.wal_region_offset));
1056 let mut relative = live_start;
1057 let mut bytes = Vec::with_capacity(remaining as usize);
1058 while remaining > 0 {
1059 let contiguous = (region_bytes - relative).min(remaining);
1060 let offset = open.manifest.wal_region_offset + relative;
1061 file.seek(SeekFrom::Start(offset))?;
1062 let mut chunk = vec![0u8; contiguous as usize];
1063 file.read_exact(&mut chunk)?;
1064 bytes.extend_from_slice(&chunk);
1065 remaining -= contiguous;
1066 relative = 0;
1067 }
1068 Ok(bytes)
1069}
1070
1071fn write_circular_wal_bytes(
1072 file: &mut File,
1073 open: &EmbeddedRdbOpen,
1074 bytes: &[u8],
1075) -> RdbFileResult<()> {
1076 let mut written = 0usize;
1077 let mut relative = wal_append_relative(open)?;
1078 while written < bytes.len() {
1079 if relative == open.manifest.wal_region_bytes {
1080 relative = 0;
1081 }
1082 let contiguous = (open.manifest.wal_region_bytes - relative) as usize;
1083 let chunk_len = contiguous.min(bytes.len() - written);
1084 write_at(
1085 file,
1086 open.manifest.wal_region_offset + relative,
1087 &bytes[written..written + chunk_len],
1088 )?;
1089 written += chunk_len;
1090 relative += chunk_len as u64;
1091 }
1092 Ok(())
1093}
1094
1095fn wal_append_relative(open: &EmbeddedRdbOpen) -> RdbFileResult<u64> {
1096 open.manifest
1097 .wal_recovery_boundary
1098 .checked_sub(open.manifest.wal_region_offset)
1099 .ok_or_else(|| RdbFileError::InvalidOperation("embedded wal boundary underflow".into()))
1100}
1101
1102fn wal_live_start_relative(open: &EmbeddedRdbOpen) -> RdbFileResult<u64> {
1103 let append = wal_append_relative(open)?;
1104 let region = open.manifest.wal_region_bytes;
1105 if region == 0 {
1106 return Err(RdbFileError::InvalidOperation(
1107 "embedded wal region is empty".into(),
1108 ));
1109 }
1110 Ok((append + region - (open.manifest.wal_live_bytes % region)) % region)
1111}
1112
1113fn wal_boundary_after_live_bytes(open: &EmbeddedRdbOpen, live_bytes: u64) -> RdbFileResult<u64> {
1114 if live_bytes > open.manifest.wal_region_bytes {
1115 return Err(RdbFileError::InvalidOperation(format!(
1116 "embedded wal live bytes {live_bytes} exceed region size {}",
1117 open.manifest.wal_region_bytes
1118 )));
1119 }
1120 let start = if open.manifest.wal_live_bytes == 0 {
1121 wal_append_relative(open)?
1122 } else {
1123 wal_live_start_relative(open)?
1124 };
1125 let relative = (start + live_bytes) % open.manifest.wal_region_bytes;
1126 open.manifest
1127 .wal_region_offset
1128 .checked_add(relative)
1129 .ok_or_else(|| RdbFileError::InvalidOperation("embedded wal boundary overflow".into()))
1130}
1131
1132fn scan_wal_bytes(bytes: &[u8]) -> WalScan {
1133 let mut scan = WalScan {
1134 next_sequence: 1,
1135 ..WalScan::default()
1136 };
1137 let mut cursor = 0usize;
1138 while cursor < bytes.len() {
1139 let Some(frame) = decode_next_wal_frame(bytes, cursor, &scan) else {
1140 break;
1141 };
1142 scan.payloads.push(frame.payload);
1143 scan.next_sequence = scan.next_sequence.saturating_add(1);
1144 scan.previous_frame_crc = frame.frame_crc;
1145 cursor = frame.end;
1146 scan.valid_bytes = cursor as u64;
1147 }
1148 scan
1149}
1150
1151struct DecodedWalFrame {
1152 payload: Vec<u8>,
1153 frame_crc: u32,
1154 end: usize,
1155}
1156
1157fn decode_next_wal_frame(bytes: &[u8], start: usize, scan: &WalScan) -> Option<DecodedWalFrame> {
1158 let remaining = bytes.len().checked_sub(start)?;
1159 if remaining < WAL_FRAME_MAGIC.len() {
1160 return None;
1161 }
1162 if &bytes[start..start + WAL_FRAME_MAGIC.len()] != WAL_FRAME_MAGIC {
1163 return None;
1164 }
1165 if remaining < WAL_FRAME_MAGIC.len() + 2 {
1166 return None;
1167 }
1168 let version_offset = start + WAL_FRAME_MAGIC.len();
1169 let version = u16::from_le_bytes(bytes[version_offset..version_offset + 2].try_into().ok()?);
1170 if version == WAL_FRAME_VERSION {
1171 decode_v2_wal_frame(bytes, start, scan)
1172 } else {
1173 decode_legacy_wal_frame(bytes, start)
1174 }
1175}
1176
1177fn decode_v2_wal_frame(bytes: &[u8], start: usize, scan: &WalScan) -> Option<DecodedWalFrame> {
1178 if bytes.len().checked_sub(start)? < WAL_FRAME_HEADER_BYTES {
1179 return None;
1180 }
1181 let header_len_offset = start + 10;
1182 let header_len = u16::from_le_bytes(
1183 bytes[header_len_offset..header_len_offset + 2]
1184 .try_into()
1185 .ok()?,
1186 ) as usize;
1187 if header_len != WAL_FRAME_HEADER_BYTES {
1188 return None;
1189 }
1190 let sequence_offset = start + 12;
1191 let sequence = u64::from_le_bytes(
1192 bytes[sequence_offset..sequence_offset + 8]
1193 .try_into()
1194 .ok()?,
1195 );
1196 if sequence != scan.next_sequence {
1197 return None;
1198 }
1199 let payload_len_offset = start + 20;
1200 let payload_len = u32::from_le_bytes(
1201 bytes[payload_len_offset..payload_len_offset + 4]
1202 .try_into()
1203 .ok()?,
1204 ) as usize;
1205 let payload_crc_offset = start + 24;
1206 let payload_crc = u32::from_le_bytes(
1207 bytes[payload_crc_offset..payload_crc_offset + 4]
1208 .try_into()
1209 .ok()?,
1210 );
1211 let previous_frame_crc_offset = start + 28;
1212 let previous_frame_crc = u32::from_le_bytes(
1213 bytes[previous_frame_crc_offset..previous_frame_crc_offset + 4]
1214 .try_into()
1215 .ok()?,
1216 );
1217 if previous_frame_crc != scan.previous_frame_crc {
1218 return None;
1219 }
1220 let header_crc_offset = start + 32;
1221 let header_crc = u32::from_le_bytes(
1222 bytes[header_crc_offset..header_crc_offset + 4]
1223 .try_into()
1224 .ok()?,
1225 );
1226 if header_crc != crc32(&bytes[start..header_crc_offset]) {
1227 return None;
1228 }
1229 let payload_start = start.checked_add(header_len)?;
1230 let end = payload_start.checked_add(payload_len)?;
1231 if end > bytes.len() {
1232 return None;
1233 }
1234 let payload = bytes[payload_start..end].to_vec();
1235 if crc32(&payload) != payload_crc {
1236 return None;
1237 }
1238 Some(DecodedWalFrame {
1239 payload,
1240 frame_crc: crc32(&bytes[start..end]),
1241 end,
1242 })
1243}
1244
1245fn decode_legacy_wal_frame(bytes: &[u8], start: usize) -> Option<DecodedWalFrame> {
1246 if bytes.len().checked_sub(start)? < LEGACY_WAL_FRAME_HEADER_BYTES {
1247 return None;
1248 }
1249 let payload_len_offset = start + WAL_FRAME_MAGIC.len();
1250 let payload_len = u32::from_le_bytes(
1251 bytes[payload_len_offset..payload_len_offset + 4]
1252 .try_into()
1253 .ok()?,
1254 ) as usize;
1255 let payload_crc_offset = payload_len_offset + 4;
1256 let payload_crc = u32::from_le_bytes(
1257 bytes[payload_crc_offset..payload_crc_offset + 4]
1258 .try_into()
1259 .ok()?,
1260 );
1261 let payload_start = start.checked_add(LEGACY_WAL_FRAME_HEADER_BYTES)?;
1262 let end = payload_start.checked_add(payload_len)?;
1263 if end > bytes.len() {
1264 return None;
1265 }
1266 let payload = bytes[payload_start..end].to_vec();
1267 if crc32(&payload) != payload_crc {
1268 return None;
1269 }
1270 Some(DecodedWalFrame {
1271 payload,
1272 frame_crc: crc32(&bytes[start..end]),
1273 end,
1274 })
1275}
1276
1277fn encode_wal_frame(
1278 sequence: u64,
1279 previous_frame_crc: u32,
1280 payload: &[u8],
1281) -> RdbFileResult<(Vec<u8>, u32)> {
1282 let payload_len = u32::try_from(payload.len())
1283 .map_err(|_| RdbFileError::InvalidOperation("embedded wal payload too large".into()))?;
1284 let mut frame = Vec::with_capacity(WAL_FRAME_HEADER_BYTES + payload.len());
1285 frame.extend_from_slice(WAL_FRAME_MAGIC);
1286 frame.extend_from_slice(&WAL_FRAME_VERSION.to_le_bytes());
1287 frame.extend_from_slice(&(WAL_FRAME_HEADER_BYTES as u16).to_le_bytes());
1288 frame.extend_from_slice(&sequence.to_le_bytes());
1289 frame.extend_from_slice(&payload_len.to_le_bytes());
1290 frame.extend_from_slice(&crc32(payload).to_le_bytes());
1291 frame.extend_from_slice(&previous_frame_crc.to_le_bytes());
1292 let header_crc = crc32(&frame);
1293 frame.extend_from_slice(&header_crc.to_le_bytes());
1294 frame.extend_from_slice(payload);
1295 let frame_crc = crc32(&frame);
1296 Ok((frame, frame_crc))
1297}
1298
1299fn encode_superblock(superblock: EmbeddedRdbSuperblock) -> RdbFileResult<Vec<u8>> {
1300 let mut bytes = vec![0u8; EMBEDDED_RDB_SUPERBLOCK_SIZE as usize];
1301 let mut cursor = 0usize;
1302 put_bytes(&mut bytes, &mut cursor, SUPERBLOCK_MAGIC);
1303 put_u32(&mut bytes, &mut cursor, SUPERBLOCK_VERSION);
1304 put_u8(&mut bytes, &mut cursor, superblock.copy_index);
1305 put_u64(&mut bytes, &mut cursor, superblock.generation);
1306 put_u32(&mut bytes, &mut cursor, superblock.format_version);
1307 put_u64(&mut bytes, &mut cursor, superblock.manifest_offset);
1308 put_u64(&mut bytes, &mut cursor, superblock.manifest_len);
1309 put_u32(&mut bytes, &mut cursor, superblock.manifest_checksum);
1310 put_u64(&mut bytes, &mut cursor, superblock.wal_region_offset);
1311 put_u64(&mut bytes, &mut cursor, superblock.wal_region_bytes);
1312 put_u64(&mut bytes, &mut cursor, superblock.wal_recovery_boundary);
1313 put_u64(&mut bytes, &mut cursor, superblock.wal_live_bytes);
1314 put_u64(&mut bytes, &mut cursor, superblock.snapshot_offset);
1315 put_u64(&mut bytes, &mut cursor, superblock.snapshot_bytes);
1316 put_u32(&mut bytes, &mut cursor, superblock.snapshot_checksum);
1317
1318 let checksum_offset = bytes.len() - CHECKSUM_LEN;
1319 let checksum = crc32(&bytes[..checksum_offset]);
1320 bytes[checksum_offset..].copy_from_slice(&checksum.to_le_bytes());
1321 Ok(bytes)
1322}
1323
1324fn decode_superblock(copy_index: u8, bytes: &[u8]) -> RdbFileResult<EmbeddedRdbSuperblock> {
1325 if bytes.len() != EMBEDDED_RDB_SUPERBLOCK_SIZE as usize {
1326 return Err(RdbFileError::InvalidOperation(
1327 "invalid embedded superblock size".into(),
1328 ));
1329 }
1330 let checksum_offset = bytes.len() - CHECKSUM_LEN;
1331 let stored_checksum = u32::from_le_bytes(bytes[checksum_offset..].try_into().unwrap());
1332 let computed_checksum = crc32(&bytes[..checksum_offset]);
1333 if stored_checksum != computed_checksum {
1334 return Err(RdbFileError::InvalidOperation(
1335 "embedded superblock checksum mismatch".into(),
1336 ));
1337 }
1338
1339 let mut cursor = 0usize;
1340 if take_bytes(bytes, &mut cursor, SUPERBLOCK_MAGIC.len())? != SUPERBLOCK_MAGIC {
1341 return Err(RdbFileError::InvalidOperation(
1342 "invalid embedded superblock magic".into(),
1343 ));
1344 }
1345 let version = take_u32(bytes, &mut cursor)?;
1346 if version != SUPERBLOCK_VERSION && version != LEGACY_SUPERBLOCK_VERSION {
1347 return Err(RdbFileError::InvalidOperation(format!(
1348 "unsupported embedded superblock version {version}"
1349 )));
1350 }
1351 let stored_copy_index = take_u8(bytes, &mut cursor)?;
1352 if stored_copy_index != copy_index {
1353 return Err(RdbFileError::InvalidOperation(
1354 "embedded superblock copy index mismatch".into(),
1355 ));
1356 }
1357
1358 let generation = take_u64(bytes, &mut cursor)?;
1359 let format_version = take_u32(bytes, &mut cursor)?;
1360 let manifest_offset = take_u64(bytes, &mut cursor)?;
1361 let manifest_len = take_u64(bytes, &mut cursor)?;
1362 let manifest_checksum = take_u32(bytes, &mut cursor)?;
1363 let wal_region_offset = take_u64(bytes, &mut cursor)?;
1364 let wal_region_bytes = take_u64(bytes, &mut cursor)?;
1365 let wal_recovery_boundary = take_u64(bytes, &mut cursor)?;
1366 let wal_live_bytes = if version == SUPERBLOCK_VERSION {
1367 take_u64(bytes, &mut cursor)?
1368 } else {
1369 wal_recovery_boundary.saturating_sub(wal_region_offset)
1370 };
1371
1372 Ok(EmbeddedRdbSuperblock {
1373 copy_index: stored_copy_index,
1374 generation,
1375 format_version,
1376 manifest_offset,
1377 manifest_len,
1378 manifest_checksum,
1379 wal_region_offset,
1380 wal_region_bytes,
1381 wal_recovery_boundary,
1382 wal_live_bytes,
1383 snapshot_offset: take_u64(bytes, &mut cursor)?,
1384 snapshot_bytes: take_u64(bytes, &mut cursor)?,
1385 snapshot_checksum: take_u32(bytes, &mut cursor)?,
1386 checksum: stored_checksum,
1387 })
1388}
1389
1390fn encode_manifest(manifest: EmbeddedRdbManifest) -> Vec<u8> {
1391 let mut bytes = vec![0u8; 8 + 4 + 8 + 8 + 8 + 8 + 8 + 8 + 4 + 16 + CHECKSUM_LEN];
1392 let mut cursor = 0usize;
1393 put_bytes(&mut bytes, &mut cursor, MANIFEST_MAGIC);
1394 put_u32(&mut bytes, &mut cursor, manifest.version);
1395 put_u64(&mut bytes, &mut cursor, manifest.wal_region_offset);
1396 put_u64(&mut bytes, &mut cursor, manifest.wal_region_bytes);
1397 put_u64(&mut bytes, &mut cursor, manifest.wal_recovery_boundary);
1398 put_u64(&mut bytes, &mut cursor, manifest.wal_live_bytes);
1399 put_u64(&mut bytes, &mut cursor, manifest.snapshot_offset);
1400 put_u64(&mut bytes, &mut cursor, manifest.snapshot_bytes);
1401 put_u32(&mut bytes, &mut cursor, manifest.snapshot_checksum);
1402 put_u128(&mut bytes, &mut cursor, manifest.created_at_unix_ms);
1403
1404 let checksum_offset = bytes.len() - CHECKSUM_LEN;
1405 let checksum = crc32(&bytes[..checksum_offset]);
1406 bytes[checksum_offset..].copy_from_slice(&checksum.to_le_bytes());
1407 bytes
1408}
1409
1410fn decode_manifest(bytes: &[u8]) -> RdbFileResult<EmbeddedRdbManifest> {
1411 let checksum_offset = bytes
1412 .len()
1413 .checked_sub(CHECKSUM_LEN)
1414 .ok_or_else(|| RdbFileError::InvalidOperation("embedded manifest too short".into()))?;
1415 let stored_checksum = u32::from_le_bytes(bytes[checksum_offset..].try_into().unwrap());
1416 let computed_checksum = crc32(&bytes[..checksum_offset]);
1417 if stored_checksum != computed_checksum {
1418 return Err(RdbFileError::InvalidOperation(
1419 "embedded manifest checksum mismatch".into(),
1420 ));
1421 }
1422
1423 let mut cursor = 0usize;
1424 if take_bytes(bytes, &mut cursor, MANIFEST_MAGIC.len())? != MANIFEST_MAGIC {
1425 return Err(RdbFileError::InvalidOperation(
1426 "invalid embedded manifest magic".into(),
1427 ));
1428 }
1429 let version = take_u32(bytes, &mut cursor)?;
1430 if version != MANIFEST_VERSION && version != LEGACY_MANIFEST_VERSION {
1431 return Err(RdbFileError::InvalidOperation(format!(
1432 "unsupported embedded manifest version {version}"
1433 )));
1434 }
1435 let wal_region_offset = take_u64(bytes, &mut cursor)?;
1436 let wal_region_bytes = take_u64(bytes, &mut cursor)?;
1437 let wal_recovery_boundary = take_u64(bytes, &mut cursor)?;
1438 let wal_live_bytes = if version == MANIFEST_VERSION {
1439 take_u64(bytes, &mut cursor)?
1440 } else {
1441 wal_recovery_boundary.saturating_sub(wal_region_offset)
1442 };
1443 Ok(EmbeddedRdbManifest {
1444 version,
1445 wal_region_offset,
1446 wal_region_bytes,
1447 wal_recovery_boundary,
1448 wal_live_bytes,
1449 snapshot_offset: take_u64(bytes, &mut cursor)?,
1450 snapshot_bytes: take_u64(bytes, &mut cursor)?,
1451 snapshot_checksum: take_u32(bytes, &mut cursor)?,
1452 created_at_unix_ms: take_u128(bytes, &mut cursor)?,
1453 checksum: stored_checksum,
1454 })
1455}
1456
1457fn trailer_checksum(bytes: &[u8]) -> u32 {
1458 let checksum_offset = bytes.len() - CHECKSUM_LEN;
1459 u32::from_le_bytes(bytes[checksum_offset..].try_into().unwrap())
1460}
1461
1462fn superblock_offset(copy_index: u8) -> RdbFileResult<u64> {
1463 match copy_index {
1464 0 => Ok(EMBEDDED_RDB_SUPERBLOCK_0_OFFSET),
1465 1 => Ok(EMBEDDED_RDB_SUPERBLOCK_1_OFFSET),
1466 _ => Err(RdbFileError::InvalidOperation(format!(
1467 "invalid embedded superblock copy index {copy_index}"
1468 ))),
1469 }
1470}
1471
1472fn write_at(file: &mut File, offset: u64, bytes: &[u8]) -> RdbFileResult<()> {
1473 file.seek(SeekFrom::Start(offset))?;
1474 file.write_all(bytes)?;
1475 Ok(())
1476}
1477
1478fn crash_inject(point: &str) {
1479 if std::env::var(CRASH_INJECT_ENV).ok().as_deref() == Some(point) {
1480 std::process::exit(173);
1481 }
1482 if crate::buggify!(CRASH_INJECT_ENV, point) {
1483 std::process::exit(173);
1484 }
1485}
1486
1487fn now_unix_ms() -> u128 {
1488 SystemTime::now()
1489 .duration_since(UNIX_EPOCH)
1490 .map(|duration| duration.as_millis())
1491 .unwrap_or(0)
1492}
1493
1494fn put_bytes(target: &mut [u8], cursor: &mut usize, value: &[u8]) {
1495 target[*cursor..*cursor + value.len()].copy_from_slice(value);
1496 *cursor += value.len();
1497}
1498
1499fn put_u8(target: &mut [u8], cursor: &mut usize, value: u8) {
1500 target[*cursor] = value;
1501 *cursor += 1;
1502}
1503
1504fn put_u32(target: &mut [u8], cursor: &mut usize, value: u32) {
1505 put_bytes(target, cursor, &value.to_le_bytes());
1506}
1507
1508fn put_u64(target: &mut [u8], cursor: &mut usize, value: u64) {
1509 put_bytes(target, cursor, &value.to_le_bytes());
1510}
1511
1512fn put_u128(target: &mut [u8], cursor: &mut usize, value: u128) {
1513 put_bytes(target, cursor, &value.to_le_bytes());
1514}
1515
1516fn take_bytes<'a>(bytes: &'a [u8], cursor: &mut usize, len: usize) -> RdbFileResult<&'a [u8]> {
1517 let end = cursor.checked_add(len).ok_or_else(|| {
1518 RdbFileError::InvalidOperation("embedded artifact cursor overflow".into())
1519 })?;
1520 if end > bytes.len() {
1521 return Err(RdbFileError::InvalidOperation(
1522 "embedded artifact truncated".into(),
1523 ));
1524 }
1525 let value = &bytes[*cursor..end];
1526 *cursor = end;
1527 Ok(value)
1528}
1529
1530fn take_u8(bytes: &[u8], cursor: &mut usize) -> RdbFileResult<u8> {
1531 Ok(take_bytes(bytes, cursor, 1)?[0])
1532}
1533
1534fn take_u32(bytes: &[u8], cursor: &mut usize) -> RdbFileResult<u32> {
1535 Ok(u32::from_le_bytes(
1536 take_bytes(bytes, cursor, 4)?.try_into().unwrap(),
1537 ))
1538}
1539
1540fn take_u64(bytes: &[u8], cursor: &mut usize) -> RdbFileResult<u64> {
1541 Ok(u64::from_le_bytes(
1542 take_bytes(bytes, cursor, 8)?.try_into().unwrap(),
1543 ))
1544}
1545
1546fn take_u128(bytes: &[u8], cursor: &mut usize) -> RdbFileResult<u128> {
1547 Ok(u128::from_le_bytes(
1548 take_bytes(bytes, cursor, 16)?.try_into().unwrap(),
1549 ))
1550}