5.7.4.1

* agent: remove external timeout to resubscribe

* liftIO

* fix tests

CHANGELOG.md

@@ -1,3 +1,10 @@
+# 5.7.4
+
+SMP agent:
+- remove re-subscription timeouts (as they are tracked per operation, and could cause failed subscriptions).
+- reconnect XFTP clients when network settings changes.
+- fix lock contention resulting in stuck subscriptions on network change.
+
 # 5.7.3
 
 SMP/NTF protocol:

package.yaml

@@ -1,5 +1,5 @@
 name: simplexmq
-version: 5.7.4.0
+version: 5.7.4.1
 synopsis: SimpleXMQ message broker
 description: |
   This package includes <./docs/Simplex-Messaging-Server.html server>,

simplexmq.cabal

@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack
 
 name:           simplexmq
-version:        5.7.4.0
+version:        5.7.4.1
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and

src/Simplex/Messaging/Agent.hs

@@ -429,24 +429,16 @@ getNetworkConfig = fmap snd . readTVarIO . useNetworkConfig
 {-# INLINE getNetworkConfig #-}
 
 setUserNetworkInfo :: AgentClient -> UserNetworkInfo -> IO ()
-setUserNetworkInfo c@AgentClient {userNetworkInfo, userNetworkDelay} netInfo = withAgentEnv' c $ do
-  ni <- asks $ userNetworkInterval . config
-  let d = initialInterval ni
-  off <- atomically $ do
-    wasOnline <- isOnline <$> swapTVar userNetworkInfo netInfo
-    let off = wasOnline && not (isOnline netInfo)
-    when off $ writeTVar userNetworkDelay d
-    pure off
-  liftIO . when off . void . forkIO $
-    growOfflineDelay 0 d ni
+setUserNetworkInfo c@AgentClient {userNetworkInfo, userNetworkUpdated} 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'
   where
-    growOfflineDelay elapsed d ni = do
-      online <- waitOnlineOrDelay c d
-      unless online $ do
-        let elapsed' = elapsed + d
-            d' = nextRetryDelay elapsed' d ni
-        atomically $ writeTVar userNetworkDelay d'
-        growOfflineDelay elapsed' d' ni
+    notRecentlyChanged ts' i =
+      maybe True (\ts -> diffUTCTime ts' ts > i) <$> readTVar userNetworkUpdated
 
 reconnectAllServers :: AgentClient -> IO ()
 reconnectAllServers c = do

src/Simplex/Messaging/Agent/Client.hs

@@ -104,7 +104,6 @@ module Simplex.Messaging.Agent.Client
     UserNetworkInfo (..),
     UserNetworkType (..),
     waitForUserNetwork,
-    waitOnlineOrDelay,
     isNetworkOnline,
     isOnline,
     throwWhenInactive,
@@ -155,7 +154,6 @@ import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Either (lefts, partitionEithers)
 import Data.Functor (($>))
-import Data.Int (Int64)
 import Data.List (deleteFirstsBy, foldl', partition, (\\))
 import Data.List.NonEmpty (NonEmpty (..), (<|))
 import qualified Data.List.NonEmpty as L
@@ -270,7 +268,7 @@ data AgentClient = AgentClient
     xftpClients :: TMap XFTPTransportSession XFTPClientVar,
     useNetworkConfig :: TVar (NetworkConfig, NetworkConfig), -- (slow, fast) networks
     userNetworkInfo :: TVar UserNetworkInfo,
-    userNetworkDelay :: TVar Int64,
+    userNetworkUpdated :: TVar (Maybe UTCTime),
     subscrConns :: TVar (Set ConnId),
     activeSubs :: TRcvQueues,
     pendingSubs :: TRcvQueues,
@@ -434,7 +432,7 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv =
   xftpClients <- TM.empty
   useNetworkConfig <- newTVar (slowNetworkConfig netCfg, netCfg)
   userNetworkInfo <- newTVar $ UserNetworkInfo UNOther True
-  userNetworkDelay <- newTVar $ initialInterval $ userNetworkInterval cfg
+  userNetworkUpdated <- newTVar Nothing
   subscrConns <- newTVar S.empty
   activeSubs <- RQ.empty
   pendingSubs <- RQ.empty
@@ -470,7 +468,7 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv =
         xftpClients,
         useNetworkConfig,
         userNetworkInfo,
-        userNetworkDelay,
+        userNetworkUpdated,
         subscrConns,
         activeSubs,
         pendingSubs,
@@ -610,12 +608,11 @@ resubscribeSMPSession c@AgentClient {smpSubWorkers, workerSeq} tSess =
       atomically $ putTMVar (sessionVar v) a
     runSubWorker = do
       ri <- asks $ reconnectInterval . config
-      timeoutCounts <- newTVarIO 0
       withRetryInterval ri $ \_ loop -> do
         pending <- atomically getPending
         forM_ (L.nonEmpty pending) $ \qs -> do
-          lift $ waitForUserNetwork c
-          void . tryAgentError' $ reconnectSMPClient timeoutCounts c tSess qs
+          liftIO $ waitForUserNetwork c
+          reconnectSMPClient c tSess qs
           loop
     getPending = RQ.getSessQueues tSess $ pendingSubs c
     cleanup :: SessionVar (Async ()) -> STM ()
@@ -625,38 +622,23 @@ resubscribeSMPSession c@AgentClient {smpSubWorkers, workerSeq} tSess =
       whenM (isEmptyTMVar $ sessionVar v) retry
       removeSessVar v tSess smpSubWorkers
 
-reconnectSMPClient :: TVar Int -> AgentClient -> SMPTransportSession -> NonEmpty RcvQueue -> AM ()
-reconnectSMPClient tc c tSess@(_, srv, _) qs = do
-  NetworkConfig {tcpTimeout} <- atomically $ getNetworkConfig c
-  -- this allows 3x of timeout per batch of subscription (90 queues per batch empirically)
-  let t = (length qs `div` 90 + 1) * tcpTimeout * 3
-  ExceptT (sequence <$> (t `timeout` runExceptT resubscribe)) >>= \case
-    Just _ -> resetTimeouts
-    -- reset and do not report consecutive timeouts while offline
-    Nothing -> ifM (atomically $ isNetworkOnline c) notifyTimeout resetTimeouts
+reconnectSMPClient :: AgentClient -> SMPTransportSession -> NonEmpty RcvQueue -> AM' ()
+reconnectSMPClient c tSess@(_, srv, _) qs = handleNotify $ do
+  cs <- readTVarIO $ RQ.getConnections $ activeSubs c
+  rs <- subscribeQueues c $ L.toList qs
+  let (errs, okConns) = partitionEithers $ map (\(RcvQueue {connId}, r) -> bimap (connId,) (const connId) r) rs
+      conns = filter (`M.notMember` cs) okConns
+  unless (null conns) $ notifySub "" $ UP srv conns
+  let (tempErrs, finalErrs) = partition (temporaryAgentError . snd) errs
+  mapM_ (\(connId, e) -> notifySub connId $ ERR e) finalErrs
+  forM_ (listToMaybe tempErrs) $ \(connId, e) -> do
+    when (null okConns && M.null cs && null finalErrs) . liftIO $
+      closeClient c smpClients tSess
+    notifySub connId $ ERR e
   where
-    resetTimeouts = atomically $ writeTVar tc 0
-    notifyTimeout = do
-      tc' <- atomically $ stateTVar tc $ \i -> (i + 1, i + 1)
-      maxTC <- asks $ maxSubscriptionTimeouts . config
-      when (tc' >= maxTC) $ do
-        let msg = show tc' <> " consecutive subscription timeouts: " <> show (length qs) <> " queues, transport session: " <> show tSess
-        atomically $ writeTBQueue (subQ c) ("", "", APC SAEConn $ ERR $ INTERNAL msg)
-    resubscribe :: AM ()
-    resubscribe = do
-      cs <- readTVarIO $ RQ.getConnections $ activeSubs c
-      rs <- lift . subscribeQueues c $ L.toList qs
-      let (errs, okConns) = partitionEithers $ map (\(RcvQueue {connId}, r) -> bimap (connId,) (const connId) r) rs
-      liftIO $ do
-        let conns = filter (`M.notMember` cs) okConns
-        unless (null conns) $ notifySub "" $ UP srv conns
-      let (tempErrs, finalErrs) = partition (temporaryAgentError . snd) errs
-      liftIO $ mapM_ (\(connId, e) -> notifySub connId $ ERR e) finalErrs
-      forM_ (listToMaybe tempErrs) $ \(_, err) -> do
-        when (null okConns && M.null cs && null finalErrs) . liftIO $
-          closeClient c smpClients tSess
-        throwError err
-    notifySub :: forall e. AEntityI e => ConnId -> ACommand 'Agent e -> IO ()
+    handleNotify :: AM' () -> AM' ()
+    handleNotify = E.handleAny $ notifySub "" . ERR . INTERNAL . show
+    notifySub :: forall e. AEntityI e => ConnId -> ACommand 'Agent e -> AM' ()
     notifySub connId cmd = atomically $ writeTBQueue (subQ c) ("", connId, APC (sAEntity @e) cmd)
 
 getNtfServerClient :: AgentClient -> NtfTransportSession -> AM NtfClient
@@ -759,23 +741,9 @@ getNetworkConfig c = do
 
 waitForUserNetwork :: AgentClient -> IO ()
 waitForUserNetwork c =
-  unlessM (atomically $ isNetworkOnline c) $
-    readTVarIO (userNetworkDelay c) >>= void . waitOnlineOrDelay c
-
-waitOnlineOrDelay :: AgentClient -> Int64 -> IO Bool
-waitOnlineOrDelay c t = do
-  let maxWait = min t $ fromIntegral (maxBound :: Int)
-      t' = t - maxWait
-  delay <- registerDelay $ fromIntegral maxWait
-  online <-
-    atomically $ do
-      expired <- readTVar delay
-      online <- isNetworkOnline c
-      unless (expired || online) retry
-      pure online
-  if online || t' <= 0
-    then pure online
-    else waitOnlineOrDelay c t'
+  unlessM (atomically $ isNetworkOnline c) $ do
+    delay <- registerDelay $ userNetworkInterval $ config $ agentEnv c
+    atomically $ unlessM (isNetworkOnline c) $ unlessM (readTVar delay) retry
 
 closeAgentClient :: AgentClient -> IO ()
 closeAgentClient c = do

src/Simplex/Messaging/Agent/Env/SQLite.hs

@@ -92,7 +92,8 @@ data AgentConfig = AgentConfig
     xftpCfg :: XFTPClientConfig,
     reconnectInterval :: RetryInterval,
     messageRetryInterval :: RetryInterval2,
-    userNetworkInterval :: RetryInterval,
+    userNetworkInterval :: Int,
+    userOfflineDelay :: NominalDiffTime,
     messageTimeout :: NominalDiffTime,
     connDeleteDeliveryTimeout :: NominalDiffTime,
     helloTimeout :: NominalDiffTime,
@@ -101,7 +102,6 @@ data AgentConfig = AgentConfig
     cleanupInterval :: Int64,
     cleanupStepInterval :: Int,
     maxWorkerRestartsPerMin :: Int,
-    maxSubscriptionTimeouts :: Int,
     storedMsgDataTTL :: NominalDiffTime,
     rcvFilesTTL :: NominalDiffTime,
     sndFilesTTL :: NominalDiffTime,
@@ -147,14 +147,6 @@ defaultMessageRetryInterval =
           }
     }
 
-defaultUserNetworkInterval :: RetryInterval
-defaultUserNetworkInterval =
-  RetryInterval
-    { initialInterval = 1200_000000, -- 20 minutes
-      increaseAfter = 0,
-      maxInterval = 7200_000000 -- 2 hours
-    }
-
 defaultAgentConfig :: AgentConfig
 defaultAgentConfig =
   AgentConfig
@@ -170,7 +162,8 @@ defaultAgentConfig =
       xftpCfg = defaultXFTPClientConfig,
       reconnectInterval = defaultReconnectInterval,
       messageRetryInterval = defaultMessageRetryInterval,
-      userNetworkInterval = defaultUserNetworkInterval,
+      userNetworkInterval = 1800_000000, -- 30 minutes, should be less than Int32 max value
+      userOfflineDelay = 2, -- if network offline event happens in less than 2 seconds after it was set online, it is ignored
       messageTimeout = 2 * nominalDay,
       connDeleteDeliveryTimeout = 2 * nominalDay,
       helloTimeout = 2 * nominalDay,
@@ -179,9 +172,6 @@ defaultAgentConfig =
       cleanupInterval = 30 * 60 * 1000000, -- 30 minutes
       cleanupStepInterval = 200000, -- 200ms
       maxWorkerRestartsPerMin = 5,
-      -- 3 consecutive subscription timeouts will result in alert to the user
-      -- this is a fallback, as the timeout set to 3x of expected timeout, to avoid potential locking.
-      maxSubscriptionTimeouts = 5,
       storedMsgDataTTL = 21 * nominalDay,
       rcvFilesTTL = 2 * nominalDay,
       sndFilesTTL = nominalDay,

tests/AgentTests.hs

@@ -95,23 +95,24 @@ type AEntityTransmission p e = (ACorrId, ConnId, ACommand p e)
 type AEntityTransmissionOrError p e = (ACorrId, ConnId, Either AgentErrorType (ACommand p e))
 
 tGetAgent :: Transport c => c -> IO (AEntityTransmissionOrError 'Agent 'AEConn)
-tGetAgent = tGetAgent'
+tGetAgent = tGetAgent' True
 
-tGetAgent' :: forall c e. (Transport c, AEntityI e) => c -> IO (AEntityTransmissionOrError 'Agent e)
-tGetAgent' h = do
-  (corrId, connId, cmdOrErr) <- pGetAgent h
+tGetAgent' :: forall c e. (Transport c, AEntityI e) => Bool -> c -> IO (AEntityTransmissionOrError 'Agent e)
+tGetAgent' skipErr h = do
+  (corrId, connId, cmdOrErr) <- pGetAgent skipErr h
   case cmdOrErr of
     Right (APC e cmd) -> case testEquality e (sAEntity @e) of
       Just Refl -> pure (corrId, connId, Right cmd)
       _ -> error $ "unexpected command " <> show cmd
     Left err -> pure (corrId, connId, Left err)
 
