smp server, journal store: check file size on append, validate queue state (#1377)

* smp server: journal store queue state validation

* validate state on update, refactor

* handle message parsing errors

* typo

* fix test, throw exception when file is smaller than expected

* fix

* fix test in ghc 8.10.7

src/Simplex/Messaging/Server.hs

@@ -93,7 +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.Journal (JournalMsgQueue (..), JMQueue (..), closeMsgQueue)
 import Simplex.Messaging.Server.MsgStore.STM
 import Simplex.Messaging.Server.MsgStore.Types
 import Simplex.Messaging.Server.NtfStore
@@ -1731,17 +1731,17 @@ saveServerMessages :: Bool -> M ()
 saveServerMessages drainMsgs =
   asks msgStore >>= \case
     AMS SMSMemory ms@STMMsgStore {storeConfig = STMStoreConfig {storePath}} -> case storePath of
-      Just f -> liftIO $ exportMessages ms f $ getQueueMessages drainMsgs
+      Just f -> liftIO $ exportMessages ms f drainMsgs
       Nothing -> logInfo "undelivered messages are not saved"
     AMS SMSJournal _ -> logInfo "closed journal message storage"
 
-exportMessages :: MsgStoreClass s => s -> FilePath -> (MsgQueue s -> IO [Message]) -> IO ()
-exportMessages ms f getMessages = do
+exportMessages :: MsgStoreClass s => s -> FilePath -> Bool -> IO ()
+exportMessages ms f drainMsgs = do
   logInfo $ "saving messages to file " <> T.pack f
   total <- liftIO $ withFile f WriteMode $ \h -> withAllMsgQueues ms (saveQueueMsgs h) 0
   logInfo $ "messages saved: " <> tshow total
   where
-    saveQueueMsgs h rId q acc = getMessages q >>= \msgs -> (acc + length msgs) <$ BLD.hPutBuilder h (encodeMessages rId msgs)
+    saveQueueMsgs h rId q acc = getQueueMessages drainMsgs 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 MessageStats
@@ -1768,7 +1768,7 @@ restoreServerMessages = do
             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
+                logInfo $ "failed expiring messages in queue " <> T.pack (queueDirectory $ queue q) <> ": " <> tshow e
                 exitFailure
             where
               expireQueue = do

src/Simplex/Messaging/Server/Main.hs

@@ -44,7 +44,7 @@ import Simplex.Messaging.Server.CLI
 import Simplex.Messaging.Server.Env.STM
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.Information
-import Simplex.Messaging.Server.MsgStore.Journal (JournalStoreConfig (..), getQueueMessages)
+import Simplex.Messaging.Server.MsgStore.Journal (JournalStoreConfig (..))
 import Simplex.Messaging.Server.MsgStore.Types (AMSType (..), SMSType (..), newMsgStore)
 import Simplex.Messaging.Transport (batchCmdsSMPVersion, sendingProxySMPVersion, simplexMQVersion, supportedServerSMPRelayVRange)
 import Simplex.Messaging.Transport.Client (SocksProxy, TransportHost (..), defaultSocksProxy)
@@ -115,7 +115,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
                 ("WARNING: journal directory " <> storeMsgsJournalDir <> " will be exported to message log file " <> storeMsgsFilePath)
                 "Journal not exported"
               ms <- newJournalMsgStore
-              exportMessages ms storeMsgsFilePath $ getQueueMessages False
+              exportMessages ms storeMsgsFilePath False
               putStrLn "Export completed"
               putStrLn $ case readMsgStoreType ini of
                 Right (AMSType SMSMemory) -> "store_messages set to `memory`"

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

@@ -15,7 +15,8 @@
 
 module Simplex.Messaging.Server.MsgStore.Journal
   ( JournalMsgStore (msgQueues, random),
-    JournalMsgQueue (queueDirectory),
+    JournalMsgQueue (queue),
+    JMQueue (queueDirectory, statePath),
     JournalStoreConfig (..),
     getQueueMessages,
     closeMsgQueue,
@@ -43,6 +44,7 @@ import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy.Char8 as LB
 import Data.Functor (($>))
+import Data.Int (Int64)
 import Data.Maybe (catMaybes, fromMaybe)
 import qualified Data.Text as T
 import Data.Time.Clock (getCurrentTime)
@@ -81,15 +83,20 @@ data JournalStoreConfig = JournalStoreConfig
     maxStateLines :: Int
   }
 
-data JournalMsgQueue = JournalMsgQueue
+data JMQueue = JMQueue
   { queueDirectory :: FilePath,
     queueLock :: Lock,
+    statePath :: FilePath
+  }
+
+data JournalMsgQueue = JournalMsgQueue
+  { queue :: JMQueue,
     state :: TVar MsgQueueState,
     -- tipMsg contains last message and length incl. newline
     -- Nothing - unknown, Just Nothing - empty queue.
     -- It  prevents reading each message twice,
     -- and reading it after it was just written.
-    tipMsg :: TVar (Maybe (Maybe (Message, Int))),
+    tipMsg :: TVar (Maybe (Maybe (Message, Int64))),
     handles :: TVar (Maybe MsgQueueHandles)
   }
 
@@ -112,8 +119,8 @@ data JournalState t = JournalState
     journalId :: ByteString,
     msgPos :: Int,
     msgCount :: Int,
-    bytePos :: Int,
-    byteCount :: Int
+    bytePos :: Int64,
+    byteCount :: Int64
   }
   deriving (Show)
 
