show client sub status

remove from client subs when removing conn
remove print
2026-10-12 02:22:45 +00:00 · 2024-05-06 20:52:54 +03:00 · 2024-05-06 20:21:19 +03:00 · 2024-05-06 19:33:54 +03:00 · 2024-05-06 18:40:28 +03:00 · 2024-05-06 13:20:42 +03:00
22 changed files with 482 additions and 299 deletions
@@ -1,3 +1,10 @@
+# 5.7.2
+
+SMP agent:
+- fix connections failing when connecting via link due to race condition on slow network.
+- remove concurrency limit when waiting for connection subscription.
+- remove TLS timeout.
+
 # 5.7.1

 SMP agent:
@@ -1,5 +1,5 @@
 name: simplexmq
-version: 5.7.1.0
+version: 5.7.2.0
 synopsis: SimpleXMQ message broker
 description: |
  This package includes <./docs/Simplex-Messaging-Server.html server>,
@@ -180,6 +180,7 @@ ghc-options:
  - -Wall
  - -Wcompat
  - -Werror=incomplete-patterns
+  - -Werror=missing-methods
  - -Wredundant-constraints
  - -Wincomplete-record-updates
  - -Wincomplete-uni-patterns
@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack

 name:           simplexmq
-version:        5.7.1.0
+version:        5.7.2.0
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                <./docs/Simplex-Messaging-Client.html client> and
@@ -174,7 +174,7 @@ library
      src
  default-extensions:
      StrictData
-  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2
+  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Werror=missing-methods -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2
  include-dirs:
      cbits
  c-sources:
@@ -255,7 +255,7 @@ executable ntf-server
      apps/ntf-server
  default-extensions:
      StrictData
-  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
+  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Werror=missing-methods -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
  build-depends:
      aeson ==2.2.*
    , ansi-terminal >=0.10 && <0.12
@@ -330,7 +330,7 @@ executable smp-agent
      apps/smp-agent
  default-extensions:
      StrictData
-  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
+  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Werror=missing-methods -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
  build-depends:
      aeson ==2.2.*
    , ansi-terminal >=0.10 && <0.12
@@ -405,7 +405,7 @@ executable smp-server
      apps/smp-server
  default-extensions:
      StrictData
-  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
+  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Werror=missing-methods -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
  build-depends:
      aeson ==2.2.*
    , ansi-terminal >=0.10 && <0.12
@@ -480,7 +480,7 @@ executable xftp
      apps/xftp
  default-extensions:
      StrictData
-  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
+  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Werror=missing-methods -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
  build-depends:
      aeson ==2.2.*
    , ansi-terminal >=0.10 && <0.12
@@ -555,7 +555,7 @@ executable xftp-server
      apps/xftp-server
  default-extensions:
      StrictData
-  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
+  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Werror=missing-methods -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
  build-depends:
      aeson ==2.2.*
    , ansi-terminal >=0.10 && <0.12
@@ -662,7 +662,7 @@ test-suite simplexmq-test
      tests
  default-extensions:
      StrictData
-  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts -with-rtsopts=-A64M -with-rtsopts=-N1
+  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Werror=missing-methods -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts -with-rtsopts=-A64M -with-rtsopts=-N1
  build-depends:
      HUnit ==1.6.*
    , QuickCheck ==2.14.*
@@ -262,9 +262,9 @@ runXFTPRcvLocalWorker c Worker {doWork} = do
      withStore' c $ \db -> updateRcvFileStatus db rcvFileId RFSDecrypting
      chunkPaths <- getChunkPaths chunks
      encSize <- liftIO $ foldM (\s path -> (s +) . fromIntegral <$> getFileSize path) 0 chunkPaths
-      when (FileSize encSize /= size) $ throwError $ XFTP XFTP.SIZE
+      when (FileSize encSize /= size) $ throwError $ XFTP "" XFTP.SIZE
      encDigest <- liftIO $ LC.sha512Hash <$> readChunks chunkPaths
-      when (FileDigest encDigest /= digest) $ throwError $ XFTP XFTP.DIGEST
+      when (FileDigest encDigest /= digest) $ throwError $ XFTP "" XFTP.DIGEST
      let destFile = CryptoFile fsSavePath cfArgs
      void $ liftError (INTERNAL . show) $ decryptChunks encSize chunkPaths key nonce $ \_ -> pure destFile
      case redirect of
@@ -281,10 +281,11 @@ runXFTPRcvLocalWorker c Worker {doWork} = do
          -- proceed with redirect
          yaml <- liftError (INTERNAL . show) (CF.readFile $ CryptoFile fsSavePath cfArgs) `agentFinally` (lift $ toFSFilePath fsSavePath >>= removePath)
          next@FileDescription {chunks = nextChunks} <- case strDecode (LB.toStrict yaml) of
-            Left _ -> throwError . XFTP $ XFTP.REDIRECT "decode error"
+            -- TODO switch to another error constructor
+            Left _ -> throwError . XFTP "" $ XFTP.REDIRECT "decode error"
            Right (ValidFileDescription fd@FileDescription {size = dstSize, digest = dstDigest})
-              | dstSize /= redirectSize -> throwError . XFTP $ XFTP.REDIRECT "size mismatch"
-              | dstDigest /= redirectDigest -> throwError . XFTP $ XFTP.REDIRECT "digest mismatch"
+              | dstSize /= redirectSize -> throwError . XFTP "" $ XFTP.REDIRECT "size mismatch"
+              | dstDigest /= redirectDigest -> throwError . XFTP "" $ XFTP.REDIRECT "digest mismatch"
              | otherwise -> pure fd
          -- register and download chunks from the actual file
          withStore c $ \db -> updateRcvFileRedirect db redirectDbId next
@@ -143,6 +143,7 @@ instance Protocol XFTPVersion XFTPErrorType FileResponse where
  type ProtoType FileResponse = 'PXFTP
  protocolClientHandshake = xftpClientHandshakeStub
  protocolPing = FileCmd SFRecipient PING
+  protocolSub _ = False
  protocolError = \case
    FRErr e -> Just e
    _ -> Nothing
@@ -55,6 +55,7 @@ module Simplex.Messaging.Agent
    deleteConnectionAsync,
    deleteConnectionsAsync,
    createConnection,
+    prepareConnectionToJoin,
    joinConnection,
    allowConnection,
    acceptContact,
@@ -117,7 +118,7 @@ module Simplex.Messaging.Agent
  )
 where

-import Control.Logger.Simple (logError, logInfo, showText)
+import Control.Logger.Simple
 import Control.Monad
 import Control.Monad.Except
 import Control.Monad.Reader
@@ -157,7 +158,7 @@ import Simplex.Messaging.Agent.Store
 import Simplex.Messaging.Agent.Store.SQLite
 import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
 import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
-import Simplex.Messaging.Client (ProtocolClient (..), ServerTransmission)
+import Simplex.Messaging.Client (ProtocolClient (..), ServerTransmission, TransmissionType (..))
 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)
