diff --git a/dash-spv/src/storage/filters.rs b/dash-spv/src/storage/filters.rs index 37a0cd827..3578a26ea 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,14 @@ 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() + } + 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); + } } 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..9521fcc81 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}` (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 { + let separator = format!("{}_", I::SEGMENT_PREFIX); + let suffix = format!(".{}", I::DATA_FILE_EXTENSION); + + file_name.strip_prefix(&separator)?.strip_suffix(&suffix)?.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,55 @@ 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. + /// + /// 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) { 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::::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().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::::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().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::::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; + + // 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 58192af13..b117b58d2 100644 --- a/dash-spv/src/sync/filters/manager.rs +++ b/dash-spv/src/sync/filters/manager.rs @@ -188,8 +188,17 @@ impl 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, + 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 +306,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 +321,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 +2194,184 @@ 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"); + } + + /// 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;