@@ -141,15 +148,7 @@ newMsgQueueState journalId =
     }
 
 newJournalState :: JournalTypeI t => ByteString -> JournalState t
-newJournalState journalId =
-  JournalState
-    { journalType = sJournalType,
-      journalId,
-      msgPos = 0,
-      msgCount = 0,
-      bytePos = 0,
-      byteCount = 0
-    }
+newJournalState journalId = JournalState sJournalType journalId 0 0 0 0
 
 journalFilePath :: FilePath -> ByteString -> FilePath
 journalFilePath dir journalId = dir </> (msgLogFileName <> "." <> B.unpack journalId <> logFileExt)
@@ -171,13 +170,16 @@ instance StrEncoding MsgQueueState where
 
 instance JournalTypeI t => StrEncoding (JournalState t) where
   strEncode JournalState {journalId, msgPos, msgCount, bytePos, byteCount} =
-    B.concat [journalId, ",", strEncodeList [msgPos, msgCount, bytePos, byteCount]]
+    B.intercalate "," [journalId, e msgPos, e msgCount, e bytePos, e byteCount]
+    where
+      e :: StrEncoding a => a -> ByteString
+      e = strEncode
   strP = do
     journalId <- A.takeTill (== ',')
-    A.char ',' *> strListP >>= \case
-      [msgPos, msgCount, bytePos, byteCount] ->
-        pure JournalState {journalType = sJournalType, journalId, msgPos, msgCount, bytePos, byteCount}
-      _ -> fail "bad JournalState"
+    JournalState sJournalType journalId <$> i <*> i <*> i <*> i
+    where
+      i :: Integral a => A.Parser a
+      i = A.char ',' *> A.decimal
 
 queueLogFileName :: String
 queueLogFileName = "queue_state"
@@ -210,6 +212,8 @@ instance MsgStoreClass JournalMsgStore where
 
   -- 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.
+  -- TODO this function requires case-sensitive file system, because it uses queue directory as recipient ID.
+  -- It can be made to support case-insensite FS by supporting more than one queue per directory, by getting recipient ID from state file name.
   withAllMsgQueues :: JournalMsgStore -> (RecipientId -> JournalMsgQueue -> a -> IO a) -> a -> IO a
   withAllMsgQueues ms@JournalMsgStore {config} action res = ifM (doesDirectoryExist storePath) processStore (pure res)
     where
@@ -223,9 +227,11 @@ instance MsgStoreClass JournalMsgStore where
         putStrLn ""
         pure res'
       JournalStoreConfig {storePath, pathParts} = config
-      processQueue lock count acc (i :: Int, (queueId, dir)) = do
+      processQueue queueLock count acc (i :: Int, (queueId, dir)) = do
         when (i `mod` 100 == 0) $ progress i count
