trim chain

Author: roberto-bayardo

storage/src/qmdb/current/batch.rs

@@ -767,6 +767,42 @@ pub(crate) struct BitmapBatchLayer<const N: usize> {
 
 impl<const N: usize> BitmapBatch<N> {
     const CHUNK_SIZE_BITS: u64 = BitMap::<N>::CHUNK_SIZE_BITS;
+
+    /// Walk to the terminal [`SharedBitmap`] at the bottom of the chain.
+    fn shared(&self) -> &Arc<SharedBitmap<N>> {
+        let mut current = self;
+        loop {
+            match current {
+                Self::Base(s) => return s,
+                Self::Layer(layer) => current = &layer.parent,
+            }
+        }
+    }
+
+    /// Return a chain equivalent to `self` with any `Layer` whose overlay is now fully committed
+    /// replaced by a direct reference to the committed bitmap. Since `apply_batch` commits
+    /// contiguous prefixes, committed `Layer`s are always at the bottom of the chain.
+    fn trim_committed(&self) -> Self {
+        let shared = self.shared();
+        let committed = shared.read().len();
+        let mut kept: Vec<Arc<ChunkOverlay<N>>> = Vec::new();
+        let mut current = self;
+        while let Self::Layer(layer) = current {
+            if layer.overlay.len <= committed {
+                let mut result = Self::Base(Arc::clone(shared));
+                for overlay in kept.into_iter().rev() {
+                    result = Self::Layer(Arc::new(BitmapBatchLayer {
+                        parent: result,
+                        overlay,
+                    }));
+                }
+                return result;
+            }
+            kept.push(Arc::clone(&layer.overlay));
+            current = &layer.parent;
+        }
+        self.clone()
+    }
 }
 
 impl<const N: usize> BitmapReadable<N> for BitmapBatch<N> {
@@ -861,7 +897,7 @@ where
         UnmerkleizedBatch::new(
             self.inner.new_batch::<H>(),
             Arc::clone(&self.grafted),
-            self.bitmap.clone(),
+            self.bitmap.trim_committed(),
         )
     }