1use std::fs::{self, File, OpenOptions};
9use std::io::Write;
10use std::path::{Path, PathBuf};
11
12use crate::embedded::{RdbFileError, RdbFileResult};
13
14mod boot;
15mod cache;
16mod extent;
17mod hydrate;
18mod lease;
19mod manifest;
20mod plan;
21mod pointer;
22mod secondary;
23mod snapshot_head;
24
25pub use boot::*;
26pub use cache::*;
27pub use extent::*;
28pub use hydrate::*;
29pub use lease::*;
30pub use manifest::*;
31pub use plan::*;
32pub use pointer::*;
33pub use secondary::*;
34pub use snapshot_head::*;
35
36const SERVERLESS_MANIFEST_MAGIC: &[u8; 8] = b"RDPKMNF1";
37const SERVERLESS_BOOT_INDEX_MAGIC: &[u8; 8] = b"RDPKBIX1";
38const SERVERLESS_GENERATION_POINTER_MAGIC: &[u8; 8] = b"RDPKCUR1";
39const SERVERLESS_EXTENT_INDEX_MAGIC: &[u8; 8] = b"RDPKEXT1";
40const SERVERLESS_SECONDARY_INDEX_MAGIC: &[u8; 8] = b"RDPKSIX1";
41const SERVERLESS_ARTIFACT_VERSION: u16 = 1;
42const CHECKSUM_LEN: usize = 4;
43const CONTENT_HASH_LEN: usize = 32;
44const SERVERLESS_CRASH_INJECT_ENV: &str = "REDDB_SERVERLESS_CRASH_AT";
45
46#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
47pub enum ServerlessPackKind {
48 Manifest,
49 BootIndex,
50 ExtentIndex,
51 HotSnapshot,
52 WalTail,
53 CollectionData,
54 SecondaryIndex,
55 ColdArchive,
56}
57
58impl ServerlessPackKind {
59 pub const fn as_str(self) -> &'static str {
60 match self {
61 Self::Manifest => "manifest",
62 Self::BootIndex => "boot-index",
63 Self::ExtentIndex => "extent-index",
64 Self::HotSnapshot => "hot-snapshot",
65 Self::WalTail => "wal-tail",
66 Self::CollectionData => "collection-data",
67 Self::SecondaryIndex => "secondary-index",
68 Self::ColdArchive => "cold-archive",
69 }
70 }
71}
72
73impl TryFrom<u8> for ServerlessPackKind {
74 type Error = RdbFileError;
75
76 fn try_from(value: u8) -> RdbFileResult<Self> {
77 match value {
78 1 => Ok(Self::Manifest),
79 2 => Ok(Self::BootIndex),
80 3 => Ok(Self::ExtentIndex),
81 4 => Ok(Self::HotSnapshot),
82 5 => Ok(Self::WalTail),
83 6 => Ok(Self::CollectionData),
84 7 => Ok(Self::SecondaryIndex),
85 8 => Ok(Self::ColdArchive),
86 other => Err(RdbFileError::InvalidOperation(format!(
87 "unknown serverless pack kind {other}"
88 ))),
89 }
90 }
91}
92
93impl From<ServerlessPackKind> for u8 {
94 fn from(value: ServerlessPackKind) -> Self {
95 match value {
96 ServerlessPackKind::Manifest => 1,
97 ServerlessPackKind::BootIndex => 2,
98 ServerlessPackKind::ExtentIndex => 3,
99 ServerlessPackKind::HotSnapshot => 4,
100 ServerlessPackKind::WalTail => 5,
101 ServerlessPackKind::CollectionData => 6,
102 ServerlessPackKind::SecondaryIndex => 7,
103 ServerlessPackKind::ColdArchive => 8,
104 }
105 }
106}
107
108#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
109pub struct ServerlessContentHash(pub [u8; CONTENT_HASH_LEN]);
110
111impl ServerlessContentHash {
112 pub const ZERO: Self = Self([0; CONTENT_HASH_LEN]);
113
114 pub fn from_bytes(bytes: &[u8]) -> Self {
115 Self(*blake3::hash(bytes).as_bytes())
116 }
117
118 pub fn is_zero(self) -> bool {
119 self.0 == [0; CONTENT_HASH_LEN]
120 }
121}
122
123fn kind_for_artifact_path(path: &Path) -> ServerlessPackKind {
124 match path.file_stem().and_then(|stem| stem.to_str()) {
125 Some("manifest") => ServerlessPackKind::Manifest,
126 Some("boot-index") => ServerlessPackKind::BootIndex,
127 Some("extent-index") => ServerlessPackKind::ExtentIndex,
128 Some("hot-snapshot") => ServerlessPackKind::HotSnapshot,
129 Some("wal-tail") => ServerlessPackKind::WalTail,
130 Some("collection-data") => ServerlessPackKind::CollectionData,
131 Some("secondary-index") => ServerlessPackKind::SecondaryIndex,
132 Some("cold-archive") => ServerlessPackKind::ColdArchive,
133 _ => ServerlessPackKind::ColdArchive,
134 }
135}
136
137fn relative_to_generation_dir(path: &Path) -> PathBuf {
138 path.file_name()
139 .map(PathBuf::from)
140 .unwrap_or_else(|| path.to_path_buf())
141}
142
143fn write_bytes(path: impl AsRef<Path>, bytes: &[u8]) -> RdbFileResult<()> {
144 let path = path.as_ref();
145 if let Some(parent) = path.parent() {
146 fs::create_dir_all(parent)?;
147 }
148 let tmp_path = crate::layout::atomic_temp_path(path);
149 {
150 let mut file = OpenOptions::new()
151 .create(true)
152 .truncate(true)
153 .write(true)
154 .open(&tmp_path)?;
155 file.write_all(bytes)?;
156 crash_inject("serverless_pack_after_tmp_write");
157 file.sync_all()?;
158 crash_inject("serverless_pack_after_tmp_sync");
159 }
160 fs::rename(&tmp_path, path)?;
161 crash_inject("serverless_pack_after_rename");
162 if let Some(parent) = path.parent() {
163 if let Ok(dir) = File::open(parent) {
164 let _ = dir.sync_all();
165 }
166 }
167 crash_inject("serverless_pack_after_dir_sync");
168 Ok(())
169}
170
171fn write_current_pointer_bytes(path: impl AsRef<Path>, bytes: &[u8]) -> RdbFileResult<()> {
172 let path = path.as_ref();
173 if let Some(parent) = path.parent() {
174 fs::create_dir_all(parent)?;
175 }
176 let tmp_path = crate::layout::atomic_temp_path(path);
177 {
178 let mut file = OpenOptions::new()
179 .create(true)
180 .truncate(true)
181 .write(true)
182 .open(&tmp_path)?;
183 file.write_all(bytes)?;
184 crash_inject("current_pointer_after_tmp_write");
185 file.sync_all()?;
186 crash_inject("current_pointer_after_tmp_sync");
187 }
188 fs::rename(&tmp_path, path)?;
189 crash_inject("current_pointer_after_rename");
190 if let Some(parent) = path.parent() {
191 if let Ok(dir) = File::open(parent) {
192 let _ = dir.sync_all();
193 }
194 }
195 crash_inject("current_pointer_after_dir_sync");
196 Ok(())
197}
198
199fn crash_inject(point: &str) {
200 if std::env::var(SERVERLESS_CRASH_INJECT_ENV).ok().as_deref() == Some(point) {
201 std::process::exit(173);
202 }
203 if crate::buggify!(SERVERLESS_CRASH_INJECT_ENV, point) {
204 std::process::exit(173);
205 }
206}
207
208fn verify_checksum(bytes: &[u8]) -> RdbFileResult<()> {
209 let Some(checksum_offset) = bytes.len().checked_sub(CHECKSUM_LEN) else {
210 return Err(RdbFileError::InvalidOperation(
211 "serverless artifact too short".into(),
212 ));
213 };
214 let stored = u32::from_le_bytes(bytes[checksum_offset..].try_into().unwrap());
215 let computed = crc32(&bytes[..checksum_offset]);
216 if stored != computed {
217 return Err(RdbFileError::InvalidOperation(format!(
218 "serverless artifact checksum mismatch: stored {stored:#010x}, computed {computed:#010x}"
219 )));
220 }
221 Ok(())
222}
223
224fn crc32(data: &[u8]) -> u32 {
225 let mut hasher = crc32fast::Hasher::new();
226 hasher.update(data);
227 hasher.finalize()
228}
229
230fn expect_magic(bytes: &[u8], cursor: &mut usize, magic: &[u8]) -> RdbFileResult<()> {
231 let actual = take_bytes(bytes, cursor, magic.len())?;
232 if actual != magic {
233 return Err(RdbFileError::InvalidOperation(
234 "invalid serverless artifact magic".into(),
235 ));
236 }
237 Ok(())
238}
239
240fn put_u16(out: &mut Vec<u8>, value: u16) {
241 out.extend_from_slice(&value.to_le_bytes());
242}
243
244fn put_u32(out: &mut Vec<u8>, value: u32) {
245 out.extend_from_slice(&value.to_le_bytes());
246}
247
248fn put_u64(out: &mut Vec<u8>, value: u64) {
249 out.extend_from_slice(&value.to_le_bytes());
250}
251
252fn put_string(out: &mut Vec<u8>, value: &str) {
253 put_u32(out, value.len() as u32);
254 out.extend_from_slice(value.as_bytes());
255}
256
257fn put_bytes(out: &mut Vec<u8>, value: &[u8]) {
258 put_u32(out, value.len() as u32);
259 out.extend_from_slice(value);
260}
261
262fn put_content_hash(out: &mut Vec<u8>, value: ServerlessContentHash) {
263 out.extend_from_slice(&value.0);
264}
265
266fn take_bytes<'a>(bytes: &'a [u8], cursor: &mut usize, len: usize) -> RdbFileResult<&'a [u8]> {
267 let end = cursor
268 .checked_add(len)
269 .ok_or_else(|| RdbFileError::InvalidOperation("serverless cursor overflow".into()))?;
270 if end > bytes.len().saturating_sub(CHECKSUM_LEN) {
271 return Err(RdbFileError::InvalidOperation(
272 "serverless artifact truncated".into(),
273 ));
274 }
275 let value = &bytes[*cursor..end];
276 *cursor = end;
277 Ok(value)
278}
279
280fn reject_trailing_bytes(bytes: &[u8], cursor: usize) -> RdbFileResult<()> {
281 if cursor != bytes.len().saturating_sub(CHECKSUM_LEN) {
282 return Err(RdbFileError::InvalidOperation(
283 "serverless artifact has trailing bytes".into(),
284 ));
285 }
286 Ok(())
287}
288
289fn take_u8(bytes: &[u8], cursor: &mut usize) -> RdbFileResult<u8> {
290 Ok(take_bytes(bytes, cursor, 1)?[0])
291}
292
293fn take_u16(bytes: &[u8], cursor: &mut usize) -> RdbFileResult<u16> {
294 Ok(u16::from_le_bytes(
295 take_bytes(bytes, cursor, 2)?.try_into().unwrap(),
296 ))
297}
298
299fn take_u32(bytes: &[u8], cursor: &mut usize) -> RdbFileResult<u32> {
300 Ok(u32::from_le_bytes(
301 take_bytes(bytes, cursor, 4)?.try_into().unwrap(),
302 ))
303}
304
305fn take_u64(bytes: &[u8], cursor: &mut usize) -> RdbFileResult<u64> {
306 Ok(u64::from_le_bytes(
307 take_bytes(bytes, cursor, 8)?.try_into().unwrap(),
308 ))
309}
310
311fn take_string(bytes: &[u8], cursor: &mut usize) -> RdbFileResult<String> {
312 let len = take_u32(bytes, cursor)? as usize;
313 let raw = take_bytes(bytes, cursor, len)?;
314 std::str::from_utf8(raw)
315 .map(|value| value.to_string())
316 .map_err(|err| RdbFileError::InvalidOperation(format!("invalid utf-8 string: {err}")))
317}
318
319fn take_vec_bytes(bytes: &[u8], cursor: &mut usize) -> RdbFileResult<Vec<u8>> {
320 let len = take_u32(bytes, cursor)? as usize;
321 Ok(take_bytes(bytes, cursor, len)?.to_vec())
322}
323
324fn take_content_hash(bytes: &[u8], cursor: &mut usize) -> RdbFileResult<ServerlessContentHash> {
325 let raw = take_bytes(bytes, cursor, CONTENT_HASH_LEN)?;
326 let mut hash = [0u8; CONTENT_HASH_LEN];
327 hash.copy_from_slice(raw);
328 Ok(ServerlessContentHash(hash))
329}
330
331#[cfg(test)]
332mod tests {
333 use super::*;
334
335 #[test]
336 fn serverless_paths_are_generation_scoped_and_deterministic() {
337 let plan = ServerlessFilePlan::new("/tmp/reddb", "tenant-a/db", 42);
338 assert_eq!(
339 plan.manifest_path(),
340 PathBuf::from("/tmp/reddb/tenant-a/db/g00000000000000000042/manifest.redpack")
341 );
342 assert_eq!(
343 plan.boot_index_path(),
344 PathBuf::from("/tmp/reddb/tenant-a/db/g00000000000000000042/boot-index.redpack")
345 );
346 assert!(ServerlessFilePlan::is_generation_dir(Path::new(
347 "g00000000000000000042"
348 )));
349 assert!(!ServerlessFilePlan::is_generation_dir(Path::new("g42")));
350 }
351
352 #[test]
353 fn cold_start_fetches_manifest_boot_snapshot_then_wal_tail() {
354 let plan = ServerlessFilePlan::new("/tmp/reddb", "db", 7);
355 let boot = ServerlessBootPlan::cold(&plan);
356 assert_eq!(boot.required_first[0], plan.manifest_path());
357 assert_eq!(boot.required_first[1], plan.boot_index_path());
358 assert_eq!(boot.required_first[2], plan.extent_index_path());
359 assert_eq!(boot.required_first[3], plan.hot_snapshot_path());
360 assert_eq!(boot.required_first[4], plan.wal_tail_path());
361 assert_eq!(boot.lazy_after_open.len(), 3);
362 }
363
364 #[test]
365 fn manifest_round_trips_with_crc_checked_binary_codec() {
366 let mut manifest = ServerlessManifest::new("tenant/db", 11);
367 manifest.push(ServerlessManifestEntry::from_bytes(
368 ServerlessPackKind::WalTail,
369 "wal-tail.redpack",
370 b"wal tail payload",
371 ));
372 manifest.push(ServerlessManifestEntry::from_bytes(
373 ServerlessPackKind::BootIndex,
374 "boot-index.redpack",
375 b"boot index payload",
376 ));
377
378 let encoded = manifest.encode();
379 let decoded = ServerlessManifest::decode(&encoded).expect("decode manifest");
380 assert_eq!(decoded, manifest);
381 assert!(!decoded.entries[0].content_hash.is_zero());
382
383 let mut corrupt = encoded;
384 let last_payload_byte = corrupt.len() - CHECKSUM_LEN - 1;
385 corrupt[last_payload_byte] ^= 0x01;
386 let err = ServerlessManifest::decode(&corrupt).expect_err("checksum catches corruption");
387 assert!(err.to_string().contains("checksum mismatch"), "{err}");
388 }
389
390 #[test]
391 fn boot_index_round_trips_and_preserves_coldstart_order() {
392 let plan = ServerlessFilePlan::new("/tmp/reddb", "db", 9);
393 let index = ServerlessBootIndex::from_plan(&plan);
394
395 assert_eq!(
396 index.required_first(),
397 vec![
398 PathBuf::from("manifest.redpack"),
399 PathBuf::from("boot-index.redpack"),
400 PathBuf::from("extent-index.redpack"),
401 PathBuf::from("hot-snapshot.redpack"),
402 PathBuf::from("wal-tail.redpack"),
403 ]
404 );
405 assert_eq!(
406 index.lazy_after_open(),
407 vec![
408 PathBuf::from("collection-data.redpack"),
409 PathBuf::from("secondary-index.redpack"),
410 PathBuf::from("cold-archive.redpack"),
411 ]
412 );
413
414 let decoded = ServerlessBootIndex::decode(&index.encode()).expect("decode boot index");
415 assert_eq!(decoded, index);
416 }
417
418 #[test]
419 fn collection_data_extent_ref_uses_canonical_pack_path() {
420 let plan = ServerlessFilePlan::new("/tmp/reddb", "db", 9);
421 let payload = b"collection snapshot bytes";
422 let extent = plan
423 .collection_data_extent_ref("events", 12, payload, true)
424 .expect("extent ref");
425
426 assert_eq!(extent.collection, "events");
427 assert_eq!(
428 extent.relative_path,
429 PathBuf::from("collection-data.redpack")
430 );
431 assert_eq!(extent.offset, 12);
432 assert_eq!(extent.bytes, payload.len() as u64);
433 assert!(extent.hot);
434 }
435
436 #[test]
437 fn manifest_rejects_trailing_payload_bytes() {
438 let manifest = ServerlessManifest::new("tenant/db", 11);
439 let mut encoded = manifest.encode();
440 encoded.truncate(encoded.len() - CHECKSUM_LEN);
441 encoded.push(0xAA);
442 let checksum = crc32(&encoded);
443 put_u32(&mut encoded, checksum);
444
445 let err = ServerlessManifest::decode(&encoded).expect_err("trailing bytes rejected");
446 assert!(err.to_string().contains("trailing bytes"), "{err}");
447 }
448
449 #[test]
450 fn generation_pointer_round_trips_and_points_to_immutable_manifest() {
451 let plan = ServerlessFilePlan::new("/tmp/reddb", "tenant/db", 19);
452 let mut manifest = ServerlessManifest::new("tenant/db", 19);
453 manifest.push(ServerlessManifestEntry::from_bytes(
454 ServerlessPackKind::HotSnapshot,
455 "hot-snapshot.redpack",
456 b"snapshot",
457 ));
458
459 let pointer = ServerlessGenerationPointer::from_manifest(&plan, &manifest);
460 assert_eq!(pointer.generation, 19);
461 assert_eq!(
462 pointer.manifest_relative_path,
463 PathBuf::from("g00000000000000000019/manifest.redpack")
464 );
465 assert!(!pointer.manifest_content_hash.is_zero());
466
467 let decoded =
468 ServerlessGenerationPointer::decode(&pointer.encode()).expect("decode pointer");
469 assert_eq!(decoded, pointer);
470 }
471
472 #[test]
473 fn extent_index_finds_key_ranges_and_hot_prefetch_paths() {
474 let mut index = ServerlessExtentIndex::new(21);
475 index.push(
476 ServerlessExtentRef::new(
477 "orders",
478 b"a".to_vec(),
479 b"m".to_vec(),
480 "orders-000.redpack",
481 0,
482 b"orders-a-m",
483 true,
484 )
485 .expect("extent"),
486 );
487 index.push(
488 ServerlessExtentRef::new(
489 "orders",
490 b"m".to_vec(),
491 b"z".to_vec(),
492 "orders-001.redpack",
493 0,
494 b"orders-m-z",
495 false,
496 )
497 .expect("extent"),
498 );
499
500 let matches = index.extents_for_key("orders", b"b");
501 assert_eq!(matches.len(), 1);
502 assert_eq!(
503 matches[0].relative_path,
504 PathBuf::from("orders-000.redpack")
505 );
506 assert_eq!(
507 index.hot_prefetch_paths(),
508 vec![PathBuf::from("orders-000.redpack")]
509 );
510
511 let decoded = ServerlessExtentIndex::decode(&index.encode()).expect("decode extent index");
512 assert_eq!(decoded, index);
513 }
514
515 #[test]
516 fn hydration_plan_uses_only_matching_extents() {
517 let mut index = ServerlessExtentIndex::new(22);
518 index.push(
519 ServerlessExtentRef::new(
520 "orders",
521 b"a".to_vec(),
522 b"m".to_vec(),
523 "orders-000.redpack",
524 64,
525 b"orders-a-m",
526 true,
527 )
528 .expect("extent"),
529 );
530 index.push(
531 ServerlessExtentRef::new(
532 "orders",
533 b"m".to_vec(),
534 b"z".to_vec(),
535 "orders-001.redpack",
536 128,
537 b"orders-m-z",
538 false,
539 )
540 .expect("extent"),
541 );
542
543 let plan = index.hydration_plan_for_key("orders", b"n");
544 assert_eq!(plan.requests.len(), 1);
545 assert_eq!(
546 plan.requests[0].relative_path,
547 PathBuf::from("orders-001.redpack")
548 );
549 assert_eq!(plan.requests[0].offset, 128);
550 assert_eq!(plan.total_bytes(), b"orders-m-z".len() as u64);
551
552 let hot = index.hot_hydration_plan();
553 assert_eq!(hot.requests.len(), 1);
554 assert_eq!(
555 hot.requests[0].relative_path,
556 PathBuf::from("orders-000.redpack")
557 );
558 }
559
560 #[test]
561 fn hydration_plan_for_range_uses_overlapping_extents() {
562 let mut index = ServerlessExtentIndex::new(27);
563 index.push(
564 ServerlessExtentRef::new(
565 "orders",
566 b"a".to_vec(),
567 b"f".to_vec(),
568 "orders-000.redpack",
569 0,
570 b"orders-a-f",
571 true,
572 )
573 .expect("extent"),
574 );
575 index.push(
576 ServerlessExtentRef::new(
577 "orders",
578 b"f".to_vec(),
579 b"p".to_vec(),
580 "orders-001.redpack",
581 64,
582 b"orders-f-p",
583 false,
584 )
585 .expect("extent"),
586 );
587 index.push(
588 ServerlessExtentRef::new(
589 "orders",
590 b"p".to_vec(),
591 b"z".to_vec(),
592 "orders-002.redpack",
593 128,
594 b"orders-p-z",
595 false,
596 )
597 .expect("extent"),
598 );
599 index.push(
600 ServerlessExtentRef::new(
601 "users",
602 b"a".to_vec(),
603 b"z".to_vec(),
604 "users-000.redpack",
605 0,
606 b"users-a-z",
607 false,
608 )
609 .expect("extent"),
610 );
611
612 let plan = index
613 .hydration_plan_for_range("orders", b"e", b"q")
614 .expect("range plan");
615 assert_eq!(plan.requests.len(), 3);
616 assert_eq!(
617 plan.requests
618 .iter()
619 .map(|request| request.relative_path.clone())
620 .collect::<Vec<_>>(),
621 vec![
622 PathBuf::from("orders-000.redpack"),
623 PathBuf::from("orders-001.redpack"),
624 PathBuf::from("orders-002.redpack"),
625 ]
626 );
627
628 let err = index
629 .hydration_plan_for_range("orders", b"q", b"e")
630 .expect_err("invalid range rejected");
631 assert!(err.to_string().contains("range_start"), "{err}");
632 }
633
634 #[test]
635 fn secondary_index_round_trips_and_builds_collection_hydration_plan() {
636 let mut extent_index = ServerlessExtentIndex::new(29);
637 extent_index.push(
638 ServerlessExtentRef::new(
639 "orders",
640 b"a".to_vec(),
641 b"m".to_vec(),
642 "collection-data.redpack",
643 0,
644 b"orders-left",
645 true,
646 )
647 .expect("orders left"),
648 );
649 extent_index.push(
650 ServerlessExtentRef::new(
651 "orders",
652 b"m".to_vec(),
653 b"z".to_vec(),
654 "collection-data.redpack",
655 11,
656 b"orders-right",
657 false,
658 )
659 .expect("orders right"),
660 );
661 extent_index.push(
662 ServerlessExtentRef::new(
663 "users",
664 b"a".to_vec(),
665 b"z".to_vec(),
666 "collection-data.redpack",
667 23,
668 b"users",
669 false,
670 )
671 .expect("users"),
672 );
673
674 let secondary = ServerlessSecondaryIndex::from_extent_index(&extent_index);
675 let decoded =
676 ServerlessSecondaryIndex::decode(&secondary.encode()).expect("decode secondary index");
677 assert_eq!(decoded, secondary);
678
679 let hydration = decoded.hydration_plan_for_collection("orders");
680 assert_eq!(hydration.generation, 29);
681 assert_eq!(hydration.requests.len(), 2);
682 assert_eq!(hydration.total_bytes(), 23);
683 assert!(hydration.requests[0].content_hash != ServerlessContentHash::ZERO);
684 }
685
686 #[test]
687 fn hydrate_local_plan_reads_only_requested_byte_ranges() {
688 let root = temp_root("serverless-hydrate-range");
689 let plan = ServerlessFilePlan::new(&root, "db", 23);
690 let collection_payload = b"aaaabbbbcccc";
691 let mut index = ServerlessExtentIndex::new(23);
692 index.push(
693 ServerlessExtentRef::new(
694 "orders",
695 b"a".to_vec(),
696 b"m".to_vec(),
697 "collection-data.redpack",
698 4,
699 b"bbbb",
700 true,
701 )
702 .expect("extent"),
703 );
704 plan.publish_core_generation(&index, collection_payload, b"secondary")
705 .expect("publish generation");
706
707 let hydration = index.hydration_plan_for_key("orders", b"b");
708 let hydrated = plan
709 .hydrate_local_plan(&hydration)
710 .expect("hydrate local range");
711 assert_eq!(hydrated.len(), 1);
712 assert_eq!(hydrated[0].payload, b"bbbb");
713 assert_eq!(hydrated[0].request.offset, 4);
714 assert_eq!(hydrated[0].request.bytes, 4);
715
716 let _ = std::fs::remove_dir_all(root);
717 }
718
719 #[test]
720 fn hydrate_local_plan_rejects_corrupt_or_out_of_bounds_ranges() {
721 let root = temp_root("serverless-hydrate-corrupt");
722 let plan = ServerlessFilePlan::new(&root, "db", 24);
723 let mut index = ServerlessExtentIndex::new(24);
724 index.push(
725 ServerlessExtentRef::new(
726 "orders",
727 b"a".to_vec(),
728 b"m".to_vec(),
729 "collection-data.redpack",
730 4,
731 b"bbbb",
732 true,
733 )
734 .expect("extent"),
735 );
736 plan.publish_core_generation(&index, b"aaaabbbbcccc", b"secondary")
737 .expect("publish generation");
738 std::fs::write(plan.collection_data_path(), b"aaaaBBBBcccc").expect("corrupt pack");
739
740 let hydration = index.hydration_plan_for_key("orders", b"b");
741 let err = plan
742 .hydrate_local_plan(&hydration)
743 .expect_err("corrupt range rejected");
744 assert!(err.to_string().contains("checksum mismatch"), "{err}");
745
746 let mut out_of_bounds = hydration.clone();
747 out_of_bounds.requests[0].offset = 11;
748 let err = plan
749 .hydrate_local_plan(&out_of_bounds)
750 .expect_err("out of bounds rejected");
751 assert!(err.to_string().contains("exceeds pack"), "{err}");
752
753 let _ = std::fs::remove_dir_all(root);
754 }
755
756 #[test]
757 fn prefetch_hot_extents_hydrates_only_hot_ranges() {
758 let root = temp_root("serverless-hot-prefetch");
759 let plan = ServerlessFilePlan::new(&root, "db", 25);
760 let mut index = ServerlessExtentIndex::new(25);
761 index.push(
762 ServerlessExtentRef::new(
763 "orders",
764 b"a".to_vec(),
765 b"m".to_vec(),
766 "collection-data.redpack",
767 0,
768 b"hot!",
769 true,
770 )
771 .expect("hot extent"),
772 );
773 index.push(
774 ServerlessExtentRef::new(
775 "orders",
776 b"m".to_vec(),
777 b"z".to_vec(),
778 "collection-data.redpack",
779 4,
780 b"cold",
781 false,
782 )
783 .expect("cold extent"),
784 );
785 plan.publish_core_generation(&index, b"hot!cold", b"secondary")
786 .expect("publish generation");
787
788 let hydrated = plan.prefetch_hot_extents(&index).expect("prefetch hot");
789 assert_eq!(hydrated.len(), 1);
790 assert_eq!(hydrated[0].payload, b"hot!");
791
792 let _ = std::fs::remove_dir_all(root);
793 }
794
795 #[test]
796 fn prefetch_hot_extents_cached_populates_only_hot_cache_entries() {
797 let root = temp_root("serverless-hot-prefetch-cache");
798 let plan = ServerlessFilePlan::new(&root, "db", 29);
799 let cache = ServerlessLocalCache::new(root.join("cache"), 29);
800 let mut index = ServerlessExtentIndex::new(29);
801 index.push(
802 ServerlessExtentRef::new(
803 "orders",
804 b"a".to_vec(),
805 b"m".to_vec(),
806 "collection-data.redpack",
807 0,
808 b"hot!",
809 true,
810 )
811 .expect("hot extent"),
812 );
813 index.push(
814 ServerlessExtentRef::new(
815 "orders",
816 b"m".to_vec(),
817 b"z".to_vec(),
818 "collection-data.redpack",
819 4,
820 b"cold",
821 false,
822 )
823 .expect("cold extent"),
824 );
825 plan.publish_core_generation(&index, b"hot!cold", b"secondary")
826 .expect("publish generation");
827
828 let hot_request = index.hydration_plan_for_key("orders", b"b").requests[0].clone();
829 let cold_request = index.hydration_plan_for_key("orders", b"n").requests[0].clone();
830
831 let hydrated = plan
832 .prefetch_hot_extents_cached(&index, &cache)
833 .expect("prefetch hot into cache");
834 assert_eq!(hydrated.len(), 1);
835 assert_eq!(hydrated[0].payload, b"hot!");
836 assert_eq!(
837 std::fs::read(cache.path_for_request(&hot_request)).expect("read hot cache"),
838 b"hot!"
839 );
840 assert!(
841 !cache.path_for_request(&cold_request).exists(),
842 "cold extent should not be prefetched into cache"
843 );
844
845 let _ = std::fs::remove_dir_all(root);
846 }
847
848 #[test]
849 fn hydrate_local_request_cached_validates_and_repairs_corrupt_cache() {
850 let root = temp_root("serverless-hydrate-cache");
851 let plan = ServerlessFilePlan::new(&root, "db", 26);
852 let cache = ServerlessLocalCache::new(root.join("cache"), 26);
853 let mut index = ServerlessExtentIndex::new(26);
854 index.push(
855 ServerlessExtentRef::new(
856 "orders",
857 b"a".to_vec(),
858 b"z".to_vec(),
859 "collection-data.redpack",
860 4,
861 b"bbbb",
862 true,
863 )
864 .expect("extent"),
865 );
866 plan.publish_core_generation(&index, b"aaaabbbbcccc", b"secondary")
867 .expect("publish generation");
868 let request = index.hydration_plan_for_key("orders", b"m").requests[0].clone();
869
870 let first = plan
871 .hydrate_local_request_cached(&request, &cache)
872 .expect("hydrate and cache");
873 assert_eq!(first.payload, b"bbbb");
874 let cache_path = cache.path_for_request(&request);
875 assert_eq!(std::fs::read(&cache_path).expect("read cache"), b"bbbb");
876
877 std::fs::write(&cache_path, b"xxxx").expect("corrupt cache");
878 let repaired = plan
879 .hydrate_local_request_cached(&request, &cache)
880 .expect("repair corrupt cache from pack");
881 assert_eq!(repaired.payload, b"bbbb");
882 assert_eq!(
883 std::fs::read(&cache_path).expect("read repaired cache"),
884 b"bbbb"
885 );
886
887 let _ = std::fs::remove_dir_all(root);
888 }
889
890 #[test]
891 fn hydrate_local_plan_cached_populates_multiple_range_entries() {
892 let root = temp_root("serverless-hydrate-plan-cache");
893 let plan = ServerlessFilePlan::new(&root, "db", 28);
894 let cache = ServerlessLocalCache::new(root.join("cache"), 28);
895 let mut index = ServerlessExtentIndex::new(28);
896 index.push(
897 ServerlessExtentRef::new(
898 "orders",
899 b"a".to_vec(),
900 b"m".to_vec(),
901 "collection-data.redpack",
902 0,
903 b"left",
904 true,
905 )
906 .expect("left extent"),
907 );
908 index.push(
909 ServerlessExtentRef::new(
910 "orders",
911 b"m".to_vec(),
912 b"z".to_vec(),
913 "collection-data.redpack",
914 4,
915 b"right",
916 false,
917 )
918 .expect("right extent"),
919 );
920 plan.publish_core_generation(&index, b"leftright", b"secondary")
921 .expect("publish generation");
922
923 let hydration = index
924 .hydration_plan_for_range("orders", b"b", b"y")
925 .expect("range hydration plan");
926 let hydrated = plan
927 .hydrate_local_plan_cached(&hydration, &cache)
928 .expect("hydrate cached plan");
929 assert_eq!(hydrated.len(), 2);
930 assert_eq!(hydrated[0].payload, b"left");
931 assert_eq!(hydrated[1].payload, b"right");
932 for range in hydrated {
933 assert!(cache.path_for_request(&range.request).exists());
934 }
935
936 let _ = std::fs::remove_dir_all(root);
937 }
938
939 #[test]
940 fn cached_hydration_enforces_max_hot_bytes_after_writes() {
941 let root = temp_root("serverless-hydrate-cache-budget");
942 let plan =
943 ServerlessFilePlan::new(&root, "db", 29).with_cache_policy(ServerlessCachePolicy {
944 max_hot_bytes: 5,
945 ..ServerlessCachePolicy::default()
946 });
947 let cache = ServerlessLocalCache::new(root.join("cache"), 29);
948 let mut index = ServerlessExtentIndex::new(29);
949 index.push(
950 ServerlessExtentRef::new(
951 "orders",
952 b"a".to_vec(),
953 b"m".to_vec(),
954 "collection-data.redpack",
955 0,
956 b"left",
957 true,
958 )
959 .expect("left extent"),
960 );
961 index.push(
962 ServerlessExtentRef::new(
963 "orders",
964 b"m".to_vec(),
965 b"z".to_vec(),
966 "collection-data.redpack",
967 4,
968 b"right",
969 true,
970 )
971 .expect("right extent"),
972 );
973 plan.publish_core_generation(&index, b"leftright", b"secondary")
974 .expect("publish generation");
975
976 let hydration = index
977 .hydration_plan_for_range("orders", b"a", b"z")
978 .expect("range hydration plan");
979 let hydrated = plan
980 .hydrate_local_plan_cached(&hydration, &cache)
981 .expect("hydrate cached plan");
982 assert_eq!(hydrated.len(), 2);
983 assert_eq!(hydrated[0].payload, b"left");
984 assert_eq!(hydrated[1].payload, b"right");
985
986 let entries = cache.cached_entries().expect("cache entries");
987 let cached_bytes: u64 = entries.iter().map(|entry| entry.bytes).sum();
988 assert!(
989 cached_bytes <= 5,
990 "cache should stay within max_hot_bytes, got {cached_bytes}"
991 );
992
993 let _ = std::fs::remove_dir_all(root);
994 }
995
996 #[test]
997 fn cache_eviction_prefers_cold_old_entries() {
998 let entries = vec![
999 ServerlessCacheEntry::new("cold-old.redpack", 100, false, 10),
1000 ServerlessCacheEntry::new("hot-old.redpack", 100, true, 1),
1001 ServerlessCacheEntry::new("cold-new.redpack", 100, false, 20),
1002 ];
1003
1004 let plan = ServerlessCacheEvictionPlan::plan(&entries, 150);
1005 assert_eq!(
1006 plan.evict,
1007 vec![
1008 PathBuf::from("cold-old.redpack"),
1009 PathBuf::from("cold-new.redpack"),
1010 ]
1011 );
1012 assert_eq!(plan.bytes_after_eviction, 100);
1013 }
1014
1015 #[test]
1016 fn extent_index_writes_and_reads_from_disk() {
1017 let root = temp_root("serverless-extent-index");
1018 let plan = ServerlessFilePlan::new(&root, "db", 5);
1019 let mut index = ServerlessExtentIndex::new(5);
1020 index.push(
1021 ServerlessExtentRef::new(
1022 "events",
1023 b"2026-01".to_vec(),
1024 b"2026-02".to_vec(),
1025 "events-2026-01.redpack",
1026 128,
1027 b"payload",
1028 true,
1029 )
1030 .expect("extent"),
1031 );
1032
1033 index
1034 .write_to_path(plan.extent_index_path())
1035 .expect("write extent index");
1036 assert_eq!(
1037 ServerlessExtentIndex::read_from_path(plan.extent_index_path())
1038 .expect("read extent index"),
1039 index
1040 );
1041 let _ = std::fs::remove_dir_all(root);
1042 }
1043
1044 #[test]
1045 fn manifest_and_boot_index_write_and_read_from_disk() {
1046 let root = temp_root("serverless-manifest");
1047 let plan = ServerlessFilePlan::new(&root, "db", 1);
1048
1049 let mut manifest = ServerlessManifest::new("db", 1);
1050 manifest.push(ServerlessManifestEntry::new(
1051 ServerlessPackKind::Manifest,
1052 "manifest.redpack",
1053 128,
1054 0xCAFE_BABE,
1055 ));
1056 manifest
1057 .write_to_path(plan.manifest_path())
1058 .expect("write manifest");
1059
1060 let boot_index = ServerlessBootIndex::from_plan(&plan);
1061 boot_index
1062 .write_to_path(plan.boot_index_path())
1063 .expect("write boot index");
1064
1065 assert_eq!(
1066 ServerlessManifest::read_from_path(plan.manifest_path()).expect("read manifest"),
1067 manifest
1068 );
1069 assert_eq!(
1070 ServerlessBootIndex::read_from_path(plan.boot_index_path()).expect("read boot index"),
1071 boot_index
1072 );
1073
1074 let _ = std::fs::remove_dir_all(root);
1075 }
1076
1077 #[test]
1078 fn publish_core_generation_writes_required_packs_before_current_pointer() {
1079 let root = temp_root("serverless-publish-core");
1080 let plan = ServerlessFilePlan::new(&root, "db", 12);
1081 let mut extent_index = ServerlessExtentIndex::new(12);
1082 extent_index.push(
1083 ServerlessExtentRef::new(
1084 "events",
1085 b"a".to_vec(),
1086 b"z".to_vec(),
1087 "collection-data.redpack",
1088 0,
1089 b"collection-bytes",
1090 true,
1091 )
1092 .expect("extent"),
1093 );
1094
1095 let pointer = plan
1096 .publish_core_generation(&extent_index, b"collection-bytes", b"secondary-bytes")
1097 .expect("publish core generation");
1098 assert_eq!(pointer.generation, 12);
1099 assert_eq!(plan.read_current_pointer().expect("read current"), pointer);
1100 assert_eq!(
1101 plan.read_current_pointer_verified()
1102 .expect("read verified current"),
1103 pointer
1104 );
1105 assert!(plan.boot_index_path().exists());
1106 assert!(plan.extent_index_path().exists());
1107 assert!(plan.collection_data_path().exists());
1108 assert!(plan.secondary_index_path().exists());
1109
1110 let manifest =
1111 ServerlessManifest::read_from_path(plan.manifest_path()).expect("read manifest");
1112 plan.validate_complete_generation(&manifest)
1113 .expect("complete generation validates");
1114 assert!(manifest
1115 .entries
1116 .iter()
1117 .any(|entry| entry.kind == ServerlessPackKind::ExtentIndex));
1118 assert!(manifest
1119 .entries
1120 .iter()
1121 .any(|entry| entry.kind == ServerlessPackKind::CollectionData));
1122 assert!(manifest
1123 .entries
1124 .iter()
1125 .any(|entry| entry.kind == ServerlessPackKind::SecondaryIndex));
1126
1127 let _ = std::fs::remove_dir_all(root);
1128 }
1129
1130 #[test]
1131 fn verified_current_pointer_rejects_missing_or_corrupt_generation() {
1132 let root = temp_root("serverless-current-verified");
1133 let plan = ServerlessFilePlan::new(&root, "db", 14);
1134 let mut extent_index = ServerlessExtentIndex::new(14);
1135 extent_index.push(
1136 ServerlessExtentRef::new(
1137 "events",
1138 b"a".to_vec(),
1139 b"z".to_vec(),
1140 "collection-data.redpack",
1141 0,
1142 b"collection-bytes",
1143 true,
1144 )
1145 .expect("extent"),
1146 );
1147 let pointer = plan
1148 .publish_core_generation(&extent_index, b"collection-bytes", b"secondary-bytes")
1149 .expect("publish complete generation");
1150 assert_eq!(
1151 plan.read_current_pointer_verified()
1152 .expect("verified pointer before corruption"),
1153 pointer
1154 );
1155
1156 std::fs::remove_file(plan.collection_data_path()).expect("remove required pack");
1157 let err = plan
1158 .read_current_pointer_verified()
1159 .expect_err("verified pointer must reject missing required pack");
1160 assert!(
1165 matches!(&err, RdbFileError::Io(io) if io.kind() == std::io::ErrorKind::NotFound),
1166 "{err}"
1167 );
1168
1169 let _ = std::fs::remove_dir_all(root);
1170 }
1171
1172 #[test]
1173 fn verified_current_pointer_rejects_manifest_hash_mismatch() {
1174 let root = temp_root("serverless-current-manifest-hash");
1175 let plan = ServerlessFilePlan::new(&root, "db", 15);
1176 let mut extent_index = ServerlessExtentIndex::new(15);
1177 extent_index.push(
1178 ServerlessExtentRef::new(
1179 "events",
1180 b"a".to_vec(),
1181 b"z".to_vec(),
1182 "collection-data.redpack",
1183 0,
1184 b"collection-bytes",
1185 true,
1186 )
1187 .expect("extent"),
1188 );
1189 plan.publish_core_generation(&extent_index, b"collection-bytes", b"secondary-bytes")
1190 .expect("publish complete generation");
1191
1192 std::fs::write(plan.manifest_path(), b"corrupt-manifest").expect("corrupt manifest");
1193 let err = plan
1194 .read_current_pointer_verified()
1195 .expect_err("verified pointer must reject manifest hash mismatch");
1196 assert!(
1197 err.to_string().contains("manifest")
1198 && (err.to_string().contains("bytes")
1199 || err.to_string().contains("checksum")
1200 || err.to_string().contains("hash")),
1201 "{err}"
1202 );
1203
1204 let _ = std::fs::remove_dir_all(root);
1205 }
1206
1207 #[test]
1208 fn publish_pointer_rejects_incomplete_or_corrupt_generation() {
1209 let root = temp_root("serverless-publish-rejects");
1210 let plan = ServerlessFilePlan::new(&root, "db", 13);
1211 let mut manifest = ServerlessManifest::new("db", 13);
1212 manifest.push(ServerlessManifestEntry::from_bytes(
1213 ServerlessPackKind::BootIndex,
1214 "boot-index.redpack",
1215 b"boot",
1216 ));
1217 manifest
1218 .write_to_path(plan.manifest_path())
1219 .expect("write incomplete manifest");
1220
1221 let err = plan
1222 .publish_generation_pointer(&manifest)
1223 .expect_err("incomplete generation rejected");
1224 assert!(err.to_string().contains("missing required"), "{err}");
1225 assert!(!plan.current_pointer_path().exists());
1226
1227 let mut extent_index = ServerlessExtentIndex::new(13);
1228 extent_index.push(
1229 ServerlessExtentRef::new(
1230 "events",
1231 b"a".to_vec(),
1232 b"z".to_vec(),
1233 "collection-data.redpack",
1234 0,
1235 b"collection-bytes",
1236 true,
1237 )
1238 .expect("extent"),
1239 );
1240 plan.publish_core_generation(&extent_index, b"collection-bytes", b"secondary-bytes")
1241 .expect("publish complete generation");
1242 std::fs::write(plan.collection_data_path(), b"collection-ByTes")
1243 .expect("corrupt collection pack");
1244 let manifest =
1245 ServerlessManifest::read_from_path(plan.manifest_path()).expect("read manifest");
1246 let err = plan
1247 .publish_generation_pointer(&manifest)
1248 .expect_err("corrupt generation rejected");
1249 assert!(err.to_string().contains("checksum mismatch"), "{err}");
1250
1251 let _ = std::fs::remove_dir_all(root);
1252 }
1253
1254 #[test]
1255 fn incomplete_generation_publish_preserves_existing_current_pointer() {
1256 let root = temp_root("serverless-current-preserved");
1257 let first = ServerlessFilePlan::new(&root, "db", 1);
1258 let mut first_index = ServerlessExtentIndex::new(1);
1259 first_index.push(
1260 ServerlessExtentRef::new(
1261 "events",
1262 b"a".to_vec(),
1263 b"z".to_vec(),
1264 "collection-data.redpack",
1265 0,
1266 b"first",
1267 true,
1268 )
1269 .expect("extent"),
1270 );
1271 let first_pointer = first
1272 .publish_core_generation(&first_index, b"first", b"secondary")
1273 .expect("publish first generation");
1274
1275 let second = ServerlessFilePlan::new(&root, "db", 2);
1276 let mut incomplete = ServerlessManifest::new("db", 2);
1277 incomplete.push(ServerlessManifestEntry::from_bytes(
1278 ServerlessPackKind::BootIndex,
1279 "boot-index.redpack",
1280 b"boot",
1281 ));
1282 incomplete
1283 .write_to_path(second.manifest_path())
1284 .expect("write incomplete manifest");
1285 let err = second
1286 .publish_generation_pointer(&incomplete)
1287 .expect_err("incomplete generation rejected");
1288 assert!(err.to_string().contains("missing required"), "{err}");
1289
1290 assert_eq!(
1291 first.read_current_pointer().expect("read current"),
1292 first_pointer
1293 );
1294
1295 let _ = std::fs::remove_dir_all(root);
1296 }
1297
1298 #[test]
1299 fn file_plan_derives_cache_generation_and_hot_boot_paths() {
1300 let plan = ServerlessFilePlan::for_data_path("/tmp/reddb/main.rdb", 41);
1301 assert_eq!(plan.root, PathBuf::from("/tmp/reddb/main.serverless"));
1302 assert_eq!(plan.namespace, "main");
1303
1304 let next = plan
1305 .for_generation(42)
1306 .with_cache_policy(ServerlessCachePolicy {
1307 keep_boot_index_local: true,
1308 keep_hot_snapshot_local: true,
1309 max_hot_bytes: 4096,
1310 });
1311 assert_eq!(next.root, plan.root);
1312 assert_eq!(next.namespace, plan.namespace);
1313 assert_eq!(next.generation, 42);
1314
1315 let cache = next.local_cache();
1316 assert_eq!(
1317 cache.root,
1318 PathBuf::from("/tmp/reddb/main.serverless/main/cache")
1319 );
1320 assert_eq!(cache.generation, 42);
1321
1322 let hot = ServerlessBootPlan::hot(&next);
1323 assert_eq!(
1324 hot.required_first,
1325 vec![
1326 next.boot_index_path(),
1327 next.hot_snapshot_path(),
1328 next.wal_tail_path(),
1329 ]
1330 );
1331 assert_eq!(
1332 hot.lazy_after_open,
1333 vec![
1334 next.manifest_path(),
1335 next.collection_data_path(),
1336 next.secondary_index_path(),
1337 ]
1338 );
1339 }
1340
1341 #[test]
1342 fn publish_generation_pointer_rejects_manifest_identity_mismatch() {
1343 let root = temp_root("serverless-pointer-identity");
1344 let plan = ServerlessFilePlan::new(&root, "db", 10);
1345
1346 let wrong_namespace = ServerlessManifest::new("other", 10);
1347 let err = plan
1348 .publish_generation_pointer(&wrong_namespace)
1349 .expect_err("namespace mismatch rejected before publish");
1350 assert!(err.to_string().contains("namespace"), "{err}");
1351
1352 let wrong_generation = ServerlessManifest::new("db", 11);
1353 let err = plan
1354 .publish_generation_pointer(&wrong_generation)
1355 .expect_err("generation mismatch rejected before publish");
1356 assert!(err.to_string().contains("generation"), "{err}");
1357
1358 assert!(!plan.current_pointer_path().exists());
1359 let _ = std::fs::remove_dir_all(root);
1360 }
1361
1362 #[test]
1363 fn verified_current_pointer_rejects_pointer_shape_and_manifest_identity() {
1364 let root = temp_root("serverless-current-pointer-shape");
1365 let plan = ServerlessFilePlan::new(&root, "db", 10);
1366 let manifest = ServerlessManifest::new("db", 10);
1367 let manifest_bytes = manifest.encode();
1368
1369 let namespace_mismatch = ServerlessGenerationPointer {
1370 namespace: "other".to_string(),
1371 generation: 10,
1372 manifest_relative_path: PathBuf::from("g00000000000000000010/manifest.redpack"),
1373 manifest_bytes: manifest_bytes.len() as u64,
1374 manifest_checksum: crc32(&manifest_bytes),
1375 manifest_content_hash: ServerlessContentHash::from_bytes(&manifest_bytes),
1376 };
1377 namespace_mismatch
1378 .write_to_path(plan.current_pointer_path())
1379 .expect("write namespace mismatch pointer");
1380 let err = plan
1381 .read_current_pointer_verified()
1382 .expect_err("pointer namespace mismatch rejected");
1383 assert!(err.to_string().contains("namespace"), "{err}");
1384
1385 let path_mismatch = ServerlessGenerationPointer {
1386 namespace: "db".to_string(),
1387 generation: 10,
1388 manifest_relative_path: PathBuf::from("g00000000000000000010/boot-index.redpack"),
1389 manifest_bytes: manifest_bytes.len() as u64,
1390 manifest_checksum: crc32(&manifest_bytes),
1391 manifest_content_hash: ServerlessContentHash::from_bytes(&manifest_bytes),
1392 };
1393 path_mismatch
1394 .write_to_path(plan.current_pointer_path())
1395 .expect("write path mismatch pointer");
1396 let err = plan
1397 .read_current_pointer_verified()
1398 .expect_err("pointer manifest path mismatch rejected");
1399 assert!(err.to_string().contains("manifest path"), "{err}");
1400
1401 let wrong_manifest_namespace = ServerlessManifest::new("other", 10);
1402 wrong_manifest_namespace
1403 .write_to_path(plan.manifest_path())
1404 .expect("write namespace mismatched manifest");
1405 let pointer = ServerlessGenerationPointer::from_manifest(&plan, &wrong_manifest_namespace);
1406 pointer
1407 .write_to_path(plan.current_pointer_path())
1408 .expect("write pointer");
1409 let err = plan
1410 .read_current_pointer_verified()
1411 .expect_err("manifest namespace mismatch rejected");
1412 assert!(err.to_string().contains("manifest namespace"), "{err}");
1413
1414 let wrong_generation_manifest = ServerlessManifest::new("db", 11);
1415 let wrong_generation_bytes = wrong_generation_manifest.encode();
1416 std::fs::write(plan.manifest_path(), &wrong_generation_bytes).expect("write manifest");
1417 let generation_mismatch = ServerlessGenerationPointer {
1418 namespace: "db".to_string(),
1419 generation: 10,
1420 manifest_relative_path: PathBuf::from("g00000000000000000010/manifest.redpack"),
1421 manifest_bytes: wrong_generation_bytes.len() as u64,
1422 manifest_checksum: crc32(&wrong_generation_bytes),
1423 manifest_content_hash: ServerlessContentHash::from_bytes(&wrong_generation_bytes),
1424 };
1425 generation_mismatch
1426 .write_to_path(plan.current_pointer_path())
1427 .expect("write generation mismatch pointer");
1428 let err = plan
1429 .read_current_pointer_verified()
1430 .expect_err("manifest generation mismatch rejected");
1431 assert!(err.to_string().contains("manifest generation"), "{err}");
1432
1433 let _ = std::fs::remove_dir_all(root);
1434 }
1435
1436 #[test]
1437 fn validate_complete_generation_rejects_pack_and_manifest_metadata_mismatch() {
1438 let root = temp_root("serverless-validate-metadata");
1439 let plan = ServerlessFilePlan::new(&root, "db", 16);
1440 let mut extent_index = ServerlessExtentIndex::new(16);
1441 extent_index.push(
1442 ServerlessExtentRef::new(
1443 "events",
1444 b"a".to_vec(),
1445 b"z".to_vec(),
1446 "collection-data.redpack",
1447 0,
1448 b"collection-bytes",
1449 true,
1450 )
1451 .expect("extent"),
1452 );
1453 plan.publish_core_generation(&extent_index, b"collection-bytes", b"secondary-bytes")
1454 .expect("publish complete generation");
1455 let manifest =
1456 ServerlessManifest::read_from_path(plan.manifest_path()).expect("read manifest");
1457
1458 let mut wrong_bytes = manifest.clone();
1459 wrong_bytes.entries[0].bytes += 1;
1460 let err = plan
1461 .validate_complete_generation(&wrong_bytes)
1462 .expect_err("pack byte length mismatch rejected");
1463 assert!(err.to_string().contains("expected"), "{err}");
1464
1465 let mut wrong_hash = manifest.clone();
1466 wrong_hash.entries[0].content_hash = ServerlessContentHash([1; CONTENT_HASH_LEN]);
1467 let err = plan
1468 .validate_complete_generation(&wrong_hash)
1469 .expect_err("pack content hash mismatch rejected");
1470 assert!(err.to_string().contains("content hash mismatch"), "{err}");
1471
1472 let mut manifest_mismatch = manifest.clone();
1473 manifest_mismatch.namespace = "other".to_string();
1474 let err = plan
1475 .validate_complete_generation(&manifest_mismatch)
1476 .expect_err("manifest-on-disk mismatch rejected");
1477 assert!(err.to_string().contains("manifest on disk"), "{err}");
1478
1479 let _ = std::fs::remove_dir_all(root);
1480 }
1481
1482 #[test]
1483 fn writer_lease_json_round_trips_and_preserves_fencing_token() {
1484 let lease = ServerlessWriterLease {
1485 database_key: "main".to_string(),
1486 holder_id: "writer-a".to_string(),
1487 term: 7,
1488 generation: 3,
1489 acquired_at_ms: 100,
1490 expires_at_ms: 200,
1491 };
1492
1493 let decoded = decode_serverless_writer_lease_json(
1494 &encode_serverless_writer_lease_json(&lease).expect("encode lease"),
1495 )
1496 .expect("decode lease");
1497
1498 assert_eq!(decoded, lease);
1499 assert_eq!(decoded.fencing_token(), (7, 3));
1500 assert!(!decoded.is_expired(199));
1501 assert!(decoded.is_expired(200));
1502 assert!(decoded.fenced_by_term(8));
1503 }
1504
1505 #[test]
1506 fn writer_lease_json_decodes_legacy_missing_term_as_base_term() {
1507 let decoded = decode_serverless_writer_lease_json(
1508 br#"{
1509 "database_key": "main",
1510 "holder_id": "writer-a",
1511 "generation": 3,
1512 "acquired_at_ms": 100,
1513 "expires_at_ms": 200
1514 }"#,
1515 )
1516 .expect("decode legacy lease");
1517
1518 assert_eq!(decoded.term, SERVERLESS_WRITER_LEASE_DEFAULT_TERM);
1519 }
1520
1521 #[test]
1522 fn writer_lease_artifact_names_are_deterministic() {
1523 assert_eq!(
1524 serverless_writer_lease_key("leases/", "main"),
1525 "leases/main.lease.json"
1526 );
1527 assert_eq!(
1528 serverless_writer_lease_temp_path("write", 10, 20, 30)
1529 .file_name()
1530 .and_then(|name| name.to_str()),
1531 Some("reddb-lease-write-10-20-30.json")
1532 );
1533 }
1534
1535 #[test]
1536 fn writer_lease_temp_file_round_trips_and_cleans_up() {
1537 let temp = ServerlessWriterLeaseTempFile::with_clock("write", 10, 20, 30);
1538 assert_eq!(
1539 temp.path().file_name().and_then(|name| name.to_str()),
1540 Some("reddb-lease-write-10-20-30.json")
1541 );
1542
1543 temp.write_bytes(b"{\"lease\":true}")
1544 .expect("write lease temp");
1545 assert_eq!(
1546 temp.read_bytes().expect("read lease temp"),
1547 b"{\"lease\":true}"
1548 );
1549 temp.cleanup().expect("cleanup lease temp");
1550 assert!(!temp.path().exists());
1551 }
1552
1553 fn temp_root(label: &str) -> PathBuf {
1554 std::env::temp_dir().join(format!(
1555 "reddb-file-{label}-{}-{}",
1556 std::process::id(),
1557 std::time::SystemTime::now()
1558 .duration_since(std::time::UNIX_EPOCH)
1559 .unwrap()
1560 .as_nanos()
1561 ))
1562 }
1563}