-        q <- openMsgQueue ms dir lock
+        let statePath = dir </> queueLogFileName <> "." <> queueId <> logFileExt
+            queue = JMQueue {queueDirectory = dir, queueLock, statePath}
+        q <- openMsgQueue ms queue
         acc' <- case strDecode $ B.pack queueId of
           Right rId -> action rId q acc
           Left e -> do
@@ -261,18 +267,20 @@ instance MsgStoreClass JournalMsgStore where
       TM.lookupIO rId msgQueues >>= maybe newQ pure
     where
       newQ = do
-        let dir = msgQueueDirectory ms rId
         queueLock <- atomically $ getMapLock queueLocks rId
-        q <- ifM (doesDirectoryExist dir) (openMsgQueue ms dir queueLock) (createQ dir queueLock)
+        let dir = msgQueueDirectory ms rId
+            statePath = dir </> (queueLogFileName <> "." <> B.unpack (strEncode rId) <> logFileExt)
+            queue = JMQueue {queueDirectory = dir, queueLock, statePath}
+        q <- ifM (doesDirectoryExist dir) (openMsgQueue ms queue) (createQ queue)
         atomically $ TM.insert rId q msgQueues
         pure q
         where
-          createQ :: FilePath -> Lock -> IO JournalMsgQueue
-          createQ dir queueLock = do
+          createQ :: JMQueue -> IO JournalMsgQueue
+          createQ queue = do
             -- folder and files are not created here,
             -- to avoid file IO for queues without messages during subscription
             journalId <- newJournalId random
-            mkJournalQueue dir queueLock (newMsgQueueState journalId, Nothing)
+            mkJournalQueue queue (newMsgQueueState journalId, Nothing)
 
   delMsgQueue :: JournalMsgStore -> RecipientId -> IO ()
   delMsgQueue ms rId = withLockMap (queueLocks ms) rId "delMsgQueue" $ do
@@ -292,16 +300,13 @@ instance MsgStoreClass JournalMsgStore where
       getMsg msgs hs = chooseReadJournal q drainMsgs hs >>= maybe (pure msgs) readMsg
         where
           readMsg (rs, h) = do
-            -- TODO handle errors
-            s <- hGetLineAt h $ bytePos rs
-            -- TODO handle errors
-            Right msg <- pure $ strDecode s
-            updateReadPos q drainMsgs (B.length s + 1) hs -- 1 is to account for new line
+            (msg, len) <- hGetMsgAt h $ bytePos rs
+            updateReadPos q drainMsgs len hs
             (msg :) <$> getMsg msgs hs
 
   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
+  writeMsg ms q@JournalMsgQueue {queue = JMQueue {queueDirectory, queueLock, statePath}, handles} logState !msg =
+    tryStore "writeMsg" $ withLock' queueLock "writeMsg" $ do
       st@MsgQueueState {canWrite, size} <- readTVarIO (state q)
       let empty = size == 0
       if canWrite || empty
@@ -316,22 +321,22 @@ instance MsgStoreClass JournalMsgStore where
       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'
-            msgLen = B.length msgStr
+            msgLen = fromIntegral $ B.length msgStr
         hs <- maybe createQueueDir pure =<< readTVarIO handles
         (ws, wh) <- case writeHandle hs of
           Nothing | msgCount writeState >= maxMsgCount -> switchWriteJournal hs
           wh_ -> pure (writeState, fromMaybe (readHandle hs) wh_)
-        let msgCount' = msgCount ws + 1
-            ws' = ws {msgPos = msgPos ws + 1, msgCount = msgCount', bytePos = bytePos ws + msgLen}
-            rs' = if journalId ws == journalId rs then rs {msgCount = msgCount'} else rs
+        let msgPos' = msgPos ws + 1
+            bytePos' = bytePos ws + msgLen
+            ws' = ws {msgPos = msgPos', msgCount = msgPos', bytePos = bytePos', byteCount = bytePos'}
+            rs' = if journalId ws == journalId rs then rs {msgCount = msgPos', byteCount = bytePos'} else rs
             !st' = st {writeState = ws', readState = rs', canWrite = canWrt', size = size + 1}
         when (size == 0) $ atomically $ writeTVar (tipMsg q) $ Just (Just (msg, msgLen))
-        hAppend wh msgStr
+        hAppend wh (bytePos ws) msgStr
         updateQueueState q logState hs st'
         where
           createQueueDir = do
             createDirectoryIfMissing True queueDirectory
