smp server: fix server-info, additional stats, allow expiring inactive clients which have prohibit subscriptions only (iOS NSE clients) (#1237)

* smp server: fix server-info * fix * faster saving messages * remove comment * move ProhibitSub out of TVar * subscription stats * stabilize test * order Co-authored-by: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com> * more notification stats * count ntf stats * update server-info --------- Co-authored-by: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com>
2026-08-02 20:10:38 +00:00 · 2024-07-25 16:06:40 +01:00
parent 9093c7b120
commit 5fa3c149e9
7 changed files with 215 additions and 82 deletions
@@ -37,6 +37,7 @@ module Simplex.Messaging.Server
  )
 where

+import Control.Concurrent.STM.TQueue (flushTQueue)
 import Control.Logger.Simple
 import Control.Monad
 import Control.Monad.Except
@@ -47,6 +48,7 @@ import Crypto.Random
 import Control.Monad.STM (retry)
 import Data.Bifunctor (first)
 import Data.ByteString.Base64 (encode)
+import qualified Data.ByteString.Builder as BLD
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy.Char8 as LB
@@ -54,6 +56,7 @@ import Data.Either (fromRight, partitionEithers)
 import Data.Functor (($>))
 import Data.Int (Int64)
 import qualified Data.IntMap.Strict as IM
+import qualified Data.IntSet as IS
 import Data.List (intercalate, mapAccumR)
 import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.List.NonEmpty as L
@@ -232,7 +235,9 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
      initialDelay <- (startAt -) . fromIntegral . (`div` 1000000_000000) . diffTimeToPicoseconds . utctDayTime <$> liftIO getCurrentTime
      liftIO $ putStrLn $ "server stats log enabled: " <> statsFilePath
      liftIO $ threadDelay' $ 1000000 * (initialDelay + if initialDelay < 0 then 86400 else 0)
-      ss@ServerStats {fromTime, qCreated, qSecured, qDeletedAll, qDeletedNew, qDeletedSecured, qSub, qSubNoMsg, qSubAuth, qSubDuplicate, qSubProhibited, msgSent, msgSentAuth, msgSentQuota, msgSentLarge, msgRecv, msgRecvGet, msgGet, msgGetNoMsg, msgGetAuth, msgGetDuplicate, msgGetProhibited, msgExpired, activeQueues, msgSentNtf, msgRecvNtf, activeQueuesNtf, qCount, msgCount, pRelays, pRelaysOwn, pMsgFwds, pMsgFwdsOwn, pMsgFwdsRecv} <- asks serverStats
+      ss@ServerStats {fromTime, qCreated, qSecured, qDeletedAll, qDeletedNew, qDeletedSecured, qSub, qSubNoMsg, qSubAuth, qSubDuplicate, qSubProhibited, ntfCreated, ntfDeleted, ntfSub, ntfSubAuth, ntfSubDuplicate, msgSent, msgSentAuth, msgSentQuota, msgSentLarge, msgRecv, msgRecvGet, msgGet, msgGetNoMsg, msgGetAuth, msgGetDuplicate, msgGetProhibited, msgExpired, activeQueues, subscribedQueues, msgSentNtf, msgRecvNtf, activeQueuesNtf, qCount, msgCount, pRelays, pRelaysOwn, pMsgFwds, pMsgFwdsOwn, pMsgFwdsRecv}
+        <- asks serverStats
+      QueueStore {queues, notifiers} <- asks queueStore
      let interval = 1000000 * logInterval
      forever $ do
        withFile statsFilePath AppendMode $ \h -> liftIO $ do
@@ -249,6 +254,11 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
          qSubAuth' <- atomically $ swapTVar qSubAuth 0
          qSubDuplicate' <- atomically $ swapTVar qSubDuplicate 0
          qSubProhibited' <- atomically $ swapTVar qSubProhibited 0
+          ntfCreated' <- atomically $ swapTVar ntfCreated 0
+          ntfDeleted' <- atomically $ swapTVar ntfDeleted 0
+          ntfSub' <- atomically $ swapTVar ntfSub 0
+          ntfSubAuth' <- atomically $ swapTVar ntfSubAuth 0
+          ntfSubDuplicate' <- atomically $ swapTVar ntfSubDuplicate 0
          msgSent' <- atomically $ swapTVar msgSent 0
          msgSentAuth' <- atomically $ swapTVar msgSentAuth 0
          msgSentQuota' <- atomically $ swapTVar msgSentQuota 0
@@ -262,6 +272,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
          msgGetProhibited' <- atomically $ swapTVar msgGetProhibited 0
          msgExpired' <- atomically $ swapTVar msgExpired 0
          ps <- atomically $ periodStatCounts activeQueues ts
+          psSub <- atomically $ periodStatCounts subscribedQueues ts
          msgSentNtf' <- atomically $ swapTVar msgSentNtf 0
          msgRecvNtf' <- atomically $ swapTVar msgRecvNtf 0
          psNtf <- atomically $ periodStatCounts activeQueuesNtf ts
