smp server: expire messages in idle message queues (including not opened)

Author: epoberezkin

src/Simplex/Messaging/Server.hs

@@ -296,7 +296,6 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
         deliverNtfs ns stats (AClient _ Client {clientId, ntfSubscriptions, sndQ, connected}) =
           whenM (currentClient readTVarIO) $ do
             subs <- readTVarIO ntfSubscriptions
-            logDebug $ "NOTIFICATIONS: client #" <> tshow clientId <> " is current with " <> tshow (M.size subs) <> " subs"
             ntfQs <- M.assocs . M.filterWithKey (\nId _ -> M.member nId subs) <$> readTVarIO ns
             tryAny (atomically $ flushSubscribedNtfs ntfQs) >>= \case
               Right len -> updateNtfStats len
@@ -321,12 +320,11 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
                   writeTVar v []
                   pure $ foldl' (\acc' ntf -> nmsg nId ntf : acc') acc ntfs -- reverses, to order by time
             nmsg nId MsgNtf {ntfNonce, ntfEncMeta} = (CorrId "", nId, NMSG ntfNonce ntfEncMeta)
-            updateNtfStats 0 = logDebug $ "NOTIFICATIONS: no ntfs for client #" <> tshow clientId
+            updateNtfStats 0 = pure ()
             updateNtfStats len = liftIO $ do
               atomicModifyIORef'_ (ntfCount stats) (subtract len)
               atomicModifyIORef'_ (msgNtfs stats) (+ len)
               atomicModifyIORef'_ (msgNtfsB stats) (+ (len `div` 80 + 1)) -- up to 80 NMSG in the batch
-              logDebug $ "NOTIFICATIONS: delivered to client #" <> tshow clientId <> " " <> tshow len <> " ntfs"
 
     sendPendingEvtsThread :: Server -> M ()
     sendPendingEvtsThread s = do
@@ -386,13 +384,13 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
       liftIO $ forever $ do
         threadDelay' interval
         old <- expireBeforeEpoch expCfg
-        Sum deleted <- withActiveMsgQueues ms $ expireQueueMsgs stats old
+        now <- systemSeconds <$> getSystemTime
+        Sum deleted <- withActiveMsgQueues ms $ expireQueueMsgs now ms old
+        atomicModifyIORef'_ (msgExpired stats) (+ deleted)
         logInfo $ "STORE: expireMessagesThread, expired " <> tshow deleted <> " messages"
       where
-        expireQueueMsgs stats old rId q =
-          runExceptT (deleteExpiredMsgs rId q True old) >>= \case
-            Right deleted -> Sum deleted <$ atomicModifyIORef'_ (msgExpired stats) (+ deleted)
-            Left _ -> pure 0
+        expireQueueMsgs now ms old rId q =
+          either (const 0) Sum <$> runExceptT (idleDeleteExpiredMsgs now ms rId q old)
 
     expireNtfsThread :: ServerConfig -> M ()
     expireNtfsThread ServerConfig {notificationExpiration = expCfg} = do
@@ -1469,7 +1467,7 @@ client
 
             expireMessages :: Maybe ExpirationConfig -> ServerStats -> ExceptT ErrorType IO ()
             expireMessages msgExp stats = do
-              deleted <- maybe (pure 0) (deleteExpiredMsgs (recipientId qr) q True <=< liftIO . expireBeforeEpoch) msgExp
+              deleted <- maybe (pure 0) (deleteExpiredMsgs ms (recipientId qr) q <=< liftIO . expireBeforeEpoch) msgExp
               liftIO $ when (deleted > 0) $ atomicModifyIORef'_ (msgExpired stats) (+ deleted)
 
             -- The condition for delivery of the message is:
@@ -1763,9 +1761,8 @@ processServerMessages = do
                   exitFailure
               where
                 expireQueue = do
-                  expired'' <- deleteExpiredMsgs rId q False old
+                  expired'' <- deleteExpiredMsgs ms rId q old
                   stored'' <- getQueueSize ms rId q
-                  liftIO $ logQueueState q
                   liftIO $ closeMsgQueue q
                   pure (stored'', expired'')
             processValidateQueue :: RecipientId -> JournalQueue -> IO MessageStats

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

