5.3.0.1

* agent: partially fix message queries, debug message times

* fix more joins

package.yaml

@@ -1,5 +1,5 @@
 name: simplexmq
-version: 5.3.0.0
+version: 5.3.0.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.3.0.0
+version:        5.3.0.1
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -85,6 +85,8 @@ library
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230531_switch_status
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230615_ratchet_sync
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230701_delivery_receipts
+      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230720_delete_expired_messages
+      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230722_indexes
       Simplex.Messaging.Agent.TAsyncs
       Simplex.Messaging.Agent.TRcvQueues
       Simplex.Messaging.Client

src/Simplex/Messaging/Agent.hs

@@ -519,9 +519,9 @@ ackMessageAsync' c corrId connId msgId rcptInfo_ = do
     enqueueAck :: m ()
     enqueueAck = do
       let mId = InternalId msgId
-      RcvMsg {msgType} <- withStore c $ \db -> getRcvMsg db connId mId
+      RcvMsg {msgType} <- withStoreCtx "ackMessageAsync': getRcvMsg" c $ \db -> getRcvMsg db connId mId
       when (isJust rcptInfo_ && msgType /= AM_A_MSG_) $ throwError $ CMD PROHIBITED
-      (RcvQueue {server}, _) <- withStore c $ \db -> setMsgUserAck db connId mId
+      (RcvQueue {server}, _) <- withStoreCtx "ackMessageAsync': setMsgUserAck" c $ \db -> setMsgUserAck db connId mId
       enqueueCommand c corrId connId (Just server) . AClientCommand $ APC SAEConn $ ACK msgId rcptInfo_
 
 deleteConnectionAsync' :: forall m. AgentMonad m => AgentClient -> ConnId -> m ()
@@ -1095,7 +1095,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl
                 SMP SMP.QUOTA -> case msgType of
                   AM_CONN_INFO -> connError msgId NOT_AVAILABLE
                   AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE
-                  _ -> retryIfNotExpired RISlow err
+                  _ -> retrySndMsg RISlow
                 SMP SMP.AUTH -> case msgType of
                   AM_CONN_INFO -> connError msgId NOT_AVAILABLE
                   AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE
@@ -1124,12 +1124,11 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl
                 _
                   -- for other operations BROKER HOST is treated as a permanent error (e.g., when connecting to the server),
                   -- the message sending would be retried
-                  | temporaryOrHostError e -> retryIfNotExpired RIFast err
+                  | temporaryOrHostError e -> do
+                    let timeoutSel = if msgType == AM_HELLO_ then helloTimeout else messageTimeout
+                    ifM (msgExpired timeoutSel) (notifyDel msgId err) (retrySndMsg RIFast)
                   | otherwise -> notifyDel msgId err
               where
-                retryIfNotExpired riMode err = do
-                  let timeoutSel = if msgType == AM_HELLO_ then helloTimeout else messageTimeout
-                  ifM (msgExpired timeoutSel) (notifyDel msgId err) (retrySndMsg riMode)
                 msgExpired timeoutSel = do
                   msgTimeout <- asks $ timeoutSel . config
                   currentTime <- liftIO getCurrentTime
@@ -1240,13 +1239,13 @@ ackMessage' c connId msgId rcptInfo_ = withConnLock c connId "ackMessage" $ do
     ack :: m ()
     ack = do
       -- the stored message was delivered via a specific queue, the rest failed to decrypt and were already acknowledged
-      (rq, srvMsgId) <- withStore c $ \db -> setMsgUserAck db connId $ InternalId msgId
+      (rq, srvMsgId) <- withStoreCtx "ackMessage': setMsgUserAck" c $ \db -> setMsgUserAck db connId $ InternalId msgId
       ackQueueMessage c rq srvMsgId
     del :: m ()
-    del = withStore' c $ \db -> deleteMsg db connId $ InternalId msgId
+    del = withStoreCtx' "ackMessage': deleteMsg" c $ \db -> deleteMsg db connId $ InternalId msgId
     sendRcpt :: Connection 'CDuplex -> m ()
     sendRcpt (DuplexConnection cData _ sqs) = do
-      msg@RcvMsg {msgType, msgReceipt} <- withStore c $ \db -> getRcvMsg db connId $ InternalId msgId
+      msg@RcvMsg {msgType, msgReceipt} <- withStoreCtx "ackMessage': getRcvMsg" c $ \db -> getRcvMsg db connId $ InternalId msgId
       case rcptInfo_ of
         Just rcptInfo -> do
           unless (msgType == AM_A_MSG_) $ throwError (CMD PROHIBITED)
