back to separate pipelines

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<Ts, Pkt, Opts, Mk, NdId>:
-    WireUnwrappingPipeline<Ts, Pkt, Mk> + WireWrappingPipeline<Ts, Pkt, Opts, NdId>
+pub trait NymNodeProcessingPipeline<Ts, Pkt, NdId>:
+    WireUnwrappingPipeline<Ts, Pkt, <Self as NymNodeProcessingPipeline<Ts, Pkt, NdId>>::MessageKind>
+    + WireWrappingPipeline<Ts, Pkt, <Self as NymNodeProcessingPipeline<Ts, Pkt, NdId>>::Options, NdId>
 where
     Ts: Clone,
     NdId: Clone,
 {
+    type Options;
+    type MessageKind;
+
     fn mix(
         &mut self,
-        message_kind: Mk,
+        message_kind: Self::MessageKind,
         payload: TimedPayload<Ts>,
         timestamp: Ts,
-    ) -> Vec<PipelinePayload<Ts, Opts, NdId>>;
+    ) -> Vec<PipelinePayload<Ts, Self::Options, NdId>>;
 
     fn process(
         &mut self,

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<SharedLpDataState>,
+    fn run_worker<P>(
+        mut pipeline: P,
         input_rx: mpsc::Receiver<WorkerInput>,
-        output_rx: mpsc::SyncSender<WorkerOutput>,
-    ) {
-        let mut pipeline = NymNodeDataPipeline::new(state.clone(), OsRng);
+        output_tx: mpsc::SyncSender<WorkerOutput>,
+    ) where
+        P: NymNodeProcessingPipeline<Instant, EncryptedLpPacket, SocketAddr>
+            + TransportUnwrap<Instant, EncryptedLpPacket, Error = MalformedLpPacketError>, // 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"
                 );

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

@@ -0,0 +1,178 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// 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<R> {
+    state: Arc<SharedLpDataState>,
+    wire: WirePipeline<R>,
+}
+
+impl<R: Rng> MixingNodeDataPipeline<R> {
+    pub(crate) fn new(state: Arc<SharedLpDataState>, rng: R) -> Self {
+        Self {
+            state: state.clone(),
+            wire: WirePipeline::new(state, rng),
+        }
+    }
+}
+
+// Processing logic - stubbed; to be implemented.
+impl<R: Rng> NymNodeProcessingPipeline<Instant, EncryptedLpPacket, SocketAddr>
+    for MixingNodeDataPipeline<R>
+{
+    type Options = MixMessage;
+    type MessageKind = MixMessage;
+
+    fn mix(
+        &mut self,
+        message_kind: MixMessage,
+        payload: TimedPayload<Instant>,
+        _: Instant,
+    ) -> Vec<PipelinePayload<Instant, MixMessage, SocketAddr>> {
+        // Everything specific to a given packet type should happen here
+        let processing_result =
+            NymNodeDataPipeline::<R>::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<R: Rng> Framing<Instant, MixMessage, SocketAddr> for MixingNodeDataPipeline<R> {
+    type Frame = LpFrame;
+    const OVERHEAD_SIZE: usize = LpFrameHeader::SIZE;
+
+    fn to_frame(
+        &mut self,
+        payload: PipelinePayload<Instant, MixMessage, SocketAddr>,
+        frame_size: usize,
+    ) -> Vec<AddressedTimedData<Instant, Self::Frame, SocketAddr>> {
+        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<R: Rng> Transport<Instant, EncryptedLpPacket, SocketAddr> for MixingNodeDataPipeline<R> {
+    type Frame = LpFrame;
+    const OVERHEAD_SIZE: usize = LpHeader::SIZE;
+
+    fn to_transport_packet(
+        &mut self,
+        frame: AddressedTimedData<Instant, Self::Frame, SocketAddr>,
+    ) -> AddressedTimedData<Instant, EncryptedLpPacket, SocketAddr> {
+        self.wire.frame_to_packet(frame)
+    }
+}
+
+impl<R: Rng> WireWrappingPipeline<Instant, EncryptedLpPacket, MixMessage, SocketAddr>
+    for MixingNodeDataPipeline<R>
+{
+    fn packet_size(&self) -> usize {
+        nym_lp_data::packet::MTU
+    }
+}
+
+impl<R: Rng> TransportUnwrap<Instant, EncryptedLpPacket> for MixingNodeDataPipeline<R> {
+    type Frame = LpFrame;
+    type Error = MalformedLpPacketError;
+
+    fn packet_to_frame(
+        &mut self,
+        packet: EncryptedLpPacket,
+        timestamp: Instant,
+    ) -> Result<TimedData<Instant, Self::Frame>, Self::Error> {
+        self.wire.packet_to_frame(packet, timestamp)
+    }
+}
+
+impl<R: Rng> FramingUnwrap<Instant, MixMessage> for MixingNodeDataPipeline<R> {
+    type Frame = LpFrame;
+
+    fn frame_to_message(
+        &mut self,
+        frame: TimedData<Instant, Self::Frame>,
+    ) -> Option<(TimedPayload<Instant>, 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<R: Rng> WireUnwrappingPipeline<Instant, EncryptedLpPacket, MixMessage>
+    for MixingNodeDataPipeline<R>
+{
+}

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;

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

@@ -44,43 +44,38 @@ impl<R: Rng> NymNodeDataPipeline<R> {
 
     // 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>,
-        _: Instant,
     ) -> Result<PipelinePayload<Instant, MixMessage, NymNodeRoutingAddress>, 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<R: Rng>
-    NymNodeProcessingPipeline<
-        Instant,
-        EncryptedLpPacket,
-        NymNodeMessage,
-        NymNodeMessage,
-        SocketAddr,
-    > for NymNodeDataPipeline<R>
+impl<R: Rng> NymNodeProcessingPipeline<Instant, EncryptedLpPacket, SocketAddr>
+    for NymNodeDataPipeline<R>
 {
+    type Options = NymNodeMessage;
+    type MessageKind = NymNodeMessage;
+
     fn mix(
         &mut self,
         message_kind: NymNodeMessage,
         payload: TimedPayload<Instant>,
-        timestamp: Instant,
+        _: Instant,
     ) -> Vec<PipelinePayload<Instant, NymNodeMessage, SocketAddr>> {
         // 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<R: Rng> FramingUnwrap<Instant, NymNodeMessage> for NymNodeDataPipeline<R> {
             .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()),