smp-server: drop relay session after timeouts

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.

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)

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