smp server: CLI option to compact store log with PostgreSQL database (#1484)

* smp server: CLI option to compact store log with PostgreSQL database

* version

* fix test

Author: epoberezkin

simplexmq.cabal

@@ -1,7 +1,7 @@
 cabal-version: 1.12
 
 name:           simplexmq
-version:        6.3.0.8
+version:        6.3.0.805
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and

src/Simplex/Messaging/Server/Env/STM.hs

@@ -135,6 +135,7 @@ data ServerConfig = ServerConfig
 
 data StartOptions = StartOptions
   { maintenance :: Bool,
+    compactLog :: Bool,
     skipWarnings :: Bool,
     confirmMigrations :: MigrationConfirmation
   }
@@ -331,7 +332,7 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smp
     ASSCfg qt mt (SSCMemory storePaths_) -> do
       let storePath = storeMsgsFile =<< storePaths_
       ms <- newMsgStore STMStoreConfig {storePath, quota = msgQueueQuota}
-      forM_ storePaths_ $ \StorePaths {storeLogFile = f} ->  loadStoreLog (mkQueue ms) f $ queueStore ms
+      forM_ storePaths_ $ \StorePaths {storeLogFile = f} -> loadStoreLog (mkQueue ms) f $ queueStore ms
       pure $ AMS qt mt ms
     ASSCfg qt mt SSCMemoryJournal {storeLogFile, storeMsgsPath} -> do
       let qsCfg = MQStoreCfg
@@ -341,9 +342,10 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smp
       pure $ AMS qt mt ms
 #if defined(dbServerPostgres)
     ASSCfg qt mt SSCDatabaseJournal {storeCfg, storeMsgsPath'} -> do
-      let StartOptions {confirmMigrations} = startOptions config
+      let StartOptions {compactLog, confirmMigrations} = startOptions config
           qsCfg = PQStoreCfg (storeCfg {confirmMigrations} :: PostgresStoreCfg)
           cfg = mkJournalStoreConfig qsCfg storeMsgsPath' msgQueueQuota maxJournalMsgCount maxJournalStateLines idleQueueInterval
+      when compactLog $ compactDbStoreLog $ dbStoreLogPath storeCfg
       ms <- newMsgStore cfg
       pure $ AMS qt mt ms
 #else
@@ -368,6 +370,16 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smp
       logInfo $ "restoring queues from file " <> T.pack f
       sl <- readWriteQueueStore False mkQ f st
       setStoreLog st sl
+    compactDbStoreLog = \case
+      Just f -> do
+        logInfo $ "compacting queues in file " <> T.pack f
+        st <- newMsgStore STMStoreConfig {storePath = Nothing, quota = msgQueueQuota}
+        sl <- readWriteQueueStore False (mkQueue st) f (queueStore st)
+        setStoreLog (queueStore st) sl
+        closeMsgStore st
+      Nothing -> do
+        logError "Error: `--compact-log` used without `db_store_log` INI option"
+        exitFailure
     getCredentials protocol creds = do
       files <- missingCreds
       unless (null files) $ do

src/Simplex/Messaging/Server/Main.hs

@@ -795,8 +795,14 @@ cliCommandP cfgPath logPath iniFile =
       maintenance <-
         switch
           ( long "maintenance"
+              <> short 'm'
               <> help "Do not start the server, only perform start and stop tasks"
           )
+      compactLog <-
+        switch
+          ( long "compact-log"
+              <> help "Compact store log (always enabled with `memory` storage for queues)"
+          )
       skipWarnings <-
         switch
           ( long "skip-warnings"
@@ -810,7 +816,7 @@ cliCommandP cfgPath logPath iniFile =
               <> help "Confirm PostgreSQL database migration: up, down (default is manual confirmation)"
               <> value MCConsole
           )
-      pure StartOptions {maintenance, skipWarnings, confirmMigrations}
+      pure StartOptions {maintenance, compactLog, skipWarnings, confirmMigrations}
     journalCmdP = storeCmdP "message log file" "journal storage"
     databaseCmdP = storeCmdP "queue store log file" "PostgreSQL database schema"
     storeCmdP src dest =

src/Simplex/Messaging/Server/QueueStore/STM.hs

@@ -63,7 +63,11 @@ instance StoreQueueClass q => QueueStoreClass q (STMQueueStore q) where
     pure STMQueueStore {queues, senders, notifiers, storeLog}
 
   closeQueueStore :: STMQueueStore q -> IO ()
-  closeQueueStore st = readTVarIO (storeLog st) >>= mapM_ closeStoreLog
+  closeQueueStore STMQueueStore {queues, senders, notifiers, storeLog} = do
+    readTVarIO storeLog >>= mapM_ closeStoreLog
+    atomically $ TM.clear queues
+    atomically $ TM.clear senders
+    atomically $ TM.clear notifiers
 
   loadedQueues = queues
   {-# INLINE loadedQueues #-}

tests/CoreTests/MsgStoreTests.hs

@@ -472,7 +472,7 @@ testReadFileMissing ms = do
 
   mq <- fromJust <$> readTVarIO (msgQueue q)
   MsgQueueState {readState = rs} <- readTVarIO $ state mq
-  closeMsgStore ms
+  closeMsgQueue q
   let path = journalFilePath (queueDirectory $ queue mq) $ journalId rs
   removeFile path
 
@@ -491,7 +491,7 @@ testReadFileMissingSwitch ms = do
 
   mq <- fromJust <$> readTVarIO (msgQueue q)
   MsgQueueState {readState = rs} <- readTVarIO $ state mq
-  closeMsgStore ms
+  closeMsgQueue q
   let path = journalFilePath (queueDirectory $ queue mq) $ journalId rs
   removeFile path
 
@@ -509,7 +509,7 @@ testWriteFileMissing ms = do
 
   mq <- fromJust <$> readTVarIO (msgQueue q)
   MsgQueueState {writeState = ws} <- readTVarIO $ state mq
-  closeMsgStore ms
+  closeMsgQueue q
   let path = journalFilePath (queueDirectory $ queue mq) $ journalId ws
   print path
   removeFile path
@@ -532,7 +532,7 @@ testReadAndWriteFilesMissing ms = do
 
   mq <- fromJust <$> readTVarIO (msgQueue q)
   MsgQueueState {readState = rs, writeState = ws} <- readTVarIO $ state mq
-  closeMsgStore ms
+  closeMsgQueue q
   removeFile $ journalFilePath (queueDirectory $ queue mq) $ journalId rs
   removeFile $ journalFilePath (queueDirectory $ queue mq) $ journalId ws
 

tests/SMPClient.hs

@@ -212,7 +212,7 @@ cfgMS msType =
       allowSMPProxy = False,
       serverClientConcurrency = 2,
       information = Nothing,
-      startOptions = StartOptions {maintenance = False, skipWarnings = False, confirmMigrations = MCYesUp}
+      startOptions = StartOptions {maintenance = False, compactLog = False, skipWarnings = False, confirmMigrations = MCYesUp}
     }
 
 serverStoreConfig :: AStoreType -> AServerStoreCfg