rotating bloomfilters

nym-api/nym-api-requests/src/models.rs

@@ -21,7 +21,7 @@ use nym_mixnet_contract_common::nym_node::Role;
 use nym_mixnet_contract_common::reward_params::{Performance, RewardingParams};
 use nym_mixnet_contract_common::rewarding::RewardEstimate;
 use nym_mixnet_contract_common::{
-    EpochId, GatewayBond, IdentityKey, Interval, KeyRotationState, MixNode, NodeId, Percent,
+    EpochId, GatewayBond, IdentityKey, Interval, MixNode, NodeId, Percent,
 };
 use nym_network_defaults::{DEFAULT_MIX_LISTENING_PORT, DEFAULT_VERLOC_LISTENING_PORT};
 use nym_node_requests::api::v1::authenticator::models::Authenticator;
@@ -41,6 +41,7 @@ use thiserror::Error;
 use time::{Date, OffsetDateTime};
 use utoipa::{IntoParams, ToResponse, ToSchema};
 
+pub use nym_mixnet_contract_common::KeyRotationState;
 pub use nym_node_requests::api::v1::node::models::BinaryBuildInformationOwned;
 
 #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]

nym-node/src/node/key_rotation/controller.rs

@@ -1,24 +1,46 @@
 // Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
+// SPDX-License-Identifier: GPL-3.0-only
 
 use crate::node::key_rotation::manager::SphinxKeyManager;
 use crate::node::nym_apis_client::NymApisClient;
 use crate::node::replay_protection::bloomfilter::ReplayProtectionBloomfilters;
+use crate::node::replay_protection::manager::ReplayProtectionBloomfiltersManager;
 use futures::pin_mut;
 use nym_task::ShutdownToken;
 use nym_validator_client::client::NymApiClientExt;
-use nym_validator_client::models::KeyRotationInfoResponse;
+use nym_validator_client::models::{KeyRotationInfoResponse, KeyRotationState};
 use std::time::Duration;
 use time::OffsetDateTime;
 use tokio::time::{interval, sleep, Instant};
 use tracing::{error, info, trace, warn};
 
