nym-lp-data crate

Cargo.lock

@@ -7487,12 +7487,12 @@ dependencies = [
 
 [[package]]
 name = "nym-lp-data"
-version = "1.21.0"
+version = "1.20.4"
 dependencies = [
- "bytes",
- "num_enum",
+ "anyhow",
+ "nym-client-core",
  "nym-common",
- "thiserror 2.0.18",
+ "nym-lp",
  "tracing",
 ]
 
@@ -7506,6 +7506,28 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "nym-mix-sim"
+version = "1.20.4"
+dependencies = [
+ "anyhow",
+ "clap",
+ "nym-bin-common",
+ "nym-common",
+ "nym-crypto",
+ "nym-lp-data",
+ "nym-sphinx",
+ "rand 0.8.5",
+ "rand_distr",
+ "serde",
+ "serde_json",
+ "strum 0.28.0",
+ "tokio",
+ "tracing",
+ "tracing-subscriber",
+ "uuid",
+]
+
 [[package]]
 name = "nym-mixnet-client"
 version = "1.21.0"

Cargo.toml

@@ -136,6 +136,7 @@ members = [
     "nym-data-observatory",
     "nym-gateway-probe",
     "nym-ip-packet-client",
+    "nym-mix-sim",
     "nym-network-monitor",
     "nym-node",
     "nym-node-status-api/nym-node-status-agent",
@@ -186,6 +187,7 @@ default-members = [
     "nym-api",
     "nym-authenticator-client",
     "nym-credential-proxy/nym-credential-proxy",
+    "nym-mix-sim",
     "nym-node",
     "nym-registration-client",
     "nym-statistics-api",
@@ -195,6 +197,8 @@ default-members = [
     "service-providers/network-requester",
     "tools/internal/localnet-orchestrator",
     "tools/nymvisor",
+    "nym-registration-client",
+    "tools/internal/localnet-orchestrator",
 ]
 
 exclude = ["contracts", "nym-wallet", "cpu-cycles"]

common/nym-lp-data/Cargo.toml

@@ -13,11 +13,16 @@ documentation.workspace = true
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+anyhow.workspace = true
+tracing.workspace = true
 bytes.workspace = true
 num_enum.workspace = true
-tracing.workspace = true
 thiserror.workspace = true
 
+
+[dev-dependencies]
+nym-client-core.workspace = true
+nym-lp.workspace = true
 nym-common.workspace = true
 
 

common/nym-lp-data/README.md

@@ -0,0 +1,103 @@
+# nym-lp-data
+
+Trait definitions and data structures for Lewes Protocol (LP) processing pipelines in the Nym mixnet.
+
+This crate is a *vocabulary* crate — it defines the traits that clients and mix nodes implement to compose a packet-processing pipeline, plus a few generic data wrappers (`TimedData`, `AddressedTimedData`, `PipelineData`) that thread per-packet state through every stage. It contains no concrete cryptography, transport, or network code. A concrete implementation live in [`nym-mix-sim`](../../nym-mix-sim).
+
+## Crate layout
+
+| Module | Purpose |
+|--------|---------|
+| [`common`](src/common)     | Wire-layer traits ([`Framing`], [`FramingUnwrap`], [`Transport`], [`TransportUnwrap`]) and their composed supertraits ([`WireWrappingPipeline`], [`WireUnwrappingPipeline`]) shared by both clients and mixnodes, plus [`NoOpWireWrapper`] / [`NoOpWireUnwrapper`] marker traits for opting into a pass-through wire layer |
+| [`clients`](src/clients)   | Client-side outbound/inbound pipeline traits: [`Chunking`], [`Reliability`], [`Obfuscation`], [`RoutingSecurity`], plus the supertraits [`ClientWrappingPipeline`] / [`ClientUnwrappingPipeline`], a `Pipeline` composition struct, no-op marker traits, and a tick-driven [`ClientWrappingPipelineDriver`] |
+| [`mixnodes`](src/mixnodes) | Mixnode processing trait [`MixnodeProcessingPipeline`] (unwrap → mix → re-wrap) and a `Pipeline` composition struct |
+
+[`Framing`]: src/common/traits.rs
+[`FramingUnwrap`]: src/common/traits.rs
+[`Transport`]: src/common/traits.rs
+[`TransportUnwrap`]: src/common/traits.rs
+[`WireWrappingPipeline`]: src/common/traits.rs
+[`WireUnwrappingPipeline`]: src/common/traits.rs
+[`NoOpWireWrapper`]: src/common/helpers.rs
+[`NoOpWireUnwrapper`]: src/common/helpers.rs
+[`Chunking`]: src/clients/traits.rs
+[`Reliability`]: src/clients/traits.rs
+[`Obfuscation`]: src/clients/traits.rs
+[`RoutingSecurity`]: src/clients/traits.rs
+[`ClientWrappingPipeline`]: src/clients/traits.rs
+[`ClientUnwrappingPipeline`]: src/clients/traits.rs
+[`ClientWrappingPipelineDriver`]: src/clients/driver.rs
+[`MixnodeProcessingPipeline`]: src/mixnodes/traits.rs
+
+## Core data types
+
+```text
+TimedData<Ts, D>           ──  pairs a value of type D with a timestamp Ts
+TimedPayload<Ts>           ──  alias for TimedData<Ts, Vec<u8>>
+
+AddressedTimedData<Ts, D, NdId>  ──  TimedData plus a destination address
+AddressedTimedPayload<Ts, NdId>  ──  alias for AddressedTimedData<Ts, Vec<u8>, NdId>
+
+PipelineData<Ts, D, Opts, NdId>  ──  TimedData plus per-message Opts
+                                     (used inside the client wrapping pipeline)
+PipelinePayload<Ts, Opts, NdId>  ──  alias for PipelineData<Ts, Vec<u8>, Opts, NdId>
+```
+
+`Ts` is the timestamp / tick-context type, `NdId` is the next-hop identifier type, and `Opts` is an [`InputOptions`](src/clients/mod.rs)-implementing per-message marker that toggles which optional pipeline stages run for a given payload (reliability, obfuscation, routing security).
+
+## Client wrapping pipeline
+
+The outbound client pipeline composes six stages, each represented by its own trait:
+
+```text
+                               Vec<u8>  ──▶  Chunking  ──▶  Reliability  ──▶  Obfuscation
+                                                                                   │
+                                                                                   ▼
+        AddressedTimedData<Ts, Pkt, NdId>  ◀──  Transport  ◀──  Framing  ◀──  RoutingSecurity
+```
+
+[`ClientWrappingPipeline`] is the supertrait that ties them together and provides a default `process()` method which runs all six stages in order on every tick. Each stage is opt-in per message via the active [`InputOptions`].
+
+### Pipeline tick semantics
+
+`process()` is intended to be called on every tick (with or without an input payload):
+
+- [`Reliability::reliable_encode`] is always called once with `Some(input)` (when present), then once more with `None` so that timer-driven retransmissions can fire even when no new payload arrived.
+- [`Obfuscation::obfuscate`] follows the same pattern — once with the real input and once with `None` so that cover-traffic loops can fire on idle ticks.
+- [`Chunking`] and [`RoutingSecurity`] only run when a payload is actually present.
+
+This convention is what allows pipelines to support Poisson cover traffic and SURB-ACK retransmission without the caller having to know whether anything is in flight.
+
+## Mixnode processing pipeline
+
+The mixnode pipeline is simpler — three stages that consume a packet and emit zero or more re-wrapped output packets:
+
+```text
+Pkt  ──▶  WireUnwrappingPipeline  ──▶  mix  ──▶  WireWrappingPipeline  ──▶  Vec<AddressedTimedData<Ts, Pkt, NdId>>
+        (TransportUnwrap +              ▲       (Framing + Transport)
+         FramingUnwrap)                 │
+                                        └──  implementor decrypts, routes,
+                                              schedules delays, etc.
+```
+
+Implementors fill in `mix()`; everything else is provided by the [`MixnodeProcessingPipeline`] supertrait's default `process()`.
+
+## Helpers
+
+- **Client-stage no-op marker traits** ([`NoOpReliability`], [`NoOpRoutingSecurity`], [`NoOpObfuscation`] in [`clients/helpers.rs`](src/clients/helpers.rs)) — implement these to opt out of a pipeline stage with zero overhead. Useful for stub or testing pipelines.
+- **Wire-layer no-op marker traits** ([`NoOpWireWrapper`], [`NoOpWireUnwrapper`] in [`common/helpers.rs`](src/common/helpers.rs)) — collapse the entire wire layer (framing + transport, or their inverses) to a pass-through. Use these when your packet type is already self-contained on the wire (e.g. a Sphinx packet) and needs no extra framing or transport header. `NoOpWireWrapper` requires `Pkt: From<Vec<u8>>`; `NoOpWireUnwrapper` requires `Pkt: Into<Vec<u8>>` and `Mk: Default`.
+- **`Pipeline` composition structs** (in [`clients/types.rs`](src/clients/types.rs) and [`mixnodes/types.rs`](src/mixnodes/types.rs)) — generic structs that aggregate one component per pipeline stage and provide blanket impls of the relevant supertraits, so you can build a working pipeline by plugging in any combination of stage implementations.
+- **[`ClientWrappingPipelineDriver`](src/clients/driver.rs)** — wraps a dyn-compatible client pipeline behind a tick-driven `tick(timestamp) -> Vec<(Pkt, NdId)>` interface, with an internal mpsc channel for application-supplied input payloads. Reads new input only when the internal buffer is empty so buffered packets do not stack additional latency on top.
+
+[`NoOpReliability`]: src/clients/helpers.rs
+[`NoOpRoutingSecurity`]: src/clients/helpers.rs
+[`NoOpObfuscation`]: src/clients/helpers.rs
+[`InputOptions`]: src/clients/mod.rs
+[`Reliability::reliable_encode`]: src/clients/traits.rs
+[`Obfuscation::obfuscate`]: src/clients/traits.rs
+
+## Example users
+
+[`nym-mix-sim`](../../nym-mix-sim) is the reference consumer: it ships two complete pipeline implementations (a pass-through `Simple*` family and a full Sphinx + Poisson + SURB-ACK family) on top of the traits defined here. See its source for end-to-end examples of implementing each pipeline stage.
+
+The integration test under [`tests/integration`](tests/integration) wires together a small synthetic pipeline (`MockChunking`, `KcpReliability`, `SphinxSecurity`, `KekwObfuscation`, `LpFraming`, `LpTransport`) against the [`nym-lp`](../nym-lp) packet types — a useful starting point if you want to read a self-contained example of every trait being implemented.

common/nym-lp-data/src/clients/driver.rs

@@ -0,0 +1,95 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use std::sync::mpsc;
+
+use crate::AddressedTimedData;
+use crate::clients::InputOptions;
+use crate::clients::traits::DynClientWrappingPipeline;
+
+/// Drives a [`DynClientWrappingPipeline`] tick-by-tick, feeding it raw application
+/// payloads and emitting transport packets whose scheduled timestamp is due.
+///
+/// ## How it works
+///
+/// 1. The caller submits raw byte payloads via [`PipelineDriver::input_sender`].
+/// 2. On each call to [`PipelineDriver::tick`], the driver reads one pending
+///    payload (only when both the packet buffer and the obfuscation buffer are
+///    empty, to avoid adding extra latency on top of buffered data), runs it
+///    through the pipeline, and appends the resulting timestamped packets to an
+///    internal buffer.
+/// 3. Packets whose `timestamp ≤ now` are extracted from the buffer and
+///    returned to the caller for sending.
+///
+/// `Ts` must implement `Clone + PartialOrd` so that timestamps can be compared
+/// to decide which packets are due.
+///
+pub struct ClientWrappingPipelineDriver<Ts, Pkt, Opts, NdId>
+where
+    Ts: Clone + PartialOrd,
+    Opts: InputOptions<NdId>,
+{
+    pipeline: Box<dyn DynClientWrappingPipeline<Ts, Pkt, Opts, NdId>>,
+
+    packet_buffer: Vec<AddressedTimedData<Ts, Pkt, NdId>>,
+
+    input: mpsc::Receiver<(Vec<u8>, Opts)>,
+
+    // Keeping a ref so we don't have problem about it being dropped
+    input_sender: mpsc::SyncSender<(Vec<u8>, Opts)>,
+}
+
+impl<Ts, Pkt, Opts, NdId> ClientWrappingPipelineDriver<Ts, Pkt, Opts, NdId>
+where
+    Ts: Clone + PartialOrd,
+    Opts: InputOptions<NdId>,
+{
+    /// Create a new driver wrapping `pipeline`.
+    ///
+    /// Internally allocates a zero-capacity `sync_channel` for input payloads.
+    pub fn new(
+        pipeline: impl DynClientWrappingPipeline<Ts, Pkt, Opts, NdId> + 'static,
+    ) -> Self {
+        let (input_sender, input_receiver) = mpsc::sync_channel(0);
+
+        Self {
+            pipeline: Box::new(pipeline),
+            packet_buffer: Vec::new(),
+            input: input_receiver,
+            input_sender,
+        }
+    }
+
+    /// Return a clone of the sender half of the input channel.
+    ///
+    /// Send raw application payloads here; they will be picked up on the next
+    /// tick when the pipeline's internal buffers are empty.
+    pub fn input_sender(&self) -> mpsc::SyncSender<(Vec<u8>, Opts)> {
+        self.input_sender.clone()
+    }
+
+    /// Advance the driver by one tick.
+    ///
+    /// Reads a pending input payload (if both the packet buffer and the
+    /// obfuscation buffer are empty), runs it through the pipeline, then
+    /// returns all packets whose `timestamp ≤ now`.
+    pub fn tick(&mut self, timestamp: Ts) -> Vec<(Pkt, NdId)> {
+        // We're reading a message only if our buffer is empty
+        // Otherwise, we will have buffers adding latencies to data
+        let next_message = if self.packet_buffer.is_empty() {
+            self.input
+                .try_recv()
+                .inspect_err(|_| tracing::trace!("No message in the queue"))
+                .ok()
+        } else {
+            None
+        };
+        self.packet_buffer
+            .extend(self.pipeline.process(next_message, timestamp.clone()));
+
+        self.packet_buffer
+            .extract_if(.., |p| p.data.timestamp <= timestamp)
+            .map(|pkt| (pkt.data.data, pkt.dst))
+            .collect()
+    }
+}

common/nym-lp-data/src/clients/helpers.rs

@@ -0,0 +1,72 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::clients::traits::{Obfuscation, Reliability, RoutingSecurity};
+use crate::clients::{InputOptions, PipelinePayload};
+
+/// Marker trait for a no-op [`Reliability`] implementation.
+///
+/// Implement this for your pipeline type to get a [`Reliability`] impl that
+/// passes the payload through unchanged with zero byte overhead.
+pub trait NoOpReliability {}
+
+impl<T, Ts, Opts, NdId> Reliability<Ts, Opts, NdId> for T
+where
+    T: NoOpReliability,
+    Opts: InputOptions<NdId>,
+{
+    const OVERHEAD_SIZE: usize = 0;
+    fn reliable_encode(
+        &mut self,
+        input: Option<PipelinePayload<Ts, Opts, NdId>>,
+        _: Ts,
+    ) -> Vec<PipelinePayload<Ts, Opts, NdId>> {
+        input.map(|payload| vec![payload]).unwrap_or_default()
+    }
+}
+
+/// Marker trait for a no-op [`RoutingSecurity`] implementation.
+///
+/// Implement this for your pipeline type to get a [`RoutingSecurity`] impl that
+/// passes the payload through unchanged with zero byte overhead and `nb_frames() == 1`.
+pub trait NoOpRoutingSecurity {}
+
+impl<T, Ts, Opts, NdId> RoutingSecurity<Ts, Opts, NdId> for T
+where
+    T: NoOpRoutingSecurity,
+    Opts: InputOptions<NdId>,
+{
+    const OVERHEAD_SIZE: usize = 0;
+
+    fn nb_frames(&self) -> usize {
+        1
+    }
+
+    fn encrypt(
+        &mut self,
+        input: PipelinePayload<Ts, Opts, NdId>,
+    ) -> PipelinePayload<Ts, Opts, NdId> {
+        input
+    }
+}
+
+/// Marker trait for a no-op [`Obfuscation`] implementation.
+///
+/// Implement this for your pipeline type to get an [`Obfuscation`] impl that
+/// passes the input through unchanged with no cover traffic, delay, or
+/// buffering.
+pub trait NoOpObfuscation {}
+
+impl<T, Ts, Opts, NdId> Obfuscation<Ts, Opts, NdId> for T
+where
+    T: NoOpObfuscation,
+    Opts: InputOptions<NdId>,
+{
+    fn obfuscate(
+        &mut self,
+        input: Option<PipelinePayload<Ts, Opts, NdId>>,
+        _: Ts,
+    ) -> Vec<PipelinePayload<Ts, Opts, NdId>> {
+        input.map(|payload| vec![payload]).unwrap_or_default()
+    }
+}

common/nym-lp-data/src/clients/mod.rs

@@ -0,0 +1,78 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::{AddressedTimedData, TimedData};
+
+pub mod driver;
+pub mod helpers;
+pub mod traits;
+pub mod types;
+
+pub trait InputOptions<NdId>: Clone {
+    fn reliability(&self) -> bool;
+    fn routing_security(&self) -> bool;
+    fn obfuscation(&self) -> bool;
+
+    fn next_hop(&self) -> NdId;
+}
+
+pub struct PipelineData<Ts, D, Opts, NdId>
+where
+    Opts: InputOptions<NdId>,
+{
+    pub data: TimedData<Ts, D>,
+    pub options: Opts,
+    _marker: std::marker::PhantomData<NdId>,
+}
+
+impl<Ts, D, Opts, NdId> PipelineData<Ts, D, Opts, NdId>
+where
+    Opts: InputOptions<NdId>,
+{
+    pub fn new(timestamp: Ts, data: D, options: Opts) -> Self {
+        PipelineData {
+            data: TimedData::new(timestamp, data),
+            options,
+            _marker: std::marker::PhantomData,
+        }
+    }
+
+    /// Apply `op` to the data component, leaving the timestamp unchanged.
+    /// `Nd` can be a different type to allow type transform as well
+    pub fn data_transform<F, Nd>(self, op: F) -> PipelineData<Ts, Nd, Opts, NdId>
+    where
+        F: FnMut(D) -> Nd,
+    {
+        PipelineData {
+            data: self.data.data_transform(op),
+            options: self.options,
+            _marker: self._marker,
+        }
+    }
+
+    /// Apply `op` to the timestamp component, leaving the data unchanged.
+    pub fn ts_transform<F>(self, op: F) -> Self
+    where
+        F: FnMut(Ts) -> Ts,
+    {
+        PipelineData {
+            data: self.data.ts_transform(op),
+            options: self.options,
+            _marker: self._marker,
+        }
+    }
+}
+
+impl<Ts, D, Opts, NdId> From<PipelineData<Ts, D, Opts, NdId>> for AddressedTimedData<Ts, D, NdId>
+where
+    Opts: InputOptions<NdId>,
+{
+    fn from(value: PipelineData<Ts, D, Opts, NdId>) -> Self {
+        AddressedTimedData {
+            dst: value.options.next_hop(),
+            data: value.data,
+        }
+    }
+}
+
+pub type PipelinePayload<Ts, Opts, NdId> = PipelineData<Ts, Vec<u8>, Opts, NdId>;

common/nym-lp-data/src/clients/traits.rs

@@ -0,0 +1,286 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::clients::{InputOptions, PipelinePayload};
+use crate::common::traits::{WireUnwrappingPipeline, WireWrappingPipeline};
+use crate::{AddressedTimedData, TimedPayload};
+
+/// Trait for splitting an incoming payload into timestamped chunks.
+///
+/// # Type Parameters
+/// - `Ts`: Timestamp type associated with each produced `TimedPayload`.
+///
+/// # Parameters
+/// - `input`: Raw payload to split into chunks.
+/// - `chunk_size`: Maximum size of each chunk in bytes.
+/// - `timestamp`: Timestamp to assign to the produced chunks.
+///
+/// # Returns
+/// - A vector of `TimedPayload`s representing the chunked payload.
+pub trait Chunking<Ts, Opts, NdId>
+where
+    Opts: InputOptions<NdId>,
+{
+    fn chunked(
+        &mut self,
+        input: Vec<u8>,
+        input_options: Opts,
+        chunk_size: usize,
+        timestamp: Ts,
+    ) -> Vec<PipelinePayload<Ts, Opts, NdId>>;
+}
+
+/// Trait for applying reliability encoding to a timed payload.
+///
+/// # Type Parameters
+/// - `Ts`: Timestamp type carried by the `TimedPayload`.
+///
+/// # Associated Constants
+/// - `OVERHEAD_SIZE`: Number of additional bytes added by the reliability scheme.
+///
+/// # Parameters
+/// - `input`: Payload to encode with the reliability mechanism.
+/// # Returns
+/// - A vector of `TimedPayload` containing the reliability-encoded data and potential retransmissions.
+pub trait Reliability<Ts, Opts, NdId>
+where
+    Opts: InputOptions<NdId>,
+{
+    const OVERHEAD_SIZE: usize;
+    fn reliable_encode(
+        &mut self,
+        input: Option<PipelinePayload<Ts, Opts, NdId>>,
+        timestamp: Ts,
+    ) -> Vec<PipelinePayload<Ts, Opts, NdId>>;
+}
+
+/// Trait for applying obfuscation to a timed payload.
+/// If obfuscation is used, `obfuscate` should be called at every `Ts` not just the ones with input
+///
+/// # Type Parameters
+/// - `Ts`: Timestamp type carried by the `TimedPayload`.
+pub trait Obfuscation<Ts, Opts, NdId>
+where
+    Opts: InputOptions<NdId>,
+{
+    /// Obfuscate a given timed payload
+    /// # Parameters
+    /// - `input`: Optional payload to obfusctate
+    /// - `timestamp` : Current timestamp
+    ///
+    /// # Returns
+    /// - An `Vec<TimedPayload>`, result of the obfuscation algorithm
+    /// - The vector can be empty if there is nothing to return right away
+    fn obfuscate(
+        &mut self,
+        input: Option<PipelinePayload<Ts, Opts, NdId>>,
+        timestamp: Ts,
+    ) -> Vec<PipelinePayload<Ts, Opts, NdId>>;
+}
+
+/// Trait for applying routing-security encryption to a timed payload.
+///
+/// # Type Parameters
+/// - `Ts`: Timestamp type carried by the `TimedPayload`.
+///
+/// # Associated Constants
+/// - `OVERHEAD_SIZE`: Number of additional bytes added by the encryption scheme.
+///
+/// # Required Methods
+/// - `encrypt`: Encrypt the given payload, returning a new `TimedPayload`.
+///
+/// # Provided Methods
+/// - `nb_frames`: Number of transport frames that one encrypted payload expands
+///   into; defaults to `1`.  Override when the encryption scheme (e.g. Sphinx)
+///   produces multiple frames per input chunk.
+pub trait RoutingSecurity<Ts, Opts, NdId>
+where
+    Opts: InputOptions<NdId>,
+{
+    const OVERHEAD_SIZE: usize;
+    fn nb_frames(&self) -> usize {
+        1
+    }
+    fn encrypt(
+        &mut self,
+        input: PipelinePayload<Ts, Opts, NdId>,
+    ) -> PipelinePayload<Ts, Opts, NdId>;
+}
+
+/// Full client-side outbound message pipeline.
+///
+/// Composes all six processing stages — [`Chunking`], [`Reliability`],
+/// [`Obfuscation`], [`RoutingSecurity`], and the shared [`WireWrappingPipeline`]
+/// (framing + transport) — into a single `process` call that takes a raw byte
+/// payload and returns a list of timestamped transport packets ready for sending.
+///
+/// # Type Parameters
+/// - `Ts`: Timestamp type carried through the pipeline.
+/// - `Pkt`: Final transport packet type produced by transport.
+///
+/// # Provided Methods
+/// - `chunk_size`: Derived from `frame_size` (via [`WireWrappingPipeline`]) minus
+///   routing-security and reliability overheads, accounting for `nb_frames` expansion.
+/// - `process`: Runs the full pipeline in order:
+///   chunk → reliability encode → obfuscate → encrypt → frame → transport.
+pub trait ClientWrappingPipeline<Ts, Pkt, Opts, NdId>:
+    Chunking<Ts, Opts, NdId>
+    + Reliability<Ts, Opts, NdId>
+    + Obfuscation<Ts, Opts, NdId>
+    + RoutingSecurity<Ts, Opts, NdId>
+    + WireWrappingPipeline<Ts, Pkt, NdId>
+where
+    Ts: Clone,
+    NdId: Clone,
+    Opts: InputOptions<NdId>,
+{
+    fn chunk_size(&self, input_options: Opts) -> usize {
+        // Frame size comes from WireWrappingPipeline
+        let mut chunk_size = self.frame_size();
+
+        if input_options.routing_security() {
+            chunk_size =
+                chunk_size * self.nb_frames() - <Self as RoutingSecurity<_, _, _>>::OVERHEAD_SIZE;
+        }
+
+        if input_options.reliability() {
+            chunk_size -= <Self as Reliability<_, _, _>>::OVERHEAD_SIZE;
+        }
+
+        chunk_size
+    }
+
+    fn process(
+        &mut self,
+        input: Option<(Vec<u8>, Opts)>, // Optional to be able to tick the pipeline without input
+        timestamp: Ts,
+    ) -> Vec<AddressedTimedData<Ts, Pkt, NdId>> {
+        let mut chunks = if let Some((input_data, input_options)) = input {
+            self.chunked(
+                input_data,
+                input_options.clone(),
+                self.chunk_size(input_options.clone()),
+                timestamp.clone(),
+            )
+        } else {
+            Vec::new()
+        };
+
+        // Reliability stage with chunks that needs reliability
+        chunks = chunks
+            .into_iter()
+            .flat_map(|chunk| {
+                if chunk.options.reliability() {
+                    self.reliable_encode(Some(chunk), timestamp.clone())
+                } else {
+                    vec![chunk]
+                }
+            })
+            .collect();
+
+        // Even if we had nothing go into the reliability stage, we need to catch potential retransmissions
+        // If we had, this should be a no-op, since it already has been called with the same timestamp
+        chunks.append(&mut self.reliable_encode(None, timestamp.clone()));
+
+        chunks = chunks
+            .into_iter()
+            .flat_map(|chunk| {
+                if chunk.options.obfuscation() {
+                    self.obfuscate(Some(chunk), timestamp.clone())
+                } else {
+                    vec![chunk]
+                }
+            })
+            .collect();
+
+        // Even if we had nothing go into the obfuscation stage, we need to catch potential cover traffic
+        // If we had, this should be a no-op, since it already has been called with the same timestamp
+        chunks.append(&mut self.obfuscate(None, timestamp.clone()));
+
+        chunks = chunks
+            .into_iter()
+            .map(|chunk| {
+                if chunk.options.routing_security() {
+                    self.encrypt(chunk)
+                } else {
+                    chunk
+                }
+            })
+            .collect();
+
+        chunks
+            .into_iter()
+            .flat_map(|payload| self.wire_wrap(payload.into()))
+            .collect::<Vec<_>>()
+    }
+}
+
+/// Dyn-compatible mirror of [`ClientWrappingPipeline`].
+///
+/// All associated constants from the sub-traits are exposed as methods so the
+/// trait can be used as `dyn DynClientWrappingPipeline<Ts, Fr, Pkt>`, erasing the
+/// concrete pipeline type while keeping `Ts`, `Fr`, and `Pkt`.
+///
+/// Implement [`ClientWrappingPipeline`] on your concrete type; the blanket impl
+/// below provides `DynClientWrappingPipeline` for free.
+pub trait DynClientWrappingPipeline<Ts, Pkt, Opts, NdId> {
+    fn packet_size(&self) -> usize;
+
+    fn process(
+        &mut self,
+        input: Option<(Vec<u8>, Opts)>,
+        timestamp: Ts,
+    ) -> Vec<AddressedTimedData<Ts, Pkt, NdId>>;
+}
+
+impl<T, Ts, Pkt, Opts, NdId> DynClientWrappingPipeline<Ts, Pkt, Opts, NdId> for T
+where
+    Ts: Clone,
+    NdId: Clone,
+    Opts: InputOptions<NdId>,
+    T: ClientWrappingPipeline<Ts, Pkt, Opts, NdId>,
+{
+    fn packet_size(&self) -> usize {
+        WireWrappingPipeline::packet_size(self)
+    }
+
+    fn process(
+        &mut self,
+        input: Option<(Vec<u8>, Opts)>,
+        timestamp: Ts,
+    ) -> Vec<AddressedTimedData<Ts, Pkt, NdId>> {
+        ClientWrappingPipeline::process(self, input, timestamp)
+    }
+}
+
+/// Full client-side inbound pipeline.
+///
+/// Combines the shared [`WireUnwrappingPipeline`] (transport + framing unwrap) with a
+/// blank [`process_unwrapped`] step that the implementor fills in (routing-security
+/// decrypt, reliability decode, chunk reassembly, etc.).
+///
+/// # Type Parameters
+/// - `Ts`: Timestamp type.
+/// - `Pkt`: Transport packet type consumed as input.
+/// - `Mk`: Message-kind marker returned alongside reassembled payloads.
+///
+/// # Required Methods
+/// - `process_unwrapped`: Called with the reassembled payload and its message kind
+///   once a complete message is available. Returns the decoded application bytes,
+///   or `None` if reassembly is still in progress.
+///
+/// # Provided Methods
+/// - `unwrap`: Strips the wire layers via [`WireUnwrappingPipeline::wire_unwrap`],
+///   then delegates to `process_unwrapped`.
+pub trait ClientUnwrappingPipeline<Ts, Pkt, Mk>: WireUnwrappingPipeline<Ts, Pkt, Mk>
+where
+    Ts: Clone,
+{
+    fn process_unwrapped(&mut self, payload: TimedPayload<Ts>, kind: Mk) -> Option<Vec<u8>>;
+
+    fn unwrap(&mut self, input: Pkt, timestamp: Ts) -> anyhow::Result<Option<Vec<u8>>> {
+        Ok(self
+            .wire_unwrap(input, timestamp)?
+            .and_then(|(payload, kind)| self.process_unwrapped(payload, kind)))
+    }
+}

common/nym-lp-data/src/clients/types.rs

@@ -0,0 +1,155 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::clients::traits::{
+    Chunking, ClientWrappingPipeline, Obfuscation, Reliability, RoutingSecurity,
+};
+use crate::clients::{InputOptions, PipelinePayload};
+use crate::common::traits::{Framing, Transport, WireWrappingPipeline};
+use crate::{AddressedTimedData, AddressedTimedPayload};
+
+/// Generic composition struct that implements [`ClientWrappingPipeline`] by
+/// delegating each stage to a held component.
+///
+/// Type parameters correspond to the six pipeline stages:
+/// - `C`: [`Chunking`]
+/// - `R`: [`Reliability`]
+/// - `O`: [`Obfuscation`]
+/// - `Rs`: [`RoutingSecurity`]
+/// - `F`: [`Framing`]
+/// - `T`: [`Transport`]
+pub struct Pipeline<C, R, O, Rs, F, T> {
+    pub packet_size: usize,
+    pub chunking: C,
+    pub reliability: R,
+    pub obfuscation: O,
+    pub security: Rs,
+    pub framing: F,
+    pub transport: T,
+}
+
+impl<Ts, Opts, NdId, C, R, O, Rs, F, T> Chunking<Ts, Opts, NdId> for Pipeline<C, R, O, Rs, F, T>
+where
+    Opts: InputOptions<NdId>,
+    C: Chunking<Ts, Opts, NdId>,
+{
+    fn chunked(
+        &mut self,
+        input: Vec<u8>,
+        input_options: Opts,
+        chunk_size: usize,
+        timestamp: Ts,
+    ) -> Vec<PipelinePayload<Ts, Opts, NdId>> {
+        self.chunking
+            .chunked(input, input_options, chunk_size, timestamp)
+    }
+}
+
+impl<Ts, Opts, NdId, C, R, O, Rs, F, T> Reliability<Ts, Opts, NdId> for Pipeline<C, R, O, Rs, F, T>
+where
+    Opts: InputOptions<NdId>,
+    R: Reliability<Ts, Opts, NdId>,
+{
+    const OVERHEAD_SIZE: usize = R::OVERHEAD_SIZE;
+
+    fn reliable_encode(
+        &mut self,
+        input: Option<PipelinePayload<Ts, Opts, NdId>>,
+        timestamp: Ts,
+    ) -> Vec<PipelinePayload<Ts, Opts, NdId>> {
+        self.reliability.reliable_encode(input, timestamp)
+    }
+}
+
+impl<Ts, Opts, NdId, C, R, O, Rs, F, T> Obfuscation<Ts, Opts, NdId> for Pipeline<C, R, O, Rs, F, T>
+where
+    Opts: InputOptions<NdId>,
+    O: Obfuscation<Ts, Opts, NdId>,
+{
+    fn obfuscate(
+        &mut self,
+        input: Option<PipelinePayload<Ts, Opts, NdId>>,
+        timestamp: Ts,
+    ) -> Vec<PipelinePayload<Ts, Opts, NdId>> {
+        self.obfuscation.obfuscate(input, timestamp)
+    }
+}
+
+impl<Ts, Opts, NdId, C, R, O, Rs, F, T> RoutingSecurity<Ts, Opts, NdId>
+    for Pipeline<C, R, O, Rs, F, T>
+where
+    Opts: InputOptions<NdId>,
+    Rs: RoutingSecurity<Ts, Opts, NdId>,
+{
+    const OVERHEAD_SIZE: usize = Rs::OVERHEAD_SIZE;
+
+    fn nb_frames(&self) -> usize {
+        self.security.nb_frames()
+    }
+
+    fn encrypt(
+        &mut self,
+        input: PipelinePayload<Ts, Opts, NdId>,
+    ) -> PipelinePayload<Ts, Opts, NdId> {
+        self.security.encrypt(input)
+    }
+}
+
+impl<Ts, NdId, C, R, O, Rs, F, T> Framing<Ts, NdId> for Pipeline<C, R, O, Rs, F, T>
+where
+    F: Framing<Ts, NdId>,
+{
+    type Frame = F::Frame;
+    const OVERHEAD_SIZE: usize = F::OVERHEAD_SIZE;
+
+    fn to_frame(
+        &self,
+        payload: AddressedTimedPayload<Ts, NdId>,
+        frame_size: usize,
+    ) -> Vec<AddressedTimedData<Ts, F::Frame, NdId>> {
+        self.framing.to_frame(payload, frame_size)
+    }
+}
+
+impl<Ts, Pkt, NdId, C, R, O, Rs, F, T> Transport<Ts, Pkt, NdId> for Pipeline<C, R, O, Rs, F, T>
+where
+    T: Transport<Ts, Pkt, NdId>,
+{
+    type Frame = T::Frame;
+    const OVERHEAD_SIZE: usize = T::OVERHEAD_SIZE;
+
+    fn to_transport_packet(
+        &self,
+        frame: AddressedTimedData<Ts, T::Frame, NdId>,
+    ) -> AddressedTimedData<Ts, Pkt, NdId> {
+        self.transport.to_transport_packet(frame)
+    }
+}
+
+impl<Ts, Pkt, NdId, C, R, O, Rs, F, T> WireWrappingPipeline<Ts, Pkt, NdId>
+    for Pipeline<C, R, O, Rs, F, T>
+where
+    Ts: Clone,
+    NdId: Clone,
+    F: Framing<Ts, NdId>,
+    T: Transport<Ts, Pkt, NdId, Frame = F::Frame>,
+{
+    fn packet_size(&self) -> usize {
+        self.packet_size
+    }
+}
+
+impl<Ts, Pkt, Opts, NdId, C, R, O, Rs, F, T> ClientWrappingPipeline<Ts, Pkt, Opts, NdId>
+    for Pipeline<C, R, O, Rs, F, T>
+where
+    Ts: Clone,
+    NdId: Clone,
+    Opts: InputOptions<NdId>,
+    C: Chunking<Ts, Opts, NdId>,
+    R: Reliability<Ts, Opts, NdId>,
+    O: Obfuscation<Ts, Opts, NdId>,
+    Rs: RoutingSecurity<Ts, Opts, NdId>,
+    F: Framing<Ts, NdId>,
+    T: Transport<Ts, Pkt, NdId, Frame = F::Frame>,
+{
+}

common/nym-lp-data/src/common/helpers.rs

@@ -0,0 +1,100 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::{
+    AddressedTimedData, AddressedTimedPayload, TimedData, TimedPayload,
+    common::traits::{
+        Framing, FramingUnwrap, Transport, TransportUnwrap, WireUnwrappingPipeline,
+        WireWrappingPipeline,
+    },
+};
+
+/// Marker trait for a no-op [`WireWrappingPipeline`] implementation.
+///
+/// Implement this for your pipeline type to get a [`WireWrappingPipeline`] impl that
+/// passes the payload through unchanged with zero byte overhead.
+pub trait NoOpWireWrapper {
+    const PACKET_SIZE: usize = 1500;
+}
+
+impl<T, Ts, NdId> Framing<Ts, NdId> for T
+where
+    T: NoOpWireWrapper,
+{
+    type Frame = Vec<u8>;
+    const OVERHEAD_SIZE: usize = 0;
+    fn to_frame(
+        &self,
+        payload: AddressedTimedPayload<Ts, NdId>,
+        _: usize,
+    ) -> Vec<AddressedTimedPayload<Ts, NdId>> {
+        vec![payload]
+    }
+}
+
+impl<T, Ts, Pkt, NdId> Transport<Ts, Pkt, NdId> for T
+where
+    T: NoOpWireWrapper,
+    Pkt: From<Vec<u8>>,
+{
+    type Frame = Vec<u8>;
+    const OVERHEAD_SIZE: usize = 0;
+    fn to_transport_packet(
+        &self,
+        frame: AddressedTimedPayload<Ts, NdId>,
+    ) -> AddressedTimedData<Ts, Pkt, NdId> {
+        frame.data_transform(|data| data.into())
+    }
+}
+
+impl<T, Ts, Pkt, NdId> WireWrappingPipeline<Ts, Pkt, NdId> for T
+where
+    T: NoOpWireWrapper,
+    Ts: Clone,
+    Pkt: From<Vec<u8>>,
+    NdId: Clone,
+{
+    fn packet_size(&self) -> usize {
+        T::PACKET_SIZE
+    }
+}
+
+/// Marker trait for a no-op [`WireUnwrappingPipeline`] implementation.
+///
+/// Implement this for your pipeline type to get a [`WireUnwrappingPipeline`] impl that
+/// passes the payload through unchanged.
+pub trait NoOpWireUnwrapper {}
+
+impl<T, Ts, Mk> FramingUnwrap<Ts, Mk> for T
+where
+    T: NoOpWireUnwrapper,
+    Mk: Default,
+{
+    type Frame = Vec<u8>;
+    fn frame_to_message(&mut self, frame: TimedPayload<Ts>) -> Option<(TimedPayload<Ts>, Mk)> {
+        Some((frame, Default::default()))
+    }
+}
+
+impl<T, Ts, Pkt> TransportUnwrap<Ts, Pkt> for T
+where
+    T: NoOpWireUnwrapper,
+    Pkt: Into<Vec<u8>>,
+{
+    type Frame = Vec<u8>;
+    fn packet_to_frame(&self, packet: Pkt, timestamp: Ts) -> anyhow::Result<TimedPayload<Ts>> {
+        Ok(TimedData {
+            timestamp,
+            data: packet.into(),
+        })
+    }
+}
+
+impl<T, Ts, Pkt, Mk> WireUnwrappingPipeline<Ts, Pkt, Mk> for T
+where
+    T: NoOpWireUnwrapper,
+    Ts: Clone,
+    Pkt: Into<Vec<u8>>,
+    Mk: Default,
+{
+}

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

@@ -0,0 +1,5 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+pub mod helpers;
+pub mod traits;

common/nym-lp-data/src/common/traits.rs

@@ -0,0 +1,169 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::{AddressedTimedData, AddressedTimedPayload, TimedData, TimedPayload};
+
+/// Trait for applying framing to a timed payload.
+///
+/// # Type Parameters
+/// - `Ts`: Timestamp type carried by the `TimedPayload`.
+///
+/// # Associated Types
+/// - `Frame`: Frame type produced by the framing operation.
+///
+/// # Associated Constants
+/// - `OVERHEAD_SIZE`: Number of additional bytes added by the framing scheme.
+///
+/// # Required Methods
+/// - `to_frame`: Splits the payload into a `Vec<TimedData<Ts, Self::Frame>>` of frames of the given size.
+pub trait Framing<Ts, NdId> {
+    type Frame;
+    const OVERHEAD_SIZE: usize;
+    fn to_frame(
+        &self,
+        payload: AddressedTimedPayload<Ts, NdId>,
+        frame_size: usize,
+    ) -> Vec<AddressedTimedData<Ts, Self::Frame, NdId>>;
+}
+
+/// Trait for unwrapping framing from a frame back into a payload.
+///
+/// # Type Parameters
+/// - `Ts`: Timestamp type carried by the `TimedPayload`.
+/// - `Mk`: Enum describing the kind of message that can be returned.
+///
+/// # Associated Types
+/// - `Frame`: Frame type consumed as input.
+///
+/// # Required Methods
+/// - `frame_to_message`: Attempts to reassemble a payload from the given frame, returning
+///   `Some((payload, kind))` when a complete message is available, or `None` otherwise.
+pub trait FramingUnwrap<Ts, Mk> {
+    type Frame;
+    fn frame_to_message(
+        &mut self,
+        frame: TimedData<Ts, Self::Frame>,
+    ) -> Option<(TimedPayload<Ts>, Mk)>;
+}
+
+/// Trait for applying a transport layer to a framed payload.
+///
+/// # Type Parameters
+/// - `Ts`: Timestamp type carried by the `TimedPayload`.
+/// - `Pkt`: Transport packet type produced as output.
+///
+/// # Associated Types
+/// - `Frame`: Frame type consumed as input.
+///
+/// # Associated Constants
+/// - `OVERHEAD_SIZE`: Number of additional bytes added by the transport scheme.
+///
+/// # Required Methods
+/// - `to_transport_packet`: Wraps a frame into a transport packet.
+pub trait Transport<Ts, Pkt, NdId> {
+    type Frame;
+    const OVERHEAD_SIZE: usize;
+    fn to_transport_packet(
+        &self,
+        frame: AddressedTimedData<Ts, Self::Frame, NdId>,
+    ) -> AddressedTimedData<Ts, Pkt, NdId>;
+}
+
+/// Trait for unwrapping a transport packet back into a frame.
+///
+/// # Type Parameters
+/// - `Ts`: Timestamp type carried by the `TimedPayload`.
+/// - `Pkt`: Transport packet type consumed as input.
+///
+/// # Associated Types
+/// - `Frame`: Frame type produced as output.
+///
+/// # Required Methods
+/// - `packet_to_frame`: Strips the transport layer from a packet, returning the inner frame
+///   tagged with the given timestamp.
+pub trait TransportUnwrap<Ts, Pkt> {
+    type Frame;
+    fn packet_to_frame(
+        &self,
+        packet: Pkt,
+        timestamp: Ts,
+    ) -> anyhow::Result<TimedData<Ts, Self::Frame>>;
+}
+
+/// Supertrait combining [`Framing`] and [`Transport`] into a reusable wire-wrapping layer.
+///
+/// Used as the bottom stage of any outbound pipeline (client or mixnode).
+///
+/// # Type Parameters
+/// - `Ts`: Timestamp type.
+/// - `Pkt`: Final transport packet type.
+///
+/// Both [`Framing`] and [`Transport`] declare their own `type Frame`; this
+/// supertrait cross-constrains them so `to_frame`'s output feeds directly into
+/// `to_transport_packet`.
+///
+/// # Required Methods
+/// - `packet_size`: Total on-wire size of an output packet in bytes.
+///
+/// # Provided Methods
+/// - `frame_size`: Derived from `packet_size` minus transport and framing overheads.
+/// - `wire_wrap`: Frames a payload and wraps each frame into a transport packet.
+pub trait WireWrappingPipeline<Ts, Pkt, NdId>:
+    Transport<Ts, Pkt, NdId>
+    + Framing<Ts, NdId, Frame = <Self as Transport<Ts, Pkt, NdId>>::Frame>
+where
+    Ts: Clone,
+    NdId: Clone,
+{
+    fn packet_size(&self) -> usize;
+
+    fn frame_size(&self) -> usize {
+        self.packet_size()
+            - <Self as Transport<Ts, Pkt, NdId>>::OVERHEAD_SIZE
+            - <Self as Framing<Ts, NdId>>::OVERHEAD_SIZE
+    }
+
+    fn wire_wrap(
+        &self,
+        payload: AddressedTimedPayload<Ts, NdId>,
+    ) -> Vec<AddressedTimedData<Ts, Pkt, NdId>> {
+        let frame_size = self.frame_size();
+        self.to_frame(payload, frame_size)
+            .into_iter()
+            .map(|frame| self.to_transport_packet(frame))
+            .collect()
+    }
+}
+
+/// Supertrait combining [`TransportUnwrap`] and [`FramingUnwrap`] into a reusable
+/// wire-unwrapping layer.
+///
+/// Used as the bottom stage of any inbound pipeline (client or mixnode).
+///
+/// # Type Parameters
+/// - `Ts`: Timestamp type.
+/// - `Pkt`: Transport packet type consumed as input.
+/// - `Mk`: Message-kind marker returned alongside the reassembled payload.
+///
+/// Both [`TransportUnwrap`] and [`FramingUnwrap`] declare their own `type Frame`;
+/// this supertrait cross-constrains them so `packet_to_frame`'s output feeds
+/// directly into `frame_to_message`.
+///
+/// # Provided Methods
+/// - `wire_unwrap`: Strips the transport layer from a packet and attempts to reassemble
+///   a payload, returning `Some((payload, kind))` when a complete message is available.
+pub trait WireUnwrappingPipeline<Ts, Pkt, Mk>:
+    TransportUnwrap<Ts, Pkt>
+    + FramingUnwrap<Ts, Mk, Frame = <Self as TransportUnwrap<Ts, Pkt>>::Frame>
+where
+    Ts: Clone,
+{
+    fn wire_unwrap(
+        &mut self,
+        input: Pkt,
+        timestamp: Ts,
+    ) -> anyhow::Result<Option<(TimedPayload<Ts>, Mk)>> {
+        let frame = self.packet_to_frame(input, timestamp)?;
+        Ok(self.frame_to_message(frame))
+    }
+}

common/nym-lp-data/src/lib.rs

@@ -1,4 +1,149 @@
 // Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
+//! Trait definitions and data structures for low-level packet (LP) processing
+//! pipelines in the Nym mixnet.
+//!
+//! ## Crate layout
+//!
+//! | Module | Purpose |
+//! |--------|---------|
+//! | [`clients`]   | Client-side pipeline traits and types: chunking, reliability, obfuscation, routing security, framing, transport |
+//! | [`common`]    | Shared framing and transport traits used by both clients and mixnodes |
+//! | [`mixnodes`]  | Mixnode-side pipeline traits: unwrap incoming packets, re-wrap and forward them |
+//!
+//! ## Core types
+//!
+//! [`TimedData`] is the foundational wrapper that pairs any piece of data with a
+//! timestamp, threading timing information through every stage of the pipeline.
+//! [`TimedPayload`] is a convenience alias for `TimedData<Ts, Vec<u8>>`.
+
+use std::fmt::Debug;
+
+pub mod clients;
+pub mod common;
+pub mod mixnodes;
 pub mod packet;
+
+/// A value of type `D` tagged with a timestamp of type `Ts`.
+///
+/// `TimedData` threads timing information through every stage of the LP
+/// pipeline.  It is produced by [`clients::traits::Chunking`] and propagated
+/// unchanged (or with the timestamp transformed) through every subsequent
+/// pipeline stage until the packet is sent on the wire.
+pub struct TimedData<Ts, D> {
+    pub timestamp: Ts,
+    pub data: D,
+}
+
+impl<Ts, D> Debug for TimedData<Ts, D>
+where
+    D: Debug,
+    Ts: Debug,
+{
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        writeln!(f, "TimedData {{")?;
+        writeln!(f, "    data:")?;
+        let data_debug = format!("{:#?}", &self.data);
+        for line in data_debug.lines() {
+            writeln!(f, "        {}", line)?;
+        }
+        writeln!(f, "    timestamp: {:#?},", &self.timestamp)?;
+        write!(f, "}}")
+    }
+}
+
+impl<Ts, D> TimedData<Ts, D> {
+    pub fn new(timestamp: Ts, data: D) -> Self {
+        TimedData { timestamp, data }
+    }
+    /// Apply `op` to the data component, leaving the timestamp unchanged.
+    /// `Nd` can be a different type to allow type transform as well
+    pub fn data_transform<F, Nd>(self, mut op: F) -> TimedData<Ts, Nd>
+    where
+        F: FnMut(D) -> Nd,
+    {
+        TimedData {
+            data: op(self.data),
+            timestamp: self.timestamp,
+        }
+    }
+
+    /// Apply `op` to the timestamp component, leaving the data unchanged.
+    pub fn ts_transform<F>(self, mut op: F) -> Self
+    where
+        F: FnMut(Ts) -> Ts,
+    {
+        TimedData {
+            data: self.data,
+            timestamp: op(self.timestamp),
+        }
+    }
+}
+
+pub struct AddressedTimedData<Ts, D, NdId> {
+    pub dst: NdId,
+    pub data: TimedData<Ts, D>,
+}
+
+impl<Ts, D, NdId> AddressedTimedData<Ts, D, NdId> {
+    pub fn new(timestamp: Ts, data: D, dst: NdId) -> Self {
+        AddressedTimedData {
+            dst,
+            data: TimedData::new(timestamp, data),
+        }
+    }
+
+    /// Apply `op` to the data component, leaving the timestamp unchanged.
+    /// `Nd` can be a different type to allow type transform as well
+    pub fn data_transform<F, Nd>(self, op: F) -> AddressedTimedData<Ts, Nd, NdId>
+    where
+        F: FnMut(D) -> Nd,
+    {
+        AddressedTimedData {
+            dst: self.dst,
+            data: self.data.data_transform(op),
+        }
+    }
+
+    /// Apply `op` to the timestamp component, leaving the data unchanged.
+    pub fn ts_transform<F>(self, op: F) -> Self
+    where
+        F: FnMut(Ts) -> Ts,
+    {
+        AddressedTimedData {
+            dst: self.dst,
+            data: self.data.ts_transform(op),
+        }
+    }
+}
+
+impl<Ts, D, NdId> Debug for AddressedTimedData<Ts, D, NdId>
+where
+    D: Debug,
+    Ts: Debug,
+    NdId: Debug,
+{
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        writeln!(f, "AddressedTimedData {{")?;
+        writeln!(f, "    dst: {:#?}", &self.dst)?;
+        writeln!(f, "    data: {:#?}", &self.data)?;
+        write!(f, "}}")
+    }
+}
+
+impl<Ts, D, NdId> From<(NdId, TimedData<Ts, D>)> for AddressedTimedData<Ts, D, NdId> {
+    fn from(value: (NdId, TimedData<Ts, D>)) -> Self {
+        AddressedTimedData {
+            dst: value.0,
+            data: value.1,
+        }
+    }
+}
+
+/// Convenience alias for a [`TimedData`] whose payload is a raw byte buffer.
+///
+/// Used as the input and output type for most pipeline stages before the data
+/// is wrapped into a typed frame or packet.
+pub type TimedPayload<Ts> = TimedData<Ts, Vec<u8>>;
+pub type AddressedTimedPayload<Ts, NdId> = AddressedTimedData<Ts, Vec<u8>, NdId>;

common/nym-lp-data/src/mixnodes/mod.rs

@@ -0,0 +1,5 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+pub mod traits;
+pub mod types;

common/nym-lp-data/src/mixnodes/traits.rs

@@ -0,0 +1,61 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::{AddressedTimedData, AddressedTimedPayload, TimedData, TimedPayload};
+
+use crate::common::traits::{WireUnwrappingPipeline, WireWrappingPipeline};
+
+/// Top-level processing trait for a mix node.
+///
+/// Combines [`WireUnwrappingPipeline`] and [`WireWrappingPipeline`] with a blank [`mix`]
+/// step that the implementor fills in (decrypt, route, re-encrypt, cover traffic, etc.).
+///
+/// # Type Parameters
+/// - `Ts`: Timestamp / tick-context type.
+/// - `Pkt`: Transport packet type; the same type is consumed and produced.
+/// - `Mk`: Message-kind marker returned by the unwrap side.
+/// - `NdId`: Identifier type for the next-hop destination.
+///
+/// Frame types are owned by the wire sub-traits as associated items and do not
+/// appear in this trait's parameter list.
+///
+/// # Required Methods
+/// - `mix`: Given a reassembled payload and the current timestamp, return zero or more
+///   `(next_hop, payload)` pairs to be re-wrapped and forwarded.
+///
+/// # Provided Methods
+/// - `process`: Unwraps the incoming packet via [`WireUnwrappingPipeline::wire_unwrap`],
+///   passes the result to [`mix`], and re-wraps each output payload via
+///   [`WireWrappingPipeline::wire_wrap`].
+pub trait MixnodeProcessingPipeline<Ts, Pkt, Mk, NdId>:
+    WireUnwrappingPipeline<Ts, Pkt, Mk> + WireWrappingPipeline<Ts, Pkt, NdId>
+where
+    Ts: Clone,
+    NdId: Clone,
+{
+    fn mix(
+        &mut self,
+        message_kind: Mk,
+        payload: TimedPayload<Ts>,
+        timestamp: Ts,
+    ) -> Vec<AddressedTimedPayload<Ts, NdId>>;
+
+    fn process(
+        &mut self,
+        input: TimedData<Ts, Pkt>,
+        timestamp: Ts,
+    ) -> anyhow::Result<Vec<AddressedTimedData<Ts, Pkt, NdId>>> {
+        let TimedData {
+            data: packet,
+            timestamp: ts,
+        } = input;
+        let Some((payload, kind)) = self.wire_unwrap(packet, ts)? else {
+            return Ok(Vec::new());
+        };
+        let mixed = self.mix(kind, payload, timestamp);
+        Ok(mixed
+            .into_iter()
+            .flat_map(|addressed_data| self.wire_wrap(addressed_data).into_iter())
+            .collect())
+    }
+}

common/nym-lp-data/src/mixnodes/types.rs

@@ -0,0 +1,56 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::common::traits::{Framing, Transport, WireWrappingPipeline};
+use crate::{AddressedTimedData, AddressedTimedPayload};
+
+/// The generic pipeline struct for a mixnode
+pub struct Pipeline<F, T, NdId> {
+    pub packet_size: usize,
+    pub framing: F,
+    pub transport: T,
+    _marker: std::marker::PhantomData<NdId>,
+}
+
+impl<Ts, F, T, NdId> Framing<Ts, NdId> for Pipeline<F, T, NdId>
+where
+    F: Framing<Ts, NdId>,
+{
+    type Frame = F::Frame;
+    const OVERHEAD_SIZE: usize = F::OVERHEAD_SIZE;
+
+    fn to_frame(
+        &self,
+        payload: AddressedTimedPayload<Ts, NdId>,
+        frame_size: usize,
+    ) -> Vec<AddressedTimedData<Ts, F::Frame, NdId>> {
+        self.framing.to_frame(payload, frame_size)
+    }
+}
+
+impl<Ts, Pkt, F, T, NdId> Transport<Ts, Pkt, NdId> for Pipeline<F, T, NdId>
+where
+    T: Transport<Ts, Pkt, NdId>,
+{
+    type Frame = T::Frame;
+    const OVERHEAD_SIZE: usize = T::OVERHEAD_SIZE;
+
+    fn to_transport_packet(
+        &self,
+        frame: AddressedTimedData<Ts, T::Frame, NdId>,
+    ) -> AddressedTimedData<Ts, Pkt, NdId> {
+        self.transport.to_transport_packet(frame)
+    }
+}
+
+impl<Ts, Pkt, F, T, NdId> WireWrappingPipeline<Ts, Pkt, NdId> for Pipeline<F, T, NdId>
+where
+    Ts: Clone,
+    NdId: Clone,
+    F: Framing<Ts, NdId>,
+    T: Transport<Ts, Pkt, NdId, Frame = F::Frame>,
+{
+    fn packet_size(&self) -> usize {
+        self.packet_size
+    }
+}

common/nym-lp-data/tests/integration/common/mod.rs

@@ -0,0 +1,220 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use nym_lp::packet::{
+    LpFrame, LpHeader, LpPacket,
+    frame::{LpFrameHeader, LpFrameKind},
+};
+
+use nym_lp_data::{
+    AddressedTimedData,
+    common::traits::{Framing, Transport},
+};
+use nym_lp_data::{
+    AddressedTimedPayload,
+    clients::{
+        InputOptions, PipelinePayload,
+        traits::{Chunking, Obfuscation, Reliability, RoutingSecurity},
+    },
+};
+
+#[derive(Clone, Copy)]
+pub struct BasicOptions {
+    pub reliability: bool,
+    pub security: bool,
+    pub obfuscation: bool,
+    pub next_hop: u8,
+}
+
+impl InputOptions<u8> for BasicOptions {
+    fn reliability(&self) -> bool {
+        self.reliability
+    }
+
+    fn routing_security(&self) -> bool {
+        self.security
+    }
+
+    fn obfuscation(&self) -> bool {
+        self.obfuscation
+    }
+
+    fn next_hop(&self) -> u8 {
+        self.next_hop
+    }
+}
+
+pub type BasicPipelinePayload<Ts> = PipelinePayload<Ts, BasicOptions, u8>;
+
+pub struct MockChunking;
+impl<Ts> Chunking<Ts, BasicOptions, u8> for MockChunking
+where
+    Ts: Clone,
+{
+    fn chunked(
+        &mut self,
+        input: Vec<u8>,
+        input_options: BasicOptions,
+        chunk_size: usize,
+        timestamp: Ts,
+    ) -> Vec<BasicPipelinePayload<Ts>> {
+        input
+            .chunks(chunk_size)
+            .map(|chunk| {
+                BasicPipelinePayload::new(timestamp.clone(), chunk.to_vec(), input_options)
+            })
+            .collect()
+    }
+}
+
+pub struct KcpReliability;
+
+impl KcpReliability {
+    const HEADER: &[u8; 5] = b"0KCP0";
+}
+
+impl<Ts> Reliability<Ts, BasicOptions, u8> for KcpReliability {
+    const OVERHEAD_SIZE: usize = Self::HEADER.len();
+    fn reliable_encode(
+        &mut self,
+        input: Option<BasicPipelinePayload<Ts>>,
+        _: Ts,
+    ) -> Vec<BasicPipelinePayload<Ts>> {
+        input
+            .map(|data| {
+                vec![data.data_transform(|data| {
+                    let mut packet = Self::HEADER.to_vec();
+                    packet.extend(data);
+                    packet
+                })]
+            })
+            .unwrap_or_default()
+    }
+}
+
+pub struct SphinxSecurity {
+    pub nb_frames: usize,
+}
+
+impl SphinxSecurity {
+    const HEADER: &[u8; 8] = b"0SPHINX0";
+}
+
+impl<Ts> RoutingSecurity<Ts, BasicOptions, u8> for SphinxSecurity {
+    const OVERHEAD_SIZE: usize = Self::HEADER.len();
+
+    fn nb_frames(&self) -> usize {
+        self.nb_frames
+    }
+
+    fn encrypt(&mut self, input: BasicPipelinePayload<Ts>) -> BasicPipelinePayload<Ts> {
+        input.data_transform(|data| {
+            let mut packet = Self::HEADER.to_vec();
+            packet.extend(data);
+            packet
+        })
+    }
+}
+
+pub struct KekwObfuscation;
+
+impl Obfuscation<u32, BasicOptions, u8> for KekwObfuscation {
+    fn obfuscate(
+        &mut self,
+        input: Option<BasicPipelinePayload<u32>>,
+        _timestamp: u32,
+    ) -> Vec<BasicPipelinePayload<u32>> {
+        if let Some(input) = input {
+            vec![input.ts_transform(|ts| ts + 1)]
+        } else {
+            Vec::new()
+        }
+    }
+}
+
+#[allow(dead_code)]
+pub struct ReallyOddObfuscation {
+    next_ts: u32,
+}
+
+impl ReallyOddObfuscation {
+    #[allow(dead_code)]
+    pub fn new(start_ts: u32) -> Self {
+        let next_ts = if !start_ts.is_multiple_of(2) {
+            start_ts
+        } else {
+            start_ts + 1
+        };
+        Self { next_ts }
+    }
+}
+
+impl Obfuscation<u32, BasicOptions, u8> for ReallyOddObfuscation {
+    fn obfuscate(
+        &mut self,
+        input: Option<BasicPipelinePayload<u32>>,
+        _timestamp: u32,
+    ) -> Vec<BasicPipelinePayload<u32>> {
+        if let Some(input) = input {
+            let pkt = input.ts_transform(|_| self.next_ts);
+            self.next_ts += 2;
+            vec![pkt]
+        } else {
+            Vec::new()
+        }
+    }
+}
+
+pub struct LpFraming;
+
+impl LpFraming {
+    const FRAME_ATTRIBUTES: &[u8; 14] = b"0LpFrameAttrs0";
+}
+
+impl<Ts> Framing<Ts, u8> for LpFraming
+where
+    Ts: Clone,
+{
+    type Frame = LpFrame;
+    const OVERHEAD_SIZE: usize = LpFrameHeader::SIZE;
+    fn to_frame(
+        &self,
+        input: AddressedTimedPayload<Ts, u8>,
+        frame_size: usize,
+    ) -> Vec<AddressedTimedData<Ts, LpFrame, u8>> {
+        input
+            .data
+            .data
+            .chunks(frame_size)
+            .map(|frame_payload| {
+                let header = LpFrameHeader::new(LpFrameKind::Opaque, *Self::FRAME_ATTRIBUTES);
+
+                AddressedTimedData::new(
+                    input.data.timestamp.clone(),
+                    LpFrame {
+                        header,
+                        content: frame_payload.to_vec().into(),
+                    },
+                    input.dst,
+                )
+            })
+            .collect()
+    }
+}
+
+pub struct LpTransport;
+
+impl<Ts> Transport<Ts, LpPacket, u8> for LpTransport {
+    type Frame = LpFrame;
+    const OVERHEAD_SIZE: usize = LpHeader::SIZE;
+    fn to_transport_packet(
+        &self,
+        input: AddressedTimedData<Ts, Self::Frame, u8>,
+    ) -> AddressedTimedData<Ts, LpPacket, u8> {
+        AddressedTimedData::new(
+            input.data.timestamp,
+            LpPacket::new(LpHeader::new(7, 7, 7), input.data.data),
+            input.dst,
+        )
+    }
+}

common/nym-lp-data/tests/integration/main.rs

@@ -0,0 +1,34 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use nym_lp_data::clients::{traits::ClientWrappingPipeline, types::Pipeline};
+
+use crate::common::{
+    KcpReliability, KekwObfuscation, LpFraming, LpTransport, MockChunking, SphinxSecurity,
+};
+
+mod common;
+
+#[test]
+fn empty_input_yields_empty_output() {
+    let packet_size = 64;
+    let security_layer_nb_frames = 2;
+
+    let mut mock_pipeline = Pipeline {
+        chunking: MockChunking,
+        reliability: KcpReliability,
+        security: SphinxSecurity {
+            nb_frames: security_layer_nb_frames,
+        },
+        obfuscation: KekwObfuscation,
+        framing: LpFraming,
+        transport: LpTransport,
+        packet_size,
+    };
+
+    let output = mock_pipeline.process(None, 1);
+
+    assert!(output.is_empty());
+}
+
+// TODO More test to come later