smp-server: limit concurrent name resolutions

src/Simplex/Messaging/Server.hs

@@ -1552,7 +1552,7 @@ client
         Nothing -> incStat (rslvDisabled st) $> Nothing
         Just nenv -> pure (Just nenv)
     -- Runs on a forked thread so RSLV does not block other commands;
-    -- concurrency is limited in forkCmd by serverClientConcurrency for direct and serverResolverConcurrency for forwarded RSLV.
+    -- concurrency is limited in forkCmd by serverClientConcurrency for direct and serverResolverConcurrency for forwarded RSLV, and globally in resolveName.
     resolveNameMsg :: VersionSMP -> NamesEnv -> NameQuery -> M s BrokerMsg
     resolveNameMsg v nenv q = do
       st <- asks (rslvStats . serverStats)

src/Simplex/Messaging/Server/Main.hs

@@ -724,7 +724,8 @@ readNamesConfig ini
               { resolverEndpoint = either (error . ("[NAMES] resolver_endpoint: " <>)) id (validateUrl endpoint resolverAuth_),
                 resolverAuth = resolverAuth_,
                 resolverTimeoutMs = boundedIniInt 3000 100 60000 "resolver_timeout_ms",
-                resolverMaxResponseBytes = boundedIniInt 16000 1024 16000 "resolver_max_response_bytes"
+                resolverMaxResponseBytes = boundedIniInt 16000 1024 16000 "resolver_max_response_bytes",
+                resolverGlobalConcurrency = boundedIniInt 32 1 1000 "resolver_global_concurrency"
               }
   where
     enabled = fromMaybe False (iniOnOff "NAMES" "enable" ini)

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

@@ -169,6 +169,8 @@ iniFileContent cfgPath logPath opts host basicAuth controlPortPwds =
         \# resolver_auth: basic <username>:<password>\n\
         \# resolver_timeout_ms: 3000\n\
         \# resolver_max_response_bytes: 16000\n\
+        \# Max concurrent requests to the resolver from all connections.\n\
+        \# resolver_global_concurrency: 32\n\
         \# Max concurrent forwarded name resolutions per proxy connection (forwarded RSLVs from many\n\
         \# clients share one proxy connection, so this is much higher than PROXY client_concurrency).\n"
     <> ("# resolver_concurrency = " <> tshow defaultNameResolverConcurrency)

src/Simplex/Messaging/Server/Names.hs

@@ -15,6 +15,7 @@ module Simplex.Messaging.Server.Names
   )
 where
 
+import Control.Concurrent.QSem (QSem, newQSem, signalQSem, waitQSem)
 import qualified Control.Exception as E
 import Control.Logger.Simple (logError)
 import Data.Bifunctor (first)
@@ -38,19 +39,22 @@ data NamesConfig = NamesConfig
   { resolverEndpoint :: String,
     resolverAuth :: Maybe RpcAuth,
     resolverTimeoutMs :: Int,
-    resolverMaxResponseBytes :: Int
+    resolverMaxResponseBytes :: Int,
+    resolverGlobalConcurrency :: Int
   }
   deriving (Show)
 
 data NamesEnv = NamesEnv
   { config :: NamesConfig,
-    resolverEnv :: ResolverEnv
+    resolverEnv :: ResolverEnv,
+    resolverSlots :: QSem
   }
 
 newNamesEnv :: NamesConfig -> IO NamesEnv
 newNamesEnv config = do
-  resolverEnv <- newResolverEnv (resolverEndpoint config) (resolverAuth config) (resolverTimeoutMs config) (resolverMaxResponseBytes config)
-  pure NamesEnv {config, resolverEnv}
+  resolverEnv <- newResolverEnv (resolverEndpoint config) (resolverAuth config) (resolverTimeoutMs config) (resolverMaxResponseBytes config) (resolverGlobalConcurrency config)
+  resolverSlots <- newQSem (resolverGlobalConcurrency config)
+  pure NamesEnv {config, resolverEnv, resolverSlots}
 
 closeNamesEnv :: NamesEnv -> IO ()
 closeNamesEnv NamesEnv {resolverEnv} = closeResolverEnv resolverEnv
@@ -60,8 +64,9 @@ pingEndpoint NamesEnv {resolverEnv, config} =
   fromMaybe (Left ResolverTimeout) <$> timeout (resolverTimeoutMs config * 1000) (healthHttp resolverEnv)
 
 resolveName :: NamesEnv -> NameQuery -> IO (Either NameErrorType NameResponse)
