diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs index 08eb64ea93..382608fdee 100644 --- a/src/Simplex/Messaging/Agent/Client.hs +++ b/src/Simplex/Messaging/Agent/Client.hs @@ -34,7 +34,6 @@ module Simplex.Messaging.Agent.Client withInvLock, withLockMap, withLocksMap, - getMapLock, ipAddressProtected, closeAgentClient, closeProtocolServerClients, diff --git a/src/Simplex/Messaging/Agent/Lock.hs b/src/Simplex/Messaging/Agent/Lock.hs index 18a6f09857..081c53b680 100644 --- a/src/Simplex/Messaging/Agent/Lock.hs +++ b/src/Simplex/Messaging/Agent/Lock.hs @@ -6,7 +6,6 @@ module Simplex.Messaging.Agent.Lock withLock', withGetLock, withGetLocks, - getPutLock, ) where diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index 8b49ef4411..813b52fe14 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/src/Simplex/Messaging/Server.hs @@ -33,9 +33,6 @@ module Simplex.Messaging.Server ( runSMPServer, runSMPServerBlocking, controlPortAuth, - importMessages, - exportMessages, - printMessageStats, disconnectTransport, verifyCmdAuthorization, dummyVerifyCmd, @@ -115,6 +112,7 @@ import Simplex.Messaging.Server.NtfStore import Simplex.Messaging.Server.Prometheus import Simplex.Messaging.Server.QueueStore import Simplex.Messaging.Server.QueueStore.QueueInfo +import Simplex.Messaging.Server.QueueStore.STM (withLoadedQueues) import Simplex.Messaging.Server.QueueStore.Types import Simplex.Messaging.Server.Stats import Simplex.Messaging.Server.StoreLog (foldLogLines) @@ -144,7 +142,7 @@ import GHC.Conc.Sync (threadLabel) #endif #if defined(dbServerPostgres) -import Simplex.Messaging.Server.MsgStore.Postgres (exportDbMessages, getDbMessageStats) +import Simplex.Messaging.Server.MsgStore.Postgres (getDbMessageStats) #endif -- | Runs an SMP server using passed configuration. @@ -2346,34 +2344,27 @@ randomId = fmap EntityId . randomId' saveServerMessages :: Bool -> MsgStore s -> IO () saveServerMessages drainMsgs ms = case ms of - StoreMemory STMMsgStore {storeConfig = STMStoreConfig {storePath}} -> case storePath of - Just f -> exportMessages False ms f drainMsgs + StoreMemory ms'@STMMsgStore {storeConfig = STMStoreConfig {storePath}} -> case storePath of + Just f -> exportMessages ms' f drainMsgs Nothing -> logNote "undelivered messages are not saved" #if defined(dbServerPostgres) StoreDatabase _ -> logNote "closed postgres message storage" #endif -exportMessages :: forall s. MsgStoreClass s => Bool -> MsgStore s -> FilePath -> Bool -> IO () -exportMessages tty st f drainMsgs = do +exportMessages :: STMMsgStore -> FilePath -> Bool -> IO () +exportMessages ms f drainMsgs = do logNote $ "saving messages to file " <> T.pack f - run $ case st of - StoreMemory ms -> exportMessages_ ms $ getMsgs ms -#if defined(dbServerPostgres) - StoreDatabase ms -> exportDbMessages tty ms -#endif + run $ fmap (\(Sum n) -> n) . withLoadedQueues (queueStore ms) . saveQueueMsgs where - exportMessages_ ms get = fmap (\(Sum n) -> n) . unsafeWithAllMsgQueues tty ms . saveQueueMsgs get run :: (Handle -> IO Int) -> IO () run a = liftIO $ withFile f WriteMode $ tryAny . a >=> \case Right n -> logNote $ "messages saved: " <> tshow n Left e -> do logError $ "error exporting messages: " <> tshow e exitFailure - 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) - saveQueueMsgs get h q = do - msgs <- get q + saveQueueMsgs :: Handle -> STMQueue -> IO (Sum Int) + saveQueueMsgs h q = do + msgs <- atomically $ getQueueMessages_ drainMsgs =<< getMsgQueue q unless (null msgs) $ BLD.hPutBuilder h $ encodeMessages (recipientId q) msgs pure $ Sum $ length msgs encodeMessages rId = mconcat . map (\msg -> BLD.byteString (strEncode $ MLRv3 rId msg) <> BLD.char8 '\n') @@ -2391,7 +2382,7 @@ processServerMessages StartOptions {skipWarnings} = do 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) + Just f -> ifM (doesFileExist f) (Just <$> importMessages ms f old_ skipWarnings) (pure Nothing) Nothing -> pure Nothing #if defined(dbServerPostgres) StoreDatabase ms -> processDbMessages old_ expire ms @@ -2406,18 +2397,17 @@ processServerMessages StartOptions {skipWarnings} = do | otherwise = logWarn "skipping message expiration" $> Nothing #endif -importMessages :: forall s. MsgStoreClass s => Bool -> s -> FilePath -> Maybe Int64 -> Bool -> IO MessageStats -importMessages tty ms f old_ skipWarnings = do +importMessages :: STMMsgStore -> FilePath -> Maybe Int64 -> Bool -> IO MessageStats +importMessages ms f old_ skipWarnings = do logNote $ "restoring messages from file " <> T.pack f (_, (storedMsgsCount, expiredMsgsCount, overQuota)) <- - foldLogLines tty f restoreMsg (Nothing, (0, 0, M.empty)) + foldLogLines False f restoreMsg (Nothing, (0, 0, M.empty)) renameFile f $ f <> ".bak" mapM_ setOverQuota_ overQuota - logQueueStates ms - EntityCounts {queueCount} <- liftIO $ getEntityCounts @(StoreQueue s) $ queueStore ms + EntityCounts {queueCount} <- liftIO $ getEntityCounts @STMQueue $ queueStore ms pure MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues = queueCount} where - restoreMsg :: (Maybe (RecipientId, StoreQueue s), (Int, Int, Map RecipientId (StoreQueue s))) -> Bool -> ByteString -> IO (Maybe (RecipientId, StoreQueue s), (Int, Int, Map RecipientId (StoreQueue s))) + restoreMsg :: (Maybe (RecipientId, STMQueue), (Int, Int, Map RecipientId STMQueue)) -> Bool -> ByteString -> IO (Maybe (RecipientId, STMQueue), (Int, Int, Map RecipientId STMQueue)) restoreMsg (q_, counts@(!stored, !expired, !overQuota)) eof s = case strDecode s of Right (MLRv3 rId msg) -> runExceptT (addToMsgQueue rId msg) >>= either (exitErr . tshow) pure Left e @@ -2425,7 +2415,6 @@ importMessages tty ms f old_ skipWarnings = do | otherwise -> exitErr $ parsingErr e where exitErr e = do - when tty $ putStrLn "" logError $ "error restoring messages: " <> e liftIO exitFailure parsingErr :: String -> Text @@ -2438,7 +2427,6 @@ importMessages tty ms f old_ skipWarnings = do case qOrErr of Right q -> addToQueue_ q rId msg Left AUTH -> liftIO $ do - when tty $ putStrLn "" warnOrExit $ "queue " <> safeDecodeUtf8 (encode $ unEntityId rId) <> " does not exist" pure (Nothing, counts) Left e -> throwE e @@ -2449,7 +2437,6 @@ importMessages tty ms f old_ skipWarnings = do writeMsg ms q False msg >>= \case Just _ -> pure (stored + 1, expired, overQuota) Nothing -> liftIO $ do - when tty $ putStrLn "" logError $ decodeLatin1 $ "message queue " <> strEncode rId <> " is full, message not restored: " <> strEncode (messageId msg) pure counts | otherwise -> pure (stored, expired + 1, overQuota) @@ -2460,8 +2447,8 @@ importMessages tty ms f old_ skipWarnings = do where -- if the first message in queue head is "quota", remove it. mergeQuotaMsgs = - withPeekMsgQueue ms q "mergeQuotaMsgs" $ maybe (pure ()) $ \case - (mq, MessageQuota {}) -> tryDeleteMsg_ q mq False + withPeekMsgQueue q $ maybe (pure ()) $ \case + (mq, MessageQuota {}) -> tryDeleteMsg_ mq _ -> pure () warnOrExit e | skipWarnings = logWarn e' diff --git a/src/Simplex/Messaging/Server/Env/STM.hs b/src/Simplex/Messaging/Server/Env/STM.hs index 40000db9dd..4c0b1d566e 100644 --- a/src/Simplex/Messaging/Server/Env/STM.hs +++ b/src/Simplex/Messaging/Server/Env/STM.hs @@ -555,7 +555,7 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smp SSCMemory storePaths_ -> do let storePath = storeMsgsFile =<< storePaths_ ms <- newMsgStore STMStoreConfig {storePath, quota = msgQueueQuota} - forM_ storePaths_ $ \StorePaths {storeLogFile = f} -> loadStoreLog (mkQueue ms True) f $ queueStore ms + forM_ storePaths_ $ \StorePaths {storeLogFile = f} -> loadStoreLog (mkQueue ms) f $ queueStore ms pure $ StoreMemory ms #if defined(dbServerPostgres) SSCDatabase storeCfg -> do @@ -614,8 +614,7 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smp Just f -> do logNote $ "compacting queues in file " <> T.pack f st <- newMsgStore STMStoreConfig {storePath = Nothing, quota = msgQueueQuota} - -- we don't need to have locks in the map - sl <- readWriteQueueStore False (mkQueue st False) f (queueStore st) + sl <- readWriteQueueStore False (mkQueue st) f (queueStore st) setStoreLog (queueStore st) sl closeMsgStore st Nothing -> do diff --git a/src/Simplex/Messaging/Server/Main.hs b/src/Simplex/Messaging/Server/Main.hs index 53864c5bd3..9966248388 100644 --- a/src/Simplex/Messaging/Server/Main.hs +++ b/src/Simplex/Messaging/Server/Main.hs @@ -28,7 +28,6 @@ module Simplex.Messaging.Server.Main importMessagesToDatabase, exportDatabaseToStoreLog, #endif - storeMsgsJournalDir', getServerSourceCode, simplexmqSource, serverPublicInfo, @@ -95,7 +94,7 @@ import Data.Semigroup (Sum (..)) import Simplex.Messaging.Agent.Store.Postgres (checkSchemaExists) import Simplex.Messaging.Agent.Store.Shared (MigrationConfirmation (..)) import Simplex.Messaging.Server.MsgStore.Postgres -import Simplex.Messaging.Server.MsgStore.STM (STMQueue, STMStoreConfig (..)) +import Simplex.Messaging.Server.MsgStore.STM (STMStoreConfig (..)) import Simplex.Messaging.Server.MsgStore.Types (MsgStoreClass (..), newMsgStore) import Simplex.Messaging.Server.QueueStore.Postgres.Config import Simplex.Messaging.Server.QueueStore.Postgres (PostgresQueueStore, batchInsertQueues, batchInsertServices, foldQueueRecs, foldServiceRecs) @@ -617,14 +616,14 @@ importStoreLogToDatabase :: FilePath -> DBOpts -> IO (Int64, Int64) importStoreLogToDatabase storeLogFile dbOpts = do ms <- newMsgStore STMStoreConfig {storePath = Nothing, quota = defaultMsgQueueQuota} let st = queueStore ms - sl <- readWriteQueueStore True (mkQueue ms False) storeLogFile st + sl <- readWriteQueueStore True (mkQueue ms) storeLogFile st closeStoreLog sl - queues <- readTVarIO $ loadedQueues st + qs <- readTVarIO $ queues st services' <- M.elems <$> readTVarIO (services st) let storeCfg = PostgresStoreCfg {dbOpts = dbOpts {createSchema = True}, dbStoreLogPath = Nothing, confirmMigrations = MCConsole, deletedTTL = 86400 * defaultDeletedTTL} - ps :: PostgresQueueStore PostgresQueue <- newQueueStore @PostgresQueue (storeCfg, False) + ps :: PostgresQueueStore PostgresQueue <- newQueueStore @PostgresQueue storeCfg sCnt <- batchInsertServices services' ps - qCnt <- batchInsertQueues @STMQueue True queues ps + qCnt <- batchInsertQueues True qs ps closeQueueStore @PostgresQueue ps renameFile storeLogFile $ storeLogFile <> ".bak" pure (sCnt, qCnt) @@ -647,7 +646,7 @@ importMessagesToDatabase msgsLogFile dbOpts = do exportDatabaseToStoreLog :: DBOpts -> FilePath -> IO (Int, Int) exportDatabaseToStoreLog dbOpts storeLogFilePath = do let storeCfg = PostgresStoreCfg {dbOpts, dbStoreLogPath = Nothing, confirmMigrations = MCConsole, deletedTTL = 86400 * defaultDeletedTTL} - ps :: PostgresQueueStore PostgresQueue <- newQueueStore @PostgresQueue (storeCfg, False) + ps :: PostgresQueueStore PostgresQueue <- newQueueStore @PostgresQueue storeCfg sl <- openWriteStoreLog False storeLogFilePath Sum sCnt <- foldServiceRecs ps $ \sr -> logNewService sl sr $> Sum (1 :: Int) Sum qCnt <- foldQueueRecs True True ps $ \(rId, qr) -> logCreateQueue sl rId qr $> Sum (1 :: Int) diff --git a/src/Simplex/Messaging/Server/MsgStore/Postgres.hs b/src/Simplex/Messaging/Server/MsgStore/Postgres.hs index 821c5e7f25..b13d1e0a53 100644 --- a/src/Simplex/Messaging/Server/MsgStore/Postgres.hs +++ b/src/Simplex/Messaging/Server/MsgStore/Postgres.hs @@ -29,7 +29,6 @@ where import Control.Concurrent.STM import qualified Control.Exception as E import Control.Monad -import Control.Monad.Reader import Control.Monad.Trans.Except import qualified Data.ByteString as B import qualified Data.ByteString.Builder as BB @@ -39,7 +38,6 @@ import Data.IORef import Data.Int (Int64) import Data.List (intersperse) import qualified Data.Map.Strict as M -import Data.Text (Text) import Data.Time.Clock.System (SystemTime (..)) import Database.PostgreSQL.Simple (Binary (..), In (..), Only (..), (:.) (..)) import qualified Database.PostgreSQL.Simple as DB @@ -56,7 +54,7 @@ import Simplex.Messaging.Server.QueueStore.Postgres import Simplex.Messaging.Server.QueueStore.Types import Simplex.Messaging.Server.StoreLog (foldLogLines) import Simplex.Messaging.Encoding.String -import Simplex.Messaging.Util (maybeFirstRow, maybeFirstRow', (<$$>)) +import Simplex.Messaging.Util (maybeFirstRow, maybeFirstRow') import System.IO (Handle, hFlush, stdout) data PostgresMsgStore = PostgresMsgStore @@ -81,32 +79,20 @@ instance StoreQueueClass PostgresQueue where {-# INLINE recipientId #-} queueRec = queueRec' {-# INLINE queueRec #-} - withQueueLock PostgresQueue {} _ = id -- TODO [messages] maybe it's just transaction? - {-# INLINE withQueueLock #-} - -newtype DBTransaction = DBTransaction {dbConn :: DB.Connection} - -type DBStoreIO a = ReaderT DBTransaction IO a instance MsgStoreClass PostgresMsgStore where - type StoreMonad PostgresMsgStore = ReaderT DBTransaction IO - type MsgQueue PostgresMsgStore = () type QueueStore PostgresMsgStore = PostgresQueueStore' type StoreQueue PostgresMsgStore = PostgresQueue type MsgStoreConfig PostgresMsgStore = PostgresMsgStoreCfg newMsgStore :: PostgresMsgStoreCfg -> IO PostgresMsgStore newMsgStore config = do - queueStore_ <- newQueueStore @PostgresQueue (queueStoreCfg config, False) + queueStore_ <- newQueueStore @PostgresQueue (queueStoreCfg config) pure PostgresMsgStore {config, queueStore_} closeMsgStore :: PostgresMsgStore -> IO () closeMsgStore = closeQueueStore @PostgresQueue . queueStore_ - withActiveMsgQueues _ _ = error "withActiveMsgQueues not used" - - unsafeWithAllMsgQueues _ _ _ = error "unsafeWithAllMsgQueues not used" - expireOldMessages :: Bool -> PostgresMsgStore -> Int64 -> Int64 -> IO MessageStats expireOldMessages _tty ms now ttl = maybeFirstRow' newMessageStats toMessageStats $ withConnection st $ \db -> @@ -150,35 +136,16 @@ instance MsgStoreClass PostgresMsgStore where msg_ = toMaybeMessage mRow in f a rId $ Right ((qr,) <$> msg_) - logQueueStates _ = error "logQueueStates not used" - - logQueueState _ = error "logQueueState not used" - queueStore = queueStore_ {-# INLINE queueStore #-} loadedQueueCounts :: PostgresMsgStore -> IO LoadedQueueCounts - loadedQueueCounts ms = do - loadedQueueCount <- M.size <$> readTVarIO queues - loadedNotifierCount <- M.size <$> readTVarIO notifiers - notifierLockCount <- M.size <$> readTVarIO notifierLocks - pure LoadedQueueCounts {loadedQueueCount, loadedNotifierCount, queueLockCount = 0, notifierLockCount} - where - PostgresQueueStore {queues, notifiers, notifierLocks} = queueStore_ ms + loadedQueueCounts _ = pure LoadedQueueCounts {loadedQueueCount = 0, loadedNotifierCount = 0} - mkQueue :: PostgresMsgStore -> Bool -> RecipientId -> QueueRec -> IO PostgresQueue - mkQueue _ _keepLock rId qr = PostgresQueue rId <$> newTVarIO (Just qr) + mkQueue :: PostgresMsgStore -> RecipientId -> QueueRec -> IO PostgresQueue + mkQueue _ rId qr = PostgresQueue rId <$> newTVarIO (Just qr) {-# INLINE mkQueue #-} - getMsgQueue _ _ _ = pure () - {-# INLINE getMsgQueue #-} - - getPeekMsgQueue :: PostgresMsgStore -> PostgresQueue -> DBStoreIO (Maybe ((), Message)) - getPeekMsgQueue _ q = ((),) <$$> tryPeekMsg_ q () - - withIdleMsgQueue :: Int64 -> PostgresMsgStore -> PostgresQueue -> (() -> DBStoreIO a) -> DBStoreIO (Maybe a, Int) - withIdleMsgQueue _ _ _ _ = error "withIdleMsgQueue not used" - deleteQueue :: PostgresMsgStore -> PostgresQueue -> IO (Either ErrorType QueueRec) deleteQueue ms q = deleteStoreQueue (queueStore_ ms) q {-# INLINE deleteQueue #-} @@ -189,8 +156,6 @@ instance MsgStoreClass PostgresMsgStore where qr <- ExceptT $ deleteStoreQueue (queueStore_ ms) q pure (qr, size) - getQueueMessages_ _ _ _ = error "getQueueMessages_ not used" - writeMsg :: PostgresMsgStore -> PostgresQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool)) writeMsg ms q _ msg = uninterruptibleMask_ $ @@ -209,43 +174,25 @@ instance MsgStoreClass PostgresMsgStore where [] -> Nothing PostgresMsgStore {config = PostgresMsgStoreCfg {quota}} = ms - setOverQuota_ :: PostgresQueue -> IO () -- can ONLY be used while restoring messages, not while server running - setOverQuota_ _ = error "TODO setOverQuota_" -- TODO [messages] - - getQueueSize_ :: () -> DBStoreIO Int - getQueueSize_ _ = error "getQueueSize_ not used" - getQueueSize :: PostgresMsgStore -> PostgresQueue -> ExceptT ErrorType IO Int getQueueSize ms q = withDB' "getQueueSize" (queueStore_ ms) $ \db -> maybeFirstRow' 0 fromOnly $ DB.query db "SELECT msg_queue_size FROM msg_queues WHERE recipient_id = ? AND deleted_at IS NULL" (Only (recipientId' q)) - tryPeekMsg_ :: PostgresQueue -> () -> DBStoreIO (Maybe Message) - tryPeekMsg_ q _ = do - db <- asks dbConn - liftIO $ maybeFirstRow toMessage $ - DB.query - db - [sql| - SELECT msg_id, msg_ts, msg_quota, msg_ntf_flag, msg_body - FROM messages - WHERE recipient_id = ? - ORDER BY message_id ASC LIMIT 1 - |] - (Only (recipientId' q)) - - tryDeleteMsg_ :: PostgresQueue -> () -> Bool -> DBStoreIO () - tryDeleteMsg_ _q _ _ = error "tryDeleteMsg_ not used" -- do - - isolateQueue :: PostgresMsgStore -> PostgresQueue -> Text -> DBStoreIO a -> ExceptT ErrorType IO a - isolateQueue ms _q op a = uninterruptibleMask_ $ withDB' op (queueStore_ ms) $ runReaderT a . DBTransaction - - unsafeRunStore _ _ _ = error "unsafeRunStore not used" - tryPeekMsg :: PostgresMsgStore -> PostgresQueue -> ExceptT ErrorType IO (Maybe Message) - tryPeekMsg ms q = isolateQueue ms q "tryPeekMsg" $ tryPeekMsg_ q () - {-# INLINE tryPeekMsg #-} + tryPeekMsg ms q = + uninterruptibleMask_ $ withDB' "tryPeekMsg" (queueStore_ ms) $ \db -> + maybeFirstRow toMessage $ + DB.query + db + [sql| + SELECT msg_id, msg_ts, msg_quota, msg_ntf_flag, msg_body + FROM messages + WHERE recipient_id = ? + ORDER BY message_id ASC LIMIT 1 + |] + (Only (recipientId' q)) tryPeekMsgs :: PostgresMsgStore -> [PostgresQueue] -> ExceptT ErrorType IO (M.Map RecipientId Message) tryPeekMsgs _ms [] = pure M.empty diff --git a/src/Simplex/Messaging/Server/MsgStore/STM.hs b/src/Simplex/Messaging/Server/MsgStore/STM.hs index 0607ef8b56..1f9b7a256b 100644 --- a/src/Simplex/Messaging/Server/MsgStore/STM.hs +++ b/src/Simplex/Messaging/Server/MsgStore/STM.hs @@ -15,6 +15,11 @@ module Simplex.Messaging.Server.MsgStore.STM ( STMMsgStore (..), STMStoreConfig (..), STMQueue, + getMsgQueue, + withPeekMsgQueue, + getQueueMessages_, + setOverQuota_, + tryDeleteMsg_, ) where @@ -24,7 +29,8 @@ import Control.Monad.Trans.Except import Data.Functor (($>)) import Data.Int (Int64) import qualified Data.Map.Strict as M -import Data.Text (Text) +import Data.Maybe (catMaybes) +import Data.Time.Clock.System (SystemTime (systemSeconds)) import Simplex.Messaging.Protocol import Simplex.Messaging.Server.MsgStore.Types import Simplex.Messaging.Server.QueueStore @@ -61,12 +67,8 @@ instance StoreQueueClass STMQueue where {-# INLINE recipientId #-} queueRec = queueRec' {-# INLINE queueRec #-} - withQueueLock _ _ = id - {-# INLINE withQueueLock #-} instance MsgStoreClass STMMsgStore where - type StoreMonad STMMsgStore = STM - type MsgQueue STMMsgStore = STMMsgQueue type QueueStore STMMsgStore = STMQueueStore STMQueue type StoreQueue STMMsgStore = STMQueue type MsgStoreConfig STMMsgStore = STMStoreConfig @@ -78,24 +80,16 @@ instance MsgStoreClass STMMsgStore where closeMsgStore = closeQueueStore @STMQueue . queueStore_ {-# INLINE closeMsgStore #-} - withActiveMsgQueues = withLoadedQueues . queueStore_ - {-# INLINE withActiveMsgQueues #-} - unsafeWithAllMsgQueues _ = withLoadedQueues . queueStore_ - {-# INLINE unsafeWithAllMsgQueues #-} expireOldMessages :: Bool -> STMMsgStore -> Int64 -> Int64 -> IO MessageStats expireOldMessages _tty ms now ttl = - withLoadedQueues (queueStore_ ms) $ atomically . expireQueueMsgs ms now (now - ttl) + withLoadedQueues (queueStore_ ms) $ atomically . expireQueueMsgs (now - ttl) foldRcvServiceMessages :: STMMsgStore -> ServiceId -> (a -> RecipientId -> Either ErrorType (Maybe (QueueRec, Message)) -> IO a) -> a -> IO (Either ErrorType a) foldRcvServiceMessages ms serviceId f = fmap Right . foldRcvServiceQueues (queueStore_ ms) serviceId f' where f' a (q, qr) = runExceptT (tryPeekMsg ms q) >>= f a (recipientId q) . ((qr,) <$$>) - logQueueStates _ = pure () - {-# INLINE logQueueStates #-} - logQueueState _ = pure () - {-# INLINE logQueueState #-} queueStore = queueStore_ {-# INLINE queueStore #-} @@ -103,34 +97,11 @@ instance MsgStoreClass STMMsgStore where loadedQueueCounts STMMsgStore {queueStore_ = st} = do loadedQueueCount <- M.size <$> readTVarIO (queues st) loadedNotifierCount <- M.size <$> readTVarIO (notifiers st) - pure LoadedQueueCounts {loadedQueueCount, loadedNotifierCount, queueLockCount = 0, notifierLockCount = 0} + pure LoadedQueueCounts {loadedQueueCount, loadedNotifierCount} - mkQueue _ _ rId qr = STMQueue rId <$> newTVarIO (Just qr) <*> newTVarIO Nothing + mkQueue _ rId qr = STMQueue rId <$> newTVarIO (Just qr) <*> newTVarIO Nothing {-# INLINE mkQueue #-} - getMsgQueue :: STMMsgStore -> STMQueue -> Bool -> STM STMMsgQueue - getMsgQueue _ STMQueue {msgQueue'} _ = readTVar msgQueue' >>= maybe newQ pure - where - newQ = do - msgTQueue <- newTQueue - canWrite <- newTVar True - size <- newTVar 0 - let q = STMMsgQueue {msgTQueue, canWrite, size} - writeTVar msgQueue' (Just q) - pure q - - getPeekMsgQueue :: STMMsgStore -> STMQueue -> STM (Maybe (STMMsgQueue, Message)) - getPeekMsgQueue _ q@STMQueue {msgQueue'} = readTVar msgQueue' $>>= \mq -> (mq,) <$$> tryPeekMsg_ q mq - - -- 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 -> STMQueue -> (STMMsgQueue -> STM a) -> STM (Maybe a, Int) - withIdleMsgQueue _ _ STMQueue {msgQueue'} action = readTVar msgQueue' >>= \case - Just q -> do - r <- action q - sz <- getQueueSize_ q - pure (Just r, sz) - Nothing -> pure (Nothing, 0) - deleteQueue :: STMMsgStore -> STMQueue -> IO (Either ErrorType QueueRec) deleteQueue ms q = fst <$$> deleteQueue_ ms q @@ -140,17 +111,9 @@ instance MsgStoreClass STMMsgStore where where getSize = maybe (pure 0) (\STMMsgQueue {size} -> readTVarIO size) - getQueueMessages_ :: Bool -> STMQueue -> STMMsgQueue -> STM [Message] - getQueueMessages_ drainMsgs _ = (if drainMsgs then flushTQueue else snapshotTQueue) . msgTQueue - where - snapshotTQueue q = do - msgs <- flushTQueue q - mapM_ (writeTQueue q) msgs - pure msgs - writeMsg :: STMMsgStore -> STMQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool)) writeMsg ms q' _logState msg = liftIO $ atomically $ do - STMMsgQueue {msgTQueue = q, canWrite, size} <- getMsgQueue ms q' True + STMMsgQueue {msgTQueue = q, canWrite, size} <- getMsgQueue q' canWrt <- readTVar canWrite empty <- isEmptyTQueue q if canWrt || empty @@ -166,29 +129,96 @@ instance MsgStoreClass STMMsgStore where STMMsgStore {storeConfig = STMStoreConfig {quota}} = ms msgQuota = MessageQuota {msgId = messageId msg, msgTs = messageTs msg} - setOverQuota_ :: STMQueue -> IO () - setOverQuota_ q = readTVarIO (msgQueue' q) >>= mapM_ (\mq -> atomically $ writeTVar (canWrite mq) False) - - getQueueSize_ :: STMMsgQueue -> STM Int - getQueueSize_ STMMsgQueue {size} = readTVar size - - tryPeekMsg_ :: STMQueue -> STMMsgQueue -> STM (Maybe Message) - tryPeekMsg_ _ = tryPeekTQueue . msgTQueue - {-# INLINE tryPeekMsg_ #-} - - tryDeleteMsg_ :: STMQueue -> STMMsgQueue -> Bool -> STM () - tryDeleteMsg_ _ STMMsgQueue {msgTQueue = q, size} _logState = - tryReadTQueue q >>= \case - Just _ -> modifyTVar' size (subtract 1) - _ -> pure () - - isolateQueue :: STMMsgStore -> STMQueue -> Text -> STM a -> ExceptT ErrorType IO a - isolateQueue _ _ _ = liftIO . atomically - {-# INLINE isolateQueue #-} - - unsafeRunStore :: STMQueue -> Text -> STM a -> IO a - unsafeRunStore _ _ = atomically - {-# INLINE unsafeRunStore #-} + tryPeekMsg :: STMMsgStore -> STMQueue -> ExceptT ErrorType IO (Maybe Message) + tryPeekMsg _ q = snd <$$> withPeekMsgQueue q pure + {-# INLINE tryPeekMsg #-} + + tryPeekMsgs :: STMMsgStore -> [STMQueue] -> ExceptT ErrorType IO (M.Map RecipientId Message) + tryPeekMsgs st qs = M.fromList . catMaybes <$> mapM (\q -> (recipientId q,) <$$> tryPeekMsg st q) qs + + tryDelMsg :: STMMsgStore -> STMQueue -> MsgId -> ExceptT ErrorType IO (Maybe Message) + tryDelMsg _ q msgId' = + withPeekMsgQueue q $ + maybe (pure Nothing) $ \(mq, msg) -> + if messageId msg == msgId' + then tryDeleteMsg_ mq $> Just msg + else pure Nothing + + tryDelPeekMsg :: STMMsgStore -> STMQueue -> MsgId -> ExceptT ErrorType IO (Maybe Message, Maybe Message) + tryDelPeekMsg _ q msgId' = + withPeekMsgQueue q $ + maybe (pure (Nothing, Nothing)) $ \(mq, msg) -> + if messageId msg == msgId' + then (Just msg,) <$> (tryDeleteMsg_ mq >> tryPeekMsg_ mq) + else pure (Nothing, Just msg) + + deleteExpiredMsgs :: STMMsgStore -> STMQueue -> Int64 -> ExceptT ErrorType IO Int + deleteExpiredMsgs _ q old = liftIO $ atomically $ getMsgQueue q >>= deleteExpireMsgs_ old + + getQueueSize :: STMMsgStore -> STMQueue -> ExceptT ErrorType IO Int + getQueueSize _ q = withPeekMsgQueue q $ maybe (pure 0) (getQueueSize_ . fst) + {-# INLINE getQueueSize #-} + +-- can ONLY be used while restoring messages, not while server running +setOverQuota_ :: STMQueue -> IO () +setOverQuota_ q = readTVarIO (msgQueue' q) >>= mapM_ (\mq -> atomically $ writeTVar (canWrite mq) False) + +getQueueSize_ :: STMMsgQueue -> STM Int +getQueueSize_ STMMsgQueue {size} = readTVar size + +tryPeekMsg_ :: STMMsgQueue -> STM (Maybe Message) +tryPeekMsg_ = tryPeekTQueue . msgTQueue +{-# INLINE tryPeekMsg_ #-} + +tryDeleteMsg_ :: STMMsgQueue -> STM () +tryDeleteMsg_ STMMsgQueue {msgTQueue = q, size} = + tryReadTQueue q >>= \case + Just _ -> modifyTVar' size (subtract 1) + _ -> pure () + +getMsgQueue :: STMQueue -> STM STMMsgQueue +getMsgQueue STMQueue {msgQueue'} = readTVar msgQueue' >>= maybe newQ pure + where + newQ = do + msgTQueue <- newTQueue + canWrite <- newTVar True + size <- newTVar 0 + let q = STMMsgQueue {msgTQueue, canWrite, size} + writeTVar msgQueue' (Just q) + pure q + +-- The action is called with Nothing when it is known that the queue is empty +withPeekMsgQueue :: STMQueue -> (Maybe (STMMsgQueue, Message) -> STM a) -> ExceptT ErrorType IO a +withPeekMsgQueue STMQueue {msgQueue'} a = + liftIO $ atomically $ (readTVar msgQueue' $>>= \mq -> (mq,) <$$> tryPeekMsg_ mq) >>= a + +getQueueMessages_ :: Bool -> STMMsgQueue -> STM [Message] +getQueueMessages_ drainMsgs = (if drainMsgs then flushTQueue else snapshotTQueue) . msgTQueue + where + snapshotTQueue q = do + msgs <- flushTQueue q + mapM_ (writeTQueue q) msgs + pure msgs + +-- does not create queue if it does not exist +expireQueueMsgs :: Int64 -> STMQueue -> STM MessageStats +expireQueueMsgs old STMQueue {msgQueue'} = + readTVar msgQueue' >>= \case + Just mq -> do + expiredMsgsCount <- deleteExpireMsgs_ old mq + storedMsgsCount <- getQueueSize_ mq + pure MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues = 1} + Nothing -> pure newMessageStats {storedQueues = 1} + +deleteExpireMsgs_ :: Int64 -> STMMsgQueue -> STM Int +deleteExpireMsgs_ old mq = loop 0 + where + loop dc = + tryPeekMsg_ mq >>= \case + Just Message {msgTs} + | systemSeconds msgTs < old -> + tryDeleteMsg_ mq >> loop (dc + 1) + _ -> pure dc deleteQueue_ :: STMMsgStore -> STMQueue -> IO (Either ErrorType (QueueRec, Maybe STMMsgQueue)) deleteQueue_ ms q = deleteStoreQueue (queueStore_ ms) q >>= mapM remove diff --git a/src/Simplex/Messaging/Server/MsgStore/Types.hs b/src/Simplex/Messaging/Server/MsgStore/Types.hs index dce3bc638f..a689469d71 100644 --- a/src/Simplex/Messaging/Server/MsgStore/Types.hs +++ b/src/Simplex/Messaging/Server/MsgStore/Types.hs @@ -4,17 +4,12 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE GADTs #-} {-# LANGUAGE LambdaCase #-} -{-# LANGUAGE MultiWayIf #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TupleSections #-} {-# LANGUAGE TypeApplications #-} {-# LANGUAGE TypeFamilyDependencies #-} -{-# OPTIONS_GHC -Wno-unrecognised-pragmas #-} - -{-# HLINT ignore "Redundant multi-way if" #-} - module Simplex.Messaging.Server.MsgStore.Types ( MsgStoreClass (..), MSType (..), @@ -30,99 +25,43 @@ module Simplex.Messaging.Server.MsgStore.Types getQueues, getQueueRecs, readQueueRec, - withPeekMsgQueue, - expireQueueMsgs, - deleteExpireMsgs_, ) where import Control.Concurrent.STM import Control.Monad import Control.Monad.Trans.Except -import Data.Functor (($>)) import Data.Int (Int64) import Data.Kind import Data.Map.Strict (Map) -import qualified Data.Map.Strict as M -import Data.Maybe (catMaybes, fromMaybe) -import Data.Text (Text) -import Data.Time.Clock.System (SystemTime (systemSeconds)) import Simplex.Messaging.Protocol import Simplex.Messaging.Server.QueueStore import Simplex.Messaging.Server.QueueStore.Types -import Simplex.Messaging.Util ((<$$>), ($>>=)) +import Simplex.Messaging.Util (($>>=)) -class (Monad (StoreMonad s), QueueStoreClass (StoreQueue s) (QueueStore s)) => MsgStoreClass s where - type StoreMonad s = (m :: Type -> Type) | m -> s +class QueueStoreClass (StoreQueue s) (QueueStore s) => MsgStoreClass s where type MsgStoreConfig s = c | c -> s - type MsgQueue s = q | q -> s type StoreQueue s = q | q -> s type QueueStore s = qs | qs -> s newMsgStore :: MsgStoreConfig s -> IO s closeMsgStore :: s -> IO () - withActiveMsgQueues :: Monoid a => s -> (StoreQueue s -> IO a) -> IO a - -- This function can only be used in server CLI commands or before server is started. - -- tty, store - unsafeWithAllMsgQueues :: Monoid a => Bool -> s -> (StoreQueue s -> IO a) -> IO a -- tty, store, now, ttl expireOldMessages :: Bool -> s -> Int64 -> Int64 -> IO MessageStats foldRcvServiceMessages :: s -> ServiceId -> (a -> RecipientId -> Either ErrorType (Maybe (QueueRec, Message)) -> IO a) -> a -> IO (Either ErrorType a) - logQueueStates :: s -> IO () - logQueueState :: StoreQueue s -> StoreMonad s () queueStore :: s -> QueueStore s loadedQueueCounts :: s -> IO LoadedQueueCounts -- message store methods - mkQueue :: s -> Bool -> RecipientId -> QueueRec -> IO (StoreQueue s) - getMsgQueue :: s -> StoreQueue s -> Bool -> StoreMonad s (MsgQueue s) - getPeekMsgQueue :: s -> StoreQueue s -> StoreMonad s (Maybe (MsgQueue s, Message)) - - -- the journal queue will be closed after action if it was initially closed or idle longer than interval in config - withIdleMsgQueue :: Int64 -> s -> StoreQueue s -> (MsgQueue s -> StoreMonad s a) -> StoreMonad s (Maybe a, Int) + mkQueue :: s -> RecipientId -> QueueRec -> IO (StoreQueue s) deleteQueue :: s -> StoreQueue s -> IO (Either ErrorType QueueRec) deleteQueueSize :: s -> StoreQueue s -> IO (Either ErrorType (QueueRec, Int)) - 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 - tryPeekMsg_ :: StoreQueue s -> MsgQueue s -> StoreMonad s (Maybe Message) - tryDeleteMsg_ :: StoreQueue s -> MsgQueue s -> Bool -> StoreMonad s () - isolateQueue :: s -> StoreQueue s -> Text -> StoreMonad s a -> ExceptT ErrorType IO a - unsafeRunStore :: StoreQueue s -> Text -> StoreMonad s a -> IO a - - -- default implementations are overridden for PostgreSQL storage of messages tryPeekMsg :: s -> StoreQueue s -> ExceptT ErrorType IO (Maybe Message) - tryPeekMsg st q = snd <$$> withPeekMsgQueue st q "tryPeekMsg" pure - {-# INLINE tryPeekMsg #-} - tryPeekMsgs :: s -> [StoreQueue s] -> ExceptT ErrorType IO (Map RecipientId Message) - tryPeekMsgs st qs = M.fromList . catMaybes <$> mapM (\q -> (recipientId q,) <$$> tryPeekMsg st q) qs - tryDelMsg :: s -> StoreQueue s -> MsgId -> ExceptT ErrorType IO (Maybe Message) - tryDelMsg st q msgId' = - withPeekMsgQueue st q "tryDelMsg" $ - maybe (pure Nothing) $ \(mq, msg) -> - if - | messageId msg == msgId' -> - tryDeleteMsg_ q mq True $> Just msg - | otherwise -> pure Nothing - -- atomic delete (== read) last and peek next message if available tryDelPeekMsg :: s -> StoreQueue s -> MsgId -> ExceptT ErrorType IO (Maybe Message, Maybe Message) - tryDelPeekMsg st q msgId' = - withPeekMsgQueue st q "tryDelPeekMsg" $ - maybe (pure (Nothing, Nothing)) $ \(mq, msg) -> - if - | messageId msg == msgId' -> (Just msg,) <$> (tryDeleteMsg_ q mq True >> tryPeekMsg_ q mq) - | otherwise -> pure (Nothing, Just msg) - deleteExpiredMsgs :: s -> StoreQueue s -> Int64 -> ExceptT ErrorType IO Int - deleteExpiredMsgs st q old = - isolateQueue st q "deleteExpiredMsgs" $ - getMsgQueue st q False >>= deleteExpireMsgs_ old q - getQueueSize :: s -> StoreQueue s -> ExceptT ErrorType IO Int - getQueueSize st q = withPeekMsgQueue st q "getQueueSize" $ maybe (pure 0) (getQueueSize_ . fst) - {-# INLINE getQueueSize #-} data MSType = MSMemory | MSPostgres @@ -153,16 +92,14 @@ instance Semigroup MessageStats where data LoadedQueueCounts = LoadedQueueCounts { loadedQueueCount :: Int, - loadedNotifierCount :: Int, - queueLockCount :: Int, - notifierLockCount :: Int + loadedNotifierCount :: Int } newMessageStats :: MessageStats newMessageStats = MessageStats 0 0 0 addQueue :: MsgStoreClass s => s -> RecipientId -> QueueRec -> IO (Either ErrorType (StoreQueue s)) -addQueue st = addQueue_ (queueStore st) (mkQueue st True) +addQueue st = addQueue_ (queueStore st) (mkQueue st) {-# INLINE addQueue #-} getQueue :: (MsgStoreClass s, QueueParty p) => s -> SParty p -> QueueId -> IO (Either ErrorType (StoreQueue s)) @@ -182,28 +119,3 @@ getQueueRecs st party qIds = getQueues st party qIds >>= mapM (fmap join . mapM readQueueRec :: StoreQueueClass q => q -> IO (Either ErrorType (q, QueueRec)) readQueueRec q = maybe (Left AUTH) (Right . (q,)) <$> readTVarIO (queueRec q) {-# INLINE readQueueRec #-} - --- The action is called with Nothing when it is known that the queue is empty -withPeekMsgQueue :: MsgStoreClass s => s -> StoreQueue s -> Text -> (Maybe (MsgQueue s, Message) -> StoreMonad s a) -> ExceptT ErrorType IO a -withPeekMsgQueue st q op a = isolateQueue st q op $ getPeekMsgQueue st q >>= a -{-# INLINE withPeekMsgQueue #-} - --- not used with PostgreSQL message store -expireQueueMsgs :: MsgStoreClass s => s -> Int64 -> Int64 -> StoreQueue s -> StoreMonad s MessageStats -expireQueueMsgs st now old q = do - (expired_, stored) <- withIdleMsgQueue now st q $ deleteExpireMsgs_ old q - pure MessageStats {storedMsgsCount = stored, expiredMsgsCount = fromMaybe 0 expired_, storedQueues = 1} - --- not used with PostgreSQL message store -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 = - tryPeekMsg_ q mq >>= \case - Just Message {msgTs} - | systemSeconds msgTs < old -> - tryDeleteMsg_ q mq False >> loop (dc + 1) - _ -> pure dc diff --git a/src/Simplex/Messaging/Server/Prometheus.hs b/src/Simplex/Messaging/Server/Prometheus.hs index b0ba2ea792..41ad5517b4 100644 --- a/src/Simplex/Messaging/Server/Prometheus.hs +++ b/src/Simplex/Messaging/Server/Prometheus.hs @@ -592,15 +592,7 @@ prometheusMetrics sm rtm ts = \\n\ \# HELP simplex_smp_loaded_queues_ntf_count Total loaded ntf credential references (all ntf credentials for memory storage)\n\ \# TYPE simplex_smp_loaded_queues_ntf_count gauge\n\ - \simplex_smp_loaded_queues_ntf_count " <> mshow (loadedNotifierCount loadedCounts) <> "\n# loadedCounts.loadedNotifierCount\n\ - \\n\ - \# HELP simplex_smp_loaded_queues_queue_lock_count Total queue locks (0 for memory storage)\n\ - \# TYPE simplex_smp_loaded_queues_queue_lock_count gauge\n\ - \simplex_smp_loaded_queues_queue_lock_count " <> mshow (queueLockCount loadedCounts) <> "\n# loadedCounts.queueLockCount\n\ - \\n\ - \# HELP simplex_smp_loaded_queues_ntf_lock_count Total notifier locks (0 for memory storage)\n\ - \# TYPE simplex_smp_loaded_queues_ntf_lock_count gauge\n\ - \simplex_smp_loaded_queues_ntf_lock_count " <> mshow (notifierLockCount loadedCounts) <> "\n# loadedCounts.notifierLockCount\n" + \simplex_smp_loaded_queues_ntf_count " <> mshow (loadedNotifierCount loadedCounts) <> "\n# loadedCounts.loadedNotifierCount\n" showTimeBuckets :: Text -> IM.IntMap Int -> Text showTimeBuckets metric = T.concat . snd . mapAccumL accumBucket (0, 0) . IM.assocs diff --git a/src/Simplex/Messaging/Server/QueueStore/Postgres.hs b/src/Simplex/Messaging/Server/QueueStore/Postgres.hs index 2a879f8cbd..23b90b1002 100644 --- a/src/Simplex/Messaging/Server/QueueStore/Postgres.hs +++ b/src/Simplex/Messaging/Server/QueueStore/Postgres.hs @@ -24,9 +24,7 @@ module Simplex.Messaging.Server.QueueStore.Postgres batchInsertServices, batchInsertQueues, foldServiceRecs, - foldRcvServiceQueueRecs, foldQueueRecs, - foldRecentQueueRecs, handleDuplicate, rowToQueueRec, withLog_, @@ -43,13 +41,12 @@ import Control.Monad import Control.Monad.Except import Control.Monad.IO.Class import Control.Monad.Trans.Except -import Data.Bifunctor (first) import Data.ByteString.Builder (Builder) import qualified Data.ByteString.Builder as BB import Data.ByteString.Char8 (ByteString) import qualified Data.ByteString.Lazy as LB import Data.Bitraversable (bimapM) -import Data.Either (fromRight, lefts) +import Data.Either (fromRight) import Data.Functor (($>)) import Data.Int (Int64) import Data.List (foldl', intersperse, partition) @@ -91,7 +88,7 @@ import Simplex.Messaging.SystemTime import Simplex.Messaging.TMap (TMap) import qualified Simplex.Messaging.TMap as TM import Simplex.Messaging.Transport (SMPServiceRole (..)) -import Simplex.Messaging.Util (eitherToMaybe, firstRow, ifM, maybeFirstRow, maybeFirstRow', tshow, (<$$>), ($>>=)) +import Simplex.Messaging.Util (eitherToMaybe, firstRow, maybeFirstRow, maybeFirstRow', tshow, (<$$>), ($>>=)) import System.Exit (exitFailure) import System.IO (IOMode (..), hFlush, stdout) import UnliftIO.STM @@ -103,35 +100,19 @@ import Simplex.Messaging.Encoding.String data PostgresQueueStore q = PostgresQueueStore { dbStore :: DBStore, dbStoreLog :: Maybe (StoreLog 'WriteMode), - -- this map caches all created and opened queues - queues :: TMap RecipientId q, - -- this map only cashes the queues that were attempted to send messages to, - senders :: TMap SenderId RecipientId, - links :: TMap LinkId RecipientId, - -- this map only cashes the queues that were attempted to be subscribed to, - notifiers :: TMap NotifierId RecipientId, - notifierLocks :: TMap NotifierId Lock, serviceLocks :: TMap CertFingerprint Lock, - deletedTTL :: Int64, - useCache :: Bool + deletedTTL :: Int64 } -type UseQueueCache = Bool - instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where - type QueueStoreCfg (PostgresQueueStore q) = (PostgresStoreCfg, UseQueueCache) + type QueueStoreCfg (PostgresQueueStore q) = PostgresStoreCfg - newQueueStore :: (PostgresStoreCfg, UseQueueCache) -> IO (PostgresQueueStore q) - newQueueStore (PostgresStoreCfg {dbOpts, dbStoreLogPath, confirmMigrations, deletedTTL}, useCache) = do + newQueueStore :: PostgresStoreCfg -> IO (PostgresQueueStore q) + newQueueStore PostgresStoreCfg {dbOpts, dbStoreLogPath, confirmMigrations, deletedTTL} = do dbStore <- either err pure =<< createDBStore dbOpts serverMigrations (MigrationConfig confirmMigrations Nothing) dbStoreLog <- mapM (openWriteStoreLog True) dbStoreLogPath - queues <- TM.emptyIO - senders <- TM.emptyIO - links <- TM.emptyIO - notifiers <- TM.emptyIO - notifierLocks <- TM.emptyIO serviceLocks <- TM.emptyIO - pure PostgresQueueStore {dbStore, dbStoreLog, queues, senders, links, notifiers, notifierLocks, serviceLocks, deletedTTL, useCache} + pure PostgresQueueStore {dbStore, dbStoreLog, serviceLocks, deletedTTL} where err e = do logError $ "STORE: newQueueStore, error opening PostgreSQL database, " <> tshow e @@ -142,9 +123,6 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where closeDBStore dbStore mapM_ closeStoreLog dbStoreLog - loadedQueues = queues - {-# INLINE loadedQueues #-} - compactQueues :: PostgresQueueStore q -> IO Int64 compactQueues st@PostgresQueueStore {deletedTTL} = do old <- subtract deletedTTL . systemSeconds <$> liftIO getSystemTime @@ -170,131 +148,44 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where (SRMessaging, SRNotifier) pure EntityCounts {queueCount, notifierCount, rcvServiceCount, ntfServiceCount, rcvServiceQueuesCount, ntfServiceQueuesCount} - -- this implementation assumes that the lock is already taken by addQueue - -- and relies on unique constraints in the database to prevent duplicate IDs. + -- this implementation relies on unique constraints in the database to prevent duplicate IDs. addQueue_ :: PostgresQueueStore q -> (RecipientId -> QueueRec -> IO q) -> RecipientId -> QueueRec -> IO (Either ErrorType q) addQueue_ st mkQ rId qr = do sq <- mkQ rId qr - withQueueLock sq "addQueue_" $ E.uninterruptibleMask_ $ runExceptT $ do + E.uninterruptibleMask_ $ runExceptT $ do void $ withDB "addQueue_" st $ \db -> E.try (DB.execute db insertQueueQuery $ queueRecToRow (rId, qr)) >>= bimapM handleDuplicate pure - when useCache $ do - atomically $ TM.insert rId sq queues - atomically $ TM.insert (senderId qr) rId senders - forM_ (notifier qr) $ \NtfCreds {notifierId = nId} -> atomically $ TM.insert nId rId notifiers - forM_ (queueData qr) $ \(lnkId, _) -> atomically $ TM.insert lnkId rId links withLog "addStoreQueue" st $ \s -> logCreateQueue s rId qr pure sq + + getQueue_ :: QueueParty p => PostgresQueueStore q -> (RecipientId -> QueueRec -> IO q) -> SParty p -> QueueId -> IO (Either ErrorType q) + getQueue_ st mkQ party qId = case party of + SRecipient -> loadQueue " WHERE recipient_id = ?" + SSender -> loadQueue " WHERE sender_id = ?" + SSenderLink -> loadQueue " WHERE link_id = ?" + SNotifier -> loadQueue " WHERE notifier_id = ?" where - PostgresQueueStore {queues, senders, links, notifiers, useCache} = st - -- Not doing duplicate checks in maps as the probability of duplicates is very low. - -- It needs to be reconsidered when IDs are supplied by the users. - -- hasId = anyM [TM.memberIO rId queues, TM.memberIO senderId senders, hasNotifier] - -- hasNotifier = maybe (pure False) (\NtfCreds {notifierId} -> TM.memberIO notifierId notifiers) notifier - - getQueue_ :: QueueParty p => PostgresQueueStore q -> (Bool -> RecipientId -> QueueRec -> IO q) -> SParty p -> QueueId -> IO (Either ErrorType q) - getQueue_ st mkQ party qId - | useCache = case party of - SRecipient -> getRcvQueue qId - SSender -> TM.lookupIO qId senders >>= maybe (mask loadSndQueue) getRcvQueue - SSenderLink -> TM.lookupIO qId links >>= maybe (mask loadLinkQueue) getRcvQueue - -- loaded queue is deleted from notifiers map to reduce cache size after queue was subscribed to by ntf server - SNotifier -> TM.lookupIO qId notifiers >>= maybe (mask loadNtfQueue) (getRcvQueue >=> (atomically (TM.delete qId notifiers) $>)) - | otherwise = case party of - SRecipient -> loadQueueNoCache " WHERE recipient_id = ?" - SSender -> loadQueueNoCache " WHERE sender_id = ?" - SSenderLink -> loadQueueNoCache " WHERE link_id = ?" - SNotifier -> loadQueueNoCache " WHERE notifier_id = ?" - where - PostgresQueueStore {queues, senders, links, notifiers, useCache} = st - getRcvQueue rId = TM.lookupIO rId queues >>= maybe (mask loadRcvQueue) (pure . Right) - loadRcvQueue = do - (rId, qRec) <- loadQueue " WHERE recipient_id = ?" - liftIO $ cacheQueue rId qRec $ \_ -> pure () -- recipient map already checked, not caching sender ref - loadSndQueue = loadSndQueue_ " WHERE sender_id = ?" - loadLinkQueue = loadSndQueue_ " WHERE link_id = ?" - loadNtfQueue = do - (rId, qRec) <- loadQueue " WHERE notifier_id = ?" - liftIO $ - TM.lookupIO rId queues -- checking recipient map first, not creating lock in map, not caching queue - >>= maybe (mkQ False rId qRec) pure - loadSndQueue_ condition = do - (rId, qRec) <- loadQueue condition - liftIO $ - TM.lookupIO rId queues -- checking recipient map first - >>= maybe (cacheQueue rId qRec cacheSender) (atomically (cacheSender rId) $>) - loadQueueNoCache cond = mask $ loadQueue cond >>= liftIO . uncurry (mkQ True) + loadQueue cond = mask $ loadQueueRec cond >>= liftIO . uncurry mkQ mask = E.uninterruptibleMask_ . runExceptT - cacheSender rId = TM.insert qId rId senders - loadQueue condition = + loadQueueRec condition = withDB "getQueue_" st $ \db -> firstRow rowToQueueRec AUTH $ DB.query db (queueRecQuery <> condition <> " AND deleted_at IS NULL") (Only qId) - cacheQueue rId qRec insertRef = do - sq <- mkQ True rId qRec -- loaded queue - -- This lock prevents the scenario when the queue is added to cache, - -- while another thread is proccessing the same queue in withAllMsgQueues - -- without adding it to cache, possibly trying to open the same files twice. - -- Alse see comment in idleDeleteExpiredMsgs. - withQueueLock sq "getQueue_" $ atomically $ - -- checking the cache again for concurrent reads, - -- use previously loaded queue if exists. - TM.lookup rId queues >>= \case - Just sq' -> pure sq' - Nothing -> do - insertRef rId - TM.insert rId sq queues - pure sq - - getQueues_ :: forall p. BatchParty p => PostgresQueueStore q -> (Bool -> RecipientId -> QueueRec -> IO q) -> SParty p -> [QueueId] -> IO [Either ErrorType q] + + getQueues_ :: forall p. BatchParty p => PostgresQueueStore q -> (RecipientId -> QueueRec -> IO q) -> SParty p -> [QueueId] -> IO [Either ErrorType q] getQueues_ st mkQ party qIds | null qIds = pure [] - | useCache = case party of - SRecipient -> do - qs <- readTVarIO queues - let qs' = map (\qId -> get qs qId qId) qIds - E.uninterruptibleMask_ $ loadQueues qs' " WHERE recipient_id IN ?" cacheRcvQueue - SNotifier -> do - ns <- readTVarIO notifiers - qs <- readTVarIO queues - let qs' = map (\qId -> get ns qId qId >>= get qs qId) qIds - E.uninterruptibleMask_ $ loadQueues qs' " WHERE notifier_id IN ?" $ \(rId, qRec) -> - forM (notifier qRec) $ \NtfCreds {notifierId = nId} -> -- it is always Just with this query - (nId,) <$> maybe (mkQ False rId qRec) pure (M.lookup rId qs) | otherwise = E.uninterruptibleMask_ $ case party of - SRecipient -> loadQueuesNoCache " WHERE recipient_id IN ?" $ \(rId, qRec) -> - Just . (rId,) <$> mkQ False rId qRec - SNotifier -> loadQueuesNoCache " WHERE notifier_id IN ?" $ \(rId, qRec) -> - forM (notifier qRec) $ \NtfCreds {notifierId = nId} -> (nId,) <$> mkQ False rId qRec + SRecipient -> loadQueues " WHERE recipient_id IN ?" $ \(rId, qRec) -> + Just . (rId,) <$> mkQ rId qRec + SNotifier -> loadQueues " WHERE notifier_id IN ?" $ \(rId, qRec) -> + forM (notifier qRec) $ \NtfCreds {notifierId = nId} -> (nId,) <$> mkQ rId qRec where - PostgresQueueStore {queues, notifiers, useCache} = st - get :: M.Map QueueId a -> QueueId -> QueueId -> Either QueueId a - get m qId = maybe (Left qId) Right . (`M.lookup` m) - loadQueues :: [Either QueueId q] -> Query -> ((RecipientId, QueueRec) -> IO (Maybe (QueueId, q))) -> IO [Either ErrorType q] - loadQueues qs' cond mkCacheQueue = do - let qIds' = lefts qs' - if null qIds' - then pure $ map (first (const INTERNAL)) qs' - else do - qs_ <- dbLoadQueues qIds' cond mkCacheQueue - pure $ map (result qs_) qs' - where - result :: Either ErrorType (M.Map QueueId q) -> Either QueueId q -> Either ErrorType q - result _ (Right q) = Right q - result qs_ (Left qId) = maybe (Left AUTH) Right . M.lookup qId =<< qs_ dbLoadQueues qIds' cond mkQueue' = runExceptT $ fmap M.fromList $ withDB' "getQueues_" st (\db -> DB.query db (queueRecQuery <> cond <> " AND deleted_at IS NULL") (Only (In qIds'))) >>= liftIO . fmap catMaybes . mapM (mkQueue' . rowToQueueRec) - cacheRcvQueue (rId, qRec) = do - sq <- mkQ True rId qRec - sq' <- withQueueLock sq "getQueue_" $ atomically $ - -- checking the cache again for concurrent reads, use previously loaded queue if exists. - TM.lookup rId queues >>= \case - Just sq' -> pure sq' - Nothing -> sq <$ TM.insert rId sq queues - pure $ Just (rId, sq') - loadQueuesNoCache cond mkQueue' = do + loadQueues cond mkQueue' = do qs_ <- dbLoadQueues qIds cond mkQueue' pure $ map (result qs_) qIds where @@ -312,7 +203,7 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where addQueueLinkData :: PostgresQueueStore q -> q -> LinkId -> QueueLinkData -> IO (Either ErrorType ()) addQueueLinkData st sq lnkId d = - withQueueRec sq "addQueueLinkData" $ \q -> case queueData q of + withQueueRec sq $ \q -> case queueData q of Nothing -> addLink q $ \db -> DB.execute db qry (d :. (lnkId, rId, QMContact)) Just (lnkId', _) | lnkId' == lnkId -> @@ -329,7 +220,7 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where deleteQueueLinkData :: PostgresQueueStore q -> q -> IO (Either ErrorType ()) deleteQueueLinkData st sq = - withQueueRec sq "deleteQueueLinkData" $ \q -> case queueData q of + withQueueRec sq $ \q -> case queueData q of Just _ -> do assertUpdated $ withDB' "deleteQueueLinkData" st $ \db -> DB.execute db "UPDATE msg_queues SET link_id = NULL, fixed_data = NULL, user_data = NULL WHERE recipient_id = ? AND deleted_at IS NULL" (Only rId) @@ -341,7 +232,7 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where secureQueue :: PostgresQueueStore q -> q -> SndPublicAuthKey -> IO (Either ErrorType ()) secureQueue st sq sKey = - withQueueRec sq "secureQueue" $ \q -> do + withQueueRec sq $ \q -> do verify q assertUpdated $ withDB' "secureQueue" st $ \db -> DB.execute db "UPDATE msg_queues SET sender_key = ? WHERE recipient_id = ? AND deleted_at IS NULL AND (sender_key IS NULL OR sender_key = ?)" (sKey, rId, sKey) @@ -355,7 +246,7 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where updateKeys :: PostgresQueueStore q -> q -> NonEmpty RcvPublicAuthKey -> IO (Either ErrorType ()) updateKeys st sq rKeys = - withQueueRec sq "updateKeys" $ \q -> do + withQueueRec sq $ \q -> do assertUpdated $ withDB' "updateKeys" st $ \db -> DB.execute db "UPDATE msg_queues SET recipient_keys = ? WHERE recipient_id = ? AND deleted_at IS NULL" (rKeys, rId) atomically $ writeTVar (queueRec sq) $ Just q {recipientKeys = rKeys} @@ -365,24 +256,14 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where addQueueNotifier :: PostgresQueueStore q -> q -> NtfCreds -> IO (Either ErrorType (Maybe NtfCreds)) addQueueNotifier st sq ntfCreds@NtfCreds {notifierId = nId, notifierKey, rcvNtfDhSecret} = - withQueueRec sq "addQueueNotifier" $ \q -> - checkCachedNotifier $ do - assertUpdated $ withDB "addQueueNotifier" st $ \db -> - E.try (update db) >>= bimapM handleDuplicate pure - nc_ <- forM (notifier q) $ \nc@NtfCreds {notifierId} -> atomically (TM.delete notifierId notifiers) $> nc - let !q' = q {notifier = Just ntfCreds} - atomically $ writeTVar (queueRec sq) $ Just q' - when useCache $ do - atomically $ TM.insert nId rId notifiers - withLog "addQueueNotifier" st $ \s -> logAddNotifier s rId ntfCreds - pure nc_ + withQueueRec sq $ \q -> do + assertUpdated $ withDB "addQueueNotifier" st $ \db -> + E.try (update db) >>= bimapM handleDuplicate pure + let !q' = q {notifier = Just ntfCreds} + atomically $ writeTVar (queueRec sq) $ Just q' + withLog "addQueueNotifier" st $ \s -> logAddNotifier s rId ntfCreds + pure $ notifier q where - checkCachedNotifier add - | useCache = - ExceptT $ withLockMap (notifierLocks st) nId "addQueueNotifier" $ - ifM (TM.memberIO nId notifiers) (pure $ Left DUPLICATE_) $ runExceptT add - | otherwise = add - PostgresQueueStore {notifiers, useCache} = st rId = recipientId sq update db = DB.execute @@ -396,18 +277,13 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where deleteQueueNotifier :: PostgresQueueStore q -> q -> IO (Either ErrorType (Maybe NtfCreds)) deleteQueueNotifier st sq = - withQueueRec sq "deleteQueueNotifier" $ \q -> - ExceptT $ fmap sequence $ forM (notifier q) $ \nc@NtfCreds {notifierId = nId} -> - withNotifierLock nId $ runExceptT $ do - assertUpdated $ withDB' "deleteQueueNotifier" st update - when (useCache st) $ atomically $ TM.delete nId $ notifiers st - atomically $ writeTVar (queueRec sq) $ Just q {notifier = Nothing} - withLog "deleteQueueNotifier" st (`logDeleteNotifier` rId) - pure nc + withQueueRec sq $ \q -> + forM (notifier q) $ \nc -> do + assertUpdated $ withDB' "deleteQueueNotifier" st update + atomically $ writeTVar (queueRec sq) $ Just q {notifier = Nothing} + withLog "deleteQueueNotifier" st (`logDeleteNotifier` rId) + pure nc where - withNotifierLock nId - | useCache st = withLockMap (notifierLocks st) nId "deleteQueueNotifier" - | otherwise = id rId = recipientId sq update db = DB.execute @@ -436,7 +312,7 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where updateQueueTime :: PostgresQueueStore q -> q -> SystemDate -> IO (Either ErrorType QueueRec) updateQueueTime st sq t = - withQueueRec sq "updateQueueTime" $ \q@QueueRec {updatedAt} -> + withQueueRec sq $ \q@QueueRec {updatedAt} -> if updatedAt == Just t then pure q else do @@ -449,7 +325,6 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where where rId = recipientId sq - -- this method is called from JournalMsgStore deleteQueue that already locks the queue deleteStoreQueue :: PostgresQueueStore q -> q -> IO (Either ErrorType QueueRec) deleteStoreQueue st sq = E.uninterruptibleMask_ $ runExceptT $ do q <- ExceptT $ readQueueRecIO qr @@ -457,11 +332,6 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where assertUpdated $ withDB' "deleteStoreQueue" st $ \db -> DB.execute db "UPDATE msg_queues SET deleted_at = ? WHERE recipient_id = ? AND deleted_at IS NULL" (ts, rId) atomically $ writeTVar qr Nothing - when (useCache st) $ do - atomically $ TM.delete (senderId q) $ senders st - forM_ (notifier q) $ \NtfCreds {notifierId} -> do - atomically $ TM.delete notifierId $ notifiers st - atomically $ TM.delete notifierId $ notifierLocks st withLog "deleteStoreQueue" st (`logDeleteQueue` rId) pure q where @@ -484,7 +354,7 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where pure serviceId setQueueService :: (PartyI p, ServiceParty p) => PostgresQueueStore q -> q -> SParty p -> Maybe ServiceId -> IO (Either ErrorType ()) - setQueueService st sq party serviceId = withQueueRec sq "setQueueService" $ \q -> case party of + setQueueService st sq party serviceId = withQueueRec sq $ \q -> case party of SRecipientService | rcvServiceId q == serviceId -> pure () | otherwise -> do @@ -609,11 +479,6 @@ foldServiceRecs st f = DB.fold_ db "SELECT service_id, service_role, service_cert, service_cert_hash, created_at FROM services" mempty $ \ !acc -> fmap (acc <>) . f . rowToServiceRec -foldRcvServiceQueueRecs :: PostgresQueueStore q -> ServiceId -> (a -> (RecipientId, QueueRec) -> IO a) -> a -> IO (Either ErrorType a) -foldRcvServiceQueueRecs st serviceId f acc = - runExceptT $ withDB' "foldRcvServiceQueueRecs" st $ \db -> - DB.fold db (queueRecQuery <> " WHERE rcv_service_id = ? AND deleted_at IS NULL") (Only serviceId) acc $ \a -> f a . rowToQueueRec - foldQueueRecs :: Monoid a => Bool -> Bool -> PostgresQueueStore q -> ((RecipientId, QueueRec) -> IO a) -> IO a foldQueueRecs withData = foldQueueRecs_ foldRecs where @@ -622,12 +487,6 @@ foldQueueRecs withData = foldQueueRecs_ foldRecs | otherwise = DB.fold_ db (queueRecQuery <> cond) acc $ \acc' -> f' acc' . rowToQueueRec cond = " WHERE deleted_at IS NULL ORDER BY recipient_id ASC" -foldRecentQueueRecs :: Monoid a => Int64 -> Bool -> PostgresQueueStore q -> ((RecipientId, QueueRec) -> IO a) -> IO a -foldRecentQueueRecs old = foldQueueRecs_ foldRecs - where - foldRecs db acc f' = DB.fold db (queueRecQuery <> cond) (Only old) acc $ \acc' -> f' acc' . rowToQueueRec - cond = " WHERE deleted_at IS NULL AND updated_at > ? ORDER BY recipient_id ASC" - foldQueueRecs_ :: Monoid a => (DB.Connection -> (Int, a) -> ((Int, a) -> (RecipientId, QueueRec) -> IO (Int, a)) -> IO (Int, a)) -> @@ -751,15 +610,15 @@ rowToServiceRec (serviceId, serviceRole, serviceCert, Binary fp, serviceCreatedA setStatusDB :: StoreQueueClass q => Text -> PostgresQueueStore q -> q -> ServerEntityStatus -> ExceptT ErrorType IO () -> IO (Either ErrorType ()) setStatusDB op st sq status writeLog = - withQueueRec sq op $ \q -> do + withQueueRec sq $ \q -> do assertUpdated $ withDB' op st $ \db -> DB.execute db "UPDATE msg_queues SET status = ? WHERE recipient_id = ? AND deleted_at IS NULL" (status, recipientId sq) atomically $ writeTVar (queueRec sq) $ Just q {status} writeLog -withQueueRec :: StoreQueueClass q => q -> Text -> (QueueRec -> ExceptT ErrorType IO a) -> IO (Either ErrorType a) -withQueueRec sq op action = - withQueueLock sq op $ E.uninterruptibleMask_ $ runExceptT $ ExceptT (readQueueRecIO $ queueRec sq) >>= action +withQueueRec :: StoreQueueClass q => q -> (QueueRec -> ExceptT ErrorType IO a) -> IO (Either ErrorType a) +withQueueRec sq action = + E.uninterruptibleMask_ $ runExceptT $ ExceptT (readQueueRecIO $ queueRec sq) >>= action assertUpdated :: ExceptT ErrorType IO Int64 -> ExceptT ErrorType IO () assertUpdated = (>>= \n -> when (n == 0) (throwE AUTH)) diff --git a/src/Simplex/Messaging/Server/QueueStore/STM.hs b/src/Simplex/Messaging/Server/QueueStore/STM.hs index 6e1a948024..1be3898219 100644 --- a/src/Simplex/Messaging/Server/QueueStore/STM.hs +++ b/src/Simplex/Messaging/Server/QueueStore/STM.hs @@ -18,6 +18,7 @@ module Simplex.Messaging.Server.QueueStore.STM ( STMQueueStore (..), STMService (..), foldRcvServiceQueues, + withLoadedQueues, setStoreLog, withLog', readQueueRecIO, @@ -91,8 +92,6 @@ instance StoreQueueClass q => QueueStoreClass q (STMQueueStore q) where atomically $ TM.clear senders atomically $ TM.clear notifiers - loadedQueues = queues - {-# INLINE loadedQueues #-} compactQueues _ = pure 0 {-# INLINE compactQueues #-} @@ -134,7 +133,7 @@ instance StoreQueueClass q => QueueStoreClass q (STMQueueStore q) where hasNotifier = maybe (pure False) (\NtfCreds {notifierId} -> TM.member notifierId notifiers) notifier hasLink = maybe (pure False) (\(lnkId, _) -> TM.member lnkId links) queueData - getQueue_ :: QueueParty p => STMQueueStore q -> (Bool -> RecipientId -> QueueRec -> IO q) -> SParty p -> QueueId -> IO (Either ErrorType q) + getQueue_ :: QueueParty p => STMQueueStore q -> (RecipientId -> QueueRec -> IO q) -> SParty p -> QueueId -> IO (Either ErrorType q) getQueue_ st _ party qId = maybe (Left AUTH) Right <$> case party of SRecipient -> TM.lookupIO qId queues @@ -144,7 +143,7 @@ instance StoreQueueClass q => QueueStoreClass q (STMQueueStore q) where where STMQueueStore {queues, senders, notifiers, links} = st - getQueues_ :: BatchParty p => STMQueueStore q -> (Bool -> RecipientId -> QueueRec -> IO q) -> SParty p -> [QueueId] -> IO [Either ErrorType q] + getQueues_ :: BatchParty p => STMQueueStore q -> (RecipientId -> QueueRec -> IO q) -> SParty p -> [QueueId] -> IO [Either ErrorType q] getQueues_ st _ party qIds = case party of SRecipient -> do qs <- readTVarIO queues @@ -379,6 +378,11 @@ foldRcvServiceQueues st serviceId f acc = where get rId = TM.lookupIO rId (queues st) $>>= \q -> (q,) <$$> readTVarIO (queueRec q) +withLoadedQueues :: Monoid a => STMQueueStore q -> (q -> IO a) -> IO a +withLoadedQueues st f = readTVarIO (queues st) >>= foldM run mempty + where + run !acc = fmap (acc <>) . f + withQueueRec :: TVar (Maybe QueueRec) -> (QueueRec -> STM a) -> IO (Either ErrorType a) withQueueRec qr a = atomically $ readQueueRec qr >>= mapM a diff --git a/src/Simplex/Messaging/Server/QueueStore/Types.hs b/src/Simplex/Messaging/Server/QueueStore/Types.hs index 415a5f33cb..e8d47751b9 100644 --- a/src/Simplex/Messaging/Server/QueueStore/Types.hs +++ b/src/Simplex/Messaging/Server/QueueStore/Types.hs @@ -1,5 +1,4 @@ {-# LANGUAGE AllowAmbiguousTypes #-} -{-# LANGUAGE BangPatterns #-} {-# LANGUAGE DataKinds #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE TypeFamilies #-} @@ -9,35 +8,29 @@ module Simplex.Messaging.Server.QueueStore.Types ( StoreQueueClass (..), QueueStoreClass (..), EntityCounts (..), - withLoadedQueues, ) where import Control.Concurrent.STM -import Control.Monad import Data.Int (Int64) import Data.List.NonEmpty (NonEmpty) import Data.Map.Strict (Map) -import Data.Text (Text) import Simplex.Messaging.Protocol import Simplex.Messaging.Server.QueueStore import Simplex.Messaging.SystemTime -import Simplex.Messaging.TMap (TMap) class StoreQueueClass q where recipientId :: q -> RecipientId queueRec :: q -> TVar (Maybe QueueRec) - withQueueLock :: q -> Text -> IO a -> IO a class StoreQueueClass q => QueueStoreClass q s where type QueueStoreCfg s newQueueStore :: QueueStoreCfg s -> IO s closeQueueStore :: s -> IO () getEntityCounts :: s -> IO EntityCounts - loadedQueues :: s -> TMap RecipientId q compactQueues :: s -> IO Int64 addQueue_ :: s -> (RecipientId -> QueueRec -> IO q) -> RecipientId -> QueueRec -> IO (Either ErrorType q) - getQueue_ :: QueueParty p => s -> (Bool -> RecipientId -> QueueRec -> IO q) -> SParty p -> QueueId -> IO (Either ErrorType q) - getQueues_ :: BatchParty p => s -> (Bool -> RecipientId -> QueueRec -> IO q) -> SParty p -> [QueueId] -> IO [Either ErrorType q] + getQueue_ :: QueueParty p => s -> (RecipientId -> QueueRec -> IO q) -> SParty p -> QueueId -> IO (Either ErrorType q) + getQueues_ :: BatchParty p => s -> (RecipientId -> QueueRec -> IO q) -> SParty p -> [QueueId] -> IO [Either ErrorType q] getQueueLinkData :: s -> q -> LinkId -> IO (Either ErrorType QueueLinkData) addQueueLinkData :: s -> q -> LinkId -> QueueLinkData -> IO (Either ErrorType ()) deleteQueueLinkData :: s -> q -> IO (Either ErrorType ()) @@ -64,8 +57,3 @@ data EntityCounts = EntityCounts rcvServiceQueuesCount :: Int, ntfServiceQueuesCount :: Int } - -withLoadedQueues :: (Monoid a, QueueStoreClass q s) => s -> (q -> IO a) -> IO a -withLoadedQueues st f = readTVarIO (loadedQueues st) >>= foldM run mempty - where - run !acc = fmap (acc <>) . f diff --git a/src/Simplex/Messaging/Server/StoreLog/ReadWrite.hs b/src/Simplex/Messaging/Server/StoreLog/ReadWrite.hs index 6bf327f70d..fb741d575f 100644 --- a/src/Simplex/Messaging/Server/StoreLog/ReadWrite.hs +++ b/src/Simplex/Messaging/Server/StoreLog/ReadWrite.hs @@ -23,7 +23,7 @@ import Data.Text.Encoding (decodeLatin1) import Simplex.Messaging.Encoding.String import Simplex.Messaging.Protocol (ASubscriberParty (..), ErrorType, RecipientId, SParty (..)) import Simplex.Messaging.Server.QueueStore (QueueRec, ServiceRec (..)) -import Simplex.Messaging.Server.QueueStore.STM (STMQueueStore (..), STMService (..)) +import Simplex.Messaging.Server.QueueStore.STM (STMQueueStore (..), STMService (..), withLoadedQueues) import Simplex.Messaging.Server.QueueStore.Types import Simplex.Messaging.Server.StoreLog import Simplex.Messaging.Util (tshow) @@ -71,7 +71,7 @@ readQueueStore tty mkQ f st = readLogLines tty f $ \_ -> processLine withQueue qId op a = runExceptT go >>= qError qId op where go = do - q <- ExceptT $ getQueue_ st (\_ -> mkQ) SRecipient qId + q <- ExceptT $ getQueue_ st mkQ SRecipient qId liftIO (readTVarIO $ queueRec q) >>= \case Nothing -> logWarn $ logPfx qId op <> "already deleted" Just _ -> void $ ExceptT $ a q diff --git a/tests/AgentTests/FunctionalAPITests.hs b/tests/AgentTests/FunctionalAPITests.hs index 028de1ab3b..65dcffadb1 100644 --- a/tests/AgentTests/FunctionalAPITests.hs +++ b/tests/AgentTests/FunctionalAPITests.hs @@ -2035,7 +2035,7 @@ testOldContactQueueShortLink ps@(_, msType) = withAgentClients2 $ \a b -> do ASSCfg _ _ (SSCMemory sp_) -> mapM_ (\StorePaths {storeLogFile} -> updateStoreLog storeLogFile) sp_ #if defined(dbServerPostgres) ASSCfg _ _ (SSCDatabase storeCfg) -> do - st :: PostgresQueueStore PostgresQueue <- newQueueStore @PostgresQueue (storeCfg, False) + st :: PostgresQueueStore PostgresQueue <- newQueueStore @PostgresQueue storeCfg updateDbStore st closeQueueStore @PostgresQueue st #endif diff --git a/tests/CoreTests/MsgStoreTests.hs b/tests/CoreTests/MsgStoreTests.hs index 558fbf96a9..0cfcc8c495 100644 --- a/tests/CoreTests/MsgStoreTests.hs +++ b/tests/CoreTests/MsgStoreTests.hs @@ -9,9 +9,7 @@ {-# LANGUAGE PatternSynonyms #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE StandaloneDeriving #-} {-# LANGUAGE TypeApplications #-} -{-# OPTIONS_GHC -Wno-orphans #-} {-# OPTIONS_GHC -fno-warn-ambiguous-fields #-} module CoreTests.MsgStoreTests where @@ -65,7 +63,6 @@ msgStoreTests = do someMsgStoreTests cfg = do it "should get queue and store/read messages" $ withMsgStore cfg testGetQueue it "should write/ack messages" $ withMsgStore cfg testWriteAckMessages - it "should not fail on EOF when changing read journal" $ withMsgStore cfg testChangeReadJournal it "should not add link data to secured messaging queue" $ withMsgStore cfg testLinkDataSecuredQueue -- TODO constrain to STM stores? @@ -193,25 +190,6 @@ testWriteAckMessages ms = do void $ ExceptT $ deleteQueue ms q1 void $ ExceptT $ deleteQueue ms q2 -testChangeReadJournal :: MsgStoreClass s => s -> IO () -testChangeReadJournal ms = do - g <- C.newRandom - (rId, qr) <- testNewQueueRec g QMMessaging - runRight_ $ do - q <- ExceptT $ addQueue ms rId qr - let write s = writeMsg ms q True =<< mkMessage s - Just (Message {msgId = mId1}, True) <- write "message 1" - (Msg "message 1", Nothing) <- tryDelPeekMsg ms q mId1 - Just (Message {msgId = mId2}, True) <- write "message 2" - (Msg "message 2", Nothing) <- tryDelPeekMsg ms q mId2 - Just (Message {msgId = mId3}, True) <- write "message 3" - (Msg "message 3", Nothing) <- tryDelPeekMsg ms q mId3 - Just (Message {msgId = mId4}, True) <- write "message 4" - (Msg "message 4", Nothing) <- tryDelPeekMsg ms q mId4 - Just (Message {msgId = mId5}, True) <- write "message 5" - (Msg "message 5", Nothing) <- tryDelPeekMsg ms q mId5 - void $ ExceptT $ deleteQueue ms q - testLinkDataSecuredQueue :: MsgStoreClass s => s -> IO () testLinkDataSecuredQueue ms = do g <- C.newRandom diff --git a/tests/CoreTests/StoreLogTests.hs b/tests/CoreTests/StoreLogTests.hs index fc47ddca85..e0b8e16fc3 100644 --- a/tests/CoreTests/StoreLogTests.hs +++ b/tests/CoreTests/StoreLogTests.hs @@ -189,7 +189,7 @@ testSMPStoreLog testSuite tests = where testReadWrite SLTC {compacted, state} = do st <- newMsgStore testSMTStoreConfig - l <- readWriteQueueStore True (mkQueue st True) testStoreLogFile $ queueStore st + l <- readWriteQueueStore True (mkQueue st) testStoreLogFile $ queueStore st storeState st `shouldReturn` state closeStoreLog l ([], compacted') <- partitionEithers . map strDecode . B.lines <$> B.readFile testStoreLogFile diff --git a/tests/SMPProxyTests.hs b/tests/SMPProxyTests.hs index e7fabf09b2..467949878d 100644 --- a/tests/SMPProxyTests.hs +++ b/tests/SMPProxyTests.hs @@ -406,12 +406,8 @@ agentViaProxyRetryOffline = do where withServer :: (ThreadId -> IO a) -> IO a withServer = withServer_ testStoreLogFile testStoreMsgsFile testStoreNtfsFile testPort - -- TODO [postgres] - -- withServer = withServer_ testStoreDBOpts testStoreMsgsDir testStoreNtfsFile testPort withServer2 :: (ThreadId -> IO a) -> IO a withServer2 = withServer_ testStoreLogFile2 testStoreMsgsFile2 testStoreNtfsFile2 testPort2 - -- TODO [postgres] - -- withServer2 = withServer_ testStoreDBOpts2 testStoreMsgsDir2 testStoreNtfsFile2 testPort2 withServer_ storeLog storeMsgs storeNtfs = let cfg' = updateCfg (memoryCfg proxyCfg storeLog storeMsgs) $ \cfg_ -> cfg_ {storeNtfsFile = Just storeNtfs} in withSmpServerConfigOn (transport @TLS) cfg' diff --git a/tests/ServerTests.hs b/tests/ServerTests.hs index 88554fcffe..55fca1122d 100644 --- a/tests/ServerTests.hs +++ b/tests/ServerTests.hs @@ -1226,7 +1226,7 @@ testRestoreExpireMessages = closeMsgStore ms' readWriteQueues = do ms <- newMsgStore testSMTStoreConfig - readWriteQueueStore True (mkQueue ms True) testStoreLogFile (queueStore ms) >>= closeStoreLog + readWriteQueueStore True (mkQueue ms) testStoreLogFile (queueStore ms) >>= closeStoreLog removeFileIfExists testStoreMsgsFile pure ms #else