5.7.5.0

* agent: remove external timeout to resubscribe

* liftIO

* fix tests

CHANGELOG.md

@@ -1,3 +1,15 @@
+# 5.7.5
+
+SMP agent:
+- fail if non-unique connection IDs are passed to sendMessages (to prevent client errors and deadlocks).
+
+# 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:

package.yaml

@@ -1,5 +1,5 @@
 name: simplexmq
-version: 5.7.4.0
+version: 5.7.5.0
 synopsis: SimpleXMQ message broker
 description: |
   This package includes <./docs/Simplex-Messaging-Server.html server>,
@@ -71,7 +71,6 @@ 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.4.0
+version:        5.7.5.0
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -230,7 +230,6 @@ library
     , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
-    , unordered-containers
     , websockets ==0.12.*
     , yaml ==0.11.*
     , zstd ==0.1.3.*
@@ -306,7 +305,6 @@ 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.*
@@ -382,7 +380,6 @@ 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.*
@@ -458,7 +455,6 @@ 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.*
@@ -534,7 +530,6 @@ executable xftp
     , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
-    , unordered-containers
     , websockets ==0.12.*
     , yaml ==0.11.*
     , zstd ==0.1.3.*
@@ -610,7 +605,6 @@ 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.*
@@ -727,7 +721,6 @@ 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/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

@@ -39,7 +39,6 @@ import Simplex.Messaging.Client
     proxyUsername,
     transportClientConfig,
   )
-import Simplex.Messaging.Client.Agent ()
 import qualified Simplex.Messaging.Crypto as C
 import qualified Simplex.Messaging.Crypto.Lazy as LC
 import Simplex.Messaging.Encoding (smpDecode, smpEncode)

src/Simplex/FileTransfer/Client/Agent.hs

@@ -18,7 +18,6 @@ import Data.Text.Encoding (decodeUtf8)
 import Simplex.FileTransfer.Client
 import Simplex.Messaging.Agent.RetryInterval
 import Simplex.Messaging.Client (NetworkConfig (..), ProtocolClientError (..), temporaryClientError)
-import Simplex.Messaging.Client.Agent ()
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (ProtocolServer (..), XFTPServer)
 import Simplex.Messaging.TMap (TMap)

src/Simplex/Messaging/Agent.hs

@@ -138,6 +138,8 @@ import qualified Data.List.NonEmpty as L
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
 import Data.Maybe (catMaybes, fromMaybe, isJust, isNothing, mapMaybe)
+import Data.Set (Set)
+import qualified Data.Set as S
 import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Time.Clock
@@ -354,12 +356,12 @@ sendMessage c = withAgentEnv c .:: sendMessage' c
 type MsgReq = (ConnId, PQEncryption, MsgFlags, MsgBody)
 
 -- | Send multiple messages to different connections (SEND command)
