feat(dash-spv): storage truncation primitives for reorg support

dash-spv/src/error.rs

@@ -80,6 +80,9 @@ pub enum StorageError {
     #[error("Read failed: {0}")]
     ReadFailed(String),
 
+    #[error("Invalid argument: {0}")]
+    InvalidArgument(String),
+
     #[error("IO error: {0}")]
     Io(#[from] io::Error),
 

dash-spv/src/storage/block_headers.rs

@@ -61,6 +61,11 @@ pub trait BlockHeaderStorage: Send + Sync + 'static {
         height: u32,
     ) -> StorageResult<()>;
 
+    /// Load a contiguous range of headers by height.
+    ///
+    /// Returns `StorageError::InvalidArgument` when the range extends into a
+    /// segment queued for deletion by a prior `truncate_above` (before the next
+    /// `persist`). Callers must clamp the range to at most `get_tip_height`.
     async fn load_headers(&self, range: Range<u32>) -> StorageResult<Vec<BlockHeader>>;
 
     async fn get_header(&self, height: u32) -> StorageResult<Option<BlockHeader>> {
@@ -95,6 +100,17 @@ pub trait BlockHeaderStorage: Send + Sync + 'static {
         &self,
         hash: &dashcore::BlockHash,
     ) -> StorageResult<Option<u32>>;
+
+    /// Drop all headers with `height > target_height`.
+    ///
+    /// Truncating above the current tip is a no-op, truncating below
+    /// `start_height` returns an error. Changes are applied in-memory and
+    /// flushed on the next `persist`.
+    ///
+    /// The truncation is not durable until the next successful `persist` call.
+    /// A crash between `truncate_above` and `persist` may leave orphaned segment
+    /// files on disk and cause the storage to reopen at the pre-truncation tip.
+    async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()>;
 }
 
 pub struct PersistentBlockHeaderStorage {
@@ -234,6 +250,17 @@ impl BlockHeaderStorage for PersistentBlockHeaderStorage {
     ) -> StorageResult<Option<u32>> {
         Ok(self.header_hash_index.get(hash).copied())
     }
+
+    async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()> {
+        let mut block_headers = self.block_headers.write().await;
+        let needs_index_prune = block_headers.tip_height().is_some_and(|tip| target_height < tip);
+        block_headers.truncate_above(target_height).await?;
+        drop(block_headers);
+        if needs_index_prune {
+            self.header_hash_index.retain(|_, h| *h <= target_height);
+        }
+        Ok(())
+    }
 }
 
 #[cfg(test)]
@@ -261,4 +288,61 @@ mod tests {
         assert_eq!(tip, expected_tip);
         assert_eq!(storage.get_tip_height().await, Some(4));
     }
