fix(dash-spv): don't preload filters below the actually stored range on filter-download start

dash-spv/src/storage/filters.rs

@@ -21,6 +21,26 @@ pub trait FilterStorage: Send + Sync + 'static {
 
     async fn filter_tip_height(&self) -> StorageResult<u32>;
 
+    /// Lowest filter height present in storage, or `None` when no filters are
+    /// stored.
+    ///
+    /// `filter_tip_height` is a tip watermark only: storage is dense within
+    /// `[filter_start_height, filter_tip_height]`, but heights below the start
+    /// were never stored. Callers must not treat the tip as proof that storage
+    /// is contiguous from an arbitrary lower height.
+    async fn filter_start_height(&self) -> Option<u32>;
+
+    /// Drop every stored filter, leaving the storage empty.
+    ///
+    /// Used when the stored range cannot serve the current scan (e.g. filters
+    /// were previously stored from a checkpoint above a wallet's rescan
+    /// start): keeping an unreachable island of tip-region filters would leave
+    /// a permanent hole below it that the contiguous store path cannot fill.
+    ///
+    /// Like `truncate_above`, the deletion is not durable until the next
+    /// successful `persist` call.
+    async fn clear_filters(&mut self) -> StorageResult<()>;
+
     /// Drop all filters with `height > target_height`.
     ///
     /// Truncating above the current tip is a no-op, truncating below
@@ -79,6 +99,14 @@ impl FilterStorage for PersistentFilterStorage {
         Ok(self.filters.read().await.tip_height().unwrap_or(0))
     }
 
+    async fn filter_start_height(&self) -> Option<u32> {
+        self.filters.read().await.start_height()
+    }
+
+    async fn clear_filters(&mut self) -> StorageResult<()> {
+        self.filters.write().await.clear()
+    }
+
     async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()> {
         self.filters.write().await.truncate_above(target_height).await
     }
@@ -107,4 +135,57 @@ mod tests {
         assert_eq!(storage.filter_tip_height().await.unwrap(), 2);
         assert!(storage.load_filters(3..4).await.is_err());
     }
+
+    /// Filters stored only in a high region (as when headers previously
+    /// synced from a checkpoint): the start accessor reports the true lowest
+    /// stored height, reads below it fail loudly instead of returning
+    /// sentinels, and `clear_filters` durably empties the storage.
+    #[tokio::test]
+    async fn test_filter_start_height_and_clear_filters() {
+        let tmp_dir = TempDir::new().unwrap();
+        let mut storage = PersistentFilterStorage::open(tmp_dir.path()).await.unwrap();
+
+        assert_eq!(storage.filter_start_height().await, None);
+
+        for height in 900..=1000 {
+            storage.store_filter(height, &filter_bytes(height as u8)).await.unwrap();
+        }
+        assert_eq!(storage.filter_start_height().await, Some(900));
+        assert_eq!(storage.filter_tip_height().await.unwrap(), 1000);
+
+        // Reads that begin below the stored range must error, not hand back
+        // sentinel data zipped against real headers.
+        assert!(storage.load_filters(100..200).await.is_err());
+        assert!(storage.load_filters(850..950).await.is_err());
+        assert_eq!(storage.load_filters(900..1001).await.unwrap().len(), 101);
+
+        storage.persist(tmp_dir.path()).await.unwrap();
+        let segment_file =
+            tmp_dir.path().join(PersistentFilterStorage::FOLDER_NAME).join("segment_0000.dat");
+        assert!(segment_file.exists());
+
+        // The start height survives a reload.
+        let mut storage = PersistentFilterStorage::open(tmp_dir.path()).await.unwrap();
+        assert_eq!(storage.filter_start_height().await, Some(900));
+
+        storage.clear_filters().await.unwrap();
+        assert_eq!(storage.filter_start_height().await, None);
+        assert_eq!(storage.filter_tip_height().await.unwrap(), 0);
+        assert!(storage.load_filters(900..901).await.is_err());
+
+        storage.persist(tmp_dir.path()).await.unwrap();
+        assert!(!segment_file.exists());
+
+        // Reopen: the clear is durable, and the storage accepts a fresh
+        // contiguous range from a lower start.
+        let mut storage = PersistentFilterStorage::open(tmp_dir.path()).await.unwrap();
+        assert_eq!(storage.filter_start_height().await, None);
+        assert_eq!(storage.filter_tip_height().await.unwrap(), 0);
+
+        for height in 100..=110 {
+            storage.store_filter(height, &filter_bytes(height as u8)).await.unwrap();
+        }
+        assert_eq!(storage.filter_start_height().await, Some(100));
+        assert_eq!(storage.load_filters(100..111).await.unwrap().len(), 11);
+    }
 }

