5.7.4.1

* agent: remove external timeout to resubscribe

* liftIO

* fix tests

CHANGELOG.md

@@ -1,3 +1,25 @@
+# 5.7.4
+
+SMP agent:
+- remove re-subscription timeouts (as they are tracked per operation, and could cause failed subscriptions).
+- reconnect XFTP clients when network settings changes.
+- fix lock contention resulting in stuck subscriptions on network change.
+
+# 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.1
 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.2.0
+version:        5.7.4.1
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and

src/Simplex/FileTransfer/Agent.hs

@@ -180,7 +180,7 @@ runXFTPRcvWorker c srv Worker {doWork} = do
         fc@RcvFileChunk {userId, rcvFileId, rcvFileEntityId, digest, fileTmpPath, replicas = replica@RcvFileChunkReplica {rcvChunkReplicaId, server, delay} : _} -> do
           let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay
           withRetryIntervalLimit xftpConsecutiveRetries ri' $ \delay' loop -> do
-            lift $ waitForUserNetwork c
+            liftIO $ waitForUserNetwork c
             downloadFileChunk fc replica
               `catchAgentError` \e -> retryOnError "XFTP rcv worker" (retryLoop loop e delay') (retryDone e) e
           where
@@ -424,7 +424,7 @@ runXFTPSndPrepareWorker c Worker {doWork} = do
             tryCreate = do
               usedSrvs <- newTVarIO ([] :: [XFTPServer])
               withRetryInterval (riFast ri) $ \_ loop -> do
-                lift $ waitForUserNetwork c
+                liftIO $ waitForUserNetwork c
                 createWithNextSrv usedSrvs
                   `catchAgentError` \e -> retryOnError "XFTP prepare worker" (retryLoop loop) (throwError e) e
               where
@@ -457,7 +457,7 @@ runXFTPSndWorker c srv Worker {doWork} = do
         fc@SndFileChunk {userId, sndFileId, sndFileEntityId, filePrefixPath, digest, replicas = replica@SndFileChunkReplica {sndChunkReplicaId, server, delay} : _} -> do
           let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay
           withRetryIntervalLimit xftpConsecutiveRetries ri' $ \delay' loop -> do
-            lift $ waitForUserNetwork c
+            liftIO $ waitForUserNetwork c
             uploadFileChunk cfg fc replica
               `catchAgentError` \e -> retryOnError "XFTP snd worker" (retryLoop loop e delay') (retryDone e) e
           where
@@ -624,7 +624,7 @@ runXFTPDelWorker c srv Worker {doWork} = do
         processDeletedReplica replica@DeletedSndChunkReplica {deletedSndChunkReplicaId, userId, server, chunkDigest, delay} = do
           let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay
           withRetryIntervalLimit xftpConsecutiveRetries ri' $ \delay' loop -> do
-            lift $ waitForUserNetwork c
+            liftIO $ waitForUserNetwork c
             deleteChunkReplica
               `catchAgentError` \e -> retryOnError "XFTP del worker" (retryLoop loop e delay') (retryDone e) e
           where

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.hs

@@ -429,22 +429,21 @@ getNetworkConfig = fmap snd . readTVarIO . useNetworkConfig
 {-# INLINE getNetworkConfig #-}
 
 setUserNetworkInfo :: AgentClient -> UserNetworkInfo -> IO ()
-setUserNetworkInfo c@AgentClient {userNetworkState} UserNetworkInfo {networkType = nt', online} = withAgentEnv' c $ do
-  d <- asks $ initialInterval . userNetworkInterval . config
-  ts <- liftIO getCurrentTime
-  atomically $ do
-    ns@UserNetworkState {networkType = nt, offline} <- readTVar userNetworkState
-    when (nt' /= nt || online /= isNothing offline) $
-      writeTVar userNetworkState $!
-        let offline'
-              | nt' /= UNNone && online = Nothing
-              | isJust offline = offline
-              | otherwise = Just UNSOffline {offlineDelay = d, offlineFrom = ts}
-         in ns {networkType = nt', offline = offline'}
+setUserNetworkInfo c@AgentClient {userNetworkInfo, userNetworkUpdated} ni = withAgentEnv' c $ do
+  ts' <- liftIO getCurrentTime
+  i <- asks $ userOfflineDelay . config
+  -- if network offline event happens in less than `userOfflineDelay` after the previous event, it is ignored
+  atomically . whenM ((isOnline ni ||) <$> notRecentlyChanged ts' i) $ do
+    writeTVar userNetworkInfo ni
+    writeTVar userNetworkUpdated $ Just ts'
+  where
+    notRecentlyChanged ts' i =
+      maybe True (\ts -> diffUTCTime ts' ts > i) <$> readTVar userNetworkUpdated
 
 reconnectAllServers :: AgentClient -> IO ()
 reconnectAllServers c = do
   reconnectServerClients c smpClients
+  reconnectServerClients c xftpClients
   reconnectServerClients c ntfClients
 
 -- | Register device notifications token
@@ -1317,7 +1316,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} ConnData {connId} sq (Worker {doWork
         let mId = unId msgId
             ri' = maybe id updateRetryInterval2 msgRetryState ri
         withRetryLock2 ri' qLock $ \riState loop -> do
-          lift $ waitForUserNetwork c
+          liftIO $ waitForUserNetwork c
           resp <- tryError $ case msgType of
             AM_CONN_INFO -> sendConfirmation c sq msgBody
             AM_CONN_INFO_REPLY -> sendConfirmation c sq msgBody

src/Simplex/Messaging/Agent/Client.hs

@@ -103,9 +103,9 @@ module Simplex.Messaging.Agent.Client
     waitUntilActive,
     UserNetworkInfo (..),
     UserNetworkType (..),
-    UserNetworkState (..),
-    UNSOffline (..),
     waitForUserNetwork,
+    isNetworkOnline,
+    isOnline,
     throwWhenInactive,
     throwWhenNoDelivery,
     beginAgentOperation,
@@ -154,7 +154,6 @@ import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Either (lefts, partitionEithers)
 import Data.Functor (($>))
-import Data.Int (Int64)
 import Data.List (deleteFirstsBy, foldl', partition, (\\))
 import Data.List.NonEmpty (NonEmpty (..), (<|))
 import qualified Data.List.NonEmpty as L
@@ -165,7 +164,7 @@ import Data.Set (Set)
 import qualified Data.Set as S
 import Data.Text (Text)
 import Data.Text.Encoding
-import Data.Time (UTCTime, defaultTimeLocale, diffUTCTime, formatTime, getCurrentTime)
+import Data.Time (UTCTime, defaultTimeLocale, formatTime, getCurrentTime)
 import Data.Time.Clock.System (getSystemTime)
 import Data.Word (Word16)
 import Network.Socket (HostName)
@@ -268,7 +267,8 @@ data AgentClient = AgentClient
     xftpServers :: TMap UserId (NonEmpty XFTPServerWithAuth),
     xftpClients :: TMap XFTPTransportSession XFTPClientVar,
     useNetworkConfig :: TVar (NetworkConfig, NetworkConfig), -- (slow, fast) networks
-    userNetworkState :: TVar UserNetworkState,
+    userNetworkInfo :: TVar UserNetworkInfo,
+    userNetworkUpdated :: TVar (Maybe UTCTime),
     subscrConns :: TVar (Set ConnId),
     activeSubs :: TRcvQueues,
     pendingSubs :: TRcvQueues,
@@ -405,22 +405,20 @@ data UserNetworkInfo = UserNetworkInfo
   }
   deriving (Show)
 
+isNetworkOnline :: AgentClient -> STM Bool
+isNetworkOnline c = isOnline <$> readTVar (userNetworkInfo c)
+
+isOnline :: UserNetworkInfo -> Bool
+isOnline UserNetworkInfo {networkType, online} = networkType /= UNNone && online
+
 data UserNetworkType = UNNone | UNCellular | UNWifi | UNEthernet | UNOther
   deriving (Eq, Show)
 
-data UserNetworkState = UserNetworkState
-  { networkType :: UserNetworkType,
-    offline :: Maybe UNSOffline
-  }
-  deriving (Show)
-
-data UNSOffline = UNSOffline {offlineDelay :: Int64, offlineFrom :: UTCTime}
-  deriving (Show)
-
 -- | Creates an SMP agent client instance that receives commands and sends responses via 'TBQueue's.
 newAgentClient :: Int -> InitialAgentServers -> Env -> STM AgentClient
 newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv = do
-  let qSize = tbqSize $ config agentEnv
+  let cfg = config agentEnv
+      qSize = tbqSize cfg
   acThread <- newTVar Nothing
   active <- newTVar True
   rcvQ <- newTBQueue qSize
@@ -433,7 +431,8 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv =
   xftpServers <- newTVar xftp
   xftpClients <- TM.empty
   useNetworkConfig <- newTVar (slowNetworkConfig netCfg, netCfg)
-  userNetworkState <- newTVar $ UserNetworkState UNOther Nothing
+  userNetworkInfo <- newTVar $ UserNetworkInfo UNOther True
+  userNetworkUpdated <- newTVar Nothing
   subscrConns <- newTVar S.empty
   activeSubs <- RQ.empty
   pendingSubs <- RQ.empty
@@ -468,7 +467,8 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv =
         xftpServers,
         xftpClients,
         useNetworkConfig,
-        userNetworkState,
+        userNetworkInfo,
+        userNetworkUpdated,
         subscrConns,
         activeSubs,
         pendingSubs,
@@ -608,12 +608,11 @@ resubscribeSMPSession c@AgentClient {smpSubWorkers, workerSeq} tSess =
       atomically $ putTMVar (sessionVar v) a
     runSubWorker = do
       ri <- asks $ reconnectInterval . config
-      timeoutCounts <- newTVarIO 0
       withRetryInterval ri $ \_ loop -> do
         pending <- atomically getPending
         forM_ (L.nonEmpty pending) $ \qs -> do
-          waitForUserNetwork c
-          void . tryAgentError' $ reconnectSMPClient timeoutCounts c tSess qs
+          liftIO $ waitForUserNetwork c
+          reconnectSMPClient c tSess qs
           loop
     getPending = RQ.getSessQueues tSess $ pendingSubs c
     cleanup :: SessionVar (Async ()) -> STM ()
@@ -623,35 +622,23 @@ resubscribeSMPSession c@AgentClient {smpSubWorkers, workerSeq} tSess =
       whenM (isEmptyTMVar $ sessionVar v) retry
       removeSessVar v tSess smpSubWorkers
 
-reconnectSMPClient :: TVar Int -> AgentClient -> SMPTransportSession -> NonEmpty RcvQueue -> AM ()
-reconnectSMPClient tc c tSess@(_, srv, _) qs = do
-  NetworkConfig {tcpTimeout} <- atomically $ getNetworkConfig c
-  -- this allows 3x of timeout per batch of subscription (90 queues per batch empirically)
-  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)
+reconnectSMPClient :: AgentClient -> SMPTransportSession -> NonEmpty RcvQueue -> AM' ()
+reconnectSMPClient c tSess@(_, srv, _) qs = handleNotify $ do
+  cs <- readTVarIO $ RQ.getConnections $ activeSubs c
+  rs <- subscribeQueues c $ L.toList qs
+  let (errs, okConns) = partitionEithers $ map (\(RcvQueue {connId}, r) -> bimap (connId,) (const connId) r) rs
+      conns = filter (`M.notMember` cs) okConns
+  unless (null conns) $ notifySub "" $ UP srv conns
+  let (tempErrs, finalErrs) = partition (temporaryAgentError . snd) errs
+  mapM_ (\(connId, e) -> notifySub connId $ ERR e) finalErrs
+  forM_ (listToMaybe tempErrs) $ \(connId, e) -> do
+    when (null okConns && M.null cs && null finalErrs) . liftIO $
+      closeClient c smpClients tSess
+    notifySub connId $ ERR e
   where
-    resubscribe :: AM ()
-    resubscribe = do
-      cs <- readTVarIO $ RQ.getConnections $ activeSubs c
-      rs <- lift . subscribeQueues c $ L.toList qs
-      let (errs, okConns) = partitionEithers $ map (\(RcvQueue {connId}, r) -> bimap (connId,) (const connId) r) rs
-      liftIO $ do
-        let conns = filter (`M.notMember` cs) okConns
-        unless (null conns) $ notifySub "" $ UP srv conns
-      let (tempErrs, finalErrs) = partition (temporaryAgentError . snd) errs
-      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
-        throwError err
-    notifySub :: forall e. AEntityI e => ConnId -> ACommand 'Agent e -> IO ()
+    handleNotify :: AM' () -> AM' ()
+    handleNotify = E.handleAny $ notifySub "" . ERR . INTERNAL . show
+    notifySub :: forall e. AEntityI e => ConnId -> ACommand 'Agent e -> AM' ()
     notifySub connId cmd = atomically $ writeTBQueue (subQ c) ("", connId, APC (sAEntity @e) cmd)
 
 getNtfServerClient :: AgentClient -> NtfTransportSession -> AM NtfClient
@@ -746,43 +733,17 @@ getClientConfig c cfgSel = do
 getNetworkConfig :: AgentClient -> STM NetworkConfig
 getNetworkConfig c = do
   (slowCfg, fastCfg) <- readTVar (useNetworkConfig c)
-  UserNetworkState {networkType} <- readTVar (userNetworkState c)
+  UserNetworkInfo {networkType} <- readTVar $ userNetworkInfo c
   pure $ case networkType of
     UNCellular -> slowCfg
     UNNone -> slowCfg
     _ -> fastCfg
 
-waitForUserNetwork :: AgentClient -> AM' ()
-waitForUserNetwork AgentClient {userNetworkState} =
-  readTVarIO userNetworkState >>= mapM_ waitWhileOffline . offline
-  where
-    waitWhileOffline UNSOffline {offlineDelay = d} =
-      unlessM (liftIO $ waitOnline d False) $ do
-        -- network delay reached, increase delay
-        ts' <- liftIO getCurrentTime
-        ni <- asks $ userNetworkInterval . config
-        atomically $ do
-          ns@UserNetworkState {offline} <- readTVar userNetworkState
-          forM_ offline $ \UNSOffline {offlineDelay = d', offlineFrom = ts} ->
-            -- Using `min` to avoid multiple updates in a short period of time
-            -- and to reset `offlineDelay` if network went `on` and `off` again.
-            writeTVar userNetworkState $!
-              let d'' = nextRetryDelay (diffToMicroseconds $ diffUTCTime ts' ts) (min d d') ni
-               in ns {offline = Just UNSOffline {offlineDelay = d'', offlineFrom = ts}}
-    waitOnline :: Int64 -> Bool -> IO Bool
-    waitOnline t online'
-      | t <= 0 = pure online'
-      | otherwise =
-          registerDelay (fromIntegral maxWait)
-            >>= atomically . onlineOrDelay
-            >>= waitOnline (t - maxWait)
-      where
-        maxWait = min t $ fromIntegral (maxBound :: Int)
-        onlineOrDelay delay = do
-          online <- isNothing . offline <$> readTVar userNetworkState
-          expired <- readTVar delay
-          unless (online || expired) retry
-          pure online
+waitForUserNetwork :: AgentClient -> IO ()
+waitForUserNetwork c =
+  unlessM (atomically $ isNetworkOnline c) $ do
+    delay <- registerDelay $ userNetworkInterval $ config $ agentEnv c
+    atomically $ unlessM (isNetworkOnline c) $ unlessM (readTVar delay) retry
 
 closeAgentClient :: AgentClient -> IO ()
 closeAgentClient c = do

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

@@ -92,7 +92,8 @@ data AgentConfig = AgentConfig
     xftpCfg :: XFTPClientConfig,
     reconnectInterval :: RetryInterval,
     messageRetryInterval :: RetryInterval2,
-    userNetworkInterval :: RetryInterval,
+    userNetworkInterval :: Int,
+    userOfflineDelay :: NominalDiffTime,
     messageTimeout :: NominalDiffTime,
     connDeleteDeliveryTimeout :: NominalDiffTime,
     helloTimeout :: NominalDiffTime,
@@ -101,7 +102,6 @@ data AgentConfig = AgentConfig
     cleanupInterval :: Int64,
     cleanupStepInterval :: Int,
     maxWorkerRestartsPerMin :: Int,
-    maxSubscriptionTimeouts :: Int,
     storedMsgDataTTL :: NominalDiffTime,
     rcvFilesTTL :: NominalDiffTime,
     sndFilesTTL :: NominalDiffTime,
@@ -147,14 +147,6 @@ defaultMessageRetryInterval =
           }
     }
 
-defaultUserNetworkInterval :: RetryInterval
-defaultUserNetworkInterval =
-  RetryInterval
-    { initialInterval = 1200_000000, -- 20 minutes
-      increaseAfter = 0,
-      maxInterval = 7200_000000 -- 2 hours
-    }
-
 defaultAgentConfig :: AgentConfig
 defaultAgentConfig =
   AgentConfig
@@ -170,7 +162,8 @@ defaultAgentConfig =
       xftpCfg = defaultXFTPClientConfig,
       reconnectInterval = defaultReconnectInterval,
       messageRetryInterval = defaultMessageRetryInterval,
-      userNetworkInterval = defaultUserNetworkInterval,
+      userNetworkInterval = 1800_000000, -- 30 minutes, should be less than Int32 max value
+      userOfflineDelay = 2, -- if network offline event happens in less than 2 seconds after it was set online, it is ignored
       messageTimeout = 2 * nominalDay,
       connDeleteDeliveryTimeout = 2 * nominalDay,
       helloTimeout = 2 * nominalDay,
@@ -179,9 +172,6 @@ defaultAgentConfig =
       cleanupInterval = 30 * 60 * 1000000, -- 30 minutes
       cleanupStepInterval = 200000, -- 200ms
       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,
       storedMsgDataTTL = 21 * nominalDay,
       rcvFilesTTL = 2 * nominalDay,
       sndFilesTTL = nominalDay,

src/Simplex/Messaging/Agent/NtfSubSupervisor.hs

@@ -161,7 +161,7 @@ runNtfWorker c srv Worker {doWork} = do
           logInfo $ "runNtfWorker, nextSub " <> tshow nextSub
           ri <- asks $ reconnectInterval . config
           withRetryInterval ri $ \_ loop -> do
-            lift $ waitForUserNetwork c
+            liftIO $ waitForUserNetwork c
             processSub nextSub
               `catchAgentError` retryOnError c "NtfWorker" loop (workerInternalError c connId . show)
     processSub :: (NtfSubscription, NtfSubNTFAction, NtfActionTs) -> AM ()
@@ -245,7 +245,7 @@ runNtfSMPWorker c srv Worker {doWork} = do
           logInfo $ "runNtfSMPWorker, nextSub " <> tshow nextSub
           ri <- asks $ reconnectInterval . config
           withRetryInterval ri $ \_ loop -> do
-            lift $ waitForUserNetwork c
+            liftIO $ waitForUserNetwork c
             processSub nextSub
               `catchAgentError` retryOnError c "NtfSMPWorker" loop (workerInternalError c connId . show)
     processSub :: (NtfSubscription, NtfSubSMPAction, NtfActionTs) -> AM ()

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

@@ -53,7 +53,8 @@ import Data.Either (fromRight, partitionEithers)
 import Data.Functor (($>))
 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 +411,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 +422,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 +460,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 +480,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
@@ -975,9 +998,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/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

src/Simplex/Messaging/Util.hs

@@ -152,12 +152,14 @@ timeoutThrow :: MonadUnliftIO m => e -> Int -> ExceptT e m a -> ExceptT e m a
 timeoutThrow e ms action = ExceptT (sequence <$> (ms `timeout` runExceptT action)) >>= maybe (throwError e) pure
 
 threadDelay' :: Int64 -> IO ()
-threadDelay' time
-  | time <= 0 = pure ()
-threadDelay' time = do
-  let maxWait = min time $ fromIntegral (maxBound :: Int)
-  threadDelay $ fromIntegral maxWait
-  when (maxWait /= time) $ threadDelay' (time - maxWait)
+threadDelay' = loop
+  where
+    loop time
+      | time <= 0 = pure ()
+      | otherwise = do
+          let maxWait = min time $ fromIntegral (maxBound :: Int)
+          threadDelay $ fromIntegral maxWait
+          loop $ time - maxWait
 
 diffToMicroseconds :: NominalDiffTime -> Int64
 diffToMicroseconds diff = fromIntegral ((truncate $ diff * 1000000) :: Integer)

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 (..))
@@ -95,23 +95,24 @@ type AEntityTransmission p e = (ACorrId, ConnId, ACommand p e)
 type AEntityTransmissionOrError p e = (ACorrId, ConnId, Either AgentErrorType (ACommand p e))
 
 tGetAgent :: Transport c => c -> IO (AEntityTransmissionOrError 'Agent 'AEConn)
-tGetAgent = tGetAgent'
+tGetAgent = tGetAgent' True
 
-tGetAgent' :: forall c e. (Transport c, AEntityI e) => c -> IO (AEntityTransmissionOrError 'Agent e)
-tGetAgent' h = do
-  (corrId, connId, cmdOrErr) <- pGetAgent h
+tGetAgent' :: forall c e. (Transport c, AEntityI e) => Bool -> c -> IO (AEntityTransmissionOrError 'Agent e)
+tGetAgent' skipErr h = do
+  (corrId, connId, cmdOrErr) <- pGetAgent skipErr h
   case cmdOrErr of
     Right (APC e cmd) -> case testEquality e (sAEntity @e) of
       Just Refl -> pure (corrId, connId, Right cmd)
       _ -> error $ "unexpected command " <> show cmd
     Left err -> pure (corrId, connId, Left err)
 
-pGetAgent :: forall c. Transport c => c -> IO (ATransmissionOrError 'Agent)
-pGetAgent h = do
+pGetAgent :: forall c. Transport c => Bool -> c -> IO (ATransmissionOrError 'Agent)
+pGetAgent skipErr h = do
   (corrId, connId, cmdOrErr) <- tGet SAgent h
   case cmdOrErr of
-    Right (APC _ CONNECT {}) -> pGetAgent h
-    Right (APC _ DISCONNECT {}) -> pGetAgent h
+    Right (APC _ CONNECT {}) -> pGetAgent skipErr h
+    Right (APC _ DISCONNECT {}) -> pGetAgent skipErr h
+    Right (APC _ (ERR (BROKER _ NETWORK))) | skipErr -> pGetAgent skipErr h
     cmd -> pure (corrId, connId, cmd)
 
 -- | receive message to handle `h`
@@ -119,15 +120,18 @@ pGetAgent h = do
 (<#:) = tGetAgent
 
 (<#:?) :: Transport c => c -> IO (ATransmissionOrError 'Agent)
-(<#:?) = pGetAgent
+(<#:?) = pGetAgent True
 
 (<#:.) :: Transport c => c -> IO (AEntityTransmissionOrError 'Agent 'AENone)
-(<#:.) = tGetAgent'
+(<#:.) = tGetAgent' True
 
 -- | send transmission `t` to handle `h` and get response
 (#:) :: Transport c => c -> (ByteString, ByteString, ByteString) -> IO (AEntityTransmissionOrError 'Agent 'AEConn)
 h #: t = tPutRaw h t >> (<#:) h
 
+(#:!) :: Transport c => c -> (ByteString, ByteString, ByteString) -> IO (AEntityTransmissionOrError 'Agent 'AEConn)
+h #:! t = tPutRaw h t >> tGetAgent' False h
+
 -- | action and expected response
 -- `h #:t #> r` is the test that sends `t` to `h` and validates that the response is `r`
 (#>) :: IO (AEntityTransmissionOrError 'Agent 'AEConn) -> AEntityTransmission 'Agent 'AEConn -> Expectation
@@ -426,8 +430,8 @@ testServerConnectionAfterError t _ = do
 
   withAgent1 $ \bob -> do
     withAgent2 $ \alice -> do
-      bob #: ("1", "alice", "SUB") =#> \("1", "alice", ERR (BROKER _ e)) -> e == NETWORK || e == TIMEOUT
-      alice #: ("1", "bob", "SUB") =#> \("1", "bob", ERR (BROKER _ e)) -> e == NETWORK || e == TIMEOUT
+      bob #:! ("1", "alice", "SUB") =#> \("1", "alice", ERR (BROKER _ e)) -> e == NETWORK || e == TIMEOUT
+      alice #:! ("1", "bob", "SUB") =#> \("1", "bob", ERR (BROKER _ e)) -> e == NETWORK || e == TIMEOUT
       withServer $ do
         alice <#=? \case ("", "bob", APC _ (SENT 4)) -> True; ("", "", APC _ (UP s ["bob"])) -> s == server; _ -> False
         alice <#=? \case ("", "bob", APC _ (SENT 4)) -> True; ("", "", APC _ (UP s ["bob"])) -> s == server; _ -> False
@@ -547,10 +551,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 +584,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

@@ -50,7 +50,7 @@ module AgentTests.FunctionalAPITests
 where
 
 import AgentTests.ConnectionRequestTests (connReqData, queueAddr, testE2ERatchetParams12)
-import Control.Concurrent (killThread, threadDelay)
+import Control.Concurrent (forkIO, killThread, threadDelay)
 import Control.Monad
 import Control.Monad.Except
 import Control.Monad.Reader
@@ -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)
@@ -77,10 +77,9 @@ import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestSte
 import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..), InitialAgentServers (..), createAgentStore)
 import Simplex.Messaging.Agent.Protocol hiding (CON, CONF, INFO, REQ)
 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 +89,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
@@ -145,6 +144,7 @@ pGet c = do
   case cmd of
     CONNECT {} -> pGet c
     DISCONNECT {} -> pGet c
+    ERR (BROKER _ NETWORK) -> pGet c
     _ -> pure t
 
 pattern CONF :: ConfirmationId -> [SMPServer] -> ConnInfo -> ACommand 'Agent e
@@ -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
@@ -428,7 +435,8 @@ functionalAPITests t = do
     it "send delivery receipts concurrently with messages" $ testDeliveryReceiptsConcurrent t
   describe "user network info" $ do
     it "should wait for user network" testWaitForUserNetwork
-    it "should not reset offline interval while offline" testDoNotResetOfflineInterval
+    it "should not reset online to offline if happens too quickly" testDoNotResetOnlineToOffline
+    it "should resume multiple threads" testResumeMultipleThreads
 
 testBasicAuth :: ATransport -> Bool -> (Maybe BasicAuth, VersionSMP) -> (Maybe BasicAuth, VersionSMP) -> (Maybe BasicAuth, VersionSMP) -> IO Int
 testBasicAuth t allowNewQueues srv@(srvAuth, srvVersion) clnt1 clnt2 = do
@@ -1530,7 +1538,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 +1597,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 +2443,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
 
@@ -2671,9 +2699,7 @@ testWaitForUserNetwork = do
   noNetworkDelay a
   setUserNetworkInfo a $ UserNetworkInfo UNNone False
   networkDelay a 100000
-  networkDelay a 150000
-  networkDelay a 200000
-  networkDelay a 200000
+  networkDelay a 100000
   setUserNetworkInfo a $ UserNetworkInfo UNCellular True
   noNetworkDelay a
   setUserNetworkInfo a $ UserNetworkInfo UNCellular False
@@ -2683,36 +2709,66 @@ testWaitForUserNetwork = do
     (networkDelay a 50000)
   noNetworkDelay a
   where
-    aCfg = agentCfg {userNetworkInterval = RetryInterval {initialInterval = 100000, increaseAfter = 0, maxInterval = 200000}}
+    aCfg = agentCfg {userNetworkInterval = 100000, userOfflineDelay = 0}
 
-testDoNotResetOfflineInterval :: IO ()
-testDoNotResetOfflineInterval = do
+testDoNotResetOnlineToOffline :: IO ()
+testDoNotResetOnlineToOffline = do
   a <- getSMPAgentClient' 1 aCfg initAgentServers testDB
   noNetworkDelay a
   setUserNetworkInfo a $ UserNetworkInfo UNWifi False
   networkDelay a 100000
-  networkDelay a 150000
-  setUserNetworkInfo a $ UserNetworkInfo UNCellular False
-  networkDelay a 200000
-  setUserNetworkInfo a $ UserNetworkInfo UNNone False
-  networkDelay a 200000
-  setUserNetworkInfo a $ UserNetworkInfo UNCellular True
+  setUserNetworkInfo a $ UserNetworkInfo UNWifi False
+  setUserNetworkInfo a $ UserNetworkInfo UNWifi True
   noNetworkDelay a
-  setUserNetworkInfo a $ UserNetworkInfo UNCellular False
+  setUserNetworkInfo a $ UserNetworkInfo UNWifi False -- ingnored
+  noNetworkDelay a
+  threadDelay 100000
+  setUserNetworkInfo a $ UserNetworkInfo UNWifi False
   networkDelay a 100000
+  setUserNetworkInfo a $ UserNetworkInfo UNNone False
+  networkDelay a 100000
+  setUserNetworkInfo a $ UserNetworkInfo UNWifi True
+  setUserNetworkInfo a $ UserNetworkInfo UNNone False -- ingnored
+  noNetworkDelay a
   where
-    aCfg = agentCfg {userNetworkInterval = RetryInterval {initialInterval = 100000, increaseAfter = 0, maxInterval = 200000}}
+    aCfg = agentCfg {userNetworkInterval = 100000, userOfflineDelay = 0.1}
+
+testResumeMultipleThreads :: IO ()
+testResumeMultipleThreads = do
+  a <- getSMPAgentClient' 1 aCfg initAgentServers testDB
+  noNetworkDelay a
+  setUserNetworkInfo a $ UserNetworkInfo UNNone False
+  vs <-
+    replicateM 50000 $ do
+      v <- newEmptyTMVarIO
+      void . forkIO $ waitNetwork a >>= atomically . putTMVar v
+      pure v
+  threadDelay 1000000
+  setUserNetworkInfo a $ UserNetworkInfo UNCellular True
+  ts <- mapM (atomically . readTMVar) vs
+  -- print $ minimum ts
+  -- print $ maximum ts
+  -- print $ sum ts `div` fromIntegral (length ts)
+  let average = sum ts `div` fromIntegral (length ts)
+  average < 3000000 `shouldBe` True
+  maximum ts < 4000000 `shouldBe` True
+  where
+    aCfg = agentCfg {userOfflineDelay = 0}
 
 noNetworkDelay :: AgentClient -> IO ()
-noNetworkDelay a = (10000 >) <$> waitNetwork a `shouldReturn` True
+noNetworkDelay a = do
+  d <- waitNetwork a
+  unless (d < 10000) $ expectationFailure $ "expected no delay, d = " <> show d
 
 networkDelay :: AgentClient -> Int64 -> IO ()
-networkDelay a d' = (\d -> d' < d && d < d' + 15000) <$> waitNetwork a `shouldReturn` True
+networkDelay a d' = do
+  d <- waitNetwork a
+  unless (d' < d && d < d' + 15000) $ expectationFailure $ "expected delay " <> show d' <> ", d = " <> show d
 
 waitNetwork :: AgentClient -> IO Int64
 waitNetwork a = do
   t <- getCurrentTime
-  waitForUserNetwork a `runReaderT` agentEnv a
+  waitForUserNetwork a
   t' <- getCurrentTime
   pure $ diffToMicroseconds $ diffUTCTime t' t
 

tests/AgentTests/NotificationTests.hs

@@ -6,8 +6,8 @@
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE PatternSynonyms #-}
 {-# LANGUAGE TypeApplications #-}
-{-# OPTIONS_GHC -fno-warn-incomplete-uni-patterns #-}
 {-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
+{-# OPTIONS_GHC -fno-warn-incomplete-uni-patterns #-}
 
 module AgentTests.NotificationTests where
 
@@ -17,10 +17,6 @@ import AgentTests.FunctionalAPITests
     createConnection,
     exchangeGreetingsMsgId,
     get,
-    withAgent,
-    withAgentClients2,
-    withAgentClientsCfgServers2,
-    withAgentClients3,
     joinConnection,
     makeConnection,
     nGet,
@@ -29,7 +25,11 @@ import AgentTests.FunctionalAPITests
     sendMessage,
     switchComplete,
     testServerMatrix2,
+    withAgent,
+    withAgentClients2,
+    withAgentClients3,
     withAgentClientsCfg2,
+    withAgentClientsCfgServers2,
     (##>),
     (=##>),
     pattern CON,
@@ -59,8 +59,8 @@ import Simplex.Messaging.Agent.Protocol hiding (CON, CONF, INFO)
 import Simplex.Messaging.Agent.Store.SQLite (getSavedNtfToken)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
-import Simplex.Messaging.Notifications.Server.Env (NtfServerConfig (..))
 import Simplex.Messaging.Notifications.Protocol
+import Simplex.Messaging.Notifications.Server.Env (NtfServerConfig (..))
 import Simplex.Messaging.Notifications.Server.Push.APNS
 import Simplex.Messaging.Notifications.Types (NtfToken (..))
 import Simplex.Messaging.Protocol (ErrorType (AUTH), MsgFlags (MsgFlags), NtfServer, ProtocolServer (..), SMPMsgMeta (..), SubscriptionMode (..))
@@ -151,7 +151,8 @@ testNtfMatrix t runTest = do
     it "next servers: SMP v7, NTF v2; curr clients: v6/v1" $ runNtfTestCfg t cfgV7 ntfServerCfgV2 agentCfg agentCfg runTest
     it "curr servers: SMP v6, NTF v1; curr clients: v6/v1" $ runNtfTestCfg t cfg ntfServerCfg agentCfg agentCfg runTest
     skip "this case cannot be supported - see RFC" $
-      it "servers: SMP v6, NTF v1; clients: v7/v2 (not supported)" $ runNtfTestCfg t cfg ntfServerCfg agentCfgV7 agentCfgV7 runTest
+      it "servers: SMP v6, NTF v1; clients: v7/v2 (not supported)" $
+        runNtfTestCfg t cfg ntfServerCfg agentCfgV7 agentCfgV7 runTest
     -- servers can be migrated in any order
     it "servers: next SMP v7, curr NTF v1; curr clients: v6/v1" $ runNtfTestCfg t cfgV7 ntfServerCfg agentCfg agentCfg runTest
     it "servers: curr SMP v6, next NTF v2; curr clients: v6/v1" $ runNtfTestCfg t cfg ntfServerCfgV2 agentCfg agentCfg runTest
@@ -258,7 +259,7 @@ testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
         atomically $ readTBQueue apnsQ
       liftIO $ sendApnsResponse APNSRespOk
       pure ntfData
-    -- the new agent is created as otherwise when running the tests in CI the old agent was keeping the connection to the server
+  -- the new agent is created as otherwise when running the tests in CI the old agent was keeping the connection to the server
   threadDelay 1000000
   withAgent 2 agentCfg initAgentServers testDB $ \a' ->
     -- server stopped before token is verified, so now the attempt to verify it will return AUTH error but re-register token,

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
     }