Merge remote-tracking branch 'origin/master' into ab/bench-target

CHANGELOG.md

@@ -1,3 +1,23 @@
+# 5.6.0
+
+Version 5.6.0.4.
+
+SMP protocol/client/server:
+- support deniable sender command authorization (to be enabled in the next version).
+- remove support for SMP protocol versions (prior to v4, 07/2022).
+
+Agent:
+- optional post-quantum key agreement using sntrup761 in double ratchet protocol.
+- improve performance of deleting multiple connections and files by batching database operations.
+- delay connection deletion to deliver pending messages.
+- API to test for notifications server.
+- remove support for client protocols versions (prior to 10/2022).
+
+XFTP server:
+- restore storage quota in case of failed uploads.
+
+Performance and stability improvements.
+
 # 5.5.3
 
 Agent:

package.yaml

@@ -1,5 +1,5 @@
 name: simplexmq
-version: 5.6.0.2
+version: 5.6.0.4
 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.6.0.2
+version:        5.6.0.4
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and

src/Simplex/FileTransfer/Server.hs

@@ -21,12 +21,12 @@ import qualified Data.ByteString.Base64.URL as B64
 import Data.ByteString.Builder (byteString)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
-import Data.Functor (($>))
 import Data.Int (Int64)
 import Data.List (intercalate)
 import Data.List.NonEmpty (NonEmpty)
 import qualified Data.List.NonEmpty as L
 import qualified Data.Map.Strict as M
+import Data.Maybe (fromMaybe, isJust)
 import qualified Data.Text as T
 import Data.Time.Clock (UTCTime (..), diffTimeToPicoseconds, getCurrentTime)
 import Data.Time.Clock.System (SystemTime (..), getSystemTime)
@@ -346,35 +346,36 @@ processXFTPRequest HTTP2Body {bodyPart} = \case
         pure $ FRRcvIds rIds
       pure $ either FRErr id r
     receiveServerFile :: FileRec -> M FileResponse
-    receiveServerFile fr@FileRec {senderId, fileInfo} = case bodyPart of
-      -- TODO do not allow repeated file upload
+    receiveServerFile FileRec {senderId, fileInfo = FileInfo {size, digest}, filePath} = case bodyPart of
       Nothing -> pure $ FRErr SIZE
