docs: mark fixed leak findings

# Conflicts:
#	src/Simplex/Messaging/Client.hs
#	tests/RSLVTests.hs

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,517 @@
+# SMP server leak findings
+
+Found with `bench/MemBench.hs` using a proxy-plus-relay topology and a transport that adds latency
+and drops replies. Journal store and PostgreSQL gave the same numbers except where a row says
+otherwise.
+
+Twelve findings: three proxy-path memory leaks (Leak 1-3), two `forkClient` bugs (Bug 3-4), an
+unauthenticated resolver fan-out (Bug 5), five PostgreSQL-backend costs (Bug 6-10), and a stuck proxy
+session (Bug 11). The TLS/TCP stack is clean (last two sections).
+
+Memory at production scale, with measurements after these fixes: `smp-server-memory.md`.
+
+## Overview
+
+| # | Issue | Reachable | Backend | Evidence | Status |
+| --- | --- | --- | --- | --- | --- |
+| Leak 1 | Forwarded commands not removed on timeout | Client, pre-auth (PRXY) | all | measured | fixed, `sh/fix-proxy-leak` |
+| Leak 2 | Failed relay connects not cleared | Client, pre-auth (PRXY) | all | measured | present |
+| Leak 3 | Proxy relay queue has no reader | Client, pre-auth (PRXY) | all | measured | fixed, #1839 |
+| Bug 3 | Proxy concurrency limit does nothing | Client, pre-auth (PRXY) | all | measured | fixed, `sh/fix-rslv-fanout` |
+| Bug 4 | Stale `endThreads` entry on fast finish | Client, limited window | all | isolation only | present |
+| Bug 5 | RSLV resolver fan-out | Client, if `[NAMES]` on | all | measured | fixed, `sh/fix-rslv-fanout` |
+| Bug 6 | SEND is three DB transactions | authenticated SEND | postgres | code review | present |
+| Bug 7 | Per-transmission verify and DB lookup | Client, pre-auth | postgres | code review | present |
+| Bug 8 | Service handshake grows `services` table | Client, pre-auth | postgres | code review | present |
+| Bug 9 | Prometheus scrape scans everything | internal, periodic | postgres | code review | present |
+| Bug 10 | Subscription churn serialized | authenticated SUB | all | code review | present |
+| Bug 11 | Proxy never drops a stuck relay session | Client, pre-auth (PRXY) | all | reproduced | fixed, `sh/fix-proxy-leak` |
+
+Leak 1, Leak 2, Leak 3, Bug 3, and Bug 11 share one entry point. `PRXY` is unauthenticated unless
+`newQueueBasicAuth` is set (`Server.hs:1534`) and it 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
+
+| | |
+| --- | --- |
+| Reachable | any client, pre-auth via PRXY |
+| Trigger | relay replies to some forwards, drops others |
+| Cost | ~20 KiB per stuck forward, unbounded |
+| Measured | yes, 64 to 1280 entries over 20 min on one session |
+| Status | present |
+
+### How it works
+
+When the agent forwards a command it stores the in-flight command in `sentCommands`, keyed by
+correlation id, so the reply can be matched later. The entry is added in `mkTransmission_`
+(`Client.hs:1414`) and removed in `processMsg` (`Client.hs:706`) when the reply arrives. Each `RFWD`
+entry holds the whole forwarded command, 16226 bytes (`Protocol.hs:319`).
+
+### The bug
+
+The timeout path, `getResponse` (`Client.hs:1378`), does not hold the map, so a forward that times
+out is never removed. The session does not die either: `monitor` (`Client.hs:668`) closes the client
+only after `timeoutErrorCount >= smpPingCount` with nothing received for 900s, and `receive`
+(`Client.hs:663`) resets both counters on any inbound transmission.
+
+### Impact
+
+About 20 KiB per unanswered forward. How long it is held depends on the relay.
+
+| 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 leak. Any arriving reply resets `lastReceived` and `timeoutErrorCount`, so the
+drop condition is never met. Measured with 1 in 3 relay writes dropped: `proxy_sentCommands` climbed
+64 to 1280 over 20 minutes, linear at 64/min, zero disconnects. That is ~1.3 MiB/min on one session.
+Traffic has to be ongoing; stop the flood and it is reclaimed after 20 minutes.
+
+The ntf server runs the same client code and is exposed the same way through unanswered `NSUB`.
+Measured with `subtmo 200`: 200 timed-out subscriptions, `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` (`Client.hs:953`) and the proxy send path does
+not, but in the unbounded case replies are arriving anyway and reset the counters.
+
+### Fix
+
+Deleting on timeout is wrong: the agent still needs late replies. `processMsg` forwards them as
+`STResponse` (`Client.hs:713`) and the agent acts on them, so a late `OK`/`SOK` to a `SUB` brings a
+connection back up and a late `MSG` is processed. Deleting would turn both into `STUnexpectedError`.
+
+Delete by age instead. Record the insert time on `Request` and remove `pending == False` entries once
+a late reply is no longer useful. The cutoff needs the agent's recovery behaviour measured, which is
+not done here. Two paths can be deleted immediately because no reply is coming: `sendRecv` returns
+early on transport error (`Client.hs:1364`) and on an oversized block (`Client.hs:1366`) without
+sending.
+
+---
+
+## Leak 2: failed relay connects are never cleared
+
+| | |
+| --- | --- |
+| Reachable | any client, pre-auth via PRXY |
+| Trigger | forward to distinct dead addresses |
+| Cost | ~19 KiB per address, held for the process lifetime |
+| Measured | yes, 300 entries from 300 dead addresses |
+| Status | present |
+
+### How it works
+
+Relay clients are cached in `smpClients`. A successful connect caches the client; a failed connect
+caches the error as `Left (error, expiry)` (`Client/Agent.hs:277`) so repeat lookups fail fast until
+the expiry passes.
+
+### The bug
+
+Nothing removes a failed entry on a timer. It is dropped only when the same server is looked up again
+and found expired (`:253`, `:414`). The other removals are `clientDisconnected` (`:314`, connected
+clients only) and shutdown (`:430`). Conditional on `persistErrorInterval > 0`; at 0 the entry is
+removed at once, but production sets 30 (`Server/Main.hs:608`).
+
+### 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, so ~19 MiB/s when the address refuses immediately. An address that
+accepts nothing and never answers waits out the 45s connect timeout, which slows it down.
+
+### Fix
+
+Scan 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
+
+> [!TIP]
+> Fixed in master by #1839 (`7d0820dd`), which took the approach proposed below. `msgQ` is now
+> `Maybe (TBQueue ...)` (`Client/Agent.hs:143`), the proxy agent is built with `msgQSize = Nothing`
+> (`Server/Main.hs:607`), and `sendMsg` logs late replies instead of enqueuing them (`Client.hs:722`).
+> The analysis below is kept for context.
+
+| | |
+| --- | --- |
+| Reachable | any client, pre-auth via PRXY |
+| Trigger | relay replies land after the 30s RFWD timeout |
+| Cost | permanent proxy stall once the queue fills |
+| Measured | yes, `msgqfill` phase |
+| Status | fixed, #1839 |
+
+### How it works
+
+`newSMPClientAgent` created one `msgQ` and gave that same queue to every relay client. The ntf server
+reads its copy (`Notifications/Server.hs`), but the SMP server never read its own: `receiveFromProxyAgent`
+reads `agentQ` only.
+
+### The bug
+
+The queue filled 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 left an entry that nothing removed. When full, `processMsgs`
+blocked in `writeTBQueue` on the `process` thread, the only reader of `rcvQ`, so the proxy stopped
+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 |
+
+The queue was never emptied, and once full the stall was permanent: the recovery forwards ran with no
+latency and got nothing back. One `msgQ` per agent and one `ProxyAgent` per server, so one slow relay
+stalled the proxy for every relay it talked to. Someone controlling the destination relay needed only
+2048 late replies. This also corrects Leak 1's "slow relay is not a leak" row: those late replies are
+what piled up here.
+
+---
+
+## Bug 3: proxy concurrency limit does nothing
+
+| | |
+| --- | --- |
+| Reachable | any client, pre-auth via PRXY |
+| Trigger | concurrent PFWDs on one connection |
+| Cost | none directly, uncaps Leak 1's growth rate |
+| Measured | yes, `conclimit` phase |
+| Status | present |
+
+### How it works
+
+`forkCmd` (`Server.hs:1591`) is meant to cap in-flight forked commands at `serverClientConcurrency`.
+`wait` blocks until a slot is free and `signal` releases it.
+
+### The bug
+
+The code is `bracket_ wait signal . forkClient clnt label $ action` (`Server.hs:1592`). `.` binds
+tighter than `$`, so it parses as `bracket_ wait signal (forkClient ... action)`. `forkClient`
+returns as soon as the thread is spawned, so `signal` runs at spawn time, not at completion. Only
+forking is serialized. Measured with `conclimit 8` and `serverClientConcurrency = 1`, all eight PFWDs
+ran at once (`first=20.0s last=20.0s spread=0.0s`); enforced, each would hold the slot for the 30s
+RFWD timeout. `procThreads` also reads near zero at any load. The counter is per-connection, so even
+when fixed the cap is per-client, not global.
+
+### Fix
+
+```haskell
+wait >> forkClient clnt label (action `finally` signal)
+```
+
+This enables the limit 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
+
+| | |
+| --- | --- |
+| Reachable | client, narrow timing window |
+| Trigger | forked child finishes before it is registered |
+| Cost | ~320 bytes per stale entry, misleading counter |
+| Measured | isolation only, not against a running server |
+| Status | present |
+
+### How it works
+
+`forkClient` (`Server.hs:1482`) registers the child in `endThreads` for shutdown tracking. It forks
+first (`:1485`), the child deletes its own key on exit (`:1487`), and the parent inserts the weak
+thread id afterward (`:1488`).
+
+### The bug
+
+If the child finishes before the parent's insert, the delete finds no key and the later insert is
+never undone, leaving a permanent entry. Reproduced over 100k forks: 20% stale at `-N1`, 13% at `-N4`,
+~320 bytes each. `deRefWeak` returns `Nothing` for all, so no thread is retained.
+
+What decides the race is whether the child yields the CPU back, not how long it runs:
+
+| 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 spins keeps the CPU; the parent wins only when the child hits a syscall, safe FFI call,
+or STM retry. Of the three call sites, `forkCmd` (`PFWD`/`PRXY`/`RSLV`) and `deliverServiceMessages`
+all do IO and yield, and the service path runs at most once per connection. Only
+`sendPendingEvtsThread.queueEvts` can finish without yielding, and it forks at most twice per client
+per 15s (`pendingENDInterval`) and only when the client's `sndQ` was full then emptied.
+`clientDisconnected` clears the map, so nothing outlives the session. The real cost is that the
+`endThreads` counter mixes stale entries with commands 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.
+
+---
+
+## Bug 5: unauthenticated RSLV resolver fan-out
+
+| | |
+| --- | --- |
+| Reachable | any client, pre-auth, when `[NAMES]` is enabled |
+| Trigger | a block full of RSLV commands |
+| Cost | up to 1000 concurrent outbound TLS handshakes per connection |
+| Measured | yes, `testRslvFanOut` (64 in flight, asserts <= 8) |
+| Status | present |
+
+### How it works
+
+`RSLV` resolves a name over an outbound HTTP/TLS request. It is verified as unauthenticated
+(`Server.hs:1393`, `vc SResolver (RSLV _) = VRVerified Nothing`) and each command forks one request
+(`Server.hs:1638`). One 16 KB block carries ~255 RSLVs. Off by default, on only with `[NAMES]`.
+
+### The bug
+
+The only bound is `serverResolverConcurrency` (default 1000, `Env/STM.hs:255`), applied through the
+per-client `procThreads` counter via the same broken `forkCmd` path as Bug 3, so it is neither global
+nor effective. `managerConnCount = 10` (`HttpResolver.hs:88`) sizes the keep-alive pool, not
+concurrency, so excess requests open extra connections rather than blocking. `resolveName` has no
+cache (`Server/Names.hs:61`). One connection drives up to `serverResolverConcurrency` concurrent
+outbound handshakes, sockets and FDs; more connections multiply it with no global bound.
+
+### Fix
+
+Add a global resolver-concurrency limit, a shared semaphore in `NamesEnv` acquired around the
+outbound call, separate from the per-client counter, and lower the default. A result cache would also
+cut repeat lookups.
+
+---
+
+The findings below are from code review of the PostgreSQL backend (`store_messages: database`), not
+bench-measured.
+
+## Bug 6: every SEND is three DB transactions behind a pool of 10
+
+| | |
+| --- | --- |
+| Reachable | authenticated SEND |
+| Trigger | normal message sending |
+| Cost | SEND rate capped at ~`poolSize / 3` per transaction latency |
+| Evidence | code review |
+| Status | present |
+
+### How it works
+
+With `store_messages: database` the queue store runs `useCache = False` (`MsgStore/Postgres.hs:100`),
+so each command hits the DB.
+
+### The bug
+
+A SEND does three separate transactions: load the queue (`QueueStore/Postgres.hs:230`),
+`delete_expired_msgs` (`MsgStore/Postgres.hs:289`), then `write_message` (`MsgStore/Postgres.hs:193`).
+`expireMessagesOnSend` defaults to `True` (`Main.hs:563`, `Server.hs:2118`), so the middle one runs on
+every SEND to a non-empty queue. All draw from one pool of `poolSize` (default 10, `Main/Init.hs:44`).
+A second pool, `dbPriorityPool`, is opened (`Agent/Store/Postgres.hs:59`) but the SMP server never
+uses it. Max SEND rate is about `poolSize / (3 x per-transaction latency)` regardless of client count;
+past that all clients block on the pool, and half the opened backends sit idle.
+
+### Fix
+
+Fewer transactions per SEND; make `expireMessagesOnSend` cheaper or default off; drop or use the
+priority pool.
+
+---
+
+## Bug 7: unauthenticated batched crypto and per-transmission DB lookup
+
+| | |
+| --- | --- |
+| Reachable | any handshake-completing peer, pre-auth |
+| Trigger | one 16 KB block of SENDs to unknown queues |
+| Cost | ~130-250 asymmetric verifications and DB SELECTs per block |
+| Evidence | code review |
+| Status | present |
+
+### How it works
+
+Recipient and Notifier commands are batch parties (`Protocol.hs:420`), so their queue lookups are
+batched. SEND is not, so it takes the per-transmission path
+`mapM (\t -> verifyTransmission ...)` (`Server.hs:1281`).
+
+### The bug
+
+Each transmission runs one crypto op and, with `useCache = False`, one `getQueueRec` SELECT
+(`Server.hs:1361`). For an unknown queue, `dummyVerifyCmd` (`Server.hs:1459`) still runs a real
+Ed25519/Ed448 verify or an X25519 DH. A 16 KB block holds up to 254 transmissions (`Protocol.hs:2292`,
+one-byte count), so one write by any peer that completed the handshake costs ~130-250 asymmetric
+verifications and ~130-250 SELECTs. The attacker chooses the auth type and that the queue is absent.
+
+### Fix
+
+Batch the sender lookups as Recipient and Notifier already are; cap unknown-queue verifications per
+block.
+
+---
+
+## Bug 8: unauthenticated service handshake grows the services table
+
+| | |
+| --- | --- |
+| Reachable | any client, pre-auth |
+| Trigger | a fresh self-signed service cert per handshake |
+| Cost | one persistent `services` row and one X509 verify per connection |
+| Evidence | code review |
+| Status | present |
+
+### How it works
+
+The SMP handshake accepts a self-signed service chain (`CCSelf`, `Transport.hs:759`) and verifies the
+supplied cert (`Transport.hs:763`). `getClientService` (`Server.hs:864`) then calls `getCreateService`.
+
+### The bug
+
+`getCreateService` inserts a `services` row on any new fingerprint (`QueueStore/Postgres.hs:478`), and
+a fresh self-signed cert is a new fingerprint every time. The only guard is a role check on an
+existing fingerprint, so each connection can create one unbounded, persistent row plus one X509 verify,
+all before authentication.
+
+### Fix
+
+Require the service to be pre-registered, or authenticate and rate-limit service creation.
+
+---
+
+## Bug 9: Prometheus scrape scans all queues and folds all subscriptions
+
+| | |
+| --- | --- |
+| Reachable | internal, fires every `prometheus_interval` |
+| Trigger | Prometheus enabled |
+| Cost | scales with queue count and live subscriptions, every scrape |
+| Evidence | code review |
+| Status | present |
+
+### How it works
+
+On each scrape the server refreshes its metrics from live state rather than incremental counters.
+
+### The bug
+
+`getEntityCounts` runs six `COUNT(1)` scans over `msg_queues` and `services`
+(`QueueStore/Postgres.hs:154`, called at `Server.hs:813`), and `getDeliveredMetrics` folds over all
+clients times all their subscriptions in memory (`Server.hs:837`, called at `Server.hs:825`). Cost
+scales with the largest dimensions and recurs every interval while Prometheus is on.
+
+### Fix
+
+Maintain the counts incrementally; drop the per-scrape full fold.
+
+---
+
+## Bug 10: subscription changes serialize through one thread and one map
+
+| | |
+| --- | --- |
+| Reachable | authenticated SUB |
+| Trigger | subscription churn, batched SUB of many queues |
+| Cost | throughput bounded by one thread and one contended TVar |
+| Evidence | code review |
+| Status | present |
+
+### How it works
+
+Subscription state lives in `queueSubscribers`, one `Map` in one TVar (`Env/STM.hs:377`). Changes are
+enqueued to `subQ` and applied by a single `serverThread` (`Server.hs:284`).
+
+### The bug
+
+All churn passes through that one thread and one TVar, and a batched SUB of N queues is N separate
+writes to `subQ` (`Server.hs:1845`), so subscription throughput does not scale with cores or clients.
+
+### Fix
+
+Shard the subscriber map, or batch `subQ` events per client.
+
+---
+
+## Bug 11: proxy never drops a stuck relay session
+
+| | |
+| --- | --- |
+| Reachable | any client, pre-auth via PRXY |
+| Trigger | relay stops answering forwards once the session is open |
+| Cost | forwards keep failing, no recovery until the ~20 min monitor drop |
+| Evidence | `testProxyForwardTimeoutStuckSession` (pending spec) |
+| Status | present |
+
+### How it works
+
+Shares Leak 1's mechanism. The forward path never calls `enablePings`, so a dead relay session is
+dropped only by `monitor` (`Client.hs:668`), which closes the client when `timeoutErrorCount >=
+smpPingCount` (3, `Client.hs:439`) and the last receive is older than `recoverWindow` (900s,
+`Client.hs:685`). `receive` (`Client.hs:663`) resets both counters on any inbound byte.
+
+### The bug
+
+A broken session is not dropped, so the proxy keeps forwarding to it and every forward fails. This is
+the reported symptom: `Error forwarding to relay ... PCEResponseTimeout` continued after the relay was
+restarted, because the proxy held the old session instead of reconnecting. Reproduced by
+`testProxyForwardTimeoutStuckSession` (`SMPProxyTests.hs:499`): 10 forwards on one timed-out session
+expecting a `PROXY NO_SESSION` drop, but all 10 return `BROKER TIMEOUT`. Unlike Leak 1 this is
+availability, not memory; the same persistent session is what makes Leak 1 unbounded.
+
+### Fix
+
+Drop the relay session after a bounded number of forward timeouts so the next forward returns
+`NO_SESSION` and the agent reconnects, instead of waiting out the 900s receive window. Enabling pings
+on the forward path would also let `monitor` detect the dead session.
+
+---
+
+## 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; one
+that stops above where it started is leaked.
+
+The peaks still matter. 200 abandoned half-open connections hold ~40 MiB for ~25s with no
+authentication. 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.

