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
2026-08-29 09:58:41 +00:00 · 2025-02-18 20:04:58 +00:00
parent fa319d798a
commit a75e138965
5 changed files with 104 additions and 40 deletions
@@ -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
@@ -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,7 +483,7 @@ 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
@@ -460,9 +491,6 @@ instance MsgStoreClass JournalMsgStore where
            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)) $
@@ -470,7 +498,7 @@ instance MsgStoreClass JournalMsgStore where
      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 =
@@ -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
@@ -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
@@ -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