From 258ceded26edd1da4d555ebf52eaf3581bf6fb89 Mon Sep 17 00:00:00 2001 From: Simon Wicky Date: Fri, 22 May 2026 15:47:23 +0200 Subject: [PATCH] back to separate pipelines --- common/nym-lp-data/src/nymnodes/traits.rs | 22 ++- nym-node/src/node/lp/data/handler/mod.rs | 38 ++-- .../lp/data/handler/pipeline/mixing_node.rs | 178 ++++++++++++++++++ .../src/node/lp/data/handler/pipeline/mod.rs | 5 + .../node/lp/data/handler/pipeline/nymnode.rs | 39 +--- 5 files changed, 236 insertions(+), 46 deletions(-) create mode 100644 nym-node/src/node/lp/data/handler/pipeline/mixing_node.rs diff --git a/common/nym-lp-data/src/nymnodes/traits.rs b/common/nym-lp-data/src/nymnodes/traits.rs index 4f52533535..4bb11ebe2f 100644 --- a/common/nym-lp-data/src/nymnodes/traits.rs +++ b/common/nym-lp-data/src/nymnodes/traits.rs @@ -13,10 +13,16 @@ use crate::common::traits::{WireUnwrappingPipeline, WireWrappingPipeline}; /// # Type Parameters /// - `Ts`: Timestamp / tick-context type. /// - `Pkt`: Transport packet type; the same type is consumed and produced. -/// - `Opts`: Per-message pipeline options carried into the re-wrapping side. -/// - `Mk`: Message-kind marker returned by the unwrap side. /// - `NdId`: Identifier type for the next-hop destination. /// +/// # Associated Types +/// - `Options`: Per-message pipeline options carried into the re-wrapping side. +/// - `MessageKind`: Message-kind marker returned by the unwrap side. +/// +/// Both are properties of the concrete pipeline rather than something a caller +/// varies, so they live as associated types. This keeps consumers (e.g. a +/// generic worker driver) free of `Options` / `MessageKind` bounds. +/// /// Frame types are owned by the wire sub-traits as associated items and do not /// appear in this trait's parameter list. /// @@ -30,18 +36,22 @@ use crate::common::traits::{WireUnwrappingPipeline, WireWrappingPipeline}; /// [`WireWrappingPipeline::wire_wrap`]. /// /// [`mix`]: NymNodeProcessingPipeline::mix -pub trait NymNodeProcessingPipeline: - WireUnwrappingPipeline + WireWrappingPipeline +pub trait NymNodeProcessingPipeline: + WireUnwrappingPipeline>::MessageKind> + + WireWrappingPipeline>::Options, NdId> where Ts: Clone, NdId: Clone, { + type Options; + type MessageKind; + fn mix( &mut self, - message_kind: Mk, + message_kind: Self::MessageKind, payload: TimedPayload, timestamp: Ts, - ) -> Vec>; + ) -> Vec>; fn process( &mut self, diff --git a/nym-node/src/node/lp/data/handler/mod.rs b/nym-node/src/node/lp/data/handler/mod.rs index fcedb68f35..fb9684e317 100644 --- a/nym-node/src/node/lp/data/handler/mod.rs +++ b/nym-node/src/node/lp/data/handler/mod.rs @@ -16,9 +16,10 @@ //! use crate::node::lp::data::PACKET_BUFFER_SIZE; -use crate::node::lp::data::handler::pipeline::NymNodeDataPipeline; +use crate::node::lp::data::handler::pipeline::{MixingNodeDataPipeline, NymNodeDataPipeline}; use crate::node::lp::data::shared::SharedLpDataState; use nym_lp_data::AddressedTimedData; +use nym_lp_data::common::traits::TransportUnwrap; use nym_lp_data::nymnodes::traits::NymNodeProcessingPipeline; use nym_lp_data::packet::{EncryptedLpPacket, MalformedLpPacketError}; use nym_metrics::inc; @@ -88,15 +89,28 @@ impl LpDataHandler { // Allow at least one worker, even if the config says 0 let worker_count = shared_state.lp_config.debug.data_worker_count.max(1); - // Create workers. They will stop naturally when worker_output_rx is dropped + // Are we running a full size node or just a mixing one + let gateway_mode = shared_state.processing_config.client_forwarding_enabled; + + // Create workers. They will stop naturally when worker_output_rx is dropped. + // The mode is decided once here; each closure picks the right pipeline type so + // the worker loop monomorphizes against a single concrete pipeline. let worker_input_txs = (0..worker_count) .map(|_| { let (worker_input_tx, worker_input_rx) = mpsc::sync_channel(WORKER_QUEUE_DEPTH); let worker_state = shared_state.clone(); let worker_output = worker_output_tx.clone(); - shutdown_tracker.spawn_blocking(move || { - Self::run_worker(worker_state, worker_input_rx, worker_output) - }); + if gateway_mode { + shutdown_tracker.spawn_blocking(move || { + let pipeline = NymNodeDataPipeline::new(worker_state, OsRng); + Self::run_worker(pipeline, worker_input_rx, worker_output); + }); + } else { + shutdown_tracker.spawn_blocking(move || { + let pipeline = MixingNodeDataPipeline::new(worker_state, OsRng); + Self::run_worker(pipeline, worker_input_rx, worker_output); + }); + } worker_input_tx }) @@ -210,15 +224,17 @@ impl LpDataHandler { start } - fn run_worker( - state: Arc, + fn run_worker

