improve test coverage

Author: roberto-bayardo

storage/src/journal/contiguous/fixed.rs

@@ -1300,6 +1300,17 @@ impl<E: Context, A: CodecFixedShared> Journal<E, A> {
         let inner = self.inner.read().await;
         inner.journal.newest_section()
     }
+
+    /// Test helper: Set and persist the recovery watermark directly.
+    #[cfg(test)]
+    pub(crate) async fn test_set_recovery_watermark(&self, watermark: u64) -> Result<(), Error> {
+        let mut inner = self.inner.write().await;
+        inner
+            .metadata
+            .put(RECOVERY_WATERMARK_KEY, watermark.to_be_bytes().to_vec());
+        inner.metadata.sync().await?;
+        Ok(())
+    }
 }
 
 // Implement Contiguous trait for fixed-length journals
@@ -2227,6 +2238,59 @@ mod tests {
         });
     }
 
+    #[test_traced]
+    fn test_fixed_journal_stale_pruning_metadata_without_watermark_walks_lengths() {
+        let executor = deterministic::Runner::default();
+        executor.start(|context| async move {
+            let cfg = test_cfg(&context, NZU64!(5));
+            let journal =
+                Journal::<_, Digest>::init_at_size(context.child("first"), cfg.clone(), 7)
+                    .await
+                    .expect("failed to initialize journal at size");
+
+            for i in 0..10u64 {
+                journal
+                    .append(&test_digest(i))
+                    .await
+                    .expect("failed to append data");
+            }
+            journal.sync().await.expect("failed to sync journal");
+            assert_eq!(journal.bounds().await, 7..17);
+
+            {
+                let mut inner = journal.inner.write().await;
+                inner.metadata.remove(&RECOVERY_WATERMARK_KEY);
+                inner
+                    .metadata
+                    .sync()
+                    .await
+                    .expect("failed to remove recovery watermark");
+            }
+            drop(journal);
+
+            // Remove the metadata's oldest section so PRUNING_BOUNDARY_KEY=7 is stale. Without a
+            // recovery watermark, recovery must still walk lengths from the recovered blob boundary.
+            context
+                .remove(&blob_partition(&cfg), Some(&1u64.to_be_bytes()))
+                .await
+                .expect("failed to remove stale oldest section");
+
+            let journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
+                .await
+                .expect("failed to recover journal");
+            assert_eq!(journal.bounds().await, 10..17);
+            assert_eq!(journal.recovery_watermark().await, 17);
+            assert_eq!(journal.read(10).await.unwrap(), test_digest(3));
+            assert_eq!(journal.read(16).await.unwrap(), test_digest(9));
+            assert!(matches!(
+                journal.read(17).await,
+                Err(Error::ItemOutOfRange(17))
+            ));
+
+            journal.destroy().await.unwrap();
+        });
+    }
+
     #[test_traced]
     fn test_fixed_journal_legacy_recovery_installs_watermark() {
         let executor = deterministic::Runner::default();
@@ -2287,6 +2351,88 @@ mod tests {
         });
     }
 
