diff --git a/src/Simplex/FileTransfer/Agent.hs b/src/Simplex/FileTransfer/Agent.hs index 8ca537e53..c82316c82 100644 --- a/src/Simplex/FileTransfer/Agent.hs +++ b/src/Simplex/FileTransfer/Agent.hs @@ -194,6 +194,7 @@ runXFTPRcvWorker c srv Worker {doWork} = do (fc@RcvFileChunk {userId, rcvFileId, rcvFileEntityId, digest, fileTmpPath, replicas = replica@RcvFileChunkReplica {rcvChunkReplicaId, server, delay} : _}, approvedRelays) -> do let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay withRetryIntervalLimit xftpConsecutiveRetries ri' $ \delay' loop -> do + atomically $ waitWhileSuspended c liftIO $ waitForUserNetwork c atomically $ incXFTPServerStat c userId srv downloadAttempts downloadFileChunk fc replica approvedRelays @@ -463,6 +464,7 @@ runXFTPSndPrepareWorker c Worker {doWork} = do let AgentClient {xftpServers} = c userSrvCount <- length <$> atomically (TM.lookup userId xftpServers) withRetryIntervalCount (riFast ri) $ \n _ loop -> do + atomically $ waitWhileSuspended c liftIO $ waitForUserNetwork c let triedAllSrvs = n > userSrvCount createWithNextSrv usedSrvs @@ -502,6 +504,7 @@ runXFTPSndWorker c srv Worker {doWork} = do fc@SndFileChunk {userId, sndFileId, sndFileEntityId, filePrefixPath, digest, replicas = replica@SndFileChunkReplica {sndChunkReplicaId, server, delay} : _} -> do let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay withRetryIntervalLimit xftpConsecutiveRetries ri' $ \delay' loop -> do + atomically $ waitWhileSuspended c liftIO $ waitForUserNetwork c atomically $ incXFTPServerStat c userId srv uploadAttempts uploadFileChunk cfg fc replica @@ -674,6 +677,7 @@ runXFTPDelWorker c srv Worker {doWork} = do processDeletedReplica replica@DeletedSndChunkReplica {deletedSndChunkReplicaId, userId, server, chunkDigest, delay} = do let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay withRetryIntervalLimit xftpConsecutiveRetries ri' $ \delay' loop -> do + atomically $ waitWhileSuspended c liftIO $ waitForUserNetwork c atomically $ incXFTPServerStat c userId srv deleteAttempts deleteChunkReplica diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index 60c98b4ea..c8ffe7a50 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -1252,7 +1252,9 @@ runCommandProcessing c@AgentClient {subQ} server_ Worker {doWork} = do withStore c (`getConn` connId) >>= \case SomeConn _ conn@DuplexConnection {} -> a conn _ -> internalErr "command requires duplex connection" - tryCommand action = withRetryInterval ri $ \_ loop -> + tryCommand action = withRetryInterval ri $ \_ loop -> do + atomically $ waitWhileSuspended c + liftIO $ waitForUserNetwork c tryError action >>= \case Left e | temporaryOrHostError e -> retrySndOp c loop @@ -1369,6 +1371,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} ConnData {connId} sq@SndQueue {userI let mId = unId msgId ri' = maybe id updateRetryInterval2 msgRetryState ri withRetryLock2 ri' qLock $ \riState loop -> do + atomically $ waitWhileSuspended c liftIO $ waitForUserNetwork c resp <- tryError $ case msgType of AM_CONN_INFO -> sendConfirmation c sq msgBody diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs index 55caf754c..d3e1f2eca 100644 --- a/src/Simplex/Messaging/Agent/Client.hs +++ b/src/Simplex/Messaging/Agent/Client.hs @@ -127,6 +127,7 @@ module Simplex.Messaging.Agent.Client beginAgentOperation, endAgentOperation, waitUntilForeground, + waitWhileSuspended, suspendSendingAndDatabase, suspendOperation, notifySuspended, @@ -716,12 +717,14 @@ resubscribeSMPSession c@AgentClient {smpSubWorkers, workerSeq} tSess = do atomically $ putTMVar (sessionVar v) a runSubWorker = do ri <- asks $ reconnectInterval . config - withRetryInterval ri $ \_ loop -> do + withRetryForeground ri isForeground (isNetworkOnline c) $ \_ loop -> do pending <- atomically getPending forM_ (L.nonEmpty pending) $ \qs -> do + atomically $ waitUntilForeground c liftIO $ waitForUserNetwork c reconnectSMPClient c tSess qs loop + isForeground = (ASForeground ==) <$> readTVar (agentState c) getPending = RQ.getSessQueues tSess $ pendingSubs c cleanup :: SessionVar (Async ()) -> STM () cleanup v = do @@ -1873,6 +1876,12 @@ waitUntilForeground :: AgentClient -> STM () waitUntilForeground c = unlessM ((ASForeground ==) <$> readTVar (agentState c)) retry {-# INLINE waitUntilForeground #-} +-- This function waits while agent is suspended, but will proceed while it is suspending, +-- to allow completing in-flight operations. +waitWhileSuspended :: AgentClient -> STM () +waitWhileSuspended c = unlessM ((ASSuspended /=) <$> readTVar (agentState c)) retry +{-# INLINE waitWhileSuspended #-} + withStore' :: AgentClient -> (DB.Connection -> IO a) -> AM a withStore' c action = withStore c $ fmap Right . action {-# INLINE withStore' #-} diff --git a/src/Simplex/Messaging/Agent/Env/SQLite.hs b/src/Simplex/Messaging/Agent/Env/SQLite.hs index 6932764c3..86203266b 100644 --- a/src/Simplex/Messaging/Agent/Env/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Env/SQLite.hs @@ -162,7 +162,7 @@ defaultReconnectInterval = RetryInterval { initialInterval = 2_000000, increaseAfter = 10_000000, - maxInterval = 60_000000 + maxInterval = 180_000000 } defaultMessageRetryInterval :: RetryInterval2 @@ -172,7 +172,7 @@ defaultMessageRetryInterval = RetryInterval { initialInterval = 2_000000, increaseAfter = 10_000000, - maxInterval = 60_000000 + maxInterval = 120_000000 }, riSlow = RetryInterval diff --git a/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs b/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs index 689e5fcce..249aad942 100644 --- a/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs +++ b/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs @@ -159,6 +159,7 @@ runNtfWorker c srv Worker {doWork} = logInfo $ "runNtfWorker, nextSub " <> tshow nextSub ri <- asks $ reconnectInterval . config withRetryInterval ri $ \_ loop -> do + atomically $ waitWhileSuspended c liftIO $ waitForUserNetwork c processSub nextSub `catchAgentError` retryOnError c "NtfWorker" loop (workerInternalError c connId . show) @@ -243,6 +244,7 @@ runNtfSMPWorker c srv Worker {doWork} = do logInfo $ "runNtfSMPWorker, nextSub " <> tshow nextSub ri <- asks $ reconnectInterval . config withRetryInterval ri $ \_ loop -> do + atomically $ waitWhileSuspended c liftIO $ waitForUserNetwork c processSub nextSub `catchAgentError` retryOnError c "NtfSMPWorker" loop (workerInternalError c connId . show) diff --git a/src/Simplex/Messaging/Agent/RetryInterval.hs b/src/Simplex/Messaging/Agent/RetryInterval.hs index 00fe4039e..35fa7c5c6 100644 --- a/src/Simplex/Messaging/Agent/RetryInterval.hs +++ b/src/Simplex/Messaging/Agent/RetryInterval.hs @@ -9,6 +9,7 @@ module Simplex.Messaging.Agent.RetryInterval RI2State (..), withRetryInterval, withRetryIntervalCount, + withRetryForeground, withRetryLock2, updateRetryInterval2, nextRetryDelay, @@ -16,10 +17,11 @@ module Simplex.Messaging.Agent.RetryInterval where import Control.Concurrent (forkIO) +import Control.Concurrent.STM (retry) import Control.Monad (void) import Control.Monad.IO.Class (MonadIO, liftIO) import Data.Int (Int64) -import Simplex.Messaging.Util (threadDelay', whenM) +import Simplex.Messaging.Util (threadDelay', unlessM, whenM) import UnliftIO.STM data RetryInterval = RetryInterval @@ -63,6 +65,27 @@ withRetryIntervalCount ri action = callAction 0 0 $ initialInterval ri let elapsed' = elapsed + delay callAction (n + 1) elapsed' $ nextRetryDelay elapsed' delay ri +withRetryForeground :: forall m a. MonadIO m => RetryInterval -> STM Bool -> STM Bool -> (Int64 -> m a -> m a) -> m a +withRetryForeground ri isForeground isOnline action = callAction 0 $ initialInterval ri + where + callAction :: Int64 -> Int64 -> m a + callAction elapsed delay = action delay loop + where + loop = do + -- limit delay to max Int value (~36 minutes on for 32 bit architectures) + d <- registerDelay $ fromIntegral $ min delay (fromIntegral (maxBound :: Int)) + (wasForeground, wasOnline) <- atomically $ (,) <$> isForeground <*> isOnline + reset <- atomically $ do + foreground <- isForeground + online <- isOnline + let reset = (not wasForeground && foreground) || (not wasOnline && online) + unlessM ((reset ||) <$> readTVar d) retry + pure reset + let (elapsed', delay') + | reset = (0, initialInterval ri) + | otherwise = (elapsed + delay, nextRetryDelay elapsed' delay ri) + callAction elapsed' delay' + -- 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 = diff --git a/tests/AgentTests/FunctionalAPITests.hs b/tests/AgentTests/FunctionalAPITests.hs index 0ab162d08..4d61d8463 100644 --- a/tests/AgentTests/FunctionalAPITests.hs +++ b/tests/AgentTests/FunctionalAPITests.hs @@ -1813,7 +1813,6 @@ testSuspendingAgentCompleteSending t = withAgentClients2 $ \a b -> do get b =##> \case ("", c, Msg "hello") -> c == aId; _ -> False ackMessage b aId 2 Nothing pure (aId, bId) - runRight_ $ do ("", "", DOWN {}) <- nGet a ("", "", DOWN {}) <- nGet b @@ -1821,15 +1820,17 @@ testSuspendingAgentCompleteSending t = withAgentClients2 $ \a b -> do 4 <- sendMessage b aId SMP.noMsgFlags "how are you?" liftIO $ threadDelay 100000 liftIO $ suspendAgent b 5000000 - withSmpServerStoreLogOn t testPort $ \_ -> runRight_ @AgentErrorType $ do - pGet b =##> \case ("", c, AEvt SAEConn (SENT 3)) -> c == aId; ("", "", AEvt _ UP {}) -> True; _ -> False - pGet b =##> \case ("", c, AEvt SAEConn (SENT 3)) -> c == aId; ("", "", AEvt _ UP {}) -> True; _ -> False - pGet b =##> \case ("", c, AEvt SAEConn (SENT 4)) -> c == aId; ("", "", AEvt _ UP {}) -> True; _ -> False - ("", "", SUSPENDED) <- nGet b - - pGet a =##> \case ("", c, AEvt _ (Msg "hello too")) -> c == bId; ("", "", AEvt _ UP {}) -> True; _ -> False - pGet a =##> \case ("", c, AEvt _ (Msg "hello too")) -> c == bId; ("", "", AEvt _ UP {}) -> True; _ -> False + -- there will be no UP event for b, because re-subscriptions are suspended until the agent is in foreground + get b =##> \case ("", c, SENT 3) -> c == aId; _ -> False + get b =##> \case ("", c, SENT 4) -> c == aId; _ -> False + nGet b ##> ("", "", SUSPENDED) + liftIO $ + getInAnyOrder + a + [ \case ("", c, AEvt _ (Msg "hello too")) -> c == bId; _ -> False, + \case ("", "", AEvt _ UP {}) -> True; _ -> False + ] ackMessage a bId 3 Nothing get a =##> \case ("", c, Msg "how are you?") -> c == bId; _ -> False ackMessage a bId 4 Nothing diff --git a/tests/CoreTests/RetryIntervalTests.hs b/tests/CoreTests/RetryIntervalTests.hs index 7097df989..da96d0208 100644 --- a/tests/CoreTests/RetryIntervalTests.hs +++ b/tests/CoreTests/RetryIntervalTests.hs @@ -2,6 +2,8 @@ module CoreTests.RetryIntervalTests where +import Control.Concurrent (threadDelay) +import Control.Concurrent.Async (concurrently_) import Control.Concurrent.STM import Control.Monad (when) import Data.Time.Clock (UTCTime, diffUTCTime, getCurrentTime, nominalDiffTimeToSeconds) @@ -13,6 +15,10 @@ retryIntervalTests = do describe "Retry interval with 2 modes and lock" $ do testRetryIntervalSameMode testRetryIntervalSwitchMode + describe "Foreground retry interval" $ do + testRetryForeground + testRetryToBackground + testRetrySkipWhenForeground testRI :: RetryInterval2 testRI = @@ -23,12 +29,15 @@ testRI = increaseAfter = 40000, maxInterval = 40000 }, - riFast = - RetryInterval - { initialInterval = 10000, - increaseAfter = 20000, - maxInterval = 40000 - } + riFast = testFastRI + } + +testFastRI :: RetryInterval +testFastRI = + RetryInterval + { initialInterval = 10000, + increaseAfter = 20000, + maxInterval = 40000 } testRetryIntervalSameMode :: Spec @@ -81,6 +90,67 @@ testRetryIntervalSwitchMode = (40000, 40000) ] +testRetryForeground :: Spec +testRetryForeground = + it "should increase elapased time and interval" $ do + intervals <- newTVarIO [] + reportedIntervals <- newTVarIO [] + ts <- newTVarIO =<< getCurrentTime + let isForeground = pure True + withRetryForeground testFastRI isForeground (pure True) $ \delay loop -> do + ints <- addInterval intervals ts + atomically $ modifyTVar' reportedIntervals (delay :) + when (length ints < 8) $ loop + (reverse <$> readTVarIO intervals) `shouldReturn` [0, 1, 1, 1, 2, 3, 4, 4] + (reverse <$> readTVarIO reportedIntervals) + `shouldReturn` [ 10000, 10000, 15000, 22500, 33750, 40000, 40000, 40000] + +testRetryToBackground :: Spec +testRetryToBackground = + it "should not change interval when moving to background" $ do + intervals <- newTVarIO [] + reportedIntervals <- newTVarIO [] + ts <- newTVarIO =<< getCurrentTime + foreground <- newTVarIO True + concurrently_ + ( do + threadDelay 50000 + atomically $ writeTVar foreground False + ) + ( withRetryForeground testFastRI (readTVar foreground) (pure True) $ \delay loop -> do + ints <- addInterval intervals ts + atomically $ modifyTVar' reportedIntervals (delay :) + when (length ints < 8) $ loop + ) + (reverse <$> readTVarIO intervals) `shouldReturn` [0, 1, 1, 1, 2, 3, 4, 4] + (reverse <$> readTVarIO reportedIntervals) + `shouldReturn` [ 10000, 10000, 15000, 22500, 33750, 40000, 40000, 40000] + +testRetrySkipWhenForeground :: Spec +testRetrySkipWhenForeground = + it "should repeat loop as soon as moving to foreground" $ do + intervals <- newTVarIO [] + reportedIntervals <- newTVarIO [] + ts <- newTVarIO =<< getCurrentTime + foreground <- newTVarIO False + concurrently_ + ( do + threadDelay 65000 + atomically $ writeTVar foreground True + threadDelay 10000 + atomically $ writeTVar foreground False + threadDelay 100000 + atomically $ writeTVar foreground True + ) + ( withRetryForeground testFastRI (readTVar foreground) (pure True) $ \delay loop -> do + ints <- addInterval intervals ts + atomically $ modifyTVar' reportedIntervals (delay :) + when (length ints < 12) $ loop + ) + (reverse <$> readTVarIO intervals) `shouldReturn` [0, 1, 1, 1, 2, 0, 1, 1, 1, 2, 3, 1] + (reverse <$> readTVarIO reportedIntervals) + `shouldReturn` [ 10000, 10000, 15000, 22500, 33750, 10000, 10000, 15000, 22500, 33750, 40000, 10000] + addInterval :: TVar [Int] -> TVar UTCTime -> IO [Int] addInterval intervals ts = do ts' <- getCurrentTime