diff --git a/bench/MemBench.hs b/bench/MemBench.hs index 2da17b1ac..626a0017e 100644 --- a/bench/MemBench.hs +++ b/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, diff --git a/docs/leak-findings.md b/docs/leak-findings.md index 6fdefe809..fc21b7997 100644 --- a/docs/leak-findings.md +++ b/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 diff --git a/src/Simplex/Messaging/Client/Agent.hs b/src/Simplex/Messaging/Client/Agent.hs index 582658601..35c41ef43 100644 --- a/src/Simplex/Messaging/Client/Agent.hs +++ b/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 diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index fd9286caa..d6b2f2925 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/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