agent: move subscriptions to pending when reconnecting servers

src/Simplex/Messaging/Agent.hs

@@ -479,7 +479,7 @@ setUserNetworkInfo c@AgentClient {userNetworkInfo, userNetworkUpdated} ni = with
 
 reconnectAllServers :: AgentClient -> IO ()
 reconnectAllServers c = do
-  reconnectServerClients c smpClients
+  withAgentEnv' c $ reconnectSMPServerClients c
   reconnectServerClients c xftpClients
   reconnectServerClients c ntfClients
 

src/Simplex/Messaging/Agent/Client.hs

@@ -33,6 +33,7 @@ module Simplex.Messaging.Agent.Client
     closeAgentClient,
     closeProtocolServerClients,
     reconnectServerClients,
+    reconnectSMPServerClients,
     reconnectSMPServer,
     closeXFTPServerClient,
     runSMPServerTest,
@@ -211,7 +212,7 @@ import Simplex.Messaging.Agent.Stats
 import Simplex.Messaging.Agent.Store
 import Simplex.Messaging.Agent.Store.SQLite (SQLiteStore (..), withTransaction)
 import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
-import Simplex.Messaging.Agent.TRcvQueues (TRcvQueues (getRcvQueues))
+import Simplex.Messaging.Agent.TRcvQueues (TRcvQueues (getRcvQueues), activeToPendingQueues)
 import qualified Simplex.Messaging.Agent.TRcvQueues as RQ
 import Simplex.Messaging.Client
 import qualified Simplex.Messaging.Crypto as C
@@ -241,6 +242,7 @@ import Simplex.Messaging.Protocol
     QueueIdsKeys (..),
     RcvMessage (..),
     RcvNtfPublicDhKey,
+    RecipientId,
     SMPMsgMeta (..),
     SProtocolType (..),
     SenderCanSecure,
@@ -922,6 +924,33 @@ reconnectServerClients :: ProtocolServerClient v err msg => AgentClient -> (Agen
 reconnectServerClients c clientsSel =
   readTVarIO (clientsSel c) >>= mapM_ (forkIO . closeClient_ c)
 
+reconnectSMPServerClients :: AgentClient -> AM' ()
+reconnectSMPServerClients c = do
+  -- 1. swap smpClients to empty map, move active subscriptions to pending
+  (clients, prevActive) <- atomically $ do
+    clients <- smpClients c `swapTVar` M.empty
+    prevActive <- activeToPendingQueues (activeSubs c) (pendingSubs c)
+    pure (clients, prevActive)
+  -- 2. notify DOWN for connections that had active subscriptions
+  let downConns = groupConnsByServer prevActive
+  forM_ (M.toList downConns) $ \(server, connIds) ->
+    liftIO $ notifyDOWN server connIds
+  -- 3. close clients
+  mapM_ (liftIO . forkIO . closeClient_ c) clients
+  -- 4. resubscribe pending subscriptions
+  pending <- readTVarIO (getRcvQueues $ pendingSubs c)
+  forM_ (M.toList pending) $ \((userId, srv, rId), _) ->
+    resubscribeSMPSession c (userId, srv, Just rId)
+  where
+    groupConnsByServer :: Map (UserId, SMPServer, RecipientId) RcvQueue -> Map SMPServer [ConnId]
+    groupConnsByServer = foldl' insertConnId M.empty
+      where
+        insertConnId :: Map SMPServer [ConnId] -> RcvQueue -> Map SMPServer [ConnId]
+        insertConnId acc RcvQueue {server, connId} =
+          M.insertWith (<>) server [connId] acc
+    notifyDOWN :: SMPServer -> [ConnId] -> IO ()
+    notifyDOWN server connIds = atomically $ writeTBQueue (subQ c) ("", "", AEvt SAENone (DOWN server connIds))
+
 reconnectSMPServer :: AgentClient -> UserId -> SMPServer -> IO ()
 reconnectSMPServer c userId srv = do
   cs <- readTVarIO $ smpClients c

src/Simplex/Messaging/Agent/TRcvQueues.hs

@@ -11,6 +11,7 @@ module Simplex.Messaging.Agent.TRcvQueues
     deleteQueue,
     getSessQueues,
     getDelSessQueues,
+    activeToPendingQueues,
     qKey,
   )
 where
@@ -19,6 +20,7 @@ import Control.Concurrent.STM
 import Data.Foldable (foldl')
 import Data.List.NonEmpty (NonEmpty (..), (<|))
 import qualified Data.List.NonEmpty as L
+import Data.Map (Map)
 import qualified Data.Map.Strict as M
 import Simplex.Messaging.Agent.Protocol (ConnId, UserId)
 import Simplex.Messaging.Agent.Store (RcvQueue, StoredRcvQueue (..))
@@ -96,6 +98,23 @@ getDelSessQueues tSess (TRcvQueues qs cs) = do
             Nothing -> (cId : removed, Nothing)
           Nothing -> (removed, Nothing) -- "impossible" in invariant holds, because we get keys from the known queues
 
+-- moves active queues to pending queues and returns queues that were active
+activeToPendingQueues :: TRcvQueues -> TRcvQueues -> STM (Map (UserId, SMPServer, RecipientId) RcvQueue)
+activeToPendingQueues (TRcvQueues aqs acs) (TRcvQueues pqs pcs) = do
+  aqs' <- mergeQueues
+  mergeConns
+  pure aqs'
+  where
+    mergeQueues :: STM (Map (UserId, SMPServer, RecipientId) RcvQueue)
+    mergeQueues = do
+      aqs' <- aqs `swapTVar` M.empty
+      modifyTVar pqs $ \pqs' -> M.union aqs' pqs'
+      pure aqs'
+    mergeConns :: STM ()
+    mergeConns = do
+      acs' <- acs `swapTVar` M.empty
+      modifyTVar pcs $ \pcs' -> M.unionWith (<>) acs' pcs'
+
 isSession :: RcvQueue -> (UserId, SMPServer, Maybe ConnId) -> Bool
 isSession rq (uId, srv, connId_) =
   userId rq == uId && server rq == srv && maybe True (connId rq ==) connId_