From 7c219455691012e4f6f18c2074df4fff1352bdd8 Mon Sep 17 00:00:00 2001 From: Evgeny Date: Sun, 3 Nov 2024 21:08:07 +0000 Subject: [PATCH] smp server: option to skip expiring messages on start, read queue state file end only (#1400) --- src/Simplex/Messaging/Server.hs | 10 ++++++---- src/Simplex/Messaging/Server/Env/STM.hs | 6 +++++- src/Simplex/Messaging/Server/Main.hs | 4 +++- src/Simplex/Messaging/Server/MsgStore/Journal.hs | 16 ++++++++++++---- tests/CoreTests/MsgStoreTests.hs | 3 ++- tests/SMPClient.hs | 1 + 6 files changed, 29 insertions(+), 11 deletions(-) diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index 9078ebace..ba91e50cf 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/src/Simplex/Messaging/Server.hs @@ -1761,20 +1761,22 @@ exportMessages tty ms f drainMsgs = do processServerMessages :: M MessageStats processServerMessages = do old_ <- asks (messageExpiration . config) $>>= (liftIO . fmap Just . expireBeforeEpoch) - asks msgStore >>= liftIO . processMessages old_ + expire <- asks $ expireMessagesOnStart . config + asks msgStore >>= liftIO . processMessages old_ expire where - processMessages :: Maybe Int64 -> AMsgStore -> IO MessageStats - processMessages old_ = \case + processMessages :: Maybe Int64 -> Bool -> AMsgStore -> IO MessageStats + processMessages old_ expire = \case AMS SMSMemory ms@STMMsgStore {storeConfig = STMStoreConfig {storePath}} -> case storePath of Just f -> ifM (doesFileExist f) (importMessages False ms f old_) (pure newMessageStats) Nothing -> pure newMessageStats - AMS SMSJournal ms -> case old_ of + AMS SMSJournal ms | expire -> case old_ of Just old -> do logInfo "expiring journal store messages..." withAllMsgQueues False ms $ \_ -> processExpireQueue old Nothing -> do logInfo "validating journal store messages..." withAllMsgQueues False ms $ \_ -> processValidateQueue + AMS SMSJournal _ -> logWarn "skipping message expiration" $> newMessageStats where processExpireQueue old q = runExceptT expireQueue >>= \case diff --git a/src/Simplex/Messaging/Server/Env/STM.hs b/src/Simplex/Messaging/Server/Env/STM.hs index b25a76a05..c46519d90 100644 --- a/src/Simplex/Messaging/Server/Env/STM.hs +++ b/src/Simplex/Messaging/Server/Env/STM.hs @@ -79,6 +79,7 @@ data ServerConfig = ServerConfig controlPortAdminAuth :: Maybe BasicAuth, -- | time after which the messages can be removed from the queues and check interval, seconds messageExpiration :: Maybe ExpirationConfig, + expireMessagesOnStart :: Bool, -- | notification expiration interval (seconds) notificationExpiration :: ExpirationConfig, -- | time after which the socket with inactive client can be disconnected (without any messages or commands, incl. PING), @@ -155,6 +156,9 @@ defaultMaxJournalMsgCount = 256 defaultMsgQueueQuota :: Int defaultMsgQueueQuota = 128 +defaultStateTailSize :: Int +defaultStateTailSize = 512 + data Env = Env { config :: ServerConfig, serverActive :: TVar Bool, @@ -284,7 +288,7 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, storeLogFile, msgSt AMSType SMSMemory -> AMS SMSMemory <$> newMsgStore STMStoreConfig {storePath = storeMsgsFile, quota = msgQueueQuota} AMSType SMSJournal -> case storeMsgsFile of Just storePath -> - let cfg = JournalStoreConfig {storePath, quota = msgQueueQuota, pathParts = journalMsgStoreDepth, maxMsgCount = maxJournalMsgCount, maxStateLines = maxJournalStateLines} + let cfg = JournalStoreConfig {storePath, quota = msgQueueQuota, pathParts = journalMsgStoreDepth, maxMsgCount = maxJournalMsgCount, maxStateLines = maxJournalStateLines, stateTailSize = defaultStateTailSize} in AMS SMSJournal <$> newMsgStore cfg Nothing -> putStrLn "Error: journal msg store require path in [STORE_LOG], restore_messages" >> exitFailure ntfStore <- NtfStore <$> TM.emptyIO diff --git a/src/Simplex/Messaging/Server/Main.hs b/src/Simplex/Messaging/Server/Main.hs index d3eab5f33..0251b6705 100644 --- a/src/Simplex/Messaging/Server/Main.hs +++ b/src/Simplex/Messaging/Server/Main.hs @@ -137,7 +137,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath = doesFileExist iniFile >>= \case True -> readIniFile iniFile >>= either exitError a _ -> exitError $ "Error: server is not initialized (" <> iniFile <> " does not exist).\nRun `" <> executableName <> " init`." - newJournalMsgStore = newMsgStore JournalStoreConfig {storePath = storeMsgsJournalDir, pathParts = journalMsgStoreDepth, quota = defaultMsgQueueQuota, maxMsgCount = defaultMaxJournalMsgCount, maxStateLines = defaultMaxJournalStateLines} + newJournalMsgStore = newMsgStore JournalStoreConfig {storePath = storeMsgsJournalDir, pathParts = journalMsgStoreDepth, quota = defaultMsgQueueQuota, maxMsgCount = defaultMaxJournalMsgCount, maxStateLines = defaultMaxJournalStateLines, stateTailSize = defaultStateTailSize} iniFile = combine cfgPath "smp-server.ini" serverVersion = "SMP server v" <> simplexMQVersion defaultServerPorts = "5223,443" @@ -238,6 +238,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath = <> ("restore_messages: " <> onOff enableStoreLog <> "\n\n") <> "# Messages and notifications expiration periods.\n" <> ("expire_messages_days: " <> tshow defMsgExpirationDays <> "\n") + <> "expire_messages_on_start: on\n" <> ("expire_ntfs_hours: " <> tshow defNtfExpirationHours <> "\n\n") <> "# Log daily server statistics to CSV file\n" <> ("log_stats: " <> onOff logStats <> "\n\n") @@ -403,6 +404,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath = defaultMessageExpiration { ttl = 86400 * readIniDefault defMsgExpirationDays "STORE_LOG" "expire_messages_days" ini }, + expireMessagesOnStart = fromMaybe True $ iniOnOff "STORE_LOG" "expire_messages_on_start" ini, notificationExpiration = defaultNtfExpiration { ttl = 3600 * readIniDefault defNtfExpirationHours "STORE_LOG" "expire_ntfs_hours" ini diff --git a/src/Simplex/Messaging/Server/MsgStore/Journal.hs b/src/Simplex/Messaging/Server/MsgStore/Journal.hs index 28e52f8e1..44297bf37 100644 --- a/src/Simplex/Messaging/Server/MsgStore/Journal.hs +++ b/src/Simplex/Messaging/Server/MsgStore/Journal.hs @@ -45,7 +45,6 @@ import qualified Data.Attoparsec.ByteString.Char8 as A import Data.Bitraversable (bimapM) 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.List (intercalate) @@ -85,7 +84,8 @@ data JournalStoreConfig = JournalStoreConfig -- When this limit is reached, the file will be changed. -- This number should be set bigger than queue quota. maxMsgCount :: Int, - maxStateLines :: Int + maxStateLines :: Int, + stateTailSize :: Int } data JMQueue = JMQueue @@ -530,7 +530,7 @@ readWriteQueueState JournalMsgStore {random, config} statePath = where tempBackup = statePath <> ".bak" readQueueState = do - ls <- LB.lines <$> LB.readFile statePath + ls <- B.lines <$> readFileTail case ls of [] -> writeNewQueueState _ -> do @@ -541,7 +541,7 @@ readWriteQueueState JournalMsgStore {random, config} statePath = logWarn $ "STORE: readWriteQueueState, empty queue state - initialized, " <> T.pack statePath st <- newMsgQueueState <$> newJournalId random writeQueueState st - useLastLine len isLastLine ls = case strDecode $ LB.toStrict $ last ls of + useLastLine len isLastLine ls = case strDecode $ last ls of Right st | len > maxStateLines config || not isLastLine -> backupWriteQueueState st @@ -571,6 +571,14 @@ readWriteQueueState JournalMsgStore {random, config} statePath = sh <- openFile statePath AppendMode closeOnException sh $ appendState sh st pure (st, sh) + readFileTail = + IO.withFile statePath ReadMode $ \h -> do + size <- IO.hFileSize h + let sz = stateTailSize config + sz' = fromIntegral sz + if size > sz' + then IO.hSeek h AbsoluteSeek (size - sz') >> B.hGet h sz + else B.hGet h (fromIntegral size) validQueueState :: MsgQueueState -> Bool validQueueState MsgQueueState {readState = rs, writeState = ws, size} diff --git a/tests/CoreTests/MsgStoreTests.hs b/tests/CoreTests/MsgStoreTests.hs index 5166ce7ef..16e3a12f0 100644 --- a/tests/CoreTests/MsgStoreTests.hs +++ b/tests/CoreTests/MsgStoreTests.hs @@ -63,7 +63,8 @@ testJournalStoreCfg = pathParts = journalMsgStoreDepth, quota = 3, maxMsgCount = 4, - maxStateLines = 2 + maxStateLines = 2, + stateTailSize = 256 } mkMessage :: MonadIO m => ByteString -> m Message diff --git a/tests/SMPClient.hs b/tests/SMPClient.hs index 0efb64c02..b28f2b804 100644 --- a/tests/SMPClient.hs +++ b/tests/SMPClient.hs @@ -132,6 +132,7 @@ cfgMS msType = controlPortUserAuth = Nothing, controlPortAdminAuth = Nothing, messageExpiration = Just defaultMessageExpiration, + expireMessagesOnStart = True, notificationExpiration = defaultNtfExpiration, inactiveClientExpiration = Just defaultInactiveClientExpiration, logStatsInterval = Nothing,