smp server: limit pending delayed AUTH responses per connection

src/Simplex/Messaging/Server.hs

@@ -1076,7 +1076,7 @@ runClientTransport h@THandle {params = thParams@THandleParams {sessionId}} = do
         expCfg <- asks $ inactiveClientExpiration . config
         th <- newMVar h -- put TH under a fair lock to interleave messages and command responses
         labelMyThread . B.unpack $ "client $" <> encode sessionId
-        raceAny_ $ [liftIO $ send th c, liftIO $ sendMsg th c, client s ms c, receive h ms c] <> disconnectThread_ c s expCfg
+        raceAny_ $ [liftIO $ send th c, liftIO $ sendMsg th c, liftIO $ sendDelayed c, client s ms c, receive h ms c] <> disconnectThread_ c s expCfg
     disconnectThread_ :: Client s -> Server s -> Maybe ExpirationConfig -> [M s ()]
     disconnectThread_ c s (Just expCfg) = [liftIO $ disconnectTransport h (rcvActiveAt c) (sndActiveAt c) expCfg (noSubscriptions c s)]
     disconnectThread_ _ _ _ = []
@@ -1202,15 +1202,25 @@ sendResponses c@Client {sndQ} stats authDelay since rs@(ts, _)
   | 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
+sendAfterDelay Client {sndQ, delayedQ} stats delay since rs
   | delay > 0 = do
-      elapsed <- subtract (microseconds since) . microseconds <$> getSystemTime
+      now <- microseconds <$> getSystemTime
+      let elapsed = now - microseconds since
       when (elapsed >= delay) $ incStat $ authDelayExceeded stats
-      forkClient c (B.unpack $ "client $" <> encode sessionId <> " auth") $ threadDelay' (delay - elapsed `mod` delay) >> write
+      atomically $ writeTBQueue delayedQ (now + delay - elapsed `mod` delay, rs)
   | otherwise = write
   where
     write = atomically $ writeTBQueue sndQ rs
 
+sendDelayed :: Client s -> IO ()
+sendDelayed Client {sndQ, delayedQ, clientTHParams = THandleParams {sessionId}} = do
+  labelMyThread . B.unpack $ "client $" <> encode sessionId <> " auth"
+  forever $ do
+    (sendAt, rs) <- atomically $ readTBQueue delayedQ
+    now <- microseconds <$> getSystemTime
+    threadDelay' (sendAt - now)
+    atomically $ writeTBQueue sndQ rs
+
 updateSlotWaitStats :: ServerStats -> SystemTime -> Bool -> IO ()
 updateSlotWaitStats ServerStats {cmdSlotWaits, cmdSlotWaitMs, cmdSlotWaitMaxMs, cmdSlotWaitTimeouts} since taken = do
   waitMs <- fromIntegral . (`div` 1000) . subtract (microseconds since) . microseconds <$> getSystemTime

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

@@ -432,6 +432,7 @@ data Client s = Client
     rcvQ :: TBQueue (NonEmpty (VerifiedTransmission s)),
     sndQ :: TBQueue (NonEmpty (Transmission BrokerMsg), [Transmission BrokerMsg]),
     msgQ :: TBQueue (NonEmpty (Transmission BrokerMsg)),
+    delayedQ :: TBQueue (Int64, (NonEmpty (Transmission BrokerMsg), [Transmission BrokerMsg])),
     procThreads :: TVar Int,
     endThreads :: TVar (IntMap (Weak ThreadId)),
     endThreadSeq :: TVar Int,
@@ -505,6 +506,7 @@ newClient clientId qSize clientTHParams createdAt = do
   rcvQ <- newTBQueueIO qSize
   sndQ <- newTBQueueIO qSize
   msgQ <- newTBQueueIO qSize
+  delayedQ <- newTBQueueIO qSize
   procThreads <- newTVarIO 0
   endThreads <- newTVarIO IM.empty
   endThreadSeq <- newTVarIO 0
