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 scrub to classify the fault and red salvage to extract every entity the \
43                 damage did not touch; red salvage never writes into the damaged file \
44                 (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 = 1;
87const MANIFEST_VERSION: u32 = 1;
88const CHECKSUM_LEN: usize = 4;
89const MANIFEST_REGION_BYTES: u64 = EMBEDDED_RDB_MANIFEST_SLOT_SIZE;
90const WAL_REGION_BYTES: u64 = 64 * 1024;
91const SNAPSHOT_ALIGNMENT: u64 = 4096;
92const SNAPSHOT_MAGIC: &[u8; 4] = b"RDST";
93const WAL_FRAME_MAGIC: &[u8; 8] = b"RDBEWAL1";
94const WAL_FRAME_VERSION: u16 = 2;
95const WAL_FRAME_HEADER_BYTES: usize = 8 + 2 + 2 + 8 + 4 + 4 + 4 + 4;
96const LEGACY_WAL_FRAME_HEADER_BYTES: usize = 8 + 4 + 4;
97const CRASH_INJECT_ENV: &str = "REDDB_EMBEDDED_RDB_CRASH_AT";
98
99#[derive(Debug, Clone, Copy, PartialEq, Eq)]
100pub struct EmbeddedRdbManifest {
101    pub version: u32,
102    pub wal_region_offset: u64,
103    pub wal_region_bytes: u64,
104    pub wal_recovery_boundary: u64,
105    pub snapshot_offset: u64,
106    pub snapshot_bytes: u64,
107    pub snapshot_checksum: u32,
108    pub created_at_unix_ms: u128,
109    pub checksum: u32,
110}
111
112#[derive(Debug, Clone, Copy, PartialEq, Eq)]
113pub struct EmbeddedRdbSuperblock {
114    pub copy_index: u8,
115    pub generation: u64,
116    pub format_version: u32,
117    pub manifest_offset: u64,
118    pub manifest_len: u64,
119    pub manifest_checksum: u32,
120    pub wal_region_offset: u64,
121    pub wal_region_bytes: u64,
122    pub wal_recovery_boundary: u64,
123    pub snapshot_offset: u64,
124    pub snapshot_bytes: u64,
125    pub snapshot_checksum: u32,
126    pub checksum: u32,
127}
128
129#[derive(Debug, Clone, PartialEq, Eq)]
130pub struct EmbeddedRdbOpen {
131    pub path: PathBuf,
132    pub selected_superblock: EmbeddedRdbSuperblock,
133    pub manifest: EmbeddedRdbManifest,
134}
135
136#[derive(Debug, Default)]
137struct WalScan {
138    payloads: Vec<Vec<u8>>,
139    next_sequence: u64,
140    previous_frame_crc: u32,
141    valid_bytes: u64,
142}
143
144pub struct EmbeddedRdbArtifact;
145
146impl EmbeddedRdbArtifact {
147    pub fn create(path: impl AsRef<Path>) -> RdbFileResult<EmbeddedRdbOpen> {
148        Self::create_with_snapshot(path, &[])
149    }
150
151    pub fn create_with_snapshot(
152        path: impl AsRef<Path>,
153        snapshot: &[u8],
154    ) -> RdbFileResult<EmbeddedRdbOpen> {
155        let path = path.as_ref();
156        if let Some(parent) = path.parent() {
157            if !parent.as_os_str().is_empty() {
158                fs::create_dir_all(parent)?;
159            }
160        }
161
162        let created_at_unix_ms = now_unix_ms();
163        let wal_region_offset = EMBEDDED_RDB_MANIFEST_ZONE_END;
164        let snapshot_offset = wal_region_offset + WAL_REGION_BYTES;
165        let manifest = EmbeddedRdbManifest {
166            version: MANIFEST_VERSION,
167            wal_region_offset,
168            wal_region_bytes: WAL_REGION_BYTES,
169            wal_recovery_boundary: wal_region_offset,
170            snapshot_offset,
171            snapshot_bytes: snapshot.len() as u64,
172            snapshot_checksum: crc32(snapshot),
173            created_at_unix_ms,
174            checksum: 0,
175        };
176        let manifest_bytes = encode_manifest(manifest);
177        let manifest_checksum = trailer_checksum(&manifest_bytes);
178
179        let mut file = OpenOptions::new()
180            .create(true)
181            .truncate(true)
182            .read(true)
183            .write(true)
184            .open(path)?;
185        file.set_len(snapshot_offset + snapshot.len() as u64)?;
186        write_at(&mut file, EMBEDDED_RDB_MANIFEST_0_OFFSET, &manifest_bytes)?;
187        if !snapshot.is_empty() {
188            write_at(&mut file, snapshot_offset, snapshot)?;
189        }
190
191        let base = EmbeddedRdbSuperblock {
192            copy_index: 0,
193            generation: 1,
194            format_version: DEFAULT_FORMAT_VERSION,
195            manifest_offset: EMBEDDED_RDB_MANIFEST_0_OFFSET,
196            manifest_len: manifest_bytes.len() as u64,
197            manifest_checksum,
198            wal_region_offset,
199            wal_region_bytes: WAL_REGION_BYTES,
200            wal_recovery_boundary: wal_region_offset,
201            snapshot_offset,
202            snapshot_bytes: snapshot.len() as u64,
203            snapshot_checksum: crc32(snapshot),
204            checksum: 0,
205        };
206        Self::write_superblock_copy(&mut file, &base)?;
207        Self::write_superblock_copy(
208            &mut file,
209            &EmbeddedRdbSuperblock {
210                copy_index: 1,
211                generation: 2,
212                ..base
213            },
214        )?;
215        file.sync_all()?;
216
217        Self::open(path)
218    }
219
220    pub fn open(path: impl AsRef<Path>) -> RdbFileResult<EmbeddedRdbOpen> {
221        Self::open_inner(path, true)
222    }
223
224    fn open_for_wal_append(path: impl AsRef<Path>) -> RdbFileResult<EmbeddedRdbOpen> {
225        Self::open_inner(path, false)
226    }
227
228    fn open_inner(
229        path: impl AsRef<Path>,
230        validate_snapshot_refs: bool,
231    ) -> RdbFileResult<EmbeddedRdbOpen> {
232        let path = path.as_ref();
233        let mut file = File::open(path)?;
234        let mut superblocks: Vec<EmbeddedRdbSuperblock> = [
235            read_superblock_copy(&mut file, 0),
236            read_superblock_copy(&mut file, 1),
237        ]
238        .into_iter()
239        .flatten()
240        .collect();
241        superblocks.sort_by_key(|superblock| std::cmp::Reverse(superblock.generation));
242
243        if superblocks.is_empty() {
244            return Err(RdbFileError::ZoneUnrecoverable {
245                zone: "superblock",
246                path: path.to_path_buf(),
247            });
248        }
249
250        for selected_superblock in superblocks {
251            // The manifest a valid superblock names is always intact: it was
252            // fsynced into an inactive slot before this generation existed. A
253            // checksum failure here is therefore bit rot, not a torn update,
254            // and ADR 0074 §2 says it fails the open by name — never falls
255            // back to a stale root that would resurrect superseded state.
256            let mut manifest = read_manifest(&mut file, selected_superblock).map_err(|_| {
257                RdbFileError::ZoneUnrecoverable {
258                    zone: "manifest",
259                    path: path.to_path_buf(),
260                }
261            })?;
262            manifest.wal_recovery_boundary = selected_superblock.wal_recovery_boundary;
263            if validate_snapshot_refs && !snapshot_reference_valid(&mut file, &manifest)? {
264                continue;
265            }
266            return Ok(EmbeddedRdbOpen {
267                path: path.to_path_buf(),
268                selected_superblock,
269                manifest,
270            });
271        }
272
273        Err(RdbFileError::ZoneUnrecoverable {
274            zone: "snapshot",
275            path: path.to_path_buf(),
276        })
277    }
278
279    pub fn wal_payloads_encoded_len(payloads: &[Vec<u8>]) -> RdbFileResult<u64> {
280        let mut len = 0u64;
281        for payload in payloads {
282            let payload_len = u32::try_from(payload.len()).map_err(|_| {
283                RdbFileError::InvalidOperation("embedded wal payload too large".into())
284            })?;
285            let frame_len = WAL_FRAME_HEADER_BYTES as u64 + payload_len as u64;
286            len = len.checked_add(frame_len).ok_or_else(|| {
287                RdbFileError::InvalidOperation("embedded wal encoded length overflow".into())
288            })?;
289        }
290        Ok(len)
291    }
292
293    pub fn write_snapshot_with_wal_capacity(
294        path: impl AsRef<Path>,
295        snapshot: &[u8],
296        min_wal_bytes: u64,
297    ) -> RdbFileResult<EmbeddedRdbOpen> {
298        let path = path.as_ref();
299        let path_lock = embedded_path_lock(path);
300        let _path_guard = path_lock
301            .lock()
302            .unwrap_or_else(|poisoned| poisoned.into_inner());
303        let lock_file = OpenOptions::new().read(true).write(true).open(path)?;
304        lock_file.lock_exclusive()?;
305
306        let open = Self::open(path)?;
307        let wal_region_bytes =
308            grow_wal_region_bytes(open.manifest.wal_region_bytes, min_wal_bytes)?;
309        let snapshot_offset = next_snapshot_offset(path, &open, wal_region_bytes, snapshot)?;
310        let snapshot_checksum = crc32(snapshot);
311        let manifest = EmbeddedRdbManifest {
312            wal_region_bytes,
313            wal_recovery_boundary: open.manifest.wal_region_offset,
314            snapshot_offset,
315            snapshot_bytes: snapshot.len() as u64,
316            snapshot_checksum,
317            checksum: 0,
318            ..open.manifest
319        };
320        let manifest_bytes = encode_manifest(manifest);
321        let manifest_checksum = trailer_checksum(&manifest_bytes);
322
323        let mut file = OpenOptions::new().read(true).write(true).open(path)?;
324        file.set_len(snapshot_offset + snapshot.len() as u64)?;
325        if !snapshot.is_empty() {
326            write_at(&mut file, snapshot_offset, snapshot)?;
327        }
328        crash_inject("snapshot_after_image_write");
329        file.sync_data()?;
330        crash_inject("snapshot_after_image_sync");
331
332        // Publish the manifest into the slot the live superblock does NOT
333        // reference, and make it durable before any superblock names it. Until
334        // the superblock write below lands, the durable state still roots
335        // through the old manifest slot, which these bytes never touched.
336        let next_manifest_offset = inactive_manifest_offset(open.selected_superblock)?;
337        write_at(&mut file, next_manifest_offset, &manifest_bytes)?;
338        crash_inject("snapshot_after_manifest_write");
339        file.sync_data()?;
340        crash_inject("snapshot_after_manifest_sync");
341
342        let next_copy_index = if open.selected_superblock.copy_index == 0 {
343            1
344        } else {
345            0
346        };
347        let next_superblock = EmbeddedRdbSuperblock {
348            copy_index: next_copy_index,
349            generation: open.selected_superblock.generation.saturating_add(1),
350            manifest_offset: next_manifest_offset,
351            manifest_len: manifest_bytes.len() as u64,
352            manifest_checksum,
353            wal_region_bytes,
354            wal_recovery_boundary: open.manifest.wal_region_offset,
355            snapshot_offset,
356            snapshot_bytes: snapshot.len() as u64,
357            snapshot_checksum,
358            checksum: 0,
359            ..open.selected_superblock
360        };
361        Self::write_superblock_copy(&mut file, &next_superblock)?;
362        crash_inject("snapshot_after_superblock_write");
363        file.sync_all()?;
364        lock_file.unlock()?;
365        Self::open(path)
366    }
367
368    pub fn open_strict_manifest(path: impl AsRef<Path>) -> RdbFileResult<EmbeddedRdbOpen> {
369        let path = path.as_ref();
370        let mut file = File::open(path)?;
371        let selected_superblock = [
372            read_superblock_copy(&mut file, 0),
373            read_superblock_copy(&mut file, 1),
374        ]
375        .into_iter()
376        .flatten()
377        .max_by_key(|superblock| superblock.generation)
378        .ok_or_else(|| RdbFileError::InvalidOperation("no valid embedded superblock".into()))?;
379
380        let mut manifest = read_manifest(&mut file, selected_superblock)?;
381        manifest.wal_recovery_boundary = selected_superblock.wal_recovery_boundary;
382        Ok(EmbeddedRdbOpen {
383            path: path.to_path_buf(),
384            selected_superblock,
385            manifest,
386        })
387    }
388
389    pub fn read_snapshot(open: &EmbeddedRdbOpen) -> RdbFileResult<Option<Vec<u8>>> {
390        if open.manifest.snapshot_bytes == 0 {
391            return Ok(None);
392        }
393        let mut file = File::open(&open.path)?;
394        let mut bytes = vec![0u8; open.manifest.snapshot_bytes as usize];
395        file.seek(SeekFrom::Start(open.manifest.snapshot_offset))?;
396        file.read_exact(&mut bytes)?;
397        let checksum = crc32(&bytes);
398        if checksum != open.manifest.snapshot_checksum {
399            return Err(RdbFileError::InvalidOperation(format!(
400                "embedded snapshot checksum mismatch: stored {:#010x}, computed {:#010x}",
401                open.manifest.snapshot_checksum, checksum
402            )));
403        }
404        if bytes.len() >= SNAPSHOT_MAGIC.len() && &bytes[..SNAPSHOT_MAGIC.len()] != SNAPSHOT_MAGIC {
405            return Err(RdbFileError::InvalidOperation(
406                "invalid embedded snapshot magic".into(),
407            ));
408        }
409        Ok(Some(bytes))
410    }
411
412    pub fn write_snapshot(
413        path: impl AsRef<Path>,
414        snapshot: &[u8],
415    ) -> RdbFileResult<EmbeddedRdbOpen> {
416        Self::write_snapshot_with_wal_capacity(path, snapshot, 0)
417    }
418
419    pub fn read_wal_payloads(open: &EmbeddedRdbOpen) -> RdbFileResult<Vec<Vec<u8>>> {
420        Ok(scan_wal(open)?.payloads)
421    }
422
423    pub fn append_wal_payloads(
424        path: impl AsRef<Path>,
425        payloads: &[Vec<u8>],
426    ) -> RdbFileResult<EmbeddedRdbOpen> {
427        let path = path.as_ref();
428        if payloads.is_empty() {
429            return Self::open(path);
430        }
431
432        let path_lock = embedded_path_lock(path);
433        let _path_guard = path_lock
434            .lock()
435            .unwrap_or_else(|poisoned| poisoned.into_inner());
436        let lock_file = OpenOptions::new().read(true).write(true).open(path)?;
437        lock_file.lock_exclusive()?;
438
439        let open = Self::open_for_wal_append(path)?;
440        let wal_scan = scan_wal(&open)?;
441        let mut sequence = wal_scan.next_sequence;
442        let mut previous_frame_crc = wal_scan.previous_frame_crc;
443        let mut encoded = Vec::new();
444        for payload in payloads {
445            let (frame, frame_crc) = encode_wal_frame(sequence, previous_frame_crc, payload)?;
446            encoded.extend_from_slice(&frame);
447            previous_frame_crc = frame_crc;
448            sequence = sequence.saturating_add(1);
449        }
450
451        let wal_start = open.manifest.wal_region_offset;
452        let wal_end = wal_start.checked_add(wal_scan.valid_bytes).ok_or_else(|| {
453            RdbFileError::InvalidOperation("embedded wal boundary overflow".into())
454        })?;
455        let max_end = open
456            .manifest
457            .wal_region_offset
458            .saturating_add(open.manifest.wal_region_bytes);
459        let next_boundary = wal_end.checked_add(encoded.len() as u64).ok_or_else(|| {
460            RdbFileError::InvalidOperation("embedded wal boundary overflow".into())
461        })?;
462        if wal_end < wal_start || next_boundary > max_end {
463            return Err(RdbFileError::InvalidOperation(
464                "embedded wal region full".into(),
465            ));
466        }
467
468        let mut file = OpenOptions::new().read(true).write(true).open(path)?;
469        write_at(&mut file, wal_end, &encoded)?;
470        crash_inject("wal_after_frame_write");
471        file.sync_data()?;
472        crash_inject("wal_after_frame_sync");
473
474        let next_copy_index = if open.selected_superblock.copy_index == 0 {
475            1
476        } else {
477            0
478        };
479        let next_superblock = EmbeddedRdbSuperblock {
480            copy_index: next_copy_index,
481            generation: open.selected_superblock.generation.saturating_add(1),
482            wal_recovery_boundary: next_boundary,
483            checksum: 0,
484            ..open.selected_superblock
485        };
486        Self::write_superblock_copy(&mut file, &next_superblock)?;
487        crash_inject("wal_after_superblock_write");
488        file.sync_all()?;
489        lock_file.unlock()?;
490        Self::open(path)
491    }
492
493    pub fn write_superblock_copy(
494        file: &mut File,
495        superblock: &EmbeddedRdbSuperblock,
496    ) -> RdbFileResult<()> {
497        let offset = superblock_offset(superblock.copy_index)?;
498        write_at(file, offset, &encode_superblock(*superblock)?)?;
499        Ok(())
500    }
501}
502
503fn read_superblock_copy(file: &mut File, copy_index: u8) -> Option<EmbeddedRdbSuperblock> {
504    let offset = superblock_offset(copy_index).ok()?;
505    let mut bytes = vec![0u8; EMBEDDED_RDB_SUPERBLOCK_SIZE as usize];
506    file.seek(SeekFrom::Start(offset)).ok()?;
507    file.read_exact(&mut bytes).ok()?;
508    decode_superblock(copy_index, &bytes).ok()
509}
510
511/// The manifest slot the given superblock does not reference.
512///
513/// With two superblock copies and two manifest slots this is always the slot
514/// belonging to the *stale* superblock copy — the same copy the update is about
515/// to overwrite. That coincidence is what makes the pair safe: a crash can only
516/// ever clobber the manifest of a superblock that was already being replaced.
517fn inactive_manifest_offset(superblock: EmbeddedRdbSuperblock) -> RdbFileResult<u64> {
518    match superblock.manifest_offset {
519        EMBEDDED_RDB_MANIFEST_0_OFFSET => Ok(EMBEDDED_RDB_MANIFEST_1_OFFSET),
520        EMBEDDED_RDB_MANIFEST_1_OFFSET => Ok(EMBEDDED_RDB_MANIFEST_0_OFFSET),
521        other => Err(RdbFileError::InvalidOperation(format!(
522            "superblock names a manifest offset outside the zone: {other}"
523        ))),
524    }
525}
526
527fn read_manifest(
528    file: &mut File,
529    superblock: EmbeddedRdbSuperblock,
530) -> RdbFileResult<EmbeddedRdbManifest> {
531    if superblock.manifest_len < CHECKSUM_LEN as u64
532        || superblock.manifest_len > MANIFEST_REGION_BYTES
533    {
534        return Err(RdbFileError::InvalidOperation(format!(
535            "invalid embedded manifest length {}",
536            superblock.manifest_len
537        )));
538    }
539    // A superblock that points outside the manifest zone is a misdirected or
540    // forged write; never chase the pointer.
541    if !matches!(
542        superblock.manifest_offset,
543        EMBEDDED_RDB_MANIFEST_0_OFFSET | EMBEDDED_RDB_MANIFEST_1_OFFSET
544    ) {
545        return Err(RdbFileError::InvalidOperation(format!(
546            "superblock names a manifest offset outside the zone: {}",
547            superblock.manifest_offset
548        )));
549    }
550
551    let mut bytes = vec![0u8; superblock.manifest_len as usize];
552    file.seek(SeekFrom::Start(superblock.manifest_offset))?;
553    file.read_exact(&mut bytes)?;
554    let checksum = trailer_checksum(&bytes);
555    if checksum != superblock.manifest_checksum {
556        return Err(RdbFileError::InvalidOperation(format!(
557            "embedded manifest checksum mismatch: stored {:#010x}, computed {:#010x}",
558            superblock.manifest_checksum, checksum
559        )));
560    }
561    decode_manifest(&bytes)
562}
563
564fn snapshot_reference_valid(
565    file: &mut File,
566    manifest: &EmbeddedRdbManifest,
567) -> RdbFileResult<bool> {
568    if manifest.snapshot_bytes == 0 {
569        return Ok(true);
570    }
571    let snapshot_end = manifest
572        .snapshot_offset
573        .checked_add(manifest.snapshot_bytes)
574        .ok_or_else(|| RdbFileError::InvalidOperation("embedded snapshot end overflow".into()))?;
575    if snapshot_end > file.metadata()?.len() {
576        return Ok(false);
577    }
578
579    let mut bytes = vec![0u8; manifest.snapshot_bytes as usize];
580    file.seek(SeekFrom::Start(manifest.snapshot_offset))?;
581    if file.read_exact(&mut bytes).is_err() {
582        return Ok(false);
583    }
584    if crc32(&bytes) != manifest.snapshot_checksum {
585        return Ok(false);
586    }
587    if bytes.len() >= SNAPSHOT_MAGIC.len() && &bytes[..SNAPSHOT_MAGIC.len()] != SNAPSHOT_MAGIC {
588        return Ok(false);
589    }
590    Ok(true)
591}
592
593fn grow_wal_region_bytes(current: u64, min_required: u64) -> RdbFileResult<u64> {
594    let mut next = current.max(WAL_REGION_BYTES);
595    while next < min_required {
596        next = next.checked_mul(2).ok_or_else(|| {
597            RdbFileError::InvalidOperation("embedded wal region size overflow".into())
598        })?;
599    }
600    Ok(next)
601}
602
603fn embedded_path_lock(path: &Path) -> Arc<Mutex<()>> {
604    static LOCKS: OnceLock<Mutex<HashMap<PathBuf, Arc<Mutex<()>>>>> = OnceLock::new();
605    let key = path.canonicalize().unwrap_or_else(|_| path.to_path_buf());
606    let mut locks = LOCKS
607        .get_or_init(|| Mutex::new(HashMap::new()))
608        .lock()
609        .unwrap_or_else(|poisoned| poisoned.into_inner());
610    locks
611        .entry(key)
612        .or_insert_with(|| Arc::new(Mutex::new(())))
613        .clone()
614}
615
616fn next_snapshot_offset(
617    path: &Path,
618    open: &EmbeddedRdbOpen,
619    wal_region_bytes: u64,
620    snapshot: &[u8],
621) -> RdbFileResult<u64> {
622    let base = open
623        .manifest
624        .wal_region_offset
625        .checked_add(wal_region_bytes)
626        .ok_or_else(|| {
627            RdbFileError::InvalidOperation("embedded snapshot offset overflow".into())
628        })?;
629    if open.manifest.snapshot_bytes == 0 && snapshot.is_empty() {
630        return Ok(base);
631    }
632
633    let file_len = std::fs::metadata(path).map(|metadata| metadata.len())?;
634    let active_snapshot_end = open
635        .manifest
636        .snapshot_offset
637        .checked_add(open.manifest.snapshot_bytes)
638        .ok_or_else(|| RdbFileError::InvalidOperation("embedded snapshot end overflow".into()))?;
639    align_up(
640        file_len.max(active_snapshot_end).max(base),
641        SNAPSHOT_ALIGNMENT,
642    )
643}
644
645fn align_up(value: u64, alignment: u64) -> RdbFileResult<u64> {
646    if alignment == 0 {
647        return Ok(value);
648    }
649    let remainder = value % alignment;
650    if remainder == 0 {
651        return Ok(value);
652    }
653    value
654        .checked_add(alignment - remainder)
655        .ok_or_else(|| RdbFileError::InvalidOperation("embedded alignment overflow".into()))
656}
657
658fn scan_wal(open: &EmbeddedRdbOpen) -> RdbFileResult<WalScan> {
659    let wal_start = open.manifest.wal_region_offset;
660    let wal_end = open.manifest.wal_recovery_boundary;
661    let max_end = open
662        .manifest
663        .wal_region_offset
664        .saturating_add(open.manifest.wal_region_bytes);
665    if wal_end < wal_start || wal_end > max_end {
666        return Err(RdbFileError::InvalidOperation(format!(
667            "invalid embedded wal boundary {wal_end}"
668        )));
669    }
670    if wal_end == wal_start {
671        return Ok(WalScan {
672            next_sequence: 1,
673            ..WalScan::default()
674        });
675    }
676
677    let mut file = File::open(&open.path)?;
678    let file_len = file.metadata()?.len();
679    if file_len <= wal_start {
680        return Ok(WalScan {
681            next_sequence: 1,
682            ..WalScan::default()
683        });
684    }
685    let read_end = wal_end.min(file_len);
686    file.seek(SeekFrom::Start(wal_start))?;
687    let mut bytes = vec![0u8; (read_end - wal_start) as usize];
688    file.read_exact(&mut bytes)?;
689    Ok(scan_wal_bytes(&bytes))
690}
691
692fn scan_wal_bytes(bytes: &[u8]) -> WalScan {
693    let mut scan = WalScan {
694        next_sequence: 1,
695        ..WalScan::default()
696    };
697    let mut cursor = 0usize;
698    while cursor < bytes.len() {
699        let Some(frame) = decode_next_wal_frame(bytes, cursor, &scan) else {
700            break;
701        };
702        scan.payloads.push(frame.payload);
703        scan.next_sequence = scan.next_sequence.saturating_add(1);
704        scan.previous_frame_crc = frame.frame_crc;
705        cursor = frame.end;
706        scan.valid_bytes = cursor as u64;
707    }
708    scan
709}
710
711struct DecodedWalFrame {
712    payload: Vec<u8>,
713    frame_crc: u32,
714    end: usize,
715}
716
717fn decode_next_wal_frame(bytes: &[u8], start: usize, scan: &WalScan) -> Option<DecodedWalFrame> {
718    let remaining = bytes.len().checked_sub(start)?;
719    if remaining < WAL_FRAME_MAGIC.len() {
720        return None;
721    }
722    if &bytes[start..start + WAL_FRAME_MAGIC.len()] != WAL_FRAME_MAGIC {
723        return None;
724    }
725    if remaining < WAL_FRAME_MAGIC.len() + 2 {
726        return None;
727    }
728    let version_offset = start + WAL_FRAME_MAGIC.len();
729    let version = u16::from_le_bytes(bytes[version_offset..version_offset + 2].try_into().ok()?);
730    if version == WAL_FRAME_VERSION {
731        decode_v2_wal_frame(bytes, start, scan)
732    } else {
733        decode_legacy_wal_frame(bytes, start)
734    }
735}
736
737fn decode_v2_wal_frame(bytes: &[u8], start: usize, scan: &WalScan) -> Option<DecodedWalFrame> {
738    if bytes.len().checked_sub(start)? < WAL_FRAME_HEADER_BYTES {
739        return None;
740    }
741    let header_len_offset = start + 10;
742    let header_len = u16::from_le_bytes(
743        bytes[header_len_offset..header_len_offset + 2]
744            .try_into()
745            .ok()?,
746    ) as usize;
747    if header_len != WAL_FRAME_HEADER_BYTES {
748        return None;
749    }
750    let sequence_offset = start + 12;
751    let sequence = u64::from_le_bytes(
752        bytes[sequence_offset..sequence_offset + 8]
753            .try_into()
754            .ok()?,
755    );
756    if sequence != scan.next_sequence {
757        return None;
758    }
759    let payload_len_offset = start + 20;
760    let payload_len = u32::from_le_bytes(
761        bytes[payload_len_offset..payload_len_offset + 4]
762            .try_into()
763            .ok()?,
764    ) as usize;
765    let payload_crc_offset = start + 24;
766    let payload_crc = u32::from_le_bytes(
767        bytes[payload_crc_offset..payload_crc_offset + 4]
768            .try_into()
769            .ok()?,
770    );
771    let previous_frame_crc_offset = start + 28;
772    let previous_frame_crc = u32::from_le_bytes(
773        bytes[previous_frame_crc_offset..previous_frame_crc_offset + 4]
774            .try_into()
775            .ok()?,
776    );
777    if previous_frame_crc != scan.previous_frame_crc {
778        return None;
779    }
780    let header_crc_offset = start + 32;
781    let header_crc = u32::from_le_bytes(
782        bytes[header_crc_offset..header_crc_offset + 4]
783            .try_into()
784            .ok()?,
785    );
786    if header_crc != crc32(&bytes[start..header_crc_offset]) {
787        return None;
788    }
789    let payload_start = start.checked_add(header_len)?;
790    let end = payload_start.checked_add(payload_len)?;
791    if end > bytes.len() {
792        return None;
793    }
794    let payload = bytes[payload_start..end].to_vec();
795    if crc32(&payload) != payload_crc {
796        return None;
797    }
798    Some(DecodedWalFrame {
799        payload,
800        frame_crc: crc32(&bytes[start..end]),
801        end,
802    })
803}
804
805fn decode_legacy_wal_frame(bytes: &[u8], start: usize) -> Option<DecodedWalFrame> {
806    if bytes.len().checked_sub(start)? < LEGACY_WAL_FRAME_HEADER_BYTES {
807        return None;
808    }
809    let payload_len_offset = start + WAL_FRAME_MAGIC.len();
810    let payload_len = u32::from_le_bytes(
811        bytes[payload_len_offset..payload_len_offset + 4]
812            .try_into()
813            .ok()?,
814    ) as usize;
815    let payload_crc_offset = payload_len_offset + 4;
816    let payload_crc = u32::from_le_bytes(
817        bytes[payload_crc_offset..payload_crc_offset + 4]
818            .try_into()
819            .ok()?,
820    );
821    let payload_start = start.checked_add(LEGACY_WAL_FRAME_HEADER_BYTES)?;
822    let end = payload_start.checked_add(payload_len)?;
823    if end > bytes.len() {
824        return None;
825    }
826    let payload = bytes[payload_start..end].to_vec();
827    if crc32(&payload) != payload_crc {
828        return None;
829    }
830    Some(DecodedWalFrame {
831        payload,
832        frame_crc: crc32(&bytes[start..end]),
833        end,
834    })
835}
836
837fn encode_wal_frame(
838    sequence: u64,
839    previous_frame_crc: u32,
840    payload: &[u8],
841) -> RdbFileResult<(Vec<u8>, u32)> {
842    let payload_len = u32::try_from(payload.len())
843        .map_err(|_| RdbFileError::InvalidOperation("embedded wal payload too large".into()))?;
844    let mut frame = Vec::with_capacity(WAL_FRAME_HEADER_BYTES + payload.len());
845    frame.extend_from_slice(WAL_FRAME_MAGIC);
846    frame.extend_from_slice(&WAL_FRAME_VERSION.to_le_bytes());
847    frame.extend_from_slice(&(WAL_FRAME_HEADER_BYTES as u16).to_le_bytes());
848    frame.extend_from_slice(&sequence.to_le_bytes());
849    frame.extend_from_slice(&payload_len.to_le_bytes());
850    frame.extend_from_slice(&crc32(payload).to_le_bytes());
851    frame.extend_from_slice(&previous_frame_crc.to_le_bytes());
852    let header_crc = crc32(&frame);
853    frame.extend_from_slice(&header_crc.to_le_bytes());
854    frame.extend_from_slice(payload);
855    let frame_crc = crc32(&frame);
856    Ok((frame, frame_crc))
857}
858
859fn encode_superblock(superblock: EmbeddedRdbSuperblock) -> RdbFileResult<Vec<u8>> {
860    let mut bytes = vec![0u8; EMBEDDED_RDB_SUPERBLOCK_SIZE as usize];
861    let mut cursor = 0usize;
862    put_bytes(&mut bytes, &mut cursor, SUPERBLOCK_MAGIC);
863    put_u32(&mut bytes, &mut cursor, SUPERBLOCK_VERSION);
864    put_u8(&mut bytes, &mut cursor, superblock.copy_index);
865    put_u64(&mut bytes, &mut cursor, superblock.generation);
866    put_u32(&mut bytes, &mut cursor, superblock.format_version);
867    put_u64(&mut bytes, &mut cursor, superblock.manifest_offset);
868    put_u64(&mut bytes, &mut cursor, superblock.manifest_len);
869    put_u32(&mut bytes, &mut cursor, superblock.manifest_checksum);
870    put_u64(&mut bytes, &mut cursor, superblock.wal_region_offset);
871    put_u64(&mut bytes, &mut cursor, superblock.wal_region_bytes);
872    put_u64(&mut bytes, &mut cursor, superblock.wal_recovery_boundary);
873    put_u64(&mut bytes, &mut cursor, superblock.snapshot_offset);
874    put_u64(&mut bytes, &mut cursor, superblock.snapshot_bytes);
875    put_u32(&mut bytes, &mut cursor, superblock.snapshot_checksum);
876
877    let checksum_offset = bytes.len() - CHECKSUM_LEN;
878    let checksum = crc32(&bytes[..checksum_offset]);
879    bytes[checksum_offset..].copy_from_slice(&checksum.to_le_bytes());
880    Ok(bytes)
881}
882
883fn decode_superblock(copy_index: u8, bytes: &[u8]) -> RdbFileResult<EmbeddedRdbSuperblock> {
884    if bytes.len() != EMBEDDED_RDB_SUPERBLOCK_SIZE as usize {
885        return Err(RdbFileError::InvalidOperation(
886            "invalid embedded superblock size".into(),
887        ));
888    }
889    let checksum_offset = bytes.len() - CHECKSUM_LEN;
890    let stored_checksum = u32::from_le_bytes(bytes[checksum_offset..].try_into().unwrap());
891    let computed_checksum = crc32(&bytes[..checksum_offset]);
892    if stored_checksum != computed_checksum {
893        return Err(RdbFileError::InvalidOperation(
894            "embedded superblock checksum mismatch".into(),
895        ));
896    }
897
898    let mut cursor = 0usize;
899    if take_bytes(bytes, &mut cursor, SUPERBLOCK_MAGIC.len())? != SUPERBLOCK_MAGIC {
900        return Err(RdbFileError::InvalidOperation(
901            "invalid embedded superblock magic".into(),
902        ));
903    }
904    let version = take_u32(bytes, &mut cursor)?;
905    if version != SUPERBLOCK_VERSION {
906        return Err(RdbFileError::InvalidOperation(format!(
907            "unsupported embedded superblock version {version}"
908        )));
909    }
910    let stored_copy_index = take_u8(bytes, &mut cursor)?;
911    if stored_copy_index != copy_index {
912        return Err(RdbFileError::InvalidOperation(
913            "embedded superblock copy index mismatch".into(),
914        ));
915    }
916
917    Ok(EmbeddedRdbSuperblock {
918        copy_index: stored_copy_index,
919        generation: take_u64(bytes, &mut cursor)?,
920        format_version: take_u32(bytes, &mut cursor)?,
921        manifest_offset: take_u64(bytes, &mut cursor)?,
922        manifest_len: take_u64(bytes, &mut cursor)?,
923        manifest_checksum: take_u32(bytes, &mut cursor)?,
924        wal_region_offset: take_u64(bytes, &mut cursor)?,
925        wal_region_bytes: take_u64(bytes, &mut cursor)?,
926        wal_recovery_boundary: take_u64(bytes, &mut cursor)?,
927        snapshot_offset: take_u64(bytes, &mut cursor)?,
928        snapshot_bytes: take_u64(bytes, &mut cursor)?,
929        snapshot_checksum: take_u32(bytes, &mut cursor)?,
930        checksum: stored_checksum,
931    })
932}
933
934fn encode_manifest(manifest: EmbeddedRdbManifest) -> Vec<u8> {
935    let mut bytes = vec![0u8; 8 + 4 + 8 + 8 + 8 + 8 + 8 + 4 + 16 + CHECKSUM_LEN];
936    let mut cursor = 0usize;
937    put_bytes(&mut bytes, &mut cursor, MANIFEST_MAGIC);
938    put_u32(&mut bytes, &mut cursor, manifest.version);
939    put_u64(&mut bytes, &mut cursor, manifest.wal_region_offset);
940    put_u64(&mut bytes, &mut cursor, manifest.wal_region_bytes);
941    put_u64(&mut bytes, &mut cursor, manifest.wal_recovery_boundary);
942    put_u64(&mut bytes, &mut cursor, manifest.snapshot_offset);
943    put_u64(&mut bytes, &mut cursor, manifest.snapshot_bytes);
944    put_u32(&mut bytes, &mut cursor, manifest.snapshot_checksum);
945    put_u128(&mut bytes, &mut cursor, manifest.created_at_unix_ms);
946
947    let checksum_offset = bytes.len() - CHECKSUM_LEN;
948    let checksum = crc32(&bytes[..checksum_offset]);
949    bytes[checksum_offset..].copy_from_slice(&checksum.to_le_bytes());
950    bytes
951}
952
953fn decode_manifest(bytes: &[u8]) -> RdbFileResult<EmbeddedRdbManifest> {
954    let checksum_offset = bytes
955        .len()
956        .checked_sub(CHECKSUM_LEN)
957        .ok_or_else(|| RdbFileError::InvalidOperation("embedded manifest too short".into()))?;
958    let stored_checksum = u32::from_le_bytes(bytes[checksum_offset..].try_into().unwrap());
959    let computed_checksum = crc32(&bytes[..checksum_offset]);
960    if stored_checksum != computed_checksum {
961        return Err(RdbFileError::InvalidOperation(
962            "embedded manifest checksum mismatch".into(),
963        ));
964    }
965
966    let mut cursor = 0usize;
967    if take_bytes(bytes, &mut cursor, MANIFEST_MAGIC.len())? != MANIFEST_MAGIC {
968        return Err(RdbFileError::InvalidOperation(
969            "invalid embedded manifest magic".into(),
970        ));
971    }
972    let version = take_u32(bytes, &mut cursor)?;
973    if version != MANIFEST_VERSION {
974        return Err(RdbFileError::InvalidOperation(format!(
975            "unsupported embedded manifest version {version}"
976        )));
977    }
978    Ok(EmbeddedRdbManifest {
979        version,
980        wal_region_offset: take_u64(bytes, &mut cursor)?,
981        wal_region_bytes: take_u64(bytes, &mut cursor)?,
982        wal_recovery_boundary: take_u64(bytes, &mut cursor)?,
983        snapshot_offset: take_u64(bytes, &mut cursor)?,
984        snapshot_bytes: take_u64(bytes, &mut cursor)?,
985        snapshot_checksum: take_u32(bytes, &mut cursor)?,
986        created_at_unix_ms: take_u128(bytes, &mut cursor)?,
987        checksum: stored_checksum,
988    })
989}
990
991fn trailer_checksum(bytes: &[u8]) -> u32 {
992    let checksum_offset = bytes.len() - CHECKSUM_LEN;
993    u32::from_le_bytes(bytes[checksum_offset..].try_into().unwrap())
994}
995
996fn superblock_offset(copy_index: u8) -> RdbFileResult<u64> {
997    match copy_index {
998        0 => Ok(EMBEDDED_RDB_SUPERBLOCK_0_OFFSET),
999        1 => Ok(EMBEDDED_RDB_SUPERBLOCK_1_OFFSET),
1000        _ => Err(RdbFileError::InvalidOperation(format!(
1001            "invalid embedded superblock copy index {copy_index}"
1002        ))),
1003    }
1004}
1005
1006fn write_at(file: &mut File, offset: u64, bytes: &[u8]) -> RdbFileResult<()> {
1007    file.seek(SeekFrom::Start(offset))?;
1008    file.write_all(bytes)?;
1009    Ok(())
1010}
1011
1012fn crash_inject(point: &str) {
1013    if std::env::var(CRASH_INJECT_ENV).ok().as_deref() == Some(point) {
1014        std::process::exit(173);
1015    }
1016    if crate::buggify!(CRASH_INJECT_ENV, point) {
1017        std::process::exit(173);
1018    }
1019}
1020
1021fn now_unix_ms() -> u128 {
1022    SystemTime::now()
1023        .duration_since(UNIX_EPOCH)
1024        .map(|duration| duration.as_millis())
1025        .unwrap_or(0)
1026}
1027
1028fn put_bytes(target: &mut [u8], cursor: &mut usize, value: &[u8]) {
1029    target[*cursor..*cursor + value.len()].copy_from_slice(value);
1030    *cursor += value.len();
1031}
1032
1033fn put_u8(target: &mut [u8], cursor: &mut usize, value: u8) {
1034    target[*cursor] = value;
1035    *cursor += 1;
1036}
1037
1038fn put_u32(target: &mut [u8], cursor: &mut usize, value: u32) {
1039    put_bytes(target, cursor, &value.to_le_bytes());
1040}
1041
1042fn put_u64(target: &mut [u8], cursor: &mut usize, value: u64) {
1043    put_bytes(target, cursor, &value.to_le_bytes());
1044}
1045
1046fn put_u128(target: &mut [u8], cursor: &mut usize, value: u128) {
1047    put_bytes(target, cursor, &value.to_le_bytes());
1048}
1049
1050fn take_bytes<'a>(bytes: &'a [u8], cursor: &mut usize, len: usize) -> RdbFileResult<&'a [u8]> {
1051    let end = cursor.checked_add(len).ok_or_else(|| {
1052        RdbFileError::InvalidOperation("embedded artifact cursor overflow".into())
1053    })?;
1054    if end > bytes.len() {
1055        return Err(RdbFileError::InvalidOperation(
1056            "embedded artifact truncated".into(),
1057        ));
1058    }
1059    let value = &bytes[*cursor..end];
1060    *cursor = end;
1061    Ok(value)
1062}
1063
1064fn take_u8(bytes: &[u8], cursor: &mut usize) -> RdbFileResult<u8> {
1065    Ok(take_bytes(bytes, cursor, 1)?[0])
1066}
1067
1068fn take_u32(bytes: &[u8], cursor: &mut usize) -> RdbFileResult<u32> {
1069    Ok(u32::from_le_bytes(
1070        take_bytes(bytes, cursor, 4)?.try_into().unwrap(),
1071    ))
1072}
1073
1074fn take_u64(bytes: &[u8], cursor: &mut usize) -> RdbFileResult<u64> {
1075    Ok(u64::from_le_bytes(
1076        take_bytes(bytes, cursor, 8)?.try_into().unwrap(),
1077    ))
1078}
1079
1080fn take_u128(bytes: &[u8], cursor: &mut usize) -> RdbFileResult<u128> {
1081    Ok(u128::from_le_bytes(
1082        take_bytes(bytes, cursor, 16)?.try_into().unwrap(),
1083    ))
1084}