-pGetAgent :: forall c. Transport c => c -> IO (ATransmissionOrError 'Agent)
-pGetAgent h = do
+pGetAgent :: forall c. Transport c => Bool -> c -> IO (ATransmissionOrError 'Agent)
+pGetAgent skipErr h = do
   (corrId, connId, cmdOrErr) <- tGet SAgent h
   case cmdOrErr of
-    Right (APC _ CONNECT {}) -> pGetAgent h
-    Right (APC _ DISCONNECT {}) -> pGetAgent h
+    Right (APC _ CONNECT {}) -> pGetAgent skipErr h
+    Right (APC _ DISCONNECT {}) -> pGetAgent skipErr h
+    Right (APC _ (ERR (BROKER _ NETWORK))) | skipErr -> pGetAgent skipErr h
     cmd -> pure (corrId, connId, cmd)
 
 -- | receive message to handle `h`
@@ -119,15 +120,18 @@ pGetAgent h = do
 (<#:) = tGetAgent
 
 (<#:?) :: Transport c => c -> IO (ATransmissionOrError 'Agent)
-(<#:?) = pGetAgent
+(<#:?) = pGetAgent True
 
 (<#:.) :: Transport c => c -> IO (AEntityTransmissionOrError 'Agent 'AENone)
-(<#:.) = tGetAgent'
+(<#:.) = tGetAgent' True
 
 -- | send transmission `t` to handle `h` and get response
 (#:) :: Transport c => c -> (ByteString, ByteString, ByteString) -> IO (AEntityTransmissionOrError 'Agent 'AEConn)
 h #: t = tPutRaw h t >> (<#:) h
 
+(#:!) :: Transport c => c -> (ByteString, ByteString, ByteString) -> IO (AEntityTransmissionOrError 'Agent 'AEConn)
+h #:! t = tPutRaw h t >> tGetAgent' False h
+
 -- | action and expected response
 -- `h #:t #> r` is the test that sends `t` to `h` and validates that the response is `r`
 (#>) :: IO (AEntityTransmissionOrError 'Agent 'AEConn) -> AEntityTransmission 'Agent 'AEConn -> Expectation
@@ -426,8 +430,8 @@ testServerConnectionAfterError t _ = do
 
   withAgent1 $ \bob -> do
     withAgent2 $ \alice -> do
-      bob #: ("1", "alice", "SUB") =#> \("1", "alice", ERR (BROKER _ e)) -> e == NETWORK || e == TIMEOUT
-      alice #: ("1", "bob", "SUB") =#> \("1", "bob", ERR (BROKER _ e)) -> e == NETWORK || e == TIMEOUT
+      bob #:! ("1", "alice", "SUB") =#> \("1", "alice", ERR (BROKER _ e)) -> e == NETWORK || e == TIMEOUT
+      alice #:! ("1", "bob", "SUB") =#> \("1", "bob", ERR (BROKER _ e)) -> e == NETWORK || e == TIMEOUT
       withServer $ do
         alice <#=? \case ("", "bob", APC _ (SENT 4)) -> True; ("", "", APC _ (UP s ["bob"])) -> s == server; _ -> False
         alice <#=? \case ("", "bob", APC _ (SENT 4)) -> True; ("", "", APC _ (UP s ["bob"])) -> s == server; _ -> False

tests/AgentTests/FunctionalAPITests.hs

@@ -77,7 +77,6 @@ import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestSte
 import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..), InitialAgentServers (..), createAgentStore)
 import Simplex.Messaging.Agent.Protocol hiding (CON, CONF, INFO, REQ)
 import qualified Simplex.Messaging.Agent.Protocol as A
-import Simplex.Messaging.Agent.RetryInterval (RetryInterval (..))
 import Simplex.Messaging.Agent.Store.SQLite (MigrationConfirmation (..), SQLiteStore (dbNew))
 import Simplex.Messaging.Agent.Store.SQLite.Common (withTransaction')
 import Simplex.Messaging.Client (NetworkConfig (..), ProtocolClientConfig (..), TransportSessionMode (TSMEntity, TSMUser), defaultClientConfig)
@@ -145,6 +144,7 @@ pGet c = do
   case cmd of
     CONNECT {} -> pGet c
     DISCONNECT {} -> pGet c
+    ERR (BROKER _ NETWORK) -> pGet c
     _ -> pure t
 
 pattern CONF :: ConfirmationId -> [SMPServer] -> ConnInfo -> ACommand 'Agent e
@@ -435,7 +435,7 @@ functionalAPITests t = do
     it "send delivery receipts concurrently with messages" $ testDeliveryReceiptsConcurrent t
   describe "user network info" $ do
     it "should wait for user network" testWaitForUserNetwork
-    it "should not reset offline interval while offline" testDoNotResetOfflineInterval
+    it "should not reset online to offline if happens too quickly" testDoNotResetOnlineToOffline
     it "should resume multiple threads" testResumeMultipleThreads
 
 testBasicAuth :: ATransport -> Bool -> (Maybe BasicAuth, VersionSMP) -> (Maybe BasicAuth, VersionSMP) -> (Maybe BasicAuth, VersionSMP) -> IO Int
@@ -2698,11 +2698,8 @@ testWaitForUserNetwork = do
   a <- getSMPAgentClient' 1 aCfg initAgentServers testDB
   noNetworkDelay a
   setUserNetworkInfo a $ UserNetworkInfo UNNone False
-  threadDelay 5000
   networkDelay a 100000
-  networkDelay a 150000
-  networkDelay a 200000
-  networkDelay a 200000
+  networkDelay a 100000
   setUserNetworkInfo a $ UserNetworkInfo UNCellular True
   noNetworkDelay a
   setUserNetworkInfo a $ UserNetworkInfo UNCellular False
@@ -2712,26 +2709,29 @@ testWaitForUserNetwork = do
     (networkDelay a 50000)
   noNetworkDelay a
   where
-    aCfg = agentCfg {userNetworkInterval = RetryInterval {initialInterval = 100000, increaseAfter = 0, maxInterval = 200000}}
+    aCfg = agentCfg {userNetworkInterval = 100000, userOfflineDelay = 0}
 
-testDoNotResetOfflineInterval :: IO ()
-testDoNotResetOfflineInterval = do
+testDoNotResetOnlineToOffline :: IO ()
+testDoNotResetOnlineToOffline = do
   a <- getSMPAgentClient' 1 aCfg initAgentServers testDB
   noNetworkDelay a
   setUserNetworkInfo a $ UserNetworkInfo UNWifi False
-  threadDelay 5000
   networkDelay a 100000
-  networkDelay a 150000
-  setUserNetworkInfo a $ UserNetworkInfo UNCellular False
-  networkDelay a 200000
-  setUserNetworkInfo a $ UserNetworkInfo UNNone False
-  networkDelay a 200000
-  setUserNetworkInfo a $ UserNetworkInfo UNCellular True
+  setUserNetworkInfo a $ UserNetworkInfo UNWifi False
+  setUserNetworkInfo a $ UserNetworkInfo UNWifi True
   noNetworkDelay a
-  setUserNetworkInfo a $ UserNetworkInfo UNCellular False
+  setUserNetworkInfo a $ UserNetworkInfo UNWifi False -- ingnored
+  noNetworkDelay a
+  threadDelay 100000
+  setUserNetworkInfo a $ UserNetworkInfo UNWifi False
   networkDelay a 100000
+  setUserNetworkInfo a $ UserNetworkInfo UNNone False
+  networkDelay a 100000
+  setUserNetworkInfo a $ UserNetworkInfo UNWifi True
+  setUserNetworkInfo a $ UserNetworkInfo UNNone False -- ingnored
+  noNetworkDelay a
   where
-    aCfg = agentCfg {userNetworkInterval = RetryInterval {initialInterval = 100000, increaseAfter = 0, maxInterval = 200000}}
+    aCfg = agentCfg {userNetworkInterval = 100000, userOfflineDelay = 0.1}
 
 testResumeMultipleThreads :: IO ()
 testResumeMultipleThreads = do
@@ -2746,14 +2746,14 @@ testResumeMultipleThreads = do
   threadDelay 1000000
   setUserNetworkInfo a $ UserNetworkInfo UNCellular True
   ts <- mapM (atomically . readTMVar) vs
-  print $ minimum ts
-  print $ maximum ts
-  print $ sum ts `div` fromIntegral (length ts)
+  -- print $ minimum ts
+  -- print $ maximum ts
+  -- print $ sum ts `div` fromIntegral (length ts)
   let average = sum ts `div` fromIntegral (length ts)
   average < 3000000 `shouldBe` True
   maximum ts < 4000000 `shouldBe` True
   where
-    aCfg = agentCfg {userNetworkInterval = RetryInterval {initialInterval = 1000000, increaseAfter = 0, maxInterval = 3600_000_000}}
+    aCfg = agentCfg {userOfflineDelay = 0}
 
 noNetworkDelay :: AgentClient -> IO ()
 noNetworkDelay a = do

tests/AgentTests/NotificationTests.hs

@@ -6,8 +6,8 @@
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE PatternSynonyms #-}
 {-# LANGUAGE TypeApplications #-}
-{-# OPTIONS_GHC -fno-warn-incomplete-uni-patterns #-}
 {-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
+{-# OPTIONS_GHC -fno-warn-incomplete-uni-patterns #-}
 
 module AgentTests.NotificationTests where
 
@@ -17,10 +17,6 @@ import AgentTests.FunctionalAPITests
     createConnection,
     exchangeGreetingsMsgId,
     get,
-    withAgent,
-    withAgentClients2,
-    withAgentClientsCfgServers2,
-    withAgentClients3,
     joinConnection,
     makeConnection,
     nGet,
@@ -29,7 +25,11 @@ import AgentTests.FunctionalAPITests
     sendMessage,
     switchComplete,
     testServerMatrix2,
+    withAgent,
+    withAgentClients2,
+    withAgentClients3,
     withAgentClientsCfg2,
+    withAgentClientsCfgServers2,
     (##>),
     (=##>),
     pattern CON,
@@ -59,8 +59,8 @@ import Simplex.Messaging.Agent.Protocol hiding (CON, CONF, INFO)
 import Simplex.Messaging.Agent.Store.SQLite (getSavedNtfToken)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
-import Simplex.Messaging.Notifications.Server.Env (NtfServerConfig (..))
 import Simplex.Messaging.Notifications.Protocol
+import Simplex.Messaging.Notifications.Server.Env (NtfServerConfig (..))
 import Simplex.Messaging.Notifications.Server.Push.APNS
 import Simplex.Messaging.Notifications.Types (NtfToken (..))
 import Simplex.Messaging.Protocol (ErrorType (AUTH), MsgFlags (MsgFlags), NtfServer, ProtocolServer (..), SMPMsgMeta (..), SubscriptionMode (..))
@@ -151,7 +151,8 @@ testNtfMatrix t runTest = do
     it "next servers: SMP v7, NTF v2; curr clients: v6/v1" $ runNtfTestCfg t cfgV7 ntfServerCfgV2 agentCfg agentCfg runTest
     it "curr servers: SMP v6, NTF v1; curr clients: v6/v1" $ runNtfTestCfg t cfg ntfServerCfg agentCfg agentCfg runTest
     skip "this case cannot be supported - see RFC" $
-      it "servers: SMP v6, NTF v1; clients: v7/v2 (not supported)" $ runNtfTestCfg t cfg ntfServerCfg agentCfgV7 agentCfgV7 runTest
+      it "servers: SMP v6, NTF v1; clients: v7/v2 (not supported)" $
+        runNtfTestCfg t cfg ntfServerCfg agentCfgV7 agentCfgV7 runTest
     -- servers can be migrated in any order
     it "servers: next SMP v7, curr NTF v1; curr clients: v6/v1" $ runNtfTestCfg t cfgV7 ntfServerCfg agentCfg agentCfg runTest
     it "servers: curr SMP v6, next NTF v2; curr clients: v6/v1" $ runNtfTestCfg t cfg ntfServerCfgV2 agentCfg agentCfg runTest
@@ -258,7 +259,7 @@ testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
         atomically $ readTBQueue apnsQ
       liftIO $ sendApnsResponse APNSRespOk
       pure ntfData
-    -- the new agent is created as otherwise when running the tests in CI the old agent was keeping the connection to the server
+  -- the new agent is created as otherwise when running the tests in CI the old agent was keeping the connection to the server
   threadDelay 1000000
   withAgent 2 agentCfg initAgentServers testDB $ \a' ->
     -- server stopped before token is verified, so now the attempt to verify it will return AUTH error but re-register token,