Files
simplexmq/tests/SMPClient.hs
T
Evgeny f0b7a4be73 messaging services (#1667)
* smp server: messaging services (#1565)

* smp server: refactor message delivery to always respond SOK to subscriptions

* refactor ntf subscribe

* cancel subscription thread and reduce service subscription count when queue is deleted

* subscribe rcv service, deliver sent messages to subscribed service

* subscribe rcv service to messages (TODO delivery on subscription)

* WIP

* efficient initial delivery of messages to subscribed service

* test: delivery to client with service certificate

* test: upgrade/downgrade to/from service subscriptions

* remove service association from agent API, add per-user flag to use the service

* agent client (WIP)

* service certificates in the client

* rfc about drift detection, and SALL to mark end of message delivery

* fix test

* fix test

* add function for postgresql message storage

* update migration

* servers: maintain xor-hash of all associated queue IDs in PostgreSQL (#1668)

* servers: maintain xor-hash of all associated queue IDs in PostgreSQL (#1615)

* ntf server: maintain xor-hash of all associated queue IDs via PostgreSQL triggers

* smp server: xor hash with triggers

* fix sql and using pgcrypto extension in tests

* track counts and hashes in smp/ntf servers via triggers, smp server stats for service subscription, update SMP protocol to pass expected count and hash in SSUB/NSSUB commands

* agent migrations with functions/triggers

* remove agent triggers

* try tracking service subs in the agent (WIP, does not compile)

* Revert "try tracking service subs in the agent (WIP, does not compile)"

This reverts commit 59e908100d.

* comment

* agent database triggers

* service subscriptions in the client

* test / fix client services

* update schema

* fix postgres migration

* update schema

* move schema test to the end

* use static function with SQLite to avoid dynamic wrapper

* agent: fail when per-connection transport isolation is used with services (#1670)

* agent: service subscription events (#1671)

* agent: use server keyhash when loading service record

* agent: process queue/service associations with delayed subscription results

* agent: service subscription events

* agent: finalize initial service subscriptions, remove associations on service ID changes (#1672)

* agent: remove service/queue associations when service ID changes

* agent: check that service ID in NEW response matches session ID in transport session

* agent subscription WIP

* test

* comment

* enable tests

* update queries

* agent: option to add SQLite aggregates to DB connection  (#1673)

* agent: add build_relations_vector function to sqlite

* update aggregate

* use static aggregate

* remove relations

---------

Co-authored-by: Evgeny Poberezkin <evgeny@poberezkin.com>

* add test, treat BAD_SERVICE as temp error, only remove queue associations on service errors

* add packZipWith for backward compatibility with GHC 8.10.7

---------

Co-authored-by: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com>

* servers: service stats and logging, allow services without option (removed), report errors during service message delivery, remove threads when service subscription ended (#1676)

* smp server: always allow services without option

* smp server: maintain IDs hash in session subscription states

* smp server: service message delivery error handling

* ntf server: log subscription count and hash differences

* smp server: remove delivery threads when service subscription ended/client disconnected

* agent: remove service queue association when service ID changed, process ENDS event, test migrating to/from service (#1677)

* agent: remove service queue association when service ID changed

* agent: process ENDS event

* agent: send service subscription error event

* agent: test migrating to/from service subscriptions, fixes

* agent: always remove service when disabled, fix service subscriptions

* ntf server: use different client certs for each SMP server, remove support for store log (#1681)

* ntf server: remove support for store log

* ntf server: use different client certificates for each SMP server

* smp protocol: fix encoding for SOKS/ENDS responses (#1683)

* agent: create user with option to enable client service (#1684)

* agent: create user with option to enable client service

* handle HTTP2 errors

* do not catch async exceptions

* agent: minor fixes

* docs: update protocol (#1705)

* docs: agent threat model

* update protocol docs

* update RFCs (#1730)

* update RFCs

* update

* update overview

* update terminology

* original language in threat model

---------

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

* docs: fix minor issues in protocols

* docs: add e2e encrypted message wire encoding to PQDR spec

* docs: add missing encodings and other protocol corrections

* docs: move implemented rfcs

* smp: service fixes (#1737)

* smp: deliver service subscription to correct client

* tests: more resilient to concurrency

* optimize PostgreSQL query

* fix service re-association after server "downgrade"

* correctly handle service removed from server (and ID changed)

* remove unused

---------

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

* prometheus: fix metrics names (#1747)

* test: rcv service re-association on restart (#1746)

* agent: correct log message

* docs: update whitepaper

* smp: fix messaging client service issues (#1751)

* services: fix minor issues

* fix accounting for subscribed service queues, add prometheus stats

* fix uncorrelated subquery

* fix potential race condition when inserting service defensively, as it is also prevented by how client is created

---------

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

* agent: refactor cleanup if no pending subs (#1757)

* smp server: batch processing of subscription messages (#1753)

* smp server: batch processing of subscription messages

* refactor

* empty line

* fix

---------

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

* smp: batch queue association updates on subscriptions (#1760)

* smp: batch queue association updates on subscriptions

* refactor to fused batching

* simpler

* batch assoc functions

* clean up

* fix

---------

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

* agent: use primary key index in setRcvServiceAssocs (#1783)

* agent: use primary key index in setRcvServiceAssocs

Previous WHERE rcv_id = ? did not match the (host, port, rcv_id)
primary key prefix and fell back to a table scan via
idx_rcv_queues_client_notice_id. With ~390k rows per queue, each
update in a 1350-row batch scanned the whole table, yielding ~290s
per batch and a multi-hour rcv-services migration.

* agent: pass SMPServer explicitly to setRcvServiceAssocs

Avoid extracting host/port from the first queue inside setRcvServiceAssocs.
The caller already has SMPServer in scope (from tSess) and the call chain
is short, so threading it through is simpler than inspecting the list.
Removes the empty-list guard from setRcvServiceAssocs (it remains in
processRcvServiceAssocs).

---------

Co-authored-by: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com>
Co-authored-by: Evgeny @ SimpleX Chat <259188159+evgeny-simplex@users.noreply.github.com>
Co-authored-by: sh <37271604+shumvgolove@users.noreply.github.com>
2026-05-21 14:14:03 +01:00

471 lines
21 KiB
Haskell

{-# LANGUAGE CPP #-}
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE DuplicateRecordFields #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE NumericUnderscores #-}
{-# LANGUAGE OverloadedLists #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE RankNTypes #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TypeApplications #-}
{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
module SMPClient where
import Control.Monad
import Control.Monad.Except (runExceptT)
import Data.ByteString.Char8 (ByteString)
import Data.List.NonEmpty (NonEmpty)
import qualified Data.X509 as X
import qualified Data.X509.Validation as XV
import Network.Socket
import qualified Network.TLS as TLS
import Simplex.Messaging.Agent.Store.Postgres.Options (DBOpts (..))
import Simplex.Messaging.Agent.Store.Shared (MigrationConfirmation (..))
import Simplex.Messaging.Client (ProtocolClientConfig (..), chooseTransportHost, defaultNetworkConfig)
import Simplex.Messaging.Client.Agent (SMPClientAgentConfig (..), defaultSMPClientAgentConfig)
import qualified Simplex.Messaging.Crypto as C
import Simplex.Messaging.Encoding
import Simplex.Messaging.Protocol
import Simplex.Messaging.Server (runSMPServerBlocking)
import Simplex.Messaging.Server.Env.STM
import Simplex.Messaging.Server.MsgStore.Types (MsgStoreClass (..), SMSType (..), SQSType (..))
import Simplex.Messaging.Server.QueueStore.Postgres.Config (PostgresStoreCfg (..))
import Simplex.Messaging.Transport
import Simplex.Messaging.Transport.Client
import Simplex.Messaging.Transport.Server
import Simplex.Messaging.Transport.Shared (ChainCertificates (..), chainIdCaCerts)
import Simplex.Messaging.Util (ifM)
import Simplex.Messaging.Version
import Simplex.Messaging.Version.Internal
import System.Info (os)
import Test.Hspec hiding (fit, it)
import UnliftIO.Concurrent
import qualified UnliftIO.Exception as E
import UnliftIO.STM (TMVar, atomically, newEmptyTMVarIO, putTMVar, takeTMVar)
import UnliftIO.Timeout (timeout)
import Util
#if defined(dbServerPostgres)
import Database.PostgreSQL.Simple (defaultConnectInfo)
#endif
#if defined(dbPostgres) || defined(dbServerPostgres)
import Database.PostgreSQL.Simple (ConnectInfo (..))
import Simplex.Messaging.Agent.Store.Postgres.Util (createDBAndUserIfNotExists, dropDatabaseAndUser)
#endif
data AServerConfig =
forall qs ms. (SupportedStore qs ms, MsgStoreClass (MsgStoreType qs ms)) =>
ASrvCfg (SQSType qs) (SMSType ms) (ServerConfig (MsgStoreType qs ms))
data AServerStoreCfg =
forall qs ms. (SupportedStore qs ms, MsgStoreClass (MsgStoreType qs ms)) =>
ASSCfg (SQSType qs) (SMSType ms) (ServerStoreCfg (MsgStoreType qs ms))
testHost :: NonEmpty TransportHost
testHost = "localhost"
testHost2 :: NonEmpty TransportHost
testHost2 = "127.0.0.1"
testPort :: ServiceName
testPort = "5001"
testPort2 :: ServiceName
testPort2 = "5002"
ntfTestPort :: ServiceName
ntfTestPort = "6001"
ntfTestPort2 :: ServiceName
ntfTestPort2 = "6002"
testKeyHash :: C.KeyHash
testKeyHash = "LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI="
testStoreLogFile :: FilePath
testStoreLogFile = "tests/tmp/smp-server-store.log"
testStoreLogFile2 :: FilePath
testStoreLogFile2 = "tests/tmp/smp-server-store.log.2"
testStoreDBOpts :: DBOpts
testStoreDBOpts =
DBOpts
{ connstr = testServerDBConnstr,
schema = "smp_server",
poolSize = 10,
createSchema = True
}
testStoreDBOpts2 :: DBOpts
testStoreDBOpts2 = testStoreDBOpts {schema = "smp_server2"}
testServerDBConnstr :: ByteString
testServerDBConnstr = "postgresql://test_server_user@/test_server_db"
#if defined(dbServerPostgres)
testServerDBConnectInfo :: ConnectInfo
testServerDBConnectInfo =
defaultConnectInfo {
connectUser = "test_server_user",
connectDatabase = "test_server_db"
}
#endif
testStoreMsgsFile :: FilePath
testStoreMsgsFile = "tests/tmp/smp-server-messages.log"
testStoreMsgsFile2 :: FilePath
testStoreMsgsFile2 = "tests/tmp/smp-server-messages.log.2"
testStoreMsgsDir :: FilePath
testStoreMsgsDir = "tests/tmp/messages"
testStoreMsgsDir2 :: FilePath
testStoreMsgsDir2 = "tests/tmp/messages.2"
testStoreNtfsFile :: FilePath
testStoreNtfsFile = "tests/tmp/smp-server-ntfs.log"
testStoreNtfsFile2 :: FilePath
testStoreNtfsFile2 = "tests/tmp/smp-server-ntfs.log.2"
testPrometheusMetricsFile :: FilePath
testPrometheusMetricsFile = "tests/tmp/smp-server-metrics.txt"
testServerStatsBackupFile :: FilePath
testServerStatsBackupFile = "tests/tmp/smp-server-stats.log"
xit' :: (HasCallStack, Example a) => String -> a -> SpecWith (Arg a)
xit' d = if os == "linux" then skip "skipped on Linux" . it d else it d
xit'' :: (HasCallStack, Example a) => String -> a -> SpecWith (Arg a)
xit'' d = skipOnCI . it d
skipOnCI :: SpecWith a -> SpecWith a
skipOnCI t = ifM (runIO envCI) (skip "skipped on CI" t) t
testSMPClient :: Transport c => (THandleSMP c 'TClient -> IO a) -> IO a
testSMPClient = testSMPClientVR supportedClientSMPRelayVRange
testSMPClientVR :: Transport c => VersionRangeSMP -> (THandleSMP c 'TClient -> IO a) -> IO a
testSMPClientVR vr client = do
Right useHost <- pure $ chooseTransportHost defaultNetworkConfig testHost
testSMPClient_ useHost testPort vr Nothing client
testSMPServiceClient :: Transport c => (TLS.Credential, C.KeyPairEd25519) -> (THandleSMP c 'TClient -> IO a) -> IO a
testSMPServiceClient serviceCreds client = do
Right useHost <- pure $ chooseTransportHost defaultNetworkConfig testHost
testSMPClient_ useHost testPort supportedClientSMPRelayVRange (Just serviceCreds) client
testSMPClient_ :: Transport c => TransportHost -> ServiceName -> VersionRangeSMP -> Maybe (TLS.Credential, C.KeyPairEd25519) -> (THandleSMP c 'TClient -> IO a) -> IO a
testSMPClient_ host port vr serviceCreds_ client = do
serviceAndKeys_ <- forM serviceCreds_ $ \(serviceCreds@(cc, pk), keys) -> do
Right serviceSignKey <- pure $ C.x509ToPrivate' pk
let idCert' = case chainIdCaCerts cc of
CCSelf cert -> cert
CCValid {idCert} -> idCert
_ -> error "bad certificate"
serviceCertHash = XV.getFingerprint idCert' X.HashSHA256
pure (ServiceCredentials {serviceRole = SRMessaging, serviceCreds, serviceCertHash, serviceSignKey}, keys)
let tcConfig = defaultTransportClientConfig {clientALPN, clientCredentials = fst <$> serviceCreds_} :: TransportClientConfig
runTransportClient tcConfig Nothing host port (Just testKeyHash) $ \h ->
runExceptT (smpClientHandshake h Nothing testKeyHash vr False serviceAndKeys_) >>= \case
Right th -> client th
Left e -> error $ show e
where
clientALPN
| authCmdsSMPVersion `isCompatible` vr = Just alpnSupportedSMPHandshakes
| otherwise = Nothing
runSMPClient :: Transport c => TProxy c 'TServer -> (THandleSMP c 'TClient -> IO a) -> IO a
runSMPClient _ test' = testSMPClient test'
runSMPServiceClient :: Transport c => TProxy c 'TServer -> (TLS.Credential, C.KeyPairEd25519) -> (THandleSMP c 'TClient -> IO a) -> IO a
runSMPServiceClient _ serviceCreds test' = testSMPServiceClient serviceCreds test'
testNtfServiceClient :: Transport c => TProxy c 'TServer -> C.KeyPairEd25519 -> (THandleSMP c 'TClient -> IO a) -> IO a
testNtfServiceClient _ keys client = do
tlsNtfServerCreds <- loadServerCredential ntfTestServerCredentials
serviceCertHash <- loadFingerprint ntfTestServerCredentials
Right serviceSignKey <- pure $ C.x509ToPrivate' $ snd tlsNtfServerCreds
let service = ServiceCredentials {serviceRole = SRNotifier, serviceCreds = tlsNtfServerCreds, serviceCertHash, serviceSignKey}
tcConfig =
defaultTransportClientConfig
{ clientCredentials = Just tlsNtfServerCreds,
clientALPN = Just alpnSupportedSMPHandshakes
}
runTransportClient tcConfig Nothing "localhost" testPort (Just testKeyHash) $ \h ->
runExceptT (smpClientHandshake h Nothing testKeyHash supportedClientSMPRelayVRange False $ Just (service, keys)) >>= \case
Right th -> client th
Left e -> error $ show e
ntfTestServerCredentials :: ServerCredentials
ntfTestServerCredentials =
ServerCredentials
{ caCertificateFile = Just "tests/fixtures/ca.crt",
privateKeyFile = "tests/fixtures/server.key",
certificateFile = "tests/fixtures/server.crt"
}
cfg :: AServerConfig
cfg = cfgMS (ASType SQSMemory SMSJournal)
cfgJ2 :: AServerConfig
cfgJ2 = journalCfg cfg testStoreLogFile2 testStoreMsgsDir2
cfgJ2QS :: SQSType s -> AServerConfig
cfgJ2QS = \case
SQSMemory -> journalCfg (cfgMS $ ASType SQSMemory SMSJournal) testStoreLogFile2 testStoreMsgsDir2
SQSPostgres -> journalCfgDB (cfgMS $ ASType SQSPostgres SMSJournal) testStoreDBOpts2 testStoreMsgsDir2
journalCfg :: AServerConfig -> FilePath -> FilePath -> AServerConfig
journalCfg (ASrvCfg _ _ cfg') storeLogFile storeMsgsPath =
ASrvCfg SQSMemory SMSJournal cfg' {serverStoreCfg = SSCMemoryJournal {storeLogFile, storeMsgsPath}}
journalCfgDB :: AServerConfig -> DBOpts -> FilePath -> AServerConfig
journalCfgDB (ASrvCfg _ _ cfg') dbOpts storeMsgsPath' =
let storeCfg = PostgresStoreCfg {dbOpts, dbStoreLogPath = Nothing, confirmMigrations = MCYesUp, deletedTTL = 86400}
in ASrvCfg SQSPostgres SMSJournal cfg' {serverStoreCfg = SSCDatabaseJournal {storeCfg, storeMsgsPath'}}
cfgMS :: AStoreType -> AServerConfig
cfgMS msType = withStoreCfg (testServerStoreConfig msType) $ \serverStoreCfg ->
ServerConfig
{ transports = [],
smpHandshakeTimeout = 60000000,
tbqSize = 4,
msgQueueQuota = 4,
maxJournalMsgCount = 5,
maxJournalStateLines = 2,
queueIdBytes = 24,
msgIdBytes = 24,
serverStoreCfg,
storeNtfsFile = Nothing,
allowNewQueues = True,
newQueueBasicAuth = Nothing,
controlPortUserAuth = Nothing,
controlPortAdminAuth = Nothing,
dailyBlockQueueQuota = 20,
messageExpiration = Just defaultMessageExpiration,
expireMessagesOnStart = True,
expireMessagesOnSend = False,
idleQueueInterval = defaultIdleQueueInterval,
notificationExpiration = defaultNtfExpiration,
inactiveClientExpiration = Just defaultInactiveClientExpiration,
logStatsInterval = Nothing,
logStatsStartTime = 0,
serverStatsLogFile = "tests/tmp/smp-server-stats.daily.log",
serverStatsBackupFile = Nothing,
prometheusInterval = Nothing,
prometheusMetricsFile = testPrometheusMetricsFile,
pendingENDInterval = 500000,
ntfDeliveryInterval = 200000,
smpCredentials =
ServerCredentials
{ caCertificateFile = Just "tests/fixtures/ca.crt",
privateKeyFile = "tests/fixtures/server.key",
certificateFile = "tests/fixtures/server.crt"
},
httpCredentials = Nothing,
smpServerVRange = supportedServerSMPRelayVRange,
transportConfig = mkTransportServerConfig True (Just alpnSupportedSMPHandshakes) True,
controlPort = Nothing,
smpAgentCfg = defaultSMPClientAgentConfig {persistErrorInterval = 1}, -- seconds
allowSMPProxy = False,
serverClientConcurrency = 2,
information = Nothing,
startOptions = defaultStartOptions
}
withStoreCfg :: AServerStoreCfg -> (forall s. ServerStoreCfg s -> ServerConfig s) -> AServerConfig
withStoreCfg (ASSCfg qt mt storeCfg) f = ASrvCfg qt mt (f storeCfg)
defaultStartOptions :: StartOptions
defaultStartOptions = StartOptions {maintenance = False, compactLog = False, logLevel = testLogLevel, skipWarnings = False, confirmMigrations = MCYesUp}
testServerStoreConfig :: AStoreType -> AServerStoreCfg
testServerStoreConfig = serverStoreConfig_ False
serverStoreConfig_ :: Bool -> AStoreType -> AServerStoreCfg
serverStoreConfig_ useDbStoreLog = \case
ASType SQSMemory SMSMemory ->
ASSCfg SQSMemory SMSMemory $ SSCMemory $ Just StorePaths {storeLogFile = testStoreLogFile, storeMsgsFile = Just testStoreMsgsFile}
ASType SQSMemory SMSJournal ->
ASSCfg SQSMemory SMSJournal $ SSCMemoryJournal {storeLogFile = testStoreLogFile, storeMsgsPath = testStoreMsgsDir}
ASType SQSPostgres SMSJournal ->
ASSCfg SQSPostgres SMSJournal SSCDatabaseJournal {storeCfg, storeMsgsPath' = testStoreMsgsDir}
#if defined(dbServerPostgres)
ASType SQSPostgres SMSPostgres ->
ASSCfg SQSPostgres SMSPostgres $ SSCDatabase storeCfg
#endif
where
dbStoreLogPath = if useDbStoreLog then Just testStoreLogFile else Nothing
storeCfg = PostgresStoreCfg {dbOpts = testStoreDBOpts, dbStoreLogPath, confirmMigrations = MCYesUp, deletedTTL = 86400}
cfgV7 :: AServerConfig
cfgV7 = updateCfg cfg $ \cfg' -> cfg' {smpServerVRange = mkVersionRange minServerSMPRelayVersion authCmdsSMPVersion}
cfgVPrev :: AStoreType -> AServerConfig
cfgVPrev msType = updateCfg (cfgMS msType) $ \cfg' -> cfg' {smpServerVRange = prevRange $ smpServerVRange cfg'}
prevRange :: VersionRange v -> VersionRange v
prevRange vr = vr {maxVersion = max (minVersion vr) (prevVersion $ maxVersion vr)}
prevVersion :: Version v -> Version v
prevVersion (Version v) = Version (v - 1)
proxyCfg :: AServerConfig
proxyCfg = proxyCfgMS (ASType SQSMemory SMSJournal)
proxyCfgMS :: AStoreType -> AServerConfig
proxyCfgMS msType =
updateCfg (cfgMS msType) $ \cfg' ->
let smpAgentCfg' = smpAgentCfg cfg'
in cfg'
{ allowSMPProxy = True,
smpAgentCfg = smpAgentCfg' {smpCfg = (smpCfg smpAgentCfg') {agreeSecret = True, proxyServer = True, serverVRange = supportedProxyClientSMPRelayVRange}}
}
proxyCfgJ2 :: AServerConfig
proxyCfgJ2 = journalCfg proxyCfg testStoreLogFile2 testStoreMsgsDir2
proxyCfgJ2QS :: SQSType qs -> AServerConfig
proxyCfgJ2QS = \case
SQSMemory -> journalCfg (proxyCfgMS $ ASType SQSMemory SMSJournal) testStoreLogFile2 testStoreMsgsDir2
SQSPostgres -> journalCfgDB (proxyCfgMS $ ASType SQSPostgres SMSJournal) testStoreDBOpts2 testStoreMsgsDir2
proxyVRangeV8 :: VersionRangeSMP
proxyVRangeV8 = mkVersionRange minServerSMPRelayVersion sendingProxySMPVersion
withSmpServerStoreMsgLogOn :: HasCallStack => (ASrvTransport, AStoreType) -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a
withSmpServerStoreMsgLogOn (t, msType) =
withSmpServerConfigOn t $ updateCfg (cfgMS msType) $ \cfg' -> cfg' {storeNtfsFile = Just testStoreNtfsFile, serverStatsBackupFile = Just testServerStatsBackupFile}
withSmpServerStoreLogOn :: HasCallStack => (ASrvTransport, AStoreType) -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a
withSmpServerStoreLogOn (t, msType) =
withSmpServerConfigOn t $ updateCfg (cfgMS msType) $ \cfg' -> cfg' {serverStatsBackupFile = Just testServerStatsBackupFile}
updateCfg :: AServerConfig -> (forall s. ServerConfig s -> ServerConfig s) -> AServerConfig
updateCfg (ASrvCfg qt mt cfg') f = ASrvCfg qt mt (f cfg')
withServerCfg :: AServerConfig -> (forall s. ServerConfig s -> a) -> a
withServerCfg (ASrvCfg _ _ cfg') f = f cfg'
withSmpServerConfigOn :: HasCallStack => ASrvTransport -> AServerConfig -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a
withSmpServerConfigOn t (ASrvCfg _ _ cfg') port' =
serverBracket
(\started -> runSMPServerBlocking started cfg' {transports = [(port', t, False)]} Nothing)
(threadDelay 10000)
withSmpServerThreadOn :: HasCallStack => (ASrvTransport, AStoreType) -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a
withSmpServerThreadOn (t, msType) = withSmpServerConfigOn t (cfgMS msType)
serverBracket :: HasCallStack => (TMVar Bool -> IO ()) -> IO () -> (HasCallStack => ThreadId -> IO a) -> IO a
serverBracket process afterProcess f = do
started <- newEmptyTMVarIO
E.bracket
(forkIOWithUnmask (\unmask -> unmask (process started) `E.catchAny` handleStartError started))
(\t -> killThread t >> afterProcess >> waitFor started "stop")
(\t -> waitFor started "start" >> f t >>= \r -> r <$ threadDelay 100000)
where
-- it putTMVar is called twise to unlock both parts of the bracket in case of start failure
handleStartError started e = do
atomically $ putTMVar started False
atomically $ putTMVar started False
E.throwIO e
waitFor started s =
5_000_000 `timeout` atomically (takeTMVar started) >>= \case
Nothing -> error $ "server did not " <> s
_ -> pure ()
withSmpServerOn :: HasCallStack => (ASrvTransport, AStoreType) -> ServiceName -> IO a -> IO a
withSmpServerOn ps port' = withSmpServerThreadOn ps port' . const
withSmpServer :: HasCallStack => (ASrvTransport, AStoreType) -> IO a -> IO a
withSmpServer ps = withSmpServerOn ps testPort
withSmpServerProxy :: HasCallStack => (ASrvTransport, AStoreType) -> IO a -> IO a
withSmpServerProxy (t, msType) = withSmpServerConfigOn t (proxyCfgMS msType) testPort . const
withSmpServers2 :: HasCallStack => (ASrvTransport, AStoreType) -> IO a -> IO a
withSmpServers2 ps@(t, ASType qs _ms) = withSmpServer ps . withSmpServerConfigOn t (cfgJ2QS qs) testPort2 . const
withSmpServersProxy2 :: HasCallStack => (ASrvTransport, AStoreType) -> IO a -> IO a
withSmpServersProxy2 ps@(t, ASType qs _ms) = withSmpServerProxy ps . withSmpServerConfigOn t (proxyCfgJ2QS qs) testPort2 . const
runSmpTest :: forall c a. (HasCallStack, Transport c) => AStoreType -> (HasCallStack => THandleSMP c 'TClient -> IO a) -> IO a
runSmpTest msType test = withSmpServerConfigOn (transport @c) (cfgMS msType) testPort $ \_ -> testSMPClient test
runSmpTestN :: forall c a. (HasCallStack, Transport c) => AStoreType -> Int -> (HasCallStack => [THandleSMP c 'TClient] -> IO a) -> IO a
runSmpTestN msType = runSmpTestNCfg (cfgMS msType) supportedClientSMPRelayVRange
runSmpTestNCfg :: forall c a. (HasCallStack, Transport c) => AServerConfig -> VersionRangeSMP -> Int -> (HasCallStack => [THandleSMP c 'TClient] -> IO a) -> IO a
runSmpTestNCfg srvCfg clntVR nClients test = withSmpServerConfigOn (transport @c) srvCfg testPort $ \_ -> run nClients []
where
run :: Int -> [THandleSMP c 'TClient] -> IO a
run 0 hs = test hs
run n hs = testSMPClientVR clntVR $ \h -> run (n - 1) (h : hs)
smpServerTest ::
forall c smp.
(Transport c, Encoding smp) =>
TProxy c 'TServer ->
(Maybe TAuthorizations, ByteString, ByteString, smp) ->
IO (Maybe TAuthorizations, ByteString, ByteString, BrokerMsg)
smpServerTest _ t = runSmpTest (ASType SQSMemory SMSJournal) $ \h -> tPut' h t >> tGet' h
where
tPut' :: THandleSMP c 'TClient -> (Maybe TAuthorizations, ByteString, ByteString, smp) -> IO ()
tPut' h@THandle {params = THandleParams {sessionId, implySessId}} (sig, corrId, queueId, smp) = do
let t' = if implySessId then smpEncode (corrId, queueId, smp) else smpEncode (sessionId, corrId, queueId, smp)
[Right ()] <- tPut h [Right (sig, t')]
pure ()
tGet' h = do
[(CorrId corrId, EntityId qId, Right cmd)] <- tGetClient h
pure (Nothing, corrId, qId, cmd)
smpTest :: (HasCallStack, Transport c) => TProxy c 'TServer -> AStoreType -> (HasCallStack => THandleSMP c 'TClient -> IO ()) -> Expectation
smpTest _ msType test' = runSmpTest msType test' `shouldReturn` ()
smpTestN :: (HasCallStack, Transport c) => AStoreType -> Int -> (HasCallStack => [THandleSMP c 'TClient] -> IO ()) -> Expectation
smpTestN msType n test' = runSmpTestN msType n test' `shouldReturn` ()
smpTest2 :: forall c. (HasCallStack, Transport c) => TProxy c 'TServer -> AStoreType -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
smpTest2 t msType = smpTest2Cfg (cfgMS msType) supportedClientSMPRelayVRange t
smpTest2Cfg :: forall c. (HasCallStack, Transport c) => AServerConfig -> VersionRangeSMP -> TProxy c 'TServer -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
smpTest2Cfg srvCfg clntVR _ test' = runSmpTestNCfg srvCfg clntVR 2 _test `shouldReturn` ()
where
_test :: HasCallStack => [THandleSMP c 'TClient] -> IO ()
_test [h1, h2] = test' h1 h2
_test _ = error "expected 2 handles"
smpTest3 :: forall c. (HasCallStack, Transport c) => TProxy c 'TServer -> AStoreType -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
smpTest3 _ msType test' = smpTestN msType 3 _test
where
_test :: HasCallStack => [THandleSMP c 'TClient] -> IO ()
_test [h1, h2, h3] = test' h1 h2 h3
_test _ = error "expected 3 handles"
smpTest4 :: forall c. (HasCallStack, Transport c) => TProxy c 'TServer -> AStoreType -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
smpTest4 _ msType test' = smpTestN msType 4 _test
where
_test :: HasCallStack => [THandleSMP c 'TClient] -> IO ()
_test [h1, h2, h3, h4] = test' h1 h2 h3 h4
_test _ = error "expected 4 handles"
unexpected :: (HasCallStack, Show a) => a -> Expectation
unexpected r = expectationFailure $ "unexpected response " <> show r
#if defined(dbPostgres) || defined(dbServerPostgres)
postgressBracket :: ConnectInfo -> IO a -> IO a
postgressBracket connInfo =
E.bracket_
(dropDatabaseAndUser connInfo >> createDBAndUserIfNotExists connInfo)
(dropDatabaseAndUser connInfo)
#endif