@@ -121,7 +121,7 @@ defaultMessageExpiration :: ExpirationConfig
 defaultMessageExpiration =
   ExpirationConfig
     { ttl = defMsgExpirationDays * 86400, -- seconds
-      checkInterval = 43200 -- seconds, 12 hours
+      checkInterval = 21600 -- seconds, 6 hours
     }
 
 defNtfExpirationHours :: Int64
@@ -290,7 +290,8 @@ 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, stateTailSize = defaultStateTailSize}
+        let idleInterval = checkInterval defaultMessageExpiration
+            cfg = JournalStoreConfig {storePath, quota = msgQueueQuota, pathParts = journalMsgStoreDepth, maxMsgCount = maxJournalMsgCount, maxStateLines = maxJournalStateLines, stateTailSize = defaultStateTailSize, idleInterval}
          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

@@ -148,7 +148,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, stateTailSize = defaultStateTailSize}
+    newJournalMsgStore = newMsgStore JournalStoreConfig {storePath = storeMsgsJournalDir, pathParts = journalMsgStoreDepth, quota = defaultMsgQueueQuota, maxMsgCount = defaultMaxJournalMsgCount, maxStateLines = defaultMaxJournalStateLines, stateTailSize = defaultStateTailSize, idleInterval = checkInterval defaultMessageExpiration}
     iniFile = combine cfgPath "smp-server.ini"
     serverVersion = "SMP server v" <> simplexMQVersion
     defaultServerPorts = "5223,443"

src/Simplex/Messaging/Server/MsgStore/Journal.hs

@@ -53,6 +53,7 @@ import Data.List (intercalate)
 import Data.Maybe (catMaybes, fromMaybe)
 import qualified Data.Text as T
 import Data.Time.Clock (getCurrentTime)
+import Data.Time.Clock.System (SystemTime (..), getSystemTime)
 import Data.Time.Format.ISO8601 (iso8601Show)
 import GHC.IO (catchAny)
 import Simplex.Messaging.Agent.Client (getMapLock, withLockMap)
@@ -92,7 +93,9 @@ data JournalStoreConfig = JournalStoreConfig
     -- This number should be set bigger than queue quota.
     maxMsgCount :: Int,
     maxStateLines :: Int,
-    stateTailSize :: Int
+    stateTailSize :: Int,
+    -- time in seconds after which the queue will be closed after message expiration
+    idleInterval :: Int64
   }
 
 data JournalQueue = JournalQueue
@@ -116,7 +119,9 @@ data JournalMsgQueue = JournalMsgQueue
     -- It  prevents reading each message twice,
     -- and reading it after it was just written.
     tipMsg :: TVar (Maybe (Maybe (Message, Int64))),
-    handles :: TVar (Maybe MsgQueueHandles)
+    handles :: TVar (Maybe MsgQueueHandles),
+    -- system time in seconds since epoch
+    activeAt :: TVar Int64
   }
 
 data MsgQueueState = MsgQueueState