( + mut pipeline: P, input_rx: mpsc::Receiver, - output_rx: mpsc::SyncSender, - ) { - let mut pipeline = NymNodeDataPipeline::new(state.clone(), OsRng); + output_tx: mpsc::SyncSender, + ) where + P: NymNodeProcessingPipeline + + TransportUnwrap, // This is needed to specify the error type + { while let Ok(input) = input_rx.recv() { // Blocking is fine, we don't want to unclog ourself and process a new packet that will be dropped anyway - if let Err(e) = output_rx.send(pipeline.process(input.packet, input.timestamp)) { + if let Err(e) = output_tx.send(pipeline.process(input.packet, input.timestamp)) { trace!( "Failed to send processing data back to handler : {e}. We are probably shutting down" ); diff --git a/nym-node/src/node/lp/data/handler/pipeline/mixing_node.rs b/nym-node/src/node/lp/data/handler/pipeline/mixing_node.rs new file mode 100644 index 0000000000..79850bfc14 --- /dev/null +++ b/nym-node/src/node/lp/data/handler/pipeline/mixing_node.rs @@ -0,0 +1,178 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +//! Pipeline for nodes operating purely as mixnodes (no client forwarding). +//! + +use std::{net::SocketAddr, sync::Arc, time::Instant}; + +use nym_lp_data::{ + AddressedTimedData, PipelinePayload, TimedData, TimedPayload, + common::traits::{ + Framing, FramingUnwrap, Transport, TransportUnwrap, WireUnwrappingPipeline, + WireWrappingPipeline, + }, + nymnodes::traits::NymNodeProcessingPipeline, + packet::{EncryptedLpPacket, LpFrame, LpHeader, MalformedLpPacketError, frame::LpFrameHeader}, +}; +use nym_sphinx_addressing::nodes::NymNodeRoutingAddress; +use rand::Rng; +use tracing::warn; + +use crate::node::{ + lp::data::{ + handler::{ + messages::MixMessage, + pipeline::{NymNodeDataPipeline, wire::WirePipeline}, + }, + shared::SharedLpDataState, + }, + routing_filter::RoutingFilter, +}; + +pub(crate) struct MixingNodeDataPipeline { + state: Arc, + wire: WirePipeline, +} + +impl MixingNodeDataPipeline { + pub(crate) fn new(state: Arc, rng: R) -> Self { + Self { + state: state.clone(), + wire: WirePipeline::new(state, rng), + } + } +} + +// Processing logic - stubbed; to be implemented. +impl NymNodeProcessingPipeline + for MixingNodeDataPipeline +{ + type Options = MixMessage; + type MessageKind = MixMessage; + + fn mix( + &mut self, + message_kind: MixMessage, + payload: TimedPayload, + _: Instant, + ) -> Vec> { + // Everything specific to a given packet type should happen here + let processing_result = + NymNodeDataPipeline::::process_mix_packet(&self.state, message_kind, payload); + + self.state.update_processing_metrics(&processing_result); + + let packet_to_forward = match processing_result { + Ok(packet) => packet, + Err(e) => { + warn!("Error processing {message_kind:?} packet : {e}"); + return Vec::new(); + } + }; + + // Now we are deciding if we are routing the packet and where + + match packet_to_forward.dst { + NymNodeRoutingAddress::Node(next_hop) => { + if !self.state.routing_filter.should_route(next_hop.ip()) { + warn!( + event = "packet.dropped.routing_filter", + next_hop = %next_hop, + "dropping packet: egress address does not belong to any known node" + ); + self.state.routing_filter_dropped(next_hop); + Vec::new() + } else { + vec![packet_to_forward.with_dst(next_hop)] + } + } + NymNodeRoutingAddress::Client(_) => { + warn!( + event = "packet.dropped.client_forwarding_disabled", + "dropping packet destined to a client_address on a client_forwarding_disabled node" + ); + self.state.client_forwarding_disabled_dropped(); + Vec::new() + } + } + } +} + +// ============== Wire wrap/unwrap: pure delegation to WirePipeline ============== + +impl Framing for MixingNodeDataPipeline { + type Frame = LpFrame; + const OVERHEAD_SIZE: usize = LpFrameHeader::SIZE; + + fn to_frame( + &mut self, + payload: PipelinePayload, + frame_size: usize, + ) -> Vec> { + let frame = LpFrame { + header: payload.options.into(), + content: payload.data.data.into(), + }; + self.wire + .message_to_frame(payload.data.timestamp, frame, payload.dst, frame_size) + } +} + +impl Transport for MixingNodeDataPipeline { + type Frame = LpFrame; + const OVERHEAD_SIZE: usize = LpHeader::SIZE; + + fn to_transport_packet( + &mut self, + frame: AddressedTimedData, + ) -> AddressedTimedData { + self.wire.frame_to_packet(frame) + } +} + +impl WireWrappingPipeline + for MixingNodeDataPipeline +{ + fn packet_size(&self) -> usize { + nym_lp_data::packet::MTU + } +} + +impl TransportUnwrap for MixingNodeDataPipeline { + type Frame = LpFrame; + type Error = MalformedLpPacketError; + + fn packet_to_frame( + &mut self, + packet: EncryptedLpPacket, + timestamp: Instant, + ) -> Result, Self::Error> { + self.wire.packet_to_frame(packet, timestamp) + } +} + +impl FramingUnwrap for MixingNodeDataPipeline { + type Frame = LpFrame; + + fn frame_to_message( + &mut self, + frame: TimedData, + ) -> Option<(TimedPayload, MixMessage)> { + let reassembled = self.wire.frame_to_maybe_message(frame)?; + let message_kind = MixMessage::from_frame_header(reassembled.data.header) + .inspect_err(|e| warn!("{e}")) + .ok()?; + + self.state.message_received(message_kind); + Some(( + TimedPayload::new(reassembled.timestamp, reassembled.data.content.to_vec()), + message_kind, + )) + } +} + +impl WireUnwrappingPipeline + for MixingNodeDataPipeline +{ +} diff --git a/nym-node/src/node/lp/data/handler/pipeline/mod.rs b/nym-node/src/node/lp/data/handler/pipeline/mod.rs index 28ae3361bf..e5ddd84706 100644 --- a/nym-node/src/node/lp/data/handler/pipeline/mod.rs +++ b/nym-node/src/node/lp/data/handler/pipeline/mod.rs @@ -4,7 +4,12 @@ //! Data pipelines. The wire wrapping/unwrapping //! shared between them lives in [`wire`]. +// Mixing node only +mod mixing_node; + +// Full blown node with client handling capacity mod nymnode; mod wire; +pub(crate) use mixing_node::MixingNodeDataPipeline; pub(crate) use nymnode::NymNodeDataPipeline; diff --git a/nym-node/src/node/lp/data/handler/pipeline/nymnode.rs b/nym-node/src/node/lp/data/handler/pipeline/nymnode.rs index 513142ab92..3874d46ca8 100644 --- a/nym-node/src/node/lp/data/handler/pipeline/nymnode.rs +++ b/nym-node/src/node/lp/data/handler/pipeline/nymnode.rs @@ -44,43 +44,38 @@ impl NymNodeDataPipeline { // Processing logic for packets supported by mixnode enabled node pub fn process_mix_packet( - &mut self, + shared_state: &SharedLpDataState, message_kind: MixMessage, payload: TimedPayload, - _: Instant, ) -> Result, LpDataHandlerError> { match message_kind { MixMessage::Sphinx(metadata) => { - processing::sphinx::process(&self.state, payload, metadata) + processing::sphinx::process(shared_state, payload, metadata) } MixMessage::Outfox(metadata) => { - processing::outfox::process(&self.state, payload, metadata) + processing::outfox::process(shared_state, payload, metadata) } } } } // Processing logic -impl - NymNodeProcessingPipeline< - Instant, - EncryptedLpPacket, - NymNodeMessage, - NymNodeMessage, - SocketAddr, - > for NymNodeDataPipeline +impl NymNodeProcessingPipeline + for NymNodeDataPipeline { + type Options = NymNodeMessage; + type MessageKind = NymNodeMessage; + fn mix( &mut self, message_kind: NymNodeMessage, payload: TimedPayload, - timestamp: Instant, + _: Instant, ) -> Vec> { // Everything specific to a given packet type should happen here let processing_result = match message_kind { - NymNodeMessage::Mix(msg) => self - .process_mix_packet(msg, payload, timestamp) + NymNodeMessage::Mix(msg) => Self::process_mix_packet(&self.state, msg, payload) .map(|payload| payload.options_transform(Into::into)), NymNodeMessage::ForwardSphinx(metadata) => { processing::sphinx::process_forward(&self.state, payload, metadata) @@ -204,20 +199,6 @@ impl FramingUnwrap for NymNodeDataPipeline { .inspect_err(|e| warn!("{e}")) .ok()?; - // Mix-only nodes must not accept gateway-role frames. Reject before processing - // so a malformed/malicious peer can't drive us through `process_forward`. - if !self.state.processing_config.client_forwarding_enabled - && !matches!(message_kind, NymNodeMessage::Mix(_)) - { - warn!( - event = "packet.dropped.unexpected_forward_frame", - ?message_kind, - "dropping unsupported frame on a node with client forwarding disabled" - ); - self.state.processing_misc_error(); - return None; - } - self.state.message_received(message_kind); Some(( TimedPayload::new(reassembled.timestamp, reassembled.data.content.to_vec()),