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