Merge branch 'release/2026.11-xynomizithra' into develop

nym-node/src/config/mod.rs

@@ -693,7 +693,12 @@ pub struct MixnetDebug {
     #[serde(with = "humantime_serde")]
     pub initial_connection_timeout: Duration,
 
-    /// Maximum number of packets that can be stored waiting to get sent to a particular connection.
+    /// Maximum number of packets buffered per egress connection awaiting a socket write.
+    /// This is a short-term burst absorber, not a queue: buffer depth converts directly into
+    /// added latency (roughly `depth / per-peer send rate`), so an oversized value is just
+    /// bufferbloat. Once it fills, further packets for that peer are dropped rather than
+    /// delayed, which is preferable in a mixnet where a packet held that long has already
+    /// missed its usefulness window. Keep worst-case queuing well under the per-hop mix delay.
     pub maximum_connection_buffer_size: usize,
 
     /// Specify whether any framed packets between nodes should use the legacy format (v7)
@@ -882,9 +887,11 @@ impl MixnetDebug {
     // which for all intents and purposes will never happen
     const DEFAULT_MAXIMUM_FORWARD_PACKET_DELAY: Duration = Duration::from_secs(10);
     const DEFAULT_PACKET_FORWARDING_INITIAL_BACKOFF: Duration = Duration::from_millis(10_000);
-    const DEFAULT_PACKET_FORWARDING_MAXIMUM_BACKOFF: Duration = Duration::from_millis(300_000);
+    const DEFAULT_PACKET_FORWARDING_MAXIMUM_BACKOFF: Duration = Duration::from_secs(16);
     const DEFAULT_INITIAL_CONNECTION_TIMEOUT: Duration = Duration::from_millis(1_500);
-    const DEFAULT_MAXIMUM_CONNECTION_BUFFER_SIZE: usize = 2000;
+    // small enough to keep worst-case egress queuing in the tens-of-ms range at a few thousand
+    // pps per peer (vs. the old 2000, which was hundreds of ms of bufferbloat)
+    const DEFAULT_MAXIMUM_CONNECTION_BUFFER_SIZE: usize = 192;
 }
 
 impl Default for MixnetDebug {

nym-node/src/node/lp/control/ingress/client_handler.rs

@@ -404,7 +404,7 @@ where
         };
 
         // Connect to target gateway with timeout
-        let stream = match timeout(Duration::from_secs(5), S::connect(target_addr)).await {
+        let mut stream = match timeout(Duration::from_secs(5), S::connect(target_addr)).await {
             Ok(Ok(stream)) => stream,
             Ok(Err(e)) => {
                 inc!("lp_forward_failed");
@@ -422,6 +422,16 @@ where
             }
         };
 
+        // Disable Nagle's algorithm: the forward stream carries small request/response
+        // handshake packets, so we want them sent immediately rather than coalesced.
+        if let Err(e) = stream.set_no_delay(true) {
+            inc!("lp_forward_failed");
+            return Err(LpHandlerError::ConnectionFailure {
+                egress: target_addr,
+                reason: format!("failed to set TCP_NODELAY: {e}"),
+            });
+        }
+
         debug!("Opened persistent exit connection to {target_addr} for forwarding");
         self.exit_stream = Some((stream, target_addr));
 

nym-node/src/node/lp/control/ingress/listener.rs

@@ -171,6 +171,14 @@ impl LpControlListener {
     }
 
     fn handle_connection(&self, stream: tokio::net::TcpStream, remote_addr: SocketAddr) {
+        // Disable Nagle's algorithm on the accepted socket so our responses are flushed
+        // immediately rather than coalesced. This is the write side of every reply we send,
+        // including handshake replies forwarded back to an entry gateway. Non-fatal: a valid
+        // connection should still be served if the option can't be set.
+        if let Err(e) = stream.set_nodelay(true) {
+            warn!("failed to set TCP_NODELAY on accepted LP connection from {remote_addr}: {e}");
+        }
+
         if let Some(initiator_details) = self
             .nodes_handler_state
             .nodes

nym-node/src/node/mixnet/handler.rs

@@ -18,6 +18,7 @@ use nym_sphinx_types::{Delay, REPLAY_TAG_SIZE};
 use std::collections::HashMap;
 use std::mem;
 use std::net::SocketAddr;
+use std::time::Duration;
 use tokio::net::TcpStream;
 use tokio::time::Instant;
 use tokio_util::codec::Framed;
@@ -59,6 +60,11 @@ impl PendingReplayCheckPackets {
         self.packets.values().map(|v| v.len()).sum()
     }
 
+    /// Instant at which the currently-deferred batch must be flushed, or `None` if nothing is pending.
+    fn flush_deadline(&self, deferral: Duration) -> Option<Instant> {
+        (self.total_count() > 0).then(|| self.last_acquired_mutex + deferral)
+    }
+
     fn replay_tags(&self) -> HashMap<u32, Vec<&[u8; REPLAY_TAG_SIZE]>> {
         let mut replay_tags = HashMap::with_capacity(self.packets.len());
         'outer: for (rotation_id, packets) in &self.packets {
@@ -289,7 +295,7 @@ impl ConnectionHandler {
                 .processing_config
                 .maximum_replay_detection_deferral;
 
-        let count_threshold = self.pending_packets.packets.len()
+        let count_threshold = self.pending_packets.total_count()
             < self
                 .shared
                 .processing_config
@@ -706,13 +712,23 @@ impl ConnectionHandler {
     ) {
         let mut packets_processed: u64 = 0;
         loop {
+            // make sure pending packets are not stuck in the queue if we don't get any more packets
+            // from this sender
+            let flush_deadline = self.pending_packets.flush_deadline(
+                self.shared
+                    .processing_config
+                    .maximum_replay_detection_deferral,
+            );
+
             tokio::select! {
                 biased;
+                // 1. check for cancellation
                 _ = self.shared.shutdown_token.cancelled() => {
                     trace!("connection handler: received shutdown");
                     Span::current().record("exit_reason", "shutdown");
                     break
                 }
+                // 2. handle any incoming packet
                 maybe_framed_nym_packet = mixnet_connection.next() => {
                     match maybe_framed_nym_packet {
                         Some(Ok(packet)) => {
@@ -732,7 +748,7 @@ impl ConnectionHandler {
                             );
                             Span::current().record("exit_reason", "corrupted");
                             Span::current().record("packets_processed", packets_processed);
-                            return
+                            break
                         }
                         None => {
                             debug!(
@@ -742,14 +758,101 @@ impl ConnectionHandler {
                             );
                             Span::current().record("exit_reason", "closed_by_remote");
                             Span::current().record("packets_processed", packets_processed);
-                            return
+                            break
                         }
                     }
                 }
+                // 3. check for the deferred pending packets
+                _ = async move {
+                    match flush_deadline {
+                        Some(d) => tokio::time::sleep_until(d).await,
+                        None => std::future::pending::<()>().await,
+                    }
+                } => {
+                    self.handle_pending_packets_batch(Instant::now()).await;
+                }
             }
         }
 
+        // drain any packets still deferred for replay-checking so they are forwarded
+        // rather than silently dropped when the connection closes, errors, or shuts down
+        self.handle_pending_packets_batch(Instant::now()).await;
+
         Span::current().record("packets_processed", packets_processed);
         debug!("exiting and closing connection");
     }
 }
+
+#[cfg(test)]
+#[allow(clippy::unwrap_used, clippy::expect_used)]
+mod tests {
+    use super::*;
+    use nym_sphinx_params::{PacketSize, PacketType};
+    use nym_sphinx_types::{
+        DESTINATION_ADDRESS_LENGTH, Destination, DestinationAddressBytes, IDENTIFIER_LENGTH,
+        NODE_ADDRESS_LENGTH, Node, NodeAddressBytes, NymPacket, PrivateKey, PublicKey,
+    };
+
+    fn random_pubkey() -> PublicKey {
+        (&PrivateKey::random()).into()
+    }
+
+    // Build a real sphinx packet whose first hop validates against `key`, then partially
+    // unwrap it - enough to land one entry in the pending replay-check batch.
+    fn pending_packet(key: &PrivateKey) -> PartialyUnwrappedPacketWithKeyRotation {
+        let route = [
+            Node::new(
+                NodeAddressBytes::from_bytes([1u8; NODE_ADDRESS_LENGTH]),
+                key.into(),
+            ),
+            Node::new(
+                NodeAddressBytes::from_bytes([2u8; NODE_ADDRESS_LENGTH]),
+                random_pubkey(),
+            ),
+        ];
+        let destination = Destination::new(
+            DestinationAddressBytes::from_bytes([3u8; DESTINATION_ADDRESS_LENGTH]),
+            [4u8; IDENTIFIER_LENGTH],
+        );
+        let delays: Vec<Delay> = std::iter::repeat_with(|| Delay::new_from_nanos(0))
+            .take(route.len())
+            .collect();
+        let packet = NymPacket::sphinx_build(
+            true,
+            PacketSize::RegularPacket.payload_size(),
+            b"x",
+            &route,
+            &destination,
+            &delays,
+        )
+        .expect("failed to build test sphinx packet");
+        let framed =
+            FramedNymPacket::new(packet, PacketType::Mix, SphinxKeyRotation::Unknown, true);
+
+        PartiallyUnwrappedPacket::new(framed, key)
+            .map_err(|(_, err)| err)
+            .expect("failed to partially unwrap test packet")
+            .with_key_rotation(0)
+    }
+
+    #[test]
+    fn no_flush_deadline_when_nothing_pending() {
+        let pending = PendingReplayCheckPackets::new();
+        assert!(pending.flush_deadline(Duration::from_millis(50)).is_none());
+    }
+
+    #[test]
+    fn flush_deadline_is_batch_start_plus_deferral() {
+        let key = PrivateKey::random();
+        let mut pending = PendingReplayCheckPackets::new();
+
+        let batch_start = Instant::now();
+        pending.push(batch_start, pending_packet(&key));
+
+        let deferral = Duration::from_millis(50);
+        assert_eq!(
+            pending.flush_deadline(deferral),
+            Some(batch_start + deferral)
+        );
+    }
+}

nym-node/src/node/mixnet/shared/mod.rs

@@ -24,7 +24,7 @@ use std::time::Duration;
 use tokio::net::TcpStream;
 use tokio::task::JoinHandle;
 use tokio::time::Instant;
-use tracing::{debug, error};
+use tracing::{debug, error, warn};
 
 pub(crate) mod final_hop;
 
@@ -192,6 +192,12 @@ impl SharedData {
         match accepted {
             Ok((socket, remote_addr)) => {
                 debug!("accepted incoming mixnet connection from: {remote_addr}");
+                // disable Nagle: mix packets are latency-sensitive and flushed one at a time.
+                if let Err(err) = socket.set_nodelay(true) {
+                    warn!(
+                        "failed to set TCP_NODELAY on mixnet connection from {remote_addr}: {err}"
+                    );
+                }
                 let mut handler = ConnectionHandler::new(self, remote_addr);
                 let join_handle =
                     tokio::spawn(async move { handler.handle_connection(socket).await });