From 85be73401ea9afc6723ec12d5134bf110f207f41 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bogdan-=C8=98tefan=20Neac=C5=9Fu?= Date: Fri, 31 Jan 2025 16:36:23 +0100 Subject: [PATCH] Add debug timeout on sink --- common/client-libs/gateway-client/src/error.rs | 3 +++ .../client-libs/gateway-client/src/socket_state.rs | 13 +++++++++++-- 2 files changed, 14 insertions(+), 2 deletions(-) diff --git a/common/client-libs/gateway-client/src/error.rs b/common/client-libs/gateway-client/src/error.rs index e1b581d30e..1769744782 100644 --- a/common/client-libs/gateway-client/src/error.rs +++ b/common/client-libs/gateway-client/src/error.rs @@ -118,6 +118,9 @@ pub enum GatewayClientError { "this operation couldn't be completed as the program is in the process of shutting down" )] ShutdownInProgress, + + #[error("Timed out on sending message(s)")] + TimeoutOnSink, } impl GatewayClientError { diff --git a/common/client-libs/gateway-client/src/socket_state.rs b/common/client-libs/gateway-client/src/socket_state.rs index 319a7bb624..85f5b1eff7 100644 --- a/common/client-libs/gateway-client/src/socket_state.rs +++ b/common/client-libs/gateway-client/src/socket_state.rs @@ -291,7 +291,10 @@ impl PartiallyDelegatedHandle { &mut self, msg: Message, ) -> Result<(), GatewayClientError> { - Ok(self.sink_half.send(msg).await?) + tokio::time::timeout(std::time::Duration::from_secs(1), self.sink_half.send(msg)) + .await + .map_err(|_| GatewayClientError::TimeoutOnSink)??; + Ok(()) } pub(crate) async fn batch_send_without_response( @@ -300,7 +303,13 @@ impl PartiallyDelegatedHandle { ) -> Result<(), GatewayClientError> { let stream_messages: Vec<_> = messages.into_iter().map(Ok).collect(); let mut send_stream = futures::stream::iter(stream_messages); - Ok(self.sink_half.send_all(&mut send_stream).await?) + tokio::time::timeout( + std::time::Duration::from_secs(1), + self.sink_half.send_all(&mut send_stream), + ) + .await + .map_err(|_| GatewayClientError::TimeoutOnSink)??; + Ok(()) } pub(crate) async fn merge(self) -> Result {