agent: expire messages failed after quota exceeded after 7 days, and expire multiple messages at once (#973)

* agent: expire messages failed after quota exceeded after 7 days, and expire multiple messages at once

* fix, test

* refactor

* catch in loop

src/Simplex/Messaging/Agent.hs

@@ -1142,7 +1142,7 @@ submitPendingMsg c cData sq = do
 
 runSmpQueueMsgDelivery :: forall m. AgentMonad m => AgentClient -> ConnData -> SndQueue -> (Worker, TMVar ()) -> m ()
 runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, duplexHandshake} sq (Worker {doWork}, qLock) = do
-  ri <- asks $ messageRetryInterval . config
+  AgentConfig {messageRetryInterval = ri, messageTimeout, helloTimeout, quotaExceededTimeout} <- asks config
   forever $ do
     atomically $ endAgentOperation c AOSndNetwork
     waitForWork doWork
@@ -1166,7 +1166,9 @@ 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
-                  _ -> retrySndMsg RISlow
+                  _ -> do
+                    expireTs <- addUTCTime (-quotaExceededTimeout) <$> liftIO getCurrentTime
+                    if internalTs < expireTs then notifyDelMsgs msgId e expireTs else retrySndMsg RISlow
                 SMP SMP.AUTH -> case msgType of
                   AM_CONN_INFO -> connError msgId NOT_AVAILABLE
                   AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE
@@ -1175,8 +1177,11 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl
                     -- in duplexHandshake mode (v2) HELLO is only sent once, without retrying,
                     -- because the queue must be secured by the time the confirmation or the first HELLO is received
                     | duplexHandshake == Just True -> connErr
-                    | otherwise ->
-                        ifM (msgExpired helloTimeout) connErr (retrySndMsg RIFast)
+                    -- otherwise branch is not used in clients with v2+ of agent protocol (since June 2022)
+                    -- TODO remove in v6
+                    | otherwise -> do
+                        expireTs <- addUTCTime (-helloTimeout) <$> liftIO getCurrentTime
+                        if internalTs < expireTs then connErr else retrySndMsg RIFast
                     where
                       connErr = case rq_ of
                         -- party initiating connection
@@ -1196,14 +1201,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 -> do
-                      let timeoutSel = if msgType == AM_HELLO_ then helloTimeout else messageTimeout
-                      ifM (msgExpired timeoutSel) (notifyDel msgId err) (retrySndMsg RIFast)
+                      let msgTimeout = if msgType == AM_HELLO_ then helloTimeout else messageTimeout
+                      expireTs <- addUTCTime (-msgTimeout) <$> liftIO getCurrentTime
+                      if internalTs < expireTs then notifyDelMsgs msgId e expireTs else retrySndMsg RIFast
                   | otherwise -> notifyDel msgId err
               where
-                msgExpired timeoutSel = do
-                  msgTimeout <- asks $ timeoutSel . config
-                  currentTime <- liftIO getCurrentTime
-                  pure $ diffUTCTime currentTime internalTs > msgTimeout
                 retrySndMsg riMode = do
                   withStore' c $ \db -> updatePendingMsgRIState db connId msgId riState
                   retrySndOp c $ loop riMode
@@ -1279,6 +1281,13 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl
                     when (isJust rq_) $ removeConfirmations db connId
                   unless (duplexHandshake == Just True) . void $ enqueueMessage c cData sq SMP.noMsgFlags HELLO
   where
+    notifyDelMsgs :: InternalId -> AgentErrorType -> UTCTime -> m ()
+    notifyDelMsgs msgId err expireTs = do
+      notifyDel msgId $ MERR (unId msgId) err
+      msgIds_ <- withStore' c $ \db -> getExpiredSndMessages db connId sq expireTs
+      forM_ (L.nonEmpty msgIds_) $ \msgIds -> do
+        notify $ MERRS (L.map unId msgIds) err
+        withStore' c $ \db -> forM_ msgIds $ \msgId' -> deleteSndMsgDelivery db connId sq msgId' False `catchAll_` pure ()
     delMsg :: InternalId -> m ()
     delMsg = delMsgKeep False
     delMsgKeep :: Bool -> InternalId -> m ()

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

