WIP

5.7.4.0
agent: remove critical error on subscription timeout (#1146 )
2026-10-12 11:39:56 +00:00 · 2024-05-12 21:19:27 +03:00 · 2024-05-11 23:39:28 +01:00 · 2024-05-11 23:38:27 +01:00 · 2024-05-10 22:24:00 +01:00 · 2024-05-10 22:19:11 +01:00
39 changed files with 444 additions and 476 deletions
@@ -1,3 +1,18 @@
+# 5.7.3
+
+SMP/NTF protocol:
+- add ALPN for handshake version negotiation, similar to XFTP (to preserve backwards compatibility with the old clients).
+- upgrade clients to versions v7/v2 of the protocols.
+
+SMP server:
+- faster responses to subscription requests.
+
+XFTP client:
+- fix network exception during file download treated as permanent file error.
+
+SMP agent:
+- do not report subscription timeouts while client is offline.
+
 # 5.7.2

 SMP agent:
@@ -1,5 +1,5 @@
 name: simplexmq
-version: 5.7.2.0
+version: 5.7.4.0
 synopsis: SimpleXMQ message broker
 description: |
  This package includes <./docs/Simplex-Messaging-Server.html server>,
@@ -71,6 +71,7 @@ dependencies:
  - transformers == 0.6.*
  - unliftio == 0.2.*
  - unliftio-core == 0.2.*
+  - unordered-containers
  - websockets == 0.12.*
  - yaml == 0.11.*
  - zstd == 0.1.3.*
@@ -180,7 +181,6 @@ 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.2.0
+version:        5.7.4.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 -Werror=missing-methods -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2
+  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2
  include-dirs:
      cbits
  c-sources:
@@ -230,6 +230,7 @@ library
    , transformers ==0.6.*
    , unliftio ==0.2.*
    , unliftio-core ==0.2.*
+    , unordered-containers
    , websockets ==0.12.*
    , yaml ==0.11.*
    , zstd ==0.1.3.*
@@ -255,7 +256,7 @@ executable ntf-server
      apps/ntf-server
  default-extensions:
      StrictData
-  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Werror=missing-methods -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
+  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -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
@@ -305,6 +306,7 @@ executable ntf-server
    , transformers ==0.6.*
    , unliftio ==0.2.*
    , unliftio-core ==0.2.*
+    , unordered-containers
    , websockets ==0.12.*
    , yaml ==0.11.*
    , zstd ==0.1.3.*
@@ -330,7 +332,7 @@ executable smp-agent
      apps/smp-agent
  default-extensions:
      StrictData
-  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Werror=missing-methods -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
+  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -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
@@ -380,6 +382,7 @@ executable smp-agent
    , transformers ==0.6.*
    , unliftio ==0.2.*
    , unliftio-core ==0.2.*
+    , unordered-containers
    , websockets ==0.12.*
    , yaml ==0.11.*
    , zstd ==0.1.3.*
@@ -405,7 +408,7 @@ executable smp-server
      apps/smp-server
  default-extensions:
      StrictData
-  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Werror=missing-methods -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
+  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -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
@@ -455,6 +458,7 @@ executable smp-server
    , transformers ==0.6.*
    , unliftio ==0.2.*
    , unliftio-core ==0.2.*
+    , unordered-containers
    , websockets ==0.12.*
    , yaml ==0.11.*
    , zstd ==0.1.3.*
@@ -480,7 +484,7 @@ executable xftp
      apps/xftp
  default-extensions:
      StrictData
-  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Werror=missing-methods -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
+  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -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
@@ -530,6 +534,7 @@ executable xftp
    , transformers ==0.6.*
    , unliftio ==0.2.*
    , unliftio-core ==0.2.*
+    , unordered-containers
    , websockets ==0.12.*
    , yaml ==0.11.*
    , zstd ==0.1.3.*
@@ -555,7 +560,7 @@ executable xftp-server
      apps/xftp-server
  default-extensions:
      StrictData
-  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Werror=missing-methods -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
+  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -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
@@ -605,6 +610,7 @@ executable xftp-server
    , transformers ==0.6.*
    , unliftio ==0.2.*
    , unliftio-core ==0.2.*
+    , unordered-containers
    , websockets ==0.12.*
    , yaml ==0.11.*
    , zstd ==0.1.3.*
@@ -662,7 +668,7 @@ test-suite simplexmq-test
      tests
  default-extensions:
      StrictData
-  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
+  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
  build-depends:
      HUnit ==1.6.*
    , QuickCheck ==2.14.*
@@ -721,6 +727,7 @@ test-suite simplexmq-test
    , transformers ==0.6.*
    , unliftio ==0.2.*
    , unliftio-core ==0.2.*
+    , unordered-containers
    , websockets ==0.12.*
    , yaml ==0.11.*
    , zstd ==0.1.3.*
@@ -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,11 +281,10 @@ 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
-            -- TODO switch to another error constructor
-            Left _ -> throwError . XFTP "" $ XFTP.REDIRECT "decode error"
+            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
@@ -245,8 +245,13 @@ downloadXFTPChunk g c@XFTPClient {config} rpKey fId chunkSpec@XFTPRcvChunkSpec {
        let dhSecret = C.dh' sDhKey rpDhKey
        cbState <- liftEither . first PCECryptoError $ LC.cbInit dhSecret cbNonce
        let t = chunkTimeout config chunkSize
-        ExceptT (sequence <$> (t `timeout` download cbState)) >>= maybe (throwError PCEResponseTimeout) pure
+        ExceptT (sequence <$> (t `timeout` (download cbState `catches` errors))) >>= maybe (throwError PCEResponseTimeout) pure
        where
+          errors =
+            [ Handler $ \(_e :: H.HTTP2Error) -> pure $ Left PCENetworkError,
+              Handler $ \(e :: IOException) -> pure $ Left (PCEIOError e),
+              Handler $ \(_e :: SomeException) -> pure $ Left PCENetworkError
+            ]
          download cbState =
            runExceptT . withExceptT PCEResponseError $
              receiveEncFile chunkPart cbState chunkSpec `catchError` \e ->
@@ -143,7 +143,6 @@ 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
@@ -13,12 +13,9 @@ import Control.Logger.Simple
 import Control.Monad
 import Control.Monad.IO.Unlift
 import Crypto.Random
-import Data.Default (def)
 import Data.Int (Int64)
-import Data.List (find)
 import Data.List.NonEmpty (NonEmpty)
 import qualified Data.Map.Strict as M
-import Data.Maybe (fromMaybe)
 import Data.Time.Clock (getCurrentTime)
 import Data.Word (Word32)
 import Data.X509.Validation (Fingerprint (..))
@@ -103,7 +100,7 @@ supportedXFTPhandshakes :: [ALPN]
 supportedXFTPhandshakes = ["xftp/1"]

 newXFTPServerEnv :: XFTPServerConfig -> IO XFTPEnv
-newXFTPServerEnv config@XFTPServerConfig {storeLogFile, fileSizeQuota, caCertificateFile, certificateFile, privateKeyFile} = do
+newXFTPServerEnv config@XFTPServerConfig {storeLogFile, fileSizeQuota, caCertificateFile, certificateFile, privateKeyFile, transportConfig} = do
  random <- liftIO C.newRandom
  store <- atomically newFileStore
  storeLog <- liftIO $ mapM (`readWriteFileStore` store) storeLogFile
@@ -112,17 +109,7 @@ newXFTPServerEnv config@XFTPServerConfig {storeLogFile, fileSizeQuota, caCertifi
  forM_ fileSizeQuota $ \quota -> do
    logInfo $ "Total / available storage: " <> tshow quota <> " / " <> tshow (quota - used)
    when (quota < used) $ logInfo "WARNING: storage quota is less than used storage, no files can be uploaded!"
-  tlsServerParams' <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile
-  let TransportServerConfig {alpn} = transportConfig config
-  let tlsServerParams = case alpn of
-        Nothing -> tlsServerParams'
-        Just supported ->
-          tlsServerParams'
-            { T.serverHooks =
-                def
-                  { T.onALPNClientSuggest = Just $ pure . fromMaybe "" . find (`elem` supported)
-                  }
-            }
+  tlsServerParams <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile (alpn transportConfig)
  Fingerprint fp <- liftIO $ loadFingerprint caCertificateFile
  serverStats <- atomically . newFileServerStats =<< liftIO getCurrentTime
  pure XFTPEnv {config, store, storeLog, random, tlsServerParams, serverIdentity = C.KeyHash fp, serverStats}
@@ -118,7 +118,7 @@ module Simplex.Messaging.Agent
  )
 where

-import Control.Logger.Simple
+import Control.Logger.Simple (logError, logInfo, showText)
 import Control.Monad
 import Control.Monad.Except
 import Control.Monad.Reader
@@ -158,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, TransmissionType (..))
+import Simplex.Messaging.Client (ProtocolClient (..), ServerTransmission)
 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)
@@ -734,9 +734,11 @@ newRcvConnSrv c userId connId enableNtfs cMode clientData pqInitKeys subMode srv
    (SCMContact, CR.IKUsePQ) -> throwError $ CMD PROHIBITED
    _ -> pure ()
  AgentConfig {smpClientVRange, smpAgentVRange, e2eEncryptVRange} <- asks config
-  (rq, qUri, tSess, sessId) <- newRcvQueue c userId connId srv smpClientVRange subMode `catchAgentError` \e -> liftIO (print e) >> throwError e
+  (rq, qUri) <- newRcvQueue c userId connId srv smpClientVRange subMode `catchAgentError` \e -> liftIO (print e) >> throwError e
  rq' <- withStore c $ \db -> updateNewConnRcv db connId rq
-  lift . when (subMode == SMSubscribe) $ addNewQueueSubscription c rq' tSess sessId
+  liftIO $ case subMode of
+    SMOnlyCreate -> pure ()
+    SMSubscribe -> addSubscription c rq'
  when enableNtfs $ do
    ns <- asks ntfSupervisor
    atomically $ sendNtfSubCommand ns (connId, NSCCreate)
@@ -861,10 +863,12 @@ joinConnSrvAsync _c _userId _connId _enableNtfs (CRContactUri _) _cInfo _subMode

 createReplyQueue :: AgentClient -> ConnData -> SndQueue -> SubscriptionMode -> SMPServerWithAuth -> AM SMPQueueInfo
 createReplyQueue c ConnData {userId, connId, enableNtfs} SndQueue {smpClientVersion} subMode srv = do
-  (rq, qUri, tSess, sessId) <- newRcvQueue c userId connId srv (versionToRange smpClientVersion) subMode
+  (rq, qUri) <- newRcvQueue c userId connId srv (versionToRange smpClientVersion) subMode
  let qInfo = toVersionT qUri smpClientVersion
  rq' <- withStore c $ \db -> upgradeSndConnToDuplex db connId rq
-  lift . when (subMode == SMSubscribe) $ addNewQueueSubscription c rq' tSess sessId
+  liftIO $ case subMode of
+    SMOnlyCreate -> pure ()
+    SMSubscribe -> addSubscription c rq'
  when enableNtfs $ do
    ns <- asks ntfSupervisor
    atomically $ sendNtfSubCommand ns (connId, NSCCreate)
@@ -924,7 +928,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 . fst <$> subscribeQueues c (concat $ M.elems rcvQs)
+  rcvRs <- lift $ connResults <$> 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
@@ -1322,13 +1326,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
@@ -1504,10 +1508,10 @@ switchDuplexConnection c (DuplexConnection cData@ConnData {connId, userId} rqs s
  -- try to get the server that is different from all queues, or at least from the primary rcv queue
  srvAuth@(ProtoServerWithAuth srv _) <- getNextServer c userId $ map qServer (L.toList rqs) <> map qServer (L.toList sqs)
  srv' <- if srv == server then getNextServer c userId [server] else pure srvAuth
-  (q, qUri, tSess, sessId) <- newRcvQueue c userId connId srv' clientVRange SMSubscribe
+  (q, qUri) <- newRcvQueue c userId connId srv' clientVRange SMSubscribe
  let rq' = (q :: NewRcvQueue) {primary = True, dbReplaceQueueId = Just dbQueueId}
  rq'' <- withStore c $ \db -> addConnRcvQueue db connId rq'
-  lift $ addNewQueueSubscription c rq'' tSess sessId
+  liftIO $ addSubscription c rq''
  void . enqueueMessages c cData sqs SMP.noMsgFlags $ QADD [(qUri, Just (server, sndId))]
  rq1 <- withStore' c $ \db -> setRcvSwitchStatus db rq $ Just RSSendingQADD
  let rqs' = updatedQs rq1 rqs <> [rq'']
@@ -1561,7 +1565,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
@@ -1623,12 +1627,6 @@ 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
@@ -1897,7 +1895,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
@@ -1914,7 +1912,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
@@ -1997,13 +1995,11 @@ getSMPServer c userId = withUserServers c userId pickServer
 {-# INLINE getSMPServer #-}

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

 cleanupManager :: AgentClient -> AM' ()
@@ -2080,10 +2076,8 @@ 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 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)
+processSMPTransmission :: AgentClient -> ServerTransmission SMPVersion BrokerMsg -> AM ()
+processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v, sessId, isResponse, rId, cmd) = do
  (rq, SomeConn _ conn) <- withStore c (\db -> getRcvConn db srv rId)
  processSMP rq conn $ toConnData conn
  where
