batch subscriptions for ntf server (#778)

* return updated ConnectionStats from switchConnectionAsync (#777)

* batch ntf server subscriptions to SMP servers

* refactor

* fix filtering queues in reconnect

* test, remove repeated grouping

* enable all tests

* reduce subscriptions count

---------

Co-authored-by: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com>

src/Simplex/Messaging/Client.hs

@@ -41,6 +41,7 @@ module Simplex.Messaging.Client
     subscribeSMPQueues,
     getSMPMessage,
     subscribeSMPQueueNotifications,
+    subscribeSMPQueuesNtfs,
     secureSMPQueue,
     enableSMPQueueNotifications,
     disableSMPQueueNotifications,
@@ -487,6 +488,16 @@ getSMPMessage c rpKey rId =
 subscribeSMPQueueNotifications :: SMPClient -> NtfPrivateSignKey -> NotifierId -> ExceptT SMPClientError IO ()
 subscribeSMPQueueNotifications = okSMPCommand NSUB
 
+-- | Subscribe to multiple SMP queues notifications batching commands if supported.
+subscribeSMPQueuesNtfs :: SMPClient -> NonEmpty (NtfPrivateSignKey, NotifierId) -> IO (NonEmpty (Either SMPClientError ()))
+subscribeSMPQueuesNtfs c qs = sendProtocolCommands c cs >>= mapM response
+  where
+    cs = L.map (\(npKey, nId) -> (Just npKey, nId, Cmd SNotifier NSUB)) qs
+    response r = pure $ case r of
+      Right OK -> Right ()
+      Right r' -> Left . PCEUnexpectedResponse $ bshow r'
+      Left e -> Left e
+
 -- | Secure the SMP queue by adding a sender public key.
 --
 -- https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#secure-queue-command

src/Simplex/Messaging/Client/Agent.hs

@@ -5,6 +5,7 @@
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TupleSections #-}
 {-# LANGUAGE TypeApplications #-}
 {-# OPTIONS_GHC -fno-warn-orphans #-}
 
@@ -16,27 +17,33 @@ import Control.Logger.Simple
 import Control.Monad.Except
 import Control.Monad.IO.Unlift
 import Control.Monad.Trans.Except
+import Data.Bifunctor (first)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
+import Data.List (find, partition)
+import Data.List.NonEmpty (NonEmpty)
+import qualified Data.List.NonEmpty as L
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
 import Data.Set (Set)
 import Data.Text.Encoding
+import Data.Tuple (swap)
 import Numeric.Natural
 import Simplex.Messaging.Agent.RetryInterval
 import Simplex.Messaging.Client
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
-import Simplex.Messaging.Protocol (BrokerMsg, ProtocolServer (..), QueueId, SMPServer)
+import Simplex.Messaging.Protocol (BrokerMsg, ProtocolServer (..), QueueId, SMPServer, NtfPrivateSignKey, NotifierId, RcvPrivateSignKey, RecipientId)
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport
-import Simplex.Messaging.Util (catchAll_, tryE, unlessM, ($>>=))
+import Simplex.Messaging.Util (catchAll_, tryE, ($>>=))
 import System.Timeout (timeout)
-import UnliftIO (async, forConcurrently_)
+import UnliftIO (async)
 import UnliftIO.Exception (Exception)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
+import Data.Either (isLeft)
 
 type SMPClientVar = TMVar (Either SMPClientError SMPClient)
 
@@ -184,9 +191,9 @@ getSMPServerClient' ca@SMPClientAgent {agentCfg, smpClients, msgQ} srv =
             _ -> TM.insert srv sVar ps
 
     serverDown :: Map SMPSub C.APrivateSignKey -> IO ()
-    serverDown ss = unless (M.null ss) . void . runExceptT $ do
+    serverDown ss = unless (M.null ss) $ do
       notify . CADisconnected srv $ M.keysSet ss
-      reconnectServer
+      void $ runExceptT reconnectServer
 
     reconnectServer :: ExceptT SMPClientError IO ()
     reconnectServer = do
@@ -201,27 +208,47 @@ getSMPServerClient' ca@SMPClientAgent {agentCfg, smpClients, msgQ} srv =
     reconnectClient :: ExceptT SMPClientError IO ()
     reconnectClient = do
       withSMP ca srv $ \smp -> do
-        notify $ CAReconnected srv
-        cs <- atomically $ mapM readTVar =<< TM.lookup srv (pendingSrvSubs ca)
-        forConcurrently_ (maybe [] M.assocs cs) $ \sub@(s, _) ->
-          unlessM (atomically $ hasSub (srvSubs ca) srv s) $
-            subscribe_ smp sub `catchE` handleError s
+        liftIO . notify $ CAReconnected srv
+        cs_ <- atomically $ mapM readTVar =<< TM.lookup srv (pendingSrvSubs ca)
+        forM_ cs_ $ \cs -> do
+          subs' <- filterM (fmap not . atomically . hasSub (srvSubs ca) srv . fst) $ M.assocs cs
+          let (nSubs, rSubs) = partition (isNotifier . fst . fst) subs'
+          nRs <- liftIO $ subscribe_ smp SPNotifier nSubs
+          rRs <- liftIO $ subscribe_ smp SPRecipient rSubs
+          case find isLeft $ nRs <> rRs of
+            Just (Left e) -> throwE e
+            _ -> pure ()
       where
-        subscribe_ :: SMPClient -> (SMPSub, C.APrivateSignKey) -> ExceptT SMPClientError IO ()
-        subscribe_ smp sub@(s, _) = do
-          smpSubscribe smp sub
-          atomically $ addSubscription ca srv sub
-          notify $ CAResubscribed srv s
+        isNotifier = \case
+          SPNotifier -> True
+          SPRecipient -> False
 
-        handleError :: SMPSub -> SMPClientError -> ExceptT SMPClientError IO ()
-        handleError s = \case
-          e@PCEResponseTimeout -> throwE e
-          e@PCENetworkError -> throwE e
-          e -> do
-            notify $ CASubError srv s e
-            atomically $ removePendingSubscription ca srv s
+        subscribe_ :: SMPClient -> SMPSubParty -> [(SMPSub, C.APrivateSignKey)] -> IO [Either SMPClientError ()]
+        subscribe_ smp party subs =
+          case L.nonEmpty subs of
+            Just subs' -> do
+              let subs'' = L.map (first snd) subs'
+              rs <- L.zip subs'' <$> smpSubscribeQueues party ca smp srv subs''
+              rs' <- forM rs $ \(sub, r) -> do
+                let sub' = first (party,) sub
+                    s = fst sub'
+                case snd r of
+                  Right () -> do
+                    atomically $ addSubscription ca srv sub'
+                    notify $ CAResubscribed srv s
+                    pure $ Right ()
+                  Left e -> do
+                    case e of
+                      PCEResponseTimeout -> pure $ Left e
+                      PCENetworkError -> pure $ Left e
+                      _ -> do
+                        notify $ CASubError srv s e
+                        atomically $ removePendingSubscription ca srv s
+                        pure $ Right ()
+              pure $ L.toList rs'
+            Nothing -> pure []
 
-    notify :: SMPClientAgentEvent -> ExceptT SMPClientError IO ()
+    notify :: SMPClientAgentEvent -> IO ()
     notify evt = atomically $ writeTBQueue (agentQ ca) evt
 
 closeSMPClientAgent :: MonadUnliftIO m => SMPClientAgent -> m ()
@@ -263,6 +290,35 @@ subscribeQueue ca srv sub = do
         removePendingSubscription ca srv $ fst sub
       throwE e
 
+subscribeQueuesSMP :: SMPClientAgent -> SMPServer -> NonEmpty (RecipientId, RcvPrivateSignKey) -> IO (NonEmpty (RecipientId, Either SMPClientError ()))
+subscribeQueuesSMP = subscribeQueues_ SPRecipient
+
+subscribeQueuesNtfs :: SMPClientAgent -> SMPServer -> NonEmpty (NotifierId, NtfPrivateSignKey) -> IO (NonEmpty (NotifierId, Either SMPClientError ()))
+subscribeQueuesNtfs = subscribeQueues_ SPNotifier
+
+subscribeQueues_ :: SMPSubParty -> SMPClientAgent -> SMPServer -> NonEmpty (QueueId, C.APrivateSignKey) -> IO (NonEmpty (QueueId, Either SMPClientError ()))
+subscribeQueues_ party ca srv subs = do
+  atomically $ forM_ subs $ addPendingSubscription ca srv . first (party,)
+  runExceptT (getSMPServerClient' ca srv) >>= \case
+    Left e -> pure $ L.map ((,Left e) . fst) subs
+    Right smp -> smpSubscribeQueues party ca smp srv subs
+
+smpSubscribeQueues :: SMPSubParty -> SMPClientAgent -> SMPClient -> SMPServer -> NonEmpty (QueueId, C.APrivateSignKey) -> IO (NonEmpty (QueueId, Either SMPClientError ()))
+smpSubscribeQueues party ca smp srv subs = do
+  rs <- L.zip subs <$> subscribe smp (L.map swap subs)
+  atomically $ forM rs $ \(sub, r) -> (fst sub,) <$> case r of
+    Right () -> do
+      addSubscription ca srv $ first (party,) sub
+      pure $ Right ()
+    Left e -> do
+      when (e /= PCENetworkError && e /= PCEResponseTimeout) $
+        removePendingSubscription ca srv $ (party,) $ fst sub
+      pure $ Left e
+  where
+    subscribe = case party of
+      SPRecipient -> subscribeSMPQueues
+      SPNotifier -> subscribeSMPQueuesNtfs
+
 showServer :: SMPServer -> ByteString
 showServer ProtocolServer {host, port} =
   strEncode host <> B.pack (if null port then "" else ':' : port)

src/Simplex/Messaging/Notifications/Server.hs

@@ -18,10 +18,14 @@ import Control.Monad.Except
 import Control.Monad.Reader
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
+import Data.Function (on)
 import Data.Functor (($>))
 import Data.Int (Int64)
 import Data.List (intercalate)
+import Data.List.NonEmpty (NonEmpty(..))
+import qualified Data.List.NonEmpty as L
 import Data.Map.Strict (Map)
+import qualified Data.Map.Strict as M
 import qualified Data.Text as T
 import Data.Text.Encoding (decodeLatin1)
 import Data.Time.Clock (UTCTime (..), diffTimeToPicoseconds, getCurrentTime)
@@ -55,6 +59,7 @@ import UnliftIO.Concurrent (forkIO, killThread, mkWeakThreadId, threadDelay)
 import UnliftIO.Directory (doesFileExist, renameFile)
 import UnliftIO.Exception
 import UnliftIO.STM
+import Data.Bifunctor (second)
 
 runNtfServer :: NtfServerConfig -> IO ()
 runNtfServer cfg = do
@@ -141,11 +146,8 @@ ntfServer cfg@NtfServerConfig {transports, logTLSErrors} started = do
 
 resubscribe :: NtfSubscriber -> Map NtfSubscriptionId NtfSubData -> M ()
 resubscribe NtfSubscriber {newSubQ} subs = do
-  d <- asks $ resubscribeDelay . config
-  forM_ subs $ \sub@NtfSubData {} ->
-    whenM (ntfShouldSubscribe <$> readTVarIO (subStatus sub)) $ do
-      atomically $ writeTBQueue newSubQ $ NtfSub sub
-      threadDelay d
+  subs' <- atomically $ filterM (fmap ntfShouldSubscribe . readTVar . subStatus) $ M.elems subs
+  mapM_ (atomically . writeTBQueue newSubQ . L.map NtfSub) $ L.nonEmpty subs'
   liftIO $ logInfo "SMP connections resubscribed"
 
 ntfSubscriber :: NtfSubscriber -> M ()
@@ -153,12 +155,18 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
   raceAny_ [subscribe, receiveSMP, receiveAgent]
   where
     subscribe :: M ()
-    subscribe =
-      forever $
-        atomically (readTBQueue newSubQ) >>= \case
-          sub@(NtfSub NtfSubData {smpQueue = SMPQueueNtf {smpServer}}) -> do
-            SMPSubscriber {newSubQ = subscriberSubQ} <- getSMPSubscriber smpServer
-            atomically $ writeTQueue subscriberSubQ sub
+    subscribe = do
+      d <- asks $ resubscribeDelay . config
+      forever $ do
+        subs <- atomically (readTBQueue newSubQ)
+        let ss = L.groupBy ((==) `on` server) subs
+        forM_ ss $ \serverSubs -> do
+          SMPSubscriber {newSubQ = subscriberSubQ} <- getSMPSubscriber $ server $ L.head serverSubs
+          atomically $ writeTQueue subscriberSubQ serverSubs
+          when (length serverSubs > 10) $ threadDelay d
+
+    server :: NtfEntityRec 'Subscription -> SMPServer
+    server (NtfSub sub) = ntfSubServer sub
 
     getSMPSubscriber :: SMPServer -> M SMPSubscriber
     getSMPSubscriber smpServer =
@@ -173,21 +181,31 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
 
     runSMPSubscriber :: SMPSubscriber -> M ()
     runSMPSubscriber SMPSubscriber {newSubQ = subscriberSubQ} =
-      forever $
-        atomically (peekTQueue subscriberSubQ)
-          >>= \(NtfSub NtfSubData {smpQueue, notifierKey}) -> do
-            updateSubStatus smpQueue NSPending
-            let SMPQueueNtf {smpServer, notifierId} = smpQueue
-            liftIO (runExceptT $ subscribeQueue ca smpServer ((SPNotifier, notifierId), notifierKey)) >>= \case
-              Right _ -> do
-                updateSubStatus smpQueue NSActive
-                void . atomically $ readTQueue subscriberSubQ
-              Left err -> do
-                handleSubError smpQueue err
-                case err of
-                  PCEResponseTimeout -> pure ()
-                  PCENetworkError -> pure ()
-                  _ -> void . atomically $ readTQueue subscriberSubQ
+      forever $ do
+        subs <- atomically (peekTQueue subscriberSubQ)
+        let subs' = L.map (\(NtfSub sub) -> sub) subs
+        mapM_ (\NtfSubData {smpQueue} -> updateSubStatus smpQueue NSPending) subs'
+        rs <- liftIO $ subscribeQueues (server $ L.head subs) subs'
+        subs_ <- L.nonEmpty <$> foldM process [] rs
+        atomically $ do
+          void $ readTQueue subscriberSubQ
+          mapM_ (writeTQueue subscriberSubQ . L.map NtfSub) subs_
+      where
+        process subs (sub@NtfSubData {smpQueue}, r) = case r of
+          Right _ -> updateSubStatus smpQueue NSActive $> subs
+          Left err -> do
+            handleSubError smpQueue err
+            pure $ case err of
+              PCEResponseTimeout -> sub : subs
+              PCENetworkError -> sub : subs
+              _ -> subs
+
+    -- | Subscribe to queues. The list of results can have a different order.
+    subscribeQueues :: SMPServer -> NonEmpty NtfSubData -> IO (NonEmpty (NtfSubData, Either SMPClientError ()))
+    subscribeQueues srv subs =
+      L.map (second snd) . L.zip subs <$> subscribeQueuesNtfs ca srv (L.map sub subs)
+      where
+        sub NtfSubData {smpQueue = SMPQueueNtf {notifierId}, notifierKey} = (notifierId, notifierKey)
 
     receiveSMP :: M ()
     receiveSMP = forever $ do
@@ -504,7 +522,7 @@ client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPu
         sub <- atomically $ mkNtfSubData subId newSub
         resp <-
           atomically (addNtfSubscription st subId sub) >>= \case
-            Just _ -> atomically (writeTBQueue newSubQ $ NtfSub sub) $> NRSubId subId
+            Just _ -> atomically (writeTBQueue newSubQ [NtfSub sub]) $> NRSubId subId
             _ -> pure $ NRErr AUTH
         withNtfLog (`logCreateSubscription` sub)
         incNtfStat subCreated

src/Simplex/Messaging/Notifications/Server/Env.hs

@@ -12,6 +12,7 @@ import Control.Monad.IO.Unlift
 import Crypto.Random
 import Data.ByteString.Char8 (ByteString)
 import Data.Int (Int64)
+import Data.List.NonEmpty (NonEmpty)
 import Data.Time.Clock (getCurrentTime)
 import Data.Time.Clock.System (SystemTime)
 import Data.Word (Word16)
@@ -93,7 +94,7 @@ newNtfServerEnv config@NtfServerConfig {subQSize, pushQSize, smpAgentCfg, apnsCo
 
 data NtfSubscriber = NtfSubscriber
   { smpSubscribers :: TMap SMPServer SMPSubscriber,
-    newSubQ :: TBQueue (NtfEntityRec 'Subscription),
+    newSubQ :: TBQueue (NonEmpty (NtfEntityRec 'Subscription)),
     smpAgent :: SMPClientAgent
   }
 
@@ -105,7 +106,7 @@ newNtfSubscriber qSize smpAgentCfg = do
   pure NtfSubscriber {smpSubscribers, newSubQ, smpAgent}
 
 data SMPSubscriber = SMPSubscriber
-  { newSubQ :: TQueue (NtfEntityRec 'Subscription),
+  { newSubQ :: TQueue (NonEmpty (NtfEntityRec 'Subscription)),
     subThreadId :: TVar (Maybe (Weak ThreadId))
   }
 

src/Simplex/Messaging/Notifications/Server/Store.hs

@@ -20,7 +20,7 @@ import qualified Data.Set as S
 import Data.Word (Word16)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Notifications.Protocol
-import Simplex.Messaging.Protocol (NtfPrivateSignKey)
+import Simplex.Messaging.Protocol (NtfPrivateSignKey, SMPServer)
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Util (whenM, ($>>=))
@@ -68,6 +68,9 @@ data NtfSubData = NtfSubData
     subStatus :: TVar NtfSubStatus
   }
 
+ntfSubServer :: NtfSubData -> SMPServer
+ntfSubServer NtfSubData {smpQueue = SMPQueueNtf {smpServer}} = smpServer
+
 data NtfEntityRec (e :: NtfEntity) where
   NtfTkn :: NtfTknData -> NtfEntityRec 'Token
   NtfSub :: NtfSubData -> NtfEntityRec 'Subscription

src/Simplex/Messaging/Util.hs

@@ -16,7 +16,7 @@ import Data.Int (Int64)
 import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Text.Encoding (decodeUtf8With)
-import Data.Time (NominalDiffTime, nominalDiffTimeToSeconds)
+import Data.Time (NominalDiffTime)
 import UnliftIO.Async
 
 raceAny_ :: MonadUnliftIO m => [m a] -> m ()

tests/AgentTests/NotificationTests.hs

@@ -19,8 +19,8 @@ import qualified Data.ByteString.Base64.URL as U
 import Data.ByteString.Char8 (ByteString)
 import Data.Text.Encoding (encodeUtf8)
 import NtfClient
-import SMPAgentClient (agentCfg, initAgentServers, testDB, testDB2)
-import SMPClient (testPort, withSmpServer, withSmpServerStoreLogOn, xit')
+import SMPAgentClient (agentCfg, initAgentServers, initAgentServers2, testDB, testDB2)
+import SMPClient (cfg, testPort, testPort2, testStoreLogFile2, withSmpServer, withSmpServerConfigOn, withSmpServerStoreLogOn, xit')
 import Simplex.Messaging.Agent
 import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..), InitialAgentServers)
 import Simplex.Messaging.Agent.Protocol
@@ -31,6 +31,7 @@ import Simplex.Messaging.Notifications.Server.Push.APNS
 import Simplex.Messaging.Notifications.Types (NtfToken (..))
 import Simplex.Messaging.Protocol (ErrorType (AUTH), MsgFlags (MsgFlags), SMPMsgMeta (..))
 import qualified Simplex.Messaging.Protocol as SMP
+import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
 import Simplex.Messaging.Transport (ATransport)
 import Simplex.Messaging.Util (tryE)
 import System.Directory (doesFileExist, removeFile)
@@ -85,6 +86,10 @@ notificationTests t =
       it "should resume subscriptions after SMP server is restarted" $ \_ ->
         withAPNSMockServer $ \apns ->
           withNtfServer t $ testNotificationsSMPRestart t apns
+    describe "Notifications after SMP server restart" $
+      it "should resume batched subscriptions after SMP server is restarted" $ \_ ->
+        withAPNSMockServer $ \apns ->
+          withNtfServer t $ testNotificationsSMPRestartBatch 100 t apns
     describe "should switch notifications to the new queue" $
       testServerMatrix2 t $ \servers ->
         withAPNSMockServer $ \apns ->
@@ -480,6 +485,52 @@ testNotificationsSMPRestart t APNSMockServer {apnsQ} = do
     get alice =##> \case ("", c, Msg "hello again") -> c == bobId; _ -> False
     liftIO $ killThread threadId
 
+testNotificationsSMPRestartBatch :: Int -> ATransport -> APNSMockServer -> IO ()
+testNotificationsSMPRestartBatch n t APNSMockServer {apnsQ} = do
+  a <- getSMPAgentClient' agentCfg initAgentServers2 testDB
+  b <- getSMPAgentClient' agentCfg initAgentServers2 testDB2
+  conns <- runServers $ do
+    conns <- forM [1 .. n :: Int] . const $ makeConnection a b
+    _ <- registerTestToken a "abcd" NMInstant apnsQ
+    liftIO $ threadDelay 1500000
+    forM_ conns $ \(aliceId, bobId) -> do
+      msgId <- sendMessage b aliceId (SMP.MsgFlags True) "hello"
+      get b ##> ("", aliceId, SENT msgId)
+      void $ messageNotification apnsQ
+      get a =##> \case ("", c, Msg "hello") -> c == bobId; _ -> False
+      ackMessage a bobId msgId
+    pure conns
+
+  runRight_ @AgentErrorType $ do
+    ("", "", DOWN _ bcs1) <- nGet a
+    ("", "", DOWN _ bcs2) <- nGet a
+    liftIO $ length (bcs1 <> bcs2) `shouldBe` length conns
+    ("", "", DOWN _ acs1) <- nGet b
+    ("", "", DOWN _ acs2) <- nGet b
+    liftIO $ length (acs1 <> acs2) `shouldBe` length conns
+
+  runServers $ do
+    ("", "", UP _ bcs1) <- nGet a
+    ("", "", UP _ bcs2) <- nGet a
+    liftIO $ length (bcs1 <> bcs2) `shouldBe` length conns
+    ("", "", UP _ acs1) <- nGet b
+    ("", "", UP _ acs2) <- nGet b
+    liftIO $ length (acs1 <> acs2) `shouldBe` length conns
+    liftIO $ threadDelay 1500000
+    forM_ conns $ \(aliceId, bobId) -> do
+      msgId <- sendMessage b aliceId (SMP.MsgFlags True) "hello again"
+      get b ##> ("", aliceId, SENT msgId)
+      _ <- messageNotificationData a apnsQ
+      get a =##> \case ("", c, Msg "hello again") -> c == bobId; _ -> False
+  where
+    runServers :: ExceptT AgentErrorType IO a -> IO a
+    runServers a = do
+      withSmpServerStoreLogOn t testPort $ \t1 -> do
+        res <- withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2} testPort2 $ \t2 ->
+          runRight a `finally` killThread t2
+        killThread t1
+        pure res
+
 testSwitchNotifications :: InitialAgentServers -> APNSMockServer -> IO ()
 testSwitchNotifications servers APNSMockServer {apnsQ} = do
   a <- getSMPAgentClient' agentCfg servers testDB
@@ -496,7 +547,7 @@ testSwitchNotifications servers APNSMockServer {apnsQ} = do
           get a =##> \case ("", c, Msg msg') -> c == bId && msg == msg'; _ -> False
           ackMessage a bId msgId
     testMessage "hello"
-    switchConnectionAsync a "" bId
+    _ <- switchConnectionAsync a "" bId
     switchComplete a bId b aId
     liftIO $ threadDelay 500000
     testMessage "hello again"