diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 385cd215d..be58c10b4 100644 --- a/.github/workflows/build.yml +++ b/.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 diff --git a/src/Simplex/Messaging/Server/MsgStore/Journal.hs b/src/Simplex/Messaging/Server/MsgStore/Journal.hs index 3b897de37..4498677af 100644 --- a/src/Simplex/Messaging/Server/MsgStore/Journal.hs +++ b/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,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 = diff --git a/src/Simplex/Messaging/Server/MsgStore/STM.hs b/src/Simplex/Messaging/Server/MsgStore/STM.hs index 05ab31475..ac462a71a 100644 --- a/src/Simplex/Messaging/Server/MsgStore/STM.hs +++ b/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 diff --git a/src/Simplex/Messaging/Server/MsgStore/Types.hs b/src/Simplex/Messaging/Server/MsgStore/Types.hs index 679945f55..ada1ca333 100644 --- a/src/Simplex/Messaging/Server/MsgStore/Types.hs +++ b/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 diff --git a/tests/CoreTests/MsgStoreTests.hs b/tests/CoreTests/MsgStoreTests.hs index 342b5b25f..72599f193 100644 --- a/tests/CoreTests/MsgStoreTests.hs +++ b/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