smp server: delayed AUTH responses (#1937)

* smp server: delayed AUTH responses

* plan

* update

* simplify

* simplify

* stat when sent without delay

* update schema

* round up response time to the nearest delay multiple

---------

Co-authored-by: Evgeny @ SimpleX Chat <259188159+evgeny-simplex@users.noreply.github.com>

src/Simplex/Messaging/Server.hs

@@ -1145,14 +1145,16 @@ cancelSub s = case subThread s of
 type VerifiedTransmissionOrError s = Either (Transmission BrokerMsg) (VerifiedTransmission s)
 
 receive :: forall c s. (Transport c, MsgStoreClass s) => THandleSMP c 'TServer -> s -> Client s -> M s ()
-receive h@THandle {params = THandleParams {thAuth, sessionId}} ms Client {rcvQ, sndQ, rcvActiveAt} = do
+receive h@THandle {params = THandleParams {thAuth, sessionId}} ms c@Client {rcvQ, rcvActiveAt} = do
   labelMyThread . B.unpack $ "client $" <> encode sessionId <> " receive"
   sa <- asks serverActive
   stats <- asks serverStats
+  authDelay <- asks $ authErrorDelay . config
   liftIO $ forever $ do
     ts <- tGetServer h
     unlessM (readTVarIO sa) $ throwIO $ userError "server stopped"
-    atomically . (writeTVar rcvActiveAt $!) =<< getSystemTime
+    receivedAt <- getSystemTime
+    atomically $ writeTVar rcvActiveAt $! receivedAt
     let (es, ts') = partitionEithers $ L.toList ts
         errs = map (second ERR) es
     errs' <- case ts' of
@@ -1168,7 +1170,7 @@ receive h@THandle {params = THandleParams {thAuth, sessionId}} ms Client {rcvQ,
         mapM_ (atomically . writeTBQueue rcvQ) $ L.nonEmpty cmds
         pure $ errs ++ errs'
       [] -> pure errs
-    mapM_ (atomically . writeTBQueue sndQ . (,[])) $ L.nonEmpty errs'
+    mapM_ (sendResponses c stats authDelay receivedAt . (,[])) $ L.nonEmpty errs'
   where
     sameParty :: SParty p -> SignedTransmission Cmd -> Bool
     sameParty p (_, _, (_, _, Cmd p' _)) = isJust $ testEquality p p'
@@ -1191,6 +1193,27 @@ receive h@THandle {params = THandleParams {thAuth, sessionId}} ms Client {rcvQ,
             GET -> incStat $ msgGetAuth stats
             _ -> pure ()
 
+isAuthError :: Transmission BrokerMsg -> Bool
+isAuthError (_, _, r) = case r of
+  ERR AUTH -> True
+  _ -> False
+
+sendResponses :: Client s -> ServerStats -> Int64 -> SystemTime -> (NonEmpty (Transmission BrokerMsg), [Transmission BrokerMsg]) -> IO ()
+sendResponses c@Client {sndQ} stats authDelay since rs@(ts, _)
+  | any isAuthError ts = sendAfterDelay c stats authDelay since rs
+  | otherwise = atomically $ writeTBQueue sndQ rs
+
+sendAfterDelay :: Client s -> ServerStats -> Int64 -> SystemTime -> (NonEmpty (Transmission BrokerMsg), [Transmission BrokerMsg]) -> IO ()
+sendAfterDelay c@Client {sndQ, clientTHParams = THandleParams {sessionId}} stats delay since rs
+  | delay > 0 = do
+      elapsed <- subtract (microseconds since) . microseconds <$> getSystemTime
+      when (elapsed >= delay) $ incStat $ authDelayExceeded stats
+      forkClient c (B.unpack $ "client $" <> encode sessionId <> " auth") $ threadDelay' (delay - elapsed `mod` delay) >> write
+  | otherwise = write
+  where
+    write = atomically $ writeTBQueue sndQ rs
+    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
   labelMyThread . B.unpack $ "client $" <> encode sessionId <> " send"
@@ -1374,15 +1397,18 @@ client
   ms
   clnt@Client {clientId, rcvQ, sndQ, msgQ, clientTHParams = thParams'@THandleParams {sessionId}, procThreads} = do
     labelMyThread . B.unpack $ "client $" <> encode sessionId <> " commands"
+    stats <- asks serverStats
+    authDelay <- asks $ authErrorDelay . config
     let clntServiceId = (\THClientService {serviceId} -> serviceId) <$> (peerClientService =<< thAuth thParams')
         process batchSubs t acc@(rs, msgs) =
           (maybe acc (\(!r, !msg_) -> (r : rs, maybe msgs (: msgs) msg_)))
             <$> processCommand clntServiceId batchSubs t
     forever $ do
       batch <- atomically (readTBQueue rcvQ)
+      processedAt <- liftIO getSystemTime
       batchSubs <- prepareBatchSubs clntServiceId batch
       foldrM (process batchSubs) ([], []) batch
-        >>= \(rs_, msgs) -> mapM_ (atomically . writeTBQueue sndQ . (,msgs)) (L.nonEmpty rs_)
+        >>= \(rs_, msgs) -> mapM_ (liftIO . sendResponses clnt stats authDelay processedAt . (,msgs)) (L.nonEmpty rs_)
   where
     prepareBatchSubs ::
       Maybe ServiceId ->
@@ -2119,6 +2145,7 @@ client
 
         processForwardedCommand :: EncFwdTransmission -> M s (Maybe BrokerMsg)
         processForwardedCommand (EncFwdTransmission s) = fmap (either (Just . ERR) id) . runExceptT $ do
+          startedAt <- liftIO getSystemTime
           THAuthServer {serverPrivKey, sessSecret'} <- maybe (throwE $ transportErr TENoServerAuth) pure (thAuth thParams')
           sessSecret <- maybe (throwE $ transportErr TENoServerAuth) pure sessSecret'
           proxyNonce <- liftEitherWith (const CRYPTO) $ C.cbNonce $ bs corrId
@@ -2162,7 +2189,11 @@ client
               _ -> Just . maybe (corrId', entId', ERR INTERNAL) fst <$> lift (processCommand Nothing (Right (M.empty, M.empty, M.empty)) t'')
           stats <- asks serverStats
           incStat $ pMsgFwdsRecv stats
-          traverse encodeResp r_
+          traverse encodeResp r_ >>= \case
+            Just rres | any isAuthError r_ -> do
+              authDelay <- asks $ authErrorDelay . config
+              Nothing <$ liftIO (sendAfterDelay clnt stats authDelay startedAt ([(corrId, NoEntity, rres)], []))
+            rres_ -> pure rres_
           where
             rejectOrVerify :: Maybe (THandleAuth 'TServer) -> SignedTransmissionOrError ErrorType Cmd -> M s (VerifiedTransmissionOrError s)
             rejectOrVerify clntThAuth = \case

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

@@ -68,6 +68,7 @@ module Simplex.Messaging.Server.Env.STM
     defaultInactiveClientExpiration,
     defaultProxyClientConcurrency,
     defaultNameResolverConcurrency,
+    defaultAuthErrorDelayMs,
     defaultMaxJournalMsgCount,
     defaultMaxJournalStateLines,
     defaultIdleQueueInterval,
@@ -187,6 +188,7 @@ data ServerConfig s = ServerConfig
     ntfDeliveryInterval :: Int,
     -- | interval between sending pending END events to unsubscribed clients, seconds
     pendingENDInterval :: Int,
+    authErrorDelay :: Int64,
     smpCredentials :: ServerCredentials,
     httpCredentials :: Maybe ServerCredentials,
     -- | SMP client-server protocol version range
@@ -255,6 +257,9 @@ defaultProxyClientConcurrency = 32
 defaultNameResolverConcurrency :: Int
 defaultNameResolverConcurrency = 1000
 
+defaultAuthErrorDelayMs :: Int64
+defaultAuthErrorDelayMs = 50
+
 journalMsgStoreDepth :: Int
 journalMsgStoreDepth = 5
 

src/Simplex/Messaging/Server/Main.hs

@@ -580,6 +580,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
               prometheusMetricsFile = combine logPath "smp-server-metrics.txt",
               pendingENDInterval = 15000000, -- 15 seconds
               ntfDeliveryInterval = 1500000, -- 1.5 second
+              authErrorDelay = 1000 * readIniDefault defaultAuthErrorDelayMs "TRANSPORT" "auth_error_delay_ms" ini,
               smpServerVRange = supportedServerSMPRelayVRange,
               transportConfig =
                 mkTransportServerConfig

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

@@ -130,7 +130,9 @@ iniFileContent cfgPath logPath opts host basicAuth controlPortPwds =
     <> ("host = " <> T.pack host <> "\n")
     <> ("port = " <> defaultServerPorts <> "\n")
     <> "log_tls_errors = off\n\n\
-        \# Use `websockets = 443` to run websockets server in addition to plain TLS.\n\
+        \# Minimum time to respond with AUTH error, milliseconds.\n"
+    <> ("# auth_error_delay_ms = " <> tshow defaultAuthErrorDelayMs <> "\n\n")
+    <> "# Use `websockets = 443` to run websockets server in addition to plain TLS.\n\
         \# This option is deprecated and should be used for testing only.\n\
         \# , port 443 should be specified in port above\n\
         \websockets = off\n"

src/Simplex/Messaging/Server/Prometheus.hs

@@ -126,6 +126,7 @@ prometheusMetrics sm rtm ts =
         _ntfServices,
         _rcvServicesSubMsg,
         _rcvServicesSubDuplicate,
+        _authDelayExceeded,
         _qCount,
         _msgCount,
         _ntfCount,
@@ -192,6 +193,10 @@ prometheusMetrics sm rtm ts =
       \# HELP simplex_smp_queues_notify_monthly Monthly active queues with notifications.\n\
       \# TYPE simplex_smp_queues_notify_monthly gauge\n\
       \simplex_smp_queues_notify_monthly " <> mstr (monthCount psNtf) <> "\n# monthCountNtf\n\
+      \\n\
+      \# 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"
     subscriptions =
       "# Subscriptions\n\

src/Simplex/Messaging/Server/QueueStore/Postgres/server_schema.sql

@@ -1,6 +1,5 @@
 
 
-
 SET statement_timeout = 0;
 SET lock_timeout = 0;
 SET idle_in_transaction_session_timeout = 0;
@@ -555,4 +554,3 @@ ALTER TABLE ONLY smp_server.msg_queues
 
 
 
-

src/Simplex/Messaging/Server/Stats.hs

@@ -128,6 +128,7 @@ data ServerStats = ServerStats
     ntfServices :: ServiceStats,
     rcvServicesSubMsg :: IORef Int,
     rcvServicesSubDuplicate :: IORef Int,
+    authDelayExceeded :: IORef Int,
     qCount :: IORef Int,
     msgCount :: IORef Int,
     ntfCount :: IORef Int,
@@ -190,6 +191,7 @@ data ServerStatsData = ServerStatsData
     _rcvServices :: ServiceStatsData,
     _rcvServicesSubMsg :: Int,
     _rcvServicesSubDuplicate :: Int,
+    _authDelayExceeded :: Int,
     _qCount :: Int,
     _msgCount :: Int,
     _ntfCount :: Int,
@@ -254,6 +256,7 @@ newServerStats ts = do
   ntfServices <- newServiceStats
   rcvServicesSubMsg <- newIORef 0
   rcvServicesSubDuplicate <- newIORef 0
+  authDelayExceeded <- newIORef 0
   qCount <- newIORef 0
   msgCount <- newIORef 0
   ntfCount <- newIORef 0
@@ -315,6 +318,7 @@ newServerStats ts = do
         ntfServices,
         rcvServicesSubMsg,
         rcvServicesSubDuplicate,
+        authDelayExceeded,
         qCount,
         msgCount,
         ntfCount,
@@ -378,6 +382,7 @@ getServerStatsData s = do
   _ntfServices <- getServiceStatsData $ ntfServices s
   _rcvServicesSubMsg <- readIORef $ rcvServicesSubMsg s
   _rcvServicesSubDuplicate <- readIORef $ rcvServicesSubDuplicate s
+  _authDelayExceeded <- readIORef $ authDelayExceeded s
   _qCount <- readIORef $ qCount s
   _msgCount <- readIORef $ msgCount s
   _ntfCount <- readIORef $ ntfCount s
@@ -439,6 +444,7 @@ getServerStatsData s = do
         _ntfServices,
         _rcvServicesSubMsg,
         _rcvServicesSubDuplicate,
+        _authDelayExceeded,
         _qCount,
         _msgCount,
         _ntfCount,
@@ -503,6 +509,7 @@ setServerStats s d = do
   setServiceStats (ntfServices s) $! _ntfServices d
   writeIORef (rcvServicesSubMsg s) $! _rcvServicesSubMsg d
   writeIORef (rcvServicesSubDuplicate s) $! _rcvServicesSubDuplicate d
+  writeIORef (authDelayExceeded s) $! _authDelayExceeded d
   writeIORef (qCount s) $! _qCount d
   writeIORef (msgCount s) $! _msgCount d
   writeIORef (ntfCount s) $! _ntfCount d
@@ -705,6 +712,7 @@ instance StrEncoding ServerStatsData where
           _ntfServices,
           _rcvServicesSubMsg = 0,
           _rcvServicesSubDuplicate = 0,
+          _authDelayExceeded = 0,
           _qCount,
           _msgCount = 0,
           _ntfCount = 0,