servers: usage statistics by command

2026-10-12 11:39:56 +00:00 · 2026-09-28 19:25:46 +00:00
43 changed files with 1509 additions and 340 deletions
@@ -0,0 +1,309 @@
+# Implementation plan: client address statistics
+
+Scope:
+
+- SMP server
+- XFTP server
+- NTF server
+
+## Rules
+
+- Counters per address are kept in memory only.
+- Counters are neither saved nor restored.
+- Prometheus receives aggregates across addresses only.
+- Addresses are not logged.
+- Addresses are shown only by a control port command.
+
+## Address key
+
+```
+data AddrKey
+  = AKIPv4 Word32
+  | AKIPv6 Word64
+```
+
+Key of a socket address:
+
+- IPv4 address: `AKIPv4`.
+- IPv6 address in `::ffff:0:0/96`: `AKIPv4` of its last 32 bits.
+- Other IPv6 address: `AKIPv6` of its first 64 bits.
+- Other address types: not counted.
+
+Text of a key:
+
+- `AKIPv4`: `a.b.c.d`.
+- `AKIPv6`: `xxxx:xxxx:xxxx:xxxx::/64`.
+
+## Counters
+
+Each server has a fixed set of counters:
+
+```
+class (Ord c, Enum c, Bounded c) => AddrCounter c where
+  counterName :: c -> Text
+  counterBounds :: c -> [Int]
+  connectionsCounter :: c
+```
+
+`counterName` is the Prometheus label value and the control port argument.
+
+Counters of every server:
+
+- `connections`: connections passed to the protocol handler, after TLS setup.
+- One counter per command tag, named by the tag: an attempt is counted, failed ones included.
+- `errors`: parse failures and authorization failures.
+
+Command tag types:
+
+- SMP: `CommandTag` in `Simplex.Messaging.Protocol`.
+- XFTP: `FileCommandTag` in `Simplex.FileTransfer.Protocol`.
+- NTF: `NtfCommandTag` in `Simplex.Messaging.Notifications.Protocol`.
+
+A function from a command to its counter is added for each server.
+
+SMP proxy counters, by result of `PRXY`:
+
+- `PRXY_own`: own server.
+- `PRXY_connected`: other server with an existing connection.
+- `PRXY_new`: other server with a new connection.
+- `PRXY_onion`: server with onion hosts only.
+- `PRXY_failed`: error.
+
+SMP proxy counters, by result of `PFWD`:
+
+- `PFWD_own`: own server.
+- `PFWD_other`: other server.
+- `PFWD_onion`: server with onion hosts only.
+- `PFWD_failed`: error other than a protocol error, or no session.
+
+Class precedence: `failed`, then `own`, then `onion`.
+
+XFTP counters:
+
+- `upload_kb`: kilobytes of chunks stored by successful `FPUT`.
+- `download_kb`: kilobytes of chunks sent for `FGET`.
+- `recipients`: recipient keys added by successful `FNEW` and `FADD`.
+
+## Periods
+
+- A period is a fixed interval of `period` seconds. The first period starts at server start.
+- Counts of the current period are accumulated as events occur.
+- Counts of the previous period are the counts of the last closed period.
+- Histograms use the counts of closed periods.
+- The control port command shows the previous and the current counts, sorted by their sum.
+
+## Module `Simplex.Messaging.Server.AddressStats`
+
+```
+data AddrStats c = AddrStats
+  { connectionsCount :: TVar Int,
+    current :: Map c (IORef Int),
+    previous :: Map c (IORef Int)
+  }
+
+type AddrStatsMap c = TMap AddrKey (AddrStats c)
+
+data AddrHistogram = AddrHistogram
+  { bucketCounts :: [Int],
+    countSum :: Int,
+    addressCount :: Int,
+    periodMax :: Int
+  }
+
+newtype AddressStatsConfig = AddressStatsConfig
+  { period :: Int64
+  }
+```
+
+- `current` and `previous` contain one cell per counter, created when the address is inserted.
+- `bucketCounts` are cumulative counts per bound, from server start.
+
+Bounds:
+
+- `countBounds`: 1 to 9, then R10 values times powers of 10, rounded half up to integers, up to 1,000,000: 60 bounds.
+- `kilobyteBounds`: the same, up to 100,000,000: 80 bounds.
+- R10 values: 1, 1.25, 1.6, 2, 2.5, 3.15, 4, 5, 6.3, 8.
+
+Functions:
+
+- `addrKey :: SockAddr -> Maybe AddrKey`
+- `addrKeyText :: AddrKey -> Text`
+- `counterByName :: AddrCounter c => Text -> Maybe c`
+- `withAddrStats :: AddrCounter c => Maybe (AddrStatsMap c) -> Socket -> (Maybe (AddrStats c) -> IO a) -> IO a`:
+  1. The peer address is read with `getPeerName`. On an exception, the connection is not counted.
+  2. In one transaction, the entry is looked up or inserted, and `connectionsCount` is incremented.
+  3. `connectionsCounter` is incremented.
+  4. The action is run. `connectionsCount` is decremented in `finally`.
+- `incAddrCounter :: AddrCounter c => AddrStats c -> c -> IO ()`
+- `addAddrCounter :: AddrCounter c => AddrStats c -> c -> Int -> IO ()`
+- `rolloverAddrStats :: AddrCounter c => AddrStatsMap c -> IORef (Map c AddrHistogram) -> IO ()`, in one pass over the entries, per counter:
+  1. The `current` cell is set to 0. Its value is the closing count.
+  2. A non-zero closing count is added to the histogram: to each bucket with a bound at or above it, to `countSum`, and to `addressCount`.
+  3. `periodMax` is the largest closing count.
+  4. The `previous` cell is set to the closing count.
+  
+  After the pass, an entry is deleted when `connectionsCount` is 0 and every closing count is 0. `connectionsCount` is read in the deleting transaction.
+- `addressStatsThread :: AddrCounter c => AddressStatsConfig -> AddrStatsMap c -> IORef (Map c AddrHistogram) -> IO ()`: `rolloverAddrStats` every `period` seconds.
+- `topAddresses :: AddrCounter c => AddrStatsMap c -> c -> Int -> IO [(AddrKey, Int, Int)]`: the key, previous count and current count of the counter, for up to the given number of entries, sorted by the sum of both counts, largest first. Entries with both counts 0 are excluded.
+- `addrHistogramMetrics :: AddrCounter c => Text -> Map c AddrHistogram -> Text`: Prometheus text for a metric name prefix.
+
+## Prometheus
+
+Written with the other metrics of each server, when both `addressStats` and `prometheus_interval` are set:
+
+```
+# TYPE <prefix>_client_address_period_count histogram
+<prefix>_client_address_period_count_bucket{counter="<name>",le="<bound>"} <addresses>
+<prefix>_client_address_period_count_bucket{counter="<name>",le="+Inf"} <addresses>
+<prefix>_client_address_period_count_sum{counter="<name>"} <sum>
+<prefix>_client_address_period_count_count{counter="<name>"} <addresses>
+# TYPE <prefix>_client_address_period_max gauge
+<prefix>_client_address_period_max{counter="<name>"} <max>
+```
+
+Prefixes:
+
+- `simplex_smp`
+- `simplex_xftp`
+- `simplex_ntf`
+
+In scrape jobs, `convert_classic_histograms_to_nhcb: true` is set.
+
+## Configuration
+
+The same INI section in `smp-server.ini`, `file-server.ini` and `ntf-server.ini`:
+
+```
+[ADDRESS_STATS]
+enable = off
+period = 300
+```
+
+- `enable = off` or an absent section: statistics are off.
+- `period`: seconds.
+
+`iniAddressStats :: Ini -> Maybe AddressStatsConfig` is added to `Simplex.Messaging.Server.CLI`.
+
+Added to `ServerConfig`, `XFTPServerConfig` and `NtfServerConfig`:
+
+```
+addressStats :: Maybe AddressStatsConfig
+```
+
+The section is written by the `init` command of each server with `enable = off`.
+
+With `addressStats` set:
+
+- an `AddrStatsMap` and a histogram reference are added to the server env
+- `addressStatsThread` is added to the server threads
+
+## Control port
+
+Command of the SMP, XFTP and NTF control ports:
+
+```
+addresses <counter> [<n>]
+```
+
+- User role.
+- `<counter>`: a counter name.
+- `<n>`: number of addresses, 10 when absent.
+
+Output of `topAddresses`:
+
+```
+address,previous,current
+<key>,<count>,<count>
+```
+
+Errors:
+
+- Unknown counter name: `error: unknown counter`.
+- Statistics off: `error: address statistics are off`.
+
+Changes:
+
+- `CPAddresses Text (Maybe Int)` is added to `ControlProtocol` in `Simplex.Messaging.Server.Control`, `Simplex.FileTransfer.Server.Control` and `Simplex.Messaging.Notifications.Server.Control`.
+- `addresses` is added to the `help` output of each control port.
+
+## SMP server
+
+In `Simplex.Messaging.Server`:
+
+- `runServer`: in the branch without HTTP, `runTransportServerState_` is called with `TLSServerCredential {credential = smpCreds, sniCredential = Nothing}`. In both branches, the socket is passed to `runClient`.
+- `runClient` is run in `withAddrStats`. The `AddrStats` is passed to `runClientTransport`.
+- `clientAddrStats :: Maybe (AddrStats SMPAddrCounter)` is added to `Client`, as a parameter of `newClient`.
+- `receive`:
+  - The counter of each parsed transmission is incremented.
+  - `errors` is incremented for each parse error and each failed verification.
+- `processProxiedCmd`: the proxy counter is incremented when the result of `PRXY` or `PFWD` is known.
+
+## NTF server
+
+In `Simplex.Messaging.Notifications.Server`:
+
+- `runServer`: `runTransportServerState_` is called with a new `SocketState` and `TLSServerCredential {credential = srvCreds, sniCredential = Nothing}`. The socket is passed to `runClient`.
+- `runClient` is run in `withAddrStats`. The `AddrStats` is passed to `runNtfClientTransport`.
+- `ntfClientAddrStats :: Maybe (AddrStats NtfAddrCounter)` is added to `NtfServerClient`, as a parameter of `newNtfServerClient`.
+- `receive`: counting as in the SMP server.
+
+## XFTP server
+
+In `Simplex.Messaging.Transport.HTTP2.Server`:
+
+- A parameter `Socket -> SessionId -> IO () -> IO ()` is added to `runHTTP2Server`. The handler of each connection is run inside it, after TLS setup.
+- In `getHTTP2Server`, the wrapper only runs the handler.
+
+In `Simplex.FileTransfer.Server`:
+
+- `TMap SessionId (AddrStats XFTPAddrCounter)` is created in `runServer`.
+- In the connection wrapper, the handler is run in `withAddrStats`. The `AddrStats` is inserted in the map before the handler, and deleted in `finally`.
+- `addrStats :: Maybe (AddrStats XFTPAddrCounter)` is added to `XFTPTransportRequest`. It is looked up by session ID for each request.
+- `processRequest`:
+  - The counter of each decoded command is incremented.
+  - `errors` is incremented for each decoding error and each failed verification.
+- `upload_kb` is increased by the chunk size, rounded up to kilobytes, after a successful `FPUT`.
+- `download_kb` is increased by the chunk size, rounded up to kilobytes, when an `FGET` response with the file is sent.
+- `recipients` is increased by the number of recipient keys after a successful `FNEW` or `FADD`.
+
+## Tests
+
+`CoreTests.AddressStatsTests`:
+
+- `addrKey`:
+  - IPv4 address
+  - IPv4-mapped IPv6 address
+  - two IPv6 addresses in one /64: one key
+  - two IPv6 addresses in different /64: two keys
+- bounds:
+  - `countBounds` values
+  - `kilobyteBounds` values
+- `counterByName` of each counter name
+- `rolloverAddrStats`:
+  - histogram buckets, sum, address count, and max
+  - zero counts not added to histograms
+  - previous counts set to closing counts
+  - entry with an open connection kept
+  - entry with closing counts kept
+  - entry without connections and closing counts deleted
+- `topAddresses`:
+  - order by the sum of previous and current counts
+  - number of entries limited
+  - entries with both counts 0 excluded
+- `addrHistogramMetrics` output
+- command to counter function of each server
+
+Server tests, with `addressStats` set and `period` 1:
+
+- SMP basic queue tests
+- SMP proxy tests
+- XFTP basic file tests
+- NTF basic token tests
+
+Control port test, SMP server with `addressStats` and a control port set:
+
+- after messages are sent, `addresses SEND` lists the test client address
+- `addresses` without authentication returns `AUTH`
+- `addresses` with an unknown counter returns `error: unknown counter`
+
+`CLITests`: `enable = off` is checked in `[ADDRESS_STATS]` of each generated INI.
@@ -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 (optional if known)
+    CI-->>CC: user confirmation
 and
    note over HC: validate server X509 credentials
    HC->>HI: session code from tlsUnique
-    HI-->>HC: user confirmation (optional if known)
+    HI-->>HC: user confirmation
 end

 note over CI, HI: 3. Session verification and protocol negotiation
@@ -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, 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).
+- 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).

 - undetectably communicate at the same time as Alice with her contacts. Doing so would result in the contact getting different messages with repeated IDs.

@@ -1120,8 +1120,6 @@ 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
@@ -121,15 +121,9 @@ 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. 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 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.

-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.
+Once the session is confirmed by the user, the host sends HELLO block to the controller.

 XRCP blocks inside TLS are padded to 16384 bytes.

@@ -311,7 +305,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 in the first session with the controller.
+- same as passive adversary, provided that user visually verified session code out-of-band.

 #### An active adversary with the access to the network:

@@ -330,15 +324,6 @@ 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:*
@@ -271,6 +271,7 @@ library
          Simplex.FileTransfer.Server.Store
          Simplex.FileTransfer.Server.StoreLog
          Simplex.Messaging.Server
+          Simplex.Messaging.Server.AddressStats
          Simplex.Messaging.Server.CLI
          Simplex.Messaging.Server.Control
          Simplex.Messaging.Server.Env.STM
@@ -542,6 +543,7 @@ test-suite simplexmq-test
      AgentTests.ServerChoice
      AgentTests.ShortLinkTests
      CLITests
+      CoreTests.AddressStatsTests
      CoreTests.BatchingTests
      CoreTests.CryptoFileTests
      CoreTests.CryptoTests
@@ -25,6 +25,7 @@ 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
@@ -64,6 +65,7 @@ import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (BlockingInfo, CommandError (..), EntityId (..), RcvPublicAuthKey, RcvPublicDhKey, RecipientId, SignedTransmission, pattern NoEntity)
 import Simplex.Messaging.Server (controlPortAuth, dummyVerifyCmd, verifyCmdAuthorization)
+import Simplex.Messaging.Server.AddressStats (AddrStats, ServerAddrStats (..), addAddrCounter, addressStatsThread, incAddrCounter, printTopAddresses, withAddrStats)
 import Simplex.Messaging.Server.Control (CPClientRole (..))
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.QueueStore (ServerEntityStatus (..))
@@ -100,7 +102,8 @@ data XFTPTransportRequest = XFTPTransportRequest
    request :: H.Request,
    sendResponse :: H.Response -> IO (),
    sniUsed :: SNICredentialUsed,
-    addCORS :: Bool
+    addCORS :: Bool,
+    clientAddrStats :: Maybe (AddrStats XFTPAddrCounter)
  }

 corsHeaders :: Bool -> [N.Header]
@@ -140,6 +143,7 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira
        : expireFiles fileExpiration
        : serverStatsThread_ cfg
          <> prometheusMetricsThread_ cfg
+          <> addressStatsThread_ cfg
          <> controlPortThread_ cfg
    )
    `finally` stopServer
@@ -153,20 +157,27 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira
        Left e -> putStrLn ("Server has no valid key: " <> show e) >> exitFailure
      env <- ask
      sessions <- liftIO TM.emptyIO
+      sessionAddrStats <- liftIO TM.emptyIO
+      stats_ <- asks addrStats
      let cleanup sessionId = atomically $ TM.delete sessionId sessions
+          withConnection sock sessionId handler =
+            withAddrStats (addrStatsMap <$> stats_) sock $ \case
+              Just s -> (atomically (TM.insert sessionId s sessionAddrStats) >> handler) `finally` atomically (TM.delete sessionId sessionAddrStats)
+              Nothing -> handler
          srvParams = if isJust httpCreds_ then defaultSupportedParamsHTTPS else defaultSupportedParams
      webCanonicalRoot_ <- liftIO $ mapM canonicalizePath (webStaticPath cfg)
-      liftIO . runHTTP2Server started xftpPort defaultHTTP2BufferSize srvParams srvCreds httpCreds_ transportConfig inactiveClientExpiration cleanup $ \sniUsed sessionId sessionALPN r sendResponse -> do
+      liftIO . runHTTP2Server started xftpPort defaultHTTP2BufferSize srvParams srvCreds httpCreds_ transportConfig inactiveClientExpiration cleanup withConnection $ \sniUsed sessionId sessionALPN r sendResponse -> do
        let addCORS' = sniUsed && addCORSHeaders transportConfig
        case H.requestMethod r of
          Just "OPTIONS" | addCORS' -> sendResponse $ H.responseNoBody N.ok200 corsPreflightHeaders
          Just "GET" | sniUsed -> forM_ webCanonicalRoot_ $ \root -> serveStaticPageH2 root r sendResponse
          _ -> do
            reqBody <- getHTTP2Body r xftpBlockSize
+            clientAddrStats <- TM.lookupIO sessionId sessionAddrStats
            let v = VersionXFTP 1
                thServerVRange = versionToRange v
                thParams0 = THandleParams {sessionId, blockSize = xftpBlockSize, thVersion = v, thServerVRange, thAuth = Nothing, implySessId = False, encryptBlock = Nothing, serviceAuth = False, serverInfo = Nothing}
