[storage] fix rewind bugs (#1521)

Co-authored-by: Patrick O'Grady <me@patrickogrady.xyz>

Author: roberto-bayardo

storage/fuzz/fuzz_targets/journal_operations.rs

@@ -121,6 +121,7 @@ fn fuzz(input: FuzzInput) {
                 JournalOperation::Rewind { size } => {
                     if *size <= journal_size && *size >= oldest_retained_pos {
                         journal.rewind(*size).await.unwrap();
+                        journal.sync().await.unwrap();
                         journal_size = *size;
                     }
                 }

storage/src/adb/any/fixed/mod.rs

@@ -220,6 +220,7 @@ impl<
             let op_count = log_size - rewind_leaf_num;
             warn!(op_count, "rewinding over uncommitted log operations");
             log.rewind(rewind_leaf_num).await?;
+            log.sync().await?;
             log_size = rewind_leaf_num;
         }
 

storage/src/adb/any/fixed/sync.rs

@@ -214,6 +214,7 @@ pub(crate) async fn init_journal<E: Storage + Metrics, A: CodecFixed<Cfg = ()>>(
             );
         journal.prune(lower_bound).await?;
         journal.rewind(upper_bound + 1).await?; // +1 because upper_bound is inclusive
+        journal.sync().await?;
         journal
     };
     let journal_size = journal.size().await?;

storage/src/adb/any/variable/mod.rs

@@ -225,6 +225,7 @@ impl<E: RStorage + Clock + Metrics, K: Array, V: Codec, H: CHasher, T: Translato
                 locations_size, "rewinding misaligned locations map"
             );
             self.locations.rewind(mmr_leaves).await?;
