Skip to main content

reddb_file/
embedded.rs

1//! Embedded single-file `.rdb` artifact.
2//!
3//! This module models the promoted path where one durable database artifact
4//! carries its superblock pair, internal manifest, WAL reservation, and current
5//! store snapshot inside the `.rdb` file itself.
6
7use 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    /// A named zone of the `.rdb` cannot be trusted, so the store does not
25    /// open. ADR 0074 §2 requires the zone be named and §4 requires the
26    /// operator be pointed at `red salvage` rather than left with a panic.
27    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
69/// The manifest zone is a ping-pong pair, exactly like the superblock zone.
70///
71/// A single in-place manifest could not satisfy "never torn": overwriting it
72/// leaves a window in which the durable superblock still names the old
73/// checksum while the bytes are already the new ones. Publishing into the
74/// *inactive* slot and only then pointing a fresh superblock at it means every
75/// valid superblock always references an intact manifest — pre-update or
76/// post-update, never a mixture.
77pub 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    // Published only after validating this generation's snapshot, or after
155    // appending to an unchanged file with an already validated snapshot.
156    stamp: FileMutationStamp,
157    superblock: EmbeddedRdbSuperblock,
158    next_sequence: u64,
159    previous_frame_crc: u32,
160    valid_bytes: u64,
161}
162
163// The superblock detects cooperating writers/checkpoints. File identity and
164// mutation times also invalidate replacement or in-place edits. On platforms
165// without this identity vocabulary, append keeps the validated full scan.
166#[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
195/// Process-lifetime writer fence for an embedded `.rdb`.
196///
197/// The per-operation lock on the data file only serializes individual appends
198/// and checkpoints. A writer process keeps the whole store in memory and
199/// checkpoints that image on close, so two processes writing the same file
200/// would each publish their own image and the last one to close would
201/// silently discard the other's commits. Holding this lock for the writer's
202/// lifetime makes the second opener fail instead.
203///
204/// It lives in a sibling lock file, not on the data file: the per-operation
205/// lock takes a fresh handle, which under POSIX `flock` would contend with a
206/// lock held through another handle of the same process, and under Windows
207/// `LockFileEx` would be denied outright (see `open_inner_with_file`).
208pub 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            // The previous owner unlinks the lock file on release. If that
241            // happened between our open and our lock, we hold an orphaned
242            // inode no later opener can see, so start over on the live path.
243            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        // Unlink while still holding the lock so a closed store is back to
253        // one operator-visible artifact; `acquire` re-checks the inode to
254        // stay correct across this. Without an inode identity check the
255        // file is left in place instead.
256        #[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    /// Read the store's roots through an already-open handle.
373    ///
374    /// The writer path holds an exclusive lock on the data file and must do
375    /// all of its I/O through that one handle: `fs2` maps onto `LockFileEx`
376    /// on Windows, whose byte-range locks are *mandatory* and scoped to the
377    /// handle that took them, so a second `File::open` of the same path is
378    /// denied with `ERROR_LOCK_VIOLATION` (os error 33). That is what made
379    /// every embedded write fail on Windows while reads kept working
380    /// (issues #2265, #2266). POSIX `flock` is advisory and scoped to the
381    /// open file description, which is why it never reproduced on Linux.
382    ///
383    /// Every helper here seeks explicitly, so sharing one cursor across the
384    /// reads and writes of a single operation is safe.
385    fn open_inner_with_file(
386        file: &mut File,
387        path: &Path,
388        validate_snapshot_refs: bool,
389    ) -> RdbFileResult<EmbeddedRdbOpen> {
390        // An I/O error propagates as an I/O error; only bytes that were read
391        // and failed to decode count towards "the zone is unrecoverable".
392        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            // The manifest a valid superblock names is always intact: it was
410            // fsynced into an inactive slot before this generation existed. A
411            // checksum failure here is therefore bit rot, not a torn update,
412            // and ADR 0074 §2 says it fails the open by name — never falls
413            // back to a stale root that would resurrect superseded state.
414            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        // One handle for the whole operation: it carries the exclusive lock,
464        // and on Windows that lock would deny any *other* handle this module
465        // opened to the same path (see `open_inner_with_file`).
466        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        // Nothing to publish when the WAL is empty, its region need not grow,
482        // and the durable image already holds exactly these bytes. A close
483        // used to checkpoint the same image two or three times, each a full
484        // write plus fsyncs.
485        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        // The image may land in a hole below a live snapshot, so only ever
510        // extend here: shrinking before publish would cut a live image.
511        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        // Publish the manifest into the slot the live superblock does NOT
523        // reference, and make it durable before any superblock names it. Until
524        // the superblock write below lands, the durable state still roots
525        // through the old manifest slot, which these bytes never touched.
526        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        // Published: the two superblock copies now name this image and the
557        // previous one. Anything past both (and past the WAL region) is
558        // unreachable, so give it back instead of growing on every checkpoint.
559        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        // Release before re-opening: `Self::open` takes its own handle, and
567        // on Windows it could not open one while this lock is held.
568        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        // The superblock is the WAL authority: appends update it without
587        // rewriting the manifest zone, so a stale manifest wal_live_bytes
588        // would make the scan see an empty region and drop live records.
589        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        // Take before any fallible I/O: a failed append must leave no cached
645        // state that could describe a partially published mutation.
646        let cached_tail = path_guard.tail.take();
647        // Single locked handle for the whole append — see
648        // `write_snapshot_with_wal_capacity` and `open_inner_with_file`.
649        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        // Recovery may salvage a valid prefix, but appending after a corrupt
674        // published frame would acknowledge an unreachable new commit.
675        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        // A warm append writes only WAL bytes and the alternate superblock:
729        // it does not change the snapshot. The matching identity/mutation
730        // stamp above proves no intervening file edit since its validation.
731        // Cache misses (including unsupported platforms) retain the full
732        // checksum pass. Normal open/read/scrub always verify snapshot bytes;
733        // this reuse is not a substitute for detecting bit rot on read.
734        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
772/// The manifest slot the given superblock does not reference.
773///
774/// With two superblock copies and two manifest slots this is always the slot
775/// belonging to the *stale* superblock copy — the same copy the update is about
776/// to overwrite. That coincidence is what makes the pair safe: a crash can only
777/// ever clobber the manifest of a superblock that was already being replaced.
778fn 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    // A superblock that points outside the manifest zone is a misdirected or
801    // forged write; never chase the pointer.
802    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
854/// Byte-for-byte comparison with the published image; the checksum match
855/// that gates this call is not proof of equality on its own.
856fn 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
894/// Lowest aligned offset past the WAL region where `snapshot` overlaps no
895/// image either superblock copy still names.
896///
897/// Both images stay intact until the new superblock is durable: the selected
898/// one is the published state, and the other copy's is the fallback when the
899/// newer superblock or its image fails validation. Any other bytes past the
900/// WAL region are unreachable and may be reused.
901fn 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    // The largest candidate starts past every live image, so it always fits.
956    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
972/// Validate the WAL geometry recorded in the manifest and return the region
973/// start. Shared by [`scan_wal`] and the writer path so both reject the same
974/// malformed manifests.
975fn 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
998/// Scan the WAL, opening the file for reading. Callers already holding a
999/// handle (the writer path) use [`scan_wal_with_file`] instead.
1000fn 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
1028/// Scan the WAL through an already-open handle — see
1029/// [`EmbeddedRdbArtifact::open_inner_with_file`] for why the writer path
1030/// cannot open a second one.
1031fn 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}