smp-server: remove journal store

docs/smp-server-db-load.md

@@ -144,9 +144,7 @@ space, so it is workload-dependent; `fillfactor = 80` reached 100% for this frac
 larger share of queues at once would need a lower value. `fillfactor` applies to pages rewritten after
 the change, so the ratio ramps up as the heap turns over, or immediately after `pg_repack`.
 
-Dropping the index regresses the deprecated journal message store's expiration (`foldRecentQueueRecs`,
-`Journal.hs:434`) to a sequential scan; this is accepted because that store is being retired. The
-Postgres message store does not read `updated_at`.
+The Postgres message store does not read `updated_at`, so no query depends on the dropped index.
 
 Autovacuum reloptions applied by the migration:
 

simplexmq.cabal

@@ -284,8 +284,6 @@ library
           Simplex.Messaging.Server.Main.Init
           Simplex.Messaging.Server.Web
           Simplex.Messaging.Server.MsgStore
-          Simplex.Messaging.Server.MsgStore.Journal
-          Simplex.Messaging.Server.MsgStore.Journal.SharedLock
           Simplex.Messaging.Server.MsgStore.STM
           Simplex.Messaging.Server.MsgStore.Types
           Simplex.Messaging.Server.Names

src/Simplex/Messaging/Server.hs

@@ -108,7 +108,6 @@ import Simplex.Messaging.Server.Control
 import Simplex.Messaging.Server.Env.STM as Env
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.MsgStore
-import Simplex.Messaging.Server.MsgStore.Journal (JournalMsgStore, JournalQueue (..), getJournalQueueMessages)
 import Simplex.Messaging.Server.Names (NamesEnv, closeNamesEnv, resolveName)
 import Simplex.Messaging.Server.MsgStore.STM
 import Simplex.Messaging.Server.MsgStore.Types
@@ -2350,7 +2349,6 @@ saveServerMessages drainMsgs ms = case ms of
   StoreMemory STMMsgStore {storeConfig = STMStoreConfig {storePath}} -> case storePath of
     Just f -> exportMessages False ms f drainMsgs
     Nothing -> logNote "undelivered messages are not saved"
-  StoreJournal _ -> logNote "closed journal message storage"
 #if defined(dbServerPostgres)
   StoreDatabase _ -> logNote "closed postgres message storage"
 #endif
@@ -2360,7 +2358,6 @@ exportMessages tty st f drainMsgs = do
   logNote $ "saving messages to file " <> T.pack f
   run $ case st of
     StoreMemory ms -> exportMessages_ ms $ getMsgs ms
-    StoreJournal ms -> exportMessages_ ms $ getJournalMsgs ms
 #if defined(dbServerPostgres)
     StoreDatabase ms -> exportDbMessages tty ms
 #endif
@@ -2372,10 +2369,6 @@ exportMessages tty st f drainMsgs = do
       Left e -> do
         logError $ "error exporting messages: " <> tshow e
         exitFailure
-    getJournalMsgs ms q =
-      readTVarIO (msgQueue' q) >>= \case
-        Just _ -> getMsgs ms q
-        Nothing -> getJournalQueueMessages ms q
     getMsgs :: MsgStoreClass s' => s' -> StoreQueue s' -> IO [Message]
     getMsgs ms q = unsafeRunStore q "saveQueueMsgs" $ getQueueMessages_ drainMsgs q =<< getMsgQueue ms q False
     saveQueueMsgs :: (StoreQueue s -> IO [Message]) -> Handle -> StoreQueue s -> IO (Sum Int)
@@ -2392,37 +2385,16 @@ processServerMessages StartOptions {skipWarnings} = do
   asks msgStore_ >>= liftIO . processMessages old_ expire
     where
       processMessages :: Maybe Int64 -> Bool -> MsgStore s' -> IO (Maybe MessageStats)
+#if defined(dbServerPostgres)
       processMessages old_ expire = \case
+#else
+      processMessages old_ _ = \case
+#endif
         StoreMemory ms@STMMsgStore {storeConfig = STMStoreConfig {storePath}} -> case storePath of
           Just f -> ifM (doesFileExist f) (Just <$> importMessages False ms f old_ skipWarnings) (pure Nothing)
           Nothing -> pure Nothing
-        StoreJournal ms -> processJournalMessages old_ expire ms
 #if defined(dbServerPostgres)
         StoreDatabase ms -> processDbMessages old_ expire ms
-#endif
-      processJournalMessages :: forall s. Maybe Int64 -> Bool -> JournalMsgStore s -> IO (Maybe MessageStats)
-      processJournalMessages old_ expire ms
-        | expire = Just <$> case old_ of
-            Just old -> do
-              logNote "expiring journal store messages..."
-              run $ processExpireQueue old
-            Nothing -> do
-              logNote "validating journal store messages..."
-              run processValidateQueue
-        | otherwise = logWarn "skipping message expiration" $> Nothing
-        where
-          run a = unsafeWithAllMsgQueues False ms a `catchAny` \_ -> exitFailure
-          processExpireQueue :: Int64 -> JournalQueue s -> IO MessageStats
-          processExpireQueue old q = unsafeRunStore q "processExpireQueue" $ do
-            mq <- getMsgQueue ms q False
-            expiredMsgsCount <- deleteExpireMsgs_ old q mq
-            storedMsgsCount <- getQueueSize_ mq
-            pure MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues = 1}
-          processValidateQueue :: JournalQueue s -> IO MessageStats
-          processValidateQueue q = unsafeRunStore q "processValidateQueue" $ do
-            storedMsgsCount <- getQueueSize_ =<< getMsgQueue ms q False
-            pure newMessageStats {storedMsgsCount, storedQueues = 1}
-#if defined(dbServerPostgres)
       processDbMessages old_ expire ms
         | expire = Just <$> case old_ of
             Just old -> do

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

