appease clippy

Merge/release/2026.5 raclette

Cargo.lock

@@ -2772,6 +2772,9 @@ name = "fastrand"
 version = "2.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "37909eebbb50d72f9059c3b6d82c0463f2ff062c9e95845c43a6c9c0355411be"
+dependencies = [
+ "getrandom 0.2.16",
+]
 
 [[package]]
 name = "ff"
@@ -2844,6 +2847,18 @@ dependencies = [
  "spin",
 ]
 
+[[package]]
+name = "flume"
+version = "0.12.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be"
+dependencies = [
+ "fastrand",
+ "futures-core",
+ "futures-sink",
+ "spin",
+]
+
 [[package]]
 name = "fnv"
 version = "1.0.7"
@@ -7149,6 +7164,7 @@ dependencies = [
  "csv",
  "cupid",
  "dashmap",
+ "flume 0.12.0",
  "futures",
  "hex",
  "hkdf",
@@ -11069,7 +11085,7 @@ checksum = "c2d12fe70b2c1b4401038055f90f151b78208de1f9f89a7dbfd41587a10c3eea"
 dependencies = [
  "atoi",
  "chrono",
- "flume",
+ "flume 0.11.1",
  "futures-channel",
  "futures-core",
  "futures-executor",

Cargo.toml

@@ -269,6 +269,7 @@ etherparse = "0.13.0"
 eyre = "0.6.9"
 fastrand = "2.1.1"
 flate2 = "1.1.1"
+flume = "0.12.0"
 futures = "0.3.31"
 futures-util = "0.3"
 generic-array = "0.14.7"

common/client-libs/validator-client/src/client.rs

@@ -23,7 +23,7 @@ use nym_api_requests::models::{
     MixnodeCoreStatusResponse, NymNodeDescriptionV1, NymNodeDescriptionV2,
 };
 use nym_api_requests::nym_nodes::{
-    NodesByAddressesResponse, SemiSkimmedNodesWithMetadata, SkimmedNode, SkimmedNodesWithMetadata,
+    NodesByAddressesResponse, SemiSkimmedNodesWithMetadata, SkimmedNodeV1, SkimmedNodesWithMetadata,
 };
 use nym_coconut_dkg_common::types::EpochId;
 use nym_http_api_client::UserAgent;
@@ -354,12 +354,12 @@ impl NymApiClient {
     }
 
     #[deprecated(note = "use get_all_basic_active_mixing_assigned_nodes instead")]
-    pub async fn get_basic_mixnodes(&self) -> Result<Vec<SkimmedNode>, ValidatorClientError> {
+    pub async fn get_basic_mixnodes(&self) -> Result<Vec<SkimmedNodeV1>, ValidatorClientError> {
         Ok(self.nym_api.get_basic_mixnodes().await?.nodes)
     }
 
     #[deprecated(note = "use get_all_basic_entry_assigned_nodes instead")]
-    pub async fn get_basic_gateways(&self) -> Result<Vec<SkimmedNode>, ValidatorClientError> {
+    pub async fn get_basic_gateways(&self) -> Result<Vec<SkimmedNodeV1>, ValidatorClientError> {
         Ok(self.nym_api.get_basic_gateways().await?.nodes)
     }
 
@@ -372,7 +372,7 @@ impl NymApiClient {
     #[deprecated(note = "use get_all_basic_entry_assigned_nodes_with_metadata instead")]
     pub async fn get_all_basic_entry_assigned_nodes(
         &self,
-    ) -> Result<Vec<SkimmedNode>, ValidatorClientError> {
+    ) -> Result<Vec<SkimmedNodeV1>, ValidatorClientError> {
         self.get_all_basic_entry_assigned_nodes_with_metadata()
             .await
             .map(|res| res.nodes)
@@ -389,7 +389,7 @@ impl NymApiClient {
     #[deprecated(note = "use get_all_basic_active_mixing_assigned_nodes_with_metadata instead")]
     pub async fn get_all_basic_active_mixing_assigned_nodes(
         &self,
-    ) -> Result<Vec<SkimmedNode>, ValidatorClientError> {
+    ) -> Result<Vec<SkimmedNodeV1>, ValidatorClientError> {
         self.get_all_basic_active_mixing_assigned_nodes_with_metadata()
             .await
             .map(|res| res.nodes)
@@ -406,7 +406,7 @@ impl NymApiClient {
     #[deprecated(note = "use get_all_basic_mixing_capable_nodes_with_metadata instead")]
     pub async fn get_all_basic_mixing_capable_nodes(
         &self,
-    ) -> Result<Vec<SkimmedNode>, ValidatorClientError> {
+    ) -> Result<Vec<SkimmedNodeV1>, ValidatorClientError> {
         self.get_all_basic_mixing_capable_nodes_with_metadata()
             .await
             .map(|res| res.nodes)
@@ -420,7 +420,7 @@ impl NymApiClient {
 
     /// retrieve basic information for all bonded nodes on the network
     #[deprecated(note = "use get_all_basic_nodes_with_metadata instead")]
-    pub async fn get_all_basic_nodes(&self) -> Result<Vec<SkimmedNode>, ValidatorClientError> {
+    pub async fn get_all_basic_nodes(&self) -> Result<Vec<SkimmedNodeV1>, ValidatorClientError> {
         self.get_all_basic_nodes_with_metadata()
             .await
             .map(|res| res.nodes)

common/client-libs/validator-client/src/nym_api/mod.rs

@@ -4,6 +4,7 @@
 use crate::nym_api::error::NymAPIError;
 use crate::nym_api::routes::{ecash, CORE_STATUS_COUNT, SINCE_ARG};
 use crate::nym_nodes::SkimmedNodesWithMetadata;
+use crate::ValidatorClientError;
 use async_trait::async_trait;
 use nym_api_requests::ecash::models::{
     AggregatedCoinIndicesSignatureResponse, AggregatedExpirationDateSignatureResponse,
@@ -20,11 +21,14 @@ use nym_api_requests::models::{
     NymNodeDescriptionV1, NymNodeDescriptionV2, PerformanceHistoryResponse, RewardedSetResponse,
     SignerInformationResponse,
 };
-use nym_api_requests::nym_nodes::{
-    NodesByAddressesRequestBody, NodesByAddressesResponse, PaginatedCachedNodesResponseV1,
-    PaginatedCachedNodesResponseV2,
-};
 use nym_api_requests::pagination::PaginatedResponse;
+use nym_http_api_client::{ApiClient, NO_PARAMS};
+use nym_mixnet_contract_common::{IdentityKeyRef, NodeId, NymNodeDetails};
+use std::net::IpAddr;
+use time::format_description::BorrowedFormatItem;
+use time::Date;
+use tracing::instrument;
+
 pub use nym_api_requests::{
     ecash::{
         models::SpentCredentialsResponse, BlindSignRequestBody, BlindedSignatureResponse,
@@ -36,17 +40,14 @@ pub use nym_api_requests::{
         MixnodeCoreStatusResponse, MixnodeStatusReportResponse, MixnodeStatusResponse,
         MixnodeUptimeHistoryResponse, StakeSaturationResponse, UptimeResponse,
     },
-    nym_nodes::{CachedNodesResponse, SemiSkimmedNode, SemiSkimmedNodesWithMetadata, SkimmedNode},
+    nym_nodes::{
+        CachedNodesResponse, NodesByAddressesRequestBody, NodesByAddressesResponse,
+        PaginatedCachedNodesResponseV1, PaginatedCachedNodesResponseV2, SemiSkimmedNodeV1,
+        SemiSkimmedNodeV3, SemiSkimmedNodesWithMetadata, SkimmedNodeV1,
+    },
     NymNetworkDetailsResponse,
 };
-use nym_http_api_client::{ApiClient, NO_PARAMS};
-use nym_mixnet_contract_common::{IdentityKeyRef, NodeId, NymNodeDetails};
-use std::net::IpAddr;
-use time::format_description::BorrowedFormatItem;
-use time::Date;
-use tracing::instrument;
 
-use crate::ValidatorClientError;
 pub use nym_coconut_dkg_common::types::EpochId;
 
 pub mod error;
@@ -390,7 +391,7 @@ pub trait NymApiClientExt: ApiClient {
 
     #[deprecated]
     #[tracing::instrument(level = "debug", skip_all)]
-    async fn get_basic_mixnodes(&self) -> Result<CachedNodesResponse<SkimmedNode>, NymAPIError> {
+    async fn get_basic_mixnodes(&self) -> Result<CachedNodesResponse<SkimmedNodeV1>, NymAPIError> {
         self.get_json(
             &[
                 routes::V1_API_VERSION,
@@ -406,7 +407,7 @@ pub trait NymApiClientExt: ApiClient {
 
     #[deprecated]
     #[instrument(level = "debug", skip(self))]
-    async fn get_basic_gateways(&self) -> Result<CachedNodesResponse<SkimmedNode>, NymAPIError> {
+    async fn get_basic_gateways(&self) -> Result<CachedNodesResponse<SkimmedNodeV1>, NymAPIError> {
         self.get_json(
             &[
                 routes::V1_API_VERSION,
@@ -443,7 +444,7 @@ pub trait NymApiClientExt: ApiClient {
         page: Option<u32>,
         per_page: Option<u32>,
         use_bincode: bool,
-    ) -> Result<PaginatedCachedNodesResponseV1<SkimmedNode>, NymAPIError> {
+    ) -> Result<PaginatedCachedNodesResponseV1<SkimmedNodeV1>, NymAPIError> {
         let mut params = Vec::new();
 
         if no_legacy {
@@ -485,7 +486,7 @@ pub trait NymApiClientExt: ApiClient {
         page: Option<u32>,
         per_page: Option<u32>,
         use_bincode: bool,
-    ) -> Result<PaginatedCachedNodesResponseV2<SkimmedNode>, NymAPIError> {
+    ) -> Result<PaginatedCachedNodesResponseV2<SkimmedNodeV1>, NymAPIError> {
         let mut params = Vec::new();
 
         if no_legacy {
@@ -527,7 +528,7 @@ pub trait NymApiClientExt: ApiClient {
         page: Option<u32>,
         per_page: Option<u32>,
         use_bincode: bool,
-    ) -> Result<PaginatedCachedNodesResponseV1<SkimmedNode>, NymAPIError> {
+    ) -> Result<PaginatedCachedNodesResponseV1<SkimmedNodeV1>, NymAPIError> {
         let mut params = Vec::new();
 
         if no_legacy {
@@ -569,7 +570,7 @@ pub trait NymApiClientExt: ApiClient {
         page: Option<u32>,
         per_page: Option<u32>,
         use_bincode: bool,
-    ) -> Result<PaginatedCachedNodesResponseV2<SkimmedNode>, NymAPIError> {
+    ) -> Result<PaginatedCachedNodesResponseV2<SkimmedNodeV1>, NymAPIError> {
         let mut params = Vec::new();
 
         if no_legacy {
@@ -612,7 +613,7 @@ pub trait NymApiClientExt: ApiClient {
         page: Option<u32>,
         per_page: Option<u32>,
         use_bincode: bool,
-    ) -> Result<PaginatedCachedNodesResponseV1<SkimmedNode>, NymAPIError> {
+    ) -> Result<PaginatedCachedNodesResponseV1<SkimmedNodeV1>, NymAPIError> {
         let mut params = Vec::new();
 
         if no_legacy {
@@ -654,7 +655,7 @@ pub trait NymApiClientExt: ApiClient {
         page: Option<u32>,
         per_page: Option<u32>,
         use_bincode: bool,
-    ) -> Result<PaginatedCachedNodesResponseV2<SkimmedNode>, NymAPIError> {
+    ) -> Result<PaginatedCachedNodesResponseV2<SkimmedNodeV1>, NymAPIError> {
         let mut params = Vec::new();
 
         if no_legacy {
@@ -695,7 +696,7 @@ pub trait NymApiClientExt: ApiClient {
         page: Option<u32>,
         per_page: Option<u32>,
         use_bincode: bool,
-    ) -> Result<PaginatedCachedNodesResponseV1<SkimmedNode>, NymAPIError> {
+    ) -> Result<PaginatedCachedNodesResponseV1<SkimmedNodeV1>, NymAPIError> {
         let mut params = Vec::new();
 
         if no_legacy {
@@ -733,7 +734,7 @@ pub trait NymApiClientExt: ApiClient {
         page: Option<u32>,
         per_page: Option<u32>,
         use_bincode: bool,
-    ) -> Result<PaginatedCachedNodesResponseV2<SkimmedNode>, NymAPIError> {
+    ) -> Result<PaginatedCachedNodesResponseV2<SkimmedNodeV1>, NymAPIError> {
         let mut params = Vec::new();
 
         if no_legacy {
@@ -770,7 +771,7 @@ pub trait NymApiClientExt: ApiClient {
         no_legacy: bool,
         page: Option<u32>,
         per_page: Option<u32>,
-    ) -> Result<PaginatedCachedNodesResponseV2<SemiSkimmedNode>, NymAPIError> {
+    ) -> Result<PaginatedCachedNodesResponseV2<SemiSkimmedNodeV1>, NymAPIError> {
         let mut params = Vec::new();
 
         if no_legacy {
@@ -797,6 +798,21 @@ pub trait NymApiClientExt: ApiClient {
         .await
     }
 
+    #[instrument(level = "debug", skip(self))]
+    async fn get_expanded_nodes_v3(
+        &self,
+        use_bincode: bool,
+    ) -> Result<PaginatedCachedNodesResponseV2<SemiSkimmedNodeV3>, NymAPIError> {
+        let mut params = Vec::new();
+
+        if use_bincode {
+            params.push(("output", "bincode".to_string()))
+        }
+
+        self.get_response("/v3/unstable/nym-nodes/semi-skimmed", &params)
+            .await
+    }
+
     #[deprecated]
     #[instrument(level = "debug", skip(self))]
     async fn get_mixnode_report(

common/client-libs/validator-client/src/nym_api/routes.rs

@@ -3,6 +3,7 @@
 
 pub const V1_API_VERSION: &str = "v1";
 pub const V2_API_VERSION: &str = "v2";
+pub const V3_API_VERSION: &str = "v3";
 pub const MIXNODES: &str = "mixnodes";
 pub const GATEWAYS: &str = "gateways";
 pub const DESCRIBED: &str = "described";

common/http-api-client/src/lib.rs

@@ -1401,6 +1401,7 @@ pub trait ApiClient: ApiClientCore {
     /// 'get' data from the segment-defined path, e.g. `["api", "v1", "mixnodes"]`, with tuple
     /// defined key-value parameters, e.g. `[("since", "12345")]`. Attempt to parse the response
     /// into the provided type `T` based on the content type header
+    #[instrument(level = "debug", skip_all, fields(path=?path))]
     async fn get_response<P, T, K, V>(
         &self,
         path: P,

common/nym-kkt/src/error.rs

@@ -3,6 +3,7 @@
 
 use crate::context::KKTStatus;
 use nym_kkt_ciphersuite::error::KKTCiphersuiteError;
+use nym_kkt_ciphersuite::{HashFunction, KEM};
 use nym_kkt_context::KKTContextEncodingError;
 use std::fmt::Debug;
 use thiserror::Error;
@@ -62,6 +63,14 @@ pub enum KKTError {
     #[error("the received encapsulation key hash does not match the expected value")]
     MismatchedKEMHash,
 
+    #[error(
+        "there are no known digests for initiator's KEM key with {kem} KEM and {hash_function} hash function"
+    )]
+    NoKnownKEMKeyDigests {
+        kem: KEM,
+        hash_function: HashFunction,
+    },
+
     #[error(transparent)]
     MalformedContext(#[from] KKTContextEncodingError),
 }

common/nym-kkt/src/initiator.rs

@@ -140,14 +140,7 @@ pub fn initiator_process(
         },
     };
 
-    Ok(KKTFrame::new(
-        context,
-        body,
-        match payload {
-            Some(payload_vec) => payload_vec,
-            None => Vec::with_capacity(0),
-        },
-    ))
+    Ok(KKTFrame::new(context, body, payload.unwrap_or_default()))
 }
 
 pub fn initiator_ingest_response(

common/nym-kkt/src/lib.rs

@@ -18,6 +18,7 @@ pub use nym_kkt_context as context;
 mod test {
     use nym_kkt_ciphersuite::{Ciphersuite, HashFunction, HashLength, KEM, SignatureScheme};
     use rand09::RngCore;
+    use std::collections::BTreeMap;
 
     use crate::keys::KEMKeys;
     use crate::{
@@ -78,9 +79,12 @@ mod test {
                 initiator_mceliece_keypair.pk.as_ref(),
             );
 
+            let init_hashes = BTreeMap::new();
+
             let responder = KKTResponder::new(
                 &responder_x25519_keypair,
                 &responder_kem,
+                &init_hashes,
                 &[
                     HashFunction::Blake3,
                     HashFunction::SHA256,

common/nym-kkt/src/responder.rs

@@ -10,7 +10,8 @@ use crate::{
     frame::KKTFrame,
 };
 use libcrux_psq::handshake::types::DHKeyPair;
-use nym_kkt_ciphersuite::{Ciphersuite, HashFunction, SignatureScheme};
+use nym_kkt_ciphersuite::{Ciphersuite, HashFunction, KEM, KEMKeyDigests, SignatureScheme};
+use std::collections::BTreeMap;
 
 /// Representation of a KKT Responder
 pub struct KKTResponder<'a> {
@@ -20,6 +21,9 @@ pub struct KKTResponder<'a> {
     /// KEM keys of this responder
     kem_keys: &'a KEMKeys,
 
+    /// Digests of the initiator's kem key
+    expected_initiator_kem_digests: &'a BTreeMap<KEM, KEMKeyDigests>,
+
     /// List of supported Hash Functions by this Responder
     supported_hash_functions: Vec<HashFunction>,
 
@@ -34,6 +38,7 @@ impl<'a> KKTResponder<'a> {
     pub fn new(
         x25519_keypair: &'a DHKeyPair,
         kem_keys: &'a KEMKeys,
+        expected_initiator_kem_digests: &'a BTreeMap<KEM, KEMKeyDigests>,
         supported_hash_functions: &[HashFunction],
         supported_signature_schemes: &[SignatureScheme],
         supported_outer_protocol_versions: &[u8],
@@ -59,12 +64,28 @@ impl<'a> KKTResponder<'a> {
         Ok(Self {
             x25519_keypair,
             kem_keys,
+            expected_initiator_kem_digests,
             supported_hash_functions: supported_hash_functions.to_vec(),
             supported_signature_schemes: supported_signature_schemes.to_vec(),
             supported_outer_protocol_versions: supported_outer_protocol_versions.to_vec(),
         })
     }
 
+    /// Attempt to retrieve expected KEM key hash of the initiator based on the received `Ciphersuite`
+    pub(crate) fn expected_initiator_kem_digest(
+        &self,
+        ciphersuite: Ciphersuite,
+    ) -> Result<&Vec<u8>, KKTError> {
+        let kem = ciphersuite.kem();
+        let hash_function = ciphersuite.hash_function();
+
+        self.expected_initiator_kem_digests
+            .get(&kem)
+            .ok_or(KKTError::NoKnownKEMKeyDigests { kem, hash_function })?
+            .get(&hash_function)
+            .ok_or(KKTError::NoKnownKEMKeyDigests { kem, hash_function })
+    }
+
     fn check_ciphersuite_compatiblity(
         &self,
         remote_ciphersuite: Ciphersuite,
@@ -102,6 +123,7 @@ impl<'a> KKTResponder<'a> {
         )?;
 
         let remote_context = *processed_req.remote_context();
+
         let remote_frame = processed_req.remote_frame;
         let request_payload = remote_frame.payload().to_vec();
         let mut carrier = processed_req.carrier;
@@ -111,12 +133,8 @@ impl<'a> KKTResponder<'a> {
         let (local_context, remote_encapsulation_key) = match remote_context.mode() {
             KKTMode::OneWay => responder_ingest_message(None, remote_frame)?,
             KKTMode::Mutual => {
-                // So we can either fetch the remote hash here using some async call to the directory,
-                // which might make registration hang or accept the sent key then verify later.
-
-                // If we choose to not accept, the response's status will be KKTStatus::UnverifiedKEMKey.
-                // The response would still contain the responder's encapsulation key.
-                responder_ingest_message(None, remote_frame)?
+                let digest = self.expected_initiator_kem_digest(remote_context.ciphersuite())?;
+                responder_ingest_message(Some(digest), remote_frame)?
             }
         };
 
@@ -162,14 +180,6 @@ pub fn responder_ingest_message(
                         own_context.update_status(KKTStatus::UnverifiedKEMKey);
                         // we don't store an unverified key
                         // changing the status notifies the initiator that we didn't
-
-                        // we could still keep it here and then verify later...
-                        // let received_encapsulation_key = EncapsulationKey::decode(
-                        //     own_context.ciphersuite().kem(),
-                        //     remote_frame.body_ref(),
-                        // )?;
-                        //  Ok((own_context, Some(received_encapsulation_key)))
-                        //
                         return Ok((own_context, None));
                     };
 

common/nym-lp/src/error.rs

@@ -65,12 +65,12 @@ pub enum LpError {
     #[error("State machine not found for lp_id: {0}")]
     StateMachineNotFound(LpReceiverIndex),
 
-    // /// Ed25519 to X25519 conversion error.
-    // #[error("Ed25519 key conversion error: {0}")]
-    // Ed25519RecoveryError(#[from] Ed25519RecoveryError),
     #[error("attempted to create an LP responder without providing a valid KEM keys")]
     ResponderWithMissingKEMKeys,
 
+    #[error("attempted to create an LP mutual initiator without providing a valid KEM key")]
+    PSQMutualInitiatorMissingKemKey,
+
     #[error(
         "there are no known digests for remote's KEM key with {kem} KEM and {hash_function} hash function"
     )]

common/nym-lp/src/lib.rs

@@ -109,6 +109,7 @@ impl SessionsMock {
             initiator_authenticator,
             responder_ecdh_pk: resp_remote.x25519_public,
             responder_pq_pk: Some(encapsulation_key),
+            initiator_pq_pk: None,
         };
 
         SessionsMock {

common/nym-lp/src/peer.rs

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::LpError;
+use nym_kkt::keys::EncapsulationKey;
 use nym_kkt_ciphersuite::{Ciphersuite, KEM, KEMKeyDigests};
 use std::collections::BTreeMap;
 use std::fmt::Debug;
@@ -41,6 +42,18 @@ impl LpLocalPeer {
         self
     }
 
+    pub fn kem_key(&self, kem: KEM) -> Option<EncapsulationKey> {
+        self.kem_keypairs
+            .as_ref()
+            .and_then(|k| k.encapsulation_key(kem))
+    }
+
+    pub fn encoded_kem_key(&self, kem: KEM) -> Option<&[u8]> {
+        self.kem_keypairs
+            .as_ref()
+            .and_then(|k| k.encoded_encapsulation_key(kem))
+    }
+
     pub fn x25519(&self) -> &Arc<DHKeyPair> {
         &self.x25519
     }

common/nym-lp/src/psq/initiator.rs

@@ -27,6 +27,7 @@ use tracing::debug;
 pub struct PSQHandshakeStateInitiator<'a, S> {
     pub(super) inner_state: PSQHandshakeState<'a, S>,
     pub(super) initiator_data: InitiatorData,
+    pub(super) mutual: bool,
 }
 
 pub(crate) fn build_psq_principal<R>(
@@ -77,6 +78,15 @@ impl<'a, S> PSQHandshakeStateInitiator<'a, S>
 where
     S: LpHandshakeChannel + Unpin,
 {
+    pub fn set_mutual_kkt(mut self) -> Result<Self, LpError> {
+        if self.inner_state.local_peer.kem_keypairs.is_none() {
+            return Err(LpError::PSQMutualInitiatorMissingKemKey);
+        }
+
+        self.mutual = true;
+        Ok(self)
+    }
+
     /// Attempt to send KKT request to begin the handshake
     async fn send_kkt_request(&mut self, request: KKTRequest) -> Result<(), LpError> {
         let kem = self.inner_state.local_peer.ciphersuite.kem();
@@ -128,16 +138,34 @@ where
             .expected_kem_key_hash(ciphersuite)?;
 
         // 2. prepare and send KKT request
-        let (mut initiator, kkt_request) = KKTInitiator::generate_one_way_request(
-            rng,
-            ciphersuite,
-            self.initiator_data.remote_peer.x25519(),
-            &dir_hash,
-            self.initiator_data.protocol_version,
-            Some(Vec::from(lp_peer_config.serialize())),
-        )?;
-        // derive the receiver index from the request
-        // let receiver_index = kkt_request
+        let (mut initiator, kkt_request) = if self.mutual {
+            // this has been verified when setting the mutual flag
+            let Some(local_encapsulation_key) = self.inner_state.local_peer.encoded_kem_key(kem)
+            else {
+                return Err(LpError::PSQMutualInitiatorMissingKemKey);
+            };
+
+            KKTInitiator::generate_mutual_request(
+                rng,
+                ciphersuite,
+                local_encapsulation_key,
+                self.initiator_data.remote_peer.x25519(),
+                &dir_hash,
+                self.initiator_data.protocol_version,
+                Some(Vec::from(lp_peer_config.serialize())),
+            )?
+        } else {
+            KKTInitiator::generate_one_way_request(
+                rng,
+                ciphersuite,
+                self.initiator_data.remote_peer.x25519(),
+                &dir_hash,
+                self.initiator_data.protocol_version,
+                Some(Vec::from(lp_peer_config.serialize())),
+            )?
+        };
+
+        let init_kem_key = self.inner_state.local_peer.kem_key(kem);
 
         debug!("sending KKT request");
         self.send_kkt_request(kkt_request).await?;
@@ -154,7 +182,7 @@ where
         let conn = self.inner_state.connection;
 
         // note: the clone is cheap due to internal Arcs
-        let encapsulation_key = response.encapsulation_key.clone();
+        let resp_encapsulation_key = response.encapsulation_key.clone();
 
         // build the PSQ initiator
         let initiator_ciphersuite = build_psq_ciphersuite(
@@ -191,13 +219,14 @@ where
 
         let initiator_authenticator = Authenticator::Dh(self.inner_state.local_peer.x25519().pk);
 
-        let receiver_index =
-            lp_peer_config.derive_receiver_index(&initiator_authenticator, &encapsulation_key)?;
+        let receiver_index = lp_peer_config
+            .derive_receiver_index(&initiator_authenticator, &resp_encapsulation_key)?;
 
         let binding = PersistentSessionBinding {
             initiator_authenticator,
             responder_ecdh_pk: self.initiator_data.remote_peer.x25519_public,
-            responder_pq_pk: Some(encapsulation_key),
+            responder_pq_pk: Some(resp_encapsulation_key),
+            initiator_pq_pk: init_kem_key,
         };
 
         let psq_session = psq_initiator.into_session()?;
@@ -218,6 +247,7 @@ mod tests {
     use nym_test_utils::helpers::{DeterministicRng09Send, u64_seeded_rng_09};
     use nym_test_utils::mocks::async_read_write::MockIOStream;
     use nym_test_utils::traits::{Leak, Timeboxed};
+    use std::collections::BTreeMap;
 
     #[tokio::test]
     async fn initiator_test_plain() -> anyhow::Result<()> {
@@ -225,6 +255,8 @@ mod tests {
             let conn_init = MockIOStream::default();
             let conn_resp = conn_init.try_get_remote_handle();
 
+            let dir_hash_init = BTreeMap::new();
+
             // leak the connections (JUST FOR THE PURPOSE OF THIS TEST!)
             // so they'd get 'static lifetime
             let conn_init = conn_init.leak();
@@ -264,6 +296,7 @@ mod tests {
             let kkt_responder = KKTResponder::new(
                 responder_x25519_keypair,
                 resp_keys,
+                &dir_hash_init,
                 &supported_hash,
                 &supported_sigs,
                 &[1],
@@ -337,4 +370,127 @@ mod tests {
         }
         Ok(())
     }
+
+    #[tokio::test]
+    async fn initiator_test_plain_mutual() -> anyhow::Result<()> {
+        for kem in KEM::iter() {
+            let conn_init = MockIOStream::default();
+            let conn_resp = conn_init.try_get_remote_handle();
+
+            // leak the connections (JUST FOR THE PURPOSE OF THIS TEST!)
+            // so they'd get 'static lifetime
+            let conn_init = conn_init.leak();
+            let conn_resp = conn_resp.leak();
+
+            let (mut init, mut resp) = mock_peers();
+            let resp_remote = resp.as_remote();
+            let init_remote = init.as_remote();
+            let dir_hash_init = init_remote.expected_kem_key_digests;
+
+            let ciphersuite = Ciphersuite::default().with_kem(kem);
+            init.ciphersuite = ciphersuite;
+            resp.ciphersuite = ciphersuite;
+            let initiator_data = InitiatorData::new(1, resp_remote);
+
+            let handshake_init = PSQHandshakeState::new(conn_init, init)
+                .as_initiator(initiator_data)
+                .set_mutual_kkt()?;
+
+            let mut init_rng = DeterministicRng09Send::new(u64_seeded_rng_09(1));
+
+            let init_fut = tokio::spawn(async move {
+                handshake_init
+                    .complete_handshake_with_rng(&mut init_rng)
+                    .timeboxed()
+                    .await
+            });
+
+            // responder:
+            let supported_sigs = [SignatureScheme::Ed25519];
+            let supported_hash = [
+                HashFunction::Blake3,
+                HashFunction::Shake256,
+                HashFunction::Shake128,
+                HashFunction::SHA256,
+            ];
+            let resp_keys = resp.kem_keypairs.as_ref().unwrap();
+            let responder_x25519_keypair = resp.x25519();
+
+            let kkt_responder = KKTResponder::new(
+                responder_x25519_keypair,
+                resp_keys,
+                &dir_hash_init,
+                &supported_hash,
+                &supported_sigs,
+                &[1],
+            )?;
+
+            // 1. read KKT request
+            let raw_kkt_req: handshake_message::KKTRequest = conn_resp
+                .receive_handshake_message(
+                    KKTRequest::size_excluding_payload(KKTMode::Mutual, kem) + LP_PEER_CONFIG_SIZE,
+                )
+                .timeboxed()
+                .await??;
+            let req = raw_kkt_req.into();
+
+            // 2. process
+            let processed_req = kkt_responder.process_request(req, LP_PEER_CONFIG_SIZE)?;
+            conn_resp
+                .send_handshake_message::<handshake_message::KKTResponse>(
+                    processed_req.response.into(),
+                    kem,
+                )
+                .timeboxed()
+                .await??;
+
+            // 3. read PSQ req
+            let responder_ciphersuite = responder::build_psq_ciphersuite(&resp, kem)?;
+            let mut responder =
+                responder::build_psq_principal(rand09::rng(), 1, responder_ciphersuite)?;
+            let response_len = psq_msg1_size(kem);
+
+            let msg: PSQMsg1 = conn_resp
+                .receive_handshake_message(response_len)
+                .timeboxed()
+                .await??;
+            responder.read_message(&msg, &mut []).unwrap();
+
+            // 4 send PSQ response
+            let mut buf = vec![0u8; PSQ_MSG2_SIZE];
+            let n = responder.write_message(&[], &mut buf).unwrap();
+            assert_eq!(n, buf.len());
+            let msg = PSQMsg2::new(buf);
+            conn_resp
+                .send_handshake_message(msg, kem)
+                .timeboxed()
+                .await??;
+
+            assert!(responder.is_handshake_finished());
+
+            let mut session_init = init_fut.await???;
+
+            let mut r_transport = responder.into_session().unwrap();
+
+            // test serialization, deserialization
+            let channel_i = session_init.active_transport();
+            let mut channel_r = r_transport.transport_channel().unwrap();
+
+            assert_eq!(channel_i.identifier(), channel_r.identifier());
+
+            let app_data_i = b"Derived session hey".as_slice();
+            let app_data_r = b"Derived session ho".as_slice();
+
+            let ct_i = encrypt_data(app_data_i, channel_i)?;
+            let pt_r = decrypt_data(&ct_i, &mut channel_r)?;
+
+            assert_eq!(app_data_i, pt_r);
+
+            let ct_r = encrypt_data(app_data_r, &mut channel_r)?;
+            let pt_i = decrypt_data(&ct_r, channel_i)?;
+
+            assert_eq!(app_data_r, pt_i);
+        }
+        Ok(())
+    }
 }

common/nym-lp/src/psq/mod.rs

@@ -4,7 +4,8 @@
 use crate::packet::version;
 use crate::peer::{LpLocalPeer, LpRemotePeer};
 use crate::transport::traits::LpHandshakeChannel;
-use nym_kkt_ciphersuite::{HashFunction, IntoEnumIterator, KEM, SignatureScheme};
+use nym_kkt_ciphersuite::{HashFunction, IntoEnumIterator, KEM, KEMKeyDigests, SignatureScheme};
+use std::collections::BTreeMap;
 
 pub(crate) mod handshake_message;
 mod helpers;
@@ -68,6 +69,19 @@ pub struct ResponderData {
 
     /// List of supported outer (LP) protocol version by this Responder
     pub supported_outer_protocol_versions: Vec<u8>,
+
+    /// Expected KEM hashes of the initiator.
+    /// It is only expected to be populated for the mutual mode of the KKT.
+    /// Otherwise the map is empty.
+    pub initiator_kem_hashes: BTreeMap<KEM, KEMKeyDigests>,
+}
+
+impl ResponderData {
+    #[must_use]
+    pub fn with_initiator_kem_hashes(mut self, kem_hashes: BTreeMap<KEM, KEMKeyDigests>) -> Self {
+        self.initiator_kem_hashes = kem_hashes;
+        self
+    }
 }
 
 impl Default for ResponderData {
@@ -77,6 +91,7 @@ impl Default for ResponderData {
             supported_hash_functions: HashFunction::iter().collect(),
             supported_signature_schemes: SignatureScheme::iter().collect(),
             supported_outer_protocol_versions: vec![version::CURRENT],
+            initiator_kem_hashes: Default::default(),
         }
     }
 }
@@ -96,6 +111,7 @@ where
         PSQHandshakeStateInitiator {
             initiator_data,
             inner_state: self,
+            mutual: false,
         }
     }
 
@@ -124,6 +140,7 @@ mod tests {
     };
     use nym_test_utils::mocks::async_read_write::MockIOStream;
     use nym_test_utils::traits::{Leak, TimeboxedSpawnable};
+    use std::collections::BTreeMap;
     use tokio::join;
 
     #[tokio::test]
@@ -197,6 +214,81 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn e2e_psq_mutual_handshake() -> anyhow::Result<()> {
+        for kem in KEM::iter() {
+            let conn_init = MockIOStream::default();
+            let conn_resp = conn_init.try_get_remote_handle();
+
+            // leak the connections (JUST FOR THE PURPOSE OF THIS TEST!)
+            // so they'd get 'static lifetime
+            let conn_init = conn_init.leak();
+            let conn_resp = conn_resp.leak();
+            let ciphersuite = Ciphersuite::default().with_kem(kem);
+
+            let (mut init, mut resp) = mock_peers();
+            init.ciphersuite = ciphersuite;
+            resp.ciphersuite = ciphersuite;
+            let resp_remote = resp.as_remote();
+            let init_remote = init.as_remote();
+
+            let handshake_init = PSQHandshakeState::new(conn_init, init)
+                .as_initiator(InitiatorData::new(1, resp_remote))
+                .set_mutual_kkt()?;
+            let handshake_resp = PSQHandshakeState::new(conn_resp, resp).as_responder(
+                ResponderData::default()
+                    .with_initiator_kem_hashes(init_remote.expected_kem_key_digests),
+            );
+
+            let init_rng = DeterministicRng09Send::new(u64_seeded_rng_09(1));
+            let resp_rng = DeterministicRng09Send::new(u64_seeded_rng_09(2));
+
+            // similarly leak the rngs to get the static lifetimes
+            let init_rng = init_rng.leak();
+            let resp_rng = resp_rng.leak();
+
+            let init_fut = handshake_init
+                .complete_handshake_with_rng(init_rng)
+                .spawn_timeboxed();
+            let resp_fut = handshake_resp
+                .complete_handshake_with_rng(resp_rng)
+                .spawn_timeboxed();
+
+            let (session_init, session_resp) = join!(init_fut, resp_fut);
+
+            let mut session_init = session_init???;
+            let mut session_resp = session_resp???;
+
+            assert_eq!(session_init.receiver_index(), session_resp.receiver_index());
+
+            assert_eq!(
+                session_init.session_identifier(),
+                session_resp.session_identifier()
+            );
+
+            // test serialization, deserialization
+            let channel_i = session_init.active_transport();
+            let channel_r = session_resp.active_transport();
+
+            assert_eq!(channel_i.identifier(), channel_r.identifier());
+
+            let app_data_i = b"Derived session hey".as_slice();
+            let app_data_r = b"Derived session ho".as_slice();
+
+            let ct_i = encrypt_data(app_data_i, channel_i)?;
+            let pt_r = decrypt_data(&ct_i, channel_r)?;
+
+            assert_eq!(app_data_i, pt_r);
+
+            let ct_r = encrypt_data(app_data_r, channel_r)?;
+            let pt_i = decrypt_data(&ct_r, channel_i)?;
+
+            assert_eq!(app_data_r, pt_i);
+        }
+
+        Ok(())
+    }
+
     // plain test without any wrappers
     #[test]
     fn e2e_test_plain() {
@@ -209,6 +301,7 @@ mod tests {
             init.ciphersuite = Ciphersuite::default().with_kem(kem);
             let resp_remote = resp.as_remote();
             let dir_hash = resp_remote.expected_kem_key_hash(init.ciphersuite).unwrap();
+            let dir_hash_init = BTreeMap::new();
 
             let resp_keys = resp.kem_keypairs.as_ref().unwrap();
             let responder_x25519_keypair = resp.x25519();
@@ -223,6 +316,7 @@ mod tests {
             let kkt_responder = KKTResponder::new(
                 responder_x25519_keypair,
                 resp_keys,
+                &dir_hash_init,
                 &supported_hash,
                 &supported_sigs,
                 &[protocol_version],
@@ -369,4 +463,188 @@ mod tests {
             assert_eq!(app_data_r, pt_i);
         }
     }
+
+    #[test]
+    fn e2e_test_plain_mutual() {
+        let mut rng = deterministic_rng_09();
+
+        for kem in KEM::iter() {
+            // SETUP START:
+            let protocol_version = 1;
+            let (mut init, resp) = mock_peers();
+            init.ciphersuite = Ciphersuite::default().with_kem(kem);
+
+            let init_remote = init.as_remote();
+            let resp_remote = resp.as_remote();
+
+            let dir_hash_init = init_remote.expected_kem_key_digests.clone();
+            let dir_hash_resp = resp_remote.expected_kem_key_hash(init.ciphersuite).unwrap();
+
+            let resp_keys = resp.kem_keypairs.as_ref().unwrap();
+            let responder_x25519_keypair = resp.x25519();
+
+            let init_keys = init.kem_keypairs.as_ref().unwrap();
+            let init_kem = init_keys.encoded_encapsulation_key(kem).unwrap();
+
+            let supported_sigs = [SignatureScheme::Ed25519];
+            let supported_hash = [
+                HashFunction::Blake3,
+                HashFunction::Shake256,
+                HashFunction::Shake128,
+                HashFunction::SHA256,
+            ];
+            let kkt_responder = KKTResponder::new(
+                responder_x25519_keypair,
+                resp_keys,
+                &dir_hash_init,
+                &supported_hash,
+                &supported_sigs,
+                &[protocol_version],
+            )
+            .unwrap();
+
+            // SETUP END
+
+            let lp_peer_config = LpPeerConfig::new_client_to_entry(&mut rng, false);
+
+            // OneWay - MlKem
+            let (mut initiator, request) = KKTInitiator::generate_mutual_request(
+                &mut rng,
+                init.ciphersuite,
+                init_kem,
+                &responder_x25519_keypair.pk,
+                &dir_hash_resp,
+                protocol_version,
+                Some(Vec::from(lp_peer_config.serialize())),
+            )
+            .unwrap();
+
+            let processed_req = kkt_responder
+                .process_request(request, LP_PEER_CONFIG_SIZE)
+                .unwrap();
+
+            let init_key = processed_req.remote_encapsulation_key.unwrap();
+            assert_eq!(init_key.as_bytes(), init_kem);
+
+            let response = initiator
+                .process_response(processed_req.response, 0)
+                .unwrap();
+            let encapsulation_key = response.encapsulation_key;
+
+            let mut payload_buf_responder = vec![0u8; 4096];
+            let mut payload_buf_initiator = vec![0u8; 4096];
+
+            let initiator_ciphersuite =
+                initiator::build_psq_ciphersuite(&init, &resp_remote, &encapsulation_key).unwrap();
+            let mut initiator = initiator::build_psq_principal(
+                rand09::rng(),
+                protocol_version,
+                initiator_ciphersuite,
+            )
+            .unwrap();
+
+            let responder_ciphersuite = responder::build_psq_ciphersuite(&resp, kem).unwrap();
+            let mut responder = responder::build_psq_principal(
+                rand09::rng(),
+                protocol_version,
+                responder_ciphersuite,
+            )
+            .unwrap();
+
+            // Send first message
+            let mut buf = vec![0u8; psq_msg1_size(kem)];
+            let len_i = initiator.write_message(&[], &mut buf).unwrap();
+            assert_eq!(len_i, buf.len());
+
+            // Read first message
+            let (_, _) = responder
+                .read_message(&buf, &mut payload_buf_responder)
+                .unwrap();
+
+            // Get the authenticator out here, so we can deserialize the session later.
+            let Some(initiator_authenticator) = responder.initiator_authenticator() else {
+                panic!("No initiator authenticator found")
+            };
+
+            // Respond
+            let mut buf = [0u8; PSQ_MSG2_SIZE];
+            let len_r = responder.write_message(&[], &mut buf).unwrap();
+            assert_eq!(len_r, buf.len());
+
+            // Finalize on registration initiator
+            let (len_i_deserialized, _) = initiator
+                .read_message(&buf, &mut payload_buf_initiator)
+                .unwrap();
+
+            // We read the same amount of data.
+            assert_eq!(len_r, len_i_deserialized);
+
+            // Ready for transport mode
+            assert!(initiator.is_handshake_finished());
+            assert!(responder.is_handshake_finished());
+
+            let i_transport = initiator.into_session().unwrap();
+            let r_transport = responder.into_session().unwrap();
+
+            // test serialization, deserialization
+            let mut session_storage = vec![0u8; 4096];
+            i_transport
+                .serialize(
+                    &mut session_storage,
+                    SessionBinding {
+                        initiator_authenticator: &Authenticator::Dh(init.x25519().pk),
+                        responder_ecdh_pk: &responder_x25519_keypair.pk,
+                        responder_pq_pk: Some(encapsulation_key.as_pq_encapsulation_key()),
+                    },
+                )
+                .unwrap();
+            let mut i_transport = Session::deserialize(
+                &session_storage,
+                SessionBinding {
+                    initiator_authenticator: &Authenticator::Dh(init.x25519().pk),
+                    responder_ecdh_pk: &responder_x25519_keypair.pk,
+                    responder_pq_pk: Some(encapsulation_key.as_pq_encapsulation_key()),
+                },
+            )
+            .unwrap();
+
+            r_transport
+                .serialize(
+                    &mut session_storage,
+                    SessionBinding {
+                        initiator_authenticator: &initiator_authenticator,
+                        responder_ecdh_pk: &responder_x25519_keypair.pk,
+                        responder_pq_pk: Some(encapsulation_key.as_pq_encapsulation_key()),
+                    },
+                )
+                .unwrap();
+            let mut r_transport = Session::deserialize(
+                &session_storage,
+                SessionBinding {
+                    initiator_authenticator: &initiator_authenticator,
+                    responder_ecdh_pk: &responder_x25519_keypair.pk,
+                    responder_pq_pk: Some(encapsulation_key.as_pq_encapsulation_key()),
+                },
+            )
+            .unwrap();
+
+            let mut channel_i = i_transport.transport_channel().unwrap();
+            let mut channel_r = r_transport.transport_channel().unwrap();
+
+            assert_eq!(channel_i.identifier(), channel_r.identifier());
+
+            let app_data_i = b"Derived session hey".as_slice();
+            let app_data_r = b"Derived session ho".as_slice();
+
+            let ct_i = encrypt_data(app_data_i, &mut channel_i).unwrap();
+            let pt_r = decrypt_data(&ct_i, &mut channel_r).unwrap();
+
+            assert_eq!(app_data_i, pt_r);
+
+            let ct_r = encrypt_data(app_data_r, &mut channel_r).unwrap();
+            let pt_i = decrypt_data(&ct_r, &mut channel_i).unwrap();
+
+            assert_eq!(app_data_r, pt_i);
+        }
+    }
 }

common/nym-lp/src/psq/responder.rs

@@ -77,12 +77,14 @@ impl<'a, S> PSQHandshakeStateResponder<'a, S>
 where
     S: LpHandshakeChannel + Unpin,
 {
-    /// Attempt to receive a KKT request from a one-way client
-    async fn receive_one_way_kkt_request(&mut self) -> Result<KKTRequest, LpError> {
-        let packet_len = KKTRequest::size_excluding_payload(
-            KKTMode::OneWay,
-            self.inner_state.local_peer.ciphersuite.kem(),
-        ) + LP_PEER_CONFIG_SIZE;
+    async fn receive_kkt_request(&mut self, mode: KKTMode) -> Result<KKTRequest, LpError> {
+        let packet_len =
+            KKTRequest::size_excluding_payload(mode, self.inner_state.local_peer.ciphersuite.kem())
+                + LP_PEER_CONFIG_SIZE;
+
+        // TODO: we have an issue here: if initiator sends us a KEM key of different type
+        // than our ciphersuite, we will fail to receive it.
+        // Surely this won't blow up in our faces later... right?
 
         let req = self
             .inner_state
@@ -93,6 +95,16 @@ where
         Ok(req.into())
     }
 
+    /// Attempt to receive a KKT request from a one-way client
+    async fn receive_one_way_kkt_request(&mut self) -> Result<KKTRequest, LpError> {
+        Self::receive_kkt_request(self, KKTMode::OneWay).await
+    }
+
+    /// Attempt to receive a KKT request from a mutual client
+    async fn receive_mutual_kkt_request(&mut self) -> Result<KKTRequest, LpError> {
+        Self::receive_kkt_request(self, KKTMode::Mutual).await
+    }
+
     /// Attempt to process the received KKT request
     fn process_kkt_request(&self, kkt_request: KKTRequest) -> Result<ProcessedKKTRequest, LpError> {
         let kem_keys = &self
@@ -105,6 +117,7 @@ where
         let processed_req = KKTResponder::new(
             &self.inner_state.local_peer.x25519,
             kem_keys,
+            &self.responder_data.initiator_kem_hashes,
             &self.responder_data.supported_hash_functions,
             &self.responder_data.supported_signature_schemes,
             &self.responder_data.supported_outer_protocol_versions,
@@ -147,11 +160,18 @@ where
         R: rand09::CryptoRng,
     {
         // 1. receive and process KKTRequest
-        let kkt_request = self.receive_one_way_kkt_request().await?;
+        let kkt_request = if self.responder_data.initiator_kem_hashes.is_empty() {
+            debug!("expecting one way KKT request");
+            self.receive_one_way_kkt_request().await?
+        } else {
+            debug!("expecting mutual KKT request");
+            self.receive_mutual_kkt_request().await?
+        };
         debug!("received KKT request");
 
         let processed_req = self.process_kkt_request(kkt_request)?;
         let kem = processed_req.requested_kem;
+        let init_kem = processed_req.remote_encapsulation_key;
 
         let lp_peer_config = LpPeerConfig::deserialize(&processed_req.request_payload)?;
 
@@ -205,6 +225,7 @@ where
             initiator_authenticator,
             responder_ecdh_pk: self.inner_state.local_peer.x25519().pk,
             responder_pq_pk: Some(kem_key),
+            initiator_pq_pk: init_kem,
         };
 
         let psq_session = psq_responder.into_session()?;
@@ -348,4 +369,124 @@ mod tests {
 
         Ok(())
     }
+
+    #[tokio::test]
+    async fn responder_test_plain_mutual() -> anyhow::Result<()> {
+        for kem in KEM::iter() {
+            let conn_init = MockIOStream::default();
+            let conn_resp = conn_init.try_get_remote_handle();
+
+            // SETUP START:
+            // leak the connections (JUST FOR THE PURPOSE OF THIS TEST!)
+            // so they'd get 'static lifetime
+            let conn_init = conn_init.leak();
+            let conn_resp = conn_resp.leak();
+
+            let (mut init, mut resp) = mock_peers();
+            let resp_remote = resp.as_remote();
+            let init_remote = init.as_remote();
+
+            let ciphersuite = Ciphersuite::default().with_kem(kem);
+            init.ciphersuite = ciphersuite;
+            resp.ciphersuite = ciphersuite;
+
+            let responder_data = ResponderData::default()
+                .with_initiator_kem_hashes(init_remote.expected_kem_key_digests);
+            let handshake_resp =
+                PSQHandshakeState::new(conn_resp, resp).as_responder(responder_data);
+
+            let mut resp_rng = DeterministicRng09Send::new(u64_seeded_rng_09(2));
+            let resp_fut = tokio::spawn(async move {
+                handshake_resp
+                    .complete_handshake_with_rng(&mut resp_rng)
+                    .timeboxed()
+                    .await
+            });
+
+            // initiator:
+
+            let mut rng = deterministic_rng_09();
+            let dir_hash = resp_remote.expected_kem_key_hash(init.ciphersuite)?;
+
+            let lp_peer_config = LpPeerConfig::new_client_to_entry(&mut rng, false);
+
+            // Mutual - MlKem
+            let (mut initiator, request) = KKTInitiator::generate_mutual_request(
+                &mut rng,
+                init.ciphersuite,
+                init.encoded_kem_key(kem).unwrap(),
+                &resp_remote.x25519_public,
+                &dir_hash,
+                1,
+                Some(Vec::from(lp_peer_config.serialize())),
+            )?;
+
+            // 1. send kkt request
+            conn_init
+                .send_handshake_message::<handshake_message::KKTRequest>(request.into(), kem)
+                .timeboxed()
+                .await??;
+
+            // 2. receive KKT response
+            let response_len = KKTResponse::size_excluding_payload(kem);
+            let resp: handshake_message::KKTResponse = conn_init
+                .receive_handshake_message(response_len)
+                .timeboxed()
+                .await??;
+            let kkt_response = resp.into();
+
+            let response = initiator.process_response(kkt_response, 0)?;
+            let encapsulation_key = response.encapsulation_key;
+
+            let initiator_ciphersuite =
+                initiator::build_psq_ciphersuite(&init, &resp_remote, &encapsulation_key)?;
+            let mut initiator =
+                initiator::build_psq_principal(rand09::rng(), 1, initiator_ciphersuite)?;
+
+            // 3. send PSQ msg1
+            // Send first message
+            let mut buf = vec![0u8; psq_msg1_size(kem)];
+            let n = initiator.write_message(&[], &mut buf).unwrap();
+            assert_eq!(n, buf.len());
+            let msg = PSQMsg1::new(buf);
+            conn_init
+                .send_handshake_message(msg, kem)
+                .timeboxed()
+                .await??;
+
+            // 4. receive PSQ msg2
+            let msg: PSQMsg2 = conn_init
+                .receive_handshake_message(PSQ_MSG2_SIZE)
+                .timeboxed()
+                .await??;
+            initiator.read_message(&msg, &mut []).unwrap();
+
+            assert!(initiator.is_handshake_finished());
+
+            let mut session_resp = resp_fut.await???;
+
+            let mut i_transport = initiator.into_session().unwrap();
+
+            // test serialization, deserialization
+            let mut channel_i = i_transport.transport_channel().unwrap();
+            let channel_r = session_resp.active_transport();
+
+            assert_eq!(channel_i.identifier(), channel_r.identifier());
+
+            let app_data_i = b"Derived session hey".as_slice();
+            let app_data_r = b"Derived session ho".as_slice();
+
+            let ct_i = encrypt_data(app_data_i, &mut channel_i)?;
+            let pt_r = decrypt_data(&ct_i, channel_r)?;
+
+            assert_eq!(app_data_i, pt_r);
+
+            let ct_r = encrypt_data(app_data_r, channel_r)?;
+            let pt_i = decrypt_data(&ct_r, &mut channel_i)?;
+
+            assert_eq!(app_data_r, pt_i);
+        }
+
+        Ok(())
+    }
 }

common/nym-lp/src/session.rs

@@ -62,6 +62,9 @@ pub struct PersistentSessionBinding {
 
     /// The responder's long term PQ-KEM public key (if any).
     pub responder_pq_pk: Option<EncapsulationKey>,
+
+    /// The initiator's long term PQ-KEM public key (if any).
+    pub initiator_pq_pk: Option<EncapsulationKey>,
 }
 
 impl Debug for PersistentSessionBinding {

common/nymsphinx/framing/src/processing.rs

@@ -110,7 +110,7 @@ pub enum PacketProcessingError {
     PacketReplay,
 }
 
-pub struct PartialyUnwrappedPacketWithKeyRotation {
+pub struct PartiallyUnwrappedPacketWithKeyRotation {
     pub packet: PartiallyUnwrappedPacket,
     pub used_key_rotation: u32,
 }
@@ -183,8 +183,8 @@ impl PartiallyUnwrappedPacket {
     pub fn with_key_rotation(
         self,
         used_key_rotation: u32,
-    ) -> PartialyUnwrappedPacketWithKeyRotation {
-        PartialyUnwrappedPacketWithKeyRotation {
+    ) -> PartiallyUnwrappedPacketWithKeyRotation {
+        PartiallyUnwrappedPacketWithKeyRotation {
             packet: self,
             used_key_rotation,
         }

common/topology/src/lib.rs

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use ::serde::{Deserialize, Serialize};
-use nym_api_requests::nym_nodes::SkimmedNode;
+use nym_api_requests::nym_nodes::SkimmedNodeV1;
 use nym_crypto::asymmetric::ed25519;
 use nym_mixnet_contract_common::EpochId;
 use nym_sphinx_addressing::nodes::NodeIdentity;
@@ -283,11 +283,11 @@ impl NymTopology {
         serde_json::from_reader(file).map_err(Into::into)
     }
 
-    pub fn add_skimmed_nodes(&mut self, nodes: &[SkimmedNode]) {
+    pub fn add_skimmed_nodes(&mut self, nodes: &[SkimmedNodeV1]) {
         self.add_additional_nodes(nodes.iter())
     }
 
-    pub fn with_skimmed_nodes(mut self, nodes: &[SkimmedNode]) -> Self {
+    pub fn with_skimmed_nodes(mut self, nodes: &[SkimmedNodeV1]) -> Self {
         self.add_skimmed_nodes(nodes);
         self
     }

common/topology/src/node.rs

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use nym_api_requests::models::DeclaredRolesV1;
-use nym_api_requests::nym_nodes::SkimmedNode;
+use nym_api_requests::nym_nodes::SkimmedNodeV1;
 use nym_crypto::asymmetric::{ed25519, x25519};
 use nym_mixnet_contract_common::NodeId;
 use nym_sphinx_addressing::nodes::NymNodeRoutingAddress;
@@ -146,10 +146,10 @@ impl<'a> From<&'a RoutingNode> for SphinxNode {
     }
 }
 
-impl<'a> TryFrom<&'a SkimmedNode> for RoutingNode {
+impl<'a> TryFrom<&'a SkimmedNodeV1> for RoutingNode {
     type Error = RoutingNodeError;
 
-    fn try_from(value: &'a SkimmedNode) -> Result<Self, Self::Error> {
+    fn try_from(value: &'a SkimmedNodeV1) -> Result<Self, Self::Error> {
         // IF YOU EVER ADD "performance" TO RoutingNode,
         // MAKE SURE TO UPDATE THE LAZY IMPLEMENTATION OF
         // `impl NodeDescriptionTopologyExt for NymNodeDescription`!!!

documentation/docs/pages/operators/changelog.mdx

@@ -212,13 +212,13 @@ chmod +x network-tunnel-manager.sh
 
 - [Deriving `Serialize` for `GatewayData`](https://github.com/nymtech/nym/pull/6314): Deriving `Serialize` for gateway data, that will be used by the diagnostic tool in the `vpn-client` repo
 
-- [DNS static table pre-resolve](https://github.com/nymtech/nym/pull/6297): This PR adds pre-resolve stage that returns addres if we have used static table previously. This ensures that we don't continually suffer the penalty of a lookup timeout, while also allowing for the possibility of going back to the default internal secure resolver if one or more nameservers becomes usable again at a future time.
+- [DNS static table pre-resolve](https://github.com/nymtech/nym/pull/6297): This PR adds pre-resolve stage that returns address if we have used static table previously. This ensures that we don't continually suffer the penalty of a lookup timeout, while also allowing for the possibility of going back to the default internal secure resolver if one or more nameservers becomes usable again at a future time.
 
 - [Add `Copy+Clone` to `nym_api_provider::Config`](https://github.com/nymtech/nym/pull/6296): Add `Copy+Clone` to `nym_client_core::client::topology_control::nym_api_provider::Config`
 
 - [LP Registration + Telescoping + Gateway Probe Localnet Mode](https://github.com/nymtech/nym/pull/6286):  Combines LP registration protocol implementation, adds telescoping/nested sessions support, adds localnet mode for `gateway-probe` testing, integrates KKT & PSQ cryptographic primitives 
 
-- [Minor DNS improvements](https://github.com/nymtech/nym/pull/6283): Increase timeouts back to 10 seconds for overall lookup and 5 seconds per query, gnore unreliable test, remove JIT resolution in http client as it is at best not useful, and at worst increasing timeout
+- [Minor DNS improvements](https://github.com/nymtech/nym/pull/6283): Increase timeouts back to 10 seconds for overall lookup and 5 seconds per query, ignore unreliable test, remove JIT resolution in http client as it is at best not useful, and at worst increasing timeout
 
 - [HTTP client without default features](https://github.com/nymtech/nym/pull/6281): Fix compile issue caused when using the http client using `default-features=false`
 
@@ -250,7 +250,7 @@ chmod +x network-tunnel-manager.sh
 
 - [Update nix to `v0.30.1`](https://github.com/nymtech/nym/pull/6316)
 
-- [Rremove repetitive words in comment](https://github.com/nymtech/nym/pull/6313)
+- [Remove repetitive words in comment](https://github.com/nymtech/nym/pull/6313)
 
 - [Clippy fixes and use fixed rust version from `REQUIRED_RUSTC_VERSION`](https://github.com/nymtech/nym/pull/6295)
 

integration-tests/src/lp_registration.rs

@@ -17,9 +17,9 @@ mod tests {
     use nym_lp::peer::LpLocalPeer;
     use nym_node::config::{LpConfig, LpDebug};
     use nym_node::node::GatewayStorage;
+    use nym_node::node::lp::control::handler::LpConnectionHandler;
     use nym_node::node::lp::error::LpHandlerError;
-    use nym_node::node::lp::handler::LpConnectionHandler;
-    use nym_node::node::lp::{LpHandlerState, MixForwardingReceiver, mix_forwarding_channels};
+    use nym_node::node::lp::{SharedLpControlState, SharedLpState};
     use nym_node::wireguard::{PeerManager, PeerRegistrator};
     use nym_registration_client::{LpClientError, LpRegistrationClient};
     use nym_test_utils::helpers::{CryptoRng09, seeded_rng};
@@ -130,11 +130,8 @@ mod tests {
 
     struct Gateway {
         base: Party,
-        lp_state: LpHandlerState,
+        lp_state: SharedLpControlState,
         ip_pool: IpPool,
-        // might be used later for mixnet registration tests
-        #[allow(unused)]
-        mix_receiver: MixForwardingReceiver,
         mock_peer_controller: SpawnedPeerController,
 
         tasks_cancellation: CancellationToken,
@@ -210,9 +207,6 @@ mod tests {
             let forward_semaphore =
                 Arc::new(Semaphore::new(lp_config.debug.max_concurrent_forwards));
 
-            // Create mix forwarding channel (unused in tests but required by struct)
-            let (mix_sender, mix_receiver) = mix_forwarding_channels();
-
             // create wireguard data
             let (wireguard_data, peer_request_rx) = Self::wireguard_data(&base);
 
@@ -231,31 +225,24 @@ mod tests {
                 upgrade_mode_details,
             );
 
-            let lp_state = LpHandlerState {
+            let lp_state = SharedLpControlState {
                 local_lp_peer: base.peer.clone(),
 
-                metrics: Default::default(),
-
-                // use default lp config (with enabled flag)
-                lp_config,
-
-                // TODO: might be needed later on for mixnet registration
-                outbound_mix_sender: mix_sender,
-
-                // we start with empty state
-                session_states: Arc::new(Default::default()),
-
                 forward_semaphore,
 
                 // handles for dealing with new peers
                 peer_registrator: Some(peer_registrator),
+                shared: SharedLpState {
+                    metrics: Default::default(),
+                    lp_config,
+                    session_states: Arc::new(Default::default()),
+                },
             };
 
             Ok(Gateway {
                 base,
                 lp_state,
                 ip_pool: Self::ip_pool(),
-                mix_receiver,
                 mock_peer_controller: SpawnedPeerController::Ready {
                     controller: mock_peer_controller,
                 },

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

@@ -6,7 +6,7 @@ use crate::models::{
     HostInformationV1, IpPacketRouterDetailsV1, NetworkRequesterDetailsV1,
     OffsetDateTimeJsonSchemaWrapper, WebSocketsV1, WireguardDetailsV1,
 };
-use crate::nym_nodes::{BasicEntryInformation, NodeRole, SemiSkimmedNode, SkimmedNode};
+use crate::nym_nodes::{BasicEntryInformation, NodeRole, SemiSkimmedNodeV1, SkimmedNodeV1};
 use nym_crypto::asymmetric::{ed25519, x25519};
 use nym_mixnet_contract_common::reward_params::Performance;
 use nym_mixnet_contract_common::NodeId;
@@ -76,7 +76,7 @@ impl NymNodeDescriptionV1 {
         current_rotation_id: u32,
         role: NodeRole,
         performance: Performance,
-    ) -> SkimmedNode {
+    ) -> SkimmedNodeV1 {
         let keys = &self.description.host_information.keys;
         let entry = if self.description.declared_role.entry {
             Some(self.entry_information())
@@ -84,7 +84,7 @@ impl NymNodeDescriptionV1 {
             None
         };
 
-        SkimmedNode {
+        SkimmedNodeV1 {
             node_id: self.node_id,
             ed25519_identity_pubkey: keys.ed25519,
             ip_addresses: self.description.host_information.ip_address.clone(),
@@ -105,10 +105,10 @@ impl NymNodeDescriptionV1 {
         current_rotation_id: u32,
         role: NodeRole,
         performance: Performance,
-    ) -> SemiSkimmedNode {
+    ) -> SemiSkimmedNodeV1 {
         let skimmed_node = self.to_skimmed_node(current_rotation_id, role, performance);
 
-        SemiSkimmedNode {
+        SemiSkimmedNodeV1 {
             basic: skimmed_node,
             x25519_noise_versioned_key: self
                 .description

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

@@ -7,7 +7,9 @@ use crate::models::{
     LewesProtocolDetailsV1, NetworkRequesterDetailsV1, NymNodeDataV1, NymNodeDescriptionV1,
     OffsetDateTimeJsonSchemaWrapper, SphinxKeyV1, WebSocketsV1, WireguardDetailsV1,
 };
-use crate::nym_nodes::{BasicEntryInformation, NodeRole, SemiSkimmedNode, SkimmedNode};
+use crate::nym_nodes::{
+    BasicEntryInformation, NodeRole, SemiSkimmedNodeV1, SemiSkimmedNodeV3, SkimmedNodeV1,
+};
 use nym_crypto::asymmetric::{ed25519, x25519};
 use nym_mixnet_contract_common::reward_params::Performance;
 use nym_mixnet_contract_common::NodeId;
@@ -92,7 +94,7 @@ impl NymNodeDescriptionV2 {
         current_rotation_id: u32,
         role: NodeRole,
         performance: Performance,
-    ) -> SkimmedNode {
+    ) -> SkimmedNodeV1 {
         let keys = &self.description.host_information.keys;
         let entry = if self.description.declared_role.entry {
             Some(self.entry_information())
@@ -100,7 +102,7 @@ impl NymNodeDescriptionV2 {
             None
         };
 
-        SkimmedNode {
+        SkimmedNodeV1 {
             node_id: self.node_id,
             ed25519_identity_pubkey: keys.ed25519,
             ip_addresses: self.description.host_information.ip_address.clone(),
@@ -121,10 +123,10 @@ impl NymNodeDescriptionV2 {
         current_rotation_id: u32,
         role: NodeRole,
         performance: Performance,
-    ) -> SemiSkimmedNode {
+    ) -> SemiSkimmedNodeV1 {
         let skimmed_node = self.to_skimmed_node(current_rotation_id, role, performance);
 
-        SemiSkimmedNode {
+        SemiSkimmedNodeV1 {
             basic: skimmed_node,
             x25519_noise_versioned_key: self
                 .description
@@ -133,6 +135,26 @@ impl NymNodeDescriptionV2 {
                 .x25519_versioned_noise,
         }
     }
+
+    pub fn to_semi_skimmed_node_v3(
+        &self,
+        current_rotation_id: u32,
+        role: NodeRole,
+        performance: Performance,
+    ) -> SemiSkimmedNodeV3 {
+        let skimmed_node = self.to_skimmed_node(current_rotation_id, role, performance);
+
+        SemiSkimmedNodeV3 {
+            basic: skimmed_node,
+            noise_key: self
+                .description
+                .host_information
+                .keys
+                .x25519_versioned_noise,
+            build_version: self.description.build_information.build_version.clone(),
+            lp: self.description.lewes_protocol.clone(),
+        }
+    }
 }
 
 // to whoever is thinking of modifying this struct.

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

@@ -1,7 +1,9 @@
 // Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use crate::models::{DeclaredRolesV1, NymNodeDataV1, OffsetDateTimeJsonSchemaWrapper};
+use crate::models::{
+    DeclaredRolesV1, LewesProtocolDetailsV1, NymNodeDataV1, OffsetDateTimeJsonSchemaWrapper,
+};
 use crate::pagination::{PaginatedResponse, Pagination};
 use nym_crypto::asymmetric::ed25519::serde_helpers::bs58_ed25519_pubkey;
 use nym_crypto::asymmetric::x25519::serde_helpers::bs58_x25519_pubkey;
@@ -18,24 +20,24 @@ use utoipa::ToSchema;
 
 #[derive(Clone, Debug, Serialize, Deserialize, schemars::JsonSchema, utoipa::ToSchema)]
 pub struct SkimmedNodesWithMetadata {
-    pub nodes: Vec<SkimmedNode>,
+    pub nodes: Vec<SkimmedNodeV1>,
     pub metadata: NodesResponseMetadata,
 }
 
 impl SkimmedNodesWithMetadata {
-    pub fn new(nodes: Vec<SkimmedNode>, metadata: NodesResponseMetadata) -> Self {
+    pub fn new(nodes: Vec<SkimmedNodeV1>, metadata: NodesResponseMetadata) -> Self {
         SkimmedNodesWithMetadata { nodes, metadata }
     }
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize, schemars::JsonSchema, utoipa::ToSchema)]
 pub struct SemiSkimmedNodesWithMetadata {
-    pub nodes: Vec<SemiSkimmedNode>,
+    pub nodes: Vec<SemiSkimmedNodeV1>,
     pub metadata: NodesResponseMetadata,
 }
 
 impl SemiSkimmedNodesWithMetadata {
-    pub fn new(nodes: Vec<SemiSkimmedNode>, metadata: NodesResponseMetadata) -> Self {
+    pub fn new(nodes: Vec<SemiSkimmedNodeV1>, metadata: NodesResponseMetadata) -> Self {
         SemiSkimmedNodesWithMetadata { nodes, metadata }
     }
 }
@@ -228,9 +230,7 @@ pub struct BasicEntryInformation {
 
 // the bare minimum information needed to construct sphinx packets
 #[derive(Clone, Debug, Serialize, Deserialize, schemars::JsonSchema, ToSchema)]
-pub struct SkimmedNode {
-    // in directory v3 all nodes (mixnodes AND gateways) will have a unique id
-    // but to keep structure consistent, introduce this field now
+pub struct SkimmedNodeV1 {
     #[schema(value_type = u32)]
     pub node_id: NodeId,
 
@@ -263,7 +263,7 @@ pub struct SkimmedNode {
     pub performance: Performance,
 }
 
-impl SkimmedNode {
+impl SkimmedNodeV1 {
     pub fn get_mix_layer(&self) -> Option<u8> {
         match self.role {
             NodeRole::Mixnode { layer } => Some(layer),
@@ -275,8 +275,8 @@ impl SkimmedNode {
 // an intermediate variant that exposes additional data such as noise keys but without
 // the full fat of the self-described data
 #[derive(Clone, Debug, Serialize, Deserialize, schemars::JsonSchema, ToSchema)]
-pub struct SemiSkimmedNode {
-    pub basic: SkimmedNode,
+pub struct SemiSkimmedNodeV1 {
+    pub basic: SkimmedNodeV1,
 
     pub x25519_noise_versioned_key: Option<VersionedNoiseKeyV1>,
     // pub location:
@@ -284,7 +284,7 @@ pub struct SemiSkimmedNode {
 
 #[derive(Clone, Debug, Serialize, Deserialize, schemars::JsonSchema, ToSchema)]
 pub struct FullFatNode {
-    pub expanded: SemiSkimmedNode,
+    pub expanded: SemiSkimmedNodeV1,
 
     // kinda temporary for now to make as few changes as possible for now
     pub self_described: Option<NymNodeDataV1>,
@@ -301,3 +301,19 @@ pub struct NodesByAddressesResponse {
     #[schema(value_type = HashMap<String, Option<u32>>)]
     pub existence: HashMap<IpAddr, Option<NodeId>>,
 }
+
+/// All the information required for sending packets between nodes (sphinx, noise, LP)
+#[derive(Clone, Debug, Serialize, Deserialize, schemars::JsonSchema, ToSchema)]
+pub struct SemiSkimmedNodeV3 {
+    /// Basic node information required for mixnet routing
+    pub basic: SkimmedNodeV1,
+
+    /// Noise key of the node
+    pub noise_key: Option<VersionedNoiseKeyV1>,
+
+    /// Build version of this node used as a hint in inferring the Ciphersuite compatibility
+    pub build_version: String,
+
+    /// Information required for establishing an LP connection
+    pub lp: Option<LewesProtocolDetailsV1>,
+}

nym-api/src/support/http/router.rs

@@ -10,6 +10,7 @@ use crate::support::http::openapi::ApiDoc;
 use crate::support::http::state::AppState;
 use crate::unstable_routes::v1::unstable_routes_v1;
 use crate::unstable_routes::v2::unstable_routes_v2;
+use crate::unstable_routes::v3::unstable_routes_v3;
 use crate::{nym_nodes, status};
 use anyhow::anyhow;
 use axum::response::Redirect;
@@ -70,7 +71,8 @@ impl RouterBuilder {
                 Router::new()
                     .nest("/unstable", unstable_routes_v2())
                     .nest("/nym-nodes", nym_nodes::handlers::v2::routes()),
-            );
+            )
+            .nest("/v3", Router::new().nest("/unstable", unstable_routes_v3()));
 
         Self {
             unfinished_router: default_routes,

nym-api/src/unstable_routes/mod.rs

@@ -4,3 +4,4 @@
 pub(crate) mod helpers;
 pub(crate) mod v1;
 pub(crate) mod v2;
+pub(crate) mod v3;

nym-api/src/unstable_routes/v1/nym_nodes/semi_skimmed/mod.rs

@@ -5,7 +5,7 @@ use crate::node_status_api::models::{AxumErrorResponse, AxumResult};
 use crate::support::http::state::AppState;
 use crate::unstable_routes::v1::nym_nodes::helpers::NodesParamsWithRole;
 use axum::extract::{Query, State};
-use nym_api_requests::nym_nodes::{CachedNodesResponse, SemiSkimmedNode};
+use nym_api_requests::nym_nodes::{CachedNodesResponse, SemiSkimmedNodeV1};
 use nym_http_api_common::FormattedResponse;
 
 #[utoipa::path(
@@ -22,6 +22,6 @@ use nym_http_api_common::FormattedResponse;
 pub(crate) async fn nodes_expanded(
     _state: State<AppState>,
     _query_params: Query<NodesParamsWithRole>,
-) -> AxumResult<FormattedResponse<CachedNodesResponse<SemiSkimmedNode>>> {
+) -> AxumResult<FormattedResponse<CachedNodesResponse<SemiSkimmedNodeV1>>> {
     Err(AxumErrorResponse::not_implemented())
 }

nym-api/src/unstable_routes/v1/nym_nodes/skimmed/handlers.rs

@@ -11,7 +11,7 @@ use crate::unstable_routes::v1::nym_nodes::skimmed::{
     PaginatedCachedNodesResponseSchema, PaginatedSkimmedNodes,
 };
 use axum::extract::{Query, State};
-use nym_api_requests::nym_nodes::{CachedNodesResponse, NodeRoleQueryParam, SkimmedNode};
+use nym_api_requests::nym_nodes::{CachedNodesResponse, NodeRoleQueryParam, SkimmedNodeV1};
 use nym_http_api_common::FormattedResponse;
 
 /// Deprecated query that gets ALL gateways
@@ -23,9 +23,9 @@ use nym_http_api_common::FormattedResponse;
     context_path = "/v1/unstable/nym-nodes",
     responses(
         (status = 200, content(
-            (CachedNodesResponse<SkimmedNode> = "application/json"),
-            (CachedNodesResponse<SkimmedNode> = "application/yaml"),
-            (CachedNodesResponse<SkimmedNode> = "application/bincode")
+            (CachedNodesResponse<SkimmedNodeV1> = "application/json"),
+            (CachedNodesResponse<SkimmedNodeV1> = "application/yaml"),
+            (CachedNodesResponse<SkimmedNodeV1> = "application/bincode")
         ))
     ),
 )]
@@ -34,7 +34,7 @@ use nym_http_api_common::FormattedResponse;
 pub(crate) async fn deprecated_gateways_basic(
     state: State<AppState>,
     query_params: Query<NodesParams>,
-) -> AxumResult<FormattedResponse<CachedNodesResponse<SkimmedNode>>> {
+) -> AxumResult<FormattedResponse<CachedNodesResponse<SkimmedNodeV1>>> {
     let output = query_params.output.unwrap_or_default();
 
     // 1. call '/v1/unstable/skimmed/entry-gateways/all'
@@ -59,9 +59,9 @@ pub(crate) async fn deprecated_gateways_basic(
     context_path = "/v1/unstable/nym-nodes",
     responses(
         (status = 200, content(
-            (CachedNodesResponse<SkimmedNode> = "application/json"),
-            (CachedNodesResponse<SkimmedNode> = "application/yaml"),
-            (CachedNodesResponse<SkimmedNode> = "application/bincode")
+            (CachedNodesResponse<SkimmedNodeV1> = "application/json"),
+            (CachedNodesResponse<SkimmedNodeV1> = "application/yaml"),
+            (CachedNodesResponse<SkimmedNodeV1> = "application/bincode")
         ))
     ),
 )]
@@ -70,7 +70,7 @@ pub(crate) async fn deprecated_gateways_basic(
 pub(crate) async fn deprecated_mixnodes_basic(
     state: State<AppState>,
     query_params: Query<NodesParams>,
-) -> AxumResult<FormattedResponse<CachedNodesResponse<SkimmedNode>>> {
+) -> AxumResult<FormattedResponse<CachedNodesResponse<SkimmedNodeV1>>> {
     let output = query_params.output.unwrap_or_default();
 
     // 1. call '/v1/unstable/nym-nodes/skimmed/mixnodes/active'

nym-api/src/unstable_routes/v1/nym_nodes/skimmed/mod.rs

@@ -3,7 +3,7 @@
 
 use crate::node_status_api::models::AxumResult;
 use nym_api_requests::models::OffsetDateTimeJsonSchemaWrapper;
-use nym_api_requests::nym_nodes::{PaginatedCachedNodesResponseV1, SkimmedNode};
+use nym_api_requests::nym_nodes::{PaginatedCachedNodesResponseV1, SkimmedNodeV1};
 use nym_api_requests::pagination::PaginatedResponse;
 use nym_http_api_common::FormattedResponse;
 use utoipa::ToSchema;
@@ -12,7 +12,7 @@ pub(crate) mod handlers;
 pub(crate) mod helpers;
 
 pub type PaginatedSkimmedNodes =
-    AxumResult<FormattedResponse<PaginatedCachedNodesResponseV1<SkimmedNode>>>;
+    AxumResult<FormattedResponse<PaginatedCachedNodesResponseV1<SkimmedNodeV1>>>;
 
 pub(crate) use handlers::*;
 
@@ -21,6 +21,6 @@ pub(crate) use handlers::*;
 #[schema(title = "PaginatedCachedNodesResponse")]
 pub struct PaginatedCachedNodesResponseSchema {
     pub refreshed_at: OffsetDateTimeJsonSchemaWrapper,
-    #[schema(value_type = SkimmedNode)]
-    pub nodes: PaginatedResponse<SkimmedNode>,
+    #[schema(value_type = SkimmedNodeV1)]
+    pub nodes: PaginatedResponse<SkimmedNodeV1>,
 }

nym-api/src/unstable_routes/v2/nym_nodes/semi_skimmed/mod.rs

@@ -9,7 +9,7 @@ use axum::extract::{Query, State};
 use nym_api_requests::models::{
     NodeAnnotation, NymNodeDescriptionV2, OffsetDateTimeJsonSchemaWrapper,
 };
-use nym_api_requests::nym_nodes::{NodeRole, PaginatedCachedNodesResponseV2, SemiSkimmedNode};
+use nym_api_requests::nym_nodes::{NodeRole, PaginatedCachedNodesResponseV2, SemiSkimmedNodeV1};
 use nym_api_requests::pagination::PaginatedResponse;
 use nym_http_api_common::FormattedResponse;
 use nym_mixnet_contract_common::NodeId;
@@ -18,7 +18,7 @@ use std::collections::HashMap;
 use utoipa::ToSchema;
 
 pub type PaginatedSemiSkimmedNodes =
-    AxumResult<FormattedResponse<PaginatedCachedNodesResponseV2<SemiSkimmedNode>>>;
+    AxumResult<FormattedResponse<PaginatedCachedNodesResponseV2<SemiSkimmedNodeV1>>>;
 
 //SW TODO : this is copied from skimmed nodes, surely we can do better than that
 fn build_nym_nodes_response<'a, NI>(
@@ -27,7 +27,7 @@ fn build_nym_nodes_response<'a, NI>(
     annotations: &HashMap<NodeId, NodeAnnotation>,
     current_key_rotation: u32,
     active_only: bool,
-) -> Vec<SemiSkimmedNode>
+) -> Vec<SemiSkimmedNodeV1>
 where
     NI: Iterator<Item = &'a NymNodeDescriptionV2> + 'a,
 {
@@ -57,11 +57,11 @@ where
 
 #[allow(dead_code)] // not dead, used in OpenAPI docs
 #[derive(ToSchema)]
-#[schema(title = "PaginatedCachedNodesExpandedResponseSchema")]
-pub struct PaginatedCachedNodesExpandedResponseSchema {
+#[schema(title = "PaginatedCachedNodesExpandedV2ResponseSchema")]
+pub struct PaginatedCachedNodesExpandedV2ResponseSchema {
     pub refreshed_at: OffsetDateTimeJsonSchemaWrapper,
-    #[schema(value_type = SemiSkimmedNode)]
-    pub nodes: PaginatedResponse<SemiSkimmedNode>,
+    #[schema(value_type = SemiSkimmedNodeV1)]
+    pub nodes: PaginatedResponse<SemiSkimmedNodeV1>,
 }
 
 /// Return all Nym Nodes and optionally legacy mixnodes/gateways (if `no-legacy` flag is not used)
@@ -75,9 +75,9 @@ pub struct PaginatedCachedNodesExpandedResponseSchema {
     context_path = "/v2/unstable/nym-nodes/semi-skimmed",
     responses(
         (status = 200, content(
-            (PaginatedCachedNodesExpandedResponseSchema = "application/json"),
-            (PaginatedCachedNodesExpandedResponseSchema = "application/yaml"),
-            (PaginatedCachedNodesExpandedResponseSchema = "application/bincode")
+            (PaginatedCachedNodesExpandedV2ResponseSchema = "application/json"),
+            (PaginatedCachedNodesExpandedV2ResponseSchema = "application/yaml"),
+            (PaginatedCachedNodesExpandedV2ResponseSchema = "application/bincode")
         ))
     )
 )]

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

@@ -9,7 +9,7 @@ use axum::extract::{Query, State};
 use nym_api_requests::models::{
     NodeAnnotation, NymNodeDescriptionV2, OffsetDateTimeJsonSchemaWrapper,
 };
-use nym_api_requests::nym_nodes::{NodeRole, PaginatedCachedNodesResponseV2, SkimmedNode};
+use nym_api_requests::nym_nodes::{NodeRole, PaginatedCachedNodesResponseV2, SkimmedNodeV1};
 use nym_http_api_common::Output;
 use nym_mixnet_contract_common::{Interval, NodeId};
 use nym_topology::CachedEpochRewardedSet;
@@ -23,7 +23,7 @@ fn build_nym_nodes_response<'a, NI>(
     annotations: &HashMap<NodeId, NodeAnnotation>,
     current_key_rotation: u32,
     active_only: bool,
-) -> Vec<SkimmedNode>
+) -> Vec<SkimmedNodeV1>
 where
     NI: Iterator<Item = &'a NymNodeDescriptionV2> + 'a,
 {
@@ -56,7 +56,7 @@ fn maybe_add_expires_header(
     interval: Interval,
     current_key_rotation: u32,
     refreshed_at: OffsetDateTimeJsonSchemaWrapper,
-    nodes: Vec<SkimmedNode>,
+    nodes: Vec<SkimmedNodeV1>,
     active_only: bool,
 ) -> PaginatedSkimmedNodes {
     let base_response = output.to_response(

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

@@ -3,7 +3,7 @@
 
 use crate::node_status_api::models::AxumResult;
 use nym_api_requests::models::OffsetDateTimeJsonSchemaWrapper;
-use nym_api_requests::nym_nodes::{PaginatedCachedNodesResponseV2, SkimmedNode};
+use nym_api_requests::nym_nodes::{PaginatedCachedNodesResponseV2, SkimmedNodeV1};
 use nym_api_requests::pagination::PaginatedResponse;
 use nym_http_api_common::FormattedResponse;
 use utoipa::ToSchema;
@@ -12,7 +12,7 @@ pub(crate) mod handlers;
 pub(crate) mod helpers;
 
 pub type PaginatedSkimmedNodes =
-    AxumResult<FormattedResponse<PaginatedCachedNodesResponseV2<SkimmedNode>>>;
+    AxumResult<FormattedResponse<PaginatedCachedNodesResponseV2<SkimmedNodeV1>>>;
 
 pub(crate) use handlers::*;
 
@@ -21,6 +21,6 @@ pub(crate) use handlers::*;
 #[schema(title = "PaginatedCachedNodesResponse")]
 pub struct PaginatedCachedNodesResponseSchema {
     pub refreshed_at: OffsetDateTimeJsonSchemaWrapper,
-    #[schema(value_type = SkimmedNode)]
-    pub nodes: PaginatedResponse<SkimmedNode>,
+    #[schema(value_type = SkimmedNodeV1)]
+    pub nodes: PaginatedResponse<SkimmedNodeV1>,
 }

nym-api/src/unstable_routes/v3/mod.rs

@@ -0,0 +1,11 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use crate::support::http::state::AppState;
+use axum::Router;
+
+pub(crate) mod nym_nodes;
+
+pub(crate) fn unstable_routes_v3() -> Router<AppState> {
+    Router::new().nest("/nym-nodes", nym_nodes::routes())
+}

nym-api/src/unstable_routes/v3/nym_nodes/mod.rs

@@ -0,0 +1,19 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use crate::support::http::state::AppState;
+use crate::unstable_routes::v3::nym_nodes::semi_skimmed::nodes_expanded;
+use axum::routing::get;
+use axum::Router;
+use tower_http::compression::CompressionLayer;
+
+pub(crate) mod semi_skimmed;
+
+pub(crate) fn routes() -> Router<AppState> {
+    Router::new()
+        .nest(
+            "/semi-skimmed",
+            Router::new().route("/", get(nodes_expanded)),
+        )
+        .layer(CompressionLayer::new())
+}

nym-api/src/unstable_routes/v3/nym_nodes/semi_skimmed/mod.rs

@@ -0,0 +1,108 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use crate::node_status_api::models::AxumResult;
+use crate::support::http::state::AppState;
+use crate::unstable_routes::helpers::refreshed_at;
+use axum::extract::{Query, State};
+use nym_api_requests::models::{
+    NodeAnnotation, NymNodeDescriptionV2, OffsetDateTimeJsonSchemaWrapper,
+};
+use nym_api_requests::nym_nodes::{NodeRole, PaginatedCachedNodesResponseV2, SemiSkimmedNodeV3};
+use nym_api_requests::pagination::PaginatedResponse;
+use nym_http_api_common::{FormattedResponse, OutputParams};
+use nym_mixnet_contract_common::NodeId;
+use nym_topology::CachedEpochRewardedSet;
+use std::collections::HashMap;
+use utoipa::ToSchema;
+
+pub type PaginatedSemiSkimmedNodes =
+    AxumResult<FormattedResponse<PaginatedCachedNodesResponseV2<SemiSkimmedNodeV3>>>;
+
+fn build_response<'a>(
+    rewarded_set: &CachedEpochRewardedSet,
+    nym_nodes: impl Iterator<Item = &'a NymNodeDescriptionV2>,
+    annotations: &HashMap<NodeId, NodeAnnotation>,
+    current_key_rotation: u32,
+) -> Vec<SemiSkimmedNodeV3> {
+    let mut nodes = Vec::new();
+    for nym_node in nym_nodes {
+        let node_id = nym_node.node_id;
+
+        let role: NodeRole = rewarded_set.role(node_id).into();
+
+        // honestly, not sure under what exact circumstances this value could be missing,
+        // but in that case just use 0 performance
+        let annotation = annotations.get(&node_id).copied().unwrap_or_default();
+
+        nodes.push(nym_node.to_semi_skimmed_node_v3(
+            current_key_rotation,
+            role,
+            annotation.last_24h_performance,
+        ));
+    }
+    nodes
+}
+
+#[allow(dead_code)] // not dead, used in OpenAPI docs
+#[derive(ToSchema)]
+#[schema(title = "PaginatedCachedNodesExpandedV3ResponseSchema")]
+pub struct PaginatedCachedNodesExpandedV3ResponseSchema {
+    pub refreshed_at: OffsetDateTimeJsonSchemaWrapper,
+    #[schema(value_type = SemiSkimmedNodeV3)]
+    pub nodes: PaginatedResponse<SemiSkimmedNodeV3>,
+}
+
+/// Return all Nym Nodes that are currently bonded.
+#[utoipa::path(
+    operation_id = "v3_nodes_expanded",
+    tag = "Unstable Nym Nodes v3",
+    get,
+    params(OutputParams),
+    path = "/semi-skimmed",
+    context_path = "/v3/unstable/nym-nodes",
+    responses(
+        (status = 200, content(
+            (PaginatedCachedNodesExpandedV3ResponseSchema = "application/json"),
+            (PaginatedCachedNodesExpandedV3ResponseSchema = "application/yaml"),
+            (PaginatedCachedNodesExpandedV3ResponseSchema = "application/bincode")
+        ))
+    )
+)]
+pub(super) async fn nodes_expanded(
+    state: State<AppState>,
+    Query(output): Query<OutputParams>,
+) -> PaginatedSemiSkimmedNodes {
+    // 1. grab all relevant described nym-nodes
+    let rewarded_set = state.rewarded_set().await?;
+
+    let describe_cache = state.describe_nodes_cache_data().await?;
+    let all_nym_nodes = describe_cache.all_nym_nodes();
+    let status_cache = &state.node_status_cache();
+    let annotations = status_cache.node_annotations().await?;
+
+    let contract_cache = state.nym_contract_cache();
+    let current_key_rotation = contract_cache.current_key_rotation_id().await?;
+    let interval = contract_cache.current_interval().await?;
+
+    let nodes = build_response(
+        &rewarded_set,
+        all_nym_nodes,
+        &annotations,
+        current_key_rotation,
+    );
+
+    // min of all caches
+    let refreshed_at = refreshed_at([
+        rewarded_set.timestamp(),
+        status_cache.cache_timestamp().await,
+        describe_cache.timestamp(),
+    ]);
+
+    Ok(output.to_response(PaginatedCachedNodesResponseV2::new_full(
+        interval.current_epoch_absolute_id(),
+        current_key_rotation,
+        refreshed_at,
+        nodes,
+    )))
+}

nym-node-status-api/nym-node-status-api/src/db/models.rs

@@ -10,7 +10,7 @@ use nym_crypto::asymmetric::{ed25519, x25519};
 use nym_network_defaults::DEFAULT_NYM_NODE_HTTP_PORT;
 use nym_node_requests::api::v1::node::models::NodeDescription;
 use nym_validator_client::{
-    client::NymNodeDetails, models::NymNodeDescriptionV1, nym_api::SkimmedNode,
+    client::NymNodeDetails, models::NymNodeDescriptionV1, nym_api::SkimmedNodeV1,
 };
 use serde::{Deserialize, Serialize};
 use sqlx::FromRow;
@@ -462,7 +462,7 @@ pub(crate) struct NymNodeInsertRecord {
 
 impl NymNodeInsertRecord {
     pub fn new(
-        skimmed_node: SkimmedNode,
+        skimmed_node: SkimmedNodeV1,
         bond_info: Option<&NymNodeDetails>,
         self_described: Option<&NymNodeDescriptionV1>,
     ) -> anyhow::Result<Self> {
@@ -503,7 +503,7 @@ impl NymNodeInsertRecord {
     }
 }
 
-impl TryFrom<NymNodeDto> for SkimmedNode {
+impl TryFrom<NymNodeDto> for SkimmedNodeV1 {
     type Error = anyhow::Error;
 
     fn try_from(other: NymNodeDto) -> Result<Self, Self::Error> {
@@ -517,7 +517,7 @@ impl TryFrom<NymNodeDto> for SkimmedNode {
             None => None,
         };
 
-        let skimmed_node = SkimmedNode {
+        let skimmed_node = SkimmedNodeV1 {
             node_id,
             ed25519_identity_pubkey: ed25519::PublicKey::from_base58_string(
                 other.ed25519_identity_pubkey,

nym-node-status-api/nym-node-status-api/src/db/queries/scraper.rs

@@ -10,7 +10,7 @@ use crate::{
     utils::now_utc,
 };
 use anyhow::Result;
-use nym_validator_client::nym_api::SkimmedNode;
+use nym_validator_client::nym_api::SkimmedNodeV1;
 
 pub(crate) async fn get_nodes_for_scraping(pool: &DbPool) -> Result<Vec<ScraperNodeInfo>> {
     let mut nodes_to_scrape = Vec::new();
@@ -24,7 +24,7 @@ pub(crate) async fn get_nodes_for_scraping(pool: &DbPool) -> Result<Vec<ScraperN
             nodes_dto.into_iter().filter_map(|node_dto| {
                 let node_id = node_dto.node_id;
                 let http_api_port = node_dto.http_api_port;
-                match SkimmedNode::try_from(node_dto) {
+                match SkimmedNodeV1::try_from(node_dto) {
                     Ok(node) => Some((node, http_api_port)),
                     Err(e) => {
                         tracing::error!("Failed to decode node_id={}: {}", node_id, e);

nym-node-status-api/nym-node-status-api/src/db/tests.rs

@@ -141,7 +141,7 @@ mod db_tests {
             http_api_port: None,
         };
 
-        let skimmed_node: nym_validator_client::nym_api::SkimmedNode =
+        let skimmed_node: nym_validator_client::nym_api::SkimmedNodeV1 =
             nym_node_dto.try_into().unwrap();
 
         assert_eq!(skimmed_node.node_id, 1);
@@ -174,7 +174,7 @@ fn test_nym_node_insert_record_new() {
     let ed25519_pk = nym_crypto::asymmetric::ed25519::PublicKey::from_bytes(&[1; 32]).unwrap();
     let x25519_pk = nym_crypto::asymmetric::x25519::PublicKey::from_bytes(&[2; 32]).unwrap();
 
-    let skimmed_node = nym_validator_client::nym_api::SkimmedNode {
+    let skimmed_node = nym_validator_client::nym_api::SkimmedNodeV1 {
         node_id: 1,
         ed25519_identity_pubkey: ed25519_pk,
         ip_addresses: vec!["1.1.1.1".parse().unwrap()],
@@ -226,7 +226,7 @@ fn test_nym_node_insert_record_with_entry() {
     let ed25519_pk = nym_crypto::asymmetric::ed25519::PublicKey::from_bytes(&[1; 32]).unwrap();
     let x25519_pk = nym_crypto::asymmetric::x25519::PublicKey::from_bytes(&[2; 32]).unwrap();
 
-    let skimmed_node = nym_validator_client::nym_api::SkimmedNode {
+    let skimmed_node = nym_validator_client::nym_api::SkimmedNodeV1 {
         node_id: 1,
         ed25519_identity_pubkey: ed25519_pk,
         ip_addresses: vec!["1.1.1.1".parse().unwrap()],
@@ -438,7 +438,7 @@ fn test_nym_node_dto_with_invalid_keys() {
         http_api_port: None,
     };
 
-    let result: Result<nym_validator_client::nym_api::SkimmedNode, _> = nym_node_dto.try_into();
+    let result: Result<nym_validator_client::nym_api::SkimmedNodeV1, _> = nym_node_dto.try_into();
     assert!(result.is_err());
     assert!(
         result
@@ -476,7 +476,7 @@ fn test_nym_node_dto_with_invalid_performance() {
         http_api_port: None,
     };
 
-    let result: Result<nym_validator_client::nym_api::SkimmedNode, _> = nym_node_dto.try_into();
+    let result: Result<nym_validator_client::nym_api::SkimmedNodeV1, _> = nym_node_dto.try_into();
     assert!(result.is_err());
     assert!(
         result

nym-node-status-api/nym-node-status-api/src/http/models/mod.rs

@@ -12,7 +12,7 @@ use nym_node_requests::api::v1::node::models::NodeDescription;
 use nym_validator_client::{
     client::NodeId,
     models::{AuthenticatorDetailsV1, BinaryBuildInformationOwned, IpPacketRouterDetailsV1},
-    nym_api::SkimmedNode,
+    nym_api::SkimmedNodeV1,
     nym_nodes::{BasicEntryInformation, NodeRole},
 };
 use serde::{Deserialize, Serialize};
@@ -139,7 +139,7 @@ impl DVpnGateway {
     #[instrument(level = tracing::Level::INFO, name = "dvpn_gw_new", skip_all, fields(gateway_key = gateway.gateway_identity_key, node_id = skimmed_node.node_id))]
     pub(crate) fn new(
         gateway: Gateway,
-        skimmed_node: &SkimmedNode,
+        skimmed_node: &SkimmedNodeV1,
         socks5_score: Option<&ScoreValue>,
     ) -> anyhow::Result<Self> {
         let location = gateway

nym-node-status-api/nym-node-status-api/src/http/state.rs

@@ -6,7 +6,7 @@ use nym_contracts_common::NaiveFloat;
 use nym_crypto::asymmetric::ed25519::PublicKey;
 use nym_mixnet_contract_common::NodeId;
 use nym_node_status_client::auth::VerifiableRequest;
-use nym_validator_client::nym_api::SkimmedNode;
+use nym_validator_client::nym_api::SkimmedNodeV1;
 use semver::Version;
 use serde::{Deserialize, Serialize};
 use std::{collections::HashMap, net::IpAddr, sync::Arc, time::Duration};
@@ -304,7 +304,7 @@ impl HttpCache {
                     Ok(records) => {
                         let mut nodes = HashMap::new();
                         for dto in records {
-                            match SkimmedNode::try_from(dto) {
+                            match SkimmedNodeV1::try_from(dto) {
                                 Ok(skimmed_node) => {
                                     let key =
                                         skimmed_node.ed25519_identity_pubkey.to_base58_string();
@@ -671,7 +671,7 @@ async fn aggregate_node_info_from_db(
     let skimmed_nodes = queries::get_all_nym_nodes(pool).await.map(|records| {
         records
             .into_iter()
-            .filter_map(|dto| SkimmedNode::try_from(dto).ok())
+            .filter_map(|dto| SkimmedNodeV1::try_from(dto).ok())
             .map(|skimmed_node| (skimmed_node.node_id, skimmed_node))
             .collect::<HashMap<_, _>>()
     })?;

nym-node-status-api/nym-node-status-api/src/monitor/mod.rs

@@ -12,7 +12,7 @@ use moka::future::Cache;
 use nym_network_defaults::NymNetworkDetails;
 use nym_validator_client::{
     QueryHttpRpcNyxdClient,
-    nym_nodes::{NodeRole, SkimmedNode},
+    nym_nodes::{NodeRole, SkimmedNodeV1},
 };
 use nym_validator_client::{
     client::{NodeId, NymApiClientExt, NymNodeDetails},
@@ -308,7 +308,7 @@ impl Monitor {
 
     fn prepare_nym_node_data(
         &self,
-        skimmed_nodes: Vec<SkimmedNode>,
+        skimmed_nodes: Vec<SkimmedNodeV1>,
         bonded_node_info: &HashMap<NodeId, NymNodeDetails>,
         described_nodes: &HashMap<NodeId, NymNodeDescriptionV1>,
     ) -> Vec<NymNodeInsertRecord> {
@@ -336,7 +336,7 @@ impl Monitor {
     async fn prepare_gateway_data(
         &mut self,
         described_gateways: &[&NymNodeDescriptionV1],
-        skimmed_gateways: &[SkimmedNode],
+        skimmed_gateways: &[SkimmedNodeV1],
         bonded_nodes: &HashMap<NodeId, NymNodeDetails>,
     ) -> anyhow::Result<Vec<GatewayInsertRecord>> {
         let mut gateway_records = Vec::new();

nym-node/Cargo.toml

@@ -30,6 +30,7 @@ console-subscriber = { workspace = true, optional = true }
 csv = { workspace = true }
 clap = { workspace = true, features = ["cargo", "env"] }
 dashmap = { workspace = true }
+flume= { workspace = true }
 futures = { workspace = true }
 hex = { workspace = true }
 humantime-serde = { workspace = true }

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

@@ -104,6 +104,12 @@ impl MixingStats {
             .dropped += 1;
     }
 
+    pub fn ingress_dropped_overflow_packet(&self) {
+        self.ingress
+            .overflow_packets_dropped
+            .fetch_add(1, Ordering::Relaxed);
+    }
+
     pub fn ingress_dropped_final_hop_packet(&self, source: IpAddr) {
         self.ingress
             .final_hop_packets_dropped
@@ -252,6 +258,9 @@ pub struct IngressMixingStats {
     // final hop packets (i.e. to gateway)
     final_hop_packets_dropped: AtomicUsize,
 
+    /// Packets that were dropped because the ingress channel queue was full
+    overflow_packets_dropped: AtomicUsize,
+
     senders: DashMap<IpAddr, IngressRecipientStats>,
 }
 
@@ -284,6 +293,10 @@ impl IngressMixingStats {
         self.final_hop_packets_dropped.load(Ordering::Relaxed)
     }
 
+    pub fn overflow_packets_dropped(&self) -> usize {
+        self.overflow_packets_dropped.load(Ordering::Relaxed)
+    }
+
     pub fn senders(&self) -> &DashMap<IpAddr, IngressRecipientStats> {
         &self.senders
     }

nym-node/src/config/mod.rs

@@ -657,13 +657,10 @@ pub struct ReplayProtectionDebug {
     /// Specifies whether this node should **NOT** use replay protection
     pub unsafe_disabled: bool,
 
-    /// How long the processing task is willing to skip mutex acquisition before it will block the thread
-    /// until it actually obtains it
-    pub maximum_replay_detection_deferral: Duration,
-
-    /// How many packets the processing task is willing to queue before it will block the thread
-    /// until it obtains the mutex
-    pub maximum_replay_detection_pending_packets: usize,
+    /// Channel capacity for the mixnet ingress channel. This determines the maximum number of
+    /// packets that can be queued waiting for ingest processing. Once the queue is full packets
+    /// will still be taken off the wire, but dropped as the node is too busy to handle them.
+    pub ingress_channel_maximum_capacity: usize,
 
     /// Probability of false positives, fraction between 0 and 1 or a number indicating 1-in-p
     pub false_positive_rate: f64,
@@ -689,9 +686,7 @@ pub struct ReplayProtectionDebug {
 }
 
 impl ReplayProtectionDebug {
-    pub const DEFAULT_MAXIMUM_REPLAY_DETECTION_DEFERRAL: Duration = Duration::from_millis(50);
-
-    pub const DEFAULT_MAXIMUM_REPLAY_DETECTION_PENDING_PACKETS: usize = 100;
+    pub const DEFAULT_INGRESS_CHANNEL_MAXIMUM_CAPACITY: usize = 2000;
 
     // 12% (completely arbitrary)
     pub const DEFAULT_BLOOMFILTER_SIZE_MULTIPLIER: f64 = 1.12;
@@ -755,9 +750,7 @@ impl Default for ReplayProtectionDebug {
     fn default() -> Self {
         ReplayProtectionDebug {
             unsafe_disabled: false,
-            maximum_replay_detection_deferral: Self::DEFAULT_MAXIMUM_REPLAY_DETECTION_DEFERRAL,
-            maximum_replay_detection_pending_packets:
-                Self::DEFAULT_MAXIMUM_REPLAY_DETECTION_PENDING_PACKETS,
+            ingress_channel_maximum_capacity: Self::DEFAULT_INGRESS_CHANNEL_MAXIMUM_CAPACITY,
             false_positive_rate: Self::DEFAULT_REPLAY_DETECTION_FALSE_POSITIVE_RATE,
             initial_expected_packets_per_second: Self::DEFAULT_INITIAL_EXPECTED_PACKETS_PER_SECOND,
             bloomfilter_minimum_packets_per_second_size:
@@ -802,7 +795,7 @@ impl MixnetDebug {
     const DEFAULT_PACKET_FORWARDING_INITIAL_BACKOFF: Duration = Duration::from_millis(10_000);
     const DEFAULT_PACKET_FORWARDING_MAXIMUM_BACKOFF: Duration = Duration::from_millis(300_000);
     const DEFAULT_INITIAL_CONNECTION_TIMEOUT: Duration = Duration::from_millis(1_500);
-    const DEFAULT_MAXIMUM_CONNECTION_BUFFER_SIZE: usize = 2000;
+    const DEFAULT_MAXIMUM_CONNECTION_BUFFER_SIZE: usize = 128;
 }
 
 impl Default for MixnetDebug {

nym-node/src/config/old_configs/old_config_v12.rs

@@ -614,16 +614,8 @@ pub async fn try_upgrade_config_v12<P: AsRef<Path>>(
                 },
                 debug: ReplayProtectionDebug {
                     unsafe_disabled: old_cfg.mixnet.replay_protection.debug.unsafe_disabled,
-                    maximum_replay_detection_deferral: old_cfg
-                        .mixnet
-                        .replay_protection
-                        .debug
-                        .maximum_replay_detection_deferral,
-                    maximum_replay_detection_pending_packets: old_cfg
-                        .mixnet
-                        .replay_protection
-                        .debug
-                        .maximum_replay_detection_pending_packets,
+                    ingress_channel_maximum_capacity:
+                        ReplayProtectionDebug::DEFAULT_INGRESS_CHANNEL_MAXIMUM_CAPACITY,
                     false_positive_rate: old_cfg.mixnet.replay_protection.debug.false_positive_rate,
                     initial_expected_packets_per_second: old_cfg
                         .mixnet

nym-node/src/node/lp/cleanup.rs

@@ -0,0 +1,153 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use crate::config::LpDebug;
+use dashmap::DashMap;
+use nym_lp::LpStateMachine;
+use nym_lp::peer_config::LpReceiverIndex;
+use nym_metrics::inc_by;
+use std::sync::Arc;
+use std::time::Duration;
+use tracing::{debug, info};
+
+/// Wrapper for state entries with timestamp tracking for cleanup
+///
+/// This wrapper adds `created_at` and `last_activity` timestamps to state entries,
+/// enabling TTL-based cleanup of stale handshakes and sessions.
+pub struct TimestampedState<T> {
+    /// The actual state (LpStateMachine or LpSession)
+    pub state: T,
+
+    /// When this state was created (never changes)
+    created_at: std::time::Instant,
+
+    /// Last activity timestamp (unix seconds, atomically updated)
+    ///
+    /// For handshakes: never updated (use created_at for TTL)
+    /// For sessions: updated on every packet received
+    last_activity: std::sync::atomic::AtomicU64,
+}
+
+impl<T> TimestampedState<T> {
+    /// Create a new timestamped state
+    pub fn new(state: T) -> Self {
+        let now_instant = std::time::Instant::now();
+        let now_unix = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap_or_default()
+            .as_secs();
+
+        Self {
+            state,
+            created_at: now_instant,
+            last_activity: std::sync::atomic::AtomicU64::new(now_unix),
+        }
+    }
+
+    /// Update last_activity timestamp (cheap, lock-free operation)
+    pub fn touch(&self) {
+        let now = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap_or_default()
+            .as_secs();
+        self.last_activity
+            .store(now, std::sync::atomic::Ordering::Relaxed);
+    }
+
+    /// Get age since creation
+    #[allow(dead_code)]
+    pub fn age(&self) -> Duration {
+        self.created_at.elapsed()
+    }
+
+    /// Get time since last activity
+    pub fn since_activity(&self) -> Duration {
+        let now = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap_or_default()
+            .as_secs();
+        let last = self
+            .last_activity
+            .load(std::sync::atomic::Ordering::Relaxed);
+        Duration::from_secs(now.saturating_sub(last))
+    }
+}
+
+pub(crate) struct CleanupTask {
+    session_states: Arc<DashMap<LpReceiverIndex, TimestampedState<LpStateMachine>>>,
+    cfg: LpDebug,
+    shutdown: nym_task::ShutdownToken,
+}
+
+impl CleanupTask {
+    pub fn new(
+        session_states: Arc<DashMap<LpReceiverIndex, TimestampedState<LpStateMachine>>>,
+        cfg: LpDebug,
+        shutdown: nym_task::ShutdownToken,
+    ) -> Self {
+        CleanupTask {
+            session_states,
+            cfg,
+            shutdown,
+        }
+    }
+
+    fn perform_cleanup(&self) {
+        let session_ttl = self.cfg.session_ttl;
+
+        let start = std::time::Instant::now();
+        let mut ss_removed = 0u64;
+
+        // Remove stale sessions (based on time since last activity)
+        // Use shorter TTL for demoted (ReadOnlyTransport) sessions
+        self.session_states.retain(|_, timestamped| {
+            if timestamped.since_activity() > session_ttl {
+                ss_removed += 1;
+                false
+            } else {
+                true
+            }
+        });
+
+        if ss_removed > 0 {
+            let duration = start.elapsed();
+            info!(
+                "LP state cleanup: {ss_removed} sessions (took {:.3}s)",
+                duration.as_secs_f64()
+            );
+
+            // Track metrics
+            if ss_removed > 0 {
+                inc_by!("lp_states_cleanup_session_removed", ss_removed as i64);
+            }
+        }
+    }
+
+    /// Background loop for cleaning up stale state entries
+    ///
+    /// Runs periodically to scan handshake_states and session_states maps,
+    /// removing entries that have exceeded their TTL.
+    ///
+    /// Demoted sessions (ReadOnlyTransport) use shorter TTL since they
+    /// only need to drain in-flight packets after subsession promotion.
+    pub(crate) async fn run(&self) {
+        let interval = self.cfg.state_cleanup_interval;
+
+        let mut cleanup_interval = tokio::time::interval(interval);
+
+        loop {
+            tokio::select! {
+                biased;
+                _ = self.shutdown.cancelled() => {
+                    debug!("LP state cleanup task: received shutdown signal");
+                    break;
+                }
+                _ = cleanup_interval.tick() => {
+                    self.perform_cleanup();
+                }
+            }
+        }
+
+        info!("LP state cleanup task shutdown complete");
+    }
+}

nym-node/src/node/lp/control/handler.rs

@@ -1,11 +1,13 @@
-// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: GPL-3.0-only
 
-use super::{LpHandlerState, LpReceiverIndex, TimestampedState};
+use crate::node::lp::cleanup::TimestampedState;
 use crate::node::lp::error::LpHandlerError;
+use crate::node::lp::state::SharedLpControlState;
 use dashmap::mapref::one::RefMut;
 use nym_lp::packet::message::LpMessageType;
 use nym_lp::packet::{EncryptedLpPacket, ForwardPacketData, LpMessage};
+use nym_lp::peer_config::LpReceiverIndex;
 use nym_lp::state_machine::{LpAction, LpInput};
 use nym_lp::transport::LpHandshakeChannel;
 use nym_lp::transport::traits::LpTransportChannel;
@@ -76,7 +78,7 @@ impl ConnectionStats {
 pub struct LpConnectionHandler<S = TcpStream> {
     stream: S,
     remote_addr: SocketAddr,
-    state: LpHandlerState,
+    state: SharedLpControlState,
     stats: ConnectionStats,
 
     // /// Flag indicating whether this is a connection from an entry gateway serving as a proxy
@@ -100,7 +102,7 @@ where
         stream: S,
         // forwarded_connection: bool,
         remote_addr: SocketAddr,
-        state: LpHandlerState,
+        state: SharedLpControlState,
     ) -> Self {
         Self {
             stream,
@@ -120,6 +122,7 @@ where
     ) -> Result<RefMut<'_, LpReceiverIndex, TimestampedState<LpStateMachine>>, LpHandlerError> {
         let receiver_index = self.bound_receiver_index()?;
         self.state
+            .shared
             .session_states
             .get_mut(&receiver_index)
             .ok_or_else(|| LpHandlerError::MissingLpSession { receiver_index })
@@ -144,7 +147,7 @@ where
 
         // 1. complete KKT/PSQ handshake before doing anything else.
         // bail if it takes too long
-        let timeout = self.state.lp_config.debug.handshake_ttl;
+        let timeout = self.state.shared.lp_config.debug.handshake_ttl;
         let local_peer = self.state.local_lp_peer.clone();
         let stream = &mut self.stream;
 
@@ -178,6 +181,7 @@ where
         // 2. insert the state machine into the shared state
         let state_machine = LpStateMachine::new(session);
         self.state
+            .shared
             .session_states
             .insert(receiver_idx, TimestampedState::new(state_machine));
         self.bound_receiver_idx = Some(receiver_idx);
@@ -665,8 +669,9 @@ where
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::config::LpConfig;
     use crate::config::lp::LpDebug;
-    use crate::node::lp::LpConfig;
+    use crate::node::lp::state::SharedLpState;
     use nym_lp::peer::{KEMKeys, LpLocalPeer, generate_keypair_mceliece, generate_keypair_mlkem};
     use nym_lp::{Ciphersuite, SessionManager, sessions_for_tests};
     use nym_test_utils::helpers::{deterministic_rng, deterministic_rng_09};
@@ -674,7 +679,7 @@ mod tests {
     // ==================== Test Helpers ====================
 
     /// Create a minimal test state for handler tests
-    async fn create_minimal_test_state() -> LpHandlerState {
+    async fn create_minimal_test_state() -> SharedLpControlState {
         use nym_crypto::asymmetric::ed25519;
 
         let mut rng = deterministic_rng();
@@ -690,9 +695,6 @@ mod tests {
             lp_config.debug.max_concurrent_forwards,
         ));
 
-        // Create mix forwarding channel (unused in tests but required by struct)
-        let (mix_sender, _mix_receiver) = nym_mixnet_client::forwarder::mix_forwarding_channels();
-
         let id_keys = Arc::new(ed25519::KeyPair::new(&mut rng));
         let x_keys = Arc::new(id_keys.to_x25519().try_into().unwrap());
 
@@ -702,14 +704,15 @@ mod tests {
         );
         let lp_peer = LpLocalPeer::new(Ciphersuite::default(), x_keys).with_kem_keys(kem_keys);
 
-        LpHandlerState {
-            lp_config,
+        SharedLpControlState {
             local_lp_peer: lp_peer,
-            metrics: nym_node_metrics::NymNodeMetrics::default(),
-            outbound_mix_sender: mix_sender,
-            session_states: Arc::new(dashmap::DashMap::new()),
             peer_registrator: None,
             forward_semaphore,
+            shared: SharedLpState {
+                lp_config,
+                metrics: nym_node_metrics::NymNodeMetrics::default(),
+                session_states: Arc::new(dashmap::DashMap::new()),
+            },
         }
     }
 

nym-node/src/node/lp/control/listener.rs

@@ -0,0 +1,130 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use crate::config::LpConfig;
+use crate::error::NymNodeError;
+use crate::node::lp::control::handler::LpConnectionHandler;
+use crate::node::lp::state::SharedLpControlState;
+use nym_task::ShutdownTracker;
+use std::net::SocketAddr;
+use tokio::net::TcpListener;
+use tracing::{debug, error, info, trace, warn};
+
+/// LP listener that accepts TCP connections on port 41264
+pub struct LpControlListener {
+    /// Address to bind to
+    bind_address: SocketAddr,
+
+    /// Shared state for connection handlers
+    handler_state: SharedLpControlState,
+
+    /// Shutdown coordination
+    shutdown: ShutdownTracker,
+}
+
+impl LpControlListener {
+    pub fn new(
+        bind_address: SocketAddr,
+        handler_state: SharedLpControlState,
+        shutdown: ShutdownTracker,
+    ) -> Self {
+        Self {
+            bind_address,
+            handler_state,
+            shutdown,
+        }
+    }
+
+    fn lp_config(&self) -> LpConfig {
+        self.handler_state.shared.lp_config
+    }
+
+    pub async fn run(&mut self) -> Result<(), NymNodeError> {
+        let bind_address = self.bind_address;
+        info!("Starting LP control listener on {bind_address}");
+
+        let listener = TcpListener::bind(bind_address).await.map_err(|source| {
+            error!("Failed to bind LP listener to {bind_address}: {source}",);
+            NymNodeError::LpBindFailure {
+                address: bind_address,
+                source,
+            }
+        })?;
+
+        let shutdown_token = self.shutdown.clone_shutdown_token();
+
+        loop {
+            tokio::select! {
+                biased;
+                _ = shutdown_token.cancelled() => {
+                    trace!("LP listener: received shutdown signal");
+                    break;
+                }
+
+                result = listener.accept() => {
+                    match result {
+                        Ok((stream, addr)) => self.handle_connection(stream, addr),
+                        Err(e) => warn!("Failed to accept LP connection: {e}")
+                    }
+                }
+            }
+        }
+
+        info!("LP listener shutdown complete");
+        Ok(())
+    }
+
+    fn handle_connection(&self, stream: tokio::net::TcpStream, remote_addr: SocketAddr) {
+        // Check connection limit
+        let active_connections = self.active_lp_connections();
+        let max_connections = self.lp_config().debug.max_connections;
+        if active_connections >= max_connections {
+            warn!(
+                "LP connection limit exceeded ({active_connections}/{max_connections}), rejecting connection from {remote_addr}"
+            );
+            return;
+        }
+
+        debug!(
+            "Accepting LP connection from {remote_addr} ({active_connections} active connections)"
+        );
+
+        // Increment connection counter
+        self.handler_state
+            .shared
+            .metrics
+            .network
+            .new_lp_connection();
+
+        // Spawn handler task
+        let handler = LpConnectionHandler::new(stream, remote_addr, self.handler_state.clone());
+
+        let metrics = self.handler_state.shared.metrics.clone();
+        self.shutdown.try_spawn_named_with_shutdown(
+            async move {
+                let result = handler.handle().await;
+
+                // Handler emits lifecycle metrics internally on success
+                // For errors, we need to emit them here since handler is consumed
+                if let Err(e) = result {
+                    warn!("LP handler error for {remote_addr}: {e}");
+                    // Note: metrics are emitted in handle() for graceful path
+                    // On error path, handle() returns early without emitting
+                    // So we track errors here
+                }
+
+                // Decrement connection counter on exit
+                metrics.network.lp_connection_closed();
+            },
+            &format!("LP::{remote_addr}"),
+        );
+    }
+
+    fn active_lp_connections(&self) -> usize {
+        self.handler_state
+            .shared
+            .metrics
+            .network
+            .active_lp_connections_count()
+    }
+}

nym-node/src/node/lp/control/mod.rs

@@ -0,0 +1,5 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+pub mod handler;
+pub(crate) mod listener;

nym-node/src/node/lp/data/handler.rs

@@ -1,5 +1,5 @@
-// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: GPL-3.0-only
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
 
 //! LP Data Handler - UDP listener for LP data plane (port 51264)
 //!
@@ -15,90 +15,24 @@
 //! ```
 //!
 
-use super::LpHandlerState;
-use crate::error::NymNodeError;
 use crate::node::lp::error::LpHandlerError;
+use crate::node::lp::state::SharedLpDataState;
 use nym_lp::packet::OuterHeader;
 use nym_metrics::inc;
 use std::net::SocketAddr;
-use std::sync::Arc;
-use tokio::net::UdpSocket;
 use tracing::*;
 
-/// Maximum UDP packet size we'll accept
-/// Sphinx packets are typically ~2KB, LP overhead is ~50 bytes, so 4KB is plenty
-const MAX_UDP_PACKET_SIZE: usize = 4096;
-
 /// LP Data Handler for UDP data plane
 pub struct LpDataHandler {
-    /// UDP socket for receiving LP-wrapped Sphinx packets
-    socket: Arc<UdpSocket>,
-
-    /// Shared state with TCP control plane
+    /// State used for handling received requests
     #[allow(dead_code)]
-    state: LpHandlerState,
-
-    /// Shutdown token
-    shutdown: nym_task::ShutdownToken,
+    state: SharedLpDataState,
 }
 
 impl LpDataHandler {
     /// Create a new LP data handler
-    pub async fn new(
-        bind_addr: SocketAddr,
-        state: LpHandlerState,
-        shutdown: nym_task::ShutdownToken,
-    ) -> Result<Self, NymNodeError> {
-        let socket = UdpSocket::bind(bind_addr).await.map_err(|source| {
-            error!("Failed to bind LP data socket to {bind_addr}: {source}");
-            NymNodeError::LpBindFailure {
-                address: bind_addr,
-                source,
-            }
-        })?;
-
-        info!("LP data handler listening on UDP {bind_addr}");
-
-        Ok(Self {
-            socket: Arc::new(socket),
-            state,
-            shutdown,
-        })
-    }
-
-    /// Run the UDP packet receive loop
-    pub async fn run(self) -> Result<(), LpHandlerError> {
-        let mut buf = vec![0u8; MAX_UDP_PACKET_SIZE];
-
-        loop {
-            tokio::select! {
-                biased;
-
-                _ = self.shutdown.cancelled() => {
-                    info!("LP data handler: received shutdown signal");
-                    break;
-                }
-
-                result = self.socket.recv_from(&mut buf) => {
-                    match result {
-                        Ok((len, src_addr)) => {
-                            // Process packet in place (no spawn - UDP is fast)
-                            if let Err(e) = self.handle_packet(&buf[..len], src_addr).await {
-                                debug!("LP data packet error from {src_addr}: {e}");
-                                inc!("lp_data_packet_errors");
-                            }
-                        }
-                        Err(e) => {
-                            warn!("LP data socket recv error: {e}");
-                            inc!("lp_data_recv_errors");
-                        }
-                    }
-                }
-            }
-        }
-
-        info!("LP data handler shutdown complete");
-        Ok(())
+    pub fn new(state: SharedLpDataState) -> Self {
+        Self { state }
     }
 
     /// Handle a single UDP packet
@@ -115,7 +49,7 @@ impl LpDataHandler {
     /// - Marking counter as used after successful decryption
     ///
     /// This prevents replay attacks where captured packets are re-sent.
-    async fn handle_packet(
+    pub(crate) async fn handle_packet(
         &self,
         packet: &[u8],
         src_addr: SocketAddr,
@@ -223,15 +157,3 @@ impl LpDataHandler {
         // }
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    // Sphinx packets are typically around 2KB
-    // LP overhead is small (~50 bytes header + AEAD tag)
-    // 4KB should be plenty with room to spare
-    const _: () = {
-        assert!(MAX_UDP_PACKET_SIZE >= 2048 + 100);
-    };
-}

nym-node/src/node/lp/data/listener.rs

@@ -0,0 +1,81 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use crate::error::NymNodeError;
+use crate::node::lp::data::MAX_UDP_PACKET_SIZE;
+use crate::node::lp::data::handler::LpDataHandler;
+use crate::node::lp::state::SharedLpDataState;
+use nym_metrics::inc;
+use std::net::SocketAddr;
+use tokio::net::UdpSocket;
+use tracing::log::warn;
+use tracing::{debug, error, info};
+
+/// LP UDP listener that accepts TCP connections on port 51264 (by default)
+pub struct LpDataListener {
+    /// Address to bind to
+    bind_address: SocketAddr,
+
+    /// State used for handling received requests
+    handler: LpDataHandler,
+
+    /// Shutdown token
+    shutdown: nym_task::ShutdownToken,
+}
+
+impl LpDataListener {
+    pub fn new(
+        bind_address: SocketAddr,
+        state: SharedLpDataState,
+        shutdown: nym_task::ShutdownToken,
+    ) -> Self {
+        Self {
+            bind_address,
+            handler: LpDataHandler::new(state),
+            shutdown,
+        }
+    }
+
+    pub async fn run(&self) -> Result<(), NymNodeError> {
+        let bind_address = self.bind_address;
+        info!("Starting LP data listener on {bind_address}");
+        let socket = UdpSocket::bind(bind_address).await.map_err(|source| {
+            error!("Failed to bind LP data socket to {bind_address}: {source}");
+            NymNodeError::LpBindFailure {
+                address: bind_address,
+                source,
+            }
+        })?;
+
+        let mut buf = vec![0u8; MAX_UDP_PACKET_SIZE];
+
+        loop {
+            tokio::select! {
+                biased;
+                _ = self.shutdown.cancelled() => {
+                    info!("LP data listener: received shutdown signal");
+                    break;
+                }
+
+                result = socket.recv_from(&mut buf) => {
+                    match result {
+                        Ok((len, src_addr)) => {
+                            // Process packet in place (no spawn - UDP is fast)
+                            if let Err(e) = self.handler.handle_packet(&buf[..len], src_addr).await {
+                                debug!("LP data packet error from {src_addr}: {e}");
+                                inc!("lp_data_packet_errors");
+                            }
+                        }
+                        Err(e) => {
+                            warn!("LP data socket recv error: {e}");
+                            inc!("lp_data_recv_errors");
+                        }
+                    }
+                }
+            }
+        }
+
+        info!("LP data handler shutdown complete");
+        Ok(())
+    }
+}

nym-node/src/node/lp/data/mod.rs

@@ -0,0 +1,20 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+/// Maximum UDP packet size we'll accept
+/// Sphinx packets are typically ~2KB, LP overhead is ~50 bytes, so 4KB is plenty
+const MAX_UDP_PACKET_SIZE: usize = 4096;
+
+pub mod handler;
+pub(crate) mod listener;
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // Sphinx packets are typically around 2KB
+    // 4KB should be plenty with room to spare
+    const _: () = {
+        assert!(MAX_UDP_PACKET_SIZE >= 2048 + 100);
+    };
+}

nym-node/src/node/lp/error.rs

@@ -1,8 +1,8 @@
 // Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use crate::node::lp::LpReceiverIndex;
 use nym_lp::packet::message::LpMessageType;
+use nym_lp::peer_config::LpReceiverIndex;
 use nym_lp::state_machine::LpAction;
 use nym_lp::transport::LpTransportError;
 use nym_lp::{LpError, packet::MalformedLpPacketError};

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

@@ -67,404 +67,125 @@
 // To view metrics, the nym-metrics registry automatically collects all metrics.
 // They can be exported via Prometheus format using the metrics endpoint.
 
-use crate::config::lp::LpConfig;
+use crate::config::LpConfig;
 use crate::error::NymNodeError;
+use crate::node::lp::cleanup::CleanupTask;
+use crate::node::lp::control::listener::LpControlListener;
+use crate::node::lp::data::listener::LpDataListener;
 use dashmap::DashMap;
 use nym_gateway::node::wireguard::PeerRegistrator;
 use nym_lp::peer::LpLocalPeer;
-use nym_lp::peer_config::LpReceiverIndex;
-use nym_lp::state_machine::LpStateMachine;
 use nym_mixnet_client::forwarder::MixForwardingSender;
 use nym_node_metrics::NymNodeMetrics;
 use nym_task::ShutdownTracker;
-use std::net::SocketAddr;
 use std::sync::Arc;
-use std::time::Duration;
-use tokio::net::TcpListener;
 use tokio::sync::Semaphore;
-use tracing::*;
+use tracing::error;
 
 pub use nym_mixnet_client::forwarder::{MixForwardingReceiver, mix_forwarding_channels};
+pub use state::{SharedLpControlState, SharedLpDataState, SharedLpState};
 
-mod data_handler;
+mod cleanup;
+pub mod control;
+mod data;
 pub mod error;
-pub mod handler;
 mod registration;
+pub mod state;
 
-/// Wrapper for state entries with timestamp tracking for cleanup
-///
-/// This wrapper adds `created_at` and `last_activity` timestamps to state entries,
-/// enabling TTL-based cleanup of stale handshakes and sessions.
-pub struct TimestampedState<T> {
-    /// The actual state (LpStateMachine or LpSession)
-    pub state: T,
-
-    /// When this state was created (never changes)
-    created_at: std::time::Instant,
-
-    /// Last activity timestamp (unix seconds, atomically updated)
-    ///
-    /// For handshakes: never updated (use created_at for TTL)
-    /// For sessions: updated on every packet received
-    last_activity: std::sync::atomic::AtomicU64,
-}
-
-impl<T> TimestampedState<T> {
-    /// Create a new timestamped state
-    pub fn new(state: T) -> Self {
-        let now_instant = std::time::Instant::now();
-        let now_unix = std::time::SystemTime::now()
-            .duration_since(std::time::UNIX_EPOCH)
-            .unwrap_or_default()
-            .as_secs();
-
-        Self {
-            state,
-            created_at: now_instant,
-            last_activity: std::sync::atomic::AtomicU64::new(now_unix),
-        }
-    }
-
-    /// Update last_activity timestamp (cheap, lock-free operation)
-    pub fn touch(&self) {
-        let now = std::time::SystemTime::now()
-            .duration_since(std::time::UNIX_EPOCH)
-            .unwrap_or_default()
-            .as_secs();
-        self.last_activity
-            .store(now, std::sync::atomic::Ordering::Relaxed);
-    }
-
-    /// Get age since creation
-    #[allow(dead_code)]
-    pub fn age(&self) -> Duration {
-        self.created_at.elapsed()
-    }
-
-    /// Get time since last activity
-    pub fn since_activity(&self) -> Duration {
-        let now = std::time::SystemTime::now()
-            .duration_since(std::time::UNIX_EPOCH)
-            .unwrap_or_default()
-            .as_secs();
-        let last = self
-            .last_activity
-            .load(std::sync::atomic::Ordering::Relaxed);
-        Duration::from_secs(now.saturating_sub(last))
-    }
-}
-
-/// Shared state for LP connection handlers
-#[derive(Clone)]
-pub struct LpHandlerState {
-    /// Encapsulates all required key information of a local Lewes Protocol Peer.
-    pub local_lp_peer: LpLocalPeer,
-
-    /// Metrics collection
-    pub metrics: NymNodeMetrics,
-
-    /// Handle registering new wireguard peers
-    pub peer_registrator: Option<PeerRegistrator>,
-
-    /// LP configuration (for timestamp validation, etc.)
-    pub lp_config: LpConfig,
-
-    /// Channel for forwarding Sphinx packets into the mixnet
-    ///
-    /// Used by the LP data handler (UDP:51264) to forward decrypted Sphinx packets
-    /// from LP clients into the mixnet for routing.
-    #[allow(dead_code)]
-    pub outbound_mix_sender: MixForwardingSender,
-
-    /// Established sessions keyed by session_id
-    ///
-    /// Used after handshake completes (session_id is deterministically computed from
-    /// both parties' X25519 keys). Enables stateless transport - each packet lookup
-    /// by session_id, decrypt/process, respond.
-    ///
-    /// Wrapped in TimestampedState for TTL-based cleanup of inactive sessions.
-    ///
-    /// Sessions are stored as LpStateMachine (not LpSession) to enable
-    /// subsession/rekeying support. The state machine handles subsession initiation
-    /// (SubsessionKK1/KK2/Ready) during transport phase, allowing long-lived connections
-    /// to rekey without re-authentication.
-    pub session_states: Arc<DashMap<LpReceiverIndex, TimestampedState<LpStateMachine>>>,
-
-    /// Semaphore limiting concurrent forward connections
-    ///
-    /// Prevents file descriptor exhaustion when forwarding LP packets during
-    /// telescope setup. When at capacity, forward requests return an error
-    /// so clients can choose a different gateway.
-    // Connection limiting (not pooling) chosen for forward requests.
-    //
-    // Why not connection pooling?
-    // 1. Forwarding is one-time per telescope setup (handshake only), not ongoing traffic.
-    //    Once telescope is established, data flows directly through the tunnel.
-    // 2. Telescope targets are distributed across many different gateways - each client
-    //    typically connects to a different exit gateway, so pooled connections would
-    //    rarely be reused.
-    // 3. Connections already go out of scope after each request-response. FD exhaustion
-    //    only happens from concurrent spikes, not accumulation.
-    // 4. A pool would accumulate one idle connection per unique destination, most of
-    //    which would never be reused before TTL expiration.
-    //
-    // Why semaphore limiting is better:
-    // 1. Directly caps concurrent forward connections regardless of destination.
-    // 2. When at capacity, returns "busy" error - client can choose another gateway.
-    //    This is better than silently queuing requests behind a pool.
-    // 3. Simple implementation: no TTL management, stale connection handling, or cleanup.
-    pub forward_semaphore: Arc<Semaphore>,
-}
-
-/// LP listener that accepts TCP connections on port 41264
-pub struct LpListener {
-    /// Shared state for connection handlers
-    handler_state: LpHandlerState,
+pub struct LpSetup {
+    control_listener: LpControlListener,
+    data_listener: LpDataListener,
+    cleanup_task: CleanupTask,
 
     /// Shutdown coordination
     shutdown: ShutdownTracker,
 }
 
-impl LpListener {
-    pub fn new(handler_state: LpHandlerState, shutdown: ShutdownTracker) -> Self {
-        Self {
-            handler_state,
+impl LpSetup {
+    pub async fn new(
+        local_lp_peer: LpLocalPeer,
+        lp_config: LpConfig,
+        metrics: NymNodeMetrics,
+        peer_registrator: Option<PeerRegistrator>,
+        mix_packet_sender: MixForwardingSender,
+        shutdown: ShutdownTracker,
+    ) -> Result<Self, NymNodeError> {
+        // TODO: this will require loading old states from disk in the future
+        let session_states = Arc::new(DashMap::new());
+
+        let shared_lp_state = SharedLpState {
+            metrics,
+            lp_config,
+            session_states: session_states.clone(),
+        };
+
+        let control_state = SharedLpControlState {
+            local_lp_peer,
+            peer_registrator,
+            forward_semaphore: Arc::new(Semaphore::new(lp_config.debug.max_concurrent_forwards)),
+            shared: shared_lp_state.clone(),
+        };
+
+        let data_state = SharedLpDataState {
+            outbound_mix_sender: mix_packet_sender,
+            shared: shared_lp_state,
+        };
+
+        let control_listener = LpControlListener::new(
+            lp_config.control_bind_address,
+            control_state,
+            shutdown.clone(),
+        );
+        let data_listener = LpDataListener::new(
+            lp_config.data_bind_address,
+            data_state,
+            shutdown.clone_shutdown_token(),
+        );
+        let cleanup_task = CleanupTask::new(
+            session_states,
+            lp_config.debug,
+            shutdown.clone_shutdown_token(),
+        );
+
+        Ok(LpSetup {
+            control_listener,
+            data_listener,
+            cleanup_task,
             shutdown,
-        }
+        })
     }
 
-    fn lp_config(&self) -> LpConfig {
-        self.handler_state.lp_config
-    }
-
-    pub async fn run(&mut self) -> Result<(), NymNodeError> {
-        let control_bind_address = self.lp_config().control_bind_address;
-        let data_bind_address = self.lp_config().data_bind_address;
-        let listener = TcpListener::bind(control_bind_address)
-            .await
-            .map_err(|source| {
-                error!("Failed to bind LP listener to {control_bind_address}: {source}",);
-                NymNodeError::LpBindFailure {
-                    address: control_bind_address,
-                    source,
-                }
-            })?;
-
+    pub fn start_tasks(mut self) {
+        // control listener
         let shutdown_token = self.shutdown.clone_shutdown_token();
-
-        // Spawn background task for state cleanup
-        let _cleanup_handle = self.spawn_state_cleanup_task();
-
-        // Spawn UDP data handler for LP data plane (port 51264)
-        let _data_handler_handle = self.spawn_data_handler().await?;
-
-        info!(
-            "LP listener started on {control_bind_address} (data handler on: {data_bind_address})",
-        );
-
-        loop {
-            tokio::select! {
-                biased;
-                _ = shutdown_token.cancelled() => {
-                    trace!("LP listener: received shutdown signal");
-                    break;
-                }
-
-                result = listener.accept() => {
-                    match result {
-                        Ok((stream, addr)) => self.handle_connection(stream, addr),
-                        Err(e) => warn!("Failed to accept LP connection: {e}")
-                    }
-                }
-            }
-        }
-
-        info!("LP listener shutdown complete");
-        Ok(())
-    }
-
-    fn handle_connection(&self, stream: tokio::net::TcpStream, remote_addr: SocketAddr) {
-        // Check connection limit
-        let active_connections = self.active_lp_connections();
-        let max_connections = self.lp_config().debug.max_connections;
-        if active_connections >= max_connections {
-            warn!(
-                "LP connection limit exceeded ({active_connections}/{max_connections}), rejecting connection from {remote_addr}"
-            );
-            return;
-        }
-
-        debug!(
-            "Accepting LP connection from {remote_addr} ({active_connections} active connections)"
-        );
-
-        // Increment connection counter
-        self.handler_state.metrics.network.new_lp_connection();
-
-        // Spawn handler task
-        let handler =
-            handler::LpConnectionHandler::new(stream, remote_addr, self.handler_state.clone());
-
-        let metrics = self.handler_state.metrics.clone();
-        self.shutdown.try_spawn_named_with_shutdown(
-            async move {
-                let result = handler.handle().await;
-
-                // Handler emits lifecycle metrics internally on success
-                // For errors, we need to emit them here since handler is consumed
-                if let Err(e) = result {
-                    warn!("LP handler error for {remote_addr}: {e}");
-                    // Note: metrics are emitted in handle() for graceful path
-                    // On error path, handle() returns early without emitting
-                    // So we track errors here
-                }
-
-                // Decrement connection counter on exit
-                metrics.network.lp_connection_closed();
-            },
-            &format!("LP::{remote_addr}"),
-        );
-    }
-
-    /// Spawn the UDP data handler for LP data plane
-    ///
-    /// The data handler listens on UDP port 51264 and processes LP-wrapped Sphinx packets
-    /// from registered clients. It decrypts the LP layer and forwards the Sphinx packets
-    /// into the mixnet.
-    async fn spawn_data_handler(&self) -> Result<tokio::task::JoinHandle<()>, NymNodeError> {
-        // Create data handler
-        let data_handler = data_handler::LpDataHandler::new(
-            self.lp_config().data_bind_address,
-            self.handler_state.clone(),
-            self.shutdown.clone_shutdown_token(),
-        )
-        .await?;
-
-        // Spawn data handler task
-        let handle = self.shutdown.try_spawn_named(
-            async move {
-                if let Err(e) = data_handler.run().await {
-                    error!("LP data handler error: {e}");
-                }
-            },
-            "LP::DataHandler",
-        );
-
-        Ok(handle)
-    }
-
-    /// Spawn background task for cleaning up stale state entries
-    ///
-    /// This task runs periodically (every `state_cleanup_interval_secs`) to remove:
-    /// - Handshake states older than `handshake_ttl_secs`
-    /// - Session states with no activity for `session_ttl_secs`
-    ///
-    /// The task automatically stops when the shutdown signal is received.
-    fn spawn_state_cleanup_task(&self) -> tokio::task::JoinHandle<()> {
-        let session_states = Arc::clone(&self.handler_state.session_states);
-        let dbg_cfg = self.handler_state.lp_config.debug;
-
-        let handshake_ttl = dbg_cfg.handshake_ttl;
-        let session_ttl = dbg_cfg.session_ttl;
-        let interval = dbg_cfg.state_cleanup_interval;
-        let shutdown = self.shutdown.clone_shutdown_token();
-        let metrics = self.handler_state.metrics.clone();
-
-        info!(
-            "Starting LP state cleanup task (handshake_ttl={}s, session_ttl={}s, interval={}s)",
-            handshake_ttl.as_secs(),
-            session_ttl.as_secs(),
-            interval.as_secs()
-        );
-
         self.shutdown.try_spawn_named(
-            cleanup_task::cleanup_loop(session_states, dbg_cfg, shutdown, metrics),
-            "LP::StateCleanup",
-        )
-    }
+            async move {
+                if let Err(err) = self.control_listener.run().await {
+                    shutdown_token.cancel();
+                    error!("LP control listener error: {err}");
+                }
+            },
+            "LP::LpControlListener",
+        );
 
-    fn active_lp_connections(&self) -> usize {
-        self.handler_state
-            .metrics
-            .network
-            .active_lp_connections_count()
-    }
-}
-
-pub(crate) mod cleanup_task {
-    use crate::config::lp::LpDebug;
-    use crate::node::lp::{LpReceiverIndex, TimestampedState};
-    use dashmap::DashMap;
-    use nym_lp::LpStateMachine;
-    use nym_metrics::inc_by;
-    use nym_node_metrics::NymNodeMetrics;
-    use std::sync::Arc;
-    use tracing::{debug, info};
-
-    async fn perform_cleanup(
-        session_states: &Arc<DashMap<LpReceiverIndex, TimestampedState<LpStateMachine>>>,
-        cfg: LpDebug,
-    ) {
-        let session_ttl = cfg.session_ttl;
-
-        let start = std::time::Instant::now();
-        let mut ss_removed = 0u64;
-
-        // Remove stale sessions (based on time since last activity)
-        // Use shorter TTL for demoted (ReadOnlyTransport) sessions
-        session_states.retain(|_, timestamped| {
-            if timestamped.since_activity() > session_ttl {
-                ss_removed += 1;
-                false
-            } else {
-                true
-            }
-        });
-
-        if ss_removed > 0 {
-            let duration = start.elapsed();
-            info!(
-                "LP state cleanup: {ss_removed} sessions (took {:.3}s)",
-                duration.as_secs_f64()
-            );
-
-            // Track metrics
-            if ss_removed > 0 {
-                inc_by!("lp_states_cleanup_session_removed", ss_removed as i64);
-            }
-        }
-    }
-
-    /// Background loop for cleaning up stale state entries
-    ///
-    /// Runs periodically to scan handshake_states and session_states maps,
-    /// removing entries that have exceeded their TTL.
-    ///
-    /// Demoted sessions (ReadOnlyTransport) use shorter TTL since they
-    /// only need to drain in-flight packets after subsession promotion.
-    pub(crate) async fn cleanup_loop(
-        session_states: Arc<DashMap<LpReceiverIndex, TimestampedState<LpStateMachine>>>,
-        cfg: LpDebug,
-        shutdown: nym_task::ShutdownToken,
-        _metrics: NymNodeMetrics,
-    ) {
-        let interval = cfg.state_cleanup_interval;
-
-        let mut cleanup_interval = tokio::time::interval(interval);
-
-        loop {
-            tokio::select! {
-                biased;
-                _ = shutdown.cancelled() => {
-                    debug!("LP state cleanup task: received shutdown signal");
-                    break;
-                }
-                _ = cleanup_interval.tick() => {
-                    perform_cleanup(&session_states, cfg).await;
-                }
-            }
-        }
-
-        info!("LP state cleanup task shutdown complete");
+        // Spawn the UDP data handler for LP data plane
+        // The data handler listens on UDP port 51264 and processes LP-wrapped Sphinx packets
+        // from registered clients. It decrypts the LP layer and forwards the Sphinx packets
+        let shutdown_token = self.shutdown.clone_shutdown_token();
+        self.shutdown.try_spawn_named(
+            async move {
+                if let Err(err) = self.data_listener.run().await {
+                    shutdown_token.cancel();
+                    error!("LP data listener error: {err}");
+                }
+            },
+            "LP::LpDataListener",
+        );
+
+        // cleanup task
+        self.shutdown.try_spawn_named(
+            async move { self.cleanup_task.run().await },
+            "LP::CleanupTask",
+        );
     }
 }

nym-node/src/node/lp/registration.rs

@@ -1,7 +1,8 @@
 // Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: GPL-3.0-only
 
-use crate::node::lp::{LpHandlerState, LpReceiverIndex};
+use crate::node::lp::state::SharedLpControlState;
+use nym_lp::peer_config::LpReceiverIndex;
 use nym_metrics::{add_histogram_obs, inc};
 use nym_registration_common::dvpn::{
     LpDvpnRegistrationFinalisation, LpDvpnRegistrationInitialRequest,
@@ -28,7 +29,7 @@ const LP_REGISTRATION_DURATION_BUCKETS: &[f64] = &[
     30.0, // 30s
 ];
 
-impl LpHandlerState {
+impl SharedLpControlState {
     async fn process_dvpn_initial_registration(
         &self,
         sender: LpReceiverIndex,

nym-node/src/node/lp/state.rs

@@ -0,0 +1,64 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use crate::config::LpConfig;
+use crate::node::lp::cleanup::TimestampedState;
+use dashmap::DashMap;
+use nym_gateway::node::wireguard::PeerRegistrator;
+use nym_lp::LpStateMachine;
+use nym_lp::peer::LpLocalPeer;
+use nym_lp::peer_config::LpReceiverIndex;
+use nym_mixnet_client::forwarder::MixForwardingSender;
+use nym_node_metrics::NymNodeMetrics;
+use std::sync::Arc;
+use tokio::sync::Semaphore;
+
+/// Shared state for LP control connections
+#[derive(Clone)]
+pub struct SharedLpControlState {
+    /// Encapsulates all required key information of a local Lewes Protocol Peer.
+    pub local_lp_peer: LpLocalPeer,
+
+    /// Handle registering new wireguard peers
+    pub peer_registrator: Option<PeerRegistrator>,
+
+    /// Semaphore limiting concurrent forward connections
+    ///
+    /// Prevents file descriptor exhaustion when forwarding LP packets during
+    /// telescope setup. When at capacity, forward requests return an error
+    /// so clients can choose a different gateway.
+    // this is temporary until there is persistent KKT/PSQ session between nodes
+    pub forward_semaphore: Arc<Semaphore>,
+
+    /// Common shared data
+    pub shared: SharedLpState,
+}
+
+/// Shared state for LP data connections
+#[derive(Clone)]
+pub struct SharedLpDataState {
+    /// Channel for forwarding Sphinx packets into the mixnet
+    ///
+    /// Used by the LP data handler (UDP:51264) to forward decrypted Sphinx packets
+    /// from LP clients into the mixnet for routing.
+    #[allow(dead_code)]
+    pub outbound_mix_sender: MixForwardingSender,
+
+    /// Common shared data
+    pub shared: SharedLpState,
+}
+
+/// Shared state for LP connection handlers
+#[derive(Clone)]
+pub struct SharedLpState {
+    /// Metrics collection
+    pub metrics: NymNodeMetrics,
+
+    /// LP configuration (for timestamp validation, etc.)
+    pub lp_config: LpConfig,
+
+    /// Established sessions keyed by receiver index
+    ///
+    /// Wrapped in TimestampedState for TTL-based cleanup of inactive sessions.
+    pub session_states: Arc<DashMap<LpReceiverIndex, TimestampedState<LpStateMachine>>>,
+}

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

@@ -1,675 +0,0 @@
-// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: GPL-3.0-only
-
-use crate::node::key_rotation::active_keys::SphinxKeyGuard;
-use crate::node::mixnet::shared::SharedData;
-use futures::StreamExt;
-use nym_noise::connection::Connection;
-use nym_noise::upgrade_noise_responder;
-use nym_sphinx_forwarding::packet::MixPacket;
-use nym_sphinx_framing::codec::NymCodec;
-use nym_sphinx_framing::packet::FramedNymPacket;
-use nym_sphinx_framing::processing::{
-    MixProcessingResult, MixProcessingResultData, PacketProcessingError, PartiallyUnwrappedPacket,
-    PartialyUnwrappedPacketWithKeyRotation, ProcessedFinalHop, process_framed_packet,
-};
-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;
-use tokio::time::Instant;
-use tokio_util::codec::Framed;
-use tracing::{Span, debug, error, instrument, trace, warn};
-
-/// How often (in packets) the stream-level span updates its packet count.
-const SPAN_UPDATE_INTERVAL: u64 = 10_000;
-
-struct PendingReplayCheckPackets {
-    // 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: Default::default(),
-            last_acquired_mutex: Instant::now(),
-        }
-    }
-
-    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: PartialyUnwrappedPacketWithKeyRotation) {
-        if self.packets.is_empty() {
-            self.last_acquired_mutex = now;
-        }
-        self.packets
-            .entry(packet.used_key_rotation)
-            .or_default()
-            .push(packet.packet)
-    }
-
-    fn total_count(&self) -> usize {
-        self.packets.values().map(|v| v.len()).sum()
-    }
-
-    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
-    }
-}
-
-pub(crate) struct ConnectionHandler {
-    shared: SharedData,
-    remote_address: SocketAddr,
-
-    // packets pending for replay detection
-    pending_packets: PendingReplayCheckPackets,
-}
-
-impl Drop for ConnectionHandler {
-    fn drop(&mut self) {
-        self.shared
-            .metrics
-            .network
-            .disconnected_ingress_mixnet_client()
-    }
-}
-
-impl ConnectionHandler {
-    pub(crate) fn new(shared: &SharedData, remote_address: SocketAddr) -> Self {
-        shared.metrics.network.new_active_ingress_mixnet_client();
-
-        ConnectionHandler {
-            shared: SharedData {
-                processing_config: shared.processing_config,
-                sphinx_keys: shared.sphinx_keys.clone(),
-                replay_protection_filter: shared.replay_protection_filter.clone(),
-                mixnet_forwarder: shared.mixnet_forwarder.clone(),
-                final_hop: shared.final_hop.clone(),
-                noise_config: shared.noise_config.clone(),
-                metrics: shared.metrics.clone(),
-                shutdown_token: shared.shutdown_token.child_token(),
-            },
-            remote_address,
-            pending_packets: PendingReplayCheckPackets::new(),
-        }
-    }
-
-    /// Determine instant at which packet should get forwarded to the next hop.
-    /// By using [`Instant`] rather than explicit [`Duration`] we minimise effects of
-    /// the skew caused by being stuck in the channel queue.
-    /// This method also clamps the maximum allowed delay so that nobody could send a bunch of packets
-    /// with, for example, delays of 1 year thus causing denial of service
-    fn create_delay_target(&self, now: Instant, delay: Option<Delay>) -> Option<Instant> {
-        let delay = delay?.to_duration();
-
-        let delay = if delay > self.shared.processing_config.maximum_packet_delay {
-            self.shared.processing_config.maximum_packet_delay
-        } else {
-            delay
-        };
-        trace!(
-            "received packet will be delayed for {}ms",
-            delay.as_millis()
-        );
-
-        Some(now + delay)
-    }
-
-    #[instrument(
-        name = "mixnode.forward_packet",
-        skip(self, mix_packet, delay),
-        level = "debug",
-        fields(
-            remote_addr = %self.remote_address,
-            delay_ms = tracing::field::Empty,
-        )
-    )]
-    fn handle_forward_packet(&self, now: Instant, mix_packet: MixPacket, delay: Option<Delay>) {
-        if !self.shared.processing_config.forward_hop_processing_enabled {
-            warn!(
-                event = "packet.dropped.forward_disabled",
-                remote_addr = %self.remote_address,
-                "dropping packet: forward hop processing disabled"
-            );
-            self.shared.dropped_forward_packet(self.remote_address.ip());
-            return;
-        }
-
-        let forward_instant = self.create_delay_target(now, delay);
-        if let Some(target) = forward_instant {
-            Span::current().record(
-                "delay_ms",
-                target.saturating_duration_since(now).as_millis() as u64,
-            );
-        }
-        self.shared.forward_mix_packet(mix_packet, forward_instant);
-    }
-
-    #[instrument(
-        name = "mixnode.final_hop",
-        skip(self, final_hop_data),
-        level = "debug",
-        fields(
-            remote_addr = %self.remote_address,
-            client_online,
-            disk_fallback = false,
-            ack_forwarded = false,
-        )
-    )]
-    async fn handle_final_hop(&self, final_hop_data: ProcessedFinalHop) {
-        if !self.shared.processing_config.final_hop_processing_enabled {
-            warn!(
-                event = "packet.dropped.final_hop_disabled",
-                remote_addr = %self.remote_address,
-                "dropping packet: final hop processing disabled"
-            );
-            self.shared
-                .dropped_final_hop_packet(self.remote_address.ip());
-            return;
-        }
-
-        let client = final_hop_data.destination;
-        let message = final_hop_data.message;
-        let has_ack = final_hop_data.forward_ack.is_some();
-
-        // if possible attempt to push message directly to the client
-        match self.shared.try_push_message_to_client(client, message) {
-            Err(unsent_plaintext) => {
-                // if that failed, store it on disk
-                Span::current().record("client_online", false);
-                match self
-                    .shared
-                    .store_processed_packet_payload(client, unsent_plaintext)
-                    .await
-                {
-                    Err(err) => error!("Failed to store client data - {err}"),
-                    Ok(_) => {
-                        Span::current().record("disk_fallback", true);
-                        self.shared
-                            .metrics
-                            .mixnet
-                            .egress
-                            .add_disk_persisted_packet();
-                        trace!("Stored packet for {client}")
-                    }
-                }
-            }
-            Ok(_) => {
-                Span::current().record("client_online", true);
-                trace!("Pushed received packet to {client}");
-            }
-        }
-
-        // if we managed to either push message directly to the [online] client or store it at
-        // disk, forward the ack
-        self.shared.forward_ack_packet(final_hop_data.forward_ack);
-        if has_ack {
-            Span::current().record("ack_forwarded", true);
-        }
-    }
-
-    fn within_deferral_threshold(&self, now: Instant) -> bool {
-        let time_threshold = now
-            .saturating_duration_since(self.pending_packets.last_acquired_mutex)
-            <= self
-                .shared
-                .processing_config
-                .maximum_replay_detection_deferral;
-
-        let count_threshold = self.pending_packets.packets.len()
-            < self
-                .shared
-                .processing_config
-                .maximum_replay_detection_pending_packets;
-
-        // time threshold is ignored if we currently have 0 packets queued up
-        if self.pending_packets.packets.is_empty() {
-            return true;
-        }
-
-        trace!(
-            "within deferral time threshold: {time_threshold}, count threshold: {count_threshold}"
-        );
-
-        if !time_threshold {
-            warn!(
-                event = "replay_detection.deferral_exceeded",
-                threshold_type = "time",
-                deferred_count = self.pending_packets.total_count(),
-                deferral_ms = now.saturating_duration_since(self.pending_packets.last_acquired_mutex).as_millis() as u64,
-                remote_addr = %self.remote_address,
-                "{}: time deferral threshold exceeded with {} pending packets",
-                self.remote_address,
-                self.pending_packets.total_count()
-            )
-        }
-
-        if !count_threshold {
-            warn!(
-                event = "replay_detection.deferral_exceeded",
-                threshold_type = "count",
-                deferred_count = self.pending_packets.total_count(),
-                remote_addr = %self.remote_address,
-                "{}: count deferral threshold exceeded",
-                self.remote_address
-            )
-        }
-
-        time_threshold && count_threshold
-    }
-
-    /// Resolve the sphinx key for the given rotation, recording the rotation
-    /// label on the current tracing span.  Returns `ExpiredKey` if the requested
-    /// odd/even key has already been rotated out.
-    fn resolve_rotation_key(
-        &self,
-        rotation: SphinxKeyRotation,
-    ) -> Result<SphinxKeyGuard, PacketProcessingError> {
-        let rotation_label = match rotation {
-            SphinxKeyRotation::Unknown => "unknown",
-            SphinxKeyRotation::OddRotation => "odd",
-            SphinxKeyRotation::EvenRotation => "even",
-        };
-        Span::current().record("key_rotation", rotation_label);
-
-        match rotation {
-            SphinxKeyRotation::Unknown => Ok(self.shared.sphinx_keys.primary()),
-            SphinxKeyRotation::OddRotation => self.shared.sphinx_keys.odd().ok_or_else(|| {
-                warn!(
-                    event = "packet.dropped.expired_key",
-                    key_rotation = "odd",
-                    remote_addr = %self.remote_address,
-                    "dropping packet: odd key rotation expired"
-                );
-                PacketProcessingError::ExpiredKey
-            }),
-            SphinxKeyRotation::EvenRotation => self.shared.sphinx_keys.even().ok_or_else(|| {
-                warn!(
-                    event = "packet.dropped.expired_key",
-                    key_rotation = "even",
-                    remote_addr = %self.remote_address,
-                    "dropping packet: even key rotation expired"
-                );
-                PacketProcessingError::ExpiredKey
-            }),
-        }
-    }
-
-    #[instrument(
-        name = "mixnode.sphinx_partial_unwrap",
-        skip(self, packet),
-        level = "debug",
-        fields(key_rotation, unwrap_result,)
-    )]
-    fn try_partially_unwrap_packet(
-        &self,
-        packet: FramedNymPacket,
-    ) -> Result<PartialyUnwrappedPacketWithKeyRotation, PacketProcessingError> {
-        let rotation = packet.header().key_rotation;
-
-        let result = match rotation {
-            SphinxKeyRotation::Unknown => {
-                // Unknown rotation: try primary, fallback to secondary
-                let primary = self.resolve_rotation_key(rotation)?;
-                let primary_rotation = primary.rotation_id();
-
-                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)
-                        }
-                    }
-                }
-            }
-            _ => {
-                let key = self.resolve_rotation_key(rotation)?;
-                let rotation_id = key.rotation_id();
-                PartiallyUnwrappedPacket::new(packet, key.inner().as_ref())
-                    .map_err(|(_, err)| err)
-                    .map(|p| p.with_key_rotation(rotation_id))
-            }
-        };
-
-        Span::current().record("unwrap_result", if result.is_ok() { "ok" } else { "err" });
-        result
-    }
-
-    async fn handle_received_packet_with_replay_detection(
-        &mut self,
-        now: Instant,
-        packet: FramedNymPacket,
-    ) {
-        // 1. derive and expand shared secret
-        // also check the header integrity
-        let partially_unwrapped = match self.try_partially_unwrap_packet(packet) {
-            Ok(unwrapped) => unwrapped,
-            Err(err) => {
-                trace!("failed to process received mix packet: {err}");
-                warn!(
-                    event = "packet.dropped.malformed",
-                    error = %err,
-                    remote_addr = %self.remote_address,
-                    "dropping malformed packet"
-                );
-                self.shared
-                    .metrics
-                    .mixnet
-                    .ingress_malformed_packet(self.remote_address.ip());
-                return;
-            }
-        };
-
-        self.pending_packets.push(now, partially_unwrapped);
-
-        // 2. check for packet replay
-        // 2.1 first try it without locking
-        if self.handle_pending_packets_batch_no_locking(now).await {
-            return;
-        }
-
-        // 2.2 if we're within deferral threshold, just leave it queued up for another call
-        if self.within_deferral_threshold(now) {
-            return;
-        }
-
-        // 2.3. otherwise block until we obtain the lock and clear the whole batch
-        self.handle_pending_packets_batch(now).await;
-    }
-
-    async fn handle_unwrapped_packet(
-        &self,
-        now: Instant,
-        unwrapped_packet: Result<MixProcessingResult, PacketProcessingError>,
-    ) {
-        // 2. increment our favourite metrics stats
-        self.shared
-            .update_metrics(&unwrapped_packet, self.remote_address.ip());
-
-        // 3. forward the packet to the relevant sink (if enabled)
-        match unwrapped_packet {
-            Err(err) => {
-                trace!("failed to process received mix packet: {err}");
-            }
-            Ok(processed_packet) => match processed_packet.processing_data {
-                MixProcessingResultData::ForwardHop { packet, delay } => {
-                    self.handle_forward_packet(now, packet, delay);
-                }
-                MixProcessingResultData::FinalHop { final_hop_data } => {
-                    self.handle_final_hop(final_hop_data).await;
-                }
-            },
-        }
-    }
-
-    async fn handle_post_replay_detection_packets(
-        &self,
-        now: Instant,
-        packets: HashMap<u32, Vec<PartiallyUnwrappedPacket>>,
-        replay_check_results: HashMap<u32, Vec<bool>>,
-    ) {
-        let mut replays_detected: u64 = 0;
-        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 {
-                    replays_detected += 1;
-                    warn!(
-                        event = "packet.dropped.replay",
-                        remote_addr = %self.remote_address,
-                        rotation_id,
-                        "dropping replayed packet"
-                    );
-                    Err(PacketProcessingError::PacketReplay)
-                } else {
-                    packet.finalise_unwrapping()
-                };
-
-                self.handle_unwrapped_packet(now, unwrapped_packet).await;
-            }
-        }
-        if replays_detected > 0 {
-            debug!(
-                replays_detected,
-                remote_addr = %self.remote_address,
-                "replay detection batch completed with replays"
-            );
-        }
-    }
-
-    async fn handle_pending_packets_batch_no_locking(&mut self, now: Instant) -> bool {
-        let replay_tags = self.pending_packets.replay_tags();
-        if replay_tags.is_empty() {
-            return false;
-        }
-
-        let replay_check_results = match self
-            .shared
-            .replay_protection_filter
-            .batch_try_check_and_set(&replay_tags)
-        {
-            None => return false,
-            Some(Ok(replay_check_results)) => replay_check_results,
-            Some(Err(_)) => {
-                // our mutex got poisoned - we have to shut down
-                error!("CRITICAL FAILURE: replay bloomfilter mutex poisoning!");
-                self.shared.shutdown_token.cancel();
-                return false;
-            }
-        };
-
-        let batch = self.pending_packets.reset(now);
-        self.handle_post_replay_detection_packets(now, batch, replay_check_results)
-            .await;
-        true
-    }
-
-    #[instrument(
-        name = "mixnode.replay_check_batch",
-        skip(self),
-        level = "debug",
-        fields(batch_size, mutex_wait_ms,)
-    )]
-    async fn handle_pending_packets_batch(&mut self, now: Instant) {
-        let replay_tags = self.pending_packets.replay_tags();
-        if replay_tags.is_empty() {
-            return;
-        }
-
-        let batch_size = self.pending_packets.total_count();
-        Span::current().record("batch_size", batch_size as u64);
-
-        let mutex_start = Instant::now();
-        let Ok(replay_check_results) = self
-            .shared
-            .replay_protection_filter
-            .batch_check_and_set(&replay_tags)
-        else {
-            // our mutex got poisoned - we have to shut down
-            error!("CRITICAL FAILURE: replay bloomfilter mutex poisoning!");
-            self.shared.shutdown_token.cancel();
-            return;
-        };
-        Span::current().record("mutex_wait_ms", mutex_start.elapsed().as_millis() as u64);
-
-        let batch = self.pending_packets.reset(now);
-        self.handle_post_replay_detection_packets(now, batch, replay_check_results)
-            .await;
-    }
-
-    #[instrument(
-        name = "mixnode.sphinx_full_unwrap",
-        skip(self, packet),
-        level = "debug",
-        fields(key_rotation)
-    )]
-    fn try_full_unwrap_packet(
-        &self,
-        packet: FramedNymPacket,
-    ) -> Result<MixProcessingResult, PacketProcessingError> {
-        let key = self.resolve_rotation_key(packet.header().key_rotation)?;
-        process_framed_packet(packet, key.inner().as_ref())
-    }
-
-    async fn handle_received_packet_with_no_replay_detection(
-        &mut self,
-        now: Instant,
-        packet: FramedNymPacket,
-    ) {
-        let unwrapped_packet = self.try_full_unwrap_packet(packet);
-        self.handle_unwrapped_packet(now, unwrapped_packet).await;
-    }
-
-    #[instrument(skip(self, packet), level = "debug")]
-    async fn handle_received_nym_packet(&mut self, packet: FramedNymPacket) {
-        let now = Instant::now();
-
-        // 1. attempt to unwrap the packet
-        // if it's a sphinx packet attempt to do pre-processing and replay detection
-        if packet.is_sphinx() && !self.shared.replay_protection_filter.disabled() {
-            self.handle_received_packet_with_replay_detection(now, packet)
-                .await;
-        } else {
-            // otherwise just skip that whole procedure and go straight to payload unwrapping
-            // (assuming the basic framing is valid)
-            self.handle_received_packet_with_no_replay_detection(now, packet)
-                .await;
-        };
-    }
-
-    #[instrument(
-        name = "mixnode.connection",
-        skip(self, socket),
-        level = "debug",
-        fields(
-            remote = %self.remote_address,
-            noise_handshake_ms = tracing::field::Empty,
-        )
-    )]
-    pub(crate) async fn handle_connection(&mut self, socket: TcpStream) {
-        let handshake_start = Instant::now();
-        let noise_stream = match upgrade_noise_responder(socket, &self.shared.noise_config).await {
-            Ok(noise_stream) => noise_stream,
-            Err(err) => {
-                Span::current().record(
-                    "noise_handshake_ms",
-                    handshake_start.elapsed().as_millis() as u64,
-                );
-                warn!(
-                    event = "connection.failed.noise",
-                    remote_addr = %self.remote_address,
-                    error = %err,
-                    "Noise responder handshake failed"
-                );
-                return;
-            }
-        };
-        Span::current().record(
-            "noise_handshake_ms",
-            handshake_start.elapsed().as_millis() as u64,
-        );
-        debug!(
-            "Noise responder handshake completed for {:?}",
-            self.remote_address
-        );
-        self.handle_stream(Framed::new(noise_stream, NymCodec))
-            .await
-    }
-
-    #[instrument(
-        name = "mixnode.stream",
-        skip(self, mixnet_connection),
-        level = "debug",
-        fields(
-            remote = %self.remote_address,
-            packets_processed = 0u64,
-            exit_reason,
-        )
-    )]
-    pub(crate) async fn handle_stream(
-        &mut self,
-        mut mixnet_connection: Framed<Connection<TcpStream>, NymCodec>,
-    ) {
-        let mut packets_processed: u64 = 0;
-        loop {
-            tokio::select! {
-                biased;
-                _ = self.shared.shutdown_token.cancelled() => {
-                    trace!("connection handler: received shutdown");
-                    Span::current().record("exit_reason", "shutdown");
-                    break
-                }
-                maybe_framed_nym_packet = mixnet_connection.next() => {
-                    match maybe_framed_nym_packet {
-                        Some(Ok(packet)) => {
-                            self.handle_received_nym_packet(packet).await;
-                            packets_processed += 1;
-                            if packets_processed.is_multiple_of(SPAN_UPDATE_INTERVAL) {
-                                Span::current().record("packets_processed", packets_processed);
-                            }
-                        }
-                        Some(Err(err)) => {
-                            warn!(
-                                event = "connection.corrupted",
-                                remote_addr = %self.remote_address,
-                                error = %err,
-                                packets_processed,
-                                "connection stream corrupted"
-                            );
-                            Span::current().record("exit_reason", "corrupted");
-                            Span::current().record("packets_processed", packets_processed);
-                            return
-                        }
-                        None => {
-                            debug!(
-                                remote_addr = %self.remote_address,
-                                packets_processed,
-                                "connection closed by remote"
-                            );
-                            Span::current().record("exit_reason", "closed_by_remote");
-                            Span::current().record("packets_processed", packets_processed);
-                            return
-                        }
-                    }
-                }
-            }
-        }
-
-        Span::current().record("packets_processed", packets_processed);
-        debug!("exiting and closing connection");
-    }
-}

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

@@ -2,20 +2,40 @@
 // SPDX-License-Identifier: GPL-3.0-only
 
 use crate::node::mixnet::SharedData;
+use crate::node::mixnet::packet_ingest::{IngressNymPacket, MixIngestSender};
+use futures::StreamExt;
+use nym_noise::connection::Connection;
+use nym_noise::upgrade_noise_responder;
+use nym_sphinx_framing::codec::NymCodec;
 use nym_task::ShutdownToken;
+use tokio::net::TcpStream;
+use tokio::task::JoinHandle;
+use tokio_util::codec::Framed;
+use tracing::{Span, debug, error, info, instrument, trace, warn};
+
+use std::io;
 use std::net::SocketAddr;
-use tracing::{debug, error, info, trace};
+use std::time::Instant;
+
+/// How often (in packets) the stream-level span updates its packet count.
+const SPAN_UPDATE_INTERVAL: u64 = 10_000;
 
 pub(crate) struct Listener {
     bind_address: SocketAddr,
     shared_data: SharedData,
+    ingest_sender: MixIngestSender,
 }
 
 impl Listener {
-    pub(crate) fn new(bind_address: SocketAddr, shared_data: SharedData) -> Self {
+    pub(crate) fn new(
+        bind_address: SocketAddr,
+        shared_data: SharedData,
+        ingest_sender: MixIngestSender,
+    ) -> Self {
         Listener {
             bind_address,
             shared_data,
+            ingest_sender,
         }
     }
 
@@ -42,10 +62,181 @@ impl Listener {
                     break
                 }
                 connection = tcp_listener.accept() => {
-                    self.shared_data.try_handle_connection(connection);
+                    self.try_handle_connection(connection);
                 }
             }
         }
         debug!("mixnet socket listener: Exiting");
     }
+
+    fn try_handle_connection(
+        &self,
+        accepted: io::Result<(TcpStream, SocketAddr)>,
+    ) -> Option<JoinHandle<()>> {
+        match accepted {
+            Ok((socket, remote_addr)) => {
+                debug!("accepted incoming mixnet connection from: {remote_addr}");
+
+                let conn_handler = ConnectionHandler::new(
+                    &self.shared_data,
+                    remote_addr,
+                    self.ingest_sender.clone(),
+                );
+                let join_handle =
+                    tokio::spawn(async move { conn_handler.handle_connection(socket).await });
+                self.shared_data.log_connected_clients();
+                Some(join_handle)
+            }
+            Err(err) => {
+                debug!("failed to accept incoming mixnet connection: {err}");
+                None
+            }
+        }
+    }
+}
+
+struct ConnectionHandler {
+    remote_addr: SocketAddr,
+    shared_data: SharedData,
+    ingest_sender: MixIngestSender,
+}
+
+impl ConnectionHandler {
+    fn new(shared: &SharedData, remote_addr: SocketAddr, ingest_sender: MixIngestSender) -> Self {
+        Self {
+            ingest_sender,
+            remote_addr,
+            shared_data: SharedData {
+                processing_config: shared.processing_config,
+                sphinx_keys: shared.sphinx_keys.clone(),
+                replay_protection_filter: shared.replay_protection_filter.clone(),
+                mixnet_forwarder: shared.mixnet_forwarder.clone(),
+                final_hop: shared.final_hop.clone(),
+                noise_config: shared.noise_config.clone(),
+                metrics: shared.metrics.clone(),
+                shutdown_token: shared.shutdown_token.child_token(),
+            },
+        }
+    }
+
+    #[instrument(
+        name = "mixnode.connection",
+        skip(self, socket),
+        level = "debug",
+        fields(
+            remote = %self.remote_addr,
+            noise_handshake_ms = tracing::field::Empty,
+        )
+    )]
+    pub(crate) async fn handle_connection(&self, socket: TcpStream) {
+        let handshake_start = Instant::now();
+        let noise_stream =
+            match upgrade_noise_responder(socket, &self.shared_data.noise_config).await {
+                Ok(noise_stream) => noise_stream,
+                Err(err) => {
+                    Span::current().record(
+                        "noise_handshake_ms",
+                        handshake_start.elapsed().as_millis() as u64,
+                    );
+                    warn!(
+                        event = "connection.failed.noise",
+                        remote_addr = %self.remote_addr,
+                        error = %err,
+                        "Noise responder handshake failed"
+                    );
+                    return;
+                }
+            };
+        Span::current().record(
+            "noise_handshake_ms",
+            handshake_start.elapsed().as_millis() as u64,
+        );
+        debug!(
+            "Noise responder handshake completed for {:?}",
+            self.remote_addr
+        );
+        self.handle_stream(Framed::new(noise_stream, NymCodec))
+            .await
+    }
+
+    #[instrument(
+        name = "mixnode.stream",
+        skip(self, mixnet_connection),
+        level = "debug",
+        fields(
+            remote = %self.remote_addr,
+            packets_processed = 0u64,
+            exit_reason,
+        )
+    )]
+    pub(crate) async fn handle_stream(
+        &self,
+        mut mixnet_connection: Framed<Connection<TcpStream>, NymCodec>,
+    ) {
+        let mut packets_processed: u64 = 0;
+        loop {
+            tokio::select! {
+                biased;
+                _ = self.shared_data.shutdown_token.cancelled() => {
+                    trace!("connection handler: received shutdown");
+                    Span::current().record("exit_reason", "shutdown");
+                    break
+                }
+                maybe_framed_nym_packet = mixnet_connection.next() => {
+                    match maybe_framed_nym_packet {
+                        Some(Ok(packet)) => {
+                            let ingress_packet = IngressNymPacket::new(packet, Instant::now(), self.remote_addr);
+                            self.handle_received_nym_packet(ingress_packet);
+                            packets_processed += 1;
+                            if packets_processed.is_multiple_of(SPAN_UPDATE_INTERVAL) {
+                                Span::current().record("packets_processed", packets_processed);
+                            }
+                        }
+                        Some(Err(err)) => {
+                            warn!(
+                                event = "connection.corrupted",
+                                remote_addr = %self.remote_addr,
+                                error = %err,
+                                packets_processed,
+                                "connection stream corrupted"
+                            );
+                            Span::current().record("exit_reason", "corrupted");
+                            Span::current().record("packets_processed", packets_processed);
+                            return
+                        }
+                        None => {
+                            debug!(
+                                remote_addr = %self.remote_addr,
+                                packets_processed,
+                                "connection closed by remote"
+                            );
+                            Span::current().record("exit_reason", "closed_by_remote");
+                            Span::current().record("packets_processed", packets_processed);
+                            return
+                        }
+                    }
+                }
+            }
+        }
+
+        Span::current().record("packets_processed", packets_processed);
+        debug!("exiting and closing connection");
+    }
+
+    /// Attempt to add the packet to the processing queue. If there is no capacity available the packet will
+    /// be dropped and the `mixnet_ingress_overflow_packets_dropped` metric will be incremented.
+    fn handle_received_nym_packet(&self, packet: IngressNymPacket) {
+        match self.ingest_sender.ingest_packet(packet) {
+            Ok(_) => {}
+            Err(err) if err.kind() == io::ErrorKind::WouldBlock => self
+                .shared_data
+                .metrics
+                .mixnet
+                .ingress_dropped_overflow_packet(),
+            Err(err) => {
+                error!("unexpected error using ingress channel - shutting down: {err}");
+                self.shared_data.shutdown_token.cancel();
+            }
+        }
+    }
 }

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

@@ -1,9 +1,9 @@
 // Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: GPL-3.0-only
 
-pub(crate) mod handler;
 pub(crate) mod listener;
 pub(crate) mod packet_forwarding;
+pub(crate) mod packet_ingest;
 pub(crate) mod shared;
 
 pub(crate) use listener::Listener;

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

@@ -0,0 +1,560 @@
+//! Mix Packet Ingest Worker implementation and tooling
+
+use std::net::SocketAddr;
+use std::time::Instant;
+use std::{io, num::NonZero};
+
+use crate::node::{
+    key_rotation::active_keys::SphinxKeyGuard,
+    mixnet::{SharedData, shared::ProcessingConfig},
+};
+use nym_node_metrics::mixnet::PacketKind;
+use nym_sphinx_forwarding::packet::MixPacket;
+use nym_sphinx_framing::packet::FramedNymPacket;
+use nym_sphinx_framing::processing::{
+    MixPacketVersion, MixProcessingResult, MixProcessingResultData, PacketProcessingError,
+    PartiallyUnwrappedPacket, PartiallyUnwrappedPacketWithKeyRotation, ProcessedFinalHop,
+    process_framed_packet,
+};
+use nym_sphinx_params::SphinxKeyRotation;
+use nym_sphinx_types::Delay;
+use nym_task::ShutdownToken;
+use tracing::{Span, debug, error, info, instrument, trace, warn};
+
+pub(crate) struct IngressNymPacket {
+    packet: FramedNymPacket,
+    received_at: Instant,
+    received_from: SocketAddr,
+}
+
+impl IngressNymPacket {
+    pub(crate) fn new(
+        packet: FramedNymPacket,
+        received_at: Instant,
+        received_from: SocketAddr,
+    ) -> Self {
+        Self {
+            packet,
+            received_at,
+            received_from,
+        }
+    }
+}
+
+pub struct MixPacketIngest {
+    shared: SharedData,
+
+    // Keep a copy of the channel to prevent accidental exit
+    packet_sender: MixIngestSender,
+    packet_receiver: MixIngestReceiver,
+}
+
+impl MixPacketIngest {
+    pub fn new(shared: &SharedData) -> Self {
+        let (packet_sender, packet_receiver) = mix_ingest_channels(&shared.processing_config);
+
+        Self {
+            shared: SharedData {
+                processing_config: shared.processing_config,
+                sphinx_keys: shared.sphinx_keys.clone(),
+                replay_protection_filter: shared.replay_protection_filter.clone(),
+                mixnet_forwarder: shared.mixnet_forwarder.clone(),
+                final_hop: shared.final_hop.clone(),
+                noise_config: shared.noise_config.clone(),
+                metrics: shared.metrics.clone(),
+                shutdown_token: shared.shutdown_token.child_token(),
+            },
+
+            packet_sender,
+            packet_receiver,
+        }
+    }
+
+    pub fn sender(&self) -> MixIngestSender {
+        self.packet_sender.clone()
+    }
+
+    pub async fn run(&mut self, shutdown_token: ShutdownToken) {
+        // this one is impossible to ever panic
+        #[allow(clippy::unwrap_used)]
+        let default_worker_count = NonZero::new(1).unwrap();
+        let num_threads = std::thread::available_parallelism()
+            .inspect_err(|e| warn!("unable to query available parallelism: {e}"))
+            .unwrap_or(default_worker_count)
+            .get();
+
+        let mut workers = tokio::task::JoinSet::new();
+
+        for _ in 0..num_threads {
+            let recv = self.packet_receiver.clone();
+            let worker = MixPacketIngestWorker {
+                packet_receiver: recv,
+                shared: SharedData {
+                    processing_config: self.shared.processing_config,
+                    sphinx_keys: self.shared.sphinx_keys.clone(),
+                    replay_protection_filter: self.shared.replay_protection_filter.clone(),
+                    mixnet_forwarder: self.shared.mixnet_forwarder.clone(),
+                    final_hop: self.shared.final_hop.clone(),
+                    noise_config: self.shared.noise_config.clone(),
+                    metrics: self.shared.metrics.clone(),
+                    shutdown_token: self.shared.shutdown_token.child_token(),
+                },
+            };
+
+            let worker_token = shutdown_token.child_token();
+            workers.spawn(worker.run(worker_token));
+        }
+        info!("launched {num_threads} mix ingest workers");
+
+        while let Some(res) = workers.join_next().await {
+            if let Err(err) = res {
+                warn!("ingest worker closed with error: {err}");
+            } else if !shutdown_token.is_cancelled() {
+                warn!("ingest worker closed unexpectedly!");
+            }
+        }
+    }
+}
+
+struct MixPacketIngestWorker {
+    shared: SharedData,
+
+    packet_receiver: MixIngestReceiver,
+}
+
+impl MixPacketIngestWorker {
+    pub async fn run(mut self, shutdown_token: ShutdownToken) {
+        trace!("starting PacketIngest");
+        loop {
+            tokio::select! {
+                biased;
+                _ = shutdown_token.cancelled() => {
+                    debug!("PacketIngest: Received shutdown");
+                    break;
+                }
+                new_packet = self.packet_receiver.recv_async() => {
+                    // this one is impossible to ever panic - the parent struct maintains a sender
+                    // and waits for this process to end. Therefore it can't happen that ALL senders
+                    // are dropped
+                    #[allow(clippy::expect_used)]
+                    let new_packet = new_packet.expect("the ingest receiver closed somehow");
+                    self.handle_ingest_packet(new_packet).await;
+                }
+            }
+        }
+    }
+
+    async fn handle_ingest_packet(&mut self, packet: IngressNymPacket) {
+        // 1. attempt to unwrap the packet
+        // if it's a sphinx packet attempt to do pre-processing and replay detection
+        if packet.packet.is_sphinx() && !self.shared.replay_protection_filter.disabled() {
+            self.handle_received_packet_with_replay_detection(packet)
+                .await;
+        } else {
+            // otherwise just skip that whole procedure and go straight to payload unwrapping
+            // (assuming the basic framing is valid)
+            self.handle_received_packet_with_no_replay_detection(packet)
+                .await;
+        };
+    }
+
+    async fn handle_received_packet_with_replay_detection(&mut self, packet: IngressNymPacket) {
+        let source = packet.received_from;
+        let received_at = packet.received_at;
+
+        // 1. derive and expand shared secret
+        // also check the header integrity
+        let partially_unwrapped = match self.try_partially_unwrap_packet(packet) {
+            Ok(unwrapped) => unwrapped,
+            Err(err) => {
+                trace!("failed to process received mix packet: {err}");
+                warn!(
+                    event = "packet.dropped.malformed",
+                    error = %err,
+                    remote_addr = %source,
+                    "dropping malformed packet"
+                );
+                self.shared
+                    .metrics
+                    .mixnet
+                    .ingress_malformed_packet(source.ip());
+                return;
+            }
+        };
+
+        let rotation_id = partially_unwrapped.used_key_rotation;
+        let Some(replay_tag) = partially_unwrapped.packet.replay_tag() else {
+            error!("corrupted packet - replay tag was missing");
+            return;
+        };
+
+        let mutex_start = Instant::now();
+        let Ok(replayed) = self
+            .shared
+            .replay_protection_filter
+            .check_and_set(rotation_id, replay_tag)
+        else {
+            // our mutex got poisoned - we have to shut down
+            error!("CRITICAL FAILURE: replay bloomfilter mutex poisoning!");
+            self.shared.shutdown_token.cancel();
+            return;
+        };
+        Span::current().record("mutex_wait_ms", mutex_start.elapsed().as_millis() as u64);
+
+        let unwrapped_packet = if replayed {
+            warn!(
+                event = "packet.dropped.replay",
+                remote_addr = %source,
+                rotation_id,
+                "dropping replayed packet"
+            );
+            Err(PacketProcessingError::PacketReplay)
+        } else {
+            partially_unwrapped.packet.finalise_unwrapping()
+        };
+
+        self.handle_unwrapped_packet(unwrapped_packet, source, received_at)
+            .await;
+    }
+
+    async fn handle_received_packet_with_no_replay_detection(&mut self, packet: IngressNymPacket) {
+        let source = packet.received_from;
+        let received_at = packet.received_at;
+        let unwrapped_packet = self.try_full_unwrap_packet(packet);
+        self.handle_unwrapped_packet(unwrapped_packet, source, received_at)
+            .await;
+    }
+
+    #[instrument(
+        name = "mixnode.sphinx_full_unwrap",
+        skip(self, packet),
+        level = "debug",
+        fields(key_rotation)
+    )]
+    fn try_full_unwrap_packet(
+        &self,
+        packet: IngressNymPacket,
+    ) -> Result<MixProcessingResult, PacketProcessingError> {
+        let key = self.resolve_rotation_key(packet.packet.header().key_rotation)?;
+        process_framed_packet(packet.packet, key.inner().as_ref())
+    }
+
+    #[instrument(
+        name = "mixnode.sphinx_partial_unwrap",
+        skip(self, packet),
+        level = "debug",
+        fields(key_rotation, unwrap_result,)
+    )]
+    fn try_partially_unwrap_packet(
+        &self,
+        packet: IngressNymPacket,
+    ) -> Result<PartiallyUnwrappedPacketWithKeyRotation, PacketProcessingError> {
+        let rotation = packet.packet.header().key_rotation;
+
+        let result = match rotation {
+            SphinxKeyRotation::Unknown => {
+                // Unknown rotation: try primary, fallback to secondary
+                let primary = self.resolve_rotation_key(rotation)?;
+                let primary_rotation = primary.rotation_id();
+
+                match PartiallyUnwrappedPacket::new(packet.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)
+                        }
+                    }
+                }
+            }
+            _ => {
+                let key = self.resolve_rotation_key(rotation)?;
+                let rotation_id = key.rotation_id();
+                PartiallyUnwrappedPacket::new(packet.packet, key.inner().as_ref())
+                    .map_err(|(_, err)| err)
+                    .map(|p| p.with_key_rotation(rotation_id))
+            }
+        };
+
+        Span::current().record("unwrap_result", if result.is_ok() { "ok" } else { "err" });
+        result
+    }
+
+    async fn handle_unwrapped_packet(
+        &self,
+        unwrapped_packet: Result<MixProcessingResult, PacketProcessingError>,
+        source: SocketAddr,
+        received_at: Instant,
+    ) {
+        // 2. increment our favourite metrics stats
+        self.update_metrics(&unwrapped_packet, source);
+
+        // 3. forward the packet to the relevant sink (if enabled)
+        match unwrapped_packet {
+            Err(err) => {
+                trace!("failed to process received mix packet: {err}");
+            }
+            Ok(processed_packet) => match processed_packet.processing_data {
+                MixProcessingResultData::ForwardHop { packet, delay } => {
+                    self.handle_forward_packet(packet, received_at, source, delay);
+                }
+                MixProcessingResultData::FinalHop { final_hop_data } => {
+                    self.handle_final_hop(final_hop_data, source).await;
+                }
+            },
+        }
+    }
+
+    fn update_metrics(
+        &self,
+        processing_result: &Result<MixProcessingResult, PacketProcessingError>,
+        source: SocketAddr,
+    ) {
+        let Ok(processing_result) = processing_result else {
+            self.shared
+                .metrics
+                .mixnet
+                .ingress_malformed_packet(source.ip());
+            return;
+        };
+
+        let packet_version = convert_to_metrics_version(processing_result.packet_version);
+
+        match processing_result.processing_data {
+            MixProcessingResultData::ForwardHop { delay, .. } => {
+                self.shared
+                    .metrics
+                    .mixnet
+                    .ingress_received_forward_packet(source.ip(), packet_version);
+
+                // check if the delay wasn't excessive
+                if let Some(delay) = delay
+                    && delay.to_duration() > self.shared.processing_config.maximum_packet_delay
+                {
+                    self.shared.metrics.mixnet.ingress_excessive_delay_packet()
+                }
+            }
+            MixProcessingResultData::FinalHop { .. } => {
+                self.shared
+                    .metrics
+                    .mixnet
+                    .ingress_received_final_hop_packet(source.ip(), packet_version);
+            }
+        }
+    }
+
+    /// Resolve the sphinx key for the given rotation, recording the rotation
+    /// label on the current tracing span.  Returns `ExpiredKey` if the requested
+    /// odd/even key has already been rotated out.
+    fn resolve_rotation_key(
+        &self,
+        rotation: SphinxKeyRotation,
+    ) -> Result<SphinxKeyGuard, PacketProcessingError> {
+        let rotation_label = match rotation {
+            SphinxKeyRotation::Unknown => "unknown",
+            SphinxKeyRotation::OddRotation => "odd",
+            SphinxKeyRotation::EvenRotation => "even",
+        };
+        Span::current().record("key_rotation", rotation_label);
+
+        match rotation {
+            SphinxKeyRotation::Unknown => Ok(self.shared.sphinx_keys.primary()),
+            SphinxKeyRotation::OddRotation => self.shared.sphinx_keys.odd().ok_or_else(|| {
+                warn!(
+                    event = "packet.dropped.expired_key",
+                    key_rotation = "odd",
+                    "dropping packet: odd key rotation expired"
+                );
+                PacketProcessingError::ExpiredKey
+            }),
+            SphinxKeyRotation::EvenRotation => self.shared.sphinx_keys.even().ok_or_else(|| {
+                warn!(
+                    event = "packet.dropped.expired_key",
+                    key_rotation = "even",
+                    "dropping packet: even key rotation expired"
+                );
+                PacketProcessingError::ExpiredKey
+            }),
+        }
+    }
+
+    #[instrument(
+        name = "mixnode.forward_packet",
+        skip(self, mix_packet, delay),
+        level = "debug",
+        fields(
+            remote_addr = %remote_addr,
+            delay_ms = tracing::field::Empty,
+        )
+    )]
+    fn handle_forward_packet(
+        &self,
+        mix_packet: MixPacket,
+        received_at: Instant,
+        remote_addr: SocketAddr,
+        delay: Option<Delay>,
+    ) {
+        if !self.shared.processing_config.forward_hop_processing_enabled {
+            warn!(
+                event = "packet.dropped.forward_disabled",
+                remote_addr = %remote_addr,
+                "dropping packet: forward hop processing disabled"
+            );
+            self.shared.dropped_forward_packet(remote_addr.ip());
+            return;
+        }
+
+        let forward_instant = self.create_delay_target(received_at, delay);
+        if let Some(target) = forward_instant {
+            Span::current().record(
+                "delay_ms",
+                target.saturating_duration_since(received_at).as_millis() as u64,
+            );
+        }
+        self.shared.forward_mix_packet(mix_packet, forward_instant);
+    }
+
+    /// Determine instant at which packet should get forwarded to the next hop.
+    /// By using [`Instant`] rather than explicit [`Duration`] we minimise effects of
+    /// the skew caused by being stuck in the channel queue.
+    /// This method also clamps the maximum allowed delay so that nobody could send a bunch of packets
+    /// with, for example, delays of 1 year thus causing denial of service
+    fn create_delay_target(&self, received_at: Instant, delay: Option<Delay>) -> Option<Instant> {
+        let delay = delay?.to_duration();
+
+        let delay = if delay > self.shared.processing_config.maximum_packet_delay {
+            self.shared.processing_config.maximum_packet_delay
+        } else {
+            delay
+        };
+        trace!(
+            "received packet will be delayed for {}ms",
+            delay.as_millis()
+        );
+
+        Some(received_at + delay)
+    }
+
+    #[instrument(
+        name = "mixnode.final_hop",
+        skip(self, final_hop_data),
+        level = "debug",
+        fields(
+            remote_addr = %remote_addr,
+            client_online,
+            disk_fallback = false,
+            ack_forwarded = false,
+        )
+    )]
+    async fn handle_final_hop(&self, final_hop_data: ProcessedFinalHop, remote_addr: SocketAddr) {
+        if !self.shared.processing_config.final_hop_processing_enabled {
+            warn!(
+                event = "packet.dropped.final_hop_disabled",
+                remote_addr = %remote_addr,
+                "dropping packet: final hop processing disabled"
+            );
+            self.shared.dropped_final_hop_packet(remote_addr.ip());
+            return;
+        }
+
+        let client = final_hop_data.destination;
+        let message = final_hop_data.message;
+        let has_ack = final_hop_data.forward_ack.is_some();
+
+        // if possible attempt to push message directly to the client
+        match self.shared.try_push_message_to_client(client, message) {
+            Err(unsent_plaintext) => {
+                // if that failed, store it on disk
+                Span::current().record("client_online", false);
+                match self
+                    .shared
+                    .store_processed_packet_payload(client, unsent_plaintext)
+                    .await
+                {
+                    Err(err) => error!("Failed to store client data - {err}"),
+                    Ok(_) => {
+                        Span::current().record("disk_fallback", true);
+                        self.shared
+                            .metrics
+                            .mixnet
+                            .egress
+                            .add_disk_persisted_packet();
+                        trace!("Stored packet for {client}")
+                    }
+                }
+            }
+            Ok(_) => {
+                Span::current().record("client_online", true);
+                trace!("Pushed received packet to {client}");
+            }
+        }
+
+        // if we managed to either push message directly to the [online] client or store it at
+        // disk, forward the ack
+        self.shared.forward_ack_packet(final_hop_data.forward_ack);
+        if has_ack {
+            Span::current().record("ack_forwarded", true);
+        }
+    }
+}
+
+fn convert_to_metrics_version(processed: MixPacketVersion) -> PacketKind {
+    match processed {
+        MixPacketVersion::Outfox => PacketKind::Outfox,
+        MixPacketVersion::Sphinx(sphinx_version) => PacketKind::Sphinx(sphinx_version.value()),
+    }
+}
+
+pub fn mix_ingest_channels(
+    processing_config: &ProcessingConfig,
+) -> (MixIngestSender, MixIngestReceiver) {
+    let (tx, rx) = flume::bounded(processing_config.ingress_channel_maximum_capacity);
+    (MixIngestSender(tx), rx)
+}
+
+#[derive(Clone)]
+pub struct MixIngestSender(flume::Sender<IngressNymPacket>);
+
+impl MixIngestSender {
+    pub fn ingest_packet(&self, packet: impl Into<IngressNymPacket>) -> io::Result<()> {
+        let sender = &self.0;
+
+        // we are using a bounded channel so unwrap is safe
+        #[allow(clippy::unwrap_used)]
+        let channel_capacity = sender.capacity().unwrap();
+        let channel_used = sender.len();
+
+        let sending_res = sender.try_send(packet.into());
+
+        sending_res.map_err(|err| match err {
+            flume::TrySendError::Full(_) => {
+                warn!(
+                    event = "mixnode.ingress_try_send",
+                    result = "full_dropped",
+                    channel_capacity,
+                    channel_used,
+                    "dropping packet: ingress buffer is full ({channel_used}/{channel_capacity})"
+                );
+                io::Error::new(io::ErrorKind::WouldBlock, "ingress queue is full")
+            }
+            flume::TrySendError::Disconnected(_) => {
+                debug!(
+                    event = "mixnode.ingress_try_send",
+                    result = "closed",
+                    channel_capacity,
+                    channel_used,
+                    "ingress queue is closed"
+                );
+                io::Error::new(io::ErrorKind::BrokenPipe, "ingress packet channel closed")
+            }
+        })
+    }
+}
+
+pub type MixIngestReceiver = flume::Receiver<IngressNymPacket>;

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

@@ -4,25 +4,16 @@
 use crate::config::Config;
 use crate::node::key_rotation::active_keys::ActiveSphinxKeys;
 use crate::node::mixnet::SharedFinalHopData;
-use crate::node::mixnet::handler::ConnectionHandler;
 use crate::node::replay_protection::bloomfilter::ReplayProtectionBloomfilters;
 use nym_gateway::node::GatewayStorageError;
 use nym_mixnet_client::forwarder::{MixForwardingSender, PacketToForward};
 use nym_node_metrics::NymNodeMetrics;
-use nym_node_metrics::mixnet::PacketKind;
 use nym_noise::config::NoiseConfig;
 use nym_sphinx_forwarding::packet::MixPacket;
-use nym_sphinx_framing::processing::{
-    MixPacketVersion, MixProcessingResult, MixProcessingResultData, PacketProcessingError,
-};
 use nym_sphinx_types::DestinationAddressBytes;
 use nym_task::ShutdownToken;
-use std::io;
-use std::net::{IpAddr, SocketAddr};
-use std::time::Duration;
-use tokio::net::TcpStream;
-use tokio::task::JoinHandle;
-use tokio::time::Instant;
+use std::net::IpAddr;
+use std::time::{Duration, Instant};
 use tracing::{debug, error};
 
 pub(crate) mod final_hop;
@@ -30,13 +21,11 @@ pub(crate) mod final_hop;
 #[derive(Clone, Copy)]
 pub(crate) struct ProcessingConfig {
     pub(crate) maximum_packet_delay: Duration,
-    /// how long the task is willing to skip mutex acquisition before it will block the thread
-    /// until it actually obtains it
-    pub(crate) maximum_replay_detection_deferral: Duration,
 
-    /// how many packets the task is willing to queue before it will block the thread
-    /// until it obtains the mutex
-    pub(crate) maximum_replay_detection_pending_packets: usize,
+    /// Channel capacity for the mixnet ingress channel. This determines the maximum number of
+    /// packets that can be queued waiting for ingest processing. Once the queue is full packets
+    /// will still be taken off the wire, but dropped as the node is too busy to handle them.
+    pub(crate) ingress_channel_maximum_capacity: usize,
 
     pub(crate) forward_hop_processing_enabled: bool,
     pub(crate) final_hop_processing_enabled: bool,
@@ -46,16 +35,11 @@ impl ProcessingConfig {
     pub(crate) fn new(config: &Config) -> Self {
         ProcessingConfig {
             maximum_packet_delay: config.mixnet.debug.maximum_forward_packet_delay,
-            maximum_replay_detection_deferral: config
+            ingress_channel_maximum_capacity: config
                 .mixnet
                 .replay_protection
                 .debug
-                .maximum_replay_detection_deferral,
-            maximum_replay_detection_pending_packets: config
-                .mixnet
-                .replay_protection
-                .debug
-                .maximum_replay_detection_pending_packets,
+                .ingress_channel_maximum_capacity,
             forward_hop_processing_enabled: config.modes.mixnode,
             final_hop_processing_enabled: config.modes.expects_final_hop_traffic()
                 || config.wireguard.enabled,
@@ -83,13 +67,6 @@ pub(crate) struct SharedData {
     pub(super) shutdown_token: ShutdownToken,
 }
 
-fn convert_to_metrics_version(processed: MixPacketVersion) -> PacketKind {
-    match processed {
-        MixPacketVersion::Outfox => PacketKind::Outfox,
-        MixPacketVersion::Sphinx(sphinx_version) => PacketKind::Sphinx(sphinx_version.value()),
-    }
-}
-
 impl SharedData {
     #[allow(clippy::too_many_arguments)]
     pub(crate) fn new(
@@ -131,64 +108,11 @@ impl SharedData {
         self.metrics.mixnet.ingress_dropped_final_hop_packet(source)
     }
 
-    pub(super) fn update_metrics(
-        &self,
-        processing_result: &Result<MixProcessingResult, PacketProcessingError>,
-        source: IpAddr,
-    ) {
-        let Ok(processing_result) = processing_result else {
-            self.metrics.mixnet.ingress_malformed_packet(source);
-            return;
-        };
-
-        let packet_version = convert_to_metrics_version(processing_result.packet_version);
-
-        match processing_result.processing_data {
-            MixProcessingResultData::ForwardHop { delay, .. } => {
-                self.metrics
-                    .mixnet
-                    .ingress_received_forward_packet(source, packet_version);
-
-                // check if the delay wasn't excessive
-                if let Some(delay) = delay
-                    && delay.to_duration() > self.processing_config.maximum_packet_delay
-                {
-                    self.metrics.mixnet.ingress_excessive_delay_packet()
-                }
-            }
-            MixProcessingResultData::FinalHop { .. } => {
-                self.metrics
-                    .mixnet
-                    .ingress_received_final_hop_packet(source, packet_version);
-            }
-        }
-    }
-
-    pub(super) fn try_handle_connection(
-        &self,
-        accepted: io::Result<(TcpStream, SocketAddr)>,
-    ) -> Option<JoinHandle<()>> {
-        match accepted {
-            Ok((socket, remote_addr)) => {
-                debug!("accepted incoming mixnet connection from: {remote_addr}");
-                let mut handler = ConnectionHandler::new(self, remote_addr);
-                let join_handle =
-                    tokio::spawn(async move { handler.handle_connection(socket).await });
-                self.log_connected_clients();
-                Some(join_handle)
-            }
-            Err(err) => {
-                debug!("failed to accept incoming mixnet connection: {err}");
-                None
-            }
-        }
-    }
-
     pub(super) fn forward_mix_packet(&self, packet: MixPacket, delay_until: Option<Instant>) {
         let has_delay = delay_until.is_some();
         if self
             .mixnet_forwarder
-            .forward_packet(PacketToForward::new(packet, delay_until))
+            .forward_packet(PacketToForward::new(packet, delay_until.map(Into::into)))
             .is_err()
             && !self.shutdown_token.is_cancelled()
         {

nym-node/src/node/mod.rs

@@ -22,7 +22,7 @@ use crate::node::http::{HttpServerConfig, NymNodeHttpServer, NymNodeRouter};
 use crate::node::key_rotation::active_keys::ActiveSphinxKeys;
 use crate::node::key_rotation::controller::KeyRotationController;
 use crate::node::key_rotation::manager::SphinxKeyManager;
-use crate::node::lp::{LpHandlerState, LpListener};
+use crate::node::lp::LpSetup;
 use crate::node::metrics::aggregator::MetricsAggregator;
 use crate::node::metrics::console_logger::ConsoleLogger;
 use crate::node::metrics::handler::client_sessions::GatewaySessionStatsHandler;
@@ -32,6 +32,7 @@ use crate::node::metrics::handler::mixnet_data_cleaner::MixnetMetricsCleaner;
 use crate::node::metrics::handler::pending_egress_packets_updater::PendingEgressPacketsUpdater;
 use crate::node::mixnet::SharedFinalHopData;
 use crate::node::mixnet::packet_forwarding::PacketForwarder;
+use crate::node::mixnet::packet_ingest::MixPacketIngest;
 use crate::node::mixnet::shared::ProcessingConfig;
 use crate::node::nym_apis_client::NymApisClient;
 use crate::node::replay_protection::background_task::ReplayProtectionDiskFlush;
@@ -80,7 +81,7 @@ use std::net::SocketAddr;
 use std::ops::Deref;
 use std::path::Path;
 use std::sync::Arc;
-use tokio::sync::{Semaphore, mpsc};
+use tokio::sync::mpsc;
 use tracing::{debug, info, trace};
 use zeroize::Zeroizing;
 
@@ -487,28 +488,23 @@ impl NymNode {
         config.save()
     }
 
-    pub async fn build_lp_listener(
-        &mut self,
+    pub async fn build_lp_tasks(
+        &self,
         peer_registrator: Option<PeerRegistrator>,
         mix_packet_sender: MixForwardingSender,
-    ) -> Result<LpListener, NymNodeError> {
-        let handler_state = LpHandlerState {
-            local_lp_peer: LpLocalPeer::new(Ciphersuite::default(), self.x25519_lp_keys.clone())
-                .with_kem_keys(self.psq_kem_keys.clone()),
-            metrics: self.metrics.clone(),
-            peer_registrator,
-            lp_config: self.config.lp,
-            outbound_mix_sender: mix_packet_sender,
-            session_states: Arc::new(dashmap::DashMap::new()),
-            forward_semaphore: Arc::new(Semaphore::new(
-                self.config.lp.debug.max_concurrent_forwards,
-            )),
-        };
+    ) -> Result<LpSetup, NymNodeError> {
+        let lp_peer = LpLocalPeer::new(Ciphersuite::default(), self.x25519_lp_keys.clone())
+            .with_kem_keys(self.psq_kem_keys.clone());
 
-        Ok(LpListener::new(
-            handler_state,
+        LpSetup::new(
+            lp_peer,
+            self.config.lp,
+            self.metrics.clone(),
+            peer_registrator,
+            mix_packet_sender,
             self.shutdown_manager.shutdown_tracker().clone(),
-        ))
+        )
+        .await
     }
 
     pub(crate) async fn new(config: Config) -> Result<Self, NymNodeError> {
@@ -770,11 +766,10 @@ impl NymNode {
                 "starting the LP listener on {} (data handler on: {})",
                 self.config.lp.control_bind_address, self.config.lp.data_bind_address,
             );
-            let mut lp_listener = self
-                .build_lp_listener(wg_peer_registrator.clone(), mix_packet_sender)
+            let lp_tasks = self
+                .build_lp_tasks(wg_peer_registrator.clone(), mix_packet_sender)
                 .await?;
-            self.shutdown_tracker()
-                .try_spawn_named(async move { lp_listener.run().await }, "LpListener");
+            lp_tasks.start_tasks();
         } else {
             info!("node not running in entry mode: the websocket and LP will remain closed");
         }
@@ -1295,7 +1290,16 @@ impl NymNode {
             self.shutdown_token(),
         );
 
-        let mut mixnet_listener = mixnet::Listener::new(self.config.mixnet.bind_address, shared);
+        let mut mix_packet_ingest = MixPacketIngest::new(&shared);
+        let ingest_sender = mix_packet_ingest.sender();
+        let ingest_shutdown = self.shutdown_token();
+        self.shutdown_tracker().try_spawn_named(
+            async move { mix_packet_ingest.run(ingest_shutdown).await },
+            "MixPacketIngest",
+        );
+
+        let mut mixnet_listener =
+            mixnet::Listener::new(self.config.mixnet.bind_address, shared, ingest_sender);
 
         let shutdown_token = self.shutdown_token();
         self.shutdown_tracker().try_spawn_named(

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

@@ -247,10 +247,11 @@ impl ReplayProtectionBloomfilters {
 // map from particular rotation id to vector of results, based on the order of requests received
 type BatchCheckResult = HashMap<u32, Vec<bool>>;
 
+#[allow(unused)]
 impl ReplayProtectionBloomfilters {
     pub(crate) fn batch_try_check_and_set(
         &self,
-        reply_tags: &HashMap<u32, Vec<&[u8; REPLAY_TAG_SIZE]>>,
+        replay_tags: &HashMap<u32, Vec<&[u8; REPLAY_TAG_SIZE]>>,
     ) -> Option<Result<BatchCheckResult, PoisonError<()>>> {
         let mut guard = match self.inner.try_lock() {
             Ok(guard) => guard,
@@ -258,18 +259,43 @@ impl ReplayProtectionBloomfilters {
             Err(TryLockError::WouldBlock) => return None,
         };
 
-        Some(Ok(guard.batch_check_and_set(reply_tags)))
+        Some(Ok(guard.batch_check_and_set(replay_tags)))
     }
 
     pub(crate) fn batch_check_and_set(
         &self,
-        reply_tags: &HashMap<u32, Vec<&[u8; REPLAY_TAG_SIZE]>>,
+        replay_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(()));
         };
 
-        Ok(guard.batch_check_and_set(reply_tags))
+        Ok(guard.batch_check_and_set(replay_tags))
+    }
+
+    pub(crate) fn try_check_and_set(
+        &self,
+        rotation_id: u32,
+        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.check_and_set(rotation_id, replay_tag)))
+    }
+
+    pub(crate) fn check_and_set(
+        &self,
+        rotation_id: u32,
+        replay_tag: &[u8; REPLAY_TAG_SIZE],
+    ) -> Result<bool, PoisonError<()>> {
+        let Ok(mut guard) = self.inner.lock() else {
+            return Err(PoisonError::new(()));
+        };
+        Ok(guard.check_and_set(rotation_id, replay_tag))
     }
 }
 
@@ -287,50 +313,47 @@ struct ReplayProtectionBloomfiltersInner {
     pre_announced: Option<RotationFilter>,
 }
 
+#[allow(unused)]
 impl ReplayProtectionBloomfiltersInner {
     fn batch_check_and_set(
         &mut self,
-        reply_tags: &HashMap<u32, Vec<&[u8; REPLAY_TAG_SIZE]>>,
+        replay_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 filter = if self.primary.metadata.rotation_id == rotation_id {
-                Some(&mut self.primary.data)
-            } 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)
-                } 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(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))
+        let mut result = HashMap::with_capacity(replay_tags.len());
+        for (&rotation_id, replay_tags) in replay_tags {
+            let mut rotation_results = Vec::with_capacity(replay_tags.len());
+            for tag in replay_tags {
+                rotation_results.push(self.check_and_set(rotation_id, tag))
             }
             result.insert(rotation_id, rotation_results);
         }
 
         result
     }
+
+    fn check_and_set(&mut self, rotation_id: u32, replay_tag: &[u8; 32]) -> bool {
+        let filter = if self.primary.metadata.rotation_id == rotation_id {
+            Some(&mut self.primary.data)
+        } else if let Some(secondary) = &mut self.overlap
+            && secondary.metadata.rotation_id == rotation_id
+        {
+            Some(&mut secondary.data)
+        } else if let Some(pre_announced) = &mut self.pre_announced
+            && pre_announced.metadata.rotation_id == rotation_id
+        {
+            Some(&mut pre_announced.data)
+        } else {
+            None
+        };
+
+        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
+            return false;
+        };
+
+        filter.check_and_set(replay_tag)
+    }
 }
 
 pub(crate) struct RotationFilter {

nym-node/src/node/shared_network.rs

@@ -20,7 +20,7 @@ use nym_topology::{
 use nym_validator_client::ValidatorClientError;
 use nym_validator_client::nym_api::NymApiClientExt;
 use nym_validator_client::nym_nodes::{
-    NodesByAddressesResponse, SemiSkimmedNode, SemiSkimmedNodesWithMetadata,
+    NodesByAddressesResponse, SemiSkimmedNodeV1, SemiSkimmedNodesWithMetadata,
 };
 use std::collections::{HashMap, HashSet};
 use std::net::{IpAddr, SocketAddr};
@@ -200,7 +200,7 @@ impl CachedNetwork {
 struct CachedNetworkInner {
     rewarded_set: EpochRewardedSet,
     topology_metadata: NymTopologyMetadata,
-    network_nodes: Vec<SemiSkimmedNode>,
+    network_nodes: Vec<SemiSkimmedNodeV1>,
 }
 
 pub struct NetworkRefresher {

tools/nym-lp-client/src/client.rs

@@ -421,7 +421,7 @@ impl SpeedtestClient {
     /// This is the primary method for sending data through the mixnet via the LP transport.
     /// Requires `init_lp_session()` to be called first to establish the LP cryptographic session.
     ///
-    /// # Data Flow (see gateway/src/node/lp_listener/data_handler.rs)
+    /// # Data Flow (see gateway/src/node/lp_listener/handler)
     /// ```text
     /// LP Client → UDP:51264 → LP Data Handler → Mixnet Entry
     ///           LP(Sphinx)      decrypt LP      forward Sphinx

tools/nym-lp-client/src/topology.rs

@@ -7,7 +7,7 @@
 
 use anyhow::{anyhow, bail, Context, Result};
 use nym_api_requests::models::{LPHashFunction, LPKEM};
-use nym_api_requests::nym_nodes::SkimmedNode;
+use nym_api_requests::nym_nodes::SkimmedNodeV1;
 use nym_crypto::asymmetric::ed25519;
 use nym_http_api_client::UserAgent;
 use nym_kkt_ciphersuite::{Ciphersuite, KEMKeyDigests, SignatureScheme, KEM};
@@ -189,7 +189,7 @@ impl SpeedtestTopology {
 }
 
 /// Extract gateway info for LP connections from a SkimmedNode
-fn gateway_info_from_skimmed(node: &SkimmedNode) -> Result<GatewayInfo> {
+fn gateway_info_from_skimmed(node: &SkimmedNodeV1) -> Result<GatewayInfo> {
     todo!("insufficient information to convert into GatewayInfo")
     // let first_ip = node
     //     .ip_addresses