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

@@ -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

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

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