WIP

* server: more efficient response to batched subscriptions

* add sndMsgQ for interleaving messages with replies

* remove redundant liftIO

* refactor

* refactor2

* rename

* fix

* diff

* remove comment

* remove comment

---------

Co-authored-by: Alexander Bondarenko <486682+dpwiz@users.noreply.github.com>

CHANGELOG.md

@@ -1,3 +1,18 @@
+# 5.7.3
+
+SMP/NTF protocol:
+- add ALPN for handshake version negotiation, similar to XFTP (to preserve backwards compatibility with the old clients).
+- upgrade clients to versions v7/v2 of the protocols.
+
+SMP server:
+- faster responses to subscription requests.
+
+XFTP client:
+- fix network exception during file download treated as permanent file error.
+
+SMP agent:
+- do not report subscription timeouts while client is offline.
+
 # 5.7.2
 
 SMP agent:

package.yaml

@@ -1,5 +1,5 @@
 name: simplexmq
-version: 5.7.2.0
+version: 5.7.4.0
 synopsis: SimpleXMQ message broker
 description: |
   This package includes <./docs/Simplex-Messaging-Server.html server>,
@@ -71,6 +71,7 @@ dependencies:
   - transformers == 0.6.*
   - unliftio == 0.2.*
   - unliftio-core == 0.2.*
+  - unordered-containers
   - websockets == 0.12.*
   - yaml == 0.11.*
   - zstd == 0.1.3.*

simplexmq.cabal

@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack
 
 name:           simplexmq
-version:        5.7.2.0
+version:        5.7.4.0
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -230,6 +230,7 @@ library
     , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
+    , unordered-containers
     , websockets ==0.12.*
     , yaml ==0.11.*
     , zstd ==0.1.3.*
@@ -305,6 +306,7 @@ executable ntf-server
     , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
+    , unordered-containers
     , websockets ==0.12.*
     , yaml ==0.11.*
     , zstd ==0.1.3.*
@@ -380,6 +382,7 @@ executable smp-agent
     , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
+    , unordered-containers
     , websockets ==0.12.*
     , yaml ==0.11.*
     , zstd ==0.1.3.*
@@ -455,6 +458,7 @@ executable smp-server
     , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
+    , unordered-containers
     , websockets ==0.12.*
     , yaml ==0.11.*
     , zstd ==0.1.3.*
@@ -530,6 +534,7 @@ executable xftp
     , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
+    , unordered-containers
     , websockets ==0.12.*
     , yaml ==0.11.*
     , zstd ==0.1.3.*
@@ -605,6 +610,7 @@ executable xftp-server
     , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
+    , unordered-containers
     , websockets ==0.12.*
     , yaml ==0.11.*
     , zstd ==0.1.3.*
@@ -721,6 +727,7 @@ test-suite simplexmq-test
     , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
+    , unordered-containers
     , websockets ==0.12.*
     , yaml ==0.11.*
     , zstd ==0.1.3.*

src/Simplex/FileTransfer/Client.hs