@@ -2093,15 +2087,14 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v,
      conn
      cData@ConnData {userId, connId, connAgentVersion, ratchetSyncState = rss} =
        withConnLock c connId "processSMP" $ case cmd of
-          Right (SMP.MSG msg@SMP.RcvMessage {msgId = srvMsgId}) ->
+          SMP.MSG msg@SMP.RcvMessage {msgId = srvMsgId} ->
            void . handleNotifyAck $ do
-              isGET <- atomically $ hasGetLock c rq
-              unless isGET checkSubscription
              msg' <- decryptSMPMessage rq msg
              ack' <- handleNotifyAck $ case msg' of
                SMP.ClientRcvMsgBody {msgTs = srvTs, msgFlags, msgBody} -> processClientMsg srvTs msgFlags msgBody
                SMP.ClientRcvMsgQuota {} -> queueDrained >> ack
-              when isGET $ notify (MSGNTF $ SMP.rcvMessageMeta srvMsgId msg')
+              whenM (atomically $ hasGetLock c rq) $
+                notify (MSGNTF $ SMP.rcvMessageMeta srvMsgId msg')
              pure ack'
            where
              queueDrained = case conn of
@@ -2244,7 +2237,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
-          Right SMP.END ->
+          SMP.END ->
            atomically (TM.lookup tSess smpClients $>>= (tryReadTMVar . sessionVar) >>= processEND)
              >>= logServer "<--" c srv rId
            where
@@ -2252,16 +2245,14 @@ 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"
-          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
+          _ -> do
+            logServer "<--" c srv rId $ "unexpected: " <> bshow cmd
+            notify . ERR $ BROKER (B.unpack $ strEncode srv) $ if isResponse then TIMEOUT else UNEXPECTED
        where
          notify :: forall e m. MonadIO m => AEntityI e => ACommand 'Agent e -> m ()
          notify = atomically . notify'
@@ -2275,25 +2266,6 @@ 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 = 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
            clientMsg <- agentCbDecrypt e2eDh cmNonce cmEncBody
@@ -41,7 +41,6 @@ module Simplex.Messaging.Agent.Client
    getQueueMessage,
    decryptSMPMessage,
    addSubscription,
-    addNewQueueSubscription,
    getSubscriptions,
    sendConfirmation,
    sendInvitation,
@@ -78,20 +77,15 @@ module Simplex.Messaging.Agent.Client
    logSecret,
    removeSubscription,
    hasActiveSubscription,
-    hasPendingSubscription,
    hasGetLock,
-    activeClientSession,
    agentClientStore,
    agentDRG,
    getAgentSubscriptions,
-    diffSubscriptions,
    slowNetworkConfig,
-    protocolClientError,
    Worker (..),
    SessionVar (..),
    SubscriptionsInfo (..),
    SubInfo (..),
-    SubscriptionsDiff (..),
    AgentOperation (..),
    AgentOpState (..),
    AgentState (..),
@@ -158,8 +152,7 @@ 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 (partitionEithers)
-import Data.Foldable (fold)
+import Data.Either (lefts, partitionEithers)
 import Data.Functor (($>))
 import Data.Int (Int64)
 import Data.List (deleteFirstsBy, foldl', partition, (\\))
@@ -175,7 +168,6 @@ 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
@@ -237,7 +229,7 @@ import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Session
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
-import Simplex.Messaging.Transport (SMPVersion, SessionId, THandleParams (sessionId))
+import Simplex.Messaging.Transport (SMPVersion)
 import Simplex.Messaging.Transport.Client (TransportHost)
 import Simplex.Messaging.Util
 import Simplex.Messaging.Version
@@ -268,7 +260,7 @@ data AgentClient = AgentClient
    active :: TVar Bool,
    rcvQ :: TBQueue (ATransmission 'Client),
    subQ :: TBQueue (ATransmission 'Agent),
-    msgQ :: TBQueue (ServerTransmission SMPVersion ErrorType BrokerMsg),
+    msgQ :: TBQueue (ServerTransmission SMPVersion BrokerMsg),
    smpServers :: TMap UserId (NonEmpty SMPServerWithAuth),
    smpClients :: TMap SMPTransportSession SMPClientVar,
    ntfServers :: TVar [NtfServer],
@@ -519,7 +511,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 :: HostName -> err -> AgentErrorType
+  clientProtocolError :: err -> AgentErrorType
  closeProtocolServerClient :: Client msg -> IO ()
  clientServer :: Client msg -> String
  clientTransportHost :: Client msg -> TransportHost
@@ -638,17 +630,21 @@ reconnectSMPClient tc c tSess@(_, srv, _) qs = do
  let t = (length qs `div` 90 + 1) * tcpTimeout * 3
  ExceptT (sequence <$> (t `timeout` runExceptT resubscribe)) >>= \case
    Just _ -> atomically $ writeTVar tc 0
-    Nothing -> do
-      tc' <- atomically $ stateTVar tc $ \i -> (i + 1, i + 1)
-      maxTC <- asks $ maxSubscriptionTimeouts . config
-      let err = if tc' >= maxTC then CRITICAL True else INTERNAL
-          msg = show tc' <> " consecutive subscription timeouts: " <> show (length qs) <> " queues, transport session: " <> show tSess
-      atomically $ writeTBQueue (subQ c) ("", "", APC SAEConn $ ERR $ err msg)
+    Nothing ->
+      (offline <$> readTVarIO (userNetworkState c)) >>= \case
+        -- reset and do not report consequitive timeouts while offline
+        Just _ -> atomically $ writeTVar tc 0
+        Nothing -> do
+          tc' <- atomically $ stateTVar tc $ \i -> (i + 1, i + 1)
+          maxTC <- asks $ maxSubscriptionTimeouts . config
+          when (tc' >= maxTC) $ do
+            let msg = show tc' <> " consecutive subscription timeouts: " <> show (length qs) <> " queues, transport session: " <> show tSess
+            atomically $ writeTBQueue (subQ c) ("", "", APC SAEConn $ ERR $ INTERNAL msg)
  where
    resubscribe :: AM ()
    resubscribe = do
      cs <- readTVarIO $ RQ.getConnections $ activeSubs c
-      (rs, sessId_) <- lift . subscribeQueues c $ L.toList qs
+      rs <- 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
@@ -657,10 +653,7 @@ 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 $
-          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_
+          closeClient c smpClients tSess
        throwError err
    notifySub :: forall e. AEntityI e => ConnId -> ACommand 'Agent e -> IO ()
    notifySub connId cmd = atomically $ writeTBQueue (subQ c) ("", connId, APC (sAEntity @e) cmd)
@@ -894,42 +887,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. (HasCallStack, ProtocolServerClient v err msg) => AgentClient -> TransportSession msg -> ByteString -> (Client msg -> AM a) -> AM a
+withClient_ :: forall a v err msg. 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` \e -> withFrozenCallStack $ logServerError cl e
+  (action cl <* stat cl "OK") `catchAgentError` logServerError cl
  where
    stat cl = liftIO . incClientStat c userId cl statCmd
-    logServerError :: HasCallStack => Client msg -> AgentErrorType -> AM a
+    logServerError :: Client msg -> AgentErrorType -> AM a
    logServerError cl e = do
-      withFrozenCallStack $ logServer "<--" c srv "" $ strEncode e
+      logServer "<--" c srv "" $ strEncode e
      stat cl $ strEncode e
      throwError e

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

-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
+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
 {-# INLINE withClient #-}

-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
+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
 {-# INLINE withLogClient #-}

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

-withSMPClient_ :: (HasCallStack, SMPQueueRec q) => AgentClient -> q -> ByteString -> (SMPClient -> AM a) -> AM a
+withSMPClient_ :: SMPQueueRec q => AgentClient -> q -> ByteString -> (SMPClient -> AM a) -> AM a
 withSMPClient_ c q cmdStr action = do
  tSess <- liftIO $ mkSMPTransportSession c q
-  withFrozenCallStack $ withLogClient_ c tSess (queueId q) cmdStr action
+  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)
@@ -945,13 +938,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) => (HostName -> err -> AgentErrorType) -> HostName -> ExceptT (ProtocolClientError err) IO a -> AM a
+liftClient :: (Show err, Encoding err) => (err -> AgentErrorType) -> HostName -> ExceptT (ProtocolClientError err) IO a -> AM a
 liftClient protocolError_ = liftError . protocolClientError protocolError_
 {-# INLINE liftClient #-}

-protocolClientError :: (Show err, Encoding err) => (HostName -> err -> AgentErrorType) -> HostName -> ProtocolClientError err -> AgentErrorType
+protocolClientError :: (Show err, Encoding err) => (err -> AgentErrorType) -> HostName -> ProtocolClientError err -> AgentErrorType
 protocolClientError protocolError_ host = \case
-  PCEProtocolError e -> protocolError_ host e
+  PCEProtocolError e -> protocolError_ e
  PCEResponseError e -> BROKER host $ RESPONSE $ B.unpack $ smpEncode e
  PCEUnexpectedResponse _ -> BROKER host UNEXPECTED
  PCEResponseTimeout -> BROKER host TIMEOUT
@@ -1030,7 +1023,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 (B.unpack $ strEncode srv) DIGEST
+          unless (digest == rcvDigest) $ throwError $ ProtocolTestFailure TSCompareFile $ XFTP DIGEST
          liftError (testErr TSDeleteFile) $ X.deleteXFTPChunk xftp spKey sId
        ok <- tcpTimeout xftpNetworkConfig `timeout` X.closeXFTPClient xftp
        incClientStat c userId xftp "XFTP_TEST" "OK"
@@ -1105,7 +1098,7 @@ getSessionMode :: AgentClient -> IO TransportSessionMode
 getSessionMode = atomically . fmap sessionMode . getNetworkConfig
 {-# INLINE getSessionMode #-}

-newRcvQueue :: AgentClient -> UserId -> ConnId -> SMPServerWithAuth -> VersionRangeSMPC -> SubscriptionMode -> AM (NewRcvQueue, SMPQueueUri, SMPTransportSession, SessionId)
+newRcvQueue :: AgentClient -> UserId -> ConnId -> SMPServerWithAuth -> VersionRangeSMPC -> SubscriptionMode -> AM (NewRcvQueue, SMPQueueUri)
 newRcvQueue c userId connId (ProtoServerWithAuth srv auth) vRange subMode = do
  C.AuthAlg a <- asks (rcvAuthAlg . config)
  g <- asks random
@@ -1114,9 +1107,8 @@ newRcvQueue c userId connId (ProtoServerWithAuth srv auth) vRange subMode = do
  (e2eDhKey, e2ePrivKey) <- atomically $ C.generateKeyPair g
  logServer "-->" c srv "" "NEW"
  tSess <- liftIO $ mkTransportSession c userId srv connId
-  (sessId, QIK {rcvId, sndId, rcvPublicDhKey}) <-
-    withClient c tSess "NEW" $ \smp ->
-      (sessionId $ thParams smp,) <$> createSMPQueue smp rKeys dhKey auth subMode
+  QIK {rcvId, sndId, rcvPublicDhKey} <-
+    withClient c tSess "NEW" $ \smp -> createSMPQueue smp rKeys dhKey auth subMode
  liftIO . logServer "<--" c srv "" $ B.unwords ["IDS", logSecret rcvId, logSecret sndId]
  let rq =
        RcvQueue
@@ -1138,18 +1130,17 @@ newRcvQueue c userId connId (ProtoServerWithAuth srv auth) vRange subMode = do
            clientNtfCreds = Nothing,
            deleteErrors = 0
          }
-      qUri = SMPQueueUri vRange $ SMPQueueAddress srv sndId e2eDhKey
-  pure (rq, qUri, tSess, sessId)
+  pure (rq, SMPQueueUri vRange $ SMPQueueAddress srv sndId e2eDhKey)

-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
+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)
+    _ -> addSubscription c rq
+  pure r

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

 -- | Subscribe to queues. The list of results can have a different order.
-subscribeQueues :: AgentClient -> [RcvQueue] -> AM' ([(RcvQueue, Either AgentErrorType ())], Maybe SessionId)
+subscribeQueues :: AgentClient -> [RcvQueue] -> AM' [(RcvQueue, Either AgentErrorType ())]
 subscribeQueues c qs = do
  (errs, qs') <- partitionEithers <$> mapM checkQueue qs
  atomically $ do
@@ -1174,43 +1165,20 @@ subscribeQueues c qs = do
    RQ.batchAddQueues (pendingSubs c) qs'
  env <- ask
  -- only "checked" queues are subscribed
-  session <- newTVarIO Nothing
-  rs <- sendTSessionBatches "SUB" 90 id (subscribeQueues_ env session) c qs'
-  (errs <> rs,) <$> readTVarIO session
+  (errs <>) <$> sendTSessionBatches "SUB" 90 id (subscribeQueues_ env) c qs'
  where
    checkQueue rq = do
      prohibited <- atomically $ hasGetLock c rq
      pure $ if prohibited then Left (rq, Left $ CMD PROHIBITED) else Right rq
-    subscribeQueues_ :: 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']
+    subscribeQueues_ :: Env -> SMPClient -> NonEmpty RcvQueue -> IO (BatchResponses SMPClientError ())
+    subscribeQueues_ env smp qs' = do
      rs <- sendBatch subscribeSMPQueues smp qs'
-      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
+      mapM_ (uncurry $ processSubResult c) rs
+      when (any temporaryClientError . lefts . map snd $ L.toList rs) $
+        runReaderT (resubscribeSMPSession c $ transportSession' smp) env
+      pure rs

-activeClientSession :: AgentClient -> SMPTransportSession -> SessionId -> STM Bool
-activeClientSession c tSess sessId = sameSess <$> tryReadSessVar tSess (smpClients c)
-  where
-    sameSess = \case
-      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)]
@@ -1245,35 +1213,16 @@ sendBatch smpCmdFunc smp qs = L.zip qs <$> smpCmdFunc smp (L.map queueCreds qs)
  where
    queueCreds RcvQueue {rcvPrivateKey, rcvId} = (rcvPrivateKey, rcvId)

-addSubscription :: AgentClient -> RcvQueue -> STM ()
-addSubscription c rq@RcvQueue {connId} = do
+addSubscription :: AgentClient -> RcvQueue -> IO ()
+addSubscription c rq@RcvQueue {connId} = atomically $ do
  modifyTVar' (subscrConns c) $ S.insert connId
  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 <-
-    atomically $
-      ifM
-        (activeClientSession c tSess sessId)
-        (True <$ addSubscription c rq)
-        (False <$ addPendingSubscription c rq)
-  unless same $ resubscribeSMPSession c tSess
-
 hasActiveSubscription :: AgentClient -> ConnId -> STM Bool
 hasActiveSubscription c connId = RQ.hasConn connId $ activeSubs c
 {-# INLINE hasActiveSubscription #-}

-hasPendingSubscription :: AgentClient -> ConnId -> STM Bool
-hasPendingSubscription c connId = RQ.hasConn connId $ pendingSubs c
-{-# INLINE hasPendingSubscription #-}
-
 removeSubscription :: AgentClient -> ConnId -> STM ()
 removeSubscription c connId = do
  modifyTVar' (subscrConns c) $ S.delete connId
@@ -1284,9 +1233,9 @@ getSubscriptions :: AgentClient -> STM (Set ConnId)
 getSubscriptions = readTVar . subscrConns
 {-# INLINE getSubscriptions #-}

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

 showServer :: ProtocolServer s -> ByteString
@@ -1737,31 +1686,6 @@ 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],
@@ -1899,8 +1823,6 @@ $(J.deriveJSON defaultJSON ''SubInfo)

 $(J.deriveJSON defaultJSON ''SubscriptionsInfo)

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

 $(J.deriveJSON defaultJSON ''WorkersSummary)
@@ -181,7 +181,7 @@ defaultAgentConfig =
      maxWorkerRestartsPerMin = 5,
      -- 3 consecutive subscription timeouts will result in alert to the user
      -- this is a fallback, as the timeout set to 3x of expected timeout, to avoid potential locking.
-      maxSubscriptionTimeouts = 3,
+      maxSubscriptionTimeouts = 5,
      storedMsgDataTTL = 21 * nominalDay,
      rcvFilesTTL = 2 * nominalDay,
      sndFilesTTL = nominalDay,
@@ -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,
-    pattern PQEncOff,
-    pattern PQSupportOff,
-    pattern PQSupportOn,
+    SndE2ERatchetParams
  )
 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,
-    VersionRangeSMPC,
+    SMPClientVersion,
    VersionSMPC,
+    VersionRangeSMPC,
    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 {serverAddress :: String, smpErr :: ErrorType}
+    SMP {smpErr :: ErrorType}
  | -- | NTF protocol errors forwarded to agent clients
-    NTF {serverAddress :: String, ntfErr :: ErrorType}
+    NTF {ntfErr :: ErrorType}
  | -- | XFTP protocol errors forwarded to agent clients
-    XFTP {serverAddress :: String, xftpErr :: XFTPErrorType}
+    XFTP {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 <$> textP <*> _strP)
-      <|> "NTF " *> (NTF <$> textP <*> _strP)
-      <|> "XFTP " *> (XFTP <$> textP <*> _strP)
+      <|> "SMP " *> (SMP <$> strP)
+      <|> "NTF " *> (NTF <$> strP)
+      <|> "XFTP " *> (XFTP <$> 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 srv e -> "SMP " <> text srv <> " " <> strEncode e
-    NTF srv e -> "NTF " <> text srv <> " " <> strEncode e
-    XFTP srv e -> "XFTP " <> text srv <> " " <> strEncode e
+    SMP e -> "SMP " <> strEncode e
+    NTF e -> "NTF " <> strEncode e
+    XFTP e -> "XFTP " <> 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
@@ -49,7 +49,7 @@ runSMPAgentBlocking (ATransport t) cfg@AgentConfig {tcpPort, caCertificateFile,
    smpAgent :: forall c. Transport c => TProxy c -> ServiceName -> Env -> IO ()
    smpAgent _ port env = do
      -- tlsServerParams is not in Env to avoid breaking functional API w/t key and certificate generation
-      tlsServerParams <- loadTLSServerParams caCertificateFile certificateFile privateKeyFile
+      tlsServerParams <- loadTLSServerParams caCertificateFile certificateFile privateKeyFile Nothing
      clientId <- newTVarIO initClientId
      runTransportServer started port tlsServerParams defaultTransportServerConfig $ \(h :: c) -> do
        putLn h $ "Welcome to SMP agent v" <> B.pack simplexMQVersion
@@ -28,7 +28,7 @@
 module Simplex.Messaging.Client
  ( -- * Connect (disconnect) client to (from) SMP server
    TransportSession,
-    ProtocolClient (thParams, sessionTs, sentSubs),
+    ProtocolClient (thParams, sessionTs),
    SMPClient,
    getProtocolClient,
    closeProtocolClient,
@@ -70,7 +70,6 @@ module Simplex.Messaging.Client
    proxyUsername,
    temporaryClientError,
    ServerTransmission,
-    TransmissionType (..),
    ClientCommand,

    -- * For testing
@@ -84,7 +83,6 @@ 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)
@@ -112,7 +110,7 @@ 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, ifM, raceAny_, threadDelay', tshow, whenM)
+import Simplex.Messaging.Util (bshow, diffToMicroseconds, raceAny_, threadDelay', whenM)
 import Simplex.Messaging.Version
 import System.Timeout (timeout)

@@ -123,7 +121,6 @@ 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
  }

@@ -140,7 +137,7 @@ data PClient v err msg = PClient
    sentCommands :: TMap CorrId (Request err msg),
    sndQ :: TBQueue (TVar Bool, ByteString),
    rcvQ :: TBQueue (NonEmpty (SignedTransmission err msg)),
-    msgQ :: Maybe (TBQueue (ServerTransmission v err msg))
+    msgQ :: Maybe (TBQueue (ServerTransmission v msg))
  }

 smpClientStub :: TVar ChaChaDRG -> ByteString -> VersionSMP -> Maybe (THandleAuth 'TClient) -> STM SMPClient
@@ -150,7 +147,6 @@ 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
@@ -168,7 +164,6 @@ smpClientStub g sessionId thVersion thAuth = do
              batch = True
            },
        sessionTs = ts,
-        sentSubs,
        client_ =
          PClient
            { connected,
@@ -193,10 +188,7 @@ 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 err msg = (TransportSession msg, Version v, SessionId, TransmissionType, EntityId, Either (ProtocolClientError err) msg)
-
-data TransmissionType = TTEvent | TTUncorrelatedResponse | TTExpiredSubResponse
-  deriving (Show)
+type ServerTransmission v msg = (TransportSession msg, Version v, SessionId, Bool, EntityId, msg)

 data HostMode
  = -- | prefer (or require) onion hosts when connecting via SOCKS proxy
@@ -268,6 +260,7 @@ data ProtocolClientConfig v = ProtocolClientConfig
    defaultTransport :: (ServiceName, ATransport),
    -- | network configuration
    networkConfig :: NetworkConfig,
+    clientALPN :: Maybe [ALPN],
    -- | client-server protocol version range
    serverVRange :: VersionRange v,
    -- | agree shared session secret (used in SMP proxy)
@@ -275,26 +268,25 @@ data ProtocolClientConfig v = ProtocolClientConfig
  }

 -- | Default protocol client configuration.
-defaultClientConfig :: VersionRange v -> ProtocolClientConfig v
-defaultClientConfig serverVRange =
+defaultClientConfig :: Maybe [ALPN] -> VersionRange v -> ProtocolClientConfig v
+defaultClientConfig clientALPN serverVRange =
  ProtocolClientConfig
    { qSize = 64,
      defaultTransport = ("443", transport @TLS),
      networkConfig = defaultNetworkConfig,
+      clientALPN,
      serverVRange,
      agreeSecret = False
    }
 {-# INLINE defaultClientConfig #-}

 defaultSMPClientConfig :: ProtocolClientConfig SMPVersion
-defaultSMPClientConfig = defaultClientConfig supportedClientSMPRelayVRange
+defaultSMPClientConfig = defaultClientConfig (Just supportedSMPHandshakes) supportedClientSMPRelayVRange
 {-# INLINE defaultSMPClientConfig #-}

 data Request err msg = Request
  { corrId :: CorrId,
    entityId :: EntityId,
-    subscription :: Bool,
-    pending :: TVar Bool,
    responseVar :: TMVar (Either (ProtocolClientError err) msg)
  }

@@ -341,8 +333,8 @@ 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 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
+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 g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize, networkConfig, clientALPN, serverVRange, agreeSecret} msgQ disconnected = do
  case chooseTransportHost networkConfig (host srv) of
    Right useHost ->
      (getCurrentTime >>= atomically . mkProtocolClient useHost >>= runClient useTransport useHost)
@@ -380,7 +372,7 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
    runClient :: (ServiceName, ATransport) -> TransportHost -> PClient v err msg -> IO (Either (ProtocolClientError err) (ProtocolClient v err msg))
    runClient (port', ATransport t) useHost c = do
      cVar <- newEmptyTMVarIO
-      let tcConfig = transportClientConfig networkConfig
+      let tcConfig = (transportClientConfig networkConfig) {alpn = clientALPN}
          username = proxyUsername transportSession
      action <-
        async $
@@ -405,19 +397,16 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
        Left e -> atomically . putTMVar cVar . Left $ PCETransportError e
        Right th@THandle {params} -> do
          sessionTs <- getCurrentTime
-          sentSubs <- atomically TM.empty
-          let c' = ProtocolClient {action = Nothing, client_ = c, thParams = params, sessionTs, sentSubs}
+          let c' = ProtocolClient {action = Nothing, client_ = c, thParams = params, sessionTs}
          atomically $ writeTVar (lastReceived c) sessionTs
          atomically $ do
            writeTVar (connected c) True
            putTMVar cVar $ Right c'
-          raceAny_ ([send c' th, process c', receive c' th] <> [monitor c' | smpPingInterval > 0])
+          raceAny_ ([send c' th, process c', receive c' th] <> [ping c' | smpPingInterval > 0])
            `finally` disconnected c'

    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 (active, s) = whenM (readTVarIO active) $ void $ tPutLog h s
+    send ProtocolClient {client_ = PClient {sndQ}} h = forever $ atomically (readTBQueue sndQ) >>= \(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
@@ -425,8 +414,8 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
      getCurrentTime >>= atomically . writeTVar lastReceived
      atomically $ writeTVar timeoutErrorCount 0

-    monitor :: ProtocolClient v err msg -> IO ()
-    monitor c@ProtocolClient {client_ = PClient {sendPings, lastReceived, timeoutErrorCount}} = loop smpPingInterval
+    ping :: ProtocolClient v err msg -> IO ()
+    ping c@ProtocolClient {client_ = PClient {sendPings, lastReceived, timeoutErrorCount}} = loop smpPingInterval
      where
        loop :: Int64 -> IO ()
        loop delay = do
@@ -450,40 +439,27 @@ 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))
-      | not $ B.null $ bs corrId =
+      | isResponse =
          atomically (TM.lookup corrId sentCommands) >>= \case
-            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
+            Nothing -> sendMsg respOrErr
+            Just Request {entityId, responseVar} -> atomically $ do
+              TM.delete corrId sentCommands
+              putTMVar responseVar $ response entityId
+      | otherwise = sendMsg respOrErr
      where
-        -- 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
+        isResponse = not $ B.null $ bs corrId
        response entityId
-          | entityId == entId = clientResp
+          | entityId == entId =
+              case respOrErr of
+                Left e -> Left $ PCEResponseError e
+                Right r -> case protocolError r of
+                  Just e -> Left $ PCEProtocolError e
+                  _ -> Right r
          | otherwise = Left . PCEUnexpectedResponse $ bshow respOrErr
-        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
+        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

 proxyUsername :: TransportSession msg -> ByteString
 proxyUsername (userId, _, entityId_) = C.sha256Hash $ bshow userId <> maybe "" (":" <>) entityId_
@@ -582,11 +558,11 @@ processSUBResponse c (Response rId r) = case r of
  Left e -> pure $ Left e

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

-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)
+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)

 -- | 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
 --
@@ -711,7 +687,7 @@ 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. Protocol v err msg => ProtocolClient v err msg -> NonEmpty (ClientCommand msg) -> IO (NonEmpty (Response err msg))
+sendProtocolCommands :: forall v err msg. ProtocolEncoding v err (ProtoCommand 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) bs
@@ -728,13 +704,13 @@ sendProtocolCommands c@ProtocolClient {thParams = THandleParams {batch, blockSiz
      where
        diff = L.length cs - length rs

-streamProtocolCommands :: forall v err msg. Protocol v err msg => ProtocolClient v err msg -> NonEmpty (ClientCommand msg) -> ([Response err msg] -> IO ()) -> IO ()
+streamProtocolCommands :: forall v err msg. ProtocolEncoding v err (ProtoCommand 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) bs

 sendBatch :: ProtocolClient v err msg -> TransportBatch (Request err msg) -> IO [Response err msg]
-sendBatch c@ProtocolClient {client_ = PClient {sndQ}} b = do
+sendBatch c@ProtocolClient {client_ = PClient {rcvConcurrency, sndQ}} b = do
  case b of
    TBError e Request {entityId} -> do
      putStrLn "send error: large message"
@@ -751,7 +727,7 @@ sendBatch c@ProtocolClient {client_ = PClient {sndQ}} b = do
      (: []) <$> getResponse c active r

 -- | Send Protocol command
-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 :: 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 c@ProtocolClient {client_ = PClient {sndQ}, thParams = THandleParams {batch, blockSize}} pKey entId cmd =
  ExceptT $ uncurry sendRecv =<< mkTransmission c (pKey, entId, cmd)
  where
@@ -772,20 +748,20 @@ sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ}, thParams = THand

 -- TODO switch to timeout or TimeManager that supports Int64
 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
+getResponse ProtocolClient {client_ = PClient {tcpTimeout, timeoutErrorCount, sentCommands}} active Request {corrId, entityId, responseVar} = do
  response <-
    timeout tcpTimeout (atomically (takeTMVar responseVar)) >>= \case
      Just r -> atomically (writeTVar timeoutErrorCount 0) $> r
      Nothing -> do
-        atomically (writeTVar active False >> writeTVar pending False)
+        atomically (writeTVar active False >> TM.delete corrId sentCommands)
        atomically $ modifyTVar' timeoutErrorCount (+ 1)
        pure $ Left PCEResponseTimeout
  pure Response {entityId, response}

-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
+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
  corrId <- atomically getNextCorrId
-  let TransmissionForAuth {tForAuth, tToSend} = encodeTransmissionForAuth thParams (corrId, entityId, cmd)
+  let TransmissionForAuth {tForAuth, tToSend} = encodeTransmissionForAuth thParams (corrId, entId, cmd)
      auth = authTransmission (thAuth thParams) pKey_ corrId tForAuth
  r <- atomically $ mkRequest corrId
  pure ((,tToSend) <$> auth, r)
@@ -794,16 +770,7 @@ 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
-      pending <- newTVar True
-      responseVar <- newEmptyTMVar
-      let r =
-            Request
-              { corrId,
-                entityId,
-                subscription = protocolSub cmd,
-                pending,
-                responseVar
-              }
+      r <- Request corrId entId <$> newEmptyTMVar
      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, ErrorType, NotifierId, NtfPrivateAuthKey, ProtocolServer (..), QueueId, RcvPrivateAuthKey, RecipientId, SMPServer)
+import Simplex.Messaging.Protocol (BrokerMsg, 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 ErrorType BrokerMsg),
+    msgQ :: TBQueue (ServerTransmission SMPVersion BrokerMsg),
    agentQ :: TBQueue SMPClientAgentEvent,
    randomDrg :: TVar ChaChaDRG,
    smpClients :: TMap SMPServer SMPClientVar,
@@ -10,7 +10,7 @@ import Data.Word (Word16)
 import Simplex.Messaging.Client
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Notifications.Protocol
-import Simplex.Messaging.Notifications.Transport (NTFVersion, supportedClientNTFVRange)
+import Simplex.Messaging.Notifications.Transport (NTFVersion, supportedClientNTFVRange, supportedNTFHandshakes)
 import Simplex.Messaging.Protocol (ErrorType)
 import Simplex.Messaging.Util (bshow)

@@ -19,7 +19,7 @@ type NtfClient = ProtocolClient NTFVersion ErrorType NtfResponse
 type NtfClientError = ProtocolClientError ErrorType

 defaultNTFClientConfig :: ProtocolClientConfig NTFVersion
-defaultNTFClientConfig = defaultClientConfig supportedClientNTFVRange
+defaultNTFClientConfig = defaultClientConfig (Just supportedNTFHandshakes) supportedClientNTFVRange

 ntfRegisterToken :: NtfClient -> C.APrivateAuthKey -> NewNtfEntity 'Token -> ExceptT NtfClientError IO (NtfTokenId, C.PublicKeyX25519)
 ntfRegisterToken c pKey newTkn =
@@ -154,7 +154,6 @@ 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, _), _, _, tType, ntfId, msgOrErr) <- atomically $ readTBQueue msgQ
+      ((_, srv, _), _, _, _, ntfId, msg) <- atomically $ readTBQueue msgQ
      let smpQueue = SMPQueueNtf srv ntfId
-      case msgOrErr of
-        Right (SMP.NMSG nmsgNonce encNMsgMeta) -> do
+      case msg of
+        SMP.NMSG nmsgNonce encNMsgMeta -> do
          ntfTs <- liftIO getSystemTime
          st <- asks store
          NtfPushServer {pushQ} <- asks pushServer
@@ -231,10 +231,8 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
            findNtfSubscriptionToken st smpQueue
              >>= mapM_ (\tkn -> writeTBQueue pushQ (tkn, PNMessage PNMessageData {smpQueue, ntfTs, nmsgNonce, encNMsgMeta}))
          incNtfStat ntfReceived
-        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
+        SMP.END -> updateSubStatus smpQueue NSEnd
+        _ -> pure ()

    receiveAgent =
      forever $
@@ -34,7 +34,7 @@ import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport (ATransport, THandleParams, TransportPeer (..))
-import Simplex.Messaging.Transport.Server (TransportServerConfig, loadFingerprint, loadTLSServerParams)
+import Simplex.Messaging.Transport.Server (TransportServerConfig, alpn, loadFingerprint, loadTLSServerParams)
 import System.IO (IOMode (..))
 import System.Mem.Weak (Weak)
 import UnliftIO.STM
@@ -84,7 +84,7 @@ data NtfEnv = NtfEnv
  }

 newNtfServerEnv :: NtfServerConfig -> IO NtfEnv
-newNtfServerEnv config@NtfServerConfig {subQSize, pushQSize, smpAgentCfg, apnsConfig, storeLogFile, caCertificateFile, certificateFile, privateKeyFile} = do
+newNtfServerEnv config@NtfServerConfig {subQSize, pushQSize, smpAgentCfg, apnsConfig, storeLogFile, caCertificateFile, certificateFile, privateKeyFile, transportConfig} = do
  random <- liftIO C.newRandom
  store <- atomically newNtfStore
  logInfo "restoring subscriptions..."
@@ -92,7 +92,7 @@ newNtfServerEnv config@NtfServerConfig {subQSize, pushQSize, smpAgentCfg, apnsCo
  logInfo "restored subscriptions"
  subscriber <- atomically $ newNtfSubscriber subQSize smpAgentCfg random
  pushServer <- atomically $ newNtfPushServer pushQSize apnsConfig
-  tlsServerParams <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile
+  tlsServerParams <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile (alpn transportConfig)
  Fingerprint fp <- liftIO $ loadFingerprint caCertificateFile
  serverStats <- atomically . newNtfServerStats =<< liftIO getCurrentTime
  pure NtfEnv {config, subscriber, pushServer, store, storeLog, random, tlsServerParams, serverIdentity = C.KeyHash fp, serverStats}
@@ -18,7 +18,7 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Notifications.Server (runNtfServer)
 import Simplex.Messaging.Notifications.Server.Env (NtfServerConfig (..), defaultInactiveClientExpiration)
 import Simplex.Messaging.Notifications.Server.Push.APNS (defaultAPNSPushClientConfig)
-import Simplex.Messaging.Notifications.Transport (supportedServerNTFVRange)
+import Simplex.Messaging.Notifications.Transport (supportedNTFHandshakes, supportedServerNTFVRange)
 import Simplex.Messaging.Protocol (ProtoServerWithAuth (..), pattern NtfServer)
 import Simplex.Messaging.Server.CLI
 import Simplex.Messaging.Server.Expiration
@@ -133,7 +133,8 @@ ntfServerCLI cfgPath logPath =
              ntfServerVRange = supportedServerNTFVRange,
              transportConfig =
                defaultTransportServerConfig
-                  { logTLSErrors = fromMaybe False $ iniOnOff "TRANSPORT" "log_tls_errors" ini
+                  { logTLSErrors = fromMaybe False $ iniOnOff "TRANSPORT" "log_tls_errors" ini,
+                    alpn = Just supportedNTFHandshakes
                  }
            }

@@ -44,7 +44,7 @@ authBatchCmdsNTFVersion :: VersionNTF
 authBatchCmdsNTFVersion = VersionNTF 2

 currentClientNTFVersion :: VersionNTF
-currentClientNTFVersion = VersionNTF 1
+currentClientNTFVersion = VersionNTF 2

 currentServerNTFVersion :: VersionNTF
 currentServerNTFVersion = VersionNTF 2
@@ -52,9 +52,15 @@ currentServerNTFVersion = VersionNTF 2
 supportedClientNTFVRange :: VersionRangeNTF
 supportedClientNTFVRange = mkVersionRange initialNTFVersion currentClientNTFVersion

+legacyServerNTFVRange :: VersionRangeNTF
+legacyServerNTFVRange = mkVersionRange initialNTFVersion initialNTFVersion
+
 supportedServerNTFVRange :: VersionRangeNTF
 supportedServerNTFVRange = mkVersionRange initialNTFVersion currentServerNTFVersion

+supportedNTFHandshakes :: [ALPN]
+supportedNTFHandshakes = ["ntf/1"]
+
 type THandleNTF c p = THandle NTFVersion c p

 data NtfServerHandshake = NtfServerHandshake
@@ -104,12 +110,13 @@ ntfServerHandshake :: forall c. Transport c => C.APrivateSignKey -> c -> C.KeyPa
 ntfServerHandshake serverSignKey c (k, pk) kh ntfVRange = do
  let th@THandle {params = THandleParams {sessionId}} = ntfTHandle c
  let sk = C.signX509 serverSignKey $ C.publicToX509 k
-  sendHandshake th $ NtfServerHandshake {sessionId, ntfVersionRange = ntfVRange, authPubKey = Just sk}
+  let ntfVersionRange = maybe legacyServerNTFVRange (const ntfVRange) $ getSessionALPN c
+  sendHandshake th $ NtfServerHandshake {sessionId, ntfVersionRange, authPubKey = Just sk}
  getHandshake th >>= \case
    NtfClientHandshake {ntfVersion = v, keyHash}
      | keyHash /= kh ->
          throwError $ TEHandshake IDENTITY
-      | v `isCompatible` ntfVRange ->
+      | v `isCompatible` ntfVersionRange ->
          pure $ ntfThHandleServer th v pk
      | otherwise -> throwError $ TEHandshake VERSION

@@ -1094,7 +1094,6 @@ 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)
@@ -1104,10 +1103,6 @@ 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
@@ -51,9 +51,11 @@ import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy.Char8 as LB
 import Data.Either (fromRight, partitionEithers)
 import Data.Functor (($>))
+import qualified Data.HashMap.Strict as HM
 import Data.Int (Int64)
 import qualified Data.IntMap.Strict as IM
-import Data.List (intercalate)
+import Data.List (intercalate, mapAccumR)
+import Data.List.NonEmpty (NonEmpty)
 import qualified Data.List.NonEmpty as L
 import qualified Data.Map.Strict as M
 import Data.Maybe (isNothing)
@@ -410,7 +412,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
                    hPutStrLn h "AUTH"

 runClientTransport :: Transport c => THandleSMP c 'TServer -> M ()
-runClientTransport th@THandle {params = THandleParams {thVersion, sessionId}} = do
+runClientTransport h@THandle {params = THandleParams {thVersion, sessionId}} = do
  q <- asks $ tbqSize . config
  ts <- liftIO getSystemTime
  active <- asks clients
@@ -421,11 +423,12 @@ runClientTransport th@THandle {params = THandleParams {thVersion, sessionId}} =
    pure new
  s <- asks server
  expCfg <- asks $ inactiveClientExpiration . config
+  th <- newMVar h -- put TH under a fair lock to interleave messages and command responses
  labelMyThread . B.unpack $ "client $" <> encode sessionId
-  raceAny_ ([liftIO $ send th c, client c s, receive th c] <> disconnectThread_ c expCfg)
+  raceAny_ ([liftIO $ send th c, liftIO $ sendMsg th c, client c s, receive h c] <> disconnectThread_ c expCfg)
    `finally` clientDisconnected c
  where
-    disconnectThread_ c (Just expCfg) = [liftIO $ disconnectTransport th (rcvActiveAt c) (sndActiveAt c) expCfg (noSubscriptions c)]
+    disconnectThread_ c (Just expCfg) = [liftIO $ disconnectTransport h (rcvActiveAt c) (sndActiveAt c) expCfg (noSubscriptions c)]
    disconnectThread_ _ _ = []
    noSubscriptions c = atomically $ (&&) <$> TM.null (subscriptions c) <*> TM.null (ntfSubscriptions c)

@@ -458,10 +461,10 @@ cancelSub sub =
    _ -> return ()

 receive :: Transport c => THandleSMP c 'TServer -> Client -> M ()
-receive th@THandle {params = THandleParams {thAuth}} Client {rcvQ, sndQ, rcvActiveAt, sessionId} = do
+receive h@THandle {params = THandleParams {thAuth}} Client {rcvQ, sndQ, rcvActiveAt, sessionId} = do
  labelMyThread . B.unpack $ "client $" <> encode sessionId <> " receive"
  forever $ do
-    ts <- L.toList <$> liftIO (tGet th)
+    ts <- L.toList <$> liftIO (tGet h)
    atomically . writeTVar rcvActiveAt =<< liftIO getSystemTime
    as <- partitionEithers <$> mapM cmdAction ts
    write sndQ $ fst as
@@ -478,20 +481,41 @@ receive th@THandle {params = THandleParams {thAuth}} Client {rcvQ, sndQ, rcvActi
              VRFailed -> Left (corrId, queueId, ERR AUTH)
    write q = mapM_ (atomically . writeTBQueue q) . L.nonEmpty

-send :: Transport c => THandleSMP c 'TServer -> Client -> IO ()
-send h@THandle {params} Client {sndQ, sessionId, sndActiveAt} = do
+send :: Transport c => MVar (THandleSMP c 'TServer) -> Client -> IO ()
+send th c@Client {sndQ, msgQ, sessionId} = do
  labelMyThread . B.unpack $ "client $" <> encode sessionId <> " send"
-  forever $ do
-    ts <- atomically $ L.sortWith tOrder <$> readTBQueue sndQ
-    -- TODO we can authorize responses as well
-    void . liftIO . tPut h $ L.map (\t -> Right (Nothing, encodeTransmission params t)) ts
-    atomically . writeTVar sndActiveAt =<< liftIO getSystemTime
+  forever $ atomically (readTBQueue sndQ) >>= sendTransmissions
  where
-    tOrder :: Transmission BrokerMsg -> Int
-    tOrder (_, _, cmd) = case cmd of
-      MSG {} -> 0
-      NMSG {} -> 0
-      _ -> 1
+    sendTransmissions :: NonEmpty (Transmission BrokerMsg) -> IO ()
+    sendTransmissions ts
+      | L.length ts <= 2 = tSend th c ts
+      | otherwise = do
+          let (msgs_, ts') = mapAccumR splitMessages [] ts
+          -- If the request had batched subscriptions (L.length ts > 2)
+          -- this will reply OK to all SUBs in the first batched transmission,
+          -- to reduce client timeouts.
+          tSend th c ts'
+          -- After that all messages will be sent in separate transmissions,
+          -- without any client response timeouts, and allowing them to interleave
+          -- with other requests responses.
+          mapM_ (atomically . writeTBQueue msgQ) $ L.nonEmpty msgs_
+      where
+        splitMessages :: [Transmission BrokerMsg] -> Transmission BrokerMsg -> ([Transmission BrokerMsg], Transmission BrokerMsg)
+        splitMessages msgs t@(corrId, entId, cmd) = case cmd of
+          -- replace MSG response with OK, accumulating MSG in a separate list.
+          MSG {} -> ((CorrId "", entId, cmd) : msgs, (corrId, entId, OK))
+          _ -> (msgs, t)
+
+sendMsg :: Transport c => MVar (THandleSMP c 'TServer) -> Client -> IO ()
+sendMsg th c@Client {msgQ, sessionId} = do
+  labelMyThread . B.unpack $ "client $" <> encode sessionId <> " sendMsg"
+  forever $ atomically (readTBQueue msgQ) >>= mapM_ (\t -> tSend th c [t])
+
+tSend :: Transport c => MVar (THandleSMP c 'TServer) -> Client -> NonEmpty (Transmission BrokerMsg) -> IO ()
+tSend th Client {sndActiveAt} ts = do
+  withMVar th $ \h@THandle {params} ->
+    void . tPut h $ L.map (\t -> Right (Nothing, encodeTransmission params t)) ts
+  atomically . writeTVar sndActiveAt =<< liftIO getSystemTime

 disconnectTransport :: Transport c => THandle v c 'TServer -> TVar SystemTime -> TVar SystemTime -> ExpirationConfig -> IO Bool -> IO ()
 disconnectTransport THandle {connection, params = THandleParams {sessionId}} rcvActiveAt sndActiveAt expCfg noSubscriptions = do
@@ -518,7 +542,11 @@ verifyTransmission :: Maybe (THandleAuth 'TServer, C.CbNonce) -> Maybe Transmiss
 verifyTransmission auth_ tAuth authorized queueId cmd =
  case cmd of
    Cmd SRecipient (NEW k _ _ _) -> pure $ Nothing `verifiedWith` k
-    Cmd SRecipient _ -> verifyQueue (\q -> Just q `verifiedWith` recipientKey q) <$> get SRecipient
+    Cmd SRecipient _ -> do
+      QueueStore {recipientKeys} <- asks queueStore
+      (queueId `HM.lookup`) <$> readTVarIO recipientKeys >>= \case
+        Nothing -> pure $! dummyVerify
+        Just rKey -> if verify rKey then slowIO >> get SRecipient >>= either (\_notGonnaHappenUnlessDeletedWhileCheckingAuth -> pure VRFailed) (pure . VRVerified . Just) else pure VRFailed
    -- SEND will be accepted without authorization before the queue is secured with KEY command
    Cmd SSender SEND {} -> verifyQueue (\q -> Just q `verified` maybe (isNothing tAuth) verify (senderKey q)) <$> get SSender
    Cmd SSender PING -> pure $ VRVerified Nothing
@@ -531,6 +559,8 @@ verifyTransmission auth_ tAuth authorized queueId cmd =
    verifyQueue = either (\_ -> dummyVerify)
    verified q cond = if cond then VRVerified q else VRFailed
    verifiedWith q k = q `verified` verify k
+    slowIO :: M ()
+    slowIO = liftIO $ threadDelay 100000 -- XXX: Some thumb twiddling to prove that fetching queue data doesn't allow to distinguish bad key and non-existent queue.
    get :: SParty p -> M (Either ErrorType QueueRec)
    get party = do
      st <- asks queueStore
@@ -871,10 +901,9 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
                Just msg ->
                  let encMsg = encryptMsg qr msg
                   in atomically (setDelivered s msg) $> (corrId, rId, MSG encMsg)
-                _ -> forkSub $> resp
-            _ -> pure resp
+                _ -> forkSub $> ok
+            _ -> pure ok
          where
-            resp = (corrId, rId, OK)
            forkSub :: M ()
            forkSub = do
              atomically . modifyTVar' sub $ \s -> s {subThread = SubPending}
@@ -976,9 +1005,10 @@ saveServerMessages keepMsgs = asks (storeMsgsFile . config) >>= mapM_ saveMessag
            >>= mapM_ (B.hPutStrLn h . strEncode . MLRv3 rId)

 restoreServerMessages :: M Int
-restoreServerMessages = asks (storeMsgsFile . config) >>= \case
-  Just f -> ifM (doesFileExist f) (restoreMessages f) (pure 0)
-  Nothing -> pure 0
+restoreServerMessages =
+  asks (storeMsgsFile . config) >>= \case
+    Just f -> ifM (doesFileExist f) (restoreMessages f) (pure 0)
+    Nothing -> pure 0
  where
    restoreMessages f = do
      logInfo $ "restoring messages from file " <> T.pack f
@@ -33,8 +33,8 @@ import Simplex.Messaging.Server.Stats
 import Simplex.Messaging.Server.StoreLog
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
-import Simplex.Messaging.Transport (ATransport, VersionSMP, VersionRangeSMP)
-import Simplex.Messaging.Transport.Server (SocketState, TransportServerConfig, loadFingerprint, loadTLSServerParams, newSocketState)
+import Simplex.Messaging.Transport (ATransport, VersionRangeSMP, VersionSMP)
+import Simplex.Messaging.Transport.Server (SocketState, TransportServerConfig, alpn, loadFingerprint, loadTLSServerParams, newSocketState)
 import System.IO (IOMode (..))
 import System.Mem.Weak (Weak)
 import UnliftIO.STM
@@ -128,6 +128,7 @@ data Client = Client
    ntfSubscriptions :: TMap NotifierId (),
    rcvQ :: TBQueue (NonEmpty (Maybe QueueRec, Transmission Cmd)),
    sndQ :: TBQueue (NonEmpty (Transmission BrokerMsg)),
+    msgQ :: TBQueue (NonEmpty (Transmission BrokerMsg)),
    endThreads :: TVar (IntMap (Weak ThreadId)),
    endThreadSeq :: TVar Int,
    thVersion :: VersionSMP,
@@ -161,12 +162,13 @@ newClient nextClientId qSize thVersion sessionId createdAt = do
  ntfSubscriptions <- TM.empty
  rcvQ <- newTBQueue qSize
  sndQ <- newTBQueue qSize
+  msgQ <- newTBQueue qSize
  endThreads <- newTVar IM.empty
  endThreadSeq <- newTVar 0
  connected <- newTVar True
  rcvActiveAt <- newTVar createdAt
  sndActiveAt <- newTVar createdAt
-  return Client {clientId, subscriptions, ntfSubscriptions, rcvQ, sndQ, endThreads, endThreadSeq, thVersion, sessionId, connected, createdAt, rcvActiveAt, sndActiveAt}
+  return Client {clientId, subscriptions, ntfSubscriptions, rcvQ, sndQ, msgQ, endThreads, endThreadSeq, thVersion, sessionId, connected, createdAt, rcvActiveAt, sndActiveAt}

 newSubscription :: SubscriptionThread -> STM Sub
 newSubscription subThread = do
@@ -174,13 +176,13 @@ newSubscription subThread = do
  return Sub {subThread, delivered}

 newEnv :: ServerConfig -> IO Env
-newEnv config@ServerConfig {caCertificateFile, certificateFile, privateKeyFile, storeLogFile} = do
+newEnv config@ServerConfig {caCertificateFile, certificateFile, privateKeyFile, storeLogFile, transportConfig} = do
  server <- atomically newServer
  queueStore <- atomically newQueueStore
  msgStore <- atomically newMsgStore
  random <- liftIO C.newRandom
  storeLog <- restoreQueues queueStore `mapM` storeLogFile
-  tlsServerParams <- loadTLSServerParams caCertificateFile certificateFile privateKeyFile
+  tlsServerParams <- loadTLSServerParams caCertificateFile certificateFile privateKeyFile (alpn transportConfig)
  Fingerprint fp <- loadFingerprint caCertificateFile
  let serverIdentity = KeyHash fp
  serverStats <- atomically . newServerStats =<< getCurrentTime
@@ -25,7 +25,7 @@ import Simplex.Messaging.Server (runSMPServer)
 import Simplex.Messaging.Server.CLI
 import Simplex.Messaging.Server.Env.STM (ServerConfig (..), defMsgExpirationDays, defaultInactiveClientExpiration, defaultMessageExpiration)
 import Simplex.Messaging.Server.Expiration
-import Simplex.Messaging.Transport (simplexMQVersion, supportedServerSMPRelayVRange)
+import Simplex.Messaging.Transport (simplexMQVersion, supportedSMPHandshakes, supportedServerSMPRelayVRange)
 import Simplex.Messaging.Transport.Client (TransportHost (..))
 import Simplex.Messaging.Transport.Server (TransportServerConfig (..), defaultTransportServerConfig)
 import Simplex.Messaging.Util (safeDecodeUtf8)
@@ -211,7 +211,8 @@ smpServerCLI cfgPath logPath =
              smpServerVRange = supportedServerSMPRelayVRange,
              transportConfig =
                defaultTransportServerConfig
-                  { logTLSErrors = fromMaybe False $ iniOnOff "TRANSPORT" "log_tls_errors" ini
+                  { logTLSErrors = fromMaybe False $ iniOnOff "TRANSPORT" "log_tls_errors" ini,
+                    alpn = Just supportedSMPHandshakes
                  },
              controlPort = either (const Nothing) (Just . T.unpack) $ lookupValue "TRANSPORT" "control_port" ini
            }
@@ -8,6 +8,9 @@ module Simplex.Messaging.Server.QueueStore where
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol

+-- normalized/storage form
+-- can be split into per-party pieces
+-- id/public key form party index, dhsecrets can use dense storage or loaded on demand
 data QueueRec = QueueRec
  { recipientId :: !RecipientId,
    recipientKey :: !RcvPublicAuthKey,
@@ -24,6 +24,8 @@ where

 import Control.Monad
 import Data.Functor (($>))
+import Data.HashMap.Strict (HashMap)
+import qualified Data.HashMap.Strict as HM
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.QueueStore
 import Simplex.Messaging.TMap (TMap)
@@ -32,24 +34,32 @@ import Simplex.Messaging.Util (ifM, ($>>=))
 import UnliftIO.STM

 data QueueStore = QueueStore
-  { queues :: TMap RecipientId (TVar QueueRec),
-    senders :: TMap SenderId RecipientId,
-    notifiers :: TMap NotifierId RecipientId
+  { recipientKeys :: TVar (HashMap RecipientId RcvPublicAuthKey), -- O(1) index for const-time auth checks
+    senderKeys :: TVar (HashMap SenderId SndPublicAuthKey), -- O(1) index for const-time auth checks
+    notifierKeys :: TVar (HashMap NotifierId NtfPublicAuthKey), -- O(1) index for const-time auth checks
+    queues :: TMap RecipientId (TVar QueueRec), -- can be stored cold; and indexed with ints (using annotated keys below)
+    senders :: TMap SenderId RecipientId, -- not needed with annotated keys
+    notifiers :: TMap NotifierId RecipientId -- not needed with annotated keys
  }

 newQueueStore :: STM QueueStore
 newQueueStore = do
+  recipientKeys <- newTVar mempty
+  senderKeys <- newTVar mempty
+  notifierKeys <- newTVar mempty
  queues <- TM.empty
  senders <- TM.empty
  notifiers <- TM.empty
-  pure QueueStore {queues, senders, notifiers}
+  pure QueueStore {recipientKeys, senderKeys, notifierKeys, queues, senders, notifiers}

 addQueue :: QueueStore -> QueueRec -> STM (Either ErrorType ())
-addQueue QueueStore {queues, senders} q@QueueRec {recipientId = rId, senderId = sId} = do
+addQueue QueueStore {recipientKeys, senderKeys, queues, senders} q@QueueRec {recipientId = rId, recipientKey, senderId = sId, senderKey} = do
  ifM hasId (pure $ Left DUPLICATE_) $ do
    qVar <- newTVar q
    TM.insert rId qVar queues
+    modifyTVar' recipientKeys $ HM.insert rId recipientKey
    TM.insert sId rId senders
+    forM_ senderKey $ modifyTVar' senderKeys . HM.insert rId -- XXX: should not exist here yet, unless testing?
    pure $ Right ()
  where
    hasId = (||) <$> TM.member rId queues <*> TM.member sId senders
@@ -64,13 +74,14 @@ getQueue QueueStore {queues, senders, notifiers} party qId =
      SNotifier -> TM.lookup qId notifiers $>>= (`TM.lookup` queues)

 secureQueue :: QueueStore -> RecipientId -> SndPublicAuthKey -> STM (Either ErrorType QueueRec)
-secureQueue QueueStore {queues} rId sKey =
+secureQueue QueueStore {senderKeys, queues} rId sKey =
  withQueue rId queues $ \qVar ->
    readTVar qVar >>= \q -> case senderKey q of
      Just k -> pure $ if sKey == k then Just q else Nothing
-      _ ->
+      _ -> do
        let q' = q {senderKey = Just sKey}
-         in writeTVar qVar q' $> Just q'
+        modifyTVar' senderKeys $ HM.insert rId sKey
+        writeTVar qVar q' $> Just q'

 addQueueNotifier :: QueueStore -> RecipientId -> NtfCreds -> STM (Either ErrorType QueueRec)
 addQueueNotifier QueueStore {queues, notifiers} rId ntfCreds@NtfCreds {notifierId = nId} = do
@@ -10,7 +10,6 @@ import Data.Composition ((.:.))
 import Data.Functor (($>))
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
-import Simplex.Messaging.Util (($>>=))

 data SessionVar a = SessionVar
  { sessionVar :: TMVar a,
@@ -37,6 +36,3 @@ removeSessVar' v sessKey vs =
  TM.lookup sessKey vs >>= \case
    Just v' | sessionVarId v == sessionVarId v' -> TM.delete sessKey vs $> True
    _ -> pure False
-
-tryReadSessVar :: Ord k => k -> TMap k (SessionVar a) -> STM (Maybe a)
-tryReadSessVar sessKey vs = TM.lookup sessKey vs $>>= (tryReadTMVar . sessionVar)
@@ -33,9 +33,12 @@ module Simplex.Messaging.Transport
    VersionSMP,
    VersionRangeSMP,
    THandleSMP,
+    supportedSMPHandshakes,
    supportedClientSMPRelayVRange,
    supportedServerSMPRelayVRange,
+    legacyServerSMPRelayVRange,
    currentClientSMPRelayVersion,
+    legacyServerSMPRelayVersion,
    currentServerSMPRelayVersion,
    batchCmdsSMPVersion,
    basicAuthSMPVersion,
@@ -124,7 +127,7 @@ smpBlockSize = 16384
 -- 4 - support command batching (7/17/2022)
 -- 5 - basic auth for SMP servers (11/12/2022)
 -- 6 - allow creating queues without subscribing (9/10/2023)
-- 7 - support authenticated encryption to verify senders' commands, imply but do NOT send session ID in signed part (4/30/2024)
+-- 7 - support authenticated encryption to verify senders' commands, imply but do NOT send session ID in signed part (2/3/2024)

 data SMPVersion

@@ -150,7 +153,10 @@ authCmdsSMPVersion :: VersionSMP
 authCmdsSMPVersion = VersionSMP 7

 currentClientSMPRelayVersion :: VersionSMP
-currentClientSMPRelayVersion = VersionSMP 6
+currentClientSMPRelayVersion = VersionSMP 7
+
+legacyServerSMPRelayVersion :: VersionSMP
+legacyServerSMPRelayVersion = VersionSMP 6

 currentServerSMPRelayVersion :: VersionSMP
 currentServerSMPRelayVersion = VersionSMP 7
@@ -160,9 +166,15 @@ currentServerSMPRelayVersion = VersionSMP 7
 supportedClientSMPRelayVRange :: VersionRangeSMP
 supportedClientSMPRelayVRange = mkVersionRange batchCmdsSMPVersion currentClientSMPRelayVersion

+legacyServerSMPRelayVRange :: VersionRangeSMP
+legacyServerSMPRelayVRange = mkVersionRange batchCmdsSMPVersion legacyServerSMPRelayVersion
+
 supportedServerSMPRelayVRange :: VersionRangeSMP
 supportedServerSMPRelayVRange = mkVersionRange batchCmdsSMPVersion currentServerSMPRelayVersion

+supportedSMPHandshakes :: [ALPN]
+supportedSMPHandshakes = ["smp/1"]
+
 simplexMQVersion :: String
 simplexMQVersion = showVersion SMQ.version

@@ -194,6 +206,9 @@ class Transport c where
  -- | tls-unique channel binding per RFC5929
  tlsUnique :: c -> SessionId

+  -- | ALPN value negotiated for the session
+  getSessionALPN :: c -> Maybe ALPN
+
  -- | Close connection
  closeConnection :: c -> IO ()

@@ -288,6 +303,7 @@ instance Transport TLS where
  getServerConnection = getTLS TServer
  getClientConnection = getTLS TClient
  getServerCerts = tlsServerCerts
+  getSessionALPN = tlsALPN
  tlsUnique = tlsUniq
  closeConnection tls = closeTLS $ tlsContext tls

@@ -468,12 +484,13 @@ smpServerHandshake serverSignKey c (k, pk) kh smpVRange = do
  let th@THandle {params = THandleParams {sessionId}} = smpTHandle c
      sk = C.signX509 serverSignKey $ C.publicToX509 k
      certChain = getServerCerts c
-  sendHandshake th $ ServerHandshake {sessionId, smpVersionRange = smpVRange, authPubKey = Just (certChain, sk)}
+      smpVersionRange = maybe legacyServerSMPRelayVRange (const smpVRange) $ getSessionALPN c
+  sendHandshake th $ ServerHandshake {sessionId, smpVersionRange, authPubKey = Just (certChain, sk)}
  getHandshake th >>= \case
    ClientHandshake {smpVersion = v, keyHash, authPubKey = k'}
      | keyHash /= kh ->
          throwE $ TEHandshake IDENTITY
-      | v `isCompatible` smpVRange ->
+      | v `isCompatible` smpVersionRange ->
          pure $ smpThHandleServer th v pk k'
      | otherwise -> throwE $ TEHandshake VERSION

@@ -51,7 +51,7 @@ data HTTP2Server = HTTP2Server
 -- This server is for testing only, it processes all requests in a single queue.
 getHTTP2Server :: HTTP2ServerConfig -> IO HTTP2Server
 getHTTP2Server HTTP2ServerConfig {qSize, http2Port, bufferSize, bodyHeadSize, serverSupported, caCertificateFile, certificateFile, privateKeyFile, transportConfig} = do
-  tlsServerParams <- loadSupportedTLSServerParams serverSupported caCertificateFile certificateFile privateKeyFile
+  tlsServerParams <- loadSupportedTLSServerParams serverSupported caCertificateFile certificateFile privateKeyFile (alpn transportConfig)
  started <- newEmptyTMVarIO
  reqQ <- newTBQueueIO qSize
  action <- async $
@@ -28,10 +28,10 @@ import Control.Logger.Simple
 import Control.Monad
 import qualified Crypto.Store.X509 as SX
 import Data.Default (def)
-import Data.List (find)
 import Data.IntMap.Strict (IntMap)
 import qualified Data.IntMap.Strict as IM
-import Data.Maybe (fromJust)
+import Data.List (find)
+import Data.Maybe (fromJust, fromMaybe)
 import qualified Data.X509 as X
 import Data.X509.Validation (Fingerprint (..))
 import qualified Data.X509.Validation as XV
@@ -152,12 +152,13 @@ startTCPServer started port = withSocketsDo $ resolve >>= open >>= setStarted
      pure sock
    setStarted sock = atomically (tryPutTMVar started True) >> pure sock

-loadTLSServerParams :: FilePath -> FilePath -> FilePath -> IO T.ServerParams
+loadTLSServerParams :: FilePath -> FilePath -> FilePath -> Maybe [ALPN] -> IO T.ServerParams
 loadTLSServerParams = loadSupportedTLSServerParams supportedParameters

-loadSupportedTLSServerParams :: T.Supported -> FilePath -> FilePath -> FilePath -> IO T.ServerParams
-loadSupportedTLSServerParams serverSupported caCertificateFile certificateFile privateKeyFile =
-  fromCredential <$> loadServerCredential
+loadSupportedTLSServerParams :: T.Supported -> FilePath -> FilePath -> FilePath -> Maybe [ALPN] -> IO T.ServerParams
+loadSupportedTLSServerParams serverSupported caCertificateFile certificateFile privateKeyFile alpn_ = do
+  tlsServerParams <- fromCredential <$> loadServerCredential
+  pure tlsServerParams {T.serverHooks = maybe def alpnHooks alpn_}
  where
    loadServerCredential :: IO T.Credential
    loadServerCredential =
@@ -172,6 +173,7 @@ loadSupportedTLSServerParams serverSupported caCertificateFile certificateFile p
          T.serverHooks = def,
          T.serverSupported = serverSupported
        }
+    alpnHooks supported = def {T.onALPNClientSuggest = Just $ pure . fromMaybe "" . find (`elem` supported)}

 loadFingerprint :: FilePath -> IO Fingerprint
 loadFingerprint certificateFile = do
@@ -14,7 +14,8 @@ import Network.WebSockets
 import Network.WebSockets.Stream (Stream)
 import qualified Network.WebSockets.Stream as S
 import Simplex.Messaging.Transport
-  ( TProxy,
+  ( ALPN,
+    TProxy,
    Transport (..),
    TransportConfig (..),
    TransportError (..),
@@ -28,6 +29,7 @@ import Simplex.Messaging.Transport.Buffer (trimCR)
 data WS = WS
  { wsPeer :: TransportPeer,
    tlsUniq :: ByteString,
+    wsALPN :: Maybe ALPN,
    wsStream :: Stream,
    wsConnection :: Connection,
    wsTransportConfig :: TransportConfig,
@@ -61,6 +63,9 @@ instance Transport WS where
  getServerCerts :: WS -> X.CertificateChain
  getServerCerts = wsServerCerts

+  getSessionALPN :: WS -> Maybe ALPN
+  getSessionALPN = wsALPN
+
  tlsUnique :: WS -> ByteString
  tlsUnique = tlsUniq

@@ -90,7 +95,8 @@ getWS wsPeer cfg wsServerCerts cxt = withTlsUnique wsPeer cxt connectWS
    connectWS tlsUniq = do
      s <- makeTLSContextStream cxt
      wsConnection <- connectPeer wsPeer s
-      pure $ WS {wsPeer, tlsUniq, wsStream = s, wsConnection, wsTransportConfig = cfg, wsServerCerts}
+      wsALPN <- T.getNegotiatedProtocol cxt
+      pure $ WS {wsPeer, tlsUniq, wsALPN, wsStream = s, wsConnection, wsTransportConfig = cfg, wsServerCerts}
    connectPeer :: TransportPeer -> Stream -> IO Connection
    connectPeer TServer = acceptClientRequest
    connectPeer TClient = sendClientRequest
@@ -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 <#= \case ("", "alice", MERR 8 (SMP _ AUTH)) -> True; _ -> False
+  bob <# ("", "alice", MERR 8 (SMP AUTH))
  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 <#= \case ("", cId, MERR 8 (SMP _ AUTH)) -> cId == aliceConn; _ -> False
+  bob <# ("", aliceConn, MERR 8 (SMP AUTH))
  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 <#= \case ("", "conn1", ERR (SMP _ AUTH)) -> True; _ -> False -- this new server does not have the queue
+    client <# ("", "conn1", ERR (SMP AUTH)) -- this new server does not have the queue

 testMsgDeliveryServerRestart :: forall c. Transport c => (c, InitialKeys) -> (c, PQSupport) -> IO ()
 testMsgDeliveryServerRestart (alice, aPQ) (bob, bPQ) = do
@@ -630,7 +630,7 @@ syntaxTests t = do
            <> " subscribe "
            <> "14\nbob's connInfo"
        )
-          >#> ("311", "a", "ERR SMP smp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:5001 AUTH")
+          >#> ("311", "a", "ERR SMP AUTH")
    describe "invalid" $ do
      it "no parameters" $ ("321", "", "JOIN") >#> ("321", "", "ERR CMD SYNTAX")
  where
@@ -58,10 +58,10 @@ import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Either (isRight)
 import Data.Int (Int64)
-import Data.List (nub)
+import Data.List (find, nub)
 import Data.List.NonEmpty (NonEmpty)
 import qualified Data.Map as M
-import Data.Maybe (isNothing)
+import Data.Maybe (isJust, isNothing)
 import qualified Data.Set as S
 import Data.Time.Clock (diffUTCTime, getCurrentTime)
 import Data.Time.Clock.System (SystemTime (..), getSystemTime)
@@ -80,7 +80,7 @@ import qualified Simplex.Messaging.Agent.Protocol as A
 import Simplex.Messaging.Agent.RetryInterval (RetryInterval (..))
 import Simplex.Messaging.Agent.Store.SQLite (MigrationConfirmation (..), SQLiteStore (dbNew))
 import Simplex.Messaging.Agent.Store.SQLite.Common (withTransaction')
-import Simplex.Messaging.Client (NetworkConfig (..), ProtocolClientConfig (..), TransportSessionMode (TSMEntity, TSMUser), defaultSMPClientConfig)
+import Simplex.Messaging.Client (NetworkConfig (..), ProtocolClientConfig (..), TransportSessionMode (TSMEntity, TSMUser), defaultClientConfig)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.Ratchet (InitialKeys (..), PQEncryption (..), PQSupport (..), pattern PQEncOff, pattern PQEncOn, pattern PQSupportOff, pattern PQSupportOn)
 import qualified Simplex.Messaging.Crypto.Ratchet as CR
@@ -90,7 +90,7 @@ import Simplex.Messaging.Protocol (BasicAuth, ErrorType (..), MsgBody, ProtocolS
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
 import Simplex.Messaging.Server.Expiration
-import Simplex.Messaging.Transport (ATransport (..), SMPVersion, VersionSMP, authCmdsSMPVersion, basicAuthSMPVersion, batchCmdsSMPVersion, currentServerSMPRelayVersion)
+import Simplex.Messaging.Transport (ATransport (..), SMPVersion, VersionSMP, authCmdsSMPVersion, basicAuthSMPVersion, batchCmdsSMPVersion, currentServerSMPRelayVersion, supportedSMPHandshakes)
 import Simplex.Messaging.Util (diffToMicroseconds)
 import Simplex.Messaging.Version (VersionRange (..))
 import qualified Simplex.Messaging.Version as V
@@ -175,13 +175,16 @@ pattern Rcvd :: AgentMsgId -> ACommand 'Agent 'AEConn
 pattern Rcvd agentMsgId <- RCVD MsgMeta {integrity = MsgOk} [MsgReceipt {agentMsgId, msgRcptStatus = MROk}]

 smpCfgVPrev :: ProtocolClientConfig SMPVersion
-smpCfgVPrev = (smpCfg agentCfg) {serverVRange = prevRange $ serverVRange $ smpCfg agentCfg}
+smpCfgVPrev = (smpCfg agentCfg) {clientALPN = Nothing, serverVRange = prevRange $ serverVRange $ smpCfg agentCfg}

 smpCfgV7 :: ProtocolClientConfig SMPVersion
 smpCfgV7 = (smpCfg agentCfg) {serverVRange = V.mkVersionRange batchCmdsSMPVersion authCmdsSMPVersion}

+ntfCfgVPrev :: ProtocolClientConfig NTFVersion
+ntfCfgVPrev = (ntfCfg agentCfg) {clientALPN = Nothing, serverVRange = V.mkVersionRange (VersionNTF 1) (VersionNTF 1)}
+
 ntfCfgV2 :: ProtocolClientConfig NTFVersion
-ntfCfgV2 = (smpCfg agentCfg) {serverVRange = V.mkVersionRange (VersionNTF 1) authBatchCmdsNTFVersion}
+ntfCfgV2 = (ntfCfg agentCfg) {serverVRange = V.mkVersionRange (VersionNTF 1) authBatchCmdsNTFVersion}

 agentCfgVPrev :: AgentConfig
 agentCfgVPrev =
@@ -190,7 +193,8 @@ agentCfgVPrev =
      smpAgentVRange = prevRange $ smpAgentVRange agentCfg,
      smpClientVRange = prevRange $ smpClientVRange agentCfg,
      e2eEncryptVRange = prevRange $ e2eEncryptVRange agentCfg,
-      smpCfg = smpCfgVPrev
+      smpCfg = smpCfgVPrev,
+      ntfCfg = ntfCfgVPrev
    }

 -- agent config for the next client version
@@ -333,6 +337,9 @@ functionalAPITests t = do
    skip "faster version of the previous test (200 subscriptions gets very slow with test coverage)" $
      it "should subscribe to multiple (6) subscriptions with batching" $
        testBatchedSubscriptions 6 3 t
+    it "should subscribe to multiple connections with pending messages" $
+      withSmpServer t $
+        testBatchedPendingMessages 10 5
  describe "Async agent commands" $ do
    it "should connect using async agent commands" $
      withSmpServer t testAsyncCommands
@@ -415,7 +422,7 @@ functionalAPITests t = do
    describe "server with password" $ do
      let auth = Just "abcd"
          srv = ProtoServerWithAuth testSMPServer2
-          authErr = Just (ProtocolTestFailure TSCreateQueue $ SMP (B.unpack $ strEncode testSMPServer2) AUTH)
+          authErr = Just (ProtocolTestFailure TSCreateQueue $ SMP 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
@@ -530,7 +537,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 =##> \case ("", cId, MERR mId (SMP _ AUTH)) -> cId == aliceId && mId == (baseId + 5); _ -> False
+    get bob ##> ("", aliceId, MERR (baseId + 5) (SMP AUTH))
    deleteConnection alice bobId
    liftIO $ noMessages alice "nothing else should be delivered to alice"
  where
@@ -662,7 +669,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 =##> \case ("", cId, MERR mId (SMP _ AUTH)) -> cId == aliceId && mId == (baseId + 5); _ -> False
+    get bob ##> ("", aliceId, MERR (baseId + 5) (SMP AUTH))
    deleteConnection alice bobId
    liftIO $ noMessages alice "nothing else should be delivered to alice"
  where
@@ -1108,7 +1115,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
@@ -1136,15 +1143,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
@@ -1395,10 +1402,10 @@ makeConnection = makeConnection_ PQSupportOn
 makeConnection_ :: PQSupport -> AgentClient -> AgentClient -> ExceptT AgentErrorType IO (ConnId, ConnId)
 makeConnection_ pqEnc alice bob = makeConnectionForUsers_ pqEnc alice 1 bob 1

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

-makeConnectionForUsers_ :: HasCallStack => PQSupport -> AgentClient -> UserId -> AgentClient -> UserId -> ExceptT AgentErrorType IO (ConnId, ConnId)
+makeConnectionForUsers_ :: 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.prepareConnectionToJoin bob bobUserId True qInfo pqSupport
@@ -1530,7 +1537,7 @@ testBatchedSubscriptions :: Int -> Int -> ATransport -> IO ()
 testBatchedSubscriptions nCreate nDel t =
  withAgentClientsCfgServers2 agentCfg agentCfg initAgentServers2 $ \a b -> do
    conns <- runServers $ do
-      conns <- replicateM (nCreate :: Int) $ makeConnection_ PQSupportOff a b
+      conns <- replicateM nCreate $ makeConnection_ PQSupportOff a b
      forM_ conns $ \(aId, bId) -> exchangeGreetings_ PQEncOff a bId b aId
      let (aIds', bIds') = unzip $ take nDel conns
      delete a bIds'
@@ -1589,6 +1596,25 @@ testBatchedSubscriptions nCreate nDel t =
        killThread t1
        pure res

+testBatchedPendingMessages :: Int -> Int -> IO ()
+testBatchedPendingMessages nCreate nMsgs =
+  withA $ \a -> do
+    conns <- withB $ \b -> runRight $ do
+      replicateM nCreate $ makeConnection a b
+    let msgConns = take nMsgs conns
+    runRight_ $ forM_ msgConns $ \(_, bId) -> sendMessage a bId SMP.noMsgFlags "hello"
+    replicateM_ nMsgs $ get a =##> \case ("", cId, SENT _) -> isJust $ find ((cId ==) . snd) msgConns; _ -> False
+    withB $ \b -> runRight_ $ do
+      r <- subscribeConnections b $ map fst conns
+      liftIO $ all isRight r `shouldBe` True
+      replicateM_ nMsgs $ do
+        ("", cId, Msg' msgId _ "hello") <- get b
+        liftIO $ isJust (find ((cId ==) . fst) msgConns) `shouldBe` True
+        ackMessage b cId msgId Nothing
+  where
+    withA = withAgent 1 agentCfg initAgentServers testDB
+    withB = withAgent 2 agentCfg initAgentServers testDB2
+
 testAsyncCommands :: IO ()
 testAsyncCommands =
  withAgentClients2 $ \alice bob -> runRight_ $ do
@@ -1683,7 +1709,7 @@ testAcceptContactAsync =
    ackMessage alice bobId (baseId + 4) Nothing
    suspendConnection alice bobId
    5 <- msgId <$> sendMessage bob aliceId SMP.noMsgFlags "message 2"
-    get bob =##> \case ("", cId, MERR mId (SMP _ AUTH)) -> cId == aliceId && mId == (baseId + 5); _ -> False
+    get bob ##> ("", aliceId, MERR (baseId + 5) (SMP AUTH))
    deleteConnection alice bobId
    liftIO $ noMessages alice "nothing else should be delivered to alice"
  where
@@ -1729,7 +1755,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 =##> \case ("", cId, MERR mId (SMP _ AUTH)) -> cId == aliceId && mId == (baseId + 3); _ -> False
+    get bob ##> ("", aliceId, MERR (baseId + 3) (SMP AUTH))

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

      withSmpServerStoreLogOn t testPort $ \_ -> do
-        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 ##> ("", bobId, MERR (baseId + 3) (SMP AUTH))
+        get alice ##> ("", bobId, MERR (baseId + 4) (SMP AUTH))
        get alice =##> \case ("", cId, DEL_CONN) -> cId == bobId; _ -> False

        liftIO $ noMessages alice "nothing else should be delivered to alice"
@@ -2396,11 +2422,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
@@ -2416,7 +2442,8 @@ testCreateQueueAuth srvVersion clnt1 clnt2 = do
  where
    getClient clientId (clntAuth, clntVersion) db =
      let servers = initAgentServers {smp = userServers [ProtoServerWithAuth testSMPServer clntAuth]}
-          smpCfg = (defaultSMPClientConfig :: ProtocolClientConfig SMPVersion) {serverVRange = V.mkVersionRange (prevVersion basicAuthSMPVersion) clntVersion}
+          alpn_ = if clntVersion >= authCmdsSMPVersion then Just supportedSMPHandshakes else Nothing
+          smpCfg = defaultClientConfig alpn_ $ V.mkVersionRange (prevVersion basicAuthSMPVersion) clntVersion
          sndAuthAlg = if srvVersion >= authCmdsSMPVersion && clntVersion >= authCmdsSMPVersion then C.AuthAlg C.SX25519 else C.AuthAlg C.SEd25519
       in getSMPAgentClient' clientId agentCfg {smpCfg, sndAuthAlg} servers db

@@ -6,8 +6,8 @@
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE PatternSynonyms #-}
 {-# LANGUAGE TypeApplications #-}
-{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
 {-# OPTIONS_GHC -fno-warn-incomplete-uni-patterns #-}
+{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}

 module AgentTests.NotificationTests where

@@ -17,6 +17,10 @@ import AgentTests.FunctionalAPITests
    createConnection,
    exchangeGreetingsMsgId,
    get,
+    withAgent,
+    withAgentClients2,
+    withAgentClientsCfgServers2,
+    withAgentClients3,
    joinConnection,
    makeConnection,
    nGet,
@@ -25,11 +29,7 @@ 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.Protocol
 import Simplex.Messaging.Notifications.Server.Env (NtfServerConfig (..))
+import Simplex.Messaging.Notifications.Protocol
 import Simplex.Messaging.Notifications.Server.Push.APNS
 import Simplex.Messaging.Notifications.Types (NtfToken (..))
 import Simplex.Messaging.Protocol (ErrorType (AUTH), MsgFlags (MsgFlags), NtfServer, ProtocolServer (..), SMPMsgMeta (..), SubscriptionMode (..))
@@ -151,8 +151,7 @@ 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
@@ -244,7 +243,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 ()
@@ -259,7 +258,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,
@@ -267,7 +266,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,9 +35,6 @@ 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)

@@ -36,6 +36,7 @@ import Simplex.Messaging.Encoding
 import Simplex.Messaging.Notifications.Protocol (NtfResponse)
 import Simplex.Messaging.Notifications.Server (runNtfServerBlocking)
 import Simplex.Messaging.Notifications.Server.Env
+import qualified Simplex.Messaging.Notifications.Server.Env as Env
 import Simplex.Messaging.Notifications.Server.Push.APNS
 import Simplex.Messaging.Notifications.Server.Push.APNS.Internal
 import Simplex.Messaging.Notifications.Transport
@@ -45,6 +46,7 @@ import Simplex.Messaging.Transport.Client
 import Simplex.Messaging.Transport.HTTP2 (HTTP2Body (..), http2TLSParams)
 import Simplex.Messaging.Transport.HTTP2.Server
 import Simplex.Messaging.Transport.Server
+import qualified Simplex.Messaging.Transport.Server as Server
 import Simplex.Messaging.Version (mkVersionRange)
 import Test.Hspec
 import UnliftIO.Async
@@ -113,7 +115,8 @@ ntfServerCfgV2 :: NtfServerConfig
 ntfServerCfgV2 =
  ntfServerCfg
    { ntfServerVRange = mkVersionRange initialNTFVersion authBatchCmdsNTFVersion,
-      smpAgentCfg = defaultSMPClientAgentConfig {smpCfg = (smpCfg defaultSMPClientAgentConfig) {serverVRange = mkVersionRange batchCmdsSMPVersion authCmdsSMPVersion}}
+      smpAgentCfg = defaultSMPClientAgentConfig {smpCfg = (smpCfg defaultSMPClientAgentConfig) {serverVRange = mkVersionRange batchCmdsSMPVersion authCmdsSMPVersion}},
+      Env.transportConfig = defaultTransportServerConfig {Server.alpn = Just supportedNTFHandshakes}
    }

 withNtfServerStoreLog :: ATransport -> (ThreadId -> IO a) -> IO a
@@ -24,8 +24,10 @@ import Simplex.Messaging.Server (runSMPServerBlocking)
 import Simplex.Messaging.Server.Env.STM
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Client
+import qualified Simplex.Messaging.Transport.Client as Client
 import Simplex.Messaging.Transport.Server
-import Simplex.Messaging.Version (mkVersionRange)
+import qualified Simplex.Messaging.Transport.Server as Server
+import Simplex.Messaging.Version
 import System.Environment (lookupEnv)
 import System.Info (os)
 import Test.Hspec
@@ -73,10 +75,15 @@ testSMPClient = testSMPClientVR supportedClientSMPRelayVRange
 testSMPClientVR :: Transport c => VersionRangeSMP -> (THandleSMP c 'TClient -> IO a) -> IO a
 testSMPClientVR vr client = do
  Right useHost <- pure $ chooseTransportHost defaultNetworkConfig testHost
-  runTransportClient defaultTransportClientConfig Nothing useHost testPort (Just testKeyHash) $ \h ->
+  let tcConfig = defaultTransportClientConfig {Client.alpn = clientALPN}
+  runTransportClient tcConfig Nothing useHost testPort (Just testKeyHash) $ \h ->
    runExceptT (smpClientHandshake h Nothing testKeyHash vr) >>= \case
      Right th -> client th
      Left e -> error $ show e
+  where
+    clientALPN
+      | authCmdsSMPVersion `isCompatible` vr = Just supportedSMPHandshakes
+      | otherwise = Nothing

 cfg :: ServerConfig
 cfg =
@@ -104,7 +111,7 @@ cfg =
      privateKeyFile = "tests/fixtures/server.key",
      certificateFile = "tests/fixtures/server.crt",
      smpServerVRange = supportedServerSMPRelayVRange,
-      transportConfig = defaultTransportServerConfig,
+      transportConfig = defaultTransportServerConfig {Server.alpn = Just supportedSMPHandshakes},
      controlPort = Nothing
    }

@@ -62,7 +62,7 @@ serverTests t@(ATransport t') = do
  describe "Store log" $ testWithStoreLog t
  describe "Restore messages" $ testRestoreMessages t
  describe "Restore messages (old / v2)" $ testRestoreExpireMessages t
-  describe "Timing of AUTH error" $ testTiming t
+  fdescribe "Timing of AUTH error" $ testTiming t
  describe "Message notifications" $ testMessageNotifications t
  describe "Message expiration" $ do
    testMsgExpireOnSend t'
@@ -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 (B.unpack $ strEncode testXFTPServer2) AUTH)
+          authErr = Just (ProtocolTestFailure TSCreateFile $ XFTP 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,8 +392,7 @@ 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 {serverAddress = \"xftp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:7000\", xftpErr = AUTH}")) <-
-        rfGet rcp'
+      ("", rfId', RFERR (INTERNAL "XFTP {xftpErr = AUTH}")) <- rfGet rcp'
      rfId' `shouldBe` rfId

  -- tmp path should be removed after permanent error
@@ -472,8 +471,7 @@ 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 {serverAddress = \"xftp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:7000\", xftpErr = AUTH}")) <-
-        sfGet sndr'
+      ("", sfId', SFERR (INTERNAL "XFTP {xftpErr = AUTH}")) <- sfGet sndr'
      sfId' `shouldBe` sfId

  -- prefix path should be removed after permanent error
@@ -508,8 +506,7 @@ 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 {serverAddress = \"xftp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:7000\", xftpErr = AUTH}")) <-
-          rfGet rcp2
+        ("", rfId', RFERR (INTERNAL "XFTP {xftpErr = AUTH}")) <- rfGet rcp2
        liftIO $ rfId' `shouldBe` rfId

 testXFTPAgentDeleteRestore :: HasCallStack => IO ()
@@ -546,8 +543,7 @@ 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 {serverAddress = \"xftp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:7000\", xftpErr = AUTH}")) <-
-          rfGet rcp2
+        ("", rfId', RFERR (INTERNAL "XFTP {xftpErr = AUTH}")) <- rfGet rcp2
        liftIO $ rfId' `shouldBe` rfId

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

          -- receive file 2
@@ -614,8 +609,7 @@ 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 {serverAddress = \"xftp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:7000\", xftpErr = AUTH}")) <-
-            rfGet rcp
+          ("", rfId', RFERR (INTERNAL "XFTP {xftpErr = AUTH}")) <- rfGet rcp
          liftIO $ rfId' `shouldBe` rfId

        -- create and send file 2
Author	SHA1	Message	Date
Alexander Bondarenko	5aa2972774	WIP	2024-05-12 21:19:27 +03:00
Evgeny Poberezkin	1339a8da11	5.7.4.0	2024-05-11 23:39:28 +01:00
Evgeny Poberezkin	103ae06d55	agent: remove critical error on subscription timeout (#1146 )	2024-05-11 23:38:27 +01:00
Evgeny Poberezkin	acc7faea11	5.7.3.1	2024-05-10 22:24:00 +01:00
Evgeny PoberezkinandAlexander Bondarenko	727fd8b8f5	server: more efficient response to batched subscriptions (#1141 ) * server: more efficient response to batched subscriptions * add sndMsgQ for interleaving messages with replies * remove redundant liftIO * refactor * refactor2 * rename * fix * diff * remove comment * remove comment --------- Co-authored-by: Alexander Bondarenko <[email protected]>	2024-05-10 22:19:11 +01:00
Evgeny Poberezkin	dc111437fd	5.7.3.0	2024-05-10 15:15:43 +01:00
Evgeny Poberezkin	5cafd9d5c4	server: more efficient responses to batch subscriptions (#1137 ) * server: more efficient responses to batch subscriptions * comments * comment * enable tests * LogError	2024-05-09 09:20:57 +01:00
Evgeny Poberezkin	ea21b296fd	agent: reset error count and do not report errors when consequitive timeouts happen while offline (#1136 ) * agent: reset error count and do not report errors when consequitive timeouts happen while offline * refactor * comment	2024-05-08 15:33:51 +01:00
spaced4ndy	3f57d54832	xftp: catch exceptions in chunk download (#1133 )	2024-05-08 16:57:04 +04:00
Evgeny Poberezkin	b40654c95d	update agent to v7/v2 SMP/NTF protocol versions (#997 ) * update agent to v7/v2 SMP/NTF protocol versions	2024-05-08 13:05:06 +01:00
Alexander BondarenkoandEvgeny Poberezkin	7a07076277	transport: require ALPN for extended handshakes (#1134 ) * transport: require ALPN for extended handshakes * fix 8.10 build * rename --------- Co-authored-by: Evgeny Poberezkin <[email protected]>	2024-05-08 00:06:09 +01:00