From 715648d6bfe15e438ed98b127d2e3e414bdfe3cd Mon Sep 17 00:00:00 2001 From: Nicolas Luck Date: Fri, 5 Dec 2025 18:54:12 +0100 Subject: [PATCH 1/4] fix: add timeout parameter to wait_for_ready() to prevent indefinite blocking Previously, wait_for_ready() would block indefinitely on a semaphore when waiting for peer connections. This caused threads to hang forever when connections failed, leading to cascading failures. Changes: - Add timeout parameter to MaybeReady::wait_for_ready() - Wrap semaphore acquire in tokio::time::timeout - Update Peer::wait_for_ready() signature to accept timeout parameter - Update all call sites to pass config.timeout This prevents threads from blocking indefinitely on failed connections. --- crates/tx5/src/ep.rs | 2 +- crates/tx5/src/peer.rs | 7 ++-- crates/tx5/src/peer/maybe_ready.rs | 54 +++++++++++++++++++++++------- 3 files changed, 47 insertions(+), 16 deletions(-) diff --git a/crates/tx5/src/ep.rs b/crates/tx5/src/ep.rs index 389c69e..824310c 100644 --- a/crates/tx5/src/ep.rs +++ b/crates/tx5/src/ep.rs @@ -286,7 +286,7 @@ impl Endpoint { let data = data.to_vec(); tokio::task::spawn(async move { let _ = tokio::time::timeout(timeout, async { - if let Some(conn) = peer.wait_for_ready().await { + if let Some(conn) = peer.wait_for_ready(timeout).await { let _ = conn.send(data).await; } }) diff --git a/crates/tx5/src/peer.rs b/crates/tx5/src/peer.rs index c5c5b2f..2a5e8ea 100644 --- a/crates/tx5/src/peer.rs +++ b/crates/tx5/src/peer.rs @@ -120,8 +120,11 @@ impl Peer { /// This future resolves when the connection is ready to use or has failed to connect. /// /// If the connection is not usable, it will return `None`. - pub async fn wait_for_ready(&self) -> Option { - self.ready.wait_for_ready().await + pub async fn wait_for_ready( + &self, + timeout: std::time::Duration, + ) -> Option { + self.ready.wait_for_ready(timeout).await } } diff --git a/crates/tx5/src/peer/maybe_ready.rs b/crates/tx5/src/peer/maybe_ready.rs index b4dd2e0..db9f5af 100644 --- a/crates/tx5/src/peer/maybe_ready.rs +++ b/crates/tx5/src/peer/maybe_ready.rs @@ -74,6 +74,10 @@ impl MaybeReady { } pub(super) async fn wait_for_ready(&self) -> Option { + pub(super) async fn wait_for_ready( + &self, + timeout: std::time::Duration, + ) -> Option { let wait = { let lock = self.0.lock().expect("poisoned"); match &*lock { @@ -83,15 +87,29 @@ impl MaybeReady { } }; - let _ = wait.acquire().await; - - let lock = self.0.lock().expect("poisoned"); - match &*lock { - MaybeReadyState::Ready(back) => Some(back.clone()), - MaybeReadyState::Failed => None, - MaybeReadyState::Wait(_) => { - // Unexpected state, this struct is supposed to prevent this being possible. - tracing::warn!("Waited for ready but still in a wait state"); + // Add timeout wrapper to prevent indefinite blocking + let result = tokio::time::timeout(timeout, async { + let _ = wait.acquire().await; + }) + .await; + + match result { + Ok(_) => { + // Semaphore acquired, check final state + let lock = self.0.lock().expect("poisoned"); + match &*lock { + MaybeReadyState::Ready(back) => Some(back.clone()), + MaybeReadyState::Failed => None, + MaybeReadyState::Wait(_) => { + // Unexpected state, this struct is supposed to prevent this being possible. + tracing::warn!("Waited for ready but still in a wait state"); + None + } + } + } + Err(_) => { + // Timeout - connection failed to become ready in time + tracing::debug!("wait_for_ready timed out after {:?}", timeout); None } } @@ -136,8 +154,13 @@ mod tests { tokio::spawn({ let maybe_ready = maybe_ready.clone(); async move { - tx.send(maybe_ready.wait_for_ready().await.is_some()) - .unwrap(); + tx.send( + maybe_ready + .wait_for_ready(std::time::Duration::from_secs(60)) + .await + .is_some(), + ) + .unwrap(); } }); @@ -178,8 +201,13 @@ mod tests { tokio::spawn({ let maybe_ready = maybe_ready.clone(); async move { - tx.send(maybe_ready.wait_for_ready().await.is_none()) - .unwrap(); + tx.send( + maybe_ready + .wait_for_ready(std::time::Duration::from_secs(60)) + .await + .is_none(), + ) + .unwrap(); } }); From 2a92be6e60d97603011ade78bbfa1d4739ec4ca3 Mon Sep 17 00:00:00 2001 From: Nicolas Luck Date: Fri, 5 Dec 2025 18:58:30 +0100 Subject: [PATCH 2/4] feat: add is_failed() methods to detect failed peer connections Add methods to check if a peer connection is in a failed state, enabling proactive cleanup of zombie connections before reuse attempts. Changes: - Add MaybeReady::is_failed() to check MaybeReadyState::Failed - Add Peer::is_failed() wrapper method - Enables detection of stale connections in peer_map --- crates/tx5/src/peer.rs | 5 +++++ crates/tx5/src/peer/maybe_ready.rs | 7 ++++++- 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/crates/tx5/src/peer.rs b/crates/tx5/src/peer.rs index 2a5e8ea..0fe97b3 100644 --- a/crates/tx5/src/peer.rs +++ b/crates/tx5/src/peer.rs @@ -117,6 +117,11 @@ impl Peer { self.ready.query_ready(|c| c.clone()) } + /// Check if this peer connection has failed + pub(crate) fn is_failed(&self) -> bool { + self.ready.is_failed() + } + /// This future resolves when the connection is ready to use or has failed to connect. /// /// If the connection is not usable, it will return `None`. diff --git a/crates/tx5/src/peer/maybe_ready.rs b/crates/tx5/src/peer/maybe_ready.rs index db9f5af..5eed332 100644 --- a/crates/tx5/src/peer/maybe_ready.rs +++ b/crates/tx5/src/peer/maybe_ready.rs @@ -73,7 +73,12 @@ impl MaybeReady { *lock = MaybeReadyState::Failed; } - pub(super) async fn wait_for_ready(&self) -> Option { + /// Check if the connection is in a failed state + pub(super) fn is_failed(&self) -> bool { + let lock = self.0.lock().expect("poisoned"); + matches!(*lock, MaybeReadyState::Failed) + } + pub(super) async fn wait_for_ready( &self, timeout: std::time::Duration, From 2b2dc39a17a8dd6e5a48ec2dc624effdd8c36770 Mon Sep 17 00:00:00 2001 From: Nicolas Luck Date: Fri, 5 Dec 2025 18:59:11 +0100 Subject: [PATCH 3/4] fix: fix event channel capacity bug and increase channel sizes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Critical fix: endpoint event channel was hardcoded to 32 slots despite config specifying internal_event_channel_size=1024. This caused channel saturation under concurrent load, resulting in "closed" errors. Changes: - ep.rs: Use config.internal_event_channel_size (32→1024, 32x increase) - hub.rs: Increase hub command channel from 32→256 (8x increase) - framed.rs: Increase cmd and msg channels from 32→256 (8x increase) These increases prevent channel saturation during concurrent connection establishment and high message throughput. --- crates/tx5-connection/src/framed.rs | 4 ++-- crates/tx5-connection/src/hub.rs | 2 +- crates/tx5/src/ep.rs | 4 +++- 3 files changed, 6 insertions(+), 4 deletions(-) diff --git a/crates/tx5-connection/src/framed.rs b/crates/tx5-connection/src/framed.rs index db6175c..ebccaf8 100644 --- a/crates/tx5-connection/src/framed.rs +++ b/crates/tx5-connection/src/framed.rs @@ -61,8 +61,8 @@ impl FramedConn { let (a, b, c, d) = crate::proto::PROTO_VER_2.encode()?; conn.send(vec![a, b, c, d]).await?; - let (cmd_send, mut cmd_recv) = tokio::sync::mpsc::channel(32); - let (msg_send, msg_recv) = tokio::sync::mpsc::channel(32); + let (cmd_send, mut cmd_recv) = tokio::sync::mpsc::channel(256); + let (msg_send, msg_recv) = tokio::sync::mpsc::channel(256); // set up the receive to just feed straight into the cmd task let cmd_send2 = cmd_send.clone(); diff --git a/crates/tx5-connection/src/hub.rs b/crates/tx5-connection/src/hub.rs index 38ec486..ede2645 100644 --- a/crates/tx5-connection/src/hub.rs +++ b/crates/tx5-connection/src/hub.rs @@ -139,7 +139,7 @@ impl Hub { let mut task_list = Vec::new(); - let (hub_cmd_send, mut cmd_recv) = tokio::sync::mpsc::channel(32); + let (hub_cmd_send, mut cmd_recv) = tokio::sync::mpsc::channel(256); // forward received messages to the cmd task let hub_cmd_send2 = hub_cmd_send.clone(); diff --git a/crates/tx5/src/ep.rs b/crates/tx5/src/ep.rs index 824310c..ce93f43 100644 --- a/crates/tx5/src/ep.rs +++ b/crates/tx5/src/ep.rs @@ -199,7 +199,9 @@ impl Endpoint { config.incoming_message_bytes_max as usize, )); - let (evt_send, evt_recv) = tokio::sync::mpsc::channel(32); + let (evt_send, evt_recv) = tokio::sync::mpsc::channel( + config.internal_event_channel_size as usize, + ); ( Self { From 0c9d81fe7e6634370758b539db884601430a6b15 Mon Sep 17 00:00:00 2001 From: Nicolas Luck Date: Fri, 5 Dec 2025 19:12:20 +0100 Subject: [PATCH 4/4] fix: add aggressive peer cleanup to prevent zombie connection reuse Prevent failed connections from remaining in peer_map and being reused by subsequent sends, which would cause them to timeout repeatedly. Changes: - Add stale peer check in connect_peer() - remove failed peers before reuse - Add early cleanup in peer task() when connection fails during negotiation - Add cleanup in send() when wait_for_ready() times out - Add debug logging for peer lifecycle events This prevents the pattern where a failed connection stays in peer_map, gets reused by subsequent sends, and times out repeatedly (20s each time). --- crates/tx5/src/ep.rs | 10 +++++++++- crates/tx5/src/peer.rs | 12 ++++++++++-- 2 files changed, 19 insertions(+), 3 deletions(-) diff --git a/crates/tx5/src/ep.rs b/crates/tx5/src/ep.rs index ce93f43..f34fcc5 100644 --- a/crates/tx5/src/ep.rs +++ b/crates/tx5/src/ep.rs @@ -139,6 +139,12 @@ impl EpInner { /// Get an existing peer connection or create a new outgoing one. pub fn connect_peer(&mut self, peer_url: PeerUrl) -> Arc { + // Check for stale failed peers before attempting reuse + if let Some(existing_peer) = self.peer_map.get(&peer_url) { + if existing_peer.is_failed() { + self.peer_map.remove(&peer_url); + } + } self.peer_map .entry(peer_url.clone()) .or_insert_with(|| { @@ -252,9 +258,11 @@ impl Endpoint { } tokio::time::timeout(self.config.timeout, async { let peer = self.inner.lock().unwrap().connect_peer(peer_url.clone()); - if let Some(conn) = peer.wait_for_ready().await { + if let Some(conn) = peer.wait_for_ready(self.config.timeout).await { conn.send(data).await } else { + // Remove the failed peer from peer_map to prevent reuse of zombie connections + self.inner.lock().unwrap().drop_peer_url(&peer_url); // Check to see if a new incoming connection was established // in the mean time. If so, we can send over that. let maybe_peer = self.inner.lock().unwrap().peer_if_ready( diff --git a/crates/tx5/src/peer.rs b/crates/tx5/src/peer.rs index 0fe97b3..68c4733 100644 --- a/crates/tx5/src/peer.rs +++ b/crates/tx5/src/peer.rs @@ -249,14 +249,22 @@ async fn task( ) { // establish our cleanup drop guard let drop_guard = DropPeer { - ep, + ep: ep.clone(), peer_url: peer_url.clone(), evt_send: evt_send.clone(), ready: ready.clone(), }; let mut wc = match conn { - None => return, + None => { + // Connection failed during negotiation - clean up immediately + // to prevent zombie connection reuse + if let Some(ep_inner) = ep.upgrade() { + ep_inner.lock().unwrap().drop_peer_url(&peer_url); + } + ready.set_failed(); + return; + } Some(wc) => wc, };