-                req0 = XFTPTransportRequest {thParams = thParams0, request = r, reqBody, sendResponse, sniUsed, addCORS = addCORS'}
+                req0 = XFTPTransportRequest {thParams = thParams0, request = r, reqBody, sendResponse, sniUsed, addCORS = addCORS', clientAddrStats}
            flip runReaderT env $ case sessionALPN of
              Nothing -> processRequest req0
              Just alpn
@@ -329,24 +340,30 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira
      [savePrometheusMetrics interval prometheusMetricsFile]
    prometheusMetricsThread_ _ = []

+    addressStatsThread_ :: XFTPServerConfig s -> [M s ()]
+    addressStatsThread_ XFTPServerConfig {addressStats = Just statsCfg} = [asks addrStats >>= mapM_ (liftIO . addressStatsThread statsCfg)]
+    addressStatsThread_ _ = []
+
    savePrometheusMetrics :: Int -> FilePath -> M s ()
    savePrometheusMetrics saveInterval metricsFile = do
      labelMyThread "savePrometheusMetrics"
      liftIO $ putStrLn $ "Prometheus metrics saved every " <> show saveInterval <> " seconds to " <> metricsFile
      ss <- asks serverStats
+      stats_ <- asks addrStats
      rtsOpts <- liftIO $ maybe ("set " <> rtsOptionsEnv) T.pack <$> lookupEnv (T.unpack rtsOptionsEnv)
      let interval = 1000000 * saveInterval
      liftIO $ forever $ do
        threadDelay interval
        ts <- getCurrentTime
-        sm <- getFileServerMetrics ss rtsOpts
+        sm <- getFileServerMetrics ss stats_ rtsOpts
        T.writeFile metricsFile $ xftpPrometheusMetrics sm ts

-    getFileServerMetrics :: FileServerStats -> T.Text -> IO FileServerMetrics
-    getFileServerMetrics ss rtsOptions = do
+    getFileServerMetrics :: FileServerStats -> Maybe (ServerAddrStats XFTPAddrCounter) -> T.Text -> IO FileServerMetrics
+    getFileServerMetrics ss stats_ rtsOptions = do
      d <- getFileServerStatsData ss
      let fd = periodStatDataCounts $ _filesDownloaded d
-      pure FileServerMetrics {statsData = d, filesDownloadedPeriods = fd, rtsOptions}
+      addressHistograms <- mapM (readIORef . addrHistograms) stats_
+      pure FileServerMetrics {statsData = d, filesDownloadedPeriods = fd, rtsOptions, addressHistograms}

    controlPortThread_ :: XFTPServerConfig s -> [M s ()]
    controlPortThread_ XFTPServerConfig {controlPort = Just port} = [runCPServer port]
@@ -401,7 +418,10 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira
                  (fr, _) <- ExceptT $ liftIO $ getFile fs SFRecipient fileId
                  ExceptT $ blockServerFile fr info
                liftIO . hPutStrLn h $ either (\e -> "error: " <> show e) (\() -> "ok") r
-              CPHelp -> hPutStrLn h "commands: stats-rts, delete, help, quit"
+              CPAddresses name n_ -> withUserRole $ do
+                stats_ <- unliftIO u $ asks addrStats
+                printTopAddresses h (addrStatsMap <$> stats_) name n_
+              CPHelp -> hPutStrLn h "commands: stats-rts, delete, addresses, help, quit"
              CPQuit -> pure ()
              CPSkip -> pure ()
              where
@@ -420,21 +440,23 @@ data ServerFile = ServerFile
  }

 processRequest :: FileStoreClass s => XFTPTransportRequest -> M s ()
-processRequest XFTPTransportRequest {thParams, reqBody = body@HTTP2Body {bodyHead}, sendResponse, addCORS}
-  | B.length bodyHead /= xftpBlockSize = sendXFTPResponse ("", NoEntity, FRErr BLOCK) Nothing
+processRequest XFTPTransportRequest {thParams, reqBody = body@HTTP2Body {bodyHead}, sendResponse, addCORS, clientAddrStats}
+  | B.length bodyHead /= xftpBlockSize = incAddrStats XACErrors >> sendXFTPResponse ("", NoEntity, FRErr BLOCK) Nothing
  | otherwise =
      case xftpDecodeTServer thParams bodyHead of
-        Right (Right t@(_, _, (corrId, fId, _))) -> do
+        Right (Right t@(_, _, (corrId, fId, cmd))) -> do
+          incAddrStats $ xftpCmdCounter cmd
          let THandleParams {thAuth} = thParams
              ent = peerEntitlement =<< thAuth
          verifyXFTPTransmission thAuth t >>= \case
-            VRVerified req -> uncurry send =<< processXFTPRequest ent body req
-            VRFailed e -> send (FRErr e) Nothing
+            VRVerified req -> uncurry send =<< processXFTPRequest clientAddrStats ent body req
+            VRFailed e -> incAddrStats XACErrors >> send (FRErr e) Nothing
          where
            send resp = sendXFTPResponse (corrId, fId, resp)
-        Right (Left (corrId, fId, e)) -> sendXFTPResponse (corrId, fId, FRErr e) Nothing
-        Left e -> sendXFTPResponse ("", NoEntity, FRErr e) Nothing
+        Right (Left (corrId, fId, e)) -> incAddrStats XACErrors >> sendXFTPResponse (corrId, fId, FRErr e) Nothing
+        Left e -> incAddrStats XACErrors >> sendXFTPResponse ("", NoEntity, FRErr e) Nothing
  where
+    incAddrStats c = liftIO $ forM_ clientAddrStats (`incAddrCounter` c)
    sendXFTPResponse t' serverFile_ = do
      let t_ = xftpEncodeTransmission thParams t'
 #ifdef slow_servers
@@ -489,8 +511,8 @@ verifyXFTPTransmission thAuth (tAuth, authorized, (corrId, fId, cmd)) =
    -- TODO verify with DH authorization
    req `verifyWith` k = if verifyCmdAuthorization thAuth tAuth authorized corrId k then VRVerified req else VRFailed AUTH

-processXFTPRequest :: forall s. FileStoreClass s => Maybe SessionEntitlement -> HTTP2Body -> XFTPRequest -> M s (FileResponse, Maybe ServerFile)
-processXFTPRequest ent HTTP2Body {bodyPart} = \case
+processXFTPRequest :: forall s. FileStoreClass s => Maybe (AddrStats XFTPAddrCounter) -> Maybe SessionEntitlement -> HTTP2Body -> XFTPRequest -> M s (FileResponse, Maybe ServerFile)
+processXFTPRequest addrStats_ ent HTTP2Body {bodyPart} = \case
  XFTPReqNew file rks auth storageHours -> noFile =<< ifM allowNew (createFile file rks storageHours) (pure $ FRErr AUTH)
    where
      allowNew = do
@@ -508,6 +530,10 @@ processXFTPRequest ent HTTP2Body {bodyPart} = \case
  XFTPReqPing -> noFile FRPong
  where
    noFile resp = pure (resp, Nothing)
+    addAddrStats :: XFTPAddrCounter -> Int -> M s ()
+    addAddrStats c n = liftIO $ forM_ addrStats_ $ \s -> addAddrCounter s c n
+    kilobytes :: Word32 -> Int
+    kilobytes bytes = (fromIntegral bytes + 1023) `div` 1024
    createFile :: FileInfo -> NonEmpty RcvPublicAuthKey -> Maybe Word32 -> M s FileResponse
    createFile file rks storageHours = do
      st <- asks fileStore
@@ -531,6 +557,7 @@ processXFTPRequest ent HTTP2Body {bodyPart} = \case
        stats <- asks serverStats
        lift $ incFileStat filesCreated
        liftIO $ atomicModifyIORef'_ (fileRecipients stats) (+ length rks)
+        lift $ addAddrStats XACRecipients $ length rks
        let rIds = L.map (\(FileRecipient rId _) -> rId) rcps
        pure $ FRSndIds sId rIds (Just (GSTExpires (roundedSeconds fileExpiresAt)))
      pure $ either FRErr id r
@@ -566,6 +593,7 @@ processXFTPRequest ent HTTP2Body {bodyPart} = \case
        lift $ withFileLog $ \sl -> logAddRecipients sl sId rcps
        stats <- asks serverStats
        liftIO $ atomicModifyIORef'_ (fileRecipients stats) (+ length rks)
+        lift $ addAddrStats XACRecipients $ length rks
        let rIds = L.map (\(FileRecipient rId _) -> rId) rcps
        pure $ FRRcvIds rIds
      pure $ either FRErr id r
@@ -593,7 +621,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 </> serverFileName senderId
+            let fPath = path </> B.unpack (B64.encode $ unEntityId senderId)
            receiveChunk (XFTPRcvChunkSpec fPath size digest) >>= \case
              Right () -> do
                stats <- asks serverStats
@@ -604,6 +632,7 @@ processXFTPRequest ent HTTP2Body {bodyPart} = \case
                    incFileStat filesUploaded
                    incFileStat filesCount
                    liftIO $ atomicModifyIORef'_ (filesSize stats) (+ fromIntegral size)
+                    addAddrStats XACUploadKb $ kilobytes size
                    pure FROk
                  Left _e -> do
                    us <- asks usedStorage
@@ -633,6 +662,7 @@ processXFTPRequest ent HTTP2Body {bodyPart} = \case
                  stats <- asks serverStats
                  incFileStat fileDownloads
                  liftIO $ updatePeriodStats (filesDownloaded stats) senderId
+                  addAddrStats XACDownloadKb $ kilobytes size
                  pure (FRFile sDhKey cbNonce, Just ServerFile {filePath = path, fileSize = size, sbState})
                _ -> pure (FRErr INTERNAL, Nothing)
        _ -> pure (FRErr NO_FILE, Nothing)
@@ -7,15 +7,18 @@ module Simplex.FileTransfer.Server.Control
 where

 import qualified Data.Attoparsec.ByteString.Char8 as A
+import Data.Text (Text)
 import Simplex.FileTransfer.Protocol (XFTPFileId)
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (BasicAuth, BlockingInfo)
+import Simplex.Messaging.Server.AddressStats (addressesArgs, addressesArgsP)

 data ControlProtocol
  = CPAuth BasicAuth
  | CPStatsRTS
  | CPDelete XFTPFileId
  | CPBlock XFTPFileId BlockingInfo
+  | CPAddresses Text (Maybe Int)
  | CPHelp
  | CPQuit
  | CPSkip
@@ -26,6 +29,7 @@ instance StrEncoding ControlProtocol where
    CPStatsRTS -> "stats-rts"
    CPDelete fId -> strEncode (Str "delete", fId)
    CPBlock fId info -> strEncode (Str "block", fId, info)
+    CPAddresses name n_ -> "addresses " <> addressesArgs name n_
    CPHelp -> "help"
    CPQuit -> "quit"
    CPSkip -> ""
@@ -35,6 +39,7 @@ instance StrEncoding ControlProtocol where
      "stats-rts" -> pure CPStatsRTS
      "delete" -> CPDelete <$> _strP
      "block" -> CPBlock <$> _strP <*> _strP
+      "addresses" -> uncurry CPAddresses <$> addressesArgsP
      "help" -> pure CPHelp
      "quit" -> pure CPQuit
      "" -> pure CPSkip
@@ -16,6 +16,8 @@ module Simplex.FileTransfer.Server.Env
    XFTPStoreConfig (..),
    XFTPEnv (..),
    XFTPRequest (..),
+    XFTPAddrCounter (..),
+    xftpCmdCounter,
    XFTPStoreType,
    FileStore (..),
    AFStoreType (..),
@@ -45,7 +47,7 @@ import Data.Word (Word16, Word32)
 import Data.X509.Validation (Fingerprint (..))
 import Network.Socket
 import qualified Network.TLS as T
-import Simplex.FileTransfer.Protocol (FileCmd, FileInfo (..), XFTPFileId)
+import Simplex.FileTransfer.Protocol (FileCmd (..), FileCommand (..), FileInfo (..), XFTPFileId)
 import Simplex.FileTransfer.Server.Stats
 import Data.Either (fromRight)
 import Data.Ini (Ini, lookupValue)
@@ -67,6 +69,7 @@ import Simplex.FileTransfer.Transport (VersionRangeXFTP)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.BBS (BBSPublicKey)
 import Simplex.Messaging.Protocol (BasicAuth, RcvPublicAuthKey)
+import Simplex.Messaging.Server.AddressStats (AddrCounter (..), AddressStatsConfig, ServerAddrStats, countBounds, kilobyteBounds, newServerAddrStats)
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.Information (ServerPublicInfo)
 import Simplex.Messaging.Transport (EntitlementConfig (..))
@@ -115,6 +118,7 @@ data XFTPServerConfig s = XFTPServerConfig
    serverStatsBackupFile :: Maybe FilePath,
    prometheusInterval :: Maybe Int,
    prometheusMetricsFile :: FilePath,
+    addressStats :: Maybe AddressStatsConfig,
    transportConfig :: TransportServerConfig,
    responseDelay :: Int,
    webStaticPath :: Maybe FilePath
@@ -136,9 +140,55 @@ data XFTPEnv s = XFTPEnv
    serverIdentity :: C.KeyHash,
    tlsServerCreds :: T.Credential,
    httpServerCreds :: Maybe T.Credential,
-    serverStats :: FileServerStats
+    serverStats :: FileServerStats,
+    addrStats :: Maybe (ServerAddrStats XFTPAddrCounter)
  }

+data XFTPAddrCounter
+  = XACConnections
+  | XACErrors
+  | XACFNew
+  | XACFAdd
+  | XACFPut
+  | XACFDel
+  | XACFGet
+  | XACFAck
+  | XACPing
+  | XACUploadKb
+  | XACDownloadKb
+  | XACRecipients
+  deriving (Eq, Ord, Enum, Bounded, Show)
+
+instance AddrCounter XFTPAddrCounter where
+  counterName = \case
+    XACConnections -> "connections"
+    XACErrors -> "errors"
+    XACFNew -> "FNEW"
+    XACFAdd -> "FADD"
+    XACFPut -> "FPUT"
+    XACFDel -> "FDEL"
+    XACFGet -> "FGET"
+    XACFAck -> "FACK"
+    XACPing -> "PING"
+    XACUploadKb -> "upload_kb"
+    XACDownloadKb -> "download_kb"
+    XACRecipients -> "recipients"
+  counterBounds = \case
+    XACUploadKb -> kilobyteBounds
+    XACDownloadKb -> kilobyteBounds
+    _ -> countBounds
+  connectionsCounter = XACConnections
+
+xftpCmdCounter :: FileCmd -> XFTPAddrCounter
+xftpCmdCounter (FileCmd _ cmd) = case cmd of
+  FNEW {} -> XACFNew
+  FADD _ -> XACFAdd
+  FPUT -> XACFPut
+  FDEL -> XACFDel
+  FGET _ -> XACFGet
+  FACK -> XACFAck
+  PING -> XACPing
+
 fileStore :: XFTPEnv s -> s
 fileStore = fromFileStore . store
 {-# INLINE fileStore #-}
@@ -182,7 +232,7 @@ defaultFileExpiration =
    }

 newXFTPServerEnv :: FileStoreClass s => XFTPServerConfig s -> IO (XFTPEnv s)
-newXFTPServerEnv config@XFTPServerConfig {serverStoreCfg, fileSizeQuota, fileExpiration, fileStorageEntitlements, xftpCredentials, httpCredentials} = do
+newXFTPServerEnv config@XFTPServerConfig {serverStoreCfg, fileSizeQuota, fileExpiration, fileStorageEntitlements, xftpCredentials, httpCredentials, addressStats} = do
  let defaultMax = ttl fileExpiration
      belowDefault = M.filter ((< defaultMax) . storageTime) fileStorageEntitlements
  unless (M.null belowDefault) $ do
@@ -210,7 +260,8 @@ newXFTPServerEnv config@XFTPServerConfig {serverStoreCfg, fileSizeQuota, fileExp
  httpServerCreds <- mapM loadServerCredential httpCredentials
  Fingerprint fp <- loadFingerprint xftpCredentials
  serverStats <- newFileServerStats =<< getCurrentTime
-  pure XFTPEnv {config, store, usedStorage, storeLog, random, tlsServerCreds, httpServerCreds, serverIdentity = C.KeyHash fp, serverStats}
+  addrStats <- mapM (const newServerAddrStats) addressStats
+  pure XFTPEnv {config, store, usedStorage, storeLog, random, tlsServerCreds, httpServerCreds, serverIdentity = C.KeyHash fp, serverStats, addrStats}

 data XFTPRequest
  = XFTPReqNew FileInfo (NonEmpty RcvPublicAuthKey) (Maybe BasicAuth) (Maybe Word32)
@@ -204,6 +204,8 @@ xftpServerCLI_ generateSite serveStaticFiles cfgPath logPath = do
               \disconnect = off\n"
            <> ("# ttl = " <> tshow (ttl defaultInactiveClientExpiration) <> "\n")
            <> ("# check_interval = " <> tshow (checkInterval defaultInactiveClientExpiration) <> "\n")
+            <> "\n"
+            <> addressStatsIniContent
            <> "\n\
               \[WEB]\n\
               \# Set path to generate static mini-site for server information\n"
@@ -325,6 +327,7 @@ xftpServerCLI_ generateSite serveStaticFiles cfgPath logPath = do
              serverStatsBackupFile = logStats $> combine logPath "file-server-stats.log",
              prometheusInterval = eitherToMaybe (lookupValue "STORE_LOG" "prometheus_interval" ini) >>= readMaybe . T.unpack,
              prometheusMetricsFile = combine logPath "xftp-server-metrics.txt",
+              addressStats = iniAddressStats ini,
              transportConfig =
                let cfg =
                      mkTransportServerConfig
@@ -11,12 +11,15 @@ module Simplex.FileTransfer.Server.Prometheus
  ) where

 import Data.Int (Int64)
+import Data.Map.Strict (Map)
 import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Time.Clock (UTCTime (..), diffUTCTime)
 import Data.Time.Clock.System (systemEpochDay)
 import Data.Time.Format.ISO8601 (iso8601Show)
+import Simplex.FileTransfer.Server.Env (XFTPAddrCounter)
 import Simplex.FileTransfer.Server.Stats
+import Simplex.Messaging.Server.AddressStats (AddrHistogram, addrHistogramMetrics)
 import Simplex.Messaging.Server.Stats (PeriodStatCounts (..))
 import Simplex.Messaging.Transport (simplexMQVersion)
 import Simplex.Messaging.Util (tshow)
@@ -24,7 +27,8 @@ import Simplex.Messaging.Util (tshow)
 data FileServerMetrics = FileServerMetrics
  { statsData :: FileServerStatsData,
    filesDownloadedPeriods :: PeriodStatCounts,
-    rtsOptions :: Text
+    rtsOptions :: Text,
+    addressHistograms :: Maybe (Map XFTPAddrCounter AddrHistogram)
  }

 rtsOptionsEnv :: Text
@@ -33,9 +37,9 @@ rtsOptionsEnv = "XFTP_RTS_OPTIONS"
 {-# FOURMOLU_DISABLE\n#-}
 xftpPrometheusMetrics :: FileServerMetrics -> UTCTime -> Text
 xftpPrometheusMetrics sm ts =
-  time <> files <> info
+  time <> files <> clientAddresses <> info
  where
-    FileServerMetrics {statsData, filesDownloadedPeriods, rtsOptions} = sm
+    FileServerMetrics {statsData, filesDownloadedPeriods, rtsOptions, addressHistograms} = sm
    FileServerStatsData
      { _fromTime,
        _filesCreated,
@@ -122,6 +126,7 @@ xftpPrometheusMetrics sm ts =
      \simplex_xftp_files_count_monthly " <> mstr (monthCount filesDownloadedPeriods) <> "\n\
      \# filesDownloaded.monthCount\n\
      \\n"
+    clientAddresses = maybe "" (addrHistogramMetrics "simplex_xftp" tsEpoch) addressHistograms
    info =
      "# Info\n\
      \# ----\n\
@@ -17,7 +17,6 @@ module Simplex.FileTransfer.Server.StoreLog
    logDeleteFile,
    logBlockFile,
    logAckFile,
-    serverFileName,
  )
 where

@@ -25,7 +24,6 @@ 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 ((.:), (.::.))
@@ -37,11 +35,10 @@ 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, EntityId (..), RcvPublicAuthKey, RecipientId, SenderId)
+import Simplex.Messaging.Protocol (BlockingInfo, 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
@@ -102,9 +99,6 @@ 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

@@ -121,9 +115,7 @@ 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
-        | takeFileName path == serverFileName qId -> setFilePath st qId path
-        | otherwise -> pure $ Left AUTH
+      PutFile qId path -> setFilePath st qId path
      AddRecipients sId rcps -> runExceptT $ addRecipients sId rcps
      DeleteFile sId -> deleteFile st sId
      BlockFile sId info -> blockFile st sId info True
@@ -4388,9 +4388,8 @@ 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
-  pure (agentMsgBody, CR.rcRcvKEM rc')
+  liftEither $ bimap (SEAgentError . cryptoError) (,CR.rcRcvKEM rc') agentMsgBody_

 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
@@ -35,8 +35,6 @@ module Simplex.Messaging.Client
    ProxiedRelay (..),
    getProtocolClient,
    closeProtocolClient,
-    closeTimedOutClient,
-    pClientSentCommandsCount,
    protocolClientServer,
    protocolClientServer',
    transportHost',
@@ -152,8 +150,7 @@ import Data.Int (Int64)
 import Data.List (find, isSuffixOf)
 import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.List.NonEmpty as L
-import qualified Data.Map.Strict as M
-import Data.Maybe (catMaybes, fromMaybe, isNothing)
+import Data.Maybe (catMaybes, fromMaybe)
 import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Time.Clock (UTCTime (..), diffUTCTime, getCurrentTime)
@@ -515,8 +512,7 @@ defaultSMPClientConfig =
 data Request err msg = Request
  { corrId :: CorrId,
    entityId :: EntityId,
-    -- only kept to report a late response to msgQ, a forwarded command holds a whole block
-    command :: Maybe (ProtoCommand msg),
+    command :: ProtoCommand msg,
    pending :: TVar Bool,
    responseVar :: TMVar (Either (ProtocolClientError err) msg)
  }
@@ -662,9 +658,10 @@ 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}} h = forever $ do
+    receive ProtocolClient {client_ = PClient {rcvQ, lastReceived, timeoutErrorCount}} 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
@@ -711,9 +708,7 @@ getProtocolClient g nm transportSession@(_, srv, _) cfg@ProtocolClientConfig {qS
                    (pure False)
              if wasPending
                then pure Nothing
-                else sendMsg $ case command of
-                  Just cmd | entityId == entId -> STResponse cmd clientResp
-                  _ -> STUnexpectedError unexpected
+                else sendMsg $ if entityId == entId then STResponse command clientResp else STUnexpectedError unexpected
      where
        unexpected = unexpectedResponse respOrErr
        clientResp = case respOrErr of
@@ -755,15 +750,6 @@ 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
-
-pClientSentCommandsCount :: ProtocolClient v err msg -> IO Int
-pClientSentCommandsCount ProtocolClient {client_ = PClient {sentCommands}} = M.size <$> readTVarIO sentCommands
-
 -- | SMP client error type.
 data ProtocolClientError err
  = -- | Correctly parsed SMP server ERR response.
@@ -1368,22 +1354,20 @@ 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, sentCommands}, thParams = THandleParams {blockSize, serviceAuth}} nm nonce_ tOut pKey entId cmd =
+sendProtocolCommand_ c@ProtocolClient {client_ = PClient {sndQ}, 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@Request {corrId} = case t_ of
-      Left e -> notSent e
+    sendRecv t_ r = case t_ of
+      Left e -> pure . Left $ PCETransportError e
      Right t
-        | B.length s > blockSize - 2 -> notSent TELargeMsg
+        | B.length s > blockSize - 2 -> pure . Left $ PCETransportError 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
@@ -1391,7 +1375,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, sentCommands, msgQ}} nm tOut Request {corrId, entityId, pending, responseVar} = do
+getResponse ProtocolClient {client_ = PClient {tcpTimeout, timeoutErrorCount}} nm tOut Request {entityId, pending, responseVar} = do
  r <- fromMaybe (netTimeoutInt tcpTimeout nm) tOut `timeout` atomically (takeTMVar responseVar)
  response <- atomically $ do
    writeTVar pending False
