5.5.1.2

agent: only retry connecting SMP clients when there are pending subscriptions (#981 )
* agent: only retry connecting SMP clients when there are pending subscriptions * fix * remove retry on creating clients * simplify
2026-08-30 20:38:23 +00:00 · 2024-02-02 08:19:02 +00:00 · 2024-02-01 16:17:37 +00:00 · 2024-02-01 11:02:31 +00:00 · 2024-02-01 10:55:50 +00:00 · 2024-01-30 11:36:07 +00:00
41 changed files with 1145 additions and 581 deletions
@@ -1,3 +1,40 @@
+# 5.5.1
+
+SMP servers:
+- do not keep stats file open
+- additional stats about currently stored messages
+
+Agent:
+- support multiple notification servers (only one can be used at a time).
+- expire messages after "quota exceeded" error after 7 days (instead of 21 days previously).
+- stabilize message delivery, remove unnecessary subscription retries and traffic.
+- improve database performance for message delivery.
+- fix sockets/memory leak - a very old bug "activated" by improvements in v5.5.0.
+
+# 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
@@ -1,5 +1,5 @@
 name: simplexmq
-version: 5.5.0.4
+version: 5.5.1.2
 synopsis: SimpleXMQ message broker
 description: |
  This package includes <./docs/Simplex-Messaging-Server.html server>,
@@ -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
@@ -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
@@ -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:
@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack

 name:           simplexmq
-version:        5.5.0.4
+version:        5.5.1.2
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                <./docs/Simplex-Messaging-Client.html client> and
@@ -101,7 +101,7 @@ library
      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230829_crypto_files
      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20231222_command_created_at
      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20231225_failed_work_items
-      Simplex.Messaging.Agent.TAsyncs
+      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240121_message_delivery_indexes
      Simplex.Messaging.Agent.TRcvQueues
      Simplex.Messaging.Client
      Simplex.Messaging.Client.Agent
@@ -48,8 +48,7 @@ import Simplex.Messaging.Protocol
    encodeTransmission,
    messageTagP,
    tDecodeParseValidate,
-    tEncode,
-    tEncodeBatch,
+    tEncodeBatch1,
    tParse,
  )
 import Simplex.Messaging.Transport (SessionId, TransportError (..))
@@ -403,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 ByteString
-xftpEncodeBatch1 (sig, t) =
-  let t' = tEncodeBatch 1 . smpEncode . Large $ 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
@@ -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.1.0"

 xftpServerCLI :: FilePath -> FilePath -> IO ()
 xftpServerCLI cfgPath logPath = do
@@ -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 =
@@ -1136,7 +1142,7 @@ submitPendingMsg c cData sq = do

 runSmpQueueMsgDelivery :: forall m. AgentMonad m => AgentClient -> ConnData -> SndQueue -> (Worker, TMVar ()) -> m ()
 runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, duplexHandshake} sq (Worker {doWork}, qLock) = do
-  ri <- asks $ messageRetryInterval . config
+  AgentConfig {messageRetryInterval = ri, messageTimeout, helloTimeout, quotaExceededTimeout} <- asks config
  forever $ do
    atomically $ endAgentOperation c AOSndNetwork
    waitForWork doWork
@@ -1160,7 +1166,9 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl
                SMP SMP.QUOTA -> case msgType of
                  AM_CONN_INFO -> connError msgId NOT_AVAILABLE
                  AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE
-                  _ -> retrySndMsg RISlow
+                  _ -> do
+                    expireTs <- addUTCTime (-quotaExceededTimeout) <$> liftIO getCurrentTime
+                    if internalTs < expireTs then notifyDelMsgs msgId e expireTs else retrySndMsg RISlow
                SMP SMP.AUTH -> case msgType of
                  AM_CONN_INFO -> connError msgId NOT_AVAILABLE
                  AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE
@@ -1169,8 +1177,11 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl
                    -- in duplexHandshake mode (v2) HELLO is only sent once, without retrying,
                    -- because the queue must be secured by the time the confirmation or the first HELLO is received
                    | duplexHandshake == Just True -> connErr
-                    | otherwise ->
-                        ifM (msgExpired helloTimeout) connErr (retrySndMsg RIFast)
+                    -- otherwise branch is not used in clients with v2+ of agent protocol (since June 2022)
+                    -- TODO remove in v6
+                    | otherwise -> do
+                        expireTs <- addUTCTime (-helloTimeout) <$> liftIO getCurrentTime
+                        if internalTs < expireTs then connErr else retrySndMsg RIFast
                    where
                      connErr = case rq_ of
                        -- party initiating connection
@@ -1190,14 +1201,11 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl
                  -- for other operations BROKER HOST is treated as a permanent error (e.g., when connecting to the server),
                  -- the message sending would be retried
                  | temporaryOrHostError e -> do
-                      let timeoutSel = if msgType == AM_HELLO_ then helloTimeout else messageTimeout
-                      ifM (msgExpired timeoutSel) (notifyDel msgId err) (retrySndMsg RIFast)
+                      let msgTimeout = if msgType == AM_HELLO_ then helloTimeout else messageTimeout
+                      expireTs <- addUTCTime (-msgTimeout) <$> liftIO getCurrentTime
+                      if internalTs < expireTs then notifyDelMsgs msgId e expireTs else retrySndMsg RIFast
                  | otherwise -> notifyDel msgId err
              where
-                msgExpired timeoutSel = do
-                  msgTimeout <- asks $ timeoutSel . config
-                  currentTime <- liftIO getCurrentTime
-                  pure $ diffUTCTime currentTime internalTs > msgTimeout
                retrySndMsg riMode = do
                  withStore' c $ \db -> updatePendingMsgRIState db connId msgId riState
                  retrySndOp c $ loop riMode
@@ -1273,6 +1281,13 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl
                    when (isJust rq_) $ removeConfirmations db connId
                  unless (duplexHandshake == Just True) . void $ enqueueMessage c cData sq SMP.noMsgFlags HELLO
  where
+    notifyDelMsgs :: InternalId -> AgentErrorType -> UTCTime -> m ()
+    notifyDelMsgs msgId err expireTs = do
+      notifyDel msgId $ MERR (unId msgId) err
+      msgIds_ <- withStore' c $ \db -> getExpiredSndMessages db connId sq expireTs
+      forM_ (L.nonEmpty msgIds_) $ \msgIds -> do
+        notify $ MERRS (L.map unId msgIds) err
+        withStore' c $ \db -> forM_ msgIds $ \msgId' -> deleteSndMsgDelivery db connId sq msgId' False `catchAll_` pure ()
    delMsg :: InternalId -> m ()
    delMsg = delMsgKeep False
    delMsgKeep :: Bool -> InternalId -> m ()
@@ -1792,12 +1807,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 +2056,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
@@ -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,8 @@ 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
@@ -165,7 +172,6 @@ import Simplex.Messaging.Agent.RetryInterval
 import Simplex.Messaging.Agent.Store
 import Simplex.Messaging.Agent.Store.SQLite (SQLiteStore (..), withTransaction)
 import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
-import Simplex.Messaging.Agent.TAsyncs
 import Simplex.Messaging.Agent.TRcvQueues (TRcvQueues (getRcvQueues))
 import qualified Simplex.Messaging.Agent.TRcvQueues as RQ
 import Simplex.Messaging.Client
@@ -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,10 +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,
-    asyncClients :: TAsyncs,
+    -- smpSubWorkers for SMP servers sessions
+    smpSubWorkers :: TMap SMPTransportSession (SessionVar (Async ())),
    agentStats :: TMap AgentStatsKey (TVar Int),
    clientId :: Int,
    agentEnv :: Env
@@ -288,11 +297,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 +370,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 +383,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 +415,8 @@ newAgentClient InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv = do
  connLocks <- TM.empty
  invLocks <- TM.empty
  deleteLock <- createLock
-  reconnectLocks <- TM.empty
-  reconnections <- newTAsyncs
-  asyncClients <- newTAsyncs
+  smpSubWorkers <- TM.empty
  agentStats <- TM.empty
-  clientId <- stateTVar (clientCounter agentEnv) $ \i -> let i' = i + 1 in (i', i')
  return
    AgentClient
      { active,
@@ -443,9 +448,7 @@ newAgentClient InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv = do
        connLocks,
        invLocks,
        deleteLock,
-        reconnectLocks,
-        reconnections,
-        asyncClients,
+        smpSubWorkers,
        agentStats,
        clientId,
        agentEnv
@@ -496,26 +499,25 @@ 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)
-    >>= either newClient (waitForProtocolClient c tSess)
+  v <- atomically (getTSessVar c tSess smpClients)
+  either newClient (waitForProtocolClient c tSess) v
+    `catchAgentError` \e -> resubscribeSMPSession c tSess >> throwError e
  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
+    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 +531,55 @@ 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 getWorkerVar >>= mapM_ (either newSubWorker (\_ -> pure ()))
  where
-    tryReconnectSMPClient aId = do
+    getWorkerVar =
+      ifM
+        (null <$> getPending)
+        (pure Nothing) -- prevent race with cleanup and adding pending queues in another call
+        (Just <$> getTSessVar c tSess smpSubWorkers)
+    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 getPending
+        forM_ (L.nonEmpty pending) $ \qs -> do
+          void . tryAgentError' $ reconnectSMPClient timeoutCounts c tSess qs
+          loop
+    getPending = RQ.getSessQueues tSess $ pendingSubs c
+    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 +588,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 ())
+      (newProtocolClient c tSess ntfClients connectClient)
      (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,78 +618,73 @@ 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 ())
+      (newProtocolClient c tSess xftpClients connectClient)
      (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) ->
-  (AgentClient -> TransportSession msg -> m ()) ->
+  (ClientVar msg -> m (Client msg)) ->
  ClientVar msg ->
  m (Client msg)
-newProtocolClient c tSess@(userId, srv, entityId_) clients connectClient reconnectClient clientVar = tryConnectClient pure tryConnectAsync
-  where
-    tryConnectClient :: (Client msg -> m a) -> m () -> m a
-    tryConnectClient successAction retryAction =
-      tryError connectClient >>= \r -> case r of
-        Right client -> do
-          logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv <> " (user " <> bshow userId <> maybe "" (" for entity " <>) entityId_ <> ")"
-          atomically $ putTMVar clientVar r
-          liftIO $ incClientStat c userId client "CLIENT" "OK"
-          atomically $ writeTBQueue (subQ c) ("", "", APC SAENone $ hostEvent CONNECT client)
-          successAction client
-        Left e -> do
-          liftIO $ incServerStat c userId srv "CLIENT" $ strEncode e
-          if temporaryAgentError e
-            then retryAction
-            else atomically $ do
-              putTMVar clientVar (Left e)
-              TM.delete 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
-      atomically . removeAsyncAction aId $ asyncClients c
+newProtocolClient c tSess@(userId, srv, entityId_) clients connectClient v =
+  tryAgentError (connectClient v) >>= \case
+    Right client -> do
+      logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv <> " (user " <> bshow userId <> maybe "" (" for entity " <>) entityId_ <> ")"
+      atomically $ putTMVar (sessionVar v) (Right client)
+      liftIO $ incClientStat c userId client "CLIENT" "OK"
+      atomically $ writeTBQueue (subQ c) ("", "", APC SAENone $ hostEvent CONNECT client)
+      pure client
+    Left e -> do
+      liftIO $ incServerStat c userId srv "CLIENT" $ strEncode e
+      atomically $ do
+        removeTSessVar v tSess clients
+        putTMVar (sessionVar v) (Left e)
+      throwError e -- signal error to caller

 hostEvent :: forall err msg. (ProtocolTypeI (ProtoType msg), ProtocolServerClient err msg) => (AProtocolType -> TransportHost -> ACommand 'Agent 'AENone) -> Client msg -> ACommand 'Agent 'AENone
 hostEvent event = event (AProtocolType $ protocolTypeI @(ProtoType msg)) . clientTransportHost
@@ -687,8 +701,7 @@ closeAgentClient c = liftIO $ do
  closeProtocolServerClients c smpClients
  closeProtocolServerClients c ntfClients
  closeProtocolServerClients c xftpClients
-  cancelActions . actions $ reconnections c
-  cancelActions . actions $ asyncClients c
+  atomically (swapTVar (smpSubWorkers c) M.empty) >>= mapM_ cancelReconnect
  clearWorkers smpDeliveryWorkers >>= mapM_ (cancelWorker . fst)
  clearWorkers asyncCmdWorkers >>= mapM_ cancelWorker
  clear connCmdsQueued
@@ -701,6 +714,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 +743,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 ()

@@ -738,9 +753,6 @@ closeXFTPServerClient :: AgentMonad' m => AgentClient -> UserId -> XFTPServer ->
 closeXFTPServerClient c userId server (FileDigest chunkDigest) =
  mkTransportSession c userId server chunkDigest >>= liftIO . closeClient c xftpClients

-cancelActions :: (Foldable f, Monoid (f (Async ()))) => TVar (f (Async ())) -> IO ()
-cancelActions as = atomically (swapTVar as mempty) >>= mapM_ (forkIO . uninterruptibleCancel)
-
 withConnLock :: MonadUnliftIO m => AgentClient -> ConnId -> String -> m a -> m a
 withConnLock _ "" _ = id
 withConnLock AgentClient {connLocks} connId name = withLockMap_ connLocks connId name
@@ -991,7 +1003,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 +1021,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 +1041,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 +1142,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 +1156,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 +1183,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 +1539,117 @@ getAgentSubscriptions c = do
    enc :: StrEncoding a => a -> Text
    enc = decodeLatin1 . strEncode

