Bugfix/bloomfilters purge (#6089)

* remove all old bloomfilters upon starting binary

* remove old bloomfilter file upon purging secondary data

Cargo.lock

@@ -2262,7 +2262,7 @@ dependencies = [
  "libc",
  "option-ext",
  "redox_users",
- "windows-sys 0.61.1",
+ "windows-sys 0.59.0",
 ]
 
 [[package]]
@@ -8777,7 +8777,7 @@ checksum = "a4e608c6638b9c18977b00b475ac1f28d14e84b27d8d42f70e0bf1e3dec127ac"
 dependencies = [
  "getrandom 0.2.16",
  "libredox",
- "thiserror 2.0.17",
+ "thiserror 2.0.12",
 ]
 
 [[package]]

nym-node/src/node/replay_protection/background_task.rs

@@ -1,10 +1,8 @@
 // Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: GPL-3.0-only
 
+use crate::config::persistence::DEFAULT_RD_BLOOMFILTER_FLUSH_FILE_EXT;
 use crate::config::Config;
-use crate::config::persistence::{
-    DEFAULT_RD_BLOOMFILTER_FILE_EXT, DEFAULT_RD_BLOOMFILTER_FLUSH_FILE_EXT,
-};
 use crate::error::NymNodeError;
 use crate::node::replay_protection::bloomfilter::RotationFilter;
 use crate::node::replay_protection::helpers::parse_rotation_id_from_filename;
@@ -24,7 +22,6 @@ use tracing::{debug, error, info, trace, warn};
 
 // background task responsible for periodically flushing the bloomfilters to disk
 pub struct ReplayProtectionDiskFlush {
-    bloomfilters_directory: PathBuf,
     disk_flushing_rate: Duration,
 
     filters_manager: ReplayProtectionBloomfiltersManager,
@@ -124,8 +121,23 @@ impl ReplayProtectionDiskFlush {
                 None
             };
 
+        // if we have any other stored bloomfilters that are neither primary nor secondary,
+        // remove them - they are an artifact from an old version that had a bug in purging code
+        for (rotation_id, path) in filter_files {
+            if rotation_id == primary_key_rotation_id {
+                continue;
+            }
+            if let Some(secondary_key_rotation_id) = secondary_key_rotation_id {
+                if secondary_key_rotation_id == rotation_id {
+                    continue;
+                }
+            }
+            info!("stale bloomfilter for rotation {rotation_id} found at: {path:?}. it is going to get removed");
+            fs::remove_file(&path)
+                .map_err(|source| NymNodeError::BloomfilterIoFailure { source, path })?;
+        }
+
         Ok(ReplayProtectionDiskFlush {
-            bloomfilters_directory,
             disk_flushing_rate: config
                 .mixnet
                 .replay_protection
@@ -142,15 +154,12 @@ impl ReplayProtectionDiskFlush {
     }
 
     fn bloomfilter_filepath(&self, rotation_id: u32) -> PathBuf {
-        self.bloomfilters_directory
-            .join(format!("rot-{rotation_id}"))
-            .with_extension(DEFAULT_RD_BLOOMFILTER_FILE_EXT)
+        self.filters_manager.bloomfilter_filepath(rotation_id)
     }
 
     fn current_bloomfilter_being_flushed_filepath(&self, rotation_id: u32) -> PathBuf {
-        self.bloomfilters_directory
-            .join(format!("rot-{rotation_id}"))
-            .with_extension(DEFAULT_RD_BLOOMFILTER_FLUSH_FILE_EXT)
+        self.filters_manager
+            .current_bloomfilter_being_flushed_filepath(rotation_id)
     }
 
     pub(crate) fn bloomfilters_manager(&self) -> ReplayProtectionBloomfiltersManager {
@@ -213,7 +222,7 @@ impl ReplayProtectionDiskFlush {
     }
 
     async fn flush_filters_to_disk(&self) -> Result<(), NymNodeError> {
-        if let Some(parent) = self.bloomfilters_directory.parent() {
+        if let Some(parent) = self.filters_manager.bloomfilters_directory().parent() {
             fs::create_dir_all(parent).map_err(|source| NymNodeError::BloomfilterIoFailure {
                 source,
                 path: parent.to_path_buf(),

nym-node/src/node/replay_protection/bloomfilter.rs

@@ -4,6 +4,7 @@
 use crate::error::NymNodeError;
 use bloomfilter::Bloom;
 use nym_sphinx_types::REPLAY_TAG_SIZE;
+use nym_validator_client::models::KeyRotationId;
 use std::collections::HashMap;
 use std::fs::File;
 use std::io::Read;
@@ -12,7 +13,6 @@ use std::path::Path;
 use std::sync::{Arc, Mutex, PoisonError, TryLockError};
 use time::OffsetDateTime;
 use tracing::{error, info, warn};
-
 // auxiliary data associated with the bloomfilter to get some statistics from the time of its creation
 // this is needed in order to more accurately resize it upon reset
 
@@ -180,15 +180,16 @@ impl ReplayProtectionBloomfilters {
         Ok(())
     }
 
-    pub(crate) fn purge_secondary(&self) -> Result<(), NymNodeError> {
+    pub(crate) fn purge_secondary(&self) -> Result<Option<KeyRotationId>, NymNodeError> {
         let mut guard = self
             .inner
             .lock()
             .map_err(|_| NymNodeError::BloomfilterFailure {
                 message: "mutex got poisoned",
             })?;
-        guard.overlap = None;
-        Ok(())
+
+        let id = guard.overlap.take().map(|f| f.metadata.rotation_id);
+        Ok(id)
     }
 
     pub(crate) fn primary_metadata(

nym-node/src/node/replay_protection/manager.rs

@@ -1,6 +1,9 @@
 // Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: GPL-3.0-only
 
+use crate::config::persistence::{
+    DEFAULT_RD_BLOOMFILTER_FILE_EXT, DEFAULT_RD_BLOOMFILTER_FLUSH_FILE_EXT,
+};
 use crate::config::Config;
 use crate::error::NymNodeError;
 use crate::node::replay_protection::bloomfilter::{ReplayProtectionBloomfilters, RotationFilter};
@@ -8,12 +11,16 @@ use crate::node::replay_protection::items_in_bloomfilter;
 use human_repr::HumanCount;
 use nym_node_metrics::NymNodeMetrics;
 use std::cmp::max;
+use std::fs;
+use std::path::PathBuf;
 use std::time::Duration;
 use time::OffsetDateTime;
 use tracing::info;
 
 #[derive(Clone)]
 pub(crate) struct ReplayProtectionBloomfiltersManager {
+    bloomfilters_directory: PathBuf,
+
     target_fp_p: f64,
     minimum_bloomfilter_packets_per_second: usize,
     bloomfilter_size_multiplier: f64,
@@ -26,6 +33,7 @@ impl ReplayProtectionBloomfiltersManager {
     pub(crate) fn new_disabled(metrics: NymNodeMetrics) -> Self {
         // the exact config values are irrelevant as the filters will never be recreated
         ReplayProtectionBloomfiltersManager {
+            bloomfilters_directory: Default::default(),
             target_fp_p: 0.001,
             minimum_bloomfilter_packets_per_second: 1,
             bloomfilter_size_multiplier: 1.0,
@@ -41,6 +49,12 @@ impl ReplayProtectionBloomfiltersManager {
         metrics: NymNodeMetrics,
     ) -> Self {
         ReplayProtectionBloomfiltersManager {
+            bloomfilters_directory: config
+                .mixnet
+                .replay_protection
+                .storage_paths
+                .current_bloomfilters_directory
+                .clone(),
             target_fp_p: config.mixnet.replay_protection.debug.false_positive_rate,
             minimum_bloomfilter_packets_per_second: config
                 .mixnet
@@ -57,6 +71,22 @@ impl ReplayProtectionBloomfiltersManager {
         }
     }
 
+    pub(crate) fn bloomfilters_directory(&self) -> &PathBuf {
+        &self.bloomfilters_directory
+    }
+
+    pub(crate) fn bloomfilter_filepath(&self, rotation_id: u32) -> PathBuf {
+        self.bloomfilters_directory
+            .join(format!("rot-{rotation_id}"))
+            .with_extension(DEFAULT_RD_BLOOMFILTER_FILE_EXT)
+    }
+
+    pub(crate) fn current_bloomfilter_being_flushed_filepath(&self, rotation_id: u32) -> PathBuf {
+        self.bloomfilters_directory
+            .join(format!("rot-{rotation_id}"))
+            .with_extension(DEFAULT_RD_BLOOMFILTER_FLUSH_FILE_EXT)
+    }
+
     pub(crate) fn bloomfilters(&self) -> ReplayProtectionBloomfilters {
         self.filters.clone()
     }
@@ -70,7 +100,14 @@ impl ReplayProtectionBloomfiltersManager {
     }
 
     pub(crate) fn purge_secondary(&self) -> Result<(), NymNodeError> {
-        self.filters.purge_secondary()
+        // remove data in memory
+        if let Some(secondary_id) = self.filters.purge_secondary()? {
+            // remove data on disk (if applicable)
+            let path = self.bloomfilter_filepath(secondary_id);
+            fs::remove_file(&path)
+                .map_err(|source| NymNodeError::BloomfilterIoFailure { source, path })?;
+        }
+        Ok(())
     }
 
     pub(crate) fn promote_pre_announced(&self) -> Result<(), NymNodeError> {