From 656df39aa0bb1885c5de767bc65a172ebb502c29 Mon Sep 17 00:00:00 2001
From: Narasimha-sc <166327228+Narasimha-sc@users.noreply.github.com>
Date: Mon, 5 Oct 2026 18:18:25 +0000
Subject: [PATCH 1/4] transport: optionally bound the time a connection with
 unacknowledged data takes to fail

TCP keep-alive probes are only sent while there is no unacknowledged data, so
a connection that died with data in flight is only reported as failed when the
OS gives up retransmitting - ~15 minutes on Linux and Android. When no network
change is reported to the agent (a silent route change, a client without a
network observer) that is how long delivery keeps failing on the dead client.

KeepAliveOpts gets unackedDataTimeout: when it is set, the sockets that set TCP
keep-alive also set TCP_USER_TIMEOUT (TCP_RXT_CONNDROPTIME on Darwin, not set
on Windows) to the keep-alive detection time, so that such a connection fails
after the same time as an idle one. On Linux this timeout also bounds the
zero-window persist timer, so a server that stops reading for that long then
fails the connection.

The option is off unless the client sets it, so that it can be enabled by the
users who test it before it becomes the default. It is a Maybe to keep the
clients that do not send it working. As it is derived from the keep-alive
options where they are set, the two detection times cannot diverge and lowering
the keep-alive settings lowers both.

It is removed for the connections via SOCKS proxy - the socket is then
connected to the proxy, its acknowledgements only mean that the proxy buffered
the data, and a slow circuit should not fail the connection.
---
 src/Simplex/Messaging/Client.hs              |  7 +++--
 src/Simplex/Messaging/Transport/KeepAlive.hs | 31 ++++++++++++++++++--
 2 files changed, 33 insertions(+), 5 deletions(-)

diff --git a/src/Simplex/Messaging/Client.hs b/src/Simplex/Messaging/Client.hs
index 7a2eedeb96..d96078804e 100644
--- a/src/Simplex/Messaging/Client.hs
+++ b/src/Simplex/Messaging/Client.hs
@@ -150,7 +150,7 @@ import Data.Int (Int64)
 import Data.List (find, isSuffixOf)
 import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.List.NonEmpty as L
-import Data.Maybe (catMaybes, fromMaybe)
+import Data.Maybe (catMaybes, fromMaybe, isJust)
 import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Time.Clock (UTCTime (..), diffUTCTime, getCurrentTime)
@@ -442,9 +442,12 @@ defaultNetworkConfig =
 
 transportClientConfig :: NetworkConfig -> NetworkRequestMode -> TransportHost -> Bool -> Maybe [ALPN] -> TransportClientConfig
 transportClientConfig NetworkConfig {socksProxy, socksMode, tcpConnectTimeout, tcpKeepAlive, logTLSErrors} nm host useSNI clientALPN =
-  TransportClientConfig {socksProxy = useSocksProxy socksMode, tcpConnectTimeout = tOut, tcpKeepAlive, logTLSErrors, clientCredentials = Nothing, clientALPN, useSNI}
+  TransportClientConfig {socksProxy = socks, tcpConnectTimeout = tOut, tcpKeepAlive = keepAlive, logTLSErrors, clientCredentials = Nothing, clientALPN, useSNI}
   where
     tOut = netTimeoutInt tcpConnectTimeout nm
+    socks = useSocksProxy socksMode
+    -- with SOCKS proxy the socket is connected to the proxy, and a slow circuit should not fail the connection
+    keepAlive = (\ka -> if isJust socks then ka {unackedDataTimeout = Nothing} else ka) <$> tcpKeepAlive
     socksProxy' = (\(SocksProxyWithAuth _ proxy) -> proxy) <$> socksProxy
     useSocksProxy SMAlways = socksProxy'
     useSocksProxy SMOnion = case host of
diff --git a/src/Simplex/Messaging/Transport/KeepAlive.hs b/src/Simplex/Messaging/Transport/KeepAlive.hs
index 1d34a27f0a..42059f22d0 100644
--- a/src/Simplex/Messaging/Transport/KeepAlive.hs
+++ b/src/Simplex/Messaging/Transport/KeepAlive.hs
@@ -11,6 +11,7 @@ module Simplex.Messaging.Transport.KeepAlive
   ) where
 
 import qualified Data.Aeson.TH as J
+import Control.Monad (when)
 import Foreign.C (CInt (..))
 import Network.Socket
 import Simplex.Messaging.Parsers (defaultJSON)
@@ -18,16 +19,25 @@ import Simplex.Messaging.Parsers (defaultJSON)
 data KeepAliveOpts = KeepAliveOpts
   { keepIdle :: Int,
     keepIntvl :: Int,
-    keepCnt :: Int
+    keepCnt :: Int,
+    -- | A connection with unacknowledged data gets no keep-alive probes, so the OS fails it only when it
+    -- gives up retransmitting - ~15 minutes on Linux and Android. When this is set such a connection is
+    -- failed after the keep-alive detection time instead. Absent or False - the OS default.
+    unackedDataTimeout :: Maybe Bool
   }
   deriving (Eq, Show)
 
