1use super::{
44 fixed::{Config as FixedConfig, Journal as FixedJournal},
45 glob::{Config as GlobConfig, Glob},
46};
47use crate::journal::Error;
48use commonware_codec::{Codec, CodecFixed, CodecShared};
49use commonware_runtime::{BufferPooler, Handle, Metrics, Storage};
50use futures::{future::try_join, stream::Stream};
51use std::{collections::HashSet, num::NonZeroUsize};
52use tracing::{debug, warn};
53
54pub trait Record: CodecFixed<Cfg = ()> + Clone {
59 fn value_location(&self) -> (u64, u32);
61
62 fn with_location(self, offset: u64, size: u32) -> Self;
66}
67
68#[derive(Clone)]
70pub struct Config<C> {
71 pub index_partition: String,
73
74 pub value_partition: String,
76
77 pub index_page_cache: commonware_runtime::buffer::paged::CacheRef,
79
80 pub index_write_buffer: NonZeroUsize,
82
83 pub value_write_buffer: NonZeroUsize,
85
86 pub compression: Option<u8>,
88
89 pub codec_config: C,
91}
92
93pub struct Oversized<E: BufferPooler + Storage + Metrics, I: Record, V: Codec> {
98 index: FixedJournal<E, I>,
99 values: Glob<E, V>,
100}
101
102impl<E: BufferPooler + Storage + Metrics, I: Record + Send + Sync, V: CodecShared>
103 Oversized<E, I, V>
104{
105 pub async fn init(context: E, cfg: Config<V::Cfg>) -> Result<Self, Error> {
111 let index_cfg = FixedConfig {
113 partition: cfg.index_partition,
114 page_cache: cfg.index_page_cache,
115 write_buffer: cfg.index_write_buffer,
116 };
117 let index = FixedJournal::init(context.child("index"), index_cfg).await?;
118
119 let value_cfg = GlobConfig {
120 partition: cfg.value_partition,
121 compression: cfg.compression,
122 codec_config: cfg.codec_config,
123 write_buffer: cfg.value_write_buffer,
124 };
125 let values = Glob::init(context.child("values"), value_cfg).await?;
126
127 let mut oversized = Self { index, values };
128
129 oversized.recover().await?;
131
132 Ok(oversized)
133 }
134
135 async fn recover(&mut self) -> Result<(), Error> {
141 let chunk_size = FixedJournal::<E, I>::CHUNK_SIZE as u64;
142 let sections: Vec<u64> = self.index.sections().collect();
143
144 for section in sections {
145 let index_size = self.index.size(section)?;
146 if index_size == 0 {
147 continue;
148 }
149
150 let glob_size = match self.values.size(section) {
151 Ok(size) => size,
152 Err(Error::AlreadyPrunedToSection(oldest)) => {
153 warn!(
158 section,
159 oldest, "index has section that glob already pruned"
160 );
161 0
162 }
163 Err(e) => return Err(e),
164 };
165
166 let entry_count = index_size / chunk_size;
168 let aligned_size = entry_count * chunk_size;
169 if aligned_size < index_size {
170 warn!(
171 section,
172 index_size, aligned_size, "trailing bytes detected: truncating"
173 );
174 self.index.rewind_section(section, aligned_size).await?;
175 }
176
177 if entry_count == 0 {
179 warn!(
180 section,
181 index_size, "trailing bytes detected: truncating to 0"
182 );
183 self.values.rewind_section(section, 0).await?;
184 continue;
185 }
186
187 let (valid_count, glob_target) = self
189 .find_last_valid_entry(section, entry_count, glob_size)
190 .await;
191
192 if valid_count < entry_count {
194 let valid_size = valid_count * chunk_size;
195 debug!(section, entry_count, valid_count, "rewinding index");
196 self.index.rewind_section(section, valid_size).await?;
197 }
198
199 if glob_size > glob_target {
202 debug!(
203 section,
204 glob_size, glob_target, "truncating glob trailing garbage"
205 );
206 self.values.rewind_section(section, glob_target).await?;
207 }
208 }
209
210 self.cleanup_orphan_value_sections().await?;
212
213 Ok(())
214 }
215
216 async fn cleanup_orphan_value_sections(&mut self) -> Result<(), Error> {
223 let index_sections: HashSet<u64> = self.index.sections().collect();
225
226 let orphan_sections: Vec<u64> = self
228 .values
229 .sections()
230 .filter(|s| !index_sections.contains(s))
231 .collect();
232
233 for section in orphan_sections {
235 warn!(section, "removing orphan value section");
236 self.values.remove_section(section).await?;
237 }
238
239 Ok(())
240 }
241
242 async fn find_last_valid_entry(
248 &self,
249 section: u64,
250 entry_count: u64,
251 glob_size: u64,
252 ) -> (u64, u64) {
253 for pos in (0..entry_count).rev() {
254 match self.index.get(section, pos).await {
255 Ok(entry) => {
256 let (offset, size) = entry.value_location();
257 let entry_end = offset.saturating_add(u64::from(size));
258 if entry_end <= glob_size {
259 return (pos + 1, entry_end);
260 }
261 if pos == entry_count - 1 {
262 warn!(
263 section,
264 pos, glob_size, entry_end, "invalid entry: glob truncated"
265 );
266 }
267 }
268 Err(_) => {
269 if pos == entry_count - 1 {
270 warn!(section, pos, "corrupted last entry, scanning backwards");
271 }
272 }
273 }
274 }
275 (0, 0)
276 }
277
278 pub async fn append(
287 &mut self,
288 section: u64,
289 entry: I,
290 value: &V,
291 ) -> Result<(u64, u64, u32), Error> {
292 let (offset, size) = self.values.append(section, value).await?;
295
296 let entry_with_location = entry.with_location(offset, size);
298 let position = self.index.append(section, &entry_with_location).await?;
299
300 Ok((position, offset, size))
301 }
302
303 pub async fn get(&self, section: u64, position: u64) -> Result<I, Error> {
305 self.index.get(section, position).await
306 }
307
308 pub async fn last(&self, section: u64) -> Result<Option<I>, Error> {
317 self.index.last(section).await
318 }
319
320 pub async fn get_value(&self, section: u64, offset: u64, size: u32) -> Result<V, Error> {
324 self.values.get(section, offset, size).await
325 }
326
327 pub async fn replay(
331 &mut self,
332 start_section: u64,
333 start_position: u64,
334 buffer: NonZeroUsize,
335 ) -> Result<impl Stream<Item = Result<(u64, u64, I), Error>> + Send + '_, Error> {
336 self.index
337 .replay(start_section, start_position, buffer)
338 .await
339 }
340
341 pub async fn sync(&mut self, sections: impl crate::Sections) -> Result<(), Error> {
343 let sections = sections.sections().collect::<Vec<_>>();
344 try_join(self.index.sync(§ions), self.values.sync(§ions))
345 .await
346 .map(|_| ())
347 }
348
349 pub async fn start_sync(
354 &mut self,
355 sections: impl crate::Sections,
356 ) -> Result<Handle<()>, Error> {
357 let sections = sections.sections().collect::<Vec<_>>();
358 let (index, values) = try_join(
359 self.index.start_sync(§ions),
360 self.values.start_sync(§ions),
361 )
362 .await?;
363 Ok(Handle::from_future(async move {
364 try_join(index, values).await.map(|_| ())
365 }))
366 }
367
368 pub async fn sync_all(&mut self) -> Result<(), Error> {
370 try_join(self.index.sync_all(), self.values.sync_all())
371 .await
372 .map(|_| ())
373 }
374
375 pub async fn prune(&mut self, min: u64) -> Result<bool, Error> {
381 let index_pruned = self.index.prune(min).await?;
382 let value_pruned = self.values.prune(min).await?;
383 Ok(index_pruned || value_pruned)
384 }
385
386 pub async fn rewind(&mut self, section: u64, index_size: u64) -> Result<(), Error> {
391 self.index.rewind(section, index_size).await?;
393
394 let value_size = match self.index.last(section).await {
396 Ok(Some(entry)) => {
397 let (offset, size) = entry.value_location();
398 offset
399 .checked_add(u64::from(size))
400 .ok_or(Error::OffsetOverflow)?
401 }
402 Ok(None) => 0,
403 Err(Error::SectionOutOfRange(_)) if index_size == 0 => 0,
404 Err(e) => return Err(e),
405 };
406
407 self.values.rewind(section, value_size).await
409 }
410
411 pub async fn rewind_section(&mut self, section: u64, index_size: u64) -> Result<(), Error> {
416 self.index.rewind_section(section, index_size).await?;
418
419 let value_size = match self.index.last(section).await {
421 Ok(Some(entry)) => {
422 let (offset, size) = entry.value_location();
423 offset
424 .checked_add(u64::from(size))
425 .ok_or(Error::OffsetOverflow)?
426 }
427 Ok(None) => 0,
428 Err(Error::SectionOutOfRange(_)) if index_size == 0 => 0,
429 Err(e) => return Err(e),
430 };
431
432 self.values.rewind_section(section, value_size).await
434 }
435
436 pub fn size(&self, section: u64) -> Result<u64, Error> {
440 self.index.size(section)
441 }
442
443 pub async fn value_size(&self, section: u64) -> Result<u64, Error> {
445 match self.index.last(section).await {
446 Ok(Some(entry)) => {
447 let (offset, size) = entry.value_location();
448 offset
449 .checked_add(u64::from(size))
450 .ok_or(Error::OffsetOverflow)
451 }
452 Ok(None) | Err(Error::SectionOutOfRange(_)) => Ok(0),
453 Err(e) => Err(e),
454 }
455 }
456
457 pub fn oldest_section(&self) -> Option<u64> {
459 self.index.oldest_section()
460 }
461
462 pub fn newest_section(&self) -> Option<u64> {
464 self.index.newest_section()
465 }
466
467 pub async fn destroy(self) -> Result<(), Error> {
469 try_join(self.index.destroy(), self.values.destroy())
470 .await
471 .map(|_| ())
472 }
473}
474
475#[cfg(test)]
476mod tests {
477 use super::*;
478 use commonware_codec::{FixedSize, Read, ReadExt, Write};
479 use commonware_cryptography::Crc32;
480 use commonware_macros::test_traced;
481 use commonware_runtime::{
482 buffer::paged::CacheRef, deterministic, Blob as _, Buf, BufMut, BufferPooler, Runner,
483 Supervisor as _,
484 };
485 use commonware_utils::{NZUsize, NZU16};
486
487 fn byte_end(offset: u64, size: u32) -> u64 {
489 offset + u64::from(size)
490 }
491
492 #[derive(Debug, Clone, PartialEq)]
494 struct TestEntry {
495 id: u64,
496 value_offset: u64,
497 value_size: u32,
498 }
499
500 impl TestEntry {
501 fn new(id: u64, value_offset: u64, value_size: u32) -> Self {
502 Self {
503 id,
504 value_offset,
505 value_size,
506 }
507 }
508 }
509
510 impl Write for TestEntry {
511 fn write(&self, buf: &mut impl BufMut) {
512 self.id.write(buf);
513 self.value_offset.write(buf);
514 self.value_size.write(buf);
515 }
516 }
517
518 impl Read for TestEntry {
519 type Cfg = ();
520
521 fn read_cfg(buf: &mut impl Buf, _: &Self::Cfg) -> Result<Self, commonware_codec::Error> {
522 let id = u64::read(buf)?;
523 let value_offset = u64::read(buf)?;
524 let value_size = u32::read(buf)?;
525 Ok(Self {
526 id,
527 value_offset,
528 value_size,
529 })
530 }
531 }
532
533 impl FixedSize for TestEntry {
534 const SIZE: usize = u64::SIZE + u64::SIZE + u32::SIZE;
535 }
536
537 impl Record for TestEntry {
538 fn value_location(&self) -> (u64, u32) {
539 (self.value_offset, self.value_size)
540 }
541
542 fn with_location(mut self, offset: u64, size: u32) -> Self {
543 self.value_offset = offset;
544 self.value_size = size;
545 self
546 }
547 }
548
549 fn test_cfg(pooler: &impl BufferPooler) -> Config<()> {
550 Config {
551 index_partition: "test-index".into(),
552 value_partition: "test-values".into(),
553 index_page_cache: CacheRef::from_pooler(pooler, NZU16!(64), NZUsize!(8)),
554 index_write_buffer: NZUsize!(1024),
555 value_write_buffer: NZUsize!(1024),
556 compression: None,
557 codec_config: (),
558 }
559 }
560
561 type TestValue = [u8; 16];
563
564 #[test_traced]
565 fn test_oversized_append_and_get() {
566 let executor = deterministic::Runner::default();
567 executor.start(|context| async move {
568 let cfg = test_cfg(&context);
569 let mut oversized: Oversized<_, TestEntry, TestValue> =
570 Oversized::init(context, cfg).await.expect("Failed to init");
571
572 let value: TestValue = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16];
574 let entry = TestEntry::new(42, 0, 0);
575 let (position, offset, size) = oversized
576 .append(1, entry, &value)
577 .await
578 .expect("Failed to append");
579
580 assert_eq!(position, 0);
581
582 let retrieved_entry = oversized.get(1, position).await.expect("Failed to get");
584 assert_eq!(retrieved_entry.id, 42);
585
586 let retrieved_value = oversized
588 .get_value(1, offset, size)
589 .await
590 .expect("Failed to get value");
591 assert_eq!(retrieved_value, value);
592
593 oversized.destroy().await.expect("Failed to destroy");
594 });
595 }
596
597 #[test_traced]
598 fn test_oversized_crash_recovery() {
599 let executor = deterministic::Runner::default();
600 executor.start(|context| async move {
601 let cfg = test_cfg(&context);
602
603 let mut oversized: Oversized<_, TestEntry, TestValue> =
605 Oversized::init(context.child("first"), cfg.clone())
606 .await
607 .expect("Failed to init");
608
609 let mut locations = Vec::new();
611 for i in 0..5u8 {
612 let value: TestValue = [i; 16];
613 let entry = TestEntry::new(i as u64, 0, 0);
614 let (position, offset, size) = oversized
615 .append(1, entry, &value)
616 .await
617 .expect("Failed to append");
618 locations.push((position, offset, size));
619 }
620 oversized.sync(1).await.expect("Failed to sync");
621 drop(oversized);
622
623 let (blob, _) = context
625 .open(&cfg.value_partition, &1u64.to_be_bytes())
626 .await
627 .expect("Failed to open blob");
628
629 let keep_size = byte_end(locations[2].1, locations[2].2);
631 blob.resize(keep_size).await.expect("Failed to truncate");
632 blob.sync().await.expect("Failed to sync");
633 drop(blob);
634
635 let oversized: Oversized<_, TestEntry, TestValue> =
637 Oversized::init(context.child("second"), cfg.clone())
638 .await
639 .expect("Failed to reinit");
640
641 for i in 0..3u8 {
643 let (position, offset, size) = locations[i as usize];
644 let entry = oversized.get(1, position).await.expect("Failed to get");
645 assert_eq!(entry.id, i as u64);
646
647 let value = oversized
648 .get_value(1, offset, size)
649 .await
650 .expect("Failed to get value");
651 assert_eq!(value, [i; 16]);
652 }
653
654 let result = oversized.get(1, 3).await;
656 assert!(result.is_err());
657
658 oversized.destroy().await.expect("Failed to destroy");
659 });
660 }
661
662 #[test_traced]
663 fn test_oversized_persistence() {
664 let executor = deterministic::Runner::default();
665 executor.start(|context| async move {
666 let cfg = test_cfg(&context);
667
668 let mut oversized: Oversized<_, TestEntry, TestValue> =
670 Oversized::init(context.child("first"), cfg.clone())
671 .await
672 .expect("Failed to init");
673
674 let value: TestValue = [42; 16];
675 let entry = TestEntry::new(123, 0, 0);
676 let (position, offset, size) = oversized
677 .append(1, entry, &value)
678 .await
679 .expect("Failed to append");
680 oversized.sync(1).await.expect("Failed to sync");
681 drop(oversized);
682
683 let oversized: Oversized<_, TestEntry, TestValue> =
685 Oversized::init(context.child("second"), cfg)
686 .await
687 .expect("Failed to reinit");
688
689 let retrieved_entry = oversized.get(1, position).await.expect("Failed to get");
690 assert_eq!(retrieved_entry.id, 123);
691
692 let retrieved_value = oversized
693 .get_value(1, offset, size)
694 .await
695 .expect("Failed to get value");
696 assert_eq!(retrieved_value, value);
697
698 oversized.destroy().await.expect("Failed to destroy");
699 });
700 }
701
702 #[test_traced]
703 fn test_oversized_sync() {
704 let executor = deterministic::Runner::default();
705 executor.start(|context| async move {
706 let cfg = test_cfg(&context);
707
708 let mut oversized: Oversized<_, TestEntry, TestValue> =
709 Oversized::init(context.child("first"), cfg.clone())
710 .await
711 .expect("Failed to init");
712
713 let mut located = Vec::new();
715 for section in 1u64..=3 {
716 let value: TestValue = [section as u8; 16];
717 let entry = TestEntry::new(section, 0, 0);
718 let (position, offset, size) = oversized
719 .append(section, entry, &value)
720 .await
721 .expect("Failed to append");
722 located.push((section, position, offset, size, value));
723 }
724
725 oversized
728 .sync(&[1, 3, 99])
729 .await
730 .expect("Failed to sync sections");
731 drop(oversized);
732
733 let oversized: Oversized<_, TestEntry, TestValue> =
735 Oversized::init(context.child("second"), cfg)
736 .await
737 .expect("Failed to reinit");
738 for &(section, position, offset, size, value) in &located {
739 let result = oversized.get(section, position).await;
740 if section == 2 {
741 assert!(result.is_err(), "unsynced section 2 must not be durable");
742 continue;
743 }
744 assert_eq!(result.expect("synced entry durable").id, section);
745 let retrieved = oversized
746 .get_value(section, offset, size)
747 .await
748 .expect("synced value durable");
749 assert_eq!(retrieved, value);
750 }
751
752 oversized.destroy().await.expect("Failed to destroy");
753 });
754 }
755
756 #[test_traced]
757 fn test_oversized_prune() {
758 let executor = deterministic::Runner::default();
759 executor.start(|context| async move {
760 let cfg = test_cfg(&context);
761 let mut oversized: Oversized<_, TestEntry, TestValue> =
762 Oversized::init(context, cfg).await.expect("Failed to init");
763
764 for section in 1u64..=5 {
766 let value: TestValue = [section as u8; 16];
767 let entry = TestEntry::new(section, 0, 0);
768 oversized
769 .append(section, entry, &value)
770 .await
771 .expect("Failed to append");
772 oversized.sync(section).await.expect("Failed to sync");
773 }
774
775 oversized.prune(3).await.expect("Failed to prune");
777
778 assert!(oversized.get(1, 0).await.is_err());
780 assert!(oversized.get(2, 0).await.is_err());
781
782 assert!(oversized.get(3, 0).await.is_ok());
784 assert!(oversized.get(4, 0).await.is_ok());
785 assert!(oversized.get(5, 0).await.is_ok());
786
787 oversized.destroy().await.expect("Failed to destroy");
788 });
789 }
790
791 #[test_traced]
792 fn test_recovery_empty_section() {
793 let executor = deterministic::Runner::default();
794 executor.start(|context| async move {
795 let cfg = test_cfg(&context);
796
797 let mut oversized: Oversized<_, TestEntry, TestValue> =
799 Oversized::init(context.child("first"), cfg.clone())
800 .await
801 .expect("Failed to init");
802
803 let value: TestValue = [42; 16];
805 let entry = TestEntry::new(1, 0, 0);
806 oversized
807 .append(2, entry, &value)
808 .await
809 .expect("Failed to append");
810 oversized.sync(2).await.expect("Failed to sync");
811 drop(oversized);
812
813 let oversized: Oversized<_, TestEntry, TestValue> =
815 Oversized::init(context.child("second"), cfg)
816 .await
817 .expect("Failed to reinit");
818
819 let entry = oversized.get(2, 0).await.expect("Failed to get");
821 assert_eq!(entry.id, 1);
822
823 oversized.destroy().await.expect("Failed to destroy");
824 });
825 }
826
827 #[test_traced]
828 fn test_recovery_all_entries_invalid() {
829 let executor = deterministic::Runner::default();
830 executor.start(|context| async move {
831 let cfg = test_cfg(&context);
832
833 let mut oversized: Oversized<_, TestEntry, TestValue> =
835 Oversized::init(context.child("first"), cfg.clone())
836 .await
837 .expect("Failed to init");
838
839 for i in 0..5u8 {
841 let value: TestValue = [i; 16];
842 let entry = TestEntry::new(i as u64, 0, 0);
843 oversized
844 .append(1, entry, &value)
845 .await
846 .expect("Failed to append");
847 }
848 oversized.sync(1).await.expect("Failed to sync");
849 drop(oversized);
850
851 let (blob, _) = context
853 .open(&cfg.value_partition, &1u64.to_be_bytes())
854 .await
855 .expect("Failed to open blob");
856 blob.resize(0).await.expect("Failed to truncate");
857 blob.sync().await.expect("Failed to sync");
858 drop(blob);
859
860 let mut oversized: Oversized<_, TestEntry, TestValue> =
862 Oversized::init(context.child("second"), cfg)
863 .await
864 .expect("Failed to reinit");
865
866 let result = oversized.get(1, 0).await;
868 assert!(result.is_err());
869
870 let value: TestValue = [99; 16];
872 let entry = TestEntry::new(100, 0, 0);
873 let (pos, offset, size) = oversized
874 .append(1, entry, &value)
875 .await
876 .expect("Failed to append after recovery");
877 assert_eq!(pos, 0);
878
879 let retrieved = oversized.get(1, 0).await.expect("Failed to get");
880 assert_eq!(retrieved.id, 100);
881 let retrieved_value = oversized
882 .get_value(1, offset, size)
883 .await
884 .expect("Failed to get value");
885 assert_eq!(retrieved_value, value);
886
887 oversized.destroy().await.expect("Failed to destroy");
888 });
889 }
890
891 #[test_traced]
892 fn test_recovery_multiple_sections_mixed_validity() {
893 let executor = deterministic::Runner::default();
894 executor.start(|context| async move {
895 let cfg = test_cfg(&context);
896
897 let mut oversized: Oversized<_, TestEntry, TestValue> =
899 Oversized::init(context.child("first"), cfg.clone())
900 .await
901 .expect("Failed to init");
902
903 let mut section1_locations = Vec::new();
905 for i in 0..3u8 {
906 let value: TestValue = [i; 16];
907 let entry = TestEntry::new(i as u64, 0, 0);
908 let loc = oversized
909 .append(1, entry, &value)
910 .await
911 .expect("Failed to append");
912 section1_locations.push(loc);
913 }
914 oversized.sync(1).await.expect("Failed to sync");
915
916 let mut section2_locations = Vec::new();
918 for i in 0..5u8 {
919 let value: TestValue = [10 + i; 16];
920 let entry = TestEntry::new(10 + i as u64, 0, 0);
921 let loc = oversized
922 .append(2, entry, &value)
923 .await
924 .expect("Failed to append");
925 section2_locations.push(loc);
926 }
927 oversized.sync(2).await.expect("Failed to sync");
928
929 for i in 0..2u8 {
931 let value: TestValue = [20 + i; 16];
932 let entry = TestEntry::new(20 + i as u64, 0, 0);
933 oversized
934 .append(3, entry, &value)
935 .await
936 .expect("Failed to append");
937 }
938 oversized.sync(3).await.expect("Failed to sync");
939 drop(oversized);
940
941 let (blob, _) = context
943 .open(&cfg.value_partition, &1u64.to_be_bytes())
944 .await
945 .expect("Failed to open blob");
946 let keep_size = byte_end(section1_locations[0].1, section1_locations[0].2);
947 blob.resize(keep_size).await.expect("Failed to truncate");
948 blob.sync().await.expect("Failed to sync");
949 drop(blob);
950
951 let (blob, _) = context
953 .open(&cfg.value_partition, &2u64.to_be_bytes())
954 .await
955 .expect("Failed to open blob");
956 let keep_size = byte_end(section2_locations[2].1, section2_locations[2].2);
957 blob.resize(keep_size).await.expect("Failed to truncate");
958 blob.sync().await.expect("Failed to sync");
959 drop(blob);
960
961 let oversized: Oversized<_, TestEntry, TestValue> =
965 Oversized::init(context.child("second"), cfg)
966 .await
967 .expect("Failed to reinit");
968
969 assert!(oversized.get(1, 0).await.is_ok());
971 assert!(oversized.get(1, 1).await.is_err());
972 assert!(oversized.get(1, 2).await.is_err());
973
974 assert!(oversized.get(2, 0).await.is_ok());
976 assert!(oversized.get(2, 1).await.is_ok());
977 assert!(oversized.get(2, 2).await.is_ok());
978 assert!(oversized.get(2, 3).await.is_err());
979 assert!(oversized.get(2, 4).await.is_err());
980
981 assert!(oversized.get(3, 0).await.is_ok());
983 assert!(oversized.get(3, 1).await.is_ok());
984
985 oversized.destroy().await.expect("Failed to destroy");
986 });
987 }
988
989 #[test_traced]
990 fn test_recovery_corrupted_last_index_entry() {
991 let executor = deterministic::Runner::default();
992 executor.start(|context| async move {
993 let cfg = Config {
997 index_partition: "test-index".into(),
998 value_partition: "test-values".into(),
999 index_page_cache: CacheRef::from_pooler(
1000 &context,
1001 NZU16!(TestEntry::SIZE as u16),
1002 NZUsize!(8),
1003 ),
1004 index_write_buffer: NZUsize!(1024),
1005 value_write_buffer: NZUsize!(1024),
1006 compression: None,
1007 codec_config: (),
1008 };
1009
1010 let mut oversized: Oversized<_, TestEntry, TestValue> =
1012 Oversized::init(context.child("first"), cfg.clone())
1013 .await
1014 .expect("Failed to init");
1015
1016 for i in 0..5u8 {
1018 let value: TestValue = [i; 16];
1019 let entry = TestEntry::new(i as u64, 0, 0);
1020 oversized
1021 .append(1, entry, &value)
1022 .await
1023 .expect("Failed to append");
1024 }
1025 oversized.sync(1).await.expect("Failed to sync");
1026 drop(oversized);
1027
1028 let (blob, size) = context
1030 .open(&cfg.index_partition, &1u64.to_be_bytes())
1031 .await
1032 .expect("Failed to open blob");
1033
1034 assert_eq!(size, 160);
1038 let last_page_crc_offset = size - 12;
1039 blob.write_at_sync(last_page_crc_offset, vec![0xFF; 12])
1040 .await
1041 .expect("Failed to corrupt");
1042 drop(blob);
1043
1044 let mut oversized: Oversized<_, TestEntry, TestValue> =
1046 Oversized::init(context.child("second"), cfg)
1047 .await
1048 .expect("Failed to reinit");
1049
1050 for i in 0..4u8 {
1052 let entry = oversized.get(1, i as u64).await.expect("Failed to get");
1053 assert_eq!(entry.id, i as u64);
1054 }
1055
1056 assert!(oversized.get(1, 4).await.is_err());
1058
1059 let value: TestValue = [99; 16];
1061 let entry = TestEntry::new(100, 0, 0);
1062 let (pos, offset, size) = oversized
1063 .append(1, entry, &value)
1064 .await
1065 .expect("Failed to append after recovery");
1066 assert_eq!(pos, 4);
1067
1068 let retrieved = oversized.get(1, 4).await.expect("Failed to get");
1069 assert_eq!(retrieved.id, 100);
1070 let retrieved_value = oversized
1071 .get_value(1, offset, size)
1072 .await
1073 .expect("Failed to get value");
1074 assert_eq!(retrieved_value, value);
1075
1076 oversized.destroy().await.expect("Failed to destroy");
1077 });
1078 }
1079
1080 #[test_traced]
1081 fn test_recovery_all_entries_valid() {
1082 let executor = deterministic::Runner::default();
1083 executor.start(|context| async move {
1084 let cfg = test_cfg(&context);
1085
1086 let mut oversized: Oversized<_, TestEntry, TestValue> =
1088 Oversized::init(context.child("first"), cfg.clone())
1089 .await
1090 .expect("Failed to init");
1091
1092 for section in 1u64..=3 {
1094 for i in 0..10u8 {
1095 let value: TestValue = [(section as u8) * 10 + i; 16];
1096 let entry = TestEntry::new(section * 100 + i as u64, 0, 0);
1097 oversized
1098 .append(section, entry, &value)
1099 .await
1100 .expect("Failed to append");
1101 }
1102 oversized.sync(section).await.expect("Failed to sync");
1103 }
1104 drop(oversized);
1105
1106 let oversized: Oversized<_, TestEntry, TestValue> =
1108 Oversized::init(context.child("second"), cfg)
1109 .await
1110 .expect("Failed to reinit");
1111
1112 for section in 1u64..=3 {
1114 for i in 0..10u8 {
1115 let entry = oversized
1116 .get(section, i as u64)
1117 .await
1118 .expect("Failed to get");
1119 assert_eq!(entry.id, section * 100 + i as u64);
1120 }
1121 }
1122
1123 oversized.destroy().await.expect("Failed to destroy");
1124 });
1125 }
1126
1127 #[test_traced]
1128 fn test_recovery_single_entry_invalid() {
1129 let executor = deterministic::Runner::default();
1130 executor.start(|context| async move {
1131 let cfg = test_cfg(&context);
1132
1133 let mut oversized: Oversized<_, TestEntry, TestValue> =
1135 Oversized::init(context.child("first"), cfg.clone())
1136 .await
1137 .expect("Failed to init");
1138
1139 let value: TestValue = [42; 16];
1140 let entry = TestEntry::new(1, 0, 0);
1141 oversized
1142 .append(1, entry, &value)
1143 .await
1144 .expect("Failed to append");
1145 oversized.sync(1).await.expect("Failed to sync");
1146 drop(oversized);
1147
1148 let (blob, _) = context
1150 .open(&cfg.value_partition, &1u64.to_be_bytes())
1151 .await
1152 .expect("Failed to open blob");
1153 blob.resize(0).await.expect("Failed to truncate");
1154 blob.sync().await.expect("Failed to sync");
1155 drop(blob);
1156
1157 let oversized: Oversized<_, TestEntry, TestValue> =
1159 Oversized::init(context.child("second"), cfg)
1160 .await
1161 .expect("Failed to reinit");
1162
1163 assert!(oversized.get(1, 0).await.is_err());
1165
1166 oversized.destroy().await.expect("Failed to destroy");
1167 });
1168 }
1169
1170 #[test_traced]
1171 fn test_recovery_last_entry_off_by_one() {
1172 let executor = deterministic::Runner::default();
1173 executor.start(|context| async move {
1174 let cfg = test_cfg(&context);
1175
1176 let mut oversized: Oversized<_, TestEntry, TestValue> =
1178 Oversized::init(context.child("first"), cfg.clone())
1179 .await
1180 .expect("Failed to init");
1181
1182 let mut locations = Vec::new();
1183 for i in 0..3u8 {
1184 let value: TestValue = [i; 16];
1185 let entry = TestEntry::new(i as u64, 0, 0);
1186 let loc = oversized
1187 .append(1, entry, &value)
1188 .await
1189 .expect("Failed to append");
1190 locations.push(loc);
1191 }
1192 oversized.sync(1).await.expect("Failed to sync");
1193 drop(oversized);
1194
1195 let (blob, _) = context
1197 .open(&cfg.value_partition, &1u64.to_be_bytes())
1198 .await
1199 .expect("Failed to open blob");
1200
1201 let last = &locations[2];
1204 let truncate_to = byte_end(last.1, last.2) - 1;
1205 blob.resize(truncate_to).await.expect("Failed to truncate");
1206 blob.sync().await.expect("Failed to sync");
1207 drop(blob);
1208
1209 let mut oversized: Oversized<_, TestEntry, TestValue> =
1211 Oversized::init(context.child("second"), cfg)
1212 .await
1213 .expect("Failed to reinit");
1214
1215 assert!(oversized.get(1, 0).await.is_ok());
1217 assert!(oversized.get(1, 1).await.is_ok());
1218
1219 assert!(oversized.get(1, 2).await.is_err());
1221
1222 let value: TestValue = [99; 16];
1224 let entry = TestEntry::new(100, 0, 0);
1225 let (pos, offset, size) = oversized
1226 .append(1, entry, &value)
1227 .await
1228 .expect("Failed to append after recovery");
1229 assert_eq!(pos, 2);
1230
1231 let retrieved = oversized.get(1, 2).await.expect("Failed to get");
1232 assert_eq!(retrieved.id, 100);
1233 let retrieved_value = oversized
1234 .get_value(1, offset, size)
1235 .await
1236 .expect("Failed to get value");
1237 assert_eq!(retrieved_value, value);
1238
1239 oversized.destroy().await.expect("Failed to destroy");
1240 });
1241 }
1242
1243 #[test_traced]
1244 fn test_recovery_glob_missing_entirely() {
1245 let executor = deterministic::Runner::default();
1246 executor.start(|context| async move {
1247 let cfg = test_cfg(&context);
1248
1249 let mut oversized: Oversized<_, TestEntry, TestValue> =
1251 Oversized::init(context.child("first"), cfg.clone())
1252 .await
1253 .expect("Failed to init");
1254
1255 for i in 0..3u8 {
1256 let value: TestValue = [i; 16];
1257 let entry = TestEntry::new(i as u64, 0, 0);
1258 oversized
1259 .append(1, entry, &value)
1260 .await
1261 .expect("Failed to append");
1262 }
1263 oversized.sync(1).await.expect("Failed to sync");
1264 drop(oversized);
1265
1266 context
1268 .remove(&cfg.value_partition, Some(&1u64.to_be_bytes()))
1269 .await
1270 .expect("Failed to remove");
1271
1272 let oversized: Oversized<_, TestEntry, TestValue> =
1274 Oversized::init(context.child("second"), cfg)
1275 .await
1276 .expect("Failed to reinit");
1277
1278 assert!(oversized.get(1, 0).await.is_err());
1280 assert!(oversized.get(1, 1).await.is_err());
1281 assert!(oversized.get(1, 2).await.is_err());
1282
1283 oversized.destroy().await.expect("Failed to destroy");
1284 });
1285 }
1286
1287 #[test_traced]
1288 fn test_recovery_can_append_after_recovery() {
1289 let executor = deterministic::Runner::default();
1290 executor.start(|context| async move {
1291 let cfg = test_cfg(&context);
1292
1293 let mut oversized: Oversized<_, TestEntry, TestValue> =
1295 Oversized::init(context.child("first"), cfg.clone())
1296 .await
1297 .expect("Failed to init");
1298
1299 let mut locations = Vec::new();
1300 for i in 0..5u8 {
1301 let value: TestValue = [i; 16];
1302 let entry = TestEntry::new(i as u64, 0, 0);
1303 let loc = oversized
1304 .append(1, entry, &value)
1305 .await
1306 .expect("Failed to append");
1307 locations.push(loc);
1308 }
1309 oversized.sync(1).await.expect("Failed to sync");
1310 drop(oversized);
1311
1312 let (blob, _) = context
1314 .open(&cfg.value_partition, &1u64.to_be_bytes())
1315 .await
1316 .expect("Failed to open blob");
1317 let keep_size = byte_end(locations[1].1, locations[1].2);
1318 blob.resize(keep_size).await.expect("Failed to truncate");
1319 blob.sync().await.expect("Failed to sync");
1320 drop(blob);
1321
1322 let mut oversized: Oversized<_, TestEntry, TestValue> =
1324 Oversized::init(context.child("second"), cfg.clone())
1325 .await
1326 .expect("Failed to reinit");
1327
1328 assert!(oversized.get(1, 0).await.is_ok());
1330 assert!(oversized.get(1, 1).await.is_ok());
1331 assert!(oversized.get(1, 2).await.is_err());
1332
1333 for i in 10..15u8 {
1335 let value: TestValue = [i; 16];
1336 let entry = TestEntry::new(i as u64, 0, 0);
1337 oversized
1338 .append(1, entry, &value)
1339 .await
1340 .expect("Failed to append after recovery");
1341 }
1342 oversized.sync(1).await.expect("Failed to sync");
1343
1344 for i in 0..5u8 {
1346 let entry = oversized
1347 .get(1, 2 + i as u64)
1348 .await
1349 .expect("Failed to get new entry");
1350 assert_eq!(entry.id, (10 + i) as u64);
1351 }
1352
1353 oversized.destroy().await.expect("Failed to destroy");
1354 });
1355 }
1356
1357 #[test_traced]
1358 fn test_recovery_glob_pruned_but_index_not() {
1359 let executor = deterministic::Runner::default();
1360 executor.start(|context| async move {
1361 let cfg = test_cfg(&context);
1362
1363 let mut oversized: Oversized<_, TestEntry, TestValue> =
1365 Oversized::init(context.child("first"), cfg.clone())
1366 .await
1367 .expect("Failed to init");
1368
1369 for section in 1u64..=3 {
1370 let value: TestValue = [section as u8; 16];
1371 let entry = TestEntry::new(section, 0, 0);
1372 oversized
1373 .append(section, entry, &value)
1374 .await
1375 .expect("Failed to append");
1376 oversized.sync(section).await.expect("Failed to sync");
1377 }
1378 drop(oversized);
1379
1380 use crate::journal::segmented::glob::{Config as GlobConfig, Glob};
1383 let glob_cfg = GlobConfig {
1384 partition: cfg.value_partition.clone(),
1385 compression: cfg.compression,
1386 codec_config: (),
1387 write_buffer: cfg.value_write_buffer,
1388 };
1389 let mut glob: Glob<_, TestValue> = Glob::init(context.child("glob"), glob_cfg)
1390 .await
1391 .expect("Failed to init glob");
1392 glob.prune(2).await.expect("Failed to prune glob");
1393 glob.sync_all().await.expect("Failed to sync glob");
1394 drop(glob);
1395
1396 let oversized: Oversized<_, TestEntry, TestValue> =
1399 Oversized::init(context.child("second"), cfg.clone())
1400 .await
1401 .expect("Failed to reinit");
1402
1403 assert!(oversized.get(1, 0).await.is_err());
1405
1406 assert!(oversized.get(2, 0).await.is_ok());
1408 assert!(oversized.get(3, 0).await.is_ok());
1409
1410 oversized.destroy().await.expect("Failed to destroy");
1411 });
1412 }
1413
1414 #[test_traced]
1415 fn test_recovery_index_partition_deleted() {
1416 let executor = deterministic::Runner::default();
1417 executor.start(|context| async move {
1418 let cfg = test_cfg(&context);
1419
1420 let mut oversized: Oversized<_, TestEntry, TestValue> =
1422 Oversized::init(context.child("first"), cfg.clone())
1423 .await
1424 .expect("Failed to init");
1425
1426 for section in 1u64..=3 {
1427 let value: TestValue = [section as u8; 16];
1428 let entry = TestEntry::new(section, 0, 0);
1429 oversized
1430 .append(section, entry, &value)
1431 .await
1432 .expect("Failed to append");
1433 oversized.sync(section).await.expect("Failed to sync");
1434 }
1435 drop(oversized);
1436
1437 context
1439 .remove(&cfg.index_partition, Some(&2u64.to_be_bytes()))
1440 .await
1441 .expect("Failed to remove index");
1442
1443 let oversized: Oversized<_, TestEntry, TestValue> =
1446 Oversized::init(context.child("second"), cfg.clone())
1447 .await
1448 .expect("Failed to reinit");
1449
1450 assert!(oversized.get(1, 0).await.is_ok());
1452 assert!(oversized.get(3, 0).await.is_ok());
1453
1454 assert!(oversized.get(2, 0).await.is_err());
1456
1457 oversized.destroy().await.expect("Failed to destroy");
1458 });
1459 }
1460
1461 #[test_traced]
1462 fn test_recovery_index_synced_but_glob_not() {
1463 let executor = deterministic::Runner::default();
1464 executor.start(|context| async move {
1465 let cfg = test_cfg(&context);
1466
1467 let mut oversized: Oversized<_, TestEntry, TestValue> =
1469 Oversized::init(context.child("first"), cfg.clone())
1470 .await
1471 .expect("Failed to init");
1472
1473 let mut locations = Vec::new();
1475 for i in 0..3u8 {
1476 let value: TestValue = [i; 16];
1477 let entry = TestEntry::new(i as u64, 0, 0);
1478 let loc = oversized
1479 .append(1, entry, &value)
1480 .await
1481 .expect("Failed to append");
1482 locations.push(loc);
1483 }
1484 oversized.sync(1).await.expect("Failed to sync");
1485
1486 for i in 10..15u8 {
1488 let value: TestValue = [i; 16];
1489 let entry = TestEntry::new(i as u64, 0, 0);
1490 oversized
1491 .append(1, entry, &value)
1492 .await
1493 .expect("Failed to append");
1494 }
1495 drop(oversized);
1497
1498 let (blob, _) = context
1501 .open(&cfg.value_partition, &1u64.to_be_bytes())
1502 .await
1503 .expect("Failed to open blob");
1504 let synced_size = byte_end(locations[2].1, locations[2].2);
1505 blob.resize(synced_size).await.expect("Failed to truncate");
1506 blob.sync().await.expect("Failed to sync");
1507 drop(blob);
1508
1509 let oversized: Oversized<_, TestEntry, TestValue> =
1511 Oversized::init(context.child("second"), cfg)
1512 .await
1513 .expect("Failed to reinit");
1514
1515 for i in 0..3u8 {
1517 let entry = oversized.get(1, i as u64).await.expect("Failed to get");
1518 assert_eq!(entry.id, i as u64);
1519 }
1520
1521 assert!(oversized.get(1, 3).await.is_err());
1523
1524 oversized.destroy().await.expect("Failed to destroy");
1525 });
1526 }
1527
1528 #[test_traced]
1529 fn test_recovery_glob_synced_but_index_not() {
1530 let executor = deterministic::Runner::default();
1531 executor.start(|context| async move {
1532 let cfg = Config {
1536 index_partition: "test-index".into(),
1537 value_partition: "test-values".into(),
1538 index_page_cache: CacheRef::from_pooler(
1539 &context,
1540 NZU16!(TestEntry::SIZE as u16),
1541 NZUsize!(8),
1542 ),
1543 index_write_buffer: NZUsize!(1024),
1544 value_write_buffer: NZUsize!(1024),
1545 compression: None,
1546 codec_config: (),
1547 };
1548
1549 let mut oversized: Oversized<_, TestEntry, TestValue> =
1551 Oversized::init(context.child("first"), cfg.clone())
1552 .await
1553 .expect("Failed to init");
1554
1555 let mut locations = Vec::new();
1557 for i in 0..3u8 {
1558 let value: TestValue = [i; 16];
1559 let entry = TestEntry::new(i as u64, 0, 0);
1560 let loc = oversized
1561 .append(1, entry, &value)
1562 .await
1563 .expect("Failed to append");
1564 locations.push(loc);
1565 }
1566 oversized.sync(1).await.expect("Failed to sync");
1567 drop(oversized);
1568
1569 let (blob, _size) = context
1572 .open(&cfg.index_partition, &1u64.to_be_bytes())
1573 .await
1574 .expect("Failed to open blob");
1575
1576 let physical_page_size = (TestEntry::SIZE + 12) as u64;
1579 blob.resize(2 * physical_page_size)
1580 .await
1581 .expect("Failed to truncate");
1582 blob.sync().await.expect("Failed to sync");
1583 drop(blob);
1584
1585 let mut oversized: Oversized<_, TestEntry, TestValue> =
1587 Oversized::init(context.child("second"), cfg.clone())
1588 .await
1589 .expect("Failed to reinit");
1590
1591 for i in 0..2u8 {
1593 let (position, offset, size) = locations[i as usize];
1594 let entry = oversized.get(1, position).await.expect("Failed to get");
1595 assert_eq!(entry.id, i as u64);
1596
1597 let value = oversized
1598 .get_value(1, offset, size)
1599 .await
1600 .expect("Failed to get value");
1601 assert_eq!(value, [i; 16]);
1602 }
1603
1604 assert!(oversized.get(1, 2).await.is_err());
1606
1607 let mut new_locations = Vec::new();
1609 for i in 10..13u8 {
1610 let value: TestValue = [i; 16];
1611 let entry = TestEntry::new(i as u64, 0, 0);
1612 let (position, offset, size) = oversized
1613 .append(1, entry, &value)
1614 .await
1615 .expect("Failed to append after recovery");
1616
1617 assert_eq!(position, (i - 10 + 2) as u64);
1619 new_locations.push((position, offset, size, i));
1620
1621 let retrieved = oversized.get(1, position).await.expect("Failed to get");
1623 assert_eq!(retrieved.id, i as u64);
1624
1625 let retrieved_value = oversized
1626 .get_value(1, offset, size)
1627 .await
1628 .expect("Failed to get value");
1629 assert_eq!(retrieved_value, value);
1630 }
1631
1632 oversized.sync(1).await.expect("Failed to sync");
1634 drop(oversized);
1635
1636 let oversized: Oversized<_, TestEntry, TestValue> =
1638 Oversized::init(context.child("third"), cfg)
1639 .await
1640 .expect("Failed to reinit after append");
1641
1642 for i in 0..2u8 {
1645 let (position, offset, size) = locations[i as usize];
1646 let entry = oversized.get(1, position).await.expect("Failed to get");
1647 assert_eq!(entry.id, i as u64);
1648
1649 let value = oversized
1650 .get_value(1, offset, size)
1651 .await
1652 .expect("Failed to get value");
1653 assert_eq!(value, [i; 16]);
1654 }
1655
1656 for (position, offset, size, expected_id) in &new_locations {
1658 let entry = oversized
1659 .get(1, *position)
1660 .await
1661 .expect("Failed to get new entry after restart");
1662 assert_eq!(entry.id, *expected_id as u64);
1663
1664 let value = oversized
1665 .get_value(1, *offset, *size)
1666 .await
1667 .expect("Failed to get new value after restart");
1668 assert_eq!(value, [*expected_id; 16]);
1669 }
1670
1671 assert!(oversized.get(1, 4).await.is_ok());
1673 assert!(oversized.get(1, 5).await.is_err());
1674
1675 oversized.destroy().await.expect("Failed to destroy");
1676 });
1677 }
1678
1679 #[test_traced]
1680 fn test_recovery_partial_index_entry() {
1681 let executor = deterministic::Runner::default();
1682 executor.start(|context| async move {
1683 let cfg = test_cfg(&context);
1684
1685 let mut oversized: Oversized<_, TestEntry, TestValue> =
1687 Oversized::init(context.child("first"), cfg.clone())
1688 .await
1689 .expect("Failed to init");
1690
1691 for i in 0..3u8 {
1693 let value: TestValue = [i; 16];
1694 let entry = TestEntry::new(i as u64, 0, 0);
1695 oversized
1696 .append(1, entry, &value)
1697 .await
1698 .expect("Failed to append");
1699 }
1700 oversized.sync(1).await.expect("Failed to sync");
1701 drop(oversized);
1702
1703 let (blob, _) = context
1707 .open(&cfg.index_partition, &1u64.to_be_bytes())
1708 .await
1709 .expect("Failed to open blob");
1710 let partial_size = 3 * 24 + 10; blob.resize(partial_size).await.expect("Failed to resize");
1712 blob.sync().await.expect("Failed to sync");
1713 drop(blob);
1714
1715 let mut oversized: Oversized<_, TestEntry, TestValue> =
1717 Oversized::init(context.child("second"), cfg.clone())
1718 .await
1719 .expect("Failed to reinit");
1720
1721 for i in 0..3u8 {
1723 let entry = oversized.get(1, i as u64).await.expect("Failed to get");
1724 assert_eq!(entry.id, i as u64);
1725 }
1726
1727 assert!(oversized.get(1, 3).await.is_err());
1729
1730 let value: TestValue = [42; 16];
1732 let entry = TestEntry::new(100, 0, 0);
1733 let (pos, offset, size) = oversized
1734 .append(1, entry, &value)
1735 .await
1736 .expect("Failed to append after recovery");
1737 assert_eq!(pos, 3);
1738
1739 let retrieved = oversized.get(1, 3).await.expect("Failed to get new entry");
1741 assert_eq!(retrieved.id, 100);
1742 let retrieved_value = oversized
1743 .get_value(1, offset, size)
1744 .await
1745 .expect("Failed to get new value");
1746 assert_eq!(retrieved_value, value);
1747
1748 oversized.destroy().await.expect("Failed to destroy");
1749 });
1750 }
1751
1752 #[test_traced]
1753 fn test_recovery_only_partial_entry() {
1754 let executor = deterministic::Runner::default();
1755 executor.start(|context| async move {
1756 let cfg = test_cfg(&context);
1757
1758 let mut oversized: Oversized<_, TestEntry, TestValue> =
1760 Oversized::init(context.child("first"), cfg.clone())
1761 .await
1762 .expect("Failed to init");
1763
1764 let value: TestValue = [42; 16];
1765 let entry = TestEntry::new(1, 0, 0);
1766 oversized
1767 .append(1, entry, &value)
1768 .await
1769 .expect("Failed to append");
1770 oversized.sync(1).await.expect("Failed to sync");
1771 drop(oversized);
1772
1773 let (blob, _) = context
1775 .open(&cfg.index_partition, &1u64.to_be_bytes())
1776 .await
1777 .expect("Failed to open blob");
1778 blob.resize(10).await.expect("Failed to resize"); blob.sync().await.expect("Failed to sync");
1780 drop(blob);
1781
1782 let mut oversized: Oversized<_, TestEntry, TestValue> =
1784 Oversized::init(context.child("second"), cfg.clone())
1785 .await
1786 .expect("Failed to reinit");
1787
1788 assert!(oversized.get(1, 0).await.is_err());
1790
1791 let value: TestValue = [99; 16];
1793 let entry = TestEntry::new(100, 0, 0);
1794 let (pos, offset, size) = oversized
1795 .append(1, entry, &value)
1796 .await
1797 .expect("Failed to append after recovery");
1798 assert_eq!(pos, 0);
1799
1800 let retrieved = oversized.get(1, 0).await.expect("Failed to get");
1801 assert_eq!(retrieved.id, 100);
1802 let retrieved_value = oversized
1803 .get_value(1, offset, size)
1804 .await
1805 .expect("Failed to get value");
1806 assert_eq!(retrieved_value, value);
1807
1808 oversized.destroy().await.expect("Failed to destroy");
1809 });
1810 }
1811
1812 #[test_traced]
1813 fn test_recovery_crash_during_rewind_index_ahead() {
1814 let executor = deterministic::Runner::default();
1816 executor.start(|context| async move {
1817 let cfg = Config {
1821 index_partition: "test-index".into(),
1822 value_partition: "test-values".into(),
1823 index_page_cache: CacheRef::from_pooler(
1824 &context,
1825 NZU16!(TestEntry::SIZE as u16),
1826 NZUsize!(8),
1827 ),
1828 index_write_buffer: NZUsize!(1024),
1829 value_write_buffer: NZUsize!(1024),
1830 compression: None,
1831 codec_config: (),
1832 };
1833
1834 let mut oversized: Oversized<_, TestEntry, TestValue> =
1836 Oversized::init(context.child("first"), cfg.clone())
1837 .await
1838 .expect("Failed to init");
1839
1840 let mut locations = Vec::new();
1841 for i in 0..5u8 {
1842 let value: TestValue = [i; 16];
1843 let entry = TestEntry::new(i as u64, 0, 0);
1844 let loc = oversized
1845 .append(1, entry, &value)
1846 .await
1847 .expect("Failed to append");
1848 locations.push(loc);
1849 }
1850 oversized.sync(1).await.expect("Failed to sync");
1851 drop(oversized);
1852
1853 let (blob, _) = context
1856 .open(&cfg.index_partition, &1u64.to_be_bytes())
1857 .await
1858 .expect("Failed to open blob");
1859 let physical_page_size = (TestEntry::SIZE + 12) as u64;
1861 blob.resize(2 * physical_page_size)
1862 .await
1863 .expect("Failed to truncate");
1864 blob.sync().await.expect("Failed to sync");
1865 drop(blob);
1866
1867 let mut oversized: Oversized<_, TestEntry, TestValue> =
1869 Oversized::init(context.child("second"), cfg.clone())
1870 .await
1871 .expect("Failed to reinit");
1872
1873 for i in 0..2u8 {
1875 let entry = oversized.get(1, i as u64).await.expect("Failed to get");
1876 assert_eq!(entry.id, i as u64);
1877 }
1878
1879 assert!(oversized.get(1, 2).await.is_err());
1881
1882 let (pos, _, _) = oversized
1884 .append(1, TestEntry::new(100, 0, 0), &[100u8; 16])
1885 .await
1886 .expect("Failed to append");
1887 assert_eq!(pos, 2);
1888
1889 oversized.destroy().await.expect("Failed to destroy");
1890 });
1891 }
1892
1893 #[test_traced]
1894 fn test_recovery_crash_during_rewind_glob_ahead() {
1895 let executor = deterministic::Runner::default();
1897 executor.start(|context| async move {
1898 let cfg = test_cfg(&context);
1899
1900 let mut oversized: Oversized<_, TestEntry, TestValue> =
1902 Oversized::init(context.child("first"), cfg.clone())
1903 .await
1904 .expect("Failed to init");
1905
1906 let mut locations = Vec::new();
1907 for i in 0..5u8 {
1908 let value: TestValue = [i; 16];
1909 let entry = TestEntry::new(i as u64, 0, 0);
1910 let loc = oversized
1911 .append(1, entry, &value)
1912 .await
1913 .expect("Failed to append");
1914 locations.push(loc);
1915 }
1916 oversized.sync(1).await.expect("Failed to sync");
1917 drop(oversized);
1918
1919 let (blob, _) = context
1922 .open(&cfg.value_partition, &1u64.to_be_bytes())
1923 .await
1924 .expect("Failed to open blob");
1925 let keep_size = byte_end(locations[1].1, locations[1].2);
1926 blob.resize(keep_size).await.expect("Failed to truncate");
1927 blob.sync().await.expect("Failed to sync");
1928 drop(blob);
1929
1930 let mut oversized: Oversized<_, TestEntry, TestValue> =
1932 Oversized::init(context.child("second"), cfg.clone())
1933 .await
1934 .expect("Failed to reinit");
1935
1936 for i in 0..2u8 {
1938 let entry = oversized.get(1, i as u64).await.expect("Failed to get");
1939 assert_eq!(entry.id, i as u64);
1940 }
1941
1942 assert!(oversized.get(1, 2).await.is_err());
1944
1945 let value: TestValue = [99; 16];
1947 let entry = TestEntry::new(100, 0, 0);
1948 let (pos, offset, size) = oversized
1949 .append(1, entry, &value)
1950 .await
1951 .expect("Failed to append after recovery");
1952 assert_eq!(pos, 2);
1953
1954 let retrieved = oversized.get(1, 2).await.expect("Failed to get");
1955 assert_eq!(retrieved.id, 100);
1956 let retrieved_value = oversized
1957 .get_value(1, offset, size)
1958 .await
1959 .expect("Failed to get value");
1960 assert_eq!(retrieved_value, value);
1961
1962 oversized.destroy().await.expect("Failed to destroy");
1963 });
1964 }
1965
1966 #[test_traced]
1967 fn test_oversized_get_value_invalid_size() {
1968 let executor = deterministic::Runner::default();
1969 executor.start(|context| async move {
1970 let cfg = test_cfg(&context);
1971 let mut oversized: Oversized<_, TestEntry, TestValue> =
1972 Oversized::init(context, cfg).await.expect("Failed to init");
1973
1974 let value: TestValue = [42; 16];
1975 let entry = TestEntry::new(1, 0, 0);
1976 let (_, offset, _size) = oversized
1977 .append(1, entry, &value)
1978 .await
1979 .expect("Failed to append");
1980 oversized.sync(1).await.expect("Failed to sync");
1981
1982 assert!(oversized.get_value(1, offset, 0).await.is_err());
1984
1985 for size in 1..4u32 {
1988 let result = oversized.get_value(1, offset, size).await;
1989 assert!(
1990 matches!(
1991 result,
1992 Err(Error::Codec(_))
1993 | Err(Error::ChecksumMismatch(_, _))
1994 | Err(Error::Runtime(_))
1995 ),
1996 "expected error, got: {:?}",
1997 result
1998 );
1999 }
2000
2001 oversized.destroy().await.expect("Failed to destroy");
2002 });
2003 }
2004
2005 #[test_traced]
2006 fn test_oversized_get_value_wrong_size() {
2007 let executor = deterministic::Runner::default();
2008 executor.start(|context| async move {
2009 let cfg = test_cfg(&context);
2010 let mut oversized: Oversized<_, TestEntry, TestValue> =
2011 Oversized::init(context, cfg).await.expect("Failed to init");
2012
2013 let value: TestValue = [42; 16];
2014 let entry = TestEntry::new(1, 0, 0);
2015 let (_, offset, correct_size) = oversized
2016 .append(1, entry, &value)
2017 .await
2018 .expect("Failed to append");
2019 oversized.sync(1).await.expect("Failed to sync");
2020
2021 let result = oversized.get_value(1, offset, correct_size - 1).await;
2024 assert!(
2025 matches!(
2026 result,
2027 Err(Error::Codec(_)) | Err(Error::ChecksumMismatch(_, _))
2028 ),
2029 "expected Codec or ChecksumMismatch error, got: {:?}",
2030 result
2031 );
2032
2033 oversized.destroy().await.expect("Failed to destroy");
2034 });
2035 }
2036
2037 #[test_traced]
2038 fn test_recovery_values_has_orphan_section() {
2039 let executor = deterministic::Runner::default();
2040 executor.start(|context| async move {
2041 let cfg = test_cfg(&context);
2042
2043 let mut oversized: Oversized<_, TestEntry, TestValue> =
2045 Oversized::init(context.child("first"), cfg.clone())
2046 .await
2047 .expect("Failed to init");
2048
2049 for section in 1u64..=2 {
2050 let value: TestValue = [section as u8; 16];
2051 let entry = TestEntry::new(section, 0, 0);
2052 oversized
2053 .append(section, entry, &value)
2054 .await
2055 .expect("Failed to append");
2056 oversized.sync(section).await.expect("Failed to sync");
2057 }
2058 drop(oversized);
2059
2060 let glob_cfg = GlobConfig {
2062 partition: cfg.value_partition.clone(),
2063 compression: cfg.compression,
2064 codec_config: (),
2065 write_buffer: cfg.value_write_buffer,
2066 };
2067 let mut glob: Glob<_, TestValue> = Glob::init(context.child("glob"), glob_cfg)
2068 .await
2069 .expect("Failed to init glob");
2070 let orphan_value: TestValue = [99; 16];
2071 glob.append(3, &orphan_value)
2072 .await
2073 .expect("Failed to append orphan");
2074 glob.sync(3).await.expect("Failed to sync glob");
2075 drop(glob);
2076
2077 let oversized: Oversized<_, TestEntry, TestValue> =
2079 Oversized::init(context.child("second"), cfg.clone())
2080 .await
2081 .expect("Failed to reinit");
2082
2083 assert!(oversized.get(1, 0).await.is_ok());
2085 assert!(oversized.get(2, 0).await.is_ok());
2086
2087 assert_eq!(oversized.newest_section(), Some(2));
2089
2090 oversized.destroy().await.expect("Failed to destroy");
2091 });
2092 }
2093
2094 #[test_traced]
2095 fn test_recovery_values_has_multiple_orphan_sections() {
2096 let executor = deterministic::Runner::default();
2097 executor.start(|context| async move {
2098 let cfg = test_cfg(&context);
2099
2100 let mut oversized: Oversized<_, TestEntry, TestValue> =
2102 Oversized::init(context.child("first"), cfg.clone())
2103 .await
2104 .expect("Failed to init");
2105
2106 let value: TestValue = [1; 16];
2107 let entry = TestEntry::new(1, 0, 0);
2108 oversized
2109 .append(1, entry, &value)
2110 .await
2111 .expect("Failed to append");
2112 oversized.sync(1).await.expect("Failed to sync");
2113 drop(oversized);
2114
2115 let glob_cfg = GlobConfig {
2117 partition: cfg.value_partition.clone(),
2118 compression: cfg.compression,
2119 codec_config: (),
2120 write_buffer: cfg.value_write_buffer,
2121 };
2122 let mut glob: Glob<_, TestValue> = Glob::init(context.child("glob"), glob_cfg)
2123 .await
2124 .expect("Failed to init glob");
2125
2126 for section in 2u64..=4 {
2127 let orphan_value: TestValue = [section as u8; 16];
2128 glob.append(section, &orphan_value)
2129 .await
2130 .expect("Failed to append orphan");
2131 glob.sync(section).await.expect("Failed to sync glob");
2132 }
2133 drop(glob);
2134
2135 let oversized: Oversized<_, TestEntry, TestValue> =
2137 Oversized::init(context.child("second"), cfg.clone())
2138 .await
2139 .expect("Failed to reinit");
2140
2141 assert!(oversized.get(1, 0).await.is_ok());
2143
2144 assert_eq!(oversized.newest_section(), Some(1));
2146
2147 oversized.destroy().await.expect("Failed to destroy");
2148 });
2149 }
2150
2151 #[test_traced]
2152 fn test_recovery_index_empty_but_values_exist() {
2153 let executor = deterministic::Runner::default();
2154 executor.start(|context| async move {
2155 let cfg = test_cfg(&context);
2156
2157 let glob_cfg = GlobConfig {
2159 partition: cfg.value_partition.clone(),
2160 compression: cfg.compression,
2161 codec_config: (),
2162 write_buffer: cfg.value_write_buffer,
2163 };
2164 let mut glob: Glob<_, TestValue> = Glob::init(context.child("glob"), glob_cfg)
2165 .await
2166 .expect("Failed to init glob");
2167
2168 for section in 1u64..=3 {
2169 let orphan_value: TestValue = [section as u8; 16];
2170 glob.append(section, &orphan_value)
2171 .await
2172 .expect("Failed to append orphan");
2173 glob.sync(section).await.expect("Failed to sync glob");
2174 }
2175 drop(glob);
2176
2177 let oversized: Oversized<_, TestEntry, TestValue> =
2179 Oversized::init(context.child("first"), cfg.clone())
2180 .await
2181 .expect("Failed to init");
2182
2183 assert_eq!(oversized.newest_section(), None);
2185 assert_eq!(oversized.oldest_section(), None);
2186
2187 oversized.destroy().await.expect("Failed to destroy");
2188 });
2189 }
2190
2191 #[test_traced]
2192 fn test_recovery_orphan_section_append_after() {
2193 let executor = deterministic::Runner::default();
2194 executor.start(|context| async move {
2195 let cfg = test_cfg(&context);
2196
2197 let mut oversized: Oversized<_, TestEntry, TestValue> =
2199 Oversized::init(context.child("first"), cfg.clone())
2200 .await
2201 .expect("Failed to init");
2202
2203 let value: TestValue = [1; 16];
2204 let entry = TestEntry::new(1, 0, 0);
2205 let (_, offset1, size1) = oversized
2206 .append(1, entry, &value)
2207 .await
2208 .expect("Failed to append");
2209 oversized.sync(1).await.expect("Failed to sync");
2210 drop(oversized);
2211
2212 let glob_cfg = GlobConfig {
2214 partition: cfg.value_partition.clone(),
2215 compression: cfg.compression,
2216 codec_config: (),
2217 write_buffer: cfg.value_write_buffer,
2218 };
2219 let mut glob: Glob<_, TestValue> = Glob::init(context.child("glob"), glob_cfg)
2220 .await
2221 .expect("Failed to init glob");
2222
2223 for section in 2u64..=3 {
2224 let orphan_value: TestValue = [section as u8; 16];
2225 glob.append(section, &orphan_value)
2226 .await
2227 .expect("Failed to append orphan");
2228 glob.sync(section).await.expect("Failed to sync glob");
2229 }
2230 drop(glob);
2231
2232 let mut oversized: Oversized<_, TestEntry, TestValue> =
2234 Oversized::init(context.child("second"), cfg.clone())
2235 .await
2236 .expect("Failed to reinit");
2237
2238 let entry = oversized.get(1, 0).await.expect("Failed to get");
2240 assert_eq!(entry.id, 1);
2241 let value = oversized
2242 .get_value(1, offset1, size1)
2243 .await
2244 .expect("Failed to get value");
2245 assert_eq!(value, [1; 16]);
2246
2247 let new_value: TestValue = [42; 16];
2249 let new_entry = TestEntry::new(42, 0, 0);
2250 let (pos, offset, size) = oversized
2251 .append(2, new_entry, &new_value)
2252 .await
2253 .expect("Failed to append after recovery");
2254 assert_eq!(pos, 0);
2255
2256 let retrieved = oversized.get(2, 0).await.expect("Failed to get");
2258 assert_eq!(retrieved.id, 42);
2259 let retrieved_value = oversized
2260 .get_value(2, offset, size)
2261 .await
2262 .expect("Failed to get value");
2263 assert_eq!(retrieved_value, new_value);
2264
2265 oversized.sync(2).await.expect("Failed to sync");
2267 drop(oversized);
2268
2269 let oversized: Oversized<_, TestEntry, TestValue> =
2270 Oversized::init(context.child("third"), cfg)
2271 .await
2272 .expect("Failed to reinit after append");
2273
2274 assert!(oversized.get(1, 0).await.is_ok());
2276 assert!(oversized.get(2, 0).await.is_ok());
2277 assert_eq!(oversized.newest_section(), Some(2));
2278
2279 oversized.destroy().await.expect("Failed to destroy");
2280 });
2281 }
2282
2283 #[test_traced]
2284 fn test_recovery_no_orphan_sections() {
2285 let executor = deterministic::Runner::default();
2286 executor.start(|context| async move {
2287 let cfg = test_cfg(&context);
2288
2289 let mut oversized: Oversized<_, TestEntry, TestValue> =
2291 Oversized::init(context.child("first"), cfg.clone())
2292 .await
2293 .expect("Failed to init");
2294
2295 for section in 1u64..=3 {
2296 let value: TestValue = [section as u8; 16];
2297 let entry = TestEntry::new(section, 0, 0);
2298 oversized
2299 .append(section, entry, &value)
2300 .await
2301 .expect("Failed to append");
2302 oversized.sync(section).await.expect("Failed to sync");
2303 }
2304 drop(oversized);
2305
2306 let oversized: Oversized<_, TestEntry, TestValue> =
2308 Oversized::init(context.child("second"), cfg)
2309 .await
2310 .expect("Failed to reinit");
2311
2312 for section in 1u64..=3 {
2314 let entry = oversized.get(section, 0).await.expect("Failed to get");
2315 assert_eq!(entry.id, section);
2316 }
2317 assert_eq!(oversized.newest_section(), Some(3));
2318
2319 oversized.destroy().await.expect("Failed to destroy");
2320 });
2321 }
2322
2323 #[test_traced]
2324 fn test_recovery_orphan_with_empty_index_section() {
2325 let executor = deterministic::Runner::default();
2326 executor.start(|context| async move {
2327 let cfg = test_cfg(&context);
2328
2329 let mut oversized: Oversized<_, TestEntry, TestValue> =
2331 Oversized::init(context.child("first"), cfg.clone())
2332 .await
2333 .expect("Failed to init");
2334
2335 let value: TestValue = [1; 16];
2336 let entry = TestEntry::new(1, 0, 0);
2337 oversized
2338 .append(1, entry, &value)
2339 .await
2340 .expect("Failed to append");
2341 oversized.sync(1).await.expect("Failed to sync");
2342 drop(oversized);
2343
2344 let glob_cfg = GlobConfig {
2346 partition: cfg.value_partition.clone(),
2347 compression: cfg.compression,
2348 codec_config: (),
2349 write_buffer: cfg.value_write_buffer,
2350 };
2351 let mut glob: Glob<_, TestValue> = Glob::init(context.child("glob"), glob_cfg)
2352 .await
2353 .expect("Failed to init glob");
2354 let orphan_value: TestValue = [2; 16];
2355 glob.append(2, &orphan_value)
2356 .await
2357 .expect("Failed to append orphan");
2358 glob.sync(2).await.expect("Failed to sync glob");
2359 drop(glob);
2360
2361 let (blob, _) = context
2363 .open(&cfg.index_partition, &1u64.to_be_bytes())
2364 .await
2365 .expect("Failed to open blob");
2366 blob.resize(0).await.expect("Failed to truncate");
2367 blob.sync().await.expect("Failed to sync");
2368 drop(blob);
2369
2370 let oversized: Oversized<_, TestEntry, TestValue> =
2372 Oversized::init(context.child("second"), cfg)
2373 .await
2374 .expect("Failed to reinit");
2375
2376 assert!(oversized.get(1, 0).await.is_err());
2378
2379 assert_eq!(oversized.newest_section(), Some(1));
2381
2382 oversized.destroy().await.expect("Failed to destroy");
2383 });
2384 }
2385
2386 #[test_traced]
2387 fn test_recovery_orphan_sections_with_gaps() {
2388 let executor = deterministic::Runner::default();
2391 executor.start(|context| async move {
2392 let cfg = test_cfg(&context);
2393
2394 let mut oversized: Oversized<_, TestEntry, TestValue> =
2396 Oversized::init(context.child("first"), cfg.clone())
2397 .await
2398 .expect("Failed to init");
2399
2400 for section in [1u64, 3, 5] {
2401 let value: TestValue = [section as u8; 16];
2402 let entry = TestEntry::new(section, 0, 0);
2403 oversized
2404 .append(section, entry, &value)
2405 .await
2406 .expect("Failed to append");
2407 oversized.sync(section).await.expect("Failed to sync");
2408 }
2409 drop(oversized);
2410
2411 let glob_cfg = GlobConfig {
2413 partition: cfg.value_partition.clone(),
2414 compression: cfg.compression,
2415 codec_config: (),
2416 write_buffer: cfg.value_write_buffer,
2417 };
2418 let mut glob: Glob<_, TestValue> = Glob::init(context.child("glob"), glob_cfg)
2419 .await
2420 .expect("Failed to init glob");
2421
2422 for section in [2u64, 4, 6] {
2423 let orphan_value: TestValue = [section as u8; 16];
2424 glob.append(section, &orphan_value)
2425 .await
2426 .expect("Failed to append orphan");
2427 glob.sync(section).await.expect("Failed to sync glob");
2428 }
2429 drop(glob);
2430
2431 let oversized: Oversized<_, TestEntry, TestValue> =
2433 Oversized::init(context.child("second"), cfg)
2434 .await
2435 .expect("Failed to reinit");
2436
2437 for section in [1u64, 3, 5] {
2439 let entry = oversized.get(section, 0).await.expect("Failed to get");
2440 assert_eq!(entry.id, section);
2441 }
2442
2443 assert_eq!(oversized.oldest_section(), Some(1));
2445 assert_eq!(oversized.newest_section(), Some(5));
2446
2447 oversized.destroy().await.expect("Failed to destroy");
2448 });
2449 }
2450
2451 #[test_traced]
2452 fn test_recovery_glob_trailing_garbage_truncated() {
2453 let executor = deterministic::Runner::default();
2457 executor.start(|context| async move {
2458 let cfg = test_cfg(&context);
2459
2460 let mut oversized: Oversized<_, TestEntry, TestValue> =
2462 Oversized::init(context.child("first"), cfg.clone())
2463 .await
2464 .expect("Failed to init");
2465
2466 let mut locations = Vec::new();
2468 for i in 0..2u8 {
2469 let value: TestValue = [i; 16];
2470 let entry = TestEntry::new(i as u64, 0, 0);
2471 let loc = oversized
2472 .append(1, entry, &value)
2473 .await
2474 .expect("Failed to append");
2475 locations.push(loc);
2476 }
2477 oversized.sync(1).await.expect("Failed to sync");
2478
2479 let expected_next_offset = byte_end(locations[1].1, locations[1].2);
2481 drop(oversized);
2482
2483 let (blob, size) = context
2485 .open(&cfg.value_partition, &1u64.to_be_bytes())
2486 .await
2487 .expect("Failed to open blob");
2488 assert_eq!(size, expected_next_offset);
2489
2490 let garbage = vec![0xDE; 100];
2492 blob.write_at_sync(size, garbage)
2493 .await
2494 .expect("Failed to write garbage");
2495 drop(blob);
2496
2497 let (blob, new_size) = context
2499 .open(&cfg.value_partition, &1u64.to_be_bytes())
2500 .await
2501 .expect("Failed to open blob");
2502 assert_eq!(new_size, expected_next_offset + 100);
2503 drop(blob);
2504
2505 let mut oversized: Oversized<_, TestEntry, TestValue> =
2507 Oversized::init(context.child("second"), cfg.clone())
2508 .await
2509 .expect("Failed to reinit");
2510
2511 for i in 0..2u8 {
2513 let entry = oversized.get(1, i as u64).await.expect("Failed to get");
2514 assert_eq!(entry.id, i as u64);
2515 }
2516
2517 let new_value: TestValue = [99; 16];
2519 let new_entry = TestEntry::new(99, 0, 0);
2520 let (pos, offset, _size) = oversized
2521 .append(1, new_entry, &new_value)
2522 .await
2523 .expect("Failed to append after recovery");
2524
2525 assert_eq!(pos, 2);
2527
2528 assert_eq!(offset, expected_next_offset);
2530
2531 let retrieved = oversized.get(1, 2).await.expect("Failed to get new entry");
2533 assert_eq!(retrieved.id, 99);
2534
2535 oversized.destroy().await.expect("Failed to destroy");
2536 });
2537 }
2538
2539 #[test_traced]
2540 fn test_recovery_entry_with_overflow_offset() {
2541 let executor = deterministic::Runner::default();
2544 executor.start(|context| async move {
2545 let cfg = Config {
2547 index_partition: "test-index".into(),
2548 value_partition: "test-values".into(),
2549 index_page_cache: CacheRef::from_pooler(
2550 &context,
2551 NZU16!(TestEntry::SIZE as u16),
2552 NZUsize!(8),
2553 ),
2554 index_write_buffer: NZUsize!(1024),
2555 value_write_buffer: NZUsize!(1024),
2556 compression: None,
2557 codec_config: (),
2558 };
2559
2560 let mut oversized: Oversized<_, TestEntry, TestValue> =
2562 Oversized::init(context.child("first"), cfg.clone())
2563 .await
2564 .expect("Failed to init");
2565
2566 let value: TestValue = [1; 16];
2567 let entry = TestEntry::new(1, 0, 0);
2568 oversized
2569 .append(1, entry, &value)
2570 .await
2571 .expect("Failed to append");
2572 oversized.sync(1).await.expect("Failed to sync");
2573 drop(oversized);
2574
2575 let (blob, _) = context
2579 .open(&cfg.index_partition, &1u64.to_be_bytes())
2580 .await
2581 .expect("Failed to open blob");
2582
2583 let mut entry_data = Vec::new();
2585 1u64.write(&mut entry_data); (u64::MAX - 10).write(&mut entry_data); 100u32.write(&mut entry_data); assert_eq!(entry_data.len(), TestEntry::SIZE);
2589
2590 let crc = Crc32::checksum(&entry_data);
2593 let len1 = TestEntry::SIZE as u16;
2594 let mut crc_record = Vec::new();
2595 crc_record.extend_from_slice(&len1.to_be_bytes()); crc_record.extend_from_slice(&crc.to_be_bytes()); crc_record.extend_from_slice(&0u16.to_be_bytes()); crc_record.extend_from_slice(&0u32.to_be_bytes()); assert_eq!(crc_record.len(), 12);
2600
2601 let mut page = entry_data;
2603 page.extend_from_slice(&crc_record);
2604 blob.write_at_sync(0, page)
2605 .await
2606 .expect("Failed to write corrupted page");
2607 drop(blob);
2608
2609 let mut oversized: Oversized<_, TestEntry, TestValue> =
2612 Oversized::init(context.child("second"), cfg.clone())
2613 .await
2614 .expect("Failed to reinit");
2615
2616 assert!(oversized.get(1, 0).await.is_err());
2618
2619 let new_value: TestValue = [99; 16];
2621 let new_entry = TestEntry::new(99, 0, 0);
2622 let (pos, new_offset, _) = oversized
2623 .append(1, new_entry, &new_value)
2624 .await
2625 .expect("Failed to append after recovery");
2626
2627 assert_eq!(pos, 0);
2629 assert_eq!(new_offset, 0);
2631
2632 oversized.destroy().await.expect("Failed to destroy");
2633 });
2634 }
2635
2636 #[test_traced]
2637 fn test_empty_section_persistence() {
2638 let executor = deterministic::Runner::default();
2641 executor.start(|context| async move {
2642 let cfg = test_cfg(&context);
2643
2644 let mut oversized: Oversized<_, TestEntry, TestValue> =
2646 Oversized::init(context.child("first"), cfg.clone())
2647 .await
2648 .expect("Failed to init");
2649
2650 for i in 0..3u8 {
2651 let value: TestValue = [i; 16];
2652 let entry = TestEntry::new(i as u64, 0, 0);
2653 oversized
2654 .append(1, entry, &value)
2655 .await
2656 .expect("Failed to append");
2657 }
2658 oversized.sync(1).await.expect("Failed to sync");
2659
2660 let value2: TestValue = [10; 16];
2662 let entry2 = TestEntry::new(10, 0, 0);
2663 oversized
2664 .append(2, entry2, &value2)
2665 .await
2666 .expect("Failed to append to section 2");
2667 oversized.sync(2).await.expect("Failed to sync section 2");
2668 drop(oversized);
2669
2670 let (blob, _) = context
2672 .open(&cfg.index_partition, &1u64.to_be_bytes())
2673 .await
2674 .expect("Failed to open blob");
2675 blob.resize(0).await.expect("Failed to truncate");
2676 blob.sync().await.expect("Failed to sync");
2677 drop(blob);
2678
2679 let mut oversized: Oversized<_, TestEntry, TestValue> =
2681 Oversized::init(context.child("second"), cfg.clone())
2682 .await
2683 .expect("Failed to reinit");
2684
2685 assert!(oversized.get(1, 0).await.is_err());
2687
2688 let entry = oversized.get(2, 0).await.expect("Failed to get section 2");
2690 assert_eq!(entry.id, 10);
2691
2692 assert_eq!(oversized.oldest_section(), Some(1));
2694
2695 let new_value: TestValue = [99; 16];
2701 let new_entry = TestEntry::new(99, 0, 0);
2702 let (pos, offset, size) = oversized
2703 .append(1, new_entry, &new_value)
2704 .await
2705 .expect("Failed to append to empty section");
2706 assert_eq!(pos, 0);
2707 assert!(offset > 0);
2709 oversized.sync(1).await.expect("Failed to sync");
2710
2711 let entry = oversized.get(1, 0).await.expect("Failed to get");
2713 assert_eq!(entry.id, 99);
2714 let value = oversized
2715 .get_value(1, offset, size)
2716 .await
2717 .expect("Failed to get value");
2718 assert_eq!(value, new_value);
2719
2720 drop(oversized);
2721
2722 let oversized: Oversized<_, TestEntry, TestValue> =
2724 Oversized::init(context.child("third"), cfg.clone())
2725 .await
2726 .expect("Failed to reinit again");
2727
2728 let entry = oversized.get(1, 0).await.expect("Failed to get");
2730 assert_eq!(entry.id, 99);
2731
2732 let entry = oversized.get(2, 0).await.expect("Failed to get section 2");
2734 assert_eq!(entry.id, 10);
2735
2736 oversized.destroy().await.expect("Failed to destroy");
2737 });
2738 }
2739
2740 #[test_traced]
2741 fn test_get_value_size_equals_crc_size() {
2742 let executor = deterministic::Runner::default();
2745 executor.start(|context| async move {
2746 let cfg = test_cfg(&context);
2747 let mut oversized: Oversized<_, TestEntry, TestValue> =
2748 Oversized::init(context, cfg).await.expect("Failed to init");
2749
2750 let value: TestValue = [42; 16];
2751 let entry = TestEntry::new(1, 0, 0);
2752 let (_, offset, _) = oversized
2753 .append(1, entry, &value)
2754 .await
2755 .expect("Failed to append");
2756 oversized.sync(1).await.expect("Failed to sync");
2757
2758 let result = oversized.get_value(1, offset, 4).await;
2761 assert!(result.is_err());
2762
2763 oversized.destroy().await.expect("Failed to destroy");
2764 });
2765 }
2766
2767 #[test_traced]
2768 fn test_get_value_size_just_over_crc() {
2769 let executor = deterministic::Runner::default();
2772 executor.start(|context| async move {
2773 let cfg = test_cfg(&context);
2774 let mut oversized: Oversized<_, TestEntry, TestValue> =
2775 Oversized::init(context, cfg).await.expect("Failed to init");
2776
2777 let value: TestValue = [42; 16];
2778 let entry = TestEntry::new(1, 0, 0);
2779 let (_, offset, _) = oversized
2780 .append(1, entry, &value)
2781 .await
2782 .expect("Failed to append");
2783 oversized.sync(1).await.expect("Failed to sync");
2784
2785 let result = oversized.get_value(1, offset, 5).await;
2788 assert!(result.is_err());
2789
2790 oversized.destroy().await.expect("Failed to destroy");
2791 });
2792 }
2793
2794 #[test_traced]
2795 fn test_recovery_maximum_section_numbers() {
2796 let executor = deterministic::Runner::default();
2799 executor.start(|context| async move {
2800 let cfg = test_cfg(&context);
2801
2802 let large_sections = [u64::MAX - 3, u64::MAX - 2, u64::MAX - 1];
2804
2805 let mut oversized: Oversized<_, TestEntry, TestValue> =
2807 Oversized::init(context.child("first"), cfg.clone())
2808 .await
2809 .expect("Failed to init");
2810
2811 let mut locations = Vec::new();
2812 for §ion in &large_sections {
2813 let value: TestValue = [(section & 0xFF) as u8; 16];
2814 let entry = TestEntry::new(section, 0, 0);
2815 let loc = oversized
2816 .append(section, entry, &value)
2817 .await
2818 .expect("Failed to append");
2819 locations.push((section, loc));
2820 oversized.sync(section).await.expect("Failed to sync");
2821 }
2822 drop(oversized);
2823
2824 let middle_section = large_sections[1];
2826 let (blob, size) = context
2827 .open(&cfg.value_partition, &middle_section.to_be_bytes())
2828 .await
2829 .expect("Failed to open blob");
2830 blob.resize(size / 2).await.expect("Failed to truncate");
2831 blob.sync().await.expect("Failed to sync");
2832 drop(blob);
2833
2834 let mut oversized: Oversized<_, TestEntry, TestValue> =
2836 Oversized::init(context.child("second"), cfg.clone())
2837 .await
2838 .expect("Failed to reinit");
2839
2840 let entry = oversized
2842 .get(large_sections[0], 0)
2843 .await
2844 .expect("Failed to get first section");
2845 assert_eq!(entry.id, large_sections[0]);
2846
2847 let entry = oversized
2848 .get(large_sections[2], 0)
2849 .await
2850 .expect("Failed to get last section");
2851 assert_eq!(entry.id, large_sections[2]);
2852
2853 assert!(oversized.get(middle_section, 0).await.is_err());
2855
2856 let new_value: TestValue = [0xAB; 16];
2858 let new_entry = TestEntry::new(999, 0, 0);
2859 oversized
2860 .append(middle_section, new_entry, &new_value)
2861 .await
2862 .expect("Failed to append after recovery");
2863
2864 oversized.destroy().await.expect("Failed to destroy");
2865 });
2866 }
2867
2868 #[test_traced]
2869 fn test_recovery_crash_during_recovery_rewind() {
2870 let executor = deterministic::Runner::default();
2874 executor.start(|context| async move {
2875 let cfg = test_cfg(&context);
2876
2877 let mut oversized: Oversized<_, TestEntry, TestValue> =
2879 Oversized::init(context.child("first"), cfg.clone())
2880 .await
2881 .expect("Failed to init");
2882
2883 let mut locations = Vec::new();
2884 for i in 0..5u8 {
2885 let value: TestValue = [i; 16];
2886 let entry = TestEntry::new(i as u64, 0, 0);
2887 let loc = oversized
2888 .append(1, entry, &value)
2889 .await
2890 .expect("Failed to append");
2891 locations.push(loc);
2892 }
2893 oversized.sync(1).await.expect("Failed to sync");
2894 drop(oversized);
2895
2896 let (blob, _) = context
2898 .open(&cfg.value_partition, &1u64.to_be_bytes())
2899 .await
2900 .expect("Failed to open blob");
2901 let keep_size = byte_end(locations[2].1, locations[2].2);
2902 blob.resize(keep_size).await.expect("Failed to truncate");
2903 blob.sync().await.expect("Failed to sync");
2904 drop(blob);
2905
2906 let chunk_size = FixedJournal::<deterministic::Context, TestEntry>::CHUNK_SIZE as u64;
2911 let (index_blob, _) = context
2912 .open(&cfg.index_partition, &1u64.to_be_bytes())
2913 .await
2914 .expect("Failed to open index blob");
2915 let partial_rewind_size = 4 * chunk_size; index_blob
2917 .resize(partial_rewind_size)
2918 .await
2919 .expect("Failed to resize");
2920 index_blob.sync().await.expect("Failed to sync");
2921 drop(index_blob);
2922
2923 let mut oversized: Oversized<_, TestEntry, TestValue> =
2926 Oversized::init(context.child("second"), cfg.clone())
2927 .await
2928 .expect("Failed to reinit after nested crash");
2929
2930 for i in 0..3u8 {
2932 let entry = oversized.get(1, i as u64).await.expect("Failed to get");
2933 assert_eq!(entry.id, i as u64);
2934
2935 let (_, offset, size) = locations[i as usize];
2936 let value = oversized
2937 .get_value(1, offset, size)
2938 .await
2939 .expect("Failed to get value");
2940 assert_eq!(value, [i; 16]);
2941 }
2942
2943 assert!(oversized.get(1, 3).await.is_err());
2945
2946 let new_value: TestValue = [0xFF; 16];
2948 let new_entry = TestEntry::new(100, 0, 0);
2949 let (pos, offset, _size) = oversized
2950 .append(1, new_entry, &new_value)
2951 .await
2952 .expect("Failed to append");
2953 assert_eq!(pos, 3); assert_eq!(offset, byte_end(locations[2].1, locations[2].2));
2957
2958 oversized.destroy().await.expect("Failed to destroy");
2959 });
2960 }
2961
2962 #[test_traced]
2963 fn test_recovery_crash_during_orphan_cleanup() {
2964 let executor = deterministic::Runner::default();
2967 executor.start(|context| async move {
2968 let cfg = test_cfg(&context);
2969
2970 let mut oversized: Oversized<_, TestEntry, TestValue> =
2972 Oversized::init(context.child("first"), cfg.clone())
2973 .await
2974 .expect("Failed to init");
2975
2976 let value: TestValue = [1; 16];
2977 let entry = TestEntry::new(1, 0, 0);
2978 let (_, offset1, size1) = oversized
2979 .append(1, entry, &value)
2980 .await
2981 .expect("Failed to append");
2982 oversized.sync(1).await.expect("Failed to sync");
2983 drop(oversized);
2984
2985 let glob_cfg = GlobConfig {
2987 partition: cfg.value_partition.clone(),
2988 compression: cfg.compression,
2989 codec_config: (),
2990 write_buffer: cfg.value_write_buffer,
2991 };
2992 let mut glob: Glob<_, TestValue> = Glob::init(context.child("glob"), glob_cfg)
2993 .await
2994 .expect("Failed to init glob");
2995
2996 for section in 2u64..=4 {
2997 let orphan_value: TestValue = [section as u8; 16];
2998 glob.append(section, &orphan_value)
2999 .await
3000 .expect("Failed to append orphan");
3001 glob.sync(section).await.expect("Failed to sync glob");
3002 }
3003 drop(glob);
3004
3005 context
3008 .remove(&cfg.value_partition, Some(&2u64.to_be_bytes()))
3009 .await
3010 .expect("Failed to remove section 2");
3011
3012 let mut oversized: Oversized<_, TestEntry, TestValue> =
3014 Oversized::init(context.child("second"), cfg.clone())
3015 .await
3016 .expect("Failed to reinit");
3017
3018 let entry = oversized.get(1, 0).await.expect("Failed to get");
3020 assert_eq!(entry.id, 1);
3021 let value = oversized
3022 .get_value(1, offset1, size1)
3023 .await
3024 .expect("Failed to get value");
3025 assert_eq!(value, [1; 16]);
3026
3027 assert_eq!(oversized.oldest_section(), Some(1));
3029 assert_eq!(oversized.newest_section(), Some(1));
3030
3031 let new_value: TestValue = [42; 16];
3033 let new_entry = TestEntry::new(42, 0, 0);
3034 let (pos, _, _) = oversized
3035 .append(2, new_entry, &new_value)
3036 .await
3037 .expect("Failed to append to section 2");
3038 assert_eq!(pos, 0); oversized.destroy().await.expect("Failed to destroy");
3041 });
3042 }
3043
3044 #[test_traced]
3045 fn test_rewind_to_zero_index_size() {
3046 let executor = deterministic::Runner::default();
3047 executor.start(|context| async move {
3048 let cfg = test_cfg(&context);
3049 let mut oversized: Oversized<_, TestEntry, TestValue> =
3050 Oversized::init(context, cfg).await.expect("Failed to init");
3051
3052 let value: TestValue = [1; 16];
3053 let entry = TestEntry::new(1, 0, 0);
3054 oversized
3055 .append(0, entry, &value)
3056 .await
3057 .expect("Failed to append");
3058 oversized.sync(0).await.expect("Failed to sync");
3059
3060 oversized
3061 .rewind(0, 0)
3062 .await
3063 .expect("rewind to zero index_size must not fail");
3064
3065 assert_eq!(oversized.last(0).await.unwrap(), None);
3066 assert_eq!(oversized.size(0).unwrap(), 0);
3067 assert_eq!(oversized.value_size(0).await.unwrap(), 0);
3068
3069 oversized.destroy().await.expect("Failed to destroy");
3070 });
3071 }
3072
3073 #[test_traced]
3074 fn test_rewind_to_zero_on_missing_section() {
3075 let executor = deterministic::Runner::default();
3076 executor.start(|context| async move {
3077 let cfg = test_cfg(&context);
3078 let mut oversized: Oversized<_, TestEntry, TestValue> =
3079 Oversized::init(context, cfg).await.expect("Failed to init");
3080
3081 oversized
3082 .rewind(0, 0)
3083 .await
3084 .expect("rewind on missing section must not fail");
3085
3086 assert!(matches!(
3087 oversized.last(0).await,
3088 Err(Error::SectionOutOfRange(0))
3089 ));
3090 assert_eq!(oversized.value_size(0).await.unwrap(), 0);
3091
3092 oversized.destroy().await.expect("Failed to destroy");
3093 });
3094 }
3095
3096 #[test_traced]
3097 fn test_rewind_nonzero_on_missing_section_errors() {
3098 let executor = deterministic::Runner::default();
3099 executor.start(|context| async move {
3100 let cfg = test_cfg(&context);
3101 let mut oversized: Oversized<_, TestEntry, TestValue> =
3102 Oversized::init(context, cfg).await.expect("Failed to init");
3103
3104 let result = oversized.rewind(0, 1).await;
3105 assert!(
3106 matches!(result, Err(Error::SectionOutOfRange(0))),
3107 "nonzero index_size on missing section must fail, got: {result:?}"
3108 );
3109
3110 oversized.destroy().await.expect("Failed to destroy");
3111 });
3112 }
3113
3114 #[test_traced]
3115 fn test_rewind_section_nonzero_on_missing_section_errors() {
3116 let executor = deterministic::Runner::default();
3117 executor.start(|context| async move {
3118 let cfg = test_cfg(&context);
3119 let mut oversized: Oversized<_, TestEntry, TestValue> =
3120 Oversized::init(context, cfg).await.expect("Failed to init");
3121
3122 let result = oversized.rewind_section(0, 1).await;
3123 assert!(
3124 matches!(result, Err(Error::SectionOutOfRange(0))),
3125 "nonzero index_size on missing section must fail, got: {result:?}"
3126 );
3127
3128 oversized.destroy().await.expect("Failed to destroy");
3129 });
3130 }
3131
3132 #[test_traced]
3133 fn test_last_pruned_section_returns_error() {
3134 let executor = deterministic::Runner::default();
3135 executor.start(|context| async move {
3136 let cfg = test_cfg(&context);
3137 let mut oversized: Oversized<_, TestEntry, TestValue> =
3138 Oversized::init(context, cfg).await.expect("Failed to init");
3139
3140 let value: TestValue = [1; 16];
3141 oversized
3142 .append(0, TestEntry::new(1, 0, 0), &value)
3143 .await
3144 .expect("Failed to append");
3145 oversized
3146 .append(1, TestEntry::new(2, 0, 0), &value)
3147 .await
3148 .expect("Failed to append");
3149 oversized.sync_all().await.expect("Failed to sync");
3150
3151 oversized.prune(1).await.expect("Failed to prune");
3152
3153 assert!(matches!(
3154 oversized.last(0).await,
3155 Err(Error::AlreadyPrunedToSection(1))
3156 ));
3157 assert!(oversized.last(1).await.unwrap().is_some());
3158
3159 oversized.destroy().await.expect("Failed to destroy");
3160 });
3161 }
3162}