diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index 3e8f556b5..214cd0790 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/src/Simplex/Messaging/Server.hs @@ -32,11 +32,13 @@ module Simplex.Messaging.Server runSMPServerBlocking, importMessages, exportMessages, + printMessageStats, disconnectTransport, verifyCmdAuthorization, dummyVerifyCmd, randomId, AttachHTTP, + MessageStats (..), ) where @@ -91,6 +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.STM import Simplex.Messaging.Server.MsgStore.Types import Simplex.Messaging.Server.NtfStore @@ -139,13 +142,24 @@ runSMPServerBlocking started cfg attachHTTP_ = newEnv cfg >>= runReaderT (smpSer type M a = ReaderT Env IO a type AttachHTTP = Socket -> TLS.Context -> IO () +data MessageStats = MessageStats + { storedMsgsCount :: Int, + expiredMsgsCount :: Int, + storedQueues :: Int + } + +newMessageStats :: MessageStats +newMessageStats = MessageStats 0 0 0 + smpServer :: TMVar Bool -> ServerConfig -> Maybe AttachHTTP -> M () smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHTTP_ = do s <- asks server pa <- asks proxyAgent - expiredMsgs <- restoreServerMessages - expiredNtfs <- restoreServerNtfs - restoreServerStats expiredMsgs expiredNtfs + msgStats <- restoreServerMessages + ntfStats <- restoreServerNtfs + liftIO $ printMessageStats "messages" msgStats + liftIO $ printMessageStats "notifications" ntfStats + restoreServerStats msgStats ntfStats raceAny_ ( serverThread s "server subscribedQ" subscribedQ subscribers subClients pendingSubEvents subscriptions cancelSub : serverThread s "server ntfSubscribedQ" ntfSubscribedQ Env.notifiers ntfSubClients pendingNtfSubEvents ntfSubscriptions (\_ -> pure ()) @@ -366,9 +380,11 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT liftIO $ forever $ do threadDelay' interval old <- expireBeforeEpoch expCfg - void $ withActiveMsgQueues ms $ \_ q -> do + void $ withActiveMsgQueues ms (\_ -> expireQueueMsgs stats old) 0 + where + expireQueueMsgs stats old q acc = runExceptT (deleteExpiredMsgs q old) >>= \case - Right deleted -> deleted <$ atomicModifyIORef'_ (msgExpired stats) (+ deleted) + Right deleted -> (acc + deleted) <$ atomicModifyIORef'_ (msgExpired stats) (+ deleted) Left _ -> pure 0 expireNtfsThread :: ServerConfig -> M () @@ -1435,7 +1451,7 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s q <- getMsgQueue ms $ recipientId qr expireMessages q messageExpiration stats msg <- liftIO $ mkMessage msgId body - writeMsg q True msg + writeMsg ms q True msg case msg_ of Left e -> pure $ err e Right Nothing -> do @@ -1722,21 +1738,47 @@ saveServerMessages drainMsgs = 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 + total <- liftIO $ withFile f WriteMode $ \h -> withAllMsgQueues ms (saveQueueMsgs h) 0 logInfo $ "messages saved: " <> tshow total where - saveQueueMsgs h rId q = getMessages q >>= \msgs -> length msgs <$ BLD.hPutBuilder h (encodeMessages rId msgs) + saveQueueMsgs h rId q acc = getMessages 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 Int -restoreServerMessages = - asks msgStore >>= \case - AMS SMSMemory ms@STMMsgStore {storeConfig = STMStoreConfig {storePath = Just f}} -> do - old_ <- asks (messageExpiration . config) $>>= (liftIO . fmap Just . expireBeforeEpoch) - liftIO $ ifM (doesFileExist f) (importMessages ms f old_) (pure 0) - _ -> logInfo "using journal message storage" $> 0 +restoreServerMessages :: M MessageStats +restoreServerMessages = do + old_ <- asks (messageExpiration . config) $>>= (liftIO . fmap Just . expireBeforeEpoch) + asks msgStore >>= liftIO . processMessages old_ + where + processMessages :: Maybe Int64 -> AMsgStore -> IO MessageStats + processMessages old_ = \case + AMS SMSMemory ms@STMMsgStore {storeConfig = STMStoreConfig {storePath}} -> case storePath of + Just f -> ifM (doesFileExist f) (importMessages ms f old_) (pure newMessageStats) + Nothing -> pure newMessageStats + AMS SMSJournal ms -> case old_ of + Just old -> do + logInfo "expiring journal store messages..." + (storedMsgsCount, expiredMsgsCount, storedQueues) <- withAllMsgQueues ms (\_ -> processExpireQueue old) (0, 0, 0) + pure MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues} + Nothing -> do + logInfo "validating journal store messages..." + (storedMsgsCount, storedQueues) <- withAllMsgQueues ms (\_ -> processValidateQueue) (0, 0) + pure MessageStats {storedMsgsCount, expiredMsgsCount = 0, storedQueues} + where + processExpireQueue old q (!stored, !expired, !qCount) = + 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 + exitFailure + where + expireQueue = do + expired'' <- deleteExpiredMsgs q old + stored'' <- liftIO $ getQueueSize q + liftIO $ closeMsgQueue q + pure (stored'', expired'') + processValidateQueue q (!stored, !qCount) = getQueueSize q >>= \stored' -> pure (stored + stored', qCount + 1) -importMessages :: MsgStoreClass s => s -> FilePath -> Maybe Int64 -> IO Int +importMessages :: MsgStoreClass s => s -> FilePath -> Maybe Int64 -> IO MessageStats 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 @@ -1744,19 +1786,18 @@ importMessages ms f old_ = do putStrLn "" logError . T.pack $ "error restoring messages: " <> e liftIO exitFailure - Right (count, total, expired) -> do - putStrLn "" + Right (lineCount, storedMsgsCount, expiredMsgsCount) -> do + putStrLn $ "Processed " <> show lineCount <> " lines" 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 + storedQueues <- M.size <$> readTVarIO (activeMsgQueues ms) + pure MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues} 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 + restoreMsg (!lineCount, !stored, !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 @@ -1766,16 +1807,20 @@ importMessages ms f old_ = do q <- getMsgQueue ms rId (isExpired, logFull) <- case msg of Message {msgTs} - | maybe True (systemSeconds msgTs >=) old_ -> (False,) . isNothing <$> writeMsg q False msg + | maybe True (systemSeconds msgTs >=) old_ -> (False,) . isNothing <$> writeMsg ms q False msg | otherwise -> pure (True, False) - MessageQuota {} -> writeMsg q False msg $> (False, False) + MessageQuota {} -> writeMsg ms 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') + let !stored' = if logFull || isExpired then stored else stored + 1 + !expired' = if isExpired then expired + 1 else expired + pure (lineCount + 1, stored', expired') msgErr :: Show e => String -> e -> String msgErr op e = op <> " error (" <> show e <> "): " <> B.unpack (B.take 100 s) +printMessageStats :: T.Text -> MessageStats -> IO () +printMessageStats name MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues} = + logInfo $ name <> " stored: " <> tshow storedMsgsCount <> ", expired: " <> tshow expiredMsgsCount <> ", queues: " <> tshow storedQueues + saveServerNtfs :: M () saveServerNtfs = asks (storeNtfsFile . config) >>= mapM_ saveNtfs where @@ -1790,33 +1835,37 @@ saveServerNtfs = asks (storeNtfsFile . config) >>= mapM_ saveNtfs saveQueueNtfs h (nId, v) = BLD.hPutBuilder h . encodeNtfs nId . reverse =<< readTVarIO v encodeNtfs nId = mconcat . map (\ntf -> BLD.byteString (strEncode $ NLRv1 nId ntf) <> BLD.char8 '\n') -restoreServerNtfs :: M Int +restoreServerNtfs :: M MessageStats restoreServerNtfs = asks (storeNtfsFile . config) >>= \case - Just f -> ifM (doesFileExist f) (restoreNtfs f) (pure 0) - Nothing -> pure 0 + Just f -> ifM (doesFileExist f) (restoreNtfs f) (pure newMessageStats) + Nothing -> pure newMessageStats where restoreNtfs f = do logInfo $ "restoring notifications from file " <> T.pack f ns <- asks ntfStore old <- asks (notificationExpiration . config) >>= liftIO . expireBeforeEpoch - runExceptT (liftIO (LB.readFile f) >>= foldM (restoreNtf ns old) 0 . LB.lines) >>= \case - Left e -> do - logError . T.pack $ "error restoring notifications: " <> e - liftIO exitFailure - Right expired -> do - renameFile f $ f <> ".bak" - logInfo "notifications restored" - pure expired + liftIO $ + LB.readFile f >>= runExceptT . foldM (restoreNtf ns old) (0, 0, 0) . LB.lines >>= \case + Left e -> do + logError . T.pack $ "error restoring notifications: " <> e + liftIO exitFailure + Right (lineCount, storedMsgsCount, expiredMsgsCount) -> do + renameFile f $ f <> ".bak" + let NtfStore ns' = ns + storedQueues <- M.size <$> readTVarIO ns' + logInfo $ "notifications restored, " <> tshow lineCount <> " lines processed" + pure MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues} where - restoreNtf ns old !expired s' = do + restoreNtf :: NtfStore -> Int64 -> (Int, Int, Int) -> LB.ByteString -> ExceptT String IO (Int, Int, Int) + restoreNtf ns old (!lineCount, !stored, !expired) s' = do NLRv1 nId ntf <- liftEither . first (ntfErr "parsing") $ strDecode s liftIO $ addToNtfs nId ntf where s = LB.toStrict s' addToNtfs nId ntf@MsgNtf {ntfTs} - | systemSeconds ntfTs < old = pure (expired + 1) - | otherwise = storeNtf ns nId ntf $> expired + | systemSeconds ntfTs < old = pure (lineCount + 1, stored, expired + 1) + | otherwise = storeNtf ns nId ntf $> (lineCount + 1, stored + 1, expired) ntfErr :: Show e => String -> e -> String ntfErr op e = op <> " error (" <> show e <> "): " <> B.unpack (B.take 100 s) @@ -1830,26 +1879,25 @@ saveServerStats = B.writeFile f $ strEncode stats logInfo "server stats saved" -restoreServerStats :: Int -> Int -> M () -restoreServerStats expiredMsgs expiredNtfs = asks (serverStatsBackupFile . config) >>= mapM_ restoreStats +restoreServerStats :: MessageStats -> MessageStats -> M () +restoreServerStats msgStats ntfStats = asks (serverStatsBackupFile . config) >>= mapM_ restoreStats where restoreStats f = whenM (doesFileExist f) $ do logInfo $ "restoring server stats from file " <> T.pack f liftIO (strDecode <$> B.readFile f) >>= \case - Right d@ServerStatsData {_qCount = statsQCount} -> do + Right d@ServerStatsData {_qCount = statsQCount, _msgCount = statsMsgCount, _ntfCount = statsNtfCount} -> do s <- asks serverStats _qCount <- fmap M.size . readTVarIO . queues =<< asks queueStore - _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} + let _msgCount = storedMsgsCount msgStats + _ntfCount = storedMsgsCount ntfStats + liftIO $ setServerStats s d {_qCount, _msgCount, _ntfCount, _msgExpired = _msgExpired d + expiredMsgsCount msgStats, _msgNtfExpired = _msgNtfExpired d + expiredMsgsCount ntfStats} renameFile f $ f <> ".bak" logInfo "server stats restored" - when (_qCount /= statsQCount) $ logWarn $ "Queue count differs: stats: " <> tshow statsQCount <> ", store: " <> tshow _qCount - logInfo $ "Restored " <> tshow _msgCount <> " messages in " <> tshow _qCount <> " queues" + compareCounts "Queue" statsQCount _qCount + compareCounts "Message" statsMsgCount _msgCount + compareCounts "Notification" statsNtfCount _ntfCount Left e -> do logInfo $ "error restoring server stats: " <> T.pack e liftIO exitFailure + compareCounts name statsCnt storeCnt = + when (statsCnt /= storeCnt) $ logWarn $ name <> " count differs: stats: " <> tshow statsCnt <> ", store: " <> tshow storeCnt \ No newline at end of file diff --git a/src/Simplex/Messaging/Server/Main.hs b/src/Simplex/Messaging/Server/Main.hs index ddd51e815..f1f0808fa 100644 --- a/src/Simplex/Messaging/Server/Main.hs +++ b/src/Simplex/Messaging/Server/Main.hs @@ -39,7 +39,7 @@ 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, exportMessages, importMessages, runSMPServer) +import Simplex.Messaging.Server (AttachHTTP, exportMessages, importMessages, printMessageStats, runSMPServer) import Simplex.Messaging.Server.CLI import Simplex.Messaging.Server.Env.STM import Simplex.Messaging.Server.Expiration @@ -99,8 +99,9 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath = ("WARNING: message log file " <> storeMsgsFilePath <> " will be imported to journal directory " <> storeMsgsJournalDir) "Messages not imported" ms <- newJournalMsgStore - void $ importMessages ms storeMsgsFilePath Nothing -- no expiration + msgStats <- importMessages ms storeMsgsFilePath Nothing -- no expiration putStrLn "Import completed" + printMessageStats "Messages" msgStats 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`" diff --git a/src/Simplex/Messaging/Server/MsgStore/Journal.hs b/src/Simplex/Messaging/Server/MsgStore/Journal.hs index 29a022668..e0ba446ad 100644 --- a/src/Simplex/Messaging/Server/MsgStore/Journal.hs +++ b/src/Simplex/Messaging/Server/MsgStore/Journal.hs @@ -10,16 +10,19 @@ {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE StandaloneDeriving #-} {-# LANGUAGE TypeFamilies #-} module Simplex.Messaging.Server.MsgStore.Journal ( JournalMsgStore (msgQueues, random), - JournalMsgQueue, + JournalMsgQueue (queueDirectory), JournalStoreConfig (..), getQueueMessages, + closeMsgQueue, -- below are exported for tests MsgQueueState (..), JournalState (..), + SJournalType (..), msgQueueDirectory, readWriteQueueState, newMsgQueueState, @@ -54,6 +57,7 @@ import Simplex.Messaging.TMap (TMap) import qualified Simplex.Messaging.TMap as TM import Simplex.Messaging.Util (ifM, tshow, ($>>=)) import System.Directory +import System.Exit import System.FilePath (()) import System.IO (BufferMode (..), Handle, IOMode (..), SeekMode (..), stdout) import qualified System.IO as IO @@ -78,22 +82,20 @@ data JournalStoreConfig = JournalStoreConfig } data JournalMsgQueue = JournalMsgQueue - { config :: JournalStoreConfig, - queueDirectory :: FilePath, + { queueDirectory :: FilePath, queueLock :: Lock, state :: TVar MsgQueueState, - -- Last message and length incl. newline + -- tipMsg contains 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. + -- It prevents reading each message twice, + -- and reading it after it was just written. tipMsg :: TVar (Maybe (Maybe (Message, Int))), - handles :: TVar (Maybe MsgQueueHandles), - random :: TVar StdGen + handles :: TVar (Maybe MsgQueueHandles) } data MsgQueueState = MsgQueueState - { writeState :: JournalState, - readState :: JournalState, + { readState :: JournalState 'JTRead, + writeState :: JournalState 'JTWrite, canWrite :: Bool, size :: Int } @@ -105,21 +107,49 @@ data MsgQueueHandles = MsgQueueHandles writeHandle :: Maybe Handle -- optional, used when write file is different from read file } -data JournalState = JournalState - { journalId :: ByteString, +data JournalState t = JournalState + { journalType :: SJournalType t, + journalId :: ByteString, msgPos :: Int, msgCount :: Int, - bytePos :: Int + bytePos :: Int, + byteCount :: Int } deriving (Show) +data JournalType = JTRead | JTWrite + +data SJournalType (t :: JournalType) where + SJTRead :: SJournalType 'JTRead + SJTWrite :: SJournalType 'JTWrite + +class JournalTypeI t where sJournalType :: SJournalType t + +instance JournalTypeI 'JTRead where sJournalType = SJTRead + +instance JournalTypeI 'JTWrite where sJournalType = SJTWrite + +deriving instance Show (SJournalType t) + newMsgQueueState :: ByteString -> MsgQueueState newMsgQueueState journalId = - let st = newJournalState journalId - in MsgQueueState {writeState = st, readState = st, canWrite = True, size = 0} + MsgQueueState + { writeState = newJournalState journalId, + readState = newJournalState journalId, + canWrite = True, + size = 0 + } -newJournalState :: ByteString -> JournalState -newJournalState journalId = JournalState {journalId, msgPos = 0, msgCount = 0, bytePos = 0} +newJournalState :: JournalTypeI t => ByteString -> JournalState t +newJournalState journalId = + JournalState + { journalType = sJournalType, + journalId, + msgPos = 0, + msgCount = 0, + bytePos = 0, + byteCount = 0 + } journalFilePath :: FilePath -> ByteString -> FilePath journalFilePath dir journalId = dir (msgLogFileName <> "." <> B.unpack journalId <> logFileExt) @@ -139,15 +169,15 @@ instance StrEncoding MsgQueueState where 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] +instance JournalTypeI t => StrEncoding (JournalState t) where + strEncode JournalState {journalId, msgPos, msgCount, bytePos, byteCount} = + B.concat [journalId, ",", strEncodeList [msgPos, msgCount, bytePos, byteCount]] strP = do journalId <- A.takeTill (== ',') - msgPos <- A.char ',' *> strP - msgCount <- A.char ',' *> strP - bytePos <- A.char ',' *> strP - pure JournalState {journalId, msgPos, msgCount, bytePos} + A.char ',' *> strListP >>= \case + [msgPos, msgCount, bytePos, byteCount] -> + pure JournalState {journalType = sJournalType, journalId, msgPos, msgCount, bytePos, byteCount} + _ -> fail "bad JournalState" queueLogFileName :: String queueLogFileName = "queue_state" @@ -158,11 +188,11 @@ msgLogFileName = "messages" logFileExt :: String logFileExt = ".log" -newtype NonAtomicIO a = NonAtomicIO (IO a) +newtype StoreIO a = StoreIO (IO a) deriving newtype (Functor, Applicative, Monad) instance MsgStoreClass JournalMsgStore where - type StoreMonad JournalMsgStore = NonAtomicIO + type StoreMonad JournalMsgStore = StoreIO type MsgQueue JournalMsgStore = JournalMsgQueue type MsgStoreConfig JournalMsgStore = JournalStoreConfig @@ -173,33 +203,36 @@ instance MsgStoreClass JournalMsgStore where msgQueues <- TM.emptyIO pure JournalMsgStore {config, random, queueLocks, msgQueues} - closeMsgStore st = readTVarIO (msgQueues st) >>= mapM_ closeMsgQueue_ + 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 + -- 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. + withAllMsgQueues :: JournalMsgStore -> (RecipientId -> JournalMsgQueue -> a -> IO a) -> a -> IO a + withAllMsgQueues ms@JournalMsgStore {config} action res = ifM (doesDirectoryExist storePath) processStore (pure res) where + processStore = do + closeMsgStore ms + lock <- createLockIO -- the same lock is used for all queues + dirs <- zip [0..] <$> listQueueDirs 0 ("", storePath) + let count = length dirs + res' <- foldM (processQueue lock count) res dirs + progress count count + putStrLn "" + pure res' JournalStoreConfig {storePath, pathParts} = config - processQueue lock count !total (i :: Int, (queueId, dir)) = do + processQueue lock count acc (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' + q <- openMsgQueue ms dir lock + acc' <- case strDecode $ B.pack queueId of + Right rId -> action rId q acc + Left e -> do + putStrLn ("Error: message queue directory " <> dir <> " is invalid: " <> e) + exitFailure + closeMsgQueue q + pure acc' progress i count = do putStr $ "Processed: " <> show i <> "/" <> show count <> " queues\r" IO.hFlush stdout @@ -216,21 +249,21 @@ instance MsgStoreClass JournalMsgStore where (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 + logQueueStates st = withActiveMsgQueues st (\_ q _ -> logState q) () + where + logState q = + readTVarIO (handles q) + >>= maybe (pure ()) (\hs -> readTVarIO (state q) >>= logQueueState (stateHandle hs)) getMsgQueue :: JournalMsgStore -> RecipientId -> ExceptT ErrorType IO JournalMsgQueue - getMsgQueue store@JournalMsgStore {queueLocks, msgQueues, random} rId = + getMsgQueue ms@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 + let dir = msgQueueDirectory ms rId queueLock <- atomically $ getMapLock queueLocks rId - q <- ifM (doesDirectoryExist dir) (openMsgQueue store dir queueLock) (createQ dir queueLock) + q <- ifM (doesDirectoryExist dir) (openMsgQueue ms dir queueLock) (createQ dir queueLock) atomically $ TM.insert rId q msgQueues pure q where @@ -239,24 +272,24 @@ instance MsgStoreClass JournalMsgStore where -- 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) + mkJournalQueue 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 + delMsgQueue ms rId = withLockMap (queueLocks ms) rId "delMsgQueue" $ do + void $ deleteMsgQueue_ ms rId + removeQueueDirectory ms rId delMsgQueueSize :: JournalMsgStore -> RecipientId -> IO Int - delMsgQueueSize st rId = withLockMap (queueLocks st) rId "delMsgQueue" $ do - state_ <- closeMsgQueue st rId + delMsgQueueSize ms rId = withLockMap (queueLocks ms) rId "delMsgQueue" $ do + state_ <- deleteMsgQueue_ ms rId sz <- maybe (pure $ -1) (fmap size . readTVarIO) state_ - removeQueueDirectory st rId + removeQueueDirectory ms 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 + getMsg msgs hs = chooseReadJournal q drainMsgs hs >>= maybe (pure msgs) readMsg where readMsg (rs, h) = do -- TODO handle errors @@ -264,10 +297,10 @@ instance MsgStoreClass JournalMsgStore where -- 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 + (msg :) <$> getMsg msgs hs - writeMsg :: JournalMsgQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool)) - writeMsg q@JournalMsgQueue {queueDirectory, handles, config, random} logState !msg = + 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 st@MsgQueueState {canWrite, size} <- readTVarIO (state q) let empty = size == 0 @@ -279,7 +312,7 @@ instance MsgStoreClass JournalMsgStore where else writeToJournal st canWrt' msgQuota $> Nothing else pure Nothing where - JournalStoreConfig {quota, maxMsgCount} = config + JournalStoreConfig {quota, maxMsgCount} = config ms 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' @@ -306,7 +339,7 @@ instance MsgStoreClass JournalMsgStore where atomically $ writeTVar handles $ Just hs pure hs switchWriteJournal hs = do - journalId <- newJournalId random + journalId <- newJournalId $ random ms wh <- createNewJournal queueDirectory journalId atomically $ writeTVar handles $ Just $ hs {writeHandle = Just wh} pure (newJournalState journalId, wh) @@ -314,9 +347,9 @@ instance MsgStoreClass JournalMsgStore where getQueueSize :: JournalMsgQueue -> IO Int getQueueSize JournalMsgQueue {state} = size <$> readTVarIO state - tryPeekMsg_ :: JournalMsgQueue -> NonAtomicIO (Maybe Message) + tryPeekMsg_ :: JournalMsgQueue -> StoreIO (Maybe Message) tryPeekMsg_ q@JournalMsgQueue {tipMsg, handles} = - NonAtomicIO $ readTVarIO handles $>>= chooseReadJournal q True $>>= peekMsg + StoreIO $ readTVarIO handles $>>= chooseReadJournal q True $>>= peekMsg where peekMsg (rs, h) = readTVarIO tipMsg >>= maybe readMsg (pure . fmap fst) where @@ -328,16 +361,16 @@ instance MsgStoreClass JournalMsgStore where 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 $ + tryDeleteMsg_ :: JournalMsgQueue -> StoreIO () + tryDeleteMsg_ q@JournalMsgQueue {tipMsg, handles} = StoreIO $ 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 + isolateQueue :: JournalMsgQueue -> String -> StoreIO a -> ExceptT ErrorType IO a + isolateQueue mq op (StoreIO a) = tryStore op $ withLock' (queueLock mq) op $ a tryStore :: String -> IO a -> ExceptT ErrorType IO a tryStore op a = @@ -347,15 +380,15 @@ tryStore op a = in logError ("STORE ERROR " <> T.pack e') $> Left (STORE e') openMsgQueue :: JournalMsgStore -> FilePath -> Lock -> IO JournalMsgQueue -openMsgQueue store dir queueLock = do +openMsgQueue ms dir queueLock = do let statePath = dir (queueLogFileName <> logFileExt) - (st, sh) <- readWriteQueueState store statePath + (st, sh) <- readWriteQueueState ms statePath (st', rh, wh_) <- openJournals dir st let hs = MsgQueueHandles {stateHandle = sh, readHandle = rh, writeHandle = wh_} - mkJournalQueue store dir queueLock (st', Just hs) + mkJournalQueue dir queueLock (st', Just hs) -mkJournalQueue :: JournalMsgStore -> FilePath -> Lock -> (MsgQueueState, Maybe MsgQueueHandles) -> IO JournalMsgQueue -mkJournalQueue JournalMsgStore {random, config} dir queueLock (st, hs_) = do +mkJournalQueue :: FilePath -> Lock -> (MsgQueueState, Maybe MsgQueueHandles) -> IO JournalMsgQueue +mkJournalQueue dir queueLock (st, hs_) = do state <- newTVarIO st tipMsg <- newTVarIO Nothing handles <- newTVarIO hs_ @@ -363,16 +396,14 @@ mkJournalQueue JournalMsgStore {random, config} dir queueLock (st, hs_) = do -- to avoid map lookup on queue operations pure JournalMsgQueue - { config, - queueDirectory = dir, + { queueDirectory = dir, queueLock, state, tipMsg, - handles, - random + handles } -chooseReadJournal :: JournalMsgQueue -> Bool -> MsgQueueHandles -> IO (Maybe (JournalState, Handle)) +chooseReadJournal :: JournalMsgQueue -> Bool -> MsgQueueHandles -> IO (Maybe (JournalState 'JTRead, Handle)) chooseReadJournal q log' hs = do st@MsgQueueState {writeState = ws, readState = rs} <- readTVarIO (state q) case writeHandle hs of @@ -390,8 +421,8 @@ chooseReadJournal q log' hs = do updateQueueState :: JournalMsgQueue -> Bool -> MsgQueueHandles -> MsgQueueState -> IO () updateQueueState q log' hs st = do - atomically $ writeTVar (state q) 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' @@ -454,9 +485,11 @@ openJournals dir st@MsgQueueState {readState = rs, writeState = ws} = do st' <- fixWriteFileSize wh 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 @@ -465,7 +498,7 @@ openJournals dir st@MsgQueueState {readState = rs, writeState = ws} = do | sz' == sz -> pure st | otherwise -> pure st -- TODO re-read file to recover what is possible ??? -removeJournal :: FilePath -> JournalState -> IO () +removeJournal :: FilePath -> JournalState t -> 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) @@ -522,13 +555,13 @@ readWriteQueueState JournalMsgStore {random, config} statePath = logQueueState sh st pure (st, sh) -closeMsgQueue :: JournalMsgStore -> RecipientId -> IO (Maybe (TVar MsgQueueState)) -closeMsgQueue st rId = +deleteMsgQueue_ :: JournalMsgStore -> RecipientId -> IO (Maybe (TVar MsgQueueState)) +deleteMsgQueue_ st rId = atomically (TM.lookupDelete rId (msgQueues st)) - >>= mapM (\q -> closeMsgQueue_ q $> state q) + >>= mapM (\q -> closeMsgQueue q $> state q) -closeMsgQueue_ :: JournalMsgQueue -> IO () -closeMsgQueue_ q = readTVarIO (handles q) >>= mapM_ closeHandles +closeMsgQueue :: JournalMsgQueue -> IO () +closeMsgQueue q = readTVarIO (handles q) >>= mapM_ closeHandles where closeHandles (MsgQueueHandles sh rh wh_) = do hClose sh diff --git a/src/Simplex/Messaging/Server/MsgStore/STM.hs b/src/Simplex/Messaging/Server/MsgStore/STM.hs index df4507175..17591650c 100644 --- a/src/Simplex/Messaging/Server/MsgStore/STM.hs +++ b/src/Simplex/Messaging/Server/MsgStore/STM.hs @@ -93,8 +93,8 @@ instance MsgStoreClass STMMsgStore where mapM_ (writeTQueue q) msgs pure msgs - writeMsg :: STMMsgQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool)) - writeMsg STMMsgQueue {msgQueue = q, quota, canWrite, size} _logState !msg = liftIO $ atomically $ do + writeMsg :: STMMsgStore -> 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 @@ -122,5 +122,5 @@ instance MsgStoreClass STMMsgStore where Just _ -> modifyTVar' size (subtract 1) _ -> pure () - atomicQueue :: STMMsgQueue -> String -> STM a -> ExceptT ErrorType IO a - atomicQueue _ _ = liftIO . atomically + isolateQueue :: STMMsgQueue -> String -> STM a -> ExceptT ErrorType IO a + isolateQueue _ _ = liftIO . atomically diff --git a/src/Simplex/Messaging/Server/MsgStore/Types.hs b/src/Simplex/Messaging/Server/MsgStore/Types.hs index 68693883d..dbaefba6f 100644 --- a/src/Simplex/Messaging/Server/MsgStore/Types.hs +++ b/src/Simplex/Messaging/Server/MsgStore/Types.hs @@ -24,17 +24,17 @@ class Monad (StoreMonad s) => MsgStoreClass s where newMsgStore :: MsgStoreConfig s -> IO s closeMsgStore :: s -> IO () activeMsgQueues :: s -> TMap RecipientId (MsgQueue s) - withAllMsgQueues :: s -> (RecipientId -> MsgQueue s -> IO Int) -> IO Int + withAllMsgQueues :: s -> (RecipientId -> MsgQueue s -> a -> IO a) -> a -> IO a 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)) + writeMsg :: s -> 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 + isolateQueue :: MsgQueue s -> String -> StoreMonad s a -> ExceptT ErrorType IO a data MSType = MSMemory | MSJournal @@ -44,16 +44,16 @@ data SMSType :: MSType -> Type where 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) +withActiveMsgQueues :: MsgStoreClass s => s -> (RecipientId -> MsgQueue s -> a -> IO a) -> a -> IO a +withActiveMsgQueues st f a = readTVarIO (activeMsgQueues st) >>= M.foldrWithKey (\k v -> (>>= f k v)) (pure a) tryPeekMsg :: MsgStoreClass s => MsgQueue s -> ExceptT ErrorType IO (Maybe Message) -tryPeekMsg mq = atomicQueue mq "tryPeekMsg" $ tryPeekMsg_ mq +tryPeekMsg mq = isolateQueue mq "tryPeekMsg" $ tryPeekMsg_ mq {-# INLINE tryPeekMsg #-} tryDelMsg :: MsgStoreClass s => MsgQueue s -> MsgId -> ExceptT ErrorType IO (Maybe Message) tryDelMsg mq msgId' = - atomicQueue mq "tryDelMsg" $ + isolateQueue mq "tryDelMsg" $ tryPeekMsg_ mq >>= \case msg_@(Just msg) | msgId msg == msgId' -> @@ -63,7 +63,7 @@ tryDelMsg mq msgId' = -- 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" $ + isolateQueue mq "tryDelPeekMsg" $ tryPeekMsg_ mq >>= \case msg_@(Just msg) | msgId msg == msgId' -> (msg_,) <$> (tryDeleteMsg_ mq >> tryPeekMsg_ mq) @@ -71,7 +71,7 @@ tryDelPeekMsg mq msgId' = _ -> pure (Nothing, Nothing) deleteExpiredMsgs :: MsgStoreClass s => MsgQueue s -> Int64 -> ExceptT ErrorType IO Int -deleteExpiredMsgs mq old = atomicQueue mq "deleteExpiredMsgs" $ loop 0 +deleteExpiredMsgs mq old = isolateQueue mq "deleteExpiredMsgs" $ loop 0 where loop dc = tryPeekMsg_ mq >>= \case diff --git a/tests/AgentTests/FunctionalAPITests.hs b/tests/AgentTests/FunctionalAPITests.hs index 38a187ec5..425e93f29 100644 --- a/tests/AgentTests/FunctionalAPITests.hs +++ b/tests/AgentTests/FunctionalAPITests.hs @@ -1042,7 +1042,7 @@ testAllowConnectionClientRestart t = do bob <- getSMPAgentClient' 2 agentCfg initAgentServersSrv2 testDB2 withSmpServerStoreLogOn t testPort $ \_ -> do (aliceId, bobId, confId) <- - withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2} testPort2 $ \_ -> do + withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2} testPort2 $ \_ -> do runRight $ do (bobId, qInfo) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe (aliceId, sqSecured) <- joinConnection bob 1 True qInfo "bob's connInfo" SMSubscribe @@ -1064,8 +1064,7 @@ testAllowConnectionClientRestart t = do alice2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB runRight_ $ subscribeConnection alice2 bobId threadDelay 500000 - - withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2} testPort2 $ \_ -> do + withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2} testPort2 $ \_ -> do runRight $ do ("", "", UP _ _) <- nGet bob get alice2 ##> ("", bobId, CON) diff --git a/tests/CoreTests/MsgStoreTests.hs b/tests/CoreTests/MsgStoreTests.hs index 150312a60..dd54b1e33 100644 --- a/tests/CoreTests/MsgStoreTests.hs +++ b/tests/CoreTests/MsgStoreTests.hs @@ -21,7 +21,7 @@ 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 (MessageStats (..), exportMessages, importMessages) import Simplex.Messaging.Server.MsgStore.Journal import Simplex.Messaging.Server.MsgStore.STM import Simplex.Messaging.Server.MsgStore.Types @@ -73,17 +73,20 @@ pattern Msg s <- Just Message {msgBody = MaxLenBS s} deriving instance Eq MsgQueueState -deriving instance Eq JournalState +deriving instance Eq (JournalState t) + +deriving instance Eq (SJournalType t) testGetQueue :: MsgStoreClass s => s -> IO () -testGetQueue st = do +testGetQueue ms = 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" + q <- getMsgQueue ms rId + let write s = writeMsg ms q True =<< mkMessage s + Just (Message {msgId = mId1}, True) <- write "message 1" + Just (Message {msgId = mId2}, False) <- write "message 2" + Just (Message {msgId = mId3}, False) <- write "message 3" Msg "message 1" <- tryPeekMsg q Msg "message 1" <- tryPeekMsg q Nothing <- tryDelMsg q mId2 @@ -99,14 +102,14 @@ testGetQueue st = do Nothing <- tryDelMsg q mId2 Nothing <- tryDelMsg q mId3 Nothing <- tryPeekMsg q - Just (Message {msgId = mId4}, True) <- writeMsg q True =<< mkMessage "message 4" + Just (Message {msgId = mId4}, True) <- write "message 4" Msg "message 4" <- tryPeekMsg q - Just (Message {msgId = mId5}, False) <- writeMsg q True =<< mkMessage "message 5" + Just (Message {msgId = mId5}, False) <- write "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" + Just (Message {msgId = mId6}, False) <- write "message 6" + Just (Message {msgId = mId7}, False) <- write "message 7" + Nothing <- write "message 8" Msg "message 5" <- tryPeekMsg q (Nothing, Msg "message 5") <- tryDelPeekMsg q mId4 (Msg "message 5", Msg "message 6") <- tryDelPeekMsg q mId5 @@ -115,65 +118,69 @@ testGetQueue st = do (Just MessageQuota {}, Nothing) <- tryDelPeekMsg q mId8 (Nothing, Nothing) <- tryDelPeekMsg q mId8 pure () - delMsgQueue st rId + delMsgQueue ms rId testChangeReadJournal :: MsgStoreClass s => s -> IO () -testChangeReadJournal st = do +testChangeReadJournal ms = 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" + q <- getMsgQueue ms rId + let write s = writeMsg ms q True =<< mkMessage s + Just (Message {msgId = mId1}, True) <- write "message 1" (Msg "message 1", Nothing) <- tryDelPeekMsg q mId1 - Just (Message {msgId = mId2}, True) <- writeMsg q True =<< mkMessage "message 2" + Just (Message {msgId = mId2}, True) <- write "message 2" (Msg "message 2", Nothing) <- tryDelPeekMsg q mId2 - Just (Message {msgId = mId3}, True) <- writeMsg q True =<< mkMessage "message 3" + Just (Message {msgId = mId3}, True) <- write "message 3" (Msg "message 3", Nothing) <- tryDelPeekMsg q mId3 - Just (Message {msgId = mId4}, True) <- writeMsg q True =<< mkMessage "message 4" + Just (Message {msgId = mId4}, True) <- write "message 4" (Msg "message 4", Nothing) <- tryDelPeekMsg q mId4 - Just (Message {msgId = mId5}, True) <- writeMsg q True =<< mkMessage "message 5" + Just (Message {msgId = mId5}, True) <- write "message 5" (Msg "message 5", Nothing) <- tryDelPeekMsg q mId5 pure () - delMsgQueue st rId + delMsgQueue ms rId testExportImportStore :: JournalMsgStore -> IO () -testExportImportStore st = do +testExportImportStore ms = 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" + let write q s = writeMsg ms q True =<< mkMessage s + q1 <- getMsgQueue ms rId1 + Just (Message {}, True) <- write q1 "message 1" + Just (Message {}, False) <- write q1 "message 2" + q2 <- getMsgQueue ms rId2 + Just (Message {}, True) <- write q2 "message 3" + Just (Message {}, False) <- write q2 "message 4" + Just (Message {}, False) <- write q2 "message 5" + Nothing <- write q2 "message 6" pure () - length <$> listDirectory (msgQueueDirectory st rId1) `shouldReturn` 2 - length <$> listDirectory (msgQueueDirectory st rId2) `shouldReturn` 2 - exportMessages st testStoreMsgsFile $ getQueueMessages False + length <$> listDirectory (msgQueueDirectory ms rId1) `shouldReturn` 2 + length <$> listDirectory (msgQueueDirectory ms rId2) `shouldReturn` 2 + exportMessages ms 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 + 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 (B.readFile testStoreMsgsFile2 `shouldReturn`) =<< B.readFile (testStoreMsgsFile <> ".bak") stmStore <- newMsgStore testSMTStoreConfig - 0 <- importMessages stmStore testStoreMsgsFile2 Nothing + MessageStats {storedMsgsCount = 6, expiredMsgsCount = 0, storedQueues = 2} <- + 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 +testQueueState ms = do g <- C.newRandom rId <- EntityId <$> atomically (C.randomBytes 24 g) - let dir = msgQueueDirectory st rId + let dir = msgQueueDirectory ms rId statePath = dir (queueLogFileName <> logFileExt) createDirectoryIfMissing True dir - state <- newMsgQueueState <$> newJournalId (random st) + state <- newMsgQueueState <$> newJournalId (random ms) withFile statePath WriteMode (`logQueueState` state) length . lines <$> readFile statePath `shouldReturn` 1 readQueueState statePath `shouldReturn` state @@ -208,7 +215,7 @@ testQueueState st = do length . lines <$> readFile statePath `shouldReturn` 1 where readQueueState statePath = do - (state, h) <- readWriteQueueState st statePath + (state, h) <- readWriteQueueState ms statePath hClose h pure state corruptFile f = do