+-- seconds after which a connection with no response to the probes is reported as failed
+keepAliveDetectionTime :: KeepAliveOpts -> Int
+keepAliveDetectionTime KeepAliveOpts {keepIdle, keepIntvl, keepCnt} = keepIdle + keepIntvl * keepCnt
+
 defaultKeepAliveOpts :: KeepAliveOpts
 defaultKeepAliveOpts =
   KeepAliveOpts
     { keepIdle = 30,
       keepIntvl = 15,
-      keepCnt = 4
+      keepCnt = 4,
+      unackedDataTimeout = Nothing
     }
 
 _SOL_TCP :: CInt
@@ -53,8 +63,12 @@ _TCP_KEEPCNT = 16
 
 #if defined(darwin_HOST_OS)
 foreign import capi "netinet/tcp.h value TCP_KEEPALIVE" _TCP_KEEPIDLE :: CInt
+
+foreign import capi "netinet/tcp.h value TCP_RXT_CONNDROPTIME" _TCP_RXT_CONNDROPTIME :: CInt
 #else
 foreign import capi "netinet/tcp.h value TCP_KEEPIDLE" _TCP_KEEPIDLE :: CInt
+
+foreign import capi "netinet/tcp.h value TCP_USER_TIMEOUT" _TCP_USER_TIMEOUT :: CInt
 #endif
 
 foreign import capi "netinet/tcp.h value TCP_KEEPINTVL" _TCP_KEEPINTVL :: CInt
@@ -63,11 +77,22 @@ foreign import capi "netinet/tcp.h value TCP_KEEPCNT" _TCP_KEEPCNT :: CInt
 
 #endif
 
+-- | seconds
+setSocketUnackedDataTimeout :: Socket -> Int -> IO ()
+#if defined(mingw32_HOST_OS)
+setSocketUnackedDataTimeout _ _ = pure ()
+#elif defined(darwin_HOST_OS)
+setSocketUnackedDataTimeout sock = setSocketOption sock (SockOpt _SOL_TCP _TCP_RXT_CONNDROPTIME)
+#else
+setSocketUnackedDataTimeout sock t = setSocketOption sock (SockOpt _SOL_TCP _TCP_USER_TIMEOUT) (t * 1000) -- TCP_USER_TIMEOUT is in milliseconds
+#endif
+
 setSocketKeepAlive :: Socket -> KeepAliveOpts -> IO ()
-setSocketKeepAlive sock KeepAliveOpts {keepCnt, keepIdle, keepIntvl} = do
+setSocketKeepAlive sock ka@KeepAliveOpts {keepCnt, keepIdle, keepIntvl, unackedDataTimeout} = do
   setSocketOption sock KeepAlive 1
   setSocketOption sock (SockOpt _SOL_TCP _TCP_KEEPIDLE) keepIdle
   setSocketOption sock (SockOpt _SOL_TCP _TCP_KEEPINTVL) keepIntvl
   setSocketOption sock (SockOpt _SOL_TCP _TCP_KEEPCNT) keepCnt
+  when (unackedDataTimeout == Just True) $ setSocketUnackedDataTimeout sock $ keepAliveDetectionTime ka
 
 $(J.deriveJSON defaultJSON ''KeepAliveOpts)

From 8e3f9a47b97754f69d77453313a712d10db6fdd1 Mon Sep 17 00:00:00 2001
From: Narasimha-sc <166327228+Narasimha-sc@users.noreply.github.com>
Date: Sun, 4 Oct 2026 15:01:53 +0000
Subject: [PATCH 2/4] agent: reconnect to the servers when the network changes

After the network changes (e.g. the IP address changed after internet loss) the
connections to the servers are dead, but the OS reports it only after TCP
retransmissions time out - ~15 minutes on Linux/Android while the agent keeps
writing to the dead socket. Until then message delivery, subscriptions and file
transfers keep failing on the same client.

setUserNetworkInfo on every online network event (the clients report one when
the network type changes and when it goes offline to online):

- closes the connected SMP and NTF clients and fails their pending requests, so
  that the workers that resume do not wait for the responses that will not
  arrive - the requests are failed only once the client's threads terminated,
  which the client now records, as its session var is also removed when a
  resubscription fails (each client type bounded by 1 second; the requests of a
  client that did not terminate in this time are left to time out as before);
- discards the errors of the clients that failed to connect, cached for
  persistErrorInterval, and removes a client that is still connecting, in the
  transaction that reads it, so that the resumed workers create a new client
  instead of waiting for a connection made via the previous network - that
  client is closed by newProtocolClient when it finds it is no longer in the map;
- closes the XFTP clients asynchronously, removing them from the map.

A client that is no longer the one in the map is rejected in the same
transaction that would publish it, so that the requests made on it do not wait
for the timeout, and it is closed in a forked thread, as closing an XFTP client
waits for its thread to terminate; it sets its session id while its var is the
one in the map and updates its service only if it was.

The clients are closed in a forked thread, so that the API does not block, and
the closes of an event that a newer one superseded are skipped. When the agent
was offline the reported network info is applied when the clients are closed,
so that the workers waiting for the network do not resume before then; when it
was online it is applied immediately.

As the clients are closed, the network change now results in DISCONNECT and
DOWN events followed by CONNECT and UP, as any other reconnection.

