diff --git a/crates/storage/src/api/traits.rs b/crates/storage/src/api/traits.rs index b1b15d4c..5b654bfc 100644 --- a/crates/storage/src/api/traits.rs +++ b/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>) -> 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) -> Result<(), Error>; } diff --git a/crates/storage/src/backend/in_memory.rs b/crates/storage/src/backend/in_memory.rs index b8af55d1..ec3fcd8f 100644 --- a/crates/storage/src/backend/in_memory.rs +++ b/crates/storage/src/backend/in_memory.rs @@ -8,13 +8,16 @@ use crate::api::{ type TableData = HashMap, Vec>; type StorageData = HashMap; -/// 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), - Delete, + Put(Vec, Vec), + Delete(Vec), + /// Delete every key in the half-open range `[from, to)`. + DeleteRange(Vec, Vec), } -type PendingOps = HashMap, 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, 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, Vec)>) -> 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>) -> 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) -> 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); } } } diff --git a/crates/storage/src/backend/rocksdb.rs b/crates/storage/src/backend/rocksdb.rs index 58f6119f..3440f94c 100644 --- a/crates/storage/src/backend/rocksdb.rs +++ b/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) -> Result<(), Error> { let mut write_opts = WriteOptions::default(); write_opts.set_sync(false); diff --git a/crates/storage/src/backend/tests.rs b/crates/storage/src/backend/tests.rs index f1a96b0a..46c90877 100644 --- a/crates/storage/src/backend/tests.rs +++ b/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 { diff --git a/crates/storage/src/store.rs b/crates/storage/src/store.rs index d5cac0ed..19193d94 100644 --- a/crates/storage/src/store.rs +++ b/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 { + ) -> Result { 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> = 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); }