diff --git a/src/Simplex/Messaging/Client.hs b/src/Simplex/Messaging/Client.hs index 9cc0d0335..b24f57099 100644 --- a/src/Simplex/Messaging/Client.hs +++ b/src/Simplex/Messaging/Client.hs @@ -35,6 +35,7 @@ module Simplex.Messaging.Client ProxiedRelay (..), getProtocolClient, closeProtocolClient, + closeTimedOutClient, pClientSentCommandsCount, protocolClientServer, protocolClientServer', @@ -660,10 +661,9 @@ getProtocolClient g nm transportSession@(_, srv, _) cfg@ProtocolClientConfig {qS responseErr = atomically . putTMVar responseVar . Left . PCETransportError receive :: Transport c => ProtocolClient v err msg -> THandle v c 'TClient -> IO () - receive ProtocolClient {client_ = PClient {rcvQ, lastReceived, timeoutErrorCount}} h = forever $ do + receive ProtocolClient {client_ = PClient {rcvQ, lastReceived}} h = forever $ do tGetClient h >>= atomically . writeTBQueue rcvQ getCurrentTime >>= atomically . writeTVar lastReceived - atomically $ writeTVar timeoutErrorCount 0 monitor :: ProtocolClient v err msg -> IO () monitor c@ProtocolClient {client_ = PClient {sendPings, lastReceived, timeoutErrorCount}} = loop smpPingInterval @@ -756,6 +756,12 @@ closeProtocolClient :: ProtocolClient v err msg -> IO () closeProtocolClient = mapM_ (deRefWeak >=> mapM_ killThread) . action {-# INLINE closeProtocolClient #-} +-- | Disconnects client when maxCnt commands in a row timed out, 0 to disable. +closeTimedOutClient :: Int -> ProtocolClient v err msg -> IO () +closeTimedOutClient maxCnt c@ProtocolClient {client_ = PClient {timeoutErrorCount}} = do + cnt <- readTVarIO timeoutErrorCount + when (maxCnt > 0 && cnt >= maxCnt) $ closeProtocolClient c + -- | SMP client error type. data ProtocolClientError err = -- | Correctly parsed SMP server ERR response. diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index 2c890e572..aac520b6f 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/src/Simplex/Messaging/Server.hs @@ -98,8 +98,8 @@ import Network.Socket (ServiceName, Socket, socketToHandle) import qualified Network.TLS as TLS import Numeric.Natural (Natural) import Simplex.Messaging.Agent.Lock -import Simplex.Messaging.Client (ProtocolClient (thParams), ProtocolClientError (..), SMPClient, SMPClientError, clientHandlers, forwardSMPTransmission, smpProxyError, temporaryClientError, transportHost') -import Simplex.Messaging.Client.Agent (AgentLeakStats (..), OwnServer, SMPClientAgent (..), SMPClientAgentEvent (..), closeSMPClientAgent, getAgentLeakStats, getSMPServerClient'', isOwnServer, lookupSMPServerClient, getConnectedSMPServerClient) +import Simplex.Messaging.Client (NetworkConfig (..), ProtocolClient (thParams), ProtocolClientConfig (..), ProtocolClientError (..), SMPClient, SMPClientError, clientHandlers, closeTimedOutClient, forwardSMPTransmission, smpProxyError, temporaryClientError, transportHost') +import Simplex.Messaging.Client.Agent (AgentLeakStats (..), OwnServer, SMPClientAgent (..), SMPClientAgentConfig (..), SMPClientAgentEvent (..), closeSMPClientAgent, getAgentLeakStats, getSMPServerClient'', isOwnServer, lookupSMPServerClient, getConnectedSMPServerClient) import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Encoding import Simplex.Messaging.Encoding.String @@ -1583,6 +1583,9 @@ client _ -> do logWarn $ "Error forwarding to relay: " <> decodeLatin1 (strEncode $ transportHost' smp) <> " own=" <> tshow own <> " " <> tshow e inc own pErrorsOther + case e of + PCEResponseTimeout -> liftIO $ closeTimedOutClient (smpPingCount $ networkConfig $ smpCfg $ agentCfg a) smp + _ -> pure () Nothing -> inc False pRequests >> inc False pErrorsConnect $> Just (ERR $ PROXY NO_SESSION) where forkProxiedCmd :: M s BrokerMsg -> M s (Maybe BrokerMsg) diff --git a/tests/SMPProxyTests.hs b/tests/SMPProxyTests.hs index 4ab300c60..a332a1a54 100644 --- a/tests/SMPProxyTests.hs +++ b/tests/SMPProxyTests.hs @@ -64,7 +64,7 @@ smpProxyTests = do testProxyRecoversWithoutDisconnect it "reconnects to relay after sender disconnects mid-connection" $ \_ -> testProxyReconnectAfterRelayRestart - xit "must drop a stuck relay session after forward timeouts" $ \_ -> + it "must drop a stuck relay session after forward timeouts" $ \_ -> testProxyForwardTimeoutStuckSession it "does not keep oversized forwarded command" $ \_ -> testForwardOversizedNotKept