From a4528561121e0fcbba95679eafa6ae34d3828492 Mon Sep 17 00:00:00 2001 From: "Evgeny @ SimpleX Chat" <259188159+evgeny-simplex@users.noreply.github.com> Date: Sun, 25 Oct 2026 20:14:15 +0000 Subject: [PATCH 1/2] smp server: limit pending delayed AUTH responses per connection --- src/Simplex/Messaging/Server.hs | 18 +++++++++++---- src/Simplex/Messaging/Server/Env/STM.hs | 3 +++ tests/ServerTests.hs | 30 ++++++++++++++++++++++++- 3 files changed, 46 insertions(+), 5 deletions(-) diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index 30133885d..cd6bdee20 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/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 + when (sendAt > now) $ 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 diff --git a/src/Simplex/Messaging/Server/Env/STM.hs b/src/Simplex/Messaging/Server/Env/STM.hs index 4c0b1d566..e343e3e93 100644 --- a/src/Simplex/Messaging/Server/Env/STM.hs +++ b/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, diff --git a/tests/ServerTests.hs b/tests/ServerTests.hs index 146a1d46f..31cbf0d90 100644 --- a/tests/ServerTests.hs +++ b/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,6 +1359,24 @@ 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 @@ -1366,6 +1384,16 @@ responseTime action = do 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 From 58f33b181bc4bf458685fea39f8d616152fe358d Mon Sep 17 00:00:00 2001 From: "Evgeny @ SimpleX Chat" <259188159+evgeny-simplex@users.noreply.github.com> Date: Mon, 26 Oct 2026 23:27:11 +0000 Subject: [PATCH 2/2] smp server: remove redundant time check before delayed AUTH responses --- src/Simplex/Messaging/Server.hs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index cd6bdee20..eb2f26e7f 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/src/Simplex/Messaging/Server.hs @@ -1218,7 +1218,7 @@ sendDelayed Client {sndQ, delayedQ, clientTHParams = THandleParams {sessionId}} forever $ do (sendAt, rs) <- atomically $ readTBQueue delayedQ now <- microseconds <$> getSystemTime - when (sendAt > now) $ threadDelay' (sendAt - now) + threadDelay' (sendAt - now) atomically $ writeTBQueue sndQ rs updateSlotWaitStats :: ServerStats -> SystemTime -> Bool -> IO ()