agent: handle cases when last message ts is not set for notifications; set last ts for "stale" notifications when messages expired and queue is empty, to prevent repeated processing (#1531)

* agent: handle cases when last message ts is not set for notifications; set last ts for "stale" notifications when messages expired and queue is empty, to prevent repeated processing

* only log errors if they exist

* only set last ts for queue that delivered notification

Author: epoberezkin

src/Simplex/Messaging/Agent.hs

@@ -224,6 +224,7 @@ import Simplex.RemoteControl.Client
 import Simplex.RemoteControl.Invitation
 import Simplex.RemoteControl.Types
 import System.Mem.Weak (deRefWeak)
+import UnliftIO.Async (mapConcurrently)
 import UnliftIO.Concurrent (forkFinally, forkIO, killThread, mkWeakThreadId, threadDelay)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
@@ -439,7 +440,7 @@ subscribeConnections c = withAgentEnv c . subscribeConnections' c
 {-# INLINE subscribeConnections #-}
 
 -- | Get messages for connections (GET commands)
-getConnectionMessages :: AgentClient -> NonEmpty ConnId -> IO (NonEmpty (Maybe SMPMsgMeta))
+getConnectionMessages :: AgentClient -> NonEmpty ConnMsgReq -> IO (NonEmpty (Maybe SMPMsgMeta))
 getConnectionMessages c = withAgentEnv' c . getConnectionMessages' c
 {-# INLINE getConnectionMessages #-}
 
@@ -1276,24 +1277,26 @@ resubscribeConnections' c connIds = do
   -- union is left-biased, so results returned by subscribeConnections' take precedence
   (`M.union` r) <$> subscribeConnections' c connIds'
 
-getConnectionMessages' :: AgentClient -> NonEmpty ConnId -> AM' (NonEmpty (Maybe SMPMsgMeta))
-getConnectionMessages' c = mapM getMsg
+getConnectionMessages' :: AgentClient -> NonEmpty ConnMsgReq -> AM' (NonEmpty (Maybe SMPMsgMeta))
+getConnectionMessages' c =
+  mapConcurrently $ \cmr ->
+    getConnectionMessage cmr `catchAgentError'` \e -> do
+      logError $ "Error loading message: " <> tshow e
+      pure Nothing
   where
-    getMsg :: ConnId -> AM' (Maybe SMPMsgMeta)
-    getMsg connId =
-      getConnectionMessage connId `catchAgentError'` \e -> do
-        logError $ "Error loading message: " <> tshow e
-        pure Nothing
-    getConnectionMessage :: ConnId -> AM (Maybe SMPMsgMeta)
-    getConnectionMessage connId = do
+    getConnectionMessage :: ConnMsgReq -> AM (Maybe SMPMsgMeta)
+    getConnectionMessage (ConnMsgReq connId dbQueueId msgTs_) = do
       whenM (atomically $ hasActiveSubscription c connId) . throwE $ CMD PROHIBITED "getConnectionMessage: subscribed"
       SomeConn _ conn <- withStore c (`getConn` connId)
-      case conn of
+      msg_ <- case conn of
         DuplexConnection _ (rq :| _) _ -> getQueueMessage c rq
         RcvConnection _ rq -> getQueueMessage c rq
         ContactConnection _ rq -> getQueueMessage c rq
         SndConnection _ _ -> throwE $ CONN SIMPLEX
         NewConnection _ -> throwE $ CMD PROHIBITED "getConnectionMessage: NewConnection"
+      when (isNothing msg_) $
+        forM_ msgTs_ $ \msgTs -> withStore' c $ \db -> setLastBrokerTs db connId (DBQueueId dbQueueId) msgTs
+      pure msg_
 
 getNotificationConns' :: AgentClient -> C.CbNonce -> ByteString -> AM (NonEmpty NotificationInfo)
 getNotificationConns' c nonce encNtfInfo =
@@ -1308,24 +1311,24 @@ getNotificationConns' c nonce encNtfInfo =
               lastNtfInfo = Just . fst <$$> getNtfInfo db lastNtf
            in initNtfInfos <> [lastNtfInfo]
       let (errs, ntfInfos_) = partitionEithers rs
-      logError $ "Error(s) loading notifications: " <> tshow errs
+      unless (null errs) $ logError $ "Error(s) loading notifications: " <> tshow errs
       case L.nonEmpty $ catMaybes ntfInfos_ of
         Just r -> pure r
         Nothing -> throwE $ INTERNAL "getNotificationConns: couldn't get conn info"
     _ -> throwE $ CMD PROHIBITED "getNotificationConns"
   where
     getNtfInfo :: DB.Connection -> PNMessageData -> IO (Either AgentErrorType (NotificationInfo, Maybe UTCTime))
     getNtfInfo db PNMessageData {smpQueue, ntfTs, nmsgNonce, encNMsgMeta} = runExceptT $ do
-      (ntfConnId, rcvNtfDhSecret, lastBrokerTs_) <- liftError' storeError $ getNtfRcvQueue db smpQueue
+      (ntfConnId, ntfDbQueueId, rcvNtfDhSecret, lastBrokerTs_) <- liftError' storeError $ getNtfRcvQueue db smpQueue
       let ntfMsgMeta = eitherToMaybe $ smpDecode =<< first show (C.cbDecrypt rcvNtfDhSecret nmsgNonce encNMsgMeta)
-          ntfInfo = NotificationInfo {ntfConnId, ntfTs, ntfMsgMeta}
+          ntfInfo = NotificationInfo {ntfConnId, ntfDbQueueId, ntfTs, ntfMsgMeta}
       pure (ntfInfo, lastBrokerTs_)
     getInitNtfInfo :: DB.Connection -> PNMessageData -> IO (Either AgentErrorType (Maybe NotificationInfo))
     getInitNtfInfo db msgData = runExceptT $ do
-      (nftInfo, lastBrokerTs_) <- ExceptT $ getNtfInfo db msgData
-      pure $ case (ntfMsgMeta nftInfo, lastBrokerTs_) of
-        (Just SMP.NMsgMeta {msgTs}, Just lastBrokerTs)
-          | systemToUTCTime msgTs > lastBrokerTs -> Just nftInfo
+      (ntfInfo, lastBrokerTs_) <- ExceptT $ getNtfInfo db msgData
+      pure $ case ntfMsgMeta ntfInfo of
+        Just SMP.NMsgMeta {msgTs}
+          | maybe True (systemToUTCTime msgTs >) lastBrokerTs_ -> Just ntfInfo
         _ -> Nothing
 
 -- | Send message to the connection (SEND command) in Reader monad

src/Simplex/Messaging/Agent/Protocol.hs

@@ -147,6 +147,7 @@ module Simplex.Messaging.Agent.Protocol
     AgentMsgId,
     NotificationsMode (..),
     NotificationInfo (..),
+    ConnMsgReq (..),
 
     -- * Encode/decode
     serializeCommand,
@@ -678,11 +679,21 @@ instance FromField NotificationsMode where fromField = blobFieldDecoder $ parseA
 
 data NotificationInfo = NotificationInfo
   { ntfConnId :: ConnId,
+    ntfDbQueueId :: Int64,
     ntfTs :: SystemTime,
+    -- Nothing means that the message failed to decrypt or to decode,
+    -- we can still show event notification
     ntfMsgMeta :: Maybe NMsgMeta
   }
   deriving (Show)
 
+data ConnMsgReq = ConnMsgReq
+  { msgConnId :: ConnId,
+    msgDbQueueId :: Int64,
+    msgTs :: Maybe UTCTime
+  }
+  deriving (Show)
+
 data ConnectionMode = CMInvitation | CMContact
   deriving (Eq, Show)
 

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

@@ -98,6 +98,7 @@ module Simplex.Messaging.Agent.Store.AgentStore
     -- Messages
     updateRcvIds,
     createRcvMsg,
+    setLastBrokerTs,
     updateRcvMsgHash,
     createSndMsgBody,
     updateSndIds,
@@ -855,7 +856,11 @@ createRcvMsg db connId rq@RcvQueue {dbQueueId} rcvMsgData@RcvMsgData {msgMeta =
   insertRcvMsgBase_ db connId rcvMsgData
   insertRcvMsgDetails_ db connId rq rcvMsgData
   updateRcvMsgHash db connId sndMsgId internalRcvId internalHash
-  DB.execute db "UPDATE rcv_queues SET last_broker_ts = ? WHERE conn_id = ? AND rcv_queue_id = ?" (brokerTs, connId, dbQueueId)
+  setLastBrokerTs db connId dbQueueId brokerTs
+
+setLastBrokerTs :: DB.Connection -> ConnId -> DBQueueId 'QSStored -> UTCTime -> IO ()
+setLastBrokerTs db connId dbQueueId brokerTs =
+  DB.execute db "UPDATE rcv_queues SET last_broker_ts = ? WHERE conn_id = ? AND rcv_queue_id = ? AND last_broker_ts < ?" (brokerTs, connId, dbQueueId, brokerTs)
 
 createSndMsgBody :: DB.Connection -> AMessage -> IO Int64
 createSndMsgBody db aMessage =
@@ -1781,19 +1786,19 @@ getActiveNtfToken db =
           ntfMode = fromMaybe NMPeriodic ntfMode_
        in NtfToken {deviceToken = DeviceToken provider dt, ntfServer, ntfTokenId, ntfPubKey, ntfPrivKey, ntfDhKeys, ntfDhSecret, ntfTknStatus, ntfTknAction, ntfMode}
 
-getNtfRcvQueue :: DB.Connection -> SMPQueueNtf -> IO (Either StoreError (ConnId, RcvNtfDhSecret, Maybe UTCTime))
+getNtfRcvQueue :: DB.Connection -> SMPQueueNtf -> IO (Either StoreError (ConnId, Int64, RcvNtfDhSecret, Maybe UTCTime))
 getNtfRcvQueue db SMPQueueNtf {smpServer = (SMPServer host port _), notifierId} =
   firstRow' res SEConnNotFound $
     DB.query
       db
       [sql|
-        SELECT conn_id, rcv_ntf_dh_secret, last_broker_ts
+        SELECT conn_id, rcv_queue_id, rcv_ntf_dh_secret, last_broker_ts
         FROM rcv_queues
         WHERE host = ? AND port = ? AND ntf_id = ? AND deleted = 0
       |]
       (host, port, notifierId)
   where
-    res (connId, Just rcvNtfDhSecret, lastBrokerTs_) = Right (connId, rcvNtfDhSecret, lastBrokerTs_)
+    res (connId, dbQueueId, Just rcvNtfDhSecret, lastBrokerTs_) = Right (connId, dbQueueId, rcvNtfDhSecret, lastBrokerTs_)
     res _ = Left SEConnNotFound
 
 setConnectionNtfs :: DB.Connection -> ConnId -> Bool -> IO ()

tests/AgentTests/FunctionalAPITests.hs

@@ -1946,7 +1946,7 @@ testOnlyCreatePullSlowHandshake = withAgentClientsCfg2 agentProxyCfgV8 agentProx
 getMsg :: AgentClient -> ConnId -> ExceptT AgentErrorType IO a -> ExceptT AgentErrorType IO a
 getMsg c cId action = do
   liftIO $ noMessages c "nothing should be delivered before GET"
-  [Just _] <- lift $ getConnectionMessages c [cId]
+  [Just _] <- lift $ getConnectionMessages c [ConnMsgReq cId 1 Nothing]
   action
 
 getMSGNTF :: AgentClient -> ConnId -> ExceptT AgentErrorType IO ()

tests/AgentTests/NotificationTests.hs

@@ -57,6 +57,7 @@ import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Text.Encoding (encodeUtf8)
 import qualified Data.Text.IO as TIO
+import Data.Time.Clock.System (systemToUTCTime)
 import NtfClient
 import SMPAgentClient (agentCfg, initAgentServers, initAgentServers2, testDB, testDB2, testNtfServer, testNtfServer2)
 import SMPClient (cfgMS, cfgJ2QS, cfgVPrev, serverStoreConfig, testPort, testPort2, withSmpServer, withSmpServerConfigOn, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn, xit'')
@@ -75,7 +76,7 @@ import Simplex.Messaging.Notifications.Server.Env (NtfServerConfig (..))
 import Simplex.Messaging.Notifications.Server.Push.APNS
 import Simplex.Messaging.Notifications.Types (NtfTknAction (..), NtfToken (..))
 import Simplex.Messaging.Parsers (parseAll)
-import Simplex.Messaging.Protocol (ErrorType (AUTH), MsgFlags (MsgFlags), NtfServer, ProtocolServer (..), SMPMsgMeta (..), SubscriptionMode (..))
+import Simplex.Messaging.Protocol (ErrorType (AUTH), MsgFlags (MsgFlags), NMsgMeta (..), NtfServer, ProtocolServer (..), SMPMsgMeta (..), SubscriptionMode (..))
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Server.Env.STM (AStoreType (..), ServerConfig (..))
 import Simplex.Messaging.Transport (ATransport)
@@ -558,11 +559,11 @@ testNotificationSubscriptionExistingConnection apns baseId alice@AgentClient {ag
     (nonce, message) <- messageNotification apns tkn
     pure (bobId, aliceId, nonce, message)
 
-  Right [NotificationInfo {ntfConnId = cId}] <- runExceptT $ getNotificationConns alice nonce message
+  Right [NotificationInfo {ntfConnId = cId, ntfMsgMeta = Just NMsgMeta {msgTs}}] <- runExceptT $ getNotificationConns alice nonce message
   cId `shouldBe` bobId
   -- alice client already has subscription for the connection,
   -- so get fails with CMD PROHIBITED (transformed into Nothing in catch)
-  [Nothing] <- getConnectionMessages alice [cId]
+  [Nothing] <- getConnectionMessages alice [ConnMsgReq cId 1 $ Just $ systemToUTCTime msgTs]
 
   threadDelay 500000
   suspendAgent alice 0
@@ -572,7 +573,7 @@ testNotificationSubscriptionExistingConnection apns baseId alice@AgentClient {ag
 
   -- aliceNtf client doesn't have subscription and is allowed to get notification message
   withAgent 3 aliceCfg initAgentServers testDB $ \aliceNtf -> do
-    (Just SMPMsgMeta {msgFlags = MsgFlags True}) :| _ <- getConnectionMessages aliceNtf [cId]
+    (Just SMPMsgMeta {msgFlags = MsgFlags True}) :| _ <- getConnectionMessages aliceNtf [ConnMsgReq cId 1 $ Just $ systemToUTCTime msgTs]
     pure ()
 
   threadDelay 1000000