Merge branch 'master' into ep/test-coverage

* docker entrypoint: DRY store log backup

Don't Repeat Yourself

Set a variable for the full path to the source file.

Create the backup path by appending an extension from `date` output.

Also fixed quoting and switched to an `if` block.

Clean up block level variables.

* docker entrypoint: use ISO 8601 format

The previous format discards information about the local time zone.

* docker entrypoint: always use UTC

Now the format always ends in +00:00, and we can ignore that part again.

* docker entrypoint: provide the date format again

* docker entrypoint: remove time zone from the date format

* docker entrypoint: use an unambiguous date format

Present a leading zero before the month:
YYYY-0MM-DD

Both YYYY-MM-DD and YYYY-DD-MM are used by people and can be confusing in the beginning of the month.

* docker entrypoint: use appropriate quoting

Without avoiding field splitting a password containing a space changes the number of arguments being set.

* docker entrypoint: use explicit braces for custom variables

Make the intentions clear and don't assume the user knows the special cases for when variables won't be extended.

Example:
word=animal
words=mistake
echo "$words vs $word vs ${word}s"

* docker entrypoint-xftp-server: braces and quoting

* docker entrypoint-xftp-server: backup block

* docker entrypoints: explain date format in a comment

* switched from long to short option to date for POSIX

* docker entrypoint-smp-server: explain date format

* docker entrypoint-xftp-server: explain date format

* docker entrypoints: further explain format

I fixed the case to match the date format letters.

Also, use words to explain since I don't want everyone to need to read about date formats to understand.

* docker entrypoints: only quote letters

I was either going to quote the dashes too or stop quoting the colons.

Having less quotes was more readable.

* Revert "docker entrypoint: use an unambiguous date format"

This reverts commit ba2a93bad9.

* docker entrypoints: remove %F

* docker entrypoint-smp-server: remove %F

Used part of the explicit ISO 8601 format,
provided by the coreutils date invocation guide.

* docker entrypoint-xftp-server: remove %F

Used part of the explicit ISO 8601 format,
provided by the coreutils date invocation guide.

.gitignore

@@ -7,3 +7,7 @@ dist-newstyle/
 
 cabal.project.local
 cabal.project.local~
+
+.hpc/
+*.tix
+.coverage

cabal.project

@@ -3,6 +3,14 @@ packages: .
 -- packages: . ../hs-socks
 -- packages: . ../http2
 
+package *
+  coverage: True
+  library-coverage: True
+
+package attoparsec
+  coverage: False
+  library-coverage: False
+
 source-repository-package
     type: git
     location: https://github.com/simplex-chat/aeson.git

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>,
@@ -144,6 +144,8 @@ tests:
       - silently == 1.2.*
       - main-tester == 0.2.*
       - timeit == 2.0.*
+    ghc-options:
+      - -fhpc
 
 ghc-options:
   # - -haddock

scripts/docker/entrypoint-smp-server

@@ -1,25 +1,25 @@
 #!/usr/bin/env sh
-confd="/etc/opt/simplex"
-logd="/var/opt/simplex/"
+confd='/etc/opt/simplex'
+logd='/var/opt/simplex/'
 
 # Check if server has been initialized
-if [ ! -f "$confd/smp-server.ini" ]; then
+if [ ! -f "${confd}/smp-server.ini" ]; then
   # If not, determine ip or domain
-  case "$ADDR" in
-    '') printf "Please specify \$ADDR environment variable.\n"; exit 1 ;;
+  case "${ADDR}" in
+    '') printf 'Please specify $ADDR environment variable.\n'; exit 1 ;;
     *[a-zA-Z]*)
-      case "$ADDR" in
-        *:*) set -- --ip "$ADDR" ;;
-        *) set -- -n "$ADDR" ;;
+      case "${ADDR}" in
+        *:*) set -- --ip "${ADDR}" ;;
+        *) set -- -n "${ADDR}" ;;
       esac
       ;;
-    *) set -- --ip "$ADDR" ;;
+    *) set -- --ip "${ADDR}" ;;
   esac
 
   # Optionally, set password
-  case "$PASS" in
+  case "${PASS}" in
     '') set -- "$@" --no-password ;;
-    *) set -- "$@" --password "$PASS" ;;
+    *) set -- "$@" --password "${PASS}" ;;
   esac
 
   # And init certificates and configs
@@ -27,7 +27,20 @@ if [ ! -f "$confd/smp-server.ini" ]; then
 fi
 
 # Backup store log just in case
