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

basic node<>node KKT

Browse Source
This commit is contained in:
Jędrzej Stuczyński
2026-03-05 10:38:13 +00:00
parent 225178f95a
commit bc0c2e5d19
15 changed files with 417 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/nym-lp/src/peer.rs
+5 -1
View File
@@ -1,5 +1,5 @@
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
// SPDX-License-Identifier: GPL-3.0-only
use crate::LpError;
use nym_kkt::keys::EncapsulationKey;
@@ -143,6 +143,10 @@ impl LpRemotePeer {
.ok_or(LpError::NoKnownKEMKeyDigests { kem, hash_function })
.cloned()
}
pub fn kem_key_digests(&self) -> &BTreeMap<KEM, KEMKeyDigests> {
&self.expected_kem_key_digests
}
}
impl From<DHPublicKey> for LpRemotePeer {
nym-node/Cargo.toml
+1
View File
@@ -139,6 +139,7 @@ harness = false
cargo_metadata = { workspace = true }
[dev-dependencies]
nym-lp = { workspace = true, features = ["mock"] }
criterion = { workspace = true, features = ["async_tokio"] }
nym-test-utils = { workspace = true }
nym-node/src/node/lp/cleanup.rs
+4 -3
View File
@@ -2,6 +2,7 @@
// SPDX-License-Identifier: GPL-3.0-only
use crate::config::LpDebug;
use crate::node::lp::state::ActiveLpSessions;
use dashmap::DashMap;
use nym_lp::LpTransportSession;
use nym_lp::peer_config::LpReceiverIndex;
@@ -74,14 +75,14 @@ impl<T> TimestampedState<T> {
}
pub(crate) struct CleanupTask {
session_states: Arc<DashMap<LpReceiverIndex, TimestampedState<LpTransportSession>>>,
session_states: ActiveLpSessions,
cfg: LpDebug,
shutdown: nym_task::ShutdownToken,
}
impl CleanupTask {
pub fn new(
session_states: Arc<DashMap<LpReceiverIndex, TimestampedState<LpTransportSession>>>,
session_states: ActiveLpSessions,
cfg: LpDebug,
shutdown: nym_task::ShutdownToken,
) -> Self {
@@ -100,7 +101,7 @@ impl CleanupTask {
// Remove stale sessions (based on time since last activity)
// Use shorter TTL for demoted (ReadOnlyTransport) sessions
self.session_states.retain(|_, timestamped| {
self.session_states.sessions.retain(|_, timestamped| {
if timestamped.since_activity() > session_ttl {
ss_removed += 1;
false
nym-node/src/node/lp/control/egress/connection.rs
+96 -1
View File
@@ -1,11 +1,17 @@
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::node::lp::control::LpConnectionStats;
use crate::node::lp::directory::LpNodeDetails;
use crate::node::lp::error::LpHandlerError;
use crate::node::lp::forwarding::client_connection::NestedClientConnectionSender;
use crate::node::lp::state::SharedLpNodeControlState;
use nym_lp::LpTransportSession;
use nym_lp::peer_config::LpReceiverIndex;
use nym_lp::transport::LpHandshakeChannel;
use std::collections::HashMap;
use std::net::SocketAddr;
use tracing::warn;
use tracing::{debug, warn};
pub(crate) type NestedNodeConnectionSender = ();
pub(crate) type NestedNodeConnectionReceiver = ();
@@ -13,6 +19,95 @@ pub(crate) type NestedNodeConnectionReceiver = ();
pub(crate) type NestedNodeControlSender = ();
pub(crate) type NestedNodeControlReceiver = ();
/// Initial connection handler for an egress LP node before completing the KKT/PSQ handshake.
pub struct InitialLpEgressNodeConnectionHandler<S> {
stream: S,
remote_addr: SocketAddr,
responder_details: LpNodeDetails,
state: SharedLpNodeControlState,
stats: LpConnectionStats,
}
impl<S> InitialLpEgressNodeConnectionHandler<S>
where
S: LpHandshakeChannel + LpHandshakeChannel + Unpin,
{
pub(crate) fn new(
stream: S,
remote_addr: SocketAddr,
responder_details: LpNodeDetails,
state: SharedLpNodeControlState,
) -> Self {
Self {
stream,
remote_addr,
responder_details,
state,
stats: LpConnectionStats::new(),
}
}
pub(crate) async fn complete_initial_handshake(
mut self,
) -> Option<Result<LpTransportSession, LpHandlerError>> {
let remote = self.remote_addr;
if self.responder_details.kem_key_hashes.is_empty() {
return Some(Err(LpHandlerError::MissingNodeKEMKeyHashes {
node_ip: self.remote_addr.ip(),
node_id: self.responder_details.node_id,
}));
}
// 1. complete KKT/PSQ handshake before doing anything else.
// bail if it takes too long
let timeout = self.state.shared.lp_config.debug.handshake_ttl;
let stream = &mut self.stream;
let handshake_state = match LpTransportSession::psq_handshake_initiator_mutual(
stream,
self.state.local_lp_peer.clone(),
self.responder_details.to_lp_peer(),
self.responder_details.supported_protocol,
) {
Ok(handshake_state) => handshake_state,
Err(err) => {
debug!("failed to initiate mutual KTT/PSQ handshake with {remote}: {err}");
self.stats.emit_lifecycle_node_metrics(false);
return None;
}
};
let session = match tokio::time::timeout(timeout, handshake_state.complete_handshake())
.await
{
Err(_timeout) => {
debug!("timed out attempting to complete mutual KTT/PSQ handshake with {remote}");
self.stats.emit_lifecycle_node_metrics(false);
return None;
}
Ok(Err(handshake_failure)) => {
debug!(
"failed to complete mutual KKT/PSQ handshake with {remote}: {handshake_failure}"
);
self.stats.emit_lifecycle_node_metrics(false);
return None;
}
Ok(Ok(session)) => session,
};
debug!(
"completed egress KKT/PSQ handshake with node {}: {remote}",
self.responder_details.node_id
);
// TODO: return session, etc.
Some(Ok(session))
}
}
pub(crate) struct NestedNodeConnectionHandler<S> {
/// Persistent connection to exit gateway for forwarding.
/// Currently, it uses raw TCP socket, later it will be wrapped with dedicated PSQ tunnel
nym-node/src/node/lp/control/ingress/client_handler.rs
+6 -10
View File
@@ -2,7 +2,7 @@
// SPDX-License-Identifier: Apache-2.0
use crate::node::lp::cleanup::TimestampedState;
use crate::node::lp::control::ingress::{LP_DURATION_BUCKETS, LpConnectionStats};
use crate::node::lp::control::{LP_DURATION_BUCKETS, LpConnectionStats};
use crate::node::lp::error::LpHandlerError;
use crate::node::lp::state::SharedLpClientControlState;
use dashmap::mapref::one::RefMut;
@@ -80,8 +80,7 @@ where
self.state
.shared
.session_states
.get_mut(&receiver_index)
.ok_or_else(|| LpHandlerError::MissingLpSession { receiver_index })
.get_state_entry_mut(receiver_index)
}
/// AIDEV-NOTE: Stream-oriented packet loop
@@ -130,10 +129,7 @@ where
let receiver_idx = session.receiver_index();
// 2. insert the state machine into the shared state
self.state
.shared
.session_states
.insert(receiver_idx, TimestampedState::new(session));
self.state.shared.session_states.insert_new_session(session);
self.bound_receiver_idx = Some(receiver_idx);
// 3. handle any new incoming packet
@@ -584,7 +580,7 @@ mod tests {
use super::*;
use crate::config::LpConfig;
use crate::config::lp::LpDebug;
use crate::node::lp::state::SharedLpState;
use crate::node::lp::state::{ActiveLpSessions, 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};
@@ -623,8 +619,8 @@ mod tests {
forward_semaphore,
shared: SharedLpState {
lp_config,
metrics: nym_node_metrics::NymNodeMetrics::default(),
session_states: Arc::new(dashmap::DashMap::new()),
metrics: NymNodeMetrics::default(),
session_states: ActiveLpSessions::new(),
},
}
}
nym-node/src/node/lp/control/ingress/listener.rs
+2 -2
View File
@@ -4,7 +4,7 @@
use crate::config::LpConfig;
use crate::error::NymNodeError;
use crate::node::lp::control::ingress::client_handler::LpClientConnectionHandler;
use crate::node::lp::control::ingress::node_handler::InitialLpNodeConnectionHandler;
use crate::node::lp::control::ingress::node_handler::InitialLpIngressNodeConnectionHandler;
use crate::node::lp::directory::{LpNodeDetails, LpNodes};
use crate::node::lp::state::{SharedLpClientControlState, SharedLpNodeControlState};
use nym_task::ShutdownTracker;
@@ -89,7 +89,7 @@ impl LpControlListener {
debug!("Accepting LP node connection from {remote_addr}");
// Spawn handler task
let mut handler = InitialLpNodeConnectionHandler::new(
let mut handler = InitialLpIngressNodeConnectionHandler::new(
stream,
remote_addr,
initiator_details,
nym-node/src/node/lp/control/ingress/mod.rs
-113
View File
@@ -1,119 +1,6 @@
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use nym_metrics::{add_histogram_obs, inc, inc_by};
pub mod client_handler;
pub(crate) mod listener;
pub mod node_handler;
// Histogram buckets for LP operation duration (legacy - used by unused forwarding methods)
const LP_DURATION_BUCKETS: &[f64] = &[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0];
// Histogram buckets for LP connection lifecycle duration
// LP connections can be very short (registration only: ~1s) or very long (dVPN sessions: hours/days)
// Covers full range from seconds to 24 hours
const LP_CONNECTION_DURATION_BUCKETS: &[f64] = &[
1.0, // 1 second
5.0, // 5 seconds
10.0, // 10 seconds
30.0, // 30 seconds
60.0, // 1 minute
300.0, // 5 minutes
600.0, // 10 minutes
1800.0, // 30 minutes
3600.0, // 1 hour
7200.0, // 2 hours
14400.0, // 4 hours
28800.0, // 8 hours
43200.0, // 12 hours
86400.0, // 24 hours
];
/// Connection lifecycle statistics tracking
struct LpConnectionStats {
/// When the connection started
start_time: std::time::Instant,
/// Total bytes received (including protocol framing)
bytes_received: u64,
/// Total bytes sent (including protocol framing)
bytes_sent: u64,
}
impl LpConnectionStats {
fn new() -> Self {
Self {
start_time: std::time::Instant::now(),
bytes_received: 0,
bytes_sent: 0,
}
}
fn duration(&self) -> std::time::Duration {
self.start_time.elapsed()
}
fn record_bytes_received(&mut self, bytes: usize) {
self.bytes_received += bytes as u64;
}
fn record_bytes_sent(&mut self, bytes: usize) {
self.bytes_sent += bytes as u64;
}
/// Emit connection lifecycle metrics for a client connection
fn emit_lifecycle_client_metrics(&self, graceful: bool) {
// Track connection duration
let duration = self.duration().as_secs_f64();
add_histogram_obs!(
"lp_client_connection_duration_seconds",
duration,
LP_CONNECTION_DURATION_BUCKETS
);
// Track bytes transferred
inc_by!(
"lp_client_connection_bytes_received_total",
self.bytes_received as i64
);
inc_by!(
"lp_client_connection_bytes_sent_total",
self.bytes_sent as i64
);
// Track completion type
if graceful {
inc!("lp_client_connections_completed_gracefully");
} else {
inc!("lp_client_connections_completed_with_error");
}
}
/// Emit connection lifecycle metrics for a node connection
fn emit_lifecycle_node_metrics(&self, graceful: bool) {
// Track connection duration
let duration = self.duration().as_secs_f64();
add_histogram_obs!(
"lp_node_connection_duration_seconds",
duration,
LP_CONNECTION_DURATION_BUCKETS
);
// Track bytes transferred
inc_by!(
"lp_node_connection_bytes_received_total",
self.bytes_received as i64
);
inc_by!(
"lp_node_connection_bytes_sent_total",
self.bytes_sent as i64
);
// Track completion type
if graceful {
inc!("lp_node_connections_completed_gracefully");
} else {
inc!("lp_node_connections_completed_with_error");
}
}
}
nym-node/src/node/lp/control/ingress/node_handler.rs
+34 -18
View File
@@ -1,7 +1,7 @@
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::node::lp::control::ingress::LpConnectionStats;
use crate::node::lp::control::LpConnectionStats;
use crate::node::lp::directory::LpNodeDetails;
use crate::node::lp::error::LpHandlerError;
use crate::node::lp::state::SharedLpNodeControlState;
@@ -14,8 +14,8 @@ use std::net::SocketAddr;
use tokio::net::TcpStream;
use tracing::debug;
/// Initial connection handler for an LP node before completing the KKT/PSQ handshake.
pub struct InitialLpNodeConnectionHandler<S = TcpStream> {
/// Initial connection handler for an ingress LP node before completing the KKT/PSQ handshake.
pub struct InitialLpIngressNodeConnectionHandler<S = TcpStream> {
stream: S,
remote_addr: SocketAddr,
initiator_details: LpNodeDetails,
@@ -24,7 +24,7 @@ pub struct InitialLpNodeConnectionHandler<S = TcpStream> {
stats: LpConnectionStats,
}
impl<S> InitialLpNodeConnectionHandler<S>
impl<S> InitialLpIngressNodeConnectionHandler<S>
where
S: LpHandshakeChannel + LpTransportChannel + Unpin,
{
@@ -47,16 +47,16 @@ where
&self.state.shared.metrics
}
pub async fn handle(mut self) -> Result<(), LpHandlerError> {
// Track total LP connections handled
inc!("lp_node_connections_total");
pub(crate) async fn complete_initial_handshake(
mut self,
) -> Option<Result<LpIngressNodeConnectionHandler<S>, LpHandlerError>> {
let remote = self.remote_addr;
if self.initiator_details.kem_key_hashes.is_empty() {
return Err(LpHandlerError::MissingNodeKEMKeyHashes {
return Some(Err(LpHandlerError::MissingNodeKEMKeyHashes {
node_ip: self.remote_addr.ip(),
node_id: self.initiator_details.node_id,
});
}));
}
// 1. complete KKT/PSQ handshake before doing anything else.
@@ -76,38 +76,50 @@ where
Err(_timeout) => {
debug!("timed out attempting to complete mutual KTT/PSQ handshake with {remote}");
self.stats.emit_lifecycle_node_metrics(false);
return Ok(());
return None;
}
Ok(Err(handshake_failure)) => {
debug!(
"failed to complete mutual KKT/PSQ handshake with {remote}: {handshake_failure}"
);
self.stats.emit_lifecycle_node_metrics(false);
return Ok(());
return None;
}
Ok(Ok(session)) => session,
};
debug!(
"completed KKT/PSQ handshake with node {}: {remote}",
"completed ingress KKT/PSQ handshake with node {}: {remote}",
self.initiator_details.node_id
);
LpNodeConnectionHandler {
Some(Ok(LpIngressNodeConnectionHandler {
stream: self.stream,
remote_addr: remote,
remote_node_id: self.initiator_details.node_id,
state: self.state,
stats: self.stats,
transport_session: session,
}
.handle()
.await
}))
}
pub async fn handle(mut self) -> Result<(), LpHandlerError> {
// Track total LP connections handled
inc!("lp_node_connections_total");
// attempt to complete initial handshake
let upgraded_handler = match self.complete_initial_handshake().await {
None => return Ok(()),
Some(handler_res) => handler_res?,
};
// continue handling the requests with the transport session
upgraded_handler.handle().await
}
}
/// Connection handler for an LP node after completing the KKT/PSQ handshake.
pub struct LpNodeConnectionHandler<S = TcpStream> {
pub struct LpIngressNodeConnectionHandler<S = TcpStream> {
stream: S,
remote_addr: SocketAddr,
remote_node_id: NodeId,
@@ -117,7 +129,7 @@ pub struct LpNodeConnectionHandler<S = TcpStream> {
transport_session: LpTransportSession,
}
impl<S> LpNodeConnectionHandler<S>
impl<S> LpIngressNodeConnectionHandler<S>
where
S: LpHandshakeChannel + LpTransportChannel + Unpin,
{
@@ -127,4 +139,8 @@ where
self.stats.emit_lifecycle_node_metrics(true);
Ok(())
}
pub(crate) fn transport_session(&self) -> &LpTransportSession {
&self.transport_session
}
}
nym-node/src/node/lp/control/mod.rs
+114
View File
@@ -1,5 +1,119 @@
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use nym_metrics::{add_histogram_obs, inc, inc_by};
pub mod egress;
pub mod ingress;
mod tests;
// Histogram buckets for LP operation duration (legacy - used by unused forwarding methods)
const LP_DURATION_BUCKETS: &[f64] = &[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0];
// Histogram buckets for LP connection lifecycle duration
// LP connections can be very short (registration only: ~1s) or very long (dVPN sessions: hours/days)
// Covers full range from seconds to 24 hours
const LP_CONNECTION_DURATION_BUCKETS: &[f64] = &[
1.0, // 1 second
5.0, // 5 seconds
10.0, // 10 seconds
30.0, // 30 seconds
60.0, // 1 minute
300.0, // 5 minutes
600.0, // 10 minutes
1800.0, // 30 minutes
3600.0, // 1 hour
7200.0, // 2 hours
14400.0, // 4 hours
28800.0, // 8 hours
43200.0, // 12 hours
86400.0, // 24 hours
];
/// Connection lifecycle statistics tracking
pub(crate) struct LpConnectionStats {
/// When the connection started
start_time: std::time::Instant,
/// Total bytes received (including protocol framing)
bytes_received: u64,
/// Total bytes sent (including protocol framing)
bytes_sent: u64,
}
impl LpConnectionStats {
fn new() -> Self {
Self {
start_time: std::time::Instant::now(),
bytes_received: 0,
bytes_sent: 0,
}
}
fn duration(&self) -> std::time::Duration {
self.start_time.elapsed()
}
fn record_bytes_received(&mut self, bytes: usize) {
self.bytes_received += bytes as u64;
}
fn record_bytes_sent(&mut self, bytes: usize) {
self.bytes_sent += bytes as u64;
}
/// Emit connection lifecycle metrics for a client connection
fn emit_lifecycle_client_metrics(&self, graceful: bool) {
// Track connection duration
let duration = self.duration().as_secs_f64();
add_histogram_obs!(
"lp_client_connection_duration_seconds",
duration,
LP_CONNECTION_DURATION_BUCKETS
);
// Track bytes transferred
inc_by!(
"lp_client_connection_bytes_received_total",
self.bytes_received as i64
);
inc_by!(
"lp_client_connection_bytes_sent_total",
self.bytes_sent as i64
);
// Track completion type
if graceful {
inc!("lp_client_connections_completed_gracefully");
} else {
inc!("lp_client_connections_completed_with_error");
}
}
/// Emit connection lifecycle metrics for a node connection
fn emit_lifecycle_node_metrics(&self, graceful: bool) {
// Track connection duration
let duration = self.duration().as_secs_f64();
add_histogram_obs!(
"lp_node_connection_duration_seconds",
duration,
LP_CONNECTION_DURATION_BUCKETS
);
// Track bytes transferred
inc_by!(
"lp_node_connection_bytes_received_total",
self.bytes_received as i64
);
inc_by!(
"lp_node_connection_bytes_sent_total",
self.bytes_sent as i64
);
// Track completion type
if graceful {
inc!("lp_node_connections_completed_gracefully");
} else {
inc!("lp_node_connections_completed_with_error");
}
}
}
nym-node/src/node/lp/control/tests.rs
+89
View File
@@ -0,0 +1,89 @@
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
#[cfg(test)]
mod tests {
use crate::node::lp::SharedLpState;
use crate::node::lp::control::egress::connection::InitialLpEgressNodeConnectionHandler;
use crate::node::lp::control::ingress::node_handler::InitialLpIngressNodeConnectionHandler;
use crate::node::lp::directory::LpNodeDetails;
use crate::node::lp::state::SharedLpNodeControlState;
use anyhow::Context;
use nym_lp::packet::version;
use nym_lp::peer::{LpLocalPeer, LpRemotePeer, mock_peers};
use nym_test_utils::helpers::seeded_rng;
use nym_test_utils::mocks::async_read_write::MockIOStream;
use nym_test_utils::traits::TimeboxedSpawnable;
use rand::RngCore;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
fn shared_node_state(peer: LpLocalPeer) -> SharedLpNodeControlState {
SharedLpNodeControlState {
local_lp_peer: peer,
nodes: Default::default(),
shared: SharedLpState {
metrics: Default::default(),
lp_config: Default::default(),
session_states: Default::default(),
},
}
}
fn lp_node_details(peer: LpRemotePeer) -> LpNodeDetails {
let key_bytes = peer.x25519().as_ref().try_into().unwrap();
let mut rng = seeded_rng(key_bytes);
LpNodeDetails::new(
rng.next_u32(),
peer.kem_key_digests().clone(),
peer.x25519().clone(),
version::CURRENT,
)
}
#[tokio::test]
async fn basic_node_to_node_handshake() -> anyhow::Result<()> {
nym_test_utils::helpers::setup_test_logger();
let (init, resp) = mock_peers();
let init_remote = resp.as_remote();
let resp_remote = init.as_remote();
let conn_init = MockIOStream::default();
let conn_resp = conn_init.try_get_remote_handle();
let init_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 1234);
let init_details = lp_node_details(init_remote);
let resp_details = lp_node_details(resp_remote);
let init_state = shared_node_state(init);
let resp_state = shared_node_state(resp);
let init_handler = InitialLpEgressNodeConnectionHandler::new(
conn_init,
init_addr,
resp_details,
init_state,
);
let resp_handler = InitialLpIngressNodeConnectionHandler::new(
conn_resp,
init_addr,
init_details,
resp_state,
);
let init_future = init_handler.complete_initial_handshake().spawn_timeboxed();
let resp_future = resp_handler.complete_initial_handshake().spawn_timeboxed();
let (init_result, resp_result) = tokio::join!(init_future, resp_future);
let init_result = init_result??.context("handshake failure")??;
let resp_result = resp_result??.context("handshake failure")??;
assert_eq!(
init_result.receiver_index(),
resp_result.transport_session().receiver_index()
);
Ok(())
}
}
nym-node/src/node/lp/directory.rs
+27 -2
View File
@@ -2,7 +2,7 @@
// SPDX-License-Identifier: GPL-3.0-only
use arc_swap::ArcSwap;
use nym_lp::peer::DHPublicKey;
use nym_lp::peer::{DHPublicKey, LpRemotePeer};
use nym_lp::{KEM, KEMKeyDigests};
use nym_topology::NodeId;
use std::collections::{BTreeMap, HashMap};
@@ -11,7 +11,7 @@ use std::ops::Deref;
use std::sync::Arc;
/// Wrapper around all known LP nodes
#[derive(Clone)]
#[derive(Clone, Default)]
pub struct LpNodes {
// map between all available ip addresses of other nodes and their details
nodes: Arc<ArcSwap<HashMap<IpAddr, LpNodeDetails>>>,
@@ -32,6 +32,24 @@ pub(crate) struct LpNodeDetails {
inner: Arc<LpNodeDetailsInner>,
}
impl LpNodeDetails {
pub(crate) fn new(
node_id: NodeId,
kem_key_hashes: BTreeMap<KEM, KEMKeyDigests>,
x25519: DHPublicKey,
supported_protocol: u8,
) -> Self {
LpNodeDetails {
inner: Arc::new(LpNodeDetailsInner {
node_id,
kem_key_hashes,
x25519,
supported_protocol,
}),
}
}
}
impl Deref for LpNodeDetails {
type Target = LpNodeDetailsInner;
@@ -44,4 +62,11 @@ pub(crate) struct LpNodeDetailsInner {
pub(crate) node_id: NodeId,
pub(crate) kem_key_hashes: BTreeMap<KEM, KEMKeyDigests>,
pub(crate) x25519: DHPublicKey,
pub(crate) supported_protocol: u8,
}
impl LpNodeDetailsInner {
pub(crate) fn to_lp_peer(&self) -> LpRemotePeer {
LpRemotePeer::new(self.x25519).with_key_digests(self.kem_key_hashes.clone())
}
}
nym-node/src/node/lp/forwarding/client_connection.rs
+1 -1
View File
@@ -1,7 +1,7 @@
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::node::lp::forwarding::node_connection::NestedNodeConnectionSender;
use crate::node::lp::control::egress::connection::NestedNodeConnectionSender;
pub(crate) type NestedClientConnectionSender = ();
pub(crate) type NestedClientConnectionReceiver = ();
nym-node/src/node/lp/forwarding/controller.rs
+1 -1
View File
@@ -1,7 +1,7 @@
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::node::lp::forwarding::node_connection::NestedNodeControlSender;
use crate::node::lp::control::egress::connection::NestedNodeControlSender;
use crate::node::lp::forwarding::{
ConnectionHandlerResponse, GetConnectionHandler, NestedConnectionControllerRequest,
};
nym-node/src/node/lp/mod.rs
+2 -1
View File
@@ -82,6 +82,7 @@ use std::sync::Arc;
use tokio::sync::Semaphore;
use tracing::error;
use crate::node::lp::state::ActiveLpSessions;
pub use nym_mixnet_client::forwarder::{MixForwardingReceiver, mix_forwarding_channels};
pub use state::{SharedLpClientControlState, SharedLpDataState, SharedLpState};
@@ -113,7 +114,7 @@ impl LpSetup {
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 session_states = ActiveLpSessions::new();
let shared_lp_state = SharedLpState {
metrics,
nym-node/src/node/lp/state.rs
+35 -4
View File
@@ -4,7 +4,9 @@
use crate::config::LpConfig;
use crate::node::lp::cleanup::TimestampedState;
use crate::node::lp::directory::LpNodes;
use crate::node::lp::error::LpHandlerError;
use dashmap::DashMap;
use dashmap::mapref::one::RefMut;
use nym_gateway::node::wireguard::PeerRegistrator;
use nym_lp::LpTransportSession;
use nym_lp::peer::LpLocalPeer;
@@ -63,6 +65,37 @@ pub struct SharedLpDataState {
pub shared: SharedLpState,
}
/// Established sessions keyed by the receiver index
///
/// Wrapped in TimestampedState for TTL-based cleanup of inactive sessions.
#[derive(Clone, Default)]
pub struct ActiveLpSessions {
// TODO: this might require split between client and node sessions. TBD
pub(crate) sessions: Arc<DashMap<LpReceiverIndex, TimestampedState<LpTransportSession>>>,
}
impl ActiveLpSessions {
pub fn new() -> Self {
Self::default()
}
pub(crate) fn get_state_entry_mut(
&self,
receiver_index: LpReceiverIndex,
) -> Result<RefMut<'_, LpReceiverIndex, TimestampedState<LpTransportSession>>, LpHandlerError>
{
self.sessions
.get_mut(&receiver_index)
.ok_or_else(|| LpHandlerError::MissingLpSession { receiver_index })
}
pub(crate) fn insert_new_session(&self, session: LpTransportSession) {
let receiver_index = session.receiver_index();
self.sessions
.insert(receiver_index, TimestampedState::new(session));
}
}
/// Shared state for LP connection handlers
#[derive(Clone)]
pub struct SharedLpState {
@@ -72,8 +105,6 @@ pub struct SharedLpState {
/// 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<LpTransportSession>>>,
/// Currently active LP sessions
pub session_states: ActiveLpSessions,
}