From 988851b83931de0d5846db84141d97b452512216 Mon Sep 17 00:00:00 2001 From: sh Date: Mon, 6 Jul 2026 12:33:48 +0000 Subject: [PATCH 1/4] smp-server: add leak diagnostics logging Add an exception-guarded periodic thread that logs a single greppable "LEAKDIAG" line censusing every growable in-memory structure: live threads, per-client endThreads and subscriptions (by SubThread state), subscriber maps, ntf store, store entity/loaded counts, and proxy agent maps with in-flight sentCommands. Interval via SMP_LEAKDIAG_SEC (default 60), no RTS flags required. Adds pClientSentCommandsCount and getAgentLeakStats accessors. --- src/Simplex/Messaging/Client.hs | 6 ++ src/Simplex/Messaging/Client/Agent.hs | 29 +++++++ src/Simplex/Messaging/Server.hs | 113 +++++++++++++++++++++++++- 3 files changed, 147 insertions(+), 1 deletion(-) diff --git a/src/Simplex/Messaging/Client.hs b/src/Simplex/Messaging/Client.hs index dc811fa0b..5cd268f21 100644 --- a/src/Simplex/Messaging/Client.hs +++ b/src/Simplex/Messaging/Client.hs @@ -35,6 +35,7 @@ module Simplex.Messaging.Client ProxiedRelay (..), getProtocolClient, closeProtocolClient, + pClientSentCommandsCount, protocolClientServer, protocolClientServer', transportHost', @@ -167,6 +168,7 @@ import Simplex.Messaging.Protocol import Simplex.Messaging.Protocol.Types import Simplex.Messaging.Server.QueueStore.QueueInfo import Simplex.Messaging.SimplexName (SimplexDomain) +import qualified Data.Map.Strict as M import Simplex.Messaging.TMap (TMap) import qualified Simplex.Messaging.TMap as TM import Simplex.Messaging.Transport @@ -737,6 +739,10 @@ useWebPort cfg presetDomains ProtocolServer {host = h :| _} = case smpWebPortSer SWPPreset -> isPresetDomain presetDomains h SWPOff -> False +-- | Count of in-flight (awaiting-response) commands on a client - for leak diagnostics. +pClientSentCommandsCount :: ProtocolClient v err msg -> IO Int +pClientSentCommandsCount ProtocolClient {client_ = PClient {sentCommands}} = M.size <$> readTVarIO sentCommands + isPresetDomain :: [HostName] -> TransportHost -> Bool isPresetDomain presetDomains = \case THDomainName h -> any (`isSuffixOf` h) presetDomains diff --git a/src/Simplex/Messaging/Client/Agent.hs b/src/Simplex/Messaging/Client/Agent.hs index e41d7a811..582658601 100644 --- a/src/Simplex/Messaging/Client/Agent.hs +++ b/src/Simplex/Messaging/Client/Agent.hs @@ -31,6 +31,8 @@ module Simplex.Messaging.Client.Agent removeActiveSubs, removePendingSub, removePendingSubs, + AgentLeakStats (..), + getAgentLeakStats, ) where @@ -158,6 +160,33 @@ data SMPClientAgent p = SMPClientAgent type OwnServer = Bool +-- | Sizes of every per-server/per-session map in the client agent, plus total in-flight +-- forwarded commands - for leak diagnostics on the proxy path. +data AgentLeakStats = AgentLeakStats + { alSmpClients :: Int, + alSmpSessions :: Int, + alActiveServiceSubs :: Int, + alActiveQueueSubs :: Int, + alPendingServiceSubs :: Int, + alPendingQueueSubs :: Int, + alSmpSubWorkers :: Int, + alSentCommands :: Int + } + +getAgentLeakStats :: SMPClientAgent p -> IO AgentLeakStats +getAgentLeakStats SMPClientAgent {smpClients, smpSessions, activeServiceSubs, activeQueueSubs, pendingServiceSubs, pendingQueueSubs, smpSubWorkers} = do + alSmpClients <- msize smpClients + sess <- readTVarIO smpSessions + alActiveServiceSubs <- msize activeServiceSubs + alActiveQueueSubs <- msize activeQueueSubs + alPendingServiceSubs <- msize pendingServiceSubs + alPendingQueueSubs <- msize pendingQueueSubs + alSmpSubWorkers <- msize smpSubWorkers + alSentCommands <- foldM (\ !a (_, c) -> (a +) <$> pClientSentCommandsCount c) 0 (M.elems sess) + pure AgentLeakStats {alSmpClients, alSmpSessions = M.size sess, alActiveServiceSubs, alActiveQueueSubs, alPendingServiceSubs, alPendingQueueSubs, alSmpSubWorkers, alSentCommands} + where + msize m = M.size <$> readTVarIO m + newSMPClientAgent :: SParty p -> SMPClientAgentConfig -> Maybe DBService -> TVar ChaChaDRG -> IO (SMPClientAgent p) newSMPClientAgent agentParty agentCfg@SMPClientAgentConfig {msgQSize, agentQSize} dbService randomDrg = do active <- newTVarIO True diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index 36ac471f9..082f5c4d1 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/src/Simplex/Messaging/Server.hs @@ -98,7 +98,7 @@ import qualified Network.TLS as TLS import Numeric.Natural (Natural) import Simplex.Messaging.Agent.Lock import Simplex.Messaging.Client (ProtocolClient (thParams), ProtocolClientError (..), SMPClient, SMPClientError, clientHandlers, forwardSMPTransmission, smpProxyError, temporaryClientError) -import Simplex.Messaging.Client.Agent (OwnServer, SMPClientAgent (..), SMPClientAgentEvent (..), closeSMPClientAgent, getSMPServerClient'', isOwnServer, lookupSMPServerClient, getConnectedSMPServerClient) +import Simplex.Messaging.Client.Agent (AgentLeakStats (..), OwnServer, SMPClientAgent (..), SMPClientAgentEvent (..), closeSMPClientAgent, getAgentLeakStats, getSMPServerClient'', isOwnServer, lookupSMPServerClient, getConnectedSMPServerClient) import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Encoding import Simplex.Messaging.Encoding.String @@ -128,6 +128,7 @@ import Simplex.Messaging.Transport.Server import Simplex.Messaging.Util import Simplex.Messaging.Version import System.Environment (lookupEnv) +import Text.Read (readMaybe) import System.Exit (exitFailure, exitSuccess) import System.IO (hPrint, hPutStrLn, hSetNewlineMode, universalNewlineMode) import System.Mem.Weak (deRefWeak) @@ -174,6 +175,23 @@ data ClientSubAction type PrevClientSub s = (Client s, ClientSubAction, (EntityId, BrokerMsg)) +-- accumulator for per-client leak diagnostics (summed across all connected clients) +data ClientAgg = ClientAgg + { aggEndThreads :: !Int, -- Weak ThreadId registrations in endThreads (forkClient leak) + aggEndThreadSeq :: !Int, -- total forkClient forks ever (fork rate) + aggProcThreads :: !Int, + aggSubs :: !Int, -- entries in per-client subscriptions map + aggNoSub :: !Int, + aggPending :: !Int, -- SubPending: forked delivery threads not yet completed (blocked-thread candidates) + aggThread :: !Int, -- SubThread: live delivery threads + aggProhibit :: !Int, + aggNtfSubs :: !Int, + aggSvcSubsCount :: !Int, + aggRcvQ :: !Int, + aggSndQ :: !Int, -- full sndQ => forkDeliver blocks (leak trigger) + aggMsgQ :: !Int + } + smpServer :: forall s. MsgStoreClass s => TMVar Bool -> ServerConfig s -> Maybe AttachHTTP -> M s () smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOptions} attachHTTP_ = do s <- asks server @@ -191,6 +209,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt ( serverThread "server subscribers" s subscribers subscriptions serviceSubsCount (Just cancelSub) : serverThread "server ntfSubscribers" s ntfSubscribers ntfSubscriptions ntfServiceSubsCount Nothing : deliverNtfsThread s + : leakDiagnosticsThread s : sendPendingEvtsThread s : receiveFromProxyAgent pa : expireNtfsThread cfg @@ -496,6 +515,98 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt printMessageStats "STORE: messages" msgStats Left e -> logError $ "STORE: expireOldMessages, error expiring messages, " <> tshow e + -- Periodic comprehensive leak diagnostics: sizes of every growable structure in the + -- server, summed across clients, plus the proxy agent. Interval seconds via SMP_LEAKDIAG_SEC + -- (default 60). A single greppable "LEAKDIAG ..." line per interval; whichever counter grows + -- monotonically over time is the leak. + leakDiagnosticsThread :: Server s -> M s () + leakDiagnosticsThread srv = do + secStr <- liftIO $ lookupEnv "SMP_LEAKDIAG_SEC" + let sec = max 5 $ fromMaybe 60 (secStr >>= readMaybe) + ms <- asks msgStore + ns <- asks ntfStore + ProxyAgent {smpAgent} <- asks proxyAgent + labelMyThread "leakDiagnosticsThread" + -- never let a diagnostics error crash the server (this thread is in raceAny_) + liftIO $ forever $ do + threadDelay $ sec * 1000000 + tryAny (logLeakStats srv ms ns smpAgent) >>= either (logError . ("LEAKDIAG error: " <>) . tshow) (const $ pure ()) + + logLeakStats :: Server s -> s -> NtfStore -> SMPClientAgent 'Sender -> IO () + logLeakStats srv ms (NtfStore nsv) smpAgent = do +#if MIN_VERSION_base(4,18,0) + nThreads <- length <$> listThreads +#else + let nThreads = 0 :: Int +#endif + cls <- IM.elems <$> getServerClients srv + cc <- foldM accClient emptyAgg cls + (smpQ, smpS, smpC, smpT, smpP) <- subAgg (subscribers srv) + (ntfQ, ntfS, ntfC, ntfT, ntfP) <- subAgg (ntfSubscribers srv) + ntfMap <- readTVarIO nsv + ntfMsgs <- foldM (\ !a v -> (a +) . length <$> readTVarIO v) (0 :: Int) (M.elems ntfMap) + EntityCounts {queueCount, notifierCount, rcvServiceCount, ntfServiceCount, rcvServiceQueuesCount, ntfServiceQueuesCount} <- getEntityCounts @(StoreQueue s) (queueStore ms) + LoadedQueueCounts {loadedQueueCount, loadedNotifierCount, openJournalCount, queueLockCount, notifierLockCount} <- loadedQueueCounts ms + AgentLeakStats {alSmpClients, alSmpSessions, alActiveServiceSubs, alActiveQueueSubs, alPendingServiceSubs, alPendingQueueSubs, alSmpSubWorkers, alSentCommands} <- getAgentLeakStats smpAgent + logNote $ + T.concat + [ "LEAKDIAG", + f "threads" nThreads, f "clients" (length cls), + f "endThreads" (aggEndThreads cc), f "endThreadSeq" (aggEndThreadSeq cc), f "procThreads" (aggProcThreads cc), + f "subs" (aggSubs cc), f "subs_nosub" (aggNoSub cc), f "subs_pending" (aggPending cc), f "subs_thread" (aggThread cc), f "subs_prohibit" (aggProhibit cc), + f "ntfSubsClient" (aggNtfSubs cc), f "svcSubsCount" (aggSvcSubsCount cc), + f "rcvQ" (aggRcvQ cc), f "sndQ" (aggSndQ cc), f "msgQ" (aggMsgQ cc), + f "smp_qSubscribers" smpQ, f "smp_svcSubscribers" smpS, f "smp_subClients" smpC, f "smp_totalSvcSubs" smpT, f "smp_pendingEvents" smpP, + f "ntf_qSubscribers" ntfQ, f "ntf_svcSubscribers" ntfS, f "ntf_subClients" ntfC, f "ntf_totalSvcSubs" ntfT, f "ntf_pendingEvents" ntfP, + f "ntfStore_keys" (M.size ntfMap), f "ntfStore_msgs" ntfMsgs, + f "store_queues" queueCount, f "store_notifiers" notifierCount, f "store_rcvServices" rcvServiceCount, f "store_ntfServices" ntfServiceCount, f "store_rcvSvcQueues" rcvServiceQueuesCount, f "store_ntfSvcQueues" ntfServiceQueuesCount, + f "loaded_queues" loadedQueueCount, f "loaded_notifiers" loadedNotifierCount, f "open_journals" openJournalCount, f "queue_locks" queueLockCount, f "notifier_locks" notifierLockCount, + f "proxy_smpClients" alSmpClients, f "proxy_smpSessions" alSmpSessions, f "proxy_activeSvcSubs" alActiveServiceSubs, f "proxy_activeQSubs" alActiveQueueSubs, f "proxy_pendingSvcSubs" alPendingServiceSubs, f "proxy_pendingQSubs" alPendingQueueSubs, f "proxy_subWorkers" alSmpSubWorkers, f "proxy_sentCommands" alSentCommands + ] + where + f :: Show a => Text -> a -> Text + f k v = " " <> k <> "=" <> tshow v + emptyAgg = ClientAgg 0 0 0 0 0 0 0 0 0 0 0 0 0 + accClient !agg Client {subscriptions, ntfSubscriptions, serviceSubsCount, procThreads, endThreads, endThreadSeq, rcvQ, sndQ, msgQ} = do + et <- IM.size <$> readTVarIO endThreads + es <- readTVarIO endThreadSeq + pt <- readTVarIO procThreads + subs <- readTVarIO subscriptions + (no, pe, th, pr) <- foldM accSub (0, 0, 0, 0) (M.elems subs) + nt <- M.size <$> readTVarIO ntfSubscriptions + sc <- fst <$> readTVarIO serviceSubsCount + (rl, sl, ml) <- atomically $ (,,) <$> lengthTBQueue rcvQ <*> lengthTBQueue sndQ <*> lengthTBQueue msgQ + pure + agg + { aggEndThreads = aggEndThreads agg + et, + aggEndThreadSeq = aggEndThreadSeq agg + es, + aggProcThreads = aggProcThreads agg + pt, + aggSubs = aggSubs agg + M.size subs, + aggNoSub = aggNoSub agg + no, + aggPending = aggPending agg + pe, + aggThread = aggThread agg + th, + aggProhibit = aggProhibit agg + pr, + aggNtfSubs = aggNtfSubs agg + nt, + aggSvcSubsCount = aggSvcSubsCount agg + fromIntegral sc, + aggRcvQ = aggRcvQ agg + fromIntegral rl, + aggSndQ = aggSndQ agg + fromIntegral sl, + aggMsgQ = aggMsgQ agg + fromIntegral ml + } + accSub (no, pe, th, pr) Sub {subThread} = case subThread of + ServerSub t -> + readTVarIO t >>= \case + NoSub -> pure (no + 1, pe, th, pr) + SubPending -> pure (no, pe + 1, th, pr) + SubThread _ -> pure (no, pe, th + 1, pr) + ProhibitSub -> pure (no, pe, th, pr + 1) + subAgg ServerSubscribers {queueSubscribers, serviceSubscribers, subClients, totalServiceSubs, pendingEvents} = do + q <- M.size <$> getSubscribedClients queueSubscribers + s' <- M.size <$> getSubscribedClients serviceSubscribers + sc <- IS.size <$> readTVarIO subClients + ts <- fst <$> readTVarIO totalServiceSubs + pe <- IM.size <$> readTVarIO pendingEvents + pure (q, s', sc, ts, pe) + expireNtfsThread :: ServerConfig s -> M s () expireNtfsThread ServerConfig {notificationExpiration = expCfg} = do ns <- asks ntfStore From a530653afe7c7d61f22a688c5ad5a3934a0a9bf7 Mon Sep 17 00:00:00 2001 From: sh Date: Mon, 6 Jul 2026 12:33:48 +0000 Subject: [PATCH 2/4] tests: add smp-server memory leak load bench Standalone smp-mem-bench executable that starts an in-process SMP server and drives churn workloads, reporting GHC live-heap residency per checkpoint after a forced major GC. Store selectable via BENCHSTORE (pgmsg/pgjournal/journal). Phases: plain, svc, svcrace, ntf, conc, svcsubs, getp, link, and leak repros - stuck (delivery threads blocked forever on a full sndQ), certchurn (serviceLocks/services grow per distinct service certificate), and ntfexp (NtfStore keys retained after notifications expire). --- bench/MemBench.hs | 452 ++++++++++++++++++++++++++++++++++++++++++++++ simplexmq.cabal | 44 +++++ 2 files changed, 496 insertions(+) create mode 100644 bench/MemBench.hs diff --git a/bench/MemBench.hs b/bench/MemBench.hs new file mode 100644 index 000000000..e1d432f79 --- /dev/null +++ b/bench/MemBench.hs @@ -0,0 +1,452 @@ +{-# LANGUAGE CPP #-} +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE DuplicateRecordFields #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE OverloadedLists #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE PatternSynonyms #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TupleSections #-} +{-# LANGUAGE TypeApplications #-} +{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-} + +-- | Memory-leak load driver for the SMP server. +-- +-- Starts an in-process SMP server (beta.2 code) and hammers a chosen command +-- path in a churn loop, printing GHC live-heap residency (measured after a +-- forced major GC) every checkpoint. A path whose residency climbs with the +-- iteration count is leaking; a flat path is clean. +-- +-- Usage: smp-mem-bench +-- phases: plain | svc | svcrace | ntf +module Main (main) where + +import Control.Concurrent (threadDelay) +import Control.Concurrent.Async (concurrently_, mapConcurrently_, withAsync) +import Control.Logger.Simple (LogConfig (..), LogLevel (..), setLogLevel, withGlobalLogging) +import Control.Concurrent.STM +import Control.Monad +import Crypto.Random (ChaChaDRG) +import qualified Data.ByteString.Char8 as B +import Data.ByteString.Char8 (ByteString) +import Data.List.NonEmpty (NonEmpty (..)) +import qualified Data.X509.Validation as XV +import GHC.Stats +import SMPClient +import qualified Simplex.Messaging.Crypto as C +import Simplex.Messaging.Protocol +import Simplex.Messaging.Server.Env.STM (AStoreType (..), ServerConfig (notificationExpiration)) +import Simplex.Messaging.Server.Expiration (ExpirationConfig (..)) +import Simplex.Messaging.Server.MsgStore.Types (SMSType (..), SQSType (..)) +import Simplex.Messaging.Transport +import Simplex.Messaging.Transport.Credentials (genCredentials, tlsCredentials) +import System.Environment (getArgs, lookupEnv) +import System.Mem (performMajorGC) +import System.Timeout (timeout) +import Text.Printf (printf) + +type H = THandleSMP TLS 'TClient + +-- store config: PostgreSQL (matches production) when built with -fserver_postgres, else journal +benchCfg :: AServerConfig +#if defined(dbServerPostgres) +benchCfg = cfgMS (ASType SQSPostgres SMSPostgres) +#else +benchCfg = cfg +#endif + +-- command helpers (copied from tests/ServerTests.hs) ------------------------ + +pattern Resp :: CorrId -> QueueId -> BrokerMsg -> Transmission (Either ErrorType BrokerMsg) +pattern Resp corrId queueId command <- (corrId, queueId, Right command) + +pattern New :: RcvPublicAuthKey -> RcvPublicDhKey -> Command 'Creator +pattern New rPub dhPub = NEW (NewQueueReq rPub dhPub Nothing SMSubscribe (Just (QRMessaging Nothing)) Nothing) + +pattern New0 :: RcvPublicAuthKey -> RcvPublicDhKey -> Command 'Creator +pattern New0 rPub dhPub = NEW (NewQueueReq rPub dhPub Nothing SMOnlyCreate (Just (QRMessaging Nothing)) Nothing) + +pattern Ids :: RecipientId -> SenderId -> RcvPublicDhKey -> BrokerMsg +pattern Ids rId sId srvDh <- IDS (QIK rId sId srvDh _sndSecure _linkId Nothing Nothing) + +pattern Ids_ :: RecipientId -> SenderId -> RcvPublicDhKey -> ServiceId -> BrokerMsg +pattern Ids_ rId sId srvDh serviceId <- IDS (QIK rId sId srvDh _sndSecure _linkId (Just serviceId) Nothing) + +pattern Msg :: MsgId -> MsgBody -> BrokerMsg +pattern Msg msgId body <- MSG RcvMessage {msgId, msgBody = EncRcvMsgBody body} + +_SEND :: MsgBody -> Command 'Sender +_SEND = SEND noMsgFlags + +_SEND' :: MsgBody -> Command 'Sender +_SEND' = SEND MsgFlags {notification = True} + +sendRecv :: forall p. PartyI p => H -> (Maybe TAuthorizations, ByteString, EntityId, Command p) -> IO (Transmission (Either ErrorType BrokerMsg)) +sendRecv h@THandle {params} (sgn, corrId, qId, cmd) = do + let TransmissionForAuth {tToSend} = encodeTransmissionForAuth params (CorrId corrId, qId, cmd) + Right () <- tPut1 h (sgn, tToSend) + tGet1 h + +signSendRecv :: forall p. PartyI p => H -> C.APrivateAuthKey -> (ByteString, EntityId, Command p) -> IO (Transmission (Either ErrorType BrokerMsg)) +signSendRecv h pk t = do + [r] <- signSendRecv_ h pk Nothing t + pure r + +serviceSignSendRecv :: forall p. PartyI p => H -> C.APrivateAuthKey -> C.PrivateKeyEd25519 -> (ByteString, EntityId, Command p) -> IO (Transmission (Either ErrorType BrokerMsg)) +serviceSignSendRecv h pk serviceKey t = do + [r] <- signSendRecv_ h pk (Just serviceKey) t + pure r + +signSendRecv_ :: forall p. PartyI p => H -> C.APrivateAuthKey -> Maybe C.PrivateKeyEd25519 -> (ByteString, EntityId, Command p) -> IO (NonEmpty (Transmission (Either ErrorType BrokerMsg))) +signSendRecv_ h@THandle {params} (C.APrivateAuthKey a pk) serviceKey_ (corrId, qId, cmd) = do + let TransmissionForAuth {tForAuth, tToSend} = encodeTransmissionForAuth params (CorrId corrId, qId, cmd) + Right () <- tPut1 h (authorize tForAuth, tToSend) + tGetClient h + where + authorize t = (,(`C.sign'` t) <$> serviceKey_) <$> case a of + C.SEd25519 -> Just . TASignature . C.ASignature C.SEd25519 $ C.sign' pk t' + C.SEd448 -> Just . TASignature . C.ASignature C.SEd448 $ C.sign' pk t' + C.SX25519 -> (\THAuthClient {peerServerPubKey = k} -> TAAuthenticator $ C.cbAuthenticate k pk (C.cbNonce corrId) t') <$> thAuth params +#if !MIN_VERSION_base(4,18,0) + _sx448 -> undefined +#endif + where + t' = case (serviceKey_, thAuth params >>= clientService) of + (Just _, Just THClientService {serviceCertHash = XV.Fingerprint fp}) -> fp <> t + _ -> t + +tPut1 :: H -> SentRawTransmission -> IO (Either TransportError ()) +tPut1 h t = do + [r] <- tPut h [Right t] + pure r + +tGet1 :: H -> IO (Transmission (Either ErrorType BrokerMsg)) +tGet1 h = do + [r] <- tGetClient h + pure r + +-- read and discard any pending transmissions until quiet, to resync after a race +drainAll :: H -> IO () +drainAll h = timeout 40000 (tGet1 h) >>= maybe (pure ()) (const $ drainAll h) + +-- measurement --------------------------------------------------------------- + +liveBytesMiB :: IO Double +liveBytesMiB = do + performMajorGC + s <- getRTSStats + pure $ fromIntegral (gcdetails_live_bytes (gc s)) / (1024 * 1024) + +report :: String -> Int -> Double -> Double -> IO () +report phase i base cur = + printf "%-8s iter=%7d live=%9.1f MiB delta=%+9.1f MiB (%+.4f KiB/iter)\n" + phase i cur (cur - base) (if i == 0 then 0 else (cur - base) * 1024 / fromIntegral i) + +withCheckpoints :: String -> Int -> Int -> (Int -> IO ()) -> IO () +withCheckpoints phase iters cp step = do + base <- liveBytesMiB + report phase 0 base base + forM_ ([1 .. iters] :: [Int]) $ \i -> do + step i + when (i `mod` cp == 0) $ liveBytesMiB >>= report phase i base + +-- phases -------------------------------------------------------------------- + +-- regular recipient: create+subscribe, send, receive, ack, delete (churn) +runPlain :: TVar ChaChaDRG -> Int -> Int -> IO () +runPlain g iters cp = + testSMPClient @TLS $ \recip -> + testSMPClient @TLS $ \sndr -> + withCheckpoints "plain" iters cp $ \i -> do + (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g + (dhPub, _dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g + let corr = B.pack (show i) + Resp _ _ (Ids rId sId _srvDh) <- signSendRecv recip rKey (corr, NoEntity, New rPub dhPub) + Resp _ _ OK <- sendRecv sndr (Nothing, corr, sId, _SEND "hello") + Resp _ _ (Msg mId _) <- tGet1 recip + Resp _ _ OK <- signSendRecv recip rKey (corr, rId, ACK mId) + Resp _ _ OK <- signSendRecv recip rKey (corr, rId, DEL) + pure () + +-- recipient-service: create queue as service, send, service receives, ack, delete (churn) +runSvc :: TVar ChaChaDRG -> Int -> Int -> IO () +runSvc g iters cp = do + creds <- genCredentials g Nothing (0, 2400) "localhost" + let (_fp, tlsCred) = tlsCredentials (creds :| []) + serviceKeys@(_, servicePK) <- atomically $ C.generateKeyPair g + testSMPClient @TLS $ \sndr -> + testSMPServiceClient @TLS (tlsCred, serviceKeys) $ \sh -> + withCheckpoints "svc" iters cp $ \i -> do + (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g + (dhPub, _dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g + let corr = B.pack (show i) + Resp _ _ (Ids_ rId sId _srvDh _serviceId) <- serviceSignSendRecv sh rKey servicePK (corr, NoEntity, New rPub dhPub) + Resp _ _ OK <- sendRecv sndr (Nothing, corr, sId, _SEND "hello") + Resp _ _ (Msg mId _) <- tGet1 sh + Resp _ _ OK <- signSendRecv sh rKey (corr, rId, ACK mId) + Resp _ _ OK <- signSendRecv sh rKey (corr, rId, DEL) + pure () + +-- recipient-service with concurrent SEND vs DEL to probe the TOCTOU orphan +runSvcRace :: TVar ChaChaDRG -> Int -> Int -> IO () +runSvcRace g iters cp = do + creds <- genCredentials g Nothing (0, 2400) "localhost" + let (_fp, tlsCred) = tlsCredentials (creds :| []) + serviceKeys@(_, servicePK) <- atomically $ C.generateKeyPair g + testSMPClient @TLS $ \sndr -> + testSMPServiceClient @TLS (tlsCred, serviceKeys) $ \sh -> + withCheckpoints "svcrace" iters cp $ \i -> do + (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g + (dhPub, _dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g + let corr = B.pack (show i) + Resp _ _ (Ids_ rId sId _srvDh _serviceId) <- serviceSignSendRecv sh rKey servicePK (corr, NoEntity, New rPub dhPub) + -- race an in-flight SEND (creates delivery Sub) against queue deletion + concurrently_ + (void $ sendRecv sndr (Nothing, corr, sId, _SEND "hello")) + (void $ signSendRecv sh rKey (corr, rId, DEL)) + drainAll sh + +-- notifications: enable ntf on live queues and send ntf-flagged messages without draining +runNtf :: TVar ChaChaDRG -> Int -> Int -> IO () +runNtf g iters cp = + testSMPClient @TLS $ \recip -> + testSMPClient @TLS $ \sndr -> + withCheckpoints "ntf" iters cp $ \i -> do + (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g + (dhPub, _dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g + (nPub, _nKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g + (rcvNtfPubDh, _dhNtfPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g + let corr = B.pack (show i) + Resp _ _ (Ids rId sId _srvDh) <- signSendRecv recip rKey (corr, NoEntity, New rPub dhPub) + Resp _ _ (NID _nId _) <- signSendRecv recip rKey (corr, rId, NKEY nPub rcvNtfPubDh) + -- ntf-flagged send stores a notification; no notifier subscribed -> stays in ntfStore + Resp _ _ OK <- sendRecv sndr (Nothing, corr, sId, _SEND' "hello") + Resp _ _ (Msg mId _) <- tGet1 recip + Resp _ _ OK <- signSendRecv recip rKey (corr, rId, ACK mId) + pure () + +-- reusable steps ------------------------------------------------------------ + +genKeys :: TVar ChaChaDRG -> IO (RcvPublicAuthKey, C.APrivateAuthKey, RcvPublicDhKey) +genKeys g = do + (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g + (dhPub, _dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g + pure (rPub, rKey, dhPub) + +plainStep :: TVar ChaChaDRG -> H -> H -> Int -> IO () +plainStep g recip sndr i = do + (rPub, rKey, dhPub) <- genKeys g + let corr = B.pack (show i) + Resp _ _ (Ids rId sId _) <- signSendRecv recip rKey (corr, NoEntity, New rPub dhPub) + Resp _ _ OK <- sendRecv sndr (Nothing, corr, sId, _SEND "hello") + Resp _ _ (Msg mId _) <- tGet1 recip + Resp _ _ OK <- signSendRecv recip rKey (corr, rId, ACK mId) + Resp _ _ OK <- signSendRecv recip rKey (corr, rId, DEL) + pure () + +ntfStep :: TVar ChaChaDRG -> H -> H -> Int -> IO () +ntfStep g recip sndr i = do + (rPub, rKey, dhPub) <- genKeys g + (nPub, _nKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g + (rcvNtfPubDh, _dh :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g + let corr = B.pack (show i) + Resp _ _ (Ids rId sId _) <- signSendRecv recip rKey (corr, NoEntity, New rPub dhPub) + Resp _ _ (NID _ _) <- signSendRecv recip rKey (corr, rId, NKEY nPub rcvNtfPubDh) + Resp _ _ OK <- sendRecv sndr (Nothing, corr, sId, _SEND' "hello") + Resp _ _ (Msg mId _) <- tGet1 recip + Resp _ _ OK <- signSendRecv recip rKey (corr, rId, ACK mId) + Resp _ _ OK <- signSendRecv recip rKey (corr, rId, DEL) + pure () + +-- concurrency/scale: many client connections running a mixed workload +runConc :: TVar ChaChaDRG -> Int -> Int -> IO () +runConc g iters _cp = do + let nW = 8 + per = max 1 (iters `div` nW) + base <- liveBytesMiB + report "conc" 0 base base + counter <- newTVarIO (0 :: Int) + let target = nW * per + monitor = do + threadDelay 2000000 + c <- readTVarIO counter + cur <- liveBytesMiB + report "conc" c base cur + when (c < target) monitor + worker w = + testSMPClient @TLS $ \recip -> + testSMPClient @TLS $ \sndr -> + forM_ ([1 .. per] :: [Int]) $ \i -> do + (if i `mod` 3 == 0 then ntfStep else plainStep) g recip sndr (w * per + i) + atomically $ modifyTVar' counter (+ 1) + withAsync monitor $ \_ -> mapConcurrently_ worker ([0 .. nW - 1] :: [Int]) + cur <- liveBytesMiB + report "conc" target base cur + +-- service SUBS reconnect: a fixed set of service queues, reconnect+resubscribe each iteration +runSvcSubs :: TVar ChaChaDRG -> Int -> Int -> IO () +runSvcSubs g iters cp = do + creds <- genCredentials g Nothing (0, 2400) "localhost" + let (_fp, tlsCred) = tlsCredentials (creds :| []) + serviceKeys@(_, servicePK) <- atomically $ C.generateKeyPair g + let aServicePK = C.APrivateAuthKey C.SEd25519 servicePK + nQueues = 20 + -- create nQueues associated with the service (persisted in the store) + (serviceId, rIds) <- testSMPServiceClient @TLS (tlsCred, serviceKeys) $ \sh -> do + xs <- forM ([1 .. nQueues] :: [Int]) $ \j -> do + (rPub, rKey, dhPub) <- genKeys g + Resp _ _ (Ids_ rId _sId _ sid) <- serviceSignSendRecv sh rKey servicePK (B.pack ("c" <> show j), NoEntity, New rPub dhPub) + pure (rId, sid) + pure (snd (head xs), map fst xs) + let idsHash = queueIdsHash rIds + base <- liveBytesMiB + report "svcsubs" 0 base base + -- each iteration: fresh service connection, SUBS to resubscribe all queues, then disconnect + forM_ ([1 .. iters] :: [Int]) $ \i -> do + testSMPServiceClient @TLS (tlsCred, serviceKeys) $ \sh -> + void $ signSendRecv_ sh aServicePK Nothing (B.pack (show i), serviceId, SUBS (fromIntegral nQueues) idsHash) + when (i `mod` cp == 0) $ liveBytesMiB >>= report "svcsubs" i base + +-- GET path: create-only (unsubscribed) queue, send, GET (poll), ack, delete +runGet :: TVar ChaChaDRG -> Int -> Int -> IO () +runGet g iters cp = + testSMPClient @TLS $ \recip -> + testSMPClient @TLS $ \sndr -> + withCheckpoints "getp" iters cp $ \i -> do + (rPub, rKey, dhPub) <- genKeys g + let corr = B.pack (show i) + Resp _ _ (Ids rId sId _) <- signSendRecv recip rKey (corr, NoEntity, New0 rPub dhPub) + Resp _ _ OK <- sendRecv sndr (Nothing, corr, sId, _SEND "hello") + Resp _ _ (Msg mId _) <- signSendRecv recip rKey (corr, rId, GET) + Resp _ _ OK <- signSendRecv recip rKey (corr, rId, ACK mId) + Resp _ _ OK <- signSendRecv recip rKey (corr, rId, DEL) + pure () + +-- forkDeliver blocked in SubPending: a subscriber that never reads its sndQ. +-- Deliveries to a full sndQ fork a deliverThread that blocks forever -> threads/subs_thread grow. +runStuck :: TVar ChaChaDRG -> Int -> Int -> IO () +runStuck g iters cp = + testSMPClient @TLS $ \sndr -> + testSMPClient @TLS $ \recip -> do + -- phase 1: create + subscribe `iters` queues (recip reads only the IDS responses, no messages yet) + qs <- forM ([1 .. iters] :: [Int]) $ \i -> do + (rPub, rKey, dhPub) <- genKeys g + Resp _ _ (Ids _rId sId _) <- signSendRecv recip rKey (B.pack ('c' : show i), NoEntity, New rPub dhPub) + pure sId + -- phase 2: send one message to each; recip never reads -> server delivery threads block on the full sndQ + base <- liveBytesMiB + report "stuck" 0 base base + forM_ (zip ([1 ..] :: [Int]) qs) $ \(i, sId) -> do + _ <- sendRecv sndr (Nothing, B.pack ('s' : show i), sId, _SEND "x") + when (i `mod` cp == 0) $ liveBytesMiB >>= report "stuck" i base + threadDelay 20000000 -- hold the subscriber open so diagnostics can sample the blocked threads + +-- serviceLocks + services never evicted: connect as a messaging service with a fresh certificate +-- each iteration. getCreateService (run in the handshake) adds a services row + serviceLocks entry +-- that is never removed -> store_rcvServices grows. +runCertChurn :: TVar ChaChaDRG -> Int -> Int -> IO () +runCertChurn g iters cp = do + base <- liveBytesMiB + report "certchurn" 0 base base + forM_ ([1 .. iters] :: [Int]) $ \i -> do + creds <- genCredentials g Nothing (0, 2400) "localhost" + let (_fp, tlsCred) = tlsCredentials (creds :| []) + serviceKeys <- atomically $ C.generateKeyPair g + testSMPServiceClient @TLS (tlsCred, serviceKeys) $ \_sh -> pure () + when (i `mod` cp == 0) $ liveBytesMiB >>= report "certchurn" i base + +-- LINK path coverage: create a queue with short-link data, update it (LSET), secure it via the +-- link (LKEY), delete the link data (LDEL), delete the queue. Exercises the links map on create +-- and delete. (Not a leak repro: the links-not-removed-on-delete bug only bites useCache=True.) +runLink :: TVar ChaChaDRG -> Int -> Int -> IO () +runLink g iters cp = + testSMPClient @TLS $ \r -> + testSMPClient @TLS $ \s -> + withCheckpoints "link" iters cp $ \i -> do + (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g + (dhPub, _dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g + (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g + C.CbNonce corrId <- atomically $ C.randomCbNonce g + let sId = EntityId $ B.take 24 $ C.sha3_384 corrId -- sender ID must be derived from corrId + ld = (EncDataBytes "fixed data", EncDataBytes "user data") + qrd = QRMessaging $ Just (sId, ld) + Resp _ NoEntity (IDS (QIK rId _sId _srvDh _qm (Just lnkId) _svc _ntf)) <- + signSendRecv r rKey (corrId, NoEntity, NEW (NewQueueReq rPub dhPub Nothing SMSubscribe (Just qrd) Nothing)) + Resp _ _ OK <- signSendRecv r rKey (B.pack ('a' : show i), rId, LSET lnkId ld) + Resp _ _ (LNK _sId2 _ld') <- signSendRecv s sKey (B.pack ('b' : show i), lnkId, LKEY sPub) + Resp _ _ OK <- signSendRecv r rKey (B.pack ('c' : show i), rId, LDEL) + Resp _ _ OK <- signSendRecv r rKey (B.pack ('d' : show i), rId, DEL) + pure () + +-- Note: the two proxy leaks are NOT reproducible in this load bench and are intentionally not +-- included. The sentCommands/PFWD-timeout leak needs a relay that keeps the session up but drops +-- RFWD (a mock relay). The empty-SessionVar leak is a precise disconnect-during-connect race +-- (reproduced deterministically by the SMPProxyTests unit test, not by a load loop). Both are +-- observable on a live proxy via the LEAKDIAG proxy_sentCommands / proxy_smpClients counters. + +-- NtfStore key-retention leak: store notifications in many notifier queues, then let them expire. +-- With the fix, deleteExpiredNtfs removes the emptied outer keys, so LEAKDIAG ntfStore_keys rises +-- during creation then falls to ~0 after expiry; without it, the empty keys are retained. +-- Run with the short-expiry config (see main) so expiry fires within the run. +runNtfExp :: TVar ChaChaDRG -> Int -> Int -> IO () +runNtfExp g iters _cp = + testSMPClient @TLS $ \recip -> + testSMPClient @TLS $ \sndr -> do + forM_ ([1 .. iters] :: [Int]) $ \i -> do + (rPub, rKey, dhPub) <- genKeys g + (nPub, _nKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g + (rcvNtfPubDh, _dh :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g + let corr = B.pack (show i) + Resp _ _ (Ids rId sId _) <- signSendRecv recip rKey (corr, NoEntity, New0 rPub dhPub) + Resp _ _ (NID _nId _) <- signSendRecv recip rKey (corr, rId, NKEY nPub rcvNtfPubDh) + Resp _ _ OK <- sendRecv sndr (Nothing, corr, sId, _SEND' "hi") + pure () + base <- liveBytesMiB + report "ntfexp" iters base base + -- hold while notifications expire; LEAKDIAG samples ntfStore_keys over this window + forM_ ([1 .. 12] :: [Int]) $ \k -> threadDelay 5000000 >> (liveBytesMiB >>= report "ntfexp" (iters * 10 + k) base) + +-- store config selectable via BENCHSTORE env: pgmsg (default, useCache=False) | pgjournal (useCache=True) | journal +srvStoreCfg :: Maybe String -> AServerConfig +srvStoreCfg = \case +#if defined(dbServerPostgres) + Just "pgjournal" -> cfgMS (ASType SQSPostgres SMSJournal) + Just "journal" -> cfgMS (ASType SQSMemory SMSJournal) + _ -> cfgMS (ASType SQSPostgres SMSPostgres) +#else + _ -> cfg +#endif + +main :: IO () +main = do + args <- getArgs + let (phase, iters) = case args of + (p : n : _) -> (p, read n) + [p] -> (p, 20000) + _ -> ("svc", 20000) + cp = max 1 (iters `div` 20) + g <- C.newRandom + storeEnv <- lookupEnv "BENCHSTORE" + -- ntfexp uses a short notification-expiration so deleteExpiredNtfs fires within the run + let srvCfg = case phase of + "ntfexp" -> updateCfg (srvStoreCfg storeEnv) $ \c -> c {notificationExpiration = ExpirationConfig {ttl = 2, checkInterval = 3}} + _ -> srvStoreCfg storeEnv + setLogLevel LogInfo + withGlobalLogging LogConfig {lc_file = Nothing, lc_stderr = True} $ + withSmpServerConfigOn (transport @TLS) srvCfg testPort $ \_ -> do + threadDelay 250000 + case phase of + "plain" -> runPlain g iters cp + "svc" -> runSvc g iters cp + "svcrace" -> runSvcRace g iters cp + "ntf" -> runNtf g iters cp + "conc" -> runConc g iters cp + "svcsubs" -> runSvcSubs g iters cp + "getp" -> runGet g iters cp + "stuck" -> runStuck g iters cp + "certchurn" -> runCertChurn g iters cp + "link" -> runLink g iters cp + "ntfexp" -> runNtfExp g iters cp + _ -> error $ "unknown phase: " <> phase diff --git a/simplexmq.cabal b/simplexmq.cabal index b017dc706..7c4fe4d61 100644 --- a/simplexmq.cabal +++ b/simplexmq.cabal @@ -652,3 +652,47 @@ test-suite simplexmq-test if flag(client_postgres) || flag(server_postgres) build-depends: postgresql-simple ==0.7.* + +executable smp-mem-bench + if flag(client_library) + buildable: False + main-is: MemBench.hs + other-modules: + SMPClient + Util + hs-source-dirs: + bench + tests + default-extensions: + StrictData + ghc-options: -Wall -Wno-unused-imports -Wno-unused-top-binds -Wno-name-shadowing -threaded -rtsopts "-with-rtsopts=-T" + build-depends: + base + , async + , bytestring + , containers + , crypton + , crypton-x509 + , crypton-x509-store + , crypton-x509-validation + , directory + , hspec ==2.11.* + , hspec-core ==2.11.* + , mtl + , network + , process + , simple-logger + , simplexmq + , sqlcipher-simple + , stm + , text + , time + , tls >=1.9.0 && <1.10 + , transformers + , unliftio + , unliftio-core + if flag(server_postgres) + cpp-options: -DdbServerPostgres + build-depends: + postgresql-simple ==0.7.* + default-language: Haskell2010 From 732f86bb01735270e8fc2e5bcbf77e6b657a2d2c Mon Sep 17 00:00:00 2001 From: sh Date: Mon, 6 Jul 2026 12:33:48 +0000 Subject: [PATCH 3/4] smp-server: fix service subscription memory leak Service subscription counters (totalServiceSubs, serviceSubsCount, ntfServiceSubsCount) are TVar (Int64, IdsHash). modifyTVar' only forces the pair to WHNF, so `n +/- n'` and `idsHash <> idsHash'` stay unevaluated and accumulate an unbounded thunk chain under subscription and delivery churn (the IdsHash chain also retains a bytestring per update) - a space leak proportional to the number of updates. Force both components in addServiceSubs/subtractServiceSubs. Verified with the load bench: svc churn drops from +5.4 KiB/iter (linear) to flat. --- src/Simplex/Messaging/Protocol.hs | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/Simplex/Messaging/Protocol.hs b/src/Simplex/Messaging/Protocol.hs index 8ef5bb04f..8730597f3 100644 --- a/src/Simplex/Messaging/Protocol.hs +++ b/src/Simplex/Messaging/Protocol.hs @@ -1546,11 +1546,14 @@ queueIdHash = IdsHash . C.md5Hash . unEntityId {-# INLINE queueIdHash #-} addServiceSubs :: (Int64, IdsHash) -> (Int64, IdsHash) -> (Int64, IdsHash) -addServiceSubs (n', idsHash') (n, idsHash) = (n + n', idsHash <> idsHash') +addServiceSubs (n', idsHash') (n, idsHash) = + let !n'' = n + n' + !h = idsHash <> idsHash' + in (n'', h) subtractServiceSubs :: (Int64, IdsHash) -> (Int64, IdsHash) -> (Int64, IdsHash) subtractServiceSubs (n', idsHash') (n, idsHash) - | n > n' = (n - n', idsHash <> idsHash') -- concat is a reversible xor: (x `xor` y) `xor` y == x + | n > n' = let !n'' = n - n'; !h = idsHash <> idsHash' in (n'', h) -- concat is a reversible xor: (x `xor` y) `xor` y == x | otherwise = (0, mempty) data ProtocolErrorType = PECmdSyntax | PECmdUnknown | PESession | PEBlock From 3cc449ba98f41cdfe8a8065a2f6ed4a8cea87e1e Mon Sep 17 00:00:00 2001 From: sh Date: Mon, 6 Jul 2026 12:33:48 +0000 Subject: [PATCH 4/4] smp-server: fix notification store key retention leak deleteExpiredNtfs trimmed each notifier's message list but never removed the outer NtfStore map key, so one empty entry per notifier queue that ever received a notification was retained forever (grows with the active notifier set, never shrinks). Remove the outer key when its list becomes empty, and make storeNtf fully atomic so it cannot race the removal and write a notification to an orphaned TVar. Verified with the load bench (ntfexp): after expiry ntfStore_keys drops from the queue count to 0 instead of staying flat. --- src/Simplex/Messaging/Server/NtfStore.hs | 28 ++++++++++-------------- 1 file changed, 12 insertions(+), 16 deletions(-) diff --git a/src/Simplex/Messaging/Server/NtfStore.hs b/src/Simplex/Messaging/Server/NtfStore.hs index b73fd4860..711072303 100644 --- a/src/Simplex/Messaging/Server/NtfStore.hs +++ b/src/Simplex/Messaging/Server/NtfStore.hs @@ -33,13 +33,10 @@ data MsgNtf = MsgNtf } storeNtf :: NtfStore -> NotifierId -> MsgNtf -> IO () -storeNtf (NtfStore ns) nId ntf = do - TM.lookupIO nId ns >>= atomically . maybe newNtfs (`modifyTVar'` (ntf :)) +storeNtf (NtfStore ns) nId ntf = -- TODO [ntfdb] coalesce messages here once the client is updated to process multiple messages -- for single notification. - -- when (isJust prevNtf) $ incStat $ msgNtfReplaced stats - where - newNtfs = TM.lookup nId ns >>= maybe (TM.insertM nId (newTVar [ntf]) ns) (`modifyTVar'` (ntf :)) + atomically $ TM.lookup nId ns >>= maybe (TM.insertM nId (newTVar [ntf]) ns) (`modifyTVar'` (ntf :)) deleteNtfs :: NtfStore -> NotifierId -> IO Int deleteNtfs (NtfStore ns) nId = atomically (TM.lookupDelete nId ns) >>= maybe (pure 0) (fmap length . readTVarIO) @@ -48,18 +45,17 @@ deleteExpiredNtfs :: NtfStore -> Int64 -> IO Int deleteExpiredNtfs (NtfStore ns) old = foldM (\expired -> fmap (expired +) . expireQueue) 0 . M.keys =<< readTVarIO ns where - expireQueue nId = TM.lookupIO nId ns >>= maybe (pure 0) expire - expire v = readTVarIO v >>= \case - [] -> pure 0 - _ -> - atomically $ readTVar v >>= \case - [] -> pure 0 - -- check the last message first, it is the earliest - ntfs | systemSeconds (ntfTs $ last $ ntfs) < old -> do + expireQueue nId = atomically $ TM.lookup nId ns >>= maybe (pure 0) (expire nId) + expire nId v = readTVar v >>= \case + [] -> TM.delete nId ns >> pure 0 + -- check the last message first, it is the earliest + ntfs + | systemSeconds (ntfTs $ last ntfs) < old -> do let !ntfs' = filter (\MsgNtf {ntfTs = ts} -> systemSeconds ts >= old) ntfs - writeTVar v ntfs' - pure $! length ntfs - length ntfs' - _ -> pure 0 + if null ntfs' + then TM.delete nId ns >> pure (length ntfs) + else writeTVar v ntfs' >> pure (length ntfs - length ntfs') + | otherwise -> pure 0 data NtfLogRecord = NLRv1 NotifierId MsgNtf