@@ -274,6 +285,8 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
          pMsgFwdsOwn' <- atomically $ getResetProxyStatsData pMsgFwdsOwn
          pMsgFwdsRecv' <- atomically $ swapTVar pMsgFwdsRecv 0
          qCount' <- readTVarIO qCount
+          qCount'' <- M.size <$> readTVarIO queues
+          ntfCount' <- M.size <$> readTVarIO notifiers
          msgCount' <- readTVarIO msgCount
          hPutStrLn h $
            intercalate
@@ -319,7 +332,17 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
                       show msgGetNoMsg',
                       show msgGetAuth',
                       show msgGetDuplicate',
-                       show msgGetProhibited'
+                       show msgGetProhibited',
+                       dayCount psSub,
+                       weekCount psSub,
+                       monthCount psSub,
+                       show qCount'',
+                       show ntfCreated',
+                       show ntfDeleted',
+                       show ntfSub',
+                       show ntfSubAuth',
+                       show ntfSubDuplicate',
+                       show ntfCount'
                     ]
              )
        liftIO $ threadDelay' interval
@@ -410,14 +433,19 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
                putStat "qDeletedAll" qDeletedAll
                putStat "qDeletedNew" qDeletedNew
                putStat "qDeletedSecured" qDeletedSecured
-                getStat (day . activeQueues) >>= \v -> hPutStrLn h $ "dayMsgQueues: " <> show (S.size v)
-                subs <- (,,,,) <$> getStat qSub <*> getStat qSubNoMsg <*> getStat qSubAuth <*> getStat qSubDuplicate <*> getStat qSubProhibited
-                hPutStrLn h $ "SUBs (count, noMsg, auth, duplicate, prohibited): " <> show subs
+                getStat (day . activeQueues) >>= \v -> hPutStrLn h $ "daily active queues: " <> show (S.size v)
+                getStat (day . subscribedQueues) >>= \v -> hPutStrLn h $ "daily subscribed queues: " <> show (S.size v)
+                putStat "qSub" qSub
+                putStat "qSubNoMsg" qSubNoMsg
+                subs <- (,,) <$> getStat qSubAuth <*> getStat qSubDuplicate <*> getStat qSubProhibited
+                hPutStrLn h $ "other SUB events (auth, duplicate, prohibited): " <> show subs
                putStat "msgSent" msgSent
                putStat "msgRecv" msgRecv
                putStat "msgRecvGet" msgRecvGet
-                gets <- (,,,,) <$> getStat msgGet <*> getStat msgGetNoMsg <*> getStat msgGetAuth <*> getStat msgGetDuplicate <*> getStat msgGetProhibited
-                hPutStrLn h $ "GETs (count, noMsg, auth, duplicate, prohibited): " <> show gets
+                putStat "msgGet" msgGet
+                putStat "msgGetNoMsg" msgGet
+                gets <- (,,) <$> getStat msgGetAuth <*> getStat msgGetDuplicate <*> getStat msgGetProhibited
+                hPutStrLn h $ "other GET events (auth, duplicate, prohibited): " <> show gets
                putStat "msgSentNtf" msgSentNtf
                putStat "msgRecvNtf" msgRecvNtf
                putStat "qCount" qCount
@@ -494,10 +522,23 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
                    putActiveClientsInfo protoName clients = do
                      activeSubs <- readTVarIO clients
                      hPutStrLn h $ protoName <> " subscriptions: " <> show (M.size activeSubs)
-                      hPutStrLn h $ protoName <> " subscribed clients: " <> show (countSubClients activeSubs)
-                      when (r == CPRAdmin) $ do
-                        clQs <- clientTBQueueLengths activeSubs
-                        hPutStrLn h $ protoName <> " subscribed clients queues (rcvQ, sndQ, msgQ): " <> show clQs
+                      clCnt <- if r == CPRAdmin then putClientQueues activeSubs else pure $ countSubClients activeSubs
+                      hPutStrLn h $ protoName <> " subscribed clients: " <> show clCnt
+                      where
+                        putClientQueues :: M.Map QueueId Client -> IO Int
+                        putClientQueues subs = do
+                          let cls = differentClients subs
+                          clQs <- clientTBQueueLengths cls
+                          hPutStrLn h $ protoName <> " subscribed clients queues (rcvQ, sndQ, msgQ): " <> show clQs
+                          pure $ length cls
+                        differentClients :: M.Map QueueId Client -> [Client]
+                        differentClients = fst . M.foldl' addClient ([], IS.empty)
+                          where
+                            addClient acc@(cls, clSet) cl@Client {clientId}
+                              | IS.member clientId clSet = acc
+                              | otherwise = (cl : cls, IS.insert clientId clSet)
+                        countSubClients :: M.Map QueueId Client -> Int
+                        countSubClients = IS.size . M.foldr' (IS.insert . clientId) IS.empty
                    countClientSubs :: (Client -> TMap QueueId a) -> Maybe (M.Map QueueId a -> IO (Int, Int, Int, Int)) -> IM.IntMap Client -> IO (Int, (Int, Int, Int, Int), Int, (Natural, Natural, Natural))
                    countClientSubs subSel countSubs_ = foldM addSubs (0, (0, 0, 0, 0), 0, (0, 0, 0))
                      where
