tests: add proxy msgQ retention bench and counter

bench/MemBench.hs

@@ -57,7 +57,8 @@ import SMPClient
 import Simplex.Messaging.Client
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Protocol
-import Simplex.Messaging.Server.Env.STM (AStoreType (..), ServerConfig (controlPort, controlPortAdminAuth, maxJournalMsgCount, msgQueueQuota, notificationExpiration, serverClientConcurrency))
+import Simplex.Messaging.Client.Agent (SMPClientAgentConfig (msgQSize))
+import Simplex.Messaging.Server.Env.STM (AStoreType (..), ServerConfig (controlPort, controlPortAdminAuth, maxJournalMsgCount, msgQueueQuota, notificationExpiration, serverClientConcurrency, smpAgentCfg))
 import Simplex.Messaging.Server.Expiration (ExpirationConfig (..))
 import Simplex.Messaging.Server.MsgStore.Types (SMSType (..), SQSType (..))
 import Simplex.Messaging.Transport
@@ -981,13 +982,18 @@ main = do
   -- them often enough to be useful over a bench run
   leakDiagSec <- fromMaybe 10 . (>>= readMaybe) <$> lookupEnv "SMP_LEAKDIAG_SEC"
   setEnv "SMP_LEAKDIAG_SEC" (show leakDiagSec)
+  -- msgqfill: proxy agent msgQ size. Default 2 so the bound is reachable; set it to the
+  -- production 2048 to run the same phase as a control that must not stall.
+  msgQSz <- fromMaybe 2 . (>>= readMaybe) <$> lookupEnv "BENCHMSGQ"
   setLogLevel LogInfo
   withGlobalLogging LogConfig {lc_file = Nothing, lc_stderr = True} $
     if phase `elem` proxyPhases
       then
