smp server: remove empty queues journals when expiring messages of idle queues (#1458)

* smp server: remove empty queues journals when expiring messages of idle queues

* remove unnecessary update

* ci: update action

* rename

Author: epoberezkin

.github/workflows/build.yml

@@ -37,7 +37,7 @@ jobs:
           cabal-version: "3.10.1.0"
 
       - name: Cache dependencies
-        uses: actions/cache@v2
+        uses: actions/cache@v3
         with:
           path: |
             ~/.cabal/store

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

@@ -108,8 +108,12 @@ data JournalQueue = JournalQueue
     msgQueue_ :: TVar (Maybe JournalMsgQueue),
     -- system time in seconds since epoch
     activeAt :: TVar Int64,
-    -- Just True - empty, Just False - non-empty, Nothing - unknown
-    isEmpty :: TVar (Maybe Bool)
+    queueState :: TVar (Maybe QState) -- Nothing - unknown
+  }
+
+data QState = QState
+  { hasPending :: Bool,
+    hasStored :: Bool
   }
 
 data JMQueue = JMQueue
@@ -152,6 +156,12 @@ data JournalState t = JournalState
   }
   deriving (Show)
 
+qState :: MsgQueueState -> QState
+qState MsgQueueState {size, readState = rs, writeState = ws} =
+  let hasPending = size > 0
+   in QState {hasPending, hasStored = hasPending || msgCount rs > 0 || msgCount ws > 0}
+{-# INLINE qState #-}
+
 data JournalType = JTRead | JTWrite
 
 data SJournalType (t :: JournalType) where
@@ -224,12 +234,20 @@ newtype StoreIO a = StoreIO {unStoreIO :: IO a}
 instance STMStoreClass JournalMsgStore where
   stmQueueStore JournalMsgStore {queueStore} = queueStore
   mkQueue st rId qr = do
-    lock <- getMapLock (queueLocks st) rId
-    q <- newTVar $ Just qr
-    mq <- newTVar Nothing
+    queueLock <- getMapLock (queueLocks st) rId
+    queueRec <- newTVar $ Just qr
+    msgQueue_ <- newTVar Nothing
     activeAt <- newTVar 0
-    isEmpty <- newTVar Nothing
-    pure $ JournalQueue rId lock q mq activeAt isEmpty
+    queueState <- newTVar Nothing
+    pure $
+      JournalQueue
+        { recipientId = rId,
+          queueLock,
+          queueRec,
+          msgQueue_,
+          activeAt,
+          queueState
+        }
   msgQueue_' = msgQueue_
 
 instance MsgStoreClass JournalMsgStore where
@@ -314,7 +332,7 @@ instance MsgStoreClass JournalMsgStore where
   {-# INLINE queueRec' #-}
 
   getMsgQueue :: JournalMsgStore -> JournalQueue -> Bool -> StoreIO JournalMsgQueue
-  getMsgQueue ms@JournalMsgStore {random} JournalQueue {recipientId = rId, msgQueue_} forWrite =
+  getMsgQueue ms@JournalMsgStore {random} q'@JournalQueue {recipientId = rId, msgQueue_} forWrite =
     StoreIO $ readTVarIO msgQueue_ >>= maybe newQ pure
     where
       newQ = do
@@ -323,6 +341,8 @@ instance MsgStoreClass JournalMsgStore where
             queue = JMQueue {queueDirectory = dir, statePath}
         q <- ifM (doesDirectoryExist dir) (openMsgQueue ms queue forWrite) (createQ queue)
         atomically $ writeTVar msgQueue_ $ Just q
+        st <- readTVarIO $ state q
+        atomically $ writeTVar (queueState q') $ Just $! qState st
         pure q
         where
           createQ :: JMQueue -> IO JournalMsgQueue
@@ -333,10 +353,9 @@ instance MsgStoreClass JournalMsgStore where
             mkJournalQueue queue (newMsgQueueState journalId) Nothing
 
   getPeekMsgQueue :: JournalMsgStore -> JournalQueue -> StoreIO (Maybe (JournalMsgQueue, Message))
-  getPeekMsgQueue ms q@JournalQueue {isEmpty} =
-    StoreIO (readTVarIO isEmpty) >>= \case
-      Just True -> pure Nothing
-      Just False -> peek
+  getPeekMsgQueue ms q@JournalQueue {queueState} =
+    StoreIO (readTVarIO queueState) >>= \case
+      Just QState {hasPending} -> if hasPending then peek else pure Nothing
       Nothing -> do
         -- We only close the queue if we just learnt it's empty.
         -- This is needed to reduce file descriptors and memory usage
@@ -353,15 +372,15 @@ instance MsgStoreClass JournalMsgStore where
 
   -- only runs action if queue is not empty
   withIdleMsgQueue :: Int64 -> JournalMsgStore -> JournalQueue -> (JournalMsgQueue -> StoreIO a) -> StoreIO (Maybe a, Int)
-  withIdleMsgQueue now ms@JournalMsgStore {config} q action =
+  withIdleMsgQueue now ms@JournalMsgStore {config} q@JournalQueue {queueState} action =
     StoreIO $ readTVarIO (msgQueue_ q) >>= \case
       Nothing ->
         E.bracket
-          (unStoreIO $ getPeekMsgQueue ms q)
+          getNonEmptyMsgQueue
           (mapM_ $ \_ -> closeMsgQueue q)
           (maybe (pure (Nothing, 0)) (unStoreIO . run))
         where
-          run (mq, _) = do
+          run mq = do
             r <- action mq
             sz <- getQueueSize_ mq
             pure (Just r, sz)
@@ -372,6 +391,19 @@ instance MsgStoreClass JournalMsgStore where
           else pure Nothing
         sz <- unStoreIO $ getQueueSize_ mq
         pure (r, sz)
+    where
+      getNonEmptyMsgQueue :: IO (Maybe JournalMsgQueue)
+      getNonEmptyMsgQueue =
+        readTVarIO queueState >>= \case
+          Just QState {hasStored}
+            | hasStored -> Just <$> unStoreIO (getMsgQueue ms q False)
+            | otherwise -> pure Nothing
+          Nothing -> do
+            mq <- unStoreIO $ getMsgQueue ms q False
+            -- queueState was updated in getMsgQueue
+            readTVarIO queueState >>= \case
+              Just QState {hasStored} | not hasStored -> closeMsgQueue q $> Nothing
+              _ -> pure $ Just mq
 
   deleteQueue :: JournalMsgStore -> JournalQueue -> IO (Either ErrorType QueueRec)
   deleteQueue ms q = fst <$$> deleteQueue_ ms q
@@ -383,15 +415,15 @@ instance MsgStoreClass JournalMsgStore where
     where
       getSize = maybe (pure (-1)) (fmap size . readTVarIO . state)
 
-  getQueueMessages_ :: Bool -> JournalMsgQueue -> StoreIO [Message]
-  getQueueMessages_ drainMsgs q = StoreIO (run [])
+  getQueueMessages_ :: Bool -> JournalQueue -> JournalMsgQueue -> StoreIO [Message]
+  getQueueMessages_ drainMsgs q' q = StoreIO (run [])
     where
       run msgs = readTVarIO (handles q) >>= maybe (pure []) (getMsg msgs)
-      getMsg msgs hs = chooseReadJournal q drainMsgs hs >>= maybe (pure msgs) readMsg
+      getMsg msgs hs = chooseReadJournal q' q drainMsgs hs >>= maybe (pure msgs) readMsg
         where
           readMsg (rs, h) = do
             (msg, len) <- hGetMsgAt h $ bytePos rs
-            updateReadPos q drainMsgs len hs
+            updateReadPos q' q drainMsgs len hs
             (msg :) <$> run msgs
 
   writeMsg :: JournalMsgStore -> JournalQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
@@ -402,7 +434,6 @@ instance MsgStoreClass JournalMsgStore where
       let empty = size == 0
       if canWrite || empty
         then do
-          atomically $ writeTVar (isEmpty q') (Just False)
           let canWrt' = quota > size
           if canWrt'
             then writeToJournal q st canWrt' msg $> Just (msg, empty)
@@ -424,7 +455,7 @@ instance MsgStoreClass JournalMsgStore where
             rs' = if journalId ws == journalId rs then rs {msgCount = msgPos', byteCount = bytePos'} else rs
             !st' = st {writeState = ws', readState = rs', canWrite = canWrt', size = size + 1}
         hAppend wh (bytePos ws) msgStr
-        updateQueueState q logState hs st' $
+        updateQueueState q' q logState hs st' $
           when (size == 0) $ writeTVar (tipMsg q) $ Just (Just (msg, msgLen))
         where
           JournalMsgQueue {queue = JMQueue {queueDirectory, statePath}, handles} = q
@@ -452,25 +483,22 @@ instance MsgStoreClass JournalMsgStore where
 
   tryPeekMsg_ :: JournalQueue -> JournalMsgQueue -> StoreIO (Maybe Message)
   tryPeekMsg_ q mq@JournalMsgQueue {tipMsg, handles} =
-    StoreIO $ (readTVarIO handles $>>= chooseReadJournal mq True $>>= peekMsg) >>= setEmpty
+    StoreIO $ (readTVarIO handles $>>= chooseReadJournal q mq True $>>= peekMsg)
     where
       peekMsg (rs, h) = readTVarIO tipMsg >>= maybe readMsg (pure . fmap fst)
         where
           readMsg = do
             ml@(msg, _) <- hGetMsgAt h $ bytePos rs
             atomically $ writeTVar tipMsg $ Just (Just ml)
             pure $ Just msg
-      setEmpty msg = do
-        atomically $ writeTVar (isEmpty q) (Just $ isNothing msg)
-        pure msg
 
   tryDeleteMsg_ :: JournalQueue -> JournalMsgQueue -> Bool -> StoreIO ()
   tryDeleteMsg_ q mq@JournalMsgQueue {tipMsg, handles} logState = StoreIO $ (`E.finally` when logState (updateActiveAt q)) $
     void $
       readTVarIO tipMsg -- if there is no cached tipMsg, do nothing
         $>>= (pure . fmap snd)
         $>>= \len -> readTVarIO handles
-        $>>= \hs -> updateReadPos mq logState len hs $> Just ()
+        $>>= \hs -> updateReadPos q mq logState len hs $> Just ()
 
   isolateQueue :: JournalQueue -> String -> StoreIO a -> ExceptT ErrorType IO a
   isolateQueue JournalQueue {recipientId, queueLock} op =
@@ -562,8 +590,8 @@ mkJournalQueue queue st hs_ = do
   -- to avoid map lookup on queue operations
   pure JournalMsgQueue {queue, state, tipMsg, handles}
 
-chooseReadJournal :: JournalMsgQueue -> Bool -> MsgQueueHandles -> IO (Maybe (JournalState 'JTRead, Handle))
-chooseReadJournal q log' hs = do
+chooseReadJournal :: JournalQueue -> JournalMsgQueue -> Bool -> MsgQueueHandles -> IO (Maybe (JournalState 'JTRead, Handle))
+chooseReadJournal q' q log' hs = do
   st@MsgQueueState {writeState = ws, readState = rs} <- readTVarIO (state q)
   case writeHandle hs of
     Just wh | msgPos rs >= msgCount rs && journalId rs /= journalId ws -> do
@@ -573,15 +601,16 @@ chooseReadJournal q log' hs = do
       when log' $ removeJournal (queueDirectory $ queue q) rs
       let !rs' = (newJournalState $ journalId ws) {msgCount = msgCount ws, byteCount = byteCount ws}
           !st' = st {readState = rs'}
-      updateQueueState q log' hs st' $ pure ()
+      updateQueueState q' q log' hs st' $ pure ()
       pure $ Just (rs', wh)
     _ | msgPos rs >= msgCount rs && journalId rs == journalId ws -> pure Nothing
     _ -> pure $ Just (rs, readHandle hs)
 
-updateQueueState :: JournalMsgQueue -> Bool -> MsgQueueHandles -> MsgQueueState -> STM () -> IO ()
-updateQueueState q log' hs st a = do
+updateQueueState :: JournalQueue -> JournalMsgQueue -> Bool -> MsgQueueHandles -> MsgQueueState -> STM () -> IO ()
+updateQueueState q' q log' hs st a = do
   unless (validQueueState st) $ E.throwIO $ userError $ "updateQueueState invalid state: " <> show st
   when log' $ appendState (stateHandle hs) st
+  atomically $ writeTVar (queueState q') $ Just $! qState st
   atomically $ writeTVar (state q) st >> a
 
 appendState :: Handle -> MsgQueueState -> IO ()
@@ -591,14 +620,14 @@ appendState h = E.uninterruptibleMask_ . appendState_ h
 appendState_ :: Handle -> MsgQueueState -> IO ()
 appendState_ h st = B.hPutStr h $ strEncode st `B.snoc` '\n'
 
-updateReadPos :: JournalMsgQueue -> Bool -> Int64 -> MsgQueueHandles -> IO ()
-updateReadPos q log' len hs = do
+updateReadPos :: JournalQueue -> JournalMsgQueue -> Bool -> Int64 -> MsgQueueHandles -> IO ()
+updateReadPos q' q log' len hs = do
   st@MsgQueueState {readState = rs, size} <- readTVarIO (state q)
   let JournalState {msgPos, bytePos} = rs
   let msgPos' = msgPos + 1
       rs' = rs {msgPos = msgPos', bytePos = bytePos + len}
       st' = st {readState = rs', size = size - 1}
-  updateQueueState q log' hs st' $ writeTVar (tipMsg q) Nothing
+  updateQueueState q' q log' hs st' $ writeTVar (tipMsg q) Nothing
 
 msgQueueDirectory :: JournalMsgStore -> RecipientId -> FilePath
 msgQueueDirectory JournalMsgStore {config = JournalStoreConfig {storePath, pathParts}} rId =

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

@@ -121,8 +121,8 @@ instance MsgStoreClass STMMsgStore where
     where
       getSize = maybe (pure 0) (\STMMsgQueue {size} -> readTVarIO size)
 
-  getQueueMessages_ :: Bool -> STMMsgQueue -> STM [Message]
-  getQueueMessages_ drainMsgs = (if drainMsgs then flushTQueue else snapshotTQueue) . msgQueue
+  getQueueMessages_ :: Bool -> STMQueue -> STMMsgQueue -> STM [Message]
+  getQueueMessages_ drainMsgs _ = (if drainMsgs then flushTQueue else snapshotTQueue) . msgQueue
     where
       snapshotTQueue q = do
         msgs <- flushTQueue q

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

@@ -60,7 +60,7 @@ class Monad (StoreMonad s) => MsgStoreClass s where
   withIdleMsgQueue :: Int64 -> s -> StoreQueue s -> (MsgQueue s -> StoreMonad s a) -> StoreMonad s (Maybe a, Int)
   deleteQueue :: s -> StoreQueue s -> IO (Either ErrorType QueueRec)
   deleteQueueSize :: s -> StoreQueue s -> IO (Either ErrorType (QueueRec, Int))
-  getQueueMessages_ :: Bool -> MsgQueue s -> StoreMonad s [Message]
+  getQueueMessages_ :: Bool -> StoreQueue s -> MsgQueue s -> StoreMonad s [Message]
   writeMsg :: s -> StoreQueue s -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
   setOverQuota_ :: StoreQueue s -> IO () -- can ONLY be used while restoring messages, not while server running
   getQueueSize_ :: MsgQueue s -> StoreMonad s Int
@@ -82,7 +82,7 @@ withActiveMsgQueues st f = readTVarIO (queues $ stmQueueStore st) >>= foldM run
     run !acc = fmap (acc <>) . f
 
 getQueueMessages :: MsgStoreClass s => Bool -> s -> StoreQueue s -> ExceptT ErrorType IO [Message]
-getQueueMessages drainMsgs st q = withPeekMsgQueue st q "getQueueSize" $ maybe (pure []) (getQueueMessages_ drainMsgs . fst)
+getQueueMessages drainMsgs st q = withPeekMsgQueue st q "getQueueSize" $ maybe (pure []) (getQueueMessages_ drainMsgs q . fst)
 {-# INLINE getQueueMessages #-}
 
 getQueueSize :: MsgStoreClass s => s -> StoreQueue s -> ExceptT ErrorType IO Int

tests/CoreTests/MsgStoreTests.hs

@@ -28,12 +28,13 @@ import qualified Data.ByteString.Base64.URL as B64
 import Data.List (isPrefixOf, isSuffixOf)
 import Data.Maybe (fromJust)
 import Data.Time.Clock (addUTCTime)
-import Data.Time.Clock.System (getSystemTime)
+import Data.Time.Clock.System (SystemTime (..), getSystemTime)
 import Simplex.Messaging.Crypto (pattern MaxLenBS)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Protocol (EntityId (..), Message (..), RecipientId, SParty (..), noMsgFlags)
 import Simplex.Messaging.Server (MessageStats (..), exportMessages, importMessages, printMessageStats)
 import Simplex.Messaging.Server.Env.STM (journalMsgStoreDepth, readWriteQueueStore)
+import Simplex.Messaging.Server.Expiration (ExpirationConfig (..), expireBeforeEpoch)
 import Simplex.Messaging.Server.MsgStore.Journal
 import Simplex.Messaging.Server.MsgStore.STM
 import Simplex.Messaging.Server.MsgStore.Types
@@ -63,6 +64,7 @@ msgStoreTests = do
       it "should create read file when read and write files are missing" testReadAndWriteFilesMissing
   describe "Journal message store: queue state backup expiration" $ do
     it "should remove old queue state backups" testRemoveQueueStateBackups
+    it "should expire messages in idle queues" testExpireIdleQueues
   where
     someMsgStoreTests :: STMStoreClass s => SpecWith s
     someMsgStoreTests = do
@@ -423,6 +425,39 @@ testRemoveQueueStateBackups = do
   where
     stateBackupCount dir = length . filter (".bak" `isSuffixOf`) <$> listDirectory dir
 
+testExpireIdleQueues :: IO ()
+testExpireIdleQueues = do
+  g <- C.newRandom
+  (rId, qr) <- testNewQueueRec g True
+
+  ms <- newMsgStore testJournalStoreCfg {idleInterval = 0}
+
+  let dir = msgQueueDirectory ms rId
+      statePath = msgQueueStatePath dir $ B.unpack (B64.encode $ unEntityId rId)
+      write q s = writeMsg ms q True =<< mkMessage s
+
+  q <- runRight $ do
+    q <- ExceptT $ addQueue ms rId qr
+    Just (Message {msgId = mId1}, True) <- write q "message 1"
+    Just (Message {msgId = mId2}, False) <- write q "message 2"
+    (Msg "message 1", Msg "message 2") <- tryDelPeekMsg ms q mId1
+    (Msg "message 2", Nothing) <- tryDelPeekMsg ms q mId2
+    liftIO $ closeMsgQueue q
+    pure q
+
+  (Just MsgQueueState {size = 0, readState = rs, writeState = ws}, True) <- readQueueState ms statePath
+  msgCount rs `shouldBe` 2
+  msgCount ws `shouldBe` 2
+
+  old <- expireBeforeEpoch ExpirationConfig {ttl = 1, checkInterval = 1} -- no old messages
+  now <- systemSeconds <$> getSystemTime
+
+  (expired_, stored) <- runRight $ idleDeleteExpiredMsgs now ms q old
+  expired_ `shouldBe` Just 0
+  stored `shouldBe` 0
+  (Nothing, False) <- readQueueState ms statePath
+  pure ()
+
 testReadFileMissing :: JournalMsgStore -> IO ()
 testReadFileMissing ms = do
   g <- C.newRandom