smp-server: remove journal message store (#1919)

Co-authored-by: sh <github.shum@liber.li>
Co-authored-by: Evgeny @ SimpleX Chat <259188159+evgeny-simplex@users.noreply.github.com>
This commit is contained in:
authored and GitHub committed 2026-10-15 17:37:49 +01:00
1 parent 9d83c0378e
commit f1f079044c
21 files changed
+167 -1942

No files matched your search

+1 -3
View File
@@ -144,9 +144,7 @@ space, so it is workload-dependent; `fillfactor = 80` reached 100% for this frac
larger share of queues at once would need a lower value. `fillfactor` applies to pages rewritten after
the change, so the ratio ramps up as the heap turns over, or immediately after `pg_repack`.
Dropping the index regresses the deprecated journal message store's expiration (`foldRecentQueueRecs`,
`Journal.hs:434`) to a sequential scan; this is accepted because that store is being retired. The
Postgres message store does not read `updated_at`.
The Postgres message store does not read `updated_at`, so no query depends on the dropped index.
Autovacuum reloptions applied by the migration:
-2
View File
@@ -284,8 +284,6 @@ library
Simplex.Messaging.Server.Main.Init
Simplex.Messaging.Server.Web
Simplex.Messaging.Server.MsgStore
Simplex.Messaging.Server.MsgStore.Journal
Simplex.Messaging.Server.MsgStore.Journal.SharedLock
Simplex.Messaging.Server.MsgStore.STM
Simplex.Messaging.Server.MsgStore.Types
Simplex.Messaging.Server.Names
+4 -32
View File
@@ -108,7 +108,6 @@ import Simplex.Messaging.Server.Control
import Simplex.Messaging.Server.Env.STM as Env
import Simplex.Messaging.Server.Expiration
import Simplex.Messaging.Server.MsgStore
import Simplex.Messaging.Server.MsgStore.Journal (JournalMsgStore, JournalQueue (..), getJournalQueueMessages)
import Simplex.Messaging.Server.Names (NamesEnv, closeNamesEnv, resolveName)
import Simplex.Messaging.Server.MsgStore.STM
import Simplex.Messaging.Server.MsgStore.Types
@@ -2350,7 +2349,6 @@ saveServerMessages drainMsgs ms = case ms of
StoreMemory STMMsgStore {storeConfig = STMStoreConfig {storePath}} -> case storePath of
Just f -> exportMessages False ms f drainMsgs
Nothing -> logNote "undelivered messages are not saved"
StoreJournal _ -> logNote "closed journal message storage"
#if defined(dbServerPostgres)
StoreDatabase _ -> logNote "closed postgres message storage"
#endif
@@ -2360,7 +2358,6 @@ exportMessages tty st f drainMsgs = do
logNote $ "saving messages to file " <> T.pack f
run $ case st of
StoreMemory ms -> exportMessages_ ms $ getMsgs ms
StoreJournal ms -> exportMessages_ ms $ getJournalMsgs ms
#if defined(dbServerPostgres)
StoreDatabase ms -> exportDbMessages tty ms
#endif
@@ -2372,10 +2369,6 @@ exportMessages tty st f drainMsgs = do
Left e -> do
logError $ "error exporting messages: " <> tshow e
exitFailure
getJournalMsgs ms q =
readTVarIO (msgQueue' q) >>= \case
Just _ -> getMsgs ms q
Nothing -> getJournalQueueMessages ms q
getMsgs :: MsgStoreClass s' => s' -> StoreQueue s' -> IO [Message]
getMsgs ms q = unsafeRunStore q "saveQueueMsgs" $ getQueueMessages_ drainMsgs q =<< getMsgQueue ms q False
saveQueueMsgs :: (StoreQueue s -> IO [Message]) -> Handle -> StoreQueue s -> IO (Sum Int)
@@ -2392,37 +2385,16 @@ processServerMessages StartOptions {skipWarnings} = do
asks msgStore_ >>= liftIO . processMessages old_ expire
where
processMessages :: Maybe Int64 -> Bool -> MsgStore s' -> IO (Maybe MessageStats)
#if defined(dbServerPostgres)
processMessages old_ expire = \case
#else
processMessages old_ _ = \case
#endif
StoreMemory ms@STMMsgStore {storeConfig = STMStoreConfig {storePath}} -> case storePath of
Just f -> ifM (doesFileExist f) (Just <$> importMessages False ms f old_ skipWarnings) (pure Nothing)
Nothing -> pure Nothing
StoreJournal ms -> processJournalMessages old_ expire ms
#if defined(dbServerPostgres)
StoreDatabase ms -> processDbMessages old_ expire ms
#endif
processJournalMessages :: forall s. Maybe Int64 -> Bool -> JournalMsgStore s -> IO (Maybe MessageStats)
processJournalMessages old_ expire ms
| expire = Just <$> case old_ of
Just old -> do
logNote "expiring journal store messages..."
run $ processExpireQueue old
Nothing -> do
logNote "validating journal store messages..."
run processValidateQueue
| otherwise = logWarn "skipping message expiration" $> Nothing
where
run a = unsafeWithAllMsgQueues False ms a `catchAny` \_ -> exitFailure
processExpireQueue :: Int64 -> JournalQueue s -> IO MessageStats
processExpireQueue old q = unsafeRunStore q "processExpireQueue" $ do
mq <- getMsgQueue ms q False
expiredMsgsCount <- deleteExpireMsgs_ old q mq
storedMsgsCount <- getQueueSize_ mq
pure MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues = 1}
processValidateQueue :: JournalQueue s -> IO MessageStats
processValidateQueue q = unsafeRunStore q "processValidateQueue" $ do
storedMsgsCount <- getQueueSize_ =<< getMsgQueue ms q False
pure newMessageStats {storedMsgsCount, storedQueues = 1}
#if defined(dbServerPostgres)
processDbMessages old_ expire ms
| expire = Just <$> case old_ of
Just old -> do
+4 -71
View File
@@ -42,7 +42,6 @@ module Simplex.Messaging.Server.Env.STM
VerifiedTransmission,
ResponseAndMessage,
newEnv,
mkJournalStoreConfig,
msgStore,
fromMsgStore,
newClient,
@@ -70,10 +69,6 @@ module Simplex.Messaging.Server.Env.STM
defaultNameResolverConcurrency,
defaultMaxClientCmdWait,
defaultAuthErrorDelayMs,
defaultMaxJournalMsgCount,
defaultMaxJournalStateLines,
defaultIdleQueueInterval,
journalMsgStoreDepth,
readWriteQueueStore,
noPostgresExitStr,
noPostgresExit,
@@ -98,7 +93,7 @@ import Data.List.NonEmpty (NonEmpty)
import Data.Map.Strict (Map)
import Data.Maybe (isJust)
import qualified Data.Text as T
import Data.Time.Clock (getCurrentTime, nominalDay)
import Data.Time.Clock (getCurrentTime)
import Data.Time.Clock.System (SystemTime)
import qualified Data.X509 as X
import Data.X509.Validation (Fingerprint (..))
@@ -115,7 +110,6 @@ import qualified Simplex.Messaging.Crypto as C
import Simplex.Messaging.Protocol
import Simplex.Messaging.Server.Expiration
import Simplex.Messaging.Server.Information
import Simplex.Messaging.Server.MsgStore.Journal
import Simplex.Messaging.Server.MsgStore.STM
import Simplex.Messaging.Server.MsgStore.Types
import Simplex.Messaging.Server.Names (NamesConfig (..), NamesEnv, newNamesEnv, pingEndpoint)
@@ -148,8 +142,6 @@ data ServerConfig s = ServerConfig
smpHandshakeTimeout :: Int,
tbqSize :: Natural,
msgQueueQuota :: Int,
maxJournalMsgCount :: Int,
maxJournalStateLines :: Int,
queueIdBytes :: Int,
msgIdBytes :: Int,
serverStoreCfg :: ServerStoreCfg s,
@@ -166,8 +158,6 @@ data ServerConfig s = ServerConfig
messageExpiration :: Maybe ExpirationConfig,
expireMessagesOnStart :: Bool,
expireMessagesOnSend :: Bool,
-- | interval of inactivity after which journal queue is closed
idleQueueInterval :: Int64,
-- | notification expiration interval (seconds)
notificationExpiration :: ExpirationConfig,
-- | time after which the socket with inactive client can be disconnected (without any messages or commands, incl. PING),
@@ -233,9 +223,6 @@ defaultMessageExpiration =
checkInterval = 7200 -- seconds, 2 hours
}
defaultIdleQueueInterval :: Int64
defaultIdleQueueInterval = 14400 -- seconds, 4 hours
defNtfExpirationHours :: Int64
defNtfExpirationHours = 24
@@ -265,21 +252,9 @@ defaultMaxClientCmdWait = 60000000 -- microseconds
defaultAuthErrorDelayMs :: Int64
defaultAuthErrorDelayMs = 50
journalMsgStoreDepth :: Int
journalMsgStoreDepth = 5
defaultMaxJournalStateLines :: Int
defaultMaxJournalStateLines = 16
defaultMaxJournalMsgCount :: Int
defaultMaxJournalMsgCount = 256
defaultMsgQueueQuota :: Int
defaultMsgQueueQuota = 128
defaultStateTailSize :: Int
defaultStateTailSize = 512
data Env s = Env
{ config :: ServerConfig s,
serverActive :: TVar Bool,
@@ -305,7 +280,6 @@ msgStore = fromMsgStore . msgStore_
fromMsgStore :: MsgStore s -> s
fromMsgStore = \case
StoreMemory s -> s
StoreJournal s -> s
#if defined(dbServerPostgres)
StoreDatabase s -> s
#endif
@@ -313,12 +287,10 @@ fromMsgStore = \case
type family SupportedStore (qs :: QSType) (ms :: MSType) :: Constraint where
SupportedStore 'QSMemory 'MSMemory = ()
SupportedStore 'QSMemory 'MSJournal = ()
SupportedStore 'QSMemory 'MSPostgres =
(Int ~ Bool, TypeError ('TE.Text "Storing messages in Postgres DB with queues in memory is not supported"))
SupportedStore 'QSPostgres 'MSMemory =
(Int ~ Bool, TypeError ('TE.Text "Storing messages in memory with queues in Postgres DB is not supported"))
SupportedStore 'QSPostgres 'MSJournal = ()
#if defined(dbServerPostgres)
SupportedStore 'QSPostgres 'MSPostgres = ()
#else
@@ -332,8 +304,6 @@ data AStoreType =
data ServerStoreCfg s where
SSCMemory :: Maybe StorePaths -> ServerStoreCfg STMMsgStore
SSCMemoryJournal :: {storeLogFile :: FilePath, storeMsgsPath :: FilePath} -> ServerStoreCfg (JournalMsgStore 'QSMemory)
SSCDatabaseJournal :: {storeCfg :: PostgresStoreCfg, storeMsgsPath' :: FilePath} -> ServerStoreCfg (JournalMsgStore 'QSPostgres)
#if defined(dbServerPostgres)
SSCDatabase :: PostgresStoreCfg -> ServerStoreCfg PostgresMsgStore
#endif
@@ -341,8 +311,6 @@ data ServerStoreCfg s where
dbStoreCfg :: ServerStoreCfg s -> Maybe PostgresStoreCfg
dbStoreCfg = \case
SSCMemory _ -> Nothing
SSCMemoryJournal {} -> Nothing
SSCDatabaseJournal {storeCfg} -> Just storeCfg
#if defined(dbServerPostgres)
SSCDatabase cfg -> Just cfg
#endif
@@ -350,8 +318,6 @@ dbStoreCfg = \case
storeLogFile' :: ServerStoreCfg s -> Maybe FilePath
storeLogFile' = \case
SSCMemory sp_ -> (\StorePaths {storeLogFile} -> storeLogFile) <$> sp_
SSCMemoryJournal {storeLogFile} -> Just storeLogFile
SSCDatabaseJournal {storeCfg = PostgresStoreCfg {dbStoreLogPath}} -> dbStoreLogPath
#if defined(dbServerPostgres)
SSCDatabase (PostgresStoreCfg {dbStoreLogPath}) -> dbStoreLogPath
#endif
@@ -360,14 +326,12 @@ data StorePaths = StorePaths {storeLogFile :: FilePath, storeMsgsFile :: Maybe F
type family MsgStoreType (qs :: QSType) (ms :: MSType) where
MsgStoreType 'QSMemory 'MSMemory = STMMsgStore
MsgStoreType qs 'MSJournal = JournalMsgStore qs
#if defined(dbServerPostgres)
MsgStoreType 'QSPostgres 'MSPostgres = PostgresMsgStore
#endif
data MsgStore s where
StoreMemory :: STMMsgStore -> MsgStore STMMsgStore
StoreJournal :: JournalMsgStore qs -> MsgStore (JournalMsgStore qs)
#if defined(dbServerPostgres)
StoreDatabase :: PostgresMsgStore -> MsgStore PostgresMsgStore
#endif
@@ -581,7 +545,7 @@ newProhibitedSub = do
return Sub {subThread = ProhibitSub, delivered}
newEnv :: ServerConfig s -> IO (Env s)
newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smpAgentCfg, information, messageExpiration, idleQueueInterval, msgQueueQuota, maxJournalMsgCount, maxJournalStateLines, namesConfig, msgIdBytes} = do
newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smpAgentCfg, information, messageExpiration, msgQueueQuota, namesConfig, msgIdBytes} = do
when (msgIdBytes < 24) $ do
logError "Error: msgIdBytes must be at least 24"
exitFailure
@@ -593,30 +557,12 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smp
ms <- newMsgStore STMStoreConfig {storePath, quota = msgQueueQuota}
forM_ storePaths_ $ \StorePaths {storeLogFile = f} -> loadStoreLog (mkQueue ms True) f $ queueStore ms
pure $ StoreMemory ms
SSCMemoryJournal {storeLogFile, storeMsgsPath} -> do
logWarn $
"Journal message store is deprecated and will be removed soon.\n"
<> "Please migrate to in-memory storage using `journal export` command.\n"
<> "After that you can migrate to PostgreSQL using `database import` command."
let qsCfg = MQStoreCfg
cfg = mkJournalStoreConfig qsCfg storeMsgsPath msgQueueQuota maxJournalMsgCount maxJournalStateLines idleQueueInterval
ms <- newMsgStore cfg
loadStoreLog (mkQueue ms True) storeLogFile $ stmQueueStore ms
pure $ StoreJournal ms
#if defined(dbServerPostgres)
SSCDatabaseJournal {storeCfg, storeMsgsPath'} -> do
let StartOptions {compactLog, confirmMigrations} = startOptions config
qsCfg = PQStoreCfg (storeCfg {confirmMigrations} :: PostgresStoreCfg)
cfg = mkJournalStoreConfig qsCfg storeMsgsPath' msgQueueQuota maxJournalMsgCount maxJournalStateLines idleQueueInterval
when compactLog $ compactDbStoreLog $ dbStoreLogPath storeCfg
StoreJournal <$> newMsgStore cfg
SSCDatabase storeCfg -> do
let StartOptions {compactLog, confirmMigrations} = startOptions config
cfg = PostgresMsgStoreCfg storeCfg {confirmMigrations} msgQueueQuota
when compactLog $ compactDbStoreLog $ dbStoreLogPath storeCfg
StoreDatabase <$> newMsgStore cfg
#else
SSCDatabaseJournal {} -> noPostgresExit
#endif
ntfStore <- NtfStore <$> TM.emptyIO
random <- C.newRandom
@@ -716,7 +662,9 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smp
Nothing -> SPMMemoryOnly
Just StorePaths {storeMsgsFile = Just _} -> SPMMessages
_ -> SPMQueues
#if defined(dbServerPostgres)
_ -> SPMMessages
#endif
noPostgresExit :: IO a
noPostgresExit = putStrLn noPostgresExitStr >> exitFailure
@@ -726,21 +674,6 @@ noPostgresExitStr =
"Error: server binary is compiled without support for PostgreSQL database.\n"
<> "Please download `smp-server-postgres` or re-compile with `cabal build -fserver_postgres`."
mkJournalStoreConfig :: QStoreCfg s -> FilePath -> Int -> Int -> Int -> Int64 -> JournalStoreConfig s
mkJournalStoreConfig queueStoreCfg storePath msgQueueQuota maxJournalMsgCount maxJournalStateLines idleQueueInterval =
JournalStoreConfig
{ storePath,
quota = msgQueueQuota,
pathParts = journalMsgStoreDepth,
queueStoreCfg,
maxMsgCount = maxJournalMsgCount,
maxStateLines = maxJournalStateLines,
stateTailSize = defaultStateTailSize,
idleInterval = idleQueueInterval,
expireBackupsAfter = 14 * nominalDay,
keepMinBackups = 2
}
newSMPProxyAgent :: SMPClientAgentConfig -> TVar ChaChaDRG -> IO ProxyAgent
newSMPProxyAgent smpAgentCfg random = do
smpAgent <- newSMPClientAgent SSender smpAgentCfg Nothing random
+47 -143
View File
@@ -28,7 +28,6 @@ module Simplex.Messaging.Server.Main
importMessagesToDatabase,
exportDatabaseToStoreLog,
#endif
newJournalMsgStore,
storeMsgsJournalDir',
getServerSourceCode,
simplexmqSource,
@@ -61,7 +60,6 @@ import qualified Data.Text.IO as T
import Options.Applicative
import Simplex.Messaging.Agent.Protocol (ConnectionLink (..), ConnectionMode (..), connReqUriP')
import Simplex.Messaging.Agent.Store.Postgres.Options (DBOpts (..))
import Simplex.Messaging.Agent.Store.Shared (MigrationConfirmation (..))
import Simplex.Messaging.Client (HostMode (..), NetworkConfig (..), ProtocolClientConfig (..), SMPWebPortServers (..), SocksMode (..), defaultNetworkConfig, textToHostMode)
import Simplex.Messaging.Client.Agent (SMPClientAgentConfig (..), defaultSMPClientAgentConfig)
import qualified Simplex.Messaging.Crypto as C
@@ -69,23 +67,20 @@ import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Parsers (parseAll)
import Network.URI (URI (..), URIAuth (..), parseAbsoluteURI)
import Simplex.Messaging.Protocol (BasicAuth (..), ProtoServerWithAuth (ProtoServerWithAuth), pattern SMPServer)
import Simplex.Messaging.Server (AttachHTTP, exportMessages, importMessages, printMessageStats, runSMPServer)
import Simplex.Messaging.Server (AttachHTTP, runSMPServer)
import Simplex.Messaging.Server.CLI
import Simplex.Messaging.Server.Env.STM
import Simplex.Messaging.Server.Expiration
import Simplex.Messaging.Server.Information
import Simplex.Messaging.Server.Main.Init
import Simplex.Messaging.Server.Web (EmbeddedWebParams (..), WebHttpsParams (..))
import Simplex.Messaging.Server.MsgStore.Journal (JournalMsgStore (..), QStoreCfg (..), stmQueueStore)
import Simplex.Messaging.Server.MsgStore.Types (MsgStoreClass (..), SQSType (..), SMSType (..), newMsgStore)
import Simplex.Messaging.Server.MsgStore.Types (SQSType (..), SMSType (..))
import Simplex.Messaging.Server.Names (NamesConfig (..), RpcAuth (..))
import Simplex.Messaging.Server.QueueStore.Postgres.Config
import Simplex.Messaging.Server.StoreLog.ReadWrite (readQueueStore)
import Simplex.Messaging.Transport (supportedProxyClientSMPRelayVRange, alpnSupportedSMPHandshakes, supportedServerSMPRelayVRange)
import Simplex.Messaging.Transport.Client (TransportHost (..), defaultSocksProxy)
import Simplex.Messaging.Transport.HTTP2 (httpALPN)
import Simplex.Messaging.Transport.Server (ServerCredentials (..), mkTransportServerConfig)
import Simplex.Messaging.Util (eitherToMaybe, ifM, safeDecodeUtf8)
import Simplex.Messaging.Util (eitherToMaybe, safeDecodeUtf8)
import System.Directory (createDirectoryIfMissing, doesDirectoryExist, doesFileExist)
import System.Exit (exitFailure)
import System.FilePath (combine)
@@ -98,15 +93,16 @@ import Data.Int (Int64)
import qualified Data.Map.Strict as M
import Data.Semigroup (Sum (..))
import Simplex.Messaging.Agent.Store.Postgres (checkSchemaExists)
import Simplex.Messaging.Server.MsgStore.Journal (JournalQueue)
import Simplex.Messaging.Server.MsgStore.Types (QSType (..))
import Simplex.Messaging.Server.MsgStore.Journal (postgresQueueStore)
import Simplex.Messaging.Agent.Store.Shared (MigrationConfirmation (..))
import Simplex.Messaging.Server.MsgStore.Postgres
import Simplex.Messaging.Server.QueueStore.Postgres (batchInsertQueues, batchInsertServices, foldQueueRecs, foldServiceRecs)
import Simplex.Messaging.Server.MsgStore.STM (STMQueue, STMStoreConfig (..))
import Simplex.Messaging.Server.MsgStore.Types (MsgStoreClass (..), newMsgStore)
import Simplex.Messaging.Server.QueueStore.Postgres.Config
import Simplex.Messaging.Server.QueueStore.Postgres (PostgresQueueStore, batchInsertQueues, batchInsertServices, foldQueueRecs, foldServiceRecs)
import Simplex.Messaging.Server.QueueStore.STM (STMQueueStore (..))
import Simplex.Messaging.Server.QueueStore.Types
import Simplex.Messaging.Server.StoreLog (closeStoreLog, logNewService, logCreateQueue, openWriteStoreLog)
import Simplex.Messaging.Util (unlessM)
import Simplex.Messaging.Util (ifM)
import System.Directory (renameFile)
import System.IO (IOMode (..), withFile)
#endif
@@ -136,79 +132,6 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
deleteDirIfExists cfgPath
deleteDirIfExists logPath
putStrLn "Deleted configuration and log files"
Journal cmd -> withIniFile $ \ini -> do
msgsDirExists <- doesDirectoryExist storeMsgsJournalDir
msgsFileExists <- doesFileExist storeMsgsFilePath
storeLogFile <- getRequiredStoreLogFile ini
case cmd of
SCImport
| msgsFileExists && msgsDirExists -> exitConfigureMsgStorage
| msgsDirExists -> do
putStrLn $ storeMsgsJournalDir <> " directory already exists."
exitFailure
| not msgsFileExists -> do
putStrLn $ storeMsgsFilePath <> " file does not exist."
exitFailure
| otherwise -> do
confirmOrExit
("WARNING: message log file " <> storeMsgsFilePath <> " will be imported to journal directory " <> storeMsgsJournalDir)
"Messages not imported"
ms <- newJournalMsgStore logPath MQStoreCfg
readQueueStore True (mkQueue ms False) storeLogFile $ stmQueueStore ms
msgStats <- importMessages True ms storeMsgsFilePath Nothing False -- no expiration
putStrLn "Import completed"
printMessageStats "Messages" msgStats
putStrLn $ case readStoreType ini of
Right (ASType SQSMemory SMSMemory) -> "store_messages set to `memory`, update it to `journal` in INI file"
Right (ASType SQSPostgres SMSPostgres) -> "store_messages set to `database`, update it to `journal` in INI file"
Right (ASType _ SMSJournal) -> "store_messages set to `journal`"
Left e -> e <> ", configure storage correctly"
SCExport
| msgsFileExists && msgsDirExists -> exitConfigureMsgStorage
| msgsFileExists -> do
putStrLn $ storeMsgsFilePath <> " file already exists."
exitFailure
| otherwise -> do
confirmOrExit
("WARNING: journal directory " <> storeMsgsJournalDir <> " will be exported to message log file " <> storeMsgsFilePath)
"Journal not exported"
case readStoreType ini of
Right (ASType SQSMemory msType) -> do
ms <- newJournalMsgStore logPath MQStoreCfg
readQueueStore True (mkQueue ms False) storeLogFile $ stmQueueStore ms
exportMessages True (StoreJournal ms) storeMsgsFilePath False
putStrLn "Export completed"
putStrLn $ case msType of
SMSMemory -> "store_messages set to `memory`, start the server."
SMSJournal -> "store_messages set to `journal`, update it to `memory` in INI file"
#if defined(dbServerPostgres)
Right (ASType SQSPostgres SMSJournal) -> do
let dbStoreLogPath = enableDbStoreLog' ini $> storeLogFilePath
dbOpts@DBOpts {connstr, schema} = iniDBOptions ini defaultDBOpts
unlessM (checkSchemaExists connstr schema) $ do
putStrLn $ "Schema " <> B.unpack schema <> " does not exist in PostrgreSQL database: " <> B.unpack connstr
exitFailure
ms <- newJournalMsgStore logPath $ PQStoreCfg PostgresStoreCfg {dbOpts, dbStoreLogPath, confirmMigrations = MCYesUp, deletedTTL = iniDeletedTTL ini}
exportMessages True (StoreJournal ms) storeMsgsFilePath False
putStrLn "Export completed"
putStrLn "store_messages set to `journal`, store_queues is set to `database`.\nExport queues to store log to use memory storage for messages (`smp-server database export`)."
Right (ASType SQSPostgres SMSPostgres) -> do
putStrLn $ "Messages can be exported with `dabatase export --table messages`."
exitFailure
#else
Right (ASType SQSPostgres SMSJournal) -> noPostgresExit
#endif
Left e -> putStrLn $ e <> ", configure storage correctly"
SCDelete
| not msgsDirExists -> do
putStrLn $ storeMsgsJournalDir <> " directory does not exists."
exitFailure
| otherwise -> do
confirmOrExit
("WARNING: journal directory " <> storeMsgsJournalDir <> " will be permanently deleted.\nTHIS CANNOT BE UNDONE!")
"Messages NOT deleted"
deleteDirIfExists storeMsgsJournalDir
putStrLn $ "Deleted all messages in journal " <> storeMsgsJournalDir
#if defined(dbServerPostgres)
Database cmd tables dbOpts@DBOpts {connstr, schema} -> withIniFile $ \ini -> do
schemaExists <- checkSchemaExists connstr schema
@@ -221,7 +144,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
confirmOrExit
("WARNING: store log file " <> storeLogFile <> " and message log file " <> storeMsgsFilePath <> " will be imported to PostrgreSQL database: " <> B.unpack connstr <> ", schema: " <> B.unpack schema)
"Store logs not imported"
(sCnt, qCnt) <- importStoreLogToDatabase logPath storeLogFile dbOpts
(sCnt, qCnt) <- importStoreLogToDatabase storeLogFile dbOpts
putStrLn $ "Imported: " <> show sCnt <> " services, " <> show qCnt <> " queues"
putStrLn "Importing messages..."
mCnt <- importMessagesToDatabase storeMsgsFilePath dbOpts
@@ -248,11 +171,10 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
confirmOrExit
("WARNING: store log file " <> storeLogFile <> " will be compacted and imported to PostrgreSQL database: " <> B.unpack connstr <> ", schema: " <> B.unpack schema)
"Queue records not imported"
(sCnt, qCnt) <- importStoreLogToDatabase logPath storeLogFile dbOpts
(sCnt, qCnt) <- importStoreLogToDatabase storeLogFile dbOpts
putStrLn $ "Import completed: " <> show sCnt <> " services, " <> show qCnt <> " queues"
putStrLn $ case readStoreType ini of
Right (ASType SQSMemory SMSMemory) -> setToDbStr <> "\nstore_messages set to `memory`, import messages to journal to use PostgreSQL database for queues (`smp-server journal import`)"
Right (ASType SQSMemory SMSJournal) -> setToDbStr
Right (ASType SQSMemory SMSMemory) -> setToDbStr <> "\nstore_messages set to `memory`, update it to `database` and import messages (`smp-server database import --table messages`)"
Right (ASType SQSPostgres _) -> "store_queues set to `database`, start the server."
Left e -> e <> ", configure storage correctly"
where
@@ -280,7 +202,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
confirmOrExit
("WARNING: PostrgreSQL schema " <> B.unpack schema <> " (database: " <> B.unpack connstr <> ") will be exported to store log file " <> storeLogFilePath <> " and to message log file " <> storeMsgsFilePath)
"Database store not exported"
(sCnt, qCnt) <- exportDatabaseToStoreLog logPath dbOpts storeLogFilePath
(sCnt, qCnt) <- exportDatabaseToStoreLog dbOpts storeLogFilePath
putStrLn $ "Exported: " <> show sCnt <> " services, " <> show qCnt <> " queues"
putStrLn "Exporting messages..."
let storeCfg = PostgresStoreCfg {dbOpts, dbStoreLogPath = Nothing, confirmMigrations = MCConsole, deletedTTL = 86400 * defaultDeletedTTL}
@@ -306,7 +228,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
confirmOrExit
("WARNING: PostrgreSQL schema " <> B.unpack schema <> " (database: " <> B.unpack connstr <> ") will be exported to store log file " <> storeLogFilePath)
"Queue records not exported"
(sCnt, qCnt) <- exportDatabaseToStoreLog logPath dbOpts storeLogFilePath
(sCnt, qCnt) <- exportDatabaseToStoreLog dbOpts storeLogFilePath
putStrLn $ "Export completed: " <> show sCnt <> " services, " <> show qCnt <> " queues"
putStrLn $ case readStoreType ini of
Right (ASType SQSPostgres _) -> "store_queues or store_messages set to `database`, update it to `memory` in INI file."
@@ -346,6 +268,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
doesFileExist iniFile >>= \case
True -> readIniFile iniFile >>= either exitError a
_ -> exitError $ "Error: server is not initialized (" <> iniFile <> " does not exist).\nRun `" <> executableName <> " init`."
#if defined(dbServerPostgres)
getRequiredStoreLogFile ini = do
case enableStoreLog' ini $> storeLogFilePath of
Just storeLogFile -> do
@@ -354,6 +277,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
(pure storeLogFile)
(putStrLn ("Store log file " <> storeLogFile <> " not found") >> exitFailure)
Nothing -> putStrLn "Store log disabled, see `[STORE_LOG] enable`" >> exitFailure
#endif
iniFile = combine cfgPath "smp-server.ini"
serverVersion = "SMP server v" <> simplexmqVersionCommit
executableName = "smp-server"
@@ -364,10 +288,10 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
readStoreType :: Ini -> Either String AStoreType
readStoreType ini = case (iniStoreQueues, iniStoreMessage) of
("memory", "memory") -> Right $ ASType SQSMemory SMSMemory
("memory", "journal") -> Right $ ASType SQSMemory SMSJournal
("memory", "journal") -> Left journalRemovedStr
("memory", "database") -> Left "Database and memory storage are not compatible."
("database", "memory") -> Left "Database and memory storage are not compatible."
("database", "journal") -> Right $ ASType SQSPostgres SMSJournal
("database", "journal") -> Left journalRemovedStr
#if defined(dbServerPostgres)
("database", "database") -> Right $ ASType SQSPostgres SMSPostgres
#else
@@ -377,10 +301,12 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
where
iniStoreQueues = fromRight "memory" $ lookupValue "STORE_LOG" "store_queues" ini
iniStoreMessage = fromRight "memory" $ lookupValue "STORE_LOG" "store_messages" ini
#if defined(dbServerPostgres)
iniDeletedTTL ini = readIniDefault (86400 * defaultDeletedTTL) "STORE_LOG" "db_deleted_ttl" ini
enableDbStoreLog' = settingIsOn "STORE_LOG" "db_store_log"
#endif
defaultStaticPath = combine logPath "www"
enableStoreLog' = settingIsOn "STORE_LOG" "enable"
enableDbStoreLog' = settingIsOn "STORE_LOG" "db_store_log"
initializeServer opts
| scripted opts = initialize opts
| otherwise = do
@@ -517,16 +443,13 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
iniStoreType = either error id $! readStoreType ini
iniStoreCfg :: SupportedStore qs ms => SQSType qs -> SMSType ms -> ServerStoreCfg (MsgStoreType qs ms)
iniStoreCfg SQSMemory SMSMemory = SSCMemory $ enableStoreLog' ini $> StorePaths {storeLogFile = storeLogFilePath, storeMsgsFile = restoreMessagesFile storeMsgsFilePath}
iniStoreCfg SQSMemory SMSJournal = SSCMemoryJournal {storeLogFile = storeLogFilePath, storeMsgsPath = storeMsgsJournalDir}
iniStoreCfg SQSPostgres SMSJournal =
let dbStoreLogPath = enableDbStoreLog' ini $> storeLogFilePath
storeCfg = PostgresStoreCfg {dbOpts = iniDBOptions ini defaultDBOpts, dbStoreLogPath, confirmMigrations = MCYesUp, deletedTTL = iniDeletedTTL ini}
in SSCDatabaseJournal {storeCfg, storeMsgsPath' = storeMsgsJournalDir}
#if defined(dbServerPostgres)
iniStoreCfg SQSPostgres SMSPostgres =
let dbStoreLogPath = enableDbStoreLog' ini $> storeLogFilePath
storeCfg = PostgresStoreCfg {dbOpts = iniDBOptions ini defaultDBOpts, dbStoreLogPath, confirmMigrations = MCYesUp, deletedTTL = iniDeletedTTL ini}
in SSCDatabase storeCfg
#else
iniStoreCfg SQSPostgres _ = error noPostgresExitStr
#endif
serverConfig :: ServerStoreCfg s -> ServerConfig s
serverConfig serverStoreCfg =
@@ -535,8 +458,6 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
smpHandshakeTimeout = 120000000,
tbqSize = 128,
msgQueueQuota = defaultMsgQueueQuota,
maxJournalMsgCount = defaultMaxJournalMsgCount,
maxJournalStateLines = defaultMaxJournalStateLines,
queueIdBytes = 24,
msgIdBytes = 24, -- must be at least 24 bytes, it is used as 192-bit nonce for XSalsa20
smpCredentials =
@@ -561,7 +482,6 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
},
expireMessagesOnStart = fromMaybe True $ iniOnOff "STORE_LOG" "expire_messages_on_start" ini,
expireMessagesOnSend = fromMaybe True $ iniOnOff "STORE_LOG" "expire_messages_on_send" ini,
idleQueueInterval = defaultIdleQueueInterval,
notificationExpiration =
defaultNtfExpiration
{ ttl = 3600 * readIniDefault defNtfExpirationHours "STORE_LOG" "expire_ntfs_hours" ini
@@ -639,7 +559,6 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
checkMsgStoreMode ini mode = do
msgsDirExists <- doesDirectoryExist storeMsgsJournalDir
msgsFileExists <- doesFileExist storeMsgsFilePath
storeLogExists <- doesFileExist storeLogFilePath
case mode of
#if defined(dbServerPostgres)
ASType SQSPostgres SMSPostgres
@@ -647,35 +566,22 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
putStrLn $ "Error: " <> storeMsgsFilePath <> " file or " <> storeMsgsJournalDir <> " directory are present."
putStrLn "Configure memory storage."
exitFailure
| otherwise -> checkDbStorage ini storeLogExists
#endif
ASType qs SMSJournal
| msgsFileExists && msgsDirExists -> exitConfigureMsgStorage
| msgsFileExists -> do
putStrLn $ "Error: store_messages is `journal` with " <> storeMsgsFilePath <> " file present."
putStrLn "Set store_messages to `memory` or use `smp-server journal export` to migrate."
exitFailure
| not msgsDirExists ->
putStrLn $ "store_messages is `journal`, " <> storeMsgsJournalDir <> " directory will be created."
| otherwise -> case qs of
SQSMemory ->
unless (storeLogExists) $ putStrLn $ "store_queues is `memory`, " <> storeLogFilePath <> " file will be created."
#if defined(dbServerPostgres)
SQSPostgres -> checkDbStorage ini storeLogExists
| otherwise -> checkDbStorage ini =<< doesFileExist storeLogFilePath
#else
SQSPostgres -> noPostgresExit
ASType SQSPostgres _ -> noPostgresExit
#endif
ASType SQSMemory SMSMemory
| msgsFileExists && msgsDirExists -> exitConfigureMsgStorage
| msgsDirExists -> do
putStrLn $ "Error: store_messages is `memory` with " <> storeMsgsJournalDir <> " directory present."
putStrLn "Set store_messages to `journal` or use `smp-server journal import` to migrate."
putStrLn journalRemovedStr
exitFailure
| otherwise -> pure ()
exitConfigureMsgStorage :: IO ()
exitConfigureMsgStorage = do
putStrLn $ "Error: both " <> storeMsgsFilePath <> " file and " <> storeMsgsJournalDir <> " directory are present."
putStrLn "Configure memory storage."
putStrLn journalRemovedStr
exitFailure
#if defined(dbServerPostgres)
@@ -707,19 +613,19 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
putStrLn "Configure queue storage."
exitFailure
importStoreLogToDatabase :: FilePath -> FilePath -> DBOpts -> IO (Int64, Int64)
importStoreLogToDatabase logPath storeLogFile dbOpts = do
ms <- newJournalMsgStore logPath MQStoreCfg
let st = stmQueueStore ms
importStoreLogToDatabase :: FilePath -> DBOpts -> IO (Int64, Int64)
importStoreLogToDatabase storeLogFile dbOpts = do
ms <- newMsgStore STMStoreConfig {storePath = Nothing, quota = defaultMsgQueueQuota}
let st = queueStore ms
sl <- readWriteQueueStore True (mkQueue ms False) storeLogFile st
closeStoreLog sl
queues <- readTVarIO $ loadedQueues st
services' <- M.elems <$> readTVarIO (services st)
let storeCfg = PostgresStoreCfg {dbOpts = dbOpts {createSchema = True}, dbStoreLogPath = Nothing, confirmMigrations = MCConsole, deletedTTL = 86400 * defaultDeletedTTL}
ps <- newJournalMsgStore logPath $ PQStoreCfg storeCfg
sCnt <- batchInsertServices services' $ postgresQueueStore ps
qCnt <- batchInsertQueues @(JournalQueue 'QSMemory) True queues $ postgresQueueStore ps
closeMsgStore ps
ps :: PostgresQueueStore PostgresQueue <- newQueueStore @PostgresQueue (storeCfg, False)
sCnt <- batchInsertServices services' ps
qCnt <- batchInsertQueues @STMQueue True queues ps
closeQueueStore @PostgresQueue ps
renameFile storeLogFile $ storeLogFile <> ".bak"
pure (sCnt, qCnt)
@@ -738,22 +644,23 @@ importMessagesToDatabase msgsLogFile dbOpts = do
renameFile msgsLogFile $ msgsLogFile <> ".bak"
pure mCnt'
exportDatabaseToStoreLog :: FilePath -> DBOpts -> FilePath -> IO (Int, Int)
exportDatabaseToStoreLog logPath dbOpts storeLogFilePath = do
exportDatabaseToStoreLog :: DBOpts -> FilePath -> IO (Int, Int)
exportDatabaseToStoreLog dbOpts storeLogFilePath = do
let storeCfg = PostgresStoreCfg {dbOpts, dbStoreLogPath = Nothing, confirmMigrations = MCConsole, deletedTTL = 86400 * defaultDeletedTTL}
ps <- newJournalMsgStore logPath $ PQStoreCfg storeCfg
ps :: PostgresQueueStore PostgresQueue <- newQueueStore @PostgresQueue (storeCfg, False)
sl <- openWriteStoreLog False storeLogFilePath
Sum sCnt <- foldServiceRecs (postgresQueueStore ps) $ \sr -> logNewService sl sr $> Sum (1 :: Int)
Sum qCnt <- foldQueueRecs True True (postgresQueueStore ps) $ \(rId, qr) -> logCreateQueue sl rId qr $> Sum (1 :: Int)
Sum sCnt <- foldServiceRecs ps $ \sr -> logNewService sl sr $> Sum (1 :: Int)
Sum qCnt <- foldQueueRecs True True ps $ \(rId, qr) -> logCreateQueue sl rId qr $> Sum (1 :: Int)
closeStoreLog sl
closeMsgStore ps
closeQueueStore @PostgresQueue ps
pure (sCnt, qCnt)
#endif
newJournalMsgStore :: FilePath -> QStoreCfg s -> IO (JournalMsgStore s)
newJournalMsgStore logPath qsCfg =
let cfg = mkJournalStoreConfig qsCfg (storeMsgsJournalDir' logPath) defaultMsgQueueQuota defaultMaxJournalMsgCount defaultMaxJournalStateLines $ checkInterval defaultMessageExpiration
in newMsgStore cfg
journalRemovedStr :: String
journalRemovedStr =
"Journal message storage is removed.\n"
<> "Export messages with `smp-server journal export` using smp-server v7.1.0.11 or earlier,\n"
<> "then set store_messages to `memory`, or to `database` and run `smp-server database import --table messages`."
storeMsgsJournalDir' :: FilePath -> FilePath
storeMsgsJournalDir' logPath = combine logPath "messages"
@@ -877,7 +784,6 @@ data CliCommand
| OnlineCert CertOptions
| Start StartOptions
| Delete
| Journal StoreCmd
| Database StoreCmd DatabaseTable DBOpts
data StoreCmd = SCImport | SCExport | SCDelete
@@ -903,7 +809,6 @@ cliCommandP cfgPath logPath iniFile =
<> command "cert" (info (OnlineCert <$> certOptionsP) (progDesc $ "Generate new online TLS server credentials (configuration: " <> iniFile <> ")"))
<> command "start" (info (Start <$> startOptionsP) (progDesc $ "Start server (configuration: " <> iniFile <> ")"))
<> command "delete" (info (pure Delete) (progDesc "Delete configuration and log files"))
<> command "journal" (info (Journal <$> journalCmdP) (progDesc "Import/export messages to/from journal storage"))
<> command "database" (info (Database <$> databaseCmdP <*> dbTableP <*> dbOptsP defaultDBOpts) (progDesc "Import/export queues to/from PostgreSQL database storage"))
)
where
@@ -1050,7 +955,6 @@ cliCommandP cfgPath logPath iniFile =
disableWeb,
scripted
}
journalCmdP = storeCmdP "message log file" "journal storage"
databaseCmdP = storeCmdP "queue store log file" "PostgreSQL database schema"
storeCmdP src dest =
hsubparser
+2 -2
View File
@@ -83,7 +83,7 @@ iniFileContent cfgPath logPath opts host basicAuth controlPortPwds =
<> ("enable = " <> onOff enableStoreLog <> "\n\n")
<> "# Queue storage mode: `memory` or `database` (to store queue records in PostgreSQL database).\n\
\# `memory` - in-memory persistence, with optional append-only log (`enable = on`).\n\
\# `database`- PostgreSQL databass (requires `store_messages = journal`).\n\
\# `database`- PostgreSQL databass (requires `store_messages = database`).\n\
\store_queues = memory\n\n\
\# Database connection settings for PostgreSQL database (`store_queues = database`).\n"
<> iniDbOpts dbOptions defaultDBOpts
@@ -91,7 +91,7 @@ iniFileContent cfgPath logPath opts host basicAuth controlPortPwds =
\# db_store_log = off\n\n\
\# Time to retain deleted queues in the database, days.\n"
<> ("# db_deleted_ttl = " <> tshow defaultDeletedTTL <> "\n\n")
<> "# Message storage mode: `memory` or `journal`.\n\
<> "# Message storage mode: `memory` or `database` (requires `store_queues = database`).\n\
\store_messages = memory\n\n\
\# When store_messages is `memory`, undelivered messages are optionally saved and restored\n\
\# when the server restarts, they are preserved in the .bak file until the next restart.\n"
File diff suppressed because it is too large. Load diff
@@ -1,44 +0,0 @@
module Simplex.Messaging.Server.MsgStore.Journal.SharedLock
( withLockWaitShared,
withLockMapWaitShared,
withSharedWaitLock,
)
where
import Control.Concurrent.STM
import qualified Control.Exception as E
import Control.Monad
import Data.Text (Text)
import Simplex.Messaging.Agent.Lock
import Simplex.Messaging.Agent.Client (getMapLock)
import Simplex.Messaging.Protocol (RecipientId)
import Simplex.Messaging.TMap (TMap)
import qualified Simplex.Messaging.TMap as TM
import Simplex.Messaging.Util (($>>), ($>>=))
-- wait until shared lock with passed ID is released and take lock
withLockWaitShared :: RecipientId -> Lock -> TMVar RecipientId -> Text -> IO a -> IO a
withLockWaitShared rId lock shared name =
E.bracket_
(atomically $ waitShared rId shared >> putTMVar lock name)
(void $ atomically $ takeTMVar lock)
-- wait until shared lock with passed ID is released and take lock from Map for this ID
withLockMapWaitShared :: RecipientId -> TMap RecipientId Lock -> TMVar RecipientId -> Text -> IO a -> IO a
withLockMapWaitShared rId locks shared name a =
E.bracket
(atomically $ waitShared rId shared >> getPutLock (getMapLock locks) rId name)
(atomically . takeTMVar)
(const a)
waitShared :: RecipientId -> TMVar RecipientId -> STM ()
waitShared rId shared = tryReadTMVar shared >>= mapM_ (\rId' -> when (rId == rId') retry)
-- wait until lock with passed ID in Map is released and take shared lock for this ID
withSharedWaitLock :: RecipientId -> TMap RecipientId Lock -> TMVar RecipientId -> IO a -> IO a
withSharedWaitLock rId locks shared =
E.bracket_
(atomically $ waitLock >> putTMVar shared rId)
(atomically $ takeTMVar shared)
where
waitLock = TM.lookup rId locks $>>= tryReadTMVar $>> retry
@@ -162,7 +162,7 @@ instance MsgStoreClass PostgresMsgStore where
loadedQueueCount <- M.size <$> readTVarIO queues
loadedNotifierCount <- M.size <$> readTVarIO notifiers
notifierLockCount <- M.size <$> readTVarIO notifierLocks
pure LoadedQueueCounts {loadedQueueCount, loadedNotifierCount, openJournalCount = 0, queueLockCount = 0, notifierLockCount}
pure LoadedQueueCounts {loadedQueueCount, loadedNotifierCount, queueLockCount = 0, notifierLockCount}
where
PostgresQueueStore {queues, notifiers, notifierLocks} = queueStore_ ms
+1 -1
View File
@@ -103,7 +103,7 @@ instance MsgStoreClass STMMsgStore where
loadedQueueCounts STMMsgStore {queueStore_ = st} = do
loadedQueueCount <- M.size <$> readTVarIO (queues st)
loadedNotifierCount <- M.size <$> readTVarIO (notifiers st)
pure LoadedQueueCounts {loadedQueueCount, loadedNotifierCount, openJournalCount = 0, queueLockCount = 0, notifierLockCount = 0}
pure LoadedQueueCounts {loadedQueueCount, loadedNotifierCount, queueLockCount = 0, notifierLockCount = 0}
mkQueue _ _ rId qr = STMQueue rId <$> newTVarIO (Just qr) <*> newTVarIO Nothing
{-# INLINE mkQueue #-}
@@ -124,13 +124,12 @@ class (Monad (StoreMonad s), QueueStoreClass (StoreQueue s) (QueueStore s)) => M
getQueueSize st q = withPeekMsgQueue st q "getQueueSize" $ maybe (pure 0) (getQueueSize_ . fst)
{-# INLINE getQueueSize #-}
data MSType = MSMemory | MSJournal | MSPostgres
data MSType = MSMemory | MSPostgres
data QSType = QSMemory | QSPostgres
data SMSType :: MSType -> Type where
SMSMemory :: SMSType 'MSMemory
SMSJournal :: SMSType 'MSJournal
SMSPostgres :: SMSType 'MSPostgres
data SQSType :: QSType -> Type where
@@ -155,7 +154,6 @@ instance Semigroup MessageStats where
data LoadedQueueCounts = LoadedQueueCounts
{ loadedQueueCount :: Int,
loadedNotifierCount :: Int,
openJournalCount :: Int,
queueLockCount :: Int,
notifierLockCount :: Int
}
+3 -7
View File
@@ -586,23 +586,19 @@ prometheusMetrics sm rtm ts =
\# TYPE simplex_smp_subscription_ntf_service_subs_total gauge\n\
\simplex_smp_subscription_ntf_service_subs_total " <> mshow (subServiceSubsCount ntfSubs) <> "\n# ntf.subServiceSubsCount\n\
\\n\
\# HELP simplex_smp_loaded_queues_queue_count Total loaded queues count (all queues for memory/journal storage)\n\
\# HELP simplex_smp_loaded_queues_queue_count Total loaded queues count (all queues for memory storage)\n\
\# TYPE simplex_smp_loaded_queues_queue_count gauge\n\
\simplex_smp_loaded_queues_queue_count " <> mshow (loadedQueueCount loadedCounts) <> "\n# loadedCounts.loadedQueueCount\n\
\\n\
\# HELP simplex_smp_loaded_queues_ntf_count Total loaded ntf credential references (all ntf credentials for memory/journal storage)\n\
\# HELP simplex_smp_loaded_queues_ntf_count Total loaded ntf credential references (all ntf credentials for memory storage)\n\
\# TYPE simplex_smp_loaded_queues_ntf_count gauge\n\
\simplex_smp_loaded_queues_ntf_count " <> mshow (loadedNotifierCount loadedCounts) <> "\n# loadedCounts.loadedNotifierCount\n\
\\n\
\# HELP simplex_smp_loaded_queues_open_journal_count Total opened queue journals (0 for memory storage)\n\
\# TYPE simplex_smp_loaded_queues_open_journal_count gauge\n\
\simplex_smp_loaded_queues_open_journal_count " <> mshow (openJournalCount loadedCounts) <> "\n# loadedCounts.openJournalCount\n\
\\n\
\# HELP simplex_smp_loaded_queues_queue_lock_count Total queue locks (0 for memory storage)\n\
\# TYPE simplex_smp_loaded_queues_queue_lock_count gauge\n\
\simplex_smp_loaded_queues_queue_lock_count " <> mshow (queueLockCount loadedCounts) <> "\n# loadedCounts.queueLockCount\n\
\\n\
\# HELP simplex_smp_loaded_queues_ntf_lock_count Total notifier locks (0 for memory/journal storage)\n\
\# HELP simplex_smp_loaded_queues_ntf_lock_count Total notifier locks (0 for memory storage)\n\
\# TYPE simplex_smp_loaded_queues_ntf_lock_count gauge\n\
\simplex_smp_loaded_queues_ntf_lock_count " <> mshow (notifierLockCount loadedCounts) <> "\n# loadedCounts.notifierLockCount\n"
+11 -20
View File
@@ -134,9 +134,7 @@ import Simplex.Messaging.Agent.Store.AgentStore (deleteClientService, getSubscri
import qualified Database.PostgreSQL.Simple as PSQL
import qualified Simplex.Messaging.Agent.Store.Postgres as Postgres
import qualified Simplex.Messaging.Agent.Store.Postgres.Common as Postgres
import Simplex.Messaging.Server.MsgStore.Journal (JournalQueue)
import Simplex.Messaging.Server.MsgStore.Postgres (PostgresQueue)
import Simplex.Messaging.Server.MsgStore.Types (QSType (..))
import Simplex.Messaging.Server.QueueStore.Postgres
import Simplex.Messaging.Server.QueueStore.Postgres.Migrations
import Simplex.Messaging.Server.QueueStore.Types (QueueStoreClass (..))
@@ -1606,7 +1604,7 @@ testAllowConnectionClientRestart ps@(t, ASType qsType _) = do
bob <- getSMPAgentClient' 2 agentCfg initAgentServersSrv2 testDB2
withSmpServerStoreLogOn ps testPort $ \_ -> do
(aliceId, bobId, confId) <-
withSmpServerConfigOn t (cfgJ2QS qsType) testPort2 $ \_ -> do
withSmpServerConfigOn t (cfgS2QS qsType) testPort2 $ \_ -> do
runRight $ do
(bobId, qInfo) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
(aliceId, sqSecured) <- joinConnection bob 1 True qInfo "bob's connInfo" SMSubscribe
@@ -1628,7 +1626,7 @@ testAllowConnectionClientRestart ps@(t, ASType qsType _) = do
alice2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
runRight_ $ subscribeConnection alice2 bobId
threadDelay 500000
withSmpServerConfigOn t (cfgJ2QS qsType) testPort2 $ \_ -> do
withSmpServerConfigOn t (cfgS2QS qsType) testPort2 $ \_ -> do
runRight $ do
("", "", UP _ _) <- nGet bob
get alice2 ##> ("", bobId, CON)
@@ -1795,7 +1793,7 @@ withServer1 :: HasTestEnv => (ASrvTransport, AStoreType) -> IO a -> IO a
withServer1 ps = withSmpServerStoreLogOn ps testPort . const
withServer2 :: HasTestEnv => (ASrvTransport, AStoreType) -> IO a -> IO a
withServer2 (t, ASType qsType _) = withSmpServerConfigOn t (cfgJ2QS qsType) testPort2 . const
withServer2 (t, ASType qsType _) = withSmpServerConfigOn t (cfgS2QS qsType) testPort2 . const
testInvitationShortLink :: (HasCallStack, HasTestEnv) => Bool -> AgentClient -> AgentClient -> IO ()
testInvitationShortLink viaProxy a b =
@@ -2035,18 +2033,11 @@ testOldContactQueueShortLink ps@(_, msType) = withAgentClients2 $ \a b -> do
#endif
() <- case testServerStoreConfig msType of
ASSCfg _ _ (SSCMemory sp_) -> mapM_ (\StorePaths {storeLogFile} -> updateStoreLog storeLogFile) sp_
ASSCfg _ _ SSCMemoryJournal {storeLogFile} -> updateStoreLog storeLogFile
#if defined(dbServerPostgres)
ASSCfg _ _ SSCDatabaseJournal {storeCfg} -> do
st :: PostgresQueueStore (JournalQueue 'QSPostgres) <- newQueueStore @(JournalQueue 'QSPostgres) (storeCfg, True)
updateDbStore st
closeQueueStore @(JournalQueue 'QSPostgres) st
ASSCfg _ _ (SSCDatabase storeCfg) -> do
st :: PostgresQueueStore PostgresQueue <- newQueueStore @PostgresQueue (storeCfg, False)
updateDbStore st
closeQueueStore @PostgresQueue st
#else
ASSCfg _ _ SSCDatabaseJournal {} -> error "no dbServerPostgres flag"
#endif
withSmpServer ps $ do
@@ -2552,7 +2543,7 @@ testExpireMessageQuota (t, msType) = withSmpServerConfigOn t cfg' testPort $ \_
ackMessage b' aId 4 Nothing
disposeAgentClient a
where
cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {msgQueueQuota = 1, maxJournalMsgCount = 2}
cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {msgQueueQuota = 1}
testExpireManyMessagesQuota :: HasTestEnv => (ASrvTransport, AStoreType) -> IO ()
testExpireManyMessagesQuota (t, msType) = withSmpServerConfigOn t cfg' testPort $ \_ -> do
@@ -2591,7 +2582,7 @@ testExpireManyMessagesQuota (t, msType) = withSmpServerConfigOn t cfg' testPort
ackMessage b' aId 4 Nothing
disposeAgentClient a
where
cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {msgQueueQuota = 1, maxJournalMsgCount = 2}
cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {msgQueueQuota = 1}
testJoinFullContactAsync :: (HasCallStack, HasTestEnv) => (ASrvTransport, AStoreType) -> IO ()
testJoinFullContactAsync (t, msType) = withSmpServerConfigOn t cfg' testPort $ \_ -> do
@@ -2610,7 +2601,7 @@ testJoinFullContactAsync (t, msType) = withSmpServerConfigOn t cfg' testPort $ \
("", _, A.REQ _ _ _ "bob's connInfo" _ _) <- get alice
pure ()
where
cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {msgQueueQuota = 1, maxJournalMsgCount = 2}
cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {msgQueueQuota = 1}
testJoinFullContactAsyncExpire :: (HasCallStack, HasTestEnv) => (ASrvTransport, AStoreType) -> IO ()
testJoinFullContactAsyncExpire (t, msType) = withSmpServerConfigOn t cfg' testPort $ \_ -> do
@@ -2624,7 +2615,7 @@ testJoinFullContactAsyncExpire (t, msType) = withSmpServerConfigOn t cfg' testPo
noMessages bob "joining should be retried until quota exceeded timeout"
get bob =##> \case ("2", c, ERR (SMP _ QUOTA)) -> c == aliceId; _ -> False
where
cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {msgQueueQuota = 1, maxJournalMsgCount = 2}
cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {msgQueueQuota = 1}
fillContactAddress :: (HasCallStack, HasTestEnv) => IO (ConnId, ConnectionRequestUri 'CMContact)
fillContactAddress = do
@@ -3202,7 +3193,7 @@ testBatchedSubscriptions nCreate nDel ps@(t, ASType qsType _) = do
runServers :: ExceptT AgentErrorType IO a -> IO a
runServers a = do
withSmpServerStoreLogOn ps testPort $ \t1 -> do
res <- withSmpServerConfigOn t (cfgJ2QS qsType) testPort2 $ \t2 ->
res <- withSmpServerConfigOn t (cfgS2QS qsType) testPort2 $ \t2 ->
runRight a `finally` killThread t2
killThread t1
pure res
@@ -3709,7 +3700,7 @@ testJoinConnectionAsyncReplyError ps@(t, ASType qsType _) = do
ConnectionStats {rcvQueuesInfo = [], sndQueuesInfo = [SndQueueInfo {}]} <- getConnectionServers b aId
pure (aId, bId)
nGet a =##> \case ("", "", DOWN _ [c]) -> c == bId; _ -> False
withSmpServerConfigOn t (cfgJ2QS qsType) testPort2 $ \_ -> do
withSmpServerConfigOn t (cfgS2QS qsType) testPort2 $ \_ -> do
confId <- withSmpServerStoreLogOn ps testPort $ \_ -> do
-- both servers need to be online for connection to progress because of SKEY
get b =##> \case ("2", c, JOINED sqSecured) -> c == aId && sqSecured; _ -> False
@@ -3842,7 +3833,7 @@ fastSwitchComplete a bId b aId = do
testFastSwitchDeadOldServer :: (HasCallStack, HasTestEnv) => (ASrvTransport, AStoreType) -> IO ()
testFastSwitchDeadOldServer ps@(t, ASType qsType _) = do
let bServers = initAgentServers {smp = userServers [testSMPServer2]}
withSmpServerConfigOn t (cfgJ2QS qsType) testPort2 $ \_ ->
withSmpServerConfigOn t (cfgS2QS qsType) testPort2 $ \_ ->
withAgent 1 aCfg initAgentServers testDB $ \a ->
withAgent 2 agentCfg bServers testDB2 $ \b -> do
(aId, bId) <- withSmpServerStoreLogOn ps testPort $ \_ -> runRight $ do
@@ -4402,7 +4393,7 @@ testDeliveryReceiptsConcurrent (t, msType) =
liftIO $ noMessages a "nothing else should be delivered to alice"
liftIO $ noMessages b "nothing else should be delivered to bob"
where
cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {msgQueueQuota = 256, maxJournalMsgCount = 512}
cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {msgQueueQuota = 256}
runClient :: String -> AgentClient -> ConnId -> IO ()
runClient _cName client connId = do
concurrently_ send receive
+1 -1
View File
@@ -877,7 +877,7 @@ testNotificationsSMPRestartBatch n ps@(t, ASType qsType _) apns =
runServers :: ExceptT AgentErrorType IO a -> IO a
runServers a = do
withSmpServerStoreLogOn ps testPort $ \t1 -> do
res <- withSmpServerConfigOn t (cfgJ2QS qsType) testPort2 $ \t2 ->
res <- withSmpServerConfigOn t (cfgS2QS qsType) testPort2 $ \t2 ->
runRight a `finally` killThread t2
killThread t1
pure res
+5 -428
View File
@@ -16,8 +16,7 @@
module CoreTests.MsgStoreTests where
import AgentTests.FunctionalAPITests (runRight, runRight_)
import Control.Concurrent (threadDelay)
import AgentTests.FunctionalAPITests (runRight_)
import Control.Concurrent.STM
import Control.Exception (bracket)
import Control.Monad
@@ -25,33 +24,23 @@ import Control.Monad.IO.Class
import Control.Monad.Trans.Except
import Crypto.Random (ChaChaDRG)
import Data.ByteString.Char8 (ByteString)
import qualified Data.ByteString.Char8 as B
import Data.Int (Int64)
import Data.List (isPrefixOf, isSuffixOf)
import Data.Maybe (fromJust)
import Data.Time.Clock (addUTCTime)
import Data.Time.Clock.System (SystemTime (..), getSystemTime)
import SMPClient (testStoreLogFile, testStoreMsgsDir, testStoreMsgsDir2, testStoreMsgsFile, testStoreMsgsFile2)
import Data.Time.Clock.System (getSystemTime)
import Simplex.Messaging.Crypto (pattern MaxLenBS)
import qualified Simplex.Messaging.Crypto as C
import Simplex.Messaging.Protocol (EncDataBytes (..), EntityId (..), ErrorType (..), LinkId, Message (..), QueueLinkData, RecipientId, SParty (..), noMsgFlags, randomMsgId)
import Simplex.Messaging.Server (exportMessages, importMessages, printMessageStats)
import Simplex.Messaging.Server.Env.STM (MsgStore (..), journalMsgStoreDepth, readWriteQueueStore)
import Simplex.Messaging.Server.Expiration (ExpirationConfig (..), expireBeforeEpoch)
import Simplex.Messaging.Server.MsgStore.Journal
import Simplex.Messaging.Server.MsgStore.STM
import Simplex.Messaging.Server.MsgStore.Types
import Simplex.Messaging.Server.QueueStore
import Simplex.Messaging.Server.QueueStore.QueueInfo
import Simplex.Messaging.Server.QueueStore.Types
import Simplex.Messaging.Server.StoreLog (closeStoreLog, logCreateQueue)
import System.Directory (copyFile, createDirectoryIfMissing, listDirectory, removeFile, renameFile)
import System.FilePath ((</>))
import System.IO (IOMode (..), withFile)
import Test.Hspec hiding (fit, it)
import Util
#if defined(dbServerPostgres)
import AgentTests.FunctionalAPITests (runRight)
import Control.Concurrent (threadDelay)
import Data.Time.Clock.System (SystemTime (..))
import Database.PostgreSQL.Simple (Only (..))
import qualified Database.PostgreSQL.Simple as DB
import Simplex.Messaging.Agent.Store.Postgres.Common
@@ -64,35 +53,14 @@ import SMPClient (testStoreDBOpts)
msgStoreTests :: Spec
msgStoreTests = do
describe "STM message store" $ someMsgStoreTests testSMTStoreConfig
describe "Journal message store" $ do
someMsgStoreTests $ testJournalStoreCfg MQStoreCfg
journalMsgStoreTests $ testJournalStoreCfg MQStoreCfg
it "should export and import journal store" $ withMsgStore (testJournalStoreCfg MQStoreCfg) testExportImportStore
#if defined(dbServerPostgres)
describe "Postgres+journal message store" $ do
someMsgStoreTests $ testJournalStoreCfg $ PQStoreCfg testPostgresStoreCfg
journalMsgStoreTests $ testJournalStoreCfg $ PQStoreCfg testPostgresStoreCfg
describe "Postgres-only message store" $ do
someMsgStoreTests testPostgresStoreConfig
it "should correctly update message counts and canWrite flag" $ withMsgStore testPostgresStoreConfig testUpdateMessageCounts
it "tryDelPeekMsg (ACK not from NSE) should reset message counts when queue is empty" $ withMsgStore testPostgresStoreConfig testResetMessageCounts
it "should expire messages across commit batches" $ withMsgStore testPostgresStoreConfig testExpireMessagesInBatches
#endif
describe "Journal message store: queue state backup expiration" $ do
it "should remove old queue state backups" testRemoveQueueStateBackups
it "should expire messages in idle queues" testExpireIdleQueues
where
journalMsgStoreTests :: (HasTestEnv => JournalStoreConfig s) -> Spec
journalMsgStoreTests cfg = do
describe "queue state" $ do
it "should restore queue state from the last line" $ withMsgStore cfg testQueueState
it "should recover when message is written and state is not" $ withMsgStore cfg testMessageState
it "should remove journal files when queue is empty" $ withMsgStore cfg testRemoveJournals
describe "missing files" $ do
it "should create read file when missing" $ withMsgStore cfg testReadFileMissing
it "should switch to write file when read file missing" $ withMsgStore cfg testReadFileMissingSwitch
it "should create write file when missing" $ withMsgStore cfg testWriteFileMissing
it "should create read file when read and write files are missing" $ withMsgStore cfg testReadAndWriteFilesMissing
someMsgStoreTests :: MsgStoreClass s => (HasTestEnv => MsgStoreConfig s) -> Spec
someMsgStoreTests cfg = do
it "should get queue and store/read messages" $ withMsgStore cfg testGetQueue
@@ -107,21 +75,6 @@ withMsgStore cfg = bracket (newMsgStore cfg) closeMsgStore
testSMTStoreConfig :: STMStoreConfig
testSMTStoreConfig = STMStoreConfig {storePath = Nothing, quota = 3}
testJournalStoreCfg :: HasTestEnv => QStoreCfg s -> JournalStoreConfig s
testJournalStoreCfg queueStoreCfg =
JournalStoreConfig
{ storePath = testStoreMsgsDir,
pathParts = journalMsgStoreDepth,
queueStoreCfg,
quota = 3,
maxMsgCount = 4,
maxStateLines = 2,
stateTailSize = 256,
idleInterval = 21600,
expireBackupsAfter = 0,
keepMinBackups = 1
}
#if defined(dbServerPostgres)
testPostgresStoreConfig :: HasTestEnv => PostgresMsgStoreCfg
testPostgresStoreConfig =
@@ -150,12 +103,6 @@ mkMessage body = liftIO $ do
pattern Msg :: ByteString -> Maybe Message
pattern Msg s <- Just Message {msgBody = MaxLenBS s}
deriving instance Eq MsgQueueState
deriving instance Eq (JournalState t)
deriving instance Eq (SJournalType t)
testNewQueueRec :: TVar ChaChaDRG -> QueueMode -> IO (RecipientId, QueueRec)
testNewQueueRec g qm = testNewQueueRecData g qm Nothing
@@ -292,58 +239,6 @@ testLinkDataSecuredQueue ms = do
void $ ExceptT $ deleteQueue ms q
void $ ExceptT $ deleteQueue ms cq
testExportImportStore :: HasTestEnv => JournalMsgStore 'QSMemory -> IO ()
testExportImportStore ms = do
g <- C.newRandom
(rId1, qr1) <- testNewQueueRec g QMMessaging
(rId2, qr2) <- testNewQueueRec g QMMessaging
sl <- readWriteQueueStore True (mkQueue ms True) testStoreLogFile $ stmQueueStore ms
runRight_ $ do
let write q s = writeMsg ms q True =<< mkMessage s
q1 <- ExceptT $ addQueue ms rId1 qr1
liftIO $ logCreateQueue sl rId1 qr1
Just (Message {}, True) <- write q1 "message 1"
Just (Message {}, False) <- write q1 "message 2"
q2 <- ExceptT $ addQueue ms rId2 qr2
liftIO $ logCreateQueue sl rId2 qr2
Just (Message {msgId = mId3}, True) <- write q2 "message 3"
Just (Message {msgId = mId4}, False) <- write q2 "message 4"
(Msg "message 3", Msg "message 4") <- tryDelPeekMsg ms q2 mId3
(Msg "message 4", Nothing) <- tryDelPeekMsg ms q2 mId4
Just (Message {}, True) <- write q2 "message 5"
Just (Message {}, False) <- write q2 "message 6"
Just (Message {}, False) <- write q2 "message 7"
Nothing <- write q2 "message 8"
pure ()
length <$> listDirectory (msgQueueDirectory ms rId1) `shouldReturn` 2
length <$> listDirectory (msgQueueDirectory ms rId2) `shouldReturn` 3
exportMessages False (StoreJournal ms) testStoreMsgsFile False
closeMsgStore ms
closeStoreLog sl
-- export with closed queues and compare
ms2 <- newMsgStore $ testJournalStoreCfg MQStoreCfg
readWriteQueueStore True (mkQueue ms2 True) testStoreLogFile (stmQueueStore ms2) >>= closeStoreLog
exportMessages False (StoreJournal ms2) (testStoreMsgsFile <> ".copy") False
s <- B.readFile testStoreMsgsFile
B.readFile (testStoreMsgsFile <> ".copy") `shouldReturn` s
let cfg = (testJournalStoreCfg MQStoreCfg :: JournalStoreConfig 'QSMemory) {storePath = testStoreMsgsDir2}
ms' <- newMsgStore cfg
readWriteQueueStore True (mkQueue ms' True) testStoreLogFile (stmQueueStore ms') >>= closeStoreLog
stats@MessageStats {storedMsgsCount = 5, expiredMsgsCount = 0, storedQueues = 2} <-
importMessages False ms' testStoreMsgsFile Nothing False
printMessageStats "Messages" stats
length <$> listDirectory (msgQueueDirectory ms rId1) `shouldReturn` 2
length <$> listDirectory (msgQueueDirectory ms rId2) `shouldReturn` 3 -- 2 message files
exportMessages False (StoreJournal ms') testStoreMsgsFile2 False
(B.readFile testStoreMsgsFile2 `shouldReturn`) =<< B.readFile (testStoreMsgsFile <> ".bak")
stmStore <- newMsgStore testSMTStoreConfig
readWriteQueueStore True (mkQueue stmStore True) testStoreLogFile (queueStore stmStore) >>= closeStoreLog
MessageStats {storedMsgsCount = 5, expiredMsgsCount = 0, storedQueues = 2} <-
importMessages False stmStore testStoreMsgsFile2 Nothing False
exportMessages False (StoreMemory stmStore) testStoreMsgsFile False
(B.sort <$> B.readFile testStoreMsgsFile `shouldReturn`) =<< (B.sort <$> B.readFile (testStoreMsgsFile2 <> ".bak"))
#if defined(dbServerPostgres)
testUpdateMessageCounts :: PostgresMsgStore -> IO ()
testUpdateMessageCounts ms = do
@@ -452,321 +347,3 @@ testExpireMessagesInBatches ms = do
where
go n = write q "fill" >>= maybe (pure n) (const $ go (n + 1))
#endif
testQueueState :: JournalMsgStore s -> IO ()
testQueueState ms = do
g <- C.newRandom
rId <- EntityId <$> atomically (C.randomBytes 24 g)
let dir = msgQueueDirectory ms rId
statePath = msgQueueStatePath dir rId
createDirectoryIfMissing True dir
state <- newMsgQueueState <$> newJournalId (random ms)
withFile statePath WriteMode (`appendState` state)
length . lines <$> readFile statePath `shouldReturn` 1
readQueueState ms statePath `shouldReturn` (Just state, False)
length <$> listDirectory dir `shouldReturn` 1 -- no backup
let state1 =
state
{ size = 1,
readState = (readState state) {msgCount = 1, byteCount = 100},
writeState = (writeState state) {msgPos = 1, msgCount = 1, bytePos = 100, byteCount = 100}
}
withFile statePath AppendMode (`appendState` state1)
length . lines <$> readFile statePath `shouldReturn` 2
readQueueState ms statePath `shouldReturn` (Just state1, False)
length <$> listDirectory dir `shouldReturn` 1 -- no backup
let state2 =
state
{ size = 2,
readState = (readState state) {msgCount = 2, byteCount = 200},
writeState = (writeState state) {msgPos = 2, msgCount = 2, bytePos = 200, byteCount = 200}
}
withFile statePath AppendMode (`appendState` state2)
length . lines <$> readFile statePath `shouldReturn` 3
copyFile statePath (statePath <> ".2")
readQueueState ms statePath `shouldReturn` (Just state2, True)
length <$> listDirectory dir `shouldReturn` 2 -- new state + copy
ls <- lines <$> readFile statePath
length ls `shouldBe` 3
-- mock compacting file
writeFile statePath $ last ls
-- corrupt the only line
corruptFile statePath
(Nothing, True) <- readQueueState ms statePath
-- corrupt the last line
renameFile (statePath <> ".2") statePath
removeOtherFiles dir statePath
length . lines <$> readFile statePath `shouldReturn` 3
corruptFile statePath
readQueueState ms statePath `shouldReturn` (Just state1, True)
length <$> listDirectory dir `shouldReturn` 1
length . lines <$> readFile statePath `shouldReturn` 3
where
corruptFile f = do
s <- readFile f
removeFile f
writeFile f $ take (length s - 4) s
removeOtherFiles dir keep = do
names <- listDirectory dir
forM_ names $ \name ->
let f = dir </> name
in unless (f == keep) $ removeFile f
testMessageState :: JournalMsgStore s -> IO ()
testMessageState ms = do
g <- C.newRandom
(rId, qr) <- testNewQueueRec g QMMessaging
let dir = msgQueueDirectory ms rId
statePath = msgQueueStatePath dir rId
write q s = writeMsg ms q True =<< mkMessage s
mId1 <- runRight $ do
q <- ExceptT $ addQueue ms rId qr
Just (Message {msgId = mId1}, True) <- write q "message 1"
Just (Message {}, False) <- write q "message 2"
liftIO $ closeMsgQueue ms q
pure mId1
ls <- B.lines <$> B.readFile statePath
B.writeFile statePath $ B.unlines $ take (length ls - 1) ls
runRight_ $ do
q <- ExceptT $ getQueue ms SRecipient rId
Just (Message {msgId = mId3}, False) <- write q "message 3"
(Msg "message 1", Msg "message 3") <- tryDelPeekMsg ms q mId1
(Msg "message 3", Nothing) <- tryDelPeekMsg ms q mId3
liftIO $ closeMsgQueue ms q
testRemoveJournals :: JournalMsgStore s -> IO ()
testRemoveJournals ms = do
g <- C.newRandom
(rId, qr) <- testNewQueueRec g QMMessaging
let dir = msgQueueDirectory ms rId
statePath = msgQueueStatePath dir rId
write q s = writeMsg ms q True =<< mkMessage s
runRight $ do
q <- ExceptT $ addQueue ms rId qr
Just (Message {msgId = mId1}, True) <- write q "message 1"
Just (Message {msgId = mId2}, False) <- write q "message 2"
(Msg "message 1", Msg "message 2") <- tryDelPeekMsg ms q mId1
(Msg "message 2", Nothing) <- tryDelPeekMsg ms q mId2
liftIO $ closeMsgQueue ms q
ls <- B.lines <$> B.readFile statePath
length ls `shouldBe` 4
journalFilesCount dir `shouldReturn` 1
stateBackupCount dir `shouldReturn` 0
runRight $ do
q <- ExceptT $ getQueue ms SRecipient rId
-- not removed yet
liftIO $ journalFilesCount dir `shouldReturn` 1
liftIO $ stateBackupCount dir `shouldReturn` 0
Nothing <- tryPeekMsg ms q
-- still not removed, queue is empty and not opened
liftIO $ journalFilesCount dir `shouldReturn` 1
_mq <- isolateQueue ms q "test" $ getMsgQueue ms q False
-- journal is removed
liftIO $ journalFilesCount dir `shouldReturn` 0
liftIO $ stateBackupCount dir `shouldReturn` 1
Just (Message {msgId = mId3}, True) <- write q "message 3"
-- journal is created
liftIO $ journalFilesCount dir `shouldReturn` 1
Just (Message {msgId = mId4}, False) <- write q "message 4"
(Msg "message 3", Msg "message 4") <- tryDelPeekMsg ms q mId3
(Msg "message 4", Nothing) <- tryDelPeekMsg ms q mId4
Just (Message {msgId = mId5}, True) <- write q "message 5"
Just (Message {msgId = mId6}, False) <- write q "message 6"
liftIO $ journalFilesCount dir `shouldReturn` 1
Just (Message {msgId = mId7}, False) <- write q "message 7"
-- separate write journal is created
liftIO $ journalFilesCount dir `shouldReturn` 2
Nothing <- write q "message 8"
(Msg "message 5", Msg "message 6") <- tryDelPeekMsg ms q mId5
liftIO $ journalFilesCount dir `shouldReturn` 2
(Msg "message 6", Msg "message 7") <- tryDelPeekMsg ms q mId6
-- read journal is removed
liftIO $ journalFilesCount dir `shouldReturn` 1
(Msg "message 7", Just MessageQuota {msgId = mId8}) <- tryDelPeekMsg ms q mId7
(Just MessageQuota {}, Nothing) <- tryDelPeekMsg ms q mId8
liftIO $ closeMsgQueue ms q
journalFilesCount dir `shouldReturn` 1
runRight $ do
q <- ExceptT $ getQueue ms SRecipient rId
Just (Message {}, True) <- write q "message 8"
liftIO $ journalFilesCount dir `shouldReturn` 1
liftIO $ stateBackupCount dir `shouldReturn` 2
liftIO $ closeMsgQueue ms q
where
journalFilesCount dir = length . filter ("messages." `isPrefixOf`) <$> listDirectory dir
stateBackupCount dir = length . filter (".bak" `isSuffixOf`) <$> listDirectory dir
testRemoveQueueStateBackups :: HasTestEnv => IO ()
testRemoveQueueStateBackups = do
g <- C.newRandom
(rId, qr) <- testNewQueueRec g QMMessaging
ms' <- newMsgStore (testJournalStoreCfg MQStoreCfg) {maxStateLines = 1, expireBackupsAfter = 0, keepMinBackups = 0}
-- set expiration time 1 second ahead
let ms = ms' {expireBackupsBefore = addUTCTime 1 $ expireBackupsBefore ms'}
let dir = msgQueueDirectory ms rId
write q s = writeMsg ms q True =<< mkMessage s
runRight $ do
q <- ExceptT $ addQueue ms rId qr
Just (Message {msgId = mId1}, True) <- write q "message 1"
Just (Message {msgId = mId2}, False) <- write q "message 2"
(Msg "message 1", Msg "message 2") <- tryDelPeekMsg ms q mId1
(Msg "message 2", Nothing) <- tryDelPeekMsg ms q mId2
liftIO $ closeMsgQueue ms q
liftIO $ stateBackupCount dir `shouldReturn` 0
q1 <- ExceptT $ getQueue ms SRecipient rId
Just (Message {}, True) <- write q1 "message 3"
Just (Message {}, False) <- write q1 "message 4"
liftIO $ closeMsgQueue ms q1
liftIO $ stateBackupCount dir `shouldReturn` 0
liftIO $ threadDelay 1000000
q2 <- ExceptT $ getQueue ms SRecipient rId
Just (Message {}, False) <- write q2 "message 5"
Nothing <- write q2 "message 5"
liftIO $ closeMsgQueue ms q2
liftIO $ stateBackupCount dir `shouldReturn` 1
where
stateBackupCount dir = length . filter (".bak" `isSuffixOf`) <$> listDirectory dir
testExpireIdleQueues :: HasTestEnv => IO ()
testExpireIdleQueues = do
g <- C.newRandom
(rId, qr) <- testNewQueueRec g QMMessaging
ms <- newMsgStore (testJournalStoreCfg MQStoreCfg) {idleInterval = 0}
let dir = msgQueueDirectory ms rId
statePath = msgQueueStatePath dir rId
write q s = writeMsg ms q True =<< mkMessage s
q <- runRight $ do
q <- ExceptT $ addQueue ms rId qr
Just (Message {msgId = mId1}, True) <- write q "message 1"
Just (Message {msgId = mId2}, False) <- write q "message 2"
(Msg "message 1", Msg "message 2") <- tryDelPeekMsg ms q mId1
(Msg "message 2", Nothing) <- tryDelPeekMsg ms q mId2
liftIO $ closeMsgQueue ms q
pure q
(Just MsgQueueState {size = 0, readState = rs, writeState = ws}, True) <- readQueueState ms statePath
msgCount rs `shouldBe` 2
msgCount ws `shouldBe` 2
old <- expireBeforeEpoch ExpirationConfig {ttl = 1, checkInterval = 1} -- no old messages
now <- systemSeconds <$> getSystemTime
(expired_, stored) <- runRight $ isolateQueue ms q "" $ withIdleMsgQueue now ms q $ deleteExpireMsgs_ old q
expired_ `shouldBe` Just 0
stored `shouldBe` 0
(Nothing, False) <- readQueueState ms statePath
pure ()
testReadFileMissing :: JournalMsgStore s -> IO ()
testReadFileMissing ms = do
g <- C.newRandom
(rId, qr) <- testNewQueueRec g QMMessaging
let write q s = writeMsg ms q True =<< mkMessage s
q <- runRight $ do
q <- ExceptT $ addQueue ms rId qr
Just (Message {}, True) <- write q "message 1"
Msg "message 1" <- tryPeekMsg ms q
pure q
mq <- fromJust <$> readTVarIO (msgQueue' q)
MsgQueueState {readState = rs} <- readTVarIO $ state mq
closeMsgQueue ms q
let path = journalFilePath (queueDirectory $ queue mq) $ journalId rs
removeFile path
runRight_ $ do
q' <- ExceptT $ getQueue ms SRecipient rId
Nothing <- tryPeekMsg ms q'
Just (Message {}, True) <- write q' "message 2"
Msg "message 2" <- tryPeekMsg ms q'
pure ()
testReadFileMissingSwitch :: JournalMsgStore s -> IO ()
testReadFileMissingSwitch ms = do
g <- C.newRandom
(rId, qr) <- testNewQueueRec g QMMessaging
q <- writeMessages ms rId qr
mq <- fromJust <$> readTVarIO (msgQueue' q)
MsgQueueState {readState = rs} <- readTVarIO $ state mq
closeMsgQueue ms q
let path = journalFilePath (queueDirectory $ queue mq) $ journalId rs
removeFile path
runRight_ $ do
q' <- ExceptT $ getQueue ms SRecipient rId
Just (Message {}, False) <- writeMsg ms q' True =<< mkMessage "message 6"
Msg "message 5" <- tryPeekMsg ms q'
pure ()
testWriteFileMissing :: JournalMsgStore s -> IO ()
testWriteFileMissing ms = do
g <- C.newRandom
(rId, qr) <- testNewQueueRec g QMMessaging
q <- writeMessages ms rId qr
mq <- fromJust <$> readTVarIO (msgQueue' q)
MsgQueueState {writeState = ws} <- readTVarIO $ state mq
closeMsgQueue ms q
let path = journalFilePath (queueDirectory $ queue mq) $ journalId ws
print path
removeFile path
runRight_ $ do
q' <- ExceptT $ getQueue ms SRecipient rId
Just Message {msgId = mId3} <- tryPeekMsg ms q'
(Msg "message 3", Msg "message 4") <- tryDelPeekMsg ms q' mId3
Just Message {msgId = mId4} <- tryPeekMsg ms q'
(Msg "message 4", Nothing) <- tryDelPeekMsg ms q' mId4
Just (Message {}, True) <- writeMsg ms q' True =<< mkMessage "message 6"
Msg "message 6" <- tryPeekMsg ms q'
pure ()
testReadAndWriteFilesMissing :: JournalMsgStore s -> IO ()
testReadAndWriteFilesMissing ms = do
g <- C.newRandom
(rId, qr) <- testNewQueueRec g QMMessaging
q <- writeMessages ms rId qr
mq <- fromJust <$> readTVarIO (msgQueue' q)
MsgQueueState {readState = rs, writeState = ws} <- readTVarIO $ state mq
closeMsgQueue ms q
removeFile $ journalFilePath (queueDirectory $ queue mq) $ journalId rs
removeFile $ journalFilePath (queueDirectory $ queue mq) $ journalId ws
runRight_ $ do
q' <- ExceptT $ getQueue ms SRecipient rId
Nothing <- tryPeekMsg ms q'
Just (Message {}, True) <- writeMsg ms q' True =<< mkMessage "message 6"
Msg "message 6" <- tryPeekMsg ms q'
pure ()
writeMessages :: JournalMsgStore s -> RecipientId -> QueueRec -> IO (JournalQueue s)
writeMessages ms rId qr = runRight $ do
q <- ExceptT $ addQueue ms rId qr
let write s = writeMsg ms q True =<< mkMessage s
Just (Message {msgId = mId1}, True) <- write "message 1"
Just (Message {msgId = mId2}, False) <- write "message 2"
Just (Message {}, False) <- write "message 3"
(Msg "message 1", Msg "message 2") <- tryDelPeekMsg ms q mId1
(Msg "message 2", Msg "message 3") <- tryDelPeekMsg ms q mId2
Just (Message {}, False) <- write "message 4"
Just (Message {}, False) <- write "message 5"
pure q
+7 -7
View File
@@ -28,7 +28,7 @@ import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Protocol
import Simplex.Messaging.Protocol.Types (ClientNotice (..))
import Simplex.Messaging.Server.Env.STM (readWriteQueueStore)
import Simplex.Messaging.Server.MsgStore.Journal
import Simplex.Messaging.Server.MsgStore.STM (STMMsgStore (..))
import Simplex.Messaging.Server.MsgStore.Types
import Simplex.Messaging.Server.QueueStore
import Simplex.Messaging.Server.QueueStore.STM (STMQueueStore (..))
@@ -177,10 +177,10 @@ testSMPStoreLog testSuite tests =
closeStoreLog l
replicateM_ 3 $ testReadWrite t
#if defined(dbServerPostgres)
(sCnt, qCnt) <- importStoreLogToDatabase testDir testStoreLogFile testStoreDBOpts
(sCnt, qCnt) <- importStoreLogToDatabase testStoreLogFile testStoreDBOpts
fromIntegral (sCnt + qCnt) `shouldBe` length (compacted t)
imported <- B.readFile $ testStoreLogFile <> ".bak"
(sCnt', qCnt') <- exportDatabaseToStoreLog testDir testStoreDBOpts testStoreLogFile
(sCnt', qCnt') <- exportDatabaseToStoreLog testStoreDBOpts testStoreLogFile
sCnt' `shouldBe` fromIntegral sCnt
qCnt' `shouldBe` fromIntegral qCnt
exported <- B.readFile testStoreLogFile
@@ -188,14 +188,14 @@ testSMPStoreLog testSuite tests =
#endif
where
testReadWrite SLTC {compacted, state} = do
st <- newMsgStore $ testJournalStoreCfg MQStoreCfg
l <- readWriteQueueStore True (mkQueue st True) testStoreLogFile $ stmQueueStore st
st <- newMsgStore testSMTStoreConfig
l <- readWriteQueueStore True (mkQueue st True) testStoreLogFile $ queueStore st
storeState st `shouldReturn` state
closeStoreLog l
([], compacted') <- partitionEithers . map strDecode . B.lines <$> B.readFile testStoreLogFile
compacted' `shouldBe` compacted
storeState :: JournalMsgStore 'QSMemory -> IO (M.Map RecipientId QueueRec)
storeState st = M.mapMaybe id <$> (readTVarIO (queues $ stmQueueStore st) >>= mapM (readTVarIO . queueRec))
storeState :: STMMsgStore -> IO (M.Map RecipientId QueueRec)
storeState st = M.mapMaybe id <$> (readTVarIO (queues $ queueStore st) >>= mapM (readTVarIO . queueRec))
type FileRecState = (FileInfo, RoundedFileTime, Maybe RoundedFileTime, ServerEntityStatus)
+3
View File
@@ -1,3 +1,4 @@
{-# LANGUAGE CPP #-}
{-# LANGUAGE DuplicateRecordFields #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE LambdaCase #-}
@@ -78,7 +79,9 @@ memCfg2 = case memCfg of
newStoreCfg :: ServerStoreCfg s -> ServerStoreCfg s
newStoreCfg = \case
SSCMemory _ -> SSCMemory (Just StorePaths {storeLogFile = testStoreLogFile2, storeMsgsFile = Just testStoreMsgsFile2})
#if defined(dbServerPostgres)
other -> other
#endif
withNames :: Int -> AServerConfig -> AServerConfig
withNames port c = updateCfg c $ \cfg_ -> cfg_ {namesConfig = Just (testNamesConfig port)}
+42 -38
View File
@@ -33,7 +33,6 @@ 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
@@ -51,6 +50,7 @@ import Util
#if defined(dbServerPostgres)
import Database.PostgreSQL.Simple (defaultConnectInfo)
import Simplex.Messaging.Server.QueueStore.Postgres.Config (PostgresStoreCfg (..))
#endif
#if defined(dbPostgres) || defined(dbServerPostgres)
@@ -126,12 +126,6 @@ testStoreMsgsFile = testPath "smp-server-messages.log"
testStoreMsgsFile2 :: HasTestEnv => FilePath
testStoreMsgsFile2 = testPath "smp-server-messages.log.2"
testStoreMsgsDir :: HasTestEnv => FilePath
testStoreMsgsDir = testPath "messages"
testStoreMsgsDir2 :: HasTestEnv => FilePath
testStoreMsgsDir2 = testPath "messages.2"
testStoreNtfsFile :: HasTestEnv => FilePath
testStoreNtfsFile = testPath "smp-server-ntfs.log"
@@ -215,24 +209,35 @@ ntfTestServerCredentials =
}
cfg :: HasTestEnv => AServerConfig
cfg = cfgMS (ASType SQSMemory SMSJournal)
cfg = cfgMS (ASType SQSMemory SMSMemory)
cfgJ2 :: HasTestEnv => AServerConfig
cfgJ2 = journalCfg cfg testStoreLogFile2 testStoreMsgsDir2
cfgS2 :: HasTestEnv => AServerConfig
cfgS2 = memoryCfg cfg testStoreLogFile2 testStoreMsgsFile2
cfgJ2QS :: HasTestEnv => SQSType s -> AServerConfig
cfgJ2QS = \case
SQSMemory -> journalCfg (cfgMS $ ASType SQSMemory SMSJournal) testStoreLogFile2 testStoreMsgsDir2
SQSPostgres -> journalCfgDB (cfgMS $ ASType SQSPostgres SMSJournal) testStoreDBOpts2 testStoreMsgsDir2
cfgS2QS :: HasTestEnv => SQSType s -> AServerConfig
cfgS2QS = \case
SQSMemory -> memoryCfg (cfgMS $ ASType SQSMemory SMSMemory) testStoreLogFile2 testStoreMsgsFile2
SQSPostgres -> databaseCfg (cfgMS postgresStoreType) testStoreDBOpts2
journalCfg :: AServerConfig -> FilePath -> FilePath -> AServerConfig
journalCfg (ASrvCfg _ _ cfg') storeLogFile storeMsgsPath =
ASrvCfg SQSMemory SMSJournal cfg' {serverStoreCfg = SSCMemoryJournal {storeLogFile, storeMsgsPath}}
memoryCfg :: AServerConfig -> FilePath -> FilePath -> AServerConfig
memoryCfg (ASrvCfg _ _ cfg') storeLogFile storeMsgsFile =
ASrvCfg SQSMemory SMSMemory cfg' {serverStoreCfg = SSCMemory $ Just StorePaths {storeLogFile, storeMsgsFile = Just storeMsgsFile}}
journalCfgDB :: AServerConfig -> DBOpts -> FilePath -> AServerConfig
journalCfgDB (ASrvCfg _ _ cfg') dbOpts storeMsgsPath' =
databaseCfg :: AServerConfig -> DBOpts -> AServerConfig
#if defined(dbServerPostgres)
databaseCfg (ASrvCfg _ _ cfg') dbOpts =
let storeCfg = PostgresStoreCfg {dbOpts, dbStoreLogPath = Nothing, confirmMigrations = MCYesUp, deletedTTL = 86400}
in ASrvCfg SQSPostgres SMSJournal cfg' {serverStoreCfg = SSCDatabaseJournal {storeCfg, storeMsgsPath'}}
in ASrvCfg SQSPostgres SMSPostgres cfg' {serverStoreCfg = SSCDatabase storeCfg}
#else
databaseCfg _ _ = error "no dbServerPostgres flag"
#endif
postgresStoreType :: AStoreType
#if defined(dbServerPostgres)
postgresStoreType = ASType SQSPostgres SMSPostgres
#else
postgresStoreType = error "no dbServerPostgres flag"
#endif
cfgMS :: HasTestEnv => AStoreType -> AServerConfig
cfgMS msType = withStoreCfg (testServerStoreConfig msType) $ \serverStoreCfg ->
@@ -241,8 +246,6 @@ cfgMS msType = withStoreCfg (testServerStoreConfig msType) $ \serverStoreCfg ->
smpHandshakeTimeout = 60000000,
tbqSize = 4,
msgQueueQuota = 4,
maxJournalMsgCount = 5,
maxJournalStateLines = 2,
queueIdBytes = 24,
msgIdBytes = 24,
serverStoreCfg,
@@ -255,7 +258,6 @@ cfgMS msType = withStoreCfg (testServerStoreConfig msType) $ \serverStoreCfg ->
messageExpiration = Just defaultMessageExpiration,
expireMessagesOnStart = True,
expireMessagesOnSend = False,
idleQueueInterval = defaultIdleQueueInterval,
notificationExpiration = defaultNtfExpiration,
inactiveClientExpiration = Just defaultInactiveClientExpiration,
logStatsInterval = Nothing,
@@ -297,20 +299,22 @@ testServerStoreConfig :: HasTestEnv => AStoreType -> AServerStoreCfg
testServerStoreConfig = serverStoreConfig_ False
serverStoreConfig_ :: HasTestEnv => Bool -> AStoreType -> AServerStoreCfg
#if defined(dbServerPostgres)
serverStoreConfig_ useDbStoreLog = \case
#else
serverStoreConfig_ _ = \case
#endif
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}
#else
ASType SQSPostgres _ -> error "no dbServerPostgres flag"
#endif
cfgVPrev :: HasTestEnv => AStoreType -> AServerConfig
cfgVPrev msType = updateCfg (cfgMS msType) $ \cfg' -> cfg' {smpServerVRange = prevRange $ smpServerVRange cfg'}
@@ -325,7 +329,7 @@ nextVersion :: Version v -> Version v
nextVersion (Version v) = Version (v + 1)
proxyCfg :: HasTestEnv => AServerConfig
proxyCfg = proxyCfgMS (ASType SQSMemory SMSJournal)
proxyCfg = proxyCfgMS (ASType SQSMemory SMSMemory)
proxyCfgMS :: HasTestEnv => AStoreType -> AServerConfig
proxyCfgMS msType =
@@ -336,13 +340,13 @@ proxyCfgMS msType =
smpAgentCfg = smpAgentCfg' {smpCfg = (smpCfg smpAgentCfg') {agreeSecret = True, proxyServer = True, serverVRange = supportedProxyClientSMPRelayVRange}}
}
proxyCfgJ2 :: HasTestEnv => AServerConfig
proxyCfgJ2 = journalCfg proxyCfg testStoreLogFile2 testStoreMsgsDir2
proxyCfgS2 :: HasTestEnv => AServerConfig
proxyCfgS2 = memoryCfg proxyCfg testStoreLogFile2 testStoreMsgsFile2
proxyCfgJ2QS :: HasTestEnv => SQSType qs -> AServerConfig
proxyCfgJ2QS = \case
SQSMemory -> journalCfg (proxyCfgMS $ ASType SQSMemory SMSJournal) testStoreLogFile2 testStoreMsgsDir2
SQSPostgres -> journalCfgDB (proxyCfgMS $ ASType SQSPostgres SMSJournal) testStoreDBOpts2 testStoreMsgsDir2
proxyCfgS2QS :: HasTestEnv => SQSType qs -> AServerConfig
proxyCfgS2QS = \case
SQSMemory -> memoryCfg (proxyCfgMS $ ASType SQSMemory SMSMemory) testStoreLogFile2 testStoreMsgsFile2
SQSPostgres -> databaseCfg (proxyCfgMS postgresStoreType) testStoreDBOpts2
-- Proxy config with a short relay-connection timeout, to bound how long a failing
-- proxy->relay connection attempt blocks in the relay reconnection tests.
@@ -430,10 +434,10 @@ withSmpServerProxy :: (HasCallStack, HasTestEnv) => (ASrvTransport, AStoreType)
withSmpServerProxy (t, msType) = withSmpServerConfigOn t (proxyCfgMS msType) testPort . const
withSmpServers2 :: (HasCallStack, HasTestEnv) => (ASrvTransport, AStoreType) -> IO a -> IO a
withSmpServers2 ps@(t, ASType qs _ms) = withSmpServer ps . withSmpServerConfigOn t (cfgJ2QS qs) testPort2 . const
withSmpServers2 ps@(t, ASType qs _ms) = withSmpServer ps . withSmpServerConfigOn t (cfgS2QS qs) testPort2 . const
withSmpServersProxy2 :: (HasCallStack, HasTestEnv) => (ASrvTransport, AStoreType) -> IO a -> IO a
withSmpServersProxy2 ps@(t, ASType qs _ms) = withSmpServerProxy ps . withSmpServerConfigOn t (proxyCfgJ2QS qs) testPort2 . const
withSmpServersProxy2 ps@(t, ASType qs _ms) = withSmpServerProxy ps . withSmpServerConfigOn t (proxyCfgS2QS qs) testPort2 . const
runSmpTest :: forall c a. (HasCallStack, HasTestEnv, Transport c) => AStoreType -> (HasCallStack => THandleSMP c 'TClient -> IO a) -> IO a
runSmpTest msType test = withSmpServerConfigOn (transport @c) (cfgMS msType) testPort $ \_ -> testSMPClient test
@@ -454,7 +458,7 @@ smpServerTest ::
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
smpServerTest _ t = runSmpTest (ASType SQSMemory SMSMemory) $ \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
+14 -14
View File
@@ -154,9 +154,9 @@ smpProxyTests = do
it "25 agents, 300 pairs, 17 messages" $ oneServer $ withNumCapabilities 4 $ deliver 25 17
where
twoServers test msType = twoServers_ (proxyCfgMS msType) (proxyCfgMS msType) test msType
twoServersFirstProxy test msType = twoServers_ (proxyCfgMS msType) (updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {msgQueueQuota = 128, maxJournalMsgCount = 256}) test msType
twoServersMoreConc test msType = twoServers_ (updateCfg (proxyCfgMS msType) $ \cfg_ -> cfg_ {serverClientConcurrency = 128}) (updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {msgQueueQuota = 128, maxJournalMsgCount = 256}) test msType
twoServersNoConc test msType = twoServers_ (updateCfg (proxyCfgMS msType) $ \cfg_ -> cfg_ {serverClientConcurrency = 1}) (updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {msgQueueQuota = 128, maxJournalMsgCount = 256}) test msType
twoServersFirstProxy test msType = twoServers_ (proxyCfgMS msType) (updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {msgQueueQuota = 128}) test msType
twoServersMoreConc test msType = twoServers_ (updateCfg (proxyCfgMS msType) $ \cfg_ -> cfg_ {serverClientConcurrency = 128}) (updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {msgQueueQuota = 128}) test msType
twoServersNoConc test msType = twoServers_ (updateCfg (proxyCfgMS msType) $ \cfg_ -> cfg_ {serverClientConcurrency = 1}) (updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {msgQueueQuota = 128}) test msType
twoServersPrevProxy test msType = twoServers_ (proxyCfgVPrev msType) (cfgMS msType) test msType
twoServersPrevRelay test msType = twoServers_ (proxyCfgMS msType) (prevServerVRange $ cfgMS msType) test msType
prevServerVRange cfg' = updateCfg cfg' $ \cfg_ -> cfg_ {smpServerVRange = prevRange $ smpServerVRange cfg_}
@@ -164,8 +164,8 @@ smpProxyTests = do
twoServers_ cfg1 cfg2 runTest (ASType qsType _) =
withSmpServerConfigOn (transport @TLS) cfg1 testPort $ \_ ->
let cfg2' = case qsType of
SQSMemory -> journalCfg cfg2 testStoreLogFile2 testStoreMsgsDir2
SQSPostgres -> journalCfgDB cfg2 testStoreDBOpts2 testStoreMsgsDir2
SQSMemory -> memoryCfg cfg2 testStoreLogFile2 testStoreMsgsFile2
SQSPostgres -> databaseCfg cfg2 testStoreDBOpts2
in withSmpServerConfigOn (transport @TLS) cfg2' testPort2 $ const runTest
smpProxyCapabilityTests :: SpecWith AStoreType
@@ -175,7 +175,7 @@ smpProxyCapabilityTests = do
it "5 agents, 10 pairs, 50 messages. N4" $ oneServer $ withNumCapabilities 4 $ deliver 5 50
oneServer :: HasTestEnv => IO a -> AStoreType -> IO a
oneServer test msType = withSmpServerConfigOn (transport @TLS) (updateCfg (proxyCfgMS msType) $ \cfg_ -> cfg_ {msgQueueQuota = 128, maxJournalMsgCount = 256}) testPort $ const test
oneServer test msType = withSmpServerConfigOn (transport @TLS) (updateCfg (proxyCfgMS msType) $ \cfg_ -> cfg_ {msgQueueQuota = 128}) testPort $ const test
deliverMessageViaProxy :: (C.AlgorithmI a, C.AuthAlgorithm a) => SMPServer -> SMPServer -> C.SAlgorithm a -> ByteString -> ByteString -> IO ()
deliverMessageViaProxy proxyServ relayServ alg msg msg' = deliverMessagesViaProxy proxyServ relayServ alg [msg] [msg']
@@ -405,15 +405,15 @@ agentViaProxyRetryOffline = do
ackMessage alice bobId (baseId + 4) Nothing
where
withServer :: (ThreadId -> IO a) -> IO a
withServer = withServer_ testStoreLogFile testStoreMsgsDir testStoreNtfsFile testPort
withServer = withServer_ testStoreLogFile testStoreMsgsFile testStoreNtfsFile testPort
-- TODO [postgres]
-- withServer = withServer_ testStoreDBOpts testStoreMsgsDir testStoreNtfsFile testPort
withServer2 :: (ThreadId -> IO a) -> IO a
withServer2 = withServer_ testStoreLogFile2 testStoreMsgsDir2 testStoreNtfsFile2 testPort2
withServer2 = withServer_ testStoreLogFile2 testStoreMsgsFile2 testStoreNtfsFile2 testPort2
-- TODO [postgres]
-- withServer2 = withServer_ testStoreDBOpts2 testStoreMsgsDir2 testStoreNtfsFile2 testPort2
withServer_ storeLog storeMsgs storeNtfs =
let cfg' = updateCfg (journalCfg proxyCfg storeLog storeMsgs) $ \cfg_ -> cfg_ {storeNtfsFile = Just storeNtfs}
let cfg' = updateCfg (memoryCfg proxyCfg storeLog storeMsgs) $ \cfg_ -> cfg_ {storeNtfsFile = Just storeNtfs}
in withSmpServerConfigOn (transport @TLS) cfg'
a `up` cId = nGet a =##> \case ("", "", UP _ [c]) -> c == cId; _ -> False
a `down` cId = nGet a =##> \case ("", "", DOWN _ [c]) -> c == cId; _ -> False
@@ -439,7 +439,7 @@ agentViaProxyRetryNoSession = do
_ <- runRight $ makeConnection b a
pure ()
where
withServer2 = withSmpServerConfigOn (transport @TLS) proxyCfgJ2 testPort2
withServer2 = withSmpServerConfigOn (transport @TLS) proxyCfgS2 testPort2
servers srv = initAgentServersProxy {smp = userServers [srv]}
testNoProxy :: HasTestEnv => AStoreType -> IO ()
@@ -483,7 +483,7 @@ testChangedFwdVersion =
testFwdCorrIdSize :: HasTestEnv => IO ()
testFwdCorrIdSize =
withSmpServerConfigOn (transport @TLS) proxyCfg testPort $ \_ ->
withSmpServerConfigOn (transport @TLS) cfgJ2 testPort2 $ \_ ->
withSmpServerConfigOn (transport @TLS) cfgS2 testPort2 $ \_ ->
testSMPClient_ "localhost" testPort supportedServerSMPRelayVRange Nothing $ \(th :: THandleSMP TLS 'TClient) -> do
(_, _, Right (SMP.PKEY sessId _ _)) <- sendRecv th (Nothing, "1", NoEntity, SMP.PRXY testSMPServer2 Nothing)
g <- C.newRandom
@@ -495,7 +495,7 @@ testFwdCorrIdSize =
testForwardedAuthErrorDelay :: HasTestEnv => IO ()
testForwardedAuthErrorDelay =
withSmpServerConfigOn (transport @TLS) proxyCfg testPort $ \_ ->
withSmpServerConfigOn (transport @TLS) (updateCfg cfgJ2 $ \cfg_ -> cfg_ {authErrorDelay = 500000}) testPort2 $ \_ -> do
withSmpServerConfigOn (transport @TLS) (updateCfg cfgS2 $ \cfg_ -> cfg_ {authErrorDelay = 500000}) testPort2 $ \_ -> do
g <- C.newRandom
ts <- getCurrentTime
pc <- either (fail . show) pure =<< getProtocolClient g NRMInteractive (1, testSMPServer, Nothing) defaultSMPClientConfig [] Nothing ts (\_ -> pure ())
@@ -510,7 +510,7 @@ testForwardedAuthErrorDelay =
-- let any stored connection error expire, then require the proxy to establish the session (PKEY).
requireProxyReconnect :: HasTestEnv => IO ()
requireProxyReconnect =
withSmpServerConfigOn (transport @TLS) proxyCfgJ2 testPort2 $ \_ -> do
withSmpServerConfigOn (transport @TLS) proxyCfgS2 testPort2 $ \_ -> do
testSMPClient_ "127.0.0.1" testPort2 supportedServerSMPRelayVRange Nothing $ \(th :: THandleSMP TLS 'TClient) -> do
(_, _, reply) <- sendRecv th (Nothing, "0", NoEntity, SMP.PING)
reply `shouldBe` Right SMP.PONG
@@ -566,7 +566,7 @@ testAgentClientReconnectAfterCancel =
t <- async $ runExceptT $ A.createConnection a NRMInteractive 1 True True SCMInvitation Nothing Nothing CR.IKPQOn False SMSubscribe
threadDelay 1000000 -- let the connect to the stalling relay start, then kill it mid-flight
cancel t
withSmpServerConfigOn (transport @TLS) cfgJ2 testPort2 $ \_ -> do
withSmpServerConfigOn (transport @TLS) cfgS2 testPort2 $ \_ -> do
testSMPClient_ "127.0.0.1" testPort2 supportedServerSMPRelayVRange Nothing $ \(th :: THandleSMP TLS 'TClient) -> do
(_, _, reply) <- sendRecv th (Nothing, "0", NoEntity, SMP.PING)
reply `shouldBe` Right SMP.PONG -- the relay is up and reachable, so a timeout can only be the poisoned var
+14 -22
View File
@@ -24,7 +24,6 @@ import Control.Concurrent.STM
import Control.Exception (SomeException, throwIO, try)
import Control.Monad
import Control.Monad.IO.Class
import CoreTests.MsgStoreTests (testJournalStoreCfg)
import Data.Bifunctor (first)
import qualified Data.ByteString.Base64 as B64
import Data.ByteString.Char8 (ByteString)
@@ -47,17 +46,14 @@ import Simplex.Messaging.Encoding
import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Parsers (parseAll, parseString)
import Simplex.Messaging.Protocol
import Simplex.Messaging.Server (exportMessages)
import Simplex.Messaging.Server.Env.STM (AStoreType (..), MsgStore (..), ServerConfig (..), ServerStoreCfg (..), readWriteQueueStore)
import Simplex.Messaging.Server.Env.STM (AStoreType (..), ServerConfig (..), ServerStoreCfg (..))
import Simplex.Messaging.Server.Expiration
import Simplex.Messaging.Server.MsgStore.Journal (JournalStoreConfig (..), QStoreCfg (..), stmQueueStore)
import Simplex.Messaging.Server.MsgStore.Types (MsgStoreClass (..), QSType (..), SMSType (..), SQSType (..), newMsgStore)
import Simplex.Messaging.Server.MsgStore.Types (SMSType (..), SQSType (..))
import Simplex.Messaging.Server.Stats (PeriodStatsData (..), ServerStatsData (..))
import Simplex.Messaging.Server.StoreLog (StoreLogRecord (..), closeStoreLog)
import Simplex.Messaging.Server.StoreLog (StoreLogRecord (..))
import Simplex.Messaging.Transport
import Simplex.Messaging.Transport.Credentials
import Simplex.Messaging.Util (whenM)
import System.Directory (doesDirectoryExist, doesFileExist, removeDirectoryRecursive, removeFile)
import System.Directory (doesFileExist, removeFile)
import System.IO (IOMode (..), withFile)
import System.Timeout
import Test.HUnit
@@ -65,8 +61,11 @@ import Test.Hspec hiding (fit, it, xit)
import Util
#if defined(dbServerPostgres)
import CoreTests.MsgStoreTests (testPostgresStoreConfig)
import CoreTests.MsgStoreTests (testPostgresStoreConfig, testSMTStoreConfig)
import Simplex.Messaging.Server.Env.STM (readWriteQueueStore)
import Simplex.Messaging.Server.MsgStore.Postgres (PostgresMsgStoreCfg (..), exportDbMessages)
import Simplex.Messaging.Server.MsgStore.Types (MsgStoreClass (..), newMsgStore)
import Simplex.Messaging.Server.StoreLog (closeStoreLog)
#endif
serverTests :: SpecWith (ASrvTransport, AStoreType)
@@ -1068,7 +1067,6 @@ testRestoreMessages =
it "should store messages on exit and restore on start" $ \(at@(ATransport t), msType) -> do
removeFileIfExists testStoreLogFile
removeFileIfExists testStoreMsgsFile
whenM (doesDirectoryExist testStoreMsgsDir) $ removeDirectoryRecursive testStoreMsgsDir
removeFileIfExists testServerStatsBackupFile
g <- C.newRandom
@@ -1138,7 +1136,6 @@ testRestoreMessages =
Right stats3 <- strDecode <$> B.readFile testServerStatsBackupFile
checkStats stats3 [rId] 5 5
removeFileIfExists testStoreMsgsFile
whenM (doesDirectoryExist testStoreMsgsDir) $ removeDirectoryRecursive testStoreMsgsDir
removeFile testServerStatsBackupFile
where
runTest :: (HasTestEnv, Transport c) => TProxy c 'TServer -> (THandleSMP c 'TClient -> IO ()) -> ThreadId -> Expectation
@@ -1218,28 +1215,23 @@ testRestoreExpireMessages =
where
exportStoreMessages :: HasTestEnv => AStoreType -> IO ()
exportStoreMessages = \case
ASType _ SMSJournal -> export
ASType _ SMSPostgres -> exportDB
ASType _ SMSMemory -> pure ()
where
export = do
ms <- readWriteQueues
exportMessages False (StoreJournal ms) testStoreMsgsFile False
closeMsgStore ms
#if defined(dbServerPostgres)
exportDB = do
readWriteQueues >>= closeMsgStore
ms' <- newMsgStore (testPostgresStoreConfig {quota = 4} :: PostgresMsgStoreCfg)
_n <- withFile testStoreMsgsFile WriteMode $ exportDbMessages False ms'
closeMsgStore ms'
readWriteQueues = do
ms <- newMsgStore testSMTStoreConfig
readWriteQueueStore True (mkQueue ms True) testStoreLogFile (queueStore ms) >>= closeStoreLog
removeFileIfExists testStoreMsgsFile
pure ms
#else
exportDB = error "compiled without server_postgres flag"
#endif
readWriteQueues = do
ms <- newMsgStore ((testJournalStoreCfg MQStoreCfg) {quota = 4} :: JournalStoreConfig 'QSMemory)
readWriteQueueStore True (mkQueue ms True) testStoreLogFile (stmQueueStore ms) >>= closeStoreLog
removeFileIfExists testStoreMsgsFile
pure ms
runTest :: (HasTestEnv, Transport c) => TProxy c 'TServer -> (THandleSMP c 'TClient -> IO ()) -> ThreadId -> Expectation
runTest _ test' server = do
testSMPClient test' `shouldReturn` ()
@@ -1576,7 +1568,7 @@ testMsgExpireOnInterval =
it "should expire messages that are not received before messageTTL after expiry interval" $ \(ATransport (t :: TProxy c 'TServer), msType) -> do
g <- C.newRandom
(sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
let cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {messageExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 1}, idleQueueInterval = 1}
let cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {messageExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 1}}
withSmpServerConfigOn (ATransport t) cfg' testPort $ \_ ->
testSMPClient @c $ \sh -> do
(sId, rId, rKey, _) <- testSMPClient @c $ \rh -> createAndSecureQueue rh sPub
+6 -18
View File
@@ -101,37 +101,27 @@ main = do
describe "RSLV functional API tests" rslvTests
describe "Agent core tests" agentCoreTests
#if defined(dbServerPostgres)
-- xdescribe "SMP server via TLS, postgres+jornal message store" $
-- before (pure (transport @TLS, ASType SQSPostgres SMSJournal)) serverTests
describe "SMP server via TLS, postgres-only message store" $
before (pure (transport @TLS, ASType SQSPostgres SMSPostgres)) serverTests
#endif
describe "SMP server via TLS, jornal message store" $ do
describe "SMP server via TLS, memory message store" $ do
describe "SMP syntax" $ serverSyntaxTests (transport @TLS)
before (pure (transport @TLS, ASType SQSMemory SMSJournal)) serverTests
describe "SMP server via TLS, memory message store" $
before (pure (transport @TLS, ASType SQSMemory SMSMemory)) serverTests
-- xdescribe "SMP server via WebSockets" $ do
-- describe "SMP syntax" $ serverSyntaxTests (transport @WS)
-- before (pure (transport @WS, ASType SQSMemory SMSJournal)) serverTests
-- before (pure (transport @WS, ASType SQSMemory SMSMemory)) serverTests
#if defined(dbServerPostgres)
describe "Notifications server (SMP server: memory store)" $
ntfServerTests (transport @TLS, ASType SQSMemory SMSMemory)
-- xdescribe "Notifications server (SMP server: postgres+jornal store)" $
-- ntfServerTests (transport @TLS, ASType SQSPostgres SMSJournal)
describe "Notifications server (SMP server: postgres-only store)" $
ntfServerTests (transport @TLS, ASType SQSPostgres SMSPostgres)
-- xdescribe "SMP client agent, postgres+jornal message store" $ agentTests (transport @TLS, ASType SQSPostgres SMSJournal)
describe "SMP client agent, server postgres-only message store" $ agentTests (transport @TLS, ASType SQSPostgres SMSPostgres)
-- xdescribe "SMP proxy, postgres+jornal message store" $
-- before (pure $ ASType SQSPostgres SMSJournal) smpProxyTests
describe "SMP proxy, postgres-only message store" $
before (pure $ ASType SQSPostgres SMSPostgres) smpProxyTests
#endif
-- xdescribe "SMP client agent, server jornal message store" $ agentTests (transport @TLS, ASType SQSMemory SMSJournal)
describe "SMP client agent, server memory message store" $ agentTests (transport @TLS, ASType SQSMemory SMSMemory)
describe "SMP proxy, jornal message store" $
before (pure $ ASType SQSMemory SMSJournal) smpProxyTests
describe "SMP proxy, memory message store" $
before (pure $ ASType SQSMemory SMSMemory) smpProxyTests
describe "XFTP" $ do
describe "XFTP server" $
before (pure $ AFSType SFSMemory) xftpServerTests
@@ -155,8 +145,6 @@ main = do
describe "SMP server via TLS, postgres-only message store" $
before (pure (transport @TLS, ASType SQSPostgres SMSPostgres)) $ describe "Timing of AUTH error" testTiming
#endif
describe "SMP server via TLS, jornal message store" $
before (pure (transport @TLS, ASType SQSMemory SMSJournal)) $ describe "Timing of AUTH error" testTiming
describe "SMP server via TLS, memory message store" $
before (pure (transport @TLS, ASType SQSMemory SMSMemory)) $ describe "Timing of AUTH error" testTiming
#if defined(dbServerPostgres)
@@ -169,8 +157,8 @@ main = do
describe "SMP proxy capabilities, postgres-only message store" $
before (pure $ ASType SQSPostgres SMSPostgres) smpProxyCapabilityTests
#endif
describe "SMP proxy capabilities, jornal message store" $
before (pure $ ASType SQSMemory SMSJournal) smpProxyCapabilityTests
describe "SMP proxy capabilities, memory message store" $
before (pure $ ASType SQSMemory SMSMemory) smpProxyCapabilityTests
describe "XFTP CLI (memory)" $
before (pure $ AFSType SFSMemory) xftpCLIFileTests
#if defined(dbServerPostgres)