5.5.0.6

* agent: fix potential race when good client can be removed instead of bad for the same transport session

* tryAgentError

* case

CHANGELOG.md

@@ -1,3 +1,27 @@
+# 5.5.0
+
+Code:
+- compatible with GHC 8.10.7 to support compilation for armv7a.
+- migrate to `crypton` from deprecated `cryptonite` (the seed for DRG is now sha512-hashed).
+- use ChaChaDRG for all random IDs, keys and nonces, only using hashed entropy as seed.
+- more efficient transaction batching in SMP protocol client and server.
+
+Agent:
+- stabilize message reception and delivery, migrate message delivery to database queue.
+- additional event MSGNTF confirming that message received via notification is processed.
+- efficient processing of messages sent to multiple recipients with batched database transactions.
+- new worker abstraction for all queued tasks resilient to race conditions and some database errors.
+- many fixed race conditions.
+- background mode for iOS NSE.
+- additional error reporting to client on critical errors (to be show as alert in the clients).
+- functional api to get worker statistics.
+
+SMP/XFTP servers:
+- fix socket and memory leak on servers with high load (inactive clients without subscriptions are disconnected after set time of inactivity).
+- control port improvements.
+- fix statistics for stored queues, messages and files.
+- make writing to store log atomic (fixes a rare bug in XFTP server).
+
 # 5.4.0
 
 Migrate to GHC 9.6.3

package.yaml

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

protocol/xftp-uri-request.txt

@@ -0,0 +1,16 @@
+Scheme name: xftp
+
+Status: Provisional
+
+Applications/protocols that use this scheme name:
+This scheme is used for URIs of XFTP (SimpleX File Transfer Protocol) servers,
+a client-server protocol for asynchronous file transfer via relays,
+preserving file meta-data (including size and name) and content privacy.
+
+Contact: Evgeny Poberezkin <ep@simplex.chat>
+
+Change controller: Evgeny Poberezkin <ep@simplex.chat>
+
+References:
+The syntax for server URIs in the provisional specification for SimpleX File Transfer Protocol:
+https://github.com/simplex-chat/simplexmq/blob/master/rfcs/2022-12-26-simplex-file-transfer.md#server-address-syntax

protocol/xrcp-uri-request.txt

@@ -0,0 +1,15 @@
+Scheme name: xrcp
+
+Status: Provisional
+
+Applications/protocols that use this scheme name:
+This scheme is used for URIs of controller sessions via SimpleX Remote Control protocol (XRCP),
+a protocol for remote access and management of hosts via insecure network.
+
+Contact: Evgeny Poberezkin <ep@simplex.chat>
+
+Change controller: Evgeny Poberezkin <ep@simplex.chat>
+
+References:
+The syntax for server URIs in the provisional specification for SimpleX File Transfer Protocol:
+https://github.com/simplex-chat/simplexmq/blob/master/rfcs/2023-10-25-remote-control.md#session-invitation

rfcs/2022-12-26-simplex-file-transfer.md

@@ -149,6 +149,19 @@ parts:
 
 This file description is sent to all recipients via normal messages, split to 15780 byte chunks if needed.
 
+### Server address syntax
+
+The server address is a URI with the following format:
+
+```abnf
+xftpServerURI = %s"xftp://" xftpServer
+xftpServer = serverIdentity "@" srvHost [":" port]
+srvHost = <hostname> ; RFC1123, RFC5891
+port = 1*DIGIT
+serverIdentity = base64url
+base64url = <base64url encoded binary> ; RFC4648, section 5
+```
+
 ### Receiving file
 
 Having received the description, the recipient will:

simplexmq.cabal

@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack
 
 name:           simplexmq
-version:        5.5.0.4
+version:        5.5.0.6
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -103,7 +103,6 @@ library
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20231225_failed_work_items
       Simplex.Messaging.Agent.TAsyncs
       Simplex.Messaging.Agent.TRcvQueues
-      Simplex.Messaging.Builder
       Simplex.Messaging.Client
       Simplex.Messaging.Client.Agent
       Simplex.Messaging.Crypto

src/Simplex/FileTransfer/Client.hs

@@ -13,7 +13,7 @@ import Control.Monad
 import Control.Monad.Except
 import Crypto.Random (ChaChaDRG)
 import Data.Bifunctor (first)
-import qualified Data.ByteString.Builder as BB
+import Data.ByteString.Builder (Builder, byteString)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Int (Int64)
@@ -25,7 +25,6 @@ import qualified Network.HTTP2.Client as H
 import Simplex.FileTransfer.Description (mb)
 import Simplex.FileTransfer.Protocol
 import Simplex.FileTransfer.Transport
