Merge branch 'master' into ep/smp-rcv-service

src/Simplex/FileTransfer/Server.hs

@@ -45,6 +45,7 @@ import Network.Socket
 import Simplex.FileTransfer.Protocol
 import Simplex.FileTransfer.Server.Control
 import Simplex.FileTransfer.Server.Env
+import Simplex.FileTransfer.Server.Prometheus
 import Simplex.FileTransfer.Server.Stats
 import Simplex.FileTransfer.Server.Store
 import Simplex.FileTransfer.Server.StoreLog
@@ -54,7 +55,7 @@ import qualified Simplex.Messaging.Crypto.Lazy as LC
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (BlockingInfo, EntityId (..), RcvPublicAuthKey, RcvPublicDhKey, RecipientId, SignedTransmission, pattern NoEntity)
-import Simplex.Messaging.Server (dummyVerifyCmd, verifyCmdAuthorization)
+import Simplex.Messaging.Server (controlPortAuth, dummyVerifyCmd, verifyCmdAuthorization)
 import Simplex.Messaging.Server.Control (CPClientRole (..))
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.QueueStore (RoundedSystemTime, ServerEntityStatus (..), getRoundedSystemTime)
@@ -69,6 +70,7 @@ import Simplex.Messaging.Transport.HTTP2.Server
 import Simplex.Messaging.Transport.Server (runLocalTCPServer)
 import Simplex.Messaging.Util
 import Simplex.Messaging.Version
+import System.Environment (lookupEnv)
 import System.Exit (exitFailure)
 import System.FilePath ((</>))
 import System.IO (hPrint, hPutStrLn, universalNewlineMode)
@@ -105,7 +107,14 @@ xftpServer :: XFTPServerConfig -> TMVar Bool -> M ()
 xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpiration, fileExpiration, xftpServerVRange} started = do
   mapM_ (expireServerFiles Nothing) fileExpiration
   restoreServerStats
-  raceAny_ (runServer : expireFilesThread_ cfg <> serverStatsThread_ cfg <> controlPortThread_ cfg) `finally` stopServer
+  raceAny_
+    ( runServer
+        : expireFilesThread_ cfg
+          <> serverStatsThread_ cfg
+          <> prometheusMetricsThread_ cfg
+          <> controlPortThread_ cfg
+    )
+    `finally` stopServer
   where
     runServer :: M ()
     runServer = do
@@ -240,6 +249,30 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira
               ]
         liftIO $ threadDelay' interval
 
+    prometheusMetricsThread_ :: XFTPServerConfig -> [M ()]
+    prometheusMetricsThread_ XFTPServerConfig {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
+      ss <- asks serverStats
+      rtsOpts <- liftIO $ maybe ("set " <> rtsOptionsEnv) T.pack <$> lookupEnv (T.unpack rtsOptionsEnv)
+      let interval = 1000000 * saveInterval
+      liftIO $ forever $ do
+        threadDelay interval
+        ts <- getCurrentTime
+        sm <- getFileServerMetrics ss rtsOpts
+        T.writeFile metricsFile $ xftpPrometheusMetrics sm ts
+
+    getFileServerMetrics :: FileServerStats -> T.Text -> IO FileServerMetrics
+    getFileServerMetrics ss rtsOptions = do
+      d <- getFileServerStatsData ss
+      let fd = periodStatDataCounts $ _filesDownloaded d
+      pure FileServerMetrics {statsData = d, filesDownloadedPeriods = fd, rtsOptions}
+
     controlPortThread_ :: XFTPServerConfig -> [M ()]
     controlPortThread_ XFTPServerConfig {controlPort = Just port} = [runCPServer port]
     controlPortThread_ _ = []
@@ -277,12 +310,9 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira
                   CPSkip -> False
                   _ -> True
             processCP h role = \case
-              CPAuth auth -> atomically $ writeTVar role $! newRole cfg
+              CPAuth auth -> controlPortAuth h user admin role auth
                 where
-                  newRole XFTPServerConfig {controlPortUserAuth = user, controlPortAdminAuth = admin}
-                    | Just auth == admin = CPRAdmin
-                    | Just auth == user = CPRUser
-                    | otherwise = CPRNone
+                  XFTPServerConfig {controlPortUserAuth = user, controlPortAdminAuth = admin} = cfg
               CPStatsRTS -> E.tryAny getRTSStats >>= either (hPrint h) (hPrint h)
               CPDelete fileId -> withUserRole $ unliftIO u $ do
                 fs <- asks store

src/Simplex/FileTransfer/Server/Env.hs

@@ -64,6 +64,8 @@ data XFTPServerConfig = XFTPServerConfig
     logStatsStartTime :: Int64,
     serverStatsLogFile :: FilePath,
     serverStatsBackupFile :: Maybe FilePath,
+    prometheusInterval :: Maybe Int,
+    prometheusMetricsFile :: FilePath,
     transportConfig :: TransportServerConfig,
     responseDelay :: Int
   }

