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

@@ -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
 

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

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.

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

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}