dash-spv/src/storage/mod.rs

@@ -375,6 +375,14 @@ impl filters::FilterStorage for DiskStorageManager {
         self.filters.read().await.filter_tip_height().await
     }
 
+    async fn filter_start_height(&self) -> Option<u32> {
+        self.filters.read().await.filter_start_height().await
+    }
+
+    async fn clear_filters(&mut self) -> StorageResult<()> {
+        self.filters.write().await.clear_filters().await
+    }
+
     async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()> {
         self.filters.write().await.truncate_above(target_height).await
     }

dash-spv/src/storage/segments.rs

@@ -107,16 +107,9 @@ impl<I: Persistable> SegmentCache<I> {
 
             for entry in entries.flatten() {
                 if let Some(name) = entry.file_name().to_str() {
-                    if name.starts_with(I::SEGMENT_PREFIX)
-                        && name.ends_with(&format!(".{}", I::DATA_FILE_EXTENSION))
-                    {
-                        let segment_id_start = I::SEGMENT_PREFIX.len() + 1;
-                        let segment_id_end = segment_id_start + 4;
-
-                        if let Ok(id) = name[segment_id_start..segment_id_end].parse::<u32>() {
-                            max_seg_id = Some(max_seg_id.map_or(id, |max: u32| max.max(id)));
-                            min_seg_id = Some(min_seg_id.map_or(id, |min: u32| min.min(id)));
-                        }
+                    if let Some(id) = Self::parse_segment_id(name) {
+                        max_seg_id = Some(max_seg_id.map_or(id, |max: u32| max.max(id)));
+                        min_seg_id = Some(min_seg_id.map_or(id, |min: u32| min.min(id)));
                     }
                 }
             }
@@ -141,6 +134,21 @@ impl<I: Persistable> SegmentCache<I> {
         Ok(cache)
     }
 
+    /// Parse the segment id out of a segment file name of the form
+    /// `{SEGMENT_PREFIX}_{id:04}.{DATA_FILE_EXTENSION}` (see
+    /// [`Persistable::segment_file_name`]).
+    ///
+    /// The entire remaining component between the `{prefix}_` and
+    /// `.{extension}` fixtures must parse as a `u32`, so trailing junk
+    /// (`segment_0000junk.dat`) is rejected and ids longer than the
+    /// zero-padding width (`segment_100000.dat`) are accepted, not truncated.
+    fn parse_segment_id(file_name: &str) -> Option<u32> {
+        let separator = format!("{}_", I::SEGMENT_PREFIX);
+        let suffix = format!(".{}", I::DATA_FILE_EXTENSION);
+
+        file_name.strip_prefix(&separator)?.strip_suffix(&suffix)?.parse::<u32>().ok()
+    }
+
     #[inline]
     fn height_to_segment_id(height: u32) -> u32 {
         height / Segment::<I>::ITEMS_PER_SEGMENT
@@ -226,6 +234,17 @@ impl<I: Persistable> SegmentCache<I> {
             )));
         }
 
+        // Reject ranges that begin below the lowest stored height: those slots
+        // were never populated and hold only sentinels. Without this guard the
+        // read would trip the `first_valid_offset` debug assertion below in
+        // debug builds, and silently hand back sentinel data in release builds.
+        if self.start_height.is_none_or(|lowest| start < lowest) {
+            return Err(StorageError::InvalidArgument(format!(
+                "get_items range {height_range:?} begins below the lowest stored height {:?}",
+                self.start_height
+            )));
+        }
+
         let mut items = Vec::with_capacity((end - start) as usize);
 
         let start_segment = Self::height_to_segment_id(start);
@@ -429,6 +448,55 @@ impl<I: Persistable> SegmentCache<I> {
         Ok(())
     }
 