@@ -1257,7 +1256,7 @@ ackMessage' c connId msgId rcptInfo_ = withConnLock c connId "ackMessage" $ do
         Nothing -> case (msgType, msgReceipt) of
           -- only remove sent message if receipt hash was Ok, both to debug and for future redundancy
           (AM_A_RCVD_, Just MsgReceipt {agentMsgId = sndMsgId, msgRcptStatus = MROk}) ->
-            withStore' c $ \db -> deleteDeliveredSndMsg db connId $ InternalId sndMsgId
+            withStoreCtx' "ackMessage': deleteDeliveredSndMsg" c $ \db -> deleteDeliveredSndMsg db connId $ InternalId sndMsgId
           _ -> pure ()
 
 switchConnection' :: AgentMonad m => AgentClient -> ConnId -> m ConnectionStats
@@ -1901,7 +1900,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
                                 | otherwise = pure conn'
                           Right _ -> prohibited >> ack
                           Left e@(AGENT A_DUPLICATE) -> do
-                            withStore' c (\db -> getLastMsg db connId srvMsgId) >>= \case
+                            withStoreCtx' "processSMP: getLastMsg" c (\db -> getLastMsg db connId srvMsgId) >>= \case
                               Just RcvMsg {internalId, msgMeta, msgBody = agentMsgBody, userAck}
                                 | userAck -> ackDel internalId
                                 | otherwise -> do
@@ -2113,7 +2112,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
               clientReceipt :: AMessageReceipt -> m (Maybe MsgReceipt)
               clientReceipt AMessageReceipt {agentMsgId, msgHash} = do
                 let sndMsgId = InternalSndId agentMsgId
-                SndMsg {internalId = InternalId msgId, msgType, internalHash, msgReceipt} <- withStore c $ \db -> getSndMsgViaRcpt db connId sndMsgId
+                SndMsg {internalId = InternalId msgId, msgType, internalHash, msgReceipt} <- withStoreCtx "messagesRcvd: getSndMsgViaRcpt" c $ \db -> getSndMsgViaRcpt db connId sndMsgId
                 if msgType /= AM_A_MSG_
                   then notify (ERR $ AGENT A_PROHIBITED) $> Nothing -- unexpected message type for receipt
                   else case msgReceipt of

src/Simplex/Messaging/Agent/Client.hs

@@ -90,6 +90,8 @@ module Simplex.Messaging.Agent.Client
     whenSuspending,
     withStore,
     withStore',
+    withStoreCtx,
+    withStoreCtx',
     storeError,
     userServers,
     pickServer,
@@ -127,6 +129,7 @@ import Data.Set (Set)
 import qualified Data.Set as S
 import Data.Text.Encoding
 import Data.Time (UTCTime, defaultTimeLocale, formatTime, getCurrentTime)
+import Data.Time.Clock (diffUTCTime)
 import Data.Word (Word16)
 import qualified Database.SQLite.Simple as DB
 import GHC.Generics (Generic)
@@ -1236,16 +1239,37 @@ waitUntilForeground :: AgentClient -> STM ()
 waitUntilForeground c = unlessM ((ASForeground ==) <$> readTVar (agentState c)) retry
 
 withStore' :: AgentMonad m => AgentClient -> (DB.Connection -> IO a) -> m a
-withStore' c action = withStore c $ fmap Right . action
+withStore' = withStoreCtx_' Nothing
 
 withStore :: AgentMonad m => AgentClient -> (DB.Connection -> IO (Either StoreError a)) -> m a
-withStore c action = do
+withStore = withStoreCtx_ Nothing
+
+withStoreCtx' :: AgentMonad m => String -> AgentClient -> (DB.Connection -> IO a) -> m a
+withStoreCtx' = withStoreCtx_' . Just
+
+withStoreCtx :: AgentMonad m => String -> AgentClient -> (DB.Connection -> IO (Either StoreError a)) -> m a
+withStoreCtx = withStoreCtx_ . Just
+
+withStoreCtx_' :: AgentMonad m => Maybe String -> AgentClient -> (DB.Connection -> IO a) -> m a
+withStoreCtx_' ctx_ c action = withStoreCtx_ ctx_ c $ fmap Right . action
+
+withStoreCtx_ :: AgentMonad m => Maybe String -> AgentClient -> (DB.Connection -> IO (Either StoreError a)) -> m a
+withStoreCtx_ ctx_ c action = do
   st <- asks store