@@ -92,6 +92,7 @@ data AgentConfig = AgentConfig
     messageRetryInterval :: RetryInterval2,
     messageTimeout :: NominalDiffTime,
     helloTimeout :: NominalDiffTime,
+    quotaExceededTimeout :: NominalDiffTime,
     initialCleanupDelay :: Int64,
     cleanupInterval :: Int64,
     cleanupStepInterval :: Int,
@@ -135,13 +136,10 @@ defaultMessageRetryInterval =
             maxInterval = 60_000000
           },
       riSlow =
-        -- TODO: these timeouts can be increased in v5.0 once most clients are updated
-        -- to resume sending on QCONT messages.
-        -- After that local message expiration period should be also increased.
         RetryInterval
-          { initialInterval = 60_000000,
+          { initialInterval = 180_000000, -- 3 minutes
             increaseAfter = 60_000000,
-            maxInterval = 3600_000000 -- 1 hour
+            maxInterval = 3 * 3600_000000 -- 3 hours
           }
     }
 
@@ -159,6 +157,7 @@ defaultAgentConfig =
       messageRetryInterval = defaultMessageRetryInterval,
       messageTimeout = 2 * nominalDay,
       helloTimeout = 2 * nominalDay,
+      quotaExceededTimeout = 7 * nominalDay,
       initialCleanupDelay = 30 * 1000000, -- 30 seconds
       cleanupInterval = 30 * 60 * 1000000, -- 30 minutes
       cleanupStepInterval = 200000, -- 200ms

src/Simplex/Messaging/Agent/Protocol.hs

@@ -337,6 +337,7 @@ data ACommand (p :: AParty) (e :: AEntity) where
   MID :: AgentMsgId -> ACommand Agent AEConn
   SENT :: AgentMsgId -> ACommand Agent AEConn
   MERR :: AgentMsgId -> AgentErrorType -> ACommand Agent AEConn
+  MERRS :: NonEmpty AgentMsgId -> AgentErrorType -> ACommand Agent AEConn
   MSG :: MsgMeta -> MsgFlags -> MsgBody -> ACommand Agent AEConn
   MSGNTF :: SMPMsgMeta -> ACommand Agent AEConn
   ACK :: AgentMsgId -> Maybe MsgReceiptInfo -> ACommand Client AEConn
@@ -398,6 +399,7 @@ data ACommandTag (p :: AParty) (e :: AEntity) where
   MID_ :: ACommandTag Agent AEConn
   SENT_ :: ACommandTag Agent AEConn
   MERR_ :: ACommandTag Agent AEConn
+  MERRS_ :: ACommandTag Agent AEConn
   MSG_ :: ACommandTag Agent AEConn
   MSGNTF_ :: ACommandTag Agent AEConn
   ACK_ :: ACommandTag Client AEConn
@@ -452,6 +454,7 @@ aCommandTag = \case
   MID _ -> MID_
   SENT _ -> SENT_
   MERR {} -> MERR_
+  MERRS {} -> MERRS_
   MSG {} -> MSG_
   MSGNTF {} -> MSGNTF_
   ACK {} -> ACK_
@@ -1611,6 +1614,7 @@ instance StrEncoding ACmdTag where
       "MID" -> ct MID_
       "SENT" -> ct SENT_
       "MERR" -> ct MERR_
+      "MERRS" -> ct MERRS_
       "MSG" -> ct MSG_
       "MSGNTF" -> ct MSGNTF_
       "ACK" -> t ACK_
@@ -1667,6 +1671,7 @@ instance (APartyI p, AEntityI e) => StrEncoding (ACommandTag p e) where
     MID_ -> "MID"
     SENT_ -> "SENT"
     MERR_ -> "MERR"