@@ -42,7 +42,6 @@ module Simplex.Messaging.Server.Env.STM
     VerifiedTransmission,
     ResponseAndMessage,
     newEnv,
-    mkJournalStoreConfig,
     msgStore,
     fromMsgStore,
     newClient,
@@ -70,10 +69,6 @@ module Simplex.Messaging.Server.Env.STM
     defaultNameResolverConcurrency,
     defaultMaxClientCmdWait,
     defaultAuthErrorDelayMs,
-    defaultMaxJournalMsgCount,
-    defaultMaxJournalStateLines,
-    defaultIdleQueueInterval,
-    journalMsgStoreDepth,
     readWriteQueueStore,
     noPostgresExitStr,
     noPostgresExit,
@@ -98,7 +93,7 @@ import Data.List.NonEmpty (NonEmpty)
 import Data.Map.Strict (Map)
 import Data.Maybe (isJust)
 import qualified Data.Text as T
-import Data.Time.Clock (getCurrentTime, nominalDay)
+import Data.Time.Clock (getCurrentTime)
 import Data.Time.Clock.System (SystemTime)
 import qualified Data.X509 as X
 import Data.X509.Validation (Fingerprint (..))
@@ -115,7 +110,6 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.Information
-import Simplex.Messaging.Server.MsgStore.Journal
 import Simplex.Messaging.Server.MsgStore.STM
 import Simplex.Messaging.Server.MsgStore.Types
 import Simplex.Messaging.Server.Names (NamesConfig (..), NamesEnv, newNamesEnv, pingEndpoint)
@@ -148,8 +142,6 @@ data ServerConfig s = ServerConfig
     smpHandshakeTimeout :: Int,
     tbqSize :: Natural,
     msgQueueQuota :: Int,
-    maxJournalMsgCount :: Int,
-    maxJournalStateLines :: Int,
     queueIdBytes :: Int,
     msgIdBytes :: Int,
     serverStoreCfg :: ServerStoreCfg s,
@@ -166,8 +158,6 @@ data ServerConfig s = ServerConfig
     messageExpiration :: Maybe ExpirationConfig,
     expireMessagesOnStart :: Bool,
     expireMessagesOnSend :: Bool,
-    -- | interval of inactivity after which journal queue is closed
-    idleQueueInterval :: Int64,
     -- | notification expiration interval (seconds)
     notificationExpiration :: ExpirationConfig,
     -- | time after which the socket with inactive client can be disconnected (without any messages or commands, incl. PING),
@@ -233,9 +223,6 @@ defaultMessageExpiration =
       checkInterval = 7200 -- seconds, 2 hours
     }
 
-defaultIdleQueueInterval :: Int64
-defaultIdleQueueInterval = 14400 -- seconds, 4 hours
-
 defNtfExpirationHours :: Int64
 defNtfExpirationHours = 24
 
@@ -265,21 +252,9 @@ defaultMaxClientCmdWait = 60000000 -- microseconds
 defaultAuthErrorDelayMs :: Int64
 defaultAuthErrorDelayMs = 50
 
