smp server: option to skip expiring messages on start, read queue state file end only (#1400)

Author: epoberezkin

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

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

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

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}

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

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,