mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-10-23 10:52:28 +00:00
smp server: updated journal store API (#1374)
* core: updated journal store API * parameterize JournalState by read/write type * collect stats when importing/expiring * compare stored stats and store for messages and notifications * simplify
This commit is contained in:
Evgeny authored and GitHub committed
1 parent
0414ea59f0
commit
dc17402896
7 files changed
+292
-204
No files matched your search
+101
-53
@@ -32,11 +32,13 @@ module Simplex.Messaging.Server
|
||||
runSMPServerBlocking,
|
||||
importMessages,
|
||||
exportMessages,
|
||||
printMessageStats,
|
||||
disconnectTransport,
|
||||
verifyCmdAuthorization,
|
||||
dummyVerifyCmd,
|
||||
randomId,
|
||||
AttachHTTP,
|
||||
MessageStats (..),
|
||||
)
|
||||
where
|
||||
|
||||
@@ -91,6 +93,7 @@ import Simplex.Messaging.Server.Control
|
||||
import Simplex.Messaging.Server.Env.STM as Env
|
||||
import Simplex.Messaging.Server.Expiration
|
||||
import Simplex.Messaging.Server.MsgStore
|
||||
import Simplex.Messaging.Server.MsgStore.Journal (JournalMsgQueue (..), closeMsgQueue)
|
||||
import Simplex.Messaging.Server.MsgStore.STM
|
||||
import Simplex.Messaging.Server.MsgStore.Types
|
||||
import Simplex.Messaging.Server.NtfStore
|
||||
@@ -139,13 +142,24 @@ runSMPServerBlocking started cfg attachHTTP_ = newEnv cfg >>= runReaderT (smpSer
|
||||
type M a = ReaderT Env IO a
|
||||
type AttachHTTP = Socket -> TLS.Context -> IO ()
|
||||
|
||||
data MessageStats = MessageStats
|
||||
{ storedMsgsCount :: Int,
|
||||
expiredMsgsCount :: Int,
|
||||
storedQueues :: Int
|
||||
}
|
||||
|
||||
newMessageStats :: MessageStats
|
||||
newMessageStats = MessageStats 0 0 0
|
||||
|
||||
smpServer :: TMVar Bool -> ServerConfig -> Maybe AttachHTTP -> M ()
|
||||
smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHTTP_ = do
|
||||
s <- asks server
|
||||
pa <- asks proxyAgent
|
||||
expiredMsgs <- restoreServerMessages
|
||||
expiredNtfs <- restoreServerNtfs
|
||||
restoreServerStats expiredMsgs expiredNtfs
|
||||
msgStats <- restoreServerMessages
|
||||
ntfStats <- restoreServerNtfs
|
||||
liftIO $ printMessageStats "messages" msgStats
|
||||
liftIO $ printMessageStats "notifications" ntfStats
|
||||
restoreServerStats msgStats ntfStats
|
||||
raceAny_
|
||||
( serverThread s "server subscribedQ" subscribedQ subscribers subClients pendingSubEvents subscriptions cancelSub
|
||||
: serverThread s "server ntfSubscribedQ" ntfSubscribedQ Env.notifiers ntfSubClients pendingNtfSubEvents ntfSubscriptions (\_ -> pure ())
|
||||
@@ -366,9 +380,11 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
|
||||
liftIO $ forever $ do
|
||||
threadDelay' interval
|
||||
old <- expireBeforeEpoch expCfg
|
||||
void $ withActiveMsgQueues ms $ \_ q -> do
|
||||
void $ withActiveMsgQueues ms (\_ -> expireQueueMsgs stats old) 0
|
||||
where
|
||||
expireQueueMsgs stats old q acc =
|
||||
runExceptT (deleteExpiredMsgs q old) >>= \case
|
||||
Right deleted -> deleted <$ atomicModifyIORef'_ (msgExpired stats) (+ deleted)
|
||||
Right deleted -> (acc + deleted) <$ atomicModifyIORef'_ (msgExpired stats) (+ deleted)
|
||||
Left _ -> pure 0
|
||||
|
||||
expireNtfsThread :: ServerConfig -> M ()
|
||||
@@ -1435,7 +1451,7 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
|
||||
q <- getMsgQueue ms $ recipientId qr
|
||||
expireMessages q messageExpiration stats
|
||||
msg <- liftIO $ mkMessage msgId body
|
||||
writeMsg q True msg
|
||||
writeMsg ms q True msg
|
||||
case msg_ of
|
||||
Left e -> pure $ err e
|
||||
Right Nothing -> do
|
||||
@@ -1722,21 +1738,47 @@ saveServerMessages drainMsgs =
|
||||
exportMessages :: MsgStoreClass s => s -> FilePath -> (MsgQueue s -> IO [Message]) -> IO ()
|
||||
exportMessages ms f getMessages = do
|
||||
logInfo $ "saving messages to file " <> T.pack f
|
||||
total <- liftIO $ withFile f WriteMode $ withAllMsgQueues ms . saveQueueMsgs
|
||||
total <- liftIO $ withFile f WriteMode $ \h -> withAllMsgQueues ms (saveQueueMsgs h) 0
|
||||
logInfo $ "messages saved: " <> tshow total
|
||||
where
|
||||
saveQueueMsgs h rId q = getMessages q >>= \msgs -> length msgs <$ BLD.hPutBuilder h (encodeMessages rId msgs)
|
||||
saveQueueMsgs h rId q acc = getMessages q >>= \msgs -> (acc + length msgs) <$ BLD.hPutBuilder h (encodeMessages rId msgs)
|
||||
encodeMessages rId = mconcat . map (\msg -> BLD.byteString (strEncode $ MLRv3 rId msg) <> BLD.char8 '\n')
|
||||
|
||||
restoreServerMessages :: M Int
|
||||
restoreServerMessages =
|
||||
asks msgStore >>= \case
|
||||
AMS SMSMemory ms@STMMsgStore {storeConfig = STMStoreConfig {storePath = Just f}} -> do
|
||||
old_ <- asks (messageExpiration . config) $>>= (liftIO . fmap Just . expireBeforeEpoch)
|
||||
liftIO $ ifM (doesFileExist f) (importMessages ms f old_) (pure 0)
|
||||
_ -> logInfo "using journal message storage" $> 0
|
||||
restoreServerMessages :: M MessageStats
|
||||
restoreServerMessages = do
|
||||
old_ <- asks (messageExpiration . config) $>>= (liftIO . fmap Just . expireBeforeEpoch)
|
||||
asks msgStore >>= liftIO . processMessages old_
|
||||
where
|
||||
processMessages :: Maybe Int64 -> AMsgStore -> IO MessageStats
|
||||
processMessages old_ = \case
|
||||
AMS SMSMemory ms@STMMsgStore {storeConfig = STMStoreConfig {storePath}} -> case storePath of
|
||||
Just f -> ifM (doesFileExist f) (importMessages ms f old_) (pure newMessageStats)
|
||||
Nothing -> pure newMessageStats
|
||||
AMS SMSJournal ms -> case old_ of
|
||||
Just old -> do
|
||||
logInfo "expiring journal store messages..."
|
||||
(storedMsgsCount, expiredMsgsCount, storedQueues) <- withAllMsgQueues ms (\_ -> processExpireQueue old) (0, 0, 0)
|
||||
pure MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues}
|
||||
Nothing -> do
|
||||
logInfo "validating journal store messages..."
|
||||
(storedMsgsCount, storedQueues) <- withAllMsgQueues ms (\_ -> processValidateQueue) (0, 0)
|
||||
pure MessageStats {storedMsgsCount, expiredMsgsCount = 0, storedQueues}
|
||||
where
|
||||
processExpireQueue old q (!stored, !expired, !qCount) =
|
||||
runExceptT expireQueue >>= \case
|
||||
Right (stored', expired') -> pure (stored + stored', expired + expired', qCount + 1)
|
||||
Left e -> do
|
||||
logInfo $ "failed expiring messages in queue " <> T.pack (queueDirectory q) <> ": " <> tshow e
|
||||
exitFailure
|
||||
where
|
||||
expireQueue = do
|
||||
expired'' <- deleteExpiredMsgs q old
|
||||
stored'' <- liftIO $ getQueueSize q
|
||||
liftIO $ closeMsgQueue q
|
||||
pure (stored'', expired'')
|
||||
processValidateQueue q (!stored, !qCount) = getQueueSize q >>= \stored' -> pure (stored + stored', qCount + 1)
|
||||
|
||||
importMessages :: MsgStoreClass s => s -> FilePath -> Maybe Int64 -> IO Int
|
||||
importMessages :: MsgStoreClass s => s -> FilePath -> Maybe Int64 -> IO MessageStats
|
||||
importMessages ms f old_ = do
|
||||
logInfo $ "restoring messages from file " <> T.pack f
|
||||
LB.readFile f >>= runExceptT . foldM restoreMsg (0, 0, 0) . zip [0..] . LB.lines >>= \case
|
||||
@@ -1744,19 +1786,18 @@ importMessages ms f old_ = do
|
||||
putStrLn ""
|
||||
logError . T.pack $ "error restoring messages: " <> e
|
||||
liftIO exitFailure
|
||||
Right (count, total, expired) -> do
|
||||
putStrLn ""
|
||||
Right (lineCount, storedMsgsCount, expiredMsgsCount) -> do
|
||||
putStrLn $ "Processed " <> show lineCount <> " lines"
|
||||
renameFile f $ f <> ".bak"
|
||||
logQueueStates ms
|
||||
qCount <- M.size <$> readTVarIO (activeMsgQueues ms)
|
||||
logInfo $ "Processed " <> tshow count <> " lines, imported " <> tshow total <> " messages into " <> tshow qCount <> " queues, expired " <> tshow expired <> " messages"
|
||||
pure expired
|
||||
storedQueues <- M.size <$> readTVarIO (activeMsgQueues ms)
|
||||
pure MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues}
|
||||
where
|
||||
progress i = do
|
||||
liftIO $ putStr $ "Processed " <> show i <> " lines\r"
|
||||
hFlush stdout
|
||||
restoreMsg :: (Int, Int, Int) -> (Int, LB.ByteString) -> ExceptT String IO (Int, Int, Int)
|
||||
restoreMsg (!count, !total, !expired) (i, s') = do
|
||||
restoreMsg (!lineCount, !stored, !expired) (i, s') = do
|
||||
when (i `mod` 1000 == 0) $ progress i
|
||||
MLRv3 rId msg <- liftEither . first (msgErr "parsing") $ strDecode s
|
||||
liftError show $ addToMsgQueue rId msg
|
||||
@@ -1766,16 +1807,20 @@ importMessages ms f old_ = do
|
||||
q <- getMsgQueue ms rId
|
||||
(isExpired, logFull) <- case msg of
|
||||
Message {msgTs}
|
||||
| maybe True (systemSeconds msgTs >=) old_ -> (False,) . isNothing <$> writeMsg q False msg
|
||||
| maybe True (systemSeconds msgTs >=) old_ -> (False,) . isNothing <$> writeMsg ms q False msg
|
||||
| otherwise -> pure (True, False)
|
||||
MessageQuota {} -> writeMsg q False msg $> (False, False)
|
||||
MessageQuota {} -> writeMsg ms q False msg $> (False, False)
|
||||
when logFull . logError . decodeLatin1 $ "message queue " <> strEncode rId <> " is full, message not restored: " <> strEncode (messageId msg)
|
||||
let total' = if logFull then total else total + 1
|
||||
expired' = if isExpired then expired + 1 else expired
|
||||
pure (count + 1, total', expired')
|
||||
let !stored' = if logFull || isExpired then stored else stored + 1
|
||||
!expired' = if isExpired then expired + 1 else expired
|
||||
pure (lineCount + 1, stored', expired')
|
||||
msgErr :: Show e => String -> e -> String
|
||||
msgErr op e = op <> " error (" <> show e <> "): " <> B.unpack (B.take 100 s)
|
||||
|
||||
printMessageStats :: T.Text -> MessageStats -> IO ()
|
||||
printMessageStats name MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues} =
|
||||
logInfo $ name <> " stored: " <> tshow storedMsgsCount <> ", expired: " <> tshow expiredMsgsCount <> ", queues: " <> tshow storedQueues
|
||||
|
||||
saveServerNtfs :: M ()
|
||||
saveServerNtfs = asks (storeNtfsFile . config) >>= mapM_ saveNtfs
|
||||
where
|
||||
@@ -1790,33 +1835,37 @@ saveServerNtfs = asks (storeNtfsFile . config) >>= mapM_ saveNtfs
|
||||
saveQueueNtfs h (nId, v) = BLD.hPutBuilder h . encodeNtfs nId . reverse =<< readTVarIO v
|
||||
encodeNtfs nId = mconcat . map (\ntf -> BLD.byteString (strEncode $ NLRv1 nId ntf) <> BLD.char8 '\n')
|
||||
|
||||
restoreServerNtfs :: M Int
|
||||
restoreServerNtfs :: M MessageStats
|
||||
restoreServerNtfs =
|
||||
asks (storeNtfsFile . config) >>= \case
|
||||
Just f -> ifM (doesFileExist f) (restoreNtfs f) (pure 0)
|
||||
Nothing -> pure 0
|
||||
Just f -> ifM (doesFileExist f) (restoreNtfs f) (pure newMessageStats)
|
||||
Nothing -> pure newMessageStats
|
||||
where
|
||||
restoreNtfs f = do
|
||||
logInfo $ "restoring notifications from file " <> T.pack f
|
||||
ns <- asks ntfStore
|
||||
old <- asks (notificationExpiration . config) >>= liftIO . expireBeforeEpoch
|
||||
runExceptT (liftIO (LB.readFile f) >>= foldM (restoreNtf ns old) 0 . LB.lines) >>= \case
|
||||
Left e -> do
|
||||
logError . T.pack $ "error restoring notifications: " <> e
|
||||
liftIO exitFailure
|
||||
Right expired -> do
|
||||
renameFile f $ f <> ".bak"
|
||||
logInfo "notifications restored"
|
||||
pure expired
|
||||
liftIO $
|
||||
LB.readFile f >>= runExceptT . foldM (restoreNtf ns old) (0, 0, 0) . LB.lines >>= \case
|
||||
Left e -> do
|
||||
logError . T.pack $ "error restoring notifications: " <> e
|
||||
liftIO exitFailure
|
||||
Right (lineCount, storedMsgsCount, expiredMsgsCount) -> do
|
||||
renameFile f $ f <> ".bak"
|
||||
let NtfStore ns' = ns
|
||||
storedQueues <- M.size <$> readTVarIO ns'
|
||||
logInfo $ "notifications restored, " <> tshow lineCount <> " lines processed"
|
||||
pure MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues}
|
||||
where
|
||||
restoreNtf ns old !expired s' = do
|
||||
restoreNtf :: NtfStore -> Int64 -> (Int, Int, Int) -> LB.ByteString -> ExceptT String IO (Int, Int, Int)
|
||||
restoreNtf ns old (!lineCount, !stored, !expired) s' = do
|
||||
NLRv1 nId ntf <- liftEither . first (ntfErr "parsing") $ strDecode s
|
||||
liftIO $ addToNtfs nId ntf
|
||||
where
|
||||
s = LB.toStrict s'
|
||||
addToNtfs nId ntf@MsgNtf {ntfTs}
|
||||
| systemSeconds ntfTs < old = pure (expired + 1)
|
||||
| otherwise = storeNtf ns nId ntf $> expired
|
||||
| systemSeconds ntfTs < old = pure (lineCount + 1, stored, expired + 1)
|
||||
| otherwise = storeNtf ns nId ntf $> (lineCount + 1, stored + 1, expired)
|
||||
ntfErr :: Show e => String -> e -> String
|
||||
ntfErr op e = op <> " error (" <> show e <> "): " <> B.unpack (B.take 100 s)
|
||||
|
||||
@@ -1830,26 +1879,25 @@ saveServerStats =
|
||||
B.writeFile f $ strEncode stats
|
||||
logInfo "server stats saved"
|
||||
|
||||
restoreServerStats :: Int -> Int -> M ()
|
||||
restoreServerStats expiredMsgs expiredNtfs = asks (serverStatsBackupFile . config) >>= mapM_ restoreStats
|
||||
restoreServerStats :: MessageStats -> MessageStats -> M ()
|
||||
restoreServerStats msgStats ntfStats = asks (serverStatsBackupFile . config) >>= mapM_ restoreStats
|
||||
where
|
||||
restoreStats f = whenM (doesFileExist f) $ do
|
||||
logInfo $ "restoring server stats from file " <> T.pack f
|
||||
liftIO (strDecode <$> B.readFile f) >>= \case
|
||||
Right d@ServerStatsData {_qCount = statsQCount} -> do
|
||||
Right d@ServerStatsData {_qCount = statsQCount, _msgCount = statsMsgCount, _ntfCount = statsNtfCount} -> do
|
||||
s <- asks serverStats
|
||||
_qCount <- fmap M.size . readTVarIO . queues =<< asks queueStore
|
||||
_msgCount <-
|
||||
asks msgStore >>= \case
|
||||
AMS SMSMemory ms -> liftIO $ readTVarIO (msgQueues ms) >>= foldM (\(!n) q -> (n +) <$> getQueueSize q) 0
|
||||
_ -> pure 0
|
||||
NtfStore ns <- asks ntfStore
|
||||
_ntfCount <- liftIO . foldM (\(!n) q -> (n +) . length <$> readTVarIO q) 0 =<< readTVarIO ns
|
||||
liftIO $ setServerStats s d {_qCount, _msgCount, _ntfCount, _msgExpired = _msgExpired d + expiredMsgs, _msgNtfExpired = _msgNtfExpired d + expiredNtfs}
|
||||
let _msgCount = storedMsgsCount msgStats
|
||||
_ntfCount = storedMsgsCount ntfStats
|
||||
liftIO $ setServerStats s d {_qCount, _msgCount, _ntfCount, _msgExpired = _msgExpired d + expiredMsgsCount msgStats, _msgNtfExpired = _msgNtfExpired d + expiredMsgsCount ntfStats}
|
||||
renameFile f $ f <> ".bak"
|
||||
logInfo "server stats restored"
|
||||
when (_qCount /= statsQCount) $ logWarn $ "Queue count differs: stats: " <> tshow statsQCount <> ", store: " <> tshow _qCount
|
||||
logInfo $ "Restored " <> tshow _msgCount <> " messages in " <> tshow _qCount <> " queues"
|
||||
compareCounts "Queue" statsQCount _qCount
|
||||
compareCounts "Message" statsMsgCount _msgCount
|
||||
compareCounts "Notification" statsNtfCount _ntfCount
|
||||
Left e -> do
|
||||
logInfo $ "error restoring server stats: " <> T.pack e
|
||||
liftIO exitFailure
|
||||
compareCounts name statsCnt storeCnt =
|
||||
when (statsCnt /= storeCnt) $ logWarn $ name <> " count differs: stats: " <> tshow statsCnt <> ", store: " <> tshow storeCnt
|
||||
@@ -39,7 +39,7 @@ import qualified Simplex.Messaging.Crypto as C
|
||||
import Simplex.Messaging.Encoding.String
|
||||
import Simplex.Messaging.Parsers (parseAll)
|
||||
import Simplex.Messaging.Protocol (BasicAuth (..), ProtoServerWithAuth (ProtoServerWithAuth), pattern SMPServer)
|
||||
import Simplex.Messaging.Server (AttachHTTP, exportMessages, importMessages, runSMPServer)
|
||||
import Simplex.Messaging.Server (AttachHTTP, exportMessages, importMessages, printMessageStats, runSMPServer)
|
||||
import Simplex.Messaging.Server.CLI
|
||||
import Simplex.Messaging.Server.Env.STM
|
||||
import Simplex.Messaging.Server.Expiration
|
||||
@@ -99,8 +99,9 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
|
||||
("WARNING: message log file " <> storeMsgsFilePath <> " will be imported to journal directory " <> storeMsgsJournalDir)
|
||||
"Messages not imported"
|
||||
ms <- newJournalMsgStore
|
||||
void $ importMessages ms storeMsgsFilePath Nothing -- no expiration
|
||||
msgStats <- importMessages ms storeMsgsFilePath Nothing -- no expiration
|
||||
putStrLn "Import completed"
|
||||
printMessageStats "Messages" msgStats
|
||||
putStrLn $ case readMsgStoreType ini of
|
||||
Right (AMSType SMSMemory) -> "store_messages set to `memory`, update it to `journal` in INI file"
|
||||
Right (AMSType SMSJournal) -> "store_messages set to `journal`"
|
||||
|
||||
@@ -10,16 +10,19 @@
|
||||
{-# LANGUAGE NamedFieldPuns #-}
|
||||
{-# LANGUAGE OverloadedStrings #-}
|
||||
{-# LANGUAGE ScopedTypeVariables #-}
|
||||
{-# LANGUAGE StandaloneDeriving #-}
|
||||
{-# LANGUAGE TypeFamilies #-}
|
||||
|
||||
module Simplex.Messaging.Server.MsgStore.Journal
|
||||
( JournalMsgStore (msgQueues, random),
|
||||
JournalMsgQueue,
|
||||
JournalMsgQueue (queueDirectory),
|
||||
JournalStoreConfig (..),
|
||||
getQueueMessages,
|
||||
closeMsgQueue,
|
||||
-- below are exported for tests
|
||||
MsgQueueState (..),
|
||||
JournalState (..),
|
||||
SJournalType (..),
|
||||
msgQueueDirectory,
|
||||
readWriteQueueState,
|
||||
newMsgQueueState,
|
||||
@@ -54,6 +57,7 @@ import Simplex.Messaging.TMap (TMap)
|
||||
import qualified Simplex.Messaging.TMap as TM
|
||||
import Simplex.Messaging.Util (ifM, tshow, ($>>=))
|
||||
import System.Directory
|
||||
import System.Exit
|
||||
import System.FilePath ((</>))
|
||||
import System.IO (BufferMode (..), Handle, IOMode (..), SeekMode (..), stdout)
|
||||
import qualified System.IO as IO
|
||||
@@ -78,22 +82,20 @@ data JournalStoreConfig = JournalStoreConfig
|
||||
}
|
||||
|
||||
data JournalMsgQueue = JournalMsgQueue
|
||||
{ config :: JournalStoreConfig,
|
||||
queueDirectory :: FilePath,
|
||||
{ queueDirectory :: FilePath,
|
||||
queueLock :: Lock,
|
||||
state :: TVar MsgQueueState,
|
||||
-- Last message and length incl. newline
|
||||
-- tipMsg contains last message and length incl. newline
|
||||
-- Nothing - unknown, Just Nothing - empty queue.
|
||||
-- This optimization prevents reading each message at least twice,
|
||||
-- or reading it after it was just written.
|
||||
-- It prevents reading each message twice,
|
||||
-- and reading it after it was just written.
|
||||
tipMsg :: TVar (Maybe (Maybe (Message, Int))),
|
||||
handles :: TVar (Maybe MsgQueueHandles),
|
||||
random :: TVar StdGen
|
||||
handles :: TVar (Maybe MsgQueueHandles)
|
||||
}
|
||||
|
||||
data MsgQueueState = MsgQueueState
|
||||
{ writeState :: JournalState,
|
||||
readState :: JournalState,
|
||||
{ readState :: JournalState 'JTRead,
|
||||
writeState :: JournalState 'JTWrite,
|
||||
canWrite :: Bool,
|
||||
size :: Int
|
||||
}
|
||||
@@ -105,21 +107,49 @@ data MsgQueueHandles = MsgQueueHandles
|
||||
writeHandle :: Maybe Handle -- optional, used when write file is different from read file
|
||||
}
|
||||
|
||||
data JournalState = JournalState
|
||||
{ journalId :: ByteString,
|
||||
data JournalState t = JournalState
|
||||
{ journalType :: SJournalType t,
|
||||
journalId :: ByteString,
|
||||
msgPos :: Int,
|
||||
msgCount :: Int,
|
||||
bytePos :: Int
|
||||
bytePos :: Int,
|
||||
byteCount :: Int
|
||||
}
|
||||
deriving (Show)
|
||||
|
||||
data JournalType = JTRead | JTWrite
|
||||
|
||||
data SJournalType (t :: JournalType) where
|
||||
SJTRead :: SJournalType 'JTRead
|
||||
SJTWrite :: SJournalType 'JTWrite
|
||||
|
||||
class JournalTypeI t where sJournalType :: SJournalType t
|
||||
|
||||
instance JournalTypeI 'JTRead where sJournalType = SJTRead
|
||||
|
||||
instance JournalTypeI 'JTWrite where sJournalType = SJTWrite
|
||||
|
||||
deriving instance Show (SJournalType t)
|
||||
|
||||
newMsgQueueState :: ByteString -> MsgQueueState
|
||||
newMsgQueueState journalId =
|
||||
let st = newJournalState journalId
|
||||
in MsgQueueState {writeState = st, readState = st, canWrite = True, size = 0}
|
||||
MsgQueueState
|
||||
{ writeState = newJournalState journalId,
|
||||
readState = newJournalState journalId,
|
||||
canWrite = True,
|
||||
size = 0
|
||||
}
|
||||
|
||||
newJournalState :: ByteString -> JournalState
|
||||
newJournalState journalId = JournalState {journalId, msgPos = 0, msgCount = 0, bytePos = 0}
|
||||
newJournalState :: JournalTypeI t => ByteString -> JournalState t
|
||||
newJournalState journalId =
|
||||
JournalState
|
||||
{ journalType = sJournalType,
|
||||
journalId,
|
||||
msgPos = 0,
|
||||
msgCount = 0,
|
||||
bytePos = 0,
|
||||
byteCount = 0
|
||||
}
|
||||
|
||||
journalFilePath :: FilePath -> ByteString -> FilePath
|
||||
journalFilePath dir journalId = dir </> (msgLogFileName <> "." <> B.unpack journalId <> logFileExt)
|
||||
@@ -139,15 +169,15 @@ instance StrEncoding MsgQueueState where
|
||||
size <- " size=" *> strP
|
||||
pure MsgQueueState {writeState, readState, canWrite, size}
|
||||
|
||||
instance StrEncoding JournalState where
|
||||
strEncode JournalState {journalId, msgPos, msgCount, bytePos} =
|
||||
B.intercalate "," [journalId, strEncode msgPos, strEncode msgCount, strEncode bytePos]
|
||||
instance JournalTypeI t => StrEncoding (JournalState t) where
|
||||
strEncode JournalState {journalId, msgPos, msgCount, bytePos, byteCount} =
|
||||
B.concat [journalId, ",", strEncodeList [msgPos, msgCount, bytePos, byteCount]]
|
||||
strP = do
|
||||
journalId <- A.takeTill (== ',')
|
||||
msgPos <- A.char ',' *> strP
|
||||
msgCount <- A.char ',' *> strP
|
||||
bytePos <- A.char ',' *> strP
|
||||
pure JournalState {journalId, msgPos, msgCount, bytePos}
|
||||
A.char ',' *> strListP >>= \case
|
||||
[msgPos, msgCount, bytePos, byteCount] ->
|
||||
pure JournalState {journalType = sJournalType, journalId, msgPos, msgCount, bytePos, byteCount}
|
||||
_ -> fail "bad JournalState"
|
||||
|
||||
queueLogFileName :: String
|
||||
queueLogFileName = "queue_state"
|
||||
@@ -158,11 +188,11 @@ msgLogFileName = "messages"
|
||||
logFileExt :: String
|
||||
logFileExt = ".log"
|
||||
|
||||
newtype NonAtomicIO a = NonAtomicIO (IO a)
|
||||
newtype StoreIO a = StoreIO (IO a)
|
||||
deriving newtype (Functor, Applicative, Monad)
|
||||
|
||||
instance MsgStoreClass JournalMsgStore where
|
||||
type StoreMonad JournalMsgStore = NonAtomicIO
|
||||
type StoreMonad JournalMsgStore = StoreIO
|
||||
type MsgQueue JournalMsgStore = JournalMsgQueue
|
||||
type MsgStoreConfig JournalMsgStore = JournalStoreConfig
|
||||
|
||||
@@ -173,33 +203,36 @@ instance MsgStoreClass JournalMsgStore where
|
||||
msgQueues <- TM.emptyIO
|
||||
pure JournalMsgStore {config, random, queueLocks, msgQueues}
|
||||
|
||||
closeMsgStore st = readTVarIO (msgQueues st) >>= mapM_ closeMsgQueue_
|
||||
closeMsgStore st = readTVarIO (msgQueues st) >>= mapM_ closeMsgQueue
|
||||
|
||||
activeMsgQueues = msgQueues
|
||||
{-# INLINE activeMsgQueues #-}
|
||||
|
||||
-- This function opens and closes all queues.
|
||||
-- It is used to export storage to a single file, not during normal server execution.
|
||||
withAllMsgQueues :: JournalMsgStore -> (RecipientId -> JournalMsgQueue -> IO Int) -> IO Int
|
||||
withAllMsgQueues st@JournalMsgStore {config} action = do
|
||||
closeMsgStore st
|
||||
lock <- createLockIO -- the same lock is used for all queues
|
||||
dirs <- zip [0..] <$> listQueueDirs 0 ("", storePath)
|
||||
let count = length dirs
|
||||
total <- foldM (processQueue lock count) 0 dirs
|
||||
progress count count
|
||||
putStrLn ""
|
||||
pure total
|
||||
-- This function is a "foldr" that opens and closes all queues, processes them as defined by action and accumulates the result.
|
||||
-- It is used to export storage to a single file and also to expire messages and validate all queues when server is started.
|
||||
withAllMsgQueues :: JournalMsgStore -> (RecipientId -> JournalMsgQueue -> a -> IO a) -> a -> IO a
|
||||
withAllMsgQueues ms@JournalMsgStore {config} action res = ifM (doesDirectoryExist storePath) processStore (pure res)
|
||||
where
|
||||
processStore = do
|
||||
closeMsgStore ms
|
||||
lock <- createLockIO -- the same lock is used for all queues
|
||||
dirs <- zip [0..] <$> listQueueDirs 0 ("", storePath)
|
||||
let count = length dirs
|
||||
res' <- foldM (processQueue lock count) res dirs
|
||||
progress count count
|
||||
putStrLn ""
|
||||
pure res'
|
||||
JournalStoreConfig {storePath, pathParts} = config
|
||||
processQueue lock count !total (i :: Int, (queueId, dir)) = do
|
||||
processQueue lock count acc (i :: Int, (queueId, dir)) = do
|
||||
when (i `mod` 100 == 0) $ progress i count
|
||||
q <- openMsgQueue st dir lock
|
||||
total' <- case strDecode $ B.pack queueId of
|
||||
Right rId -> (total +) <$> action rId q
|
||||
Left e -> total <$ putStrLn ("Error: message queue directory " <> dir <> " is invalid: " <> e)
|
||||
closeMsgQueue_ q
|
||||
pure total'
|
||||
q <- openMsgQueue ms dir lock
|
||||
acc' <- case strDecode $ B.pack queueId of
|
||||
Right rId -> action rId q acc
|
||||
Left e -> do
|
||||
putStrLn ("Error: message queue directory " <> dir <> " is invalid: " <> e)
|
||||
exitFailure
|
||||
closeMsgQueue q
|
||||
pure acc'
|
||||
progress i count = do
|
||||
putStr $ "Processed: " <> show i <> "/" <> show count <> " queues\r"
|
||||
IO.hFlush stdout
|
||||
@@ -216,21 +249,21 @@ instance MsgStoreClass JournalMsgStore where
|
||||
(Nothing <$ putStrLn ("Error: path " <> path' <> " is not a directory, skipping"))
|
||||
|
||||
logQueueStates :: JournalMsgStore -> IO ()
|
||||
logQueueStates st =
|
||||
void $ withActiveMsgQueues st $ \_ q ->
|
||||
readTVarIO (handles q)
|
||||
>>= maybe (pure ()) (\hs -> readTVarIO (state q) >>= logQueueState (stateHandle hs))
|
||||
>> pure 0
|
||||
logQueueStates st = withActiveMsgQueues st (\_ q _ -> logState q) ()
|
||||
where
|
||||
logState q =
|
||||
readTVarIO (handles q)
|
||||
>>= maybe (pure ()) (\hs -> readTVarIO (state q) >>= logQueueState (stateHandle hs))
|
||||
|
||||
getMsgQueue :: JournalMsgStore -> RecipientId -> ExceptT ErrorType IO JournalMsgQueue
|
||||
getMsgQueue store@JournalMsgStore {queueLocks, msgQueues, random} rId =
|
||||
getMsgQueue ms@JournalMsgStore {queueLocks, msgQueues, random} rId =
|
||||
tryStore "getMsgQueue" $ withLockMap queueLocks rId "getMsgQueue" $
|
||||
TM.lookupIO rId msgQueues >>= maybe newQ pure
|
||||
where
|
||||
newQ = do
|
||||
let dir = msgQueueDirectory store rId
|
||||
let dir = msgQueueDirectory ms rId
|
||||
queueLock <- atomically $ getMapLock queueLocks rId
|
||||
q <- ifM (doesDirectoryExist dir) (openMsgQueue store dir queueLock) (createQ dir queueLock)
|
||||
q <- ifM (doesDirectoryExist dir) (openMsgQueue ms dir queueLock) (createQ dir queueLock)
|
||||
atomically $ TM.insert rId q msgQueues
|
||||
pure q
|
||||
where
|
||||
@@ -239,24 +272,24 @@ instance MsgStoreClass JournalMsgStore where
|
||||
-- folder and files are not created here,
|
||||
-- to avoid file IO for queues without messages during subscription
|
||||
journalId <- newJournalId random
|
||||
mkJournalQueue store dir queueLock (newMsgQueueState journalId, Nothing)
|
||||
mkJournalQueue dir queueLock (newMsgQueueState journalId, Nothing)
|
||||
|
||||
delMsgQueue :: JournalMsgStore -> RecipientId -> IO ()
|
||||
delMsgQueue st rId = withLockMap (queueLocks st) rId "delMsgQueue" $ do
|
||||
void $ closeMsgQueue st rId
|
||||
removeQueueDirectory st rId
|
||||
delMsgQueue ms rId = withLockMap (queueLocks ms) rId "delMsgQueue" $ do
|
||||
void $ deleteMsgQueue_ ms rId
|
||||
removeQueueDirectory ms rId
|
||||
|
||||
delMsgQueueSize :: JournalMsgStore -> RecipientId -> IO Int
|
||||
delMsgQueueSize st rId = withLockMap (queueLocks st) rId "delMsgQueue" $ do
|
||||
state_ <- closeMsgQueue st rId
|
||||
delMsgQueueSize ms rId = withLockMap (queueLocks ms) rId "delMsgQueue" $ do
|
||||
state_ <- deleteMsgQueue_ ms rId
|
||||
sz <- maybe (pure $ -1) (fmap size . readTVarIO) state_
|
||||
removeQueueDirectory st rId
|
||||
removeQueueDirectory ms rId
|
||||
pure sz
|
||||
|
||||
getQueueMessages :: Bool -> JournalMsgQueue -> IO [Message]
|
||||
getQueueMessages drainMsgs q = readTVarIO (handles q) >>= maybe (pure []) (getMsg [])
|
||||
where
|
||||
getMsg ms hs = chooseReadJournal q drainMsgs hs >>= maybe (pure ms) readMsg
|
||||
getMsg msgs hs = chooseReadJournal q drainMsgs hs >>= maybe (pure msgs) readMsg
|
||||
where
|
||||
readMsg (rs, h) = do
|
||||
-- TODO handle errors
|
||||
@@ -264,10 +297,10 @@ instance MsgStoreClass JournalMsgStore where
|
||||
-- TODO handle errors
|
||||
Right msg <- pure $ strDecode s
|
||||
updateReadPos q drainMsgs (B.length s + 1) hs -- 1 is to account for new line
|
||||
(msg :) <$> getMsg ms hs
|
||||
(msg :) <$> getMsg msgs hs
|
||||
|
||||
writeMsg :: JournalMsgQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
|
||||
writeMsg q@JournalMsgQueue {queueDirectory, handles, config, random} logState !msg =
|
||||
writeMsg :: JournalMsgStore -> JournalMsgQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
|
||||
writeMsg ms q@JournalMsgQueue {queueDirectory, handles} logState !msg =
|
||||
tryStore "writeMsg" $ withLock' (queueLock q) "writeMsg" $ do
|
||||
st@MsgQueueState {canWrite, size} <- readTVarIO (state q)
|
||||
let empty = size == 0
|
||||
@@ -279,7 +312,7 @@ instance MsgStoreClass JournalMsgStore where
|
||||
else writeToJournal st canWrt' msgQuota $> Nothing
|
||||
else pure Nothing
|
||||
where
|
||||
JournalStoreConfig {quota, maxMsgCount} = config
|
||||
JournalStoreConfig {quota, maxMsgCount} = config ms
|
||||
msgQuota = MessageQuota {msgId = msgId msg, msgTs = msgTs msg}
|
||||
writeToJournal st@MsgQueueState {writeState, readState = rs, size} canWrt' msg' = do
|
||||
let msgStr = strEncode msg' `B.snoc` '\n'
|
||||
@@ -306,7 +339,7 @@ instance MsgStoreClass JournalMsgStore where
|
||||
atomically $ writeTVar handles $ Just hs
|
||||
pure hs
|
||||
switchWriteJournal hs = do
|
||||
journalId <- newJournalId random
|
||||
journalId <- newJournalId $ random ms
|
||||
wh <- createNewJournal queueDirectory journalId
|
||||
atomically $ writeTVar handles $ Just $ hs {writeHandle = Just wh}
|
||||
pure (newJournalState journalId, wh)
|
||||
@@ -314,9 +347,9 @@ instance MsgStoreClass JournalMsgStore where
|
||||
getQueueSize :: JournalMsgQueue -> IO Int
|
||||
getQueueSize JournalMsgQueue {state} = size <$> readTVarIO state
|
||||
|
||||
tryPeekMsg_ :: JournalMsgQueue -> NonAtomicIO (Maybe Message)
|
||||
tryPeekMsg_ :: JournalMsgQueue -> StoreIO (Maybe Message)
|
||||
tryPeekMsg_ q@JournalMsgQueue {tipMsg, handles} =
|
||||
NonAtomicIO $ readTVarIO handles $>>= chooseReadJournal q True $>>= peekMsg
|
||||
StoreIO $ readTVarIO handles $>>= chooseReadJournal q True $>>= peekMsg
|
||||
where
|
||||
peekMsg (rs, h) = readTVarIO tipMsg >>= maybe readMsg (pure . fmap fst)
|
||||
where
|
||||
@@ -328,16 +361,16 @@ instance MsgStoreClass JournalMsgStore where
|
||||
atomically $ writeTVar tipMsg $ Just (Just (msg, B.length s + 1)) -- 1 is to account for new line
|
||||
pure $ Just msg
|
||||
|
||||
tryDeleteMsg_ :: JournalMsgQueue -> NonAtomicIO ()
|
||||
tryDeleteMsg_ q@JournalMsgQueue {tipMsg, handles} = NonAtomicIO $
|
||||
tryDeleteMsg_ :: JournalMsgQueue -> StoreIO ()
|
||||
tryDeleteMsg_ q@JournalMsgQueue {tipMsg, handles} = StoreIO $
|
||||
void $
|
||||
readTVarIO tipMsg -- if there is no cached tipMsg, do nothing
|
||||
$>>= (pure . fmap snd)
|
||||
$>>= \len -> readTVarIO handles
|
||||
$>>= \hs -> updateReadPos q True len hs $> Just ()
|
||||
|
||||
atomicQueue :: JournalMsgQueue -> String -> NonAtomicIO a -> ExceptT ErrorType IO a
|
||||
atomicQueue mq op (NonAtomicIO a) = tryStore op $ withLock' (queueLock mq) op $ a
|
||||
isolateQueue :: JournalMsgQueue -> String -> StoreIO a -> ExceptT ErrorType IO a
|
||||
isolateQueue mq op (StoreIO a) = tryStore op $ withLock' (queueLock mq) op $ a
|
||||
|
||||
tryStore :: String -> IO a -> ExceptT ErrorType IO a
|
||||
tryStore op a =
|
||||
@@ -347,15 +380,15 @@ tryStore op a =
|
||||
in logError ("STORE ERROR " <> T.pack e') $> Left (STORE e')
|
||||
|
||||
openMsgQueue :: JournalMsgStore -> FilePath -> Lock -> IO JournalMsgQueue
|
||||
openMsgQueue store dir queueLock = do
|
||||
openMsgQueue ms dir queueLock = do
|
||||
let statePath = dir </> (queueLogFileName <> logFileExt)
|
||||
(st, sh) <- readWriteQueueState store statePath
|
||||
(st, sh) <- readWriteQueueState ms statePath
|
||||
(st', rh, wh_) <- openJournals dir st
|
||||
let hs = MsgQueueHandles {stateHandle = sh, readHandle = rh, writeHandle = wh_}
|
||||
mkJournalQueue store dir queueLock (st', Just hs)
|
||||
mkJournalQueue dir queueLock (st', Just hs)
|
||||
|
||||
mkJournalQueue :: JournalMsgStore -> FilePath -> Lock -> (MsgQueueState, Maybe MsgQueueHandles) -> IO JournalMsgQueue
|
||||
mkJournalQueue JournalMsgStore {random, config} dir queueLock (st, hs_) = do
|
||||
mkJournalQueue :: FilePath -> Lock -> (MsgQueueState, Maybe MsgQueueHandles) -> IO JournalMsgQueue
|
||||
mkJournalQueue dir queueLock (st, hs_) = do
|
||||
state <- newTVarIO st
|
||||
tipMsg <- newTVarIO Nothing
|
||||
handles <- newTVarIO hs_
|
||||
@@ -363,16 +396,14 @@ mkJournalQueue JournalMsgStore {random, config} dir queueLock (st, hs_) = do
|
||||
-- to avoid map lookup on queue operations
|
||||
pure
|
||||
JournalMsgQueue
|
||||
{ config,
|
||||
queueDirectory = dir,
|
||||
{ queueDirectory = dir,
|
||||
queueLock,
|
||||
state,
|
||||
tipMsg,
|
||||
handles,
|
||||
random
|
||||
handles
|
||||
}
|
||||
|
||||
chooseReadJournal :: JournalMsgQueue -> Bool -> MsgQueueHandles -> IO (Maybe (JournalState, Handle))
|
||||
chooseReadJournal :: JournalMsgQueue -> Bool -> MsgQueueHandles -> IO (Maybe (JournalState 'JTRead, Handle))
|
||||
chooseReadJournal q log' hs = do
|
||||
st@MsgQueueState {writeState = ws, readState = rs} <- readTVarIO (state q)
|
||||
case writeHandle hs of
|
||||
@@ -390,8 +421,8 @@ chooseReadJournal q log' hs = do
|
||||
|
||||
updateQueueState :: JournalMsgQueue -> Bool -> MsgQueueHandles -> MsgQueueState -> IO ()
|
||||
updateQueueState q log' hs st = do
|
||||
atomically $ writeTVar (state q) st
|
||||
when log' $ logQueueState (stateHandle hs) st
|
||||
atomically $ writeTVar (state q) st
|
||||
|
||||
logQueueState :: Handle -> MsgQueueState -> IO ()
|
||||
logQueueState h st = B.hPutStr h $ strEncode st `B.snoc` '\n'
|
||||
@@ -454,9 +485,11 @@ openJournals dir st@MsgQueueState {readState = rs, writeState = ws} = do
|
||||
st' <- fixWriteFileSize wh
|
||||
pure (st', rh, Just wh)
|
||||
where
|
||||
openJournal :: JournalState t -> IO (Either FilePath Handle)
|
||||
openJournal JournalState {journalId} =
|
||||
let path = journalFilePath dir journalId
|
||||
in ifM (doesFileExist path) (Right <$> openFile path ReadWriteMode) (pure $ Left path)
|
||||
-- do that for all append operations
|
||||
fixWriteFileSize h = do
|
||||
let sz = fromIntegral $ bytePos ws
|
||||
sz' <- IO.hFileSize h
|
||||
@@ -465,7 +498,7 @@ openJournals dir st@MsgQueueState {readState = rs, writeState = ws} = do
|
||||
| sz' == sz -> pure st
|
||||
| otherwise -> pure st -- TODO re-read file to recover what is possible ???
|
||||
|
||||
removeJournal :: FilePath -> JournalState -> IO ()
|
||||
removeJournal :: FilePath -> JournalState t -> IO ()
|
||||
removeJournal dir JournalState {journalId} = do
|
||||
let path = journalFilePath dir journalId
|
||||
removeFile path `catchAny` (\e -> logError $ "STORE ERROR removing file " <> T.pack path <> ": " <> tshow e)
|
||||
@@ -522,13 +555,13 @@ readWriteQueueState JournalMsgStore {random, config} statePath =
|
||||
logQueueState sh st
|
||||
pure (st, sh)
|
||||
|
||||
closeMsgQueue :: JournalMsgStore -> RecipientId -> IO (Maybe (TVar MsgQueueState))
|
||||
closeMsgQueue st rId =
|
||||
deleteMsgQueue_ :: JournalMsgStore -> RecipientId -> IO (Maybe (TVar MsgQueueState))
|
||||
deleteMsgQueue_ st rId =
|
||||
atomically (TM.lookupDelete rId (msgQueues st))
|
||||
>>= mapM (\q -> closeMsgQueue_ q $> state q)
|
||||
>>= mapM (\q -> closeMsgQueue q $> state q)
|
||||
|
||||
closeMsgQueue_ :: JournalMsgQueue -> IO ()
|
||||
closeMsgQueue_ q = readTVarIO (handles q) >>= mapM_ closeHandles
|
||||
closeMsgQueue :: JournalMsgQueue -> IO ()
|
||||
closeMsgQueue q = readTVarIO (handles q) >>= mapM_ closeHandles
|
||||
where
|
||||
closeHandles (MsgQueueHandles sh rh wh_) = do
|
||||
hClose sh
|
||||
|
||||
@@ -93,8 +93,8 @@ instance MsgStoreClass STMMsgStore where
|
||||
mapM_ (writeTQueue q) msgs
|
||||
pure msgs
|
||||
|
||||
writeMsg :: STMMsgQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
|
||||
writeMsg STMMsgQueue {msgQueue = q, quota, canWrite, size} _logState !msg = liftIO $ atomically $ do
|
||||
writeMsg :: STMMsgStore -> STMMsgQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
|
||||
writeMsg _ STMMsgQueue {msgQueue = q, quota, canWrite, size} _logState !msg = liftIO $ atomically $ do
|
||||
canWrt <- readTVar canWrite
|
||||
empty <- isEmptyTQueue q
|
||||
if canWrt || empty
|
||||
@@ -122,5 +122,5 @@ instance MsgStoreClass STMMsgStore where
|
||||
Just _ -> modifyTVar' size (subtract 1)
|
||||
_ -> pure ()
|
||||
|
||||
atomicQueue :: STMMsgQueue -> String -> STM a -> ExceptT ErrorType IO a
|
||||
atomicQueue _ _ = liftIO . atomically
|
||||
isolateQueue :: STMMsgQueue -> String -> STM a -> ExceptT ErrorType IO a
|
||||
isolateQueue _ _ = liftIO . atomically
|
||||
@@ -24,17 +24,17 @@ class Monad (StoreMonad s) => MsgStoreClass s where
|
||||
newMsgStore :: MsgStoreConfig s -> IO s
|
||||
closeMsgStore :: s -> IO ()
|
||||
activeMsgQueues :: s -> TMap RecipientId (MsgQueue s)
|
||||
withAllMsgQueues :: s -> (RecipientId -> MsgQueue s -> IO Int) -> IO Int
|
||||
withAllMsgQueues :: s -> (RecipientId -> MsgQueue s -> a -> IO a) -> a -> IO a
|
||||
logQueueStates :: s -> IO ()
|
||||
getMsgQueue :: s -> RecipientId -> ExceptT ErrorType IO (MsgQueue s)
|
||||
delMsgQueue :: s -> RecipientId -> IO ()
|
||||
delMsgQueueSize :: s -> RecipientId -> IO Int
|
||||
getQueueMessages :: Bool -> MsgQueue s -> IO [Message]
|
||||
writeMsg :: MsgQueue s -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
|
||||
writeMsg :: s -> MsgQueue s -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
|
||||
getQueueSize :: MsgQueue s -> IO Int
|
||||
tryPeekMsg_ :: MsgQueue s -> StoreMonad s (Maybe Message)
|
||||
tryDeleteMsg_ :: MsgQueue s -> StoreMonad s ()
|
||||
atomicQueue :: MsgQueue s -> String -> StoreMonad s a -> ExceptT ErrorType IO a
|
||||
isolateQueue :: MsgQueue s -> String -> StoreMonad s a -> ExceptT ErrorType IO a
|
||||
|
||||
data MSType = MSMemory | MSJournal
|
||||
|
||||
@@ -44,16 +44,16 @@ data SMSType :: MSType -> Type where
|
||||
|
||||
data AMSType = forall s. AMSType (SMSType s)
|
||||
|
||||
withActiveMsgQueues :: MsgStoreClass s => s -> (RecipientId -> MsgQueue s -> IO Int) -> IO Int
|
||||
withActiveMsgQueues st f = readTVarIO (activeMsgQueues st) >>= M.foldrWithKey (\k v -> ((+) <$> f k v <*>)) (pure 0)
|
||||
withActiveMsgQueues :: MsgStoreClass s => s -> (RecipientId -> MsgQueue s -> a -> IO a) -> a -> IO a
|
||||
withActiveMsgQueues st f a = readTVarIO (activeMsgQueues st) >>= M.foldrWithKey (\k v -> (>>= f k v)) (pure a)
|
||||
|
||||
tryPeekMsg :: MsgStoreClass s => MsgQueue s -> ExceptT ErrorType IO (Maybe Message)
|
||||
tryPeekMsg mq = atomicQueue mq "tryPeekMsg" $ tryPeekMsg_ mq
|
||||
tryPeekMsg mq = isolateQueue mq "tryPeekMsg" $ tryPeekMsg_ mq
|
||||
{-# INLINE tryPeekMsg #-}
|
||||
|
||||
tryDelMsg :: MsgStoreClass s => MsgQueue s -> MsgId -> ExceptT ErrorType IO (Maybe Message)
|
||||
tryDelMsg mq msgId' =
|
||||
atomicQueue mq "tryDelMsg" $
|
||||
isolateQueue mq "tryDelMsg" $
|
||||
tryPeekMsg_ mq >>= \case
|
||||
msg_@(Just msg)
|
||||
| msgId msg == msgId' ->
|
||||
@@ -63,7 +63,7 @@ tryDelMsg mq msgId' =
|
||||
-- atomic delete (== read) last and peek next message if available
|
||||
tryDelPeekMsg :: MsgStoreClass s => MsgQueue s -> MsgId -> ExceptT ErrorType IO (Maybe Message, Maybe Message)
|
||||
tryDelPeekMsg mq msgId' =
|
||||
atomicQueue mq "tryDelPeekMsg" $
|
||||
isolateQueue mq "tryDelPeekMsg" $
|
||||
tryPeekMsg_ mq >>= \case
|
||||
msg_@(Just msg)
|
||||
| msgId msg == msgId' -> (msg_,) <$> (tryDeleteMsg_ mq >> tryPeekMsg_ mq)
|
||||
@@ -71,7 +71,7 @@ tryDelPeekMsg mq msgId' =
|
||||
_ -> pure (Nothing, Nothing)
|
||||
|
||||
deleteExpiredMsgs :: MsgStoreClass s => MsgQueue s -> Int64 -> ExceptT ErrorType IO Int
|
||||
deleteExpiredMsgs mq old = atomicQueue mq "deleteExpiredMsgs" $ loop 0
|
||||
deleteExpiredMsgs mq old = isolateQueue mq "deleteExpiredMsgs" $ loop 0
|
||||
where
|
||||
loop dc =
|
||||
tryPeekMsg_ mq >>= \case
|
||||
|
||||
@@ -1042,7 +1042,7 @@ testAllowConnectionClientRestart t = do
|
||||
bob <- getSMPAgentClient' 2 agentCfg initAgentServersSrv2 testDB2
|
||||
withSmpServerStoreLogOn t testPort $ \_ -> do
|
||||
(aliceId, bobId, confId) <-
|
||||
withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2} testPort2 $ \_ -> do
|
||||
withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2} testPort2 $ \_ -> do
|
||||
runRight $ do
|
||||
(bobId, qInfo) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
|
||||
(aliceId, sqSecured) <- joinConnection bob 1 True qInfo "bob's connInfo" SMSubscribe
|
||||
@@ -1064,8 +1064,7 @@ testAllowConnectionClientRestart t = do
|
||||
alice2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
|
||||
runRight_ $ subscribeConnection alice2 bobId
|
||||
threadDelay 500000
|
||||
|
||||
withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2} testPort2 $ \_ -> do
|
||||
withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2} testPort2 $ \_ -> do
|
||||
runRight $ do
|
||||
("", "", UP _ _) <- nGet bob
|
||||
get alice2 ##> ("", bobId, CON)
|
||||
|
||||
@@ -21,7 +21,7 @@ import Data.Time.Clock.System (getSystemTime)
|
||||
import Simplex.Messaging.Crypto (pattern MaxLenBS)
|
||||
import qualified Simplex.Messaging.Crypto as C
|
||||
import Simplex.Messaging.Protocol (EntityId (..), Message (..), noMsgFlags)
|
||||
import Simplex.Messaging.Server (exportMessages, importMessages)
|
||||
import Simplex.Messaging.Server (MessageStats (..), exportMessages, importMessages)
|
||||
import Simplex.Messaging.Server.MsgStore.Journal
|
||||
import Simplex.Messaging.Server.MsgStore.STM
|
||||
import Simplex.Messaging.Server.MsgStore.Types
|
||||
@@ -73,17 +73,20 @@ pattern Msg s <- Just Message {msgBody = MaxLenBS s}
|
||||
|
||||
deriving instance Eq MsgQueueState
|
||||
|
||||
deriving instance Eq JournalState
|
||||
deriving instance Eq (JournalState t)
|
||||
|
||||
deriving instance Eq (SJournalType t)
|
||||
|
||||
testGetQueue :: MsgStoreClass s => s -> IO ()
|
||||
testGetQueue st = do
|
||||
testGetQueue ms = do
|
||||
g <- C.newRandom
|
||||
rId <- EntityId <$> atomically (C.randomBytes 24 g)
|
||||
runRight_ $ do
|
||||
q <- getMsgQueue st rId
|
||||
Just (Message {msgId = mId1}, True) <- writeMsg q True =<< mkMessage "message 1"
|
||||
Just (Message {msgId = mId2}, False) <- writeMsg q True =<< mkMessage "message 2"
|
||||
Just (Message {msgId = mId3}, False) <- writeMsg q True =<< mkMessage "message 3"
|
||||
q <- getMsgQueue ms rId
|
||||
let write s = writeMsg ms q True =<< mkMessage s
|
||||
Just (Message {msgId = mId1}, True) <- write "message 1"
|
||||
Just (Message {msgId = mId2}, False) <- write "message 2"
|
||||
Just (Message {msgId = mId3}, False) <- write "message 3"
|
||||
Msg "message 1" <- tryPeekMsg q
|
||||
Msg "message 1" <- tryPeekMsg q
|
||||
Nothing <- tryDelMsg q mId2
|
||||
@@ -99,14 +102,14 @@ testGetQueue st = do
|
||||
Nothing <- tryDelMsg q mId2
|
||||
Nothing <- tryDelMsg q mId3
|
||||
Nothing <- tryPeekMsg q
|
||||
Just (Message {msgId = mId4}, True) <- writeMsg q True =<< mkMessage "message 4"
|
||||
Just (Message {msgId = mId4}, True) <- write "message 4"
|
||||
Msg "message 4" <- tryPeekMsg q
|
||||
Just (Message {msgId = mId5}, False) <- writeMsg q True =<< mkMessage "message 5"
|
||||
Just (Message {msgId = mId5}, False) <- write "message 5"
|
||||
(Nothing, Msg "message 4") <- tryDelPeekMsg q mId3
|
||||
(Msg "message 4", Msg "message 5") <- tryDelPeekMsg q mId4
|
||||
Just (Message {msgId = mId6}, False) <- writeMsg q True =<< mkMessage "message 6"
|
||||
Just (Message {msgId = mId7}, False) <- writeMsg q True =<< mkMessage "message 7"
|
||||
Nothing <- writeMsg q True =<< mkMessage "message 8"
|
||||
Just (Message {msgId = mId6}, False) <- write "message 6"
|
||||
Just (Message {msgId = mId7}, False) <- write "message 7"
|
||||
Nothing <- write "message 8"
|
||||
Msg "message 5" <- tryPeekMsg q
|
||||
(Nothing, Msg "message 5") <- tryDelPeekMsg q mId4
|
||||
(Msg "message 5", Msg "message 6") <- tryDelPeekMsg q mId5
|
||||
@@ -115,65 +118,69 @@ testGetQueue st = do
|
||||
(Just MessageQuota {}, Nothing) <- tryDelPeekMsg q mId8
|
||||
(Nothing, Nothing) <- tryDelPeekMsg q mId8
|
||||
pure ()
|
||||
delMsgQueue st rId
|
||||
delMsgQueue ms rId
|
||||
|
||||
testChangeReadJournal :: MsgStoreClass s => s -> IO ()
|
||||
testChangeReadJournal st = do
|
||||
testChangeReadJournal ms = do
|
||||
g <- C.newRandom
|
||||
rId <- EntityId <$> atomically (C.randomBytes 24 g)
|
||||
runRight_ $ do
|
||||
q <- getMsgQueue st rId
|
||||
Just (Message {msgId = mId1}, True) <- writeMsg q True =<< mkMessage "message 1"
|
||||
q <- getMsgQueue ms rId
|
||||
let write s = writeMsg ms q True =<< mkMessage s
|
||||
Just (Message {msgId = mId1}, True) <- write "message 1"
|
||||
(Msg "message 1", Nothing) <- tryDelPeekMsg q mId1
|
||||
Just (Message {msgId = mId2}, True) <- writeMsg q True =<< mkMessage "message 2"
|
||||
Just (Message {msgId = mId2}, True) <- write "message 2"
|
||||
(Msg "message 2", Nothing) <- tryDelPeekMsg q mId2
|
||||
Just (Message {msgId = mId3}, True) <- writeMsg q True =<< mkMessage "message 3"
|
||||
Just (Message {msgId = mId3}, True) <- write "message 3"
|
||||
(Msg "message 3", Nothing) <- tryDelPeekMsg q mId3
|
||||
Just (Message {msgId = mId4}, True) <- writeMsg q True =<< mkMessage "message 4"
|
||||
Just (Message {msgId = mId4}, True) <- write "message 4"
|
||||
(Msg "message 4", Nothing) <- tryDelPeekMsg q mId4
|
||||
Just (Message {msgId = mId5}, True) <- writeMsg q True =<< mkMessage "message 5"
|
||||
Just (Message {msgId = mId5}, True) <- write "message 5"
|
||||
(Msg "message 5", Nothing) <- tryDelPeekMsg q mId5
|
||||
pure ()
|
||||
delMsgQueue st rId
|
||||
delMsgQueue ms rId
|
||||
|
||||
testExportImportStore :: JournalMsgStore -> IO ()
|
||||
testExportImportStore st = do
|
||||
testExportImportStore ms = do
|
||||
g <- C.newRandom
|
||||
rId1 <- EntityId <$> atomically (C.randomBytes 24 g)
|
||||
rId2 <- EntityId <$> atomically (C.randomBytes 24 g)
|
||||
runRight_ $ do
|
||||
q1 <- getMsgQueue st rId1
|
||||
Just (Message {}, True) <- writeMsg q1 True =<< mkMessage "message 1"
|
||||
Just (Message {}, False) <- writeMsg q1 True =<< mkMessage "message 2"
|
||||
q2 <- getMsgQueue st rId2
|
||||
Just (Message {}, True) <- writeMsg q2 True =<< mkMessage "message 3"
|
||||
Just (Message {}, False) <- writeMsg q2 True =<< mkMessage "message 4"
|
||||
Just (Message {}, False) <- writeMsg q2 True =<< mkMessage "message 5"
|
||||
Nothing <- writeMsg q2 True =<< mkMessage "message 6"
|
||||
let write q s = writeMsg ms q True =<< mkMessage s
|
||||
q1 <- getMsgQueue ms rId1
|
||||
Just (Message {}, True) <- write q1 "message 1"
|
||||
Just (Message {}, False) <- write q1 "message 2"
|
||||
q2 <- getMsgQueue ms rId2
|
||||
Just (Message {}, True) <- write q2 "message 3"
|
||||
Just (Message {}, False) <- write q2 "message 4"
|
||||
Just (Message {}, False) <- write q2 "message 5"
|
||||
Nothing <- write q2 "message 6"
|
||||
pure ()
|
||||
length <$> listDirectory (msgQueueDirectory st rId1) `shouldReturn` 2
|
||||
length <$> listDirectory (msgQueueDirectory st rId2) `shouldReturn` 2
|
||||
exportMessages st testStoreMsgsFile $ getQueueMessages False
|
||||
length <$> listDirectory (msgQueueDirectory ms rId1) `shouldReturn` 2
|
||||
length <$> listDirectory (msgQueueDirectory ms rId2) `shouldReturn` 2
|
||||
exportMessages ms testStoreMsgsFile $ getQueueMessages False
|
||||
let cfg = (testJournalStoreCfg :: JournalStoreConfig) {storePath = testStoreMsgsDir2}
|
||||
st' <- newMsgStore cfg
|
||||
0 <- importMessages st' testStoreMsgsFile Nothing
|
||||
length <$> listDirectory (msgQueueDirectory st rId1) `shouldReturn` 2
|
||||
length <$> listDirectory (msgQueueDirectory st rId2) `shouldReturn` 3 -- state file is backed up
|
||||
exportMessages st' testStoreMsgsFile2 $ getQueueMessages False
|
||||
ms' <- newMsgStore cfg
|
||||
MessageStats {storedMsgsCount = 6, expiredMsgsCount = 0, storedQueues = 2} <-
|
||||
importMessages ms' testStoreMsgsFile Nothing
|
||||
length <$> listDirectory (msgQueueDirectory ms rId1) `shouldReturn` 2
|
||||
length <$> listDirectory (msgQueueDirectory ms rId2) `shouldReturn` 3 -- state file is backed up
|
||||
exportMessages ms' testStoreMsgsFile2 $ getQueueMessages False
|
||||
(B.readFile testStoreMsgsFile2 `shouldReturn`) =<< B.readFile (testStoreMsgsFile <> ".bak")
|
||||
stmStore <- newMsgStore testSMTStoreConfig
|
||||
0 <- importMessages stmStore testStoreMsgsFile2 Nothing
|
||||
MessageStats {storedMsgsCount = 6, expiredMsgsCount = 0, storedQueues = 2} <-
|
||||
importMessages stmStore testStoreMsgsFile2 Nothing
|
||||
exportMessages stmStore testStoreMsgsFile $ getQueueMessages False
|
||||
(B.sort <$> B.readFile testStoreMsgsFile `shouldReturn`) =<< (B.sort <$> B.readFile (testStoreMsgsFile2 <> ".bak"))
|
||||
|
||||
testQueueState :: JournalMsgStore -> IO ()
|
||||
testQueueState st = do
|
||||
testQueueState ms = do
|
||||
g <- C.newRandom
|
||||
rId <- EntityId <$> atomically (C.randomBytes 24 g)
|
||||
let dir = msgQueueDirectory st rId
|
||||
let dir = msgQueueDirectory ms rId
|
||||
statePath = dir </> (queueLogFileName <> logFileExt)
|
||||
createDirectoryIfMissing True dir
|
||||
state <- newMsgQueueState <$> newJournalId (random st)
|
||||
state <- newMsgQueueState <$> newJournalId (random ms)
|
||||
withFile statePath WriteMode (`logQueueState` state)
|
||||
length . lines <$> readFile statePath `shouldReturn` 1
|
||||
readQueueState statePath `shouldReturn` state
|
||||
@@ -208,7 +215,7 @@ testQueueState st = do
|
||||
length . lines <$> readFile statePath `shouldReturn` 1
|
||||
where
|
||||
readQueueState statePath = do
|
||||
(state, h) <- readWriteQueueState st statePath
|
||||
(state, h) <- readWriteQueueState ms statePath
|
||||
hClose h
|
||||
pure state
|
||||
corruptFile f = do
|
||||
|
||||
Reference in new issue
Block a user