5.1.1

* server: expire messages when restoring them

* add test

* specify constructor

Co-authored-by: spaced4ndy <[email protected]>

* improve test

---------

Co-authored-by: spaced4ndy <[email protected]>

package.yaml

@@ -1,5 +1,5 @@
 name: simplexmq
-version: 5.1.0
+version: 5.1.1
 synopsis: SimpleXMQ message broker
 description: |
   This package includes <./docs/Simplex-Messaging-Server.html server>,

simplexmq.cabal

@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack
 
 name:           simplexmq
-version:        5.1.0
+version:        5.1.1
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -81,6 +81,7 @@ library
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230320_retry_state
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230401_snd_files
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230510_files_pending_replicas_indexes
+      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230516_encrypted_rcv_message_hashes
       Simplex.Messaging.Agent.TAsyncs
       Simplex.Messaging.Agent.TRcvQueues
       Simplex.Messaging.Client

src/Simplex/FileTransfer/Agent.hs

@@ -415,7 +415,7 @@ runXFTPSndPrepareWorker c doWork = do
         createChunk :: Int -> SndFileChunk -> m ()
         createChunk numRecipients' ch = do
           atomically $ assertAgentForeground c
-          (replica, ProtoServerWithAuth srv _) <- agentOperationBracket c AOSndNetwork throwWhenInactive tryCreate
+          (replica, ProtoServerWithAuth srv _) <- tryCreate
           withStore' c $ \db -> createSndFileReplica db ch replica
           addXFTPSndWorker c $ Just srv
           where
@@ -445,7 +445,7 @@ runXFTPSndWorker c srv doWork = do
   forever $ do
     void . atomically $ readTMVar doWork
     atomically $ assertAgentForeground c
-    agentOperationBracket c AOSndNetwork throwWhenInactive runXFTPOperation
+    runXFTPOperation
   where
     noWorkToDo = void . atomically $ tryTakeTMVar doWork
     runXFTPOperation :: m ()

src/Simplex/Messaging/Agent.hs

@@ -1601,6 +1601,7 @@ cleanupManager c@AgentClient {subQ} = do
   forever $ do
     void . runExceptT $ do
       deleteConns `catchError` (notify "" . ERR)
+      deleteRcvMsgHashes `catchError` (notify "" . ERR)
       deleteRcvFilesExpired `catchError` (notify "" . RFERR)
       deleteRcvFilesDeleted `catchError` (notify "" . RFERR)
       deleteRcvFilesTmpPaths `catchError` (notify "" . RFERR)
@@ -1614,6 +1615,9 @@ cleanupManager c@AgentClient {subQ} = do
       withLock (deleteLock c) "cleanupManager" $ do
         void $ withStore' c getDeletedConnIds >>= deleteDeletedConns c
         withStore' c deleteUsersWithoutConns >>= mapM_ (notify "" . DEL_USER)
+    deleteRcvMsgHashes = do
+      rcvMsgHashesTTL <- asks $ rcvMsgHashesTTL . config
+      withStore' c (`deleteRcvMsgHashesExpired` rcvMsgHashesTTL)
     deleteRcvFilesExpired = do
       rcvFilesTTL <- asks $ rcvFilesTTL . config
       rcvExpired <- withStore' c (`getRcvFilesExpired` rcvFilesTTL)
@@ -1697,7 +1701,8 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
                               enqueueCommand c "" connId (Just server) $ AInternalCommand $ ICQDelete rcvId
                             _ -> notify . ERR . AGENT $ A_QUEUE "replaced RcvQueue not found in connection"
                         _ -> pure ()
-                      tryError agentClientMsg >>= \case
+                      let encryptedMsgHash = C.sha256Hash encAgentMsg
+                      tryError (agentClientMsg encryptedMsgHash) >>= \case
                         Right (Just (msgId, msgMeta, aMessage)) -> case aMessage of
                           HELLO -> helloMsg >> ackDel msgId
                           REPLY cReq -> replyMsg cReq >> ackDel msgId
@@ -1728,11 +1733,15 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
                                     logServer "<--" c srv rId "MSG <MSG>"
                                     notify $ MSG msgMeta msgFlags body
                                   _ -> pure ()
-                            _ -> throwError e
-                        Left e -> throwError e
+                            _ -> checkDuplicateHash e encryptedMsgHash
+                        Left e -> checkDuplicateHash e encryptedMsgHash
                       where
