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/5] 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/5] 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/5] 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 925545c7bb15edc7dbfd6aebe571afe6b78bec70 Mon Sep 17 00:00:00 2001
From: Narasimha-sc <166327228+Narasimha-sc@users.noreply.github.com>
Date: Sun, 4 Oct 2026 15:02:27 +0000
Subject: [PATCH 4/5] agent: restart the retry loops when the network changes

Closing the clients when the network changes makes the workers fail their
current attempt, but a worker already waiting out its retry interval resumes
only when it expires - up to the maximum interval, so delivery and
subscriptions can take minutes after the network is usable again.

The network epoch is incremented when the clients are closed, and the agent
retry loops restart from the loop's initial interval when it changes - for
delivery and XFTP workers resumed with persisted retry state that is the
persisted interval (withRetryEpoch). Message delivery restarts the fast
interval only, the slow interval for QUOTA errors is unchanged; the
subscription worker keeps its foreground and online resets; retry attempt
counters are not reset. A loop that was waiting for the network sees the epoch
change after its first attempt, so if that attempt fails it retries once more
without delay.

withRetryWait2 now takes the value to sample before the action and a wait that
receives the retry interval mode and returns whether the interval restarts, so
one loop serves the plain, the lock and the epoch variants - the slow interval
retries in STM and falls through to the plain wait, so it is unchanged.
withRetryEpoch2 is withRetryLock2 without the lock, for the command worker,
which retries with two intervals.
---
 src/Simplex/FileTransfer/Agent.hs             |  14 +-
 src/Simplex/Messaging/Agent.hs                |   6 +-
 src/Simplex/Messaging/Agent/Client.hs         |   6 +-
 .../Messaging/Agent/NtfSubSupervisor.hs       |   4 +-
 src/Simplex/Messaging/Agent/RetryInterval.hs  |  93 +++++++++---
 tests/AgentTests/FunctionalAPITests.hs        |  22 +++
 tests/CoreTests/RetryIntervalTests.hs         | 136 +++++++++++++++++-
 7 files changed, 244 insertions(+), 37 deletions(-)

diff --git a/src/Simplex/FileTransfer/Agent.hs b/src/Simplex/FileTransfer/Agent.hs
index aa6bbabf77..6d07ed6c32 100644
--- a/src/Simplex/FileTransfer/Agent.hs
+++ b/src/Simplex/FileTransfer/Agent.hs
@@ -207,7 +207,7 @@ runXFTPRcvWorker c srv Worker {doWork} = do
           rcvWorkerInternalError c rcvFileId rcvFileEntityId redirectEntityId_ (Just fileTmpPath) (INTERNAL "chunk has no replicas")
         (fc@RcvFileChunk {userId, rcvFileId, rcvFileEntityId, digest, fileTmpPath, replicas = replica@RcvFileChunkReplica {rcvChunkReplicaId, server, delay} : _}, approvedRelays, redirectEntityId_) -> do
           let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay
-          withRetryIntervalLimit xftpConsecutiveRetries ri' $ \delay' loop -> do
+          withRetryIntervalLimit xftpConsecutiveRetries ri' (readTVar $ networkEpoch c) $ \delay' loop -> do
             liftIO $ waitWhileSuspended c
             liftIO $ waitForUserNetwork c
             atomically $ incXFTPServerStat c userId srv downloadAttempts
@@ -265,9 +265,9 @@ runXFTPRcvWorker c srv Worker {doWork} = do
         chunkReceived RcvFileChunk {replicas} = any received replicas
 
 -- The first call of action has n == 0, maxN is max number of retries
-withRetryIntervalLimit :: forall m. MonadIO m => Int -> RetryInterval -> (Int64 -> m () -> m ()) -> m ()
-withRetryIntervalLimit maxN ri action =
-  withRetryIntervalCount ri $ \n delay loop ->
+withRetryIntervalLimit :: forall m. MonadIO m => Int -> RetryInterval -> STM Int -> (Int64 -> m () -> m ()) -> m ()
+withRetryIntervalLimit maxN ri getEpoch action =
+  withRetryEpochCount ri getEpoch $ \n delay loop ->
     when (n < maxN) $ action delay loop
 
 retryOnError :: Text -> AM a -> AM a -> AgentErrorType -> AM a