@@ -1400,18 +1384,14 @@ getResponse ProtocolClient {client_ = PClient {tcpTimeout, timeoutErrorCount, se
    -- See `processMsg`.
    ((r <|>) <$> tryTakeTMVar responseVar) >>= \case
      Just r' -> writeTVar timeoutErrorCount 0 $> r'
-      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
+      Nothing -> modifyTVar' timeoutErrorCount (+ 1) $> Left PCEResponseTimeout
  pure Response {entityId, response}

 mkTransmission :: Protocol v err msg => ProtocolClient v err msg ->  ClientCommand msg -> IO (PCTransmission err msg)
 mkTransmission c = mkTransmission_ c Nothing

 mkTransmission_ :: forall v err msg. Protocol v err msg => ProtocolClient v err msg -> Maybe C.CbNonce -> ClientCommand msg -> IO (PCTransmission err msg)
-mkTransmission_ ProtocolClient {thParams, client_ = PClient {clientCorrId, sentCommands, msgQ}} nonce_ (entityId, pKey_, command) = do
+mkTransmission_ ProtocolClient {thParams, client_ = PClient {clientCorrId, sentCommands}} nonce_ (entityId, pKey_, command) = do
  nonce@(C.CbNonce corrId) <- maybe (atomically $ C.randomCbNonce clientCorrId) pure nonce_
  let TransmissionForAuth {tForAuth, tToSend} = encodeTransmissionForAuth thParams (CorrId corrId, entityId, command)
      auth = authTransmission (thAuth thParams) (useServiceAuth command) pKey_ nonce tForAuth
@@ -1426,7 +1406,7 @@ mkTransmission_ ProtocolClient {thParams, client_ = PClient {clientCorrId, sentC
            Request
              { corrId,
                entityId,
-                command = command <$ msgQ,
+                command,
                pending,
                responseVar
              }
@@ -72,6 +72,7 @@ import Simplex.Messaging.Notifications.Transport
 import Simplex.Messaging.Protocol (EntityId (..), ErrorType (..), NotifierId, Party (..), ProtocolServer (host), SMPServer, ServiceSub (..), SignedTransmission, Transmission, pattern NoEntity, pattern SMPServer, encodeTransmission, tGetServer, tPut)
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Server
+import Simplex.Messaging.Server.AddressStats (AddrStats, ServerAddrStats (..), addressStatsThread, incAddrCounter, printTopAddresses, withAddrStats)
 import Simplex.Messaging.Server.Control (CPClientRole (..))
 import Simplex.Messaging.Server.Env.STM (StartOptions (..))
 import Simplex.Messaging.Server.Stats (PeriodStats (..), PeriodStatCounts (..), periodStatCounts, periodStatDataCounts, updatePeriodStats)
@@ -80,7 +81,7 @@ import Simplex.Messaging.SystemTime
 import Simplex.Messaging.TMap (TMap)
 import Simplex.Messaging.Transport (ASrvTransport, ATransport (..), THandle (..), THandleAuth (..), THandleParams (..), TProxy, Transport (..), TransportPeer (..), defaultSupportedParams)
 import Simplex.Messaging.Transport.Buffer (trimCR)
-import Simplex.Messaging.Transport.Server (AddHTTP, runTransportServer, runLocalTCPServer)
+import Simplex.Messaging.Transport.Server (AddHTTP, TLSServerCredential (..), newSocketState, runLocalTCPServer, runTransportServerState_)
 import Simplex.Messaging.Util
 import System.Environment (lookupEnv)
 import System.Exit (exitFailure, exitSuccess)
@@ -122,6 +123,7 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg, startOptions}
        : map runServer transports
          <> serverStatsThread_ cfg
          <> prometheusMetricsThread_ cfg
+          <> addressStatsThread_ cfg
          <> controlPortThread_ cfg
    )
    `finally` stopServer
@@ -131,16 +133,20 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg, startOptions}
      srvCreds <- asks tlsServerCreds
      serverSignKey <- either fail pure $ C.x509ToPrivate' $ snd srvCreds
      env <- ask
-      liftIO $ runTransportServer started tcpPort defaultSupportedParams srvCreds tCfg $ \h -> runClient serverSignKey t h `runReaderT` env
+      ss <- liftIO newSocketState
+      liftIO $ runTransportServerState_ ss started tcpPort defaultSupportedParams TLSServerCredential {credential = srvCreds, sniCredential = Nothing} tCfg $ \sock (_, h) ->
+        runClient serverSignKey t sock h `runReaderT` env

-    runClient :: Transport c => C.APrivateSignKey -> TProxy c 'TServer -> c 'TServer -> M ()
-    runClient signKey _ h = do
-      kh <- asks serverIdentity
-      ks <- atomically . C.generateKeyPair =<< asks random
-      NtfServerConfig {ntfServerVRange} <- asks config
-      liftIO (runExceptT $ ntfServerHandshake signKey h ks kh ntfServerVRange) >>= \case
-        Right th -> runNtfClientTransport th
-        Left _ -> pure ()
+    runClient :: Transport c => C.APrivateSignKey -> TProxy c 'TServer -> Socket -> c 'TServer -> M ()
+    runClient signKey _ sock h = do
+      stats_ <- asks addrStats
+      withAddrStats (addrStatsMap <$> stats_) sock $ \addrStats_ -> do
+        kh <- asks serverIdentity
+        ks <- atomically . C.generateKeyPair =<< asks random
+        NtfServerConfig {ntfServerVRange} <- asks config
+        liftIO (runExceptT $ ntfServerHandshake signKey h ks kh ntfServerVRange) >>= \case
+          Right th -> runNtfClientTransport addrStats_ th
+          Left _ -> pure ()

    stopServer :: M ()
    stopServer = do
@@ -231,6 +237,10 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg, startOptions}
      [savePrometheusMetrics interval prometheusMetricsFile]
    prometheusMetricsThread_ _ = []

+    addressStatsThread_ :: NtfServerConfig -> [M ()]
+    addressStatsThread_ NtfServerConfig {addressStats = Just statsCfg} = [asks addrStats >>= mapM_ (liftIO . addressStatsThread statsCfg)]
+    addressStatsThread_ _ = []
+
    savePrometheusMetrics :: Int -> FilePath -> M ()
    savePrometheusMetrics saveInterval metricsFile = do
      labelMyThread "savePrometheusMetrics"
@@ -256,7 +266,7 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg, startOptions}
      pure NtfServerMetrics {statsData = d, activeTokensCounts = psTkns, activeSubsCounts = psSubs, tokenCount, approxSubCount, lastNtfCount, rtsOptions}

    getNtfRealTimeMetrics :: NtfEnv -> IO NtfRealTimeMetrics
-    getNtfRealTimeMetrics NtfEnv {subscriber, pushServer} = do
+    getNtfRealTimeMetrics NtfEnv {subscriber, pushServer, addrStats} = do
 #if MIN_VERSION_base(4,18,0)
      threadsCount <- length <$> listThreads
 #else
@@ -274,6 +284,7 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg, startOptions}
      ntfPendingQueueSubs <- getSMPSubMetrics a pendingQueueSubs
      smpSessionCount <- M.size <$> readTVarIO smpSessions
      apnsPushQLength <- pushWorkersQLength pushWorkers
+      addressHistograms <- mapM (readIORef . addrHistograms) addrStats
      pure
        NtfRealTimeMetrics
          { threadsCount,
@@ -285,7 +296,8 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg, startOptions}
            ntfPendingServiceSubs,
            ntfPendingQueueSubs,
            smpSessionCount,
-            apnsPushQLength
+            apnsPushQLength,
+            addressHistograms
          }
      where
        getSMPServiceSubMetrics :: forall sub. SMPClientAgent 'NotifierService -> (SMPClientAgent 'NotifierService -> TMap SMPServer (TVar (Maybe sub))) -> (sub -> Int64) -> IO NtfSMPSubMetrics
@@ -441,7 +453,10 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg, startOptions}
                      T.hPutStrLn h $ name <> " own servers count: " <> tshow (length ownServers)
                      when (r == CPRAdmin) $ T.hPutStrLn h $ name <> " own servers: " <> T.intercalate "," ownServers
                      T.hPutStrLn h $ name <> " other servers count: " <> tshow otherServers
-              CPHelp -> hPutStrLn h "commands: stats, stats-rts, server-info, help, quit"
+              CPAddresses name n_ -> withUserRole $ do
+                stats_ <- unliftIO u $ asks addrStats
+                printTopAddresses h (addrStatsMap <$> stats_) name n_
+              CPHelp -> hPutStrLn h "commands: stats, stats-rts, server-info, addresses, help, quit"
              CPQuit -> pure ()
              CPSkip -> pure ()
              where
@@ -756,11 +771,11 @@ periodicNtfsThread s = do
    cnt <- withPeriodicNtfTokens st now $ \tkn -> unlift $ pushNotification s Nothing False tkn PNCheckMessages
    logNote $ "Scheduled periodic notifications: " <> tshow cnt

-runNtfClientTransport :: Transport c => THandleNTF c 'TServer -> M ()
-runNtfClientTransport th@THandle {params} = do
+runNtfClientTransport :: Transport c => Maybe (AddrStats NtfAddrCounter) -> THandleNTF c 'TServer -> M ()
+runNtfClientTransport addrStats_ th@THandle {params} = do
  qSize <- asks $ clientQSize . config
  ts <- liftIO getSystemTime
-  c <- liftIO $ newNtfServerClient qSize params ts
+  c <- liftIO $ newNtfServerClient qSize params ts addrStats_
  s <- asks subscriber
  ps <- asks pushServer
  expCfg <- asks $ inactiveClientExpiration . config
@@ -775,7 +790,7 @@ clientDisconnected :: NtfServerClient -> IO ()
 clientDisconnected NtfServerClient {connected} = atomically $ writeTVar connected False

 receive :: Transport c => NtfPostgresStore -> THandleNTF c 'TServer -> NtfServerClient -> IO ()
-receive st th@THandle {params = THandleParams {thAuth}} NtfServerClient {rcvQ, sndQ, rcvActiveAt} = forever $ do
+receive st th@THandle {params = THandleParams {thAuth}} NtfServerClient {rcvQ, sndQ, rcvActiveAt, ntfClientAddrStats} = forever $ do
  ts <- L.toList <$> tGetServer th
  atomically . (writeTVar rcvActiveAt $!) =<< getSystemTime
  (errs, cmds) <- partitionEithers <$> mapM cmdAction ts
@@ -784,17 +799,21 @@ receive st th@THandle {params = THandleParams {thAuth}} NtfServerClient {rcvQ, s
  where
    cmdAction = \case
      Left (corrId, entId, e) -> do
+        incAddrStats NACErrors
        logError $ "invalid client request: " <> tshow e
        pure $ Left (corrId, entId, NRErr e)
-      Right t@(_, _, (corrId, entId, _)) ->
+      Right t@(_, _, (corrId, entId, cmd)) -> do
+        incAddrStats $ ntfCmdCounter cmd
        verified =<< verifyNtfTransmission st thAuth t
        where
          verified = \case
            VRVerified req -> pure $ Right req
            VRFailed e -> do
+              incAddrStats NACErrors
              logError "unauthorized client request"
              pure $ Left (corrId, entId, NRErr e)
    write q = mapM_ (atomically . writeTBQueue q) . L.nonEmpty
+    incAddrStats c = forM_ ntfClientAddrStats (`incAddrCounter` c)

 send :: Transport c => THandleNTF c 'TServer -> NtfServerClient -> IO ()
 send h@THandle {params} NtfServerClient {sndQ, sndActiveAt} = forever $ do
@@ -7,14 +7,17 @@ module Simplex.Messaging.Notifications.Server.Control
 where

 import qualified Data.Attoparsec.ByteString.Char8 as A
+import Data.Text (Text)
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (BasicAuth)
+import Simplex.Messaging.Server.AddressStats (addressesArgs, addressesArgsP)

 data ControlProtocol
  = CPAuth BasicAuth
  | CPStats
  | CPStatsRTS
  | CPServerInfo
+  | CPAddresses Text (Maybe Int)
  | CPHelp
  | CPQuit
  | CPSkip
@@ -25,6 +28,7 @@ instance StrEncoding ControlProtocol where
    CPStats -> "stats"
    CPStatsRTS -> "stats-rts"
    CPServerInfo -> "server-info"
+    CPAddresses name n_ -> "addresses " <> addressesArgs name n_
    CPHelp -> "help"
    CPQuit -> "quit"
    CPSkip -> ""
@@ -34,6 +38,7 @@ instance StrEncoding ControlProtocol where
      "stats" -> pure CPStats
      "stats-rts" -> pure CPStatsRTS
      "server-info" -> pure CPServerInfo
+      "addresses" -> uncurry CPAddresses <$> addressesArgsP
      "help" -> pure CPHelp
      "quit" -> pure CPQuit
      "" -> pure CPSkip
@@ -19,6 +19,8 @@ module Simplex.Messaging.Notifications.Server.Env
    PushWorkerVar,
    NtfRequest (..),
    NtfServerClient (..),
+    NtfAddrCounter (..),
+    ntfCmdCounter,
    defaultInactiveClientExpiration,
    newNtfServerEnv,
    newNtfSubscriber,
@@ -55,6 +57,7 @@ import Simplex.Messaging.Notifications.Server.Store.Postgres
 import Simplex.Messaging.Notifications.Server.Store.Types
 import Simplex.Messaging.Notifications.Transport (NTFVersion, VersionRangeNTF)
 import Simplex.Messaging.Protocol (BasicAuth, CorrId, Party (..), SMPServer, SParty (..), ServiceId, Transmission)
+import Simplex.Messaging.Server.AddressStats (AddrCounter (..), AddrStats, AddressStatsConfig, ServerAddrStats, newServerAddrStats)
 import Simplex.Messaging.Server.Env.STM (StartOptions (..))
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.QueueStore.Postgres.Config (PostgresStoreCfg (..))
@@ -97,6 +100,7 @@ data NtfServerConfig = NtfServerConfig
    -- | interval and file to save prometheus metrics
    prometheusInterval :: Maybe Int,
    prometheusMetricsFile :: FilePath,
+    addressStats :: Maybe AddressStatsConfig,
    ntfServerVRange :: VersionRangeNTF,
    transportConfig :: TransportServerConfig,
    startOptions :: StartOptions
@@ -117,11 +121,56 @@ data NtfEnv = NtfEnv
    random :: TVar ChaChaDRG,
    tlsServerCreds :: TLS.Credential,
    serverIdentity :: C.KeyHash,
-    serverStats :: NtfServerStats
+    serverStats :: NtfServerStats,
+    addrStats :: Maybe (ServerAddrStats NtfAddrCounter)
  }

+data NtfAddrCounter
+  = NACConnections
+  | NACErrors
+  | NACTNew
+  | NACTVfy
+  | NACTChk
+  | NACTRpl
+  | NACTDel
+  | NACTCrn
+  | NACSNew
+  | NACSChk
+  | NACSDel
+  | NACPing
+  deriving (Eq, Ord, Enum, Bounded, Show)
+
+instance AddrCounter NtfAddrCounter where
+  counterName = \case
+    NACConnections -> "connections"
+    NACErrors -> "errors"
+    NACTNew -> "TNEW"
+    NACTVfy -> "TVFY"
+    NACTChk -> "TCHK"
+    NACTRpl -> "TRPL"
+    NACTDel -> "TDEL"
+    NACTCrn -> "TCRN"
+    NACSNew -> "SNEW"
+    NACSChk -> "SCHK"
+    NACSDel -> "SDEL"
+    NACPing -> "PING"
+  connectionsCounter = NACConnections
+
+ntfCmdCounter :: NtfCmd -> NtfAddrCounter
+ntfCmdCounter (NtfCmd _ cmd) = case cmd of
+  TNEW _ -> NACTNew
+  TVFY _ -> NACTVfy
+  TCHK -> NACTChk
+  TRPL _ -> NACTRpl
+  TDEL -> NACTDel
+  TCRN _ -> NACTCrn
+  SNEW _ -> NACSNew
+  SCHK -> NACSChk
+  SDEL -> NACSDel
+  PING -> NACPing
+
 newNtfServerEnv :: NtfServerConfig -> IO NtfEnv
-newNtfServerEnv config@NtfServerConfig {pushQSize, smpAgentCfg, apnsConfig, dbStoreConfig, ntfCredentials, useServiceCreds} = do
+newNtfServerEnv config@NtfServerConfig {pushQSize, smpAgentCfg, apnsConfig, dbStoreConfig, ntfCredentials, useServiceCreds, addressStats} = do
  random <- C.newRandom
  store <- newNtfDbStore dbStoreConfig
  tlsServerCreds <- loadServerCredential ntfCredentials
@@ -130,7 +179,8 @@ newNtfServerEnv config@NtfServerConfig {pushQSize, smpAgentCfg, apnsConfig, dbSt
  subscriber <- newNtfSubscriber smpAgentCfg dbService random
  pushServer <- newNtfPushServer pushQSize apnsConfig
  serverStats <- newNtfServerStats =<< getCurrentTime
-  pure NtfEnv {config, subscriber, pushServer, store, random, tlsServerCreds, serverIdentity = C.KeyHash fp, serverStats}
+  addrStats <- mapM (const newServerAddrStats) addressStats
+  pure NtfEnv {config, subscriber, pushServer, store, random, tlsServerCreds, serverIdentity = C.KeyHash fp, serverStats, addrStats}
  where
    mkDbService g st = DBService {getCredentials, updateServiceId}
      where
@@ -242,14 +292,15 @@ data NtfServerClient = NtfServerClient
    ntfThParams :: THandleParams NTFVersion 'TServer,
    connected :: TVar Bool,
    rcvActiveAt :: TVar SystemTime,
-    sndActiveAt :: TVar SystemTime
+    sndActiveAt :: TVar SystemTime,
+    ntfClientAddrStats :: Maybe (AddrStats NtfAddrCounter)
  }

-newNtfServerClient :: Natural -> THandleParams NTFVersion 'TServer -> SystemTime -> IO NtfServerClient
-newNtfServerClient qSize ntfThParams ts = do
+newNtfServerClient :: Natural -> THandleParams NTFVersion 'TServer -> SystemTime -> Maybe (AddrStats NtfAddrCounter) -> IO NtfServerClient
+newNtfServerClient qSize ntfThParams ts ntfClientAddrStats = do
  rcvQ <- newTBQueueIO qSize
  sndQ <- newTBQueueIO qSize
  connected <- newTVarIO True
  rcvActiveAt <- newTVarIO ts
  sndActiveAt <- newTVarIO ts
-  return NtfServerClient {rcvQ, sndQ, ntfThParams, connected, rcvActiveAt, sndActiveAt}
+  return NtfServerClient {rcvQ, sndQ, ntfThParams, connected, rcvActiveAt, sndActiveAt, ntfClientAddrStats}
@@ -141,6 +141,8 @@ ntfServerCLI cfgPath logPath =
               \disconnect = off\n"
            <> ("# ttl = " <> tshow (ttl defaultInactiveClientExpiration) <> "\n")
            <> ("# check_interval = " <> tshow (checkInterval defaultInactiveClientExpiration) <> "\n")
+            <> "\n"
+            <> addressStatsIniContent
    enableStoreLog' = settingIsOn "STORE_LOG" "enable"
    runServer startOptions ini = do
      setLogLevel $ logLevel startOptions
@@ -216,6 +218,7 @@ ntfServerCLI cfgPath logPath =
              serverStatsBackupFile = logStats $> combine logPath "ntf-server-stats.log",
              prometheusInterval = eitherToMaybe $ read . T.unpack <$> lookupValue "STORE_LOG" "prometheus_interval" ini,
              prometheusMetricsFile = combine logPath "ntf-server-metrics.txt",
+              addressStats = iniAddressStats ini,
              ntfServerVRange = supportedServerNTFVRange,
              transportConfig =
                mkTransportServerConfig
@@ -21,7 +21,9 @@ import Data.Time.Clock (UTCTime (..), diffUTCTime)
 import Data.Time.Clock.System (systemEpochDay)
 import Data.Time.Format.ISO8601 (iso8601Show)
 import Numeric.Natural (Natural)