-            let statePath = queueDirectory </> (queueLogFileName <> logFileExt)
             sh <- openFile statePath AppendMode
             B.hPutStr sh ""
             rh <- createNewJournal queueDirectory $ journalId rs
@@ -354,11 +359,8 @@ instance MsgStoreClass JournalMsgStore where
       peekMsg (rs, h) = readTVarIO tipMsg >>= maybe readMsg (pure . fmap fst)
         where
           readMsg = do
-            -- TODO handle errors
-            s <- hGetLineAt h $ bytePos rs
-            -- TODO handle errors
-            Right msg <- pure $ strDecode s
-            atomically $ writeTVar tipMsg $ Just (Just (msg, B.length s + 1)) -- 1 is to account for new line
+            ml@(msg, _) <- hGetMsgAt h $ bytePos rs
+            atomically $ writeTVar tipMsg $ Just (Just ml)
             pure $ Just msg
 
   tryDeleteMsg_ :: JournalMsgQueue -> StoreIO ()
@@ -370,7 +372,7 @@ instance MsgStoreClass JournalMsgStore where
         $>>= \hs -> updateReadPos q True len hs $> Just ()
 
   isolateQueue :: JournalMsgQueue -> String -> StoreIO a -> ExceptT ErrorType IO a
-  isolateQueue mq op (StoreIO a) = tryStore op $ withLock' (queueLock mq) op $ a
+  isolateQueue q op (StoreIO a) = tryStore op $ withLock' (queueLock $ queue q) op $ a
 
 tryStore :: String -> IO a -> ExceptT ErrorType IO a
 tryStore op a =
@@ -379,29 +381,21 @@ tryStore op a =
       let e' = op <> " " <> show e
        in logError ("STORE ERROR " <> T.pack e') $> Left (STORE e')
 