docs/smp-server-memory.md

@@ -0,0 +1,243 @@
+# SMP server memory at production scale
+
+Production: ~40k client connections per server, 32 GB RAM shared with PostgreSQL, pure PostgreSQL
+store, 14 servers proxying to each other, GHC 9.6.3, `+RTS -N -F1.2 -A16m -I0.01 -Iw15 -s -RTS`.
+With `-F1.2` the copying GC needs about 2.2x the live heap plus the nursery, so the server's live
+heap has to stay well under ~10 GB.
+
+A busy connection (20 batch-subscribed queues, 2 created with NEW, message traffic) cost 237 KiB of
+server live heap, 9.0 GiB at 40k connections. With the changes below it costs 43 KiB, 1.6 GiB.
+
+Earlier findings on leaks and PostgreSQL load are in `leak-findings.md` and `smp-server-db-load.md`.
+
+## Method
+
+All numbers are server-only: the bench client runs in a child process, so the measuring process holds
+only the server. PostgreSQL store, production RTS flags, 16-core machine. Live heap is measured after
+two major GCs with finalizers allowed to run in between (`liveBytesMiB` in `bench/MemBench.hs`).
+
+| Phase | Measures |
+| --- | --- |
+| `prodmix` | 2000 connections, each SUBs 20 queues in one block and creates 2 with NEW, then 30 s of SEND/MSG/ACK traffic |
+| `conns N` | N idle connections after the SMP handshake |
+| `subslice N` | per subscription: `SUBMODE=sub` (`SUBBATCH`, `SUBKEEP`), `new`, `newthensub` |
+| `load` | server CPU per operation under a fixed mixed workload |
+| `ntfloop`, `ntfdeliver` | notification store and its delivery loop |
+| `mesh` | proxy and relay in separate processes: relay sessions, forwards in flight, forward rate (`MESH=`) |
+| `pfwdbig` | oversized forwards left in the proxy's `sentCommands` |
+
+```sh
+cabal build -fserver_postgres exe:smp-mem-bench
+BENCHID=1 PROD_CONNS=2000 PROD_QUEUES=20 PROD_NEW=2 PROD_SEC=30 \
+  $(cabal list-bin -fserver_postgres exe:smp-mem-bench) prodmix 0 \
+  +RTS -N -F1.2 -A16m -I0.01 -Iw15 -T -ki2k -RTS
+```
+
+`BENCHID` offsets ports and the database, so runs do not collide with each other or with tests.
+
+## Results
+
+`prodmix`, live heap per connection after subscribing and peak memory in use under traffic. Each row
+adds to the previous one.
+
+| Change | KiB/conn | Peak in use | 40k conns |
+| --- | --- | --- | --- |
+| base (`sh/fix-leak`) | 236.6 | 1075 MiB | 9.0 GiB |
+| `-ki2k` | 111.5 | 719 MiB | 4.3 GiB |
+| connection commits (`sh/fix-conn-mem`) | 89.8 | 635 MiB | 3.4 GiB |
+| unpinned subscription keys, proxy commits (`sh/mem-combined`) | 80.0 | 604-622 MiB | 3.1 GiB |
+| patched tls 1.9 | 42.9 | 469 MiB | 1.6 GiB |
+
+Idle connections (`conns 2000`): 153 KiB at default flags, 68.4 KiB on `sh/mem-combined` with
+`-ki2k`, 34.8 KiB with the patched tls.
+
+Message throughput did not drop in any row. `-ki2k` costs some GC time (below).
+
+## Findings
+
+### 1. Thread stacks: 125 KiB per busy connection
+
+Each connection has several threads that block in shallow loops. A thread starts with a 1 KiB stack
+(`-ki1k`). On first overflow the RTS copies up to `-kb` (1 KiB) of frames into a new 32 KiB chunk
+(`-kc32k`), which with a 1 KiB initial stack moves the loop frame itself, so the thread never returns
+to its first chunk and keeps the 32 KiB chunk while the connection lives.
+
+| Flags | idle KiB/conn | prodmix KiB/conn |
+| --- | --- | --- |
+| default | 153 | 236.6 |
+| `-kc8k` | 100 | 135.0 |
+| `-ki2k` | 91 | 111.5 |
+
+`-ki2k` leaves the loop frame in the first chunk, so the overflow chunk is freed on return. Cost, 3
+alternating 60 s `load` runs: GC CPU +15% (4.8-5.2 s to 5.7-5.8 s per run), major GCs +43% (the live
+heap is smaller, so `-F1.2` triggers major GCs sooner), CPU per operation +0-5% (2.74-3.02 ms to
+2.91-3.35 ms, one noisy run). One `load` run with `-kc2k` grew to 6.8 GB RSS; not investigated, so
+`-kc` is left at its default.
+
+Branch `sh/rts-ki2k` sets `-with-rtsopts=-ki2k` for smp-server. Command-line `+RTS` flags still apply.
+
+### 2. Pinned blocks kept alive by small long-lived objects
+
+`ByteString`s and crypton's `ScrubbedBytes` are pinned and share 4 KiB blocks. GHC treats a pinned
+block as live if any object in it is live, and treats every object in a live pinned block as alive,
+including dead `ScrubbedBytes`, whose weak pointers and finalizers then stay allocated (GHC 9.6.3
+`Evac.c:443`, `GCAux.c:73`). A 24-byte value created during request processing, among crypto
+temporaries, therefore keeps ~4 KiB and several dead keys alive for as long as it lives. A standalone
+program reproduces it (pinned key 2742 B live per key, unpinned 80 B).
+
+Subscriptions. The subscription key is such a value.
+
+| `subslice 5000` | before | unpinned keys |
+| --- | --- | --- |
+| NEW-created | 5.25 KiB | 0.46 KiB |
+| one SUB per block | 1.67-2.42 KiB | 0.47 KiB |
+| 100 SUBs per block | 0.49 KiB | 0.46 KiB |
+| NEW0, then SUB on the same connection | 4.98 KiB | 0.47 KiB |
+
+Branch `sh/fix-sub-mem` stores the keys of `subscriptions`, `ntfSubscriptions` and `queueSubscribers`
+as `ShortByteString` and shares one key between the maps. CPU unchanged.
+
+TLS session state. tls 1.9 keeps record keys, IVs, TLS 1.3 traffic secrets, verify data and
+handshake state for the connection's lifetime, all allocated during the handshake. A heap census per
+connection before and after re-allocating them together at the end of the handshake:
+
+| | tls 1.9 | patched |
+| --- | --- | --- |
+| weak pointers | ~138 | ~45 |
+| memory in pinned blocks, outside the census | ~39 KiB | ~13 KiB |
+| idle KiB/conn | 68.4 | 34.8 |
+| prodmix KiB/conn | 80.0 | 42.9 |
+
+The patch (`tls-1.9-compact-state.patch`: `Network.TLS.Handshake.Compact`, called at the end of
+`handshake` under the context locks) re-derives the TLS 1.3 record keys from the kept traffic
+secrets, copies the IVs, secrets, session ID, verify data, randoms, main and resumption secrets and
+the transcript hash, and reseeds the context RNG. It needs a tls fork. Not compacted: the TLS 1.2 bulk
+key, keys installed later by KeyUpdate. The transcript hash copy coerces crypton's hidden `Context`
+newtype, so it must be checked on crypton upgrades.
+
+Full test suite with the patched tls: 1431 examples, 0 failures. The same run on unpatched tls had 1
+failure, the XFTP agent's "should resume sending file after restart".
+
+SMP handshake state (session secret, chain keys) has the same pattern and is not done yet; the
+remaining ~13 KiB outside the census is likely there.
+
+### 3. IDs sliced from the received block: 16 KiB per subscription
+
+Parsed `corrId` and `entityId` are slices of the ~16 KB decrypted block, and the entity ID was stored
+as the subscription key, so one live key kept the block. One SUB per block cost 16.43 KiB per
+subscription, one survivor of 100 cost 18.04 KiB. `sh/fix-sub-keys` copies both IDs in
+`tDecodeServer` (16.43 to 1.67-2.42 KiB). Unpinned keys (finding 2) also remove it for the maps.
+
+### 4. Connection structure: 21 KiB per busy connection
+
+- tls 1.9 stores the receive record state lazily, so each connection kept its last received 16 KB
+  record. `recvTLS` forces the state after each read. Not needed with tls 2.x.
+- The send and message-send loops are one thread, and one server-wide thread expires inactive clients
+  instead of one per client: 6 threads per connection become 4.
+
+`sh/fix-conn-mem`: `prodmix` 111.5 to 89.8 KiB with `-ki2k`, 236.6 to 210.5 KiB without.
+
+### 5. Notification store and its delivery loop
+
+Keys of notifiers whose notifications were delivered stayed in the store, and the loop in
+`deliverNtfsThread` sends every key to `getQueueNtfServices` (one `notifier_id IN ?` query read in
+full) every 1.5 s.
+
+| `ntfloop` (keys, no traffic) | allocation | CPU | peak in use |
+| --- | --- | --- | --- |
+| 0 | 0 MiB/s | 0 | 280 MiB |
+| 100k | 168 MiB/s | 0.42 cores | 530 MiB |
+| 300k | 263 MiB/s | 0.76 cores | 913 MiB |
+
+`ntfdeliver` with 20k delivered notifications: 20000 keys kept, idle server allocating 36 MiB/s at
+0.18 cores; with `sh/fix-ntf-store` 0 keys, 0.2 MiB/s, 0.01 cores.
+
+### 6. Proxy mesh
+
+The relay processed forwarded commands from a proxy one at a time in the connection's command loop,
+with a PostgreSQL round trip each; all users of a proxy share its connection. Forwards the relay has
+not reached wait on the proxy, each with a thread and its ~16 KB block. `mesh`, 1000 forwards per
+second for 20 s, proxy and relay in separate processes, 2 runs each:
+
+| | serial | concurrent |
+| --- | --- | --- |
+| forwards ok / failed | 19,968 / 0 | 19,968 / 0 |
+| proxy live heap at end of sending | 192-200 MiB | 13 MiB |
+| proxy memory in use | 653-672 MiB | 325 MiB (idle baseline) |
+| proxy threads | 4,026-4,137 | 420 |
+
+Here the backlog drained within the 30 s forward timeout. With a slower relay (a busy database, a
+remote relay over SOCKS) the queueing delay exceeds it and forwards fail; not reproduced.
+
+`sh/fix-proxy-mesh` forks each forwarded command under the per-connection concurrency limit, stops
+keeping the forwarded block on the proxy while waiting (53.0 to 36.8 KiB per in-flight forward,
+relay not answering) and caps forwards in flight per relay at 512 (`[PROXY] relay_concurrency`).
+Forwarded commands use per-command nonces and secrets and are matched by correlation ID, so concurrent
+processing is safe; the SimpleX agent keeps one message in flight per queue. The spec requires
+responses in order per queue within a connection (`simplex-messaging.md`, "same order within each
+queue ID"); serializing forwarded commands per queue ID is not done yet.
+
+Also in the proxy path (`sh/fix-proxy-leak`): a PFWD of 16260-16266 bytes from a client that declares
+`proxyServer = True` failed with `TELargeMsg` before sending and left its 20.2 KiB request in
+`sentCommands` for the life of the relay session (2001 PFWDs, 2001 entries); timed-out requests were
+kept too; and a relay session that stopped answering was never dropped (10 of 10 forwards timed out).
+
+### 7. Concurrency limit and name resolution
+
+`forkCmd` released its slot when the thread was forked, not when the command finished, so
+`serverClientConcurrency` and `serverResolverConcurrency` limited nothing: 16 RSLVs with a limit of 4
+all reached the resolver, 64 RSLVs from one connection were 64 concurrent requests, and 5 lookups
+answered with 502 opened 5 connections because error bodies were not read. `sh/fix-rslv-fanout` holds
+the slot until completion, adds a global limit (`[NAMES] resolver_global_concurrency`, 32) and reads
+error bodies. With the limit working, a client that hits it blocks its own command loop.
+
+### 8. Smaller findings
+
+- Active-queue statistics are `IntSet`s at 64 B per element (15M elements, 916 MiB for one set), six
+  of them, reset only when `log_stats` is on.
+- Control port `save` closes the PostgreSQL pool while the server keeps running; every later DB
+  operation blocks (`cpsave` phase: NEW after `save` gets no response).
+
+### 9. tls 2.x
+
+tls 2.1.6 (with tls-session-manager 0.0.6, crypton-connection 0.4.3, http-client-tls 0.3.6.4, an
+index-state bump and version pins) saves 2-5 KiB per connection: `prodmix` 75.4 KiB against 80.0.
+Problems found:
+
+- After a failed handshake a tls 2 client and a tls 2 server wait for each other in `bye`; this hung
+  `testServerMultipleIdentities`. Fixed on `sh/tls2` by closing the context without `bye`.
+- A tls 2 client that receives a CertificateRequest runs a timed read inside `handshake`; when the
+  timeout fires in the middle of a record the stream fails ("bad record mac") or hangs. SMP servers
+  always request client certificates. 1-2 failures per 2000 concurrent connections; present in tls
+  2.4.3. A `conns 2000` run with tls 2 on both sides stopped at 53 connections.
+
+tls 2.x is not recommended; `sh/tls2` is kept for reference.
+
+## Open
+
+- Serialize forwarded commands per queue ID on the relay (spec response order).
+- A client that sends PFWDs and never reads the responses costs the proxy 6.3 MiB live (11.4 MiB in
+  use) until the inactive-client expiry (up to 6 hours), 6 GiB per 1000 such clients.
+- PRXY to many aliases of one relay opens an unbounded number of relay sessions, each 207 KiB on the
+  proxy and 149 KiB on the relay: 2.0 GiB and 1.4 GiB per 10k.
+- Compact the SMP handshake state after the handshake, as in finding 2.
+- Publish the tls fork and reference it from `cabal.project`.
+- A SEND right after NEW or SUB can find no subscriber yet (`queueSubscribers` is updated through
+  `subQ`), so the message waits for the next SUB or ACK. `deliverIfSame` may leave a subscription
+  without a delivery thread. Both seen as occasional stuck steps in `load`; not memory.
+- `[PROXY] relay_concurrency` and `[NAMES] resolver_global_concurrency` are new INI keys.
+
+## Branches
+
+| Branch | Base | Content |
+| --- | --- | --- |
+| `sh/fix-ntf-store` | master | finding 5 |
+| `sh/fix-sub-keys` | master | finding 3 |
+| `sh/fix-proxy-leak` | master | finding 6, request leak and stuck session |
+| `sh/fix-rslv-fanout` | master | finding 7 |
+| `sh/fix-conn-mem` | master | finding 4 |
+| `sh/fix-sub-mem` | master | finding 2, subscription keys |
+| `sh/fix-proxy-mesh` | `sh/fix-proxy-leak` | finding 6, relay concurrency, per-relay limit |
+| `sh/rts-ki2k` | master | finding 1 |
+| `sh/mem-combined` | `sh/fix-leak` | findings 1-6 together, as measured in Results (local) |
+| `sh/tls2` | `sh/mem-combined` | finding 9, not for merge (local) |

docs/tls-1.9-compact-state.patch

@@ -0,0 +1,184 @@
+diff -ruN -x 'dist*' tls-1.9.0-orig/Network/TLS/Crypto.hs tls-fork/Network/TLS/Crypto.hs
+--- tls-1.9.0-orig/Network/TLS/Crypto.hs	2026-09-30 18:44:40.083085887 +0000
++++ tls-fork/Network/TLS/Crypto.hs	2026-09-30 18:44:40.106519923 +0000
+@@ -8,6 +8,7 @@
+     , hashUpdate
+     , hashUpdateSSL
+     , hashFinal
++    , hashCopy
+ 
+     , module Network.TLS.Crypto.DH
+     , module Network.TLS.Crypto.IES
+@@ -44,6 +45,8 @@
+ import qualified Crypto.Hash as H
+ import qualified Data.ByteString as B
+ import qualified Data.ByteArray as B (convert)
++import qualified Data.ByteArray as BA
++import Unsafe.Coerce (unsafeCoerce)
+ import Crypto.Error
+ import Crypto.Number.Basic (numBits)
+ import Crypto.Random
+@@ -135,6 +138,17 @@
+ hashUpdate (HashContextSSL sha1Ctx md5Ctx) b =
+     HashContextSSL (H.hashUpdate sha1Ctx b) (H.hashUpdate md5Ctx b)
+ 
++-- | Copies the context into a new allocation.
++hashCopy :: HashContext -> IO HashContext
++hashCopy (HashContext (ContextSimple h)) = HashContext . ContextSimple <$> copyContext h
++hashCopy (HashContextSSL sha1Ctx md5Ctx) = HashContextSSL <$> copyContext sha1Ctx <*> copyContext md5Ctx
++
++-- H.Context is a newtype over Bytes, with the constructor in a hidden crypton module
++copyContext :: H.Context alg -> IO (H.Context alg)
++copyContext h = do
++    b <- BA.copy h (\_ -> return ()) :: IO BA.Bytes
++    return $! unsafeCoerce b
++
+ hashUpdateSSL :: HashCtx
+               -> (B.ByteString,B.ByteString) -- ^ (for the md5 context, for the sha1 context)
+               -> HashCtx
+diff -ruN -x 'dist*' tls-1.9.0-orig/Network/TLS/Handshake/Compact.hs tls-fork/Network/TLS/Handshake/Compact.hs
+--- tls-1.9.0-orig/Network/TLS/Handshake/Compact.hs	1970-01-01 00:00:00.000000000 +0000
++++ tls-fork/Network/TLS/Handshake/Compact.hs	2026-09-30 18:45:10.137984909 +0000
+@@ -0,0 +1,111 @@
++{-# LANGUAGE OverloadedStrings #-}
++{-# OPTIONS_HADDOCK hide #-}
++module Network.TLS.Handshake.Compact
++    ( compactContext
++    ) where
++
++import Control.Concurrent.MVar
++import Control.Exception (evaluate)
++import Crypto.Random (drgNew)
++import qualified Data.ByteString as B
++import Data.IORef
++
++import Network.TLS.Cipher
++import Network.TLS.Context.Internal
++import Network.TLS.Crypto
++import Network.TLS.Extension (Cookie (..))
++import Network.TLS.Handshake.State
++import Network.TLS.Imports
++import Network.TLS.KeySchedule (hkdfExpandLabel)
++import Network.TLS.Record.State
++import Network.TLS.RNG
++import Network.TLS.State
++import Network.TLS.Struct
++import Network.TLS.Types
++
++-- | Re-allocates the state a context keeps after the handshake, next to each other.
++-- Each small pinned value created during the handshake shares a 4K block with short-lived
++-- crypto temporaries, and keeps the whole block alive, with every dead ScrubbedBytes in it.
++compactContext :: Context -> IO ()
++compactContext ctx = do
++    tx <- readMVar $ ctxTxState ctx
++    rx <- readMVar $ ctxRxState ctx
++    st <- readMVar $ ctxState ctx
++    hst <- readMVar $ ctxHandshake ctx
++    finished <- readIORef $ ctxFinished ctx
++    peerFinished <- readIORef $ ctxPeerFinished ctx
++    -- key derivation and reseeding allocate temporaries, so they precede the copies
++    txKey <- trafficKey tx
++    rxKey <- trafficKey rx
++    rng <- evaluate $ StateRNG $ fst $ withTLSRNG (stRandomGen st) drgNew
++    tx' <- copyRecordState BulkEncrypt txKey tx
++    rx' <- copyRecordState BulkDecrypt rxKey rx
++    st' <- copyTLSState rng st
++    hst' <- traverse copyHandshakeState hst
++    finished' <- traverse copyBytes finished
++    peerFinished' <- traverse copyBytes peerFinished
++    modifyMVar_ (ctxTxState ctx) $ \_ -> return tx'
++    modifyMVar_ (ctxRxState ctx) $ \_ -> return rx'
++    modifyMVar_ (ctxState ctx) $ \_ -> return st'
++    modifyMVar_ (ctxHandshake ctx) $ \_ -> return hst'
++    writeIORef (ctxFinished ctx) finished'
++    writeIORef (ctxPeerFinished ctx) peerFinished'
++
++copyBytes :: ByteString -> IO ByteString
++copyBytes = evaluate . B.copy
++
++-- TLS 1.3 record keys are derived from the traffic secret kept in the record state;
++-- earlier versions keep the bulk key only inside the cipher state.
++trafficKey :: RecordState -> IO (Maybe ByteString)
++trafficKey RecordState{stCipher = Just cipher, stCryptLevel = CryptApplicationSecret, stCryptState = cst} =
++    Just <$> evaluate (hkdfExpandLabel (cipherHash cipher) (cstMacSecret cst) "key" "" (bulkKeySize $ cipherBulk cipher))
++trafficKey _ = return Nothing
++
++copyRecordState :: BulkDirection -> Maybe ByteString -> RecordState -> IO RecordState
++copyRecordState dir key_ rs@RecordState{stCryptState = cst} = do
++    iv <- copyBytes $ cstIV cst
++    secret <- copyBytes $ cstMacSecret cst
++    bulkState <- case (key_, stCipher rs) of
++        (Just key, Just cipher) -> bulkInit (cipherBulk cipher) dir <$> copyBytes key
++        _ -> return $ cstKey cst
++    evaluate rs{stCryptState = CryptState{cstKey = bulkState, cstIV = iv, cstMacSecret = secret}}
++
++copyTLSState :: StateRNG -> TLSState -> IO TLSState
++copyTLSState rng st = do
++    session <- case stSession st of
++        Session sid -> Session <$> traverse copyBytes sid
++    clientVerified <- copyBytes $ stClientVerifiedData st
++    serverVerified <- copyBytes $ stServerVerifiedData st
++    proto <- traverse copyBytes $ stNegotiatedProtocol st
++    alpn <- traverse (traverse copyBytes) $ stClientALPNSuggest st
++    cookie <- traverse (\(Cookie c) -> Cookie <$> copyBytes c) $ stTLS13Cookie st
++    exporter <- traverse copyBytes $ stExporterMasterSecret st
++    evaluate
++        st
++            { stSession = session
++            , stClientVerifiedData = clientVerified
++            , stServerVerifiedData = serverVerified
++            , stNegotiatedProtocol = proto
++            , stClientALPNSuggest = alpn
++            , stTLS13Cookie = cookie
++            , stExporterMasterSecret = exporter
++            , stRandomGen = rng
++            }
++
++copyHandshakeState :: HandshakeState -> IO HandshakeState
++copyHandshakeState hst = do
++    clientRandom <- ClientRandom <$> copyBytes (unClientRandom $ hstClientRandom hst)
++    serverRandom <- traverse (fmap ServerRandom . copyBytes . unServerRandom) $ hstServerRandom hst
++    mainSecret <- traverse copyBytes $ hstMasterSecret hst
++    digest <- case hstHandshakeDigest hst of
++        HandshakeMessages msgs -> HandshakeMessages <$> traverse copyBytes msgs
++        HandshakeDigestContext h -> HandshakeDigestContext <$> hashCopy h
++    resumption <- traverse (\(BaseSecret s) -> BaseSecret <$> copyBytes s) $ hstTLS13ResumptionSecret hst
++    evaluate
++        hst
++            { hstClientRandom = clientRandom
++            , hstServerRandom = serverRandom
++            , hstMasterSecret = mainSecret
++            , hstHandshakeDigest = digest
++            , hstTLS13ResumptionSecret = resumption
++            }
+diff -ruN -x 'dist*' tls-1.9.0-orig/Network/TLS/Handshake.hs tls-fork/Network/TLS/Handshake.hs
+--- tls-1.9.0-orig/Network/TLS/Handshake.hs	2026-09-30 18:44:40.083878116 +0000
++++ tls-fork/Network/TLS/Handshake.hs	2026-09-30 18:45:14.390780619 +0000
+@@ -18,6 +18,7 @@
+ import Network.TLS.Struct
+ 
+ import Network.TLS.Handshake.Common
++import Network.TLS.Handshake.Compact
+ import Network.TLS.Handshake.Client
+ import Network.TLS.Handshake.Server
+ 
+@@ -27,7 +28,7 @@
+ -- This is to be called at the beginning of a connection, and during renegotiation
+ handshake :: MonadIO m => Context -> m ()
+ handshake ctx =
+-    liftIO $ withRWLock ctx $ handleException ctx (ctxDoHandshake ctx ctx)
++    liftIO $ withRWLock ctx $ handleException ctx (ctxDoHandshake ctx ctx) >> compactContext ctx
+ 
+ -- Handshake when requested by the remote end
+ -- This is called automatically by 'recvData', in a context where the read lock
+diff -ruN -x 'dist*' tls-1.9.0-orig/tls.cabal tls-fork/tls.cabal
+--- tls-1.9.0-orig/tls.cabal	2026-09-30 18:44:40.090442445 +0000
++++ tls-fork/tls.cabal	2026-09-30 19:12:16.099341583 +0000
+@@ -85,6 +85,7 @@
+         Network.TLS.Handshake.Signature
+         Network.TLS.Handshake.State
+         Network.TLS.Handshake.State13
++        Network.TLS.Handshake.Compact
+         Network.TLS.Hooks
+         Network.TLS.IO
+         Network.TLS.Imports

protocol/diagrams/xrcp/session.mmd

@@ -17,11 +17,11 @@ note over CC, HC: TLS handshake
 par
     note over CC: validate client X509 credentials
     CC->>CI: session code from tlsUnique
-    CI-->>CC: user confirmation
+    CI-->>CC: user confirmation (optional if known)
 and
     note over HC: validate server X509 credentials
     HC->>HI: session code from tlsUnique
-    HI-->>HC: user confirmation
+    HI-->>HC: user confirmation (optional if known)
 end
 
 note over CI, HI: 3. Session verification and protocol negotiation

protocol/security.md

@@ -168,7 +168,7 @@ This document describes the cryptographic primitives and threat model for the Si
 
 *cannot:*
 
-- impersonate a sender and send messages to the user whose database was stolen. Doing so requires also compromising the router (to place the message in the queue, that is possible until the Double-Ratchet advances forward) or the user's device at a subsequent time (to place the message in the database).
+- impersonate a sender and send messages to the user whose database was stolen, unless the database contains a new rotation queue that is not secured yet. Doing so requires also compromising the router (to place the message in the queue, that is possible until the Double-Ratchet advances forward) or the user's device at a subsequent time (to place the message in the database).
 
 - undetectably communicate at the same time as Alice with her contacts. Doing so would result in the contact getting different messages with repeated IDs.
 

protocol/simplex-messaging.md

@@ -1120,6 +1120,8 @@ When the client receives PKEY response it MUST validate that:
 - the fingerprint of the received certificate matches fingerprint in the router address - it mitigates MITM attack by proxy.
 - the router session key is correctly signed with the received certificate.
 
+`smpVersionRange` in `PKEY` is not signed.
+
 The proxy router may respond with error response in case the destination router is not available or in case it has an earlier version that does not support proxied commands.
 
 #### Send command via proxy

protocol/xrcp.md

@@ -121,9 +121,15 @@ During TLS handshake, parties validate certificate chains against previously kno
 
 ### Session verification and protocol negotiation
 
-Once TLS session is established, both the host and controller devices present a "session security code" to the user who must match them (e.g., visually or via QR code scan) and confirm on the host device. The session security code must be a digest of tlsunique channel binding. As it is computed as a digest of the TLS handshake for both the controller and the host, it will validate that the same TLS certificates are used on both sides, and that the same TLS session is established, mitigating the possibility of MITM attack in the connection.
+Once TLS session is established, both the host and controller devices present a "session security code" to the user. The session security code must be a digest of tlsunique channel binding. As it is computed as a digest of the TLS handshake for both the controller and the host, it will validate that the same TLS certificates are used on both sides, and that the same TLS session is established, mitigating the possibility of MITM attack in the connection.
 
-Once the session is confirmed by the user, the host sends HELLO block to the controller.
+To verify the session, the user matches the codes (e.g., visually or via QR code scan) and confirms the session on the host device.
+
+The user MUST verify the first session with a new controller. The host stores the long-term identity of the controller (CA fingerprint and Ed25519 public key) only after this session is verified. A controller with the stored long-term identity is a known controller.
+
+The user MAY verify the subsequent sessions with a known controller. In a session that the host confirms without user action, the controller is authenticated only by its long-term identity stored in the first session (see [Threat model](#threat-model)). SimpleX Chat mobile apps require user verification of sessions with known controllers only when the user enables the "Verify connections" option.
+
+Once the session is confirmed, the host sends HELLO block to the controller.
 
 XRCP blocks inside TLS are padded to 16384 bytes.
 
@@ -305,7 +311,7 @@ to receive: (rcvKey', sk, nonce) = HKDF(rcvKey, "SimpleXSbChain", 88)
 - prevent host and controller devices from establishing the session
 
 *cannot:*
-- same as passive adversary, provided that user visually verified session code out-of-band.
+- same as passive adversary, provided that user visually verified session code out-of-band in the first session with the controller.
 
 #### An active adversary with the access to the network:
 
@@ -324,6 +330,15 @@ to receive: (rcvKey', sk, nonce) = HKDF(rcvKey, "SimpleXSbChain", 88)
 *cannot:*
 - connect to the host or make host connect to itself.
 
+#### An active adversary with the access to the network who also obtained the long-term private keys of a known controller (e.g., from the controller device):
+
+*can:*
+- make host connect to itself instead of the controller.
+- access any data of the controlled host application, within the capabilities of the provided API, if the host confirms the session without user action.
+
+*cannot:*
+- access host application data, provided that user visually verified session code out-of-band.
+
 #### Compromised controller device:
 
 *can:*

simplexmq.cabal

@@ -659,3 +659,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/FileTransfer/Server.hs

@@ -25,7 +25,6 @@ import Control.Monad.Reader
 import Control.Monad.Trans.Except
 import qualified Data.Aeson as J
 import Data.Bifunctor (first)
-import qualified Data.ByteString.Base64.URL as B64
 import Data.ByteString.Builder (Builder, byteString)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
@@ -594,7 +593,7 @@ processXFTPRequest ent HTTP2Body {bodyPart} = \case
               \used -> let used' = used + fromIntegral size in if used' <= quota then (True, used') else (False, used)
           receive = do
             path <- asks $ filesPath . config
-            let fPath = path </> B.unpack (B64.encode $ unEntityId senderId)
+            let fPath = path </> serverFileName senderId
             receiveChunk (XFTPRcvChunkSpec fPath size digest) >>= \case
               Right () -> do
                 stats <- asks serverStats

src/Simplex/FileTransfer/Server/StoreLog.hs

@@ -17,6 +17,7 @@ module Simplex.FileTransfer.Server.StoreLog
     logDeleteFile,
     logBlockFile,
     logAckFile,
+    serverFileName,
   )
 where
 
@@ -24,6 +25,7 @@ import Control.Applicative (optional, (<|>))
 import Control.Concurrent.STM
 import Control.Monad.Except
 import qualified Data.Attoparsec.ByteString.Char8 as A
+import qualified Data.ByteString.Base64.URL as B64
 import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy.Char8 as LB
 import Data.Composition ((.:), (.::.))
@@ -35,10 +37,11 @@ import Simplex.FileTransfer.Protocol (FileInfo (..))
 import Simplex.FileTransfer.Server.Store
 import Simplex.FileTransfer.Transport (XFTPErrorType (..))
 import Simplex.Messaging.Encoding.String
-import Simplex.Messaging.Protocol (BlockingInfo, RcvPublicAuthKey, RecipientId, SenderId)
+import Simplex.Messaging.Protocol (BlockingInfo, EntityId (..), RcvPublicAuthKey, RecipientId, SenderId)
 import Simplex.Messaging.Server.QueueStore (ServerEntityStatus (..))
 import Simplex.Messaging.Server.StoreLog
 import Simplex.Messaging.Util (bshow)
+import System.FilePath (takeFileName)
 import System.IO
 
 data FileStoreLogRecord
@@ -99,6 +102,9 @@ logBlockFile s fId = logFileStoreRecord s . BlockFile fId
 logAckFile :: StoreLog 'WriteMode -> RecipientId -> IO ()
 logAckFile s = logFileStoreRecord s . AckFile
 
+serverFileName :: SenderId -> FilePath
+serverFileName = B.unpack . B64.encode . unEntityId
+
 readWriteFileStore :: FilePath -> STMFileStore -> IO (StoreLog 'WriteMode)
 readWriteFileStore = readWriteStoreLog readFileStore writeFileStore
 
@@ -115,7 +121,9 @@ readFileStore f st = mapM_ (addFileLogRecord . LB.toStrict) . LB.lines =<< LB.re
       AddFile sId file createdAt expiresAt status
         | size file > 0 -> addFile st sId file createdAt expiresAt status
         | otherwise -> pure $ Left SIZE
-      PutFile qId path -> setFilePath st qId path
+      PutFile qId path
+        | takeFileName path == serverFileName qId -> setFilePath st qId path
+        | otherwise -> pure $ Left AUTH
       AddRecipients sId rcps -> runExceptT $ addRecipients sId rcps
       DeleteFile sId -> deleteFile st sId
       BlockFile sId info -> blockFile st sId info True

src/Simplex/Messaging/Agent.hs

@@ -4388,8 +4388,9 @@ agentRatchetDecrypt' :: TVar ChaChaDRG -> DB.Connection -> ConnId -> CR.RatchetX
 agentRatchetDecrypt' g db connId rc encAgentMsg = do
   skipped <- liftIO $ getSkippedMsgKeys db connId
   (agentMsgBody_, rc', skippedDiff) <- withExceptT (SEAgentError . cryptoError) $ CR.rcDecrypt g rc skipped encAgentMsg
+  agentMsgBody <- liftEither $ first (SEAgentError . cryptoError) agentMsgBody_
   liftIO $ updateRatchet db connId rc' skippedDiff
-  liftEither $ bimap (SEAgentError . cryptoError) (,CR.rcRcvKEM rc') agentMsgBody_
+  pure (agentMsgBody, CR.rcRcvKEM rc')
 
 newSndQueue :: UserId -> ConnId -> Compatible SMPQueueInfo -> Maybe (C.APrivateAuthKey) -> AM' (NewSndQueue, C.PublicKeyX25519)
 newSndQueue userId connId (Compatible (SMPQueueInfo smpClientVersion SMPQueueAddress {smpServer, senderId, queueMode, dhPublicKey = rcvE2ePubDhKey})) sndKey_ = do

src/Simplex/Messaging/Client.hs

@@ -35,6 +35,8 @@ module Simplex.Messaging.Client
     ProxiedRelay (..),
     getProtocolClient,
     closeProtocolClient,
+    closeTimedOutClient,
+    pClientSentCommandsCount,
     protocolClientServer,
     protocolClientServer',
     transportHost',
@@ -150,7 +152,8 @@ import Data.Int (Int64)
 import Data.List (find, isSuffixOf)
 import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.List.NonEmpty as L
-import Data.Maybe (catMaybes, fromMaybe)
+import qualified Data.Map.Strict as M
+import Data.Maybe (catMaybes, fromMaybe, isNothing)
 import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Time.Clock (UTCTime (..), diffUTCTime, getCurrentTime)
@@ -658,10 +661,9 @@ getProtocolClient g nm transportSession@(_, srv, _) cfg@ProtocolClientConfig {qS
               responseErr = atomically . putTMVar responseVar . Left . PCETransportError
 
     receive :: Transport c => ProtocolClient v err msg -> THandle v c 'TClient -> IO ()
-    receive ProtocolClient {client_ = PClient {rcvQ, lastReceived, timeoutErrorCount}} h = forever $ do
+    receive ProtocolClient {client_ = PClient {rcvQ, lastReceived}} h = forever $ do
       tGetClient h >>= atomically . writeTBQueue rcvQ
       getCurrentTime >>= atomically . writeTVar lastReceived
-      atomically $ writeTVar timeoutErrorCount 0
 
     monitor :: ProtocolClient v err msg -> IO ()
     monitor c@ProtocolClient {client_ = PClient {sendPings, lastReceived, timeoutErrorCount}} = loop smpPingInterval
@@ -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
@@ -750,6 +756,12 @@ closeProtocolClient :: ProtocolClient v err msg -> IO ()
 closeProtocolClient = mapM_ (deRefWeak >=> mapM_ killThread) . action
 {-# INLINE closeProtocolClient #-}
 
+-- | Disconnects client when maxCnt commands in a row timed out, 0 to disable.
+closeTimedOutClient :: Int -> ProtocolClient v err msg -> IO ()
+closeTimedOutClient maxCnt c@ProtocolClient {client_ = PClient {timeoutErrorCount}} = do
+  cnt <- readTVarIO timeoutErrorCount
+  when (maxCnt > 0 && cnt >= maxCnt) $ closeProtocolClient c
+
 -- | SMP client error type.
 data ProtocolClientError err
   = -- | Correctly parsed SMP server ERR response.
@@ -1354,20 +1366,22 @@ sendProtocolCommand c nm = sendProtocolCommand_ c nm Nothing Nothing
 --
 -- Please note: if nonce is passed it is also used as a correlation ID
 sendProtocolCommand_ :: forall v err msg. Protocol v err msg => ProtocolClient v err msg -> NetworkRequestMode -> Maybe C.CbNonce -> Maybe Int -> Maybe C.APrivateAuthKey -> EntityId -> ProtoCommand msg -> ExceptT (ProtocolClientError err) IO msg
-sendProtocolCommand_ c@ProtocolClient {client_ = PClient {sndQ}, thParams = THandleParams {blockSize, serviceAuth}} nm nonce_ tOut pKey entId cmd =
+sendProtocolCommand_ c@ProtocolClient {client_ = PClient {sndQ, sentCommands}, thParams = THandleParams {blockSize, serviceAuth}} nm nonce_ tOut pKey entId cmd =
   ExceptT $ uncurry sendRecv =<< mkTransmission_ c nonce_ (entId, pKey, cmd)
   where
     -- two separate "atomically" needed to avoid blocking
     sendRecv :: Either TransportError SentRawTransmission -> Request err msg -> IO (Either (ProtocolClientError err) msg)
-    sendRecv t_ r = case t_ of
-      Left e -> pure . Left $ PCETransportError e
+    sendRecv t_ r@Request {corrId} = case t_ of
+      Left e -> notSent e
       Right t
-        | B.length s > blockSize - 2 -> pure . Left $ PCETransportError TELargeMsg
+        | B.length s > blockSize - 2 -> notSent TELargeMsg
         | otherwise -> do
             nonBlockingWriteTBQueue sndQ (Just r, s)
             response <$> getResponse c nm tOut r
         where
           s = tEncodeBatch1 serviceAuth t
+      where
+        notSent e = Left (PCETransportError e) <$ atomically (TM.delete corrId sentCommands)
 
 nonBlockingWriteTBQueue :: TBQueue a -> a -> IO ()
 nonBlockingWriteTBQueue q x = do
@@ -1375,7 +1389,7 @@ nonBlockingWriteTBQueue q x = do
   unless sent $ void $ forkIO $ atomically $ writeTBQueue q x
 
 getResponse :: ProtocolClient v err msg -> NetworkRequestMode -> Maybe Int -> Request err msg -> IO (Response err msg)
-getResponse ProtocolClient {client_ = PClient {tcpTimeout, timeoutErrorCount}} nm tOut Request {entityId, pending, responseVar} = do
+getResponse ProtocolClient {client_ = PClient {tcpTimeout, timeoutErrorCount, sentCommands, msgQ}} nm tOut Request {corrId, entityId, pending, responseVar} = do
   r <- fromMaybe (netTimeoutInt tcpTimeout nm) tOut `timeout` atomically (takeTMVar responseVar)
   response <- atomically $ do
     writeTVar pending False
@@ -1384,7 +1398,11 @@ getResponse ProtocolClient {client_ = PClient {tcpTimeout, timeoutErrorCount}} n
     -- See `processMsg`.
     ((r <|>) <$> tryTakeTMVar responseVar) >>= \case
       Just r' -> writeTVar timeoutErrorCount 0 $> r'
-      Nothing -> modifyTVar' timeoutErrorCount (+ 1) $> Left PCEResponseTimeout
+      Nothing -> do
+        modifyTVar' timeoutErrorCount (+ 1)
+        -- a late response is delivered to msgQ, without msgQ it is only logged
+        when (isNothing msgQ) $ TM.delete corrId sentCommands
+        pure $ Left PCEResponseTimeout
   pure Response {entityId, response}
 
 mkTransmission :: Protocol v err msg => ProtocolClient v err msg ->  ClientCommand msg -> IO (PCTransmission err msg)

src/Simplex/Messaging/Client/Agent.hs

@@ -31,6 +31,8 @@ module Simplex.Messaging.Client.Agent
     removeActiveSubs,
     removePendingSub,
     removePendingSubs,
+    AgentLeakStats (..),
+    getAgentLeakStats,
   )
 where
 
@@ -158,6 +160,35 @@ 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 <$> maybe (pure 0) (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

src/Simplex/Messaging/Protocol.hs

@@ -241,6 +241,7 @@ import Data.Attoparsec.ByteString.Char8 (Parser, (<?>))
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.Bifunctor (bimap, first)
 import Data.Bits (xor)
+import qualified Data.ByteArray as BA
 import qualified Data.ByteString as BS
 import qualified Data.ByteString.Base64 as B64
 import Data.ByteString.Char8 (ByteString)
@@ -1323,7 +1324,10 @@ instance ProtocolTypeI p => FromJSON (ProtocolServer p) where
   parseJSON = strParseJSON "ProtocolServer"
 
 newtype BasicAuth = BasicAuth {unBasicAuth :: ByteString}
-  deriving (Eq, Ord, Show)
+  deriving (Ord, Show)
+
+instance Eq BasicAuth where
+  BasicAuth s == BasicAuth s' = BA.constEq s s'
 
 instance IsString BasicAuth where fromString = BasicAuth . B.pack
 
@@ -2407,7 +2411,9 @@ tDecodeServer THandleParams {sessionId, thVersion = v, implySessId} = \case
           where
             cmdOrErr = parseProtocol @v @err @cmd v command >>= checkCredentials tAuth entityId
             t :: a -> (CorrId, EntityId, a)
-            t = (corrId,entityId,)
+            -- IDs are slices of the ~16 KB received block and are kept as subscription keys,
+            -- so without a copy one live key retains the whole block
+            t = (CorrId $ B.copy $ bs corrId,EntityId $ B.copy $ unEntityId entityId,)
         Left _ -> tError corrId PEBlock
     | otherwise -> tError corrId PESession
   Left _ -> tError "" PEBlock

src/Simplex/Messaging/Server.hs

@@ -76,7 +76,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,8 +98,8 @@ import Network.Socket (ServiceName, Socket, socketToHandle)
 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 (NetworkConfig (..), ProtocolClient (thParams), ProtocolClientConfig (..), ProtocolClientError (..), SMPClient, SMPClientError, clientHandlers, closeTimedOutClient, forwardSMPTransmission, smpProxyError, temporaryClientError, transportHost')
+import Simplex.Messaging.Client.Agent (AgentLeakStats (..), OwnServer, SMPClientAgent (..), SMPClientAgentConfig (..), 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
@@ -369,7 +388,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
         runDeliverNtfs ms ns' stats
       where
         runDeliverNtfs :: s -> NtfStore -> ServerStats -> IO ()
-        runDeliverNtfs ms (NtfStore ns) stats = do
+        runDeliverNtfs ms ns'@(NtfStore ns) stats = do
           ntfs <- M.assocs <$> readTVarIO ns
           unless (null ntfs) $
             getQueueNtfServices @(StoreQueue s) (queueStore ms) ntfs >>= \case
@@ -381,6 +400,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
                     cIds <- IS.toList <$> readTVarIO subClients
                     forM_ cIds $ \cId -> getServerClient cId srv >>= mapM_ (deliverQueueNtfs ntfs')
                 atomically $ modifyTVar' ns (`M.withoutKeys` S.fromList (map fst deleted))
+                deleteEmptyNtfs ns' ntfs
           where
             deliverQueueNtfs ntfs' c@Client {ntfSubscriptions} =
               whenM (currentClient readTVarIO c) $ do
@@ -496,6 +516,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
@@ -1092,6 +1207,7 @@ controlPortAuth h user admin role auth = do
   readTVarIO role >>= \case
     CPRNone -> do
       atomically $ writeTVar role $! newRole
+      when (newRole == CPRNone) $ logWarn "ControlPort: failed auth"
       hPutStrLn h $ currentRole newRole
     r -> hPutStrLn h $ currentRole r <> if r == newRole then "" else ", start new session to change."
   where
@@ -1464,7 +1580,12 @@ client
                 Right r -> PRES r <$ inc own pSuccesses
                 Left e -> ERR (smpProxyError e) <$ case e of
                   PCEProtocolError {} -> inc own pSuccesses
-                  _ -> inc own pErrorsOther
+                  _ -> do
+                    logWarn $ "Error forwarding to relay: " <> decodeLatin1 (strEncode $ transportHost' smp) <> " own=" <> tshow own <> " " <> tshow e
+                    inc own pErrorsOther
+                    case e of
+                      PCEResponseTimeout -> liftIO $ closeTimedOutClient (smpPingCount $ networkConfig $ smpCfg $ agentCfg a) smp
+                      _ -> pure ()
           Nothing -> inc False pRequests >> inc False pErrorsConnect $> Just (ERR $ PROXY NO_SESSION)
       where
         forkProxiedCmd :: M s BrokerMsg -> M s (Maybe BrokerMsg)
@@ -1472,11 +1593,15 @@ client
     -- Run a slow command on a thread
     forkCmd :: (ServerConfig s -> Int) -> CorrId -> EntityId -> M s BrokerMsg -> M s (Maybe a)
     forkCmd concurrency corrId entId cmdAction = do
-      bracket_ wait signal . forkClient clnt (B.unpack $ "client $" <> encode sessionId <> " cmd") $
-        -- commands MUST be processed under a reasonable timeout or the client would halt
-        cmdAction >>= \t -> atomically $ writeTBQueue sndQ ([(corrId, entId, t)], [])
+      -- the forked thread releases the slot when the command completes, the caller only if the fork failed
+      mask $ \restore -> do
+        wait
+        forkClient clnt (B.unpack $ "client $" <> encode sessionId <> " cmd") (restore cmd `finally` signal)
+          `onException` signal
       pure Nothing
       where
+        -- commands MUST be processed under a reasonable timeout or the client would halt
+        cmd = cmdAction >>= \t -> atomically $ writeTBQueue sndQ ([(corrId, entId, t)], [])
         wait = do
           limit <- asks (concurrency . config)
           atomically $ do
@@ -1492,7 +1617,7 @@ client
         Nothing -> incStat (rslvDisabled st) $> Nothing
         Just nenv -> pure (Just nenv)
     -- Runs on a forked thread so RSLV does not block other commands;
-    -- concurrency is limited by serverResolverConcurrency in forkCmd.
+    -- concurrency is limited per connection by serverResolverConcurrency in forkCmd and globally in resolveName.
     resolveNameMsg :: VersionSMP -> NamesEnv -> NameQuery -> M s BrokerMsg
     resolveNameMsg v nenv q = do
       st <- asks (rslvStats . serverStats)
@@ -2127,6 +2252,7 @@ client
           let proxyNonce = C.cbNonce $ bs corrId
           s' <- liftEitherWith (const CRYPTO) $ C.cbDecryptNoPad sessSecret proxyNonce s
           FwdTransmission {fwdCorrId, fwdVersion, fwdKey, fwdTransmission = EncTransmission et} <- liftEitherWith (const $ CMD SYNTAX) $ smpDecode s'
+          unless (fwdVersion `isCompatible` thServerVRange thParams') $ throwE $ transportErr TEVersion
           let clientSecret = C.dh' fwdKey serverPrivKey
               clientNonce = C.cbNonce $ bs fwdCorrId
           b <- liftEitherWith (const CRYPTO) $ C.cbDecrypt clientSecret clientNonce et

src/Simplex/Messaging/Server/Main.hs

@@ -814,7 +814,8 @@ readNamesConfig ini
               { resolverEndpoint = either (error . ("[NAMES] resolver_endpoint: " <>)) id (validateUrl endpoint resolverAuth_),
                 resolverAuth = resolverAuth_,
                 resolverTimeoutMs = boundedIniInt 3000 100 60000 "resolver_timeout_ms",
-                resolverMaxResponseBytes = boundedIniInt 16000 1024 16000 "resolver_max_response_bytes"
+                resolverMaxResponseBytes = boundedIniInt 16000 1024 16000 "resolver_max_response_bytes",
+                resolverGlobalConcurrency = boundedIniInt 32 1 1000 "resolver_global_concurrency"
               }
   where
     enabled = fromMaybe False (iniOnOff "NAMES" "enable" ini)

src/Simplex/Messaging/Server/Main/Init.hs

@@ -167,6 +167,8 @@ iniFileContent cfgPath logPath opts host basicAuth controlPortPwds =
         \# resolver_auth: basic <username>:<password>\n\
         \# resolver_timeout_ms: 3000\n\
         \# resolver_max_response_bytes: 16000\n\
+        \# Max concurrent requests to the resolver from all connections.\n\
+        \# resolver_global_concurrency: 32\n\
         \# Max concurrent name resolutions per connection (forwarded RSLVs from many\n\
         \# clients share one proxy connection, so this is much higher than PROXY client_concurrency).\n"
     <> ("# resolver_concurrency = " <> tshow defaultNameResolverConcurrency)

src/Simplex/Messaging/Server/Names.hs

@@ -15,6 +15,7 @@ module Simplex.Messaging.Server.Names
   )
 where
 
+import Control.Concurrent.QSem (QSem, newQSem, signalQSem, waitQSem)
 import qualified Control.Exception as E
 import Control.Logger.Simple (logError)
 import Data.Bifunctor (first)
@@ -38,19 +39,22 @@ data NamesConfig = NamesConfig
   { resolverEndpoint :: String,
     resolverAuth :: Maybe RpcAuth,
     resolverTimeoutMs :: Int,
-    resolverMaxResponseBytes :: Int
+    resolverMaxResponseBytes :: Int,
+    resolverGlobalConcurrency :: Int
   }
   deriving (Show)
 
 data NamesEnv = NamesEnv
   { config :: NamesConfig,
-    resolverEnv :: ResolverEnv
+    resolverEnv :: ResolverEnv,
+    resolverSlots :: QSem
   }
 
 newNamesEnv :: NamesConfig -> IO NamesEnv
 newNamesEnv config = do
-  resolverEnv <- newResolverEnv (resolverEndpoint config) (resolverAuth config) (resolverTimeoutMs config) (resolverMaxResponseBytes config)
-  pure NamesEnv {config, resolverEnv}
+  resolverEnv <- newResolverEnv (resolverEndpoint config) (resolverAuth config) (resolverTimeoutMs config) (resolverMaxResponseBytes config) (resolverGlobalConcurrency config)
+  resolverSlots <- newQSem (resolverGlobalConcurrency config)
+  pure NamesEnv {config, resolverEnv, resolverSlots}
 
 closeNamesEnv :: NamesEnv -> IO ()
 closeNamesEnv NamesEnv {resolverEnv} = closeResolverEnv resolverEnv
@@ -60,8 +64,9 @@ pingEndpoint NamesEnv {resolverEnv, config} =
   fromMaybe (Left ResolverTimeout) <$> timeout (resolverTimeoutMs config * 1000) (healthHttp resolverEnv)
 
 resolveName :: NamesEnv -> NameQuery -> IO (Either NameErrorType NameResponse)
-resolveName env q = do
-  r <- E.try (timeout (resolverTimeoutMs (config env) * 1000) (fetch env q))
+resolveName env@NamesEnv {resolverSlots} q = do
+  -- waiting for a slot counts towards the timeout, so lookups do not queue longer than resolverTimeoutMs
+  r <- E.try (timeout (resolverTimeoutMs (config env) * 1000) (E.bracket_ (waitQSem resolverSlots) (signalQSem resolverSlots) (fetch env q)))
   case r of
     Right result -> pure (fromMaybe (Left (RESOLVER "timeout")) result)
     Left e

src/Simplex/Messaging/Server/Names/HttpResolver.hs

@@ -85,9 +85,9 @@ data ResolverError
   | ResolverTimeout
   deriving (Show)
 
-newResolverEnv :: String -> Maybe RpcAuth -> Int -> Int -> IO ResolverEnv
-newResolverEnv baseUrl auth_ timeoutMs maxResponseBytes = do
-  manager <- HC.newManager tlsManagerSettings {managerConnCount = 10}
+newResolverEnv :: String -> Maybe RpcAuth -> Int -> Int -> Int -> IO ResolverEnv
+newResolverEnv baseUrl auth_ timeoutMs maxResponseBytes maxConcurrency = do
+  manager <- HC.newManager tlsManagerSettings {managerConnCount = maxConcurrency}
   pure
     ResolverEnv
       { manager,
@@ -136,8 +136,9 @@ httpGet ResolverEnv {manager, baseUrl, authHdr, timeoutMicro, maxResponseBytes}
           }
   result <- E.try $ withResponse req manager $ \res -> do
     let status = HT.statusCode (responseStatus res)
+    -- http-client closes the connection unless the body is read to the end
     if status >= 400
-      then pure (Left (HttpStatusErr status))
+      then Left (HttpStatusErr status) <$ brReadSome (responseBody res) (maxResponseBytes + 1)
       else do
         bs <- brReadSome (responseBody res) (maxResponseBytes + 1)
         pure $ if BL.length bs > fromIntegral maxResponseBytes then Left BodyTooLarge else Right bs

src/Simplex/Messaging/Server/NtfStore.hs

@@ -10,6 +10,7 @@ module Simplex.Messaging.Server.NtfStore
     storeNtf,
     deleteNtfs,
     deleteExpiredNtfs,
+    deleteEmptyNtfs,
   ) where
 
 import Control.Concurrent.STM
@@ -22,6 +23,7 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (EncNMsgMeta, MsgId, NotifierId)
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
+import Simplex.Messaging.Util (whenM)
 
 newtype NtfStore = NtfStore (TMap NotifierId (TVar [MsgNtf]))
 
@@ -33,13 +35,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 +47,25 @@ 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
+
+deleteEmptyNtfs :: NtfStore -> [(NotifierId, TVar [MsgNtf])] -> IO ()
+deleteEmptyNtfs (NtfStore ns) = mapM_ deleteEmpty
+  where
+    deleteEmpty (nId, v) =
+      whenM (null <$> readTVarIO v) $
+        atomically $
+          TM.lookup nId ns >>= mapM_ (\v' -> whenM (null <$> readTVar v') $ TM.delete nId ns)
 
 data NtfLogRecord = NLRv1 NotifierId MsgNtf
 

src/Simplex/Messaging/Transport.hs

@@ -865,7 +865,6 @@ forceCertChain :: CertChainPubKey -> CertChainPubKey
 forceCertChain cert@(CertChainPubKey (X.CertificateChain cc) signedKey) = length (show cc) `seq` show signedKey `seq` cert
 {-# INLINE forceCertChain #-}
 
--- This function is only used with v >= 8, so currently it's a simple record update.
 -- * Note: it requires updating version-based parameters, to be consistent with smpTHandle_.
 smpTHParamsSetVersion :: VersionSMP -> THandleParams SMPVersion p -> THandleParams SMPVersion p
 smpTHParamsSetVersion v params =

tests/CoreTests/BatchingTests.hs

@@ -14,11 +14,13 @@ import Control.Monad
 import Crypto.Random (ChaChaDRG)
 import qualified Data.ByteString as B
 import Data.ByteString.Char8 (ByteString)
+import Data.ByteString.Unsafe (unsafeUseAsCString)
 import qualified Data.List.NonEmpty as L
 import Data.Time.Clock.System (SystemTime, getSystemTime)
 import qualified Data.X509 as X
 import qualified Data.X509.CertificateStore as XS
 import qualified Data.X509.File as XF
+import Foreign.Ptr (plusPtr)
 import Simplex.Messaging.Client
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
@@ -40,6 +42,8 @@ batchingTests = do
     it "should batch subscription responses with message" testBatchSubResponses
     it "should break on message that does not fit" testClientBatchWithMessage
     it "should break on large message" testClientBatchWithLargeMessage
+  describe "tDecodeServer" $
+    it "should copy IDs out of received block" testDecodeServerCopiesIds
 
 testBatchSubscriptions :: IO ()
 testBatchSubscriptions = do
@@ -166,6 +170,26 @@ testClientBatchWithLargeMessage = do
   (length rs1', length rs2') `shouldBe` (75, 135)
   all lenOk [s1', s2'] `shouldBe` True
 
+testDecodeServerCopiesIds :: IO ()
+testDecodeServerCopiesIds = do
+  sessId <- atomically . C.randomBytes 32 =<< C.newRandom
+  subs <- replicateM 2 $ randomSUB sessId
+  let thParams = testTHandleParams sessId
+  [TBTransmissions s 2 _] <- pure $ batchTransmissions thParams $ L.fromList subs
+  forM_ (tParse thParams s) $ \t -> do
+    let (corrId, entId, _) = tDecodeClient @SMPVersion @ErrorType @Cmd thParams t
+    Right (_, _, (corrId', entId', _)) <- pure $ tDecodeServer @SMPVersion @ErrorType @Cmd (testTHandleParams sessId) t
+    (corrId', entId') `shouldBe` (corrId, entId)
+    sharesBuffer s (bs corrId) `shouldReturn` True
+    sharesBuffer s (unEntityId entId) `shouldReturn` True
+    sharesBuffer s (bs corrId') `shouldReturn` False
+    sharesBuffer s (unEntityId entId') `shouldReturn` False
+
+sharesBuffer :: ByteString -> ByteString -> IO Bool
+sharesBuffer block s =
+  unsafeUseAsCString block $ \blockPtr -> unsafeUseAsCString s $ \ptr ->
+    pure $ ptr >= blockPtr && ptr < blockPtr `plusPtr` B.length block
+
 testClientStub :: IO (ProtocolClient SMPVersion ErrorType BrokerMsg)
 testClientStub = do
   g <- C.newRandom
@@ -237,7 +261,7 @@ randomSEND sessId len = do
       TransmissionForAuth {tForAuth, tToSend} = encodeTransmissionForAuth thParams (CorrId corrId, EntityId sId, Cmd SSender $ SEND noMsgFlags msg)
   pure $ (,tToSend) <$> authTransmission thAuth_ False (Just spKey) nonce tForAuth
 
-testTHandleParams :: ByteString -> THandleParams SMPVersion 'TClient
+testTHandleParams :: ByteString -> THandleParams SMPVersion p
 testTHandleParams sessionId =
   THandleParams
     { sessionId,

tests/CoreTests/MsgStoreTests.hs

@@ -28,22 +28,25 @@ import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Int (Int64)
 import Data.List (isPrefixOf, isSuffixOf)
+import qualified Data.Map.Strict as M
 import Data.Maybe (fromJust)
 import Data.Time.Clock (addUTCTime)
 import Data.Time.Clock.System (SystemTime (..), getSystemTime)
 import SMPClient (testStoreLogFile, testStoreMsgsDir, testStoreMsgsDir2, testStoreMsgsFile, testStoreMsgsFile2)
 import Simplex.Messaging.Crypto (pattern MaxLenBS)
 import qualified Simplex.Messaging.Crypto as C
-import Simplex.Messaging.Protocol (EntityId (..), ErrorType, LinkId, Message (..), QueueLinkData, RecipientId, SParty (..), noMsgFlags)
+import Simplex.Messaging.Protocol (EntityId (..), ErrorType, LinkId, Message (..), NotifierId, QueueLinkData, RecipientId, SParty (..), noMsgFlags)
 import Simplex.Messaging.Server (exportMessages, importMessages, printMessageStats)
 import Simplex.Messaging.Server.Env.STM (MsgStore (..), journalMsgStoreDepth, readWriteQueueStore)
 import Simplex.Messaging.Server.Expiration (ExpirationConfig (..), expireBeforeEpoch)
 import Simplex.Messaging.Server.MsgStore.Journal
 import Simplex.Messaging.Server.MsgStore.STM
 import Simplex.Messaging.Server.MsgStore.Types
+import Simplex.Messaging.Server.NtfStore
 import Simplex.Messaging.Server.QueueStore
 import Simplex.Messaging.Server.QueueStore.QueueInfo
 import Simplex.Messaging.Server.StoreLog (closeStoreLog, logCreateQueue)
+import qualified Simplex.Messaging.TMap as TM
 import System.Directory (copyFile, createDirectoryIfMissing, listDirectory, removeFile, renameFile)
 import System.FilePath ((</>))
 import System.IO (IOMode (..), withFile)
@@ -83,6 +86,9 @@ msgStoreTests = do
   describe "Journal message store: queue state backup expiration" $ do
     it "should remove old queue state backups" testRemoveQueueStateBackups
     it "should expire messages in idle queues" testExpireIdleQueues
+  describe "Notification store" $ do
+    it "should remove keys of expired notifications" testExpireNtfs
+    it "should remove keys of delivered notifications" testDeleteEmptyNtfs
   where
     journalMsgStoreTests :: SpecWith (JournalMsgStore s)
     journalMsgStoreTests = do
@@ -611,6 +617,54 @@ testExpireIdleQueues = do
   (Nothing, False) <- readQueueState ms statePath
   pure ()
 
+mkMsgNtf :: TVar ChaChaDRG -> Int64 -> IO MsgNtf
+mkMsgNtf g ts = do
+  ntfMsgId <- atomically $ C.randomBytes 24 g
+  ntfNonce <- atomically $ C.randomCbNonce g
+  pure MsgNtf {ntfMsgId, ntfTs = MkSystemTime ts 0, ntfNonce, ntfEncMeta = "meta"}
+
+ntfCounts :: NtfStore -> IO [(NotifierId, Int)]
+ntfCounts (NtfStore ns) = M.assocs <$> (mapM (fmap length . readTVarIO) =<< readTVarIO ns)
+
+testExpireNtfs :: IO ()
+testExpireNtfs = do
+  g <- C.newRandom
+  st@(NtfStore ns) <- NtfStore <$> TM.emptyIO
+  now <- systemSeconds <$> getSystemTime
+  let old = now - 100
+      nId1 = EntityId "notifier 1"
+      nId2 = EntityId "notifier 2"
+      nId3 = EntityId "notifier 3"
+  storeNtf st nId1 =<< mkMsgNtf g old
+  storeNtf st nId2 =<< mkMsgNtf g old
+  storeNtf st nId2 =<< mkMsgNtf g now
+  atomically $ TM.insertM nId3 (newTVar []) ns
+  ntfCounts st `shouldReturn` [(nId1, 1), (nId2, 2), (nId3, 0)]
+  deleteExpiredNtfs st (now - 50) `shouldReturn` 2
+  ntfCounts st `shouldReturn` [(nId2, 1)]
+  storeNtf st nId1 =<< mkMsgNtf g now
+  deleteExpiredNtfs st (now - 50) `shouldReturn` 0
+  ntfCounts st `shouldReturn` [(nId1, 1), (nId2, 1)]
+
+testDeleteEmptyNtfs :: IO ()
+testDeleteEmptyNtfs = do
+  g <- C.newRandom
+  st@(NtfStore ns) <- NtfStore <$> TM.emptyIO
+  now <- systemSeconds <$> getSystemTime
+  let nId1 = EntityId "notifier 1"
+      nId2 = EntityId "notifier 2"
+      nId3 = EntityId "notifier 3"
+      nId4 = EntityId "notifier 4"
+  forM_ ([nId1, nId2, nId3, nId4] :: [NotifierId]) $ \nId -> storeNtf st nId =<< mkMsgNtf g now
+  ntfs <- M.assocs <$> readTVarIO ns
+  let flush nId = forM_ (lookup nId ntfs) $ \v -> atomically $ writeTVar v []
+  mapM_ flush ([nId1, nId2, nId4] :: [NotifierId])
+  storeNtf st nId2 =<< mkMsgNtf g now
+  deleteNtfs st nId4 `shouldReturn` 0
+  storeNtf st nId4 =<< mkMsgNtf g now
+  deleteEmptyNtfs st ntfs
+  ntfCounts st `shouldReturn` [(nId2, 1), (nId3, 1), (nId4, 1)]
+
 testReadFileMissing :: JournalMsgStore s -> IO ()
 testReadFileMissing ms = do
   g <- C.newRandom

tests/CoreTests/StoreLogTests.hs

@@ -22,7 +22,7 @@ import qualified Data.X509.Validation as XV
 import SMPClient
 import Simplex.FileTransfer.Protocol (FileInfo (..))
 import Simplex.FileTransfer.Server.Store (FileRec (..), FileRecipient (..), FileStoreClass (..), RoundedFileTime, STMFileStore (..))
-import Simplex.FileTransfer.Server.StoreLog (FileStoreLogRecord (..), readWriteFileStore)
+import Simplex.FileTransfer.Server.StoreLog (FileStoreLogRecord (..), readWriteFileStore, serverFileName)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol
@@ -37,6 +37,7 @@ import Simplex.Messaging.Server.StoreLog
 import Simplex.Messaging.SystemTime
 import Simplex.Messaging.Transport (SMPServiceRole (..))
 import Simplex.Messaging.Transport.Credentials (genCredentials)
+import System.FilePath ((</>))
 import Test.Hspec hiding (fit, it)
 import Util
 
@@ -269,6 +270,18 @@ fileStoreLogTests = do
           saved = [AddFile sId file createdAt Nothing EntityActive, BlockFile sId blockedWithNotice],
           compacted = [AddFile sId file createdAt Nothing (EntityBlocked blockedWithNotice)],
           state = M.fromList [(sId, (file, createdAt, Nothing, EntityBlocked blockedWithNotice))]
+        },
+      SLTC
+        { name = "put file with wrong file name",
+          saved = [AddFile sId file createdAt (Just expiresAt) EntityActive, PutFile sId "/etc/passwd"],
+          compacted = [AddFile sId file createdAt (Just expiresAt) EntityActive],
+          state = M.fromList [(sId, (file, createdAt, Just expiresAt, EntityActive))]
+        },
+      SLTC
+        { name = "put file in another directory",
+          saved = [AddFile sId file createdAt (Just expiresAt) EntityActive, PutFile sId ("/srv/xftp" </> serverFileName sId)],
+          compacted = [AddFile sId file createdAt (Just expiresAt) EntityActive, PutFile sId ("/srv/xftp" </> serverFileName sId)],
+          state = M.fromList [(sId, (file, createdAt, Just expiresAt, EntityActive))]
         }
     ]
 

tests/CoreTests/XFTPStoreTests.hs

@@ -9,7 +9,7 @@ import Data.Word (Word32)
 import Simplex.FileTransfer.Protocol (FileInfo (..), SFileParty (..))
 import Simplex.FileTransfer.Server.Store
 import Simplex.FileTransfer.Server.Store.Postgres (PostgresFileStore, importFileStore, exportFileStore)
-import Simplex.FileTransfer.Server.StoreLog (closeStoreLog, readWriteFileStore, writeFileStore)
+import Simplex.FileTransfer.Server.StoreLog (closeStoreLog, readWriteFileStore, serverFileName, writeFileStore)
 import Simplex.FileTransfer.Transport (XFTPErrorType (..))
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Protocol (BlockingInfo (..), BlockingReason (..), EntityId (..))
@@ -17,6 +17,7 @@ import Simplex.Messaging.Server.QueueStore (ServerEntityStatus (..))
 import Simplex.Messaging.Server.StoreLog (openWriteStoreLog)
 import Simplex.Messaging.SystemTime (RoundedSystemTime (..))
 import System.Directory (doesFileExist, removeFile)
+import System.FilePath ((</>))
 import Test.Hspec hiding (fit, it)
 import UnliftIO.STM
 import Util
@@ -265,7 +266,7 @@ testMigrationRoundTrip = do
       sId2 = EntityId "migration_file_2"
       rId1 = EntityId "migration_rcp_1_"
   addFile stmStore sId1 fileInfo1 testCreatedAt (Just testExpiresAt) EntityActive `shouldReturn` Right ()
-  void $ setFilePath stmStore sId1 "/tmp/file1"
+  void $ setFilePath stmStore sId1 ("/tmp" </> serverFileName sId1)
   addRecipient stmStore sId1 (FileRecipient rId1 rcpKey1) `shouldReturn` Right ()
   let testBlockInfo = BlockingInfo {reason = BRSpam, notice = Nothing}
   addFile stmStore sId2 fileInfo2 testCreatedAt Nothing (EntityBlocked testBlockInfo) `shouldReturn` Right ()
@@ -289,7 +290,7 @@ testMigrationRoundTrip = do
   case result1 of
     Right (FileRec {fileInfo = fi, filePath, expiresAt, fileStatus}, _) -> do
       size fi `shouldBe` 128000
-      readTVarIO filePath `shouldReturn` Just "/tmp/file1"
+      readTVarIO filePath `shouldReturn` Just ("/tmp" </> serverFileName sId1)
       expiresAt `shouldBe` Just testExpiresAt
       readTVarIO fileStatus `shouldReturn` EntityActive
     Left e -> expectationFailure $ "getFile sId1 failed: " <> show e

tests/NamesResolverServer.hs

@@ -9,6 +9,7 @@
 module NamesResolverServer
   ( withResolverServer,
     withResolverServerDelayed,
+    withResolverServerConns,
     resolveResp,
     testNamesConfig,
     memCfg,
@@ -36,9 +37,18 @@ withResolverServer :: ([Text] -> (Status, LB.ByteString)) -> (Int -> IORef [[Tex
 withResolverServer = withResolverServerDelayed 0
 
 withResolverServerDelayed :: Int -> ([Text] -> (Status, LB.ByteString)) -> (Int -> IORef [[Text]] -> IO a) -> IO a
-withResolverServerDelayed delayMs handler action = do
+withResolverServerDelayed delayMs handler action = withResolverServer_ delayMs handler $ \port reqs _ -> action port reqs
+
+-- | Also counts the TCP connections the resolver accepted.
+withResolverServerConns :: ([Text] -> (Status, LB.ByteString)) -> (Int -> IORef [[Text]] -> IORef Int -> IO a) -> IO a
+withResolverServerConns = withResolverServer_ 0
+
+withResolverServer_ :: Int -> ([Text] -> (Status, LB.ByteString)) -> (Int -> IORef [[Text]] -> IORef Int -> IO a) -> IO a
+withResolverServer_ delayMs handler action = do
   reqs <- newIORef []
-  Warp.withApplication (pure (app reqs)) $ \port -> action port reqs
+  conns <- newIORef 0
+  let settings = Warp.setOnOpen (\_ -> True <$ atomicModifyIORef' conns (\n -> (n + 1, ()))) Warp.defaultSettings
+  Warp.withApplicationSettings settings (pure (app reqs)) $ \port -> action port reqs conns
   where
     app :: IORef [[Text]] -> Application
     app reqs req send = do
@@ -61,7 +71,8 @@ testNamesConfig port =
     { resolverEndpoint = "http://127.0.0.1:" <> show port,
       resolverAuth = Nothing,
       resolverTimeoutMs = 1000,
-      resolverMaxResponseBytes = 65536
+      resolverMaxResponseBytes = 65536,
+      resolverGlobalConcurrency = 8
     }
 
 memCfg :: AServerConfig

tests/RSLVTests.hs

@@ -11,11 +11,14 @@
 
 module RSLVTests (rslvTests) where
 
+import Control.Concurrent (threadDelay)
+import Control.Monad (forM_)
 import Control.Monad.Trans.Except (ExceptT, runExceptT)
 import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy as LB
 import Data.IORef (IORef, readIORef)
 import Data.List.NonEmpty (NonEmpty (..))
+import qualified Data.List.NonEmpty as L
 import Data.Text (Text)
 import Data.Text.Encoding (encodeUtf8)
 import Data.Time.Clock (getCurrentTime)
@@ -47,6 +50,8 @@ import Simplex.Messaging.Protocol
     tPut,
   )
 import qualified Simplex.Messaging.Protocol as SMP
+import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
+import Simplex.Messaging.Server.Names (NamesConfig (..))
 import Simplex.Messaging.SimplexName (SimplexDomain)
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Version (mkVersionRange)
@@ -102,6 +107,9 @@ rslvTests = do
     it "RSLV sends the 2LD as its hash" testRslvSendsTheHash
     it "a name with subnames is sent as text" testSubnameKeepsItsLabels
     it "a record naming a different name is rejected" testRslvWrongName
+  describe "RSLV resource use" $ do
+    it "one connection has at most resolver_concurrency lookups in flight" testRslvConnectionCap
+    it "all connections have at most resolver_global_concurrency lookups in flight" testRslvFanOut
 
 -- | /v2/resolve answers 200, 400 or 502, so a 404 is a resolver that predates
 -- the route, not a name that does not exist.
@@ -275,5 +283,47 @@ testRslvWrongName =
       Left (PCEUnexpectedResponse _) -> pure ()
       _ -> expectationFailure $ "expected Left (PCEUnexpectedResponse ..), got: " <> show r
 
+-- The resolver answers after 3s and the server gives up after 1s, so a request
+-- that reached the resolver stays in flight for the whole check.
+testRslvConnectionCap :: IO ()
+testRslvConnectionCap =
+  NRS.withResolverServerDelayed 3000 (NRS.resolveResp status200 "{}") $ \port reqs ->
+    withSmpServerConfigOn (transport @TLS) (updateCfg (withNames port memCfg) $ \c -> c {serverResolverConcurrency = connCap}) testPort $ const $
+      testSMPClient @TLS $ \h -> do
+        sendRslvs h "cap" 16
+        threadDelay 800000
+        length <$> resolvePaths reqs `shouldReturn` connCap
+        recvResponses h 16 `shouldReturn` replicate 16 (Right (ERR (NAME (RESOLVER "timeout"))))
+  where
+    connCap = 4
+
+testRslvFanOut :: IO ()
+testRslvFanOut =
+  NRS.withResolverServerDelayed 3000 (NRS.resolveResp status200 "{}") $ \port reqs ->
+    withSmpServerConfigOn (transport @TLS) (withNames port memCfg) testPort $ const $
+      testSMPClient @TLS $ \h1 -> testSMPClient @TLS $ \h2 -> do
+        sendRslvs h1 "a" 32
+        sendRslvs h2 "b" 32
+        threadDelay 800000
+        length <$> resolvePaths reqs `shouldReturn` resolverGlobalConcurrency (NRS.testNamesConfig port)
+        let timedOut = replicate 32 (Right (ERR (NAME (RESOLVER "timeout"))))
+        recvResponses h1 32 `shouldReturn` timedOut
+        recvResponses h2 32 `shouldReturn` timedOut
+
+-- | One RSLV per block, so no batch limit applies.
+sendRslvs :: THandleSMP TLS 'TClient -> String -> Int -> IO ()
+sendRslvs h@THandle {params} prefix n =
+  forM_ [1 .. n] $ \i -> do
+    let TransmissionForAuth {tToSend} = encodeTransmissionForAuth params (CorrId (B.pack $ prefix <> show i), NoEntity, Cmd SResolver (RSLV (NQDomain (domain "alice.simplex"))))
+    [Right ()] <- tPut h (Right (Nothing, tToSend) :| [])
+    pure ()
+
+recvResponses :: THandleSMP TLS 'TClient -> Int -> IO [Either ErrorType BrokerMsg]
+recvResponses h n
+  | n <= 0 = pure []
+  | otherwise = do
+      rs <- map (\(_, _, r) -> r) . L.toList <$> tGetClient h
+      (rs <>) <$> recvResponses h (n - length rs)
+
 runExceptT' :: Show e => ExceptT e IO a -> IO a
 runExceptT' a = runExceptT a >>= either (fail . show) pure

tests/SMPClient.hs

@@ -316,6 +316,9 @@ prevRange vr = vr {maxVersion = max (minVersion vr) (prevVersion $ maxVersion vr
 prevVersion :: Version v -> Version v
 prevVersion (Version v) = Version (v - 1)
 
+nextVersion :: Version v -> Version v
+nextVersion (Version v) = Version (v + 1)
+
 proxyCfg :: AServerConfig
 proxyCfg = proxyCfgMS (ASType SQSMemory SMSJournal)
 
@@ -346,6 +349,14 @@ proxyCfgShortTimeout =
         nt = NetworkTimeout {backgroundTimeout = 4_000000, interactiveTimeout = 4_000000}
      in cfg' {smpAgentCfg = aCfg {smpCfg = cCfg {networkConfig = (networkConfig cCfg) {tcpConnectTimeout = nt}}}}
 
+proxyCfgForwardTimeout :: AServerConfig
+proxyCfgForwardTimeout =
+  updateCfg proxyCfg $ \cfg' ->
+    let aCfg = smpAgentCfg cfg'
+        cCfg = smpCfg aCfg
+        nt = NetworkTimeout {backgroundTimeout = 1, interactiveTimeout = 1}
+     in cfg' {smpAgentCfg = aCfg {smpCfg = cCfg {networkConfig = (networkConfig cCfg) {tcpTimeout = nt}}}}
+
 withSmpServerStoreMsgLogOn :: HasCallStack => (ASrvTransport, AStoreType) -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a
 withSmpServerStoreMsgLogOn (t, msType) =
   withSmpServerConfigOn t $ updateCfg (cfgMS msType) $ \cfg' -> cfg' {storeNtfsFile = Just testStoreNtfsFile, serverStatsBackupFile = Just testServerStatsBackupFile}

tests/SMPNamesTests.hs

@@ -5,6 +5,7 @@
 
 module SMPNamesTests (smpNamesTests, testNameRecord, testPricing, registeredBody, availableBody, reservedBody, responseBody, resolved) where
 
+import Control.Monad (forM_)
 import qualified Data.Aeson as J
 import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy as LB
@@ -15,7 +16,7 @@ import qualified Data.Map.Strict as M
 import qualified Data.Text as T
 import Data.Text.Encoding (encodeUtf8)
 import Network.HTTP.Types (status200, status400, status404, status500, status502)
-import NamesResolverServer (resolveResp, testNamesConfig, withResolverServer, withResolverServerDelayed)
+import NamesResolverServer (resolveResp, testNamesConfig, withResolverServer, withResolverServerConns, withResolverServerDelayed)
 import Simplex.Messaging.Encoding (smpDecode, smpEncode)
 import Simplex.Messaging.Encoding.String (strDecode)
 import Simplex.Messaging.Protocol (Command (..), ErrorType (..), NameErrorType (..), NamePricing (..), NameQuery (..), NameRecord (..), NameRegistration (..), NameResponse (..), NameReservedReason (..), ProtocolEncoding (..), USDCents (..))
@@ -302,6 +303,13 @@ resolverSpec = do
       _ <- resolveName env aliceDomain
       readIORef reqs >>= \rs -> length rs `shouldBe` 2
 
+  it "keeps the resolver connection alive across error responses" $
+    withResolverServerConns (resolveResp status502 "{\"error\":\"upstream\"}") $ \port reqs conns -> do
+      env <- newNamesEnv (testNamesConfig port)
+      forM_ [1 .. 5 :: Int] $ \_ -> resolveName env aliceDomain `shouldReturn` Left (RESOLVER "HTTP 502")
+      length <$> readIORef reqs `shouldReturn` 5
+      readIORef conns `shouldReturn` 1
+
   it "addresses the resolver with the full canonical domain name" $
     withResolverServer (resolveResp status200 (registeredBody testNameRecord)) $ \port reqs -> do
       env <- newNamesEnv (testNamesConfig port)

tests/SMPProxyTests.hs

@@ -21,6 +21,7 @@ import Control.Logger.Simple
 import Control.Monad (forM, forM_, forever, replicateM_)
 import Control.Monad.Trans.Except (ExceptT, runExceptT)
 import Data.ByteString.Char8 (ByteString)
+import qualified Data.ByteString.Char8 as B
 import Data.List.NonEmpty (NonEmpty)
 import qualified Data.List.NonEmpty as L
 import Data.Time.Clock (getCurrentTime)
@@ -63,6 +64,10 @@ smpProxyTests = do
       testProxyRecoversWithoutDisconnect
     it "reconnects to relay after sender disconnects mid-connection" $ \_ ->
       testProxyReconnectAfterRelayRestart
+    it "must drop a stuck relay session after forward timeouts" $ \_ ->
+      testProxyForwardTimeoutStuckSession
+    it "does not keep oversized forwarded command" $ \_ ->
+      testForwardOversizedNotKept
   describe "agent client reconnection" $ do
     it "reconnects after a connect is cancelled mid-flight" $ \_ ->
       testAgentClientReconnectAfterCancel
@@ -191,6 +196,8 @@ deliverMessagesViaProxy proxyServ relayServ alg unsecuredMsgs securedMsgs = do
   forM_ unsecuredMsgs $ \msg -> do
     runExceptT' (proxySMPMessage pc NRMInteractive sess Nothing sndId noMsgFlags msg) `shouldReturn` Right ()
     runExceptT' (proxySMPMessage pc NRMInteractive sess {prSessionId = "bad session"} Nothing sndId noMsgFlags msg) `shouldReturn` Left (ProxyProtocolError $ SMP.PROXY SMP.NO_SESSION)
+    forM_ ([prevVersion minServerSMPRelayVersion, nextVersion currentServerSMPRelayVersion] :: [VersionSMP]) $ \v ->
+      runExceptT' (proxySMPMessage pc NRMInteractive sess {prVersion = v} Nothing sndId noMsgFlags msg) `shouldReturn` Left (ProxyProtocolError $ SMP.PROXY $ SMP.PROTOCOL $ SMP.PROXY $ SMP.BROKER $ SMP.TRANSPORT TEVersion)
     -- receive 1
     (_tSess, _, [(_entId, STEvent (Right (SMP.MSG RcvMessage {msgId, msgBody = EncRcvMsgBody encBody})))]) <- atomically $ readTBQueue msgQ
     dec msgId encBody `shouldBe` Right msg
@@ -494,6 +501,32 @@ testProxyReconnectAfterRelayRestart =
       race_ (threadDelay 1000000) requestRelaySession
     requireProxyReconnect
 
+testProxyForwardTimeoutStuckSession :: IO ()
+testProxyForwardTimeoutStuckSession =
+  withSmpServerConfigOn (transport @TLS) proxyCfgForwardTimeout testPort $ \_ -> do
+    g <- C.newRandom
+    ts <- getCurrentTime
+    let srv = SMPServer testHost testPort testKeyHash
+        vr = mkVersionRange minServerSMPRelayVersion currentClientSMPRelayVersion
+    pc <- either (fail . show) pure =<< getProtocolClient g NRMInteractive (1, srv, Nothing) defaultSMPClientConfig {serverVRange = vr} [] Nothing ts (\_ -> pure ())
+    sess <- runExceptT' $ connectSMPProxiedRelay pc NRMInteractive srv (Just "correct")
+    sId <- atomically $ SMP.EntityId <$> C.randomBytes 24 g
+    rs <- forM ([1 .. 10] :: [Int]) $ \_ -> runExceptT' (proxySMPMessage pc NRMInteractive sess Nothing sId noMsgFlags "hi")
+    rs `shouldSatisfy` elem (Left (ProxyProtocolError (SMP.PROXY SMP.NO_SESSION)))
+
+testForwardOversizedNotKept :: IO ()
+testForwardOversizedNotKept =
+  withSmpServerConfigOn (transport @TLS) proxyCfg testPort $ \_ -> do
+    g <- C.newRandom
+    ts <- getCurrentTime
+    let proxyClientCfg = defaultSMPClientConfig {serverVRange = supportedProxyClientSMPRelayVRange, agreeSecret = True, proxyServer = True}
+    c <- either (fail . show) pure =<< getProtocolClient g NRMBackground (1, testSMPServer, Nothing) proxyClientCfg [] Nothing ts (\_ -> pure ())
+    (k, _) <- atomically $ C.generateKeyPair @'C.X25519 g
+    let et = SMP.EncTransmission $ B.replicate smpBlockSize 'a'
+    runExceptT (forwardSMPTransmission c (SMP.CorrId "123456789012345678901234") currentClientSMPRelayVersion k et)
+      `shouldReturn` Left (PCETransportError TELargeMsg)
+    pClientSentCommandsCount c `shouldReturn` 0
+
 -- Bug B (same root cause as the proxy, in the messaging agent): getSMPServerClient inserts an
 -- empty SessionVar into smpClients, then connects inside newProtocolClient's tryAllErrors, which
 -- rethrows async exceptions. If the connecting thread is cancelled mid-connect, putTMVar is