smp-server: enforce command concurrency limit (#1944)

* smp-server: hold command slot until completion

* smp-server: command slot wait timeout and stats

* smp-server: clarify command slot comments

* smp-server: simplify command slot fork

* smp-server: limit direct RSLV per client

* smp-server: distinct error for RSLV slot timeout

* smp-server: client command slot wait timeout

* smp-server: reply to forwarded RSLV slot timeout

* smp-server: raise default concurrency limits

* smp-server: simplify command slot wait

src/Simplex/Messaging/Server.hs

@@ -98,8 +98,8 @@ 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.Agent (OwnServer, SMPClientAgent (..), SMPClientAgentEvent (..), closeSMPClientAgent, getSMPServerClient'', isOwnServer, lookupSMPServerClient, getConnectedSMPServerClient)
+import Simplex.Messaging.Client (NetworkConfig (..), NetworkRequestMode (..), ProtocolClient (thParams), ProtocolClientConfig (..), ProtocolClientError (..), SMPClient, SMPClientError, clientHandlers, forwardSMPTransmission, netTimeoutInt, smpProxyError, temporaryClientError)
+import Simplex.Messaging.Client.Agent (OwnServer, SMPClientAgent (..), SMPClientAgentConfig (..), SMPClientAgentEvent (..), closeSMPClientAgent, getSMPServerClient'', isOwnServer, lookupSMPServerClient, getConnectedSMPServerClient)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
@@ -1212,7 +1212,23 @@ sendAfterDelay c@Client {sndQ, clientTHParams = THandleParams {sessionId}} stats
   | otherwise = write
   where
     write = atomically $ writeTBQueue sndQ rs
-    microseconds (MkSystemTime s ns) = s * 1000000 + fromIntegral (ns `div` 1000)
+
+updateSlotWaitStats :: ServerStats -> SystemTime -> Bool -> IO ()
+updateSlotWaitStats ServerStats {cmdSlotWaits, cmdSlotWaitMs, cmdSlotWaitMaxMs, cmdSlotWaitTimeouts} since taken = do
+  waitMs <- fromIntegral . (`div` 1000) . subtract (microseconds since) . microseconds <$> getSystemTime
+  incStat cmdSlotWaits
+  atomicModifyIORef'_ cmdSlotWaitMs (+ waitMs)
+  atomicModifyIORef'_ cmdSlotWaitMaxMs (max waitMs)
+  unless taken $ incStat cmdSlotWaitTimeouts
+
+rslvTooManyRequests :: BrokerMsg
+rslvTooManyRequests = ERR $ NAME $ RESOLVER "too many requests"
+
+forwardedCmdWait :: ServerConfig s -> Int
+forwardedCmdWait = (`netTimeoutInt` NRMBackground) . tcpTimeout . networkConfig . smpCfg . smpAgentCfg
+
+microseconds :: SystemTime -> Int64
+microseconds (MkSystemTime s ns) = s * 1000000 + fromIntegral (ns `div` 1000)
 
 send :: Transport c => MVar (THandleSMP c 'TServer) -> Client s -> IO ()
 send th c@Client {sndQ, msgQ, clientTHParams = THandleParams {sessionId}} = do
@@ -1491,21 +1507,45 @@ client
           Nothing -> 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)
-    -- 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
-      bracket_ wait signal . forkClient clnt (B.unpack $ "client $" <> encode sessionId <> " cmd") $
-        -- commands MUST be processed under a reasonable timeout or the client would halt
-        cmdAction >>= \t -> atomically $ writeTBQueue sndQ ([(corrId, entId, t)], [])
+        forkProxiedCmd = forkCmd serverClientConcurrency corrId (EntityId sessId) (ERR $ smpProxyError PCEResponseTimeout)
+    -- Run a slow command on a thread, responding with timeoutErr if it was not started before the client timeout
+    forkCmd :: (ServerConfig s -> Int) -> CorrId -> EntityId -> BrokerMsg -> M s BrokerMsg -> M s (Maybe a)
+    forkCmd concurrency corrId entId timeoutErr cmdAction =
+      -- commands MUST be processed under a reasonable timeout or the client would halt
+      forkCmd_ concurrency maxClientCmdWait (sendResp timeoutErr) $ cmdAction >>= sendResp
+      where
+        sendResp t = atomically $ writeTBQueue sndQ ([(corrId, entId, t)], [])
+    -- Run a slow command on a thread that sends its response
+    forkCmd_ :: (ServerConfig s -> Int) -> (ServerConfig s -> Int) -> M s () -> M s () -> M s (Maybe a)
+    forkCmd_ concurrency waitTimeout onTimeout cmd = do
+      let label = B.unpack $ "client $" <> encode sessionId <> " cmd"
+      -- Exceptions are masked so the slot is never lost between taking it and starting the thread.
+      -- The thread frees the slot when the command finishes; the caller frees it only if forking failed.
+      mask $ \restore ->
+        ifM
+          wait
+          (forkClient clnt label (restore cmd `finally` signal) `onException` signal)
+          (restore onTimeout)
       pure Nothing
       where
+        -- Takes a free slot at once if there is one. Otherwise waits for a slot until the client
+        -- timeout, after which the client no longer expects the response.
+        -- Other commands of this client are not processed while it waits.
         wait = do
           limit <- asks (concurrency . config)
-          atomically $ do
-            used <- readTVar procThreads
-            when (used >= limit) retry
-            writeTVar procThreads $! used + 1
+          ifM (atomically $ takeSlot limit) (pure True) $ do
+            clientTimeout <- asks $ waitTimeout . config
+            waitStarted <- liftIO getSystemTime
+            expired <- liftIO $ registerDelay clientTimeout
+            taken <- atomically $ ifM (takeSlot limit) (pure True) (ifM (readTVar expired) (pure False) retry)
+            stats <- asks serverStats
+            liftIO $ updateSlotWaitStats stats waitStarted taken
+            pure taken
+        takeSlot limit = do
+          used <- readTVar procThreads
+          let free = used < limit
+          when free $ writeTVar procThreads $! used + 1
+          pure free
         signal = atomically $ modifyTVar' procThreads (\t -> t - 1)
     rslvNamesEnv :: M s (Maybe NamesEnv)
     rslvNamesEnv = do
@@ -1515,7 +1555,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 by serverResolverConcurrency in forkCmd.
+    -- concurrency is limited in forkCmd by serverClientConcurrency for direct and serverResolverConcurrency for forwarded RSLV.
     resolveNameMsg :: VersionSMP -> NamesEnv -> NameQuery -> M s BrokerMsg
     resolveNameMsg v nenv q = do
       st <- asks (rslvStats . serverStats)
@@ -1545,7 +1585,7 @@ client
       Cmd SProxyService (RFWD encBlock) -> (response . (corrId, NoEntity,) =<<) <$> processForwardedCommand encBlock
       Cmd SResolver (RSLV d) -> rslvNamesEnv >>= \case
         Nothing -> pure $ response (corrId, NoEntity, ERR (NAME NO_RESOLVER))
-        Just nenv -> forkCmd serverResolverConcurrency corrId NoEntity (resolveNameMsg (thVersion thParams') nenv d)
+        Just nenv -> forkCmd serverClientConcurrency corrId NoEntity rslvTooManyRequests (resolveNameMsg (thVersion thParams') nenv d)
       Cmd SSenderLink command -> case command of
         LKEY k -> withQueue $ \q qr -> checkMode QMMessaging qr $ secureQueue_ q k $>> getQueueLink_ q qr
         LGET -> withQueue $ \q qr -> checkContact qr $ getQueueLink_ q qr
@@ -2181,9 +2221,12 @@ client
             Right t''@(_, (corrId', entId', cmd')) -> case cmd' of
               Cmd SResolver (RSLV d) -> lift $ rslvNamesEnv >>= \case
                 Nothing -> pure $ Just (corrId', entId', ERR (NAME NO_RESOLVER))
-                Just nenv -> forkCmd serverResolverConcurrency corrId NoEntity $ do
-                  msg <- resolveNameMsg (thVersion clntTHParams) nenv d
-                  either ERR id <$> runExceptT (encodeResp (corrId', entId', msg))
+                Just nenv -> do
+                  let sendResolved msg = do
+                        rres <- either ERR id <$> runExceptT (encodeResp (corrId', entId', msg))
+                        atomically $ writeTBQueue sndQ ([(corrId, NoEntity, rres)], [])
+                  forkCmd_ serverResolverConcurrency forwardedCmdWait (sendResolved rslvTooManyRequests) $
+                    resolveNameMsg (thVersion clntTHParams) nenv d >>= sendResolved
               -- INTERNAL because processCommand never returns Nothing for sender commands;
               -- `fst` drops the empty message only returned for SUB.
               _ -> Just . maybe (corrId', entId', ERR INTERNAL) fst <$> lift (processCommand Nothing (Right (M.empty, M.empty, M.empty)) t'')

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

@@ -68,6 +68,7 @@ module Simplex.Messaging.Server.Env.STM
     defaultInactiveClientExpiration,
     defaultProxyClientConcurrency,
     defaultNameResolverConcurrency,
+    defaultMaxClientCmdWait,
     defaultAuthErrorDelayMs,
     defaultMaxJournalMsgCount,
     defaultMaxJournalStateLines,
@@ -201,11 +202,12 @@ data ServerConfig s = ServerConfig
     smpAgentCfg :: SMPClientAgentConfig,
     allowSMPProxy :: Bool, -- auth is the same with `newQueueBasicAuth`
     serverClientConcurrency :: Int,
-    -- | max concurrent name resolutions per connection, enforced in forkCmd.
+    -- | max concurrent forwarded name resolutions per proxy connection, enforced in forkCmd.
     -- Much higher than serverClientConcurrency: forwarded RSLVs from many clients
     -- aggregate over a single proxy->relay connection (only servers send proxied
     -- requests), so bounding them by the per-client limit would throttle unduly.
     serverResolverConcurrency :: Int,
+    maxClientCmdWait :: Int,
     -- | public-namespace resolver config; Nothing disables the names role
     namesConfig :: Maybe NamesConfig,
     -- | server public information
@@ -252,10 +254,13 @@ defaultInactiveClientExpiration =
     }
 
 defaultProxyClientConcurrency :: Int
-defaultProxyClientConcurrency = 32
+defaultProxyClientConcurrency = 500
 
 defaultNameResolverConcurrency :: Int
-defaultNameResolverConcurrency = 1000
+defaultNameResolverConcurrency = 5000
+
+defaultMaxClientCmdWait :: Int
+defaultMaxClientCmdWait = 60000000 -- microseconds
 
 defaultAuthErrorDelayMs :: Int64
 defaultAuthErrorDelayMs = 50

src/Simplex/Messaging/Server/Main.hs

@@ -611,6 +611,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
               allowSMPProxy = True,
               serverClientConcurrency = readIniDefault defaultProxyClientConcurrency "PROXY" "client_concurrency" ini,
               serverResolverConcurrency = readIniDefault defaultNameResolverConcurrency "NAMES" "resolver_concurrency" ini,
+              maxClientCmdWait = defaultMaxClientCmdWait,
               namesConfig = readNamesConfig ini,
               information = serverPublicInfo ini,
               startOptions

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

@@ -154,7 +154,7 @@ iniFileContent cfgPath logPath opts host basicAuth controlPortPwds =
         \# `socks_mode` can be 'onion' for SOCKS proxy to be used for .onion destination hosts only (default)\n\
         \# or 'always' to be used for all destination hosts (can be used if it is an .onion server).\n\
         \# socks_mode = onion\n\n\
-        \# Limit number of threads a client can spawn to process proxy commands in parrallel.\n"
+        \# Limit number of threads a client can spawn to process proxy commands and name resolutions in parallel.\n"
     <> ("# client_concurrency = " <> tshow defaultProxyClientConcurrency)
     <> "\n\n\
         \[NAMES]\n\
@@ -169,7 +169,7 @@ 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 name resolutions per connection (forwarded RSLVs from many\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)
     <> "\n\n\

src/Simplex/Messaging/Server/Prometheus.hs

@@ -127,6 +127,10 @@ prometheusMetrics sm rtm ts =
         _rcvServicesSubMsg,
         _rcvServicesSubDuplicate,
         _authDelayExceeded,
+        _cmdSlotWaits,
+        _cmdSlotWaitMs,
+        _cmdSlotWaitMaxMs,
+        _cmdSlotWaitTimeouts,
         _qCount,
         _msgCount,
         _ntfCount,
@@ -197,6 +201,19 @@ prometheusMetrics sm rtm ts =
       \# HELP simplex_smp_auth_delay_exceeded Responses with AUTH errors sent without delay, as processing took longer than the delay.\n\
       \# TYPE simplex_smp_auth_delay_exceeded counter\n\
       \simplex_smp_auth_delay_exceeded " <> mshow _authDelayExceeded <> "\n# authDelayExceeded\n\
+      \\n\
+      \# HELP simplex_smp_cmd_slot_wait_ms Time commands waited for a free slot of the connection concurrency limit, in ms.\n\
+      \# TYPE simplex_smp_cmd_slot_wait_ms summary\n\
+      \simplex_smp_cmd_slot_wait_ms_sum " <> mshow _cmdSlotWaitMs <> "\n# cmdSlotWaitMs\n\
+      \simplex_smp_cmd_slot_wait_ms_count " <> mshow _cmdSlotWaits <> "\n# cmdSlotWaits\n\
+      \\n\
+      \# HELP simplex_smp_cmd_slot_wait_ms_max The longest wait for a free command slot, in ms.\n\
+      \# TYPE simplex_smp_cmd_slot_wait_ms_max gauge\n\
+      \simplex_smp_cmd_slot_wait_ms_max " <> mshow _cmdSlotWaitMaxMs <> "\n# cmdSlotWaitMaxMs\n\
+      \\n\
+      \# HELP simplex_smp_cmd_slot_wait_timeouts Commands not processed because the client timed out while they waited for a free slot.\n\
+      \# TYPE simplex_smp_cmd_slot_wait_timeouts counter\n\
+      \simplex_smp_cmd_slot_wait_timeouts " <> mshow _cmdSlotWaitTimeouts <> "\n# cmdSlotWaitTimeouts\n\
       \\n"
     subscriptions =
       "# Subscriptions\n\

src/Simplex/Messaging/Server/Stats.hs

@@ -129,6 +129,11 @@ data ServerStats = ServerStats
     rcvServicesSubMsg :: IORef Int,
     rcvServicesSubDuplicate :: IORef Int,
     authDelayExceeded :: IORef Int,
+    -- commands that waited for a free slot in forkCmd, total and max wait in ms, and waits that timed out
+    cmdSlotWaits :: IORef Int,
+    cmdSlotWaitMs :: IORef Int,
+    cmdSlotWaitMaxMs :: IORef Int,
+    cmdSlotWaitTimeouts :: IORef Int,
     qCount :: IORef Int,
     msgCount :: IORef Int,
     ntfCount :: IORef Int,
@@ -192,6 +197,10 @@ data ServerStatsData = ServerStatsData
     _rcvServicesSubMsg :: Int,
     _rcvServicesSubDuplicate :: Int,
     _authDelayExceeded :: Int,
+    _cmdSlotWaits :: Int,
+    _cmdSlotWaitMs :: Int,
+    _cmdSlotWaitMaxMs :: Int,
+    _cmdSlotWaitTimeouts :: Int,
     _qCount :: Int,
     _msgCount :: Int,
     _ntfCount :: Int,
@@ -257,6 +266,10 @@ newServerStats ts = do
   rcvServicesSubMsg <- newIORef 0
   rcvServicesSubDuplicate <- newIORef 0
   authDelayExceeded <- newIORef 0
+  cmdSlotWaits <- newIORef 0
+  cmdSlotWaitMs <- newIORef 0
+  cmdSlotWaitMaxMs <- newIORef 0
+  cmdSlotWaitTimeouts <- newIORef 0
   qCount <- newIORef 0
   msgCount <- newIORef 0
   ntfCount <- newIORef 0
@@ -319,6 +332,10 @@ newServerStats ts = do
         rcvServicesSubMsg,
         rcvServicesSubDuplicate,
         authDelayExceeded,
+        cmdSlotWaits,
+        cmdSlotWaitMs,
+        cmdSlotWaitMaxMs,
+        cmdSlotWaitTimeouts,
         qCount,
         msgCount,
         ntfCount,
@@ -383,6 +400,10 @@ getServerStatsData s = do
   _rcvServicesSubMsg <- readIORef $ rcvServicesSubMsg s
   _rcvServicesSubDuplicate <- readIORef $ rcvServicesSubDuplicate s
   _authDelayExceeded <- readIORef $ authDelayExceeded s
+  _cmdSlotWaits <- readIORef $ cmdSlotWaits s
+  _cmdSlotWaitMs <- readIORef $ cmdSlotWaitMs s
+  _cmdSlotWaitMaxMs <- readIORef $ cmdSlotWaitMaxMs s
+  _cmdSlotWaitTimeouts <- readIORef $ cmdSlotWaitTimeouts s
   _qCount <- readIORef $ qCount s
   _msgCount <- readIORef $ msgCount s
   _ntfCount <- readIORef $ ntfCount s
@@ -445,6 +466,10 @@ getServerStatsData s = do
         _rcvServicesSubMsg,
         _rcvServicesSubDuplicate,
         _authDelayExceeded,
+        _cmdSlotWaits,
+        _cmdSlotWaitMs,
+        _cmdSlotWaitMaxMs,
+        _cmdSlotWaitTimeouts,
         _qCount,
         _msgCount,
         _ntfCount,
@@ -510,6 +535,10 @@ setServerStats s d = do
   writeIORef (rcvServicesSubMsg s) $! _rcvServicesSubMsg d
   writeIORef (rcvServicesSubDuplicate s) $! _rcvServicesSubDuplicate d
   writeIORef (authDelayExceeded s) $! _authDelayExceeded d
+  writeIORef (cmdSlotWaits s) $! _cmdSlotWaits d
+  writeIORef (cmdSlotWaitMs s) $! _cmdSlotWaitMs d
+  writeIORef (cmdSlotWaitMaxMs s) $! _cmdSlotWaitMaxMs d
+  writeIORef (cmdSlotWaitTimeouts s) $! _cmdSlotWaitTimeouts d
   writeIORef (qCount s) $! _qCount d
   writeIORef (msgCount s) $! _msgCount d
   writeIORef (ntfCount s) $! _ntfCount d
@@ -713,6 +742,10 @@ instance StrEncoding ServerStatsData where
           _rcvServicesSubMsg = 0,
           _rcvServicesSubDuplicate = 0,
           _authDelayExceeded = 0,
+          _cmdSlotWaits = 0,
+          _cmdSlotWaitMs = 0,
+          _cmdSlotWaitMaxMs = 0,
+          _cmdSlotWaitTimeouts = 0,
           _qCount,
           _msgCount = 0,
           _ntfCount = 0,