Merge branch 'master' into ep/disable-background

build: move type to client_library module
version
2026-10-12 02:22:45 +00:00 · 2024-12-20 13:22:04 +00:00 · 2024-12-20 13:12:11 +00:00 · 2024-12-20 12:12:32 +00:00 · 2024-12-20 12:10:18 +00:00 · 2024-12-19 20:45:17 +00:00
35 changed files with 1125 additions and 284 deletions
@@ -1,3 +1,19 @@
+# 6.2.0
+
+Version 6.2.0.7
+
+Build:
+- client_library flag to build only used modules in the clients, remove package yaml
+
+SMP server:
+- journal storage for messages (BETA).
+- prevent race condition when deleting queue and to avoid "orphan" messages (#1395).
+
+SMP agent:
+- support SMP and XFTP server roles (storage/proxy) and operators (#1343).
+- treat blocked STM and other critical errors that offer restart as temporary for message delivery (#1405).
+- fix inconsistent state after app restart while accepting contact request (#1412).
+
 # 6.1.3

 SMP server: fix restoring notification credentials.
@@ -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
@@ -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.
@@ -1,7 +1,7 @@
 cabal-version: 1.12

 name:           simplexmq
-version:        6.2.0.2
+version:        6.2.1.100
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                <./docs/Simplex-Messaging-Client.html client> and
@@ -73,7 +73,6 @@ library
      Simplex.FileTransfer.Chunks
      Simplex.FileTransfer.Client
      Simplex.FileTransfer.Client.Agent
-      Simplex.FileTransfer.Client.Main
      Simplex.FileTransfer.Client.Presets
      Simplex.FileTransfer.Crypto
      Simplex.FileTransfer.Description
@@ -152,7 +151,6 @@ library
      Simplex.Messaging.Notifications.Types
      Simplex.Messaging.Parsers
      Simplex.Messaging.Protocol
-      Simplex.Messaging.Server.CLI
      Simplex.Messaging.Server.Expiration
      Simplex.Messaging.Server.QueueStore.QueueInfo
      Simplex.Messaging.ServiceScheme
@@ -168,7 +166,6 @@ library
      Simplex.Messaging.Transport.HTTP2.Server
      Simplex.Messaging.Transport.KeepAlive
      Simplex.Messaging.Transport.Server
-      Simplex.Messaging.Transport.WebSockets
      Simplex.Messaging.Util
      Simplex.Messaging.Version
      Simplex.Messaging.Version.Internal
@@ -179,6 +176,7 @@ library
      Simplex.RemoteControl.Types
  if !flag(client_library)
      exposed-modules:
+          Simplex.FileTransfer.Client.Main
          Simplex.FileTransfer.Server
          Simplex.FileTransfer.Server.Control
          Simplex.FileTransfer.Server.Env
@@ -186,21 +184,6 @@ library
          Simplex.FileTransfer.Server.Stats
          Simplex.FileTransfer.Server.Store
          Simplex.FileTransfer.Server.StoreLog
-          Simplex.Messaging.Server
-          Simplex.Messaging.Server.Control
-          Simplex.Messaging.Server.Env.STM
-          Simplex.Messaging.Server.Information
-          Simplex.Messaging.Server.Main
-          Simplex.Messaging.Server.MsgStore
-          Simplex.Messaging.Server.MsgStore.Journal
-          Simplex.Messaging.Server.MsgStore.STM
-          Simplex.Messaging.Server.MsgStore.Types
-          Simplex.Messaging.Server.NtfStore
-          Simplex.Messaging.Server.QueueStore
-          Simplex.Messaging.Server.QueueStore.STM
-          Simplex.Messaging.Server.Stats
-          Simplex.Messaging.Server.StoreLog
-          Simplex.Messaging.Server.StoreLog.Types
          Simplex.Messaging.Notifications.Server
          Simplex.Messaging.Notifications.Server.Control
          Simplex.Messaging.Notifications.Server.Env
@@ -210,6 +193,24 @@ library
          Simplex.Messaging.Notifications.Server.Stats
          Simplex.Messaging.Notifications.Server.Store
          Simplex.Messaging.Notifications.Server.StoreLog
+          Simplex.Messaging.Server
+          Simplex.Messaging.Server.CLI
+          Simplex.Messaging.Server.Control
+          Simplex.Messaging.Server.Env.STM
+          Simplex.Messaging.Server.Information
+          Simplex.Messaging.Server.Main
+          Simplex.Messaging.Server.MsgStore
+          Simplex.Messaging.Server.MsgStore.Journal
+          Simplex.Messaging.Server.MsgStore.STM
+          Simplex.Messaging.Server.MsgStore.Types
+          Simplex.Messaging.Server.NtfStore
+          Simplex.Messaging.Server.Prometheus
+          Simplex.Messaging.Server.QueueStore
+          Simplex.Messaging.Server.QueueStore.STM
+          Simplex.Messaging.Server.Stats
+          Simplex.Messaging.Server.StoreLog
+          Simplex.Messaging.Server.StoreLog.Types
+          Simplex.Messaging.Transport.WebSockets
  other-modules:
      Paths_simplexmq
  hs-source-dirs:
@@ -270,7 +271,6 @@ library
    , transformers ==0.6.*
    , unliftio ==0.2.*
    , unliftio-core ==0.2.*
-    , websockets ==0.12.*
    , yaml ==0.11.*
    , zstd ==0.1.3.*
  default-language: Haskell2010
@@ -280,6 +280,7 @@ library
    build-depends:
        case-insensitive ==1.2.*
      , hashable ==1.4.*
+      , websockets ==0.12.*
  if impl(ghc >= 9.6.2)
    build-depends:
        bytestring ==0.11.*
@@ -384,6 +385,7 @@ test-suite simplexmq-test
      AgentTests.MigrationTests
      AgentTests.NotificationTests
      AgentTests.SchemaDump
+      AgentTests.ServerChoice
      AgentTests.SQLiteTests
      CLITests
      CoreTests.BatchingTests
@@ -51,8 +51,7 @@ import Data.Text (Text)
 import Data.Time.Clock (getCurrentTime)
 import Data.Time.Format (defaultTimeLocale, formatTime)
 import Simplex.FileTransfer.Chunks (toKB)
-import Simplex.FileTransfer.Client (XFTPChunkSpec (..))
-import Simplex.FileTransfer.Client.Main
+import Simplex.FileTransfer.Client (XFTPChunkSpec (..), getChunkDigest, prepareChunkSizes, prepareChunkSpecs, singleChunkSize)
 import Simplex.FileTransfer.Crypto
 import Simplex.FileTransfer.Description
 import Simplex.FileTransfer.Protocol (FileParty (..), SFileParty (..))
@@ -461,14 +460,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 +476,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 +545,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
@@ -20,14 +20,18 @@ import Data.Bifunctor (first)
 import Data.ByteString.Builder (Builder, byteString)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
+import qualified Data.ByteString.Lazy as LB
 import Data.Int (Int64)
+import Data.List (foldl')
 import Data.List.NonEmpty (NonEmpty (..))
+import Data.Maybe (listToMaybe)
 import Data.Time.Clock (UTCTime)
 import Data.Word (Word32)
 import qualified Data.X509 as X
 import qualified Data.X509.Validation as XV
 import qualified Network.HTTP.Types as N
 import qualified Network.HTTP2.Client as H
+import Simplex.FileTransfer.Chunks
 import Simplex.FileTransfer.Protocol
 import Simplex.FileTransfer.Transport
 import Simplex.Messaging.Client
@@ -298,3 +302,41 @@ noFile HTTP2Body {bodyPart} a = case bodyPart of

 -- FACK :: FileCommand Recipient
 -- PING :: FileCommand Recipient
+
+singleChunkSize :: Int64 -> Maybe Word32
+singleChunkSize size' =
+  listToMaybe $ dropWhile (< chunkSize) serverChunkSizes
+  where
+    chunkSize = fromIntegral size'
+
+prepareChunkSizes :: Int64 -> [Word32]
+prepareChunkSizes size' = prepareSizes size'
+  where
+    (smallSize, bigSize)
+      | size' > size34 chunkSize3 = (chunkSize2, chunkSize3)
+      | size' > size34 chunkSize2 = (chunkSize1, chunkSize2)
+      | otherwise = (chunkSize0, chunkSize1)
+    size34 sz = (fromIntegral sz * 3) `div` 4
+    prepareSizes 0 = []
+    prepareSizes size
+      | size >= fromIntegral bigSize = replicate (fromIntegral n1) bigSize <> prepareSizes remSz
+      | size > size34 bigSize = [bigSize]
+      | otherwise = replicate (fromIntegral n2') smallSize
+      where
+        (n1, remSz) = size `divMod` fromIntegral bigSize
+        n2' = let (n2, remSz2) = (size `divMod` fromIntegral smallSize) in if remSz2 == 0 then n2 else n2 + 1
+
+prepareChunkSpecs :: FilePath -> [Word32] -> [XFTPChunkSpec]
+prepareChunkSpecs filePath chunkSizes = reverse . snd $ foldl' addSpec (0, []) chunkSizes
+  where
+    addSpec :: (Int64, [XFTPChunkSpec]) -> Word32 -> (Int64, [XFTPChunkSpec])
+    addSpec (chunkOffset, specs) sz =
+      let spec = XFTPChunkSpec {filePath, chunkOffset, chunkSize = sz}
+       in (chunkOffset + fromIntegral sz, spec : specs)
+
+getChunkDigest :: XFTPChunkSpec -> IO ByteString
+getChunkDigest XFTPChunkSpec {filePath = chunkPath, chunkOffset, chunkSize} =
+  withFile chunkPath ReadMode $ \h -> do
+    hSeek h AbsoluteSeek $ fromIntegral chunkOffset
+    chunk <- LB.hGet h (fromIntegral chunkSize)
+    pure $! LC.sha256Hash chunk
@@ -19,11 +19,7 @@ module Simplex.FileTransfer.Client.Main
    singleChunkSize,
    prepareChunkSizes,
    prepareChunkSpecs,
-    maxFileSize,
-    maxFileSizeHard,
-    fileSizeLen,
    getChunkDigest,
-    SentRecipientReplica (..),
  )
 where

@@ -34,7 +30,6 @@ import Control.Monad.Trans.Except
 import Crypto.Random (ChaChaDRG)
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.Bifunctor (first)
-import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy.Char8 as LB
 import Data.Char (toLower)
@@ -45,7 +40,7 @@ import Data.List.NonEmpty (NonEmpty (..), nonEmpty)
 import qualified Data.List.NonEmpty as L
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
-import Data.Maybe (fromMaybe, listToMaybe)
+import Data.Maybe (fromMaybe)
 import qualified Data.Text as T
 import Data.Word (Word32)
 import GHC.Records (HasField (getField))
@@ -80,20 +75,6 @@ import UnliftIO.Directory
 xftpClientVersion :: String
 xftpClientVersion = "1.0.1"

-- | Soft limit for XFTP clients. Should be checked and reported to user.
-maxFileSize :: Int64
-maxFileSize = gb 1
-
-maxFileSizeStr :: String
-maxFileSizeStr = B.unpack . strEncode $ FileSize maxFileSize
-
-- | Hard internal limit for XFTP agent after which it refuses to prepare chunks.
-maxFileSizeHard :: Int64
-maxFileSizeHard = gb 5
-
-fileSizeLen :: Int64
-fileSizeLen = 8
-
 newtype CLIError = CLIError String
  deriving (Eq, Show, Exception)

@@ -231,16 +212,6 @@ data SentFileChunkReplica = SentFileChunkReplica
  }
  deriving (Show)

-data SentRecipientReplica = SentRecipientReplica
-  { chunkNo :: Int,
-    server :: XFTPServer,
-    rcvNo :: Int,
-    replicaId :: ChunkReplicaId,
-    replicaKey :: C.APrivateAuthKey,
-    digest :: FileDigest,
-    chunkSize :: FileSize Word32
-  }
-
 logCfg :: LogConfig
 logCfg = LogConfig {lc_file = Nothing, lc_stderr = True}

@@ -414,13 +385,6 @@ cliSendFileOpts SendOptions {filePath, outputDir, numRecipients, xftpServers, re
      B.writeFile fdSndPath $ strEncode fdSnd
      pure (fdRcvPaths, fdSndPath)

-getChunkDigest :: XFTPChunkSpec -> IO ByteString
-getChunkDigest XFTPChunkSpec {filePath = chunkPath, chunkOffset, chunkSize} =
-  withFile chunkPath ReadMode $ \h -> do
-    hSeek h AbsoluteSeek $ fromIntegral chunkOffset
-    chunk <- LB.hGet h (fromIntegral chunkSize)
-    pure $! LC.sha256Hash chunk
-
 cliReceiveFile :: ReceiveOptions -> ExceptT CLIError IO ()
 cliReceiveFile ReceiveOptions {fileDescription, filePath, retryCount, tempPath, verbose, yes} =
  getFileDescription' fileDescription >>= receive
@@ -536,37 +500,6 @@ getFileDescription' path =
  getFileDescription path >>= \case
    AVFD fd -> either (throwE . CLIError) pure $ checkParty fd

-singleChunkSize :: Int64 -> Maybe Word32
-singleChunkSize size' =
-  listToMaybe $ dropWhile (< chunkSize) serverChunkSizes
-  where
-    chunkSize = fromIntegral size'
-
-prepareChunkSizes :: Int64 -> [Word32]
-prepareChunkSizes size' = prepareSizes size'
-  where
-    (smallSize, bigSize)
-      | size' > size34 chunkSize3 = (chunkSize2, chunkSize3)
-      | size' > size34 chunkSize2 = (chunkSize1, chunkSize2)
-      | otherwise = (chunkSize0, chunkSize1)
-    size34 sz = (fromIntegral sz * 3) `div` 4
-    prepareSizes 0 = []
-    prepareSizes size
-      | size >= fromIntegral bigSize = replicate (fromIntegral n1) bigSize <> prepareSizes remSz
-      | size > size34 bigSize = [bigSize]
-      | otherwise = replicate (fromIntegral n2') smallSize
-      where
-        (n1, remSz) = size `divMod` fromIntegral bigSize
-        n2' = let (n2, remSz2) = (size `divMod` fromIntegral smallSize) in if remSz2 == 0 then n2 else n2 + 1
-
-prepareChunkSpecs :: FilePath -> [Word32] -> [XFTPChunkSpec]
-prepareChunkSpecs filePath chunkSizes = reverse . snd $ foldl' addSpec (0, []) chunkSizes
-  where
-    addSpec :: (Int64, [XFTPChunkSpec]) -> Word32 -> (Int64, [XFTPChunkSpec])
-    addSpec (chunkOffset, specs) sz =
-      let spec = XFTPChunkSpec {filePath, chunkOffset, chunkSize = sz}
-       in (chunkOffset + fromIntegral sz, spec : specs)
-
 getEncPath :: MonadIO m => Maybe FilePath -> String -> m FilePath
 getEncPath path name = (`uniqueCombine` (name <> ".encrypted")) =<< maybe (liftIO getCanonicalTemporaryDirectory) pure path

@@ -37,6 +37,10 @@ module Simplex.FileTransfer.Description
    FileClientData,
    fileDescriptionURI,
    qrSizeLimit,
+    maxFileSize,
+    maxFileSizeStr,
+    maxFileSizeHard,
+    fileSizeLen,
  )
 where

@@ -266,6 +270,21 @@ instance StrEncoding FileDescriptionURI where
 qrSizeLimit :: Int
 qrSizeLimit = 1002 -- ~2 chunks in URLencoded YAML with some spare size for server hosts

+-- | Soft limit for XFTP clients. Should be checked and reported to user.
+maxFileSize :: Int64
+maxFileSize = gb 1
+
+maxFileSizeStr :: String
+maxFileSizeStr = B.unpack . strEncode $ FileSize maxFileSize
+
+-- | Hard internal limit for XFTP agent after which it refuses to prepare chunks.
+maxFileSizeHard :: Int64
+maxFileSizeHard = gb 5
+
+fileSizeLen :: Int64
+fileSizeLen = 8
+
+
 instance (Integral a, Show a) => StrEncoding (FileSize a) where
  strEncode (FileSize b)
    | b' /= 0 = bshow b
@@ -246,6 +246,16 @@ data DeletedSndChunkReplica = DeletedSndChunkReplica
  }
  deriving (Show)

+data SentRecipientReplica = SentRecipientReplica
+  { chunkNo :: Int,
+    server :: XFTPServer,
+    rcvNo :: Int,
+    replicaId :: ChunkReplicaId,
+    replicaKey :: C.APrivateAuthKey,
+    digest :: FileDigest,
+    chunkSize :: FileSize Word32
+  }
+
 data FileErrorType
  = -- | cannot proceed with download from not approved relays without proxy
    NOT_APPROVED
@@ -120,6 +120,8 @@ module Simplex.Messaging.Agent
    debugAgentLocks,
    getAgentSubscriptions,
    logConnection,
+    -- for tests
+    withAgentEnv,
  )
 where

@@ -219,13 +221,14 @@ getSMPAgentClient_ clientId cfg initServers@InitialAgentServers {smp, xftp} stor
      | backgroundMode = run c "subscriber" $ subscriber c
      | otherwise = do
          restoreServersStats c
-          raceAny_
-            [ run c "subscriber" $ subscriber c,
-              run c "runNtfSupervisor" $ runNtfSupervisor c,
-              run c "cleanupManager" $ cleanupManager c,
-              run c "logServersStats" $ logServersStats c
-            ]
-            `E.finally` saveServersStats c
+          run c "subscriber" $ subscriber c
+          -- raceAny_
+          --   [ run c "subscriber" $ subscriber c
+          --     run c "runNtfSupervisor" $ runNtfSupervisor c,
+          --     run c "cleanupManager" $ cleanupManager c,
+          --     run c "logServersStats" $ logServersStats c
+          --   ]
+          --   `E.finally` saveServersStats c
    run AgentClient {subQ, acThread} name a =
      a `E.catchAny` \e -> whenM (isJust <$> readTVarIO acThread) $ do
        logError $ "Agent thread " <> name <> " crashed: " <> tshow e
@@ -685,9 +688,15 @@ allowConnectionAsync' c corrId connId confId ownConnInfo =
      enqueueCommand c corrId connId (Just server) $ AClientCommand $ LET confId ownConnInfo
    _ -> throwE $ CMD PROHIBITED "allowConnectionAsync"

+-- TODO
+-- Unlike `acceptContact` (synchronous version), `acceptContactAsync` uses `unacceptInvitation` in case of error,
+-- because we're not taking lock here. In practice it is less likely to fail because it doesn't involve network IO,
+-- and also it can't be triggered by user concurrently several times in a row. It could be improved similarly to
+-- `acceptContact` by creating a new map for invitation locks and taking lock here, and removing `unacceptInvitation`
+-- while marking invitation as accepted inside "lock level transaction" after successful `joinConnAsync`.
 acceptContactAsync' :: AgentClient -> ACorrId -> Bool -> InvitationId -> ConnInfo -> PQSupport -> SubscriptionMode -> AM ConnId
 acceptContactAsync' c corrId enableNtfs invId ownConnInfo pqSupport subMode = do
-  Invitation {contactConnId, connReq} <- withStore c (`getInvitation` invId)
+  Invitation {contactConnId, connReq} <- withStore c $ \db -> getInvitation db "acceptContactAsync'" invId
  withStore c (`getConn` contactConnId) >>= \case
    SomeConn _ (ContactConnection ConnData {userId} _) -> do
      withStore' c $ \db -> acceptInvitation db invId ownConnInfo
@@ -807,7 +816,7 @@ newConnToJoin c userId connId enableNtfs cReq pqSup = case cReq of

 newConnToAccept :: AgentClient -> ConnId -> Bool -> ConfirmationId -> PQSupport -> AM ConnId
 newConnToAccept c connId enableNtfs invId pqSup = do
-  Invitation {connReq, contactConnId} <- withStore c (`getInvitation` invId)
+  Invitation {connReq, contactConnId} <- withStore c $ \db -> getInvitation db "newConnToAccept" invId
  withStore c (`getConn` contactConnId) >>= \case
    SomeConn _ (ContactConnection ConnData {userId} _) ->
      newConnToJoin c userId connId enableNtfs connReq pqSup
@@ -815,11 +824,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 =
@@ -937,13 +948,12 @@ allowConnection' c connId confId ownConnInfo = withConnLock c connId "allowConne
 -- | Accept contact (ACPT command) in Reader monad
 acceptContact' :: AgentClient -> ConnId -> Bool -> InvitationId -> ConnInfo -> PQSupport -> SubscriptionMode -> AM SndQueueSecured
 acceptContact' c connId enableNtfs invId ownConnInfo pqSupport subMode = withConnLock c connId "acceptContact" $ do
-  Invitation {contactConnId, connReq} <- withStore c (`getInvitation` invId)
+  Invitation {contactConnId, connReq} <- withStore c $ \db -> getInvitation db "acceptContact'" invId
  withStore c (`getConn` contactConnId) >>= \case
    SomeConn _ (ContactConnection ConnData {userId} _) -> do
+      sqSecured <- joinConn c userId connId enableNtfs connReq ownConnInfo pqSupport subMode
      withStore' c $ \db -> acceptInvitation db invId ownConnInfo
-      joinConn c userId connId enableNtfs connReq ownConnInfo pqSupport subMode `catchAgentError` \err -> do
-        withStore' c (`unacceptInvitation` invId)
-        throwE err
+      pure sqSecured
    _ -> throwE $ CMD PROHIBITED "acceptContact"

 -- | Reject contact (RJCT command) in Reader monad
@@ -1194,14 +1204,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 +1658,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 +2167,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
@@ -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
@@ -2021,7 +2020,7 @@ storeError = \case
  SEConnDuplicate -> CONN DUPLICATE
  SEBadConnType CRcv -> CONN SIMPLEX
  SEBadConnType CSnd -> CONN SIMPLEX
-  SEInvitationNotFound -> CMD PROHIBITED "SEInvitationNotFound"
+  SEInvitationNotFound cxt invId -> CMD PROHIBITED $ "SEInvitationNotFound " <> cxt <> ", invitationId = " <> show invId
  -- this error is never reported as store error,
  -- it is used to wrap agent operations when "transaction-like" store access is needed
  -- NOTE: network IO should NOT be used inside AgentStoreMonad
@@ -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 ()
@@ -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)
@@ -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,
@@ -623,7 +623,7 @@ data StoreError
  | -- | Confirmation not found.
    SEConfirmationNotFound
  | -- | Invitation not found
-    SEInvitationNotFound
+    SEInvitationNotFound String InvitationId
  | -- | Message not found
    SEMsgNotFound
  | -- | Command not found
@@ -926,9 +926,9 @@ createInvitation db gVar NewInvitation {contactConnId, connReq, recipientConnInf
      |]
      (invitationId, contactConnId, connReq, recipientConnInfo)

-getInvitation :: DB.Connection -> InvitationId -> IO (Either StoreError Invitation)
-getInvitation db invitationId =
-  firstRow invitation SEInvitationNotFound $
+getInvitation :: DB.Connection -> String -> InvitationId -> IO (Either StoreError Invitation)
+getInvitation db cxt invitationId =
+  firstRow invitation (SEInvitationNotFound cxt invitationId) $
    DB.query
      db
      [sql|
@@ -145,7 +145,6 @@ import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Client (SocksAuth (..), SocksProxyWithAuth (..), TransportClientConfig (..), TransportHost (..), defaultSMPPort, defaultTcpConnectTimeout, runTransportClient)
 import Simplex.Messaging.Transport.KeepAlive
-import Simplex.Messaging.Transport.WebSockets (WS)
 import Simplex.Messaging.Util (bshow, diffToMicroseconds, ifM, liftEitherWith, raceAny_, threadDelay', tshow, whenM)
 import Simplex.Messaging.Version
 import System.Mem.Weak (Weak, deRefWeak)
@@ -544,7 +543,6 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
      "" -> case protocolTypeI @(ProtoType msg) of
        SPSMP | smpWebPort -> ("443", transport @TLS)
        _ -> defaultTransport cfg
-      "80" -> ("80", transport @WS)
      p -> (p, transport @TLS)

    client :: forall c. Transport c => TProxy c -> PClient v err msg -> TMVar (Either (ProtocolClientError err) (ProtocolClient v err msg)) -> c -> IO ()
@@ -71,6 +71,7 @@ import Data.Maybe (catMaybes, fromMaybe, isJust, isNothing)
 import Data.Semigroup (Sum (..))
 import qualified Data.Text as T
 import Data.Text.Encoding (decodeLatin1)
+import qualified Data.Text.IO as T
 import Data.Time.Clock (UTCTime (..), diffTimeToPicoseconds, getCurrentTime)
 import Data.Time.Clock.System (SystemTime (..), getSystemTime)
 import Data.Time.Format.ISO8601 (iso8601Show)
@@ -98,6 +99,7 @@ import Simplex.Messaging.Server.MsgStore.Journal (JournalQueue, closeMsgQueue)
 import Simplex.Messaging.Server.MsgStore.STM
 import Simplex.Messaging.Server.MsgStore.Types
 import Simplex.Messaging.Server.NtfStore
+import Simplex.Messaging.Server.Prometheus
 import Simplex.Messaging.Server.QueueStore
 import Simplex.Messaging.Server.QueueStore.QueueInfo
 import Simplex.Messaging.Server.QueueStore.STM
@@ -163,11 +165,11 @@ smpServer :: TMVar Bool -> ServerConfig -> Maybe AttachHTTP -> M ()
 smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHTTP_ = do
  s <- asks server
  pa <- asks proxyAgent
-  msgStats <- processServerMessages
+  msgStats_ <- processServerMessages
  ntfStats <- restoreServerNtfs
-  liftIO $ printMessageStats "messages" msgStats
+  liftIO $ mapM_ (printMessageStats "messages") msgStats_
  liftIO $ printMessageStats "notifications" ntfStats
-  restoreServerStats msgStats ntfStats
+  restoreServerStats msgStats_ ntfStats
  raceAny_
    ( serverThread s "server subscribedQ" subscribedQ subscribers subClients pendingSubEvents subscriptions cancelSub
        : serverThread s "server ntfSubscribedQ" ntfSubscribedQ Env.notifiers ntfSubClients pendingNtfSubEvents ntfSubscriptions (\_ -> pure ())
@@ -176,7 +178,11 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
        : receiveFromProxyAgent pa
        : expireNtfsThread cfg
        : sigIntHandlerThread
-        : map runServer transports <> expireMessagesThread_ cfg <> serverStatsThread_ cfg <> controlPortThread_ cfg
+        : map runServer transports
+            <> expireMessagesThread_ cfg
+            <> serverStatsThread_ cfg
+            <> prometheusMetricsThread_ cfg
+            <> controlPortThread_ cfg
    )
    `finally` stopServer s
  where
@@ -296,7 +302,6 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
        deliverNtfs ns stats (AClient _ Client {clientId, ntfSubscriptions, sndQ, connected}) =
          whenM (currentClient readTVarIO) $ do
            subs <- readTVarIO ntfSubscriptions
-            logDebug $ "NOTIFICATIONS: client #" <> tshow clientId <> " is current with " <> tshow (M.size subs) <> " subs"
            ntfQs <- M.assocs . M.filterWithKey (\nId _ -> M.member nId subs) <$> readTVarIO ns
            tryAny (atomically $ flushSubscribedNtfs ntfQs) >>= \case
              Right len -> updateNtfStats len
@@ -321,12 +326,11 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
                  writeTVar v []
                  pure $ foldl' (\acc' ntf -> nmsg nId ntf : acc') acc ntfs -- reverses, to order by time
            nmsg nId MsgNtf {ntfNonce, ntfEncMeta} = (CorrId "", nId, NMSG ntfNonce ntfEncMeta)
-            updateNtfStats 0 = logDebug $ "NOTIFICATIONS: no ntfs for client #" <> tshow clientId
+            updateNtfStats 0 = pure ()
            updateNtfStats len = liftIO $ do
              atomicModifyIORef'_ (ntfCount stats) (subtract len)
              atomicModifyIORef'_ (msgNtfs stats) (+ len)
              atomicModifyIORef'_ (msgNtfsB stats) (+ (len `div` 80 + 1)) -- up to 80 NMSG in the batch
-              logDebug $ "NOTIFICATIONS: delivered to client #" <> tshow clientId <> " " <> tshow len <> " ntfs"

    sendPendingEvtsThread :: Server -> M ()
    sendPendingEvtsThread s = do
@@ -386,13 +390,16 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
      liftIO $ forever $ do
        threadDelay' interval
        old <- expireBeforeEpoch expCfg
-        Sum deleted <- withActiveMsgQueues ms $ expireQueueMsgs stats old
-        logInfo $ "STORE: expireMessagesThread, expired " <> tshow deleted <> " messages"
+        now <- systemSeconds <$> getSystemTime
+        msgStats@MessageStats {storedMsgsCount = stored, expiredMsgsCount = expired} <-
+          withActiveMsgQueues ms $ expireQueueMsgs now ms old
+        atomicWriteIORef (msgCount stats) stored
+        atomicModifyIORef'_ (msgExpired stats) (+ expired)
+        printMessageStats "STORE: messages" msgStats
      where
-        expireQueueMsgs stats old rId q =
-          runExceptT (deleteExpiredMsgs rId q True old) >>= \case
-            Right deleted -> Sum deleted <$ atomicModifyIORef'_ (msgExpired stats) (+ deleted)
-            Left _ -> pure 0
+        expireQueueMsgs now ms old rId q = fmap (fromRight newMessageStats) . runExceptT $ do
+          (expired_, stored) <- idleDeleteExpiredMsgs now ms rId q old
+          pure MessageStats {storedMsgsCount = stored, expiredMsgsCount = fromMaybe 0 expired_, storedQueues = 1}

    expireNtfsThread :: ServerConfig -> M ()
    expireNtfsThread ServerConfig {notificationExpiration = expCfg} = do
@@ -554,6 +561,50 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
        showProxyStats ProxyStatsData {_pRequests, _pSuccesses, _pErrorsConnect, _pErrorsCompat, _pErrorsOther} =
          [show _pRequests, show _pSuccesses, show _pErrorsConnect, show _pErrorsCompat, show _pErrorsOther]

+    prometheusMetricsThread_ :: ServerConfig -> [M ()]
+    prometheusMetricsThread_ ServerConfig {prometheusInterval = Just interval, prometheusMetricsFile} =
+      [savePrometheusMetrics interval prometheusMetricsFile]
+    prometheusMetricsThread_ _ = []
+
+    savePrometheusMetrics :: Int -> FilePath -> M ()
+    savePrometheusMetrics saveInterval metricsFile = do
+      labelMyThread "savePrometheusMetrics"
+      liftIO $ putStrLn $ "Prometheus metrics saved every " <> show saveInterval <> " seconds to " <> metricsFile
+      AMS _ st <- asks msgStore
+      ss <- asks serverStats
+      env <- ask
+      let interval = 1000000 * saveInterval
+      liftIO $ forever $ do
+        threadDelay interval
+        ts <- getCurrentTime
+        sm <- getServerMetrics st ss
+        rtm <- getRealTimeMetrics env
+        T.writeFile metricsFile $ prometheusMetrics sm rtm ts
+
+    getServerMetrics :: STMQueueStore s => s -> ServerStats -> IO ServerMetrics
+    getServerMetrics st ss = do
+      d <- getServerStatsData ss
+      let ps = periodStatDataCounts $ _activeQueues d
+          psNtf = periodStatDataCounts $ _activeQueuesNtf d
+      queueCount <- M.size <$> readTVarIO (activeMsgQueues st)
+      notifierCount <- M.size <$> readTVarIO (notifiers' st)
+      pure ServerMetrics {statsData = d, activeQueueCounts = ps, activeNtfCounts = psNtf, queueCount, notifierCount}
+
+    getRealTimeMetrics :: Env -> IO RealTimeMetrics
+    getRealTimeMetrics Env {clients, sockets, server = Server {subscribers, notifiers, subClients, ntfSubClients}} = do
+      socketStats <- mapM (traverse getSocketStats) =<< readTVarIO sockets
+#if MIN_VERSION_base(4,18,0)
+      threadsCount <- length <$> listThreads
+#else
+      let threadsCount = 0
+#endif
+      clientsCount <- IM.size <$> readTVarIO clients
+      smpSubsCount <- M.size <$> readTVarIO subscribers
+      smpSubClientsCount <- IM.size <$> readTVarIO subClients
+      ntfSubsCount <- M.size <$> readTVarIO notifiers
+      ntfSubClientsCount <- IM.size <$> readTVarIO ntfSubClients
+      pure RealTimeMetrics {socketStats, threadsCount, clientsCount, smpSubsCount, smpSubClientsCount, ntfSubsCount, ntfSubClientsCount}
+
    runClient :: Transport c => C.APrivateSignKey -> TProxy c -> c -> M ()
    runClient signKey tp h = do
      kh <- asks serverIdentity
@@ -694,13 +745,13 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
 #endif
              CPSockets -> withUserRole $ unliftIO u (asks sockets) >>= readTVarIO >>= mapM_ putSockets
                where
-                  putSockets (tcpPort, (accepted', closed', active')) = do
-                    (accepted, closed, active) <- (,,) <$> readTVarIO accepted' <*> readTVarIO closed' <*> readTVarIO active'
+                  putSockets (tcpPort, socketsState) = do
+                    ss <- getSocketStats socketsState
                    hPutStrLn h $ "Sockets for port " <> tcpPort <> ":"
-                    hPutStrLn h $ "accepted: " <> show accepted
-                    hPutStrLn h $ "closed: " <> show closed
-                    hPutStrLn h $ "active: " <> show (IM.size active)
-                    hPutStrLn h $ "leaked: " <> show (accepted - closed - IM.size active)
+                    hPutStrLn h $ "accepted: " <> show (socketsAccepted ss)
+                    hPutStrLn h $ "closed: " <> show (socketsClosed ss)
+                    hPutStrLn h $ "active: " <> show (socketsActive ss)
+                    hPutStrLn h $ "leaked: " <> show (socketsLeaked ss)
              CPSocketThreads -> withAdminRole $ do
 #if MIN_VERSION_base(4,18,0)
                unliftIO u (asks sockets) >>= readTVarIO >>= mapM_ putSocketThreads
@@ -1469,7 +1520,7 @@ client

            expireMessages :: Maybe ExpirationConfig -> ServerStats -> ExceptT ErrorType IO ()
            expireMessages msgExp stats = do
-              deleted <- maybe (pure 0) (deleteExpiredMsgs (recipientId qr) q True <=< liftIO . expireBeforeEpoch) msgExp
+              deleted <- maybe (pure 0) (deleteExpiredMsgs ms (recipientId qr) q <=< liftIO . expireBeforeEpoch) msgExp
              liftIO $ when (deleted > 0) $ atomicModifyIORef'_ (msgExpired stats) (+ deleted)

            -- The condition for delivery of the message is:
@@ -1733,26 +1784,26 @@ exportMessages tty ms f drainMsgs = do
          exitFailure
    encodeMessages rId = mconcat . map (\msg -> BLD.byteString (strEncode $ MLRv3 rId msg) <> BLD.char8 '\n')

-processServerMessages :: M MessageStats
+processServerMessages :: M (Maybe MessageStats)
 processServerMessages = do
  old_ <- asks (messageExpiration . config) $>>= (liftIO . fmap Just . expireBeforeEpoch)
  expire <- asks $ expireMessagesOnStart . config
  asks msgStore >>= liftIO . processMessages old_ expire
    where
-      processMessages :: Maybe Int64 -> Bool -> AMsgStore -> IO MessageStats
+      processMessages :: Maybe Int64 -> Bool -> AMsgStore -> IO (Maybe MessageStats)
      processMessages old_ expire = \case
        AMS SMSMemory ms@STMMsgStore {storeConfig = STMStoreConfig {storePath}} -> case storePath of
-          Just f -> ifM (doesFileExist f) (importMessages False ms f old_) (pure newMessageStats)
-          Nothing -> pure newMessageStats
+          Just f -> ifM (doesFileExist f) (Just <$> importMessages False ms f old_) (pure Nothing)
+          Nothing -> pure Nothing
        AMS SMSJournal ms
-          | expire -> case old_ of
+          | expire -> Just <$> case old_ of
              Just old -> do
                logInfo "expiring journal store messages..."
                withAllMsgQueues False ms $ processExpireQueue old
              Nothing -> do
                logInfo "validating journal store messages..."
                withAllMsgQueues False ms $ processValidateQueue
-          | otherwise -> logWarn "skipping message expiration" $> newMessageStats
+          | otherwise -> logWarn "skipping message expiration" $> Nothing
          where
            processExpireQueue old rId q =
              runExceptT expireQueue >>= \case
@@ -1763,9 +1814,8 @@ processServerMessages = do
                  exitFailure
              where
                expireQueue = do
-                  expired'' <- deleteExpiredMsgs rId q False old
+                  expired'' <- deleteExpiredMsgs ms rId q old
                  stored'' <- getQueueSize ms rId q
-                  liftIO $ logQueueState q
                  liftIO $ closeMsgQueue q
                  pure (stored'', expired'')
            processValidateQueue :: RecipientId -> JournalQueue -> IO MessageStats
@@ -1821,9 +1871,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_ 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)
@@ -1890,8 +1940,8 @@ saveServerStats =
      B.writeFile f $ strEncode stats
      logInfo "server stats saved"

-restoreServerStats :: MessageStats -> MessageStats -> M ()
-restoreServerStats msgStats ntfStats = asks (serverStatsBackupFile . config) >>= mapM_ restoreStats
+restoreServerStats :: Maybe MessageStats -> MessageStats -> M ()
+restoreServerStats msgStats_ ntfStats = asks (serverStatsBackupFile . config) >>= mapM_ restoreStats
  where
    restoreStats f = whenM (doesFileExist f) $ do
      logInfo $ "restoring server stats from file " <> T.pack f
@@ -1900,9 +1950,11 @@ restoreServerStats msgStats ntfStats = asks (serverStatsBackupFile . config) >>=
          s <- asks serverStats
          AMS _ st <- asks msgStore
          _qCount <- M.size <$> readTVarIO (activeMsgQueues st)
-          let _msgCount = storedMsgsCount msgStats
+          let _msgCount = maybe statsMsgCount storedMsgsCount msgStats_
              _ntfCount = storedMsgsCount ntfStats
-          liftIO $ setServerStats s d {_qCount, _msgCount, _ntfCount, _msgExpired = _msgExpired d + expiredMsgsCount msgStats, _msgNtfExpired = _msgNtfExpired d + expiredMsgsCount ntfStats}
+              _msgExpired' = _msgExpired d + maybe 0 expiredMsgsCount msgStats_
+              _msgNtfExpired' = _msgNtfExpired d + expiredMsgsCount ntfStats
+          liftIO $ setServerStats s d {_qCount, _msgCount, _ntfCount, _msgExpired = _msgExpired', _msgNtfExpired = _msgNtfExpired'}
          renameFile f $ f <> ".bak"
          logInfo "server stats restored"
          compareCounts "Queue" statsQCount _qCount
@@ -80,6 +80,8 @@ data ServerConfig = ServerConfig
    -- | time after which the messages can be removed from the queues and check interval, seconds
    messageExpiration :: Maybe ExpirationConfig,
    expireMessagesOnStart :: Bool,
+    -- | interval of inactivity after which journal queue is closed
+    idleQueueInterval :: Int64,
    -- | notification expiration interval (seconds)
    notificationExpiration :: ExpirationConfig,
    -- | time after which the socket with inactive client can be disconnected (without any messages or commands, incl. PING),
@@ -94,6 +96,9 @@ data ServerConfig = ServerConfig
    serverStatsLogFile :: FilePath,
    -- | file to save and restore stats
    serverStatsBackupFile :: Maybe FilePath,
+    -- | interval and file to save prometheus metrics
+    prometheusInterval :: Maybe Int,
+    prometheusMetricsFile :: FilePath,
    -- | notification delivery interval
    ntfDeliveryInterval :: Int,
    -- | interval between sending pending END events to unsubscribed clients, seconds
@@ -121,9 +126,12 @@ defaultMessageExpiration :: ExpirationConfig
 defaultMessageExpiration =
  ExpirationConfig
    { ttl = defMsgExpirationDays * 86400, -- seconds
-      checkInterval = 43200 -- seconds, 12 hours
+      checkInterval = 14400 -- seconds, 4 hours
    }

+defaultIdleQueueInterval :: Int64
+defaultIdleQueueInterval = 28800 -- seconds, 8 hours
+
 defNtfExpirationHours :: Int64
 defNtfExpirationHours = 24

@@ -283,14 +291,14 @@ newProhibitedSub = do
  return Sub {subThread = ProhibitSub, delivered}

 newEnv :: ServerConfig -> IO Env
-newEnv config@ServerConfig {smpCredentials, httpCredentials, storeLogFile, msgStoreType, storeMsgsFile, smpAgentCfg, information, messageExpiration, msgQueueQuota, maxJournalMsgCount, maxJournalStateLines} = do
+newEnv config@ServerConfig {smpCredentials, httpCredentials, storeLogFile, msgStoreType, storeMsgsFile, smpAgentCfg, information, messageExpiration, idleQueueInterval, msgQueueQuota, maxJournalMsgCount, maxJournalStateLines} = do
  serverActive <- newTVarIO True
  server <- newServer
  msgStore@(AMS _ store) <- case msgStoreType of
    AMSType SMSMemory -> AMS SMSMemory <$> newMsgStore STMStoreConfig {storePath = storeMsgsFile, quota = msgQueueQuota}
    AMSType SMSJournal -> case storeMsgsFile of
      Just storePath -> 
-        let cfg = JournalStoreConfig {storePath, quota = msgQueueQuota, pathParts = journalMsgStoreDepth, maxMsgCount = maxJournalMsgCount, maxStateLines = maxJournalStateLines, stateTailSize = defaultStateTailSize}
+        let cfg = JournalStoreConfig {storePath, quota = msgQueueQuota, pathParts = journalMsgStoreDepth, maxMsgCount = maxJournalMsgCount, maxStateLines = maxJournalStateLines, stateTailSize = defaultStateTailSize, idleInterval = idleQueueInterval}
         in AMS SMSJournal <$> newMsgStore cfg
      Nothing -> putStrLn "Error: journal msg store require path in [STORE_LOG], restore_messages" >> exitFailure
  ntfStore <- NtfStore <$> TM.emptyIO
@@ -148,7 +148,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
      doesFileExist iniFile >>= \case
        True -> readIniFile iniFile >>= either exitError a
        _ -> exitError $ "Error: server is not initialized (" <> iniFile <> " does not exist).\nRun `" <> executableName <> " init`."
-    newJournalMsgStore = newMsgStore JournalStoreConfig {storePath = storeMsgsJournalDir, pathParts = journalMsgStoreDepth, quota = defaultMsgQueueQuota, maxMsgCount = defaultMaxJournalMsgCount, maxStateLines = defaultMaxJournalStateLines, stateTailSize = defaultStateTailSize}
+    newJournalMsgStore = newMsgStore JournalStoreConfig {storePath = storeMsgsJournalDir, pathParts = journalMsgStoreDepth, quota = defaultMsgQueueQuota, maxMsgCount = defaultMaxJournalMsgCount, maxStateLines = defaultMaxJournalStateLines, stateTailSize = defaultStateTailSize, idleInterval = checkInterval defaultMessageExpiration}
    iniFile = combine cfgPath "smp-server.ini"
    serverVersion = "SMP server v" <> simplexMQVersion
    defaultServerPorts = "5223,443"
@@ -253,7 +253,9 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
                <> ("expire_ntfs_hours: " <> tshow defNtfExpirationHours <> "\n\n")
                <> "# Log daily server statistics to CSV file\n"
                <> ("log_stats: " <> onOff logStats <> "\n\n")
-                <> "[AUTH]\n\
+                <> "# Log interval for real-time Prometheus metrics\n\
+                   \# prometheus_interval: 300\n\n\
+                   \[AUTH]\n\
                   \# Set new_queues option to off to completely prohibit creating new messaging queues.\n\
                   \# This can be useful when you want to decommission the server, but not all connections are switched yet.\n\
                   \new_queues: on\n\n\
@@ -416,6 +418,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
                    { ttl = 86400 * readIniDefault defMsgExpirationDays "STORE_LOG" "expire_messages_days" ini
                    },
              expireMessagesOnStart = fromMaybe True $ iniOnOff "STORE_LOG" "expire_messages_on_start" ini,
+              idleQueueInterval = defaultIdleQueueInterval,
              notificationExpiration =
                defaultNtfExpiration
                  { ttl = 3600 * readIniDefault defNtfExpirationHours "STORE_LOG" "expire_ntfs_hours" ini
@@ -430,6 +433,8 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
              logStatsStartTime = 0, -- seconds from 00:00 UTC
              serverStatsLogFile = combine logPath "smp-server-stats.daily.log",
              serverStatsBackupFile = logStats $> combine logPath "smp-server-stats.log",
+              prometheusInterval = eitherToMaybe $ read . T.unpack <$> lookupValue "STORE_LOG" "prometheus_interval" ini,
+              prometheusMetricsFile = combine logPath "smp-server-metrics.txt",
              pendingENDInterval = 15000000, -- 15 seconds
              ntfDeliveryInterval = 3000000, -- 3 seconds
              smpServerVRange = supportedServerSMPRelayVRange,
@@ -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,9 +49,10 @@ 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)
 import Data.Time.Format.ISO8601 (iso8601Show)
 import GHC.IO (catchAny)
 import Simplex.Messaging.Agent.Client (getMapLock, withLockMap)
@@ -92,7 +92,9 @@ data JournalStoreConfig = JournalStoreConfig
    -- This number should be set bigger than queue quota.
    maxMsgCount :: Int,
    maxStateLines :: Int,
-    stateTailSize :: Int
+    stateTailSize :: Int,
+    -- time in seconds after which the queue will be closed after message expiration
+    idleInterval :: Int64
  }

 data JournalQueue = JournalQueue
@@ -100,7 +102,11 @@ data JournalQueue = JournalQueue
    -- To avoid race conditions and errors when restoring queues,
    -- Nothing is written to TVar when queue is deleted.
    queueRec :: TVar (Maybe QueueRec),
-    msgQueue_ :: TVar (Maybe JournalMsgQueue)
+    msgQueue_ :: TVar (Maybe JournalMsgQueue),
+    -- system time in seconds since epoch
+    activeAt :: TVar Int64,
+    -- Just True - empty, Just False - non-empty, Nothing - unknown
+    isEmpty :: TVar (Maybe Bool)
  }

 data JMQueue = JMQueue
@@ -219,9 +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
-    pure $ JournalQueue lock q mq
+    activeAt <- newTVar 0
+    isEmpty <- newTVar Nothing
+    pure $ JournalQueue lock q mq activeAt isEmpty
  msgQueue_' = msgQueue_

 instance MsgStoreClass JournalMsgStore where
@@ -295,11 +303,11 @@ instance MsgStoreClass JournalMsgStore where
              (Nothing <$ putStrLn ("Error: path " <> path' <> " is not a directory, skipping"))

  logQueueStates :: JournalMsgStore -> IO ()
-  logQueueStates ms = withActiveMsgQueues ms $ \_ -> logQueueState
+  logQueueStates ms = withActiveMsgQueues ms $ \_ -> unStoreIO . logQueueState

-  logQueueState :: JournalQueue -> IO ()
+  logQueueState :: JournalQueue -> StoreIO ()
  logQueueState q =
-    void $
+    StoreIO . void $
      readTVarIO (msgQueue_ q)
        $>>= \mq -> readTVarIO (handles mq)
        $>>= (\hs -> (readTVarIO (state mq) >>= appendState (stateHandle hs)) $> Just ())
@@ -316,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
@@ -326,9 +334,46 @@ instance MsgStoreClass JournalMsgStore where
            journalId <- newJournalId random
            mkJournalQueue queue (newMsgQueueState journalId) Nothing

-  openedMsgQueue :: JournalQueue -> StoreIO (Maybe JournalMsgQueue)
-  openedMsgQueue = StoreIO . readTVarIO . msgQueue_
-  {-# INLINE openedMsgQueue #-}
+  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 $ 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
+          then Just <$> unStoreIO (action mq) `E.finally` closeMsgQueue q
+          else pure Nothing
+        sz <- unStoreIO $ getQueueSize_ mq
+        pure (r, sz)

  deleteQueue :: JournalMsgStore -> RecipientId -> JournalQueue -> IO (Either ErrorType QueueRec)
  deleteQueue ms rId q =
@@ -355,11 +400,12 @@ instance MsgStoreClass JournalMsgStore where
  writeMsg :: JournalMsgStore -> RecipientId -> JournalQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
  writeMsg ms rId q' logState msg = isolateQueue rId q' "writeMsg" $ do
    q <- getMsgQueue ms rId q'
-    StoreIO $ do
+    StoreIO $ (`E.finally` updateActiveAt q') $ do
      st@MsgQueueState {canWrite, size} <- readTVarIO (state q)
      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)
@@ -408,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
@@ -418,19 +464,25 @@ 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_ :: JournalMsgQueue -> Bool -> StoreIO ()
-  tryDeleteMsg_ q@JournalMsgQueue {tipMsg, handles} logState = StoreIO $
+  tryDeleteMsg_ :: JournalQueue -> JournalMsgQueue -> Bool -> StoreIO ()
+  tryDeleteMsg_ q mq@JournalMsgQueue {tipMsg, handles} logState = StoreIO $ (`E.finally` when logState (updateActiveAt q)) $
    void $
      readTVarIO tipMsg -- if there is no cached tipMsg, do nothing
        $>>= (pure . fmap snd)
        $>>= \len -> readTVarIO handles
-        $>>= \hs -> updateReadPos q logState len hs $> Just ()
+        $>>= \hs -> updateReadPos mq logState len hs $> Just ()

  isolateQueue :: RecipientId -> JournalQueue -> String -> StoreIO a -> ExceptT ErrorType IO a
  isolateQueue rId JournalQueue {queueLock} op =
    tryStore' op rId . withLock' queueLock op . unStoreIO

+updateActiveAt :: JournalQueue -> IO ()
+updateActiveAt q = atomically . writeTVar (activeAt q) . systemSeconds =<< getSystemTime
+
 tryStore' :: String -> RecipientId -> IO a -> ExceptT ErrorType IO a
 tryStore' op rId = tryStore op rId . fmap Right

@@ -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 (..),
@@ -21,6 +20,7 @@ import Control.Concurrent.STM
 import Control.Monad.IO.Class
 import Control.Monad.Trans.Except
 import Data.Functor (($>))
+import Data.Int (Int64)
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.MsgStore.Types
 import Simplex.Messaging.Server.QueueStore
@@ -28,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
@@ -62,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
@@ -105,12 +105,20 @@ 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

-  openedMsgQueue :: STMQueue -> STM (Maybe STMMsgQueue)
-  openedMsgQueue = readTVar . msgQueue_
-  {-# INLINE openedMsgQueue #-}
+  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
+    Just q -> do
+      r <- action q
+      sz <- getQueueSize_ q
+      pure (Just r, sz)
+    Nothing -> pure (Nothing, 0)

  deleteQueue :: STMMsgStore -> RecipientId -> STMQueue -> IO (Either ErrorType QueueRec)
  deleteQueue ms rId q = fst <$$> deleteQueue' ms rId q
@@ -153,12 +161,12 @@ 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_ :: STMMsgQueue -> Bool -> STM ()
-  tryDeleteMsg_ STMMsgQueue {msgQueue = q, size} _logState =
+  tryDeleteMsg_ :: STMQueue -> STMMsgQueue -> Bool -> STM ()
+  tryDeleteMsg_ _ STMMsgQueue {msgQueue = q, size} _logState =
    tryReadTQueue q >>= \case
      Just _ -> modifyTVar' size (subtract 1)
      _ -> pure ()
@@ -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
@@ -42,18 +48,21 @@ class Monad (StoreMonad s) => MsgStoreClass s where
  activeMsgQueues :: s -> TMap RecipientId (StoreQueue s)
  withAllMsgQueues :: Monoid a => Bool -> s -> (RecipientId -> StoreQueue s -> IO a) -> IO a
  logQueueStates :: s -> IO ()
-  logQueueState :: StoreQueue 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)
-  openedMsgQueue :: StoreQueue s -> StoreMonad s (Maybe (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)
  deleteQueueSize :: s -> RecipientId -> StoreQueue s -> IO (Either ErrorType (QueueRec, Int))
  getQueueMessages_ :: Bool -> MsgQueue s -> StoreMonad s [Message]
  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)
-  tryDeleteMsg_ :: MsgQueue s -> Bool -> StoreMonad s ()
+  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

 data MSType = MSMemory | MSJournal
@@ -72,47 +81,61 @@ 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_ 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_ 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 => RecipientId -> StoreQueue s -> Bool -> Int64 -> ExceptT ErrorType IO Int
-deleteExpiredMsgs rId q logState old =
-  isolateQueue rId q "deleteExpiredMsgs" $ openedMsgQueue q >>= maybe (pure 0) (loop 0)
+deleteExpiredMsgs :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> Int64 -> ExceptT ErrorType IO Int
+deleteExpiredMsgs st rId q old =
+  isolateQueue rId q "deleteExpiredMsgs" $
+    getMsgQueue st rId q >>= deleteExpireMsgs_ old q
+
+-- closed and idle queues will be closed after expiration
+-- returns (expired count, queue size after expiration)
+idleDeleteExpiredMsgs :: MsgStoreClass s => Int64 -> s -> RecipientId -> StoreQueue s -> Int64 -> ExceptT ErrorType IO (Maybe Int, Int)
+idleDeleteExpiredMsgs now st rId q old =
+  isolateQueue rId q "idleDeleteExpiredMsgs" $
+    withIdleMsgQueue now st rId q (deleteExpireMsgs_ old q)
+
+deleteExpireMsgs_ :: MsgStoreClass s => Int64 -> StoreQueue s -> MsgQueue s -> StoreMonad s Int
+deleteExpireMsgs_ old q mq = do
+  n <- loop 0
+  logQueueState q
+  pure n
  where
-    loop dc mq =
-      tryPeekMsg_ mq >>= \case
+    loop dc =
+      tryPeekMsg_ q mq >>= \case
        Just Message {msgTs}
          | systemSeconds msgTs < old ->
-              tryDeleteMsg_ mq logState >> loop (dc + 1) mq
+              tryDeleteMsg_ q mq False >> loop (dc + 1)
        _ -> pure dc
@@ -0,0 +1,379 @@
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE TypeApplications #-}
+{-# OPTIONS_GHC -fno-warn-unrecognised-pragmas #-}
+
+module Simplex.Messaging.Server.Prometheus where
+
+import Data.Int (Int64)
+import Data.Text (Text)
+import qualified Data.Text as T
+import Data.Time.Clock (UTCTime (..), diffUTCTime)
+import Data.Time.Clock.System (systemEpochDay)
+import Data.Time.Format.ISO8601 (iso8601Show)
+import Network.Socket (ServiceName)
+import Simplex.Messaging.Server.Stats
+import Simplex.Messaging.Transport.Server (SocketStats (..))
+
+data ServerMetrics = ServerMetrics
+  { statsData :: ServerStatsData,
+    activeQueueCounts :: PeriodStatCounts,
+    activeNtfCounts :: PeriodStatCounts,
+    queueCount :: Int,
+    notifierCount :: Int
+  }
+
+data RealTimeMetrics = RealTimeMetrics
+  { socketStats :: [(ServiceName, SocketStats)],
+    threadsCount :: Int,
+    clientsCount :: Int,
+    smpSubsCount :: Int,
+    smpSubClientsCount :: Int,
+    ntfSubsCount :: Int,
+    ntfSubClientsCount :: Int
+  }
+
+{-# FOURMOLU_DISABLE\n#-}
+prometheusMetrics :: ServerMetrics -> RealTimeMetrics -> UTCTime -> Text
+prometheusMetrics sm rtm ts =
+  time <> queues <> subscriptions <> messages <> ntfMessages <> ntfs <> relays <> info
+  where
+    ServerMetrics {statsData, activeQueueCounts = ps, activeNtfCounts = psNtf, queueCount, notifierCount} = sm
+    RealTimeMetrics
+      { socketStats,
+        threadsCount,
+        clientsCount,
+        smpSubsCount,
+        smpSubClientsCount,
+        ntfSubsCount,
+        ntfSubClientsCount
+      } = rtm
+    ServerStatsData
+      { _fromTime,
+        _qCreated,
+        _qSecured,
+        _qDeletedAll,
+        _qDeletedAllB,
+        _qDeletedNew,
+        _qDeletedSecured,
+        _qSub,
+        _qSubAllB,
+        _qSubAuth,
+        _qSubDuplicate,
+        _qSubProhibited,
+        _qSubEnd,
+        _qSubEndB,
+        _ntfCreated,
+        _ntfDeleted,
+        _ntfDeletedB,
+        _ntfSub,
+        _ntfSubB,
+        _ntfSubAuth,
+        _ntfSubDuplicate,
+        _msgSent,
+        _msgSentAuth,
+        _msgSentQuota,
+        _msgSentLarge,
+        _msgRecv,
+        _msgRecvGet,
+        _msgGet,
+        _msgGetNoMsg,
+        _msgGetAuth,
+        _msgGetDuplicate,
+        _msgGetProhibited,
+        _msgExpired,
+        _activeQueues,
+        _msgSentNtf,
+        _msgRecvNtf,
+        _activeQueuesNtf,
+        _msgNtfs,
+        _msgNtfsB,
+        _msgNtfNoSub,
+        _msgNtfLost,
+        _msgNtfExpired,
+        _pRelays,
+        _pRelaysOwn,
+        _pMsgFwds,
+        _pMsgFwdsOwn,
+        _pMsgFwdsRecv,
+        _qCount,
+        _msgCount,
+        _ntfCount
+      } = statsData
+    time =
+      "# Recorded at: " <> T.pack (iso8601Show ts) <> "\n\
+      \# Stats from: " <> T.pack (iso8601Show _fromTime) <> "\n\
+      \\n"
+    queues =
+      "# Queues\n\
+      \# ------\n\
+      \\n\
+      \# HELP simplex_smp_queues_created Created queues\n\
+      \# TYPE simplex_smp_queues_created counter\n\
+      \simplex_smp_queues_created " <> mshow _qCreated <> "\n# qCreated\n\
+      \\n\
+      \# HELP simplex_smp_queues_secured Secured queues\n\
+      \# TYPE simplex_smp_queues_secured counter\n\
+      \simplex_smp_queues_secured " <> mshow _qSecured <> "\n# qSecured\n\
+      \\n\
+      \# HELP simplex_smp_queues_deleted Deleted queues\n\
+      \# TYPE simplex_smp_queues_deleted counter\n\
+      \simplex_smp_queues_deleted{type=\"all\"} " <> mshow _qDeletedAll <> "\n# qDeleted\n\
+      \simplex_smp_queues_deleted{type=\"new\"} " <> mshow _qDeletedNew <> "\n# qDeletedNew\n\
+      \simplex_smp_queues_deleted{type=\"secured\"} " <> mshow _qDeletedSecured <> "\n# qDeletedSecured\n\
+      \\n\
+      \# HELP simplex_smp_queues_deleted_batch Batched requests to delete queues\n\
+      \# TYPE simplex_smp_queues_deleted_batch counter\n\
+      \simplex_smp_queues_deleted_batch " <> mshow _qDeletedAllB <> "\n# qDeletedAllB\n\
+      \\n\
+      \# HELP simplex_smp_queues_total1 Total number of stored queues (first type of count).\n\
+      \# TYPE simplex_smp_queues_total1 gauge\n\
+      \simplex_smp_queues_total1 " <> mshow _qCount <> "\n# qCount\n\
+      \\n\
+      \# HELP simplex_smp_queues_total2 Total number of stored queues (second type of count).\n\
+      \# TYPE simplex_smp_queues_total2 gauge\n\
+      \simplex_smp_queues_total2 " <> mshow queueCount <> "\n# qCount2\n\
+      \\n\
+      \# HELP simplex_smp_queues_daily Daily active queues.\n\
+      \# TYPE simplex_smp_queues_daily gauge\n\
+      \simplex_smp_queues_daily " <> mstr (dayCount ps) <> "\n# dayMsgQueues\n\
+      \\n\
+      \# HELP simplex_smp_queues_weekly Weekly active queues.\n\
+      \# TYPE simplex_smp_queues_weekly gauge\n\
+      \simplex_smp_queues_weekly " <> mstr (weekCount ps) <> "\n# weekMsgQueues\n\
+      \\n\
+      \# HELP simplex_smp_queues_monthly Monthly active queues.\n\
+      \# TYPE simplex_smp_queues_monthly gauge\n\
+      \simplex_smp_queues_monthly " <> mstr (monthCount ps) <> "\n# monthMsgQueues\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_daily Daily active queues with notifications.\n\
+      \# TYPE simplex_smp_queues_notify_daily gauge\n\
+      \simplex_smp_queues_notify_daily " <> mstr (dayCount psNtf) <> "\n# dayCountNtf\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_weekly Weekly active queues with notifications.\n\
+      \# TYPE simplex_smp_queues_notify_weekly gauge\n\
+      \simplex_smp_queues_notify_weekly " <> mstr (weekCount psNtf) <> "\n# weekCountNtf\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_monthly Monthly active queues with notifications.\n\
+      \# TYPE simplex_smp_queues_notify_monthly gauge\n\
+      \simplex_smp_queues_notify_monthly " <> mstr (monthCount psNtf) <> "\n# monthCountNtf\n\
+      \\n"
+    subscriptions =
+      "# Subscriptions\n\
+      \# -------------\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_successes Successful subscriptions.\n\
+      \# TYPE simplex_smp_subscribtion_successes counter\n\
+      \simplex_smp_subscribtion_successes " <> mshow _qSub <> "\n# qSub\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_successes_batch Batched successful subscriptions.\n\
+      \# TYPE simplex_smp_subscribtion_successes_batch counter\n\
+      \simplex_smp_subscribtion_successes_batch " <> mshow _qSubAllB <> "\n# qSubAllB\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_end Ended subscriptions.\n\
+      \# TYPE simplex_smp_subscribtion_end counter\n\
+      \simplex_smp_subscribtion_end " <> mshow _qSubEnd <> "\n# qSubEnd\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_end_batch Batched ended subscriptions.\n\
+      \# TYPE simplex_smp_subscribtion_end_batch counter\n\
+      \simplex_smp_subscribtion_end_batch " <> mshow _qSubEndB <> "\n# qSubEndB\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_errors Subscription errors.\n\
+      \# TYPE simplex_smp_subscribtion_errors counter\n\
+      \simplex_smp_subscribtion_errors{type=\"auth\"} " <> mshow _qSubAuth <> "\n# qSubAuth\n\
+      \simplex_smp_subscribtion_errors{type=\"duplicate\"} " <> mshow _qSubDuplicate <> "\n# qSubDuplicate\n\
+      \simplex_smp_subscribtion_errors{type=\"prohibited\"} " <> mshow _qSubProhibited <> "\n# qSubProhibited\n\
+      \\n"
+    messages =
+      "# Messages\n\
+      \# --------\n\
+      \\n\
+      \# HELP simplex_smp_messages_sent Sent messages.\n\
+      \# TYPE simplex_smp_messages_sent counter\n\
+      \simplex_smp_messages_sent " <> mshow _msgSent <> "\n# msgSent\n\
+      \\n\
+      \# HELP simplex_smp_messages_sent_errors Total number of messages errors by type.\n\
+      \# TYPE simplex_smp_messages_sent_errors counter\n\
+      \simplex_smp_messages_sent_errors{type=\"auth\"} " <> mshow _msgSentAuth <> "\n# msgSentAuth\n\
+      \simplex_smp_messages_sent_errors{type=\"quota\"} " <> mshow _msgSentQuota <> "\n# msgSentQuota\n\
+      \simplex_smp_messages_sent_errors{type=\"large\"} " <> mshow _msgSentLarge <> "\n# msgSentLarge\n\
+      \\n\
+      \# HELP simplex_smp_messages_received Received messages.\n\
+      \# TYPE simplex_smp_messages_received counter\n\
+      \simplex_smp_messages_received " <> mshow _msgRecv <> "\n# msgRecv\n\
+      \\n\
+      \# HELP simplex_smp_messages_expired Expired messages.\n\
+      \# TYPE simplex_smp_messages_expired counter\n\
+      \simplex_smp_messages_expired " <> mshow _msgExpired <> "\n# msgExpired\n\
+      \\n\
+      \# HELP simplex_smp_messages_total Total number of messages stored.\n\
+      \# TYPE simplex_smp_messages_total gauge\n\
+      \simplex_smp_messages_total " <> mshow _msgCount <> "\n# msgCount\n\
+      \\n"
+    ntfMessages =
+      "# Notification messages (client)\n\
+      \# ------------------------------\n\
+      \\n\
+      \# HELP simplex_smp_messages_notify_sent Sent messages with notification flag (cleint).\n\
+      \# TYPE simplex_smp_messages_notify_sent counter\n\
+      \simplex_smp_messages_notify_sent " <> mshow _msgSentNtf <> "\n# msgSentNtf\n\
+      \\n\
+      \# HELP simplex_smp_messages_notify_received Received messages with notification flag (client).\n\
+      \# TYPE simplex_smp_messages_notify_received counter\n\
+      \simplex_smp_messages_notify_received " <> mshow _msgRecvNtf <> "\n# msgRecvNtf\n\
+      \\n\
+      \# HELP simplex_smp_messages_notify_get_sent Requests to get messages with notification flag (client).\n\
+      \# TYPE simplex_smp_messages_notify_get_sent counter\n\
+      \simplex_smp_messages_notify_get_sent " <> mshow _msgGet <> "\n# msgGet\n\
+      \\n\
+      \# HELP simplex_smp_messages_notify_get_received Succesfully received get requests messages with notification flag (client).\n\
+      \# TYPE simplex_smp_messages_notify_get_received counter\n\
+      \simplex_smp_messages_notify_get_received " <> mshow _msgRecvGet <> "\n# msgRecvGet\n\
+      \\n\
+      \# HELP simplex_smp_messages_notify_get_errors Error events with messages with notification flag (client). \n\
+      \# TYPE simplex_smp_messages_notify_get_errors counter\n\
+      \simplex_smp_messages_notify_get_errors{type=\"nomsg\"} " <> mshow _msgGetNoMsg <> "\n# msgGetNoMsg\n\
+      \simplex_smp_messages_notify_get_errors{type=\"auth\"} " <> mshow _msgGetAuth <> "\n# msgGetAuth\n\
+      \simplex_smp_messages_notify_get_errors{type=\"duplicate\"} " <> mshow _msgGetDuplicate <> "\n# msgGetDuplicate\n\
+      \simplex_smp_messages_notify_get_errors{type=\"prohibited\"} " <> mshow _msgGetProhibited <> "\n# msgGetProhibited\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_created Created queues with notification flag (client).\n\
+      \# TYPE simplex_smp_queues_notify_created counter\n\
+      \simplex_smp_queues_notify_created " <> mshow _ntfCreated <> "\n# ntfCreated\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_deleted Deleted queues with notification flag (client).\n\
+      \# TYPE simplex_smp_queues_notify_deleted counter\n\
+      \simplex_smp_queues_notify_deleted " <> mshow _ntfDeleted <> "\n# ntfDeleted\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_deleted_batch Deleted batched queues with notification flag (client).\n\
+      \# TYPE simplex_smp_queues_notify_deleted_batch counter\n\
+      \simplex_smp_queues_notify_deleted_batch " <> mshow _ntfDeletedB <> "\n# ntfDeletedB\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_total1 Total number of stored queues with notification flag (first type of count).\n\
+      \# TYPE simplex_smp_queues_notify_total1 gauge\n\
+      \simplex_smp_queues_notify_total1 " <> mshow _ntfCount <> "\n# ntfCount1\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_total2 Total number of stored queues with notification flag (second type of count).\n\
+      \# TYPE simplex_smp_queues_notify_total2 gauge\n\
+      \simplex_smp_queues_notify_total2 " <> mshow notifierCount <> "\n# ntfCount2\n\
+      \\n"
+    ntfs =
+      "# Notifications (server)\n\
+      \# ----------------------\n\
+      \\n\
+      \# HELP simplex_smp_messages_ntf_successes Successful events with notification messages (to ntf server). \n\
+      \# TYPE simplex_smp_messages_ntf_successes counter\n\
+      \simplex_smp_messages_ntf_successes " <> mshow _msgNtfs <> "\n# msgNtfs\n\
+      \\n\
+      \# HELP simplex_smp_messages_ntf_successes_batch Successful batched events with notification messages (to ntf server). \n\
+      \# TYPE simplex_smp_messages_ntf_successes_batch counter\n\
+      \simplex_smp_messages_ntf_successes_batch " <> mshow _msgNtfsB <> "\n# msgNtfsB\n\
+      \\n\
+      \# HELP simplex_smp_messages_ntf_errors Error events with notification messages (to ntf server). \n\
+      \# TYPE simplex_smp_messages_ntf_errors counter\n\
+      \simplex_smp_messages_ntf_errors{type=\"nosub\"} " <> mshow _msgNtfNoSub <> "\n# msgNtfNoSub\n\
+      \simplex_smp_messages_ntf_errors{type=\"lost\"} " <> mshow _msgNtfLost <> "\n# msgNtfLost\n\
+      \simplex_smp_messages_ntf_errors{type=\"expired\"} " <> mshow _msgNtfExpired <> "\n# msgNtfExpired\n\
+      \\n\
+      \# HELP simplex_smp_subscription_ntf_requests Subscription requests with notification flag (from ntf server). \n\
+      \# TYPE simplex_smp_subscription_ntf_requests counter\n\
+      \simplex_smp_subscription_ntf_requests " <> mshow _ntfSub <> "\n# ntfSub\n\
+      \\n\
+      \# HELP simplex_smp_subscription_ntf_requests_batch Batched subscription requests with notification flag (from ntf server). \n\
+      \# TYPE simplex_smp_subscription_ntf_requests_batch counter\n\
+      \simplex_smp_subscription_ntf_requests_batch " <> mshow _ntfSubB <> "\n# ntfSubB\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_ntf_errors Subscription errors with notification flag (from ntf server). \n\
+      \# TYPE simplex_smp_subscribtion_ntf_errors counter\n\
+      \simplex_smp_subscribtion_ntf_errors{type=\"auth\"} " <> mshow _ntfSubAuth <> "\n# ntfSubAuth\n\
+      \simplex_smp_subscribtion_ntf_errors{type=\"duplicate\"} " <> mshow _ntfSubDuplicate <> "\n# ntfSubDuplicate\n\
+      \\n"
+    relays =
+      "# Relays\n\
+      \# ------\n\
+      \\n\
+      \# HELP simplex_smp_relay_sessions_requests Session requests through relay.\n\
+      \# TYPE simplex_smp_relay_sessions_requests counter\n\
+      \simplex_smp_relay_sessions_requests{source=\"all\"} " <> mshow (_pRequests _pRelays) <> "\n# pRelays_pRequests\n\
+      \simplex_smp_relay_sessions_requests{source=\"own\"} " <> mshow (_pRequests _pRelaysOwn) <> "\n# pRelaysOwn_pRequests\n\
+      \\n\
+      \# HELP simplex_smp_relay_sessions_successes Successful session events through relay.\n\
+      \# TYPE simplex_smp_relay_sessions_successes counter\n\
+      \simplex_smp_relay_sessions_successes{source=\"all\"} " <> mshow (_pSuccesses _pRelays) <> "\n# pRelays_pSuccesses\n\
+      \simplex_smp_relay_sessions_successes{source=\"own\"} " <> mshow (_pSuccesses _pRelaysOwn) <> "\n# pRelaysOwn_pSuccesses\n\
+      \\n\
+      \# HELP simplex_smp_relay_sessions_errors Error session events through relay.\n\
+      \# TYPE simplex_smp_relay_sessions_errors counter\n\
+      \simplex_smp_relay_sessions_errors{source=\"all\",type=\"connect\"} " <> mshow (_pErrorsConnect _pRelays) <> "\n# pRelays_pErrorsConnect\n\
+      \simplex_smp_relay_sessions_errors{source=\"all\",type=\"compat\"} " <> mshow (_pErrorsCompat _pRelays) <> "\n# pRelays_pErrorsCompat\n\
+      \simplex_smp_relay_sessions_errors{source=\"all\",type=\"other\"} " <> mshow (_pErrorsOther _pRelays) <> "\n# pRelays_pErrorsOther\n\
+      \simplex_smp_relay_sessions_errors{source=\"own\",type=\"connect\"} " <> mshow (_pErrorsConnect _pRelaysOwn) <> "\n# pRelaysOwn_pErrorsConnect\n\
+      \simplex_smp_relay_sessions_errors{source=\"own\",type=\"compat\"} " <> mshow (_pErrorsCompat _pRelaysOwn) <> "\n# pRelaysOwn_pErrorsCompat\n\
+      \simplex_smp_relay_sessions_errors{source=\"own\",type=\"other\"} " <> mshow (_pErrorsOther _pRelaysOwn) <> "\n# pRelaysOwn_pErrorsOther\n\
+      \\n\
+      \# HELP simplex_smp_relay_messages_requests Message requests sent through relay.\n\
+      \# TYPE simplex_smp_relay_messages_requests counter\n\
+      \simplex_smp_relay_messages_requests{source=\"all\"} " <> mshow (_pRequests _pMsgFwds) <> "\n# pMsgFwds_pRequests\n\
+      \simplex_smp_relay_messages_requests{source=\"own\"} " <> mshow (_pRequests _pMsgFwdsOwn) <> "\n# pMsgFwdsOwn_pRequests\n\
+      \\n\
+      \# HELP simplex_smp_relay_messages_successes Successful messages sent through relay.\n\
+      \# TYPE simplex_smp_relay_messages_successes counter\n\
+      \simplex_smp_relay_messages_successes{source=\"all\"} " <> mshow (_pSuccesses _pMsgFwds) <> "\n# pMsgFwds_pSuccesses\n\
+      \simplex_smp_relay_messages_successes{source=\"own\"} " <> mshow (_pSuccesses _pMsgFwdsOwn) <> "\n# pMsgFwdsOwn_pSuccesses\n\
+      \\n\
+      \# HELP simplex_smp_relay_messages_errors Error events with messages sent through relay.\n\
+      \# TYPE simplex_smp_relay_messages_errors counter\n\
+      \simplex_smp_relay_messages_errors{source=\"all\",type=\"connect\"} " <> mshow (_pErrorsConnect _pMsgFwds) <> "\n# pMsgFwds_pErrorsConnect\n\
+      \simplex_smp_relay_messages_errors{source=\"all\",type=\"compat\"} " <> mshow (_pErrorsCompat _pMsgFwds) <> "\n# pMsgFwds_pErrorsCompat\n\
+      \simplex_smp_relay_messages_errors{source=\"all\",type=\"other\"} " <> mshow (_pErrorsOther _pMsgFwds) <> "\n# pMsgFwds_pErrorsOther\n\
+      \simplex_smp_relay_messages_errors{source=\"own\",type=\"connect\"} " <> mshow (_pErrorsConnect _pMsgFwdsOwn) <> "\n# pMsgFwdsOwn_pErrorsConnect\n\
+      \simplex_smp_relay_messages_errors{source=\"own\",type=\"compat\"} " <> mshow (_pErrorsCompat _pMsgFwdsOwn) <> "\n# pMsgFwdsOwn_pErrorsCompat\n\
+      \simplex_smp_relay_messages_errors{source=\"own\",type=\"other\"} " <> mshow (_pErrorsOther _pMsgFwdsOwn) <> "\n# pMsgFwdsOwn_pErrorsOther\n\
+      \\n\
+      \# HELP simplex_smp_relay_messages_received Relay messages statistics.\n\
+      \# TYPE simplex_smp_relay_messages_received counter\n\
+      \simplex_smp_relay_messages_received " <> mshow _pMsgFwdsRecv <> "\n# pMsgFwdsRecv\n\
+      \\n"
+    info =
+      "# Info\n\
+      \# ----\n\
+      \\n"
+      <> socketsMetric socketsAccepted "simplex_smp_sockets_accepted" "Accepted sockets"
+      <> socketsMetric socketsClosed "simplex_smp_sockets_closed" "Closed sockets"
+      <> socketsMetric socketsActive "simplex_smp_sockets_active" "Active sockets"
+      <> socketsMetric socketsLeaked "simplex_smp_sockets_leaked" "Leaked sockets"
+      <> "# HELP simplex_smp_threads_total Threads\n\
+      \# TYPE simplex_smp_threads_total gauge\n\
+      \simplex_smp_threads_total " <> mshow threadsCount <> "\n\
+      \\n\
+      \# HELP simplex_smp_clients_total Clients\n\
+      \# TYPE simplex_smp_clients_total gauge\n\
+      \simplex_smp_clients_total " <> mshow clientsCount <> "\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_total Total subscriptions\n\
+      \# TYPE simplex_smp_subscribtion_total gauge\n\
+      \simplex_smp_subscribtion_total " <> mshow smpSubsCount <> "\n# smpSubs\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_clients_total Subscribed clients, first counting method\n\
+      \# TYPE simplex_smp_subscribtion_clients_total gauge\n\
+      \simplex_smp_subscribtion_clients_total " <> mshow smpSubClientsCount <> "\n# smpSubClients\n\
+      \\n\
+      \# HELP simplex_smp_subscription_ntf_total Total notification subscripbtions (from ntf server)\n\
+      \# TYPE simplex_smp_subscription_ntf_total gauge\n\
+      \simplex_smp_subscription_ntf_total " <> mshow ntfSubsCount <> "\n# ntfSubs\n\
+      \\n\
+      \# HELP simplex_smp_subscription_ntf_clients_total Total subscribed NTF servers, first counting method\n\
+      \# TYPE simplex_smp_subscription_ntf_clients_total gauge\n\
+      \simplex_smp_subscription_ntf_clients_total " <> mshow ntfSubClientsCount <> "\n# ntfSubClients\n"
+    socketsMetric :: (SocketStats -> Int) -> Text -> Text -> Text
+    socketsMetric sel metric descr =
+      "# HELP " <> metric <> " " <> descr <> "\n"
+        <> "# TYPE " <> metric <> " gauge\n"
+        <> T.concat (map (\(port, ss) -> metric <> "{port=\"" <> T.pack port <> "\"} " <> mshow (sel ss) <> "\n") socketStats)
+        <> "\n"
+    mstr a = T.pack a <> " " <> tsEpoch
+    mshow :: Show a => a -> Text
+    mshow = mstr . show
+    tsEpoch = T.pack $ show @Int64 $ floor @Double $ realToFrac (ts `diffUTCTime` epoch) * 1000
+    epoch = UTCTime systemEpochDay 0
+{-# FOURMOLU_ENABLE\n#-}
@@ -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_)

@@ -638,6 +638,14 @@ data PeriodStatCounts = PeriodStatCounts
    monthCount :: String
  }

+periodStatDataCounts :: PeriodStatsData -> PeriodStatCounts
+periodStatDataCounts PeriodStatsData {_day, _week, _month} =
+  PeriodStatCounts
+    { dayCount = show $ IS.size _day,
+      weekCount = show $ IS.size _week,
+      monthCount = show $ IS.size _month
+    }
+
 periodStatCounts :: PeriodStats -> UTCTime -> IO PeriodStatCounts
 periodStatCounts ps ts = do
  let d = utctDay ts
@@ -12,7 +12,9 @@ module Simplex.Messaging.Transport.Server
    runTransportServerState,
    runTransportServerState_,
    SocketState,
+    SocketStats (..),
    newSocketState,
+    getSocketStats,
    runTransportServer,
    runTransportServerSocket,
    runLocalTCPServer,
@@ -163,9 +165,24 @@ safeAccept sock =

 type SocketState = (TVar Int, TVar Int, TVar (IntMap (Weak ThreadId)))

+data SocketStats = SocketStats
+  { socketsAccepted :: Int,
+    socketsClosed :: Int,
+    socketsActive :: Int,
+    socketsLeaked :: Int
+  }
+
 newSocketState :: IO SocketState
 newSocketState = (,,) <$> newTVarIO 0 <*> newTVarIO 0 <*> newTVarIO mempty

+getSocketStats :: SocketState -> IO SocketStats
+getSocketStats (accepted, closed, active) = do
+  socketsAccepted <- readTVarIO accepted
+  socketsClosed <- readTVarIO closed
+  socketsActive <- IM.size <$> readTVarIO active
+  let socketsLeaked = socketsAccepted - socketsClosed - socketsActive
+  pure SocketStats {socketsAccepted, socketsClosed, socketsActive, socketsLeaked}
+
 closeServer :: TMVar Bool -> TVar (IntMap (Weak ThreadId)) -> Socket -> IO ()
 closeServer started clients sock = do
  close sock
@@ -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
@@ -3076,8 +3076,9 @@ testTwoUsers = withAgentClients2 $ \a b -> do
    ("", "", DOWN _ _) <- nGet a
    ("", "", DOWN _ _) <- nGet a
    ("", "", DOWN _ _) <- nGet a
-    ("", "", DOWN _ _) <- nGet a
-    ("", "", UP _ _) <- nGet a
+    -- to avoice race condition
+    nGet a =##> \case ("", "", DOWN _ _) -> True; ("", "", UP _ _) -> True; _ -> False
+    nGet a =##> \case ("", "", UP _ _) -> True; ("", "", DOWN _ _) -> True; _ -> False
    ("", "", UP _ _) <- nGet a
    ("", "", UP _ _) <- nGet a
    ("", "", UP _ _) <- nGet a
@@ -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://[email protected]"
+
+testOp1Srv2 :: ProtoServerWithAuth 'PSMP
+testOp1Srv2 = "smp://[email protected]"
+
+testOp2Srv1 :: ProtoServerWithAuth 'PSMP
+testOp2Srv1 = "smp://[email protected]"
+
+testOp2Srv2 :: ProtoServerWithAuth 'PSMP
+testOp2Srv2 = "smp://[email protected]"
+
+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
@@ -77,7 +77,8 @@ testJournalStoreCfg =
      quota = 3,
      maxMsgCount = 4,
      maxStateLines = 2,
-      stateTailSize = 256
+      stateTailSize = 256,
+      idleInterval = 21600
    }

 mkMessage :: MonadIO m => ByteString -> m Message
@@ -66,7 +66,7 @@ testRetryIntervalSameMode =

 testRetryIntervalSwitchMode :: Spec
 testRetryIntervalSwitchMode =
-  it "should increase elapased time and interval when the mode stays the same" $ do
+  it "should increase elapased time and interval when the mode switches" $ do
    lock <- newEmptyTMVarIO
    intervals <- newTVarIO []
    reportedIntervals <- newTVarIO []
@@ -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
@@ -43,6 +43,9 @@ import Util
 testHost :: NonEmpty TransportHost
 testHost = "localhost"

+testHost2 :: NonEmpty TransportHost
+testHost2 = "127.0.0.1"
+
 testPort :: ServiceName
 testPort = "5001"

@@ -76,6 +79,9 @@ testStoreNtfsFile = "tests/tmp/smp-server-ntfs.log"
 testStoreNtfsFile2 :: FilePath
 testStoreNtfsFile2 = "tests/tmp/smp-server-ntfs.log.2"

+testPrometheusMetricsFile :: FilePath
+testPrometheusMetricsFile = "tests/tmp/smp-server-metrics.txt"
+
 testServerStatsBackupFile :: FilePath
 testServerStatsBackupFile = "tests/tmp/smp-server-stats.log"

@@ -133,12 +139,15 @@ cfgMS msType =
      controlPortAdminAuth = Nothing,
      messageExpiration = Just defaultMessageExpiration,
      expireMessagesOnStart = True,
+      idleQueueInterval = defaultIdleQueueInterval,
      notificationExpiration = defaultNtfExpiration,
      inactiveClientExpiration = Just defaultInactiveClientExpiration,
      logStatsInterval = Nothing,
      logStatsStartTime = 0,
-      serverStatsLogFile = "tests/smp-server-stats.daily.log",
+      serverStatsLogFile = "tests/tmp/smp-server-stats.daily.log",
      serverStatsBackupFile = Nothing,
+      prometheusInterval = Nothing,
+      prometheusMetricsFile = testPrometheusMetricsFile,
      pendingENDInterval = 500000,
      ntfDeliveryInterval = 200000,
      smpCredentials =
@@ -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
@@ -46,7 +46,7 @@ import Simplex.Messaging.Server.StoreLog (closeStoreLog)
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Util (whenM)
 import Simplex.Messaging.Version (mkVersionRange)
-import System.Directory (doesDirectoryExist, removeDirectoryRecursive, removeFile)
+import System.Directory (doesDirectoryExist, doesFileExist, removeDirectoryRecursive, removeFile)
 import System.TimeIt (timeItT)
 import System.Timeout
 import Test.HUnit
@@ -71,6 +71,7 @@ serverTests  = do
  describe "Store log" testWithStoreLog
  describe "Restore messages" testRestoreMessages
  describe "Restore messages (old / v2)" testRestoreExpireMessages
+  describe "Save prometheus metrics" testPrometheusMetrics
  describe "Timing of AUTH error" testTiming
  describe "Message notifications" testMessageNotifications
  describe "Message expiration" $ do
@@ -825,6 +826,13 @@ testRestoreExpireMessages =
    runClient :: Transport c => TProxy c -> (THandleSMP c 'TClient -> IO ()) -> Expectation
    runClient _ test' = testSMPClient test' `shouldReturn` ()

+testPrometheusMetrics :: SpecWith (ATransport, AMSType)
+testPrometheusMetrics =
+  it "should save Prometheus metrics" $ \(at, msType) -> do
+    let cfg' = (cfgMS msType) {prometheusInterval = Just 1}
+    withSmpServerConfigOn at cfg' testPort $ \_ -> threadDelay 1000000
+    doesFileExist testPrometheusMetricsFile `shouldReturn` True
+
 createAndSecureQueue :: Transport c => THandleSMP c 'TClient -> SndPublicAuthKey -> IO (SenderId, RecipientId, RcvPrivateAuthKey, RcvDhSecret)
 createAndSecureQueue h sPub = do
  g <- C.newRandom
@@ -972,7 +980,7 @@ testMsgExpireOnInterval =
  xit' "should expire messages that are not received before messageTTL after expiry interval" $ \(ATransport (t :: TProxy c), msType) -> do
    g <- C.newRandom
    (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
-    let cfg' = (cfgMS msType) {messageExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 1}}
+    let cfg' = (cfgMS msType) {messageExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 1}, idleQueueInterval = 1}
    withSmpServerConfigOn (ATransport t) cfg' testPort $ \_ ->
      testSMPClient @c $ \sh -> do
        (sId, rId, rKey, _) <- testSMPClient @c $ \rh -> createAndSecureQueue rh sPub
Author	SHA1	Message	Date
Evgeny Poberezkin	4e4f8f2ad1	Merge branch 'master' into ep/disable-background	2024-12-20 13:22:04 +00:00
Evgeny Poberezkin	f5cef2556b	build: move type to client_library module	2024-12-20 13:12:11 +00:00
Evgeny Poberezkin	fef63f9083	version	2024-12-20 12:12:32 +00:00
Evgeny Poberezkin	457575f86f	Merge branch 'master' into ep/disable-background	2024-12-20 12:10:18 +00:00
sh	77a5ed2ec6	6.2.1.0 (#1424 )	2024-12-19 20:45:17 +00:00
Evgenyandsh	4d640c16aa	smp server: log prometheus metrics (#1411 ) * smp server: log prometheus metrics * save metrics * diff * lines * version * do not include Prometheus into client * corrections Co-authored-by: sh <[email protected]> * corrections Co-authored-by: sh <[email protected]> * corrections Co-authored-by: sh <[email protected]> * add timestamp to metrics * remove type * remove version --------- Co-authored-by: sh <[email protected]>	2024-12-18 11:19:13 +00:00
Evgeny Poberezkin	8647c24a95	agent: disable background processes	2024-12-14 15:24:56 +00:00
Evgeny	0a827307d6	build: remove some modules from the client build (#1418 ) * build: remove some modules from the client build * remove websockets from client_library	2024-12-14 15:22:06 +00:00
Evgeny Poberezkin	79e9447b73	6.2.0.7	2024-12-07 14:36:50 +00:00
spaced4ndy	9893935e7c	agent: fix acceptContact inconsistent state after app restart (#1412 )	2024-12-05 19:27:21 +04:00
Evgeny Poberezkin	966b9990e0	6.2.0.6	2024-12-04 19:57:11 +00:00
spaced4ndy	bef1e38295	xftp: add debug info to "too many recipients" error (#1410 )	2024-12-04 19:56:33 +00:00
Evgeny Poberezkin	4b43cb8054	6.2.0.5	2024-12-03 12:33:32 +00:00
Evgeny	38ad3c046e	agent: function to check if contact address is the same (#1409 )	2024-12-02 11:41:06 +00:00
Evgeny	b2afe6f40c	rfc: moving queue records to journal storage and queue blobs, expiring messages in journal storage (#1407 ) * rfc: moving queue records to journal storage and queue blobs * rfc: expiring messages in journal storage * add comment	2024-11-28 12:22:52 +00:00
Evgeny Poberezkin	601620bdde	6.2.0.4	2024-11-22 19:50:04 +00:00
Evgeny	97104988a3	server: support server roles and operators (#1343 ) * server: support server roles and operators * make server operator optional * allRoles * fix test * different server host in tests * remove ServerCfg fields used only in UI * comments * choose different server for invitation when connecting via address * fix test in ghc8107 * simplify	2024-11-22 16:43:37 +00:00
Evgeny	45333bd340	smp server: do not open/read journal message queues that are known to be empty (#1406 ) * smp server: do not open/read journal message queues that are known to be empty * cleanup * version * close empty queues on first subscription * revert version	2024-11-22 12:29:31 +00:00
Evgeny	bbcb1abfda	agent: treat blocked STM and other critical errors that offer restart as temporary (#1405 )	2024-11-21 22:01:19 +00:00
Evgeny Poberezkin	a64c1aa2c4	6.2.0.3	2024-11-17 09:45:17 +00:00
Evgeny	21fbbf9106	smp server: update message counts during message expiration, increase idle interval (#1404 ) * smp server: update message counts during message expiration, increase idle interval * version * fix * flip results * version	2024-11-17 09:44:40 +00:00
Evgeny	17a0be10fa	smp server: expire messages in idle message queues (including not opened) (#1403 ) * smp server: expire messages in idle message queues (including not opened) * use message expiration interval * simpler * version * remove version	2024-11-15 10:26:24 +00:00