From 56ca7bf30e813648d13cea6f3d13336bb9af25ad Mon Sep 17 00:00:00 2001 From: Evgeny Date: Sat, 26 Oct 2024 10:33:18 +0100 Subject: [PATCH] smp server: do not log progress when server starts (#1390) * smp server: do not log progress when server starts * test --- src/Simplex/Messaging/Server.hs | 26 +++++++++---------- src/Simplex/Messaging/Server/Main.hs | 4 +-- .../Messaging/Server/MsgStore/Journal.hs | 13 ++++------ src/Simplex/Messaging/Server/MsgStore/STM.hs | 2 +- .../Messaging/Server/MsgStore/Types.hs | 2 +- tests/CoreTests/MsgStoreTests.hs | 12 ++++----- tests/ServerTests.hs | 2 +- 7 files changed, 28 insertions(+), 33 deletions(-) diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index 81c995bb2..38d718fd2 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/src/Simplex/Messaging/Server.hs @@ -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 diff --git a/src/Simplex/Messaging/Server/Main.hs b/src/Simplex/Messaging/Server/Main.hs index f728e6be8..d3eab5f33 100644 --- a/src/Simplex/Messaging/Server/Main.hs +++ b/src/Simplex/Messaging/Server/Main.hs @@ -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`" diff --git a/src/Simplex/Messaging/Server/MsgStore/Journal.hs b/src/Simplex/Messaging/Server/MsgStore/Journal.hs index f98b5222a..5324c9997 100644 --- a/src/Simplex/Messaging/Server/MsgStore/Journal.hs +++ b/src/Simplex/Messaging/Server/MsgStore/Journal.hs @@ -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 diff --git a/src/Simplex/Messaging/Server/MsgStore/STM.hs b/src/Simplex/Messaging/Server/MsgStore/STM.hs index f70a1e71d..1b27b6a10 100644 --- a/src/Simplex/Messaging/Server/MsgStore/STM.hs +++ b/src/Simplex/Messaging/Server/MsgStore/STM.hs @@ -57,7 +57,7 @@ instance MsgStoreClass STMMsgStore where activeMsgQueues = msgQueues {-# INLINE activeMsgQueues #-} - withAllMsgQueues = withActiveMsgQueues + withAllMsgQueues _ = withActiveMsgQueues {-# INLINE withAllMsgQueues #-} logQueueStates _ = pure () diff --git a/src/Simplex/Messaging/Server/MsgStore/Types.hs b/src/Simplex/Messaging/Server/MsgStore/Types.hs index 718aa95ca..0486804f3 100644 --- a/src/Simplex/Messaging/Server/MsgStore/Types.hs +++ b/src/Simplex/Messaging/Server/MsgStore/Types.hs @@ -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) diff --git a/tests/CoreTests/MsgStoreTests.hs b/tests/CoreTests/MsgStoreTests.hs index fc00317a0..4a5d95eb2 100644 --- a/tests/CoreTests/MsgStoreTests.hs +++ b/tests/CoreTests/MsgStoreTests.hs @@ -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 () diff --git a/tests/ServerTests.hs b/tests/ServerTests.hs index c133d94fc..e80d39897 100644 --- a/tests/ServerTests.hs +++ b/tests/ServerTests.hs @@ -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