+    /// Drop every stored item, leaving the cache empty.
+    ///
+    /// All backing segment files — including segments not currently resident
+    /// in memory — are queued for deletion on the next `persist`. Like
+    /// `truncate_above`, the deletion is not durable until that `persist`
+    /// succeeds; a crash in between leaves the old files on disk and the
+    /// cache reopens with the pre-clear contents.
+    ///
+    /// The directory scan runs to completion *before* any cache state is
+    /// mutated: a failed scan must not leave the cache reporting empty while
+    /// persisted segment files survive on disk, since those would resurrect
+    /// after restart. A missing segments directory is treated as an empty
+    /// (already-cleared) cache; any other I/O error is propagated.
+    pub fn clear(&mut self) -> StorageResult<()> {
+        // Queue every on-disk segment file, discovered by directory scan since
+        // only up to MAX_ACTIVE_SEGMENTS of them are resident in memory.
+        let mut to_delete = HashSet::new();
+        match fs::read_dir(&self.segments_dir) {
+            Ok(entries) => {
+                for entry in entries {
+                    let entry = entry?;
+                    if let Some(name) = entry.file_name().to_str() {
+                        if let Some(id) = Self::parse_segment_id(name) {
+                            to_delete.insert(id);
+                        }
+                    }
+                }
+            }
+            // No directory yet means nothing was ever persisted here.
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
+            Err(e) => return Err(StorageError::Io(e)),
+        }
+
+        // Scan succeeded — now it is safe to mutate cache state. Resident and
+        // evicted segments may be dirty and not persisted yet, so the scan
+        // above cannot see them. Queue their ids too; `persist` ignores
+        // missing files.
+        to_delete.extend(self.segments.keys().copied());
+        to_delete.extend(self.evicted.keys().copied());
+
+        self.to_delete.extend(to_delete);
+        self.segments.clear();
+        self.evicted.clear();
+        self.tip_height = None;
+        self.start_height = None;
+
+        Ok(())
+    }
+
     pub async fn persist(&mut self, segments_dir: impl Into<PathBuf>) {
         let segments_dir = segments_dir.into();
 
@@ -1092,6 +1160,153 @@ mod tests {
         );
     }
 
