diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index 80adf05cb..79e411bc5 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/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 diff --git a/src/Simplex/Messaging/Agent/Env/SQLite.hs b/src/Simplex/Messaging/Agent/Env/SQLite.hs index bb066f35b..fc7af667f 100644 --- a/src/Simplex/Messaging/Agent/Env/SQLite.hs +++ b/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 diff --git a/src/Simplex/Messaging/Agent/RetryInterval.hs b/src/Simplex/Messaging/Agent/RetryInterval.hs index 35fa7c5c6..4ddc220da 100644 --- a/src/Simplex/Messaging/Agent/RetryInterval.hs +++ b/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} = diff --git a/src/Simplex/Messaging/Agent/Store.hs b/src/Simplex/Messaging/Agent/Store.hs index 607e6d1da..e75cdb05e 100644 --- a/src/Simplex/Messaging/Agent/Store.hs +++ b/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 diff --git a/src/Simplex/Messaging/Agent/Store/AgentStore.hs b/src/Simplex/Messaging/Agent/Store/AgentStore.hs index 3d82b5ac2..a55d10815 100644 --- a/src/Simplex/Messaging/Agent/Store/AgentStore.hs +++ b/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 ()) diff --git a/tests/AgentTests/FunctionalAPITests.hs b/tests/AgentTests/FunctionalAPITests.hs index af02a700c..319f885ea 100644 --- a/tests/AgentTests/FunctionalAPITests.hs +++ b/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 diff --git a/tests/CoreTests/RetryIntervalTests.hs b/tests/CoreTests/RetryIntervalTests.hs index 36a50f30d..d1ab97124 100644 --- a/tests/CoreTests/RetryIntervalTests.hs +++ b/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