additional bugfixes and debugging nym-api deadlock

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

@@ -26,9 +26,7 @@ impl KeyRotationState {
     }
 
     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)
+        self.current_rotation_starting_epoch_id(current_epoch_id) + self.validity_epochs
     }
 
     pub fn current_rotation_starting_epoch_id(&self, current_epoch_id: EpochId) -> EpochId {

gateway/src/node/client_handling/websocket/connection_handler/authenticated.rs

@@ -314,7 +314,8 @@ impl<R, S> AuthenticatedHandler<R, S> {
             }
             Ok(request) => match request {
                 // currently only a single type exists
-                BinaryRequest::ForwardSphinx { packet } => {
+                BinaryRequest::ForwardSphinx { packet }
+                | BinaryRequest::ForwardSphinxV2 { packet } => {
                     self.handle_forward_sphinx(packet).await.into_ws_message()
                 }
                 _ => RequestHandlingError::UnknownBinaryRequest.into_error_message(),

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

@@ -32,6 +32,7 @@ use schemars::gen::SchemaGenerator;
 use schemars::schema::{InstanceType, Schema, SchemaObject};
 use schemars::JsonSchema;
 use serde::{Deserialize, Deserializer, Serialize};
+use std::cmp::Ordering;
 use std::collections::BTreeMap;
 use std::fmt::{Debug, Display, Formatter};
 use std::net::IpAddr;
@@ -39,7 +40,7 @@ use std::ops::{Deref, DerefMut};
 use std::{fmt, time::Duration};
 use thiserror::Error;
 use time::{Date, OffsetDateTime};
-use tracing::warn;
+use tracing::{error, warn};
 use utoipa::{IntoParams, ToResponse, ToSchema};
 
 pub use nym_mixnet_contract_common::KeyRotationState;
@@ -1476,18 +1477,22 @@ impl KeyRotationInfoResponse {
             .current_rotation_starting_epoch_id(self.current_absolute_epoch_id)
     }
 
+    fn current_epoch_progress(&self, now: OffsetDateTime) -> f32 {
+        let elapsed = (now - self.current_epoch_start).as_seconds_f32();
+        elapsed / self.epoch_duration.as_secs_f32()
+    }
+
     pub fn is_epoch_stuck(&self) -> bool {
         let now = OffsetDateTime::now_utc();
-        let expected_epoch_end = self.current_epoch_start + self.epoch_duration;
-        if now > expected_epoch_end {
-            let diff = now - expected_epoch_end;
+        let progress = self.current_epoch_progress(now);
+        if progress > 1. {
+            let into_next = 1. - progress;
             // if epoch hasn't progressed for more than 20% of its duration, mark is as stuck
-            let threshold = Duration::from_secs_f32(self.epoch_duration.as_secs_f32() * 0.2);
-            if diff > threshold {
-                // SAFETY: the value is positive
-                #[allow(clippy::unwrap_used)]
-                let std_dur = std::time::Duration::try_from(diff).unwrap();
-                warn!("the current epoch is expected to have been over by {expected_epoch_end}. it's already {} overdue!", humantime_serde::re::humantime::format_duration(std_dur));
+            if into_next > 0.2 {
+                let diff_time =
+                    Duration::from_secs_f32(into_next * self.epoch_duration.as_secs_f32());
+                let expected_epoch_end = self.current_epoch_start + self.epoch_duration;
+                warn!("the current epoch is expected to have been over by {expected_epoch_end}. it's already {} overdue!", humantime_serde::re::humantime::format_duration(diff_time));
                 return true;
             }
         }
@@ -1509,12 +1514,40 @@ impl KeyRotationInfoResponse {
         let passed_epochs = diff / self.epoch_duration;
         let expected_current_epoch = self.current_absolute_epoch_id + passed_epochs.floor() as u32;
 
-        println!("expected_current_epoch: {}", expected_current_epoch);
-        println!("current_epoch: {}", self.current_absolute_epoch_id);
-
         self.key_rotation_state
             .key_rotation_id(expected_current_epoch)
     }
+
+    pub fn until_next_rotation(&self) -> Option<Duration> {
+        let current_epoch_progress = self.current_epoch_progress(OffsetDateTime::now_utc());
+        if current_epoch_progress > 1. {
+            return None;
+        }
+
+        let next_rotation_epoch = self.next_rotation_starting_epoch_id();
+        let full_remaining =
+            (next_rotation_epoch - self.current_absolute_epoch_id).checked_add(1)?;
+
+        let epochs_until_next_rotation = (1. - current_epoch_progress) + full_remaining as f32;
+
+        Some(Duration::from_secs_f32(
+            epochs_until_next_rotation * self.epoch_duration.as_secs_f32(),
+        ))
+    }
+
+    pub fn epoch_start_time(&self, absolute_epoch_id: EpochId) -> OffsetDateTime {
+        match absolute_epoch_id.cmp(&self.current_absolute_epoch_id) {
+            Ordering::Less => {
+                let diff = self.current_absolute_epoch_id - absolute_epoch_id;
+                self.current_epoch_start - diff * self.epoch_duration
+            }
+            Ordering::Equal => self.current_epoch_start,
+            Ordering::Greater => {
+                let diff = absolute_epoch_id - self.current_absolute_epoch_id;
+                self.current_epoch_start + diff * self.epoch_duration
+            }
+        }
+    }
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize, schemars::JsonSchema, ToSchema)]

nym-api/src/epoch_operations/rewarded_set_assignment.rs

@@ -212,13 +212,18 @@ impl EpochAdvancer {
 
         // SAFETY: the cache MUST HAVE been initialised before now
         #[allow(clippy::unwrap_used)]
+        warn!("⚠️⚠️⚠️⚠️⚠️ TRYING TO GET DESCRIBE CACHE");
         let described_cache = self.described_cache.get().await.unwrap();
 
+        warn!("⚠️⚠️⚠️⚠️⚠️ TRYING TO GET STATUS CACHE");
+
         let Some(status_cache) = self.status_cache.node_annotations().await else {
             warn!("there are no node annotations available");
             return Vec::new();
         };
 
+        warn!("⚠️⚠️⚠️⚠️⚠️ GOT CACHES");
+
         for nym_node in nym_nodes {
             let node_id = nym_node.node_id();
             let saturation = nym_node.rewarding_details.bond_saturation(reward_params);

nym-api/src/key_rotation/mod.rs

@@ -90,6 +90,13 @@ impl KeyRotationController {
     async fn handle_contract_cache_update(&mut self) {
         let updated = self.get_contract_data().await;
 
+        info!(
+            "current rotation: {}",
+            updated
+                .key_rotation_state
+                .key_rotation_id(updated.interval.current_epoch_absolute_id())
+        );
+
         // if we're only 1/4 epoch away from the next rotation, and we haven't yet performed the refresh,
         // update the self-described cache, as all nodes should have already pre-announced their new sphinx keys
         if let Some(remaining) = updated.epochs_until_next_rotation() {

nym-api/src/network_monitor/monitor/preparer.rs

@@ -159,13 +159,6 @@ impl PacketPreparer {
         self.contract_cache.naive_wait_for_initial_values().await;
         self.described_cache.naive_wait_for_initial_values().await;
 
-        #[allow(clippy::expect_used)]
-        let described_nodes = self
-            .described_cache
-            .get()
-            .await
-            .expect("the self-describe cache should have been initialised!");
-
         // now wait for at least `minimum_full_routes` mixnodes per layer and `minimum_full_routes` gateway to be online
         info!("Waiting for minimal topology to be online");
         let initialisation_backoff = Duration::from_secs(30);
@@ -174,6 +167,13 @@ impl PacketPreparer {
             let mixnodes = self.contract_cache.legacy_mixnodes_all_basic().await;
             let nym_nodes = self.contract_cache.nym_nodes().await;
 
+            #[allow(clippy::expect_used)]
+            let described_nodes = self
+                .described_cache
+                .get()
+                .await
+                .expect("the self-describe cache should have been initialised!");
+
             let mut gateways_count = gateways.len();
             let mut mixnodes_count = mixnodes.len();
 

nym-api/src/nym_contract_cache/handlers.rs

@@ -490,7 +490,7 @@ async fn get_current_epoch(
 #[utoipa::path(
     tag = "contract-cache",
     get,
-    path = "/epoch/key-rotation-info",
+    path = "/key-rotation-info",
     context_path = "/v1/epoch",
     responses(
         (status = 200, content(

nym-api/src/support/caching/cache.rs

@@ -7,6 +7,7 @@ use std::time::Duration;
 use thiserror::Error;
 use time::OffsetDateTime;
 use tokio::sync::{RwLock, RwLockMappedWriteGuard, RwLockReadGuard, RwLockWriteGuard};
+use tracing::{debug, warn};
 
 #[derive(Debug, Error)]
 #[error("the cache item has not been initialised")]
@@ -31,13 +32,23 @@ impl<T> SharedCache<T> {
         SharedCache::default()
     }
 
-    pub(crate) async fn update(&self, value: impl Into<T>) {
-        let mut guard = self.0.write().await;
+    pub(crate) async fn try_update(&self, value: impl Into<T>, typ: &str) -> Result<(), T> {
+        let value = value.into();
+        let mut guard = match tokio::time::timeout(Duration::from_millis(200), self.0.write()).await
+        {
+            Ok(guard) => guard,
+            Err(_) => {
+                debug!("failed to obtain write permit for {typ} cache");
+                return Err(value);
+            }
+        };
+
         if let Some(ref mut existing) = guard.inner {
             existing.unchecked_update(value)
         } else {
-            guard.inner = Some(Cache::new(value.into()))
-        }
+            guard.inner = Some(Cache::new(value))
+        };
+        Ok(())
     }
 
     pub(crate) async fn get(&self) -> Result<RwLockReadGuard<'_, Cache<T>>, UninitialisedCache> {

nym-api/src/support/caching/refresher.rs

@@ -18,6 +18,7 @@ pub struct RefreshRequester(Arc<Notify>);
 
 impl RefreshRequester {
     pub(crate) fn request_cache_refresh(&self) {
+        warn!("REQUESTING SELF DESCRIBED REFRESH");
         self.0.notify_waiters()
     }
 }
@@ -108,21 +109,44 @@ where
     // TODO: in the future offer 2 options of refreshing cache. either provide `T` directly
     // or via `FnMut(&mut T)` closure
     async fn do_refresh_cache(&self) {
-        match self.provider.try_refresh().await {
-            Ok(updated_items) => {
-                self.shared_cache.update(updated_items).await;
-                if !self.refresh_notification_sender.is_closed()
-                    && self
-                        .refresh_notification_sender
-                        .send(CacheNotification::Updated)
-                        .is_err()
-                {
-                    warn!("failed to send cache update notification");
-                }
-            }
+        let mut updated_items = match self.provider.try_refresh().await {
             Err(err) => {
-                error!("{}: failed to refresh the cache: {err}", self.name)
+                error!("{}: failed to refresh the cache: {err}", self.name);
+                return;
             }
+            Ok(items) => items,
+        };
+
+        let mut failures = 0;
+        loop {
+            match self
+                .shared_cache
+                .try_update(updated_items, &self.name)
+                .await
+            {
+                Ok(_) => break,
+                Err(returned) => {
+                    failures += 1;
+                    updated_items = returned
+                }
+            };
+            if failures % 10 == 0 {
+                warn!(
+                    "failed to obtain write permit for {} cache {failures} times in a row!",
+                    self.name
+                );
+            }
+
+            tokio::time::sleep(Duration::from_secs_f32(0.5)).await
+        }
+
+        if !self.refresh_notification_sender.is_closed()
+            && self
+                .refresh_notification_sender
+                .send(CacheNotification::Updated)
+                .is_err()
+        {
+            warn!("failed to send cache update notification");
         }
     }
 
@@ -152,6 +176,8 @@ where
                 // note: `Notify` is not cancellation safe, HOWEVER, there's only one listener,
                 // so it doesn't matter if we lose our queue position
                 _ = self.refresh_requester.0.notified() => {
+                    warn!("RECEIVED SELF DESCRIBED REFRESH REQUEST");
+
                     self.refresh(&mut task_client).await;
                     // since we just performed the full request, we can reset our existing interval
                     refresh_interval.reset();

nym-api/src/unstable_routes/v2/nym_nodes/skimmed/helpers.rs

@@ -291,8 +291,8 @@ pub(crate) async fn nodes_basic(
     ]);
 
     Ok(output.to_response(PaginatedCachedNodesResponseV2::new_full(
-        current_key_rotation,
         interval.current_epoch_absolute_id(),
+        current_key_rotation,
         refreshed_at,
         nodes,
     )))

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

@@ -51,6 +51,10 @@ struct NextAction {
 }
 
 impl NextAction {
+    fn new(typ: KeyRotationActionState, deadline: OffsetDateTime) -> Self {
+        NextAction { typ, deadline }
+    }
+
     fn until_deadline(&self) -> Duration {
         let now = OffsetDateTime::now_utc();
         Duration::try_from(self.deadline - now).unwrap_or_else(|_| {
@@ -60,10 +64,30 @@ impl NextAction {
     }
 
     fn wait(duration: Duration) -> NextAction {
-        NextAction {
-            typ: KeyRotationActionState::Wait,
-            deadline: OffsetDateTime::now_utc() + duration,
-        }
+        NextAction::new(
+            KeyRotationActionState::Wait,
+            OffsetDateTime::now_utc() + duration,
+        )
+    }
+
+    fn pre_announce(rotation_id: u32, deadline: OffsetDateTime) -> Self {
+        NextAction::new(
+            KeyRotationActionState::PreAnnounce { rotation_id },
+            deadline,
+        )
+    }
+
+    fn swap_default(expected_new_rotation: u32, deadline: OffsetDateTime) -> Self {
+        NextAction::new(
+            KeyRotationActionState::SwapDefault {
+                expected_new_rotation,
+            },
+            deadline,
+        )
+    }
+
+    fn purge_secondary(deadline: OffsetDateTime) -> Self {
+        NextAction::new(KeyRotationActionState::PurgeOld, deadline)
     }
 }
 
@@ -108,107 +132,137 @@ impl KeyRotationController {
     }
 
     async fn try_determine_next_action(&self) -> NextAction {
+        let now = OffsetDateTime::now_utc();
         let Some(key_rotation_info) = self.try_get_key_rotation_info().await else {
             warn!("failed to retrieve key rotation information");
             return NextAction::wait(Duration::from_secs(240));
         };
 
+        // check if we think the epoch is stuck (we're already 20% or more into following epoch with no advancement)
         if key_rotation_info.is_epoch_stuck() {
             warn!("the epoch is stuck - can't progress with key rotation");
             return NextAction::wait(Duration::from_secs(240));
         }
 
-        let current_epoch = key_rotation_info.current_absolute_epoch_id;
+        // >>>>> START: determine if we called this method pre-maturely due to clock skew
 
         // current rotation id as determined by the current epoch id
-        let current_rotation = key_rotation_info.current_key_rotation_id();
+        let current_rotation_id = key_rotation_info.current_key_rotation_id();
 
         // expected rotation id as determined by the current TIME
         // used to determined epoch stalling or clocks being slightly out of sync
-        let expected_current_rotation = key_rotation_info.expected_current_rotation_id();
+        let expected_current_rotation_id = key_rotation_info.expected_current_rotation_id();
 
-        if current_rotation != expected_current_rotation {
-            warn!("the current rotation is {current_rotation} whilst we expected {expected_current_rotation}");
+        if current_rotation_id != expected_current_rotation_id {
+            warn!("the current rotation is {current_rotation_id} whilst we expected {expected_current_rotation_id}");
             // if we got here, it means epoch is most likely NOT stuck (we're within the threshold)
             // so probably we prematurely called this method before nym-api(s) got to advancing
             // the epoch and thus the rotation, so wait a bit instead.
             return NextAction::wait(Duration::from_secs(30));
         }
+        // >>>>> END: determine if we called this method pre-maturely due to clock skew
 
-        // epoch id of when the new rotation id is meant to start
-        let next_rotation_epoch = key_rotation_info.next_rotation_starting_epoch_id();
+        // if we're less than 30s until next rotation, we probably started our binary in a rather
+        // unfortunate time, just wait until the next rotation rather than do all the work only to throw it
+        // away immediately
+        let Some(until_next_rotation) = key_rotation_info.until_next_rotation() else {
+            warn!("failed to determine time remaining until the next key rotation");
+            return NextAction::wait(Duration::from_secs(30));
+        };
+        if until_next_rotation < Duration::from_secs(30) {
+            debug!("less than 30s until next rotation - waiting until until then");
+            return NextAction::wait(Duration::from_secs(30));
+        }
+
+        let current_epoch = key_rotation_info.current_absolute_epoch_id;
 
         // epoch id of when the current rotation has started
-        let current_rotation_epoch = key_rotation_info.current_rotation_starting_epoch_id();
+        let current_rotation_start_epoch = key_rotation_info.current_rotation_starting_epoch_id();
 
-        let secondary_rotation_id = self.managed_keys.keys.secondary_key_rotation_id();
+        // epoch id of when the new rotation id is meant to start
+        let next_rotation_start_epoch = key_rotation_info.next_rotation_starting_epoch_id();
+
+        let secondary_key_rotation_id = self.managed_keys.keys.secondary_key_rotation_id();
+        let primary_key_rotation_id = self.managed_keys.keys.primary_key_rotation_id();
 
         debug!(
-            "current: {current_rotation}, primary: {}, secondary: {secondary_rotation_id:?}",
+            "current rotation: {current_rotation_id}, primary: {}, secondary: {secondary_key_rotation_id:?}",
             self.managed_keys.keys.primary_key_rotation_id()
         );
 
-        let next_rotation_id = current_rotation + 1;
-        // edge case for rotation 0
-        let prev_rotation_id = current_rotation.checked_sub(1);
+        let rotates_next_epoch = next_rotation_start_epoch == current_epoch + 1;
+        let next_rotation_id = current_rotation_id + 1;
 
-        let (action, execution_epoch) = match secondary_rotation_id {
-            None => {
-                debug!("no secondary key - need to pre-announce one");
-                // 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 = next_rotation_id;
+        let Some(secondary_key_rotation_id) = secondary_key_rotation_id else {
+            debug!("we don't have a secondary key");
+            // figure out if we already have appropriate key (like we crashed or this is the first time node is running)
+            // or whether we have to regenerate anything or, which is the most likely case, we're waiting to
+            // pre-announce new key for the following rotation
 
-                (
-                    KeyRotationActionState::PreAnnounce { rotation_id },
-                    next_rotation_epoch - 1,
-                )
-            }
-            Some(secondary_id) if Some(secondary_id) == prev_rotation_id => {
-                debug!("secondary key is from the previous rotation - need to remove it now");
-                // 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(secondary_id) if secondary_id == current_rotation => {
-                debug!("secondary key is from THIS rotation - we need to swap into it");
-                // 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 {
-                        expected_new_rotation: current_rotation,
-                    },
-                    current_rotation_epoch,
-                )
-            }
-            Some(secondary_id) if secondary_id == next_rotation_id => {
-                debug!("secondary key is for the NEXT rotation - we need to swap into it");
-                // 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 {
-                        expected_new_rotation: next_rotation_id,
-                    },
-                    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)
+            if primary_key_rotation_id != current_rotation_id {
+                warn!("current primary key does not correspond to the current rotation - immediately pre-announcing new key (rotates next epoch: {rotates_next_epoch}");
+                // we don't have a secondary key and our current key is already outdated -
+                // preannounce a key for either this or the next rotation
+                // (and next time this method is called, it will be promoted to primary)
+                return if rotates_next_epoch {
+                    NextAction::pre_announce(next_rotation_id, now)
+                } else {
+                    NextAction::pre_announce(current_rotation_id, now)
+                };
             }
+
+            // we have a primary key corresponding to the current rotation, so we just have to pre-announce
+            // a key for the next rotation an epoch before the rotation
+            let deadline = key_rotation_info.epoch_start_time(next_rotation_start_epoch - 1);
+            debug!(
+                "going to pre-announce secondary key for rotation {next_rotation_id} on {deadline}"
+            );
+            return NextAction::pre_announce(next_rotation_id, deadline);
         };
 
-        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;
+        // the current secondary key corresponds to the next rotation, i.e. this is the pre-announced key
+        if secondary_key_rotation_id == next_rotation_id {
+            debug!("secondary key is for the NEXT rotation - we need to swap into it");
 
-        NextAction {
-            typ: action,
-            deadline: now - since_epoch_start + until_execution_epoch,
+            let deadline = key_rotation_info.epoch_start_time(next_rotation_start_epoch);
+            return NextAction::swap_default(next_rotation_id, deadline);
         }
+
+        if secondary_key_rotation_id == current_rotation_id {
+            debug!("secondary key is for the CURRENT rotation - we need to swap into it");
+
+            return NextAction::swap_default(current_rotation_id, now);
+        }
+
+        if secondary_key_rotation_id < current_rotation_id {
+            let deadline = if secondary_key_rotation_id == current_rotation_id - 1 {
+                debug!("secondary key is from the PREVIOUS rotations - we need to purge it");
+                // we purge the key after the end of overlap period, i.e. during the 2nd epoch of a rotation
+                key_rotation_info.epoch_start_time(current_rotation_start_epoch + 1)
+            } else {
+                debug!("secondary key is from AN OLD rotation - we need to purge it");
+                // the key is from some old rotation, we were probably offline for some time - we need to pre-announce new key
+                // for the upcoming rotation, so start off by purging this key immediately
+                now
+            };
+
+            return NextAction::purge_secondary(deadline);
+        }
+
+        // at this point all branches should have been covered, i.e. missing secondary key,
+        // secondary key == next rotation
+        // secondary key == current rotation
+        // secondary key < current rotation
+        // the only, theoretical, branch is if secondary key was from few rotations in the future,
+        // but this would require some weird chain shenanigans
+        error!("this code branch should have been unreachable - please report if you see this error with the following information:\
+            primary_key_rotation = {primary_key_rotation_id},
+            secondary_key_rotation = {secondary_key_rotation_id},
+            current_rotation = {current_rotation_id},
+            next_rotation = {next_rotation_id},
+            raw_response = {key_rotation_info:?}");
+
+        NextAction::wait(Duration::from_secs(240))
     }
 
     async fn try_get_key_rotation_info(&self) -> Option<KeyRotationInfoResponse> {

nym-node/src/node/nym_apis_client.rs

@@ -24,6 +24,8 @@ pub struct NymApisClient {
     inner: Arc<RwLock<InnerClient>>,
 }
 
+const TODO: &str = "add shutdown signal to cancel any queries if received";
+
 struct InnerClient {
     active_client: NymApiClient,
     available_urls: Vec<Url>,
@@ -42,6 +44,7 @@ impl NymApisClient {
         let active_client = nym_http_api_client::Client::builder(urls[0].clone())?
             .no_hickory_dns()
             .with_user_agent(NymNode::user_agent())
+            .with_timeout(Duration::from_secs(5))
             .build()?;
 
         Ok(NymApisClient {

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

@@ -80,13 +80,16 @@ impl ReplayProtectionBloomfilters {
     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 next = primary_id + 1;
+        let previous = primary_id.checked_sub(1);
         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 {
+                if secondary_id == next {
                     (None, Some(secondary_filter))
-                } else if secondary_id == primary_id - 1 {
+                } else if Some(secondary_id) == previous {
                     (Some(secondary_filter), None)
                 } else {
                     warn!("{secondary_id} is not valid for either pre_announced or overlap bloomfilter given primary rotation of {primary_id}");

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

@@ -90,7 +90,7 @@ impl ReplayProtectionBloomfiltersManager {
 
         let primary = self.filters.primary_metadata()?;
         let time_delta = OffsetDateTime::now_utc() - primary.creation_time;
-        let received_since_creation = received - primary.packets_received_at_creation;
+        let received_since_creation = received.saturating_sub(primary.packets_received_at_creation);
         let received_per_second =
             (received_since_creation as f64 / time_delta.as_seconds_f64()).round() as usize;