-openMsgQueue :: JournalMsgStore -> FilePath -> Lock -> IO JournalMsgQueue
-openMsgQueue ms dir queueLock = do
-  let statePath = dir </> (queueLogFileName <> logFileExt)
+openMsgQueue :: JournalMsgStore -> JMQueue -> IO JournalMsgQueue
+openMsgQueue ms q@JMQueue {queueDirectory = dir, statePath} = do
   (st, sh) <- readWriteQueueState ms statePath
   (st', rh, wh_) <- openJournals dir st
   let hs = MsgQueueHandles {stateHandle = sh, readHandle = rh, writeHandle = wh_}
-  mkJournalQueue dir queueLock (st', Just hs)
+  mkJournalQueue q (st', Just hs)
 
-mkJournalQueue :: FilePath -> Lock -> (MsgQueueState, Maybe MsgQueueHandles) -> IO JournalMsgQueue
-mkJournalQueue dir queueLock (st, hs_) = do
+mkJournalQueue :: JMQueue -> (MsgQueueState, Maybe MsgQueueHandles) -> IO JournalMsgQueue
+mkJournalQueue queue (st, hs_) = do
   state <- newTVarIO st
   tipMsg <- newTVarIO Nothing
   handles <- newTVarIO hs_
   -- using the same queue lock which is currently locked,
   -- to avoid map lookup on queue operations
-  pure
-    JournalMsgQueue
-      { queueDirectory = dir,
-        queueLock,
-        state,
-        tipMsg,
-        handles
-      }
+  pure JournalMsgQueue {queue, state, tipMsg, handles}
 
 chooseReadJournal :: JournalMsgQueue -> Bool -> MsgQueueHandles -> IO (Maybe (JournalState 'JTRead, Handle))
 chooseReadJournal q log' hs = do
@@ -411,8 +405,8 @@ chooseReadJournal q log' hs = do
       -- switching to write journal
       atomically $ writeTVar (handles q) $ Just hs {readHandle = wh, writeHandle = Nothing}
       hClose $ readHandle hs
-      removeJournal (queueDirectory q) rs
-      let !rs' = (newJournalState $ journalId ws) {msgCount = msgCount ws}
+      removeJournal (queueDirectory $ queue q) rs
+      let !rs' = (newJournalState $ journalId ws) {msgCount = msgCount ws, byteCount = byteCount ws}
           !st' = st {readState = rs'}
       updateQueueState q log' hs st'
       pure $ Just (rs', wh)
@@ -421,13 +415,14 @@ chooseReadJournal q log' hs = do
 
 updateQueueState :: JournalMsgQueue -> Bool -> MsgQueueHandles -> MsgQueueState -> IO ()
 updateQueueState q log' hs st = do
+  unless (validQueueState st) $ E.throwIO $ userError $ "updating to invalid state: " <> show 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'
 
-updateReadPos :: JournalMsgQueue -> Bool -> Int -> MsgQueueHandles -> IO ()
+updateReadPos :: JournalMsgQueue -> Bool -> Int64 -> MsgQueueHandles -> IO ()
 updateReadPos q log' len hs = do
   st@MsgQueueState {readState = rs, size} <- readTVarIO (state q)
   let JournalState {msgPos, bytePos} = rs
@@ -471,32 +466,39 @@ openJournals dir st@MsgQueueState {readState = rs, writeState = ws} = do
       pure (st', rh, Nothing)
     Right rh
       | rjId == wjId -> do
-          st' <- fixWriteFileSize rh
-          pure (st', rh, Nothing)
-      | otherwise ->
+          fixFileSize rh $ bytePos ws
+          pure (st, rh, Nothing)
+      | otherwise -> do
+          fixFileSize rh $ byteCount rs
           openJournal ws >>= \case
             Left path -> do
               logError $ "STORE ERROR no write file " <> T.pack path <> ", creating new file"
               wh <- createNewJournal dir wjId
-              let size' = msgCount rs + msgPos rs
+              let size' = msgCount rs - msgPos rs
                   st' = st {writeState = newJournalState wjId, size = size'} -- we don't amend canWrite to trigger QCONT
               pure (st', rh, Just wh)
             Right wh -> do
-              st' <- fixWriteFileSize wh
-              pure (st', rh, Just wh)
+              fixFileSize wh $ bytePos ws
+              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
-      if
-        | sz' > sz -> logWarn "STORE WARNING" >> IO.hSetFileSize h sz $> st
-        | sz' == sz -> pure st
-        | otherwise -> pure st -- TODO re-read file to recover what is possible ???
+
+fixFileSize :: Handle -> Int64 -> IO ()
+fixFileSize h pos = do
+  let pos' = fromIntegral pos
+  size <- IO.hFileSize h
+  if
+    | size > pos' -> do
+        logWarn $ "STORE WARNING truncating file size from " <> tshow size <> " to " <> tshow pos
+        IO.hSetFileSize h pos'
+    | size < pos' ->
+        -- From code logic this can't happen.
+        E.throwIO $ userError $ "file size " <> show size <> " is smaller than position " <> show pos
+    | otherwise -> pure ()
 
 removeJournal :: FilePath -> JournalState t -> IO ()
 removeJournal dir JournalState {journalId} = do
@@ -517,7 +519,10 @@ readWriteQueueState JournalMsgStore {random, config} statePath =
       ls <- LB.lines <$> LB.readFile statePath
       case ls of
         [] -> writeNewQueueState
-        _ -> useLastLine (length ls) True ls
+        _ -> do
+          r@(st, _) <- useLastLine (length ls) True ls
+          unless (validQueueState st) $ E.throwIO $ userError $ "read invalid invalid: " <> show st
+          pure r
     writeNewQueueState = do
       logWarn $ "STORE WARNING: empty queue state in " <> T.pack statePath <> ", initialized"
       st <- newMsgQueueState <$> newJournalId random
@@ -539,9 +544,7 @@ readWriteQueueState JournalMsgStore {random, config} statePath =
             ls' -> do
               logWarn $ "STORE WARNING: invalid last line in queue state " <> T.pack statePath <> ", using the previous line"
               useLastLine len False ls'
-        | otherwise -> do
-            logError $ "STORE ERROR invalid queue state in " <> T.pack statePath <> ": " <> tshow e
-            E.throwIO $ userError $ "Error reading queue state " <> statePath <> ": " <> show e
+        | otherwise -> E.throwIO $ userError $ "reading queue state " <> statePath <> ": " <> show e
     backupWriteQueueState st = do
       -- State backup is made in two steps to mitigate the crash during the backup.
       -- Temporary backup file will be used when it is present.
@@ -555,6 +558,25 @@ readWriteQueueState JournalMsgStore {random, config} statePath =
       logQueueState sh st
       pure (st, sh)
 
+validQueueState :: MsgQueueState -> Bool
+validQueueState MsgQueueState {readState = rs, writeState = ws, size}
+  | journalId rs == journalId ws =
+      alwaysValid
+        && msgPos rs <= msgPos ws
+        && msgCount rs == msgCount ws
+        && bytePos rs <= bytePos ws
+        && byteCount rs == byteCount ws
+        && size == msgCount rs - msgPos rs
+  | otherwise =
+      alwaysValid
+        && size == msgCount ws + msgCount rs - msgPos rs
+  where
+    alwaysValid =
+      msgPos rs <= msgCount rs
+        && bytePos rs <= byteCount rs
+        && msgPos ws == msgCount ws
+        && bytePos ws == byteCount ws
+
 deleteMsgQueue_ :: JournalMsgStore -> RecipientId -> IO (Maybe (TVar MsgQueueState))
 deleteMsgQueue_ st rId =
   atomically (TM.lookupDelete rId (msgQueues st))
@@ -573,11 +595,21 @@ removeQueueDirectory st rId =
   let dir = msgQueueDirectory st rId
    in removePathForcibly dir `catchAny` (\e -> logError $ "STORE ERROR removeQueueDirectory " <> T.pack dir <> ": " <> tshow e)
 
-hAppend :: Handle -> ByteString -> IO ()
-hAppend h s = IO.hSeek h SeekFromEnd 0 >> B.hPutStr h s
+hAppend :: Handle -> Int64 -> ByteString -> IO ()
+hAppend h pos s = do
+  fixFileSize h pos
+  IO.hSeek h SeekFromEnd 0
+  B.hPutStr h s
 
-hGetLineAt :: Handle -> Int -> IO ByteString
-hGetLineAt h pos = IO.hSeek h AbsoluteSeek (fromIntegral pos) >> B.hGetLine h
+hGetMsgAt :: Handle -> Int64 -> IO (Message, Int64)
+hGetMsgAt h pos = do
+  IO.hSeek h AbsoluteSeek $ fromIntegral pos
+  s <- B.hGetLine h
+  case strDecode s of
+    Right !msg ->
+      let !len = fromIntegral (B.length s) + 1
+       in pure (msg, len)
+    Left e -> E.throwIO $ userError $ "Error parsing message: " <> e
 
 openFile :: FilePath -> IOMode -> IO Handle
 openFile f mode = do

tests/AgentTests/FunctionalAPITests.hs

@@ -1946,7 +1946,7 @@ testBatchedSubscriptions nCreate nDel t =
     runServers :: ExceptT AgentErrorType IO a -> IO a
     runServers a = do
       withSmpServerStoreLogOn t testPort $ \t1 -> do
-        res <- withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2} testPort2 $ \t2 ->
+        res <- withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2} testPort2 $ \t2 ->
           runRight a `finally` killThread t2
         killThread t1
         pure res

tests/AgentTests/NotificationTests.hs

@@ -54,7 +54,7 @@ import Data.Text.Encoding (encodeUtf8)
 import Database.SQLite.Simple.QQ (sql)
 import NtfClient
 import SMPAgentClient (agentCfg, initAgentServers, initAgentServers2, testDB, testDB2, testNtfServer, testNtfServer2)
-import SMPClient (cfg, cfgVPrev, testPort, testPort2, testStoreLogFile2, withSmpServer, withSmpServerConfigOn, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn)
+import SMPClient (cfg, cfgVPrev, testPort, testPort2, testStoreLogFile2, testStoreMsgsDir2, withSmpServer, withSmpServerConfigOn, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn)
 import Simplex.Messaging.Agent hiding (createConnection, joinConnection, sendMessage)
 import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestStep (..), withStore')
 import Simplex.Messaging.Agent.Env.SQLite (AgentConfig, Env (..), InitialAgentServers)
@@ -785,7 +785,7 @@ testNotificationsSMPRestartBatch n t apns =
     runServers :: ExceptT AgentErrorType IO a -> IO a
     runServers a = do
       withSmpServerStoreLogOn t testPort $ \t1 -> do
-        res <- withSmpServerConfigOn t (cfg :: ServerConfig) {storeLogFile = Just testStoreLogFile2} testPort2 $ \t2 ->
+        res <- withSmpServerConfigOn t (cfg :: ServerConfig) {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2} testPort2 $ \t2 ->
           runRight a `finally` killThread t2
         killThread t1
         pure res

tests/CoreTests/MsgStoreTests.hs

@@ -1,5 +1,6 @@
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE GADTs #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE PatternSynonyms #-}
@@ -22,6 +23,7 @@ import Simplex.Messaging.Crypto (pattern MaxLenBS)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Protocol (EntityId (..), Message (..), noMsgFlags)
 import Simplex.Messaging.Server (MessageStats (..), exportMessages, importMessages)
+import Simplex.Messaging.Server.Env.STM (journalMsgStoreDepth)
 import Simplex.Messaging.Server.MsgStore.Journal
 import Simplex.Messaging.Server.MsgStore.STM
 import Simplex.Messaging.Server.MsgStore.Types
@@ -55,7 +57,7 @@ testJournalStoreCfg :: JournalStoreConfig
 testJournalStoreCfg =
   JournalStoreConfig
     { storePath = testStoreMsgsDir,
-      pathParts = 4,
+      pathParts = journalMsgStoreDepth,
       quota = 3,
       maxMsgCount = 4,
       maxStateLines = 2
@@ -158,19 +160,19 @@ testExportImportStore ms = do
     pure ()
   length <$> listDirectory (msgQueueDirectory ms rId1) `shouldReturn` 2
   length <$> listDirectory (msgQueueDirectory ms rId2) `shouldReturn` 2
-  exportMessages ms testStoreMsgsFile $ getQueueMessages False
+  exportMessages ms testStoreMsgsFile False
   let cfg = (testJournalStoreCfg :: JournalStoreConfig) {storePath = testStoreMsgsDir2}
   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
+  exportMessages ms' testStoreMsgsFile2 False
   (B.readFile testStoreMsgsFile2 `shouldReturn`) =<< B.readFile (testStoreMsgsFile <> ".bak")
   stmStore <- newMsgStore testSMTStoreConfig
   MessageStats {storedMsgsCount = 6, expiredMsgsCount = 0, storedQueues = 2} <-
     importMessages stmStore testStoreMsgsFile2 Nothing
-  exportMessages stmStore testStoreMsgsFile $ getQueueMessages False
+  exportMessages stmStore testStoreMsgsFile False
   (B.sort <$> B.readFile testStoreMsgsFile `shouldReturn`) =<< (B.sort <$> B.readFile (testStoreMsgsFile2 <> ".bak"))
 
 testQueueState :: JournalMsgStore -> IO ()
@@ -186,13 +188,23 @@ testQueueState ms = do
   readQueueState statePath `shouldReturn` state
   length <$> listDirectory dir `shouldReturn` 1 -- no backup
 
-  let state1 = state {size = 1, writeState = (writeState state) {msgPos = 1, msgCount = 1, bytePos = 100}}
+  let state1 =
+        state
+          { size = 1,
+            readState = (readState state) {msgCount = 1, byteCount = 100},
+            writeState = (writeState state) {msgPos = 1, msgCount = 1, bytePos = 100, byteCount = 100}
+          }
   withFile statePath AppendMode (`logQueueState` state1)
   length . lines <$> readFile statePath `shouldReturn` 2
   readQueueState statePath `shouldReturn` state1
   length <$> listDirectory dir `shouldReturn` 1 -- no backup
 
-  let state2 = state {size = 2, writeState = (writeState state) {msgPos = 2, msgCount = 2, bytePos = 200}}
+  let state2 =
+        state
+          { size = 2,
+            readState = (readState state) {msgCount = 2, byteCount = 200},
+            writeState = (writeState state) {msgPos = 2, msgCount = 2, bytePos = 200, byteCount = 200}
+          }
   withFile statePath AppendMode (`logQueueState` state2)
   length . lines <$> readFile statePath `shouldReturn` 3
   copyFile statePath (statePath <> ".2")

tests/ServerTests.hs

@@ -16,6 +16,7 @@
 module ServerTests where
 
 import AgentTests.NotificationTests (removeFileIfExists)
+import CoreTests.MsgStoreTests (testJournalStoreCfg)
 import Control.Concurrent (ThreadId, killThread, threadDelay)
 import Control.Concurrent.STM
 import Control.Exception (SomeException, try)
@@ -35,9 +36,11 @@ import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (parseAll)
 import Simplex.Messaging.Protocol
+import Simplex.Messaging.Server (exportMessages)
 import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
 import Simplex.Messaging.Server.Expiration
-import Simplex.Messaging.Server.MsgStore.Types (AMSType (..), SMSType (..))
+import Simplex.Messaging.Server.MsgStore.Journal (JournalStoreConfig (..))
+import Simplex.Messaging.Server.MsgStore.Types (AMSType (..), SMSType (..), newMsgStore)
 import Simplex.Messaging.Server.Stats (PeriodStatsData (..), ServerStatsData (..))
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Util (whenM)
@@ -66,8 +69,8 @@ serverTests t@(ATransport t') = do
     describe "GET & SUB commands" $ testGetSubCommands t'
     describe "Exceeding queue quota" $ testExceedQueueQuota t'
   describe "Store log" $ testWithStoreLog t
-  xdescribe "Restore messages" $ testRestoreMessages t
-  xdescribe "Restore messages (old / v2)" $ testRestoreExpireMessages t
+  describe "Restore messages" $ testRestoreMessages t
+  describe "Restore messages (old / v2)" $ testRestoreExpireMessages t
   describe "Timing of AUTH error" $ testTiming t
   describe "Message notifications" $ testMessageNotifications t
   describe "Message expiration" $ do
@@ -780,6 +783,7 @@ testRestoreExpireMessages at@(ATransport t) =
       pure ()
 
     logSize testStoreLogFile `shouldReturn` 2
+    exportStoreMessages
     msgs <- B.readFile testStoreMsgsFile
     length (B.lines msgs) `shouldBe` 4
 
@@ -788,6 +792,7 @@ testRestoreExpireMessages at@(ATransport t) =
     withSmpServerConfigOn at cfg1 testPort . runTest t $ \_ -> pure ()
 
     logSize testStoreLogFile `shouldReturn` 1
+    exportStoreMessages
     msgs' <- B.readFile testStoreMsgsFile
     msgs' `shouldBe` msgs
 
@@ -797,12 +802,17 @@ testRestoreExpireMessages at@(ATransport t) =
 
     logSize testStoreLogFile `shouldReturn` 1
     -- two messages expired
-    -- msgs'' <- B.readFile testStoreMsgsFile
-    -- length (B.lines msgs'') `shouldBe` 2
-    -- B.lines msgs'' `shouldBe` drop 2 (B.lines msgs)
+    exportStoreMessages
+    msgs'' <- B.readFile testStoreMsgsFile
+    length (B.lines msgs'') `shouldBe` 2
+    B.lines msgs'' `shouldBe` drop 2 (B.lines msgs)
     Right ServerStatsData {_msgExpired} <- strDecode <$> B.readFile testServerStatsBackupFile
     _msgExpired `shouldBe` 2
   where
+    exportStoreMessages = do
+      ms <- newMsgStore testJournalStoreCfg {quota = 4}
+      removeFileIfExists testStoreMsgsFile
+      exportMessages ms testStoreMsgsFile False
     runTest :: Transport c => TProxy c -> (THandleSMP c 'TClient -> IO ()) -> ThreadId -> Expectation
     runTest _ test' server = do
       testSMPClient test' `shouldReturn` ()

tests/Test.hs

@@ -61,7 +61,7 @@ main = do
           describe "TRcvQueues tests" tRcvQueuesTests
           describe "Util tests" utilTests
         describe "SMP server via TLS" $ serverTests (transport @TLS)
-        describe "SMP server via WebSockets" $ serverTests (transport @WS)
+        xdescribe "SMP server via WebSockets" $ serverTests (transport @WS)
         describe "Notifications server" $ ntfServerTests (transport @TLS)
         describe "SMP client agent" $ agentTests (transport @TLS)
         describe "SMP proxy" smpProxyTests