From 23225a82b935910ca7ec738f007fbf84e14c37b9 Mon Sep 17 00:00:00 2001 From: Quantum Explorer Date: Wed, 15 Jul 2026 03:02:05 +0700 Subject: [PATCH 1/2] fix(dash-spv): don't preload filters below the actually stored range on filter-download start MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit FiltersManager::start_download treated stored_filters_tip — a tip watermark — as proof that filter storage is contiguous from scan_start. When only tip-region filters were ever stored (e.g. headers previously synced from a checkpoint above the wallet's birth height), the preload called load_filters over never-populated segments: a debug-assertion abort in SegmentCache::get_items (crash-loop on every launch, since the restart resumes the same scan state) and silent sentinel filter data zipped against real headers in release builds. - FilterStorage gains filter_start_height() (lowest stored height, backed by the SegmentCache start it already tracks) and clear_filters() (drop all filters; files removed on next persist). - start_download only treats the stored range as usable when it reaches down to scan_start. Otherwise the unreachable tip-region filters are discarded and the download restarts from scan_start exactly as if no filters were stored — resuming from their tip would leave a permanent hole below them that the in-order store loop can never fill. Preload and the initial batch's mark_verified are gated on the same check, and progress.stored_height is reset so batch scan gating no longer trusts the stale watermark. - SegmentCache::get_items now returns a typed InvalidArgument error for ranges that begin below the lowest stored height, replacing the debug-only panic / release sentinel reads for that case. Fixes #892 Co-Authored-By: Claude Fable 5 --- dash-spv/src/storage/filters.rs | 82 ++++++++++++ dash-spv/src/storage/mod.rs | 8 ++ dash-spv/src/storage/segments.rs | 179 +++++++++++++++++++++++++-- dash-spv/src/sync/filters/manager.rs | 165 +++++++++++++++++++++++- 4 files changed, 419 insertions(+), 15 deletions(-) diff --git a/dash-spv/src/storage/filters.rs b/dash-spv/src/storage/filters.rs index 37a0cd827..07f449ea7 100644 --- a/dash-spv/src/storage/filters.rs +++ b/dash-spv/src/storage/filters.rs @@ -21,6 +21,26 @@ pub trait FilterStorage: Send + Sync + 'static { async fn filter_tip_height(&self) -> StorageResult; + /// 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; + + /// 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,15 @@ impl FilterStorage for PersistentFilterStorage { Ok(self.filters.read().await.tip_height().unwrap_or(0)) } + async fn filter_start_height(&self) -> Option { + self.filters.read().await.start_height() + } + + async fn clear_filters(&mut self) -> StorageResult<()> { + self.filters.write().await.clear(); + Ok(()) + } + async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()> { self.filters.write().await.truncate_above(target_height).await } @@ -107,4 +136,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); + } } diff --git a/dash-spv/src/storage/mod.rs b/dash-spv/src/storage/mod.rs index 73135493b..248ab10d0 100644 --- a/dash-spv/src/storage/mod.rs +++ b/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 { + 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 } diff --git a/dash-spv/src/storage/segments.rs b/dash-spv/src/storage/segments.rs index 49f291add..bbfe634b5 100644 --- a/dash-spv/src/storage/segments.rs +++ b/dash-spv/src/storage/segments.rs @@ -107,16 +107,9 @@ impl SegmentCache { 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::() { - 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 SegmentCache { Ok(cache) } + /// Parse the segment id out of a segment file name of the form + /// `{SEGMENT_PREFIX}_{id:04}.{DATA_FILE_EXTENSION}`. + fn parse_segment_id(file_name: &str) -> Option { + if !file_name.starts_with(I::SEGMENT_PREFIX) + || !file_name.ends_with(&format!(".{}", I::DATA_FILE_EXTENSION)) + { + return None; + } + + let segment_id_start = I::SEGMENT_PREFIX.len() + 1; + let segment_id_end = segment_id_start + 4; + + file_name.get(segment_id_start..segment_id_end)?.parse::().ok() + } + #[inline] fn height_to_segment_id(height: u32) -> u32 { height / Segment::::ITEMS_PER_SEGMENT @@ -226,6 +234,17 @@ impl SegmentCache { ))); } + // 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,38 @@ impl SegmentCache { 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. + pub fn clear(&mut self) { + // Queue every on-disk segment file, discovered by directory scan since + // only up to MAX_ACTIVE_SEGMENTS of them are resident in memory. + if let Ok(entries) = fs::read_dir(&self.segments_dir) { + for entry in entries.flatten() { + if let Some(name) = entry.file_name().to_str() { + if let Some(id) = Self::parse_segment_id(name) { + self.to_delete.insert(id); + } + } + } + } + + // 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. + self.to_delete.extend(self.segments.keys().copied()); + self.to_delete.extend(self.evicted.keys().copied()); + + self.segments.clear(); + self.evicted.clear(); + self.tip_height = None; + self.start_height = None; + } + pub async fn persist(&mut self, segments_dir: impl Into) { let segments_dir = segments_dir.into(); @@ -1092,6 +1143,114 @@ 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::::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::::ITEMS_PER_SEGMENT; + let mut cache = SegmentCache::::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::::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::::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::::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::::load_or_new(tmp_dir.path()).await.unwrap(); + cache.clear(); + + 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::::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::::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(); + + 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::::load_or_new(tmp_dir.path()).await.unwrap(); + assert_eq!(reloaded.tip_height(), None); + assert_eq!(reloaded.start_height(), None); + } + #[test] fn test_segment_insert_get() { let segment_id = 10; diff --git a/dash-spv/src/sync/filters/manager.rs b/dash-spv/src/sync/filters/manager.rs index 58192af13..7b50dc3e9 100644 --- a/dash-spv/src/sync/filters/manager.rs +++ b/dash-spv/src/sync/filters/manager.rs @@ -188,8 +188,13 @@ impl 0 && stored_filters_start.is_some_and(|start| start <= scan_start); + + // Stored filters that don't reach down to `scan_start` cannot seed + // this scan, and resuming the download from their tip would leave a + // permanent hole below them that the in-order store loop can never + // fill. Drop them and re-download from `scan_start`, exactly as if no + // filters were stored, keeping storage dense within its stored range. + if stored_filters_tip > 0 && !stored_covers_scan { + tracing::warn!( + "Stored filters {:?}..={} do not reach scan start {}; discarding them and re-downloading", + stored_filters_start, + stored_filters_tip, + scan_start + ); + self.filter_storage.write().await.clear_filters().await?; + self.progress.update_stored_height(0); + } + // Determine download start (where we need to download from) // Must be at least header_start_height for checkpoint-based sync - let download_start = if stored_filters_tip > 0 { + let download_start = if stored_covers_scan { (stored_filters_tip + 1).max(header_start_height) } else { scan_start @@ -268,7 +298,7 @@ impl 0 && scan_start <= stored_filters_tip { + let filters = if stored_covers_scan && scan_start <= stored_filters_tip { let end_height = stored_filters_tip.min(batch_end); tracing::info!( "Loading stored filters {} to {} into current batch", @@ -283,7 +313,7 @@ impl= batch_end { + if stored_covers_scan && stored_filters_tip >= batch_end { batch.mark_verified(); } self.active_batches.insert(scan_start, batch); @@ -2156,6 +2186,131 @@ mod tests { assert_eq!(batch.end_height(), 100); } + /// Reproduces #892: filters were only ever stored near a high tip (e.g. + /// headers previously synced from a checkpoint), and the wallet's scan + /// start is far below that region. `start_download` must not treat the + /// stored tip watermark as contiguous coverage from `scan_start`: + /// preloading `load_filters(scan_start, ..)` would read never-populated + /// segments (debug abort in `SegmentCache::get_items`, sentinel filter + /// data in release builds). The unreachable tip-region filters are + /// discarded and the download restarts from `scan_start`. + #[tokio::test] + async fn test_start_download_discards_stored_filters_above_scan_start() { + let mut manager = create_test_manager().await; + + // Block headers cover the whole range so send_pending can resolve stop hashes. + let headers = dashcore::block::Header::dummy_batch(0..1001); + manager + .header_storage + .write() + .await + .store_headers( + &headers.iter().map(crate::types::HashedBlockHeader::from).collect::>(), + ) + .await + .unwrap(); + + // Filters were only ever stored near the tip: 900..=1000. The heights + // below 900 were never populated. + { + let mut filter_storage = manager.filter_storage.write().await; + for height in 900..=1000u32 { + filter_storage.store_filter(height, &[height as u8; 8]).await.unwrap(); + } + } + // Restart shape: `new()` seeds stored_height from the tip watermark. + manager.progress.update_stored_height(1000); + manager.progress.update_filter_header_tip_height(1000); + manager.progress.update_target_height(1000); + + // Wallet committed far below the stored region, so scan_start = 100. + manager.wallet.write().await.update_wallet_synced_height(&MOCK_WALLET_ID, 99); + + let (tx, mut rx) = unbounded_channel(); + let events = manager.start_download(&RequestSender::new(tx)).await.unwrap(); + + assert!(events.is_empty()); + assert_eq!(manager.state(), SyncState::Syncing); + + // The unreachable tip-region filters were discarded entirely... + assert_eq!(manager.filter_storage.read().await.filter_tip_height().await.unwrap(), 0); + assert_eq!(manager.filter_storage.read().await.filter_start_height().await, None); + + // ...nothing was preloaded into the initial batch... + let batch = manager.active_batches.get(&100).expect("initial batch at scan_start"); + assert!(batch.filters().is_empty()); + assert!(!batch.verified()); + assert!(!batch.scanned()); + assert_eq!(batch.end_height(), 1000); + + // ...scan gating no longer sees the stale tip watermark... + assert_eq!(manager.progress.stored_height(), 0); + + // ...and both the store cursor and the download restart from + // scan_start rather than stored_filters_tip + 1. + assert_eq!(manager.next_batch_to_store, 100); + match rx.try_recv().expect("a filter request must have been sent") { + NetworkRequest::SendMessage(NetworkMessage::GetCFilters(gcf)) => { + assert_eq!(gcf.start_height, 100); + } + other => panic!("Expected GetCFilters, got {:?}", other), + } + } + + /// Counterpart to the sparse-storage case: when the stored filter range + /// actually reaches down to `scan_start`, the preload happens exactly as + /// before and nothing is discarded. + #[tokio::test] + async fn test_start_download_preloads_when_stored_filters_cover_scan_start() { + let mut manager = create_test_manager().await; + + let headers = dashcore::block::Header::dummy_batch(0..1001); + manager + .header_storage + .write() + .await + .store_headers( + &headers.iter().map(crate::types::HashedBlockHeader::from).collect::>(), + ) + .await + .unwrap(); + + // Filters stored densely from genesis through the tip. + { + let mut filter_storage = manager.filter_storage.write().await; + for height in 0..=1000u32 { + filter_storage.store_filter(height, &[height as u8; 8]).await.unwrap(); + } + } + manager.progress.update_stored_height(1000); + manager.progress.update_filter_header_tip_height(1000); + manager.progress.update_target_height(1000); + + manager.wallet.write().await.update_wallet_synced_height(&MOCK_WALLET_ID, 99); + + let (tx, mut rx) = unbounded_channel(); + manager.start_download(&RequestSender::new(tx)).await.unwrap(); + + assert_eq!(manager.state(), SyncState::Syncing); + + // Storage is untouched. + assert_eq!(manager.filter_storage.read().await.filter_tip_height().await.unwrap(), 1000); + assert_eq!(manager.filter_storage.read().await.filter_start_height().await, Some(0)); + + // The stored range 100..=1000 was preloaded and the batch is verified + // and scanned immediately. + let batch = manager.active_batches.get(&100).expect("initial batch at scan_start"); + assert_eq!(batch.filters().len(), 901); + assert!(batch.verified()); + assert!(batch.scanned()); + assert_eq!(manager.progress.stored_height(), 1000); + + // Nothing left to download: everything through the filter header tip + // is already stored. + assert_eq!(manager.next_batch_to_store, 1001); + assert!(rx.try_recv().is_err(), "no filter request expected"); + } + #[tokio::test] async fn test_handle_new_filter_headers_transitions_synced_to_syncing() { let mut manager = create_test_manager().await; From 828f707d5d91301a1587cd3460622b5a2435b99e Mon Sep 17 00:00:00 2001 From: Quantum Explorer Date: Wed, 15 Jul 2026 03:17:32 +0700 Subject: [PATCH 2/2] fix(dash-spv): address CodeRabbit review on sparse-filter preload fix Three follow-ups to the #892 fix: - parse_segment_id: parse the entire component between the `{prefix}_` and `.{ext}` fixtures via strip_prefix/strip_suffix instead of a fixed 4-byte slice. Trailing junk (`segment_0000junk.dat`) is now rejected and ids wider than the zero-padding (`segment_100000.dat`) are accepted rather than silently truncated. Adds a dedicated parse_segment_id test covering round-trips, junk tails, 5+ digit ids, and malformed names. - SegmentCache::clear now returns StorageResult<()>: the directory scan runs to completion before any cache state is mutated, io NotFound on read_dir is treated as empty-ok, and every other error is propagated. A failed scan no longer leaves the cache reporting empty while persisted segment files survive to resurrect the sparse-storage bug after restart. The Result is threaded through clear_filters and its callers. - start_download gates stored_covers_scan on the stored start height alone (`stored_filters_start.is_some_and(|s| s <= scan_start)`). filter_tip_height collapses "empty" and "only height 0 stored" to 0, so the previous `stored_filters_tip > 0` guard misread a genesis-only store as empty and re-downloaded height 0 instead of preloading it. The discard branch is made consistent via `is_some()`. Adds a genesis-only restart test asserting the lone filter is preloaded with no clear and no re-request. Co-Authored-By: Claude Fable 5 --- dash-spv/src/storage/filters.rs | 3 +- dash-spv/src/storage/segments.rs | 102 +++++++++++++++++++++------ dash-spv/src/sync/filters/manager.rs | 69 ++++++++++++++++-- 3 files changed, 145 insertions(+), 29 deletions(-) diff --git a/dash-spv/src/storage/filters.rs b/dash-spv/src/storage/filters.rs index 07f449ea7..3578a26ea 100644 --- a/dash-spv/src/storage/filters.rs +++ b/dash-spv/src/storage/filters.rs @@ -104,8 +104,7 @@ impl FilterStorage for PersistentFilterStorage { } async fn clear_filters(&mut self) -> StorageResult<()> { - self.filters.write().await.clear(); - Ok(()) + self.filters.write().await.clear() } async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()> { diff --git a/dash-spv/src/storage/segments.rs b/dash-spv/src/storage/segments.rs index bbfe634b5..9521fcc81 100644 --- a/dash-spv/src/storage/segments.rs +++ b/dash-spv/src/storage/segments.rs @@ -135,18 +135,18 @@ impl SegmentCache { } /// Parse the segment id out of a segment file name of the form - /// `{SEGMENT_PREFIX}_{id:04}.{DATA_FILE_EXTENSION}`. + /// `{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 { - if !file_name.starts_with(I::SEGMENT_PREFIX) - || !file_name.ends_with(&format!(".{}", I::DATA_FILE_EXTENSION)) - { - return None; - } + let separator = format!("{}_", I::SEGMENT_PREFIX); + let suffix = format!(".{}", I::DATA_FILE_EXTENSION); - let segment_id_start = I::SEGMENT_PREFIX.len() + 1; - let segment_id_end = segment_id_start + 4; - - file_name.get(segment_id_start..segment_id_end)?.parse::().ok() + file_name.strip_prefix(&separator)?.strip_suffix(&suffix)?.parse::().ok() } #[inline] @@ -455,29 +455,46 @@ impl SegmentCache { /// `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. - pub fn clear(&mut self) { + /// + /// 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. - if let Ok(entries) = fs::read_dir(&self.segments_dir) { - for entry in entries.flatten() { - if let Some(name) = entry.file_name().to_str() { - if let Some(id) = Self::parse_segment_id(name) { - self.to_delete.insert(id); + 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)), } - // 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. - self.to_delete.extend(self.segments.keys().copied()); - self.to_delete.extend(self.evicted.keys().copied()); + // 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) { @@ -1197,7 +1214,7 @@ mod tests { // Reopen (so only the min/max segments are resident) and clear. let mut cache = SegmentCache::::load_or_new(tmp_dir.path()).await.unwrap(); - cache.clear(); + cache.clear().unwrap(); assert_eq!(cache.tip_height(), None); assert_eq!(cache.start_height(), None); @@ -1238,7 +1255,7 @@ mod tests { // without an intervening persist. let more = FilterHeader::dummy_batch(100..105); cache.store_items_at_height(&more, 10).await.unwrap(); - cache.clear(); + cache.clear().unwrap(); assert_eq!(cache.tip_height(), None); assert_eq!(cache.start_height(), None); @@ -1251,6 +1268,45 @@ mod tests { assert_eq!(reloaded.start_height(), None); } + #[test] + fn test_parse_segment_id() { + type Cache = SegmentCache; + + // 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::>::parse_segment_id(&>::segment_file_name(7)), + Some(7) + ); + } + #[test] fn test_segment_insert_get() { let segment_id = 10; diff --git a/dash-spv/src/sync/filters/manager.rs b/dash-spv/src/sync/filters/manager.rs index 7b50dc3e9..b117b58d2 100644 --- a/dash-spv/src/sync/filters/manager.rs +++ b/dash-spv/src/sync/filters/manager.rs @@ -191,6 +191,10 @@ impl 0 && stored_filters_start.is_some_and(|start| start <= scan_start); + // sentinel filter data in release builds). Gate purely on the stored + // start so a genesis-only store (start = tip = 0) is recognized as + // covering a scan from 0 rather than being misread as empty. + let stored_covers_scan = stored_filters_start.is_some_and(|start| start <= scan_start); // Stored filters that don't reach down to `scan_start` cannot seed // this scan, and resuming the download from their tip would leave a // permanent hole below them that the in-order store loop can never // fill. Drop them and re-download from `scan_start`, exactly as if no // filters were stored, keeping storage dense within its stored range. - if stored_filters_tip > 0 && !stored_covers_scan { + // `is_some()` (not `tip > 0`) so this covers a lone above-scan filter + // stored at height 0, though that cannot arise while the start is also + // above `scan_start`. + if stored_filters_start.is_some() && !stored_covers_scan { tracing::warn!( "Stored filters {:?}..={} do not reach scan start {}; discarding them and re-downloading", stored_filters_start, @@ -2311,6 +2319,59 @@ mod tests { assert!(rx.try_recv().is_err(), "no filter request expected"); } + /// Genesis-only storage: a single filter at height 0, scanning from 0. + /// `filter_tip_height` collapses to 0 for both an empty store and this + /// one, so gating the preload on `tip > 0` would misread it as empty and + /// needlessly re-download height 0. The stored start (Some(0)) must drive + /// the decision: the filter is preloaded and nothing is discarded or + /// re-requested. Regtest can produce exactly this shape. + #[tokio::test] + async fn test_start_download_preloads_genesis_only_stored_filter() { + let mut manager = create_test_manager().await; + + let headers = dashcore::block::Header::dummy_batch(0..1); + manager + .header_storage + .write() + .await + .store_headers( + &headers.iter().map(crate::types::HashedBlockHeader::from).collect::>(), + ) + .await + .unwrap(); + + // Exactly one filter stored, at height 0. + manager.filter_storage.write().await.store_filter(0, &[0u8; 8]).await.unwrap(); + assert_eq!(manager.filter_storage.read().await.filter_tip_height().await.unwrap(), 0); + assert_eq!(manager.filter_storage.read().await.filter_start_height().await, Some(0)); + + // Restart shape at the genesis tip. + manager.progress.update_stored_height(0); + manager.progress.update_filter_header_tip_height(0); + manager.progress.update_target_height(0); + // Wallet at genesis: scan_start = 0. + + let (tx, mut rx) = unbounded_channel(); + manager.start_download(&RequestSender::new(tx)).await.unwrap(); + + // The lone filter was NOT discarded... + assert_eq!(manager.filter_storage.read().await.filter_tip_height().await.unwrap(), 0); + assert_eq!(manager.filter_storage.read().await.filter_start_height().await, Some(0)); + + // ...it was preloaded into the initial batch, which is verified and + // scanned since the whole (single-height) range is covered... + let batch = manager.active_batches.get(&0).expect("initial batch at scan_start"); + assert_eq!(batch.filters().len(), 1); + assert!(batch.verified()); + assert!(batch.scanned()); + assert_eq!(manager.progress.stored_height(), 0); + + // ...and the download frontier sits above the stored tip, so no + // filter request goes out for the already-stored genesis height. + assert_eq!(manager.next_batch_to_store, 1); + assert!(rx.try_recv().is_err(), "no filter request expected for the genesis-only store"); + } + #[tokio::test] async fn test_handle_new_filter_headers_transitions_synced_to_syncing() { let mut manager = create_test_manager().await;