@@ -526,13 +567,14 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
                    countSMPSubs :: M.Map QueueId Sub -> IO (Int, Int, Int, Int)
                    countSMPSubs = foldM countSubs (0, 0, 0, 0)
                      where
-                        countSubs (c1, c2, c3, c4) Sub {subThread} =
-                          readTVarIO subThread >>= \st -> pure $ case st of
-                            NoSub -> (c1 + 1, c2, c3, c4)
-                            SubPending -> (c1, c2 + 1, c3, c4)
-                            SubThread _ -> (c1, c2, c3 + 1, c4)
-                            ProhibitSub -> (c1, c2, c3, c4 + 1)
-                    countSubClients = S.size . M.foldr' (S.insert . clientId) S.empty
+                        countSubs (c1, c2, c3, c4) Sub {subThread} = case subThread of
+                          ServerSub t -> do
+                            st <- readTVarIO t
+                            pure $ case st of
+                              NoSub -> (c1 + 1, c2, c3, c4)
+                              SubPending -> (c1, c2 + 1, c3, c4)
+                              SubThread _ -> (c1, c2, c3 + 1, c4)
+                          ProhibitSub -> pure (c1, c2, c3, c4 + 1)
              CPDelete queueId' -> withUserRole $ unliftIO u $ do
                st <- asks queueStore
                ms <- asks msgStore
