diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index 7a60ff1e3..15d4d7ce9 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/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 diff --git a/src/Simplex/Messaging/Server/Prometheus.hs b/src/Simplex/Messaging/Server/Prometheus.hs index 85d2624fd..c1420dc7a 100644 --- a/src/Simplex/Messaging/Server/Prometheus.hs +++ b/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\ diff --git a/src/Simplex/Messaging/Server/Stats.hs b/src/Simplex/Messaging/Server/Stats.hs index 11a7de666..c62218094 100644 --- a/src/Simplex/Messaging/Server/Stats.hs +++ b/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, diff --git a/tests/SMPProxyTests.hs b/tests/SMPProxyTests.hs index a332a1a54..56cd62068 100644 --- a/tests/SMPProxyTests.hs +++ b/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