dandri/simplexmq
1
0
Fork 0
mirror of https://github.com/simplex-chat/simplexmq.git synced 2026-03-30 14:16:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

smp server: journal message store (#1370)

Browse Source 
* smp server: remove STM function from MsgStore

* polymorphic MsgStore

* jourmal storage for messages (WIP)

* more journal, test setup

* writeMsg

* test

* tryDelMsg

* delMsgQueue

* remove MsgStoreClass instance of existential wrapper for Msg stores

* store config

* extract common logic out of store instances

* add store type to config

* open journals, cache last message, tests pass

* CLI commands

* refactor import/export messages

* cli commands to import/export journal message store

* export journal without draining, import/export tests

* journal command

* import/export progress

* better progress info

* only log queue state once when importing

* logs

* handle IO errors in journal store, return as STORE error

* recover from state file errors

* fix message files after crash

* fix messages folder
This commit is contained in:
Evgeny
2024-10-21 11:50:30 +01:00
committed by GitHub
parent 967afaf802
commit 0414ea59f0
21 changed files with 1360 additions and 315 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
rfcs/2024-09-01-smp-message-storage.md
View File

@@ -96,7 +96,7 @@ else if write_msg = max_queue_messages:
add quota_exceeded message to write_file
update queue state: write_msg += 1
append updated queue state to queue.log
else ]
else
// It is required that `max_queue_messages < max_file_messages`,
// so that we never need more than one additional write file.
if write_msg >= max_file_messages: // queue file rotation

3
simplexmq.cabal
View File

@@ -174,7 +174,9 @@ library
Simplex.Messaging.Server.Information
Simplex.Messaging.Server.Main
Simplex.Messaging.Server.MsgStore
Simplex.Messaging.Server.MsgStore.Journal
Simplex.Messaging.Server.MsgStore.STM
Simplex.Messaging.Server.MsgStore.Types
Simplex.Messaging.Server.NtfStore
Simplex.Messaging.Server.QueueStore
Simplex.Messaging.Server.QueueStore.QueueInfo
@@ -613,6 +615,7 @@ test-suite simplexmq-test
CoreTests.CryptoFileTests
CoreTests.CryptoTests
CoreTests.EncodingTests
CoreTests.MsgStoreTests
CoreTests.RetryIntervalTests
CoreTests.SOCKSSettings
CoreTests.StoreLogTests

4
src/Simplex/FileTransfer/Server/Main.hs
View File

@@ -51,7 +51,9 @@ xftpServerCLI cfgPath logPath = do
True -> readIniFile iniFile >>= either exitError runServer
_ -> exitError $ "Error: server is not initialized (" <> iniFile <> " does not exist).\nRun `" <> executableName <> " init`."
Delete -> do
confirmOrExit "WARNING: deleting the server will make all queues inaccessible, because the server identity (certificate fingerprint) will change.\nTHIS CANNOT BE UNDONE!"
confirmOrExit
"WARNING: deleting the server will make all queues inaccessible, because the server identity (certificate fingerprint) will change.\nTHIS CANNOT BE UNDONE!"
"Server NOT deleted"
deleteDirIfExists cfgPath
deleteDirIfExists logPath
putStrLn "Deleted configuration and log files"

3
src/Simplex/Messaging/Agent/Client.hs
View File

@@ -30,6 +30,7 @@ module Simplex.Messaging.Agent.Client
withConnLocks,
withInvLock,
withLockMap,
getMapLock,
ipAddressProtected,
closeAgentClient,
closeProtocolServerClients,
@@ -500,7 +501,7 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} currentTs a
getMsgLocks <- TM.emptyIO
connLocks <- TM.emptyIO
invLocks <- TM.emptyIO
deleteLock <- atomically createLock
deleteLock <- createLockIO
smpSubWorkers <- TM.emptyIO
smpServersStats <- TM.emptyIO
xftpServersStats <- TM.emptyIO

5
src/Simplex/Messaging/Agent/Lock.hs
View File