-[ -f "$logd/smp-server-store.log" ] && cp "$logd"/smp-server-store.log "$logd"/smp-server-store.log."$(date +'%FT%T')"
+#
+# Uses the UTC (universal) time zone and this
+# format: YYYY-mm-dd'T'HH:MM:SS
+# year, month, day, letter T, hour, minute, second
+#
+# This is the ISO 8601 format without the time zone at the end.
+#
+_file="${logd}/smp-server-store.log"
+if [ -f "${_file}" ]; then
+  _backup_extension="$(date -u '+%Y-%m-%dT%H:%M:%S')"
+  cp -v -p "${_file}" "${_file}.${_backup_extension:-date-failed}"
+  unset -v _backup_extension
+fi
+unset -v _file
 
 # Finally, run smp-sever. Notice that "exec" here is important:
 # smp-server replaces our helper script, so that it can catch INT signal

scripts/docker/entrypoint-xftp-server

@@ -1,25 +1,25 @@
 #!/usr/bin/env sh
-confd="/etc/opt/simplex-xftp"
-logd="/var/opt/simplex-xftp"
+confd='/etc/opt/simplex-xftp'
+logd='/var/opt/simplex-xftp'
 
 # Check if server has been initialized
-if [ ! -f "$confd/file-server.ini" ]; then
+if [ ! -f "${confd}/file-server.ini" ]; then
   # If not, determine ip or domain
-  case "$ADDR" in
-    '') printf "Please specify \$ADDR environment variable.\n"; exit 1 ;;
+  case "${ADDR}" in
+    '') printf 'Please specify $ADDR environment variable.\n'; exit 1 ;;
     *[a-zA-Z]*)
-      case "$ADDR" in
-        *:*) set -- --ip "$ADDR" ;;
-        *) set -- -n "$ADDR" ;;
+      case "${ADDR}" in
+        *:*) set -- --ip "${ADDR}" ;;
+        *) set -- -n "${ADDR}" ;;
       esac
       ;;
-    *) set -- --ip "$ADDR" ;;
+    *) set -- --ip "${ADDR}" ;;
   esac
 
   # Set quota
-  case "$QUOTA" in
-    '') printf "Please specify \$QUOTA environment variable.\n"; exit 1 ;;
-    *) set -- "$@" --quota "$QUOTA" ;;
+  case "${QUOTA}" in
+    '') printf 'Please specify $QUOTA environment variable.\n'; exit 1 ;;
+    *) set -- "$@" --quota "${QUOTA}" ;;
   esac
 
   # Init the certificates and configs
@@ -27,7 +27,20 @@ if [ ! -f "$confd/file-server.ini" ]; then
 fi
 
 # Backup store log just in case
-[ -f "$logd/file-server-store.log" ] && cp "$logd"/file-server-store.log "$logd"/file-server-store.log."$(date +'%FT%T')"
+#
+# Uses the UTC (universal) time zone and this
+# format: YYYY-mm-dd'T'HH:MM:SS
+# year, month, day, letter T, hour, minute, second
+#
+# This is the ISO 8601 format without the time zone at the end.
+#
+_file="${logd}/file-server-store.log"
+if [ -f "${_file}" ]; then
+  _backup_extension="$(date -u '+%Y-%m-%dT%H:%M:%S')"
+  cp -v -p "${_file}" "${_file}.${_backup_extension:-date-failed}"
+  unset -v _backup_extension
+fi
+unset -v _file
 
 # Finally, run xftp-sever. Notice that "exec" here is important:
 # smp-server replaces our helper script, so that it can catch INT signal

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
@@ -544,7 +545,7 @@ test-suite simplexmq-test
       Paths_simplexmq
   hs-source-dirs:
       tests
-  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns
+  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -fhpc
   build-depends:
       HUnit ==1.6.*
     , QuickCheck ==2.14.*

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)
@@ -1652,6 +1656,8 @@ cleanupManager c@AgentClient {subQ} = do
     notify :: forall e. AEntityI e => EntityId -> ACommand 'Agent e -> ExceptT AgentErrorType m ()
     notify entId cmd = atomically $ writeTBQueue subQ ("", entId, APC (sAEntity @e) cmd)
 
+-- | make sure to ACK or throw in each message processing branch
+-- it cannot be finally, unfortunately, as sometimes it needs to be ACK+DEL
 processSMPTransmission :: forall m. AgentMonad m => AgentClient -> ServerTransmission BrokerMsg -> m ()
 processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, sessId, rId, cmd) = do
   (rq, SomeConn _ conn) <- withStore c (\db -> getRcvConn db srv rId)
@@ -1697,7 +1703,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 +1735,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 >> ack
+                        Left e -> checkDuplicateHash e encryptedMsgHash >> ack
                       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 +1755,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/FunctionalAPITests.hs

