add client tracking of SUB'd connections in it

* upgrade SMP/NTF servers to v7/v2 protocol versions

* 5.6.0.0

---------

Co-authored-by: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com>

CHANGELOG.md

@@ -1,3 +1,28 @@
+# 5.7.1
+
+SMP agent:
+- increase timeout for TLS connection via SOCKS
+
+# 5.7.0
+
+Version 5.7.0.4
+
+_Please note_: the earliest SimpleX Chat clients supported by this version of the servers is 5.5.3 (released on February 11, 2024).
+
+SMP server:
+- increase max SMP protocol version to 7 (support for deniable authenticators).
+
+NTF server:
+- increase max NTF protocol version to 2 (support for deniable authenticators).
+
+XFTP server:
+- version handshake using ALPN.
+
+SMP agent:
+- increase timeouts for XFTP files.
+- don't send commands after timeout.
+- PQ encryption support.
+
 # 5.6.2
 
 Version 5.6.2.2.

package.yaml

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

simplexmq.cabal

@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack
 
 name:           simplexmq
-version:        5.7.0.0
+version:        5.7.1.0
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and

src/Simplex/Messaging/Agent.hs

@@ -117,7 +117,7 @@ module Simplex.Messaging.Agent
   )
 where
 
-import Control.Logger.Simple (logError, logInfo, showText)
+import Control.Logger.Simple
 import Control.Monad
 import Control.Monad.Except
 import Control.Monad.Reader
@@ -149,7 +149,7 @@ import Simplex.FileTransfer.Protocol (FileParty (..))
 import Simplex.FileTransfer.Util (removePath)
 import Simplex.Messaging.Agent.Client
 import Simplex.Messaging.Agent.Env.SQLite