@@ -1,6 +1,7 @@
module Simplex.Messaging.Agent.Lock
( Lock,
createLock,
createLockIO,
withLock,
withLock',
withGetLock,
@@ -24,6 +25,10 @@ createLock :: STM Lock
createLock = newEmptyTMVar
{-# INLINE createLock #-}
createLockIO :: IO Lock
createLockIO = newEmptyTMVarIO
{-# INLINE createLockIO #-}
withLock :: MonadUnliftIO m => Lock -> String -> ExceptT e m a -> ExceptT e m a
withLock lock name = ExceptT . withLock' lock name . runExceptT
{-# INLINE withLock #-}

4
src/Simplex/Messaging/Notifications/Server/Main.hs
View File

@@ -53,7 +53,9 @@ ntfServerCLI cfgPath logPath =
True -> readIniFile iniFile >>= either exitError runServer
_ -> exitError $ "Error: server is not initialized (" <> iniFile <> " does not exist).\nRun `" <> executableName <> " init`."
Delete -> do
confirmOrExit "WARNING: deleting the server will make all queues inaccessible, because the server identity (certificate fingerprint) will change.\nTHIS CANNOT BE UNDONE!"
confirmOrExit
"WARNING: deleting the server will make all queues inaccessible, because the server identity (certificate fingerprint) will change.\nTHIS CANNOT BE UNDONE!"
"Server NOT deleted"
deleteDirIfExists cfgPath
deleteDirIfExists logPath
putStrLn "Deleted configuration and log files"

2
src/Simplex/Messaging/Protocol.hs
View File

@@ -399,8 +399,6 @@ data Command (p :: Party) where
NKEY :: NtfPublicAuthKey -> RcvNtfPublicDhKey -> Command Recipient
NDEL :: Command Recipient
GET :: Command Recipient
-- ACK v1 has to be supported for encoding/decoding
-- ACK :: Command Recipient
ACK :: MsgId -> Command Recipient
OFF :: Command Recipient
DEL :: Command Recipient

281
src/Simplex/Messaging/Server.hs
View File

@@ -30,6 +30,8 @@
module Simplex.Messaging.Server
( runSMPServer,
runSMPServerBlocking,
importMessages,
exportMessages,
disconnectTransport,
verifyCmdAuthorization,
dummyVerifyCmd,
@@ -39,7 +41,6 @@ module Simplex.Messaging.Server
where
import Control.Concurrent.STM (throwSTM)
import Control.Concurrent.STM.TQueue (flushTQueue)
import Control.Logger.Simple
import Control.Monad
import Control.Monad.Except
@@ -91,6 +92,7 @@ import Simplex.Messaging.Server.Env.STM as Env
import Simplex.Messaging.Server.Expiration
import Simplex.Messaging.Server.MsgStore
import Simplex.Messaging.Server.MsgStore.STM
import Simplex.Messaging.Server.MsgStore.Types
import Simplex.Messaging.Server.NtfStore
import Simplex.Messaging.Server.QueueStore
import Simplex.Messaging.Server.QueueStore.QueueInfo
@@ -193,11 +195,17 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
stopServer s = do
asks serverActive >>= atomically . (`writeTVar` False)
logInfo "Saving server state..."
withLock' (savingLock s) "final" $ saveServer False >> closeServer
withLock' (savingLock s) "final" $ saveServer True >> closeServer
logInfo "Server stopped"
saveServer :: Bool -> M ()
saveServer keepMsgs = withLog closeStoreLog >> saveServerMessages keepMsgs >> saveServerNtfs >> saveServerStats
saveServer drainMsgs = do
withLog closeStoreLog
AMS _ ms <- asks msgStore
liftIO $ closeMsgStore ms
saveServerMessages drainMsgs
saveServerNtfs
saveServerStats
closeServer :: M ()
closeServer = asks (smpAgent . proxyAgent) >>= liftIO . closeSMPClientAgent
@@ -351,19 +359,17 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
expireMessagesThread :: ExpirationConfig -> M ()
expireMessagesThread expCfg = do
ms <- asks msgStore
quota <- asks $ msgQueueQuota . config
AMS _ ms <- asks msgStore
let interval = checkInterval expCfg * 1000000
stats <- asks serverStats
labelMyThread "expireMessagesThread"
forever $ do
liftIO $ threadDelay' interval
old <- liftIO $ expireBeforeEpoch expCfg
rIds <- M.keysSet <$> readTVarIO ms
forM_ rIds $ \rId -> do
q <- liftIO $ getMsgQueue ms rId quota
deleted <- liftIO $ deleteExpiredMsgs q old
liftIO $ atomicModifyIORef'_ (msgExpired stats) (+ deleted)
liftIO $ forever $ do
threadDelay' interval
old <- expireBeforeEpoch expCfg
void $ withActiveMsgQueues ms $ \_ q -> do
runExceptT (deleteExpiredMsgs q old) >>= \case
Right deleted -> deleted <$ atomicModifyIORef'_ (msgExpired stats) (+ deleted)
Left _ -> pure 0
expireNtfsThread :: ServerConfig -> M ()
expireNtfsThread ServerConfig {notificationExpiration = expCfg} = do
@@ -767,7 +773,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
ProhibitSub -> pure (c1, c2, c3, c4 + 1)
CPDelete queueId' -> withUserRole $ unliftIO u $ do
st <- asks queueStore
ms <- asks msgStore
AMS _ ms <- asks msgStore
queueId <- liftIO (getQueue st SSender queueId') >>= \case
Left _ -> pure queueId' -- fallback to using as recipientId directly
Right QueueRec {recipientId} -> pure recipientId
@@ -782,7 +788,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
liftIO . hPutStrLn h $ "ok, " <> show numDeleted <> " messages deleted"
CPSave -> withAdminRole $ withLock' (savingLock srv) "control" $ do
hPutStrLn h "saving server state..."
unliftIO u $ saveServer True
unliftIO u $ saveServer False
hPutStrLn h "server state saved!"
CPHelp -> hPutStrLn h "commands: stats, stats-rts, clients, sockets, socket-threads, threads, server-info, delete, save, help, quit"
CPQuit -> pure ()
@@ -1166,7 +1172,7 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
NDEL -> deleteQueueNotifier_ st
OFF -> suspendQueue_ st
DEL -> delQueueAndMsgs st
QUE -> withQueue getQueueInfo
QUE -> withQueue $ fmap (corrId,entId,) . getQueueInfo
where
createQueue :: QueueStore -> RcvPublicAuthKey -> RcvPublicDhKey -> SubscriptionMode -> SenderCanSecure -> M (Transmission BrokerMsg)
createQueue st recipientKey dhKey subMode sndSecure = time "NEW" $ do
@@ -1262,7 +1268,7 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
okResp <$> liftIO (suspendQueue st entId)
subscribeQueue :: QueueRec -> RecipientId -> M (Transmission BrokerMsg)
subscribeQueue qr rId = do
subscribeQueue qr rId =
atomically (TM.lookup rId subscriptions) >>= \case
Nothing -> newSub >>= deliver True
Just s@Sub {subThread} -> do
@@ -1284,11 +1290,13 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
pure sub
deliver :: Bool -> Sub -> M (Transmission BrokerMsg)
deliver inc sub = do
q <- getStoreMsgQueue "SUB" rId
msg_ <- liftIO $ tryPeekMsgIO q
when (inc && isJust msg_) $
incStat . qSub =<< asks serverStats
deliverMessage "SUB" qr rId sub msg_
AMS _ ms <- asks msgStore
stats <- asks serverStats
fmap (either (\e -> (corrId, rId, ERR e)) id) $ liftIO $ runExceptT $ do
q <- getMsgQueue ms rId
msg_ <- tryPeekMsg q
liftIO $ when (inc && isJust msg_) $ incStat (qSub stats)
liftIO $ deliverMessage "SUB" qr rId sub msg_
-- clients that use GET are not added to server subscribers
getMessage :: QueueRec -> M (Transmission BrokerMsg)
@@ -1305,7 +1313,7 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
_ -> do
stats <- asks serverStats
incStat $ msgGetProhibited stats
pure (corrId, entId, ERR $ CMD PROHIBITED)
pure $ err $ CMD PROHIBITED
where
newSub :: STM Sub
newSub = do
@@ -1318,19 +1326,16 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
pure s
getMessage_ :: Sub -> Maybe MsgId -> M (Transmission BrokerMsg)
getMessage_ s delivered_ = do
q <- getStoreMsgQueue "GET" entId
AMS _ ms <- asks msgStore
stats <- asks serverStats
(statCnt, r) <-
-- TODO split STM, use tryPeekMsgIO
atomically $
tryPeekMsg q >>= \case
Just msg ->
let encMsg = encryptMsg qr msg
cnt = if isJust delivered_ then msgGetDuplicate else msgGet
in setDelivered s msg $> (cnt, (corrId, entId, MSG encMsg))
_ -> pure (msgGetNoMsg, (corrId, entId, OK))
incStat $ statCnt stats
pure r
fmap (either err id) $ liftIO $ runExceptT $ do
q <- getMsgQueue ms entId
tryPeekMsg q >>= \case
Just msg -> do
let encMsg = encryptMsg qr msg
incStat $ (if isJust delivered_ then msgGetDuplicate else msgGet) stats
atomically $ setDelivered s msg $> (corrId, entId, MSG encMsg)
Nothing -> incStat (msgGetNoMsg stats) $> ok
withQueue :: (QueueRec -> M (Transmission BrokerMsg)) -> M (Transmission BrokerMsg)
withQueue action = case qr_ of
@@ -1367,16 +1372,19 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
Just sub ->
atomically (getDelivered sub) >>= \case
Just st -> do
q <- getStoreMsgQueue "ACK" entId
case st of
ProhibitSub -> do
deletedMsg_ <- liftIO $ tryDelMsg q msgId
mapM_ (updateStats True) deletedMsg_
pure ok
_ -> do
(deletedMsg_, msg_) <- liftIO $ tryDelPeekMsg q msgId
mapM_ (updateStats False) deletedMsg_
deliverMessage "ACK" qr entId sub msg_
AMS _ ms <- asks msgStore
stats <- asks serverStats
fmap (either err id) $ liftIO $ runExceptT $ do
q <- getMsgQueue ms entId
case st of
ProhibitSub -> do
deletedMsg_ <- tryDelMsg q msgId
liftIO $ mapM_ (updateStats stats True) deletedMsg_
pure ok
_ -> do
(deletedMsg_, msg_) <- tryDelPeekMsg q msgId
liftIO $ mapM_ (updateStats stats False) deletedMsg_
liftIO $ deliverMessage "ACK" qr entId sub msg_
_ -> pure $ err NO_MSG
where
getDelivered :: Sub -> STM (Maybe ServerSub)
@@ -1385,11 +1393,10 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
if msgId == msgId' || B.null msgId
then pure $ Just subThread
else putTMVar delivered msgId' $> Nothing
updateStats :: Bool -> Message -> M ()
updateStats isGet = \case
updateStats :: ServerStats -> Bool -> Message -> IO ()
updateStats stats isGet = \case
MessageQuota {} -> pure ()
Message {msgFlags} -> do
stats <- asks serverStats
incStat $ msgRecv stats
if isGet
then incStat $ msgRecvGet stats
@@ -1399,11 +1406,11 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
-- ns <- asks ntfStore
-- atomically $ TM.lookup nId ns >>=
-- mapM_ (\MsgNtf {ntfMsgId} -> when (msgId == msgId') $ TM.delete nId ns)
liftIO $ atomicModifyIORef'_ (msgCount stats) (subtract 1)
liftIO $ updatePeriodStats (activeQueues stats) entId
atomicModifyIORef'_ (msgCount stats) (subtract 1)
updatePeriodStats (activeQueues stats) entId
when (notification msgFlags) $ do
incStat $ msgRecvNtf stats
liftIO $ updatePeriodStats (activeQueuesNtf stats) entId
updatePeriodStats (activeQueuesNtf stats) entId
sendMessage :: QueueRec -> MsgFlags -> MsgBody -> M (Transmission BrokerMsg)
sendMessage qr msgFlags msgBody
@@ -1421,15 +1428,20 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
case C.maxLenBS msgBody of
Left _ -> pure $ err LARGE_MSG
Right body -> do
msg_ <- time "SEND" $ do
q <- getStoreMsgQueue "SEND" $ recipientId qr
expireMessages q
liftIO . writeMsg q =<< mkMessage body
AMS _ ms <- asks msgStore
ServerConfig {messageExpiration, msgIdBytes} <- asks config
msgId <- randomId' msgIdBytes
msg_ <- liftIO $ time "SEND" $ runExceptT $ do
q <- getMsgQueue ms $ recipientId qr
expireMessages q messageExpiration stats
msg <- liftIO $ mkMessage msgId body
writeMsg q True msg
case msg_ of
Nothing -> do
Left e -> pure $ err e
Right Nothing -> do
incStat $ msgSentQuota stats
pure $ err QUOTA
Just (msg, wasEmpty) -> time "SEND ok" $ do
Right (Just (msg, wasEmpty)) -> time "SEND ok" $ do
when wasEmpty $ liftIO $ tryDeliverMessage msg
when (notification msgFlags) $ do
mapM_ (`enqueueNotification` msg) (notifier qr)
@@ -1441,20 +1453,15 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
pure ok
where
THandleParams {thVersion} = thParams'
mkMessage :: C.MaxLenBS MaxMessageLen -> M Message
mkMessage body = do
msgId <- randomId' =<< asks (msgIdBytes . config)
msgTs <- liftIO getSystemTime
mkMessage :: MsgId -> C.MaxLenBS MaxMessageLen -> IO Message
mkMessage msgId body = do
msgTs <- getSystemTime
pure $! Message msgId msgTs msgFlags body
expireMessages :: MsgQueue -> M ()
expireMessages q = do
msgExp <- asks $ messageExpiration . config
old <- liftIO $ mapM expireBeforeEpoch msgExp
deleted <- liftIO $ sum <$> mapM (deleteExpiredMsgs q) old
when (deleted > 0) $ do
stats <- asks serverStats
liftIO $ atomicModifyIORef'_ (msgExpired stats) (+ deleted)
expireMessages :: MsgStoreClass s => MsgQueue s -> Maybe ExpirationConfig -> ServerStats -> ExceptT ErrorType IO ()
expireMessages q msgExp stats = do
deleted <- maybe (pure 0) (deleteExpiredMsgs q <=< liftIO . expireBeforeEpoch) msgExp
liftIO $ when (deleted > 0) $ atomicModifyIORef'_ (msgExpired stats) (+ deleted)
-- The condition for delivery of the message is:
-- - the queue was empty when the message was sent,
@@ -1472,7 +1479,6 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
atomically deliverToSub >>= mapM_ forkDeliver
where
rId = recipientId qr
-- remove tryPeekMsg
deliverToSub =
-- lookup has ot be in the same transaction,
-- so that if subscription ends, it re-evalutates
@@ -1586,7 +1592,7 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
verified = \case
VRVerified qr -> Right (qr, (corrId', entId', cmd'))
VRFailed -> Left (corrId', entId', ERR AUTH)
deliverMessage :: T.Text -> QueueRec -> RecipientId -> Sub -> Maybe Message -> M (Transmission BrokerMsg)
deliverMessage :: T.Text -> QueueRec -> RecipientId -> Sub -> Maybe Message -> IO (Transmission BrokerMsg)
deliverMessage name qr rId s@Sub {subThread} msg_ = time (name <> " deliver") . atomically $
case subThread of
ProhibitSub -> pure resp
@@ -1614,16 +1620,10 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
setDelivered :: Sub -> Message -> STM Bool
setDelivered s msg = tryPutTMVar (delivered s) $! messageId msg
getStoreMsgQueue :: T.Text -> RecipientId -> M MsgQueue
getStoreMsgQueue name rId = time (name <> " getMsgQueue") $ do
ms <- asks msgStore
quota <- asks $ msgQueueQuota . config
liftIO $ getMsgQueue ms rId quota
delQueueAndMsgs :: QueueStore -> M (Transmission BrokerMsg)
delQueueAndMsgs st = do
withLog (`logDeleteQueue` entId)
ms <- asks msgStore
AMS _ ms <- asks msgStore
liftIO (deleteQueue st entId $>>= \q -> delMsgQueue ms entId $> Right q) >>= \case
Right q -> do
-- Possibly, the same should be done if the queue is suspended, but currently we do not use it
@@ -1643,14 +1643,16 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
pure ok
Left e -> pure $ err e
getQueueInfo :: QueueRec -> M (Transmission BrokerMsg)
getQueueInfo :: QueueRec -> M BrokerMsg
getQueueInfo QueueRec {senderKey, notifier} = do
q <- getStoreMsgQueue "getQueueInfo" entId
qiSub <- liftIO $ TM.lookupIO entId subscriptions >>= mapM mkQSub
qiSize <- liftIO $ getQueueSize q
qiMsg <- liftIO $ toMsgInfo <$$> tryPeekMsgIO q
let info = QueueInfo {qiSnd = isJust senderKey, qiNtf = isJust notifier, qiSub, qiSize, qiMsg}
pure (corrId, entId, INFO info)
AMS _ ms <- asks msgStore
fmap (either ERR id) $ liftIO $ runExceptT $ do
q <- getMsgQueue ms entId
qiSub <- liftIO $ TM.lookupIO entId subscriptions >>= mapM mkQSub
qiSize <- liftIO $ getQueueSize q
qiMsg <- toMsgInfo <$$> tryPeekMsg q
let info = QueueInfo {qiSnd = isJust senderKey, qiNtf = isJust notifier, qiSub, qiSize, qiMsg}
pure $ INFO info
where
mkQSub Sub {subThread, delivered} = do
qSubThread <- case subThread of
@@ -1710,59 +1712,69 @@ randomId = fmap EntityId . randomId'
{-# INLINE randomId #-}
saveServerMessages :: Bool -> M ()
saveServerMessages keepMsgs = asks (storeMsgsFile . config) >>= mapM_ saveMessages
saveServerMessages drainMsgs =
asks msgStore >>= \case
AMS SMSMemory ms@STMMsgStore {storeConfig = STMStoreConfig {storePath}} -> case storePath of
Just f -> liftIO $ exportMessages ms f $ getQueueMessages drainMsgs
Nothing -> logInfo "undelivered messages are not saved"
AMS SMSJournal _ -> logInfo "closed journal message storage"
exportMessages :: MsgStoreClass s => s -> FilePath -> (MsgQueue s -> IO [Message]) -> IO ()
exportMessages ms f getMessages = do
logInfo $ "saving messages to file " <> T.pack f
total <- liftIO $ withFile f WriteMode $ withAllMsgQueues ms . saveQueueMsgs
logInfo $ "messages saved: " <> tshow total
where
saveMessages f = do
logInfo $ "saving messages to file " <> T.pack f
ms <- asks msgStore
liftIO . withFile f WriteMode $ \h ->
readTVarIO ms >>= mapM_ (saveQueueMsgs h) . M.assocs
logInfo "messages saved"
where
saveQueueMsgs h (rId, q) = BLD.hPutBuilder h . encodeMessages rId =<< atomically (getMessages $ msgQueue q)
getMessages = if keepMsgs then snapshotTQueue else flushTQueue
snapshotTQueue q = do
msgs <- flushTQueue q
mapM_ (writeTQueue q) msgs
pure msgs
encodeMessages rId = mconcat . map (\msg -> BLD.byteString (strEncode $ MLRv3 rId msg) <> BLD.char8 '\n')
saveQueueMsgs h rId q = getMessages q >>= \msgs -> length msgs <$ BLD.hPutBuilder h (encodeMessages rId msgs)
encodeMessages rId = mconcat . map (\msg -> BLD.byteString (strEncode $ MLRv3 rId msg) <> BLD.char8 '\n')
restoreServerMessages :: M Int
restoreServerMessages =
asks (storeMsgsFile . config) >>= \case
Just f -> ifM (doesFileExist f) (restoreMessages f) (pure 0)
Nothing -> pure 0
where
restoreMessages f = do
logInfo $ "restoring messages from file " <> T.pack f
ms <- asks msgStore
quota <- asks $ msgQueueQuota . config
asks msgStore >>= \case
AMS SMSMemory ms@STMMsgStore {storeConfig = STMStoreConfig {storePath = Just f}} -> do
old_ <- asks (messageExpiration . config) $>>= (liftIO . fmap Just . expireBeforeEpoch)
runExceptT (liftIO (LB.readFile f) >>= foldM (\expired -> restoreMsg expired ms quota old_) 0 . LB.lines) >>= \case
Left e -> do
logError . T.pack $ "error restoring messages: " <> e
liftIO exitFailure
Right expired -> do
renameFile f $ f <> ".bak"
logInfo "messages restored"
pure expired
liftIO $ ifM (doesFileExist f) (importMessages ms f old_) (pure 0)
_ -> logInfo "using journal message storage" $> 0
importMessages :: MsgStoreClass s => s -> FilePath -> Maybe Int64 -> IO Int
importMessages ms f old_ = do
logInfo $ "restoring messages from file " <> T.pack f
LB.readFile f >>= runExceptT . foldM restoreMsg (0, 0, 0) . zip [0..] . LB.lines >>= \case
Left e -> do
putStrLn ""
logError . T.pack $ "error restoring messages: " <> e
liftIO exitFailure
Right (count, total, expired) -> do
putStrLn ""
renameFile f $ f <> ".bak"
logQueueStates ms
qCount <- M.size <$> readTVarIO (activeMsgQueues ms)
logInfo $ "Processed " <> tshow count <> " lines, imported " <> tshow total <> " messages into " <> tshow qCount <> " queues, expired " <> tshow expired <> " messages"
pure expired
where
progress i = do
liftIO $ putStr $ "Processed " <> show i <> " lines\r"
hFlush stdout
restoreMsg :: (Int, Int, Int) -> (Int, LB.ByteString) -> ExceptT String IO (Int, Int, Int)
restoreMsg (!count, !total, !expired) (i, s') = do
when (i `mod` 1000 == 0) $ progress i
MLRv3 rId msg <- liftEither . first (msgErr "parsing") $ strDecode s
liftError show $ addToMsgQueue rId msg
where
restoreMsg !expired ms quota old_ s' = do
MLRv3 rId msg <- liftEither . first (msgErr "parsing") $ strDecode s
addToMsgQueue rId msg
where
s = LB.toStrict s'
addToMsgQueue rId msg = do
q <- liftIO $ getMsgQueue ms rId quota
(isExpired, logFull) <- liftIO $ case msg of
Message {msgTs}
| maybe True (systemSeconds msgTs >=) old_ -> (False,) . isNothing <$> writeMsg q msg
| otherwise -> pure (True, False)
MessageQuota {} -> writeMsg q msg $> (False, False)
when logFull . logError . decodeLatin1 $ "message queue " <> strEncode rId <> " is full, message not restored: " <> strEncode (messageId msg)
pure $ if isExpired then expired + 1 else expired
msgErr :: Show e => String -> e -> String
msgErr op e = op <> " error (" <> show e <> "): " <> B.unpack (B.take 100 s)
s = LB.toStrict s'
addToMsgQueue rId msg = do
q <- getMsgQueue ms rId
(isExpired, logFull) <- case msg of
Message {msgTs}
| maybe True (systemSeconds msgTs >=) old_ -> (False,) . isNothing <$> writeMsg q False msg
| otherwise -> pure (True, False)
MessageQuota {} -> writeMsg q False msg $> (False, False)
when logFull . logError . decodeLatin1 $ "message queue " <> strEncode rId <> " is full, message not restored: " <> strEncode (messageId msg)
let total' = if logFull then total else total + 1
expired' = if isExpired then expired + 1 else expired
pure (count + 1, total', expired')
msgErr :: Show e => String -> e -> String
msgErr op e = op <> " error (" <> show e <> "): " <> B.unpack (B.take 100 s)
saveServerNtfs :: M ()
saveServerNtfs = asks (storeNtfsFile . config) >>= mapM_ saveNtfs
@@ -1827,7 +1839,10 @@ restoreServerStats expiredMsgs expiredNtfs = asks (serverStatsBackupFile . confi
Right d@ServerStatsData {_qCount = statsQCount} -> do
s <- asks serverStats
_qCount <- fmap M.size . readTVarIO . queues =<< asks queueStore
_msgCount <- liftIO . foldM (\(!n) q -> (n +) <$> getQueueSize q) 0 =<< readTVarIO =<< asks msgStore
_msgCount <-
asks msgStore >>= \case
AMS SMSMemory ms -> liftIO $ readTVarIO (msgQueues ms) >>= foldM (\(!n) q -> (n +) <$> getQueueSize q) 0
_ -> pure 0
NtfStore ns <- asks ntfStore
_ntfCount <- liftIO . foldM (\(!n) q -> (n +) . length <$> readTVarIO q) 0 =<< readTVarIO ns
liftIO $ setServerStats s d {_qCount, _msgCount, _ntfCount, _msgExpired = _msgExpired d + expiredMsgs, _msgNtfExpired = _msgNtfExpired d + expiredNtfs}

6
src/Simplex/Messaging/Server/CLI.hs
View File

@@ -44,11 +44,11 @@ import Text.Read (readMaybe)
exitError :: String -> IO a
exitError msg = putStrLn msg >> exitFailure
confirmOrExit :: String -> IO ()
confirmOrExit s =
confirmOrExit :: String -> String -> IO ()
confirmOrExit s no =
withPrompt (s <> "\nContinue (Y/n): ") $ do
ok <- getLine
when (ok /= "Y") $ putStrLn "Server NOT deleted" >> exitFailure
when (ok /= "Y") $ putStrLn no >> exitFailure
data SignAlgorithm = ED448 | ED25519
deriving (Read, Show)

51
src/Simplex/Messaging/Server/Env/STM.hs
View File

@@ -1,10 +1,14 @@
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE DuplicateRecordFields #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE KindSignatures #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE StrictData #-}
{-# LANGUAGE TypeFamilies #-}
module Simplex.Messaging.Server.Env.STM where
@@ -37,7 +41,9 @@ 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.NtfStore
import Simplex.Messaging.Server.QueueStore (NtfCreds (..), QueueRec (..))
import Simplex.Messaging.Server.QueueStore.STM
@@ -57,7 +63,10 @@ data ServerConfig = ServerConfig
{ transports :: [(ServiceName, ATransport, AddHTTP)],
smpHandshakeTimeout :: Int,
tbqSize :: Natural,
msgStoreType :: AMSType,
msgQueueQuota :: Int,
maxJournalMsgCount :: Int,
maxJournalStateLines :: Int,
queueIdBytes :: Int,
msgIdBytes :: Int,
storeLogFile :: Maybe FilePath,
@@ -136,6 +145,18 @@ defaultInactiveClientExpiration =
defaultProxyClientConcurrency :: Int
defaultProxyClientConcurrency = 32
journalMsgStoreDepth :: Int
journalMsgStoreDepth = 5
defaultMaxJournalStateLines :: Int
defaultMaxJournalStateLines = 16
defaultMaxJournalMsgCount :: Int
defaultMaxJournalMsgCount = 256
defaultMsgQueueQuota :: Int
defaultMsgQueueQuota = 128
data Env = Env
{ config :: ServerConfig,
serverActive :: TVar Bool,
@@ -143,7 +164,7 @@ data Env = Env
server :: Server,
serverIdentity :: KeyHash,
queueStore :: QueueStore,
msgStore :: STMMsgStore,
msgStore :: AMsgStore,
ntfStore :: NtfStore,
random :: TVar ChaChaDRG,
storeLog :: Maybe (StoreLog 'WriteMode),
@@ -156,6 +177,20 @@ data Env = Env
proxyAgent :: ProxyAgent -- senders served on this proxy
}
type family MsgStore s where
MsgStore 'MSMemory = STMMsgStore
MsgStore 'MSJournal = JournalMsgStore
data AMsgStore = forall s. MsgStoreClass (MsgStore s) => AMS (SMSType s) (MsgStore s)
data AMsgQueue = forall s. MsgStoreClass (MsgStore s) => AMQ (SMSType s) (MsgQueue (MsgStore s))
data AMsgStoreCfg = forall s. MsgStoreClass (MsgStore s) => AMSC (SMSType s) (MsgStoreConfig (MsgStore s))
msgPersistence :: AMsgStoreCfg -> Bool
msgPersistence (AMSC SMSMemory (STMStoreConfig {storePath})) = isJust storePath
msgPersistence (AMSC SMSJournal _) = True
type Subscribed = Bool
data Server = Server
@@ -213,7 +248,7 @@ newServer = do
ntfSubClients <- newTVarIO IM.empty
pendingSubEvents <- newTVarIO IM.empty
pendingNtfSubEvents <- newTVarIO IM.empty
savingLock <- atomically createLock
savingLock <- createLockIO
return Server {subscribedQ, subscribers, ntfSubscribedQ, notifiers, subClients, ntfSubClients, pendingSubEvents, pendingNtfSubEvents, savingLock}
newClient :: ClientId -> Natural -> VersionSMP -> ByteString -> SystemTime -> IO Client
@@ -243,11 +278,17 @@ newProhibitedSub = do
return Sub {subThread = ProhibitSub, delivered}
newEnv :: ServerConfig -> IO Env
newEnv config@ServerConfig {smpCredentials, httpCredentials, storeLogFile, smpAgentCfg, information, messageExpiration} = do
newEnv config@ServerConfig {smpCredentials, httpCredentials, storeLogFile, msgStoreType, storeMsgsFile, smpAgentCfg, information, messageExpiration, msgQueueQuota, maxJournalMsgCount, maxJournalStateLines} = do
serverActive <- newTVarIO True
server <- newServer
queueStore <- newQueueStore
msgStore <- newMsgStore
msgStore <- case msgStoreType of
AMSType SMSMemory -> AMS SMSMemory <$> newMsgStore STMStoreConfig {storePath = storeMsgsFile, quota = msgQueueQuota}
AMSType SMSJournal -> case storeMsgsFile of
Just storePath ->
let cfg = JournalStoreConfig {storePath, quota = msgQueueQuota, pathParts = journalMsgStoreDepth, maxMsgCount = maxJournalMsgCount, maxStateLines = maxJournalStateLines}
in AMS SMSJournal <$> newMsgStore cfg
Nothing -> putStrLn "Error: journal msg store require path in [STORE_LOG], restore_messages" >> exitFailure
ntfStore <- NtfStore <$> TM.emptyIO
random <- C.newRandom
storeLog <-
@@ -314,7 +355,7 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, storeLogFile, smpAg
where
persistence
| isNothing storeLogFile = SPMMemoryOnly
| isJust (storeMsgsFile config) = SPMMessages
| isJust storeMsgsFile = SPMMessages
| otherwise = SPMQueues
newSMPProxyAgent :: SMPClientAgentConfig -> TVar ChaChaDRG -> IO ProxyAgent

170
src/Simplex/Messaging/Server/Main.hs
View File

@@ -1,6 +1,6 @@
{-# LANGUAGE ApplicativeDo #-}
{-# LANGUAGE BangPatterns #-}
{-# LANGUAGE DuplicateRecordFields #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedLists #-}
@@ -20,6 +20,7 @@ import Control.Monad
import Data.ByteString.Char8 (ByteString)
import qualified Data.ByteString.Char8 as B
import Data.Char (isAlpha, isAscii, toUpper)
import Data.Either (fromRight)
import Data.Functor (($>))
import Data.Ini (Ini, lookupValue, readIniFile)
import Data.List (find, isPrefixOf)
@@ -27,7 +28,7 @@ import qualified Data.List.NonEmpty as L
import Data.Maybe (fromMaybe, isJust, isNothing)
import Data.Text (Text)
import qualified Data.Text as T
import Data.Text.Encoding (encodeUtf8, decodeLatin1)
import Data.Text.Encoding (decodeLatin1, encodeUtf8)
import qualified Data.Text.IO as T
import Network.Socket (HostName)
import Options.Applicative
@@ -38,17 +39,19 @@ import qualified Simplex.Messaging.Crypto as C
import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Parsers (parseAll)
import Simplex.Messaging.Protocol (BasicAuth (..), ProtoServerWithAuth (ProtoServerWithAuth), pattern SMPServer)
import Simplex.Messaging.Server (AttachHTTP, runSMPServer)
import Simplex.Messaging.Server (AttachHTTP, exportMessages, importMessages, 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.MsgStore.Journal (JournalStoreConfig (..), getQueueMessages)
import Simplex.Messaging.Server.MsgStore.Types (AMSType (..), SMSType (..), newMsgStore)
import Simplex.Messaging.Transport (batchCmdsSMPVersion, sendingProxySMPVersion, simplexMQVersion, supportedServerSMPRelayVRange)
import Simplex.Messaging.Transport.Client (SocksProxy, TransportHost (..), defaultSocksProxy)
import Simplex.Messaging.Transport.Server (ServerCredentials (..), TransportServerConfig (..), defaultTransportServerConfig)
import Simplex.Messaging.Util (eitherToMaybe, safeDecodeUtf8, tshow)
import Simplex.Messaging.Version (mkVersionRange)
import System.Directory (createDirectoryIfMissing, doesFileExist)
import System.Directory (createDirectoryIfMissing, doesDirectoryExist, doesFileExist)
import System.Exit (exitFailure)
import System.FilePath (combine)
import System.IO (BufferMode (..), hSetBuffering, stderr, stdout)
@@ -70,25 +73,73 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
doesFileExist iniFile >>= \case
True -> exitError $ "Error: server is already initialized (" <> iniFile <> " exists).\nRun `" <> executableName <> " start`."
_ -> initializeServer opts
OnlineCert certOpts ->
doesFileExist iniFile >>= \case
True -> genOnline cfgPath certOpts
_ -> exitError $ "Error: server is not initialized (" <> iniFile <> " does not exist).\nRun `" <> executableName <> " init`."
Start ->
doesFileExist iniFile >>= \case
True -> readIniFile iniFile >>= either exitError runServer
_ -> exitError $ "Error: server is not initialized (" <> iniFile <> " does not exist).\nRun `" <> executableName <> " init`."
OnlineCert certOpts -> withIniFile $ \_ -> genOnline cfgPath certOpts
Start -> withIniFile runServer
Delete -> do
confirmOrExit "WARNING: deleting the server will make all queues inaccessible, because the server identity (certificate fingerprint) will change.\nTHIS CANNOT BE UNDONE!"
confirmOrExit
"WARNING: deleting the server will make all queues inaccessible, because the server identity (certificate fingerprint) will change.\nTHIS CANNOT BE UNDONE!"
"Server NOT deleted"
deleteDirIfExists cfgPath
deleteDirIfExists logPath
putStrLn "Deleted configuration and log files"
Journal cmd -> withIniFile $ \ini -> do
msgsDirExists <- doesDirectoryExist storeMsgsJournalDir
msgsFileExists <- doesFileExist storeMsgsFilePath
when (msgsFileExists && msgsDirExists) exitConfigureMsgStorage
case cmd of
JCImport
| msgsDirExists -> do
putStrLn $ storeMsgsJournalDir <> " directory already exists."
exitFailure
| not msgsFileExists -> do
putStrLn $ storeMsgsFilePath <> " file does not exists."
exitFailure
| otherwise -> do
confirmOrExit
("WARNING: message log file " <> storeMsgsFilePath <> " will be imported to journal directory " <> storeMsgsJournalDir)
"Messages not imported"
ms <- newJournalMsgStore
void $ importMessages ms storeMsgsFilePath Nothing -- no expiration
putStrLn "Import completed"
putStrLn $ case readMsgStoreType ini of
Right (AMSType SMSMemory) -> "store_messages set to `memory`, update it to `journal` in INI file"
Right (AMSType SMSJournal) -> "store_messages set to `journal`"
Left e -> e <> ", update it to `journal` in INI file"
JCExport
| 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"
ms <- newJournalMsgStore
exportMessages ms storeMsgsFilePath $ getQueueMessages False
putStrLn "Export completed"
putStrLn $ case readMsgStoreType ini of
Right (AMSType SMSMemory) -> "store_messages set to `memory`"
Right (AMSType SMSJournal) -> "store_messages set to `journal`, update it to `memory` in INI file"
Left e -> e <> ", update it to `memory` in INI file"
where
withIniFile a =
doesFileExist iniFile >>= \case
True -> readIniFile iniFile >>= either exitError a
_ -> exitError $ "Error: server is not initialized (" <> iniFile <> " does not exist).\nRun `" <> executableName <> " init`."
newJournalMsgStore = newMsgStore JournalStoreConfig {storePath = storeMsgsJournalDir, pathParts = journalMsgStoreDepth, quota = defaultMsgQueueQuota, maxMsgCount = defaultMaxJournalMsgCount, maxStateLines = defaultMaxJournalStateLines}
iniFile = combine cfgPath "smp-server.ini"
serverVersion = "SMP server v" <> simplexMQVersion
defaultServerPorts = "5223,443"
executableName = "smp-server"
storeLogFilePath = combine logPath "smp-server-store.log"
storeMsgsFilePath = combine logPath "smp-server-messages.log"
storeMsgsJournalDir = combine logPath "messages"
storeNtfsFilePath = combine logPath "smp-server-ntfs.log"
readMsgStoreType :: Ini -> Either String AMSType
readMsgStoreType = textToMsgStoreType . fromRight "memory" . lookupValue "STORE_LOG" "store_messages"
textToMsgStoreType = \case
"memory" -> Right $ AMSType SMSMemory
"journal" -> Right $ AMSType SMSJournal
s -> Left $ "invalid store_messages: " <> T.unpack s
httpsCertFile = combine cfgPath "web.crt"
httpsKeyFile = combine cfgPath "web.key"
defaultStaticPath = combine logPath "www"
@@ -128,7 +179,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
_ -> putStrLn "Invalid password. Only latin letters, digits and symbols other than '@' and ':' are allowed" >> serverPassword
checkInitOptions InitOptions {sourceCode, serverInfo, operatorCountry, hostingCountry} = do
let err_
| isNothing sourceCode && hasServerInfo serverInfo =
| isNothing sourceCode && hasServerInfo serverInfo =
Just "Error: passing any server information requires passing --source-code"
| isNothing (operator serverInfo) && isJust operatorCountry =
Just "Error: passing --operator-country requires passing --operator"
@@ -168,9 +219,12 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
\# and restoring it when the server is started.\n\
\# Log is compacted on start (deleted objects are removed).\n"
<> ("enable: " <> onOff enableStoreLog <> "\n\n")
<> "# Undelivered messages are optionally saved and restored when the server restarts,\n\
\# they are preserved in the .bak file until the next restart.\n"
<> ("restore_messages: " <> onOff enableStoreLog <> "\n")
<> "# Message storage mode: `memory` or `journal`.\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"
<> ("restore_messages: " <> onOff enableStoreLog <> "\n\n")
<> "# Messages and notifications expiration periods.\n"
<> ("expire_messages_days: " <> tshow defMsgExpirationDays <> "\n")
<> ("expire_ntfs_hours: " <> tshow defNtfExpirationHours <> "\n\n")
<> "# Log daily server statistics to CSV file\n"
@@ -248,12 +302,13 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
fp <- checkSavedFingerprint cfgPath defaultX509Config
let host = either (const "<hostnames>") T.unpack $ lookupValue "TRANSPORT" "host" ini
port = T.unpack $ strictIni "TRANSPORT" "port" ini
cfg@ServerConfig {information, storeLogFile, newQueueBasicAuth, messageExpiration, inactiveClientExpiration} = serverConfig
cfg@ServerConfig {information, storeLogFile, msgStoreType, newQueueBasicAuth, messageExpiration, inactiveClientExpiration} = serverConfig
sourceCode' = (\ServerPublicInfo {sourceCode} -> sourceCode) <$> information
srv = ProtoServerWithAuth (SMPServer [THDomainName host] (if port == "5223" then "" else port) (C.KeyHash fp)) newQueueBasicAuth
printServiceInfo serverVersion srv
printSourceCode sourceCode'
printServerConfig transports storeLogFile
checkMsgStoreMode msgStoreType
putStrLn $ case messageExpiration of
Just ExpirationConfig {ttl} -> "expiring messages after " <> showTTL ttl
_ -> "not expiring messages"
@@ -302,12 +357,16 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
_ -> enableStoreLog $> path
transports = iniTransports ini
sharedHTTP = any (\(_, _, addHTTP) -> addHTTP) transports
iniMsgStoreType = either error id $! readMsgStoreType ini
serverConfig =
ServerConfig
{ transports,
smpHandshakeTimeout = 120000000,
tbqSize = 128,
msgQueueQuota = 128,
msgStoreType = iniMsgStoreType,
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 =
@@ -318,8 +377,10 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
},
httpCredentials = (\WebHttpsParams {key, cert} -> ServerCredentials {caCertificateFile = Nothing, privateKeyFile = key, certificateFile = cert}) <$> webHttpsParams',
storeLogFile = enableStoreLog $> storeLogFilePath,
storeMsgsFile = restoreMessagesFile $ combine logPath "smp-server-messages.log",
storeNtfsFile = restoreMessagesFile $ combine logPath "smp-server-ntfs.log",
storeMsgsFile = case iniMsgStoreType of
AMSType SMSMemory -> restoreMessagesFile storeMsgsFilePath
AMSType SMSJournal -> Just storeMsgsJournalDir,
storeNtfsFile = restoreMessagesFile storeNtfsFilePath,
-- allow creating new queues by default
allowNewQueues = fromMaybe True $ iniOnOff "AUTH" "new_queues" ini,
newQueueBasicAuth = either error id <$!> strDecodeIni "AUTH" "create_password" ini,
@@ -393,6 +454,31 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
pure WebHttpsParams {port, cert, key}
webStaticPath' = eitherToMaybe $ T.unpack <$> lookupValue "WEB" "static_path" ini
checkMsgStoreMode :: AMSType -> IO ()
checkMsgStoreMode mode = do
msgsDirExists <- doesDirectoryExist storeMsgsJournalDir
msgsFileExists <- doesFileExist storeMsgsFilePath
case mode of
_ | msgsFileExists && msgsDirExists -> exitConfigureMsgStorage
AMSType SMSJournal
| 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."
AMSType SMSMemory
| 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."
exitFailure
_ -> pure ()
exitConfigureMsgStorage = do
putStrLn $ "Error: both " <> storeMsgsFilePath <> " file and " <> storeMsgsJournalDir <> " directory are present."
putStrLn "Configure memory storage."
exitFailure
data EmbeddedWebParams = EmbeddedWebParams
{ webStaticPath :: FilePath,
webHttpPort :: Maybe Int,
@@ -458,12 +544,13 @@ informationIniContent InitOptions {sourceCode, serverInfo} =
\# Hosting type can be `virtual`, `dedicated`, `colocation`, `owned`\n"
<> ("hosting_type: " <> maybe "virtual" (decodeLatin1 . strEncode) hostingType <> "\n\n")
where
ServerPublicInfo {operator, website, hosting, hostingType, serverCountry}= serverInfo
ServerPublicInfo {operator, website, hosting, hostingType, serverCountry} = serverInfo
countryStr optName country = optDisabled country <> optName <> "_country: " <> fromMaybe "ISO-3166 2-letter code" country <> "\n"
enitiyStrs optName entity =
optDisabled entity
<> optName <> ": "
<> maybe ("entity (organization or person name)") name entity
<> optName
<> ": "
<> maybe "entity (organization or person name)" name entity
<> "\n"
<> countryStr optName (country =<< entity)
@@ -519,6 +606,9 @@ data CliCommand
| OnlineCert CertOptions
| Start
| Delete
| Journal JournalCmd
data JournalCmd = JCImport | JCExport
data InitOptions = InitOptions
{ enableStoreLog :: Bool,
@@ -550,6 +640,7 @@ cliCommandP cfgPath logPath iniFile =
<> command "cert" (info (OnlineCert <$> certOptionsP) (progDesc $ "Generate new online TLS server credentials (configuration: " <> iniFile <> ")"))
<> command "start" (info (pure Start) (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"))
)
where
initP :: Parser InitOptions
@@ -607,22 +698,21 @@ cliCommandP cfgPath logPath iniFile =
<|> option strParse (long "control-port" <> help "Enable control port" <> metavar "PORT" <> value Nothing)
socksProxy <-
flag' (Just defaultSocksProxy) (long "socks-proxy" <> help "Outgoing SOCKS proxy on port 9050")
<|>
option
strParse
( long "socks-proxy"
<> metavar "PROXY"
<> help "Outgoing SOCKS proxy to forward messages to onion-only servers"
<> value Nothing
)
<|> option
strParse
( long "socks-proxy"
<> metavar "PROXY"
<> help "Outgoing SOCKS proxy to forward messages to onion-only servers"
<> value Nothing
)
ownDomains :: Maybe (L.NonEmpty TransportHost) <-
option
strParse
( long "own-domains"
<> metavar "DOMAINS"
<> help "Own server domain names (comma-separated)"
<> value Nothing
)
( long "own-domains"
<> metavar "DOMAINS"
<> help "Own server domain names (comma-separated)"
<> value Nothing
)
sourceCode <-
flag' (Just simplexmqSource) (long "source-code" <> help ("Server source code (default: " <> simplexmqSource <> ")"))
<|> (optional . strOption) (long "source-code" <> metavar "URI" <> help "Server source code")
@@ -690,6 +780,12 @@ cliCommandP cfgPath logPath iniFile =
disableWeb,
scripted
}
journalCmdP =
hsubparser
( command "import" (info (pure JCImport) (progDesc "Import message log file into a new journal storage"))
<> command "export" (info (pure JCExport) (progDesc "Export journal storage to message log file"))
)
parseBasicAuth :: ReadM ServerPassword
parseBasicAuth = eitherReader $ fmap ServerPassword . strDecode . B.pack
entityP :: String -> String -> String -> Parser (Maybe Entity, Maybe Text)

556
src/Simplex/Messaging/Server/MsgStore/Journal.hs Normal file
View File

@@ -0,0 +1,556 @@
{-# LANGUAGE BangPatterns #-}
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE DerivingStrategies #-}
{-# LANGUAGE DuplicateRecordFields #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE GeneralizedNewtypeDeriving #-}
{-# LANGUAGE InstanceSigs #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE MultiWayIf #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TypeFamilies #-}
module Simplex.Messaging.Server.MsgStore.Journal
( JournalMsgStore (msgQueues, random),
JournalMsgQueue,
JournalStoreConfig (..),
getQueueMessages,
-- below are exported for tests
MsgQueueState (..),
JournalState (..),
msgQueueDirectory,
readWriteQueueState,
newMsgQueueState,
newJournalId,
logQueueState,
queueLogFileName,
logFileExt,
)
where
import Control.Concurrent.STM
import qualified Control.Exception as E
import Control.Logger.Simple
import Control.Monad
import Control.Monad.Trans.Except
import qualified Data.Attoparsec.ByteString.Char8 as A
import Data.ByteString.Char8 (ByteString)
import qualified Data.ByteString.Char8 as B
import qualified Data.ByteString.Lazy.Char8 as LB
import Data.Functor (($>))
import Data.Maybe (catMaybes, fromMaybe)
import qualified Data.Text as T
import Data.Time.Clock (getCurrentTime)
import Data.Time.Format.ISO8601 (iso8601Show)
import GHC.IO (catchAny)
import Simplex.Messaging.Agent.Client (getMapLock, withLockMap)
import Simplex.Messaging.Agent.Lock
import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Protocol (ErrorType (..), Message (..), RecipientId)
import Simplex.Messaging.Server.MsgStore.Types
import Simplex.Messaging.TMap (TMap)
import qualified Simplex.Messaging.TMap as TM
import Simplex.Messaging.Util (ifM, tshow, ($>>=))
import System.Directory
import System.FilePath ((</>))
import System.IO (BufferMode (..), Handle, IOMode (..), SeekMode (..), stdout)
import qualified System.IO as IO
import System.Random (StdGen, genByteString, newStdGen)
data JournalMsgStore = JournalMsgStore
{ config :: JournalStoreConfig,
random :: TVar StdGen,
queueLocks :: TMap RecipientId Lock,
msgQueues :: TMap RecipientId JournalMsgQueue
}
data JournalStoreConfig = JournalStoreConfig
{ storePath :: FilePath,
pathParts :: Int,
quota :: Int,
-- Max number of messages per journal file - ignored in STM store.
-- When this limit is reached, the file will be changed.
-- This number should be set bigger than queue quota.
maxMsgCount :: Int,
maxStateLines :: Int
}
data JournalMsgQueue = JournalMsgQueue
{ config :: JournalStoreConfig,
queueDirectory :: FilePath,
queueLock :: Lock,
state :: TVar MsgQueueState,
-- Last message and length incl. newline
-- Nothing - unknown, Just Nothing - empty queue.
-- This optimization prevents reading each message at least twice,
-- or reading it after it was just written.
tipMsg :: TVar (Maybe (Maybe (Message, Int))),
handles :: TVar (Maybe MsgQueueHandles),
random :: TVar StdGen
}
data MsgQueueState = MsgQueueState
{ writeState :: JournalState,
readState :: JournalState,
canWrite :: Bool,
size :: Int
}
deriving (Show)
data MsgQueueHandles = MsgQueueHandles
{ stateHandle :: Handle, -- handle to queue state log file, rotates and removes old backups when server is restarted
readHandle :: Handle,
writeHandle :: Maybe Handle -- optional, used when write file is different from read file
}
data JournalState = JournalState
{ journalId :: ByteString,
msgPos :: Int,
msgCount :: Int,
bytePos :: Int
}
deriving (Show)
newMsgQueueState :: ByteString -> MsgQueueState
newMsgQueueState journalId =
let st = newJournalState journalId
in MsgQueueState {writeState = st, readState = st, canWrite = True, size = 0}
newJournalState :: ByteString -> JournalState
newJournalState journalId = JournalState {journalId, msgPos = 0, msgCount = 0, bytePos = 0}
journalFilePath :: FilePath -> ByteString -> FilePath
journalFilePath dir journalId = dir </> (msgLogFileName <> "." <> B.unpack journalId <> logFileExt)
instance StrEncoding MsgQueueState where
strEncode MsgQueueState {writeState, readState, canWrite, size} =
B.unwords
[ "write=" <> strEncode writeState,
"read=" <> strEncode readState,
"canWrite=" <> strEncode canWrite,
"size=" <> strEncode size
]
strP = do
writeState <- "write=" *> strP
readState <- " read=" *> strP
canWrite <- " canWrite=" *> strP
size <- " size=" *> strP
pure MsgQueueState {writeState, readState, canWrite, size}
instance StrEncoding JournalState where
strEncode JournalState {journalId, msgPos, msgCount, bytePos} =
B.intercalate "," [journalId, strEncode msgPos, strEncode msgCount, strEncode bytePos]
strP = do
journalId <- A.takeTill (== ',')
msgPos <- A.char ',' *> strP
msgCount <- A.char ',' *> strP
bytePos <- A.char ',' *> strP
pure JournalState {journalId, msgPos, msgCount, bytePos}
queueLogFileName :: String
queueLogFileName = "queue_state"
msgLogFileName :: String
msgLogFileName = "messages"
logFileExt :: String
logFileExt = ".log"
newtype NonAtomicIO a = NonAtomicIO (IO a)
deriving newtype (Functor, Applicative, Monad)
instance MsgStoreClass JournalMsgStore where
type StoreMonad JournalMsgStore = NonAtomicIO
type MsgQueue JournalMsgStore = JournalMsgQueue
type MsgStoreConfig JournalMsgStore = JournalStoreConfig
newMsgStore :: JournalStoreConfig -> IO JournalMsgStore
newMsgStore config = do
random <- newTVarIO =<< newStdGen
queueLocks <- TM.emptyIO
msgQueues <- TM.emptyIO
pure JournalMsgStore {config, random, queueLocks, msgQueues}
closeMsgStore st = readTVarIO (msgQueues st) >>= mapM_ closeMsgQueue_
activeMsgQueues = msgQueues
{-# INLINE activeMsgQueues #-}
-- This function opens and closes all queues.
-- It is used to export storage to a single file, not during normal server execution.
withAllMsgQueues :: JournalMsgStore -> (RecipientId -> JournalMsgQueue -> IO Int) -> IO Int
withAllMsgQueues st@JournalMsgStore {config} action = do
closeMsgStore st
lock <- createLockIO -- the same lock is used for all queues
dirs <- zip [0..] <$> listQueueDirs 0 ("", storePath)
let count = length dirs
total <- foldM (processQueue lock count) 0 dirs
progress count count
putStrLn ""
pure total
where
JournalStoreConfig {storePath, pathParts} = config
processQueue lock count !total (i :: Int, (queueId, dir)) = do
when (i `mod` 100 == 0) $ progress i count
q <- openMsgQueue st dir lock
total' <- case strDecode $ B.pack queueId of
Right rId -> (total +) <$> action rId q
Left e -> total <$ putStrLn ("Error: message queue directory " <> dir <> " is invalid: " <> e)
closeMsgQueue_ q
pure total'
progress i count = do
putStr $ "Processed: " <> show i <> "/" <> show count <> " queues\r"
IO.hFlush stdout
listQueueDirs depth (queueId, path)
| depth == pathParts - 1 = listDirs
| otherwise = fmap concat . mapM (listQueueDirs (depth + 1)) =<< listDirs
where
listDirs = fmap catMaybes . mapM queuePath =<< listDirectory path
queuePath dir = do
let path' = path </> dir
ifM
(doesDirectoryExist path')
(pure $ Just (queueId <> dir, path'))
(Nothing <$ putStrLn ("Error: path " <> path' <> " is not a directory, skipping"))
logQueueStates :: JournalMsgStore -> IO ()
logQueueStates st =
void $ withActiveMsgQueues st $ \_ q ->
readTVarIO (handles q)
>>= maybe (pure ()) (\hs -> readTVarIO (state q) >>= logQueueState (stateHandle hs))
>> pure 0
getMsgQueue :: JournalMsgStore -> RecipientId -> ExceptT ErrorType IO JournalMsgQueue
getMsgQueue store@JournalMsgStore {queueLocks, msgQueues, random} rId =
tryStore "getMsgQueue" $ withLockMap queueLocks rId "getMsgQueue" $
TM.lookupIO rId msgQueues >>= maybe newQ pure
where
newQ = do
let dir = msgQueueDirectory store rId
queueLock <- atomically $ getMapLock queueLocks rId
q <- ifM (doesDirectoryExist dir) (openMsgQueue store dir queueLock) (createQ dir queueLock)
atomically $ TM.insert rId q msgQueues
pure q
where
createQ :: FilePath -> Lock -> IO JournalMsgQueue
createQ dir queueLock = do
-- folder and files are not created here,
-- to avoid file IO for queues without messages during subscription
journalId <- newJournalId random
mkJournalQueue store dir queueLock (newMsgQueueState journalId, Nothing)
delMsgQueue :: JournalMsgStore -> RecipientId -> IO ()
delMsgQueue st rId = withLockMap (queueLocks st) rId "delMsgQueue" $ do
void $ closeMsgQueue st rId
removeQueueDirectory st rId
delMsgQueueSize :: JournalMsgStore -> RecipientId -> IO Int
delMsgQueueSize st rId = withLockMap (queueLocks st) rId "delMsgQueue" $ do
state_ <- closeMsgQueue st rId
sz <- maybe (pure $ -1) (fmap size . readTVarIO) state_
removeQueueDirectory st rId
pure sz
getQueueMessages :: Bool -> JournalMsgQueue -> IO [Message]
getQueueMessages drainMsgs q = readTVarIO (handles q) >>= maybe (pure []) (getMsg [])
where
getMsg ms hs = chooseReadJournal q drainMsgs hs >>= maybe (pure ms) readMsg
where
readMsg (rs, h) = do
-- TODO handle errors
s <- hGetLineAt h $ bytePos rs
-- TODO handle errors
Right msg <- pure $ strDecode s
updateReadPos q drainMsgs (B.length s + 1) hs -- 1 is to account for new line
(msg :) <$> getMsg ms hs
writeMsg :: JournalMsgQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
writeMsg q@JournalMsgQueue {queueDirectory, handles, config, random} logState !msg =
tryStore "writeMsg" $ withLock' (queueLock q) "writeMsg" $ do
st@MsgQueueState {canWrite, size} <- readTVarIO (state q)
let empty = size == 0
if canWrite || empty
then do
let canWrt' = quota > size
if canWrt'
then writeToJournal st canWrt' msg $> Just (msg, empty)
else writeToJournal st canWrt' msgQuota $> Nothing
else pure Nothing
where
JournalStoreConfig {quota, maxMsgCount} = config
msgQuota = MessageQuota {msgId = msgId msg, msgTs = msgTs msg}
writeToJournal st@MsgQueueState {writeState, readState = rs, size} canWrt' msg' = do
let msgStr = strEncode msg' `B.snoc` '\n'
msgLen = B.length msgStr
hs <- maybe createQueueDir pure =<< readTVarIO handles
(ws, wh) <- case writeHandle hs of
Nothing | msgCount writeState >= maxMsgCount -> switchWriteJournal hs
wh_ -> pure (writeState, fromMaybe (readHandle hs) wh_)
let msgCount' = msgCount ws + 1
ws' = ws {msgPos = msgPos ws + 1, msgCount = msgCount', bytePos = bytePos ws + msgLen}
rs' = if journalId ws == journalId rs then rs {msgCount = msgCount'} else rs
!st' = st {writeState = ws', readState = rs', canWrite = canWrt', size = size + 1}
when (size == 0) $ atomically $ writeTVar (tipMsg q) $ Just (Just (msg, msgLen))
hAppend wh msgStr
updateQueueState q logState hs st'
where
createQueueDir = do
createDirectoryIfMissing True queueDirectory
let statePath = queueDirectory </> (queueLogFileName <> logFileExt)
sh <- openFile statePath AppendMode
B.hPutStr sh ""
rh <- createNewJournal queueDirectory $ journalId rs
let hs = MsgQueueHandles {stateHandle = sh, readHandle = rh, writeHandle = Nothing}
atomically $ writeTVar handles $ Just hs
pure hs
switchWriteJournal hs = do
journalId <- newJournalId random
wh <- createNewJournal queueDirectory journalId
atomically $ writeTVar handles $ Just $ hs {writeHandle = Just wh}
pure (newJournalState journalId, wh)
getQueueSize :: JournalMsgQueue -> IO Int
getQueueSize JournalMsgQueue {state} = size <$> readTVarIO state
tryPeekMsg_ :: JournalMsgQueue -> NonAtomicIO (Maybe Message)
tryPeekMsg_ q@JournalMsgQueue {tipMsg, handles} =
NonAtomicIO $ readTVarIO handles $>>= chooseReadJournal q True $>>= peekMsg
where
peekMsg (rs, h) = readTVarIO tipMsg >>= maybe readMsg (pure . fmap fst)
where
readMsg = do
-- TODO handle errors
s <- hGetLineAt h $ bytePos rs
-- TODO handle errors
Right msg <- pure $ strDecode s
atomically $ writeTVar tipMsg $ Just (Just (msg, B.length s + 1)) -- 1 is to account for new line
pure $ Just msg
tryDeleteMsg_ :: JournalMsgQueue -> NonAtomicIO ()
tryDeleteMsg_ q@JournalMsgQueue {tipMsg, handles} = NonAtomicIO $
void $
readTVarIO tipMsg -- if there is no cached tipMsg, do nothing
$>>= (pure . fmap snd)
$>>= \len -> readTVarIO handles
$>>= \hs -> updateReadPos q True len hs $> Just ()
atomicQueue :: JournalMsgQueue -> String -> NonAtomicIO a -> ExceptT ErrorType IO a
atomicQueue mq op (NonAtomicIO a) = tryStore op $ withLock' (queueLock mq) op $ a
tryStore :: String -> IO a -> ExceptT ErrorType IO a
tryStore op a =
ExceptT $
(Right <$> a) `catchAny` \e ->
let e' = op <> " " <> show e
in logError ("STORE ERROR " <> T.pack e') $> Left (STORE e')
openMsgQueue :: JournalMsgStore -> FilePath -> Lock -> IO JournalMsgQueue
openMsgQueue store dir queueLock = do
let statePath = dir </> (queueLogFileName <> logFileExt)
(st, sh) <- readWriteQueueState store statePath
(st', rh, wh_) <- openJournals dir st
let hs = MsgQueueHandles {stateHandle = sh, readHandle = rh, writeHandle = wh_}
mkJournalQueue store dir queueLock (st', Just hs)
mkJournalQueue :: JournalMsgStore -> FilePath -> Lock -> (MsgQueueState, Maybe MsgQueueHandles) -> IO JournalMsgQueue
mkJournalQueue JournalMsgStore {random, config} dir queueLock (st, hs_) = do
state <- newTVarIO st
tipMsg <- newTVarIO Nothing
handles <- newTVarIO hs_
-- using the same queue lock which is currently locked,
-- to avoid map lookup on queue operations
pure
JournalMsgQueue
{ config,
queueDirectory = dir,
queueLock,
state,
tipMsg,
handles,
random
}
chooseReadJournal :: JournalMsgQueue -> Bool -> MsgQueueHandles -> IO (Maybe (JournalState, Handle))
chooseReadJournal q log' hs = do
st@MsgQueueState {writeState = ws, readState = rs} <- readTVarIO (state q)
case writeHandle hs of
Just wh | msgPos rs >= msgCount rs && journalId rs /= journalId ws -> do
-- switching to write journal
atomically $ writeTVar (handles q) $ Just hs {readHandle = wh, writeHandle = Nothing}
hClose $ readHandle hs
removeJournal (queueDirectory q) rs
let !rs' = (newJournalState $ journalId ws) {msgCount = msgCount ws}
!st' = st {readState = rs'}
updateQueueState q log' hs st'
pure $ Just (rs', wh)
_ | msgPos rs >= msgCount rs && journalId rs == journalId ws -> pure Nothing
_ -> pure $ Just (rs, readHandle hs)
updateQueueState :: JournalMsgQueue -> Bool -> MsgQueueHandles -> MsgQueueState -> IO ()
updateQueueState q log' hs st = do
atomically $ writeTVar (state q) st
when log' $ logQueueState (stateHandle hs) st
logQueueState :: Handle -> MsgQueueState -> IO ()
logQueueState h st = B.hPutStr h $ strEncode st `B.snoc` '\n'
updateReadPos :: JournalMsgQueue -> Bool -> Int -> MsgQueueHandles -> IO ()
updateReadPos q log' len hs = do
st@MsgQueueState {readState = rs, size} <- readTVarIO (state q)
let JournalState {msgPos, bytePos} = rs
let msgPos' = msgPos + 1
rs' = rs {msgPos = msgPos', bytePos = bytePos + len}
st' = st {readState = rs', size = size - 1}
updateQueueState q log' hs st'
atomically $ writeTVar (tipMsg q) Nothing
msgQueueDirectory :: JournalMsgStore -> RecipientId -> FilePath
msgQueueDirectory JournalMsgStore {config = JournalStoreConfig {storePath, pathParts}} rId =
storePath </> B.unpack (B.intercalate "/" $ splitSegments pathParts $ strEncode rId)
where
splitSegments _ "" = []
splitSegments 1 s = [s]
splitSegments n s =
let (seg, s') = B.splitAt 2 s
in seg : splitSegments (n - 1) s'
createNewJournal :: FilePath -> ByteString -> IO Handle
createNewJournal dir journalId = do
let path = journalFilePath dir journalId -- TODO retry if file exists
h <- openFile path ReadWriteMode
B.hPutStr h ""
pure h
newJournalId :: TVar StdGen -> IO ByteString
newJournalId g = strEncode <$> atomically (stateTVar g $ genByteString 12)
openJournals :: FilePath -> MsgQueueState -> IO (MsgQueueState, Handle, Maybe Handle)
openJournals dir st@MsgQueueState {readState = rs, writeState = ws} = do
-- TODO verify that file exists, what to do if it's not, or if its state diverges
-- TODO check current position matches state, fix if not
let rjId = journalId rs
wjId = journalId ws
openJournal rs >>= \case
Left path -> do
logError $ "STORE ERROR no read file " <> T.pack path <> ", creating new file"
rh <- createNewJournal dir rjId
let st' = newMsgQueueState rjId
pure (st', rh, Nothing)
Right rh
| rjId == wjId -> do
st' <- fixWriteFileSize rh
pure (st', rh, Nothing)
| otherwise ->
openJournal ws >>= \case
Left path -> do
logError $ "STORE ERROR no write file " <> T.pack path <> ", creating new file"
wh <- createNewJournal dir wjId
let size' = msgCount rs + msgPos rs
st' = st {writeState = newJournalState wjId, size = size'} -- we don't amend canWrite to trigger QCONT
pure (st', rh, Just wh)
Right wh -> do
st' <- fixWriteFileSize wh
pure (st', rh, Just wh)
where
openJournal JournalState {journalId} =
let path = journalFilePath dir journalId
in ifM (doesFileExist path) (Right <$> openFile path ReadWriteMode) (pure $ Left path)
fixWriteFileSize h = do
let sz = fromIntegral $ bytePos ws
sz' <- IO.hFileSize h
if
| sz' > sz -> logWarn "STORE WARNING" >> IO.hSetFileSize h sz $> st
| sz' == sz -> pure st
| otherwise -> pure st -- TODO re-read file to recover what is possible ???
removeJournal :: FilePath -> JournalState -> IO ()
removeJournal dir JournalState {journalId} = do
let path = journalFilePath dir journalId
removeFile path `catchAny` (\e -> logError $ "STORE ERROR removing file " <> T.pack path <> ": " <> tshow e)
-- This function is supposed to be resilient to crashes while updating state files,
-- and also resilient to crashes during its execution.
readWriteQueueState :: JournalMsgStore -> FilePath -> IO (MsgQueueState, Handle)
readWriteQueueState JournalMsgStore {random, config} statePath =
ifM
(doesFileExist tempBackup)
(renameFile tempBackup statePath >> readQueueState)
(ifM (doesFileExist statePath) readQueueState writeNewQueueState)
where
tempBackup = statePath <> ".bak"
readQueueState = do
ls <- LB.lines <$> LB.readFile statePath
case ls of
[] -> writeNewQueueState
_ -> useLastLine (length ls) True ls
writeNewQueueState = do
logWarn $ "STORE WARNING: empty queue state in " <> T.pack statePath <> ", initialized"
st <- newMsgQueueState <$> newJournalId random
writeQueueState st
useLastLine len isLastLine ls = case strDecode $ LB.toStrict $ last ls of
Right st
| len > maxStateLines config || not isLastLine ->
backupWriteQueueState st
| otherwise -> do
-- when state file has fewer than maxStateLines, we don't compact it
sh <- openFile statePath AppendMode
pure (st, sh)
Left e -- if the last line failed to parse
| isLastLine -> case init ls of -- or use the previous line
[] -> do
logWarn $ "STORE WARNING: invalid 1-line queue state " <> T.pack statePath <> ", initialized"
st <- newMsgQueueState <$> newJournalId random
backupWriteQueueState st
ls' -> do
logWarn $ "STORE WARNING: invalid last line in queue state " <> T.pack statePath <> ", using the previous line"
useLastLine len False ls'
| otherwise -> do
logError $ "STORE ERROR invalid queue state in " <> T.pack statePath <> ": " <> tshow e
E.throwIO $ userError $ "Error reading queue state " <> statePath <> ": " <> show e
backupWriteQueueState st = do
-- State backup is made in two steps to mitigate the crash during the backup.
-- Temporary backup file will be used when it is present.
renameFile statePath tempBackup -- 1) temp backup
r <- writeQueueState st -- 2) save state
ts <- getCurrentTime
renameFile tempBackup (statePath <> "." <> iso8601Show ts <> ".bak") -- 3) timed backup
pure r
writeQueueState st = do
sh <- openFile statePath AppendMode
logQueueState sh st
pure (st, sh)
closeMsgQueue :: JournalMsgStore -> RecipientId -> IO (Maybe (TVar MsgQueueState))
closeMsgQueue st rId =
atomically (TM.lookupDelete rId (msgQueues st))
>>= mapM (\q -> closeMsgQueue_ q $> state q)
closeMsgQueue_ :: JournalMsgQueue -> IO ()
closeMsgQueue_ q = readTVarIO (handles q) >>= mapM_ closeHandles
where
closeHandles (MsgQueueHandles sh rh wh_) = do
hClose sh
hClose rh
mapM_ hClose wh_
removeQueueDirectory :: JournalMsgStore -> RecipientId -> IO ()
removeQueueDirectory st rId =
let dir = msgQueueDirectory st rId
in removePathForcibly dir `catchAny` (\e -> logError $ "STORE ERROR removeQueueDirectory " <> T.pack dir <> ": " <> tshow e)
hAppend :: Handle -> ByteString -> IO ()
hAppend h s = IO.hSeek h SeekFromEnd 0 >> B.hPutStr h s
hGetLineAt :: Handle -> Int -> IO ByteString
hGetLineAt h pos = IO.hSeek h AbsoluteSeek (fromIntegral pos) >> B.hGetLine h
openFile :: FilePath -> IOMode -> IO Handle
openFile f mode = do
h <- IO.openFile f mode
IO.hSetBuffering h LineBuffering
pure h
hClose :: Handle -> IO ()
hClose h = IO.hClose h `catchAny` (\e -> logError $ "Error closing file" <> tshow e)

187
src/Simplex/Messaging/Server/MsgStore/STM.hs
View File

@@ -1,129 +1,126 @@
{-# LANGUAGE BangPatterns #-}
{-# LANGUAGE ConstraintKinds #-}
{-# LANGUAGE DuplicateRecordFields #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE FlexibleInstances #-}
{-# LANGUAGE InstanceSigs #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE MultiParamTypeClasses #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE TupleSections #-}
{-# LANGUAGE TypeFamilies #-}
module Simplex.Messaging.Server.MsgStore.STM
( STMMsgStore,
MsgQueue (msgQueue),
newMsgStore,
getMsgQueue,
delMsgQueue,
delMsgQueueSize,
writeMsg,
tryPeekMsg,
tryPeekMsgIO,
tryDelMsg,
tryDelPeekMsg,
deleteExpiredMsgs,
getQueueSize,
( STMMsgStore (..),
STMMsgQueue (msgQueue),
STMStoreConfig (..),
)
where
import qualified Data.ByteString.Char8 as B
import Control.Concurrent.STM
import Control.Monad.IO.Class
import Control.Monad.Trans.Except
import Data.Functor (($>))
import Data.Int (Int64)
import Data.Time.Clock.System (SystemTime (systemSeconds))
import Simplex.Messaging.Protocol (Message (..), MsgId, RecipientId)
import Simplex.Messaging.Protocol (ErrorType, Message (..), RecipientId)
import Simplex.Messaging.Server.MsgStore.Types
import Simplex.Messaging.TMap (TMap)
import qualified Simplex.Messaging.TMap as TM
import UnliftIO.STM
data MsgQueue = MsgQueue
data STMMsgQueue = STMMsgQueue
{ msgQueue :: TQueue Message,
quota :: Int,
canWrite :: TVar Bool,
size :: TVar Int
}
type STMMsgStore = TMap RecipientId MsgQueue
data STMMsgStore = STMMsgStore
{ storeConfig :: STMStoreConfig,
msgQueues :: TMap RecipientId STMMsgQueue
}
newMsgStore :: IO STMMsgStore
newMsgStore = TM.emptyIO
data STMStoreConfig = STMStoreConfig
{ storePath :: Maybe FilePath,
quota :: Int
}
-- The reason for double lookup is that majority of messaging queues exist,
-- because multiple messages are sent to the same queue,
-- so the first lookup without STM transaction will return the queue faster.
-- In case the queue does not exist, it needs to be looked-up again inside transaction.
getMsgQueue :: STMMsgStore -> RecipientId -> Int -> IO MsgQueue
getMsgQueue st rId quota = TM.lookupIO rId st >>= maybe (atomically maybeNewQ) pure
where
maybeNewQ = TM.lookup rId st >>= maybe newQ pure
newQ = do
msgQueue <- newTQueue
canWrite <- newTVar True
size <- newTVar 0
let q = MsgQueue {msgQueue, quota, canWrite, size}
TM.insert rId q st
pure q
instance MsgStoreClass STMMsgStore where
type StoreMonad STMMsgStore = STM
type MsgQueue STMMsgStore = STMMsgQueue
type MsgStoreConfig STMMsgStore = STMStoreConfig
delMsgQueue :: STMMsgStore -> RecipientId -> IO ()
delMsgQueue st rId = atomically $ TM.delete rId st
newMsgStore :: STMStoreConfig -> IO STMMsgStore
newMsgStore storeConfig = do
msgQueues <- TM.emptyIO
pure STMMsgStore {storeConfig, msgQueues}
delMsgQueueSize :: STMMsgStore -> RecipientId -> IO Int
delMsgQueueSize st rId = atomically (TM.lookupDelete rId st) >>= maybe (pure 0) (\MsgQueue {size} -> readTVarIO size)
closeMsgStore _ = pure ()
writeMsg :: MsgQueue -> Message -> IO (Maybe (Message, Bool))
writeMsg MsgQueue {msgQueue = q, quota, canWrite, size} !msg = atomically $ do
canWrt <- readTVar canWrite
empty <- isEmptyTQueue q
if canWrt || empty
then do
canWrt' <- (quota >) <$> readTVar size
writeTVar canWrite $! canWrt'
modifyTVar' size (+ 1)
if canWrt'
then writeTQueue q msg $> Just (msg, empty)
else (writeTQueue q $! msgQuota) $> Nothing
else pure Nothing
where
msgQuota = MessageQuota {msgId = msgId msg, msgTs = msgTs msg}
activeMsgQueues = msgQueues
{-# INLINE activeMsgQueues #-}
tryPeekMsgIO :: MsgQueue -> IO (Maybe Message)
tryPeekMsgIO = atomically . tryPeekTQueue . msgQueue
{-# INLINE tryPeekMsgIO #-}
withAllMsgQueues = withActiveMsgQueues
{-# INLINE withAllMsgQueues #-}
-- TODO remove once deliverToSub is split
tryPeekMsg :: MsgQueue -> STM (Maybe Message)
tryPeekMsg = tryPeekTQueue . msgQueue
{-# INLINE tryPeekMsg #-}
logQueueStates _ = pure ()
tryDelMsg :: MsgQueue -> MsgId -> IO (Maybe Message)
tryDelMsg mq msgId' = atomically $
tryPeekMsg mq >>= \case
msg_@(Just msg)
| msgId msg == msgId' || B.null msgId' -> tryDeleteMsg_ mq >> pure msg_
| otherwise -> pure Nothing
_ -> pure Nothing
-- The reason for double lookup is that majority of messaging queues exist,
-- because multiple messages are sent to the same queue,
-- so the first lookup without STM transaction will return the queue faster.
-- In case the queue does not exist, it needs to be looked-up again inside transaction.
getMsgQueue :: STMMsgStore -> RecipientId -> ExceptT ErrorType IO STMMsgQueue
getMsgQueue STMMsgStore {msgQueues = qs, storeConfig = STMStoreConfig {quota}} rId =
liftIO $ TM.lookupIO rId qs >>= maybe (atomically maybeNewQ) pure
where
maybeNewQ = TM.lookup rId qs >>= maybe newQ pure
newQ = do
msgQueue <- newTQueue
canWrite <- newTVar True
size <- newTVar 0
let q = STMMsgQueue {msgQueue, quota, canWrite, size}
TM.insert rId q qs
pure q
-- atomic delete (== read) last and peek next message if available
tryDelPeekMsg :: MsgQueue -> MsgId -> IO (Maybe Message, Maybe Message)
tryDelPeekMsg mq msgId' = atomically $
tryPeekMsg mq >>= \case
msg_@(Just msg)
| msgId msg == msgId' || B.null msgId' -> (msg_,) <$> (tryDeleteMsg_ mq >> tryPeekMsg mq)
| otherwise -> pure (Nothing, msg_)
_ -> pure (Nothing, Nothing)
delMsgQueue :: STMMsgStore -> RecipientId -> IO ()
delMsgQueue st rId = atomically $ TM.delete rId $ msgQueues st
deleteExpiredMsgs :: MsgQueue -> Int64 -> IO Int
deleteExpiredMsgs mq old = atomically $ loop 0
where
loop dc =
tryPeekMsg mq >>= \case
Just Message {msgTs}
| systemSeconds msgTs < old ->
tryDeleteMsg_ mq >> loop (dc + 1)
_ -> pure dc
delMsgQueueSize :: STMMsgStore -> RecipientId -> IO Int
delMsgQueueSize st rId = atomically (TM.lookupDelete rId $ msgQueues st) >>= maybe (pure 0) (\STMMsgQueue {size} -> readTVarIO size)
tryDeleteMsg_ :: MsgQueue -> STM ()
tryDeleteMsg_ MsgQueue {msgQueue = q, size} =
tryReadTQueue q >>= \case
Just _ -> modifyTVar' size (subtract 1)
_ -> pure ()
getQueueMessages :: Bool -> STMMsgQueue -> IO [Message]
getQueueMessages drainMsgs = atomically . (if drainMsgs then flushTQueue else snapshotTQueue) . msgQueue
where
snapshotTQueue q = do
msgs <- flushTQueue q
mapM_ (writeTQueue q) msgs
pure msgs
getQueueSize :: MsgQueue -> IO Int
getQueueSize MsgQueue {size} = readTVarIO size
writeMsg :: STMMsgQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
writeMsg STMMsgQueue {msgQueue = q, quota, canWrite, size} _logState !msg = liftIO $ atomically $ do
canWrt <- readTVar canWrite
empty <- isEmptyTQueue q
if canWrt || empty
then do
canWrt' <- (quota >) <$> readTVar size
writeTVar canWrite $! canWrt'
modifyTVar' size (+ 1)
if canWrt'
then writeTQueue q msg $> Just (msg, empty)
else (writeTQueue q $! msgQuota) $> Nothing
else pure Nothing
where
msgQuota = MessageQuota {msgId = msgId msg, msgTs = msgTs msg}
getQueueSize :: STMMsgQueue -> IO Int
getQueueSize STMMsgQueue {size} = readTVarIO size
tryPeekMsg_ :: STMMsgQueue -> STM (Maybe Message)
tryPeekMsg_ = tryPeekTQueue . msgQueue
{-# INLINE tryPeekMsg_ #-}
tryDeleteMsg_ :: STMMsgQueue -> STM ()
tryDeleteMsg_ STMMsgQueue {msgQueue = q, size} =
tryReadTQueue q >>= \case
Just _ -> modifyTVar' size (subtract 1)
_ -> pure ()
atomicQueue :: STMMsgQueue -> String -> STM a -> ExceptT ErrorType IO a
atomicQueue _ _ = liftIO . atomically

81
src/Simplex/Messaging/Server/MsgStore/Types.hs Normal file
View File

@@ -0,0 +1,81 @@
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE TupleSections #-}
{-# LANGUAGE TypeFamilyDependencies #-}
module Simplex.Messaging.Server.MsgStore.Types where
import Control.Concurrent.STM
import Control.Monad.Trans.Except
import Data.Int (Int64)
import Data.Kind
import qualified Data.Map.Strict as M
import Data.Time.Clock.System (SystemTime (systemSeconds))
import Simplex.Messaging.Protocol (ErrorType, Message (..), MsgId, RecipientId)
import Simplex.Messaging.TMap (TMap)
class Monad (StoreMonad s) => MsgStoreClass s where
type StoreMonad s = (m :: Type -> Type) | m -> s
type MsgStoreConfig s = c | c -> s
type MsgQueue s = q | q -> s
newMsgStore :: MsgStoreConfig s -> IO s
closeMsgStore :: s -> IO ()
activeMsgQueues :: s -> TMap RecipientId (MsgQueue s)
withAllMsgQueues :: s -> (RecipientId -> MsgQueue s -> IO Int) -> IO Int
logQueueStates :: s -> IO ()
getMsgQueue :: s -> RecipientId -> ExceptT ErrorType IO (MsgQueue s)
delMsgQueue :: s -> RecipientId -> IO ()
delMsgQueueSize :: s -> RecipientId -> IO Int
getQueueMessages :: Bool -> MsgQueue s -> IO [Message]
writeMsg :: MsgQueue s -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
getQueueSize :: MsgQueue s -> IO Int
tryPeekMsg_ :: MsgQueue s -> StoreMonad s (Maybe Message)
tryDeleteMsg_ :: MsgQueue s -> StoreMonad s ()
atomicQueue :: MsgQueue s -> String -> StoreMonad s a -> ExceptT ErrorType IO a
data MSType = MSMemory | MSJournal
data SMSType :: MSType -> Type where
SMSMemory :: SMSType 'MSMemory
SMSJournal :: SMSType 'MSJournal
data AMSType = forall s. AMSType (SMSType s)
withActiveMsgQueues :: MsgStoreClass s => s -> (RecipientId -> MsgQueue s -> IO Int) -> IO Int
withActiveMsgQueues st f = readTVarIO (activeMsgQueues st) >>= M.foldrWithKey (\k v -> ((+) <$> f k v <*>)) (pure 0)
tryPeekMsg :: MsgStoreClass s => MsgQueue s -> ExceptT ErrorType IO (Maybe Message)
tryPeekMsg mq = atomicQueue mq "tryPeekMsg" $ tryPeekMsg_ mq
{-# INLINE tryPeekMsg #-}
tryDelMsg :: MsgStoreClass s => MsgQueue s -> MsgId -> ExceptT ErrorType IO (Maybe Message)
tryDelMsg mq msgId' =
atomicQueue mq "tryDelMsg" $
tryPeekMsg_ mq >>= \case
msg_@(Just msg)
| msgId msg == msgId' ->
tryDeleteMsg_ mq >> pure msg_
_ -> pure Nothing
-- atomic delete (== read) last and peek next message if available
tryDelPeekMsg :: MsgStoreClass s => MsgQueue s -> MsgId -> ExceptT ErrorType IO (Maybe Message, Maybe Message)
tryDelPeekMsg mq msgId' =
atomicQueue mq "tryDelPeekMsg" $
tryPeekMsg_ mq >>= \case
msg_@(Just msg)
| msgId msg == msgId' -> (msg_,) <$> (tryDeleteMsg_ mq >> tryPeekMsg_ mq)
| otherwise -> pure (Nothing, msg_)
_ -> pure (Nothing, Nothing)
deleteExpiredMsgs :: MsgStoreClass s => MsgQueue s -> Int64 -> ExceptT ErrorType IO Int
deleteExpiredMsgs mq old = atomicQueue mq "deleteExpiredMsgs" $ loop 0
where
loop dc =
tryPeekMsg_ mq >>= \case
Just Message {msgTs}
| systemSeconds msgTs < old ->
tryDeleteMsg_ mq >> loop (dc + 1)
_ -> pure dc

8
src/Simplex/Messaging/Util.hs
View File

@@ -14,11 +14,13 @@ import Data.Bifunctor (first)
import Data.ByteString.Char8 (ByteString)
import qualified Data.ByteString.Char8 as B
import qualified Data.ByteString.Lazy.Char8 as LB
import Data.Int (Int64)
import Data.IORef
import Data.Int (Int64)
import Data.List (groupBy, sortOn)
import Data.List.NonEmpty (NonEmpty)
import qualified Data.List.NonEmpty as L
import Data.Map.Strict (Map)
import qualified Data.Map.Strict as M
import Data.Text (Text)
import qualified Data.Text as T
import Data.Text.Encoding (decodeUtf8With, encodeUtf8)
@@ -185,3 +187,7 @@ encodeJSON = safeDecodeUtf8 . LB.toStrict . J.encode
decodeJSON :: FromJSON a => Text -> Maybe a
decodeJSON = J.decode . LB.fromStrict . encodeUtf8
traverseWithKey_ :: Monad m => (k -> v -> m ()) -> Map k v -> m ()
traverseWithKey_ f = M.foldrWithKey (\k v -> (f k v >>)) (pure ())
{-# INLINE traverseWithKey_ #-}

21
tests/AgentTests/FunctionalAPITests.hs
View File

@@ -77,7 +77,7 @@ import Data.Word (Word16)
import qualified Database.SQLite.Simple as SQL
import GHC.Stack (withFrozenCallStack)
import SMPAgentClient
import SMPClient (cfg, prevRange, prevVersion, testPort, testPort2, testStoreLogFile2, withSmpServer, withSmpServerConfigOn, withSmpServerOn, withSmpServerProxy, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn)
import SMPClient (cfg, prevRange, prevVersion, testPort, testPort2, testStoreLogFile2, testStoreMsgsDir2, withSmpServer, withSmpServerConfigOn, withSmpServerProxy, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn)
import Simplex.Messaging.Agent hiding (createConnection, joinConnection, sendMessage)
import qualified Simplex.Messaging.Agent as A
import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestStep (..), ServerQueueInfo (..), UserNetworkInfo (..), UserNetworkType (..), waitForUserNetwork)
@@ -96,6 +96,7 @@ import Simplex.Messaging.Protocol (BasicAuth, ErrorType (..), MsgBody, ProtocolS
import qualified Simplex.Messaging.Protocol as SMP
import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
import Simplex.Messaging.Server.Expiration
import Simplex.Messaging.Server.MsgStore.Types (AMSType (..), SMSType (..))
import Simplex.Messaging.Server.QueueStore.QueueInfo
import Simplex.Messaging.Transport (ATransport (..), SMPVersion, VersionSMP, authCmdsSMPVersion, basicAuthSMPVersion, batchCmdsSMPVersion, currentServerSMPRelayVersion, sndAuthKeySMPVersion, supportedSMPHandshakes)
import Simplex.Messaging.Util (bshow, diffToMicroseconds)
@@ -530,7 +531,7 @@ testRatchetMatrix2 t runTest = do
testServerMatrix2 :: HasCallStack => ATransport -> (InitialAgentServers -> IO ()) -> Spec
testServerMatrix2 t runTest = do
it "1 server" $ withSmpServer t $ runTest initAgentServers
it "2 servers" $ withSmpServer t . withSmpServerOn t testPort2 $ runTest initAgentServers2
it "2 servers" $ withSmpServer t $ withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2} testPort2 $ \_ -> runTest initAgentServers2
testPQMatrix2 :: HasCallStack => ATransport -> (HasCallStack => (AgentClient, InitialKeys) -> (AgentClient, PQSupport) -> AgentMsgId -> IO ()) -> Spec
testPQMatrix2 = pqMatrix2_ True
@@ -1282,7 +1283,7 @@ testSkippedMessages :: HasCallStack => ATransport -> IO ()
testSkippedMessages t = do
alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
(aliceId, bobId) <- withSmpServerStoreLogOn t testPort $ \_ -> do
(aliceId, bobId) <- withSmpServerConfigOn t cfg' testPort $ \_ -> do
(aliceId, bobId) <- runRight $ makeConnection alice bob
runRight_ $ do
2 <- sendMessage alice bobId SMP.noMsgFlags "hello"
@@ -1310,7 +1311,7 @@ testSkippedMessages t = do
alice2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
bob2 <- getSMPAgentClient' 4 agentCfg initAgentServers testDB2
withSmpServerStoreLogOn t testPort $ \_ -> do
withSmpServerConfigOn t cfg' testPort $ \_ -> do
runRight_ $ do
subscribeConnection bob2 aliceId
subscribeConnection alice2 bobId
@@ -1326,6 +1327,8 @@ testSkippedMessages t = do
ackMessage bob2 aliceId 4 Nothing
disposeAgentClient alice2
disposeAgentClient bob2
where
cfg' = cfg {msgStoreType = AMSType SMSMemory, storeMsgsFile = Nothing}
testDeliveryAfterSubscriptionError :: HasCallStack => ATransport -> IO ()
testDeliveryAfterSubscriptionError t = do
@@ -1429,7 +1432,7 @@ withUP a bId p =
]
testExpireMessageQuota :: HasCallStack => ATransport -> IO ()
testExpireMessageQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1} testPort $ \_ -> do
testExpireMessageQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1, maxJournalMsgCount = 2} testPort $ \_ -> do
a <- getSMPAgentClient' 1 agentCfg {quotaExceededTimeout = 1, messageRetryInterval = fastMessageRetryInterval} initAgentServers testDB
b <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
(aId, bId) <- runRight $ do
@@ -1455,7 +1458,7 @@ testExpireMessageQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1} testP
disposeAgentClient a
testExpireManyMessagesQuota :: ATransport -> IO ()
testExpireManyMessagesQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1} testPort $ \_ -> do
testExpireManyMessagesQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1, maxJournalMsgCount = 2} testPort $ \_ -> do
a <- getSMPAgentClient' 1 agentCfg {quotaExceededTimeout = 2, messageRetryInterval = fastMessageRetryInterval} initAgentServers testDB
b <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
(aId, bId) <- runRight $ do
@@ -2356,7 +2359,7 @@ testJoinConnectionAsyncReplyErrorV8 t = do
ConnectionStats {rcvQueuesInfo = [], sndQueuesInfo = [SndQueueInfo {}]} <- getConnectionServers b aId
pure (aId, bId)
nGet a =##> \case ("", "", DOWN _ [c]) -> c == bId; _ -> False
withSmpServerOn t testPort2 $ do
withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2} testPort2 $ \_ -> do
get b =##> \case ("2", c, JOINED sqSecured) -> c == aId && not sqSecured; _ -> False
confId <- withSmpServerStoreLogOn t testPort $ \_ -> do
pGet a >>= \case
@@ -2395,7 +2398,7 @@ testJoinConnectionAsyncReplyError t = do
ConnectionStats {rcvQueuesInfo = [], sndQueuesInfo = [SndQueueInfo {}]} <- getConnectionServers b aId
pure (aId, bId)
nGet a =##> \case ("", "", DOWN _ [c]) -> c == bId; _ -> False
withSmpServerOn t testPort2 $ do
withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2} testPort2 $ \_ -> do
confId <- withSmpServerStoreLogOn t 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
@@ -2974,7 +2977,7 @@ testDeliveryReceiptsVersion t = do
testDeliveryReceiptsConcurrent :: HasCallStack => ATransport -> IO ()
testDeliveryReceiptsConcurrent t =
withSmpServerConfigOn t cfg {msgQueueQuota = 256} testPort $ \_ -> do
withSmpServerConfigOn t cfg {msgQueueQuota = 256, maxJournalMsgCount = 512} testPort $ \_ -> do
withAgentClients2 $ \a b -> do
(aId, bId) <- runRight $ makeConnection a b
t1 <- liftIO getCurrentTime

222
tests/CoreTests/MsgStoreTests.hs Normal file
View File

@@ -0,0 +1,222 @@
{-# LANGUAGE DuplicateRecordFields #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE PatternSynonyms #-}
{-# LANGUAGE RankNTypes #-}
{-# LANGUAGE StandaloneDeriving #-}
{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
{-# OPTIONS_GHC -Wno-orphans #-}
module CoreTests.MsgStoreTests where
import AgentTests.FunctionalAPITests (runRight_)
import Control.Concurrent.STM
import Control.Exception (bracket)
import Control.Monad
import Control.Monad.IO.Class
import Data.ByteString.Char8 (ByteString)
import qualified Data.ByteString.Char8 as B
import Data.Time.Clock.System (getSystemTime)
import Simplex.Messaging.Crypto (pattern MaxLenBS)
import qualified Simplex.Messaging.Crypto as C
import Simplex.Messaging.Protocol (EntityId (..), Message (..), noMsgFlags)
import Simplex.Messaging.Server (exportMessages, importMessages)
import Simplex.Messaging.Server.MsgStore.Journal
import Simplex.Messaging.Server.MsgStore.STM
import Simplex.Messaging.Server.MsgStore.Types
import SMPClient (testStoreMsgsDir, testStoreMsgsDir2, testStoreMsgsFile, testStoreMsgsFile2)
import System.Directory (copyFile, createDirectoryIfMissing, listDirectory, removeFile, renameFile)
import System.FilePath ((</>))
import System.IO (IOMode (..), hClose, withFile)
import Test.Hspec
msgStoreTests :: Spec
msgStoreTests = do
around (withMsgStore testSMTStoreConfig) $ describe "STM message store" someMsgStoreTests
around (withMsgStore testJournalStoreCfg) $ describe "Journal message store" $ do
someMsgStoreTests
it "should export and import journal store" testExportImportStore
describe "queue state" $ do
it "should restore queue state from the last line" testQueueState
where
someMsgStoreTests :: MsgStoreClass s => SpecWith s
someMsgStoreTests = do
it "should get queue and store/read messages" testGetQueue
it "should not fail on EOF when changing read journal" testChangeReadJournal
withMsgStore :: MsgStoreClass s => MsgStoreConfig s -> (s -> IO ()) -> IO ()
withMsgStore cfg = bracket (newMsgStore cfg) closeMsgStore
testSMTStoreConfig :: STMStoreConfig
testSMTStoreConfig = STMStoreConfig {storePath = Nothing, quota = 3}
testJournalStoreCfg :: JournalStoreConfig
testJournalStoreCfg =
JournalStoreConfig
{ storePath = testStoreMsgsDir,
pathParts = 4,
quota = 3,
maxMsgCount = 4,
maxStateLines = 2
}
mkMessage :: MonadIO m => ByteString -> m Message
mkMessage body = liftIO $ do
g <- C.newRandom
msgTs <- getSystemTime
msgId <- atomically $ C.randomBytes 24 g
pure Message {msgId, msgTs, msgFlags = noMsgFlags, msgBody = C.unsafeMaxLenBS body}
pattern Msg :: ByteString -> Maybe Message
pattern Msg s <- Just Message {msgBody = MaxLenBS s}
deriving instance Eq MsgQueueState
deriving instance Eq JournalState
testGetQueue :: MsgStoreClass s => s -> IO ()
testGetQueue st = do
g <- C.newRandom
rId <- EntityId <$> atomically (C.randomBytes 24 g)
runRight_ $ do
q <- getMsgQueue st rId
Just (Message {msgId = mId1}, True) <- writeMsg q True =<< mkMessage "message 1"
Just (Message {msgId = mId2}, False) <- writeMsg q True =<< mkMessage "message 2"
Just (Message {msgId = mId3}, False) <- writeMsg q True =<< mkMessage "message 3"
Msg "message 1" <- tryPeekMsg q
Msg "message 1" <- tryPeekMsg q
Nothing <- tryDelMsg q mId2
Msg "message 1" <- tryDelMsg q mId1
Nothing <- tryDelMsg q mId1
Msg "message 2" <- tryPeekMsg q
Nothing <- tryDelMsg q mId1
(Nothing, Msg "message 2") <- tryDelPeekMsg q mId1
(Msg "message 2", Msg "message 3") <- tryDelPeekMsg q mId2
(Nothing, Msg "message 3") <- tryDelPeekMsg q mId2
Msg "message 3" <- tryPeekMsg q
(Msg "message 3", Nothing) <- tryDelPeekMsg q mId3
Nothing <- tryDelMsg q mId2
Nothing <- tryDelMsg q mId3
Nothing <- tryPeekMsg q
Just (Message {msgId = mId4}, True) <- writeMsg q True =<< mkMessage "message 4"
Msg "message 4" <- tryPeekMsg q
Just (Message {msgId = mId5}, False) <- writeMsg q True =<< mkMessage "message 5"
(Nothing, Msg "message 4") <- tryDelPeekMsg q mId3
(Msg "message 4", Msg "message 5") <- tryDelPeekMsg q mId4
Just (Message {msgId = mId6}, False) <- writeMsg q True =<< mkMessage "message 6"
Just (Message {msgId = mId7}, False) <- writeMsg q True =<< mkMessage "message 7"
Nothing <- writeMsg q True =<< mkMessage "message 8"
Msg "message 5" <- tryPeekMsg q
(Nothing, Msg "message 5") <- tryDelPeekMsg q mId4
(Msg "message 5", Msg "message 6") <- tryDelPeekMsg q mId5
(Msg "message 6", Msg "message 7") <- tryDelPeekMsg q mId6
(Msg "message 7", Just MessageQuota {msgId = mId8}) <- tryDelPeekMsg q mId7
(Just MessageQuota {}, Nothing) <- tryDelPeekMsg q mId8
(Nothing, Nothing) <- tryDelPeekMsg q mId8
pure ()
delMsgQueue st rId
testChangeReadJournal :: MsgStoreClass s => s -> IO ()
testChangeReadJournal st = do
g <- C.newRandom
rId <- EntityId <$> atomically (C.randomBytes 24 g)
runRight_ $ do
q <- getMsgQueue st rId
Just (Message {msgId = mId1}, True) <- writeMsg q True =<< mkMessage "message 1"
(Msg "message 1", Nothing) <- tryDelPeekMsg q mId1
Just (Message {msgId = mId2}, True) <- writeMsg q True =<< mkMessage "message 2"
(Msg "message 2", Nothing) <- tryDelPeekMsg q mId2
Just (Message {msgId = mId3}, True) <- writeMsg q True =<< mkMessage "message 3"
(Msg "message 3", Nothing) <- tryDelPeekMsg q mId3
Just (Message {msgId = mId4}, True) <- writeMsg q True =<< mkMessage "message 4"
(Msg "message 4", Nothing) <- tryDelPeekMsg q mId4
Just (Message {msgId = mId5}, True) <- writeMsg q True =<< mkMessage "message 5"
(Msg "message 5", Nothing) <- tryDelPeekMsg q mId5
pure ()
delMsgQueue st rId
testExportImportStore :: JournalMsgStore -> IO ()
testExportImportStore st = do
g <- C.newRandom
rId1 <- EntityId <$> atomically (C.randomBytes 24 g)
rId2 <- EntityId <$> atomically (C.randomBytes 24 g)
runRight_ $ do
q1 <- getMsgQueue st rId1
Just (Message {}, True) <- writeMsg q1 True =<< mkMessage "message 1"
Just (Message {}, False) <- writeMsg q1 True =<< mkMessage "message 2"
q2 <- getMsgQueue st rId2
Just (Message {}, True) <- writeMsg q2 True =<< mkMessage "message 3"
Just (Message {}, False) <- writeMsg q2 True =<< mkMessage "message 4"
Just (Message {}, False) <- writeMsg q2 True =<< mkMessage "message 5"
Nothing <- writeMsg q2 True =<< mkMessage "message 6"
pure ()
length <$> listDirectory (msgQueueDirectory st rId1) `shouldReturn` 2
length <$> listDirectory (msgQueueDirectory st rId2) `shouldReturn` 2
exportMessages st testStoreMsgsFile $ getQueueMessages False
let cfg = (testJournalStoreCfg :: JournalStoreConfig) {storePath = testStoreMsgsDir2}
st' <- newMsgStore cfg
0 <- importMessages st' testStoreMsgsFile Nothing
length <$> listDirectory (msgQueueDirectory st rId1) `shouldReturn` 2
length <$> listDirectory (msgQueueDirectory st rId2) `shouldReturn` 3 -- state file is backed up
exportMessages st' testStoreMsgsFile2 $ getQueueMessages False
(B.readFile testStoreMsgsFile2 `shouldReturn`) =<< B.readFile (testStoreMsgsFile <> ".bak")
stmStore <- newMsgStore testSMTStoreConfig
0 <- importMessages stmStore testStoreMsgsFile2 Nothing
exportMessages stmStore testStoreMsgsFile $ getQueueMessages False
(B.sort <$> B.readFile testStoreMsgsFile `shouldReturn`) =<< (B.sort <$> B.readFile (testStoreMsgsFile2 <> ".bak"))
testQueueState :: JournalMsgStore -> IO ()
testQueueState st = do
g <- C.newRandom
rId <- EntityId <$> atomically (C.randomBytes 24 g)
let dir = msgQueueDirectory st rId
statePath = dir </> (queueLogFileName <> logFileExt)
createDirectoryIfMissing True dir
state <- newMsgQueueState <$> newJournalId (random st)
withFile statePath WriteMode (`logQueueState` state)
length . lines <$> readFile statePath `shouldReturn` 1
readQueueState statePath `shouldReturn` state
length <$> listDirectory dir `shouldReturn` 1 -- no backup
let state1 = state {size = 1, writeState = (writeState state) {msgPos = 1, msgCount = 1, bytePos = 100}}
withFile statePath AppendMode (`logQueueState` state1)
length . lines <$> readFile statePath `shouldReturn` 2
readQueueState statePath `shouldReturn` state1
length <$> listDirectory dir `shouldReturn` 1 -- no backup
let state2 = state {size = 2, writeState = (writeState state) {msgPos = 2, msgCount = 2, bytePos = 200}}
withFile statePath AppendMode (`logQueueState` state2)
length . lines <$> readFile statePath `shouldReturn` 3
copyFile statePath (statePath <> ".2")
readQueueState statePath `shouldReturn` state2
length <$> listDirectory dir `shouldReturn` 3 -- new state, copy + backup
length . lines <$> readFile statePath `shouldReturn` 1
-- corrupt the only line
corruptFile statePath
newState <- readQueueState statePath
newState `shouldBe` newMsgQueueState (journalId $ writeState newState)
-- corrupt the last line
renameFile (statePath <> ".2") statePath
removeOtherFiles dir statePath
length . lines <$> readFile statePath `shouldReturn` 3
corruptFile statePath
readQueueState statePath `shouldReturn` state1
length <$> listDirectory dir `shouldReturn` 2
length . lines <$> readFile statePath `shouldReturn` 1
where
readQueueState statePath = do
(state, h) <- readWriteQueueState st statePath
hClose h
pure state
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

16
tests/SMPClient.hs
View File

@@ -23,6 +23,7 @@ import Simplex.Messaging.Encoding
import Simplex.Messaging.Protocol
import Simplex.Messaging.Server (runSMPServerBlocking)
import Simplex.Messaging.Server.Env.STM
import Simplex.Messaging.Server.MsgStore.Types (AMSType (..), SMSType (..))
import Simplex.Messaging.Transport
import Simplex.Messaging.Transport.Client
import qualified Simplex.Messaging.Transport.Client as Client
@@ -62,6 +63,12 @@ testStoreMsgsFile = "tests/tmp/smp-server-messages.log"
testStoreMsgsFile2 :: FilePath
testStoreMsgsFile2 = "tests/tmp/smp-server-messages.log.2"
testStoreMsgsDir :: FilePath
testStoreMsgsDir = "tests/tmp/messages"
testStoreMsgsDir2 :: FilePath
testStoreMsgsDir2 = "tests/tmp/messages.2"
testStoreNtfsFile :: FilePath
testStoreNtfsFile = "tests/tmp/smp-server-ntfs.log"
@@ -105,11 +112,14 @@ cfg =
{ transports = [],
smpHandshakeTimeout = 60000000,
tbqSize = 1,
msgStoreType = AMSType SMSJournal,
msgQueueQuota = 4,
maxJournalMsgCount = 5,
maxJournalStateLines = 2,
queueIdBytes = 24,
msgIdBytes = 24,
storeLogFile = Nothing,
storeMsgsFile = Nothing,
storeLogFile = Just testStoreLogFile,
storeMsgsFile = Just testStoreMsgsDir,
storeNtfsFile = Nothing,
allowNewQueues = True,
newQueueBasicAuth = Nothing,
@@ -173,7 +183,7 @@ withSmpServerStoreMsgLogOn t =
t
cfg
{ storeLogFile = Just testStoreLogFile,
storeMsgsFile = Just testStoreMsgsFile,
storeMsgsFile = Just testStoreMsgsDir,
storeNtfsFile = Just testStoreNtfsFile,
serverStatsBackupFile = Just testServerStatsBackupFile
}

19
tests/SMPProxyTests.hs
View File

@@ -135,14 +135,15 @@ smpProxyTests = do
let deliver nAgents nMsgs = agentDeliverMessagesViaProxyConc (replicate nAgents [srv1]) (map bshow [1 :: Int .. nMsgs])
it "25 agents, 300 pairs, 17 messages" . oneServer . withNumCapabilities 4 $ deliver 25 17
where
oneServer = withSmpServerConfigOn (transport @TLS) proxyCfg {msgQueueQuota = 128} testPort . const
oneServer = withSmpServerConfigOn (transport @TLS) proxyCfg {msgQueueQuota = 128, maxJournalMsgCount = 256} testPort . const
twoServers = twoServers_ proxyCfg proxyCfg
twoServersFirstProxy = twoServers_ proxyCfg cfgV8 {msgQueueQuota = 128}
twoServersMoreConc = twoServers_ proxyCfg {serverClientConcurrency = 128} cfgV8 {msgQueueQuota = 128}
twoServersNoConc = twoServers_ proxyCfg {serverClientConcurrency = 1} cfgV8 {msgQueueQuota = 128}
twoServersFirstProxy = twoServers_ proxyCfg cfgV8 {msgQueueQuota = 128, maxJournalMsgCount = 256}
twoServersMoreConc = twoServers_ proxyCfg {serverClientConcurrency = 128} cfgV8 {msgQueueQuota = 128, maxJournalMsgCount = 256}
twoServersNoConc = twoServers_ proxyCfg {serverClientConcurrency = 1} cfgV8 {msgQueueQuota = 128, maxJournalMsgCount = 256}
twoServers_ cfg1 cfg2 runTest =
withSmpServerConfigOn (transport @TLS) cfg1 testPort $ \_ ->
withSmpServerConfigOn (transport @TLS) cfg2 testPort2 $ const runTest
let cfg2' = cfg2 {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2}
in withSmpServerConfigOn (transport @TLS) cfg2' testPort2 $ const runTest
deliverMessageViaProxy :: (C.AlgorithmI a, C.AuthAlgorithm a) => SMPServer -> SMPServer -> C.SAlgorithm a -> ByteString -> ByteString -> IO ()
deliverMessageViaProxy proxyServ relayServ alg msg msg' = deliverMessagesViaProxy proxyServ relayServ alg [msg] [msg']
@@ -379,11 +380,11 @@ agentViaProxyRetryOffline = do
ackMessage alice bobId (baseId + 4) Nothing
where
withServer :: (ThreadId -> IO a) -> IO a
withServer = withServer_ testStoreLogFile testStoreMsgsFile testStoreNtfsFile testPort
withServer = withServer_ testStoreLogFile testStoreMsgsDir testStoreNtfsFile testPort
withServer2 :: (ThreadId -> IO a) -> IO a
withServer2 = withServer_ testStoreLogFile2 testStoreMsgsFile2 testStoreNtfsFile2 testPort2
withServer_ storeLog storeMsgs storeNtfs port =
withSmpServerConfigOn (transport @TLS) proxyCfg {storeLogFile = Just storeLog, storeMsgsFile = Just storeMsgs, storeNtfsFile = Just storeNtfs} port
withServer2 = withServer_ testStoreLogFile2 testStoreMsgsDir2 testStoreNtfsFile2 testPort2
withServer_ storeLog storeMsgs storeNtfs =
withSmpServerConfigOn (transport @TLS) proxyCfg {storeLogFile = Just storeLog, storeMsgsFile = Just storeMsgs, storeNtfsFile = Just storeNtfs}
a `up` cId = nGet a =##> \case ("", "", UP _ [c]) -> c == cId; _ -> False
a `down` cId = nGet a =##> \case ("", "", DOWN _ [c]) -> c == cId; _ -> False
aCfg = agentCfg {messageRetryInterval = fastMessageRetryInterval}

32
tests/ServerTests.hs
View File

@@ -37,10 +37,12 @@ import Simplex.Messaging.Parsers (parseAll)
import Simplex.Messaging.Protocol
import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
import Simplex.Messaging.Server.Expiration
import Simplex.Messaging.Server.MsgStore.Types (AMSType (..), SMSType (..))
import Simplex.Messaging.Server.Stats (PeriodStatsData (..), ServerStatsData (..))
import Simplex.Messaging.Transport
import Simplex.Messaging.Util (whenM)
import Simplex.Messaging.Version (mkVersionRange)
import System.Directory (removeFile)
import System.Directory (doesDirectoryExist, removeDirectoryRecursive, removeFile)
import System.TimeIt (timeItT)
import System.Timeout
import Test.HUnit
@@ -64,8 +66,8 @@ serverTests t@(ATransport t') = do
describe "GET & SUB commands" $ testGetSubCommands t'
describe "Exceeding queue quota" $ testExceedQueueQuota t'
describe "Store log" $ testWithStoreLog t
describe "Restore messages" $ testRestoreMessages t
describe "Restore messages (old / v2)" $ testRestoreExpireMessages t
xdescribe "Restore messages" $ testRestoreMessages t
xdescribe "Restore messages (old / v2)" $ testRestoreExpireMessages t
describe "Timing of AUTH error" $ testTiming t
describe "Message notifications" $ testMessageNotifications t
describe "Message expiration" $ do
@@ -601,7 +603,8 @@ testWithStoreLog at@(ATransport t) =
logSize testStoreLogFile `shouldReturn` 6
withSmpServerThreadOn at testPort . runTest t $ \h -> do
let cfg' = cfg {msgStoreType = AMSType SMSMemory, storeLogFile = Nothing, storeMsgsFile = Nothing}
withSmpServerConfigOn at cfg' testPort . runTest t $ \h -> do
sId1 <- readTVarIO senderId1
-- fails if store log is disabled
Resp "bcda" _ (ERR AUTH) <- signSendRecv h sKey1 ("bcda", sId1, _SEND "hello")
@@ -646,6 +649,7 @@ testRestoreMessages at@(ATransport t) =
it "should store messages on exit and restore on start" $ do
removeFileIfExists testStoreLogFile
removeFileIfExists testStoreMsgsFile
whenM (doesDirectoryExist testStoreMsgsDir) $ removeDirectoryRecursive testStoreMsgsDir
removeFileIfExists testServerStatsBackupFile
g <- C.newRandom
@@ -679,7 +683,7 @@ testRestoreMessages at@(ATransport t) =
rId <- readTVarIO recipientId
logSize testStoreLogFile `shouldReturn` 2
logSize testStoreMsgsFile `shouldReturn` 5
-- logSize testStoreMsgsFile `shouldReturn` 5
logSize testServerStatsBackupFile `shouldReturn` 74
Right stats1 <- strDecode <$> B.readFile testServerStatsBackupFile
checkStats stats1 [rId] 5 1
@@ -697,7 +701,7 @@ testRestoreMessages at@(ATransport t) =
logSize testStoreLogFile `shouldReturn` 1
-- the last message is not removed because it was not ACK'd
logSize testStoreMsgsFile `shouldReturn` 3
-- logSize testStoreMsgsFile `shouldReturn` 3
logSize testServerStatsBackupFile `shouldReturn` 74
Right stats2 <- strDecode <$> B.readFile testServerStatsBackupFile
checkStats stats2 [rId] 5 3
@@ -714,15 +718,15 @@ testRestoreMessages at@(ATransport t) =
(dec mId6 msg6, Left "ClientRcvMsgQuota") #== "restored message delivered"
Resp "7" _ OK <- signSendRecv h rKey ("7", rId, ACK mId6)
pure ()
logSize testStoreLogFile `shouldReturn` 1
logSize testStoreMsgsFile `shouldReturn` 0
-- logSize testStoreMsgsFile `shouldReturn` 0
logSize testServerStatsBackupFile `shouldReturn` 74
Right stats3 <- strDecode <$> B.readFile testServerStatsBackupFile
checkStats stats3 [rId] 5 5
removeFile testStoreLogFile
removeFile testStoreMsgsFile
removeFileIfExists testStoreMsgsFile
whenM (doesDirectoryExist testStoreMsgsDir) $ removeDirectoryRecursive testStoreMsgsDir
removeFile testServerStatsBackupFile
where
runTest :: Transport c => TProxy c -> (THandleSMP c 'TClient -> IO ()) -> ThreadId -> Expectation
@@ -780,7 +784,7 @@ testRestoreExpireMessages at@(ATransport t) =
length (B.lines msgs) `shouldBe` 4
let expCfg1 = Just ExpirationConfig {ttl = 86400, checkInterval = 43200}
cfg1 = cfg {storeLogFile = Just testStoreLogFile, storeMsgsFile = Just testStoreMsgsFile, messageExpiration = expCfg1, serverStatsBackupFile = Just testServerStatsBackupFile}
cfg1 = cfg {messageExpiration = expCfg1, serverStatsBackupFile = Just testServerStatsBackupFile}
withSmpServerConfigOn at cfg1 testPort . runTest t $ \_ -> pure ()
logSize testStoreLogFile `shouldReturn` 1
@@ -788,14 +792,14 @@ testRestoreExpireMessages at@(ATransport t) =
msgs' `shouldBe` msgs
let expCfg2 = Just ExpirationConfig {ttl = 2, checkInterval = 43200}
cfg2 = cfg {storeLogFile = Just testStoreLogFile, storeMsgsFile = Just testStoreMsgsFile, messageExpiration = expCfg2, serverStatsBackupFile = Just testServerStatsBackupFile}
cfg2 = cfg {messageExpiration = expCfg2, serverStatsBackupFile = Just testServerStatsBackupFile}
withSmpServerConfigOn at cfg2 testPort . runTest t $ \_ -> pure ()
logSize testStoreLogFile `shouldReturn` 1
-- two messages expired
msgs'' <- B.readFile testStoreMsgsFile
length (B.lines msgs'') `shouldBe` 2
B.lines msgs'' `shouldBe` drop 2 (B.lines msgs)
-- msgs'' <- B.readFile testStoreMsgsFile
-- length (B.lines msgs'') `shouldBe` 2
-- B.lines msgs'' `shouldBe` drop 2 (B.lines msgs)
Right ServerStatsData {_msgExpired} <- strDecode <$> B.readFile testServerStatsBackupFile
_msgExpired `shouldBe` 2
where

2
tests/Test.hs
View File

@@ -11,6 +11,7 @@ import CoreTests.BatchingTests
import CoreTests.CryptoFileTests
import CoreTests.CryptoTests
import CoreTests.EncodingTests
import CoreTests.MsgStoreTests
import CoreTests.RetryIntervalTests
import CoreTests.SOCKSSettings
import CoreTests.StoreLogTests
@@ -53,6 +54,7 @@ main = do
describe "Version range" versionRangeTests
describe "Encryption tests" cryptoTests
describe "Encrypted files tests" cryptoFileTests
describe "Message store tests" msgStoreTests
describe "Retry interval tests" retryIntervalTests
describe "SOCKS settings tests" socksSettingsTests
describe "Store log tests" storeLogTests