From 0b6166d20ebb79070dc731694b10db520c4280c8 Mon Sep 17 00:00:00 2001
From: mfahampshire <maxhampshire@pm.me>
Date: Tue, 17 Mar 2026 12:02:12 +0000
Subject: [PATCH] Add LpFrameKind::Stream variant with StreamFrameAttributes -
 Define LP wire format for stream multiplexing - Handle new variant in entry
 gateway match arm

---
 common/nym-lp/src/packet/frame.rs             | 63 +++++++++++++++++++
 .../node/lp/control/ingress/client_handler.rs |  2 +-
 2 files changed, 64 insertions(+), 1 deletion(-)

diff --git a/common/nym-lp/src/packet/frame.rs b/common/nym-lp/src/packet/frame.rs
index 5472b45d1b..85bddcde5b 100644
--- a/common/nym-lp/src/packet/frame.rs
+++ b/common/nym-lp/src/packet/frame.rs
@@ -103,6 +103,13 @@ impl LpFrame {
         Self::new(LpFrameKind::Forward, data)
     }
 
+    pub fn new_stream(attrs: StreamFrameAttributes, content: impl Into<Bytes>) -> Self {
+        Self {
+            header: LpFrameHeader::new(LpFrameKind::Stream, attrs.encode()),
+            content: content.into(),
+        }
+    }
+
     pub(crate) fn len(&self) -> usize {
         LpFrameHeader::SIZE + self.content.len()
     }
@@ -115,6 +122,62 @@ pub enum LpFrameKind {
     Opaque = 0,
     Registration = 1,
     Forward = 2,
+    Stream = 3,
+}
+
+/// Message type within a `LpFrameKind::Stream` frame.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum StreamMsgType {
+    /// Open a new stream. Content is optional initial data.
+    Open = 0,
+    /// Data on an existing stream.
+    Data = 1,
+}
+
+/// Parsed form of the 14-byte `frame_attributes` for `LpFrameKind::Stream`.
+///
+/// Wire layout (big-endian):
+/// ```text
+/// [0..8 ) stream_id    : u64
+/// [8    ) msg_type      : u8   (0 = Open, 1 = Data)
+/// [9..13) sequence_num  : u32
+/// [13   ) reserved      : u8
+/// ```
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct StreamFrameAttributes {
+    pub stream_id: u64,
+    pub msg_type: StreamMsgType,
+    pub sequence_num: u32,
+}
+
+impl StreamFrameAttributes {
+    pub fn encode(&self) -> [u8; 14] {
+        let mut buf = [0u8; 14];
+        buf[0..8].copy_from_slice(&self.stream_id.to_be_bytes());
+        buf[8] = self.msg_type as u8;
+        buf[9..13].copy_from_slice(&self.sequence_num.to_be_bytes());
+        buf
+    }
+
+    pub fn parse(attrs: &[u8; 14]) -> Result<Self, MalformedLpPacketError> {
+        let stream_id = u64::from_be_bytes(attrs[0..8].try_into().unwrap());
+        let msg_type = match attrs[8] {
+            0 => StreamMsgType::Open,
+            1 => StreamMsgType::Data,
+            other => {
+                return Err(MalformedLpPacketError::DeserialisationFailure(format!(
+                    "invalid stream msg_type: {other}"
+                )));
+            }
+        };
+        let sequence_num = u32::from_be_bytes(attrs[9..13].try_into().unwrap());
+        Ok(Self {
+            stream_id,
+            msg_type,
+            sequence_num,
+        })
+    }
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
diff --git a/nym-node/src/node/lp/control/ingress/client_handler.rs b/nym-node/src/node/lp/control/ingress/client_handler.rs
index 5fc23caf0d..2ac4955417 100644
--- a/nym-node/src/node/lp/control/ingress/client_handler.rs
+++ b/nym-node/src/node/lp/control/ingress/client_handler.rs
@@ -261,7 +261,7 @@ where
                 self.handle_forwarding_request(receiver_idx, forward_data)
                     .await
             }
-            typ @ LpFrameKind::Opaque => {
+            typ @ (LpFrameKind::Opaque | LpFrameKind::Stream) => {
                 // Neither registration nor forwarding - unknown payload type
                 warn!(
                     "Unknown transport payload type from {remote} (receiver_idx={receiver_idx}). dropping {} bytes",