+            self.locations.sync().await?;
         } else if mmr_leaves > locations_size {
             warn!(mmr_leaves, locations_size, "rewinding misaligned mmr");
             self.mmr.pop((mmr_leaves - locations_size) as usize).await?;
@@ -347,6 +348,7 @@ impl<E: RStorage + Clock + Metrics, K: Array, V: Codec, H: CHasher, T: Translato
         // Pop any MMR elements that are ahead of the last log commit point.
         if mmr_leaves > self.log_size {
             self.locations.rewind(self.log_size).await?;
+            self.locations.sync().await?;
 
             let op_count = mmr_leaves - self.log_size;
             warn!(op_count, "popping uncommitted MMR operations");

storage/src/adb/immutable/mod.rs

@@ -297,6 +297,7 @@ impl<E: RStorage + Clock + Metrics, K: Array, V: Codec, H: CHasher, T: Translato
                 locations_size, "rewinding misaligned locations map"
             );
             locations.rewind(mmr_leaves).await?;
+            locations.sync().await?;
         }
         if mmr_leaves > locations_size {
             warn!(mmr_leaves, locations_size, "rewinding misaligned mmr");
@@ -391,6 +392,7 @@ impl<E: RStorage + Clock + Metrics, K: Array, V: Codec, H: CHasher, T: Translato
         // Pop any MMR elements that are ahead of the last log commit point.
         if mmr_leaves > log_size {
             locations.rewind(log_size).await?;
+            locations.sync().await?;
 
             let op_count = mmr_leaves - log_size;
             warn!(op_count, "popping uncommitted MMR operations");

storage/src/adb/keyless.rs

@@ -146,6 +146,7 @@ impl<E: Storage + Clock + Metrics, V: Codec, H: CHasher> Keyless<E, V, H> {
                 locations_size, "rewinding misaligned locations journal"
             );
             locations.rewind(mmr_leaves).await?;
+            locations.sync().await?;
             locations_size = mmr_leaves;
         } else if mmr_leaves > locations_size {
             warn!(mmr_leaves, locations_size, "rewinding misaligned mmr");
@@ -195,6 +196,7 @@ impl<E: Storage + Clock + Metrics, V: Codec, H: CHasher> Keyless<E, V, H> {
                     "rewinding to last commit point"
                 );
                 locations.rewind(last_commit_loc + 1).await?;
+                locations.sync().await?;
                 mmr.pop((locations_size - last_commit_loc - 1) as usize)
                     .await?;
                 locations_size = last_commit_loc + 1;
@@ -203,16 +205,19 @@ impl<E: Storage + Clock + Metrics, V: Codec, H: CHasher> Keyless<E, V, H> {
                 let rewind_point = rewind_point.expect("no rewind point found");
                 let section = rewind_point.0 / cfg.log_items_per_section.get();
                 log.rewind_to_offset(section, rewind_point.1).await?;
+                log.sync(section).await?;
             }
         } else if locations_size > 0 {
             warn!(
                 old_size = locations_size,
                 "no commit point found, rewinding to start"
             );
             locations.rewind(0).await?;
+            locations.sync().await?;
             mmr.pop(locations_size as usize).await?;
             locations_size = 0;
             log.rewind_section(0, 0).await?;
+            log.sync(0).await?;
         }
 
         Ok(Self {

storage/src/freezer/storage.rs

@@ -578,6 +578,7 @@ impl<E: Storage + Metrics + Clock, K: Array, V: Codec> Freezer<E, K, V> {
 
                 // Rewind the journal to the committed section and offset
                 journal.rewind(checkpoint.section, checkpoint.size).await?;
+                journal.sync(checkpoint.section).await?;
 
                 // Resize table if needed
                 let expected_table_len = Self::table_offset(checkpoint.table_size);

storage/src/journal/fixed.rs

@@ -362,11 +362,14 @@ impl<E: Storage + Metrics, A: CodecFixed<Cfg = ()>> Journal<E, A> {
         Ok(item_pos)
     }
 
-    /// Rewind the journal to the given `size`. Returns [Error::MissingBlob] if the rewind
-    /// point precedes the oldest retained element point. The journal is not synced after rewinding.
+    /// Rewind the journal to the given `size`. Returns [Error::MissingBlob] if the rewind point
+    /// precedes the oldest retained element point. The journal is not synced after rewinding.
     ///
-    /// Note that this operation is not atomic, but it will always leave the journal in a consistent
-    /// state in the event of failure since blobs are always removed from newest to oldest.
+    /// # Warnings
+    ///
+    /// * This operation is not guaranteed to survive restarts until sync is called.
+    /// * This operation is not atomic, but it will always leave the journal in a consistent state
+    ///   in the event of failure since blobs are always removed from newest to oldest.
     pub async fn rewind(&mut self, size: u64) -> Result<(), Error> {
         match size.cmp(&self.size().await?) {
             std::cmp::Ordering::Greater => return Err(Error::InvalidRewind(size)),

storage/src/journal/variable.rs

@@ -652,13 +652,25 @@ impl<E: Storage + Metrics, V: Codec> Journal<E, V> {
     }
 
     /// Rewinds the journal to the given `section` and `offset`, removing any data beyond it.
+    ///
+    /// # Warnings
+    ///
+    /// * This operation is not guaranteed to survive restarts until sync is called.
+    /// * This operation is not atomic, but it will always leave the journal in a consistent state
+    ///   in the event of failure since blobs are always removed in reverse order of section.
     pub async fn rewind_to_offset(&mut self, section: u64, offset: u32) -> Result<(), Error> {
         self.rewind(section, offset as u64 * ITEM_ALIGNMENT).await
     }
 
     /// Rewinds the journal to the given `section` and `size`.
     ///
     /// This removes any data beyond the specified `section` and `size`.
+    ///
+    /// # Warnings
+    ///
+    /// * This operation is not guaranteed to survive restarts until sync is called.
+    /// * This operation is not atomic, but it will always leave the journal in a consistent state
+    ///   in the event of failure since blobs are always removed in reverse order of section.
     pub async fn rewind(&mut self, section: u64, size: u64) -> Result<(), Error> {
         self.prune_guard(section, false)?;
 
@@ -707,6 +719,10 @@ impl<E: Storage + Metrics, V: Codec> Journal<E, V> {
     /// Rewinds the `section` to the given `size`.
     ///
     /// Unlike [Self::rewind], this method does not modify anything other than the given `section`.
+    ///
+    /// # Warning
+    ///
+    /// This operation is not guaranteed to survive restarts until sync is called.
     pub async fn rewind_section(&mut self, section: u64, size: u64) -> Result<(), Error> {
         self.prune_guard(section, false)?;
 

storage/src/mmr/journaled.rs

@@ -266,6 +266,7 @@ impl<E: RStorage + Clock + Metrics, H: CHasher> Mmr<E, H> {
                 orphaned_leaf = Some(item);
             }
             journal.rewind(last_valid_size).await?;
+            journal.sync().await?;
             journal_size = last_valid_size
         }
 
@@ -483,7 +484,8 @@ impl<E: RStorage + Clock + Metrics, H: CHasher> Mmr<E, H> {
     }
 
     /// Pop the given number of elements from the tip of the MMR assuming they exist, and otherwise
-    /// return Empty or ElementPruned errors.
+    /// return Empty or ElementPruned errors. The backing journal is synced to disk before
+    /// returning.
     ///
     /// # Warning
     ///
@@ -522,6 +524,7 @@ impl<E: RStorage + Clock + Metrics, H: CHasher> Mmr<E, H> {
         }
 
         self.journal.rewind(new_size).await?;
+        self.journal.sync().await?;
         self.journal_size = new_size;
 
         // Reset the mem_mmr to one of the new_size in the "prune_all" state.
@@ -561,10 +564,6 @@ impl<E: RStorage + Clock + Metrics, H: CHasher> Mmr<E, H> {
     /// Process all batched updates and sync the MMR to disk. If `pool` is non-null, then it will be
     /// used to parallelize the sync.
     pub async fn sync(&mut self, h: &mut impl Hasher<H>) -> Result<(), Error> {
-        if self.size() == 0 {
-            return Ok(());
-        }
-
         // Write the nodes cached in the memory-resident MMR to the journal.
         self.mem_mmr.sync(h);
 
@@ -852,6 +851,20 @@ mod tests {
             assert!(mmr.prune_to_pos(&mut hasher, 0).await.is_ok());
             assert!(mmr.sync(&mut hasher).await.is_ok());
             assert!(matches!(mmr.pop(1).await, Err(Error::Empty)));
+
+            mmr.add(&mut hasher, &test_digest(0)).await.unwrap();
+            assert_eq!(mmr.size(), 1);
+            mmr.sync(&mut hasher).await.unwrap();
+            assert!(mmr.get_node(0).await.is_ok());
+            assert!(mmr.pop(1).await.is_ok());
+            assert_eq!(mmr.size(), 0);
+            mmr.sync(&mut hasher).await.unwrap();
+
+            let mmr = Mmr::init(context.clone(), &mut hasher, test_config())
+                .await
+                .unwrap();
+            assert_eq!(mmr.size(), 0);
+
             mmr.destroy().await.unwrap();
         });
     }