-import Simplex.Messaging.Builder (Builder, builder)
 import Simplex.Messaging.Client
   ( NetworkConfig (..),
     ProtocolClientError (..),
@@ -139,7 +138,7 @@ sendXFTPCommand c@XFTPClient {http2Client = HTTP2Client {sessionId}} pKey fId cm
       xftpEncodeTransmission sessionId (Just pKey) ("", fId, FileCmd (sFileParty @p) cmd)
   sendXFTPTransmission c t chunkSpec_
 
-sendXFTPTransmission :: XFTPClient -> Builder -> Maybe XFTPChunkSpec -> ExceptT XFTPClientError IO (FileResponse, HTTP2Body)
+sendXFTPTransmission :: XFTPClient -> ByteString -> Maybe XFTPChunkSpec -> ExceptT XFTPClientError IO (FileResponse, HTTP2Body)
 sendXFTPTransmission XFTPClient {config, http2Client = http2@HTTP2Client {sessionId}} t chunkSpec_ = do
   let req = H.requestStreaming N.methodPost "/" [] streamBody
       reqTimeout = (\XFTPChunkSpec {chunkSize} -> chunkTimeout config chunkSize) <$> chunkSpec_
@@ -153,9 +152,9 @@ sendXFTPTransmission XFTPClient {config, http2Client = http2@HTTP2Client {sessio
       _ -> pure (r, body)
     Left e -> throwError $ PCEResponseError e
   where
-    streamBody :: (BB.Builder -> IO ()) -> IO () -> IO ()
+    streamBody :: (Builder -> IO ()) -> IO () -> IO ()
     streamBody send done = do
-      send $ builder t
+      send $ byteString t
       forM_ chunkSpec_ $ \XFTPChunkSpec {filePath, chunkOffset, chunkSize} ->
         withFile filePath ReadMode $ \h -> do
           hSeek h AbsoluteSeek $ fromIntegral chunkOffset

src/Simplex/FileTransfer/Protocol.hs

@@ -24,7 +24,6 @@ import Data.List.NonEmpty (NonEmpty (..))
 import Data.Maybe (isNothing)
 import Data.Type.Equality
 import Data.Word (Word32)
-import Simplex.Messaging.Builder (Builder)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
@@ -49,8 +48,7 @@ import Simplex.Messaging.Protocol
     encodeTransmission,
     messageTagP,
     tDecodeParseValidate,
-    tEncode,
-    tEncodeBatch,
+    tEncodeBatch1,
     tParse,
   )
 import Simplex.Messaging.Transport (SessionId, TransportError (..))
@@ -395,7 +393,7 @@ checkParty' c = case testEquality (sFileParty @p) (sFileParty @p') of
   Just Refl -> Just c
   _ -> Nothing
 
-xftpEncodeTransmission :: ProtocolEncoding e c => SessionId -> Maybe C.APrivateSignKey -> Transmission c -> Either TransportError Builder
+xftpEncodeTransmission :: ProtocolEncoding e c => SessionId -> Maybe C.APrivateSignKey -> Transmission c -> Either TransportError ByteString
 xftpEncodeTransmission sessionId pKey (corrId, fId, msg) = do
   let t = encodeTransmission currentXFTPVersion sessionId (corrId, fId, msg)
   xftpEncodeBatch1 $ signTransmission t
@@ -404,10 +402,8 @@ xftpEncodeTransmission sessionId pKey (corrId, fId, msg) = do
     signTransmission t = ((`C.sign` t) <$> pKey, t)
 
 -- this function uses batch syntax but puts only one transmission in the batch
-xftpEncodeBatch1 :: (Maybe C.ASignature, ByteString) -> Either TransportError Builder
-xftpEncodeBatch1 (sig, t) =
-  let t' = tEncodeBatch 1 . encodeLarge $ tEncode (sig, t)
-   in first (const TELargeMsg) $ C.pad' t' xftpBlockSize
+xftpEncodeBatch1 :: SentRawTransmission -> Either TransportError ByteString
+xftpEncodeBatch1 t = first (const TELargeMsg) $ C.pad (tEncodeBatch1 t) xftpBlockSize
 
 xftpDecodeTransmission :: ProtocolEncoding e c => SessionId -> ByteString -> Either XFTPErrorType (SignedTransmission e c)
 xftpDecodeTransmission sessionId t = do

src/Simplex/FileTransfer/Server.hs

@@ -18,6 +18,7 @@ import Control.Monad.Except
 import Control.Monad.Reader
 import Data.Bifunctor (first)
 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 (($>))
@@ -43,7 +44,6 @@ import Simplex.FileTransfer.Server.Stats
 import Simplex.FileTransfer.Server.Store
 import Simplex.FileTransfer.Server.StoreLog
 import Simplex.FileTransfer.Transport
-import Simplex.Messaging.Builder (builder)
 import qualified Simplex.Messaging.Crypto as C
 import qualified Simplex.Messaging.Crypto.Lazy as LC
 import Simplex.Messaging.Encoding.String
@@ -242,7 +242,7 @@ processRequest HTTP2Request {sessionId, reqBody = body@HTTP2Body {bodyHead}, sen
               send "padding error" -- TODO respond with BLOCK error?
               done
             Right t -> do
-              send $ builder t
+              send $ byteString t
               -- timeout sending file in the same way as receiving
               forM_ serverFile_ $ \ServerFile {filePath, fileSize, sbState} -> do
                 withFile filePath ReadMode $ \h -> sendEncFile h send sbState (fromIntegral fileSize)

src/Simplex/FileTransfer/Server/Main.hs

@@ -33,7 +33,7 @@ import System.IO (BufferMode (..), hSetBuffering, stderr, stdout)
 import Text.Read (readMaybe)
 
 xftpServerVersion :: String
-xftpServerVersion = "1.2.0.4"
+xftpServerVersion = "1.2.0.6"
 
 xftpServerCLI :: FilePath -> FilePath -> IO ()
 xftpServerCLI cfgPath logPath = do

src/Simplex/Messaging/Agent.hs

@@ -38,6 +38,7 @@ module Simplex.Messaging.Agent
     AgentErrorMonad,
     SubscriptionsInfo (..),
     getSMPAgentClient,
+    getSMPAgentClient_,
     disconnectAgentClient,
     resumeAgentClient,
     withConnLock,
@@ -176,11 +177,15 @@ import UnliftIO.STM
 
 -- | Creates an SMP agent client instance
 getSMPAgentClient :: (MonadRandom m, MonadUnliftIO m) => AgentConfig -> InitialAgentServers -> SQLiteStore -> Bool -> m AgentClient
-getSMPAgentClient cfg initServers store backgroundMode =
+getSMPAgentClient = getSMPAgentClient_ 1
+{-# INLINE getSMPAgentClient #-}
+
+getSMPAgentClient_ :: (MonadRandom m, MonadUnliftIO m) => Int -> AgentConfig -> InitialAgentServers -> SQLiteStore -> Bool -> m AgentClient
+getSMPAgentClient_ clientId cfg initServers store backgroundMode =
   liftIO (newSMPAgentEnv cfg store) >>= runReaderT runAgent
   where
     runAgent = do
-      c <- getAgentClient initServers
+      c <- getAgentClient clientId initServers
       void $ runAgentThreads c `forkFinally` const (disconnectAgentClient c)
       pure c
     runAgentThreads c
@@ -461,8 +466,9 @@ withAgentEnv :: AgentClient -> ReaderT Env m a -> m a
 withAgentEnv c = (`runReaderT` agentEnv c)
 
 -- | Creates an SMP agent client instance that receives commands and sends responses via 'TBQueue's.
-getAgentClient :: AgentMonad' m => InitialAgentServers -> m AgentClient
-getAgentClient initServers = ask >>= atomically . newAgentClient initServers
+getAgentClient :: AgentMonad' m => Int -> InitialAgentServers -> m AgentClient
+getAgentClient clientId initServers = ask >>= atomically . newAgentClient clientId initServers
+{-# INLINE getAgentClient #-}
 
 logConnection :: MonadUnliftIO m => AgentClient -> Bool -> m ()
 logConnection c connected =
@@ -1792,12 +1798,11 @@ getAgentMigrations' :: AgentMonad m => AgentClient -> m [UpMigration]
 getAgentMigrations' c = map upMigration <$> withStore' c (Migrations.getCurrent . DB.conn)
 
 debugAgentLocks' :: AgentMonad' m => AgentClient -> m AgentLocks
-debugAgentLocks' AgentClient {connLocks = cs, invLocks = is, reconnectLocks = rs, deleteLock = d} = do
+debugAgentLocks' AgentClient {connLocks = cs, invLocks = is, deleteLock = d} = do
   connLocks <- getLocks cs
   invLocks <- getLocks is
-  srvLocks <- getLocks rs
   delLock <- atomically $ tryReadTMVar d
-  pure AgentLocks {connLocks, invLocks, srvLocks, delLock}
+  pure AgentLocks {connLocks, invLocks, delLock}
   where
     getLocks ls = atomically $ M.mapKeys (B.unpack . strEncode) . M.mapMaybe id <$> (mapM tryReadTMVar =<< readTVar ls)
 
@@ -2042,7 +2047,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
               handleNotifyAck :: m () -> m ()
               handleNotifyAck m = m `catchAgentError` \e -> notify (ERR e) >> ack
           SMP.END ->
-            atomically (TM.lookup tSess smpClients $>>= tryReadTMVar >>= processEND)
+            atomically (TM.lookup tSess smpClients $>>= (tryReadTMVar . sessionVar) >>= processEND)
               >>= logServer "<--" c srv rId
             where
               processEND = \case

src/Simplex/Messaging/Agent/Client.hs

@@ -78,6 +78,7 @@ module Simplex.Messaging.Agent.Client
     agentDRG,
     getAgentSubscriptions,
     Worker (..),
+    SessionVar (..),
     SubscriptionsInfo (..),
     SubInfo (..),
     AgentOperation (..),
@@ -116,6 +117,10 @@ module Simplex.Messaging.Agent.Client
     getNextServer,
     withUserServers,
     withNextSrv,
+    AgentWorkersDetails (..),
+    getAgentWorkersDetails,
+    AgentWorkersSummary (..),
+    getAgentWorkersSummary,
   )
 where
 
@@ -150,6 +155,7 @@ import Data.Text.Encoding
 import Data.Time (UTCTime, defaultTimeLocale, formatTime, getCurrentTime)
 import Data.Time.Clock.System (getSystemTime)
 import Data.Word (Word16)
+-- import GHC.Conc (unsafeIOToSTM)
 import Network.Socket (HostName)
 import Simplex.FileTransfer.Client (XFTPChunkSpec (..), XFTPClient, XFTPClientConfig (..), XFTPClientError)
 import qualified Simplex.FileTransfer.Client as X
@@ -217,13 +223,18 @@ import UnliftIO.Exception (bracket)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
 
-type ClientVar msg = TMVar (Either AgentErrorType (Client msg))
+data SessionVar a = SessionVar
+  { sessionVar :: TMVar a,
+    sessionVarId :: Int
+  }
+
+type ClientVar msg = SessionVar (Either AgentErrorType (Client msg)) 
 
 type SMPClientVar = ClientVar SMP.BrokerMsg
 
 type NtfClientVar = ClientVar NtfResponse
 
-type XFTPClientVar = TMVar (Either AgentErrorType XFTPClient)
+type XFTPClientVar = ClientVar FileResponse
 
 type SMPTransportSession = TransportSession SMP.BrokerMsg
 
@@ -264,9 +275,8 @@ data AgentClient = AgentClient
     invLocks :: TMap ByteString Lock,
     -- lock to prevent concurrency between periodic and async connection deletions
     deleteLock :: Lock,
-    -- locks to prevent concurrent reconnections to SMP servers
-    reconnectLocks :: TMap SMPTransportSession Lock,
-    reconnections :: TAsyncs,
+    -- smpSubWorkers for SMP servers sessions
+    smpSubWorkers :: TMap SMPTransportSession (SessionVar (Async ())),
     asyncClients :: TAsyncs,
     agentStats :: TMap AgentStatsKey (TVar Int),
     clientId :: Int,
@@ -288,11 +298,11 @@ getAgentWorker' toW fromW name hasWork c key ws work = do
     whenExists w
       | hasWork = hasWorkToDo (toW w) $> w
       | otherwise = pure w
-    runWorker w = runWorkerAsync (toW w) . void $ runExceptT runWork
+    runWorker w = runWorkerAsync (toW w) runWork
       where
-        runWork :: ExceptT AgentErrorType m ()
-        runWork = tryAgentError (work w) >>= restartOrDelete
-        restartOrDelete :: Either AgentErrorType () -> ExceptT AgentErrorType m ()
+        runWork :: m ()
+        runWork = tryAgentError' (work w) >>= restartOrDelete
+        restartOrDelete :: Either AgentErrorType () -> m ()
         restartOrDelete e_ = do
           t <- liftIO getSystemTime
           maxRestarts <- asks $ maxWorkerRestartsPerMin . config
@@ -361,7 +371,6 @@ data AgentState = ASForeground | ASSuspending | ASSuspended
 data AgentLocks = AgentLocks
   { connLocks :: Map String String,
     invLocks :: Map String String,
-    srvLocks :: Map String String,
     delLock :: Maybe String
   }
   deriving (Show)
@@ -375,8 +384,8 @@ data AgentStatsKey = AgentStatsKey
   }
   deriving (Eq, Ord, Show)
 
-newAgentClient :: InitialAgentServers -> Env -> STM AgentClient
-newAgentClient InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv = do
+newAgentClient :: Int -> InitialAgentServers -> Env -> STM AgentClient
+newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv = do
   let qSize = tbqSize $ config agentEnv
   active <- newTVar True
   rcvQ <- newTBQueue qSize
@@ -407,11 +416,9 @@ newAgentClient InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv = do
   connLocks <- TM.empty
   invLocks <- TM.empty
   deleteLock <- createLock
-  reconnectLocks <- TM.empty
-  reconnections <- newTAsyncs
+  smpSubWorkers <- TM.empty
   asyncClients <- newTAsyncs
   agentStats <- TM.empty
-  clientId <- stateTVar (clientCounter agentEnv) $ \i -> let i' = i + 1 in (i', i')
   return
     AgentClient
       { active,
@@ -443,8 +450,7 @@ newAgentClient InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv = do
         connLocks,
         invLocks,
         deleteLock,
-        reconnectLocks,
-        reconnections,
+        smpSubWorkers,
         asyncClients,
         agentStats,
         clientId,
@@ -496,26 +502,24 @@ instance ProtocolServerClient XFTPErrorType FileResponse where
 getSMPServerClient :: forall m. AgentMonad m => AgentClient -> SMPTransportSession -> m SMPClient
 getSMPServerClient c@AgentClient {active, smpClients, msgQ} tSess@(userId, srv, _) = do
   unlessM (readTVarIO active) . throwError $ INACTIVE
-  atomically (getClientVar tSess smpClients)
+  atomically (getTSessVar c tSess smpClients)
     >>= either newClient (waitForProtocolClient c tSess)
   where
-    newClient v = do
-      tc <- newTVarIO 0
-      newProtocolClient c tSess smpClients connectClient (reconnectSMPClient 0 tc) v
-    connectClient :: m SMPClient
-    connectClient = do
+    newClient = newProtocolClient c tSess smpClients connectClient resubscribeSMPSession
+    connectClient :: SMPClientVar -> m SMPClient
+    connectClient v = do
       cfg <- getClientConfig c smpCfg
       u <- askUnliftIO
-      liftEitherError (protocolClientError SMP $ B.unpack $ strEncode srv) (getProtocolClient tSess cfg (Just msgQ) $ clientDisconnected u)
+      liftEitherError (protocolClientError SMP $ B.unpack $ strEncode srv) (getProtocolClient tSess cfg (Just msgQ) $ clientDisconnected u v)
 
-    clientDisconnected :: UnliftIO m -> SMPClient -> IO ()
-    clientDisconnected u client = do
+    clientDisconnected :: UnliftIO m -> SMPClientVar -> SMPClient -> IO ()
+    clientDisconnected u v client = do
       removeClientAndSubs >>= serverDown
       logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv
       where
         removeClientAndSubs :: IO ([RcvQueue], [ConnId])
         removeClientAndSubs = atomically $ do
-          TM.delete tSess smpClients
+          removeTSessVar v tSess smpClients
           qs <- RQ.getDelSessQueues tSess $ activeSubs c
           mapM_ (`RQ.addQueue` pendingSubs c) qs
           let cs = S.fromList $ map qConnId qs
@@ -529,41 +533,49 @@ 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 $ reconnectServer c tSess
+            unliftIO u $ resubscribeSMPSession c tSess
 
         notifySub :: forall e. AEntityI e => ConnId -> ACommand 'Agent e -> IO ()
         notifySub connId cmd = atomically $ writeTBQueue (subQ c) ("", connId, APC (sAEntity @e) cmd)
 
-reconnectServer :: AgentMonad m => AgentClient -> SMPTransportSession -> m ()
-reconnectServer c tSess = newAsyncAction tryReconnectSMPClient $ reconnections c
+resubscribeSMPSession :: AgentMonad' m => AgentClient -> SMPTransportSession -> m ()
+resubscribeSMPSession c@AgentClient {smpSubWorkers} tSess =
+  atomically (getTSessVar c tSess smpSubWorkers) >>= either newSubWorker (\_ -> pure ())
   where
-    tryReconnectSMPClient aId = do
+    newSubWorker v = do
+      a <- async $ void (E.tryAny runSubWorker) >> atomically (cleanup v)
+      atomically $ putTMVar (sessionVar v) a
+    runSubWorker = do
       ri <- asks $ reconnectInterval . config
       timeoutCounts <- newTVarIO 0
-      withRetryIntervalCount ri $ \n _ loop ->
-        reconnectSMPClient n timeoutCounts c tSess `catchAgentError` const loop
-      atomically . removeAsyncAction aId $ reconnections c
+      withRetryInterval ri $ \_ loop -> do
+        pending <- atomically . RQ.getSessQueues tSess $ pendingSubs c
+        forM_ (L.nonEmpty pending) $ \qs -> do
+          void . tryAgentError' $ reconnectSMPClient timeoutCounts c tSess qs
+          loop
+    cleanup :: SessionVar (Async ()) -> STM ()
+    cleanup v = do
+      -- Here we wait until TMVar is not empty to prevent worker cleanup happening before worker is added to TMVar.
+      -- Not waiting may result in terminated worker remaining in the map.
+      whenM (isEmptyTMVar $ sessionVar v) retry
+      removeTSessVar v tSess smpSubWorkers
 
-reconnectSMPClient :: forall m. AgentMonad m => Int -> TVar Int -> AgentClient -> SMPTransportSession -> m ()
-reconnectSMPClient n tc c tSess@(_, srv, _) = do
-  ts <- liftIO getCurrentTime
-  let label = unwords ["reconnect", show n, show ts]
-  withLockMap_ (reconnectLocks c) tSess label $ do
-    qs <- atomically (RQ.getSessQueues tSess $ pendingSubs c)
-    NetworkConfig {tcpTimeout} <- readTVarIO $ useNetworkConfig c
-    -- this allows 3x of timeout per batch of subscription (90 queues per batch empirically)
-    let t = (length qs `div` 90 + 1) * tcpTimeout * 3
-    t `timeout` mapM_ resubscribe (L.nonEmpty qs) >>= \case
-      Just _ -> atomically $ writeTVar tc 0
-      Nothing -> do
-        tc' <- atomically $ stateTVar tc $ \i -> (i + 1, i + 1)
-        maxTC <- asks $ maxSubscriptionTimeouts . config
-        let err = if tc' >= maxTC then CRITICAL True else INTERNAL
-            msg = show tc' <> " consecutive subscription timeouts: " <> show (length qs) <> " queues, transport session: " <> show tSess
-        atomically $ writeTBQueue (subQ c) ("", "", APC SAEConn $ ERR $ err msg)
+reconnectSMPClient :: forall m. AgentMonad m => TVar Int -> AgentClient -> SMPTransportSession -> NonEmpty RcvQueue -> m ()
+reconnectSMPClient tc c tSess@(_, srv, _) qs = do
+  NetworkConfig {tcpTimeout} <- readTVarIO $ useNetworkConfig c
+  -- this allows 3x of timeout per batch of subscription (90 queues per batch empirically)
+  let t = (length qs `div` 90 + 1) * tcpTimeout * 3
+  t `timeout` resubscribe >>= \case
+    Just _ -> atomically $ writeTVar tc 0
+    Nothing -> do
+      tc' <- atomically $ stateTVar tc $ \i -> (i + 1, i + 1)
+      maxTC <- asks $ maxSubscriptionTimeouts . config
+      let err = if tc' >= maxTC then CRITICAL True else INTERNAL
+          msg = show tc' <> " consecutive subscription timeouts: " <> show (length qs) <> " queues, transport session: " <> show tSess
+      atomically $ writeTBQueue (subQ c) ("", "", APC SAEConn $ ERR $ err msg)
   where
-    resubscribe :: NonEmpty RcvQueue -> m ()
-    resubscribe qs = do
+    resubscribe :: m ()
+    resubscribe = do
       cs <- atomically . RQ.getConns $ activeSubs c
       rs <- subscribeQueues c $ L.toList qs
       let (errs, okConns) = partitionEithers $ map (\(RcvQueue {connId}, r) -> bimap (connId,) (const connId) r) rs
@@ -572,26 +584,29 @@ reconnectSMPClient n tc c tSess@(_, srv, _) = do
         unless (null conns) $ notifySub "" $ UP srv conns
       let (tempErrs, finalErrs) = partition (temporaryAgentError . snd) errs
       liftIO $ mapM_ (\(connId, e) -> notifySub connId $ ERR e) finalErrs
-      mapM_ (throwError . snd) $ listToMaybe tempErrs
+      forM_ (listToMaybe tempErrs) $ \(_, err) -> do
+        when (null okConns && S.null cs && null finalErrs) . liftIO $
+          closeClient c smpClients tSess
+        throwError err
     notifySub :: forall e. AEntityI e => ConnId -> ACommand 'Agent e -> IO ()
     notifySub connId cmd = atomically $ writeTBQueue (subQ c) ("", connId, APC (sAEntity @e) cmd)
 
 getNtfServerClient :: forall m. AgentMonad m => AgentClient -> NtfTransportSession -> m NtfClient
 getNtfServerClient c@AgentClient {active, ntfClients} tSess@(userId, srv, _) = do
   unlessM (readTVarIO active) . throwError $ INACTIVE
-  atomically (getClientVar tSess ntfClients)
+  atomically (getTSessVar c tSess ntfClients)
     >>= either
       (newProtocolClient c tSess ntfClients connectClient $ \_ _ -> pure ())
       (waitForProtocolClient c tSess)
   where
-    connectClient :: m NtfClient
-    connectClient = do
+    connectClient :: NtfClientVar -> m NtfClient
+    connectClient v = do
       cfg <- getClientConfig c ntfCfg
-      liftEitherError (protocolClientError NTF $ B.unpack $ strEncode srv) (getProtocolClient tSess cfg Nothing clientDisconnected)
+      liftEitherError (protocolClientError NTF $ B.unpack $ strEncode srv) (getProtocolClient tSess cfg Nothing $ clientDisconnected v)
 
-    clientDisconnected :: NtfClient -> IO ()
-    clientDisconnected client = do
-      atomically $ TM.delete tSess ntfClients
+    clientDisconnected :: NtfClientVar -> NtfClient -> IO ()
+    clientDisconnected v client = do
+      atomically $ removeTSessVar v tSess ntfClients
       incClientStat c userId client "DISCONNECT" ""
       atomically $ writeTBQueue (subQ c) ("", "", APC SAENone $ hostEvent DISCONNECT client)
       logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv
@@ -599,60 +614,68 @@ getNtfServerClient c@AgentClient {active, ntfClients} tSess@(userId, srv, _) = d
 getXFTPServerClient :: forall m. AgentMonad m => AgentClient -> XFTPTransportSession -> m XFTPClient
 getXFTPServerClient c@AgentClient {active, xftpClients, useNetworkConfig} tSess@(userId, srv, _) = do
   unlessM (readTVarIO active) . throwError $ INACTIVE
-  atomically (getClientVar tSess xftpClients)
+  atomically (getTSessVar c tSess xftpClients)
     >>= either
       (newProtocolClient c tSess xftpClients connectClient $ \_ _ -> pure ())
       (waitForProtocolClient c tSess)
   where
-    connectClient :: m XFTPClient
-    connectClient = do
+    connectClient :: XFTPClientVar -> m XFTPClient
+    connectClient v = do
       cfg <- asks $ xftpCfg . config
       xftpNetworkConfig <- readTVarIO useNetworkConfig
-      liftEitherError (protocolClientError XFTP $ B.unpack $ strEncode srv) (X.getXFTPClient tSess cfg {xftpNetworkConfig} clientDisconnected)
+      liftEitherError (protocolClientError XFTP $ B.unpack $ strEncode srv) (X.getXFTPClient tSess cfg {xftpNetworkConfig} $ clientDisconnected v)
 
-    clientDisconnected :: XFTPClient -> IO ()
-    clientDisconnected client = do
-      atomically $ TM.delete tSess xftpClients
+    clientDisconnected :: XFTPClientVar -> XFTPClient -> IO ()
+    clientDisconnected v client = do
+      atomically $ removeTSessVar v tSess xftpClients
       incClientStat c userId client "DISCONNECT" ""
       atomically $ writeTBQueue (subQ c) ("", "", APC SAENone $ hostEvent DISCONNECT client)
       logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv
 
-getClientVar :: forall a s. TransportSession s -> TMap (TransportSession s) (TMVar a) -> STM (Either (TMVar a) (TMVar a))
-getClientVar tSess clients = maybe (Left <$> newClientVar) (pure . Right) =<< TM.lookup tSess clients
+getTSessVar :: forall a s. AgentClient -> TransportSession s -> TMap (TransportSession s) (SessionVar a) -> STM (Either (SessionVar a) (SessionVar a))
+getTSessVar c tSess vs = maybe (Left <$> newSessionVar) (pure . Right) =<< TM.lookup tSess vs
   where
-    newClientVar :: STM (TMVar a)
-    newClientVar = do
-      var <- newEmptyTMVar
-      TM.insert tSess var clients
-      pure var
+    newSessionVar :: STM (SessionVar a)
+    newSessionVar = do
+      sessionVar <- newEmptyTMVar
+      sessionVarId <- stateTVar (workerSeq c) $ \next -> (next, next + 1)
+      let v = SessionVar {sessionVar, sessionVarId}
+      TM.insert tSess v vs
+      pure v
+
+removeTSessVar :: SessionVar a -> TransportSession msg -> TMap (TransportSession msg) (SessionVar a) -> STM ()
+removeTSessVar v tSess vs =
+  TM.lookup tSess vs
+    >>= mapM_ (\v' -> when (sessionVarId v == sessionVarId v') $ TM.delete tSess vs)
 
 waitForProtocolClient :: (AgentMonad m, ProtocolTypeI (ProtoType msg)) => AgentClient -> TransportSession msg -> ClientVar msg -> m (Client msg)
-waitForProtocolClient c (_, srv, _) clientVar = do
+waitForProtocolClient c (_, srv, _) v = do
   NetworkConfig {tcpConnectTimeout} <- readTVarIO $ useNetworkConfig c
-  client_ <- liftIO $ tcpConnectTimeout `timeout` atomically (readTMVar clientVar)
+  client_ <- liftIO $ tcpConnectTimeout `timeout` atomically (readTMVar $ sessionVar v)
   liftEither $ case client_ of
     Just (Right smpClient) -> Right smpClient
     Just (Left e) -> Left e
     Nothing -> Left $ BROKER (B.unpack $ strEncode srv) TIMEOUT
 
+-- clientConnected arg is only passed for SMP server
 newProtocolClient ::
   forall err msg m.
   (AgentMonad m, ProtocolTypeI (ProtoType msg), ProtocolServerClient err msg) =>
   AgentClient ->
   TransportSession msg ->
   TMap (TransportSession msg) (ClientVar msg) ->
-  m (Client msg) ->
+  (ClientVar msg -> m (Client msg)) ->
   (AgentClient -> TransportSession msg -> m ()) ->
   ClientVar msg ->
   m (Client msg)
-newProtocolClient c tSess@(userId, srv, entityId_) clients connectClient reconnectClient clientVar = tryConnectClient pure tryConnectAsync
+newProtocolClient c tSess@(userId, srv, entityId_) clients connectClient clientConnected v = tryConnectClient pure tryConnectAsync
   where
     tryConnectClient :: (Client msg -> m a) -> m () -> m a
     tryConnectClient successAction retryAction =
-      tryError connectClient >>= \r -> case r of
+      tryAgentError (connectClient v) >>= \case
         Right client -> do
           logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv <> " (user " <> bshow userId <> maybe "" (" for entity " <>) entityId_ <> ")"
-          atomically $ putTMVar clientVar r
+          atomically $ putTMVar (sessionVar v) (Right client)
           liftIO $ incClientStat c userId client "CLIENT" "OK"
           atomically $ writeTBQueue (subQ c) ("", "", APC SAENone $ hostEvent CONNECT client)
           successAction client
@@ -661,15 +684,15 @@ newProtocolClient c tSess@(userId, srv, entityId_) clients connectClient reconne
           if temporaryAgentError e
             then retryAction
             else atomically $ do
-              putTMVar clientVar (Left e)
-              TM.delete tSess clients
+              putTMVar (sessionVar v) (Left e)
+              removeTSessVar v tSess clients
           throwError e
     tryConnectAsync :: m ()
     tryConnectAsync = newAsyncAction connectAsync $ asyncClients c
     connectAsync :: Int -> m ()
     connectAsync aId = do
       ri <- asks $ reconnectInterval . config
-      withRetryInterval ri $ \_ loop -> void $ tryConnectClient (const $ reconnectClient c tSess) loop
+      withRetryInterval ri $ \_ loop -> void $ tryConnectClient (const $ clientConnected c tSess) loop
       atomically . removeAsyncAction aId $ asyncClients c
 
 hostEvent :: forall err msg. (ProtocolTypeI (ProtoType msg), ProtocolServerClient err msg) => (AProtocolType -> TransportHost -> ACommand 'Agent 'AENone) -> Client msg -> ACommand 'Agent 'AENone
@@ -687,7 +710,7 @@ closeAgentClient c = liftIO $ do
   closeProtocolServerClients c smpClients
   closeProtocolServerClients c ntfClients
   closeProtocolServerClients c xftpClients
-  cancelActions . actions $ reconnections c
+  atomically (swapTVar (smpSubWorkers c) M.empty) >>= mapM_ cancelReconnect
   cancelActions . actions $ asyncClients c
   clearWorkers smpDeliveryWorkers >>= mapM_ (cancelWorker . fst)
   clearWorkers asyncCmdWorkers >>= mapM_ cancelWorker
@@ -701,6 +724,8 @@ closeAgentClient c = liftIO $ do
     clearWorkers workers = atomically $ swapTVar (workers c) mempty
     clear :: Monoid m => (AgentClient -> TVar m) -> IO ()
     clear sel = atomically $ writeTVar (sel c) mempty
+    cancelReconnect :: SessionVar (Async ())  -> IO ()
+    cancelReconnect v = void . forkIO $ atomically (readTMVar $ sessionVar v) >>= uninterruptibleCancel
 
 cancelWorker :: Worker -> IO ()
 cancelWorker Worker {doWork, action} = do
@@ -728,9 +753,9 @@ closeClient c clientSel tSess =
   atomically (TM.lookupDelete tSess $ clientSel c) >>= mapM_ (closeClient_ c)
 
 closeClient_ :: ProtocolServerClient err msg => AgentClient -> ClientVar msg -> IO ()
-closeClient_ c cVar = do
+closeClient_ c v = do
   NetworkConfig {tcpConnectTimeout} <- readTVarIO $ useNetworkConfig c
-  tcpConnectTimeout `timeout` atomically (readTMVar cVar) >>= \case
+  tcpConnectTimeout `timeout` atomically (readTMVar $ sessionVar v) >>= \case
     Just (Right client) -> closeProtocolServerClient client `catchAll_` pure ()
     _ -> pure ()
 
@@ -991,7 +1016,7 @@ temporaryOrHostError = \case
   e -> temporaryAgentError e
 
 -- | Subscribe to queues. The list of results can have a different order.
-subscribeQueues :: forall m. AgentMonad m => AgentClient -> [RcvQueue] -> m [(RcvQueue, Either AgentErrorType ())]
+subscribeQueues :: forall m. AgentMonad' m => AgentClient -> [RcvQueue] -> m [(RcvQueue, Either AgentErrorType ())]
 subscribeQueues c qs = do
   (errs, qs') <- partitionEithers <$> mapM checkQueue qs
   forM_ qs' $ \rq@RcvQueue {connId} -> atomically $ do
@@ -1009,13 +1034,13 @@ subscribeQueues c qs = do
       rs <- sendBatch subscribeSMPQueues smp qs'
       mapM_ (uncurry $ processSubResult c) rs
       when (any temporaryClientError . lefts . map snd $ L.toList rs) . unliftIO u $
-        reconnectServer c (transportSession' smp)
+        resubscribeSMPSession c (transportSession' smp)
       pure rs
 
 type BatchResponses e r = (NonEmpty (RcvQueue, Either e r))
 
 -- statBatchSize is not used to batch the commands, only for traffic statistics
-sendTSessionBatches :: forall m q r. AgentMonad m => ByteString -> Int -> (q -> RcvQueue) -> (SMPClient -> NonEmpty q -> IO (BatchResponses SMPClientError r)) -> AgentClient -> [q] -> m [(RcvQueue, Either AgentErrorType r)]
+sendTSessionBatches :: forall m q r. AgentMonad' m => ByteString -> Int -> (q -> RcvQueue) -> (SMPClient -> NonEmpty q -> IO (BatchResponses SMPClientError r)) -> AgentClient -> [q] -> m [(RcvQueue, Either AgentErrorType r)]
 sendTSessionBatches statCmd statBatchSize toRQ action c qs =
   concatMap L.toList <$> (mapConcurrently sendClientBatch =<< batchQueues)
   where
@@ -1029,7 +1054,7 @@ sendTSessionBatches statCmd statBatchSize toRQ action c qs =
            in M.alter (Just . maybe [q] (q <|)) tSess m
     sendClientBatch :: (SMPTransportSession, NonEmpty q) -> m (BatchResponses AgentErrorType r)
     sendClientBatch (tSess@(userId, srv, _), qs') =
-      tryError (getSMPServerClient c tSess) >>= \case
+      tryAgentError' (getSMPServerClient c tSess) >>= \case
         Left e -> pure $ L.map ((,Left e) . toRQ) qs'
         Right smp -> liftIO $ do
           logServer "-->" c srv (bshow (length qs') <> " queues") statCmd
@@ -1130,7 +1155,7 @@ enableQueueNotifications c rq@RcvQueue {rcvId, rcvPrivateKey} notifierKey rcvNtf
   withSMPClient c rq "NKEY <nkey>" $ \smp ->
     enableSMPQueueNotifications smp rcvPrivateKey rcvId notifierKey rcvNtfPublicDhKey
 
-enableQueuesNtfs :: forall m. AgentMonad m => AgentClient -> [(RcvQueue, SMP.NtfPublicVerifyKey, SMP.RcvNtfPublicDhKey)] -> m [(RcvQueue, Either AgentErrorType (SMP.NotifierId, SMP.RcvNtfPublicDhKey))]
+enableQueuesNtfs :: forall m. AgentMonad' m => AgentClient -> [(RcvQueue, SMP.NtfPublicVerifyKey, SMP.RcvNtfPublicDhKey)] -> m [(RcvQueue, Either AgentErrorType (SMP.NotifierId, SMP.RcvNtfPublicDhKey))]
 enableQueuesNtfs = sendTSessionBatches "NKEY" 90 fst3 enableQueues_
   where
     fst3 (x, _, _) = x
@@ -1144,7 +1169,7 @@ disableQueueNotifications c rq@RcvQueue {rcvId, rcvPrivateKey} =
   withSMPClient c rq "NDEL" $ \smp ->
     disableSMPQueueNotifications smp rcvPrivateKey rcvId
 
-disableQueuesNtfs :: forall m. AgentMonad m => AgentClient -> [RcvQueue] -> m [(RcvQueue, Either AgentErrorType ())]
+disableQueuesNtfs :: forall m. AgentMonad' m => AgentClient -> [RcvQueue] -> m [(RcvQueue, Either AgentErrorType ())]
 disableQueuesNtfs = sendTSessionBatches "NDEL" 90 id $ sendBatch disableSMPQueuesNtfs
 
 sendAck :: AgentMonad m => AgentClient -> RcvQueue -> MsgId -> m ()
@@ -1171,7 +1196,7 @@ deleteQueue c rq@RcvQueue {rcvId, rcvPrivateKey} = do
   withSMPClient c rq "DEL" $ \smp ->
     deleteSMPQueue smp rcvPrivateKey rcvId
 
-deleteQueues :: forall m. AgentMonad m => AgentClient -> [RcvQueue] -> m [(RcvQueue, Either AgentErrorType ())]
+deleteQueues :: forall m. AgentMonad' m => AgentClient -> [RcvQueue] -> m [(RcvQueue, Either AgentErrorType ())]
 deleteQueues = sendTSessionBatches "DEL" 90 id $ sendBatch deleteSMPQueues
 
 sendAgentMessage :: AgentMonad m => AgentClient -> SndQueue -> MsgFlags -> ByteString -> m ()
@@ -1527,6 +1552,114 @@ getAgentSubscriptions c = do
     enc :: StrEncoding a => a -> Text
     enc = decodeLatin1 . strEncode
 
+data AgentWorkersDetails = AgentWorkersDetails
+  { smpClients_ :: [Text],
+    ntfClients_ :: [Text],
+    xftpClients_ :: [Text],
+    smpDeliveryWorkers_ :: Map Text Int,
+    asyncCmdWorkers_ :: Map Text Int,
+    smpSubWorkers_ :: [Text],
+    asyncCients_ :: [Int],
+    ntfWorkers_ :: Map Text Int,
+    ntfSMPWorkers_ :: Map Text Int,
+    xftpRcvWorkers_ :: Map Text Int,
+    xftpSndWorkers_ :: Map Text Int,
+    xftpDelWorkers_ :: Map Text Int
+  }
+  deriving (Show)
+
+getAgentWorkersDetails :: MonadIO m => AgentClient -> m AgentWorkersDetails
+getAgentWorkersDetails AgentClient {smpClients, ntfClients, xftpClients, smpDeliveryWorkers, asyncCmdWorkers, smpSubWorkers, asyncClients = TAsyncs {actions}, agentEnv} = do
+  smpClients_ <- textKeys <$> readTVarIO smpClients
+  ntfClients_ <- textKeys <$> readTVarIO ntfClients
+  xftpClients_ <- textKeys <$> readTVarIO xftpClients
+  smpDeliveryWorkers_ <- workerStats . fmap fst =<< readTVarIO smpDeliveryWorkers
+  asyncCmdWorkers_ <- workerStats =<< readTVarIO asyncCmdWorkers
+  smpSubWorkers_ <- textKeys <$> readTVarIO smpSubWorkers
+  asyncCients_ <- M.keys <$> readTVarIO actions
+  ntfWorkers_ <- workerStats =<< readTVarIO ntfWorkers
+  ntfSMPWorkers_ <- workerStats =<< readTVarIO ntfSMPWorkers
+  xftpRcvWorkers_ <- workerStats =<< readTVarIO xftpRcvWorkers
+  xftpSndWorkers_ <- workerStats =<< readTVarIO xftpSndWorkers
+  xftpDelWorkers_ <- workerStats =<< readTVarIO xftpDelWorkers
+  pure
+    AgentWorkersDetails
+      { smpClients_,
+        ntfClients_,
+        xftpClients_,
+        smpDeliveryWorkers_,
+        asyncCmdWorkers_,
+        smpSubWorkers_,
+        asyncCients_,
+        ntfWorkers_,
+        ntfSMPWorkers_,
+        xftpRcvWorkers_,
+        xftpSndWorkers_,
+        xftpDelWorkers_
+      }
+  where
+    textKeys :: StrEncoding k => Map k v -> [Text]
+    textKeys = map textKey . M.keys
+    textKey :: StrEncoding k => k -> Text
+    textKey = decodeASCII . strEncode
+    workerStats :: (StrEncoding k, MonadIO m) => Map k Worker -> m (Map Text Int)
+    workerStats ws = fmap M.fromList . forM (M.toList ws) $ \(qa, Worker {restarts}) -> do
+      RestartCount {restartCount} <- readTVarIO restarts
+      pure (textKey qa, restartCount)
+    Env {ntfSupervisor, xftpAgent} = agentEnv
+    NtfSupervisor {ntfWorkers, ntfSMPWorkers} = ntfSupervisor
+    XFTPAgent {xftpRcvWorkers, xftpSndWorkers, xftpDelWorkers} = xftpAgent
+
+data AgentWorkersSummary = AgentWorkersSummary
+  { smpClientsCount :: Int,
+    ntfClientsCount :: Int,
+    xftpClientsCount :: Int,
+    smpDeliveryWorkersCount :: Int,
+    asyncCmdWorkersCount :: Int,
+    smpSubWorkersCount :: Int,
+    asyncCientsCount :: Int,
+    ntfWorkersCount :: Int,
+    ntfSMPWorkersCount :: Int,
+    xftpRcvWorkersCount :: Int,
+    xftpSndWorkersCount :: Int,
+    xftpDelWorkersCount :: Int
+  }
+  deriving (Show)
+
+getAgentWorkersSummary :: MonadIO m => AgentClient -> m AgentWorkersSummary
+getAgentWorkersSummary AgentClient {smpClients, ntfClients, xftpClients, smpDeliveryWorkers, asyncCmdWorkers, smpSubWorkers, asyncClients = TAsyncs {actions}, agentEnv} = do
+  smpClientsCount <- M.size <$> readTVarIO smpClients
+  ntfClientsCount <- M.size <$> readTVarIO ntfClients
+  xftpClientsCount <- M.size <$> readTVarIO xftpClients
+  smpDeliveryWorkersCount <- M.size <$> readTVarIO smpDeliveryWorkers
+  asyncCmdWorkersCount <- M.size <$> readTVarIO asyncCmdWorkers
+  smpSubWorkersCount <- M.size <$> readTVarIO smpSubWorkers
+  asyncCientsCount <- M.size <$> readTVarIO actions
+  ntfWorkersCount <- M.size <$> readTVarIO ntfWorkers
+  ntfSMPWorkersCount <- M.size <$> readTVarIO ntfSMPWorkers
+  xftpRcvWorkersCount <- M.size <$> readTVarIO xftpRcvWorkers
+  xftpSndWorkersCount <- M.size <$> readTVarIO xftpSndWorkers
+  xftpDelWorkersCount <- M.size <$> readTVarIO xftpDelWorkers
+  pure
+    AgentWorkersSummary
+      { smpClientsCount,
+        ntfClientsCount,
+        xftpClientsCount,
+        smpDeliveryWorkersCount,
+        asyncCmdWorkersCount,
+        smpSubWorkersCount,
+        asyncCientsCount,
+        ntfWorkersCount,
+        ntfSMPWorkersCount,
+        xftpRcvWorkersCount,
+        xftpSndWorkersCount,
+        xftpDelWorkersCount
+      }
+  where
+    Env {ntfSupervisor, xftpAgent} = agentEnv
+    NtfSupervisor {ntfWorkers, ntfSMPWorkers} = ntfSupervisor
+    XFTPAgent {xftpRcvWorkers, xftpSndWorkers, xftpDelWorkers} = xftpAgent
+
 $(J.deriveJSON defaultJSON ''AgentLocks)
 
 $(J.deriveJSON (enumJSON $ dropPrefix "TS") ''ProtocolTestStep)
@@ -1536,3 +1669,7 @@ $(J.deriveJSON defaultJSON ''ProtocolTestFailure)
 $(J.deriveJSON defaultJSON ''SubInfo)
 
 $(J.deriveJSON defaultJSON ''SubscriptionsInfo)
+
+$(J.deriveJSON defaultJSON {J.fieldLabelModifier = takeWhile (/= '_')} ''AgentWorkersDetails)
+
+$(J.deriveJSON defaultJSON ''AgentWorkersSummary)

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

@@ -19,6 +19,7 @@ module Simplex.Messaging.Agent.Env.SQLite
     defaultAgentConfig,
     defaultReconnectInterval,
     tryAgentError,
+    tryAgentError',
     catchAgentError,
     agentFinally,
     Env (..),
@@ -33,6 +34,7 @@ module Simplex.Messaging.Agent.Env.SQLite
   )
 where
 
+import Control.Monad
 import Control.Monad.Except
 import Control.Monad.IO.Unlift
 import Control.Monad.Reader
@@ -112,8 +114,7 @@ data AgentConfig = AgentConfig
     certificateFile :: FilePath,
     e2eEncryptVRange :: VersionRange,
     smpAgentVRange :: VersionRange,
-    smpClientVRange :: VersionRange,
-    initialClientId :: Int
+    smpClientVRange :: VersionRange
   }
 
 defaultReconnectInterval :: RetryInterval
@@ -184,15 +185,13 @@ defaultAgentConfig =
       certificateFile = "/etc/opt/simplex-agent/agent.crt",
       e2eEncryptVRange = supportedE2EEncryptVRange,
       smpAgentVRange = supportedSMPAgentVRange,
-      smpClientVRange = supportedSMPClientVRange,
-      initialClientId = 0
+      smpClientVRange = supportedSMPClientVRange
     }
 
 data Env = Env
   { config :: AgentConfig,
     store :: SQLiteStore,
     random :: TVar ChaChaDRG,
-    clientCounter :: TVar Int,
     randomServer :: TVar StdGen,
     ntfSupervisor :: NtfSupervisor,
     xftpAgent :: XFTPAgent,
@@ -200,14 +199,13 @@ data Env = Env
   }
 
 newSMPAgentEnv :: AgentConfig -> SQLiteStore -> IO Env
-newSMPAgentEnv config@AgentConfig {initialClientId} store = do
+newSMPAgentEnv config store = do
   random <- C.newRandom
-  clientCounter <- newTVarIO initialClientId
   randomServer <- newTVarIO =<< liftIO newStdGen
   ntfSupervisor <- atomically . newNtfSubSupervisor $ tbqSize config
   xftpAgent <- atomically newXFTPAgent
   multicastSubscribers <- newTMVarIO 0
-  pure Env {config, store, random, clientCounter, randomServer, ntfSupervisor, xftpAgent, multicastSubscribers}
+  pure Env {config, store, random, randomServer, ntfSupervisor, xftpAgent, multicastSubscribers}
 
 createAgentStore :: FilePath -> ScrubbedBytes -> Bool -> MigrationConfirmation -> IO (Either MigrationError SQLiteStore)
 createAgentStore dbFilePath dbKey keepKey = createSQLiteStore dbFilePath dbKey keepKey Migrations.app
@@ -250,6 +248,11 @@ tryAgentError :: AgentMonad m => m a -> m (Either AgentErrorType a)
 tryAgentError = tryAllErrors mkInternal
 {-# INLINE tryAgentError #-}
 
+-- unlike runExceptT, this ensures we catch IO exceptions as well
+tryAgentError' :: AgentMonad' m => ExceptT AgentErrorType m a -> m (Either AgentErrorType a)
+tryAgentError' = fmap join . runExceptT . tryAgentError
+{-# INLINE tryAgentError' #-}
+
 catchAgentError :: AgentMonad m => m a -> (AgentErrorType -> m a) -> m a
 catchAgentError = catchAllErrors mkInternal
 {-# INLINE catchAgentError #-}

src/Simplex/Messaging/Agent/Server.hs

@@ -48,9 +48,11 @@ runSMPAgentBlocking (ATransport t) cfg@AgentConfig {tcpPort, caCertificateFile,
     smpAgent _ = do
       -- tlsServerParams is not in Env to avoid breaking functional API w/t key and certificate generation
       tlsServerParams <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile
+      clientId <- newTVarIO 0
       runTransportServer started tcpPort tlsServerParams defaultTransportServerConfig $ \(h :: c) -> do
         liftIO . putLn h $ "Welcome to SMP agent v" <> B.pack simplexMQVersion
-        c <- getAgentClient initServers
+        cId <- atomically $ stateTVar clientId $ \i -> (i + 1, i + 1)
+        c <- getAgentClient cId initServers
         logConnection c True
         race_ (connectClient h c) (runAgentClient c)
           `E.finally` disconnectAgentClient c

src/Simplex/Messaging/Builder.hs

@@ -1,52 +0,0 @@
-{-# LANGUAGE StrictData #-}
-
-module Simplex.Messaging.Builder
-  ( Builder (length, builder),
-    byteString,
-    lazyByteString,
-    word16BE,
-    char8,
-    toLazyByteString,
-  )
-where
-
-import qualified Data.ByteString as B
-import qualified Data.ByteString.Builder as BB
-import qualified Data.ByteString.Lazy as LB
-import Data.Word (Word16)
-
-
--- length-aware builder
-data Builder = Builder {length :: Int, builder :: BB.Builder}
-
-instance Semigroup Builder where
-  Builder l1 b1 <> Builder l2 b2 = Builder (l1 + l2) (b1 <> b2)
-  {-# INLINE (<>) #-}
-
-instance Monoid Builder where
-  mempty = Builder 0 mempty
-  {-# INLINE mempty #-}
-  mconcat bs = Builder (sum ls) (mconcat bbs)
-    where
-      (ls, bbs) = foldr (\(Builder l b) ~(ls', bbs') -> (l : ls', b : bbs')) ([], []) bs
-  {-# INLINE mconcat #-}
-
-byteString :: B.ByteString -> Builder
-byteString s = Builder (B.length s) (BB.byteString s)
-{-# INLINE byteString #-}
-
-lazyByteString :: LB.ByteString -> Builder
-lazyByteString s = Builder (fromIntegral $ LB.length s) (BB.lazyByteString s)
-{-# INLINE lazyByteString #-}
-
-word16BE :: Word16 -> Builder
-word16BE = Builder 2 . BB.word16BE
-{-# INLINE word16BE #-}
-
-char8 :: Char -> Builder
-char8 = Builder 1 . BB.char8
-{-# INLINE char8 #-}
-
-toLazyByteString :: Builder -> LB.ByteString
-toLazyByteString = BB.toLazyByteString . builder
-{-# INLINE toLazyByteString #-}

src/Simplex/Messaging/Client.hs

@@ -96,10 +96,7 @@ import Data.Maybe (fromMaybe)
 import Data.Time.Clock (UTCTime, getCurrentTime)
 import Network.Socket (ServiceName)
 import Numeric.Natural
-import Simplex.Messaging.Builder (Builder)
-import qualified Simplex.Messaging.Builder as BB
 import qualified Simplex.Messaging.Crypto as C
-import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (defaultJSON, dropPrefix, enumJSON)
 import Simplex.Messaging.Protocol
@@ -136,7 +133,7 @@ data PClient err msg = PClient
     pingErrorCount :: TVar Int,
     clientCorrId :: TVar Natural,
     sentCommands :: TMap CorrId (Request err msg),
-    sndQ :: TBQueue Builder,
+    sndQ :: TBQueue ByteString,
     rcvQ :: TBQueue (NonEmpty (SignedTransmission err msg)),
     msgQ :: Maybe (TBQueue (ServerTransmission msg))
   }
@@ -660,9 +657,11 @@ sendBatch c@ProtocolClient {client_ = PClient {sndQ}} b = do
     TBLargeTransmission Request {entityId} -> do
       putStrLn "send error: large message"
       pure [Response entityId $ Left $ PCETransportError TELargeMsg]
-    TBTransmissions s n rs -> do
-      when (n > 0) $ atomically $ writeTBQueue sndQ $ tEncodeBatch n s
-      mapConcurrently (getResponse c) rs
+    TBTransmissions s n rs
+      | n > 0 -> do
+          atomically $ writeTBQueue sndQ s
+          mapConcurrently (getResponse c) rs
+      | otherwise -> pure []
     TBTransmission s r -> do
       atomically $ writeTBQueue sndQ s
       (: []) <$> getResponse c r
@@ -675,11 +674,11 @@ sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ}, batch, blockSize
     -- two separate "atomically" needed to avoid blocking
     sendRecv :: SentRawTransmission -> Request err msg -> IO (Either (ProtocolClientError err) msg)
     sendRecv t r
-      | BB.length s > blockSize - 2 = pure $ Left $ PCETransportError TELargeMsg
+      | B.length s > blockSize - 2 = pure $ Left $ PCETransportError TELargeMsg
       | otherwise = atomically (writeTBQueue sndQ s) >> response <$> getResponse c r
       where
         s
-          | batch = tEncodeBatch 1 . encodeLarge $ tEncode t
+          | batch = tEncodeBatch1 t
           | otherwise = tEncode t
 
 -- TODO switch to timeout or TimeManager that supports Int64

src/Simplex/Messaging/Crypto.hs

@@ -140,7 +140,6 @@ module Simplex.Messaging.Crypto
 
     -- * Message padding / un-padding
     pad,
-    pad',
     unPad,
 
     -- * X509 Certificates
@@ -206,8 +205,6 @@ import Database.SQLite.Simple.FromField (FromField (..))
 import Database.SQLite.Simple.ToField (ToField (..))
 import GHC.TypeLits (ErrorMessage (..), KnownNat, Nat, TypeError, natVal, type (+))
 import Network.Transport.Internal (decodeWord16, encodeWord16)
-import Simplex.Messaging.Builder (Builder, byteString, word16BE)
-import qualified Simplex.Messaging.Builder as BB
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (blobFieldDecoder, parseAll, parseString)
@@ -922,14 +919,6 @@ pad msg paddedLen
     len = B.length msg
     padLen = paddedLen - len - 2
 
-pad' :: Builder -> Int -> Either CryptoError Builder
-pad' msg paddedLen
-  | len <= maxMsgLen && padLen >= 0 = Right $ word16BE (fromIntegral len) <> msg <> byteString (B.replicate padLen '#')
-  | otherwise = Left CryptoLargeMsgError
-  where
-    len = BB.length msg
-    padLen = paddedLen - len - 2
-
 unPad :: ByteString -> Either CryptoError ByteString
 unPad padded
   | B.length lenWrd == 2 && B.length rest >= len = Right $ B.take len rest

src/Simplex/Messaging/Encoding.hs

@@ -11,7 +11,6 @@ module Simplex.Messaging.Encoding
   ( Encoding (..),
     Tail (..),
     Large (..),
-    encodeLarge,
     _smpP,
     smpEncodeList,
     smpListP,
@@ -30,8 +29,6 @@ import qualified Data.List.NonEmpty as L
 import Data.Time.Clock.System (SystemTime (..))
 import Data.Word (Word16, Word32)
 import Network.Transport.Internal (decodeWord16, decodeWord32, encodeWord16, encodeWord32)
-import Simplex.Messaging.Builder (Builder, word16BE)
-import qualified Simplex.Messaging.Builder as BB
 import Simplex.Messaging.Parsers (parseAll)
 import Simplex.Messaging.Util ((<$?>))
 
@@ -141,10 +138,6 @@ instance Encoding Large where
     Large <$> A.take len
   {-# INLINE smpP #-}
 
-encodeLarge :: Builder -> Builder
-encodeLarge s = word16BE (fromIntegral $ BB.length s) <> s
-{-# INLINE encodeLarge #-}
-
 instance Encoding SystemTime where
   smpEncode = smpEncode . systemSeconds
   {-# INLINE smpEncode #-}

src/Simplex/Messaging/Notifications/Server.hs

@@ -369,7 +369,7 @@ receive th NtfServerClient {rcvQ, sndQ, rcvActiveAt} = forever $ do
 send :: Transport c => THandle c -> NtfServerClient -> IO ()
 send h@THandle {thVersion = v} NtfServerClient {sndQ, sessionId, sndActiveAt} = forever $ do
   t <- atomically $ readTBQueue sndQ
-  void . liftIO $ tPut h Nothing [(Nothing, encodeTransmission v sessionId t)]
+  void . liftIO $ tPut h [(Nothing, encodeTransmission v sessionId t)]
   atomically . writeTVar sndActiveAt =<< liftIO getSystemTime
 
 -- instance Show a => Show (TVar a) where

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

@@ -30,7 +30,7 @@ import System.IO (BufferMode (..), hSetBuffering, stderr, stdout)
 import Text.Read (readMaybe)
 
 ntfServerVersion :: String
-ntfServerVersion = "1.7.0.4"
+ntfServerVersion = "1.7.0.6"
 
 defaultSMPBatchDelay :: Int
 defaultSMPBatchDelay = 10000

src/Simplex/Messaging/Protocol.hs

@@ -144,7 +144,7 @@ module Simplex.Messaging.Protocol
     tParse,
     tDecodeParseValidate,
     tEncode,
-    tEncodeBatch,
+    tEncodeBatch1,
     batchTransmissions,
     batchTransmissions',
 
@@ -155,7 +155,6 @@ module Simplex.Messaging.Protocol
 where
 
 import Control.Applicative (optional, (<|>))
-import Control.Concurrent (threadDelay)
 import Control.Monad.Except
 import Data.Aeson (FromJSON (..), ToJSON (..))
 import qualified Data.Aeson.TH as J
@@ -163,8 +162,6 @@ import Data.Attoparsec.ByteString.Char8 (Parser, (<?>))
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
-import qualified Data.ByteString.Lazy.Char8 as LB
-import qualified Data.ByteString.Lazy.Internal as LB
 import Data.Char (isPrint, isSpace)
 import Data.Constraint (Dict (..))
 import Data.Functor (($>))
@@ -177,8 +174,6 @@ import Data.Time.Clock.System (SystemTime (..))
 import Data.Type.Equality
 import GHC.TypeLits (ErrorMessage (..), TypeError, type (+))
 import Network.Socket (HostName, ServiceName)
-import Simplex.Messaging.Builder (Builder, char8, lazyByteString)
-import qualified Simplex.Messaging.Builder as BB
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
@@ -1288,16 +1283,16 @@ instance Encoding CommandError where
       _ -> fail "bad command error type"
 
 -- | Send signed SMP transmission to TCP transport.
-tPut :: Transport c => THandle c -> Maybe Int -> NonEmpty SentRawTransmission -> IO [Either TransportError ()]
-tPut th delay_ = fmap concat . mapM tPutBatch . batchTransmissions (batch th) (blockSize th)
+tPut :: Transport c => THandle c -> NonEmpty SentRawTransmission -> IO [Either TransportError ()]
+tPut th = fmap concat . mapM tPutBatch . batchTransmissions (batch th) (blockSize th)
   where
     tPutBatch :: TransportBatch () -> IO [Either TransportError ()]
     tPutBatch = \case
       TBLargeTransmission _ -> [Left TELargeMsg] <$ putStrLn "tPut error: large message"
-      TBTransmissions s n _ -> replicate n <$> (tPutLog th (tEncodeBatch n s) <* mapM_ threadDelay delay_)
+      TBTransmissions s n _ -> replicate n <$> (tPutLog th s)
       TBTransmission s _ -> (: []) <$> tPutLog th s
 
-tPutLog :: Transport c => THandle c -> Builder -> IO (Either TransportError ())
+tPutLog :: Transport c => THandle c -> ByteString -> IO (Either TransportError ())
 tPutLog th s = do
   r <- tPutBlock th s
   case r of
@@ -1305,8 +1300,8 @@ tPutLog th s = do
     _ -> pure ()
   pure r
 
--- Builder in TBTransmissions does not include byte with transmissions count, it is added by tEncodeBatch
-data TransportBatch r = TBTransmissions Builder Int [r] | TBTransmission Builder r | TBLargeTransmission r
+-- ByteString in TBTransmissions includes byte with transmissions count
+data TransportBatch r = TBTransmissions ByteString Int [r] | TBTransmission ByteString r | TBLargeTransmission r
 
 batchTransmissions :: Bool -> Int -> NonEmpty SentRawTransmission -> [TransportBatch ()]
 batchTransmissions batch bSize = batchTransmissions' batch bSize . L.map (,())
@@ -1314,35 +1309,43 @@ batchTransmissions batch bSize = batchTransmissions' batch bSize . L.map (,())
 -- | encodes and batches transmissions into blocks,
 batchTransmissions' :: forall r. Bool -> Int -> NonEmpty (SentRawTransmission, r) -> [TransportBatch r]
 batchTransmissions' batch bSize
-  | batch = addBatch . foldr addTransmission ([], mempty, 0, [])
+  | batch = addBatch . foldr addTransmission ([], 0, 0, [], [])
   | otherwise = map mkBatch1 . L.toList
   where
     mkBatch1 :: (SentRawTransmission, r) -> TransportBatch r
     mkBatch1 (t, r)
       -- 2 bytes are reserved for pad size
-      | BB.length s <= bSize - 2 = TBTransmission s r
+      | B.length s <= bSize - 2 = TBTransmission s r
       | otherwise = TBLargeTransmission r
       where
         s = tEncode t
-    addTransmission :: (SentRawTransmission, r) -> ([TransportBatch r], Builder, Int, [r]) -> ([TransportBatch r], Builder, Int, [r])
-    addTransmission (t, r) acc@(bs, b, n, rs)
-      -- 3 = 2 bytes reserved for pad size + 1 for transmission count
-      | len + BB.length b <= bSize - 3 && n < 255 = (bs, s <> b, 1 + n, r : rs)
-      | len <= bSize - 3 = (addBatch acc, s, 1, [r])
-      | otherwise = (TBLargeTransmission r : addBatch acc, mempty, 0, [])
+    -- 3 = 2 bytes reserved for pad size + 1 for transmission count
+    bSize' = bSize - 3
+    addTransmission :: (SentRawTransmission, r) -> ([TransportBatch r], Int, Int, [ByteString], [r]) -> ([TransportBatch r], Int, Int, [ByteString], [r])
+    addTransmission (t, r) acc@(bs, len, n, ss, rs)
+      | len' <= bSize' && n < 255 = (bs, len', 1 + n, s : ss, r : rs)
+      | sLen <= bSize' = (addBatch acc, sLen, 1, [s], [r])
+      | otherwise = (TBLargeTransmission r : addBatch acc, 0, 0, [], [])
       where
-        s = encodeLarge $ tEncode t
-        len = BB.length s
-    addBatch :: ([TransportBatch r], Builder, Int, [r]) -> [TransportBatch r]
-    addBatch (bs, b, n, rs) = if n == 0 then bs else TBTransmissions b n rs : bs
+        s = tEncodeForBatch t
+        sLen = B.length s
+        len' = len + sLen
+    addBatch :: ([TransportBatch r], Int, Int, [ByteString], [r]) -> [TransportBatch r]
+    addBatch (bs, _len, n, ss, rs) = if n == 0 then bs else TBTransmissions b n rs : bs
+      where
+        b = B.concat $ B.singleton (lenEncode n) : ss
 
-tEncode :: SentRawTransmission -> Builder
-tEncode (sig, t) = lazyByteString $ LB.chunk (smpEncode $ C.signatureBytes sig) (LB.fromStrict t)
+tEncode :: SentRawTransmission -> ByteString
+tEncode (sig, t) = smpEncode (C.signatureBytes sig) <> t
 {-# INLINE tEncode #-}
 
-tEncodeBatch :: Int -> Builder -> Builder
-tEncodeBatch n s = char8 (lenEncode n) <> s
-{-# INLINE tEncodeBatch #-}
+tEncodeForBatch :: SentRawTransmission -> ByteString
+tEncodeForBatch = smpEncode . Large . tEncode
+{-# INLINE tEncodeForBatch #-}
+
+tEncodeBatch1 :: SentRawTransmission -> ByteString
+tEncodeBatch1 t = lenEncode 1 `B.cons` tEncodeForBatch t
+{-# INLINE tEncodeBatch1 #-}
 
 encodeTransmission :: ProtocolEncoding e c => Version -> ByteString -> Transmission c -> ByteString
 encodeTransmission v sessionId (CorrId corrId, queueId, command) =

src/Simplex/Messaging/Server.hs

@@ -439,7 +439,7 @@ send h@THandle {thVersion = v} Client {sndQ, sessionId, sndActiveAt} = do
   labelMyThread . B.unpack $ "client $" <> encode sessionId <> " send"
   forever $ do
     ts <- atomically $ L.sortWith tOrder <$> readTBQueue sndQ
-    void . liftIO . tPut h Nothing $ L.map ((Nothing,) . encodeTransmission v sessionId) ts
+    void . liftIO . tPut h $ L.map ((Nothing,) . encodeTransmission v sessionId) ts
     atomically . writeTVar sndActiveAt =<< liftIO getSystemTime
   where
     tOrder :: Transmission BrokerMsg -> Int

src/Simplex/Messaging/Transport.hs

@@ -78,7 +78,6 @@ import Network.Socket
 import qualified Network.TLS as T
 import qualified Network.TLS.Extra as TE
 import qualified Paths_simplexmq as SMQ
-import Simplex.Messaging.Builder (Builder, byteString, toLazyByteString)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Parsers (dropPrefix, parse, parseRead1, sumTypeJSON)
@@ -135,9 +134,6 @@ class Transport c where
   -- | Write bytes to connection
   cPut :: c -> ByteString -> IO ()
 
-  -- | Write bytes to connection
-  cPut' :: c -> LB.ByteString -> IO ()
-
   -- | Receive ByteString from connection, allowing LF or CRLF termination.
   getLn :: c -> IO ByteString
 
@@ -221,11 +217,8 @@ instance Transport TLS where
     getBuffered tlsBuffer n t_ (T.recvData tlsContext)
 
   cPut :: TLS -> ByteString -> IO ()
-  cPut cxt = cPut' cxt . LB.fromStrict
-
-  cPut' :: TLS -> LB.ByteString -> IO ()
-  cPut' TLS {tlsContext, tlsTransportConfig = TransportConfig {transportTimeout = t_}} s =
-    withTimedErr t_ $ T.sendData tlsContext s
+  cPut TLS {tlsContext, tlsTransportConfig = TransportConfig {transportTimeout = t_}} =
+    withTimedErr t_ . T.sendData tlsContext . LB.fromStrict
 
   getLn :: TLS -> IO ByteString
   getLn TLS {tlsContext, tlsBuffer} = do
@@ -316,10 +309,10 @@ serializeTransportError = \case
   TEHandshake e -> "HANDSHAKE " <> bshow e
 
 -- | Pad and send block to SMP transport.
-tPutBlock :: Transport c => THandle c -> Builder -> IO (Either TransportError ())
+tPutBlock :: Transport c => THandle c -> ByteString -> IO (Either TransportError ())
 tPutBlock THandle {connection = c, blockSize} block =
-  bimapM (const $ pure TELargeMsg) (cPut' c . toLazyByteString) $
-    C.pad' block blockSize
+  bimapM (const $ pure TELargeMsg) (cPut c) $
+    C.pad block blockSize
 
 -- | Receive block from SMP transport.
 tGetBlock :: Transport c => THandle c -> IO (Either TransportError ByteString)
@@ -363,7 +356,7 @@ smpThHandle :: forall c. THandle c -> Version -> THandle c
 smpThHandle th v = (th :: THandle c) {thVersion = v, batch = v >= 4}
 
 sendHandshake :: (Transport c, Encoding smp) => THandle c -> smp -> ExceptT TransportError IO ()
-sendHandshake th = ExceptT . tPutBlock th . byteString . smpEncode
+sendHandshake th = ExceptT . tPutBlock th . smpEncode
 
 getHandshake :: (Transport c, Encoding smp) => THandle c -> ExceptT TransportError IO smp
 getHandshake th = ExceptT $ (parse smpP (TEHandshake PARSE) =<<) <$> tGetBlock th

src/Simplex/Messaging/Transport/WebSockets.hs

@@ -72,9 +72,6 @@ instance Transport WS where
   cPut :: WS -> ByteString -> IO ()
   cPut = sendBinaryData . wsConnection
 
-  cPut' :: WS -> LB.ByteString -> IO ()
-  cPut' = sendBinaryData . wsConnection
-
   getLn :: WS -> IO ByteString
   getLn c = do
     s <- trimCR <$> receiveData (wsConnection c)

tests/AgentTests/FunctionalAPITests.hs

@@ -375,8 +375,8 @@ runTestCfg2 aCfg bCfg baseMsgId runTest =
 
 withAgentClientsCfg2 :: AgentConfig -> AgentConfig -> (AgentClient -> AgentClient -> IO ()) -> IO ()
 withAgentClientsCfg2 aCfg bCfg runTest = do
-  a <- getSMPAgentClient' aCfg initAgentServers testDB
-  b <- getSMPAgentClient' bCfg initAgentServers testDB2
+  a <- getSMPAgentClient' 1 aCfg initAgentServers testDB
+  b <- getSMPAgentClient' 2 bCfg initAgentServers testDB2
   runTest a b
   disconnectAgentClient a
   disconnectAgentClient b
@@ -421,9 +421,9 @@ runAgentClientTest alice bob baseId = do
 
 testAgentClient3 :: HasCallStack => IO ()
 testAgentClient3 = do
-  a <- getSMPAgentClient' agentCfg initAgentServers testDB
-  b <- getSMPAgentClient' agentCfg initAgentServers testDB2
-  c <- getSMPAgentClient' agentCfg initAgentServers testDB3
+  a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  b <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
+  c <- getSMPAgentClient' 3 agentCfg initAgentServers testDB3
   runRight_ $ do
     (aIdForB, bId) <- makeConnection a b
     (aIdForC, cId) <- makeConnection a c
@@ -496,7 +496,7 @@ testAsyncInitiatingOffline =
     (bobId, cReq) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
     disconnectAgentClient alice
     aliceId <- joinConnection bob 1 True cReq "bob's connInfo" SMSubscribe
-    alice' <- liftIO $ getSMPAgentClient' agentCfg initAgentServers testDB
+    alice' <- liftIO $ getSMPAgentClient' 3 agentCfg initAgentServers testDB
     subscribeConnection alice' bobId
     ("", _, CONF confId _ "bob's connInfo") <- get alice'
     allowConnection alice' bobId confId "alice's connInfo"
@@ -513,7 +513,7 @@ testAsyncJoiningOfflineBeforeActivation =
     disconnectAgentClient bob
     ("", _, CONF confId _ "bob's connInfo") <- get alice
     allowConnection alice bobId confId "alice's connInfo"
-    bob' <- liftIO $ getSMPAgentClient' agentCfg initAgentServers testDB2
+    bob' <- liftIO $ getSMPAgentClient' 3 agentCfg initAgentServers testDB2
     subscribeConnection bob' aliceId
     get alice ##> ("", bobId, CON)
     get bob' ##> ("", aliceId, INFO "alice's connInfo")
@@ -527,11 +527,11 @@ testAsyncBothOffline =
     disconnectAgentClient alice
     aliceId <- joinConnection bob 1 True cReq "bob's connInfo" SMSubscribe
     disconnectAgentClient bob
-    alice' <- liftIO $ getSMPAgentClient' agentCfg initAgentServers testDB
+    alice' <- liftIO $ getSMPAgentClient' 3 agentCfg initAgentServers testDB
     subscribeConnection alice' bobId
     ("", _, CONF confId _ "bob's connInfo") <- get alice'
     allowConnection alice' bobId confId "alice's connInfo"
-    bob' <- liftIO $ getSMPAgentClient' agentCfg initAgentServers testDB2
+    bob' <- liftIO $ getSMPAgentClient' 4 agentCfg initAgentServers testDB2
     subscribeConnection bob' aliceId
     get alice' ##> ("", bobId, CON)
     get bob' ##> ("", aliceId, INFO "alice's connInfo")
@@ -574,8 +574,8 @@ testAsyncHelloTimeout = do
 testAllowConnectionClientRestart :: HasCallStack => ATransport -> IO ()
 testAllowConnectionClientRestart t = do
   let initAgentServersSrv2 = initAgentServers {smp = userServers [noAuthSrv testSMPServer2]}
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServersSrv2 testDB2
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg initAgentServersSrv2 testDB2
   withSmpServerStoreLogOn t testPort $ \_ -> do
     (aliceId, bobId, confId) <-
       withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2} testPort2 $ \_ -> do
@@ -595,7 +595,7 @@ testAllowConnectionClientRestart t = do
     threadDelay 100000 -- give time to enqueue confirmation (enqueueConfirmation)
     disconnectAgentClient alice
 
-    alice2 <- getSMPAgentClient' agentCfg initAgentServers testDB
+    alice2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
 
     withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2} testPort2 $ \_ -> do
       runRight $ do
@@ -613,8 +613,8 @@ testAllowConnectionClientRestart t = do
 
 testIncreaseConnAgentVersion :: HasCallStack => ATransport -> IO ()
 testIncreaseConnAgentVersion t = do
-  alice <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB2
   withSmpServerStoreMsgLogOn t testPort $ \_ -> do
     (aliceId, bobId) <- runRight $ do
       (aliceId, bobId) <- makeConnection alice bob
@@ -626,7 +626,7 @@ testIncreaseConnAgentVersion t = do
     -- version doesn't increase if incompatible
 
     disconnectAgentClient alice
-    alice2 <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
+    alice2 <- getSMPAgentClient' 3 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
 
     runRight_ $ do
       subscribeConnection alice2 bobId
@@ -637,7 +637,7 @@ testIncreaseConnAgentVersion t = do
     -- version increases if compatible
 
     disconnectAgentClient bob
-    bob2 <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB2
+    bob2 <- getSMPAgentClient' 4 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB2
 
     runRight_ $ do
       subscribeConnection bob2 aliceId
@@ -648,7 +648,7 @@ testIncreaseConnAgentVersion t = do
     -- version doesn't decrease, even if incompatible
 
     disconnectAgentClient alice2
-    alice3 <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 2 2} initAgentServers testDB
+    alice3 <- getSMPAgentClient' 5 agentCfg {smpAgentVRange = mkVersionRange 2 2} initAgentServers testDB
 
     runRight_ $ do
       subscribeConnection alice3 bobId
@@ -657,7 +657,7 @@ testIncreaseConnAgentVersion t = do
       checkVersion bob2 aliceId 3
 
     disconnectAgentClient bob2
-    bob3 <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 1} initAgentServers testDB2
+    bob3 <- getSMPAgentClient' 6 agentCfg {smpAgentVRange = mkVersionRange 1 1} initAgentServers testDB2
 
     runRight_ $ do
       subscribeConnection bob3 aliceId
@@ -674,8 +674,8 @@ checkVersion c connId v = do
 
 testIncreaseConnAgentVersionMaxCompatible :: HasCallStack => ATransport -> IO ()
 testIncreaseConnAgentVersionMaxCompatible t = do
-  alice <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB2
   withSmpServerStoreMsgLogOn t testPort $ \_ -> do
     (aliceId, bobId) <- runRight $ do
       (aliceId, bobId) <- makeConnection alice bob
@@ -687,9 +687,9 @@ testIncreaseConnAgentVersionMaxCompatible t = do
     -- version increases to max compatible
 
     disconnectAgentClient alice
-    alice2 <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
+    alice2 <- getSMPAgentClient' 3 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
     disconnectAgentClient bob
-    bob2 <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 4} initAgentServers testDB2
+    bob2 <- getSMPAgentClient' 4 agentCfg {smpAgentVRange = mkVersionRange 1 4} initAgentServers testDB2
 
     runRight_ $ do
       subscribeConnection alice2 bobId
@@ -702,8 +702,8 @@ testIncreaseConnAgentVersionMaxCompatible t = do
 
 testIncreaseConnAgentVersionStartDifferentVersion :: HasCallStack => ATransport -> IO ()
 testIncreaseConnAgentVersionStartDifferentVersion t = do
-  alice <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB2
   withSmpServerStoreMsgLogOn t testPort $ \_ -> do
     (aliceId, bobId) <- runRight $ do
       (aliceId, bobId) <- makeConnection alice bob
@@ -715,7 +715,7 @@ testIncreaseConnAgentVersionStartDifferentVersion t = do
     -- version increases to max compatible
 
     disconnectAgentClient alice
-    alice2 <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
+    alice2 <- getSMPAgentClient' 3 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
 
     runRight_ $ do
       subscribeConnection alice2 bobId
@@ -727,8 +727,8 @@ testIncreaseConnAgentVersionStartDifferentVersion t = do
 
 testDeliverClientRestart :: HasCallStack => ATransport -> IO ()
 testDeliverClientRestart t = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
 
   (aliceId, bobId) <- withSmpServerStoreMsgLogOn t testPort $ \_ -> do
     runRight $ do
@@ -743,7 +743,7 @@ testDeliverClientRestart t = do
 
   disconnectAgentClient bob
 
-  bob2 <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  bob2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
 
   withSmpServerStoreMsgLogOn t testPort $ \_ -> do
     runRight_ $ do
@@ -758,8 +758,8 @@ testDeliverClientRestart t = do
 
 testDuplicateMessage :: HasCallStack => ATransport -> IO ()
 testDuplicateMessage t = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   (aliceId, bobId, bob1) <- withSmpServerStoreMsgLogOn t testPort $ \_ -> do
     (aliceId, bobId) <- runRight $ makeConnection alice bob
     runRight_ $ do
@@ -769,7 +769,7 @@ testDuplicateMessage t = do
     disconnectAgentClient bob
 
     -- if the agent user did not send ACK, the message will be delivered again
-    bob1 <- getSMPAgentClient' agentCfg initAgentServers testDB2
+    bob1 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
     runRight_ $ do
       subscribeConnection bob1 aliceId
       get bob1 =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
@@ -785,13 +785,13 @@ testDuplicateMessage t = do
   -- commenting two lines below and uncommenting further two lines would also runRight_,
   -- it is the scenario tested above, when the message was not acknowledged by the user
   threadDelay 200000
-  Left (BROKER _ TIMEOUT) <- runExceptT $ ackMessage bob1 aliceId 5 Nothing
+  Left (BROKER _ NETWORK) <- runExceptT $ ackMessage bob1 aliceId 5 Nothing
 
   disconnectAgentClient alice
   disconnectAgentClient bob1
 
-  alice2 <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob2 <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice2 <- getSMPAgentClient' 4 agentCfg initAgentServers testDB
+  bob2 <- getSMPAgentClient' 5 agentCfg initAgentServers testDB2
 
   withSmpServerStoreMsgLogOn t testPort $ \_ -> do
     runRight_ $ do
@@ -808,8 +808,8 @@ testDuplicateMessage t = do
 
 testSkippedMessages :: HasCallStack => ATransport -> IO ()
 testSkippedMessages t = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   (aliceId, bobId) <- withSmpServerStoreLogOn t testPort $ \_ -> do
     (aliceId, bobId) <- runRight $ makeConnection alice bob
     runRight_ $ do
@@ -835,8 +835,8 @@ testSkippedMessages t = do
 
   disconnectAgentClient alice
 
-  alice2 <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob2 <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
+  bob2 <- getSMPAgentClient' 4 agentCfg initAgentServers testDB2
 
   withSmpServerStoreLogOn t testPort $ \_ -> do
     runRight_ $ do
@@ -899,7 +899,7 @@ setupDesynchronizedRatchet alice bob = do
   -- importing database backup after progressing ratchet de-synchronizes ratchet
   liftIO $ renameFile (testDB2 <> ".bak") testDB2
 
-  bob2 <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  bob2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
 
   runRight_ $ do
     subscribeConnection bob2 aliceId
@@ -959,8 +959,8 @@ serverUpP = \case
 
 testRatchetSyncClientRestart :: HasCallStack => ATransport -> IO ()
 testRatchetSyncClientRestart t = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   (aliceId, bobId, bob2) <- withSmpServerStoreMsgLogOn t testPort $ \_ ->
     setupDesynchronizedRatchet alice bob
   ("", "", DOWN _ _) <- nGet alice
@@ -968,7 +968,7 @@ testRatchetSyncClientRestart t = do
   ConnectionStats {ratchetSyncState} <- runRight $ synchronizeRatchet bob2 aliceId False
   liftIO $ ratchetSyncState `shouldBe` RSStarted
   disconnectAgentClient bob2
-  bob3 <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  bob3 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
   withSmpServerStoreMsgLogOn t testPort $ \_ -> do
     runRight_ $ do
       ("", "", UP _ _) <- nGet alice
@@ -984,8 +984,8 @@ testRatchetSyncClientRestart t = do
 
 testRatchetSyncSuspendForeground :: HasCallStack => ATransport -> IO ()
 testRatchetSyncSuspendForeground t = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   (aliceId, bobId, bob2) <- withSmpServerStoreMsgLogOn t testPort $ \_ ->
     setupDesynchronizedRatchet alice bob
 
@@ -1018,8 +1018,8 @@ testRatchetSyncSuspendForeground t = do
 
 testRatchetSyncSimultaneous :: HasCallStack => ATransport -> IO ()
 testRatchetSyncSimultaneous t = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   (aliceId, bobId, bob2) <- withSmpServerStoreMsgLogOn t testPort $ \_ ->
     setupDesynchronizedRatchet alice bob
 
@@ -1101,7 +1101,7 @@ testInactiveNoSubs :: ATransport -> IO ()
 testInactiveNoSubs t = do
   let cfg' = cfg {inactiveClientExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 1}}
   withSmpServerConfigOn t cfg' testPort $ \_ -> do
-    alice <- getSMPAgentClient' agentCfg initAgentServers testDB
+    alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
     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)
@@ -1111,7 +1111,7 @@ testInactiveWithSubs :: ATransport -> IO ()
 testInactiveWithSubs t = do
   let cfg' = cfg {inactiveClientExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 1}}
   withSmpServerConfigOn t cfg' testPort $ \_ -> do
-    alice <- getSMPAgentClient' agentCfg initAgentServers testDB
+    alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
     runRight_ . void $ createConnection alice 1 True SCMInvitation Nothing SMSubscribe
     Nothing <- 800000 `timeout` get alice
     liftIO $ threadDelay 1200000
@@ -1123,7 +1123,7 @@ testActiveClientNotDisconnected :: ATransport -> IO ()
 testActiveClientNotDisconnected t = do
   let cfg' = cfg {inactiveClientExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 1}}
   withSmpServerConfigOn t cfg' testPort $ \_ -> do
-    alice <- getSMPAgentClient' agentCfg initAgentServers testDB
+    alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
     ts <- getSystemTime
     runRight_ $ do
       (connId, _cReq) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
@@ -1214,8 +1214,8 @@ testSuspendingAgentTimeout t = withAgentClients2 $ \a b -> do
 
 testBatchedSubscriptions :: Int -> Int -> ATransport -> IO ()
 testBatchedSubscriptions nCreate nDel t = do
-  a <- getSMPAgentClient' agentCfg initAgentServers2 testDB
-  b <- getSMPAgentClient' agentCfg initAgentServers2 testDB2
+  a <- getSMPAgentClient' 1 agentCfg initAgentServers2 testDB
+  b <- getSMPAgentClient' 2 agentCfg initAgentServers2 testDB2
   conns <- runServers $ do
     conns <- replicateM (nCreate :: Int) $ makeConnection a b
     forM_ conns $ \(aId, bId) -> exchangeGreetings a bId b aId
@@ -1324,11 +1324,11 @@ testAsyncCommands =
 
 testAsyncCommandsRestore :: ATransport -> IO ()
 testAsyncCommandsRestore t = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
   bobId <- runRight $ createConnectionAsync alice 1 "1" True SCMInvitation SMSubscribe
   liftIO $ noMessages alice "alice doesn't receive INV because server is down"
   disconnectAgentClient alice
-  alice' <- liftIO $ getSMPAgentClient' agentCfg initAgentServers testDB
+  alice' <- liftIO $ getSMPAgentClient' 2 agentCfg initAgentServers testDB
   withSmpServerStoreLogOn t testPort $ \_ -> do
     runRight_ $ do
       subscribeConnection alice' bobId
@@ -1378,7 +1378,7 @@ testAcceptContactAsync =
 
 testDeleteConnectionAsync :: ATransport -> IO ()
 testDeleteConnectionAsync t = do
-  a <- getSMPAgentClient' agentCfg {initialCleanupDelay = 10000, cleanupInterval = 10000, deleteErrorCount = 3} initAgentServers testDB
+  a <- getSMPAgentClient' 1 agentCfg {initialCleanupDelay = 10000, cleanupInterval = 10000, deleteErrorCount = 3} initAgentServers testDB
   connIds <- withSmpServerStoreLogOn t testPort $ \_ -> runRight $ do
     (bId1, _inv) <- createConnection a 1 True SCMInvitation Nothing SMSubscribe
     (bId2, _inv) <- createConnection a 1 True SCMInvitation Nothing SMSubscribe
@@ -1398,8 +1398,8 @@ testDeleteConnectionAsync t = do
 testJoinConnectionAsyncReplyError :: HasCallStack => ATransport -> IO ()
 testJoinConnectionAsyncReplyError t = do
   let initAgentServersSrv2 = initAgentServers {smp = userServers [noAuthSrv testSMPServer2]}
-  a <- getSMPAgentClient' agentCfg initAgentServers testDB
-  b <- getSMPAgentClient' agentCfg initAgentServersSrv2 testDB2
+  a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  b <- getSMPAgentClient' 2 agentCfg initAgentServersSrv2 testDB2
   (aId, bId) <- withSmpServerStoreLogOn t testPort $ \_ -> runRight $ do
     bId <- createConnectionAsync a 1 "1" True SCMInvitation SMSubscribe
     ("1", bId', INV (ACR _ qInfo)) <- get a
@@ -1491,8 +1491,8 @@ testUsersNoServer t = withAgentClientsCfg2 aCfg agentCfg $ \a b -> do
 
 testSwitchConnection :: InitialAgentServers -> IO ()
 testSwitchConnection servers = do
-  a <- getSMPAgentClient' agentCfg servers testDB
-  b <- getSMPAgentClient' agentCfg {initialClientId = 1} servers testDB2
+  a <- getSMPAgentClient' 1 agentCfg servers testDB
+  b <- getSMPAgentClient' 2 agentCfg servers testDB2
   runRight_ $ do
     (aId, bId) <- makeConnection a b
     exchangeGreetingsMsgId 4 a bId b aId
@@ -1575,12 +1575,12 @@ testSwitchAsync servers = do
     testFullSwitch a bId b aId 16
   where
     withA :: (AgentClient -> IO a) -> IO a
-    withA = withAgent agentCfg servers testDB
+    withA = withAgent 1 agentCfg servers testDB
     withB :: (AgentClient -> IO a) -> IO a
-    withB = withAgent agentCfg {initialClientId = 1} servers testDB2
+    withB = withAgent 2 agentCfg servers testDB2
 
-withAgent :: AgentConfig -> InitialAgentServers -> FilePath -> (AgentClient -> IO a) -> IO a
-withAgent cfg' servers dbPath = bracket (getSMPAgentClient' cfg' servers dbPath) disconnectAgentClient
+withAgent :: Int -> AgentConfig -> InitialAgentServers -> FilePath -> (AgentClient -> IO a) -> IO a
+withAgent clientId cfg' servers dbPath = bracket (getSMPAgentClient' clientId cfg' servers dbPath) disconnectAgentClient
 
 sessionSubscribe :: (forall a. (AgentClient -> IO a) -> IO a) -> [ConnId] -> (AgentClient -> ExceptT AgentErrorType IO ()) -> IO ()
 sessionSubscribe withC connIds a =
@@ -1593,8 +1593,8 @@ sessionSubscribe withC connIds a =
 
 testSwitchDelete :: InitialAgentServers -> IO ()
 testSwitchDelete servers = do
-  a <- getSMPAgentClient' agentCfg servers testDB
-  b <- getSMPAgentClient' agentCfg {initialClientId = 1} servers testDB2
+  a <- getSMPAgentClient' 1 agentCfg servers testDB
+  b <- getSMPAgentClient' 2 agentCfg servers testDB2
   runRight_ $ do
     (aId, bId) <- makeConnection a b
     exchangeGreetingsMsgId 4 a bId b aId
@@ -1656,9 +1656,9 @@ testAbortSwitchStarted servers = do
     testFullSwitch a bId b aId 18
   where
     withA :: (AgentClient -> IO a) -> IO a
-    withA = withAgent agentCfg servers testDB
+    withA = withAgent 1 agentCfg servers testDB
     withB :: (AgentClient -> IO a) -> IO a
-    withB = withAgent agentCfg {initialClientId = 1} servers testDB2
+    withB = withAgent 2 agentCfg servers testDB2
 
 testAbortSwitchStartedReinitiate :: HasCallStack => InitialAgentServers -> IO ()
 testAbortSwitchStartedReinitiate servers = do
@@ -1707,9 +1707,9 @@ testAbortSwitchStartedReinitiate servers = do
     testFullSwitch a bId b aId 18
   where
     withA :: (AgentClient -> IO a) -> IO a
-    withA = withAgent agentCfg servers testDB
+    withA = withAgent 1 agentCfg servers testDB
     withB :: (AgentClient -> IO a) -> IO a
-    withB = withAgent agentCfg {initialClientId = 1} servers testDB2
+    withB = withAgent 2 agentCfg servers testDB2
 
 switchPhaseRcvP :: ConnId -> SwitchPhase -> [Maybe RcvSwitchStatus] -> ATransmission 'Agent -> Bool
 switchPhaseRcvP cId sphase swchStatuses = switchPhaseP cId QDRcv sphase (\stats -> rcvSwchStatuses' stats == swchStatuses)
@@ -1761,9 +1761,9 @@ testCannotAbortSwitchSecured servers = do
     testFullSwitch a bId b aId 16
   where
     withA :: (AgentClient -> IO a) -> IO a
-    withA = withAgent agentCfg servers testDB
+    withA = withAgent 1 agentCfg servers testDB
     withB :: (AgentClient -> IO a) -> IO a
-    withB = withAgent agentCfg {initialClientId = 1} servers testDB2
+    withB = withAgent 2 agentCfg servers testDB2
 
 testSwitch2Connections :: HasCallStack => InitialAgentServers -> IO ()
 testSwitch2Connections servers = do
@@ -1819,9 +1819,9 @@ testSwitch2Connections servers = do
     testFullSwitch a bId2 b aId2 16
   where
     withA :: (AgentClient -> IO a) -> IO a
-    withA = withAgent agentCfg servers testDB
+    withA = withAgent 1 agentCfg servers testDB
     withB :: (AgentClient -> IO a) -> IO a
-    withB = withAgent agentCfg {initialClientId = 1} servers testDB2
+    withB = withAgent 2 agentCfg servers testDB2
 
 testSwitch2ConnectionsAbort1 :: HasCallStack => InitialAgentServers -> IO ()
 testSwitch2ConnectionsAbort1 servers = do
@@ -1872,14 +1872,14 @@ testSwitch2ConnectionsAbort1 servers = do
     testFullSwitch a bId2 b aId2 14
   where
     withA :: (AgentClient -> IO a) -> IO a
-    withA = withAgent agentCfg servers testDB
+    withA = withAgent 1 agentCfg servers testDB
     withB :: (AgentClient -> IO a) -> IO a
-    withB = withAgent agentCfg {initialClientId = 1} servers testDB2
+    withB = withAgent 2 agentCfg servers testDB2
 
 testCreateQueueAuth :: HasCallStack => (Maybe BasicAuth, Version) -> (Maybe BasicAuth, Version) -> IO Int
 testCreateQueueAuth clnt1 clnt2 = do
-  a <- getClient clnt1 testDB
-  b <- getClient clnt2 testDB2
+  a <- getClient 1 clnt1 testDB
+  b <- getClient 2 clnt2 testDB2
   r <- runRight $ do
     tryError (createConnection a 1 True SCMInvitation Nothing SMSubscribe) >>= \case
       Left (SMP AUTH) -> pure 0
@@ -1900,15 +1900,15 @@ testCreateQueueAuth clnt1 clnt2 = do
   disconnectAgentClient b
   pure r
   where
-    getClient (clntAuth, clntVersion) db =
+    getClient clientId (clntAuth, clntVersion) db =
       let servers = initAgentServers {smp = userServers [ProtoServerWithAuth testSMPServer clntAuth]}
           smpCfg = (defaultClientConfig :: ProtocolClientConfig) {serverVRange = mkVersionRange 4 clntVersion}
-       in getSMPAgentClient' agentCfg {smpCfg} servers db
+       in getSMPAgentClient' clientId agentCfg {smpCfg} servers db
 
 testSMPServerConnectionTest :: ATransport -> Maybe BasicAuth -> SMPServerWithAuth -> IO (Maybe ProtocolTestFailure)
 testSMPServerConnectionTest t newQueueBasicAuth srv =
   withSmpServerConfigOn t cfg {newQueueBasicAuth} testPort2 $ \_ -> do
-    a <- getSMPAgentClient' agentCfg initAgentServers testDB -- initially passed server is not running
+    a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB -- initially passed server is not running
     runRight $ testProtocolServer a 1 srv
 
 testRatchetAdHash :: HasCallStack => IO ()
@@ -1941,8 +1941,8 @@ testDeliveryReceipts =
 
 testDeliveryReceiptsVersion :: HasCallStack => ATransport -> IO ()
 testDeliveryReceiptsVersion t = do
-  a <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
-  b <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB2
+  a <- getSMPAgentClient' 1 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
+  b <- getSMPAgentClient' 2 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB2
   withSmpServerStoreMsgLogOn t testPort $ \_ -> do
     (aId, bId) <- runRight $ do
       (aId, bId) <- makeConnection a b
@@ -1962,8 +1962,8 @@ testDeliveryReceiptsVersion t = do
 
     disconnectAgentClient a
     disconnectAgentClient b
-    a' <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 4} initAgentServers testDB
-    b' <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 4} initAgentServers testDB2
+    a' <- getSMPAgentClient' 3 agentCfg {smpAgentVRange = mkVersionRange 1 4} initAgentServers testDB
+    b' <- getSMPAgentClient' 4 agentCfg {smpAgentVRange = mkVersionRange 1 4} initAgentServers testDB2
 
     runRight_ $ do
       subscribeConnection a' bId
@@ -2103,10 +2103,10 @@ testTwoUsers = withAgentClients2 $ \a b -> do
     hasClients :: HasCallStack => AgentClient -> Int -> ExceptT AgentErrorType IO ()
     hasClients c n = liftIO $ M.size <$> readTVarIO (smpClients c) `shouldReturn` n
 
-getSMPAgentClient' :: AgentConfig -> InitialAgentServers -> FilePath -> IO AgentClient
-getSMPAgentClient' cfg' initServers dbPath = do
+getSMPAgentClient' :: Int -> AgentConfig -> InitialAgentServers -> FilePath -> IO AgentClient
+getSMPAgentClient' clientId cfg' initServers dbPath = do
   Right st <- liftIO $ createAgentStore dbPath "" False MCError
-  getSMPAgentClient cfg' initServers st False
+  getSMPAgentClient_ clientId cfg' initServers st False
 
 testServerMultipleIdentities :: HasCallStack => IO ()
 testServerMultipleIdentities =
@@ -2122,7 +2122,7 @@ testServerMultipleIdentities =
     -- this saves queue with second server identity
     Left (BROKER _ NETWORK) <- runExceptT $ joinConnection bob 1 True secondIdentityCReq "bob's connInfo" SMSubscribe
     disconnectAgentClient bob
-    bob' <- liftIO $ getSMPAgentClient' agentCfg initAgentServers testDB2
+    bob' <- liftIO $ getSMPAgentClient' 3 agentCfg initAgentServers testDB2
     subscribeConnection bob' aliceId
     exchangeGreetingsMsgId 6 alice bobId bob' aliceId
   where

tests/AgentTests/NotificationTests.hs

@@ -24,7 +24,7 @@ import NtfClient
 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.Env.SQLite (InitialAgentServers)
 import Simplex.Messaging.Agent.Protocol
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
@@ -98,7 +98,7 @@ notificationTests t =
 
 testNotificationToken :: APNSMockServer -> IO ()
 testNotificationToken APNSMockServer {apnsQ} = do
-  a <- getSMPAgentClient' agentCfg initAgentServers testDB
+  a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
   runRight_ $ do
     let tkn = DeviceToken PPApnsTest "abcd"
     NTRegistered <- registerNtfToken a tkn NMPeriodic
@@ -126,7 +126,7 @@ testNtfTokenRepeatRegistration :: APNSMockServer -> IO ()
 testNtfTokenRepeatRegistration APNSMockServer {apnsQ} = do
   -- setLogLevel LogError -- LogDebug
   -- withGlobalLogging logCfg $ do
-  a <- getSMPAgentClient' agentCfg initAgentServers testDB
+  a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
   runRight_ $ do
     let tkn = DeviceToken PPApnsTest "abcd"
     NTRegistered <- registerNtfToken a tkn NMPeriodic
@@ -150,8 +150,8 @@ testNtfTokenSecondRegistration :: APNSMockServer -> IO ()
 testNtfTokenSecondRegistration APNSMockServer {apnsQ} = do
   -- setLogLevel LogError -- LogDebug
   -- withGlobalLogging logCfg $ do
-  a <- getSMPAgentClient' agentCfg initAgentServers testDB
-  a' <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  a' <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   runRight_ $ do
     let tkn = DeviceToken PPApnsTest "abcd"
     NTRegistered <- registerNtfToken a tkn NMPeriodic
@@ -185,7 +185,7 @@ testNtfTokenSecondRegistration APNSMockServer {apnsQ} = do
 
 testNtfTokenServerRestart :: ATransport -> APNSMockServer -> IO ()
 testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
-  a <- getSMPAgentClient' agentCfg initAgentServers testDB
+  a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
   let tkn = DeviceToken PPApnsTest "abcd"
   ntfData <- withNtfServer t . runRight $ do
     NTRegistered <- registerNtfToken a tkn NMPeriodic
@@ -196,7 +196,7 @@ testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
   -- 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
-  a' <- getSMPAgentClient' agentCfg initAgentServers testDB
+  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
   withNtfServer t . runRight_ $ do
@@ -214,8 +214,8 @@ testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
 
 testNotificationSubscriptionExistingConnection :: APNSMockServer -> IO ()
 testNotificationSubscriptionExistingConnection APNSMockServer {apnsQ} = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   (bobId, aliceId, nonce, message) <- runRight $ do
     -- establish connection
     (bobId, qInfo) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
@@ -247,7 +247,7 @@ testNotificationSubscriptionExistingConnection APNSMockServer {apnsQ} = do
   Left (CMD PROHIBITED) <- runExceptT $ getNotificationMessage alice nonce message
 
   -- aliceNtf client doesn't have subscription and is allowed to get notification message
-  aliceNtf <- getSMPAgentClient' agentCfg initAgentServers testDB
+  aliceNtf <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
   runRight_ $ do
     (_, [SMPMsgMeta {msgFlags = MsgFlags True}]) <- getNotificationMessage aliceNtf nonce message
     pure ()
@@ -272,8 +272,8 @@ testNotificationSubscriptionExistingConnection APNSMockServer {apnsQ} = do
 
 testNotificationSubscriptionNewConnection :: APNSMockServer -> IO ()
 testNotificationSubscriptionNewConnection APNSMockServer {apnsQ} = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   runRight_ $ do
     -- alice registers notification token
     DeviceToken {} <- registerTestToken alice "abcd" NMInstant apnsQ
@@ -330,8 +330,8 @@ registerTestToken a token mode apnsQ = do
 
 testChangeNotificationsMode :: APNSMockServer -> IO ()
 testChangeNotificationsMode APNSMockServer {apnsQ} = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   runRight_ $ do
     -- establish connection
     (bobId, qInfo) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
@@ -396,8 +396,8 @@ testChangeNotificationsMode APNSMockServer {apnsQ} = do
 
 testChangeToken :: APNSMockServer -> IO ()
 testChangeToken APNSMockServer {apnsQ} = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   (aliceId, bobId) <- runRight $ do
     -- establish connection
     (bobId, qInfo) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
@@ -419,7 +419,7 @@ testChangeToken APNSMockServer {apnsQ} = do
     pure (aliceId, bobId)
   disconnectAgentClient alice
 
-  alice1 <- getSMPAgentClient' agentCfg initAgentServers testDB
+  alice1 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
   runRight_ $ do
     subscribeConnection alice1 bobId
     -- change notification token
@@ -441,8 +441,8 @@ testChangeToken APNSMockServer {apnsQ} = do
 
 testNotificationsStoreLog :: ATransport -> APNSMockServer -> IO ()
 testNotificationsStoreLog t APNSMockServer {apnsQ} = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   (aliceId, bobId) <- withNtfServerStoreLog t $ \threadId -> runRight $ do
     (aliceId, bobId) <- makeConnection alice bob
     _ <- registerTestToken alice "abcd" NMInstant apnsQ
@@ -469,8 +469,8 @@ testNotificationsStoreLog t APNSMockServer {apnsQ} = do
 
 testNotificationsSMPRestart :: ATransport -> APNSMockServer -> IO ()
 testNotificationsSMPRestart t APNSMockServer {apnsQ} = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   (aliceId, bobId) <- withSmpServerStoreLogOn t testPort $ \threadId -> runRight $ do
     (aliceId, bobId) <- makeConnection alice bob
     _ <- registerTestToken alice "abcd" NMInstant apnsQ
@@ -501,8 +501,8 @@ testNotificationsSMPRestart t APNSMockServer {apnsQ} = do
 
 testNotificationsSMPRestartBatch :: Int -> ATransport -> APNSMockServer -> IO ()
 testNotificationsSMPRestartBatch n t APNSMockServer {apnsQ} = do
-  a <- getSMPAgentClient' agentCfg initAgentServers2 testDB
-  b <- getSMPAgentClient' agentCfg initAgentServers2 testDB2
+  a <- getSMPAgentClient' 1 agentCfg initAgentServers2 testDB
+  b <- getSMPAgentClient' 2 agentCfg initAgentServers2 testDB2
   threadDelay 1000000
   conns <- runServers $ do
     conns <- replicateM (n :: Int) $ makeConnection a b
@@ -550,8 +550,8 @@ testNotificationsSMPRestartBatch n t APNSMockServer {apnsQ} = do
 
 testSwitchNotifications :: InitialAgentServers -> APNSMockServer -> IO ()
 testSwitchNotifications servers APNSMockServer {apnsQ} = do
-  a <- getSMPAgentClient' agentCfg servers testDB
-  b <- getSMPAgentClient' agentCfg {initialClientId = 1} servers testDB2
+  a <- getSMPAgentClient' 1 agentCfg servers testDB
+  b <- getSMPAgentClient' 2 agentCfg servers testDB2
   runRight_ $ do
     (aId, bId) <- makeConnection a b
     exchangeGreetingsMsgId 4 a bId b aId

tests/CoreTests/BatchingTests.hs

@@ -5,9 +5,8 @@ module CoreTests.BatchingTests (batchingTests) where
 import Control.Concurrent.STM
 import Control.Monad
 import Data.ByteString.Char8 (ByteString)
+import qualified Data.ByteString as B
 import qualified Data.List.NonEmpty as L
-import Simplex.Messaging.Builder (Builder)
-import qualified Simplex.Messaging.Builder as BB
 import Simplex.Messaging.Client
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Protocol
@@ -170,8 +169,8 @@ randomSENDCmd c len = do
   msg <- atomically $ C.randomBytes len g
   mkTransmission c (Just rpKey, sId, Cmd SSender $ SEND noMsgFlags msg)
 
-lenOk :: Builder -> Bool
-lenOk s = 0 < BB.length s && BB.length s <= smpBlockSize - 2
+lenOk :: ByteString -> Bool
+lenOk s = 0 < B.length s && B.length s <= smpBlockSize - 2
 
 lenOk1 :: TransportBatch r -> Bool
 lenOk1 = \case

tests/NtfClient.hs

@@ -136,7 +136,7 @@ ntfServerTest _ t = runNtfTest $ \h -> tPut' h t >> tGet' h
     tPut' :: THandle c -> (Maybe C.ASignature, ByteString, ByteString, smp) -> IO ()
     tPut' h@THandle {sessionId} (sig, corrId, queueId, smp) = do
       let t' = smpEncode (sessionId, corrId, queueId, smp)
-      [Right ()] <- tPut h Nothing [(sig, t')]
+      [Right ()] <- tPut h [(sig, t')]
       pure ()
     tGet' h = do
       [(Nothing, _, (CorrId corrId, qId, Right cmd))] <- tGet h

tests/SMPAgentClient.hs

@@ -1,6 +1,7 @@
 {-# LANGUAGE ConstraintKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE GADTs #-}
+{-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE NumericUnderscores #-}
 {-# LANGUAGE OverloadedLists #-}
 {-# LANGUAGE OverloadedStrings #-}
@@ -194,8 +195,8 @@ agentCfg =
     { tcpPort = agentTestPort,
       tbqSize = 4,
       -- database = testDB,
-      smpCfg = defaultClientConfig {qSize = 1, defaultTransport = (testPort, transport @TLS)},
-      ntfCfg = defaultClientConfig {qSize = 1, defaultTransport = (ntfTestPort, transport @TLS)},
+      smpCfg = defaultClientConfig {qSize = 1, defaultTransport = (testPort, transport @TLS), networkConfig},
+      ntfCfg = defaultClientConfig {qSize = 1, defaultTransport = (ntfTestPort, transport @TLS), networkConfig},
       reconnectInterval = defaultReconnectInterval {initialInterval = 50_000},
       xftpNotifyErrsOnRetry = False,
       ntfWorkerDelay = 100,
@@ -204,6 +205,8 @@ agentCfg =
       privateKeyFile = "tests/fixtures/server.key",
       certificateFile = "tests/fixtures/server.crt"
     }
+  where
+    networkConfig = defaultNetworkConfig {tcpConnectTimeout = 3_000_000, tcpTimeout = 2_000_000}
 
 type AgentTestMonad m = (MonadUnliftIO m, MonadRandom m, MonadFail m)
 

tests/SMPClient.hs

@@ -164,7 +164,7 @@ smpServerTest _ t = runSmpTest $ \h -> tPut' h t >> tGet' h
     tPut' :: THandle c -> (Maybe C.ASignature, ByteString, ByteString, smp) -> IO ()
     tPut' h@THandle {sessionId} (sig, corrId, queueId, smp) = do
       let t' = smpEncode (sessionId, corrId, queueId, smp)
-      [Right ()] <- tPut h Nothing [(sig, t')]
+      [Right ()] <- tPut h [(sig, t')]
       pure ()
     tGet' h = do
       [(Nothing, _, (CorrId corrId, qId, Right cmd))] <- tGet h

tests/ServerTests.hs

@@ -89,7 +89,7 @@ signSendRecv h@THandle {thVersion, sessionId} pk (corrId, qId, cmd) = do
 
 tPut1 :: Transport c => THandle c -> SentRawTransmission -> IO (Either TransportError ())
 tPut1 h t = do
-  [r] <- tPut h Nothing [t]
+  [r] <- tPut h [t]
   pure r
 
 tGet1 :: (ProtocolEncoding err cmd, Transport c, MonadIO m, MonadFail m) => THandle c -> m (SignedTransmission err cmd)

tests/XFTPAgent.hs

@@ -94,18 +94,18 @@ testXFTPAgentSendReceive :: HasCallStack => IO ()
 testXFTPAgentSendReceive = withXFTPServer $ do
   filePath <- createRandomFile
   -- send file, delete snd file internally
-  sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
+  sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
   (rfd1, rfd2) <- runRight $ do
     (sfId, _, rfd1, rfd2) <- testSend sndr filePath
     xftpDeleteSndFileInternal sndr sfId
     pure (rfd1, rfd2)
 
   -- receive file, delete rcv file
-  testReceiveDelete rfd1 filePath
-  testReceiveDelete rfd2 filePath
+  testReceiveDelete 2 rfd1 filePath
+  testReceiveDelete 3 rfd2 filePath
   where
-    testReceiveDelete rfd originalFilePath = do
-      rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
+    testReceiveDelete clientId rfd originalFilePath = do
+      rcp <- getSMPAgentClient' clientId agentCfg initAgentServers testDB2
       runRight_ $ do
         rfId <- testReceive rcp rfd originalFilePath
         xftpDeleteRcvFile rcp rfId
@@ -118,17 +118,17 @@ testXFTPAgentSendReceiveEncrypted = withXFTPServer $ do
   s <- LB.readFile filePath
   file <- atomically $ CryptoFile (senderFiles </> "encrypted_testfile") . Just <$> CF.randomArgs g
   runRight_ $ CF.writeFile file s
-  sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
+  sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
   (rfd1, rfd2) <- runRight $ do
     (sfId, _, rfd1, rfd2) <- testSendCF sndr file
     xftpDeleteSndFileInternal sndr sfId
     pure (rfd1, rfd2)
   -- receive file, delete rcv file
-  testReceiveDelete rfd1 filePath g
-  testReceiveDelete rfd2 filePath g
+  testReceiveDelete 2 rfd1 filePath g
+  testReceiveDelete 3 rfd2 filePath g
   where
-    testReceiveDelete rfd originalFilePath g = do
-      rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
+    testReceiveDelete clientId rfd originalFilePath g = do
+      rcp <- getSMPAgentClient' clientId agentCfg initAgentServers testDB2
       cfArgs <- atomically $ Just <$> CF.randomArgs g
       runRight_ $ do
         rfId <- testReceiveCF rcp rfd cfArgs originalFilePath
@@ -188,13 +188,13 @@ testXFTPAgentReceiveRestore = withGlobalLogging logCfgNoLogs $ do
 
   rfd <- withXFTPServerStoreLogOn $ \_ -> do
     -- send file
-    sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
+    sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
     runRight $ do
       (_, _, rfd, _) <- testSend sndr filePath
       pure rfd
 
   -- receive file - should not succeed with server down
-  rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  rcp <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   rfId <- runRight $ do
     xftpStartWorkers rcp (Just recipientFiles)
     rfId <- xftpReceiveFile rcp 1 rfd Nothing
@@ -208,7 +208,7 @@ testXFTPAgentReceiveRestore = withGlobalLogging logCfgNoLogs $ do
 
   withXFTPServerStoreLogOn $ \_ -> do
     -- receive file - should start downloading with server up
-    rcp' <- getSMPAgentClient' agentCfg initAgentServers testDB2
+    rcp' <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
     runRight_ $ xftpStartWorkers rcp' (Just recipientFiles)
     ("", rfId', RFPROG _ _) <- rfGet rcp'
     liftIO $ rfId' `shouldBe` rfId
@@ -218,7 +218,7 @@ testXFTPAgentReceiveRestore = withGlobalLogging logCfgNoLogs $ do
 
   withXFTPServerStoreLogOn $ \_ -> do
     -- receive file - should continue downloading with server up
-    rcp' <- getSMPAgentClient' agentCfg initAgentServers testDB2
+    rcp' <- getSMPAgentClient' 4 agentCfg initAgentServers testDB2
     runRight_ $ xftpStartWorkers rcp' (Just recipientFiles)
     rfProgress rcp' $ mb 18
     ("", rfId', RFDONE path) <- rfGet rcp'
@@ -236,13 +236,13 @@ testXFTPAgentReceiveCleanup = withGlobalLogging logCfgNoLogs $ do
 
   rfd <- withXFTPServerStoreLogOn $ \_ -> do
     -- send file
-    sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
+    sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
     runRight $ do
       (_, _, rfd, _) <- testSend sndr filePath
       pure rfd
 
   -- receive file - should not succeed with server down
-  rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  rcp <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   rfId <- runRight $ do
     xftpStartWorkers rcp (Just recipientFiles)
     rfId <- xftpReceiveFile rcp 1 rfd Nothing
@@ -256,7 +256,7 @@ testXFTPAgentReceiveCleanup = withGlobalLogging logCfgNoLogs $ do
 
   withXFTPServerThreadOn $ \_ -> do
     -- receive file - should fail with AUTH error
-    rcp' <- getSMPAgentClient' agentCfg initAgentServers testDB2
+    rcp' <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
     runRight_ $ xftpStartWorkers rcp' (Just recipientFiles)
     ("", rfId', RFERR (INTERNAL "XFTP {xftpErr = AUTH}")) <- rfGet rcp'
     rfId' `shouldBe` rfId
@@ -269,7 +269,7 @@ testXFTPAgentSendRestore = withGlobalLogging logCfgNoLogs $ do
   filePath <- createRandomFile
 
   -- send file - should not succeed with server down
-  sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
+  sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
   sfId <- runRight $ do
     xftpStartWorkers sndr (Just senderFiles)
     sfId <- xftpSendFile sndr 1 (CF.plain filePath) 2
@@ -286,7 +286,7 @@ testXFTPAgentSendRestore = withGlobalLogging logCfgNoLogs $ do
 
   withXFTPServerStoreLogOn $ \_ -> do
     -- send file - should start uploading with server up
-    sndr' <- getSMPAgentClient' agentCfg initAgentServers testDB
+    sndr' <- getSMPAgentClient' 2 agentCfg initAgentServers testDB
     runRight_ $ xftpStartWorkers sndr' (Just senderFiles)
     ("", sfId', SFPROG _ _) <- sfGet sndr'
     liftIO $ sfId' `shouldBe` sfId
@@ -296,7 +296,7 @@ testXFTPAgentSendRestore = withGlobalLogging logCfgNoLogs $ do
 
   withXFTPServerStoreLogOn $ \_ -> do
     -- send file - should continue uploading with server up
-    sndr' <- getSMPAgentClient' agentCfg initAgentServers testDB
+    sndr' <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
     runRight_ $ xftpStartWorkers sndr' (Just senderFiles)
     sfProgress sndr' $ mb 18
     ("", sfId', SFDONE _sndDescr [rfd1, _rfd2]) <- sfGet sndr'
@@ -308,7 +308,7 @@ testXFTPAgentSendRestore = withGlobalLogging logCfgNoLogs $ do
     doesFileExist encPath `shouldReturn` False
 
     -- receive file
-    rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
+    rcp <- getSMPAgentClient' 4 agentCfg initAgentServers testDB2
     runRight_ . void $
       testReceive rcp rfd1 filePath
 
@@ -318,7 +318,7 @@ testXFTPAgentSendCleanup = withGlobalLogging logCfgNoLogs $ do
 
   sfId <- withXFTPServerStoreLogOn $ \_ -> do
     -- send file
-    sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
+    sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
     sfId <- runRight $ do
       xftpStartWorkers sndr (Just senderFiles)
       sfId <- xftpSendFile sndr 1 (CF.plain filePath) 2
@@ -339,7 +339,7 @@ testXFTPAgentSendCleanup = withGlobalLogging logCfgNoLogs $ do
 
   withXFTPServerThreadOn $ \_ -> do
     -- send file - should fail with AUTH error
-    sndr' <- getSMPAgentClient' agentCfg initAgentServers testDB
+    sndr' <- getSMPAgentClient' 2 agentCfg initAgentServers testDB
     runRight_ $ xftpStartWorkers sndr' (Just senderFiles)
     ("", sfId', SFERR (INTERNAL "XFTP {xftpErr = AUTH}")) <- sfGet sndr'
     sfId' `shouldBe` sfId
@@ -354,11 +354,11 @@ testXFTPAgentDelete = withGlobalLogging logCfgNoLogs $
     filePath <- createRandomFile
 
     -- send file
-    sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
+    sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
     (sfId, sndDescr, rfd1, rfd2) <- runRight $ testSend sndr filePath
 
     -- receive file
-    rcp1 <- getSMPAgentClient' agentCfg initAgentServers testDB2
+    rcp1 <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
     runRight_ . void $
       testReceive rcp1 rfd1 filePath
 
@@ -376,7 +376,7 @@ testXFTPAgentDelete = withGlobalLogging logCfgNoLogs $
     length <$> listDirectory xftpServerFiles `shouldReturn` 0
 
     -- receive file - should fail with AUTH error
-    rcp2 <- getSMPAgentClient' agentCfg initAgentServers testDB2
+    rcp2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
     runRight $ do
       xftpStartWorkers rcp2 (Just recipientFiles)
       rfId <- xftpReceiveFile rcp2 1 rfd2 Nothing
@@ -389,11 +389,11 @@ testXFTPAgentDeleteRestore = withGlobalLogging logCfgNoLogs $ do
 
   (sfId, sndDescr, rfd2) <- withXFTPServerStoreLogOn $ \_ -> do
     -- send file
-    sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
+    sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
     (sfId, sndDescr, rfd1, rfd2) <- runRight $ testSend sndr filePath
 
     -- receive file
-    rcp1 <- getSMPAgentClient' agentCfg initAgentServers testDB2
+    rcp1 <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
     runRight_ . void $
       testReceive rcp1 rfd1 filePath
     disconnectAgentClient rcp1
@@ -401,7 +401,7 @@ testXFTPAgentDeleteRestore = withGlobalLogging logCfgNoLogs $ do
     pure (sfId, sndDescr, rfd2)
 
   -- delete file - should not succeed with server down
-  sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
+  sndr <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
   runRight $ do
     xftpStartWorkers sndr (Just senderFiles)
     xftpDeleteSndFileRemote sndr 1 sfId sndDescr
@@ -413,14 +413,14 @@ testXFTPAgentDeleteRestore = withGlobalLogging logCfgNoLogs $ do
 
   withXFTPServerStoreLogOn $ \_ -> do
     -- delete file - should succeed with server up
-    sndr' <- getSMPAgentClient' agentCfg initAgentServers testDB
+    sndr' <- getSMPAgentClient' 4 agentCfg initAgentServers testDB
     runRight_ $ xftpStartWorkers sndr' (Just senderFiles)
 
     threadDelay 1000000
     length <$> listDirectory xftpServerFiles `shouldReturn` 0
 
     -- receive file - should fail with AUTH error
-    rcp2 <- getSMPAgentClient' agentCfg initAgentServers testDB3
+    rcp2 <- getSMPAgentClient' 5 agentCfg initAgentServers testDB3
     runRight $ do
       xftpStartWorkers rcp2 (Just recipientFiles)
       rfId <- xftpReceiveFile rcp2 1 rfd2 Nothing
@@ -433,11 +433,11 @@ testXFTPAgentDeleteOnServer = withGlobalLogging logCfgNoLogs $
     filePath1 <- createRandomFile' "testfile1"
 
     -- send file 1
-    sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
+    sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
     (_, _, rfd1_1, rfd1_2) <- runRight $ testSend sndr filePath1
 
     -- receive file 1 successfully
-    rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
+    rcp <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
     runRight_ . void $
       testReceive rcp rfd1_1 filePath1
 
@@ -471,11 +471,11 @@ testXFTPAgentExpiredOnServer = withGlobalLogging logCfgNoLogs $ do
     filePath1 <- createRandomFile' "testfile1"
 
     -- send file 1
-    sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
+    sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
     (_, _, rfd1_1, rfd1_2) <- runRight $ testSend sndr filePath1
 
     -- receive file 1 successfully
-    rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
+    rcp <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
     runRight_ . void $
       testReceive rcp rfd1_1 filePath1
 
@@ -509,7 +509,7 @@ testXFTPAgentRequestAdditionalRecipientIDs = withXFTPServer $ do
   filePath <- createRandomFile
 
   -- send file
-  sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
+  sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
   rfds <- runRight $ do
     xftpStartWorkers sndr (Just senderFiles)
     sfId <- xftpSendFile sndr 1 (CF.plain filePath) 500
@@ -522,7 +522,7 @@ testXFTPAgentRequestAdditionalRecipientIDs = withXFTPServer $ do
 
   -- receive file using different descriptions
   -- ! revise number of recipients and indexes if xftpMaxRecipientsPerRequest is changed
-  rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  rcp <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   runRight_ $ do
     void $ testReceive rcp (head rfds) filePath
     void $ testReceive rcp (rfds !! 99) filePath
@@ -532,5 +532,5 @@ testXFTPAgentRequestAdditionalRecipientIDs = withXFTPServer $ do
 testXFTPServerTest :: HasCallStack => Maybe BasicAuth -> XFTPServerWithAuth -> IO (Maybe ProtocolTestFailure)
 testXFTPServerTest newFileBasicAuth srv =
   withXFTPServerCfg testXFTPServerConfig {newFileBasicAuth, xftpPort = xftpTestPort2} $ \_ -> do
-    a <- getSMPAgentClient' agentCfg initAgentServers testDB -- initially passed server is not running
+    a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB -- initially passed server is not running
     runRight $ testProtocolServer a 1 srv