smp-server: client command slot wait timeout

src/Simplex/Messaging/Server.hs

@@ -11,7 +11,6 @@
 {-# LANGUAGE NumericUnderscores #-}
 {-# LANGUAGE OverloadedLists #-}
 {-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE PatternSynonyms #-}
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TupleSections #-}
@@ -99,7 +98,7 @@ 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 (NetworkConfig (..), NetworkRequestMode (..), ProtocolClient (thParams), ProtocolClientConfig (..), ProtocolClientError (..), SMPClient, SMPClientError, clientHandlers, forwardSMPTransmission, netTimeoutInt, smpProxyError, temporaryClientError, pattern NRMInteractive)
+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
@@ -1222,6 +1221,9 @@ updateSlotWaitStats ServerStats {cmdSlotWaits, cmdSlotWaitMs, cmdSlotWaitMaxMs,
   atomicModifyIORef'_ cmdSlotWaitMaxMs (max waitMs)
   unless taken $ incStat cmdSlotWaitTimeouts
 
+forwardedCmdWait :: ServerConfig s -> Int
+forwardedCmdWait = (`netTimeoutInt` NRMBackground) . tcpTimeout . networkConfig . smpCfg . smpAgentCfg
+
 microseconds :: SystemTime -> Int64
 microseconds (MkSystemTime s ns) = s * 1000000 + fromIntegral (ns `div` 1000)
 
@@ -1507,12 +1509,12 @@ client
     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 NRMInteractive (sendResp timeoutErr) $ cmdAction >>= sendResp
+      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) -> NetworkRequestMode -> M s () -> M s () -> M s (Maybe a)
-    forkCmd_ concurrency nm onTimeout cmd = do
+    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.
@@ -1529,7 +1531,7 @@ client
         wait = do
           limit <- asks (concurrency . config)
           ifM (atomically $ takeSlot limit) (pure True) $ do
-            clientTimeout <- asks $ (`netTimeoutInt` nm) . tcpTimeout . networkConfig . smpCfg . smpAgentCfg . config
+            clientTimeout <- asks $ waitTimeout . config
             waitStarted <- liftIO getSystemTime
             expired <- liftIO $ registerDelay clientTimeout
             taken <- atomically $ (True <$ (takeSlot limit >>= check)) `orElse` (False <$ (readTVar expired >>= check))
@@ -2218,7 +2220,7 @@ client
                 Nothing -> pure $ Just (corrId', entId', ERR (NAME NO_RESOLVER))
                 -- The proxy waits for forwarded commands with the background timeout.
                 -- If no slot is free before it, there is no response, as the proxy has already timed out.
-                Just nenv -> forkCmd_ serverResolverConcurrency NRMBackground (pure ()) $ do
+                Just nenv -> forkCmd_ serverResolverConcurrency forwardedCmdWait (pure ()) $ do
                   msg <- resolveNameMsg (thVersion clntTHParams) nenv d
                   rres <- either ERR id <$> runExceptT (encodeResp (corrId', entId', msg))
                   atomically $ writeTBQueue sndQ ([(corrId, NoEntity, rres)], [])

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

@@ -68,6 +68,7 @@ module Simplex.Messaging.Server.Env.STM
     defaultInactiveClientExpiration,
     defaultProxyClientConcurrency,
     defaultNameResolverConcurrency,
+    defaultMaxClientCmdWait,
     defaultAuthErrorDelayMs,
     defaultMaxJournalMsgCount,
     defaultMaxJournalStateLines,
@@ -206,6 +207,7 @@ data ServerConfig s = ServerConfig
     -- 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
@@ -257,6 +259,9 @@ defaultProxyClientConcurrency = 32
 defaultNameResolverConcurrency :: Int
 defaultNameResolverConcurrency = 1000
 
+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

tests/RSLVTests.hs

@@ -27,7 +27,6 @@ import NamesResolverServer (memCfg, memCfg2, memProxyCfg, withNames)
 import qualified NamesResolverServer as NRS
 import SMPClient
 import Simplex.Messaging.Client
-import Simplex.Messaging.Client.Agent (SMPClientAgentConfig (..))
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String (strDecode)
 import SMPNamesTests (availableBody, registeredBody, reservedBody, resolved, testNameRecord, testPricing)
@@ -331,11 +330,7 @@ testRslvSlotWaitTimeout =
 
 slotWaitCfg :: Int -> Int -> AServerConfig
 slotWaitCfg port clientTimeout =
-  updateCfg (withNames port memCfg) $ \c ->
-    let aCfg = smpAgentCfg c
-        cCfg = smpCfg aCfg
-        nt = NetworkTimeout {backgroundTimeout = clientTimeout, interactiveTimeout = clientTimeout}
-     in c {serverClientConcurrency = 4, prometheusInterval = Just 1, smpAgentCfg = aCfg {smpCfg = cCfg {networkConfig = (networkConfig cCfg) {tcpTimeout = nt}}}}
+  updateCfg (withNames port memCfg) $ \c -> c {serverClientConcurrency = 4, prometheusInterval = Just 1, maxClientCmdWait = clientTimeout}
 
 data SlotWaitMetrics = SlotWaitMetrics {waits :: Int, waitTimeouts :: Int, waitMaxMs :: Int, waitSumMs :: Int}
 

tests/SMPClient.hs

@@ -278,6 +278,7 @@ cfgMS msType = withStoreCfg (testServerStoreConfig msType) $ \serverStoreCfg ->
       allowSMPProxy = False,
       serverClientConcurrency = 2,
       serverResolverConcurrency = defaultNameResolverConcurrency,
+      maxClientCmdWait = defaultMaxClientCmdWait,
       namesConfig = Nothing,
       information = Nothing,
       startOptions = defaultStartOptions