@@ -61,6 +61,7 @@ import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Transport (ATransport (..))
 import Simplex.Messaging.Util (tryError)
 import Simplex.Messaging.Version
+import System.Directory (copyFile, renameFile)
 import Test.Hspec
 import UnliftIO
 import XFTPClient (testXFTPServer)
@@ -147,9 +148,13 @@ functionalAPITests t = do
       testAsyncServerOffline t
     it "should notify after HELLO timeout" $
       withSmpServer t testAsyncHelloTimeout
-  describe "Duplicate message delivery" $
+  describe "Message delivery" $ do
     it "should deliver messages to the user once, even if repeat delivery is made by the server (no ACK)" $
       testDuplicateMessage t
+    it "should report error via msg integrity on skipped messages" $
+      testSkippedMessages t
+    it "should report decryption error on ratchet becoming out of sync" $
+      testDecryptionError t
   describe "Inactive client disconnection" $ do
     it "should disconnect clients if it was inactive longer than TTL" $
       testInactiveClientDisconnected t
@@ -163,8 +168,10 @@ functionalAPITests t = do
     it "should suspend agent on timeout, even if pending messages not sent" $
       testSuspendingAgentTimeout t
   describe "Batching SMP commands" $ do
-    it "should subscribe to multiple subscriptions with batching" $
-      testBatchedSubscriptions t
+    xit "should subscribe to multiple (200) subscriptions with batching" $
+      testBatchedSubscriptions 200 10 t
+    it "should subscribe to multiple (6) subscriptions with batching" $
+      testBatchedSubscriptions 6 3 t
   describe "Async agent commands" $ do
     it "should connect using async agent commands" $
       withSmpServer t testAsyncCommands
@@ -491,6 +498,101 @@ testDuplicateMessage t = do
       get alice2 ##> ("", bobId, SENT 6)
       get bob2 =##> \case ("", c, Msg "hello 3") -> c == aliceId; _ -> False
 
+testSkippedMessages :: HasCallStack => ATransport -> IO ()
+testSkippedMessages t = do
+  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  (aliceId, bobId) <- withSmpServerStoreLogOn t testPort $ \_ -> do
+    (aliceId, bobId) <- runRight $ makeConnection alice bob
+    runRight_ $ do
+      4 <- sendMessage alice bobId SMP.noMsgFlags "hello"
+      get alice ##> ("", bobId, SENT 4)
+      get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
+      ackMessage bob aliceId 4
+
+    disconnectAgentClient bob
+
+    runRight_ $ do
+      5 <- sendMessage alice bobId SMP.noMsgFlags "hello 2"
+      get alice ##> ("", bobId, SENT 5)
+      6 <- sendMessage alice bobId SMP.noMsgFlags "hello 3"
+      get alice ##> ("", bobId, SENT 6)
+      7 <- sendMessage alice bobId SMP.noMsgFlags "hello 4"
+      get alice ##> ("", bobId, SENT 7)
+
+    pure (aliceId, bobId)
+
+  nGet alice =##> \case ("", "", DOWN _ [c]) -> c == bobId; _ -> False
+  threadDelay 200000
+
+  disconnectAgentClient alice
+
+  alice2 <- getSMPAgentClient' agentCfg initAgentServers testDB
+  bob2 <- getSMPAgentClient' agentCfg initAgentServers testDB2
+
+  withSmpServerStoreLogOn t testPort $ \_ -> do
+    runRight_ $ do
+      subscribeConnection bob2 aliceId
+      subscribeConnection alice2 bobId
+
+      8 <- sendMessage alice2 bobId SMP.noMsgFlags "hello 5"
+      get alice2 ##> ("", bobId, SENT 8)
+      get bob2 =##> \case ("", c, MSG MsgMeta {integrity = MsgError {errorInfo = MsgSkipped {fromMsgId = 4, toMsgId = 6}}} _ "hello 5") -> c == aliceId; _ -> False
+      ackMessage bob2 aliceId 5
+
+      9 <- sendMessage alice2 bobId SMP.noMsgFlags "hello 6"
+      get alice2 ##> ("", bobId, SENT 9)
+      get bob2 =##> \case ("", c, Msg "hello 6") -> c == aliceId; _ -> False
+      ackMessage bob2 aliceId 6
+
+testDecryptionError :: HasCallStack => ATransport -> IO ()
+testDecryptionError t = do
+  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  withSmpServerStoreMsgLogOn t testPort $ \_ -> do
+    (aliceId, bobId) <- runRight $ makeConnection alice bob
+    runRight_ $ do
+      4 <- sendMessage alice bobId SMP.noMsgFlags "hello"
+      get alice ##> ("", bobId, SENT 4)
+      get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
+      ackMessage bob aliceId 4
+
+      5 <- sendMessage bob aliceId SMP.noMsgFlags "hello 2"
+      get bob ##> ("", aliceId, SENT 5)
+      get alice =##> \case ("", c, Msg "hello 2") -> c == bobId; _ -> False
+      ackMessage alice bobId 5
+
+      liftIO $ copyFile testDB2 (testDB2 <> ".bak")
+
+      6 <- sendMessage alice bobId SMP.noMsgFlags "hello 3"
+      get alice ##> ("", bobId, SENT 6)
+      get bob =##> \case ("", c, Msg "hello 3") -> c == aliceId; _ -> False
+      ackMessage bob aliceId 6
+
+      7 <- sendMessage bob aliceId SMP.noMsgFlags "hello 4"
+      get bob ##> ("", aliceId, SENT 7)
+      get alice =##> \case ("", c, Msg "hello 4") -> c == bobId; _ -> False
+      ackMessage alice bobId 7
+
+    disconnectAgentClient bob
+
+    -- importing database backup after progressing ratchet de-synchronizes ratchet,
+    -- this will be fixed by ratchet re-negotiation
+    liftIO $ renameFile (testDB2 <> ".bak") testDB2
+
+    bob2 <- getSMPAgentClient' agentCfg initAgentServers testDB2
+
+    runRight_ $ do
+      subscribeConnection bob2 aliceId
+
+      8 <- sendMessage alice bobId SMP.noMsgFlags "hello 5"
+      get alice ##> ("", bobId, SENT 8)
+      get bob2 =##> \case ("", c, ERR AGENT {agentErr = A_CRYPTO {cryptoErr = RATCHET_HEADER}}) -> c == aliceId; _ -> False
+
+      6 <- sendMessage bob2 aliceId SMP.noMsgFlags "hello 6"
+      get bob2 ##> ("", aliceId, SENT 6)
+      get alice =##> \case ("", c, ERR AGENT {agentErr = A_CRYPTO {cryptoErr = RATCHET_HEADER}}) -> c == bobId; _ -> False
+
 makeConnection :: AgentClient -> AgentClient -> ExceptT AgentErrorType IO (ConnId, ConnId)
 makeConnection alice bob = makeConnectionForUsers alice 1 bob 1
 
