servers: usage statistics by command

plans/2026-09-26-client-address-stats.md

@@ -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.

simplexmq.cabal

@@ -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

src/Simplex/FileTransfer/Server.hs

@@ -65,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 (..))
@@ -101,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]
@@ -141,6 +143,7 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira
         : expireFiles fileExpiration
         : serverStatsThread_ cfg
           <> prometheusMetricsThread_ cfg
+          <> addressStatsThread_ cfg
           <> controlPortThread_ cfg
     )
     `finally` stopServer
@@ -154,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
@@ -330,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]
@@ -402,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
@@ -421,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
@@ -490,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
@@ -509,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
@@ -532,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
@@ -567,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
@@ -605,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
@@ -634,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)

src/Simplex/FileTransfer/Server/Control.hs

@@ -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

src/Simplex/FileTransfer/Server/Env.hs

@@ -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)

src/Simplex/FileTransfer/Server/Main.hs

@@ -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

src/Simplex/FileTransfer/Server/Prometheus.hs

@@ -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\

src/Simplex/Messaging/Notifications/Server.hs

@@ -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

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

@@ -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

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

@@ -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}

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

@@ -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

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

@@ -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\

src/Simplex/Messaging/Server.hs

@@ -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 (ProtocolClient (thParams), ProtocolClientError (..), SMPClient, SMPClientError, clientHandlers, forwardSMPTransmission, smpProxyError, temporaryClientError)
+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 ()
@@ -1144,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
@@ -1154,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
@@ -1181,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
@@ -1376,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) =
@@ -1421,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
@@ -1459,16 +1483,24 @@ client
         liftIO (lookupSMPServerClient a sessId) >>= \case
           Just (own, smp) -> do
             inc own pRequests
+            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 <$ inc own pSuccesses
+                Right r -> PRES r <$ delivered
                 Left e -> ERR (smpProxyError e) <$ case e of
-                  PCEProtocolError {} -> inc own pSuccesses
-                  _ -> inc own pErrorsOther
-          Nothing -> inc False pRequests >> inc False pErrorsConnect $> Just (ERR $ PROXY NO_SESSION)
+                  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)
+        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

src/Simplex/Messaging/Server/AddressStats.hs

@@ -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"

src/Simplex/Messaging/Server/CLI.hs

@@ -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

src/Simplex/Messaging/Server/Control.hs

@@ -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

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

@@ -31,6 +31,8 @@ module Simplex.Messaging.Server.Env.STM
     SubscribedClients,
     ProxyAgent (..),
     Client (..),
+    SMPAddrCounter (..),
+    smpCmdCounter,
     ClientId,
     ClientSub (..),
     Sub (..),
@@ -111,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
@@ -183,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
@@ -285,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
@@ -465,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))
@@ -520,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
@@ -556,7 +669,8 @@ newClient clientId qSize clientTHParams createdAt = do
         connected,
         createdAt,
         rcvActiveAt,
-        sndActiveAt
+        sndActiveAt,
+        clientAddrStats
       }
 
 newSubscription :: SubscriptionThread -> STM Sub
@@ -571,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
@@ -626,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,
@@ -642,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 ()

src/Simplex/Messaging/Server/Main.hs

@@ -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,

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

@@ -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 (..))
@@ -176,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")

src/Simplex/Messaging/Server/Prometheus.hs

@@ -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\

src/Simplex/Messaging/Transport/HTTP2/Server.hs

@@ -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

tests/CLITests.hs

@@ -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]

tests/CoreTests/AddressStatsTests.hs

@@ -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

tests/NtfClient.hs

@@ -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

tests/SMPClient.hs

@@ -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 =
@@ -396,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
 

tests/SMPProxyTests.hs

@@ -33,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
@@ -149,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

tests/ServerTests.hs

@@ -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

tests/Test.hs

@@ -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

tests/XFTPClient.hs

@@ -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

tests/XFTPServerTests.hs

@@ -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}) $