smp-server: count forwards above relay limit

src/Simplex/Messaging/Server.hs

@@ -1455,7 +1455,7 @@ client
                       _ -> ERR $ transportErr TEVersion
       PFWD fwdV pubKey encBlock -> do
         ProxyAgent {smpAgent = a, relayForwards} <- asks proxyAgent
-        ServerStats {pMsgFwds, pMsgFwdsOwn} <- asks serverStats
+        ServerStats {pMsgFwds, pMsgFwdsOwn, pMsgFwdsRelayLimit} <- asks serverStats
         let inc = mkIncProxyStats pMsgFwds pMsgFwdsOwn
         liftIO (lookupSMPServerClient a sessId) >>= \case
           Just (own, smp) -> do
@@ -1465,7 +1465,7 @@ client
             forkProxiedCmd $
               bracket (atomically $ reserve maxFwds relayForwards) (\r -> when r $ atomically $ release relayForwards) $ \case
                 -- the relay is not keeping up, the client retries TIMEOUT later
-                False -> ERR (smpProxyError PCEResponseTimeout) <$ inc own pErrorsOther
+                False -> ERR (smpProxyError PCEResponseTimeout) <$ (inc own pErrorsOther >> incStat pMsgFwdsRelayLimit)
                 True ->
                   liftIO (runExceptT (forwardSMPTransmission smp corrId fwdV pubKey encBlock) `E.catches` clientHandlers) >>= \case
                     Right r -> PRES r <$ inc own pSuccesses

src/Simplex/Messaging/Server/Prometheus.hs

@@ -122,6 +122,7 @@ prometheusMetrics sm rtm ts =
         _pMsgFwds,
         _pMsgFwdsOwn,
         _pMsgFwdsRecv,
+        _pMsgFwdsRelayLimit,
         _rcvServices,
         _ntfServices,
         _rcvServicesSubMsg,