+
+    #[tokio::test]
+    async fn test_truncate_above_drops_index_entries_and_allows_restore() {
+        let tmp_dir = TempDir::new().unwrap();
+        let mut storage = PersistentBlockHeaderStorage::open(tmp_dir.path()).await.unwrap();
+
+        let headers = BlockHeader::dummy_batch(0..10);
+        storage.store_headers(&headers).await.unwrap();
+
+        let orphaned_hash = headers[7].block_hash();
+        assert_eq!(storage.get_header_height_by_hash(&orphaned_hash).await.unwrap(), Some(7));
+
+        storage.truncate_above(5).await.unwrap();
+
+        assert_eq!(storage.get_tip_height().await, Some(5));
+        assert_eq!(storage.get_header_height_by_hash(&orphaned_hash).await.unwrap(), None);
+
+        let kept_hash = headers[3].block_hash();
+        assert_eq!(storage.get_header_height_by_hash(&kept_hash).await.unwrap(), Some(3));
+
+        let replacement = BlockHeader::dummy_batch(100..105);
+        storage.store_headers_at_height(&replacement, 6).await.unwrap();
+        assert_eq!(storage.get_tip_height().await, Some(10));
+
+        let reloaded = storage.load_headers(6..11).await.unwrap();
+        assert_eq!(reloaded, replacement);
+
+        let new_hash = replacement[0].block_hash();
+        assert_eq!(storage.get_header_height_by_hash(&new_hash).await.unwrap(), Some(6));
+
+        // Exercise the durability contract: persist, drop, reopen, and verify
+        // the rebuilt index does not resurrect orphaned hashes from stale files.
+        storage.persist(tmp_dir.path()).await.unwrap();
+        drop(storage);
+
+        let reopened = PersistentBlockHeaderStorage::open(tmp_dir.path()).await.unwrap();
+        assert_eq!(reopened.get_tip_height().await, Some(10));
+        assert_eq!(reopened.get_header_height_by_hash(&orphaned_hash).await.unwrap(), None);
+        assert_eq!(reopened.get_header_height_by_hash(&kept_hash).await.unwrap(), Some(3));
+        assert_eq!(reopened.get_header_height_by_hash(&new_hash).await.unwrap(), Some(6));
+    }
+
+    #[tokio::test]
+    async fn test_truncate_above_tip_is_noop_block_headers() {
+        let tmp_dir = TempDir::new().unwrap();
+        let mut storage = PersistentBlockHeaderStorage::open(tmp_dir.path()).await.unwrap();
+
+        let headers = BlockHeader::dummy_batch(0..5);
+        storage.store_headers(&headers).await.unwrap();
+
+        storage.truncate_above(100).await.unwrap();
+        assert_eq!(storage.get_tip_height().await, Some(4));
+
+        let still_indexed =
+            storage.get_header_height_by_hash(&headers[4].block_hash()).await.unwrap();
+        assert_eq!(still_indexed, Some(4));
+    }
 }

dash-spv/src/storage/blocks.rs

@@ -22,6 +22,17 @@ pub trait BlockStorage: Send + Sync + 'static {
 
     /// Load a single block by height.
     async fn load_block(&self, height: CoreBlockHeight) -> StorageResult<Option<HashedBlock>>;
+
+    /// Drop all blocks with `height > target_height`.
+    ///
+    /// Truncating above the current tip is a no-op, truncating below
+    /// `start_height` returns an error. Changes are applied in-memory and
+    /// flushed on the next `persist`.
+    ///
+    /// The truncation is not durable until the next successful `persist` call.
+    /// A crash between `truncate_above` and `persist` may leave orphaned segment
+    /// files on disk and cause the storage to reopen at the pre-truncation tip.
+    async fn truncate_above(&mut self, target_height: CoreBlockHeight) -> StorageResult<()>;
 }
 
 /// Persistent storage for full blocks using segmented files.
