remove StrictData

* rfc: moving queue records to journal storage and queue blobs

* rfc: expiring messages in journal storage

* add comment

rfcs/2024-11-25-journal-expiration.md

@@ -0,0 +1,15 @@
+# Expiring messages in journal storage
+
+## Problem
+
+The journal storage servers recently migrated to do not delete delivered or expired messages, they only update pointers to journal file lines. The messages are actually deleted when the whole journal file is deleted (when fully deleted or fully expired).
+
+The problem is that in case the queue stops receiving the new messages then writing of messages won't switch to the new journal file, and the current journal file containing delivered or expired messages would never be deleted.
+
+## Solution
+
+Remove current journal file and update queue_state.log during message expiration of "idle" queue (that is, without any new messages received or delivered within 3 hours) in case when:
+- the queue is "empty" after the expiration
+- the queue contains only quota marker(s), in which case move them to a new journal file and update the queue_state accordingly. Quota markers can be kept indefinitely to prevent writing the new messages to the dormant queues that reached capacity, so it's important to handle this case.
+
+Also remove current journal file when the queue is opened in case it is empty (as it would not be ever expired in case it remains empty), and also update queue_state.log

rfcs/2024-11-25-queue-blobs-2.md

@@ -0,0 +1,58 @@
+# Blob extensions for SMP queues 2 and queue storage
+
+This document evolves the design proposed [here](./2024-09-09-smp-blobs.md).
+
+## Problems
+
+In addition to problems in the first doc, we have these issues with in-memory queue record storage:
+- many queues are idle or rarely used, but they are loaded to memory, and currently just loading all queues uses 20gb RAM on each server, and takes 10 min to process, increasing downtimes during restarts.
+- adding blobs to memory would make this problem much worse.
+
+## Proposed solution
+
+Move queues to the same journalling approach as [used for messages](./2024-09-01-smp-message-storage.md) now, with independent file names in the same folders.
+
+Each queue change would be logged to its own file, and every time the queue is opened the whole file will be read and compacted to a single line - replacing one store log for all queues, with individual log files for each queue.
+
+Queue deletion would not be making a record in the file, instead it would be deleting the entire folder - it would reduce retention period for any metadata of deleted queues.
+
+We could additionally record deletions to the central log, for debugging, and reset it on every start. But in this case we should not remove folders at the point of deletion, but rather mark them as deleted and delete on restart. TBC
+
+It would also allow simplifying blob storage by having only one blob per queue - for example, limied to 16kb (a bit smaller to fit in block) for contact address queues and 4-8kb for invitations (to fit PQ keys and conversation preferences).
+
+We would also need to be able to lookup recipient ID via sender/notifier/link IDs.
+
+One possible solution is to use and load to memory a central index file. But it is likely to also consume a lot of memory and result in slow starts.
+
+Another solution that is probably better is to use the same folder structure and put notifier/sender/link files with the ID of the recipient queue inside the files. So to locate recipient queue the sender would have to locate folder containing the reference file pointing to the recipient queue and then to locate the actual queue data.
+
+## Implementation details
+
+Each queue folder would these files:
+
+- queue_state.log (and timestamped backups) - to store pointers to message journals (already implemented)
+- messages.randomBase64.log - message journals (already implemented)
+- queue_rec.log (and timestamped backups) - to log complete queue record every time it is changed (so only the last line needs to be read following the same logic as with queue_state.log, to prevent file corruption).
+- blob.data, blob.data.bak, blob.timestamp.data - files for data blobs (to make sure some copy of this file is readable/correct in case of write corruption) - the same two step overwrite process will be used as currently with store log compacting:
+  - on write: 1. if file exists, move it to .bak, 2. store new blob to .data, 3. move .bak to .timestamp.data
+  - on read: 1. if .bak exists, move it to .data 2. use .data
+
+Additional suggestion to reduce probability of queue_state.log and queue_rec.log file corruption is to do one of the following:
+- log end of lines in the beginning of the output, not in the end, to prevent the last line from being corrupted in case the previous line was not fully stored. The downside is that the file will not be EOL terminated, and there will be no confirmation that the output was fully made.
+- log EOL both in the beginning and at the end of output, and ignore empty lines in between - this would both confirm that the last line is fully logged and prevent corruption of the next line in case it was not.
+- check the last byte of the file and log EOL if it is not EOL. Probably cleanest approach, but with a small performance cost.
+
+If queue folder is a reference to the queue, it may have one of these files:
+- notifier.id
+- sender.id
+- link.id
+
+These files would contain a one line with the recipient ID of the queue. These files would never change, they can only be deleted when queue is deleted or when notifier/link is deleted.
+
+There is logic in code preventing using the same ID in different contexts, and the ID size is large enough to make any collisions unlikely (192 bits), so with correctly working code the queue folder would either have one of reference files, and nothing else, or the queue and message files from the beginning of this section. But even if the same ID is re-used in different context, it should not cause any problems as file names don't overlap.
+
+While we could store different types of references in different types of folders, it would have additional costs of maintaining 4 folder hierarchies. Instead we could use the fact that it is one hierarchy to prevent using the same ID in different contexts.
+
+## Protocol
+
+The only change in protocol is that there will be only one blob per queue, without markers (see the previous doc). Otherwise the protocol and proposed privacy improvement seem reasonable.

simplexmq.cabal

@@ -1,7 +1,7 @@
 cabal-version: 1.12
 
 name:           simplexmq
-version:        6.2.0.3
+version:        6.2.0.6
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -214,8 +214,6 @@ library
       Paths_simplexmq
   hs-source-dirs:
       src
-  default-extensions:
-      StrictData
   ghc-options: -Weverything -Wno-missing-exported-signatures -Wno-missing-import-lists -Wno-missed-specialisations -Wno-all-missed-specialisations -Wno-unsafe -Wno-safe -Wno-missing-local-signatures -Wno-missing-kind-signatures -Wno-missing-deriving-strategies -Wno-monomorphism-restriction -Wno-prepositive-qualified-module -Wno-implicit-prelude -Wno-missing-safe-haskell-mode -Wno-missing-export-lists -Wno-partial-fields -Wcompat -Werror=incomplete-record-updates -Werror=incomplete-patterns -Werror=incomplete-uni-patterns -Werror=missing-home-modules -Werror=missing-methods -Werror=tabs -Wredundant-constraints -Wincomplete-record-updates -Wunused-type-patterns -O2
   include-dirs:
       cbits
@@ -297,8 +295,6 @@ executable ntf-server
       Paths_simplexmq
   hs-source-dirs:
       apps/ntf-server
-  default-extensions:
-      StrictData
   ghc-options: -Weverything -Wno-missing-exported-signatures -Wno-missing-import-lists -Wno-missed-specialisations -Wno-all-missed-specialisations -Wno-unsafe -Wno-safe -Wno-missing-local-signatures -Wno-missing-kind-signatures -Wno-missing-deriving-strategies -Wno-monomorphism-restriction -Wno-prepositive-qualified-module -Wno-implicit-prelude -Wno-missing-safe-haskell-mode -Wno-missing-export-lists -Wno-partial-fields -Wcompat -Werror=incomplete-record-updates -Werror=incomplete-patterns -Werror=incomplete-uni-patterns -Werror=missing-methods -Werror=tabs -Wredundant-constraints -Wincomplete-record-updates -Wunused-type-patterns -O2 -threaded -rtsopts
   build-depends:
       base
@@ -317,8 +313,6 @@ executable smp-server
   hs-source-dirs:
       apps/smp-server
       apps/smp-server/web
-  default-extensions:
-      StrictData
   ghc-options: -Weverything -Wno-missing-exported-signatures -Wno-missing-import-lists -Wno-missed-specialisations -Wno-all-missed-specialisations -Wno-unsafe -Wno-safe -Wno-missing-local-signatures -Wno-missing-kind-signatures -Wno-missing-deriving-strategies -Wno-monomorphism-restriction -Wno-prepositive-qualified-module -Wno-implicit-prelude -Wno-missing-safe-haskell-mode -Wno-missing-export-lists -Wno-partial-fields -Wcompat -Werror=incomplete-record-updates -Werror=incomplete-patterns -Werror=incomplete-uni-patterns -Werror=missing-methods -Werror=tabs -Wredundant-constraints -Wincomplete-record-updates -Wunused-type-patterns -O2 -threaded -rtsopts
   build-depends:
       base