+data AgentWorkersDetails = AgentWorkersDetails
+  { smpClients_ :: [Text],
+    ntfClients_ :: [Text],
+    xftpClients_ :: [Text],
+    smpDeliveryWorkers_ :: Map Text WorkersDetails,
+    asyncCmdWorkers_ :: Map Text WorkersDetails,
+    smpSubWorkers_ :: [Text],
+    ntfWorkers_ :: Map Text WorkersDetails,
+    ntfSMPWorkers_ :: Map Text WorkersDetails,
+    xftpRcvWorkers_ :: Map Text WorkersDetails,
+    xftpSndWorkers_ :: Map Text WorkersDetails,
+    xftpDelWorkers_ :: Map Text WorkersDetails
+  }
+  deriving (Show)
+
+data WorkersDetails = WorkersDetails
+  { restarts :: Int,
+    hasWork :: Bool,
+    hasAction :: Bool
+  }
+  deriving (Show)
+
+getAgentWorkersDetails :: MonadIO m => AgentClient -> m AgentWorkersDetails
+getAgentWorkersDetails AgentClient {smpClients, ntfClients, xftpClients, smpDeliveryWorkers, asyncCmdWorkers, smpSubWorkers, 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
+  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_,
+        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 WorkersDetails)
+    workerStats ws = fmap M.fromList . forM (M.toList ws) $ \(qa, Worker {restarts, doWork, action}) -> do
+      RestartCount {restartCount} <- readTVarIO restarts
+      hasWork <- atomically $ not <$> isEmptyTMVar doWork
+      hasAction <- atomically $ not <$> isEmptyTMVar action
+      pure (textKey qa, WorkersDetails {restarts = restartCount, hasWork, hasAction})
+    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,
+    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, 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
+  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,
+        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 +1659,8 @@ $(J.deriveJSON defaultJSON ''ProtocolTestFailure)
 $(J.deriveJSON defaultJSON ''SubInfo)

 $(J.deriveJSON defaultJSON ''SubscriptionsInfo)
+
+$(J.deriveJSON defaultJSON ''WorkersDetails)
+$(J.deriveJSON defaultJSON {J.fieldLabelModifier = takeWhile (/= '_')} ''AgentWorkersDetails)
+
+$(J.deriveJSON defaultJSON ''AgentWorkersSummary)
@@ -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
@@ -90,6 +92,7 @@ data AgentConfig = AgentConfig
    messageRetryInterval :: RetryInterval2,
    messageTimeout :: NominalDiffTime,
    helloTimeout :: NominalDiffTime,
+    quotaExceededTimeout :: NominalDiffTime,
    initialCleanupDelay :: Int64,
    cleanupInterval :: Int64,
    cleanupStepInterval :: Int,
@@ -112,8 +115,7 @@ data AgentConfig = AgentConfig
    certificateFile :: FilePath,
    e2eEncryptVRange :: VersionRange,
    smpAgentVRange :: VersionRange,
-    smpClientVRange :: VersionRange,
-    initialClientId :: Int
+    smpClientVRange :: VersionRange
  }

 defaultReconnectInterval :: RetryInterval
@@ -134,13 +136,10 @@ defaultMessageRetryInterval =
            maxInterval = 60_000000
          },
      riSlow =
-        -- TODO: these timeouts can be increased in v5.0 once most clients are updated
-        -- to resume sending on QCONT messages.
-        -- After that local message expiration period should be also increased.
        RetryInterval
-          { initialInterval = 60_000000,
+          { initialInterval = 180_000000, -- 3 minutes
            increaseAfter = 60_000000,
-            maxInterval = 3600_000000 -- 1 hour
+            maxInterval = 3 * 3600_000000 -- 3 hours
          }
    }

@@ -158,6 +157,7 @@ defaultAgentConfig =
      messageRetryInterval = defaultMessageRetryInterval,
      messageTimeout = 2 * nominalDay,
      helloTimeout = 2 * nominalDay,
+      quotaExceededTimeout = 7 * nominalDay,
      initialCleanupDelay = 30 * 1000000, -- 30 seconds
      cleanupInterval = 30 * 60 * 1000000, -- 30 minutes
      cleanupStepInterval = 200000, -- 200ms
@@ -184,15 +184,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 +198,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 +247,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 #-}
@@ -72,7 +72,7 @@ processNtfSub c (connId, cmd) = do
      logInfo $ "processNtfSub, NSCCreate - a = " <> tshow a
      case a of
        Nothing -> do
-          withNtfServer c $ \ntfServer -> do
+          withTokenServer $ \ntfServer -> do
            case clientNtfCreds of
              Just ClientNtfCreds {notifierId} -> do
                let newSub = newNtfSubscription connId smpServer (Just notifierId) ntfServer NASKey
@@ -99,7 +99,7 @@ processNtfSub c (connId, cmd) = do
                | isDeleteNtfSubAction action -> do
                    if ntfSubStatus == NASNew || ntfSubStatus == NASOff || ntfSubStatus == NASDeleted
                      then resetSubscription
-                      else withNtfServer c $ \ntfServer -> do
+                      else withTokenServer $ \ntfServer -> do
                        withStore' c $ \db -> supervisorUpdateNtfSub db sub {ntfServer} (NtfSubNTFAction NSACreate)
                        void $ getNtfNTFWorker True c ntfServer
                | otherwise -> case action of
@@ -111,7 +111,7 @@ processNtfSub c (connId, cmd) = do
              void $ getNtfNTFWorker True c subNtfServer
            resetSubscription :: m ()
            resetSubscription =
-              withNtfServer c $ \ntfServer -> do
+              withTokenServer $ \ntfServer -> do
                let sub' = sub {ntfQueueId = Nothing, ntfServer, ntfSubId = Nothing, ntfSubStatus = NASNew}
                withStore' c $ \db -> supervisorUpdateNtfSub db sub' (NtfSubSMPAction NSASmpKey)
                void $ getNtfSMPWorker True c smpServer
@@ -143,8 +143,8 @@ getNtfSMPWorker hasWork c server = do
  ws <- asks $ ntfSMPWorkers . ntfSupervisor
  getAgentWorker "ntf_smp" hasWork c server ws $ runNtfSMPWorker c server

-withNtfServer :: AgentMonad' m => AgentClient -> (NtfServer -> m ()) -> m ()
-withNtfServer c action = getNtfServer c >>= mapM_ action
+withTokenServer :: AgentMonad' m => (NtfServer -> m ()) -> m ()
+withTokenServer action = getNtfToken >>= mapM_ (\NtfToken {ntfServer} -> action ntfServer)

 runNtfWorker :: forall m. AgentMonad m => AgentClient -> NtfServer -> Worker -> m ()
 runNtfWorker c srv Worker {doWork} = do
@@ -337,6 +337,7 @@ data ACommand (p :: AParty) (e :: AEntity) where
  MID :: AgentMsgId -> ACommand Agent AEConn
  SENT :: AgentMsgId -> ACommand Agent AEConn
  MERR :: AgentMsgId -> AgentErrorType -> ACommand Agent AEConn
+  MERRS :: NonEmpty AgentMsgId -> AgentErrorType -> ACommand Agent AEConn
  MSG :: MsgMeta -> MsgFlags -> MsgBody -> ACommand Agent AEConn
  MSGNTF :: SMPMsgMeta -> ACommand Agent AEConn
  ACK :: AgentMsgId -> Maybe MsgReceiptInfo -> ACommand Client AEConn
@@ -398,6 +399,7 @@ data ACommandTag (p :: AParty) (e :: AEntity) where
  MID_ :: ACommandTag Agent AEConn
  SENT_ :: ACommandTag Agent AEConn
  MERR_ :: ACommandTag Agent AEConn
+  MERRS_ :: ACommandTag Agent AEConn
  MSG_ :: ACommandTag Agent AEConn
  MSGNTF_ :: ACommandTag Agent AEConn
  ACK_ :: ACommandTag Client AEConn
@@ -452,6 +454,7 @@ aCommandTag = \case
  MID _ -> MID_
  SENT _ -> SENT_
  MERR {} -> MERR_
+  MERRS {} -> MERRS_
  MSG {} -> MSG_
  MSGNTF {} -> MSGNTF_
  ACK {} -> ACK_
@@ -1611,6 +1614,7 @@ instance StrEncoding ACmdTag where
      "MID" -> ct MID_
      "SENT" -> ct SENT_
      "MERR" -> ct MERR_
+      "MERRS" -> ct MERRS_
      "MSG" -> ct MSG_
      "MSGNTF" -> ct MSGNTF_
      "ACK" -> t ACK_
@@ -1667,6 +1671,7 @@ instance (APartyI p, AEntityI e) => StrEncoding (ACommandTag p e) where
    MID_ -> "MID"
    SENT_ -> "SENT"
    MERR_ -> "MERR"
+    MERRS_ -> "MERRS"
    MSG_ -> "MSG"
    MSGNTF_ -> "MSGNTF"
    ACK_ -> "ACK"
@@ -1736,6 +1741,7 @@ commandP binaryP =
          MID_ -> s (MID <$> A.decimal)
          SENT_ -> s (SENT <$> A.decimal)
          MERR_ -> s (MERR <$> A.decimal <* A.space <*> strP)
+          MERRS_ -> s (MERRS <$> strP_ <*> strP)
          MSG_ -> s (MSG <$> strP <* A.space <*> smpP <* A.space <*> binaryP)
          MSGNTF_ -> s (MSGNTF <$> strP)
          RCVD_ -> s (RCVD <$> strP <* A.space <*> strP)
@@ -1788,12 +1794,13 @@ serializeCommand = \case
  SWITCH dir phase srvs -> s (SWITCH_, dir, phase, srvs)
  RSYNC rrState cryptoErr cstats -> s (RSYNC_, rrState, cryptoErr, cstats)
  SEND msgFlags msgBody -> B.unwords [s SEND_, smpEncode msgFlags, serializeBinary msgBody]
-  MID mId -> s (MID_, Str $ bshow mId)
-  SENT mId -> s (SENT_, Str $ bshow mId)
-  MERR mId e -> s (MERR_, Str $ bshow mId, e)
+  MID mId -> s (MID_, mId)
+  SENT mId -> s (SENT_, mId)
+  MERR mId e -> s (MERR_, mId, e)
+  MERRS mIds e -> s (MERRS_, mIds, e)
  MSG msgMeta msgFlags msgBody -> B.unwords [s MSG_, s msgMeta, smpEncode msgFlags, serializeBinary msgBody]
  MSGNTF smpMsgMeta -> s (MSGNTF_, smpMsgMeta)
-  ACK mId rcptInfo_ -> s (ACK_, Str $ bshow mId) <> maybe "" (B.cons ' ' . serializeBinary) rcptInfo_
+  ACK mId rcptInfo_ -> s (ACK_, mId) <> maybe "" (B.cons ' ' . serializeBinary) rcptInfo_
  RCVD msgMeta rcpts -> s (RCVD_, msgMeta, rcpts)
  SWCH -> s SWCH_
  OFF -> s OFF_
@@ -34,23 +34,25 @@ import UnliftIO.STM
 -- See a full agent executable here: https://github.com/simplex-chat/simplexmq/blob/master/apps/smp-agent/Main.hs
 runSMPAgent :: (MonadRandom m, MonadUnliftIO m) => ATransport -> AgentConfig -> InitialAgentServers -> SQLiteStore -> m ()
 runSMPAgent t cfg initServers store =
-  runSMPAgentBlocking t cfg initServers store =<< newEmptyTMVarIO
+  runSMPAgentBlocking t cfg initServers store 0 =<< newEmptyTMVarIO

 -- | Runs an SMP agent as a TCP service using passed configuration with signalling.
 --
 -- This function uses passed TMVar to signal when the server is ready to accept TCP requests (True)
 -- and when it is disconnected from the TCP socket once the server thread is killed (False).
-runSMPAgentBlocking :: (MonadRandom m, MonadUnliftIO m) => ATransport -> AgentConfig -> InitialAgentServers -> SQLiteStore -> TMVar Bool -> m ()
-runSMPAgentBlocking (ATransport t) cfg@AgentConfig {tcpPort, caCertificateFile, certificateFile, privateKeyFile} initServers store started = do
+runSMPAgentBlocking :: (MonadRandom m, MonadUnliftIO m) => ATransport -> AgentConfig -> InitialAgentServers -> SQLiteStore -> Int -> TMVar Bool -> m ()
+runSMPAgentBlocking (ATransport t) cfg@AgentConfig {tcpPort, caCertificateFile, certificateFile, privateKeyFile} initServers store initClientId started = do
  liftIO (newSMPAgentEnv cfg store) >>= runReaderT (smpAgent t)
  where
    smpAgent :: forall c m'. (Transport c, MonadUnliftIO m', MonadReader Env m') => TProxy c -> m' ()
    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 initClientId
      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
@@ -110,6 +110,7 @@ module Simplex.Messaging.Agent.Store.SQLite
    getPendingQueueMsg,
    updatePendingMsgRIState,
    deletePendingMsgs,
+    getExpiredSndMessages,
    setMsgUserAck,
    getRcvMsg,
    getLastMsg,
@@ -1041,6 +1042,33 @@ deletePendingMsgs :: DB.Connection -> ConnId -> SndQueue -> IO ()
 deletePendingMsgs db connId SndQueue {dbQueueId} =
  DB.execute db "DELETE FROM snd_message_deliveries WHERE conn_id = ? AND snd_queue_id = ?" (connId, dbQueueId)

+getExpiredSndMessages :: DB.Connection -> ConnId -> SndQueue -> UTCTime -> IO [InternalId]
+getExpiredSndMessages db connId SndQueue {dbQueueId} expireTs = do
+  -- type is Maybe InternalId because MAX always returns one row, possibly with NULL value
+  maxId :: [Maybe InternalId] <-
+    map fromOnly
+      <$> DB.query
+        db
+        [sql|
+          SELECT MAX(internal_id)
+          FROM messages
+          WHERE conn_id = ? AND internal_snd_id IS NOT NULL AND internal_ts < ?
+        |]
+        (connId, expireTs)
+  case maxId of
+    Just msgId : _ ->
+      map fromOnly
+        <$> DB.query
+          db
+          [sql|
+            SELECT internal_id
+            FROM snd_message_deliveries
+            WHERE conn_id = ? AND snd_queue_id = ? AND failed = 0 AND internal_id <= ?
+            ORDER BY internal_id ASC
+          |]
+          (connId, dbQueueId, msgId)
+    _ -> pure []
+
 setMsgUserAck :: DB.Connection -> ConnId -> InternalId -> IO (Either StoreError (RcvQueue, SMP.MsgId))
 setMsgUserAck db connId agentMsgId = runExceptT $ do
  (dbRcvId, srvMsgId) <-
@@ -67,6 +67,7 @@ import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230814_indexes
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230829_crypto_files
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20231222_command_created_at
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20231225_failed_work_items
+import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240121_message_delivery_indexes
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (dropPrefix, sumTypeJSON)
 import Simplex.Messaging.Transport.Client (TransportHost)
@@ -102,7 +103,8 @@ schemaMigrations =
    ("m20230814_indexes", m20230814_indexes, Just down_m20230814_indexes),
    ("m20230829_crypto_files", m20230829_crypto_files, Just down_m20230829_crypto_files),
    ("m20231222_command_created_at", m20231222_command_created_at, Just down_m20231222_command_created_at),