-  liftEitherError storeError . agentOperationBracket c AODatabase (\_ -> pure ()) $
-    withTransaction st action `E.catch` handleInternal
+  liftEitherError storeError . agentOperationBracket c AODatabase (\_ -> pure ()) $ case ctx_ of
+    Nothing -> withTransaction st action `E.catch` handleInternal ""
+    -- uncomment to debug store performance
+    -- Just ctx -> do
+    --   t1 <- liftIO getCurrentTime
+    --   putStrLn $ "agent withStoreCtx start       :: " <> show t1 <> " :: " <> ctx
+    --   r <- withTransaction st action `E.catch` handleInternal (" (" <> ctx <> ")")
+    --   t2 <- liftIO getCurrentTime
+    --   putStrLn $ "agent withStoreCtx end         :: " <> show t2 <> " :: " <> ctx <> " :: duration=" <> show (diffToMilliseconds $ diffUTCTime t2 t1)
+    --   pure r
+    Just _ -> withTransaction st action `E.catch` handleInternal ""
   where
-    handleInternal :: E.SomeException -> IO (Either StoreError a)
-    handleInternal = pure . Left . SEInternal . bshow
+    handleInternal :: String -> E.SomeException -> IO (Either StoreError a)
+    handleInternal ctxStr e = pure . Left . SEInternal . B.pack $ show e <> ctxStr
 
 storeError :: StoreError -> AgentErrorType
 storeError = \case

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

@@ -906,7 +906,7 @@ getSndMsgViaRcpt db connId sndMsgId =
       [sql|
         SELECT s.internal_id, m.msg_type, s.internal_hash, s.rcpt_internal_id, s.rcpt_status
         FROM snd_messages s
-        JOIN messages m ON s.internal_id = m.internal_id
+        JOIN messages m ON s.conn_id = m.conn_id AND s.internal_id = m.internal_id
         WHERE s.conn_id = ? AND s.internal_snd_id = ?
       |]
       (connId, sndMsgId)
@@ -976,8 +976,8 @@ getRcvMsg db connId agentMsgId =
           r.internal_id, m.internal_ts, r.broker_id, r.broker_ts, r.external_snd_id, r.integrity, r.internal_hash,
           m.msg_type, m.msg_body, s.internal_id, s.rcpt_status, r.user_ack
         FROM rcv_messages r
-        JOIN messages m ON r.internal_id = m.internal_id
-        LEFT JOIN snd_messages s ON s.rcpt_internal_id = r.internal_id
+        JOIN messages m ON r.conn_id = m.conn_id AND r.internal_id = m.internal_id
+        LEFT JOIN snd_messages s ON s.conn_id = r.conn_id AND s.rcpt_internal_id = r.internal_id
         WHERE r.conn_id = ? AND r.internal_id = ?
       |]
       (connId, agentMsgId)
@@ -992,9 +992,9 @@ getLastMsg db connId msgId =
           r.internal_id, m.internal_ts, r.broker_id, r.broker_ts, r.external_snd_id, r.integrity, r.internal_hash,
           m.msg_type, m.msg_body, s.internal_id, s.rcpt_status, r.user_ack
         FROM rcv_messages r
-        JOIN messages m ON r.internal_id = m.internal_id
+        JOIN messages m ON r.conn_id = m.conn_id AND r.internal_id = m.internal_id
         JOIN connections c ON r.conn_id = c.conn_id AND c.last_internal_msg_id = r.internal_id
-        LEFT JOIN snd_messages s ON s.rcpt_internal_id = r.internal_id
+        LEFT JOIN snd_messages s ON s.conn_id = r.conn_id AND s.rcpt_internal_id = r.internal_id
         WHERE r.conn_id = ? AND r.broker_id = ?
       |]
       (connId, msgId)
@@ -1058,15 +1058,7 @@ deleteSndMsgsExpired db ttl = do
   cutoffTs <- addUTCTime (- ttl) <$> getCurrentTime
   DB.execute
     db
-    [sql|
-      DELETE FROM messages
-      WHERE internal_id IN (
-        SELECT s.internal_id
-        FROM snd_messages s
-        JOIN messages m USING (internal_id)
-        WHERE m.internal_ts < ?
-      )
-    |]
+    "DELETE FROM messages WHERE internal_snd_id IS NOT NULL AND internal_ts < ?"
     (Only cutoffTs)
 
 createRatchetX3dhKeys :: DB.Connection -> ConnId -> C.PrivateKeyX448 -> C.PrivateKeyX448 -> IO ()

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

@@ -64,6 +64,8 @@ import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230516_encrypted_rcv_m
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230531_switch_status
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230615_ratchet_sync
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230701_delivery_receipts
+import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230720_delete_expired_messages
+import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230722_indexes
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (dropPrefix, sumTypeJSON)
 import Simplex.Messaging.Transport.Client (TransportHost)
