Merge branch 'master' into ep/conc-msgs

plans/2026-05-18-parallel-message-processing.md

@@ -0,0 +1,161 @@
+# Parallel Message Processing - Eliminate Single-Thread Bottlenecks
+
+## Problem
+
+Message reception flows through two single-thread bottlenecks:
+
+1. **Agent `msgQ` bottleneck**: Multiple SMP server connections write to one shared `TBQueue` (`AgentClient.msgQ` / `SMPClientAgent.msgQ`). A single `subscriber` thread reads and processes all messages sequentially - DB lookups, double-ratchet decryption, DB writes - regardless of which connection they came from.
+
+2. **Chat `subQ` bottleneck**: The agent's `subscriber` thread writes processed events to one shared `TBQueue` (`AgentClient.subQ`). A single `agentSubscriber` thread in simplex-chat reads and processes all events sequentially.
+
+Both bottlenecks serialize work that could run in parallel, since messages from different connections are independent.
+
+## Solution
+
+Replace queues with callbacks at both layers. The producer calls a processing function directly in its own thread.
+
+### Layer 1: SMP client - eliminate `msgQ`
+
+**Current flow:**
+```
+SMP connection thread -> writeTBQueue msgQ -> subscriber thread -> processSMPTransmissions
+```
+
+**New flow:**
+```
+SMP connection thread -> processMsg callback (with per-client MVar lock)
+```
+
+**Why the MVar lock:** Within one SMP client, two threads produce messages:
+- The receive loop (`processMsgs` in `Client.hs:686`)
+- `writeSMPMessage` (`Client.hs:874`) - called from `processSUBResponse_` when a SUB response includes an inline MSG
+
+These two must be serialized within one client. An MVar lock ensures they take turns calling the callback. Across different clients (different server connections), no lock is shared - natural parallelism.
+
+#### Changes
+
+**`src/Simplex/Messaging/Client.hs`:**
+- In `PClient`: replace `msgQ :: Maybe (TBQueue ...)` with `processServerMsg :: Maybe (ServerTransmissionBatch v err msg -> IO ())` and `processLock :: MVar ()`
+- `processMsgs`: acquire `processLock`, call `processServerMsg` with the batch
+- `writeSMPMessage`: acquire `processLock`, call `processServerMsg`
+- `getProtocolClient`: takes `Maybe (ServerTransmissionBatch v err msg -> IO ())` instead of `Maybe (TBQueue ...)`
+- `smpClientStub`: sets `processServerMsg = Nothing`
+- `serverTransmission`: unchanged
+
+**`src/Simplex/Messaging/Agent/Client.hs`:**
+- Remove `msgQ` field from `AgentClient`
+- `smpConnectClient`: pass `processSMPTransmissions` wrapper as callback instead of `Just msgQ`
+- Remove `AgentQueuesInfo` and `getAgentQueuesInfo` entirely (dead with no queues to monitor)
+- Add `inflightCallbacks :: TVar Int` for monitoring instead - increment before callback, decrement in bracket
+
+**`src/Simplex/Messaging/Agent.hs`:**
+- Remove `subscriber` function
+- Remove `subscriber` from `runAgentThreads`
+- `processSMPTransmissions` stays, called directly from SMP client threads
+- `agentOperationBracket c AORcvNetwork` moves into the callback wrapper
+- Exception handling: wrap callback with `catchOwn` matching current `subscriber`'s error handling
+
+**`src/Simplex/Messaging/Client/Agent.hs`:**
+- `SMPClientAgent`: replace `msgQ` with callback field `processServerMsg :: ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> IO ()`
+- `newSMPClientAgent`: takes callback parameter instead of creating `msgQ`
+- `connectClient`: passes callback to `getProtocolClient`
+
+**`src/Simplex/Messaging/Notifications/Server.hs`:**
+- `ntfSubscriber`: remove `receiveSMP` loop; the processing logic becomes the callback passed via `SMPClientAgent`
+- Processing stays in M (via `UnliftIO` or pre-bound env)
+
+**Tests (`tests/SMPProxyTests.hs`):**
+- 2 sites: change `getProtocolClient ... (Just msgQ) ...` to pass a callback that writes to a local test TBQueue
+
+### Layer 2: Agent to chat - eliminate `subQ`
+
+**Current flow:**
+```
+agent processSMPTransmissions -> writeTBQueue subQ -> chat agentSubscriber -> process
+```
+
+**New flow:**
+```
+agent processSMPTransmissions -> processEvent callback [events]
+```
+
+**Key design decisions:**
+- Callback takes `[ATransmission]` (list), not single event. All events from one connection batch are passed together to maintain ordering within a connection.
+- Error notifications (currently `nonBlockingWriteTBQueue`) use `forkIO $ callback [event]` - fire-and-forget, order doesn't matter for errors.
+- The `isFullTBQueue subQ` / pending mechanism disappears - the callback receives the full list directly, no need to buffer/flush.
+- `AgentClient` keeps `testQ :: Maybe (TBQueue ATransmission)` for tests only.
+
+#### Changes
+
+**`src/Simplex/Messaging/Agent/Client.hs`:**
+- Replace `subQ :: TBQueue ATransmission` with:
+  - `processEvent :: [ATransmission] -> IO ()` - callback, accepts event list
+  - `testQ :: Maybe (TBQueue ATransmission)` - test-only, `Nothing` in production
+- Remove `AgentQueuesInfo` / `getAgentQueuesInfo`
+- Add `inflightCallbacks :: TVar Int` with bracket: `withInflight c $ processEvent c events`
+
+**`src/Simplex/Messaging/Agent.hs`:**
+- `processSMPTransmissions`: accumulate events in a local list (currently uses `pendingMsgs` TVar + flush pattern). Call `processEvent` once at end with the full list.
+- `runCommandProcessing`: same - call `processEvent` once with all events for the command batch. Remove `isFullTBQueue`/pending logic.
+- All `notify`/`notify'` helpers within `processSMPTransmissions` write to a local `TVar [ATransmission]` instead of directly to `subQ`. Flushed at end as single `processEvent` call.
+- Error sites (currently `nonBlockingWriteTBQueue`): use `forkIO $ processEvent c [event]`
+- Other direct `writeTBQueue subQ` sites (CONNECT/DISCONNECT events, SUSPENDED, etc.): call `processEvent c [event]` directly.
+- Remove `subscriber` function entirely.
+- Exception safety: `processEvent` call wrapped in bracket that catches "own" exceptions and logs them.
+
+**`src/Simplex/Messaging/Agent/Client.hs`:**
+- `notifySub'` (line 838): change to `forkIO $ processEvent c [event]` (non-blocking error notification)
+
+**`src/Simplex/Messaging/Agent/NtfSubSupervisor.hs`:**
+- 1 site: change `nonBlockingWriteTBQueue subQ event` to `forkIO $ processEvent c [event]`
+
+**`src/Simplex/FileTransfer/Agent.hs`:**
+- 1 site (line 351): `notify` helper changes to `processEvent c [event]`
+
+**`simplex-chat/src/Simplex/Chat/Library/Commands.hs`:**
+- Remove `agentSubscriber` thread
+- Pass chat's `process` function (adapted to accept `[ATransmission]`) as `processEvent` callback at agent initialization
+
+**Tests:**
+- `pGet` changes from `readTBQueue (subQ c)` to `readTBQueue (fromJust $ testQ c)` - 1 line
+- Agent test setup: `processEvent = mapM_ (atomically . writeTBQueue q)` where `q` is `testQ`
+- ~348 test call sites unchanged
+
+## Concurrency Safety
+
+- **Per-SMP-connection:** MVar in each SMP client serializes `processMsgs` and `writeSMPMessage`
+- **Cross-connection:** Different SMP clients have different MVars, run in different threads - fully parallel
+- **Per-connection-id:** `withConnLock connId` in `processSMPTransmissions` handles per-connection locking
+- **Chat callback:** Must be safe for concurrent calls from different agent threads. Chat dispatches by entity type and connection ID; individual handlers use their own locks.
+- **Exception safety:** Callback wrapped with bracket pattern - catches own exceptions, logs, decrements inflight counter. Exceptions don't kill SMP client threads.
+
+## Implementation Order
+
+Both layers change in one PR since they share `Client.hs`.
+
+### Phase 1: SMP client callback (`Client.hs` + both agent types)
+
+- [ ] 1.1 `Client.hs`: Replace `msgQ` with `processServerMsg` callback + `processLock` MVar in `PClient`
+- [ ] 1.2 `Client.hs`: Update `processMsgs`, `writeSMPMessage`, `getProtocolClient`, `smpClientStub`
+- [ ] 1.3 `Client/Agent.hs`: Replace `msgQ` in `SMPClientAgent` with callback field, update `newSMPClientAgent`, `connectClient`
+- [ ] 1.4 `Agent/Client.hs`: Remove `msgQ` from `AgentClient`, update `smpConnectClient` to pass `processSMPTransmissions` as callback
+- [ ] 1.5 `Agent.hs`: Remove `subscriber` thread from `runAgentThreads`, add exception wrapper to callback
+- [ ] 1.6 `Notifications/Server.hs`: Convert `receiveSMP` from loop to callback passed to `SMPClientAgent`
+- [ ] 1.7 `SMPProxyTests.hs`: Update 2 call sites to use callback + local test queue
+
+### Phase 2: Agent event callback (`subQ` -> `processEvent`)
+
+- [ ] 2.1 `Agent/Client.hs`: Add `processEvent :: [ATransmission] -> IO ()` and `testQ :: Maybe (TBQueue ATransmission)`, remove `subQ`, remove `AgentQueuesInfo`
+- [ ] 2.2 `Agent.hs`: Rewrite `processSMPTransmissions` to accumulate events in local list and call `processEvent` once at end
+- [ ] 2.3 `Agent.hs`: Update `runCommandProcessing` - remove pending/isFullTBQueue pattern, call `processEvent` with list
+- [ ] 2.4 `Agent.hs`, `Agent/Client.hs`, `NtfSubSupervisor.hs`, `FileTransfer/Agent.hs`: Update all `writeTBQueue subQ` / `nonBlockingWriteTBQueue subQ` sites (~32 total)
+- [ ] 2.5 `Agent/Client.hs`: Add inflight counter with bracket
+- [ ] 2.6 Update `pGet` to read from `testQ` (1 line), update test agent setup
+- [ ] 2.7 `simplex-chat`: Pass chat's `process` as callback, remove `agentSubscriber`
+- [ ] 2.8 Fix any multi-server test ordering issues
+
+## Risks
+
+- **Chat thread safety:** Chat's `process` may not be safe for concurrent calls. Audit needed.
+- **Backpressure:** Slow callback blocks SMP client receive thread. Acceptable - the connection that produced the message waits. Cross-connection interference eliminated.
+- **Ordering:** Within one SMP connection - preserved (MVar + list callback). Across connections - non-deterministic (same as today, since `msgQ` interleaving was arbitrary). Most tests use 1 server.

simplexmq.cabal

@@ -1,7 +1,7 @@
 cabal-version: 3.0
 
 name:           simplexmq
-version:        7.0.1.0
+version:        7.0.0.5
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and

src/Simplex/FileTransfer/Agent.hs

@@ -348,7 +348,7 @@ xftpDeleteRcvFiles' c rcvFileEntityIds = do
     batchFiles f rcvFiles = withStoreBatch' c $ \db -> map (\RcvFile {rcvFileId} -> f db rcvFileId) rcvFiles
 
 notify :: forall m e. (MonadIO m, AEntityI e) => AgentClient -> AEntityId -> AEvent e -> m ()
-notify c entId cmd = atomically $ writeTBQueue (subQ c) ("", entId, AEvt (sAEntity @e) cmd)
+notify c entId cmd = liftIO $ notifyEvent c ("", entId, AEvt (sAEntity @e) cmd)
 
 xftpSendFile' :: AgentClient -> UserId -> CryptoFile -> Int -> AM SndFileId
 xftpSendFile' c userId file numRecipients = do

src/Simplex/Messaging/Agent.hs

@@ -40,6 +40,7 @@ module Simplex.Messaging.Agent
     vrValue,
     getSMPAgentClient,
     getSMPAgentClient_,
+    startSMPAgentClient,
     disconnectAgentClient,
     disposeAgentClient,
     resumeAgentClient,
@@ -201,7 +202,7 @@ import Simplex.Messaging.Agent.Store.Entity
 import Simplex.Messaging.Agent.Store.Interface (closeDBStore, execSQL, getCurrentMigrations)
 import Simplex.Messaging.Agent.Store.Shared (UpMigration (..), upMigration)
 import qualified Simplex.Messaging.Agent.TSessionSubs as SS
-import Simplex.Messaging.Client (NetworkRequestMode (..), ProtocolClientError (..), SMPClientError, ServerTransmission (..), ServerTransmissionBatch, TransportSessionMode (..), nonBlockingWriteTBQueue, smpErrorClientNotice, temporaryClientError, unexpectedResponse)
+import Simplex.Messaging.Client (NetworkRequestMode (..), ProtocolClientError (..), SMPClientError, ServerTransmission (..), ServerTransmissionBatch, TransportSessionMode (..), smpErrorClientNotice, temporaryClientError, unexpectedResponse)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.File (CryptoFile, CryptoFileArgs)
 import Simplex.Messaging.Crypto.Ratchet (PQEncryption, PQSupport (..), pattern PQEncOff, pattern PQEncOn, pattern PQSupportOff, pattern PQSupportOn)
@@ -259,41 +260,44 @@ import UnliftIO.STM
 type AE a = ExceptT AgentErrorType IO a
 
 -- | Creates an SMP agent client instance
-getSMPAgentClient :: AgentConfig -> InitialAgentServers -> DBStore -> Bool -> AE AgentClient
+getSMPAgentClient :: AgentConfig -> InitialAgentServers -> DBStore -> (Bool -> ATransmission -> IO ()) -> AE AgentClient
 getSMPAgentClient = getSMPAgentClient_ 1
 {-# INLINE getSMPAgentClient #-}
 
-getSMPAgentClient_ :: Int -> AgentConfig -> InitialAgentServers -> DBStore -> Bool -> AE AgentClient
-getSMPAgentClient_ clientId cfg initServers@InitialAgentServers {smp, xftp, netCfg, useServices, presetServers} store backgroundMode = do
+getSMPAgentClient_ :: Int -> AgentConfig -> InitialAgentServers -> DBStore -> (Bool -> ATransmission -> IO ()) -> AE AgentClient
+getSMPAgentClient_ clientId cfg initServers@InitialAgentServers {smp, xftp, netCfg, useServices, presetServers} store processEvent = do
   -- This error should be prevented in the app
   when (any id useServices && sessionMode netCfg == TSMEntity) $ throwE $ CMD PROHIBITED "newAgentClient"
-  liftIO $ newSMPAgentEnv cfg store >>= runReaderT runAgent
+  liftIO $ newSMPAgentEnv cfg store >>= runReaderT createAgent
   where
-    runAgent = do
+    createAgent = do
       liftIO $ checkServers "SMP" smp >> checkServers "XFTP" xftp
       currentTs <- liftIO getCurrentTime
       notices <- liftIO $ withTransaction store (`getClientNotices` presetServers) `catchAll_` pure []
-      c@AgentClient {acThread} <- liftIO . newAgentClient clientId initServers currentTs notices =<< ask
-      t <- runAgentThreads c `forkFinally` const (liftIO $ disconnectAgentClient c)
-      atomically . writeTVar acThread . Just =<< mkWeakThreadId t
-      pure c
+      env <- ask
+      let processMsg c t = subscriber c t `runReaderT` env
+      liftIO $ newAgentClient clientId initServers currentTs notices processEvent processMsg env
     checkServers protocol srvs =
       forM_ (M.assocs srvs) $ \(userId, srvs') -> checkUserServers ("getSMPAgentClient " <> protocol <> " " <> tshow userId) srvs'
-    runAgentThreads c
-      | backgroundMode = run c "subscriber" $ subscriber c
-      | otherwise = do
-          restoreServersStats c
-          raceAny_
-            [ run c "subscriber" $ subscriber c,
-              run c "runNtfSupervisor" $ runNtfSupervisor c,
-              run c "cleanupManager" $ cleanupManager c,
-              run c "logServersStats" $ logServersStats c
-            ]
-            `E.finally` saveServersStats c
-    run AgentClient {subQ, acThread} name a =
+
+startSMPAgentClient :: AgentClient -> Bool -> IO ()
+startSMPAgentClient c@AgentClient {acThread, agentEnv} backgroundMode = do
+  unless backgroundMode $ do
+    t <- runAgentThreads `forkFinally` const (disconnectAgentClient c)
+    atomically . writeTVar acThread . Just =<< mkWeakThreadId t
+  where
+    runAgentThreads = flip runReaderT agentEnv $ do
+      restoreServersStats c
+      raceAny_
+        [ run "runNtfSupervisor" $ runNtfSupervisor c,
+          run "cleanupManager" $ cleanupManager c,
+          run "logServersStats" $ logServersStats c
+        ]
+        `E.finally` saveServersStats c
+    run name a =
       a `E.catchAny` \e -> whenM (isJust <$> readTVarIO acThread) $ do
         logError $ "Agent thread " <> name <> " crashed: " <> tshow e
-        atomically $ writeTBQueue subQ ("", "", AEvt SAEConn $ ERR $ CRITICAL True $ show e)
+        liftIO $ notifyEvent c ("", "", AEvt SAEConn $ ERR $ CRITICAL True $ show e)
 
 logServersStats :: AgentClient -> AM' ()
 logServersStats c = do
@@ -306,19 +310,19 @@ logServersStats c = do
     liftIO $ threadDelay' int
 
 saveServersStats :: AgentClient -> AM' ()
-saveServersStats c@AgentClient {subQ, smpServersStats, xftpServersStats, ntfServersStats} = do
+saveServersStats c@AgentClient {smpServersStats, xftpServersStats, ntfServersStats} = do
   sss <- mapM (liftIO . getAgentSMPServerStats) =<< readTVarIO smpServersStats
   xss <- mapM (liftIO . getAgentXFTPServerStats) =<< readTVarIO xftpServersStats
   nss <- mapM (liftIO . getAgentNtfServerStats) =<< readTVarIO ntfServersStats
   let stats = AgentPersistedServerStats {smpServersStats = sss, xftpServersStats = xss, ntfServersStats = OptionalMap nss}
   tryAllErrors' (withStore' c (`updateServersStats` stats)) >>= \case
-    Left e -> atomically $ writeTBQueue subQ ("", "", AEvt SAEConn $ ERR $ INTERNAL $ show e)
+    Left e -> liftIO $ notifyEvent c ("", "", AEvt SAEConn $ ERR $ INTERNAL $ show e)
     Right () -> pure ()
 
 restoreServersStats :: AgentClient -> AM' ()
 restoreServersStats c@AgentClient {smpServersStats, xftpServersStats, ntfServersStats, srvStatsStartedAt} = do
   tryAllErrors' (withStore c getServersStats) >>= \case
-    Left e -> atomically $ writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ INTERNAL $ show e)
+    Left e -> liftIO $ notifyEvent c ("", "", AEvt SAEConn $ ERR $ INTERNAL $ show e)
     Right (startedAt, Nothing) -> atomically $ writeTVar srvStatsStartedAt startedAt
     Right (startedAt, Just AgentPersistedServerStats {smpServersStats = sss, xftpServersStats = xss, ntfServersStats = OptionalMap nss}) -> do
       atomically $ writeTVar srvStatsStartedAt startedAt
@@ -835,8 +839,8 @@ deleteUser' c@AgentClient {smpServersStats, xftpServersStats} userId delSMPQueue
   lift $ saveServersStats c
   where
     delUser =