@@ -66,12 +77,20 @@ impl BlockStorage for PersistentBlockStorage {
     async fn load_block(&self, height: u32) -> StorageResult<Option<HashedBlock>> {
         self.blocks.write().await.get_item(height).await
     }
+
+    async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()> {
+        self.blocks.write().await.truncate_above(target_height).await
+    }
 }
 
 #[cfg(test)]
 mod tests {
-    use super::*;
+    use std::sync::Arc;
+
     use tempfile::TempDir;
+    use tokio::sync::Barrier;
+
+    use super::*;
 
     #[tokio::test]
     async fn test_store_and_load_block() {
@@ -112,6 +131,101 @@ mod tests {
         assert!(loaded.is_none());
     }
 
+    #[tokio::test]
+    async fn test_truncate_above_drops_blocks_and_allows_restore() {
+        let temp_dir = TempDir::new().unwrap();
+        let mut storage = PersistentBlockStorage::open(temp_dir.path()).await.unwrap();
+
+        for height in 100..110 {
+            storage.store_block(height, HashedBlock::dummy(height, vec![])).await.unwrap();
+        }
+
+        storage.truncate_above(104).await.unwrap();
+
+        assert_eq!(storage.load_block(104).await.unwrap(), Some(HashedBlock::dummy(104, vec![])));
+        for height in 105..110 {
+            assert_eq!(storage.load_block(height).await.unwrap(), None);
+        }
+
+        let replacement = HashedBlock::dummy(105, vec![]);
+        storage.store_block(105, replacement.clone()).await.unwrap();
+        assert_eq!(storage.load_block(105).await.unwrap(), Some(replacement));
+    }
+
+    #[tokio::test]
+    async fn test_truncate_above_persist_reopen_blocks() {
+        let temp_dir = TempDir::new().unwrap();
+        {
+            let mut storage = PersistentBlockStorage::open(temp_dir.path()).await.unwrap();
+            for height in 100..110 {
+                storage.store_block(height, HashedBlock::dummy(height, vec![])).await.unwrap();
+            }
+            storage.truncate_above(104).await.unwrap();
+            storage.persist(temp_dir.path()).await.unwrap();
+        }
+
+        let storage = PersistentBlockStorage::open(temp_dir.path()).await.unwrap();
+        assert_eq!(storage.load_block(104).await.unwrap(), Some(HashedBlock::dummy(104, vec![])));
+        for height in 105..110 {
+            assert_eq!(storage.load_block(height).await.unwrap(), None);
+        }
+    }
+
+    #[tokio::test]
+    async fn test_truncate_above_tip_noop_blocks() {
+        let temp_dir = TempDir::new().unwrap();
+        let mut storage = PersistentBlockStorage::open(temp_dir.path()).await.unwrap();
+
+        storage.store_block(50, HashedBlock::dummy(50, vec![])).await.unwrap();
+        storage.truncate_above(1_000).await.unwrap();
+        assert_eq!(storage.load_block(50).await.unwrap(), Some(HashedBlock::dummy(50, vec![])));
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn test_truncate_and_read_under_contention() {
+        let temp_dir = TempDir::new().unwrap();
+        let mut storage = PersistentBlockStorage::open(temp_dir.path()).await.unwrap();
+        for height in 0..20 {
+            storage.store_block(height, HashedBlock::dummy(height, vec![])).await.unwrap();
+        }
+
+        // The outer `RwLock` is required because `truncate_above` takes `&mut self`
+        // and so the writer and reader cannot literally execute in parallel, but a
+        // `Barrier` guarantees both tasks are scheduled and racing for the lock
+        // before either acquires it. Under the multi-thread runtime this exercises
+        // the read/write contention path while keeping the post-state assertions
+        // deterministic.
+        let shared = Arc::new(RwLock::new(storage));
+        let barrier = Arc::new(Barrier::new(2));
+
+        let reader = {
+            let shared = Arc::clone(&shared);
+            let barrier = Arc::clone(&barrier);
+            tokio::spawn(async move {
+                barrier.wait().await;
+                for _ in 0..50 {
+                    let _ = shared.read().await.load_block(5).await.unwrap();
+                }
+            })
+        };
+
+        let writer = {
+            let shared = Arc::clone(&shared);
+            let barrier = Arc::clone(&barrier);
+            tokio::spawn(async move {
+                barrier.wait().await;
+                shared.write().await.truncate_above(10).await.unwrap();
+            })
+        };
+
+        reader.await.unwrap();
+        writer.await.unwrap();
+
+        let guard = shared.read().await;
+        assert_eq!(guard.load_block(5).await.unwrap(), Some(HashedBlock::dummy(5, vec![])));
+        assert_eq!(guard.load_block(15).await.unwrap(), None);
+    }
+
     #[tokio::test]
     async fn test_returns_none_for_gaps() {
         let temp_dir = TempDir::new().unwrap();

dash-spv/src/storage/filter_headers.rs

@@ -17,6 +17,11 @@ pub trait FilterHeaderStorage: Send + Sync + 'static {
         height: u32,
     ) -> StorageResult<()>;
 
+    /// Load a contiguous range of filter headers by height.
+    ///
+    /// Returns `StorageError::InvalidArgument` when the range extends into a
+    /// segment queued for deletion by a prior `truncate_above` (before the next
+    /// `persist`). Callers must clamp the range to at most `get_filter_tip_height`.
     async fn load_filter_headers(&self, range: Range<u32>) -> StorageResult<Vec<FilterHeader>>;
 
     async fn get_filter_header(&self, height: u32) -> StorageResult<Option<FilterHeader>> {
@@ -42,6 +47,17 @@ pub trait FilterHeaderStorage: Send + Sync + 'static {
     async fn get_filter_tip_height(&self) -> StorageResult<Option<u32>>;
 
     async fn get_filter_start_height(&self) -> Option<u32>;
+
+    /// Drop all filter headers with `height > target_height`.
+    ///
+    /// Truncating above the current tip is a no-op, truncating below
+    /// `start_height` returns an error. Changes are applied in-memory and
+    /// flushed on the next `persist`.
+    ///
+    /// The truncation is not durable until the next successful `persist` call.
+    /// A crash between `truncate_above` and `persist` may leave orphaned segment
+    /// files on disk and cause the storage to reopen at the pre-truncation tip.
+    async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()>;
 }
 
 pub struct PersistentFilterHeaderStorage {
@@ -100,4 +116,69 @@ impl FilterHeaderStorage for PersistentFilterHeaderStorage {
     async fn get_filter_start_height(&self) -> Option<u32> {
         self.filter_headers.read().await.start_height()
     }
+
+    async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()> {
+        self.filter_headers.write().await.truncate_above(target_height).await
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tempfile::TempDir;
+
+    #[tokio::test]
+    async fn test_truncate_above_and_restore() {
+        let tmp_dir = TempDir::new().unwrap();
+        let mut storage = PersistentFilterHeaderStorage::open(tmp_dir.path()).await.unwrap();
+
+        let headers = FilterHeader::dummy_batch(0..10);
+        storage.store_filter_headers(&headers).await.unwrap();
+        assert_eq!(storage.get_filter_tip_height().await.unwrap(), Some(9));
+
+        storage.truncate_above(4).await.unwrap();
+        assert_eq!(storage.get_filter_tip_height().await.unwrap(), Some(4));
+
+        let kept = storage.load_filter_headers(0..5).await.unwrap();
+        assert_eq!(kept, headers[0..5]);
+
+        let replacement = FilterHeader::dummy_batch(100..105);
+        storage.store_filter_headers_at_height(&replacement, 5).await.unwrap();
+        assert_eq!(storage.get_filter_tip_height().await.unwrap(), Some(9));
+
+        let reloaded = storage.load_filter_headers(5..10).await.unwrap();
+        assert_eq!(reloaded, replacement);
+    }
+
+    #[tokio::test]
+    async fn test_truncate_above_persist_reopen_filter_headers() {
+        let tmp_dir = TempDir::new().unwrap();
+        {
+            let mut storage = PersistentFilterHeaderStorage::open(tmp_dir.path()).await.unwrap();
+            let headers = FilterHeader::dummy_batch(0..10);
+            storage.store_filter_headers(&headers).await.unwrap();
+            storage.truncate_above(4).await.unwrap();
+            storage.persist(tmp_dir.path()).await.unwrap();
+        }
+
+        let storage = PersistentFilterHeaderStorage::open(tmp_dir.path()).await.unwrap();
+        assert_eq!(storage.get_filter_tip_height().await.unwrap(), Some(4));
+        let kept = storage.load_filter_headers(0..5).await.unwrap();
+        assert_eq!(kept, FilterHeader::dummy_batch(0..5));
+        for h in 5..10 {
+            assert_eq!(storage.get_filter_header(h).await.unwrap(), None);
+        }
+    }
+
+    #[tokio::test]
+    async fn test_truncate_above_tip_noop() {
+        let tmp_dir = TempDir::new().unwrap();
+        let mut storage = PersistentFilterHeaderStorage::open(tmp_dir.path()).await.unwrap();
+
+        let headers = FilterHeader::dummy_batch(0..5);
+        storage.store_filter_headers(&headers).await.unwrap();
+
+        storage.truncate_above(100).await.unwrap();
+        assert_eq!(storage.get_filter_tip_height().await.unwrap(), Some(4));
+    }
 }