docs: drop non-findings sections from leak report

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.

bench/NetLag.hs

@@ -0,0 +1,111 @@
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE FlexibleInstances #-}
+{-# LANGUAGE InstanceSigs #-}
+{-# LANGUAGE KindSignatures #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TypeApplications #-}
+
+-- | A latency-injecting Transport for the memory-leak bench.
+--
+-- 'LagTLS' is a newtype over 'TLS' that delegates every 'Transport' method, adding a
+-- configurable delay before each read and write and optionally swallowing writes. It is
+-- wire-identical to 'TLS' - 'transportName' is only used for thread labels and logging, so a
+-- peer speaking plain TLS interoperates with it unchanged.
+--
+-- Used as the destination relay's listener transport so that proxy->relay traffic can be
+-- delayed without touching the proxy, the clients, or any production code:
+--
+-- > withSmpServerConfigOn (transport @TLS) proxyCfg testPort $ \_ ->
+-- >   withSmpServerConfigOn (transport @LagTLS) cfgJ2 testPort2 $ \_ -> ...
+--
+-- 'setDropSnd' keeps the TLS session healthy while responses vanish, which is what the
+-- proxy sentCommands leak needs: the relay must stay up and stop answering, so the proxy's
+-- RFWD requests time out without the client being torn down.
+--
+-- Delays apply to the SMP handshake as well as to post-handshake traffic (both go through
+-- cGet/cPut), so phases that need an established session must connect first and arm the lag
+-- afterwards.
+module NetLag
+  ( LagTLS,
+    setLag,
+    setDropSnd,
+    setDropEvery,
+    clearLag,
+  )
+where
+
+import Control.Concurrent (threadDelay)
+import Control.Concurrent.STM
+import Control.Monad (unless, when)
+import Data.ByteString.Char8 (ByteString)
+import Simplex.Messaging.Transport
+import System.IO.Unsafe (unsafePerformIO)
+
+newtype LagTLS (p :: TransportPeer) = LagTLS (TLS p)
+
+data LagCtl = LagCtl
+  { rcvDelayUs :: TVar Int,
+    sndDelayUs :: TVar Int,
+    dropSnd :: TVar Bool,
+    -- drop every nth write, 0 disables. Distinct from dropSnd: dropping everything makes the
+    -- peer's monitor eventually tear the session down, while dropping a fraction keeps the
+    -- session healthy indefinitely because any reply resets its counters.
+    dropEvery :: TVar Int,
+    sndSeq :: TVar Int
+  }
+
+-- A single process-wide control: getTransportConnection has nowhere to thread per-listener
+-- state through, and the bench runs one lagged relay at a time.
+lagCtl :: LagCtl
+lagCtl =
+  unsafePerformIO $
+    LagCtl <$> newTVarIO 0 <*> newTVarIO 0 <*> newTVarIO False <*> newTVarIO 0 <*> newTVarIO 0
+{-# NOINLINE lagCtl #-}
+
+-- | One-way delays in microseconds: inbound (peer -> this transport) and outbound.
+setLag :: Int -> Int -> IO ()
+setLag rcv snd' = atomically $ do
+  writeTVar (rcvDelayUs lagCtl) rcv
+  writeTVar (sndDelayUs lagCtl) snd'
+
+-- | Silently discard everything written. The session stays open and the peer keeps waiting.
+setDropSnd :: Bool -> IO ()
+setDropSnd b = atomically $ writeTVar (dropSnd lagCtl) b
+
+-- | Silently discard every nth write, passing the rest. 0 disables.
+setDropEvery :: Int -> IO ()
+setDropEvery n = atomically $ writeTVar (dropEvery lagCtl) n
+
+clearLag :: IO ()
+clearLag = setLag 0 0 >> setDropSnd False >> setDropEvery 0
+
+delayBy :: TVar Int -> IO ()
+delayBy v = do
+  d <- readTVarIO v
+  when (d > 0) $ threadDelay d
+
+instance Transport LagTLS where
+  transportName _ = "LagTLS"
+  transportConfig (LagTLS t) = transportConfig t
+  getTransportConnection cfg sent chain ctx = LagTLS <$> getTransportConnection cfg sent chain ctx
+  certificateSent (LagTLS t) = certificateSent t
+  getPeerCertChain (LagTLS t) = getPeerCertChain t
+  getSessionALPN (LagTLS t) = getSessionALPN t
+  tlsUnique (LagTLS t) = tlsUnique t
+  closeConnection (LagTLS t) = closeConnection t
+
+  cGet :: LagTLS p -> Int -> IO ByteString
+  cGet (LagTLS t) n = delayBy (rcvDelayUs lagCtl) >> cGet t n
+
+  cPut :: LagTLS p -> ByteString -> IO ()
+  cPut (LagTLS t) s = do
+    delayBy (sndDelayUs lagCtl)
+    drop' <- atomically $ do
+      always <- readTVar (dropSnd lagCtl)
+      every <- readTVar (dropEvery lagCtl)
+      i <- stateTVar (sndSeq lagCtl) $ \n -> let n' = n + 1 in (n', n')
+      pure $ always || (every > 0 && i `mod` every == 0)
+    unless drop' $ cPut t s
+
+  getLn :: LagTLS p -> IO ByteString
+  getLn (LagTLS t) = delayBy (rcvDelayUs lagCtl) >> getLn t

docs/leak-findings.md

@@ -0,0 +1,283 @@
+# SMP server leak findings
+
+Found with `bench/MemBench.hs`, extended with a proxy plus relay topology and a transport that
+adds latency and drops replies. Measured on both the journal store and PostgreSQL, same results.
+
+Three leaks on the proxy path, all client reachable. Two related bugs. TLS/TCP stack is clean.
+
+All three leaks are reached the same way: `PRXY` is unauthenticated unless `newQueueBasicAuth` is
+set (`Server.hs:1534`) and names an arbitrary destination, so a client can point the proxy at a
+relay it controls.
+
+---
+
+## Leak 1: forwarded commands are never removed on timeout
+
+### Issue
+
+Entries go into `sentCommands` in `mkTransmission_` (`Client.hs:1418`). The only removal is in
+`processMsg` (`Client.hs:706`), which runs when a reply arrives. `getResponse` (`Client.hs:1383`)
+handles the timeout but never gets the map, so it cannot delete.
+
+Each entry holds the forwarded command: 16226 bytes for `RFWD`.
+
+The session survives too. `monitor` (`Client.hs:668`) drops the client only when
+`timeoutErrorCount >= smpPingCount` and nothing has arrived for 900s, and `receive`
+(`Client.hs:663`) resets both on every inbound transmission.
+
+### Impact
+
+About 20 KiB per unanswered forward. How long it is held depends on how the relay misbehaves.
+
+| relay behaviour | result |
+| --- | --- |
+| slow but still replies | not a leak here, the late reply deletes the entry (but see Leak 3) |
+| goes fully silent | bounded, `monitor` closes the client at ~20 min |
+| replies to some, drops others | **unbounded** |
+
+The third case is the problem. Any arriving reply resets `lastReceived` and `timeoutErrorCount`,
+so the drop condition is never met. Measured with 1 in 3 relay writes dropped and forwarding
+running: `proxy_sentCommands` climbed 64 to 1280 over 20 minutes, linear at 64/min, zero
+disconnects. That is ~1.3 MiB/min, ~77 MiB/hour, on one session.
+
+Traffic has to be ongoing. Flood and stop and it is reclaimed after 20 minutes.
+
+The ntf server uses the same client code and has the same exposure through unanswered `NSUB`.
+Measured with `subtmo 200`: 200 queues, 200 timed out, `sentCommands` 0 to 200, ~1.76 KiB each.
+At the ntf batch size of 1360 that is ~2.3 MiB per unanswered batch.
+
+Pings do not help. Subscribe paths call `enablePings` and the proxy send path does not, but in
+the unbounded case replies are arriving anyway, which resets the counters either way.
+
+### Fix
+
+Do not just delete on timeout. The agent still needs late replies. `processMsg` forwards them as
+`STResponse` (`Client.hs:713`), and `Agent.hs:3093` acts on them. A late `OK`/`SOK` to a `SUB`
+calls `processSubOk`, which is what brings a connection back UP, and a late `MSG` is processed as
+a real message. Deleting on timeout turns both into `STUnexpectedError` (`Client.hs:702`), so the
+agent would report an error instead of recovering, and drop the message.
+
+Delete by age instead: record the time on `Request` when it is added, and remove entries with
+`pending == False` that are older than the point where a late reply is no longer useful. Choosing
+that age needs the agent's recovery behaviour measured, which is not done here.
+
+Two entry points can be fixed by deleting, because no reply is ever coming. `mkTransmission_`
+inserts before sending (`Client.hs:1361`) and `sendRecv` returns early at `Client.hs:1366`
+(transport error) and `Client.hs:1368` (oversized block) without sending or deleting.
+
+---
+
+## Leak 2: failed relay connects are never cleared
+
+### Issue
+
+A failed connect is cached in `smpClients` as `Left (error, expiry)` (`Client/Agent.hs:275`) and
+removed only on a later lookup of the same server (`:250`, `:411`). Nothing removes it on a timer.
+The other removals are `clientDisconnected` (`:311`, connected clients only) and shutdown
+(`:427`).
+
+Conditional on `persistErrorInterval > 0`. At 0 the entry goes immediately, but production sets
+30 (`Server/Main.hs:607`).
+
+### Impact
+
+Host, port and key hash come from the client, so distinct addresses are unlimited. Measured:
+`proxy_smpClients = 300` after 300 dead addresses, ~19 KiB each, never freed while the process
+runs. 1000 entries created in about 1s.
+
+That is ~19 MiB/s when the address refuses the connection immediately. An address that accepts
+nothing and never answers waits out the 45s connect timeout, which slows it right down.
+
+### Fix
+
+Check the map on a timer and remove entries past their expiry. The timestamp is already stored.
+
+---
+
+## Leak 3: the proxy's relay message queue has no reader
+
+### Issue
+
+`newSMPClientAgent` creates one `msgQ` (`Client/Agent.hs:194`) and `connectClient` gives that same
+queue to every relay client (`:296`). The ntf server reads its copy
+(`Notifications/Server.hs:537`). The SMP server never reads its own: `receiveFromProxyAgent`
+reads `agentQ` only (`Server.hs:475`). There are three `readTBQueue` sites on a `msgQ` in `src/`
+and none is the proxy's.
+
+It fills from late replies. `processMsg` routes a response to `msgQ` when the request is still in
+`sentCommands` but `pending` is already `False` (`Client.hs:713`), so every reply arriving after
+the proxy's 30s RFWD timeout leaves an entry that nothing takes out.
+
+When it is full, `processMsgs` blocks in `writeTBQueue` (`Client.hs:694`). That is the `process`
+thread, the only reader of `rcvQ`, so the proxy stops handling responses entirely.
+
+### Impact
+
+Measured with the `msgqfill` phase: 4 forwards at 40s each way so replies land after the timeout,
+then lag cleared and 3 more attempted. Only `msgQSize` differs.
+
+| `msgQSize` | `proxy_msgQ` at end | `sentCommands` at end | recovery forwards |
+| --- | --- | --- | --- |
+| 2 | 2 (at cap) | 4 and climbing | **0 of 3** |
+| 2048 (production) | 4 | 0 | 3 of 3 |
+
+Two things. The queue is never emptied: at production size it still holds the 4 late replies at
+the end of the run. And when it fills the stall is permanent, not slow: the recovery forwards ran
+with no latency at all and got nothing back.
+
+One `msgQ` per agent and one `ProxyAgent` per server, so one slow relay stalls the proxy for every
+relay it talks to. That part is from the code, not measured: the bench has one relay.
+
+Someone who controls the destination relay only needs 2048 late replies to do this.
+
+This also corrects Leak 1's "slow relay is not a leak" row. That is right about `sentCommands` and
+wrong about the session: the late replies that clear `sentCommands` are the ones that pile up
+here.
+
+### Fix
+
+Make `msgQ` optional in `SMPClientAgent` and pass `Nothing` for the proxy. `getProtocolClient`
+already takes a `Maybe` and `sendMsg` logs instead when it is `Nothing`. A discarding reader would
+also work but still allocates and copies every batch.
+
+---
+
+## Bug 3: proxy concurrency limit does nothing
+
+### Issue
+
+`Server.hs:1590`:
+
+```haskell
+bracket_ wait signal . forkClient clnt label $ action
+```
+
+`.` binds tighter than `$`, so `signal` runs when the thread starts, not when it finishes. Only
+forking is limited.
+
+Measured with `conclimit 8` and `serverClientConcurrency = 1`, eight concurrent PFWDs on one
+connection, relay silent:
+
+```
+conclimit: n=8 cap=1 completions first=20.0s last=20.0s spread=0.0s
+```
+
+All eight ran at once. Enforced, each would hold the slot for the 30s RFWD timeout, needing ~240s.
+
+### Impact
+
+No memory cost of its own. Removes the cap on how fast Leak 1 grows, and `procThreads` reads near
+zero at any load.
+
+### Fix
+
+```haskell
+wait >> forkClient clnt label (action `finally` signal)
+```
+
+This turns the limit on for the first time. Default is 32 and `wait` blocks the client's whole
+command loop when hit, so check that value first.
+
+---
+
+## Bug 4: stale endThreads entry when a command finishes fast
+
+### Issue
+
+`forkClient` (`Server.hs:1480`) registers the thread after `forkIO`. If the action finishes first,
+its delete misses and the insert is never undone.
+
+Reproduced in isolation, 100k forks: 20% stale at `-N1`, 13% at `-N4`, ~320 bytes each.
+`deRefWeak` returns `Nothing` for all of them, so no thread is retained.
+
+What decides it is not how long the child takes. Measured over 20k forks, varying only the child's
+work before its delete:
+
+| child does | -N1 | -N4 |
+| --- | --- | --- |
+| nothing | 17.5% | 10.7% |
+| spins 1us | 19.2% | 9.7% |
+| spins 100us | 17.8% | 9.8% |
+| one failing `connect()` | **0%** | **0.1%** |
+
+A child that just spins does not lose the race, it keeps the CPU from the parent. The parent only
+wins when the child hands the CPU back, which happens on a syscall, a safe FFI call, or an STM
+retry.
+
+Against that rule, of the three call sites:
+
+- `forkCmd` (`Server.hs:1593`) for `PFWD`/`PRXY`/`RSLV`: all do network IO, all yield. `RSLV` was
+  checked separately since it is client driven at command rate, but `resolveName` has no cache
+  (`Server/Names.hs:62`).
+- `deliverServiceMessages` (`Server.hs:1977`): `clientServiceSubscribed` is set to `True` once
+  (`Server.hs:2031`) and never reset, so this runs at most once per connection.
+- `sendPendingEvtsThread.queueEvts` (`Server.hs:463`): the only one that can finish without
+  yielding. The child does `writeTBQueue sndQ` and three `IORef` updates, so if space appeared in
+  the queue it finishes straight away.
+
+### Impact
+
+Small, and not reproduced against a running server.
+
+The one path that can skip yielding forks at most twice per client every 15s
+(`pendingENDInterval`, `Server/Main.hs:581`), and only when that client's `sndQ` was full at the
+check and had emptied by the time the child ran. `clientDisconnected` (`Server.hs:1237`) clears
+the whole map, so nothing outlives the session.
+
+A client can affect both of those by pausing and resuming its socket reads, so this is not out of
+reach, but hitting a sub-millisecond window at two tries per 15s was not shown.
+
+The real cost is that the `endThreads` counter is misleading: it mixes stale entries with commands
+that are actually still running.
+
+### Fix
+
+```haskell
+atomically $ modifyTVar' endThreads $ IM.insert tId Nothing        -- before forkIO
+atomically $ modifyTVar' endThreads $ IM.adjust (const (Just w)) tId
+```
+
+`adjust` is a no-op if the action already removed the key.
+
+---
+
+## Clean: TLS/TCP stack
+
+200 connections opened at once, closed, then measured again:
+
+| test | peak per conn | after 25s |
+| --- | --- | --- |
+| TCP connect, never start TLS | 48.2 KiB | 0.31 KiB |
+| TLS done, no SMP handshake | 203.1 KiB | 0.71 KiB |
+| handshake done, one byte, then quiet | 264.6 KiB | 0.87 KiB |
+
+All recovered. Also clean: 400 connect/disconnect rounds, and steady forwarding at 50ms each way.
+
+Measure well after closing. At +5s the middle two still read ~120 KiB per connection, which looks
+like a 24 MiB leak but is just connections still closing. A number that keeps falling is being
+freed; a number that stops above where it started is leaked.
+
+The peaks are still worth knowing. 200 abandoned half open connections hold ~40 MiB for ~25s, with
+no authentication needed. A client that finishes the handshake then sends one byte holds ~265 KiB
+for as long as it stays connected, because there is no read timeout: `transportTimeout` is
+hardcoded `Nothing` (`Transport/Server.hs:104`).
+
+## Clean: connectivity and sockets under latency
+
+Latency set with `BENCHLAG_MS` on `proxyfwd`, one way. Sockets counted from `/proc/<pid>/fd`.
+
+| lag each way | delivered | sockets | relay connects | reconnects | timeouts |
+| --- | --- | --- | --- | --- | --- |
+| 0ms | 12/12 | 8 | 1 | 0 | 0 |
+| 500ms | 10/10 | 8 | 1 | 0 | 0 |
+| 5s | 6/6 | 8 | 1 | 0 | 0 |
+| 16s | 4/4 | 8 | 1 | 0 | 0 |
+| 40s | 0/2 | 8 | 1 | 0 | 1 |
+
+Nothing builds up. The socket count is the same whether forwards succeed or time out, the session
+is opened once and reused, and there are no reconnects at any latency.
+
+This is why Leak 1 has no upper bound: the session holding the stuck entries never closes.
+
+Forwards work to 16s each way and fail at 40s, because of the 30s RFWD timeout. The exact cutoff is
+not measured, since the test transport adds its delay per read/write rather than per message.

simplexmq.cabal

@@ -1,7 +1,7 @@
 cabal-version: 3.0
 
 name:           simplexmq
-version:        7.0.1.0
+version:        7.0.0.6
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -652,3 +652,48 @@ 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:
+      NetLag
+      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

src/Simplex/Messaging/Agent/Store/SQLite.hs

@@ -50,6 +50,7 @@ import Data.Bits (xor)
 import Data.ByteArray (ScrubbedBytes)
 import qualified Data.ByteArray as BA
 import Data.ByteString (ByteString)
+import qualified Data.ByteString as B
 import Data.Functor (($>))
 import Data.IORef
 import Data.Maybe (fromMaybe)

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
 
@@ -107,7 +109,7 @@ data SMPClientAgentConfig = SMPClientAgentConfig
   { smpCfg :: ProtocolClientConfig SMPVersion,
     reconnectInterval :: RetryInterval,
     persistErrorInterval :: NominalDiffTime,
-    msgQSize :: Maybe Natural,
+    msgQSize :: Natural,
     agentQSize :: Natural,
     agentSubsBatchSize :: Int,
     ownServerDomains :: [ByteString]
@@ -124,7 +126,7 @@ defaultSMPClientAgentConfig =
             maxInterval = 10 * second
           },
       persistErrorInterval = 30, -- seconds
-      msgQSize = Just 2048,
+      msgQSize = 2048,
       agentQSize = 2048,
       agentSubsBatchSize = 1360,
       ownServerDomains = []
@@ -138,7 +140,7 @@ data SMPClientAgent p = SMPClientAgent
     dbService :: Maybe DBService,
     active :: TVar Bool,
     startedAt :: UTCTime,
-    msgQ :: Maybe (TBQueue (ServerTransmissionBatch SMPVersion ErrorType BrokerMsg)),
+    msgQ :: TBQueue (ServerTransmissionBatch SMPVersion ErrorType BrokerMsg),
     agentQ :: TBQueue SMPClientAgentEvent,
     randomDrg :: TVar ChaChaDRG,
     smpClients :: TMap SMPServer SMPClientVar,
@@ -158,12 +160,40 @@ 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,
+    alMsgQ :: Int
+  }
+
+getAgentLeakStats :: SMPClientAgent p -> IO AgentLeakStats
+getAgentLeakStats SMPClientAgent {smpClients, smpSessions, activeServiceSubs, activeQueueSubs, pendingServiceSubs, pendingQueueSubs, smpSubWorkers, msgQ} = 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)
+  alMsgQ <- fromIntegral <$> atomically (lengthTBQueue msgQ)
+  pure AgentLeakStats {alSmpClients, alSmpSessions = M.size sess, alActiveServiceSubs, alActiveQueueSubs, alPendingServiceSubs, alPendingQueueSubs, alSmpSubWorkers, alSentCommands, alMsgQ}
+  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
   startedAt <- getCurrentTime
-  -- Only subscribing agents receive server transmissions, should not be created until processed to prevent deadlock.
-  msgQ <- mapM newTBQueueIO msgQSize
+  msgQ <- newTBQueueIO msgQSize
   agentQ <- newTBQueueIO agentQSize
   smpClients <- TM.emptyIO
   smpSessions <- TM.emptyIO
@@ -265,7 +295,7 @@ connectClient ca@SMPClientAgent {agentCfg, dbService, smpClients, smpSessions, m
   Nothing -> getClient cfg
   where
     cfg = smpCfg agentCfg
-    getClient cfg' = getProtocolClient randomDrg NRMBackground (1, srv, Nothing) cfg' [] msgQ startedAt clientDisconnected
+    getClient cfg' = getProtocolClient randomDrg NRMBackground (1, srv, Nothing) cfg' [] (Just msgQ) startedAt clientDisconnected
 
     clientDisconnected :: SMPClient -> IO ()
     clientDisconnected smp = do

src/Simplex/Messaging/Notifications/Server.hs

@@ -34,7 +34,6 @@ import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Either (partitionEithers)
 import Data.Functor (($>))
-import Data.Hashable (hash)
 import Data.IORef
 import Data.Int (Int64)
 import qualified Data.IntSet as IS
@@ -527,10 +526,10 @@ subscribeNtfs NtfSubscriber {smpSubscribers, subscriberSeq, smpAgent = ca} st sm
       subscribeQueuesNtfs ca smpServer' [sub]
 
 ntfSubscriber :: NtfSubscriber -> M ()
-ntfSubscriber NtfSubscriber {smpAgent = ca@SMPClientAgent {msgQ = msgQ_, agentQ}} =
+ntfSubscriber NtfSubscriber {smpAgent = ca@SMPClientAgent {msgQ, agentQ}} =
   race_ receiveSMP receiveAgent
   where
-    receiveSMP = forM_ msgQ_ $ \msgQ -> do
+    receiveSMP = do
       st <- asks store
       ps <- asks pushServer
       stats <- asks serverStats
@@ -641,13 +640,11 @@ showServer' :: SMPServer -> Text
 showServer' = decodeLatin1 . strEncode . host
 
 pushNotification :: NtfPushServer -> Maybe T.Text -> OwnServer -> NtfTknRec -> PushNotification -> M ()
-pushNotification s srvHost_ isOwn tkn@NtfTknRec {ntfTknId, token = token@(DeviceToken pp _)} ntf =
+pushNotification s srvHost_ isOwn tkn@NtfTknRec {token = token@(DeviceToken pp _)} ntf =
   ifM
     (pushProviderAllowed token)
-    (getOrCreatePushWorker s (srvHost_, pp, hash (unEntityId ntfTknId) `mod` pushWorkersPerServer) isOwn >>= atomically . (`writeTBQueue` (tkn, ntf)))
+    (getOrCreatePushWorker s (srvHost_, pp) isOwn >>= atomically . (`writeTBQueue` (tkn, ntf)))
     (logWarn "skipping disabled APNS test push provider")
-  where
-    pushWorkersPerServer = 8
 
 pushProviderAllowed :: DeviceToken -> M Bool
 pushProviderAllowed (DeviceToken PPApnsTest _) = asks (allowTestPushProvider . config)
@@ -660,8 +657,8 @@ guardPushProvider token action =
     action
     (pure $ NRErr $ CMD SMP.PROHIBITED)
 
-getOrCreatePushWorker :: NtfPushServer -> (Maybe T.Text, PushProvider, Int) -> OwnServer -> M (TBQueue (NtfTknRec, PushNotification))
-getOrCreatePushWorker s@NtfPushServer {pushWorkers, pushWorkerSeq, pushQSize} key@(srvHost_, _, _) isOwn = do
+getOrCreatePushWorker :: NtfPushServer -> (Maybe T.Text, PushProvider) -> OwnServer -> M (TBQueue (NtfTknRec, PushNotification))
+getOrCreatePushWorker s@NtfPushServer {pushWorkers, pushWorkerSeq, pushQSize} key@(srvHost_, _) isOwn = do
   ts <- liftIO getCurrentTime
   withGetSessVar' pushWorkerSeq key pushWorkers ts createWorker existingWorker
   where
@@ -734,7 +731,7 @@ runPushWorker s srvHost_ isOwn q = forever $ do
               _ -> err e
         err e = logError ("Push provider error (" <> tshow pp <> ", " <> tshow ntfTknId <> "): " <> tshow e) $> Left e
 
-pushWorkersQLength :: TMap (Maybe T.Text, PushProvider, Int) PushWorkerVar -> IO Natural
+pushWorkersQLength :: TMap (Maybe T.Text, PushProvider) PushWorkerVar -> IO Natural
 pushWorkersQLength workers = do
   ws <- readTVarIO workers
   foldM addQLength 0 ws

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

@@ -174,7 +174,7 @@ data SMPSubscriber = SMPSubscriber
   }
 
 data NtfPushServer = NtfPushServer
-  { pushWorkers :: TMap (Maybe T.Text, PushProvider, Int) PushWorkerVar, -- Int is the worker shard
+  { pushWorkers :: TMap (Maybe T.Text, PushProvider) PushWorkerVar,
     pushWorkerSeq :: TVar Int,
     pushQSize :: Natural,
     pushClients :: TMap PushProvider PushClientVar,

src/Simplex/Messaging/Server.hs

@@ -75,7 +75,7 @@ import Data.List.NonEmpty (NonEmpty (..), (<|))
 import qualified Data.List.NonEmpty as L
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
-import Data.Maybe (fromMaybe, isJust, isNothing)
+import Data.Maybe (fromMaybe, isJust, isNothing, listToMaybe)
 import Data.Semigroup (Sum (..))
 import qualified Data.Set as S
 import Data.Text (Text)
@@ -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,101 @@ 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
+      -- listening port identifies the server when several run in one process (bench topologies)
+      let srvPort = maybe "?" (\(p, _, _) -> p) $ listToMaybe transports
+      labelMyThread "leakDiagnosticsThread"
+      -- never let a diagnostics error crash the server (this thread is in raceAny_)
+      liftIO $ forever $ do
+        threadDelay $ sec * 1000000
+        tryAny (logLeakStats srvPort srv ms ns smpAgent) >>= either (logError . ("LEAKDIAG error: " <>) . tshow) (const $ pure ())
+
+    logLeakStats :: ServiceName -> Server s -> s -> NtfStore -> SMPClientAgent 'Sender -> IO ()
+    logLeakStats srvPort 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, alMsgQ} <- getAgentLeakStats smpAgent
+      logNote $
+        T.concat
+          [ "LEAKDIAG",
+            " srv=" <> T.pack srvPort,
+            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, f "proxy_msgQ" alMsgQ
+          ]
+      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

src/Simplex/Messaging/Server/Main.hs

@@ -604,7 +604,6 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
                               }
                         },
                     ownServerDomains = either (const []) textToOwnServers $ lookupValue "PROXY" "own_server_domains" ini,
-                    msgQSize = Nothing, -- to prevent accumulation of late responses, and deadlocks in SMP proxy
                     persistErrorInterval = 30 -- seconds
                   },
               allowSMPProxy = True,

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
 

tests/SMPClient.hs

@@ -275,7 +275,7 @@ cfgMS msType = withStoreCfg (testServerStoreConfig msType) $ \serverStoreCfg ->
       smpServerVRange = supportedServerSMPRelayVRange,
       transportConfig = mkTransportServerConfig True (Just alpnSupportedSMPHandshakes) True,
       controlPort = Nothing,
-      smpAgentCfg = defaultSMPClientAgentConfig {persistErrorInterval = 1, msgQSize = Nothing}, -- seconds
+      smpAgentCfg = defaultSMPClientAgentConfig {persistErrorInterval = 1}, -- seconds
       allowSMPProxy = False,
       serverClientConcurrency = 2,
       serverResolverConcurrency = defaultNameResolverConcurrency,