src/Simplex/FileTransfer/Server/Main.hs

@@ -30,7 +30,7 @@ import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Transport (simplexMQVersion)
 import Simplex.Messaging.Transport.Client (TransportHost (..))
 import Simplex.Messaging.Transport.Server (ServerCredentials (..), mkTransportServerConfig)
-import Simplex.Messaging.Util (safeDecodeUtf8, tshow)
+import Simplex.Messaging.Util (eitherToMaybe, safeDecodeUtf8, tshow)
 import System.Directory (createDirectoryIfMissing, doesFileExist)
 import System.FilePath (combine)
 import System.IO (BufferMode (..), hSetBuffering, stderr, stdout)
@@ -89,6 +89,9 @@ xftpServerCLI cfgPath logPath = do
             <> "# Expire files after the specified number of hours.\n"
             <> ("expire_files_hours: " <> tshow defFileExpirationHours <> "\n\n")
             <> "log_stats: off\n\
+               \\n\
+               \# Log interval for real-time Prometheus metrics\n\
+               \# prometheus_interval: 60\n\
                \\n\
                \[AUTH]\n\
                \# Set new_files option to off to completely prohibit uploading new files.\n\
@@ -188,6 +191,8 @@ xftpServerCLI cfgPath logPath = do
               logStatsStartTime = 0, -- seconds from 00:00 UTC
               serverStatsLogFile = combine logPath "file-server-stats.daily.log",
               serverStatsBackupFile = logStats $> combine logPath "file-server-stats.log",
+              prometheusInterval = eitherToMaybe $ read . T.unpack <$> lookupValue "STORE_LOG" "prometheus_interval" ini,
+              prometheusMetricsFile = combine logPath "xftp-server-metrics.txt",
               transportConfig =
                 mkTransportServerConfig
                   (fromMaybe False $ iniOnOff "TRANSPORT" "log_tls_errors" ini)

src/Simplex/FileTransfer/Server/Prometheus.hs