-import Simplex.Messaging.Agent.Lock (withLock', withLock)
+import Simplex.Messaging.Agent.Lock (withLock, withLock')
 import Simplex.Messaging.Agent.NtfSubSupervisor
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval
@@ -160,7 +160,7 @@ import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
 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 PQEncOn, pattern PQEncOff, pattern PQSupportOn, pattern PQSupportOff)
+import Simplex.Messaging.Crypto.Ratchet (PQEncryption, PQSupport (..), pattern PQEncOff, pattern PQEncOn, pattern PQSupportOff, pattern PQSupportOn)
 import qualified Simplex.Messaging.Crypto.Ratchet as CR
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
@@ -198,7 +198,7 @@ getSMPAgentClient_ clientId cfg initServers store backgroundMode =
   liftIO $ newSMPAgentEnv cfg store >>= runReaderT runAgent
   where
     runAgent = do
-      c@AgentClient {acThread}  <- atomically . newAgentClient clientId initServers =<< ask
+      c@AgentClient {acThread} <- atomically . newAgentClient clientId initServers =<< ask
       t <- runAgentThreads c `forkFinally` const (liftIO $ disconnectAgentClient c)
       atomically . writeTVar acThread . Just =<< mkWeakThreadId t
       pure c
@@ -239,7 +239,7 @@ createUser c = withAgentEnv c .: createUser' c
 {-# INLINE createUser #-}
 
 -- | Delete user record optionally deleting all user's connections on SMP servers
-deleteUser :: AgentClient -> UserId -> Bool -> AE  ()
+deleteUser :: AgentClient -> UserId -> Bool -> AE ()
 deleteUser c = withAgentEnv c .: deleteUser' c
 {-# INLINE deleteUser #-}
 
@@ -618,14 +618,14 @@ newConnAsync c userId corrId enableNtfs cMode pqInitKeys subMode = do
 newConnNoQueues :: AgentClient -> UserId -> ConnId -> Bool -> SConnectionMode c -> PQSupport -> AM ConnId
 newConnNoQueues c userId connId enableNtfs cMode pqSupport = do
   g <- asks random
-  connAgentVersion <- asks $ maxVersion . ($ pqSupport) . smpAgentVRange . config
+  connAgentVersion <- asks $ maxVersion . smpAgentVRange . config
   let cData = ConnData {userId, connId, connAgentVersion, enableNtfs, lastExternalSndId = 0, deleted = False, ratchetSyncState = RSOk, pqSupport}
   withStore c $ \db -> createNewConn db g cData cMode
 
 joinConnAsync :: AgentClient -> UserId -> ACorrId -> Bool -> ConnectionRequestUri c -> ConnInfo -> PQSupport -> SubscriptionMode -> AM ConnId
 joinConnAsync c userId corrId enableNtfs cReqUri@CRInvitationUri {} cInfo pqSup subMode = do
   withInvLock c (strEncode cReqUri) "joinConnAsync" $ do
-    lift (compatibleInvitationUri cReqUri pqSup) >>= \case
+    lift (compatibleInvitationUri cReqUri) >>= \case
       Just (_, Compatible (CR.E2ERatchetParams v _ _ _), Compatible connAgentVersion) -> do
         g <- asks random
         let pqSupport = pqSup `CR.pqSupportAnd` versionPQSupport_ connAgentVersion (Just v)
@@ -721,24 +721,20 @@ 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) <- newRcvQueue c userId connId srv smpClientVRange subMode `catchAgentError` \e -> liftIO (print e) >> throwError e
+  (rq, qUri, tSess, sessId) <- newRcvQueue c userId connId srv smpClientVRange subMode `catchAgentError` \e -> liftIO (print e) >> throwError e
   rq' <- withStore c $ \db -> updateNewConnRcv db connId rq
-  liftIO $ case subMode of
-    SMOnlyCreate -> pure ()
-    SMSubscribe -> addSubscription c rq'
+  lift . when (subMode == SMSubscribe) $ addNewQueueSubscription c rq' tSess sessId
   when enableNtfs $ do
     ns <- asks ntfSupervisor
     atomically $ sendNtfSubCommand ns (connId, NSCCreate)
-  let pqEnc = CR.connPQEncryption pqInitKeys
-      crData = ConnReqUriData SSSimplex (smpAgentVRange pqEnc) [qUri] clientData
-      e2eVRange = e2eEncryptVRange pqEnc
+  let crData = ConnReqUriData SSSimplex smpAgentVRange [qUri] clientData
   case cMode of
     SCMContact -> pure (connId, CRContactUri crData)
     SCMInvitation -> do
       g <- asks random
-      (pk1, pk2, pKem, e2eRcvParams) <- liftIO $ CR.generateRcvE2EParams g (maxVersion e2eVRange) (CR.initialPQEncryption pqInitKeys)
+      (pk1, pk2, pKem, e2eRcvParams) <- liftIO $ CR.generateRcvE2EParams g (maxVersion e2eEncryptVRange) (CR.initialPQEncryption pqInitKeys)
       withStore' c $ \db -> createRatchetX3dhKeys db connId pk1 pk2 pKem
-      pure (connId, CRInvitationUri crData $ toVersionRangeT e2eRcvParams e2eVRange)
+      pure (connId, CRInvitationUri crData $ toVersionRangeT e2eRcvParams e2eEncryptVRange)
 
 joinConn :: AgentClient -> UserId -> ConnId -> Bool -> ConnectionRequestUri c -> ConnInfo -> PQSupport -> SubscriptionMode -> AM ConnId
 joinConn c userId connId enableNtfs cReq cInfo pqSupport subMode = do
@@ -750,14 +746,14 @@ joinConn c userId connId enableNtfs cReq cInfo pqSupport subMode = do
 
 startJoinInvitation :: UserId -> ConnId -> Bool -> ConnectionRequestUri 'CMInvitation -> PQSupport -> AM (Compatible VersionSMPA, ConnData, NewSndQueue, CR.Ratchet 'C.X448, CR.SndE2ERatchetParams 'C.X448)
 startJoinInvitation userId connId enableNtfs cReqUri pqSup =
-  lift (compatibleInvitationUri cReqUri pqSup) >>= \case
+  lift (compatibleInvitationUri cReqUri) >>= \case
     Just (qInfo, (Compatible e2eRcvParams@(CR.E2ERatchetParams v _ rcDHRr kem_)), aVersion@(Compatible connAgentVersion)) -> do
       g <- asks random
       let pqSupport = pqSup `CR.pqSupportAnd` versionPQSupport_ connAgentVersion (Just v)
       (pk1, pk2, pKem, e2eSndParams) <- liftIO $ CR.generateSndE2EParams g v (CR.replyKEM_ v kem_ pqSupport)
       (_, rcDHRs) <- atomically $ C.generateKeyPair g
       rcParams <- liftEitherWith cryptoError $ CR.pqX3dhSnd pk1 pk2 pKem e2eRcvParams
-      maxSupported <- asks $ maxVersion . ($ pqSup) . e2eEncryptVRange . config
+      maxSupported <- asks $ maxVersion . e2eEncryptVRange . config
       let rcVs = CR.RatchetVersions {current = v, maxSupported}
           rc = CR.initSndRatchet rcVs rcDHRr rcDHRs rcParams
       q <- lift $ newSndQueue userId "" qInfo
@@ -767,29 +763,29 @@ startJoinInvitation userId connId enableNtfs cReqUri pqSup =
 
 connRequestPQSupport :: AgentClient -> PQSupport -> ConnectionRequestUri c -> IO (Maybe (VersionSMPA, PQSupport))
 connRequestPQSupport c pqSup cReq = withAgentEnv' c $ case cReq of
-  CRInvitationUri {} -> invPQSupported <$$> compatibleInvitationUri cReq pqSup
+  CRInvitationUri {} -> invPQSupported <$$> compatibleInvitationUri cReq
     where
       invPQSupported (_, Compatible (CR.E2ERatchetParams e2eV _ _ _), Compatible agentV) = (agentV, pqSup `CR.pqSupportAnd` versionPQSupport_ agentV (Just e2eV))
-  CRContactUri {} -> ctPQSupported <$$> compatibleContactUri cReq pqSup
+  CRContactUri {} -> ctPQSupported <$$> compatibleContactUri cReq
     where
       ctPQSupported (_, Compatible agentV) = (agentV, pqSup `CR.pqSupportAnd` versionPQSupport_ agentV Nothing)
 
-compatibleInvitationUri :: ConnectionRequestUri 'CMInvitation -> PQSupport -> AM' (Maybe (Compatible SMPQueueInfo, Compatible (CR.RcvE2ERatchetParams 'C.X448), Compatible VersionSMPA))
-compatibleInvitationUri (CRInvitationUri ConnReqUriData {crAgentVRange, crSmpQueues = (qUri :| _)} e2eRcvParamsUri) pqSup = do
+compatibleInvitationUri :: ConnectionRequestUri 'CMInvitation -> AM' (Maybe (Compatible SMPQueueInfo, Compatible (CR.RcvE2ERatchetParams 'C.X448), Compatible VersionSMPA))
+compatibleInvitationUri (CRInvitationUri ConnReqUriData {crAgentVRange, crSmpQueues = (qUri :| _)} e2eRcvParamsUri) = do
   AgentConfig {smpClientVRange, smpAgentVRange, e2eEncryptVRange} <- asks config
   pure $
     (,,)
       <$> (qUri `compatibleVersion` smpClientVRange)
-      <*> (e2eRcvParamsUri `compatibleVersion` e2eEncryptVRange pqSup)
-      <*> (crAgentVRange `compatibleVersion` smpAgentVRange pqSup)
+      <*> (e2eRcvParamsUri `compatibleVersion` e2eEncryptVRange)
+      <*> (crAgentVRange `compatibleVersion` smpAgentVRange)
 
-compatibleContactUri :: ConnectionRequestUri 'CMContact -> PQSupport -> AM' (Maybe (Compatible SMPQueueInfo, Compatible VersionSMPA))
-compatibleContactUri (CRContactUri ConnReqUriData {crAgentVRange, crSmpQueues = (qUri :| _)}) pqSup = do
+compatibleContactUri :: ConnectionRequestUri 'CMContact -> AM' (Maybe (Compatible SMPQueueInfo, Compatible VersionSMPA))
+compatibleContactUri (CRContactUri ConnReqUriData {crAgentVRange, crSmpQueues = (qUri :| _)}) = do
   AgentConfig {smpClientVRange, smpAgentVRange} <- asks config
   pure $
     (,)
-      <$> (qUri `compatibleVersion` smpClientVRange) 
-      <*> (crAgentVRange `compatibleVersion` smpAgentVRange pqSup)
+      <$> (qUri `compatibleVersion` smpClientVRange)
+      <*> (crAgentVRange `compatibleVersion` smpAgentVRange)
 
 versionPQSupport_ :: VersionSMPA -> Maybe CR.VersionE2E -> PQSupport
 versionPQSupport_ agentV e2eV_ = PQSupport $ agentV >= pqdrSMPAgentVersion && maybe True (>= CR.pqRatchetE2EEncryptVersion) e2eV_
@@ -812,7 +808,7 @@ joinConnSrv c userId connId enableNtfs inv@CRInvitationUri {} cInfo pqSup subMod
         void $ withStore' c $ \db -> deleteConn db Nothing connId'
         throwError e
 joinConnSrv c userId connId enableNtfs cReqUri@CRContactUri {} cInfo pqSup subMode srv =
-  lift (compatibleContactUri cReqUri pqSup) >>= \case
+  lift (compatibleContactUri cReqUri) >>= \case
     Just (qInfo, vrsn) -> do
       (connId', cReq) <- newConnSrv c userId connId enableNtfs SCMInvitation Nothing (CR.IKNoPQ pqSup) subMode srv
       sendInvitation c userId qInfo vrsn cReq cInfo
@@ -831,12 +827,10 @@ 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) <- newRcvQueue c userId connId srv (versionToRange smpClientVersion) subMode
+  (rq, qUri, tSess, sessId) <- newRcvQueue c userId connId srv (versionToRange smpClientVersion) subMode
   let qInfo = toVersionT qUri smpClientVersion
   rq' <- withStore c $ \db -> upgradeSndConnToDuplex db connId rq
-  liftIO $ case subMode of
-    SMOnlyCreate -> pure ()
-    SMSubscribe -> addSubscription c rq'
+  lift . when (subMode == SMSubscribe) $ addNewQueueSubscription c rq' tSess sessId
   when enableNtfs $ do
     ns <- asks ntfSupervisor
     atomically $ sendNtfSubCommand ns (connId, NSCCreate)
@@ -896,7 +890,7 @@ subscribeConnections' c connIds = do
       (subRs, rcvQs) = M.mapEither rcvQueueOrResult cs
   mapM_ (mapM_ (\(cData, sqs) -> mapM_ (lift . resumeMsgDelivery c cData) sqs) . sndQueue) cs
   mapM_ (resumeConnCmds c) $ M.keys cs
-  rcvRs <- lift $ connResults <$> subscribeQueues c (concat $ M.elems rcvQs)
+  rcvRs <- lift $ connResults . fst <$> subscribeQueues c (concat $ M.elems rcvQs)
   ns <- asks ntfSupervisor
   tkn <- readTVarIO (ntfTkn ns)
   when (instantNotifications tkn) . void . lift . forkIO . void . runExceptT $ sendNtfCreate ns rcvRs conns
@@ -1209,7 +1203,7 @@ enqueueMessage c cData sq msgFlags aMessage =
 {-# INLINE enqueueMessage #-}
 
 -- this function is used only for sending messages in batch, it returns the list of successes to enqueue additional deliveries
-enqueueMessageB :: forall t. (Traversable t) => AgentClient -> t (Either AgentErrorType (ConnData, NonEmpty SndQueue, Maybe PQEncryption, MsgFlags, AMessage)) -> AM' (t (Either AgentErrorType ((AgentMsgId, PQEncryption), Maybe (ConnData, [SndQueue], AgentMsgId))))
+enqueueMessageB :: forall t. Traversable t => AgentClient -> t (Either AgentErrorType (ConnData, NonEmpty SndQueue, Maybe PQEncryption, MsgFlags, AMessage)) -> AM' (t (Either AgentErrorType ((AgentMsgId, PQEncryption), Maybe (ConnData, [SndQueue], AgentMsgId))))
 enqueueMessageB c reqs = do
   cfg <- asks config
   reqMids <- withStoreBatch c $ \db -> fmap (bindRight $ storeSentMsg db cfg) reqs
@@ -1219,7 +1213,7 @@ enqueueMessageB c reqs = do
     pure $ Right ((msgId, pqSecr), if null sqs' then Nothing else Just (cData, sqs', msgId))
   where
     storeSentMsg :: DB.Connection -> AgentConfig -> (ConnData, NonEmpty SndQueue, Maybe PQEncryption, MsgFlags, AMessage) -> IO (Either AgentErrorType ((ConnData, NonEmpty SndQueue, Maybe PQEncryption, MsgFlags, AMessage), InternalId, PQEncryption))
-    storeSentMsg db cfg req@(cData@ConnData {connId, pqSupport}, sq :| _, pqEnc_, msgFlags, aMessage) = fmap (first storeError) $ runExceptT $ do
+    storeSentMsg db cfg req@(cData@ConnData {connId}, sq :| _, pqEnc_, msgFlags, aMessage) = fmap (first storeError) $ runExceptT $ do
       let AgentConfig {smpAgentVRange, e2eEncryptVRange} = cfg
       internalTs <- liftIO getCurrentTime
       (internalId, internalSndId, prevMsgHash) <- liftIO $ updateSndIds db connId
@@ -1227,10 +1221,9 @@ enqueueMessageB c reqs = do
           agentMsg = AgentMessage privHeader aMessage
           agentMsgStr = smpEncode agentMsg
           internalHash = C.sha256Hash agentMsgStr
-          currentE2EVersion = maxVersion $ e2eEncryptVRange PQSupportOff
-      (encAgentMessage, pqEnc) <- agentRatchetEncrypt db cData agentMsgStr e2eEncUserMsgLength pqEnc_ currentE2EVersion
-      -- agent version range is determined by the connection suppport of PQ encryption, that is may be enabled when message is sent
-      let agentVersion = maxVersion $ smpAgentVRange pqSupport
+          currentE2EVersion = maxVersion e2eEncryptVRange
+      (encAgentMessage, pqEnc) <- agentRatchetEncrypt db cData agentMsgStr e2eEncAgentMsgLength pqEnc_ currentE2EVersion
+      let agentVersion = maxVersion smpAgentVRange
           msgBody = smpEncode $ AgentMsgEnvelope {agentVersion, encAgentMessage}
           msgType = agentMessageType agentMsg
           msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgFlags, msgBody, pqEncryption = pqEnc, internalHash, prevMsgHash}
@@ -1242,7 +1235,7 @@ enqueueSavedMessage :: AgentClient -> ConnData -> AgentMsgId -> SndQueue -> AM'
 enqueueSavedMessage c cData msgId sq = enqueueSavedMessageB c $ Identity (cData, [sq], msgId)
 {-# INLINE enqueueSavedMessage #-}
 
-enqueueSavedMessageB :: (Foldable t) => AgentClient -> t (ConnData, [SndQueue], AgentMsgId) -> AM' ()
+enqueueSavedMessageB :: Foldable t => AgentClient -> t (ConnData, [SndQueue], AgentMsgId) -> AM' ()
 enqueueSavedMessageB c reqs = do
   -- saving to the database is in the start to avoid race conditions when delivery is read from queue before it is saved
   void $ withStoreBatch' c $ \db -> concatMap (storeDeliveries db) reqs
@@ -1477,10 +1470,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) <- newRcvQueue c userId connId srv' clientVRange SMSubscribe
+  (q, qUri, tSess, sessId) <- newRcvQueue c userId connId srv' clientVRange SMSubscribe
   let rq' = (q :: NewRcvQueue) {primary = True, dbReplaceQueueId = Just dbQueueId}
   rq'' <- withStore c $ \db -> addConnRcvQueue db connId rq'
-  liftIO $ addSubscription c rq''
+  lift $ addNewQueueSubscription c rq'' tSess sessId
   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'']
@@ -1520,7 +1513,7 @@ synchronizeRatchet' c connId pqSupport' force = withConnLock c connId "synchroni
           let cData' = cData {pqSupport = pqSupport'} :: ConnData
           AgentConfig {e2eEncryptVRange} <- asks config
           g <- asks random
-          (pk1, pk2, pKem, e2eParams) <- liftIO $ CR.generateRcvE2EParams g (maxVersion $ e2eEncryptVRange pqSupport') pqSupport'
+          (pk1, pk2, pKem, e2eParams) <- liftIO $ CR.generateRcvE2EParams g (maxVersion e2eEncryptVRange) pqSupport'
           enqueueRatchetKeyMsgs c cData' sqs e2eParams
           withStore' c $ \db -> do
             setConnRatchetSync db connId RSStarted
@@ -1964,11 +1957,13 @@ getSMPServer c userId = withUserServers c userId pickServer
 {-# INLINE getSMPServer #-}
 
 subscriber :: AgentClient -> AM' ()
-subscriber c@AgentClient {msgQ} = forever $ do
+subscriber c@AgentClient {subQ, msgQ} = forever $ do
   t <- atomically $ readTBQueue msgQ
   agentOperationBracket c AORcvNetwork waitUntilActive $
     runExceptT (processSMPTransmission c t) >>= \case
-      Left e -> liftIO $ print e
+      Left e -> do
+        logError $ tshow e
+        atomically $ writeTBQueue subQ ("", "", APC SAEConn $ ERR e)
       Right _ -> return ()
 
 cleanupManager :: AgentClient -> AM' ()
@@ -2046,8 +2041,10 @@ data ACKd = ACKd | ACKPending
 -- | make sure to ACK or throw in each message processing branch
 -- it cannot be finally, unfortunately, as sometimes it needs to be ACK+DEL
 processSMPTransmission :: AgentClient -> ServerTransmission SMPVersion BrokerMsg -> AM ()
-processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v, sessId, rId, cmd) = do
-  (rq, SomeConn _ conn) <- withStore c (\db -> getRcvConn db srv rId)
+processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v, sessId, isResponse, rId, cmd) = do
+  active <- atomically $ activeClientSession c tSess sessId
+  unless active $ logNote $ "Inactive client for " <> tshow (strEncode srv)
+  (rq, SomeConn _ conn) <- withStore c (\db -> getRcvConn db srv rId) `catchAgentError` \e -> logError ("getRcvConn: " <> tshow e) >> throwError e
   processSMP rq conn $ toConnData conn
   where
     processSMP :: forall c. RcvQueue -> Connection c -> ConnData -> AM ()
@@ -2058,12 +2055,13 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v,
         withConnLock c connId "processSMP" $ case cmd of
           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
-              whenM (atomically $ hasGetLock c rq) $
-                notify (MSGNTF $ SMP.rcvMessageMeta srvMsgId msg')
+              when isGET $ notify (MSGNTF $ SMP.rcvMessageMeta srvMsgId msg')
               pure ack'
             where
               queueDrained = case conn of
@@ -2189,8 +2187,8 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v,
                       _ -> prohibited >> ack
                   _ -> prohibited >> ack
               updateConnVersion :: Connection c -> ConnData -> VersionSMPA -> AM (Connection c)
-              updateConnVersion conn' cData'@ConnData {pqSupport} msgAgentVersion = do
-                aVRange <- asks $ ($ pqSupport) . smpAgentVRange . config
+              updateConnVersion conn' cData' msgAgentVersion = do
+                aVRange <- asks $ smpAgentVRange . config
                 let msgAVRange = fromMaybe (versionToRange msgAgentVersion) $ safeVersionRange (minVersion aVRange) msgAgentVersion
                 case msgAVRange `compatibleVersion` aVRange of
                   Just (Compatible av)
@@ -2219,9 +2217,10 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v,
                   | otherwise -> ignored
                 _ -> ignored
               ignored = pure "END from disconnected client - ignored"
+          SMP.SUBOK -> checkSubscription
           _ -> do
             logServer "<--" c srv rId $ "unexpected: " <> bshow cmd
-            notify . ERR $ BROKER (B.unpack $ strEncode srv) UNEXPECTED
+            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'
@@ -2235,6 +2234,16 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v,
           enqueueCmd :: InternalCommand -> AM ()
           enqueueCmd = enqueueCommand c "" connId (Just srv) . AInternalCommand
 
+          checkSubscription :: AM ()
+          checkSubscription = do
+            added <-
+              atomically $
+                ifM
+                  ((&&) <$> hasPendingSubscription c connId <*> activeClientSession c tSess sessId)
+                  (True <$ addSubscription c rq)
+                  (pure False)
+            when (added && isResponse) $ 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
@@ -2256,19 +2265,17 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v,
             logServer "<--" c srv rId $ "MSG <CONF>:" <> logSecret srvMsgId
             AgentConfig {smpClientVRange, smpAgentVRange, e2eEncryptVRange} <- asks config
             let ConnData {pqSupport} = toConnData conn'
-                aVRange = smpAgentVRange pqSupport
-                e2eVRange = e2eEncryptVRange pqSupport
             unless
-              (agentVersion `isCompatible` aVRange && smpClientVersion `isCompatible` smpClientVRange)
+              (agentVersion `isCompatible` smpAgentVRange && smpClientVersion `isCompatible` smpClientVRange)
               (throwError $ AGENT A_VERSION)
             case status of
               New -> case (conn', e2eEncryption) of
                 -- party initiating connection
                 (RcvConnection _ _, Just (CR.AE2ERatchetParams _ e2eSndParams@(CR.E2ERatchetParams e2eVersion _ _ _))) -> do
-                  unless (e2eVersion `isCompatible` e2eVRange) (throwError $ AGENT A_VERSION)
+                  unless (e2eVersion `isCompatible` e2eEncryptVRange) (throwError $ AGENT A_VERSION)
                   (pk1, rcDHRs, pKem) <- withStore c (`getRatchetX3dhKeys` connId)
                   rcParams <- liftError cryptoError $ CR.pqX3dhRcv pk1 rcDHRs pKem e2eSndParams
-                  let rcVs = CR.RatchetVersions {current = e2eVersion, maxSupported = maxVersion e2eVRange}
+                  let rcVs = CR.RatchetVersions {current = e2eVersion, maxSupported = maxVersion e2eEncryptVRange}
                       pqSupport' = pqSupport `CR.pqSupportAnd` versionPQSupport_ agentVersion (Just e2eVersion)
                       rc = CR.initRcvRatchet rcVs rcDHRs rcParams pqSupport'
                   g <- asks random
@@ -2450,7 +2457,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v,
               ContactConnection {} -> do
                 -- show connection request even if invitaion via contact address is not compatible.
                 -- in case invitation not compatible, assume there is no PQ encryption support.
-                pqSupport <- lift $ maybe PQSupportOff pqSupported <$> compatibleInvitationUri connReq PQSupportOn
+                pqSupport <- lift $ maybe PQSupportOff pqSupported <$> compatibleInvitationUri connReq
                 g <- asks random
                 let newInv = NewInvitation {contactConnId = connId, connReq, recipientConnInfo = cInfo}
                 invId <- withStore c $ \db -> createInvitation db g newInv
@@ -2470,10 +2477,9 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v,
           newRatchetKey e2eOtherPartyParams@(CR.E2ERatchetParams e2eVersion k1Rcv k2Rcv _) conn'@(DuplexConnection cData'@ConnData {lastExternalSndId, pqSupport} _ sqs) =
             unlessM ratchetExists $ do
               AgentConfig {e2eEncryptVRange} <- asks config
-              let connE2EVRange = e2eEncryptVRange pqSupport
-              unless (e2eVersion `isCompatible` connE2EVRange) (throwError $ AGENT A_VERSION)
+              unless (e2eVersion `isCompatible` e2eEncryptVRange) (throwError $ AGENT A_VERSION)
               keys <- getSendRatchetKeys
-              let rcVs = CR.RatchetVersions {current = e2eVersion, maxSupported = maxVersion connE2EVRange}
+              let rcVs = CR.RatchetVersions {current = e2eVersion, maxSupported = maxVersion e2eEncryptVRange}
               initRatchet rcVs keys
               notifyAgreed
             where
@@ -2579,8 +2585,7 @@ confirmQueue (Compatible agentVersion) c cData@ConnData {connId, pqSupport} sq s
   where
     mkConfirmation :: AgentMessage -> AM MsgBody
     mkConfirmation aMessage = do
-      -- the version to be used when PQSupport is disabled
-      currentE2EVersion <- asks $ maxVersion . ($ PQSupportOff) . e2eEncryptVRange . config
+      currentE2EVersion <- asks $ maxVersion . e2eEncryptVRange . config
       withStore c $ \db -> runExceptT $ do
         void . liftIO $ updateSndIds db connId
         let pqEnc = CR.pqSupportToEnc pqSupport
@@ -2599,8 +2604,7 @@ enqueueConfirmation c cData sq connInfo e2eEncryption_ = do
 
 storeConfirmation :: AgentClient -> ConnData -> SndQueue -> Maybe (CR.SndE2ERatchetParams 'C.X448) -> AgentMessage -> AM ()
 storeConfirmation c cData@ConnData {connId, pqSupport, connAgentVersion = v} sq e2eEncryption_ agentMsg = do
-  -- the version to be used when PQSupport is disabled
-  currentE2EVersion <- asks $ maxVersion . ($ PQSupportOff) . e2eEncryptVRange . config
+  currentE2EVersion <- asks $ maxVersion . e2eEncryptVRange . config
   withStore c $ \db -> runExceptT $ do
     internalTs <- liftIO getCurrentTime
     (internalId, internalSndId, prevMsgHash) <- liftIO $ updateSndIds db connId
@@ -2620,8 +2624,8 @@ enqueueRatchetKeyMsgs c cData (sq :| sqs) e2eEncryption = do
   mapM_ (lift . enqueueSavedMessage c cData msgId) $ filter isActiveSndQ sqs
 
 enqueueRatchetKey :: AgentClient -> ConnData -> SndQueue -> CR.RcvE2ERatchetParams 'C.X448 -> AM AgentMsgId
-enqueueRatchetKey c cData@ConnData {connId, pqSupport} sq e2eEncryption = do
-  aVRange <- asks $ ($ pqSupport) . smpAgentVRange . config
+enqueueRatchetKey c cData@ConnData {connId} sq e2eEncryption = do
+  aVRange <- asks $ smpAgentVRange . config
   msgId <- storeRatchetKey $ maxVersion aVRange
   lift $ submitPendingMsg c cData sq
   pure $ unId msgId

src/Simplex/Messaging/Agent/Client.hs

@@ -41,6 +41,7 @@ module Simplex.Messaging.Agent.Client
     getQueueMessage,
     decryptSMPMessage,
     addSubscription,
+    addNewQueueSubscription,
     getSubscriptions,
     sendConfirmation,
     sendInvitation,
@@ -77,7 +78,9 @@ module Simplex.Messaging.Agent.Client
     logSecret,
     removeSubscription,
     hasActiveSubscription,
+    hasPendingSubscription,
     hasGetLock,
+    activeClientSession,
     agentClientStore,
     agentDRG,
     getAgentSubscriptions,
@@ -152,7 +155,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 (lefts, partitionEithers)
+import Data.Either (partitionEithers)
 import Data.Functor (($>))
 import Data.Int (Int64)
 import Data.List (deleteFirstsBy, foldl', partition, (\\))
@@ -168,6 +171,7 @@ import Data.Text.Encoding
 import Data.Time (UTCTime, defaultTimeLocale, diffUTCTime, formatTime, getCurrentTime)
 import Data.Time.Clock.System (getSystemTime)
 import Data.Word (Word16)
+import GHC.Stack (HasCallStack, withFrozenCallStack)
 import Network.Socket (HostName)
 import Simplex.FileTransfer.Client (XFTPChunkSpec (..), XFTPClient, XFTPClientConfig (..), XFTPClientError)
 import qualified Simplex.FileTransfer.Client as X
@@ -229,7 +233,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)
+import Simplex.Messaging.Transport (SMPVersion, SessionId, THandleParams (sessionId))
 import Simplex.Messaging.Transport.Client (TransportHost)
 import Simplex.Messaging.Util
 import Simplex.Messaging.Version
@@ -640,7 +644,7 @@ reconnectSMPClient tc c tSess@(_, srv, _) qs = do
     resubscribe :: AM ()
     resubscribe = do
       cs <- readTVarIO $ RQ.getConnections $ activeSubs c
-      rs <- lift . subscribeQueues c $ L.toList qs
+      (rs, sessId_) <- lift . subscribeQueues c $ L.toList qs
       let (errs, okConns) = partitionEithers $ map (\(RcvQueue {connId}, r) -> bimap (connId,) (const connId) r) rs
       liftIO $ do
         let conns = filter (`M.notMember` cs) okConns
@@ -649,7 +653,10 @@ reconnectSMPClient tc c tSess@(_, srv, _) qs = do
       liftIO $ mapM_ (\(connId, e) -> notifySub connId $ ERR e) finalErrs
       forM_ (listToMaybe tempErrs) $ \(_, err) -> do
         when (null okConns && M.null cs && null finalErrs) . liftIO $
-          closeClient c smpClients tSess
+          forM_ sessId_ $ \sessId -> do
+            -- We only close the client session that was used to subscribe.
+            v_ <- atomically $ ifM (activeClientSession c tSess sessId) (TM.lookupDelete tSess $ smpClients c) (pure Nothing)
+            mapM_ (closeClient_ c) v_
         throwError err
     notifySub :: forall e. AEntityI e => ConnId -> ACommand 'Agent e -> IO ()
     notifySub connId cmd = atomically $ writeTBQueue (subQ c) ("", connId, APC (sAEntity @e) cmd)
@@ -702,8 +709,8 @@ getXFTPServerClient c@AgentClient {active, xftpClients, workerSeq} tSess@(userId
 
 waitForProtocolClient :: ProtocolTypeI (ProtoType msg) => AgentClient -> TransportSession msg -> ClientVar msg -> AM (Client msg)
 waitForProtocolClient c (_, srv, _) v = do
-  NetworkConfig {tcpConnectTimeout, tcpTimeout} <- atomically $ getNetworkConfig c
-  client_ <- liftIO $ (tcpConnectTimeout + tcpTimeout) `timeout` atomically (readTMVar $ sessionVar v)
+  NetworkConfig {tcpConnectTimeout} <- atomically $ getNetworkConfig c
+  client_ <- liftIO $ tcpConnectTimeout `timeout` atomically (readTMVar $ sessionVar v)
   liftEither $ case client_ of
     Just (Right smpClient) -> Right smpClient
     Just (Left e) -> Left e
@@ -883,42 +890,42 @@ getMapLock locks key = TM.lookup key locks >>= maybe newLock pure
   where
     newLock = createLock >>= \l -> TM.insert key l locks $> l
 
-withClient_ :: forall a v err msg. ProtocolServerClient v err msg => AgentClient -> TransportSession msg -> ByteString -> (Client msg -> AM a) -> AM a
+withClient_ :: forall a v err msg. (HasCallStack, ProtocolServerClient v err msg) => AgentClient -> TransportSession msg -> ByteString -> (Client msg -> AM a) -> AM a
 withClient_ c tSess@(userId, srv, _) statCmd action = do
   cl <- getProtocolServerClient c tSess
-  (action cl <* stat cl "OK") `catchAgentError` logServerError cl
+  (action cl <* stat cl "OK") `catchAgentError` \e -> withFrozenCallStack $ logServerError cl e
   where
     stat cl = liftIO . incClientStat c userId cl statCmd
-    logServerError :: Client msg -> AgentErrorType -> AM a
+    logServerError :: HasCallStack => Client msg -> AgentErrorType -> AM a
     logServerError cl e = do
-      logServer "<--" c srv "" $ strEncode e
+      withFrozenCallStack $ logServer "<--" c srv "" $ strEncode e
       stat cl $ strEncode e
       throwError e
 
-withLogClient_ :: ProtocolServerClient v err msg => AgentClient -> TransportSession msg -> EntityId -> ByteString -> (Client msg -> AM a) -> AM a
+withLogClient_ :: (HasCallStack, ProtocolServerClient v err msg) => AgentClient -> TransportSession msg -> EntityId -> ByteString -> (Client msg -> AM a) -> AM a
 withLogClient_ c tSess@(_, srv, _) entId cmdStr action = do
-  logServer "-->" c srv entId cmdStr
+  withFrozenCallStack $ logServer "-->" c srv entId cmdStr
   res <- withClient_ c tSess cmdStr action
-  logServer "<--" c srv entId "OK"
+  withFrozenCallStack $ logServer "<--" c srv entId "OK"
   return res
 
-withClient :: forall v err msg a. ProtocolServerClient v err msg => AgentClient -> TransportSession msg -> ByteString -> (Client msg -> ExceptT (ProtocolClientError err) IO a) -> AM a
-withClient c tSess statKey action = withClient_ c tSess statKey $ \client -> liftClient (clientProtocolError @v @err @msg) (clientServer client) $ action client
+withClient :: forall v err msg a. (HasCallStack, ProtocolServerClient v err msg) => AgentClient -> TransportSession msg -> ByteString -> (Client msg -> ExceptT (ProtocolClientError err) IO a) -> AM a
+withClient c tSess statKey action = withFrozenCallStack $ withClient_ c tSess statKey $ \client -> liftClient (clientProtocolError @v @err @msg) (clientServer client) $ action client
 {-# INLINE withClient #-}
 
-withLogClient :: forall v err msg a. ProtocolServerClient v err msg => AgentClient -> TransportSession msg -> EntityId -> ByteString -> (Client msg -> ExceptT (ProtocolClientError err) IO a) -> AM a
-withLogClient c tSess entId cmdStr action = withLogClient_ c tSess entId cmdStr $ \client -> liftClient (clientProtocolError @v @err @msg) (clientServer client) $ action client
+withLogClient :: forall v err msg a. (HasCallStack, ProtocolServerClient v err msg) => AgentClient -> TransportSession msg -> EntityId -> ByteString -> (Client msg -> ExceptT (ProtocolClientError err) IO a) -> AM a
+withLogClient c tSess entId cmdStr action = withFrozenCallStack $ withLogClient_ c tSess entId cmdStr $ \client -> liftClient (clientProtocolError @v @err @msg) (clientServer client) $ action client
 {-# INLINE withLogClient #-}
 
-withSMPClient :: SMPQueueRec q => AgentClient -> q -> ByteString -> (SMPClient -> ExceptT SMPClientError IO a) -> AM a
+withSMPClient :: (HasCallStack, SMPQueueRec q) => AgentClient -> q -> ByteString -> (SMPClient -> ExceptT SMPClientError IO a) -> AM a
 withSMPClient c q cmdStr action = do
   tSess <- liftIO $ mkSMPTransportSession c q
-  withLogClient c tSess (queueId q) cmdStr action
+  withFrozenCallStack $ withLogClient c tSess (queueId q) cmdStr action
 
-withSMPClient_ :: SMPQueueRec q => AgentClient -> q -> ByteString -> (SMPClient -> AM a) -> AM a
+withSMPClient_ :: (HasCallStack, SMPQueueRec q) => AgentClient -> q -> ByteString -> (SMPClient -> AM a) -> AM a
 withSMPClient_ c q cmdStr action = do
   tSess <- liftIO $ mkSMPTransportSession c q
-  withLogClient_ c tSess (queueId q) cmdStr action
+  withFrozenCallStack $ withLogClient_ c tSess (queueId q) cmdStr action
 
 withNtfClient :: AgentClient -> NtfServer -> EntityId -> ByteString -> (NtfClient -> ExceptT NtfClientError IO a) -> AM a
 withNtfClient c srv = withLogClient c (0, srv, Nothing)
@@ -1094,7 +1101,7 @@ getSessionMode :: AgentClient -> IO TransportSessionMode
 getSessionMode = atomically . fmap sessionMode . getNetworkConfig
 {-# INLINE getSessionMode #-}
 
-newRcvQueue :: AgentClient -> UserId -> ConnId -> SMPServerWithAuth -> VersionRangeSMPC -> SubscriptionMode -> AM (NewRcvQueue, SMPQueueUri)
+newRcvQueue :: AgentClient -> UserId -> ConnId -> SMPServerWithAuth -> VersionRangeSMPC -> SubscriptionMode -> AM (NewRcvQueue, SMPQueueUri, SMPTransportSession, SessionId)
 newRcvQueue c userId connId (ProtoServerWithAuth srv auth) vRange subMode = do
   C.AuthAlg a <- asks (rcvAuthAlg . config)
   g <- asks random
@@ -1103,8 +1110,9 @@ 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
-  QIK {rcvId, sndId, rcvPublicDhKey} <-
-    withClient c tSess "NEW" $ \smp -> createSMPQueue smp rKeys dhKey auth subMode
+  (sessId, QIK {rcvId, sndId, rcvPublicDhKey}) <-
+    withClient c tSess "NEW" $ \smp ->
+      (sessionId $ thParams smp,) <$> createSMPQueue smp rKeys dhKey auth subMode
   liftIO . logServer "<--" c srv "" $ B.unwords ["IDS", logSecret rcvId, logSecret sndId]
   let rq =
         RcvQueue
@@ -1126,17 +1134,16 @@ newRcvQueue c userId connId (ProtoServerWithAuth srv auth) vRange subMode = do
             clientNtfCreds = Nothing,
             deleteErrors = 0
           }
-  pure (rq, SMPQueueUri vRange $ SMPQueueAddress srv sndId e2eDhKey)
+      qUri = SMPQueueUri vRange $ SMPQueueAddress srv sndId e2eDhKey
+  pure (rq, qUri, tSess, sessId)
 
-processSubResult :: AgentClient -> RcvQueue -> Either SMPClientError () -> IO (Either SMPClientError ())
-processSubResult c rq r = do
-  case r of
-    Left e ->
-      unless (temporaryClientError e) . atomically $ do
-        RQ.deleteQueue rq (pendingSubs c)
-        TM.insert (RQ.qKey rq) e (removedSubs c)
-    _ -> addSubscription c rq
-  pure r
+processSubResult :: AgentClient -> RcvQueue -> Either SMPClientError () -> STM ()
+processSubResult c rq = \case
+  Left e ->
+    unless (temporaryClientError e) $ do
+      RQ.deleteQueue rq (pendingSubs c)
+      TM.insert (RQ.qKey rq) e (removedSubs c)
+  Right () -> addSubscription c rq
 
 temporaryAgentError :: AgentErrorType -> Bool
 temporaryAgentError = \case
@@ -1153,7 +1160,7 @@ temporaryOrHostError = \case
 {-# INLINE temporaryOrHostError #-}
 
 -- | Subscribe to queues. The list of results can have a different order.
-subscribeQueues :: AgentClient -> [RcvQueue] -> AM' [(RcvQueue, Either AgentErrorType ())]
+subscribeQueues :: AgentClient -> [RcvQueue] -> AM' ([(RcvQueue, Either AgentErrorType ())], Maybe SessionId)
 subscribeQueues c qs = do
   (errs, qs') <- partitionEithers <$> mapM checkQueue qs
   atomically $ do
@@ -1161,20 +1168,43 @@ subscribeQueues c qs = do
     RQ.batchAddQueues (pendingSubs c) qs'
   env <- ask
   -- only "checked" queues are subscribed
-  (errs <>) <$> sendTSessionBatches "SUB" 90 id (subscribeQueues_ env) c qs'
+  session <- newTVarIO Nothing
+  rs <- sendTSessionBatches "SUB" 90 id (subscribeQueues_ env session) c qs'
+  (errs <> rs,) <$> readTVarIO session
   where
     checkQueue rq = do
       prohibited <- atomically $ hasGetLock c rq
       pure $ if prohibited then Left (rq, Left $ CMD PROHIBITED) else Right rq
-    subscribeQueues_ :: Env -> SMPClient -> NonEmpty RcvQueue -> IO (BatchResponses SMPClientError ())
-    subscribeQueues_ env smp qs' = do
+    subscribeQueues_ :: Env -> TVar (Maybe SessionId) -> SMPClient -> NonEmpty RcvQueue -> IO (BatchResponses SMPClientError ())
+    subscribeQueues_ env session smp qs' = do
+      atomically . modifyTVar (sentSubs smp) . M.union $ M.fromList [(connId, False) | RcvQueue {connId} <- L.toList qs']
       rs <- sendBatch subscribeSMPQueues smp qs'
-      mapM_ (uncurry $ processSubResult c) rs
-      when (any temporaryClientError . lefts . map snd $ L.toList rs) $
-        runReaderT (resubscribeSMPSession c $ transportSession' smp) env
-      pure rs
+      atomically . modifyTVar (sentSubs smp) . M.union $ M.fromList [(connId, True) | (RcvQueue {connId}, Right ()) <- L.toList rs]
+      active <-
+        atomically $
+          ifM
+            (activeClientSession c tSess sessId)
+            (writeTVar session (Just sessId) >> mapM_ (uncurry $ processSubResult c) rs $> True)
+            (pure False)
+      if active
+        then when (hasTempErrors rs) resubscribe $> rs
+        else do
+          unless active $ 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
 
-type BatchResponses e r = (NonEmpty (RcvQueue, Either e r))
+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)
 
 -- 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)]
@@ -1209,16 +1239,30 @@ sendBatch smpCmdFunc smp qs = L.zip qs <$> smpCmdFunc smp (L.map queueCreds qs)
   where
     queueCreds RcvQueue {rcvPrivateKey, rcvId} = (rcvPrivateKey, rcvId)
 
-addSubscription :: AgentClient -> RcvQueue -> IO ()
-addSubscription c rq@RcvQueue {connId} = atomically $ do
+addSubscription :: AgentClient -> RcvQueue -> STM ()
+addSubscription c rq@RcvQueue {connId} = do
   modifyTVar' (subscrConns c) $ S.insert connId
   RQ.addQueue rq $ activeSubs c
   RQ.deleteQueue 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 <$ RQ.addQueue rq (pendingSubs c))
+  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
@@ -1229,9 +1273,9 @@ getSubscriptions :: AgentClient -> STM (Set ConnId)
 getSubscriptions = readTVar . subscrConns
 {-# INLINE getSubscriptions #-}
 
-logServer :: MonadIO m => ByteString -> AgentClient -> ProtocolServer s -> QueueId -> ByteString -> m ()
+logServer :: (HasCallStack, MonadIO m) => ByteString -> AgentClient -> ProtocolServer s -> QueueId -> ByteString -> m ()
 logServer dir AgentClient {clientId} srv qId cmdStr =
-  logInfo . decodeUtf8 $ B.unwords ["A", "(" <> bshow clientId <> ")", dir, showServer srv, ":", logSecret qId, cmdStr]
+  withFrozenCallStack $ logInfo . decodeUtf8 $ B.unwords ["A", "(" <> bshow clientId <> ")", dir, showServer srv, ":", logSecret qId, cmdStr]
 {-# INLINE logServer #-}
 
 showServer :: ProtocolServer s -> ByteString

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

@@ -56,7 +56,7 @@ import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
 import Simplex.Messaging.Client
 import Simplex.Messaging.Client.Agent ()
 import qualified Simplex.Messaging.Crypto as C
-import Simplex.Messaging.Crypto.Ratchet (PQSupport, VersionRangeE2E, supportedE2EEncryptVRange)
+import Simplex.Messaging.Crypto.Ratchet (VersionRangeE2E, supportedE2EEncryptVRange)
 import Simplex.Messaging.Notifications.Client (defaultNTFClientConfig)
 import Simplex.Messaging.Notifications.Transport (NTFVersion)
 import Simplex.Messaging.Notifications.Types
@@ -117,8 +117,8 @@ data AgentConfig = AgentConfig
     caCertificateFile :: FilePath,
     privateKeyFile :: FilePath,
     certificateFile :: FilePath,
-    e2eEncryptVRange :: PQSupport -> VersionRangeE2E,
-    smpAgentVRange :: PQSupport -> VersionRangeSMPA,
+    e2eEncryptVRange :: VersionRangeE2E,
+    smpAgentVRange :: VersionRangeSMPA,
     smpClientVRange :: VersionRangeSMPC
   }
 

src/Simplex/Messaging/Agent/Protocol.hs

@@ -44,7 +44,7 @@ module Simplex.Messaging.Agent.Protocol
     currentSMPAgentVersion,
     supportedSMPAgentVRange,
     e2eEncConnInfoLength,
-    e2eEncUserMsgLength,
+    e2eEncAgentMsgLength,
 
     -- * SMP agent protocol types
     ConnInfo,
@@ -272,16 +272,11 @@ deliveryRcptsSMPAgentVersion = VersionSMPA 4
 pqdrSMPAgentVersion :: VersionSMPA
 pqdrSMPAgentVersion = VersionSMPA 5
 
--- TODO v5.7 increase to 5
 currentSMPAgentVersion :: VersionSMPA
-currentSMPAgentVersion = VersionSMPA 4
+currentSMPAgentVersion = VersionSMPA 5
 
--- TODO v5.7 remove dependency of version range on whether PQ support is needed
-supportedSMPAgentVRange :: PQSupport -> VersionRangeSMPA
-supportedSMPAgentVRange pq =
-  mkVersionRange duplexHandshakeSMPAgentVersion $ case pq of
-    PQSupportOn -> pqdrSMPAgentVersion
-    PQSupportOff -> currentSMPAgentVersion
+supportedSMPAgentVRange :: VersionRangeSMPA
+supportedSMPAgentVRange = mkVersionRange duplexHandshakeSMPAgentVersion currentSMPAgentVersion
 
 -- it is shorter to allow all handshake headers,
 -- including E2E (double-ratchet) parameters and
@@ -292,8 +287,8 @@ e2eEncConnInfoLength v = \case
   PQSupportOn | v >= pqdrSMPAgentVersion -> 11122
   _ -> 14848
 
-e2eEncUserMsgLength :: VersionSMPA -> PQSupport -> Int
-e2eEncUserMsgLength v = \case
+e2eEncAgentMsgLength :: VersionSMPA -> PQSupport -> Int
+e2eEncAgentMsgLength v = \case
   -- reduced by 2222 (the increase of message ratchet header size)
   PQSupportOn | v >= pqdrSMPAgentVersion -> 13634
   _ -> 15856

src/Simplex/Messaging/Client.hs

@@ -28,7 +28,7 @@
 module Simplex.Messaging.Client
   ( -- * Connect (disconnect) client to (from) SMP server
     TransportSession,
-    ProtocolClient (thParams, sessionTs),
+    ProtocolClient (thParams, sessionTs, sentSubs),
     SMPClient,
     getProtocolClient,
     closeProtocolClient,
@@ -97,7 +97,7 @@ import Data.List (find)
 import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.List.NonEmpty as L
 import Data.Maybe (fromMaybe)
-import Data.Time.Clock (UTCTime (..), getCurrentTime)
+import Data.Time.Clock (UTCTime (..), diffUTCTime, getCurrentTime)
 import Network.Socket (ServiceName)
 import Numeric.Natural
 import qualified Simplex.Messaging.Crypto as C
@@ -110,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, raceAny_, threadDelay', whenM)
+import Simplex.Messaging.Util (bshow, diffToMicroseconds, raceAny_, threadDelay', whenM)
 import Simplex.Messaging.Version
 import System.Timeout (timeout)
 import UnliftIO (pooledMapConcurrentlyN)
@@ -122,6 +122,7 @@ data ProtocolClient v err msg = ProtocolClient
   { action :: Maybe (Async ()),
     thParams :: THandleParams v 'TClient,
     sessionTs :: UTCTime,
+    sentSubs :: TMap ByteString Bool, -- DEBUG: ConnId -> sub status
     client_ :: PClient v err msg
   }
 
@@ -131,20 +132,26 @@ data PClient v err msg = PClient
     transportHost :: TransportHost,
     tcpTimeout :: Int,
     rcvConcurrency :: Int,
-    pingErrorCount :: TVar Int,
+    sendPings :: TVar Bool,
+    lastReceived :: TVar UTCTime,
+    timeoutErrorCount :: TVar Int,
     clientCorrId :: TVar ChaChaDRG,
     sentCommands :: TMap CorrId (Request err msg),
-    sndQ :: TBQueue (TVar Bool, ByteString),
+    sndQ :: TBQueue (Maybe (TVar Bool), ByteString),
     rcvQ :: TBQueue (NonEmpty (SignedTransmission err msg)),
     msgQ :: Maybe (TBQueue (ServerTransmission v msg))
   }
 
 smpClientStub :: TVar ChaChaDRG -> ByteString -> VersionSMP -> Maybe (THandleAuth 'TClient) -> STM SMPClient
 smpClientStub g sessionId thVersion thAuth = do
+  let ts = UTCTime (read "2024-03-31") 0
   connected <- newTVar False
   clientCorrId <- C.newRandomDRG g
   sentCommands <- TM.empty
-  pingErrorCount <- newTVar 0
+  sendPings <- newTVar False
+  sentSubs <- TM.empty
+  lastReceived <- newTVar ts
+  timeoutErrorCount <- newTVar 0
   sndQ <- newTBQueue 100
   rcvQ <- newTBQueue 100
   return
@@ -159,7 +166,8 @@ smpClientStub g sessionId thVersion thAuth = do
               implySessId = thVersion >= authCmdsSMPVersion,
               batch = True
             },
-        sessionTs = UTCTime (read "2024-03-31") 0,
+        sessionTs = ts,
+        sentSubs,
         client_ =
           PClient
             { connected,
@@ -167,7 +175,9 @@ smpClientStub g sessionId thVersion thAuth = do
               transportHost = "localhost",
               tcpTimeout = 15_000_000,
               rcvConcurrency = 8,
-              pingErrorCount,
+              sendPings,
+              lastReceived,
+              timeoutErrorCount,
               clientCorrId,
               sentCommands,
               sndQ,
@@ -182,7 +192,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 msg = (TransportSession msg, Version v, SessionId, EntityId, msg)
+type ServerTransmission v msg = (TransportSession msg, Version v, SessionId, Bool, EntityId, msg)
 
 data HostMode
   = -- | prefer (or require) onion hosts when connecting via SOCKS proxy
@@ -215,7 +225,7 @@ data NetworkConfig = NetworkConfig
     tcpKeepAlive :: Maybe KeepAliveOpts,
     -- | period for SMP ping commands (microseconds, 0 to disable)
     smpPingInterval :: Int64,
-    -- | the count of PING errors after which SMP client terminates (and will be reconnected), 0 to disable
+    -- | the count of timeout errors after which SMP client terminates (and will be reconnected), 0 to disable
     smpPingCount :: Int,
     logTLSErrors :: Bool
   }
@@ -329,15 +339,17 @@ getProtocolClient :: forall v err msg. Protocol v err msg => TVar ChaChaDRG -> T
 getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize, networkConfig, serverVRange, agreeSecret} msgQ disconnected = do
   case chooseTransportHost networkConfig (host srv) of
     Right useHost ->
-      (atomically (mkProtocolClient useHost) >>= runClient useTransport useHost)
+      (getCurrentTime >>= atomically . mkProtocolClient useHost >>= runClient useTransport useHost)
         `catch` \(e :: IOException) -> pure . Left $ PCEIOError e
     Left e -> pure $ Left e
   where
     NetworkConfig {tcpConnectTimeout, tcpTimeout, rcvConcurrency, smpPingInterval} = networkConfig
-    mkProtocolClient :: TransportHost -> STM (PClient v err msg)
-    mkProtocolClient transportHost = do
+    mkProtocolClient :: TransportHost -> UTCTime -> STM (PClient v err msg)
+    mkProtocolClient transportHost ts = do
       connected <- newTVar False
-      pingErrorCount <- newTVar 0
+      sendPings <- newTVar False
+      lastReceived <- newTVar ts
+      timeoutErrorCount <- newTVar 0
       clientCorrId <- C.newRandomDRG g
       sentCommands <- TM.empty
       sndQ <- newTBQueue qSize
@@ -348,7 +360,9 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
             transportSession,
             transportHost,
             tcpTimeout,
-            pingErrorCount,
+            sendPings,
+            lastReceived,
+            timeoutErrorCount,
             clientCorrId,
             sentCommands,
             rcvConcurrency,
@@ -366,7 +380,7 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
         async $
           runTransportClient tcConfig (Just username) useHost port' (Just $ keyHash srv) (client t c cVar)
             `finally` atomically (tryPutTMVar cVar $ Left PCENetworkError)
-      c_ <- (tcpConnectTimeout + tcpTimeout) `timeout` atomically (takeTMVar cVar)
+      c_ <- tcpConnectTimeout `timeout` atomically (takeTMVar cVar)
       case c_ of
         Just (Right c') -> pure $ Right c' {action = Just action}
         Just (Left e) -> pure $ Left e
@@ -385,44 +399,62 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
         Left e -> atomically . putTMVar cVar . Left $ PCETransportError e
         Right th@THandle {params} -> do
           sessionTs <- getCurrentTime
-          let c' = ProtocolClient {action = Nothing, client_ = c, thParams = params, sessionTs}
+          sentSubs <- atomically TM.empty
+          let c' = ProtocolClient {action = Nothing, client_ = c, thParams = params, sessionTs, sentSubs}
+          atomically $ writeTVar (lastReceived c) sessionTs
           atomically $ do
             writeTVar (connected c) True
             putTMVar cVar $ Right c'
-          raceAny_ ([send c' th, process c', receive c' th] <> [ping c' | smpPingInterval > 0])
+          raceAny_ ([send c' th, process c', receive c' th] <> [monitor 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) >>= \(active, s) -> whenM (readTVarIO active) (void $ tPutLog h s)
+    send ProtocolClient {client_ = PClient {sndQ}} h = forever $ atomically (readTBQueue sndQ) >>= sendActive
+      where
+        sendActive (Nothing, s) = send_ s
+        sendActive (Just active, s) = whenM (readTVarIO active) $ send_ s
+        send_ = void . tPutLog h
 
     receive :: Transport c => ProtocolClient v err msg -> THandle v c 'TClient -> IO ()
-    receive ProtocolClient {client_ = PClient {rcvQ}} h = forever $ tGet h >>= atomically . writeTBQueue rcvQ
+    receive ProtocolClient {client_ = PClient {rcvQ, lastReceived, timeoutErrorCount}} h = forever $ do
+      tGet h >>= atomically . writeTBQueue rcvQ
+      getCurrentTime >>= atomically . writeTVar lastReceived
+      atomically $ writeTVar timeoutErrorCount 0
 
-    ping :: ProtocolClient v err msg -> IO ()
-    ping c@ProtocolClient {client_ = PClient {pingErrorCount}} = do
-      threadDelay' smpPingInterval
-      runExceptT (sendProtocolCommand c Nothing "" $ protocolPing @v @err @msg) >>= \case
-        Left PCEResponseTimeout -> do
-          cnt <- atomically $ stateTVar pingErrorCount $ \cnt -> (cnt + 1, cnt + 1)
-          when (maxCnt == 0 || cnt < maxCnt) $ ping c
-        _ -> ping c -- sendProtocolCommand resets pingErrorCount
+    monitor :: ProtocolClient v err msg -> IO ()
+    monitor c@ProtocolClient {client_ = PClient {sendPings, lastReceived, timeoutErrorCount}} = loop smpPingInterval
       where
+        loop :: Int64 -> IO ()
+        loop delay = do
+          threadDelay' delay
+          diff <- diffUTCTime <$> getCurrentTime <*> readTVarIO lastReceived
+          let idle = diffToMicroseconds diff
+              remaining = smpPingInterval - idle
+          if remaining > 1_000_000 -- delay pings only for significant time
+            then loop remaining
+            else do
+              whenM (readTVarIO sendPings) $ void . runExceptT $ sendProtocolCommand c Nothing "" (protocolPing @v @err @msg)
+              -- sendProtocolCommand/getResponse updates counter for each command
+              cnt <- readTVarIO timeoutErrorCount
+              -- drop client when maxCnt of commands have timed out in sequence, but only after some time has passed after last received response
+              when (maxCnt == 0 || cnt < maxCnt || diff < recoverWindow) $ loop smpPingInterval
+        recoverWindow = 15 * 60 -- seconds
         maxCnt = smpPingCount networkConfig
 
     process :: ProtocolClient v err msg -> IO ()
     process c = forever $ atomically (readTBQueue $ rcvQ $ client_ c) >>= mapM_ (processMsg c)
 
     processMsg :: ProtocolClient v err msg -> SignedTransmission err msg -> IO ()
-    processMsg c@ProtocolClient {client_ = PClient {sentCommands}} (_, _, (corrId, entId, respOrErr)) =
-      if B.null $ bs corrId
-        then sendMsg respOrErr
-        else do
+    processMsg c@ProtocolClient {client_ = PClient {sentCommands}} (_, _, (corrId, entId, respOrErr))
+      | isResponse =
           atomically (TM.lookup corrId sentCommands) >>= \case
             Nothing -> sendMsg respOrErr
             Just Request {entityId, responseVar} -> atomically $ do
               TM.delete corrId sentCommands
               putTMVar responseVar $ response entityId
+      | otherwise = sendMsg respOrErr
       where
+        isResponse = not $ B.null $ bs corrId
         response entityId
           | entityId == entId =
               case respOrErr of
@@ -433,7 +465,7 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
           | otherwise = Left . PCEUnexpectedResponse $ bshow respOrErr
         sendMsg :: Either err msg -> IO ()
         sendMsg = \case
-          Right msg -> atomically $ mapM_ (`writeTBQueue` serverTransmission c entId msg) msgQ
+          Right msg -> atomically $ mapM_ (`writeTBQueue` serverTransmission c isResponse entId msg) msgQ
           Left e -> putStrLn $ "SMP client error: " <> show e
 
 proxyUsername :: TransportSession msg -> ByteString
@@ -504,15 +536,22 @@ createSMPQueue c (rKey, rpKey) dhKey auth subMode =
 --
 -- https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#subscribe-to-queue
 subscribeSMPQueue :: SMPClient -> RcvPrivateAuthKey -> RecipientId -> ExceptT SMPClientError IO ()
-subscribeSMPQueue c rpKey rId =
-  sendSMPCommand c (Just rpKey) rId SUB >>= \case
-    OK -> return ()
+subscribeSMPQueue c@ProtocolClient {client_ = PClient {sendPings}} rpKey rId = do
+  liftIO . atomically $ writeTVar sendPings True
+  -- We are not expiring sending subscriptions even if it expires to increase chances of subscription
+  -- succeeding without retries - the subscription is registered in the agent when uncorrelated MSG response
+  -- is received via the active client for pending queue subscription - it also prevents unnecessary retries.
+  sendProtocolCommand' False c (Just rpKey) rId (Cmd SRecipient SUB) >>= \case
+    OK -> pure ()
     cmd@MSG {} -> liftIO $ writeSMPMessage c rId cmd
     r -> throwE . PCEUnexpectedResponse $ bshow r
 
 -- | Subscribe to multiple SMP queues batching commands if supported.
 subscribeSMPQueues :: SMPClient -> NonEmpty (RcvPrivateAuthKey, RecipientId) -> IO (NonEmpty (Either SMPClientError ()))
-subscribeSMPQueues c qs = sendProtocolCommands c cs >>= mapM (processSUBResponse c)
+subscribeSMPQueues c@ProtocolClient {client_ = PClient {sendPings}} qs = do
+  atomically $ writeTVar sendPings True
+  -- See comment in subscribeSMPQueue
+  sendProtocolCommands' False c cs >>= mapM (processSUBResponse c)
   where
     cs = L.map (\(rpKey, rId) -> (Just rpKey, rId, Cmd SRecipient SUB)) qs
 
@@ -525,16 +564,17 @@ streamSubscribeSMPQueues c qs cb = streamProtocolCommands c cs $ mapM process >=
 processSUBResponse :: SMPClient -> Response ErrorType BrokerMsg -> IO (Either SMPClientError ())
 processSUBResponse c (Response rId r) = case r of
   Right OK -> pure $ Right ()
+  Right SUBOK -> pure $ Right () -- from v8 of SMP protocol
   Right cmd@MSG {} -> writeSMPMessage c rId cmd $> Right ()
   Right r' -> pure . Left . PCEUnexpectedResponse $ bshow r'
   Left e -> pure $ Left e
 
 writeSMPMessage :: SMPClient -> RecipientId -> BrokerMsg -> IO ()
-writeSMPMessage c rId msg = atomically $ mapM_ (`writeTBQueue` serverTransmission c rId msg) (msgQ $ client_ c)
+writeSMPMessage c rId msg = atomically $ mapM_ (`writeTBQueue` serverTransmission c False rId msg) (msgQ $ client_ c)
 
-serverTransmission :: ProtocolClient v err msg -> RecipientId -> msg -> ServerTransmission v msg
-serverTransmission ProtocolClient {thParams = THandleParams {thVersion, sessionId}, client_ = PClient {transportSession}} entityId message =
-  (transportSession, thVersion, sessionId, entityId, message)
+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
 --
@@ -660,9 +700,13 @@ type PCTransmission err msg = (Either TransportError SentRawTransmission, Reques
 
 -- | Send multiple commands with batching and collect responses
 sendProtocolCommands :: forall v err msg. ProtocolEncoding v err (ProtoCommand msg) => ProtocolClient v err msg -> NonEmpty (ClientCommand msg) -> IO (NonEmpty (Response err msg))
-sendProtocolCommands c@ProtocolClient {thParams = THandleParams {batch, blockSize}} cs = do
+sendProtocolCommands = sendProtocolCommands' True
+{-# INLINE sendProtocolCommands #-}
+
+sendProtocolCommands' :: forall v err msg. ProtocolEncoding v err (ProtoCommand msg) => Bool -> ProtocolClient v err msg -> NonEmpty (ClientCommand msg) -> IO (NonEmpty (Response err msg))
+sendProtocolCommands' expire c@ProtocolClient {thParams = THandleParams {batch, blockSize}} cs = do
   bs <- batchTransmissions' batch blockSize <$> mapM (mkTransmission c) cs
-  validate . concat =<< mapM (sendBatch c) bs
+  validate . concat =<< mapM (sendBatch c expire) bs
   where
     validate :: [Response err msg] -> IO (NonEmpty (Response err msg))
     validate rs
@@ -679,28 +723,36 @@ sendProtocolCommands c@ProtocolClient {thParams = THandleParams {batch, blockSiz
 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
+  mapM_ (cb <=< sendBatch c True) bs
 
-sendBatch :: ProtocolClient v err msg -> TransportBatch (Request err msg) -> IO [Response err msg]
-sendBatch c@ProtocolClient {client_ = PClient {rcvConcurrency, sndQ}} b = do
+sendBatch :: ProtocolClient v err msg -> Bool -> TransportBatch (Request err msg) -> IO [Response err msg]
+sendBatch c@ProtocolClient {client_ = PClient {rcvConcurrency, sndQ}} expire b = do
   case b of
     TBError e Request {entityId} -> do
       putStrLn "send error: large message"
       pure [Response entityId $ Left $ PCETransportError e]
     TBTransmissions s n rs
       | n > 0 -> do
-          active <- newTVarIO True
+          active <- mkActive_ expire
           atomically $ writeTBQueue sndQ (active, s)
           pooledMapConcurrentlyN rcvConcurrency (getResponse c active) rs
       | otherwise -> pure []
     TBTransmission s r -> do
-      active <- newTVarIO True
+      active <- mkActive_ expire
       atomically $ writeTBQueue sndQ (active, s)
       (: []) <$> getResponse c active r
 
+mkActive_ :: Bool -> IO (Maybe (TVar Bool))
+mkActive_ True = Just <$> newTVarIO True
+mkActive_ False = pure Nothing
+
 -- | Send Protocol command
 sendProtocolCommand :: forall v err msg. ProtocolEncoding v err (ProtoCommand msg) => ProtocolClient v err msg -> Maybe C.APrivateAuthKey -> EntityId -> ProtoCommand msg -> ExceptT (ProtocolClientError err) IO msg
-sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ}, thParams = THandleParams {batch, blockSize}} pKey entId cmd =
+sendProtocolCommand = sendProtocolCommand' True
+{-# INLINE sendProtocolCommand #-}
+
+sendProtocolCommand' :: forall v err msg. ProtocolEncoding v err (ProtoCommand msg) => Bool -> ProtocolClient v err msg -> Maybe C.APrivateAuthKey -> EntityId -> ProtoCommand msg -> ExceptT (ProtocolClientError err) IO msg
+sendProtocolCommand' expire c@ProtocolClient {client_ = PClient {sndQ}, thParams = THandleParams {batch, blockSize}} pKey entId cmd =
   ExceptT $ uncurry sendRecv =<< mkTransmission c (pKey, entId, cmd)
   where
     -- two separate "atomically" needed to avoid blocking
@@ -710,7 +762,7 @@ sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ}, thParams = THand
       Right t
         | B.length s > blockSize - 2 -> pure . Left $ PCETransportError TELargeMsg
         | otherwise -> do
-            active <- newTVarIO True
+            active <- mkActive_ expire
             atomically (writeTBQueue sndQ (active, s))
             response <$> getResponse c active r
         where
@@ -719,12 +771,15 @@ sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ}, thParams = THand
             | otherwise = tEncode t
 
 -- 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, pingErrorCount, sentCommands}} active Request {corrId, entityId, responseVar} = do
+getResponse :: ProtocolClient v err msg -> Maybe (TVar Bool) -> Request err msg -> IO (Response err msg)
+getResponse ProtocolClient {client_ = PClient {tcpTimeout, timeoutErrorCount, sentCommands}} active Request {corrId, entityId, responseVar} = do
   response <-
     timeout tcpTimeout (atomically (takeTMVar responseVar)) >>= \case
-      Just r -> atomically (writeTVar pingErrorCount 0) $> r
-      Nothing -> atomically (writeTVar active False >> TM.delete corrId sentCommands) $> Left PCEResponseTimeout
+      Just r -> atomically (writeTVar timeoutErrorCount 0) $> r
+      Nothing -> do
+        atomically (mapM_ (`writeTVar` False) active >> TM.delete corrId sentCommands)
+        atomically $ modifyTVar' timeoutErrorCount (+ 1)
+        pure $ Left PCEResponseTimeout
   pure Response {entityId, response}
 
 mkTransmission :: forall v err msg. ProtocolEncoding v err (ProtoCommand msg) => ProtocolClient v err msg -> ClientCommand msg -> IO (PCTransmission err msg)

src/Simplex/Messaging/Client/Agent.hs

@@ -163,8 +163,8 @@ getSMPServerClient' ca@SMPClientAgent {agentCfg, smpClients, msgQ, randomDrg, wo
 
     waitForSMPClient :: SMPClientVar -> ExceptT SMPClientError IO SMPClient
     waitForSMPClient v = do
-      let ProtocolClientConfig {networkConfig = NetworkConfig {tcpConnectTimeout, tcpTimeout}} = smpCfg agentCfg
-      smpClient_ <- liftIO $ (tcpConnectTimeout + tcpTimeout) `timeout` atomically (readTMVar $ sessionVar v)
+      let ProtocolClientConfig {networkConfig = NetworkConfig {tcpConnectTimeout}} = smpCfg agentCfg
+      smpClient_ <- liftIO $ tcpConnectTimeout `timeout` atomically (readTMVar $ sessionVar v)
       liftEither $ case smpClient_ of
         Just (Right smpClient) -> Right smpClient
         Just (Left e) -> Left e

src/Simplex/Messaging/Compression.hs

@@ -4,20 +4,9 @@
 module Simplex.Messaging.Compression where
 
 import qualified Codec.Compression.Zstd as Z1
-import qualified Codec.Compression.Zstd.FFI as Z
-import Control.Monad (forM)
-import Control.Monad.Except
-import Control.Monad.IO.Class
 import Data.ByteString (ByteString)
 import qualified Data.ByteString as B
-import qualified Data.ByteString.Unsafe as B
-import Data.Either (fromRight)
-import Data.List.NonEmpty (NonEmpty)
-import Foreign
-import Foreign.C.Types
-import GHC.IO (unsafePerformIO)
 import Simplex.Messaging.Encoding
-import UnliftIO.Exception (bracket)
 
 data Compressed
   = -- | Short messages are left intact to skip copying and FFI festivities.
@@ -42,49 +31,15 @@ instance Encoding Compressed where
       '1' -> Compressed <$> smpP
       x -> fail $ "unknown Compressed tag: " <> show x
 
--- | Compress as single chunk using stack-allocated context.
 compress1 :: ByteString -> Compressed
 compress1 bs
   | B.length bs <= maxLengthPassthrough = Passthrough bs
   | otherwise = Compressed . Large $ Z1.compress compressionLevel bs
 
-type CompressCtx = (Ptr Z.CCtx, Ptr CChar, CSize)
-
-withCompressCtx :: CSize -> (CompressCtx -> IO a) -> IO a
-withCompressCtx scratchSize action =
-  bracket Z.createCCtx Z.freeCCtx $ \cctx ->
-    allocaBytes (fromIntegral scratchSize) $ \scratchPtr ->
-      action (cctx, scratchPtr, scratchSize)
-
--- | Compress bytes, falling back to Passthrough in case of some internal error.
-compress :: CompressCtx -> ByteString -> IO Compressed
-compress ctx bs = fromRight (Passthrough bs) <$> compress_ ctx bs
-
-compress_ :: CompressCtx -> ByteString -> IO (Either String Compressed)
-compress_ (cctx, scratchPtr, scratchSize) bs
-  | B.length bs <= maxLengthPassthrough = pure . Right $ Passthrough bs
-  | otherwise =
-      B.unsafeUseAsCStringLen bs $ \(sourcePtr, sourceSize) -> runExceptT $ do
-        -- should not fail, unless input buffer is too short
-        dstSize <- ExceptT $ Z.checkError $ Z.compressCCtx cctx scratchPtr scratchSize sourcePtr (fromIntegral sourceSize) compressionLevel
-        liftIO $ Compressed . Large <$> B.packCStringLen (scratchPtr, fromIntegral dstSize)
-
-type DecompressCtx = (Ptr Z.DCtx, Ptr CChar, CSize)
-
-withDecompressCtx :: Int -> (DecompressCtx -> IO a) -> IO a
-withDecompressCtx maxUnpackedSize action =
-  bracket Z.createDCtx Z.freeDCtx $ \dctx ->
-    allocaBytes maxUnpackedSize $ \scratchPtr ->
-      action (dctx, scratchPtr, fromIntegral maxUnpackedSize)
-
-decompress :: DecompressCtx -> Compressed -> IO (Either String ByteString)
-decompress (dctx, scratchPtr, scratchSize) = \case
-  Passthrough bs -> pure $ Right bs
-  Compressed (Large bs) ->
-    B.unsafeUseAsCStringLen bs $ \(sourcePtr, sourceSize) -> do
-      res <- Z.checkError $ Z.decompressDCtx dctx scratchPtr scratchSize sourcePtr (fromIntegral sourceSize)
-      forM res $ \dstSize -> B.packCStringLen (scratchPtr, fromIntegral dstSize)
-
-decompressBatch :: Int -> NonEmpty Compressed -> NonEmpty (Either String ByteString)
-decompressBatch maxUnpackedSize items = unsafePerformIO $ withDecompressCtx maxUnpackedSize $ forM items . decompress
-{-# NOINLINE decompressBatch #-} -- prevent double-evaluation under unsafePerformIO
+decompress1 :: Compressed -> Either String ByteString
+decompress1 = \case
+  Passthrough bs -> Right bs
+  Compressed (Large bs) -> case Z1.decompress bs of
+    Z1.Error e -> Left e
+    Z1.Skip -> Right mempty
+    Z1.Decompress bs' -> Right bs'

src/Simplex/Messaging/Crypto/Ratchet.hs

@@ -143,16 +143,11 @@ kdfX3DHE2EEncryptVersion = VersionE2E 2
 pqRatchetE2EEncryptVersion :: VersionE2E
 pqRatchetE2EEncryptVersion = VersionE2E 3
 
--- TODO v5.7 increase to 3
 currentE2EEncryptVersion :: VersionE2E
-currentE2EEncryptVersion = VersionE2E 2
+currentE2EEncryptVersion = VersionE2E 3
 
--- TODO v5.7 remove dependency of version range on whether PQ encryption is used
-supportedE2EEncryptVRange :: PQSupport -> VersionRangeE2E
-supportedE2EEncryptVRange pq =
-  mkVersionRange kdfX3DHE2EEncryptVersion $ case pq of 
-    PQSupportOn -> pqRatchetE2EEncryptVersion
-    PQSupportOff -> currentE2EEncryptVersion
+supportedE2EEncryptVRange :: VersionRangeE2E
+supportedE2EEncryptVRange = mkVersionRange kdfX3DHE2EEncryptVersion currentE2EEncryptVersion
 
 data RatchetKEMState
   = RKSProposed -- only KEM encapsulation key

src/Simplex/Messaging/Notifications/Server.hs

@@ -218,7 +218,7 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
 
     receiveSMP :: M ()
     receiveSMP = forever $ do
-      ((_, srv, _), _, _, ntfId, msg) <- atomically $ readTBQueue msgQ
+      ((_, srv, _), _, _, _, ntfId, msg) <- atomically $ readTBQueue msgQ
       let smpQueue = SMPQueueNtf srv ntfId
       case msg of
         SMP.NMSG nmsgNonce encNMsgMeta -> do

src/Simplex/Messaging/Notifications/Transport.hs

@@ -47,7 +47,7 @@ currentClientNTFVersion :: VersionNTF
 currentClientNTFVersion = VersionNTF 1
 
 currentServerNTFVersion :: VersionNTF
-currentServerNTFVersion = VersionNTF 1
+currentServerNTFVersion = VersionNTF 2
 
 supportedClientNTFVRange :: VersionRangeNTF
 supportedClientNTFVRange = mkVersionRange initialNTFVersion currentClientNTFVersion

src/Simplex/Messaging/Protocol.hs

@@ -238,10 +238,10 @@ type MaxRcvMessageLen = MaxMessageLen + 16 -- 16104, the padded size is 16106
 
 -- it is shorter to allow per-queue e2e encryption DH key in the "public" header
 e2eEncConfirmationLength :: Int
-e2eEncConfirmationLength = 15936
+e2eEncConfirmationLength = 15920 -- 15881 .. 15976
 
 e2eEncMessageLength :: Int
-e2eEncMessageLength = 16032
+e2eEncMessageLength = 16016 -- 16004 .. 16021
 
 -- | SMP protocol clients
 data Party = Recipient | Sender | Notifier
@@ -395,6 +395,7 @@ data BrokerMsg where
   NMSG :: C.CbNonce -> EncNMsgMeta -> BrokerMsg
   END :: BrokerMsg
   OK :: BrokerMsg
+  SUBOK :: BrokerMsg
   ERR :: ErrorType -> BrokerMsg
   PONG :: BrokerMsg
   deriving (Eq, Show)
@@ -582,6 +583,7 @@ data BrokerMsgTag
   | NMSG_
   | END_
   | OK_
+  | SUBOK_
   | ERR_
   | PONG_
   deriving (Show)
@@ -641,6 +643,7 @@ instance Encoding BrokerMsgTag where
     NMSG_ -> "NMSG"
     END_ -> "END"
     OK_ -> "OK"
+    SUBOK_ -> "SUBOK"
     ERR_ -> "ERR"
     PONG_ -> "PONG"
   smpP = messageTagP
@@ -653,6 +656,7 @@ instance ProtocolMsgTag BrokerMsgTag where
     "NMSG" -> Just NMSG_
     "END" -> Just END_
     "OK" -> Just OK_
+    "SUBOK" -> Just SUBOK_
     "ERR" -> Just ERR_
     "PONG" -> Just PONG_
     _ -> Nothing
@@ -1198,7 +1202,7 @@ instance ProtocolEncoding SMPVersion ErrorType Cmd where
 
 instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where
   type Tag BrokerMsg = BrokerMsgTag
-  encodeProtocol _v = \case
+  encodeProtocol v = \case
     IDS (QIK rcvId sndId srvDh) -> e (IDS_, ' ', rcvId, sndId, srvDh)
     MSG RcvMessage {msgId, msgBody = EncRcvMsgBody body} ->
       e (MSG_, ' ', msgId, Tail body)
@@ -1206,6 +1210,9 @@ instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where
     NMSG nmsgNonce encNMsgMeta -> e (NMSG_, ' ', nmsgNonce, encNMsgMeta)
     END -> e END_
     OK -> e OK_
+    SUBOK
+      | v >= sendingProxySMPVersion -> e SUBOK_
+      | otherwise -> e OK_
     ERR err -> e (ERR_, ' ', err)
     PONG -> e PONG_
     where
@@ -1223,6 +1230,7 @@ instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where
     NMSG_ -> NMSG <$> _smpP <*> smpP
     END_ -> pure END
     OK_ -> pure OK
+    SUBOK_ -> pure SUBOK
     ERR_ -> ERR <$> _smpP
     PONG_ -> pure PONG
 

src/Simplex/Messaging/Server.hs

@@ -724,7 +724,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
             deliver sub = do
               q <- getStoreMsgQueue "SUB" rId
               msg_ <- atomically $ tryPeekMsg q
-              deliverMessage "SUB" qr rId sub q msg_
+              deliverMessage "SUB" qr rId sub q msg_ SUBOK
 
         getMessage :: QueueRec -> M (Transmission BrokerMsg)
         getMessage qr = time "GET" $ do
@@ -781,7 +781,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
                     _ -> do
                       (deletedMsg_, msg_) <- atomically $ tryDelPeekMsg q msgId
                       mapM_ updateStats deletedMsg_
-                      deliverMessage "ACK" qr queueId sub q msg_
+                      deliverMessage "ACK" qr queueId sub q msg_ OK
                 _ -> pure $ err NO_MSG
           where
             getDelivered :: TVar Sub -> STM (Maybe Sub)
@@ -863,17 +863,18 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
                   encNMsgMeta = C.cbEncrypt rcvNtfDhSecret cbNonce (smpEncode msgMeta) 128
               pure . (cbNonce,) $ fromRight "" encNMsgMeta
 
-        deliverMessage :: T.Text -> QueueRec -> RecipientId -> TVar Sub -> MsgQueue -> Maybe Message -> M (Transmission BrokerMsg)
-        deliverMessage name qr rId sub q msg_ = time (name <> " deliver") $ do
+        deliverMessage :: T.Text -> QueueRec -> RecipientId -> TVar Sub -> MsgQueue -> Maybe Message -> BrokerMsg -> M (Transmission BrokerMsg)
+        deliverMessage name qr rId sub q msg_ respCmd = time (name <> " deliver") $ do
           readTVarIO sub >>= \case
             s@Sub {subThread = NoSub} ->
               case msg_ of
                 Just msg ->
                   let encMsg = encryptMsg qr msg
                    in atomically (setDelivered s msg) $> (corrId, rId, MSG encMsg)
-                _ -> forkSub $> ok
-            _ -> pure ok
+                _ -> forkSub $> resp
+            _ -> pure resp
           where
+            resp = (corrId, rId, respCmd)
             forkSub :: M ()
             forkSub = do
               atomically . modifyTVar' sub $ \s -> s {subThread = SubPending}

src/Simplex/Messaging/Session.hs

@@ -10,6 +10,7 @@ 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,
@@ -36,3 +37,6 @@ 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)

src/Simplex/Messaging/Transport.hs

@@ -41,6 +41,7 @@ module Simplex.Messaging.Transport
     basicAuthSMPVersion,
     subModeSMPVersion,
     authCmdsSMPVersion,
+    sendingProxySMPVersion,
     simplexMQVersion,
     smpBlockSize,
     TransportConfig (..),
@@ -124,7 +125,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 (2/3/2024)
+-- 7 - support authenticated encryption to verify senders' commands, imply but do NOT send session ID in signed part (4/30/2024)
 
 data SMPVersion
 
@@ -149,11 +150,14 @@ subModeSMPVersion = VersionSMP 6
 authCmdsSMPVersion :: VersionSMP
 authCmdsSMPVersion = VersionSMP 7
 
+sendingProxySMPVersion :: VersionSMP
+sendingProxySMPVersion = VersionSMP 8
+
 currentClientSMPRelayVersion :: VersionSMP
 currentClientSMPRelayVersion = VersionSMP 6
 
 currentServerSMPRelayVersion :: VersionSMP
-currentServerSMPRelayVersion = VersionSMP 6
+currentServerSMPRelayVersion = VersionSMP 7
 
 -- minimal supported protocol version is 4
 -- TODO remove code that supports sending commands without batching

src/Simplex/Messaging/Transport/Client.hs

@@ -2,6 +2,7 @@
 {-# LANGUAGE FlexibleInstances #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE NumericUnderscores #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 
@@ -18,7 +19,7 @@ module Simplex.Messaging.Transport.Client
     TransportHost (..),
     TransportHosts (..),
     TransportHosts_ (..),
-    validateCertificateChain
+    validateCertificateChain,
   )
 where
 
@@ -51,7 +52,7 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (parseAll, parseString)
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.KeepAlive
-import Simplex.Messaging.Util (bshow, (<$?>), catchAll, tshow)
+import Simplex.Messaging.Util (bshow, catchAll, tshow, (<$?>))
 import System.IO.Error
 import System.Timeout (timeout)
 import Text.Read (readMaybe)
@@ -124,7 +125,7 @@ data TransportClientConfig = TransportClientConfig
 
 -- time to resolve host, connect socket, set up TLS
 defaultTcpConnectTimeout :: Int
-defaultTcpConnectTimeout = 10000000
+defaultTcpConnectTimeout = 25_000_000
 
 defaultTransportClientConfig :: TransportClientConfig
 defaultTransportClientConfig = TransportClientConfig Nothing defaultTcpConnectTimeout (Just defaultKeepAliveOpts) True Nothing Nothing
@@ -142,14 +143,19 @@ runTLSTransportClient tlsParams caStore_ cfg@TransportClientConfig {socksProxy,
   serverCert <- newEmptyTMVarIO
   let hostName = B.unpack $ strEncode host
       clientParams = mkTLSClientParams tlsParams caStore_ hostName port keyHash clientCredentials alpn serverCert
-      connectTCP = case socksProxy of
-        Just proxy -> connectSocksClient proxy proxyUsername $ hostAddr host
-        _ -> connectTCPClient hostName
+      (connectTCP, tlsTimeout) = case socksProxy of
+        -- We use a much larger timeout for connections via SOCKS proxy, to allow the circuits created
+        -- in the socket connection that would otherwise timeout to be used in the next connection attempt.
+        -- Using standard timeout results in permanent timeout for the clients using SOCKS in cases
+        -- when SOCKS proxy is very slow (bad network, congestion in underlying network, etc.),
+        -- because SOCKS proxy destroys circuits when the last session using them is closed.
+        Just proxy -> (connectSocksClient proxy proxyUsername (hostAddr host), tcpConnectTimeout * 10)
+        _ -> (connectTCPClient hostName, tcpConnectTimeout)
   c <- do
     sock <- connectTCP port
     mapM_ (setSocketKeepAlive sock) tcpKeepAlive `catchAll` \e -> logError ("Error setting TCP keep-alive" <> tshow e)
     let tCfg = clientTransportConfig cfg
-    timeout tcpConnectTimeout (connectTLS (Just hostName) tCfg clientParams sock) >>= \case
+    tlsTimeout `timeout` connectTLS (Just hostName) tCfg clientParams sock >>= \case
       Nothing -> do
         close sock
         logError "connection timed out"

src/Simplex/Messaging/Transport/HTTP2/Client.hs

@@ -70,7 +70,7 @@ defaultHTTP2ClientConfig :: HTTP2ClientConfig
 defaultHTTP2ClientConfig =
   HTTP2ClientConfig
     { qSize = 64,
-      connTimeout = 10000000,
+      connTimeout = defaultTcpConnectTimeout,
       transportConfig =
         TransportClientConfig
           { socksProxy = Nothing,

tests/AgentTests.hs

@@ -214,14 +214,13 @@ testDuplexConnection _ alice bob = testDuplexConnection' (alice, IKPQOn) (bob, P
 testDuplexConnection' :: (HasCallStack, Transport c) => (c, InitialKeys) -> (c, PQSupport) -> IO ()
 testDuplexConnection' (alice, aPQ) (bob, bPQ) = do
   let pq = pqConnectionMode aPQ bPQ
-      pqSup = CR.pqEncToSupport pq
   ("1", "bob", Right (INV cReq)) <- alice #: ("1", "bob", "NEW T INV" <> pqConnModeStr aPQ <> " subscribe")
   let cReq' = strEncode cReq
   bob #: ("11", "alice", "JOIN T " <> cReq' <> enableKEMStr bPQ <> " subscribe 14\nbob's connInfo") #> ("11", "alice", OK)
   ("", "bob", Right (A.CONF confId pqSup' _ "bob's connInfo")) <- (alice <#:)
-  pqSup' `shouldBe` pqSup
+  pqSup' `shouldBe` CR.connPQEncryption aPQ
   alice #: ("2", "bob", "LET " <> confId <> " 16\nalice's connInfo") #> ("2", "bob", OK)
-  bob <# ("", "alice", A.INFO pqSup "alice's connInfo")
+  bob <# ("", "alice", A.INFO bPQ "alice's connInfo")
   bob <# ("", "alice", CON pq)
   alice <# ("", "bob", CON pq)
   -- message IDs 1 to 3 get assigned to control messages, so first MSG is assigned ID 4
@@ -253,15 +252,14 @@ testDuplexConnRandomIds _ alice bob = testDuplexConnRandomIds' (alice, IKPQOn) (
 testDuplexConnRandomIds' :: (HasCallStack, Transport c) => (c, InitialKeys) -> (c, PQSupport) -> IO ()
 testDuplexConnRandomIds' (alice, aPQ) (bob, bPQ) = do
   let pq = pqConnectionMode aPQ bPQ
-      pqSup = CR.pqEncToSupport pq
   ("1", bobConn, Right (INV cReq)) <- alice #: ("1", "", "NEW T INV" <> pqConnModeStr aPQ <> " subscribe")
   let cReq' = strEncode cReq
   ("11", aliceConn, Right OK) <- bob #: ("11", "", "JOIN T " <> cReq' <> enableKEMStr bPQ <> " subscribe 14\nbob's connInfo")
   ("", bobConn', Right (A.CONF confId pqSup' _ "bob's connInfo")) <- (alice <#:)
-  pqSup' `shouldBe` pqSup
+  pqSup' `shouldBe` CR.connPQEncryption aPQ
   bobConn' `shouldBe` bobConn
   alice #: ("2", bobConn, "LET " <> confId <> " 16\nalice's connInfo") =#> \case ("2", c, OK) -> c == bobConn; _ -> False
-  bob <# ("", aliceConn, A.INFO pqSup "alice's connInfo")
+  bob <# ("", aliceConn, A.INFO bPQ "alice's connInfo")
   bob <# ("", aliceConn, CON pq)
   alice <# ("", bobConn, CON pq)
   alice #: ("2", bobConn, "SEND F :hello") #> ("2", bobConn, A.MID 4 pq)
@@ -291,17 +289,15 @@ testContactConnection (alice, aPQ) (bob, bPQ) (tom, tPQ) = do
   ("1", "alice_contact", Right (INV cReq)) <- alice #: ("1", "alice_contact", "NEW T CON" <> pqConnModeStr aPQ <> " subscribe")
   let cReq' = strEncode cReq
       abPQ = pqConnectionMode aPQ bPQ
-      abPQSup = CR.pqEncToSupport abPQ
       aPQMode = CR.connPQEncryption aPQ
 
   bob #: ("11", "alice", "JOIN T " <> cReq' <> enableKEMStr bPQ <> " subscribe 14\nbob's connInfo") #> ("11", "alice", OK)
-  ("", "alice_contact", Right (A.REQ aInvId pqSup' _ "bob's connInfo")) <- (alice <#:)
-  pqSup' `shouldBe` bPQ
+  ("", "alice_contact", Right (A.REQ aInvId PQSupportOn _ "bob's connInfo")) <- (alice <#:)
   alice #: ("2", "bob", "ACPT " <> aInvId <> enableKEMStr aPQMode <> " 16\nalice's connInfo") #> ("2", "bob", OK)
   ("", "alice", Right (A.CONF bConfId pqSup'' _ "alice's connInfo")) <- (bob <#:)
-  pqSup'' `shouldBe` abPQSup
+  pqSup'' `shouldBe` bPQ
   bob #: ("12", "alice", "LET " <> bConfId <> " 16\nbob's connInfo 2") #> ("12", "alice", OK)
-  alice <# ("", "bob", A.INFO abPQSup "bob's connInfo 2")
+  alice <# ("", "bob", A.INFO (CR.connPQEncryption aPQ) "bob's connInfo 2")
   alice <# ("", "bob", CON abPQ)
   bob <# ("", "alice", CON abPQ)
   alice #: ("3", "bob", "SEND F :hi") #> ("3", "bob", A.MID 4 abPQ)
@@ -310,15 +306,13 @@ testContactConnection (alice, aPQ) (bob, bPQ) (tom, tPQ) = do
   bob #: ("13", "alice", "ACK 4") #> ("13", "alice", OK)
 
   let atPQ = pqConnectionMode aPQ tPQ
-      atPQSup = CR.pqEncToSupport atPQ
   tom #: ("21", "alice", "JOIN T " <> cReq' <> enableKEMStr tPQ <> " subscribe 14\ntom's connInfo") #> ("21", "alice", OK)
-  ("", "alice_contact", Right (A.REQ aInvId' pqSup3 _ "tom's connInfo")) <- (alice <#:)
-  pqSup3 `shouldBe` tPQ
+  ("", "alice_contact", Right (A.REQ aInvId' PQSupportOn _ "tom's connInfo")) <- (alice <#:)
   alice #: ("4", "tom", "ACPT " <> aInvId' <> enableKEMStr aPQMode <> " 16\nalice's connInfo") #> ("4", "tom", OK)
   ("", "alice", Right (A.CONF tConfId pqSup4 _ "alice's connInfo")) <- (tom <#:)
-  pqSup4 `shouldBe` atPQSup
+  pqSup4 `shouldBe` tPQ
   tom #: ("22", "alice", "LET " <> tConfId <> " 16\ntom's connInfo 2") #> ("22", "alice", OK)
-  alice <# ("", "tom", A.INFO atPQSup "tom's connInfo 2")
+  alice <# ("", "tom", A.INFO (CR.connPQEncryption aPQ) "tom's connInfo 2")
   alice <# ("", "tom", CON atPQ)
   tom <# ("", "alice", CON atPQ)
   alice #: ("5", "tom", "SEND F :hi there") #> ("5", "tom", A.MID 4 atPQ)
@@ -329,22 +323,20 @@ testContactConnection (alice, aPQ) (bob, bPQ) (tom, tPQ) = do
 testContactConnRandomIds :: Transport c => (c, InitialKeys) -> (c, PQSupport) -> IO ()
 testContactConnRandomIds (alice, aPQ) (bob, bPQ) = do
   let pq = pqConnectionMode aPQ bPQ
-      pqSup = CR.pqEncToSupport pq
   ("1", aliceContact, Right (INV cReq)) <- alice #: ("1", "", "NEW T CON" <> pqConnModeStr aPQ <> " subscribe")
   let cReq' = strEncode cReq
 
   ("11", aliceConn, Right OK) <- bob #: ("11", "", "JOIN T " <> cReq' <> enableKEMStr bPQ <> " subscribe 14\nbob's connInfo")
-  ("", aliceContact', Right (A.REQ aInvId pqSup' _ "bob's connInfo")) <- (alice <#:)
-  pqSup' `shouldBe` bPQ
+  ("", aliceContact', Right (A.REQ aInvId PQSupportOn _ "bob's connInfo")) <- (alice <#:)
   aliceContact' `shouldBe` aliceContact
 
   ("2", bobConn, Right OK) <- alice #: ("2", "", "ACPT " <> aInvId <> enableKEMStr (CR.connPQEncryption aPQ) <> " 16\nalice's connInfo")
   ("", aliceConn', Right (A.CONF bConfId pqSup'' _ "alice's connInfo")) <- (bob <#:)
-  pqSup'' `shouldBe` pqSup
+  pqSup'' `shouldBe` bPQ
   aliceConn' `shouldBe` aliceConn
 
   bob #: ("12", aliceConn, "LET " <> bConfId <> " 16\nbob's connInfo 2") #> ("12", aliceConn, OK)
-  alice <# ("", bobConn, A.INFO pqSup "bob's connInfo 2")
+  alice <# ("", bobConn, A.INFO (CR.connPQEncryption aPQ) "bob's connInfo 2")
   alice <# ("", bobConn, CON pq)
   bob <# ("", aliceConn, CON pq)
 
@@ -358,7 +350,7 @@ testRejectContactRequest _ alice bob = do
   ("1", "a_contact", Right (INV cReq)) <- alice #: ("1", "a_contact", "NEW T CON subscribe")
   let cReq' = strEncode cReq
   bob #: ("11", "alice", "JOIN T " <> cReq' <> " subscribe 10\nbob's info") #> ("11", "alice", OK)
-  ("", "a_contact", Right (A.REQ aInvId PQSupportOff _ "bob's info")) <- (alice <#:)
+  ("", "a_contact", Right (A.REQ aInvId PQSupportOn _ "bob's info")) <- (alice <#:)
   -- RJCT must use correct contact connection
   alice #: ("2a", "bob", "RJCT " <> aInvId) #> ("2a", "bob", ERR $ CONN NOT_FOUND)
   alice #: ("2b", "a_contact", "RJCT " <> aInvId) #> ("2b", "a_contact", OK)
@@ -571,12 +563,11 @@ connect' (h1, name1, pqMode1) (h2, name2, pqMode2) = do
   ("c1", _, Right (INV cReq)) <- h1 #: ("c1", name2, "NEW T INV" <> pqConnModeStr pqMode1 <> " subscribe")
   let cReq' = strEncode cReq
       pq = pqConnectionMode pqMode1 pqMode2
-      pqSup = CR.pqEncToSupport pq
   h2 #: ("c2", name1, "JOIN T " <> cReq' <> enableKEMStr pqMode2 <> " subscribe 5\ninfo2") #> ("c2", name1, OK)
   ("", _, Right (A.CONF connId pqSup' _ "info2")) <- (h1 <#:)
-  pqSup' `shouldBe` pqSup
+  pqSup' `shouldBe` CR.connPQEncryption pqMode1
   h1 #: ("c3", name2, "LET " <> connId <> " 5\ninfo1") #> ("c3", name2, OK)
-  h2 <# ("", name1, A.INFO pqSup "info1")
+  h2 <# ("", name1, A.INFO pqMode2 "info1")
   h2 <# ("", name1, CON pq)
   h1 <# ("", name2, CON pq)
 

tests/AgentTests/ConnectionRequestTests.hs

@@ -68,7 +68,7 @@ testE2ERatchetParams :: RcvE2ERatchetParamsUri 'C.X448
 testE2ERatchetParams = E2ERatchetParamsUri (mkVersionRange (VersionE2E 1) (VersionE2E 1)) testDhPubKey testDhPubKey Nothing
 
 testE2ERatchetParams12 :: RcvE2ERatchetParamsUri 'C.X448
-testE2ERatchetParams12 = E2ERatchetParamsUri (supportedE2EEncryptVRange PQSupportOn) testDhPubKey testDhPubKey Nothing
+testE2ERatchetParams12 = E2ERatchetParamsUri supportedE2EEncryptVRange testDhPubKey testDhPubKey Nothing
 
 connectionRequest :: AConnectionRequestUri
 connectionRequest =
@@ -82,7 +82,7 @@ connectionRequestCurrentRange :: AConnectionRequestUri
 connectionRequestCurrentRange =
   ACR SCMInvitation $
     CRInvitationUri
-      connReqData {crAgentVRange = supportedSMPAgentVRange PQSupportOn, crSmpQueues = [queueV1, queueV1]}
+      connReqData {crAgentVRange = supportedSMPAgentVRange, crSmpQueues = [queueV1, queueV1]}
       testE2ERatchetParams12
 
 connectionRequestClientDataEmpty :: AConnectionRequestUri

tests/AgentTests/DoubleRatchetTests.hs

@@ -93,9 +93,9 @@ fullMsgLen :: Ratchet a -> Int
 fullMsgLen Ratchet {rcSupportKEM, rcVersion} = headerLenLength + fullHeaderLen v rcSupportKEM + C.authTagSize + paddedMsgLen
   where
     v = current rcVersion
-    headerLenLength = case rcSupportKEM of
-      PQSupportOn | v >= pqRatchetE2EEncryptVersion -> 3 -- two bytes are added because of two Large used in new encoding
-      _ -> 1
+    headerLenLength
+      | v >= pqRatchetE2EEncryptVersion = 3 -- two bytes are added because of two Large used in new encoding
+      | otherwise = 1
 
 testMessageHeader :: forall a. AlgorithmI a => VersionE2E -> C.SAlgorithm a -> Expectation
 testMessageHeader v _ = do
@@ -520,7 +520,7 @@ initRatchets = do
   Right paramsBob <- pure $ pqX3dhSnd pkBob1 pkBob2 Nothing e2eAlice
   Right paramsAlice <- runExceptT $ pqX3dhRcv pkAlice1 pkAlice2 Nothing e2eBob
   (_, pkBob3) <- atomically $ C.generateKeyPair g
-  let vs = testRatchetVersions PQSupportOff
+  let vs = testRatchetVersions
       bob = initSndRatchet vs (C.publicKey pkAlice2) pkBob3 paramsBob
       alice = initRcvRatchet vs pkAlice2 paramsAlice PQSupportOff
   pure (alice, bob, encrypt' noSndKEM, decrypt' noRcvKEM, (\#>))
@@ -537,7 +537,7 @@ initRatchetsKEMProposed = do
   Right paramsBob <- pure $ pqX3dhSnd pkBob1 pkBob2 pKemParams_ e2eAlice
   Right paramsAlice <- runExceptT $ pqX3dhRcv pkAlice1 pkAlice2 Nothing e2eBob
   (_, pkBob3) <- atomically $ C.generateKeyPair g
-  let vs = testRatchetVersions PQSupportOn
+  let vs = testRatchetVersions
       bob = initSndRatchet vs (C.publicKey pkAlice2) pkBob3 paramsBob
       alice = initRcvRatchet vs pkAlice2 paramsAlice PQSupportOn
   pure (alice, bob, encrypt' hasSndKEM, decrypt' hasRcvKEM, (!#>))
@@ -555,7 +555,7 @@ initRatchetsKEMAccepted = do
   Right paramsBob <- pure $ pqX3dhSnd pkBob1 pkBob2 pKemParams_ e2eAlice
   Right paramsAlice <- runExceptT $ pqX3dhRcv pkAlice1 pkAlice2 pKem_ e2eBob
   (_, pkBob3) <- atomically $ C.generateKeyPair g
-  let vs = testRatchetVersions PQSupportOn
+  let vs = testRatchetVersions
       bob = initSndRatchet vs (C.publicKey pkAlice2) pkBob3 paramsBob
       alice = initRcvRatchet vs pkAlice2 paramsAlice PQSupportOn
   pure (alice, bob, encrypt' hasSndKEM, decrypt' hasRcvKEM, (!#>))
@@ -572,14 +572,14 @@ initRatchetsKEMProposedAgain = do
   Right paramsBob <- pure $ pqX3dhSnd pkBob1 pkBob2 pKemParams_ e2eAlice
   Right paramsAlice <- runExceptT $ pqX3dhRcv pkAlice1 pkAlice2 pKem_ e2eBob
   (_, pkBob3) <- atomically $ C.generateKeyPair g
-  let vs = testRatchetVersions PQSupportOn
+  let vs = testRatchetVersions
       bob = initSndRatchet vs (C.publicKey pkAlice2) pkBob3 paramsBob
       alice = initRcvRatchet vs pkAlice2 paramsAlice PQSupportOn
   pure (alice, bob, encrypt' hasSndKEM, decrypt' hasRcvKEM, (!#>))
 
-testRatchetVersions :: PQSupport -> RatchetVersions
-testRatchetVersions pq =
-  let v = maxVersion $ supportedE2EEncryptVRange pq
+testRatchetVersions :: RatchetVersions
+testRatchetVersions =
+  let v = maxVersion supportedE2EEncryptVRange
    in RatchetVersions v v
 
 encrypt_ :: AlgorithmI a => Maybe PQEncryption -> (TVar ChaChaDRG, Ratchet a, SkippedMsgKeys) -> ByteString -> IO (Either CryptoError (ByteString, Ratchet a, SkippedMsgDiff))

tests/AgentTests/FunctionalAPITests.hs

@@ -187,9 +187,9 @@ agentCfgVPrev :: AgentConfig
 agentCfgVPrev =
   agentCfg
     { sndAuthAlg = C.AuthAlg C.SEd25519,
-      smpAgentVRange = \_ -> prevRange $ smpAgentVRange agentCfg PQSupportOff,
+      smpAgentVRange = prevRange $ smpAgentVRange agentCfg,
       smpClientVRange = prevRange $ smpClientVRange agentCfg,
-      e2eEncryptVRange = \_ -> prevRange $ e2eEncryptVRange agentCfg PQSupportOff,
+      e2eEncryptVRange = prevRange $ e2eEncryptVRange agentCfg,
       smpCfg = smpCfgVPrev
     }
 
@@ -198,14 +198,14 @@ agentCfgV7 :: AgentConfig
 agentCfgV7 =
   agentCfg
     { sndAuthAlg = C.AuthAlg C.SX25519,
-      smpAgentVRange = \_ -> V.mkVersionRange duplexHandshakeSMPAgentVersion $ max pqdrSMPAgentVersion currentSMPAgentVersion,
-      e2eEncryptVRange = \_ -> V.mkVersionRange CR.kdfX3DHE2EEncryptVersion $ max CR.pqRatchetE2EEncryptVersion CR.currentE2EEncryptVersion,
+      smpAgentVRange = V.mkVersionRange duplexHandshakeSMPAgentVersion $ max pqdrSMPAgentVersion currentSMPAgentVersion,
+      e2eEncryptVRange = V.mkVersionRange CR.kdfX3DHE2EEncryptVersion $ max CR.pqRatchetE2EEncryptVersion CR.currentE2EEncryptVersion,
       smpCfg = smpCfgV7,
       ntfCfg = ntfCfgV2
     }
 
 agentCfgRatchetVPrev :: AgentConfig
-agentCfgRatchetVPrev = agentCfg {e2eEncryptVRange = \_ -> prevRange $ e2eEncryptVRange agentCfg PQSupportOff}
+agentCfgRatchetVPrev = agentCfg {e2eEncryptVRange = prevRange $ e2eEncryptVRange agentCfg}
 
 prevRange :: VersionRange v -> VersionRange v
 prevRange vr = vr {maxVersion = max (minVersion vr) (prevVersion $ maxVersion vr)}
@@ -546,28 +546,27 @@ testEnablePQEncryption =
     (a, 4, "msg 1") \#>\ b
     (b, 5, "msg 2") \#>\ a
     -- 45 bytes is used by agent message envelope inside double ratchet message envelope
-    let largeMsg g' pqEnc = atomically $ C.randomBytes (e2eEncUserMsgLength pqdrSMPAgentVersion pqEnc - 45) g'
+    let largeMsg g' pqEnc = atomically $ C.randomBytes (e2eEncAgentMsgLength pqdrSMPAgentVersion pqEnc - 45) g'
     lrg <- largeMsg g PQSupportOff
     (a, 6, lrg) \#>\ b
     (b, 7, lrg) \#>\ a
-    -- enabling PQ encryption
-    (a, 8, lrg) \#>! b
-    (b, 9, lrg) \#>! a
     -- switched to smaller envelopes (before reporting PQ encryption enabled)
     sml <- largeMsg g PQSupportOn
     -- fail because of message size
     Left (A.CMD LARGE) <- tryError $ A.sendMessage ca bId PQEncOn SMP.noMsgFlags lrg
-    (11, PQEncOff) <- A.sendMessage ca bId PQEncOn SMP.noMsgFlags sml
-    get ca =##> \case ("", connId, SENT 11) -> connId == bId; _ -> False
-    get cb =##> \case ("", connId, MsgErr' 10 MsgSkipped {} PQEncOff msg') -> connId == aId && msg' == sml; _ -> False
-    ackMessage cb aId 10 Nothing
+    (9, PQEncOff) <- A.sendMessage ca bId PQEncOn SMP.noMsgFlags sml
+    get ca =##> \case ("", connId, SENT 9) -> connId == bId; _ -> False
+    get cb =##> \case ("", connId, MsgErr' 8 MsgSkipped {} PQEncOff msg') -> connId == aId && msg' == sml; _ -> False
+    ackMessage cb aId 8 Nothing
     -- -- fail in reply to sync IDss
     Left (A.CMD LARGE) <- tryError $ A.sendMessage cb aId PQEncOn SMP.noMsgFlags lrg
-    (12, PQEncOn) <- A.sendMessage cb aId PQEncOn SMP.noMsgFlags sml
-    get cb =##> \case ("", connId, SENT 12) -> connId == aId; _ -> False
-    get ca =##> \case ("", connId, MsgErr' 12 MsgSkipped {} PQEncOn msg') -> connId == bId && msg' == sml; _ -> False
-    ackMessage ca bId 12 Nothing
+    (10, PQEncOff) <- A.sendMessage cb aId PQEncOn SMP.noMsgFlags sml
+    get cb =##> \case ("", connId, SENT 10) -> connId == aId; _ -> False
+    get ca =##> \case ("", connId, MsgErr' 10 MsgSkipped {} PQEncOff msg') -> connId == bId && msg' == sml; _ -> False
+    ackMessage ca bId 10 Nothing
+    (a, 11, sml) \#>! b
     -- PQ encryption now enabled
+    (b, 12, sml) !#>! a
     (a, 13, sml) !#>! b
     (b, 14, sml) !#>! a
     -- disabling PQ encryption
@@ -797,8 +796,8 @@ testAllowConnectionClientRestart t = do
 
 testIncreaseConnAgentVersion :: HasCallStack => ATransport -> IO ()
 testIncreaseConnAgentVersion t = do
-  alice <- getSMPAgentClient' 1 agentCfg {smpAgentVRange = \_ -> mkVersionRange 1 2} initAgentServers testDB
-  bob <- getSMPAgentClient' 2 agentCfg {smpAgentVRange = \_ -> mkVersionRange 1 2} initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB2
   withSmpServerStoreMsgLogOn t testPort $ \_ -> do
     (aliceId, bobId) <- runRight $ do
       (aliceId, bobId) <- makeConnection_ PQSupportOff alice bob
@@ -810,7 +809,7 @@ testIncreaseConnAgentVersion t = do
     -- version doesn't increase if incompatible
 
     disposeAgentClient alice
-    alice2 <- getSMPAgentClient' 3 agentCfg {smpAgentVRange = \_ -> mkVersionRange 1 3} initAgentServers testDB
+    alice2 <- getSMPAgentClient' 3 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
 
     runRight_ $ do
       subscribeConnection alice2 bobId
@@ -821,7 +820,7 @@ testIncreaseConnAgentVersion t = do
     -- version increases if compatible
 
     disposeAgentClient bob
-    bob2 <- getSMPAgentClient' 4 agentCfg {smpAgentVRange = \_ -> mkVersionRange 1 3} initAgentServers testDB2
+    bob2 <- getSMPAgentClient' 4 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB2
 
     runRight_ $ do
       subscribeConnection bob2 aliceId
@@ -832,7 +831,7 @@ testIncreaseConnAgentVersion t = do
     -- version doesn't decrease, even if incompatible
 
     disposeAgentClient alice2
-    alice3 <- getSMPAgentClient' 5 agentCfg {smpAgentVRange = \_ -> mkVersionRange 2 2} initAgentServers testDB
+    alice3 <- getSMPAgentClient' 5 agentCfg {smpAgentVRange = mkVersionRange 2 2} initAgentServers testDB
 
     runRight_ $ do
       subscribeConnection alice3 bobId
@@ -841,7 +840,7 @@ testIncreaseConnAgentVersion t = do
       checkVersion bob2 aliceId 3
 
     disposeAgentClient bob2
-    bob3 <- getSMPAgentClient' 6 agentCfg {smpAgentVRange = \_ -> mkVersionRange 1 1} initAgentServers testDB2
+    bob3 <- getSMPAgentClient' 6 agentCfg {smpAgentVRange = mkVersionRange 1 1} initAgentServers testDB2
 
     runRight_ $ do
       subscribeConnection bob3 aliceId
@@ -858,8 +857,8 @@ checkVersion c connId v = do
 
 testIncreaseConnAgentVersionMaxCompatible :: HasCallStack => ATransport -> IO ()
 testIncreaseConnAgentVersionMaxCompatible t = do
-  alice <- getSMPAgentClient' 1 agentCfg {smpAgentVRange = \_ -> mkVersionRange 1 2} initAgentServers testDB
-  bob <- getSMPAgentClient' 2 agentCfg {smpAgentVRange = \_ -> mkVersionRange 1 2} initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB2
   withSmpServerStoreMsgLogOn t testPort $ \_ -> do
     (aliceId, bobId) <- runRight $ do
       (aliceId, bobId) <- makeConnection_ PQSupportOff alice bob
@@ -871,7 +870,7 @@ testIncreaseConnAgentVersionMaxCompatible t = do
     -- version increases to max compatible
 
     disposeAgentClient alice
-    alice2 <- getSMPAgentClient' 3 agentCfg {smpAgentVRange = \_ -> mkVersionRange 1 3} initAgentServers testDB
+    alice2 <- getSMPAgentClient' 3 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
     disposeAgentClient bob
     bob2 <- getSMPAgentClient' 4 agentCfg {smpAgentVRange = supportedSMPAgentVRange} initAgentServers testDB2
 
@@ -886,8 +885,8 @@ testIncreaseConnAgentVersionMaxCompatible t = do
 
 testIncreaseConnAgentVersionStartDifferentVersion :: HasCallStack => ATransport -> IO ()
 testIncreaseConnAgentVersionStartDifferentVersion t = do
-  alice <- getSMPAgentClient' 1 agentCfg {smpAgentVRange = \_ -> mkVersionRange 1 2} initAgentServers testDB
-  bob <- getSMPAgentClient' 2 agentCfg {smpAgentVRange = \_ -> mkVersionRange 1 3} initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB2
   withSmpServerStoreMsgLogOn t testPort $ \_ -> do
     (aliceId, bobId) <- runRight $ do
       (aliceId, bobId) <- makeConnection_ PQSupportOff alice bob
@@ -899,7 +898,7 @@ testIncreaseConnAgentVersionStartDifferentVersion t = do
     -- version increases to max compatible
 
     disposeAgentClient alice
-    alice2 <- getSMPAgentClient' 3 agentCfg {smpAgentVRange = \_ -> mkVersionRange 1 3} initAgentServers testDB
+    alice2 <- getSMPAgentClient' 3 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
 
     runRight_ $ do
       subscribeConnection alice2 bobId
@@ -1394,10 +1393,10 @@ makeConnection = makeConnection_ PQSupportOn
 makeConnection_ :: PQSupport -> AgentClient -> AgentClient -> ExceptT AgentErrorType IO (ConnId, ConnId)
 makeConnection_ pqEnc alice bob = makeConnectionForUsers_ pqEnc alice 1 bob 1
 
-makeConnectionForUsers :: AgentClient -> UserId -> AgentClient -> UserId -> ExceptT AgentErrorType IO (ConnId, ConnId)
+makeConnectionForUsers :: HasCallStack => AgentClient -> UserId -> AgentClient -> UserId -> ExceptT AgentErrorType IO (ConnId, ConnId)
 makeConnectionForUsers = makeConnectionForUsers_ PQSupportOn
 
-makeConnectionForUsers_ :: PQSupport -> AgentClient -> UserId -> AgentClient -> UserId -> ExceptT AgentErrorType IO (ConnId, ConnId)
+makeConnectionForUsers_ :: HasCallStack => PQSupport -> AgentClient -> UserId -> AgentClient -> UserId -> ExceptT AgentErrorType IO (ConnId, ConnId)
 makeConnectionForUsers_ pqSupport alice aliceUserId bob bobUserId = do
   (bobId, qInfo) <- A.createConnection alice aliceUserId True SCMInvitation Nothing (CR.IKNoPQ pqSupport) SMSubscribe
   aliceId <- A.joinConnection bob bobUserId True qInfo "bob's connInfo" pqSupport SMSubscribe
@@ -2454,8 +2453,8 @@ testDeliveryReceipts =
 
 testDeliveryReceiptsVersion :: HasCallStack => ATransport -> IO ()
 testDeliveryReceiptsVersion t = do
-  a <- getSMPAgentClient' 1 agentCfg {smpAgentVRange = \_ -> mkVersionRange 1 3} initAgentServers testDB
-  b <- getSMPAgentClient' 2 agentCfg {smpAgentVRange = \_ -> mkVersionRange 1 3} initAgentServers testDB2
+  a <- getSMPAgentClient' 1 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
+  b <- getSMPAgentClient' 2 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB2
   withSmpServerStoreMsgLogOn t testPort $ \_ -> do
     (aId, bId) <- runRight $ do
       (aId, bId) <- makeConnection_ PQSupportOff a b
@@ -2482,8 +2481,8 @@ testDeliveryReceiptsVersion t = do
       subscribeConnection a' bId
       subscribeConnection b' aId
       exchangeGreetingsMsgId_ PQEncOff 6 a' bId b' aId
-      checkVersion a' bId 4
-      checkVersion b' aId 4
+      checkVersion a' bId 5
+      checkVersion b' aId 5
       (8, PQEncOff) <- A.sendMessage a' bId PQEncOn SMP.noMsgFlags "hello"
       get a' ##> ("", bId, SENT 8)
       get b' =##> \case ("", c, Msg' 8 PQEncOff "hello") -> c == aId; _ -> False