From 8bcbb6988981d005f5516d04a5ebd4c0a536bc80 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Mon, 12 May 2025 14:13:49 +0100 Subject: [PATCH] flushing bloomfilters to disk and loading --- Cargo.toml | 2 +- nym-node/src/config/mod.rs | 64 ++-- nym-node/src/config/persistence.rs | 14 - nym-node/src/error.rs | 3 + nym-node/src/node/key_rotation/controller.rs | 3 +- nym-node/src/node/mod.rs | 22 +- .../node/replay_protection/background_task.rs | 306 ++++++++++------- .../src/node/replay_protection/bloomfilter.rs | 322 +++++++++++------- .../src/node/replay_protection/helpers.rs | 34 ++ .../src/node/replay_protection/manager.rs | 55 ++- nym-node/src/node/replay_protection/mod.rs | 1 + 11 files changed, 520 insertions(+), 306 deletions(-) create mode 100644 nym-node/src/node/replay_protection/helpers.rs diff --git a/Cargo.toml b/Cargo.toml index c2f5b97078..38a771deab 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -349,7 +349,6 @@ utoipauto = "0.2" uuid = "*" vergen = { version = "=8.3.1", default-features = false } walkdir = "2" -wasm-bindgen-test = "0.3.49" x25519-dalek = "2.0.0" zeroize = "1.7.0" @@ -397,6 +396,7 @@ serde-wasm-bindgen = "0.6.5" tsify = "0.4.5" wasm-bindgen = "0.2.99" wasm-bindgen-futures = "0.4.49" +wasm-bindgen-test = "0.3.49" wasmtimer = "0.4.1" web-sys = "0.3.76" diff --git a/nym-node/src/config/mod.rs b/nym-node/src/config/mod.rs index 3b9d251351..f86eefba5b 100644 --- a/nym-node/src/config/mod.rs +++ b/nym-node/src/config/mod.rs @@ -6,6 +6,7 @@ use crate::config::template::CONFIG_TEMPLATE; use crate::error::NymNodeError; use celes::Country; use clap::ValueEnum; +use human_repr::HumanCount; use nym_bin_common::logging::LoggingSettings; use nym_config::defaults::{ mainnet, var_names, DEFAULT_MIX_LISTENING_PORT, DEFAULT_NYM_NODE_HTTP_PORT, @@ -26,6 +27,7 @@ use std::fmt::{Display, Formatter}; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; use std::path::{Path, PathBuf}; use std::time::Duration; +use sysinfo::System; use tracing::{debug, error}; use url::Url; @@ -42,6 +44,7 @@ pub mod upgrade_helpers; pub use crate::config::gateway_tasks::GatewayTasksConfig; pub use crate::config::metrics::MetricsConfig; pub use crate::config::service_providers::ServiceProvidersConfig; +use crate::node::replay_protection::{bitmap_size, items_in_bloomfilter}; const DEFAULT_NYMNODES_DIR: &str = "nym-nodes"; @@ -662,35 +665,38 @@ impl ReplayProtectionDebug { pub const DEFAULT_BLOOMFILTER_MINIMUM_PACKETS_PER_SECOND_SIZE: usize = 200; pub fn validate(&self) -> Result<(), NymNodeError> { - todo!() - // if self.false_positive_rate >= 1.0 || self.false_positive_rate <= 0.0 { - // return Err(NymNodeError::config_validation_failure( - // "false positive rate for replay detection can't be larger than (or equal to) 1 or smaller than (or equal to) 0", - // )); - // } - // - // let items_in_filter = items_in_bloomfilter( - // self.bloomfilter_reset_rate, - // self.initial_expected_packets_per_second, - // ); - // let bitmap_size = bitmap_size(self.false_positive_rate, items_in_filter); - // let bloomfilter_size = bitmap_size / 8; - // - // let mut sys_info = System::new(); - // sys_info.refresh_memory(); - // - // // we'll need 2x size of the bloomfilter - // // as during key transition we'll have to simultaneously use two filters - // // plus we also need to make a memcopy during disk flush - // let required_memory = 2 * bloomfilter_size; - // - // let memory = sys_info.available_memory(); - // if (memory as usize) < required_memory { - // return Err(NymNodeError::config_validation_failure( - // format!("system does not have sufficient memory to allocate required replay protection bloomfilters. {} is available whilst at least {} is needed",memory.human_count_bytes(), required_memory.human_count_bytes()))); - // } - // - // Ok(()) + if self.false_positive_rate >= 1.0 || self.false_positive_rate <= 0.0 { + return Err(NymNodeError::config_validation_failure( + "false positive rate for replay detection can't be larger than (or equal to) 1 or smaller than (or equal to) 0", + )); + } + + // ideally we would have pulled the exact information from the network, + // but making async calls really doesn't play around with this method + // so we do second best: assume 24h rotation with 1h overlap (which realistically won't ever change) + + let items_in_filter = items_in_bloomfilter( + Duration::from_secs(25 * 60 * 60), + self.initial_expected_packets_per_second, + ); + let bitmap_size = bitmap_size(self.false_positive_rate, items_in_filter); + let bloomfilter_size = bitmap_size / 8; + + let mut sys_info = System::new(); + sys_info.refresh_memory(); + + // we'll need 2x size of the bloomfilter + // as during key transition we'll have to simultaneously use two filters + // plus we also need to make a memcopy during disk flush + let required_memory = 2 * bloomfilter_size; + + let memory = sys_info.available_memory(); + if (memory as usize) < required_memory { + return Err(NymNodeError::config_validation_failure( + format!("system does not have sufficient memory to allocate required replay protection bloomfilters. {} is available whilst at least {} is needed",memory.human_count_bytes(), required_memory.human_count_bytes()))); + } + + Ok(()) } } diff --git a/nym-node/src/config/persistence.rs b/nym-node/src/config/persistence.rs index 57da186548..087bcd5b76 100644 --- a/nym-node/src/config/persistence.rs +++ b/nym-node/src/config/persistence.rs @@ -496,20 +496,6 @@ pub struct ReplayProtectionPaths { pub current_bloomfilters_directory: PathBuf, } -impl ReplayProtectionPaths { - pub fn bloomfilter_filepath(&self, rotation_id: u32) -> PathBuf { - self.current_bloomfilters_directory - .join(format!("rot-{rotation_id}")) - .with_extension(DEFAULT_RD_BLOOMFILTER_FILE_EXT) - } - - pub fn current_bloomfilter_being_flushed_filepath(&self, rotation_id: u32) -> PathBuf { - self.current_bloomfilters_directory - .join(format!("rot-{rotation_id}")) - .with_extension(DEFAULT_RD_BLOOMFILTER_FLUSH_FILE_EXT) - } -} - impl ReplayProtectionPaths { pub fn new>(data_dir: P) -> Self { ReplayProtectionPaths { diff --git a/nym-node/src/error.rs b/nym-node/src/error.rs index 5eda797d41..dee03e107e 100644 --- a/nym-node/src/error.rs +++ b/nym-node/src/error.rs @@ -198,6 +198,9 @@ pub enum NymNodeError { #[error("failed to save/load the bloomfilter: {source} using path: {}", path.display())] BloomfilterIoFailure { source: io::Error, path: PathBuf }, + #[error("failed to deserialise bloomfilter metadata")] + BloomfilterMetadataDeserialisationFailure, + #[error(transparent)] GatewayFailure(Box), diff --git a/nym-node/src/node/key_rotation/controller.rs b/nym-node/src/node/key_rotation/controller.rs index 172e6cfda7..e4fd372f30 100644 --- a/nym-node/src/node/key_rotation/controller.rs +++ b/nym-node/src/node/key_rotation/controller.rs @@ -4,7 +4,6 @@ use crate::config::Config; 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_node_metrics::NymNodeMetrics; @@ -80,7 +79,7 @@ impl KeyRotationController { pub(crate) fn new( config: &Config, client: NymApisClient, - replay_protection_manager: ReplayProtectionBloomfilters, + replay_protection_manager: ReplayProtectionBloomfiltersManager, metrics: NymNodeMetrics, managed_keys: SphinxKeyManager, shutdown_token: ShutdownToken, diff --git a/nym-node/src/node/mod.rs b/nym-node/src/node/mod.rs index 20f3078a1f..ddd43d79b0 100644 --- a/nym-node/src/node/mod.rs +++ b/nym-node/src/node/mod.rs @@ -33,6 +33,7 @@ use crate::node::mixnet::SharedFinalHopData; use crate::node::nym_apis_client::NymApisClient; use crate::node::replay_protection::background_task::ReplayProtectionDiskFlush; use crate::node::replay_protection::bloomfilter::ReplayProtectionBloomfilters; +use crate::node::replay_protection::manager::ReplayProtectionBloomfiltersManager; use crate::node::routing_filter::{OpenFilter, RoutingFilter}; use crate::node::shared_network::{ CachedNetwork, CachedTopologyProvider, LocalGatewayNode, NetworkRefresher, @@ -953,9 +954,11 @@ impl NymNode { pub(crate) async fn setup_replay_detection( &self, - ) -> Result { + ) -> Result { if self.config.mixnet.replay_protection.debug.unsafe_disabled { - return Ok(ReplayProtectionBloomfilters::new_disabled()); + return Ok(ReplayProtectionBloomfiltersManager::new_disabled( + self.metrics.clone(), + )); } // create the background task for the bloomfilter @@ -965,15 +968,16 @@ impl NymNode { &self.config, sphinx_keys.keys.primary_key_rotation_id(), sphinx_keys.keys.secondary_key_rotation_id(), + self.metrics.clone(), self.shutdown_manager .clone_token("replay-detection-background-flush"), ) .await?; - let replay_protection_bloomfilter = replay_detection_background.global_bloomfilters(); + let bloomfilters_manager = replay_detection_background.bloomfilters_manager(); self.shutdown_manager .spawn(async move { replay_detection_background.run().await }); - Ok(replay_protection_bloomfilter) + Ok(bloomfilters_manager) } // I'm assuming this will be needed in other places, so it's explicitly extracted @@ -996,13 +1000,13 @@ impl NymNode { pub(crate) async fn setup_key_rotation( &mut self, nym_apis_client: NymApisClient, - replay_protection_bloomfilters: ReplayProtectionBloomfilters, + replay_protection_manager: ReplayProtectionBloomfiltersManager, ) -> Result<(), NymNodeError> { let managed_keys = self.take_managed_sphinx_keys()?; let rotation_controller = KeyRotationController::new( &self.config, nym_apis_client, - replay_protection_bloomfilters, + replay_protection_manager, self.metrics.clone(), managed_keys, self.shutdown_manager.clone_token("key-rotation-controller"), @@ -1119,18 +1123,18 @@ impl NymNode { let active_clients_store = ActiveClientsStore::new(); let nym_apis_client = self.setup_nym_apis_client()?; - let replay_protection_bloomfilter = self.setup_replay_detection().await?; + let bloomfilters_manager = self.setup_replay_detection().await?; let (mix_packet_sender, active_egress_mixnet_connections) = self .start_mixnet_listener( &active_clients_store, - replay_protection_bloomfilter.clone(), + bloomfilters_manager.bloomfilters(), network_refresher.routing_filter(), self.shutdown_manager.clone_token("mixnet-traffic"), ) .await?; - self.setup_key_rotation(nym_apis_client, replay_protection_bloomfilter) + self.setup_key_rotation(nym_apis_client, bloomfilters_manager) .await?; let metrics_sender = self.setup_metrics_backend( diff --git a/nym-node/src/node/replay_protection/background_task.rs b/nym-node/src/node/replay_protection/background_task.rs index bcd77b2fec..1c8d6ad033 100644 --- a/nym-node/src/node/replay_protection/background_task.rs +++ b/nym-node/src/node/replay_protection/background_task.rs @@ -1,67 +1,33 @@ // Copyright 2025 - Nym Technologies SA // 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; +use crate::node::replay_protection::bloomfilter::RotationFilter; +use crate::node::replay_protection::helpers::parse_rotation_id_from_filename; +use crate::node::replay_protection::items_in_bloomfilter; +use crate::node::replay_protection::manager::ReplayProtectionBloomfiltersManager; +use human_repr::HumanDuration; +use nym_node_metrics::NymNodeMetrics; use nym_task::ShutdownToken; +use std::collections::HashMap; use std::fs; use std::path::PathBuf; use std::time::Duration; -use tokio::time::interval; -use tracing::{error, info, trace, warn}; - -struct ReplayProtectionBackgroundTaskConfig { - current_bloomfilter_path: PathBuf, - current_bloomfilter_temp_flush_path: PathBuf, - - false_positive_rate: f64, - filter_reset_rate: Duration, - disk_flushing_rate: Duration, - bloomfilter_size_multiplier: f64, - minimum_bloomfilter_packets_per_second: usize, -} - -impl From<&Config> for ReplayProtectionBackgroundTaskConfig { - fn from(config: &Config) -> Self { - todo!() - // ReplayProtectionBackgroundTaskConfig { - // current_bloomfilter_path: config - // .mixnet - // .replay_protection - // .storage_paths - // .current_bloomfilter_filepath(), - // current_bloomfilter_temp_flush_path: config - // .mixnet - // .replay_protection - // .storage_paths - // .current_bloomfilter_being_flushed_filepath(), - // false_positive_rate: config.mixnet.replay_protection.debug.false_positive_rate, - // filter_reset_rate: config.mixnet.replay_protection.debug.bloomfilter_reset_rate, - // disk_flushing_rate: config - // .mixnet - // .replay_protection - // .debug - // .bloomfilter_disk_flushing_rate, - // bloomfilter_size_multiplier: config - // .mixnet - // .replay_protection - // .debug - // .bloomfilter_size_multiplier, - // minimum_bloomfilter_packets_per_second: config - // .mixnet - // .replay_protection - // .debug - // .bloomfilter_minimum_packets_per_second_size, - // } - } -} +use tokio::fs::File; +use tokio::io::AsyncWriteExt; +use tokio::time::{interval, Instant}; +use tracing::{debug, error, info, trace, warn}; // background task responsible for periodically flushing the bloomfilters to disk pub struct ReplayProtectionDiskFlush { - config: ReplayProtectionBackgroundTaskConfig, + bloomfilters_directory: PathBuf, + disk_flushing_rate: Duration, - filters: ReplayProtectionBloomfilters, + filters_manager: ReplayProtectionBloomfiltersManager, shutdown_token: ShutdownToken, } @@ -70,86 +36,194 @@ impl ReplayProtectionDiskFlush { config: &Config, primary_key_rotation_id: u32, secondary_key_rotation_id: Option, + metrics: NymNodeMetrics, shutdown_token: ShutdownToken, ) -> Result { - // based on current rotation id, figure out which filter is which and also purge old ones, if exist. - todo!() - // - // let task_config: ReplayProtectionBackgroundTaskConfig = config.into(); - // - // if task_config.current_bloomfilter_temp_flush_path.exists() { - // error!( - // "bloomfilter didn't get successfully flushed to disk and its data got corrupted" - // ); - // fs::remove_file(&task_config.current_bloomfilter_temp_flush_path).map_err(|source| { - // NymNodeError::BloomfilterIoFailure { - // source, - // path: task_config.current_bloomfilter_temp_flush_path.clone(), - // } - // })? - // } - // - // // if there's nothing on disk, we must create a new filter - // let bloomfilter = if task_config.current_bloomfilter_path.exists() { - // ReplayProtectionBloomfilters::load(&task_config.current_bloomfilter_path).await? - // } else { - // let bf_items = items_in_bloomfilter( - // task_config.filter_reset_rate, - // config - // .mixnet - // .replay_protection - // .debug - // .initial_expected_packets_per_second, - // ); - // - // ReplayProtectionBloomfilters::new_empty(bf_items, task_config.false_positive_rate)? - // }; - // - // Ok(ReplayProtectionDiskFlush { - // config: task_config, - // filters: bloomfilter, - // shutdown_token, - // }) - } + let bloomfilters_directory = config + .mixnet + .replay_protection + .storage_paths + .current_bloomfilters_directory + .clone(); - pub(crate) fn global_bloomfilters(&self) -> ReplayProtectionBloomfilters { - self.filters.clone() - } + let dir_read_err = |source| NymNodeError::BloomfilterIoFailure { + source, + path: bloomfilters_directory.clone(), + }; - async fn flush_to_disk(&self) -> Result<(), NymNodeError> { - if let Some(temp_parent) = self.config.current_bloomfilter_temp_flush_path.parent() { - fs::create_dir_all(temp_parent).map_err(|source| { - NymNodeError::BloomfilterIoFailure { - source, - path: temp_parent.to_path_buf(), + let available_filters_dir = fs::read_dir(&bloomfilters_directory).map_err(dir_read_err)?; + + // figure out what bloomfilters we have available on disk + let mut filter_files = HashMap::new(); + for entry in available_filters_dir.into_iter() { + let entry = entry.map_err(dir_read_err)?; + let path = entry.path(); + + let Some(rotation) = entry + .file_name() + .to_str() + .and_then(parse_rotation_id_from_filename) + else { + warn!("invalid bloomfilter file at '{}'", path.display()); + continue; + }; + + // if any bloomfilter has the temp extension, we can't trust its data as it hasn't completed the flush + if let Some(ext) = entry.path().extension() { + if ext == DEFAULT_RD_BLOOMFILTER_FLUSH_FILE_EXT { + error!( + "bloomfilter {rotation} didn't get successfully flushed to disk and its data got corrupted" + ); + fs::remove_file(&path) + .map_err(|source| NymNodeError::BloomfilterIoFailure { source, path })?; + continue; } - })? - } - if let Some(current_parent) = self.config.current_bloomfilter_temp_flush_path.parent() { - fs::create_dir_all(current_parent).map_err(|source| { - NymNodeError::BloomfilterIoFailure { - source, - path: current_parent.to_path_buf(), - } - })? + } + + filter_files.insert(rotation, path); } + let rebuild_items_in_filter = items_in_bloomfilter( + Duration::from_secs(25 * 60 * 60), + config + .mixnet + .replay_protection + .debug + .initial_expected_packets_per_second, + ); + let fp_r = config.mixnet.replay_protection.debug.false_positive_rate; + + // if filters do not exist on disk, we must make new ones + let primary_bloomfilter = match filter_files.get(&primary_key_rotation_id) { + Some(primary_path) => RotationFilter::load(primary_path)?, + None => { + info!("no stored bloomfilter for rotation {primary_key_rotation_id}"); + RotationFilter::new(rebuild_items_in_filter, fp_r, 0, primary_key_rotation_id)? + } + }; + + let secondary_bloomfilter = + if let Some(secondary_key_rotation_id) = secondary_key_rotation_id { + match filter_files.get(&secondary_key_rotation_id) { + Some(secondary_path) => Some(RotationFilter::load(secondary_path)?), + None => { + info!("no stored bloomfilter for rotation {secondary_key_rotation_id}"); + Some(RotationFilter::new( + rebuild_items_in_filter, + fp_r, + 0, + secondary_key_rotation_id, + )?) + } + } + } else { + None + }; + + Ok(ReplayProtectionDiskFlush { + bloomfilters_directory, + disk_flushing_rate: config + .mixnet + .replay_protection + .debug + .bloomfilter_disk_flushing_rate, + filters_manager: ReplayProtectionBloomfiltersManager::new( + config, + primary_bloomfilter, + secondary_bloomfilter, + metrics, + ), + shutdown_token, + }) + } + + fn bloomfilter_filepath(&self, rotation_id: u32) -> PathBuf { + self.bloomfilters_directory + .join(format!("rot-{rotation_id}")) + .with_extension(DEFAULT_RD_BLOOMFILTER_FILE_EXT) + } + + 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_manager(&self) -> ReplayProtectionBloomfiltersManager { + self.filters_manager.clone() + } + + async fn flush(&self, data: Vec, rotation_id: u32) -> Result<(), NymNodeError> { // because it takes a while to actually write the file to disk, // we first write bytes to temporary location, // and then we move it to the correct path - let temp = &self.config.current_bloomfilter_temp_flush_path; - self.filters.flush_to_disk(temp).await?; - fs::rename(temp, &self.config.current_bloomfilter_path).map_err(|source| { + let temp_path = self.current_bloomfilter_being_flushed_filepath(rotation_id); + let final_path = self.bloomfilter_filepath(rotation_id); + debug!("flushing replay protection bloomfilter {rotation_id} to disk..."); + let start = Instant::now(); + + let mut file = File::create(&temp_path).await.map_err(|source| { NymNodeError::BloomfilterIoFailure { source, - path: self.config.current_bloomfilter_path.clone(), + path: temp_path.clone(), } })?; + + file.write_all(&data) + .await + .map_err(|source| NymNodeError::BloomfilterIoFailure { + source, + path: temp_path.to_path_buf(), + })?; + + fs::rename(temp_path, &final_path).map_err(|source| { + NymNodeError::BloomfilterIoFailure { + source, + path: final_path, + } + })?; + + let elapsed = start.elapsed(); + + info!( + "flushed replay protection bloomfilter {rotation_id} to disk. it took: {}", + elapsed.human_duration() + ); + + Ok(()) + } + + // average HDD has the write speed of ~80MB/s so a 2GB bloomfilter would take almost 30s to write... + // and this function is explicitly async and using tokio's async operations, because otherwise + // we'd have to go through the whole hassle of using spawn_blocking and awaiting that one instead + async fn flush_primary(&self) -> Result<(), NymNodeError> { + let (bytes, id) = self.filters_manager.primary_bytes_and_id()?; + self.flush(bytes, id).await + } + + async fn flush_secondary(&self) -> Result<(), NymNodeError> { + let Some((bytes, id)) = self.filters_manager.secondary_bytes_and_id()? else { + return Ok(()); + }; + self.flush(bytes, id).await + } + + async fn flush_filters_to_disk(&self) -> Result<(), NymNodeError> { + if let Some(parent) = self.bloomfilters_directory.parent() { + fs::create_dir_all(parent).map_err(|source| NymNodeError::BloomfilterIoFailure { + source, + path: parent.to_path_buf(), + })? + } + + self.flush_primary().await?; + self.flush_secondary().await?; + Ok(()) } pub(crate) async fn run(&mut self) { - let mut flush_timer = interval(self.config.disk_flushing_rate); + let mut flush_timer = interval(self.disk_flushing_rate); flush_timer.reset(); loop { @@ -160,7 +234,7 @@ impl ReplayProtectionDiskFlush { break; } _ = flush_timer.tick() => { - if let Err(err) = self.flush_to_disk().await { + if let Err(err) = self.flush_filters_to_disk().await { error!("failed to flush bloomfilter to disk: {err}") } } @@ -168,8 +242,8 @@ impl ReplayProtectionDiskFlush { } info!("SHUTDOWN: flushing replay detection bloomfilter to disk. this might take a while. DO NOT INTERRUPT THIS PROCESS"); - if let Err(err) = self.flush_to_disk().await { - warn!("failed to flush replay detection bloom filter on shutdown: {err}"); + if let Err(err) = self.flush_filters_to_disk().await { + warn!("failed to flush replay detection bloom filters on shutdown: {err}"); } } } diff --git a/nym-node/src/node/replay_protection/bloomfilter.rs b/nym-node/src/node/replay_protection/bloomfilter.rs index d375bb06fc..47b12e0b31 100644 --- a/nym-node/src/node/replay_protection/bloomfilter.rs +++ b/nym-node/src/node/replay_protection/bloomfilter.rs @@ -5,19 +5,22 @@ use crate::error::NymNodeError; use bloomfilter::Bloom; use nym_sphinx_types::REPLAY_TAG_SIZE; use std::collections::HashMap; +use std::fs::File; +use std::io::Read; use std::mem; use std::path::Path; -use std::sync::{Arc, PoisonError, TryLockError}; -use tokio::time::Instant; -use tracing::error; +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 + #[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 - pub(crate) creation_time: Instant, + pub(crate) creation_time: OffsetDateTime, /// 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 @@ -26,25 +29,80 @@ pub(crate) struct ReplayProtectionBloomfilterMetadata { pub(crate) rotation_id: u32, } +impl ReplayProtectionBloomfilterMetadata { + const SERIALIZED_LEN: usize = size_of::() + size_of::() + size_of::(); + + // UNIX_TIMESTAMP || PACKETS_RECEIVED || ROTATION_ID + pub(crate) fn bytes(&self) -> Vec { + self.creation_time + .unix_timestamp() + .to_be_bytes() + .into_iter() + .chain((self.packets_received_at_creation as u64).to_be_bytes()) + .chain(self.rotation_id.to_be_bytes()) + .collect() + } + pub(crate) fn try_from_bytes(bytes: &[u8]) -> Result { + if bytes.len() != Self::SERIALIZED_LEN { + return Err(NymNodeError::BloomfilterMetadataDeserialisationFailure); + } + + // SAFETY: we just checked we have correct number of bytes + #[allow(clippy::unwrap_used)] + let creation_timestamp = i64::from_be_bytes(bytes[0..8].try_into().unwrap()); + + #[allow(clippy::unwrap_used)] + let packets_received_at_creation = + u64::from_be_bytes(bytes[8..16].try_into().unwrap()) as usize; + + #[allow(clippy::unwrap_used)] + let rotation_id = u32::from_be_bytes(bytes[16..].try_into().unwrap()); + + Ok(ReplayProtectionBloomfilterMetadata { + creation_time: OffsetDateTime::from_unix_timestamp(creation_timestamp) + .map_err(|_| NymNodeError::BloomfilterMetadataDeserialisationFailure)?, + packets_received_at_creation, + rotation_id, + }) + } +} + // it appears that now std Mutex is faster (or comparable) to parking_lot // in high contention situations: https://github.com/rust-lang/rust/pull/95035#issuecomment-1073966631 // (tokio's async Mutex has too much overhead due to the number of access required) #[derive(Clone)] pub(crate) struct ReplayProtectionBloomfilters { disabled: bool, - inner: Arc>, + inner: Arc>, } impl ReplayProtectionBloomfilters { - pub(crate) fn new_empty(items_count: usize, fp_p: f64) -> Result { - todo!() - // Ok(ReplayProtectionBloomfilter { - // disabled: false, - // inner: Arc::new(std::sync::Mutex::new(ReplayProtectionBloomfilterInner { - // current_filter: Bloom::new_for_fp_rate(items_count, fp_p) - // .map_err(NymNodeError::bloomfilter_failure)?, - // })), - // }) + pub(crate) fn new(primary: RotationFilter, secondary: Option) -> Self { + // figure out if the secondary filter is the overlap or pre_announced filter + let primary_id = primary.metadata.rotation_id; + let (overlap, pre_announced) = match secondary { + None => (None, None), + Some(secondary_filter) => { + let secondary_id = secondary_filter.metadata.rotation_id; + if secondary_id == primary_id + 1 { + (None, Some(secondary_filter)) + } else if secondary_id == primary_id - 1 { + (Some(secondary_filter), None) + } else { + warn!("{secondary_id} is not valid for either pre_announced or overlap bloomfilter given primary rotation of {primary_id}"); + (None, None) + } + } + }; + + ReplayProtectionBloomfilters { + disabled: false, + inner: Arc::new(Mutex::new(ReplayProtectionBloomfiltersInner { + primary, + overlap, + pre_announced, + })), + } } // SAFETY: the hardcoded values of 1,1 are valid @@ -56,13 +114,13 @@ impl ReplayProtectionBloomfilters { inner: Arc::new(std::sync::Mutex::new(ReplayProtectionBloomfiltersInner { primary: RotationFilter { metadata: ReplayProtectionBloomfilterMetadata { - creation_time: Instant::now(), + creation_time: OffsetDateTime::now_utc(), packets_received_at_creation: 0, rotation_id: u32::MAX, }, data: Bloom::new(1, 1).unwrap(), }, - secondary: None, + overlap: None, pre_announced: None, })), } @@ -79,10 +137,6 @@ impl ReplayProtectionBloomfilters { 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() @@ -90,14 +144,12 @@ impl ReplayProtectionBloomfilters { message: "mutex got poisoned", })?; - guard.pre_announced = Some(RotationFilter { - metadata: ReplayProtectionBloomfilterMetadata { - creation_time: Instant::now(), - packets_received_at_creation, - rotation_id, - }, - data: filter, - }); + guard.pre_announced = Some(RotationFilter::new( + items_count, + fp_p, + packets_received_at_creation, + rotation_id, + )?); Ok(()) } @@ -119,7 +171,7 @@ impl ReplayProtectionBloomfilters { mem::swap(&mut guard.primary, &mut pre_announced); // temp (pre_announced) -> secondary - guard.secondary = Some(pre_announced); + guard.overlap = Some(pre_announced); Ok(()) } @@ -130,7 +182,7 @@ impl ReplayProtectionBloomfilters { .map_err(|_| NymNodeError::BloomfilterFailure { message: "mutex got poisoned", })?; - guard.secondary = None; + guard.overlap = None; Ok(()) } @@ -149,97 +201,41 @@ impl ReplayProtectionBloomfilters { Ok(metadata) } - pub(crate) fn reset(&self, items_count: usize, fp_p: f64) -> Result<(), NymNodeError> { - // 1. build the new filter - todo!() - // let new_inner = ReplayProtectionBloomfilterInner { - // current_filter: Bloom::new_for_fp_rate(items_count, fp_p) - // .map_err(NymNodeError::bloomfilter_failure)?, - // }; - // - // // 2. swap it - // let mut guard = self - // .inner - // .lock() - // .map_err(|_| NymNodeError::BloomfilterFailure { - // message: "mutex got poisoned", - // })?; - // - // *guard = new_inner; - // Ok(()) + pub(crate) fn primary_bytes_and_id(&self) -> Result<(Vec, u32), NymNodeError> { + let guard = self + .inner + .lock() + .map_err(|_| NymNodeError::BloomfilterFailure { + message: "mutex got poisoned", + })?; + + let id = guard.primary.metadata.rotation_id; + let bytes = guard.primary.data.to_bytes(); + Ok((bytes, id)) } - // NOTE: with key rotations we'll have to check whether the file is still valid and which - // key it corresponds to, but that's a future problem - pub(crate) async fn load>(path: P) -> Result { - todo!() - // info!("attempting to load prior replay detection bloomfilter..."); - // let path = path.as_ref(); - // let mut file = - // File::open(path) - // .await - // .map_err(|source| NymNodeError::BloomfilterIoFailure { - // source, - // path: path.to_path_buf(), - // })?; - // - // let mut buf = Vec::new(); - // file.read_to_end(&mut buf) - // .await - // .map_err(|source| NymNodeError::BloomfilterIoFailure { - // source, - // path: path.to_path_buf(), - // })?; - // - // Ok(ReplayProtectionBloomfilter { - // disabled: false, - // inner: Arc::new(std::sync::Mutex::new(ReplayProtectionBloomfilterInner { - // current_filter: Bloom::from_bytes(buf) - // .map_err(NymNodeError::bloomfilter_failure)?, - // })), - // }) - } + pub(crate) fn secondary_bytes_and_id(&self) -> Result, u32)>, NymNodeError> { + let guard = self + .inner + .lock() + .map_err(|_| NymNodeError::BloomfilterFailure { + message: "mutex got poisoned", + })?; - // average HDD has the write speed of ~80MB/s so a 2GB bloomfilter would take almost 30s to write... - // and this function is explicitly async and using tokio's async operations, because otherwise - // we'd have to go through the whole hassle of using spawn_blocking and awaiting that one instead - pub(crate) async fn flush_to_disk>(&self, path: P) -> Result<(), NymNodeError> { - todo!() - // debug!("flushing replay protection bloomfilter to disk..."); - // let start = Instant::now(); - // let path = path.as_ref(); - // - // let mut file = - // File::create(path) - // .await - // .map_err(|source| NymNodeError::BloomfilterIoFailure { - // source, - // path: path.to_path_buf(), - // })?; - // let data = self.bytes().map_err(|_| NymNodeError::BloomfilterFailure { - // message: "mutex got poisoned", - // })?; - // file.write_all(&data) - // .await - // .map_err(|source| NymNodeError::BloomfilterIoFailure { - // source, - // path: path.to_path_buf(), - // })?; - // - // let elapsed = start.elapsed(); - // - // info!( - // "flushed replay protection bloomfilter to disk. it took: {}", - // elapsed.human_duration() - // ); - // - // Ok(()) - } -} + let secondary = match guard.overlap.as_ref() { + Some(overlap) => overlap, + None => { + let Some(pre_announced) = guard.pre_announced.as_ref() else { + return Ok(None); + }; + pre_announced + } + }; -struct RotationFilter { - metadata: ReplayProtectionBloomfilterMetadata, - data: Bloom<[u8; REPLAY_TAG_SIZE]>, + let id = secondary.metadata.rotation_id; + let bytes = secondary.data.to_bytes(); + Ok(Some((bytes, id))) + } } impl ReplayProtectionBloomfilters { @@ -266,28 +262,19 @@ impl ReplayProtectionBloomfilters { Ok(guard.batch_check_and_set(&reply_tags)) } - - // due to the size of the bloomfilter, extra caution has to be applied when using this method - // note: we're not getting reference to bytes as this method is used when flushing data to the disk - // (which takes ~30s) and we can't block the mutex for that long. - fn bytes(&self) -> Result, PoisonError<()>> { - todo!() - // let guard = self.inner.lock().map_err(|_| PoisonError::new(()))?; - // Ok(guard.current_filter.to_bytes()) - } } struct ReplayProtectionBloomfiltersInner { primary: RotationFilter, // don't worry, we'll never have 3 active filters at once, - // we will either have a secondary (during the first epoch of a new rotation) + // we will either have a overlap (during the first epoch of a new rotation) // or a pre_announced (during the last epoch of the current rotation) // during epoch transition, the following change will happen: - // primary -> secondary + // primary -> overlap // pre_announced -> primary // I'm not using an enum because it's easier to reason about those as separate fields - secondary: Option, + overlap: Option, pre_announced: Option, } @@ -300,9 +287,9 @@ impl ReplayProtectionBloomfiltersInner { for (&rotation_id, reply_tags) in reply_tags { // try to 'find' the relevant filter. we might be doing 3 reads here, but realistically it's // going to be 'primary' most of the time and even if not, it's just few ns of overhead... - let mut filter = if self.primary.metadata.rotation_id == rotation_id { + let filter = if self.primary.metadata.rotation_id == rotation_id { Some(&mut self.primary.data) - } else if let Some(secondary) = &mut self.secondary { + } else if let Some(secondary) = &mut self.overlap { // if let chaining won't be stable until 1.88 so we have to do the Option workaround if secondary.metadata.rotation_id == rotation_id { Some(&mut secondary.data) @@ -319,7 +306,7 @@ impl ReplayProtectionBloomfiltersInner { None }; - let Some(mut filter) = filter else { + let Some(filter) = filter else { // if we've received a packet from an unknown rotation, it most likely means it has been replayed // from an older rotation, so mark it as such result.insert(rotation_id, vec![false; reply_tags.len()]); @@ -336,3 +323,74 @@ impl ReplayProtectionBloomfiltersInner { result } } + +pub(crate) struct RotationFilter { + metadata: ReplayProtectionBloomfilterMetadata, + data: Bloom<[u8; REPLAY_TAG_SIZE]>, +} + +impl RotationFilter { + pub(crate) fn new( + items_count: usize, + fp_p: f64, + packets_received_at_creation: usize, + rotation_id: u32, + ) -> Result { + let filter = + Bloom::new_for_fp_rate(items_count, fp_p).map_err(NymNodeError::bloomfilter_failure)?; + + Ok(RotationFilter { + metadata: ReplayProtectionBloomfilterMetadata { + creation_time: OffsetDateTime::now_utc(), + packets_received_at_creation, + rotation_id, + }, + data: filter, + }) + } + + // due to the size of the bloomfilter, extra caution has to be applied when using this method + // note: we're not getting reference to bytes as this method is used when flushing data to the disk + // (which takes ~30s) and we can't block the mutex for that long. + fn bytes(&self) -> Vec { + // attach metadata bytes at the end as it would make deserialisation cheaper (as we could avoid + // copying the bloomfilter bytes twice) + let mut bloom_bytes = self.data.to_bytes(); + bloom_bytes.extend_from_slice(&self.metadata.bytes()); + bloom_bytes + } + + pub(crate) fn try_from_bytes(bytes: Vec) -> Result { + let len = bytes.len(); + if bytes.len() < ReplayProtectionBloomfilterMetadata::SERIALIZED_LEN { + return Err(NymNodeError::BloomfilterMetadataDeserialisationFailure); + } + + let mut bloom_bytes = bytes; + let metadata_bytes = + bloom_bytes.split_off(len - ReplayProtectionBloomfilterMetadata::SERIALIZED_LEN); + + Ok(RotationFilter { + metadata: ReplayProtectionBloomfilterMetadata::try_from_bytes(&metadata_bytes)?, + data: Bloom::from_bytes(bloom_bytes).map_err(NymNodeError::bloomfilter_failure)?, + }) + } + + pub(crate) fn load>(path: P) -> Result { + info!("attempting to load prior replay detection bloomfilter..."); + let path = path.as_ref(); + let mut file = File::open(path).map_err(|source| NymNodeError::BloomfilterIoFailure { + source, + path: path.to_path_buf(), + })?; + + let mut buf = Vec::new(); + file.read_to_end(&mut buf) + .map_err(|source| NymNodeError::BloomfilterIoFailure { + source, + path: path.to_path_buf(), + })?; + + RotationFilter::try_from_bytes(buf) + } +} diff --git a/nym-node/src/node/replay_protection/helpers.rs b/nym-node/src/node/replay_protection/helpers.rs new file mode 100644 index 0000000000..d46e994919 --- /dev/null +++ b/nym-node/src/node/replay_protection/helpers.rs @@ -0,0 +1,34 @@ +// Copyright 2025 - Nym Technologies SA +// SPDX-License-Identifier: GPL-3.0-only + +pub(crate) fn parse_rotation_id_from_filename(name: &str) -> Option { + let stripped = name.strip_prefix("rot-")?; + let ext_idx = stripped.rfind(".").unwrap_or(stripped.len()); + let rotation = stripped.chars().take(ext_idx).collect::(); + rotation.parse::().ok() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parsing_rotation_id() { + let test_cases = vec![ + ("rot", None), + ("rot-123", Some(123)), + ("foo-123", None), + ("rot-123.ext", Some(123)), + ("rot-123.different-ext", Some(123)), + ("rot.123.aaa", None), + ]; + + for (raw, expected) in test_cases { + assert_eq!( + parse_rotation_id_from_filename(raw), + expected, + "failed: {raw} to {expected:?}" + ); + } + } +} diff --git a/nym-node/src/node/replay_protection/manager.rs b/nym-node/src/node/replay_protection/manager.rs index 0265a8fb78..b923ba535c 100644 --- a/nym-node/src/node/replay_protection/manager.rs +++ b/nym-node/src/node/replay_protection/manager.rs @@ -1,15 +1,18 @@ // Copyright 2025 - Nym Technologies SA // SPDX-License-Identifier: GPL-3.0-only +use crate::config::Config; use crate::error::NymNodeError; -use crate::node::replay_protection::bloomfilter::ReplayProtectionBloomfilters; +use crate::node::replay_protection::bloomfilter::{ReplayProtectionBloomfilters, RotationFilter}; 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 time::OffsetDateTime; use tracing::info; +#[derive(Clone)] pub(crate) struct ReplayProtectionBloomfiltersManager { target_fp_p: f64, minimum_bloomfilter_packets_per_second: usize, @@ -20,6 +23,52 @@ pub(crate) struct ReplayProtectionBloomfiltersManager { } impl ReplayProtectionBloomfiltersManager { + pub(crate) fn new_disabled(metrics: NymNodeMetrics) -> Self { + // the exact config values are irrelevant as the filters will never be recreated + ReplayProtectionBloomfiltersManager { + target_fp_p: 0.001, + minimum_bloomfilter_packets_per_second: 1, + bloomfilter_size_multiplier: 1.0, + metrics, + filters: ReplayProtectionBloomfilters::new_disabled(), + } + } + + pub(crate) fn new( + config: &Config, + primary: RotationFilter, + secondary: Option, + metrics: NymNodeMetrics, + ) -> Self { + ReplayProtectionBloomfiltersManager { + target_fp_p: config.mixnet.replay_protection.debug.false_positive_rate, + minimum_bloomfilter_packets_per_second: config + .mixnet + .replay_protection + .debug + .bloomfilter_minimum_packets_per_second_size, + bloomfilter_size_multiplier: config + .mixnet + .replay_protection + .debug + .bloomfilter_size_multiplier, + metrics, + filters: ReplayProtectionBloomfilters::new(primary, secondary), + } + } + + pub(crate) fn bloomfilters(&self) -> ReplayProtectionBloomfilters { + self.filters.clone() + } + + pub(crate) fn primary_bytes_and_id(&self) -> Result<(Vec, u32), NymNodeError> { + self.filters.primary_bytes_and_id() + } + + pub(crate) fn secondary_bytes_and_id(&self) -> Result, u32)>, NymNodeError> { + self.filters.secondary_bytes_and_id() + } + pub(crate) fn purge_secondary(&self) -> Result<(), NymNodeError> { self.filters.purge_secondary() } @@ -40,10 +89,10 @@ impl ReplayProtectionBloomfiltersManager { + self.metrics.mixnet.ingress.final_hop_packets_received(); let primary = self.filters.primary_metadata()?; - let time_delta = primary.creation_time.elapsed(); + let time_delta = OffsetDateTime::now_utc() - primary.creation_time; 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; + (received_since_creation as f64 / time_delta.as_seconds_f64()).round() as usize; let bf_received = max( received_per_second, diff --git a/nym-node/src/node/replay_protection/mod.rs b/nym-node/src/node/replay_protection/mod.rs index 00b2548f1a..969e4f5fd0 100644 --- a/nym-node/src/node/replay_protection/mod.rs +++ b/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; +mod helpers; pub(crate) mod manager; pub fn bitmap_size(false_positive_rate: f64, items_in_filter: usize) -> usize {