+    /// Ranges that begin below the lowest stored height read never-populated
+    /// sentinel slots and must fail with a typed error rather than panicking
+    /// on the `first_valid_offset` debug assertion (or silently returning
+    /// sentinels in release builds).
+    #[tokio::test]
+    async fn test_get_items_below_start_height_errors() {
+        let tmp_dir = TempDir::new().unwrap();
+
+        let items = FilterHeader::dummy_batch(0..10);
+
+        let mut cache = SegmentCache::<FilterHeader>::load_or_new(tmp_dir.path()).await.unwrap();
+        cache.store_items_at_height(&items, 900).await.unwrap();
+        assert_eq!(cache.start_height(), Some(900));
+
+        // Entirely below the stored range (the #892 shape: a scan start far
+        // below tip-region storage, within the same segment).
+        assert!(matches!(cache.get_items(100..200).await, Err(StorageError::InvalidArgument(_))));
+        // Straddling the lowest stored height.
+        assert!(matches!(cache.get_items(895..905).await, Err(StorageError::InvalidArgument(_))));
+        // The stored range itself is readable.
+        assert_eq!(cache.get_items(900..910).await.unwrap(), items);
+
+        // Same result when the never-populated slots live in a lower segment
+        // than the stored data.
+        const ITEMS_PER_SEGMENT: u32 = Segment::<FilterHeader>::ITEMS_PER_SEGMENT;
+        let mut cache = SegmentCache::<FilterHeader>::load_or_new(tmp_dir.path()).await.unwrap();
+        cache.store_items_at_height(&items, ITEMS_PER_SEGMENT + 5).await.unwrap();
+        assert!(matches!(cache.get_items(100..200).await, Err(StorageError::InvalidArgument(_))));
+
+        // An empty cache rejects every range.
+        let mut cache =
+            SegmentCache::<FilterHeader>::load_or_new(tmp_dir.path().join("empty")).await.unwrap();
+        assert!(matches!(cache.get_items(0..1).await, Err(StorageError::InvalidArgument(_))));
+    }
+
+    #[tokio::test]
+    async fn test_segment_cache_clear() {
+        let tmp_dir = TempDir::new().unwrap();
+
+        const ITEMS_PER_SEGMENT: u32 = Segment::<FilterHeader>::ITEMS_PER_SEGMENT;
+
+        // Two segments' worth of items, persisted to disk.
+        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 file_0 = tmp_dir.path().join(FilterHeader::segment_file_name(0));
+        let file_1 = tmp_dir.path().join(FilterHeader::segment_file_name(1));
+        assert!(file_0.exists());
+        assert!(file_1.exists());
+
+        // Reopen (so only the min/max segments are resident) and clear.
+        let mut cache = SegmentCache::<FilterHeader>::load_or_new(tmp_dir.path()).await.unwrap();
+        cache.clear().unwrap();
+
+        assert_eq!(cache.tip_height(), None);
+        assert_eq!(cache.start_height(), None);
+        assert_eq!(cache.get_item(3).await.unwrap(), None);
+        assert!(cache.get_items(0..5).await.is_err());
+
+        // The deletion becomes durable on persist.
+        cache.persist(tmp_dir.path()).await;
+        assert!(!file_0.exists());
+        assert!(!file_1.exists());
+
+        // Storing a fresh range from a lower origin after clear is sound.
+        let replacement = FilterHeader::dummy_batch(100..110);
+        cache.store_items_at_height(&replacement, 100).await.unwrap();
+        assert_eq!(cache.start_height(), Some(100));
+        assert_eq!(cache.tip_height(), Some(109));
+        assert_eq!(cache.get_items(100..110).await.unwrap(), replacement);
+
+        cache.persist(tmp_dir.path()).await;
+        let mut reloaded = SegmentCache::<FilterHeader>::load_or_new(tmp_dir.path()).await.unwrap();
+        assert_eq!(reloaded.start_height(), Some(100));
+        assert_eq!(reloaded.tip_height(), Some(109));
+        assert_eq!(reloaded.get_items(100..110).await.unwrap(), replacement);
+    }
+
+    /// `clear` must also drop dirty in-memory state that was never persisted,
+    /// and queue previously-persisted files of those same segments.
+    #[tokio::test]
+    async fn test_segment_cache_clear_unpersisted_state() {
+        let tmp_dir = TempDir::new().unwrap();
+
+        let items = FilterHeader::dummy_batch(0..10);
+        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;
+
+        // Dirty the resident segment again so it differs from disk, then clear
+        // without an intervening persist.
+        let more = FilterHeader::dummy_batch(100..105);
+        cache.store_items_at_height(&more, 10).await.unwrap();
+        cache.clear().unwrap();
+
+        assert_eq!(cache.tip_height(), None);
+        assert_eq!(cache.start_height(), None);
+
+        cache.persist(tmp_dir.path()).await;
+        assert!(!tmp_dir.path().join(FilterHeader::segment_file_name(0)).exists());
+
+        let reloaded = SegmentCache::<FilterHeader>::load_or_new(tmp_dir.path()).await.unwrap();
+        assert_eq!(reloaded.tip_height(), None);
+        assert_eq!(reloaded.start_height(), None);
+    }
+
+    #[test]
+    fn test_parse_segment_id() {
+        type Cache = SegmentCache<FilterHeader>;
+
+        // Round-trips the writer's `{prefix}_{id:04}.{ext}` format for a range
+        // of ids, including ones wider than the zero-padding.
+        for id in [0u32, 1, 42, 9999, 10_000, 100_000, u32::MAX] {
+            assert_eq!(Cache::parse_segment_id(&FilterHeader::segment_file_name(id)), Some(id));
+        }
+
+        // Zero-padded low ids parse to their numeric value, not truncated.
+        assert_eq!(Cache::parse_segment_id("segment_0000.dat"), Some(0));
+        assert_eq!(Cache::parse_segment_id("segment_0005.dat"), Some(5));
+
+        // Ids wider than the padding are accepted whole, never truncated to
+        // the first four digits.
+        assert_eq!(Cache::parse_segment_id("segment_100000.dat"), Some(100_000));
+
+        // Trailing junk between the id and the extension is rejected rather
+        // than silently parsed as a prefix of the component.
+        assert_eq!(Cache::parse_segment_id("segment_0000junk.dat"), None);
+        assert_eq!(Cache::parse_segment_id("segment_12ab.dat"), None);
+
+        // Wrong prefix, wrong extension, missing separator, or a
+        // non-numeric/empty id component are all rejected.
+        assert_eq!(Cache::parse_segment_id("segment_0000.tmp"), None);
+        assert_eq!(Cache::parse_segment_id("other_0000.dat"), None);
+        assert_eq!(Cache::parse_segment_id("segment0000.dat"), None);
+        assert_eq!(Cache::parse_segment_id("segment_.dat"), None);
+        assert_eq!(Cache::parse_segment_id("segment_-1.dat"), None);
+        assert_eq!(Cache::parse_segment_id("segment_0000.dat.bak"), None);
+
+        // The filter storage's item type round-trips through the same helper.
+        assert_eq!(
+            SegmentCache::<Vec<u8>>::parse_segment_id(&<Vec<u8>>::segment_file_name(7)),
+            Some(7)
+        );
+    }
+
     #[test]
     fn test_segment_insert_get() {
         let segment_id = 10;