agent: make QUOTA error temporary for async commands (#1881)

* agent: make QUOTA error temporary for async commands

* retry commands on quota

---------

Co-authored-by: Evgeny @ SimpleX Chat <[email protected]>

src/Simplex/Messaging/Agent.hs

@@ -2168,20 +2168,21 @@ data CommandCompletion = CCMoved | CCCompleted
 
 runCommandProcessing :: AgentClient -> ConnId -> Maybe SMPServer -> Worker -> AM ()
 runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
-  ri <- asks $ messageRetryInterval . config -- different retry interval?
+  AgentConfig {messageRetryInterval = RetryInterval2 {riFast}, commandQuotaRetryInterval} <- asks config
+  let ri = RetryInterval2 {riSlow = commandQuotaRetryInterval, riFast}
   forever $ do
     atomically $ endAgentOperation c AOSndNetwork
     lift $ waitForWork doWork
     liftIO $ throwWhenInactive c
     atomically $ beginAgentOperation c AOSndNetwork
-    withWork c doWork (\db -> getPendingServerCommand db connId server_) $ runProcessCmd (riFast ri)
+    withWork c doWork (\db -> getPendingServerCommand db connId server_) $ runProcessCmd ri
   where
     runProcessCmd ri cmd = do
       pending <- newTVarIO []
       processCmd ri cmd pending
       mapM_ (atomically . writeTBQueue subQ) . reverse =<< readTVarIO pending
-    processCmd :: RetryInterval -> PendingCommand -> TVar [ATransmission] -> AM ()
-    processCmd ri PendingCommand {cmdId, corrId, userId, command} pendingCmds = case command of
+    processCmd :: RetryInterval2 -> PendingCommand -> TVar [ATransmission] -> AM ()
+    processCmd ri PendingCommand {cmdId, corrId, userId, command, createdAt} pendingCmds = case command of
       AClientCommand cmd -> case cmd of
         NEW enableNtfs (ACM cMode) pqEnc subMode useDR -> noServer $ do
           triedHosts <- newTVarIO S.empty
@@ -2376,12 +2377,18 @@ runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
             SomeConn _ conn@DuplexConnection {} -> a conn
             _ -> internalErr "command requires duplex connection"
         tryCommand action = tryMoveableCommand (action $> CCCompleted)
-        tryMoveableCommand action = withRetryInterval ri $ \_ loop -> do
+        tryMoveableCommand action = withRetryInterval2 ri $ \_ loop -> do
           liftIO $ waitWhileSuspended c
           liftIO $ waitForUserNetwork c
           tryAllErrors action >>= \case
+            Left e@(SMP _ SMP.QUOTA) -> do
+              AgentConfig {quotaExceededTimeout} <- asks config
+              expireTs <- addUTCTime (-quotaExceededTimeout) <$> liftIO getCurrentTime
+              if createdAt < expireTs
+                then cmdError e
+                else retrySndOp c $ loop RISlow
             Left e
-              | temporaryOrHostError e -> retrySndOp c loop
+              | temporaryOrHostError e -> retrySndOp c $ loop RIFast
               | otherwise -> cmdError e
             Right CCCompleted -> withStore' c (`deleteCommand` cmdId)
             Right CCMoved -> pure () -- command processing moved to another command queue

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

@@ -162,6 +162,7 @@ data AgentConfig = AgentConfig
     connDeleteDeliveryTimeout :: NominalDiffTime,
     helloTimeout :: NominalDiffTime,
     quotaExceededTimeout :: NominalDiffTime,
+    commandQuotaRetryInterval :: RetryInterval,
     persistErrorInterval :: NominalDiffTime,
     initialCleanupDelay :: Int64,
     cleanupInterval :: Int64,
@@ -241,6 +242,7 @@ defaultAgentConfig =
       connDeleteDeliveryTimeout = 2 * nominalDay,
       helloTimeout = 2 * nominalDay,
       quotaExceededTimeout = 7 * nominalDay,
+      commandQuotaRetryInterval = RetryInterval {initialInterval = 30_000000, increaseAfter = 0, maxInterval = 3600_000000},
       persistErrorInterval = 3, -- seconds
       initialCleanupDelay = 30 * 1000000, -- 30 seconds
       cleanupInterval = 5 * 60 * 1000000, -- 5 minutes

src/Simplex/Messaging/Agent/RetryInterval.hs

@@ -10,6 +10,7 @@ module Simplex.Messaging.Agent.RetryInterval
     withRetryInterval,
     withRetryIntervalCount,
     withRetryForeground,
+    withRetryInterval2,
     withRetryLock2,
     updateRetryInterval2,
     nextRetryDelay,
@@ -86,9 +87,24 @@ withRetryForeground ri isForeground isOnline action = callAction 0 $ initialInte
                 | otherwise = (elapsed + delay, nextRetryDelay elapsed' delay ri)
           callAction elapsed' delay'
 
+withRetryInterval2 :: forall m. MonadIO m => RetryInterval2 -> (RI2State -> (RetryIntervalMode -> m ()) -> m ()) -> m ()
+withRetryInterval2 = withRetryWait2 $ liftIO . threadDelay'
+
 -- This function allows action to toggle between slow and fast retry intervals.
 withRetryLock2 :: forall m. MonadIO m => RetryInterval2 -> TMVar () -> (RI2State -> (RetryIntervalMode -> m ()) -> m ()) -> m ()
-withRetryLock2 RetryInterval2 {riSlow, riFast} lock action =
+withRetryLock2 ri lock = withRetryWait2 wait ri
+  where
+    wait delay = do
+      waiting <- newTVarIO True
+      _ <- liftIO . forkIO $ do
+        threadDelay' delay
+        atomically $ whenM (readTVar waiting) $ void $ tryPutTMVar lock ()
+      atomically $ do
+        takeTMVar lock
+        writeTVar waiting False
+
+withRetryWait2 :: forall m. Monad m => (Int64 -> m ()) -> RetryInterval2 -> (RI2State -> (RetryIntervalMode -> m ()) -> m ()) -> m ()
+withRetryWait2 wait RetryInterval2 {riSlow, riFast} action =
   callAction (0, initialInterval riSlow) (0, initialInterval riFast)
   where
     callAction :: (Int64, Int64) -> (Int64, Int64) -> m ()
@@ -102,14 +118,6 @@ withRetryLock2 RetryInterval2 {riSlow, riFast} lock action =
           let elapsed' = elapsed + delay
               delay' = nextRetryDelay elapsed' delay ri
           call (elapsed', delay')
-        wait delay = do
-          waiting <- newTVarIO True
-          _ <- liftIO . forkIO $ do
-            threadDelay' delay
-            atomically $ whenM (readTVar waiting) $ void $ tryPutTMVar lock ()
-          atomically $ do
-            takeTMVar lock
-            writeTVar waiting False
 
 nextRetryDelay :: Int64 -> Int64 -> RetryInterval -> Int64
 nextRetryDelay elapsed delay RetryInterval {increaseAfter, maxInterval} =

src/Simplex/Messaging/Agent/Store.hs

@@ -489,7 +489,8 @@ data PendingCommand = PendingCommand
     corrId :: ACorrId,
     userId :: UserId,
     connId :: ConnId,
-    command :: AgentCommand
+    command :: AgentCommand,
+    createdAt :: UTCTime
   }
 
 data AgentCmdType = ACClient | ACInternal

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

@@ -1718,7 +1718,7 @@ getPendingServerCommand db connId srv_ = getWorkItem "command" getCmdId getComma
         DB.query
           db
           [sql|
-            SELECT c.corr_id, cs.user_id, c.command
+            SELECT c.corr_id, cs.user_id, c.command, c.created_at
             FROM commands c
             JOIN connections cs USING (conn_id)
             WHERE c.command_id = ?
@@ -1726,7 +1726,7 @@ getPendingServerCommand db connId srv_ = getWorkItem "command" getCmdId getComma
           (Only cmdId)
       where
         err = SEInternal $ "command  " <> bshow cmdId <> " returned []"
-        pendingCommand (corrId, userId, command) = PendingCommand {cmdId, corrId, userId, connId, command}
+        pendingCommand (corrId, userId, command, createdAt) = PendingCommand {cmdId, corrId, userId, connId, command, createdAt}
     markCommandFailed cmdId = DB.execute db "UPDATE commands SET failed = 1 WHERE command_id = ?" (Only cmdId)
 
 updateCommandServer :: DB.Connection -> AsyncCmdId -> SMPServer -> IO (Either StoreError ())

tests/AgentTests/FunctionalAPITests.hs

@@ -442,6 +442,9 @@ functionalAPITests ps = do
       it "should expire multiple messages" $ testExpireManyMessages ps
       it "should expire one message if quota is exceeded" $ testExpireMessageQuota ps
       it "should expire multiple messages if quota is exceeded" $ testExpireManyMessagesQuota ps
+    describe "joining a full contact address" $ do
+      it "should retry joining until the address has space" $ testJoinFullContactAsync ps
+      it "should fail joining after quota exceeded timeout" $ testJoinFullContactAsyncExpire ps
     it "should drop message after too many receive attempts" $ testDropMsgAfterRcvAttempts ps
 #if !defined(dbPostgres)
     -- TODO [postgres] restore from outdated db backup (we use copyFile/renameFile for sqlite)
@@ -2484,6 +2487,49 @@ testExpireManyMessagesQuota (t, msType) = withSmpServerConfigOn t cfg' testPort
   where
     cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {msgQueueQuota = 1, maxJournalMsgCount = 2}
 
+testJoinFullContactAsync :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
+testJoinFullContactAsync (t, msType) = withSmpServerConfigOn t cfg' testPort $ \_ -> do
+  (contactId, qInfo) <- fillContactAddress
+  withAgent 2 agentCfg {commandQuotaRetryInterval = fastRetryInterval} initAgentServers testDB2 $ \bob -> do
+    aliceId <- runRight $ do
+      (aliceId, _) <- A.prepareConnectionToJoin bob 1 True qInfo PQSupportOn
+      A.joinConnectionAsync bob "2" False aliceId True qInfo "bob's connInfo" PQSupportOn SMSubscribe
+      pure aliceId
+    threadDelay 500000
+    noMessages bob "joining should be retried while the address is full"
+    withAgent 1 agentCfg initAgentServers testDB $ \alice -> runRight_ $ do
+      subscribeConnection alice contactId
+      ("", _, A.REQ _ _ _ "carol's connInfo" _ _) <- get alice
+      get bob =##> \case ("2", c, JOINED False) -> c == aliceId; _ -> False
+      ("", _, A.REQ _ _ _ "bob's connInfo" _ _) <- get alice
+      pure ()
+  where
+    cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {msgQueueQuota = 1, maxJournalMsgCount = 2}
+
+testJoinFullContactAsyncExpire :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
+testJoinFullContactAsyncExpire (t, msType) = withSmpServerConfigOn t cfg' testPort $ \_ -> do
+  (_, qInfo) <- fillContactAddress
+  withAgent 2 agentCfg {quotaExceededTimeout = 2, commandQuotaRetryInterval = fastRetryInterval} initAgentServers testDB2 $ \bob -> do
+    aliceId <- runRight $ do
+      (aliceId, _) <- A.prepareConnectionToJoin bob 1 True qInfo PQSupportOn
+      A.joinConnectionAsync bob "2" False aliceId True qInfo "bob's connInfo" PQSupportOn SMSubscribe
+      pure aliceId
+    threadDelay 500000
+    noMessages bob "joining should be retried until quota exceeded timeout"
+    get bob =##> \case ("2", c, ERR (SMP _ QUOTA)) -> c == aliceId; _ -> False
+  where
+    cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {msgQueueQuota = 1, maxJournalMsgCount = 2}
+
+fillContactAddress :: HasCallStack => IO (ConnId, ConnectionRequestUri 'CMContact)
+fillContactAddress = do
+  (contactId, qInfo) <- withAgent 1 agentCfg initAgentServers testDB $ \alice -> runRight $ do
+    (contactId, CCLink qInfo Nothing) <- A.createConnection alice NRMInteractive 1 True True SCMContact Nothing Nothing IKPQOn False SMSubscribe
+    pure (contactId, qInfo)
+  withAgent 3 agentCfg initAgentServers testDB3 $ \carol -> runRight_ $ do
+    (aliceId, _) <- A.prepareConnectionToJoin carol 1 True qInfo PQSupportOn
+    void $ A.joinConnection carol NRMInteractive 1 aliceId True qInfo "carol's connInfo" PQSupportOn SMSubscribe
+  pure (contactId, qInfo)
+
 testDropMsgAfterRcvAttempts :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testDropMsgAfterRcvAttempts ps =
   withSmpServerStoreLogOn ps testPort $ \_ -> do

tests/CoreTests/RetryIntervalTests.hs

@@ -14,8 +14,11 @@ import Util
 retryIntervalTests :: Spec
 retryIntervalTests = do
   describe "Retry interval with 2 modes and lock" $ do
-    testRetryIntervalSameMode
-    testRetryIntervalSwitchMode
+    testRetryIntervalSameMode withRetryNewLock2
+    testRetryIntervalSwitchMode withRetryNewLock2
+  describe "Retry interval with 2 modes" $ do
+    testRetryIntervalSameMode withRetryInterval2
+    testRetryIntervalSwitchMode withRetryInterval2
   describe "Foreground retry interval" $ do
     testRetryForeground
     testRetryToBackground
@@ -41,14 +44,18 @@ testFastRI =
       maxInterval = 40000
     }
 
-testRetryIntervalSameMode :: Spec
-testRetryIntervalSameMode =
+type WithRetry2 = RetryInterval2 -> (RI2State -> (RetryIntervalMode -> IO ()) -> IO ()) -> IO ()
+
+withRetryNewLock2 :: WithRetry2
+withRetryNewLock2 ri action = newEmptyTMVarIO >>= \lock -> withRetryLock2 ri lock action
+
+testRetryIntervalSameMode :: WithRetry2 -> Spec
+testRetryIntervalSameMode withRetry =
   it "should increase elapased time and interval when the mode stays the same" $ do
-    lock <- newEmptyTMVarIO
     intervals <- newTVarIO []
     reportedIntervals <- newTVarIO []
     ts <- newTVarIO =<< getCurrentTime
-    withRetryLock2 testRI lock $ \(RI2State slow fast) loop -> do
+    withRetry testRI $ \(RI2State slow fast) loop -> do
       ints <- addInterval intervals ts
       atomically $ modifyTVar' reportedIntervals ((slow, fast) :)
       when (length ints < 9) $ loop RIFast
@@ -65,14 +72,13 @@ testRetryIntervalSameMode =
                        (20000, 40000)
                      ]
 
-testRetryIntervalSwitchMode :: Spec
-testRetryIntervalSwitchMode =
+testRetryIntervalSwitchMode :: WithRetry2 -> Spec
+testRetryIntervalSwitchMode withRetry =
   it "should increase elapased time and interval when the mode switches" $ do
-    lock <- newEmptyTMVarIO
     intervals <- newTVarIO []
     reportedIntervals <- newTVarIO []
     ts <- newTVarIO =<< getCurrentTime
-    withRetryLock2 testRI lock $ \(RI2State slow fast) loop -> do
+    withRetry testRI $ \(RI2State slow fast) loop -> do
       ints <- addInterval intervals ts
       atomically $ modifyTVar' reportedIntervals ((slow, fast) :)
       when (length ints < 11) $ loop $ if length ints <= 5 then RIFast else RISlow