From 54995ac2f8a48778d90180ce5d8537d38fd3824c Mon Sep 17 00:00:00 2001 From: Narasimha-sc <166327228+Narasimha-sc@users.noreply.github.com> Date: Mon, 19 Oct 2026 00:10:06 +0000 Subject: [PATCH] agent: choose another proxy when the chosen proxy keeps failing The proxy chosen for a destination relay is kept in smpProxiedRelays and reused on every retry. It was forgotten only when the proxy could not create a session to the relay (other than a host error), when the proxy connection was closed, or when the agent was closed. Two failures kept it: - connecting to the proxy itself fails (proxy offline): with no proxy connection to close, retries used this proxy until it was reachable again or the app restarted; - a forwarded command fails with PROXY (BROKER e) other than a host error (e.g. TIMEOUT: the proxy has a session to the relay but cannot reach it): retries used this proxy while its connection stayed up. A failed connection to the proxy now forgets it at once: without a live connection the proxy normally has not received this destination, so choosing again adds no privacy cost. A request that only waits for another request's connection does not forget the proxy when its wait times out. A forwarded command failing with PROXY (BROKER e) other than a host error starts a failure streak, stored next to the chosen proxy; a successful command, an error from the destination relay returned through the proxy, or PROXY (PROTOCOL e), the relay's error to the proxy, clears it. A failure that comes at least proxyFailureInterval (30 s) after the start of the streak forgets the proxy, so a single error or a burst of errors keeps it as long as a later forward reaches the relay. The streak functions are exported for tests. Failures, successes and deletes change the entry only while it still names that proxy, also when the proxy disconnects or its relay session fails, so that they do not change a new choice. After PROXY NO_SESSION, the dead relay session is removed from the proxy client that handled the request, so that the retry through the same proxy does not use it again, and that retry no longer records the proxy as the choice when there is none, so that a concurrently forgotten proxy is not put back. --- src/Simplex/Messaging/Agent/Client.hs | 79 ++++++++++++++--------- src/Simplex/Messaging/Agent/Env/SQLite.hs | 2 + 2 files changed, 52 insertions(+), 29 deletions(-) diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs index 08eb64ea9..34fae881b 100644 --- a/src/Simplex/Messaging/Agent/Client.hs +++ b/src/Simplex/Messaging/Agent/Client.hs @@ -171,6 +171,8 @@ module Simplex.Messaging.Agent.Client userServers, pickServer, getNextServer, + proxiedRelayFailed, + proxiedRelaySucceeded, withNextSrv, incSMPServerStat, incSMPServerStat', @@ -226,7 +228,7 @@ import Data.Set (Set) import qualified Data.Set as S import Data.Text (Text) import Data.Text.Encoding -import Data.Time (UTCTime, addUTCTime, defaultTimeLocale, formatTime, getCurrentTime) +import Data.Time (UTCTime, addUTCTime, defaultTimeLocale, diffUTCTime, formatTime, getCurrentTime) import Data.Time.Clock.System (getSystemTime) import Data.Word (Word16, Word32) import qualified Data.X509.Validation as XV @@ -351,8 +353,9 @@ data AgentClient = AgentClient useClientServices :: TMap UserId Bool, -- smpProxiedRelays: -- SMPTransportSession defines connection from proxy to relay, - -- SMPServerWithAuth defines client connected to SMP proxy (with the same userId and entityId in TransportSession) - smpProxiedRelays :: TMap SMPTransportSession SMPServerWithAuth, + -- SMPServerWithAuth defines client connected to SMP proxy (with the same userId and entityId in TransportSession), + -- Maybe UTCTime is the start of the current failure streak of this proxy + smpProxiedRelays :: TMap SMPTransportSession (SMPServerWithAuth, Maybe UTCTime), ntfServers :: TVar [NtfServer], ntfClients :: TMap NtfTransportSession NtfClientVar, xftpServers :: TMap UserId (UserServers 'PXFTP), @@ -698,12 +701,14 @@ getSMPProxyClient c@AgentClient {active, smpClients, smpProxiedRelays, workerSeq unlessM (readTVarIO active) $ throwE INACTIVE proxySrv <- maybe (getNextServer c userId proxySrvs [destSrv]) pure proxySrv_ ts <- liftIO getCurrentTime - (tSess, auth) <- atomically $ proxiedTransportSession proxySrv - withGetSessVar workerSeq tSess smpClients ts (newProxyClient tSess auth ts) (waitForProxyClient tSess auth) + (tSess@(_, srv, _), auth) <- atomically $ proxiedTransportSession proxySrv + withGetSessVar workerSeq tSess smpClients ts (newProxyClient tSess auth ts) (waitForProxyClient tSess auth) `catchAllErrors` \case + e@(BROKER _ TIMEOUT) -> throwE e + e -> atomically (deleteProxiedRelay c destSess srv) >> throwE e where proxiedTransportSession :: SMPServerWithAuth -> STM (SMPTransportSession, Maybe SMP.BasicAuth) proxiedTransportSession proxySrv = do - ProtoServerWithAuth srv auth <- TM.lookup destSess smpProxiedRelays >>= maybe (TM.insert destSess proxySrv smpProxiedRelays $> proxySrv) pure + ProtoServerWithAuth srv auth <- TM.lookup destSess smpProxiedRelays >>= maybe (when (isNothing proxySrv_) (TM.insert destSess (proxySrv, Nothing) smpProxiedRelays) $> proxySrv) (pure . fst) pure ((userId, srv, qId), auth) newProxyClient :: SMPTransportSession -> Maybe SMP.BasicAuth -> UTCTime -> SMPClientVar -> AM (SMPConnectedClient, Either AgentErrorType ProxiedRelay) newProxyClient tSess auth ts v = do @@ -728,7 +733,7 @@ getSMPProxyClient c@AgentClient {active, smpClients, smpProxiedRelays, workerSeq atomically $ do unless (serverHostError e) $ do removeSessVar rv destSrv prs - TM.delete destSess smpProxiedRelays + deleteProxiedRelay c destSess $ protocolClientServer' smp putTMVar (sessionVar rv) (Left e) pure $ Left e where @@ -779,7 +784,7 @@ smpConnectClient c@AgentClient {smpClients, msgQ, proxySessTs, presetDomains} nm (Nothing, Nothing) -> pure () smpClientDisconnected :: AgentClient -> SMPTransportSession -> Env -> SMPClientVar -> TMap SMPServer ProxiedRelayVar -> SMPClient -> IO () -smpClientDisconnected c@AgentClient {active, smpClients, smpProxiedRelays} tSess@(userId, srv, cId) env v prs client = do +smpClientDisconnected c@AgentClient {active, smpClients} tSess@(userId, srv, cId) env v prs client = do removeClientAndSubs >>= serverDown logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv where @@ -799,7 +804,7 @@ smpClientDisconnected c@AgentClient {active, smpClients, smpProxiedRelays} tSess cs = nubOrd $ map qConnId qs -- this removes proxied relays that this client created sessions to destSrvs <- M.keys <$> readTVar prs - forM_ destSrvs $ \destSrv -> TM.delete (userId, destSrv, cId) smpProxiedRelays + forM_ destSrvs $ \destSrv -> deleteProxiedRelay c (userId, destSrv, cId) srv pure (qs, cs, serviceSub_) serverDown :: ([RcvQueueSub], [ConnId], Maybe ServiceSub) -> IO () @@ -1208,12 +1213,15 @@ sendOrProxySMPCommand c nm userId destSrv@ProtocolServer {host = destHosts} conn SPFProhibit -> False unknownServer = liftIO $ maybe True (\srvs -> all (`S.notMember` knownHosts srvs) destHosts) <$> TM.lookupIO userId (smpServers c) sendViaProxy :: Maybe SMPServerWithAuth -> SMPTransportSession -> AM (Maybe SMPServer, a) - sendViaProxy proxySrv_ destSess@(_, _, connId_) = do - r <- tryAllErrors . withProxySession c nm proxySrv_ destSess entId ("PFWD " <> cmdStr) $ \(SMPConnectedClient smp _, proxySess@ProxiedRelay {prBasicAuth}) -> do - r' <- liftClient SMP (clientServer smp) $ sendCmdViaProxy smp proxySess + sendViaProxy proxySrv_ destSess = do + r <- tryAllErrors . withProxySession c nm proxySrv_ destSess entId ("PFWD " <> cmdStr) $ \(SMPConnectedClient smp prs, proxySess@ProxiedRelay {prBasicAuth}) -> do let proxySrv = protocolClientServer' smp + r' <- + liftClient SMP (clientServer smp) (sendCmdViaProxy smp proxySess) `catchAllErrors` \case + e@SMP {} -> atomically (proxiedRelaySucceeded c destSess proxySrv) >> throwE e + e -> throwE e case r' of - Right r -> pure (Just proxySrv, r) + Right r -> atomically (proxiedRelaySucceeded c destSess proxySrv) $> (Just proxySrv, r) Left proxyErr -> do case proxyErr of ProxyProtocolError (SMP.PROXY SMP.NO_SESSION) -> do @@ -1223,6 +1231,9 @@ sendOrProxySMPCommand c nm userId destSrv@ProtocolServer {host = destHosts} conn -- sendViaProxy is called recursively here to re-create the session via the same server -- to avoid failure in interactive calls that don't retry after the session disconnection. Nothing -> sendViaProxy (Just $ ProtoServerWithAuth proxySrv prBasicAuth) destSess + ProxyProtocolError (SMP.PROXY (SMP.BROKER e)) + | not (brokerHostError e) -> liftIO (proxiedRelayFailed c destSess proxySrv) >> proxyError + ProxyProtocolError (SMP.PROXY (SMP.PROTOCOL _)) -> atomically (proxiedRelaySucceeded c destSess proxySrv) >> proxyError _ -> proxyError where proxyError = @@ -1234,17 +1245,9 @@ sendOrProxySMPCommand c nm userId destSrv@ProtocolServer {host = destHosts} conn } -- checks that the current proxied relay session is the same one that was used to send the message and removes it deleteRelaySession = - ( TM.lookup destSess (smpProxiedRelays c) - $>>= \(ProtoServerWithAuth srv _) -> tryReadSessVar (userId, srv, connId_) (smpClients c) - ) - >>= \case - Just (Right (SMPConnectedClient smp' prs)) - | sameClient smp' -> - tryReadSessVar destSrv prs >>= \case - Just (Right proxySess') | sameProxiedRelay proxySess' -> TM.delete destSrv prs - _ -> pure () - _ -> pure () - sameClient smp' = sessionId (thParams smp) == sessionId (thParams smp') + tryReadSessVar destSrv prs >>= \case + Just (Right proxySess') | sameProxiedRelay proxySess' -> TM.delete destSrv prs + _ -> pure () sameProxiedRelay proxySess' = prSessionId proxySess == prSessionId proxySess' case r of Right r'@(srv_, _) -> do @@ -1630,11 +1633,29 @@ serverHostError = \case SMP _ (SMP.PROXY (SMP.BROKER e)) -> brokerHostError e PROXY _ _ (ProxyProtocolError (SMP.PROXY (SMP.BROKER e))) -> brokerHostError e _ -> False - where - brokerHostError = \case - HOST -> True - SMP.TRANSPORT TEVersion -> True - _ -> False + +brokerHostError :: BrokerErrorType -> Bool +brokerHostError = \case + HOST -> True + SMP.TRANSPORT TEVersion -> True + _ -> False + +deleteProxiedRelay :: AgentClient -> SMPTransportSession -> SMPServer -> STM () +deleteProxiedRelay AgentClient {smpProxiedRelays} destSess proxySrv = + TM.lookup destSess smpProxiedRelays >>= mapM_ (\(ProtoServerWithAuth srv _, _) -> when (srv == proxySrv) $ TM.delete destSess smpProxiedRelays) + +proxiedRelayFailed :: AgentClient -> SMPTransportSession -> SMPServer -> IO () +proxiedRelayFailed AgentClient {agentEnv, smpProxiedRelays} destSess proxySrv = do + let failureInterval = proxyFailureInterval $ config agentEnv + ts <- getCurrentTime + atomically $ + TM.lookup destSess smpProxiedRelays >>= mapM_ (\(p@(ProtoServerWithAuth srv _), failedAt_) -> when (srv == proxySrv) $ case failedAt_ of + Nothing -> TM.insert destSess (p, Just ts) smpProxiedRelays + Just failedAt -> when (diffUTCTime ts failedAt >= failureInterval) $ TM.delete destSess smpProxiedRelays) + +proxiedRelaySucceeded :: AgentClient -> SMPTransportSession -> SMPServer -> STM () +proxiedRelaySucceeded AgentClient {smpProxiedRelays} destSess proxySrv = + TM.lookup destSess smpProxiedRelays >>= mapM_ (\(p@(ProtoServerWithAuth srv _), failedAt_) -> when (srv == proxySrv && isJust failedAt_) $ TM.insert destSess (p, Nothing) smpProxiedRelays) -- | Batch by transport session and subscribe queues. The list of results can have a different order. subscribeQueues :: AgentClient -> Bool -> [RcvQueueSub] -> AM' [(RcvQueueSub, Either AgentErrorType (Maybe ServiceId))] diff --git a/src/Simplex/Messaging/Agent/Env/SQLite.hs b/src/Simplex/Messaging/Agent/Env/SQLite.hs index fc7af667f..d27b76102 100644 --- a/src/Simplex/Messaging/Agent/Env/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Env/SQLite.hs @@ -164,6 +164,7 @@ data AgentConfig = AgentConfig quotaExceededTimeout :: NominalDiffTime, commandQuotaRetryInterval :: RetryInterval, persistErrorInterval :: NominalDiffTime, + proxyFailureInterval :: NominalDiffTime, initialCleanupDelay :: Int64, cleanupInterval :: Int64, cleanupBatchSize :: Int, @@ -244,6 +245,7 @@ defaultAgentConfig = quotaExceededTimeout = 7 * nominalDay, commandQuotaRetryInterval = RetryInterval {initialInterval = 30_000000, increaseAfter = 0, maxInterval = 3600_000000}, persistErrorInterval = 3, -- seconds + proxyFailureInterval = 30, -- seconds initialCleanupDelay = 30 * 1000000, -- 30 seconds cleanupInterval = 5 * 60 * 1000000, -- 5 minutes cleanupBatchSize = 10000,