@@ -587,7 +629,8 @@ runClientTransport h@THandle {params = thParams@THandleParams {thVersion, sessio
  where
    disconnectThread_ c (Just expCfg) = [liftIO $ disconnectTransport h (rcvActiveAt c) (sndActiveAt c) expCfg (noSubscriptions c)]
    disconnectThread_ _ _ = []
-    noSubscriptions c = atomically $ (&&) <$> TM.null (subscriptions c) <*> TM.null (ntfSubscriptions c)
+    noSubscriptions c = atomically $ (&&) <$> TM.null (ntfSubscriptions c) <*> (not . hasSubs <$> readTVar (subscriptions c))
+    hasSubs = any $ (\case ServerSub _ -> True; ProhibitSub -> False) . subThread

 clientDisconnected :: Client -> M ()
 clientDisconnected c@Client {clientId, subscriptions, ntfSubscriptions, connected, sessionId, endThreads} = do
@@ -615,10 +658,12 @@ sameClientId :: Client -> Client -> Bool
 sameClientId Client {clientId} Client {clientId = cId'} = clientId == cId'

 cancelSub :: Sub -> IO ()
-cancelSub s =
-  readTVarIO (subThread s) >>= \case
-    SubThread t -> liftIO $ deRefWeak t >>= mapM_ killThread
-    _ -> pure ()
+cancelSub s = case subThread s of
+  ServerSub st ->
+    readTVarIO st >>= \case
+      SubThread t -> liftIO $ deRefWeak t >>= mapM_ killThread
+      _ -> pure ()
+  ProhibitSub -> pure ()

 receive :: Transport c => THandleSMP c 'TServer -> Client -> M ()
 receive h@THandle {params = THandleParams {thAuth}} Client {rcvQ, sndQ, rcvActiveAt, sessionId} = do
@@ -643,6 +688,7 @@ receive h@THandle {params = THandleParams {thAuth}} Client {rcvQ, sndQ, rcvActiv
                case cmd of
                  Cmd _ SEND {} -> incStat $ msgSentAuth stats
                  Cmd _ SUB -> incStat $ qSubAuth stats
+                  Cmd _ NSUB -> incStat $ ntfSubAuth stats
                  Cmd _ GET -> incStat $ msgGetAuth stats
                  _ -> pure ()
                pure $ Left (corrId, entId, ERR AUTH)
@@ -983,6 +1029,7 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
                Left e -> pure $ ERR e
                Right _ -> do
                  withLog $ \s -> logAddNotifier s entId ntfCreds
+                  incStat . ntfCreated =<< asks serverStats
                  pure $ NID notifierId rcvPublicDhKey

        deleteQueueNotifier_ :: QueueStore -> M (Transmission BrokerMsg)
@@ -992,6 +1039,7 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
            Right () -> do
              -- Possibly, the same should be done if the queue is suspended, but currently we do not use it
              atomically $ writeTQueue ntfSubscribedQ (entId, clnt, False)
+              incStat . ntfDeleted =<< asks serverStats
              pure ok
            Left e -> pure $ err e

@@ -1006,7 +1054,7 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
            Nothing -> newSub >>= deliver True
            Just s@Sub {subThread} -> do
              stats <- asks serverStats
-              readTVarIO subThread >>= \case
+              case subThread of
                ProhibitSub -> do
                  -- cannot use SUB in the same connection where GET was used
                  incStat $ qSubProhibited stats
@@ -1028,6 +1076,7 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
              when inc $ do
                stats <- asks serverStats
                incStat $ (if isJust msg_ then qSub else qSubNoMsg) stats
+                atomically $ updatePeriodStats (subscribedQueues stats) rId
              deliverMessage "SUB" qr rId sub msg_

        getMessage :: QueueRec -> M (Transmission BrokerMsg)
@@ -1036,7 +1085,7 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
            Nothing ->
              atomically newSub >>= (`getMessage_` Nothing)
            Just s@Sub {subThread} ->
-              readTVarIO subThread >>= \case
+              case subThread of
                ProhibitSub ->
                  atomically (tryTakeTMVar $ delivered s)
                    >>= getMessage_ s
@@ -1048,7 +1097,7 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
          where
            newSub :: STM Sub
            newSub = do
-              s <- newSubscription ProhibitSub
+              s <- newProhibitedSub
              TM.insert entId s subscriptions
              pure s
            getMessage_ :: Sub -> Maybe MsgId -> M (Transmission BrokerMsg)
@@ -1070,11 +1119,19 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
        withQueue action = maybe (pure $ err AUTH) action qr_

        subscribeNotifications :: M (Transmission BrokerMsg)
-        subscribeNotifications = time "NSUB" . atomically $ do
-          unlessM (TM.member entId ntfSubscriptions) $ do
-            writeTQueue ntfSubscribedQ (entId, clnt, True)
-            TM.insert entId () ntfSubscriptions
+        subscribeNotifications = do
+          statCount <-
+            time "NSUB" . atomically $ do
+              ifM
+                (TM.member entId ntfSubscriptions)
+                (pure ntfSubDuplicate)
+                (newSub $> ntfSub)
+          incStat . statCount =<< asks serverStats
          pure ok
+          where
+            newSub = do
+              writeTQueue ntfSubscribedQ (entId, clnt, True)
+              TM.insert entId () ntfSubscriptions

        acknowledgeMsg :: QueueRec -> MsgId -> M (Transmission BrokerMsg)
        acknowledgeMsg qr msgId = time "ACK" $ do
@@ -1095,11 +1152,11 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
                      deliverMessage "ACK" qr entId sub msg_
                _ -> pure $ err NO_MSG
          where
-            getDelivered :: Sub -> STM (Maybe SubscriptionThread)
+            getDelivered :: Sub -> STM (Maybe ServerSub)
            getDelivered Sub {delivered, subThread} = do
              tryTakeTMVar delivered $>>= \msgId' ->
                if msgId == msgId' || B.null msgId
-                  then Just <$> readTVar subThread
+                  then pure $ Just subThread
                  else putTMVar delivered msgId' $> Nothing
            updateStats :: Bool -> Message -> M ()
            updateStats isGet = \case
@@ -1188,26 +1245,28 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
                deliverToSub =
                  TM.lookup rId subscribers
                    $>>= \rc@Client {subscriptions = subs, sndQ = q} -> TM.lookup rId subs
-                    $>>= \s@Sub {subThread, delivered} -> readTVar subThread >>= \case
-                      NoSub ->
-                        tryTakeTMVar delivered >>= \case
-                          Just _ -> pure Nothing -- if a message was already delivered, should not deliver more
-                          Nothing ->
-                            ifM
-                              (isFullTBQueue q)
-                              (writeTVar subThread SubPending $> Just (rc, s))
-                              (deliver q s $> Nothing)
-                      _ -> pure Nothing
+                    $>>= \s@Sub {subThread, delivered} -> case subThread of
+                      ProhibitSub -> pure Nothing
+                      ServerSub st -> readTVar st >>= \case
+                        NoSub ->
+                          tryTakeTMVar delivered >>= \case
+                            Just _ -> pure Nothing -- if a message was already delivered, should not deliver more
+                            Nothing ->
+                              ifM
+                                (isFullTBQueue q)
+                                (writeTVar st SubPending $> Just (rc, s, st))
+                                (deliver q s $> Nothing)
+                        _ -> pure Nothing
                deliver q s = do
                  let encMsg = encryptMsg qr msg
                  writeTBQueue q [(CorrId "", rId, MSG encMsg)]
                  void $ setDelivered s msg
-                forkDeliver (rc@Client {sndQ = q}, s@Sub {subThread, delivered}) = do
+                forkDeliver (rc@Client {sndQ = q}, s@Sub {delivered}, st) = do
                  t <- mkWeakThreadId =<< forkIO deliverThread
-                  atomically . modifyTVar' subThread $ \case
+                  atomically . modifyTVar' st $ \case
                    -- this case is needed because deliverThread can exit before it
                    SubPending -> SubThread t
-                    st -> st
+                    st' -> st'
                  where
                    deliverThread = do
                      labelMyThread $ B.unpack ("client $" <> encode sessionId) <> " deliver/SEND"
@@ -1217,7 +1276,7 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
                            Just _ -> pure () -- if a message was already delivered, should not deliver more
                            Nothing -> do
                              deliver q s
-                              writeTVar subThread NoSub
+                              writeTVar st NoSub

            trySendNotification :: NtfCreds -> Message -> TVar ChaChaDRG -> STM (Maybe Bool)
            trySendNotification NtfCreds {notifierId, rcvNtfDhSecret} msg ntfNonceDrg =
@@ -1295,7 +1354,7 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
                      VRFailed -> Left (corrId', entId', ERR AUTH)
        deliverMessage :: T.Text -> QueueRec -> RecipientId -> Sub -> Maybe Message -> M (Transmission BrokerMsg)
        deliverMessage name qr rId s@Sub {subThread} msg_ = time (name <> " deliver") . atomically $
-          readTVar subThread >>= \case
+          case subThread of
            ProhibitSub -> pure resp
            _ -> case msg_ of
              Just msg ->
@@ -1351,12 +1410,14 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
          pure (corrId, entId, INFO info)
          where
            mkQSub Sub {subThread, delivered} = do
-              st <- readTVar subThread
-              let qSubThread = case st of
+              qSubThread <- case subThread of
+                ServerSub t -> do
+                  st <- readTVar t
+                  pure $ case st of
                    NoSub -> QNoSub
                    SubPending -> QSubPending
                    SubThread _ -> QSubThread
-                    ProhibitSub -> QProhibitSub
+                ProhibitSub -> pure QProhibitSub
              qDelivered <- decodeLatin1 . encode <$$> tryReadTMVar delivered
              pure QSub {qSubThread, qDelivered}

@@ -1408,13 +1469,16 @@ saveServerMessages keepMsgs = asks (storeMsgsFile . config) >>= mapM_ saveMessag
      logInfo $ "saving messages to file " <> T.pack f
      ms <- asks msgStore
      liftIO . withFile f WriteMode $ \h ->
-        readTVarIO ms >>= mapM_ (saveQueueMsgs ms h) . M.keys
+        readTVarIO ms >>= mapM_ (saveQueueMsgs h) . M.assocs
      logInfo "messages saved"
      where
-        getMessages = if keepMsgs then snapshotMsgQueue else flushMsgQueue
-        saveQueueMsgs ms h rId =
-          atomically (getMessages ms rId)
-            >>= mapM_ (B.hPutStrLn h . strEncode . MLRv3 rId)
+        saveQueueMsgs h (rId, q) = BLD.hPutBuilder h . encodeMessages rId =<< atomically (getMessages $ msgQueue q)
+        getMessages = if keepMsgs then snapshotTQueue else flushTQueue
+        snapshotTQueue q = do
+          msgs <- flushTQueue q
+          mapM_ (writeTQueue q) msgs
+          pure msgs
+        encodeMessages rId = mconcat . map (\msg -> BLD.byteString (strEncode $ MLRv3 rId msg) <> BLD.char8 '\n')

 restoreServerMessages :: M Int
 restoreServerMessages =
@@ -1,12 +1,15 @@
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE StrictData #-}

 module Simplex.Messaging.Server.Env.STM where

 import Control.Concurrent (ThreadId)
+import Control.Logger.Simple
+import Control.Monad
 import Control.Monad.IO.Unlift
 import Crypto.Random
 import Data.ByteString.Char8 (ByteString)
@@ -17,6 +20,7 @@ import Data.List.NonEmpty (NonEmpty)
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
 import Data.Maybe (isJust, isNothing)
+import qualified Data.Text as T
 import Data.Time.Clock (getCurrentTime)
 import Data.Time.Clock.System (SystemTime)
 import Data.X509.Validation (Fingerprint (..))
@@ -104,7 +108,7 @@ defaultMessageExpiration =
 defaultInactiveClientExpiration :: ExpirationConfig
 defaultInactiveClientExpiration =
  ExpirationConfig
-    { ttl = 43200, -- seconds, 12 hours
+    { ttl = 21600, -- seconds, 6 hours
      checkInterval = 3600 -- seconds, 1 hours
    }

@@ -162,10 +166,12 @@ data Client = Client
    sndActiveAt :: TVar SystemTime
  }

-data SubscriptionThread = NoSub | SubPending | SubThread (Weak ThreadId) | ProhibitSub
+data ServerSub = ServerSub (TVar SubscriptionThread) | ProhibitSub
+
+data SubscriptionThread = NoSub | SubPending | SubThread (Weak ThreadId)

 data Sub = Sub
-  { subThread :: TVar SubscriptionThread,
+  { subThread :: ServerSub, -- Nothing value indicates that sub
    delivered :: TMVar MsgId
  }

@@ -197,16 +203,24 @@ newClient nextClientId qSize thVersion sessionId createdAt = do
 newSubscription :: SubscriptionThread -> STM Sub
 newSubscription st = do
  delivered <- newEmptyTMVar
-  subThread <- newTVar st
+  subThread <- ServerSub <$> newTVar st
  return Sub {subThread, delivered}

+newProhibitedSub :: STM Sub
+newProhibitedSub = do
+  delivered <- newEmptyTMVar
+  return Sub {subThread = ProhibitSub, delivered}
+
 newEnv :: ServerConfig -> IO Env
 newEnv config@ServerConfig {caCertificateFile, certificateFile, privateKeyFile, storeLogFile, smpAgentCfg, transportConfig, information, messageExpiration} = do
  server <- atomically newServer
  queueStore <- atomically newQueueStore
  msgStore <- atomically newMsgStore
  random <- liftIO C.newRandom
-  storeLog <- restoreQueues queueStore `mapM` storeLogFile
+  storeLog <-
+    forM storeLogFile $ \f -> do
+      logInfo $ "restoring queues from file " <> T.pack f
+      restoreQueues queueStore f
  tlsServerParams <- loadTLSServerParams caCertificateFile certificateFile privateKeyFile (alpn transportConfig)
  Fingerprint fp <- loadFingerprint caCertificateFile
  let serverIdentity = KeyHash fp
@@ -14,8 +14,6 @@ module Simplex.Messaging.Server.MsgStore.STM
    getMsgQueue,
    delMsgQueue,
    delMsgQueueSize,
-    flushMsgQueue,
-    snapshotMsgQueue,
    writeMsg,
    tryPeekMsg,
    peekMsg,
@@ -25,7 +23,6 @@ module Simplex.Messaging.Server.MsgStore.STM
  )
 where

-import Control.Concurrent.STM.TQueue (flushTQueue)
 import qualified Data.ByteString.Char8 as B
 import Data.Functor (($>))
 import Data.Int (Int64)
@@ -64,17 +61,6 @@ delMsgQueue st rId = TM.delete rId st
 delMsgQueueSize :: STMMsgStore -> RecipientId -> STM Int
 delMsgQueueSize st rId = TM.lookupDelete rId st >>= maybe (pure 0) (\MsgQueue {size} -> readTVar size)

-flushMsgQueue :: STMMsgStore -> RecipientId -> STM [Message]
-flushMsgQueue st rId = TM.lookupDelete rId st >>= maybe (pure []) (flushTQueue . msgQueue)
-
-snapshotMsgQueue :: STMMsgStore -> RecipientId -> STM [Message]
-snapshotMsgQueue st rId = TM.lookup rId st >>= maybe (pure []) (snapshotTQueue . msgQueue)
-  where
-    snapshotTQueue q = do
-      msgs <- flushTQueue q
-      mapM_ (writeTQueue q) msgs
-      pure msgs
-
 writeMsg :: MsgQueue -> Message -> STM (Maybe (Message, Bool))
 writeMsg MsgQueue {msgQueue = q, quota, canWrite, size} !msg = do
  canWrt <- readTVar canWrite
@@ -31,6 +31,11 @@ data ServerStats = ServerStats
    qSubAuth :: TVar Int,
    qSubDuplicate :: TVar Int,
    qSubProhibited :: TVar Int,
+    ntfCreated :: TVar Int,
+    ntfDeleted :: TVar Int,
+    ntfSub :: TVar Int,
+    ntfSubAuth :: TVar Int,
+    ntfSubDuplicate :: TVar Int,
    msgSent :: TVar Int,
    msgSentAuth :: TVar Int,
    msgSentQuota :: TVar Int,
@@ -44,6 +49,7 @@ data ServerStats = ServerStats
    msgGetProhibited :: TVar Int,
    msgExpired :: TVar Int,
    activeQueues :: PeriodStats RecipientId,
+    subscribedQueues :: PeriodStats RecipientId,
    msgSentNtf :: TVar Int, -- sent messages with NTF flag
    msgRecvNtf :: TVar Int, -- received messages with NTF flag
    activeQueuesNtf :: PeriodStats RecipientId,
@@ -71,6 +77,11 @@ data ServerStatsData = ServerStatsData
    _qSubAuth :: Int,
    _qSubDuplicate :: Int,
    _qSubProhibited :: Int,
+    _ntfCreated :: Int,
+    _ntfDeleted :: Int,
+    _ntfSub :: Int,
+    _ntfSubAuth :: Int,
+    _ntfSubDuplicate :: Int,
    _msgSent :: Int,
    _msgSentAuth :: Int,
    _msgSentQuota :: Int,
@@ -84,6 +95,7 @@ data ServerStatsData = ServerStatsData
    _msgGetProhibited :: Int,
    _msgExpired :: Int,
    _activeQueues :: PeriodStatsData RecipientId,
+    _subscribedQueues :: PeriodStatsData RecipientId,
    _msgSentNtf :: Int,
    _msgRecvNtf :: Int,
    _activeQueuesNtf :: PeriodStatsData RecipientId,
@@ -113,6 +125,11 @@ newServerStats ts = do
  qSubAuth <- newTVar 0
  qSubDuplicate <- newTVar 0
  qSubProhibited <- newTVar 0
+  ntfCreated <- newTVar 0
+  ntfDeleted <- newTVar 0
+  ntfSub <- newTVar 0
+  ntfSubAuth <- newTVar 0
+  ntfSubDuplicate <- newTVar 0
  msgSent <- newTVar 0
  msgSentAuth <- newTVar 0
  msgSentQuota <- newTVar 0
@@ -126,6 +143,7 @@ newServerStats ts = do
  msgGetProhibited <- newTVar 0
  msgExpired <- newTVar 0
  activeQueues <- newPeriodStats
+  subscribedQueues <- newPeriodStats
  msgSentNtf <- newTVar 0
  msgRecvNtf <- newTVar 0
  activeQueuesNtf <- newPeriodStats
@@ -152,6 +170,11 @@ newServerStats ts = do
        qSubAuth,
        qSubDuplicate,
        qSubProhibited,
+        ntfCreated,
+        ntfDeleted,
+        ntfSub,
+        ntfSubAuth,
+        ntfSubDuplicate,
        msgSent,
        msgSentAuth,
        msgSentQuota,
@@ -165,6 +188,7 @@ newServerStats ts = do
        msgGetProhibited,
        msgExpired,
        activeQueues,
+        subscribedQueues,
        msgSentNtf,
        msgRecvNtf,
        activeQueuesNtf,
@@ -193,6 +217,11 @@ getServerStatsData s = do
  _qSubAuth <- readTVar $ qSubAuth s
  _qSubDuplicate <- readTVar $ qSubDuplicate s
  _qSubProhibited <- readTVar $ qSubProhibited s
+  _ntfCreated <- readTVar $ ntfCreated s
+  _ntfDeleted <- readTVar $ ntfDeleted s
+  _ntfSub <- readTVar $ ntfSub s
+  _ntfSubAuth <- readTVar $ ntfSubAuth s
+  _ntfSubDuplicate <- readTVar $ ntfSubDuplicate s
  _msgSent <- readTVar $ msgSent s
  _msgSentAuth <- readTVar $ msgSentAuth s
  _msgSentQuota <- readTVar $ msgSentQuota s
@@ -206,6 +235,7 @@ getServerStatsData s = do
  _msgGetProhibited <- readTVar $ msgGetProhibited s
  _msgExpired <- readTVar $ msgExpired s
  _activeQueues <- getPeriodStatsData $ activeQueues s
+  _subscribedQueues <- getPeriodStatsData $ subscribedQueues s
  _msgSentNtf <- readTVar $ msgSentNtf s
  _msgRecvNtf <- readTVar $ msgRecvNtf s
  _activeQueuesNtf <- getPeriodStatsData $ activeQueuesNtf s
@@ -232,6 +262,11 @@ getServerStatsData s = do
        _qSubAuth,
        _qSubDuplicate,
        _qSubProhibited,
+        _ntfCreated,
+        _ntfDeleted,
+        _ntfSub,
+        _ntfSubAuth,
+        _ntfSubDuplicate,
        _msgSent,
        _msgSentAuth,
        _msgSentQuota,
@@ -245,6 +280,7 @@ getServerStatsData s = do
        _msgGetProhibited,
        _msgExpired,
        _activeQueues,
+        _subscribedQueues,
        _msgSentNtf,
        _msgRecvNtf,
        _activeQueuesNtf,
@@ -273,6 +309,11 @@ setServerStats s d = do
  writeTVar (qSubAuth s) $! _qSubAuth d
  writeTVar (qSubDuplicate s) $! _qSubDuplicate d
  writeTVar (qSubProhibited s) $! _qSubProhibited d
+  writeTVar (ntfCreated s) $! _ntfCreated d
+  writeTVar (ntfDeleted s) $! _ntfDeleted d
+  writeTVar (ntfSub s) $! _ntfSub d
+  writeTVar (ntfSubAuth s) $! _ntfSubAuth d
+  writeTVar (ntfSubDuplicate s) $! _ntfSubDuplicate d
  writeTVar (msgSent s) $! _msgSent d
  writeTVar (msgSentAuth s) $! _msgSentAuth d
  writeTVar (msgSentQuota s) $! _msgSentQuota d
@@ -286,6 +327,7 @@ setServerStats s d = do
  writeTVar (msgGetProhibited s) $! _msgGetProhibited d
  writeTVar (msgExpired s) $! _msgExpired d
  setPeriodStats (activeQueues s) (_activeQueues d)
+  setPeriodStats (subscribedQueues s) (_subscribedQueues d)
  writeTVar (msgSentNtf s) $! _msgSentNtf d
  writeTVar (msgRecvNtf s) $! _msgRecvNtf d
  setPeriodStats (activeQueuesNtf s) (_activeQueuesNtf d)
@@ -315,6 +357,11 @@ instance StrEncoding ServerStatsData where
        "qSubAuth=" <> strEncode (_qSubAuth d),
        "qSubDuplicate=" <> strEncode (_qSubDuplicate d),
        "qSubProhibited=" <> strEncode (_qSubProhibited d),
+        "ntfCreated=" <> strEncode (_ntfCreated d),
+        "ntfDeleted=" <> strEncode (_ntfDeleted d),
+        "ntfSub=" <> strEncode (_ntfSub d),
+        "ntfSubAuth=" <> strEncode (_ntfSubAuth d),
+        "ntfSubDuplicate=" <> strEncode (_ntfSubDuplicate d),
        "msgSent=" <> strEncode (_msgSent d),
        "msgSentAuth=" <> strEncode (_msgSentAuth d),
        "msgSentQuota=" <> strEncode (_msgSentQuota d),
@@ -334,6 +381,8 @@ instance StrEncoding ServerStatsData where
        "msgNtfLost=" <> strEncode (_msgNtfLost d),
        "activeQueues:",
        strEncode (_activeQueues d),
+        "subscribedQueues:",
+        strEncode (_subscribedQueues d),
        "activeQueuesNtf:",
        strEncode (_activeQueuesNtf d),
        "pRelays:",
@@ -359,6 +408,11 @@ instance StrEncoding ServerStatsData where
    _qSubAuth <- opt "qSubAuth="
    _qSubDuplicate <- opt "qSubDuplicate="
    _qSubProhibited <- opt "qSubProhibited="
+    _ntfCreated <- opt "ntfCreated="
+    _ntfDeleted <- opt "ntfDeleted="
+    _ntfSub <- opt "ntfSub="
+    _ntfSubAuth <- opt "ntfSubAuth="
+    _ntfSubDuplicate <- opt "ntfSubDuplicate="
    _msgSent <- "msgSent=" *> strP <* A.endOfLine
    _msgSentAuth <- opt "msgSentAuth="
    _msgSentQuota <- opt "msgSentQuota="
@@ -384,6 +438,10 @@ instance StrEncoding ServerStatsData where
          _week <- "weekMsgQueues=" *> strP <* A.endOfLine
          _month <- "monthMsgQueues=" *> strP <* optional A.endOfLine
          pure PeriodStatsData {_day, _week, _month}
+    _subscribedQueues <-
+      optional ("subscribedQueues:" <* A.endOfLine) >>= \case
+        Just _ -> strP <* optional A.endOfLine
+        _ -> pure newPeriodStatsData
    _activeQueuesNtf <-
      optional ("activeQueuesNtf:" <* A.endOfLine) >>= \case
        Just _ -> strP <* optional A.endOfLine
@@ -406,6 +464,11 @@ instance StrEncoding ServerStatsData where
          _qSubAuth,
          _qSubDuplicate,
          _qSubProhibited,
+          _ntfCreated,
+          _ntfDeleted,
+          _ntfSub,
+          _ntfSubAuth,
+          _ntfSubDuplicate,
          _msgSent,
          _msgSentAuth,
          _msgSentQuota,
@@ -424,6 +487,7 @@ instance StrEncoding ServerStatsData where
          _msgNtfNoSub,
          _msgNtfLost,
          _activeQueues,
+          _subscribedQueues,
          _activeQueuesNtf,
          _pRelays,
          _pRelaysOwn,
@@ -38,6 +38,7 @@ module AgentTests.FunctionalAPITests
    rfGet,
    sfGet,
    nGet,
+    getInAnyOrder,
    (##>),
    (=##>),
    pattern CON,
@@ -361,11 +361,15 @@ agentViaProxyRetryOffline = do
          -- proxy relay down
          4 <- msgId <$> A.sendMessage bob aliceId pqEnc noMsgFlags msg2
          bob `down` aliceId
-        withServer2 $ \_ -> runRight_ $ do
-          bob `up` aliceId
-          get bob ##> ("", aliceId, A.SENT (baseId + 4) bProxySrv)
-          get alice =##> \case ("", c, Msg' _ pq msg2') -> c == bobId && pq == pqEnc && msg2 == msg2'; _ -> False
-          ackMessage alice bobId (baseId + 4) Nothing
+        withServer2 $ \_ -> do
+          getInAnyOrder
+            bob
+            [ \case ("", "", AEvt SAENone (UP _ [c])) -> c == aliceId; _ -> False,
+              \case ("", c, AEvt SAEConn (A.SENT mId srv)) -> c == aliceId && mId == baseId + 4 && srv == bProxySrv; _ -> False
+            ]
+          runRight_ $ do
+            get alice =##> \case ("", c, Msg' _ pq msg2') -> c == bobId && pq == pqEnc && msg2 == msg2'; _ -> False
+            ackMessage alice bobId (baseId + 4) Nothing
  where
    withServer :: (ThreadId -> IO a) -> IO a
    withServer = withServer_ testStoreLogFile testStoreMsgsFile testPort
@@ -610,7 +610,7 @@ testRestoreMessages at@(ATransport t) =

    logSize testStoreLogFile `shouldReturn` 2
    logSize testStoreMsgsFile `shouldReturn` 5
-    logSize testServerStatsBackupFile `shouldReturn` 62
+    logSize testServerStatsBackupFile `shouldReturn` 71
    Right stats1 <- strDecode <$> B.readFile testServerStatsBackupFile
    checkStats stats1 [rId] 5 1

@@ -628,7 +628,7 @@ testRestoreMessages at@(ATransport t) =
    logSize testStoreLogFile `shouldReturn` 1
    -- the last message is not removed because it was not ACK'd
    logSize testStoreMsgsFile `shouldReturn` 3
-    logSize testServerStatsBackupFile `shouldReturn` 62
+    logSize testServerStatsBackupFile `shouldReturn` 71
    Right stats2 <- strDecode <$> B.readFile testServerStatsBackupFile
    checkStats stats2 [rId] 5 3

@@ -647,7 +647,7 @@ testRestoreMessages at@(ATransport t) =

    logSize testStoreLogFile `shouldReturn` 1
    logSize testStoreMsgsFile `shouldReturn` 0
-    logSize testServerStatsBackupFile `shouldReturn` 62
+    logSize testServerStatsBackupFile `shouldReturn` 71
    Right stats3 <- strDecode <$> B.readFile testServerStatsBackupFile
    checkStats stats3 [rId] 5 5