@@ -288,9 +289,18 @@ createConnection :: AgentClient -> UserId -> Bool -> SConnectionMode c -> Maybe
 createConnection c userId enableNtfs = withAgentEnv c .:: newConn c userId "" enableNtfs
 {-# INLINE createConnection #-}

-- | Join SMP agent connection (JOIN command)
-joinConnection :: AgentClient -> UserId -> Bool -> ConnectionRequestUri c -> ConnInfo -> PQSupport -> SubscriptionMode -> AE ConnId
-joinConnection c userId enableNtfs = withAgentEnv c .:: joinConn c userId "" enableNtfs
+-- | Create SMP agent connection without queue (to be joined with joinConnection passing connection ID).
+-- This method is required to prevent race condition when confirmation from peer is received before
+-- the caller of joinConnection saves connection ID to the database.
+-- Instead of it we could send confirmation asynchronously, but then it would be harder to report
+-- "link deleted" (SMP AUTH) interactively, so this approach is simpler overall.
+prepareConnectionToJoin :: AgentClient -> UserId -> Bool -> ConnectionRequestUri c -> PQSupport -> AE ConnId
+prepareConnectionToJoin c userId enableNtfs = withAgentEnv c .: newConnToJoin c userId "" enableNtfs
+
+-- | Join SMP agent connection (JOIN command).
+joinConnection :: AgentClient -> UserId -> Maybe ConnId -> Bool -> ConnectionRequestUri c -> ConnInfo -> PQSupport -> SubscriptionMode -> AE ConnId
+joinConnection c userId Nothing enableNtfs = withAgentEnv c .:: joinConn c userId "" False enableNtfs
+joinConnection c userId (Just connId) enableNtfs = withAgentEnv c .:: joinConn c userId connId True enableNtfs
 {-# INLINE joinConnection #-}

 -- | Allow connection to continue after CONF notification (LET command)
@@ -575,7 +585,7 @@ processCommand :: AgentClient -> (EntityId, APartyCmd 'Client) -> AM (EntityId,
 processCommand c (connId, APC e cmd) =
  second (APC e) <$> case cmd of
    NEW enableNtfs (ACM cMode) pqIK subMode -> second (INV . ACR cMode) <$> newConn c userId connId enableNtfs cMode Nothing pqIK subMode
-    JOIN enableNtfs (ACR _ cReq) pqEnc subMode connInfo -> (,OK) <$> joinConn c userId connId enableNtfs cReq connInfo pqEnc subMode
+    JOIN enableNtfs (ACR _ cReq) pqEnc subMode connInfo -> (,OK) <$> joinConn c userId connId False enableNtfs cReq connInfo pqEnc subMode
    LET confId ownCInfo -> allowConnection' c connId confId ownCInfo $> (connId, OK)
    ACPT invId pqEnc ownCInfo -> (,OK) <$> acceptContact' c connId True invId ownCInfo pqEnc SMSubscribe
    RJCT invId -> rejectContact' c connId invId $> (connId, OK)
@@ -708,11 +718,14 @@ switchConnectionAsync' c corrId connId =

 newConn :: AgentClient -> UserId -> ConnId -> Bool -> SConnectionMode c -> Maybe CRClientData -> CR.InitialKeys -> SubscriptionMode -> AM (ConnId, ConnectionRequestUri c)
 newConn c userId connId enableNtfs cMode clientData pqInitKeys subMode =
-  getSMPServer c userId >>= newConnSrv c userId connId enableNtfs cMode clientData pqInitKeys subMode
+  getSMPServer c userId >>= newConnSrv c userId connId False enableNtfs cMode clientData pqInitKeys subMode

-newConnSrv :: AgentClient -> UserId -> ConnId -> Bool -> SConnectionMode c -> Maybe CRClientData -> CR.InitialKeys -> SubscriptionMode -> SMPServerWithAuth -> AM (ConnId, ConnectionRequestUri c)
-newConnSrv c userId connId enableNtfs cMode clientData pqInitKeys subMode srv = do
-  connId' <- newConnNoQueues c userId connId enableNtfs cMode (CR.connPQEncryption pqInitKeys)
+newConnSrv :: AgentClient -> UserId -> ConnId -> Bool -> Bool -> SConnectionMode c -> Maybe CRClientData -> CR.InitialKeys -> SubscriptionMode -> SMPServerWithAuth -> AM (ConnId, ConnectionRequestUri c)
+newConnSrv c userId connId hasNewConn enableNtfs cMode clientData pqInitKeys subMode srv = do
+  connId' <-
+    if hasNewConn
+      then pure connId
+      else newConnNoQueues c userId connId enableNtfs cMode (CR.connPQEncryption pqInitKeys)
  newRcvConnSrv c userId connId' enableNtfs cMode clientData pqInitKeys subMode srv

 newRcvConnSrv :: AgentClient -> UserId -> ConnId -> Bool -> SConnectionMode c -> Maybe CRClientData -> CR.InitialKeys -> SubscriptionMode -> SMPServerWithAuth -> AM (ConnId, ConnectionRequestUri c)
@@ -736,18 +749,36 @@ newRcvConnSrv c userId connId enableNtfs cMode clientData pqInitKeys subMode srv
      withStore' c $ \db -> createRatchetX3dhKeys db connId pk1 pk2 pKem
      pure (connId, CRInvitationUri crData $ toVersionRangeT e2eRcvParams e2eEncryptVRange)

-joinConn :: AgentClient -> UserId -> ConnId -> Bool -> ConnectionRequestUri c -> ConnInfo -> PQSupport -> SubscriptionMode -> AM ConnId
-joinConn c userId connId enableNtfs cReq cInfo pqSupport subMode = do
+newConnToJoin :: forall c. AgentClient -> UserId -> ConnId -> Bool -> ConnectionRequestUri c -> PQSupport -> AM ConnId
+newConnToJoin c userId connId enableNtfs cReq pqSup = case cReq of
+  CRInvitationUri {} ->
+    lift (compatibleInvitationUri cReq) >>= \case
+      Just (_, (Compatible (CR.E2ERatchetParams v _ _ _)), aVersion) -> create aVersion (Just v)
+      Nothing -> throwError $ AGENT A_VERSION
+  CRContactUri {} ->
+    lift (compatibleContactUri cReq) >>= \case
+      Just (_, aVersion) -> create aVersion Nothing
+      Nothing -> throwError $ AGENT A_VERSION
+  where
+    create :: Compatible VersionSMPA -> Maybe CR.VersionE2E -> AM ConnId
+    create (Compatible connAgentVersion) e2eV_ = do
+      g <- asks random
+      let pqSupport = pqSup `CR.pqSupportAnd` versionPQSupport_ connAgentVersion e2eV_
+          cData = ConnData {userId, connId, connAgentVersion, enableNtfs, lastExternalSndId = 0, deleted = False, ratchetSyncState = RSOk, pqSupport}
+      withStore c $ \db -> createNewConn db g cData SCMInvitation
+
+joinConn :: AgentClient -> UserId -> ConnId -> Bool -> Bool -> ConnectionRequestUri c -> ConnInfo -> PQSupport -> SubscriptionMode -> AM ConnId
+joinConn c userId connId hasNewConn enableNtfs cReq cInfo pqSupport subMode = do
  srv <- case cReq of
    CRInvitationUri ConnReqUriData {crSmpQueues = q :| _} _ ->
      getNextServer c userId [qServer q]
    _ -> getSMPServer c userId
-  joinConnSrv c userId connId enableNtfs cReq cInfo pqSupport subMode srv
+  joinConnSrv c userId connId hasNewConn enableNtfs cReq cInfo pqSupport subMode srv

-startJoinInvitation :: UserId -> ConnId -> Bool -> ConnectionRequestUri 'CMInvitation -> PQSupport -> AM (Compatible VersionSMPA, ConnData, NewSndQueue, CR.Ratchet 'C.X448, CR.SndE2ERatchetParams 'C.X448)
+startJoinInvitation :: UserId -> ConnId -> Bool -> ConnectionRequestUri 'CMInvitation -> PQSupport -> AM (ConnData, NewSndQueue, CR.Ratchet 'C.X448, CR.SndE2ERatchetParams 'C.X448)
 startJoinInvitation userId connId enableNtfs cReqUri pqSup =
  lift (compatibleInvitationUri cReqUri) >>= \case
-    Just (qInfo, (Compatible e2eRcvParams@(CR.E2ERatchetParams v _ rcDHRr kem_)), aVersion@(Compatible connAgentVersion)) -> do
+    Just (qInfo, (Compatible e2eRcvParams@(CR.E2ERatchetParams v _ rcDHRr kem_)), Compatible connAgentVersion) -> do
      g <- asks random
      let pqSupport = pqSup `CR.pqSupportAnd` versionPQSupport_ connAgentVersion (Just v)
      (pk1, pk2, pKem, e2eSndParams) <- liftIO $ CR.generateSndE2EParams g v (CR.replyKEM_ v kem_ pqSupport)
@@ -758,7 +789,7 @@ startJoinInvitation userId connId enableNtfs cReqUri pqSup =
          rc = CR.initSndRatchet rcVs rcDHRr rcDHRs rcParams
      q <- lift $ newSndQueue userId "" qInfo
      let cData = ConnData {userId, connId, connAgentVersion, enableNtfs, lastExternalSndId = 0, deleted = False, ratchetSyncState = RSOk, pqSupport}
-      pure (aVersion, cData, q, rc, e2eSndParams)
+      pure (cData, q, rc, e2eSndParams)
    Nothing -> throwError $ AGENT A_VERSION

 connRequestPQSupport :: AgentClient -> PQSupport -> ConnectionRequestUri c -> IO (Maybe (VersionSMPA, PQSupport))
@@ -791,33 +822,36 @@ versionPQSupport_ :: VersionSMPA -> Maybe CR.VersionE2E -> PQSupport
 versionPQSupport_ agentV e2eV_ = PQSupport $ agentV >= pqdrSMPAgentVersion && maybe True (>= CR.pqRatchetE2EEncryptVersion) e2eV_
 {-# INLINE versionPQSupport_ #-}

-joinConnSrv :: AgentClient -> UserId -> ConnId -> Bool -> ConnectionRequestUri c -> ConnInfo -> PQSupport -> SubscriptionMode -> SMPServerWithAuth -> AM ConnId
-joinConnSrv c userId connId enableNtfs inv@CRInvitationUri {} cInfo pqSup subMode srv =
+joinConnSrv :: AgentClient -> UserId -> ConnId -> Bool -> Bool -> ConnectionRequestUri c -> ConnInfo -> PQSupport -> SubscriptionMode -> SMPServerWithAuth -> AM ConnId
+joinConnSrv c userId connId hasNewConn enableNtfs inv@CRInvitationUri {} cInfo pqSup subMode srv =
  withInvLock c (strEncode inv) "joinConnSrv" $ do
-    (aVersion, cData, q, rc, e2eSndParams) <- startJoinInvitation userId connId enableNtfs inv pqSup
+    (cData, q, rc, e2eSndParams) <- startJoinInvitation userId connId enableNtfs inv pqSup
    g <- asks random
    (connId', sq) <- withStore c $ \db -> runExceptT $ do
-      r@(connId', _) <- ExceptT $ createSndConn db g cData q
+      r@(connId', _) <-
+        if hasNewConn
+          then (connId,) <$> ExceptT (updateNewConnSnd db connId q)
+          else ExceptT $ createSndConn db g cData q
      liftIO $ createRatchet db connId' rc
      pure r
    let cData' = (cData :: ConnData) {connId = connId'}
-    tryError (confirmQueue aVersion c cData' sq srv cInfo (Just e2eSndParams) subMode) >>= \case
+    tryError (confirmQueue c cData' sq srv cInfo (Just e2eSndParams) subMode) >>= \case
      Right _ -> pure connId'
      Left e -> do
        -- possible improvement: recovery for failure on network timeout, see rfcs/2022-04-20-smp-conf-timeout-recovery.md
        void $ withStore' c $ \db -> deleteConn db Nothing connId'
        throwError e
-joinConnSrv c userId connId enableNtfs cReqUri@CRContactUri {} cInfo pqSup subMode srv =
+joinConnSrv c userId connId hasNewConn enableNtfs cReqUri@CRContactUri {} cInfo pqSup subMode srv =
  lift (compatibleContactUri cReqUri) >>= \case
    Just (qInfo, vrsn) -> do
-      (connId', cReq) <- newConnSrv c userId connId enableNtfs SCMInvitation Nothing (CR.IKNoPQ pqSup) subMode srv
+      (connId', cReq) <- newConnSrv c userId connId hasNewConn enableNtfs SCMInvitation Nothing (CR.IKNoPQ pqSup) subMode srv
      sendInvitation c userId qInfo vrsn cReq cInfo
      pure connId'
    Nothing -> throwError $ AGENT A_VERSION

 joinConnSrvAsync :: AgentClient -> UserId -> ConnId -> Bool -> ConnectionRequestUri c -> ConnInfo -> PQSupport -> SubscriptionMode -> SMPServerWithAuth -> AM ()
 joinConnSrvAsync c userId connId enableNtfs inv@CRInvitationUri {} cInfo pqSupport subMode srv = do
-  (_aVersion, cData, q, rc, e2eSndParams) <- startJoinInvitation userId connId enableNtfs inv pqSupport
+  (cData, q, rc, e2eSndParams) <- startJoinInvitation userId connId enableNtfs inv pqSupport
  q' <- withStore c $ \db -> runExceptT $ do
    liftIO $ createRatchet db connId rc
    ExceptT $ updateNewConnSnd db connId q
@@ -857,7 +891,7 @@ acceptContact' c connId enableNtfs invId ownConnInfo pqSupport subMode = withCon
  withStore c (`getConn` contactConnId) >>= \case
    SomeConn _ (ContactConnection ConnData {userId} _) -> do
      withStore' c $ \db -> acceptInvitation db invId ownConnInfo
-      joinConn c userId connId enableNtfs connReq ownConnInfo pqSupport subMode `catchAgentError` \err -> do
+      joinConn c userId connId False enableNtfs connReq ownConnInfo pqSupport subMode `catchAgentError` \err -> do
        withStore' c (`unacceptInvitation` invId)
        throwError err
    _ -> throwError $ CMD PROHIBITED
@@ -890,7 +924,7 @@ subscribeConnections' c connIds = do
      (subRs, rcvQs) = M.mapEither rcvQueueOrResult cs
  mapM_ (mapM_ (\(cData, sqs) -> mapM_ (lift . resumeMsgDelivery c cData) sqs) . sndQueue) cs
  mapM_ (resumeConnCmds c) $ M.keys cs
-  rcvRs <- lift $ connResults <$> subscribeQueues c (concat $ M.elems rcvQs)
+  rcvRs <- lift $ connResults . fst <$> subscribeQueues c (concat $ M.elems rcvQs)
  ns <- asks ntfSupervisor
  tkn <- readTVarIO (ntfTkn ns)
  when (instantNotifications tkn) . void . lift . forkIO . void . runExceptT $ sendNtfCreate ns rcvRs conns
@@ -1288,13 +1322,13 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} ConnData {connId} sq (Worker {doWork
            Left e -> do
              let err = if msgType == AM_A_MSG_ then MERR mId e else ERR e
              case e of
-                SMP SMP.QUOTA -> case msgType of
+                SMP _ SMP.QUOTA -> case msgType of
                  AM_CONN_INFO -> connError msgId NOT_AVAILABLE
                  AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE
                  _ -> do
                    expireTs <- addUTCTime (-quotaExceededTimeout) <$> liftIO getCurrentTime
                    if internalTs < expireTs then notifyDelMsgs msgId e expireTs else retrySndMsg RISlow
-                SMP SMP.AUTH -> case msgType of
+                SMP _ SMP.AUTH -> case msgType of
                  AM_CONN_INFO -> connError msgId NOT_AVAILABLE
                  AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE
                  AM_RATCHET_INFO -> connError msgId NOT_AVAILABLE
@@ -1527,7 +1561,7 @@ synchronizeRatchet' c connId pqSupport' force = withConnLock c connId "synchroni
 ackQueueMessage :: AgentClient -> RcvQueue -> SMP.MsgId -> AM ()
 ackQueueMessage c rq srvMsgId =
  sendAck c rq srvMsgId `catchAgentError` \case
-    SMP SMP.NO_MSG -> pure ()
+    SMP _ SMP.NO_MSG -> pure ()
    e -> throwError e

 -- | Suspend SMP agent connection (OFF command) in Reader monad
@@ -1589,6 +1623,12 @@ prepareDeleteConnections_ getConnections c waitDelivery connIds = do
      rqs = concat $ M.elems rcvQs
      connIds' = M.keys rcvQs
  lift . forM_ connIds' $ disableConn c
+  -- remove from subs tracking in clients
+  clients <- readTVarIO (smpClients c)
+  forM_ clients $ \SessionVar {sessionVar} ->
+    atomically $ tryReadTMVar sessionVar >>= \case
+      Just (Right smp) -> modifyTVar' (sentSubs smp) . M.filterWithKey $ \k _v -> k `M.notMember` rcvQs
+      _ -> pure ()
  -- ! delRs is not used to notify about the result in any of the calling functions,
  -- ! it is only used to check results count in deleteConnections_;
  -- ! if it was used to notify about the result, it might be necessary to differentiate
@@ -1857,7 +1897,7 @@ deleteToken_ c tkn@NtfToken {ntfTokenId, ntfTknStatus} = do
    withStore' c $ \db -> updateNtfToken db tkn ntfTknStatus ntfTknAction
    atomically $ nsUpdateToken ns tkn {ntfTknStatus, ntfTknAction}
    agentNtfDeleteToken c tknId tkn `catchAgentError` \case
-      NTF AUTH -> pure ()
+      NTF _ AUTH -> pure ()
      e -> throwError e
  withStore' c $ \db -> removeNtfToken db tkn
  atomically $ nsRemoveNtfToken ns
@@ -1874,7 +1914,7 @@ withToken c tkn@NtfToken {deviceToken, ntfMode} from_ (toStatus, toAction_) f =
      let updatedToken = tkn {ntfTknStatus = toStatus, ntfTknAction = toAction_}
      atomically $ nsUpdateToken ns updatedToken
      pure toStatus
-    Left e@(NTF AUTH) -> do
+    Left e@(NTF _ AUTH) -> do
      withStore' c $ \db -> removeNtfToken db tkn
      atomically $ nsRemoveNtfToken ns
      void $ registerNtfToken' c deviceToken ntfMode
@@ -1957,11 +1997,13 @@ getSMPServer c userId = withUserServers c userId pickServer
 {-# INLINE getSMPServer #-}

 subscriber :: AgentClient -> AM' ()
-subscriber c@AgentClient {msgQ} = forever $ do
+subscriber c@AgentClient {subQ, msgQ} = forever $ do
  t <- atomically $ readTBQueue msgQ
  agentOperationBracket c AORcvNetwork waitUntilActive $
    runExceptT (processSMPTransmission c t) >>= \case
-      Left e -> liftIO $ print e
+      Left e -> do
+        logError $ tshow e
+        atomically $ writeTBQueue subQ ("", "", APC SAEConn $ ERR e)
      Right _ -> return ()

 cleanupManager :: AgentClient -> AM' ()
@@ -2038,8 +2080,10 @@ data ACKd = ACKd | ACKPending

 -- | make sure to ACK or throw in each message processing branch
 -- it cannot be finally, unfortunately, as sometimes it needs to be ACK+DEL
-processSMPTransmission :: AgentClient -> ServerTransmission SMPVersion BrokerMsg -> AM ()
-processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v, sessId, isResponse, rId, cmd) = do
+processSMPTransmission :: AgentClient -> ServerTransmission SMPVersion ErrorType BrokerMsg -> AM ()
+processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v, sessId, tType, rId, cmd) = do
+  active <- atomically $ activeClientSession c tSess sessId
+  unless active $ logNote $ "Inactive client for " <> tshow (strEncode srv)
  (rq, SomeConn _ conn) <- withStore c (\db -> getRcvConn db srv rId)
  processSMP rq conn $ toConnData conn
  where
@@ -2049,7 +2093,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v,
      conn
      cData@ConnData {userId, connId, connAgentVersion, ratchetSyncState = rss} =
        withConnLock c connId "processSMP" $ case cmd of
-          SMP.MSG msg@SMP.RcvMessage {msgId = srvMsgId} ->
+          Right (SMP.MSG msg@SMP.RcvMessage {msgId = srvMsgId}) ->
            void . handleNotifyAck $ do
              isGET <- atomically $ hasGetLock c rq
              unless isGET checkSubscription
@@ -2200,7 +2244,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v,
              ackDel aId = enqueueCmd (ICAckDel rId srvMsgId aId) $> ACKd
              handleNotifyAck :: AM ACKd -> AM ACKd
              handleNotifyAck m = m `catchAgentError` \e -> notify (ERR e) >> ack
-          SMP.END ->
+          Right SMP.END ->
            atomically (TM.lookup tSess smpClients $>>= (tryReadTMVar . sessionVar) >>= processEND)
              >>= logServer "<--" c srv rId
            where
@@ -2208,15 +2252,16 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v,
                Just (Right clnt)
                  | sessId == sessionId (thParams clnt) -> do
                      removeSubscription c connId
+                      modifyTVar (sentSubs clnt) $ M.delete connId
                      notify' END
                      pure "END"
                  | otherwise -> ignored
                _ -> ignored
              ignored = pure "END from disconnected client - ignored"
-          SMP.SUBOK -> checkSubscription
-          _ -> do
-            logServer "<--" c srv rId $ "unexpected: " <> bshow cmd
-            notify . ERR $ BROKER (B.unpack $ strEncode srv) $ if isResponse then TIMEOUT else UNEXPECTED
+          Right (SMP.ERR e) -> notify $ ERR $ SMP (B.unpack $ strEncode srv) e
+          Right SMP.OK -> checkSubscription
+          Right _ -> unexpected
+          Left e -> notify $ ERR $ protocolClientError SMP (B.unpack $ strEncode srv) e
        where
          notify :: forall e m. MonadIO m => AEntityI e => ACommand 'Agent e -> m ()
          notify = atomically . notify'
@@ -2230,15 +2275,24 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v,
          enqueueCmd :: InternalCommand -> AM ()
          enqueueCmd = enqueueCommand c "" connId (Just srv) . AInternalCommand

+          unexpected :: AM ()
+          unexpected = do
+            logServer "<--" c srv rId $ "unexpected: " <> bshow cmd
+            -- TODO add extended information about transmission type once UNEXPECTED has string
+            notify . ERR $ BROKER (B.unpack $ strEncode srv) UNEXPECTED
+
          checkSubscription :: AM ()
-          checkSubscription = do
-            added <-
-              atomically $
-                ifM
-                  ((&&) <$> hasPendingSubscription c connId <*> activeClientSession c tSess sessId)
-                  (True <$ addSubscription c rq)
-                  (pure False)
-            when added . notify $ UP srv [connId]
+          checkSubscription = case tType of
+            TTEvent -> pure ()
+            TTUncorrelatedResponse -> unexpected
+            TTExpiredSubResponse -> do
+              added <-
+                atomically $
+                  ifM
+                    ((&&) <$> hasPendingSubscription c connId <*> activeClientSession c tSess sessId)
+                    (True <$ addSubscription c rq)
+                    (pure False)
+              when added $ notify $ UP srv [connId]

          decryptClientMessage :: C.DhSecretX25519 -> SMP.ClientMsgEnvelope -> AM (SMP.PrivHeader, AgentMsgEnvelope)
          decryptClientMessage e2eDh SMP.ClientMsgEnvelope {cmNonce, cmEncBody} = do
@@ -2573,8 +2627,8 @@ confirmQueueAsync c cData sq srv connInfo e2eEncryption_ subMode = do
  storeConfirmation c cData sq e2eEncryption_ =<< mkAgentConfirmation c cData sq srv connInfo subMode
  lift $ submitPendingMsg c cData sq

-confirmQueue :: Compatible VersionSMPA -> AgentClient -> ConnData -> SndQueue -> SMPServerWithAuth -> ConnInfo -> Maybe (CR.SndE2ERatchetParams 'C.X448) -> SubscriptionMode -> AM ()
-confirmQueue (Compatible agentVersion) c cData@ConnData {connId, pqSupport} sq srv connInfo e2eEncryption_ subMode = do
+confirmQueue :: AgentClient -> ConnData -> SndQueue -> SMPServerWithAuth -> ConnInfo -> Maybe (CR.SndE2ERatchetParams 'C.X448) -> SubscriptionMode -> AM ()
+confirmQueue c cData@ConnData {connId, connAgentVersion, pqSupport} sq srv connInfo e2eEncryption_ subMode = do
  msg <- mkConfirmation =<< mkAgentConfirmation c cData sq srv connInfo subMode
  sendConfirmation c sq msg
  withStore' c $ \db -> setSndQueueStatus db sq Confirmed
@@ -2586,7 +2640,7 @@ confirmQueue (Compatible agentVersion) c cData@ConnData {connId, pqSupport} sq s
        void . liftIO $ updateSndIds db connId
        let pqEnc = CR.pqSupportToEnc pqSupport
        (encConnInfo, _) <- agentRatchetEncrypt db cData (smpEncode aMessage) e2eEncConnInfoLength (Just pqEnc) currentE2EVersion
-        pure . smpEncode $ AgentConfirmation {agentVersion, e2eEncryption_, encConnInfo}
+        pure . smpEncode $ AgentConfirmation {agentVersion = connAgentVersion, e2eEncryption_, encConnInfo}

 mkAgentConfirmation :: AgentClient -> ConnData -> SndQueue -> SMPServerWithAuth -> ConnInfo -> SubscriptionMode -> AM AgentMessage
 mkAgentConfirmation c cData sq srv connInfo subMode = do
@@ -84,11 +84,14 @@ module Simplex.Messaging.Agent.Client
    agentClientStore,
    agentDRG,
    getAgentSubscriptions,
+    diffSubscriptions,
    slowNetworkConfig,
+    protocolClientError,
    Worker (..),
    SessionVar (..),
    SubscriptionsInfo (..),
    SubInfo (..),
+    SubscriptionsDiff (..),
    AgentOperation (..),
    AgentOpState (..),
    AgentState (..),
@@ -155,7 +158,8 @@ import Data.Bifunctor (bimap, first, second)
 import Data.ByteString.Base64
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
-import Data.Either (lefts, partitionEithers)
+import Data.Either (partitionEithers)
+import Data.Foldable (fold)
 import Data.Functor (($>))
 import Data.Int (Int64)
 import Data.List (deleteFirstsBy, foldl', partition, (\\))
@@ -171,6 +175,7 @@ import Data.Text.Encoding
 import Data.Time (UTCTime, defaultTimeLocale, diffUTCTime, formatTime, getCurrentTime)
 import Data.Time.Clock.System (getSystemTime)
 import Data.Word (Word16)
+import GHC.Stack (HasCallStack, withFrozenCallStack)
 import Network.Socket (HostName)
 import Simplex.FileTransfer.Client (XFTPChunkSpec (..), XFTPClient, XFTPClientConfig (..), XFTPClientError)
 import qualified Simplex.FileTransfer.Client as X
@@ -263,7 +268,7 @@ data AgentClient = AgentClient
    active :: TVar Bool,
    rcvQ :: TBQueue (ATransmission 'Client),
    subQ :: TBQueue (ATransmission 'Agent),
-    msgQ :: TBQueue (ServerTransmission SMPVersion BrokerMsg),
+    msgQ :: TBQueue (ServerTransmission SMPVersion ErrorType BrokerMsg),
    smpServers :: TMap UserId (NonEmpty SMPServerWithAuth),
    smpClients :: TMap SMPTransportSession SMPClientVar,
    ntfServers :: TVar [NtfServer],
@@ -514,7 +519,7 @@ agentDRG AgentClient {agentEnv = Env {random}} = random
 class (Encoding err, Show err) => ProtocolServerClient v err msg | msg -> v, msg -> err where
  type Client msg = c | c -> msg
  getProtocolServerClient :: AgentClient -> TransportSession msg -> AM (Client msg)
-  clientProtocolError :: err -> AgentErrorType
+  clientProtocolError :: HostName -> err -> AgentErrorType
  closeProtocolServerClient :: Client msg -> IO ()
  clientServer :: Client msg -> String
  clientTransportHost :: Client msg -> TransportHost
@@ -643,7 +648,7 @@ reconnectSMPClient tc c tSess@(_, srv, _) qs = do
    resubscribe :: AM ()
    resubscribe = do
      cs <- readTVarIO $ RQ.getConnections $ activeSubs c
-      rs <- lift . subscribeQueues c $ L.toList qs
+      (rs, sessId_) <- lift . subscribeQueues c $ L.toList qs
      let (errs, okConns) = partitionEithers $ map (\(RcvQueue {connId}, r) -> bimap (connId,) (const connId) r) rs
      liftIO $ do
        let conns = filter (`M.notMember` cs) okConns
@@ -652,7 +657,10 @@ reconnectSMPClient tc c tSess@(_, srv, _) qs = do
      liftIO $ mapM_ (\(connId, e) -> notifySub connId $ ERR e) finalErrs
      forM_ (listToMaybe tempErrs) $ \(_, err) -> do
        when (null okConns && M.null cs && null finalErrs) . liftIO $
-          closeClient c smpClients tSess
+          forM_ sessId_ $ \sessId -> do
+            -- We only close the client session that was used to subscribe.
+            v_ <- atomically $ ifM (activeClientSession c tSess sessId) (TM.lookupDelete tSess $ smpClients c) (pure Nothing)
+            mapM_ (closeClient_ c) v_
        throwError err
    notifySub :: forall e. AEntityI e => ConnId -> ACommand 'Agent e -> IO ()
    notifySub connId cmd = atomically $ writeTBQueue (subQ c) ("", connId, APC (sAEntity @e) cmd)
@@ -886,42 +894,42 @@ getMapLock locks key = TM.lookup key locks >>= maybe newLock pure
  where
    newLock = createLock >>= \l -> TM.insert key l locks $> l

-withClient_ :: forall a v err msg. ProtocolServerClient v err msg => AgentClient -> TransportSession msg -> ByteString -> (Client msg -> AM a) -> AM a
+withClient_ :: forall a v err msg. (HasCallStack, ProtocolServerClient v err msg) => AgentClient -> TransportSession msg -> ByteString -> (Client msg -> AM a) -> AM a
 withClient_ c tSess@(userId, srv, _) statCmd action = do
  cl <- getProtocolServerClient c tSess
-  (action cl <* stat cl "OK") `catchAgentError` logServerError cl
+  (action cl <* stat cl "OK") `catchAgentError` \e -> withFrozenCallStack $ logServerError cl e
  where
    stat cl = liftIO . incClientStat c userId cl statCmd
-    logServerError :: Client msg -> AgentErrorType -> AM a
+    logServerError :: HasCallStack => Client msg -> AgentErrorType -> AM a
    logServerError cl e = do
-      logServer "<--" c srv "" $ strEncode e
+      withFrozenCallStack $ logServer "<--" c srv "" $ strEncode e
      stat cl $ strEncode e
      throwError e

-withLogClient_ :: ProtocolServerClient v err msg => AgentClient -> TransportSession msg -> EntityId -> ByteString -> (Client msg -> AM a) -> AM a
+withLogClient_ :: (HasCallStack, ProtocolServerClient v err msg) => AgentClient -> TransportSession msg -> EntityId -> ByteString -> (Client msg -> AM a) -> AM a
 withLogClient_ c tSess@(_, srv, _) entId cmdStr action = do
-  logServer "-->" c srv entId cmdStr
+  withFrozenCallStack $ logServer "-->" c srv entId cmdStr
  res <- withClient_ c tSess cmdStr action
-  logServer "<--" c srv entId "OK"
+  withFrozenCallStack $ logServer "<--" c srv entId "OK"
  return res

-withClient :: forall v err msg a. ProtocolServerClient v err msg => AgentClient -> TransportSession msg -> ByteString -> (Client msg -> ExceptT (ProtocolClientError err) IO a) -> AM a
-withClient c tSess statKey action = withClient_ c tSess statKey $ \client -> liftClient (clientProtocolError @v @err @msg) (clientServer client) $ action client
+withClient :: forall v err msg a. (HasCallStack, ProtocolServerClient v err msg) => AgentClient -> TransportSession msg -> ByteString -> (Client msg -> ExceptT (ProtocolClientError err) IO a) -> AM a
+withClient c tSess statKey action = withFrozenCallStack $ withClient_ c tSess statKey $ \client -> liftClient (clientProtocolError @v @err @msg) (clientServer client) $ action client
 {-# INLINE withClient #-}

-withLogClient :: forall v err msg a. ProtocolServerClient v err msg => AgentClient -> TransportSession msg -> EntityId -> ByteString -> (Client msg -> ExceptT (ProtocolClientError err) IO a) -> AM a
-withLogClient c tSess entId cmdStr action = withLogClient_ c tSess entId cmdStr $ \client -> liftClient (clientProtocolError @v @err @msg) (clientServer client) $ action client
+withLogClient :: forall v err msg a. (HasCallStack, ProtocolServerClient v err msg) => AgentClient -> TransportSession msg -> EntityId -> ByteString -> (Client msg -> ExceptT (ProtocolClientError err) IO a) -> AM a
+withLogClient c tSess entId cmdStr action = withFrozenCallStack $ withLogClient_ c tSess entId cmdStr $ \client -> liftClient (clientProtocolError @v @err @msg) (clientServer client) $ action client
 {-# INLINE withLogClient #-}

-withSMPClient :: SMPQueueRec q => AgentClient -> q -> ByteString -> (SMPClient -> ExceptT SMPClientError IO a) -> AM a
+withSMPClient :: (HasCallStack, SMPQueueRec q) => AgentClient -> q -> ByteString -> (SMPClient -> ExceptT SMPClientError IO a) -> AM a
 withSMPClient c q cmdStr action = do
  tSess <- liftIO $ mkSMPTransportSession c q
-  withLogClient c tSess (queueId q) cmdStr action
+  withFrozenCallStack $ withLogClient c tSess (queueId q) cmdStr action

-withSMPClient_ :: SMPQueueRec q => AgentClient -> q -> ByteString -> (SMPClient -> AM a) -> AM a
+withSMPClient_ :: (HasCallStack, SMPQueueRec q) => AgentClient -> q -> ByteString -> (SMPClient -> AM a) -> AM a
 withSMPClient_ c q cmdStr action = do
  tSess <- liftIO $ mkSMPTransportSession c q
-  withLogClient_ c tSess (queueId q) cmdStr action
+  withFrozenCallStack $ withLogClient_ c tSess (queueId q) cmdStr action

 withNtfClient :: AgentClient -> NtfServer -> EntityId -> ByteString -> (NtfClient -> ExceptT NtfClientError IO a) -> AM a
 withNtfClient c srv = withLogClient c (0, srv, Nothing)
@@ -937,13 +945,13 @@ withXFTPClient c (userId, srv, entityId) cmdStr action = do
  tSess <- liftIO $ mkTransportSession c userId srv entityId
  withLogClient c tSess entityId cmdStr action

-liftClient :: (Show err, Encoding err) => (err -> AgentErrorType) -> HostName -> ExceptT (ProtocolClientError err) IO a -> AM a
+liftClient :: (Show err, Encoding err) => (HostName -> err -> AgentErrorType) -> HostName -> ExceptT (ProtocolClientError err) IO a -> AM a
 liftClient protocolError_ = liftError . protocolClientError protocolError_
 {-# INLINE liftClient #-}

-protocolClientError :: (Show err, Encoding err) => (err -> AgentErrorType) -> HostName -> ProtocolClientError err -> AgentErrorType
+protocolClientError :: (Show err, Encoding err) => (HostName -> err -> AgentErrorType) -> HostName -> ProtocolClientError err -> AgentErrorType
 protocolClientError protocolError_ host = \case
-  PCEProtocolError e -> protocolError_ e
+  PCEProtocolError e -> protocolError_ host e
  PCEResponseError e -> BROKER host $ RESPONSE $ B.unpack $ smpEncode e
  PCEUnexpectedResponse _ -> BROKER host UNEXPECTED
  PCEResponseTimeout -> BROKER host TIMEOUT
@@ -1022,7 +1030,7 @@ runXFTPServerTest c userId (ProtoServerWithAuth srv auth) = do
          liftError (testErr TSUploadFile) $ X.uploadXFTPChunk xftp spKey sId chunkSpec
          liftError (testErr TSDownloadFile) $ X.downloadXFTPChunk g xftp rpKey rId $ XFTPRcvChunkSpec rcvPath chSize digest
          rcvDigest <- liftIO $ C.sha256Hash <$> B.readFile rcvPath
-          unless (digest == rcvDigest) $ throwError $ ProtocolTestFailure TSCompareFile $ XFTP DIGEST
+          unless (digest == rcvDigest) $ throwError $ ProtocolTestFailure TSCompareFile $ XFTP (B.unpack $ strEncode srv) DIGEST
          liftError (testErr TSDeleteFile) $ X.deleteXFTPChunk xftp spKey sId
        ok <- tcpTimeout xftpNetworkConfig `timeout` X.closeXFTPClient xftp
        incClientStat c userId xftp "XFTP_TEST" "OK"
@@ -1133,15 +1141,15 @@ newRcvQueue c userId connId (ProtoServerWithAuth srv auth) vRange subMode = do
      qUri = SMPQueueUri vRange $ SMPQueueAddress srv sndId e2eDhKey
  pure (rq, qUri, tSess, sessId)

-processSubResult :: AgentClient -> RcvQueue -> Either SMPClientError () -> IO (Either SMPClientError ())
-processSubResult c rq r = do
-  case r of
-    Left e ->
-      unless (temporaryClientError e) . atomically $ do
-        RQ.deleteQueue rq (pendingSubs c)
-        TM.insert (RQ.qKey rq) e (removedSubs c)
-    _ -> atomically $ addSubscription c rq
-  pure r
+processSubResult :: AgentClient -> RcvQueue -> Either SMPClientError () -> STM ()
+processSubResult c rq@RcvQueue {connId} = \case
+  Left e ->
+    unless (temporaryClientError e) $ do
+      RQ.deleteQueue rq (pendingSubs c)
+      TM.insert (RQ.qKey rq) e (removedSubs c)
+  Right () ->
+    whenM (hasPendingSubscription c connId) $
+      addSubscription c rq

 temporaryAgentError :: AgentErrorType -> Bool
 temporaryAgentError = \case
@@ -1158,7 +1166,7 @@ temporaryOrHostError = \case
 {-# INLINE temporaryOrHostError #-}

 -- | Subscribe to queues. The list of results can have a different order.
-subscribeQueues :: AgentClient -> [RcvQueue] -> AM' [(RcvQueue, Either AgentErrorType ())]
+subscribeQueues :: AgentClient -> [RcvQueue] -> AM' ([(RcvQueue, Either AgentErrorType ())], Maybe SessionId)
 subscribeQueues c qs = do
  (errs, qs') <- partitionEithers <$> mapM checkQueue qs
  atomically $ do
@@ -1166,18 +1174,34 @@ subscribeQueues c qs = do
    RQ.batchAddQueues (pendingSubs c) qs'
  env <- ask
  -- only "checked" queues are subscribed
-  (errs <>) <$> sendTSessionBatches "SUB" 90 id (subscribeQueues_ env) c qs'
+  session <- newTVarIO Nothing
+  rs <- sendTSessionBatches "SUB" 90 id (subscribeQueues_ env session) c qs'
+  (errs <> rs,) <$> readTVarIO session
  where
    checkQueue rq = do
      prohibited <- atomically $ hasGetLock c rq
      pure $ if prohibited then Left (rq, Left $ CMD PROHIBITED) else Right rq
-    subscribeQueues_ :: Env -> SMPClient -> NonEmpty RcvQueue -> IO (BatchResponses SMPClientError ())
-    subscribeQueues_ env smp qs' = do
+    subscribeQueues_ :: Env -> TVar (Maybe SessionId) -> SMPClient -> NonEmpty RcvQueue -> IO (BatchResponses SMPClientError ())
+    subscribeQueues_ env session smp qs' = do
+      atomically . modifyTVar (sentSubs smp) . M.union $ M.fromList [(connId, False) | RcvQueue {connId} <- L.toList qs']
      rs <- sendBatch subscribeSMPQueues smp qs'
-      mapM_ (uncurry $ processSubResult c) rs
-      when (any temporaryClientError . lefts . map snd $ L.toList rs) $
-        runReaderT (resubscribeSMPSession c $ transportSession' smp) env
-      pure rs
+      atomically . modifyTVar (sentSubs smp) . M.union $ M.fromList [(connId, True) | (RcvQueue {connId}, Right ()) <- L.toList rs]
+      active <-
+        atomically $
+          ifM
+            (activeClientSession c tSess sessId)
+            (writeTVar session (Just sessId) >> mapM_ (uncurry $ processSubResult c) rs $> True)
+            (pure False)
+      if active
+        then when (hasTempErrors rs) resubscribe $> rs
+        else do
+          logWarn "subcription batch result for replaced SMP client, resubscribing"
+          resubscribe $> L.map (second $ \_ -> Left PCENetworkError) rs
+      where
+        tSess = transportSession' smp
+        sessId = sessionId $ thParams smp
+        hasTempErrors = any (either temporaryClientError (const False) . snd)
+        resubscribe = resubscribeSMPSession c tSess `runReaderT` env

 activeClientSession :: AgentClient -> SMPTransportSession -> SessionId -> STM Bool
 activeClientSession c tSess sessId = sameSess <$> tryReadSessVar tSess (smpClients c)
@@ -1186,7 +1210,7 @@ activeClientSession c tSess sessId = sameSess <$> tryReadSessVar tSess (smpClien
      Just (Right smp) -> sessId == sessionId (thParams smp)
      _ -> False

-type BatchResponses e r = (NonEmpty (RcvQueue, Either e r))
+type BatchResponses e r = NonEmpty (RcvQueue, Either e r)

 -- statBatchSize is not used to batch the commands, only for traffic statistics
 sendTSessionBatches :: forall q r. ByteString -> Int -> (q -> RcvQueue) -> (SMPClient -> NonEmpty q -> IO (BatchResponses SMPClientError r)) -> AgentClient -> [q] -> AM' [(RcvQueue, Either AgentErrorType r)]
@@ -1227,6 +1251,11 @@ addSubscription c rq@RcvQueue {connId} = do
  RQ.addQueue rq $ activeSubs c
  RQ.deleteQueue rq $ pendingSubs c

+addPendingSubscription :: AgentClient -> RcvQueue -> STM ()
+addPendingSubscription c rq@RcvQueue {connId} = do
+  modifyTVar' (subscrConns c) $ S.insert connId
+  RQ.addQueue rq $ pendingSubs c
+
 addNewQueueSubscription :: AgentClient -> RcvQueue -> SMPTransportSession -> SessionId -> AM' ()
 addNewQueueSubscription c rq tSess sessId = do
  same <-
@@ -1234,7 +1263,7 @@ addNewQueueSubscription c rq tSess sessId = do
      ifM
        (activeClientSession c tSess sessId)
        (True <$ addSubscription c rq)
-        (False <$ RQ.addQueue rq (pendingSubs c))
+        (False <$ addPendingSubscription c rq)
  unless same $ resubscribeSMPSession c tSess

 hasActiveSubscription :: AgentClient -> ConnId -> STM Bool
@@ -1255,9 +1284,9 @@ getSubscriptions :: AgentClient -> STM (Set ConnId)
 getSubscriptions = readTVar . subscrConns
 {-# INLINE getSubscriptions #-}

-logServer :: MonadIO m => ByteString -> AgentClient -> ProtocolServer s -> QueueId -> ByteString -> m ()
+logServer :: (HasCallStack, MonadIO m) => ByteString -> AgentClient -> ProtocolServer s -> QueueId -> ByteString -> m ()
 logServer dir AgentClient {clientId} srv qId cmdStr =
-  logInfo . decodeUtf8 $ B.unwords ["A", "(" <> bshow clientId <> ")", dir, showServer srv, ":", logSecret qId, cmdStr]
+  withFrozenCallStack $ logInfo . decodeUtf8 $ B.unwords ["A", "(" <> bshow clientId <> ")", dir, showServer srv, ":", logSecret qId, cmdStr]
 {-# INLINE logServer #-}

 showServer :: ProtocolServer s -> ByteString
@@ -1708,6 +1737,31 @@ getAgentSubscriptions c = do
    enc :: StrEncoding a => a -> Text
    enc = decodeLatin1 . strEncode

+diffSubscriptions :: AgentClient -> IO SubscriptionsDiff
+diffSubscriptions AgentClient {smpClients, subscrConns} = do
+  agentConns <- readTVarIO subscrConns
+  clients <- readTVarIO smpClients
+  clientsSubs <- fmap fold . forM (M.assocs clients) $ \((_, srv, _), SessionVar {sessionVar}) ->
+    atomically (tryReadTMVar sessionVar) >>= \case
+      Just (Right smp) -> readTVarIO (sentSubs smp)
+      _ -> mempty <$ putStrLn ("no client for " <> show srv)
+  let allClientsSubs = M.keysSet clientsSubs
+  pure
+    SubscriptionsDiff
+      { inBoth = S.size $ agentConns `S.intersection` allClientsSubs,
+        inAgent = map textId . S.toList $ agentConns `S.difference` allClientsSubs,
+        inClients = M.fromList . map (\k -> (textId k, M.lookup k clientsSubs == Just True)) . S.toList $ allClientsSubs `S.difference` agentConns
+      }
+  where
+    textId = decodeLatin1 . strEncode
+
+data SubscriptionsDiff = SubscriptionsDiff
+  { inBoth :: Int,
+    inAgent :: [Text],
+    inClients :: Map Text Bool
+  }
+  deriving (Show)
+
 data AgentWorkersDetails = AgentWorkersDetails
  { smpClients_ :: [Text],
    ntfClients_ :: [Text],
@@ -1845,6 +1899,8 @@ $(J.deriveJSON defaultJSON ''SubInfo)

 $(J.deriveJSON defaultJSON ''SubscriptionsInfo)

+$(J.deriveJSON defaultJSON ''SubscriptionsDiff)
+
 $(J.deriveJSON defaultJSON ''WorkersDetails)

 $(J.deriveJSON defaultJSON ''WorkersSummary)
@@ -193,13 +193,13 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.Ratchet
  ( InitialKeys (..),
    PQEncryption (..),
-    pattern PQEncOff,
    PQSupport,
-    pattern PQSupportOn,
-    pattern PQSupportOff,
    RcvE2ERatchetParams,
    RcvE2ERatchetParamsUri,
-    SndE2ERatchetParams
+    SndE2ERatchetParams,
+    pattern PQEncOff,
+    pattern PQSupportOff,
+    pattern PQSupportOn,
  )
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
@@ -213,14 +213,14 @@ import Simplex.Messaging.Protocol
    MsgId,
    NMsgMeta,
    ProtocolServer (..),
+    SMPClientVersion,
    SMPMsgMeta,
    SMPServer,
    SMPServerWithAuth,
    SndPublicAuthKey,
    SubscriptionMode,
-    SMPClientVersion,
-    VersionSMPC,
    VersionRangeSMPC,
+    VersionSMPC,
    initialSMPClientVersion,
    legacyEncodeServer,
    legacyServerP,
@@ -908,7 +908,7 @@ instance Encoding AgentMsgEnvelope where
 -- AgentRatchetInfo is not encrypted with double ratchet, but with per-queue E2E encryption
 data AgentMessage
  = -- used by the initiating party when confirming reply queue
-  AgentConnInfo ConnInfo
+    AgentConnInfo ConnInfo
  | -- AgentConnInfoReply is used by accepting party in duplexHandshake mode (v2), allowing to include reply queue(s) in the initial confirmation.
    -- It made removed REPLY message unnecessary.
    AgentConnInfoReply (NonEmpty SMPQueueInfo) ConnInfo
@@ -1382,9 +1382,9 @@ deriving instance Show (ConnectionRequestUri m)
 data AConnectionRequestUri = forall m. ConnectionModeI m => ACR (SConnectionMode m) (ConnectionRequestUri m)

 instance Eq AConnectionRequestUri where
-   ACR m cr == ACR m' cr' = case testEquality m m' of
-     Just Refl -> cr == cr'
-     _ -> False
+  ACR m cr == ACR m' cr' = case testEquality m m' of
+    Just Refl -> cr == cr'
+    _ -> False

 deriving instance Show AConnectionRequestUri

@@ -1469,11 +1469,11 @@ data AgentErrorType
  | -- | connection errors
    CONN {connErr :: ConnectionErrorType}
  | -- | SMP protocol errors forwarded to agent clients
-    SMP {smpErr :: ErrorType}
+    SMP {serverAddress :: String, smpErr :: ErrorType}
  | -- | NTF protocol errors forwarded to agent clients
-    NTF {ntfErr :: ErrorType}
+    NTF {serverAddress :: String, ntfErr :: ErrorType}
  | -- | XFTP protocol errors forwarded to agent clients
-    XFTP {xftpErr :: XFTPErrorType}
+    XFTP {serverAddress :: String, xftpErr :: XFTPErrorType}
  | -- | XRCP protocol errors forwarded to agent clients
    RCP {rcpErr :: RCErrorType}
  | -- | SMP server errors
@@ -1584,9 +1584,9 @@ instance StrEncoding AgentErrorType where
  strP =
    "CMD " *> (CMD <$> parseRead1)
      <|> "CONN " *> (CONN <$> parseRead1)
-      <|> "SMP " *> (SMP <$> strP)
-      <|> "NTF " *> (NTF <$> strP)
-      <|> "XFTP " *> (XFTP <$> strP)
+      <|> "SMP " *> (SMP <$> textP <*> _strP)
+      <|> "NTF " *> (NTF <$> textP <*> _strP)
+      <|> "XFTP " *> (XFTP <$> textP <*> _strP)
      <|> "RCP " *> (RCP <$> strP)
      <|> "BROKER " *> (BROKER <$> textP <* " RESPONSE " <*> (RESPONSE <$> textP))
      <|> "BROKER " *> (BROKER <$> textP <* " TRANSPORT " <*> (TRANSPORT <$> transportErrorP))
@@ -1602,9 +1602,9 @@ instance StrEncoding AgentErrorType where
  strEncode = \case
    CMD e -> "CMD " <> bshow e
    CONN e -> "CONN " <> bshow e
-    SMP e -> "SMP " <> strEncode e
-    NTF e -> "NTF " <> strEncode e
-    XFTP e -> "XFTP " <> strEncode e
+    SMP srv e -> "SMP " <> text srv <> " " <> strEncode e
+    NTF srv e -> "NTF " <> text srv <> " " <> strEncode e
+    XFTP srv e -> "XFTP " <> text srv <> " " <> strEncode e
    RCP e -> "RCP " <> strEncode e
    BROKER srv (RESPONSE e) -> "BROKER " <> text srv <> " RESPONSE " <> text e
    BROKER srv (TRANSPORT e) -> "BROKER " <> text srv <> " TRANSPORT " <> serializeTransportError e
@@ -28,7 +28,7 @@
 module Simplex.Messaging.Client
  ( -- * Connect (disconnect) client to (from) SMP server
    TransportSession,
-    ProtocolClient (thParams, sessionTs),
+    ProtocolClient (thParams, sessionTs, sentSubs),
    SMPClient,
    getProtocolClient,
    closeProtocolClient,
@@ -70,6 +70,7 @@ module Simplex.Messaging.Client
    proxyUsername,
    temporaryClientError,
    ServerTransmission,
+    TransmissionType (..),
    ClientCommand,

    -- * For testing
@@ -83,6 +84,7 @@ where
 import Control.Concurrent.Async
 import Control.Concurrent.STM
 import Control.Exception
+import Control.Logger.Simple
 import Control.Monad
 import Control.Monad.Except
 import Control.Monad.IO.Class (liftIO)
@@ -110,10 +112,9 @@ import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Client (SocksProxy, TransportClientConfig (..), TransportHost (..), defaultTcpConnectTimeout, runTransportClient)
 import Simplex.Messaging.Transport.KeepAlive
 import Simplex.Messaging.Transport.WebSockets (WS)
-import Simplex.Messaging.Util (bshow, diffToMicroseconds, raceAny_, threadDelay', whenM)
+import Simplex.Messaging.Util (bshow, diffToMicroseconds, ifM, raceAny_, threadDelay', tshow, whenM)
 import Simplex.Messaging.Version
 import System.Timeout (timeout)
-import UnliftIO (pooledMapConcurrentlyN)

 -- | 'SMPClient' is a handle used to send commands to a specific SMP server.
 --
@@ -122,6 +123,7 @@ data ProtocolClient v err msg = ProtocolClient
  { action :: Maybe (Async ()),
    thParams :: THandleParams v 'TClient,
    sessionTs :: UTCTime,
+    sentSubs :: TMap ByteString Bool, -- DEBUG: ConnId -> sub status
    client_ :: PClient v err msg
  }

@@ -136,9 +138,9 @@ data PClient v err msg = PClient
    timeoutErrorCount :: TVar Int,
    clientCorrId :: TVar ChaChaDRG,
    sentCommands :: TMap CorrId (Request err msg),
-    sndQ :: TBQueue (Maybe (TVar Bool), ByteString),
+    sndQ :: TBQueue (TVar Bool, ByteString),
    rcvQ :: TBQueue (NonEmpty (SignedTransmission err msg)),
-    msgQ :: Maybe (TBQueue (ServerTransmission v msg))
+    msgQ :: Maybe (TBQueue (ServerTransmission v err msg))
  }

 smpClientStub :: TVar ChaChaDRG -> ByteString -> VersionSMP -> Maybe (THandleAuth 'TClient) -> STM SMPClient
@@ -148,6 +150,7 @@ smpClientStub g sessionId thVersion thAuth = do
  clientCorrId <- C.newRandomDRG g
  sentCommands <- TM.empty
  sendPings <- newTVar False
+  sentSubs <- TM.empty
  lastReceived <- newTVar ts
  timeoutErrorCount <- newTVar 0
  sndQ <- newTBQueue 100
@@ -165,6 +168,7 @@ smpClientStub g sessionId thVersion thAuth = do
              batch = True
            },
        sessionTs = ts,
+        sentSubs,
        client_ =
          PClient
            { connected,
@@ -189,7 +193,10 @@ type SMPClient = ProtocolClient SMPVersion ErrorType BrokerMsg
 type ClientCommand msg = (Maybe C.APrivateAuthKey, EntityId, ProtoCommand msg)

 -- | Type synonym for transmission from some SPM server queue.
-type ServerTransmission v msg = (TransportSession msg, Version v, SessionId, Bool, EntityId, msg)
+type ServerTransmission v err msg = (TransportSession msg, Version v, SessionId, TransmissionType, EntityId, Either (ProtocolClientError err) msg)
+
+data TransmissionType = TTEvent | TTUncorrelatedResponse | TTExpiredSubResponse
+  deriving (Show)

 data HostMode
  = -- | prefer (or require) onion hosts when connecting via SOCKS proxy
@@ -286,6 +293,8 @@ defaultSMPClientConfig = defaultClientConfig supportedClientSMPRelayVRange
 data Request err msg = Request
  { corrId :: CorrId,
    entityId :: EntityId,
+    subscription :: Bool,
+    pending :: TVar Bool,
    responseVar :: TMVar (Either (ProtocolClientError err) msg)
  }

@@ -332,7 +341,7 @@ type TransportSession msg = (UserId, ProtoServer msg, Maybe EntityId)
 --
 -- A single queue can be used for multiple 'SMPClient' instances,
 -- as 'SMPServerTransmission' includes server information.
-getProtocolClient :: forall v err msg. Protocol v err msg => TVar ChaChaDRG -> TransportSession msg -> ProtocolClientConfig v -> Maybe (TBQueue (ServerTransmission v msg)) -> (ProtocolClient v err msg -> IO ()) -> IO (Either (ProtocolClientError err) (ProtocolClient v err msg))
+getProtocolClient :: forall v err msg. Protocol v err msg => TVar ChaChaDRG -> TransportSession msg -> ProtocolClientConfig v -> Maybe (TBQueue (ServerTransmission v err msg)) -> (ProtocolClient v err msg -> IO ()) -> IO (Either (ProtocolClientError err) (ProtocolClient v err msg))
 getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize, networkConfig, serverVRange, agreeSecret} msgQ disconnected = do
  case chooseTransportHost networkConfig (host srv) of
    Right useHost ->
@@ -396,7 +405,8 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
        Left e -> atomically . putTMVar cVar . Left $ PCETransportError e
        Right th@THandle {params} -> do
          sessionTs <- getCurrentTime
-          let c' = ProtocolClient {action = Nothing, client_ = c, thParams = params, sessionTs}
+          sentSubs <- atomically TM.empty
+          let c' = ProtocolClient {action = Nothing, client_ = c, thParams = params, sessionTs, sentSubs}
          atomically $ writeTVar (lastReceived c) sessionTs
          atomically $ do
            writeTVar (connected c) True
@@ -407,9 +417,7 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
    send :: Transport c => ProtocolClient v err msg -> THandle v c 'TClient -> IO ()
    send ProtocolClient {client_ = PClient {sndQ}} h = forever $ atomically (readTBQueue sndQ) >>= sendActive
      where
-        sendActive (Nothing, s) = send_ s
-        sendActive (Just active, s) = whenM (readTVarIO active) $ send_ s
-        send_ = void . tPutLog h
+        sendActive (active, s) = whenM (readTVarIO active) $ void $ tPutLog h s

    receive :: Transport c => ProtocolClient v err msg -> THandle v c 'TClient -> IO ()
    receive ProtocolClient {client_ = PClient {rcvQ, lastReceived, timeoutErrorCount}} h = forever $ do
@@ -442,27 +450,40 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize

    processMsg :: ProtocolClient v err msg -> SignedTransmission err msg -> IO ()
    processMsg c@ProtocolClient {client_ = PClient {sentCommands}} (_, _, (corrId, entId, respOrErr))
-      | isResponse =
+      | not $ B.null $ bs corrId =
          atomically (TM.lookup corrId sentCommands) >>= \case
-            Nothing -> sendMsg respOrErr
-            Just Request {entityId, responseVar} -> atomically $ do
-              TM.delete corrId sentCommands
-              putTMVar responseVar $ response entityId
-      | otherwise = sendMsg respOrErr
+            Nothing -> sendMsg TTUncorrelatedResponse
+            Just Request {entityId, subscription, pending, responseVar} -> do
+              wasPending <-
+                atomically $ do
+                  TM.delete corrId sentCommands
+                  ifM
+                    (readTVar pending)
+                    (True <$ tryPutTMVar responseVar (response entityId))
+                    (pure False)
+              unless wasPending $ mapM_ sendMsg $ transmissionType entityId subscription
+      | otherwise = sendMsg TTEvent
      where
-        isResponse = not $ B.null $ bs corrId
+        -- we don't send expired events that are not subscription responses to avoid double error reporting,
+        -- as the expired command already reported them.
+        transmissionType entityId subscription
+          | entityId /= entId = Just TTUncorrelatedResponse
+          | subscription = Just TTExpiredSubResponse
+          | otherwise = Nothing
        response entityId
-          | entityId == entId =
-              case respOrErr of
-                Left e -> Left $ PCEResponseError e
-                Right r -> case protocolError r of
-                  Just e -> Left $ PCEProtocolError e
-                  _ -> Right r
+          | entityId == entId = clientResp
          | otherwise = Left . PCEUnexpectedResponse $ bshow respOrErr
-        sendMsg :: Either err msg -> IO ()
-        sendMsg = \case
-          Right msg -> atomically $ mapM_ (`writeTBQueue` serverTransmission c isResponse entId msg) msgQ
-          Left e -> putStrLn $ "SMP client error: " <> show e
+        clientResp = case respOrErr of
+          Left e -> Left $ PCEResponseError e
+          Right r -> case protocolError r of
+            Just e -> Left $ PCEProtocolError e
+            _ -> Right r
+        sendMsg :: TransmissionType -> IO ()
+        sendMsg tType = case msgQ of
+          Just q -> atomically $ writeTBQueue q $ serverTransmission c tType entId clientResp
+          Nothing -> case clientResp of
+            Left e -> logError $ "SMP client error: " <> tshow e
+            Right _ -> logWarn $ "SMP client unprocessed event: " <> tshow tType

 proxyUsername :: TransportSession msg -> ByteString
 proxyUsername (userId, _, entityId_) = C.sha256Hash $ bshow userId <> maybe "" (":" <>) entityId_
@@ -534,10 +555,7 @@ createSMPQueue c (rKey, rpKey) dhKey auth subMode =
 subscribeSMPQueue :: SMPClient -> RcvPrivateAuthKey -> RecipientId -> ExceptT SMPClientError IO ()
 subscribeSMPQueue c@ProtocolClient {client_ = PClient {sendPings}} rpKey rId = do
  liftIO . atomically $ writeTVar sendPings True
-  -- We are not expiring sending subscriptions even if it expires to increase chances of subscription
-  -- succeeding without retries - the subscription is registered in the agent when uncorrelated MSG response
-  -- is received via the active client for pending queue subscription - it also prevents unnecessary retries.
-  sendProtocolCommand' False c (Just rpKey) rId (Cmd SRecipient SUB) >>= \case
+  sendSMPCommand c (Just rpKey) rId SUB >>= \case
    OK -> pure ()
    cmd@MSG {} -> liftIO $ writeSMPMessage c rId cmd
    r -> throwE . PCEUnexpectedResponse $ bshow r
@@ -546,8 +564,7 @@ subscribeSMPQueue c@ProtocolClient {client_ = PClient {sendPings}} rpKey rId = d
 subscribeSMPQueues :: SMPClient -> NonEmpty (RcvPrivateAuthKey, RecipientId) -> IO (NonEmpty (Either SMPClientError ()))
 subscribeSMPQueues c@ProtocolClient {client_ = PClient {sendPings}} qs = do
  atomically $ writeTVar sendPings True
-  -- See comment in subscribeSMPQueue
-  sendProtocolCommands' False c cs >>= mapM (processSUBResponse c)
+  sendProtocolCommands c cs >>= mapM (processSUBResponse c)
  where
    cs = L.map (\(rpKey, rId) -> (Just rpKey, rId, Cmd SRecipient SUB)) qs

@@ -560,17 +577,16 @@ streamSubscribeSMPQueues c qs cb = streamProtocolCommands c cs $ mapM process >=
 processSUBResponse :: SMPClient -> Response ErrorType BrokerMsg -> IO (Either SMPClientError ())
 processSUBResponse c (Response rId r) = case r of
  Right OK -> pure $ Right ()
-  Right SUBOK -> pure $ Right () -- from v8 of SMP protocol
  Right cmd@MSG {} -> writeSMPMessage c rId cmd $> Right ()
  Right r' -> pure . Left . PCEUnexpectedResponse $ bshow r'
  Left e -> pure $ Left e

 writeSMPMessage :: SMPClient -> RecipientId -> BrokerMsg -> IO ()
-writeSMPMessage c rId msg = atomically $ mapM_ (`writeTBQueue` serverTransmission c False rId msg) (msgQ $ client_ c)
+writeSMPMessage c rId msg = atomically $ mapM_ (`writeTBQueue` serverTransmission c TTEvent rId (Right msg)) (msgQ $ client_ c)

-serverTransmission :: ProtocolClient v err msg -> Bool -> RecipientId -> msg -> ServerTransmission v msg
-serverTransmission ProtocolClient {thParams = THandleParams {thVersion, sessionId}, client_ = PClient {transportSession}} isResponse entityId message =
-  (transportSession, thVersion, sessionId, isResponse, entityId, message)
+serverTransmission :: ProtocolClient v err msg -> TransmissionType -> RecipientId -> Either (ProtocolClientError err) msg -> ServerTransmission v err msg
+serverTransmission ProtocolClient {thParams = THandleParams {thVersion, sessionId}, client_ = PClient {transportSession}} tType entityId msgOrErr =
+  (transportSession, thVersion, sessionId, tType, entityId, msgOrErr)

 -- | Get message from SMP queue. The server returns ERR PROHIBITED if a client uses SUB and GET via the same transport connection for the same queue
 --
@@ -695,14 +711,10 @@ sendSMPCommand c pKey qId cmd = sendProtocolCommand c pKey qId (Cmd sParty cmd)
 type PCTransmission err msg = (Either TransportError SentRawTransmission, Request err msg)

 -- | Send multiple commands with batching and collect responses
-sendProtocolCommands :: forall v err msg. ProtocolEncoding v err (ProtoCommand msg) => ProtocolClient v err msg -> NonEmpty (ClientCommand msg) -> IO (NonEmpty (Response err msg))
-sendProtocolCommands = sendProtocolCommands' True
-{-# INLINE sendProtocolCommands #-}
-
-sendProtocolCommands' :: forall v err msg. ProtocolEncoding v err (ProtoCommand msg) => Bool -> ProtocolClient v err msg -> NonEmpty (ClientCommand msg) -> IO (NonEmpty (Response err msg))
-sendProtocolCommands' expire c@ProtocolClient {thParams = THandleParams {batch, blockSize}} cs = do
+sendProtocolCommands :: forall v err msg. Protocol v err msg => ProtocolClient v err msg -> NonEmpty (ClientCommand msg) -> IO (NonEmpty (Response err msg))
+sendProtocolCommands c@ProtocolClient {thParams = THandleParams {batch, blockSize}} cs = do
  bs <- batchTransmissions' batch blockSize <$> mapM (mkTransmission c) cs
-  validate . concat =<< mapM (sendBatch c expire) bs
+  validate . concat =<< mapM (sendBatch c) bs
  where
    validate :: [Response err msg] -> IO (NonEmpty (Response err msg))
    validate rs
@@ -716,39 +728,31 @@ sendProtocolCommands' expire c@ProtocolClient {thParams = THandleParams {batch,
      where
        diff = L.length cs - length rs

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

-sendBatch :: ProtocolClient v err msg -> Bool -> TransportBatch (Request err msg) -> IO [Response err msg]
-sendBatch c@ProtocolClient {client_ = PClient {rcvConcurrency, sndQ}} expire b = do
+sendBatch :: ProtocolClient v err msg -> TransportBatch (Request err msg) -> IO [Response err msg]
+sendBatch c@ProtocolClient {client_ = PClient {sndQ}} b = do
  case b of
    TBError e Request {entityId} -> do
      putStrLn "send error: large message"
      pure [Response entityId $ Left $ PCETransportError e]
    TBTransmissions s n rs
      | n > 0 -> do
-          active <- mkActive_ expire
+          active <- newTVarIO True
          atomically $ writeTBQueue sndQ (active, s)
-          pooledMapConcurrentlyN rcvConcurrency (getResponse c active) rs
+          mapConcurrently (getResponse c active) rs
      | otherwise -> pure []
    TBTransmission s r -> do
-      active <- mkActive_ expire
+      active <- newTVarIO True
      atomically $ writeTBQueue sndQ (active, s)
      (: []) <$> getResponse c active r

-mkActive_ :: Bool -> IO (Maybe (TVar Bool))
-mkActive_ True = Just <$> newTVarIO True
-mkActive_ False = pure Nothing
-
 -- | Send Protocol command
-sendProtocolCommand :: forall v err msg. ProtocolEncoding v err (ProtoCommand msg) => ProtocolClient v err msg -> Maybe C.APrivateAuthKey -> EntityId -> ProtoCommand msg -> ExceptT (ProtocolClientError err) IO msg
-sendProtocolCommand = sendProtocolCommand' True
-{-# INLINE sendProtocolCommand #-}
-
-sendProtocolCommand' :: forall v err msg. ProtocolEncoding v err (ProtoCommand msg) => Bool -> ProtocolClient v err msg -> Maybe C.APrivateAuthKey -> EntityId -> ProtoCommand msg -> ExceptT (ProtocolClientError err) IO msg
-sendProtocolCommand' expire c@ProtocolClient {client_ = PClient {sndQ}, thParams = THandleParams {batch, blockSize}} pKey entId cmd =
+sendProtocolCommand :: forall v err msg. Protocol v err msg => ProtocolClient v err msg -> Maybe C.APrivateAuthKey -> EntityId -> ProtoCommand msg -> ExceptT (ProtocolClientError err) IO msg
+sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ}, thParams = THandleParams {batch, blockSize}} pKey entId cmd =
  ExceptT $ uncurry sendRecv =<< mkTransmission c (pKey, entId, cmd)
  where
    -- two separate "atomically" needed to avoid blocking
@@ -758,7 +762,7 @@ sendProtocolCommand' expire c@ProtocolClient {client_ = PClient {sndQ}, thParams
      Right t
        | B.length s > blockSize - 2 -> pure . Left $ PCETransportError TELargeMsg
        | otherwise -> do
-            active <- mkActive_ expire
+            active <- newTVarIO True
            atomically (writeTBQueue sndQ (active, s))
            response <$> getResponse c active r
        where
@@ -767,21 +771,21 @@ sendProtocolCommand' expire c@ProtocolClient {client_ = PClient {sndQ}, thParams
            | otherwise = tEncode t

 -- TODO switch to timeout or TimeManager that supports Int64
-getResponse :: ProtocolClient v err msg -> Maybe (TVar Bool) -> Request err msg -> IO (Response err msg)
-getResponse ProtocolClient {client_ = PClient {tcpTimeout, timeoutErrorCount, sentCommands}} active Request {corrId, entityId, responseVar} = do
+getResponse :: ProtocolClient v err msg -> TVar Bool -> Request err msg -> IO (Response err msg)
+getResponse ProtocolClient {client_ = PClient {tcpTimeout, timeoutErrorCount}} active Request {entityId, pending, responseVar} = do
  response <-
    timeout tcpTimeout (atomically (takeTMVar responseVar)) >>= \case
      Just r -> atomically (writeTVar timeoutErrorCount 0) $> r
      Nothing -> do
-        atomically (mapM_ (`writeTVar` False) active >> TM.delete corrId sentCommands)
+        atomically (writeTVar active False >> writeTVar pending False)
        atomically $ modifyTVar' timeoutErrorCount (+ 1)
        pure $ Left PCEResponseTimeout
  pure Response {entityId, response}

-mkTransmission :: forall v err msg. ProtocolEncoding v err (ProtoCommand msg) => ProtocolClient v err msg -> ClientCommand msg -> IO (PCTransmission err msg)
-mkTransmission ProtocolClient {thParams, client_ = PClient {clientCorrId, sentCommands}} (pKey_, entId, cmd) = do
+mkTransmission :: forall v err msg. Protocol v err msg => ProtocolClient v err msg -> ClientCommand msg -> IO (PCTransmission err msg)
+mkTransmission ProtocolClient {thParams, client_ = PClient {clientCorrId, sentCommands}} (pKey_, entityId, cmd) = do
  corrId <- atomically getNextCorrId
-  let TransmissionForAuth {tForAuth, tToSend} = encodeTransmissionForAuth thParams (corrId, entId, cmd)
+  let TransmissionForAuth {tForAuth, tToSend} = encodeTransmissionForAuth thParams (corrId, entityId, cmd)
      auth = authTransmission (thAuth thParams) pKey_ corrId tForAuth
  r <- atomically $ mkRequest corrId
  pure ((,tToSend) <$> auth, r)
@@ -790,7 +794,16 @@ mkTransmission ProtocolClient {thParams, client_ = PClient {clientCorrId, sentCo
    getNextCorrId = CorrId <$> C.randomBytes 24 clientCorrId -- also used as nonce
    mkRequest :: CorrId -> STM (Request err msg)
    mkRequest corrId = do
-      r <- Request corrId entId <$> newEmptyTMVar
+      pending <- newTVar True
+      responseVar <- newEmptyTMVar
+      let r =
+            Request
+              { corrId,
+                entityId,
+                subscription = protocolSub cmd,
+                pending,
+                responseVar
+              }
      TM.insert corrId r sentCommands
      pure r

@@ -39,7 +39,7 @@ import Simplex.Messaging.Agent.RetryInterval
 import Simplex.Messaging.Client
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
-import Simplex.Messaging.Protocol (BrokerMsg, NotifierId, NtfPrivateAuthKey, ProtocolServer (..), QueueId, RcvPrivateAuthKey, RecipientId, SMPServer)
+import Simplex.Messaging.Protocol (BrokerMsg, ErrorType, NotifierId, NtfPrivateAuthKey, ProtocolServer (..), QueueId, RcvPrivateAuthKey, RecipientId, SMPServer)
 import Simplex.Messaging.Session
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
@@ -94,7 +94,7 @@ defaultSMPClientAgentConfig =

 data SMPClientAgent = SMPClientAgent
  { agentCfg :: SMPClientAgentConfig,
-    msgQ :: TBQueue (ServerTransmission SMPVersion BrokerMsg),
+    msgQ :: TBQueue (ServerTransmission SMPVersion ErrorType BrokerMsg),
    agentQ :: TBQueue SMPClientAgentEvent,
    randomDrg :: TVar ChaChaDRG,
    smpClients :: TMap SMPServer SMPClientVar,
@@ -154,6 +154,7 @@ instance Protocol NTFVersion ErrorType NtfResponse where
  type ProtoType NtfResponse = 'PNTF
  protocolClientHandshake c _ks = ntfClientHandshake c
  protocolPing = NtfCmd SSubscription PING
+  protocolSub _ = False
  protocolError = \case
    NRErr e -> Just e
    _ -> Nothing
@@ -218,10 +218,10 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge

    receiveSMP :: M ()
    receiveSMP = forever $ do
-      ((_, srv, _), _, _, _, ntfId, msg) <- atomically $ readTBQueue msgQ
+      ((_, srv, _), _, _, tType, ntfId, msgOrErr) <- atomically $ readTBQueue msgQ
      let smpQueue = SMPQueueNtf srv ntfId
-      case msg of
-        SMP.NMSG nmsgNonce encNMsgMeta -> do
+      case msgOrErr of
+        Right (SMP.NMSG nmsgNonce encNMsgMeta) -> do
          ntfTs <- liftIO getSystemTime
          st <- asks store
          NtfPushServer {pushQ} <- asks pushServer
@@ -231,8 +231,10 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
            findNtfSubscriptionToken st smpQueue
              >>= mapM_ (\tkn -> writeTBQueue pushQ (tkn, PNMessage PNMessageData {smpQueue, ntfTs, nmsgNonce, encNMsgMeta}))
          incNtfStat ntfReceived
-        SMP.END -> updateSubStatus smpQueue NSEnd
-        _ -> pure ()
+        Right SMP.END -> updateSubStatus smpQueue NSEnd
+        Right (SMP.ERR e) -> logError $ "SMP server error: " <> tshow e
+        Right _ -> logError $ "SMP server unexpected response: " <> tshow tType
+        Left e -> logError $ "SMP client error: " <> tshow e

    receiveAgent =
      forever $
@@ -395,7 +395,6 @@ data BrokerMsg where
  NMSG :: C.CbNonce -> EncNMsgMeta -> BrokerMsg
  END :: BrokerMsg
  OK :: BrokerMsg
-  SUBOK :: BrokerMsg
  ERR :: ErrorType -> BrokerMsg
  PONG :: BrokerMsg
  deriving (Eq, Show)
@@ -583,7 +582,6 @@ data BrokerMsgTag
  | NMSG_
  | END_
  | OK_
-  | SUBOK_
  | ERR_
  | PONG_
  deriving (Show)
@@ -643,7 +641,6 @@ instance Encoding BrokerMsgTag where
    NMSG_ -> "NMSG"
    END_ -> "END"
    OK_ -> "OK"
-    SUBOK_ -> "SUBOK"
    ERR_ -> "ERR"
    PONG_ -> "PONG"
  smpP = messageTagP
@@ -656,7 +653,6 @@ instance ProtocolMsgTag BrokerMsgTag where
    "NMSG" -> Just NMSG_
    "END" -> Just END_
    "OK" -> Just OK_
-    "SUBOK" -> Just SUBOK_
    "ERR" -> Just ERR_
    "PONG" -> Just PONG_
    _ -> Nothing
@@ -1098,6 +1094,7 @@ class (ProtocolEncoding v err msg, ProtocolEncoding v err (ProtoCommand msg), Sh
  type ProtoType msg = (sch :: ProtocolType) | sch -> msg
  protocolClientHandshake :: forall c. Transport c => c -> Maybe C.KeyPairX25519 -> C.KeyHash -> VersionRange v -> ExceptT TransportError IO (THandle v c 'TClient)
  protocolPing :: ProtoCommand msg
+  protocolSub :: ProtoCommand msg -> Bool
  protocolError :: msg -> Maybe err

 type ProtoServer msg = ProtocolServer (ProtoType msg)
@@ -1107,6 +1104,10 @@ instance Protocol SMPVersion ErrorType BrokerMsg where
  type ProtoType BrokerMsg = 'PSMP
  protocolClientHandshake = smpClientHandshake
  protocolPing = Cmd SSender PING
+  protocolSub = \case
+    Cmd SRecipient SUB -> True
+    Cmd SNotifier NSUB -> True
+    _ -> False
  protocolError = \case
    ERR e -> Just e
    _ -> Nothing
@@ -1202,7 +1203,7 @@ instance ProtocolEncoding SMPVersion ErrorType Cmd where

 instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where
  type Tag BrokerMsg = BrokerMsgTag
-  encodeProtocol v = \case
+  encodeProtocol _v = \case
    IDS (QIK rcvId sndId srvDh) -> e (IDS_, ' ', rcvId, sndId, srvDh)
    MSG RcvMessage {msgId, msgBody = EncRcvMsgBody body} ->
      e (MSG_, ' ', msgId, Tail body)
@@ -1210,9 +1211,6 @@ instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where
    NMSG nmsgNonce encNMsgMeta -> e (NMSG_, ' ', nmsgNonce, encNMsgMeta)
    END -> e END_
    OK -> e OK_
-    SUBOK
-      | v >= sendingProxySMPVersion -> e SUBOK_
-      | otherwise -> e OK_
    ERR err -> e (ERR_, ' ', err)
    PONG -> e PONG_
    where
@@ -1230,7 +1228,6 @@ instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where
    NMSG_ -> NMSG <$> _smpP <*> smpP
    END_ -> pure END
    OK_ -> pure OK
-    SUBOK_ -> pure SUBOK
    ERR_ -> ERR <$> _smpP
    PONG_ -> pure PONG

@@ -724,7 +724,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
            deliver sub = do
              q <- getStoreMsgQueue "SUB" rId
              msg_ <- atomically $ tryPeekMsg q
-              deliverMessage "SUB" qr rId sub q msg_ SUBOK
+              deliverMessage "SUB" qr rId sub q msg_

        getMessage :: QueueRec -> M (Transmission BrokerMsg)
        getMessage qr = time "GET" $ do
@@ -781,7 +781,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
                    _ -> do
                      (deletedMsg_, msg_) <- atomically $ tryDelPeekMsg q msgId
                      mapM_ updateStats deletedMsg_
-                      deliverMessage "ACK" qr queueId sub q msg_ OK
+                      deliverMessage "ACK" qr queueId sub q msg_
                _ -> pure $ err NO_MSG
          where
            getDelivered :: TVar Sub -> STM (Maybe Sub)
@@ -863,8 +863,8 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
                  encNMsgMeta = C.cbEncrypt rcvNtfDhSecret cbNonce (smpEncode msgMeta) 128
              pure . (cbNonce,) $ fromRight "" encNMsgMeta

-        deliverMessage :: T.Text -> QueueRec -> RecipientId -> TVar Sub -> MsgQueue -> Maybe Message -> BrokerMsg -> M (Transmission BrokerMsg)
-        deliverMessage name qr rId sub q msg_ respCmd = time (name <> " deliver") $ do
+        deliverMessage :: T.Text -> QueueRec -> RecipientId -> TVar Sub -> MsgQueue -> Maybe Message -> M (Transmission BrokerMsg)
+        deliverMessage name qr rId sub q msg_ = time (name <> " deliver") $ do
          readTVarIO sub >>= \case
            s@Sub {subThread = NoSub} ->
              case msg_ of
@@ -874,7 +874,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
                _ -> forkSub $> resp
            _ -> pure resp
          where
-            resp = (corrId, rId, respCmd)
+            resp = (corrId, rId, OK)
            forkSub :: M ()
            forkSub = do
              atomically . modifyTVar' sub $ \s -> s {subThread = SubPending}
@@ -41,7 +41,6 @@ module Simplex.Messaging.Transport
    basicAuthSMPVersion,
    subModeSMPVersion,
    authCmdsSMPVersion,
-    sendingProxySMPVersion,
    simplexMQVersion,
    smpBlockSize,
    TransportConfig (..),
@@ -150,9 +149,6 @@ subModeSMPVersion = VersionSMP 6
 authCmdsSMPVersion :: VersionSMP
 authCmdsSMPVersion = VersionSMP 7

-sendingProxySMPVersion :: VersionSMP
-sendingProxySMPVersion = VersionSMP 8
-
 currentClientSMPRelayVersion :: VersionSMP
 currentClientSMPRelayVersion = VersionSMP 6

@@ -54,7 +54,6 @@ import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.KeepAlive
 import Simplex.Messaging.Util (bshow, catchAll, tshow, (<$?>))
 import System.IO.Error
-import System.Timeout (timeout)
 import Text.Read (readMaybe)
 import UnliftIO.Exception (IOException)
 import qualified UnliftIO.Exception as E
@@ -139,35 +138,26 @@ runTransportClient :: Transport c => TransportClientConfig -> Maybe ByteString -
 runTransportClient = runTLSTransportClient supportedParameters Nothing

 runTLSTransportClient :: Transport c => T.Supported -> Maybe XS.CertificateStore -> TransportClientConfig -> Maybe ByteString -> TransportHost -> ServiceName -> Maybe C.KeyHash -> (c -> IO a) -> IO a
-runTLSTransportClient tlsParams caStore_ cfg@TransportClientConfig {socksProxy, tcpConnectTimeout, tcpKeepAlive, clientCredentials, alpn} proxyUsername host port keyHash client = do
+runTLSTransportClient tlsParams caStore_ cfg@TransportClientConfig {socksProxy, tcpKeepAlive, clientCredentials, alpn} proxyUsername host port keyHash client = do
  serverCert <- newEmptyTMVarIO
  let hostName = B.unpack $ strEncode host
      clientParams = mkTLSClientParams tlsParams caStore_ hostName port keyHash clientCredentials alpn serverCert
-      (connectTCP, tlsTimeout) = case socksProxy of
-        -- We use a much larger timeout for connections via SOCKS proxy, to allow the circuits created
-        -- in the socket connection that would otherwise timeout to be used in the next connection attempt.
-        -- Using standard timeout results in permanent timeout for the clients using SOCKS in cases
-        -- when SOCKS proxy is very slow (bad network, congestion in underlying network, etc.),
-        -- because SOCKS proxy destroys circuits when the last session using them is closed.
-        Just proxy -> (connectSocksClient proxy proxyUsername (hostAddr host), tcpConnectTimeout * 10)
-        _ -> (connectTCPClient hostName, tcpConnectTimeout)
+      connectTCP = case socksProxy of
+        Just proxy -> connectSocksClient proxy proxyUsername (hostAddr host)
+        _ -> connectTCPClient hostName
  c <- do
    sock <- connectTCP port
    mapM_ (setSocketKeepAlive sock) tcpKeepAlive `catchAll` \e -> logError ("Error setting TCP keep-alive" <> tshow e)
    let tCfg = clientTransportConfig cfg
-    tlsTimeout `timeout` connectTLS (Just hostName) tCfg clientParams sock >>= \case
-      Nothing -> do
-        close sock
-        logError "connection timed out"
-        fail "connection timed out"
-      Just tls -> do
-        chain <-
-          atomically (tryTakeTMVar serverCert) >>= \case
-            Nothing -> do
-              logError "onServerCertificate didn't fire or failed to get cert chain"
-              closeTLS tls >> error "onServerCertificate failed"
-            Just c -> pure c
-        getClientConnection tCfg chain tls
+    -- No TLS timeout to avoid failing connections via SOCKS
+    tls <- connectTLS (Just hostName) tCfg clientParams sock
+    chain <-
+      atomically (tryTakeTMVar serverCert) >>= \case
+        Nothing -> do
+          logError "onServerCertificate didn't fire or failed to get cert chain"
+          closeTLS tls >> error "onServerCertificate failed"
+        Just c -> pure c
+    getClientConnection tCfg chain tls
  client c `E.finally` closeConnection c
  where
    hostAddr = \case
@@ -27,9 +27,9 @@ import GHC.Stack (withFrozenCallStack)
 import Network.HTTP.Types (urlEncode)
 import SMPAgentClient
 import SMPClient (testKeyHash, testPort, testPort2, testStoreLogFile, withSmpServer, withSmpServerStoreLogOn)
-import Simplex.Messaging.Agent.Protocol hiding (MID, CONF, INFO, REQ)
+import Simplex.Messaging.Agent.Protocol hiding (CONF, INFO, MID, REQ)
 import qualified Simplex.Messaging.Agent.Protocol as A
-import Simplex.Messaging.Crypto.Ratchet (InitialKeys (..), PQEncryption (..), PQSupport (..), pattern IKPQOn, pattern IKPQOff, pattern PQEncOn, pattern PQSupportOn, pattern PQSupportOff)
+import Simplex.Messaging.Crypto.Ratchet (InitialKeys (..), PQEncryption (..), PQSupport (..), pattern IKPQOff, pattern IKPQOn, pattern PQEncOn, pattern PQSupportOff, pattern PQSupportOn)
 import qualified Simplex.Messaging.Crypto.Ratchet as CR
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (ErrorType (..))
@@ -242,7 +242,7 @@ testDuplexConnection' (alice, aPQ) (bob, bPQ) = do
  alice #: ("4a", "bob", "ACK 7") #> ("4a", "bob", OK)
  alice #: ("5", "bob", "OFF") #> ("5", "bob", OK)
  bob #: ("17", "alice", "SEND F 9\nmessage 3") #> ("17", "alice", A.MID 8 pq)
-  bob <# ("", "alice", MERR 8 (SMP AUTH))
+  bob <#= \case ("", "alice", MERR 8 (SMP _ AUTH)) -> True; _ -> False
  alice #: ("6", "bob", "DEL") #> ("6", "bob", OK)
  alice #:# "nothing else should be delivered to alice"

@@ -280,7 +280,7 @@ testDuplexConnRandomIds' (alice, aPQ) (bob, bPQ) = do
  alice #: ("4a", bobConn, "ACK 7") #> ("4a", bobConn, OK)
  alice #: ("5", bobConn, "OFF") #> ("5", bobConn, OK)
  bob #: ("17", aliceConn, "SEND F 9\nmessage 3") #> ("17", aliceConn, A.MID 8 pq)
-  bob <# ("", aliceConn, MERR 8 (SMP AUTH))
+  bob <#= \case ("", cId, MERR 8 (SMP _ AUTH)) -> cId == aliceConn; _ -> False
  alice #: ("6", bobConn, "DEL") #> ("6", bobConn, OK)
  alice #:# "nothing else should be delivered to alice"

@@ -383,7 +383,7 @@ testSubscrNotification t (server, _) client = do
  killThread server
  client <#. ("", "", DOWN testSMPServer ["conn1"])
  withSmpServer (ATransport t) $
-    client <# ("", "conn1", ERR (SMP AUTH)) -- this new server does not have the queue
+    client <#= \case ("", "conn1", ERR (SMP _ AUTH)) -> True; _ -> False -- this new server does not have the queue

 testMsgDeliveryServerRestart :: forall c. Transport c => (c, InitialKeys) -> (c, PQSupport) -> IO ()
 testMsgDeliveryServerRestart (alice, aPQ) (bob, bPQ) = do
@@ -547,10 +547,10 @@ testResumeDeliveryQuotaExceeded _ alice bob = do
  bob <#= \case ("", "alice", Msg "message 4") -> True; _ -> False
  bob #: ("4", "alice", "ACK 7") #> ("4", "alice", OK)
  inAnyOrder
-      (tGetAgent alice)
-      [ \case ("", c, Right (SENT 8)) -> c == "bob"; _ -> False,
-        \case ("", c, Right QCONT) -> c == "bob"; _ -> False
-      ]
+    (tGetAgent alice)
+    [ \case ("", c, Right (SENT 8)) -> c == "bob"; _ -> False,
+      \case ("", c, Right QCONT) -> c == "bob"; _ -> False
+    ]
  bob <#= \case ("", "alice", Msg "over quota") -> True; _ -> False
  -- message 8 is skipped because of alice agent sending "QCONT" message
  bob #: ("5", "alice", "ACK 9") #> ("5", "alice", OK)
@@ -580,7 +580,7 @@ enableKEMStr _ = ""

 pqConnModeStr :: InitialKeys -> ByteString
 pqConnModeStr (IKNoPQ PQSupportOff) = ""
-pqConnModeStr pq =  " " <> strEncode pq
+pqConnModeStr pq = " " <> strEncode pq

 sendMessage :: Transport c => (c, ConnId) -> (c, ConnId) -> ByteString -> IO ()
 sendMessage (h1, name1) (h2, name2) msg = do
@@ -630,7 +630,7 @@ syntaxTests t = do
            <> " subscribe "
            <> "14\nbob's connInfo"
        )
-          >#> ("311", "a", "ERR SMP AUTH")
+          >#> ("311", "a", "ERR SMP smp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:5001 AUTH")
    describe "invalid" $ do
      it "no parameters" $ ("321", "", "JOIN") >#> ("321", "", "ERR CMD SYNTAX")
  where
@@ -244,7 +244,7 @@ createConnection :: AgentClient -> UserId -> Bool -> SConnectionMode c -> Maybe
 createConnection c userId enableNtfs cMode clientData = A.createConnection c userId enableNtfs cMode clientData (IKNoPQ PQSupportOn)

 joinConnection :: AgentClient -> UserId -> Bool -> ConnectionRequestUri c -> ConnInfo -> SubscriptionMode -> AE ConnId
-joinConnection c userId enableNtfs cReq connInfo = A.joinConnection c userId enableNtfs cReq connInfo PQSupportOn
+joinConnection c userId enableNtfs cReq connInfo = A.joinConnection c userId Nothing enableNtfs cReq connInfo PQSupportOn

 sendMessage :: AgentClient -> ConnId -> SMP.MsgFlags -> MsgBody -> AE AgentMsgId
 sendMessage c connId msgFlags msgBody = do
@@ -415,7 +415,7 @@ functionalAPITests t = do
    describe "server with password" $ do
      let auth = Just "abcd"
          srv = ProtoServerWithAuth testSMPServer2
-          authErr = Just (ProtocolTestFailure TSCreateQueue $ SMP AUTH)
+          authErr = Just (ProtocolTestFailure TSCreateQueue $ SMP (B.unpack $ strEncode testSMPServer2) AUTH)
      it "should pass with correct password" $ testSMPServerConnectionTest t auth (srv auth) `shouldReturn` Nothing
      it "should fail without password" $ testSMPServerConnectionTest t auth (srv Nothing) `shouldReturn` authErr
      it "should fail with incorrect password" $ testSMPServerConnectionTest t auth (srv $ Just "wrong") `shouldReturn` authErr
@@ -503,7 +503,7 @@ runAgentClientTest :: HasCallStack => PQSupport -> AgentClient -> AgentClient ->
 runAgentClientTest pqSupport alice@AgentClient {} bob baseId =
  runRight_ $ do
    (bobId, qInfo) <- A.createConnection alice 1 True SCMInvitation Nothing (IKNoPQ pqSupport) SMSubscribe
-    aliceId <- A.joinConnection bob 1 True qInfo "bob's connInfo" pqSupport SMSubscribe
+    aliceId <- A.joinConnection bob 1 Nothing True qInfo "bob's connInfo" pqSupport SMSubscribe
    ("", _, A.CONF confId pqSup' _ "bob's connInfo") <- get alice
    liftIO $ pqSup' `shouldBe` pqSupport
    allowConnection alice bobId confId "alice's connInfo"
@@ -530,7 +530,7 @@ runAgentClientTest pqSupport alice@AgentClient {} bob baseId =
    ackMessage alice bobId (baseId + 4) Nothing
    suspendConnection alice bobId
    5 <- msgId <$> A.sendMessage bob aliceId pqEnc SMP.noMsgFlags "message 2"
-    get bob ##> ("", aliceId, MERR (baseId + 5) (SMP AUTH))
+    get bob =##> \case ("", cId, MERR mId (SMP _ AUTH)) -> cId == aliceId && mId == (baseId + 5); _ -> False
    deleteConnection alice bobId
    liftIO $ noMessages alice "nothing else should be delivered to alice"
  where
@@ -630,7 +630,9 @@ runAgentClientContactTest :: HasCallStack => PQSupport -> AgentClient -> AgentCl
 runAgentClientContactTest pqSupport alice bob baseId =
  runRight_ $ do
    (_, qInfo) <- A.createConnection alice 1 True SCMContact Nothing (IKNoPQ pqSupport) SMSubscribe
-    aliceId <- A.joinConnection bob 1 True qInfo "bob's connInfo" pqSupport SMSubscribe
+    aliceId <- A.prepareConnectionToJoin bob 1 True qInfo pqSupport
+    aliceId' <- A.joinConnection bob 1 (Just aliceId) True qInfo "bob's connInfo" pqSupport SMSubscribe
+    liftIO $ aliceId' `shouldBe` aliceId
    ("", _, A.REQ invId pqSup' _ "bob's connInfo") <- get alice
    liftIO $ pqSup' `shouldBe` pqSupport
    bobId <- acceptContact alice True invId "alice's connInfo" PQSupportOn SMSubscribe
@@ -660,7 +662,7 @@ runAgentClientContactTest pqSupport alice bob baseId =
    ackMessage alice bobId (baseId + 4) Nothing
    suspendConnection alice bobId
    5 <- msgId <$> A.sendMessage bob aliceId pqEnc SMP.noMsgFlags "message 2"
-    get bob ##> ("", aliceId, MERR (baseId + 5) (SMP AUTH))
+    get bob =##> \case ("", cId, MERR mId (SMP _ AUTH)) -> cId == aliceId && mId == (baseId + 5); _ -> False
    deleteConnection alice bobId
    liftIO $ noMessages alice "nothing else should be delivered to alice"
  where
@@ -1106,7 +1108,7 @@ testExpireMessageQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1} testP
    5 <- sendMessage a bId SMP.noMsgFlags "2"
    liftIO $ threadDelay 1000000
    6 <- sendMessage a bId SMP.noMsgFlags "3" -- this won't expire
-    get a =##> \case ("", c, MERR 5 (SMP QUOTA)) -> bId == c; _ -> False
+    get a =##> \case ("", c, MERR 5 (SMP _ QUOTA)) -> bId == c; _ -> False
    pure (aId, bId)
  withAgent 3 agentCfg initAgentServers testDB2 $ \b' -> runRight_ $ do
    subscribeConnection b' aId
@@ -1134,15 +1136,15 @@ testExpireManyMessagesQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1}
    7 <- sendMessage a bId SMP.noMsgFlags "4"
    liftIO $ threadDelay 1000000
    8 <- sendMessage a bId SMP.noMsgFlags "5" -- this won't expire
-    get a =##> \case ("", c, MERR 5 (SMP QUOTA)) -> bId == c; _ -> False
+    get a =##> \case ("", c, MERR 5 (SMP _ QUOTA)) -> bId == c; _ -> False
    get a >>= \case
-      ("", c, MERR 6 (SMP QUOTA)) -> do
+      ("", c, MERR 6 (SMP _ QUOTA)) -> do
        liftIO $ bId `shouldBe` c
-        get a =##> \case ("", c', MERR 7 (SMP QUOTA)) -> bId == c'; ("", c', MERRS [7] (SMP QUOTA)) -> bId == c'; _ -> False
-      ("", c, MERRS [6] (SMP QUOTA)) -> do
+        get a =##> \case ("", c', MERR 7 (SMP _ QUOTA)) -> bId == c'; ("", c', MERRS [7] (SMP _ QUOTA)) -> bId == c'; _ -> False
+      ("", c, MERRS [6] (SMP _ QUOTA)) -> do
        liftIO $ bId `shouldBe` c
-        get a =##> \case ("", c', MERR 7 (SMP QUOTA)) -> bId == c'; _ -> False
-      ("", c, MERRS [6, 7] (SMP QUOTA)) -> liftIO $ bId `shouldBe` c
+        get a =##> \case ("", c', MERR 7 (SMP _ QUOTA)) -> bId == c'; _ -> False
+      ("", c, MERRS [6, 7] (SMP _ QUOTA)) -> liftIO $ bId `shouldBe` c
      r -> error $ show r
    pure (aId, bId)
  withAgent 3 agentCfg initAgentServers testDB2 $ \b' -> runRight_ $ do
@@ -1393,13 +1395,15 @@ makeConnection = makeConnection_ PQSupportOn
 makeConnection_ :: PQSupport -> AgentClient -> AgentClient -> ExceptT AgentErrorType IO (ConnId, ConnId)
 makeConnection_ pqEnc alice bob = makeConnectionForUsers_ pqEnc alice 1 bob 1

-makeConnectionForUsers :: AgentClient -> UserId -> AgentClient -> UserId -> ExceptT AgentErrorType IO (ConnId, ConnId)
+makeConnectionForUsers :: HasCallStack => AgentClient -> UserId -> AgentClient -> UserId -> ExceptT AgentErrorType IO (ConnId, ConnId)
 makeConnectionForUsers = makeConnectionForUsers_ PQSupportOn

-makeConnectionForUsers_ :: PQSupport -> AgentClient -> UserId -> AgentClient -> UserId -> ExceptT AgentErrorType IO (ConnId, ConnId)
+makeConnectionForUsers_ :: HasCallStack => PQSupport -> AgentClient -> UserId -> AgentClient -> UserId -> ExceptT AgentErrorType IO (ConnId, ConnId)
 makeConnectionForUsers_ pqSupport alice aliceUserId bob bobUserId = do
  (bobId, qInfo) <- A.createConnection alice aliceUserId True SCMInvitation Nothing (CR.IKNoPQ pqSupport) SMSubscribe
-  aliceId <- A.joinConnection bob bobUserId True qInfo "bob's connInfo" pqSupport SMSubscribe
+  aliceId <- A.prepareConnectionToJoin bob bobUserId True qInfo pqSupport
+  aliceId' <- A.joinConnection bob bobUserId (Just aliceId) True qInfo "bob's connInfo" pqSupport SMSubscribe
+  liftIO $ aliceId' `shouldBe` aliceId
  ("", _, A.CONF confId pqSup' _ "bob's connInfo") <- get alice
  liftIO $ pqSup' `shouldBe` pqSupport
  allowConnection alice bobId confId "alice's connInfo"
@@ -1679,7 +1683,7 @@ testAcceptContactAsync =
    ackMessage alice bobId (baseId + 4) Nothing
    suspendConnection alice bobId
    5 <- msgId <$> sendMessage bob aliceId SMP.noMsgFlags "message 2"
-    get bob ##> ("", aliceId, MERR (baseId + 5) (SMP AUTH))
+    get bob =##> \case ("", cId, MERR mId (SMP _ AUTH)) -> cId == aliceId && mId == (baseId + 5); _ -> False
    deleteConnection alice bobId
    liftIO $ noMessages alice "nothing else should be delivered to alice"
  where
@@ -1725,7 +1729,7 @@ testWaitDeliveryNoPending t = withAgentClients2 $ \alice bob ->
    get alice =##> \case ("", cId, DEL_CONN) -> cId == bobId; _ -> False

    3 <- msgId <$> sendMessage bob aliceId SMP.noMsgFlags "message 2"
-    get bob ##> ("", aliceId, MERR (baseId + 3) (SMP AUTH))
+    get bob =##> \case ("", cId, MERR mId (SMP _ AUTH)) -> cId == aliceId && mId == (baseId + 3); _ -> False

    liftIO $ noMessages alice "nothing else should be delivered to alice"
    liftIO $ noMessages bob "nothing else should be delivered to bob"
@@ -1820,8 +1824,8 @@ testWaitDeliveryAUTHErr t =
        liftIO $ noMessages bob "nothing else should be delivered to bob"

      withSmpServerStoreLogOn t testPort $ \_ -> do
-        get alice ##> ("", bobId, MERR (baseId + 3) (SMP AUTH))
-        get alice ##> ("", bobId, MERR (baseId + 4) (SMP AUTH))
+        get alice =##> \case ("", cId, MERR mId (SMP _ AUTH)) -> cId == bobId && mId == (baseId + 3); _ -> False
+        get alice =##> \case ("", cId, MERR mId (SMP _ AUTH)) -> cId == bobId && mId == (baseId + 4); _ -> False
        get alice =##> \case ("", cId, DEL_CONN) -> cId == bobId; _ -> False

        liftIO $ noMessages alice "nothing else should be delivered to alice"
@@ -2392,11 +2396,11 @@ testCreateQueueAuth srvVersion clnt1 clnt2 = do
  b <- getClient 2 clnt2 testDB2
  r <- runRight $ do
    tryError (createConnection a 1 True SCMInvitation Nothing SMSubscribe) >>= \case
-      Left (SMP AUTH) -> pure 0
+      Left (SMP _ AUTH) -> pure 0
      Left e -> throwError e
      Right (bId, qInfo) ->
        tryError (joinConnection b 1 True qInfo "bob's connInfo" SMSubscribe) >>= \case
-          Left (SMP AUTH) -> pure 1
+          Left (SMP _ AUTH) -> pure 1
          Left e -> throwError e
          Right aId -> do
            ("", _, CONF confId _ "bob's connInfo") <- get a
@@ -6,8 +6,8 @@
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE PatternSynonyms #-}
 {-# LANGUAGE TypeApplications #-}
-{-# OPTIONS_GHC -fno-warn-incomplete-uni-patterns #-}
 {-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
+{-# OPTIONS_GHC -fno-warn-incomplete-uni-patterns #-}

 module AgentTests.NotificationTests where

@@ -17,10 +17,6 @@ import AgentTests.FunctionalAPITests
    createConnection,
    exchangeGreetingsMsgId,
    get,
-    withAgent,
-    withAgentClients2,
-    withAgentClientsCfgServers2,
-    withAgentClients3,
    joinConnection,
    makeConnection,
    nGet,
@@ -29,7 +25,11 @@ import AgentTests.FunctionalAPITests
    sendMessage,
    switchComplete,
    testServerMatrix2,
+    withAgent,
+    withAgentClients2,
+    withAgentClients3,
    withAgentClientsCfg2,
+    withAgentClientsCfgServers2,
    (##>),
    (=##>),
    pattern CON,
@@ -59,8 +59,8 @@ import Simplex.Messaging.Agent.Protocol hiding (CON, CONF, INFO)
 import Simplex.Messaging.Agent.Store.SQLite (getSavedNtfToken)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
-import Simplex.Messaging.Notifications.Server.Env (NtfServerConfig (..))
 import Simplex.Messaging.Notifications.Protocol
+import Simplex.Messaging.Notifications.Server.Env (NtfServerConfig (..))
 import Simplex.Messaging.Notifications.Server.Push.APNS
 import Simplex.Messaging.Notifications.Types (NtfToken (..))
 import Simplex.Messaging.Protocol (ErrorType (AUTH), MsgFlags (MsgFlags), NtfServer, ProtocolServer (..), SMPMsgMeta (..), SubscriptionMode (..))
@@ -151,7 +151,8 @@ testNtfMatrix t runTest = do
    it "next servers: SMP v7, NTF v2; curr clients: v6/v1" $ runNtfTestCfg t cfgV7 ntfServerCfgV2 agentCfg agentCfg runTest
    it "curr servers: SMP v6, NTF v1; curr clients: v6/v1" $ runNtfTestCfg t cfg ntfServerCfg agentCfg agentCfg runTest
    skip "this case cannot be supported - see RFC" $
-      it "servers: SMP v6, NTF v1; clients: v7/v2 (not supported)" $ runNtfTestCfg t cfg ntfServerCfg agentCfgV7 agentCfgV7 runTest
+      it "servers: SMP v6, NTF v1; clients: v7/v2 (not supported)" $
+        runNtfTestCfg t cfg ntfServerCfg agentCfgV7 agentCfgV7 runTest
    -- servers can be migrated in any order
    it "servers: next SMP v7, curr NTF v1; curr clients: v6/v1" $ runNtfTestCfg t cfgV7 ntfServerCfg agentCfg agentCfg runTest
    it "servers: curr SMP v6, next NTF v2; curr clients: v6/v1" $ runNtfTestCfg t cfg ntfServerCfgV2 agentCfg agentCfg runTest
@@ -243,7 +244,7 @@ testNtfTokenSecondRegistration APNSMockServer {apnsQ} =
    -- now the second token registration is verified
    verifyNtfToken a' tkn nonce' verification'
    -- the first registration is removed
-    Left (NTF AUTH) <- tryE $ checkNtfToken a tkn
+    Left (NTF _ AUTH) <- tryE $ checkNtfToken a tkn
    -- and the second is active
    NTActive <- checkNtfToken a' tkn
    pure ()
@@ -258,7 +259,7 @@ testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
        atomically $ readTBQueue apnsQ
      liftIO $ sendApnsResponse APNSRespOk
      pure ntfData
-    -- the new agent is created as otherwise when running the tests in CI the old agent was keeping the connection to the server
+  -- the new agent is created as otherwise when running the tests in CI the old agent was keeping the connection to the server
  threadDelay 1000000
  withAgent 2 agentCfg initAgentServers testDB $ \a' ->
    -- server stopped before token is verified, so now the attempt to verify it will return AUTH error but re-register token,
@@ -266,7 +267,7 @@ testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
    withNtfServer t . runRight_ $ do
      verification <- ntfData .-> "verification"
      nonce <- C.cbNonce <$> ntfData .-> "nonce"
-      Left (NTF AUTH) <- tryE $ verifyNtfToken a' tkn nonce verification
+      Left (NTF _ AUTH) <- tryE $ verifyNtfToken a' tkn nonce verification
      APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData'}, sendApnsResponse = sendApnsResponse'} <-
        atomically $ readTBQueue apnsQ
      verification' <- ntfData' .-> "verification"
@@ -35,6 +35,9 @@ protocolErrorTests = modifyMaxSuccess (const 1000) $ do
    errHasSpaces = \case
      BROKER srv (RESPONSE e) -> hasSpaces srv || hasSpaces e
      BROKER srv _ -> hasSpaces srv
+      SMP srv _ -> hasSpaces srv
+      NTF srv _ -> hasSpaces srv
+      XFTP srv _ -> hasSpaces srv
      _ -> False
    hasSpaces s = ' ' `B.elem` encodeUtf8 (T.pack s)

@@ -22,10 +22,13 @@ tRcvQueuesTests = do
  describe "connection API" $ do
    it "hasConn" hasConnTest
    it "hasConn, batch add" hasConnTestBatch
+    it "hasConn, batch idempotent" batchIdempotentTest
    it "deleteConn" deleteConnTest
  describe "session API" $ do
    it "getSessQueues" getSessQueuesTest
    it "getDelSessQueues" getDelSessQueuesTest
+  describe "queue transfer" $ do
+    it "getDelSessQueues-batchAddQueues preserves total length" removeSubsTest

 checkDataInvariant :: RQ.TRcvQueues -> IO Bool
 checkDataInvariant trq = atomically $ do
@@ -62,6 +65,19 @@ hasConnTestBatch = do
  atomically (RQ.hasConn "c3" trq) `shouldReturn` True
  atomically (RQ.hasConn "nope" trq) `shouldReturn` False

+batchIdempotentTest :: IO ()
+batchIdempotentTest = do
+  trq <- atomically RQ.empty
+  let qs = [dummyRQ 0 "smp://1234-w==@alpha" "c1", dummyRQ 0 "smp://1234-w==@alpha" "c2", dummyRQ 0 "smp://1234-w==@beta" "c3"]
+  atomically $ RQ.batchAddQueues trq qs
+  checkDataInvariant trq `shouldReturn` True
+  qs' <- readTVarIO $ RQ.getRcvQueues trq
+  cs' <- readTVarIO $ RQ.getConnections trq
+  atomically $ RQ.batchAddQueues trq qs
+  checkDataInvariant trq `shouldReturn` True
+  readTVarIO (RQ.getRcvQueues trq) `shouldReturn` qs'
+  fmap L.nub <$> readTVarIO (RQ.getConnections trq) `shouldReturn`cs' -- connections get duplicated, but that doesn't appear to affect anybody
+
 deleteConnTest :: IO ()
 deleteConnTest = do
  trq <- atomically RQ.empty
@@ -121,6 +137,40 @@ getDelSessQueuesTest = do
  atomically (RQ.hasConn "c3" trq) `shouldReturn` True
  atomically (RQ.hasConn "c4" trq) `shouldReturn` True

+removeSubsTest :: IO ()
+removeSubsTest = do
+  aq <- atomically RQ.empty
+  let qs =
+        [ dummyRQ 0 "smp://1234-w==@alpha" "c1",
+          dummyRQ 0 "smp://1234-w==@alpha" "c2",
+          dummyRQ 0 "smp://1234-w==@beta" "c3",
+          dummyRQ 1 "smp://1234-w==@beta" "c4"
+        ]
+  atomically $ RQ.batchAddQueues aq qs
+
+  pq <- atomically RQ.empty
+  atomically (totalSize aq pq) `shouldReturn` (4, 4)
+
+  atomically $ RQ.getDelSessQueues (0, "smp://1234-w==@alpha", Nothing) aq >>= RQ.batchAddQueues pq . fst
+  atomically (totalSize aq pq) `shouldReturn` (4, 4)
+
+  atomically $ RQ.getDelSessQueues (0, "smp://1234-w==@beta", Just "non-existent") aq >>= RQ.batchAddQueues pq . fst
+  atomically (totalSize aq pq) `shouldReturn` (4, 4)
+
+  atomically $ RQ.getDelSessQueues (0, "smp://1234-w==@localhost", Nothing) aq >>= RQ.batchAddQueues pq . fst
+  atomically (totalSize aq pq) `shouldReturn` (4, 4)
+
+  atomically $ RQ.getDelSessQueues (0, "smp://1234-w==@beta", Just "c3") aq >>= RQ.batchAddQueues pq . fst
+  atomically (totalSize aq pq) `shouldReturn` (4, 4)
+
+totalSize :: RQ.TRcvQueues -> RQ.TRcvQueues -> STM (Int, Int)
+totalSize a b = do
+  qsizeA <- M.size <$> readTVar (RQ.getRcvQueues a)
+  qsizeB <- M.size <$> readTVar (RQ.getRcvQueues b)
+  csizeA <- M.size <$> readTVar (RQ.getConnections a)
+  csizeB <- M.size <$> readTVar (RQ.getConnections b)
+  pure (qsizeA + qsizeB, csizeA + csizeB)
+
 dummyRQ :: UserId -> SMPServer -> ConnId -> RcvQueue
 dummyRQ userId server connId =
  RcvQueue
@@ -69,7 +69,7 @@ xftpAgentTests = around_ testBracket . describe "agent XFTP API" $ do
    describe "server with password" $ do
      let auth = Just "abcd"
          srv = ProtoServerWithAuth testXFTPServer2
-          authErr = Just (ProtocolTestFailure TSCreateFile $ XFTP AUTH)
+          authErr = Just (ProtocolTestFailure TSCreateFile $ XFTP (B.unpack $ strEncode testXFTPServer2) AUTH)
      it "should pass with correct password" $ testXFTPServerTest auth (srv auth) `shouldReturn` Nothing
      it "should fail without password" $ testXFTPServerTest auth (srv Nothing) `shouldReturn` authErr
      it "should fail with incorrect password" $ testXFTPServerTest auth (srv $ Just "wrong") `shouldReturn` authErr
@@ -392,7 +392,8 @@ testXFTPAgentReceiveCleanup = withGlobalLogging logCfgNoLogs $ do
    -- receive file - should fail with AUTH error
    withAgent 3 agentCfg initAgentServers testDB2 $ \rcp' -> do
      runRight_ $ xftpStartWorkers rcp' (Just recipientFiles)
-      ("", rfId', RFERR (INTERNAL "XFTP {xftpErr = AUTH}")) <- rfGet rcp'
+      ("", rfId', RFERR (INTERNAL "XFTP {serverAddress = \"xftp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:7000\", xftpErr = AUTH}")) <-
+        rfGet rcp'
      rfId' `shouldBe` rfId

  -- tmp path should be removed after permanent error
@@ -471,7 +472,8 @@ testXFTPAgentSendCleanup = withGlobalLogging logCfgNoLogs $ do
    -- send file - should fail with AUTH error
    withAgent 2 agentCfg initAgentServers testDB $ \sndr' -> do
      runRight_ $ xftpStartWorkers sndr' (Just senderFiles)
-      ("", sfId', SFERR (INTERNAL "XFTP {xftpErr = AUTH}")) <- sfGet sndr'
+      ("", sfId', SFERR (INTERNAL "XFTP {serverAddress = \"xftp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:7000\", xftpErr = AUTH}")) <-
+        sfGet sndr'
      sfId' `shouldBe` sfId

  -- prefix path should be removed after permanent error
@@ -506,7 +508,8 @@ testXFTPAgentDelete = withGlobalLogging logCfgNoLogs $
      withAgent 3 agentCfg initAgentServers testDB2 $ \rcp2 -> runRight $ do
        xftpStartWorkers rcp2 (Just recipientFiles)
        rfId <- xftpReceiveFile rcp2 1 rfd2 Nothing
-        ("", rfId', RFERR (INTERNAL "XFTP {xftpErr = AUTH}")) <- rfGet rcp2
+        ("", rfId', RFERR (INTERNAL "XFTP {serverAddress = \"xftp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:7000\", xftpErr = AUTH}")) <-
+          rfGet rcp2
        liftIO $ rfId' `shouldBe` rfId

 testXFTPAgentDeleteRestore :: HasCallStack => IO ()
@@ -543,7 +546,8 @@ testXFTPAgentDeleteRestore = withGlobalLogging logCfgNoLogs $ do
      withAgent 5 agentCfg initAgentServers testDB3 $ \rcp2 -> runRight $ do
        xftpStartWorkers rcp2 (Just recipientFiles)
        rfId <- xftpReceiveFile rcp2 1 rfd2 Nothing
-        ("", rfId', RFERR (INTERNAL "XFTP {xftpErr = AUTH}")) <- rfGet rcp2
+        ("", rfId', RFERR (INTERNAL "XFTP {serverAddress = \"xftp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:7000\", xftpErr = AUTH}")) <-
+          rfGet rcp2
        liftIO $ rfId' `shouldBe` rfId

 testXFTPAgentDeleteOnServer :: HasCallStack => IO ()
@@ -577,7 +581,8 @@ testXFTPAgentDeleteOnServer = withGlobalLogging logCfgNoLogs $
        runRight_ . void $ do
          -- receive file 1 again
          rfId1 <- xftpReceiveFile rcp 1 rfd1_2 Nothing
-          ("", rfId1', RFERR (INTERNAL "XFTP {xftpErr = AUTH}")) <- rfGet rcp
+          ("", rfId1', RFERR (INTERNAL "XFTP {serverAddress = \"xftp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:7000\", xftpErr = AUTH}")) <-
+            rfGet rcp
          liftIO $ rfId1 `shouldBe` rfId1'

          -- receive file 2
@@ -609,7 +614,8 @@ testXFTPAgentExpiredOnServer = withGlobalLogging logCfgNoLogs $ do
        -- receive file 1 again - should fail with AUTH error
        runRight $ do
          rfId <- xftpReceiveFile rcp 1 rfd1_2 Nothing
-          ("", rfId', RFERR (INTERNAL "XFTP {xftpErr = AUTH}")) <- rfGet rcp
+          ("", rfId', RFERR (INTERNAL "XFTP {serverAddress = \"xftp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:7000\", xftpErr = AUTH}")) <-
+            rfGet rcp
          liftIO $ rfId' `shouldBe` rfId

        -- create and send file 2
Author	SHA1	Message	Date
IC Rainbow	757742f5ed	show client sub status	2024-05-06 20:52:54 +03:00
IC Rainbow	d1b3b5ff54	remove from client subs when removing conn	2024-05-06 20:21:19 +03:00
Alexander Bondarenko	f2af239c30	remove print	2024-05-06 19:33:54 +03:00
Alexander Bondarenko	1c73ac638f	diff against agent subs	2024-05-06 18:40:28 +03:00
IC Rainbow	29c673dd64	Add subs to ProtocolClient	2024-05-06 13:20:42 +03:00
Evgeny Poberezkin	fe8e8366c8	Merge branch 'master' into ep/sessions-subs-2	2024-05-05 21:25:49 +01:00
Evgeny Poberezkin	93fd424f86	5.7.2.0	2024-05-05 17:13:26 +01:00
Evgeny Poberezkin	e13b0df539	client: remove TLS handshake timeout (#1129 ) * client: remove TLS handshake timeout * remove comment	2024-05-05 17:05:51 +01:00
Evgeny Poberezkin	5f740cd1c0	Merge branch 'master' into ep/sessions-subs-2	2024-05-05 12:15:06 +01:00
Evgeny Poberezkin	ee8e4067b0	agent: prepare connection record before joining to prevent race conditions (#1128 ) * agent: prepare connection record before joining to prevent race conditions * prepare connection for contact address as well * clean up	2024-05-05 12:12:19 +01:00
Evgeny Poberezkin	443d260e73	add pending connection	2024-05-04 20:33:48 +01:00
Evgeny Poberezkin	4b5a88f407	track subscription responses better	2024-05-04 17:12:02 +01:00
Evgeny Poberezkin	6ee7082269	fixes	2024-05-04 09:57:25 +01:00
Evgeny Poberezkin	4310a69391	revert not expiring sending subs	2024-05-04 09:38:01 +01:00
Evgeny Poberezkin	ca4a7b561e	do not forward agent error	2024-05-04 00:19:09 +01:00
Evgeny Poberezkin	acaa5d1d9f	Merge branch 'master' into ep/old-sessions-subs	2024-05-03 23:39:50 +01:00
Alexander BondarenkoandEvgeny Poberezkin	0e205e70ad	add TRcvQueues tests (#1117 ) Co-authored-by: Evgeny Poberezkin <[email protected]>	2024-05-03 23:39:00 +01:00
Alexander BondarenkoandEvgeny Poberezkin	1104b44136	client: prevent sub actions from zombie sessions (#1122 ) * client: prevent sub actions from zombie sessions * error handling * add AERR to pass background errors to client * switch to activeClientSession * put closeClient under activeClientSession * rename * remove AERR, do not skip processing * move check and state update to one transaction * catch extra UPs * fix * check queue is still pending before making it active --------- Co-authored-by: Evgeny Poberezkin <[email protected]>	2024-05-03 23:29:19 +01:00
Evgeny Poberezkin	82bf52f79e	Merge branch 'master' into ep/old-sessions-subs	2024-05-03 23:21:06 +01:00
Evgeny Poberezkin	b586a6e90a	client: removed concurrency limit when waiting for subscription results (#1126 )	2024-05-03 22:16:52 +01:00