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

smp server: do not log progress when server starts (#1390)

Browse Source 
* smp server: do not log progress when server starts

* test
This commit is contained in:
Evgeny
2024-10-26 10:33:18 +01:00
committed by GitHub
parent 2a218675b2
commit 56ca7bf30e
7 changed files with 28 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Simplex/Messaging/Server.hs
+12 -14
View File
@@ -1739,17 +1739,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 drainMsgs
Just f -> liftIO $ exportMessages False ms f drainMsgs
Nothing -> logInfo "undelivered messages are not saved"
AMS SMSJournal ms -> do
liftIO $ closeMsgStore ms
logInfo "closed journal message storage"
exportMessages :: MsgStoreClass s => s -> FilePath -> Bool -> IO ()
exportMessages ms f drainMsgs = do
exportMessages :: MsgStoreClass s => Bool -> s -> FilePath -> Bool -> IO ()
exportMessages tty ms f drainMsgs = do
logInfo $ "saving messages to file " <> T.pack f
liftIO $ withFile f WriteMode $ \h ->
tryAny (withAllMsgQueues ms $ saveQueueMsgs h) >>= \case
tryAny (withAllMsgQueues tty ms $ saveQueueMsgs h) >>= \case
Right (Sum total) -> logInfo $ "messages saved: " <> tshow total
Left e -> do
logError $ "error exporting messages: " <> tshow e
@@ -1766,15 +1766,15 @@ processServerMessages = do
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)
Just f -> ifM (doesFileExist f) (importMessages False ms f old_) (pure newMessageStats)
Nothing -> pure newMessageStats
AMS SMSJournal ms -> case old_ of
Just old -> do
logInfo "expiring journal store messages..."
withAllMsgQueues ms $ \_ -> processExpireQueue old
withAllMsgQueues False ms $ \_ -> processExpireQueue old
Nothing -> do
logInfo "validating journal store messages..."
withAllMsgQueues ms $ \_ -> processValidateQueue
withAllMsgQueues False ms $ \_ -> processValidateQueue
where
processExpireQueue old q =
runExceptT expireQueue >>= \case
@@ -1793,8 +1793,8 @@ processServerMessages = do
processValidateQueue q =
getQueueSize q >>= \storedMsgsCount -> pure mempty {storedMsgsCount, storedQueues = 1}
importMessages :: forall s. MsgStoreClass s => s -> FilePath -> Maybe Int64 -> IO MessageStats
importMessages ms f old_ = do
importMessages :: forall s. MsgStoreClass s => Bool -> s -> FilePath -> Maybe Int64 -> IO MessageStats
importMessages tty ms f old_ = do
logInfo $ "restoring messages from file " <> T.pack f
LB.readFile f >>= runExceptT . foldM restoreMsg (0, Nothing, (0, 0, M.empty)) . zip [0..] . LB.lines >>= \case
Left e -> do
@@ -1802,19 +1802,17 @@ importMessages ms f old_ = do
logError . T.pack $ "error restoring messages: " <> e
liftIO exitFailure
Right (lineCount, _, (storedMsgsCount, expiredMsgsCount, overQuota)) -> do
putStrLn $ "Processed " <> show lineCount <> " lines"
putStrLn $ progress lineCount
renameFile f $ f <> ".bak"
mapM_ setOverQuota_ overQuota
logQueueStates ms
storedQueues <- M.size <$> readTVarIO (activeMsgQueues ms)
pure MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues}
where
progress i = do
liftIO $ putStr $ "Processed " <> show i <> " lines\r"
hFlush stdout
progress i = "Processed " <> show i <> " lines"
restoreMsg :: (Int, Maybe (RecipientId, MsgQueue s), (Int, Int, M.Map RecipientId (MsgQueue s))) -> (Int, LB.ByteString) -> ExceptT String IO (Int, Maybe (RecipientId, MsgQueue s), (Int, Int, M.Map RecipientId (MsgQueue s)))
restoreMsg (!lineCount, q_, (!stored, !expired, !overQuota)) (i, s') = do
when (i `mod` 1000 == 0) $ progress i
when (tty && i `mod` 1000 == 0) $ liftIO $ putStr (progress i <> "\r")
MLRv3 rId msg <- liftEither . first (msgErr "parsing") $ strDecode s
liftError show $ addToMsgQueue rId msg
where
src/Simplex/Messaging/Server/Main.hs
+2 -2
View File
@@ -99,7 +99,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
("WARNING: message log file " <> storeMsgsFilePath <> " will be imported to journal directory " <> storeMsgsJournalDir)
"Messages not imported"
ms <- newJournalMsgStore
msgStats <- importMessages ms storeMsgsFilePath Nothing -- no expiration
msgStats <- importMessages True ms storeMsgsFilePath Nothing -- no expiration
putStrLn "Import completed"
printMessageStats "Messages" msgStats
putStrLn $ case readMsgStoreType ini of
@@ -116,7 +116,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 False
exportMessages True ms storeMsgsFilePath False
putStrLn "Export completed"
putStrLn $ case readMsgStoreType ini of
Right (AMSType SMSMemory) -> "store_messages set to `memory`"
src/Simplex/Messaging/Server/MsgStore/Journal.hs
+5 -8
View File
@@ -216,20 +216,19 @@ instance MsgStoreClass JournalMsgStore where
-- 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 :: forall a. Monoid a => JournalMsgStore -> (RecipientId -> JournalMsgQueue -> IO a) -> IO a
withAllMsgQueues ms@JournalMsgStore {config} action = ifM (doesDirectoryExist storePath) processStore (pure mempty)
withAllMsgQueues :: forall a. Monoid a => Bool -> JournalMsgStore -> (RecipientId -> JournalMsgQueue -> IO a) -> IO a
withAllMsgQueues tty ms@JournalMsgStore {config} action = ifM (doesDirectoryExist storePath) processStore (pure mempty)
where
processStore = do
closeMsgStore ms
lock <- createLockIO -- the same lock is used for all queues
(!count, !res) <- foldQueues 0 (processQueue lock) (0, mempty) ("", storePath)
progress count
putStrLn ""
putStrLn $ progress count
pure res
JournalStoreConfig {storePath, pathParts} = config
processQueue :: Lock -> (Int, a) -> (String, FilePath) -> IO (Int, a)
processQueue queueLock (!i, !r) (queueId, dir) = do
when (i `mod` 100 == 0) $ progress i
when (tty && i `mod` 100 == 0) $ putStr (progress i <> "\r") >> IO.hFlush stdout
let statePath = msgQueueStatePath dir queueId
q <- openMsgQueue ms JMQueue {queueDirectory = dir, queueLock, statePath}
r' <- case strDecode $ B.pack queueId of
@@ -239,9 +238,7 @@ instance MsgStoreClass JournalMsgStore where
exitFailure
closeMsgQueue q
pure (i + 1, r <> r')
progress i = do
putStr $ "Processed: " <> show i <> " queues\r"
IO.hFlush stdout
progress i = "Processed: " <> show i <> " queues"
foldQueues depth f acc (queueId, path) = do
let f' = if depth == pathParts - 1 then f else foldQueues (depth + 1) f
listDirs >>= foldM f' acc
src/Simplex/Messaging/Server/MsgStore/STM.hs
+1 -1
View File
@@ -57,7 +57,7 @@ instance MsgStoreClass STMMsgStore where
activeMsgQueues = msgQueues
{-# INLINE activeMsgQueues #-}
withAllMsgQueues = withActiveMsgQueues
withAllMsgQueues _ = withActiveMsgQueues
{-# INLINE withAllMsgQueues #-}
logQueueStates _ = pure ()
src/Simplex/Messaging/Server/MsgStore/Types.hs
+1 -1
View File
@@ -26,7 +26,7 @@ class Monad (StoreMonad s) => MsgStoreClass s where
newMsgStore :: MsgStoreConfig s -> IO s
closeMsgStore :: s -> IO ()
activeMsgQueues :: s -> TMap RecipientId (MsgQueue s)
withAllMsgQueues :: Monoid a => s -> (RecipientId -> MsgQueue s -> IO a) -> IO a
withAllMsgQueues :: Monoid a => Bool -> s -> (RecipientId -> MsgQueue s -> IO a) -> IO a
logQueueStates :: s -> IO ()
logQueueState :: MsgQueue s -> IO ()
getMsgQueue :: s -> RecipientId -> ExceptT ErrorType IO (MsgQueue s)
tests/CoreTests/MsgStoreTests.hs
+6 -6
View File
@@ -164,24 +164,24 @@ testExportImportStore ms = do
pure ()
length <$> listDirectory (msgQueueDirectory ms rId1) `shouldReturn` 2
length <$> listDirectory (msgQueueDirectory ms rId2) `shouldReturn` 3
exportMessages ms testStoreMsgsFile False
exportMessages False ms testStoreMsgsFile False
renameFile testStoreMsgsFile (testStoreMsgsFile <> ".copy")
closeMsgStore ms
exportMessages ms testStoreMsgsFile False
exportMessages False ms testStoreMsgsFile False
(B.readFile testStoreMsgsFile `shouldReturn`) =<< B.readFile (testStoreMsgsFile <> ".copy")
let cfg = (testJournalStoreCfg :: JournalStoreConfig) {storePath = testStoreMsgsDir2}
ms' <- newMsgStore cfg
stats@MessageStats {storedMsgsCount = 5, expiredMsgsCount = 0, storedQueues = 2} <-
importMessages ms' testStoreMsgsFile Nothing
importMessages False ms' testStoreMsgsFile Nothing
printMessageStats "Messages" stats
length <$> listDirectory (msgQueueDirectory ms rId1) `shouldReturn` 2
length <$> listDirectory (msgQueueDirectory ms rId2) `shouldReturn` 4 -- state file is backed up, 2 message files
exportMessages ms' testStoreMsgsFile2 False
exportMessages False ms' testStoreMsgsFile2 False
(B.readFile testStoreMsgsFile2 `shouldReturn`) =<< B.readFile (testStoreMsgsFile <> ".bak")
stmStore <- newMsgStore testSMTStoreConfig
MessageStats {storedMsgsCount = 5, expiredMsgsCount = 0, storedQueues = 2} <-
importMessages stmStore testStoreMsgsFile2 Nothing
exportMessages stmStore testStoreMsgsFile False
importMessages False stmStore testStoreMsgsFile2 Nothing
exportMessages False stmStore testStoreMsgsFile False
(B.sort <$> B.readFile testStoreMsgsFile `shouldReturn`) =<< (B.sort <$> B.readFile (testStoreMsgsFile2 <> ".bak"))
testQueueState :: JournalMsgStore -> IO ()
tests/ServerTests.hs
+1 -1
View File
@@ -813,7 +813,7 @@ testRestoreExpireMessages =
AMSType SMSJournal -> do
ms <- newMsgStore testJournalStoreCfg {quota = 4}
removeFileIfExists testStoreMsgsFile
exportMessages ms testStoreMsgsFile False
exportMessages False ms testStoreMsgsFile False
AMSType SMSMemory -> pure ()
runTest :: Transport c => TProxy c -> (THandleSMP c 'TClient -> IO ()) -> ThreadId -> Expectation
runTest _ test' server = do