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