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 (..))