diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index c7b5f5390..5973e9c0d 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/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 diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs index 0467c31f8..d8da2786f 100644 --- a/src/Simplex/Messaging/Agent/Client.hs +++ b/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 diff --git a/src/Simplex/Messaging/Agent/TRcvQueues.hs b/src/Simplex/Messaging/Agent/TRcvQueues.hs index 9ffe325b2..d93de646e 100644 --- a/src/Simplex/Messaging/Agent/TRcvQueues.hs +++ b/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_