@@ -523,6 +525,7 @@ newClient clientId qSize clientTHParams createdAt = do
         rcvQ,
         sndQ,
         msgQ,
+        delayedQ,
         procThreads,
         endThreads,
         endThreadSeq,

tests/ServerTests.hs

@@ -1340,7 +1340,7 @@ testTiming =
           ok `shouldBe` True
 
 testAuthErrorDelay :: SpecWith (ASrvTransport, AStoreType)
-testAuthErrorDelay =
+testAuthErrorDelay = do
   it "should send AUTH errors after the delay and other responses without it" $ \(ATransport t, msType) ->
     smpTest2Cfg (updateCfg (cfgMS msType) $ \cfg' -> cfg' {authErrorDelay = 500000}) supportedClientSMPRelayVRange t $ \rh sh -> do
       g <- C.newRandom
@@ -1359,13 +1359,41 @@ testAuthErrorDelay =
       suspendedTime `shouldSatisfy` (>= 0.5)
       noLinkTime `shouldSatisfy` (>= 0.5)
       subTime `shouldSatisfy` (< 0.5)
+  it "should not read client commands while its delayed AUTH errors fill the queue" $ \(ATransport t, msType) ->
+    smpTest2Cfg (updateCfg (cfgMS msType) $ \cfg' -> cfg' {authErrorDelay = 500000, tbqSize = 1}) supportedClientSMPRelayVRange t $ \h _ -> do
+      g <- C.newRandom
+      (_, badKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
+      (pongTime, authCorrIds) <- pongTimeAfter h $
+        forM_ (["1", "2", "3"] :: [ByteString]) $ \corrId -> signSend_ h badKey Nothing (corrId, EntityId "1234", SUB)
+      authCorrIds `shouldMatchList` ["1", "2", "3"]
+      pongTime `shouldSatisfy` (>= 0.5)
+  it "should not process client commands while its delayed AUTH errors fill the queue" $ \(ATransport t, msType) ->
+    smpTest2Cfg (updateCfg (cfgMS msType) $ \cfg' -> cfg' {authErrorDelay = 500000, tbqSize = 1}) supportedClientSMPRelayVRange t $ \rh sh -> do
+      g <- C.newRandom
+      (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
+      (sId, rId, rKey, _) <- createAndSecureQueue rh sPub
+      Resp "1" _ OK <- signSendRecv rh rKey ("1", rId, OFF)
+      (pongTime, authCorrIds) <- pongTimeAfter sh $
+        forM_ (["2", "3", "4"] :: [ByteString]) $ \corrId -> signSend_ sh sKey Nothing (corrId, sId, _SEND "hello")
+      authCorrIds `shouldMatchList` ["2", "3", "4"]
+      pongTime `shouldSatisfy` (>= 0.5)
 
 responseTime :: IO a -> IO (NominalDiffTime, a)
 responseTime action = do
   t <- getCurrentTime
   r <- action
   (,r) . (`diffUTCTime` t) <$> getCurrentTime
 
+pongTimeAfter :: Transport c => THandleSMP c 'TClient -> IO () -> IO (NominalDiffTime, [CorrId])
+pongTimeAfter h@THandle {params} sendCmds = do
+  start <- getCurrentTime
+  sendCmds
+  let TransmissionForAuth {tToSend} = encodeTransmissionForAuth params (CorrId "ping", NoEntity, PING)
+  Right () <- tPut1 h (Nothing, tToSend)
+  rs <- replicateM 4 $ (,) <$> tGet1 h <*> getCurrentTime
+  [pongAt] <- pure [at | (Resp "ping" _ PONG, at) <- rs]
+  pure (pongAt `diffUTCTime` start, [corrId | (Resp corrId _ (ERR AUTH), _) <- rs])
+
 testMessageNotifications :: SpecWith (ASrvTransport, AStoreType)
 testMessageNotifications =
   it "should create simplex connection, subscribe notifier and deliver notifications" $ \(ATransport t, msType) -> do