perf(storage): prune old block signatures with a range delete

crates/storage/src/api/traits.rs

@@ -43,6 +43,15 @@ pub trait StorageWriteBatch: Send {
     /// Delete multiple keys from a table.
     fn delete_batch(&mut self, table: Table, keys: Vec<Vec<u8>>) -> Result<(), Error>;
 
+    /// Delete every key in the half-open range `[from, to)` from a table.
+    ///
+    /// Unlike [`delete_batch`](Self::delete_batch), the caller does not need to
+    /// know the keys, so the write cost need not scale with the number of
+    /// entries covered: RocksDB records a single range tombstone instead of one
+    /// delete per key. Operations within a batch apply in call order, so a later
+    /// `put_batch` for a key inside the range still wins.
+    fn delete_range(&mut self, table: Table, from: &[u8], to: &[u8]) -> Result<(), Error>;
+
     /// Commit the batch, consuming it.
     fn commit(self: Box<Self>) -> Result<(), Error>;
 }

crates/storage/src/backend/in_memory.rs

@@ -8,13 +8,16 @@ use crate::api::{
 type TableData = HashMap<Vec<u8>, Vec<u8>>;
 type StorageData = HashMap<Table, TableData>;
 
-/// Pending operation for a key - last operation wins.
+/// Pending operation in a batch, replayed in call order on commit so the last
+/// operation touching a key wins (matching how RocksDB applies a `WriteBatch`).
 enum PendingOp {
-    Put(Vec<u8>),
-    Delete,
+    Put(Vec<u8>, Vec<u8>),
+    Delete(Vec<u8>),
+    /// Delete every key in the half-open range `[from, to)`.
+    DeleteRange(Vec<u8>, Vec<u8>),
 }
 
-type PendingOps = HashMap<Table, HashMap<Vec<u8>, PendingOp>>;
+type PendingOps = Vec<(Table, PendingOp)>;
 
 /// In-memory storage backend using HashMaps.
 ///
@@ -52,7 +55,7 @@ impl StorageBackend for InMemoryBackend {
     fn begin_write(&self) -> Result<Box<dyn StorageWriteBatch + 'static>, Error> {
         Ok(Box::new(InMemoryWriteBatch {
             data: Arc::clone(&self.data),
-            ops: HashMap::new(),
+            ops: PendingOps::new(),
         }))
     }
 }
