diff --git a/common/client-core/src/client/real_messages_control/real_traffic_stream.rs b/common/client-core/src/client/real_messages_control/real_traffic_stream.rs index 07b3506c45..59e233a8be 100644 --- a/common/client-core/src/client/real_messages_control/real_traffic_stream.rs +++ b/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,13 @@ where tracing::error!( "failed to send mixnet packet due to closed channel (outside of shutdown!)" ); + // Set the flag to break out of the main loop + // This prevents an 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 +611,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 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 +636,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;