agent: move subscriptions to pending when reconnecting servers

src/Simplex/Messaging/Agent.hs

@@ -14,7 +14,7 @@
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TupleSections #-}
 {-# LANGUAGE TypeApplications #-}
 {-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
 
 -- |
 -- Module      : Simplex.Messaging.Agent
@@ -479,7 +479,7 @@
 
 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 @@
     closeAgentClient,
     closeProtocolServerClients,
     reconnectServerClients,
+    reconnectSMPServerClients,
     reconnectSMPServer,
     closeXFTPServerClient,
     runSMPServerTest,
@@ -922,6 +923,30 @@
 reconnectServerClients c clientsSel =
   readTVarIO (clientsSel c) >>= mapM_ (forkIO . closeClient_ c)
 
+reconnectSMPServerClients :: AgentClient -> AM' ()
+reconnectSMPServerClients c = do
+  (clients, qs) <- atomically $ do
+    clients <- swapTVar (smpClients c) M.empty
+    qs <- RQ.getDelAllQueues (activeSubs c)
+    qs' <- RQ.getDelAllQueues (pendingSubs c)
+    pure (clients, qs <> qs')
+  atomically $ writeTBQueue (subQ c) ("", "", AEvt SAENone DOWN_ALL)
+  mapM_ (liftIO . forkIO . closeClient_ c) clients
+  (qSubRs, _) <- subscribeQueues c qs
+  let upConns = subscribedConnsByServer qSubRs
+  forM_ (M.toList upConns) $ \(server, connIds) ->
+    liftIO $ notifyUP server (S.toList . S.fromList $ connIds)
+  where
+    subscribedConnsByServer :: [(RcvQueue, Either AgentErrorType ())] -> Map SMPServer [ConnId]
+    subscribedConnsByServer = foldl' insertConnId M.empty
+      where
+        insertConnId :: Map SMPServer [ConnId] -> (RcvQueue, Either AgentErrorType ()) -> Map SMPServer [ConnId]
+        insertConnId acc (RcvQueue {server, connId}, qSubResult) = case qSubResult of
+          Right _ -> M.insertWith (<>) server [connId] acc
+          Left _ -> acc
+    notifyUP :: SMPServer -> [ConnId] -> IO ()
+    notifyUP server connIds = atomically $ writeTBQueue (subQ c) ("", "", AEvt SAENone (UP server connIds))
+
 reconnectSMPServer :: AgentClient -> UserId -> SMPServer -> IO ()
 reconnectSMPServer c userId srv = do
   cs <- readTVarIO $ smpClients c
@@ -1904,7 +1929,7 @@
   SEDatabaseBusy e -> CRITICAL True $ B.unpack e
   e -> INTERNAL $ show e
 
 userServers :: forall p. (ProtocolTypeI p, UserProtocol p) => AgentClient -> TMap UserId (NonEmpty (ProtoServerWithAuth p))
 userServers c = case protocolTypeI @p of
   SPSMP -> smpServers c
   SPXFTP -> xftpServers c

src/Simplex/Messaging/Agent/Protocol.hs

@@ -338,6 +338,7 @@ data AEvent (e :: AEntity) where
   CONNECT :: AProtocolType -> TransportHost -> AEvent AENone
   DISCONNECT :: AProtocolType -> TransportHost -> AEvent AENone
   DOWN :: SMPServer -> [ConnId] -> AEvent AENone
+  DOWN_ALL :: AEvent AENone
   UP :: SMPServer -> [ConnId] -> AEvent AENone
   SWITCH :: QueueDirection -> SwitchPhase -> ConnectionStats -> AEvent AEConn
   RSYNC :: RatchetSyncState -> Maybe AgentCryptoError -> ConnectionStats -> AEvent AEConn
@@ -406,6 +407,7 @@ data AEventTag (e :: AEntity) where
   CONNECT_ :: AEventTag AENone
   DISCONNECT_ :: AEventTag AENone
   DOWN_ :: AEventTag AENone
+  DOWN_ALL_ :: AEventTag AENone
   UP_ :: AEventTag AENone
   SWITCH_ :: AEventTag AEConn
   RSYNC_ :: AEventTag AEConn
@@ -458,6 +460,7 @@ aEventTag = \case
   CONNECT {} -> CONNECT_
   DISCONNECT {} -> DISCONNECT_
   DOWN {} -> DOWN_
+  DOWN_ALL {} -> DOWN_ALL_
   UP {} -> UP_
   SWITCH {} -> SWITCH_
   RSYNC {} -> RSYNC_

src/Simplex/Messaging/Agent/TRcvQueues.hs

@@ -11,6 +11,7 @@ module Simplex.Messaging.Agent.TRcvQueues
     deleteQueue,
     getSessQueues,
     getDelSessQueues,
+    getDelAllQueues,
     qKey,
   )
 where
@@ -96,6 +97,11 @@ 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
 
+getDelAllQueues :: TRcvQueues -> STM [RcvQueue]
+getDelAllQueues (TRcvQueues qs cs) = do
+  writeTVar cs M.empty
+  M.elems <$> swapTVar qs M.empty
+
 isSession :: RcvQueue -> (UserId, SMPServer, Maybe ConnId) -> Bool
 isSession rq (uId, srv, connId_) =
   userId rq == uId && server rq == srv && maybe True (connId rq ==) connId_

tests/AgentTests/FunctionalAPITests.hs

@@ -14,6 +14,7 @@
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TypeApplications #-}
 {-# OPTIONS_GHC -Wno-orphans #-}
+{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
 
 module AgentTests.FunctionalAPITests
   ( functionalAPITests,
@@ -2904,7 +2905,7 @@
             _ -> error "timeout"
 
 testTwoUsers :: HasCallStack => IO ()
-testTwoUsers = withAgentClients2 $ \a b -> do
+testTwoUsers = withAgentClientsCfg2 aCfg aCfg $ \a b -> do
   let nc = netCfg initAgentServers
   sessionMode nc `shouldBe` TSMUser
   runRight_ $ do
@@ -2916,7 +2917,7 @@
     b `hasClients` 1
     liftIO $ setNetworkConfig a nc {sessionMode = TSMEntity}
     liftIO $ threadDelay 250000
-    ("", "", DOWN _ _) <- nGet a
+    ("", "", DOWN_ALL) <- nGet a
     ("", "", UP _ _) <- nGet a
     a `hasClients` 2
 
@@ -2925,9 +2926,7 @@
     liftIO $ threadDelay 250000
     liftIO $ setNetworkConfig a nc {sessionMode = TSMUser}
     liftIO $ threadDelay 250000
-    ("", "", DOWN _ _) <- nGet a
-    ("", "", DOWN _ _) <- nGet a
-    ("", "", UP _ _) <- nGet a
+    ("", "", DOWN_ALL) <- nGet a
     ("", "", UP _ _) <- nGet a
     a `hasClients` 1
 
@@ -2940,9 +2939,7 @@
     b `hasClients` 1
     liftIO $ setNetworkConfig a nc {sessionMode = TSMEntity}
     liftIO $ threadDelay 250000
-    ("", "", DOWN _ _) <- nGet a
-    ("", "", DOWN _ _) <- nGet a
-    ("", "", UP _ _) <- nGet a
+    ("", "", DOWN_ALL) <- nGet a
     ("", "", UP _ _) <- nGet a
     a `hasClients` 4
     exchangeGreetingsMsgId 6 a bId1 b aId1
@@ -2952,20 +2949,16 @@
     liftIO $ threadDelay 250000
     liftIO $ setNetworkConfig a nc {sessionMode = TSMUser}
     liftIO $ threadDelay 250000
-    ("", "", DOWN _ _) <- nGet a
-    ("", "", DOWN _ _) <- nGet a
-    ("", "", DOWN _ _) <- nGet a
-    ("", "", DOWN _ _) <- nGet a
-    ("", "", UP _ _) <- nGet a
-    ("", "", UP _ _) <- nGet a
-    ("", "", UP _ _) <- nGet a
+    ("", "", DOWN_ALL) <- nGet a
     ("", "", UP _ _) <- nGet a
     a `hasClients` 2
     exchangeGreetingsMsgId 8 a bId1 b aId1
     exchangeGreetingsMsgId 8 a bId1' b aId1'
     exchangeGreetingsMsgId 6 a bId2 b aId2
     exchangeGreetingsMsgId 6 a bId2' b aId2'
   where
+    aCfg :: AgentConfig
+    aCfg = agentCfg {tbqSize = 16}
     hasClients :: HasCallStack => AgentClient -> Int -> ExceptT AgentErrorType IO ()
     hasClients c n = liftIO $ M.size <$> readTVarIO (smpClients c) `shouldReturn` n
 

tests/CoreTests/TRcvQueuesTests.hs

@@ -76,7 +76,7 @@ batchIdempotentTest = do
   atomically $ RQ.batchAddQueues trq qs
   checkDataInvariant trq `shouldReturn` True
   readTVarIO (RQ.getRcvQueues trq) `shouldReturn` qs'
-  fmap L.nub <$> readTVarIO (RQ.getConnections trq) `shouldReturn`cs' -- connections get duplicated, but that doesn't appear to affect anybody
+  fmap L.nub <$> readTVarIO (RQ.getConnections trq) `shouldReturn` cs' -- connections get duplicated, but that doesn't appear to affect anybody
 
 deleteConnTest :: IO ()
 deleteConnTest = do