+import Simplex.Messaging.Notifications.Server.Env (NtfAddrCounter)
 import Simplex.Messaging.Notifications.Server.Stats
+import Simplex.Messaging.Server.AddressStats (AddrHistogram, addrHistogramMetrics)
 import Simplex.Messaging.Server.Stats (PeriodStatCounts (..))
 import Simplex.Messaging.Transport (simplexMQVersion)
 import Simplex.Messaging.Util (tshow)
@@ -49,7 +51,8 @@ data NtfRealTimeMetrics = NtfRealTimeMetrics
    ntfPendingServiceSubs :: NtfSMPSubMetrics,
    ntfPendingQueueSubs :: NtfSMPSubMetrics,
    smpSessionCount :: Int,
-    apnsPushQLength :: Natural
+    apnsPushQLength :: Natural,
+    addressHistograms :: Maybe (M.Map NtfAddrCounter AddrHistogram)
  }

 data NtfSMPWorkerMetrics = NtfSMPWorkerMetrics {ownServers :: [Text], otherServers :: Int}
@@ -59,7 +62,7 @@ data NtfSMPSubMetrics = NtfSMPSubMetrics {ownSrvSubs :: M.Map Text Int, otherSer
 {-# FOURMOLU_DISABLE\n#-}
 ntfPrometheusMetrics :: NtfServerMetrics -> NtfRealTimeMetrics -> UTCTime -> Text
 ntfPrometheusMetrics sm rtm ts =
-  time <> tokens <> subscriptions <> notifications <> info
+  time <> tokens <> subscriptions <> notifications <> clientAddresses <> info
  where
    NtfServerMetrics {statsData, activeTokensCounts = psTkns, activeSubsCounts = psSubs, tokenCount, approxSubCount, lastNtfCount, rtsOptions} = sm
    NtfRealTimeMetrics
@@ -72,7 +75,8 @@ ntfPrometheusMetrics sm rtm ts =
        ntfPendingServiceSubs,
        ntfPendingQueueSubs,
        smpSessionCount,
-        apnsPushQLength
+        apnsPushQLength,
+        addressHistograms
      } = rtm
    NtfServerStatsData
      { _fromTime,
@@ -221,6 +225,7 @@ ntfPrometheusMetrics sm rtm ts =
      <> showNtfsByServer _ntfReceivedAuthOwn "simplex_ntf_notifications_received_auth_own" "Received notifications without token or subscription (AUTH error)" "ntfReceivedAuthOwn"
      <> showNtfsByServer _ntfDeliveredOwn "simplex_ntf_notifications_delivered_own" "Delivered notifications" "ntfDeliveredOwn"
      <> showNtfsByServer _ntfFailedOwn "simplex_ntf_notifications_failed_own" "Failed notifications" "ntfFailedOwn"
+    clientAddresses = maybe "" (addrHistogramMetrics "simplex_ntf" tsEpoch) addressHistograms
    info =
      "# Info\n\
      \# ----\n\
@@ -241,7 +241,6 @@ 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)
@@ -1324,10 +1323,7 @@ instance ProtocolTypeI p => FromJSON (ProtocolServer p) where
  parseJSON = strParseJSON "ProtocolServer"

 newtype BasicAuth = BasicAuth {unBasicAuth :: ByteString}
-  deriving (Ord, Show)
-
-instance Eq BasicAuth where
-  BasicAuth s == BasicAuth s' = BA.constEq s s'
+  deriving (Eq, Ord, Show)

 instance IsString BasicAuth where fromString = BasicAuth . B.pack

@@ -98,12 +98,13 @@ 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 (NetworkConfig (..), ProtocolClient (thParams), ProtocolClientConfig (..), ProtocolClientError (..), SMPClient, SMPClientError, clientHandlers, closeTimedOutClient, forwardSMPTransmission, smpProxyError, temporaryClientError)
-import Simplex.Messaging.Client.Agent (OwnServer, SMPClientAgent (..), SMPClientAgentConfig (..), SMPClientAgentEvent (..), closeSMPClientAgent, getSMPServerClient'', isOwnServer, lookupSMPServerClient, getConnectedSMPServerClient)
+import Simplex.Messaging.Client (ProtocolClient (thParams), ProtocolClientError (..), SMPClient, SMPClientError, clientHandlers, forwardSMPTransmission, smpProxyError, temporaryClientError, transportHost')
+import Simplex.Messaging.Client.Agent (OwnServer, SMPClientAgent (..), SMPClientAgentEvent (..), closeSMPClientAgent, getSMPServerClient'', isOwnServer, lookupSMPServerClient, getConnectedSMPServerClient)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol
+import Simplex.Messaging.Server.AddressStats (AddrStats, ServerAddrStats (..), addAddrCounter, addressStatsThread, incAddrCounter, printTopAddresses, withAddrStats)
 import Simplex.Messaging.Server.Control
 import Simplex.Messaging.Server.Env.STM as Env
 import Simplex.Messaging.Server.Expiration
@@ -124,6 +125,7 @@ import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Buffer (trimCR)
+import Simplex.Messaging.Transport.Client (TransportHost (..))
 import Simplex.Messaging.Transport.Server
 import Simplex.Messaging.Util
 import Simplex.Messaging.Version
@@ -199,6 +201,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
            <> expireMessagesThread_ cfg
            <> serverStatsThread_ cfg
            <> prometheusMetricsThread_ cfg
+            <> addressStatsThread_ cfg
            <> controlPortThread_ cfg
    )
    `finally` stopServer s
@@ -216,11 +219,12 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
          runTransportServerState_ ss started tcpPort defaultSupportedParamsHTTPS combinedCreds tCfg $ \s (sniUsed, h) ->
            case cast h of
              Just (TLS {tlsContext} :: TLS 'TServer) | sniUsed -> labelMyThread "https client" >> attachHTTP s tlsContext
-              _ -> runClient srvCert srvSignKey t h `runReaderT` env
+              _ -> runClient srvCert srvSignKey t s h `runReaderT` env
          where
            combinedCreds = TLSServerCredential {credential = smpCreds, sniCredential = Just httpCreds}
        _ ->
-          runTransportServerState ss started tcpPort defaultSupportedParams smpCreds tCfg $ \h -> runClient srvCert srvSignKey t h `runReaderT` env
+          runTransportServerState_ ss started tcpPort defaultSupportedParams TLSServerCredential {credential = smpCreds, sniCredential = Nothing} tCfg $ \s (_, h) ->
+            runClient srvCert srvSignKey t s h `runReaderT` env

    sigIntHandlerThread :: M s ()
    sigIntHandlerThread = do
@@ -515,6 +519,10 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
      [logServerStats logStatsStartTime interval serverStatsLogFile]
    serverStatsThread_ _ = []

+    addressStatsThread_ :: ServerConfig s -> [M s ()]
+    addressStatsThread_ ServerConfig {addressStats = Just statsCfg} = [asks addrStats >>= mapM_ (liftIO . addressStatsThread statsCfg)]
+    addressStatsThread_ _ = []
+
    logServerStats :: Int64 -> Int64 -> FilePath -> M s ()
    logServerStats startAt logInterval statsFilePath = do
      labelMyThread "logServerStats"
@@ -699,7 +707,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
      pure ServerMetrics {statsData = d, activeQueueCounts = ps, activeNtfCounts = psNtf, entityCounts, rtsOptions}

    getRealTimeMetrics :: Env s -> IO RealTimeMetrics
-    getRealTimeMetrics Env {sockets, msgStore_ = ms, server = srv@Server {subscribers, ntfSubscribers}} = do
+    getRealTimeMetrics Env {sockets, msgStore_ = ms, server = srv@Server {subscribers, ntfSubscribers}, addrStats} = do
      socketStats <- mapM (traverse getSocketStats) =<< readTVarIO sockets
 #if MIN_VERSION_base(4,18,0)
      threadsCount <- length <$> listThreads
@@ -711,7 +719,8 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
      smpSubs <- getSubscribersMetrics subscribers
      ntfSubs <- getSubscribersMetrics ntfSubscribers
      loadedCounts <- loadedQueueCounts $ fromMsgStore ms
-      pure RealTimeMetrics {socketStats, threadsCount, clientsCount, deliveredSubs, deliveredTimes, smpSubs, ntfSubs, loadedCounts}
+      addressHistograms <- mapM (readIORef . addrHistograms) addrStats
+      pure RealTimeMetrics {socketStats, threadsCount, clientsCount, deliveredSubs, deliveredTimes, smpSubs, ntfSubs, loadedCounts, addressHistograms}
      where
        getSubscribersMetrics ServerSubscribers {queueSubscribers, serviceSubscribers, totalServiceSubs, subClients} = do
          subsCount <- M.size <$> getSubscribedClients queueSubscribers
@@ -732,19 +741,21 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
                Nothing -> acc
                Just (_, ts) -> (cnt + 1, updateTimeBuckets ts ts' times)

-    runClient :: Transport c => X.CertificateChain -> C.APrivateSignKey -> TProxy c 'TServer -> c 'TServer -> M s ()
-    runClient srvCert srvSignKey tp h = do
-      ms <- asks msgStore
-      g <- asks random
-      idSize <- asks $ queueIdBytes . config
-      kh <- asks serverIdentity
-      ks <- atomically . C.generateKeyPair =<< asks random
-      ServerConfig {smpServerVRange, smpHandshakeTimeout, information} <- asks config
-      let serverInfo = LB.toStrict . J.encode <$> information
-      labelMyThread $ "smp handshake for " <> transportName tp
-      liftIO (timeout smpHandshakeTimeout . runExceptT $ smpServerHandshake srvCert srvSignKey h ks kh smpServerVRange serverInfo $ getClientService ms g idSize) >>= \case
-        Just (Right th) -> runClientTransport th
-        _ -> pure ()
+    runClient :: Transport c => X.CertificateChain -> C.APrivateSignKey -> TProxy c 'TServer -> Socket -> c 'TServer -> M s ()
+    runClient srvCert srvSignKey tp sock h = do
+      stats_ <- asks addrStats
+      withAddrStats (addrStatsMap <$> stats_) sock $ \addrStats_ -> do
+        ms <- asks msgStore
+        g <- asks random
+        idSize <- asks $ queueIdBytes . config
+        kh <- asks serverIdentity
+        ks <- atomically . C.generateKeyPair =<< asks random
+        ServerConfig {smpServerVRange, smpHandshakeTimeout, information} <- asks config
+        let serverInfo = LB.toStrict . J.encode <$> information
+        labelMyThread $ "smp handshake for " <> transportName tp
+        liftIO (timeout smpHandshakeTimeout . runExceptT $ smpServerHandshake srvCert srvSignKey h ks kh smpServerVRange serverInfo $ getClientService ms g idSize) >>= \case
+          Just (Right th) -> runClientTransport addrStats_ th
+          _ -> pure ()

    getClientService :: s -> TVar ChaChaDRG -> Int -> SMPServiceRole -> X.CertificateChain -> XV.Fingerprint -> ExceptT TransportError IO ServiceId
    getClientService ms g idSize role cert fp = do
@@ -1040,7 +1051,10 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
                hPutStrLn h "saving server state..."
                unliftIO u $ saveServer False
                hPutStrLn h "server state saved!"
-              CPHelp -> hPutStrLn h "commands: stats, stats-rts, clients, sockets, socket-threads, threads, server-info, delete, save, help, quit"
+              CPAddresses name n_ -> withUserRole $ do
+                stats_ <- unliftIO u $ asks addrStats
+                printTopAddresses h (addrStatsMap <$> stats_) name n_
+              CPHelp -> hPutStrLn h "commands: stats, stats-rts, clients, sockets, socket-threads, threads, server-info, delete, addresses, save, help, quit"
              CPQuit -> pure ()
              CPSkip -> pure ()
              where
@@ -1061,13 +1075,13 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
                    logError "Unauthorized control port command"
                    hPutStrLn h "AUTH"

-runClientTransport :: forall c s. (Transport c, MsgStoreClass s) => THandleSMP c 'TServer -> M s ()
-runClientTransport h@THandle {params = thParams@THandleParams {sessionId}} = do
+runClientTransport :: forall c s. (Transport c, MsgStoreClass s) => Maybe (AddrStats SMPAddrCounter) -> THandleSMP c 'TServer -> M s ()
+runClientTransport addrStats_ h@THandle {params = thParams@THandleParams {sessionId}} = do
  q <- asks $ tbqSize . config
  ts <- liftIO getSystemTime
  nextClientId <- asks clientSeq
  clientId <- atomically $ stateTVar nextClientId $ \next -> (next, next + 1)
-  c <- liftIO $ newClient clientId q thParams ts
+  c <- liftIO $ newClient clientId q thParams ts addrStats_
  runClientThreads c `finally` clientDisconnected c
  where
    runClientThreads :: Client s -> M s ()
@@ -1092,7 +1106,6 @@ 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
@@ -1145,7 +1158,7 @@ cancelSub s = case subThread s of
 type VerifiedTransmissionOrError s = Either (Transmission BrokerMsg) (VerifiedTransmission s)

 receive :: forall c s. (Transport c, MsgStoreClass s) => THandleSMP c 'TServer -> s -> Client s -> M s ()
-receive h@THandle {params = THandleParams {thAuth, sessionId}} ms Client {rcvQ, sndQ, rcvActiveAt} = do
+receive h@THandle {params = THandleParams {thAuth, sessionId}} ms Client {rcvQ, sndQ, rcvActiveAt, clientAddrStats} = do
  labelMyThread . B.unpack $ "client $" <> encode sessionId <> " receive"
  sa <- asks serverActive
  stats <- asks serverStats
@@ -1155,6 +1168,9 @@ receive h@THandle {params = THandleParams {thAuth, sessionId}} ms Client {rcvQ,
    atomically . (writeTVar rcvActiveAt $!) =<< getSystemTime
    let (es, ts') = partitionEithers $ L.toList ts
        errs = map (second ERR) es
+    forM_ clientAddrStats $ \s -> do
+      unless (null es) $ addAddrCounter s SACErrors $ length es
+      forM_ ts' $ \(_, _, (_, _, cmd)) -> incAddrCounter s $ smpCmdCounter cmd
    errs' <- case ts' of
      (_, _, (_, _, Cmd p cmd)) : rest -> do
        let service = peerClientService =<< thAuth
@@ -1182,7 +1198,9 @@ receive h@THandle {params = THandleParams {thAuth, sessionId}} ms Client {rcvQ,
    verified :: ServerStats -> SignedTransmission Cmd -> VerificationResult s -> IO (VerifiedTransmissionOrError s)
    verified stats (_, _, t@(corrId, entId, Cmd _ command)) = \case
      VRVerified q -> pure $ Right (q, t)
-      VRFailed e -> Left (corrId, entId, ERR e) <$ when (e == AUTH) incAuthStat
+      VRFailed e -> do
+        forM_ clientAddrStats (`incAddrCounter` SACErrors)
+        Left (corrId, entId, ERR e) <$ when (e == AUTH) incAuthStat
        where
          incAuthStat = case command of
            SEND {} -> incStat $ msgSentAuth stats
@@ -1377,7 +1395,7 @@ client :: forall s. MsgStoreClass s => Server s -> s -> Client s -> M s ()
 client
  Server {subscribers, ntfSubscribers}
  ms
-  clnt@Client {clientId, rcvQ, sndQ, msgQ, clientTHParams = thParams'@THandleParams {sessionId}, procThreads} = do
+  clnt@Client {clientId, rcvQ, sndQ, msgQ, clientTHParams = thParams'@THandleParams {sessionId}, procThreads, clientAddrStats} = do
    labelMyThread . B.unpack $ "client $" <> encode sessionId <> " commands"
    let clntServiceId = (\THClientService {serviceId} -> serviceId) <$> (peerClientService =<< thAuth thParams')
        process batchSubs t acc@(rs, msgs) =
@@ -1422,25 +1440,30 @@ client
          getRelay = do
            ProxyAgent {smpAgent = a} <- asks proxyAgent
            liftIO (getConnectedSMPServerClient a srv) >>= \case
-              Just r -> Just <$> proxyServerResponse a r
+              Just r -> Just <$> proxyServerResponse a SACPrxyConnected r
              Nothing ->
                forkProxiedCmd $
                  liftIO (runExceptT (getSMPServerClient'' a srv) `E.catches` clientHandlers)
-                    >>= proxyServerResponse a
-          proxyServerResponse :: SMPClientAgent 'Sender -> Either SMPClientError (OwnServer, SMPClient) -> M s BrokerMsg
-          proxyServerResponse a smp_ = do
+                    >>= proxyServerResponse a SACPrxyNew
+          proxyServerResponse :: SMPClientAgent 'Sender -> SMPAddrCounter -> Either SMPClientError (OwnServer, SMPClient) -> M s BrokerMsg
+          proxyServerResponse a otherCounter smp_ = do
            ServerStats {pRelays, pRelaysOwn} <- asks serverStats
            let inc = mkIncProxyStats pRelays pRelaysOwn
            case smp_ of
              Right (own, smp) -> do
                inc own pRequests
                case proxyResp smp of
-                  r@PKEY {} -> r <$ inc own pSuccesses
-                  r -> r <$ inc own pErrorsCompat
+                  r@PKEY {} -> do
+                    incAddrStats $ proxyDestCounter own smp SACPrxyOwn SACPrxyOnion otherCounter
+                    r <$ inc own pSuccesses
+                  r -> do
+                    incAddrStats SACPrxyFailed
+                    r <$ inc own pErrorsCompat
              Left e -> do
                let own = isOwnServer a srv
                inc own pRequests
                inc own $ if temporaryClientError e then pErrorsConnect else pErrorsOther
+                incAddrStats SACPrxyFailed
                logWarn $ "Error connecting: " <> decodeLatin1 (strEncode $ host srv) <> " " <> tshow e
                pure . ERR $ smpProxyError e
            where
@@ -1454,47 +1477,38 @@ client
                        Nothing -> ERR $ transportErr TENoServerAuth
                      _ -> ERR $ transportErr TEVersion
      PFWD fwdV pubKey encBlock -> do
-        ProxyAgent {smpAgent = a, relayForwards} <- asks proxyAgent
+        ProxyAgent {smpAgent = a} <- asks proxyAgent
        ServerStats {pMsgFwds, pMsgFwdsOwn} <- asks serverStats
        let inc = mkIncProxyStats pMsgFwds pMsgFwdsOwn
        liftIO (lookupSMPServerClient a sessId) >>= \case
          Just (own, smp) -> do
            inc own pRequests
-            maxFwds <- asks $ proxyRelayConcurrency . config
-            -- reserved on the forked thread, so that the slot is released when the forward completes
-            forkProxiedCmd $
-              bracket (atomically $ reserve maxFwds relayForwards) (\r -> when r $ atomically $ release relayForwards) $ \case
-                -- the relay is not keeping up, the client retries TIMEOUT later
-                False -> ERR (smpProxyError PCEResponseTimeout) <$ inc own pErrorsOther
-                True ->
-                  liftIO (runExceptT (forwardSMPTransmission smp corrId fwdV pubKey encBlock) `E.catches` clientHandlers) >>= \case
-                    Right r -> PRES r <$ inc own pSuccesses
-                    Left e -> ERR (smpProxyError e) <$ case e of
-                      PCEProtocolError {} -> inc own pSuccesses
-                      PCEResponseTimeout -> do
-                        inc own pErrorsOther
-                        liftIO $ closeTimedOutClient (smpPingCount $ networkConfig $ smpCfg $ agentCfg a) smp
-                      _ -> inc own pErrorsOther
-          Nothing -> inc False pRequests >> inc False pErrorsConnect $> Just (ERR $ PROXY NO_SESSION)
+            let delivered = incAddrStats (proxyDestCounter own smp SACPfwdOwn SACPfwdOnion SACPfwdOther) >> inc own pSuccesses
+            forkProxiedCmd $ do
+              liftIO (runExceptT (forwardSMPTransmission smp corrId fwdV pubKey encBlock) `E.catches` clientHandlers)  >>= \case
+                Right r -> PRES r <$ delivered
+                Left e -> ERR (smpProxyError e) <$ case e of
+                  PCEProtocolError {} -> delivered
+                  _ -> incAddrStats SACPfwdFailed >> inc own pErrorsOther
+          Nothing -> incAddrStats SACPfwdFailed >> inc False pRequests >> inc False pErrorsConnect $> Just (ERR $ PROXY NO_SESSION)
      where
        forkProxiedCmd :: M s BrokerMsg -> M s (Maybe BrokerMsg)
        forkProxiedCmd = forkCmd serverClientConcurrency corrId (EntityId sessId)
-        reserve maxFwds v = stateTVar v $ \fwds -> case M.lookup sessId fwds of
-          Just n | n >= maxFwds -> (False, fwds)
-          n_ -> (True, M.insert sessId (maybe 1 (+ 1) n_) fwds)
-        release v = modifyTVar' v $ M.update (\n -> if n > 1 then Just (n - 1) else Nothing) sessId
+        proxyDestCounter :: OwnServer -> SMPClient -> SMPAddrCounter -> SMPAddrCounter -> SMPAddrCounter -> SMPAddrCounter
+        proxyDestCounter own smp ownCounter onionCounter otherCounter
+          | own = ownCounter
+          | THOnionHost _ <- transportHost' smp = onionCounter
+          | otherwise = otherCounter
+    incAddrStats :: SMPAddrCounter -> M s ()
+    incAddrStats c = liftIO $ forM_ clientAddrStats (`incAddrCounter` c)
    -- 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
-      -- 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
+      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)], [])
      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
@@ -2145,7 +2159,6 @@ 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
@@ -2164,26 +2177,22 @@ client
                r2 <- liftEitherWith (const BLOCK) $ EncResponse <$> C.cbEncrypt clientSecret (C.reverseNonce clientNonce) r' paddedProxiedTLength
                let fr = FwdResponse {fwdCorrId, fwdResponse = r2}
                pure $ RRES $ EncFwdResponse $ C.cbEncryptNoPad sessSecret (C.reverseNonce proxyNonce) (smpEncode fr)
-          let forwarded =
-                lift (rejectOrVerify clntThAuth t') >>= \case
-                  -- rejectOrVerify filters allowed commands, no need to repeat it here.
-                  Left r -> pure r
-                  Right t''@(_, (corrId', entId', cmd')) -> case cmd' of
-                    Cmd SResolver (RSLV d) ->
-                      lift $
-                        rslvNamesEnv >>= \case
-                          Nothing -> pure (corrId', entId', ERR (NAME NO_RESOLVER))
-                          Just nenv -> (corrId',entId',) <$> resolveNameMsg (thVersion clntTHParams) nenv d
-                    -- INTERNAL because processCommand never returns Nothing for sender commands;
-                    -- `fst` drops the empty message only returned for SUB.
-                    _ -> maybe (corrId', entId', ERR INTERNAL) fst <$> lift (processCommand Nothing (Right (M.empty, M.empty, M.empty)) t'')
-              -- commands of all clients of a proxy share its connection, so they are processed concurrently
-              concurrency = case t' of
-                Right (_, _, (_, _, Cmd SResolver _)) -> serverResolverConcurrency
-                _ -> serverClientConcurrency
+          -- the inner response, or Nothing if forked (RSLV).
+          r_ <- lift (rejectOrVerify clntThAuth t') >>= \case
+            -- rejectOrVerify filters allowed commands, no need to repeat it here.
+            Left r -> pure $ Just r
+            Right t''@(_, (corrId', entId', cmd')) -> case cmd' of
+              Cmd SResolver (RSLV d) -> lift $ rslvNamesEnv >>= \case
+                Nothing -> pure $ Just (corrId', entId', ERR (NAME NO_RESOLVER))
+                Just nenv -> forkCmd serverResolverConcurrency corrId NoEntity $ do
+                  msg <- resolveNameMsg (thVersion clntTHParams) nenv d
+                  either ERR id <$> runExceptT (encodeResp (corrId', entId', msg))
+              -- INTERNAL because processCommand never returns Nothing for sender commands;
+              -- `fst` drops the empty message only returned for SUB.
+              _ -> Just . maybe (corrId', entId', ERR INTERNAL) fst <$> lift (processCommand Nothing (Right (M.empty, M.empty, M.empty)) t'')
          stats <- asks serverStats
          incStat $ pMsgFwdsRecv stats
-          lift $ forkCmd concurrency corrId NoEntity $ either ERR id <$> runExceptT (forwarded >>= encodeResp)
+          traverse encodeResp r_
          where
            rejectOrVerify :: Maybe (THandleAuth 'TServer) -> SignedTransmissionOrError ErrorType Cmd -> M s (VerifiedTransmissionOrError s)
            rejectOrVerify clntThAuth = \case
@@ -0,0 +1,250 @@
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE TupleSections #-}
+
+module Simplex.Messaging.Server.AddressStats
+  ( AddrKey (..),
+    AddrCounter (..),
+    AddrStats (..),
+    AddrStatsMap,
+    AddrHistogram (..),
+    ServerAddrStats (..),
+    AddressStatsConfig (..),
+    defaultAddressStatsPeriod,
+    countBounds,
+    kilobyteBounds,
+    addrKey,
+    addrKeyText,
+    counterByName,
+    newServerAddrStats,
+    withAddrStats,
+    openAddrStats,
+    closeAddrStats,
+    incAddrCounter,
+    addAddrCounter,
+    rolloverAddrStats,
+    addressStatsThread,
+    topAddresses,
+    printTopAddresses,
+    addressesArgsP,
+    addressesArgs,
+    addrHistogramMetrics,
+  )
+where
+
+import Control.Applicative (optional)
+import Control.Monad
+import Control.Monad.IO.Unlift
+import qualified Data.Attoparsec.ByteString.Char8 as A
+import Data.Bits (shiftL, shiftR, (.&.), (.|.))
+import Data.ByteString.Char8 (ByteString)
+import Data.IORef
+import Data.Int (Int64)
+import qualified Data.IntMap.Strict as IM
+import Data.List (find, foldl', sortOn)
+import Data.Map.Strict (Map)
+import qualified Data.Map.Strict as M
+import Data.Maybe (fromMaybe)
+import Data.Ord (Down (..))
+import Data.Text (Text)
+import qualified Data.Text as T
+import Data.Text.Encoding (encodeUtf8)
+import qualified Data.Text.IO as T
+import Data.Word (Word32, Word64)
+import Network.Socket (SockAddr (..), Socket, getPeerName, hostAddress6ToTuple, hostAddressToTuple)
+import Numeric (showHex)
+import Simplex.Messaging.TMap (TMap)
+import qualified Simplex.Messaging.TMap as TM
+import Simplex.Messaging.Util (atomicModifyIORef'_, bshow, labelMyThread, safeDecodeUtf8, threadDelay', tshow, whenM, (<$$>))
+import System.IO (Handle, hPutStrLn)
+import UnliftIO.Exception (finally, tryAny)
+import UnliftIO.STM
+
+data AddrKey
+  = AKIPv4 Word32
+  | AKIPv6 Word64
+  deriving (Eq, Ord, Show)
+
+class (Ord c, Enum c, Bounded c) => AddrCounter c where
+  counterName :: c -> Text
+  counterBounds :: c -> [Int]
+  counterBounds _ = countBounds
+  connectionsCounter :: c
+
+data AddrStats c = AddrStats
+  { connectionsCount :: TVar Int,
+    current :: Map c (IORef Int),
+    previous :: Map c (IORef Int)
+  }
+
+type AddrStatsMap c = TMap AddrKey (AddrStats c)
+
+data AddrHistogram = AddrHistogram
+  { bucketCounts :: [Int],
+    countSum :: Int,
+    addressCount :: Int,
+    periodMax :: Int
+  }
+  deriving (Eq, Show)
+
+data ServerAddrStats c = ServerAddrStats
+  { addrStatsMap :: AddrStatsMap c,
+    addrHistograms :: IORef (Map c AddrHistogram)
+  }
+
+newtype AddressStatsConfig = AddressStatsConfig
+  { period :: Int64
+  }
+  deriving (Show)
+
+defaultAddressStatsPeriod :: Int64
+defaultAddressStatsPeriod = 300
+
+countBounds :: [Int]
+countBounds = r10Bounds 5
+
+kilobyteBounds :: [Int]
+kilobyteBounds = r10Bounds 7
+
+r10Bounds :: Int -> [Int]
+r10Bounds decades =
+  [1 .. 9]
+    <> [(r * 10 ^ d + 50) `div` 100 | d <- [1 .. decades], r <- [100, 125, 160, 200, 250, 315, 400, 500, 630, 800]]
+    <> [10 ^ (decades + 1)]
+
+addrKey :: SockAddr -> Maybe AddrKey
+addrKey = \case
+  SockAddrInet _ a ->
+    let (b1, b2, b3, b4) = hostAddressToTuple a
+     in Just $ AKIPv4 $ foldl' (\acc b -> acc `shiftL` 8 .|. fromIntegral b) 0 [b1, b2, b3, b4]
+  SockAddrInet6 _ _ a _ -> Just $ case hostAddress6ToTuple a of
+    (0, 0, 0, 0, 0, 0xffff, w1, w2) -> AKIPv4 $ fromIntegral w1 `shiftL` 16 .|. fromIntegral w2
+    (w1, w2, w3, w4, _, _, _, _) -> AKIPv6 $ foldl' (\acc w -> acc `shiftL` 16 .|. fromIntegral w) 0 [w1, w2, w3, w4]
+  _ -> Nothing
+
+addrKeyText :: AddrKey -> Text
+addrKeyText = \case
+  AKIPv4 a -> T.intercalate "." $ map (\s -> tshow $ (a `shiftR` s) .&. 0xff) [24, 16, 8, 0]
+  AKIPv6 p -> T.intercalate ":" (map (\s -> T.pack $ showHex ((p `shiftR` s) .&. 0xffff) "") [48, 32, 16, 0]) <> "::/64"
+
+counterByName :: AddrCounter c => Text -> Maybe c
+counterByName name = find ((name ==) . counterName) [minBound .. maxBound]
+
+newServerAddrStats :: AddrCounter c => IO (ServerAddrStats c)
+newServerAddrStats = ServerAddrStats <$> TM.emptyIO <*> newIORef (M.fromList $ map (\c -> (c, emptyHistogram c)) [minBound .. maxBound])
+  where
+    emptyHistogram c = AddrHistogram {bucketCounts = map (const 0) (counterBounds c), countSum = 0, addressCount = 0, periodMax = 0}
+
+newAddrStats :: AddrCounter c => IO (AddrStats c)
+newAddrStats = AddrStats <$> newTVarIO 0 <*> newCells <*> newCells
+  where
+    newCells = M.fromList <$> mapM (\c -> (c,) <$> newIORef 0) [minBound .. maxBound]
+
+withAddrStats :: (AddrCounter c, MonadUnliftIO m) => Maybe (AddrStatsMap c) -> Socket -> (Maybe (AddrStats c) -> m a) -> m a
+withAddrStats m_ sock action = case m_ of
+  Just m ->
+    liftIO (addrKey <$$> tryAny (getPeerName sock)) >>= \case
+      Right (Just k) -> do
+        s <- liftIO $ openAddrStats m k
+        action (Just s) `finally` liftIO (closeAddrStats s)
+      _ -> action Nothing
+  Nothing -> action Nothing
+
+openAddrStats :: AddrCounter c => AddrStatsMap c -> AddrKey -> IO (AddrStats c)
+openAddrStats m k = do
+  new <- maybe newAddrStats pure =<< TM.lookupIO k m
+  s <- atomically $ TM.lookup k m >>= maybe (new <$ TM.insert k new m) pure >>= \entry -> entry <$ modifyTVar' (connectionsCount entry) (+ 1)
+  s <$ incAddrCounter s connectionsCounter
+
+closeAddrStats :: AddrStats c -> IO ()
+closeAddrStats AddrStats {connectionsCount} = atomically $ modifyTVar' connectionsCount (subtract 1)
+
+incAddrCounter :: AddrCounter c => AddrStats c -> c -> IO ()
+incAddrCounter s c = addAddrCounter s c 1
+
+addAddrCounter :: AddrCounter c => AddrStats c -> c -> Int -> IO ()
+addAddrCounter AddrStats {current} c n = forM_ (M.lookup c current) $ \cell -> atomicModifyIORef'_ cell (+ n)
+
+rolloverAddrStats :: AddrCounter c => ServerAddrStats c -> IO ()
+rolloverAddrStats ServerAddrStats {addrStatsMap, addrHistograms} = do
+  entries <- M.toList <$> readTVarIO addrStatsMap
+  closingCounts <- forM entries $ \(k, s) -> do
+    counts <- closePeriod s
+    when (all (== 0) counts) $ atomically $ deleteIdle k s
+    pure counts
+  let periodCounts = M.unionsWith (<>) $ map (M.map (: []) . M.filter (> 0)) closingCounts
+  atomicModifyIORef'_ addrHistograms $ M.mapWithKey $ \c -> addPeriodCounts (counterBounds c) (M.findWithDefault [] c periodCounts)
+  where
+    closePeriod AddrStats {current, previous} = M.traverseWithKey closeCell current
+      where
+        closeCell c cell = do
+          n <- atomicModifyIORef' cell (0,)
+          forM_ (M.lookup c previous) (`writeIORef` n)
+          pure n
+    deleteIdle k AddrStats {connectionsCount} =
+      whenM ((0 ==) <$> readTVar connectionsCount) $ TM.delete k addrStatsMap
+
+addPeriodCounts :: [Int] -> [Int] -> AddrHistogram -> AddrHistogram
+addPeriodCounts bounds ns AddrHistogram {bucketCounts, countSum, addressCount} =
+  AddrHistogram
+    { bucketCounts = zipWith (+) bucketCounts $ scanl1 (+) $ map (\i -> IM.findWithDefault 0 i boundCounts) [0 .. length bounds - 1],
+      countSum = countSum + sum ns,
+      addressCount = addressCount + length ns,
+      periodMax = maximum (0 : ns)
+    }
+  where
+    boundCounts = IM.fromListWith (+) $ map (\n -> (length $ takeWhile (< n) bounds, 1)) ns
+
+addressStatsThread :: AddrCounter c => AddressStatsConfig -> ServerAddrStats c -> IO ()
+addressStatsThread AddressStatsConfig {period} stats = do
+  labelMyThread "addressStatsThread"
+  forever $ threadDelay' (period * 1000000) >> rolloverAddrStats stats
+
+topAddresses :: AddrCounter c => AddrStatsMap c -> c -> Int -> IO [(AddrKey, Int, Int)]
+topAddresses m c n = do
+  entries <- M.toList <$> readTVarIO m
+  counts <- forM entries $ \(k, AddrStats {previous, current}) -> (k,,) <$> cellValue previous <*> cellValue current
+  pure $ take n $ sortOn (\(_, p, cur) -> Down (p + cur)) $ filter (\(_, p, cur) -> p + cur > 0) counts
+  where
+    cellValue = maybe (pure 0) readIORef . M.lookup c
+
+printTopAddresses :: AddrCounter c => Handle -> Maybe (AddrStatsMap c) -> Text -> Maybe Int -> IO ()
+printTopAddresses h m_ name n_ = case m_ of
+  Just m -> case counterByName name of
+    Just c -> do
+      rows <- topAddresses m c (fromMaybe 10 n_)
+      hPutStrLn h "address,previous,current"
+      forM_ rows $ \(k, p, cur) -> T.hPutStrLn h $ addrKeyText k <> "," <> tshow p <> "," <> tshow cur
+    Nothing -> hPutStrLn h "error: unknown counter"
+  Nothing -> hPutStrLn h "error: address statistics are off"
+
+addressesArgsP :: A.Parser (Text, Maybe Int)
+addressesArgsP = (,) <$> (A.space *> (safeDecodeUtf8 <$> A.takeTill (== ' '))) <*> optional (A.space *> A.decimal)
+
+addressesArgs :: Text -> Maybe Int -> ByteString
+addressesArgs name n_ = encodeUtf8 name <> maybe "" ((" " <>) . bshow) n_
+
+addrHistogramMetrics :: AddrCounter c => Text -> Text -> Map c AddrHistogram -> Text
+addrHistogramMetrics prefix tsEpoch histograms =
+  "# Client addresses\n\
+  \# ----------------\n\
+  \\n"
+    <> "# HELP " <> countMetric <> " Counts of client addresses per period\n"
+    <> "# TYPE " <> countMetric <> " histogram\n"
+    <> T.concat (map countLines hs)
+    <> "\n# HELP " <> maxMetric <> " Maximum count of a client address in the last period\n"
+    <> "# TYPE " <> maxMetric <> " gauge\n"
+    <> T.concat (map (\(c, AddrHistogram {periodMax}) -> sample maxMetric c [] periodMax) hs)
+    <> "\n"
+  where
+    hs = M.toList histograms
+    countMetric = prefix <> "_client_address_period_count"
+    maxMetric = prefix <> "_client_address_period_max"
+    countLines (c, AddrHistogram {bucketCounts, countSum, addressCount}) =
+      T.concat (zipWith (\b cnt -> sample (countMetric <> "_bucket") c [("le", tshow b)] cnt) (counterBounds c) bucketCounts)
+        <> sample (countMetric <> "_bucket") c [("le", "+Inf")] addressCount
+        <> sample (countMetric <> "_sum") c [] countSum
+        <> sample (countMetric <> "_count") c [] addressCount
+    sample metric c labels v =
+      metric <> "{" <> T.intercalate "," (map (\(l, lv) -> l <> "=\"" <> lv <> "\"") (("counter", counterName c) : labels)) <> "} " <> tshow v <> " " <> tsEpoch <> "\n"
@@ -44,6 +44,8 @@ module Simplex.Messaging.Server.CLI
    checkSavedFingerprint,
    iniTransports,
    iniDBOptions,
+    iniAddressStats,
+    addressStatsIniContent,
    printServerConfig,
    printServerTransports,
    printSMPServerConfig,
@@ -73,13 +75,14 @@ import Simplex.Messaging.Agent.Store.Postgres.Options (DBOpts (..))
 import Simplex.Messaging.Agent.Store.Shared (MigrationConfirmation (..))
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (ProtoServerWithAuth (..), ProtocolServer (..), ProtocolTypeI)
+import Simplex.Messaging.Server.AddressStats (AddressStatsConfig (..), defaultAddressStatsPeriod)
 import Simplex.Messaging.Server.Env.STM (ServerStoreCfg (..), StartOptions (..), dbStoreCfg, storeLogFile')
 import Simplex.Messaging.Server.Main.GitCommit
 import Simplex.Messaging.Server.QueueStore.Postgres.Config (PostgresStoreCfg (..))
 import Simplex.Messaging.Transport (ASrvTransport, ATransport (..), TLS, Transport (..), simplexMQVersion)
 import Simplex.Messaging.Transport.Server (AddHTTP, loadFileFingerprint)
 import Simplex.Messaging.Transport.WebSockets (WS)
-import Simplex.Messaging.Util (eitherToMaybe, whenM)
+import Simplex.Messaging.Util (eitherToMaybe, tshow, whenM)
 import System.Directory (doesDirectoryExist, listDirectory, removeDirectoryRecursive, removePathForcibly)
 import System.Environment (lookupEnv)
 import System.Exit (exitFailure)
@@ -435,6 +438,19 @@ iniDBOptions ini _default@DBOpts {connstr, schema, poolSize} =
      createSchema = False
    }

+iniAddressStats :: Ini -> Maybe AddressStatsConfig
+iniAddressStats ini =
+  AddressStatsConfig {period = readIniDefault defaultAddressStatsPeriod "ADDRESS_STATS" "period" ini}
+    <$ settingIsOn "ADDRESS_STATS" "enable" ini
+
+addressStatsIniContent :: Text
+addressStatsIniContent =
+  "[ADDRESS_STATS]\n\
+  \# Command counts per client address, shown by the control port command `addresses`\n\
+  \# and aggregated in Prometheus metrics.\n\
+  \enable = off\n"
+    <> ("period = " <> tshow defaultAddressStatsPeriod <> "\n")
+
 printServerConfig :: String -> [(ServiceName, ASrvTransport, AddHTTP)] -> Maybe FilePath -> IO ()
 printServerConfig protocol transports logFile = do
  putStrLn $ case logFile of
@@ -7,8 +7,10 @@ module Simplex.Messaging.Server.Control
  ) where

 import qualified Data.Attoparsec.ByteString.Char8 as A
+import Data.Text (Text)
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (BasicAuth, BlockingInfo, SenderId)
+import Simplex.Messaging.Server.AddressStats (addressesArgs, addressesArgsP)

 data CPClientRole = CPRNone | CPRUser | CPRAdmin
  deriving (Eq, Show)
@@ -28,6 +30,7 @@ data ControlProtocol
  | CPStatus SenderId
  | CPBlock SenderId BlockingInfo
  | CPUnblock SenderId
+  | CPAddresses Text (Maybe Int)
  | CPSave
  | CPHelp
  | CPQuit
@@ -49,6 +52,7 @@ instance StrEncoding ControlProtocol where
    CPStatus sId -> "status " <> strEncode sId
    CPBlock sId info -> "block " <> strEncode sId <> " " <> strEncode info
    CPUnblock sId -> "unblock " <> strEncode sId
+    CPAddresses name n_ -> "addresses " <> addressesArgs name n_
    CPSave -> "save"
    CPHelp -> "help"
    CPQuit -> "quit"
@@ -69,6 +73,7 @@ instance StrEncoding ControlProtocol where
      "status" -> CPStatus <$> _strP
      "block" -> CPBlock <$> _strP <*> _strP
      "unblock" -> CPUnblock <$> _strP
+      "addresses" -> uncurry CPAddresses <$> addressesArgsP
      "save" -> pure CPSave
      "help" -> pure CPHelp
      "quit" -> pure CPQuit
@@ -31,6 +31,8 @@ module Simplex.Messaging.Server.Env.STM
    SubscribedClients,
    ProxyAgent (..),
    Client (..),
+    SMPAddrCounter (..),
+    smpCmdCounter,
    ClientId,
    ClientSub (..),
    Sub (..),
@@ -67,7 +69,6 @@ module Simplex.Messaging.Server.Env.STM
    defaultNtfExpiration,
    defaultInactiveClientExpiration,
    defaultProxyClientConcurrency,
-    defaultProxyRelayConcurrency,
    defaultNameResolverConcurrency,
    defaultMaxJournalMsgCount,
    defaultMaxJournalStateLines,
@@ -112,6 +113,7 @@ import Simplex.Messaging.Client.Agent (SMPClientAgent, SMPClientAgentConfig, new
 import Simplex.Messaging.Crypto (KeyHash (..))
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Protocol
+import Simplex.Messaging.Server.AddressStats (AddrCounter (..), AddrStats, AddressStatsConfig, ServerAddrStats, newServerAddrStats)
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.Information
 import Simplex.Messaging.Server.MsgStore.Journal
@@ -129,7 +131,7 @@ import Simplex.Messaging.Server.StoreLog.ReadWrite
 import Simplex.Messaging.SystemTime
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
-import Simplex.Messaging.Transport (ASrvTransport, SMPVersion, SessionId, THandleParams, TransportPeer (..), VersionRangeSMP)
+import Simplex.Messaging.Transport (ASrvTransport, SMPVersion, THandleParams, TransportPeer (..), VersionRangeSMP)
 import Simplex.Messaging.Transport.Server
 import Simplex.Messaging.Util (ifM, tshow, whenM, ($>>=))
 import System.Directory (doesFileExist)
@@ -184,6 +186,7 @@ data ServerConfig s = ServerConfig
    -- | interval and file to save prometheus metrics
    prometheusInterval :: Maybe Int,
    prometheusMetricsFile :: FilePath,
+    addressStats :: Maybe AddressStatsConfig,
    -- | notification delivery interval
    ntfDeliveryInterval :: Int,
    -- | interval between sending pending END events to unsubscribed clients, seconds
@@ -200,8 +203,6 @@ data ServerConfig s = ServerConfig
    smpAgentCfg :: SMPClientAgentConfig,
    allowSMPProxy :: Bool, -- auth is the same with `newQueueBasicAuth`
    serverClientConcurrency :: Int,
-    -- | max forwarded commands awaiting a response from one destination relay
-    proxyRelayConcurrency :: Int,
    -- | max concurrent name resolutions per connection, enforced in forkCmd.
    -- Much higher than serverClientConcurrency: forwarded RSLVs from many clients
    -- aggregate over a single proxy->relay connection (only servers send proxied
@@ -255,9 +256,6 @@ defaultInactiveClientExpiration =
 defaultProxyClientConcurrency :: Int
 defaultProxyClientConcurrency = 32

-defaultProxyRelayConcurrency :: Int
-defaultProxyRelayConcurrency = 512
-
 defaultNameResolverConcurrency :: Int
 defaultNameResolverConcurrency = 1000

@@ -291,7 +289,8 @@ data Env s = Env
    sockets :: TVar [(ServiceName, SocketState)],
    clientSeq :: TVar ClientId,
    proxyAgent :: ProxyAgent, -- senders served on this proxy
-    namesEnv :: Maybe NamesEnv -- public-namespace resolver, present when [NAMES] enable: on
+    namesEnv :: Maybe NamesEnv, -- public-namespace resolver, present when [NAMES] enable: on
+    addrStats :: Maybe (ServerAddrStats SMPAddrCounter)
  }

 msgStore :: Env s -> s
@@ -447,10 +446,8 @@ data ClientSub
  | CSDeleted QueueId (Maybe ServiceId) -- includes previously associated service IDs
  | CSService ServiceId (Int64, IdsHash) -- only send END to idividual client subs on message delivery, not of SSUB/NSSUB

-data ProxyAgent = ProxyAgent
-  { smpAgent :: SMPClientAgent 'Sender,
-    -- forwarded commands awaiting a response, per relay session
-    relayForwards :: TVar (Map SessionId Int)
+newtype ProxyAgent = ProxyAgent
+  { smpAgent :: SMPClientAgent 'Sender
  }

 type ClientId = Int
@@ -473,9 +470,117 @@ data Client s = Client
    connected :: TVar Bool,
    createdAt :: SystemTime,
    rcvActiveAt :: TVar SystemTime,
-    sndActiveAt :: TVar SystemTime
+    sndActiveAt :: TVar SystemTime,
+    clientAddrStats :: Maybe (AddrStats SMPAddrCounter)
  }

+data SMPAddrCounter
+  = SACConnections
+  | SACErrors
+  | SACNew
+  | SACSub
+  | SACSubs
+  | SACKey
+  | SACRKey
+  | SACLSet
+  | SACLDel
+  | SACNKey
+  | SACNDel
+  | SACGet
+  | SACAck
+  | SACOff
+  | SACDel
+  | SACQue
+  | SACSKey
+  | SACSend
+  | SACPing
+  | SACLKey
+  | SACLGet
+  | SACNSub
+  | SACNSubs
+  | SACPrxy
+  | SACPfwd
+  | SACRfwd
+  | SACRslv
+  | SACPrxyOwn
+  | SACPrxyConnected
+  | SACPrxyNew
+  | SACPrxyOnion
+  | SACPrxyFailed
+  | SACPfwdOwn
+  | SACPfwdOther
+  | SACPfwdOnion
+  | SACPfwdFailed
+  deriving (Eq, Ord, Enum, Bounded, Show)
+
+instance AddrCounter SMPAddrCounter where
+  counterName = \case
+    SACConnections -> "connections"
+    SACErrors -> "errors"
+    SACNew -> "NEW"
+    SACSub -> "SUB"
+    SACSubs -> "SUBS"
+    SACKey -> "KEY"
+    SACRKey -> "RKEY"
+    SACLSet -> "LSET"
+    SACLDel -> "LDEL"
+    SACNKey -> "NKEY"
+    SACNDel -> "NDEL"
+    SACGet -> "GET"
+    SACAck -> "ACK"
+    SACOff -> "OFF"
+    SACDel -> "DEL"
+    SACQue -> "QUE"
+    SACSKey -> "SKEY"
+    SACSend -> "SEND"
+    SACPing -> "PING"
+    SACLKey -> "LKEY"
+    SACLGet -> "LGET"
+    SACNSub -> "NSUB"
+    SACNSubs -> "NSUBS"
+    SACPrxy -> "PRXY"
+    SACPfwd -> "PFWD"
+    SACRfwd -> "RFWD"
+    SACRslv -> "RSLV"
+    SACPrxyOwn -> "PRXY_own"
+    SACPrxyConnected -> "PRXY_connected"
+    SACPrxyNew -> "PRXY_new"
+    SACPrxyOnion -> "PRXY_onion"
+    SACPrxyFailed -> "PRXY_failed"
+    SACPfwdOwn -> "PFWD_own"
+    SACPfwdOther -> "PFWD_other"
+    SACPfwdOnion -> "PFWD_onion"
+    SACPfwdFailed -> "PFWD_failed"
+  connectionsCounter = SACConnections
+
+smpCmdCounter :: Cmd -> SMPAddrCounter
+smpCmdCounter (Cmd _ cmd) = case cmd of
+  NEW _ -> SACNew
+  SUB -> SACSub
+  SUBS {} -> SACSubs
+  KEY _ -> SACKey
+  RKEY _ -> SACRKey
+  LSET {} -> SACLSet
+  LDEL -> SACLDel
+  NKEY {} -> SACNKey
+  NDEL -> SACNDel
+  GET -> SACGet
+  ACK _ -> SACAck
+  OFF -> SACOff
+  DEL -> SACDel
+  QUE -> SACQue
+  SKEY _ -> SACSKey
+  SEND {} -> SACSend
+  PING -> SACPing
+  LKEY _ -> SACLKey
+  LGET -> SACLGet
+  NSUB -> SACNSub
+  NSUBS {} -> SACNSubs
+  PRXY {} -> SACPrxy
+  PFWD {} -> SACPfwd
+  RFWD _ -> SACRfwd
+  RSLV _ -> SACRslv
+
 type VerifiedTransmission s = (Maybe (StoreQueue s, QueueRec), Transmission Cmd)

 type ResponseAndMessage = (Transmission BrokerMsg, Maybe (Transmission BrokerMsg))
@@ -528,8 +633,8 @@ newServerSubscribers = do
  pendingEvents <- newTVarIO IM.empty
  pure ServerSubscribers {subQ, queueSubscribers, serviceSubscribers, totalServiceSubs, subClients, pendingEvents}

-newClient :: ClientId -> Natural -> THandleParams SMPVersion 'TServer -> SystemTime -> IO (Client s)
-newClient clientId qSize clientTHParams createdAt = do
+newClient :: ClientId -> Natural -> THandleParams SMPVersion 'TServer -> SystemTime -> Maybe (AddrStats SMPAddrCounter) -> IO (Client s)
+newClient clientId qSize clientTHParams createdAt clientAddrStats = do
  subscriptions <- TM.emptyIO
  ntfSubscriptions <- TM.emptyIO
  serviceSubscribed <- newTVarIO False
@@ -564,7 +669,8 @@ newClient clientId qSize clientTHParams createdAt = do
        connected,
        createdAt,
        rcvActiveAt,
-        sndActiveAt
+        sndActiveAt,
+        clientAddrStats
      }

 newSubscription :: SubscriptionThread -> STM Sub
@@ -579,7 +685,7 @@ newProhibitedSub = do
  return Sub {subThread = ProhibitSub, delivered}

 newEnv :: ServerConfig s -> IO (Env s)
-newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smpAgentCfg, information, messageExpiration, idleQueueInterval, msgQueueQuota, maxJournalMsgCount, maxJournalStateLines, namesConfig} = do
+newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smpAgentCfg, information, messageExpiration, idleQueueInterval, msgQueueQuota, maxJournalMsgCount, maxJournalStateLines, namesConfig, addressStats} = do
  serverActive <- newTVarIO True
  server <- newServer
  msgStore_ <- case serverStoreCfg of
@@ -634,6 +740,7 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smp
      Right _ -> logInfo "[NAMES] endpoint probe ok"
      Left e -> logWarn $ "[NAMES] endpoint probe failed (server will still start, RSLV will return ERR (NAME ...) until reachable): " <> tshow e
    pure env
+  addrStats <- mapM (const newServerAddrStats) addressStats
  pure
    Env
      { serverActive,
@@ -650,7 +757,8 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smp
        sockets,
        clientSeq,
        proxyAgent,
-        namesEnv
+        namesEnv,
+        addrStats
      }
  where
    loadStoreLog :: StoreQueueClass q => (RecipientId -> QueueRec -> IO q) -> FilePath -> STMQueueStore q -> IO ()
@@ -739,8 +847,7 @@ mkJournalStoreConfig queueStoreCfg storePath msgQueueQuota maxJournalMsgCount ma
 newSMPProxyAgent :: SMPClientAgentConfig -> TVar ChaChaDRG -> IO ProxyAgent
 newSMPProxyAgent smpAgentCfg random = do
  smpAgent <- newSMPClientAgent SSender smpAgentCfg Nothing random
-  relayForwards <- newTVarIO mempty
-  pure ProxyAgent {smpAgent, relayForwards}
+  pure ProxyAgent {smpAgent}

 readWriteQueueStore :: forall q. StoreQueueClass q => Bool -> (RecipientId -> QueueRec -> IO q) -> FilePath -> STMQueueStore q -> IO (StoreLog 'WriteMode)
 readWriteQueueStore tty mkQ = readWriteStoreLog (readQueueStore tty mkQ) (writeQueueStore @q)
@@ -578,6 +578,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
              serverStatsBackupFile = logStats $> combine logPath "smp-server-stats.log",
              prometheusInterval = eitherToMaybe (lookupValue "STORE_LOG" "prometheus_interval" ini) >>= readMaybe . T.unpack,
              prometheusMetricsFile = combine logPath "smp-server-metrics.txt",
+              addressStats = iniAddressStats ini,
              pendingENDInterval = 15000000, -- 15 seconds
              ntfDeliveryInterval = 1500000, -- 1.5 second
              smpServerVRange = supportedServerSMPRelayVRange,
@@ -609,7 +610,6 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
                  },
              allowSMPProxy = True,
              serverClientConcurrency = readIniDefault defaultProxyClientConcurrency "PROXY" "client_concurrency" ini,
-              proxyRelayConcurrency = readIniDefault defaultProxyRelayConcurrency "PROXY" "relay_concurrency" ini,
              serverResolverConcurrency = readIniDefault defaultNameResolverConcurrency "NAMES" "resolver_concurrency" ini,
              namesConfig = readNamesConfig ini,
              information = serverPublicInfo ini,
@@ -25,7 +25,7 @@ import Network.Socket (HostName)
 import Simplex.Messaging.Agent.Store.Postgres.Options (DBOpts (..))
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (BasicAuth)
-import Simplex.Messaging.Server.CLI (SignAlgorithm, onOff)
+import Simplex.Messaging.Server.CLI (SignAlgorithm, addressStatsIniContent, onOff)
 import Simplex.Messaging.Server.Env.STM
 import Simplex.Messaging.Server.Expiration (ExpirationConfig (..))
 import Simplex.Messaging.Server.Information (Entity (..), ServerPublicInfo (..))
@@ -154,9 +154,6 @@ iniFileContent cfgPath logPath opts host basicAuth controlPortPwds =
        \# socks_mode = onion\n\n\
        \# Limit number of threads a client can spawn to process proxy commands in parrallel.\n"
    <> ("# client_concurrency = " <> tshow defaultProxyClientConcurrency)
-    <> "\n\n\
-        \# Limit number of messages forwarded to one destination server and awaiting its response.\n"
-    <> ("# relay_concurrency = " <> tshow defaultProxyRelayConcurrency)
    <> "\n\n\
        \[NAMES]\n\
        \# Public-namespace resolution via the snrc-resolve.py REST resolver.\n\
@@ -179,7 +176,9 @@ iniFileContent cfgPath logPath opts host basicAuth controlPortPwds =
        \disconnect = on\n"
    <> ("ttl = " <> tshow (ttl defaultInactiveClientExpiration) <> "\n")
    <> ("check_interval = " <> tshow (checkInterval defaultInactiveClientExpiration))
-    <> "\n\n\
+    <> "\n\n"
+    <> addressStatsIniContent
+    <> "\n\
        \[WEB]\n\
        \# Set path to generate static mini-site for server information and qr codes/links\n"
    <> ("static_path = " <> T.pack (fromMaybe defaultStaticPath webStaticPath) <> "\n\n")
@@ -14,12 +14,15 @@ module Simplex.Messaging.Server.Prometheus
 import Data.Int (Int64)
 import qualified Data.IntMap.Strict as IM
 import Data.List (mapAccumL)
+import Data.Map.Strict (Map)
 import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Time.Clock (UTCTime (..), diffUTCTime)
 import Data.Time.Clock.System (systemEpochDay)
 import Data.Time.Format.ISO8601 (iso8601Show)
 import Network.Socket (ServiceName)
+import Simplex.Messaging.Server.AddressStats (AddrHistogram, addrHistogramMetrics)
+import Simplex.Messaging.Server.Env.STM (SMPAddrCounter)
 import Simplex.Messaging.Server.MsgStore.Types (LoadedQueueCounts (..))
 import Simplex.Messaging.Server.QueueStore.Types (EntityCounts (..))
 import Simplex.Messaging.Server.Stats
@@ -46,7 +49,8 @@ data RealTimeMetrics = RealTimeMetrics
    deliveredTimes :: TimeBuckets,
    smpSubs :: RTSubscriberMetrics,
    ntfSubs :: RTSubscriberMetrics,
-    loadedCounts :: LoadedQueueCounts
+    loadedCounts :: LoadedQueueCounts,
+    addressHistograms :: Maybe (Map SMPAddrCounter AddrHistogram)
  }

 data RTSubscriberMetrics = RTSubscriberMetrics
@@ -59,7 +63,7 @@ data RTSubscriberMetrics = RTSubscriberMetrics
 {-# FOURMOLU_DISABLE\n#-}
 prometheusMetrics :: ServerMetrics -> RealTimeMetrics -> UTCTime -> Text
 prometheusMetrics sm rtm ts =
-  time <> queues <> subscriptions <> messages <> ntfMessages <> ntfs <> relays <> services <> names <> info
+  time <> queues <> subscriptions <> messages <> ntfMessages <> ntfs <> relays <> services <> names <> clientAddresses <> info
  where
    ServerMetrics {statsData, activeQueueCounts = ps, activeNtfCounts = psNtf, entityCounts, rtsOptions} = sm
    RealTimeMetrics
@@ -70,7 +74,8 @@ prometheusMetrics sm rtm ts =
        deliveredTimes,
        smpSubs,
        ntfSubs,
-        loadedCounts
+        loadedCounts,
+        addressHistograms
      } = rtm
    ServerStatsData
      { _fromTime,
@@ -485,6 +490,7 @@ prometheusMetrics sm rtm ts =
          \# TYPE simplex_smp_names_disabled counter\n\
          \simplex_smp_names_disabled " <> mshow _rslvDisabled <> "\n# rslvDisabled\n\
          \\n"
+    clientAddresses = maybe "" (addrHistogramMetrics "simplex_smp" $ tsEpoch ts) addressHistograms
    info =
      "# Info\n\
      \# ----\n\
@@ -865,6 +865,7 @@ 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 =
@@ -54,7 +54,7 @@ getHTTP2Server HTTP2ServerConfig {qSize, http2Port, bufferSize, bodyHeadSize, se
  started <- newEmptyTMVarIO
  reqQ <- newTBQueueIO qSize
  action <- async $
-    runHTTP2Server started http2Port bufferSize serverSupported srvCreds Nothing transportConfig Nothing (const $ pure ()) $ \_sniUsed sessionId sessionALPN r sendResponse -> do
+    runHTTP2Server started http2Port bufferSize serverSupported srvCreds Nothing transportConfig Nothing (const $ pure ()) (\_ _ handler -> handler) $ \_sniUsed sessionId sessionALPN r sendResponse -> do
      reqBody <- getHTTP2Body r bodyHeadSize
      atomically $ writeTBQueue reqQ HTTP2Request {sessionId, sessionALPN, request = r, reqBody, sendResponse}
  void . atomically $ takeTMVar started
@@ -63,13 +63,13 @@ getHTTP2Server HTTP2ServerConfig {qSize, http2Port, bufferSize, bodyHeadSize, se
 closeHTTP2Server :: HTTP2Server -> IO ()
 closeHTTP2Server = uninterruptibleCancel . action

-runHTTP2Server :: TMVar Bool -> ServiceName -> BufferSize -> T.Supported -> T.Credential -> Maybe T.Credential -> TransportServerConfig -> Maybe ExpirationConfig -> (SessionId -> IO ()) -> (SNICredentialUsed -> HTTP2ServerFunc) -> IO ()
-runHTTP2Server started port bufferSize srvSupported srvCreds httpCreds_ transportConfig expCfg_ clientFinished = runHTTP2ServerWith_ expCfg_ clientFinished bufferSize setup
+runHTTP2Server :: TMVar Bool -> ServiceName -> BufferSize -> T.Supported -> T.Credential -> Maybe T.Credential -> TransportServerConfig -> Maybe ExpirationConfig -> (SessionId -> IO ()) -> (Socket -> SessionId -> IO () -> IO ()) -> (SNICredentialUsed -> HTTP2ServerFunc) -> IO ()
+runHTTP2Server started port bufferSize srvSupported srvCreds httpCreds_ transportConfig expCfg_ clientFinished withConnection = runHTTP2ServerWith_ expCfg_ clientFinished bufferSize setup
  where
    setup handler = do
      ss <- newSocketState
      let combinedCreds = TLSServerCredential {credential = srvCreds, sniCredential = httpCreds_}
-      runTransportServerState_ ss started port srvSupported combinedCreds transportConfig $ \_ -> handler
+      runTransportServerState_ ss started port srvSupported combinedCreds transportConfig $ \sock h@(_, tls) -> withConnection sock (tlsUniq tls) (handler h)

 -- HTTP2 server can be run on both client and server TLS connections.
 runHTTP2ServerWith :: BufferSize -> ((TLS p -> IO ()) -> a) -> HTTP2ServerFunc -> a
@@ -108,6 +108,8 @@ smpServerTest storeLog basicAuth = do
  lookupValue "TRANSPORT" "websockets" ini `shouldBe` Right "off"
  lookupValue "AUTH" "new_queues" ini `shouldBe` Right "on"
  lookupValue "INACTIVE_CLIENTS" "disconnect" ini `shouldBe` Right "on"
+  lookupValue "ADDRESS_STATS" "enable" ini `shouldBe` Right "off"
+  lookupValue "ADDRESS_STATS" "period" ini `shouldBe` Right "300"
  doesFileExist (cfgPath <> "/ca.key") `shouldReturn` True
  -- start
  r <- lines <$> capture_ (withArgs ["start"] $ (100000 `timeout` smpServerCLI cfgPath logPath) `catchAll_` pure (Just ()))
@@ -234,6 +236,8 @@ ntfServerTest storeLog = do
  lookupValue "STORE_LOG" "log_stats" ini `shouldBe` Right "off"
  lookupValue "TRANSPORT" "port" ini `shouldBe` Right "443"
  lookupValue "TRANSPORT" "websockets" ini `shouldBe` Right "off"
+  lookupValue "ADDRESS_STATS" "enable" ini `shouldBe` Right "off"
+  lookupValue "ADDRESS_STATS" "period" ini `shouldBe` Right "300"
  doesFileExist (ntfCfgPath <> "/ca.key") `shouldReturn` True
  r <- lines <$> capture_ (withArgs ["start"] $ (100000 `timeout` ntfServerCLI ntfCfgPath ntfLogPath) `catchAll_` pure (Just ()))
  r `shouldContain` ["SMP notifications server v" <> simplexmqVersionCommit]
@@ -252,6 +256,8 @@ xftpServerTest storeLog = do
  lookupValue "STORE_LOG" "enable" ini `shouldBe` Right (if storeLog then "on" else "off")
  lookupValue "STORE_LOG" "log_stats" ini `shouldBe` Right "off"
  lookupValue "TRANSPORT" "port" ini `shouldBe` Right "443"
+  lookupValue "ADDRESS_STATS" "enable" ini `shouldBe` Right "off"
+  lookupValue "ADDRESS_STATS" "period" ini `shouldBe` Right "300"
  doesFileExist (fileCfgPath <> "/ca.key") `shouldReturn` True
  r <- lines <$> capture_ (withArgs ["start"] $ (100000 `timeout` xftpServerCLI fileCfgPath fileLogPath) `catchAll_` pure (Just ()))
  r `shouldContain` ["SimpleX XFTP server v" <> simplexmqVersionCommit]
@@ -0,0 +1,326 @@
+{-# LANGUAGE CPP #-}
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+
+module CoreTests.AddressStatsTests where
+
+import Data.ByteString.Char8 (ByteString)
+import qualified Data.ByteString.Char8 as B
+import Data.IORef (readIORef)
+import Data.List.NonEmpty (NonEmpty (..))
+import qualified Data.Map.Strict as M
+import qualified Data.Set as S
+import Data.Text (Text)
+import qualified Data.Text as T
+import Data.Text.Encoding (decodeLatin1)
+import Network.Socket (SockAddr (..), tupleToHostAddress, tupleToHostAddress6)
+import qualified Simplex.FileTransfer.Protocol as XFTP
+import Simplex.FileTransfer.Server.Env (XFTPAddrCounter (..), xftpCmdCounter)
+import Simplex.FileTransfer.Transport (supportedFileServerVRange)
+import qualified Simplex.Messaging.Crypto as C
+import Simplex.Messaging.Protocol
+import Simplex.Messaging.Server.AddressStats
+import Simplex.Messaging.Server.Env.STM (SMPAddrCounter (..), smpCmdCounter)
+import Simplex.Messaging.SimplexName (SimplexDomain (..), SimplexTLD (..))
+import Simplex.Messaging.Transport (supportedClientSMPRelayVRange, supportedServerSMPRelayVRange)
+import Simplex.Messaging.Version (VersionRange (..))
+import Test.Hspec hiding (fit, it)
+import UnliftIO.STM
+import Util
+#if defined(dbServerPostgres)
+import qualified Simplex.Messaging.Notifications.Protocol as NTF
+import Simplex.Messaging.Notifications.Server.Env (NtfAddrCounter (..), ntfCmdCounter)
+import Simplex.Messaging.Notifications.Transport (supportedServerNTFVRange)
+#endif
+
+data TestCounter = TCConnections | TCSend | TCGet
+  deriving (Eq, Ord, Enum, Bounded, Show)
+
+instance AddrCounter TestCounter where
+  counterName = \case
+    TCConnections -> "connections"
+    TCSend -> "SEND"
+    TCGet -> "GET"
+  counterBounds _ = [1, 2, 5, 10]
+  connectionsCounter = TCConnections
+
+addr1, addr2, addr3, addr4 :: AddrKey
+addr1 = AKIPv4 1
+addr2 = AKIPv4 2
+addr3 = AKIPv4 3
+addr4 = AKIPv4 4
+
+addressStatsTests :: Spec
+addressStatsTests = do
+  describe "addrKey" addrKeyTests
+  describe "bounds" boundsTests
+  describe "counterByName" counterByNameTests
+  describe "openAddrStats" openAddrStatsTests
+  describe "rolloverAddrStats" rolloverTests
+  describe "topAddresses" topAddressesTests
+  describe "addrHistogramMetrics" metricsTests
+  describe "command counters" commandCounterTests
+
+addrKeyTests :: Spec
+addrKeyTests = do
+  it "should use IPv4 address" $
+    addrKeyText <$> addrKey (ipv4 (192, 168, 1, 2)) `shouldBe` Just "192.168.1.2"
+  it "should use IPv4 address of IPv4-mapped IPv6 address" $
+    addrKey (ipv6 (0, 0, 0, 0, 0, 0xffff, 0xc0a8, 0x0102)) `shouldBe` addrKey (ipv4 (192, 168, 1, 2))
+  it "should use one key for IPv6 addresses in one /64" $ do
+    addrKey (ipv6 (0x2001, 0xdb8, 1, 2, 3, 4, 5, 6)) `shouldBe` addrKey (ipv6 (0x2001, 0xdb8, 1, 2, 7, 8, 9, 10))
+    addrKeyText <$> addrKey (ipv6 (0x2001, 0xdb8, 1, 2, 3, 4, 5, 6)) `shouldBe` Just "2001:db8:1:2::/64"
+  it "should use different keys for IPv6 addresses in different /64" $
+    addrKey (ipv6 (0x2001, 0xdb8, 1, 2, 3, 4, 5, 6)) `shouldNotBe` addrKey (ipv6 (0x2001, 0xdb8, 1, 3, 3, 4, 5, 6))
+  it "should not use other address types" $
+    addrKey (SockAddrUnix "tests/tmp/socket") `shouldBe` Nothing
+  where
+    ipv4 a = SockAddrInet 443 $ tupleToHostAddress a
+    ipv6 a = SockAddrInet6 443 0 (tupleToHostAddress6 a) 0
+
+boundsTests :: Spec
+boundsTests = do
+  it "should use R10 values up to 1000000 for counts" $
+    countBounds
+      `shouldBe` [ 1, 2, 3, 4, 5, 6, 7, 8, 9,
+                   10, 13, 16, 20, 25, 32, 40, 50, 63, 80,
+                   100, 125, 160, 200, 250, 315, 400, 500, 630, 800,
+                   1000, 1250, 1600, 2000, 2500, 3150, 4000, 5000, 6300, 8000,
+                   10000, 12500, 16000, 20000, 25000, 31500, 40000, 50000, 63000, 80000,
+                   100000, 125000, 160000, 200000, 250000, 315000, 400000, 500000, 630000, 800000,
+                   1000000
+                 ]
+  it "should use R10 values up to 100000000 for kilobytes" $ do
+    take 60 kilobyteBounds `shouldBe` countBounds
+    drop 59 kilobyteBounds
+      `shouldBe` [ 1000000, 1250000, 1600000, 2000000, 2500000, 3150000, 4000000, 5000000, 6300000, 8000000,
+                   10000000, 12500000, 16000000, 20000000, 25000000, 31500000, 40000000, 50000000, 63000000, 80000000,
+                   100000000
+                 ]
+
+counterByNameTests :: Spec
+counterByNameTests = do
+  it "should find SMP counters" $ testCounterNames [minBound .. maxBound :: SMPAddrCounter]
+  it "should find XFTP counters" $ testCounterNames [minBound .. maxBound :: XFTPAddrCounter]
+#if defined(dbServerPostgres)
+  it "should find NTF counters" $ testCounterNames [minBound .. maxBound :: NtfAddrCounter]
+#endif
+  it "should not find unknown counter" $
+    counterByName "unknown" `shouldBe` (Nothing :: Maybe SMPAddrCounter)
+  where
+    testCounterNames :: (AddrCounter c, Show c) => [c] -> Expectation
+    testCounterNames cs = map (counterByName . counterName) cs `shouldBe` map Just cs
+
+openAddrStatsTests :: Spec
+openAddrStatsTests =
+  it "should count connections of one address in one entry" $ do
+    ServerAddrStats {addrStatsMap = m} <- newServerAddrStats
+    s <- openAddrStats m addr1
+    s' <- openAddrStats m addr1
+    readTVarIO (connectionsCount s) `shouldReturn` 2
+    counts s TCConnections `shouldReturn` (0, 2)
+    closeAddrStats s'
+    readTVarIO (connectionsCount s) `shouldReturn` 1
+
+rolloverTests :: Spec
+rolloverTests = do
+  it "should add non-zero closing counts to histograms" $ do
+    stats@ServerAddrStats {addrStatsMap = m, addrHistograms} <- newServerAddrStats
+    s1 <- openAddrStats m addr1
+    s2 <- openAddrStats m addr2
+    addAddrCounter s1 TCSend 3
+    addAddrCounter s2 TCSend 7
+    rolloverAddrStats stats
+    readIORef addrHistograms
+      `shouldReturn` M.fromList
+        [ (TCConnections, AddrHistogram [2, 2, 2, 2] 2 2 1),
+          (TCSend, AddrHistogram [0, 0, 1, 2] 10 2 7),
+          (TCGet, AddrHistogram [0, 0, 0, 0] 0 0 0)
+        ]
+    addAddrCounter s1 TCSend 20
+    rolloverAddrStats stats
+    readIORef addrHistograms
+      `shouldReturn` M.fromList
+        [ (TCConnections, AddrHistogram [2, 2, 2, 2] 2 2 0),
+          (TCSend, AddrHistogram [0, 0, 1, 2] 30 3 20),
+          (TCGet, AddrHistogram [0, 0, 0, 0] 0 0 0)
+        ]
+  it "should set previous counts to closing counts" $ do
+    stats@ServerAddrStats {addrStatsMap = m} <- newServerAddrStats
+    s <- openAddrStats m addr1
+    addAddrCounter s TCSend 3
+    counts s TCSend `shouldReturn` (0, 3)
+    rolloverAddrStats stats
+    counts s TCSend `shouldReturn` (3, 0)
+    addAddrCounter s TCSend 5
+    rolloverAddrStats stats
+    counts s TCSend `shouldReturn` (5, 0)
+    rolloverAddrStats stats
+    counts s TCSend `shouldReturn` (0, 0)
+  it "should keep entries with open connections or closing counts, and delete other entries" $ do
+    stats@ServerAddrStats {addrStatsMap = m} <- newServerAddrStats
+    _ <- openAddrStats m addr1
+    s2 <- openAddrStats m addr2
+    s3 <- openAddrStats m addr3
+    closeAddrStats s3
+    rolloverAddrStats stats
+    M.keys <$> readTVarIO m `shouldReturn` [addr1, addr2, addr3]
+    addAddrCounter s2 TCSend 1
+    closeAddrStats s2
+    rolloverAddrStats stats
+    M.keys <$> readTVarIO m `shouldReturn` [addr1, addr2]
+    rolloverAddrStats stats
+    M.keys <$> readTVarIO m `shouldReturn` [addr1]
+
+topAddressesTests :: Spec
+topAddressesTests =
+  it "should return addresses with non-zero counts, sorted by the sum of previous and current counts" $ do
+    stats@ServerAddrStats {addrStatsMap = m} <- newServerAddrStats
+    [s1, s2, s3, _] <- mapM (openAddrStats m) [addr1, addr2, addr3, addr4]
+    addAddrCounter s1 TCSend 3
+    addAddrCounter s2 TCSend 7
+    rolloverAddrStats stats
+    addAddrCounter s1 TCSend 5
+    addAddrCounter s3 TCSend 10
+    topAddresses m TCSend 10 `shouldReturn` [(addr3, 0, 10), (addr1, 3, 5), (addr2, 7, 0)]
+    topAddresses m TCSend 2 `shouldReturn` [(addr3, 0, 10), (addr1, 3, 5)]
+
+metricsTests :: Spec
+metricsTests =
+  it "should encode histograms and maximums as Prometheus metrics" $
+    addrHistogramMetrics "simplex_test" "1767225600000" histograms
+      `shouldBe` T.unlines
+        [ "# Client addresses",
+          "# ----------------",
+          "",
+          "# HELP simplex_test_client_address_period_count Counts of client addresses per period",
+          "# TYPE simplex_test_client_address_period_count histogram",
+          "simplex_test_client_address_period_count_bucket{counter=\"SEND\",le=\"1\"} 0 1767225600000",
+          "simplex_test_client_address_period_count_bucket{counter=\"SEND\",le=\"2\"} 0 1767225600000",
+          "simplex_test_client_address_period_count_bucket{counter=\"SEND\",le=\"5\"} 1 1767225600000",
+          "simplex_test_client_address_period_count_bucket{counter=\"SEND\",le=\"10\"} 2 1767225600000",
+          "simplex_test_client_address_period_count_bucket{counter=\"SEND\",le=\"+Inf\"} 2 1767225600000",
+          "simplex_test_client_address_period_count_sum{counter=\"SEND\"} 10 1767225600000",
+          "simplex_test_client_address_period_count_count{counter=\"SEND\"} 2 1767225600000",
+          "simplex_test_client_address_period_count_bucket{counter=\"GET\",le=\"1\"} 0 1767225600000",
+          "simplex_test_client_address_period_count_bucket{counter=\"GET\",le=\"2\"} 0 1767225600000",
+          "simplex_test_client_address_period_count_bucket{counter=\"GET\",le=\"5\"} 0 1767225600000",
+          "simplex_test_client_address_period_count_bucket{counter=\"GET\",le=\"10\"} 0 1767225600000",
+          "simplex_test_client_address_period_count_bucket{counter=\"GET\",le=\"+Inf\"} 1 1767225600000",
+          "simplex_test_client_address_period_count_sum{counter=\"GET\"} 12 1767225600000",
+          "simplex_test_client_address_period_count_count{counter=\"GET\"} 1 1767225600000",
+          "",
+          "# HELP simplex_test_client_address_period_max Maximum count of a client address in the last period",
+          "# TYPE simplex_test_client_address_period_max gauge",
+          "simplex_test_client_address_period_max{counter=\"SEND\"} 7 1767225600000",
+          "simplex_test_client_address_period_max{counter=\"GET\"} 12 1767225600000",
+          ""
+        ]
+  where
+    histograms =
+      M.fromList
+        [ (TCSend, AddrHistogram [0, 0, 1, 2] 10 2 7),
+          (TCGet, AddrHistogram [0, 0, 0, 0] 12 1 12)
+        ]
+
+commandCounterTests :: Spec
+commandCounterTests = do
+  it "should count each SMP command in the counter named by its tag" $ do
+    cmds <- smpCommands
+    map (counterName . smpCmdCounter) cmds `shouldBe` map (commandTag . encodeProtocol (maxVersion supportedServerSMPRelayVRange)) cmds
+    S.fromList (map smpCmdCounter cmds) `shouldBe` S.fromList [SACNew .. SACRslv]
+  it "should count each XFTP command in the counter named by its tag" $ do
+    cmds <- xftpCommands
+    map (counterName . xftpCmdCounter) cmds `shouldBe` map (commandTag . encodeProtocol (maxVersion supportedFileServerVRange)) cmds
+    S.fromList (map xftpCmdCounter cmds) `shouldBe` S.fromList [XACFNew .. XACPing]
+#if defined(dbServerPostgres)
+  it "should count each NTF command in the counter named by its tag" $ do
+    cmds <- ntfCommands
+    map (counterName . ntfCmdCounter) cmds `shouldBe` map (commandTag . encodeProtocol (maxVersion supportedServerNTFVRange)) cmds
+    S.fromList (map ntfCmdCounter cmds) `shouldBe` S.fromList [NACTNew .. NACPing]
+#endif
+  where
+    commandTag :: ByteString -> Text
+    commandTag = decodeLatin1 . B.takeWhile (/= ' ')
+
+counts :: AddrStats TestCounter -> TestCounter -> IO (Int, Int)
+counts AddrStats {previous, current} c = (,) <$> readIORef (previous M.! c) <*> readIORef (current M.! c)
+
+testKeys :: IO (C.APublicAuthKey, C.APrivateAuthKey, C.PublicKeyX25519)
+testKeys = do
+  g <- C.newRandom
+  (authKey, authPrivKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
+  (dhKey, _ :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g
+  pure (authKey, authPrivKey, dhKey)
+
+testSMPServer :: SMPServer
+testSMPServer = SMPServer "localhost" "5001" (C.KeyHash "hash")
+
+smpCommands :: IO [Cmd]
+smpCommands = do
+  (authKey, _, dhKey) <- testKeys
+  let newQueue = NewQueueReq {rcvAuthKey = authKey, rcvDhKey = dhKey, auth_ = Nothing, subMode = SMSubscribe, queueReqData = Nothing, ntfCreds = Nothing}
+      linkData = (EncDataBytes "fixed", EncDataBytes "user")
+      name = NQDomain SimplexDomain {nameTLD = TLDSimplex, domain = "alice", subDomain = []}
+  pure
+    [ Cmd SCreator $ NEW newQueue,
+      Cmd SRecipient SUB,
+      Cmd SRecipientService $ SUBS 1 mempty,
+      Cmd SRecipient $ KEY authKey,
+      Cmd SRecipient $ RKEY (authKey :| []),
+      Cmd SRecipient $ LSET (EntityId "link") linkData,
+      Cmd SRecipient LDEL,
+      Cmd SRecipient $ NKEY authKey dhKey,
+      Cmd SRecipient NDEL,
+      Cmd SRecipient GET,
+      Cmd SRecipient $ ACK "message",
+      Cmd SRecipient OFF,
+      Cmd SRecipient DEL,
+      Cmd SRecipient QUE,
+      Cmd SSender $ SKEY authKey,
+      Cmd SSender $ SEND noMsgFlags "hello",
+      Cmd SIdleClient PING,
+      Cmd SSenderLink $ LKEY authKey,
+      Cmd SSenderLink LGET,
+      Cmd SNotifier NSUB,
+      Cmd SNotifierService $ NSUBS 1 mempty,
+      Cmd SProxiedClient $ PRXY testSMPServer Nothing,
+      Cmd SProxiedClient $ PFWD (maxVersion supportedClientSMPRelayVRange) dhKey (EncTransmission "transmission"),
+      Cmd SProxyService $ RFWD (EncFwdTransmission "transmission"),
+      Cmd SResolver $ RSLV name
+    ]
+
+xftpCommands :: IO [XFTP.FileCmd]
+xftpCommands = do
+  (authKey, _, dhKey) <- testKeys
+  pure
+    [ XFTP.FileCmd XFTP.SFSender $ XFTP.FNEW (XFTP.FileInfo authKey 65536 "digest") (authKey :| []) Nothing Nothing,
+      XFTP.FileCmd XFTP.SFSender $ XFTP.FADD (authKey :| []),
+      XFTP.FileCmd XFTP.SFSender XFTP.FPUT,
+      XFTP.FileCmd XFTP.SFSender XFTP.FDEL,
+      XFTP.FileCmd XFTP.SFRecipient $ XFTP.FGET dhKey,
+      XFTP.FileCmd XFTP.SFRecipient XFTP.FACK,
+      XFTP.FileCmd XFTP.SFRecipient XFTP.PING
+    ]
+
+#if defined(dbServerPostgres)
+ntfCommands :: IO [NTF.NtfCmd]
+ntfCommands = do
+  (authKey, authPrivKey, dhKey) <- testKeys
+  let token = NTF.DeviceToken NTF.PPApnsTest "token"
+      queue = NTF.SMPQueueNtf testSMPServer (EntityId "notifier")
+  pure
+    [ NTF.NtfCmd NTF.SToken $ NTF.TNEW $ NTF.NewNtfTkn token authKey dhKey,
+      NTF.NtfCmd NTF.SToken $ NTF.TVFY $ NTF.NtfRegCode "code",
+      NTF.NtfCmd NTF.SToken NTF.TCHK,
+      NTF.NtfCmd NTF.SToken $ NTF.TRPL token,
+      NTF.NtfCmd NTF.SToken NTF.TDEL,
+      NTF.NtfCmd NTF.SToken $ NTF.TCRN 20,
+      NTF.NtfCmd NTF.SSubscription $ NTF.SNEW $ NTF.NewNtfSub (EntityId "token") queue authPrivKey,
+      NTF.NtfCmd NTF.SSubscription NTF.SCHK,
+      NTF.NtfCmd NTF.SSubscription NTF.SDEL,
+      NTF.NtfCmd NTF.SSubscription NTF.PING
+    ]
+#endif
@@ -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, serverFileName)
+import Simplex.FileTransfer.Server.StoreLog (FileStoreLogRecord (..), readWriteFileStore)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol
@@ -37,7 +37,6 @@ 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

@@ -270,18 +269,6 @@ 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))]
        }
    ]

@@ -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, serverFileName, writeFileStore)
+import Simplex.FileTransfer.Server.StoreLog (closeStoreLog, readWriteFileStore, writeFileStore)
 import Simplex.FileTransfer.Transport (XFTPErrorType (..))
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Protocol (BlockingInfo (..), BlockingReason (..), EntityId (..))
@@ -17,7 +17,6 @@ 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
@@ -266,7 +265,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" </> serverFileName sId1)
+  void $ setFilePath stmStore sId1 "/tmp/file1"
  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 ()
@@ -290,7 +289,7 @@ testMigrationRoundTrip = do
  case result1 of
    Right (FileRec {fileInfo = fi, filePath, expiresAt, fileStatus}, _) -> do
      size fi `shouldBe` 128000
-      readTVarIO filePath `shouldReturn` Just ("/tmp" </> serverFileName sId1)
+      readTVarIO filePath `shouldReturn` Just "/tmp/file1"
      expiresAt `shouldBe` Just testExpiresAt
      readTVarIO fileStatus `shouldReturn` EntityActive
    Left e -> expectationFailure $ "getFile sId1 failed: " <> show e
@@ -48,6 +48,7 @@ import Simplex.Messaging.Notifications.Server.Push.APNS
 import Simplex.Messaging.Notifications.Server.Push.APNS.Internal
 import Simplex.Messaging.Notifications.Transport
 import Simplex.Messaging.Protocol
+import Simplex.Messaging.Server.AddressStats (AddressStatsConfig (..))
 import Simplex.Messaging.Server.QueueStore.Postgres.Config (PostgresStoreCfg (..))
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport
@@ -155,6 +156,7 @@ ntfServerCfg =
      serverStatsBackupFile = Nothing,
      prometheusInterval = Nothing,
      prometheusMetricsFile = ntfTestPrometheusMetricsFile,
+      addressStats = Just AddressStatsConfig {period = 1},
      ntfServerVRange = supportedServerNTFVRange,
      transportConfig = mkTransportServerConfig True (Just alpnSupportedNTFHandshakes) False,
      startOptions = defaultStartOptions
@@ -11,14 +11,11 @@

 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)
@@ -50,7 +47,6 @@ import Simplex.Messaging.Protocol
    tPut,
  )
 import qualified Simplex.Messaging.Protocol as SMP
-import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
 import Simplex.Messaging.SimplexName (SimplexDomain)
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Version (mkVersionRange)
@@ -106,8 +102,6 @@ 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" $
-    it "one connection has at most resolver_concurrency lookups in flight" testRslvConnectionCap

 -- | /v2/resolve answers 200, 400 or 502, so a 404 is a resolver that predates
 -- the route, not a name that does not exist.
@@ -281,34 +275,5 @@ 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
-
-- | 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
@@ -18,6 +18,7 @@ module SMPClient where
 import Control.Monad
 import Control.Monad.Except (runExceptT)
 import Data.ByteString.Char8 (ByteString)
+import qualified Data.ByteString.Char8 as B
 import Data.List.NonEmpty (NonEmpty)
 import qualified Data.X509 as X
 import qualified Data.X509.Validation as XV
@@ -31,16 +32,19 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server (runSMPServerBlocking)
+import Simplex.Messaging.Server.AddressStats (AddressStatsConfig (..))
 import Simplex.Messaging.Server.Env.STM
 import Simplex.Messaging.Server.MsgStore.Types (MsgStoreClass (..), SMSType (..), SQSType (..))
 import Simplex.Messaging.Server.QueueStore.Postgres.Config (PostgresStoreCfg (..))
 import Simplex.Messaging.Transport
+import Simplex.Messaging.Transport.Buffer (trimCR)
 import Simplex.Messaging.Transport.Client
 import Simplex.Messaging.Transport.Server
 import Simplex.Messaging.Transport.Shared (ChainCertificates (..), chainIdCaCerts)
 import Simplex.Messaging.Util (ifM)
 import Simplex.Messaging.Version
 import Simplex.Messaging.Version.Internal
+import System.IO (Handle, IOMode (..), hClose, hFlush)
 import System.Info (os)
 import Test.Hspec hiding (fit, it)
 import UnliftIO.Concurrent
@@ -84,6 +88,12 @@ ntfTestPort = "6001"
 ntfTestPort2 :: ServiceName
 ntfTestPort2 = "6002"

+testControlPort :: ServiceName
+testControlPort = "5226"
+
+testControlPortUserAuth :: BasicAuth
+testControlPortUserAuth = "user_password"
+
 testKeyHash :: C.KeyHash
 testKeyHash = "LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI="

@@ -261,6 +271,7 @@ cfgMS msType = withStoreCfg (testServerStoreConfig msType) $ \serverStoreCfg ->
      serverStatsBackupFile = Nothing,
      prometheusInterval = Nothing,
      prometheusMetricsFile = testPrometheusMetricsFile,
+      addressStats = Just AddressStatsConfig {period = 1},
      pendingENDInterval = 500000,
      ntfDeliveryInterval = 200000,
      smpCredentials =
@@ -276,7 +287,6 @@ cfgMS msType = withStoreCfg (testServerStoreConfig msType) $ \serverStoreCfg ->
      smpAgentCfg = defaultSMPClientAgentConfig {persistErrorInterval = 1, msgQSize = Nothing}, -- seconds
      allowSMPProxy = False,
      serverClientConcurrency = 2,
-      proxyRelayConcurrency = defaultProxyRelayConcurrency,
      serverResolverConcurrency = defaultNameResolverConcurrency,
      namesConfig = Nothing,
      information = Nothing,
@@ -317,9 +327,6 @@ 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)

@@ -350,14 +357,6 @@ 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}
@@ -408,6 +407,22 @@ withStallingServerOn port action =
    (pure ())
    (const action)

+withControlPort :: (Handle -> IO a) -> IO a
+withControlPort action = do
+  addr : _ <- getAddrInfo (Just defaultHints {addrSocketType = Stream}) (Just "127.0.0.1") (Just testControlPort)
+  E.bracket (connectHandle addr) hClose $ \h -> replicateM_ 2 (B.hGetLine h) >> action h
+  where
+    connectHandle addr = do
+      sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)
+      connect sock $ addrAddress addr
+      socketToHandle sock ReadWriteMode
+
+controlPortCommand :: Handle -> ByteString -> Int -> IO [ByteString]
+controlPortCommand h cmd n = do
+  B.hPutStrLn h cmd
+  hFlush h
+  replicateM n $ trimCR <$> B.hGetLine h
+
 withSmpServerOn :: HasCallStack => (ASrvTransport, AStoreType) -> ServiceName -> IO a -> IO a
 withSmpServerOn ps port' = withSmpServerThreadOn ps port' . const

@@ -21,7 +21,6 @@ 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)
@@ -34,11 +33,13 @@ import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..), InitialAgentServers
 import Simplex.Messaging.Agent.Protocol hiding (CON, CONF, INFO, REQ)
 import qualified Simplex.Messaging.Agent.Protocol as A
 import Simplex.Messaging.Client
+import Simplex.Messaging.Client.Agent (SMPClientAgentConfig (..))
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.Ratchet (pattern PQSupportOn)
 import qualified Simplex.Messaging.Crypto.Ratchet as CR
 import Simplex.Messaging.Protocol (EncRcvMsgBody (..), MsgBody, QueueReqData (..), RcvMessage (..), SubscriptionMode (..), maxMessageLength, noMsgFlags)
 import qualified Simplex.Messaging.Protocol as SMP
+import Simplex.Messaging.Server.AddressStats (AddressStatsConfig (..), defaultAddressStatsPeriod)
 import Simplex.Messaging.Server.Env.STM (AStoreType (..), ServerConfig (..))
 import Simplex.Messaging.Server.MsgStore.Types (SQSType (..))
 import Simplex.Messaging.Transport
@@ -64,10 +65,6 @@ 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
@@ -154,7 +151,29 @@ smpProxyTests = do
      xdescribe "stress test 10k" $ do
        let deliver nAgents nMsgs = agentDeliverMessagesViaProxyConc (replicate nAgents [srv1]) (map bshow [1 :: Int .. nMsgs])
        it "25 agents, 300 pairs, 17 messages" . oneServer . withNumCapabilities 4 $ deliver 25 17
+  describe "client address statistics" $ do
+    it "should count proxy commands to other servers" $
+      proxyAddressStats [] [("PRXY", "2"), ("PRXY_new", "1"), ("PRXY_connected", "1"), ("PFWD", "3"), ("PFWD_other", "2"), ("PFWD_failed", "1")]
+    it "should count proxy commands to own servers" $
+      proxyAddressStats ["127.0.0.1"] [("PRXY", "2"), ("PRXY_own", "2"), ("PFWD", "3"), ("PFWD_own", "2"), ("PFWD_failed", "1")]
  where
+    proxyAddressStats :: [ByteString] -> [(ByteString, ByteString)] -> AStoreType -> IO ()
+    proxyAddressStats ownDomains counts msType =
+      twoServers_ proxyCfg' (cfgMS msType) testProxyCounts msType
+      where
+        proxyCfg' = updateCfg (proxyCfgMS msType) $ \cfg_ ->
+          cfg_
+            { addressStats = Just AddressStatsConfig {period = defaultAddressStatsPeriod},
+              controlPort = Just testControlPort,
+              controlPortUserAuth = Just testControlPortUserAuth,
+              smpAgentCfg = (smpAgentCfg cfg_) {ownServerDomains = ownDomains}
+            }
+        testProxyCounts = do
+          deliverMessageViaProxy (SMPServer testHost2 testPort testKeyHash) (SMPServer testHost2 testPort2 testKeyHash) C.SEd448 "hello 1" "hello 2"
+          withControlPort $ \h -> do
+            controlPortCommand h ("auth " <> SMP.unBasicAuth testControlPortUserAuth) 1 `shouldReturn` ["Current role is CPRUser"]
+            forM_ counts $ \(counter, n) ->
+              controlPortCommand h ("addresses " <> counter) 2 `shouldReturn` ["address,previous,current", "127.0.0.1,0," <> n]
    oneServer test msType = withSmpServerConfigOn (transport @TLS) (updateCfg (proxyCfgMS msType) $ \cfg_ -> cfg_ {msgQueueQuota = 128, maxJournalMsgCount = 256}) testPort $ const test
    twoServers test msType = twoServers_ (proxyCfgMS msType) (proxyCfgMS msType) test msType
    twoServersFirstProxy test msType = twoServers_ (proxyCfgMS msType) (updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {msgQueueQuota = 128, maxJournalMsgCount = 256}) test msType
@@ -196,8 +215,6 @@ 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
@@ -501,32 +518,6 @@ 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
@@ -47,6 +47,7 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (parseAll, parseString)
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server (exportMessages)
+import Simplex.Messaging.Server.AddressStats (AddressStatsConfig (..), defaultAddressStatsPeriod)
 import Simplex.Messaging.Server.Env.STM (AStoreType (..), MsgStore (..), ServerConfig (..), ServerStoreCfg (..), readWriteQueueStore)
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.MsgStore.Journal (JournalStoreConfig (..), QStoreCfg (..), stmQueueStore)
@@ -96,6 +97,7 @@ serverTests = do
  describe "Restore messages" testRestoreMessages
  describe "Restore messages (old / v2)" testRestoreExpireMessages
  describe "Save prometheus metrics" testPrometheusMetrics
+  describe "Client address statistics" testAddressStatsControlPort
  describe "Timing of AUTH error" testTiming
  describe "Message notifications" $ do
    testMessageNotifications
@@ -1254,6 +1256,27 @@ testPrometheusMetrics =
    let cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {prometheusInterval = Just 1}
    withSmpServerConfigOn at cfg' testPort $ \_ -> threadDelay 1000000
    doesFileExist testPrometheusMetricsFile `shouldReturn` True
+    metrics <- B.readFile testPrometheusMetricsFile
+    metrics `shouldSatisfy` ("# TYPE simplex_smp_client_address_period_count histogram" `B.isInfixOf`)
+
+testAddressStatsControlPort :: SpecWith (ASrvTransport, AStoreType)
+testAddressStatsControlPort =
+  it "should list client addresses by counter via control port" $ \(at@(ATransport (_ :: TProxy c 'TServer)), msType) -> do
+    let cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {addressStats = Just AddressStatsConfig {period = defaultAddressStatsPeriod}, controlPort = Just testControlPort, controlPortUserAuth = Just testControlPortUserAuth}
+        ipv4Client = testSMPClient_ @c "127.0.0.1" testPort supportedClientSMPRelayVRange Nothing
+    withSmpServerConfigOn at cfg' testPort $ \_ ->
+      ipv4Client $ \r -> ipv4Client $ \s -> do
+        g <- C.newRandom
+        (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd448 g
+        (sId, _, _, _) <- createAndSecureQueue r sPub
+        Resp "bcda" _ OK <- signSendRecv s sKey ("bcda", sId, _SEND "hello")
+        Resp "cdab" _ OK <- signSendRecv s sKey ("cdab", sId, _SEND "hello again")
+        withControlPort $ \h -> do
+          controlPortCommand h "addresses SEND" 1 `shouldReturn` ["AUTH"]
+          controlPortCommand h ("auth " <> unBasicAuth testControlPortUserAuth) 1 `shouldReturn` ["Current role is CPRUser"]
+          controlPortCommand h "addresses unknown" 1 `shouldReturn` ["error: unknown counter"]
+          controlPortCommand h "addresses SEND" 2 `shouldReturn` ["address,previous,current", "127.0.0.1,0,2"]
+          controlPortCommand h "addresses connections" 2 `shouldReturn` ["address,previous,current", "127.0.0.1,0,2"]

 createAndSecureQueue :: Transport c => THandleSMP c 'TClient -> SndPublicAuthKey -> IO (SenderId, RecipientId, RcvPrivateAuthKey, RcvDhSecret)
 createAndSecureQueue h sPub = do
@@ -7,6 +7,7 @@ import CLITests
 import Control.Concurrent (threadDelay)
 import qualified Control.Exception as E
 import Control.Logger.Simple
+import CoreTests.AddressStatsTests
 import CoreTests.BatchingTests
 import CoreTests.CryptoFileTests
 import CoreTests.CryptoTests
@@ -83,6 +84,7 @@ main = do
      . after_ (eventuallyRemove "tests/tmp" 3)
      $ do
        describe "Core tests" $ do
+          describe "Address statistics tests" addressStatsTests
          describe "Batching tests" batchingTests
          describe "Encoding tests" encodingTests
          describe "Version range" versionRangeTests
@@ -24,6 +24,7 @@ import qualified Simplex.Messaging.Agent as A
 import Simplex.Messaging.Agent.Protocol (UserId)
 import Simplex.Messaging.Crypto.File (CryptoFile)
 import Simplex.Messaging.Protocol (XFTPServer)
+import Simplex.Messaging.Server.AddressStats (AddressStatsConfig (..))
 import Simplex.Messaging.Transport.HTTP2 (httpALPN)
 import Simplex.Messaging.Transport.Server
 import Test.Hspec hiding (fit, it)
@@ -208,6 +209,7 @@ testXFTPServerConfig =
      serverStatsBackupFile = Nothing,
      prometheusInterval = Nothing,
      prometheusMetricsFile = xftpTestPrometheusMetricsFile,
+      addressStats = Just AddressStatsConfig {period = 1},
      transportConfig = mkTransportServerConfig True (Just alpnSupportedXFTPhandshakes) False,
      responseDelay = 0,
      webStaticPath = Nothing
@@ -28,6 +28,7 @@ import qualified Data.X509 as X
 import Data.X509.Validation (Fingerprint (..), getFingerprint)
 import Network.HPACK.Token (tokenKey)
 import qualified Network.HTTP2.Client as H2
+import SMPClient (controlPortCommand, testControlPort, testControlPortUserAuth, withControlPort)
 import ServerTests (logSize)
 import Simplex.FileTransfer.Client hiding (createXFTPChunk)
 import qualified Simplex.FileTransfer.Client as A
@@ -39,7 +40,8 @@ import Simplex.Messaging.Client (ProtocolClientError (..))
 import qualified Simplex.Messaging.Crypto as C
 import qualified Simplex.Messaging.Crypto.Lazy as LC
 import Simplex.Messaging.Encoding (smpDecode, smpEncode)
-import Simplex.Messaging.Protocol (BasicAuth, EntityId (..), RecipientId, SenderId, pattern NoEntity)
+import Simplex.Messaging.Protocol (BasicAuth (..), EntityId (..), RecipientId, SenderId, pattern NoEntity)
+import Simplex.Messaging.Server.AddressStats (AddressStatsConfig (..), defaultAddressStatsPeriod)
 import Simplex.Messaging.Server.Expiration (ExpirationConfig (..))
 import Simplex.Messaging.Transport (CertChainPubKey (..), TLS (..), TransportPeer (..), defaultSupportedParams, defaultSupportedParamsHTTPS)
 import Simplex.Messaging.Transport.Client (TransportClientConfig (..), TransportHost (..), defaultTransportClientConfig, runTLSTransportClient)
@@ -68,6 +70,7 @@ xftpServerTests =
      it "should disconnect inactive clients" testInactiveClientExpiration
      it "should not allow uploading chunks after specified storage quota" testFileStorageQuota
      it "should store file records to log and restore them after server restart" testFileLog
+      it "should count commands, transfers and recipients by client address" testFileAddressStats
      describe "XFTP basic auth" $ do
        --                                               allow FNEW | server auth | clnt auth | success
        it "prohibited without basic auth" $ testFileBasicAuth True (Just "pwd") Nothing False
@@ -334,6 +337,23 @@ testFileLog _ = do
      downloadXFTPChunk g c rpKey rId $ XFTPRcvChunkSpec "tests/tmp/received_chunk1" chSize digest
      liftIO $ B.readFile "tests/tmp/received_chunk1" `shouldReturn` bytes

+testFileAddressStats :: AFStoreType -> Expectation
+testFileAddressStats fsType =
+  withXFTPServerConfigOn (updateXFTPCfg (cfgFS fsType) $ \c -> c {addressStats, controlPort = Just testControlPort, controlPortUserAuth = Just testControlPortUserAuth}) $ \_ -> do
+    ts <- getCurrentTime
+    runRight_ $ do
+      c <- ExceptT $ getXFTPClient (1, "xftp://[email protected]:8000", Nothing) testXFTPClientConfig [] ts (\_ -> pure Nothing) (\_ -> pure ())
+      runTestFileChunkDelivery c c
+    withControlPort $ \h -> do
+      controlPortCommand h ("auth " <> unBasicAuth testControlPortUserAuth) 1 `shouldReturn` ["Current role is CPRUser"]
+      controlPortCommand h "addresses errors" 1 `shouldReturn` ["address,previous,current"]
+      forM_ counts $ \(counter, n) ->
+        controlPortCommand h ("addresses " <> counter) 2 `shouldReturn` ["address,previous,current", "127.0.0.1,0," <> n]
+  where
+    addressStats = Just AddressStatsConfig {period = defaultAddressStatsPeriod}
+    counts :: [(ByteString, ByteString)]
+    counts = [("connections", "1"), ("FNEW", "2"), ("FPUT", "2"), ("FGET", "2"), ("recipients", "2"), ("upload_kb", "128"), ("download_kb", "256")]
+
 testFileBasicAuth :: Bool -> Maybe BasicAuth -> Maybe BasicAuth -> Bool -> AFStoreType -> IO ()
 testFileBasicAuth allowNewFiles newFileBasicAuth clntAuth success fsType =
  withXFTPServerConfigOn (updateXFTPCfg (cfgFS fsType) $ \c -> c {allowNewFiles, newFileBasicAuth}) $