-resolveName env q = do
-  r <- E.try (timeout (resolverTimeoutMs (config env) * 1000) (fetch env q))
+resolveName env@NamesEnv {resolverSlots} q = do
+  -- waiting for a slot counts towards the timeout, so lookups do not queue longer than resolverTimeoutMs
+  r <- E.try (timeout (resolverTimeoutMs (config env) * 1000) (E.bracket_ (waitQSem resolverSlots) (signalQSem resolverSlots) (fetch env q)))
   case r of
     Right result -> pure (fromMaybe (Left (RESOLVER "timeout")) result)
     Left e

src/Simplex/Messaging/Server/Names/HttpResolver.hs

@@ -85,9 +85,9 @@ data ResolverError
   | ResolverTimeout
   deriving (Show)
 
-newResolverEnv :: String -> Maybe RpcAuth -> Int -> Int -> IO ResolverEnv
-newResolverEnv baseUrl auth_ timeoutMs maxResponseBytes = do
-  manager <- HC.newManager tlsManagerSettings {managerConnCount = 10}
+newResolverEnv :: String -> Maybe RpcAuth -> Int -> Int -> Int -> IO ResolverEnv
+newResolverEnv baseUrl auth_ timeoutMs maxResponseBytes maxConcurrency = do
+  manager <- HC.newManager tlsManagerSettings {managerConnCount = maxConcurrency}
   pure
     ResolverEnv
       { manager,
@@ -136,8 +136,9 @@ httpGet ResolverEnv {manager, baseUrl, authHdr, timeoutMicro, maxResponseBytes}
           }
   result <- E.try $ withResponse req manager $ \res -> do
     let status = HT.statusCode (responseStatus res)
+    -- http-client closes the connection unless the body is read to the end
     if status >= 400
-      then pure (Left (HttpStatusErr status))
+      then Left (HttpStatusErr status) <$ brReadSome (responseBody res) (maxResponseBytes + 1)
       else do
         bs <- brReadSome (responseBody res) (maxResponseBytes + 1)
         pure $ if BL.length bs > fromIntegral maxResponseBytes then Left BodyTooLarge else Right bs

tests/NamesResolverServer.hs