-journalMsgStoreDepth :: Int
-journalMsgStoreDepth = 5
-
-defaultMaxJournalStateLines :: Int
-defaultMaxJournalStateLines = 16
-
-defaultMaxJournalMsgCount :: Int
-defaultMaxJournalMsgCount = 256
-
 defaultMsgQueueQuota :: Int
 defaultMsgQueueQuota = 128
 
-defaultStateTailSize :: Int
-defaultStateTailSize = 512
-
 data Env s = Env
   { config :: ServerConfig s,
     serverActive :: TVar Bool,
@@ -305,20 +280,17 @@ msgStore = fromMsgStore . msgStore_
 fromMsgStore :: MsgStore s -> s
 fromMsgStore = \case
   StoreMemory s -> s
-  StoreJournal s -> s
 #if defined(dbServerPostgres)
   StoreDatabase s -> s
 #endif
 {-# INLINE fromMsgStore #-}
 
 type family SupportedStore (qs :: QSType) (ms :: MSType) :: Constraint where
   SupportedStore 'QSMemory 'MSMemory = ()
-  SupportedStore 'QSMemory 'MSJournal = ()
   SupportedStore 'QSMemory 'MSPostgres =
     (Int ~ Bool, TypeError ('TE.Text "Storing messages in Postgres DB with queues in memory is not supported"))
   SupportedStore 'QSPostgres 'MSMemory =
     (Int ~ Bool, TypeError ('TE.Text "Storing messages in memory with queues in Postgres DB is not supported"))
-  SupportedStore 'QSPostgres 'MSJournal = ()
 #if defined(dbServerPostgres)
   SupportedStore 'QSPostgres 'MSPostgres = ()
 #else
@@ -332,26 +304,20 @@ data AStoreType =
 
 data ServerStoreCfg s where
   SSCMemory :: Maybe StorePaths -> ServerStoreCfg STMMsgStore
-  SSCMemoryJournal :: {storeLogFile :: FilePath, storeMsgsPath :: FilePath} -> ServerStoreCfg (JournalMsgStore 'QSMemory)
-  SSCDatabaseJournal :: {storeCfg :: PostgresStoreCfg, storeMsgsPath' :: FilePath} -> ServerStoreCfg (JournalMsgStore 'QSPostgres)
 #if defined(dbServerPostgres)
   SSCDatabase :: PostgresStoreCfg -> ServerStoreCfg PostgresMsgStore
 #endif
 
 dbStoreCfg :: ServerStoreCfg s -> Maybe PostgresStoreCfg
 dbStoreCfg = \case
   SSCMemory _ -> Nothing
-  SSCMemoryJournal {} -> Nothing
-  SSCDatabaseJournal {storeCfg} -> Just storeCfg
 #if defined(dbServerPostgres)
   SSCDatabase cfg -> Just cfg
 #endif
 
 storeLogFile' :: ServerStoreCfg s -> Maybe FilePath
 storeLogFile' = \case
   SSCMemory sp_ -> (\StorePaths {storeLogFile} -> storeLogFile) <$> sp_
-  SSCMemoryJournal {storeLogFile} -> Just storeLogFile
-  SSCDatabaseJournal {storeCfg = PostgresStoreCfg {dbStoreLogPath}} -> dbStoreLogPath
 #if defined(dbServerPostgres)
   SSCDatabase (PostgresStoreCfg {dbStoreLogPath}) -> dbStoreLogPath
 #endif
@@ -360,14 +326,12 @@ data StorePaths = StorePaths {storeLogFile :: FilePath, storeMsgsFile :: Maybe F
 
 type family MsgStoreType (qs :: QSType) (ms :: MSType) where
   MsgStoreType 'QSMemory 'MSMemory = STMMsgStore
-  MsgStoreType qs 'MSJournal = JournalMsgStore qs
 #if defined(dbServerPostgres)
   MsgStoreType 'QSPostgres 'MSPostgres = PostgresMsgStore
 #endif
 
 data MsgStore s where
   StoreMemory :: STMMsgStore -> MsgStore STMMsgStore
-  StoreJournal :: JournalMsgStore qs -> MsgStore (JournalMsgStore qs)
 #if defined(dbServerPostgres)
   StoreDatabase :: PostgresMsgStore -> MsgStore PostgresMsgStore
 #endif
@@ -581,7 +545,7 @@ newProhibitedSub = do
   return Sub {subThread = ProhibitSub, delivered}
 
 newEnv :: ServerConfig s -> IO (Env s)
-newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smpAgentCfg, information, messageExpiration, idleQueueInterval, msgQueueQuota, maxJournalMsgCount, maxJournalStateLines, namesConfig, msgIdBytes} = do
+newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smpAgentCfg, information, messageExpiration, msgQueueQuota, namesConfig, msgIdBytes} = do
   when (msgIdBytes < 24) $ do
     logError "Error: msgIdBytes must be at least 24"
     exitFailure
@@ -593,30 +557,12 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smp
       ms <- newMsgStore STMStoreConfig {storePath, quota = msgQueueQuota}
       forM_ storePaths_ $ \StorePaths {storeLogFile = f} -> loadStoreLog (mkQueue ms True) f $ queueStore ms
       pure $ StoreMemory ms
-    SSCMemoryJournal {storeLogFile, storeMsgsPath} -> do
-      logWarn $
-        "Journal message store is deprecated and will be removed soon.\n"
-          <> "Please migrate to in-memory storage using `journal export` command.\n"
-          <> "After that you can migrate to PostgreSQL using `database import` command."
-      let qsCfg = MQStoreCfg
-          cfg = mkJournalStoreConfig qsCfg storeMsgsPath msgQueueQuota maxJournalMsgCount maxJournalStateLines idleQueueInterval
-      ms <- newMsgStore cfg
-      loadStoreLog (mkQueue ms True) storeLogFile $ stmQueueStore ms
-      pure $ StoreJournal ms
 #if defined(dbServerPostgres)
-    SSCDatabaseJournal {storeCfg, storeMsgsPath'} -> do
-      let StartOptions {compactLog, confirmMigrations} = startOptions config
-          qsCfg = PQStoreCfg (storeCfg {confirmMigrations} :: PostgresStoreCfg)
-          cfg = mkJournalStoreConfig qsCfg storeMsgsPath' msgQueueQuota maxJournalMsgCount maxJournalStateLines idleQueueInterval
-      when compactLog $ compactDbStoreLog $ dbStoreLogPath storeCfg
-      StoreJournal <$> newMsgStore cfg
     SSCDatabase storeCfg -> do
       let StartOptions {compactLog, confirmMigrations} = startOptions config
           cfg = PostgresMsgStoreCfg storeCfg {confirmMigrations} msgQueueQuota
       when compactLog $ compactDbStoreLog $ dbStoreLogPath storeCfg
       StoreDatabase <$> newMsgStore cfg
-#else
-    SSCDatabaseJournal {} -> noPostgresExit
 #endif
   ntfStore <- NtfStore <$> TM.emptyIO
   random <- C.newRandom
@@ -716,7 +662,9 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smp
             Nothing -> SPMMemoryOnly
             Just StorePaths {storeMsgsFile = Just _} -> SPMMessages
             _ -> SPMQueues
+#if defined(dbServerPostgres)
           _ -> SPMMessages
+#endif
 
 noPostgresExit :: IO a
 noPostgresExit = putStrLn noPostgresExitStr >> exitFailure
@@ -726,21 +674,6 @@ noPostgresExitStr =
   "Error: server binary is compiled without support for PostgreSQL database.\n"
     <> "Please download `smp-server-postgres` or re-compile with `cabal build -fserver_postgres`."
 
-mkJournalStoreConfig :: QStoreCfg s -> FilePath -> Int -> Int -> Int -> Int64 -> JournalStoreConfig s
-mkJournalStoreConfig queueStoreCfg storePath msgQueueQuota maxJournalMsgCount maxJournalStateLines idleQueueInterval =
-  JournalStoreConfig
-    { storePath,
-      quota = msgQueueQuota,
-      pathParts = journalMsgStoreDepth,
-      queueStoreCfg,
-      maxMsgCount = maxJournalMsgCount,
-      maxStateLines = maxJournalStateLines,
-      stateTailSize = defaultStateTailSize,
-      idleInterval = idleQueueInterval,
-      expireBackupsAfter = 14 * nominalDay,
-      keepMinBackups = 2
-    }
-
 newSMPProxyAgent :: SMPClientAgentConfig -> TVar ChaChaDRG -> IO ProxyAgent
 newSMPProxyAgent smpAgentCfg random = do
   smpAgent <- newSMPClientAgent SSender smpAgentCfg Nothing random