@@ -490,7 +490,7 @@ runXFTPSndPrepareWorker c Worker {doWork} = do
               triedHosts <- newTVarIO S.empty
               let AgentClient {xftpServers} = c
               userSrvCount <- liftIO $ length <$> TM.lookupIO userId xftpServers
-              withRetryIntervalCount (riFast ri) $ \n _ loop -> do
+              withRetryEpochCount (riFast ri) (readTVar $ networkEpoch c) $ \n _ loop -> do
                 liftIO $ waitWhileSuspended c
                 liftIO $ waitForUserNetwork c
                 let triedAllSrvs = n > userSrvCount
@@ -530,7 +530,7 @@ runXFTPSndWorker c srv Worker {doWork} = do
         SndFileChunk {sndFileId, sndFileEntityId, filePrefixPath, replicas = []} -> sndWorkerInternalError c sndFileId sndFileEntityId (Just filePrefixPath) (INTERNAL "chunk has no replicas")
         fc@SndFileChunk {userId, sndFileId, sndFileEntityId, filePrefixPath, digest, replicas = replica@SndFileChunkReplica {sndChunkReplicaId, server, delay} : _} -> do
           let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay
-          withRetryIntervalLimit xftpConsecutiveRetries ri' $ \delay' loop -> do
+          withRetryIntervalLimit xftpConsecutiveRetries ri' (readTVar $ networkEpoch c) $ \delay' loop -> do
             liftIO $ waitWhileSuspended c
             liftIO $ waitForUserNetwork c
             atomically $ incXFTPServerStat c userId srv uploadAttempts
@@ -708,7 +708,7 @@ runXFTPDelWorker c srv Worker {doWork} = do
       where
         processDeletedReplica replica@DeletedSndChunkReplica {deletedSndChunkReplicaId, userId, server, chunkDigest, delay} = do
           let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay
-          withRetryIntervalLimit xftpConsecutiveRetries ri' $ \delay' loop -> do
+          withRetryIntervalLimit xftpConsecutiveRetries ri' (readTVar $ networkEpoch c) $ \delay' loop -> do
             liftIO $ waitWhileSuspended c
             liftIO $ waitForUserNetwork c
             atomically $ incXFTPServerStat c userId srv deleteAttempts
diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs
index d378566b98..e3c062fbf2 100644
--- a/src/Simplex/Messaging/Agent.hs
+++ b/src/Simplex/Messaging/Agent.hs
@@ -744,7 +744,7 @@ setUserNetworkInfo c@AgentClient {userNetworkInfo, userNetworkUpdated, networkEv
       whenCurrent $ closeProtocolServerClients c xftpClients
       where
         whenCurrent = whenM (atomically $ isCurrent e)
-    networkOnline e = atomically $ whenM (isCurrent e) $ writeTVar userNetworkInfo ni
+    networkOnline e = atomically $ whenM (isCurrent e) $ writeTVar userNetworkInfo ni >> modifyTVar' (networkEpoch c) (+ 1)
     isCurrent e = (e ==) <$> readTVar networkEventSeq
     notRecentlyChanged ts' i =
       maybe True (\ts -> diffUTCTime ts' ts > i) <$> readTVar userNetworkUpdated
@@ -2398,7 +2398,7 @@ runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
             SomeConn _ conn@DuplexConnection {} -> a conn
             _ -> internalErr "command requires duplex connection"
         tryCommand action = tryMoveableCommand (action $> CCCompleted)
-        tryMoveableCommand action = withRetryInterval2 ri $ \_ loop -> do
+        tryMoveableCommand action = withRetryEpoch2 ri (readTVar $ networkEpoch c) $ \_ loop -> do
           liftIO $ waitWhileSuspended c
           liftIO $ waitForUserNetwork c
           tryAllErrors action >>= \case
@@ -2576,7 +2576,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} sq@SndQueue {userId, connId, server,
         atomically $ endAgentOperation c AOMsgDelivery -- this operation begins in submitPendingMsg
         let mId = unId msgId
             ri' = maybe id updateRetryInterval2 msgRetryState ri
-        withRetryLock2 ri' qLock $ \riState loop -> do
+        withRetryLock2 ri' (readTVar $ networkEpoch c) qLock $ \riState loop -> do
           liftIO $ waitWhileSuspended c
           liftIO $ waitForUserNetwork c
           resp <- tryAllErrors $ case msgType of
diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs
index 227663df44..8b11ae0360 100644
--- a/src/Simplex/Messaging/Agent/Client.hs
+++ b/src/Simplex/Messaging/Agent/Client.hs
@@ -365,6 +365,8 @@ data AgentClient = AgentClient
     presetServers :: [SMPServer],
     userNetworkInfo :: TVar UserNetworkInfo,
     userNetworkUpdated :: TVar (Maybe UTCTime),
+    -- incremented when the clients are closed after the network changed, see withRetryEpoch
+    networkEpoch :: TVar Int,
     -- incremented for each accepted network info, to identify the most recent one
     networkEventSeq :: TVar Int,
     subscrConns :: TVar (Set ConnId),
@@ -539,6 +541,7 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, entitlements, netCf
   useNetworkConfig <- newTVarIO (slowNetworkConfig netCfg, netCfg)
   userNetworkInfo <- newTVarIO $ UserNetworkInfo UNOther True
   userNetworkUpdated <- newTVarIO Nothing
+  networkEpoch <- newTVarIO 0
   networkEventSeq <- newTVarIO 0
   subscrConns <- newTVarIO S.empty
   currentSubs <- SS.emptyIO
@@ -584,6 +587,7 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, entitlements, netCf
         presetServers,
         userNetworkInfo,
         userNetworkUpdated,
+        networkEpoch,
         networkEventSeq,
         subscrConns,
         currentSubs,
@@ -846,7 +850,7 @@ resubscribeSMPSession c@AgentClient {smpSubWorkers, workerSeq} tSess = do
       atomically $ putTMVar (sessionVar v) a
     runSubWorker v = do
       ri <- asks $ reconnectInterval . config
-      withRetryForeground ri isForeground (isNetworkOnline c) $ \_ loop -> do
+      withRetryForeground ri isForeground (isNetworkOnline c) (readTVar $ networkEpoch c) $ \_ loop -> do
         pending_ <- atomically $ do
           pending@(pendingSubs, pendingSS) <- SS.getPendingSubs tSess $ currentSubs c
           if M.null pendingSubs && isNothing pendingSS
diff --git a/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs b/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs
index 06c9b4ca44..9c3145cbf6 100644
--- a/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs
+++ b/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs
@@ -456,7 +456,7 @@ retrySubActions _ [] _ = pure ()
 retrySubActions c subs action = do
   v <- newTVarIO subs
   ri <- asks $ reconnectInterval . config
-  withRetryInterval ri $ \_ loop -> do
+  withRetryEpoch ri (readTVar $ networkEpoch c) $ \_ loop -> do
     liftIO $ waitWhileSuspended c
     liftIO $ waitForUserNetwork c
     subs' <- readTVarIO v
@@ -563,7 +563,7 @@ runNtfTknDelWorker c srv Worker {doWork} =
         \nextTknToDelete -> do
           logInfo $ "runNtfTknDelWorker, nextTknToDelete " <> tshow nextTknToDelete
           ri <- asks $ reconnectInterval . config
-          withRetryInterval ri $ \_ loop -> do
+          withRetryEpoch ri (readTVar $ networkEpoch c) $ \_ loop -> do
             liftIO $ waitWhileSuspended c
             liftIO $ waitForUserNetwork c
             processTknToDelete nextTknToDelete `catchAllErrors` retryTmpError loop nextTknToDelete
diff --git a/src/Simplex/Messaging/Agent/RetryInterval.hs b/src/Simplex/Messaging/Agent/RetryInterval.hs
index 4ddc220daf..f6447fd149 100644
--- a/src/Simplex/Messaging/Agent/RetryInterval.hs
+++ b/src/Simplex/Messaging/Agent/RetryInterval.hs
@@ -9,6 +9,9 @@ module Simplex.Messaging.Agent.RetryInterval
     RI2State (..),
     withRetryInterval,
     withRetryIntervalCount,
+    withRetryEpoch,
+    withRetryEpochCount,
+    withRetryEpoch2,
     withRetryForeground,
     withRetryInterval2,
     withRetryLock2,
@@ -66,20 +69,50 @@ withRetryIntervalCount ri action = callAction 0 0 $ initialInterval ri
           let elapsed' = elapsed + delay
           callAction (n + 1) elapsed' $ nextRetryDelay elapsed' delay ri
 
-withRetryForeground :: forall m a. MonadIO m => RetryInterval -> STM Bool -> STM Bool -> (Int64 -> m a -> m a) -> m a
-withRetryForeground ri isForeground isOnline action = callAction 0 $ initialInterval ri
+-- The delay restarts from the initial interval when the epoch changes (the agent increments it when the network changes)
+-- while the loop waits; when it changed while the action was running the next attempt is made without any delay.
+withRetryEpoch :: forall m a. MonadIO m => RetryInterval -> STM Int -> (Int64 -> m a -> m a) -> m a
+withRetryEpoch ri getEpoch = withRetryEpochCount ri getEpoch . const
+
+withRetryEpochCount :: forall m a. MonadIO m => RetryInterval -> STM Int -> (Int -> Int64 -> m a -> m a) -> m a
+withRetryEpochCount ri getEpoch action = callAction 0 0 $ initialInterval ri
+  where
+    callAction :: Int -> Int64 -> Int64 -> m a
+    callAction n elapsed delay = do
+      epoch <- atomically getEpoch
+      action n delay $ do
+        reset <- waitEpoch' epoch delay
+        let (elapsed', delay')
+              | reset = (0, initialInterval ri)
+              | otherwise = (elapsed + delay, nextRetryDelay elapsed' delay ri)
+        callAction (n + 1) elapsed' delay'
+    waitEpoch' = waitEpoch getEpoch
+
+-- returns True when the epoch changed before the delay expired
+waitEpoch :: MonadIO m => STM Int -> Int -> Int64 -> m Bool
+waitEpoch getEpoch epoch delay = do
+  -- limit delay to max Int value (~36 minutes on for 32 bit architectures)
+  d <- registerDelay $ fromIntegral $ min delay (fromIntegral (maxBound :: Int))
+  atomically $ do
+    changed <- (epoch /=) <$> getEpoch
+    unlessM ((changed ||) <$> readTVar d) retry
+    pure changed
+
+withRetryForeground :: forall m a. MonadIO m => RetryInterval -> STM Bool -> STM Bool -> STM Int -> (Int64 -> m a -> m a) -> m a
+withRetryForeground ri isForeground isOnline getEpoch action = callAction 0 $ initialInterval ri
   where
     callAction :: Int64 -> Int64 -> m a
-    callAction elapsed delay = action delay loop
+    callAction elapsed delay = action delay . loop =<< atomically getEpoch
       where
-        loop = do
+        loop epoch = do
           -- limit delay to max Int value (~36 minutes on for 32 bit architectures)
           d <- registerDelay $ fromIntegral $ min delay (fromIntegral (maxBound :: Int))
           (wasForeground, wasOnline) <- atomically $ (,) <$> isForeground <*> isOnline
           reset <- atomically $ do
             foreground <- isForeground
             online <- isOnline
-            let reset = (not wasForeground && foreground) || (not wasOnline && online)
+            epochChanged <- (epoch /=) <$> getEpoch
+            let reset = (not wasForeground && foreground) || (not wasOnline && online) || epochChanged
             unlessM ((reset ||) <$> readTVar d) retry
             pure reset
           let (elapsed', delay')
@@ -88,35 +121,55 @@ withRetryForeground ri isForeground isOnline action = callAction 0 $ initialInte
           callAction elapsed' delay'
 
 withRetryInterval2 :: forall m. MonadIO m => RetryInterval2 -> (RI2State -> (RetryIntervalMode -> m ()) -> m ()) -> m ()
-withRetryInterval2 = withRetryWait2 $ liftIO . threadDelay'
+withRetryInterval2 = withRetryWait2 (pure ()) $ \_ _ delay -> False <$ liftIO (threadDelay' delay)
 
 -- This function allows action to toggle between slow and fast retry intervals.
-withRetryLock2 :: forall m. MonadIO m => RetryInterval2 -> TMVar () -> (RI2State -> (RetryIntervalMode -> m ()) -> m ()) -> m ()
-withRetryLock2 ri lock = withRetryWait2 wait ri
+-- The fast interval restarts on the epoch change, as in withRetryEpoch; once the action chose the slow
+-- interval (recipient queue quota) the epoch change is not applied to the fast interval - neither the one
+-- that happened during that action nor during the wait.
+withRetryLock2 :: forall m. MonadIO m => RetryInterval2 -> STM Int -> TMVar () -> (RI2State -> (RetryIntervalMode -> m ()) -> m ()) -> m ()
+withRetryLock2 ri getEpoch lock = withRetryWait2 (atomically getEpoch) wait ri
   where
-    wait delay = do
+    wait mode epoch delay = do
       waiting <- newTVarIO True
       _ <- liftIO . forkIO $ do
         threadDelay' delay
         atomically $ whenM (readTVar waiting) $ void $ tryPutTMVar lock ()
       atomically $ do
-        takeTMVar lock
+        reset <- (epochChanged <* tryTakeTMVar lock) `orElse` (False <$ takeTMVar lock)
         writeTVar waiting False
+        pure reset
+      where
+        epochChanged = case mode of
+          RIFast -> unlessM ((epoch /=) <$> getEpoch) retry >> pure True
+          RISlow -> retry
 
-withRetryWait2 :: forall m. Monad m => (Int64 -> m ()) -> RetryInterval2 -> (RI2State -> (RetryIntervalMode -> m ()) -> m ()) -> m ()
-withRetryWait2 wait RetryInterval2 {riSlow, riFast} action =
+-- As withRetryLock2, without the lock: the fast interval restarts on the epoch change, the slow one
+-- (recipient queue quota) does not.
+withRetryEpoch2 :: forall m. MonadIO m => RetryInterval2 -> STM Int -> (RI2State -> (RetryIntervalMode -> m ()) -> m ()) -> m ()
+withRetryEpoch2 ri getEpoch = withRetryWait2 (atomically getEpoch) wait ri
+  where
+    wait mode epoch delay = case mode of
+      RISlow -> False <$ liftIO (threadDelay' delay)
+      RIFast -> waitEpoch getEpoch epoch delay
+
+-- the value sampled before the action is passed to wait, so that a change that happened during the
+-- action resets the interval as one during the wait does
+withRetryWait2 :: forall m e. Monad m => m e -> (RetryIntervalMode -> e -> Int64 -> m Bool) -> RetryInterval2 -> (RI2State -> (RetryIntervalMode -> m ()) -> m ()) -> m ()
+withRetryWait2 sample wait RetryInterval2 {riSlow, riFast} action =
   callAction (0, initialInterval riSlow) (0, initialInterval riFast)
   where
     callAction :: (Int64, Int64) -> (Int64, Int64) -> m ()
-    callAction slow fast = action (RI2State (snd slow) (snd fast)) loop
+    callAction slow fast = sample >>= \e -> action (RI2State (snd slow) (snd fast)) (loop e)
       where
-        loop = \case
-          RISlow -> run slow riSlow (`callAction` fast)
-          RIFast -> run fast riFast (callAction slow)
-        run (elapsed, delay) ri call = do
-          wait delay
-          let elapsed' = elapsed + delay
-              delay' = nextRetryDelay elapsed' delay ri
+        loop e = \case
+          RISlow -> run RISlow e slow riSlow (`callAction` fast)
+          RIFast -> run RIFast e fast riFast (callAction slow)
+        run mode e (elapsed, delay) ri call = do
+          reset <- wait mode e delay
+          let (elapsed', delay')
+                | reset = (0, initialInterval ri)
+                | otherwise = (elapsed + delay, nextRetryDelay elapsed' delay ri)
           call (elapsed', delay')
 
 nextRetryDelay :: Int64 -> Int64 -> RetryInterval -> Int64
diff --git a/tests/AgentTests/FunctionalAPITests.hs b/tests/AgentTests/FunctionalAPITests.hs
index 0cd91c4461..88769674c4 100644
--- a/tests/AgentTests/FunctionalAPITests.hs
+++ b/tests/AgentTests/FunctionalAPITests.hs
@@ -608,6 +608,7 @@ functionalAPITests ps = 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 retry delivery and subscription immediately when network changes" $ testNetworkChangeRetry ps
     it "should resume multiple threads" testResumeMultipleThreads
   describe "SMP queue info" $ do
     it "server should respond with queue and subscription information" $
@@ -4817,6 +4818,27 @@ testNetworkChangeReconnect ps =
     nGet a =##> \case ("", "", UP _ [c]) -> c == bId; _ -> False
     runRight_ $ exchangeGreetings a bId b aId
 
+testNetworkChangeRetry :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
+testNetworkChangeRetry ps =
+  withAgentClientsCfg2 aCfg agentCfg $ \a b -> do
+    (aId, bId) <- withSmpServerStoreLogOn ps testPort $ \_ -> runRight $ makeConnection a b
+    nGet a =##> \case ("", "", DOWN _ [c]) -> c == bId; _ -> False
+    nGet b =##> \case ("", "", DOWN _ [c]) -> c == aId; _ -> False
+    -- the first delivery attempt fails, and the delivery worker waits for the configured 60 seconds
+    2 <- runRight $ sendMessage a bId SMP.noMsgFlags "hello"
+    threadDelay 500000
+    withSmpServerStoreLogOn ps testPort $ \_ -> do
+      -- without network change event both the message delivery and the subscription
+      -- would be retried in 60 seconds, the test would fail with timeout
+      setUserNetworkInfo a $ UserNetworkInfo UNCellular True
+      runRight_ $ do
+        withUP a bId $ \case ("", c, SENT 2) -> c == bId; _ -> False
+        withUP b aId $ \case ("", c, Msg "hello") -> c == aId; _ -> False
+        ackMessage b aId 2 Nothing
+  where
+    slowRI = RetryInterval {initialInterval = 60_000000, increaseAfter = 0, maxInterval = 60_000000}
+    aCfg = agentCfg {reconnectInterval = slowRI, messageRetryInterval = RetryInterval2 {riSlow = slowRI, riFast = slowRI}}
+
 testDoNotResetOnlineToOffline :: IO ()
 testDoNotResetOnlineToOffline = withAgent 1 aCfg initAgentServers testDB $ \a -> do
   noNetworkDelay a
diff --git a/tests/CoreTests/RetryIntervalTests.hs b/tests/CoreTests/RetryIntervalTests.hs
index d1ab971249..780866d7dc 100644
--- a/tests/CoreTests/RetryIntervalTests.hs
+++ b/tests/CoreTests/RetryIntervalTests.hs
@@ -6,6 +6,7 @@ import Control.Concurrent (threadDelay)
 import Control.Concurrent.Async (concurrently_)
 import Control.Concurrent.STM
 import Control.Monad (when)
+import Data.Int (Int64)
 import Data.Time.Clock (UTCTime, diffUTCTime, getCurrentTime, nominalDiffTimeToSeconds)
 import Simplex.Messaging.Agent.RetryInterval
 import Test.Hspec hiding (fit, it)
@@ -16,13 +17,23 @@ retryIntervalTests = do
   describe "Retry interval with 2 modes and lock" $ do
     testRetryIntervalSameMode withRetryNewLock2
     testRetryIntervalSwitchMode withRetryNewLock2
+    testRetryIntervalEpochReset
   describe "Retry interval with 2 modes" $ do
     testRetryIntervalSameMode withRetryInterval2
     testRetryIntervalSwitchMode withRetryInterval2
+  describe "Retry interval with 2 modes and epoch" $ do
+    testRetryIntervalSameMode (`withRetryEpoch2` pure 0)
+    testRetryIntervalSwitchMode (`withRetryEpoch2` pure 0)
   describe "Foreground retry interval" $ do
     testRetryForeground
     testRetryToBackground
     testRetrySkipWhenForeground
+    testRetryForegroundEpoch
+  describe "Retry interval with epoch" $ do
+    testRetryEpochSameValue
+    testRetrySkipOnEpochChange
+    testRetryEpochChangedInAction
+    testRetryEpochKeepsCount
 
 testRI :: RetryInterval2
 testRI =
@@ -47,7 +58,7 @@ testFastRI =
 type WithRetry2 = RetryInterval2 -> (RI2State -> (RetryIntervalMode -> IO ()) -> IO ()) -> IO ()
 
 withRetryNewLock2 :: WithRetry2
-withRetryNewLock2 ri action = newEmptyTMVarIO >>= \lock -> withRetryLock2 ri lock action
+withRetryNewLock2 ri action = newEmptyTMVarIO >>= \lock -> withRetryLock2 ri (pure 0) lock action
 
 testRetryIntervalSameMode :: WithRetry2 -> Spec
 testRetryIntervalSameMode withRetry =
@@ -97,6 +108,41 @@ testRetryIntervalSwitchMode withRetry =
                        (40000, 40000)
                      ]
 
+testRetryIntervalEpochReset :: Spec
+testRetryIntervalEpochReset =
+  it "should restart fast interval when epoch changes, but not slow interval" $ do
+    lock <- newEmptyTMVarIO
+    epoch <- newTVarIO (0 :: Int)
+    intervals <- newTVarIO []
+    reportedIntervals <- newTVarIO []
+    ts <- newTVarIO =<< getCurrentTime
+    concurrently_
+      ( do
+          -- fast mode: waiting for 33750, restarts immediately at 65ms with the initial interval
+          threadDelay 65000
+          atomically $ modifyTVar' epoch (+ 1)
+          -- slow mode: waiting for 20000 from 95ms, the change is ignored
+          threadDelay 40000
+          atomically $ modifyTVar' epoch (+ 1)
+      )
+      ( withRetryLock2 testRI (readTVar epoch) lock $ \(RI2State slow fast) loop -> do
+          ints <- addInterval intervals ts
+          atomically $ modifyTVar' reportedIntervals ((slow, fast) :)
+          when (length ints < 9) $ loop $ if length ints <= 6 then RIFast else RISlow
+      )
+    (reverse <$> readTVarIO intervals) `shouldReturn` [0, 1, 1, 1, 2, 0, 1, 2, 2]
+    (reverse <$> readTVarIO reportedIntervals)
+      `shouldReturn` [ (20000, 10000),
+                       (20000, 10000),
+                       (20000, 15000),
+                       (20000, 22500),
+                       (20000, 33750),
+                       (20000, 10000),
+                       (20000, 10000),
+                       (20000, 10000),
+                       (30000, 10000)
+                     ]
+
 testRetryForeground :: Spec
 testRetryForeground =
   it "should increase elapased time and interval" $ do
@@ -104,7 +150,7 @@ testRetryForeground =
     reportedIntervals <- newTVarIO []
     ts <- newTVarIO =<< getCurrentTime
     let isForeground = pure True
-    withRetryForeground testFastRI isForeground (pure True) $ \delay loop -> do
+    withRetryForeground testFastRI isForeground (pure True) (pure 0) $ \delay loop -> do
       ints <- addInterval intervals ts
       atomically $ modifyTVar' reportedIntervals (delay :)
       when (length ints < 8) $ loop
@@ -124,7 +170,7 @@ testRetryToBackground =
           threadDelay 50000
           atomically $ writeTVar foreground False
       )
-      ( withRetryForeground testFastRI (readTVar foreground) (pure True) $ \delay loop -> do
+      ( withRetryForeground testFastRI (readTVar foreground) (pure True) (pure 0) $ \delay loop -> do
           ints <- addInterval intervals ts
           atomically $ modifyTVar' reportedIntervals (delay :)
           when (length ints < 8) $ loop
@@ -149,7 +195,89 @@ testRetrySkipWhenForeground =
           threadDelay 100000
           atomically $ writeTVar foreground True
       )
-      ( withRetryForeground testFastRI (readTVar foreground) (pure True) $ \delay loop -> do
+      ( withRetryForeground testFastRI (readTVar foreground) (pure True) (pure 0) $ \delay loop -> do
+          ints <- addInterval intervals ts
+          atomically $ modifyTVar' reportedIntervals (delay :)
+          when (length ints < 12) $ loop
+      )
+    (reverse <$> readTVarIO intervals) `shouldReturn` [0, 1, 1, 1, 2, 0, 1, 1, 1, 2, 3, 1]
+    (reverse <$> readTVarIO reportedIntervals)
+      `shouldReturn` [10000, 10000, 15000, 22500, 33750, 10000, 10000, 15000, 22500, 33750, 40000, 10000]
+
+testRetryForegroundEpoch :: Spec
+testRetryForegroundEpoch =
+  testEpochRestartsLoop "should repeat foreground loop as soon as epoch changes" $ \e ->
+    withRetryForeground testFastRI (pure True) (pure True) (readTVar e)
+
+testRetryEpochSameValue :: Spec
+testRetryEpochSameValue =
+  it "should not change interval when epoch is written with the same value" $ do
+    intervals <- newTVarIO []
+    reportedIntervals <- newTVarIO []
+    ts <- newTVarIO =<< getCurrentTime
+    epoch <- newTVarIO (0 :: Int)
+    concurrently_
+      ( do
+          threadDelay 50000
+          atomically $ writeTVar epoch 0
+      )
+      ( withRetryEpoch testFastRI (readTVar epoch) $ \delay loop -> do
+          ints <- addInterval intervals ts
+          atomically $ modifyTVar' reportedIntervals (delay :)
+          when (length ints < 8) $ loop
+      )
+    (reverse <$> readTVarIO intervals) `shouldReturn` [0, 1, 1, 1, 2, 3, 4, 4]
+    (reverse <$> readTVarIO reportedIntervals)
+      `shouldReturn` [10000, 10000, 15000, 22500, 33750, 40000, 40000, 40000]
+
+testRetryEpochChangedInAction :: Spec
+testRetryEpochChangedInAction =
+  it "should restart interval when epoch changes while action runs" $ do
+    intervals <- newTVarIO []
+    reportedIntervals <- newTVarIO []
+    ts <- newTVarIO =<< getCurrentTime
+    epoch <- newTVarIO (0 :: Int)
+    withRetryEpoch testFastRI (readTVar epoch) $ \delay loop -> do
+      ints <- addInterval intervals ts
+      atomically $ modifyTVar' reportedIntervals (delay :)
+      when (length ints == 3) $ atomically $ modifyTVar' epoch (+ 1)
+      when (length ints < 6) $ loop
+    (reverse <$> readTVarIO intervals) `shouldReturn` [0, 1, 1, 0, 1, 1]
+    (reverse <$> readTVarIO reportedIntervals)
+      `shouldReturn` [10000, 10000, 15000, 10000, 10000, 15000]
+
+testRetryEpochKeepsCount :: Spec
+testRetryEpochKeepsCount =
+  it "should not reset attempt count when epoch changes" $ do
+    counts <- newTVarIO ([] :: [Int])
+    epoch <- newTVarIO (0 :: Int)
+    withRetryEpochCount testFastRI (readTVar epoch) $ \n _ loop -> do
+      ns <- atomically $ stateTVar counts $ \ns -> (n : ns, n : ns)
+      when (length ns == 2) $ atomically $ modifyTVar' epoch (+ 1)
+      when (length ns < 4) loop
+    (reverse <$> readTVarIO counts) `shouldReturn` [0, 1, 2, 3]
+
+testRetrySkipOnEpochChange :: Spec
+testRetrySkipOnEpochChange =
+  testEpochRestartsLoop "should repeat loop as soon as epoch changes" $ \e ->
+    withRetryEpoch testFastRI (readTVar e)
+
+-- the epoch changes twice while the loop waits, so the interval restarts from the initial one twice
+testEpochRestartsLoop :: String -> (TVar Int -> (Int64 -> IO () -> IO ()) -> IO ()) -> Spec
+testEpochRestartsLoop name withRetry =
+  it name $ do
+    intervals <- newTVarIO []
+    reportedIntervals <- newTVarIO []
+    ts <- newTVarIO =<< getCurrentTime
+    epoch <- newTVarIO (0 :: Int)
+    concurrently_
+      ( do
+          threadDelay 65000
+          atomically $ modifyTVar' epoch (+ 1)
+          threadDelay 110000
+          atomically $ modifyTVar' epoch (+ 1)
+      )
+      ( withRetry epoch $ \delay loop -> do
           ints <- addInterval intervals ts
           atomically $ modifyTVar' reportedIntervals (delay :)
           when (length ints < 12) $ loop

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 5/5] 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)