agent: restart retry loops when reconnecting all servers

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
@@ -492,7 +492,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
@@ -533,7 +533,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
@@ -712,7 +712,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

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 >> modifyTVar' (networkEpoch c) (+ 1)
+    isCurrent e = (e ==) <$> readTVar networkEventSeq
     notRecentlyChanged ts' i =
       maybe True (\ts -> diffUTCTime ts' ts > i) <$> readTVar userNetworkUpdated
 
@@ -749,6 +774,7 @@ reconnectAllServers c = do
   reconnectServerClients c smpClients
   reconnectServerClients c xftpClients
   reconnectServerClients c ntfClients
+  atomically $ modifyTVar' (networkEpoch c) (+ 1)
 
 -- | Register device notifications token
 registerNtfToken :: AgentClient -> NetworkRequestMode -> DeviceToken -> NotificationsMode -> AE NtfTknStatus
@@ -2443,7 +2469,7 @@ runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
             _ -> internalErr "command requires duplex connection"
         tryCommand action = tryMoveableCommand (action $> CCCompleted)
         tryAtomicCommand action = tryMoveableCommand (action $> CCDeleted)
-        tryMoveableCommand action = withRetryInterval2 ri $ \_ loop -> do
+        tryMoveableCommand action = withRetryEpoch2 ri (readTVar $ networkEpoch c) $ \_ loop -> do
           liftIO $ waitWhileSuspended c
           liftIO $ waitForUserNetwork c
           tryAllErrors action >>= \case
@@ -2625,7 +2651,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

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,10 @@ 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),
     currentSubs :: TSessionSubs,
     removedSubs :: TMap (UserId, SMPServer) (TMap SMP.RecipientId SMPClientError),
@@ -533,6 +539,8 @@ 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
   removedSubs <- TM.emptyIO
@@ -577,6 +585,8 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, entitlements, netCf
         presetServers,
         userNetworkInfo,
         userNetworkUpdated,
+        networkEpoch,
+        networkEventSeq,
         subscrConns,
         currentSubs,
         removedSubs,
@@ -765,11 +775,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 +811,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 ()
@@ -830,7 +848,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
@@ -952,8 +970,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 +1075,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/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