@@ -10,6 +10,7 @@
 module NamesResolverServer
   ( withResolverServer,
     withResolverServerDelayed,
+    withResolverServerConns,
     resolveResp,
     testNamesConfig,
     memCfg,
@@ -38,9 +39,18 @@ withResolverServer :: ([Text] -> (Status, LB.ByteString)) -> (Int -> IORef [[Tex
 withResolverServer = withResolverServerDelayed 0
 
 withResolverServerDelayed :: Int -> ([Text] -> (Status, LB.ByteString)) -> (Int -> IORef [[Text]] -> IO a) -> IO a
-withResolverServerDelayed delayMs handler action = do
+withResolverServerDelayed delayMs handler action = withResolverServer_ delayMs handler $ \port reqs _ -> action port reqs
+
+-- | Also counts the TCP connections the resolver accepted.
+withResolverServerConns :: ([Text] -> (Status, LB.ByteString)) -> (Int -> IORef [[Text]] -> IORef Int -> IO a) -> IO a
+withResolverServerConns = withResolverServer_ 0
+
+withResolverServer_ :: Int -> ([Text] -> (Status, LB.ByteString)) -> (Int -> IORef [[Text]] -> IORef Int -> IO a) -> IO a
+withResolverServer_ delayMs handler action = do
   reqs <- newIORef []
-  Warp.withApplication (pure (app reqs)) $ \port -> action port reqs
+  conns <- newIORef 0
+  let settings = Warp.setOnOpen (\_ -> True <$ atomicModifyIORef' conns (\n -> (n + 1, ()))) Warp.defaultSettings
+  Warp.withApplicationSettings settings (pure (app reqs)) $ \port -> action port reqs conns
   where
     app :: IORef [[Text]] -> Application
     app reqs req send = do
@@ -63,7 +73,8 @@ testNamesConfig port =
     { resolverEndpoint = "http://127.0.0.1:" <> show port,
       resolverAuth = Nothing,
       resolverTimeoutMs = 1000,
-      resolverMaxResponseBytes = 65536
+      resolverMaxResponseBytes = 65536,
+      resolverGlobalConcurrency = 8
     }
 
 memCfg :: HasTestEnv => AServerConfig

tests/RSLVTests.hs

@@ -51,6 +51,7 @@ import Simplex.Messaging.Protocol
   )
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
+import Simplex.Messaging.Server.Names (NamesConfig (..))
 import Simplex.Messaging.SimplexName (SimplexDomain)
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Version (mkVersionRange)
@@ -108,6 +109,7 @@ rslvTests = do
     it "a record naming a different name is rejected" testRslvWrongName
   describe "RSLV resource use" $ do
     it "one connection has at most client_concurrency lookups in flight" testRslvConnectionCap
+    it "all connections have at most resolver_global_concurrency lookups in flight" testRslvFanOut
     it "a lookup waiting for a slot runs when a slot is released" testRslvSlotWait
     it "a lookup waiting for a slot fails when the client times out" testRslvSlotWaitTimeout
 
@@ -297,6 +299,19 @@ testRslvConnectionCap =
   where
     connCap = 4
 
+testRslvFanOut :: HasTestEnv => IO ()
+testRslvFanOut =
+  NRS.withResolverServerDelayed 3000 (NRS.resolveResp status200 "{}") $ \port reqs ->
+    withSmpServerConfigOn (transport @TLS) (updateCfg (withNames port memCfg) $ \c -> c {serverClientConcurrency = 32}) testPort $ const $
+      testSMPClient @TLS $ \h1 -> testSMPClient @TLS $ \h2 -> do
+        sendRslvs h1 "a" 32
+        sendRslvs h2 "b" 32
+        threadDelay 800000
+        length <$> resolvePaths reqs `shouldReturn` resolverGlobalConcurrency (NRS.testNamesConfig port)
+        let timedOut = replicate 32 (Right (ERR (NAME (RESOLVER "timeout"))))
+        recvResponses h1 32 `shouldReturn` timedOut
+        recvResponses h2 32 `shouldReturn` timedOut
+
 -- 4 lookups take the slots for the 1s resolver timeout, the 5th waits for a slot.
 testRslvSlotWait :: HasTestEnv => IO ()
 testRslvSlotWait =

tests/SMPNamesTests.hs

@@ -5,6 +5,7 @@
 
 module SMPNamesTests (smpNamesTests, testNameRecord, testPricing, registeredBody, availableBody, reservedBody, responseBody, resolved) where
 
+import Control.Monad (forM_)
 import qualified Data.Aeson as J
 import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy as LB
@@ -15,7 +16,7 @@ import qualified Data.Map.Strict as M
 import qualified Data.Text as T
 import Data.Text.Encoding (encodeUtf8)
 import Network.HTTP.Types (status200, status400, status404, status500, status502)
-import NamesResolverServer (resolveResp, testNamesConfig, withResolverServer, withResolverServerDelayed)
+import NamesResolverServer (resolveResp, testNamesConfig, withResolverServer, withResolverServerConns, withResolverServerDelayed)
 import Simplex.Messaging.Encoding (smpDecode, smpEncode)
 import Simplex.Messaging.Encoding.String (strDecode)
 import Simplex.Messaging.Protocol (Command (..), ErrorType (..), NameErrorType (..), NamePricing (..), NameQuery (..), NameRecord (..), NameRegistration (..), NameResponse (..), NameReservedReason (..), ProtocolEncoding (..), USDCents (..))
@@ -302,6 +303,13 @@ resolverSpec = do
       _ <- resolveName env aliceDomain
       readIORef reqs >>= \rs -> length rs `shouldBe` 2
 
+  it "keeps the resolver connection alive across error responses" $
+    withResolverServerConns (resolveResp status502 "{\"error\":\"upstream\"}") $ \port reqs conns -> do
+      env <- newNamesEnv (testNamesConfig port)
+      forM_ [1 .. 5 :: Int] $ \_ -> resolveName env aliceDomain `shouldReturn` Left (RESOLVER "HTTP 502")
+      length <$> readIORef reqs `shouldReturn` 5
+      readIORef conns `shouldReturn` 1
+
   it "addresses the resolver with the full canonical domain name" $
     withResolverServer (resolveResp status200 (registeredBody testNameRecord)) $ \port reqs -> do
       env <- newNamesEnv (testNamesConfig port)