@@ -345,8 +339,6 @@ executable xftp
       Paths_simplexmq
   hs-source-dirs:
       apps/xftp
-  default-extensions:
-      StrictData
   ghc-options: -Weverything -Wno-missing-exported-signatures -Wno-missing-import-lists -Wno-missed-specialisations -Wno-all-missed-specialisations -Wno-unsafe -Wno-safe -Wno-missing-local-signatures -Wno-missing-kind-signatures -Wno-missing-deriving-strategies -Wno-monomorphism-restriction -Wno-prepositive-qualified-module -Wno-implicit-prelude -Wno-missing-safe-haskell-mode -Wno-missing-export-lists -Wno-partial-fields -Wcompat -Werror=incomplete-record-updates -Werror=incomplete-patterns -Werror=incomplete-uni-patterns -Werror=missing-methods -Werror=tabs -Wredundant-constraints -Wincomplete-record-updates -Wunused-type-patterns -O2 -threaded -rtsopts
   build-depends:
       base
@@ -361,8 +353,6 @@ executable xftp-server
       Paths_simplexmq
   hs-source-dirs:
       apps/xftp-server
-  default-extensions:
-      StrictData
   ghc-options: -Weverything -Wno-missing-exported-signatures -Wno-missing-import-lists -Wno-missed-specialisations -Wno-all-missed-specialisations -Wno-unsafe -Wno-safe -Wno-missing-local-signatures -Wno-missing-kind-signatures -Wno-missing-deriving-strategies -Wno-monomorphism-restriction -Wno-prepositive-qualified-module -Wno-implicit-prelude -Wno-missing-safe-haskell-mode -Wno-missing-export-lists -Wno-partial-fields -Wcompat -Werror=incomplete-record-updates -Werror=incomplete-patterns -Werror=incomplete-uni-patterns -Werror=missing-methods -Werror=tabs -Wredundant-constraints -Wincomplete-record-updates -Wunused-type-patterns -O2 -threaded -rtsopts
   build-depends:
       base
@@ -384,6 +374,7 @@ test-suite simplexmq-test
       AgentTests.MigrationTests
       AgentTests.NotificationTests
       AgentTests.SchemaDump
+      AgentTests.ServerChoice
       AgentTests.SQLiteTests
       CLITests
       CoreTests.BatchingTests
@@ -416,8 +407,6 @@ test-suite simplexmq-test
   hs-source-dirs:
       tests
       apps/smp-server/web
-  default-extensions:
-      StrictData
   ghc-options: -Weverything -Wno-missing-exported-signatures -Wno-missing-import-lists -Wno-missed-specialisations -Wno-all-missed-specialisations -Wno-unsafe -Wno-safe -Wno-missing-local-signatures -Wno-missing-kind-signatures -Wno-missing-deriving-strategies -Wno-monomorphism-restriction -Wno-prepositive-qualified-module -Wno-implicit-prelude -Wno-missing-safe-haskell-mode -Wno-missing-export-lists -Wno-partial-fields -Wcompat -Werror=incomplete-record-updates -Werror=incomplete-patterns -Werror=incomplete-uni-patterns -Werror=missing-methods -Werror=tabs -Wredundant-constraints -Wincomplete-record-updates -Wunused-type-patterns -O2 -threaded -rtsopts -with-rtsopts=-A64M -with-rtsopts=-N1
   build-depends:
       base

src/Simplex/FileTransfer/Agent.hs

@@ -461,14 +461,14 @@ runXFTPSndPrepareWorker c Worker {doWork} = do
           pure srv
           where
             tryCreate = do
-              usedSrvs <- newTVarIO ([] :: [XFTPServer])
+              triedHosts <- newTVarIO S.empty
               let AgentClient {xftpServers} = c
               userSrvCount <- liftIO $ length <$> TM.lookupIO userId xftpServers
               withRetryIntervalCount (riFast ri) $ \n _ loop -> do
                 liftIO $ waitWhileSuspended c
                 liftIO $ waitForUserNetwork c
                 let triedAllSrvs = n > userSrvCount
-                createWithNextSrv usedSrvs
+                createWithNextSrv triedHosts
                   `catchAgentError` \e -> retryOnError "XFTP prepare worker" (retryLoop loop triedAllSrvs e) (throwE e) e
               where
                 -- we don't do closeXFTPServerClient here to not risk closing connection for concurrent chunk upload
@@ -477,10 +477,10 @@ runXFTPSndPrepareWorker c Worker {doWork} = do
                     when (triedAllSrvs && serverHostError e) $ notify c sndFileEntityId $ SFWARN e
                   liftIO $ assertAgentForeground c
                   loop
-            createWithNextSrv usedSrvs = do
+            createWithNextSrv triedHosts = do
               deleted <- withStore' c $ \db -> getSndFileDeleted db sndFileId
               when deleted $ throwE $ FILE NO_FILE
-              withNextSrv c userId usedSrvs [] $ \srvAuth -> do
+              withNextSrv c userId storageSrvs triedHosts [] $ \srvAuth -> do
                 replica <- agentXFTPNewChunk c ch numRecipients' srvAuth
                 pure (replica, srvAuth)
 
@@ -546,8 +546,8 @@ runXFTPSndWorker c srv Worker {doWork} = do
         withStore' c $ \db -> updateSndFileComplete db sndFileId
       where
         addRecipients :: SndFileChunk -> SndFileChunkReplica -> AM SndFileChunkReplica
-        addRecipients ch@SndFileChunk {numRecipients} cr@SndFileChunkReplica {rcvIdsKeys}
-          | length rcvIdsKeys > numRecipients = throwE $ INTERNAL "too many recipients"
+        addRecipients ch@SndFileChunk {numRecipients} cr@SndFileChunkReplica {sndChunkReplicaId, rcvIdsKeys}
+          | length rcvIdsKeys > numRecipients = throwE $ INTERNAL ("too many recipients, sndChunkReplicaId = " <> show sndChunkReplicaId)
           | length rcvIdsKeys == numRecipients = pure cr
           | otherwise = do
               let numRecipients' = min (numRecipients - length rcvIdsKeys) maxRecipients

src/Simplex/Messaging/Agent.hs

@@ -120,6 +120,8 @@ module Simplex.Messaging.Agent
     debugAgentLocks,
     getAgentSubscriptions,
     logConnection,
+    -- for tests
+    withAgentEnv,
   )
 where
 
@@ -815,11 +817,13 @@ newConnToAccept c connId enableNtfs invId pqSup = do
 
 joinConn :: AgentClient -> UserId -> ConnId -> Bool -> ConnectionRequestUri c -> ConnInfo -> PQSupport -> SubscriptionMode -> AM SndQueueSecured
 joinConn c userId connId enableNtfs cReq cInfo pqSupport subMode = do
-  srv <- case cReq of
-    CRInvitationUri ConnReqUriData {crSmpQueues = q :| _} _ ->
-      getNextServer c userId [qServer q]
-    _ -> getSMPServer c userId
+  srv <- getNextSMPServer c userId [qServer cReqQueue]
   joinConnSrv c userId connId enableNtfs cReq cInfo pqSupport subMode srv
+  where
+    cReqQueue :: SMPQueueUri
+    cReqQueue = case cReq of
+      CRInvitationUri ConnReqUriData {crSmpQueues = q :| _} _ -> q
+      CRContactUri ConnReqUriData {crSmpQueues = q :| _} -> q
 
 startJoinInvitation :: AgentClient -> UserId -> ConnId -> Maybe SndQueue -> Bool -> ConnectionRequestUri 'CMInvitation -> PQSupport -> AM (ConnData, SndQueue, CR.SndE2ERatchetParams 'C.X448)
 startJoinInvitation c userId connId sq_ enableNtfs cReqUri pqSup =
@@ -1194,14 +1198,13 @@ runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
     processCmd ri PendingCommand {cmdId, corrId, userId, command} pendingCmds = case command of
       AClientCommand cmd -> case cmd of
         NEW enableNtfs (ACM cMode) pqEnc subMode -> noServer $ do
