processing loop of KeyRotationController

common/cosmwasm-smart-contracts/mixnet-contract/src/key_rotation.rs

@@ -25,6 +25,18 @@ impl KeyRotationState {
         let full_rots = diff / self.validity_epochs;
         full_rots
     }
+
+    pub fn next_rotation_starting_epoch_id(&self, current_epoch_id: EpochId) -> EpochId {
+        let current_rotation_id = self.key_rotation_id(current_epoch_id);
+
+        self.initial_epoch_id + self.validity_epochs * (current_rotation_id + 1)
+    }
+
+    pub fn current_rotation_starting_epoch_id(&self, current_epoch_id: EpochId) -> EpochId {
+        let current_rotation_id = self.key_rotation_id(current_epoch_id);
+
+        self.initial_epoch_id + self.validity_epochs * current_rotation_id
+    }
 }
 
 #[cw_serde]
@@ -71,4 +83,76 @@ mod tests {
         assert_eq!(2, state.key_rotation_id(10048));
         assert_eq!(2, state.key_rotation_id(10060));
     }
+
+    #[test]
+    fn next_rotation_starting_epoch_id() {
+        let state = KeyRotationState {
+            validity_epochs: 24,
+            initial_epoch_id: 0,
+        };
+        assert_eq!(24, state.next_rotation_starting_epoch_id(0));
+        assert_eq!(24, state.next_rotation_starting_epoch_id(23));
+        assert_eq!(48, state.next_rotation_starting_epoch_id(24));
+        assert_eq!(48, state.next_rotation_starting_epoch_id(47));
+        assert_eq!(72, state.next_rotation_starting_epoch_id(48));
+
+        let state = KeyRotationState {
+            validity_epochs: 12,
+            initial_epoch_id: 0,
+        };
+        assert_eq!(12, state.next_rotation_starting_epoch_id(0));
+        assert_eq!(12, state.next_rotation_starting_epoch_id(11));
+        assert_eq!(24, state.next_rotation_starting_epoch_id(12));
+        assert_eq!(24, state.next_rotation_starting_epoch_id(23));
+        assert_eq!(36, state.next_rotation_starting_epoch_id(24));
+
+        let state = KeyRotationState {
+            validity_epochs: 24,
+            initial_epoch_id: 10000,
+        };
+        assert_eq!(10024, state.next_rotation_starting_epoch_id(123));
+        assert_eq!(10024, state.next_rotation_starting_epoch_id(10000));
+        assert_eq!(10024, state.next_rotation_starting_epoch_id(10001));
+        assert_eq!(10024, state.next_rotation_starting_epoch_id(10023));
+        assert_eq!(10048, state.next_rotation_starting_epoch_id(10024));
+        assert_eq!(10048, state.next_rotation_starting_epoch_id(10047));
+        assert_eq!(10072, state.next_rotation_starting_epoch_id(10048));
+        assert_eq!(10072, state.next_rotation_starting_epoch_id(10060));
+    }
+
+    #[test]
+    fn current_rotation_starting_epoch_id() {
+        let state = KeyRotationState {
+            validity_epochs: 24,
+            initial_epoch_id: 0,
+        };
+        assert_eq!(0, state.current_rotation_starting_epoch_id(0));
+        assert_eq!(0, state.current_rotation_starting_epoch_id(23));
+        assert_eq!(24, state.current_rotation_starting_epoch_id(24));
+        assert_eq!(24, state.current_rotation_starting_epoch_id(47));
+        assert_eq!(48, state.current_rotation_starting_epoch_id(48));
+
+        let state = KeyRotationState {
+            validity_epochs: 12,
+            initial_epoch_id: 0,
+        };
+        assert_eq!(0, state.current_rotation_starting_epoch_id(0));
+        assert_eq!(0, state.current_rotation_starting_epoch_id(11));
+        assert_eq!(12, state.current_rotation_starting_epoch_id(12));
+        assert_eq!(12, state.current_rotation_starting_epoch_id(23));
+        assert_eq!(24, state.current_rotation_starting_epoch_id(24));
+
+        let state = KeyRotationState {
+            validity_epochs: 24,
+            initial_epoch_id: 10000,
+        };
+        assert_eq!(10000, state.current_rotation_starting_epoch_id(123));
+        assert_eq!(10000, state.current_rotation_starting_epoch_id(10000));
+        assert_eq!(10000, state.current_rotation_starting_epoch_id(10001));
+        assert_eq!(10000, state.current_rotation_starting_epoch_id(10023));
+        assert_eq!(10024, state.current_rotation_starting_epoch_id(10024));
+        assert_eq!(10024, state.current_rotation_starting_epoch_id(10047));
+        assert_eq!(10048, state.current_rotation_starting_epoch_id(10048));
+        assert_eq!(10048, state.current_rotation_starting_epoch_id(10060));
+    }
 }

common/nymsphinx/framing/src/processing.rs

@@ -110,6 +110,11 @@ pub enum PacketProcessingError {
     PacketReplay,
 }
 