The retry loops are not restarted, so a worker already waiting out its retry
interval resumes only when it expires.
---
 src/Simplex/Messaging/Agent.hs         | 35 ++++++++++++++---
 src/Simplex/Messaging/Agent/Client.hs  | 52 ++++++++++++++++++++++++--
 src/Simplex/Messaging/Client.hs        | 11 +++++-
 src/Simplex/Messaging/Session.hs       |  5 +++
 tests/AgentTests/FunctionalAPITests.hs | 26 +++++++++++++
 5 files changed, 120 insertions(+), 9 deletions(-)

diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs
index c5f57cdd96..d378566b98 100644
--- a/src/Simplex/Messaging/Agent.hs
+++ b/src/Simplex/Messaging/Agent.hs
@@ -266,7 +266,7 @@ import Simplex.RemoteControl.Invitation
 import Simplex.RemoteControl.Types
 import System.Mem.Weak (deRefWeak)
 import UnliftIO.Async (mapConcurrently)
-import UnliftIO.Concurrent (forkFinally, forkIO, killThread, mkWeakThreadId, threadDelay)
+import UnliftIO.Concurrent (forkFinally, forkIO, forkIOWithUnmask, killThread, mkWeakThreadId, threadDelay)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
 
@@ -713,14 +713,39 @@ setNetworkConfig c@AgentClient {useNetworkConfig, proxySessTs} cfg' = do
   when changed $ liftIO $ reconnectAllServers c
 
 setUserNetworkInfo :: AgentClient -> UserNetworkInfo -> IO ()
-setUserNetworkInfo c@AgentClient {userNetworkInfo, userNetworkUpdated} ni = withAgentEnv' c $ do
+setUserNetworkInfo c@AgentClient {userNetworkInfo, userNetworkUpdated, networkEventSeq} ni = withAgentEnv' c $ do
   ts' <- liftIO getCurrentTime
   i <- asks $ userOfflineDelay . config
   -- if network offline event happens in less than `userOfflineDelay` after the previous event, it is ignored
