GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

flushing bloomfilters to disk and loading

Browse Source
This commit is contained in:
Jędrzej Stuczyński
2025-05-12 14:13:49 +01:00
parent 8b0d9a00ec
commit 8bcbb69889
11 changed files with 520 additions and 306 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.toml
+1 -1
View File
@@ -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"
nym-node/src/config/mod.rs
+35 -29
View File
@@ -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(())
}
}
nym-node/src/config/persistence.rs
-14
View File
@@ -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<P: AsRef<Path>>(data_dir: P) -> Self {
ReplayProtectionPaths {
nym-node/src/error.rs
+3
View File
@@ -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<nym_gateway::GatewayError>),
nym-node/src/node/key_rotation/controller.rs
+1 -2
View File
@@ -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,
nym-node/src/node/mod.rs
+13 -9
View File
@@ -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<ReplayProtectionBloomfilters, NymNodeError> {
) -> Result<ReplayProtectionBloomfiltersManager, NymNodeError> {
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(
nym-node/src/node/replay_protection/background_task.rs
+190 -116
View File
@@ -1,67 +1,33 @@
// 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;
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<u32>,
metrics: NymNodeMetrics,
shutdown_token: ShutdownToken,
) -> Result<Self, NymNodeError> {
// 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<u8>, 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}");
}
}
}
nym-node/src/node/replay_protection/bloomfilter.rs
+190 -132
View File
@@ -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::<i64>() + size_of::<u64>() + size_of::<u32>();
// UNIX_TIMESTAMP || PACKETS_RECEIVED || ROTATION_ID
pub(crate) fn bytes(&self) -> Vec<u8> {
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<Self, NymNodeError> {
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<std::sync::Mutex<ReplayProtectionBloomfiltersInner>>,
inner: Arc<Mutex<ReplayProtectionBloomfiltersInner>>,
}
impl ReplayProtectionBloomfilters {
pub(crate) fn new_empty(items_count: usize, fp_p: f64) -> Result<Self, NymNodeError> {
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<RotationFilter>) -> 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<u8>, 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<P: AsRef<Path>>(path: P) -> Result<Self, NymNodeError> {
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<Option<(Vec<u8>, 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<P: AsRef<Path>>(&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<Vec<u8>, 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<RotationFilter>,
overlap: Option<RotationFilter>,
pre_announced: Option<RotationFilter>,
}
@@ -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<Self, NymNodeError> {
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<u8> {
// 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<u8>) -> Result<Self, NymNodeError> {
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<P: AsRef<Path>>(path: P) -> Result<Self, NymNodeError> {
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)
}
}
nym-node/src/node/replay_protection/helpers.rs
+34
View File
@@ -0,0 +1,34 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
pub(crate) fn parse_rotation_id_from_filename(name: &str) -> Option<u32> {
let stripped = name.strip_prefix("rot-")?;
let ext_idx = stripped.rfind(".").unwrap_or(stripped.len());
let rotation = stripped.chars().take(ext_idx).collect::<String>();
rotation.parse::<u32>().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:?}"
);
}
}
}
nym-node/src/node/replay_protection/manager.rs
+52 -3
View File
@@ -1,15 +1,18 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// 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<RotationFilter>,
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<u8>, u32), NymNodeError> {
self.filters.primary_bytes_and_id()
}
pub(crate) fn secondary_bytes_and_id(&self) -> Result<Option<(Vec<u8>, 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,
nym-node/src/node/replay_protection/mod.rs
+1
View File
@@ -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 {