-                        agentClientMsg :: m (Maybe (InternalId, MsgMeta, AMessage))
-                        agentClientMsg = withStore c $ \db -> runExceptT $ do
+                        checkDuplicateHash :: AgentErrorType -> ByteString -> m ()
+                        checkDuplicateHash e encryptedMsgHash =
+                          unlessM (withStore' c $ \db -> checkRcvMsgHashExists db connId encryptedMsgHash) $
+                            throwError e
+                        agentClientMsg :: ByteString -> m (Maybe (InternalId, MsgMeta, AMessage))
+                        agentClientMsg encryptedMsgHash = withStore c $ \db -> runExceptT $ do
                           agentMsgBody <- agentRatchetDecrypt db connId encAgentMsg
                           liftEither (parse smpP (SEAgentError $ AGENT A_MESSAGE) agentMsgBody) >>= \case
                             agentMsg@(AgentMessage APrivHeader {sndMsgId, prevMsgHash} aMessage) -> do
@@ -1744,7 +1753,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
                                   recipient = (unId internalId, internalTs)
                                   broker = (srvMsgId, systemToUTCTime srvTs)
                                   msgMeta = MsgMeta {integrity, recipient, broker, sndMsgId}
-                                  rcvMsg = RcvMsgData {msgMeta, msgType, msgFlags, msgBody = agentMsgBody, internalRcvId, internalHash, externalPrevSndHash = prevMsgHash}
+                                  rcvMsg = RcvMsgData {msgMeta, msgType, msgFlags, msgBody = agentMsgBody, internalRcvId, internalHash, externalPrevSndHash = prevMsgHash, encryptedMsgHash}
                               liftIO $ createRcvMsg db connId rq rcvMsg
                               pure $ Just (internalId, msgMeta, aMessage)
                             _ -> pure Nothing

src/Simplex/Messaging/Agent/Client.hs

@@ -115,7 +115,6 @@ import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Either (lefts, partitionEithers)
 import Data.Functor (($>))
-import Data.Int (Int64)
 import Data.List (deleteFirstsBy, foldl', partition, (\\))
 import Data.List.NonEmpty (NonEmpty (..), (<|))
 import qualified Data.List.NonEmpty as L

src/Simplex/Messaging/Agent/Env/SQLite.hs

@@ -82,6 +82,7 @@ data AgentConfig = AgentConfig
     helloTimeout :: NominalDiffTime,
     initialCleanupDelay :: Int64,
     cleanupInterval :: Int64,
+    rcvMsgHashesTTL :: NominalDiffTime,
     rcvFilesTTL :: NominalDiffTime,
     sndFilesTTL :: NominalDiffTime,
     xftpNotifyErrsOnRetry :: Bool,
@@ -145,6 +146,7 @@ defaultAgentConfig =
       helloTimeout = 2 * nominalDay,
       initialCleanupDelay = 30 * 1000000, -- 30 seconds
       cleanupInterval = 30 * 60 * 1000000, -- 30 minutes
+      rcvMsgHashesTTL = 30 * nominalDay,
       rcvFilesTTL = 2 * nominalDay,
       sndFilesTTL = nominalDay,
       xftpNotifyErrsOnRetry = True,

src/Simplex/Messaging/Agent/NtfSubSupervisor.hs

@@ -26,6 +26,7 @@ import Data.Bifunctor (first)
 import qualified Data.Map.Strict as M
 import Data.Text (Text)
 import Data.Time (UTCTime, addUTCTime, getCurrentTime)
+import Data.Time.Clock (diffUTCTime)
 import Simplex.Messaging.Agent.Client
 import Simplex.Messaging.Agent.Env.SQLite
 import Simplex.Messaging.Agent.Protocol (ACommand (..), APartyCmd (..), AgentErrorType (..), BrokerErrorType (..), ConnId, NotificationsMode (..), SAEntity (..))
@@ -39,7 +40,7 @@ import Simplex.Messaging.Notifications.Types
 import Simplex.Messaging.Protocol (NtfServer, ProtocolServer, SMPServer, sameSrvAddr)
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
-import Simplex.Messaging.Util (diffInMicros, threadDelay', tshow, unlessM)
+import Simplex.Messaging.Util (diffToMicroseconds, threadDelay', tshow, unlessM)
 import System.Random (randomR)
 import UnliftIO
 import UnliftIO.Concurrent (forkIO, threadDelay)
@@ -291,7 +292,7 @@ rescheduleAction doWork ts actionTs
   | otherwise = do
     void . atomically $ tryTakeTMVar doWork
     void . forkIO $ do
-      liftIO $ threadDelay' $ diffInMicros actionTs ts
+      liftIO $ threadDelay' $ diffToMicroseconds $ diffUTCTime actionTs ts
       void . atomically $ tryPutTMVar doWork ()
     pure True
 

src/Simplex/Messaging/Agent/Store.hs

@@ -428,7 +428,8 @@ data RcvMsgData = RcvMsgData
     msgBody :: MsgBody,
     internalRcvId :: InternalRcvId,
     internalHash :: MsgHash,
-    externalPrevSndHash :: MsgHash
+    externalPrevSndHash :: MsgHash,
+    encryptedMsgHash :: MsgHash
   }
 
 data RcvMsg = RcvMsg

src/Simplex/Messaging/Agent/Store/SQLite.hs

@@ -94,8 +94,10 @@ module Simplex.Messaging.Agent.Store.SQLite
     deletePendingMsgs,
     setMsgUserAck,
     getLastMsg,
+    checkRcvMsgHashExists,
     deleteMsg,
     deleteSndMsgDelivery,
+    deleteRcvMsgHashesExpired,
     -- Double ratchet persistence
     createRatchetX3dhKeys,
     getRatchetX3dhKeys,
@@ -182,6 +184,7 @@ module Simplex.Messaging.Agent.Store.SQLite
     -- * utilities
     withConnection,
     withTransaction,
+    withTransactionCtx,
     firstRow,
     firstRow',
     maybeFirstRow,
@@ -212,7 +215,7 @@ import Data.Ord (Down (..))
 import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Text.Encoding (decodeLatin1, encodeUtf8)
-import Data.Time.Clock (NominalDiffTime, UTCTime, addUTCTime, getCurrentTime)
+import Data.Time.Clock (NominalDiffTime, UTCTime, addUTCTime, diffUTCTime, getCurrentTime)
 import Data.Word (Word32)
 import Database.SQLite.Simple (FromRow, NamedParam (..), Only (..), Query (..), SQLError, ToRow, field, (:.) (..))
 import qualified Database.SQLite.Simple as DB
@@ -241,7 +244,7 @@ import Simplex.Messaging.Parsers (blobFieldParser, dropPrefix, fromTextField_, s
 import Simplex.Messaging.Protocol
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Transport.Client (TransportHost)
-import Simplex.Messaging.Util (bshow, eitherToMaybe, ($>>=), (<$$>))
+import Simplex.Messaging.Util (bshow, diffToMilliseconds, eitherToMaybe, ($>>=), (<$$>))
 import Simplex.Messaging.Version
 import System.Directory (copyFile, createDirectoryIfMissing, doesFileExist)
 import System.Exit (exitFailure)
@@ -428,16 +431,29 @@ withConnection SQLiteStore {dbConnection} =
     (atomically . putTMVar dbConnection)
 
 withTransaction :: forall a. SQLiteStore -> (DB.Connection -> IO a) -> IO a
-withTransaction st action = withConnection st $ loop 500 3_000_000
+withTransaction = withTransactionCtx Nothing
+
+withTransactionCtx :: forall a. Maybe String -> SQLiteStore -> (DB.Connection -> IO a) -> IO a
+withTransactionCtx ctx_ st action = withConnection st $ loop 500 3_000_000
   where
     loop :: Int -> Int -> DB.Connection -> IO a
     loop t tLim db =
-      DB.withImmediateTransaction db (action db) `E.catch` \(e :: SQLError) ->
+      transactionWithCtx `E.catch` \(e :: SQLError) ->
         if tLim > t && DB.sqlError e == DB.ErrorBusy
           then do
             threadDelay t
             loop (t * 9 `div` 8) (tLim - t) db
           else E.throwIO e
+      where
+        transactionWithCtx = case ctx_ of
+          Nothing -> DB.withImmediateTransaction db (action db)
+          Just ctx -> do
+            t1 <- getCurrentTime
+            r <- DB.withImmediateTransaction db (action db)
+            t2 <- getCurrentTime
+            putStrLn $ "withTransactionCtx start :: " <> show t1 <> " :: " <> ctx
+            putStrLn $ "withTransactionCtx end   :: " <> show t2 <> " :: " <> ctx <> " :: duration=" <> show (diffToMilliseconds $ diffUTCTime t2 t1)
+            pure r
 
 createUserRecord :: DB.Connection -> IO UserId
 createUserRecord db = do
@@ -925,6 +941,17 @@ getLastMsg db connId msgId =
       let msgMeta = MsgMeta {recipient = (agentMsgId, internalTs), broker = (brokerId, brokerTs), sndMsgId, integrity}
        in RcvMsg {internalId = InternalId agentMsgId, msgMeta, msgBody, userAck}
 
+checkRcvMsgHashExists :: DB.Connection -> ConnId -> ByteString -> IO Bool
+checkRcvMsgHashExists db connId hash = do
+  fromMaybe False
+    <$> maybeFirstRow
+      fromOnly
+      ( DB.query
+          db
+          "SELECT 1 FROM encrypted_rcv_message_hashes WHERE conn_id = ? AND hash = ? LIMIT 1"
+          (connId, hash)
+      )
+
 deleteMsg :: DB.Connection -> ConnId -> InternalId -> IO ()
 deleteMsg db connId msgId =
   DB.execute db "DELETE FROM messages WHERE conn_id = ? AND internal_id = ?;" (connId, msgId)
@@ -938,6 +965,11 @@ deleteSndMsgDelivery db connId SndQueue {dbQueueId} msgId = do
   (Only (cnt :: Int) : _) <- DB.query db "SELECT count(*) FROM snd_message_deliveries WHERE conn_id = ? AND internal_id = ?" (connId, msgId)
   when (cnt == 0) $ deleteMsg db connId msgId
 
+deleteRcvMsgHashesExpired :: DB.Connection -> NominalDiffTime -> IO ()
+deleteRcvMsgHashesExpired db ttl = do
+  cutoffTs <- addUTCTime (- ttl) <$> getCurrentTime
+  DB.execute db "DELETE FROM encrypted_rcv_message_hashes WHERE created_at < ?" (Only cutoffTs)
+
 createRatchetX3dhKeys :: DB.Connection -> ConnId -> C.PrivateKeyX448 -> C.PrivateKeyX448 -> IO ()
 createRatchetX3dhKeys db connId x3dhPrivKey1 x3dhPrivKey2 =
   DB.execute db "INSERT INTO ratchets (conn_id, x3dh_priv_key_1, x3dh_priv_key_2) VALUES (?, ?, ?)" (connId, x3dhPrivKey1, x3dhPrivKey2)
@@ -1690,10 +1722,10 @@ insertRcvMsgBase_ dbConn connId RcvMsgData {msgMeta, msgType, msgFlags, msgBody,
     ]
 
 insertRcvMsgDetails_ :: DB.Connection -> ConnId -> RcvQueue -> RcvMsgData -> IO ()
-insertRcvMsgDetails_ dbConn connId RcvQueue {dbQueueId} RcvMsgData {msgMeta, internalRcvId, internalHash, externalPrevSndHash} = do
+insertRcvMsgDetails_ db connId RcvQueue {dbQueueId} RcvMsgData {msgMeta, internalRcvId, internalHash, externalPrevSndHash, encryptedMsgHash} = do
   let MsgMeta {integrity, recipient, broker, sndMsgId} = msgMeta
   DB.executeNamed
-    dbConn
+    db
     [sql|
       INSERT INTO rcv_messages
         ( conn_id, rcv_queue_id, internal_rcv_id, internal_id, external_snd_id,
@@ -1715,6 +1747,7 @@ insertRcvMsgDetails_ dbConn connId RcvQueue {dbQueueId} RcvMsgData {msgMeta, int
       ":external_prev_snd_hash" := externalPrevSndHash,
       ":integrity" := integrity
     ]
+  DB.execute db "INSERT INTO encrypted_rcv_message_hashes (conn_id, hash) VALUES (?,?)" (connId, encryptedMsgHash)
 
 updateHashRcv_ :: DB.Connection -> ConnId -> RcvMsgData -> IO ()
 updateHashRcv_ dbConn connId RcvMsgData {msgMeta, internalHash, internalRcvId} =

src/Simplex/Messaging/Agent/Store/SQLite/Migrations.hs

@@ -59,6 +59,7 @@ import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230223_files
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230320_retry_state
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230401_snd_files
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230510_files_pending_replicas_indexes
+import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230516_encrypted_rcv_message_hashes
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (dropPrefix, sumTypeJSON)
 import Simplex.Messaging.Transport.Client (TransportHost)
@@ -84,7 +85,8 @@ schemaMigrations =
     ("m20230223_files", m20230223_files, Just down_m20230223_files),
     ("m20230320_retry_state", m20230320_retry_state, Just down_m20230320_retry_state),
     ("m20230401_snd_files", m20230401_snd_files, Just down_m20230401_snd_files),
-    ("m20230510_files_pending_replicas_indexes", m20230510_files_pending_replicas_indexes, Just down_m20230510_files_pending_replicas_indexes)
+    ("m20230510_files_pending_replicas_indexes", m20230510_files_pending_replicas_indexes, Just down_m20230510_files_pending_replicas_indexes),
+    ("m20230516_encrypted_rcv_message_hashes", m20230516_encrypted_rcv_message_hashes, Just down_m20230516_encrypted_rcv_message_hashes)
   ]
 
 -- | The list of migrations in ascending order by date

src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20230516_encrypted_rcv_message_hashes.hs

@@ -0,0 +1,28 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230516_encrypted_rcv_message_hashes where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20230516_encrypted_rcv_message_hashes :: Query
+m20230516_encrypted_rcv_message_hashes =
+  [sql|
+CREATE TABLE encrypted_rcv_message_hashes(
+  encrypted_rcv_message_hash_id INTEGER PRIMARY KEY,
+  conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,
+  hash BLOB NOT NULL,
+  created_at TEXT NOT NULL DEFAULT (datetime('now')),
+  updated_at TEXT NOT NULL DEFAULT (datetime('now'))
+);
+
+CREATE INDEX idx_encrypted_rcv_message_hashes_hash ON encrypted_rcv_message_hashes(conn_id, hash);
+|]
+
+down_m20230516_encrypted_rcv_message_hashes :: Query
+down_m20230516_encrypted_rcv_message_hashes =
+  [sql|
+DROP INDEX idx_encrypted_rcv_message_hashes_hash;
+
+DROP TABLE encrypted_rcv_message_hashes;
+|]

src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql

@@ -346,6 +346,13 @@ CREATE TABLE deleted_snd_chunk_replicas(
   created_at TEXT NOT NULL DEFAULT(datetime('now')),
   updated_at TEXT NOT NULL DEFAULT(datetime('now'))
 );
+CREATE TABLE encrypted_rcv_message_hashes(
+  encrypted_rcv_message_hash_id INTEGER PRIMARY KEY,
+  conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,
+  hash BLOB NOT NULL,
+  created_at TEXT NOT NULL DEFAULT(datetime('now')),
+  updated_at TEXT NOT NULL DEFAULT(datetime('now'))
+);
 CREATE UNIQUE INDEX idx_rcv_queues_ntf ON rcv_queues(host, port, ntf_id);
 CREATE UNIQUE INDEX idx_rcv_queue_id ON rcv_queues(conn_id, rcv_queue_id);
 CREATE UNIQUE INDEX idx_snd_queue_id ON snd_queues(conn_id, snd_queue_id);
@@ -432,3 +439,7 @@ CREATE INDEX idx_snd_file_chunk_replicas_pending ON snd_file_chunk_replicas(
 CREATE INDEX idx_deleted_snd_chunk_replicas_pending ON deleted_snd_chunk_replicas(
   created_at
 );
+CREATE INDEX idx_encrypted_rcv_message_hashes_hash ON encrypted_rcv_message_hashes(
+  conn_id,
+  hash
+);

src/Simplex/Messaging/Server.hs

@@ -741,7 +741,8 @@ restoreServerMessages = asks (storeMsgsFile . config) >>= mapM_ restoreMessages
       st <- asks queueStore
       ms <- asks msgStore
       quota <- asks $ msgQueueQuota . config
-      runExceptT (liftIO (B.readFile f) >>= mapM_ (restoreMsg st ms quota) . B.lines) >>= \case
+      old_ <- asks (messageExpiration . config) $>>= (liftIO . fmap Just . expireBeforeEpoch)
+      runExceptT (liftIO (B.readFile f) >>= mapM_ (restoreMsg st ms quota old_) . B.lines) >>= \case
         Left e -> do
           logError . T.pack $ "error restoring messages: " <> e
           liftIO exitFailure
@@ -749,7 +750,7 @@ restoreServerMessages = asks (storeMsgsFile . config) >>= mapM_ restoreMessages
           renameFile f $ f <> ".bak"
           logInfo "messages restored"
       where
-        restoreMsg st ms quota s = do
+        restoreMsg st ms quota old_ s = do
           r <- liftEither . first (msgErr "parsing") $ strDecode s
           case r of
             MLRv3 rId msg -> addToMsgQueue rId msg
@@ -759,13 +760,14 @@ restoreServerMessages = asks (storeMsgsFile . config) >>= mapM_ restoreMessages
               addToMsgQueue rId msg'
           where
             addToMsgQueue rId msg = do
-              full <- atomically $ do
+              logFull <- atomically $ do
                 q <- getMsgQueue ms rId quota
-                isNothing <$> writeMsg q msg
-              case msg of
-                Message {} ->
-                  when full . logError . decodeLatin1 $ "message queue " <> strEncode rId <> " is full, message not restored: " <> strEncode (msgId (msg :: Message))
-                MessageQuota {} -> pure ()
+                case msg of
+                  Message {msgTs}
+                    | maybe True (systemSeconds msgTs >=) old_ -> isNothing <$> writeMsg q msg
+                    | otherwise -> pure False
+                  MessageQuota {} -> writeMsg q msg $> False
+              when logFull . logError . decodeLatin1 $ "message queue " <> strEncode rId <> " is full, message not restored: " <> strEncode (msgId (msg :: Message))
             updateMsgV1toV3 QueueRec {rcvDhSecret} RcvMessage {msgId, msgTs, msgFlags, msgBody = EncRcvMsgBody body} = do
               let nonce = C.cbNonce msgId
               msgBody <- liftEither . first (msgErr "v1 message decryption") $ C.maxLenBS =<< C.cbDecrypt rcvDhSecret nonce body

src/Simplex/Messaging/Server/Main.hs

@@ -165,7 +165,7 @@ smpServerCLI cfgPath logPath =
           ServerConfig
             { transports = iniTransports ini,
               tbqSize = 32,
-              serverTbqSize = 128,
+              serverTbqSize = 1024,
               msgQueueQuota = 128,
               queueIdBytes = 24,
               msgIdBytes = 24, -- must be at least 24 bytes, it is used as 192-bit nonce for XSalsa20

src/Simplex/Messaging/Util.hs

@@ -12,13 +12,11 @@ import Control.Monad.Trans.Except
 import Data.Bifunctor (first)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
-import Data.Fixed (Fixed (MkFixed), Pico)
 import Data.Int (Int64)
 import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Text.Encoding (decodeUtf8With)
-import Data.Time (nominalDiffTimeToSeconds)
-import Data.Time.Clock (UTCTime, diffUTCTime)
+import Data.Time (NominalDiffTime, nominalDiffTimeToSeconds)
 import UnliftIO.Async
 
 raceAny_ :: MonadUnliftIO m => [m a] -> m ()
@@ -117,14 +115,8 @@ threadDelay' time = do
   threadDelay $ fromIntegral maxWait
   when (maxWait /= time) $ threadDelay' (time - maxWait)
 
-diffInSeconds :: UTCTime -> UTCTime -> Int64
-diffInSeconds a b = (`div` 1000000_000000) $ diffInPicos a b
+diffToMicroseconds :: NominalDiffTime -> Int64
+diffToMicroseconds diff = fromIntegral ((truncate $ diff * 1000000) :: Integer)
 
-diffInMicros :: UTCTime -> UTCTime -> Int64
-diffInMicros a b = (`div` 1000000) $ diffInPicos a b
-
-diffInPicos :: UTCTime -> UTCTime -> Int64
-diffInPicos a b = fromInteger . fromPico . nominalDiffTimeToSeconds $ diffUTCTime a b
-
-fromPico :: Pico -> Integer
-fromPico (MkFixed i) = i
+diffToMilliseconds :: NominalDiffTime -> Int64
+diffToMilliseconds diff = fromIntegral ((truncate $ diff * 1000) :: Integer)

tests/AgentTests/SQLiteTests.hs

@@ -428,7 +428,8 @@ mkRcvMsgData internalId internalRcvId externalSndId brokerId internalHash =
       msgFlags = SMP.noMsgFlags,
       msgBody = hw,
       internalHash,
-      externalPrevSndHash = "hash_from_sender"
+      externalPrevSndHash = "hash_from_sender",
+      encryptedMsgHash = "encrypted_msg_hash"
     }
 
 testCreateRcvMsg_ :: DB.Connection -> PrevExternalSndId -> PrevRcvMsgHash -> ConnId -> RcvQueue -> RcvMsgData -> Expectation

tests/ServerTests.hs

@@ -55,7 +55,9 @@ serverTests t@(ATransport t') = do
     describe "Exceeding queue quota" $ testExceedQueueQuota t'
   describe "Store log" $ testWithStoreLog t
   describe "Restore messages" $ testRestoreMessages t
-  describe "Restore messages (old / v2)" $ testRestoreMessagesV2 t
+  describe "Restore messages (old / v2)" $ do
+    testRestoreMessagesV2 t
+    testRestoreExpireMessages t
   describe "Timing of AUTH error" $ testTiming t
   describe "Message notifications" $ testMessageNotifications t
   describe "Message expiration" $ do
@@ -779,6 +781,62 @@ testRestoreMessagesV2 at@(ATransport t) =
     runClient :: Transport c => TProxy c -> (THandle c -> IO ()) -> Expectation
     runClient _ test' = testSMPClient test' `shouldReturn` ()
 
+testRestoreExpireMessages :: ATransport -> Spec
+testRestoreExpireMessages at@(ATransport t) =
+  it "should store messages on exit and restore on start" $ do
+    (sPub, sKey) <- C.generateSignatureKeyPair C.SEd25519
+    recipientId <- newTVarIO ""
+    recipientKey <- newTVarIO Nothing
+    dhShared <- newTVarIO Nothing
+    senderId <- newTVarIO ""
+
+    withSmpServerStoreMsgLogOnV2 at testPort . runTest t $ \h -> do
+      runClient t $ \h1 -> do
+        (sId, rId, rKey, dh) <- createAndSecureQueue h1 sPub
+        atomically $ do
+          writeTVar recipientId rId
+          writeTVar recipientKey $ Just rKey
+          writeTVar dhShared $ Just dh
+          writeTVar senderId sId
+      sId <- readTVarIO senderId
+      Resp "1" _ OK <- signSendRecv h sKey ("1", sId, _SEND "hello 1")
+      Resp "2" _ OK <- signSendRecv h sKey ("2", sId, _SEND "hello 2")
+      threadDelay 3000000
+      Resp "3" _ OK <- signSendRecv h sKey ("3", sId, _SEND "hello 3")
+      Resp "4" _ OK <- signSendRecv h sKey ("4", sId, _SEND "hello 4")
+      pure ()
+
+    logSize testStoreLogFile `shouldReturn` 2
+    msgs <- B.readFile testStoreMsgsFile
+    length (B.lines msgs) `shouldBe` 4
+
+    let expCfg1 = Just ExpirationConfig {ttl = 86400, checkInterval = 43200}
+        cfg1 = cfgV2 {storeLogFile = Just testStoreLogFile, storeMsgsFile = Just testStoreMsgsFile, messageExpiration = expCfg1}
+    withSmpServerConfigOn at cfg1 testPort . runTest t $ \_ -> pure ()
+
+    logSize testStoreLogFile `shouldReturn` 1
+    msgs' <- B.readFile testStoreMsgsFile
+    msgs' `shouldBe` msgs
+
+    let expCfg2 = Just ExpirationConfig {ttl = 2, checkInterval = 43200}
+        cfg2 = cfgV2 {storeLogFile = Just testStoreLogFile, storeMsgsFile = Just testStoreMsgsFile, messageExpiration = expCfg2}
+    withSmpServerConfigOn at cfg2 testPort . runTest t $ \_ -> pure ()
+
+    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)
+
+  where
+    runTest :: Transport c => TProxy c -> (THandle c -> IO ()) -> ThreadId -> Expectation
+    runTest _ test' server = do
+      testSMPClient test' `shouldReturn` ()
+      killThread server
+
+    runClient :: Transport c => TProxy c -> (THandle c -> IO ()) -> Expectation
+    runClient _ test' = testSMPClient test' `shouldReturn` ()
+
 createAndSecureQueue :: Transport c => THandle c -> SndPublicVerifyKey -> IO (SenderId, RecipientId, RcvPrivateSignKey, RcvDhSecret)
 createAndSecureQueue h sPub = do
   (rPub, rKey) <- C.generateSignatureKeyPair C.SEd448