diff --git a/dash-spv/src/error.rs b/dash-spv/src/error.rs
index 76645a06b..4edee6019 100644
--- a/dash-spv/src/error.rs
+++ b/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),
 
diff --git a/dash-spv/src/storage/block_headers.rs b/dash-spv/src/storage/block_headers.rs
index f75332f3a..e7c07b36d 100644
--- a/dash-spv/src/storage/block_headers.rs
+++ b/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,52 @@ 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(100).await.unwrap();
+        assert_eq!(storage.get_tip_height().await, Some(9));
+        assert_eq!(
+            storage.get_header_height_by_hash(&headers[4].block_hash()).await.unwrap(),
+            Some(4)
+        );
+
+        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));
+    }
 }
diff --git a/dash-spv/src/storage/blocks.rs b/dash-spv/src/storage/blocks.rs
index ad4981e95..27324d5c9 100644
--- a/dash-spv/src/storage/blocks.rs
+++ b/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,13 +77,18 @@ 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 tempfile::TempDir;
 
+    use super::*;
+
     #[tokio::test]
     async fn test_store_and_load_block() {
         let temp_dir = TempDir::new().unwrap();
@@ -112,6 +128,21 @@ mod tests {
         assert!(loaded.is_none());
     }
 
+    #[tokio::test]
+    async fn test_truncate_above_wrapper_smoke() {
+        let temp_dir = TempDir::new().unwrap();
+        let mut storage = PersistentBlockStorage::open(temp_dir.path()).await.unwrap();
+
+        for height in 0..5 {
+            storage.store_block(height, HashedBlock::dummy(height, vec![])).await.unwrap();
+        }
+
+        storage.truncate_above(2).await.unwrap();
+
+        assert_eq!(storage.load_block(2).await.unwrap(), Some(HashedBlock::dummy(2, vec![])));
+        assert_eq!(storage.load_block(3).await.unwrap(), None);
+    }
+
     #[tokio::test]
     async fn test_returns_none_for_gaps() {
         let temp_dir = TempDir::new().unwrap();
diff --git a/dash-spv/src/storage/filter_headers.rs b/dash-spv/src/storage/filter_headers.rs
index da077871c..869a82f16 100644
--- a/dash-spv/src/storage/filter_headers.rs
+++ b/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,28 @@ 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_wrapper_smoke() {
+        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(2).await.unwrap();
+
+        assert_eq!(storage.get_filter_tip_height().await.unwrap(), Some(2));
+        assert_eq!(storage.get_filter_header(3).await.unwrap(), None);
+    }
 }
diff --git a/dash-spv/src/storage/filters.rs b/dash-spv/src/storage/filters.rs
index 33bab100d..37a0cd827 100644
--- a/dash-spv/src/storage/filters.rs
+++ b/dash-spv/src/storage/filters.rs
@@ -12,9 +12,25 @@ use crate::{
 pub trait FilterStorage: Send + Sync + 'static {
     async fn store_filter(&mut self, height: u32, filter: &[u8]) -> StorageResult<()>;
 
+    /// Load a contiguous range of filters 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 `filter_tip_height`.
     async fn load_filters(&self, range: Range<u32>) -> StorageResult<Vec<Vec<u8>>>;
 
     async fn filter_tip_height(&self) -> StorageResult<u32>;
+
+    /// Drop all filters 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 PersistentFilterStorage {
@@ -62,4 +78,33 @@ impl FilterStorage for PersistentFilterStorage {
     async fn filter_tip_height(&self) -> StorageResult<u32> {
         Ok(self.filters.read().await.tip_height().unwrap_or(0))
     }
+
+    async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()> {
+        self.filters.write().await.truncate_above(target_height).await
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tempfile::TempDir;
+
+    fn filter_bytes(seed: u8) -> Vec<u8> {
+        vec![seed; 8]
+    }
+
+    #[tokio::test]
+    async fn test_truncate_above_wrapper_smoke() {
+        let tmp_dir = TempDir::new().unwrap();
+        let mut storage = PersistentFilterStorage::open(tmp_dir.path()).await.unwrap();
+
+        for height in 0..5 {
+            storage.store_filter(height, &filter_bytes(height as u8)).await.unwrap();
+        }
+
+        storage.truncate_above(2).await.unwrap();
+
+        assert_eq!(storage.filter_tip_height().await.unwrap(), 2);
+        assert!(storage.load_filters(3..4).await.is_err());
+    }
 }
diff --git a/dash-spv/src/storage/mod.rs b/dash-spv/src/storage/mod.rs
index 208ba977b..57e4fb736 100644
--- a/dash-spv/src/storage/mod.rs
+++ b/dash-spv/src/storage/mod.rs
@@ -319,6 +319,10 @@ impl BlockHeaderStorage for DiskStorageManager {
     ) -> StorageResult<Option<u32>> {
         self.block_headers.read().await.get_header_height_by_hash(hash).await
     }
+
+    async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()> {
+        self.block_headers.write().await.truncate_above(target_height).await
+    }
 }
 
 #[async_trait]
@@ -346,6 +350,10 @@ impl FilterHeaderStorage for DiskStorageManager {
     async fn get_filter_start_height(&self) -> Option<u32> {
         self.filter_headers.read().await.get_filter_start_height().await
     }
+
+    async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()> {
+        self.filter_headers.write().await.truncate_above(target_height).await
+    }
 }
 
 #[async_trait]
@@ -361,6 +369,10 @@ impl filters::FilterStorage for DiskStorageManager {
     async fn filter_tip_height(&self) -> StorageResult<u32> {
         self.filters.read().await.filter_tip_height().await
     }
+
+    async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()> {
+        self.filters.write().await.truncate_above(target_height).await
+    }
 }
 
 #[async_trait]
@@ -372,6 +384,10 @@ impl BlockStorage for DiskStorageManager {
     async fn load_block(&self, height: u32) -> StorageResult<Option<HashedBlock>> {
         self.blocks.read().await.load_block(height).await
     }
+
+    async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()> {
+        self.blocks.write().await.truncate_above(target_height).await
+    }
 }
 
 #[async_trait]
@@ -568,6 +584,27 @@ mod tests {
         assert!(storage.get_header_height_by_hash(&hash).await.unwrap().is_none());
     }
 
+    #[tokio::test]
+    async fn test_disk_storage_manager_truncate_above_block_headers() {
+        let temp_dir = TempDir::new().unwrap();
+        let config = ClientConfig::regtest().with_storage_path(temp_dir.path());
+        let mut mgr = DiskStorageManager::new(&config).await.unwrap();
+
+        let headers = BlockHeader::dummy_batch(0..10);
+        mgr.store_headers(&headers).await.unwrap();
+
+        let orphaned_hash = headers[7].block_hash();
+        assert_eq!(mgr.get_header_height_by_hash(&orphaned_hash).await.unwrap(), Some(7));
+
+        <DiskStorageManager as BlockHeaderStorage>::truncate_above(&mut mgr, 5).await.unwrap();
+
+        assert_eq!(mgr.get_tip_height().await, Some(5));
+        assert_eq!(mgr.get_header_height_by_hash(&orphaned_hash).await.unwrap(), None);
+
+        let kept_hash = headers[3].block_hash();
+        assert_eq!(mgr.get_header_height_by_hash(&kept_hash).await.unwrap(), Some(3));
+    }
+
     #[tokio::test]
     async fn test_lock_lifecycle() {
         let temp_dir = TempDir::new().expect("Failed to create temp directory");
diff --git a/dash-spv/src/storage/segments.rs b/dash-spv/src/storage/segments.rs
index f3247b56d..49f291add 100644
--- a/dash-spv/src/storage/segments.rs
+++ b/dash-spv/src/storage/segments.rs
@@ -1,7 +1,7 @@
 //! Segment management and persistence for items implementing the Persistable trait.
 
 use std::{
-    collections::HashMap,
+    collections::{HashMap, HashSet},
     fs::{self, File},
     io::BufReader,
     ops::Range,
@@ -80,6 +80,9 @@ pub struct SegmentCache<I: Persistable> {
     tip_height: Option<u32>,
     start_height: Option<u32>,
     segments_dir: PathBuf,
+    /// Segment ids whose backing files must be removed on the next `persist`.
+    /// Populated by `truncate_above` for segments that are dropped entirely.
+    to_delete: HashSet<u32>,
 }
 
 impl<I: Persistable> SegmentCache<I> {
@@ -94,6 +97,7 @@ impl<I: Persistable> SegmentCache<I> {
             tip_height: None,
             start_height: None,
             segments_dir: segments_dir.clone(),
+            to_delete: HashSet::new(),
         };
 
         // Building the metadata
@@ -184,9 +188,13 @@ impl<I: Persistable> SegmentCache<I> {
         }
 
         // If the segment is already in the to_persist map, load it from there.
-        // If the segment is not in the to_persist map, load it from disk.
+        // If the segment is queued for deletion, return a fresh empty segment.
+        // The next `persist` will atomically overwrite the stale file.
+        // Otherwise, load it from disk.
         let segment = if let Some(segment) = self.evicted.remove(segment_id) {
             segment
+        } else if self.to_delete.remove(segment_id) {
+            Segment::new(*segment_id, vec![], SegmentState::Dirty)
         } else {
             Segment::load(&self.segments_dir, *segment_id).await?
         };
@@ -195,12 +203,29 @@ impl<I: Persistable> SegmentCache<I> {
         Ok(segment)
     }
 
+    /// Load a contiguous range of items by height.
+    ///
+    /// Returns `StorageError::InvalidArgument` when the requested range extends
+    /// into a segment queued for deletion by a prior `truncate_above` (before
+    /// the next `persist`). Callers reading across a truncation boundary must
+    /// clamp their range to `tip_height` first, or fall back to per-item reads
+    /// via `get_item`, which returns `Ok(None)` for those slots.
     pub async fn get_items(&mut self, height_range: Range<u32>) -> StorageResult<Vec<I>> {
         debug_assert!(height_range.start < height_range.end);
 
         let start = height_range.start;
         let end = height_range.end;
 
+        // Reject ranges that extend past the current tip. After a within-segment
+        // `truncate_above`, the boundary segment is not in `to_delete` so the
+        // loop guard below cannot catch overruns into its sentinel tail.
+        if end > self.next_height() {
+            return Err(StorageError::InvalidArgument(format!(
+                "get_items range {height_range:?} extends above tip {:?}",
+                self.tip_height
+            )));
+        }
+
         let mut items = Vec::with_capacity((end - start) as usize);
 
         let start_segment = Self::height_to_segment_id(start);
@@ -215,6 +240,16 @@ impl<I: Persistable> SegmentCache<I> {
         let end_segment = Self::height_to_segment_id(end - 1);
 
         for segment_id in start_segment..=end_segment {
+            // Same guard as `get_item`: a segment queued for deletion holds
+            // nothing but sentinels by construction. Loading it via
+            // `get_segment_mut` would consume the deletion intent and queue
+            // an all-sentinel rewrite, so refuse the read instead.
+            if self.to_delete.contains(&segment_id) {
+                return Err(StorageError::InvalidArgument(format!(
+                    "get_items range {height_range:?} extends into segment {segment_id} queued for deletion"
+                )));
+            }
+
             let segment = self.get_segment_mut(&segment_id).await?;
 
             let seg_start = if segment_id == start_segment {
@@ -261,9 +296,24 @@ impl<I: Persistable> SegmentCache<I> {
     }
 
     /// Get a single item by height. Returns `None` for sentinel (empty) slots.
-    /// Unlike `get_items()`, this does not assert dense storage — safe for sparse data.
+    /// Unlike `get_items()`, this does not assert dense storage, safe for sparse data.
+    ///
+    /// For heights in a segment queued for deletion by a prior `truncate_above`
+    /// (before the next `persist`), this returns `Ok(None)` rather than the
+    /// `StorageError::InvalidArgument` that `get_items` returns. Callers must
+    /// not interpret `Ok(None)` here as a fallback for an error from `get_items`.
+    /// The two APIs report truncated slots differently by design.
     pub async fn get_item(&mut self, height: u32) -> StorageResult<Option<I>> {
         let segment_id = Self::height_to_segment_id(height);
+
+        // A segment queued for deletion holds nothing but sentinels by
+        // construction. Short-circuit so a read above the truncated tip does
+        // not consume the deletion intent in `get_segment_mut` and replace
+        // the stale on-disk file with a fresh all-sentinel blob.
+        if self.to_delete.contains(&segment_id) {
+            return Ok(None);
+        }
+
         let offset = Self::height_to_offset(height);
         let segment = self.get_segment_mut(&segment_id).await?;
         let item = segment.get_single(offset);
@@ -322,9 +372,80 @@ impl<I: Persistable> SegmentCache<I> {
         Ok(())
     }
 
+    /// Truncate the cache so that no items above `target_height` remain.
+    ///
+    /// Segments entirely above the target are dropped from memory and queued for
+    /// deletion on the next `persist`. The segment containing `target_height + 1`
+    /// has its tail slots reset to `I::sentinel()` so subsequent `Segment::insert`
+    /// calls into the same range remain sound.
+    ///
+    /// Returns an error if `target_height` is below `start_height`, since the
+    /// resulting cache would have a hole below its origin. Callers must guard
+    /// against truncating an empty cache except as a no-op (no error).
+    ///
+    /// 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, causing `tip_height` to be recomputed from stale data on
+    /// restart and the cache to reopen at the pre-truncation tip.
+    pub async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()> {
+        let tip = match self.tip_height {
+            Some(tip) => tip,
+            None => return Ok(()),
+        };
+
+        if target_height >= tip {
+            return Ok(());
+        }
+
+        if let Some(start) = self.start_height {
+            if target_height < start {
+                return Err(StorageError::InvalidArgument(format!(
+                    "truncate_above({target_height}) below start_height ({start})"
+                )));
+            }
+        }
+
+        let boundary_segment_id = Self::height_to_segment_id(target_height);
+        let boundary_offset = Self::height_to_offset(target_height);
+        let max_segment_id = Self::height_to_segment_id(tip);
+
+        // Load the boundary segment first so any disk I/O error is surfaced
+        // before mutating cache state. After this point only infallible
+        // in-memory operations run, so the function cannot leave the cache
+        // in a half-truncated state.
+        if boundary_offset + 1 < Segment::<I>::ITEMS_PER_SEGMENT {
+            let segment = self.get_segment_mut(&boundary_segment_id).await?;
+            segment.reset_above(boundary_offset);
+        }
+
+        for segment_id in (boundary_segment_id + 1)..=max_segment_id {
+            self.segments.remove(&segment_id);
+            self.evicted.remove(&segment_id);
+            self.to_delete.insert(segment_id);
+        }
+
+        self.tip_height = Some(target_height);
+
+        Ok(())
+    }
+
     pub async fn persist(&mut self, segments_dir: impl Into<PathBuf>) {
         let segments_dir = segments_dir.into();
 
+        let mut failed = HashSet::new();
+        for id in self.to_delete.drain() {
+            let path = segments_dir.join(I::segment_file_name(id));
+            match tokio::fs::remove_file(&path).await {
+                Ok(_) => {}
+                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
+                Err(e) => {
+                    tracing::error!("Failed to delete segment file {:?}: {}", path, e);
+                    failed.insert(id);
+                }
+            }
+        }
+        self.to_delete.extend(failed);
+
         for (id, segments) in self.evicted.iter_mut() {
             if let Err(e) = segments.persist(&segments_dir).await {
                 tracing::error!("Failed to persist segment with id {id}: {e}");
@@ -476,6 +597,29 @@ impl<I: Persistable> Segment<I> {
         Ok(())
     }
 
+    /// Reset all slots strictly above `offset` to the sentinel value.
+    /// The slot at `offset` is preserved. Marks the segment dirty if any
+    /// slot was changed.
+    fn reset_above(&mut self, offset: u32) {
+        debug_assert!(offset < Self::ITEMS_PER_SEGMENT);
+
+        let sentinel = I::sentinel();
+        let start = (offset as usize) + 1;
+        let mut changed = false;
+
+        for slot in &mut self.items[start..] {
+            if *slot != sentinel {
+                *slot = sentinel.clone();
+                changed = true;
+            }
+        }
+
+        if changed {
+            self.state = SegmentState::Dirty;
+            self.last_accessed = Instant::now();
+        }
+    }
+
     pub fn insert(&mut self, item: I, offset: u32) {
         debug_assert!(offset < Self::ITEMS_PER_SEGMENT);
 
@@ -654,6 +798,300 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_truncate_above_within_segment() {
+        let tmp_dir = TempDir::new().unwrap();
+
+        let items = FilterHeader::dummy_batch(0..20);
+
+        let mut cache = SegmentCache::<FilterHeader>::load_or_new(tmp_dir.path()).await.unwrap();
+        cache.store_items_at_height(&items, 0).await.unwrap();
+        assert_eq!(cache.tip_height(), Some(19));
+
+        cache.truncate_above(9).await.unwrap();
+        assert_eq!(cache.tip_height(), Some(9));
+        assert_eq!(cache.start_height(), Some(0));
+
+        let kept = cache.get_items(0..10).await.unwrap();
+        assert_eq!(kept, items[0..10]);
+
+        // Re-store into the truncated range — must not panic on the sentinel debug_assert.
+        let replacement = FilterHeader::dummy_batch(100..110);
+        cache.store_items_at_height(&replacement, 10).await.unwrap();
+        assert_eq!(cache.tip_height(), Some(19));
+
+        let reread = cache.get_items(10..20).await.unwrap();
+        assert_eq!(reread, replacement);
+    }
+
+    #[tokio::test]
+    async fn test_truncate_within_segment_persist_reload() {
+        let tmp_dir = TempDir::new().unwrap();
+
+        let items = FilterHeader::dummy_batch(0..20);
+
+        let mut cache = SegmentCache::<FilterHeader>::load_or_new(tmp_dir.path()).await.unwrap();
+        cache.store_items_at_height(&items, 0).await.unwrap();
+        cache.persist(tmp_dir.path()).await;
+
+        let mut cache = SegmentCache::<FilterHeader>::load_or_new(tmp_dir.path()).await.unwrap();
+        cache.truncate_above(9).await.unwrap();
+        cache.persist(tmp_dir.path()).await;
+
+        // Reopen without re-storing to verify the boundary segment was rewritten
+        // with sentinel slots above the cut. If `reset_above`'s dirty flag were
+        // skipped, the stale on-disk slots would survive and `tip_height` would
+        // come back as 19.
+        let mut reloaded = SegmentCache::<FilterHeader>::load_or_new(tmp_dir.path()).await.unwrap();
+        assert_eq!(reloaded.tip_height(), Some(9));
+        assert_eq!(reloaded.get_items(0..10).await.unwrap(), items[0..10]);
+        for height in 10..20u32 {
+            assert_eq!(reloaded.get_item(height).await.unwrap(), None);
+        }
+    }
+
+    #[tokio::test]
+    async fn test_truncate_above_segment_boundary() {
+        let tmp_dir = TempDir::new().unwrap();
+
+        const ITEMS_PER_SEGMENT: u32 = Segment::<FilterHeader>::ITEMS_PER_SEGMENT;
+
+        let items = FilterHeader::dummy_batch(0..ITEMS_PER_SEGMENT + 5);
+
+        let mut cache = SegmentCache::<FilterHeader>::load_or_new(tmp_dir.path()).await.unwrap();
+        cache.store_items_at_height(&items, 0).await.unwrap();
+        assert_eq!(cache.tip_height(), Some(ITEMS_PER_SEGMENT + 4));
+
+        // Persist first so segment 1's file is actually on disk, otherwise the
+        // post-truncate `!exists()` check below would be trivially true.
+        cache.persist(tmp_dir.path()).await;
+        let dropped_segment_file = tmp_dir.path().join(FilterHeader::segment_file_name(1));
+        assert!(dropped_segment_file.exists());
+
+        cache.truncate_above(ITEMS_PER_SEGMENT - 1).await.unwrap();
+        assert_eq!(cache.tip_height(), Some(ITEMS_PER_SEGMENT - 1));
+
+        let kept = cache.get_items(0..ITEMS_PER_SEGMENT).await.unwrap();
+        assert_eq!(kept, items[0..ITEMS_PER_SEGMENT as usize]);
+
+        cache.persist(tmp_dir.path()).await;
+        assert!(!dropped_segment_file.exists());
+
+        // Reload and verify the truncation is durable.
+        let mut reloaded = SegmentCache::<FilterHeader>::load_or_new(tmp_dir.path()).await.unwrap();
+        assert_eq!(reloaded.tip_height(), Some(ITEMS_PER_SEGMENT - 1));
+
+        // Re-storing into the dropped segment is sound.
+        let new_items = FilterHeader::dummy_batch(500..505);
+        reloaded.store_items_at_height(&new_items, ITEMS_PER_SEGMENT).await.unwrap();
+        assert_eq!(reloaded.tip_height(), Some(ITEMS_PER_SEGMENT + 4));
+        let reread = reloaded.get_items(ITEMS_PER_SEGMENT..ITEMS_PER_SEGMENT + 5).await.unwrap();
+        assert_eq!(reread, new_items);
+    }
+
+    #[tokio::test]
+    async fn test_truncate_above_then_store_into_dropped_segment_without_persist() {
+        let tmp_dir = TempDir::new().unwrap();
+
+        const ITEMS_PER_SEGMENT: u32 = Segment::<FilterHeader>::ITEMS_PER_SEGMENT;
+
+        let items = FilterHeader::dummy_batch(0..ITEMS_PER_SEGMENT + 5);
+
+        let mut cache = SegmentCache::<FilterHeader>::load_or_new(tmp_dir.path()).await.unwrap();
+        cache.store_items_at_height(&items, 0).await.unwrap();
+        cache.persist(tmp_dir.path()).await;
+
+        let mut cache = SegmentCache::<FilterHeader>::load_or_new(tmp_dir.path()).await.unwrap();
+        cache.truncate_above(ITEMS_PER_SEGMENT - 1).await.unwrap();
+
+        // Re-store into the dropped segment BEFORE persist runs. Without the
+        // to_delete check in get_segment_mut, the stale on-disk file would be
+        // loaded and the insert would hit the sentinel debug_assert.
+        let replacement = FilterHeader::dummy_batch(500..505);
+        cache.store_items_at_height(&replacement, ITEMS_PER_SEGMENT).await.unwrap();
+        assert_eq!(cache.tip_height(), Some(ITEMS_PER_SEGMENT + 4));
+
+        let reread = cache.get_items(ITEMS_PER_SEGMENT..ITEMS_PER_SEGMENT + 5).await.unwrap();
+        assert_eq!(reread, replacement);
+
+        cache.persist(tmp_dir.path()).await;
+        let mut reloaded = SegmentCache::<FilterHeader>::load_or_new(tmp_dir.path()).await.unwrap();
+        assert_eq!(reloaded.tip_height(), Some(ITEMS_PER_SEGMENT + 4));
+        let reread = reloaded.get_items(ITEMS_PER_SEGMENT..ITEMS_PER_SEGMENT + 5).await.unwrap();
+        assert_eq!(reread, replacement);
+    }
+
+    #[tokio::test]
+    async fn test_read_above_truncated_tip_preserves_deletion() {
+        let tmp_dir = TempDir::new().unwrap();
+
+        const ITEMS_PER_SEGMENT: u32 = Segment::<FilterHeader>::ITEMS_PER_SEGMENT;
+
+        let items = FilterHeader::dummy_batch(0..ITEMS_PER_SEGMENT + 5);
+
+        let mut cache = SegmentCache::<FilterHeader>::load_or_new(tmp_dir.path()).await.unwrap();
+        cache.store_items_at_height(&items, 0).await.unwrap();
+        cache.persist(tmp_dir.path()).await;
+
+        let dropped_segment_file = tmp_dir.path().join(FilterHeader::segment_file_name(1));
+        assert!(dropped_segment_file.exists());
+
+        let mut cache = SegmentCache::<FilterHeader>::load_or_new(tmp_dir.path()).await.unwrap();
+        cache.truncate_above(ITEMS_PER_SEGMENT - 1).await.unwrap();
+
+        // Reading above the truncated tip must not cancel the pending deletion.
+        // Without the to_delete short-circuit in get_item, this read would
+        // remove the entry from to_delete and queue an all-sentinel rewrite.
+        assert_eq!(cache.get_item(ITEMS_PER_SEGMENT + 2).await.unwrap(), None);
+        assert_eq!(cache.get_item(ITEMS_PER_SEGMENT).await.unwrap(), None);
+
+        cache.persist(tmp_dir.path()).await;
+        assert!(!dropped_segment_file.exists());
+    }
+
+    #[tokio::test]
+    async fn test_get_items_above_truncated_tip_preserves_deletion() {
+        let tmp_dir = TempDir::new().unwrap();
+
+        const ITEMS_PER_SEGMENT: u32 = Segment::<FilterHeader>::ITEMS_PER_SEGMENT;
+
+        let items = FilterHeader::dummy_batch(0..ITEMS_PER_SEGMENT + 5);
+
+        let mut cache = SegmentCache::<FilterHeader>::load_or_new(tmp_dir.path()).await.unwrap();
+        cache.store_items_at_height(&items, 0).await.unwrap();
+        cache.persist(tmp_dir.path()).await;
+
+        let dropped_segment_file = tmp_dir.path().join(FilterHeader::segment_file_name(1));
+        assert!(dropped_segment_file.exists());
+
+        let mut cache = SegmentCache::<FilterHeader>::load_or_new(tmp_dir.path()).await.unwrap();
+        cache.truncate_above(ITEMS_PER_SEGMENT - 1).await.unwrap();
+
+        // A range read that spans into a segment queued for deletion must
+        // error rather than consuming the deletion intent and handing back
+        // sentinels.
+        assert!(matches!(
+            cache.get_items(ITEMS_PER_SEGMENT..ITEMS_PER_SEGMENT + 5).await,
+            Err(StorageError::InvalidArgument(_))
+        ));
+
+        cache.persist(tmp_dir.path()).await;
+        assert!(!dropped_segment_file.exists());
+    }
+
+    #[tokio::test]
+    async fn test_get_items_above_truncated_tip_within_segment_errors() {
+        let tmp_dir = TempDir::new().unwrap();
+
+        let items = FilterHeader::dummy_batch(0..20);
+
+        let mut cache = SegmentCache::<FilterHeader>::load_or_new(tmp_dir.path()).await.unwrap();
+        cache.store_items_at_height(&items, 0).await.unwrap();
+        cache.truncate_above(9).await.unwrap();
+
+        // The boundary segment is not in `to_delete`, but reading past the
+        // truncated tip would land in its sentinel tail. Fail fast instead of
+        // panicking on the `last_valid_offset` debug_assert.
+        assert!(matches!(cache.get_items(0..15).await, Err(StorageError::InvalidArgument(_))));
+        assert!(matches!(cache.get_items(8..11).await, Err(StorageError::InvalidArgument(_))));
+
+        let kept = cache.get_items(0..10).await.unwrap();
+        assert_eq!(kept, items[0..10]);
+    }
+
+    #[tokio::test]
+    async fn test_truncate_above_noop_cases() {
+        let tmp_dir = TempDir::new().unwrap();
+
+        let mut cache = SegmentCache::<FilterHeader>::load_or_new(tmp_dir.path()).await.unwrap();
+
+        cache.truncate_above(0).await.unwrap();
+        cache.truncate_above(100).await.unwrap();
+        assert_eq!(cache.tip_height(), None);
+
+        let items = FilterHeader::dummy_batch(0..10);
+        cache.store_items_at_height(&items, 0).await.unwrap();
+
+        cache.truncate_above(9).await.unwrap();
+        assert_eq!(cache.tip_height(), Some(9));
+
+        cache.truncate_above(100).await.unwrap();
+        assert_eq!(cache.tip_height(), Some(9));
+
+        let kept = cache.get_items(0..10).await.unwrap();
+        assert_eq!(kept, items);
+    }
+
+    #[tokio::test]
+    async fn test_truncate_below_start_errors() {
+        let tmp_dir = TempDir::new().unwrap();
+
+        let items = FilterHeader::dummy_batch(0..5);
+
+        let mut cache = SegmentCache::<FilterHeader>::load_or_new(tmp_dir.path()).await.unwrap();
+        cache.store_items_at_height(&items, 10).await.unwrap();
+        assert_eq!(cache.start_height(), Some(10));
+
+        assert!(cache.truncate_above(5).await.is_err());
+        assert_eq!(cache.tip_height(), Some(14));
+        assert_eq!(cache.start_height(), Some(10));
+    }
+
+    #[tokio::test]
+    async fn test_sequential_truncations_within_same_segment() {
+        let tmp_dir = TempDir::new().unwrap();
+        let items = FilterHeader::dummy_batch(0..30);
+
+        let mut cache = SegmentCache::<FilterHeader>::load_or_new(tmp_dir.path()).await.unwrap();
+        cache.store_items_at_height(&items, 0).await.unwrap();
+
+        cache.truncate_above(20).await.unwrap();
+        assert_eq!(cache.tip_height(), Some(20));
+
+        cache.truncate_above(10).await.unwrap();
+        assert_eq!(cache.tip_height(), Some(10));
+        assert_eq!(cache.start_height(), Some(0));
+
+        // Re-store into the doubly-truncated range. Both `reset_above` passes
+        // must have left the slots as sentinels for `insert` to remain sound.
+        let replacement = FilterHeader::dummy_batch(200..220);
+        cache.store_items_at_height(&replacement, 11).await.unwrap();
+        assert_eq!(cache.tip_height(), Some(30));
+        assert_eq!(cache.get_items(11..31).await.unwrap(), replacement);
+    }
+
+    #[tokio::test]
+    async fn test_sequential_truncations_across_segment_boundaries() {
+        let tmp_dir = TempDir::new().unwrap();
+
+        const ITEMS_PER_SEGMENT: u32 = Segment::<FilterHeader>::ITEMS_PER_SEGMENT;
+
+        let items = FilterHeader::dummy_batch(0..ITEMS_PER_SEGMENT * 3);
+
+        let mut cache = SegmentCache::<FilterHeader>::load_or_new(tmp_dir.path()).await.unwrap();
+        cache.store_items_at_height(&items, 0).await.unwrap();
+        cache.persist(tmp_dir.path()).await;
+
+        let mut cache = SegmentCache::<FilterHeader>::load_or_new(tmp_dir.path()).await.unwrap();
+        cache.truncate_above(ITEMS_PER_SEGMENT * 2 + 5).await.unwrap();
+        cache.truncate_above(ITEMS_PER_SEGMENT + 5).await.unwrap();
+        cache.persist(tmp_dir.path()).await;
+
+        // Segment 2 must be gone on disk (dropped by the second cut),
+        // segment 1 stays (its tail was reset by `reset_above`).
+        assert!(!tmp_dir.path().join(FilterHeader::segment_file_name(2)).exists());
+        assert!(tmp_dir.path().join(FilterHeader::segment_file_name(1)).exists());
+
+        let mut reloaded = SegmentCache::<FilterHeader>::load_or_new(tmp_dir.path()).await.unwrap();
+        assert_eq!(reloaded.tip_height(), Some(ITEMS_PER_SEGMENT + 5));
+        let replacement = FilterHeader::dummy_batch(500..510);
+        reloaded.store_items_at_height(&replacement, ITEMS_PER_SEGMENT + 6).await.unwrap();
+        assert_eq!(
+            reloaded.get_items(ITEMS_PER_SEGMENT + 6..ITEMS_PER_SEGMENT + 16).await.unwrap(),
+            replacement
+        );
+    }
+
     #[test]
     fn test_segment_insert_get() {
         let segment_id = 10;
diff --git a/dash-spv/src/sync/masternodes/sync_manager.rs b/dash-spv/src/sync/masternodes/sync_manager.rs
index e72652fa1..120bb0d78 100644
--- a/dash-spv/src/sync/masternodes/sync_manager.rs
+++ b/dash-spv/src/sync/masternodes/sync_manager.rs
@@ -710,6 +710,10 @@ mod tests {
         async fn get_header_height_by_hash(&self, hash: &BlockHash) -> StorageResult<Option<u32>> {
             Ok(self.0.get(hash).copied())
         }
+        async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()> {
+            self.0.retain(|_, h| *h <= target_height);
+            Ok(())
+        }
     }
 
     fn make_diff(base_byte: u8, tip_byte: u8) -> MnListDiff {