-    ("m20231225_failed_work_items", m20231225_failed_work_items, Just down_m20231225_failed_work_items)
+    ("m20231225_failed_work_items", m20231225_failed_work_items, Just down_m20231225_failed_work_items),
+    ("m20240121_message_delivery_indexes", m20240121_message_delivery_indexes, Just down_m20240121_message_delivery_indexes)
  ]

 -- | The list of migrations in ascending order by date
@@ -128,9 +130,12 @@ run st = \case
  where
    runUp Migration {name, up, down} = withTransaction' st $ \db -> do
      when (name == "m20220811_onion_hosts") $ updateServers db
-      insert db >> execSQL db up
+      insert db >> execSQL db up'
      where
        insert db = DB.execute db "INSERT INTO migrations (name, down, ts) VALUES (?,?,?)" . (name,down,) =<< getCurrentTime
+        up'
+          | dbNew st && name == "m20230110_users" = fromQuery new_m20230110_users
+          | otherwise = up
        updateServers db = forM_ (M.assocs extraSMPServerHosts) $ \(h, h') ->
          let hs = decodeLatin1 . strEncode $ ([h, h'] :: NonEmpty TransportHost)
           in DB.execute db "UPDATE servers SET host = ? WHERE host = ?" (hs, decodeLatin1 $ strEncode h)
@@ -27,3 +27,24 @@ UPDATE connections SET user_id = 1;

 PRAGMA ignore_check_constraints=OFF;
 |]
+
+-- This is executed in the new database
+-- It does not create new user record
+new_m20230110_users :: Query
+new_m20230110_users =
+  [sql|
+PRAGMA ignore_check_constraints=ON;
+
+CREATE TABLE users (
+  user_id INTEGER PRIMARY KEY AUTOINCREMENT
+);
+
+ALTER TABLE connections ADD COLUMN user_id INTEGER CHECK (user_id NOT NULL)
+REFERENCES users ON DELETE CASCADE;
+
+CREATE INDEX idx_connections_user ON connections(user_id);
+
+CREATE INDEX idx_commands_conn_id ON commands(conn_id);
+
+PRAGMA ignore_check_constraints=OFF;
+|]
@@ -0,0 +1,20 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240121_message_delivery_indexes where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20240121_message_delivery_indexes :: Query
+m20240121_message_delivery_indexes =
+    [sql|
+CREATE INDEX idx_messages_snd_expired ON messages(conn_id, internal_snd_id, internal_ts);
+CREATE INDEX idx_snd_message_deliveries_expired ON snd_message_deliveries(conn_id, snd_queue_id, failed, internal_id);
+|]
+
+down_m20240121_message_delivery_indexes :: Query
+down_m20240121_message_delivery_indexes =
+    [sql|
+DROP INDEX idx_messages_snd_expired;
+DROP INDEX idx_snd_message_deliveries_expired;
+|]
@@ -497,3 +497,14 @@ CREATE INDEX idx_commands_server_commands ON commands(
 CREATE INDEX idx_rcv_files_status_created_at ON rcv_files(status, created_at);
 CREATE INDEX idx_snd_files_status_created_at ON snd_files(status, created_at);
 CREATE INDEX idx_snd_files_snd_file_entity_id ON snd_files(snd_file_entity_id);
+CREATE INDEX idx_messages_snd_expired ON messages(
+  conn_id,
+  internal_snd_id,
+  internal_ts
+);
+CREATE INDEX idx_snd_message_deliveries_expired ON snd_message_deliveries(
+  conn_id,
+  snd_queue_id,
+  failed,
+  internal_id
+);
@@ -1,24 +0,0 @@
-module Simplex.Messaging.Agent.TAsyncs where
-
-import Control.Monad.IO.Unlift (MonadUnliftIO)
-import Simplex.Messaging.TMap (TMap)
-import qualified Simplex.Messaging.TMap as TM
-import UnliftIO.Async (Async, async)
-import UnliftIO.STM
-
-data TAsyncs = TAsyncs
-  { actionId :: TVar Int,
-    actions :: TMap Int (Async ())
-  }
-
-newTAsyncs :: STM TAsyncs
-newTAsyncs = TAsyncs <$> newTVar 0 <*> TM.empty
-
-newAsyncAction :: MonadUnliftIO m => (Int -> m ()) -> TAsyncs -> m ()
-newAsyncAction action as = do
-  aId <- atomically $ stateTVar (actionId as) $ \i -> (i + 1, i + 1)
-  a <- async $ action aId
-  atomically $ TM.insert aId a $ actions as
-
-removeAsyncAction :: Int -> TAsyncs -> STM ()
-removeAsyncAction aId = TM.delete aId . actions
@@ -72,9 +72,7 @@ module Simplex.Messaging.Client
    ClientCommand,

    -- * For testing
-    ClientBatch (..),
    PCTransmission,
-    batchClientTransmissions,
    mkTransmission,
    clientStub,
  )
@@ -99,10 +97,9 @@ import Data.Time.Clock (UTCTime, getCurrentTime)
 import Network.Socket (ServiceName)
 import Numeric.Natural
 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 as SMP
+import Simplex.Messaging.Protocol
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport
@@ -173,7 +170,7 @@ clientStub sessionId = do
            }
      }

-type SMPClient = ProtocolClient ErrorType SMP.BrokerMsg
+type SMPClient = ProtocolClient ErrorType BrokerMsg

 -- | Type for client command data
 type ClientCommand msg = (Maybe C.APrivateSignKey, EntityId, ProtoCommand msg)
@@ -349,12 +346,12 @@ getProtocolClient transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize,
      action <-
        async $
          runTransportClient tcConfig (Just username) useHost port' (Just $ keyHash srv) (client t c cVar)
-            `finally` atomically (putTMVar cVar $ Left PCENetworkError)
+            `finally` atomically (tryPutTMVar cVar $ Left PCENetworkError)
      c_ <- tcpConnectTimeout `timeout` atomically (takeTMVar cVar)
-      pure $ case c_ of
-        Just (Right c') -> Right c' {action = Just action}
-        Just (Left e) -> Left e
-        Nothing -> Left PCENetworkError
+      case c_ of
+        Just (Right c') -> pure $ Right c' {action = Just action}
+        Just (Left e) -> pure $ Left e
+        Nothing -> cancel action $> Left PCENetworkError

    useTransport :: (ServiceName, ATransport)
    useTransport = case port srv of
@@ -634,7 +631,7 @@ type PCTransmission err msg = (SentRawTransmission, Request err msg)
 -- | Send multiple commands with batching and collect responses
 sendProtocolCommands :: forall err msg. ProtocolEncoding err (ProtoCommand msg) => ProtocolClient err msg -> NonEmpty (ClientCommand msg) -> IO (NonEmpty (Response err msg))
 sendProtocolCommands c@ProtocolClient {batch, blockSize} cs = do
-  bs <- batchClientTransmissions batch blockSize <$> mapM (mkTransmission c) cs
+  bs <- batchTransmissions' batch blockSize <$> mapM (mkTransmission c) cs
  validate . concat =<< mapM (sendBatch c) bs
  where
    validate :: [Response err msg] -> IO (NonEmpty (Response err msg))
@@ -651,58 +648,24 @@ sendProtocolCommands c@ProtocolClient {batch, blockSize} cs = do

 streamProtocolCommands :: forall err msg. ProtocolEncoding err (ProtoCommand msg) => ProtocolClient err msg -> NonEmpty (ClientCommand msg) -> ([Response err msg] -> IO ()) -> IO ()
 streamProtocolCommands c@ProtocolClient {batch, blockSize} cs cb = do
-  bs <- batchClientTransmissions batch blockSize <$> mapM (mkTransmission c) cs
+  bs <- batchTransmissions' batch blockSize <$> mapM (mkTransmission c) cs
  mapM_ (cb <=< sendBatch c) bs

-sendBatch :: ProtocolClient err msg -> ClientBatch err msg -> IO [Response err msg]
+sendBatch :: ProtocolClient err msg -> TransportBatch (Request err msg) -> IO [Response err msg]
 sendBatch c@ProtocolClient {client_ = PClient {sndQ}} b = do
  case b of
-    CBLargeTransmission Request {entityId} -> do
+    TBLargeTransmission Request {entityId} -> do
      putStrLn "send error: large message"
      pure [Response entityId $ Left $ PCETransportError TELargeMsg]
-    CBTransmissions s n rs -> do
-      when (n > 0) $ atomically $ writeTBQueue sndQ $ tEncodeBatch n s
-      mapConcurrently (getResponse c) rs
-    CBTransmission s r -> do
+    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

-data ClientBatch err msg
-  = -- ByteString in CBTransmissions does not include count byte, it is added by tEncodeBatch
-    CBTransmissions ByteString Int [Request err msg]
-  | CBTransmission ByteString (Request err msg)
-  | CBLargeTransmission (Request err msg)
-
-- | encodes and batches transmissions into blocks
-batchClientTransmissions :: forall err msg. Bool -> Int -> NonEmpty (PCTransmission err msg) -> [ClientBatch err msg]
-batchClientTransmissions batch blkSize
-  | batch = reverse . mkBatch []
-  | otherwise = map mkBatch1 . L.toList
-  where
-    mkBatch :: [ClientBatch err msg] -> NonEmpty (PCTransmission err msg) -> [ClientBatch err msg]
-    mkBatch bs ts =
-      let (b, ts_) = encodeBatch "" 0 [] ts
-          bs' = b : bs
-       in maybe bs' (mkBatch bs') ts_
-    mkBatch1 :: PCTransmission err msg -> ClientBatch err msg
-    mkBatch1 (t, r)
-      | B.length s <= blkSize - 2 = CBTransmission s r
-      | otherwise = CBLargeTransmission r
-      where
-        s = tEncode t
-    encodeBatch :: ByteString -> Int -> [Request err msg] -> NonEmpty (PCTransmission err msg) -> (ClientBatch err msg, Maybe (NonEmpty (PCTransmission err msg)))
-    encodeBatch s n rs ts@((t, r) :| ts_)
-      | B.length s' <= blkSize - 3 && n < 255 =
-          case L.nonEmpty ts_ of
-            Just ts' -> encodeBatch s' n' rs' ts'
-            Nothing -> (CBTransmissions s' n' (reverse rs'), Nothing)
-      | n == 0 = (CBLargeTransmission r, L.nonEmpty ts_)
-      | otherwise = (CBTransmissions s n (reverse rs), Just ts)
-      where
-        s' = s <> smpEncode (Large $ tEncode t)
-        n' = n + 1
-        rs' = r : rs
-
 -- | Send Protocol command
 sendProtocolCommand :: forall err msg. ProtocolEncoding err (ProtoCommand msg) => ProtocolClient err msg -> Maybe C.APrivateSignKey -> EntityId -> ProtoCommand msg -> ExceptT (ProtocolClientError err) IO msg
 sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ}, batch, blockSize} pKey entId cmd =
@@ -715,7 +678,7 @@ sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ}, batch, blockSize
      | otherwise = atomically (writeTBQueue sndQ s) >> response <$> getResponse c r
      where
        s
-          | batch = tEncodeBatch 1 . smpEncode . Large $ tEncode t
+          | batch = tEncodeBatch1 t
          | otherwise = tEncode t

 -- TODO switch to timeout or TimeManager that supports Int64
@@ -41,7 +41,6 @@ import qualified Crypto.MAC.Poly1305 as Poly1305
 import Data.Bifunctor (first)
 import Data.ByteArray (ByteArrayAccess)
 import qualified Data.ByteArray as BA
-import qualified Data.ByteString as S
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy.Char8 as LB
@@ -175,7 +174,7 @@ secretBoxTailTag sbProcess secret nonce msg = run <$> sbInit_ secret nonce

 -- passes lazy bytestring via initialized secret box returning the reversed list of chunks
 secretBoxLazy_ :: (SbState -> ByteString -> (ByteString, SbState)) -> SbState -> LazyByteString -> ([ByteString], SbState)
-secretBoxLazy_ sbProcess state = foldlChunks update ([], state)
+secretBoxLazy_ sbProcess state = LB.foldlChunks update ([], state)
  where
    update (cs, st) chunk = let (!c, !st') = sbProcess st chunk in (c : cs, st')

@@ -231,10 +230,3 @@ cryptoPassed :: CE.CryptoFailable b -> Either CryptoError b
 cryptoPassed = \case
  CE.CryptoPassed a -> Right a
  CE.CryptoFailed e -> Left $ CryptoPoly1305Error e