@@ -295,11 +300,11 @@ instance MsgStoreClass JournalMsgStore where
               (Nothing <$ putStrLn ("Error: path " <> path' <> " is not a directory, skipping"))
 
   logQueueStates :: JournalMsgStore -> IO ()
-  logQueueStates ms = withActiveMsgQueues ms $ \_ -> logQueueState
+  logQueueStates ms = withActiveMsgQueues ms $ \_ -> unStoreIO . logQueueState
 
-  logQueueState :: JournalQueue -> IO ()
+  logQueueState :: JournalQueue -> StoreIO ()
   logQueueState q =
-    void $
+    StoreIO . void $
       readTVarIO (msgQueue_ q)
         $>>= \mq -> readTVarIO (handles mq)
         $>>= (\hs -> (readTVarIO (state mq) >>= appendState (stateHandle hs)) $> Just ())
@@ -326,9 +331,21 @@ instance MsgStoreClass JournalMsgStore where
             journalId <- newJournalId random
             mkJournalQueue queue (newMsgQueueState journalId) Nothing
 
-  openedMsgQueue :: JournalQueue -> StoreIO (Maybe JournalMsgQueue)
-  openedMsgQueue = StoreIO . readTVarIO . msgQueue_
-  {-# INLINE openedMsgQueue #-}
+  withIdleMsgQueue :: Int64 -> JournalMsgStore -> RecipientId -> JournalQueue -> (JournalMsgQueue -> StoreIO a) -> StoreIO (Maybe a)
+  withIdleMsgQueue now ms@JournalMsgStore {config} rId q action =
+    StoreIO $ readTVarIO (msgQueue_ q) >>= maybe runQ idleQ
+    where
+      runQ =
+        Just <$>
+          E.bracket
+            (unStoreIO $ getMsgQueue ms rId q)
+            (\_ -> closeMsgQueue q)
+            (unStoreIO . action)
+      idleQ mq = do
+        ts <- readTVarIO $ activeAt mq
+        if now - ts > idleInterval config
+          then Just <$> unStoreIO (action mq) `E.finally` closeMsgQueue q
+          else pure Nothing
 
   deleteQueue :: JournalMsgStore -> RecipientId -> JournalQueue -> IO (Either ErrorType QueueRec)
   deleteQueue ms rId q =
@@ -355,7 +372,7 @@ instance MsgStoreClass JournalMsgStore where
   writeMsg :: JournalMsgStore -> RecipientId -> JournalQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
   writeMsg ms rId q' logState msg = isolateQueue rId q' "writeMsg" $ do
     q <- getMsgQueue ms rId q'
-    StoreIO $ do
+    StoreIO $ (`E.finally` updateActiveAt q) $ do
       st@MsgQueueState {canWrite, size} <- readTVarIO (state q)
       let empty = size == 0
       if canWrite || empty
@@ -420,7 +437,7 @@ instance MsgStoreClass JournalMsgStore where
             pure $ Just msg
 
   tryDeleteMsg_ :: JournalMsgQueue -> Bool -> StoreIO ()
-  tryDeleteMsg_ q@JournalMsgQueue {tipMsg, handles} logState = StoreIO $
+  tryDeleteMsg_ q@JournalMsgQueue {tipMsg, handles} logState = StoreIO $ (`E.finally` updateActiveAt q) $
     void $
       readTVarIO tipMsg -- if there is no cached tipMsg, do nothing
         $>>= (pure . fmap snd)
@@ -431,6 +448,9 @@ instance MsgStoreClass JournalMsgStore where
   isolateQueue rId JournalQueue {queueLock} op =
     tryStore' op rId . withLock' queueLock op . unStoreIO
 
+updateActiveAt :: JournalMsgQueue -> IO ()
+updateActiveAt q = atomically . writeTVar (activeAt q) . systemSeconds =<< getSystemTime
+
 tryStore' :: String -> RecipientId -> IO a -> ExceptT ErrorType IO a
 tryStore' op rId = tryStore op rId . fmap Right
 
@@ -457,9 +477,10 @@ mkJournalQueue queue st hs_ = do
   state <- newTVarIO st
   tipMsg <- newTVarIO Nothing
   handles <- newTVarIO hs_
+  activeAt <- newTVarIO . systemSeconds =<< getSystemTime
   -- using the same queue lock which is currently locked,
   -- to avoid map lookup on queue operations
-  pure JournalMsgQueue {queue, state, tipMsg, handles}
+  pure JournalMsgQueue {queue, state, tipMsg, handles, activeAt}
 
 chooseReadJournal :: JournalMsgQueue -> Bool -> MsgQueueHandles -> IO (Maybe (JournalState 'JTRead, Handle))
 chooseReadJournal q log' hs = do

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

@@ -21,6 +21,7 @@ import Control.Concurrent.STM
 import Control.Monad.IO.Class
 import Control.Monad.Trans.Except
 import Data.Functor (($>))
+import Data.Int (Int64)
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.MsgStore.Types
 import Simplex.Messaging.Server.QueueStore
@@ -108,9 +109,10 @@ instance MsgStoreClass STMMsgStore where
         writeTVar msgQueue_ $! Just q
         pure q
 
-  openedMsgQueue :: STMQueue -> STM (Maybe STMMsgQueue)
-  openedMsgQueue = readTVar . msgQueue_
-  {-# INLINE openedMsgQueue #-}
+  -- does not create queue if it does not exist, does not delete it if it does (can't just close in-memory queue)
+  withIdleMsgQueue :: Int64 -> STMMsgStore -> RecipientId -> STMQueue -> (STMMsgQueue -> STM a) -> STM (Maybe a)
+  withIdleMsgQueue _ _ _ STMQueue {msgQueue_} action = readTVar msgQueue_ >>= mapM action
+  {-# INLINE withIdleMsgQueue #-}
 
   deleteQueue :: STMMsgStore -> RecipientId -> STMQueue -> IO (Either ErrorType QueueRec)
   deleteQueue ms rId q = fst <$$> deleteQueue' ms rId q

src/Simplex/Messaging/Server/MsgStore/Types.hs

@@ -15,6 +15,7 @@ import Control.Monad (foldM)
 import Control.Monad.Trans.Except
 import Data.Int (Int64)
 import Data.Kind
+import Data.Maybe (fromMaybe)
 import qualified Data.Map.Strict as M
 import Data.Time.Clock.System (SystemTime (systemSeconds))
 import Simplex.Messaging.Protocol
@@ -42,10 +43,11 @@ class Monad (StoreMonad s) => MsgStoreClass s where
   activeMsgQueues :: s -> TMap RecipientId (StoreQueue s)
   withAllMsgQueues :: Monoid a => Bool -> s -> (RecipientId -> StoreQueue s -> IO a) -> IO a
   logQueueStates :: s -> IO ()
-  logQueueState :: StoreQueue s -> IO ()
+  logQueueState :: StoreQueue s -> StoreMonad s ()
   queueRec' :: StoreQueue s -> TVar (Maybe QueueRec)
   getMsgQueue :: s -> RecipientId -> StoreQueue s -> StoreMonad s (MsgQueue s)
-  openedMsgQueue :: StoreQueue s -> StoreMonad s (Maybe (MsgQueue s))
+  -- the journal queue will be closed after action if it was initially closed or idle longer than interval in config
+  withIdleMsgQueue :: Int64 -> s -> RecipientId -> StoreQueue s -> (MsgQueue s -> StoreMonad s a) -> StoreMonad s (Maybe a)
   deleteQueue :: s -> RecipientId -> StoreQueue s -> IO (Either ErrorType QueueRec)
   deleteQueueSize :: s -> RecipientId -> StoreQueue s -> IO (Either ErrorType (QueueRec, Int))
   getQueueMessages_ :: Bool -> MsgQueue s -> StoreMonad s [Message]
@@ -106,13 +108,26 @@ withMsgQueue :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> String ->
 withMsgQueue st rId q op a = isolateQueue rId q op $ getMsgQueue st rId q >>= a
 {-# INLINE withMsgQueue #-}
 
-deleteExpiredMsgs :: MsgStoreClass s => RecipientId -> StoreQueue s -> Bool -> Int64 -> ExceptT ErrorType IO Int
-deleteExpiredMsgs rId q logState old =
-  isolateQueue rId q "deleteExpiredMsgs" $ openedMsgQueue q >>= maybe (pure 0) (loop 0)
+deleteExpiredMsgs :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> Int64 -> ExceptT ErrorType IO Int
+deleteExpiredMsgs st rId q old =
+  isolateQueue rId q "deleteExpiredMsgs" $
+    getMsgQueue st rId q >>= deleteExpireMsgs_ old q
+
+-- closed and idle queues will be closed after expiration
+idleDeleteExpiredMsgs :: MsgStoreClass s => Int64 -> s -> RecipientId -> StoreQueue s -> Int64 -> ExceptT ErrorType IO Int
+idleDeleteExpiredMsgs now st rId q old =
+  isolateQueue rId q "idleDeleteExpiredMsgs" $ 
+    fromMaybe 0 <$> withIdleMsgQueue now st rId q (deleteExpireMsgs_ old q)
+
+deleteExpireMsgs_ :: MsgStoreClass s => Int64 -> StoreQueue s -> MsgQueue s -> StoreMonad s Int
+deleteExpireMsgs_ old q mq = do
+  n <- loop 0
+  logQueueState q
+  pure n
   where
-    loop dc mq =
+    loop dc =
       tryPeekMsg_ mq >>= \case
         Just Message {msgTs}
           | systemSeconds msgTs < old ->
-              tryDeleteMsg_ mq logState >> loop (dc + 1) mq
+              tryDeleteMsg_ mq False >> loop (dc + 1)
         _ -> pure dc

tests/CoreTests/MsgStoreTests.hs

@@ -77,7 +77,8 @@ testJournalStoreCfg =
       quota = 3,
       maxMsgCount = 4,
       maxStateLines = 2,
-      stateTailSize = 256
+      stateTailSize = 256,
+      idleInterval = 21600
     }
 
 mkMessage :: MonadIO m => ByteString -> m Message