From 2653d12e550757a55ec0fcf5ebd6f4d8804e0df6 Mon Sep 17 00:00:00 2001 From: benedettadavico Date: Fri, 24 Apr 2026 16:07:49 +0200 Subject: [PATCH] fix ipr msg, and unit tests --- Cargo.lock | 1 + common/ip-packet-requests/src/lib.rs | 20 +++++ nym-gateway-probe/src/common/helpers.rs | 8 +- nym-gateway-probe/src/common/icmp.rs | 29 +++++-- nym-gateway-probe/src/common/probe_tests.rs | 23 ++++-- nym-ip-packet-client/Cargo.toml | 1 + nym-ip-packet-client/src/connect.rs | 46 ++++++++--- nym-ip-packet-client/src/helpers.rs | 26 +++--- nym-ip-packet-client/src/lib.rs | 1 + nym-ip-packet-client/src/lp_stream.rs | 79 +++++++++++++++++++ .../nym-sdk/src/ipr_wrapper/ip_mix_stream.rs | 10 +-- .../ip-packet-router/src/mixnet_listener.rs | 24 ++++-- 12 files changed, 219 insertions(+), 49 deletions(-) create mode 100644 nym-ip-packet-client/src/lp_stream.rs diff --git a/Cargo.lock b/Cargo.lock index a1015be7fb..970150dff0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7028,6 +7028,7 @@ dependencies = [ "bytes", "futures", "nym-ip-packet-requests", + "nym-lp", "nym-sdk", "thiserror 2.0.12", "tokio", diff --git a/common/ip-packet-requests/src/lib.rs b/common/ip-packet-requests/src/lib.rs index f3014a69ec..2ff84dddee 100644 --- a/common/ip-packet-requests/src/lib.rs +++ b/common/ip-packet-requests/src/lib.rs @@ -12,6 +12,26 @@ pub mod v7; pub mod v8; pub mod v9; +/// Highest IPR protocol version that is allowed to be sent as a **non-stream** mixnet payload +/// (i.e. not wrapped in `LpFrameKind::SphinxStream`). +pub const MAX_NON_STREAM_VERSION: u8 = v8::VERSION; + +/// First IPR protocol version that **requires** the SphinxStream (LP) transport for non-stream +/// mixnet sends, matching the node-side enforcement in `ip-packet-router`. +pub const SPHINX_STREAM_VERSION_THRESHOLD: u8 = v9::VERSION; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn stream_transport_threshold_is_consistent() { + assert_eq!(MAX_NON_STREAM_VERSION, v8::VERSION); + assert_eq!(SPHINX_STREAM_VERSION_THRESHOLD, v9::VERSION); + assert!(SPHINX_STREAM_VERSION_THRESHOLD > MAX_NON_STREAM_VERSION); + } +} + // version 3: initial version // version 4: IPv6 support // version 5: Add severity level to info response diff --git a/nym-gateway-probe/src/common/helpers.rs b/nym-gateway-probe/src/common/helpers.rs index 5dc330e9cc..54822c3987 100644 --- a/nym-gateway-probe/src/common/helpers.rs +++ b/nym-gateway-probe/src/common/helpers.rs @@ -1,9 +1,10 @@ // Copyright 2026 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use nym_ip_packet_requests::v8::response::{ +use nym_ip_packet_client::current::response::{ ControlResponse, DataResponse, InfoLevel, IpPacketResponse, IpPacketResponseData, }; +use nym_ip_packet_client::lp_stream; use nym_sdk::{ DebugConfig, NymApiTopologyProvider, NymApiTopologyProviderConfig, NymNetworkDetails, TopologyProvider, mixnet::ReconstructedMessage, @@ -32,7 +33,10 @@ pub fn mixnet_debug_config( } pub fn unpack_data_response(reconstructed_message: &ReconstructedMessage) -> Option { - match IpPacketResponse::from_reconstructed_message(reconstructed_message) { + let payload = + lp_stream::maybe_unwrap_lp_stream_payload_from_reconstructed(reconstructed_message); + + match IpPacketResponse::from_bytes(payload) { Ok(response) => match response.data { IpPacketResponseData::Data(data_response) => Some(data_response), IpPacketResponseData::Control(control) => match *control { diff --git a/nym-gateway-probe/src/common/icmp.rs b/nym-gateway-probe/src/common/icmp.rs index ec757623f1..2d96a27da5 100644 --- a/nym-gateway-probe/src/common/icmp.rs +++ b/nym-gateway-probe/src/common/icmp.rs @@ -9,7 +9,7 @@ use nym_connection_monitor::{ wrap_icmp_in_ipv6, }, }; -use nym_ip_packet_requests::{IpPair, codec::MultiIpPacketCodec, v8::request::IpPacketRequest}; +use nym_ip_packet_requests::{IpPair, codec::MultiIpPacketCodec}; use nym_sdk::mixnet::{ InputMessage, MixnetClient, MixnetMessageSender, Recipient, TransmissionLane, }; @@ -25,6 +25,7 @@ pub async fn send_ping_v4( sequence_number: u16, destination: Ipv4Addr, exit_router_address: Recipient, + stream_id: u64, ) -> anyhow::Result<()> { let icmp_identifier = icmp_identifier(); let icmp_echo_request = create_icmpv4_echo_request(sequence_number, icmp_identifier)?; @@ -35,7 +36,12 @@ pub async fn send_ping_v4( MultiIpPacketCodec::bundle_one_packet(ipv4_packet.packet().to_vec().into()); // Wrap into a mixnet input message addressed to the IPR - let mixnet_message = create_input_message(exit_router_address, bundled_packet)?; + let mixnet_message = create_input_message( + exit_router_address, + bundled_packet, + stream_id, + sequence_number, + )?; mixnet_client.send(mixnet_message).await?; Ok(()) @@ -47,6 +53,7 @@ pub async fn send_ping_v6( sequence_number: u16, destination: Ipv6Addr, exit_router_address: Recipient, + stream_id: u64, ) -> anyhow::Result<()> { let icmp_identifier = icmp_identifier(); let icmp_echo_request = create_icmpv6_echo_request( @@ -62,7 +69,12 @@ pub async fn send_ping_v6( MultiIpPacketCodec::bundle_one_packet(ipv6_packet.packet().to_vec().into()); // Wrap into a mixnet input message addressed to the IPR - let mixnet_message = create_input_message(exit_router_address, bundled_packet)?; + let mixnet_message = create_input_message( + exit_router_address, + bundled_packet, + stream_id, + sequence_number, + )?; // Send across the mixnet mixnet_client.send(mixnet_message).await?; @@ -72,15 +84,22 @@ pub async fn send_ping_v6( fn create_input_message( recipient: impl Into, bundled_packets: Bytes, + stream_id: u64, + sequence_number: u16, ) -> anyhow::Result { - let packet = IpPacketRequest::new_data_request(bundled_packets).to_bytes()?; + let packet = nym_ip_packet_client::current::new_data_request(bundled_packets).to_bytes()?; + let framed_packet = nym_ip_packet_client::lp_stream::encode_stream_frame( + stream_id, + sequence_number as u32, + packet, + ); let lane = TransmissionLane::General; let packet_type = None; let surbs = 0; Ok(InputMessage::new_anonymous( recipient.into(), - packet, + framed_packet, surbs, lane, packet_type, diff --git a/nym-gateway-probe/src/common/probe_tests.rs b/nym-gateway-probe/src/common/probe_tests.rs index d37a1042d2..081fba9c21 100644 --- a/nym-gateway-probe/src/common/probe_tests.rs +++ b/nym-gateway-probe/src/common/probe_tests.rs @@ -256,8 +256,8 @@ pub async fn do_ping( let (maybe_ip_pair, mut mixnet_client) = connect_exit(mixnet_client, exit_router_address).await; match maybe_ip_pair { - Some(ip_pair) => ( - do_ping_exit(&mut mixnet_client, ip_pair, exit_router_address).await, + Some((ip_pair, stream_id)) => ( + do_ping_exit(&mut mixnet_client, ip_pair, stream_id, exit_router_address).await, mixnet_client, ), None => (Ok(Some(Exit::fail_to_connect())), mixnet_client), @@ -304,7 +304,7 @@ async fn do_ping_entry( async fn connect_exit( mixnet_client: MixnetClient, exit_router_address: Recipient, -) -> (Option, MixnetClient) { +) -> (Option<(IpPair, u64)>, MixnetClient) { // Step 2: connect to the exit gateway info!( "Connecting to exit gateway: {}", @@ -315,12 +315,19 @@ async fn connect_exit( let mut ipr_client = IprClientConnect::new(mixnet_client, cancel_token); let maybe_ip_pair = ipr_client.connect(exit_router_address).await; + let stream_id = ipr_client.active_stream_id(); let mixnet_client = ipr_client.into_mixnet_client(); if let Ok(our_ips) = maybe_ip_pair { info!("Successfully connected to exit gateway"); info!("Using mixnet VPN IP addresses: {our_ips}"); - (Some(our_ips), mixnet_client) + let Some(stream_id) = stream_id else { + tracing::warn!( + "No active IPR stream id set after connect; cannot run IPR data-plane tests" + ); + return (None, mixnet_client); + }; + (Some((our_ips, stream_id)), mixnet_client) } else { (None, mixnet_client) } @@ -329,10 +336,11 @@ async fn connect_exit( pub async fn do_ping_exit( mixnet_client: &mut MixnetClient, our_ips: IpPair, + stream_id: u64, exit_router_address: Recipient, ) -> anyhow::Result> { // Step 3: perform ICMP connectivity checks for the exit gateway - send_icmp_pings(mixnet_client, our_ips, exit_router_address).await?; + send_icmp_pings(mixnet_client, our_ips, exit_router_address, stream_id).await?; listen_for_icmp_ping_replies(mixnet_client, our_ips).await } @@ -340,6 +348,7 @@ async fn send_icmp_pings( mixnet_client: &MixnetClient, our_ips: IpPair, exit_router_address: Recipient, + stream_id: u64, ) -> anyhow::Result<()> { // ipv4 addresses for testing let ipr_tun_ip_v4 = NYM_TUN_DEVICE_ADDRESS_V4; @@ -361,6 +370,7 @@ async fn send_icmp_pings( ii, ipr_tun_ip_v4, exit_router_address, + stream_id, ) .await?; icmp::send_ping_v4( @@ -369,6 +379,7 @@ async fn send_icmp_pings( ii, external_ip_v4, exit_router_address, + stream_id, ) .await?; } @@ -381,6 +392,7 @@ async fn send_icmp_pings( ii, ipr_tun_ip_v6, exit_router_address, + stream_id, ) .await?; icmp::send_ping_v6( @@ -389,6 +401,7 @@ async fn send_icmp_pings( ii, external_ip_v6, exit_router_address, + stream_id, ) .await?; } diff --git a/nym-ip-packet-client/Cargo.toml b/nym-ip-packet-client/Cargo.toml index 47b2de02c3..4117d80350 100644 --- a/nym-ip-packet-client/Cargo.toml +++ b/nym-ip-packet-client/Cargo.toml @@ -26,3 +26,4 @@ tracing.workspace = true nym-sdk = { workspace = true } nym-ip-packet-requests = { workspace = true } +nym-lp = { workspace = true } diff --git a/nym-ip-packet-client/src/connect.rs b/nym-ip-packet-client/src/connect.rs index 78ca7b3dff..6ca04e2227 100644 --- a/nym-ip-packet-client/src/connect.rs +++ b/nym-ip-packet-client/src/connect.rs @@ -14,7 +14,7 @@ use tracing::{debug, error}; use nym_ip_packet_requests::response_helpers::{self, IprResponseError}; use crate::{ - current::{request::IpPacketRequest, response::IpPacketResponse}, + current::{self, response::IpPacketResponse}, error::{Error, Result}, helpers::check_ipr_message_version, }; @@ -36,6 +36,7 @@ pub struct IprClientConnect { mixnet_client: MixnetClient, connected: ConnectionState, cancel_token: CancellationToken, + active_stream_id: Option, } impl IprClientConnect { @@ -44,6 +45,7 @@ impl IprClientConnect { mixnet_client, connected: ConnectionState::Disconnected, cancel_token, + active_stream_id: None, } } @@ -51,6 +53,10 @@ impl IprClientConnect { self.mixnet_client } + pub fn active_stream_id(&self) -> Option { + self.active_stream_id + } + pub async fn connect(&mut self, ip_packet_router_address: Recipient) -> Result { if self.connected != ConnectionState::Disconnected { return Err(Error::AlreadyConnected); @@ -79,8 +85,8 @@ impl IprClientConnect { self.listen_for_connect_response(request_id).await } - async fn send_connect_request(&self, ip_packet_router_address: Recipient) -> Result { - let (request, request_id) = nym_ip_packet_requests::v9::new_connect_request(None); + async fn send_connect_request(&mut self, ip_packet_router_address: Recipient) -> Result { + let (request, request_id) = current::new_connect_request(None); tracing::info!( request_id = request_id, protocol_version = request.protocol.version, @@ -95,10 +101,13 @@ impl IprClientConnect { // We use 20 surbs for the connect request because typically the IPR is configured to have // a min threshold of 10 surbs that it reserves for itself to request additional surbs. let surbs = 20; + let request_bytes = request.to_bytes()?; + let framed_bytes = maybe_wrap_stream_frame(request_id, 0, request_bytes); + self.active_stream_id = Some(request_id); self.mixnet_client - .send(create_input_message( + .send(create_input_message_bytes( ip_packet_router_address, - request, + framed_bytes, surbs, )?) .await @@ -151,7 +160,7 @@ impl IprClientConnect { // Then we deserialize the message tracing::debug!("IprClient: got message while waiting for connect response"); - let Ok(response) = IpPacketResponse::from_reconstructed_message(&msg) else { + let Ok(response) = ipr_response_from_reconstructed_message(&msg) else { // This is ok, it's likely just one of our self-pings tracing::debug!("Failed to deserialize mixnet message"); continue; @@ -176,14 +185,33 @@ impl IprClientConnect { } } -fn create_input_message( +fn maybe_wrap_stream_frame(stream_id: u64, sequence_num: u32, payload: Vec) -> Vec { + if !crate::lp_stream::current_requires_sphinx_stream_transport() { + return payload; + } + + crate::lp_stream::encode_stream_frame(stream_id, sequence_num, payload) +} + +fn ipr_response_from_reconstructed_message( + msg: &nym_sdk::mixnet::ReconstructedMessage, +) -> std::result::Result { + let payload = if crate::lp_stream::current_requires_sphinx_stream_transport() { + crate::lp_stream::maybe_unwrap_lp_stream_payload_from_reconstructed(msg) + } else { + &msg.message + }; + IpPacketResponse::from_bytes(payload) +} + +fn create_input_message_bytes( recipient: Recipient, - request: IpPacketRequest, + bytes: Vec, surbs: u32, ) -> Result { Ok(InputMessage::new_anonymous( recipient, - request.to_bytes()?, + bytes, surbs, TransmissionLane::General, None, diff --git a/nym-ip-packet-client/src/helpers.rs b/nym-ip-packet-client/src/helpers.rs index 1b31870ff8..cfa0957aec 100644 --- a/nym-ip-packet-client/src/helpers.rs +++ b/nym-ip-packet-client/src/helpers.rs @@ -7,18 +7,16 @@ use nym_sdk::mixnet::ReconstructedMessage; use crate::{current::VERSION as CURRENT_VERSION, error::Result}; pub(crate) fn check_ipr_message_version(message: &ReconstructedMessage) -> Result<()> { - nym_ip_packet_requests::response_helpers::check_ipr_message_version( - &message.message, - CURRENT_VERSION, - ) - .map_err(|e| match e { - IprResponseError::NoVersionByte => crate::Error::NoVersionInMessage, - IprResponseError::VersionMismatch { expected, received } if received < expected => { - crate::Error::ReceivedResponseWithOldVersion { expected, received } - } - IprResponseError::VersionMismatch { expected, received } => { - crate::Error::ReceivedResponseWithNewVersion { expected, received } - } - _ => crate::Error::NoVersionInMessage, - }) + let payload = crate::lp_stream::maybe_unwrap_lp_stream_payload_from_reconstructed(message); + nym_ip_packet_requests::response_helpers::check_ipr_message_version(payload, CURRENT_VERSION) + .map_err(|e| match e { + IprResponseError::NoVersionByte => crate::Error::NoVersionInMessage, + IprResponseError::VersionMismatch { expected, received } if received < expected => { + crate::Error::ReceivedResponseWithOldVersion { expected, received } + } + IprResponseError::VersionMismatch { expected, received } => { + crate::Error::ReceivedResponseWithNewVersion { expected, received } + } + _ => crate::Error::NoVersionInMessage, + }) } diff --git a/nym-ip-packet-client/src/lib.rs b/nym-ip-packet-client/src/lib.rs index 0ab5c2eb24..40478fc650 100644 --- a/nym-ip-packet-client/src/lib.rs +++ b/nym-ip-packet-client/src/lib.rs @@ -5,6 +5,7 @@ mod connect; mod error; mod helpers; mod listener; +pub mod lp_stream; pub use connect::IprClientConnect; pub use error::Error; diff --git a/nym-ip-packet-client/src/lp_stream.rs b/nym-ip-packet-client/src/lp_stream.rs new file mode 100644 index 0000000000..821ef39d0d --- /dev/null +++ b/nym-ip-packet-client/src/lp_stream.rs @@ -0,0 +1,79 @@ +use bytes::BytesMut; +use nym_ip_packet_requests::SPHINX_STREAM_VERSION_THRESHOLD; +use nym_lp::packet::frame::{ + LpFrame, LpFrameHeader, LpFrameKind, SphinxStreamFrameAttributes, SphinxStreamMsgType, +}; +use nym_sdk::mixnet::ReconstructedMessage; + +/// Whether the "current" IPR client is operating at a version where the node expects +/// non-stream mixnet IPR messages to be LP Stream framed (see `SPHINX_STREAM_VERSION_THRESHOLD`). +pub(crate) fn current_requires_sphinx_stream_transport() -> bool { + crate::current::VERSION >= SPHINX_STREAM_VERSION_THRESHOLD +} + +pub fn maybe_unwrap_lp_stream_payload(data: &[u8]) -> &[u8] { + if data.len() < LpFrameHeader::SIZE { + return data; + } + let Ok(header) = LpFrameHeader::parse(data) else { + return data; + }; + if header.kind == LpFrameKind::SphinxStream { + &data[LpFrameHeader::SIZE..] + } else { + data + } +} + +pub fn maybe_unwrap_lp_stream_payload_from_reconstructed(message: &ReconstructedMessage) -> &[u8] { + maybe_unwrap_lp_stream_payload(&message.message) +} + +pub fn encode_stream_frame(stream_id: u64, sequence_num: u32, payload: Vec) -> Vec { + let attrs = SphinxStreamFrameAttributes { + stream_id, + msg_type: SphinxStreamMsgType::Data, + sequence_num, + }; + let frame = LpFrame::new_stream(attrs, payload); + let mut buf = BytesMut::with_capacity(LpFrameHeader::SIZE + frame.content.len()); + frame.encode(&mut buf); + buf.to_vec() +} + +#[cfg(test)] +mod tests { + use super::*; + use nym_lp::packet::frame::SphinxStreamFrameAttributes; + + #[test] + fn stream_frame_roundtrip_unwraps_payload() { + let stream_id = 0x0123_4567_89ab_cdef; + let seq = 42u32; + let payload = b"hello-ipr".to_vec(); + + let framed = encode_stream_frame(stream_id, seq, payload.clone()); + + let header = LpFrameHeader::parse(&framed).expect("valid lp header"); + assert_eq!(header.kind, LpFrameKind::SphinxStream); + + let attrs = + SphinxStreamFrameAttributes::parse(&header.frame_attributes).expect("valid attrs"); + assert_eq!(attrs.stream_id, stream_id); + assert_eq!(attrs.sequence_num, seq); + assert_eq!(attrs.msg_type, SphinxStreamMsgType::Data); + + let unwrapped = maybe_unwrap_lp_stream_payload(&framed); + assert_eq!(unwrapped, payload.as_slice()); + } + + #[test] + fn unwrap_noops_on_non_stream_or_malformed_data() { + let raw = b"\x09\x00\x01\x02\x03"; + assert_eq!(maybe_unwrap_lp_stream_payload(raw), raw); + + // malformed header: not enough bytes for LP header + let short = b"\x00\x01"; + assert_eq!(maybe_unwrap_lp_stream_payload(short), short); + } +} diff --git a/sdk/rust/nym-sdk/src/ipr_wrapper/ip_mix_stream.rs b/sdk/rust/nym-sdk/src/ipr_wrapper/ip_mix_stream.rs index 13cfb56e23..b4e75746e0 100644 --- a/sdk/rust/nym-sdk/src/ipr_wrapper/ip_mix_stream.rs +++ b/sdk/rust/nym-sdk/src/ipr_wrapper/ip_mix_stream.rs @@ -9,11 +9,9 @@ use crate::ip_packet_client::{ use crate::mixnet::{MixnetClient, MixnetStream, Recipient}; use crate::Error; use bytes::Bytes; +use current_ipr::response::IpPacketResponse; use nym_ip_packet_requests::response_helpers; -use nym_ip_packet_requests::{ - v9::{self, response::IpPacketResponse}, - IpPair, -}; +use nym_ip_packet_requests::{v9 as current_ipr, IpPair}; use nym_network_defaults::NymNetworkDetails; use std::time::Duration; use tokio::io::AsyncWriteExt; @@ -107,7 +105,7 @@ impl IpMixStream { } async fn connect_tunnel(stream: &mut MixnetStream) -> Result { - let (request, request_id) = v9::new_connect_request(None); + let (request, request_id) = current_ipr::new_connect_request(None); debug!("Sending connect request with ID: {}", request_id); let request_bytes = request.to_bytes()?; @@ -146,7 +144,7 @@ impl IpMixStream { /// Send an IP packet through the tunnel. pub async fn send_ip_packet(&mut self, packet: &[u8]) -> Result<(), Error> { self.check_connected()?; - let request = v9::new_data_request(packet.to_vec().into()); + let request = current_ipr::new_data_request(packet.to_vec().into()); let request_bytes = request.to_bytes()?; self.stream .write_all(&request_bytes) diff --git a/service-providers/ip-packet-router/src/mixnet_listener.rs b/service-providers/ip-packet-router/src/mixnet_listener.rs index fa25adbf71..3295dfeeb1 100644 --- a/service-providers/ip-packet-router/src/mixnet_listener.rs +++ b/service-providers/ip-packet-router/src/mixnet_listener.rs @@ -23,6 +23,7 @@ use crate::{ }; use futures::StreamExt; use nym_ip_packet_requests::codec::MultiIpPacketCodec; +use nym_ip_packet_requests::{MAX_NON_STREAM_VERSION, SPHINX_STREAM_VERSION_THRESHOLD}; use nym_lp::packet::frame::{LpFrameHeader, LpFrameKind, SphinxStreamFrameAttributes}; use nym_sdk::mixnet::MixnetMessageSender; use nym_sphinx::receiver::ReconstructedMessage; @@ -559,8 +560,9 @@ impl MixnetListener { /// /// # Version / transport enforcement /// - /// - LP Stream frames (`stream_id` is `Some`) **must** carry v9+ payloads. - /// - Non-stream messages (`stream_id` is `None`) **must** be v8 or lower. + /// - LP Stream frames (`stream_id` is `Some`) **must** carry payloads with version + /// `>= SPHINX_STREAM_VERSION_THRESHOLD` (see `nym_ip_packet_requests`). + /// - Non-stream messages (`stream_id` is `None`) **must** be `<= MAX_NON_STREAM_VERSION`. /// /// Messages that violate these rules are dropped. async fn on_ipr_message( @@ -578,16 +580,22 @@ impl MixnetListener { }?; // Enforce version/transport consistency: - // - LP Stream frames must carry v9+ payloads - // - Non-stream messages must be v8 or lower + // - LP Stream frames must carry payloads at/above the SphinxStream threshold + // - Non-stream messages must be at/below the max non-stream version let version_num = request.version().into_u8(); - if stream_id.is_some() && version_num < 9 { - log::warn!("LP Stream frame contains v{version_num} payload, expected v9+; dropping",); + if stream_id.is_some() && version_num < SPHINX_STREAM_VERSION_THRESHOLD { + log::warn!( + "LP Stream frame contains v{version_num} payload, expected v{expected}+; dropping", + expected = SPHINX_STREAM_VERSION_THRESHOLD + ); return Ok(vec![]); } - if stream_id.is_none() && version_num >= 9 { - log::warn!("Non-stream message claims v{version_num}, expected v8 or lower; dropping",); + if stream_id.is_none() && version_num > MAX_NON_STREAM_VERSION { + log::warn!( + "Non-stream message claims v{version_num}, expected v{expected} or lower; dropping", + expected = MAX_NON_STREAM_VERSION + ); return Ok(vec![]); }