-          usedSrvs <- newTVarIO ([] :: [SMPServer])
-          tryCommand . withNextSrv c userId usedSrvs [] $ \srv -> do
+          triedHosts <- newTVarIO S.empty
+          tryCommand . withNextSrv c userId storageSrvs triedHosts [] $ \srv -> do
             cReq <- newRcvConnSrv c userId connId enableNtfs cMode Nothing pqEnc subMode srv
             notify $ INV (ACR cMode cReq)
         JOIN enableNtfs (ACR _ cReq@(CRInvitationUri ConnReqUriData {crSmpQueues = q :| _} _)) pqEnc subMode connInfo -> noServer $ do
-          let initUsed = [qServer q]
-          usedSrvs <- newTVarIO initUsed
-          tryCommand . withNextSrv c userId usedSrvs initUsed $ \srv -> do
+          triedHosts <- newTVarIO S.empty
+          tryCommand . withNextSrv c userId storageSrvs triedHosts [qServer q] $ \srv -> do
             sqSecured <- joinConnSrvAsync c userId connId enableNtfs cReq connInfo pqEnc subMode srv
             notify $ JOINED sqSecured
         LET confId ownCInfo -> withServer' . tryCommand $ allowConnection' c connId confId ownCInfo >> notify OK
@@ -1649,8 +1652,8 @@ switchDuplexConnection c (DuplexConnection cData@ConnData {connId, userId} rqs s
   checkRQSwchStatus rq RSSwitchStarted
   clientVRange <- asks $ smpClientVRange . config
   -- try to get the server that is different from all queues, or at least from the primary rcv queue
-  srvAuth@(ProtoServerWithAuth srv _) <- getNextServer c userId $ map qServer (L.toList rqs) <> map qServer (L.toList sqs)
-  srv' <- if srv == server then getNextServer c userId [server] else pure srvAuth
+  srvAuth@(ProtoServerWithAuth srv _) <- getNextSMPServer c userId $ map qServer (L.toList rqs) <> map qServer (L.toList sqs)
+  srv' <- if srv == server then getNextSMPServer c userId [server] else pure srvAuth
   (q, qUri, tSess, sessId) <- newRcvQueue c userId connId srv' clientVRange SMSubscribe False
   let rq' = (q :: NewRcvQueue) {primary = True, dbReplaceQueueId = Just dbQueueId}
   rq'' <- withStore c $ \db -> addConnRcvQueue db connId rq'
@@ -2158,9 +2161,13 @@ debugAgentLocks AgentClient {connLocks = cs, invLocks = is, deleteLock = d} = do
     getLocks ls = atomically $ M.mapKeys (B.unpack . strEncode) . M.mapMaybe id <$> (mapM tryReadTMVar =<< readTVar ls)
 
 getSMPServer :: AgentClient -> UserId -> AM SMPServerWithAuth
-getSMPServer c userId = withUserServers c userId pickServer
+getSMPServer c userId = getNextSMPServer c userId []
 {-# INLINE getSMPServer #-}
 
+getNextSMPServer :: AgentClient -> UserId -> [SMPServer] -> AM SMPServerWithAuth
+getNextSMPServer c userId = getNextServer c userId storageSrvs
+{-# INLINE getNextSMPServer #-}
+
 subscriber :: AgentClient -> AM' ()
 subscriber c@AgentClient {msgQ} = forever $ do
   t <- atomically $ readTBQueue msgQ

src/Simplex/Messaging/Agent/Client.hs

@@ -149,7 +149,6 @@ module Simplex.Messaging.Agent.Client
     userServers,
     pickServer,
     getNextServer,
-    withUserServers,
     withNextSrv,
     incSMPServerStat,
     incSMPServerStat',
@@ -193,12 +192,12 @@ import qualified Data.ByteString.Char8 as B
 import Data.Either (isRight, partitionEithers)
 import Data.Functor (($>))
 import Data.Int (Int64)
-import Data.List (deleteFirstsBy, find, foldl', partition, (\\))
+import Data.List (find, foldl', partition)
 import Data.List.NonEmpty (NonEmpty (..), (<|))
 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, listToMaybe)
+import Data.Maybe (catMaybes, fromMaybe, isJust, isNothing, listToMaybe, mapMaybe)
 import Data.Set (Set)
 import qualified Data.Set as S
 import Data.Text (Text)
@@ -264,7 +263,6 @@ import Simplex.Messaging.Protocol
     VersionSMPC,
     XFTPServer,
     XFTPServerWithAuth,
-    sameSrvAddr',
     pattern NoEntity,
   )
 import qualified Simplex.Messaging.Protocol as SMP
@@ -619,7 +617,7 @@ getSMPServerClient c@AgentClient {active, smpClients, workerSeq} tSess = do
 getSMPProxyClient :: AgentClient -> Maybe SMPServerWithAuth -> SMPTransportSession -> AM (SMPConnectedClient, Either AgentErrorType ProxiedRelay)
 getSMPProxyClient c@AgentClient {active, smpClients, smpProxiedRelays, workerSeq} proxySrv_ destSess@(userId, destSrv, qId) = do
   unlessM (readTVarIO active) $ throwE INACTIVE
-  proxySrv <- maybe (getNextServer c userId [destSrv]) pure proxySrv_
+  proxySrv <- maybe (getNextServer c userId proxySrvs [destSrv]) pure proxySrv_
   ts <- liftIO getCurrentTime
   atomically (getClientVar proxySrv ts) >>= \(tSess, auth, v) ->
     either (newProxyClient tSess auth ts) (waitForProxyClient tSess auth) v
@@ -1074,7 +1072,7 @@ sendOrProxySMPCommand ::
   (SMPClient -> ProxiedRelay -> ExceptT SMPClientError IO (Either ProxyClientError ())) ->
   (SMPClient -> ExceptT SMPClientError IO ()) ->
   AM (Maybe SMPServer)
-sendOrProxySMPCommand c userId destSrv connId cmdStr senderId sendCmdViaProxy sendCmdDirectly = do
+sendOrProxySMPCommand c userId destSrv@ProtocolServer {host = destHosts} connId cmdStr senderId sendCmdViaProxy sendCmdDirectly = do
   tSess <- mkTransportSession c userId destSrv connId
   ifM shouldUseProxy (sendViaProxy Nothing tSess) (sendDirectly tSess $> Nothing)
   where
@@ -1093,7 +1091,7 @@ sendOrProxySMPCommand c userId destSrv connId cmdStr senderId sendCmdViaProxy se
         SPFAllow -> True
         SPFAllowProtected -> ipAddressProtected cfg destSrv
         SPFProhibit -> False
-    unknownServer = liftIO $ maybe True (notElem destSrv . knownSrvs) <$> TM.lookupIO userId (smpServers c)
+    unknownServer = liftIO $ maybe True (\srvs -> all (`S.notMember` knownHosts srvs) destHosts) <$> TM.lookupIO userId (smpServers c)
     sendViaProxy :: Maybe SMPServerWithAuth -> SMPTransportSession -> AM (Maybe SMPServer)
     sendViaProxy proxySrv_ destSess@(_, _, connId_) = do
       r <- tryAgentError . withProxySession c proxySrv_ destSess senderId ("PFWD " <> cmdStr) $ \(SMPConnectedClient smp _, proxySess@ProxiedRelay {prBasicAuth}) -> do
@@ -1388,6 +1386,7 @@ temporaryAgentError = \case
   PROXY _ _ (ProxyProtocolError (SMP.PROXY (SMP.BROKER e))) -> tempBrokerError e
   PROXY _ _ (ProxyProtocolError (SMP.PROXY SMP.NO_SESSION)) -> True
   INACTIVE -> True
+  CRITICAL True _ -> True -- critical errors that do not show restart button are likely to be permanent
   _ -> False
   where
     tempBrokerError = \case
@@ -2035,33 +2034,82 @@ userServers c = case protocolTypeI @p of
   SPXFTP -> xftpServers c
 {-# INLINE userServers #-}
 
-pickServer :: forall p. NonEmpty (ProtoServerWithAuth p) -> AM (ProtoServerWithAuth p)
+pickServer :: NonEmpty (Maybe OperatorId, ProtoServerWithAuth p) -> AM (ProtoServerWithAuth p)
 pickServer = \case
-  srv :| [] -> pure srv
+  (_, srv) :| [] -> pure srv
   servers -> do
     gen <- asks randomServer
-    atomically $ (servers L.!!) <$> stateTVar gen (randomR (0, L.length servers - 1))
+    atomically $ snd . (servers L.!!) <$> stateTVar gen (randomR (0, L.length servers - 1))
 
-getNextServer :: forall p. (ProtocolTypeI p, UserProtocol p) => AgentClient -> UserId -> [ProtocolServer p] -> AM (ProtoServerWithAuth p)
-getNextServer c userId usedSrvs = withUserServers c userId $ \srvs ->
-  case L.nonEmpty $ deleteFirstsBy sameSrvAddr' (L.toList srvs) (map noAuthSrv usedSrvs) of
-    Just srvs' -> pickServer srvs'
-    _ -> pickServer srvs
+getNextServer ::
+  (ProtocolTypeI p, UserProtocol p) =>
+  AgentClient ->
+  UserId -> 
+  (UserServers p -> NonEmpty (Maybe OperatorId, ProtoServerWithAuth p)) ->
+  [ProtocolServer p] ->
+  AM (ProtoServerWithAuth p)
+getNextServer c userId srvsSel usedSrvs = do
+  srvs <- getUserServers_ c userId srvsSel
+  snd <$> getNextServer_ srvs (usedOperatorsHosts srvs usedSrvs)
 
-withUserServers :: forall p a. (ProtocolTypeI p, UserProtocol p) => AgentClient -> UserId -> (NonEmpty (ProtoServerWithAuth p) -> AM a) -> AM a
-withUserServers c userId action =
+usedOperatorsHosts :: NonEmpty (Maybe OperatorId, ProtoServerWithAuth p) -> [ProtocolServer p] -> (Set (Maybe OperatorId), Set TransportHost)
+usedOperatorsHosts srvs usedSrvs = (usedOperators, usedHosts)
+  where
+    usedHosts = S.unions $ map serverHosts usedSrvs
+    usedOperators = S.fromList $ mapMaybe usedOp $ L.toList srvs
+    usedOp (op, srv) = if hasUsedHost srv then Just op else Nothing
+    hasUsedHost (ProtoServerWithAuth srv _) = any (`S.member` usedHosts) $ serverHosts srv
+
+getNextServer_ ::
+  (ProtocolTypeI p, UserProtocol p) =>
+  NonEmpty (Maybe OperatorId, ProtoServerWithAuth p) ->
+  (Set (Maybe OperatorId), Set TransportHost) ->
+  AM (NonEmpty (Maybe OperatorId, ProtoServerWithAuth p), ProtoServerWithAuth p)
+getNextServer_ servers (usedOperators, usedHosts) = do
+  -- choose from servers of unused operators, when possible
+  let otherOpsSrvs = filterOrAll ((`S.notMember` usedOperators) . fst) servers
+      -- choose from servers with unused hosts when possible
+      unusedSrvs = filterOrAll (isUnusedServer usedHosts) otherOpsSrvs
+  (otherOpsSrvs,) <$> pickServer unusedSrvs
+  where
+    filterOrAll p srvs = fromMaybe srvs $ L.nonEmpty $ L.filter p srvs
+
+isUnusedServer :: Set TransportHost -> (Maybe OperatorId, ProtoServerWithAuth p) -> Bool
+isUnusedServer usedHosts (_, ProtoServerWithAuth ProtocolServer {host} _) = all (`S.notMember` usedHosts) host
+
+getUserServers_ ::
+  (ProtocolTypeI p, UserProtocol p) =>
+  AgentClient ->
+  UserId ->
+  (UserServers p -> NonEmpty (Maybe OperatorId, ProtoServerWithAuth p)) ->
+  AM (NonEmpty (Maybe OperatorId, ProtoServerWithAuth p))
+getUserServers_ c userId srvsSel =
   liftIO (TM.lookupIO userId $ userServers c) >>= \case
-    Just srvs -> action $ enabledSrvs srvs
+    Just srvs -> pure $ srvsSel srvs
     _ -> throwE $ INTERNAL "unknown userId - no user servers"
 
-withNextSrv :: forall p a. (ProtocolTypeI p, UserProtocol p) => AgentClient -> UserId -> TVar [ProtocolServer p] -> [ProtocolServer p] -> (ProtoServerWithAuth p -> AM a) -> AM a
-withNextSrv c userId usedSrvs initUsed action = do
-  used <- readTVarIO usedSrvs
-  srvAuth@(ProtoServerWithAuth srv _) <- getNextServer c userId used
-  srvs_ <- liftIO $ TM.lookupIO userId $ userServers c
-  let unused = maybe [] ((\\ used) . map protoServer . L.toList . enabledSrvs) srvs_
-      used' = if null unused then initUsed else srv : used
-  atomically $ writeTVar usedSrvs $! used'
+-- This function checks used servers and operators every time to allow
+-- changing configuration while retry look is executing.
+-- This function is not thread safe.
+withNextSrv ::
+  (ProtocolTypeI p, UserProtocol p) =>
+  AgentClient ->
+  UserId ->
+  (UserServers p -> NonEmpty (Maybe OperatorId, ProtoServerWithAuth p)) ->
+  TVar (Set TransportHost) ->
+  [ProtocolServer p] -> 
+  (ProtoServerWithAuth p -> AM a) ->
+  AM a
+withNextSrv c userId srvsSel triedHosts usedSrvs action = do
+  srvs <- getUserServers_ c userId srvsSel
+  let (usedOperators, usedHosts) = usedOperatorsHosts srvs usedSrvs
+  tried <- readTVarIO triedHosts
+  let triedOrUsed = S.union tried usedHosts
+  (otherOpsSrvs, srvAuth@(ProtoServerWithAuth srv _)) <- getNextServer_ srvs (usedOperators, triedOrUsed)
+  let newHosts = serverHosts srv
+      unusedSrvs = L.filter (isUnusedServer $ S.union triedOrUsed newHosts) otherOpsSrvs
+      !tried' = if null unusedSrvs then S.empty else S.union tried newHosts
+  atomically $ writeTVar triedHosts tried'
   action srvAuth
 
 incSMPServerStat :: AgentClient -> UserId -> SMPServer -> (AgentSMPServerStats -> TVar Int) -> STM ()

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

@@ -17,11 +17,14 @@ module Simplex.Messaging.Agent.Env.SQLite
     AgentConfig (..),
     InitialAgentServers (..),
     ServerCfg (..),
+    ServerRoles (..),
+    OperatorId,
     UserServers (..),
     NetworkConfig (..),
     presetServerCfg,
-    enabledServerCfg,
+    allRoles,
     mkUserServers,
+    serverHosts,
     defaultAgentConfig,
     defaultReconnectInterval,
     tryAgentError,
@@ -42,6 +45,7 @@ module Simplex.Messaging.Agent.Env.SQLite
 where
 
 import Control.Concurrent (ThreadId)
+import Control.Exception (BlockedIndefinitelyOnSTM (..), SomeException, fromException)
 import Control.Monad.Except
 import Control.Monad.IO.Unlift
 import Control.Monad.Reader
@@ -54,6 +58,8 @@ import Data.List.NonEmpty (NonEmpty)
 import qualified Data.List.NonEmpty as L
 import Data.Map.Strict (Map)
 import Data.Maybe (fromMaybe)
+import Data.Set (Set)
+import qualified Data.Set as S
 import Data.Time.Clock (NominalDiffTime, nominalDay)
 import Data.Time.Clock.System (SystemTime (..))
 import Data.Word (Word16)
@@ -71,14 +77,14 @@ import Simplex.Messaging.Notifications.Client (defaultNTFClientConfig)
 import Simplex.Messaging.Notifications.Transport (NTFVersion)
 import Simplex.Messaging.Notifications.Types
 import Simplex.Messaging.Parsers (defaultJSON)
-import Simplex.Messaging.Protocol (NtfServer, ProtoServerWithAuth, ProtocolServer, ProtocolType (..), ProtocolTypeI, VersionRangeSMPC, XFTPServer, supportedSMPClientVRange)
+import Simplex.Messaging.Protocol (NtfServer, ProtoServerWithAuth (..), ProtocolServer (..), ProtocolType (..), ProtocolTypeI, VersionRangeSMPC, XFTPServer, supportedSMPClientVRange)
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport (SMPVersion)
+import Simplex.Messaging.Transport.Client (TransportHost)
 import Simplex.Messaging.Util (allFinally, catchAllErrors, catchAllErrors', tryAllErrors, tryAllErrors')
 import System.Mem.Weak (Weak)
 import System.Random (StdGen, newStdGen)
-import UnliftIO (SomeException)
 import UnliftIO.STM
 
 type AM' a = ReaderT Env IO a
@@ -94,29 +100,42 @@ data InitialAgentServers = InitialAgentServers
 
 data ServerCfg p = ServerCfg
   { server :: ProtoServerWithAuth p,
-    preset :: Bool,
-    tested :: Maybe Bool,
-    enabled :: Bool
+    operator :: Maybe OperatorId,
+    enabled :: Bool,
+    roles :: ServerRoles
   }
   deriving (Show)
 
-enabledServerCfg :: ProtoServerWithAuth p -> ServerCfg p
-enabledServerCfg server = ServerCfg {server, preset = False, tested = Nothing, enabled = True}
+data ServerRoles = ServerRoles
+  { storage :: Bool,
+    proxy :: Bool
+  }
+  deriving (Show)
 
-presetServerCfg :: Bool -> ProtoServerWithAuth p -> ServerCfg p
-presetServerCfg enabled server = ServerCfg {server, preset = True, tested = Nothing, enabled}
+allRoles :: ServerRoles
+allRoles = ServerRoles True True
+
+presetServerCfg :: Bool -> ServerRoles -> Maybe OperatorId -> ProtoServerWithAuth p -> ServerCfg p
+presetServerCfg enabled roles operator server =
+  ServerCfg {server, operator, enabled, roles}
 
 data UserServers p = UserServers
-  { enabledSrvs :: NonEmpty (ProtoServerWithAuth p),
-    knownSrvs :: NonEmpty (ProtocolServer p)
+  { storageSrvs :: NonEmpty (Maybe OperatorId, ProtoServerWithAuth p),
+    proxySrvs :: NonEmpty (Maybe OperatorId, ProtoServerWithAuth p),
+    knownHosts :: Set TransportHost
   }
 
+type OperatorId = Int64
+
 -- This function sets all servers as enabled in case all passed servers are disabled.
 mkUserServers :: NonEmpty (ServerCfg p) -> UserServers p
-mkUserServers srvs = UserServers {enabledSrvs, knownSrvs}
+mkUserServers srvs = UserServers {storageSrvs = filterSrvs storage, proxySrvs = filterSrvs proxy, knownHosts}
   where
-    enabledSrvs = L.map (\ServerCfg {server} -> server) $ fromMaybe srvs $ L.nonEmpty $ L.filter (\ServerCfg {enabled} -> enabled) srvs
-    knownSrvs = L.map (\ServerCfg {server = ProtoServerWithAuth srv _} -> srv) srvs
+    filterSrvs role = L.map (\ServerCfg {operator, server} -> (operator, server)) $ fromMaybe srvs $ L.nonEmpty $ L.filter (\ServerCfg {enabled, roles} -> enabled && role roles) srvs
+    knownHosts = S.unions $ L.map (\ServerCfg {server = ProtoServerWithAuth srv _} -> serverHosts srv) srvs
+
+serverHosts :: ProtocolServer p -> Set TransportHost
+serverHosts ProtocolServer {host} = S.fromList $ L.toList host
 
 data AgentConfig = AgentConfig
   { tcpPort :: Maybe ServiceName,
@@ -313,7 +332,9 @@ agentFinally = allFinally mkInternal
 {-# INLINE agentFinally #-}
 
 mkInternal :: SomeException -> AgentErrorType
-mkInternal = INTERNAL . show
+mkInternal e = case fromException e of
+  Just BlockedIndefinitelyOnSTM -> CRITICAL True "Thread blocked indefinitely in STM transaction"
+  _ -> INTERNAL $ show e
 {-# INLINE mkInternal #-}
 
 data Worker = Worker
@@ -335,6 +356,8 @@ updateRestartCount t (RestartCount minute count) = do
 
 $(pure [])
 
+$(JQ.deriveJSON defaultJSON ''ServerRoles)
+
 instance ProtocolTypeI p => ToJSON (ServerCfg p) where
   toEncoding = $(JQ.mkToEncoding defaultJSON ''ServerCfg)
   toJSON = $(JQ.mkToJSON defaultJSON ''ServerCfg)

src/Simplex/Messaging/Agent/Protocol.hs

@@ -108,6 +108,7 @@ module Simplex.Messaging.Agent.Protocol
     ConnReqUriData (..),
     CRClientData,
     ServiceScheme,
+    sameConnReqContact,
     simplexChat,
     connReqUriP',
     AgentErrorType (..),
@@ -1263,6 +1264,12 @@ instance Eq AConnectionRequestUri where
 
 deriving instance Show AConnectionRequestUri
 
+sameConnReqContact :: ConnectionRequestUri 'CMContact -> ConnectionRequestUri 'CMContact -> Bool
+sameConnReqContact (CRContactUri ConnReqUriData {crSmpQueues = qs}) (CRContactUri ConnReqUriData {crSmpQueues = qs'}) =
+  L.length qs == L.length qs' && all same (L.zip qs qs')
+  where
+    same (q, q') = sameQAddress (qAddress q) (qAddress q')
+
 data ConnReqUriData = ConnReqUriData
   { crScheme :: ServiceScheme,
     crAgentVRange :: VersionRangeSMPA,

src/Simplex/Messaging/Server.hs

@@ -1821,9 +1821,9 @@ importMessages tty ms f old_ = do
               mergeQuotaMsgs >> writeMsg ms rId q False msg $> (stored, expired, M.insert rId q overQuota)
               where
                 -- if the first message in queue head is "quota", remove it.
-                mergeQuotaMsgs = withMsgQueue ms rId q "mergeQuotaMsgs" $ \mq ->
-                  tryPeekMsg_ mq >>= \case
-                    Just MessageQuota {} -> tryDeleteMsg_ q mq False
+                mergeQuotaMsgs =
+                  withPeekMsgQueue ms rId q "mergeQuotaMsgs" $ maybe (pure ()) $ \case
+                    (mq, MessageQuota {}) -> tryDeleteMsg_ q mq False
                     _ -> pure ()
         msgErr :: Show e => String -> e -> String
         msgErr op e = op <> " error (" <> show e <> "): " <> B.unpack (B.take 100 s)

src/Simplex/Messaging/Server/MsgStore/Journal.hs

@@ -11,8 +11,8 @@
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE StandaloneDeriving #-}
-{-# LANGUAGE TupleSections #-}
 {-# LANGUAGE TypeFamilies #-}
+{-# LANGUAGE TupleSections #-}
 
 module Simplex.Messaging.Server.MsgStore.Journal
   ( JournalMsgStore (queues, senders, notifiers, random),
@@ -20,7 +20,6 @@ module Simplex.Messaging.Server.MsgStore.Journal
     JournalMsgQueue (queue, state),
     JMQueue (queueDirectory, statePath),
     JournalStoreConfig (..),
-    getQueueMessages,
     closeMsgQueue,
     closeMsgQueueHandles,
     -- below are exported for tests
@@ -50,7 +49,7 @@ import qualified Data.ByteString.Char8 as B
 import Data.Functor (($>))
 import Data.Int (Int64)
 import Data.List (intercalate)
-import Data.Maybe (catMaybes, fromMaybe)
+import Data.Maybe (catMaybes, fromMaybe, isNothing)
 import qualified Data.Text as T
 import Data.Time.Clock (getCurrentTime)
 import Data.Time.Clock.System (SystemTime (..), getSystemTime)
@@ -105,7 +104,9 @@ data JournalQueue = JournalQueue
     queueRec :: TVar (Maybe QueueRec),
     msgQueue_ :: TVar (Maybe JournalMsgQueue),
     -- system time in seconds since epoch
-    activeAt :: TVar Int64
+    activeAt :: TVar Int64,
+    -- Just True - empty, Just False - non-empty, Nothing - unknown
+    isEmpty :: TVar (Maybe Bool)
   }
 
 data JMQueue = JMQueue
@@ -224,10 +225,11 @@ instance STMQueueStore JournalMsgStore where
   storeLog' = storeLog
   mkQueue st qr = do
     lock <- getMapLock (queueLocks st) $ recipientId qr
-    q <- newTVar $! Just qr
+    q <- newTVar $ Just qr
     mq <- newTVar Nothing
     activeAt <- newTVar 0
-    pure $ JournalQueue lock q mq activeAt
+    isEmpty <- newTVar Nothing
+    pure $ JournalQueue lock q mq activeAt isEmpty
   msgQueue_' = msgQueue_
 
 instance MsgStoreClass JournalMsgStore where
@@ -322,7 +324,7 @@ instance MsgStoreClass JournalMsgStore where
             statePath = msgQueueStatePath dir $ B.unpack (strEncode rId)
             queue = JMQueue {queueDirectory = dir, statePath}
         q <- ifM (doesDirectoryExist dir) (openMsgQueue ms queue) (createQ queue)
-        atomically $ writeTVar msgQueue_ $! Just q
+        atomically $ writeTVar msgQueue_ $ Just q
         pure q
         where
           createQ :: JMQueue -> IO JournalMsgQueue
@@ -332,14 +334,39 @@ instance MsgStoreClass JournalMsgStore where
             journalId <- newJournalId random
             mkJournalQueue queue (newMsgQueueState journalId) Nothing
 
+  getPeekMsgQueue :: JournalMsgStore -> RecipientId -> JournalQueue -> StoreIO (Maybe (JournalMsgQueue, Message))
+  getPeekMsgQueue ms rId q@JournalQueue {isEmpty} =
+    StoreIO (readTVarIO isEmpty) >>= \case
+      Just True -> pure Nothing
+      Just False -> peek
+      Nothing -> do
+        -- We only close the queue if we just learnt it's empty.
+        -- This is needed to reduce file descriptors and memory usage
+        -- after the server just started and many clients subscribe.
+        -- In case the queue became non-empty on write and then again empty on read
+        -- we won't be closing it, to avoid frequent open/close on active queues.
+        r <- peek
+        when (isNothing r) $ StoreIO $ closeMsgQueue q
+        pure r
+    where
+      peek = do
+        mq <- getMsgQueue ms rId q
+        (mq,) <$$> tryPeekMsg_ q mq
+
+  -- only runs action if queue is not empty
   withIdleMsgQueue :: Int64 -> JournalMsgStore -> RecipientId -> JournalQueue -> (JournalMsgQueue -> StoreIO a) -> StoreIO (Maybe a, Int)
   withIdleMsgQueue now ms@JournalMsgStore {config} rId q action =
     StoreIO $ readTVarIO (msgQueue_ q) >>= \case
       Nothing ->
-        E.bracket (unStoreIO $ getMsgQueue ms rId q) (\_ -> closeMsgQueue q) $ \mq -> unStoreIO $ do
-          r <- action mq
-          sz <- getQueueSize_ mq
-          pure (Just r, sz)
+        E.bracket
+          (unStoreIO $ getPeekMsgQueue ms rId q)
+          (mapM_ $ \_ -> closeMsgQueue q)
+          (maybe (pure (Nothing, 0)) (unStoreIO . run))
+        where
+          run (mq, _) = do
+            r <- action mq
+            sz <- getQueueSize_ mq
+            pure (Just r, sz)
       Just mq -> do
         ts <- readTVarIO $ activeAt q
         r <- if now - ts >= idleInterval config
@@ -378,6 +405,7 @@ instance MsgStoreClass JournalMsgStore where
       let empty = size == 0
       if canWrite || empty
         then do
+          atomically $ writeTVar (isEmpty q') (Just False)
           let canWrt' = quota > size
           if canWrt'
             then writeToJournal q st canWrt' msg $> Just (msg, empty)
@@ -426,9 +454,9 @@ instance MsgStoreClass JournalMsgStore where
   getQueueSize_ :: JournalMsgQueue -> StoreIO Int
   getQueueSize_ JournalMsgQueue {state} = StoreIO $ size <$> readTVarIO state
 
-  tryPeekMsg_ :: JournalMsgQueue -> StoreIO (Maybe Message)
-  tryPeekMsg_ q@JournalMsgQueue {tipMsg, handles} =
-    StoreIO $ readTVarIO handles $>>= chooseReadJournal q True $>>= peekMsg
+  tryPeekMsg_ :: JournalQueue -> JournalMsgQueue -> StoreIO (Maybe Message)
+  tryPeekMsg_ q mq@JournalMsgQueue {tipMsg, handles} =
+    StoreIO $ (readTVarIO handles $>>= chooseReadJournal mq True $>>= peekMsg) >>= setEmpty
     where
       peekMsg (rs, h) = readTVarIO tipMsg >>= maybe readMsg (pure . fmap fst)
         where
@@ -436,6 +464,9 @@ instance MsgStoreClass JournalMsgStore where
             ml@(msg, _) <- hGetMsgAt h $ bytePos rs
             atomically $ writeTVar tipMsg $ Just (Just ml)
             pure $ Just msg
+      setEmpty msg = do
+        atomically $ writeTVar (isEmpty q) (Just $ isNothing msg)
+        pure msg
 
   tryDeleteMsg_ :: JournalQueue -> JournalMsgQueue -> Bool -> StoreIO ()
   tryDeleteMsg_ q mq@JournalMsgQueue {tipMsg, handles} logState = StoreIO $ (`E.finally` when logState (updateActiveAt q)) $

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

@@ -1,4 +1,3 @@
-{-# LANGUAGE BangPatterns #-}
 {-# LANGUAGE ConstraintKinds #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
@@ -8,8 +7,8 @@
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE NamedFieldPuns #-}
-{-# LANGUAGE TupleSections #-}
 {-# LANGUAGE TypeFamilies #-}
+{-# LANGUAGE TupleSections #-}
 
 module Simplex.Messaging.Server.MsgStore.STM
   ( STMMsgStore (..),
@@ -29,7 +28,7 @@ import Simplex.Messaging.Server.QueueStore.STM
 import Simplex.Messaging.Server.StoreLog
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
-import Simplex.Messaging.Util ((<$$>))
+import Simplex.Messaging.Util ((<$$>), ($>>=))
 import System.IO (IOMode (..))
 
 data STMMsgStore = STMMsgStore
@@ -63,7 +62,7 @@ instance STMQueueStore STMMsgStore where
   senders' = senders
   notifiers' = notifiers
   storeLog' = storeLog
-  mkQueue _ qr = STMQueue <$> (newTVar $! Just qr) <*> newTVar Nothing
+  mkQueue _ qr = STMQueue <$> newTVar (Just qr) <*> newTVar Nothing
   msgQueue_' = msgQueue_
 
 instance MsgStoreClass STMMsgStore where
@@ -106,9 +105,12 @@ instance MsgStoreClass STMMsgStore where
         canWrite <- newTVar True
         size <- newTVar 0
         let q = STMMsgQueue {msgQueue, canWrite, size}
-        writeTVar msgQueue_ $! Just q
+        writeTVar msgQueue_ (Just q)
         pure q
 
+  getPeekMsgQueue :: STMMsgStore -> RecipientId -> STMQueue -> STM (Maybe (STMMsgQueue, Message))
+  getPeekMsgQueue _ _ q@STMQueue {msgQueue_} = readTVar msgQueue_ $>>= \mq -> (mq,) <$$> tryPeekMsg_ q mq
+
   -- does not create queue if it does not exist, does not delete it if it does (can't just close in-memory queue)
   withIdleMsgQueue :: Int64 -> STMMsgStore -> RecipientId -> STMQueue -> (STMMsgQueue -> STM a) -> STM (Maybe a, Int)
   withIdleMsgQueue _ _ _ STMQueue {msgQueue_} action = readTVar msgQueue_ >>= \case
@@ -159,8 +161,8 @@ instance MsgStoreClass STMMsgStore where
   getQueueSize_ :: STMMsgQueue -> STM Int
   getQueueSize_ STMMsgQueue {size} = readTVar size
 
-  tryPeekMsg_ :: STMMsgQueue -> STM (Maybe Message)
-  tryPeekMsg_ = tryPeekTQueue . msgQueue
+  tryPeekMsg_ :: STMQueue -> STMMsgQueue -> STM (Maybe Message)
+  tryPeekMsg_ _ = tryPeekTQueue . msgQueue
   {-# INLINE tryPeekMsg_ #-}
 
   tryDeleteMsg_ :: STMQueue -> STMMsgQueue -> Bool -> STM ()

src/Simplex/Messaging/Server/MsgStore/Types.hs

@@ -4,15 +4,20 @@
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE MultiWayIf #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE TupleSections #-}
 {-# LANGUAGE TypeFamilyDependencies #-}
+{-# OPTIONS_GHC -Wno-unrecognised-pragmas #-}
+
+{-# HLINT ignore "Redundant multi-way if" #-}
 
 module Simplex.Messaging.Server.MsgStore.Types where
 
 import Control.Concurrent.STM
 import Control.Monad (foldM)
 import Control.Monad.Trans.Except
+import Data.Functor (($>))
 import Data.Int (Int64)
 import Data.Kind
 import qualified Data.Map.Strict as M
@@ -21,6 +26,7 @@ import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.QueueStore
 import Simplex.Messaging.Server.StoreLog.Types
 import Simplex.Messaging.TMap (TMap)
+import Simplex.Messaging.Util ((<$$>))
 import System.IO (IOMode (..))
 
 class MsgStoreClass s => STMQueueStore s where
@@ -44,7 +50,9 @@ class Monad (StoreMonad s) => MsgStoreClass s where
   logQueueStates :: s -> IO ()
   logQueueState :: StoreQueue s -> StoreMonad s ()
   queueRec' :: StoreQueue s -> TVar (Maybe QueueRec)
+  getPeekMsgQueue :: s -> RecipientId -> StoreQueue s -> StoreMonad s (Maybe (MsgQueue s, Message))
   getMsgQueue :: s -> RecipientId -> StoreQueue s -> StoreMonad s (MsgQueue s)
+
   -- the journal queue will be closed after action if it was initially closed or idle longer than interval in config
   withIdleMsgQueue :: Int64 -> s -> RecipientId -> StoreQueue s -> (MsgQueue s -> StoreMonad s a) -> StoreMonad s (Maybe a, Int)
   deleteQueue :: s -> RecipientId -> StoreQueue s -> IO (Either ErrorType QueueRec)
@@ -53,7 +61,7 @@ class Monad (StoreMonad s) => MsgStoreClass s where
   writeMsg :: s -> RecipientId -> StoreQueue s -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
   setOverQuota_ :: StoreQueue s -> IO () -- can ONLY be used while restoring messages, not while server running
   getQueueSize_ :: MsgQueue s -> StoreMonad s Int
-  tryPeekMsg_ :: MsgQueue s -> StoreMonad s (Maybe Message)
+  tryPeekMsg_ :: StoreQueue s -> MsgQueue s -> StoreMonad s (Maybe Message)
   tryDeleteMsg_ :: StoreQueue s -> MsgQueue s -> Bool -> StoreMonad s ()
   isolateQueue :: RecipientId -> StoreQueue s -> String -> StoreMonad s a -> ExceptT ErrorType IO a
 
@@ -73,39 +81,39 @@ withActiveMsgQueues st f = readTVarIO (activeMsgQueues st) >>= foldM run mempty
       pure $! acc <> r
 
 getQueueMessages :: MsgStoreClass s => Bool -> s -> RecipientId -> StoreQueue s -> ExceptT ErrorType IO [Message]
-getQueueMessages drainMsgs st rId q = withMsgQueue st rId q "getQueueSize" $ getQueueMessages_ drainMsgs
+getQueueMessages drainMsgs st rId q = withPeekMsgQueue st rId q "getQueueSize" $ maybe (pure []) (getQueueMessages_ drainMsgs . fst)
 {-# INLINE getQueueMessages #-}
 
 getQueueSize :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> ExceptT ErrorType IO Int
-getQueueSize st rId q = withMsgQueue st rId q "getQueueSize" $ getQueueSize_
+getQueueSize st rId q = withPeekMsgQueue st rId q "getQueueSize" $ maybe (pure 0) (getQueueSize_ . fst)
 {-# INLINE getQueueSize #-}
 
 tryPeekMsg :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> ExceptT ErrorType IO (Maybe Message)
-tryPeekMsg st rId q = withMsgQueue st rId q "tryPeekMsg" $ tryPeekMsg_
+tryPeekMsg st rId q = snd <$$> withPeekMsgQueue st rId q "tryPeekMsg" pure
 {-# INLINE tryPeekMsg #-}
 
 tryDelMsg :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> MsgId -> ExceptT ErrorType IO (Maybe Message)
 tryDelMsg st rId q msgId' =
-  withMsgQueue st rId q "tryDelMsg" $ \mq ->
-    tryPeekMsg_ mq >>= \case
-      msg_@(Just msg)
+  withPeekMsgQueue st rId q "tryDelMsg" $
+    maybe (pure Nothing) $ \(mq, msg) ->
+      if
         | messageId msg == msgId' ->
-            tryDeleteMsg_ q mq True >> pure msg_
-      _ -> pure Nothing
+            tryDeleteMsg_ q mq True $> Just msg
+        | otherwise -> pure Nothing
 
 -- atomic delete (== read) last and peek next message if available
 tryDelPeekMsg :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> MsgId -> ExceptT ErrorType IO (Maybe Message, Maybe Message)
 tryDelPeekMsg st rId q msgId' =
-  withMsgQueue st rId q "tryDelPeekMsg" $ \mq ->
-    tryPeekMsg_ mq >>= \case
-      msg_@(Just msg)
-        | messageId msg == msgId' -> (msg_,) <$> (tryDeleteMsg_ q mq True >> tryPeekMsg_ mq)
-        | otherwise -> pure (Nothing, msg_)
-      _ -> pure (Nothing, Nothing)
+  withPeekMsgQueue st rId q "tryDelPeekMsg" $
+    maybe (pure (Nothing, Nothing)) $ \(mq, msg) ->
+      if
+        | messageId msg == msgId' -> (Just msg,) <$> (tryDeleteMsg_ q mq True >> tryPeekMsg_ q mq)
+        | otherwise -> pure (Nothing, Just msg)
 
-withMsgQueue :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> String -> (MsgQueue s -> StoreMonad s a) -> ExceptT ErrorType IO a
-withMsgQueue st rId q op a = isolateQueue rId q op $ getMsgQueue st rId q >>= a
-{-# INLINE withMsgQueue #-}
+-- The action is called with Nothing when it is known that the queue is empty
+withPeekMsgQueue :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> String -> (Maybe (MsgQueue s, Message) -> StoreMonad s a) -> ExceptT ErrorType IO a
+withPeekMsgQueue st rId q op a = isolateQueue rId q op $ getPeekMsgQueue st rId q >>= a
+{-# INLINE withPeekMsgQueue #-}
 
 deleteExpiredMsgs :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> Int64 -> ExceptT ErrorType IO Int
 deleteExpiredMsgs st rId q old =
@@ -126,7 +134,7 @@ deleteExpireMsgs_ old q mq = do
   pure n
   where
     loop dc =
-      tryPeekMsg_ mq >>= \case
+      tryPeekMsg_ q mq >>= \case
         Just Message {msgTs}
           | systemSeconds msgTs < old ->
               tryDeleteMsg_ q mq False >> loop (dc + 1)

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

@@ -84,7 +84,7 @@ secureQueue st sq sKey =
     secure q@QueueRec {recipientId = rId} = case senderKey q of
       Just k -> pure $ if sKey == k then Right rId else Left AUTH
       Nothing -> do
-        writeTVar qr $! Just q {senderKey = Just sKey}
+        writeTVar qr $ Just q {senderKey = Just sKey}
         pure $ Right rId
 
 addQueueNotifier :: STMQueueStore s => s -> StoreQueue s -> NtfCreds -> IO (Either ErrorType (Maybe NotifierId))
@@ -96,7 +96,7 @@ addQueueNotifier st sq ntfCreds@NtfCreds {notifierId = nId} =
     add q@QueueRec {recipientId = rId} = ifM (TM.member nId (notifiers' st)) (pure $ Left DUPLICATE_) $ do
       nId_ <- forM (notifier q) $ \NtfCreds {notifierId} -> TM.delete notifierId (notifiers' st) $> notifierId
       let !q' = q {notifier = Just ntfCreds}
-      writeTVar qr $! Just q'
+      writeTVar qr $ Just q'
       TM.insert nId rId $ notifiers' st
       pure $ Right (rId, nId_)
 

tests/AgentTests.hs

@@ -16,6 +16,7 @@ import AgentTests.FunctionalAPITests (functionalAPITests)
 import AgentTests.MigrationTests (migrationTests)
 import AgentTests.NotificationTests (notificationTests)
 import AgentTests.SQLiteTests (storeTests)
+import AgentTests.ServerChoice (serverChoiceTests)
 import Simplex.Messaging.Transport (ATransport (..))
 import Test.Hspec
 
@@ -26,4 +27,5 @@ agentTests (ATransport t) = do
   describe "Functional API" $ functionalAPITests (ATransport t)
   describe "Notification tests" $ notificationTests (ATransport t)
   describe "SQLite store" storeTests
+  describe "Chosen servers" serverChoiceTests
   describe "Migration tests" migrationTests

tests/AgentTests/ServerChoice.hs

@@ -0,0 +1,85 @@
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE OverloadedLists #-}
+{-# LANGUAGE OverloadedStrings #-}
+
+module AgentTests.ServerChoice where
+
+import AgentTests.FunctionalAPITests
+import Control.Monad.IO.Class
+import Data.List.NonEmpty (NonEmpty (..))
+import qualified Data.Map.Strict as M
+import SMPAgentClient
+import Simplex.Messaging.Agent (withAgentEnv)
+import Simplex.Messaging.Agent.Client hiding (userServers)
+import Simplex.Messaging.Agent.Env.SQLite
+import Simplex.Messaging.Client (defaultNetworkConfig)
+import Simplex.Messaging.Protocol
+import Test.Hspec
+import Test.QuickCheck
+import XFTPClient (testXFTPServer)
+
+serverChoiceTests :: Spec
+serverChoiceTests = do
+  describe "Server operators" $ do
+    it "should choose server of different operator" $ ioProperty $ testChooseDifferentOperator
+
+operatorSimpleX :: Maybe OperatorId
+operatorSimpleX = Just 1
+
+operator2 :: Maybe OperatorId
+operator2 = Just 2
+
+testOp1Srv1 :: ProtoServerWithAuth 'PSMP
+testOp1Srv1 = "smp://LcJU@test1.simplex.im"
+
+testOp1Srv2 :: ProtoServerWithAuth 'PSMP
+testOp1Srv2 = "smp://LcJU@test2.simplex.im"
+
+testOp2Srv1 :: ProtoServerWithAuth 'PSMP
+testOp2Srv1 = "smp://LcJU@srv1.example.com"
+
+testOp2Srv2 :: ProtoServerWithAuth 'PSMP
+testOp2Srv2 = "smp://LcJU@srv2.example.com"
+
+testSMPServers :: NonEmpty (ServerCfg 'PSMP)
+testSMPServers =
+  [ presetServerCfg True allRoles operatorSimpleX testOp1Srv1,
+    presetServerCfg True allRoles operatorSimpleX testOp1Srv2,
+    presetServerCfg True proxyOnly operator2 testOp2Srv1,
+    presetServerCfg True proxyOnly operator2 testOp2Srv2
+  ]
+
+storageOnly :: ServerRoles
+storageOnly = ServerRoles {storage = True, proxy = False}
+
+proxyOnly :: ServerRoles
+proxyOnly = ServerRoles {storage = False, proxy = True}
+
+initServers :: InitialAgentServers
+initServers =
+  InitialAgentServers
+    { smp = M.fromList [(1, testSMPServers)],
+      ntf = [testNtfServer],
+      xftp = userServers [testXFTPServer],
+      netCfg = defaultNetworkConfig
+    }
+
+testChooseDifferentOperator :: IO ()
+testChooseDifferentOperator = do
+  c <- getSMPAgentClient' 1 agentCfg initServers testDB
+  runRight_ $ do
+    -- chooses the only operator with storage role
+    srv1 <- withAgentEnv c $ getNextServer c 1 storageSrvs []
+    liftIO $ srv1 == testOp1Srv1 || srv1 == testOp1Srv2 `shouldBe` True
+    -- chooses another server for storage
+    srv2 <- withAgentEnv c $ getNextServer c 1 storageSrvs [protoServer testOp1Srv1]
+    liftIO $ srv2 `shouldBe` testOp1Srv2
+    -- chooses another operator for proxy
+    srv3 <- withAgentEnv c $ getNextServer c 1 proxySrvs [protoServer srv1]
+    liftIO $ srv3 == testOp2Srv1 || srv3 == testOp2Srv2 `shouldBe` True
+    -- chooses another operator for proxy
+    srv3' <- withAgentEnv c $ getNextServer c 1 proxySrvs [protoServer testOp1Srv1, protoServer testOp1Srv2]
+    liftIO $ srv3' == testOp2Srv1 || srv3' == testOp2Srv2 `shouldBe` True
+    -- chooses any other server
+    srv4 <- withAgentEnv c $ getNextServer c 1 proxySrvs [protoServer testOp1Srv1, protoServer testOp2Srv1]
+    liftIO $ srv4 == testOp1Srv2 || srv4 == testOp2Srv2 `shouldBe` True

tests/SMPAgentClient.hs

@@ -38,7 +38,7 @@ testSMPServer :: SMPServer
 testSMPServer = "smp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:5001"
 
 testSMPServer2 :: SMPServer
-testSMPServer2 = "smp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:5002"
+testSMPServer2 = "smp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@127.0.0.1:5002"
 
 testNtfServer :: NtfServer
 testNtfServer = "ntf://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:6001"
@@ -92,7 +92,7 @@ userServers :: NonEmpty (ProtocolServer p) -> Map UserId (NonEmpty (ServerCfg p)
 userServers = userServers' . L.map noAuthSrv
 
 userServers' :: NonEmpty (ProtoServerWithAuth p) -> Map UserId (NonEmpty (ServerCfg p))
-userServers' srvs = M.fromList [(1, L.map (presetServerCfg True) srvs)]
+userServers' srvs = M.fromList [(1, L.map (presetServerCfg True (ServerRoles True True) (Just 1)) srvs)]
 
 noAuthSrvCfg :: ProtocolServer p -> ServerCfg p
-noAuthSrvCfg = presetServerCfg True . noAuthSrv
+noAuthSrvCfg = presetServerCfg True (ServerRoles True True) (Just 1) . noAuthSrv

tests/SMPClient.hs

@@ -43,6 +43,9 @@ import Util
 testHost :: NonEmpty TransportHost
 testHost = "localhost"
 
+testHost2 :: NonEmpty TransportHost
+testHost2 = "127.0.0.1"
+
 testPort :: ServiceName
 testPort = "5001"
 

tests/SMPProxyTests.hs

@@ -61,7 +61,7 @@ smpProxyTests = do
     xit "batching proxy requests" todo
   describe "deliver message via SMP proxy" $ do
     let srv1 = SMPServer testHost testPort testKeyHash
-        srv2 = SMPServer testHost testPort2 testKeyHash
+        srv2 = SMPServer testHost2 testPort2 testKeyHash
     describe "client API" $ do
       let maxLen = maxMessageLength encryptedBlockSMPVersion
       describe "one server" $ do
@@ -316,7 +316,7 @@ agentViaProxyVersionError :: IO ()
 agentViaProxyVersionError =
   withAgent 1 agentCfg (servers [SMPServer testHost testPort testKeyHash]) testDB $ \alice -> do
     Left (A.BROKER _ (TRANSPORT TEVersion)) <-
-      withAgent 2 agentCfg (servers [SMPServer testHost testPort2 testKeyHash]) testDB2 $ \bob -> runExceptT $ do
+      withAgent 2 agentCfg (servers [SMPServer testHost2 testPort2 testKeyHash]) testDB2 $ \bob -> runExceptT $ do
         (_bobId, qInfo) <- A.createConnection alice 1 True SCMInvitation Nothing (CR.IKNoPQ PQSupportOn) SMSubscribe
         aliceId <- A.prepareConnectionToJoin bob 1 True qInfo PQSupportOn
         A.joinConnection bob 1 aliceId True qInfo "bob's connInfo" PQSupportOn SMSubscribe