+    MERRS_ -> "MERRS"
     MSG_ -> "MSG"
     MSGNTF_ -> "MSGNTF"
     ACK_ -> "ACK"
@@ -1736,6 +1741,7 @@ commandP binaryP =
           MID_ -> s (MID <$> A.decimal)
           SENT_ -> s (SENT <$> A.decimal)
           MERR_ -> s (MERR <$> A.decimal <* A.space <*> strP)
+          MERRS_ -> s (MERRS <$> strP_ <*> strP)
           MSG_ -> s (MSG <$> strP <* A.space <*> smpP <* A.space <*> binaryP)
           MSGNTF_ -> s (MSGNTF <$> strP)
           RCVD_ -> s (RCVD <$> strP <* A.space <*> strP)
@@ -1788,12 +1794,13 @@ serializeCommand = \case
   SWITCH dir phase srvs -> s (SWITCH_, dir, phase, srvs)
   RSYNC rrState cryptoErr cstats -> s (RSYNC_, rrState, cryptoErr, cstats)
   SEND msgFlags msgBody -> B.unwords [s SEND_, smpEncode msgFlags, serializeBinary msgBody]
-  MID mId -> s (MID_, Str $ bshow mId)
-  SENT mId -> s (SENT_, Str $ bshow mId)
-  MERR mId e -> s (MERR_, Str $ bshow mId, e)
+  MID mId -> s (MID_, mId)
+  SENT mId -> s (SENT_, mId)
+  MERR mId e -> s (MERR_, mId, e)
+  MERRS mIds e -> s (MERRS_, mIds, e)
   MSG msgMeta msgFlags msgBody -> B.unwords [s MSG_, s msgMeta, smpEncode msgFlags, serializeBinary msgBody]
   MSGNTF smpMsgMeta -> s (MSGNTF_, smpMsgMeta)
-  ACK mId rcptInfo_ -> s (ACK_, Str $ bshow mId) <> maybe "" (B.cons ' ' . serializeBinary) rcptInfo_
+  ACK mId rcptInfo_ -> s (ACK_, mId) <> maybe "" (B.cons ' ' . serializeBinary) rcptInfo_
   RCVD msgMeta rcpts -> s (RCVD_, msgMeta, rcpts)
   SWCH -> s SWCH_
   OFF -> s OFF_

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

@@ -110,6 +110,7 @@ module Simplex.Messaging.Agent.Store.SQLite
     getPendingQueueMsg,
     updatePendingMsgRIState,
     deletePendingMsgs,
+    getExpiredSndMessages,
     setMsgUserAck,
     getRcvMsg,
     getLastMsg,
@@ -1041,6 +1042,20 @@ deletePendingMsgs :: DB.Connection -> ConnId -> SndQueue -> IO ()
 deletePendingMsgs db connId SndQueue {dbQueueId} =
   DB.execute db "DELETE FROM snd_message_deliveries WHERE conn_id = ? AND snd_queue_id = ?" (connId, dbQueueId)
 
+getExpiredSndMessages :: DB.Connection -> ConnId -> SndQueue -> UTCTime -> IO [InternalId]
+getExpiredSndMessages db connId SndQueue {dbQueueId} expireTs =
+  map fromOnly
+    <$> DB.query
+      db
+      [sql|
+        SELECT d.internal_id
+        FROM snd_message_deliveries d
+        JOIN messages m ON m.conn_id = d.conn_id AND m.internal_id = d.internal_id
+        WHERE d.conn_id = ? AND d.snd_queue_id = ? AND d.failed = 0 AND m.internal_ts < ?
+        ORDER BY d.internal_id ASC
+      |]
+      (connId, dbQueueId, expireTs)
+
 setMsgUserAck :: DB.Connection -> ConnId -> InternalId -> IO (Either StoreError (RcvQueue, SMP.MsgId))
 setMsgUserAck db connId agentMsgId = runExceptT $ do
   (dbRcvId, srvMsgId) <-