@@ -105,34 +108,41 @@ struct InMemoryWriteBatch {
 
 impl StorageWriteBatch for InMemoryWriteBatch {
     fn put_batch(&mut self, table: Table, batch: Vec<(Vec<u8>, Vec<u8>)>) -> Result<(), Error> {
-        let table_ops = self.ops.entry(table).or_default();
         for (key, value) in batch {
-            table_ops.insert(key, PendingOp::Put(value));
+            self.ops.push((table, PendingOp::Put(key, value)));
         }
         Ok(())
     }
 
     fn delete_batch(&mut self, table: Table, keys: Vec<Vec<u8>>) -> Result<(), Error> {
-        let table_ops = self.ops.entry(table).or_default();
         for key in keys {
-            table_ops.insert(key, PendingOp::Delete);
+            self.ops.push((table, PendingOp::Delete(key)));
         }
         Ok(())
     }
 
+    fn delete_range(&mut self, table: Table, from: &[u8], to: &[u8]) -> Result<(), Error> {
+        let range = PendingOp::DeleteRange(from.to_vec(), to.to_vec());
+        self.ops.push((table, range));
+        Ok(())
+    }
+
     fn commit(self: Box<Self>) -> Result<(), Error> {
         let mut guard = self.data.write().map_err(|e| e.to_string())?;
 
-        for (table, ops) in self.ops {
+        for (table, op) in self.ops {
             let table_data = guard.get_mut(&table).expect("table exists");
-            for (key, op) in ops {
-                match op {
-                    PendingOp::Put(value) => {
-                        table_data.insert(key, value);
-                    }
-                    PendingOp::Delete => {
-                        table_data.remove(&key);
-                    }
+            match op {
+                PendingOp::Put(key, value) => {
+                    table_data.insert(key, value);
+                }
+                PendingOp::Delete(key) => {
+                    table_data.remove(&key);
+                }
+                // Keys are unordered here, so the range is applied by scanning
+                // the table rather than by seeking to `from`.
+                PendingOp::DeleteRange(from, to) => {
+                    table_data.retain(|key, _| key < &from || key >= &to);
                 }
             }
         }

crates/storage/src/backend/rocksdb.rs

@@ -167,6 +167,16 @@ impl StorageWriteBatch for RocksDBWriteBatch {
         Ok(())
     }
 
+    fn delete_range(&mut self, table: Table, from: &[u8], to: &[u8]) -> Result<(), Error> {
+        let cf = self
+            .db
+            .cf_handle(cf_name(table))
+            .ok_or_else(|| format!("Column family {} not found", cf_name(table)))?;
+
+        self.batch.delete_range_cf(&cf, from, to);
+        Ok(())
+    }
+
     fn commit(self: Box<Self>) -> Result<(), Error> {
         let mut write_opts = WriteOptions::default();
         write_opts.set_sync(false);

crates/storage/src/backend/tests.rs

@@ -19,6 +19,8 @@ pub fn run_backend_tests(backend: &dyn StorageBackend) {
     test_nonexistent_key(backend);
     test_delete_then_put(backend);
     test_put_then_delete(backend);
+    test_delete_range(backend);
+    test_delete_range_then_put(backend);
     test_multiple_tables(backend);
 }
 
@@ -172,6 +174,66 @@ fn test_put_then_delete(backend: &dyn StorageBackend) {
     );
 }
 
+fn test_delete_range(backend: &dyn StorageBackend) {
+    // Keys sort lexicographically, so `b` and `c` fall inside [b, d) while the
+    // bounds' neighbours `a` and `d` stay: the range is half-open.
+    {
+        let mut batch = backend.begin_write().unwrap();
+        let entries = ["a", "b", "c", "d"]
+            .iter()
+            .map(|suffix| (format!("test_range:{suffix}").into_bytes(), b"v".to_vec()))
+            .collect();
+        batch.put_batch(Table::LiveChain, entries).unwrap();
+        batch.commit().unwrap();
+    }
+
+    {
+        let mut batch = backend.begin_write().unwrap();
+        batch
+            .delete_range(Table::LiveChain, b"test_range:b", b"test_range:d")
+            .unwrap();
+        batch.commit().unwrap();
+    }
+
+    let view = backend.begin_read().unwrap();
+    let mut keys: Vec<_> = view
+        .prefix_iterator(Table::LiveChain, b"test_range:")
+        .unwrap()
+        .map(|entry| entry.unwrap().0)
+        .collect();
+
+    keys.sort();
+    assert_eq!(keys.len(), 2);
+    assert_eq!(&*keys[0], b"test_range:a");
+    assert_eq!(&*keys[1], b"test_range:d");
+}
+
+fn test_delete_range_then_put(backend: &dyn StorageBackend) {
+    // Range delete then put of a key inside the range - put should win.
+    {
+        let mut batch = backend.begin_write().unwrap();
+        let entries = vec![(b"test_range_put:b".to_vec(), b"old".to_vec())];
+        batch.put_batch(Table::LiveChain, entries).unwrap();
+        batch.commit().unwrap();
+    }
+
+    {
+        let mut batch = backend.begin_write().unwrap();
+        batch
+            .delete_range(Table::LiveChain, b"test_range_put:a", b"test_range_put:z")
+            .unwrap();
+        let entries = vec![(b"test_range_put:b".to_vec(), b"new".to_vec())];
+        batch.put_batch(Table::LiveChain, entries).unwrap();
+        batch.commit().unwrap();
+    }
+
+    let view = backend.begin_read().unwrap();
+    assert_eq!(
+        view.get(Table::LiveChain, b"test_range_put:b").unwrap(),
+        Some(b"new".to_vec())
+    );
+}
+
 fn test_multiple_tables(backend: &dyn StorageBackend) {
     // Write to different tables
     {

crates/storage/src/store.rs

@@ -909,11 +909,11 @@ impl Store {
             .map_or(finalized_slot, |header| {
                 header.expect("Failed to get block header").slot
             });
-        let pruned_signatures = self
+        let pruned_below_slot = self
             .prune_old_block_signatures(finalized_slot, tip_slot)
             .expect("prune old block signatures");
-        if pruned_signatures > 0 {
-            info!(pruned_signatures, "Pruned old finalized block signatures");
+        if pruned_below_slot > 0 {
+            info!(pruned_below_slot, "Pruned old finalized block signatures");
         }
         Ok(())
     }
@@ -1078,41 +1078,37 @@ impl Store {
     /// reverted, so their signatures are not needed for fork choice, re-org
     /// safety, or re-aggregation once outside the window.
     ///
-    /// Returns the number of signatures pruned.
+    /// Returns the exclusive slot below which signatures were dropped, or 0 when
+    /// nothing was pruned. This is a range delete, so the count of removed keys
+    /// is not known without reading the table back.
     pub fn prune_old_block_signatures(
         &mut self,
         finalized_slot: u64,
         tip_slot: u64,
-    ) -> Result<usize, Error> {
+    ) -> Result<u64, Error> {
         let cutoff = tip_slot.saturating_sub(SIGNATURE_PRUNING_RANGE);
         // Only prune when the whole window is finalized; never touch
-        // non-finalized signatures.
-        if cutoff > finalized_slot {
+        // non-finalized signatures. A zero cutoff covers nothing.
+        if cutoff > finalized_slot || cutoff == 0 {
             return Ok(0);
         }
 
-        let view = self.backend.begin_read().expect("read view");
-
-        // Keys are slot||root in big-endian slot order, so iteration ascends by
-        // slot: take entries below the cutoff and stop at the first one past it.
-        let keys_to_delete: Vec<Vec<u8>> = view
-            .prefix_iterator(Table::BlockSignatures, &[])
-            .expect("iterator")
-            .filter_map(|res| res.ok())
-            .map(|(key, _)| key.to_vec())
-            .take_while(|key| decode_slot_root_key(key).0 < cutoff)
-            .collect();
-        drop(view);
+        // Keys are slot||root in big-endian slot order, so the cutoff's bare
+        // slot prefix is an exact upper bound: keys below the cutoff sort
+        // before it, and keys at the cutoff sort after it (they extend it with
+        // a root). A single range delete drops them all without reading the
+        // table (and without walking the tombstones left by earlier prunes).
+        let mut batch = self.backend.begin_write().expect("write batch");
+        batch
+            .delete_range(
+                Table::BlockSignatures,
+                &0u64.to_be_bytes(),
+                &cutoff.to_be_bytes(),
+            )
+            .expect("delete finalized block signatures");
+        batch.commit().expect("commit");
 
-        let count = keys_to_delete.len();
-        if count > 0 {
-            let mut batch = self.backend.begin_write().expect("write batch");
-            batch
-                .delete_batch(Table::BlockSignatures, keys_to_delete)
-                .expect("delete finalized block signatures");
-            batch.commit().expect("commit");
-        }
-        Ok(count)
+        Ok(cutoff)
     }
 
     /// Get the block header by root.
@@ -1930,12 +1926,12 @@ mod tests {
         // tip = range + 10, finalized = range + 5, so cutoff = tip - range = 10.
         let tip_slot = SIGNATURE_PRUNING_RANGE + 10;
         let finalized_slot = SIGNATURE_PRUNING_RANGE + 5;
-        let pruned = store
+        let pruned_below_slot = store
             .prune_old_block_signatures(finalized_slot, tip_slot)
             .expect("prune");
 
         // cutoff = 10: slots 0..9 pruned, slots 10..12 kept (within the window).
-        assert_eq!(pruned, 10);
+        assert_eq!(pruned_below_slot, 10);
         assert_eq!(count_entries(backend.as_ref(), Table::BlockSignatures), 3);
 
         // Oldest signatures are gone, but headers, bodies, and roots stay queryable.
@@ -1965,10 +1961,10 @@ mod tests {
         // cutoff = tip - range > finalized → prune nothing.
         let tip_slot = SIGNATURE_PRUNING_RANGE + 100;
         let finalized_slot = 5;
-        let pruned = store
+        let pruned_below_slot = store
             .prune_old_block_signatures(finalized_slot, tip_slot)
             .expect("prune");
-        assert_eq!(pruned, 0);
+        assert_eq!(pruned_below_slot, 0);
         assert_eq!(count_entries(backend.as_ref(), Table::BlockSignatures), 10);
     }
 
@@ -1983,8 +1979,8 @@ mod tests {
 
         // Early chain: tip < SIGNATURE_PRUNING_RANGE → cutoff saturates to 0,
         // so nothing is old enough to prune even though slots are finalized.
-        let pruned = store.prune_old_block_signatures(9, 9).expect("prune");
-        assert_eq!(pruned, 0);
+        let pruned_below_slot = store.prune_old_block_signatures(9, 9).expect("prune");
+        assert_eq!(pruned_below_slot, 0);
         assert_eq!(count_entries(backend.as_ref(), Table::BlockSignatures), 10);
     }