@@ -93,7 +95,9 @@ schemaMigrations =
     ("m20230516_encrypted_rcv_message_hashes", m20230516_encrypted_rcv_message_hashes, Just down_m20230516_encrypted_rcv_message_hashes),
     ("m20230531_switch_status", m20230531_switch_status, Just down_m20230531_switch_status),
     ("m20230615_ratchet_sync", m20230615_ratchet_sync, Just down_m20230615_ratchet_sync),
-    ("m20230701_delivery_receipts", m20230701_delivery_receipts, Just down_m20230701_delivery_receipts)
+    ("m20230701_delivery_receipts", m20230701_delivery_receipts, Just down_m20230701_delivery_receipts),
+    ("m20230720_delete_expired_messages", m20230720_delete_expired_messages, Just down_m20230720_delete_expired_messages),
+    ("m20230722_indexes", m20230722_indexes, Just down_m20230722_indexes)
   ]
 
 -- | The list of migrations in ascending order by date

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

@@ -0,0 +1,20 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230720_delete_expired_messages where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20230720_delete_expired_messages :: Query
+m20230720_delete_expired_messages =
+  [sql|
+CREATE INDEX idx_messages_internal_snd_id_ts ON messages(internal_snd_id, internal_ts);
+
+DELETE FROM messages WHERE internal_snd_id IS NOT NULL AND internal_ts < datetime('now', '-3 days');
+|]
+
+down_m20230720_delete_expired_messages :: Query
+down_m20230720_delete_expired_messages =
+  [sql|
+DROP INDEX idx_messages_internal_snd_id_ts;
+|]

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

@@ -0,0 +1,20 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230722_indexes where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20230722_indexes :: Query
+m20230722_indexes =
+  [sql|
+CREATE INDEX idx_processed_ratchet_key_hashes_created_at ON processed_ratchet_key_hashes(created_at);
+CREATE INDEX idx_encrypted_rcv_message_hashes_created_at ON encrypted_rcv_message_hashes(created_at);
+|]
+
+down_m20230722_indexes :: Query
+down_m20230722_indexes =
+  [sql|
+DROP INDEX idx_encrypted_rcv_message_hashes_created_at;
+DROP INDEX idx_processed_ratchet_key_hashes_created_at;
+|]

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

@@ -467,3 +467,13 @@ CREATE INDEX idx_snd_messages_rcpt_internal_id ON snd_messages(
   conn_id,
   rcpt_internal_id
 );
+CREATE INDEX idx_messages_internal_snd_id_ts ON messages(
+  internal_snd_id,
+  internal_ts
+);
+CREATE INDEX idx_processed_ratchet_key_hashes_created_at ON processed_ratchet_key_hashes(
+  created_at
+);
+CREATE INDEX idx_encrypted_rcv_message_hashes_created_at ON encrypted_rcv_message_hashes(
+  created_at
+);

tests/XFTPAgent.hs

@@ -16,7 +16,6 @@ import Data.Int (Int64)
 import Data.List (find, isSuffixOf)
 import Data.Maybe (fromJust)
 import SMPAgentClient (agentCfg, initAgentServers, testDB)
-import SMPClient (xit'')
 import Simplex.FileTransfer.Description
 import Simplex.FileTransfer.Protocol (FileParty (..), XFTPErrorType (AUTH))
 import Simplex.FileTransfer.Server.Env (XFTPServerConfig (..))
@@ -37,7 +36,7 @@ xftpAgentTests = around_ testBracket . describe "Functional API" $ do
   it "should send and receive file" testXFTPAgentSendReceive
   it "should resume receiving file after restart" testXFTPAgentReceiveRestore
   it "should cleanup rcv tmp path after permanent error" testXFTPAgentReceiveCleanup
-  xit'' "should resume sending file after restart" testXFTPAgentSendRestore
+  it "should resume sending file after restart" testXFTPAgentSendRestore
   it "should cleanup snd prefix path after permanent error" testXFTPAgentSendCleanup
   it "should delete sent file on server" testXFTPAgentDelete
   it "should resume deleting file after restart" testXFTPAgentDeleteRestore
@@ -255,6 +254,7 @@ testXFTPAgentSendRestore = withGlobalLogging logCfgNoLogs $ do
     liftIO $ sfId' `shouldBe` sfId
 
     -- prefix path should be removed after sending file
+    threadDelay 100000
     doesDirectoryExist prefixPath `shouldReturn` False
     doesFileExist encPath `shouldReturn` False