1#[cfg(all(test, feature = "arbitrary"))]
213mod conformance;
214mod storage;
215use commonware_runtime::buffer::paged::CacheRef;
216use commonware_utils::Array;
217use std::num::NonZeroUsize;
218pub use storage::{Checkpoint, Cursor, Freezer};
219use thiserror::Error;
220
221pub enum Identifier<'a, K: Array> {
223 Cursor(Cursor),
224 Key(&'a K),
225}
226
227#[derive(Debug, Error)]
229pub enum Error {
230 #[error("runtime error: {0}")]
231 Runtime(#[from] commonware_runtime::Error),
232 #[error("journal error: {0}")]
233 Journal(#[from] crate::journal::Error),
234 #[error("codec error: {0}")]
235 Codec(#[from] commonware_codec::Error),
236 #[error("checkpoint does not match stored data")]
237 CheckpointMismatch,
238}
239
240#[derive(Clone)]
242pub struct Config<C> {
243 pub key_partition: String,
245
246 pub key_write_buffer: NonZeroUsize,
248
249 pub key_page_cache: CacheRef,
251
252 pub value_partition: String,
254
255 pub value_compression: Option<u8>,
257
258 pub value_write_buffer: NonZeroUsize,
260
261 pub value_target_size: u64,
263
264 pub table_partition: String,
266
267 pub table_initial_size: u32,
269
270 pub table_resize_frequency: u8,
273
274 pub table_resize_chunk_size: u32,
276
277 pub table_replay_buffer: NonZeroUsize,
279
280 pub codec_config: C,
282}
283
284#[cfg(test)]
285mod tests {
286 use super::*;
287 use commonware_codec::DecodeExt;
288 use commonware_formatting::hex;
289 use commonware_macros::{test_group, test_traced};
290 use commonware_runtime::{deterministic, Blob, Metrics as _, Runner, Storage, Supervisor as _};
291 use commonware_utils::{sequence::FixedBytes, NZUsize, NZU16};
292 use rand::{Rng, RngExt as _};
293 use std::num::NonZeroU16;
294
295 fn test_key(key: &str) -> FixedBytes<64> {
296 let mut buf = [0u8; 64];
297 let key = key.as_bytes();
298 assert!(key.len() <= buf.len());
299 buf[..key.len()].copy_from_slice(key);
300 FixedBytes::decode(buf.as_ref()).unwrap()
301 }
302
303 const DEFAULT_WRITE_BUFFER: usize = 1024;
304 const DEFAULT_VALUE_TARGET_SIZE: u64 = 10 * 1024 * 1024;
305 const DEFAULT_TABLE_INITIAL_SIZE: u32 = 256;
306 const DEFAULT_TABLE_RESIZE_FREQUENCY: u8 = 4;
307 const DEFAULT_TABLE_RESIZE_CHUNK_SIZE: u32 = 128; const DEFAULT_TABLE_REPLAY_BUFFER: usize = 64 * 1024; const PAGE_SIZE: NonZeroU16 = NZU16!(1024);
310 const PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(10);
311
312 fn test_put_get(compression: Option<u8>) {
313 let executor = deterministic::Runner::default();
315 executor.start(|context| async move {
316 let cfg = Config {
318 key_partition: "test-key-index".into(),
319 key_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
320 key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
321 value_partition: "test-value-journal".into(),
322 value_compression: compression,
323 value_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
324 value_target_size: DEFAULT_VALUE_TARGET_SIZE,
325 table_partition: "test-table".into(),
326 table_initial_size: DEFAULT_TABLE_INITIAL_SIZE,
327 table_resize_frequency: DEFAULT_TABLE_RESIZE_FREQUENCY,
328 table_resize_chunk_size: DEFAULT_TABLE_RESIZE_CHUNK_SIZE,
329 table_replay_buffer: NZUsize!(DEFAULT_TABLE_REPLAY_BUFFER),
330 codec_config: (),
331 };
332 let mut freezer = Freezer::<_, FixedBytes<64>, i32>::init(
333 context.child("storage"),
334 cfg.clone(),
335 None,
336 )
337 .await
338 .expect("Failed to initialize freezer");
339
340 let key = test_key("testkey");
341 let data = 42;
342
343 let value = freezer
345 .get(Identifier::Key(&key))
346 .await
347 .expect("Failed to check key");
348 assert!(value.is_none());
349
350 freezer
352 .put(key.clone(), data)
353 .await
354 .expect("Failed to put data");
355
356 let value = freezer
358 .get(Identifier::Key(&key))
359 .await
360 .expect("Failed to get data")
361 .expect("Data not found");
362 assert_eq!(value, data);
363
364 let buffer = context.encode();
366 assert!(buffer.contains("gets_total 2"), "{}", buffer);
367 assert!(buffer.contains("puts_total 1"), "{}", buffer);
368 assert!(buffer.contains("unnecessary_reads_total 0"), "{}", buffer);
369
370 freezer.sync().await.expect("Failed to sync data");
372 });
373 }
374
375 #[test_traced]
376 fn test_put_get_no_compression() {
377 test_put_get(None);
378 }
379
380 #[test_traced]
381 fn test_put_get_compression() {
382 test_put_get(Some(3));
383 }
384
385 #[test_traced]
386 fn test_has() {
387 let executor = deterministic::Runner::default();
389 executor.start(|context| async move {
390 let cfg = Config {
392 key_partition: "test-key-index".into(),
393 key_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
394 key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
395 value_partition: "test-value-journal".into(),
396 value_compression: None,
397 value_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
398 value_target_size: DEFAULT_VALUE_TARGET_SIZE,
399 table_partition: "test-table".into(),
400 table_initial_size: DEFAULT_TABLE_INITIAL_SIZE,
401 table_resize_frequency: DEFAULT_TABLE_RESIZE_FREQUENCY,
402 table_resize_chunk_size: DEFAULT_TABLE_RESIZE_CHUNK_SIZE,
403 table_replay_buffer: NZUsize!(DEFAULT_TABLE_REPLAY_BUFFER),
404 codec_config: (),
405 };
406 let mut freezer =
407 Freezer::<_, FixedBytes<64>, i32>::init(context.child("storage"), cfg, None)
408 .await
409 .expect("Failed to initialize freezer");
410
411 let key = test_key("testkey");
413 assert!(!freezer.has(&key).await.expect("Failed to check key"));
414
415 freezer
417 .put(key.clone(), 42)
418 .await
419 .expect("Failed to put data");
420 assert!(freezer.has(&key).await.expect("Failed to check key"));
421
422 assert!(!freezer
424 .has(&test_key("otherkey"))
425 .await
426 .expect("Failed to check key"));
427
428 let buffer = context.encode();
430 assert!(buffer.contains("has_total 3"), "{}", buffer);
431 assert!(buffer.contains("gets_total 0"), "{}", buffer);
432 });
433 }
434
435 #[test_traced]
436 fn test_multiple_keys() {
437 let executor = deterministic::Runner::default();
439 executor.start(|context| async move {
440 let cfg = Config {
442 key_partition: "test-key-index".into(),
443 key_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
444 key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
445 value_partition: "test-value-journal".into(),
446 value_compression: None,
447 value_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
448 value_target_size: DEFAULT_VALUE_TARGET_SIZE,
449 table_partition: "test-table".into(),
450 table_initial_size: DEFAULT_TABLE_INITIAL_SIZE,
451 table_resize_frequency: DEFAULT_TABLE_RESIZE_FREQUENCY,
452 table_resize_chunk_size: DEFAULT_TABLE_RESIZE_CHUNK_SIZE,
453 table_replay_buffer: NZUsize!(DEFAULT_TABLE_REPLAY_BUFFER),
454 codec_config: (),
455 };
456 let mut freezer = Freezer::<_, FixedBytes<64>, i32>::init(
457 context.child("storage"),
458 cfg.clone(),
459 None,
460 )
461 .await
462 .expect("Failed to initialize freezer");
463
464 let keys = vec![
466 (test_key("key1"), 1),
467 (test_key("key2"), 2),
468 (test_key("key3"), 3),
469 (test_key("key4"), 4),
470 (test_key("key5"), 5),
471 ];
472
473 for (key, data) in &keys {
474 freezer
475 .put(key.clone(), *data)
476 .await
477 .expect("Failed to put data");
478 }
479
480 for (key, data) in &keys {
482 let retrieved = freezer
483 .get(Identifier::Key(key))
484 .await
485 .expect("Failed to get data")
486 .expect("Data not found");
487 assert_eq!(retrieved, *data);
488 }
489 });
490 }
491
492 #[test_traced]
493 fn test_collision_handling() {
494 let executor = deterministic::Runner::default();
496 executor.start(|context| async move {
497 let cfg = Config {
499 key_partition: "test-key-index".into(),
500 key_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
501 key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
502 value_partition: "test-value-journal".into(),
503 value_compression: None,
504 value_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
505 value_target_size: DEFAULT_VALUE_TARGET_SIZE,
506 table_partition: "test-table".into(),
507 table_initial_size: 4, table_resize_frequency: DEFAULT_TABLE_RESIZE_FREQUENCY,
509 table_resize_chunk_size: DEFAULT_TABLE_RESIZE_CHUNK_SIZE,
510 table_replay_buffer: NZUsize!(DEFAULT_TABLE_REPLAY_BUFFER),
511 codec_config: (),
512 };
513 let mut freezer = Freezer::<_, FixedBytes<64>, i32>::init(
514 context.child("storage"),
515 cfg.clone(),
516 None,
517 )
518 .await
519 .expect("Failed to initialize freezer");
520
521 let keys = vec![
523 (test_key("key1"), 1),
524 (test_key("key2"), 2),
525 (test_key("key3"), 3),
526 (test_key("key4"), 4),
527 (test_key("key5"), 5),
528 (test_key("key6"), 6),
529 (test_key("key7"), 7),
530 (test_key("key8"), 8),
531 ];
532
533 for (key, data) in &keys {
534 freezer
535 .put(key.clone(), *data)
536 .await
537 .expect("Failed to put data");
538 }
539
540 freezer.sync().await.expect("Failed to sync");
542
543 for (key, data) in &keys {
545 let retrieved = freezer
546 .get(Identifier::Key(key))
547 .await
548 .expect("Failed to get data")
549 .expect("Data not found");
550 assert_eq!(retrieved, *data);
551 }
552
553 let buffer = context.encode();
555 assert!(buffer.contains("gets_total 8"), "{}", buffer);
556 assert!(buffer.contains("unnecessary_reads_total 5"), "{}", buffer);
557 });
558 }
559
560 #[test_traced]
561 fn test_restart() {
562 let executor = deterministic::Runner::default();
564 executor.start(|context| async move {
565 let cfg = Config {
566 key_partition: "test-key-index".into(),
567 key_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
568 key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
569 value_partition: "test-value-journal".into(),
570 value_compression: None,
571 value_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
572 value_target_size: DEFAULT_VALUE_TARGET_SIZE,
573 table_partition: "test-table".into(),
574 table_initial_size: DEFAULT_TABLE_INITIAL_SIZE,
575 table_resize_frequency: DEFAULT_TABLE_RESIZE_FREQUENCY,
576 table_resize_chunk_size: DEFAULT_TABLE_RESIZE_CHUNK_SIZE,
577 table_replay_buffer: NZUsize!(DEFAULT_TABLE_REPLAY_BUFFER),
578 codec_config: (),
579 };
580
581 let checkpoint = {
583 let mut freezer = Freezer::<_, FixedBytes<64>, i32>::init(
584 context.child("first"),
585 cfg.clone(),
586 None,
587 )
588 .await
589 .expect("Failed to initialize freezer");
590
591 let keys = vec![
592 (test_key("persist1"), 100),
593 (test_key("persist2"), 200),
594 (test_key("persist3"), 300),
595 ];
596
597 for (key, data) in &keys {
598 freezer
599 .put(key.clone(), *data)
600 .await
601 .expect("Failed to put data");
602 }
603
604 freezer.close().await.expect("Failed to close freezer")
605 };
606
607 {
609 let freezer = Freezer::<_, FixedBytes<64>, i32>::init(
610 context.child("second"),
611 cfg.clone(),
612 Some(checkpoint),
613 )
614 .await
615 .expect("Failed to initialize freezer");
616
617 let keys = vec![
618 (test_key("persist1"), 100),
619 (test_key("persist2"), 200),
620 (test_key("persist3"), 300),
621 ];
622
623 for (key, data) in &keys {
624 let retrieved = freezer
625 .get(Identifier::Key(key))
626 .await
627 .expect("Failed to get data")
628 .expect("Data not found");
629 assert_eq!(retrieved, *data);
630 }
631 }
632 });
633 }
634
635 #[test_traced]
636 fn test_crash_consistency() {
637 let executor = deterministic::Runner::default();
639 executor.start(|context| async move {
640 let cfg = Config {
641 key_partition: "test-key-index".into(),
642 key_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
643 key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
644 value_partition: "test-value-journal".into(),
645 value_compression: None,
646 value_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
647 value_target_size: DEFAULT_VALUE_TARGET_SIZE,
648 table_partition: "test-table".into(),
649 table_initial_size: DEFAULT_TABLE_INITIAL_SIZE,
650 table_resize_frequency: DEFAULT_TABLE_RESIZE_FREQUENCY,
651 table_resize_chunk_size: DEFAULT_TABLE_RESIZE_CHUNK_SIZE,
652 table_replay_buffer: NZUsize!(DEFAULT_TABLE_REPLAY_BUFFER),
653 codec_config: (),
654 };
655
656 let checkpoint = {
658 let mut freezer = Freezer::<_, FixedBytes<64>, i32>::init(
659 context.child("first"),
660 cfg.clone(),
661 None,
662 )
663 .await
664 .expect("Failed to initialize freezer");
665
666 freezer
667 .put(test_key("committed1"), 1)
668 .await
669 .expect("Failed to put data");
670 freezer
671 .put(test_key("committed2"), 2)
672 .await
673 .expect("Failed to put data");
674
675 freezer.sync().await.expect("Failed to sync");
677
678 freezer
680 .put(test_key("uncommitted1"), 3)
681 .await
682 .expect("Failed to put data");
683 freezer
684 .put(test_key("uncommitted2"), 4)
685 .await
686 .expect("Failed to put data");
687
688 freezer.close().await.expect("Failed to close")
690 };
691
692 {
694 let freezer = Freezer::<_, FixedBytes<64>, i32>::init(
695 context.child("second"),
696 cfg.clone(),
697 Some(checkpoint),
698 )
699 .await
700 .expect("Failed to initialize freezer");
701
702 assert_eq!(
704 freezer
705 .get(Identifier::Key(&test_key("committed1")))
706 .await
707 .unwrap(),
708 Some(1)
709 );
710 assert_eq!(
711 freezer
712 .get(Identifier::Key(&test_key("committed2")))
713 .await
714 .unwrap(),
715 Some(2)
716 );
717
718 if let Some(val) = freezer
721 .get(Identifier::Key(&test_key("uncommitted1")))
722 .await
723 .unwrap()
724 {
725 assert_eq!(val, 3);
726 }
727 if let Some(val) = freezer
728 .get(Identifier::Key(&test_key("uncommitted2")))
729 .await
730 .unwrap()
731 {
732 assert_eq!(val, 4);
733 }
734 }
735 });
736 }
737
738 #[test_traced]
739 fn test_destroy() {
740 let executor = deterministic::Runner::default();
742 executor.start(|context| async move {
743 let cfg = Config {
745 key_partition: "test-key-index".into(),
746 key_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
747 key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
748 value_partition: "test-value-journal".into(),
749 value_compression: None,
750 value_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
751 value_target_size: DEFAULT_VALUE_TARGET_SIZE,
752 table_partition: "test-table".into(),
753 table_initial_size: DEFAULT_TABLE_INITIAL_SIZE,
754 table_resize_frequency: DEFAULT_TABLE_RESIZE_FREQUENCY,
755 table_resize_chunk_size: DEFAULT_TABLE_RESIZE_CHUNK_SIZE,
756 table_replay_buffer: NZUsize!(DEFAULT_TABLE_REPLAY_BUFFER),
757 codec_config: (),
758 };
759 {
760 let mut freezer = Freezer::<_, FixedBytes<64>, i32>::init(
761 context.child("first"),
762 cfg.clone(),
763 None,
764 )
765 .await
766 .expect("Failed to initialize freezer");
767
768 freezer
769 .put(test_key("destroy1"), 1)
770 .await
771 .expect("Failed to put data");
772 freezer
773 .put(test_key("destroy2"), 2)
774 .await
775 .expect("Failed to put data");
776
777 freezer.destroy().await.expect("Failed to destroy freezer");
779 }
780
781 {
783 let freezer = Freezer::<_, FixedBytes<64>, i32>::init(
784 context.child("second"),
785 cfg.clone(),
786 None,
787 )
788 .await
789 .expect("Failed to initialize freezer");
790
791 assert!(freezer
793 .get(Identifier::Key(&test_key("destroy1")))
794 .await
795 .unwrap()
796 .is_none());
797 assert!(freezer
798 .get(Identifier::Key(&test_key("destroy2")))
799 .await
800 .unwrap()
801 .is_none());
802 }
803 });
804 }
805
806 #[test_traced]
807 fn test_partial_table_entry_write() {
808 let executor = deterministic::Runner::default();
810 executor.start(|context| async move {
811 let cfg = Config {
813 key_partition: "test-key-index".into(),
814 key_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
815 key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
816 value_partition: "test-value-journal".into(),
817 value_compression: None,
818 value_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
819 value_target_size: DEFAULT_VALUE_TARGET_SIZE,
820 table_partition: "test-table".into(),
821 table_initial_size: DEFAULT_TABLE_INITIAL_SIZE,
822 table_resize_frequency: DEFAULT_TABLE_RESIZE_FREQUENCY,
823 table_resize_chunk_size: DEFAULT_TABLE_RESIZE_CHUNK_SIZE,
824 table_replay_buffer: NZUsize!(DEFAULT_TABLE_REPLAY_BUFFER),
825 codec_config: (),
826 };
827 let checkpoint = {
828 let mut freezer = Freezer::<_, FixedBytes<64>, i32>::init(
829 context.child("first"),
830 cfg.clone(),
831 None,
832 )
833 .await
834 .expect("Failed to initialize freezer");
835
836 freezer.put(test_key("key1"), 42).await.unwrap();
837 freezer.sync().await.unwrap();
838 freezer.close().await.unwrap()
839 };
840
841 {
843 let (blob, _) = context.open(&cfg.table_partition, b"table").await.unwrap();
844 blob.write_at_sync(0, vec![0xFF; 10]).await.unwrap();
846 }
847
848 {
850 let freezer = Freezer::<_, FixedBytes<64>, i32>::init(
851 context.child("second"),
852 cfg.clone(),
853 Some(checkpoint),
854 )
855 .await
856 .expect("Failed to initialize freezer");
857
858 let result = freezer
861 .get(Identifier::Key(&test_key("key1")))
862 .await
863 .unwrap();
864 assert!(result.is_none() || result == Some(42));
865 }
866 });
867 }
868
869 #[test_traced]
870 fn test_table_entry_invalid_crc() {
871 let executor = deterministic::Runner::default();
873 executor.start(|context| async move {
874 let cfg = Config {
875 key_partition: "test-key-index".into(),
876 key_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
877 key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
878 value_partition: "test-value-journal".into(),
879 value_compression: None,
880 value_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
881 value_target_size: DEFAULT_VALUE_TARGET_SIZE,
882 table_partition: "test-table".into(),
883 table_initial_size: DEFAULT_TABLE_INITIAL_SIZE,
884 table_resize_frequency: DEFAULT_TABLE_RESIZE_FREQUENCY,
885 table_resize_chunk_size: DEFAULT_TABLE_RESIZE_CHUNK_SIZE,
886 table_replay_buffer: NZUsize!(DEFAULT_TABLE_REPLAY_BUFFER),
887 codec_config: (),
888 };
889
890 let checkpoint = {
892 let mut freezer = Freezer::<_, FixedBytes<64>, i32>::init(
893 context.child("first"),
894 cfg.clone(),
895 None,
896 )
897 .await
898 .expect("Failed to initialize freezer");
899
900 freezer.put(test_key("key1"), 42).await.unwrap();
901 freezer.sync().await.unwrap();
902 freezer.close().await.unwrap()
903 };
904
905 {
907 let (blob, _) = context.open(&cfg.table_partition, b"table").await.unwrap();
908 let entry_data = blob.read_at(0, 24).await.unwrap();
910 let mut corrupted = entry_data.coalesce();
911 corrupted.as_mut()[20] ^= 0xFF;
913 blob.write_at_sync(0, corrupted).await.unwrap();
914 }
915
916 {
918 let freezer = Freezer::<_, FixedBytes<64>, i32>::init(
919 context.child("second"),
920 cfg.clone(),
921 Some(checkpoint),
922 )
923 .await
924 .expect("Failed to initialize freezer");
925
926 let result = freezer
928 .get(Identifier::Key(&test_key("key1")))
929 .await
930 .unwrap();
931 assert!(result.is_none() || result == Some(42));
933 }
934 });
935 }
936
937 #[test_traced]
938 fn test_table_extra_bytes() {
939 let executor = deterministic::Runner::default();
941 executor.start(|context| async move {
942 let cfg = Config {
943 key_partition: "test-key-index".into(),
944 key_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
945 key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
946 value_partition: "test-value-journal".into(),
947 value_compression: None,
948 value_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
949 value_target_size: DEFAULT_VALUE_TARGET_SIZE,
950 table_partition: "test-table".into(),
951 table_initial_size: DEFAULT_TABLE_INITIAL_SIZE,
952 table_resize_frequency: DEFAULT_TABLE_RESIZE_FREQUENCY,
953 table_resize_chunk_size: DEFAULT_TABLE_RESIZE_CHUNK_SIZE,
954 table_replay_buffer: NZUsize!(DEFAULT_TABLE_REPLAY_BUFFER),
955 codec_config: (),
956 };
957
958 let checkpoint = {
960 let mut freezer = Freezer::<_, FixedBytes<64>, i32>::init(
961 context.child("first"),
962 cfg.clone(),
963 None,
964 )
965 .await
966 .expect("Failed to initialize freezer");
967
968 freezer.put(test_key("key1"), 42).await.unwrap();
969 freezer.sync().await.unwrap();
970 freezer.close().await.unwrap()
971 };
972
973 {
975 let (blob, size) = context.open(&cfg.table_partition, b"table").await.unwrap();
976 blob.write_at_sync(size, hex!("0xdeadbeef").to_vec())
978 .await
979 .unwrap();
980 }
981
982 {
984 let freezer = Freezer::<_, FixedBytes<64>, i32>::init(
985 context.child("second"),
986 cfg.clone(),
987 Some(checkpoint),
988 )
989 .await
990 .expect("Failed to initialize freezer");
991
992 assert_eq!(
994 freezer
995 .get(Identifier::Key(&test_key("key1")))
996 .await
997 .unwrap(),
998 Some(42)
999 );
1000
1001 let mut freezer_mut = freezer;
1003 freezer_mut.put(test_key("key2"), 43).await.unwrap();
1004 assert_eq!(
1005 freezer_mut
1006 .get(Identifier::Key(&test_key("key2")))
1007 .await
1008 .unwrap(),
1009 Some(43)
1010 );
1011 }
1012 });
1013 }
1014
1015 #[test_traced]
1016 fn test_indexing_across_resizes() {
1017 let executor = deterministic::Runner::default();
1019 executor.start(|context| async move {
1020 let cfg = Config {
1022 key_partition: "test-key-index".into(),
1023 key_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
1024 key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
1025 value_partition: "test-value-journal".into(),
1026 value_compression: None,
1027 value_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
1028 value_target_size: DEFAULT_VALUE_TARGET_SIZE,
1029 table_partition: "test-table".into(),
1030 table_initial_size: 2, table_resize_frequency: 2, table_resize_chunk_size: DEFAULT_TABLE_RESIZE_CHUNK_SIZE,
1033 table_replay_buffer: NZUsize!(DEFAULT_TABLE_REPLAY_BUFFER),
1034 codec_config: (),
1035 };
1036 let mut freezer =
1037 Freezer::<_, FixedBytes<64>, i32>::init(context.child("first"), cfg.clone(), None)
1038 .await
1039 .expect("Failed to initialize freezer");
1040
1041 let mut keys = Vec::new();
1044 for i in 0..1000 {
1045 let key = test_key(&format!("key{i}"));
1046 keys.push((key.clone(), i));
1047
1048 freezer.put(key, i).await.expect("Failed to put data");
1050 freezer.sync().await.expect("Failed to sync");
1051 }
1052
1053 for (key, value) in &keys {
1055 let retrieved = freezer
1056 .get(Identifier::Key(key))
1057 .await
1058 .expect("Failed to get data")
1059 .expect("Data not found");
1060 assert_eq!(retrieved, *value, "Value mismatch for key after resizes");
1061 }
1062
1063 let buffer = context.encode();
1067 assert!(buffer.contains("first_resizes_total 8"), "{}", buffer);
1068
1069 let checkpoint = freezer.close().await.expect("Failed to close");
1071 let freezer = Freezer::<_, FixedBytes<64>, i32>::init(
1072 context.child("second"),
1073 cfg.clone(),
1074 Some(checkpoint),
1075 )
1076 .await
1077 .expect("Failed to reinitialize freezer");
1078
1079 for (key, value) in &keys {
1081 let retrieved = freezer
1082 .get(Identifier::Key(key))
1083 .await
1084 .expect("Failed to get data")
1085 .expect("Data not found");
1086 assert_eq!(retrieved, *value, "Value mismatch for key after restart");
1087 }
1088 });
1089 }
1090
1091 #[test_traced]
1092 fn test_insert_during_resize() {
1093 let executor = deterministic::Runner::default();
1094 executor.start(|context| async move {
1095 let cfg = Config {
1096 key_partition: "test-key-index".into(),
1097 key_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
1098 key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
1099 value_partition: "test-value-journal".into(),
1100 value_compression: None,
1101 value_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
1102 value_target_size: DEFAULT_VALUE_TARGET_SIZE,
1103 table_partition: "test-table".into(),
1104 table_initial_size: 2,
1105 table_resize_frequency: 1,
1106 table_resize_chunk_size: 1, table_replay_buffer: NZUsize!(DEFAULT_TABLE_REPLAY_BUFFER),
1108 codec_config: (),
1109 };
1110 let mut freezer = Freezer::<_, FixedBytes<64>, i32>::init(
1111 context.child("storage"),
1112 cfg.clone(),
1113 None,
1114 )
1115 .await
1116 .unwrap();
1117
1118 freezer.put(test_key("key0"), 0).await.unwrap();
1121 freezer.put(test_key("key2"), 1).await.unwrap();
1122 freezer.sync().await.unwrap(); assert!(freezer.resizing().is_some());
1126
1127 freezer.put(test_key("key6"), 2).await.unwrap();
1130 assert!(context.encode().contains("unnecessary_writes_total 1"));
1131 assert_eq!(freezer.resizable(), 3);
1132
1133 freezer.put(test_key("key3"), 3).await.unwrap();
1136 assert!(context.encode().contains("unnecessary_writes_total 1"));
1137 assert_eq!(freezer.resizable(), 3);
1138
1139 freezer.sync().await.unwrap();
1141 assert!(freezer.resizing().is_none());
1142 assert_eq!(freezer.resizable(), 2);
1143
1144 freezer.put(test_key("key4"), 4).await.unwrap();
1147 freezer.put(test_key("key7"), 5).await.unwrap();
1148 freezer.sync().await.unwrap();
1149
1150 assert!(freezer.resizing().is_some());
1152
1153 let keys = ["key0", "key2", "key6", "key3", "key4", "key7"];
1155 for (i, k) in keys.iter().enumerate() {
1156 assert_eq!(
1157 freezer.get(Identifier::Key(&test_key(k))).await.unwrap(),
1158 Some(i as i32)
1159 );
1160 }
1161
1162 while freezer.resizing().is_some() {
1164 freezer.sync().await.unwrap();
1165 }
1166
1167 assert_eq!(freezer.resizable(), 0);
1169 });
1170 }
1171
1172 #[test_traced]
1173 fn test_resize_after_startup() {
1174 let executor = deterministic::Runner::default();
1175 executor.start(|context| async move {
1176 let cfg = Config {
1177 key_partition: "test-key-index".into(),
1178 key_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
1179 key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
1180 value_partition: "test-value-journal".into(),
1181 value_compression: None,
1182 value_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
1183 value_target_size: DEFAULT_VALUE_TARGET_SIZE,
1184 table_partition: "test-table".into(),
1185 table_initial_size: 2,
1186 table_resize_frequency: 1,
1187 table_resize_chunk_size: 1, table_replay_buffer: NZUsize!(DEFAULT_TABLE_REPLAY_BUFFER),
1189 codec_config: (),
1190 };
1191
1192 let checkpoint = {
1194 let mut freezer = Freezer::<_, FixedBytes<64>, i32>::init(
1195 context.child("first"),
1196 cfg.clone(),
1197 None,
1198 )
1199 .await
1200 .unwrap();
1201
1202 freezer.put(test_key("key0"), 0).await.unwrap();
1205 freezer.put(test_key("key2"), 1).await.unwrap();
1206 let checkpoint = freezer.sync().await.unwrap();
1207
1208 assert!(freezer.resizing().is_some());
1210
1211 checkpoint
1212 };
1213
1214 let mut freezer = Freezer::<_, FixedBytes<64>, i32>::init(
1216 context.child("second"),
1217 cfg.clone(),
1218 Some(checkpoint),
1219 )
1220 .await
1221 .unwrap();
1222 assert_eq!(freezer.resizable(), 1);
1223 assert_eq!(freezer.resizing(), None);
1224
1225 freezer.sync().await.unwrap();
1227 assert_eq!(freezer.resizing(), Some(1));
1228
1229 while freezer.resizing().is_some() {
1231 freezer.sync().await.unwrap();
1232 }
1233
1234 assert_eq!(freezer.resizable(), 0);
1236 });
1237 }
1238
1239 fn test_operations_and_restart(num_keys: usize) -> String {
1240 let executor = deterministic::Runner::default();
1241 executor.start(|mut context| async move {
1242 let cfg = Config {
1243 key_partition: "test-key-index".into(),
1244 key_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
1245 key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
1246 value_partition: "test-value-journal".into(),
1247 value_compression: None,
1248 value_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
1249 value_target_size: 128, table_partition: "test-table".into(),
1251 table_initial_size: 8, table_resize_frequency: 2, table_resize_chunk_size: DEFAULT_TABLE_RESIZE_CHUNK_SIZE,
1254 table_replay_buffer: NZUsize!(DEFAULT_TABLE_REPLAY_BUFFER),
1255 codec_config: (),
1256 };
1257 let mut freezer = Freezer::<_, FixedBytes<96>, FixedBytes<256>>::init(
1258 context.child("init").with_attribute("index", 1),
1259 cfg.clone(),
1260 None,
1261 )
1262 .await
1263 .expect("Failed to initialize freezer");
1264
1265 let mut pairs = Vec::new();
1267
1268 for _ in 0..num_keys {
1269 let mut key = [0u8; 96];
1271 context.fill_bytes(&mut key);
1272 let key = FixedBytes::<96>::new(key);
1273
1274 let mut value = [0u8; 256];
1276 context.fill_bytes(&mut value);
1277 let value = FixedBytes::<256>::new(value);
1278
1279 freezer
1281 .put(key.clone(), value.clone())
1282 .await
1283 .expect("Failed to put data");
1284 pairs.push((key, value));
1285
1286 if context.random_bool(0.1) {
1288 freezer.sync().await.expect("Failed to sync");
1289 }
1290 }
1291
1292 freezer.sync().await.expect("Failed to sync");
1294
1295 for (key, value) in &pairs {
1297 let retrieved = freezer
1298 .get(Identifier::Key(key))
1299 .await
1300 .expect("Failed to get data")
1301 .expect("Data not found");
1302 assert_eq!(&retrieved, value);
1303 }
1304
1305 for (key, _) in &pairs {
1307 assert!(freezer
1308 .get(Identifier::Key(key))
1309 .await
1310 .expect("Failed to check key")
1311 .is_some());
1312 }
1313
1314 for _ in 0..10 {
1316 let mut key = [0u8; 96];
1317 context.fill_bytes(&mut key);
1318 let key = FixedBytes::<96>::new(key);
1319 assert!(freezer
1320 .get(Identifier::Key(&key))
1321 .await
1322 .expect("Failed to check key")
1323 .is_none());
1324 }
1325
1326 let checkpoint = freezer.close().await.expect("Failed to close freezer");
1328
1329 let mut freezer = Freezer::<_, FixedBytes<96>, FixedBytes<256>>::init(
1331 context.child("init").with_attribute("index", 2),
1332 cfg.clone(),
1333 Some(checkpoint),
1334 )
1335 .await
1336 .expect("Failed to initialize freezer");
1337
1338 for (key, value) in &pairs {
1340 let retrieved = freezer
1341 .get(Identifier::Key(key))
1342 .await
1343 .expect("Failed to get data")
1344 .expect("Data not found");
1345 assert_eq!(&retrieved, value);
1346 }
1347
1348 for _ in 0..20 {
1350 let mut key = [0u8; 96];
1351 context.fill_bytes(&mut key);
1352 let key = FixedBytes::<96>::new(key);
1353
1354 let mut value = [0u8; 256];
1355 context.fill_bytes(&mut value);
1356 let value = FixedBytes::<256>::new(value);
1357
1358 freezer.put(key, value).await.expect("Failed to put data");
1359 }
1360
1361 for _ in 0..3 {
1363 freezer.sync().await.expect("Failed to sync");
1364
1365 for _ in 0..5 {
1367 let mut key = [0u8; 96];
1368 context.fill_bytes(&mut key);
1369 let key = FixedBytes::<96>::new(key);
1370
1371 let mut value = [0u8; 256];
1372 context.fill_bytes(&mut value);
1373 let value = FixedBytes::<256>::new(value);
1374
1375 freezer.put(key, value).await.expect("Failed to put data");
1376 }
1377 }
1378
1379 freezer.sync().await.expect("Failed to sync");
1381
1382 context.auditor().state()
1384 })
1385 }
1386
1387 #[test_group("slow")]
1388 #[test_traced]
1389 fn test_determinism() {
1390 let state1 = test_operations_and_restart(1_000);
1391 let state2 = test_operations_and_restart(1_000);
1392 assert_eq!(state1, state2);
1393 }
1394
1395 #[test_traced]
1396 fn test_put_multiple_updates() {
1397 let executor = deterministic::Runner::default();
1399 executor.start(|context| async move {
1400 let cfg = Config {
1402 key_partition: "test-key-index".into(),
1403 key_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
1404 key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
1405 value_partition: "test-value-journal".into(),
1406 value_compression: None,
1407 value_write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
1408 value_target_size: DEFAULT_VALUE_TARGET_SIZE,
1409 table_partition: "test-table".into(),
1410 table_initial_size: DEFAULT_TABLE_INITIAL_SIZE,
1411 table_resize_frequency: DEFAULT_TABLE_RESIZE_FREQUENCY,
1412 table_resize_chunk_size: DEFAULT_TABLE_RESIZE_CHUNK_SIZE,
1413 table_replay_buffer: NZUsize!(DEFAULT_TABLE_REPLAY_BUFFER),
1414 codec_config: (),
1415 };
1416 let mut freezer = Freezer::<_, FixedBytes<64>, i32>::init(
1417 context.child("storage"),
1418 cfg.clone(),
1419 None,
1420 )
1421 .await
1422 .expect("Failed to initialize freezer");
1423
1424 let key = test_key("key1");
1425
1426 freezer
1427 .put(key.clone(), 1)
1428 .await
1429 .expect("Failed to put data");
1430 freezer
1431 .put(key.clone(), 2)
1432 .await
1433 .expect("Failed to put data");
1434 freezer.sync().await.expect("Failed to sync");
1435 assert_eq!(
1436 freezer
1437 .get(Identifier::Key(&key))
1438 .await
1439 .expect("Failed to get data")
1440 .unwrap(),
1441 2
1442 );
1443 });
1444 }
1445}