@@ -370,6 +371,10 @@ prometheusMetrics sm rtm ts =
       \simplex_smp_relay_messages_errors{source=\"own\",type=\"compat\"} " <> mshow (_pErrorsCompat _pMsgFwdsOwn) <> "\n# pMsgFwdsOwn_pErrorsCompat\n\
       \simplex_smp_relay_messages_errors{source=\"own\",type=\"other\"} " <> mshow (_pErrorsOther _pMsgFwdsOwn) <> "\n# pMsgFwdsOwn_pErrorsOther\n\
       \\n\
+      \# HELP simplex_smp_relay_messages_concurrency_limit Messages not sent through relay, as the limit of messages in flight to relay was reached.\n\
+      \# TYPE simplex_smp_relay_messages_concurrency_limit counter\n\
+      \simplex_smp_relay_messages_concurrency_limit " <> mshow _pMsgFwdsRelayLimit <> "\n# pMsgFwdsRelayLimit\n\
+      \\n\
       \# HELP simplex_smp_relay_messages_received Relay messages statistics.\n\
       \# TYPE simplex_smp_relay_messages_received counter\n\
       \simplex_smp_relay_messages_received " <> mshow _pMsgFwdsRecv <> "\n# pMsgFwdsRecv\n\

src/Simplex/Messaging/Server/Stats.hs

@@ -124,6 +124,7 @@ data ServerStats = ServerStats
     pMsgFwds :: ProxyStats,
     pMsgFwdsOwn :: ProxyStats,
     pMsgFwdsRecv :: IORef Int,
+    pMsgFwdsRelayLimit :: IORef Int,
     rcvServices :: ServiceStats,
     ntfServices :: ServiceStats,
     rcvServicesSubMsg :: IORef Int,
@@ -186,6 +187,7 @@ data ServerStatsData = ServerStatsData
     _pMsgFwds :: ProxyStatsData,
     _pMsgFwdsOwn :: ProxyStatsData,
     _pMsgFwdsRecv :: Int,
+    _pMsgFwdsRelayLimit :: Int,
     _ntfServices :: ServiceStatsData,
     _rcvServices :: ServiceStatsData,
     _rcvServicesSubMsg :: Int,
@@ -250,6 +252,7 @@ newServerStats ts = do
   pMsgFwds <- newProxyStats
   pMsgFwdsOwn <- newProxyStats
   pMsgFwdsRecv <- newIORef 0
+  pMsgFwdsRelayLimit <- newIORef 0
   rcvServices <- newServiceStats
   ntfServices <- newServiceStats
   rcvServicesSubMsg <- newIORef 0
@@ -311,6 +314,7 @@ newServerStats ts = do
         pMsgFwds,
         pMsgFwdsOwn,
         pMsgFwdsRecv,
+        pMsgFwdsRelayLimit,
         rcvServices,
         ntfServices,
         rcvServicesSubMsg,
@@ -374,6 +378,7 @@ getServerStatsData s = do
   _pMsgFwds <- getProxyStatsData $ pMsgFwds s
   _pMsgFwdsOwn <- getProxyStatsData $ pMsgFwdsOwn s
   _pMsgFwdsRecv <- readIORef $ pMsgFwdsRecv s
+  _pMsgFwdsRelayLimit <- readIORef $ pMsgFwdsRelayLimit s
   _rcvServices <- getServiceStatsData $ rcvServices s
   _ntfServices <- getServiceStatsData $ ntfServices s
   _rcvServicesSubMsg <- readIORef $ rcvServicesSubMsg s
@@ -435,6 +440,7 @@ getServerStatsData s = do
         _pMsgFwds,
         _pMsgFwdsOwn,
         _pMsgFwdsRecv,
+        _pMsgFwdsRelayLimit,
         _rcvServices,
         _ntfServices,
         _rcvServicesSubMsg,
@@ -499,6 +505,7 @@ setServerStats s d = do
   setProxyStats (pMsgFwds s) $! _pMsgFwds d
   setProxyStats (pMsgFwdsOwn s) $! _pMsgFwdsOwn d
   writeIORef (pMsgFwdsRecv s) $! _pMsgFwdsRecv d
+  writeIORef (pMsgFwdsRelayLimit s) $! _pMsgFwdsRelayLimit d
   setServiceStats (rcvServices s) $! _rcvServices d
   setServiceStats (ntfServices s) $! _ntfServices d
   writeIORef (rcvServicesSubMsg s) $! _rcvServicesSubMsg d
@@ -701,6 +708,7 @@ instance StrEncoding ServerStatsData where
           _pMsgFwds,
           _pMsgFwdsOwn,
           _pMsgFwdsRecv,
+          _pMsgFwdsRelayLimit = 0,
           _rcvServices,
           _ntfServices,
           _rcvServicesSubMsg = 0,

tests/SMPProxyTests.hs

@@ -46,6 +46,7 @@ import Simplex.Messaging.Util (bshow, tshow)
 import Simplex.Messaging.Version (mkVersionRange)
 import System.FilePath (splitExtensions)
 import System.Random (randomRIO)
+import Text.Read (readMaybe)
 import Test.Hspec hiding (fit, it)
 import UnliftIO
 import Util
@@ -68,6 +69,8 @@ smpProxyTests = do
       testProxyForwardTimeoutStuckSession
     it "does not keep oversized forwarded command" $ \_ ->
       testForwardOversizedNotKept
+    it "rejects forwards above the relay concurrency limit" $ \_ ->
+      testProxyRelayConcurrencyLimit
   describe "agent client reconnection" $ do
     it "reconnects after a connect is cancelled mid-flight" $ \_ ->
       testAgentClientReconnectAfterCancel
@@ -514,6 +517,27 @@ testProxyForwardTimeoutStuckSession =
     rs <- forM ([1 .. 10] :: [Int]) $ \_ -> runExceptT' (proxySMPMessage pc NRMInteractive sess Nothing sId noMsgFlags "hi")
     rs `shouldSatisfy` elem (Left (ProxyProtocolError (SMP.PROXY SMP.NO_SESSION)))
 
+testProxyRelayConcurrencyLimit :: IO ()
+testProxyRelayConcurrencyLimit =
+  withSmpServerConfigOn (transport @TLS) cfg' testPort $ \_ -> do
+    g <- C.newRandom
+    ts <- getCurrentTime
+    let srv = SMPServer testHost testPort testKeyHash
+        vr = mkVersionRange minServerSMPRelayVersion currentClientSMPRelayVersion
+    pc <- either (fail . show) pure =<< getProtocolClient g NRMInteractive (1, srv, Nothing) defaultSMPClientConfig {serverVRange = vr} [] Nothing ts (\_ -> pure ())
+    sess <- runExceptT' $ connectSMPProxiedRelay pc NRMInteractive srv (Just "correct")
+    sId <- atomically $ SMP.EntityId <$> C.randomBytes 24 g
+    rs <- forConcurrently ([1 .. 20] :: [Int]) $ \_ -> runExceptT (proxySMPMessage pc NRMInteractive sess Nothing sId noMsgFlags "hi")
+    let rejected = length $ filter (== Right (Left (ProxyProtocolError (SMP.PROXY (SMP.BROKER SMP.TIMEOUT))))) rs
+    rejected `shouldSatisfy` (> 0)
+    threadDelay 1500000
+    readRelayLimitMetric <$> readFile testPrometheusMetricsFile `shouldReturn` Just rejected
+  where
+    cfg' = updateCfg proxyCfg $ \cfg_ -> cfg_ {proxyRelayConcurrency = 1, prometheusInterval = Just 1}
+    readRelayLimitMetric m = case [v | "simplex_smp_relay_messages_concurrency_limit" : v : _ <- map words (lines m)] of
+      v : _ -> readMaybe v
+      [] -> Nothing
+
 testForwardOversizedNotKept :: IO ()
 testForwardOversizedNotKept =
   withSmpServerConfigOn (transport @TLS) proxyCfg testPort $ \_ -> do