From 5fa3c149e9ac10e1d3a08130dc67074f5962077a Mon Sep 17 00:00:00 2001 From: Evgeny Poberezkin Date: Thu, 25 Jul 2024 16:06:40 +0100 Subject: [PATCH] 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> --- src/Simplex/Messaging/Server.hs | 174 +++++++++++++------ src/Simplex/Messaging/Server/Env/STM.hs | 24 ++- src/Simplex/Messaging/Server/MsgStore/STM.hs | 14 -- src/Simplex/Messaging/Server/Stats.hs | 64 +++++++ tests/AgentTests/FunctionalAPITests.hs | 1 + tests/SMPProxyTests.hs | 14 +- tests/ServerTests.hs | 6 +- 7 files changed, 215 insertions(+), 82 deletions(-) diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index 34c7dd561..f8927d21d 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/src/Simplex/Messaging/Server.hs @@ -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 = diff --git a/src/Simplex/Messaging/Server/Env/STM.hs b/src/Simplex/Messaging/Server/Env/STM.hs index dc77b7481..7a3bcc362 100644 --- a/src/Simplex/Messaging/Server/Env/STM.hs +++ b/src/Simplex/Messaging/Server/Env/STM.hs @@ -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 diff --git a/src/Simplex/Messaging/Server/MsgStore/STM.hs b/src/Simplex/Messaging/Server/MsgStore/STM.hs index c8f78e2fb..6f9c40e58 100644 --- a/src/Simplex/Messaging/Server/MsgStore/STM.hs +++ b/src/Simplex/Messaging/Server/MsgStore/STM.hs @@ -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 diff --git a/src/Simplex/Messaging/Server/Stats.hs b/src/Simplex/Messaging/Server/Stats.hs index 2f4ed0613..377fddd1a 100644 --- a/src/Simplex/Messaging/Server/Stats.hs +++ b/src/Simplex/Messaging/Server/Stats.hs @@ -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, diff --git a/tests/AgentTests/FunctionalAPITests.hs b/tests/AgentTests/FunctionalAPITests.hs index 75593be73..53991ed12 100644 --- a/tests/AgentTests/FunctionalAPITests.hs +++ b/tests/AgentTests/FunctionalAPITests.hs @@ -38,6 +38,7 @@ module AgentTests.FunctionalAPITests rfGet, sfGet, nGet, + getInAnyOrder, (##>), (=##>), pattern CON, diff --git a/tests/SMPProxyTests.hs b/tests/SMPProxyTests.hs index 2aa89e06f..8044d23f7 100644 --- a/tests/SMPProxyTests.hs +++ b/tests/SMPProxyTests.hs @@ -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 diff --git a/tests/ServerTests.hs b/tests/ServerTests.hs index 9534b7902..60aa1dd1c 100644 --- a/tests/ServerTests.hs +++ b/tests/ServerTests.hs @@ -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