From 5d38ad03af14348febf48a4771d2c79f807fbc0f Mon Sep 17 00:00:00 2001 From: Alexander Bondarenko <486682+dpwiz@users.noreply.github.com> Date: Thu, 23 May 2024 17:34:25 +0300 Subject: [PATCH] tests: add proxy stress tests (#1163) * tests: add proxy stress tests * organize benches * add agent tests * move prints to logNote * fix stuck agent tests --- tests/SMPProxyTests.hs | 172 ++++++++++++++++++++++++++++++++++------- tests/Util.hs | 22 ++++++ 2 files changed, 165 insertions(+), 29 deletions(-) diff --git a/tests/SMPProxyTests.hs b/tests/SMPProxyTests.hs index 1c458a062..748eb34e7 100644 --- a/tests/SMPProxyTests.hs +++ b/tests/SMPProxyTests.hs @@ -8,13 +8,14 @@ {-# LANGUAGE PatternSynonyms #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE TupleSections #-} {-# LANGUAGE TypeApplications #-} module SMPProxyTests where import AgentTests.FunctionalAPITests -import Control.Monad.Trans.Except (runExceptT) +import Control.Logger.Simple +import Control.Monad (forM, forM_, forever) +import Control.Monad.Trans.Except (ExceptT, runExceptT) import Data.ByteString.Char8 (ByteString) import Data.List.NonEmpty (NonEmpty) import qualified Data.List.NonEmpty as L @@ -33,9 +34,12 @@ import qualified Simplex.Messaging.Crypto.Ratchet as CR import Simplex.Messaging.Protocol as SMP import Simplex.Messaging.Server.Env.STM (ServerConfig (..)) import Simplex.Messaging.Transport +import Simplex.Messaging.Util (bshow, tshow) import Simplex.Messaging.Version (mkVersionRange) +import System.FilePath (splitExtensions) import Test.Hspec import UnliftIO +import Util smpProxyTests :: Spec smpProxyTests = do @@ -71,6 +75,22 @@ smpProxyTests = do deliverMessageViaProxy proxyServ relayServ C.SEd25519 msg1 msg2 it "max message size, X25519 keys" . twoServersFirstProxy $ deliverMessageViaProxy proxyServ relayServ C.SX25519 msg1 msg2 + describe "stress test 1k" $ do + let deliver n = deliverMessagesViaProxy srv1 srv2 C.SEd448 [] (map bshow [1 :: Int .. n]) + it "1x1000" . twoServersFirstProxy $ deliver 1000 + it "5x200" . twoServersFirstProxy $ 5 `inParrallel` deliver 200 + it "10x100" . twoServersFirstProxy $ 10 `inParrallel` deliver 100 + xdescribe "stress test 10k" $ do + let deliver n = deliverMessagesViaProxy srv1 srv2 C.SEd448 [] (map bshow [1 :: Int .. n]) + it "1x10000" . twoServersFirstProxy $ deliver 10000 + it "5x2000" . twoServersFirstProxy $ 5 `inParrallel` deliver 2000 + it "10x1000" . twoServersFirstProxy $ 10 `inParrallel` deliver 1000 + it "100x100 N1" . twoServersFirstProxy $ withNumCapabilities 1 $ 100 `inParrallel` deliver 100 + it "100x100 N4 C1" . twoServersNoConc $ withNumCapabilities 4 $ 100 `inParrallel` deliver 100 + it "100x100 N4 C2" . twoServersFirstProxy $ withNumCapabilities 4 $ 100 `inParrallel` deliver 100 + it "100x100 N4 C16" . twoServersMoreConc $ withNumCapabilities 4 $ 100 `inParrallel` deliver 100 + it "100x100 N" . twoServersFirstProxy $ withNCPUCapabilities $ 100 `inParrallel` deliver 100 + it "500x20" . twoServersFirstProxy $ 500 `inParrallel` deliver 20 describe "agent API" $ do describe "one server" $ do it "always via proxy" . oneServer $ @@ -92,46 +112,65 @@ smpProxyTests = do agentDeliverMessageViaProxy ([srv1], SPMUnknown, False) ([srv2], SPMUnknown, False) C.SEd448 "hello 1" "hello 2" it "fails when fallback is prohibited" . twoServers_ proxyCfg cfgV7 $ agentViaProxyVersionError + describe "stress test 1k" $ do + let deliver nAgents nMsgs = agentDeliverMessagesViaProxyConc (replicate nAgents [srv1]) (map bshow [1 :: Int .. nMsgs]) + it "2 agents, 250 messages" . oneServer $ deliver 2 250 + it "5 agents, 10 pairs, 50 messages, N1" . oneServer . withNumCapabilities 1 $ deliver 5 50 + it "5 agents, 10 pairs, 50 messages. N4" . oneServer . withNumCapabilities 4 $ deliver 5 50 + xdescribe "stress test 10k" $ do + let deliver nAgents nMsgs = agentDeliverMessagesViaProxyConc (replicate nAgents [srv1]) (map bshow [1 :: Int .. nMsgs]) + it "25 agents, 300 pairs, 17 messages" . oneServer . withNumCapabilities 4 $ deliver 25 17 where - oneServer = withSmpServerConfigOn (transport @TLS) proxyCfg testPort . const + oneServer = withSmpServerConfigOn (transport @TLS) proxyCfg {msgQueueQuota = 128} testPort . const twoServers = twoServers_ proxyCfg proxyCfg - twoServersFirstProxy = twoServers_ proxyCfg cfgV8 + twoServersFirstProxy = twoServers_ proxyCfg cfgV8 {msgQueueQuota = 128} + twoServersMoreConc = twoServers_ proxyCfg {serverClientConcurrency = 128} cfgV8 {msgQueueQuota = 128} + twoServersNoConc = twoServers_ proxyCfg {serverClientConcurrency = 1} cfgV8 {msgQueueQuota = 128} twoServers_ cfg1 cfg2 runTest = withSmpServerConfigOn (transport @TLS) cfg1 testPort $ \_ -> withSmpServerConfigOn (transport @TLS) cfg2 testPort2 $ const runTest deliverMessageViaProxy :: (C.AlgorithmI a, C.AuthAlgorithm a) => SMPServer -> SMPServer -> C.SAlgorithm a -> ByteString -> ByteString -> IO () -deliverMessageViaProxy proxyServ relayServ alg msg msg' = do +deliverMessageViaProxy proxyServ relayServ alg msg msg' = deliverMessagesViaProxy proxyServ relayServ alg [msg] [msg'] + +deliverMessagesViaProxy :: (C.AlgorithmI a, C.AuthAlgorithm a) => SMPServer -> SMPServer -> C.SAlgorithm a -> [ByteString] -> [ByteString] -> IO () +deliverMessagesViaProxy proxyServ relayServ alg unsecuredMsgs securedMsgs = do g <- C.newRandom -- set up proxy - Right pc <- getProtocolClient g (1, proxyServ, Nothing) defaultSMPClientConfig {serverVRange = mkVersionRange batchCmdsSMPVersion sendingProxySMPVersion} Nothing (\_ -> pure ()) + pc' <- getProtocolClient g (1, proxyServ, Nothing) defaultSMPClientConfig {serverVRange = mkVersionRange batchCmdsSMPVersion sendingProxySMPVersion} Nothing (\_ -> pure ()) + pc <- either (fail . show) pure pc' THAuthClient {} <- maybe (fail "getProtocolClient returned no thAuth") pure $ thAuth $ thParams pc -- set up relay - msgQ <- newTBQueueIO 4 - Right rc <- getProtocolClient g (2, relayServ, Nothing) defaultSMPClientConfig {serverVRange = mkVersionRange batchCmdsSMPVersion authCmdsSMPVersion} (Just msgQ) (\_ -> pure ()) - runRight_ $ do - -- prepare receiving queue - (rPub, rPriv) <- atomically $ C.generateAuthKeyPair alg g - (rdhPub, rdhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g - QIK {rcvId, sndId, rcvPublicDhKey = srvDh} <- createSMPQueue rc (rPub, rPriv) rdhPub (Just "correct") SMSubscribe - let dec = decryptMsgV3 $ C.dh' srvDh rdhPriv - -- get proxy session - sess <- connectSMPProxiedRelay pc relayServ (Just "correct") - -- send via proxy to unsecured queue - Right () <- proxySMPMessage pc sess Nothing sndId noMsgFlags msg + msgQ <- newTBQueueIO 1024 + rc' <- getProtocolClient g (2, relayServ, Nothing) defaultSMPClientConfig {serverVRange = mkVersionRange batchCmdsSMPVersion authCmdsSMPVersion} (Just msgQ) (\_ -> pure ()) + rc <- either (fail . show) pure rc' + -- prepare receiving queue + (rPub, rPriv) <- atomically $ C.generateAuthKeyPair alg g + (rdhPub, rdhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g + QIK {rcvId, sndId, rcvPublicDhKey = srvDh} <- runExceptT' $ createSMPQueue rc (rPub, rPriv) rdhPub (Just "correct") SMSubscribe + let dec = decryptMsgV3 $ C.dh' srvDh rdhPriv + -- get proxy session + sess <- runExceptT' $ connectSMPProxiedRelay pc relayServ (Just "correct") + -- send via proxy to unsecured queue + forM_ unsecuredMsgs $ \msg -> do + runExceptT' (proxySMPMessage pc sess Nothing sndId noMsgFlags msg) `shouldReturn` Right () -- receive 1 (_tSess, _v, _sid, [(_entId, STEvent (Right (SMP.MSG RcvMessage {msgId, msgBody = EncRcvMsgBody encBody})))]) <- atomically $ readTBQueue msgQ - liftIO $ dec msgId encBody `shouldBe` Right msg - ackSMPMessage rc rPriv rcvId msgId - -- secure queue - (sPub, sPriv) <- atomically $ C.generateAuthKeyPair alg g - secureSMPQueue rc rPriv rcvId sPub - -- send via proxy to secured queue - Right () <- proxySMPMessage pc sess (Just sPriv) sndId noMsgFlags msg' - -- receive 2 - (_tSess, _v, _sid, [(_entId, STEvent (Right (SMP.MSG RcvMessage {msgId = msgId', msgBody = EncRcvMsgBody encBody'})))]) <- atomically $ readTBQueue msgQ - liftIO $ dec msgId' encBody' `shouldBe` Right msg' - ackSMPMessage rc rPriv rcvId msgId' + dec msgId encBody `shouldBe` Right msg + runExceptT' $ ackSMPMessage rc rPriv rcvId msgId + -- secure queue + (sPub, sPriv) <- atomically $ C.generateAuthKeyPair alg g + runExceptT' $ secureSMPQueue rc rPriv rcvId sPub + -- send via proxy to secured queue + waitSendRecv + ( forM_ securedMsgs $ \msg' -> + runExceptT' (proxySMPMessage pc sess (Just sPriv) sndId noMsgFlags msg') `shouldReturn` Right () + ) + ( forM_ securedMsgs $ \msg' -> do + (_tSess, _v, _sid, [(_entId, STEvent (Right (SMP.MSG RcvMessage {msgId = msgId', msgBody = EncRcvMsgBody encBody'})))]) <- atomically $ readTBQueue msgQ + dec msgId' encBody' `shouldBe` Right msg' + runExceptT' $ ackSMPMessage rc rPriv rcvId msgId' + ) agentDeliverMessageViaProxy :: (C.AlgorithmI a, C.AuthAlgorithm a) => (NonEmpty SMPServer, SMPProxyMode, Bool) -> (NonEmpty SMPServer, SMPProxyMode, Bool) -> C.SAlgorithm a -> ByteString -> ByteString -> IO () agentDeliverMessageViaProxy aTestCfg@(aSrvs, _, aViaProxy) bTestCfg@(bSrvs, _, bViaProxy) alg msg1 msg2 = @@ -171,6 +210,71 @@ agentDeliverMessageViaProxy aTestCfg@(aSrvs, _, aViaProxy) bTestCfg@(bSrvs, _, b aCfg = agentProxyCfg {sndAuthAlg = C.AuthAlg alg, rcvAuthAlg = C.AuthAlg alg} servers (srvs, smpProxyMode, _) = (initAgentServersProxy smpProxyMode SPFAllow) {smp = userServers $ L.map noAuthSrv srvs} +agentDeliverMessagesViaProxyConc :: [NonEmpty SMPServer] -> [MsgBody] -> IO () +agentDeliverMessagesViaProxyConc agentServers msgs = + withAgents $ \agents -> do + let pairs = combinations 2 agents + logNote $ "Pairing " <> tshow (length agents) <> " agents into " <> tshow (length pairs) <> " connections" + connections <- forM pairs $ \case + [a, b] -> prePair a b + _ -> error "agents must be paired" + logNote "Running..." + mapConcurrently_ run connections + where + withAgents :: ([AgentClient] -> IO ()) -> IO () + withAgents action = go [] (zip [1 :: Int ..] agentServers) + where + go agents = \case + [] -> action agents + (aId, aSrvs) : next -> withAgent aId aCfg (servers aSrvs) (dbPrefix <> show aId <> dbSuffix) $ \a -> (a : agents) `go` next + (dbPrefix, dbSuffix) = splitExtensions testDB + -- agent connections have to be set up in advance + -- otherwise the CONF messages would get mixed with MSG + prePair alice bob = do + (bobId, qInfo) <- runExceptT' $ A.createConnection alice 1 True SCMInvitation Nothing (CR.IKNoPQ PQSupportOn) SMSubscribe + aliceId <- runExceptT' $ A.joinConnection bob 1 Nothing True qInfo "bob's connInfo" PQSupportOn SMSubscribe + confId <- + get alice >>= \case + ("", _, A.CONF confId pqSup' _ "bob's connInfo") -> do + pqSup' `shouldBe` PQSupportOn + pure confId + huh -> fail $ show huh + runExceptT' $ allowConnection alice bobId confId "alice's connInfo" + get alice ##> ("", bobId, A.CON pqEnc) + get bob ##> ("", aliceId, A.INFO PQSupportOn "alice's connInfo") + get bob ##> ("", aliceId, A.CON pqEnc) + pure (alice, bobId, bob, aliceId) + -- stream messages in opposite directions, while getting deliveries and sending ACKs + run (alice, bobId, bob, aliceId) = do + aSender <- async $ forM_ msgs $ runExceptT' . A.sendMessage alice bobId pqEnc noMsgFlags + bRecipient <- + async $ + forever $ + get bob >>= \case + ("", _, A.SENT _ _) -> pure () + ("", _, Msg' mId' _ _) -> runExceptT' $ ackMessage alice bobId mId' Nothing + huh -> fail (show huh) + bSender <- async $ forM_ msgs $ runExceptT' . A.sendMessage bob aliceId pqEnc noMsgFlags + aRecipient <- + async $ + forever $ + get alice >>= \case + ("", _, A.SENT _ _) -> pure () + ("", _, Msg' mId' _ _) -> runExceptT' $ ackMessage alice bobId mId' Nothing + huh -> fail (show huh) + logDebug "run waiting..." + a2b <- async $ (waitCatch aSender >>= either throwIO pure) `finally` cancel bRecipient -- stopped sender cancels paired recipient loop + b2a <- async $ (waitCatch bSender >>= either throwIO pure) `finally` cancel aRecipient + waitEitherCatch a2b b2a >>= \case + Right (Right ()) -> wait b2a + Right (Left e) -> cancel bSender >> throwIO e + Left (Right ()) -> wait a2b + Left (Left e) -> cancel aSender >> throwIO e + logDebug "run finished" + pqEnc = CR.PQEncOn + aCfg = agentProxyCfg {sndAuthAlg = C.AuthAlg C.SEd448, rcvAuthAlg = C.AuthAlg C.SEd448} + servers srvs = (initAgentServersProxy SPMAlways SPFAllow) {smp = userServers $ L.map noAuthSrv srvs} + agentViaProxyVersionError :: IO () agentViaProxyVersionError = withAgent 1 agentProxyCfg (servers [SMPServer testHost testPort testKeyHash]) testDB $ \alice -> do @@ -201,3 +305,13 @@ testProxyAuth = do todo :: IO () todo = do fail "TODO" + +runExceptT' :: Exception e => ExceptT e IO a -> IO a +runExceptT' a = runExceptT a >>= either throwIO pure + +waitSendRecv :: IO () -> IO () -> IO () +waitSendRecv s r = do + s' <- async s + r' <- async r + waitCatch s' >>= either (\e -> cancel r' >> fail (show e)) pure + waitCatch r' >>= either (\e -> cancel s' >> fail (show e)) pure diff --git a/tests/Util.hs b/tests/Util.hs index a52fee32c..6ad6d054f 100644 --- a/tests/Util.hs +++ b/tests/Util.hs @@ -1,6 +1,28 @@ module Util where +import Control.Monad (replicateM) +import Data.Either (partitionEithers) +import Data.List (tails) +import GHC.Conc (getNumCapabilities, getNumProcessors, setNumCapabilities) import Test.Hspec +import UnliftIO skip :: String -> SpecWith a -> SpecWith a skip = before_ . pendingWith + +withNumCapabilities :: Int -> IO a -> IO a +withNumCapabilities new a = getNumCapabilities >>= \old -> bracket_ (setNumCapabilities new) (setNumCapabilities old) a + +withNCPUCapabilities :: IO a -> IO a +withNCPUCapabilities a = getNumProcessors >>= \p -> withNumCapabilities p a + +inParrallel :: Int -> IO () -> IO () +inParrallel n action = do + streams <- replicateM n $ async action + (es, rs) <- partitionEithers <$> mapM waitCatch streams + map show es `shouldBe` [] + length rs `shouldBe` n + +combinations :: Int -> [a] -> [[a]] +combinations 0 _ = [[]] +combinations k xs = [y : ys | y : xs' <- tails xs, ys <- combinations (k - 1) xs']