-      whenM (withStore' c (`deleteUserWithoutConns` userId)) . atomically $
-        writeTBQueue (subQ c) ("", "", AEvt SAENone $ DEL_USER userId)
+      whenM (withStore' c (`deleteUserWithoutConns` userId)) . liftIO $
+        notifyEvent c ("", "", AEvt SAENone $ DEL_USER userId)
 
 setUserService' :: AgentClient -> UserId -> Bool -> AM ()
 setUserService' c userId enable = do
@@ -1330,7 +1334,7 @@ startJoinInvitation c userId connId sq_ enableNtfs cReqUri pqSup =
             getSndRatchet db connId v >>= \case
               Right r -> pure $ Right $ snd r
               Left e -> do
-                nonBlockingWriteTBQueue (subQ c) ("", connId, AEvt SAEConn (ERR $ INTERNAL $ "no snd ratchet " <> show e))
+                nonBlockingNotifyEvent c ("", connId, AEvt SAEConn (ERR $ INTERNAL $ "no snd ratchet " <> show e))
                 runExceptT $ createRatchet_ db g maxSupported pqSupport e2eRcvParams
           pure (cData, sq, e2eSndParams, Nothing)
         _ -> do
@@ -1424,7 +1428,7 @@ joinConnSrv c nm userId connId enableNtfs cReqUri@CRContactUri {} cInfo pqSup su
             getRatchetX3dhKeys db connId >>= \case
               Right keys -> pure $ CR.mkRcvE2ERatchetParams (maxVersion e2eVR) keys
               Left e -> do
-                nonBlockingWriteTBQueue (subQ c) ("", connId, AEvt SAEConn (ERR $ INTERNAL $ "no rcv ratchet " <> show e))
+                nonBlockingNotifyEvent c ("", connId, AEvt SAEConn (ERR $ INTERNAL $ "no rcv ratchet " <> show e))
                 let pqEnc = CR.initialPQEncryption False pqInitKeys
                 (pks, e2eRcvParams) <- liftIO $ CR.generateRcvE2EParams g (maxVersion e2eVR) pqEnc
                 createRatchetX3dhKeys db connId pks
@@ -1436,7 +1440,7 @@ joinConnSrv c nm userId connId enableNtfs cReqUri@CRContactUri {} cInfo pqSup su
 delInvSL :: AgentClient -> ConnId -> SMPServerWithAuth -> SMP.LinkId -> AM ()
 delInvSL c connId srv lnkId =
   withStore' c (\db -> deleteInvShortLink db (protoServer srv) lnkId) `catchE` \e ->
-    liftIO $ nonBlockingWriteTBQueue (subQ c) ("", connId, AEvt SAEConn (ERR $ INTERNAL $ "error deleting short link " <> show e))
+    liftIO $ nonBlockingNotifyEvent c ("", connId, AEvt SAEConn (ERR $ INTERNAL $ "error deleting short link " <> show e))
 
 joinConnSrvAsync :: AgentClient -> UserId -> ConnId -> Bool -> ConnectionRequestUri c -> ConnInfo -> PQSupport -> SubscriptionMode -> SMPServerWithAuth -> AM SndQueueSecured
 joinConnSrvAsync c userId connId enableNtfs inv@CRInvitationUri {} cInfo pqSupport subMode srv = do
@@ -1609,8 +1613,8 @@ subscribeConnections_ c conns = do
     notifyResultError rs = do
       let actual = M.size rs
           expected = length conns
-      when (actual /= expected) . atomically $
-        writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ INTERNAL $ "subscribeConnections result size: " <> show actual <> ", expected " <> show expected)
+      when (actual /= expected) . liftIO $
+        notifyEvent c ("", "", AEvt SAEConn $ ERR $ INTERNAL $ "subscribeConnections result size: " <> show actual <> ", expected " <> show expected)
 
 subscribeAllConnections' :: AgentClient -> Bool -> Maybe UserId -> AM ()
 subscribeAllConnections' c onlyNeeded activeUserId_ = handleErr $ do
@@ -1657,7 +1661,7 @@ subscribeAllConnections' c onlyNeeded activeUserId_ = handleErr $ do
                 Just SSErrorQueueCount {expectedQueueCount = n, subscribedQueueCount = n'} | n > 0 && n' == 0 -> unassocQueues
                 _ -> pure True
               Left e -> do
-                atomically $ writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR e)
+                liftIO $ notifyEvent c ("", "", AEvt SAEConn $ ERR e)
                 if clientServiceError e
                   then False <$ withStore' c (\db -> unassocUserServerRcvQueueSubs' db userId srv)
                   else pure True
@@ -1866,21 +1870,17 @@ getAsyncCmdWorker hasWork c connId server =
 data CommandCompletion = CCMoved | CCCompleted
 
 runCommandProcessing :: AgentClient -> ConnId -> Maybe SMPServer -> Worker -> AM ()
-runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
+runCommandProcessing c connId server_ Worker {doWork} = do
   ri <- asks $ messageRetryInterval . config -- different retry interval?
   forever $ do
-    atomically $ endAgentOperation c AOSndNetwork
+    endAgentOp c AOSndNetwork
     lift $ waitForWork doWork
     liftIO $ throwWhenInactive c
     atomically $ beginAgentOperation c AOSndNetwork
-    withWork c doWork (\db -> getPendingServerCommand db connId server_) $ runProcessCmd (riFast ri)
+    withWork c doWork (\db -> getPendingServerCommand db connId server_) $ processCmd (riFast ri)
   where
-    runProcessCmd ri cmd = do
-      pending <- newTVarIO []
-      processCmd ri cmd pending
-      mapM_ (atomically . writeTBQueue subQ) . reverse =<< readTVarIO pending
-    processCmd :: RetryInterval -> PendingCommand -> TVar [ATransmission] -> AM ()
-    processCmd ri PendingCommand {cmdId, corrId, userId, command} pendingCmds = case command of
+    processCmd :: RetryInterval -> PendingCommand -> AM ()
+    processCmd ri PendingCommand {cmdId, corrId, userId, command} = case command of
       AClientCommand cmd -> case cmd of
         NEW enableNtfs (ACM cMode) pqEnc subMode -> noServer $ do
           triedHosts <- newTVarIO S.empty
@@ -2034,9 +2034,7 @@ runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
         internalErr s = cmdError $ INTERNAL $ s <> ": " <> show (agentCommandTag command)
         cmdError e = notify (ERR e) >> withStore' c (`deleteCommand` cmdId)
         notify :: forall e. AEntityI e => AEvent e -> AM ()
