implement

2026-10-12 02:22:45 +00:00 · 2026-09-27 17:01:06 +00:00
parent 629d5d8204
commit f93ed11d1a
27 changed files with 567 additions and 105 deletions
@@ -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
@@ -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)
@@ -7,15 +7,18 @@ module Simplex.FileTransfer.Server.Control
 where

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

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

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

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

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

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

 rtsOptionsEnv :: Text
@@ -33,9 +37,9 @@ rtsOptionsEnv = "XFTP_RTS_OPTIONS"
 {-# FOURMOLU_DISABLE\n#-}
 xftpPrometheusMetrics :: FileServerMetrics -> UTCTime -> Text
 xftpPrometheusMetrics sm ts =
-  time <> files <> info
+  time <> files <> clientAddresses <> info
  where
-    FileServerMetrics {statsData, filesDownloadedPeriods, rtsOptions} = sm
+    FileServerMetrics {statsData, filesDownloadedPeriods, rtsOptions, addressHistograms} = sm
    FileServerStatsData
      { _fromTime,
        _filesCreated,
@@ -122,6 +126,7 @@ xftpPrometheusMetrics sm ts =
      \simplex_xftp_files_count_monthly " <> mstr (monthCount filesDownloadedPeriods) <> "\n\
      \# filesDownloaded.monthCount\n\
      \\n"
+    clientAddresses = maybe "" (addrHistogramMetrics "simplex_xftp" tsEpoch) addressHistograms
    info =
      "# Info\n\
      \# ----\n\
@@ -72,6 +72,7 @@ import Simplex.Messaging.Notifications.Transport
 import Simplex.Messaging.Protocol (EntityId (..), ErrorType (..), NotifierId, Party (..), ProtocolServer (host), SMPServer, ServiceSub (..), SignedTransmission, Transmission, pattern NoEntity, pattern SMPServer, encodeTransmission, tGetServer, tPut)
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Server
+import Simplex.Messaging.Server.AddressStats (AddrStats, ServerAddrStats (..), addressStatsThread, incAddrCounter, printTopAddresses, withAddrStats)
 import Simplex.Messaging.Server.Control (CPClientRole (..))
 import Simplex.Messaging.Server.Env.STM (StartOptions (..))
 import Simplex.Messaging.Server.Stats (PeriodStats (..), PeriodStatCounts (..), periodStatCounts, periodStatDataCounts, updatePeriodStats)
@@ -80,7 +81,7 @@ import Simplex.Messaging.SystemTime
 import Simplex.Messaging.TMap (TMap)
 import Simplex.Messaging.Transport (ASrvTransport, ATransport (..), THandle (..), THandleAuth (..), THandleParams (..), TProxy, Transport (..), TransportPeer (..), defaultSupportedParams)
 import Simplex.Messaging.Transport.Buffer (trimCR)
-import Simplex.Messaging.Transport.Server (AddHTTP, runTransportServer, runLocalTCPServer)
+import Simplex.Messaging.Transport.Server (AddHTTP, TLSServerCredential (..), newSocketState, runLocalTCPServer, runTransportServerState_)
 import Simplex.Messaging.Util
 import System.Environment (lookupEnv)
 import System.Exit (exitFailure, exitSuccess)
@@ -122,6 +123,7 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg, startOptions}
        : map runServer transports
          <> serverStatsThread_ cfg
          <> prometheusMetricsThread_ cfg
+          <> addressStatsThread_ cfg
          <> controlPortThread_ cfg
    )
    `finally` stopServer
@@ -131,16 +133,20 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg, startOptions}
      srvCreds <- asks tlsServerCreds
      serverSignKey <- either fail pure $ C.x509ToPrivate' $ snd srvCreds
      env <- ask
-      liftIO $ runTransportServer started tcpPort defaultSupportedParams srvCreds tCfg $ \h -> runClient serverSignKey t h `runReaderT` env
+      ss <- liftIO newSocketState
+      liftIO $ runTransportServerState_ ss started tcpPort defaultSupportedParams TLSServerCredential {credential = srvCreds, sniCredential = Nothing} tCfg $ \sock (_, h) ->
+        runClient serverSignKey t sock h `runReaderT` env

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

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

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

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

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

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

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

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

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

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

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

 data NtfSMPWorkerMetrics = NtfSMPWorkerMetrics {ownServers :: [Text], otherServers :: Int}
