1use std::collections::HashMap;
8use std::fs::{self, File, OpenOptions};
9use std::io::{Read, Seek, SeekFrom, Write};
10use std::path::{Path, PathBuf};
11use std::sync::{Arc, Mutex, OnceLock};
12use std::time::{SystemTime, UNIX_EPOCH};
13
14use fs2::FileExt;
15
16pub type RdbFileResult<T> = Result<T, RdbFileError>;
17
18pub const DEFAULT_FORMAT_VERSION: u32 = 1;
19
20#[derive(Debug)]
21pub enum RdbFileError {
22 InvalidOperation(String),
23 Io(std::io::Error),
24 ZoneUnrecoverable {
28 zone: &'static str,
29 path: PathBuf,
30 },
31}
32
33impl std::fmt::Display for RdbFileError {
34 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
35 match self {
36 Self::InvalidOperation(msg) => write!(f, "INVALID_OPERATION: {msg}"),
37 Self::Io(err) => write!(f, "io error: {err}"),
38 Self::ZoneUnrecoverable { zone, path } => write!(
39 f,
40 "{zone} zone of {} failed validation, so the store will not be opened \
41 (opening it could only return data the zone can no longer vouch for). \
42 Run 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
69pub const EMBEDDED_RDB_MANIFEST_SLOT_SIZE: u64 = 4096;
78pub const EMBEDDED_RDB_MANIFEST_0_OFFSET: u64 = EMBEDDED_RDB_SUPERBLOCK_SIZE * 2;
79pub const EMBEDDED_RDB_MANIFEST_1_OFFSET: u64 =
80 EMBEDDED_RDB_MANIFEST_0_OFFSET + EMBEDDED_RDB_MANIFEST_SLOT_SIZE;
81pub const EMBEDDED_RDB_MANIFEST_ZONE_END: u64 =
82 EMBEDDED_RDB_MANIFEST_1_OFFSET + EMBEDDED_RDB_MANIFEST_SLOT_SIZE;
83
84const SUPERBLOCK_MAGIC: &[u8; 8] = b"RDBSBLK1";
85const MANIFEST_MAGIC: &[u8; 8] = b"RDBMNFS1";
86const SUPERBLOCK_VERSION: u32 = 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 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 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
511fn 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 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}