feat: filter storage using segmentscache (#267)

* Implement segmented filter storage

* Store filters during initial sync

* Add startup initialization

* Add tests

* filter data storage uses SegmentCache generic struct to persist teh data

* fixed clippy warning

* correctly setting sync_base_height using the chain state in the new filters segment cache

---------

Co-authored-by: xdustinface <xdustinfacex@gmail.com>

Author: ZocoLini

dash-spv/src/storage/disk/filters.rs

@@ -24,6 +24,9 @@ pub(super) enum WorkerCommand {
     SaveFilterHeaderSegmentCache {
         segment_id: u32,
     },
+    SaveFilterSegmentCache {
+        segment_id: u32,
+    },
     SaveIndex {
         index: HashMap<BlockHash, u32>,
     },
@@ -37,6 +40,7 @@ pub struct DiskStorageManager {
     // Segmented header storage
     pub(super) block_headers: Arc<RwLock<SegmentCache<BlockHeader>>>,
     pub(super) filter_headers: Arc<RwLock<SegmentCache<FilterHeader>>>,
+    pub(super) filters: Arc<RwLock<SegmentCache<Vec<u8>>>>,
 
     // Reverse index for O(1) lookups
     pub(super) header_hash_index: Arc<RwLock<HashMap<BlockHash, u32>>>,
@@ -107,6 +111,9 @@ impl DiskStorageManager {
             filter_headers: Arc::new(RwLock::new(
                 SegmentCache::load_or_new(base_path.clone(), sync_base_height).await?,
             )),
+            filters: Arc::new(RwLock::new(
+                SegmentCache::load_or_new(base_path.clone(), sync_base_height).await?,
+            )),
             header_hash_index: Arc::new(RwLock::new(HashMap::new())),
             worker_tx: None,
             worker_handle: None,
@@ -120,6 +127,7 @@ impl DiskStorageManager {
         if let Ok(Some(state)) = storage.load_chain_state().await {
             storage.filter_headers.write().await.set_sync_base_height(state.sync_base_height);
             storage.block_headers.write().await.set_sync_base_height(state.sync_base_height);
+            storage.filters.write().await.set_sync_base_height(state.sync_base_height);
             tracing::debug!("Loaded sync_base_height: {}", state.sync_base_height);
         }
 
@@ -151,6 +159,7 @@ impl DiskStorageManager {
 
         let block_headers = Arc::clone(&self.block_headers);
         let filter_headers = Arc::clone(&self.filter_headers);
+        let cfilters = Arc::clone(&self.filters);
 
         let worker_handle = tokio::spawn(async move {
             while let Some(cmd) = worker_rx.recv().await {
@@ -203,6 +212,30 @@ impl DiskStorageManager {
                             }
                         }
                     }
+                    WorkerCommand::SaveFilterSegmentCache {
+                        segment_id,
+                    } => {
+                        let mut cache = cfilters.write().await;
+                        let segment = match cache.get_segment_mut(&segment_id).await {
+                            Ok(segment) => segment,
+                            Err(e) => {
+                                eprintln!("Failed to get segment {}: {}", segment_id, e);
+                                continue;
+                            }
+                        };
+
+                        match segment.persist(&base_path).await {
+                            Ok(()) => {
+                                tracing::trace!(
+                                    "Background worker completed saving filter segment {}",
+                                    segment_id
+                                );
+                            }
+                            Err(e) => {
+                                eprintln!("Failed to save segment {}: {}", segment_id, e);
+                            }
+                        }
+                    }
                     WorkerCommand::SaveIndex {
                         index,
                     } => {

dash-spv/src/storage/disk/manager.rs

@@ -19,7 +19,6 @@
 //! - Better concurrency
 //! - Simpler code
 
-mod filters;
 mod headers;
 pub(crate) mod io;
 mod lockfile;

dash-spv/src/storage/disk/mod.rs

@@ -51,6 +51,16 @@ pub trait Persistable: Sized + Encodable + Decodable + PartialEq + Clone {
     fn make_save_command(segment: &Segment<Self>) -> WorkerCommand;
 }
 
+impl Persistable for Vec<u8> {
+    const FOLDER_NAME: &'static str = "filters";
+
+    fn make_save_command(segment: &Segment<Self>) -> WorkerCommand {
+        WorkerCommand::SaveFilterSegmentCache {
+            segment_id: segment.segment_id,
+        }
+    }
+}
+
 impl Persistable for BlockHeader {
     const FOLDER_NAME: &'static str = "block_headers";
 

dash-spv/src/storage/disk/segments.rs

@@ -21,6 +21,7 @@ impl DiskStorageManager {
         // Update our sync_base_height
         self.filter_headers.write().await.set_sync_base_height(state.sync_base_height);
         self.block_headers.write().await.set_sync_base_height(state.sync_base_height);
+        self.filters.write().await.set_sync_base_height(state.sync_base_height);
 
         // First store all headers
         // For checkpoint sync, we need to store headers starting from the checkpoint height
@@ -355,6 +356,8 @@ impl DiskStorageManager {
         // Clear in-memory state
         self.block_headers.write().await.clear_in_memory();
         self.filter_headers.write().await.clear_in_memory();
+        self.filters.write().await.clear_in_memory();
+
         self.header_hash_index.write().await.clear();
         self.mempool_transactions.write().await.clear();
         *self.mempool_state.write().await = None;
@@ -449,8 +452,8 @@ impl DiskStorageManager {
     /// Save all dirty segments to disk via background worker.
     pub(super) async fn save_dirty(&self) {
         self.filter_headers.write().await.persist_dirty(self).await;
-
         self.block_headers.write().await.persist_dirty(self).await;
+        self.filters.write().await.persist_dirty(self).await;
 
         if let Some(tx) = &self.worker_tx {
             // Save the index only if it has grown significantly (every 10k new entries)
@@ -594,11 +597,16 @@ impl StorageManager for DiskStorageManager {
     }
 
     async fn store_filter(&mut self, height: u32, filter: &[u8]) -> StorageResult<()> {
-        Self::store_filter(self, height, filter).await
+        self.filters.write().await.store_items_at_height(&[filter.to_vec()], height, self).await
     }
 
     async fn load_filter(&self, height: u32) -> StorageResult<Option<Vec<u8>>> {
-        Self::load_filter(self, height).await
+        self.filters
+            .write()
+            .await
+            .get_items(height..height + 1)
+            .await
+            .map(|items| items.first().cloned())
     }
 
     async fn store_metadata(&mut self, key: &str, value: &[u8]) -> StorageResult<()> {
@@ -614,7 +622,17 @@ impl StorageManager for DiskStorageManager {
     }
 
     async fn clear_filters(&mut self) -> StorageResult<()> {
-        Self::clear_filters(self).await
+        // Stop worker to prevent concurrent writes to filter directories
+        self.stop_worker().await;
+
+        // Clear in-memory and on-disk filter headers segments
+        self.filter_headers.write().await.clear_all().await?;
+        self.filters.write().await.clear_all().await?;
+
+        // Restart background worker for future operations
+        self.start_worker().await;
+
+        Ok(())
     }
 
     async fn stats(&self) -> StorageResult<StorageStats> {

dash-spv/src/storage/disk/state.rs

@@ -611,6 +611,12 @@ impl<
             return Ok(());
         }
 
+        // Store the verified filter to disk
+        storage
+            .store_filter(height, &cfilter.filter)
+            .await
+            .map_err(|e| SyncError::Storage(format!("Failed to store filter: {}", e)))?;
+
         let matches = self
             .filter_sync
             .check_filter_for_matches(

dash-spv/src/sync/message_handlers.rs

@@ -302,7 +302,7 @@ async fn test_mixed_operations() {
     storage.store_filter_headers(&filter_headers).await.unwrap();
 
     // Store some filters
-    for height in [1000, 5000, 50_000, 70_000] {
+    for height in 0..75_000 {
         let filter_data = vec![height as u8; 100];
         storage.store_filter(height, &filter_data).await.unwrap();
     }