dandri/simplexmq
1
0
Fork 0
mirror of https://github.com/simplex-chat/simplexmq.git synced 2026-08-22 21:00:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

tests: add proxy stress tests (#1163)

Browse Source 
* tests: add proxy stress tests

* organize benches

* add agent tests

* move prints to logNote

* fix stuck agent tests
This commit is contained in:
Alexander Bondarenko
2024-05-23 15:34:25 +01:00
committed by GitHub
parent 6c86aa302f
commit 5d38ad03af
2 changed files with 165 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/SMPProxyTests.hs
+143 -29
View File
@@ -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
tests/Util.hs
+22
View File
@@ -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']