+    #[test_traced]
+    fn test_fixed_journal_update_metadata_watermark_before_clear_lowers_only() {
+        let executor = deterministic::Runner::default();
+        executor.start(|context| async move {
+            let cfg = test_cfg(&context, NZU64!(5));
+            let meta_cfg = MetadataConfig {
+                partition: format!("{}-metadata", cfg.partition),
+                codec_config: ((0..).into(), ()),
+            };
+            let mut metadata =
+                Metadata::<_, u64, Vec<u8>>::init(context.child("metadata"), meta_cfg)
+                    .await
+                    .expect("failed to initialize metadata");
+            metadata.put(RECOVERY_WATERMARK_KEY, 7u64.to_be_bytes().to_vec());
+
+            let changed =
+                Journal::<_, Digest>::update_metadata_watermark_before_clear(&mut metadata, 9)
+                    .expect("failed to update metadata watermark");
+            assert!(!changed);
+            let raw_watermark = metadata
+                .get(&RECOVERY_WATERMARK_KEY)
+                .expect("missing recovery watermark");
+            let persisted_watermark =
+                u64::from_be_bytes(raw_watermark.as_slice().try_into().unwrap());
+            assert_eq!(persisted_watermark, 7);
+
+            let changed =
+                Journal::<_, Digest>::update_metadata_watermark_before_clear(&mut metadata, 5)
+                    .expect("failed to update metadata watermark");
+            assert!(changed);
+            let raw_watermark = metadata
+                .get(&RECOVERY_WATERMARK_KEY)
+                .expect("missing recovery watermark");
+            let persisted_watermark =
+                u64::from_be_bytes(raw_watermark.as_slice().try_into().unwrap());
+            assert_eq!(persisted_watermark, 5);
+        });
+    }
+
+    #[test_traced]
+    fn test_fixed_journal_prune_to_blob_boundary_removes_pruning_metadata() {
+        let executor = deterministic::Runner::default();
+        executor.start(|context| async move {
+            let cfg = test_cfg(&context, NZU64!(5));
+            let journal =
+                Journal::<_, Digest>::init_at_size(context.child("first"), cfg.clone(), 7)
+                    .await
+                    .expect("failed to initialize journal at size");
+
+            for i in 0..8u64 {
+                journal
+                    .append(&test_digest(i))
+                    .await
+                    .expect("failed to append data");
+            }
+            journal.sync().await.expect("failed to sync journal");
+            assert_eq!(journal.bounds().await, 7..15);
+
+            journal.prune(10).await.expect("failed to prune journal");
+            journal.sync().await.expect("failed to sync pruned journal");
+            assert_eq!(journal.bounds().await, 10..15);
+            drop(journal);
+
+            let meta_cfg = MetadataConfig {
+                partition: format!("{}-metadata", cfg.partition),
+                codec_config: ((0..).into(), ()),
+            };
+            let metadata = Metadata::<_, u64, Vec<u8>>::init(context.child("metadata"), meta_cfg)
+                .await
+                .expect("failed to reopen metadata");
+            assert!(metadata.get(&PRUNING_BOUNDARY_KEY).is_none());
+            drop(metadata);
+
+            let journal = Journal::<_, Digest>::init(context.child("second"), cfg.clone())
+                .await
+                .expect("failed to reopen journal");
+            assert_eq!(journal.bounds().await, 10..15);
+            assert_eq!(journal.read(10).await.unwrap(), test_digest(3));
+            journal.destroy().await.unwrap();
+        });
+    }
+
     #[test_traced]
     fn test_fixed_journal_recover_rejects_overlong_section() {
         let executor = deterministic::Runner::default();

storage/src/journal/contiguous/variable.rs

@@ -1156,10 +1156,6 @@ impl<E: Context, V: CodecShared> Journal<E, V> {
         let start_section = position_to_section(anchor, items_per_section);
         let first_position = pruning_boundary.max(start_section * items_per_section);
 
-        if anchor < first_position {
-            return Ok(None);
-        }
-
         let skip = anchor - first_position;
         let stream = data.replay(start_section, 0, REPLAY_BUFFER_SIZE).await?;
         futures::pin_mut!(stream);
@@ -1301,11 +1297,30 @@ impl<E: Context, V: CodecShared> Journal<E, V> {
         self.offsets.rewind(position).await
     }
 
+    /// Test helper: Set and persist the offsets recovery watermark directly.
+    pub(crate) async fn test_set_offsets_recovery_watermark(
+        &self,
+        watermark: u64,
+    ) -> Result<(), Error> {
+        self.offsets.test_set_recovery_watermark(watermark).await
+    }
+
     /// Test helper: Get the size of the internal offsets journal.
     pub(crate) async fn test_offsets_size(&self) -> u64 {
         self.offsets.size().await
     }
 
+    /// Test helper: Rewind the internal data journal to the item at `position`.
+    pub(crate) async fn test_rewind_data_to_position(&self, position: u64) -> Result<(), Error> {
+        let offset = {
+            let offsets_reader = self.offsets.reader().await;
+            offsets_reader.read(position).await?
+        };
+        let section = position_to_section(position, self.items_per_section);
+        let mut inner = self.inner.write().await;
+        inner.data.rewind_to_offset(section, offset).await
+    }
+
     /// Test helper: Append directly to the internal data journal (simulates crash scenario).
     pub(crate) async fn test_append_data(
         &self,
@@ -2196,6 +2211,137 @@ mod tests {
         });
     }
 
+    #[test_traced]
+    fn test_variable_recovery_offsets_watermark_outside_bounds_rebuilds_from_start() {
+        let executor = deterministic::Runner::default();
+        executor.start(|context| async move {
+            let cfg = Config {
+                partition: "recovery-watermark-outside-bounds".into(),
+                items_per_section: NZU64!(10),
+                compression: None,
+                codec_config: (),
+                page_cache: CacheRef::from_pooler(&context, LARGE_PAGE_SIZE, NZUsize!(10)),
+                write_buffer: NZUsize!(1024),
+            };
+
+            let journal = Journal::<_, u64>::init(context.child("first"), cfg.clone())
+                .await
+                .unwrap();
+
+            for i in 0..15u64 {
+                journal.append(&(i * 100)).await.unwrap();
+            }
+            journal.sync().await.unwrap();
+
+            // Simulate stale metadata that points past the recovered offsets bounds.
+            journal
+                .test_set_offsets_recovery_watermark(30)
+                .await
+                .unwrap();
+            drop(journal);
+
+            let journal = Journal::<_, u64>::init(context.child("second"), cfg.clone())
+                .await
+                .unwrap();
+            assert_eq!(journal.bounds().await, 0..15);
+            assert_eq!(journal.test_offsets_size().await, 15);
+            for i in 0..15u64 {
+                assert_eq!(journal.read(i).await.unwrap(), i * 100);
+            }
+
+            journal.destroy().await.unwrap();
+        });
+    }
+
+    #[test_traced]
+    fn test_variable_rebuild_offsets_anchor_outside_bounds_returns_none() {
+        let executor = deterministic::Runner::default();
+        executor.start(|context| async move {
+            let data_cfg = variable::Config {
+                partition: "rebuild-anchor-outside-data".into(),
+                compression: None,
+                codec_config: (),
+                page_cache: CacheRef::from_pooler(&context, LARGE_PAGE_SIZE, NZUsize!(10)),
+                write_buffer: NZUsize!(1024),
+            };
+            let offsets_cfg = fixed::Config {
+                partition: "rebuild-anchor-outside-offsets".into(),
+                items_per_blob: NZU64!(10),
+                page_cache: CacheRef::from_pooler(&context, LARGE_PAGE_SIZE, NZUsize!(10)),
+                write_buffer: NZUsize!(1024),
+            };
+
+            let mut data = variable::Journal::<_, u64>::init(context.child("data"), data_cfg)
+                .await
+                .unwrap();
+            let mut offsets = fixed::Journal::<_, u64>::init(context.child("offsets"), offsets_cfg)
+                .await
+                .unwrap();
+
+            let (offset, _) = data.append(0, &100).await.unwrap();
+            offsets.append(&offset).await.unwrap();
+
+            let result =
+                Journal::<_, u64>::rebuild_offsets_from_anchor(&data, &mut offsets, 10, 0, 2)
+                    .await
+                    .unwrap();
+            assert!(result.is_none());
+
+            data.destroy().await.unwrap();
+            offsets.destroy().await.unwrap();
+        });
+    }
+
+    #[test_traced]
+    fn test_variable_recovery_retries_from_pruning_boundary_when_anchor_too_far() {
+        let executor = deterministic::Runner::default();
+        executor.start(|context| async move {
+            let cfg = Config {
+                partition: "recovery-anchor-too-far".into(),
+                items_per_section: NZU64!(10),
+                compression: None,
+                codec_config: (),
+                page_cache: CacheRef::from_pooler(&context, LARGE_PAGE_SIZE, NZUsize!(10)),
+                write_buffer: NZUsize!(1024),
+            };
+
+            let journal = Journal::<_, u64>::init(context.child("first"), cfg.clone())
+                .await
+                .unwrap();
+
+            for i in 0..20u64 {
+                journal.append(&(i * 100)).await.unwrap();
+            }
+            journal.sync().await.unwrap();
+
+            // The offsets watermark is in-bounds, but the data journal is shorter than that
+            // anchor. Recovery should retry from the pruning boundary and rebuild only the
+            // retained data prefix.
+            journal
+                .test_set_offsets_recovery_watermark(15)
+                .await
+                .unwrap();
+            journal.test_rewind_data_to_position(12).await.unwrap();
+            journal.test_sync_data().await.unwrap();
+            drop(journal);
+
+            let journal = Journal::<_, u64>::init(context.child("second"), cfg.clone())
+                .await
+                .unwrap();
+            assert_eq!(journal.bounds().await, 0..12);
+            assert_eq!(journal.test_offsets_size().await, 12);
+            for i in 0..12u64 {
+                assert_eq!(journal.read(i).await.unwrap(), i * 100);
+            }
+            assert!(matches!(
+                journal.read(12).await,
+                Err(Error::ItemOutOfRange(12))
+            ));
+
+            journal.destroy().await.unwrap();
+        });
+    }
+
     #[test_traced]
     fn test_variable_rewind_commit_reopen() {
         let executor = deterministic::Runner::default();