-        notify cmd =
-          let t = (corrId, connId, AEvt (sAEntity @e) cmd)
-           in atomically $ ifM (isFullTBQueue subQ) (modifyTVar' pendingCmds (t :)) (writeTBQueue subQ t)
+        notify cmd = liftIO $ notifyEvent c (corrId, connId, AEvt (sAEntity @e) cmd)
 -- ^ ^ ^ async command processing /
 
 enqueueMessages :: AgentClient -> ConnData -> NonEmpty SndQueue -> MsgFlags -> AMessage -> AM (AgentMsgId, PQEncryption)
@@ -2169,17 +2167,17 @@ submitPendingMsg c sq = do
   void $ getDeliveryWorker True c sq
 
 runSmpQueueMsgDelivery :: AgentClient -> SndQueue -> (Worker, TMVar ()) -> AM ()
-runSmpQueueMsgDelivery c@AgentClient {subQ} sq@SndQueue {userId, connId, server, queueMode} (Worker {doWork}, qLock) = do
+runSmpQueueMsgDelivery c sq@SndQueue {userId, connId, server, queueMode} (Worker {doWork}, qLock) = do
   AgentConfig {messageRetryInterval = ri, messageTimeout, helloTimeout, quotaExceededTimeout} <- asks config
   forever $ do
-    atomically $ endAgentOperation c AOSndNetwork
+    endAgentOp c AOSndNetwork
     lift $ waitForWork doWork
     liftIO $ throwWhenInactive c
     liftIO $ throwWhenNoDelivery c sq
     atomically $ beginAgentOperation c AOSndNetwork
     withWork c doWork (\db -> getPendingQueueMsg db connId sq) $
       \(rq_, PendingMsgData {msgId, msgType, msgBody, pqEncryption, msgFlags, msgRetryState, internalTs, internalSndId, prevMsgHash, pendingMsgPrepData_}) -> do
-        atomically $ endAgentOperation c AOMsgDelivery -- this operation begins in submitPendingMsg
+        endAgentOp c AOMsgDelivery -- this operation begins in submitPendingMsg
         let mId = unId msgId
             ri' = maybe id updateRetryInterval2 msgRetryState ri
         withRetryLock2 ri' qLock $ \riState loop -> do
@@ -2337,7 +2335,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} sq@SndQueue {userId, connId, server,
     delMsgKeep :: Bool -> InternalId -> AM ()
     delMsgKeep keepForReceipt msgId = withStore' c $ \db -> deleteSndMsgDelivery db connId sq msgId keepForReceipt
     notify :: forall e. AEntityI e => AEvent e -> AM ()
-    notify cmd = atomically $ writeTBQueue subQ ("", connId, AEvt (sAEntity @e) cmd)
+    notify cmd = liftIO $ notifyEvent c ("", connId, AEvt (sAEntity @e) cmd)
     notifyDel :: AEntityI e => InternalId -> AEvent e -> AM ()
     notifyDel msgId cmd = notify cmd >> delMsg msgId
     connError msgId = notifyDel msgId . ERR . (`CONN` "")
@@ -2349,17 +2347,22 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} sq@SndQueue {userId, connId, server,
 retrySndOp :: AgentClient -> AM () -> AM ()
 retrySndOp c loop = do
   -- end... is in a separate atomically because if begin... blocks, SUSPENDED won't be sent
-  atomically $ endAgentOperation c AOSndNetwork
+  endAgentOp c AOSndNetwork
   liftIO $ throwWhenInactive c
   atomically $ beginAgentOperation c AOSndNetwork
   loop
 
+endAgentOp :: MonadIO m => AgentClient -> AgentOperation -> m ()
+endAgentOp c op = do
+  suspended <- atomically $ endAgentOperation c op
+  when suspended $ liftIO $ notifyEvent c ("", "", AEvt SAENone SUSPENDED)
+
 -- | Like 'withConnLock', but writes the returned 'ATransmission' to 'subQ'
 -- after releasing the lock, preventing deadlock with agentSubscriber.
 withConnLockNotify :: AgentClient -> ConnId -> Text -> AM (Maybe ATransmission) -> AM ()
 withConnLockNotify c connId name action = do
   t_ <- withConnLock c connId name action
-  forM_ t_ $ atomically . writeTBQueue (subQ c)
+  forM_ t_ $ liftIO . notifyEvent c
 
 ackMessage' :: AgentClient -> ConnId -> AgentMsgId -> Maybe MsgReceiptInfo -> AM ()
 ackMessage' c connId msgId rcptInfo_ = withConnLockNotify c connId "ackMessage" $ do
@@ -2576,7 +2579,7 @@ prepareDeleteConnections_ getConnections c waitDelivery connIds = do
     unsubNtfConnIds connIds' = do
       ns <- asks ntfSupervisor
       atomically $ writeTBQueue (ntfSubQ ns) (NSCDeleteSub, connIds')
-    notify = atomically . writeTBQueue (subQ c)
+    notify = liftIO . notifyEvent c
 
 deleteConnQueues :: AgentClient -> NetworkRequestMode -> Bool -> Bool -> [RcvQueue] -> AM' (Map ConnId (Either AgentErrorType ()))
 deleteConnQueues c nm waitDelivery ntf rqs = do
@@ -2610,7 +2613,7 @@ deleteConnQueues c nm waitDelivery ntf rqs = do
                 -- attempts and successes are counted in deleteQueues function
                 atomically $ incSMPServerStat c userId server connDeleted
                 pure ((rq, Right ()), Just (Just e))
-    notify = when ntf . atomically . writeTBQueue (subQ c)
+    notify = when ntf . liftIO . notifyEvent c
     connResults :: [(RcvQueue, Either AgentErrorType ())] -> Map ConnId (Either AgentErrorType ())
     connResults = M.map snd . foldl' addResult M.empty
       where
@@ -2646,8 +2649,8 @@ deleteConnections_ getConnections ntf waitDelivery c nm connIds = do
     notifyResultError rs = do
       let actual = M.size rs
           expected = length connIds
-      when (actual /= expected) . atomically $
-        writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ INTERNAL $ "deleteConnections result size: " <> show actual <> ", expected " <> show expected)
+      when (actual /= expected) . liftIO $
+        notifyEvent c ("", "", AEvt SAEConn $ ERR $ INTERNAL $ "deleteConnections result size: " <> show actual <> ", expected " <> show expected)
 
 getConnectionServers' :: AgentClient -> ConnId -> AM ConnectionStats
 getConnectionServers' c connId = do
@@ -2950,20 +2953,19 @@ suspendAgent c 0 = do
   where
     suspend opSel = atomically $ modifyTVar' (opSel c) $ \s -> s {opSuspended = True}
 suspendAgent c@AgentClient {agentState = as} maxDelay = do
-  state <-
-    atomically $ do
-      writeTVar as ASSuspending
-      suspendOperation c AONtfNetwork $ pure ()
-      suspendOperation c AORcvNetwork $
-        suspendOperation c AOMsgDelivery $
-          suspendSendingAndDatabase c
-      readTVar as
+  (state, suspended) <- atomically $ do
+    writeTVar as ASSuspending
+    void $ suspendOperation c AONtfNetwork $ pure False
+    suspended <- suspendOperation c AORcvNetwork $
+      suspendOperation c AOMsgDelivery $
+        suspendSendingAndDatabase c
+    (,suspended) <$> readTVar as
+  when suspended $ notifyEvent c ("", "", AEvt SAENone SUSPENDED)
   when (state == ASSuspending) . void . forkIO $ do
     threadDelay maxDelay
-    -- liftIO $ putStrLn "suspendAgent after timeout"
-    atomically . whenSuspending c $ do
-      -- unsafeIOToSTM $ putStrLn $ "in timeout: suspendSendingAndDatabase"
+    suspended' <- atomically . whenSuspendingB c $
       suspendSendingAndDatabase c
+    when suspended' $ notifyEvent c ("", "", AEvt SAENone SUSPENDED)
 
 execAgentStoreSQL :: AgentClient -> Text -> AE [Text]
 execAgentStoreSQL c sql = withAgentEnv c $ withStore' c (`execSQL` sql)
@@ -2988,17 +2990,17 @@ getNextSMPServer :: AgentClient -> UserId -> [SMPServer] -> AM SMPServerWithAuth
 getNextSMPServer c userId = getNextServer c userId storageSrvs
 {-# INLINE getNextSMPServer #-}
 
-subscriber :: AgentClient -> AM' ()
-subscriber c@AgentClient {msgQ, subQ} = run $ forever $ do
-  t <- atomically $ readTBQueue msgQ
+subscriber :: AgentClient -> ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> AM' ()
+subscriber c t = run $
   agentOperationBracket c AORcvNetwork waitUntilActive $
     processSMPTransmissions c t
   where
-    run a = a `catchOwn` \e -> notify $ CRITICAL True $ "Agent subscriber stopped: " <> show e
-    notify err = atomically $ writeTBQueue subQ ("", "", AEvt SAEConn $ ERR err)
+    run a = a `catchOwn` \e -> notify $ CRITICAL True $ "subscriber error: " <> show e
+    notify err = liftIO $ notifyEvent c ("", "", AEvt SAEConn $ ERR err)
+
 
 cleanupManager :: AgentClient -> AM' ()
-cleanupManager c@AgentClient {subQ} = do
+cleanupManager c = do
   AgentConfig {initialCleanupDelay, cleanupInterval = int, storedMsgDataTTL = ttl, cleanupBatchSize = limit} <-
     asks config
   liftIO $ threadDelay' initialCleanupDelay
@@ -3066,7 +3068,7 @@ cleanupManager c@AgentClient {subQ} = do
       rcvFilesTTL <- asks $ rcvFilesTTL . config
       withStore' c (`deleteDeletedSndChunkReplicasExpired` rcvFilesTTL)
     notify :: forall e. AEntityI e => AEntityId -> AEvent e -> AM ()
-    notify entId cmd = atomically $ writeTBQueue subQ ("", entId, AEvt (sAEntity @e) cmd)
+    notify entId cmd = liftIO $ notifyEvent c ("", entId, AEvt (sAEntity @e) cmd)
 
 data ACKd = ACKd | ACKPending
 
@@ -3074,10 +3076,10 @@ data ACKd = ACKd | ACKPending
 -- It cannot be finally, as sometimes it needs to be ACK+DEL,
 -- and sometimes ACK has to be sent from the consumer.
 processSMPTransmissions :: AgentClient -> ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> AM' ()
-processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandleParams {thAuth, sessionId = sessId}, ts) = do
+processSMPTransmissions c (tSess@(userId, srv, _), THandleParams {thAuth, sessionId = sessId}, ts) = do
   upConnIds <- newTVarIO []
   serviceRQs <- newTVarIO ([] :: [RcvQueue])
-  forM_ ts $ \(entId, t) -> case t of
+  forM_ ts $ \(entId, t) -> E.uninterruptibleMask_ $ case t of
     STEvent msgOrErr
       | entId == SMP.NoEntity -> case msgOrErr of
           Right msg -> case msg of
@@ -3086,7 +3088,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
             _ -> logError $ "unexpected event: " <> tshow msg
           Left e -> notifyErr "" e
       | otherwise -> withRcvConn entId $ \rq@RcvQueue {connId} conn -> case msgOrErr of
-          Right msg -> runProcessSMP rq conn (toConnData conn) msg
+          Right msg -> processSMP rq conn (toConnData conn) msg
           Left e -> lift $ do
             processClientNotice rq e
             notifyErr connId e
@@ -3097,11 +3099,11 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
           Right (SMP.SOK serviceId_) -> liftIO $ processSubOk rq upConnIds serviceRQs serviceId_
           Right msg@SMP.MSG {} -> do
             liftIO $ processSubOk rq upConnIds serviceRQs Nothing -- the connection is UP even when processing this particular message fails
-            runProcessSMP rq conn (toConnData conn) msg
+            processSMP rq conn (toConnData conn) msg
           Right r -> lift $ processSubErr rq $ unexpectedResponse r
           Left e -> lift $ unless (temporaryClientError e) $ processSubErr rq e -- timeout/network was already reported
         SMP.ACK _ -> case respOrErr of
-          Right msg@SMP.MSG {} -> runProcessSMP rq conn (toConnData conn) msg
+          Right msg@SMP.MSG {} -> processSMP rq conn (toConnData conn) msg
           _ -> pure () -- TODO process OK response to ACK
         _ -> pure () -- TODO process expired response to DEL
     STResponse {} -> pure () -- TODO process expired responses to sent messages
@@ -3147,21 +3149,15 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
           (atomically $ putTMVar (clientNoticesLock c) ())
           (processClientNotices c tSess [(rcvQueueSub rq, notice_)])
     notify' :: forall e m. (AEntityI e, MonadIO m) => ConnId -> AEvent e -> m ()
-    notify' connId msg = atomically $ writeTBQueue subQ ("", connId, AEvt (sAEntity @e) msg)
+    notify' connId msg = liftIO $ notifyEvent c ("", connId, AEvt (sAEntity @e) msg)
     notifyErr :: ConnId -> SMPClientError -> AM' ()
     notifyErr connId = notify' connId . ERR . protocolClientError SMP (B.unpack $ strEncode srv)
-    runProcessSMP :: RcvQueue -> Connection c -> ConnData -> BrokerMsg -> AM ()
-    runProcessSMP rq conn cData msg = do
-      pending <- newTVarIO []
-      processSMP rq conn cData msg pending
-      mapM_ (atomically . writeTBQueue subQ) . reverse =<< readTVarIO pending
-    processSMP :: forall c. RcvQueue -> Connection c -> ConnData -> BrokerMsg -> TVar [ATransmission] -> AM ()
+    processSMP :: forall c. RcvQueue -> Connection c -> ConnData -> BrokerMsg -> AM ()
     processSMP
       rq@RcvQueue {rcvId = rId, queueMode, e2ePrivKey, e2eDhSecret, status, smpClientVersion = agreedClientVerion}
       conn
       cData@ConnData {connId, connAgentVersion = agreedAgentVersion, ratchetSyncState = rss}
-      smpMsg
-      pendingMsgs =
+      smpMsg =
         withConnLock c connId "processSMP" $ case smpMsg of
           SMP.MSG msg@SMP.RcvMessage {msgId = srvMsgId} -> do
             atomically $ incSMPServerStat c userId srv recvMsgs
@@ -3361,9 +3357,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
           notify :: forall e m. (AEntityI e, MonadIO m) => AEvent e -> m ()
           notify = notify_ connId
           notify_ :: forall e m. (AEntityI e, MonadIO m) => ConnId -> AEvent e -> m ()
-          notify_ connId' msg =
-            let t = ("", connId', AEvt (sAEntity @e) msg)
-             in atomically $ ifM (isFullTBQueue subQ) (modifyTVar' pendingMsgs (t :)) (writeTBQueue subQ t)
+          notify_ connId' msg = liftIO $ notifyEvent c ("", connId', AEvt (sAEntity @e) msg)
 
           prohibited :: Text -> AM ()
           prohibited s = do

src/Simplex/Messaging/Agent/Client.hs

@@ -159,6 +159,7 @@ module Simplex.Messaging.Agent.Client
     suspendOperation,
     notifySuspended,
     whenSuspending,
+    whenSuspendingB,
     withStore,
     withStore',
     withStoreBatch,
@@ -167,6 +168,8 @@ module Simplex.Messaging.Agent.Client
     storeError,
     notifySub,
     notifySub',
+    notifyEvent,
+    nonBlockingNotifyEvent,
     userServers,
     pickServer,
     getNextServer,
@@ -317,7 +320,7 @@ import System.Mem.Weak (Weak, deRefWeak)
 import System.Random (randomR)
 import UnliftIO (mapConcurrently, timeout)
 import UnliftIO.Async (async)
-import UnliftIO.Concurrent (forkIO, mkWeakThreadId)
+import UnliftIO.Concurrent (forkIO, forkIOWithUnmask, mkWeakThreadId)
 import UnliftIO.Directory (doesFileExist, getTemporaryDirectory, removeFile)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
@@ -340,8 +343,8 @@ type XFTPTransportSession = TransportSession FileResponse
 data AgentClient = AgentClient
   { acThread :: TVar (Maybe (Weak ThreadId)),
     active :: TVar Bool,
-    subQ :: TBQueue ATransmission,
-    msgQ :: TBQueue (ServerTransmissionBatch SMPVersion ErrorType BrokerMsg),
+    processEvent :: Bool -> ATransmission -> IO (),
+    processServerMsg :: AgentClient -> ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> IO (),
     smpServers :: TMap UserId (UserServers 'PSMP),
     smpClients :: TMap SMPTransportSession SMPClientVar,
     useClientServices :: TMap UserId Bool,
@@ -422,7 +425,8 @@ getAgentWorker' toW fromW name hasWork c@AgentClient {agentEnv} key ws work = do
           t <- liftIO getSystemTime
           let maxRestarts = maxWorkerRestartsPerMin $ config agentEnv
           -- worker may terminate because it was deleted from the map (getWorker returns Nothing), then it won't restart
-          restart <- atomically $ getWorker >>= maybe (pure False) (shouldRestart e_ (toW w) t maxRestarts)
+          (restart, notify_) <- atomically $ getWorker >>= maybe (pure (False, Nothing)) (shouldRestart e_ (toW w) t maxRestarts)
+          forM_ notify_ $ liftIO . notifyEvent c
           when restart runWork
         shouldRestart e_ Worker {workerId = wId, doWork, action, restarts} t maxRestarts w'
           | wId == workerId (toW w') = do
@@ -430,24 +434,21 @@ getAgentWorker' toW fromW name hasWork c@AgentClient {agentEnv} key ws work = do
               isActive <- readTVar $ active c
               checkRestarts isActive $ updateRestartCount t rc
           | otherwise =
-              pure False -- there is a new worker in the map, no action
+              pure (False, Nothing) -- there is a new worker in the map, no action
           where
             checkRestarts isActive rc
               | isActive && restartCount rc < maxRestarts = do
                   writeTVar restarts rc
                   hasWorkToDo' doWork
                   void $ tryPutTMVar action Nothing
-                  notifyErr INTERNAL
-                  pure True
+                  pure (True, Just $ notifyMsg rc INTERNAL)
               | otherwise = do
                   TM.delete key ws
-                  when isActive $ notifyErr $ CRITICAL True
-                  pure False
-              where
-                notifyErr err = do
-                  let e = either ((", error: " <>) . show) (\_ -> ", no error") e_
-                      msg = "Worker " <> name <> " for " <> show key <> " terminated " <> show (restartCount rc) <> " times" <> e
-                  writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ err msg)
+                  pure (False, if isActive then Just (notifyMsg rc $ CRITICAL True) else Nothing)
+            notifyMsg rc err =
+              let e = either ((", error: " <>) . show) (\_ -> ", no error") e_
+                  msg = "Worker " <> name <> " for " <> show key <> " terminated " <> show (restartCount rc) <> " times" <> e
+               in ("", "", AEvt SAEConn $ ERR $ err msg)
 
 newWorker :: AgentClient -> STM Worker
 newWorker c = do
@@ -464,7 +465,8 @@ runWorkerAsync Worker {action} work =
     (atomically . tryPutTMVar action) -- if it was running (or if start crashes), put it back and unlock (don't lock if it was just started)
     (\a -> when (isNothing a) start) -- start worker if it's not running
   where
-    start = atomically . putTMVar action . Just =<< mkWeakThreadId =<< forkIO work
+    -- unmask so the worker doesn't inherit an uninterruptibleMask from the enqueueing thread (else killThread hangs on teardown)
+    start = atomically . putTMVar action . Just =<< mkWeakThreadId =<< forkIOWithUnmask ($ work)
 
 data AgentOperation = AONtfNetwork | AORcvNetwork | AOMsgDelivery | AOSndNetwork | AODatabase
   deriving (Eq, Show)
@@ -508,15 +510,13 @@ data UserNetworkType = UNNone | UNCellular | UNWifi | UNEthernet | UNOther
   deriving (Eq, Show)
 
 -- | Creates an SMP agent client instance that receives commands and sends responses via 'TBQueue's.
-newAgentClient :: Int -> InitialAgentServers -> UTCTime -> Map (Maybe SMPServer) (Maybe SystemSeconds) -> Env -> IO AgentClient
-newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg, useServices, presetDomains, presetServers} currentTs notices agentEnv = do
+newAgentClient :: Int -> InitialAgentServers -> UTCTime -> Map (Maybe SMPServer) (Maybe SystemSeconds) -> (Bool -> ATransmission -> IO ()) -> (AgentClient -> ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> IO ()) -> Env -> IO AgentClient
+newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg, useServices, presetDomains, presetServers} currentTs notices processEvent processServerMsg agentEnv = do
   let cfg = config agentEnv
       qSize = tbqSize cfg
   proxySessTs <- newTVarIO =<< getCurrentTime
   acThread <- newTVarIO Nothing
   active <- newTVarIO True
-  subQ <- newTBQueueIO qSize
-  msgQ <- newTBQueueIO qSize
   smpServers <- newTVarIO $ M.map mkUserServers smp
   smpClients <- TM.emptyIO
   useClientServices <- newTVarIO useServices
@@ -555,8 +555,8 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg, useServices
     AgentClient
       { acThread,
         active,
-        subQ,
-        msgQ,
+        processEvent,
+        processServerMsg,
         smpServers,
         smpClients,
         useClientServices,
@@ -731,7 +731,7 @@ getSMPProxyClient c@AgentClient {active, smpClients, smpProxiedRelays, workerSeq
         Nothing -> Left $ BROKER (B.unpack $ strEncode srv) TIMEOUT
 
 smpConnectClient :: AgentClient -> NetworkRequestMode -> SMPTransportSession -> TMap SMPServer ProxiedRelayVar -> SMPClientVar -> AM SMPConnectedClient
-smpConnectClient c@AgentClient {smpClients, msgQ, proxySessTs, presetDomains} nm tSess@(userId, srv, _) prs v =
+smpConnectClient c@AgentClient {processServerMsg, smpClients, proxySessTs, presetDomains} nm tSess@(userId, srv, _) prs v =
   newProtocolClient c tSess smpClients connectClient v
     `catchAllErrors` \e -> lift (resubscribeSMPSession c tSess) >> throwE e
   where
@@ -744,7 +744,7 @@ smpConnectClient c@AgentClient {smpClients, msgQ, proxySessTs, presetDomains} nm
       env <- ask
       smp <- liftError (protocolClientError SMP $ B.unpack $ strEncode srv) $ do
         ts <- readTVarIO proxySessTs
-        ExceptT $ getProtocolClient g nm tSess cfg' presetDomains (Just msgQ) ts $ smpClientDisconnected c tSess env v' prs
+        ExceptT $ getProtocolClient g nm tSess cfg' presetDomains (Just $ processServerMsg c) ts $ smpClientDisconnected c tSess env v' prs
       atomically $ SS.setSessionId tSess (sessionId $ thParams smp) $ currentSubs c
       updateClientService service smp
       pure SMPConnectedClient {connectedClient = smp, proxiedRelays = prs}
@@ -833,7 +833,7 @@ resubscribeSMPSession c@AgentClient {smpSubWorkers, workerSeq} tSess = do
     handleNotify = E.handleAny $ notifySub' c "" . ERR . INTERNAL . show
 
 notifySub' :: forall e m. (AEntityI e, MonadIO m) => AgentClient -> ConnId -> AEvent e -> m ()
-notifySub' c connId cmd = liftIO $ nonBlockingWriteTBQueue (subQ c) (B.empty, connId, AEvt (sAEntity @e) cmd)
+notifySub' c connId cmd = liftIO $ nonBlockingNotifyEvent c (B.empty, connId, AEvt (sAEntity @e) cmd)
 {-# INLINE notifySub' #-}
 
 notifySub :: MonadIO m => AgentClient -> AEvent 'AENone -> m ()
@@ -858,7 +858,7 @@ getNtfServerClient c@AgentClient {active, ntfClients, workerSeq, proxySessTs, pr
     clientDisconnected :: NtfClientVar -> NtfClient -> IO ()
     clientDisconnected v client = do
       atomically $ removeSessVar v tSess ntfClients
-      atomically $ writeTBQueue (subQ c) ("", "", AEvt SAENone $ hostEvent DISCONNECT client)
+      notifyEvent c ("", "", AEvt SAENone $ hostEvent DISCONNECT client)
       logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv
 
 getXFTPServerClient :: AgentClient -> XFTPTransportSession -> AM XFTPClient
@@ -879,7 +879,7 @@ getXFTPServerClient c@AgentClient {active, xftpClients, workerSeq, proxySessTs,
     clientDisconnected :: XFTPClientVar -> XFTPClient -> IO ()
     clientDisconnected v client = do
       atomically $ removeSessVar v tSess xftpClients
-      atomically $ writeTBQueue (subQ c) ("", "", AEvt SAENone $ hostEvent DISCONNECT client)
+      notifyEvent c ("", "", AEvt SAENone $ hostEvent DISCONNECT client)
       logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv
 
 waitForProtocolClient ::
@@ -918,7 +918,7 @@ newProtocolClient c tSess@(userId, srv, entityId_) clients connectClient v =
     Right client -> do
       logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv <> " (user " <> bshow userId <> maybe "" (" for entity " <>) entityId_ <> ")"
       atomically $ putTMVar (sessionVar v) (Right client)
-      liftIO $ nonBlockingWriteTBQueue (subQ c) ("", "", AEvt SAENone $ hostEvent CONNECT client)
+      liftIO $ nonBlockingNotifyEvent c ("", "", AEvt SAENone $ hostEvent CONNECT client)
       pure client
     Left e -> do
       ei <- asks $ persistErrorInterval . config
@@ -1046,6 +1046,12 @@ withConnLock' _ "" _ = id
 withConnLock' AgentClient {connLocks} connId name = withLockMap connLocks connId name
 {-# INLINE withConnLock' #-}
 
+notifyEvent :: AgentClient -> ATransmission -> IO ()
+notifyEvent AgentClient {processEvent} = processEvent True
+
+nonBlockingNotifyEvent :: AgentClient -> ATransmission -> IO ()
+nonBlockingNotifyEvent AgentClient {processEvent} = processEvent False
+
 withInvLock :: AgentClient -> ByteString -> Text -> AM a -> AM a
 withInvLock c key name = ExceptT . withInvLock' c key name . runExceptT
 {-# INLINE withInvLock #-}
@@ -1732,7 +1738,7 @@ resubscribeClientService c tSess@(userId, srv, _) serviceSub =
         r <$ withStore' c (\db -> removeRcvServiceAssocs db userId srv)
       _ -> pure r
     Left e -> do
-      atomically $ writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR e)
+      liftIO $ notifyEvent c ("", "", AEvt SAEConn $ ERR e)
       when (clientServiceError e) $ do
         atomically $ SS.deleteServiceSub tSess $ currentSubs c
         unassocSubscribeQueues
@@ -2280,7 +2286,7 @@ withWork_ c doWork getWork action =
     noWork = liftIO $ noWorkToDo doWork
     notifyErr err e = do
       logError $ "withWork_ error: " <> tshow e
-      atomically $ writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ err $ show e)
+      liftIO $ notifyEvent c ("", "", AEvt SAEConn $ ERR $ err $ show e)
 
 withWorkItems :: (AnyStoreError e', MonadIO m) => AgentClient -> TMVar () -> ExceptT e m (Either e' [Either e' a]) -> (NonEmpty a -> ExceptT e m ()) -> ExceptT e m ()
 withWorkItems c doWork getWork action = do
@@ -2305,7 +2311,7 @@ withWorkItems c doWork getWork action = do
     noWork = liftIO $ noWorkToDo doWork
     notifyErr err e = do
       logError $ "withWorkItems error: " <> tshow e
-      atomically $ writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ err $ show e)
+      liftIO $ notifyEvent c ("", "", AEvt SAEConn $ ERR $ err $ show e)
 
 noWorkToDo :: TMVar () -> IO ()
 noWorkToDo = void . atomically . tryTakeTMVar
@@ -2319,9 +2325,9 @@ hasWorkToDo' :: TMVar () -> STM ()
 hasWorkToDo' = void . (`tryPutTMVar` ())
 {-# INLINE hasWorkToDo' #-}
 
-endAgentOperation :: AgentClient -> AgentOperation -> STM ()
+endAgentOperation :: AgentClient -> AgentOperation -> STM Bool
 endAgentOperation c op = endOperation c op $ case op of
-  AONtfNetwork -> pure ()
+  AONtfNetwork -> pure False
   AORcvNetwork ->
     suspendOperation c AOMsgDelivery $
       suspendSendingAndDatabase c
@@ -2333,36 +2339,37 @@ endAgentOperation c op = endOperation c op $ case op of
   AODatabase ->
     notifySuspended c
 
-suspendSendingAndDatabase :: AgentClient -> STM ()
+suspendSendingAndDatabase :: AgentClient -> STM Bool
 suspendSendingAndDatabase c =
   suspendOperation c AOSndNetwork $
     suspendOperation c AODatabase $
       notifySuspended c
 
-suspendOperation :: AgentClient -> AgentOperation -> STM () -> STM ()
+suspendOperation :: AgentClient -> AgentOperation -> STM Bool -> STM Bool
 suspendOperation c op endedAction = do
   n <- stateTVar (agentOpSel op c) $ \s -> (opsInProgress s, s {opSuspended = True})
-  -- unsafeIOToSTM $ putStrLn $ "suspendOperation_ " <> show op <> " " <> show n
-  when (n == 0) $ whenSuspending c endedAction
+  if n == 0 then whenSuspendingB c endedAction else pure False
 
-notifySuspended :: AgentClient -> STM ()
+notifySuspended :: AgentClient -> STM Bool
 notifySuspended c = do
-  -- unsafeIOToSTM $ putStrLn "notifySuspended"
-  writeTBQueue (subQ c) ("", "", AEvt SAENone SUSPENDED)
   writeTVar (agentState c) ASSuspended
+  pure True
 
-endOperation :: AgentClient -> AgentOperation -> STM () -> STM ()
+endOperation :: AgentClient -> AgentOperation -> STM Bool -> STM Bool
 endOperation c op endedAction = do
   (suspended, n) <- stateTVar (agentOpSel op c) $ \s ->
     let n = max 0 (opsInProgress s - 1)
      in ((opSuspended s, n), s {opsInProgress = n})
-  -- unsafeIOToSTM $ putStrLn $ "endOperation: " <> show op <> " " <> show suspended <> " " <> show n
-  when (suspended && n == 0) $ whenSuspending c endedAction
+  if suspended && n == 0 then whenSuspendingB c endedAction else pure False
 
 whenSuspending :: AgentClient -> STM () -> STM ()
 whenSuspending c = whenM ((== ASSuspending) <$> readTVar (agentState c))
 {-# INLINE whenSuspending #-}
 
+whenSuspendingB :: AgentClient -> STM Bool -> STM Bool
+whenSuspendingB c action =
+  ifM ((== ASSuspending) <$> readTVar (agentState c)) action (pure False)
+
 beginAgentOperation :: AgentClient -> AgentOperation -> STM ()
 beginAgentOperation c op = do
   let opVar = agentOpSel op c
@@ -2376,7 +2383,9 @@ agentOperationBracket :: MonadUnliftIO m => AgentClient -> AgentOperation -> (Ag
 agentOperationBracket c op check action =
   E.bracket
     (liftIO (check c) >> atomically (beginAgentOperation c op))
-    (\_ -> atomically $ endAgentOperation c op)
+    (\_ -> do
+      suspended <- atomically $ endAgentOperation c op
+      when suspended $ liftIO $ notifyEvent c ("", "", AEvt SAENone SUSPENDED))
     (const action)
 
 waitUntilForeground :: AgentClient -> IO ()
@@ -2849,9 +2858,9 @@ data ClientInfo
   deriving (Show)
 
 getAgentQueuesInfo :: AgentClient -> IO AgentQueuesInfo
-getAgentQueuesInfo AgentClient {msgQ, subQ, smpClients} = do
-  msgQInfo <- atomically $ getTBQueueInfo msgQ
-  subQInfo <- atomically $ getTBQueueInfo subQ
+getAgentQueuesInfo AgentClient {smpClients} = do
+  let msgQInfo = TBQueueInfo {qLength = 0, qFull = False}
+      subQInfo = TBQueueInfo {qLength = 0, qFull = False}
   smpClientsMap <- readTVarIO smpClients
   let smpClientsMap' = M.mapKeys (decodeLatin1 . strEncode) smpClientsMap
   smpClientsQueues <- mapM getClientQueuesInfo smpClientsMap'

src/Simplex/Messaging/Agent/NtfSubSupervisor.hs

@@ -502,9 +502,9 @@ workerInternalError c connId internalErrStr = do
 
 -- TODO change error
 notifyInternalError :: MonadIO m => AgentClient -> ConnId -> String -> m ()
-notifyInternalError AgentClient {subQ} connId internalErrStr = do
+notifyInternalError c connId internalErrStr = do
   logError $ T.pack internalErrStr
-  liftIO $ nonBlockingWriteTBQueue subQ ("", connId, AEvt SAEConn $ ERR $ INTERNAL internalErrStr)
+  liftIO $ nonBlockingNotifyEvent c ("", connId, AEvt SAEConn $ ERR $ INTERNAL internalErrStr)
 
 notifyInternalError' :: MonadIO m => AgentClient -> String -> m ()
 notifyInternalError' c = notifyInternalError c ""

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

@@ -50,6 +50,7 @@ import Data.Bits (xor)
 import Data.ByteArray (ScrubbedBytes)
 import qualified Data.ByteArray as BA
 import Data.ByteString (ByteString)
+import qualified Data.ByteString as B
 import Data.Functor (($>))
 import Data.IORef
 import Data.Maybe (fromMaybe)

src/Simplex/Messaging/Client.hs

@@ -565,10 +565,10 @@ type SMPTransportSession = TransportSession BrokerMsg
 -- | Connects to 'ProtocolServer' using passed client configuration
 -- and queue for messages and notifications.
 --
--- A single queue can be used for multiple 'SMPClient' instances,
+-- A single callback can be used for multiple 'SMPClient' instances,
 -- as 'SMPServerTransmission' includes server information.
-getProtocolClient :: forall v err msg. Protocol v err msg => TVar ChaChaDRG -> NetworkRequestMode -> TransportSession msg -> ProtocolClientConfig v -> [HostName] -> Maybe (TBQueue (ServerTransmissionBatch v err msg)) -> UTCTime -> (ProtocolClient v err msg -> IO ()) -> IO (Either (ProtocolClientError err) (ProtocolClient v err msg))
-getProtocolClient g nm transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize, networkConfig, clientALPN, serviceCredentials, serverVRange, agreeSecret, proxyServer, useSNI} presetDomains msgQ proxySessTs disconnected = do
+getProtocolClient :: forall v err msg. Protocol v err msg => TVar ChaChaDRG -> NetworkRequestMode -> TransportSession msg -> ProtocolClientConfig v -> [HostName] -> Maybe (ServerTransmissionBatch v err msg -> IO ()) -> UTCTime -> (ProtocolClient v err msg -> IO ()) -> IO (Either (ProtocolClientError err) (ProtocolClient v err msg))
+getProtocolClient g nm transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize, networkConfig, clientALPN, serviceCredentials, serverVRange, agreeSecret, proxyServer, useSNI} presetDomains processServerMsg proxySessTs disconnected = do
   case chooseTransportHost networkConfig (host srv) of
     Right useHost ->
       (getCurrentTime >>= mkProtocolClient useHost >>= runClient useTransport useHost)
@@ -586,6 +586,7 @@ getProtocolClient g nm transportSession@(_, srv, _) cfg@ProtocolClientConfig {qS
       sentCommands <- TM.emptyIO
       sndQ <- newTBQueueIO qSize
       rcvQ <- newTBQueueIO qSize
+      msgQ <- mapM (const $ newTBQueueIO qSize) processServerMsg
       return
         PClient
           { connected,
@@ -644,7 +645,7 @@ getProtocolClient g nm transportSession@(_, srv, _) cfg@ProtocolClientConfig {qS
           atomically $ do
             writeTVar (connected c) True
             putTMVar cVar $ Right c'
-          raceAny_ ([send c' th, process c', receive c' th] <> [monitor c' | smpPingInterval > 0])
+          raceAny_ ([send c' th, process c', receive c' th] <> readMsgs c' <> [monitor c' | smpPingInterval > 0])
             `E.finally` disconnected c'
 
     send :: Transport c => ProtocolClient v err msg -> THandle v c 'TClient -> IO ()
@@ -683,13 +684,18 @@ getProtocolClient g nm transportSession@(_, srv, _) cfg@ProtocolClientConfig {qS
         recoverWindow = 15 * 60 -- seconds
         maxCnt = smpPingCount networkConfig
 
+    readMsgs :: ProtocolClient v err msg -> [IO ()]
+    readMsgs c = case (processServerMsg, msgQ $ client_ c) of
+      (Just cb, Just q) -> [forever $ atomically (readTBQueue q) >>= cb]
+      _ -> []
+
     process :: ProtocolClient v err msg -> IO ()
     process c = forever $ atomically (readTBQueue $ rcvQ $ client_ c) >>= processMsgs c
 
     processMsgs :: ProtocolClient v err msg -> NonEmpty (Transmission (Either err msg)) -> IO ()
     processMsgs c ts = do
       ts' <- catMaybes <$> mapM (processMsg c) (L.toList ts)
-      forM_ msgQ $ \q ->
+      forM_ (msgQ $ client_ c) $ \q ->
         mapM_ (atomically . writeTBQueue q . serverTransmission c) (L.nonEmpty ts')
 
     processMsg :: ProtocolClient v err msg -> Transmission (Either err msg) -> IO (Maybe (EntityId, ServerTransmission err msg))
@@ -717,7 +723,7 @@ getProtocolClient g nm transportSession@(_, srv, _) cfg@ProtocolClientConfig {qS
             Just e -> Left $ PCEProtocolError e
             _ -> Right r
         sendMsg :: ServerTransmission err msg -> IO (Maybe (EntityId, ServerTransmission err msg))
-        sendMsg t = case msgQ of
+        sendMsg t = case processServerMsg of
           Just _ -> pure $ Just (entId, t)
           Nothing ->
             Nothing <$ case clientResp of

src/Simplex/Messaging/Client/Agent.hs

@@ -107,7 +107,7 @@ data SMPClientAgentConfig = SMPClientAgentConfig
   { smpCfg :: ProtocolClientConfig SMPVersion,
     reconnectInterval :: RetryInterval,
     persistErrorInterval :: NominalDiffTime,
-    msgQSize :: Maybe Natural,
+    msgQSize :: Natural,
     agentQSize :: Natural,
     agentSubsBatchSize :: Int,
     ownServerDomains :: [ByteString]
@@ -124,7 +124,7 @@ defaultSMPClientAgentConfig =
             maxInterval = 10 * second
           },
       persistErrorInterval = 30, -- seconds
-      msgQSize = Just 2048,
+      msgQSize = 2048,
       agentQSize = 2048,
       agentSubsBatchSize = 1360,
       ownServerDomains = []
@@ -138,7 +138,7 @@ data SMPClientAgent p = SMPClientAgent
     dbService :: Maybe DBService,
     active :: TVar Bool,
     startedAt :: UTCTime,
-    msgQ :: Maybe (TBQueue (ServerTransmissionBatch SMPVersion ErrorType BrokerMsg)),
+    processMsg :: SMPClientAgent p -> ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> IO (),
     agentQ :: TBQueue SMPClientAgentEvent,
     randomDrg :: TVar ChaChaDRG,
     smpClients :: TMap SMPServer SMPClientVar,
@@ -158,12 +158,10 @@ data SMPClientAgent p = SMPClientAgent
 
 type OwnServer = Bool
 
-newSMPClientAgent :: SParty p -> SMPClientAgentConfig -> Maybe DBService -> TVar ChaChaDRG -> IO (SMPClientAgent p)
-newSMPClientAgent agentParty agentCfg@SMPClientAgentConfig {msgQSize, agentQSize} dbService randomDrg = do
+newSMPClientAgent :: SParty p -> SMPClientAgentConfig -> (SMPClientAgent p -> ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> IO ()) -> Maybe DBService -> TVar ChaChaDRG -> IO (SMPClientAgent p)
+newSMPClientAgent agentParty agentCfg@SMPClientAgentConfig {agentQSize} processMsg dbService randomDrg = do
   active <- newTVarIO True
   startedAt <- getCurrentTime
-  -- Only subscribing agents receive server transmissions, should not be created until processed to prevent deadlock.
-  msgQ <- mapM newTBQueueIO msgQSize
   agentQ <- newTBQueueIO agentQSize
   smpClients <- TM.emptyIO
   smpSessions <- TM.emptyIO
@@ -180,7 +178,7 @@ newSMPClientAgent agentParty agentCfg@SMPClientAgentConfig {msgQSize, agentQSize
         dbService,
         active,
         startedAt,
-        msgQ,
+        processMsg,
         agentQ,
         randomDrg,
         smpClients,
@@ -255,7 +253,7 @@ isOwnServer SMPClientAgent {agentCfg} ProtocolServer {host} =
 
 -- | Run an SMP client for SMPClientVar
 connectClient :: SMPClientAgent p -> SMPServer -> SMPClientVar -> IO (Either SMPClientError SMPClient)
-connectClient ca@SMPClientAgent {agentCfg, dbService, smpClients, smpSessions, msgQ, randomDrg, startedAt} srv v = case dbService of
+connectClient ca@SMPClientAgent {agentCfg, dbService, smpClients, smpSessions, processMsg, randomDrg, startedAt} srv v =  case dbService of
   Just dbs -> runExceptT $ do
     creds <- ExceptT $ getCredentials dbs srv
     smp <- ExceptT $ getClient cfg {serviceCredentials = Just creds}
@@ -265,7 +263,7 @@ connectClient ca@SMPClientAgent {agentCfg, dbService, smpClients, smpSessions, m
   Nothing -> getClient cfg
   where
     cfg = smpCfg agentCfg
-    getClient cfg' = getProtocolClient randomDrg NRMBackground (1, srv, Nothing) cfg' [] msgQ startedAt clientDisconnected
+    getClient cfg' = getProtocolClient randomDrg NRMBackground (1, srv, Nothing) cfg' [] (Just $ processMsg ca) startedAt clientDisconnected
 
     clientDisconnected :: SMPClient -> IO ()
     clientDisconnected smp = do

src/Simplex/Messaging/Notifications/Server.hs

@@ -34,7 +34,6 @@ import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Either (partitionEithers)
 import Data.Functor (($>))
-import Data.Hashable (hash)
 import Data.IORef
 import Data.Int (Int64)
 import qualified Data.IntSet as IS
@@ -55,7 +54,7 @@ import GHC.IORef (atomicSwapIORef)
 import GHC.Stats (getRTSStats)
 import Network.Socket (ServiceName, Socket, socketToHandle)
 import Numeric.Natural (Natural)
-import Simplex.Messaging.Client (ProtocolClientError (..), SMPClientError, ServerTransmission (..))
+import Simplex.Messaging.Client (ProtocolClientError (..), SMPClientError, ServerTransmission (..), ServerTransmissionBatch)
 import Simplex.Messaging.Client.Agent
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
@@ -69,7 +68,7 @@ import Simplex.Messaging.Notifications.Server.Store (NtfSTMStore, TokenNtfMessag
 import Simplex.Messaging.Notifications.Server.Store.Postgres
 import Simplex.Messaging.Notifications.Server.Store.Types
 import Simplex.Messaging.Notifications.Transport
-import Simplex.Messaging.Protocol (EntityId (..), ErrorType (..), NotifierId, Party (..), ProtocolServer (host), SMPServer, ServiceSub (..), SignedTransmission, Transmission, pattern NoEntity, pattern SMPServer, encodeTransmission, tGetServer, tPut)
+import Simplex.Messaging.Protocol (BrokerMsg, EntityId (..), ErrorType (..), NotifierId, Party (..), ProtocolServer (host), SMPServer, ServiceSub (..), SignedTransmission, Transmission, pattern NoEntity, pattern SMPServer, encodeTransmission, tGetServer, tPut)
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Server
 import Simplex.Messaging.Server.Control (CPClientRole (..))
@@ -78,7 +77,7 @@ import Simplex.Messaging.Server.Stats (PeriodStats (..), PeriodStatCounts (..),
 import Simplex.Messaging.Session
 import Simplex.Messaging.SystemTime
 import Simplex.Messaging.TMap (TMap)
-import Simplex.Messaging.Transport (ASrvTransport, ATransport (..), THandle (..), THandleAuth (..), THandleParams (..), TProxy, Transport (..), TransportPeer (..), defaultSupportedParams)
+import Simplex.Messaging.Transport (ASrvTransport, ATransport (..), SMPVersion, THandle (..), THandleAuth (..), THandleParams (..), TProxy, Transport (..), TransportPeer (..), defaultSupportedParams)
 import Simplex.Messaging.Transport.Buffer (trimCR)
 import Simplex.Messaging.Transport.Server (AddHTTP, runTransportServer, runLocalTCPServer)
 import Simplex.Messaging.Util
@@ -102,7 +101,7 @@ runNtfServer cfg = do
   runNtfServerBlocking started cfg
 
 runNtfServerBlocking :: TMVar Bool -> NtfServerConfig -> IO ()
-runNtfServerBlocking started cfg = runReaderT (ntfServer cfg started) =<< newNtfServerEnv cfg
+runNtfServerBlocking started cfg = runReaderT (ntfServer cfg started) =<< newNtfServerEnv cfg receiveSMPMessage
 
 type M a = ReaderT NtfEnv IO a
 
@@ -110,7 +109,6 @@ ntfServer :: NtfServerConfig -> TMVar Bool -> M ()
 ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg, startOptions} started = do
   restoreServerStats
   s <- asks subscriber
-  ps <- asks pushServer
   when (maintenance startOptions) $ do
     liftIO $ putStrLn "Server started in 'maintenance' mode, exiting"
     stopServer
@@ -118,7 +116,7 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg, startOptions}
   void $ forkIO $ resubscribe s
   raceAny_
     ( ntfSubscriber s
-        : periodicNtfsThread ps
+        : periodicNtfsThread
         : map runServer transports
           <> serverStatsThread_ cfg
           <> prometheusMetricsThread_ cfg
@@ -526,92 +524,83 @@ subscribeNtfs NtfSubscriber {smpSubscribers, subscriberSeq, smpAgent = ca} st sm
       void $ updateSubStatus st srvId' nId NSPending
       subscribeQueuesNtfs ca smpServer' [sub]
 
+receiveSMPMessage :: NtfEnv -> SMPClientAgent 'NotifierService -> ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> IO ()
+receiveSMPMessage env@NtfEnv {store = st, serverStats = stats} ca ((_, srv@(SMPServer (h :| _) _ _), _), THandleParams {sessionId}, ts) =
+  forM_ ts $ \(ntfId, t) -> case t of
+    STUnexpectedError e -> logError $ "SMP client unexpected error: " <> tshow e -- uncorrelated response, should not happen
+    STResponse {} -> pure () -- it was already reported as timeout error
+    STEvent msgOrErr -> do
+      let smpQueue = SMPQueueNtf srv ntfId
+      case msgOrErr of
+        Right (SMP.NMSG nmsgNonce encNMsgMeta) -> do
+          ntfTs <- getSystemTime
+          updatePeriodStats (activeSubs stats) ntfId
+          let newNtf = PNMessageData {smpQueue, ntfTs, nmsgNonce, encNMsgMeta}
+              srvHost = safeDecodeUtf8 $ strEncode h
+              isOwn = isOwnServer ca srv
+          addTokenLastNtf st newNtf >>= \case
+            Right (tkn, lastNtfs) -> do
+              pushNotification env (Just srvHost) isOwn tkn $ PNMessage lastNtfs
+              incNtfStat_ stats ntfReceived
+              when isOwn $ incServerStat srvHost (ntfReceivedOwn stats)
+            Left AUTH -> do
+              incNtfStat_ stats ntfReceivedAuth
+              when isOwn $ incServerStat srvHost (ntfReceivedAuthOwn stats)
+            Left _ -> pure ()
+        Right SMP.END ->
+          whenM (atomically $ activeClientSession' ca sessionId srv) $
+            void $ updateSrvSubStatus st smpQueue NSEnd
+        Right SMP.DELD ->
+          void $ updateSrvSubStatus st smpQueue NSDeleted
+        Right (SMP.ERR e) -> logError $ "SMP server error: " <> tshow e
+        Right _ -> logError "SMP server unexpected response"
+        Left e -> logError $ "SMP client error: " <> tshow e
+
 ntfSubscriber :: NtfSubscriber -> M ()
-ntfSubscriber NtfSubscriber {smpAgent = ca@SMPClientAgent {msgQ = msgQ_, agentQ}} =
-  race_ receiveSMP receiveAgent
+ntfSubscriber NtfSubscriber {smpAgent = ca@SMPClientAgent {agentQ}} = do
+  st <- asks store
+  batchSize <- asks $ subsBatchSize . config
+  liftIO $ forever $
+    atomically (readTBQueue agentQ) >>= \case
+      CAConnected srv serviceId -> do
+        let asService = if isJust serviceId then "as service " else ""
+        logInfo $ "SMP server reconnected " <> asService <> showServer' srv
+      CADisconnected srv nIds -> do
+        updated <- batchUpdateSrvSubStatus st srv Nothing nIds NSInactive
+        logSubStatus srv "disconnected" (L.length nIds) updated
+      CASubscribed srv serviceId nIds -> do
+        updated <- batchUpdateSrvSubStatus st srv serviceId nIds NSActive
+        let asService = if isJust serviceId then " as service" else ""
+        logSubStatus srv ("subscribed" <> asService) (L.length nIds) updated
+      CASubError srv errs -> do
+        forM_ (L.nonEmpty $ mapMaybe (\(nId, err) -> (nId,) <$> queueSubErrorStatus err) $ L.toList errs) $ \subStatuses -> do
+          updated <- batchUpdateSrvSubErrors st srv subStatuses
+          logSubErrors srv subStatuses updated
+          -- TODO [certs rcv] resubscribe queues with statuses NSErr and NSService
+      CAServiceDisconnected srv serviceSub ->
+        logNote $ "SMP server service disconnected " <> showService srv serviceSub
+      CAServiceSubscribed srv serviceSub@(ServiceSub _ n idsHash) (ServiceSub _ n' idsHash')
+        | n /= n' -> logWarn $ msg <> ", confirmed subs: " <> tshow n'
+        | idsHash /= idsHash' -> logWarn $ msg <> ", different IDs hash"
+        | otherwise -> logNote msg
+        where
+          msg = "SMP server service subscribed " <> showService srv serviceSub
+      CAServiceSubError srv serviceSub e ->
+        -- Errors that require re-subscribing queues directly are reported as CAServiceUnavailable.
+        -- See smpSubscribeService in Simplex.Messaging.Client.Agent
+        logError $ "SMP server service subscription error " <> showService srv serviceSub <> ": " <> tshow e
+      CAServiceUnavailable srv serviceSub -> do
+        logError $ "SMP server service unavailable: " <> showService srv serviceSub
+        removeServiceAndAssociations st srv >>= \case
+          Right (srvId, updated) -> do
+            logSubStatus srv "removed service association" updated updated
+            void $ subscribeSrvSubs ca st batchSize (srv, srvId, Nothing)
+          Left e -> logError $ "SMP server update and resubscription error " <> tshow e
   where
-    receiveSMP = forM_ msgQ_ $ \msgQ -> do
-      st <- asks store
-      ps <- asks pushServer
-      stats <- asks serverStats
-      forever $ do
-        ((_, srv@(SMPServer (h :| _) _ _), _), THandleParams {sessionId}, ts) <- atomically $ readTBQueue msgQ
-        forM_ ts $ \(ntfId, t) -> case t of
-          STUnexpectedError e -> logError $ "SMP client unexpected error: " <> tshow e -- uncorrelated response, should not happen
-          STResponse {} -> pure () -- it was already reported as timeout error
-          STEvent msgOrErr -> do
-            let smpQueue = SMPQueueNtf srv ntfId
-            case msgOrErr of
-              Right (SMP.NMSG nmsgNonce encNMsgMeta) -> do
-                ntfTs <- liftIO getSystemTime
-                liftIO $ updatePeriodStats (activeSubs stats) ntfId
-                let newNtf = PNMessageData {smpQueue, ntfTs, nmsgNonce, encNMsgMeta}
-                    srvHost = safeDecodeUtf8 $ strEncode h
-                    isOwn = isOwnServer ca srv
-                liftIO (addTokenLastNtf st newNtf) >>= \case
-                  Right (tkn, lastNtfs) -> do
-                    pushNotification ps (Just srvHost) isOwn tkn $ PNMessage lastNtfs
-                    liftIO $ incNtfStat_ stats ntfReceived
-                    when isOwn $ liftIO $ incServerStat srvHost (ntfReceivedOwn stats)
-                  Left AUTH -> liftIO $ do
-                    incNtfStat_ stats ntfReceivedAuth
-                    when isOwn $ incServerStat srvHost (ntfReceivedAuthOwn stats)
-                  Left _ -> pure ()
-              Right SMP.END ->
-                whenM (atomically $ activeClientSession' ca sessionId srv) $
-                  void $ liftIO $ updateSrvSubStatus st smpQueue NSEnd
-              Right SMP.DELD ->
-                void $ liftIO $ updateSrvSubStatus st smpQueue NSDeleted
-              Right (SMP.ERR e) -> logError $ "SMP server error: " <> tshow e
-              Right _ -> logError "SMP server unexpected response"
-              Left e -> logError $ "SMP client error: " <> tshow e
-
-    receiveAgent = do
-      st <- asks store
-      batchSize <- asks $ subsBatchSize . config
-      liftIO $ forever $
-        atomically (readTBQueue agentQ) >>= \case
-          CAConnected srv serviceId -> do
-            let asService = if isJust serviceId then "as service " else ""
-            logInfo $ "SMP server reconnected " <> asService <> showServer' srv
-          CADisconnected srv nIds -> do
-            updated <- batchUpdateSrvSubStatus st srv Nothing nIds NSInactive
-            logSubStatus srv "disconnected" (L.length nIds) updated
-          CASubscribed srv serviceId nIds -> do
-            updated <- batchUpdateSrvSubStatus st srv serviceId nIds NSActive
-            let asService = if isJust serviceId then " as service" else ""
-            logSubStatus srv ("subscribed" <> asService) (L.length nIds) updated
-          CASubError srv errs -> do
-            forM_ (L.nonEmpty $ mapMaybe (\(nId, err) -> (nId,) <$> queueSubErrorStatus err) $ L.toList errs) $ \subStatuses -> do
-              updated <- batchUpdateSrvSubErrors st srv subStatuses
-              logSubErrors srv subStatuses updated
-              -- TODO [certs rcv] resubscribe queues with statuses NSErr and NSService
-          CAServiceDisconnected srv serviceSub ->
-            logNote $ "SMP server service disconnected " <> showService srv serviceSub
-          CAServiceSubscribed srv serviceSub@(ServiceSub _ n idsHash) (ServiceSub _ n' idsHash')
-            | n /= n' -> logWarn $ msg <> ", confirmed subs: " <> tshow n'
-            | idsHash /= idsHash' -> logWarn $ msg <> ", different IDs hash"
-            | otherwise -> logNote msg
-            where
-              msg = "SMP server service subscribed " <> showService srv serviceSub
-          CAServiceSubError srv serviceSub e ->
-            -- Errors that require re-subscribing queues directly are reported as CAServiceUnavailable.
-            -- See smpSubscribeService in Simplex.Messaging.Client.Agent
-            logError $ "SMP server service subscription error " <> showService srv serviceSub <> ": " <> tshow e
-          CAServiceUnavailable srv serviceSub -> do
-            logError $ "SMP server service unavailable: " <> showService srv serviceSub
-            removeServiceAndAssociations st srv >>= \case
-              Right (srvId, updated) -> do
-                logSubStatus srv "removed service association" updated updated
-                void $ subscribeSrvSubs ca st batchSize (srv, srvId, Nothing)
-              Left e -> logError $ "SMP server update and resubscription error " <> tshow e
-      where
-        showService srv (ServiceSub serviceId n _) = showServer' srv  <> ", service ID " <> decodeLatin1 (strEncode serviceId) <> ", " <> tshow n <> " subs"
-
+    showService srv (ServiceSub serviceId n _) = showServer' srv <> ", service ID " <> decodeLatin1 (strEncode serviceId) <> ", " <> tshow n <> " subs"
     logSubErrors :: SMPServer -> NonEmpty (SMP.NotifierId, NtfSubStatus) -> Int -> IO ()
-    logSubErrors srv subs updated = forM_ (L.group $ L.sort $ L.map snd subs) $ \ss -> do
+    logSubErrors srv subs updated = forM_ (L.group $ L.sort $ L.map snd subs) $ \ss ->
       logError $ "SMP server subscription errors " <> showServer' srv <> ": " <> tshow (L.head ss) <> " (" <> tshow (length ss) <> " errors, " <> tshow updated <> " subs updated)"
-
     queueSubErrorStatus :: SMPClientError -> Maybe NtfSubStatus
     queueSubErrorStatus = \case
       PCEProtocolError AUTH -> Just NSAuth
@@ -640,71 +629,66 @@ logSubStatus srv event n updated =
 showServer' :: SMPServer -> Text
 showServer' = decodeLatin1 . strEncode . host
 
-pushNotification :: NtfPushServer -> Maybe T.Text -> OwnServer -> NtfTknRec -> PushNotification -> M ()
-pushNotification s srvHost_ isOwn tkn@NtfTknRec {ntfTknId, token = token@(DeviceToken pp _)} ntf =
-  ifM
-    (pushProviderAllowed token)
-    (getOrCreatePushWorker s (srvHost_, pp, hash (unEntityId ntfTknId) `mod` pushWorkersPerServer) isOwn >>= atomically . (`writeTBQueue` (tkn, ntf)))
-    (logWarn "skipping disabled APNS test push provider")
-  where
-    pushWorkersPerServer = 8
+pushNotification :: NtfEnv -> Maybe T.Text -> OwnServer -> NtfTknRec -> PushNotification -> IO ()
+pushNotification env srvHost_ isOwn tkn@NtfTknRec {token = token@(DeviceToken pp _)} ntf =
+  if pushProviderAllowed env token
+    then getOrCreatePushWorker env (srvHost_, pp) isOwn >>= atomically . (`writeTBQueue` (tkn, ntf))
+    else logWarn "skipping disabled APNS test push provider"
 
-pushProviderAllowed :: DeviceToken -> M Bool
-pushProviderAllowed (DeviceToken PPApnsTest _) = asks (allowTestPushProvider . config)
-pushProviderAllowed _ = pure True
+pushProviderAllowed :: NtfEnv -> DeviceToken -> Bool
+pushProviderAllowed NtfEnv {config} (DeviceToken PPApnsTest _) = allowTestPushProvider config
+pushProviderAllowed _ _ = True
 
 guardPushProvider :: DeviceToken -> M NtfResponse -> M NtfResponse
 guardPushProvider token action =
   ifM
-    (pushProviderAllowed token)
+    ((`pushProviderAllowed` token) <$> ask)
     action
     (pure $ NRErr $ CMD SMP.PROHIBITED)
 
-getOrCreatePushWorker :: NtfPushServer -> (Maybe T.Text, PushProvider, Int) -> OwnServer -> M (TBQueue (NtfTknRec, PushNotification))
-getOrCreatePushWorker s@NtfPushServer {pushWorkers, pushWorkerSeq, pushQSize} key@(srvHost_, _, _) isOwn = do
-  ts <- liftIO getCurrentTime
+getOrCreatePushWorker :: NtfEnv -> (Maybe T.Text, PushProvider) -> OwnServer -> IO (TBQueue (NtfTknRec, PushNotification))
+getOrCreatePushWorker env@NtfEnv {pushServer = NtfPushServer {pushWorkers, pushWorkerSeq, pushQSize}} key@(srvHost_, _) isOwn = do
+  ts <- getCurrentTime
   withGetSessVar' pushWorkerSeq key pushWorkers ts createWorker existingWorker
   where
     createWorker v = do
-      q <- liftIO $ newTBQueueIO pushQSize
-      tId <- mkWeakThreadId =<< forkIO (runPushWorker s srvHost_ isOwn q)
+      q <- newTBQueueIO pushQSize
+      tId <- mkWeakThreadId =<< forkIO (runPushWorker env srvHost_ isOwn q)
       atomically $ putTMVar (sessionVar v) PushWorker {workerQ = q, workerThreadId = tId}
       pure q
     existingWorker v = workerQ <$> atomically (readTMVar $ sessionVar v)
 
-runPushWorker :: NtfPushServer -> Maybe T.Text -> OwnServer -> TBQueue (NtfTknRec, PushNotification) -> M ()
-runPushWorker s srvHost_ isOwn q = forever $ do
+runPushWorker :: NtfEnv -> Maybe T.Text -> OwnServer -> TBQueue (NtfTknRec, PushNotification) -> IO ()
+runPushWorker NtfEnv {store = st, serverStats = stats, pushServer = s} srvHost_ isOwn q = forever $ do
   (tkn@NtfTknRec {ntfTknId, token = t@(DeviceToken pp _), tknStatus}, ntf) <- atomically (readTBQueue q)
-  liftIO $ logDebug $ "sending push notification to " <> T.pack (show pp)
-  st <- asks store
+  logDebug $ "sending push notification to " <> T.pack (show pp)
   case ntf of
     PNVerification _ ->
-      liftIO (deliverNotification st pp tkn ntf) >>= \case
+      deliverNotification st pp tkn ntf >>= \case
         Right _ -> do
-          void $ liftIO $ setTknStatusConfirmed st tkn
-          incNtfStatT t ntfVrfDelivered
-        Left _ -> incNtfStatT t ntfVrfFailed
+          void $ setTknStatusConfirmed st tkn
+          incNtfStatT_ stats t ntfVrfDelivered
+        Left _ -> incNtfStatT_ stats t ntfVrfFailed
     PNCheckMessages ->
-      liftIO (deliverNotification st pp tkn ntf) >>= \case
+      deliverNotification st pp tkn ntf >>= \case
         Right _ -> do
-          void $ liftIO $ updateTokenCronSentAt st ntfTknId . systemSeconds =<< getSystemTime
-          incNtfStatT t ntfCronDelivered
-        Left _ -> incNtfStatT t ntfCronFailed
+          void $ updateTokenCronSentAt st ntfTknId . systemSeconds =<< getSystemTime
+          incNtfStatT_ stats t ntfCronDelivered
+        Left _ -> incNtfStatT_ stats t ntfCronFailed
     PNMessage {} -> checkActiveTkn tknStatus $ do
-      stats <- asks serverStats
-      liftIO $ updatePeriodStats (activeTokens stats) ntfTknId
-      liftIO (deliverNotification st pp tkn ntf) >>= \case
+      updatePeriodStats (activeTokens stats) ntfTknId
+      deliverNotification st pp tkn ntf >>= \case
         Left _ -> do
-          incNtfStatT t ntfFailed
-          when isOwn $ liftIO $ mapM_ (`incServerStat` ntfFailedOwn stats) srvHost_
+          incNtfStatT_ stats t ntfFailed
+          when isOwn $ mapM_ (`incServerStat` ntfFailedOwn stats) srvHost_
         Right () -> do
-          incNtfStatT t ntfDelivered
-          when isOwn $ liftIO $ mapM_ (`incServerStat` ntfDeliveredOwn stats) srvHost_
+          incNtfStatT_ stats t ntfDelivered
+          when isOwn $ mapM_ (`incServerStat` ntfDeliveredOwn stats) srvHost_
   where
-    checkActiveTkn :: NtfTknStatus -> M () -> M ()
+    checkActiveTkn :: NtfTknStatus -> IO () -> IO ()
     checkActiveTkn status action
       | status == NTActive = action
-      | otherwise = liftIO $ logError "bad notification token status"
+      | otherwise = logError "bad notification token status"
     deliverNotification :: NtfPostgresStore -> PushProvider -> NtfTknRec -> PushNotification -> IO (Either PushProviderError ())
     deliverNotification st pp tkn@NtfTknRec {ntfTknId} ntf' = do
       (deliver, clientVar) <- getPushClient s pp
@@ -734,7 +718,7 @@ runPushWorker s srvHost_ isOwn q = forever $ do
               _ -> err e
         err e = logError ("Push provider error (" <> tshow pp <> ", " <> tshow ntfTknId <> "): " <> tshow e) $> Left e
 
-pushWorkersQLength :: TMap (Maybe T.Text, PushProvider, Int) PushWorkerVar -> IO Natural
+pushWorkersQLength :: TMap (Maybe T.Text, PushProvider) PushWorkerVar -> IO Natural
 pushWorkersQLength workers = do
   ws <- readTVarIO workers
   foldM addQLength 0 ws
@@ -744,16 +728,16 @@ pushWorkersQLength workers = do
         Just PushWorker {workerQ} -> (acc +) <$> atomically (lengthTBQueue workerQ)
         Nothing -> pure acc
 
-periodicNtfsThread :: NtfPushServer -> M ()
-periodicNtfsThread s = do
+periodicNtfsThread :: M ()
+periodicNtfsThread = do
+  env <- ask
   st <- asks store
   ntfsInterval <- asks $ periodicNtfsInterval . config
   let interval = 1000000 * ntfsInterval
-  UnliftIO unlift <- askUnliftIO
   liftIO $ forever $ do
     threadDelay interval
     now <- systemSeconds <$> getSystemTime
-    cnt <- withPeriodicNtfTokens st now $ \tkn -> unlift $ pushNotification s Nothing False tkn PNCheckMessages
+    cnt <- withPeriodicNtfTokens st now $ \tkn -> pushNotification env Nothing False tkn PNCheckMessages
     logNote $ "Scheduled periodic notifications: " <> tshow cnt
 
 runNtfClientTransport :: Transport c => THandleNTF c 'TServer -> M ()
@@ -762,10 +746,9 @@ runNtfClientTransport th@THandle {params} = do
   ts <- liftIO getSystemTime
   c <- liftIO $ newNtfServerClient qSize params ts
   s <- asks subscriber
-  ps <- asks pushServer
   expCfg <- asks $ inactiveClientExpiration . config
   st <- asks store
-  raceAny_ ([liftIO $ send th c, client c s ps, liftIO $ receive st th c] <> disconnectThread_ c expCfg)
+  raceAny_ ([liftIO $ send th c, client c s, liftIO $ receive st th c] <> disconnectThread_ c expCfg)
     `finally` liftIO (clientDisconnected c)
   where
     disconnectThread_ c (Just expCfg) = [liftIO $ disconnectTransport th (rcvActiveAt c) (sndActiveAt c) expCfg (pure True)]
@@ -842,16 +825,17 @@ verifyNtfTransmission st thAuth (tAuth, authorized, (corrId, entId, cmd)) = case
       AUTH -> dummyVerifyCmd thAuth tAuth authorized corrId `seq` VRFailed AUTH
       e -> VRFailed e
 
-client :: NtfServerClient -> NtfSubscriber -> NtfPushServer -> M ()
-client NtfServerClient {rcvQ, sndQ} ns@NtfSubscriber {smpAgent = ca} ps =
+client :: NtfServerClient -> NtfSubscriber -> M ()
+client NtfServerClient {rcvQ, sndQ} ns@NtfSubscriber {smpAgent = ca} = do
+  env <- ask
   forever $
     atomically (readTBQueue rcvQ)
-      >>= mapM processCommand
+      >>= mapM (processCommand env)
       >>= atomically . writeTBQueue sndQ
   where
-    processCommand :: NtfRequest -> M (Transmission NtfResponse)
-    processCommand = \case
-      NtfReqNew corrId (ANE SToken newTkn@(NewNtfTkn token _ dhPubKey)) -> (corrId,NoEntity,) <$> guardPushProvider token (do
+    processCommand :: NtfEnv -> NtfRequest -> M (Transmission NtfResponse)
+    processCommand env = \case
+      NtfReqNew corrId (ANE SToken newTkn@(NewNtfTkn token _ dhPubKey)) -> fmap (corrId,NoEntity,) $ guardPushProvider token $ do
         logDebug "TNEW - new token"
         (srvDhPubKey, srvDhPrivKey) <- atomically . C.generateKeyPair =<< asks random
         let dhSecret = C.dh' dhPubKey srvDhPrivKey
@@ -860,10 +844,10 @@ client NtfServerClient {rcvQ, sndQ} ns@NtfSubscriber {smpAgent = ca} ps =
         ts <- liftIO $ getSystemDate
         let tkn = mkNtfTknRec tknId newTkn srvDhPrivKey dhSecret regCode ts
         withNtfStore (`addNtfToken` tkn) $ \_ -> do
-          pushNotification ps Nothing False tkn $ PNVerification regCode
+          liftIO $ pushNotification env Nothing False tkn $ PNVerification regCode
           incNtfStatT token ntfVrfQueued
           incNtfStatT token tknCreated
-          pure $ NRTknId tknId srvDhPubKey)
+          pure $ NRTknId tknId srvDhPubKey
       NtfReqCmd SToken (NtfTkn tkn@NtfTknRec {token, ntfTknId, tknStatus, tknRegCode, tknDhSecret, tknDhPrivKey}) (corrId, tknId, cmd) -> do
         (corrId,tknId,) <$> case cmd of
           TNEW (NewNtfTkn _ _ dhPubKey) -> guardPushProvider token $ do
@@ -876,7 +860,7 @@ client NtfServerClient {rcvQ, sndQ} ns@NtfSubscriber {smpAgent = ca} ps =
               | otherwise -> withNtfStore (\st -> updateTknStatus st tkn NTRegistered) $ \_ -> sendVerification
             where
               sendVerification = do
-                pushNotification ps Nothing False tkn $ PNVerification tknRegCode
+                liftIO $ pushNotification env Nothing False tkn $ PNVerification tknRegCode
                 incNtfStatT token ntfVrfQueued
                 pure $ NRTknId ntfTknId $ C.publicKey tknDhPrivKey
           TVFY code -- this allows repeated verification for cases when client connection dropped before server response
@@ -894,7 +878,7 @@ client NtfServerClient {rcvQ, sndQ} ns@NtfSubscriber {smpAgent = ca} ps =
             regCode <- getRegCode
             let tkn' = tkn {token = token', tknStatus = NTRegistered, tknRegCode = regCode}
             withNtfStore (`replaceNtfToken` tkn') $ \_ -> do
-              pushNotification ps Nothing False tkn' $ PNVerification regCode
+              liftIO $ pushNotification env Nothing False tkn' $ PNVerification regCode
               incNtfStatT token ntfVrfQueued
               incNtfStatT token tknReplaced
               pure NROk
@@ -966,6 +950,11 @@ incNtfStatT (DeviceToken PPApnsNull _) _ = pure ()
 incNtfStatT _ statSel = incNtfStat statSel
 {-# INLINE incNtfStatT #-}
 
+incNtfStatT_ :: NtfServerStats -> DeviceToken -> (NtfServerStats -> IORef Int) -> IO ()
+incNtfStatT_ _ (DeviceToken PPApnsNull _) _ = pure ()
+incNtfStatT_ stats _ statSel = incNtfStat_ stats statSel
+{-# INLINE incNtfStatT_ #-}
+
 incNtfStat :: (NtfServerStats -> IORef Int) -> M ()
 incNtfStat statSel = asks serverStats >>= liftIO . (`incNtfStat_` statSel)
 {-# INLINE incNtfStat #-}

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

@@ -6,6 +6,7 @@
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedLists #-}
 {-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE RecursiveDo #-}
 
 module Simplex.Messaging.Notifications.Server.Env
   ( NtfServerConfig (..),
@@ -45,7 +46,7 @@ import qualified Data.X509.Validation as XV
 import Network.Socket
 import qualified Network.TLS as TLS
 import Numeric.Natural
-import Simplex.Messaging.Client (ProtocolClientError (..), SMPClientError)
+import Simplex.Messaging.Client (ProtocolClientError (..), SMPClientError, ServerTransmissionBatch)
 import Simplex.Messaging.Client.Agent
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Notifications.Protocol
@@ -54,14 +55,14 @@ import Simplex.Messaging.Notifications.Server.Stats
 import Simplex.Messaging.Notifications.Server.Store.Postgres
 import Simplex.Messaging.Notifications.Server.Store.Types
 import Simplex.Messaging.Notifications.Transport (NTFVersion, VersionRangeNTF)
-import Simplex.Messaging.Protocol (BasicAuth, CorrId, Party (..), SMPServer, SParty (..), ServiceId, Transmission)
+import Simplex.Messaging.Protocol (BasicAuth, BrokerMsg, CorrId, ErrorType, Party (..), SMPServer, SParty (..), ServiceId, Transmission)
 import Simplex.Messaging.Server.Env.STM (StartOptions (..))
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.QueueStore.Postgres.Config (PostgresStoreCfg (..))
 import Simplex.Messaging.Session
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
-import Simplex.Messaging.Transport (ASrvTransport, SMPServiceRole (..), ServiceCredentials (..), THandleParams, TransportPeer (..))
+import Simplex.Messaging.Transport (ASrvTransport, SMPServiceRole (..), SMPVersion, ServiceCredentials (..), THandleParams, TransportPeer (..))
 import Simplex.Messaging.Transport.Credentials (genCredentials, tlsCredentials)
 import Simplex.Messaging.Transport.Server (AddHTTP, ServerCredentials, TransportServerConfig, loadFingerprint, loadServerCredential)
 import Simplex.Messaging.Util (liftEitherWith, tshow)
@@ -120,17 +121,18 @@ data NtfEnv = NtfEnv
     serverStats :: NtfServerStats
   }
 
-newNtfServerEnv :: NtfServerConfig -> IO NtfEnv
-newNtfServerEnv config@NtfServerConfig {pushQSize, smpAgentCfg, apnsConfig, dbStoreConfig, ntfCredentials, useServiceCreds} = do
+newNtfServerEnv :: NtfServerConfig -> (NtfEnv -> SMPClientAgent 'NotifierService -> ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> IO ()) -> IO NtfEnv
+newNtfServerEnv config@NtfServerConfig {pushQSize, smpAgentCfg, apnsConfig, dbStoreConfig, ntfCredentials, useServiceCreds} mkProcessMsg = do
   random <- C.newRandom
   store <- newNtfDbStore dbStoreConfig
   tlsServerCreds <- loadServerCredential ntfCredentials
   XV.Fingerprint fp <- loadFingerprint ntfCredentials
   let dbService = if useServiceCreds then Just $ mkDbService random store else Nothing
-  subscriber <- newNtfSubscriber smpAgentCfg dbService random
   pushServer <- newNtfPushServer pushQSize apnsConfig
   serverStats <- newNtfServerStats =<< getCurrentTime
-  pure NtfEnv {config, subscriber, pushServer, store, random, tlsServerCreds, serverIdentity = C.KeyHash fp, serverStats}
+  rec subscriber <- newNtfSubscriber smpAgentCfg (mkProcessMsg env) dbService random
+      let env = NtfEnv {config, subscriber, pushServer, store, random, tlsServerCreds, serverIdentity = C.KeyHash fp, serverStats}
+  pure env
   where
     mkDbService g st = DBService {getCredentials, updateServiceId}
       where
@@ -159,11 +161,11 @@ data NtfSubscriber = NtfSubscriber
 
 type SMPSubscriberVar = SessionVar SMPSubscriber
 
-newNtfSubscriber :: SMPClientAgentConfig -> Maybe DBService -> TVar ChaChaDRG -> IO NtfSubscriber
-newNtfSubscriber smpAgentCfg dbService random = do
+newNtfSubscriber :: SMPClientAgentConfig -> (SMPClientAgent 'NotifierService -> ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> IO ()) -> Maybe DBService -> TVar ChaChaDRG -> IO NtfSubscriber
+newNtfSubscriber smpAgentCfg processMsg dbService random = do
   smpSubscribers <- TM.emptyIO
   subscriberSeq <- newTVarIO 0
-  smpAgent <- newSMPClientAgent SNotifierService smpAgentCfg dbService random
+  smpAgent <- newSMPClientAgent SNotifierService smpAgentCfg processMsg dbService random
   pure NtfSubscriber {smpSubscribers, subscriberSeq, smpAgent}
 
 data SMPSubscriber = SMPSubscriber
@@ -174,7 +176,7 @@ data SMPSubscriber = SMPSubscriber
   }
 
 data NtfPushServer = NtfPushServer
-  { pushWorkers :: TMap (Maybe T.Text, PushProvider, Int) PushWorkerVar, -- Int is the worker shard
+  { pushWorkers :: TMap (Maybe T.Text, PushProvider) PushWorkerVar,
     pushWorkerSeq :: TVar Int,
     pushQSize :: Natural,
     pushClients :: TMap PushProvider PushClientVar,

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

@@ -730,7 +730,7 @@ mkJournalStoreConfig queueStoreCfg storePath msgQueueQuota maxJournalMsgCount ma
 
 newSMPProxyAgent :: SMPClientAgentConfig -> TVar ChaChaDRG -> IO ProxyAgent
 newSMPProxyAgent smpAgentCfg random = do
-  smpAgent <- newSMPClientAgent SSender smpAgentCfg Nothing random
+  smpAgent <- newSMPClientAgent SSender smpAgentCfg (\_ _ -> pure ()) Nothing random
   pure ProxyAgent {smpAgent}
 
 readWriteQueueStore :: forall q. StoreQueueClass q => Bool -> (RecipientId -> QueueRec -> IO q) -> FilePath -> STMQueueStore q -> IO (StoreLog 'WriteMode)

src/Simplex/Messaging/Server/Main.hs

@@ -604,7 +604,6 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
                               }
                         },
                     ownServerDomains = either (const []) textToOwnServers $ lookupValue "PROXY" "own_server_domains" ini,
-                    msgQSize = Nothing, -- to prevent accumulation of late responses, and deadlocks in SMP proxy
                     persistErrorInterval = 30 -- seconds
                   },
               allowSMPProxy = True,

tests/AgentTests/NotificationTests.hs

@@ -17,7 +17,10 @@ module AgentTests.NotificationTests where
 
 -- import Control.Logger.Simple (LogConfig (..), LogLevel (..), setLogLevel, withGlobalLogging)
 import AgentTests.FunctionalAPITests
-  ( agentCfgVPrevPQ,
+  ( AgentClient (..),
+    ackMessage,
+    agentCfgVPrevPQ,
+    allowConnection,
     createConnection,
     exchangeGreetings,
     get,
@@ -27,6 +30,7 @@ import AgentTests.FunctionalAPITests
     runRight,
     runRight_,
     sendMessage,
+    subscribeConnection,
     switchComplete,
     testServerMatrix2,
     withAgent,
@@ -61,7 +65,7 @@ import qualified Database.PostgreSQL.Simple as PSQL
 import NtfClient
 import SMPAgentClient (agentCfg, initAgentServers, initAgentServers2, testDB, testDB2, testNtfServer, testNtfServer2)
 import SMPClient
-import Simplex.Messaging.Agent hiding (checkNtfToken, createConnection, joinConnection, registerNtfToken, sendMessage, verifyNtfToken)
+import Simplex.Messaging.Agent hiding (AgentClient, ackMessage, allowConnection, checkNtfToken, createConnection, deleteNtfToken, foregroundAgent, getConnectionMessages, getNtfTokenData, getNotificationConns, joinConnection, registerNtfToken, sendMessage, setNtfServers, subscribeConnection, suspendAgent, switchConnectionAsync, toggleConnectionNtfs, verifyNtfToken)
 import qualified Simplex.Messaging.Agent as A
 import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestStep (..), withStore')
 import Simplex.Messaging.Agent.Env.SQLite (AgentConfig, Env (..), InitialAgentServers)
@@ -197,13 +201,41 @@ testNtfMatrix ps@(_, msType) runTest = do
     cfgVPrev' = cfgVPrev msType
 
 registerNtfToken :: AgentClient -> DeviceToken -> NotificationsMode -> AE NtfTknStatus
-registerNtfToken c = A.registerNtfToken c NRMInteractive
+registerNtfToken c = A.registerNtfToken (client c) NRMInteractive
 
 checkNtfToken :: AgentClient -> DeviceToken -> AE NtfTknStatus
-checkNtfToken c = A.checkNtfToken c NRMInteractive
+checkNtfToken c = A.checkNtfToken (client c) NRMInteractive
 
 verifyNtfToken :: AgentClient -> DeviceToken -> C.CbNonce -> ByteString -> AE ()
-verifyNtfToken c = A.verifyNtfToken c NRMInteractive
+verifyNtfToken c = A.verifyNtfToken (client c) NRMInteractive
+
+deleteNtfToken :: AgentClient -> DeviceToken -> AE ()
+deleteNtfToken c = A.deleteNtfToken (client c)
+
+getNtfTokenData :: AgentClient -> AE NtfToken
+getNtfTokenData = A.getNtfTokenData . client
+
+setNtfServers :: AgentClient -> [NtfServer] -> IO ()
+setNtfServers c = A.setNtfServers (client c)
+
+foregroundAgent :: AgentClient -> IO ()
+foregroundAgent = A.foregroundAgent . client
+
+suspendAgent :: AgentClient -> Int -> IO ()
+suspendAgent c = A.suspendAgent (client c)
+
+toggleConnectionNtfs :: AgentClient -> ConnId -> Bool -> AE ()
+toggleConnectionNtfs c = A.toggleConnectionNtfs (client c)
+
+getNotificationConns :: AgentClient -> C.CbNonce -> ByteString -> AE (NonEmpty NotificationInfo)
+getNotificationConns c = A.getNotificationConns (client c)
+
+getConnectionMessages :: AgentClient -> NonEmpty ConnMsgReq -> IO (NonEmpty (Either AgentErrorType (Maybe SMPMsgMeta)))
+getConnectionMessages c = A.getConnectionMessages (client c)
+
+switchConnectionAsync :: AgentClient -> ACorrId -> ConnId -> AE ConnectionStats
+switchConnectionAsync c = A.switchConnectionAsync (client c)
+
 
 runNtfTestCfg :: HasCallStack => (ASrvTransport, AStoreType) -> AgentMsgId -> AServerConfig -> NtfServerConfig -> AgentConfig -> AgentConfig -> (APNSMockServer -> AgentMsgId -> AgentClient -> AgentClient -> IO ()) -> IO ()
 runNtfTestCfg (t, msType) baseId smpCfg ntfCfg aCfg bCfg runTest = do
@@ -335,7 +367,7 @@ testNtfTokenServerRestartReverify t apns = do
 testNtfTokenServerRestartReverifyTimeout :: ASrvTransport -> APNSMockServer -> IO ()
 testNtfTokenServerRestartReverifyTimeout t apns = do
   let tkn = DeviceToken PPApnsTest "abcd"
-  withAgent 1 agentCfg initAgentServers testDB $ \a@AgentClient {agentEnv = Env {store}} -> do
+  withAgent 1 agentCfg initAgentServers testDB $ \a@AgentClient {client = A.AgentClient {agentEnv = Env {store}}} -> do
     (nonce, verification) <- withNtfServer t $ runRight $ do
       NTRegistered <- registerNtfToken a tkn NMPeriodic
       APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}} <-
@@ -394,7 +426,7 @@ testNtfTokenServerRestartReregister t apns = do
 testNtfTokenServerRestartReregisterTimeout :: ASrvTransport -> APNSMockServer -> IO ()
 testNtfTokenServerRestartReregisterTimeout t apns = do
   let tkn = DeviceToken PPApnsTest "abcd"
-  withAgent 1 agentCfg initAgentServers testDB $ \a@AgentClient {agentEnv = Env {store}} -> do
+  withAgent 1 agentCfg initAgentServers testDB $ \a@AgentClient {client = A.AgentClient {agentEnv = Env {store}}} -> do
     withNtfServer t $ runRight $ do
       NTRegistered <- registerNtfToken a tkn NMPeriodic
       APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just _}} <-
@@ -428,7 +460,7 @@ testNtfTokenServerRestartReregisterTimeout t apns = do
 
 getTestNtfTokenPort :: AgentClient -> AE String
 getTestNtfTokenPort a =
-  ExceptT (runExceptT (withStore' a getSavedNtfToken) `runReaderT` agentEnv a) >>= \case
+  ExceptT (runExceptT (withStore' (client a) getSavedNtfToken) `runReaderT` agentEnv (client a)) >>= \case
     Just NtfToken {ntfServer = ProtocolServer {port}} -> pure port
     Nothing -> error "no active NtfToken"
 
@@ -540,10 +572,10 @@ testRunNTFServerTests :: ASrvTransport -> NtfServer -> IO (Maybe ProtocolTestFai
 testRunNTFServerTests t srv =
   withNtfServer t $
     withAgent 1 agentCfg initAgentServers testDB $ \a ->
-      testProtocolServer a NRMInteractive 1 $ ProtoServerWithAuth srv Nothing
+      A.testProtocolServer (client a) NRMInteractive 1 $ ProtoServerWithAuth srv Nothing
 
 testNotificationSubscriptionExistingConnection :: APNSMockServer -> AgentMsgId -> AgentClient -> AgentClient -> IO ()
-testNotificationSubscriptionExistingConnection apns baseId alice@AgentClient {agentEnv = Env {config = aliceCfg, store}} bob = do
+testNotificationSubscriptionExistingConnection apns baseId alice@AgentClient {client = A.AgentClient {agentEnv = Env {config = aliceCfg, store}}} bob = do
   (bobId, aliceId, nonce, message) <- runRight $ do
     -- establish connection
     (bobId, qInfo) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe

tests/AgentTests/ResolveNameTests.hs

@@ -12,7 +12,7 @@
 
 module AgentTests.ResolveNameTests (resolveNameTests) where
 
-import AgentTests.FunctionalAPITests (withAgent)
+import AgentTests.FunctionalAPITests (AgentClient (..), withAgent)
 import Control.Monad.Except (runExceptT)
 import qualified Data.Aeson as J
 import qualified Data.ByteString.Lazy as LB
@@ -24,7 +24,6 @@ import SMPAgentClient
 import SMPClient
 import SMPNamesTests (testNameRecord)
 import Simplex.Messaging.Agent (resolveSimplexName)
-import Simplex.Messaging.Agent.Client (AgentClient)
 import Simplex.Messaging.Agent.Env.SQLite (InitialAgentServers (..), ServerCfg, ServerRoles (..), presetServerCfg)
 import Simplex.Messaging.Agent.Protocol (AgentErrorType (..))
 import Simplex.Messaging.Client (SMPProxyFallback (..), SMPProxyMode (..), pattern NRMInteractive)
@@ -90,7 +89,7 @@ resolveNameTests = do
 testDirectNotFound :: HasCallStack => IO ()
 testDirectNotFound =
   withDirectResolver (status404, "{}") $ \c -> do
-    r <- runExceptT $ resolveSimplexName c NRMInteractive 1 (SimplexDomain TLDSimplex "alice" [])
+    r <- runExceptT $ resolveSimplexName (client c) NRMInteractive 1 (SimplexDomain TLDSimplex "alice" [])
     case r of
       Left (SMP _ (SMP.NAME SMP.NOT_FOUND)) -> pure ()
       _ -> expectationFailure $ "expected Left (SMP _ (NAME NOT_FOUND)), got: " <> show r
@@ -98,7 +97,7 @@ testDirectNotFound =
 testProxyNotFound :: HasCallStack => IO ()
 testProxyNotFound =
   withProxyAndResolver (status404, "{}") $ \c -> do
-    r <- runExceptT $ resolveSimplexName c NRMInteractive 1 (SimplexDomain TLDSimplex "alice" [])
+    r <- runExceptT $ resolveSimplexName (client c) NRMInteractive 1 (SimplexDomain TLDSimplex "alice" [])
     case r of
       Left (SMP host (SMP.NAME SMP.NOT_FOUND)) | testPort `isInfixOf` host -> pure ()
       _ -> expectationFailure $ "expected Left (SMP <proxyHost:" <> testPort <> "> (NAME NOT_FOUND)), got: " <> show r
@@ -106,7 +105,7 @@ testProxyNotFound =
 testTestingTldNotFound :: HasCallStack => IO ()
 testTestingTldNotFound =
   withDirectResolver (status404, "{}") $ \c -> do
-    r <- runExceptT $ resolveSimplexName c NRMInteractive 1 (SimplexDomain TLDTesting "bob" [])
+    r <- runExceptT $ resolveSimplexName (client c) NRMInteractive 1 (SimplexDomain TLDTesting "bob" [])
     case r of
       Left (SMP _ (SMP.NAME SMP.NOT_FOUND)) -> pure ()
       _ -> expectationFailure $ "expected Left (SMP _ (NAME NOT_FOUND)), got: " <> show r
@@ -114,7 +113,7 @@ testTestingTldNotFound =
 testWebTldNotFound :: HasCallStack => IO ()
 testWebTldNotFound =
   withDirectResolver (status404, "{}") $ \c -> do
-    r <- runExceptT $ resolveSimplexName c NRMInteractive 1 (SimplexDomain TLDWeb "example.com" [])
+    r <- runExceptT $ resolveSimplexName (client c) NRMInteractive 1 (SimplexDomain TLDWeb "example.com" [])
     case r of
       Left (SMP _ (SMP.NAME SMP.NOT_FOUND)) -> pure ()
       _ -> expectationFailure $ "expected Left (SMP _ (NAME NOT_FOUND)), got: " <> show r
@@ -122,7 +121,7 @@ testWebTldNotFound =
 testNoResolver :: HasCallStack => IO ()
 testNoResolver =
   withNoResolver $ \c -> do
-    r <- runExceptT $ resolveSimplexName c NRMInteractive 1 (SimplexDomain TLDSimplex "alice" [])
+    r <- runExceptT $ resolveSimplexName (client c) NRMInteractive 1 (SimplexDomain TLDSimplex "alice" [])
     case r of
       Left (SMP _ (SMP.NAME SMP.NO_RESOLVER)) -> pure ()
       _ -> expectationFailure $ "expected Left (SMP _ (NAME NO_RESOLVER)), got: " <> show r
@@ -130,7 +129,7 @@ testNoResolver =
 testNoNameServers :: HasCallStack => IO ()
 testNoNameServers =
   withNoNameServers $ \c -> do
-    r <- runExceptT $ resolveSimplexName c NRMInteractive 1 (SimplexDomain TLDSimplex "alice" [])
+    r <- runExceptT $ resolveSimplexName (client c) NRMInteractive 1 (SimplexDomain TLDSimplex "alice" [])
     case r of
       Left NO_NAME_SERVERS -> pure ()
       _ -> expectationFailure $ "expected Left NO_NAME_SERVERS, got: " <> show r
@@ -138,7 +137,7 @@ testNoNameServers =
 testBackendError :: HasCallStack => IO ()
 testBackendError =
   withDirectResolver (status502, "{}") $ \c -> do
-    r <- runExceptT $ resolveSimplexName c NRMInteractive 1 (SimplexDomain TLDSimplex "alice" [])
+    r <- runExceptT $ resolveSimplexName (client c) NRMInteractive 1 (SimplexDomain TLDSimplex "alice" [])
     case r of
       Left (SMP _ (SMP.NAME (SMP.RESOLVER _))) -> pure ()
       _ -> expectationFailure $ "expected Left (SMP _ (NAME (RESOLVER ..))), got: " <> show r
@@ -146,7 +145,7 @@ testBackendError =
 testDirectSuccess :: HasCallStack => IO ()
 testDirectSuccess =
   withDirectResolver (status200, J.encode testNameRecord) $ \c -> do
-    r <- runExceptT $ resolveSimplexName c NRMInteractive 1 (SimplexDomain TLDSimplex "alice" [])
+    r <- runExceptT $ resolveSimplexName (client c) NRMInteractive 1 (SimplexDomain TLDSimplex "alice" [])
     case r of
       Right nr -> nr `shouldBe` testNameRecord
       _ -> expectationFailure $ "expected Right NameRecord, got: " <> show r

tests/AgentTests/ServerChoice.hs

@@ -74,17 +74,17 @@ testChooseDifferentOperator = do
   c <- getSMPAgentClient' 1 agentCfg initServers testDB
   runRight_ $ do
     -- chooses the only operator with storage role
-    srv1 <- withAgentEnv c $ getNextServer c 1 storageSrvs []
+    srv1 <- withAgentEnv (client c) $ getNextServer (client c) 1 storageSrvs []
     liftIO $ srv1 == testOp1Srv1 || srv1 == testOp1Srv2 `shouldBe` True
     -- chooses another server for storage
-    srv2 <- withAgentEnv c $ getNextServer c 1 storageSrvs [protoServer testOp1Srv1]
+    srv2 <- withAgentEnv (client c) $ getNextServer (client c) 1 storageSrvs [protoServer testOp1Srv1]
     liftIO $ srv2 `shouldBe` testOp1Srv2
     -- chooses another operator for proxy
-    srv3 <- withAgentEnv c $ getNextServer c 1 proxySrvs [protoServer srv1]
+    srv3 <- withAgentEnv (client c) $ getNextServer (client c) 1 proxySrvs [protoServer srv1]
     liftIO $ srv3 == testOp2Srv1 || srv3 == testOp2Srv2 `shouldBe` True
     -- chooses another operator for proxy
-    srv3' <- withAgentEnv c $ getNextServer c 1 proxySrvs [protoServer testOp1Srv1, protoServer testOp1Srv2]
+    srv3' <- withAgentEnv (client c) $ getNextServer (client c) 1 proxySrvs [protoServer testOp1Srv1, protoServer testOp1Srv2]
     liftIO $ srv3' == testOp2Srv1 || srv3' == testOp2Srv2 `shouldBe` True
     -- chooses any other server
-    srv4 <- withAgentEnv c $ getNextServer c 1 proxySrvs [protoServer testOp1Srv1, protoServer testOp2Srv1]
+    srv4 <- withAgentEnv (client c) $ getNextServer (client c) 1 proxySrvs [protoServer testOp1Srv1, protoServer testOp2Srv1]
     liftIO $ srv4 == testOp1Srv2 || srv4 == testOp2Srv2 `shouldBe` True

tests/SMPClient.hs

@@ -275,7 +275,7 @@ cfgMS msType = withStoreCfg (testServerStoreConfig msType) $ \serverStoreCfg ->
       smpServerVRange = supportedServerSMPRelayVRange,
       transportConfig = mkTransportServerConfig True (Just alpnSupportedSMPHandshakes) True,
       controlPort = Nothing,
-      smpAgentCfg = defaultSMPClientAgentConfig {persistErrorInterval = 1, msgQSize = Nothing}, -- seconds
+      smpAgentCfg = defaultSMPClientAgentConfig {persistErrorInterval = 1}, -- seconds
       allowSMPProxy = False,
       serverClientConcurrency = 2,
       serverResolverConcurrency = defaultNameResolverConcurrency,

tests/SMPProxyTests.hs

@@ -27,7 +27,7 @@ import Data.Time.Clock (getCurrentTime)
 import SMPAgentClient
 import SMPClient
 import ServerTests (decryptMsgV3, sendRecv)
-import Simplex.Messaging.Agent hiding (createConnection, joinConnection, sendMessage)
+import Simplex.Messaging.Agent hiding (AgentClient, createConnection, joinConnection, sendMessage, allowConnection, ackMessage)
 import qualified Simplex.Messaging.Agent as A
 import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..), InitialAgentServers (..))
 import Simplex.Messaging.Agent.Protocol hiding (CON, CONF, INFO, REQ)
@@ -180,7 +180,7 @@ deliverMessagesViaProxy proxyServ relayServ alg unsecuredMsgs securedMsgs = do
   THAuthClient {} <- maybe (fail "getProtocolClient returned no thAuth") pure $ thAuth $ thParams pc
   -- set up relay
   msgQ <- newTBQueueIO 1024
-  rc' <- getProtocolClient g NRMInteractive (2, relayServ, Nothing) defaultSMPClientConfig {serverVRange = mkVersionRange minServerSMPRelayVersion currentClientSMPRelayVersion} [] (Just msgQ) ts (\_ -> pure ())
+  rc' <- getProtocolClient g NRMInteractive (2, relayServ, Nothing) defaultSMPClientConfig {serverVRange = mkVersionRange minServerSMPRelayVersion currentClientSMPRelayVersion} [] (Just $ atomically . writeTBQueue msgQ) ts (\_ -> pure ())
   rc <- either (fail . show) pure rc'
   -- prepare receiving queue
   (rPub, rPriv) <- atomically $ C.generateAuthKeyPair alg g
@@ -232,9 +232,9 @@ agentDeliverMessageViaProxy :: (C.AlgorithmI a, C.AuthAlgorithm a) => (NonEmpty
 agentDeliverMessageViaProxy aTestCfg@(aSrvs, _, aViaProxy) bTestCfg@(bSrvs, _, bViaProxy) alg msg1 msg2 baseId =
   withAgent 1 aCfg (servers aTestCfg) testDB $ \alice ->
     withAgent 2 aCfg (servers bTestCfg) testDB2 $ \bob -> runRight_ $ do
-      (bobId, CCLink qInfo Nothing) <- A.createConnection alice NRMInteractive 1 True True SCMInvitation Nothing Nothing CR.IKPQOn SMSubscribe
-      aliceId <- A.prepareConnectionToJoin bob 1 True qInfo PQSupportOn
-      sqSecured <- A.joinConnection bob NRMInteractive 1 aliceId True qInfo "bob's connInfo" PQSupportOn SMSubscribe
+      (bobId, CCLink qInfo Nothing) <- A.createConnection (client alice) NRMInteractive 1 True True SCMInvitation Nothing Nothing CR.IKPQOn SMSubscribe
+      aliceId <- A.prepareConnectionToJoin (client bob) 1 True qInfo PQSupportOn
+      sqSecured <- A.joinConnection (client bob) NRMInteractive 1 aliceId True qInfo "bob's connInfo" PQSupportOn SMSubscribe
       liftIO $ sqSecured `shouldBe` True
       ("", _, A.CONF confId pqSup' _ "bob's connInfo") <- get alice
       liftIO $ pqSup' `shouldBe` PQSupportOn
@@ -245,18 +245,18 @@ agentDeliverMessageViaProxy aTestCfg@(aSrvs, _, aViaProxy) bTestCfg@(bSrvs, _, b
       get bob ##> ("", aliceId, A.CON pqEnc)
       -- message IDs 1 to 3 (or 1 to 4 in v1) get assigned to control messages, so first MSG is assigned ID 4
       let aProxySrv = if aViaProxy then Just $ L.head aSrvs else Nothing
-      1 <- msgId <$> A.sendMessage alice bobId pqEnc noMsgFlags msg1
+      1 <- msgId <$> A.sendMessage (client alice) bobId pqEnc noMsgFlags msg1
       get alice ##> ("", bobId, A.SENT (baseId + 1) aProxySrv)
-      2 <- msgId <$> A.sendMessage alice bobId pqEnc noMsgFlags msg2
+      2 <- msgId <$> A.sendMessage (client alice) bobId pqEnc noMsgFlags msg2
       get alice ##> ("", bobId, A.SENT (baseId + 2) aProxySrv)
       get bob =##> \case ("", c, Msg' _ pq msg1') -> c == aliceId && pq == pqEnc && msg1 == msg1'; _ -> False
       ackMessage bob aliceId (baseId + 1) Nothing
       get bob =##> \case ("", c, Msg' _ pq msg2') -> c == aliceId && pq == pqEnc && msg2 == msg2'; _ -> False
       ackMessage bob aliceId (baseId + 2) Nothing
       let bProxySrv = if bViaProxy then Just $ L.head bSrvs else Nothing
-      3 <- msgId <$> A.sendMessage bob aliceId pqEnc noMsgFlags msg1
+      3 <- msgId <$> A.sendMessage (client bob) aliceId pqEnc noMsgFlags msg1
       get bob ##> ("", aliceId, A.SENT (baseId + 3) bProxySrv)
-      4 <- msgId <$> A.sendMessage bob aliceId pqEnc noMsgFlags msg2
+      4 <- msgId <$> A.sendMessage (client bob) aliceId pqEnc noMsgFlags msg2
       get bob ##> ("", aliceId, A.SENT (baseId + 4) bProxySrv)
       get alice =##> \case ("", c, Msg' _ pq msg1') -> c == bobId && pq == pqEnc && msg1 == msg1'; _ -> False
       ackMessage alice bobId (baseId + 3) Nothing
@@ -288,9 +288,9 @@ agentDeliverMessagesViaProxyConc agentServers msgs =
     -- agent connections have to be set up in advance
     -- otherwise the CONF messages would get mixed with MSG
     prePair alice bob = do
-      (bobId, CCLink qInfo Nothing) <- runExceptT' $ A.createConnection alice NRMInteractive 1 True True SCMInvitation Nothing Nothing CR.IKPQOn SMSubscribe
-      aliceId <- runExceptT' $ A.prepareConnectionToJoin bob 1 True qInfo PQSupportOn
-      sqSecured <- runExceptT' $ A.joinConnection bob NRMInteractive 1 aliceId True qInfo "bob's connInfo" PQSupportOn SMSubscribe
+      (bobId, CCLink qInfo Nothing) <- runExceptT' $ A.createConnection (client alice) NRMInteractive 1 True True SCMInvitation Nothing Nothing CR.IKPQOn SMSubscribe
+      aliceId <- runExceptT' $ A.prepareConnectionToJoin (client bob) 1 True qInfo PQSupportOn
+      sqSecured <- runExceptT' $ A.joinConnection (client bob) NRMInteractive 1 aliceId True qInfo "bob's connInfo" PQSupportOn SMSubscribe
       liftIO $ sqSecured `shouldBe` True
       confId <-
         get alice >>= \case
@@ -305,7 +305,7 @@ agentDeliverMessagesViaProxyConc agentServers msgs =
       pure (alice, bobId, bob, aliceId)
     -- stream messages in opposite directions, while getting deliveries and sending ACKs
     run (alice, bobId, bob, aliceId) = do
-      aSender <- async $ forM_ msgs $ runExceptT' . A.sendMessage alice bobId pqEnc noMsgFlags
+      aSender <- async $ forM_ msgs $ runExceptT' . A.sendMessage (client alice) bobId pqEnc noMsgFlags
       bRecipient <-
         async $
           forever $
@@ -313,7 +313,7 @@ agentDeliverMessagesViaProxyConc agentServers msgs =
               ("", _, A.SENT _ _) -> pure ()
               ("", _, Msg' mId' _ _) -> runExceptT' $ ackMessage alice bobId mId' Nothing
               huh -> fail (show huh)
-      bSender <- async $ forM_ msgs $ runExceptT' . A.sendMessage bob aliceId pqEnc noMsgFlags
+      bSender <- async $ forM_ msgs $ runExceptT' . A.sendMessage (client bob) aliceId pqEnc noMsgFlags
       aRecipient <-
         async $
           forever $
@@ -339,9 +339,9 @@ agentViaProxyVersionError =
   withAgent 1 agentCfg (servers [SMPServer testHost testPort testKeyHash]) testDB $ \alice -> do
     Left (A.BROKER _ (TRANSPORT TEVersion)) <-
       withAgent 2 agentCfg (servers [SMPServer testHost2 testPort2 testKeyHash]) testDB2 $ \bob -> runExceptT $ do
-        (_bobId, CCLink qInfo Nothing) <- A.createConnection alice NRMInteractive 1 True True SCMInvitation Nothing Nothing CR.IKPQOn SMSubscribe
-        aliceId <- A.prepareConnectionToJoin bob 1 True qInfo PQSupportOn
-        A.joinConnection bob NRMInteractive 1 aliceId True qInfo "bob's connInfo" PQSupportOn SMSubscribe
+        (_bobId, CCLink qInfo Nothing) <- A.createConnection (client alice) NRMInteractive 1 True True SCMInvitation Nothing Nothing CR.IKPQOn SMSubscribe
+        aliceId <- A.prepareConnectionToJoin (client bob) 1 True qInfo PQSupportOn
+        A.joinConnection (client bob) NRMInteractive 1 aliceId True qInfo "bob's connInfo" PQSupportOn SMSubscribe
     pure ()
   where
     servers srvs = (initAgentServersProxy_ SPMUnknown SPFProhibit) {smp = userServers srvs}
@@ -359,9 +359,9 @@ agentViaProxyRetryOffline = do
       let pqEnc = CR.PQEncOn
       withServer $ \_ -> do
         (aliceId, bobId) <- withServer2 $ \_ -> runRight $ do
-          (bobId, CCLink qInfo Nothing) <- A.createConnection alice NRMInteractive 1 True True SCMInvitation Nothing Nothing CR.IKPQOn SMSubscribe
-          aliceId <- A.prepareConnectionToJoin bob 1 True qInfo PQSupportOn
-          sqSecured <- A.joinConnection bob NRMInteractive 1 aliceId True qInfo "bob's connInfo" PQSupportOn SMSubscribe
+          (bobId, CCLink qInfo Nothing) <- A.createConnection (client alice) NRMInteractive 1 True True SCMInvitation Nothing Nothing CR.IKPQOn SMSubscribe
+          aliceId <- A.prepareConnectionToJoin (client bob) 1 True qInfo PQSupportOn
+          sqSecured <- A.joinConnection (client bob) NRMInteractive 1 aliceId True qInfo "bob's connInfo" PQSupportOn SMSubscribe
           liftIO $ sqSecured `shouldBe` True
           ("", _, A.CONF confId pqSup' _ "bob's connInfo") <- get alice
           liftIO $ pqSup' `shouldBe` PQSupportOn
@@ -369,18 +369,18 @@ agentViaProxyRetryOffline = do
           get alice ##> ("", bobId, A.CON pqEnc)
           get bob ##> ("", aliceId, A.INFO PQSupportOn "alice's connInfo")
           get bob ##> ("", aliceId, A.CON pqEnc)
-          1 <- msgId <$> A.sendMessage alice bobId pqEnc noMsgFlags msg1
+          1 <- msgId <$> A.sendMessage (client alice) bobId pqEnc noMsgFlags msg1
           get alice ##> ("", bobId, A.SENT (baseId + 1) aProxySrv)
           get bob =##> \case ("", c, Msg' _ pq msg1') -> c == aliceId && pq == pqEnc && msg1 == msg1'; _ -> False
           ackMessage bob aliceId (baseId + 1) Nothing
-          2 <- msgId <$> A.sendMessage bob aliceId pqEnc noMsgFlags msg2
+          2 <- msgId <$> A.sendMessage (client bob) aliceId pqEnc noMsgFlags msg2
           get bob ##> ("", aliceId, A.SENT (baseId + 2) bProxySrv)
           get alice =##> \case ("", c, Msg' _ pq msg2') -> c == bobId && pq == pqEnc && msg2 == msg2'; _ -> False
           ackMessage alice bobId (baseId + 2) Nothing
           pure (aliceId, bobId)
         runRight_ $ do
           -- destination relay down
-          3 <- msgId <$> A.sendMessage alice bobId pqEnc noMsgFlags msg1
+          3 <- msgId <$> A.sendMessage (client alice) bobId pqEnc noMsgFlags msg1
           bob `down` aliceId
         withServer2 $ \_ -> runRight_ $ do
           bob `up` aliceId
@@ -389,7 +389,7 @@ agentViaProxyRetryOffline = do
           ackMessage bob aliceId (baseId + 3) Nothing
         runRight_ $ do
           -- proxy relay down
-          4 <- msgId <$> A.sendMessage bob aliceId pqEnc noMsgFlags msg2
+          4 <- msgId <$> A.sendMessage (client bob) aliceId pqEnc noMsgFlags msg2
           bob `down` aliceId
         withServer2 $ \_ -> do
           getInAnyOrder
@@ -520,14 +520,14 @@ testAgentClientReconnectAfterCancel :: IO ()
 testAgentClientReconnectAfterCancel =
   withAgent 1 agentCfg agentServersLeak testDB $ \a -> do
     withStallingServerOn testPort2 $ do
-      t <- async $ runExceptT $ A.createConnection a NRMInteractive 1 True True SCMInvitation Nothing Nothing CR.IKPQOn SMSubscribe
+      t <- async $ runExceptT $ A.createConnection (client a) NRMInteractive 1 True True SCMInvitation Nothing Nothing CR.IKPQOn SMSubscribe
       threadDelay 1000000 -- let the connect to the stalling relay start, then kill it mid-flight
       cancel t
     withSmpServerConfigOn (transport @TLS) cfgJ2 testPort2 $ \_ -> do
       testSMPClient_ "127.0.0.1" testPort2 proxyVRangeV8 Nothing $ \(th :: THandleSMP TLS 'TClient) -> do
         (_, _, reply) <- sendRecv th (Nothing, "0", NoEntity, SMP.PING)
         reply `shouldBe` Right SMP.PONG -- the relay is up and reachable, so a timeout can only be the poisoned var
-      r <- timeout 8000000 $ runExceptT $ A.createConnection a NRMInteractive 1 True True SCMInvitation Nothing Nothing CR.IKPQOn SMSubscribe
+      r <- timeout 8000000 $ runExceptT $ A.createConnection (client a) NRMInteractive 1 True True SCMInvitation Nothing Nothing CR.IKPQOn SMSubscribe
       case r of
         Just (Right _) -> pure ()
         _ -> expectationFailure $ "agent failed to connect after a cancelled connect; got: " <> show r

tests/XFTPAgent.hs

@@ -10,7 +10,7 @@
 
 module XFTPAgent where
 
-import AgentTests.FunctionalAPITests (get, rfGet, runRight, runRight_, sfGet, withAgent)
+import AgentTests.FunctionalAPITests (AgentClient (..), get, rfGet, runRight, runRight_, sfGet, withAgent)
 
 import Control.Logger.Simple
 import Control.Monad
@@ -30,7 +30,7 @@ import Simplex.FileTransfer.Server.Env (AFStoreType, XFTPServerConfig (..))
 import Simplex.FileTransfer.Server.Store (STMFileStore)
 import Simplex.FileTransfer.Transport (XFTPErrorType (AUTH))
 import Simplex.FileTransfer.Types (RcvFileId, SndFileId)
-import Simplex.Messaging.Agent (AgentClient, testProtocolServer, xftpDeleteRcvFile, xftpDeleteSndFileInternal, xftpDeleteSndFileRemote, xftpReceiveFile, xftpSendDescription, xftpSendFile, xftpStartWorkers)
+import qualified Simplex.Messaging.Agent as A
 import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestStep (..))
 import Simplex.Messaging.Agent.Env.SQLite (AgentConfig, xftpCfg)
 import Simplex.Messaging.Agent.Protocol (AEvent (..), AgentErrorType (..), BrokerErrorType (..), noAuthSrv)
@@ -99,7 +99,7 @@ testXFTPServerTest newFileBasicAuth srv =
   withXFTPServerCfg testXFTPServerConfig {newFileBasicAuth, xftpPort = xftpTestPort2} $ \_ ->
     -- initially passed server is not running
     withAgent 1 agentCfg initAgentServers testDB $ \a ->
-      testProtocolServer a NRMInteractive 1 srv
+      A.testProtocolServer (client a) NRMInteractive 1 srv
 
 rfProgress :: forall m. (HasCallStack, MonadIO m, MonadFail m) => AgentClient -> Int64 -> m ()
 rfProgress c expected = loop 0
@@ -132,7 +132,7 @@ testXFTPAgentSendReceive = do
   -- send file, delete snd file internally
   (rfd1, rfd2) <- withAgent 1 agentCfg initAgentServers testDB $ \sndr -> runRight $ do
     (sfId, _, rfd1, rfd2) <- testSend sndr filePath
-    liftIO $ xftpDeleteSndFileInternal sndr sfId
+    liftIO $ A.xftpDeleteSndFileInternal (client sndr) sfId
     pure (rfd1, rfd2)
   -- receive file, delete rcv file
   testReceiveDelete 2 rfd1 filePath
@@ -141,7 +141,7 @@ testXFTPAgentSendReceive = do
     testReceiveDelete clientId rfd originalFilePath =
       withAgent clientId agentCfg initAgentServers testDB2 $ \rcp -> do
         rfId <- runRight $ testReceive rcp rfd originalFilePath
-        xftpDeleteRcvFile rcp rfId
+        A.xftpDeleteRcvFile (client rcp) rfId
 
 testXFTPAgentSendReceiveEncrypted :: HasCallStack => AFStoreType -> IO ()
 testXFTPAgentSendReceiveEncrypted = withXFTPServer $ do
@@ -152,7 +152,7 @@ testXFTPAgentSendReceiveEncrypted = withXFTPServer $ do
   runRight_ $ CF.writeFile file s
   (rfd1, rfd2) <- withAgent 1 agentCfg initAgentServers testDB $ \sndr -> runRight $ do
     (sfId, _, rfd1, rfd2) <- testSendCF sndr file
-    liftIO $ xftpDeleteSndFileInternal sndr sfId
+    liftIO $ A.xftpDeleteSndFileInternal (client sndr) sfId
     pure (rfd1, rfd2)
   -- receive file, delete rcv file
   testReceiveDelete 2 rfd1 filePath g
@@ -162,7 +162,7 @@ testXFTPAgentSendReceiveEncrypted = withXFTPServer $ do
       withAgent clientId agentCfg initAgentServers testDB2 $ \rcp -> do
         cfArgs <- atomically $ Just <$> CF.randomArgs g
         rfId <- runRight $ testReceiveCF rcp rfd cfArgs originalFilePath
-        xftpDeleteRcvFile rcp rfId
+        A.xftpDeleteRcvFile (client rcp) rfId
 
 testXFTPAgentSendReceiveRedirect :: HasCallStack => AFStoreType -> IO ()
 testXFTPAgentSendReceiveRedirect = withXFTPServer $ do
@@ -171,7 +171,7 @@ testXFTPAgentSendReceiveRedirect = withXFTPServer $ do
   let fileSize = mb 17
       totalSize = fileSize + mb 1
   withAgent 1 agentCfg initAgentServers testDB $ \sndr -> do
-    directFileId <- runRight $ xftpSendFile sndr 1 (CryptoFile filePathIn Nothing) 1
+    directFileId <- runRight $ A.xftpSendFile (client sndr) 1 (CryptoFile filePathIn Nothing) 1
     sfGet sndr `shouldReturn` ("", directFileId, SFPROG 4194304 totalSize)
     sfGet sndr `shouldReturn` ("", directFileId, SFPROG 8388608 totalSize)
     sfGet sndr `shouldReturn` ("", directFileId, SFPROG 12582912 totalSize)
@@ -185,7 +185,7 @@ testXFTPAgentSendReceiveRedirect = withXFTPServer $ do
 
     testNoRedundancy vfdDirect
 
-    redirectFileId <- runRight $ xftpSendDescription sndr 1 vfdDirect 1
+    redirectFileId <- runRight $ A.xftpSendDescription (client sndr) 1 vfdDirect 1
     logInfo $ "File sent, sending redirect: " <> tshow redirectFileId
     sfGet sndr `shouldReturn` ("", redirectFileId, SFPROG 65536 65536)
     vfdRedirect@(ValidFileDescription fdRedirect) <-
@@ -206,7 +206,7 @@ testXFTPAgentSendReceiveRedirect = withXFTPServer $ do
     withAgent 2 agentCfg initAgentServers testDB2 $ \rcp -> do
       FileDescriptionURI {description} <- either fail pure $ strDecode uri
 
-      rcvFileId <- runRight $ xftpReceiveFile rcp 1 description Nothing True
+      rcvFileId <- runRight $ A.xftpReceiveFile (client rcp) 1 description Nothing True
       rfGet rcp `shouldReturn` ("", rcvFileId, RFPROG 65536 totalSize) -- extra RFPROG before switching to real file
       rfGet rcp `shouldReturn` ("", rcvFileId, RFPROG 4194304 totalSize)
       rfGet rcp `shouldReturn` ("", rcvFileId, RFPROG 8388608 totalSize)
@@ -228,7 +228,7 @@ testXFTPAgentSendReceiveNoRedirect = withXFTPServer $ do
   let fileSize = mb 5
   filePathIn <- createRandomFile_ fileSize "testfile"
   withAgent 1 agentCfg initAgentServers testDB $ \sndr -> do
-    directFileId <- runRight $ xftpSendFile sndr 1 (CryptoFile filePathIn Nothing) 1
+    directFileId <- runRight $ A.xftpSendFile (client sndr) 1 (CryptoFile filePathIn Nothing) 1
     let totalSize = fileSize + mb 1
     sfGet sndr `shouldReturn` ("", directFileId, SFPROG 4194304 totalSize)
     sfGet sndr `shouldReturn` ("", directFileId, SFPROG 5242880 totalSize)
@@ -250,7 +250,7 @@ testXFTPAgentSendReceiveNoRedirect = withXFTPServer $ do
       FileDescriptionURI {description} <- either fail pure $ strDecode uri
       let ValidFileDescription FileDescription {redirect} = description
       redirect `shouldBe` Nothing
-      rcvFileId <- runRight $ xftpReceiveFile rcp 1 description Nothing True
+      rcvFileId <- runRight $ A.xftpReceiveFile (client rcp) 1 description Nothing True
       -- NO extra "RFPROG 65k 65k" before switching to real file
       rfGet rcp `shouldReturn` ("", rcvFileId, RFPROG 4194304 totalSize)
       rfGet rcp `shouldReturn` ("", rcvFileId, RFPROG 5242880 totalSize)
@@ -286,10 +286,10 @@ testXFTPAgentSendReceiveMatrix = do
         filePath <- createRandomFile_ (kb 319 :: Integer) "testfile"
         rfd <- withAgent 1 sender initAgentServers testDB $ \sndr -> do
           (sfId, _, rfd1, _) <- runRight $ testSendCF' sndr (CF.plain filePath) (kb 320)
-          rfd1 <$ xftpDeleteSndFileInternal sndr sfId
+          rfd1 <$ A.xftpDeleteSndFileInternal (client sndr) sfId
         withAgent 2 receiver initAgentServers testDB2 $ \rcp -> do
           rfId <- runRight $ testReceiveCF' rcp rfd Nothing filePath (kb 320)
-          xftpDeleteRcvFile rcp rfId
+          A.xftpDeleteRcvFile (client rcp) rfId
 
 createRandomFile :: HasCallStack => IO FilePath
 createRandomFile = createRandomFile' "testfile"
@@ -312,8 +312,8 @@ testSendCF sndr file = testSendCF' sndr file $ mb 18
 
 testSendCF' :: HasCallStack => AgentClient -> CryptoFile -> Int64 -> ExceptT AgentErrorType IO (SndFileId, ValidFileDescription 'FSender, ValidFileDescription 'FRecipient, ValidFileDescription 'FRecipient)
 testSendCF' sndr file size = do
-  xftpStartWorkers sndr (Just senderFiles)
-  sfId <- xftpSendFile sndr 1 file 2
+  A.xftpStartWorkers (client sndr) (Just senderFiles)
+  sfId <- A.xftpSendFile (client sndr) 1 file 2
   sfProgress sndr size
   ("", sfId', SFDONE sndDescr [rfd1, rfd2]) <- sfGet sndr
   liftIO $ testNoRedundancy rfd1
@@ -330,7 +330,7 @@ testReceive rcp rfd = testReceiveCF rcp rfd Nothing
 
 testReceiveCF :: HasCallStack => AgentClient -> ValidFileDescription 'FRecipient -> Maybe CryptoFileArgs -> FilePath -> ExceptT AgentErrorType IO RcvFileId
 testReceiveCF rcp rfd cfArgs originalFilePath = do
-  xftpStartWorkers rcp (Just recipientFiles)
+  A.xftpStartWorkers (client rcp) (Just recipientFiles)
   testReceiveCF' rcp rfd cfArgs originalFilePath $ mb 18
 
 testReceive' :: HasCallStack => AgentClient -> ValidFileDescription 'FRecipient -> FilePath -> ExceptT AgentErrorType IO RcvFileId
@@ -338,7 +338,7 @@ testReceive' rcp rfd originalFilePath = testReceiveCF' rcp rfd Nothing originalF
 
 testReceiveCF' :: HasCallStack => AgentClient -> ValidFileDescription 'FRecipient -> Maybe CryptoFileArgs -> FilePath -> Int64 -> ExceptT AgentErrorType IO RcvFileId
 testReceiveCF' rcp rfd cfArgs originalFilePath size = do
-  rfId <- xftpReceiveFile rcp 1 rfd cfArgs True
+  rfId <- A.xftpReceiveFile (client rcp) 1 rfd cfArgs True
   rfProgress rcp size
   ("", rfId', RFDONE path) <- rfGet rcp
   liftIO $ do
@@ -362,8 +362,8 @@ testXFTPAgentReceiveRestore = do
 
   -- receive file - should not succeed with server down
   rfId <- withAgent 2 agentCfg initAgentServers testDB2 $ \rcp -> runRight $ do
-    xftpStartWorkers rcp (Just recipientFiles)
-    rfId <- xftpReceiveFile rcp 1 rfd Nothing True
+    A.xftpStartWorkers (client rcp) (Just recipientFiles)
+    rfId <- A.xftpReceiveFile (client rcp) 1 rfd Nothing True
     liftIO $ timeout 300000 (get rcp) `shouldReturn` Nothing -- wait for worker attempt
     pure rfId
 
@@ -374,7 +374,7 @@ testXFTPAgentReceiveRestore = do
   withXFTPServerStoreLogOn $ \_ ->
     -- receive file - should start downloading with server up
     withAgent 3 agentCfg initAgentServers testDB2 $ \rcp' -> do
-      runRight_ $ xftpStartWorkers rcp' (Just recipientFiles)
+      runRight_ $ A.xftpStartWorkers (client rcp') (Just recipientFiles)
       ("", rfId', RFPROG _ _) <- rfGet rcp'
       liftIO $ rfId' `shouldBe` rfId
   threadDelay 100000
@@ -382,7 +382,7 @@ testXFTPAgentReceiveRestore = do
   withXFTPServerStoreLogOn $ \_ ->
     -- receive file - should continue downloading with server up
     withAgent 4 agentCfg initAgentServers testDB2 $ \rcp' -> do
-      runRight_ $ xftpStartWorkers rcp' (Just recipientFiles)
+      runRight_ $ A.xftpStartWorkers (client rcp') (Just recipientFiles)
       rfProgress rcp' $ mb 18
       ("", rfId', RFDONE path) <- rfGet rcp'
       liftIO $ do
@@ -406,8 +406,8 @@ testXFTPAgentReceiveCleanup = withGlobalLogging logCfgNoLogs $ do
 
   -- receive file - should not succeed with server down
   rfId <- withAgent 2 agentCfg initAgentServers testDB2 $ \rcp -> runRight $ do
-    xftpStartWorkers rcp (Just recipientFiles)
-    rfId <- xftpReceiveFile rcp 1 rfd Nothing True
+    A.xftpStartWorkers (client rcp) (Just recipientFiles)
+    rfId <- A.xftpReceiveFile (client rcp) 1 rfd Nothing True
     liftIO $ timeout 300000 (get rcp) `shouldReturn` Nothing -- wait for worker attempt
     pure rfId
 
@@ -418,7 +418,7 @@ testXFTPAgentReceiveCleanup = withGlobalLogging logCfgNoLogs $ do
   withXFTPServerThreadOn $ \_ ->
     -- receive file - should fail with AUTH error
     withAgent 3 agentCfg initAgentServers testDB2 $ \rcp' -> do
-      runRight_ $ xftpStartWorkers rcp' (Just recipientFiles)
+      runRight_ $ A.xftpStartWorkers (client rcp') (Just recipientFiles)
       ("", rfId', RFERR (XFTP "xftp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:8000" AUTH)) <- rfGet rcp'
       rfId' `shouldBe` rfId
 
@@ -431,8 +431,8 @@ testXFTPAgentSendRestore = withGlobalLogging logCfgNoLogs $ do
 
   -- send file - should not succeed with server down
   sfId <- withAgent 1 agentCfg initAgentServers testDB $ \sndr -> runRight $ do
-    xftpStartWorkers sndr (Just senderFiles)
-    sfId <- xftpSendFile sndr 1 (CF.plain filePath) 2
+    A.xftpStartWorkers (client sndr) (Just senderFiles)
+    sfId <- A.xftpSendFile (client sndr) 1 (CF.plain filePath) 2
     liftIO $ timeout 1000000 (get sndr) `shouldReturn` Nothing -- wait for worker to encrypt and attempt to create file
     pure sfId
 
@@ -446,7 +446,7 @@ testXFTPAgentSendRestore = withGlobalLogging logCfgNoLogs $ do
   withXFTPServerStoreLogOn $ \_ ->
     -- send file - should start uploading with server up
     withAgent 2 agentCfg initAgentServers testDB $ \sndr' -> do
-      runRight_ $ xftpStartWorkers sndr' (Just senderFiles)
+      runRight_ $ A.xftpStartWorkers (client sndr') (Just senderFiles)
       ("", sfId', SFPROG _ _) <- sfGet sndr'
       liftIO $ sfId' `shouldBe` sfId
 
@@ -455,7 +455,7 @@ testXFTPAgentSendRestore = withGlobalLogging logCfgNoLogs $ do
   withXFTPServerStoreLogOn $ \_ -> do
     -- send file - should continue uploading with server up
     rfd1 <- withAgent 3 agentCfg initAgentServers testDB $ \sndr' -> do
-      runRight_ $ xftpStartWorkers sndr' (Just senderFiles)
+      runRight_ $ A.xftpStartWorkers (client sndr') (Just senderFiles)
       sfProgress sndr' $ mb 18
       ("", sfId', SFDONE _sndDescr [rfd1, rfd2]) <- sfGet sndr'
       liftIO $ testNoRedundancy rfd1
@@ -479,8 +479,8 @@ testXFTPAgentSendCleanup = withGlobalLogging logCfgNoLogs $ do
   sfId <- withXFTPServerStoreLogOn $ \_ ->
     -- send file
     withAgent 1 agentCfg initAgentServers testDB $ \sndr -> runRight $ do
-      xftpStartWorkers sndr (Just senderFiles)
-      sfId <- xftpSendFile sndr 1 (CF.plain filePath) 2
+      A.xftpStartWorkers (client sndr) (Just senderFiles)
+      sfId <- A.xftpSendFile (client sndr) 1 (CF.plain filePath) 2
       -- wait for progress events for 5 out of 6 chunks - at this point all chunks should be created on the server
       forM_ [1 .. 5 :: Integer] $ \_ -> do
         (_, _, SFPROG _ _) <- sfGet sndr
@@ -497,7 +497,7 @@ testXFTPAgentSendCleanup = withGlobalLogging logCfgNoLogs $ do
   withXFTPServerThreadOn $ \_ ->
     -- send file - should fail with AUTH error
     withAgent 2 agentCfg initAgentServers testDB $ \sndr' -> do
-      runRight_ $ xftpStartWorkers sndr' (Just senderFiles)
+      runRight_ $ A.xftpStartWorkers (client sndr') (Just senderFiles)
       ("", sfId', SFERR (XFTP "xftp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:8000" AUTH)) <-
         sfGet sndr'
       sfId' `shouldBe` sfId
@@ -523,8 +523,8 @@ testXFTPAgentDelete = withGlobalLogging logCfgNoLogs . withXFTPServer test
           length <$> listDirectory xftpServerFiles `shouldReturn` 6
 
           -- delete file
-          runRight_ $ xftpStartWorkers sndr (Just senderFiles)
-          xftpDeleteSndFileRemote sndr 1 sfId sndDescr
+          runRight_ $ A.xftpStartWorkers (client sndr) (Just senderFiles)
+          A.xftpDeleteSndFileRemote (client sndr) 1 sfId sndDescr
           Nothing <- 100000 `timeout` sfGet sndr
           pure ()
 
@@ -533,8 +533,8 @@ testXFTPAgentDelete = withGlobalLogging logCfgNoLogs . withXFTPServer test
 
         -- receive file - should fail with AUTH error
         withAgent 3 agentCfg initAgentServers testDB2 $ \rcp2 -> runRight $ do
-          xftpStartWorkers rcp2 (Just recipientFiles)
-          rfId <- xftpReceiveFile rcp2 1 rfd2 Nothing True
+          A.xftpStartWorkers (client rcp2) (Just recipientFiles)
+          rfId <- A.xftpReceiveFile (client rcp2) 1 rfd2 Nothing True
           ("", rfId', RFERR (XFTP "xftp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:8000" AUTH)) <-
             rfGet rcp2
           liftIO $ rfId' `shouldBe` rfId
@@ -555,8 +555,8 @@ testXFTPAgentDeleteRestore = withGlobalLogging logCfgNoLogs $ do
 
   -- delete file - should not succeed with server down
   withAgent 3 agentCfg initAgentServers testDB $ \sndr -> do
-    runRight_ $ xftpStartWorkers sndr (Just senderFiles)
-    xftpDeleteSndFileRemote sndr 1 sfId sndDescr
+    runRight_ $ A.xftpStartWorkers (client sndr) (Just senderFiles)
+    A.xftpDeleteSndFileRemote (client sndr) 1 sfId sndDescr
     timeout 300000 (get sndr) `shouldReturn` Nothing -- wait for worker attempt
   threadDelay 300000
   length <$> listDirectory xftpServerFiles `shouldReturn` 6
@@ -564,15 +564,15 @@ testXFTPAgentDeleteRestore = withGlobalLogging logCfgNoLogs $ do
   withXFTPServerStoreLogOn $ \_ -> do
     -- delete file - should succeed with server up
     withAgent 4 agentCfg initAgentServers testDB $ \sndr' -> do
-      runRight_ $ xftpStartWorkers sndr' (Just senderFiles)
+      runRight_ $ A.xftpStartWorkers (client sndr') (Just senderFiles)
 
       threadDelay 1000000
       length <$> listDirectory xftpServerFiles `shouldReturn` 0
 
       -- receive file - should fail with AUTH error
       withAgent 5 agentCfg initAgentServers testDB3 $ \rcp2 -> runRight $ do
-        xftpStartWorkers rcp2 (Just recipientFiles)
-        rfId <- xftpReceiveFile rcp2 1 rfd2 Nothing True
+        A.xftpStartWorkers (client rcp2) (Just recipientFiles)
+        rfId <- A.xftpReceiveFile (client rcp2) 1 rfd2 Nothing True
         ("", rfId', RFERR (XFTP "xftp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:8000" AUTH)) <-
           rfGet rcp2
         liftIO $ rfId' `shouldBe` rfId
@@ -608,7 +608,7 @@ testXFTPAgentDeleteOnServer = withGlobalLogging logCfgNoLogs . withXFTPServer te
 
           runRight_ . void $ do
             -- receive file 1 again
-            rfId1 <- xftpReceiveFile rcp 1 rfd1_2 Nothing True
+            rfId1 <- A.xftpReceiveFile (client rcp) 1 rfd1_2 Nothing True
             ("", rfId1', RFERR (XFTP "xftp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:8000" AUTH)) <-
               rfGet rcp
             liftIO $ rfId1 `shouldBe` rfId1'
@@ -640,7 +640,7 @@ testXFTPAgentExpiredOnServer fsType = withGlobalLogging logCfgNoLogs $
 
         -- receive file 1 again - should fail with AUTH error
         runRight $ do
-          rfId <- xftpReceiveFile rcp 1 rfd1_2 Nothing True
+          rfId <- A.xftpReceiveFile (client rcp) 1 rfd1_2 Nothing True
           ("", rfId', RFERR (XFTP "xftp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:8000" AUTH)) <-
             rfGet rcp
           liftIO $ rfId' `shouldBe` rfId
@@ -662,8 +662,8 @@ testXFTPAgentRequestAdditionalRecipientIDs = withXFTPServer $ do
 
   -- send file
   rfds <- withAgent 1 agentCfg initAgentServers testDB $ \sndr -> runRight $ do
-    xftpStartWorkers sndr (Just senderFiles)
-    sfId <- xftpSendFile sndr 1 (CF.plain filePath) 500
+    A.xftpStartWorkers (client sndr) (Just senderFiles)
+    sfId <- A.xftpSendFile (client sndr) 1 (CF.plain filePath) 500
     sfProgress sndr $ mb 18
     ("", sfId', SFDONE _sndDescr rfds) <- sfGet sndr
     liftIO $ do
@@ -685,4 +685,4 @@ testXFTPServerTest_ :: HasCallStack => XFTPServerWithAuth -> IO (Maybe ProtocolT
 testXFTPServerTest_ srv =
   -- initially passed server is not running
   withAgent 1 agentCfg initAgentServers testDB $ \a ->
-    testProtocolServer a NRMInteractive 1 srv
+    A.testProtocolServer (client a) NRMInteractive 1 srv

tests/XFTPWebTests.hs

@@ -47,8 +47,8 @@ import Util
 import Simplex.FileTransfer.Server.Env (XFTPServerConfig)
 import Simplex.FileTransfer.Server.Store (STMFileStore)
 import XFTPClient (testXFTPServerConfigEd25519SNI, testXFTPServerConfigSNI, withXFTPServerCfg, xftpTestPort)
-import AgentTests.FunctionalAPITests (rfGet, runRight, runRight_, sfGet, withAgent)
-import Simplex.Messaging.Agent (AgentClient, xftpReceiveFile, xftpSendFile, xftpStartWorkers)
+import AgentTests.FunctionalAPITests (AgentClient (..), rfGet, runRight, runRight_, sfGet, withAgent)
+import qualified Simplex.Messaging.Agent as A
 import Simplex.Messaging.Agent.Protocol (AEvent (..))
 import SMPAgentClient (agentCfg, initAgentServers, testDB)
 import XFTPCLI (recipientFiles, senderFiles, testBracket)
@@ -3143,8 +3143,8 @@ tsUploadHaskellDownloadTest cfg caFile = do
           <> jsOut2 "Buffer.from(yaml)" "Buffer.from(originalData)"
     let vfd :: ValidFileDescription 'FRecipient = either error id $ strDecode yamlDesc
     withAgent 1 agentCfg initAgentServers testDB $ \rcp -> do
-      runRight_ $ xftpStartWorkers rcp (Just recipientFiles)
-      _ <- runRight $ xftpReceiveFile rcp 1 vfd Nothing True
+      runRight_ $ A.xftpStartWorkers (client rcp) (Just recipientFiles)
+      _ <- runRight $ A.xftpReceiveFile (client rcp) 1 vfd Nothing True
       rfProgress rcp 50000
       (_, _, RFDONE outPath) <- rfGet rcp
       downloadedData <- B.readFile outPath
@@ -3179,8 +3179,8 @@ tsUploadRedirectHaskellDownloadTest cfg caFile = do
     let vfd@(ValidFileDescription fd) :: ValidFileDescription 'FRecipient = either error id $ strDecode yamlDesc
     redirect fd `shouldSatisfy` (/= Nothing)
     withAgent 1 agentCfg initAgentServers testDB $ \rcp -> do
-      runRight_ $ xftpStartWorkers rcp (Just recipientFiles)
-      _ <- runRight $ xftpReceiveFile rcp 1 vfd Nothing True
+      runRight_ $ A.xftpStartWorkers (client rcp) (Just recipientFiles)
+      _ <- runRight $ A.xftpReceiveFile (client rcp) 1 vfd Nothing True
       outPath <- waitRfDone rcp
       downloadedData <- B.readFile outPath
       downloadedData `shouldBe` originalData
@@ -3194,8 +3194,8 @@ haskellUploadTsDownloadTest cfg = do
   B.writeFile filePath originalData
   withXFTPServerCfg cfg $ \_ -> do
     vfd <- withAgent 1 agentCfg initAgentServers testDB $ \sndr -> do
-      runRight_ $ xftpStartWorkers sndr (Just senderFiles)
-      _ <- runRight $ xftpSendFile sndr 1 (CF.plain filePath) 1
+      runRight_ $ A.xftpStartWorkers (client sndr) (Just senderFiles)
+      _ <- runRight $ A.xftpSendFile (client sndr) 1 (CF.plain filePath) 1
       sfProgress sndr 50000
       (_, _, SFDONE _ [rfd]) <- sfGet sndr
       pure rfd