+pub struct PartialyUnwrappedPacketWithKeyRotation {
+    pub packet: PartiallyUnwrappedPacket,
+    pub used_key_rotation: u32,
+}
+
 pub struct PartiallyUnwrappedPacket {
     received_data: FramedNymPacket,
     partial_result: PartialMixProcessingResult,
@@ -174,6 +179,16 @@ impl PartiallyUnwrappedPacket {
     pub fn replay_tag(&self) -> Option<&[u8; REPLAY_TAG_SIZE]> {
         self.partial_result.replay_tag()
     }
+
+    pub fn with_key_rotation(
+        self,
+        used_key_rotation: u32,
+    ) -> PartialyUnwrappedPacketWithKeyRotation {
+        PartialyUnwrappedPacketWithKeyRotation {
+            packet: self,
+            used_key_rotation,
+        }
+    }
 }
 
 impl From<(FramedNymPacket, PartialMixProcessingResult)> for PartiallyUnwrappedPacket {

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

@@ -1454,6 +1454,23 @@ pub struct KeyRotationInfoResponse {
     pub epoch_duration: Duration,
 }
 
+impl KeyRotationInfoResponse {
+    pub fn current_key_rotation_id(&self) -> u32 {
+        self.key_rotation_state
+            .key_rotation_id(self.current_epoch_id)
+    }
+
+    pub fn next_rotation_starting_epoch_id(&self) -> EpochId {
+        self.key_rotation_state
+            .next_rotation_starting_epoch_id(self.current_epoch_id)
+    }
+
+    pub fn current_rotation_starting_epoch_id(&self) -> EpochId {
+        self.key_rotation_state
+            .current_rotation_starting_epoch_id(self.current_epoch_id)
+    }
+}
+
 #[derive(Clone, Debug, Serialize, Deserialize, schemars::JsonSchema, ToSchema)]
 pub struct RewardedSetResponse {
     #[serde(default)]

nym-node/nym-node-requests/src/api/mod.rs

@@ -10,7 +10,6 @@ use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
 use std::fmt::{Display, Formatter};
 use std::ops::Deref;
-use utoipa::ToSchema;
 
 #[cfg(feature = "client")]
 pub mod client;
@@ -22,7 +21,7 @@ pub use client::Client;
 // create the type alias manually if openapi is not enabled
 pub type SignedHostInformation = SignedData<crate::api::v1::node::models::HostInformation>;
 
-#[derive(ToSchema)]
+#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
 pub struct SignedDataHostInfo {
     // #[serde(flatten)]
     pub data: crate::api::v1::node::models::HostInformation,
@@ -341,8 +340,7 @@ mod tests {
         assert!(!current_struct_no_noise.verify(ed22519.public_key()));
         assert!(current_struct_no_noise.verify_host_information());
 
-        // if noise key is present, the signature is actually valid
-        assert!(current_struct_noise.verify(ed22519.public_key()));
+        assert!(!current_struct_noise.verify(ed22519.public_key()));
         assert!(current_struct_noise.verify_host_information())
     }
 

nym-node/src/error.rs

@@ -56,6 +56,15 @@ pub enum KeyIOFailure {
         err: io::Error,
     },
 
+    #[error("failed to copy {key} key from '{}' to '{}': {err}", source.display(), destination.display())]
+    KeyCopyFailure {
+        key: String,
+        source: PathBuf,
+        destination: PathBuf,
+        #[source]
+        err: io::Error,
+    },
+
     #[error("failed to remove {key} key from '{}': {err}", path.display())]
     KeyRemovalFailure {
         key: String,

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

@@ -16,11 +16,9 @@ struct ActiveSphinxKeysInner {
     /// Key that's currently used as the default when processing packets with no explicit rotation information
     primary_key: ArcSwap<SphinxPrivateKey>,
 
-    /// Optionally, a key from the previous rotation during the overlap period when the keys are rotated.
+    /// Optionally, a secondary key associated with this node. depending on the context it could either be
+    /// the pre-announced key for the following rotation or a key from the previous rotation for the overlap period
     secondary_key: ArcSwapOption<SphinxPrivateKey>,
-
-    /// Optionally, a key for the upcoming rotation that's being pre-announced to other network entities
-    pre_announced_key: ArcSwapOption<SphinxPrivateKey>,
 }
 
 impl ActiveSphinxKeys {
@@ -37,12 +35,13 @@ impl ActiveSphinxKeys {
         primary: SphinxPrivateKey,
         secondary: Option<SphinxPrivateKey>,
     ) -> Self {
-        ActiveSphinxKeys {
-            inner: Arc::new(ActiveSphinxKeysInner {
-                primary_key: ArcSwap::from_pointee(primary),
-                secondary_key: ArcSwapOption::from_pointee(secondary),
-            }),
-        }
+        todo!()
+        // ActiveSphinxKeys {
+        //     inner: Arc::new(ActiveSphinxKeysInner {
+        //         primary_key: ArcSwap::from_pointee(primary),
+        //         secondary_key: ArcSwapOption::from_pointee(secondary),
+        //     }),
+        // }
     }
 
     pub(crate) fn even(&self) -> Option<SphinxKeyGuard> {
@@ -74,17 +73,39 @@ impl ActiveSphinxKeys {
         Some(SphinxKeyGuard::Secondary(SecondaryKeyGuard { guard }))
     }
 
-    pub(crate) fn rotate(&self, new_primary: SphinxPrivateKey) {
-        if self.inner.secondary_key.load().is_some() {
-            // this should NEVER happen, but technically nothing should blow up
-            error!("somehow our secondary key was still set during the rotation!")
-        }
-
-        let old_primary = self.inner.primary_key.swap(Arc::new(new_primary));
-        self.inner.secondary_key.store(Some(old_primary));
+    pub(crate) fn set_secondary(&self, new_key: SphinxPrivateKey) {
+        self.inner.secondary_key.store(Some(Arc::new(new_key)))
     }
 
-    fn deactivate_secondary(&self) {
+    pub(crate) fn secondary_key_rotation_id(&self) -> Option<u32> {
+        self.inner
+            .secondary_key
+            .load()
+            .as_ref()
+            .map(|k| k.rotation_id())
+    }
+
+    // set the secondary (pre-announced key) as the primary
+    // and the current primary as the secondary (for the overlap epoch)
+    pub(crate) fn rotate(&self) -> bool {
+        let primary = self.inner.primary_key.load();
+
+        let Some(pre_announced) = self.inner.secondary_key.load_full() else {
+            error!("sphinx key inconsistency - attempted to perform key rotation without having pre-announced new key");
+            return false;
+        };
+
+        if pre_announced.rotation_id() != primary.rotation_id() + 1 {
+            error!("sphinx key inconsistency - pre-announced key rotation id != primary + 1");
+            return false;
+        }
+
+        let old_primary = self.inner.primary_key.swap(pre_announced);
+        self.inner.secondary_key.store(Some(old_primary));
+        true
+    }
+
+    pub(crate) fn deactivate_secondary(&self) {
         self.inner.secondary_key.store(None);
     }
 }
@@ -106,8 +127,25 @@ impl Deref for SphinxKeyGuard {
     }
 }
 
+// enum SecondaryKey {
+//     PreAnnounced(SphinxPrivateKey),
+//     PreviousOverlap(SphinxPrivateKey),
+// }
+
+// impl Deref for SecondaryKey {
+//     type Target = SphinxPrivateKey;
+//
+//     fn deref(&self) -> &Self::Target {
+//         match self {
+//             SecondaryKey::PreAnnounced(key) => &key,
+//             SecondaryKey::PreviousOverlap(key) => &key,
+//         }
+//     }
+// }
+
 pub(crate) struct SecondaryKeyGuard {
     guard: Guard<Option<Arc<SphinxPrivateKey>>>,
+    // guard: Guard<Option<Arc<SecondaryKey>>>,
 }
 
 impl Deref for SecondaryKeyGuard {

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

@@ -3,26 +3,53 @@
 
 use crate::node::key_rotation::manager::SphinxKeyManager;
 use crate::node::nym_apis_client::NymApisClient;
+use crate::node::replay_protection::bloomfilter::ReplayProtectionBloomfilters;
+use futures::pin_mut;
 use nym_task::ShutdownToken;
+use nym_validator_client::client::NymApiClientExt;
+use nym_validator_client::models::KeyRotationInfoResponse;
 use std::time::Duration;
-use tokio::time::interval;
-use tracing::{info, trace};
+use time::OffsetDateTime;
+use tokio::time::{interval, sleep, Instant};
+use tracing::{error, info, trace, warn};
 
 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,
     client: NymApisClient,
     managed_keys: SphinxKeyManager,
     shutdown_token: ShutdownToken,
 }
 
-enum KeyRotationActionState {
-    // perform key-rotation and pre-announce new key to the nym-api(s)
-    PreAnnounce,
+struct NextAction {
+    typ: KeyRotationActionState,
+    deadline: OffsetDateTime,
+}
 
-    // remove the old key and purge associated data like the replay detection bloomfilter
+impl NextAction {
+    fn until_deadline(&self) -> Duration {
+        let now = OffsetDateTime::now_utc();
+        Duration::try_from(self.deadline - now).unwrap_or_else(|_| {
+            // deadline is already in the past
+            Duration::from_nanos(0)
+        })
+    }
+}
+
+#[derive(Clone, Copy)]
+enum KeyRotationActionState {
+    // generate and pre-announce new key to the nym-api(s)
+    PreAnnounce { rotation_id: u32 },
+
+    // perform the following exchange
+    // primary -> secondary
+    // pre_announced -> primary
+    SwapDefault,
+
+    // remove the old overlap key and purge associated data like the replay detection bloomfilter
     PurgeOld,
 }
 
@@ -31,8 +58,108 @@ impl KeyRotationController {
         todo!()
     }
 
-    async fn regular_poll(&self) {
-        todo!()
+    async fn determine_next_action(&self) -> NextAction {
+        loop {
+            if let Some(next) = self.try_determine_next_action().await {
+                return next;
+            }
+
+            warn!("failed to determine next key rotation action; will try again in 2min");
+            sleep(Duration::from_secs(120)).await;
+        }
+    }
+
+    async fn try_determine_next_action(&self) -> Option<NextAction> {
+        let key_rotation_info = self.try_get_key_rotation_info().await?;
+        let current_rotation = key_rotation_info.current_key_rotation_id();
+
+        let current_epoch = key_rotation_info.current_epoch_id;
+        let next_rotation_epoch = key_rotation_info.next_rotation_starting_epoch_id();
+        let current_rotation_epoch = key_rotation_info.current_rotation_starting_epoch_id();
+
+        let secondary_rotation_id = self.managed_keys.keys.secondary_key_rotation_id();
+        let (action, execution_epoch) = match secondary_rotation_id {
+            None => {
+                // we don't have any secondary key, meaning the next thing we could possibly do is to pre-announce new key
+                // an epoch before next rotation
+                let rotation_id = current_rotation + 1;
+
+                (
+                    KeyRotationActionState::PreAnnounce { rotation_id },
+                    next_rotation_epoch - 1,
+                )
+            }
+            Some(id) if id == current_rotation - 1 => {
+                // our secondary key is from the previous rotation, meaning the next thing we have to do
+                // is to remove it (we have clearly already rotated)
+                (KeyRotationActionState::PurgeOld, current_rotation_epoch + 1)
+            }
+            Some(id) if id == current_rotation => {
+                // our secondary key is from the current epoch, meaning (hopefully) we just have gone into the
+                // next rotation, and we have to swap it into the primary
+                (KeyRotationActionState::SwapDefault, current_rotation_epoch)
+            }
+            Some(id) if id == current_rotation + 1 => {
+                // our secondary key is from the upcoming rotation, meaning it's the pre-announced key, meaning
+                // the next thing we have to do is to swap it into the primary
+                (KeyRotationActionState::SwapDefault, next_rotation_epoch)
+            }
+            Some(other) => {
+                // this situation should have never occurred, our secondary key is completely unusable,
+                // so we should just remove it immediately and try again
+                error!("inconsistent secondary key state. it's marked for rotation {other} while the current value is {current_rotation}");
+                (KeyRotationActionState::PurgeOld, current_epoch)
+            }
+        };
+
+        let now = OffsetDateTime::now_utc();
+        let since_epoch_start = now - key_rotation_info.current_epoch_start;
+        let until_execution_epoch =
+            execution_epoch.saturating_sub(current_epoch) * key_rotation_info.epoch_duration;
+
+        Some(NextAction {
+            typ: action,
+            deadline: now - since_epoch_start + until_execution_epoch,
+        })
+    }
+
+    async fn try_get_key_rotation_info(&self) -> Option<KeyRotationInfoResponse> {
+        let Ok(rotation_info) = self
+            .client
+            .query_exhaustively(
+                async |c| c.get_key_rotation_info().await,
+                Duration::from_secs(5),
+            )
+            .await
+        else {
+            warn!("failed to retrieve key rotation information from ANY nym-api - we might miss configuration changes");
+            return None;
+        };
+
+        Some(rotation_info)
+    }
+
+    async fn execute_next_action(&self, action: KeyRotationActionState) {
+        match action {
+            KeyRotationActionState::PreAnnounce { rotation_id } => {
+                let public_key = match self.managed_keys.generate_key_for_new_rotation(rotation_id)
+                {
+                    Err(err) => {
+                        error!("failed to generate and store new sphinx key: {err}");
+                        return;
+                    }
+                    Ok(key) => key,
+                };
+
+                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}")
+                }
+            }
+            KeyRotationActionState::PurgeOld => {}
+        }
     }
 
     pub(crate) async fn run(&self) {
@@ -41,17 +168,27 @@ impl KeyRotationController {
         let mut polling_interval = interval(self.regular_polling_interval);
         polling_interval.reset();
 
+        let mut next_action = self.determine_next_action().await;
+        let mut state_update_future = sleep(next_action.until_deadline());
+        pin_mut!(state_update_future);
+
         while !self.shutdown_token.is_cancelled() {
             tokio::select! {
                 biased;
                 _ = self.shutdown_token.cancelled() => {
-                   trace!("KeyRotationController: Received shutdown");
+                    trace!("KeyRotationController: Received shutdown");
+                    break;
                 }
-                _ = polling_interval.tick() => {
-                    self.regular_poll().await;
+                _ = polling_interval.tick() => {}
+                _ = &mut state_update_future => {
+                    self.execute_next_action(next_action.typ).await
                 }
-                // TODO:
             }
+
+            next_action = self.determine_next_action().await;
+            state_update_future
+                .as_mut()
+                .reset(Instant::now() + next_action.until_deadline());
         }
 
         trace!("KeyRotationController: exiting")

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

@@ -4,7 +4,7 @@
 use crate::error::{KeyIOFailure, NymNodeError};
 use crate::node::helpers::{load_key, store_key};
 use crate::node::key_rotation::active_keys::ActiveSphinxKeys;
-use crate::node::key_rotation::key::SphinxPrivateKey;
+use crate::node::key_rotation::key::{SphinxPrivateKey, SphinxPublicKey};
 use rand::rngs::OsRng;
 use rand::{CryptoRng, RngCore};
 use std::fs;
@@ -43,14 +43,17 @@ impl SphinxKeyManager {
         })
     }
 
-    fn replace_key_files<P: AsRef<Path>>(
+    // moves the primary key to the secondary file
+    // and vice verse, i.e. secondary to the primary
+    fn swap_key_files<P: AsRef<Path>>(
         primary_path: P,
         secondary_path: P,
     ) -> Result<(), NymNodeError> {
         let tmp_path = primary_path.as_ref().with_extension("tmp");
 
-        fs::rename(primary_path.as_ref(), secondary_path.as_ref()).map_err(|err| {
-            KeyIOFailure::KeyMoveFailure {
+        // 1. COPY: primary -> temp
+        fs::copy(primary_path.as_ref(), secondary_path.as_ref()).map_err(|err| {
+            KeyIOFailure::KeyCopyFailure {
                 key: "old x25519 sphinx primary".to_string(),
                 source: primary_path.as_ref().to_path_buf(),
                 destination: secondary_path.as_ref().to_path_buf(),
@@ -58,34 +61,72 @@ impl SphinxKeyManager {
             }
         })?;
 
-        fs::rename(&tmp_path, primary_path.as_ref()).map_err(|err| {
+        // 2. MOVE: secondary -> primary
+        fs::rename(secondary_path.as_ref(), primary_path.as_ref()).map_err(|err| {
             KeyIOFailure::KeyMoveFailure {
-                key: "new x25519 sphinx primary".to_string(),
-                source: tmp_path,
+                key: "x25519 sphinx secondary".to_string(),
+                source: secondary_path.as_ref().to_path_buf(),
                 destination: primary_path.as_ref().to_path_buf(),
                 err,
             }
         })?;
+
+        // 3. MOVE temp -> secondary
+        fs::rename(&tmp_path, secondary_path.as_ref()).map_err(|err| {
+            KeyIOFailure::KeyMoveFailure {
+                key: "old x25519 sphinx primary".to_string(),
+                source: tmp_path.clone(),
+                destination: primary_path.as_ref().to_path_buf(),
+                err,
+            }
+        })?;
+
+        // 4. REMOVE: temp
+        fs::remove_file(&tmp_path).map_err(|err| KeyIOFailure::KeyRemovalFailure {
+            key: "old x25519 sphinx primary (temp location)".to_string(),
+            path: tmp_path,
+            err,
+        })?;
         Ok(())
     }
 
-    // 1. generate new key
-    // 2. save it in a temp file
-    // 3. move primary key file to the secondary file location (thus losing the secondary)
-    // 4. move the temp file to the primary file location
-    // 5. set primary as the secondary
-    // 6. set new key as the primary
-    // 7. (outside this method) broadcast update to nym-apis
-    pub(crate) fn rotate_keys(&mut self, current_rotation_id: u32) -> Result<(), NymNodeError> {
+    pub(crate) fn generate_key_for_new_rotation(
+        &self,
+        expected_rotation: u32,
+    ) -> Result<SphinxPublicKey, NymNodeError> {
         let mut rng = OsRng;
-        let new_primary = SphinxPrivateKey::new(&mut rng, current_rotation_id);
+        let new = SphinxPrivateKey::new(&mut rng, expected_rotation);
+        let pub_key = (&new).into();
+        store_key(
+            &new,
+            &self.secondary_key_path,
+            "x22519 (pre-announced) sphinx",
+        )?;
 
-        let tmp_path = self.primary_key_path.with_extension("tmp");
-        store_key(&new_primary, &tmp_path, "x25519 sphinx")?;
+        self.keys.set_secondary(new);
+        Ok(pub_key)
+    }
 
-        Self::replace_key_files(&self.primary_key_path, &self.secondary_key_path)?;
+    pub(crate) fn rotate_keys(&self) -> Result<(), NymNodeError> {
+        if !self.keys.rotate() {
+            // we failed to perform the rotation because the secondary key somehow didn't exist
+            // we can't do much here, but just generate a brand-new key to rotate into
+            let primary = self.keys.primary().rotation_id();
+            self.generate_key_for_new_rotation(primary + 1)?;
+            self.keys.rotate();
+        }
+        Self::swap_key_files(&self.primary_key_path, &self.secondary_key_path)
+    }
 
-        self.keys.rotate(new_primary);
+    pub(crate) fn remove_overlap_key(&self) -> Result<(), NymNodeError> {
+        self.keys.deactivate_secondary();
+        fs::remove_file(&self.secondary_key_path).map_err(|err| {
+            KeyIOFailure::KeyRemovalFailure {
+                key: "old x25519 sphinx secondary".to_string(),
+                path: self.secondary_key_path.clone(),
+                err,
+            }
+        })?;
         Ok(())
     }
 
@@ -94,49 +135,51 @@ impl SphinxKeyManager {
         primary_key_path: P,
         secondary_key_path: P,
     ) -> Result<Self, NymNodeError> {
-        // check the temporary location in case we crashed in the middle of rotating the key
-        let tmp_location = primary_key_path.as_ref().with_extension("tmp");
-        if tmp_location.exists() {
-            warn!("we seem to have crashed in the middle of rotating the sphinx key");
-            // if temporary key exists, it means it has never overwritten the primary;
-            // secondary key might or might have not gotten overwritten, but that doesn't matter,
-            // we can do it again
-            Self::replace_key_files(primary_key_path.as_ref(), secondary_key_path.as_ref())?;
-        }
+        todo!("check if primary and secondary are correct - we might have crashed during the file swap");
 
-        // primary key should always be present
-        let primary: SphinxPrivateKey =
-            load_key(primary_key_path.as_ref(), "x25519 sphinx primary")?;
-
-        // if upon loading it turns out that the node has been inactive for a long time,
-        // immediately rotate keys (but leave 1h grace period for current primary, i.e. set it as secondary)
-        if primary.rotation_id() != current_rotation_id {
-            warn!("this node has been inactive for more than a key rotation duration. the current primary key was generated for rotation {} while the current rotation is {current_rotation_id}. new key will be generated now.", primary.rotation_id());
-            let mut this = SphinxKeyManager {
-                keys: ActiveSphinxKeys::new_loaded(primary, None),
-                primary_key_path: primary_key_path.as_ref().to_path_buf(),
-                secondary_key_path: secondary_key_path.as_ref().to_path_buf(),
-            };
-            this.rotate_keys(current_rotation_id)?;
-            return Ok(this);
-        }
-
-        // secondary key **might** be present
-        let secondary_path = secondary_key_path.as_ref();
-
-        let secondary = if secondary_path.exists() {
-            Some(load_key::<SphinxPrivateKey, _>(
-                secondary_key_path.as_ref(),
-                "x25519 sphinx secondary",
-            )?)
-        } else {
-            None
-        };
-
-        Ok(SphinxKeyManager {
-            keys: ActiveSphinxKeys::new_loaded(primary, secondary),
-            primary_key_path: primary_key_path.as_ref().to_path_buf(),
-            secondary_key_path: secondary_key_path.as_ref().to_path_buf(),
-        })
+        // // check the temporary location in case we crashed in the middle of rotating the key
+        // let tmp_location = primary_key_path.as_ref().with_extension("tmp");
+        // if tmp_location.exists() {
+        //     warn!("we seem to have crashed in the middle of rotating the sphinx key");
+        //     // if temporary key exists, it means it has never overwritten the primary;
+        //     // secondary key might or might have not gotten overwritten, but that doesn't matter,
+        //     // we can do it again
+        //     Self:swape_key_files(primary_key_path.as_ref(), secondary_key_path.as_ref())?;
+        // }
+        //
+        // // primary key should always be present
+        // let primary: SphinxPrivateKey =
+        //     load_key(primary_key_path.as_ref(), "x25519 sphinx primary")?;
+        //
+        // // if upon loading it turns out that the node has been inactive for a long time,
+        // // immediately rotate keys (but leave 1h grace period for current primary, i.e. set it as secondary)
+        // if primary.rotation_id() != current_rotation_id {
+        //     warn!("this node has been inactive for more than a key rotation duration. the current primary key was generated for rotation {} while the current rotation is {current_rotation_id}. new key will be generated now.", primary.rotation_id());
+        //     let mut this = SphinxKeyManager {
+        //         keys: ActiveSphinxKeys::new_loaded(primary, None),
+        //         primary_key_path: primary_key_path.as_ref().to_path_buf(),
+        //         secondary_key_path: secondary_key_path.as_ref().to_path_buf(),
+        //     };
+        //     this.rotate_keys(current_rotation_id)?;
+        //     return Ok(this);
+        // }
+        //
+        // // secondary key **might** be present
+        // let secondary_path = secondary_key_path.as_ref();
+        //
+        // let secondary = if secondary_path.exists() {
+        //     Some(load_key::<SphinxPrivateKey, _>(
+        //         secondary_key_path.as_ref(),
+        //         "x25519 sphinx secondary",
+        //     )?)
+        // } else {
+        //     None
+        // };
+        //
+        // Ok(SphinxKeyManager {
+        //     keys: ActiveSphinxKeys::new_loaded(primary, secondary),
+        //     primary_key_path: primary_key_path.as_ref().to_path_buf(),
+        //     secondary_key_path: secondary_key_path.as_ref().to_path_buf(),
+        // })
     }
 }

nym-node/src/node/mixnet/handler.rs

@@ -8,10 +8,11 @@ use nym_sphinx_framing::codec::NymCodec;
 use nym_sphinx_framing::packet::FramedNymPacket;
 use nym_sphinx_framing::processing::{
     process_framed_packet, MixProcessingResult, MixProcessingResultData, PacketProcessingError,
-    PartiallyUnwrappedPacket, ProcessedFinalHop,
+    PartiallyUnwrappedPacket, PartialyUnwrappedPacketWithKeyRotation, ProcessedFinalHop,
 };
 use nym_sphinx_params::SphinxKeyRotation;
 use nym_sphinx_types::{Delay, REPLAY_TAG_SIZE};
+use std::collections::HashMap;
 use std::mem;
 use std::net::SocketAddr;
 use tokio::net::TcpStream;
@@ -20,41 +21,50 @@ use tokio_util::codec::Framed;
 use tracing::{debug, error, instrument, trace, warn};
 
 struct PendingReplayCheckPackets {
-    packets: Vec<PartiallyUnwrappedPacket>,
+    // map of rotation id used for packet creation to the packets
+    packets: HashMap<u32, Vec<PartiallyUnwrappedPacket>>,
     last_acquired_mutex: Instant,
 }
 
 impl PendingReplayCheckPackets {
     fn new() -> PendingReplayCheckPackets {
         PendingReplayCheckPackets {
-            packets: vec![],
+            packets: Default::default(),
             last_acquired_mutex: Instant::now(),
         }
     }
 
-    fn reset(&mut self, now: Instant) -> Vec<PartiallyUnwrappedPacket> {
+    fn reset(&mut self, now: Instant) -> HashMap<u32, Vec<PartiallyUnwrappedPacket>> {
         self.last_acquired_mutex = now;
         mem::take(&mut self.packets)
     }
 
-    fn push(&mut self, now: Instant, packet: PartiallyUnwrappedPacket) {
+    fn push(&mut self, now: Instant, packet: PartialyUnwrappedPacketWithKeyRotation) {
         if self.packets.is_empty() {
             self.last_acquired_mutex = now;
         }
-        self.packets.push(packet);
+        self.packets
+            .entry(packet.used_key_rotation)
+            .or_default()
+            .push(packet.packet)
     }
 
-    fn replay_tags(&self) -> Vec<&[u8; REPLAY_TAG_SIZE]> {
-        let mut replay_tags = Vec::with_capacity(self.packets.len());
-        for packet in &self.packets {
-            let Some(replay_tag) = packet.replay_tag() else {
-                error!(
-                    "corrupted batch of {} packets - replay tag was missing",
-                    self.packets.len()
-                );
-                return Vec::new();
-            };
-            replay_tags.push(replay_tag);
+    fn replay_tags(&self) -> HashMap<u32, Vec<&[u8; REPLAY_TAG_SIZE]>> {
+        let mut replay_tags = HashMap::with_capacity(self.packets.len());
+        'outer: for (rotation_id, packets) in &self.packets {
+            let mut rotation_replay_tags = Vec::with_capacity(packets.len());
+            for packet in packets {
+                let Some(replay_tag) = packet.replay_tag() else {
+                    error!(
+                        "corrupted batch of {} packets - replay tag was missing",
+                        self.packets.len()
+                    );
+                    replay_tags.insert(*rotation_id, Vec::new());
+                    continue 'outer;
+                };
+                rotation_replay_tags.push(replay_tag);
+            }
+            replay_tags.insert(*rotation_id, rotation_replay_tags);
         }
         replay_tags
     }
@@ -216,22 +226,26 @@ impl ConnectionHandler {
     fn try_partially_unwrap_packet(
         &self,
         packet: FramedNymPacket,
-    ) -> Result<PartiallyUnwrappedPacket, PacketProcessingError> {
+    ) -> Result<PartialyUnwrappedPacketWithKeyRotation, PacketProcessingError> {
         // based on the received sphinx key rotation information,
         // attempt to choose appropriate key for processing the packet
         match packet.header().key_rotation {
             SphinxKeyRotation::Unknown => {
+                let primary = self.shared.sphinx_keys.primary();
+                let primary_rotation = primary.rotation_id();
+
                 // we have to try both keys, start with the primary as it has higher likelihood of being correct
                 // if let Ok(partially_unwrapped) = PartiallyUnwrappedPacket::new()
-                match PartiallyUnwrappedPacket::new(
-                    packet,
-                    self.shared.sphinx_keys.primary().inner().as_ref(),
-                ) {
-                    Ok(unwrapped_packet) => Ok(unwrapped_packet),
+                match PartiallyUnwrappedPacket::new(packet, primary.inner().as_ref()) {
+                    Ok(unwrapped_packet) => {
+                        Ok(unwrapped_packet.with_key_rotation(primary_rotation))
+                    }
                     Err((packet, err)) => {
                         if let Some(secondary) = self.shared.sphinx_keys.secondary() {
+                            let secondary_rotation = secondary.rotation_id();
                             PartiallyUnwrappedPacket::new(packet, secondary.inner().as_ref())
                                 .map_err(|(_, err)| err)
+                                .map(|p| p.with_key_rotation(secondary_rotation))
                         } else {
                             Err(err)
                         }
@@ -242,15 +256,19 @@ impl ConnectionHandler {
                 let Some(odd_key) = self.shared.sphinx_keys.odd() else {
                     return Err(PacketProcessingError::ExpiredKey);
                 };
+                let odd_rotation = odd_key.rotation_id();
                 PartiallyUnwrappedPacket::new(packet, odd_key.inner().as_ref())
                     .map_err(|(_, err)| err)
+                    .map(|p| p.with_key_rotation(odd_rotation))
             }
             SphinxKeyRotation::EvenRotation => {
                 let Some(even_key) = self.shared.sphinx_keys.even() else {
                     return Err(PacketProcessingError::ExpiredKey);
                 };
+                let even_rotation = even_key.rotation_id();
                 PartiallyUnwrappedPacket::new(packet, even_key.inner().as_ref())
                     .map_err(|(_, err)| err)
+                    .map(|p| p.with_key_rotation(even_rotation))
             }
         }
     }
@@ -317,17 +335,24 @@ impl ConnectionHandler {
     async fn handle_post_replay_detection_packets(
         &self,
         now: Instant,
-        packets: Vec<PartiallyUnwrappedPacket>,
-        replay_check_results: Vec<bool>,
+        packets: HashMap<u32, Vec<PartiallyUnwrappedPacket>>,
+        replay_check_results: HashMap<u32, Vec<bool>>,
     ) {
-        for (packet, replayed) in packets.into_iter().zip(replay_check_results) {
-            let unwrapped_packet = if replayed {
-                Err(PacketProcessingError::PacketReplay)
-            } else {
-                packet.finalise_unwrapping()
+        for (rotation_id, packets) in packets {
+            let Some(replay_checks) = replay_check_results.get(&rotation_id) else {
+                // this should never happen, but if we messed up, and it does, don't panic, just drop the packets
+                error!("inconsistent replay check result - no values for rotation {rotation_id}");
+                continue;
             };
+            for (packet, &replayed) in packets.into_iter().zip(replay_checks) {
+                let unwrapped_packet = if replayed {
+                    Err(PacketProcessingError::PacketReplay)
+                } else {
+                    packet.finalise_unwrapping()
+                };
 
-            self.handle_unwrapped_packet(now, unwrapped_packet).await;
+                self.handle_unwrapped_packet(now, unwrapped_packet).await;
+            }
         }
     }
 

nym-node/src/node/mixnet/shared/mod.rs

@@ -5,7 +5,7 @@ use crate::config::Config;
 use crate::node::key_rotation::active_keys::ActiveSphinxKeys;
 use crate::node::mixnet::handler::ConnectionHandler;
 use crate::node::mixnet::SharedFinalHopData;
-use crate::node::replay_protection::bloomfilter::ReplayProtectionBloomfilter;
+use crate::node::replay_protection::bloomfilter::ReplayProtectionBloomfilters;
 use nym_gateway::node::GatewayStorageError;
 use nym_mixnet_client::forwarder::{MixForwardingSender, PacketToForward};
 use nym_node_metrics::mixnet::PacketKind;
@@ -66,7 +66,7 @@ impl ProcessingConfig {
 pub(crate) struct SharedData {
     pub(super) processing_config: ProcessingConfig,
     pub(super) sphinx_keys: ActiveSphinxKeys,
-    pub(super) replay_protection_filter: ReplayProtectionBloomfilter,
+    pub(super) replay_protection_filter: ReplayProtectionBloomfilters,
 
     // used for FORWARD mix packets and FINAL ack packets
     pub(super) mixnet_forwarder: MixForwardingSender,
@@ -89,7 +89,7 @@ impl SharedData {
     pub(crate) fn new(
         processing_config: ProcessingConfig,
         sphinx_keys: ActiveSphinxKeys,
-        replay_protection_filter: ReplayProtectionBloomfilter,
+        replay_protection_filter: ReplayProtectionBloomfilters,
         mixnet_forwarder: MixForwardingSender,
         final_hop: SharedFinalHopData,
         metrics: NymNodeMetrics,

nym-node/src/node/mod.rs

@@ -31,7 +31,7 @@ use crate::node::mixnet::shared::ProcessingConfig;
 use crate::node::mixnet::SharedFinalHopData;
 use crate::node::nym_apis_client::NymApisClient;
 use crate::node::replay_protection::background_task::ReplayProtectionBackgroundTask;
-use crate::node::replay_protection::bloomfilter::ReplayProtectionBloomfilter;
+use crate::node::replay_protection::bloomfilter::ReplayProtectionBloomfilters;
 use crate::node::routing_filter::{OpenFilter, RoutingFilter};
 use crate::node::shared_network::{
     CachedNetwork, CachedTopologyProvider, LocalGatewayNode, NetworkRefresher,
@@ -956,9 +956,9 @@ impl NymNode {
 
     pub(crate) async fn setup_replay_detection(
         &self,
-    ) -> Result<ReplayProtectionBloomfilter, NymNodeError> {
+    ) -> Result<ReplayProtectionBloomfilters, NymNodeError> {
         if self.config.mixnet.replay_protection.debug.unsafe_disabled {
-            return Ok(ReplayProtectionBloomfilter::new_disabled());
+            return Ok(ReplayProtectionBloomfilters::new_disabled());
         }
 
         // create the background task for the bloomfilter

nym-node/src/node/nym_apis_client.rs

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::error::NymNodeError;
+use crate::node::key_rotation::key::SphinxPublicKey;
 use crate::node::NymNode;
 use futures::{stream, StreamExt};
 use nym_crypto::asymmetric::ed25519;
@@ -55,17 +56,17 @@ impl NymApisClient {
         })
     }
 
-    async fn use_next_endpoint(&mut self) {
-        let mut guard = self.inner.write().await;
-        if guard.available_urls.len() == 1 {
-            return;
-        }
-
-        let next_index = (guard.currently_used_api + 1) % guard.available_urls.len();
-        let next = guard.available_urls[next_index].clone();
-        guard.currently_used_api = next_index;
-        guard.active_client.change_nym_api(next)
-    }
+    // async fn use_next_endpoint(&self) {
+    //     let mut guard = self.inner.write().await;
+    //     if guard.available_urls.len() == 1 {
+    //         return;
+    //     }
+    //
+    //     let next_index = (guard.currently_used_api + 1) % guard.available_urls.len();
+    //     let next = guard.available_urls[next_index].clone();
+    //     guard.currently_used_api = next_index;
+    //     guard.active_client.change_nym_api(next)
+    // }
 
     pub(crate) async fn query_exhaustively<R, T>(
         &self,
@@ -75,11 +76,19 @@ impl NymApisClient {
     where
         R: AsyncFn(Client) -> Result<T, NymAPIError>,
     {
-        self.inner
-            .read()
-            .await
-            .query_exhaustively(req, timeout_duration)
-            .await
+        let guard = self.inner.read().await;
+        let (res, last_working_endpoint) = guard.query_exhaustively(req, timeout_duration).await?;
+
+        // if we had to use a different api, update our starting point for the future calls
+        if guard.currently_used_api != last_working_endpoint {
+            drop(guard);
+            let mut guard = self.inner.write().await;
+            let next_url = guard.available_urls[last_working_endpoint].clone();
+            guard.currently_used_api = last_working_endpoint;
+            guard.active_client.change_nym_api(next_url);
+        }
+
+        Ok(res)
     }
 
     pub(crate) async fn broadcast_force_refresh(&self, private_key: &ed25519::PrivateKey) {
@@ -90,8 +99,12 @@ impl NymApisClient {
             .await;
     }
 
-    pub(crate) async fn broadcast_key_rotation(&self) {
-        self.inner.read().await.broadcast_key_rotation().await;
+    pub(crate) async fn broadcast_pre_announced_key(&self, public_key: SphinxPublicKey) {
+        self.inner
+            .read()
+            .await
+            .broadcast_pre_announced_key(public_key)
+            .await;
     }
 }
 
@@ -117,20 +130,30 @@ impl InnerClient {
         }
     }
 
-    pub(crate) async fn query_exhaustively<R, T>(
+    async fn query_exhaustively<R, T>(
         &self,
         req: R,
         timeout_duration: Duration,
-    ) -> Result<T, NymNodeError>
+    ) -> Result<(T, usize), NymNodeError>
     where
         R: AsyncFn(Client) -> Result<T, NymAPIError>,
     {
-        // this is DESIGNED to query sequentially (but exhaustively) and not to try to send queries to ALL apis at once
+        let last_working = self.currently_used_api;
+
+        // start from the last working api and progress from there
+        // also, note this is DESIGNED to query sequentially (but exhaustively)
+        // and not to try to send queries to ALL apis at once
         // and check which resolves first
-        for url in &self.available_urls {
+        for (idx, url) in self
+            .available_urls
+            .iter()
+            .enumerate()
+            .skip(last_working)
+            .chain(self.available_urls.iter().enumerate().take(last_working))
+        {
             let nym_api = self.active_client.nym_api.clone_with_new_url(url.clone());
             match timeout(timeout_duration, req(nym_api)).await {
-                Ok(Ok(res)) => return Ok(res),
+                Ok(Ok(res)) => return Ok((res, idx)),
                 Ok(Err(err)) => {
                     warn!("failed to resolve query for {url}: {err}")
                 }
@@ -143,7 +166,7 @@ impl InnerClient {
         Err(NymNodeError::NymApisExhausted)
     }
 
-    pub(crate) async fn broadcast_force_refresh(&self, private_key: &ed25519::PrivateKey) {
+    async fn broadcast_force_refresh(&self, private_key: &ed25519::PrivateKey) {
         let request = NodeRefreshBody::new(private_key);
 
         self.broadcast(
@@ -154,7 +177,7 @@ impl InnerClient {
         .await;
     }
 
-    pub(crate) async fn broadcast_key_rotation(&self) {
+    async fn broadcast_pre_announced_key(&self, public_key: SphinxPublicKey) {
         todo!()
     }
 }

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

@@ -3,7 +3,7 @@
 
 use crate::config::Config;
 use crate::error::NymNodeError;
-use crate::node::replay_protection::bloomfilter::ReplayProtectionBloomfilter;
+use crate::node::replay_protection::bloomfilter::ReplayProtectionBloomfilters;
 use crate::node::replay_protection::items_in_bloomfilter;
 use human_repr::HumanCount;
 use nym_node_metrics::NymNodeMetrics;
@@ -72,7 +72,7 @@ pub struct ReplayProtectionBackgroundTask {
     config: ReplayProtectionBackgroundTaskConfig,
     last_reset: LastResetData,
 
-    filter: ReplayProtectionBloomfilter,
+    filter: ReplayProtectionBloomfilters,
     metrics: NymNodeMetrics,
     shutdown_token: ShutdownToken,
 }
@@ -99,7 +99,7 @@ impl ReplayProtectionBackgroundTask {
 
         // if there's nothing on disk, we must create a new filter
         let bloomfilter = if task_config.current_bloomfilter_path.exists() {
-            ReplayProtectionBloomfilter::load(&task_config.current_bloomfilter_path).await?
+            ReplayProtectionBloomfilters::load(&task_config.current_bloomfilter_path).await?
         } else {
             let bf_items = items_in_bloomfilter(
                 task_config.filter_reset_rate,
@@ -110,7 +110,7 @@ impl ReplayProtectionBackgroundTask {
                     .initial_expected_packets_per_second,
             );
 
-            ReplayProtectionBloomfilter::new_empty(bf_items, task_config.false_positive_rate)?
+            ReplayProtectionBloomfilters::new_empty(bf_items, task_config.false_positive_rate)?
         };
 
         Ok(ReplayProtectionBackgroundTask {
@@ -125,7 +125,7 @@ impl ReplayProtectionBackgroundTask {
         })
     }
 
-    pub(crate) fn global_bloomfilter(&self) -> ReplayProtectionBloomfilter {
+    pub(crate) fn global_bloomfilter(&self) -> ReplayProtectionBloomfilters {
         self.filter.clone()
     }
 

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

@@ -5,6 +5,7 @@ use crate::error::NymNodeError;
 use bloomfilter::Bloom;
 use human_repr::HumanDuration;
 use nym_sphinx_types::REPLAY_TAG_SIZE;
+use std::collections::HashMap;
 use std::path::Path;
 use std::sync::{Arc, PoisonError, TryLockError};
 use tokio::fs::File;
@@ -12,34 +13,58 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::time::Instant;
 use tracing::{debug, 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 {
+    // 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,
+
+    /// 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,
+
+    rotation_id: u32,
+}
+
 // 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 ReplayProtectionBloomfilter {
+pub(crate) struct ReplayProtectionBloomfilters {
     disabled: bool,
-    inner: Arc<std::sync::Mutex<ReplayProtectionBloomfilterInner>>,
+    inner: Arc<std::sync::Mutex<ReplayProtectionBloomfiltersInner>>,
 }
 
-impl ReplayProtectionBloomfilter {
+impl ReplayProtectionBloomfilters {
     pub(crate) fn new_empty(items_count: usize, fp_p: f64) -> Result<Self, NymNodeError> {
-        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)?,
-            })),
-        })
+        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)?,
+        //     })),
+        // })
     }
 
     // SAFETY: the hardcoded values of 1,1 are valid
     #[allow(clippy::unwrap_used)]
     pub(crate) fn new_disabled() -> Self {
         // well, technically it's not fully empty, but the memory footprint is negligible
-        ReplayProtectionBloomfilter {
+        ReplayProtectionBloomfilters {
             disabled: true,
-            inner: Arc::new(std::sync::Mutex::new(ReplayProtectionBloomfilterInner {
-                current_filter: Bloom::new(1, 1).unwrap(),
+            inner: Arc::new(std::sync::Mutex::new(ReplayProtectionBloomfiltersInner {
+                primary: RotationFilter {
+                    metadata: ReplayProtectionBloomfilterMetadata {
+                        creation_time: Instant::now(),
+                        packets_received_at_creation: 0,
+                        rotation_id: u32::MAX,
+                    },
+                    data: Bloom::new(1, 1).unwrap(),
+                },
+                secondary: None,
+                pre_announced: None,
             })),
         }
     }
@@ -50,174 +75,188 @@ impl ReplayProtectionBloomfilter {
 
     pub(crate) fn reset(&self, items_count: usize, fp_p: f64) -> Result<(), NymNodeError> {
         // 1. build the new filter
-        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(())
+        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(())
     }
 
     // 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> {
-        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)?,
-            })),
-        })
+        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)?,
+        //     })),
+        // })
     }
 
     // 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> {
-        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(())
+        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(())
     }
 }
 
-struct ReplayProtectionBloomfilterInner {
-    // metadata to do with epochs, etc.
-    current_filter: Bloom<[u8; REPLAY_TAG_SIZE]>,
-    // overlap_filter: bloomfilter::Bloom<[u8; REPLAY_TAG_SIZE]>,
+struct RotationFilter {
+    metadata: ReplayProtectionBloomfilterMetadata,
+    data: Bloom<[u8; REPLAY_TAG_SIZE]>,
 }
 
-impl ReplayProtectionBloomfilter {
-    #[allow(dead_code)]
-    pub(crate) fn check_and_set(
-        &self,
-        replay_tag: &[u8; REPLAY_TAG_SIZE],
-    ) -> Result<bool, PoisonError<()>> {
-        let Ok(mut guard) = self.inner.lock() else {
-            return Err(PoisonError::new(()));
-        };
-
-        Ok(guard.current_filter.check_and_set(replay_tag))
-    }
-
-    #[allow(dead_code)]
-    pub(crate) fn try_check_and_set(
-        &self,
-        replay_tag: &[u8; REPLAY_TAG_SIZE],
-    ) -> Option<Result<bool, PoisonError<()>>> {
-        let mut guard = match self.inner.try_lock() {
-            Ok(guard) => guard,
-            Err(TryLockError::Poisoned(_)) => return Some(Err(PoisonError::new(()))),
-            Err(TryLockError::WouldBlock) => return None,
-        };
-
-        Some(Ok(guard.current_filter.check_and_set(replay_tag)))
-    }
-
+impl ReplayProtectionBloomfilters {
     pub(crate) fn batch_try_check_and_set(
         &self,
-        reply_tags: &[&[u8; REPLAY_TAG_SIZE]],
-    ) -> Option<Result<Vec<bool>, PoisonError<()>>> {
+        reply_tags: &HashMap<u32, Vec<&[u8; REPLAY_TAG_SIZE]>>,
+    ) -> Option<Result<HashMap<u32, Vec<bool>>, PoisonError<()>>> {
         let mut guard = match self.inner.try_lock() {
             Ok(guard) => guard,
             Err(TryLockError::Poisoned(_)) => return Some(Err(PoisonError::new(()))),
             Err(TryLockError::WouldBlock) => return None,
         };
 
-        let mut result = Vec::with_capacity(reply_tags.len());
-        for tag in reply_tags {
-            result.push(guard.current_filter.check_and_set(tag));
-        }
-
-        // for testing throughput without disabling checks:
-        // return Some(Ok(vec![false; reply_tags.len()]));
-
-        Some(Ok(result))
+        Some(Ok(guard.batch_check_and_set(&reply_tags)))
     }
 
     pub(crate) fn batch_check_and_set(
         &self,
-        reply_tags: &[&[u8; REPLAY_TAG_SIZE]],
-    ) -> Result<Vec<bool>, PoisonError<()>> {
+        reply_tags: &HashMap<u32, Vec<&[u8; REPLAY_TAG_SIZE]>>,
+    ) -> Result<HashMap<u32, Vec<bool>>, PoisonError<()>> {
         let Ok(mut guard) = self.inner.lock() else {
             return Err(PoisonError::new(()));
         };
 
-        let mut result = Vec::with_capacity(reply_tags.len());
-        for tag in reply_tags {
-            result.push(guard.current_filter.check_and_set(tag));
-        }
-
-        // for testing throughput without disabling checks:
-        // return Ok(vec![false; reply_tags.len()]);
-
-        Ok(result)
-    }
-
-    #[allow(dead_code)]
-    pub(crate) fn clear(&self) -> Result<(), PoisonError<()>> {
-        let mut guard = self.inner.lock().map_err(|_| PoisonError::new(()))?;
-        guard.current_filter.clear();
-        Ok(())
+        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<()>> {
-        let guard = self.inner.lock().map_err(|_| PoisonError::new(()))?;
-        Ok(guard.current_filter.to_bytes())
+        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)
+    // or a pre_announced (during the last epoch of the current rotation)
+    // during epoch transition, the following change will happen:
+    // primary -> secondary
+    // pre_announced -> primary
+    // I'm not using an enum because it's easier to reason about those as separate fields
+    secondary: Option<RotationFilter>,
+    pre_announced: Option<RotationFilter>,
+}
+
+impl ReplayProtectionBloomfiltersInner {
+    fn batch_check_and_set(
+        &mut self,
+        reply_tags: &HashMap<u32, Vec<&[u8; REPLAY_TAG_SIZE]>>,
+    ) -> HashMap<u32, Vec<bool>> {
+        let mut result = HashMap::with_capacity(reply_tags.len());
+        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 {
+                Some(&mut self.primary.data)
+            } else if let Some(secondary) = &mut self.secondary {
+                // 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)
+                } else {
+                    None
+                }
+            } else if let Some(pre_announced) = &mut self.pre_announced {
+                if pre_announced.metadata.rotation_id == rotation_id {
+                    Some(&mut pre_announced.data)
+                } else {
+                    None
+                }
+            } else {
+                None
+            };
+
+            let Some(mut 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()]);
+                continue;
+            };
+
+            let mut rotation_results = Vec::with_capacity(reply_tags.len());
+            for tag in reply_tags {
+                rotation_results.push(filter.check_and_set(tag))
+            }
+            result.insert(rotation_id, rotation_results);
+        }
+
+        result
     }
 }