@@ -59,7 +62,7 @@ data NtfSMPSubMetrics = NtfSMPSubMetrics {ownSrvSubs :: M.Map Text Int, otherSer
 {-# FOURMOLU_DISABLE\n#-}
 ntfPrometheusMetrics :: NtfServerMetrics -> NtfRealTimeMetrics -> UTCTime -> Text
 ntfPrometheusMetrics sm rtm ts =
-  time <> tokens <> subscriptions <> notifications <> info
+  time <> tokens <> subscriptions <> notifications <> clientAddresses <> info
  where
    NtfServerMetrics {statsData, activeTokensCounts = psTkns, activeSubsCounts = psSubs, tokenCount, approxSubCount, lastNtfCount, rtsOptions} = sm
    NtfRealTimeMetrics
@@ -72,7 +75,8 @@ ntfPrometheusMetrics sm rtm ts =
        ntfPendingServiceSubs,
        ntfPendingQueueSubs,
        smpSessionCount,
-        apnsPushQLength
+        apnsPushQLength,
+        addressHistograms
      } = rtm
    NtfServerStatsData
      { _fromTime,
@@ -221,6 +225,7 @@ ntfPrometheusMetrics sm rtm ts =
      <> showNtfsByServer _ntfReceivedAuthOwn "simplex_ntf_notifications_received_auth_own" "Received notifications without token or subscription (AUTH error)" "ntfReceivedAuthOwn"
      <> showNtfsByServer _ntfDeliveredOwn "simplex_ntf_notifications_delivered_own" "Delivered notifications" "ntfDeliveredOwn"
      <> showNtfsByServer _ntfFailedOwn "simplex_ntf_notifications_failed_own" "Failed notifications" "ntfFailedOwn"
+    clientAddresses = maybe "" (addrHistogramMetrics "simplex_ntf" tsEpoch) addressHistograms
    info =
      "# Info\n\
      \# ----\n\
@@ -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
@@ -44,6 +44,8 @@ module Simplex.Messaging.Server.CLI
    checkSavedFingerprint,
    iniTransports,
    iniDBOptions,
+    iniAddressStats,
+    addressStatsIniContent,
    printServerConfig,
    printServerTransports,
    printSMPServerConfig,
@@ -73,13 +75,14 @@ import Simplex.Messaging.Agent.Store.Postgres.Options (DBOpts (..))
 import Simplex.Messaging.Agent.Store.Shared (MigrationConfirmation (..))
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (ProtoServerWithAuth (..), ProtocolServer (..), ProtocolTypeI)
+import Simplex.Messaging.Server.AddressStats (AddressStatsConfig (..), defaultAddressStatsPeriod)
 import Simplex.Messaging.Server.Env.STM (ServerStoreCfg (..), StartOptions (..), dbStoreCfg, storeLogFile')
 import Simplex.Messaging.Server.Main.GitCommit
 import Simplex.Messaging.Server.QueueStore.Postgres.Config (PostgresStoreCfg (..))
 import Simplex.Messaging.Transport (ASrvTransport, ATransport (..), TLS, Transport (..), simplexMQVersion)
 import Simplex.Messaging.Transport.Server (AddHTTP, loadFileFingerprint)
 import Simplex.Messaging.Transport.WebSockets (WS)
-import Simplex.Messaging.Util (eitherToMaybe, whenM)
+import Simplex.Messaging.Util (eitherToMaybe, tshow, whenM)
 import System.Directory (doesDirectoryExist, listDirectory, removeDirectoryRecursive, removePathForcibly)
 import System.Environment (lookupEnv)
 import System.Exit (exitFailure)
@@ -435,6 +438,19 @@ iniDBOptions ini _default@DBOpts {connstr, schema, poolSize} =
      createSchema = False
    }

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

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

 data CPClientRole = CPRNone | CPRUser | CPRAdmin
  deriving (Eq, Show)
@@ -28,6 +30,7 @@ data ControlProtocol
  | CPStatus SenderId
  | CPBlock SenderId BlockingInfo
  | CPUnblock SenderId
+  | CPAddresses Text (Maybe Int)
  | CPSave
  | CPHelp
  | CPQuit
@@ -49,6 +52,7 @@ instance StrEncoding ControlProtocol where
    CPStatus sId -> "status " <> strEncode sId
    CPBlock sId info -> "block " <> strEncode sId <> " " <> strEncode info
    CPUnblock sId -> "unblock " <> strEncode sId
+    CPAddresses name n_ -> "addresses " <> addressesArgs name n_
    CPSave -> "save"
    CPHelp -> "help"
    CPQuit -> "quit"
@@ -69,6 +73,7 @@ instance StrEncoding ControlProtocol where
      "status" -> CPStatus <$> _strP
      "block" -> CPBlock <$> _strP <*> _strP
      "unblock" -> CPUnblock <$> _strP
+      "addresses" -> uncurry CPAddresses <$> addressesArgsP
      "save" -> pure CPSave
      "help" -> pure CPHelp
      "quit" -> pure CPQuit
@@ -31,6 +31,8 @@ module Simplex.Messaging.Server.Env.STM
    SubscribedClients,
    ProxyAgent (..),
    Client (..),
+    SMPAddrCounter (..),
+    smpCmdCounter,
    ClientId,
    ClientSub (..),
    Sub (..),
@@ -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 ()
@@ -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,
@@ -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")
@@ -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\
@@ -54,7 +54,7 @@ getHTTP2Server HTTP2ServerConfig {qSize, http2Port, bufferSize, bodyHeadSize, se
  started <- newEmptyTMVarIO
  reqQ <- newTBQueueIO qSize
  action <- async $
-    runHTTP2Server started http2Port bufferSize serverSupported srvCreds Nothing transportConfig Nothing (const $ pure ()) $ \_sniUsed sessionId sessionALPN r sendResponse -> do
+    runHTTP2Server started http2Port bufferSize serverSupported srvCreds Nothing transportConfig Nothing (const $ pure ()) (\_ _ handler -> handler) $ \_sniUsed sessionId sessionALPN r sendResponse -> do
      reqBody <- getHTTP2Body r bodyHeadSize
      atomically $ writeTBQueue reqQ HTTP2Request {sessionId, sessionALPN, request = r, reqBody, sendResponse}
  void . atomically $ takeTMVar started
@@ -63,13 +63,13 @@ getHTTP2Server HTTP2ServerConfig {qSize, http2Port, bufferSize, bodyHeadSize, se
 closeHTTP2Server :: HTTP2Server -> IO ()
 closeHTTP2Server = uninterruptibleCancel . action

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

 -- HTTP2 server can be run on both client and server TLS connections.
 runHTTP2ServerWith :: BufferSize -> ((TLS p -> IO ()) -> a) -> HTTP2ServerFunc -> a
@@ -108,6 +108,8 @@ smpServerTest storeLog basicAuth = do
  lookupValue "TRANSPORT" "websockets" ini `shouldBe` Right "off"
  lookupValue "AUTH" "new_queues" ini `shouldBe` Right "on"
  lookupValue "INACTIVE_CLIENTS" "disconnect" ini `shouldBe` Right "on"
+  lookupValue "ADDRESS_STATS" "enable" ini `shouldBe` Right "off"
+  lookupValue "ADDRESS_STATS" "period" ini `shouldBe` Right "300"
  doesFileExist (cfgPath <> "/ca.key") `shouldReturn` True
  -- start
  r <- lines <$> capture_ (withArgs ["start"] $ (100000 `timeout` smpServerCLI cfgPath logPath) `catchAll_` pure (Just ()))
@@ -234,6 +236,8 @@ ntfServerTest storeLog = do
  lookupValue "STORE_LOG" "log_stats" ini `shouldBe` Right "off"
  lookupValue "TRANSPORT" "port" ini `shouldBe` Right "443"
  lookupValue "TRANSPORT" "websockets" ini `shouldBe` Right "off"
+  lookupValue "ADDRESS_STATS" "enable" ini `shouldBe` Right "off"
+  lookupValue "ADDRESS_STATS" "period" ini `shouldBe` Right "300"
  doesFileExist (ntfCfgPath <> "/ca.key") `shouldReturn` True
  r <- lines <$> capture_ (withArgs ["start"] $ (100000 `timeout` ntfServerCLI ntfCfgPath ntfLogPath) `catchAll_` pure (Just ()))
  r `shouldContain` ["SMP notifications server v" <> simplexmqVersionCommit]
@@ -252,6 +256,8 @@ xftpServerTest storeLog = do
  lookupValue "STORE_LOG" "enable" ini `shouldBe` Right (if storeLog then "on" else "off")
  lookupValue "STORE_LOG" "log_stats" ini `shouldBe` Right "off"
  lookupValue "TRANSPORT" "port" ini `shouldBe` Right "443"
+  lookupValue "ADDRESS_STATS" "enable" ini `shouldBe` Right "off"
+  lookupValue "ADDRESS_STATS" "period" ini `shouldBe` Right "300"
  doesFileExist (fileCfgPath <> "/ca.key") `shouldReturn` True
  r <- lines <$> capture_ (withArgs ["start"] $ (100000 `timeout` xftpServerCLI fileCfgPath fileLogPath) `catchAll_` pure (Just ()))
  r `shouldContain` ["SimpleX XFTP server v" <> simplexmqVersionCommit]
@@ -49,6 +49,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
@@ -159,6 +160,7 @@ ntfServerCfg =
      serverStatsBackupFile = Nothing,
      prometheusInterval = Nothing,
      prometheusMetricsFile = ntfTestPrometheusMetricsFile,
+      addressStats = Just AddressStatsConfig {period = 1},
      ntfServerVRange = supportedServerNTFVRange,
      transportConfig = mkTransportServerConfig True (Just alpnSupportedNTFHandshakes) False,
      startOptions = defaultStartOptions
@@ -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 Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
@@ -37,16 +38,19 @@ import Simplex.Messaging.Crypto.Entitlement (Entitlement (..), EntitlementCreden
 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
@@ -90,6 +94,12 @@ ntfTestPort = "6001"
 ntfTestPort2 :: ServiceName
 ntfTestPort2 = "6002"

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

@@ -288,6 +298,7 @@ cfgMS msType = withStoreCfg (testServerStoreConfig msType) $ \serverStoreCfg ->
      serverStatsBackupFile = Nothing,
      prometheusInterval = Nothing,
      prometheusMetricsFile = testPrometheusMetricsFile,
+      addressStats = Just AddressStatsConfig {period = 1},
      pendingENDInterval = 500000,
      ntfDeliveryInterval = 200000,
      smpCredentials =
@@ -423,6 +434,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

@@ -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
@@ -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)
@@ -97,6 +98,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
@@ -1268,6 +1270,27 @@ testPrometheusMetrics =
    let cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {prometheusInterval = Just 1}
    withSmpServerConfigOn at cfg' testPort $ \_ -> threadDelay 1000000
    doesFileExist testPrometheusMetricsFile `shouldReturn` True
+    metrics <- B.readFile testPrometheusMetricsFile
+    metrics `shouldSatisfy` ("# TYPE simplex_smp_client_address_period_count histogram" `B.isInfixOf`)
+
+testAddressStatsControlPort :: SpecWith (ASrvTransport, AStoreType)
+testAddressStatsControlPort =
+  it "should list client addresses by counter via control port" $ \(at@(ATransport (_ :: TProxy c 'TServer)), msType) -> do
+    let cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {addressStats = Just AddressStatsConfig {period = defaultAddressStatsPeriod}, controlPort = Just testControlPort, controlPortUserAuth = Just testControlPortUserAuth}
+        ipv4Client = testSMPClient_ @c "127.0.0.1" testPort supportedClientSMPRelayVRange Nothing
+    withSmpServerConfigOn at cfg' testPort $ \_ ->
+      ipv4Client $ \r -> ipv4Client $ \s -> do
+        g <- C.newRandom
+        (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd448 g
+        (sId, _, _, _) <- createAndSecureQueue r sPub
+        Resp "bcda" _ OK <- signSendRecv s sKey ("bcda", sId, _SEND "hello")
+        Resp "cdab" _ OK <- signSendRecv s sKey ("cdab", sId, _SEND "hello again")
+        withControlPort $ \h -> do
+          controlPortCommand h "addresses SEND" 1 `shouldReturn` ["AUTH"]
+          controlPortCommand h ("auth " <> unBasicAuth testControlPortUserAuth) 1 `shouldReturn` ["Current role is CPRUser"]
+          controlPortCommand h "addresses unknown" 1 `shouldReturn` ["error: unknown counter"]
+          controlPortCommand h "addresses SEND" 2 `shouldReturn` ["address,previous,current", "127.0.0.1,0,2"]
+          controlPortCommand h "addresses connections" 2 `shouldReturn` ["address,previous,current", "127.0.0.1,0,2"]

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

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