@@ -612,14 +714,14 @@ testSuspendingAgentTimeout t = do
     ("", "", SUSPENDED) <- nGet b
     pure ()
 
-testBatchedSubscriptions :: ATransport -> IO ()
-testBatchedSubscriptions t = do
+testBatchedSubscriptions :: Int -> Int -> ATransport -> IO ()
+testBatchedSubscriptions nCreate nDel t = do
   a <- getSMPAgentClient' agentCfg initAgentServers2 testDB
   b <- getSMPAgentClient' agentCfg initAgentServers2 testDB2
   conns <- runServers $ do
-    conns <- forM [1 .. 200 :: Int] . const $ makeConnection a b
+    conns <- forM [1 .. nCreate :: Int] . const $ makeConnection a b
     forM_ conns $ \(aId, bId) -> exchangeGreetings a bId b aId
-    let (aIds', bIds') = unzip $ take 10 conns
+    let (aIds', bIds') = unzip $ take nDel conns
     delete a bIds'
     delete b aIds'
     liftIO $ threadDelay 1000000
@@ -635,11 +737,14 @@ testBatchedSubscriptions t = do
     ("", "", UP {}) <- nGet b
     liftIO $ threadDelay 1000000
     let (aIds, bIds) = unzip conns
-        conns' = drop 10 conns
+        conns' = drop nDel conns
         (aIds', bIds') = unzip conns'
     subscribe a bIds
     subscribe b aIds
     forM_ conns' $ \(aId, bId) -> exchangeGreetingsMsgId 6 a bId b aId
+    void $ resubscribeConnections a bIds
+    void $ resubscribeConnections b aIds
+    forM_ conns' $ \(aId, bId) -> exchangeGreetingsMsgId 8 a bId b aId
     delete a bIds'
     delete b aIds'
     deleteFail a bIds'
@@ -649,7 +754,7 @@ testBatchedSubscriptions t = do
     subscribe c cs = do
       r <- subscribeConnections c cs
       liftIO $ do
-        let dc = S.fromList $ take 10 cs
+        let dc = S.fromList $ take nDel cs
         all isRight (M.withoutKeys r dc) `shouldBe` True
         all (== Left (CONN NOT_FOUND)) (M.restrictKeys r dc) `shouldBe` True
         M.keys r `shouldMatchList` cs

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