+struct RotationConfig {
+    epoch_duration: Duration,
+    rotation_state: KeyRotationState,
+}
+
+impl RotationConfig {
+    fn rotation_lifetime(&self) -> Duration {
+        (self.rotation_state.validity_epochs + 1) * self.epoch_duration
+    }
+}
+
+impl From<KeyRotationInfoResponse> for RotationConfig {
+    fn from(value: KeyRotationInfoResponse) -> Self {
+        RotationConfig {
+            epoch_duration: value.epoch_duration,
+            rotation_state: value.key_rotation_state,
+        }
+    }
+}
+
 pub(crate) struct KeyRotationController {
     // regular polling rate to catch any changes in the system config. they shouldn't happen too often
     // so the requests can be sent quite infrequently
     regular_polling_interval: Duration,
 
-    replay_protection_bloomfilters: ReplayProtectionBloomfilters,
+    rotation_config: RotationConfig,
+    replay_protection_manager: ReplayProtectionBloomfiltersManager,
     client: NymApisClient,
     managed_keys: SphinxKeyManager,
     shutdown_token: ShutdownToken,
@@ -54,8 +76,21 @@ enum KeyRotationActionState {
 }
 
 impl KeyRotationController {
-    pub(crate) fn new(client: NymApisClient, shutdown_token: ShutdownToken) -> Self {
+    pub(crate) async fn new(
+        regular_polling_interval: Duration,
+        client: NymApisClient,
+        replay_protection_manager: ReplayProtectionBloomfiltersManager,
+        managed_keys: SphinxKeyManager,
+        shutdown_token: ShutdownToken,
+    ) -> Self {
         todo!()
+        // KeyRotationController {
+        //     regular_polling_interval,
+        //     replay_protection_manager,
+        //     client,
+        //     managed_keys,
+        //     shutdown_token,
+        // }
     }
 
     async fn determine_next_action(&self) -> NextAction {
@@ -151,14 +186,41 @@ impl KeyRotationController {
                     Ok(key) => key,
                 };
 
+                if self
+                    .replay_protection_manager
+                    .allocate_pre_announced(rotation_id, self.rotation_config.rotation_lifetime())
+                    .is_err()
+                {
+                    // mutex poisoning - we have to exit
+                    self.shutdown_token.cancel();
+                    return;
+                }
                 self.client.broadcast_pre_announced_key(public_key).await;
             }
             KeyRotationActionState::SwapDefault => {
                 if let Err(err) = self.managed_keys.rotate_keys() {
                     error!("failed to perform sphinx key swap: {err}")
+                };
+                if self
+                    .replay_protection_manager
+                    .promote_pre_announced()
+                    .is_err()
+                {
+                    // mutex poisoning - we have to exit
+                    self.shutdown_token.cancel();
+                    return;
+                }
+            }
+            KeyRotationActionState::PurgeOld => {
+                if let Err(err) = self.managed_keys.remove_overlap_key() {
+                    error!("failed to remove old sphinx key: {err}");
+                };
+                if self.replay_protection_manager.purge_secondary().is_err() {
+                    // mutex poisoning - we have to exit
+                    self.shutdown_token.cancel();
+                    return;
                 }
             }
-            KeyRotationActionState::PurgeOld => {}
         }
     }
 

nym-node/src/node/nym_apis_client.rs

@@ -1,5 +1,5 @@
 // Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
+// SPDX-License-Identifier: GPL-3.0-only
 
 use crate::error::NymNodeError;
 use crate::node::key_rotation::key::SphinxPublicKey;

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

@@ -6,25 +6,27 @@ use bloomfilter::Bloom;
 use human_repr::HumanDuration;
 use nym_sphinx_types::REPLAY_TAG_SIZE;
 use std::collections::HashMap;
+use std::mem;
 use std::path::Path;
 use std::sync::{Arc, PoisonError, TryLockError};
 use tokio::fs::File;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::time::Instant;
-use tracing::{debug, info};
+use tracing::{debug, error, info};
 
 // 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
-struct ReplayProtectionBloomfilterMetadata {
+#[derive(Copy, Clone)]
+pub(crate) struct ReplayProtectionBloomfilterMetadata {
     // used in the unlikely case of epoch durations being changed. it doesn't really cost us anything
     // to include it, so might as well
-    creation_time: Instant,
+    pub(crate) creation_time: Instant,
 
     /// Number of packets that this node has received since startup, as recorded when this bloomfilter was created.
     /// Used for determining the approximate packet rate and thus number of entries in the bloomfilter
-    packets_received_at_creation: usize,
+    pub(crate) packets_received_at_creation: usize,
 
-    rotation_id: u32,
+    pub(crate) rotation_id: u32,
 }
 
 // it appears that now std Mutex is faster (or comparable) to parking_lot
@@ -73,6 +75,83 @@ impl ReplayProtectionBloomfilters {
         self.disabled
     }
 
+    pub(crate) fn allocate_pre_announced(
+        &self,
+        items_count: usize,
+        fp_p: f64,
+        packets_received_at_creation: usize,
+        rotation_id: u32,
+    ) -> Result<(), NymNodeError> {
+        // build the new filter
+        let filter =
+            Bloom::new_for_fp_rate(items_count, fp_p).map_err(NymNodeError::bloomfilter_failure)?;
+
+        let mut guard = self
+            .inner
+            .lock()
+            .map_err(|_| NymNodeError::BloomfilterFailure {
+                message: "mutex got poisoned",
+            })?;
+
+        guard.pre_announced = Some(RotationFilter {
+            metadata: ReplayProtectionBloomfilterMetadata {
+                creation_time: Instant::now(),
+                packets_received_at_creation,
+                rotation_id,
+            },
+            data: filter,
+        });
+        Ok(())
+    }
+
+    pub(crate) fn promote_pre_announced(&self) -> Result<(), NymNodeError> {
+        let mut guard = self
+            .inner
+            .lock()
+            .map_err(|_| NymNodeError::BloomfilterFailure {
+                message: "mutex got poisoned",
+            })?;
+
+        let Some(mut pre_announced) = guard.pre_announced.take() else {
+            error!("there was no pre-announced bloomfilter to promote");
+            return Ok(());
+        };
+
+        // pre_announced -> primary
+        // primary -> temp (pre_announced)
+        mem::swap(&mut guard.primary, &mut pre_announced);
+
+        // temp (pre_announced) -> secondary
+        guard.secondary = Some(pre_announced);
+        Ok(())
+    }
+
+    pub(crate) fn purge_secondary(&self) -> Result<(), NymNodeError> {
+        let mut guard = self
+            .inner
+            .lock()
+            .map_err(|_| NymNodeError::BloomfilterFailure {
+                message: "mutex got poisoned",
+            })?;
+        guard.secondary = None;
+        Ok(())
+    }
+
+    pub(crate) fn primary_metadata(
+        &self,
+    ) -> Result<ReplayProtectionBloomfilterMetadata, NymNodeError> {
+        let metadata = self
+            .inner
+            .lock()
+            .map_err(|_| NymNodeError::BloomfilterFailure {
+                message: "mutex got poisoned",
+            })?
+            .primary
+            .metadata;
+
+        Ok(metadata)
+    }
+
     pub(crate) fn reset(&self, items_count: usize, fp_p: f64) -> Result<(), NymNodeError> {
         // 1. build the new filter
         todo!()

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

@@ -0,0 +1,65 @@
+// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use crate::error::NymNodeError;
+use crate::node::replay_protection::bloomfilter::ReplayProtectionBloomfilters;
+use crate::node::replay_protection::items_in_bloomfilter;
+use human_repr::HumanCount;
+use nym_node_metrics::NymNodeMetrics;
+use std::cmp::max;
+use std::time::Duration;
+use tracing::info;
+
+pub(crate) struct ReplayProtectionBloomfiltersManager {
+    target_fp_p: f64,
+    minimum_bloomfilter_packets_per_second: usize,
+    bloomfilter_size_multiplier: f64,
+
+    metrics: NymNodeMetrics,
+    filters: ReplayProtectionBloomfilters,
+}
+
+impl ReplayProtectionBloomfiltersManager {
+    pub(crate) fn purge_secondary(&self) -> Result<(), NymNodeError> {
+        self.filters.purge_secondary()
+    }
+
+    pub(crate) fn promote_pre_announced(&self) -> Result<(), NymNodeError> {
+        self.filters.promote_pre_announced()
+    }
+
+    // TODO: actually do add some metrics
+    pub(crate) fn allocate_pre_announced(
+        &self,
+        rotation_id: u32,
+        rotation_lifetime: Duration,
+    ) -> Result<(), NymNodeError> {
+        // 1. estimated the number of items in the filter based on the extrapolated items received
+        // by the primary filter
+        let received = self.metrics.mixnet.ingress.forward_hop_packets_received()
+            + self.metrics.mixnet.ingress.final_hop_packets_received();
+
+        let primary = self.filters.primary_metadata()?;
+        let time_delta = primary.creation_time.elapsed();
+        let received_since_creation = received - primary.packets_received_at_creation;
+        let received_per_second =
+            (received_since_creation as f64 / time_delta.as_secs_f64()).round() as usize;
+
+        let bf_received = max(
+            received_per_second,
+            self.minimum_bloomfilter_packets_per_second,
+        );
+        let items_in_new_filter = items_in_bloomfilter(rotation_lifetime, bf_received);
+        let adjusted =
+            (items_in_new_filter as f64 * self.bloomfilter_size_multiplier).round() as usize;
+
+        info!(
+            "allocating new bloom filter. new expected number of packets: {} that preserve fp rate of {}",
+            adjusted.human_count_bare(),
+            self.target_fp_p
+        );
+
+        self.filters
+            .allocate_pre_announced(adjusted, self.target_fp_p, received, rotation_id)
+    }
+}

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

@@ -6,6 +6,7 @@ use std::time::Duration;
 
 pub(crate) mod background_task;
 pub(crate) mod bloomfilter;
+pub(crate) mod manager;
 
 pub fn bitmap_size(false_positive_rate: f64, items_in_filter: usize) -> usize {
     /// Equivalent to ln(1 / 2^ln(2)) = −ln^2(2)