@@ -0,0 +1,134 @@
+{-# LANGUAGE DuplicateRecordFields #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE TypeApplications #-}
+{-# OPTIONS_GHC -fno-warn-unrecognised-pragmas #-}
+
+module Simplex.FileTransfer.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 Simplex.FileTransfer.Server.Stats
+import Simplex.Messaging.Server.Stats (PeriodStatCounts (..))
+import Simplex.Messaging.Transport (simplexMQVersion)
+import Simplex.Messaging.Util (tshow)
+
+data FileServerMetrics = FileServerMetrics
+  { statsData :: FileServerStatsData,
+    filesDownloadedPeriods :: PeriodStatCounts,
+    rtsOptions :: Text
+  }
+
+rtsOptionsEnv :: Text
+rtsOptionsEnv = "XFTP_RTS_OPTIONS"
+
+{-# FOURMOLU_DISABLE\n#-}
+xftpPrometheusMetrics :: FileServerMetrics -> UTCTime -> Text
+xftpPrometheusMetrics sm ts =
+  time <> files <> info
+  where
+    FileServerMetrics {statsData, filesDownloadedPeriods, rtsOptions} = sm
+    FileServerStatsData
+      { _fromTime,
+        _filesCreated,
+        _fileRecipients,
+        _filesUploaded,
+        _filesExpired,
+        _filesDeleted,
+        _filesBlocked,
+        _fileDownloads,
+        _fileDownloadAcks,
+        _filesCount,
+        _filesSize
+      } = statsData
+    time =
+      "# Recorded at: " <> T.pack (iso8601Show ts) <> "\n\
+      \# Stats from: " <> T.pack (iso8601Show _fromTime) <> "\n\
+      \\n"
+    files =
+      "# Files\n\
+      \# -----\n\
+      \\n\
+      \# HELP simplex_xftp_files_created Created files\n\
+      \# TYPE simplex_xftp_files_created counter\n\
+      \simplex_xftp_files_created " <> mshow _filesCreated <> "\n\
+      \# filesCreated\n\
+      \\n\
+      \# HELP simplex_xftp_files_recipients Files recipients\n\
+      \# TYPE simplex_xftp_files_recipients counter\n\
+      \simplex_xftp_files_recipients " <> mshow _fileRecipients <> "\n\
+      \# fileRecipients\n\
+      \\n\
+      \# HELP simplex_xftp_files_uploaded Uploaded files\n\
+      \# TYPE simplex_xftp_files_uploaded counter\n\
+      \simplex_xftp_files_uploaded " <> mshow _filesUploaded <> "\n\
+      \# filesUploaded\n\
+      \\n\
+      \# HELP simplex_xftp_files_expired Expired files\n\
+      \# TYPE simplex_xftp_files_expired counter\n\
+      \simplex_xftp_files_expired " <> mshow _filesExpired <> "\n\
+      \# filesExpired\n\
+      \\n\
+      \# HELP simplex_xftp_files_deleted Deleted files\n\
+      \# TYPE simplex_xftp_files_deleted counter\n\
+      \simplex_xftp_files_deleted " <> mshow _filesDeleted <> "\n\
+      \# filesDeleted\n\
+      \\n\
+      \# HELP simplex_xftp_files_blocked Blocked files\n\
+      \# TYPE simplex_xftp_files_blocked counter\n\
+      \simplex_xftp_files_blocked " <> mshow _filesBlocked <> "\n\
+      \# filesBlocked\n\
+      \\n\
+      \# HELP simplex_xftp_file_downloads File downloads\n\
+      \# TYPE simplex_xftp_file_downloads counter\n\
+      \simplex_xftp_file_downloads " <> mshow _fileDownloads <> "\n\
+      \# fileDownloads\n\
+      \\n\
+      \# HELP simplex_xftp_file_download_acks File download ACKs\n\
+      \# TYPE simplex_xftp_file_download_acks counter\n\
+      \simplex_xftp_file_download_acks " <> mshow _fileDownloadAcks <> "\n\
+      \# fileDownloadAcks\n\
+      \\n\
+      \# HELP simplex_xftp_files_count_total Total files count \n\
+      \# TYPE simplex_xftp_files_count_total gauge\n\
+      \simplex_xftp_files_count_total " <> mshow _filesCount <> "\n\
+      \# filesCount\n\
+      \\n\
+      \# HELP simplex_xftp_files_size Size of files \n\
+      \# TYPE simplex_xftp_files_size gauge\n\
+      \simplex_xftp_files_size " <> mshow _filesSize <> "\n\
+      \# filesSize \n\
+      \\n\
+      \# HELP simplex_xftp_files_count_daily Daily files count\n\
+      \# TYPE simplex_xftp_files_count_daily gauge\n\
+      \simplex_xftp_files_count_daily " <> mstr (dayCount filesDownloadedPeriods) <> "\n\
+      \# filesDownloaded.dayCount\n\
+      \\n\
+      \# HELP simplex_xftp_files_count_weekly Weekly files count\n\
+      \# TYPE simplex_xftp_files_count_weekly gauge\n\
+      \simplex_xftp_files_count_weekly " <> mstr (weekCount filesDownloadedPeriods) <> "\n\
+      \# filesDownloaded.weekCount\n\
+      \\n\
+      \# HELP simplex_xftp_files_count_monthly Monthly files count\n\
+      \# TYPE simplex_xftp_files_count_monthly gauge\n\
+      \simplex_xftp_files_count_monthly " <> mstr (monthCount filesDownloadedPeriods) <> "\n\
+      \# filesDownloaded.monthCount\n\
+      \\n"
+    info =
+      "# Info\n\
+      \# ----\n\
+      \\n\
+      \# HELP simplex_xftp_info Server information. RTS options have to be passed via " <> rtsOptionsEnv <> " env var\n\
+      \# TYPE simplex_xftp_info gauge\n\
+      \simplex_xftp_info{version=\"" <> T.pack simplexMQVersion <> "\",rts_options=\"" <> rtsOptions <> "\"} 1\n\
+      \\n"
+    mstr a = a <> " " <> tsEpoch
+    mshow :: Show a => a -> Text
+    mshow = mstr . tshow
+    tsEpoch = tshow @Int64 $ floor @Double $ realToFrac (ts `diffUTCTime` epoch) * 1000
+    epoch = UTCTime systemEpochDay 0
+{-# FOURMOLU_ENABLE\n#-}

src/Simplex/Messaging/Agent.hs

@@ -912,12 +912,19 @@ getConnShortLink' c nm userId = \case
   where
     decryptData :: ConnectionModeI c => SMPServer -> LinkKey -> C.SbKey -> (SMP.SenderId, QueueLinkData) -> AM (ConnectionRequestUri c, ConnLinkData c)
     decryptData srv linkKey k (sndId, d) = do
-      r@(cReq, _) <- liftEither $ SL.decryptLinkData @c linkKey k d
+      r@(cReq, clData) <- liftEither $ SL.decryptLinkData @c linkKey k d
       let (srv', sndId') = qAddress (connReqQueue cReq)
       unless (srv `sameSrvHost` srv' && sndId == sndId') $
         throwE $ AGENT $ A_LINK "different address"
-      pure r
+      pure $ if srv' == srv then r else (updateConnReqServer srv cReq, clData)
     sameSrvHost ProtocolServer {host = h :| _} ProtocolServer {host = hs} = h `elem` hs
+    updateConnReqServer :: SMPServer -> ConnectionRequestUri c -> ConnectionRequestUri c
+    updateConnReqServer srv = \case
+      CRInvitationUri crData e2eParams -> CRInvitationUri (updateQueues crData) e2eParams
+      CRContactUri crData -> CRContactUri $ updateQueues crData
+      where
+        updateQueues crData@(ConnReqUriData {crSmpQueues = SMPQueueUri vr addr :| qs}) =
+          crData {crSmpQueues = SMPQueueUri vr addr {smpServer = srv} :| qs}
 
 deleteLocalInvShortLink' :: AgentClient -> ConnShortLink 'CMInvitation -> AM ()
 deleteLocalInvShortLink' c (CSLInvitation _ srv linkId _) = withStore' c $ \db -> deleteInvShortLink db srv linkId

src/Simplex/Messaging/Notifications/Server.hs

@@ -364,12 +364,9 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg, startOptions}
                   CPSkip -> False
                   _ -> True
             processCP h role = \case
-              CPAuth auth -> atomically $ writeTVar role $! newRole cfg
+              CPAuth auth -> controlPortAuth h user admin role auth
                 where
-                  newRole NtfServerConfig {controlPortUserAuth = user, controlPortAdminAuth = admin}
-                    | Just auth == admin = CPRAdmin
-                    | Just auth == user = CPRUser
-                    | otherwise = CPRNone
+                  NtfServerConfig {controlPortUserAuth = user, controlPortAdminAuth = admin} = cfg
               CPStats -> withUserRole $ do
                 ss <- unliftIO u $ asks serverStats
                 let getStat :: (NtfServerStats -> IORef a) -> IO a

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

@@ -167,6 +167,9 @@ ntfServerCLI cfgPath logPath =
             <> "Time to retain deleted entities in the database, days.\n"
             <> ("# db_deleted_ttl: " <> tshow defaultDeletedTTL <> "\n\n")
             <> "log_stats: off\n\n\
+               \# Log interval for real-time Prometheus metrics\n\
+               \# prometheus_interval: 60\n\
+               \\n\
                \[AUTH]\n\
                \# control_port_admin_password:\n\
                \# control_port_user_password:\n\

src/Simplex/Messaging/Server.hs

@@ -31,6 +31,7 @@
 module Simplex.Messaging.Server
   ( runSMPServer,
     runSMPServerBlocking,
+    controlPortAuth,
     importMessages,
     exportMessages,
     printMessageStats,
@@ -558,6 +559,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
           msgNtfNoSub' <- atomicSwapIORef (msgNtfNoSub ss) 0
           msgNtfLost' <- atomicSwapIORef (msgNtfLost ss) 0
           msgNtfExpired' <- atomicSwapIORef (msgNtfExpired ss) 0
+          _qBlocked <- atomicSwapIORef (qBlocked ss) 0 -- not logged, only reset
           pRelays' <- getResetProxyStatsData pRelays
           pRelaysOwn' <- getResetProxyStatsData pRelaysOwn
           pMsgFwds' <- getResetProxyStatsData pMsgFwds
@@ -770,12 +772,9 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
                   CPSkip -> False
                   _ -> True
             processCP h role = \case
-              CPAuth auth -> atomically $ writeTVar role $! newRole cfg
+              CPAuth auth -> controlPortAuth h user admin role auth
                 where
-                  newRole ServerConfig {controlPortUserAuth = user, controlPortAdminAuth = admin}
-                    | Just auth == admin = CPRAdmin
-                    | Just auth == user = CPRUser
-                    | otherwise = CPRNone
+                  ServerConfig {controlPortUserAuth = user, controlPortAdminAuth = admin} = cfg
               CPSuspend -> withAdminRole $ hPutStrLn h "suspend not implemented"
               CPResume -> withAdminRole $ hPutStrLn h "resume not implemented"
               CPClients -> withAdminRole $ do
@@ -964,41 +963,53 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
                               SubPending -> (c1, c2 + 1, c3, c4)
                               SubThread _ -> (c1, c2, c3 + 1, c4)
                           ProhibitSub -> pure (c1, c2, c3, c4 + 1)
-              CPDelete sId -> withUserRole $ unliftIO u $ do
+              CPDelete qId -> withAdminRole $ unliftIO u $ do
                 st <- asks msgStore
                 r <- liftIO $ runExceptT $ do
-                  q <- ExceptT $ getQueue st SSender sId
+                  (q, _) <- ExceptT $ getSenderQueue st qId
                   ExceptT $ deleteQueueSize st q
                 case r of
                   Left e -> liftIO $ hPutStrLn h $ "error: " <> show e
                   Right (qr, numDeleted) -> do
                     updateDeletedStats qr
                     liftIO $ hPutStrLn h $ "ok, " <> show numDeleted <> " messages deleted"
-              CPStatus sId -> withUserRole $ unliftIO u $ do
+              CPStatus qId -> withUserRole $ unliftIO u $ do
                 st <- asks msgStore
-                q <- liftIO $ getQueueRec st SSender sId
+                q <- liftIO $ getSenderQueue st qId
                 liftIO $ hPutStrLn h $ case q of
                   Left e -> "error: " <> show e
                   Right (_, QueueRec {queueMode, status, updatedAt}) ->
                     "status: " <> show status <> ", updatedAt: " <> show updatedAt <> ", queueMode: " <> show queueMode
-              CPBlock sId info -> withUserRole $ unliftIO u $ do
+              CPBlock qId info -> withUserRole $ unliftIO u $ do
+                st <- asks msgStore
+                stats <- asks serverStats
+                blocked <- liftIO $ readIORef $ qBlocked stats
+                let quota = dailyBlockQueueQuota cfg
+                if blocked >= quota && quota /= 0
+                  then liftIO $ hPutStrLn h $ "error: reached limit of " <> show quota <> " queues blocked daily"
+                  else do
+                    r <- liftIO $ runExceptT $ do
+                      (q, QueueRec {status}) <- ExceptT $ getSenderQueue st qId
+                      when (status == EntityActive) $ ExceptT $ blockQueue (queueStore st) q info
+                      pure status
+                    case r of
+                      Left e -> liftIO $ hPutStrLn h $ "error: " <> show e
+                      Right EntityActive -> do
+                        incStat $ qBlocked stats
+                        liftIO $ hPutStrLn h "ok, queue blocked"
+                      Right status -> liftIO $ hPutStrLn h $ "ok, already inactive: " <> show status
+              CPUnblock qId -> withUserRole $ unliftIO u $ do
                 st <- asks msgStore
                 r <- liftIO $ runExceptT $ do
-                  q <- ExceptT $ getQueue st SSender sId
-                  ExceptT $ blockQueue (queueStore st) q info
-                case r of
-                  Left e -> liftIO $ hPutStrLn h $ "error: " <> show e
-                  Right () -> do
-                    incStat . qBlocked =<< asks serverStats
-                    liftIO $ hPutStrLn h "ok"
-              CPUnblock sId -> withUserRole $ unliftIO u $ do
-                st <- asks msgStore
-                r <- liftIO $ runExceptT $ do
-                  q <- ExceptT $ getQueue st SSender sId
-                  ExceptT $ unblockQueue (queueStore st) q
+                  (q, QueueRec {status}) <- ExceptT $ getSenderQueue st qId
+                  case status of
+                    EntityBlocked info -> Right info <$ ExceptT (unblockQueue (queueStore st) q)
+                    EntityActive -> pure $ Left True
+                    EntityOff -> pure $ Left False
                 liftIO $ hPutStrLn h $ case r of
                   Left e -> "error: " <> show e
-                  Right () -> "ok"
+                  Right (Right info) -> "ok, queue unblocked, reason to block was: " <> show info
+                  Right (Left unblocked) -> if unblocked then "ok, queue was active" else "error, queue is inactive"
               CPSave -> withAdminRole $ withLock' (savingLock srv) "control" $ do
                 hPutStrLn h "saving server state..."
                 unliftIO u $ saveServer False
@@ -1007,6 +1018,11 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
               CPQuit -> pure ()
               CPSkip -> pure ()
               where
+                getSenderQueue st qId =
+                  getQueueRec st SSender qId >>= \case
+                    Right r -> pure $ Right r
+                    Left AUTH -> getQueueRec st SSenderLink qId
+                    Left e ->  pure $ Left e
                 withUserRole action = readTVarIO role >>= \case
                   CPRAdmin -> action
                   CPRUser -> action
@@ -1045,6 +1061,20 @@ runClientTransport h@THandle {params = thParams@THandleParams {sessionId}} = do
       where
         hasSubs ServerSubscribers {subClients} = IS.member clientId <$> readTVarIO subClients
 
+controlPortAuth :: Handle -> Maybe BasicAuth -> Maybe BasicAuth -> TVar CPClientRole -> BasicAuth -> IO ()
+controlPortAuth h user admin role auth = do
+  readTVarIO role >>= \case
+    CPRNone -> do
+      atomically $ writeTVar role $! newRole
+      hPutStrLn h $ currentRole newRole
+    r -> hPutStrLn h $ currentRole r <> if r == newRole then "" else ", start new session to change."
+  where
+    currentRole r = "Current role is " <> show r
+    newRole
+      | Just auth == admin = CPRAdmin
+      | Just auth == user = CPRUser
+      | otherwise = CPRNone
+
 clientDisconnected :: forall s. Client s -> M s ()
 clientDisconnected c@Client {clientId, subscriptions, ntfSubscriptions, serviceSubsCount, ntfServiceSubsCount, connected, clientTHParams = THandleParams {sessionId, thAuth}, endThreads} = do
   labelMyThread . B.unpack $ "client $" <> encode sessionId <> " disc"
@@ -1514,7 +1544,7 @@ client
                             stats <- asks serverStats
                             incStat $ qCreated stats
                             incStat $ qCount stats
-                            when (isJust ntf) $ incStat $ ntfCreated stats
+                            when (isJust ntf) $ incStat $ ntfNewCreated stats
                             case subMode of
                               SMOnlyCreate -> pure ()
                               SMSubscribe -> subscribeNewQueue rcvId qr -- no need to check if message is available, it's a new queue

src/Simplex/Messaging/Server/Control.hs

@@ -8,7 +8,7 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (BasicAuth, BlockingInfo, SenderId)
 
 data CPClientRole = CPRNone | CPRUser | CPRAdmin
-  deriving (Eq)
+  deriving (Eq, Show)
 
 data ControlProtocol
   = CPAuth BasicAuth

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

@@ -149,6 +149,7 @@ data ServerConfig s = ServerConfig
     -- | control port passwords,
     controlPortUserAuth :: Maybe BasicAuth,
     controlPortAdminAuth :: Maybe BasicAuth,
+    dailyBlockQueueQuota :: Int,
     -- | time after which the messages can be removed from the queues and check interval, seconds
     messageExpiration :: Maybe ExpirationConfig,
     expireMessagesOnStart :: Bool,

src/Simplex/Messaging/Server/Main.hs

@@ -420,6 +420,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
               newQueueBasicAuth = either error id <$!> strDecodeIni "AUTH" "create_password" ini,
               controlPortAdminAuth = either error id <$!> strDecodeIni "AUTH" "control_port_admin_password" ini,
               controlPortUserAuth = either error id <$!> strDecodeIni "AUTH" "control_port_user_password" ini,
+              dailyBlockQueueQuota = readIniDefault 20 "AUTH" "daily_block_queue_quota" ini,
               messageExpiration =
                 Just
                   defaultMessageExpiration

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

@@ -92,7 +92,7 @@ iniFileContent cfgPath logPath opts host basicAuth controlPortPwds =
     <> "# Log daily server statistics to CSV file\n"
     <> ("log_stats: " <> onOff logStats <> "\n\n")
     <> "# Log interval for real-time Prometheus metrics\n\
-        \# prometheus_interval: 300\n\n\
+        \# prometheus_interval: 60\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\
@@ -107,8 +107,11 @@ iniFileContent cfgPath logPath opts host basicAuth controlPortPwds =
         )
     <> "\n\n"
     <> (optDisabled controlPortPwds <> "control_port_admin_password: " <> maybe "" fst controlPortPwds <> "\n")
-    <> (optDisabled controlPortPwds <> "control_port_user_password: " <> maybe "" snd controlPortPwds <> "\n")
-    <> "\n\
+    <> (optDisabled controlPortPwds <> "control_port_user_password: " <> maybe "" snd controlPortPwds <> "\n\n")
+    <> "# The limit for queues that can be blocked via control port per day, resets at 0:00 UTC.\n\
+        \# Set to 0 to disable limit, to -1 to prohibit blocking. Default is 20.\n\
+        \# daily_block_queue_quota: 20\n\
+        \\n\
         \[TRANSPORT]\n\
         \# Host is only used to print server address on start.\n\
         \# You can specify multiple server ports.\n"

src/Simplex/Messaging/Server/Prometheus.hs

@@ -82,6 +82,7 @@ prometheusMetrics sm rtm ts =
         _qSubEnd,
         _qSubEndB,
         _ntfCreated,
+        _ntfNewCreated,
         _ntfDeleted,
         _ntfDeletedB,
         _ntfSub,
@@ -262,15 +263,19 @@ prometheusMetrics sm rtm ts =
       \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\
+      \# HELP simplex_smp_queues_notify_created Created queue notification credentials.\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\
+      \# HELP simplex_smp_queues_notify_new_created Created new queues with notification credentials.\n\
+      \# TYPE simplex_smp_queues_notify_new_created counter\n\
+      \simplex_smp_queues_notify_new_created " <> mshow _ntfNewCreated <> "\n# ntfNewCreated\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_deleted Deleted queue notification credentials.\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\
+      \# HELP simplex_smp_queues_notify_deleted_batch Deleted batched queue notification credentials.\n\
       \# TYPE simplex_smp_queues_notify_deleted_batch counter\n\
       \simplex_smp_queues_notify_deleted_batch " <> mshow _ntfDeletedB <> "\n# ntfDeletedB\n\
       \\n\

src/Simplex/Messaging/Server/Stats.hs

@@ -48,6 +48,7 @@ data ServerStats = ServerStats
     qSubEnd :: IORef Int,
     qSubEndB :: IORef Int,
     ntfCreated :: IORef Int,
+    ntfNewCreated :: IORef Int, -- credentials created at the time of queue creation
     ntfDeleted :: IORef Int,
     ntfDeletedB :: IORef Int,
     ntfSub :: IORef Int,
@@ -107,6 +108,7 @@ data ServerStatsData = ServerStatsData
     _qSubEnd :: Int,
     _qSubEndB :: Int,
     _ntfCreated :: Int,
+    _ntfNewCreated :: Int,
     _ntfDeleted :: Int,
     _ntfDeletedB :: Int,
     _ntfSub :: Int,
@@ -167,6 +169,7 @@ newServerStats ts = do
   qSubEnd <- newIORef 0
   qSubEndB <- newIORef 0
   ntfCreated <- newIORef 0
+  ntfNewCreated <- newIORef 0
   ntfDeleted <- newIORef 0
   ntfDeletedB <- newIORef 0
   ntfSub <- newIORef 0
@@ -224,6 +227,7 @@ newServerStats ts = do
         qSubEnd,
         qSubEndB,
         ntfCreated,
+        ntfNewCreated,
         ntfDeleted,
         ntfDeletedB,
         ntfSub,
@@ -283,6 +287,7 @@ getServerStatsData s = do
   _qSubEnd <- readIORef $ qSubEnd s
   _qSubEndB <- readIORef $ qSubEndB s
   _ntfCreated <- readIORef $ ntfCreated s
+  _ntfNewCreated <- readIORef $ ntfNewCreated s
   _ntfDeleted <- readIORef $ ntfDeleted s
   _ntfDeletedB <- readIORef $ ntfDeletedB s
   _ntfSub <- readIORef $ ntfSub s
@@ -340,6 +345,7 @@ getServerStatsData s = do
         _qSubEnd,
         _qSubEndB,
         _ntfCreated,
+        _ntfNewCreated,
         _ntfDeleted,
         _ntfDeletedB,
         _ntfSub,
@@ -400,6 +406,7 @@ setServerStats s d = do
   writeIORef (qSubEnd s) $! _qSubEnd d
   writeIORef (qSubEndB s) $! _qSubEndB d
   writeIORef (ntfCreated s) $! _ntfCreated d
+  writeIORef (ntfNewCreated s) $! _ntfNewCreated d
   writeIORef (ntfDeleted s) $! _ntfDeleted d
   writeIORef (ntfDeletedB s) $! _ntfDeletedB d
   writeIORef (ntfSub s) $! _ntfSub d
@@ -460,6 +467,7 @@ instance StrEncoding ServerStatsData where
         "qSubEnd=" <> strEncode (_qSubEnd d),
         "qSubEndB=" <> strEncode (_qSubEndB d),
         "ntfCreated=" <> strEncode (_ntfCreated d),
+        "ntfNewCreated=" <> strEncode (_ntfNewCreated d),
         "ntfDeleted=" <> strEncode (_ntfDeleted d),
         "ntfDeletedB=" <> strEncode (_ntfDeletedB d),
         "ntfSub=" <> strEncode (_ntfSub d),
@@ -523,6 +531,7 @@ instance StrEncoding ServerStatsData where
     _qSubEnd <- opt "qSubEnd="
     _qSubEndB <- opt "qSubEndB="
     _ntfCreated <- opt "ntfCreated="
+    _ntfNewCreated <- opt "ntfNewCreated="
     _ntfDeleted <- opt "ntfDeleted="
     _ntfDeletedB <- opt "ntfDeletedB="
     _ntfSub <- opt "ntfSub="
@@ -590,6 +599,7 @@ instance StrEncoding ServerStatsData where
           _qSubEnd,
           _qSubEndB,
           _ntfCreated,
+          _ntfNewCreated,
           _ntfDeleted,
           _ntfDeletedB,
           _ntfSub,

src/Simplex/Messaging/Transport/Server.hs

@@ -49,7 +49,7 @@ import Network.Socket
 import qualified Network.TLS as T
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Shared
-import Simplex.Messaging.Util (catchAll_, labelMyThread, tshow)
+import Simplex.Messaging.Util (catchAll_, labelMyThread, tshow, unlessM)
 import System.Exit (exitFailure)
 import System.IO.Error (tryIOError)
 import System.Mem.Weak (Weak, deRefWeak)
@@ -172,12 +172,13 @@ runTCPServerSocket (accepted, gracefullyClosed, clients) started getSocket serve
   E.bracket getSocket (closeServer started clients) $ \sock ->
     forever . E.bracketOnError (safeAccept sock) (close . fst) $ \(conn, _peer) -> do
       cId <- atomically $ stateTVar accepted $ \cId -> let cId' = cId + 1 in cId' `seq` (cId', cId')
+      closed <- newTVarIO False
       let closeConn _ = do
-            atomically $ modifyTVar' clients $ IM.delete cId
+            atomically $ writeTVar closed True >> modifyTVar' clients (IM.delete cId)
             gracefulClose conn 5000 `catchAll_` pure () -- catchAll_ is needed here in case the connection was closed earlier
             atomically $ modifyTVar' gracefullyClosed (+ 1)
       tId <- mkWeakThreadId =<< server conn `forkFinally` closeConn
-      atomically $ modifyTVar' clients $ IM.insert cId tId
+      atomically $ unlessM (readTVar closed) $ modifyTVar' clients $ IM.insert cId tId
 
 -- | Recover from errors in `accept` whenever it is safe.
 -- Some errors are safe to ignore, while blindly restaring `accept` may trigger a busy loop.