1use super::{
60 fixed::{
61 Config as FixedConfig, Journal as FixedJournal, RecoveryPreflight, Replay as FixedReplay,
62 },
63 glob::{Config as GlobConfig, Glob},
64};
65use crate::{
66 Context, SyncCompletion,
67 journal::{Error, durability::Barrier},
68 metadata::{Config as MetadataConfig, Metadata},
69};
70use commonware_codec::{Codec, CodecFixed, CodecShared};
71use commonware_runtime::{Error as RError, Handle, ReadOptions};
72use commonware_utils::sequence::U64 as SectionKey;
73use futures::{FutureExt as _, future::try_join};
74use std::{
75 collections::{BTreeMap, BTreeSet, HashSet},
76 num::NonZeroUsize,
77};
78use tracing::{debug, warn};
79
80pub trait Record: CodecFixed<Cfg = ()> + Clone {
85 fn value_location(&self) -> (u64, u32);
87
88 fn with_location(self, offset: u64, size: u32) -> Self;
92}
93
94#[derive(Clone)]
96pub struct Config<C> {
97 pub index_partition: String,
99
100 pub value_partition: String,
102
103 pub index_page_cache: commonware_runtime::buffer::paged::CacheRef,
105
106 pub index_write_buffer: NonZeroUsize,
108
109 pub value_write_buffer: NonZeroUsize,
111
112 pub replay_buffer: NonZeroUsize,
114
115 pub compression: Option<u8>,
117
118 pub codec_config: C,
120}
121
122enum Recovery<'a> {
128 Restore { section: u64, index_size: u64 },
129 Floors(&'a BTreeMap<u64, u64>),
130 Infer,
131}
132
133struct Tracking<E: Context> {
135 metadata: Metadata<E, SectionKey, u64>,
137
138 marker_sync_pending: Option<SyncCompletion>,
140
141 barriers: BTreeMap<u64, Barrier>,
143}
144
145impl<E: Context> Tracking<E> {
146 fn stage_marker(&mut self, section: u64, floor: u64) -> bool {
148 let key = SectionKey::new(section);
149 match self.metadata.get(&key) {
150 None if floor == 0 => false,
151 Some(stored) if *stored == floor => false,
152 _ => {
153 self.metadata.put(key, floor);
154 true
155 }
156 }
157 }
158
159 fn barrier(&mut self, section: u64) -> &mut Barrier {
164 let floor = self
165 .metadata
166 .get(&SectionKey::new(section))
167 .copied()
168 .unwrap_or(0);
169 self.barriers
170 .entry(section)
171 .or_insert_with(|| Barrier::new(floor))
172 }
173
174 fn observe_marker_sync(&mut self) -> Result<bool, Error> {
178 let Some(completion) = self.marker_sync_pending.as_mut() else {
179 return Ok(false);
180 };
181 let Some(result) = completion.now_or_never() else {
182 return Ok(true);
183 };
184 result.map_err(|err| Error::Metadata(crate::metadata::Error::Runtime(err)))?;
185 self.marker_sync_pending = None;
186
187 let metadata = &self.metadata;
190 self.barriers.retain(|section, barrier| {
191 let published = metadata
192 .get(&SectionKey::new(*section))
193 .copied()
194 .unwrap_or(0);
195 barrier.boundary() > published || !barrier.settled()
196 });
197 Ok(false)
198 }
199}
200
201struct Validation {
203 current_section: Option<u64>,
205 floor: u64,
206 truncated: bool,
207
208 rewinds: Vec<(u64, u64)>,
210
211 failed: bool,
213}
214
215impl Validation {
216 const fn new() -> Self {
217 Self {
218 current_section: None,
219 floor: 0,
220 truncated: false,
221 rewinds: Vec::new(),
222 failed: false,
223 }
224 }
225}
226
227pub struct Oversized<E: Context, I: Record, V: Codec> {
238 index: FixedJournal<E, I>,
239 values: Glob<E, V>,
240 tracking: Option<Tracking<E>>,
241}
242
243impl<E: Context, I: Record + Send + Sync, V: CodecShared> std::fmt::Debug for Oversized<E, I, V> {
244 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
245 f.debug_struct("Oversized")
246 .field("oldest_section", &self.oldest_section())
247 .field("newest_section", &self.newest_section())
248 .finish_non_exhaustive()
249 }
250}
251
252impl<E: Context, I: Record + Send + Sync, V: CodecShared> Oversized<E, I, V> {
253 pub async fn init(context: E, cfg: Config<V::Cfg>) -> Result<Self, Error> {
259 let replay_buffer = cfg.replay_buffer;
260 let journal = Self::init_inner(context, cfg, Recovery::Infer).await?;
261 journal.recover_inferred(replay_buffer).await
262 }
263
264 pub async fn init_with_checkpoint(
275 context: E,
276 cfg: Config<V::Cfg>,
277 checkpoint: (u64, u64),
278 ) -> Result<Self, Error> {
279 let (section, index_size) = checkpoint;
280 Self::init_inner(
281 context,
282 cfg,
283 Recovery::Restore {
284 section,
285 index_size,
286 },
287 )
288 .await
289 }
290
291 pub async fn init_with_metadata(
297 context: &E,
298 cfg: Config<V::Cfg>,
299 metadata_partition: String,
300 read_options: ReadOptions,
301 ) -> Result<Replay<E, I, V>, Error> {
302 let replay_buffer = cfg.replay_buffer;
303
304 let metadata = Metadata::init(
306 context.child("metadata"),
307 MetadataConfig {
308 partition: metadata_partition,
309 codec_config: (),
310 },
311 )
312 .await?;
313 let floors = metadata
314 .keys()
315 .map(|key| {
316 (
317 u64::from(key),
318 *metadata.get(key).expect("metadata key must have a value"),
319 )
320 })
321 .collect::<BTreeMap<_, _>>();
322
323 let recovery = if floors.is_empty() {
326 Recovery::Infer
327 } else {
328 Recovery::Floors(&floors)
329 };
330 let mut journal = Self::init_inner(context.child("oversized"), cfg, recovery).await?;
331 journal.tracking = Some(Tracking {
332 metadata,
333 marker_sync_pending: None,
334 barriers: BTreeMap::new(),
335 });
336 let mut replay = journal.replay(0, 0, replay_buffer, read_options).await?;
337 replay.validation = Some(Validation::new());
338 Ok(replay)
339 }
340
341 async fn init_inner(
343 context: E,
344 cfg: Config<V::Cfg>,
345 recovery: Recovery<'_>,
346 ) -> Result<Self, Error> {
347 let index_cfg = FixedConfig {
348 partition: cfg.index_partition,
349 page_cache: cfg.index_page_cache,
350 write_buffer: cfg.index_write_buffer,
351 };
352 let index_context = context.child("index");
353 let value_cfg = GlobConfig {
354 partition: cfg.value_partition,
355 compression: cfg.compression,
356 codec_config: cfg.codec_config,
357 write_buffer: cfg.value_write_buffer,
358 };
359 let value_context = context.child("values");
360
361 let (index, values) = match recovery {
362 Recovery::Infer => {
363 let index = FixedJournal::init(index_context, index_cfg).await?;
364 (index, Glob::init(value_context, value_cfg).await?)
365 }
366 Recovery::Floors(minimum_items) => {
367 let preflight =
368 FixedJournal::preflight_floors(index_context, index_cfg, minimum_items).await?;
369 let values = Glob::init(value_context, value_cfg).await?;
370 Self::validate_value_floors(&values, &preflight)?;
371 (preflight.finish().await?, values)
372 }
373 Recovery::Restore {
374 section,
375 index_size,
376 } => {
377 let preflight =
378 FixedJournal::preflight_restore(index_context, index_cfg, section, index_size)
379 .await?;
380 let values = Glob::init(value_context, value_cfg).await?;
381 let value_size = Self::validate_restore_values(&values, &preflight, section)?;
382 let index = preflight.finish().await?;
383
384 let values = values.rewind(section, value_size).await?;
387 (index, values.sync(section).await?)
388 }
389 };
390 Ok(Self {
391 index,
392 values,
393 tracking: None,
394 })
395 }
396
397 async fn recover_inferred(self, buffer: NonZeroUsize) -> Result<Self, Error> {
400 let mut replay = self.replay(0, 0, buffer, ReadOptions::default()).await?;
401 while let Some(result) = replay.next().await {
402 result?;
403 }
404 replay.finish()?.repair().await
405 }
406
407 fn boundary_value_end(section: u64, entry: &Option<I>) -> Result<u64, Error> {
409 let Some(entry) = entry else {
410 return Ok(0);
411 };
412 let (offset, size) = entry.value_location();
413 offset.checked_add(u64::from(size)).ok_or_else(|| {
414 Error::Corruption(format!(
415 "section {section} has an overflowing terminal value range"
416 ))
417 })
418 }
419
420 async fn repair(mut self) -> Result<Self, Error> {
427 let chunk_size = FixedJournal::<E, I>::CHUNK_SIZE as u64;
428 let sections: Vec<u64> = self.index.sections().collect();
429
430 let mut rewound_index = Vec::new();
431 let mut rewound_values = Vec::new();
432 for section in sections {
433 let index_size = self.index.size(section)?;
434 let glob_size = match self.values.size(section) {
435 Ok(size) => size,
436 Err(Error::AlreadyPrunedToSection(oldest)) => {
437 warn!(
442 section,
443 oldest, "index has section that glob already pruned"
444 );
445 0
446 }
447 Err(e) => return Err(e),
448 };
449
450 let entry_count = index_size / chunk_size;
452 let aligned_size = entry_count * chunk_size;
453 if aligned_size < index_size {
454 warn!(
455 section,
456 index_size, aligned_size, "trailing bytes detected: truncating"
457 );
458 self.index = self.index.rewind_section(section, aligned_size).await?;
459 rewound_index.push(section);
460 }
461
462 if entry_count == 0 {
464 if glob_size > 0 {
465 debug!(section, glob_size, "truncating orphaned value bytes");
466 self.values = self.values.rewind_section(section, 0).await?;
467 rewound_values.push(section);
468 }
469 continue;
470 }
471
472 let (valid_count, glob_target) = self
474 .find_last_valid_entry(section, entry_count, glob_size)
475 .await?;
476
477 if valid_count < entry_count {
479 let valid_size = valid_count * chunk_size;
480 debug!(section, entry_count, valid_count, "rewinding index");
481 self.index = self.index.rewind_section(section, valid_size).await?;
482 rewound_index.push(section);
483 }
484
485 if glob_size > glob_target {
488 debug!(
489 section,
490 glob_size, glob_target, "truncating glob trailing garbage"
491 );
492 self.values = self.values.rewind_section(section, glob_target).await?;
493 rewound_values.push(section);
494 }
495 }
496
497 self.values = self.values.sync(&rewound_values).await?;
503 self.index = self.index.sync(&rewound_index).await?;
504
505 self.cleanup_orphan_value_sections().await
507 }
508
509 fn validate_value_floors(
511 values: &Glob<E, V>,
512 preflight: &RecoveryPreflight<E, I>,
513 ) -> Result<(), Error> {
514 for (§ion, entry) in preflight.boundaries() {
515 let required = Self::boundary_value_end(section, entry)?;
516 if required == 0 {
517 continue;
518 }
519 let retained = values.size(section)?;
520 if retained < required {
521 return Err(Error::Corruption(format!(
522 "section {section} retains {retained} value bytes, below the validation \
523 floor of {required}"
524 )));
525 }
526 }
527 Ok(())
528 }
529
530 fn validate_restore_values(
533 values: &Glob<E, V>,
534 preflight: &RecoveryPreflight<E, I>,
535 section: u64,
536 ) -> Result<u64, Error> {
537 for (&candidate, entry) in preflight.boundaries().range(..section) {
540 let required = Self::boundary_value_end(candidate, entry)?;
541 let retained = values.size(candidate)?;
542 if retained != required {
543 return Err(Error::Corruption(format!(
544 "section {candidate} index ends at value byte {required}, but its glob size is {retained}"
545 )));
546 }
547 }
548
549 if let Some(orphan) = values.sections().find(|candidate| {
551 *candidate < section && !preflight.boundaries().contains_key(candidate)
552 }) {
553 return Err(Error::Corruption(format!(
554 "section {orphan} has values but no index"
555 )));
556 }
557
558 let required = Self::boundary_value_end(section, &preflight.boundaries()[§ion])?;
560 if required > 0 {
561 let retained = values.size(section)?;
562 if retained < required {
563 return Err(Error::Corruption(format!(
564 "section {section} retains {retained} of {required} committed value bytes"
565 )));
566 }
567 }
568
569 Ok(required)
570 }
571
572 async fn cleanup_orphan_value_sections(mut self) -> Result<Self, Error> {
579 let index_sections: HashSet<u64> = self.index.sections().collect();
581
582 let orphan_sections: Vec<u64> = self
584 .values
585 .sections()
586 .filter(|s| !index_sections.contains(s))
587 .collect();
588
589 for section in orphan_sections {
591 warn!(section, "removing orphan value section");
592 (self.values, _) = self.values.remove_section(section).await?;
593 }
594
595 Ok(self)
596 }
597
598 async fn align_values_to_index(mut self) -> Result<Self, Error> {
600 let sections = self.index.sections().collect::<Vec<_>>();
601 let mut rewound = Vec::new();
602 for section in sections {
603 let target = Self::boundary_value_end(section, &self.index.last(section).await?)?;
604 let retained = self.values.size(section)?;
605 if retained < target {
606 return Err(Error::Corruption(format!(
607 "section {section} retains {retained} of {target} indexed value bytes"
608 )));
609 }
610 if retained > target {
611 self.values = self.values.rewind_section(section, target).await?;
612 rewound.push(section);
613 }
614 }
615 self.values = self.values.sync(&rewound).await?;
616 self.cleanup_orphan_value_sections().await
617 }
618
619 async fn find_last_valid_entry(
626 &self,
627 section: u64,
628 entry_count: u64,
629 glob_size: u64,
630 ) -> Result<(u64, u64), Error> {
631 for pos in (0..entry_count).rev() {
632 match self.index.get(section, pos).await {
633 Ok(entry) => {
634 let (offset, size) = entry.value_location();
635 let entry_end = offset.saturating_add(u64::from(size));
636 if entry_end <= glob_size && self.values.verify(section, offset, size).await? {
637 return Ok((pos + 1, entry_end));
638 }
639 if pos == entry_count - 1 {
640 warn!(
641 section,
642 pos, glob_size, entry_end, "invalid entry: glob truncated or corrupt"
643 );
644 }
645 }
646 Err(Error::ItemOutOfRange(_) | Error::Runtime(RError::InvalidChecksum)) => {
647 if pos == entry_count - 1 {
648 warn!(section, pos, "corrupted last entry, scanning backwards");
649 }
650 }
651 Err(err) => return Err(err),
652 }
653 }
654 Ok((0, 0))
655 }
656
657 async fn reconcile_markers(mut self) -> Result<Self, Error> {
659 let mut tracking = self
660 .tracking
661 .take()
662 .expect("tracked replay preserves its recovery state");
663
664 let mut dirty = false;
665 for section in self.index.sections() {
666 let items = self.index.section_len(section)?;
667 dirty |= tracking.stage_marker(section, items);
668 }
669 if dirty {
670 let marker;
671 (tracking.metadata, marker) = tracking.metadata.start_sync().await?;
672 tracking.marker_sync_pending = Some(marker.boxed().shared());
673 }
674 self.tracking = Some(tracking);
675 Ok(self)
676 }
677
678 async fn prepare_rewind(
680 &mut self,
681 section: u64,
682 index_size: u64,
683 remove_later: bool,
684 ) -> Result<(), Error> {
685 let Some(mut tracking) = self.tracking.take() else {
686 return Ok(());
687 };
688 let items = index_size / FixedJournal::<E, I>::CHUNK_SIZE as u64;
689 let mut dirty = false;
690
691 if remove_later {
692 tracking.metadata.retain(|key, _| {
693 let keep = u64::from(key) <= section;
694 dirty |= !keep;
695 keep
696 });
697 }
698 let key = SectionKey::new(section);
699 if tracking
700 .metadata
701 .get(&key)
702 .is_some_and(|floor| *floor > items)
703 {
704 tracking.metadata.put(key, items);
705 dirty = true;
706 }
707 if dirty {
708 tracking.metadata = tracking.metadata.sync().await?;
709 tracking.marker_sync_pending = None;
710 }
711
712 if remove_later {
713 tracking
714 .barriers
715 .retain(|candidate, _| *candidate <= section);
716 }
717 if let Some(barrier) = tracking.barriers.get_mut(§ion) {
718 barrier.truncate(items);
719 }
720 self.tracking = Some(tracking);
721 Ok(())
722 }
723
724 pub async fn append(
733 mut self,
734 section: u64,
735 entry: I,
736 value: &V,
737 ) -> Result<(Self, u64, u64, u32), Error> {
738 let (offset, size);
741 (self.values, offset, size) = self.values.append(section, value).await?;
742
743 let entry_with_location = entry.with_location(offset, size);
745 let position;
746 (self.index, position) = self.index.append(section, &entry_with_location).await?;
747
748 if let Some(tracking) = &mut self.tracking {
751 tracking.barrier(section);
752 }
753
754 Ok((self, position, offset, size))
755 }
756
757 pub async fn get(&self, section: u64, position: u64) -> Result<I, Error> {
759 self.index.get(section, position).await
760 }
761
762 pub async fn last(&self, section: u64) -> Result<Option<I>, Error> {
771 self.index.last(section).await
772 }
773
774 pub async fn get_value(&self, section: u64, offset: u64, size: u32) -> Result<V, Error> {
778 self.values.get(section, offset, size).await
779 }
780
781 pub async fn replay(
789 self,
790 start_section: u64,
791 start_position: u64,
792 buffer: NonZeroUsize,
793 read_options: ReadOptions,
794 ) -> Result<Replay<E, I, V>, Error> {
795 let Self {
796 index,
797 values,
798 tracking,
799 } = self;
800 let index = index
801 .replay(start_section, start_position, buffer, read_options)
802 .await?;
803 Ok(Replay {
804 index,
805 values,
806 tracking,
807 validation: None,
808 })
809 }
810
811 pub(crate) async fn start_sync_tracked(
817 mut self,
818 sections: &BTreeSet<u64>,
819 active: &BTreeSet<u64>,
820 ) -> Result<(Self, Handle<()>), Error> {
821 let mut tracking = self
822 .tracking
823 .take()
824 .expect("tracked sync preserves its recovery state");
825 let marker_pending = tracking.observe_marker_sync()?;
826 let lengths = sections
827 .iter()
828 .map(|§ion| Ok((section, self.index.section_len(section)?)))
829 .collect::<Result<Vec<_>, Error>>()?;
830 let ((index, index_handle), (values, values_handle)) = try_join(
831 self.index.start_sync(sections),
832 self.values.start_sync(sections),
833 )
834 .await?;
835 self.index = index;
836 self.values = values;
837 let completion: SyncCompletion =
838 async move { try_join(index_handle, values_handle).await.map(|_| ()) }
839 .boxed()
840 .shared();
841
842 for (section, length) in lengths {
845 tracking.barrier(section).record(length, completion.clone());
846 }
847
848 if !marker_pending {
851 let publish = tracking
852 .barriers
853 .iter_mut()
854 .filter(|(section, _)| !active.contains(section))
855 .map(|(§ion, barrier)| (section, barrier.boundary()))
856 .collect::<Vec<_>>();
857 let mut metadata_dirty = false;
858 for (section, floor) in publish {
859 metadata_dirty |= tracking.stage_marker(section, floor);
860 }
861 if metadata_dirty {
862 let handle;
863 (tracking.metadata, handle) = tracking.metadata.start_sync().await?;
864 tracking.marker_sync_pending = Some(handle.boxed().shared());
865 }
866 }
867
868 self.tracking = Some(tracking);
869 Ok((self, Handle::from_future(completion)))
870 }
871
872 pub(crate) async fn sync_tracked(
875 self,
876 sections: &BTreeSet<u64>,
877 active: &BTreeSet<u64>,
878 ) -> Result<Self, Error> {
879 let (mut journal, handle) = self.start_sync_tracked(sections, active).await?;
880 handle.await?;
881 journal
882 .tracking
883 .as_mut()
884 .expect("tracking mode is preserved")
885 .observe_marker_sync()?;
886 Ok(journal)
887 }
888
889 pub async fn sync(mut self, sections: impl crate::Sections) -> Result<Self, Error> {
891 if self.tracking.is_some() {
892 let sections = sections.sections().collect::<BTreeSet<_>>();
893 return self.sync_tracked(§ions, §ions).await;
894 }
895
896 let sections = sections.sections().collect::<Vec<_>>();
897 (self.index, self.values) =
898 try_join(self.index.sync(§ions), self.values.sync(§ions)).await?;
899 Ok(self)
900 }
901
902 pub async fn start_sync(
908 mut self,
909 sections: impl crate::Sections,
910 ) -> Result<(Self, Handle<()>), Error> {
911 if self.tracking.is_some() {
912 let sections = sections.sections().collect::<BTreeSet<_>>();
913 return self.start_sync_tracked(§ions, §ions).await;
914 }
915
916 let sections = sections.sections().collect::<Vec<_>>();
917 let ((index, index_handle), (values, values_handle)) = try_join(
918 self.index.start_sync(§ions),
919 self.values.start_sync(§ions),
920 )
921 .await?;
922 self.index = index;
923 self.values = values;
924 Ok((
925 self,
926 Handle::from_future(
927 async move { try_join(index_handle, values_handle).await.map(|_| ()) },
928 ),
929 ))
930 }
931
932 pub async fn sync_all(mut self) -> Result<Self, Error> {
934 if self.tracking.is_some() {
935 let sections = self.index.sections().collect::<BTreeSet<_>>();
936 return self.sync(sections).await;
937 }
938 (self.index, self.values) = try_join(self.index.sync_all(), self.values.sync_all()).await?;
939 Ok(self)
940 }
941
942 pub async fn prune(mut self, min: u64) -> Result<(Self, bool), Error> {
948 if let Some(mut tracking) = self.tracking.take() {
951 let mut removed = false;
952 tracking.metadata.retain(|key, _| {
953 let keep = u64::from(key) >= min;
954 removed |= !keep;
955 keep
956 });
957 if removed {
958 tracking.metadata = tracking.metadata.sync().await?;
959 tracking.marker_sync_pending = None;
960 }
961 tracking.barriers = tracking.barriers.split_off(&min);
962 self.tracking = Some(tracking);
963 }
964
965 let index_pruned;
966 (self.index, index_pruned) = self.index.prune(min).await?;
967 let value_pruned;
968 (self.values, value_pruned) = self.values.prune(min).await?;
969 Ok((self, index_pruned || value_pruned))
970 }
971
972 async fn rewound_value_end(&self, section: u64, index_size: u64) -> Result<u64, Error> {
976 match self.index.last(section).await {
977 Ok(Some(entry)) => {
978 let (offset, size) = entry.value_location();
979 offset
980 .checked_add(u64::from(size))
981 .ok_or(Error::OffsetOverflow)
982 }
983 Ok(None) => Ok(0),
984 Err(Error::SectionOutOfRange(_)) if index_size == 0 => Ok(0),
985 Err(e) => Err(e),
986 }
987 }
988
989 pub async fn rewind(mut self, section: u64, index_size: u64) -> Result<Self, Error> {
999 self.prepare_rewind(section, index_size, true).await?;
1000
1001 self.index = self.index.rewind(section, index_size).await?;
1003
1004 let value_size = self.rewound_value_end(section, index_size).await?;
1006
1007 self.index = self.index.sync(section).await?;
1012
1013 self.values = self.values.rewind(section, value_size).await?;
1015 self.values = self.values.sync(section).await?;
1016 Ok(self)
1017 }
1018
1019 pub async fn rewind_section(mut self, section: u64, index_size: u64) -> Result<Self, Error> {
1026 self.prepare_rewind(section, index_size, false).await?;
1027
1028 self.index = self.index.rewind_section(section, index_size).await?;
1030
1031 let value_size = self.rewound_value_end(section, index_size).await?;
1033
1034 self.index = self.index.sync(section).await?;
1036
1037 self.values = self.values.rewind_section(section, value_size).await?;
1039 self.values = self.values.sync(section).await?;
1040 Ok(self)
1041 }
1042
1043 pub fn size(&self, section: u64) -> Result<u64, Error> {
1047 self.index.size(section)
1048 }
1049
1050 pub async fn value_size(&self, section: u64) -> Result<u64, Error> {
1052 match self.index.last(section).await {
1053 Ok(Some(entry)) => {
1054 let (offset, size) = entry.value_location();
1055 offset
1056 .checked_add(u64::from(size))
1057 .ok_or(Error::OffsetOverflow)
1058 }
1059 Ok(None) | Err(Error::SectionOutOfRange(_)) => Ok(0),
1060 Err(e) => Err(e),
1061 }
1062 }
1063
1064 pub fn pruned(&self, section: u64) -> bool {
1069 self.index.pruned(section)
1070 }
1071
1072 pub fn oldest_section(&self) -> Option<u64> {
1074 self.index.oldest_section()
1075 }
1076
1077 pub fn newest_section(&self) -> Option<u64> {
1079 self.index.newest_section()
1080 }
1081
1082 pub async fn destroy(mut self) -> Result<(), Error> {
1084 if let Some(tracking) = self.tracking.take() {
1087 tracking.metadata.destroy().await?;
1088 }
1089 try_join(self.index.destroy(), self.values.destroy())
1090 .await
1091 .map(|_| ())
1092 }
1093}
1094
1095pub struct Replay<E: Context, I: Record, V: Codec> {
1101 index: FixedReplay<E, I>,
1102 values: Glob<E, V>,
1103 tracking: Option<Tracking<E>>,
1104 validation: Option<Validation>,
1105}
1106
1107impl<E: Context, I: Record + Send + Sync, V: CodecShared> Replay<E, I, V> {
1108 pub async fn next(&mut self) -> Option<Result<(u64, u64, I), Error>> {
1117 loop {
1118 let result = self.index.next().await?;
1119 let (section, position, entry) = match result {
1120 Ok(entry) => entry,
1121 Err(err) => return Some(Err(err)),
1122 };
1123 let (tracking, validation) = (&self.tracking, &mut self.validation);
1124 let Some(validation) = validation else {
1125 return Some(Ok((section, position, entry)));
1126 };
1127
1128 if validation.current_section != Some(section) {
1131 validation.current_section = Some(section);
1132 validation.floor = tracking
1133 .as_ref()
1134 .expect("tracked replay preserves its recovery state")
1135 .metadata
1136 .get(&SectionKey::new(section))
1137 .copied()
1138 .unwrap_or(0);
1139 validation.truncated = false;
1140 }
1141 if validation.truncated {
1142 continue;
1143 }
1144 if position < validation.floor {
1145 return Some(Ok((section, position, entry)));
1146 }
1147
1148 let (offset, size) = entry.value_location();
1149 match self.values.verify(section, offset, size).await {
1150 Ok(true) => return Some(Ok((section, position, entry))),
1151 Ok(false) => {
1152 validation.rewinds.push((section, position));
1153 validation.truncated = true;
1154 }
1155 Err(err) => {
1156 validation.failed = true;
1157 return Some(Err(err));
1158 }
1159 }
1160 }
1161 }
1162
1163 pub fn finish(self) -> Result<Oversized<E, I, V>, Error> {
1168 if self.validation.is_some() {
1169 return Err(Error::ReplayFailed);
1170 }
1171 Ok(Oversized {
1172 index: self.index.finish()?,
1173 values: self.values,
1174 tracking: self.tracking,
1175 })
1176 }
1177
1178 pub async fn finish_tracked(self) -> Result<Oversized<E, I, V>, Error> {
1180 let Some(validation) = self.validation else {
1181 return Err(Error::ReplayFailed);
1182 };
1183 if validation.failed {
1184 return Err(Error::ReplayFailed);
1185 }
1186 let mut journal = Oversized {
1187 index: self.index.finish()?,
1188 values: self.values,
1189 tracking: self.tracking,
1190 };
1191
1192 let chunk_size = FixedJournal::<E, I>::CHUNK_SIZE as u64;
1195 for (section, items) in validation.rewinds {
1196 let index_size = items.checked_mul(chunk_size).ok_or(Error::OffsetOverflow)?;
1197 journal = journal.rewind_section(section, index_size).await?;
1198 }
1199 journal
1200 .align_values_to_index()
1201 .await?
1202 .reconcile_markers()
1203 .await
1204 }
1205}
1206
1207#[cfg(test)]
1208mod tests {
1209 use super::*;
1210 use commonware_codec::{FixedSize, Read, ReadExt, Write};
1211 use commonware_cryptography::Crc32;
1212 use commonware_macros::test_traced;
1213 use commonware_runtime::{
1214 Blob as _, Buf, BufMut, BufferPooler, Runner, Storage as _, Supervisor as _, WriteOptions,
1215 buffer::paged::{CacheRef, corrupt_page},
1216 deterministic,
1217 mocks::{DelayedSyncContext, PendingSyncs, SyncFaultContext, drive_pending_syncs},
1218 };
1219 use commonware_utils::{NZU16, NZUsize};
1220
1221 fn byte_end(offset: u64, size: u32) -> u64 {
1223 offset + u64::from(size)
1224 }
1225
1226 #[derive(Debug, Clone, PartialEq)]
1228 struct TestEntry {
1229 id: u64,
1230 value_offset: u64,
1231 value_size: u32,
1232 }
1233
1234 impl TestEntry {
1235 fn new(id: u64, value_offset: u64, value_size: u32) -> Self {
1236 Self {
1237 id,
1238 value_offset,
1239 value_size,
1240 }
1241 }
1242 }
1243
1244 impl Write for TestEntry {
1245 fn write(&self, buf: &mut impl BufMut) {
1246 self.id.write(buf);
1247 self.value_offset.write(buf);
1248 self.value_size.write(buf);
1249 }
1250 }
1251
1252 impl Read for TestEntry {
1253 type Cfg = ();
1254
1255 fn read_cfg(buf: &mut impl Buf, _: &Self::Cfg) -> Result<Self, commonware_codec::Error> {
1256 let id = u64::read(buf)?;
1257 let value_offset = u64::read(buf)?;
1258 let value_size = u32::read(buf)?;
1259 Ok(Self {
1260 id,
1261 value_offset,
1262 value_size,
1263 })
1264 }
1265 }
1266
1267 impl FixedSize for TestEntry {
1268 const SIZE: usize = u64::SIZE + u64::SIZE + u32::SIZE;
1269 }
1270
1271 impl Record for TestEntry {
1272 fn value_location(&self) -> (u64, u32) {
1273 (self.value_offset, self.value_size)
1274 }
1275
1276 fn with_location(mut self, offset: u64, size: u32) -> Self {
1277 self.value_offset = offset;
1278 self.value_size = size;
1279 self
1280 }
1281 }
1282
1283 fn test_cfg(pooler: &impl BufferPooler) -> Config<()> {
1284 Config {
1285 index_partition: "test-index".into(),
1286 value_partition: "test-values".into(),
1287 index_page_cache: CacheRef::from_pooler(pooler, NZU16!(64), NZUsize!(8)),
1288 index_write_buffer: NZUsize!(1024),
1289 value_write_buffer: NZUsize!(1024),
1290 replay_buffer: NZUsize!(4096),
1291 compression: None,
1292 codec_config: (),
1293 }
1294 }
1295
1296 fn entry_cfg(pooler: &impl BufferPooler) -> Config<()> {
1298 let mut cfg = test_cfg(pooler);
1299 cfg.index_page_cache =
1300 CacheRef::from_pooler(pooler, NZU16!(TestEntry::SIZE as u16), NZUsize!(8));
1301 cfg
1302 }
1303
1304 type TestValue = [u8; 16];
1306
1307 #[test_traced]
1308 fn test_oversized_append_and_get() {
1309 let executor = deterministic::Runner::default();
1310 executor.start(|context| async move {
1311 let cfg = test_cfg(&context);
1312 let mut oversized: Oversized<_, TestEntry, TestValue> =
1313 Oversized::init(context, cfg).await.expect("Failed to init");
1314
1315 let value: TestValue = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16];
1317 let entry = TestEntry::new(42, 0, 0);
1318 let (position, offset, size);
1319 (oversized, position, offset, size) = oversized
1320 .append(1, entry, &value)
1321 .await
1322 .expect("Failed to append");
1323
1324 assert_eq!(position, 0);
1325
1326 let retrieved_entry = oversized.get(1, position).await.expect("Failed to get");
1328 assert_eq!(retrieved_entry.id, 42);
1329
1330 let retrieved_value = oversized
1332 .get_value(1, offset, size)
1333 .await
1334 .expect("Failed to get value");
1335 assert_eq!(retrieved_value, value);
1336
1337 oversized.destroy().await.expect("Failed to destroy");
1338 });
1339 }
1340
1341 #[test_traced]
1342 fn test_oversized_crash_recovery() {
1343 let executor = deterministic::Runner::default();
1344 executor.start(|context| async move {
1345 let cfg = test_cfg(&context);
1346
1347 let mut oversized: Oversized<_, TestEntry, TestValue> =
1349 Oversized::init(context.child("first"), cfg.clone())
1350 .await
1351 .expect("Failed to init");
1352
1353 let mut locations = Vec::new();
1355 for i in 0..5u8 {
1356 let value: TestValue = [i; 16];
1357 let entry = TestEntry::new(i as u64, 0, 0);
1358 let (position, offset, size);
1359 (oversized, position, offset, size) = oversized
1360 .append(1, entry, &value)
1361 .await
1362 .expect("Failed to append");
1363 locations.push((position, offset, size));
1364 }
1365 oversized = oversized.sync(1).await.expect("Failed to sync");
1366 drop(oversized);
1367
1368 let (blob, _) = context
1370 .open(&cfg.value_partition, &1u64.to_be_bytes())
1371 .await
1372 .expect("Failed to open blob");
1373
1374 let keep_size = byte_end(locations[2].1, locations[2].2);
1376 blob.resize(keep_size).await.expect("Failed to truncate");
1377 blob.sync().await.expect("Failed to sync");
1378 drop(blob);
1379
1380 let oversized: Oversized<_, TestEntry, TestValue> =
1382 Oversized::init(context.child("second"), cfg.clone())
1383 .await
1384 .expect("Failed to reinit");
1385
1386 for i in 0..3u8 {
1388 let (position, offset, size) = locations[i as usize];
1389 let entry = oversized.get(1, position).await.expect("Failed to get");
1390 assert_eq!(entry.id, i as u64);
1391
1392 let value = oversized
1393 .get_value(1, offset, size)
1394 .await
1395 .expect("Failed to get value");
1396 assert_eq!(value, [i; 16]);
1397 }
1398
1399 let result = oversized.get(1, 3).await;
1401 assert!(result.is_err());
1402
1403 oversized.destroy().await.expect("Failed to destroy");
1404 });
1405 }
1406
1407 #[test_traced]
1408 fn test_oversized_persistence() {
1409 let executor = deterministic::Runner::default();
1410 executor.start(|context| async move {
1411 let cfg = test_cfg(&context);
1412
1413 let mut oversized: Oversized<_, TestEntry, TestValue> =
1415 Oversized::init(context.child("first"), cfg.clone())
1416 .await
1417 .expect("Failed to init");
1418
1419 let value: TestValue = [42; 16];
1420 let entry = TestEntry::new(123, 0, 0);
1421 let (position, offset, size);
1422 (oversized, position, offset, size) = oversized
1423 .append(1, entry, &value)
1424 .await
1425 .expect("Failed to append");
1426 oversized = oversized.sync(1).await.expect("Failed to sync");
1427 drop(oversized);
1428
1429 let oversized: Oversized<_, TestEntry, TestValue> =
1431 Oversized::init(context.child("second"), cfg)
1432 .await
1433 .expect("Failed to reinit");
1434
1435 let retrieved_entry = oversized.get(1, position).await.expect("Failed to get");
1436 assert_eq!(retrieved_entry.id, 123);
1437
1438 let retrieved_value = oversized
1439 .get_value(1, offset, size)
1440 .await
1441 .expect("Failed to get value");
1442 assert_eq!(retrieved_value, value);
1443
1444 oversized.destroy().await.expect("Failed to destroy");
1445 });
1446 }
1447
1448 #[test_traced]
1449 fn test_oversized_sync() {
1450 let executor = deterministic::Runner::default();
1451 executor.start(|context| async move {
1452 let cfg = test_cfg(&context);
1453
1454 let mut oversized: Oversized<_, TestEntry, TestValue> =
1455 Oversized::init(context.child("first"), cfg.clone())
1456 .await
1457 .expect("Failed to init");
1458
1459 let mut located = Vec::new();
1461 for section in 1u64..=3 {
1462 let value: TestValue = [section as u8; 16];
1463 let entry = TestEntry::new(section, 0, 0);
1464 let (position, offset, size);
1465 (oversized, position, offset, size) = oversized
1466 .append(section, entry, &value)
1467 .await
1468 .expect("Failed to append");
1469 located.push((section, position, offset, size, value));
1470 }
1471
1472 oversized = oversized
1475 .sync(&[1, 3, 99])
1476 .await
1477 .expect("Failed to sync sections");
1478 drop(oversized);
1479
1480 let oversized: Oversized<_, TestEntry, TestValue> =
1482 Oversized::init(context.child("second"), cfg)
1483 .await
1484 .expect("Failed to reinit");
1485 for &(section, position, offset, size, value) in &located {
1486 let result = oversized.get(section, position).await;
1487 if section == 2 {
1488 assert!(result.is_err(), "unsynced section 2 must not be durable");
1489 continue;
1490 }
1491 assert_eq!(result.expect("synced entry durable").id, section);
1492 let retrieved = oversized
1493 .get_value(section, offset, size)
1494 .await
1495 .expect("synced value durable");
1496 assert_eq!(retrieved, value);
1497 }
1498
1499 oversized.destroy().await.expect("Failed to destroy");
1500 });
1501 }
1502
1503 async fn assert_adopted_entries_consistent(
1506 oversized: &Oversized<deterministic::Context, TestEntry, TestValue>,
1507 ) {
1508 let chunk = FixedJournal::<deterministic::Context, TestEntry>::CHUNK_SIZE as u64;
1509 for position in 0..oversized.size(1).expect("size") / chunk {
1510 let entry = oversized.get(1, position).await.expect("Failed to get");
1511 let (offset, size) = entry.value_location();
1512 let value = oversized
1513 .get_value(1, offset, size)
1514 .await
1515 .expect("adopted entry must reference durable bytes");
1516 assert_eq!(
1517 value, [entry.id as u8; 16],
1518 "entry {} must read back the value appended with it",
1519 entry.id
1520 );
1521 }
1522 }
1523
1524 #[test_traced]
1525 fn test_oversized_rewind_truncation_durable_before_offset_reuse() {
1526 let executor = deterministic::Runner::default();
1527 let (_, checkpoint) = executor.start_and_recover(|context| async move {
1528 let mut oversized: Oversized<_, TestEntry, TestValue> =
1530 Oversized::init(context.child("first"), test_cfg(&context))
1531 .await
1532 .expect("Failed to init");
1533 (oversized, _, _, _) = oversized
1534 .append(1, TestEntry::new(1, 0, 0), &[1; 16])
1535 .await
1536 .expect("Failed to append");
1537 oversized = oversized.sync(1).await.expect("Failed to sync");
1538
1539 oversized = oversized.rewind(1, 0).await.expect("Failed to rewind");
1545 (oversized, _, _, _) = oversized
1546 .append(1, TestEntry::new(2, 0, 0), &[2; 16])
1547 .await
1548 .expect("Failed to append");
1549 oversized.values = oversized
1550 .values
1551 .sync(1)
1552 .await
1553 .expect("Failed to sync values");
1554 });
1555
1556 deterministic::Runner::from(checkpoint).start(|context| async move {
1557 let oversized: Oversized<_, TestEntry, TestValue> =
1558 Oversized::init(context.child("second"), test_cfg(&context))
1559 .await
1560 .expect("Failed to reinit");
1561 assert_eq!(
1562 oversized.size(1).expect("size"),
1563 0,
1564 "rewound entry must not be revived over reused value bytes"
1565 );
1566 oversized.destroy().await.expect("Failed to destroy");
1567 });
1568 }
1569
1570 #[test_traced]
1571 fn test_oversized_recovery_never_adopts_entries_for_lost_values() {
1572 let executor = deterministic::Runner::default();
1575 let (_, checkpoint) = executor.start_and_recover(|context| async move {
1576 let mut oversized: Oversized<_, TestEntry, TestValue> =
1577 Oversized::init(context.child("first"), test_cfg(&context))
1578 .await
1579 .expect("Failed to init");
1580 (oversized, _, _, _) = oversized
1581 .append(1, TestEntry::new(1, 0, 0), &[1; 16])
1582 .await
1583 .expect("Failed to append");
1584 oversized.index = oversized.index.sync(1).await.expect("Failed to sync index");
1585 });
1586
1587 let (_, checkpoint) =
1591 deterministic::Runner::from(checkpoint).start_and_recover(|context| async move {
1592 let mut oversized: Oversized<_, TestEntry, TestValue> =
1593 Oversized::init(context.child("second"), test_cfg(&context))
1594 .await
1595 .expect("Failed to reinit");
1596 assert_eq!(
1597 oversized.size(1).expect("size"),
1598 0,
1599 "entry without durable value bytes must be rewound"
1600 );
1601 (oversized, _, _, _) = oversized
1602 .append(1, TestEntry::new(2, 0, 0), &[2; 16])
1603 .await
1604 .expect("Failed to append");
1605 oversized.values = oversized
1606 .values
1607 .sync(1)
1608 .await
1609 .expect("Failed to sync values");
1610 });
1611
1612 deterministic::Runner::from(checkpoint).start(|context| async move {
1615 let oversized: Oversized<_, TestEntry, TestValue> =
1616 Oversized::init(context.child("third"), test_cfg(&context))
1617 .await
1618 .expect("Failed to reinit");
1619 assert_adopted_entries_consistent(&oversized).await;
1620 oversized.destroy().await.expect("Failed to destroy");
1621 });
1622 }
1623
1624 #[test_traced]
1625 fn test_oversized_rewind_fails_when_truncation_cannot_be_made_durable() {
1626 let executor = deterministic::Runner::default();
1627 let (_, checkpoint) = executor.start_and_recover(|context| async move {
1628 let mut oversized: Oversized<_, TestEntry, TestValue> =
1630 Oversized::init(context.child("first"), test_cfg(&context))
1631 .await
1632 .expect("Failed to init");
1633 (oversized, _, _, _) = oversized
1634 .append(1, TestEntry::new(1, 0, 0), &[1; 16])
1635 .await
1636 .expect("Failed to append");
1637 oversized = oversized.sync(1).await.expect("Failed to sync");
1638 drop(oversized);
1639
1640 let faulty_values = SyncFaultContext {
1644 inner: context.child("second"),
1645 fail_partition: "test-values".into(),
1646 };
1647 let oversized: Oversized<_, TestEntry, TestValue> =
1648 Oversized::init(faulty_values, test_cfg(&context))
1649 .await
1650 .expect("Failed to reinit");
1651 assert!(
1652 oversized.rewind(1, 0).await.is_err(),
1653 "rewind must fail when its truncation cannot be made durable"
1654 );
1655 });
1656
1657 deterministic::Runner::from(checkpoint).start(|context| async move {
1660 let oversized: Oversized<_, TestEntry, TestValue> =
1661 Oversized::init(context.child("third"), test_cfg(&context))
1662 .await
1663 .expect("Failed to reinit");
1664 assert_eq!(oversized.size(1).expect("size"), 0);
1665 oversized.destroy().await.expect("Failed to destroy");
1666 });
1667 }
1668
1669 #[test_traced]
1670 fn test_oversized_recovery_glob_truncation_durable_before_offset_reuse() {
1671 let executor = deterministic::Runner::default();
1674 let (_, checkpoint) = executor.start_and_recover(|context| async move {
1675 let mut oversized: Oversized<_, TestEntry, TestValue> =
1676 Oversized::init(context.child("first"), test_cfg(&context))
1677 .await
1678 .expect("Failed to init");
1679 (oversized, _, _, _) = oversized
1680 .append(1, TestEntry::new(1, 0, 0), &[1; 16])
1681 .await
1682 .expect("Failed to append");
1683 (oversized, _, _, _) = oversized
1684 .append(1, TestEntry::new(2, 0, 0), &[2; 16])
1685 .await
1686 .expect("Failed to append");
1687 oversized = oversized.sync(1).await.expect("Failed to sync");
1688
1689 let chunk = FixedJournal::<deterministic::Context, TestEntry>::CHUNK_SIZE as u64;
1690 oversized.index = oversized
1691 .index
1692 .rewind(1, chunk)
1693 .await
1694 .expect("Failed to rewind index");
1695 oversized.index = oversized.index.sync(1).await.expect("Failed to sync index");
1696 });
1697
1698 let (_, checkpoint) =
1702 deterministic::Runner::from(checkpoint).start_and_recover(|context| async move {
1703 let mut oversized: Oversized<_, TestEntry, TestValue> =
1704 Oversized::init(context.child("second"), test_cfg(&context))
1705 .await
1706 .expect("Failed to reinit");
1707 (oversized, _, _, _) = oversized
1708 .append(1, TestEntry::new(3, 0, 0), &[3; 16])
1709 .await
1710 .expect("Failed to append");
1711 oversized.index = oversized.index.sync(1).await.expect("Failed to sync index");
1712 });
1713
1714 deterministic::Runner::from(checkpoint).start(|context| async move {
1717 let oversized: Oversized<_, TestEntry, TestValue> =
1718 Oversized::init(context.child("third"), test_cfg(&context))
1719 .await
1720 .expect("Failed to reinit");
1721 assert_adopted_entries_consistent(&oversized).await;
1722 oversized.destroy().await.expect("Failed to destroy");
1723 });
1724 }
1725
1726 #[test_traced]
1727 fn test_oversized_rewind_crash_between_truncations_recovers_post_rewind() {
1728 let executor = deterministic::Runner::default();
1729 let (_, checkpoint) = executor.start_and_recover(|context| async move {
1730 let mut oversized: Oversized<_, TestEntry, TestValue> =
1732 Oversized::init(context.child("first"), test_cfg(&context))
1733 .await
1734 .expect("Failed to init");
1735 (oversized, _, _, _) = oversized
1736 .append(1, TestEntry::new(1, 0, 0), &[1; 16])
1737 .await
1738 .expect("Failed to append");
1739 (oversized, _, _, _) = oversized
1740 .append(1, TestEntry::new(2, 0, 0), &[2; 16])
1741 .await
1742 .expect("Failed to append");
1743 oversized = oversized.sync(1).await.expect("Failed to sync");
1744
1745 let chunk = FixedJournal::<deterministic::Context, TestEntry>::CHUNK_SIZE as u64;
1748 oversized.index = oversized
1749 .index
1750 .rewind(1, chunk)
1751 .await
1752 .expect("Failed to rewind index");
1753 oversized.index = oversized.index.sync(1).await.expect("Failed to sync index");
1754 });
1755
1756 deterministic::Runner::from(checkpoint).start(|context| async move {
1759 let oversized: Oversized<_, TestEntry, TestValue> =
1760 Oversized::init(context.child("second"), test_cfg(&context))
1761 .await
1762 .expect("Failed to reinit");
1763 let chunk = FixedJournal::<deterministic::Context, TestEntry>::CHUNK_SIZE as u64;
1764 assert_eq!(oversized.size(1).expect("size"), chunk);
1765 let entry = oversized.get(1, 0).await.expect("Failed to get");
1766 let (offset, size) = entry.value_location();
1767 assert_eq!(
1768 oversized.values.size(1).expect("glob size"),
1769 byte_end(offset, size),
1770 "orphaned value bytes must be truncated"
1771 );
1772 assert_adopted_entries_consistent(&oversized).await;
1773 oversized.destroy().await.expect("Failed to destroy");
1774 });
1775 }
1776
1777 #[test_traced]
1778 fn test_oversized_start_sync_completion_means_recoverable() {
1779 let executor = deterministic::Runner::default();
1780 let (_, checkpoint) = executor.start_and_recover(|context| async move {
1781 let mut oversized: Oversized<_, TestEntry, TestValue> =
1782 Oversized::init(context.child("first"), test_cfg(&context))
1783 .await
1784 .expect("Failed to init");
1785 (oversized, _, _, _) = oversized
1786 .append(1, TestEntry::new(1, 0, 0), &[1; 16])
1787 .await
1788 .expect("Failed to append");
1789 let (_oversized, handle) = oversized.start_sync(1).await.expect("Failed to start sync");
1790 handle.await.expect("sync must complete");
1791 });
1793
1794 deterministic::Runner::from(checkpoint).start(|context| async move {
1795 let oversized: Oversized<_, TestEntry, TestValue> =
1796 Oversized::init(context.child("second"), test_cfg(&context))
1797 .await
1798 .expect("Failed to reinit");
1799 let chunk = FixedJournal::<deterministic::Context, TestEntry>::CHUNK_SIZE as u64;
1800 assert_eq!(oversized.size(1).expect("size"), chunk);
1801 assert_adopted_entries_consistent(&oversized).await;
1802 oversized.destroy().await.expect("Failed to destroy");
1803 });
1804 }
1805
1806 #[test_traced]
1807 fn test_oversized_sync_values_failure_recovers_clean() {
1808 let executor = deterministic::Runner::default();
1809 let (_, checkpoint) = executor.start_and_recover(|context| async move {
1810 let faulty_values = SyncFaultContext {
1811 inner: context.child("first"),
1812 fail_partition: "test-values".into(),
1813 };
1814 let mut oversized: Oversized<_, TestEntry, TestValue> =
1815 Oversized::init(faulty_values, test_cfg(&context))
1816 .await
1817 .expect("Failed to init");
1818 (oversized, _, _, _) = oversized
1819 .append(1, TestEntry::new(1, 0, 0), &[1; 16])
1820 .await
1821 .expect("Failed to append");
1822
1823 assert!(oversized.sync(1).await.is_err(), "value sync must fail");
1827 });
1828
1829 deterministic::Runner::from(checkpoint).start(|context| async move {
1830 let oversized: Oversized<_, TestEntry, TestValue> =
1831 Oversized::init(context.child("second"), test_cfg(&context))
1832 .await
1833 .expect("Failed to reinit");
1834 assert_eq!(
1835 oversized.size(1).expect("size"),
1836 0,
1837 "entry without durable value bytes must be rewound"
1838 );
1839 oversized.destroy().await.expect("Failed to destroy");
1840 });
1841 }
1842
1843 #[test_traced]
1844 fn test_oversized_start_sync_values_failure_recovers_clean() {
1845 let executor = deterministic::Runner::default();
1846 let (_, checkpoint) = executor.start_and_recover(|context| async move {
1847 let faulty_values = SyncFaultContext {
1848 inner: context.child("first"),
1849 fail_partition: "test-values".into(),
1850 };
1851 let mut oversized: Oversized<_, TestEntry, TestValue> =
1852 Oversized::init(faulty_values, test_cfg(&context))
1853 .await
1854 .expect("Failed to init");
1855 (oversized, _, _, _) = oversized
1856 .append(1, TestEntry::new(1, 0, 0), &[1; 16])
1857 .await
1858 .expect("Failed to append");
1859
1860 let (_oversized, handle) = oversized.start_sync(1).await.expect("Failed to start sync");
1863 assert!(handle.await.is_err(), "value sync must fail");
1864 });
1865
1866 deterministic::Runner::from(checkpoint).start(|context| async move {
1867 let oversized: Oversized<_, TestEntry, TestValue> =
1868 Oversized::init(context.child("second"), test_cfg(&context))
1869 .await
1870 .expect("Failed to reinit");
1871 assert_eq!(
1872 oversized.size(1).expect("size"),
1873 0,
1874 "entry without durable value bytes must be rewound"
1875 );
1876 oversized.destroy().await.expect("Failed to destroy");
1877 });
1878 }
1879
1880 #[test_traced]
1881 fn test_oversized_start_sync_dropped_handle_driven_by_next_sync() {
1882 let executor = deterministic::Runner::default();
1883 let (_, checkpoint) = executor.start_and_recover(|context| async move {
1884 let mut oversized: Oversized<_, TestEntry, TestValue> =
1885 Oversized::init(context.child("first"), test_cfg(&context))
1886 .await
1887 .expect("Failed to init");
1888 (oversized, _, _, _) = oversized
1889 .append(1, TestEntry::new(1, 0, 0), &[1; 16])
1890 .await
1891 .expect("Failed to append");
1892
1893 let (oversized, handle) = oversized.start_sync(1).await.expect("Failed to start sync");
1896 drop(handle);
1897 oversized.sync(1).await.expect("Failed to sync");
1898 });
1899
1900 deterministic::Runner::from(checkpoint).start(|context| async move {
1901 let oversized: Oversized<_, TestEntry, TestValue> =
1902 Oversized::init(context.child("second"), test_cfg(&context))
1903 .await
1904 .expect("Failed to reinit");
1905 let chunk = FixedJournal::<deterministic::Context, TestEntry>::CHUNK_SIZE as u64;
1906 assert_eq!(oversized.size(1).expect("size"), chunk);
1907 assert_adopted_entries_consistent(&oversized).await;
1908 oversized.destroy().await.expect("Failed to destroy");
1909 });
1910 }
1911
1912 #[test_traced]
1913 fn test_oversized_recovery_rejects_entry_with_torn_value_bytes() {
1914 let executor = deterministic::Runner::default();
1915 let (_, checkpoint) = executor.start_and_recover(|context| async move {
1916 let mut oversized: Oversized<_, TestEntry, TestValue> =
1918 Oversized::init(context.child("first"), test_cfg(&context))
1919 .await
1920 .expect("Failed to init");
1921 (oversized, _, _, _) = oversized
1922 .append(1, TestEntry::new(1, 0, 0), &[1; 16])
1923 .await
1924 .expect("Failed to append");
1925 oversized = oversized.sync(1).await.expect("Failed to sync");
1926
1927 let (offset, size);
1931 (oversized, _, offset, size) = oversized
1932 .append(1, TestEntry::new(2, 0, 0), &[2; 16])
1933 .await
1934 .expect("Failed to append");
1935 oversized.index = oversized.index.sync(1).await.expect("Failed to sync index");
1936 oversized
1937 .values
1938 .inject(1, offset, vec![0xFF; size as usize])
1939 .await
1940 .expect("Failed to overwrite value bytes");
1941 oversized.values = oversized
1942 .values
1943 .sync(1)
1944 .await
1945 .expect("Failed to sync values");
1946 });
1947
1948 deterministic::Runner::from(checkpoint).start(|context| async move {
1950 let oversized: Oversized<_, TestEntry, TestValue> =
1951 Oversized::init(context.child("second"), test_cfg(&context))
1952 .await
1953 .expect("Failed to reinit");
1954 let chunk = FixedJournal::<deterministic::Context, TestEntry>::CHUNK_SIZE as u64;
1955 assert_eq!(
1956 oversized.size(1).expect("size"),
1957 chunk,
1958 "entry with torn value bytes must be rewound"
1959 );
1960 assert_adopted_entries_consistent(&oversized).await;
1961 oversized.destroy().await.expect("Failed to destroy");
1962 });
1963 }
1964
1965 #[test_traced]
1966 fn test_recovery_scans_past_torn_interior_index_page() {
1967 let executor = deterministic::Runner::default();
1968 executor.start(|context| async move {
1969 let cfg = entry_cfg(&context);
1971
1972 let mut oversized: Oversized<_, TestEntry, TestValue> =
1974 Oversized::init(context.child("first"), cfg.clone())
1975 .await
1976 .expect("Failed to init");
1977 for i in 1..=5u8 {
1978 (oversized, _, _, _) = oversized
1979 .append(1, TestEntry::new(i as u64, 0, 0), &[i; 16])
1980 .await
1981 .expect("Failed to append");
1982 }
1983 oversized = oversized.sync(1).await.expect("Failed to sync");
1984 drop(oversized);
1985
1986 let physical_page = TestEntry::SIZE as u64 + 12;
1989 let (index_blob, size) = context
1990 .open(&cfg.index_partition, &1u64.to_be_bytes())
1991 .await
1992 .expect("Failed to open index blob");
1993 assert_eq!(size, 5 * physical_page);
1994 index_blob
1995 .write_at(
1996 2 * physical_page + TestEntry::SIZE as u64,
1997 vec![0xFF; 12],
1998 WriteOptions::SYNC,
1999 )
2000 .await
2001 .expect("Failed to corrupt index page");
2002 drop(index_blob);
2003
2004 let (values_blob, _) = context
2007 .open(&cfg.value_partition, &1u64.to_be_bytes())
2008 .await
2009 .expect("Failed to open values blob");
2010 values_blob
2011 .write_at(60, vec![0xFF; 20], WriteOptions::SYNC)
2012 .await
2013 .expect("Failed to corrupt value");
2014 values_blob
2015 .write_at(80, vec![0xFF; 20], WriteOptions::SYNC)
2016 .await
2017 .expect("Failed to corrupt value");
2018 drop(values_blob);
2019
2020 let oversized: Oversized<_, TestEntry, TestValue> =
2023 Oversized::init(context.child("second"), cfg)
2024 .await
2025 .expect("Failed to reinit");
2026 let chunk = FixedJournal::<deterministic::Context, TestEntry>::CHUNK_SIZE as u64;
2027 assert_eq!(oversized.size(1).expect("size"), 2 * chunk);
2028 assert_adopted_entries_consistent(&oversized).await;
2029 oversized.destroy().await.expect("Failed to destroy");
2030 });
2031 }
2032
2033 #[test_traced]
2034 fn test_oversized_restore_discards_beyond_checkpoint() {
2035 let executor = deterministic::Runner::default();
2036 let (_, checkpoint) = executor.start_and_recover(|context| async move {
2037 let mut oversized: Oversized<_, TestEntry, TestValue> =
2040 Oversized::init(context.child("first"), test_cfg(&context))
2041 .await
2042 .expect("Failed to init");
2043 (oversized, _, _, _) = oversized
2044 .append(1, TestEntry::new(1, 0, 0), &[1; 16])
2045 .await
2046 .expect("Failed to append");
2047 oversized = oversized.sync(1).await.expect("Failed to sync");
2048 (oversized, _, _, _) = oversized
2049 .append(1, TestEntry::new(2, 0, 0), &[2; 16])
2050 .await
2051 .expect("Failed to append");
2052 (oversized, _, _, _) = oversized
2053 .append(2, TestEntry::new(3, 0, 0), &[3; 16])
2054 .await
2055 .expect("Failed to append");
2056 oversized.index = oversized
2057 .index
2058 .sync(&[1, 2])
2059 .await
2060 .expect("Failed to sync index");
2061 });
2062
2063 deterministic::Runner::from(checkpoint).start(|context| async move {
2065 let chunk = FixedJournal::<deterministic::Context, TestEntry>::CHUNK_SIZE as u64;
2066 let oversized: Oversized<_, TestEntry, TestValue> = Oversized::init_with_checkpoint(
2067 context.child("second"),
2068 test_cfg(&context),
2069 (1, chunk),
2070 )
2071 .await
2072 .expect("Failed to reinit");
2073 assert_eq!(oversized.size(1).expect("size"), chunk);
2074 assert_eq!(oversized.newest_section(), Some(1));
2075 assert_adopted_entries_consistent(&oversized).await;
2076 oversized.destroy().await.expect("Failed to destroy");
2077 });
2078 }
2079
2080 #[test_traced]
2081 fn test_oversized_restore_does_not_repair_discarded_sections() {
2082 let executor = deterministic::Runner::default();
2083 executor.start(|context| async move {
2084 let cfg = entry_cfg(&context);
2085 let mut oversized: Oversized<_, TestEntry, TestValue> =
2086 Oversized::init(context.child("seed"), cfg.clone())
2087 .await
2088 .expect("failed to init");
2089 (oversized, _, _, _) = oversized
2090 .append(1, TestEntry::new(0, 0, 0), &[0; 16])
2091 .await
2092 .expect("failed to append checkpoint entry");
2093 for id in 1..=3 {
2094 (oversized, _, _, _) = oversized
2095 .append(2, TestEntry::new(id, 0, 0), &[id as u8; 16])
2096 .await
2097 .expect("failed to append discardable entry");
2098 }
2099 oversized = oversized.sync_all().await.expect("failed to sync");
2100 drop(oversized);
2101
2102 corrupt_page(
2105 &context,
2106 &cfg.index_partition,
2107 &2u64.to_be_bytes(),
2108 1,
2109 TestEntry::SIZE as u64,
2110 )
2111 .await;
2112
2113 let pending = PendingSyncs::default();
2114 pending.arm();
2115 let delayed = DelayedSyncContext {
2116 inner: context,
2117 pending: pending.clone(),
2118 };
2119 let chunk = TestEntry::SIZE as u64;
2120 let oversized: Oversized<_, TestEntry, TestValue> = drive_pending_syncs(
2121 &pending,
2122 Oversized::init_with_checkpoint(delayed.child("restore"), cfg, (1, chunk)),
2123 )
2124 .await
2125 .expect("checkpoint restore failed");
2126
2127 assert_eq!(pending.calls(), 2);
2130 oversized.destroy().await.expect("failed to destroy");
2131 });
2132 }
2133
2134 #[test_traced]
2135 fn test_oversized_restore_incomplete_section_errors() {
2136 let executor = deterministic::Runner::default();
2137 executor.start(|context| async move {
2138 let mut oversized: Oversized<_, TestEntry, TestValue> =
2140 Oversized::init(context.child("first"), test_cfg(&context))
2141 .await
2142 .expect("Failed to init");
2143 (oversized, _, _, _) = oversized
2144 .append(1, TestEntry::new(1, 0, 0), &[1; 16])
2145 .await
2146 .expect("Failed to append");
2147 (oversized, _, _, _) = oversized
2148 .append(2, TestEntry::new(2, 0, 0), &[2; 16])
2149 .await
2150 .expect("Failed to append");
2151 oversized = oversized.sync_all().await.expect("Failed to sync");
2152 drop(oversized);
2153
2154 let (blob, len) = context
2157 .open("test-values", &1u64.to_be_bytes())
2158 .await
2159 .expect("Failed to open values blob");
2160 blob.resize(len - 1).await.expect("Failed to resize");
2161 blob.sync().await.expect("Failed to sync");
2162 drop(blob);
2163
2164 let chunk = FixedJournal::<deterministic::Context, TestEntry>::CHUNK_SIZE as u64;
2167 for instance in ["second", "third"] {
2168 let result: Result<Oversized<_, TestEntry, TestValue>, Error> =
2169 Oversized::init_with_checkpoint(
2170 context.child(instance),
2171 test_cfg(&context),
2172 (2, chunk),
2173 )
2174 .await;
2175 assert!(matches!(result, Err(Error::Corruption(_))));
2176 }
2177 });
2178 }
2179
2180 #[test_traced]
2181 fn test_oversized_restore_adopts_interior_index_corruption() {
2182 let executor = deterministic::Runner::default();
2183 executor.start(|context| async move {
2184 for (child, seed_child, restore_child, checkpoint) in [
2185 (
2186 "checkpoint_section",
2187 "seed_checkpoint_section",
2188 "restore_checkpoint_section",
2189 (1, 3 * TestEntry::SIZE as u64),
2190 ),
2191 (
2192 "earlier_section",
2193 "seed_earlier_section",
2194 "restore_earlier_section",
2195 (2, TestEntry::SIZE as u64),
2196 ),
2197 ] {
2198 let mut cfg = entry_cfg(&context);
2199 cfg.index_partition = format!("test-index-{child}");
2200 cfg.value_partition = format!("test-values-{child}");
2201
2202 let mut oversized: Oversized<_, TestEntry, TestValue> =
2205 Oversized::init(context.child(seed_child), cfg.clone())
2206 .await
2207 .expect("failed to init");
2208 for id in 0..3 {
2209 (oversized, _, _, _) = oversized
2210 .append(1, TestEntry::new(id, 0, 0), &[id as u8; 16])
2211 .await
2212 .expect("failed to append");
2213 }
2214 (oversized, _, _, _) = oversized
2215 .append(2, TestEntry::new(3, 0, 0), &[3; 16])
2216 .await
2217 .expect("failed to append later section");
2218 oversized = oversized.sync_all().await.expect("failed to sync");
2219 drop(oversized);
2220
2221 corrupt_page(
2224 &context,
2225 &cfg.index_partition,
2226 &1u64.to_be_bytes(),
2227 1,
2228 TestEntry::SIZE as u64,
2229 )
2230 .await;
2231 let (blob, expected_size) = context
2232 .open(&cfg.index_partition, &1u64.to_be_bytes())
2233 .await
2234 .expect("failed to open index");
2235 let expected = blob
2236 .read_at(0, expected_size as usize, ReadOptions::default())
2237 .await
2238 .expect("failed to snapshot damaged index")
2239 .coalesce();
2240 drop(blob);
2241
2242 let oversized: Oversized<_, TestEntry, TestValue> =
2243 Oversized::init_with_checkpoint(
2244 context.child(restore_child),
2245 cfg.clone(),
2246 checkpoint,
2247 )
2248 .await
2249 .expect("checkpoint restore should adopt the covered prefix");
2250 assert!(matches!(
2251 oversized.get(1, 1).await,
2252 Err(Error::Runtime(RError::InvalidChecksum))
2253 ));
2254 drop(oversized);
2255
2256 let (blob, actual_size) = context
2257 .open(&cfg.index_partition, &1u64.to_be_bytes())
2258 .await
2259 .expect("failed to reopen index");
2260 assert_eq!(actual_size, expected_size);
2261 let actual = blob
2262 .read_at(0, actual_size as usize, ReadOptions::default())
2263 .await
2264 .expect("failed to read damaged index")
2265 .coalesce();
2266 assert_eq!(actual.as_ref(), expected.as_ref());
2267 }
2268 });
2269 }
2270
2271 #[test_traced]
2272 fn test_oversized_restore_adopts_rotted_committed_value() {
2273 let executor = deterministic::Runner::default();
2274 executor.start(|context| async move {
2275 let mut oversized: Oversized<_, TestEntry, TestValue> =
2277 Oversized::init(context.child("first"), test_cfg(&context))
2278 .await
2279 .expect("Failed to init");
2280 let (offset, size);
2281 (oversized, _, offset, size) = oversized
2282 .append(1, TestEntry::new(1, 0, 0), &[1; 16])
2283 .await
2284 .expect("Failed to append");
2285 (oversized, _, _, _) = oversized
2286 .append(2, TestEntry::new(2, 0, 0), &[2; 16])
2287 .await
2288 .expect("Failed to append");
2289 oversized = oversized.sync_all().await.expect("Failed to sync");
2290
2291 oversized
2293 .values
2294 .inject(1, offset, vec![0xFF; size as usize])
2295 .await
2296 .expect("Failed to corrupt value");
2297 oversized.values = oversized.values.sync(1).await.expect("Failed to sync");
2298 drop(oversized);
2299
2300 let chunk = FixedJournal::<deterministic::Context, TestEntry>::CHUNK_SIZE as u64;
2303 let oversized: Oversized<_, TestEntry, TestValue> = Oversized::init_with_checkpoint(
2304 context.child("second"),
2305 test_cfg(&context),
2306 (2, chunk),
2307 )
2308 .await
2309 .expect("Failed to reinit");
2310 assert!(matches!(
2311 oversized.get_value(1, offset, size).await,
2312 Err(Error::ChecksumMismatch(_, _))
2313 ));
2314 let entry = oversized.get(2, 0).await.expect("Failed to get");
2315 let (offset, size) = entry.value_location();
2316 assert_eq!(
2317 oversized
2318 .get_value(2, offset, size)
2319 .await
2320 .expect("Failed to get value"),
2321 [2; 16]
2322 );
2323 oversized.destroy().await.expect("Failed to destroy");
2324 });
2325 }
2326
2327 #[test_traced]
2328 fn test_oversized_replay_empty_finishes_immediately() {
2329 let executor = deterministic::Runner::default();
2330 executor.start(|context| async move {
2331 let cfg = test_cfg(&context);
2332 let oversized: Oversized<_, TestEntry, TestValue> =
2333 Oversized::init(context, cfg).await.expect("Failed to init");
2334
2335 let replay = oversized
2337 .replay(0, 0, NZUsize!(1024), ReadOptions::default())
2338 .await
2339 .expect("Failed to replay");
2340 let oversized = replay.finish().expect("failed to finish replay");
2341 oversized.destroy().await.expect("Failed to destroy");
2342 });
2343 }
2344
2345 #[test_traced]
2346 fn test_oversized_replay_propagates_read_options() {
2347 let executor = deterministic::Runner::default();
2348 executor.start(|context| async move {
2349 let (context, recordings) = commonware_runtime::mocks::RecordingContext::new(context);
2350 let cfg = test_cfg(&context);
2351 let page_cache = cfg.index_page_cache.clone();
2352 let mut oversized: Oversized<_, TestEntry, TestValue> =
2353 Oversized::init(context, cfg).await.expect("Failed to init");
2354 (oversized, _, _, _) = oversized
2355 .append(1, TestEntry::new(1, 0, 0), &[1; 16])
2356 .await
2357 .expect("Failed to append");
2358 oversized = oversized.sync(1).await.expect("Failed to sync");
2359
2360 page_cache.clear();
2362 let mut replay = oversized
2363 .replay(1, 0, NZUsize!(1024), ReadOptions::DONT_CACHE)
2364 .await
2365 .expect("Failed to replay");
2366 recordings.clear();
2367
2368 let (section, position, entry) = replay
2370 .next()
2371 .await
2372 .expect("missing replay item")
2373 .expect("Failed to read replay item");
2374 assert_eq!((section, position, entry.id), (1, 0, 1));
2375
2376 let reads = recordings.snapshot().reads;
2377 assert!(!reads.is_empty());
2378 assert!(
2379 reads
2380 .iter()
2381 .all(|options| *options == ReadOptions::DONT_CACHE)
2382 );
2383
2384 assert!(replay.next().await.is_none());
2385 replay
2386 .finish()
2387 .expect("failed to finish replay")
2388 .destroy()
2389 .await
2390 .expect("Failed to destroy");
2391 });
2392 }
2393
2394 #[test_traced]
2395 fn test_oversized_replay_finish_before_drain_fails() {
2396 let executor = deterministic::Runner::default();
2397 executor.start(|context| async move {
2398 let cfg = test_cfg(&context);
2399 let mut oversized: Oversized<_, TestEntry, TestValue> =
2400 Oversized::init(context, cfg).await.expect("Failed to init");
2401 (oversized, _, _, _) = oversized
2402 .append(1, TestEntry::new(1, 0, 0), &[1; 16])
2403 .await
2404 .expect("Failed to append");
2405 oversized = oversized.sync(1).await.expect("Failed to sync");
2406
2407 let replay = oversized
2408 .replay(0, 0, NZUsize!(1024), ReadOptions::default())
2409 .await
2410 .expect("Failed to replay");
2411 assert!(matches!(replay.finish(), Err(Error::ReplayFailed)));
2412 });
2413 }
2414
2415 #[test_traced]
2416 fn test_oversized_prune() {
2417 let executor = deterministic::Runner::default();
2418 executor.start(|context| async move {
2419 let cfg = test_cfg(&context);
2420 let mut oversized: Oversized<_, TestEntry, TestValue> =
2421 Oversized::init(context, cfg).await.expect("Failed to init");
2422
2423 for section in 1u64..=5 {
2425 let value: TestValue = [section as u8; 16];
2426 let entry = TestEntry::new(section, 0, 0);
2427 (oversized, _, _, _) = oversized
2428 .append(section, entry, &value)
2429 .await
2430 .expect("Failed to append");
2431 oversized = oversized.sync(section).await.expect("Failed to sync");
2432 }
2433
2434 (oversized, _) = oversized.prune(3).await.expect("Failed to prune");
2436
2437 assert!(oversized.pruned(1));
2439 assert!(oversized.pruned(2));
2440 assert!(!oversized.pruned(3));
2441
2442 assert!(oversized.get(1, 0).await.is_err());
2444 assert!(oversized.get(2, 0).await.is_err());
2445
2446 assert!(oversized.get(3, 0).await.is_ok());
2448 assert!(oversized.get(4, 0).await.is_ok());
2449 assert!(oversized.get(5, 0).await.is_ok());
2450
2451 oversized.destroy().await.expect("Failed to destroy");
2452 });
2453 }
2454
2455 #[test_traced]
2456 fn test_recovery_empty_section() {
2457 let executor = deterministic::Runner::default();
2458 executor.start(|context| async move {
2459 let cfg = test_cfg(&context);
2460
2461 let mut oversized: Oversized<_, TestEntry, TestValue> =
2463 Oversized::init(context.child("first"), cfg.clone())
2464 .await
2465 .expect("Failed to init");
2466
2467 let value: TestValue = [42; 16];
2469 let entry = TestEntry::new(1, 0, 0);
2470 (oversized, _, _, _) = oversized
2471 .append(2, entry, &value)
2472 .await
2473 .expect("Failed to append");
2474 oversized = oversized.sync(2).await.expect("Failed to sync");
2475 drop(oversized);
2476
2477 let oversized: Oversized<_, TestEntry, TestValue> =
2479 Oversized::init(context.child("second"), cfg)
2480 .await
2481 .expect("Failed to reinit");
2482
2483 let entry = oversized.get(2, 0).await.expect("Failed to get");
2485 assert_eq!(entry.id, 1);
2486
2487 oversized.destroy().await.expect("Failed to destroy");
2488 });
2489 }
2490
2491 #[test_traced]
2492 fn test_recovery_all_entries_invalid() {
2493 let executor = deterministic::Runner::default();
2494 executor.start(|context| async move {
2495 let cfg = test_cfg(&context);
2496
2497 let mut oversized: Oversized<_, TestEntry, TestValue> =
2499 Oversized::init(context.child("first"), cfg.clone())
2500 .await
2501 .expect("Failed to init");
2502
2503 for i in 0..5u8 {
2505 let value: TestValue = [i; 16];
2506 let entry = TestEntry::new(i as u64, 0, 0);
2507 (oversized, _, _, _) = oversized
2508 .append(1, entry, &value)
2509 .await
2510 .expect("Failed to append");
2511 }
2512 oversized = oversized.sync(1).await.expect("Failed to sync");
2513 drop(oversized);
2514
2515 let (blob, _) = context
2517 .open(&cfg.value_partition, &1u64.to_be_bytes())
2518 .await
2519 .expect("Failed to open blob");
2520 blob.resize(0).await.expect("Failed to truncate");
2521 blob.sync().await.expect("Failed to sync");
2522 drop(blob);
2523
2524 let mut oversized: Oversized<_, TestEntry, TestValue> =
2526 Oversized::init(context.child("second"), cfg)
2527 .await
2528 .expect("Failed to reinit");
2529
2530 let result = oversized.get(1, 0).await;
2532 assert!(result.is_err());
2533
2534 let value: TestValue = [99; 16];
2536 let entry = TestEntry::new(100, 0, 0);
2537 let (pos, offset, size);
2538 (oversized, pos, offset, size) = oversized
2539 .append(1, entry, &value)
2540 .await
2541 .expect("Failed to append after recovery");
2542 assert_eq!(pos, 0);
2543
2544 let retrieved = oversized.get(1, 0).await.expect("Failed to get");
2545 assert_eq!(retrieved.id, 100);
2546 let retrieved_value = oversized
2547 .get_value(1, offset, size)
2548 .await
2549 .expect("Failed to get value");
2550 assert_eq!(retrieved_value, value);
2551
2552 oversized.destroy().await.expect("Failed to destroy");
2553 });
2554 }
2555
2556 #[test_traced]
2557 fn test_recovery_multiple_sections_mixed_validity() {
2558 let executor = deterministic::Runner::default();
2559 executor.start(|context| async move {
2560 let cfg = test_cfg(&context);
2561
2562 let mut oversized: Oversized<_, TestEntry, TestValue> =
2564 Oversized::init(context.child("first"), cfg.clone())
2565 .await
2566 .expect("Failed to init");
2567
2568 let mut section1_locations = Vec::new();
2570 for i in 0..3u8 {
2571 let value: TestValue = [i; 16];
2572 let entry = TestEntry::new(i as u64, 0, 0);
2573 let (position, offset, size);
2574 (oversized, position, offset, size) = oversized
2575 .append(1, entry, &value)
2576 .await
2577 .expect("Failed to append");
2578 section1_locations.push((position, offset, size));
2579 }
2580 oversized = oversized.sync(1).await.expect("Failed to sync");
2581
2582 let mut section2_locations = Vec::new();
2584 for i in 0..5u8 {
2585 let value: TestValue = [10 + i; 16];
2586 let entry = TestEntry::new(10 + i as u64, 0, 0);
2587 let (position, offset, size);
2588 (oversized, position, offset, size) = oversized
2589 .append(2, entry, &value)
2590 .await
2591 .expect("Failed to append");
2592 section2_locations.push((position, offset, size));
2593 }
2594 oversized = oversized.sync(2).await.expect("Failed to sync");
2595
2596 for i in 0..2u8 {
2598 let value: TestValue = [20 + i; 16];
2599 let entry = TestEntry::new(20 + i as u64, 0, 0);
2600 (oversized, _, _, _) = oversized
2601 .append(3, entry, &value)
2602 .await
2603 .expect("Failed to append");
2604 }
2605 oversized = oversized.sync(3).await.expect("Failed to sync");
2606 drop(oversized);
2607
2608 let (blob, _) = context
2610 .open(&cfg.value_partition, &1u64.to_be_bytes())
2611 .await
2612 .expect("Failed to open blob");
2613 let keep_size = byte_end(section1_locations[0].1, section1_locations[0].2);
2614 blob.resize(keep_size).await.expect("Failed to truncate");
2615 blob.sync().await.expect("Failed to sync");
2616 drop(blob);
2617
2618 let (blob, _) = context
2620 .open(&cfg.value_partition, &2u64.to_be_bytes())
2621 .await
2622 .expect("Failed to open blob");
2623 let keep_size = byte_end(section2_locations[2].1, section2_locations[2].2);
2624 blob.resize(keep_size).await.expect("Failed to truncate");
2625 blob.sync().await.expect("Failed to sync");
2626 drop(blob);
2627
2628 let oversized: Oversized<_, TestEntry, TestValue> =
2632 Oversized::init(context.child("second"), cfg)
2633 .await
2634 .expect("Failed to reinit");
2635
2636 assert!(oversized.get(1, 0).await.is_ok());
2638 assert!(oversized.get(1, 1).await.is_err());
2639 assert!(oversized.get(1, 2).await.is_err());
2640
2641 assert!(oversized.get(2, 0).await.is_ok());
2643 assert!(oversized.get(2, 1).await.is_ok());
2644 assert!(oversized.get(2, 2).await.is_ok());
2645 assert!(oversized.get(2, 3).await.is_err());
2646 assert!(oversized.get(2, 4).await.is_err());
2647
2648 assert!(oversized.get(3, 0).await.is_ok());
2650 assert!(oversized.get(3, 1).await.is_ok());
2651
2652 oversized.destroy().await.expect("Failed to destroy");
2653 });
2654 }
2655
2656 #[test_traced]
2657 fn test_recovery_corrupted_last_index_entry() {
2658 let executor = deterministic::Runner::default();
2659 executor.start(|context| async move {
2660 let cfg = entry_cfg(&context);
2664
2665 let mut oversized: Oversized<_, TestEntry, TestValue> =
2667 Oversized::init(context.child("first"), cfg.clone())
2668 .await
2669 .expect("Failed to init");
2670
2671 for i in 0..5u8 {
2673 let value: TestValue = [i; 16];
2674 let entry = TestEntry::new(i as u64, 0, 0);
2675 (oversized, _, _, _) = oversized
2676 .append(1, entry, &value)
2677 .await
2678 .expect("Failed to append");
2679 }
2680 oversized = oversized.sync(1).await.expect("Failed to sync");
2681 drop(oversized);
2682
2683 let (blob, size) = context
2685 .open(&cfg.index_partition, &1u64.to_be_bytes())
2686 .await
2687 .expect("Failed to open blob");
2688
2689 assert_eq!(size, 160);
2693 let last_page_crc_offset = size - 12;
2694 blob.write_at(last_page_crc_offset, vec![0xFF; 12], WriteOptions::SYNC)
2695 .await
2696 .expect("Failed to corrupt");
2697 drop(blob);
2698
2699 let mut oversized: Oversized<_, TestEntry, TestValue> =
2701 Oversized::init(context.child("second"), cfg)
2702 .await
2703 .expect("Failed to reinit");
2704
2705 for i in 0..4u8 {
2707 let entry = oversized.get(1, i as u64).await.expect("Failed to get");
2708 assert_eq!(entry.id, i as u64);
2709 }
2710
2711 assert!(oversized.get(1, 4).await.is_err());
2713
2714 let value: TestValue = [99; 16];
2716 let entry = TestEntry::new(100, 0, 0);
2717 let (pos, offset, size);
2718 (oversized, pos, offset, size) = oversized
2719 .append(1, entry, &value)
2720 .await
2721 .expect("Failed to append after recovery");
2722 assert_eq!(pos, 4);
2723
2724 let retrieved = oversized.get(1, 4).await.expect("Failed to get");
2725 assert_eq!(retrieved.id, 100);
2726 let retrieved_value = oversized
2727 .get_value(1, offset, size)
2728 .await
2729 .expect("Failed to get value");
2730 assert_eq!(retrieved_value, value);
2731
2732 oversized.destroy().await.expect("Failed to destroy");
2733 });
2734 }
2735
2736 #[test_traced]
2737 fn test_recovery_all_entries_valid() {
2738 let executor = deterministic::Runner::default();
2739 executor.start(|context| async move {
2740 let cfg = test_cfg(&context);
2741
2742 let mut oversized: Oversized<_, TestEntry, TestValue> =
2744 Oversized::init(context.child("first"), cfg.clone())
2745 .await
2746 .expect("Failed to init");
2747
2748 for section in 1u64..=3 {
2750 for i in 0..10u8 {
2751 let value: TestValue = [(section as u8) * 10 + i; 16];
2752 let entry = TestEntry::new(section * 100 + i as u64, 0, 0);
2753 (oversized, _, _, _) = oversized
2754 .append(section, entry, &value)
2755 .await
2756 .expect("Failed to append");
2757 }
2758 oversized = oversized.sync(section).await.expect("Failed to sync");
2759 }
2760 drop(oversized);
2761
2762 let oversized: Oversized<_, TestEntry, TestValue> =
2764 Oversized::init(context.child("second"), cfg)
2765 .await
2766 .expect("Failed to reinit");
2767
2768 for section in 1u64..=3 {
2770 for i in 0..10u8 {
2771 let entry = oversized
2772 .get(section, i as u64)
2773 .await
2774 .expect("Failed to get");
2775 assert_eq!(entry.id, section * 100 + i as u64);
2776 }
2777 }
2778
2779 oversized.destroy().await.expect("Failed to destroy");
2780 });
2781 }
2782
2783 #[test_traced]
2784 fn test_recovery_single_entry_invalid() {
2785 let executor = deterministic::Runner::default();
2786 executor.start(|context| async move {
2787 let cfg = test_cfg(&context);
2788
2789 let mut oversized: Oversized<_, TestEntry, TestValue> =
2791 Oversized::init(context.child("first"), cfg.clone())
2792 .await
2793 .expect("Failed to init");
2794
2795 let value: TestValue = [42; 16];
2796 let entry = TestEntry::new(1, 0, 0);
2797 (oversized, _, _, _) = oversized
2798 .append(1, entry, &value)
2799 .await
2800 .expect("Failed to append");
2801 oversized = oversized.sync(1).await.expect("Failed to sync");
2802 drop(oversized);
2803
2804 let (blob, _) = context
2806 .open(&cfg.value_partition, &1u64.to_be_bytes())
2807 .await
2808 .expect("Failed to open blob");
2809 blob.resize(0).await.expect("Failed to truncate");
2810 blob.sync().await.expect("Failed to sync");
2811 drop(blob);
2812
2813 let oversized: Oversized<_, TestEntry, TestValue> =
2815 Oversized::init(context.child("second"), cfg)
2816 .await
2817 .expect("Failed to reinit");
2818
2819 assert!(oversized.get(1, 0).await.is_err());
2821
2822 oversized.destroy().await.expect("Failed to destroy");
2823 });
2824 }
2825
2826 #[test_traced]
2827 fn test_recovery_last_entry_off_by_one() {
2828 let executor = deterministic::Runner::default();
2829 executor.start(|context| async move {
2830 let cfg = test_cfg(&context);
2831
2832 let mut oversized: Oversized<_, TestEntry, TestValue> =
2834 Oversized::init(context.child("first"), cfg.clone())
2835 .await
2836 .expect("Failed to init");
2837
2838 let mut locations = Vec::new();
2839 for i in 0..3u8 {
2840 let value: TestValue = [i; 16];
2841 let entry = TestEntry::new(i as u64, 0, 0);
2842 let (position, offset, size);
2843 (oversized, position, offset, size) = oversized
2844 .append(1, entry, &value)
2845 .await
2846 .expect("Failed to append");
2847 locations.push((position, offset, size));
2848 }
2849 oversized = oversized.sync(1).await.expect("Failed to sync");
2850 drop(oversized);
2851
2852 let (blob, _) = context
2854 .open(&cfg.value_partition, &1u64.to_be_bytes())
2855 .await
2856 .expect("Failed to open blob");
2857
2858 let last = &locations[2];
2861 let truncate_to = byte_end(last.1, last.2) - 1;
2862 blob.resize(truncate_to).await.expect("Failed to truncate");
2863 blob.sync().await.expect("Failed to sync");
2864 drop(blob);
2865
2866 let mut oversized: Oversized<_, TestEntry, TestValue> =
2868 Oversized::init(context.child("second"), cfg)
2869 .await
2870 .expect("Failed to reinit");
2871
2872 assert!(oversized.get(1, 0).await.is_ok());
2874 assert!(oversized.get(1, 1).await.is_ok());
2875
2876 assert!(oversized.get(1, 2).await.is_err());
2878
2879 let value: TestValue = [99; 16];
2881 let entry = TestEntry::new(100, 0, 0);
2882 let (pos, offset, size);
2883 (oversized, pos, offset, size) = oversized
2884 .append(1, entry, &value)
2885 .await
2886 .expect("Failed to append after recovery");
2887 assert_eq!(pos, 2);
2888
2889 let retrieved = oversized.get(1, 2).await.expect("Failed to get");
2890 assert_eq!(retrieved.id, 100);
2891 let retrieved_value = oversized
2892 .get_value(1, offset, size)
2893 .await
2894 .expect("Failed to get value");
2895 assert_eq!(retrieved_value, value);
2896
2897 oversized.destroy().await.expect("Failed to destroy");
2898 });
2899 }
2900
2901 #[test_traced]
2902 fn test_recovery_glob_missing_entirely() {
2903 let executor = deterministic::Runner::default();
2904 executor.start(|context| async move {
2905 let cfg = test_cfg(&context);
2906
2907 let mut oversized: Oversized<_, TestEntry, TestValue> =
2909 Oversized::init(context.child("first"), cfg.clone())
2910 .await
2911 .expect("Failed to init");
2912
2913 for i in 0..3u8 {
2914 let value: TestValue = [i; 16];
2915 let entry = TestEntry::new(i as u64, 0, 0);
2916 (oversized, _, _, _) = oversized
2917 .append(1, entry, &value)
2918 .await
2919 .expect("Failed to append");
2920 }
2921 oversized = oversized.sync(1).await.expect("Failed to sync");
2922 drop(oversized);
2923
2924 context
2926 .remove(&cfg.value_partition, Some(&1u64.to_be_bytes()))
2927 .await
2928 .expect("Failed to remove");
2929
2930 let oversized: Oversized<_, TestEntry, TestValue> =
2932 Oversized::init(context.child("second"), cfg)
2933 .await
2934 .expect("Failed to reinit");
2935
2936 assert!(oversized.get(1, 0).await.is_err());
2938 assert!(oversized.get(1, 1).await.is_err());
2939 assert!(oversized.get(1, 2).await.is_err());
2940
2941 oversized.destroy().await.expect("Failed to destroy");
2942 });
2943 }
2944
2945 #[test_traced]
2946 fn test_recovery_can_append_after_recovery() {
2947 let executor = deterministic::Runner::default();
2948 executor.start(|context| async move {
2949 let cfg = test_cfg(&context);
2950
2951 let mut oversized: Oversized<_, TestEntry, TestValue> =
2953 Oversized::init(context.child("first"), cfg.clone())
2954 .await
2955 .expect("Failed to init");
2956
2957 let mut locations = Vec::new();
2958 for i in 0..5u8 {
2959 let value: TestValue = [i; 16];
2960 let entry = TestEntry::new(i as u64, 0, 0);
2961 let (position, offset, size);
2962 (oversized, position, offset, size) = oversized
2963 .append(1, entry, &value)
2964 .await
2965 .expect("Failed to append");
2966 locations.push((position, offset, size));
2967 }
2968 oversized = oversized.sync(1).await.expect("Failed to sync");
2969 drop(oversized);
2970
2971 let (blob, _) = context
2973 .open(&cfg.value_partition, &1u64.to_be_bytes())
2974 .await
2975 .expect("Failed to open blob");
2976 let keep_size = byte_end(locations[1].1, locations[1].2);
2977 blob.resize(keep_size).await.expect("Failed to truncate");
2978 blob.sync().await.expect("Failed to sync");
2979 drop(blob);
2980
2981 let mut oversized: Oversized<_, TestEntry, TestValue> =
2983 Oversized::init(context.child("second"), cfg.clone())
2984 .await
2985 .expect("Failed to reinit");
2986
2987 assert!(oversized.get(1, 0).await.is_ok());
2989 assert!(oversized.get(1, 1).await.is_ok());
2990 assert!(oversized.get(1, 2).await.is_err());
2991
2992 for i in 10..15u8 {
2994 let value: TestValue = [i; 16];
2995 let entry = TestEntry::new(i as u64, 0, 0);
2996 (oversized, _, _, _) = oversized
2997 .append(1, entry, &value)
2998 .await
2999 .expect("Failed to append after recovery");
3000 }
3001 oversized = oversized.sync(1).await.expect("Failed to sync");
3002
3003 for i in 0..5u8 {
3005 let entry = oversized
3006 .get(1, 2 + i as u64)
3007 .await
3008 .expect("Failed to get new entry");
3009 assert_eq!(entry.id, (10 + i) as u64);
3010 }
3011
3012 oversized.destroy().await.expect("Failed to destroy");
3013 });
3014 }
3015
3016 #[test_traced]
3017 fn test_recovery_glob_pruned_but_index_not() {
3018 let executor = deterministic::Runner::default();
3019 executor.start(|context| async move {
3020 let cfg = test_cfg(&context);
3021
3022 let mut oversized: Oversized<_, TestEntry, TestValue> =
3024 Oversized::init(context.child("first"), cfg.clone())
3025 .await
3026 .expect("Failed to init");
3027
3028 for section in 1u64..=3 {
3029 let value: TestValue = [section as u8; 16];
3030 let entry = TestEntry::new(section, 0, 0);
3031 (oversized, _, _, _) = oversized
3032 .append(section, entry, &value)
3033 .await
3034 .expect("Failed to append");
3035 oversized = oversized.sync(section).await.expect("Failed to sync");
3036 }
3037 drop(oversized);
3038
3039 use crate::journal::segmented::glob::{Config as GlobConfig, Glob};
3042 let glob_cfg = GlobConfig {
3043 partition: cfg.value_partition.clone(),
3044 compression: cfg.compression,
3045 codec_config: (),
3046 write_buffer: cfg.value_write_buffer,
3047 };
3048 let mut glob: Glob<_, TestValue> = Glob::init(context.child("glob"), glob_cfg)
3049 .await
3050 .expect("Failed to init glob");
3051 (glob, _) = glob.prune(2).await.expect("Failed to prune glob");
3052 glob = glob.sync_all().await.expect("Failed to sync glob");
3053 drop(glob);
3054
3055 let oversized: Oversized<_, TestEntry, TestValue> =
3058 Oversized::init(context.child("second"), cfg.clone())
3059 .await
3060 .expect("Failed to reinit");
3061
3062 assert!(oversized.get(1, 0).await.is_err());
3064
3065 assert!(oversized.get(2, 0).await.is_ok());
3067 assert!(oversized.get(3, 0).await.is_ok());
3068
3069 oversized.destroy().await.expect("Failed to destroy");
3070 });
3071 }
3072
3073 #[test_traced]
3074 fn test_recovery_index_partition_deleted() {
3075 let executor = deterministic::Runner::default();
3076 executor.start(|context| async move {
3077 let cfg = test_cfg(&context);
3078
3079 let mut oversized: Oversized<_, TestEntry, TestValue> =
3081 Oversized::init(context.child("first"), cfg.clone())
3082 .await
3083 .expect("Failed to init");
3084
3085 for section in 1u64..=3 {
3086 let value: TestValue = [section as u8; 16];
3087 let entry = TestEntry::new(section, 0, 0);
3088 (oversized, _, _, _) = oversized
3089 .append(section, entry, &value)
3090 .await
3091 .expect("Failed to append");
3092 oversized = oversized.sync(section).await.expect("Failed to sync");
3093 }
3094 drop(oversized);
3095
3096 context
3098 .remove(&cfg.index_partition, Some(&2u64.to_be_bytes()))
3099 .await
3100 .expect("Failed to remove index");
3101
3102 let oversized: Oversized<_, TestEntry, TestValue> =
3105 Oversized::init(context.child("second"), cfg.clone())
3106 .await
3107 .expect("Failed to reinit");
3108
3109 assert!(oversized.get(1, 0).await.is_ok());
3111 assert!(oversized.get(3, 0).await.is_ok());
3112
3113 assert!(oversized.get(2, 0).await.is_err());
3115
3116 oversized.destroy().await.expect("Failed to destroy");
3117 });
3118 }
3119
3120 #[test_traced]
3121 fn test_recovery_index_synced_but_glob_not() {
3122 let executor = deterministic::Runner::default();
3123 executor.start(|context| async move {
3124 let cfg = test_cfg(&context);
3125
3126 let mut oversized: Oversized<_, TestEntry, TestValue> =
3128 Oversized::init(context.child("first"), cfg.clone())
3129 .await
3130 .expect("Failed to init");
3131
3132 let mut locations = Vec::new();
3134 for i in 0..3u8 {
3135 let value: TestValue = [i; 16];
3136 let entry = TestEntry::new(i as u64, 0, 0);
3137 let (position, offset, size);
3138 (oversized, position, offset, size) = oversized
3139 .append(1, entry, &value)
3140 .await
3141 .expect("Failed to append");
3142 locations.push((position, offset, size));
3143 }
3144 oversized = oversized.sync(1).await.expect("Failed to sync");
3145
3146 for i in 10..15u8 {
3148 let value: TestValue = [i; 16];
3149 let entry = TestEntry::new(i as u64, 0, 0);
3150 (oversized, _, _, _) = oversized
3151 .append(1, entry, &value)
3152 .await
3153 .expect("Failed to append");
3154 }
3155 drop(oversized);
3157
3158 let (blob, _) = context
3161 .open(&cfg.value_partition, &1u64.to_be_bytes())
3162 .await
3163 .expect("Failed to open blob");
3164 let synced_size = byte_end(locations[2].1, locations[2].2);
3165 blob.resize(synced_size).await.expect("Failed to truncate");
3166 blob.sync().await.expect("Failed to sync");
3167 drop(blob);
3168
3169 let oversized: Oversized<_, TestEntry, TestValue> =
3171 Oversized::init(context.child("second"), cfg)
3172 .await
3173 .expect("Failed to reinit");
3174
3175 for i in 0..3u8 {
3177 let entry = oversized.get(1, i as u64).await.expect("Failed to get");
3178 assert_eq!(entry.id, i as u64);
3179 }
3180
3181 assert!(oversized.get(1, 3).await.is_err());
3183
3184 oversized.destroy().await.expect("Failed to destroy");
3185 });
3186 }
3187
3188 #[test_traced]
3189 fn test_recovery_glob_synced_but_index_not() {
3190 let executor = deterministic::Runner::default();
3191 executor.start(|context| async move {
3192 let cfg = entry_cfg(&context);
3196
3197 let mut oversized: Oversized<_, TestEntry, TestValue> =
3199 Oversized::init(context.child("first"), cfg.clone())
3200 .await
3201 .expect("Failed to init");
3202
3203 let mut locations = Vec::new();
3205 for i in 0..3u8 {
3206 let value: TestValue = [i; 16];
3207 let entry = TestEntry::new(i as u64, 0, 0);
3208 let (position, offset, size);
3209 (oversized, position, offset, size) = oversized
3210 .append(1, entry, &value)
3211 .await
3212 .expect("Failed to append");
3213 locations.push((position, offset, size));
3214 }
3215 oversized = oversized.sync(1).await.expect("Failed to sync");
3216 drop(oversized);
3217
3218 let (blob, _size) = context
3221 .open(&cfg.index_partition, &1u64.to_be_bytes())
3222 .await
3223 .expect("Failed to open blob");
3224
3225 let physical_page_size = (TestEntry::SIZE + 12) as u64;
3228 blob.resize(2 * physical_page_size)
3229 .await
3230 .expect("Failed to truncate");
3231 blob.sync().await.expect("Failed to sync");
3232 drop(blob);
3233
3234 let mut oversized: Oversized<_, TestEntry, TestValue> =
3236 Oversized::init(context.child("second"), cfg.clone())
3237 .await
3238 .expect("Failed to reinit");
3239
3240 for i in 0..2u8 {
3242 let (position, offset, size) = locations[i as usize];
3243 let entry = oversized.get(1, position).await.expect("Failed to get");
3244 assert_eq!(entry.id, i as u64);
3245
3246 let value = oversized
3247 .get_value(1, offset, size)
3248 .await
3249 .expect("Failed to get value");
3250 assert_eq!(value, [i; 16]);
3251 }
3252
3253 assert!(oversized.get(1, 2).await.is_err());
3255
3256 let mut new_locations = Vec::new();
3258 for i in 10..13u8 {
3259 let value: TestValue = [i; 16];
3260 let entry = TestEntry::new(i as u64, 0, 0);
3261 let (position, offset, size);
3262 (oversized, position, offset, size) = oversized
3263 .append(1, entry, &value)
3264 .await
3265 .expect("Failed to append after recovery");
3266
3267 assert_eq!(position, (i - 10 + 2) as u64);
3269 new_locations.push((position, offset, size, i));
3270
3271 let retrieved = oversized.get(1, position).await.expect("Failed to get");
3273 assert_eq!(retrieved.id, i as u64);
3274
3275 let retrieved_value = oversized
3276 .get_value(1, offset, size)
3277 .await
3278 .expect("Failed to get value");
3279 assert_eq!(retrieved_value, value);
3280 }
3281
3282 oversized = oversized.sync(1).await.expect("Failed to sync");
3284 drop(oversized);
3285
3286 let oversized: Oversized<_, TestEntry, TestValue> =
3288 Oversized::init(context.child("third"), cfg)
3289 .await
3290 .expect("Failed to reinit after append");
3291
3292 for i in 0..2u8 {
3295 let (position, offset, size) = locations[i as usize];
3296 let entry = oversized.get(1, position).await.expect("Failed to get");
3297 assert_eq!(entry.id, i as u64);
3298
3299 let value = oversized
3300 .get_value(1, offset, size)
3301 .await
3302 .expect("Failed to get value");
3303 assert_eq!(value, [i; 16]);
3304 }
3305
3306 for (position, offset, size, expected_id) in &new_locations {
3308 let entry = oversized
3309 .get(1, *position)
3310 .await
3311 .expect("Failed to get new entry after restart");
3312 assert_eq!(entry.id, *expected_id as u64);
3313
3314 let value = oversized
3315 .get_value(1, *offset, *size)
3316 .await
3317 .expect("Failed to get new value after restart");
3318 assert_eq!(value, [*expected_id; 16]);
3319 }
3320
3321 assert!(oversized.get(1, 4).await.is_ok());
3323 assert!(oversized.get(1, 5).await.is_err());
3324
3325 oversized.destroy().await.expect("Failed to destroy");
3326 });
3327 }
3328
3329 #[test_traced]
3330 fn test_recovery_partial_index_entry() {
3331 let executor = deterministic::Runner::default();
3332 executor.start(|context| async move {
3333 let cfg = test_cfg(&context);
3334
3335 let mut oversized: Oversized<_, TestEntry, TestValue> =
3337 Oversized::init(context.child("first"), cfg.clone())
3338 .await
3339 .expect("Failed to init");
3340
3341 for i in 0..3u8 {
3343 let value: TestValue = [i; 16];
3344 let entry = TestEntry::new(i as u64, 0, 0);
3345 (oversized, _, _, _) = oversized
3346 .append(1, entry, &value)
3347 .await
3348 .expect("Failed to append");
3349 }
3350 oversized = oversized.sync(1).await.expect("Failed to sync");
3351 drop(oversized);
3352
3353 let (blob, _) = context
3357 .open(&cfg.index_partition, &1u64.to_be_bytes())
3358 .await
3359 .expect("Failed to open blob");
3360 let partial_size = 3 * 24 + 10; blob.resize(partial_size).await.expect("Failed to resize");
3362 blob.sync().await.expect("Failed to sync");
3363 drop(blob);
3364
3365 let mut oversized: Oversized<_, TestEntry, TestValue> =
3367 Oversized::init(context.child("second"), cfg.clone())
3368 .await
3369 .expect("Failed to reinit");
3370
3371 for i in 0..3u8 {
3373 let entry = oversized.get(1, i as u64).await.expect("Failed to get");
3374 assert_eq!(entry.id, i as u64);
3375 }
3376
3377 assert!(oversized.get(1, 3).await.is_err());
3379
3380 let value: TestValue = [42; 16];
3382 let entry = TestEntry::new(100, 0, 0);
3383 let (pos, offset, size);
3384 (oversized, pos, offset, size) = oversized
3385 .append(1, entry, &value)
3386 .await
3387 .expect("Failed to append after recovery");
3388 assert_eq!(pos, 3);
3389
3390 let retrieved = oversized.get(1, 3).await.expect("Failed to get new entry");
3392 assert_eq!(retrieved.id, 100);
3393 let retrieved_value = oversized
3394 .get_value(1, offset, size)
3395 .await
3396 .expect("Failed to get new value");
3397 assert_eq!(retrieved_value, value);
3398
3399 oversized.destroy().await.expect("Failed to destroy");
3400 });
3401 }
3402
3403 #[test_traced]
3404 fn test_recovery_only_partial_entry() {
3405 let executor = deterministic::Runner::default();
3406 executor.start(|context| async move {
3407 let cfg = test_cfg(&context);
3408
3409 let mut oversized: Oversized<_, TestEntry, TestValue> =
3411 Oversized::init(context.child("first"), cfg.clone())
3412 .await
3413 .expect("Failed to init");
3414
3415 let value: TestValue = [42; 16];
3416 let entry = TestEntry::new(1, 0, 0);
3417 (oversized, _, _, _) = oversized
3418 .append(1, entry, &value)
3419 .await
3420 .expect("Failed to append");
3421 oversized = oversized.sync(1).await.expect("Failed to sync");
3422 drop(oversized);
3423
3424 let (blob, _) = context
3426 .open(&cfg.index_partition, &1u64.to_be_bytes())
3427 .await
3428 .expect("Failed to open blob");
3429 blob.resize(10).await.expect("Failed to resize"); blob.sync().await.expect("Failed to sync");
3431 drop(blob);
3432
3433 let mut oversized: Oversized<_, TestEntry, TestValue> =
3435 Oversized::init(context.child("second"), cfg.clone())
3436 .await
3437 .expect("Failed to reinit");
3438
3439 assert!(oversized.get(1, 0).await.is_err());
3441
3442 let value: TestValue = [99; 16];
3444 let entry = TestEntry::new(100, 0, 0);
3445 let (pos, offset, size);
3446 (oversized, pos, offset, size) = oversized
3447 .append(1, entry, &value)
3448 .await
3449 .expect("Failed to append after recovery");
3450 assert_eq!(pos, 0);
3451
3452 let retrieved = oversized.get(1, 0).await.expect("Failed to get");
3453 assert_eq!(retrieved.id, 100);
3454 let retrieved_value = oversized
3455 .get_value(1, offset, size)
3456 .await
3457 .expect("Failed to get value");
3458 assert_eq!(retrieved_value, value);
3459
3460 oversized.destroy().await.expect("Failed to destroy");
3461 });
3462 }
3463
3464 #[test_traced]
3465 fn test_recovery_crash_during_rewind_index_ahead() {
3466 let executor = deterministic::Runner::default();
3468 executor.start(|context| async move {
3469 let cfg = entry_cfg(&context);
3473
3474 let mut oversized: Oversized<_, TestEntry, TestValue> =
3476 Oversized::init(context.child("first"), cfg.clone())
3477 .await
3478 .expect("Failed to init");
3479
3480 let mut locations = Vec::new();
3481 for i in 0..5u8 {
3482 let value: TestValue = [i; 16];
3483 let entry = TestEntry::new(i as u64, 0, 0);
3484 let (position, offset, size);
3485 (oversized, position, offset, size) = oversized
3486 .append(1, entry, &value)
3487 .await
3488 .expect("Failed to append");
3489 locations.push((position, offset, size));
3490 }
3491 oversized = oversized.sync(1).await.expect("Failed to sync");
3492 drop(oversized);
3493
3494 let (blob, _) = context
3497 .open(&cfg.index_partition, &1u64.to_be_bytes())
3498 .await
3499 .expect("Failed to open blob");
3500 let physical_page_size = (TestEntry::SIZE + 12) as u64;
3502 blob.resize(2 * physical_page_size)
3503 .await
3504 .expect("Failed to truncate");
3505 blob.sync().await.expect("Failed to sync");
3506 drop(blob);
3507
3508 let mut oversized: Oversized<_, TestEntry, TestValue> =
3510 Oversized::init(context.child("second"), cfg.clone())
3511 .await
3512 .expect("Failed to reinit");
3513
3514 for i in 0..2u8 {
3516 let entry = oversized.get(1, i as u64).await.expect("Failed to get");
3517 assert_eq!(entry.id, i as u64);
3518 }
3519
3520 assert!(oversized.get(1, 2).await.is_err());
3522
3523 let pos;
3525 (oversized, pos, _, _) = oversized
3526 .append(1, TestEntry::new(100, 0, 0), &[100u8; 16])
3527 .await
3528 .expect("Failed to append");
3529 assert_eq!(pos, 2);
3530
3531 oversized.destroy().await.expect("Failed to destroy");
3532 });
3533 }
3534
3535 #[test_traced]
3536 fn test_recovery_crash_during_rewind_glob_ahead() {
3537 let executor = deterministic::Runner::default();
3539 executor.start(|context| async move {
3540 let cfg = test_cfg(&context);
3541
3542 let mut oversized: Oversized<_, TestEntry, TestValue> =
3544 Oversized::init(context.child("first"), cfg.clone())
3545 .await
3546 .expect("Failed to init");
3547
3548 let mut locations = Vec::new();
3549 for i in 0..5u8 {
3550 let value: TestValue = [i; 16];
3551 let entry = TestEntry::new(i as u64, 0, 0);
3552 let (position, offset, size);
3553 (oversized, position, offset, size) = oversized
3554 .append(1, entry, &value)
3555 .await
3556 .expect("Failed to append");
3557 locations.push((position, offset, size));
3558 }
3559 oversized = oversized.sync(1).await.expect("Failed to sync");
3560 drop(oversized);
3561
3562 let (blob, _) = context
3565 .open(&cfg.value_partition, &1u64.to_be_bytes())
3566 .await
3567 .expect("Failed to open blob");
3568 let keep_size = byte_end(locations[1].1, locations[1].2);
3569 blob.resize(keep_size).await.expect("Failed to truncate");
3570 blob.sync().await.expect("Failed to sync");
3571 drop(blob);
3572
3573 let mut oversized: Oversized<_, TestEntry, TestValue> =
3575 Oversized::init(context.child("second"), cfg.clone())
3576 .await
3577 .expect("Failed to reinit");
3578
3579 for i in 0..2u8 {
3581 let entry = oversized.get(1, i as u64).await.expect("Failed to get");
3582 assert_eq!(entry.id, i as u64);
3583 }
3584
3585 assert!(oversized.get(1, 2).await.is_err());
3587
3588 let value: TestValue = [99; 16];
3590 let entry = TestEntry::new(100, 0, 0);
3591 let (pos, offset, size);
3592 (oversized, pos, offset, size) = oversized
3593 .append(1, entry, &value)
3594 .await
3595 .expect("Failed to append after recovery");
3596 assert_eq!(pos, 2);
3597
3598 let retrieved = oversized.get(1, 2).await.expect("Failed to get");
3599 assert_eq!(retrieved.id, 100);
3600 let retrieved_value = oversized
3601 .get_value(1, offset, size)
3602 .await
3603 .expect("Failed to get value");
3604 assert_eq!(retrieved_value, value);
3605
3606 oversized.destroy().await.expect("Failed to destroy");
3607 });
3608 }
3609
3610 #[test_traced]
3611 fn test_oversized_get_value_invalid_size() {
3612 let executor = deterministic::Runner::default();
3613 executor.start(|context| async move {
3614 let cfg = test_cfg(&context);
3615 let mut oversized: Oversized<_, TestEntry, TestValue> =
3616 Oversized::init(context, cfg).await.expect("Failed to init");
3617
3618 let value: TestValue = [42; 16];
3619 let entry = TestEntry::new(1, 0, 0);
3620 let (offset, _size);
3621 (oversized, _, offset, _size) = oversized
3622 .append(1, entry, &value)
3623 .await
3624 .expect("Failed to append");
3625 oversized = oversized.sync(1).await.expect("Failed to sync");
3626
3627 assert!(oversized.get_value(1, offset, 0).await.is_err());
3629
3630 for size in 1..4u32 {
3633 let result = oversized.get_value(1, offset, size).await;
3634 assert!(
3635 matches!(
3636 result,
3637 Err(Error::Codec(_))
3638 | Err(Error::ChecksumMismatch(_, _))
3639 | Err(Error::Runtime(_))
3640 ),
3641 "expected error, got: {:?}",
3642 result
3643 );
3644 }
3645
3646 oversized.destroy().await.expect("Failed to destroy");
3647 });
3648 }
3649
3650 #[test_traced]
3651 fn test_oversized_get_value_wrong_size() {
3652 let executor = deterministic::Runner::default();
3653 executor.start(|context| async move {
3654 let cfg = test_cfg(&context);
3655 let mut oversized: Oversized<_, TestEntry, TestValue> =
3656 Oversized::init(context, cfg).await.expect("Failed to init");
3657
3658 let value: TestValue = [42; 16];
3659 let entry = TestEntry::new(1, 0, 0);
3660 let (offset, correct_size);
3661 (oversized, _, offset, correct_size) = oversized
3662 .append(1, entry, &value)
3663 .await
3664 .expect("Failed to append");
3665 oversized = oversized.sync(1).await.expect("Failed to sync");
3666
3667 let result = oversized.get_value(1, offset, correct_size - 1).await;
3670 assert!(
3671 matches!(
3672 result,
3673 Err(Error::Codec(_)) | Err(Error::ChecksumMismatch(_, _))
3674 ),
3675 "expected Codec or ChecksumMismatch error, got: {:?}",
3676 result
3677 );
3678
3679 oversized.destroy().await.expect("Failed to destroy");
3680 });
3681 }
3682
3683 #[test_traced]
3684 fn test_recovery_values_has_orphan_section() {
3685 let executor = deterministic::Runner::default();
3686 executor.start(|context| async move {
3687 let cfg = test_cfg(&context);
3688
3689 let mut oversized: Oversized<_, TestEntry, TestValue> =
3691 Oversized::init(context.child("first"), cfg.clone())
3692 .await
3693 .expect("Failed to init");
3694
3695 for section in 1u64..=2 {
3696 let value: TestValue = [section as u8; 16];
3697 let entry = TestEntry::new(section, 0, 0);
3698 (oversized, _, _, _) = oversized
3699 .append(section, entry, &value)
3700 .await
3701 .expect("Failed to append");
3702 oversized = oversized.sync(section).await.expect("Failed to sync");
3703 }
3704 drop(oversized);
3705
3706 let glob_cfg = GlobConfig {
3708 partition: cfg.value_partition.clone(),
3709 compression: cfg.compression,
3710 codec_config: (),
3711 write_buffer: cfg.value_write_buffer,
3712 };
3713 let mut glob: Glob<_, TestValue> = Glob::init(context.child("glob"), glob_cfg)
3714 .await
3715 .expect("Failed to init glob");
3716 let orphan_value: TestValue = [99; 16];
3717 (glob, _, _) = glob
3718 .append(3, &orphan_value)
3719 .await
3720 .expect("Failed to append orphan");
3721 glob = glob.sync(3).await.expect("Failed to sync glob");
3722 drop(glob);
3723
3724 let oversized: Oversized<_, TestEntry, TestValue> =
3726 Oversized::init(context.child("second"), cfg.clone())
3727 .await
3728 .expect("Failed to reinit");
3729
3730 assert!(oversized.get(1, 0).await.is_ok());
3732 assert!(oversized.get(2, 0).await.is_ok());
3733
3734 assert_eq!(oversized.newest_section(), Some(2));
3736
3737 oversized.destroy().await.expect("Failed to destroy");
3738 });
3739 }
3740
3741 #[test_traced]
3742 fn test_recovery_values_has_multiple_orphan_sections() {
3743 let executor = deterministic::Runner::default();
3744 executor.start(|context| async move {
3745 let cfg = test_cfg(&context);
3746
3747 let mut oversized: Oversized<_, TestEntry, TestValue> =
3749 Oversized::init(context.child("first"), cfg.clone())
3750 .await
3751 .expect("Failed to init");
3752
3753 let value: TestValue = [1; 16];
3754 let entry = TestEntry::new(1, 0, 0);
3755 (oversized, _, _, _) = oversized
3756 .append(1, entry, &value)
3757 .await
3758 .expect("Failed to append");
3759 oversized = oversized.sync(1).await.expect("Failed to sync");
3760 drop(oversized);
3761
3762 let glob_cfg = GlobConfig {
3764 partition: cfg.value_partition.clone(),
3765 compression: cfg.compression,
3766 codec_config: (),
3767 write_buffer: cfg.value_write_buffer,
3768 };
3769 let mut glob: Glob<_, TestValue> = Glob::init(context.child("glob"), glob_cfg)
3770 .await
3771 .expect("Failed to init glob");
3772
3773 for section in 2u64..=4 {
3774 let orphan_value: TestValue = [section as u8; 16];
3775 (glob, _, _) = glob
3776 .append(section, &orphan_value)
3777 .await
3778 .expect("Failed to append orphan");
3779 glob = glob.sync(section).await.expect("Failed to sync glob");
3780 }
3781 drop(glob);
3782
3783 let oversized: Oversized<_, TestEntry, TestValue> =
3785 Oversized::init(context.child("second"), cfg.clone())
3786 .await
3787 .expect("Failed to reinit");
3788
3789 assert!(oversized.get(1, 0).await.is_ok());
3791
3792 assert_eq!(oversized.newest_section(), Some(1));
3794
3795 oversized.destroy().await.expect("Failed to destroy");
3796 });
3797 }
3798
3799 #[test_traced]
3800 fn test_recovery_index_empty_but_values_exist() {
3801 let executor = deterministic::Runner::default();
3802 executor.start(|context| async move {
3803 let cfg = test_cfg(&context);
3804
3805 let glob_cfg = GlobConfig {
3807 partition: cfg.value_partition.clone(),
3808 compression: cfg.compression,
3809 codec_config: (),
3810 write_buffer: cfg.value_write_buffer,
3811 };
3812 let mut glob: Glob<_, TestValue> = Glob::init(context.child("glob"), glob_cfg)
3813 .await
3814 .expect("Failed to init glob");
3815
3816 for section in 1u64..=3 {
3817 let orphan_value: TestValue = [section as u8; 16];
3818 (glob, _, _) = glob
3819 .append(section, &orphan_value)
3820 .await
3821 .expect("Failed to append orphan");
3822 glob = glob.sync(section).await.expect("Failed to sync glob");
3823 }
3824 drop(glob);
3825
3826 let oversized: Oversized<_, TestEntry, TestValue> =
3828 Oversized::init(context.child("first"), cfg.clone())
3829 .await
3830 .expect("Failed to init");
3831
3832 assert_eq!(oversized.newest_section(), None);
3834 assert_eq!(oversized.oldest_section(), None);
3835
3836 oversized.destroy().await.expect("Failed to destroy");
3837 });
3838 }
3839
3840 #[test_traced]
3841 fn test_recovery_orphan_section_append_after() {
3842 let executor = deterministic::Runner::default();
3843 executor.start(|context| async move {
3844 let cfg = test_cfg(&context);
3845
3846 let mut oversized: Oversized<_, TestEntry, TestValue> =
3848 Oversized::init(context.child("first"), cfg.clone())
3849 .await
3850 .expect("Failed to init");
3851
3852 let value: TestValue = [1; 16];
3853 let entry = TestEntry::new(1, 0, 0);
3854 let (offset1, size1);
3855 (oversized, _, offset1, size1) = oversized
3856 .append(1, entry, &value)
3857 .await
3858 .expect("Failed to append");
3859 oversized = oversized.sync(1).await.expect("Failed to sync");
3860 drop(oversized);
3861
3862 let glob_cfg = GlobConfig {
3864 partition: cfg.value_partition.clone(),
3865 compression: cfg.compression,
3866 codec_config: (),
3867 write_buffer: cfg.value_write_buffer,
3868 };
3869 let mut glob: Glob<_, TestValue> = Glob::init(context.child("glob"), glob_cfg)
3870 .await
3871 .expect("Failed to init glob");
3872
3873 for section in 2u64..=3 {
3874 let orphan_value: TestValue = [section as u8; 16];
3875 (glob, _, _) = glob
3876 .append(section, &orphan_value)
3877 .await
3878 .expect("Failed to append orphan");
3879 glob = glob.sync(section).await.expect("Failed to sync glob");
3880 }
3881 drop(glob);
3882
3883 let mut oversized: Oversized<_, TestEntry, TestValue> =
3885 Oversized::init(context.child("second"), cfg.clone())
3886 .await
3887 .expect("Failed to reinit");
3888
3889 let entry = oversized.get(1, 0).await.expect("Failed to get");
3891 assert_eq!(entry.id, 1);
3892 let value = oversized
3893 .get_value(1, offset1, size1)
3894 .await
3895 .expect("Failed to get value");
3896 assert_eq!(value, [1; 16]);
3897
3898 let new_value: TestValue = [42; 16];
3900 let new_entry = TestEntry::new(42, 0, 0);
3901 let (pos, offset, size);
3902 (oversized, pos, offset, size) = oversized
3903 .append(2, new_entry, &new_value)
3904 .await
3905 .expect("Failed to append after recovery");
3906 assert_eq!(pos, 0);
3907
3908 let retrieved = oversized.get(2, 0).await.expect("Failed to get");
3910 assert_eq!(retrieved.id, 42);
3911 let retrieved_value = oversized
3912 .get_value(2, offset, size)
3913 .await
3914 .expect("Failed to get value");
3915 assert_eq!(retrieved_value, new_value);
3916
3917 oversized = oversized.sync(2).await.expect("Failed to sync");
3919 drop(oversized);
3920
3921 let oversized: Oversized<_, TestEntry, TestValue> =
3922 Oversized::init(context.child("third"), cfg)
3923 .await
3924 .expect("Failed to reinit after append");
3925
3926 assert!(oversized.get(1, 0).await.is_ok());
3928 assert!(oversized.get(2, 0).await.is_ok());
3929 assert_eq!(oversized.newest_section(), Some(2));
3930
3931 oversized.destroy().await.expect("Failed to destroy");
3932 });
3933 }
3934
3935 #[test_traced]
3936 fn test_recovery_no_orphan_sections() {
3937 let executor = deterministic::Runner::default();
3938 executor.start(|context| async move {
3939 let cfg = test_cfg(&context);
3940
3941 let mut oversized: Oversized<_, TestEntry, TestValue> =
3943 Oversized::init(context.child("first"), cfg.clone())
3944 .await
3945 .expect("Failed to init");
3946
3947 for section in 1u64..=3 {
3948 let value: TestValue = [section as u8; 16];
3949 let entry = TestEntry::new(section, 0, 0);
3950 (oversized, _, _, _) = oversized
3951 .append(section, entry, &value)
3952 .await
3953 .expect("Failed to append");
3954 oversized = oversized.sync(section).await.expect("Failed to sync");
3955 }
3956 drop(oversized);
3957
3958 let oversized: Oversized<_, TestEntry, TestValue> =
3960 Oversized::init(context.child("second"), cfg)
3961 .await
3962 .expect("Failed to reinit");
3963
3964 for section in 1u64..=3 {
3966 let entry = oversized.get(section, 0).await.expect("Failed to get");
3967 assert_eq!(entry.id, section);
3968 }
3969 assert_eq!(oversized.newest_section(), Some(3));
3970
3971 oversized.destroy().await.expect("Failed to destroy");
3972 });
3973 }
3974
3975 #[test_traced]
3976 fn test_recovery_orphan_with_empty_index_section() {
3977 let executor = deterministic::Runner::default();
3978 executor.start(|context| async move {
3979 let cfg = test_cfg(&context);
3980
3981 let mut oversized: Oversized<_, TestEntry, TestValue> =
3983 Oversized::init(context.child("first"), cfg.clone())
3984 .await
3985 .expect("Failed to init");
3986
3987 let value: TestValue = [1; 16];
3988 let entry = TestEntry::new(1, 0, 0);
3989 (oversized, _, _, _) = oversized
3990 .append(1, entry, &value)
3991 .await
3992 .expect("Failed to append");
3993 oversized = oversized.sync(1).await.expect("Failed to sync");
3994 drop(oversized);
3995
3996 let glob_cfg = GlobConfig {
3998 partition: cfg.value_partition.clone(),
3999 compression: cfg.compression,
4000 codec_config: (),
4001 write_buffer: cfg.value_write_buffer,
4002 };
4003 let mut glob: Glob<_, TestValue> = Glob::init(context.child("glob"), glob_cfg)
4004 .await
4005 .expect("Failed to init glob");
4006 let orphan_value: TestValue = [2; 16];
4007 (glob, _, _) = glob
4008 .append(2, &orphan_value)
4009 .await
4010 .expect("Failed to append orphan");
4011 glob = glob.sync(2).await.expect("Failed to sync glob");
4012 drop(glob);
4013
4014 let (blob, _) = context
4016 .open(&cfg.index_partition, &1u64.to_be_bytes())
4017 .await
4018 .expect("Failed to open blob");
4019 blob.resize(0).await.expect("Failed to truncate");
4020 blob.sync().await.expect("Failed to sync");
4021 drop(blob);
4022
4023 let oversized: Oversized<_, TestEntry, TestValue> =
4025 Oversized::init(context.child("second"), cfg)
4026 .await
4027 .expect("Failed to reinit");
4028
4029 assert!(oversized.get(1, 0).await.is_err());
4031
4032 assert_eq!(oversized.newest_section(), Some(1));
4034
4035 oversized.destroy().await.expect("Failed to destroy");
4036 });
4037 }
4038
4039 #[test_traced]
4040 fn test_recovery_orphan_sections_with_gaps() {
4041 let executor = deterministic::Runner::default();
4044 executor.start(|context| async move {
4045 let cfg = test_cfg(&context);
4046
4047 let mut oversized: Oversized<_, TestEntry, TestValue> =
4049 Oversized::init(context.child("first"), cfg.clone())
4050 .await
4051 .expect("Failed to init");
4052
4053 for section in [1u64, 3, 5] {
4054 let value: TestValue = [section as u8; 16];
4055 let entry = TestEntry::new(section, 0, 0);
4056 (oversized, _, _, _) = oversized
4057 .append(section, entry, &value)
4058 .await
4059 .expect("Failed to append");
4060 oversized = oversized.sync(section).await.expect("Failed to sync");
4061 }
4062 drop(oversized);
4063
4064 let glob_cfg = GlobConfig {
4066 partition: cfg.value_partition.clone(),
4067 compression: cfg.compression,
4068 codec_config: (),
4069 write_buffer: cfg.value_write_buffer,
4070 };
4071 let mut glob: Glob<_, TestValue> = Glob::init(context.child("glob"), glob_cfg)
4072 .await
4073 .expect("Failed to init glob");
4074
4075 for section in [2u64, 4, 6] {
4076 let orphan_value: TestValue = [section as u8; 16];
4077 (glob, _, _) = glob
4078 .append(section, &orphan_value)
4079 .await
4080 .expect("Failed to append orphan");
4081 glob = glob.sync(section).await.expect("Failed to sync glob");
4082 }
4083 drop(glob);
4084
4085 let oversized: Oversized<_, TestEntry, TestValue> =
4087 Oversized::init(context.child("second"), cfg)
4088 .await
4089 .expect("Failed to reinit");
4090
4091 for section in [1u64, 3, 5] {
4093 let entry = oversized.get(section, 0).await.expect("Failed to get");
4094 assert_eq!(entry.id, section);
4095 }
4096
4097 assert_eq!(oversized.oldest_section(), Some(1));
4099 assert_eq!(oversized.newest_section(), Some(5));
4100
4101 oversized.destroy().await.expect("Failed to destroy");
4102 });
4103 }
4104
4105 #[test_traced]
4108 fn test_recovery_discards_index_extension_after_prefix_loss() {
4109 let executor = deterministic::Runner::default();
4110 executor.start(|context| async move {
4111 let cfg = test_cfg(&context);
4113 let mut oversized: Oversized<_, TestEntry, TestValue> =
4114 Oversized::init(context.child("first"), cfg.clone())
4115 .await
4116 .expect("Failed to init");
4117 for section in 1..=2 {
4118 (oversized, _, _, _) = oversized
4119 .append(section, TestEntry::new(section, 0, 0), &[section as u8; 16])
4120 .await
4121 .expect("Failed to append");
4122 oversized = oversized.sync(section).await.expect("Failed to sync");
4123 }
4124 drop(oversized);
4125
4126 for section in 1..=2u64 {
4129 let (blob, original_size) = context
4130 .open(&cfg.index_partition, §ion.to_be_bytes())
4131 .await
4132 .expect("Failed to open index blob");
4133 if section == 1 {
4134 blob.resize(0).await.expect("Failed to truncate index");
4135 blob.sync().await.expect("Failed to sync index truncation");
4136 } else {
4137 drop(blob);
4138 context
4139 .remove(&cfg.index_partition, Some(§ion.to_be_bytes()))
4140 .await
4141 .expect("Failed to remove index");
4142 }
4143 let (blob, _) = context
4144 .open(&cfg.index_partition, §ion.to_be_bytes())
4145 .await
4146 .expect("Failed to recreate index blob");
4147 blob.write_at(
4148 0,
4149 vec![0; usize::try_from(original_size * 2).unwrap()],
4150 WriteOptions::SYNC,
4151 )
4152 .await
4153 .expect("Failed to extend index");
4154 }
4155
4156 let mut oversized: Oversized<_, TestEntry, TestValue> =
4160 Oversized::init(context.child("second"), cfg.clone())
4161 .await
4162 .expect("Failed to recover extended index");
4163 for section in 1..=2u64 {
4164 assert!(matches!(
4165 oversized.get(section, 0).await,
4166 Err(Error::ItemOutOfRange(0))
4167 ));
4168 let (_, recovered_size) = context
4169 .open(&cfg.index_partition, §ion.to_be_bytes())
4170 .await
4171 .expect("Failed to reopen index blob");
4172 assert_eq!(recovered_size, 0);
4173
4174 let position;
4176 (oversized, position, _, _) = oversized
4177 .append(section, TestEntry::new(section, 0, 0), &[section as u8; 16])
4178 .await
4179 .expect("Failed to append after recovery");
4180 assert_eq!(position, 0);
4181 }
4182 oversized = oversized.sync_all().await.expect("Failed to sync sentinel");
4183 drop(oversized);
4184
4185 let oversized: Oversized<_, TestEntry, TestValue> =
4187 Oversized::init(context.child("third"), cfg)
4188 .await
4189 .expect("Failed to reopen sentinel");
4190 for section in 1..=2u64 {
4191 let entry = oversized
4192 .get(section, 0)
4193 .await
4194 .expect("Sentinel index missing");
4195 let (offset, size) = entry.value_location();
4196 assert_eq!(entry.id, section);
4197 assert_eq!(
4198 oversized
4199 .get_value(section, offset, size)
4200 .await
4201 .expect("Sentinel value missing"),
4202 [section as u8; 16]
4203 );
4204 }
4205 oversized.destroy().await.expect("Failed to destroy");
4206 });
4207 }
4208
4209 #[test_traced]
4210 fn test_recovery_glob_trailing_garbage_truncated() {
4211 let executor = deterministic::Runner::default();
4215 executor.start(|context| async move {
4216 let cfg = test_cfg(&context);
4217
4218 let mut oversized: Oversized<_, TestEntry, TestValue> =
4220 Oversized::init(context.child("first"), cfg.clone())
4221 .await
4222 .expect("Failed to init");
4223
4224 let mut locations = Vec::new();
4226 for i in 0..2u8 {
4227 let value: TestValue = [i; 16];
4228 let entry = TestEntry::new(i as u64, 0, 0);
4229 let (position, offset, size);
4230 (oversized, position, offset, size) = oversized
4231 .append(1, entry, &value)
4232 .await
4233 .expect("Failed to append");
4234 locations.push((position, offset, size));
4235 }
4236 oversized = oversized.sync(1).await.expect("Failed to sync");
4237
4238 let expected_next_offset = byte_end(locations[1].1, locations[1].2);
4240 drop(oversized);
4241
4242 let (blob, size) = context
4244 .open(&cfg.value_partition, &1u64.to_be_bytes())
4245 .await
4246 .expect("Failed to open blob");
4247 assert_eq!(size, expected_next_offset);
4248
4249 let garbage = vec![0xDE; 100];
4251 blob.write_at(size, garbage, WriteOptions::SYNC)
4252 .await
4253 .expect("Failed to write garbage");
4254 drop(blob);
4255
4256 let (blob, new_size) = context
4258 .open(&cfg.value_partition, &1u64.to_be_bytes())
4259 .await
4260 .expect("Failed to open blob");
4261 assert_eq!(new_size, expected_next_offset + 100);
4262 drop(blob);
4263
4264 let mut oversized: Oversized<_, TestEntry, TestValue> =
4266 Oversized::init(context.child("second"), cfg.clone())
4267 .await
4268 .expect("Failed to reinit");
4269
4270 for i in 0..2u8 {
4272 let entry = oversized.get(1, i as u64).await.expect("Failed to get");
4273 assert_eq!(entry.id, i as u64);
4274 }
4275
4276 let new_value: TestValue = [99; 16];
4278 let new_entry = TestEntry::new(99, 0, 0);
4279 let (pos, offset, _size);
4280 (oversized, pos, offset, _size) = oversized
4281 .append(1, new_entry, &new_value)
4282 .await
4283 .expect("Failed to append after recovery");
4284
4285 assert_eq!(pos, 2);
4287
4288 assert_eq!(offset, expected_next_offset);
4290
4291 let retrieved = oversized.get(1, 2).await.expect("Failed to get new entry");
4293 assert_eq!(retrieved.id, 99);
4294
4295 oversized.destroy().await.expect("Failed to destroy");
4296 });
4297 }
4298
4299 #[test_traced]
4300 fn test_recovery_entry_with_overflow_offset() {
4301 let executor = deterministic::Runner::default();
4304 executor.start(|context| async move {
4305 let cfg = entry_cfg(&context);
4307
4308 let mut oversized: Oversized<_, TestEntry, TestValue> =
4310 Oversized::init(context.child("first"), cfg.clone())
4311 .await
4312 .expect("Failed to init");
4313
4314 let value: TestValue = [1; 16];
4315 let entry = TestEntry::new(1, 0, 0);
4316 (oversized, _, _, _) = oversized
4317 .append(1, entry, &value)
4318 .await
4319 .expect("Failed to append");
4320 oversized = oversized.sync(1).await.expect("Failed to sync");
4321 drop(oversized);
4322
4323 let (blob, _) = context
4327 .open(&cfg.index_partition, &1u64.to_be_bytes())
4328 .await
4329 .expect("Failed to open blob");
4330
4331 let mut entry_data = Vec::new();
4333 1u64.write(&mut entry_data); (u64::MAX - 10).write(&mut entry_data); 100u32.write(&mut entry_data); assert_eq!(entry_data.len(), TestEntry::SIZE);
4337
4338 let crc = Crc32::checksum(&entry_data);
4341 let len1 = TestEntry::SIZE as u16;
4342 let mut crc_record = Vec::new();
4343 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);
4348
4349 let mut page = entry_data;
4351 page.extend_from_slice(&crc_record);
4352 blob.write_at(0, page, WriteOptions::SYNC)
4353 .await
4354 .expect("Failed to write corrupted page");
4355 drop(blob);
4356
4357 let mut oversized: Oversized<_, TestEntry, TestValue> =
4360 Oversized::init(context.child("second"), cfg.clone())
4361 .await
4362 .expect("Failed to reinit");
4363
4364 assert!(oversized.get(1, 0).await.is_err());
4366
4367 let new_value: TestValue = [99; 16];
4369 let new_entry = TestEntry::new(99, 0, 0);
4370 let (pos, new_offset);
4371 (oversized, pos, new_offset, _) = oversized
4372 .append(1, new_entry, &new_value)
4373 .await
4374 .expect("Failed to append after recovery");
4375
4376 assert_eq!(pos, 0);
4378 assert_eq!(new_offset, 0);
4380
4381 oversized.destroy().await.expect("Failed to destroy");
4382 });
4383 }
4384
4385 #[test_traced]
4386 fn test_empty_section_persistence() {
4387 let executor = deterministic::Runner::default();
4390 executor.start(|context| async move {
4391 let cfg = test_cfg(&context);
4392
4393 let mut oversized: Oversized<_, TestEntry, TestValue> =
4395 Oversized::init(context.child("first"), cfg.clone())
4396 .await
4397 .expect("Failed to init");
4398
4399 for i in 0..3u8 {
4400 let value: TestValue = [i; 16];
4401 let entry = TestEntry::new(i as u64, 0, 0);
4402 (oversized, _, _, _) = oversized
4403 .append(1, entry, &value)
4404 .await
4405 .expect("Failed to append");
4406 }
4407 oversized = oversized.sync(1).await.expect("Failed to sync");
4408
4409 let value2: TestValue = [10; 16];
4411 let entry2 = TestEntry::new(10, 0, 0);
4412 (oversized, _, _, _) = oversized
4413 .append(2, entry2, &value2)
4414 .await
4415 .expect("Failed to append to section 2");
4416 oversized = oversized.sync(2).await.expect("Failed to sync section 2");
4417 drop(oversized);
4418
4419 let (blob, _) = context
4421 .open(&cfg.index_partition, &1u64.to_be_bytes())
4422 .await
4423 .expect("Failed to open blob");
4424 blob.resize(0).await.expect("Failed to truncate");
4425 blob.sync().await.expect("Failed to sync");
4426 drop(blob);
4427
4428 let mut oversized: Oversized<_, TestEntry, TestValue> =
4430 Oversized::init(context.child("second"), cfg.clone())
4431 .await
4432 .expect("Failed to reinit");
4433
4434 assert!(oversized.get(1, 0).await.is_err());
4436
4437 let entry = oversized.get(2, 0).await.expect("Failed to get section 2");
4439 assert_eq!(entry.id, 10);
4440
4441 assert_eq!(oversized.oldest_section(), Some(1));
4443
4444 let new_value: TestValue = [99; 16];
4447 let new_entry = TestEntry::new(99, 0, 0);
4448 let (pos, offset, size);
4449 (oversized, pos, offset, size) = oversized
4450 .append(1, new_entry, &new_value)
4451 .await
4452 .expect("Failed to append to empty section");
4453 assert_eq!(pos, 0);
4454 assert_eq!(offset, 0);
4455 oversized = oversized.sync(1).await.expect("Failed to sync");
4456
4457 let entry = oversized.get(1, 0).await.expect("Failed to get");
4459 assert_eq!(entry.id, 99);
4460 let value = oversized
4461 .get_value(1, offset, size)
4462 .await
4463 .expect("Failed to get value");
4464 assert_eq!(value, new_value);
4465
4466 drop(oversized);
4467
4468 let oversized: Oversized<_, TestEntry, TestValue> =
4470 Oversized::init(context.child("third"), cfg.clone())
4471 .await
4472 .expect("Failed to reinit again");
4473
4474 let entry = oversized.get(1, 0).await.expect("Failed to get");
4476 assert_eq!(entry.id, 99);
4477
4478 let entry = oversized.get(2, 0).await.expect("Failed to get section 2");
4480 assert_eq!(entry.id, 10);
4481
4482 oversized.destroy().await.expect("Failed to destroy");
4483 });
4484 }
4485
4486 #[test_traced]
4487 fn test_get_value_size_equals_crc_size() {
4488 let executor = deterministic::Runner::default();
4491 executor.start(|context| async move {
4492 let cfg = test_cfg(&context);
4493 let mut oversized: Oversized<_, TestEntry, TestValue> =
4494 Oversized::init(context, cfg).await.expect("Failed to init");
4495
4496 let value: TestValue = [42; 16];
4497 let entry = TestEntry::new(1, 0, 0);
4498 let offset;
4499 (oversized, _, offset, _) = oversized
4500 .append(1, entry, &value)
4501 .await
4502 .expect("Failed to append");
4503 oversized = oversized.sync(1).await.expect("Failed to sync");
4504
4505 let result = oversized.get_value(1, offset, 4).await;
4508 assert!(result.is_err());
4509
4510 oversized.destroy().await.expect("Failed to destroy");
4511 });
4512 }
4513
4514 #[test_traced]
4515 fn test_get_value_size_just_over_crc() {
4516 let executor = deterministic::Runner::default();
4519 executor.start(|context| async move {
4520 let cfg = test_cfg(&context);
4521 let mut oversized: Oversized<_, TestEntry, TestValue> =
4522 Oversized::init(context, cfg).await.expect("Failed to init");
4523
4524 let value: TestValue = [42; 16];
4525 let entry = TestEntry::new(1, 0, 0);
4526 let offset;
4527 (oversized, _, offset, _) = oversized
4528 .append(1, entry, &value)
4529 .await
4530 .expect("Failed to append");
4531 oversized = oversized.sync(1).await.expect("Failed to sync");
4532
4533 let result = oversized.get_value(1, offset, 5).await;
4536 assert!(result.is_err());
4537
4538 oversized.destroy().await.expect("Failed to destroy");
4539 });
4540 }
4541
4542 #[test_traced]
4543 fn test_recovery_maximum_section_numbers() {
4544 let executor = deterministic::Runner::default();
4547 executor.start(|context| async move {
4548 let cfg = test_cfg(&context);
4549
4550 let large_sections = [u64::MAX - 3, u64::MAX - 2, u64::MAX - 1];
4552
4553 let mut oversized: Oversized<_, TestEntry, TestValue> =
4555 Oversized::init(context.child("first"), cfg.clone())
4556 .await
4557 .expect("Failed to init");
4558
4559 let mut locations = Vec::new();
4560 for §ion in &large_sections {
4561 let value: TestValue = [(section & 0xFF) as u8; 16];
4562 let entry = TestEntry::new(section, 0, 0);
4563 let (position, offset, size);
4564 (oversized, position, offset, size) = oversized
4565 .append(section, entry, &value)
4566 .await
4567 .expect("Failed to append");
4568 locations.push((section, (position, offset, size)));
4569 oversized = oversized.sync(section).await.expect("Failed to sync");
4570 }
4571 drop(oversized);
4572
4573 let middle_section = large_sections[1];
4575 let (blob, size) = context
4576 .open(&cfg.value_partition, &middle_section.to_be_bytes())
4577 .await
4578 .expect("Failed to open blob");
4579 blob.resize(size / 2).await.expect("Failed to truncate");
4580 blob.sync().await.expect("Failed to sync");
4581 drop(blob);
4582
4583 let mut oversized: Oversized<_, TestEntry, TestValue> =
4585 Oversized::init(context.child("second"), cfg.clone())
4586 .await
4587 .expect("Failed to reinit");
4588
4589 let entry = oversized
4591 .get(large_sections[0], 0)
4592 .await
4593 .expect("Failed to get first section");
4594 assert_eq!(entry.id, large_sections[0]);
4595
4596 let entry = oversized
4597 .get(large_sections[2], 0)
4598 .await
4599 .expect("Failed to get last section");
4600 assert_eq!(entry.id, large_sections[2]);
4601
4602 assert!(oversized.get(middle_section, 0).await.is_err());
4604
4605 let new_value: TestValue = [0xAB; 16];
4607 let new_entry = TestEntry::new(999, 0, 0);
4608 (oversized, _, _, _) = oversized
4609 .append(middle_section, new_entry, &new_value)
4610 .await
4611 .expect("Failed to append after recovery");
4612
4613 oversized.destroy().await.expect("Failed to destroy");
4614 });
4615 }
4616
4617 #[test_traced]
4618 fn test_recovery_crash_during_recovery_rewind() {
4619 let executor = deterministic::Runner::default();
4623 executor.start(|context| async move {
4624 let cfg = test_cfg(&context);
4625
4626 let mut oversized: Oversized<_, TestEntry, TestValue> =
4628 Oversized::init(context.child("first"), cfg.clone())
4629 .await
4630 .expect("Failed to init");
4631
4632 let mut locations = Vec::new();
4633 for i in 0..5u8 {
4634 let value: TestValue = [i; 16];
4635 let entry = TestEntry::new(i as u64, 0, 0);
4636 let (position, offset, size);
4637 (oversized, position, offset, size) = oversized
4638 .append(1, entry, &value)
4639 .await
4640 .expect("Failed to append");
4641 locations.push((position, offset, size));
4642 }
4643 oversized = oversized.sync(1).await.expect("Failed to sync");
4644 drop(oversized);
4645
4646 let (blob, _) = context
4648 .open(&cfg.value_partition, &1u64.to_be_bytes())
4649 .await
4650 .expect("Failed to open blob");
4651 let keep_size = byte_end(locations[2].1, locations[2].2);
4652 blob.resize(keep_size).await.expect("Failed to truncate");
4653 blob.sync().await.expect("Failed to sync");
4654 drop(blob);
4655
4656 let chunk_size = FixedJournal::<deterministic::Context, TestEntry>::CHUNK_SIZE as u64;
4661 let (index_blob, _) = context
4662 .open(&cfg.index_partition, &1u64.to_be_bytes())
4663 .await
4664 .expect("Failed to open index blob");
4665 let partial_rewind_size = 4 * chunk_size; index_blob
4667 .resize(partial_rewind_size)
4668 .await
4669 .expect("Failed to resize");
4670 index_blob.sync().await.expect("Failed to sync");
4671 drop(index_blob);
4672
4673 let mut oversized: Oversized<_, TestEntry, TestValue> =
4676 Oversized::init(context.child("second"), cfg.clone())
4677 .await
4678 .expect("Failed to reinit after nested crash");
4679
4680 for i in 0..3u8 {
4682 let entry = oversized.get(1, i as u64).await.expect("Failed to get");
4683 assert_eq!(entry.id, i as u64);
4684
4685 let (_, offset, size) = locations[i as usize];
4686 let value = oversized
4687 .get_value(1, offset, size)
4688 .await
4689 .expect("Failed to get value");
4690 assert_eq!(value, [i; 16]);
4691 }
4692
4693 assert!(oversized.get(1, 3).await.is_err());
4695
4696 let new_value: TestValue = [0xFF; 16];
4698 let new_entry = TestEntry::new(100, 0, 0);
4699 let (pos, offset, _size);
4700 (oversized, pos, offset, _size) = oversized
4701 .append(1, new_entry, &new_value)
4702 .await
4703 .expect("Failed to append");
4704 assert_eq!(pos, 3); assert_eq!(offset, byte_end(locations[2].1, locations[2].2));
4708
4709 oversized.destroy().await.expect("Failed to destroy");
4710 });
4711 }
4712
4713 #[test_traced]
4714 fn test_recovery_crash_during_orphan_cleanup() {
4715 let executor = deterministic::Runner::default();
4718 executor.start(|context| async move {
4719 let cfg = test_cfg(&context);
4720
4721 let mut oversized: Oversized<_, TestEntry, TestValue> =
4723 Oversized::init(context.child("first"), cfg.clone())
4724 .await
4725 .expect("Failed to init");
4726
4727 let value: TestValue = [1; 16];
4728 let entry = TestEntry::new(1, 0, 0);
4729 let (offset1, size1);
4730 (oversized, _, offset1, size1) = oversized
4731 .append(1, entry, &value)
4732 .await
4733 .expect("Failed to append");
4734 oversized = oversized.sync(1).await.expect("Failed to sync");
4735 drop(oversized);
4736
4737 let glob_cfg = GlobConfig {
4739 partition: cfg.value_partition.clone(),
4740 compression: cfg.compression,
4741 codec_config: (),
4742 write_buffer: cfg.value_write_buffer,
4743 };
4744 let mut glob: Glob<_, TestValue> = Glob::init(context.child("glob"), glob_cfg)
4745 .await
4746 .expect("Failed to init glob");
4747
4748 for section in 2u64..=4 {
4749 let orphan_value: TestValue = [section as u8; 16];
4750 (glob, _, _) = glob
4751 .append(section, &orphan_value)
4752 .await
4753 .expect("Failed to append orphan");
4754 glob = glob.sync(section).await.expect("Failed to sync glob");
4755 }
4756 drop(glob);
4757
4758 context
4761 .remove(&cfg.value_partition, Some(&2u64.to_be_bytes()))
4762 .await
4763 .expect("Failed to remove section 2");
4764
4765 let mut oversized: Oversized<_, TestEntry, TestValue> =
4767 Oversized::init(context.child("second"), cfg.clone())
4768 .await
4769 .expect("Failed to reinit");
4770
4771 let entry = oversized.get(1, 0).await.expect("Failed to get");
4773 assert_eq!(entry.id, 1);
4774 let value = oversized
4775 .get_value(1, offset1, size1)
4776 .await
4777 .expect("Failed to get value");
4778 assert_eq!(value, [1; 16]);
4779
4780 assert_eq!(oversized.oldest_section(), Some(1));
4782 assert_eq!(oversized.newest_section(), Some(1));
4783
4784 let new_value: TestValue = [42; 16];
4786 let new_entry = TestEntry::new(42, 0, 0);
4787 let pos;
4788 (oversized, pos, _, _) = oversized
4789 .append(2, new_entry, &new_value)
4790 .await
4791 .expect("Failed to append to section 2");
4792 assert_eq!(pos, 0); oversized.destroy().await.expect("Failed to destroy");
4795 });
4796 }
4797
4798 #[test_traced]
4799 fn test_rewind_to_zero_index_size() {
4800 let executor = deterministic::Runner::default();
4801 executor.start(|context| async move {
4802 let cfg = test_cfg(&context);
4803 let mut oversized: Oversized<_, TestEntry, TestValue> =
4804 Oversized::init(context, cfg).await.expect("Failed to init");
4805
4806 let value: TestValue = [1; 16];
4807 let entry = TestEntry::new(1, 0, 0);
4808 (oversized, _, _, _) = oversized
4809 .append(0, entry, &value)
4810 .await
4811 .expect("Failed to append");
4812 oversized = oversized.sync(0).await.expect("Failed to sync");
4813
4814 oversized = oversized
4815 .rewind(0, 0)
4816 .await
4817 .expect("rewind to zero index_size must not fail");
4818
4819 assert_eq!(oversized.last(0).await.unwrap(), None);
4820 assert_eq!(oversized.size(0).unwrap(), 0);
4821 assert_eq!(oversized.value_size(0).await.unwrap(), 0);
4822
4823 oversized.destroy().await.expect("Failed to destroy");
4824 });
4825 }
4826
4827 #[test_traced]
4828 fn test_rewind_to_zero_on_missing_section() {
4829 let executor = deterministic::Runner::default();
4830 executor.start(|context| async move {
4831 let cfg = test_cfg(&context);
4832 let mut oversized: Oversized<_, TestEntry, TestValue> =
4833 Oversized::init(context, cfg).await.expect("Failed to init");
4834
4835 oversized = oversized
4836 .rewind(0, 0)
4837 .await
4838 .expect("rewind on missing section must not fail");
4839
4840 assert!(matches!(
4841 oversized.last(0).await,
4842 Err(Error::SectionOutOfRange(0))
4843 ));
4844 assert_eq!(oversized.value_size(0).await.unwrap(), 0);
4845
4846 oversized.destroy().await.expect("Failed to destroy");
4847 });
4848 }
4849
4850 #[test_traced]
4851 fn test_rewind_nonzero_on_missing_section_errors() {
4852 let executor = deterministic::Runner::default();
4853 executor.start(|context| async move {
4854 let cfg = test_cfg(&context);
4855 let oversized: Oversized<_, TestEntry, TestValue> =
4856 Oversized::init(context, cfg).await.expect("Failed to init");
4857
4858 let result = oversized.rewind(0, 1).await;
4859 assert!(
4860 matches!(result, Err(Error::SectionOutOfRange(0))),
4861 "nonzero index_size on missing section must fail, got: {result:?}"
4862 );
4863 });
4864 }
4865
4866 #[test_traced]
4867 fn test_rewind_section_nonzero_on_missing_section_errors() {
4868 let executor = deterministic::Runner::default();
4869 executor.start(|context| async move {
4870 let cfg = test_cfg(&context);
4871 let oversized: Oversized<_, TestEntry, TestValue> =
4872 Oversized::init(context, cfg).await.expect("Failed to init");
4873
4874 let result = oversized.rewind_section(0, 1).await;
4875 assert!(
4876 matches!(result, Err(Error::SectionOutOfRange(0))),
4877 "nonzero index_size on missing section must fail, got: {result:?}"
4878 );
4879 });
4880 }
4881
4882 #[test_traced]
4883 fn test_last_pruned_section_returns_error() {
4884 let executor = deterministic::Runner::default();
4885 executor.start(|context| async move {
4886 let cfg = test_cfg(&context);
4887 let mut oversized: Oversized<_, TestEntry, TestValue> =
4888 Oversized::init(context, cfg).await.expect("Failed to init");
4889
4890 let value: TestValue = [1; 16];
4891 (oversized, _, _, _) = oversized
4892 .append(0, TestEntry::new(1, 0, 0), &value)
4893 .await
4894 .expect("Failed to append");
4895 (oversized, _, _, _) = oversized
4896 .append(1, TestEntry::new(2, 0, 0), &value)
4897 .await
4898 .expect("Failed to append");
4899 oversized = oversized.sync_all().await.expect("Failed to sync");
4900
4901 (oversized, _) = oversized.prune(1).await.expect("Failed to prune");
4902
4903 assert!(matches!(
4904 oversized.last(0).await,
4905 Err(Error::AlreadyPrunedToSection(1))
4906 ));
4907 assert!(oversized.last(1).await.unwrap().is_some());
4908
4909 oversized.destroy().await.expect("Failed to destroy");
4910 });
4911 }
4912}