Better message to come in the PR description

common/client-core/src/client/mix_traffic/mod.rs

@@ -20,7 +20,10 @@ pub mod transceiver;
 
 // We remind ourselves that 32 x 32kb = 1024kb, a reasonable size for a network buffer.
 pub const MIX_MESSAGE_RECEIVER_BUFFER_SIZE: usize = 32;
-const MAX_FAILURE_COUNT: usize = 100;
+
+/// Reduced from 100 to 20 to fail fast (~1-2 seconds instead of ~6 seconds).
+/// If we can't send 20 packets in a row, the gateway is unreachable.
+const MAX_FAILURE_COUNT: usize = 20;
 
 // that's also disgusting.
 pub struct Empty;

common/client-core/src/client/real_messages_control/real_traffic_stream.rs

@@ -120,6 +120,9 @@ where
     stats_tx: ClientStatsSender,
 
     shutdown_token: ShutdownToken,
+
+    /// Flag to indicate that the mix_tx channel is closed and we should stop processing
+    mix_tx_closed: bool,
 }
 
 #[derive(Debug)]
@@ -195,6 +198,7 @@ where
             lane_queue_lengths,
             stats_tx,
             shutdown_token,
+            mix_tx_closed: false,
         }
     }
 
@@ -297,7 +301,12 @@ where
                     tracing::error!(
                         "failed to send mixnet packet due to closed channel (outside of shutdown!)"
                     );
+                    // This prevents an infinite loop where we keep trying to send
+                    // packets through a closed channel
+                    self.mix_tx_closed = true;
                 }
+                // Early return to avoid further processing when channel is closed
+                return;
             }
             Ok(_) => {
                 let event = if fragment_id.is_some() {
@@ -601,6 +610,12 @@ where
                     }
                     next_message = self.next() => if let Some(next_message) = next_message {
                         self.on_message(next_message).await;
+                        // Check if mix_tx channel was closed during on_message
+                        // and break immediately to prevent infinite loop
+                        if self.mix_tx_closed {
+                            tracing::error!("OutQueueControl: mix_tx channel closed, stopping traffic stream");
+                            break;
+                        }
                     } else {
                         tracing::trace!("OutQueueControl: Stopping since channel closed");
                         break;
@@ -620,6 +635,12 @@ where
                     }
                     next_message = self.next() => if let Some(next_message) = next_message {
                         self.on_message(next_message).await;
+                        // Check if mix_tx channel was closed during on_message
+                        // and break immediately to prevent infinite loop
+                        if self.mix_tx_closed {
+                            tracing::error!("OutQueueControl: mix_tx channel closed, stopping traffic stream");
+                            break;
+                        }
                     } else {
                         tracing::trace!("OutQueueControl: Stopping since channel closed");
                         break;