-sendMessages :: AgentClient -> [MsgReq] -> IO [Either AgentErrorType (AgentMsgId, PQEncryption)]
-sendMessages c = withAgentEnv' c . sendMessages' c
+sendMessages :: AgentClient -> [MsgReq] -> AE [Either AgentErrorType (AgentMsgId, PQEncryption)]
+sendMessages c = withAgentEnv c . sendMessages' c
 {-# INLINE sendMessages #-}
 
-sendMessagesB :: Traversable t => AgentClient -> t (Either AgentErrorType MsgReq) -> IO (t (Either AgentErrorType (AgentMsgId, PQEncryption)))
-sendMessagesB c = withAgentEnv' c . sendMessagesB' c
+sendMessagesB :: Traversable t => AgentClient -> t (Either AgentErrorType MsgReq) -> AE (t (Either AgentErrorType (AgentMsgId, PQEncryption)))
+sendMessagesB c = withAgentEnv c . sendMessagesB' c
 {-# INLINE sendMessagesB #-}
 
 ackMessage :: AgentClient -> ConnId -> AgentMsgId -> Maybe MsgReceiptInfo -> AE ()
@@ -429,22 +431,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
@@ -1034,16 +1035,27 @@ getNotificationMessage' c nonce encNtfInfo = do
 
 -- | Send message to the connection (SEND command) in Reader monad
 sendMessage' :: AgentClient -> ConnId -> PQEncryption -> MsgFlags -> MsgBody -> AM (AgentMsgId, PQEncryption)
-sendMessage' c connId pqEnc msgFlags msg = ExceptT $ runIdentity <$> sendMessagesB' c (Identity (Right (connId, pqEnc, msgFlags, msg)))
+sendMessage' c connId pqEnc msgFlags msg = ExceptT $ runIdentity <$> sendMessagesB_ c (Identity (Right (connId, pqEnc, msgFlags, msg))) (S.singleton connId)
 {-# INLINE sendMessage' #-}
 
 -- | Send multiple messages to different connections (SEND command) in Reader monad
-sendMessages' :: AgentClient -> [MsgReq] -> AM' [Either AgentErrorType (AgentMsgId, PQEncryption)]
+sendMessages' :: AgentClient -> [MsgReq] -> AM [Either AgentErrorType (AgentMsgId, PQEncryption)]
 sendMessages' c = sendMessagesB' c . map Right
 {-# INLINE sendMessages' #-}
 
-sendMessagesB' :: forall t. Traversable t => AgentClient -> t (Either AgentErrorType MsgReq) -> AM' (t (Either AgentErrorType (AgentMsgId, PQEncryption)))
-sendMessagesB' c reqs = withConnLocks c connIds "sendMessages" $ do
+sendMessagesB' :: forall t. Traversable t => AgentClient -> t (Either AgentErrorType MsgReq) -> AM (t (Either AgentErrorType (AgentMsgId, PQEncryption)))
+sendMessagesB' c reqs = do
+  connIds <- liftEither $ foldl' addConnId (Right S.empty) reqs
+  lift $ sendMessagesB_ c reqs connIds
+  where
+    addConnId s@(Right s') (Right (connId, _, _, _))
+      | B.null connId = s
+      | connId `S.notMember` s' = Right $ S.insert connId s'
+      | otherwise = Left $ INTERNAL "sendMessages: duplicate connection ID"
+    addConnId s _ = s
+
+sendMessagesB_ :: forall t. Traversable t => AgentClient -> t (Either AgentErrorType MsgReq) -> Set ConnId -> AM' (t (Either AgentErrorType (AgentMsgId, PQEncryption)))
+sendMessagesB_ c reqs connIds = withConnLocks c connIds "sendMessages" $ do
   reqs' <- withStoreBatch c (\db -> fmap (bindRight $ \req@(connId, _, _, _) -> bimap storeError (req,) <$> getConn db connId) reqs)
   let (toEnable, reqs'') = mapAccumL prepareConn [] reqs'
   void $ withStoreBatch' c $ \db -> map (\connId -> setConnPQSupport db connId PQSupportOn) toEnable
@@ -1065,7 +1077,6 @@ sendMessagesB' c reqs = withConnLocks c connIds "sendMessages" $ do
               let cData' = cData {pqSupport = PQSupportOn} :: ConnData
                in (connId : acc, Right (cData', sqs, Just pqEnc, msgFlags, A_MSG msg))
           | otherwise = (acc, Right (cData, sqs, Just pqEnc, msgFlags, A_MSG msg))
-    connIds = map (\(connId, _, _, _) -> connId) $ rights $ toList reqs
 
 -- / async command processing v v v
 
@@ -1317,7 +1328,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)
@@ -186,7 +185,6 @@ import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
 import Simplex.Messaging.Agent.TRcvQueues (TRcvQueues (getRcvQueues))
 import qualified Simplex.Messaging.Agent.TRcvQueues as RQ
 import Simplex.Messaging.Client
-import Simplex.Messaging.Client.Agent ()
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
@@ -268,7 +266,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 +404,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 +430,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 +466,8 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv =
         xftpServers,
         xftpClients,
         useNetworkConfig,
-        userNetworkState,
+        userNetworkInfo,
+        userNetworkUpdated,
         subscrConns,
         activeSubs,
         pendingSubs,
@@ -608,12 +607,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,39 +621,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 ->
-      (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)
+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
@@ -750,43 +732,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
@@ -870,15 +826,15 @@ withInvLock' :: AgentClient -> ByteString -> String -> AM' a -> AM' a
 withInvLock' AgentClient {invLocks} = withLockMap invLocks
 {-# INLINE withInvLock' #-}
 
-withConnLocks :: AgentClient -> [ConnId] -> String -> AM' a -> AM' a
-withConnLocks AgentClient {connLocks} = withLocksMap_ connLocks . filter (not . B.null)
+withConnLocks :: AgentClient -> Set ConnId -> String -> AM' a -> AM' a
+withConnLocks AgentClient {connLocks} = withLocksMap_ connLocks
 {-# INLINE withConnLocks #-}
 
 withLockMap :: (Ord k, MonadUnliftIO m) => TMap k Lock -> k -> String -> m a -> m a
 withLockMap = withGetLock . getMapLock
 {-# INLINE withLockMap #-}
 
-withLocksMap_ :: (Ord k, MonadUnliftIO m) => TMap k Lock -> [k] -> String -> m a -> m a
+withLocksMap_ :: (Ord k, MonadUnliftIO m) => TMap k Lock -> Set k -> String -> m a -> m a
 withLocksMap_ = withGetLocks . getMapLock
 {-# INLINE withLocksMap_ #-}
 

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

@@ -54,7 +54,6 @@ import Simplex.Messaging.Agent.RetryInterval
 import Simplex.Messaging.Agent.Store.SQLite
 import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
 import Simplex.Messaging.Client
-import Simplex.Messaging.Client.Agent ()
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.Ratchet (VersionRangeE2E, supportedE2EEncryptVRange)
 import Simplex.Messaging.Notifications.Client (defaultNTFClientConfig)
@@ -92,7 +91,8 @@ data AgentConfig = AgentConfig
     xftpCfg :: XFTPClientConfig,
     reconnectInterval :: RetryInterval,
     messageRetryInterval :: RetryInterval2,
-    userNetworkInterval :: RetryInterval,
+    userNetworkInterval :: Int,
+    userOfflineDelay :: NominalDiffTime,
     messageTimeout :: NominalDiffTime,
     connDeleteDeliveryTimeout :: NominalDiffTime,
     helloTimeout :: NominalDiffTime,
@@ -101,7 +101,6 @@ data AgentConfig = AgentConfig
     cleanupInterval :: Int64,
     cleanupStepInterval :: Int,
     maxWorkerRestartsPerMin :: Int,
-    maxSubscriptionTimeouts :: Int,
     storedMsgDataTTL :: NominalDiffTime,
     rcvFilesTTL :: NominalDiffTime,
     sndFilesTTL :: NominalDiffTime,
@@ -147,14 +146,6 @@ defaultMessageRetryInterval =
           }
     }
 
-defaultUserNetworkInterval :: RetryInterval
-defaultUserNetworkInterval =
-  RetryInterval
-    { initialInterval = 1200_000000, -- 20 minutes
-      increaseAfter = 0,
-      maxInterval = 7200_000000 -- 2 hours
-    }
-
 defaultAgentConfig :: AgentConfig
 defaultAgentConfig =
   AgentConfig
@@ -170,7 +161,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 +171,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 = 5,
       storedMsgDataTTL = 21 * nominalDay,
       rcvFilesTTL = 2 * nominalDay,
       sndFilesTTL = nominalDay,

src/Simplex/Messaging/Agent/Lock.hs

@@ -12,6 +12,8 @@ import Control.Monad (void)
 import Control.Monad.Except (ExceptT (..), runExceptT)
 import Control.Monad.IO.Unlift
 import Data.Functor (($>))
+import Data.Set (Set)
+import qualified Data.Set as S
 import UnliftIO.Async (forConcurrently)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
@@ -39,13 +41,11 @@ withGetLock getLock key name a =
     (atomically . takeTMVar)
     (const a)
 
-withGetLocks :: MonadUnliftIO m => (k -> STM Lock) -> [k] -> String -> m a -> m a
+withGetLocks :: MonadUnliftIO m => (k -> STM Lock) -> Set k -> String -> m a -> m a
 withGetLocks getLock keys name = E.bracket holdLocks releaseLocks . const
   where
-    holdLocks = forConcurrently keys $ \key -> atomically $ getPutLock getLock key name
-    -- only this withGetLocks would be holding the locks,
-    -- so it's safe to combine all lock releases into one transaction
-    releaseLocks = atomically . mapM_ takeTMVar
+    holdLocks = forConcurrently (S.toList keys) $ \key -> atomically $ getPutLock getLock key name
+    releaseLocks = mapM_ (atomically . takeTMVar)
 
 -- getLock and putTMVar can be in one transaction on the assumption that getLock doesn't write in case the lock already exists,
 -- and in case it is created and added to some shared resource (we use TMap) it also helps avoid contention for the newly created lock.

src/Simplex/Messaging/Agent/NtfSubSupervisor.hs

@@ -33,7 +33,6 @@ import Simplex.Messaging.Agent.Protocol (ACommand (..), APartyCmd (..), AgentErr
 import Simplex.Messaging.Agent.RetryInterval
 import Simplex.Messaging.Agent.Store
 import Simplex.Messaging.Agent.Store.SQLite
-import Simplex.Messaging.Client.Agent ()
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Notifications.Protocol (NtfSubStatus (..), NtfTknStatus (..), SMPQueueNtf (..))
 import Simplex.Messaging.Notifications.Types
@@ -161,7 +160,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 +244,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/Client/Agent.hs

@@ -8,7 +8,6 @@
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TupleSections #-}
 {-# LANGUAGE TypeApplications #-}
-{-# OPTIONS_GHC -fno-warn-orphans #-}
 
 module Simplex.Messaging.Client.Agent where
 
@@ -19,7 +18,6 @@ import Control.Monad
 import Control.Monad.Except
 import Control.Monad.IO.Unlift
 import Control.Monad.Trans.Except
-import Control.Monad.Trans.Reader
 import Crypto.Random (ChaChaDRG)
 import Data.Bifunctor (bimap, first)
 import Data.ByteString.Char8 (ByteString)
@@ -47,8 +45,6 @@ import Simplex.Messaging.Transport
 import Simplex.Messaging.Util (catchAll_, toChunks, ($>>=))
 import System.Timeout (timeout)
 import UnliftIO (async)
-import UnliftIO.Exception (Exception)
-import qualified UnliftIO.Exception as E
 import UnliftIO.STM
 
 type SMPClientVar = SessionVar (Either SMPClientError SMPClient)
@@ -105,31 +101,6 @@ data SMPClientAgent = SMPClientAgent
     workerSeq :: TVar Int
   }
 
-newtype InternalException e = InternalException {unInternalException :: e}
-  deriving (Eq, Show)
-
-instance Exception e => Exception (InternalException e)
-
-instance Exception e => MonadUnliftIO (ExceptT e IO) where
-  {-# INLINE withRunInIO #-}
-  withRunInIO :: ((forall a. ExceptT e IO a -> IO a) -> IO b) -> ExceptT e IO b
-  withRunInIO inner =
-    ExceptT . fmap (first unInternalException) . E.try $
-      withRunInIO $ \run ->
-        inner $ run . (either (E.throwIO . InternalException) pure <=< runExceptT)
-
--- as MonadUnliftIO instance for IO is `withRunInIO inner = inner id`,
--- the last two lines could be replaced with:
--- inner $ either (E.throwIO . InternalException) pure <=< runExceptT
-
-instance Exception e => MonadUnliftIO (ExceptT e (ReaderT r IO)) where
-  {-# INLINE withRunInIO #-}
-  withRunInIO :: ((forall a. ExceptT e (ReaderT r IO) a -> IO a) -> IO b) -> ExceptT e (ReaderT r IO) b
-  withRunInIO inner =
-    withExceptT unInternalException . ExceptT . E.try $
-      withRunInIO $ \run ->
-        inner $ run . (either (E.throwIO . InternalException) pure <=< runExceptT)
-
 newSMPClientAgent :: SMPClientAgentConfig -> TVar ChaChaDRG -> STM SMPClientAgent
 newSMPClientAgent agentCfg@SMPClientAgentConfig {msgQSize, agentQSize} randomDrg = do
   msgQ <- newTBQueue msgQSize

src/Simplex/Messaging/Crypto/File.hs

@@ -31,7 +31,6 @@ import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy as LB
 import Data.List.NonEmpty (NonEmpty (..))
 import Data.Maybe (isJust)
-import Simplex.Messaging.Client.Agent ()
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.Lazy (LazyByteString)
 import qualified Simplex.Messaging.Crypto.Lazy as LC

src/Simplex/Messaging/Server.hs

@@ -51,7 +51,6 @@ 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, mapAccumR)
@@ -542,11 +541,7 @@ 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 _ -> 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
+    Cmd SRecipient _ -> verifyQueue (\q -> Just q `verifiedWith` recipientKey q) <$> get SRecipient
     -- 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
@@ -559,8 +554,6 @@ 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

src/Simplex/Messaging/Server/QueueStore.hs

@@ -8,9 +8,6 @@ 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,8 +24,6 @@ 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)
@@ -34,32 +32,24 @@ import Simplex.Messaging.Util (ifM, ($>>=))
 import UnliftIO.STM
 
 data QueueStore = QueueStore
-  { 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
+  { queues :: TMap RecipientId (TVar QueueRec),
+    senders :: TMap SenderId RecipientId,
+    notifiers :: TMap NotifierId RecipientId
   }
 
 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 {recipientKeys, senderKeys, notifierKeys, queues, senders, notifiers}
+  pure QueueStore {queues, senders, notifiers}
 
 addQueue :: QueueStore -> QueueRec -> STM (Either ErrorType ())
-addQueue QueueStore {recipientKeys, senderKeys, queues, senders} q@QueueRec {recipientId = rId, recipientKey, senderId = sId, senderKey} = do
+addQueue QueueStore {queues, senders} q@QueueRec {recipientId = rId, senderId = sId} = 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
@@ -74,14 +64,13 @@ 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 {senderKeys, queues} rId sKey =
+secureQueue QueueStore {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}
-        modifyTVar' senderKeys $ HM.insert rId sKey
-        writeTVar qVar q' $> Just q'
+         in 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/Util.hs

@@ -97,15 +97,15 @@ catchAll_ :: IO a -> IO a -> IO a
 catchAll_ a = catchAll a . const
 {-# INLINE catchAll_ #-}
 
-tryAllErrors :: (MonadUnliftIO m, MonadError e m) => (E.SomeException -> e) -> m a -> m (Either e a)
-tryAllErrors err action = tryError action `UE.catch` (pure . Left . err)
+tryAllErrors :: MonadUnliftIO m => (E.SomeException -> e) -> ExceptT e m a -> ExceptT e m (Either e a)
+tryAllErrors err action = ExceptT $ Right <$> runExceptT action `UE.catch` (pure . Left . err)
 {-# INLINE tryAllErrors #-}
 
 tryAllErrors' :: MonadUnliftIO m => (E.SomeException -> e) -> ExceptT e m a -> m (Either e a)
 tryAllErrors' err action = runExceptT action `UE.catch` (pure . Left . err)
 {-# INLINE tryAllErrors' #-}
 
-catchAllErrors :: (MonadUnliftIO m, MonadError e m) => (E.SomeException -> e) -> m a -> (e -> m a) -> m a
+catchAllErrors :: MonadUnliftIO m => (E.SomeException -> e) -> ExceptT e m a -> (e -> ExceptT e m a) -> ExceptT e m a
 catchAllErrors err action handler = tryAllErrors err action >>= either handler pure
 {-# INLINE catchAllErrors #-}
 
@@ -113,11 +113,11 @@ catchAllErrors' :: MonadUnliftIO m => (E.SomeException -> e) -> ExceptT e m a ->
 catchAllErrors' err action handler = tryAllErrors' err action >>= either handler pure
 {-# INLINE catchAllErrors' #-}
 
-catchThrow :: (MonadUnliftIO m, MonadError e m) => m a -> (E.SomeException -> e) -> m a
+catchThrow :: MonadUnliftIO m => ExceptT e m a -> (E.SomeException -> e) -> ExceptT e m a
 catchThrow action err = catchAllErrors err action throwError
 {-# INLINE catchThrow #-}
 
-allFinally :: (MonadUnliftIO m, MonadError e m) => (E.SomeException -> e) -> m a -> m b -> m a
+allFinally :: MonadUnliftIO m => (E.SomeException -> e) -> ExceptT e m a -> ExceptT e m b -> ExceptT e m a
 allFinally err action final = tryAllErrors err action >>= \r -> final >> either throwError pure r
 {-# INLINE allFinally #-}
 
@@ -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

@@ -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

tests/AgentTests/FunctionalAPITests.hs

@@ -1,7 +1,9 @@
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE FlexibleInstances #-}
 {-# LANGUAGE GADTs #-}
+{-# LANGUAGE InstanceSigs #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE NumericUnderscores #-}
@@ -50,10 +52,11 @@ 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
+import Data.Bifunctor (first)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Either (isRight)
@@ -65,7 +68,7 @@ import Data.Maybe (isJust, isNothing)
 import qualified Data.Set as S
 import Data.Time.Clock (diffUTCTime, getCurrentTime)
 import Data.Time.Clock.System (SystemTime (..), getSystemTime)
-import Data.Type.Equality
+import Data.Type.Equality (testEquality, (:~:) (Refl))
 import Data.Word (Word16)
 import qualified Database.SQLite.Simple as SQL
 import GHC.Stack (withFrozenCallStack)
@@ -77,7 +80,6 @@ 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), defaultClientConfig)
@@ -145,6 +147,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
@@ -435,7 +438,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
@@ -2698,9 +2702,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
@@ -2710,36 +2712,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
 
@@ -2780,3 +2812,16 @@ exchangeGreetingsMsgIds alice bobId aliceMsgId bob aliceId bobMsgId = do
   get bob ##> ("", aliceId, SENT bobMsgId')
   get alice =##> \case ("", c, Msg "hello too") -> c == bobId; _ -> False
   ackMessage alice bobId aliceMsgId' Nothing
+
+newtype InternalException e = InternalException {unInternalException :: e}
+  deriving (Eq, Show)
+
+instance Exception e => Exception (InternalException e)
+
+instance Exception e => MonadUnliftIO (ExceptT e IO) where
+  {-# INLINE withRunInIO #-}
+  withRunInIO :: ((forall a. ExceptT e IO a -> IO a) -> IO b) -> ExceptT e IO b
+  withRunInIO inner =
+    ExceptT . fmap (first unInternalException) . try $
+      withRunInIO $ \run ->
+        inner $ run . (either (throwIO . InternalException) pure <=< runExceptT)

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/CoreTests/UtilTests.hs

@@ -2,11 +2,11 @@
 
 module CoreTests.UtilTests where
 
+import AgentTests.FunctionalAPITests ()
 import Control.Exception (Exception, SomeException, throwIO)
 import Control.Monad.Except
 import Control.Monad.IO.Class
 import Data.IORef
-import Simplex.Messaging.Client.Agent ()
 import Simplex.Messaging.Util
 import Test.Hspec
 import qualified UnliftIO.Exception as UE

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
-  fdescribe "Timing of AUTH error" $ testTiming t
+  describe "Timing of AUTH error" $ testTiming t
   describe "Message notifications" $ testMessageNotifications t
   describe "Message expiration" $ do
     testMsgExpireOnSend t'