-  atomically . whenM ((isOnline ni ||) <$> notRecentlyChanged ts' i) $ do
-    writeTVar userNetworkInfo ni
-    writeTVar userNetworkUpdated $ Just ts'
+  -- mask, so that the thread restoring the deferred online network info is always forked
+  liftIO $ E.mask_ $ do
+    event <- atomically $ do
+      update <- (isOnline ni ||) <$> notRecentlyChanged ts' i
+      if update
+        then do
+          writeTVar userNetworkUpdated $ Just ts'
+          -- when the agent was offline the reported info is applied when the clients are closed,
+          -- so that the workers waiting for the network do not resume before then
+          wasOnline <- isOnline <$> readTVar userNetworkInfo
+          writeTVar userNetworkInfo $ if wasOnline || not (isOnline ni) then ni else ni {online = False}
+          modifyTVar' networkEventSeq (+ 1)
+          Just <$> readTVar networkEventSeq
+        else pure Nothing
+    -- The client reports online network when it changes. The connections to the servers made via
+    -- the previous network are likely to be dead, but the OS may only report it after several minutes.
+    forM_ event $ \e -> when (isOnline ni) $ void $ forkIOWithUnmask $ \unmask ->
+      unmask (closeClients e) `E.finally` networkOnline e
   where
+    -- skipped when a newer info superseded this one - it closes the clients itself, and the ones
+    -- that stay connected while the network is offline are closed by the next online info
+    closeClients e = do
+      whenCurrent $ closeNetworkClients c smpClients
+      whenCurrent $ closeNetworkClients c ntfClients
+      whenCurrent $ closeProtocolServerClients c xftpClients
+      where
+        whenCurrent = whenM (atomically $ isCurrent e)
+    networkOnline e = atomically $ whenM (isCurrent e) $ writeTVar userNetworkInfo ni
+    isCurrent e = (e ==) <$> readTVar networkEventSeq
     notRecentlyChanged ts' i =
       maybe True (\ts -> diffUTCTime ts' ts > i) <$> readTVar userNetworkUpdated
 
diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs
index 3d8f8c1bd3..0601628a70 100644
--- a/src/Simplex/Messaging/Agent/Client.hs
+++ b/src/Simplex/Messaging/Agent/Client.hs
@@ -21,6 +21,7 @@
 {-# LANGUAGE TupleSections #-}
 {-# LANGUAGE TypeApplications #-}
 {-# LANGUAGE TypeFamilyDependencies #-}
+{-# LANGUAGE TypeOperators #-}
 
 module Simplex.Messaging.Agent.Client
   ( AgentClient (..),
@@ -148,6 +149,7 @@ module Simplex.Messaging.Agent.Client
     UserNetworkType (..),
     getFastNetworkConfig,
     waitForUserNetwork,
+    closeNetworkClients,
     isNetworkOnline,
     isOnline,
     throwWhenInactive,
@@ -363,6 +365,8 @@ data AgentClient = AgentClient
     presetServers :: [SMPServer],
     userNetworkInfo :: TVar UserNetworkInfo,
     userNetworkUpdated :: TVar (Maybe UTCTime),
+    -- incremented for each accepted network info, to identify the most recent one
+    networkEventSeq :: TVar Int,
     subscrConns :: TVar (Set ConnId),
     currentSubs :: TSessionSubs,
     removedSubs :: TMap (UserId, SMPServer) (TMap SMP.RecipientId SMPClientError),
@@ -535,6 +539,7 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, entitlements, netCf
   useNetworkConfig <- newTVarIO (slowNetworkConfig netCfg, netCfg)
   userNetworkInfo <- newTVarIO $ UserNetworkInfo UNOther True
   userNetworkUpdated <- newTVarIO Nothing
+  networkEventSeq <- newTVarIO 0
   subscrConns <- newTVarIO S.empty
   currentSubs <- SS.emptyIO
   removedSubs <- TM.emptyIO
@@ -579,6 +584,7 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, entitlements, netCf
         presetServers,
         userNetworkInfo,
         userNetworkUpdated,
+        networkEventSeq,
         subscrConns,
         currentSubs,
         removedSubs,
@@ -767,8 +773,12 @@ smpConnectClient c@AgentClient {smpClients, msgQ, proxySessTs, presetDomains} nm
       smp <- liftError (protocolClientError SMP $ B.unpack $ strEncode srv) $ do
         ts <- readTVarIO proxySessTs
         ExceptT $ getProtocolClient g nm tSess cfg' presetDomains (Just msgQ) ts $ smpClientDisconnected c tSess env v' prs
-      atomically $ SS.setSessionId tSess (sessionId $ thParams smp) $ currentSubs c
-      updateClientService service smp
+      -- the client removed from the map while it was connecting (see newProtocolClient) must not
+      -- replace the session or the service of the client that is current now
+      current <- atomically $ do
+        cur <- currentSessVar v' tSess smpClients
+        cur <$ when cur (SS.setSessionId tSess (sessionId $ thParams smp) $ currentSubs c)
+      when current $ updateClientService service smp
       pure SMPConnectedClient {connectedClient = smp, proxiedRelays = prs}
     updateClientService service smp = case (service, smpClientServiceId smp) of
       (Just (_, serviceId_), Just serviceId) -> withStore' c $ \db -> do
@@ -954,8 +964,18 @@ newProtocolClient ::
 newProtocolClient c tSess@(userId, srv, entityId_) clients connectClient v =
   tryAllErrors (connectClient v) >>= \case
     Right client -> do
+      -- the client removed from the map while it was connecting, e.g. when the network changed, must
+      -- not be used - the requests made on it would wait for the responses that will not arrive; it is
+      -- published and rejected in one transaction, so that the caller always receives the one in the map
+      let e = BROKER (B.unpack $ strEncode srv) (NETWORK NEFailedError)
+      current <- atomically $ do
+        cur <- currentSessVar v tSess clients
+        cur <$ putTMVar (sessionVar v) (if cur then Right client else Left (e, Nothing))
+      unless current $ do
+        -- closed in another thread, as closing an XFTP client waits for its thread to terminate
+        void $ liftIO $ forkIO $ closeProtocolServerClient (protocolClient client) `catchAll_` pure ()
+        throwE e
       logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv <> " (user " <> bshow userId <> maybe "" (" for entity " <>) entityId_ <> ")"
-      atomically $ putTMVar (sessionVar v) (Right client)
       liftIO $ nonBlockingWriteTBQueue (subQ c) ("", "", AEvt SAENone $ hostEvent CONNECT client)
       pure client
     Left e -> do
@@ -1049,6 +1069,32 @@ reconnectServerClients :: ProtocolServerClient v err msg => AgentClient -> (Agen
 reconnectServerClients c clientsSel =
   readTVarIO (clientsSel c) >>= mapM_ (forkIO . closeClient_ c)
 
+-- The pending requests of the clients that are closed fail before this function returns, so that
+-- the workers that resume do not wait for the responses that will not arrive; the requests of
+-- a client that did not terminate within the timeout are left to time out as before.
+closeNetworkClients :: (ProtocolServerClient v err msg, ProtoClient msg ~ ProtocolClient v err msg) => AgentClient -> (AgentClient -> TMap (TransportSession msg) (ClientVar msg)) -> IO ()
+closeNetworkClients c clientsSel = do
+  closed <- readTVarIO (clientsSel c) >>= fmap catMaybes . mapM closeNetworkClient . M.toList
+  void $ 1000000 `timeout` atomically (mapM_ (\pc -> whenM (clientConnected pc) retry) closed)
+  -- the requests are failed after the client threads terminate, so that the responses that arrive
+  -- before that are processed as usual rather than as the responses to the failed requests
+  forM_ closed $ \pc -> unlessM (atomically $ clientConnected pc) $ failPendingRequests pc
+  where
+    closeNetworkClient (tSess, v) =
+      atomically (tryReadTMVar (sessionVar v) >>= \r -> r <$ removeUnconnected r) >>= \case
+        Just (Right client) -> do
+          let pc = protocolClient client
+          Just pc <$ forkIO (closeProtocolServerClient pc `catchAll_` pure ())
+        _ -> pure Nothing
+      where
+        -- the client that failed to connect is removed so that the resumed workers do not get its cached
+        -- error, and the one that is connecting - so that they create a new client instead of waiting for
+        -- the connection made via the previous network; it is read and removed in one transaction, so that
+        -- it cannot publish itself in between - newProtocolClient then rejects and closes it
+        removeUnconnected = \case
+          Just (Right _) -> pure ()
+          _ -> removeSessVar v tSess (clientsSel c)
+
 reconnectSMPServer :: AgentClient -> UserId -> SMPServer -> IO ()
 reconnectSMPServer c userId srv = do
   cs <- readTVarIO $ smpClients c
diff --git a/src/Simplex/Messaging/Client.hs b/src/Simplex/Messaging/Client.hs
index d96078804e..c1f0bc03c2 100644
--- a/src/Simplex/Messaging/Client.hs
+++ b/src/Simplex/Messaging/Client.hs
@@ -35,6 +35,8 @@ module Simplex.Messaging.Client
     ProxiedRelay (..),
     getProtocolClient,
     closeProtocolClient,
+    failPendingRequests,
+    clientConnected,
     protocolClientServer,
     protocolClientServer',
     transportHost',
@@ -654,7 +656,7 @@ getProtocolClient g nm transportSession@(_, srv, _) cfg@ProtocolClientConfig {qS
             writeTVar (connected c) True
             putTMVar cVar $ Right c'
           raceAny_ ([send c' th, process c', receive c' th] <> [monitor c' | smpPingInterval > 0])
-            `E.finally` disconnected c'
+            `E.finally` (atomically (writeTVar (connected c) False) >> disconnected c')
 
     send :: Transport c => ProtocolClient v err msg -> THandle v c 'TClient -> IO ()
     send ProtocolClient {client_ = PClient {sndQ}} h = forever $ atomically (readTBQueue sndQ) >>= sendPending
@@ -759,6 +761,13 @@ closeProtocolClient :: ProtocolClient v err msg -> IO ()
 closeProtocolClient = mapM_ (deRefWeak >=> mapM_ killThread) . action
 {-# INLINE closeProtocolClient #-}
 
+clientConnected :: ProtocolClient v err msg -> STM Bool
+clientConnected ProtocolClient {client_ = PClient {connected}} = readTVar connected
+
+failPendingRequests :: ProtocolClient v err msg -> IO ()
+failPendingRequests ProtocolClient {client_ = PClient {sentCommands}} =
+  readTVarIO sentCommands >>= mapM_ (\Request {pending, responseVar} -> atomically $ whenM (swapTVar pending False) $ void $ tryPutTMVar responseVar $ Left $ PCENetworkError NEFailedError)
+
 -- | SMP client error type.
 data ProtocolClientError err
   = -- | Correctly parsed SMP server ERR response.
diff --git a/src/Simplex/Messaging/Session.hs b/src/Simplex/Messaging/Session.hs
index 16ff3c934e..67c9202f8e 100644
--- a/src/Simplex/Messaging/Session.hs
+++ b/src/Simplex/Messaging/Session.hs
@@ -6,6 +6,7 @@ module Simplex.Messaging.Session
   ( SessionVar (..),
     getSessVar,
     removeSessVar,
+    currentSessVar,
     withGetSessVar,
     withGetSessVar',
     tryReadSessVar,
@@ -44,6 +45,10 @@ removeSessVar v sessKey vs =
     Just v' | sessionVarId v == sessionVarId v' -> TM.delete sessKey vs
     _ -> pure ()
 
+-- the client is current while its session var is the one in the map
+currentSessVar :: Ord k => SessionVar a -> k -> TMap k (SessionVar a) -> STM Bool
+currentSessVar v sessKey vs = maybe False ((sessionVarId v ==) . sessionVarId) <$> TM.lookup sessKey vs
+
 -- | Get or create a session var and route to onNew (newly created) or onExisting. The new-var
 -- branch is bracketed from the point of creation: if it is interrupted before filling the var
 -- (e.g. an async exception during connect), the still-empty var is dropped from the map so the
diff --git a/tests/AgentTests/FunctionalAPITests.hs b/tests/AgentTests/FunctionalAPITests.hs
index 0467306f5e..0cd91c4461 100644
--- a/tests/AgentTests/FunctionalAPITests.hs
+++ b/tests/AgentTests/FunctionalAPITests.hs
@@ -91,6 +91,7 @@ import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestSte
 import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..), Env (..), InitialAgentServers (..), createAgentStore)
 import Simplex.Messaging.Agent.Protocol hiding (CON, CONF, INFO, REQ, SENT)
 import qualified Simplex.Messaging.Agent.Protocol as A
+import Simplex.Messaging.Agent.RetryInterval (RetryInterval (..), RetryInterval2 (..))
 import Simplex.Messaging.Agent.Store (Connection' (..), SomeConn' (..), StoredRcvQueue (..))
 import Simplex.Messaging.Agent.Store.AgentStore (deleteRatchetKeyHashesExpired, getConn, getRatchetX3dhKeys)
 import Simplex.Messaging.Agent.Store.Common (DBStore (..), withTransaction)
@@ -606,6 +607,7 @@ functionalAPITests ps = do
   describe "user network info" $ do
     it "should wait for user network" testWaitForUserNetwork
     it "should not reset online to offline if happens too quickly" testDoNotResetOnlineToOffline
+    it "should reconnect to servers when network changes" $ testNetworkChangeReconnect ps
     it "should resume multiple threads" testResumeMultipleThreads
   describe "SMP queue info" $ do
     it "server should respond with queue and subscription information" $
@@ -4790,9 +4792,31 @@ testWaitForUserNetwork = withAgent 1 aCfg initAgentServers testDB $ \a -> do
     (threadDelay 50000 >> setUserNetworkInfo a (UserNetworkInfo UNCellular True))
     (networkDelay a 50000)
   noNetworkDelay a
+  -- the agent is disposed, so that the workers the network events resumed do not use the database
+  -- of the test that runs next - they share the file
+  disposeAgentClient a
   where
     aCfg = agentCfg {userNetworkInterval = 100000, userOfflineDelay = 0}
 
+testNetworkChangeReconnect :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
+testNetworkChangeReconnect ps =
+  withAgentClientsCfg2 agentCfg {userOfflineDelay = 0} agentCfg $ \a b -> withSmpServer ps $ do
+    (aId, bId) <- runRight $ makeConnection a b
+    -- network type changed while online - connections to servers are re-established
+    setUserNetworkInfo a $ UserNetworkInfo UNCellular True
+    nGet a =##> \case ("", "", DOWN _ [c]) -> c == bId; _ -> False
+    nGet a =##> \case ("", "", UP _ [c]) -> c == bId; _ -> False
+    -- the same network type reported again (the client only reports it when the network changes) - reconnects
+    setUserNetworkInfo a $ UserNetworkInfo UNCellular True
+    nGet a =##> \case ("", "", DOWN _ [c]) -> c == bId; _ -> False
+    nGet a =##> \case ("", "", UP _ [c]) -> c == bId; _ -> False
+    setUserNetworkInfo a $ UserNetworkInfo UNNone False
+    noMessages a "no reconnection when offline"
+    setUserNetworkInfo a $ UserNetworkInfo UNWifi True
+    nGet a =##> \case ("", "", DOWN _ [c]) -> c == bId; _ -> False
+    nGet a =##> \case ("", "", UP _ [c]) -> c == bId; _ -> False
+    runRight_ $ exchangeGreetings a bId b aId
+
 testDoNotResetOnlineToOffline :: IO ()
 testDoNotResetOnlineToOffline = withAgent 1 aCfg initAgentServers testDB $ \a -> do
   noNetworkDelay a
@@ -4811,6 +4835,7 @@ testDoNotResetOnlineToOffline = withAgent 1 aCfg initAgentServers testDB $ \a ->
   setUserNetworkInfo a $ UserNetworkInfo UNWifi True
   setUserNetworkInfo a $ UserNetworkInfo UNNone False -- ingnored
   noNetworkDelay a
+  disposeAgentClient a
   where
     aCfg = agentCfg {userNetworkInterval = 100000, userOfflineDelay = 0.1}
 
@@ -4832,6 +4857,7 @@ testResumeMultipleThreads = withAgent 1 aCfg initAgentServers testDB $ \a -> do
   let average = sum ts `div` fromIntegral (length ts)
   average < 3000000 `shouldBe` True
   maximum ts < 4000000 `shouldBe` True
+  disposeAgentClient a
   where
     aCfg = agentCfg {userOfflineDelay = 0}
 

From 2c4ea1f0a4e2bcb7a9d3fa16f255788184ad1716 Mon Sep 17 00:00:00 2001
From: Narasimha-sc <166327228+Narasimha-sc@users.noreply.github.com>
Date: Sat, 26 Sep 2026 09:50:17 +0000
Subject: [PATCH 3/4] agent: fix three races that closing clients on a network
 change makes reachable

Closing the clients on a network change makes a client that is no longer the one
in the session map a routine occurrence rather than a rare one, which makes these
pre-existing races reachable in normal use.

- setSubsPending removes the subscriptions of a transport session only when the
  session id matches. A client that is no longer the one in the map never set
  its session id, and when the session mode stopped matching the transport
  session its disconnection removed the subscriptions of the client that
  replaced it without making them pending, so they were neither subscribed nor
  scheduled and nothing corrected it until the app restarted.
- smpClientDisconnected removes the sessions to the proxied relays only when no
  other client replaced this one, as those sessions are then the sessions of the
  client that replaced it.
- updateClientService reads the stored service id in the transaction that writes
  it. The id read before the handshake can be stale, as the transaction of a
  client that is no longer current can commit after the one of the client that
  replaced it, and removeRcvServiceAssocs would then run against a service id
  that changed since, removing the associations of the current client.
---
 src/Simplex/Messaging/Agent/Client.hs       | 12 ++++++++----
 src/Simplex/Messaging/Agent/TSessionSubs.hs |  4 +++-
 2 files changed, 11 insertions(+), 5 deletions(-)

diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs
index 0601628a70..227663df44 100644
--- a/src/Simplex/Messaging/Agent/Client.hs
+++ b/src/Simplex/Messaging/Agent/Client.hs
@@ -781,7 +781,9 @@ smpConnectClient c@AgentClient {smpClients, msgQ, proxySessTs, presetDomains} nm
       when current $ updateClientService service smp
       pure SMPConnectedClient {connectedClient = smp, proxiedRelays = prs}
     updateClientService service smp = case (service, smpClientServiceId smp) of
-      (Just (_, serviceId_), Just serviceId) -> withStore' c $ \db -> do
+      (Just _, Just serviceId) -> withStore' c $ \db -> do
+        -- the id read before the handshake can be stale - this transaction can commit after another client's
+        serviceId_ <- (>>= snd) <$> getClientServiceCredentials db userId srv
         setClientServiceId db userId srv serviceId
         forM_ serviceId_ $ \sId -> when (sId /= serviceId) $ removeRcvServiceAssocs db userId srv
       (Just _, Nothing) -> withStore' c $ \db -> deleteClientService db userId srv -- e.g., server version downgrade
@@ -807,9 +809,11 @@ smpClientDisconnected c@AgentClient {active, smpClients, smpProxiedRelays} tSess
           (subs, serviceSub_) <- SS.setSubsPending mode tSess sessId $ currentSubs c
           let qs = M.elems subs
               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
+          -- this removes proxied relays that this client created sessions to, unless another client
+          -- replaced this one, e.g. when the network changed - those sessions are now its sessions
+          unlessM (TM.member tSess smpClients) $ do
+            destSrvs <- M.keys <$> readTVar prs
+            forM_ destSrvs $ \destSrv -> TM.delete (userId, destSrv, cId) smpProxiedRelays
           pure (qs, cs, serviceSub_)
 
     serverDown :: ([RcvQueueSub], [ConnId], Maybe ServiceSub) -> IO ()
diff --git a/src/Simplex/Messaging/Agent/TSessionSubs.hs b/src/Simplex/Messaging/Agent/TSessionSubs.hs
index a1db48c9ea..42a3dc01a7 100644
--- a/src/Simplex/Messaging/Agent/TSessionSubs.hs
+++ b/src/Simplex/Messaging/Agent/TSessionSubs.hs
@@ -216,7 +216,9 @@ setSubsPending mode tSess@(uId, srv, connId_) sessId tss@(TSessionSubs ss)
   | entitySession == isJust connId_ =
       TM.lookup tSess ss >>= withSessSubs (`setSubsPending_` Nothing)
   | otherwise =
-      TM.lookupDelete tSess ss >>= withSessSubs setPendingChangeMode
+      -- the subscriptions are removed only when the session matches - a client that is no longer
+      -- the one in the map never set its session id, and its disconnection must not remove them
+      TM.lookup tSess ss >>= withSessSubs (\s -> TM.delete tSess ss >> setPendingChangeMode s)
   where
     entitySession = mode == TSMEntity
     sessEntId = if entitySession then Just else const Nothing

From 84381e0ea528f7812ebe725d8f0b5dc638efa6b4 Mon Sep 17 00:00:00 2001
From: Narasimha-sc <166327228+Narasimha-sc@users.noreply.github.com>
Date: Mon, 5 Oct 2026 18:18:25 +0000
Subject: [PATCH 4/4] transport: optionally bound the time a connection with
 unacknowledged data takes to fail

TCP keep-alive probes are only sent while there is no unacknowledged data, so
a connection that died with data in flight is only reported as failed when the
OS gives up retransmitting - ~15 minutes on Linux and Android. When no network
change is reported to the agent (a silent route change, a client without a
network observer) that is how long delivery keeps failing on the dead client.

KeepAliveOpts gets unackedDataTimeout: when it is set, the sockets that set TCP
keep-alive also set TCP_USER_TIMEOUT (TCP_RXT_CONNDROPTIME on Darwin, not set
on Windows) to the keep-alive detection time, so that such a connection fails
after the same time as an idle one. On Linux this timeout also bounds the
zero-window persist timer, so a server that stops reading for that long then
fails the connection.

The option is off unless the client sets it, so that it can be enabled by the
users who test it before it becomes the default. It is a Maybe to keep the
clients that do not send it working. As it is derived from the keep-alive
options where they are set, the two detection times cannot diverge and lowering
the keep-alive settings lowers both.

It is removed for the connections via SOCKS proxy - the socket is then
connected to the proxy, its acknowledgements only mean that the proxy buffered
the data, and a slow circuit should not fail the connection.
---
 src/Simplex/Messaging/Client.hs              |  7 +++--
 src/Simplex/Messaging/Transport/KeepAlive.hs | 31 ++++++++++++++++++--
 2 files changed, 33 insertions(+), 5 deletions(-)

diff --git a/src/Simplex/Messaging/Client.hs b/src/Simplex/Messaging/Client.hs
index 9989a96eb7..07d565261e 100644
--- a/src/Simplex/Messaging/Client.hs
+++ b/src/Simplex/Messaging/Client.hs
@@ -151,7 +151,7 @@ import Data.Int (Int64)
 import Data.List (find, isSuffixOf)
 import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.List.NonEmpty as L
-import Data.Maybe (catMaybes, fromMaybe)
+import Data.Maybe (catMaybes, fromMaybe, isJust)
 import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Time.Clock (UTCTime (..), diffUTCTime, getCurrentTime)
@@ -443,9 +443,12 @@ defaultNetworkConfig =
 
 transportClientConfig :: NetworkConfig -> NetworkRequestMode -> TransportHost -> Bool -> Maybe [ALPN] -> TransportClientConfig
 transportClientConfig NetworkConfig {socksProxy, socksMode, tcpConnectTimeout, tcpKeepAlive, logTLSErrors} nm host useSNI clientALPN =
-  TransportClientConfig {socksProxy = useSocksProxy socksMode, tcpConnectTimeout = tOut, tcpKeepAlive, logTLSErrors, clientCredentials = Nothing, clientALPN, useSNI}
+  TransportClientConfig {socksProxy = socks, tcpConnectTimeout = tOut, tcpKeepAlive = keepAlive, logTLSErrors, clientCredentials = Nothing, clientALPN, useSNI}
   where
     tOut = netTimeoutInt tcpConnectTimeout nm
+    socks = useSocksProxy socksMode
+    -- with SOCKS proxy the socket is connected to the proxy, and a slow circuit should not fail the connection
+    keepAlive = (\ka -> if isJust socks then ka {unackedDataTimeout = Nothing} else ka) <$> tcpKeepAlive
     socksProxy' = (\(SocksProxyWithAuth _ proxy) -> proxy) <$> socksProxy
     useSocksProxy SMAlways = socksProxy'
     useSocksProxy SMOnion = case host of
diff --git a/src/Simplex/Messaging/Transport/KeepAlive.hs b/src/Simplex/Messaging/Transport/KeepAlive.hs
index 1d34a27f0a..42059f22d0 100644
--- a/src/Simplex/Messaging/Transport/KeepAlive.hs
+++ b/src/Simplex/Messaging/Transport/KeepAlive.hs
@@ -11,6 +11,7 @@ module Simplex.Messaging.Transport.KeepAlive
   ) where
 
 import qualified Data.Aeson.TH as J
+import Control.Monad (when)
 import Foreign.C (CInt (..))
 import Network.Socket
 import Simplex.Messaging.Parsers (defaultJSON)
@@ -18,16 +19,25 @@ import Simplex.Messaging.Parsers (defaultJSON)
 data KeepAliveOpts = KeepAliveOpts
   { keepIdle :: Int,
     keepIntvl :: Int,
-    keepCnt :: Int
+    keepCnt :: Int,
+    -- | A connection with unacknowledged data gets no keep-alive probes, so the OS fails it only when it
+    -- gives up retransmitting - ~15 minutes on Linux and Android. When this is set such a connection is
+    -- failed after the keep-alive detection time instead. Absent or False - the OS default.
+    unackedDataTimeout :: Maybe Bool
   }
   deriving (Eq, Show)
 
+-- seconds after which a connection with no response to the probes is reported as failed
+keepAliveDetectionTime :: KeepAliveOpts -> Int
+keepAliveDetectionTime KeepAliveOpts {keepIdle, keepIntvl, keepCnt} = keepIdle + keepIntvl * keepCnt
+
 defaultKeepAliveOpts :: KeepAliveOpts
 defaultKeepAliveOpts =
   KeepAliveOpts
     { keepIdle = 30,
       keepIntvl = 15,
-      keepCnt = 4
+      keepCnt = 4,
+      unackedDataTimeout = Nothing
     }
 
 _SOL_TCP :: CInt
@@ -53,8 +63,12 @@ _TCP_KEEPCNT = 16
 
 #if defined(darwin_HOST_OS)
 foreign import capi "netinet/tcp.h value TCP_KEEPALIVE" _TCP_KEEPIDLE :: CInt
+
+foreign import capi "netinet/tcp.h value TCP_RXT_CONNDROPTIME" _TCP_RXT_CONNDROPTIME :: CInt
 #else
 foreign import capi "netinet/tcp.h value TCP_KEEPIDLE" _TCP_KEEPIDLE :: CInt
+
+foreign import capi "netinet/tcp.h value TCP_USER_TIMEOUT" _TCP_USER_TIMEOUT :: CInt
 #endif
 
 foreign import capi "netinet/tcp.h value TCP_KEEPINTVL" _TCP_KEEPINTVL :: CInt
@@ -63,11 +77,22 @@ foreign import capi "netinet/tcp.h value TCP_KEEPCNT" _TCP_KEEPCNT :: CInt
 
 #endif
 
+-- | seconds
+setSocketUnackedDataTimeout :: Socket -> Int -> IO ()
+#if defined(mingw32_HOST_OS)
+setSocketUnackedDataTimeout _ _ = pure ()
+#elif defined(darwin_HOST_OS)
+setSocketUnackedDataTimeout sock = setSocketOption sock (SockOpt _SOL_TCP _TCP_RXT_CONNDROPTIME)
+#else
+setSocketUnackedDataTimeout sock t = setSocketOption sock (SockOpt _SOL_TCP _TCP_USER_TIMEOUT) (t * 1000) -- TCP_USER_TIMEOUT is in milliseconds
+#endif
+
 setSocketKeepAlive :: Socket -> KeepAliveOpts -> IO ()
-setSocketKeepAlive sock KeepAliveOpts {keepCnt, keepIdle, keepIntvl} = do
+setSocketKeepAlive sock ka@KeepAliveOpts {keepCnt, keepIdle, keepIntvl, unackedDataTimeout} = do
   setSocketOption sock KeepAlive 1
   setSocketOption sock (SockOpt _SOL_TCP _TCP_KEEPIDLE) keepIdle
   setSocketOption sock (SockOpt _SOL_TCP _TCP_KEEPINTVL) keepIntvl
   setSocketOption sock (SockOpt _SOL_TCP _TCP_KEEPCNT) keepCnt
+  when (unackedDataTimeout == Just True) $ setSocketUnackedDataTimeout sock $ keepAliveDetectionTime ka
 
 $(J.deriveJSON defaultJSON ''KeepAliveOpts)