From 1484523164445487486ac2372b03808ed01b3dc6 Mon Sep 17 00:00:00 2001 From: Evgeny Date: Tue, 22 Oct 2024 12:09:51 +0100 Subject: [PATCH] smp server, journal store: check file size on append, validate queue state (#1377) * smp server: journal store queue state validation * validate state on update, refactor * handle message parsing errors * typo * fix test, throw exception when file is smaller than expected * fix * fix test in ghc 8.10.7 --- src/Simplex/Messaging/Server.hs | 12 +- src/Simplex/Messaging/Server/Main.hs | 4 +- .../Messaging/Server/MsgStore/Journal.hs | 198 ++++++++++-------- tests/AgentTests/FunctionalAPITests.hs | 2 +- tests/AgentTests/NotificationTests.hs | 4 +- tests/CoreTests/MsgStoreTests.hs | 24 ++- tests/ServerTests.hs | 22 +- tests/Test.hs | 2 +- 8 files changed, 161 insertions(+), 107 deletions(-) diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index 214cd0790..efde4e667 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/src/Simplex/Messaging/Server.hs @@ -93,7 +93,7 @@ import Simplex.Messaging.Server.Control import Simplex.Messaging.Server.Env.STM as Env import Simplex.Messaging.Server.Expiration import Simplex.Messaging.Server.MsgStore -import Simplex.Messaging.Server.MsgStore.Journal (JournalMsgQueue (..), closeMsgQueue) +import Simplex.Messaging.Server.MsgStore.Journal (JournalMsgQueue (..), JMQueue (..), closeMsgQueue) import Simplex.Messaging.Server.MsgStore.STM import Simplex.Messaging.Server.MsgStore.Types import Simplex.Messaging.Server.NtfStore @@ -1731,17 +1731,17 @@ saveServerMessages :: Bool -> M () saveServerMessages drainMsgs = asks msgStore >>= \case AMS SMSMemory ms@STMMsgStore {storeConfig = STMStoreConfig {storePath}} -> case storePath of - Just f -> liftIO $ exportMessages ms f $ getQueueMessages drainMsgs + Just f -> liftIO $ exportMessages ms f 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 +exportMessages :: MsgStoreClass s => s -> FilePath -> Bool -> IO () +exportMessages ms f drainMsgs = do logInfo $ "saving messages to file " <> T.pack f total <- liftIO $ withFile f WriteMode $ \h -> withAllMsgQueues ms (saveQueueMsgs h) 0 logInfo $ "messages saved: " <> tshow total where - saveQueueMsgs h rId q acc = getMessages q >>= \msgs -> (acc + length msgs) <$ BLD.hPutBuilder h (encodeMessages rId msgs) + saveQueueMsgs h rId q acc = getQueueMessages drainMsgs q >>= \msgs -> (acc + length msgs) <$ BLD.hPutBuilder h (encodeMessages rId msgs) encodeMessages rId = mconcat . map (\msg -> BLD.byteString (strEncode $ MLRv3 rId msg) <> BLD.char8 '\n') restoreServerMessages :: M MessageStats @@ -1768,7 +1768,7 @@ restoreServerMessages = do runExceptT expireQueue >>= \case Right (stored', expired') -> pure (stored + stored', expired + expired', qCount + 1) Left e -> do - logInfo $ "failed expiring messages in queue " <> T.pack (queueDirectory q) <> ": " <> tshow e + logInfo $ "failed expiring messages in queue " <> T.pack (queueDirectory $ queue q) <> ": " <> tshow e exitFailure where expireQueue = do diff --git a/src/Simplex/Messaging/Server/Main.hs b/src/Simplex/Messaging/Server/Main.hs index f1f0808fa..5abe0b331 100644 --- a/src/Simplex/Messaging/Server/Main.hs +++ b/src/Simplex/Messaging/Server/Main.hs @@ -44,7 +44,7 @@ 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.Journal (JournalStoreConfig (..)) 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) @@ -115,7 +115,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath = ("WARNING: journal directory " <> storeMsgsJournalDir <> " will be exported to message log file " <> storeMsgsFilePath) "Journal not exported" ms <- newJournalMsgStore - exportMessages ms storeMsgsFilePath $ getQueueMessages False + exportMessages ms storeMsgsFilePath False putStrLn "Export completed" putStrLn $ case readMsgStoreType ini of Right (AMSType SMSMemory) -> "store_messages set to `memory`" diff --git a/src/Simplex/Messaging/Server/MsgStore/Journal.hs b/src/Simplex/Messaging/Server/MsgStore/Journal.hs index e0ba446ad..13df614a8 100644 --- a/src/Simplex/Messaging/Server/MsgStore/Journal.hs +++ b/src/Simplex/Messaging/Server/MsgStore/Journal.hs @@ -15,7 +15,8 @@ module Simplex.Messaging.Server.MsgStore.Journal ( JournalMsgStore (msgQueues, random), - JournalMsgQueue (queueDirectory), + JournalMsgQueue (queue), + JMQueue (queueDirectory, statePath), JournalStoreConfig (..), getQueueMessages, closeMsgQueue, @@ -43,6 +44,7 @@ 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.Int (Int64) import Data.Maybe (catMaybes, fromMaybe) import qualified Data.Text as T import Data.Time.Clock (getCurrentTime) @@ -81,15 +83,20 @@ data JournalStoreConfig = JournalStoreConfig maxStateLines :: Int } -data JournalMsgQueue = JournalMsgQueue +data JMQueue = JMQueue { queueDirectory :: FilePath, queueLock :: Lock, + statePath :: FilePath + } + +data JournalMsgQueue = JournalMsgQueue + { queue :: JMQueue, state :: TVar MsgQueueState, -- tipMsg contains last message and length incl. newline -- Nothing - unknown, Just Nothing - empty queue. -- It prevents reading each message twice, -- and reading it after it was just written. - tipMsg :: TVar (Maybe (Maybe (Message, Int))), + tipMsg :: TVar (Maybe (Maybe (Message, Int64))), handles :: TVar (Maybe MsgQueueHandles) } @@ -112,8 +119,8 @@ data JournalState t = JournalState journalId :: ByteString, msgPos :: Int, msgCount :: Int, - bytePos :: Int, - byteCount :: Int + bytePos :: Int64, + byteCount :: Int64 } deriving (Show) @@ -141,15 +148,7 @@ newMsgQueueState journalId = } newJournalState :: JournalTypeI t => ByteString -> JournalState t -newJournalState journalId = - JournalState - { journalType = sJournalType, - journalId, - msgPos = 0, - msgCount = 0, - bytePos = 0, - byteCount = 0 - } +newJournalState journalId = JournalState sJournalType journalId 0 0 0 0 journalFilePath :: FilePath -> ByteString -> FilePath journalFilePath dir journalId = dir (msgLogFileName <> "." <> B.unpack journalId <> logFileExt) @@ -171,13 +170,16 @@ instance StrEncoding MsgQueueState where instance JournalTypeI t => StrEncoding (JournalState t) where strEncode JournalState {journalId, msgPos, msgCount, bytePos, byteCount} = - B.concat [journalId, ",", strEncodeList [msgPos, msgCount, bytePos, byteCount]] + B.intercalate "," [journalId, e msgPos, e msgCount, e bytePos, e byteCount] + where + e :: StrEncoding a => a -> ByteString + e = strEncode strP = do journalId <- A.takeTill (== ',') - A.char ',' *> strListP >>= \case - [msgPos, msgCount, bytePos, byteCount] -> - pure JournalState {journalType = sJournalType, journalId, msgPos, msgCount, bytePos, byteCount} - _ -> fail "bad JournalState" + JournalState sJournalType journalId <$> i <*> i <*> i <*> i + where + i :: Integral a => A.Parser a + i = A.char ',' *> A.decimal queueLogFileName :: String queueLogFileName = "queue_state" @@ -210,6 +212,8 @@ instance MsgStoreClass JournalMsgStore where -- This function is a "foldr" that opens and closes all queues, processes them as defined by action and accumulates the result. -- It is used to export storage to a single file and also to expire messages and validate all queues when server is started. + -- TODO this function requires case-sensitive file system, because it uses queue directory as recipient ID. + -- It can be made to support case-insensite FS by supporting more than one queue per directory, by getting recipient ID from state file name. withAllMsgQueues :: JournalMsgStore -> (RecipientId -> JournalMsgQueue -> a -> IO a) -> a -> IO a withAllMsgQueues ms@JournalMsgStore {config} action res = ifM (doesDirectoryExist storePath) processStore (pure res) where @@ -223,9 +227,11 @@ instance MsgStoreClass JournalMsgStore where putStrLn "" pure res' JournalStoreConfig {storePath, pathParts} = config - processQueue lock count acc (i :: Int, (queueId, dir)) = do + processQueue queueLock count acc (i :: Int, (queueId, dir)) = do when (i `mod` 100 == 0) $ progress i count - q <- openMsgQueue ms dir lock + let statePath = dir queueLogFileName <> "." <> queueId <> logFileExt + queue = JMQueue {queueDirectory = dir, queueLock, statePath} + q <- openMsgQueue ms queue acc' <- case strDecode $ B.pack queueId of Right rId -> action rId q acc Left e -> do @@ -261,18 +267,20 @@ instance MsgStoreClass JournalMsgStore where TM.lookupIO rId msgQueues >>= maybe newQ pure where newQ = do - let dir = msgQueueDirectory ms rId queueLock <- atomically $ getMapLock queueLocks rId - q <- ifM (doesDirectoryExist dir) (openMsgQueue ms dir queueLock) (createQ dir queueLock) + let dir = msgQueueDirectory ms rId + statePath = dir (queueLogFileName <> "." <> B.unpack (strEncode rId) <> logFileExt) + queue = JMQueue {queueDirectory = dir, queueLock, statePath} + q <- ifM (doesDirectoryExist dir) (openMsgQueue ms queue) (createQ queue) atomically $ TM.insert rId q msgQueues pure q where - createQ :: FilePath -> Lock -> IO JournalMsgQueue - createQ dir queueLock = do + createQ :: JMQueue -> IO JournalMsgQueue + createQ queue = do -- folder and files are not created here, -- to avoid file IO for queues without messages during subscription journalId <- newJournalId random - mkJournalQueue dir queueLock (newMsgQueueState journalId, Nothing) + mkJournalQueue queue (newMsgQueueState journalId, Nothing) delMsgQueue :: JournalMsgStore -> RecipientId -> IO () delMsgQueue ms rId = withLockMap (queueLocks ms) rId "delMsgQueue" $ do @@ -292,16 +300,13 @@ instance MsgStoreClass JournalMsgStore where getMsg msgs hs = chooseReadJournal q drainMsgs hs >>= maybe (pure msgs) 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, len) <- hGetMsgAt h $ bytePos rs + updateReadPos q drainMsgs len hs (msg :) <$> getMsg msgs hs writeMsg :: JournalMsgStore -> JournalMsgQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool)) - writeMsg ms q@JournalMsgQueue {queueDirectory, handles} logState !msg = - tryStore "writeMsg" $ withLock' (queueLock q) "writeMsg" $ do + writeMsg ms q@JournalMsgQueue {queue = JMQueue {queueDirectory, queueLock, statePath}, handles} logState !msg = + tryStore "writeMsg" $ withLock' queueLock "writeMsg" $ do st@MsgQueueState {canWrite, size} <- readTVarIO (state q) let empty = size == 0 if canWrite || empty @@ -316,22 +321,22 @@ instance MsgStoreClass JournalMsgStore where 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 + msgLen = fromIntegral $ 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 + let msgPos' = msgPos ws + 1 + bytePos' = bytePos ws + msgLen + ws' = ws {msgPos = msgPos', msgCount = msgPos', bytePos = bytePos', byteCount = bytePos'} + rs' = if journalId ws == journalId rs then rs {msgCount = msgPos', byteCount = bytePos'} 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 + hAppend wh (bytePos ws) 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 @@ -354,11 +359,8 @@ instance MsgStoreClass JournalMsgStore 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 + ml@(msg, _) <- hGetMsgAt h $ bytePos rs + atomically $ writeTVar tipMsg $ Just (Just ml) pure $ Just msg tryDeleteMsg_ :: JournalMsgQueue -> StoreIO () @@ -370,7 +372,7 @@ instance MsgStoreClass JournalMsgStore where $>>= \hs -> updateReadPos q True len hs $> Just () isolateQueue :: JournalMsgQueue -> String -> StoreIO a -> ExceptT ErrorType IO a - isolateQueue mq op (StoreIO a) = tryStore op $ withLock' (queueLock mq) op $ a + isolateQueue q op (StoreIO a) = tryStore op $ withLock' (queueLock $ queue q) op $ a tryStore :: String -> IO a -> ExceptT ErrorType IO a tryStore op a = @@ -379,29 +381,21 @@ tryStore op a = let e' = op <> " " <> show e in logError ("STORE ERROR " <> T.pack e') $> Left (STORE e') -openMsgQueue :: JournalMsgStore -> FilePath -> Lock -> IO JournalMsgQueue -openMsgQueue ms dir queueLock = do - let statePath = dir (queueLogFileName <> logFileExt) +openMsgQueue :: JournalMsgStore -> JMQueue -> IO JournalMsgQueue +openMsgQueue ms q@JMQueue {queueDirectory = dir, statePath} = do (st, sh) <- readWriteQueueState ms statePath (st', rh, wh_) <- openJournals dir st let hs = MsgQueueHandles {stateHandle = sh, readHandle = rh, writeHandle = wh_} - mkJournalQueue dir queueLock (st', Just hs) + mkJournalQueue q (st', Just hs) -mkJournalQueue :: FilePath -> Lock -> (MsgQueueState, Maybe MsgQueueHandles) -> IO JournalMsgQueue -mkJournalQueue dir queueLock (st, hs_) = do +mkJournalQueue :: JMQueue -> (MsgQueueState, Maybe MsgQueueHandles) -> IO JournalMsgQueue +mkJournalQueue queue (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 - { queueDirectory = dir, - queueLock, - state, - tipMsg, - handles - } + pure JournalMsgQueue {queue, state, tipMsg, handles} chooseReadJournal :: JournalMsgQueue -> Bool -> MsgQueueHandles -> IO (Maybe (JournalState 'JTRead, Handle)) chooseReadJournal q log' hs = do @@ -411,8 +405,8 @@ chooseReadJournal q log' hs = 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} + removeJournal (queueDirectory $ queue q) rs + let !rs' = (newJournalState $ journalId ws) {msgCount = msgCount ws, byteCount = byteCount ws} !st' = st {readState = rs'} updateQueueState q log' hs st' pure $ Just (rs', wh) @@ -421,13 +415,14 @@ chooseReadJournal q log' hs = do updateQueueState :: JournalMsgQueue -> Bool -> MsgQueueHandles -> MsgQueueState -> IO () updateQueueState q log' hs st = do + unless (validQueueState st) $ E.throwIO $ userError $ "updating to invalid state: " <> show st when log' $ logQueueState (stateHandle hs) st atomically $ writeTVar (state q) st logQueueState :: Handle -> MsgQueueState -> IO () logQueueState h st = B.hPutStr h $ strEncode st `B.snoc` '\n' -updateReadPos :: JournalMsgQueue -> Bool -> Int -> MsgQueueHandles -> IO () +updateReadPos :: JournalMsgQueue -> Bool -> Int64 -> MsgQueueHandles -> IO () updateReadPos q log' len hs = do st@MsgQueueState {readState = rs, size} <- readTVarIO (state q) let JournalState {msgPos, bytePos} = rs @@ -471,32 +466,39 @@ openJournals dir st@MsgQueueState {readState = rs, writeState = ws} = do pure (st', rh, Nothing) Right rh | rjId == wjId -> do - st' <- fixWriteFileSize rh - pure (st', rh, Nothing) - | otherwise -> + fixFileSize rh $ bytePos ws + pure (st, rh, Nothing) + | otherwise -> do + fixFileSize rh $ byteCount rs 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 + 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) + fixFileSize wh $ bytePos ws + pure (st, rh, Just wh) where openJournal :: JournalState t -> IO (Either FilePath Handle) openJournal JournalState {journalId} = let path = journalFilePath dir journalId in ifM (doesFileExist path) (Right <$> openFile path ReadWriteMode) (pure $ Left path) -- do that for all append operations - 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 ??? + +fixFileSize :: Handle -> Int64 -> IO () +fixFileSize h pos = do + let pos' = fromIntegral pos + size <- IO.hFileSize h + if + | size > pos' -> do + logWarn $ "STORE WARNING truncating file size from " <> tshow size <> " to " <> tshow pos + IO.hSetFileSize h pos' + | size < pos' -> + -- From code logic this can't happen. + E.throwIO $ userError $ "file size " <> show size <> " is smaller than position " <> show pos + | otherwise -> pure () removeJournal :: FilePath -> JournalState t -> IO () removeJournal dir JournalState {journalId} = do @@ -517,7 +519,10 @@ readWriteQueueState JournalMsgStore {random, config} statePath = ls <- LB.lines <$> LB.readFile statePath case ls of [] -> writeNewQueueState - _ -> useLastLine (length ls) True ls + _ -> do + r@(st, _) <- useLastLine (length ls) True ls + unless (validQueueState st) $ E.throwIO $ userError $ "read invalid invalid: " <> show st + pure r writeNewQueueState = do logWarn $ "STORE WARNING: empty queue state in " <> T.pack statePath <> ", initialized" st <- newMsgQueueState <$> newJournalId random @@ -539,9 +544,7 @@ readWriteQueueState JournalMsgStore {random, config} statePath = 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 + | otherwise -> E.throwIO $ userError $ "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. @@ -555,6 +558,25 @@ readWriteQueueState JournalMsgStore {random, config} statePath = logQueueState sh st pure (st, sh) +validQueueState :: MsgQueueState -> Bool +validQueueState MsgQueueState {readState = rs, writeState = ws, size} + | journalId rs == journalId ws = + alwaysValid + && msgPos rs <= msgPos ws + && msgCount rs == msgCount ws + && bytePos rs <= bytePos ws + && byteCount rs == byteCount ws + && size == msgCount rs - msgPos rs + | otherwise = + alwaysValid + && size == msgCount ws + msgCount rs - msgPos rs + where + alwaysValid = + msgPos rs <= msgCount rs + && bytePos rs <= byteCount rs + && msgPos ws == msgCount ws + && bytePos ws == byteCount ws + deleteMsgQueue_ :: JournalMsgStore -> RecipientId -> IO (Maybe (TVar MsgQueueState)) deleteMsgQueue_ st rId = atomically (TM.lookupDelete rId (msgQueues st)) @@ -573,11 +595,21 @@ 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 +hAppend :: Handle -> Int64 -> ByteString -> IO () +hAppend h pos s = do + fixFileSize h pos + 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 +hGetMsgAt :: Handle -> Int64 -> IO (Message, Int64) +hGetMsgAt h pos = do + IO.hSeek h AbsoluteSeek $ fromIntegral pos + s <- B.hGetLine h + case strDecode s of + Right !msg -> + let !len = fromIntegral (B.length s) + 1 + in pure (msg, len) + Left e -> E.throwIO $ userError $ "Error parsing message: " <> e openFile :: FilePath -> IOMode -> IO Handle openFile f mode = do diff --git a/tests/AgentTests/FunctionalAPITests.hs b/tests/AgentTests/FunctionalAPITests.hs index 425e93f29..646edb4fa 100644 --- a/tests/AgentTests/FunctionalAPITests.hs +++ b/tests/AgentTests/FunctionalAPITests.hs @@ -1946,7 +1946,7 @@ testBatchedSubscriptions nCreate nDel t = runServers :: ExceptT AgentErrorType IO a -> IO a runServers a = do withSmpServerStoreLogOn t testPort $ \t1 -> do - res <- withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2} testPort2 $ \t2 -> + res <- withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2} testPort2 $ \t2 -> runRight a `finally` killThread t2 killThread t1 pure res diff --git a/tests/AgentTests/NotificationTests.hs b/tests/AgentTests/NotificationTests.hs index 272c6d6d6..37b490f0d 100644 --- a/tests/AgentTests/NotificationTests.hs +++ b/tests/AgentTests/NotificationTests.hs @@ -54,7 +54,7 @@ import Data.Text.Encoding (encodeUtf8) import Database.SQLite.Simple.QQ (sql) import NtfClient import SMPAgentClient (agentCfg, initAgentServers, initAgentServers2, testDB, testDB2, testNtfServer, testNtfServer2) -import SMPClient (cfg, cfgVPrev, testPort, testPort2, testStoreLogFile2, withSmpServer, withSmpServerConfigOn, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn) +import SMPClient (cfg, cfgVPrev, testPort, testPort2, testStoreLogFile2, testStoreMsgsDir2, withSmpServer, withSmpServerConfigOn, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn) import Simplex.Messaging.Agent hiding (createConnection, joinConnection, sendMessage) import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestStep (..), withStore') import Simplex.Messaging.Agent.Env.SQLite (AgentConfig, Env (..), InitialAgentServers) @@ -785,7 +785,7 @@ testNotificationsSMPRestartBatch n t apns = runServers :: ExceptT AgentErrorType IO a -> IO a runServers a = do withSmpServerStoreLogOn t testPort $ \t1 -> do - res <- withSmpServerConfigOn t (cfg :: ServerConfig) {storeLogFile = Just testStoreLogFile2} testPort2 $ \t2 -> + res <- withSmpServerConfigOn t (cfg :: ServerConfig) {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2} testPort2 $ \t2 -> runRight a `finally` killThread t2 killThread t1 pure res diff --git a/tests/CoreTests/MsgStoreTests.hs b/tests/CoreTests/MsgStoreTests.hs index dd54b1e33..c3344bfa6 100644 --- a/tests/CoreTests/MsgStoreTests.hs +++ b/tests/CoreTests/MsgStoreTests.hs @@ -1,5 +1,6 @@ {-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE GADTs #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE PatternSynonyms #-} @@ -22,6 +23,7 @@ import Simplex.Messaging.Crypto (pattern MaxLenBS) import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Protocol (EntityId (..), Message (..), noMsgFlags) import Simplex.Messaging.Server (MessageStats (..), exportMessages, importMessages) +import Simplex.Messaging.Server.Env.STM (journalMsgStoreDepth) import Simplex.Messaging.Server.MsgStore.Journal import Simplex.Messaging.Server.MsgStore.STM import Simplex.Messaging.Server.MsgStore.Types @@ -55,7 +57,7 @@ testJournalStoreCfg :: JournalStoreConfig testJournalStoreCfg = JournalStoreConfig { storePath = testStoreMsgsDir, - pathParts = 4, + pathParts = journalMsgStoreDepth, quota = 3, maxMsgCount = 4, maxStateLines = 2 @@ -158,19 +160,19 @@ testExportImportStore ms = do pure () length <$> listDirectory (msgQueueDirectory ms rId1) `shouldReturn` 2 length <$> listDirectory (msgQueueDirectory ms rId2) `shouldReturn` 2 - exportMessages ms testStoreMsgsFile $ getQueueMessages False + exportMessages ms testStoreMsgsFile False let cfg = (testJournalStoreCfg :: JournalStoreConfig) {storePath = testStoreMsgsDir2} ms' <- newMsgStore cfg MessageStats {storedMsgsCount = 6, expiredMsgsCount = 0, storedQueues = 2} <- importMessages ms' testStoreMsgsFile Nothing length <$> listDirectory (msgQueueDirectory ms rId1) `shouldReturn` 2 length <$> listDirectory (msgQueueDirectory ms rId2) `shouldReturn` 3 -- state file is backed up - exportMessages ms' testStoreMsgsFile2 $ getQueueMessages False + exportMessages ms' testStoreMsgsFile2 False (B.readFile testStoreMsgsFile2 `shouldReturn`) =<< B.readFile (testStoreMsgsFile <> ".bak") stmStore <- newMsgStore testSMTStoreConfig MessageStats {storedMsgsCount = 6, expiredMsgsCount = 0, storedQueues = 2} <- importMessages stmStore testStoreMsgsFile2 Nothing - exportMessages stmStore testStoreMsgsFile $ getQueueMessages False + exportMessages stmStore testStoreMsgsFile False (B.sort <$> B.readFile testStoreMsgsFile `shouldReturn`) =<< (B.sort <$> B.readFile (testStoreMsgsFile2 <> ".bak")) testQueueState :: JournalMsgStore -> IO () @@ -186,13 +188,23 @@ testQueueState ms = do 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}} + let state1 = + state + { size = 1, + readState = (readState state) {msgCount = 1, byteCount = 100}, + writeState = (writeState state) {msgPos = 1, msgCount = 1, bytePos = 100, byteCount = 100} + } withFile statePath AppendMode (`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}} + let state2 = + state + { size = 2, + readState = (readState state) {msgCount = 2, byteCount = 200}, + writeState = (writeState state) {msgPos = 2, msgCount = 2, bytePos = 200, byteCount = 200} + } withFile statePath AppendMode (`logQueueState` state2) length . lines <$> readFile statePath `shouldReturn` 3 copyFile statePath (statePath <> ".2") diff --git a/tests/ServerTests.hs b/tests/ServerTests.hs index 38db97f3f..965967878 100644 --- a/tests/ServerTests.hs +++ b/tests/ServerTests.hs @@ -16,6 +16,7 @@ module ServerTests where import AgentTests.NotificationTests (removeFileIfExists) +import CoreTests.MsgStoreTests (testJournalStoreCfg) import Control.Concurrent (ThreadId, killThread, threadDelay) import Control.Concurrent.STM import Control.Exception (SomeException, try) @@ -35,9 +36,11 @@ import Simplex.Messaging.Encoding import Simplex.Messaging.Encoding.String import Simplex.Messaging.Parsers (parseAll) import Simplex.Messaging.Protocol +import Simplex.Messaging.Server (exportMessages) import Simplex.Messaging.Server.Env.STM (ServerConfig (..)) import Simplex.Messaging.Server.Expiration -import Simplex.Messaging.Server.MsgStore.Types (AMSType (..), SMSType (..)) +import Simplex.Messaging.Server.MsgStore.Journal (JournalStoreConfig (..)) +import Simplex.Messaging.Server.MsgStore.Types (AMSType (..), SMSType (..), newMsgStore) import Simplex.Messaging.Server.Stats (PeriodStatsData (..), ServerStatsData (..)) import Simplex.Messaging.Transport import Simplex.Messaging.Util (whenM) @@ -66,8 +69,8 @@ serverTests t@(ATransport t') = do describe "GET & SUB commands" $ testGetSubCommands t' describe "Exceeding queue quota" $ testExceedQueueQuota t' describe "Store log" $ testWithStoreLog t - xdescribe "Restore messages" $ testRestoreMessages t - xdescribe "Restore messages (old / v2)" $ testRestoreExpireMessages t + describe "Restore messages" $ testRestoreMessages t + describe "Restore messages (old / v2)" $ testRestoreExpireMessages t describe "Timing of AUTH error" $ testTiming t describe "Message notifications" $ testMessageNotifications t describe "Message expiration" $ do @@ -780,6 +783,7 @@ testRestoreExpireMessages at@(ATransport t) = pure () logSize testStoreLogFile `shouldReturn` 2 + exportStoreMessages msgs <- B.readFile testStoreMsgsFile length (B.lines msgs) `shouldBe` 4 @@ -788,6 +792,7 @@ testRestoreExpireMessages at@(ATransport t) = withSmpServerConfigOn at cfg1 testPort . runTest t $ \_ -> pure () logSize testStoreLogFile `shouldReturn` 1 + exportStoreMessages msgs' <- B.readFile testStoreMsgsFile msgs' `shouldBe` msgs @@ -797,12 +802,17 @@ testRestoreExpireMessages at@(ATransport t) = 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) + exportStoreMessages + 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 + exportStoreMessages = do + ms <- newMsgStore testJournalStoreCfg {quota = 4} + removeFileIfExists testStoreMsgsFile + exportMessages ms testStoreMsgsFile False runTest :: Transport c => TProxy c -> (THandleSMP c 'TClient -> IO ()) -> ThreadId -> Expectation runTest _ test' server = do testSMPClient test' `shouldReturn` () diff --git a/tests/Test.hs b/tests/Test.hs index 89672cd4f..6ae4edb41 100644 --- a/tests/Test.hs +++ b/tests/Test.hs @@ -61,7 +61,7 @@ main = do describe "TRcvQueues tests" tRcvQueuesTests describe "Util tests" utilTests describe "SMP server via TLS" $ serverTests (transport @TLS) - describe "SMP server via WebSockets" $ serverTests (transport @WS) + xdescribe "SMP server via WebSockets" $ serverTests (transport @WS) describe "Notifications server" $ ntfServerTests (transport @TLS) describe "SMP client agent" $ agentTests (transport @TLS) describe "SMP proxy" smpProxyTests