-
-foldlChunks :: (a -> S.ByteString -> a) -> a -> LazyByteString -> a
-foldlChunks f = go
-  where
-    go !a LB.Empty = a
-    go !a (LB.Chunk c cs) = go (f a c) cs
-{-# INLINE foldlChunks #-}
@@ -174,37 +174,37 @@ instance (Encoding a, Encoding b) => Encoding (a, b) where
  {-# INLINE smpP #-}

 instance (Encoding a, Encoding b, Encoding c) => Encoding (a, b, c) where
-  smpEncode (a, b, c) = smpEncode a <> smpEncode b <> smpEncode c
+  smpEncode (a, b, c) = B.concat [smpEncode a, smpEncode b, smpEncode c]
  {-# INLINE smpEncode #-}
  smpP = (,,) <$> smpP <*> smpP <*> smpP
  {-# INLINE smpP #-}

 instance (Encoding a, Encoding b, Encoding c, Encoding d) => Encoding (a, b, c, d) where
-  smpEncode (a, b, c, d) = smpEncode a <> smpEncode b <> smpEncode c <> smpEncode d
+  smpEncode (a, b, c, d) = B.concat [smpEncode a, smpEncode b, smpEncode c, smpEncode d]
  {-# INLINE smpEncode #-}
  smpP = (,,,) <$> smpP <*> smpP <*> smpP <*> smpP
  {-# INLINE smpP #-}

 instance (Encoding a, Encoding b, Encoding c, Encoding d, Encoding e) => Encoding (a, b, c, d, e) where
-  smpEncode (a, b, c, d, e) = smpEncode a <> smpEncode b <> smpEncode c <> smpEncode d <> smpEncode e
+  smpEncode (a, b, c, d, e) = B.concat [smpEncode a, smpEncode b, smpEncode c, smpEncode d, smpEncode e]
  {-# INLINE smpEncode #-}
  smpP = (,,,,) <$> smpP <*> smpP <*> smpP <*> smpP <*> smpP
  {-# INLINE smpP #-}

 instance (Encoding a, Encoding b, Encoding c, Encoding d, Encoding e, Encoding f) => Encoding (a, b, c, d, e, f) where
-  smpEncode (a, b, c, d, e, f) = smpEncode a <> smpEncode b <> smpEncode c <> smpEncode d <> smpEncode e <> smpEncode f
+  smpEncode (a, b, c, d, e, f) = B.concat [smpEncode a, smpEncode b, smpEncode c, smpEncode d, smpEncode e, smpEncode f]
  {-# INLINE smpEncode #-}
  smpP = (,,,,,) <$> smpP <*> smpP <*> smpP <*> smpP <*> smpP <*> smpP
  {-# INLINE smpP #-}

 instance (Encoding a, Encoding b, Encoding c, Encoding d, Encoding e, Encoding f, Encoding g) => Encoding (a, b, c, d, e, f, g) where
-  smpEncode (a, b, c, d, e, f, g) = smpEncode a <> smpEncode b <> smpEncode c <> smpEncode d <> smpEncode e <> smpEncode f <> smpEncode g
+  smpEncode (a, b, c, d, e, f, g) = B.concat [smpEncode a, smpEncode b, smpEncode c, smpEncode d, smpEncode e, smpEncode f, smpEncode g]
  {-# INLINE smpEncode #-}
  smpP = (,,,,,,) <$> smpP <*> smpP <*> smpP <*> smpP <*> smpP <*> smpP <*> smpP
  {-# INLINE smpP #-}

 instance (Encoding a, Encoding b, Encoding c, Encoding d, Encoding e, Encoding f, Encoding g, Encoding h) => Encoding (a, b, c, d, e, f, g, h) where
-  smpEncode (a, b, c, d, e, f, g, h) = smpEncode a <> smpEncode b <> smpEncode c <> smpEncode d <> smpEncode e <> smpEncode f <> smpEncode g <> smpEncode h
+  smpEncode (a, b, c, d, e, f, g, h) = B.concat [smpEncode a, smpEncode b, smpEncode c, smpEncode d, smpEncode e, smpEncode f, smpEncode g, smpEncode h]
  {-# INLINE smpEncode #-}
  smpP = (,,,,,,,) <$> smpP <*> smpP <*> smpP <*> smpP <*> smpP <*> smpP <*> smpP <*> smpP
  {-# INLINE smpP #-}
@@ -109,9 +109,9 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg} started = do
      liftIO $ threadDelay' $ 1000000 * (initialDelay + if initialDelay < 0 then 86400 else 0)
      NtfServerStats {fromTime, tknCreated, tknVerified, tknDeleted, subCreated, subDeleted, ntfReceived, ntfDelivered, activeTokens, activeSubs} <- asks serverStats
      let interval = 1000000 * logInterval
-      withFile statsFilePath AppendMode $ \h -> liftIO $ do
-        hSetBuffering h LineBuffering
-        forever $ do
+      forever $ do
+        withFile statsFilePath AppendMode $ \h -> liftIO $ do
+          hSetBuffering h LineBuffering
          ts <- getCurrentTime
          fromTime' <- atomically $ swapTVar fromTime ts
          tknCreated' <- atomically $ swapTVar tknCreated 0
@@ -141,7 +141,7 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg} started = do
                weekCount sub,
                monthCount sub
              ]
-          threadDelay' interval
+        liftIO $ threadDelay' interval

 resubscribe :: NtfSubscriber -> Map NtfSubscriptionId NtfSubData -> M ()
 resubscribe NtfSubscriber {newSubQ} subs = do
@@ -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
@@ -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.1.0"

 defaultSMPBatchDelay :: Int
 defaultSMPBatchDelay = 10000
@@ -144,8 +144,9 @@ module Simplex.Messaging.Protocol
    tParse,
    tDecodeParseValidate,
    tEncode,
-    tEncodeBatch,
+    tEncodeBatch1,
    batchTransmissions,
+    batchTransmissions',

    -- * exports for tests
    CommandTag (..),
@@ -154,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
@@ -1283,14 +1283,14 @@ 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 :: TransportBatch () -> IO [Either TransportError ()]
    tPutBatch = \case
-      TBLargeTransmission -> [Left TELargeMsg] <$ putStrLn "tPut error: large message"
-      TBTransmissions n s -> replicate n <$> (tPutLog th (tEncodeBatch n s) <* mapM_ threadDelay delay_)
-      TBTransmission s -> (: []) <$> tPutLog th s
+      TBLargeTransmission _ -> [Left TELargeMsg] <$ putStrLn "tPut error: large message"
+      TBTransmissions s n _ -> replicate n <$> (tPutLog th s)
+      TBTransmission s _ -> (: []) <$> tPutLog th s

 tPutLog :: Transport c => THandle c -> ByteString -> IO (Either TransportError ())
 tPutLog th s = do
@@ -1300,44 +1300,52 @@ tPutLog th s = do
    _ -> pure ()
  pure r

-- ByteString does not include length byte, it is added by tEncodeBatch
-data TransportBatch = TBTransmissions Int ByteString | TBTransmission ByteString | TBLargeTransmission
+-- 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 (,())

 -- | encodes and batches transmissions into blocks,
-batchTransmissions :: Bool -> Int -> NonEmpty SentRawTransmission -> [TransportBatch]
-batchTransmissions batch bSize
-  | batch = reverse . mkBatch [] . L.map tEncode
-  | otherwise = map (mkBatch1 . tEncode) . L.toList
+batchTransmissions' :: forall r. Bool -> Int -> NonEmpty (SentRawTransmission, r) -> [TransportBatch r]
+batchTransmissions' batch bSize
+  | batch = addBatch . foldr addTransmission ([], 0, 0, [], [])
+  | otherwise = map mkBatch1 . L.toList
  where
-    mkBatch :: [TransportBatch] -> NonEmpty ByteString -> [TransportBatch]
-    mkBatch rs ts =
-      let (n, s, ts_) = encodeBatch 0 "" ts
-          r = if n == 0 then TBLargeTransmission else TBTransmissions n s
-          rs' = r : rs
-       in case ts_ of
-            Just ts' -> mkBatch rs' ts'
-            _ -> rs'
-    mkBatch1 :: ByteString -> TransportBatch
-    mkBatch1 s = if B.length s > bSize - 2 then TBLargeTransmission else TBTransmission s
-    encodeBatch :: Int -> ByteString -> NonEmpty ByteString -> (Int, ByteString, Maybe (NonEmpty ByteString))
-    encodeBatch n s ts@(t :| ts_)
-      | n == 255 = (n, s, Just ts)
-      | otherwise =
-          let s' = s <> smpEncode (Large t)
-              n' = n + 1
-           in if B.length s' > bSize - 3 -- one byte is reserved for the number of messages in the batch
-                then (n,s,) $ if n == 0 then L.nonEmpty ts_ else Just ts
-                else case L.nonEmpty ts_ of
-                  Just ts' -> encodeBatch n' s' ts'
-                  _ -> (n', s', Nothing)
+    mkBatch1 :: (SentRawTransmission, r) -> TransportBatch r
+    mkBatch1 (t, r)
+      -- 2 bytes are reserved for pad size
+      | B.length s <= bSize - 2 = TBTransmission s r
+      | otherwise = TBLargeTransmission r
+      where
+        s = tEncode t
+    -- 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 = 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 -> ByteString
 tEncode (sig, t) = smpEncode (C.signatureBytes sig) <> t
 {-# INLINE tEncode #-}

-tEncodeBatch :: Int -> ByteString -> ByteString
-tEncodeBatch n s = lenEncode n `B.cons` 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) =
@@ -203,9 +203,9 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
      liftIO $ threadDelay' $ 1000000 * (initialDelay + if initialDelay < 0 then 86400 else 0)
      ServerStats {fromTime, qCreated, qSecured, qDeleted, msgSent, msgRecv, msgExpired, activeQueues, msgSentNtf, msgRecvNtf, activeQueuesNtf, qCount, msgCount} <- asks serverStats
      let interval = 1000000 * logInterval
-      withFile statsFilePath AppendMode $ \h -> liftIO $ do
-        hSetBuffering h LineBuffering
-        forever $ do
+      forever $ do
+        withFile statsFilePath AppendMode $ \h -> liftIO $ do
+          hSetBuffering h LineBuffering
          ts <- getCurrentTime
          fromTime' <- atomically $ swapTVar fromTime ts
          qCreated' <- atomically $ swapTVar qCreated 0
@@ -241,7 +241,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
                show msgCount',
                show msgExpired'
              ]
-          threadDelay' interval
+        liftIO $ threadDelay' interval

    runClient :: Transport c => TProxy c -> c -> M ()
    runClient tp h = do
@@ -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
@@ -69,7 +69,7 @@ import Data.Bifunctor (first)
 import Data.Bitraversable (bimapM)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
-import qualified Data.ByteString.Lazy as BL
+import qualified Data.ByteString.Lazy.Char8 as LB
 import Data.Default (def)
 import Data.Functor (($>))
 import Data.Version (showVersion)
@@ -217,8 +217,8 @@ instance Transport TLS where
    getBuffered tlsBuffer n t_ (T.recvData tlsContext)

  cPut :: TLS -> ByteString -> IO ()
-  cPut TLS {tlsContext, tlsTransportConfig = TransportConfig {transportTimeout = t_}} s =
-    withTimedErr t_ . T.sendData tlsContext $ BL.fromStrict s
+  cPut TLS {tlsContext, tlsTransportConfig = TransportConfig {transportTimeout = t_}} =
+    withTimedErr t_ . T.sendData tlsContext . LB.fromStrict

  getLn :: TLS -> IO ByteString
  getLn TLS {tlsContext, tlsBuffer} = do
@@ -21,6 +21,7 @@ module Simplex.Messaging.Transport.Client
 where

 import Control.Applicative (optional)
+import Control.Logger.Simple (logError)
 import Control.Monad.IO.Unlift
 import Data.Aeson (FromJSON (..), ToJSON (..))
 import qualified Data.Attoparsec.ByteString.Char8 as A
@@ -48,7 +49,7 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (parseAll, parseString)
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.KeepAlive
-import Simplex.Messaging.Util (bshow, (<$?>))
+import Simplex.Messaging.Util (bshow, (<$?>), catchAll, tshow)
 import System.IO.Error
 import Text.Read (readMaybe)
 import UnliftIO.Exception (IOException)
@@ -135,7 +136,7 @@ runTLSTransportClient tlsParams caStore_ cfg@TransportClientConfig {socksProxy,
        _ -> connectTCPClient hostName
  c <- liftIO $ do
    sock <- connectTCP port
-    mapM_ (setSocketKeepAlive sock) tcpKeepAlive
+    mapM_ (setSocketKeepAlive sock) tcpKeepAlive `catchAll` \e -> logError ("Error setting TCP keep-alive" <> tshow e)
    let tCfg = clientTransportConfig cfg
    connectTLS (Just hostName) tCfg clientParams sock >>= getClientConnection tCfg
  client c `E.finally` liftIO (closeConnection c)
@@ -7,7 +7,7 @@ module Simplex.Messaging.Transport.WebSockets (WS (..)) where
 import qualified Control.Exception as E
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
-import qualified Data.ByteString.Lazy as BL
+import qualified Data.ByteString.Lazy as LB
 import qualified Network.TLS as T
 import Network.WebSockets
 import Network.WebSockets.Stream (Stream)
@@ -101,5 +101,5 @@ makeTLSContextStream cxt =
      (Just <$> T.recvData cxt) `E.catch` \case
        T.Error_EOF -> pure Nothing
        e -> E.throwIO e
-    writeStream :: Maybe BL.ByteString -> IO ()
+    writeStream :: Maybe LB.ByteString -> IO ()
    writeStream = maybe (closeTLS cxt) (T.sendData cxt)
@@ -359,7 +359,6 @@ testServerConnectionAfterError t _ = do
    withAgent2 $ \alice -> do
      withServer $ do
        connect (bob, "bob") (alice, "alice")
-
      bob <#. ("", "", DOWN server ["alice"])
      alice <#. ("", "", DOWN server ["bob"])
      alice #: ("1", "bob", "SEND F 5\nhello") #> ("1", "bob", MID 4)
@@ -386,10 +385,10 @@ testServerConnectionAfterError t _ = do
  where
    server = SMPServer "localhost" testPort2 testKeyHash
    withServer test' = withSmpServerStoreLogOn (ATransport t) testPort2 (const test') `shouldReturn` ()
-    withAgent1 = withAgent agentTestPort testDB
-    withAgent2 = withAgent agentTestPort2 testDB2
-    withAgent :: String -> FilePath -> (c -> IO a) -> IO a
-    withAgent agentPort agentDB = withSmpAgentThreadOn_ (ATransport t) (agentPort, testPort2, agentDB) (pure ()) . const . testSMPAgentClientOn agentPort
+    withAgent1 = withAgent agentTestPort testDB 0 
+    withAgent2 = withAgent agentTestPort2 testDB2 10
+    withAgent :: String -> FilePath -> Int -> (c -> IO a) -> IO a
+    withAgent agentPort agentDB initClientId = withSmpAgentThreadOn_ (ATransport t) (agentPort, testPort2, agentDB) initClientId (pure ()) . const . testSMPAgentClientOn agentPort

 testMsgDeliveryAgentRestart :: Transport c => TProxy c -> c -> IO ()
 testMsgDeliveryAgentRestart t bob = do
@@ -424,7 +423,7 @@ testMsgDeliveryAgentRestart t bob = do
  removeFile testDB
  where
    withServer test' = withSmpServerStoreLogOn (ATransport t) testPort2 (const test') `shouldReturn` ()
-    withAgent = withSmpAgentThreadOn_ (ATransport t) (agentTestPort, testPort, testDB) (pure ()) . const . testSMPAgentClientOn agentTestPort
+    withAgent = withSmpAgentThreadOn_ (ATransport t) (agentTestPort, testPort, testDB) 0 (pure ()) . const . testSMPAgentClientOn agentTestPort

 testConcurrentMsgDelivery :: Transport c => TProxy c -> c -> c -> IO ()
 testConcurrentMsgDelivery _ alice bob = do
@@ -36,6 +36,7 @@ import AgentTests.ConnectionRequestTests (connReqData, queueAddr, testE2ERatchet
 import Control.Concurrent (killThread, threadDelay)
 import Control.Monad
 import Control.Monad.Except
+import Control.Monad.Reader
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Either (isRight)
@@ -46,13 +47,15 @@ import qualified Data.Set as S
 import Data.Time.Clock (diffUTCTime, getCurrentTime)
 import Data.Time.Clock.System (SystemTime (..), getSystemTime)
 import Data.Type.Equality
+import qualified Database.SQLite.Simple as SQL
 import SMPAgentClient
 import SMPClient (cfg, testPort, testPort2, testStoreLogFile2, withSmpServer, withSmpServerConfigOn, withSmpServerOn, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn)
 import Simplex.Messaging.Agent
 import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestStep (..))
 import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..), InitialAgentServers (..), createAgentStore)
 import Simplex.Messaging.Agent.Protocol as Agent
-import Simplex.Messaging.Agent.Store.SQLite (MigrationConfirmation (..))
+import Simplex.Messaging.Agent.Store.SQLite (MigrationConfirmation (..), SQLiteStore (dbNew))
+import Simplex.Messaging.Agent.Store.SQLite.Common (withTransaction')
 import Simplex.Messaging.Client (NetworkConfig (..), ProtocolClientConfig (..), TransportSessionMode (TSMEntity, TSMUser), defaultClientConfig)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
@@ -73,9 +76,12 @@ type AEntityTransmission e = (ACorrId, ConnId, ACommand 'Agent e)
 a ##> t = withTimeout a (`shouldBe` t)

 (=##>) :: (Show a, HasCallStack, MonadUnliftIO m) => m a -> (a -> Bool) -> m ()
-a =##> p = withTimeout a (`shouldSatisfy` p)
+a =##> p =
+  withTimeout a $ \r -> do
+    unless (p r) $ liftIO $ putStrLn $ "value failed predicate: " <> show r
+    r `shouldSatisfy` p

-withTimeout :: MonadUnliftIO m => m a -> (a -> Expectation) -> m ()
+withTimeout :: (HasCallStack, MonadUnliftIO m) => m a -> (a -> Expectation) -> m ()
 withTimeout a test =
  timeout 10_000000 a >>= \case
    Nothing -> error "operation timed out"
@@ -111,6 +117,9 @@ pGet c = do
 pattern Msg :: MsgBody -> ACommand 'Agent e
 pattern Msg msgBody <- MSG MsgMeta {integrity = MsgOk} _ msgBody

+pattern MsgErr :: AgentMsgId -> MsgErrorType -> MsgBody -> ACommand 'Agent e
+pattern MsgErr msgId err msgBody <- MSG MsgMeta {recipient = (msgId, _), integrity = MsgError err} _ msgBody
+
 pattern Rcvd :: AgentMsgId -> ACommand 'Agent e
 pattern Rcvd agentMsgId <- RCVD MsgMeta {integrity = MsgOk} [MsgReceipt {agentMsgId, msgRcptStatus = MROk}]

@@ -160,15 +169,18 @@ runRight action =
    Left e -> error $ "Unexpected error: " <> show e

 getInAnyOrder :: HasCallStack => AgentClient -> [ATransmission 'Agent -> Bool] -> Expectation
-getInAnyOrder _ [] = pure ()
-getInAnyOrder c rs = do
-  r <- pGet c
+getInAnyOrder c = inAnyOrder (pGet c)
+
+inAnyOrder :: (Show a, MonadIO m, HasCallStack) => m a -> [a -> Bool] -> m ()
+inAnyOrder _ [] = pure ()
+inAnyOrder g rs = do
+  r <- g
  let rest = filter (not . expected r) rs
  if length rest < length rs
-    then getInAnyOrder c rest
+    then inAnyOrder g rest
    else error $ "unexpected event: " <> show r
  where
-    expected :: ATransmission 'Agent -> (ATransmission 'Agent -> Bool) -> Bool
+    expected :: a -> (a -> Bool) -> Bool
    expected r rp = rp r

 functionalAPITests :: ATransport -> Spec
@@ -212,6 +224,11 @@ functionalAPITests t = do
      testDuplicateMessage t
    it "should report error via msg integrity on skipped messages" $
      testSkippedMessages t
+    describe "message expiration" $ do
+      it "should expire one message" $ testExpireMessage t
+      it "should expire multiple messages" $ testExpireManyMessages t
+      it "should expire one message if quota is exceeded" $ testExpireMessageQuota t
+      it "should expire multiple messages if quota is exceeded" $ testExpireManyMessagesQuota t
    describe "Ratchet synchronization" $ do
      it "should report ratchet de-synchronization, synchronize ratchets" $
        testRatchetSync t
@@ -375,8 +392,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
@@ -385,7 +402,7 @@ withAgentClients2 :: (AgentClient -> AgentClient -> IO ()) -> IO ()
 withAgentClients2 = withAgentClientsCfg2 agentCfg agentCfg

 runAgentClientTest :: HasCallStack => AgentClient -> AgentClient -> AgentMsgId -> IO ()
-runAgentClientTest alice bob baseId = do
+runAgentClientTest alice bob baseId =
  runRight_ $ do
    (bobId, qInfo) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
    aliceId <- joinConnection bob 1 True qInfo "bob's connInfo" SMSubscribe
@@ -421,9 +438,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
@@ -446,7 +463,7 @@ testAgentClient3 = do
    ackMessage c aIdForC 5 Nothing

 runAgentClientContactTest :: HasCallStack => AgentClient -> AgentClient -> AgentMsgId -> IO ()
-runAgentClientContactTest alice bob baseId = do
+runAgentClientContactTest alice bob baseId =
  runRight_ $ do
    (_, qInfo) <- createConnection alice 1 True SCMContact Nothing SMSubscribe
    aliceId <- joinConnection bob 1 True qInfo "bob's connInfo" SMSubscribe
@@ -496,7 +513,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 +530,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 +544,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 +591,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
@@ -589,13 +606,13 @@ testAllowConnectionClientRestart t = do

    runRight_ $ do
      allowConnectionAsync alice "1" bobId confId "alice's connInfo"
-      ("1", _, OK) <- get alice
+      get alice =##> \case ("1", _, OK) -> True; _ -> False
      pure ()

    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 +630,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 +643,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 +654,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 +665,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 +674,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 +691,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 +704,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 +719,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 +732,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 +744,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 +760,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 +775,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 +786,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 +802,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 +825,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 +852,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
@@ -855,6 +872,102 @@ testSkippedMessages t = do
  disconnectAgentClient alice2
  disconnectAgentClient bob2

+testExpireMessage :: HasCallStack => ATransport -> IO ()
+testExpireMessage t = do
+  a <- getSMPAgentClient' 1 agentCfg {messageTimeout = 1, messageRetryInterval = fastMessageRetryInterval} initAgentServers testDB
+  b <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
+  (aId, bId) <- withSmpServerStoreLogOn t testPort $ \_ -> runRight $ makeConnection a b
+  nGet a =##> \case ("", "", DOWN _ [c]) -> c == bId; _ -> False
+  nGet b =##> \case ("", "", DOWN _ [c]) -> c == aId; _ -> False
+  4 <- runRight $ sendMessage a bId SMP.noMsgFlags "1"
+  threadDelay 1000000
+  5 <- runRight $ sendMessage a bId SMP.noMsgFlags "2" -- this won't expire
+  get a =##> \case ("", c, MERR 4 (BROKER _ e)) -> bId == c && (e == TIMEOUT || e == NETWORK); _ -> False
+  withSmpServerStoreLogOn t testPort $ \_ -> runRight_ $ do
+    withUP a bId $ \case ("", _, SENT 5) -> True; _ -> False
+    withUP b aId $ \case ("", _, MsgErr 4 (MsgSkipped 3 3) "2") -> True; _ -> False
+    ackMessage b aId 4 Nothing
+
+testExpireManyMessages :: HasCallStack => ATransport -> IO ()
+testExpireManyMessages t = do
+  a <- getSMPAgentClient' 1 agentCfg {messageTimeout = 1, messageRetryInterval = fastMessageRetryInterval} initAgentServers testDB
+  b <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
+  (aId, bId) <- withSmpServerStoreLogOn t testPort $ \_ -> runRight $ makeConnection a b
+  runRight_ $ do
+    nGet a =##> \case ("", "", DOWN _ [c]) -> c == bId; _ -> False
+    nGet b =##> \case ("", "", DOWN _ [c]) -> c == aId; _ -> False
+    4 <- sendMessage a bId SMP.noMsgFlags "1"
+    5 <- sendMessage a bId SMP.noMsgFlags "2"
+    6 <- sendMessage a bId SMP.noMsgFlags "3"
+    liftIO $ threadDelay 1000000
+    7 <- sendMessage a bId SMP.noMsgFlags "4" -- this won't expire
+    get a =##> \case ("", c, MERR 4 (BROKER _ e)) -> bId == c && (e == TIMEOUT || e == NETWORK); _ -> False
+    get a =##> \case ("", c, MERRS [5, 6] (BROKER _ e)) -> bId == c && (e == TIMEOUT || e == NETWORK); _ -> False
+  withSmpServerStoreLogOn t testPort $ \_ -> runRight_ $ do
+    withUP a bId $ \case ("", _, SENT 7) -> True; _ -> False
+    withUP b aId $ \case ("", _, MsgErr 4 (MsgSkipped 3 5) "4") -> True; _ -> False
+    ackMessage b aId 4 Nothing
+
+withUP :: AgentClient -> ConnId -> (AEntityTransmission 'AEConn -> Bool) -> ExceptT AgentErrorType IO ()
+withUP a bId p =
+  liftIO $
+    getInAnyOrder
+      a
+      [ \case ("", "", APC SAENone (UP _ [c])) -> c == bId; _ -> False,
+        \case (corrId, c, APC SAEConn cmd) -> c == bId && p (corrId, c, cmd); _ -> False
+      ]
+
+testExpireMessageQuota :: HasCallStack => ATransport -> IO ()
+testExpireMessageQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1} testPort $ \_ -> do
+  a <- getSMPAgentClient' 1 agentCfg {quotaExceededTimeout = 1, messageRetryInterval = fastMessageRetryInterval} initAgentServers testDB
+  b <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
+  (aId, bId) <- runRight $ do
+    (aId, bId) <- makeConnection a b
+    liftIO $ threadDelay 500000
+    disconnectAgentClient b
+    4 <- sendMessage a bId SMP.noMsgFlags "1"
+    get a ##> ("", bId, SENT 4)
+    5 <- sendMessage a bId SMP.noMsgFlags "2"
+    liftIO $ threadDelay 1000000
+    6 <- sendMessage a bId SMP.noMsgFlags "3" -- this won't expire
+    get a =##> \case ("", c, MERR 5 (SMP QUOTA)) -> bId == c; _ -> False
+    pure (aId, bId)
+  b' <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
+  runRight_ $ do
+    subscribeConnection b' aId
+    get b' =##> \case ("", c, Msg "1") -> c == aId; _ -> False
+    ackMessage b' aId 4 Nothing
+    get a ##> ("", bId, SENT 6)
+    get b' =##> \case ("", c, MsgErr 6 (MsgSkipped 4 4) "3") -> c == aId; _ -> False
+    ackMessage b' aId 6 Nothing
+
+testExpireManyMessagesQuota :: HasCallStack => ATransport -> IO ()
+testExpireManyMessagesQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1} testPort $ \_ -> do
+  a <- getSMPAgentClient' 1 agentCfg {quotaExceededTimeout = 1, messageRetryInterval = fastMessageRetryInterval} initAgentServers testDB
+  b <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
+  (aId, bId) <- runRight $ do
+    (aId, bId) <- makeConnection a b
+    liftIO $ threadDelay 500000
+    disconnectAgentClient b
+    4 <- sendMessage a bId SMP.noMsgFlags "1"
+    get a ##> ("", bId, SENT 4)
+    5 <- sendMessage a bId SMP.noMsgFlags "2"
+    6 <- sendMessage a bId SMP.noMsgFlags "3"
+    7 <- sendMessage a bId SMP.noMsgFlags "4"
+    liftIO $ threadDelay 1000000
+    8 <- sendMessage a bId SMP.noMsgFlags "5" -- this won't expire
+    get a =##> \case ("", c, MERR 5 (SMP QUOTA)) -> bId == c; _ -> False
+    get a =##> \case ("", c, MERRS [6, 7] (SMP QUOTA)) -> bId == c; _ -> False
+    pure (aId, bId)
+  b' <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
+  runRight_ $ do
+    subscribeConnection b' aId
+    get b' =##> \case ("", c, Msg "1") -> c == aId; _ -> False
+    ackMessage b' aId 4 Nothing
+    get a ##> ("", bId, SENT 8)
+    get b' =##> \case ("", c, MsgErr 6 (MsgSkipped 4 6) "5") -> c == aId; _ -> False
+    ackMessage b' aId 6 Nothing
+
 testRatchetSync :: HasCallStack => ATransport -> IO ()
 testRatchetSync t = withAgentClients2 $ \alice bob ->
  withSmpServerStoreMsgLogOn t testPort $ \_ -> do
@@ -899,7 +1012,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 +1072,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 +1081,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 +1097,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 +1131,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 +1214,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 +1224,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 +1236,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 +1327,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
@@ -1289,7 +1402,7 @@ testAsyncCommands =
    liftIO $ aliceId' `shouldBe` aliceId
    ("", _, CONF confId _ "bob's connInfo") <- get alice
    allowConnectionAsync alice "3" bobId confId "alice's connInfo"
-    ("3", _, OK) <- get alice
+    get alice =##> \case ("3", _, OK) -> True; _ -> False
    get alice ##> ("", bobId, CON)
    get bob ##> ("", aliceId, INFO "alice's connInfo")
    get bob ##> ("", aliceId, CON)
@@ -1300,20 +1413,26 @@ testAsyncCommands =
    get alice ##> ("", bobId, SENT $ baseId + 2)
    get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
    ackMessageAsync bob "4" aliceId (baseId + 1) Nothing
-    ("4", _, OK) <- get bob
-    get bob =##> \case ("", c, Msg "how are you?") -> c == aliceId; _ -> False
+    inAnyOrder
+      (get bob)
+      [ \case ("4", _, OK) -> True; _ -> False,
+        \case ("", c, Msg "how are you?") -> c == aliceId; _ -> False
+      ]
    ackMessageAsync bob "5" aliceId (baseId + 2) Nothing
-    ("5", _, OK) <- get bob
+    get bob =##> \case ("5", _, OK) -> True; _ -> False
    3 <- msgId <$> sendMessage bob aliceId SMP.noMsgFlags "hello too"
    get bob ##> ("", aliceId, SENT $ baseId + 3)
    4 <- msgId <$> sendMessage bob aliceId SMP.noMsgFlags "message 1"
    get bob ##> ("", aliceId, SENT $ baseId + 4)
    get alice =##> \case ("", c, Msg "hello too") -> c == bobId; _ -> False
    ackMessageAsync alice "6" bobId (baseId + 3) Nothing
-    ("6", _, OK) <- get alice
-    get alice =##> \case ("", c, Msg "message 1") -> c == bobId; _ -> False
+    inAnyOrder
+      (get alice)
+      [ \case ("6", _, OK) -> True; _ -> False,
+        \case ("", c, Msg "message 1") -> c == bobId; _ -> False
+      ]
    ackMessageAsync alice "7" bobId (baseId + 4) Nothing
-    ("7", _, OK) <- get alice
+    get alice =##> \case ("7", _, OK) -> True; _ -> False
    deleteConnectionAsync alice bobId
    get alice =##> \case ("", c, DEL_RCVQ _ _ Nothing) -> c == bobId; _ -> False
    get alice =##> \case ("", c, DEL_CONN) -> c == bobId; _ -> False
@@ -1324,15 +1443,15 @@ 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
-      ("1", _, INV _) <- get alice'
+      get alice' =##> \case ("1", _, INV _) -> True; _ -> False
      pure ()
  disconnectAgentClient alice'

@@ -1343,8 +1462,7 @@ testAcceptContactAsync =
    aliceId <- joinConnection bob 1 True qInfo "bob's connInfo" SMSubscribe
    ("", _, REQ invId _ "bob's connInfo") <- get alice
    bobId <- acceptContactAsync alice "1" True invId "alice's connInfo" SMSubscribe
-    ("1", bobId', OK) <- get alice
-    liftIO $ bobId' `shouldBe` bobId
+    get alice =##> \case ("1", c, OK) -> c == bobId; _ -> False
    ("", _, CONF confId _ "alice's connInfo") <- get bob
    allowConnection bob aliceId confId "bob's connInfo"
    get alice ##> ("", bobId, INFO "bob's connInfo")
@@ -1378,7 +1496,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 +1516,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
@@ -1410,8 +1528,7 @@ testJoinConnectionAsyncReplyError t = do
    pure (aId, bId)
  nGet a =##> \case ("", "", DOWN _ [c]) -> c == bId; _ -> False
  withSmpServerOn t testPort2 $ do
-    ("2", aId', OK) <- get b
-    liftIO $ aId' `shouldBe` aId
+    get b =##> \case ("2", c, OK) -> c == aId; _ -> False
    confId <- withSmpServerStoreLogOn t testPort $ \_ -> do
      pGet a >>= \case
        ("", "", APC _ (UP _ [_])) -> do
@@ -1491,8 +1608,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 +1692,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 +1710,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 +1773,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 +1824,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 +1878,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 +1936,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 +1989,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 +2017,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 +2058,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 +2079,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 +2220,12 @@ 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
+  c <- getSMPAgentClient_ clientId cfg' initServers st False
+  when (dbNew st) $ withTransaction' st (`SQL.execute_` "INSERT INTO users (user_id) VALUES (1)")
+  pure c

 testServerMultipleIdentities :: HasCallStack => IO ()
 testServerMultipleIdentities =
@@ -2122,7 +2241,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
@@ -1,3 +1,4 @@
+{-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
@@ -10,9 +11,10 @@ module AgentTests.NotificationTests where

 -- import Control.Logger.Simple (LogConfig (..), LogLevel (..), setLogLevel, withGlobalLogging)
 import AgentTests.FunctionalAPITests (exchangeGreetingsMsgId, get, getSMPAgentClient', makeConnection, nGet, runRight, runRight_, switchComplete, testServerMatrix2, (##>), (=##>), pattern Msg)
-import Control.Concurrent (killThread, threadDelay)
+import Control.Concurrent (ThreadId, killThread, threadDelay)
 import Control.Monad
 import Control.Monad.Except
+import Control.Monad.Reader (runReaderT)
 import Control.Monad.Trans.Except
 import qualified Data.Aeson as J
 import qualified Data.Aeson.Types as JT
@@ -21,17 +23,19 @@ import qualified Data.ByteString.Base64.URL as U
 import Data.ByteString.Char8 (ByteString)
 import Data.Text.Encoding (encodeUtf8)
 import NtfClient
-import SMPAgentClient (agentCfg, initAgentServers, initAgentServers2, testDB, testDB2)
+import SMPAgentClient (agentCfg, initAgentServers, initAgentServers2, testDB, testDB2, testDB3, testNtfServer2)
 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.Client (withStore')
+import Simplex.Messaging.Agent.Env.SQLite (InitialAgentServers)
 import Simplex.Messaging.Agent.Protocol
+import Simplex.Messaging.Agent.Store.SQLite (getSavedNtfToken)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Notifications.Protocol
 import Simplex.Messaging.Notifications.Server.Push.APNS
 import Simplex.Messaging.Notifications.Types (NtfToken (..))
-import Simplex.Messaging.Protocol (ErrorType (AUTH), MsgFlags (MsgFlags), SMPMsgMeta (..), SubscriptionMode (..))
+import Simplex.Messaging.Protocol (ErrorType (AUTH), MsgFlags (MsgFlags), ProtocolServer (..), SMPMsgMeta (..), SubscriptionMode (..))
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
 import Simplex.Messaging.Transport (ATransport)
@@ -60,6 +64,12 @@ notificationTests t =
      it "should re-register token when notification server is restarted" $ \_ ->
        withAPNSMockServer $ \apns ->
          testNtfTokenServerRestart t apns
+      it "should work with multiple configured servers" $ \_ ->
+        withAPNSMockServer $ \apns ->
+          testNtfTokenMultipleServers t apns
+      it "should keep working with active token until replaced" $ \_ ->
+        withAPNSMockServer $ \apns ->
+          testNtfTokenChangeServers t apns
    describe "Managing notification subscriptions" $ do
      -- fails on Ubuntu CI?
      xit' "should create notification subscription for existing connection" $ \_ -> do
@@ -95,10 +105,20 @@ notificationTests t =
      testServerMatrix2 t $ \servers ->
        withAPNSMockServer $ \apns ->
          withNtfServer t $ testSwitchNotifications servers apns
+    it "should keep sending notifications for old token" $
+      withSmpServer t $
+        withAPNSMockServer $ \apns ->
+          withNtfServerOn t ntfTestPort $
+            testNotificationsOldToken apns
+    it "should update server from new token" $
+      withSmpServer t $
+        withAPNSMockServer $ \apns ->
+          withNtfServerOn t ntfTestPort2 . withNtfServerThreadOn t ntfTestPort $ \ntf ->
+            testNotificationsNewToken apns ntf

 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 +146,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 +170,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 +205,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 +216,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
@@ -212,10 +232,72 @@ testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
    NTActive <- checkNtfToken a' tkn
    disconnectAgentClient a'

+getTestNtfTokenPort :: (MonadUnliftIO m, MonadError AgentErrorType m) => AgentClient -> m String
+getTestNtfTokenPort a =
+  runReaderT (withStore' a getSavedNtfToken) (agentEnv a) >>= \case
+    Just NtfToken {ntfServer = ProtocolServer {port}} -> pure port
+    Nothing -> error "no active NtfToken"
+
+testNtfTokenMultipleServers :: ATransport -> APNSMockServer -> IO ()
+testNtfTokenMultipleServers t APNSMockServer {apnsQ} = do
+  let tkn = DeviceToken PPApnsTest "abcd"
+  a <- getSMPAgentClient' 1 agentCfg initAgentServers2 testDB
+  withNtfServerThreadOn t ntfTestPort $ \ntf ->
+    withNtfServerThreadOn t ntfTestPort2 $ \ntf2 -> runRight_ $ do
+      -- register a new token, the agent picks a server and stores its choice
+      NTRegistered <- registerNtfToken a tkn NMPeriodic
+      APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}, sendApnsResponse} <-
+        atomically $ readTBQueue apnsQ
+      verification <- ntfData .-> "verification"
+      nonce <- C.cbNonce <$> ntfData .-> "nonce"
+      liftIO $ sendApnsResponse APNSRespOk
+      verifyNtfToken a tkn nonce verification
+      NTActive <- checkNtfToken a tkn
+      -- shut down the "other" server
+      port <- getTestNtfTokenPort a
+      liftIO . killThread $ if port == ntfTestPort then ntf2 else ntf
+      -- still works
+      NTActive <- checkNtfToken a tkn
+      liftIO . killThread $ if port == ntfTestPort then ntf else ntf2
+      -- negative test, the correct server is now gone
+      Left _ <- tryError (checkNtfToken a tkn)
+      pure ()
+
+testNtfTokenChangeServers :: ATransport -> APNSMockServer -> IO ()
+testNtfTokenChangeServers t APNSMockServer {apnsQ} =
+  withNtfServerThreadOn t ntfTestPort $ \ntf -> do
+    tkn1 <- runRight $ do
+      a <- liftIO $ getSMPAgentClient' 1 agentCfg initAgentServers testDB
+      tkn <- registerTestToken a "abcd" NMInstant apnsQ
+      NTActive <- checkNtfToken a tkn
+      setNtfServers a [testNtfServer2]
+      NTActive <- checkNtfToken a tkn -- still works on old server
+      disconnectAgentClient a
+      pure tkn
+
+    threadDelay 1000000
+
+    a <- liftIO $ getSMPAgentClient' 2 agentCfg initAgentServers testDB
+    runRight_ $ do
+      getTestNtfTokenPort a >>= \port -> liftIO $ port `shouldBe` ntfTestPort
+      NTActive <- checkNtfToken a tkn1
+      setNtfServers a [testNtfServer2] -- just change configured server list
+      getTestNtfTokenPort a >>= \port -> liftIO $ port `shouldBe` ntfTestPort -- not yet changed
+      -- trigger token replace
+      tkn2 <- registerTestToken a "xyzw" NMInstant apnsQ
+      getTestNtfTokenPort a >>= \port -> liftIO $ port `shouldBe` ntfTestPort -- not yet changed
+      deleteNtfToken a tkn2 -- force server switch
+      Left BROKER {brokerErr = NETWORK} <- tryError $ registerTestToken a "qwer" NMInstant apnsQ -- ok, it's down for now
+      getTestNtfTokenPort a >>= \port2 -> liftIO $ port2 `shouldBe` ntfTestPort2 -- but the token got updated
+    killThread ntf
+    withNtfServerOn t ntfTestPort2 $ runRight_ $ do
+      tkn <- registerTestToken a "qwer" NMInstant apnsQ
+      checkNtfToken a tkn >>= \r -> liftIO $ r `shouldBe` NTActive
+
 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 +329,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 +354,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
@@ -319,8 +401,8 @@ registerTestToken :: AgentClient -> ByteString -> NotificationsMode -> TBQueue A
 registerTestToken a token mode apnsQ = do
  let tkn = DeviceToken PPApnsTest token
  NTRegistered <- registerNtfToken a tkn mode
-  APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData'}, sendApnsResponse = sendApnsResponse'} <-
-    atomically $ readTBQueue apnsQ
+  Just APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData'}, sendApnsResponse = sendApnsResponse'} <-
+    timeout 1000000 . atomically $ readTBQueue apnsQ
  verification' <- ntfData' .-> "verification"
  nonce' <- C.cbNonce <$> ntfData' .-> "nonce"
  liftIO $ sendApnsResponse' APNSRespOk
@@ -330,8 +412,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 +478,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 +501,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 +523,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 +551,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,12 +583,13 @@ 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
    _ <- registerTestToken a "abcd" NMInstant apnsQ
-    liftIO $ threadDelay 1500000
+    liftIO $ threadDelay 5000000
    forM_ conns $ \(aliceId, bobId) -> do
      msgId <- sendMessage b aliceId (SMP.MsgFlags True) "hello"
      get b ##> ("", aliceId, SENT msgId)
@@ -549,8 +632,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
@@ -570,9 +653,70 @@ testSwitchNotifications servers APNSMockServer {apnsQ} = do
  disconnectAgentClient a
  disconnectAgentClient b

-messageNotification :: TBQueue APNSMockRequest -> ExceptT AgentErrorType IO (C.CbNonce, ByteString)
+testNotificationsOldToken :: APNSMockServer -> IO ()
+testNotificationsOldToken APNSMockServer {apnsQ} = do
+  a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  b <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
+  c <- getSMPAgentClient' 3 agentCfg initAgentServers testDB3
+  runRight_ $ do
+    (abId, baId) <- makeConnection a b
+    let testMessageAB = testMessage_ apnsQ a abId b baId
+    _ <- registerTestToken a "abcd" NMInstant apnsQ
+    liftIO $ threadDelay 250000
+    testMessageAB "hello"
+    -- change server
+    setNtfServers a [testNtfServer2] -- server 2 isn't running now, don't use
+    -- replacing token keeps server
+    _ <- registerTestToken a "xyzw" NMInstant apnsQ
+    getTestNtfTokenPort a >>= \port -> liftIO $ port `shouldBe` ntfTestPort
+    testMessageAB "still there"
+    -- new connections keep server
+    (acId, caId) <- makeConnection a c
+    let testMessageAC = testMessage_ apnsQ a acId c caId
+    testMessageAC "greetings"
+  disconnectAgentClient a
+  disconnectAgentClient b
+  disconnectAgentClient c
+
+testNotificationsNewToken :: APNSMockServer -> ThreadId -> IO ()
+testNotificationsNewToken APNSMockServer {apnsQ} oldNtf = do
+  a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  b <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
+  c <- getSMPAgentClient' 3 agentCfg initAgentServers testDB3
+  runRight_ $ do
+    (abId, baId) <- makeConnection a b
+    let testMessageAB = testMessage_ apnsQ a abId b baId
+    tkn <- registerTestToken a "abcd" NMInstant apnsQ
+    getTestNtfTokenPort a >>= \port -> liftIO $ port `shouldBe` ntfTestPort
+    liftIO $ threadDelay 250000
+    testMessageAB "hello"
+    -- switch
+    setNtfServers a [testNtfServer2]
+    deleteNtfToken a tkn
+    _ <- registerTestToken a "abcd" NMInstant apnsQ
+    getTestNtfTokenPort a >>= \port -> liftIO $ port `shouldBe` ntfTestPort2
+    liftIO $ threadDelay 250000
+    liftIO $ killThread oldNtf
+    -- -- back to work
+    testMessageAB "hello again"
+    (acId, caId) <- makeConnection a c
+    let testMessageAC = testMessage_ apnsQ a acId c caId
+    testMessageAC "greetings"
+  disconnectAgentClient a
+  disconnectAgentClient b
+  disconnectAgentClient c
+
+testMessage_ :: HasCallStack => TBQueue APNSMockRequest -> AgentClient -> ConnId -> AgentClient -> ConnId -> SMP.MsgBody -> ExceptT AgentErrorType IO ()
+testMessage_ apnsQ a aId b bId msg = do
+  msgId <- sendMessage b aId (SMP.MsgFlags True) msg
+  get b ##> ("", aId, SENT msgId)
+  void $ messageNotification apnsQ
+  get a =##> \case ("", c, Msg msg') -> c == bId && msg == msg'; _ -> False
+  ackMessage a bId msgId Nothing
+
+messageNotification :: HasCallStack => TBQueue APNSMockRequest -> ExceptT AgentErrorType IO (C.CbNonce, ByteString)
 messageNotification apnsQ = do
-  750000 `timeout` atomically (readTBQueue apnsQ) >>= \case
+  1000000 `timeout` atomically (readTBQueue apnsQ) >>= \case
    Nothing -> error "no notification"
    Just APNSMockRequest {notification = APNSNotification {aps = APNSMutableContent {}, notificationData = Just ntfData}, sendApnsResponse} -> do
      nonce <- C.cbNonce <$> ntfData .-> "nonce"
@@ -34,6 +34,7 @@ import Simplex.Messaging.Agent.Client ()
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.Store
 import Simplex.Messaging.Agent.Store.SQLite
+import Simplex.Messaging.Agent.Store.SQLite.Common (withTransaction')
 import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
 import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
 import qualified Simplex.Messaging.Crypto as C
@@ -69,6 +70,7 @@ createEncryptedStore key keepKey = do
  -- IO operations on multiple similarly named files; error seems to be environment specific
  r <- randomIO :: IO Word32
  Right st <- createSQLiteStore (testDB <> show r) key keepKey Migrations.app MCError
+  withTransaction' st (`SQL.execute_` "INSERT INTO users (user_id) VALUES (1);")
  pure st

 removeStore :: SQLiteStore -> IO ()
@@ -7,7 +7,10 @@ import Control.DeepSeq
 import Control.Monad (unless, void)
 import Data.List (dropWhileEnd)
 import Data.Maybe (fromJust, isJust)
+import Database.SQLite.Simple (Only (..))
+import qualified Database.SQLite.Simple as SQL
 import Simplex.Messaging.Agent.Store.SQLite
+import Simplex.Messaging.Agent.Store.SQLite.Common (withTransaction')
 import Simplex.Messaging.Agent.Store.SQLite.Migrations (Migration (..), MigrationsToRun (..), toDownMigration)
 import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
 import Simplex.Messaging.Util (ifM)
@@ -28,6 +31,8 @@ schemaDumpTest :: Spec
 schemaDumpTest = do
  it "verify and overwrite schema dump" testVerifySchemaDump
  it "verify schema down migrations" testSchemaMigrations
+  it "should NOT create user record for new database" testUsersMigrationNew
+  it "should create user record for old database" testUsersMigrationOld

 testVerifySchemaDump :: IO ()
 testVerifySchemaDump = do
@@ -61,6 +66,25 @@ testSchemaMigrations = do
      schema''' <- getSchema testDB testSchema
      schema''' `shouldBe` schema'

+testUsersMigrationNew :: IO ()
+testUsersMigrationNew = do
+  Right st <- createSQLiteStore testDB "" False Migrations.app MCError
+  withTransaction' st (`SQL.query_` "SELECT user_id FROM users;")
+    `shouldReturn` ([] :: [Only Int])
+  closeSQLiteStore st
+
+testUsersMigrationOld :: IO ()
+testUsersMigrationOld = do
+  let beforeUsers = takeWhile (("m20230110_users" /=) . name) Migrations.app
+  Right st <- createSQLiteStore testDB "" False beforeUsers MCError
+  withTransaction' st (`SQL.query_` "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'users';")
+    `shouldReturn` ([] :: [Only String])
+  closeSQLiteStore st
+  Right st' <- createSQLiteStore testDB "" False Migrations.app MCYesUp
+  withTransaction' st' (`SQL.query_` "SELECT user_id FROM users;")
+    `shouldReturn` ([Only (1 :: Int)])
+  closeSQLiteStore st'
+
 skipComparisonForDownMigrations :: [String]
 skipComparisonForDownMigrations =
  [ -- on down migration idx_messages_internal_snd_id_ts index moves down to the end of the file
@@ -5,7 +5,7 @@ module CoreTests.BatchingTests (batchingTests) where
 import Control.Concurrent.STM
 import Control.Monad
 import Data.ByteString.Char8 (ByteString)
-import qualified Data.ByteString.Char8 as B
+import qualified Data.ByteString as B
 import qualified Data.List.NonEmpty as L
 import Simplex.Messaging.Client
 import qualified Simplex.Messaging.Crypto as C
@@ -20,7 +20,7 @@ batchingTests = do
    it "should batch with 90 subscriptions per batch" testBatchSubscriptions
    it "should break on message that does not fit" testBatchWithMessage
    it "should break on large message" testBatchWithLargeMessage
-  describe "batchClientTransmissions" $ do
+  describe "batchTransmissions'" $ do
    it "should batch with 90 subscriptions per batch" testClientBatchSubscriptions
    it "should break on message that does not fit" testClientBatchWithMessage
    it "should break on large message" testClientBatchWithLargeMessage
@@ -34,8 +34,8 @@ testBatchSubscriptions = do
  length batches1 `shouldBe` 200
  let batches = batchTransmissions True smpBlockSize $ L.fromList subs
  length batches `shouldBe` 3
-  [TBTransmissions n1 s1, TBTransmissions n2 s2, TBTransmissions n3 s3] <- pure batches
-  (n1, n2, n3) `shouldBe` (90, 90, 20)
+  [TBTransmissions s1 n1 _, TBTransmissions s2 n2 _, TBTransmissions s3 n3 _] <- pure batches
+  (n1, n2, n3) `shouldBe` (20, 90, 90)
  all lenOk [s1, s2, s3] `shouldBe` True

 testBatchWithMessage :: IO ()
@@ -50,8 +50,8 @@ testBatchWithMessage = do
  length batches1 `shouldBe` 101
  let batches = batchTransmissions True smpBlockSize $ L.fromList cmds
  length batches `shouldBe` 2
-  [TBTransmissions n1 s1, TBTransmissions n2 s2] <- pure batches
-  (n1, n2) `shouldBe` (60, 41)
+  [TBTransmissions s1 n1 _, TBTransmissions s2 n2 _] <- pure batches
+  (n1, n2) `shouldBe` (55, 46)
  all lenOk [s1, s2] `shouldBe` True

 testBatchWithLargeMessage :: IO ()
@@ -69,8 +69,8 @@ testBatchWithLargeMessage = do
  length batches1' `shouldBe` 160
  let batches = batchTransmissions True smpBlockSize $ L.fromList cmds
  length batches `shouldBe` 4
-  [TBTransmissions n1 s1, TBLargeTransmission, TBTransmissions n2 s2, TBTransmissions n3 s3] <- pure batches
-  (n1, n2, n3) `shouldBe` (60, 90, 10)
+  [TBTransmissions s1 n1 _, TBLargeTransmission _, TBTransmissions s2 n2 _, TBTransmissions s3 n3 _] <- pure batches
+  (n1, n2, n3) `shouldBe` (60, 10, 90)
  all lenOk [s1, s2, s3] `shouldBe` True

 testClientBatchSubscriptions :: IO ()
@@ -78,13 +78,13 @@ testClientBatchSubscriptions = do
  sessId <- atomically . C.randomBytes 32 =<< C.newRandom
  client <- atomically $ clientStub sessId
  subs <- replicateM 200 $ randomSUBCmd client
-  let batches1 = batchClientTransmissions False smpBlockSize $ L.fromList subs
-  all lenOk1' batches1 `shouldBe` True
-  let batches = batchClientTransmissions True smpBlockSize $ L.fromList subs
+  let batches1 = batchTransmissions' False smpBlockSize $ L.fromList subs
+  all lenOk1 batches1 `shouldBe` True
+  let batches = batchTransmissions' True smpBlockSize $ L.fromList subs
  length batches `shouldBe` 3
-  [CBTransmissions s1 n1 rs1, CBTransmissions s2 n2 rs2, CBTransmissions s3 n3 rs3] <- pure batches
-  (n1, n2, n3) `shouldBe` (90, 90, 20)
-  (length rs1, length rs2, length rs3) `shouldBe` (90, 90, 20)
+  [TBTransmissions s1 n1 rs1, TBTransmissions s2 n2 rs2, TBTransmissions s3 n3 rs3] <- pure batches
+  (n1, n2, n3) `shouldBe` (20, 90, 90)
+  (length rs1, length rs2, length rs3) `shouldBe` (20, 90, 90)
  all lenOk [s1, s2, s3] `shouldBe` True

 testClientBatchWithMessage :: IO ()
@@ -95,14 +95,14 @@ testClientBatchWithMessage = do
  send <- randomSENDCmd client 8000
  subs2 <- replicateM 40 $ randomSUBCmd client
  let cmds = subs1 <> [send] <> subs2
-      batches1 = batchClientTransmissions False smpBlockSize $ L.fromList cmds
-  all lenOk1' batches1 `shouldBe` True
+      batches1 = batchTransmissions' False smpBlockSize $ L.fromList cmds
+  all lenOk1 batches1 `shouldBe` True
  length batches1 `shouldBe` 101
-  let batches = batchClientTransmissions True smpBlockSize $ L.fromList cmds
+  let batches = batchTransmissions' True smpBlockSize $ L.fromList cmds
  length batches `shouldBe` 2
-  [CBTransmissions s1 n1 rs1, CBTransmissions s2 n2 rs2] <- pure batches
-  (n1, n2) `shouldBe` (60, 41)
-  (length rs1, length rs2) `shouldBe` (60, 41)
+  [TBTransmissions s1 n1 rs1, TBTransmissions s2 n2 rs2] <- pure batches
+  (n1, n2) `shouldBe` (55, 46)
+  (length rs1, length rs2) `shouldBe` (55, 46)
  all lenOk [s1, s2] `shouldBe` True

 testClientBatchWithLargeMessage :: IO ()
@@ -113,26 +113,26 @@ testClientBatchWithLargeMessage = do
  send <- randomSENDCmd client 17000
  subs2 <- replicateM 100 $ randomSUBCmd client
  let cmds = subs1 <> [send] <> subs2
-      batches1 = batchClientTransmissions False smpBlockSize $ L.fromList cmds
-  all lenOk1' batches1 `shouldBe` False
+      batches1 = batchTransmissions' False smpBlockSize $ L.fromList cmds
+  all lenOk1 batches1 `shouldBe` False
  length batches1 `shouldBe` 161
  let batches1' = take 60 batches1 <> drop 61 batches1
-  all lenOk1' batches1' `shouldBe` True
+  all lenOk1 batches1' `shouldBe` True
  length batches1' `shouldBe` 160
  --
-  let batches = batchClientTransmissions True smpBlockSize $ L.fromList cmds
+  let batches = batchTransmissions' True smpBlockSize $ L.fromList cmds
  length batches `shouldBe` 4
-  [CBTransmissions s1 n1 rs1, CBLargeTransmission _, CBTransmissions s2 n2 rs2, CBTransmissions s3 n3 rs3] <- pure batches
-  (n1, n2, n3) `shouldBe` (60, 90, 10)
-  (length rs1, length rs2, length rs3) `shouldBe` (60, 90, 10)
+  [TBTransmissions s1 n1 rs1, TBLargeTransmission _, TBTransmissions s2 n2 rs2, TBTransmissions s3 n3 rs3] <- pure batches
+  (n1, n2, n3) `shouldBe` (60, 10, 90)
+  (length rs1, length rs2, length rs3) `shouldBe` (60, 10, 90)
  all lenOk [s1, s2, s3] `shouldBe` True
  --
  let cmds' = [send] <> subs1 <> subs2
-  let batches' = batchClientTransmissions True smpBlockSize $ L.fromList cmds'
+  let batches' = batchTransmissions' True smpBlockSize $ L.fromList cmds'
  length batches' `shouldBe` 3
-  [CBLargeTransmission _, CBTransmissions s1' n1' rs1', CBTransmissions s2' n2' rs2'] <- pure batches'
-  (n1', n2') `shouldBe` (90, 70)
-  (length rs1', length rs2') `shouldBe` (90, 70)
+  [TBLargeTransmission _, TBTransmissions s1' n1' rs1', TBTransmissions s2' n2' rs2'] <- pure batches'
+  (n1', n2') `shouldBe` (70, 90)
+  (length rs1', length rs2') `shouldBe` (70, 90)
  all lenOk [s1', s2'] `shouldBe` True

 randomSUB :: ByteString -> IO (Maybe C.ASignature, ByteString)
@@ -172,12 +172,7 @@ randomSENDCmd c len = do
 lenOk :: ByteString -> Bool
 lenOk s = 0 < B.length s && B.length s <= smpBlockSize - 2

-lenOk1 :: TransportBatch -> Bool
+lenOk1 :: TransportBatch r -> Bool
 lenOk1 = \case
-  TBTransmission s -> lenOk s
-  _ -> False
-
-lenOk1' :: ClientBatch err msg -> Bool
-lenOk1' = \case
-  CBTransmission s _ -> lenOk s
+  TBTransmission s _ -> lenOk s
  _ -> False
@@ -57,6 +57,9 @@ testHost = "localhost"
 ntfTestPort :: ServiceName
 ntfTestPort = "6001"

+ntfTestPort2 :: ServiceName
+ntfTestPort2 = "6002"
+
 apnsTestPort :: ServiceName
 apnsTestPort = "6010"

@@ -77,7 +80,7 @@ testNtfClient client = do
 ntfServerCfg :: NtfServerConfig
 ntfServerCfg =
  NtfServerConfig
-    { transports = undefined,
+    { transports = [],
      subIdBytes = 24,
      regCodeBytes = 32,
      clientQSize = 1,
@@ -105,16 +108,19 @@ ntfServerCfg =
    }

 withNtfServerStoreLog :: ATransport -> (ThreadId -> IO a) -> IO a
-withNtfServerStoreLog t = withNtfServerCfg t ntfServerCfg {storeLogFile = Just ntfTestStoreLogFile}
+withNtfServerStoreLog t = withNtfServerCfg ntfServerCfg {storeLogFile = Just ntfTestStoreLogFile, transports = [(ntfTestPort, t)]}

 withNtfServerThreadOn :: ATransport -> ServiceName -> (ThreadId -> IO a) -> IO a
-withNtfServerThreadOn t port' = withNtfServerCfg t ntfServerCfg {transports = [(port', t)]}
+withNtfServerThreadOn t port' = withNtfServerCfg ntfServerCfg {transports = [(port', t)]}

-withNtfServerCfg :: ATransport -> NtfServerConfig -> (ThreadId -> IO a) -> IO a
-withNtfServerCfg t cfg =
-  serverBracket
-    (\started -> runNtfServerBlocking started cfg {transports = [(ntfTestPort, t)]})
-    (pure ())
+withNtfServerCfg :: HasCallStack => NtfServerConfig -> (ThreadId -> IO a) -> IO a
+withNtfServerCfg cfg@NtfServerConfig {transports} =
+  case transports of
+    [] -> error "no transports configured"
+    _ ->
+      serverBracket
+        (\started -> runNtfServerBlocking started cfg)
+        (pure ())

 withNtfServerOn :: ATransport -> ServiceName -> IO a -> IO a
 withNtfServerOn t port' = withNtfServerThreadOn t port' . const
@@ -136,7 +142,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
@@ -1,6 +1,7 @@
 {-# LANGUAGE ConstraintKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE GADTs #-}
+{-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE NumericUnderscores #-}
 {-# LANGUAGE OverloadedLists #-}
 {-# LANGUAGE OverloadedStrings #-}
@@ -9,12 +10,14 @@

 module SMPAgentClient where

+import Control.Monad
 import Control.Monad.IO.Unlift
 import Crypto.Random
 import qualified Data.ByteString.Char8 as B
 import Data.List.NonEmpty (NonEmpty)
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
+import qualified Database.SQLite.Simple as SQL
 import Network.Socket (ServiceName)
 import NtfClient (ntfTestPort)
 import SMPClient
@@ -30,10 +33,11 @@ import Simplex.Messaging.Agent.Env.SQLite
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval
 import Simplex.Messaging.Agent.Server (runSMPAgentBlocking)
-import Simplex.Messaging.Agent.Store.SQLite (MigrationConfirmation (..))
+import Simplex.Messaging.Agent.Store.SQLite (MigrationConfirmation (..), SQLiteStore (dbNew))
+import Simplex.Messaging.Agent.Store.SQLite.Common (withTransaction')
 import Simplex.Messaging.Client (ProtocolClientConfig (..), chooseTransportHost, defaultClientConfig, defaultNetworkConfig)
 import Simplex.Messaging.Parsers (parseAll)
-import Simplex.Messaging.Protocol (ProtoServerWithAuth)
+import Simplex.Messaging.Protocol (NtfServer, ProtoServerWithAuth)
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Client
 import Test.Hspec
@@ -176,11 +180,17 @@ testSMPServer = "smp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:50
 testSMPServer2 :: SMPServer
 testSMPServer2 = "smp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:5002"

+testNtfServer :: NtfServer
+testNtfServer = "ntf://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:6001"
+
+testNtfServer2 :: NtfServer
+testNtfServer2 = "ntf://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:6002"
+
 initAgentServers :: InitialAgentServers
 initAgentServers =
  InitialAgentServers
    { smp = userServers [noAuthSrv testSMPServer],
-      ntf = ["ntf://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:6001"],
+      ntf = [testNtfServer],
      xftp = userServers [noAuthSrv testXFTPServer],
      netCfg = defaultNetworkConfig {tcpTimeout = 500_000, tcpConnectTimeout = 500_000}
    }
@@ -194,27 +204,36 @@ agentCfg =
    { tcpPort = agentTestPort,
      tbqSize = 4,
      -- database = testDB,
-      smpCfg = defaultClientConfig {qSize = 1, defaultTransport = (testPort, transport @TLS)},
-      ntfCfg = defaultClientConfig {qSize = 1, defaultTransport = (ntfTestPort, transport @TLS)},
-      reconnectInterval = defaultReconnectInterval {initialInterval = 50_000},
+      smpCfg = defaultClientConfig {qSize = 1, defaultTransport = (testPort, transport @TLS), networkConfig},
+      ntfCfg = defaultClientConfig {qSize = 1, defaultTransport = (ntfTestPort, transport @TLS), networkConfig},
+      reconnectInterval = fastRetryInterval,
      xftpNotifyErrsOnRetry = False,
-      ntfWorkerDelay = 1000,
-      ntfSMPWorkerDelay = 1000,
+      ntfWorkerDelay = 100,
+      ntfSMPWorkerDelay = 100,
      caCertificateFile = "tests/fixtures/ca.crt",
      privateKeyFile = "tests/fixtures/server.key",
      certificateFile = "tests/fixtures/server.crt"
    }
+  where
+    networkConfig = defaultNetworkConfig {tcpConnectTimeout = 3_000_000, tcpTimeout = 2_000_000}
+
+fastRetryInterval :: RetryInterval
+fastRetryInterval = defaultReconnectInterval {initialInterval = 50_000}
+
+fastMessageRetryInterval :: RetryInterval2
+fastMessageRetryInterval = RetryInterval2 {riFast = fastRetryInterval, riSlow = fastRetryInterval}

 type AgentTestMonad m = (MonadUnliftIO m, MonadRandom m, MonadFail m)

-withSmpAgentThreadOn_ :: AgentTestMonad m => ATransport -> (ServiceName, ServiceName, FilePath) -> m () -> (ThreadId -> m a) -> m a
-withSmpAgentThreadOn_ t (port', smpPort', db') afterProcess =
+withSmpAgentThreadOn_ :: AgentTestMonad m => ATransport -> (ServiceName, ServiceName, FilePath) -> Int -> m () -> (ThreadId -> m a) -> m a
+withSmpAgentThreadOn_ t (port', smpPort', db') initClientId afterProcess =
  let cfg' = agentCfg {tcpPort = port'}
      initServers' = initAgentServers {smp = userServers [ProtoServerWithAuth (SMPServer "localhost" smpPort' testKeyHash) Nothing]}
   in serverBracket
        ( \started -> do
            Right st <- liftIO $ createAgentStore db' "" False MCError
-            runSMPAgentBlocking t cfg' initServers' st started
+            when (dbNew st) . liftIO $ withTransaction' st (`SQL.execute_` "INSERT INTO users (user_id) VALUES (1)")
+            runSMPAgentBlocking t cfg' initServers' st initClientId started
        )
        afterProcess

@@ -222,7 +241,7 @@ userServers :: NonEmpty (ProtoServerWithAuth p) -> Map UserId (NonEmpty (ProtoSe
 userServers srvs = M.fromList [(1, srvs)]

 withSmpAgentThreadOn :: AgentTestMonad m => ATransport -> (ServiceName, ServiceName, FilePath) -> (ThreadId -> m a) -> m a
-withSmpAgentThreadOn t a@(_, _, db') = withSmpAgentThreadOn_ t a $ removeFile db'
+withSmpAgentThreadOn t a@(_, _, db') = withSmpAgentThreadOn_ t a 0 $ removeFile db'

 withSmpAgentOn :: AgentTestMonad m => ATransport -> (ServiceName, ServiceName, FilePath) -> m a -> m a
 withSmpAgentOn t (port', smpPort', db') = withSmpAgentThreadOn t (port', smpPort', db') . const
@@ -130,7 +130,7 @@ serverBracket process afterProcess f = do
  E.bracket
    (forkIOWithUnmask ($ process started))
    (\t -> killThread t >> afterProcess >> waitFor started "stop")
-    (\t -> waitFor started "start" >> f t)
+    (\t -> waitFor started "start" >> f t >>= \r -> r <$ threadDelay 100000)
  where
    waitFor started s =
      5_000_000 `timeout` atomically (takeTMVar started) >>= \case
@@ -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
@@ -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)
@@ -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
Author	SHA1	Message	Date
Evgeny Poberezkin	a516c2f72c	5.5.1.2	2024-02-02 08:19:02 +00:00
Evgeny Poberezkin	2ae1c9f79d	agent: only retry connecting SMP clients when there are pending subscriptions (#981 ) * agent: only retry connecting SMP clients when there are pending subscriptions * fix * remove retry on creating clients * simplify	2024-02-01 16:17:37 +00:00
Alexander BondarenkoandEvgeny Poberezkin	24b84106a6	core: add activity tests to workers details (#980 ) * core: add activity tests to workers details * remove names --------- Co-authored-by: Evgeny Poberezkin <evgeny@poberezkin.com>	2024-02-01 11:02:31 +00:00
Alexander BondarenkoandEvgeny Poberezkin	15bc027f23	core: fix reconnection bugs (#979 ) * core: fix reconnection bugs * untangle newProtocolClient * refactor * report busy clientVar error * log error * comments --------- Co-authored-by: Evgeny Poberezkin <evgeny@poberezkin.com>	2024-02-01 10:55:50 +00:00
Alexander Bondarenko	cb64dabf75	tests: fix sporadic failures from reading stats (#975 )	2024-01-30 11:36:07 +00:00
Evgeny Poberezkin	7a0cd8041b	5.5.1.1	2024-01-22 14:06:52 +00:00
Evgeny Poberezkinandspaced4ndy	fd4eeb36db	agent: optimize expired messages query (#976 ) * agent: optimize expired messages query * schema * fix query * fix * typo Co-authored-by: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com> * fix * refactor * comment * refactor2 --------- Co-authored-by: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com>	2024-01-22 14:04:57 +00:00
Evgeny Poberezkin	1e49f1c92d	5.5.1.0	2024-01-19 19:47:39 +00:00
Alexander Bondarenko Evgeny Poberezkin Evgeny Poberezkin	f7cdec2f08	fix: support multiple notification servers in configuration (#971 ) * tests: add trpl-keeps-server check * add smp server switch check * add connection test and fix withNtfServer * Update src/Simplex/Messaging/Agent/NtfSubSupervisor.hs Co-authored-by: Evgeny Poberezkin <e.poberezkin@me.com> * use ntfServer from token * rename --------- Co-authored-by: Evgeny Poberezkin <e.poberezkin@me.com> Co-authored-by: Evgeny Poberezkin <evgeny@poberezkin.com>	2024-01-19 19:42:28 +00:00
Evgeny Poberezkin	8ff89c19dc	servers: do not keep stats file open during delays (#974 )	2024-01-19 17:47:47 +00:00
Evgeny Poberezkin	baf2c47065	agent: expire messages failed after quota exceeded after 7 days, and expire multiple messages at once (#973 ) * agent: expire messages failed after quota exceeded after 7 days, and expire multiple messages at once * fix, test * refactor * catch in loop	2024-01-19 17:21:25 +00:00
Alexander Bondarenko	40fa34c2d5	tests: test multiple Ntf servers (#966 ) * tests: add ntf case with multiple Ntf servers * simplify test * fix for master * add server switch test * add server switch test * add message test for ntf server switch	2024-01-17 19:47:47 +00:00
Alexander Bondarenko	eb41abfb8f	tests: fix random order in async commands case (#969 )	2024-01-17 11:48:41 +00:00
Evgeny Poberezkin	b547f34cc0	Revert "Revert "agent: do not create user record in new databases (#957 )" (#963 )" (#964 ) * Revert "Revert "agent: do not create user record in new databases (#957)" (#963)" This reverts commit `aee9088417`. * tests * remove logging	2024-01-17 11:01:04 +00:00
Evgeny Poberezkin	f6ed4640d4	5.5.0.6	2024-01-16 17:07:15 +00:00
Evgeny Poberezkin	a0b35cec4f	agent: fix potential race when good client can be removed instead of bad for the same transport session (#967 ) * agent: fix potential race when good client can be removed instead of bad for the same transport session * tryAgentError * case	2024-01-16 13:45:51 +00:00
Evgeny Poberezkin	00c4ff4a21	agent: fix race condition in subscription worker (results in stuck delivery), better error handling in batch operations (#962 ) * debugging test: join connection when reply queue creation fails * more logs * more logs * more logs, fixed bug * cleanup * catch all errors * simplify * comment * remove client ID counter	2024-01-16 10:29:35 +00:00
Evgeny Poberezkin	aee9088417	Revert "agent: do not create user record in new databases (#957 )" (#963 ) This reverts commit `9b9a0bd0df`.	2024-01-14 22:28:28 +00:00
Evgeny Poberezkin	7f7a77c4eb	SMP protocol: optimize batching transactions, remove Builder (#961 ) * remove Builder * fewer chunks * remove lazy bytestrings * optimize * pad	2024-01-14 20:42:47 +00:00
Alexander Bondarenko	cd4329f2de	agent: collect worker stats (#959 ) * agent: collect worker stats * add more workers * process asyncCmdWorkers as a Map	2024-01-12 23:38:35 +00:00
Alexander BondarenkoandEvgeny Poberezkin	68f5e189a6	smp: check for progress when resubscribing (#956 ) * smp: check for progress when resubscribing * add allowClose to reconnectServer to distinguish entry points * resolve error todo * make reconnect async per-session * remove allowClose * deregister reconnecter when it finishes * signal/react more work explicitly * fix restart condition * wait for reconnecter to finish * remove redundant reconnect locks * rename getClientVar for expanded scope * formatting * remove withPending * move pending check to tryReconnectSMPClient loop * combine pending check and slot release transactions * actually reserve the slot for async * simplify * refactor * refactor * use removeClientVar * rename * refactor * test * reduce MonadError scope --------- Co-authored-by: Evgeny Poberezkin <2769109+epoberezkin@users.noreply.github.com>	2024-01-12 19:29:18 +00:00
Evgeny Poberezkin	17f64e1565	docs: URI schemes (#958 )	2024-01-12 17:21:57 +00:00
Evgeny Poberezkin	9b9a0bd0df	agent: do not create user record in new databases (#957 ) * agent: do not create user record in new databases * update	2024-01-12 16:57:57 +00:00
Evgeny Poberezkin	ad8cd1d515	5.5.0.5	2024-01-09 19:44:15 +00:00
Evgeny Poberezkin	ca527b4d6c	use Builder for batching commands (#953 ) * use Builder for batching commands * refactor * refactor * increase timeout in test * do not materialize builder * use concat * debug test * debug2 * delays * more delay, less worker delay * remove logs * reduce delay * tests * sized builder * builder * comment * remove unsafe * style * remove fromIntegral	2024-01-08 22:54:27 +00:00