-      Just getBody -> do
-        -- TODO validate body size before downloading, once it's populated
-        path <- asks $ filesPath . config
-        let fPath = path </> B.unpack (B64.encode senderId)
-            FileInfo {size, digest} = fileInfo
-        withFileLog $ \sl -> logPutFile sl senderId fPath
-        st <- asks store
-        quota_ <- asks $ fileSizeQuota . config
-        -- TODO timeout file upload, remove partially uploaded files
-        stats <- asks serverStats
-        liftIO $
-          runExceptT (receiveFile getBody (XFTPRcvChunkSpec fPath size digest)) >>= \case
-            Right () -> do
-              used <- readTVarIO $ usedStorage st
-              if maybe False (used + fromIntegral size >) quota_
-                then remove fPath $> FRErr QUOTA
-                else do
-                  atomically (setFilePath' st fr fPath)
-                  atomically $ modifyTVar' (filesUploaded stats) (+ 1)
-                  atomically $ modifyTVar' (filesCount stats) (+ 1)
-                  atomically $ modifyTVar' (filesSize stats) (+ fromIntegral size)
-                  pure FROk
-            Left e -> remove fPath $> FRErr e
+      -- TODO validate body size from request before downloading, once it's populated
+      Just getBody -> ifM reserve receive (pure $ FRErr QUOTA) -- TODO: handle duplicate uploads
         where
-          remove fPath = whenM (doesFileExist fPath) (removeFile fPath) `catch` logFileError
-
+          reserve = do
+            us <- asks $ usedStorage . store
+            quota <- asks $ fromMaybe maxBound . fileSizeQuota . config
+            atomically . stateTVar us $
+              \used -> let used' = used + fromIntegral size in if used' <= quota then (True, used') else (False, used)
+          receive = do
+            path <- asks $ filesPath . config
+            let fPath = path </> B.unpack (B64.encode senderId)
+            receiveChunk (XFTPRcvChunkSpec fPath size digest) >>= \case
+              Right () -> do
+                stats <- asks serverStats
+                withFileLog $ \sl -> logPutFile sl senderId fPath
+                atomically $ writeTVar filePath (Just fPath)
+                atomically $ modifyTVar' (filesUploaded stats) (+ 1)
+                atomically $ modifyTVar' (filesCount stats) (+ 1)
+                atomically $ modifyTVar' (filesSize stats) (+ fromIntegral size)
+                pure FROk
+              Left e -> do
+                us <- asks $ usedStorage . store
+                atomically . modifyTVar' us $ subtract (fromIntegral size)
+                liftIO $ whenM (doesFileExist fPath) (removeFile fPath) `catch` logFileError
+                pure $ FRErr e
+          receiveChunk spec = do
+            t <- asks $ fileTimeout . config
+            liftIO $ fromMaybe (Left TIMEOUT) <$> timeout t (runExceptT (receiveFile getBody spec) `catchAll_` pure (Left FILE_IO))
     sendServerFile :: FileRec -> RcvPublicDhKey -> M (FileResponse, Maybe ServerFile)
     sendServerFile FileRec {senderId, filePath, fileInfo = FileInfo {size}} rDhKey = do
       readTVarIO filePath >>= \case

src/Simplex/FileTransfer/Server/Env.hs

@@ -14,12 +14,13 @@ import Control.Monad.IO.Unlift
 import Crypto.Random
 import Data.Int (Int64)
 import Data.List.NonEmpty (NonEmpty)
+import qualified Data.Map.Strict as M
 import Data.Time.Clock (getCurrentTime)
 import Data.Word (Word32)
 import Data.X509.Validation (Fingerprint (..))
 import Network.Socket
 import qualified Network.TLS as T
-import Simplex.FileTransfer.Protocol (FileCmd, FileInfo, XFTPFileId)
+import Simplex.FileTransfer.Protocol (FileCmd, FileInfo (..), XFTPFileId)
 import Simplex.FileTransfer.Server.Stats
 import Simplex.FileTransfer.Server.Store
 import Simplex.FileTransfer.Server.StoreLog
@@ -47,6 +48,8 @@ data XFTPServerConfig = XFTPServerConfig
     newFileBasicAuth :: Maybe BasicAuth,
     -- | time after which the files can be removed and check interval, seconds
     fileExpiration :: Maybe ExpirationConfig,
+    -- | timeout to receive file
+    fileTimeout :: Int,
     -- | time after which inactive clients can be disconnected and check interval, seconds
     inactiveClientExpiration :: Maybe ExpirationConfig,
     -- CA certificate private key is not needed for initialization
@@ -93,7 +96,8 @@ newXFTPServerEnv config@XFTPServerConfig {storeLogFile, fileSizeQuota, caCertifi
   random <- liftIO C.newRandom
   store <- atomically newFileStore
   storeLog <- liftIO $ mapM (`readWriteFileStore` store) storeLogFile
-  used <- readTVarIO (usedStorage store)
+  used <- countUsedStorage <$> readTVarIO (files store)
+  atomically $ writeTVar (usedStorage store) used
   forM_ fileSizeQuota $ \quota -> do
     logInfo $ "Total / available storage: " <> tshow quota <> " / " <> tshow (quota - used)
     when (quota < used) $ logInfo "WARNING: storage quota is less than used storage, no files can be uploaded!"
@@ -102,6 +106,9 @@ newXFTPServerEnv config@XFTPServerConfig {storeLogFile, fileSizeQuota, caCertifi
   serverStats <- atomically . newFileServerStats =<< liftIO getCurrentTime
   pure XFTPEnv {config, store, storeLog, random, tlsServerParams, serverIdentity = C.KeyHash fp, serverStats}
 
+countUsedStorage :: M.Map k FileRec -> Int64
+countUsedStorage = M.foldl' (\acc FileRec {fileInfo = FileInfo {size}} -> acc + fromIntegral size) 0
+
 data XFTPRequest
   = XFTPReqNew FileInfo (NonEmpty RcvPublicAuthKey) (Maybe BasicAuth)
   | XFTPReqCmd XFTPFileId FileRec FileCmd

src/Simplex/FileTransfer/Server/Main.hs

@@ -160,6 +160,7 @@ xftpServerCLI cfgPath logPath = do
                   defaultFileExpiration
                     { ttl = 3600 * readIniDefault defFileExpirationHours "STORE_LOG" "expire_files_hours" ini
                     },
+              fileTimeout = 10 * 60 * 1000000, -- 10 mins to send 4mb chunk
               inactiveClientExpiration =
                 settingIsOn "INACTIVE_CLIENTS" "disconnect" ini
                   $> ExpirationConfig

src/Simplex/FileTransfer/Server/Store.hs

@@ -11,7 +11,6 @@ module Simplex.FileTransfer.Server.Store
     newFileStore,
     addFile,
     setFilePath,
-    setFilePath',
     addRecipient,
     deleteFile,
     deleteRecipient,
@@ -79,12 +78,10 @@ newFileRec senderId fileInfo createdAt = do
 
 setFilePath :: FileStore -> SenderId -> FilePath -> STM (Either XFTPErrorType ())
 setFilePath st sId fPath =
-  withFile st sId $ \fr -> setFilePath' st fr fPath $> Right ()
-
-setFilePath' :: FileStore -> FileRec -> FilePath -> STM ()
-setFilePath' st FileRec {fileInfo, filePath} fPath = do
-  writeTVar filePath (Just fPath)
-  modifyTVar' (usedStorage st) (+ fromIntegral (size fileInfo))
+  withFile st sId $ \FileRec {fileInfo, filePath} -> do
+    writeTVar filePath (Just fPath)
+    modifyTVar' (usedStorage st) (+ fromIntegral (size fileInfo))
+    pure $ Right ()
 
 addRecipient :: FileStore -> SenderId -> FileRecipient -> STM (Either XFTPErrorType ())
 addRecipient st@FileStore {recipients} senderId (FileRecipient rId rKey) =

src/Simplex/FileTransfer/Transport.hs

@@ -157,6 +157,8 @@ data XFTPErrorType
     HAS_FILE
   | -- | file IO error
     FILE_IO
+  | -- | file sending timeout
+    TIMEOUT
   | -- | bad redirect data
     REDIRECT {redirectError :: String}
   | -- | internal server error
@@ -188,6 +190,7 @@ instance Encoding XFTPErrorType where
     NO_FILE -> "NO_FILE"
     HAS_FILE -> "HAS_FILE"
     FILE_IO -> "FILE_IO"
+    TIMEOUT -> "TIMEOUT"
     REDIRECT err -> "REDIRECT " <> smpEncode err
     INTERNAL -> "INTERNAL"
     DUPLICATE_ -> "DUPLICATE_"
@@ -205,6 +208,7 @@ instance Encoding XFTPErrorType where
       "NO_FILE" -> pure NO_FILE
       "HAS_FILE" -> pure HAS_FILE
       "FILE_IO" -> pure FILE_IO
+      "TIMEOUT" -> pure TIMEOUT
       "REDIRECT" -> REDIRECT <$> _smpP
       "INTERNAL" -> pure INTERNAL
       "DUPLICATE_" -> pure DUPLICATE_

src/Simplex/Messaging/Agent.hs

@@ -41,6 +41,7 @@ module Simplex.Messaging.Agent
     getSMPAgentClient,
     getSMPAgentClient_,
     disconnectAgentClient,
+    disposeAgentClient,
     resumeAgentClient,
     withConnLock,
     withInvLock,
@@ -122,7 +123,7 @@ import Control.Monad
 import Control.Monad.Except
 import Control.Monad.IO.Unlift (MonadUnliftIO)
 import Control.Monad.Reader
-import Crypto.Random (ChaChaDRG, MonadRandom)
+import Crypto.Random (ChaChaDRG)
 import qualified Data.Aeson as J
 import Data.Bifunctor (bimap, first, second)
 import Data.ByteString.Char8 (ByteString)
@@ -179,28 +180,40 @@ import Simplex.Messaging.Version
 import Simplex.RemoteControl.Client
 import Simplex.RemoteControl.Invitation
 import Simplex.RemoteControl.Types
+import System.Mem.Weak (deRefWeak)
 import UnliftIO.Async (race_)
-import UnliftIO.Concurrent (forkFinally, forkIO, threadDelay)
+import UnliftIO.Concurrent (forkFinally, forkIO, killThread, mkWeakThreadId, threadDelay)
+import qualified UnliftIO.Exception as E
 import UnliftIO.STM
 
 -- import GHC.Conc (unsafeIOToSTM)
 
 -- | Creates an SMP agent client instance
-getSMPAgentClient :: (MonadRandom m, MonadUnliftIO m) => AgentConfig -> InitialAgentServers -> SQLiteStore -> Bool -> m AgentClient
+getSMPAgentClient :: MonadIO m => AgentConfig -> InitialAgentServers -> SQLiteStore -> Bool -> m AgentClient
 getSMPAgentClient = getSMPAgentClient_ 1
 {-# INLINE getSMPAgentClient #-}
 
-getSMPAgentClient_ :: (MonadRandom m, MonadUnliftIO m) => Int -> AgentConfig -> InitialAgentServers -> SQLiteStore -> Bool -> m AgentClient
+getSMPAgentClient_ :: MonadIO m => Int -> AgentConfig -> InitialAgentServers -> SQLiteStore -> Bool -> m AgentClient
 getSMPAgentClient_ clientId cfg initServers store backgroundMode =
-  liftIO (newSMPAgentEnv cfg store) >>= runReaderT runAgent
+  liftIO $ newSMPAgentEnv cfg store >>= runReaderT runAgent
   where
     runAgent = do
-      c <- getAgentClient clientId initServers
-      void $ runAgentThreads c `forkFinally` const (disconnectAgentClient c)
+      c@AgentClient {acThread}  <- getAgentClient clientId initServers
+      t <- runAgentThreads c `forkFinally` const (disconnectAgentClient c)
+      atomically . writeTVar acThread . Just =<< mkWeakThreadId t
       pure c
     runAgentThreads c
-      | backgroundMode = subscriber c
-      | otherwise = raceAny_ [subscriber c, runNtfSupervisor c, cleanupManager c]
+      | backgroundMode = run c "subscriber" $ subscriber c
+      | otherwise =
+          raceAny_
+            [ run c "subscriber" $ subscriber c,
+              run c "runNtfSupervisor" $ runNtfSupervisor c,
+              run c "cleanupManager" $ cleanupManager c
+            ]
+    run AgentClient {subQ, acThread} name a =
+      a `E.catchAny` \e -> whenM (isJust <$> readTVarIO acThread) $ do
+        logError $ "Agent thread " <> name <> " crashed: " <> tshow e
+        atomically $ writeTBQueue subQ ("", "", APC SAEConn $ ERR $ CRITICAL True $ show e)
 
 disconnectAgentClient :: MonadUnliftIO m => AgentClient -> m ()
 disconnectAgentClient c@AgentClient {agentEnv = Env {ntfSupervisor = ns, xftpAgent = xa}} = do
@@ -209,6 +222,14 @@ disconnectAgentClient c@AgentClient {agentEnv = Env {ntfSupervisor = ns, xftpAge
   closeXFTPAgent xa
   logConnection c False
 
+-- only used in the tests
+disposeAgentClient :: MonadUnliftIO m => AgentClient -> m ()
+disposeAgentClient c@AgentClient {acThread, agentEnv = Env {store}} = do
+  t_ <- atomically (swapTVar acThread Nothing) $>>= (liftIO . deRefWeak)
+  disconnectAgentClient c
+  mapM_ killThread t_
+  liftIO $ closeSQLiteStore store
+
 resumeAgentClient :: MonadIO m => AgentClient -> m ()
 resumeAgentClient c = atomically $ writeTVar (active c) True
 
@@ -1916,9 +1937,11 @@ cleanupManager c@AgentClient {subQ} = do
   where
     run :: forall e. AEntityI e => (AgentErrorType -> ACommand 'Agent e) -> ExceptT AgentErrorType m () -> m ()
     run err a = do
-      void . runExceptT $ a `catchAgentError` (notify "" . err)
+      waitActive . runExceptT $ a `catchAgentError` (notify "" . err)
       step <- asks $ cleanupStepInterval . config
       liftIO $ threadDelay step
+    -- we are catching it to avoid CRITICAL errors in tests when this is the only remaining handle to active
+    waitActive a = liftIO (E.tryAny . atomically $ waitUntilActive c) >>= either (\_ -> pure ()) (\_ -> void a)
     deleteConns =
       withLock (deleteLock c) "cleanupManager" $ do
         void $ withStore' c getDeletedConnIds >>= deleteDeletedConns c

src/Simplex/Messaging/Agent/Client.hs

@@ -129,7 +129,7 @@ module Simplex.Messaging.Agent.Client
 where
 
 import Control.Applicative ((<|>))
-import Control.Concurrent (forkIO, threadDelay)
+import Control.Concurrent (ThreadId, forkIO, threadDelay)
 import Control.Concurrent.Async (Async, uninterruptibleCancel)
 import Control.Concurrent.STM (retry, throwSTM)
 import Control.Exception (AsyncException (..))
@@ -225,6 +225,7 @@ import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport.Client (TransportHost)
 import Simplex.Messaging.Util
 import Simplex.Messaging.Version
+import System.Mem.Weak (Weak)
 import System.Random (randomR)
 import UnliftIO (mapConcurrently, timeout)
 import UnliftIO.Async (async)
@@ -252,7 +253,8 @@ type NtfTransportSession = TransportSession NtfResponse
 type XFTPTransportSession = TransportSession FileResponse
 
 data AgentClient = AgentClient
-  { active :: TVar Bool,
+  { acThread :: TVar (Maybe (Weak ThreadId)),
+    active :: TVar Bool,
     rcvQ :: TBQueue (ATransmission 'Client),
     subQ :: TBQueue (ATransmission 'Agent),
     msgQ :: TBQueue (ServerTransmission SMPVersion BrokerMsg),
@@ -395,6 +397,7 @@ data AgentStatsKey = AgentStatsKey
 newAgentClient :: Int -> InitialAgentServers -> Env -> STM AgentClient
 newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv = do
   let qSize = tbqSize $ config agentEnv
+  acThread <- newTVar Nothing
   active <- newTVar True
   rcvQ <- newTBQueue qSize
   subQ <- newTBQueue qSize
@@ -428,7 +431,8 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv =
   agentStats <- TM.empty
   return
     AgentClient
-      { active,
+      { acThread,
+        active,
         rcvQ,
         subQ,
         msgQ,
@@ -521,11 +525,11 @@ getSMPServerClient c@AgentClient {active, smpClients, msgQ} tSess@(userId, srv,
     connectClient v = do
       cfg <- getClientConfig c smpCfg
       g <- asks random
-      u <- askUnliftIO
-      liftEitherError (protocolClientError SMP $ B.unpack $ strEncode srv) (getProtocolClient g tSess cfg (Just msgQ) $ clientDisconnected u v)
+      env <- ask
+      liftEitherError (protocolClientError SMP $ B.unpack $ strEncode srv) (getProtocolClient g tSess cfg (Just msgQ) $ clientDisconnected env v)
 
-    clientDisconnected :: UnliftIO m -> SMPClientVar -> SMPClient -> IO ()
-    clientDisconnected u v client = do
+    clientDisconnected :: Env -> SMPClientVar -> SMPClient -> IO ()
+    clientDisconnected env v client = do
       removeClientAndSubs >>= serverDown
       logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv
       where
@@ -548,7 +552,7 @@ getSMPServerClient c@AgentClient {active, smpClients, msgQ} tSess@(userId, srv,
           unless (null conns) $ notifySub "" $ DOWN srv conns
           unless (null qs) $ do
             atomically $ mapM_ (releaseGetLock c) qs
-            unliftIO u $ resubscribeSMPSession c tSess
+            runReaderT (resubscribeSMPSession c tSess) env
 
         notifySub :: forall e. AEntityI e => ConnId -> ACommand 'Agent e -> IO ()
         notifySub connId cmd = atomically $ writeTBQueue (subQ c) ("", connId, APC (sAEntity @e) cmd)
@@ -1066,19 +1070,19 @@ subscribeQueues c qs = do
   atomically $ do
     modifyTVar' (subscrConns c) (`S.union` S.fromList (map qConnId qs'))
     RQ.batchAddQueues (pendingSubs c) qs'
-  u <- askUnliftIO
+  env <- ask
   -- only "checked" queues are subscribed
-  (errs <>) <$> sendTSessionBatches "SUB" 90 id (subscribeQueues_ u) c qs'
+  (errs <>) <$> sendTSessionBatches "SUB" 90 id (subscribeQueues_ env) c qs'
   where
     checkQueue rq = do
       prohibited <- atomically $ hasGetLock c rq
       pure $ if prohibited then Left (rq, Left $ CMD PROHIBITED) else Right rq
-    subscribeQueues_ :: UnliftIO m -> SMPClient -> NonEmpty RcvQueue -> IO (BatchResponses SMPClientError ())
-    subscribeQueues_ u smp qs' = do
+    subscribeQueues_ :: Env -> SMPClient -> NonEmpty RcvQueue -> IO (BatchResponses SMPClientError ())
+    subscribeQueues_ env smp qs' = do
       rs <- sendBatch subscribeSMPQueues smp qs'
       mapM_ (uncurry $ processSubResult c) rs
-      when (any temporaryClientError . lefts . map snd $ L.toList rs) . unliftIO u $
-        resubscribeSMPSession c (transportSession' smp)
+      when (any temporaryClientError . lefts . map snd $ L.toList rs) $
+        runReaderT (resubscribeSMPSession c $ transportSession' smp) env
       pure rs
 
 type BatchResponses e r = (NonEmpty (RcvQueue, Either e r))

src/Simplex/Messaging/Compression.hs

@@ -3,6 +3,7 @@
 
 module Simplex.Messaging.Compression where
 
+import qualified Codec.Compression.Zstd as Z1
 import qualified Codec.Compression.Zstd.FFI as Z
 import Control.Monad (forM)
 import Control.Monad.Except
@@ -28,6 +29,9 @@ data Compressed
 maxLengthPassthrough :: Int
 maxLengthPassthrough = 180 -- Sampled from real client data. Messages with length > 180 rapidly gain compression ratio.
 
+compressionLevel :: Num a => a
+compressionLevel = 3
+
 instance Encoding Compressed where
   smpEncode = \case
     Passthrough bytes -> "0" <> smpEncode bytes
@@ -38,7 +42,11 @@ instance Encoding Compressed where
       '1' -> Compressed <$> smpP
       x -> fail $ "unknown Compressed tag: " <> show x
 
--- ** Batch compression context
+-- | Compress as single chunk using stack-allocated context.
+compress1 :: ByteString -> Compressed
+compress1 bs
+  | B.length bs <= maxLengthPassthrough = Passthrough bs
+  | otherwise = Compressed . Large $ Z1.compress compressionLevel bs
 
 type CompressCtx = (Ptr Z.CCtx, Ptr CChar, CSize)
 
@@ -66,7 +74,7 @@ compress_ (cctx, scratchPtr, scratchSize) bs
   | otherwise =
       B.unsafeUseAsCStringLen bs $ \(sourcePtr, sourceSize) -> runExceptT $ do
         -- should not fail, unless input buffer is too short
-        dstSize <- ExceptT $ Z.checkError $ Z.compressCCtx cctx scratchPtr scratchSize sourcePtr (fromIntegral sourceSize) 3
+        dstSize <- ExceptT $ Z.checkError $ Z.compressCCtx cctx scratchPtr scratchSize sourcePtr (fromIntegral sourceSize) compressionLevel
         liftIO $ Compressed . Large <$> B.packCStringLen (scratchPtr, fromIntegral dstSize)
 
 type DecompressCtx = (Ptr Z.DCtx, Ptr CChar, CSize)

tests/AgentTests/FunctionalAPITests.hs

@@ -473,8 +473,8 @@ withAgentClientsCfg2 aCfg bCfg runTest = do
   a <- getSMPAgentClient' 1 aCfg initAgentServers testDB
   b <- getSMPAgentClient' 2 bCfg initAgentServers testDB2
   runTest a b
-  disconnectAgentClient a
-  disconnectAgentClient b
+  disposeAgentClient a
+  disposeAgentClient b
 
 withAgentClients2 :: (AgentClient -> AgentClient -> IO ()) -> IO ()
 withAgentClients2 = withAgentClientsCfg2 agentCfg agentCfg
@@ -664,7 +664,7 @@ testAsyncInitiatingOffline :: HasCallStack => IO ()
 testAsyncInitiatingOffline =
   withAgentClients2 $ \alice bob -> runRight_ $ do
     (bobId, cReq) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
-    disconnectAgentClient alice
+    disposeAgentClient alice
     aliceId <- joinConnection bob 1 True cReq "bob's connInfo" SMSubscribe
     alice' <- liftIO $ getSMPAgentClient' 3 agentCfg initAgentServers testDB
     subscribeConnection alice' bobId
@@ -680,7 +680,7 @@ testAsyncJoiningOfflineBeforeActivation =
   withAgentClients2 $ \alice bob -> runRight_ $ do
     (bobId, qInfo) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
     aliceId <- joinConnection bob 1 True qInfo "bob's connInfo" SMSubscribe
-    disconnectAgentClient bob
+    disposeAgentClient bob
     ("", _, CONF confId _ "bob's connInfo") <- get alice
     allowConnection alice bobId confId "alice's connInfo"
     bob' <- liftIO $ getSMPAgentClient' 3 agentCfg initAgentServers testDB2
@@ -694,9 +694,9 @@ testAsyncBothOffline :: HasCallStack => IO ()
 testAsyncBothOffline =
   withAgentClients2 $ \alice bob -> runRight_ $ do
     (bobId, cReq) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
-    disconnectAgentClient alice
+    disposeAgentClient alice
     aliceId <- joinConnection bob 1 True cReq "bob's connInfo" SMSubscribe
-    disconnectAgentClient bob
+    disposeAgentClient bob
     alice' <- liftIO $ getSMPAgentClient' 3 agentCfg initAgentServers testDB
     subscribeConnection alice' bobId
     ("", _, CONF confId _ "bob's connInfo") <- get alice'
@@ -754,7 +754,7 @@ testAllowConnectionClientRestart t = do
       pure ()
 
     threadDelay 100000 -- give time to enqueue confirmation (enqueueConfirmation)
-    disconnectAgentClient alice
+    disposeAgentClient alice
 
     alice2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
 
@@ -769,8 +769,8 @@ testAllowConnectionClientRestart t = do
         get bob ##> ("", aliceId, CON)
 
         exchangeGreetingsMsgId 4 alice2 bobId bob aliceId
-    disconnectAgentClient alice2
-    disconnectAgentClient bob
+    disposeAgentClient alice2
+    disposeAgentClient bob
 
 testIncreaseConnAgentVersion :: HasCallStack => ATransport -> IO ()
 testIncreaseConnAgentVersion t = do
@@ -786,7 +786,7 @@ testIncreaseConnAgentVersion t = do
 
     -- version doesn't increase if incompatible
 
-    disconnectAgentClient alice
+    disposeAgentClient alice
     alice2 <- getSMPAgentClient' 3 agentCfg {smpAgentVRange = \_ -> mkVersionRange 1 3} initAgentServers testDB
 
     runRight_ $ do
@@ -797,7 +797,7 @@ testIncreaseConnAgentVersion t = do
 
     -- version increases if compatible
 
-    disconnectAgentClient bob
+    disposeAgentClient bob
     bob2 <- getSMPAgentClient' 4 agentCfg {smpAgentVRange = \_ -> mkVersionRange 1 3} initAgentServers testDB2
 
     runRight_ $ do
@@ -808,7 +808,7 @@ testIncreaseConnAgentVersion t = do
 
     -- version doesn't decrease, even if incompatible
 
-    disconnectAgentClient alice2
+    disposeAgentClient alice2
     alice3 <- getSMPAgentClient' 5 agentCfg {smpAgentVRange = \_ -> mkVersionRange 2 2} initAgentServers testDB
 
     runRight_ $ do
@@ -817,7 +817,7 @@ testIncreaseConnAgentVersion t = do
       checkVersion alice3 bobId 3
       checkVersion bob2 aliceId 3
 
-    disconnectAgentClient bob2
+    disposeAgentClient bob2
     bob3 <- getSMPAgentClient' 6 agentCfg {smpAgentVRange = \_ -> mkVersionRange 1 1} initAgentServers testDB2
 
     runRight_ $ do
@@ -825,8 +825,8 @@ testIncreaseConnAgentVersion t = do
       exchangeGreetingsMsgId_ PQEncOff 12 alice3 bobId bob3 aliceId
       checkVersion alice3 bobId 3
       checkVersion bob3 aliceId 3
-    disconnectAgentClient alice3
-    disconnectAgentClient bob3
+    disposeAgentClient alice3
+    disposeAgentClient bob3
 
 checkVersion :: AgentClient -> ConnId -> Word16 -> ExceptT AgentErrorType IO ()
 checkVersion c connId v = do
@@ -847,9 +847,9 @@ testIncreaseConnAgentVersionMaxCompatible t = do
 
     -- version increases to max compatible
 
-    disconnectAgentClient alice
+    disposeAgentClient alice
     alice2 <- getSMPAgentClient' 3 agentCfg {smpAgentVRange = \_ -> mkVersionRange 1 3} initAgentServers testDB
-    disconnectAgentClient bob
+    disposeAgentClient bob
     bob2 <- getSMPAgentClient' 4 agentCfg {smpAgentVRange = supportedSMPAgentVRange} initAgentServers testDB2
 
     runRight_ $ do
@@ -858,8 +858,8 @@ testIncreaseConnAgentVersionMaxCompatible t = do
       exchangeGreetingsMsgId_ PQEncOff 6 alice2 bobId bob2 aliceId
       checkVersion alice2 bobId 3
       checkVersion bob2 aliceId 3
-    disconnectAgentClient alice2
-    disconnectAgentClient bob2
+    disposeAgentClient alice2
+    disposeAgentClient bob2
 
 testIncreaseConnAgentVersionStartDifferentVersion :: HasCallStack => ATransport -> IO ()
 testIncreaseConnAgentVersionStartDifferentVersion t = do
@@ -875,7 +875,7 @@ testIncreaseConnAgentVersionStartDifferentVersion t = do
 
     -- version increases to max compatible
 
-    disconnectAgentClient alice
+    disposeAgentClient alice
     alice2 <- getSMPAgentClient' 3 agentCfg {smpAgentVRange = \_ -> mkVersionRange 1 3} initAgentServers testDB
 
     runRight_ $ do
@@ -883,8 +883,8 @@ testIncreaseConnAgentVersionStartDifferentVersion t = do
       exchangeGreetingsMsgId_ PQEncOff 6 alice2 bobId bob aliceId
       checkVersion alice2 bobId 3
       checkVersion bob aliceId 3
-    disconnectAgentClient alice2
-    disconnectAgentClient bob
+    disposeAgentClient alice2
+    disposeAgentClient bob
 
 testDeliverClientRestart :: HasCallStack => ATransport -> IO ()
 testDeliverClientRestart t = do
@@ -902,7 +902,7 @@ testDeliverClientRestart t = do
 
   6 <- runRight $ sendMessage bob aliceId SMP.noMsgFlags "hello"
 
-  disconnectAgentClient bob
+  disposeAgentClient bob
 
   bob2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
 
@@ -914,8 +914,8 @@ testDeliverClientRestart t = do
 
       get bob2 ##> ("", aliceId, SENT 6)
       get alice =##> \case ("", c, Msg "hello") -> c == bobId; _ -> False
-  disconnectAgentClient alice
-  disconnectAgentClient bob2
+  disposeAgentClient alice
+  disposeAgentClient bob2
 
 testDuplicateMessage :: HasCallStack => ATransport -> IO ()
 testDuplicateMessage t = do
@@ -927,7 +927,7 @@ testDuplicateMessage t = do
       4 <- sendMessage alice bobId SMP.noMsgFlags "hello"
       get alice ##> ("", bobId, SENT 4)
       get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
-    disconnectAgentClient bob
+    disposeAgentClient bob
 
     -- if the agent user did not send ACK, the message will be delivered again
     bob1 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
@@ -948,8 +948,8 @@ testDuplicateMessage t = do
   threadDelay 200000
   Left (BROKER _ NETWORK) <- runExceptT $ ackMessage bob1 aliceId 5 Nothing
 
-  disconnectAgentClient alice
-  disconnectAgentClient bob1
+  disposeAgentClient alice
+  disposeAgentClient bob1
 
   alice2 <- getSMPAgentClient' 4 agentCfg initAgentServers testDB
   bob2 <- getSMPAgentClient' 5 agentCfg initAgentServers testDB2
@@ -964,8 +964,8 @@ testDuplicateMessage t = do
       6 <- sendMessage alice2 bobId SMP.noMsgFlags "hello 3"
       get alice2 ##> ("", bobId, SENT 6)
       get bob2 =##> \case ("", c, Msg "hello 3") -> c == aliceId; _ -> False
-  disconnectAgentClient alice2
-  disconnectAgentClient bob2
+  disposeAgentClient alice2
+  disposeAgentClient bob2
 
 testSkippedMessages :: HasCallStack => ATransport -> IO ()
 testSkippedMessages t = do
@@ -979,7 +979,7 @@ testSkippedMessages t = do
       get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
       ackMessage bob aliceId 4 Nothing
 
-    disconnectAgentClient bob
+    disposeAgentClient bob
 
     runRight_ $ do
       5 <- sendMessage alice bobId SMP.noMsgFlags "hello 2"
@@ -994,7 +994,7 @@ testSkippedMessages t = do
   nGet alice =##> \case ("", "", DOWN _ [c]) -> c == bobId; _ -> False
   threadDelay 200000
 
-  disconnectAgentClient alice
+  disposeAgentClient alice
 
   alice2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
   bob2 <- getSMPAgentClient' 4 agentCfg initAgentServers testDB2
@@ -1013,8 +1013,8 @@ testSkippedMessages t = do
       get alice2 ##> ("", bobId, SENT 9)
       get bob2 =##> \case ("", c, Msg "hello 6") -> c == aliceId; _ -> False
       ackMessage bob2 aliceId 6 Nothing
-  disconnectAgentClient alice2
-  disconnectAgentClient bob2
+  disposeAgentClient alice2
+  disposeAgentClient bob2
 
 testExpireMessage :: HasCallStack => ATransport -> IO ()
 testExpireMessage t = do
@@ -1068,7 +1068,7 @@ testExpireMessageQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1} testP
   (aId, bId) <- runRight $ do
     (aId, bId) <- makeConnection a b
     liftIO $ threadDelay 500000
-    disconnectAgentClient b
+    disposeAgentClient b
     4 <- sendMessage a bId SMP.noMsgFlags "1"
     get a ##> ("", bId, SENT 4)
     5 <- sendMessage a bId SMP.noMsgFlags "2"
@@ -1092,7 +1092,7 @@ testExpireManyMessagesQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1}
   (aId, bId) <- runRight $ do
     (aId, bId) <- makeConnection a b
     liftIO $ threadDelay 500000
-    disconnectAgentClient b
+    disposeAgentClient b
     4 <- sendMessage a bId SMP.noMsgFlags "1"
     get a ##> ("", bId, SENT 4)
     5 <- sendMessage a bId SMP.noMsgFlags "2"
@@ -1151,7 +1151,7 @@ setupDesynchronizedRatchet alice bob = do
     get alice =##> \case ("", c, Msg "hello 4") -> c == bobId; _ -> False
     ackMessage alice bobId 7 Nothing
 
-  disconnectAgentClient bob
+  disposeAgentClient bob
 
   -- importing database backup after progressing ratchet de-synchronizes ratchet
   liftIO $ renameFile (testDB2 <> ".bak") testDB2
@@ -1224,7 +1224,7 @@ testRatchetSyncClientRestart t = do
   ("", "", DOWN _ _) <- nGet bob2
   ConnectionStats {ratchetSyncState} <- runRight $ synchronizeRatchet bob2 aliceId PQSupportOn False
   liftIO $ ratchetSyncState `shouldBe` RSStarted
-  disconnectAgentClient bob2
+  disposeAgentClient bob2
   bob3 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
   withSmpServerStoreMsgLogOn t testPort $ \_ -> do
     runRight_ $ do
@@ -1235,9 +1235,9 @@ testRatchetSyncClientRestart t = do
       get alice =##> ratchetSyncP bobId RSOk
       get bob3 =##> ratchetSyncP aliceId RSOk
       exchangeGreetingsMsgIds alice bobId 12 bob3 aliceId 9
-  disconnectAgentClient alice
-  disconnectAgentClient bob
-  disconnectAgentClient bob3
+  disposeAgentClient alice
+  disposeAgentClient bob
+  disposeAgentClient bob3
 
 testRatchetSyncSuspendForeground :: HasCallStack => ATransport -> IO ()
 testRatchetSyncSuspendForeground t = do
@@ -1269,9 +1269,9 @@ testRatchetSyncSuspendForeground t = do
       get alice =##> ratchetSyncP bobId RSOk
       get bob2 =##> ratchetSyncP aliceId RSOk
       exchangeGreetingsMsgIds alice bobId 12 bob2 aliceId 9
-  disconnectAgentClient alice
-  disconnectAgentClient bob
-  disconnectAgentClient bob2
+  disposeAgentClient alice
+  disposeAgentClient bob
+  disposeAgentClient bob2
 
 testRatchetSyncSimultaneous :: HasCallStack => ATransport -> IO ()
 testRatchetSyncSimultaneous t = do
@@ -1302,9 +1302,9 @@ testRatchetSyncSimultaneous t = do
       get alice =##> ratchetSyncP bobId RSOk
       get bob2 =##> ratchetSyncP aliceId RSOk
       exchangeGreetingsMsgIds alice bobId 12 bob2 aliceId 9
-  disconnectAgentClient alice
-  disconnectAgentClient bob
-  disconnectAgentClient bob2
+  disposeAgentClient alice
+  disposeAgentClient bob
+  disposeAgentClient bob2
 
 testOnlyCreatePull :: IO ()
 testOnlyCreatePull = withAgentClients2 $ \alice bob -> runRight_ $ do
@@ -1370,7 +1370,7 @@ testInactiveNoSubs t = do
     runRight_ . void $ createConnection alice 1 True SCMInvitation Nothing SMOnlyCreate -- do not subscribe to pass noSubscriptions check
     Just (_, _, APC SAENone (CONNECT _ _)) <- timeout 2000000 $ atomically (readTBQueue $ subQ alice)
     Just (_, _, APC SAENone (DISCONNECT _ _)) <- timeout 5000000 $ atomically (readTBQueue $ subQ alice)
-    disconnectAgentClient alice
+    disposeAgentClient alice
 
 testInactiveWithSubs :: ATransport -> IO ()
 testInactiveWithSubs t = do
@@ -1382,7 +1382,7 @@ testInactiveWithSubs t = do
     liftIO $ threadDelay 1200000
     -- and after 2 sec of inactivity no DOWN is sent as we have a live subscription
     liftIO $ timeout 1200000 (get alice) `shouldReturn` Nothing
-    disconnectAgentClient alice
+    disposeAgentClient alice
 
 testActiveClientNotDisconnected :: ATransport -> IO ()
 testActiveClientNotDisconnected t = do
@@ -1393,7 +1393,7 @@ testActiveClientNotDisconnected t = do
     runRight_ $ do
       (connId, _cReq) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
       keepSubscribing alice connId ts
-    disconnectAgentClient alice
+    disposeAgentClient alice
   where
     keepSubscribing :: AgentClient -> ConnId -> SystemTime -> ExceptT AgentErrorType IO ()
     keepSubscribing alice connId ts = do
@@ -1512,8 +1512,8 @@ testBatchedSubscriptions nCreate nDel t = do
     delete b aIds'
     deleteFail a bIds'
     deleteFail b aIds'
-  disconnectAgentClient a
-  disconnectAgentClient b
+  disposeAgentClient a
+  disposeAgentClient b
   where
     subscribe :: AgentClient -> [ConnId] -> ExceptT AgentErrorType IO ()
     subscribe c cs = do
@@ -1598,14 +1598,14 @@ testAsyncCommandsRestore t = do
   alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
   bobId <- runRight $ createConnectionAsync alice 1 "1" True SCMInvitation (IKNoPQ PQSupportOn) SMSubscribe
   liftIO $ noMessages alice "alice doesn't receive INV because server is down"
-  disconnectAgentClient alice
+  disposeAgentClient alice
   alice' <- liftIO $ getSMPAgentClient' 2 agentCfg initAgentServers testDB
   withSmpServerStoreLogOn t testPort $ \_ -> do
     runRight_ $ do
       subscribeConnection alice' bobId
       get alice' =##> \case ("1", _, INV _) -> True; _ -> False
       pure ()
-  disconnectAgentClient alice'
+  disposeAgentClient alice'
 
 testAcceptContactAsync :: IO ()
 testAcceptContactAsync =
@@ -1663,7 +1663,7 @@ testDeleteConnectionAsync t = do
     get a =##> \case ("", c, DEL_CONN) -> c `elem` connIds; _ -> False
     get a =##> \case ("", c, DEL_CONN) -> c `elem` connIds; _ -> False
     liftIO $ noMessages a "nothing else should be delivered to alice"
-  disconnectAgentClient a
+  disposeAgentClient a
 
 testWaitDeliveryNoPending :: ATransport -> IO ()
 testWaitDeliveryNoPending t = do
@@ -1692,8 +1692,8 @@ testWaitDeliveryNoPending t = do
     liftIO $ noMessages alice "nothing else should be delivered to alice"
     liftIO $ noMessages bob "nothing else should be delivered to bob"
 
-  disconnectAgentClient alice
-  disconnectAgentClient bob
+  disposeAgentClient alice
+  disposeAgentClient bob
   where
     baseId = 3
     msgId = subtract baseId
@@ -1749,8 +1749,8 @@ testWaitDelivery t = do
     liftIO $ noMessages alice "nothing else should be delivered to alice"
     liftIO $ noMessages bob "nothing else should be delivered to bob"
 
-  disconnectAgentClient alice
-  disconnectAgentClient bob
+  disposeAgentClient alice
+  disposeAgentClient bob
   where
     baseId = 3
     msgId = subtract baseId
@@ -1795,8 +1795,8 @@ testWaitDeliveryAUTHErr t = do
     liftIO $ noMessages alice "nothing else should be delivered to alice"
     liftIO $ noMessages bob "nothing else should be delivered to bob"
 
-  disconnectAgentClient alice
-  disconnectAgentClient bob
+  disposeAgentClient alice
+  disposeAgentClient bob
   where
     baseId = 3
     msgId = subtract baseId
@@ -1838,8 +1838,8 @@ testWaitDeliveryTimeout t = do
     liftIO $ noMessages alice "nothing else should be delivered to alice"
     liftIO $ noMessages bob "nothing else should be delivered to bob"
 
-  disconnectAgentClient alice
-  disconnectAgentClient bob
+  disposeAgentClient alice
+  disposeAgentClient bob
   where
     baseId = 3
     msgId = subtract baseId
@@ -1887,8 +1887,8 @@ testWaitDeliveryTimeout2 t = do
     liftIO $ noMessages alice "nothing else should be delivered to alice"
     liftIO $ noMessages bob "nothing else should be delivered to bob"
 
-  disconnectAgentClient alice
-  disconnectAgentClient bob
+  disposeAgentClient alice
+  disposeAgentClient bob
   where
     baseId = 3
     msgId = subtract baseId
@@ -1931,8 +1931,8 @@ testJoinConnectionAsyncReplyError t = do
       get b ##> ("", aId, INFO "alice's connInfo")
       get b ##> ("", aId, CON)
       exchangeGreetings a bId b aId
-  disconnectAgentClient a
-  disconnectAgentClient b
+  disposeAgentClient a
+  disposeAgentClient b
 
 testUsers :: IO ()
 testUsers =
@@ -1995,8 +1995,8 @@ testSwitchConnection servers = do
     exchangeGreetingsMsgId 4 a bId b aId
     testFullSwitch a bId b aId 10
     testFullSwitch a bId b aId 16
-  disconnectAgentClient a
-  disconnectAgentClient b
+  disposeAgentClient a
+  disposeAgentClient b
 
 testFullSwitch :: AgentClient -> ByteString -> AgentClient -> ByteString -> Int64 -> ExceptT AgentErrorType IO ()
 testFullSwitch a bId b aId msgId = do
@@ -2077,7 +2077,7 @@ testSwitchAsync servers = do
     withB = withAgent 2 agentCfg servers testDB2
 
 withAgent :: Int -> AgentConfig -> InitialAgentServers -> FilePath -> (AgentClient -> IO a) -> IO a
-withAgent clientId cfg' servers dbPath = bracket (getSMPAgentClient' clientId cfg' servers dbPath) disconnectAgentClient
+withAgent clientId cfg' servers dbPath = bracket (getSMPAgentClient' clientId cfg' servers dbPath) disposeAgentClient
 
 sessionSubscribe :: (forall a. (AgentClient -> IO a) -> IO a) -> [ConnId] -> (AgentClient -> ExceptT AgentErrorType IO ()) -> IO ()
 sessionSubscribe withC connIds a =
@@ -2095,7 +2095,7 @@ testSwitchDelete servers = do
   runRight_ $ do
     (aId, bId) <- makeConnection a b
     exchangeGreetingsMsgId 4 a bId b aId
-    disconnectAgentClient b
+    disposeAgentClient b
     stats <- switchConnectionAsync a "" bId
     liftIO $ rcvSwchStatuses' stats `shouldMatchList` [Just RSSwitchStarted]
     phaseRcv a bId SPStarted [Just RSSendingQADD, Nothing]
@@ -2104,8 +2104,8 @@ testSwitchDelete servers = do
     get a =##> \case ("", c, DEL_RCVQ _ _ Nothing) -> c == bId; _ -> False
     get a =##> \case ("", c, DEL_CONN) -> c == bId; _ -> False
     liftIO $ noMessages a "nothing else should be delivered to alice"
-  disconnectAgentClient a
-  disconnectAgentClient b
+  disposeAgentClient a
+  disposeAgentClient b
 
 testAbortSwitchStarted :: HasCallStack => InitialAgentServers -> IO ()
 testAbortSwitchStarted servers = do
@@ -2393,8 +2393,8 @@ testCreateQueueAuth srvVersion clnt1 clnt2 = do
             get b ##> ("", aId, CON)
             exchangeGreetings a bId b aId
             pure 2
-  disconnectAgentClient a
-  disconnectAgentClient b
+  disposeAgentClient a
+  disposeAgentClient b
   pure r
   where
     getClient clientId (clntAuth, clntVersion) db =
@@ -2458,8 +2458,8 @@ testDeliveryReceiptsVersion t = do
       liftIO $ noMessages b "no delivery receipt (unsupported version)"
       pure (aId, bId)
 
-    disconnectAgentClient a
-    disconnectAgentClient b
+    disposeAgentClient a
+    disposeAgentClient b
     a' <- getSMPAgentClient' 3 agentCfg {smpAgentVRange = supportedSMPAgentVRange} initAgentServers testDB
     b' <- getSMPAgentClient' 4 agentCfg {smpAgentVRange = supportedSMPAgentVRange} initAgentServers testDB2
 
@@ -2487,8 +2487,8 @@ testDeliveryReceiptsVersion t = do
       ackMessage b' aId 12 $ Just ""
       get a' =##> \case ("", c, Rcvd 12) -> c == bId; _ -> False
       ackMessage a' bId 13 Nothing
-    disconnectAgentClient a'
-    disconnectAgentClient b'
+    disposeAgentClient a'
+    disposeAgentClient b'
 
 testDeliveryReceiptsConcurrent :: HasCallStack => ATransport -> IO ()
 testDeliveryReceiptsConcurrent t =
@@ -2626,7 +2626,7 @@ testServerMultipleIdentities =
     exchangeGreetings alice bobId bob aliceId
     -- this saves queue with second server identity
     Left (BROKER _ NETWORK) <- runExceptT $ joinConnection bob 1 True secondIdentityCReq "bob's connInfo" SMSubscribe
-    disconnectAgentClient bob
+    disposeAgentClient bob
     bob' <- liftIO $ getSMPAgentClient' 3 agentCfg initAgentServers testDB2
     subscribeConnection bob' aliceId
     exchangeGreetingsMsgId 6 alice bobId bob' aliceId

tests/AgentTests/NotificationTests.hs

@@ -179,7 +179,7 @@ testNotificationToken APNSMockServer {apnsQ} = do
     deleteNtfToken a tkn
     -- agent deleted this token
     Left (CMD PROHIBITED) <- tryE $ checkNtfToken a tkn
-    disconnectAgentClient a
+    disposeAgentClient a
 
 (.->) :: J.Value -> J.Key -> ExceptT AgentErrorType IO ByteString
 v .-> key = do
@@ -211,7 +211,7 @@ testNtfTokenRepeatRegistration APNSMockServer {apnsQ} = do
     -- can still use the first verification code, it is the same after decryption
     verifyNtfToken a tkn nonce verification
     NTActive <- checkNtfToken a tkn
-    disconnectAgentClient a
+    disposeAgentClient a
 
 testNtfTokenSecondRegistration :: APNSMockServer -> IO ()
 testNtfTokenSecondRegistration APNSMockServer {apnsQ} = do
@@ -247,8 +247,8 @@ testNtfTokenSecondRegistration APNSMockServer {apnsQ} = do
     Left (NTF AUTH) <- tryE $ checkNtfToken a tkn
     -- and the second is active
     NTActive <- checkNtfToken a' tkn
-    disconnectAgentClient a
-    disconnectAgentClient a'
+    disposeAgentClient a
+    disposeAgentClient a'
 
 testNtfTokenServerRestart :: ATransport -> APNSMockServer -> IO ()
 testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
@@ -262,7 +262,7 @@ testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
     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
   threadDelay 1000000
-  disconnectAgentClient a
+  disposeAgentClient a
   a' <- getSMPAgentClient' 2 agentCfg initAgentServers testDB
   -- server stopped before token is verified, so now the attempt to verify it will return AUTH error but re-register token,
   -- so that repeat verification happens without restarting the clients, when notification arrives
@@ -277,7 +277,7 @@ testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
     liftIO $ sendApnsResponse' APNSRespOk
     verifyNtfToken a' tkn nonce' verification'
     NTActive <- checkNtfToken a' tkn
-    disconnectAgentClient a'
+    disposeAgentClient a'
 
 getTestNtfTokenPort :: (MonadUnliftIO m, MonadError AgentErrorType m) => AgentClient -> m String
 getTestNtfTokenPort a =
@@ -319,7 +319,7 @@ testNtfTokenChangeServers t APNSMockServer {apnsQ} =
       NTActive <- checkNtfToken a tkn
       setNtfServers a [testNtfServer2]
       NTActive <- checkNtfToken a tkn -- still works on old server
-      disconnectAgentClient a
+      disposeAgentClient a
       pure tkn
 
     threadDelay 1000000
@@ -386,7 +386,7 @@ testNotificationSubscriptionExistingConnection APNSMockServer {apnsQ} alice@Agen
   runRight_ $ do
     (_, [SMPMsgMeta {msgFlags = MsgFlags True}]) <- getNotificationMessage aliceNtf nonce message
     pure ()
-  disconnectAgentClient aliceNtf
+  disposeAgentClient aliceNtf
 
   runRight_ $ do
     get alice =##> \case ("", c, Msg "hello") -> c == bobId; _ -> False
@@ -517,8 +517,8 @@ testChangeNotificationsMode APNSMockServer {apnsQ} = do
     ackMessage alice bobId (baseId + 5) Nothing
     -- no notifications should follow
     noNotification apnsQ
-  disconnectAgentClient alice
-  disconnectAgentClient bob
+  disposeAgentClient alice
+  disposeAgentClient bob
   where
     baseId = 3
     msgId = subtract baseId
@@ -546,7 +546,7 @@ testChangeToken APNSMockServer {apnsQ} = do
     get alice =##> \case ("", c, Msg "hello") -> c == bobId; _ -> False
     ackMessage alice bobId (baseId + 1) Nothing
     pure (aliceId, bobId)
-  disconnectAgentClient alice
+  disposeAgentClient alice
 
   alice1 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
   runRight_ $ do
@@ -562,8 +562,8 @@ testChangeToken APNSMockServer {apnsQ} = do
     ackMessage alice1 bobId (baseId + 2) Nothing
     -- no notifications should follow
     noNotification apnsQ
-  disconnectAgentClient alice1
-  disconnectAgentClient bob
+  disposeAgentClient alice1
+  disposeAgentClient bob
   where
     baseId = 3
     msgId = subtract baseId
@@ -593,8 +593,8 @@ testNotificationsStoreLog t APNSMockServer {apnsQ} = do
     void $ messageNotificationData alice apnsQ
     get alice =##> \case ("", c, Msg "hello again") -> c == bobId; _ -> False
     liftIO $ killThread threadId
-  disconnectAgentClient alice
-  disconnectAgentClient bob
+  disposeAgentClient alice
+  disposeAgentClient bob
 
 testNotificationsSMPRestart :: ATransport -> APNSMockServer -> IO ()
 testNotificationsSMPRestart t APNSMockServer {apnsQ} = do
@@ -625,8 +625,8 @@ testNotificationsSMPRestart t APNSMockServer {apnsQ} = do
     _ <- messageNotificationData alice apnsQ
     get alice =##> \case ("", c, Msg "hello again") -> c == bobId; _ -> False
     liftIO $ killThread threadId
-  disconnectAgentClient alice
-  disconnectAgentClient bob
+  disposeAgentClient alice
+  disposeAgentClient bob
 
 testNotificationsSMPRestartBatch :: Int -> ATransport -> APNSMockServer -> IO ()
 testNotificationsSMPRestartBatch n t APNSMockServer {apnsQ} = do
@@ -666,8 +666,8 @@ testNotificationsSMPRestartBatch n t APNSMockServer {apnsQ} = do
       get b ##> ("", aliceId, SENT msgId)
       _ <- messageNotificationData a apnsQ
       get a =##> \case ("", c, Msg "hello again") -> c == bobId; _ -> False
-  disconnectAgentClient a
-  disconnectAgentClient b
+  disposeAgentClient a
+  disposeAgentClient b
   where
     runServers :: ExceptT AgentErrorType IO a -> IO a
     runServers a = do
@@ -697,8 +697,8 @@ testSwitchNotifications servers APNSMockServer {apnsQ} = do
     switchComplete a bId b aId
     liftIO $ threadDelay 500000
     testMessage "hello again"
-  disconnectAgentClient a
-  disconnectAgentClient b
+  disposeAgentClient a
+  disposeAgentClient b
 
 testNotificationsOldToken :: APNSMockServer -> IO ()
 testNotificationsOldToken APNSMockServer {apnsQ} = do
@@ -721,9 +721,9 @@ testNotificationsOldToken APNSMockServer {apnsQ} = do
     (acId, caId) <- makeConnection a c
     let testMessageAC = testMessage_ apnsQ a acId c caId
     testMessageAC "greetings"
-  disconnectAgentClient a
-  disconnectAgentClient b
-  disconnectAgentClient c
+  disposeAgentClient a
+  disposeAgentClient b
+  disposeAgentClient c
 
 testNotificationsNewToken :: APNSMockServer -> ThreadId -> IO ()
 testNotificationsNewToken APNSMockServer {apnsQ} oldNtf = do
@@ -749,9 +749,9 @@ testNotificationsNewToken APNSMockServer {apnsQ} oldNtf = do
     (acId, caId) <- makeConnection a c
     let testMessageAC = testMessage_ apnsQ a acId c caId
     testMessageAC "greetings"
-  disconnectAgentClient a
-  disconnectAgentClient b
-  disconnectAgentClient c
+  disposeAgentClient a
+  disposeAgentClient b
+  disposeAgentClient c
 
 testMessage_ :: HasCallStack => TBQueue APNSMockRequest -> AgentClient -> ConnId -> AgentClient -> ConnId -> SMP.MsgBody -> ExceptT AgentErrorType IO ()
 testMessage_ apnsQ a aId b bId msg = do

tests/XFTPAgent.hs

@@ -24,7 +24,7 @@ import Simplex.FileTransfer.Description (FileDescription (..), FileDescriptionUR
 import Simplex.FileTransfer.Protocol (FileParty (..))
 import Simplex.FileTransfer.Transport (XFTPErrorType (AUTH))
 import Simplex.FileTransfer.Server.Env (XFTPServerConfig (..))
-import Simplex.Messaging.Agent (AgentClient, disconnectAgentClient, testProtocolServer, xftpDeleteRcvFile, xftpDeleteSndFileInternal, xftpDeleteSndFileRemote, xftpReceiveFile, xftpSendDescription, xftpSendFile, xftpStartWorkers)
+import Simplex.Messaging.Agent (AgentClient, disposeAgentClient, testProtocolServer, xftpDeleteRcvFile, xftpDeleteSndFileInternal, xftpDeleteSndFileRemote, xftpReceiveFile, xftpSendDescription, xftpSendFile, xftpStartWorkers)
 import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestStep (..))
 import Simplex.Messaging.Agent.Protocol (ACommand (..), AgentErrorType (..), BrokerErrorType (..), RcvFileId, SndFileId, noAuthSrv)
 import qualified Simplex.Messaging.Crypto as C
@@ -115,7 +115,7 @@ testXFTPAgentSendReceive = withXFTPServer $ do
       runRight_ $ do
         rfId <- testReceive rcp rfd originalFilePath
         xftpDeleteRcvFile rcp rfId
-      disconnectAgentClient rcp
+      disposeAgentClient rcp
 
 testXFTPAgentSendReceiveEncrypted :: HasCallStack => IO ()
 testXFTPAgentSendReceiveEncrypted = withXFTPServer $ do
@@ -139,7 +139,7 @@ testXFTPAgentSendReceiveEncrypted = withXFTPServer $ do
       runRight_ $ do
         rfId <- testReceiveCF rcp rfd cfArgs originalFilePath
         xftpDeleteRcvFile rcp rfId
-      disconnectAgentClient rcp
+      disposeAgentClient rcp
 
 testXFTPAgentSendReceiveRedirect :: HasCallStack => IO ()
 testXFTPAgentSendReceiveRedirect = withXFTPServer $ do
@@ -173,7 +173,7 @@ testXFTPAgentSendReceiveRedirect = withXFTPServer $ do
   case strDecode uri of
     Left err -> fail err
     Right ok -> ok `shouldBe` fileDescriptionURI vfdRedirect
-  disconnectAgentClient sndr
+  disposeAgentClient sndr
   --- recipient
   rcp <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   FileDescriptionURI {description} <- either fail pure $ strDecode uri
@@ -190,7 +190,7 @@ testXFTPAgentSendReceiveRedirect = withXFTPServer $ do
     rfGet rcp >>= \case
       (_, _, RFDONE out) -> pure out
       r -> error $ "Expected RFDONE, got " <> show r
-  disconnectAgentClient rcp
+  disposeAgentClient rcp
 
   inBytes <- B.readFile filePathIn
   B.readFile out `shouldReturn` inBytes
@@ -215,7 +215,7 @@ testXFTPAgentSendReceiveNoRedirect = withXFTPServer $ do
   case strDecode uri of
     Left err -> fail err
     Right ok -> ok `shouldBe` fileDescriptionURI vfdDirect
-  disconnectAgentClient sndr
+  disposeAgentClient sndr
   --- recipient
   rcp <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   FileDescriptionURI {description} <- either fail pure $ strDecode uri
@@ -230,7 +230,7 @@ testXFTPAgentSendReceiveNoRedirect = withXFTPServer $ do
     rfGet rcp >>= \case
       (_, _, RFDONE out) -> pure out
       r -> error $ "Expected RFDONE, got " <> show r
-  disconnectAgentClient rcp
+  disposeAgentClient rcp
 
   inBytes <- B.readFile filePathIn
   B.readFile out `shouldReturn` inBytes
@@ -303,7 +303,7 @@ testXFTPAgentReceiveRestore = withGlobalLogging logCfgNoLogs $ do
     rfId <- xftpReceiveFile rcp 1 rfd Nothing
     liftIO $ timeout 300000 (get rcp) `shouldReturn` Nothing -- wait for worker attempt
     pure rfId
-  disconnectAgentClient rcp
+  disposeAgentClient rcp
 
   [prefixDir] <- listDirectory recipientFiles
   let tmpPath = recipientFiles </> prefixDir </> "xftp.encrypted"
@@ -315,7 +315,7 @@ testXFTPAgentReceiveRestore = withGlobalLogging logCfgNoLogs $ do
     runRight_ $ xftpStartWorkers rcp' (Just recipientFiles)
     ("", rfId', RFPROG _ _) <- rfGet rcp'
     liftIO $ rfId' `shouldBe` rfId
-    disconnectAgentClient rcp'
+    disposeAgentClient rcp'
 
   threadDelay 100000
 
@@ -351,7 +351,7 @@ testXFTPAgentReceiveCleanup = withGlobalLogging logCfgNoLogs $ do
     rfId <- xftpReceiveFile rcp 1 rfd Nothing
     liftIO $ timeout 300000 (get rcp) `shouldReturn` Nothing -- wait for worker attempt
     pure rfId
-  disconnectAgentClient rcp
+  disposeAgentClient rcp
 
   [prefixDir] <- listDirectory recipientFiles
   let tmpPath = recipientFiles </> prefixDir </> "xftp.encrypted"
@@ -378,7 +378,7 @@ testXFTPAgentSendRestore = withGlobalLogging logCfgNoLogs $ do
     sfId <- xftpSendFile sndr 1 (CF.plain filePath) 2
     liftIO $ timeout 1000000 (get sndr) `shouldReturn` Nothing -- wait for worker to encrypt and attempt to create file
     pure sfId
-  disconnectAgentClient sndr
+  disposeAgentClient sndr
 
   dirEntries <- listDirectory senderFiles
   let prefixDir = fromJust $ find (isSuffixOf "_snd.xftp") dirEntries
@@ -393,7 +393,7 @@ testXFTPAgentSendRestore = withGlobalLogging logCfgNoLogs $ do
     runRight_ $ xftpStartWorkers sndr' (Just senderFiles)
     ("", sfId', SFPROG _ _) <- sfGet sndr'
     liftIO $ sfId' `shouldBe` sfId
-    disconnectAgentClient sndr'
+    disposeAgentClient sndr'
 
   threadDelay 100000
 
@@ -430,7 +430,7 @@ testXFTPAgentSendCleanup = withGlobalLogging logCfgNoLogs $ do
         (_, _, SFPROG _ _) <- sfGet sndr
         pure ()
       pure sfId
-    disconnectAgentClient sndr
+    disposeAgentClient sndr
     pure sfId
 
   dirEntries <- listDirectory senderFiles
@@ -473,7 +473,7 @@ testXFTPAgentDelete = withGlobalLogging logCfgNoLogs $
       xftpDeleteSndFileRemote sndr 1 sfId sndDescr
       Nothing <- liftIO $ 100000 `timeout` sfGet sndr
       pure ()
-    disconnectAgentClient rcp1
+    disposeAgentClient rcp1
 
     threadDelay 1000000
     length <$> listDirectory xftpServerFiles `shouldReturn` 0
@@ -499,8 +499,8 @@ testXFTPAgentDeleteRestore = withGlobalLogging logCfgNoLogs $ do
     rcp1 <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
     runRight_ . void $
       testReceive rcp1 rfd1 filePath
-    disconnectAgentClient rcp1
-    disconnectAgentClient sndr
+    disposeAgentClient rcp1
+    disposeAgentClient sndr
     pure (sfId, sndDescr, rfd2)
 
   -- delete file - should not succeed with server down
@@ -509,7 +509,7 @@ testXFTPAgentDeleteRestore = withGlobalLogging logCfgNoLogs $ do
     xftpStartWorkers sndr (Just senderFiles)
     xftpDeleteSndFileRemote sndr 1 sfId sndDescr
     liftIO $ timeout 300000 (get sndr) `shouldReturn` Nothing -- wait for worker attempt
-  disconnectAgentClient sndr
+  disposeAgentClient sndr
 
   threadDelay 300000
   length <$> listDirectory xftpServerFiles `shouldReturn` 6

tests/XFTPClient.hs

@@ -106,6 +106,7 @@ testXFTPServerConfig =
       allowNewFiles = True,
       newFileBasicAuth = Nothing,
       fileExpiration = Just defaultFileExpiration,
+      fileTimeout = 10000000,
       inactiveClientExpiration = Just defaultInactiveClientExpiration,
       caCertificateFile = "tests/fixtures/ca.crt",
       privateKeyFile = "tests/fixtures/server.key",