-        ( if phase == "conclimit"
-            then withProxyTopologyCfg (updateCfg (proxySrvCfg storeEnv) $ \c -> c {serverClientConcurrency = 1}) storeEnv
-            else withProxyTopology storeEnv
+        ( case phase of
+            "conclimit" -> withProxyTopologyCfg (updateCfg (proxySrvCfg storeEnv) $ \c -> c {serverClientConcurrency = 1}) storeEnv
+            -- shrink the proxy agent's msgQ so its bound is reachable in one run
+            "msgqfill" -> withProxyTopologyCfg (updateCfg (proxySrvCfg storeEnv) $ \c -> c {smpAgentCfg = (smpAgentCfg c) {msgQSize = msgQSz}}) storeEnv
+            _ -> withProxyTopology storeEnv
         )
           $ settle leakDiagSec
           $ case phase of
@@ -997,6 +1003,7 @@ main = do
         "subtmo" -> runSubTmo g iters cp
         "conclimit" -> runConcLimit g iters cp
         "fastfwd" -> runFastFwd g iters cp
+        "msgqfill" -> runMsgQFill g iters cp
         _ -> error $ "unknown proxy phase: " <> phase
       else withSmpServerConfigOn (transport @TLS) srvCfg testPort $ \_ -> settle leakDiagSec $ do
         threadDelay 250000
@@ -1018,8 +1025,50 @@ main = do
           "tlspartial" -> runTlsPartial iters cp
           _ -> error $ "unknown phase: " <> phase
 
+-- Leak 3: the proxy agent's msgQ has no reader.
+--
+-- newSMPClientAgent creates one msgQ (Client/Agent.hs:194) and connectClient passes that same
+-- queue to every relay client (:296). The ntf server drains its copy (Notifications/Server.hs:537);
+-- the SMP server does not - receiveFromProxyAgent reads agentQ only (Server.hs:475). So on a
+-- proxy the queue only ever fills.
+--
+-- What fills it: processMsg routes a response to msgQ when the request is found but `pending` is
+-- already False (Client.hs:713), i.e. the reply arrived after the proxy's own RFWD timeout. A
+-- relay that answers late therefore deposits one entry per late reply, permanently.
+--
+-- Once full, `process` blocks in writeTBQueue (Client.hs:694). It is the only reader of rcvQ, so
+-- every later response from that relay goes unprocessed and every forward times out. msgQ is
+-- shared across relays, so this is not confined to the relay that caused it.
+--
+-- Phase: force late replies until the queue is full, then clear the lag and try to forward again.
+-- A healthy proxy answers; a stalled one cannot. Run with BENCHMSGQ=2048 as the control.
+runMsgQFill :: TVar ChaChaDRG -> Int -> Int -> IO ()
+runMsgQFill g iters _cp = do
+  rq <- newRelayQueue g
+  pc <- proxyClient g 1
+  sess <- runExceptT' $ connectSMPProxiedRelay pc NRMInteractive relaySrv Nothing
+  -- must exceed the proxy's 30s RFWD timeout each way, or the reply is matched while still
+  -- pending and never reaches msgQ (measured: 20s each way is too little, 40s is enough)
+  lagMs <- fromMaybe 40000 . (>>= readMaybe) <$> lookupEnv "BENCHLAG_MS"
+  printf "msgqfill: lag=%dms each way, %d forwards to strand\n" lagMs iters
+  setLag (lagMs * 1000) (lagMs * 1000)
+  forM_ ([1 .. iters] :: [Int]) $ \_ ->
+    void $ runExceptT (proxySMPMessage pc NRMInteractive sess Nothing (rqSndId rq) noMsgFlags "late")
+  clearLag
+  -- let the stranded replies arrive and land in msgQ
+  printf "msgqfill: lag cleared, waiting %ds for late replies\n" (3 * lagMs `div` 1000)
+  threadDelay $ 3 * lagMs * 1000
+  -- recovery probe: no lag now, so a proxy whose process thread still runs must answer
+  ok <- newTVarIO (0 :: Int)
+  forM_ ([1 .. 3] :: [Int]) $ \_ ->
+    runExceptT (proxySMPMessage pc NRMInteractive sess Nothing (rqSndId rq) noMsgFlags "probe") >>= \case
+      Right (Right ()) -> atomically $ modifyTVar' ok (+ 1)
+      _ -> pure ()
+  o <- readTVarIO ok
+  printf "msgqfill: recovery forwards succeeded=%d/3 (0 means the proxy stalled)\n" o
+
 proxyPhases :: [String]
-proxyPhases = ["proxyfwd", "proxytmo", "proxychurn", "subtmo", "conclimit", "fastfwd"]
+proxyPhases = ["proxyfwd", "proxytmo", "proxychurn", "subtmo", "conclimit", "fastfwd", "msgqfill"]
 
 -- Hold the servers up past one LEAKDIAG interval after the phase finishes, so the end state is
 -- always sampled at least once. Short phases would otherwise exit before any line is emitted,

docs/leak-findings.md

@@ -3,7 +3,7 @@
 From `bench/MemBench.hs` extended with a proxy plus relay topology and a transport that adds
 latency and drops replies.
 
-Two leaks on the proxy path, both client reachable. Two related bugs. TLS/TCP stack clean.
+Three leaks on the proxy path, all client reachable. Two related bugs. TLS/TCP stack clean.
 
 Measured on both the journal store and the PostgreSQL queue and message store, which is the
 production configuration. Results are the same on both.
@@ -89,18 +89,27 @@ sustained traffic and some replies do arrive, and every arrival resets `lastRece
 
 ### Fix
 
-```haskell
-Nothing -> do
-  TM.delete corrId sentCommands                              -- new
-  modifyTVar' timeoutErrorCount (+ 1) $> Left PCEResponseTimeout
-```
+**Do not simply delete on timeout.** I had that here and it is wrong.
 
-Pass `sentCommands` and the request's `corrId` into `getResponse`. Double delete is harmless.
+Late responses are load bearing for the agent. When a reply arrives for a request that already
+timed out, `processMsg` takes the `wasPending == False` branch and forwards it as `STResponse`
+(`Client.hs:713`). `Agent.hs:3093` acts on those: a late `OK`/`SOK` to a `SUB` calls
+`processSubOk`, which is what brings the connection back UP, and a late `MSG` is processed as a
+real message. Deleting the entry on timeout turns both into `STUnexpectedError`
+(`Client.hs:702`), so the agent would report an error instead of recovering the subscription,
+and would drop the message. The ntf server ignores `STResponse` (`Notifications/Server.hs:540`)
+so it would only gain log noise, but the agent regression is real.
 
-Two more entry points leak with no timeout involved. `mkTransmission_` inserts the request
-before it is sent (`Client.hs:1361`), and `sendRecv` then returns early at `Client.hs:1366`
-(transport error) and `Client.hs:1368` (block over `blockSize - 2`) without sending or deleting.
-Both need the same delete.
+So the entry has to stay reachable for as long as a late reply is still useful, which rules out
+deleting it at the timeout. The fix has to bound the map by age instead: stamp `Request` on
+insert and sweep entries whose `pending` is `False` and whose stamp is older than the window in
+which a late reply could still matter. That needs a window chosen against the agent's
+subscription recovery behaviour, which I have not measured, so I am not proposing a number here.
+
+Two entry points do leak with no timeout involved and can be fixed as written, because no reply
+is ever coming. `mkTransmission_` inserts the request before it is sent (`Client.hs:1361`) and
+`sendRecv` then returns early at `Client.hs:1366` (transport error) and `Client.hs:1368` (block
+over `blockSize - 2`) without sending or deleting. Both should delete.
 
 ---
 
@@ -138,6 +147,64 @@ Sweep the map on a timer, dropping entries past their expiry. The timestamp is a
 
 ---
 
+## Leak 3: the proxy's relay message queue has no reader
+
+### Issue
+
+`newSMPClientAgent` creates one `msgQ` (`Client/Agent.hs:194`) and `connectClient` hands that
+same queue to every relay client it opens (`:296`). The ntf server drains its copy
+(`Notifications/Server.hs:537`). The SMP server never drains its own:
+`receiveFromProxyAgent` reads `agentQ` only (`Server.hs:475`). Grepping every `readTBQueue` on a
+`msgQ` in `src/` returns three sites, and none of them is the proxy's.
+
+What fills it: `processMsg` routes a response to `msgQ` when the request is still in
+`sentCommands` but `pending` is already `False` (`Client.hs:713`), meaning the reply arrived
+after the proxy's own RFWD timeout. So every late reply from a relay deposits one entry that is
+never taken out.
+
+When the queue is full, `processMsgs` blocks in `writeTBQueue` (`Client.hs:694`). That is the
+`process` thread, and it is the only reader of `rcvQ`, so once it blocks the proxy stops
+handling responses entirely and every subsequent forward times out.
+
+### Impact
+
+Measured with the `msgqfill` phase, 4 forwards at 40s each way so the replies land after the
+proxy's 30s timeout, then the lag is cleared and 3 more forwards are attempted. Only
+`msgQSize` differs between the runs.
+
+| `msgQSize` | `proxy_msgQ` at end | `proxy_sentCommands` at end | recovery forwards |
+|---|---|---|---|
+| 2 | 2 (at cap) | 4 and climbing | **0 of 3** |
+| 2048 (production) | 4 | 0 | 3 of 3 |
+
+Two separate things are shown. The queue never drains: at production size it holds the 4 late
+replies for the rest of the run. And when it does fill, the stall is permanent, not a slowdown:
+the recovery forwards ran with no latency at all and still got nothing back.
+
+The queue is shared, not per relay. There is one `msgQ` per `SMPClientAgent` and one
+`ProxyAgent` per server, so a single relay that answers slowly can stall the proxy's response
+handling for every relay it talks to. This part is from the code, not measured: the bench
+topology has one relay.
+
+Reachable the same way as Leak 1. `PRXY` is unauthenticated unless `newQueueBasicAuth` is set,
+and names an arbitrary destination, so a client can point the proxy at a relay it controls that
+answers just late enough. 2048 late replies is a cheap budget for that.
+
+Note this is the same relay behaviour I recorded under Leak 1 as "slow relay that still replies:
+transient, not a leak". That verdict was right about `sentCommands` and wrong about the session:
+the late replies that clear `sentCommands` are exactly the ones that accumulate here.
+
+### Fix
+
+Drain it, or do not create it. The SMP server has no use for these transmissions, so the honest
+options are to give `ProxyAgent` a reader that discards them, or to make `msgQ` optional in
+`SMPClientAgent` and pass `Nothing` for the proxy, which is already supported
+(`getProtocolClient` takes `Maybe`, and `sendMsg` logs instead when it is `Nothing`).
+
+The second is better: a discarding reader would still allocate and copy every batch.
+
+---
+
 ## Bug 3: proxy concurrency limit is inert
 
 ### Issue

src/Simplex/Messaging/Client/Agent.hs

@@ -170,11 +170,12 @@ data AgentLeakStats = AgentLeakStats
     alPendingServiceSubs :: Int,
     alPendingQueueSubs :: Int,
     alSmpSubWorkers :: Int,
-    alSentCommands :: Int
+    alSentCommands :: Int,
+    alMsgQ :: Int
   }
 
 getAgentLeakStats :: SMPClientAgent p -> IO AgentLeakStats
-getAgentLeakStats SMPClientAgent {smpClients, smpSessions, activeServiceSubs, activeQueueSubs, pendingServiceSubs, pendingQueueSubs, smpSubWorkers} = do
+getAgentLeakStats SMPClientAgent {smpClients, smpSessions, activeServiceSubs, activeQueueSubs, pendingServiceSubs, pendingQueueSubs, smpSubWorkers, msgQ} = do
   alSmpClients <- msize smpClients
   sess <- readTVarIO smpSessions
   alActiveServiceSubs <- msize activeServiceSubs
@@ -183,7 +184,8 @@ getAgentLeakStats SMPClientAgent {smpClients, smpSessions, activeServiceSubs, ac
   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}
+  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
 

src/Simplex/Messaging/Server.hs

@@ -549,7 +549,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
       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
+      AgentLeakStats {alSmpClients, alSmpSessions, alActiveServiceSubs, alActiveQueueSubs, alPendingServiceSubs, alPendingQueueSubs, alSmpSubWorkers, alSentCommands, alMsgQ} <- getAgentLeakStats smpAgent
       logNote $
         T.concat
           [ "LEAKDIAG",
@@ -564,7 +564,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
             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_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