agent: fix three races that closing clients on a network change makes reachable

src/Simplex/Messaging/Agent.hs

@@ -269,7 +269,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
 
@@ -733,14 +733,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
 

src/Simplex/Messaging/Agent/Client.hs

@@ -21,6 +21,7 @@
 {-# LANGUAGE TupleSections #-}
 {-# LANGUAGE TypeApplications #-}
 {-# LANGUAGE TypeFamilyDependencies #-}
+{-# LANGUAGE TypeOperators #-}
 
 module Simplex.Messaging.Agent.Client
   ( AgentClient (..),
@@ -147,6 +148,7 @@ module Simplex.Messaging.Agent.Client
     UserNetworkType (..),
     getFastNetworkConfig,
     waitForUserNetwork,
+    closeNetworkClients,
     isNetworkOnline,
     isOnline,
     throwWhenInactive,
@@ -362,6 +364,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),
@@ -533,6 +537,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
@@ -577,6 +582,7 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, entitlements, netCf
         presetServers,
         userNetworkInfo,
         userNetworkUpdated,
+        networkEventSeq,
         subscrConns,
         currentSubs,
         removedSubs,
@@ -765,11 +771,17 @@ 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
+      (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
@@ -795,9 +807,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 ()
@@ -952,8 +966,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
@@ -1047,6 +1071,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

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

src/Simplex/Messaging/Client.hs

@@ -35,6 +35,8 @@ module Simplex.Messaging.Client
     ProxiedRelay (..),
     getProtocolClient,
     closeProtocolClient,
+    failPendingRequests,
+    clientConnected,
     protocolClientServer,
     protocolClientServer',
     transportHost',
@@ -151,7 +153,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 +445,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
@@ -652,10 +657,10 @@ 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` (failPendingRequests c >> disconnected c')
+            `E.finally` (atomically (writeTVar (connected c) False) >> failSentCommands c >> disconnected c')
 
-    failPendingRequests :: PClient v err msg -> IO ()
-    failPendingRequests PClient {sentCommands} = atomically $ swapTVar sentCommands mempty >>= mapM_ failRequest
+    failSentCommands :: PClient v err msg -> IO ()
+    failSentCommands PClient {sentCommands} = atomically $ swapTVar sentCommands mempty >>= mapM_ failRequest
       where
         failRequest Request {pending, responseVar} =
           whenM (swapTVar pending False) $ void $ tryPutTMVar responseVar $ Left $ PCENetworkError NEFailedError
@@ -763,6 +768,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.

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

src/Simplex/Messaging/Transport/KeepAlive.hs

@@ -11,23 +11,33 @@ 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)
 
 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)