@@ -245,8 +245,13 @@ downloadXFTPChunk g c@XFTPClient {config} rpKey fId chunkSpec@XFTPRcvChunkSpec {
         let dhSecret = C.dh' sDhKey rpDhKey
         cbState <- liftEither . first PCECryptoError $ LC.cbInit dhSecret cbNonce
         let t = chunkTimeout config chunkSize
-        ExceptT (sequence <$> (t `timeout` download cbState)) >>= maybe (throwError PCEResponseTimeout) pure
+        ExceptT (sequence <$> (t `timeout` (download cbState `catches` errors))) >>= maybe (throwError PCEResponseTimeout) pure
         where
+          errors =
+            [ Handler $ \(_e :: H.HTTP2Error) -> pure $ Left PCENetworkError,
+              Handler $ \(e :: IOException) -> pure $ Left (PCEIOError e),
+              Handler $ \(_e :: SomeException) -> pure $ Left PCENetworkError
+            ]
           download cbState =
             runExceptT . withExceptT PCEResponseError $
               receiveEncFile chunkPart cbState chunkSpec `catchError` \e ->

src/Simplex/FileTransfer/Server/Env.hs

@@ -13,12 +13,9 @@ import Control.Logger.Simple
 import Control.Monad
 import Control.Monad.IO.Unlift
 import Crypto.Random
-import Data.Default (def)
 import Data.Int (Int64)
-import Data.List (find)
 import Data.List.NonEmpty (NonEmpty)
 import qualified Data.Map.Strict as M
-import Data.Maybe (fromMaybe)
 import Data.Time.Clock (getCurrentTime)
 import Data.Word (Word32)
 import Data.X509.Validation (Fingerprint (..))
@@ -103,7 +100,7 @@ supportedXFTPhandshakes :: [ALPN]
 supportedXFTPhandshakes = ["xftp/1"]
 
 newXFTPServerEnv :: XFTPServerConfig -> IO XFTPEnv
-newXFTPServerEnv config@XFTPServerConfig {storeLogFile, fileSizeQuota, caCertificateFile, certificateFile, privateKeyFile} = do
+newXFTPServerEnv config@XFTPServerConfig {storeLogFile, fileSizeQuota, caCertificateFile, certificateFile, privateKeyFile, transportConfig} = do
   random <- liftIO C.newRandom
   store <- atomically newFileStore
   storeLog <- liftIO $ mapM (`readWriteFileStore` store) storeLogFile
@@ -112,17 +109,7 @@ newXFTPServerEnv config@XFTPServerConfig {storeLogFile, fileSizeQuota, caCertifi
   forM_ fileSizeQuota $ \quota -> do
     logInfo $ "Total / available storage: " <> tshow quota <> " / " <> tshow (quota - used)
     when (quota < used) $ logInfo "WARNING: storage quota is less than used storage, no files can be uploaded!"
-  tlsServerParams' <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile
-  let TransportServerConfig {alpn} = transportConfig config
-  let tlsServerParams = case alpn of
-        Nothing -> tlsServerParams'
-        Just supported ->
-          tlsServerParams'
-            { T.serverHooks =
-                def
-                  { T.onALPNClientSuggest = Just $ pure . fromMaybe "" . find (`elem` supported)
-                  }
-            }
+  tlsServerParams <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile (alpn transportConfig)
   Fingerprint fp <- liftIO $ loadFingerprint caCertificateFile
   serverStats <- atomically . newFileServerStats =<< liftIO getCurrentTime
   pure XFTPEnv {config, store, storeLog, random, tlsServerParams, serverIdentity = C.KeyHash fp, serverStats}

src/Simplex/Messaging/Agent/Client.hs

@@ -630,12 +630,16 @@ reconnectSMPClient tc c tSess@(_, srv, _) qs = do
   let t = (length qs `div` 90 + 1) * tcpTimeout * 3
   ExceptT (sequence <$> (t `timeout` runExceptT resubscribe)) >>= \case
     Just _ -> atomically $ writeTVar tc 0
-    Nothing -> do
-      tc' <- atomically $ stateTVar tc $ \i -> (i + 1, i + 1)
-      maxTC <- asks $ maxSubscriptionTimeouts . config
-      let err = if tc' >= maxTC then CRITICAL True else INTERNAL
-          msg = show tc' <> " consecutive subscription timeouts: " <> show (length qs) <> " queues, transport session: " <> show tSess
-      atomically $ writeTBQueue (subQ c) ("", "", APC SAEConn $ ERR $ err msg)
+    Nothing ->
+      (offline <$> readTVarIO (userNetworkState c)) >>= \case
+        -- reset and do not report consequitive timeouts while offline
+        Just _ -> atomically $ writeTVar tc 0
+        Nothing -> do
+          tc' <- atomically $ stateTVar tc $ \i -> (i + 1, i + 1)
+          maxTC <- asks $ maxSubscriptionTimeouts . config
+          when (tc' >= maxTC) $ do
+            let msg = show tc' <> " consecutive subscription timeouts: " <> show (length qs) <> " queues, transport session: " <> show tSess
+            atomically $ writeTBQueue (subQ c) ("", "", APC SAEConn $ ERR $ INTERNAL msg)
   where
     resubscribe :: AM ()
     resubscribe = do

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

@@ -181,7 +181,7 @@ defaultAgentConfig =
       maxWorkerRestartsPerMin = 5,
       -- 3 consecutive subscription timeouts will result in alert to the user
       -- this is a fallback, as the timeout set to 3x of expected timeout, to avoid potential locking.
-      maxSubscriptionTimeouts = 3,
+      maxSubscriptionTimeouts = 5,
       storedMsgDataTTL = 21 * nominalDay,
       rcvFilesTTL = 2 * nominalDay,
       sndFilesTTL = nominalDay,

src/Simplex/Messaging/Agent/Server.hs

@@ -49,7 +49,7 @@ runSMPAgentBlocking (ATransport t) cfg@AgentConfig {tcpPort, caCertificateFile,
     smpAgent :: forall c. Transport c => TProxy c -> ServiceName -> Env -> IO ()
     smpAgent _ port env = do
       -- tlsServerParams is not in Env to avoid breaking functional API w/t key and certificate generation
-      tlsServerParams <- loadTLSServerParams caCertificateFile certificateFile privateKeyFile
+      tlsServerParams <- loadTLSServerParams caCertificateFile certificateFile privateKeyFile Nothing
       clientId <- newTVarIO initClientId
       runTransportServer started port tlsServerParams defaultTransportServerConfig $ \(h :: c) -> do
         putLn h $ "Welcome to SMP agent v" <> B.pack simplexMQVersion

src/Simplex/Messaging/Client.hs

@@ -260,6 +260,7 @@ data ProtocolClientConfig v = ProtocolClientConfig
     defaultTransport :: (ServiceName, ATransport),
     -- | network configuration
     networkConfig :: NetworkConfig,
+    clientALPN :: Maybe [ALPN],
     -- | client-server protocol version range
     serverVRange :: VersionRange v,
     -- | agree shared session secret (used in SMP proxy)
@@ -267,19 +268,20 @@ data ProtocolClientConfig v = ProtocolClientConfig
   }
 
 -- | Default protocol client configuration.
-defaultClientConfig :: VersionRange v -> ProtocolClientConfig v
-defaultClientConfig serverVRange =
+defaultClientConfig :: Maybe [ALPN] -> VersionRange v -> ProtocolClientConfig v
+defaultClientConfig clientALPN serverVRange =
   ProtocolClientConfig
     { qSize = 64,
       defaultTransport = ("443", transport @TLS),
       networkConfig = defaultNetworkConfig,
+      clientALPN,
       serverVRange,
       agreeSecret = False
     }
 {-# INLINE defaultClientConfig #-}
 
 defaultSMPClientConfig :: ProtocolClientConfig SMPVersion
-defaultSMPClientConfig = defaultClientConfig supportedClientSMPRelayVRange
+defaultSMPClientConfig = defaultClientConfig (Just supportedSMPHandshakes) supportedClientSMPRelayVRange
 {-# INLINE defaultSMPClientConfig #-}
 
 data Request err msg = Request
@@ -332,7 +334,7 @@ type TransportSession msg = (UserId, ProtoServer msg, Maybe EntityId)
 -- A single queue can be used for multiple 'SMPClient' instances,
 -- as 'SMPServerTransmission' includes server information.
 getProtocolClient :: forall v err msg. Protocol v err msg => TVar ChaChaDRG -> TransportSession msg -> ProtocolClientConfig v -> Maybe (TBQueue (ServerTransmission v msg)) -> (ProtocolClient v err msg -> IO ()) -> IO (Either (ProtocolClientError err) (ProtocolClient v err msg))
-getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize, networkConfig, serverVRange, agreeSecret} msgQ disconnected = do
+getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize, networkConfig, clientALPN, serverVRange, agreeSecret} msgQ disconnected = do
   case chooseTransportHost networkConfig (host srv) of
     Right useHost ->
       (getCurrentTime >>= atomically . mkProtocolClient useHost >>= runClient useTransport useHost)
@@ -370,7 +372,7 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
     runClient :: (ServiceName, ATransport) -> TransportHost -> PClient v err msg -> IO (Either (ProtocolClientError err) (ProtocolClient v err msg))
     runClient (port', ATransport t) useHost c = do
       cVar <- newEmptyTMVarIO
-      let tcConfig = transportClientConfig networkConfig
+      let tcConfig = (transportClientConfig networkConfig) {alpn = clientALPN}
           username = proxyUsername transportSession
       action <-
         async $

src/Simplex/Messaging/Notifications/Client.hs

@@ -10,7 +10,7 @@ import Data.Word (Word16)
 import Simplex.Messaging.Client
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Notifications.Protocol
-import Simplex.Messaging.Notifications.Transport (NTFVersion, supportedClientNTFVRange)
+import Simplex.Messaging.Notifications.Transport (NTFVersion, supportedClientNTFVRange, supportedNTFHandshakes)
 import Simplex.Messaging.Protocol (ErrorType)
 import Simplex.Messaging.Util (bshow)
 
@@ -19,7 +19,7 @@ type NtfClient = ProtocolClient NTFVersion ErrorType NtfResponse
 type NtfClientError = ProtocolClientError ErrorType
 
 defaultNTFClientConfig :: ProtocolClientConfig NTFVersion
-defaultNTFClientConfig = defaultClientConfig supportedClientNTFVRange
+defaultNTFClientConfig = defaultClientConfig (Just supportedNTFHandshakes) supportedClientNTFVRange
 
 ntfRegisterToken :: NtfClient -> C.APrivateAuthKey -> NewNtfEntity 'Token -> ExceptT NtfClientError IO (NtfTokenId, C.PublicKeyX25519)
 ntfRegisterToken c pKey newTkn =

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

@@ -34,7 +34,7 @@ import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport (ATransport, THandleParams, TransportPeer (..))
-import Simplex.Messaging.Transport.Server (TransportServerConfig, loadFingerprint, loadTLSServerParams)
+import Simplex.Messaging.Transport.Server (TransportServerConfig, alpn, loadFingerprint, loadTLSServerParams)
 import System.IO (IOMode (..))
 import System.Mem.Weak (Weak)
 import UnliftIO.STM
@@ -84,7 +84,7 @@ data NtfEnv = NtfEnv
   }
 
 newNtfServerEnv :: NtfServerConfig -> IO NtfEnv
-newNtfServerEnv config@NtfServerConfig {subQSize, pushQSize, smpAgentCfg, apnsConfig, storeLogFile, caCertificateFile, certificateFile, privateKeyFile} = do
+newNtfServerEnv config@NtfServerConfig {subQSize, pushQSize, smpAgentCfg, apnsConfig, storeLogFile, caCertificateFile, certificateFile, privateKeyFile, transportConfig} = do
   random <- liftIO C.newRandom
   store <- atomically newNtfStore
   logInfo "restoring subscriptions..."
@@ -92,7 +92,7 @@ newNtfServerEnv config@NtfServerConfig {subQSize, pushQSize, smpAgentCfg, apnsCo
   logInfo "restored subscriptions"
   subscriber <- atomically $ newNtfSubscriber subQSize smpAgentCfg random
   pushServer <- atomically $ newNtfPushServer pushQSize apnsConfig
-  tlsServerParams <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile
+  tlsServerParams <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile (alpn transportConfig)
   Fingerprint fp <- liftIO $ loadFingerprint caCertificateFile
   serverStats <- atomically . newNtfServerStats =<< liftIO getCurrentTime
   pure NtfEnv {config, subscriber, pushServer, store, storeLog, random, tlsServerParams, serverIdentity = C.KeyHash fp, serverStats}

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

@@ -18,7 +18,7 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Notifications.Server (runNtfServer)
 import Simplex.Messaging.Notifications.Server.Env (NtfServerConfig (..), defaultInactiveClientExpiration)
 import Simplex.Messaging.Notifications.Server.Push.APNS (defaultAPNSPushClientConfig)
-import Simplex.Messaging.Notifications.Transport (supportedServerNTFVRange)
+import Simplex.Messaging.Notifications.Transport (supportedNTFHandshakes, supportedServerNTFVRange)
 import Simplex.Messaging.Protocol (ProtoServerWithAuth (..), pattern NtfServer)
 import Simplex.Messaging.Server.CLI
 import Simplex.Messaging.Server.Expiration
@@ -133,7 +133,8 @@ ntfServerCLI cfgPath logPath =
               ntfServerVRange = supportedServerNTFVRange,
               transportConfig =
                 defaultTransportServerConfig
-                  { logTLSErrors = fromMaybe False $ iniOnOff "TRANSPORT" "log_tls_errors" ini
+                  { logTLSErrors = fromMaybe False $ iniOnOff "TRANSPORT" "log_tls_errors" ini,
+                    alpn = Just supportedNTFHandshakes
                   }
             }
 

src/Simplex/Messaging/Notifications/Transport.hs

@@ -44,7 +44,7 @@ authBatchCmdsNTFVersion :: VersionNTF
 authBatchCmdsNTFVersion = VersionNTF 2
 
 currentClientNTFVersion :: VersionNTF
-currentClientNTFVersion = VersionNTF 1
+currentClientNTFVersion = VersionNTF 2
 
 currentServerNTFVersion :: VersionNTF
 currentServerNTFVersion = VersionNTF 2
@@ -52,9 +52,15 @@ currentServerNTFVersion = VersionNTF 2
 supportedClientNTFVRange :: VersionRangeNTF
 supportedClientNTFVRange = mkVersionRange initialNTFVersion currentClientNTFVersion
 
+legacyServerNTFVRange :: VersionRangeNTF
+legacyServerNTFVRange = mkVersionRange initialNTFVersion initialNTFVersion
+
 supportedServerNTFVRange :: VersionRangeNTF
 supportedServerNTFVRange = mkVersionRange initialNTFVersion currentServerNTFVersion
 
+supportedNTFHandshakes :: [ALPN]
+supportedNTFHandshakes = ["ntf/1"]
+
 type THandleNTF c p = THandle NTFVersion c p
 
 data NtfServerHandshake = NtfServerHandshake
@@ -104,12 +110,13 @@ ntfServerHandshake :: forall c. Transport c => C.APrivateSignKey -> c -> C.KeyPa
 ntfServerHandshake serverSignKey c (k, pk) kh ntfVRange = do
   let th@THandle {params = THandleParams {sessionId}} = ntfTHandle c
   let sk = C.signX509 serverSignKey $ C.publicToX509 k
-  sendHandshake th $ NtfServerHandshake {sessionId, ntfVersionRange = ntfVRange, authPubKey = Just sk}
+  let ntfVersionRange = maybe legacyServerNTFVRange (const ntfVRange) $ getSessionALPN c
+  sendHandshake th $ NtfServerHandshake {sessionId, ntfVersionRange, authPubKey = Just sk}
   getHandshake th >>= \case
     NtfClientHandshake {ntfVersion = v, keyHash}
       | keyHash /= kh ->
           throwError $ TEHandshake IDENTITY
-      | v `isCompatible` ntfVRange ->
+      | v `isCompatible` ntfVersionRange ->
           pure $ ntfThHandleServer th v pk
       | otherwise -> throwError $ TEHandshake VERSION
 

src/Simplex/Messaging/Server.hs

@@ -51,9 +51,11 @@ import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy.Char8 as LB
 import Data.Either (fromRight, partitionEithers)
 import Data.Functor (($>))
+import qualified Data.HashMap.Strict as HM
 import Data.Int (Int64)
 import qualified Data.IntMap.Strict as IM
-import Data.List (intercalate)
+import Data.List (intercalate, mapAccumR)
+import Data.List.NonEmpty (NonEmpty)
 import qualified Data.List.NonEmpty as L
 import qualified Data.Map.Strict as M
 import Data.Maybe (isNothing)
@@ -410,7 +412,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
                     hPutStrLn h "AUTH"
 
 runClientTransport :: Transport c => THandleSMP c 'TServer -> M ()
-runClientTransport th@THandle {params = THandleParams {thVersion, sessionId}} = do
+runClientTransport h@THandle {params = THandleParams {thVersion, sessionId}} = do
   q <- asks $ tbqSize . config
   ts <- liftIO getSystemTime
   active <- asks clients
@@ -421,11 +423,12 @@ runClientTransport th@THandle {params = THandleParams {thVersion, sessionId}} =
     pure new
   s <- asks server
   expCfg <- asks $ inactiveClientExpiration . config
+  th <- newMVar h -- put TH under a fair lock to interleave messages and command responses
   labelMyThread . B.unpack $ "client $" <> encode sessionId
-  raceAny_ ([liftIO $ send th c, client c s, receive th c] <> disconnectThread_ c expCfg)
+  raceAny_ ([liftIO $ send th c, liftIO $ sendMsg th c, client c s, receive h c] <> disconnectThread_ c expCfg)
     `finally` clientDisconnected c
   where
-    disconnectThread_ c (Just expCfg) = [liftIO $ disconnectTransport th (rcvActiveAt c) (sndActiveAt c) expCfg (noSubscriptions c)]
+    disconnectThread_ c (Just expCfg) = [liftIO $ disconnectTransport h (rcvActiveAt c) (sndActiveAt c) expCfg (noSubscriptions c)]
     disconnectThread_ _ _ = []
     noSubscriptions c = atomically $ (&&) <$> TM.null (subscriptions c) <*> TM.null (ntfSubscriptions c)
 
@@ -458,10 +461,10 @@ cancelSub sub =
     _ -> return ()
 
 receive :: Transport c => THandleSMP c 'TServer -> Client -> M ()
-receive th@THandle {params = THandleParams {thAuth}} Client {rcvQ, sndQ, rcvActiveAt, sessionId} = do
+receive h@THandle {params = THandleParams {thAuth}} Client {rcvQ, sndQ, rcvActiveAt, sessionId} = do
   labelMyThread . B.unpack $ "client $" <> encode sessionId <> " receive"
   forever $ do
-    ts <- L.toList <$> liftIO (tGet th)
+    ts <- L.toList <$> liftIO (tGet h)
     atomically . writeTVar rcvActiveAt =<< liftIO getSystemTime
     as <- partitionEithers <$> mapM cmdAction ts
     write sndQ $ fst as
@@ -478,20 +481,41 @@ receive th@THandle {params = THandleParams {thAuth}} Client {rcvQ, sndQ, rcvActi
               VRFailed -> Left (corrId, queueId, ERR AUTH)
     write q = mapM_ (atomically . writeTBQueue q) . L.nonEmpty
 
-send :: Transport c => THandleSMP c 'TServer -> Client -> IO ()
-send h@THandle {params} Client {sndQ, sessionId, sndActiveAt} = do
+send :: Transport c => MVar (THandleSMP c 'TServer) -> Client -> IO ()
+send th c@Client {sndQ, msgQ, sessionId} = do
   labelMyThread . B.unpack $ "client $" <> encode sessionId <> " send"
-  forever $ do
-    ts <- atomically $ L.sortWith tOrder <$> readTBQueue sndQ
-    -- TODO we can authorize responses as well
-    void . liftIO . tPut h $ L.map (\t -> Right (Nothing, encodeTransmission params t)) ts
-    atomically . writeTVar sndActiveAt =<< liftIO getSystemTime
+  forever $ atomically (readTBQueue sndQ) >>= sendTransmissions
   where
-    tOrder :: Transmission BrokerMsg -> Int
-    tOrder (_, _, cmd) = case cmd of
-      MSG {} -> 0
-      NMSG {} -> 0
-      _ -> 1
+    sendTransmissions :: NonEmpty (Transmission BrokerMsg) -> IO ()
+    sendTransmissions ts
+      | L.length ts <= 2 = tSend th c ts
+      | otherwise = do
+          let (msgs_, ts') = mapAccumR splitMessages [] ts
+          -- If the request had batched subscriptions (L.length ts > 2)
+          -- this will reply OK to all SUBs in the first batched transmission,
+          -- to reduce client timeouts.
+          tSend th c ts'
+          -- After that all messages will be sent in separate transmissions,
+          -- without any client response timeouts, and allowing them to interleave
+          -- with other requests responses.
+          mapM_ (atomically . writeTBQueue msgQ) $ L.nonEmpty msgs_
+      where
+        splitMessages :: [Transmission BrokerMsg] -> Transmission BrokerMsg -> ([Transmission BrokerMsg], Transmission BrokerMsg)
+        splitMessages msgs t@(corrId, entId, cmd) = case cmd of
+          -- replace MSG response with OK, accumulating MSG in a separate list.
+          MSG {} -> ((CorrId "", entId, cmd) : msgs, (corrId, entId, OK))
+          _ -> (msgs, t)
+
+sendMsg :: Transport c => MVar (THandleSMP c 'TServer) -> Client -> IO ()
+sendMsg th c@Client {msgQ, sessionId} = do
+  labelMyThread . B.unpack $ "client $" <> encode sessionId <> " sendMsg"
+  forever $ atomically (readTBQueue msgQ) >>= mapM_ (\t -> tSend th c [t])
+
+tSend :: Transport c => MVar (THandleSMP c 'TServer) -> Client -> NonEmpty (Transmission BrokerMsg) -> IO ()
+tSend th Client {sndActiveAt} ts = do
+  withMVar th $ \h@THandle {params} ->
+    void . tPut h $ L.map (\t -> Right (Nothing, encodeTransmission params t)) ts
+  atomically . writeTVar sndActiveAt =<< liftIO getSystemTime
 
 disconnectTransport :: Transport c => THandle v c 'TServer -> TVar SystemTime -> TVar SystemTime -> ExpirationConfig -> IO Bool -> IO ()
 disconnectTransport THandle {connection, params = THandleParams {sessionId}} rcvActiveAt sndActiveAt expCfg noSubscriptions = do
@@ -518,7 +542,11 @@ verifyTransmission :: Maybe (THandleAuth 'TServer, C.CbNonce) -> Maybe Transmiss
 verifyTransmission auth_ tAuth authorized queueId cmd =
   case cmd of
     Cmd SRecipient (NEW k _ _ _) -> pure $ Nothing `verifiedWith` k
-    Cmd SRecipient _ -> verifyQueue (\q -> Just q `verifiedWith` recipientKey q) <$> get SRecipient
+    Cmd SRecipient _ -> do
+      QueueStore {recipientKeys} <- asks queueStore
+      (queueId `HM.lookup`) <$> readTVarIO recipientKeys >>= \case
+        Nothing -> pure $! dummyVerify
+        Just rKey -> if verify rKey then slowIO >> get SRecipient >>= either (\_notGonnaHappenUnlessDeletedWhileCheckingAuth -> pure VRFailed) (pure . VRVerified . Just) else pure VRFailed
     -- SEND will be accepted without authorization before the queue is secured with KEY command
     Cmd SSender SEND {} -> verifyQueue (\q -> Just q `verified` maybe (isNothing tAuth) verify (senderKey q)) <$> get SSender
     Cmd SSender PING -> pure $ VRVerified Nothing
@@ -531,6 +559,8 @@ verifyTransmission auth_ tAuth authorized queueId cmd =
     verifyQueue = either (\_ -> dummyVerify)
     verified q cond = if cond then VRVerified q else VRFailed
     verifiedWith q k = q `verified` verify k
+    slowIO :: M ()
+    slowIO = liftIO $ threadDelay 100000 -- XXX: Some thumb twiddling to prove that fetching queue data doesn't allow to distinguish bad key and non-existent queue.
     get :: SParty p -> M (Either ErrorType QueueRec)
     get party = do
       st <- asks queueStore
@@ -975,9 +1005,10 @@ saveServerMessages keepMsgs = asks (storeMsgsFile . config) >>= mapM_ saveMessag
             >>= mapM_ (B.hPutStrLn h . strEncode . MLRv3 rId)
 
 restoreServerMessages :: M Int
-restoreServerMessages = asks (storeMsgsFile . config) >>= \case
-  Just f -> ifM (doesFileExist f) (restoreMessages f) (pure 0)
-  Nothing -> pure 0
+restoreServerMessages =
+  asks (storeMsgsFile . config) >>= \case
+    Just f -> ifM (doesFileExist f) (restoreMessages f) (pure 0)
+    Nothing -> pure 0
   where
     restoreMessages f = do
       logInfo $ "restoring messages from file " <> T.pack f

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

@@ -33,8 +33,8 @@ import Simplex.Messaging.Server.Stats
 import Simplex.Messaging.Server.StoreLog
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
-import Simplex.Messaging.Transport (ATransport, VersionSMP, VersionRangeSMP)
-import Simplex.Messaging.Transport.Server (SocketState, TransportServerConfig, loadFingerprint, loadTLSServerParams, newSocketState)
+import Simplex.Messaging.Transport (ATransport, VersionRangeSMP, VersionSMP)
+import Simplex.Messaging.Transport.Server (SocketState, TransportServerConfig, alpn, loadFingerprint, loadTLSServerParams, newSocketState)
 import System.IO (IOMode (..))
 import System.Mem.Weak (Weak)
 import UnliftIO.STM
@@ -128,6 +128,7 @@ data Client = Client
     ntfSubscriptions :: TMap NotifierId (),
     rcvQ :: TBQueue (NonEmpty (Maybe QueueRec, Transmission Cmd)),
     sndQ :: TBQueue (NonEmpty (Transmission BrokerMsg)),
+    msgQ :: TBQueue (NonEmpty (Transmission BrokerMsg)),
     endThreads :: TVar (IntMap (Weak ThreadId)),
     endThreadSeq :: TVar Int,
     thVersion :: VersionSMP,
@@ -161,12 +162,13 @@ newClient nextClientId qSize thVersion sessionId createdAt = do
   ntfSubscriptions <- TM.empty
   rcvQ <- newTBQueue qSize
   sndQ <- newTBQueue qSize
+  msgQ <- newTBQueue qSize
   endThreads <- newTVar IM.empty
   endThreadSeq <- newTVar 0
   connected <- newTVar True
   rcvActiveAt <- newTVar createdAt
   sndActiveAt <- newTVar createdAt
-  return Client {clientId, subscriptions, ntfSubscriptions, rcvQ, sndQ, endThreads, endThreadSeq, thVersion, sessionId, connected, createdAt, rcvActiveAt, sndActiveAt}
+  return Client {clientId, subscriptions, ntfSubscriptions, rcvQ, sndQ, msgQ, endThreads, endThreadSeq, thVersion, sessionId, connected, createdAt, rcvActiveAt, sndActiveAt}
 
 newSubscription :: SubscriptionThread -> STM Sub
 newSubscription subThread = do
@@ -174,13 +176,13 @@ newSubscription subThread = do
   return Sub {subThread, delivered}
 
 newEnv :: ServerConfig -> IO Env
-newEnv config@ServerConfig {caCertificateFile, certificateFile, privateKeyFile, storeLogFile} = do
+newEnv config@ServerConfig {caCertificateFile, certificateFile, privateKeyFile, storeLogFile, transportConfig} = do
   server <- atomically newServer
   queueStore <- atomically newQueueStore
   msgStore <- atomically newMsgStore
   random <- liftIO C.newRandom
   storeLog <- restoreQueues queueStore `mapM` storeLogFile
-  tlsServerParams <- loadTLSServerParams caCertificateFile certificateFile privateKeyFile
+  tlsServerParams <- loadTLSServerParams caCertificateFile certificateFile privateKeyFile (alpn transportConfig)
   Fingerprint fp <- loadFingerprint caCertificateFile
   let serverIdentity = KeyHash fp
   serverStats <- atomically . newServerStats =<< getCurrentTime

src/Simplex/Messaging/Server/Main.hs

@@ -25,7 +25,7 @@ import Simplex.Messaging.Server (runSMPServer)
 import Simplex.Messaging.Server.CLI
 import Simplex.Messaging.Server.Env.STM (ServerConfig (..), defMsgExpirationDays, defaultInactiveClientExpiration, defaultMessageExpiration)
 import Simplex.Messaging.Server.Expiration
-import Simplex.Messaging.Transport (simplexMQVersion, supportedServerSMPRelayVRange)
+import Simplex.Messaging.Transport (simplexMQVersion, supportedSMPHandshakes, supportedServerSMPRelayVRange)
 import Simplex.Messaging.Transport.Client (TransportHost (..))
 import Simplex.Messaging.Transport.Server (TransportServerConfig (..), defaultTransportServerConfig)
 import Simplex.Messaging.Util (safeDecodeUtf8)
@@ -211,7 +211,8 @@ smpServerCLI cfgPath logPath =
               smpServerVRange = supportedServerSMPRelayVRange,
               transportConfig =
                 defaultTransportServerConfig
-                  { logTLSErrors = fromMaybe False $ iniOnOff "TRANSPORT" "log_tls_errors" ini
+                  { logTLSErrors = fromMaybe False $ iniOnOff "TRANSPORT" "log_tls_errors" ini,
+                    alpn = Just supportedSMPHandshakes
                   },
               controlPort = either (const Nothing) (Just . T.unpack) $ lookupValue "TRANSPORT" "control_port" ini
             }

src/Simplex/Messaging/Server/QueueStore.hs

@@ -8,6 +8,9 @@ module Simplex.Messaging.Server.QueueStore where
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol
 
+-- normalized/storage form
+-- can be split into per-party pieces
+-- id/public key form party index, dhsecrets can use dense storage or loaded on demand
 data QueueRec = QueueRec
   { recipientId :: !RecipientId,
     recipientKey :: !RcvPublicAuthKey,

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

@@ -24,6 +24,8 @@ where
 
 import Control.Monad
 import Data.Functor (($>))
+import Data.HashMap.Strict (HashMap)
+import qualified Data.HashMap.Strict as HM
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.QueueStore
 import Simplex.Messaging.TMap (TMap)
@@ -32,24 +34,32 @@ import Simplex.Messaging.Util (ifM, ($>>=))
 import UnliftIO.STM
 
 data QueueStore = QueueStore
-  { queues :: TMap RecipientId (TVar QueueRec),
-    senders :: TMap SenderId RecipientId,
-    notifiers :: TMap NotifierId RecipientId
+  { recipientKeys :: TVar (HashMap RecipientId RcvPublicAuthKey), -- O(1) index for const-time auth checks
+    senderKeys :: TVar (HashMap SenderId SndPublicAuthKey), -- O(1) index for const-time auth checks
+    notifierKeys :: TVar (HashMap NotifierId NtfPublicAuthKey), -- O(1) index for const-time auth checks
+    queues :: TMap RecipientId (TVar QueueRec), -- can be stored cold; and indexed with ints (using annotated keys below)
+    senders :: TMap SenderId RecipientId, -- not needed with annotated keys
+    notifiers :: TMap NotifierId RecipientId -- not needed with annotated keys
   }
 
 newQueueStore :: STM QueueStore
 newQueueStore = do
+  recipientKeys <- newTVar mempty
+  senderKeys <- newTVar mempty
+  notifierKeys <- newTVar mempty
   queues <- TM.empty
   senders <- TM.empty
   notifiers <- TM.empty
-  pure QueueStore {queues, senders, notifiers}
+  pure QueueStore {recipientKeys, senderKeys, notifierKeys, queues, senders, notifiers}
 
 addQueue :: QueueStore -> QueueRec -> STM (Either ErrorType ())
-addQueue QueueStore {queues, senders} q@QueueRec {recipientId = rId, senderId = sId} = do
+addQueue QueueStore {recipientKeys, senderKeys, queues, senders} q@QueueRec {recipientId = rId, recipientKey, senderId = sId, senderKey} = do
   ifM hasId (pure $ Left DUPLICATE_) $ do
     qVar <- newTVar q
     TM.insert rId qVar queues
+    modifyTVar' recipientKeys $ HM.insert rId recipientKey
     TM.insert sId rId senders
+    forM_ senderKey $ modifyTVar' senderKeys . HM.insert rId -- XXX: should not exist here yet, unless testing?
     pure $ Right ()
   where
     hasId = (||) <$> TM.member rId queues <*> TM.member sId senders
@@ -64,13 +74,14 @@ getQueue QueueStore {queues, senders, notifiers} party qId =
       SNotifier -> TM.lookup qId notifiers $>>= (`TM.lookup` queues)
 
 secureQueue :: QueueStore -> RecipientId -> SndPublicAuthKey -> STM (Either ErrorType QueueRec)
-secureQueue QueueStore {queues} rId sKey =
+secureQueue QueueStore {senderKeys, queues} rId sKey =
   withQueue rId queues $ \qVar ->
     readTVar qVar >>= \q -> case senderKey q of
       Just k -> pure $ if sKey == k then Just q else Nothing
-      _ ->
+      _ -> do
         let q' = q {senderKey = Just sKey}
-         in writeTVar qVar q' $> Just q'
+        modifyTVar' senderKeys $ HM.insert rId sKey
+        writeTVar qVar q' $> Just q'
 
 addQueueNotifier :: QueueStore -> RecipientId -> NtfCreds -> STM (Either ErrorType QueueRec)
 addQueueNotifier QueueStore {queues, notifiers} rId ntfCreds@NtfCreds {notifierId = nId} = do

src/Simplex/Messaging/Transport.hs

@@ -33,9 +33,12 @@ module Simplex.Messaging.Transport
     VersionSMP,
     VersionRangeSMP,
     THandleSMP,
+    supportedSMPHandshakes,
     supportedClientSMPRelayVRange,
     supportedServerSMPRelayVRange,
+    legacyServerSMPRelayVRange,
     currentClientSMPRelayVersion,
+    legacyServerSMPRelayVersion,
     currentServerSMPRelayVersion,
     batchCmdsSMPVersion,
     basicAuthSMPVersion,
@@ -150,7 +153,10 @@ authCmdsSMPVersion :: VersionSMP
 authCmdsSMPVersion = VersionSMP 7
 
 currentClientSMPRelayVersion :: VersionSMP
-currentClientSMPRelayVersion = VersionSMP 6
+currentClientSMPRelayVersion = VersionSMP 7
+
+legacyServerSMPRelayVersion :: VersionSMP
+legacyServerSMPRelayVersion = VersionSMP 6
 
 currentServerSMPRelayVersion :: VersionSMP
 currentServerSMPRelayVersion = VersionSMP 7
@@ -160,9 +166,15 @@ currentServerSMPRelayVersion = VersionSMP 7
 supportedClientSMPRelayVRange :: VersionRangeSMP
 supportedClientSMPRelayVRange = mkVersionRange batchCmdsSMPVersion currentClientSMPRelayVersion
 
+legacyServerSMPRelayVRange :: VersionRangeSMP
+legacyServerSMPRelayVRange = mkVersionRange batchCmdsSMPVersion legacyServerSMPRelayVersion
+
 supportedServerSMPRelayVRange :: VersionRangeSMP
 supportedServerSMPRelayVRange = mkVersionRange batchCmdsSMPVersion currentServerSMPRelayVersion
 
+supportedSMPHandshakes :: [ALPN]
+supportedSMPHandshakes = ["smp/1"]
+
 simplexMQVersion :: String
 simplexMQVersion = showVersion SMQ.version
 
@@ -194,6 +206,9 @@ class Transport c where
   -- | tls-unique channel binding per RFC5929
   tlsUnique :: c -> SessionId
 
+  -- | ALPN value negotiated for the session
+  getSessionALPN :: c -> Maybe ALPN
+
   -- | Close connection
   closeConnection :: c -> IO ()
 
@@ -288,6 +303,7 @@ instance Transport TLS where
   getServerConnection = getTLS TServer
   getClientConnection = getTLS TClient
   getServerCerts = tlsServerCerts
+  getSessionALPN = tlsALPN
   tlsUnique = tlsUniq
   closeConnection tls = closeTLS $ tlsContext tls
 
@@ -468,12 +484,13 @@ smpServerHandshake serverSignKey c (k, pk) kh smpVRange = do
   let th@THandle {params = THandleParams {sessionId}} = smpTHandle c
       sk = C.signX509 serverSignKey $ C.publicToX509 k
       certChain = getServerCerts c
-  sendHandshake th $ ServerHandshake {sessionId, smpVersionRange = smpVRange, authPubKey = Just (certChain, sk)}
+      smpVersionRange = maybe legacyServerSMPRelayVRange (const smpVRange) $ getSessionALPN c
+  sendHandshake th $ ServerHandshake {sessionId, smpVersionRange, authPubKey = Just (certChain, sk)}
   getHandshake th >>= \case
     ClientHandshake {smpVersion = v, keyHash, authPubKey = k'}
       | keyHash /= kh ->
           throwE $ TEHandshake IDENTITY
-      | v `isCompatible` smpVRange ->
+      | v `isCompatible` smpVersionRange ->
           pure $ smpThHandleServer th v pk k'
       | otherwise -> throwE $ TEHandshake VERSION
 

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

@@ -51,7 +51,7 @@ data HTTP2Server = HTTP2Server
 -- This server is for testing only, it processes all requests in a single queue.
 getHTTP2Server :: HTTP2ServerConfig -> IO HTTP2Server
 getHTTP2Server HTTP2ServerConfig {qSize, http2Port, bufferSize, bodyHeadSize, serverSupported, caCertificateFile, certificateFile, privateKeyFile, transportConfig} = do
-  tlsServerParams <- loadSupportedTLSServerParams serverSupported caCertificateFile certificateFile privateKeyFile
+  tlsServerParams <- loadSupportedTLSServerParams serverSupported caCertificateFile certificateFile privateKeyFile (alpn transportConfig)
   started <- newEmptyTMVarIO
   reqQ <- newTBQueueIO qSize
   action <- async $

src/Simplex/Messaging/Transport/Server.hs

@@ -28,10 +28,10 @@ import Control.Logger.Simple
 import Control.Monad
 import qualified Crypto.Store.X509 as SX
 import Data.Default (def)
-import Data.List (find)
 import Data.IntMap.Strict (IntMap)
 import qualified Data.IntMap.Strict as IM
-import Data.Maybe (fromJust)
+import Data.List (find)
+import Data.Maybe (fromJust, fromMaybe)
 import qualified Data.X509 as X
 import Data.X509.Validation (Fingerprint (..))
 import qualified Data.X509.Validation as XV
@@ -152,12 +152,13 @@ startTCPServer started port = withSocketsDo $ resolve >>= open >>= setStarted
       pure sock
     setStarted sock = atomically (tryPutTMVar started True) >> pure sock
 
-loadTLSServerParams :: FilePath -> FilePath -> FilePath -> IO T.ServerParams
+loadTLSServerParams :: FilePath -> FilePath -> FilePath -> Maybe [ALPN] -> IO T.ServerParams
 loadTLSServerParams = loadSupportedTLSServerParams supportedParameters
 
-loadSupportedTLSServerParams :: T.Supported -> FilePath -> FilePath -> FilePath -> IO T.ServerParams
-loadSupportedTLSServerParams serverSupported caCertificateFile certificateFile privateKeyFile =
-  fromCredential <$> loadServerCredential
+loadSupportedTLSServerParams :: T.Supported -> FilePath -> FilePath -> FilePath -> Maybe [ALPN] -> IO T.ServerParams
+loadSupportedTLSServerParams serverSupported caCertificateFile certificateFile privateKeyFile alpn_ = do
+  tlsServerParams <- fromCredential <$> loadServerCredential
+  pure tlsServerParams {T.serverHooks = maybe def alpnHooks alpn_}
   where
     loadServerCredential :: IO T.Credential
     loadServerCredential =
@@ -172,6 +173,7 @@ loadSupportedTLSServerParams serverSupported caCertificateFile certificateFile p
           T.serverHooks = def,
           T.serverSupported = serverSupported
         }
+    alpnHooks supported = def {T.onALPNClientSuggest = Just $ pure . fromMaybe "" . find (`elem` supported)}
 
 loadFingerprint :: FilePath -> IO Fingerprint
 loadFingerprint certificateFile = do

src/Simplex/Messaging/Transport/WebSockets.hs

@@ -14,7 +14,8 @@ import Network.WebSockets
 import Network.WebSockets.Stream (Stream)
 import qualified Network.WebSockets.Stream as S
 import Simplex.Messaging.Transport
-  ( TProxy,
+  ( ALPN,
+    TProxy,
     Transport (..),
     TransportConfig (..),
     TransportError (..),
@@ -28,6 +29,7 @@ import Simplex.Messaging.Transport.Buffer (trimCR)
 data WS = WS
   { wsPeer :: TransportPeer,
     tlsUniq :: ByteString,
+    wsALPN :: Maybe ALPN,
     wsStream :: Stream,
     wsConnection :: Connection,
     wsTransportConfig :: TransportConfig,
@@ -61,6 +63,9 @@ instance Transport WS where
   getServerCerts :: WS -> X.CertificateChain
   getServerCerts = wsServerCerts
 
+  getSessionALPN :: WS -> Maybe ALPN
+  getSessionALPN = wsALPN
+
   tlsUnique :: WS -> ByteString
   tlsUnique = tlsUniq
 
@@ -90,7 +95,8 @@ getWS wsPeer cfg wsServerCerts cxt = withTlsUnique wsPeer cxt connectWS
     connectWS tlsUniq = do
       s <- makeTLSContextStream cxt
       wsConnection <- connectPeer wsPeer s
-      pure $ WS {wsPeer, tlsUniq, wsStream = s, wsConnection, wsTransportConfig = cfg, wsServerCerts}
+      wsALPN <- T.getNegotiatedProtocol cxt
+      pure $ WS {wsPeer, tlsUniq, wsALPN, wsStream = s, wsConnection, wsTransportConfig = cfg, wsServerCerts}
     connectPeer :: TransportPeer -> Stream -> IO Connection
     connectPeer TServer = acceptClientRequest
     connectPeer TClient = sendClientRequest

tests/AgentTests.hs

@@ -27,9 +27,9 @@ import GHC.Stack (withFrozenCallStack)
 import Network.HTTP.Types (urlEncode)
 import SMPAgentClient
 import SMPClient (testKeyHash, testPort, testPort2, testStoreLogFile, withSmpServer, withSmpServerStoreLogOn)
-import Simplex.Messaging.Agent.Protocol hiding (MID, CONF, INFO, REQ)
+import Simplex.Messaging.Agent.Protocol hiding (CONF, INFO, MID, REQ)
 import qualified Simplex.Messaging.Agent.Protocol as A
-import Simplex.Messaging.Crypto.Ratchet (InitialKeys (..), PQEncryption (..), PQSupport (..), pattern IKPQOn, pattern IKPQOff, pattern PQEncOn, pattern PQSupportOn, pattern PQSupportOff)
+import Simplex.Messaging.Crypto.Ratchet (InitialKeys (..), PQEncryption (..), PQSupport (..), pattern IKPQOff, pattern IKPQOn, pattern PQEncOn, pattern PQSupportOff, pattern PQSupportOn)
 import qualified Simplex.Messaging.Crypto.Ratchet as CR
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (ErrorType (..))
@@ -547,10 +547,10 @@ testResumeDeliveryQuotaExceeded _ alice bob = do
   bob <#= \case ("", "alice", Msg "message 4") -> True; _ -> False
   bob #: ("4", "alice", "ACK 7") #> ("4", "alice", OK)
   inAnyOrder
-      (tGetAgent alice)
-      [ \case ("", c, Right (SENT 8)) -> c == "bob"; _ -> False,
-        \case ("", c, Right QCONT) -> c == "bob"; _ -> False
-      ]
+    (tGetAgent alice)
+    [ \case ("", c, Right (SENT 8)) -> c == "bob"; _ -> False,
+      \case ("", c, Right QCONT) -> c == "bob"; _ -> False
+    ]
   bob <#= \case ("", "alice", Msg "over quota") -> True; _ -> False
   -- message 8 is skipped because of alice agent sending "QCONT" message
   bob #: ("5", "alice", "ACK 9") #> ("5", "alice", OK)
@@ -580,7 +580,7 @@ enableKEMStr _ = ""
 
 pqConnModeStr :: InitialKeys -> ByteString
 pqConnModeStr (IKNoPQ PQSupportOff) = ""
-pqConnModeStr pq =  " " <> strEncode pq
+pqConnModeStr pq = " " <> strEncode pq
 
 sendMessage :: Transport c => (c, ConnId) -> (c, ConnId) -> ByteString -> IO ()
 sendMessage (h1, name1) (h2, name2) msg = do

tests/AgentTests/FunctionalAPITests.hs

@@ -58,10 +58,10 @@ import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Either (isRight)
 import Data.Int (Int64)
-import Data.List (nub)
+import Data.List (find, nub)
 import Data.List.NonEmpty (NonEmpty)
 import qualified Data.Map as M
-import Data.Maybe (isNothing)
+import Data.Maybe (isJust, isNothing)
 import qualified Data.Set as S
 import Data.Time.Clock (diffUTCTime, getCurrentTime)
 import Data.Time.Clock.System (SystemTime (..), getSystemTime)
@@ -80,7 +80,7 @@ import qualified Simplex.Messaging.Agent.Protocol as A
 import Simplex.Messaging.Agent.RetryInterval (RetryInterval (..))
 import Simplex.Messaging.Agent.Store.SQLite (MigrationConfirmation (..), SQLiteStore (dbNew))
 import Simplex.Messaging.Agent.Store.SQLite.Common (withTransaction')
-import Simplex.Messaging.Client (NetworkConfig (..), ProtocolClientConfig (..), TransportSessionMode (TSMEntity, TSMUser), defaultSMPClientConfig)
+import Simplex.Messaging.Client (NetworkConfig (..), ProtocolClientConfig (..), TransportSessionMode (TSMEntity, TSMUser), defaultClientConfig)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.Ratchet (InitialKeys (..), PQEncryption (..), PQSupport (..), pattern PQEncOff, pattern PQEncOn, pattern PQSupportOff, pattern PQSupportOn)
 import qualified Simplex.Messaging.Crypto.Ratchet as CR
@@ -90,7 +90,7 @@ import Simplex.Messaging.Protocol (BasicAuth, ErrorType (..), MsgBody, ProtocolS
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
 import Simplex.Messaging.Server.Expiration
-import Simplex.Messaging.Transport (ATransport (..), SMPVersion, VersionSMP, authCmdsSMPVersion, basicAuthSMPVersion, batchCmdsSMPVersion, currentServerSMPRelayVersion)
+import Simplex.Messaging.Transport (ATransport (..), SMPVersion, VersionSMP, authCmdsSMPVersion, basicAuthSMPVersion, batchCmdsSMPVersion, currentServerSMPRelayVersion, supportedSMPHandshakes)
 import Simplex.Messaging.Util (diffToMicroseconds)
 import Simplex.Messaging.Version (VersionRange (..))
 import qualified Simplex.Messaging.Version as V
@@ -175,13 +175,16 @@ pattern Rcvd :: AgentMsgId -> ACommand 'Agent 'AEConn
 pattern Rcvd agentMsgId <- RCVD MsgMeta {integrity = MsgOk} [MsgReceipt {agentMsgId, msgRcptStatus = MROk}]
 
 smpCfgVPrev :: ProtocolClientConfig SMPVersion
-smpCfgVPrev = (smpCfg agentCfg) {serverVRange = prevRange $ serverVRange $ smpCfg agentCfg}
+smpCfgVPrev = (smpCfg agentCfg) {clientALPN = Nothing, serverVRange = prevRange $ serverVRange $ smpCfg agentCfg}
 
 smpCfgV7 :: ProtocolClientConfig SMPVersion
 smpCfgV7 = (smpCfg agentCfg) {serverVRange = V.mkVersionRange batchCmdsSMPVersion authCmdsSMPVersion}
 
+ntfCfgVPrev :: ProtocolClientConfig NTFVersion
+ntfCfgVPrev = (ntfCfg agentCfg) {clientALPN = Nothing, serverVRange = V.mkVersionRange (VersionNTF 1) (VersionNTF 1)}
+
 ntfCfgV2 :: ProtocolClientConfig NTFVersion
-ntfCfgV2 = (smpCfg agentCfg) {serverVRange = V.mkVersionRange (VersionNTF 1) authBatchCmdsNTFVersion}
+ntfCfgV2 = (ntfCfg agentCfg) {serverVRange = V.mkVersionRange (VersionNTF 1) authBatchCmdsNTFVersion}
 
 agentCfgVPrev :: AgentConfig
 agentCfgVPrev =
@@ -190,7 +193,8 @@ agentCfgVPrev =
       smpAgentVRange = prevRange $ smpAgentVRange agentCfg,
       smpClientVRange = prevRange $ smpClientVRange agentCfg,
       e2eEncryptVRange = prevRange $ e2eEncryptVRange agentCfg,
-      smpCfg = smpCfgVPrev
+      smpCfg = smpCfgVPrev,
+      ntfCfg = ntfCfgVPrev
     }
 
 -- agent config for the next client version
@@ -333,6 +337,9 @@ functionalAPITests t = do
     skip "faster version of the previous test (200 subscriptions gets very slow with test coverage)" $
       it "should subscribe to multiple (6) subscriptions with batching" $
         testBatchedSubscriptions 6 3 t
+    it "should subscribe to multiple connections with pending messages" $
+      withSmpServer t $
+        testBatchedPendingMessages 10 5
   describe "Async agent commands" $ do
     it "should connect using async agent commands" $
       withSmpServer t testAsyncCommands
@@ -1530,7 +1537,7 @@ testBatchedSubscriptions :: Int -> Int -> ATransport -> IO ()
 testBatchedSubscriptions nCreate nDel t =
   withAgentClientsCfgServers2 agentCfg agentCfg initAgentServers2 $ \a b -> do
     conns <- runServers $ do
-      conns <- replicateM (nCreate :: Int) $ makeConnection_ PQSupportOff a b
+      conns <- replicateM nCreate $ makeConnection_ PQSupportOff a b
       forM_ conns $ \(aId, bId) -> exchangeGreetings_ PQEncOff a bId b aId
       let (aIds', bIds') = unzip $ take nDel conns
       delete a bIds'
@@ -1589,6 +1596,25 @@ testBatchedSubscriptions nCreate nDel t =
         killThread t1
         pure res
 
+testBatchedPendingMessages :: Int -> Int -> IO ()
+testBatchedPendingMessages nCreate nMsgs =
+  withA $ \a -> do
+    conns <- withB $ \b -> runRight $ do
+      replicateM nCreate $ makeConnection a b
+    let msgConns = take nMsgs conns
+    runRight_ $ forM_ msgConns $ \(_, bId) -> sendMessage a bId SMP.noMsgFlags "hello"
+    replicateM_ nMsgs $ get a =##> \case ("", cId, SENT _) -> isJust $ find ((cId ==) . snd) msgConns; _ -> False
+    withB $ \b -> runRight_ $ do
+      r <- subscribeConnections b $ map fst conns
+      liftIO $ all isRight r `shouldBe` True
+      replicateM_ nMsgs $ do
+        ("", cId, Msg' msgId _ "hello") <- get b
+        liftIO $ isJust (find ((cId ==) . fst) msgConns) `shouldBe` True
+        ackMessage b cId msgId Nothing
+  where
+    withA = withAgent 1 agentCfg initAgentServers testDB
+    withB = withAgent 2 agentCfg initAgentServers testDB2
+
 testAsyncCommands :: IO ()
 testAsyncCommands =
   withAgentClients2 $ \alice bob -> runRight_ $ do
@@ -2416,7 +2442,8 @@ testCreateQueueAuth srvVersion clnt1 clnt2 = do
   where
     getClient clientId (clntAuth, clntVersion) db =
       let servers = initAgentServers {smp = userServers [ProtoServerWithAuth testSMPServer clntAuth]}
-          smpCfg = (defaultSMPClientConfig :: ProtocolClientConfig SMPVersion) {serverVRange = V.mkVersionRange (prevVersion basicAuthSMPVersion) clntVersion}
+          alpn_ = if clntVersion >= authCmdsSMPVersion then Just supportedSMPHandshakes else Nothing
+          smpCfg = defaultClientConfig alpn_ $ V.mkVersionRange (prevVersion basicAuthSMPVersion) clntVersion
           sndAuthAlg = if srvVersion >= authCmdsSMPVersion && clntVersion >= authCmdsSMPVersion then C.AuthAlg C.SX25519 else C.AuthAlg C.SEd25519
        in getSMPAgentClient' clientId agentCfg {smpCfg, sndAuthAlg} servers db
 

tests/NtfClient.hs

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

tests/SMPClient.hs

@@ -24,8 +24,10 @@ import Simplex.Messaging.Server (runSMPServerBlocking)
 import Simplex.Messaging.Server.Env.STM
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Client
+import qualified Simplex.Messaging.Transport.Client as Client
 import Simplex.Messaging.Transport.Server
-import Simplex.Messaging.Version (mkVersionRange)
+import qualified Simplex.Messaging.Transport.Server as Server
+import Simplex.Messaging.Version
 import System.Environment (lookupEnv)
 import System.Info (os)
 import Test.Hspec
@@ -73,10 +75,15 @@ testSMPClient = testSMPClientVR supportedClientSMPRelayVRange
 testSMPClientVR :: Transport c => VersionRangeSMP -> (THandleSMP c 'TClient -> IO a) -> IO a
 testSMPClientVR vr client = do
   Right useHost <- pure $ chooseTransportHost defaultNetworkConfig testHost
-  runTransportClient defaultTransportClientConfig Nothing useHost testPort (Just testKeyHash) $ \h ->
+  let tcConfig = defaultTransportClientConfig {Client.alpn = clientALPN}
+  runTransportClient tcConfig Nothing useHost testPort (Just testKeyHash) $ \h ->
     runExceptT (smpClientHandshake h Nothing testKeyHash vr) >>= \case
       Right th -> client th
       Left e -> error $ show e
+  where
+    clientALPN
+      | authCmdsSMPVersion `isCompatible` vr = Just supportedSMPHandshakes
+      | otherwise = Nothing
 
 cfg :: ServerConfig
 cfg =
@@ -104,7 +111,7 @@ cfg =
       privateKeyFile = "tests/fixtures/server.key",
       certificateFile = "tests/fixtures/server.crt",
       smpServerVRange = supportedServerSMPRelayVRange,
-      transportConfig = defaultTransportServerConfig,
+      transportConfig = defaultTransportServerConfig {Server.alpn = Just supportedSMPHandshakes},
       controlPort = Nothing
     }
 

tests/ServerTests.hs

@@ -62,7 +62,7 @@ serverTests t@(ATransport t') = do
   describe "Store log" $ testWithStoreLog t
   describe "Restore messages" $ testRestoreMessages t
   describe "Restore messages (old / v2)" $ testRestoreExpireMessages t
-  describe "Timing of AUTH error" $ testTiming t
+  fdescribe "Timing of AUTH error" $ testTiming t
   describe "Message notifications" $ testMessageNotifications t
   describe "Message expiration" $ do
     testMsgExpireOnSend t'