From 925545c7bb15edc7dbfd6aebe571afe6b78bec70 Mon Sep 17 00:00:00 2001 From: Narasimha-sc <166327228+Narasimha-sc@users.noreply.github.com> Date: Sun, 4 Oct 2026 15:02:27 +0000 Subject: [PATCH] agent: restart the retry loops when the network changes Closing the clients when the network changes makes the workers fail their current attempt, but a worker already waiting out its retry interval resumes only when it expires - up to the maximum interval, so delivery and subscriptions can take minutes after the network is usable again. The network epoch is incremented when the clients are closed, and the agent retry loops restart from the loop's initial interval when it changes - for delivery and XFTP workers resumed with persisted retry state that is the persisted interval (withRetryEpoch). Message delivery restarts the fast interval only, the slow interval for QUOTA errors is unchanged; the subscription worker keeps its foreground and online resets; retry attempt counters are not reset. A loop that was waiting for the network sees the epoch change after its first attempt, so if that attempt fails it retries once more without delay. withRetryWait2 now takes the value to sample before the action and a wait that receives the retry interval mode and returns whether the interval restarts, so one loop serves the plain, the lock and the epoch variants - the slow interval retries in STM and falls through to the plain wait, so it is unchanged. withRetryEpoch2 is withRetryLock2 without the lock, for the command worker, which retries with two intervals. --- src/Simplex/FileTransfer/Agent.hs | 14 +- src/Simplex/Messaging/Agent.hs | 6 +- src/Simplex/Messaging/Agent/Client.hs | 6 +- .../Messaging/Agent/NtfSubSupervisor.hs | 4 +- src/Simplex/Messaging/Agent/RetryInterval.hs | 93 +++++++++--- tests/AgentTests/FunctionalAPITests.hs | 22 +++ tests/CoreTests/RetryIntervalTests.hs | 136 +++++++++++++++++- 7 files changed, 244 insertions(+), 37 deletions(-) diff --git a/src/Simplex/FileTransfer/Agent.hs b/src/Simplex/FileTransfer/Agent.hs index aa6bbabf7..6d07ed6c3 100644 --- a/src/Simplex/FileTransfer/Agent.hs +++ b/src/Simplex/FileTransfer/Agent.hs @@ -207,7 +207,7 @@ runXFTPRcvWorker c srv Worker {doWork} = do rcvWorkerInternalError c rcvFileId rcvFileEntityId redirectEntityId_ (Just fileTmpPath) (INTERNAL "chunk has no replicas") (fc@RcvFileChunk {userId, rcvFileId, rcvFileEntityId, digest, fileTmpPath, replicas = replica@RcvFileChunkReplica {rcvChunkReplicaId, server, delay} : _}, approvedRelays, redirectEntityId_) -> do let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay - withRetryIntervalLimit xftpConsecutiveRetries ri' $ \delay' loop -> do + withRetryIntervalLimit xftpConsecutiveRetries ri' (readTVar $ networkEpoch c) $ \delay' loop -> do liftIO $ waitWhileSuspended c liftIO $ waitForUserNetwork c atomically $ incXFTPServerStat c userId srv downloadAttempts @@ -265,9 +265,9 @@ runXFTPRcvWorker c srv Worker {doWork} = do chunkReceived RcvFileChunk {replicas} = any received replicas -- The first call of action has n == 0, maxN is max number of retries -withRetryIntervalLimit :: forall m. MonadIO m => Int -> RetryInterval -> (Int64 -> m () -> m ()) -> m () -withRetryIntervalLimit maxN ri action = - withRetryIntervalCount ri $ \n delay loop -> +withRetryIntervalLimit :: forall m. MonadIO m => Int -> RetryInterval -> STM Int -> (Int64 -> m () -> m ()) -> m () +withRetryIntervalLimit maxN ri getEpoch action = + withRetryEpochCount ri getEpoch $ \n delay loop -> when (n < maxN) $ action delay loop retryOnError :: Text -> AM a -> AM a -> AgentErrorType -> AM a @@ -490,7 +490,7 @@ runXFTPSndPrepareWorker c Worker {doWork} = do triedHosts <- newTVarIO S.empty let AgentClient {xftpServers} = c userSrvCount <- liftIO $ length <$> TM.lookupIO userId xftpServers - withRetryIntervalCount (riFast ri) $ \n _ loop -> do + withRetryEpochCount (riFast ri) (readTVar $ networkEpoch c) $ \n _ loop -> do liftIO $ waitWhileSuspended c liftIO $ waitForUserNetwork c let triedAllSrvs = n > userSrvCount @@ -530,7 +530,7 @@ runXFTPSndWorker c srv Worker {doWork} = do SndFileChunk {sndFileId, sndFileEntityId, filePrefixPath, replicas = []} -> sndWorkerInternalError c sndFileId sndFileEntityId (Just filePrefixPath) (INTERNAL "chunk has no replicas") 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 + withRetryIntervalLimit xftpConsecutiveRetries ri' (readTVar $ networkEpoch c) $ \delay' loop -> do liftIO $ waitWhileSuspended c liftIO $ waitForUserNetwork c atomically $ incXFTPServerStat c userId srv uploadAttempts @@ -708,7 +708,7 @@ runXFTPDelWorker c srv Worker {doWork} = do where 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 + withRetryIntervalLimit xftpConsecutiveRetries ri' (readTVar $ networkEpoch c) $ \delay' loop -> do liftIO $ waitWhileSuspended c liftIO $ waitForUserNetwork c atomically $ incXFTPServerStat c userId srv deleteAttempts diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index d378566b9..e3c062fbf 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -744,7 +744,7 @@ setUserNetworkInfo c@AgentClient {userNetworkInfo, userNetworkUpdated, networkEv whenCurrent $ closeProtocolServerClients c xftpClients where whenCurrent = whenM (atomically $ isCurrent e) - networkOnline e = atomically $ whenM (isCurrent e) $ writeTVar userNetworkInfo ni + networkOnline e = atomically $ whenM (isCurrent e) $ writeTVar userNetworkInfo ni >> modifyTVar' (networkEpoch c) (+ 1) isCurrent e = (e ==) <$> readTVar networkEventSeq notRecentlyChanged ts' i = maybe True (\ts -> diffUTCTime ts' ts > i) <$> readTVar userNetworkUpdated @@ -2398,7 +2398,7 @@ 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 = withRetryInterval2 ri $ \_ loop -> do + tryMoveableCommand action = withRetryEpoch2 ri (readTVar $ networkEpoch c) $ \_ loop -> do liftIO $ waitWhileSuspended c liftIO $ waitForUserNetwork c tryAllErrors action >>= \case @@ -2576,7 +2576,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} sq@SndQueue {userId, connId, server, atomically $ endAgentOperation c AOMsgDelivery -- this operation begins in submitPendingMsg let mId = unId msgId ri' = maybe id updateRetryInterval2 msgRetryState ri - withRetryLock2 ri' qLock $ \riState loop -> do + withRetryLock2 ri' (readTVar $ networkEpoch c) qLock $ \riState loop -> do liftIO $ waitWhileSuspended c liftIO $ waitForUserNetwork c resp <- tryAllErrors $ case msgType of diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs index 227663df4..8b11ae036 100644 --- a/src/Simplex/Messaging/Agent/Client.hs +++ b/src/Simplex/Messaging/Agent/Client.hs @@ -365,6 +365,8 @@ data AgentClient = AgentClient presetServers :: [SMPServer], userNetworkInfo :: TVar UserNetworkInfo, userNetworkUpdated :: TVar (Maybe UTCTime), + -- incremented when the clients are closed after the network changed, see withRetryEpoch + networkEpoch :: TVar Int, -- incremented for each accepted network info, to identify the most recent one networkEventSeq :: TVar Int, subscrConns :: TVar (Set ConnId), @@ -539,6 +541,7 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, entitlements, netCf useNetworkConfig <- newTVarIO (slowNetworkConfig netCfg, netCfg) userNetworkInfo <- newTVarIO $ UserNetworkInfo UNOther True userNetworkUpdated <- newTVarIO Nothing + networkEpoch <- newTVarIO 0 networkEventSeq <- newTVarIO 0 subscrConns <- newTVarIO S.empty currentSubs <- SS.emptyIO @@ -584,6 +587,7 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, entitlements, netCf presetServers, userNetworkInfo, userNetworkUpdated, + networkEpoch, networkEventSeq, subscrConns, currentSubs, @@ -846,7 +850,7 @@ resubscribeSMPSession c@AgentClient {smpSubWorkers, workerSeq} tSess = do atomically $ putTMVar (sessionVar v) a runSubWorker v = do ri <- asks $ reconnectInterval . config - withRetryForeground ri isForeground (isNetworkOnline c) $ \_ loop -> do + withRetryForeground ri isForeground (isNetworkOnline c) (readTVar $ networkEpoch c) $ \_ loop -> do pending_ <- atomically $ do pending@(pendingSubs, pendingSS) <- SS.getPendingSubs tSess $ currentSubs c if M.null pendingSubs && isNothing pendingSS diff --git a/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs b/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs index 06c9b4ca4..9c3145cbf 100644 --- a/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs +++ b/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs @@ -456,7 +456,7 @@ retrySubActions _ [] _ = pure () retrySubActions c subs action = do v <- newTVarIO subs ri <- asks $ reconnectInterval . config - withRetryInterval ri $ \_ loop -> do + withRetryEpoch ri (readTVar $ networkEpoch c) $ \_ loop -> do liftIO $ waitWhileSuspended c liftIO $ waitForUserNetwork c subs' <- readTVarIO v @@ -563,7 +563,7 @@ runNtfTknDelWorker c srv Worker {doWork} = \nextTknToDelete -> do logInfo $ "runNtfTknDelWorker, nextTknToDelete " <> tshow nextTknToDelete ri <- asks $ reconnectInterval . config - withRetryInterval ri $ \_ loop -> do + withRetryEpoch ri (readTVar $ networkEpoch c) $ \_ loop -> do liftIO $ waitWhileSuspended c liftIO $ waitForUserNetwork c processTknToDelete nextTknToDelete `catchAllErrors` retryTmpError loop nextTknToDelete diff --git a/src/Simplex/Messaging/Agent/RetryInterval.hs b/src/Simplex/Messaging/Agent/RetryInterval.hs index 4ddc220da..f6447fd14 100644 --- a/src/Simplex/Messaging/Agent/RetryInterval.hs +++ b/src/Simplex/Messaging/Agent/RetryInterval.hs @@ -9,6 +9,9 @@ module Simplex.Messaging.Agent.RetryInterval RI2State (..), withRetryInterval, withRetryIntervalCount, + withRetryEpoch, + withRetryEpochCount, + withRetryEpoch2, withRetryForeground, withRetryInterval2, withRetryLock2, @@ -66,20 +69,50 @@ 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 +-- The delay restarts from the initial interval when the epoch changes (the agent increments it when the network changes) +-- while the loop waits; when it changed while the action was running the next attempt is made without any delay. +withRetryEpoch :: forall m a. MonadIO m => RetryInterval -> STM Int -> (Int64 -> m a -> m a) -> m a +withRetryEpoch ri getEpoch = withRetryEpochCount ri getEpoch . const + +withRetryEpochCount :: forall m a. MonadIO m => RetryInterval -> STM Int -> (Int -> Int64 -> m a -> m a) -> m a +withRetryEpochCount ri getEpoch action = callAction 0 0 $ initialInterval ri + where + callAction :: Int -> Int64 -> Int64 -> m a + callAction n elapsed delay = do + epoch <- atomically getEpoch + action n delay $ do + reset <- waitEpoch' epoch delay + let (elapsed', delay') + | reset = (0, initialInterval ri) + | otherwise = (elapsed + delay, nextRetryDelay elapsed' delay ri) + callAction (n + 1) elapsed' delay' + waitEpoch' = waitEpoch getEpoch + +-- returns True when the epoch changed before the delay expired +waitEpoch :: MonadIO m => STM Int -> Int -> Int64 -> m Bool +waitEpoch getEpoch epoch delay = do + -- limit delay to max Int value (~36 minutes on for 32 bit architectures) + d <- registerDelay $ fromIntegral $ min delay (fromIntegral (maxBound :: Int)) + atomically $ do + changed <- (epoch /=) <$> getEpoch + unlessM ((changed ||) <$> readTVar d) retry + pure changed + +withRetryForeground :: forall m a. MonadIO m => RetryInterval -> STM Bool -> STM Bool -> STM Int -> (Int64 -> m a -> m a) -> m a +withRetryForeground ri isForeground isOnline getEpoch action = callAction 0 $ initialInterval ri where callAction :: Int64 -> Int64 -> m a - callAction elapsed delay = action delay loop + callAction elapsed delay = action delay . loop =<< atomically getEpoch where - loop = do + loop epoch = 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) + epochChanged <- (epoch /=) <$> getEpoch + let reset = (not wasForeground && foreground) || (not wasOnline && online) || epochChanged unlessM ((reset ||) <$> readTVar d) retry pure reset let (elapsed', delay') @@ -88,35 +121,55 @@ withRetryForeground ri isForeground isOnline action = callAction 0 $ initialInte callAction elapsed' delay' withRetryInterval2 :: forall m. MonadIO m => RetryInterval2 -> (RI2State -> (RetryIntervalMode -> m ()) -> m ()) -> m () -withRetryInterval2 = withRetryWait2 $ liftIO . threadDelay' +withRetryInterval2 = withRetryWait2 (pure ()) $ \_ _ delay -> False <$ liftIO (threadDelay' 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 ri lock = withRetryWait2 wait ri +-- The fast interval restarts on the epoch change, as in withRetryEpoch; once the action chose the slow +-- interval (recipient queue quota) the epoch change is not applied to the fast interval - neither the one +-- that happened during that action nor during the wait. +withRetryLock2 :: forall m. MonadIO m => RetryInterval2 -> STM Int -> TMVar () -> (RI2State -> (RetryIntervalMode -> m ()) -> m ()) -> m () +withRetryLock2 ri getEpoch lock = withRetryWait2 (atomically getEpoch) wait ri where - wait delay = do + wait mode epoch delay = do waiting <- newTVarIO True _ <- liftIO . forkIO $ do threadDelay' delay atomically $ whenM (readTVar waiting) $ void $ tryPutTMVar lock () atomically $ do - takeTMVar lock + reset <- (epochChanged <* tryTakeTMVar lock) `orElse` (False <$ takeTMVar lock) writeTVar waiting False + pure reset + where + epochChanged = case mode of + RIFast -> unlessM ((epoch /=) <$> getEpoch) retry >> pure True + RISlow -> retry -withRetryWait2 :: forall m. Monad m => (Int64 -> m ()) -> RetryInterval2 -> (RI2State -> (RetryIntervalMode -> m ()) -> m ()) -> m () -withRetryWait2 wait RetryInterval2 {riSlow, riFast} action = +-- As withRetryLock2, without the lock: the fast interval restarts on the epoch change, the slow one +-- (recipient queue quota) does not. +withRetryEpoch2 :: forall m. MonadIO m => RetryInterval2 -> STM Int -> (RI2State -> (RetryIntervalMode -> m ()) -> m ()) -> m () +withRetryEpoch2 ri getEpoch = withRetryWait2 (atomically getEpoch) wait ri + where + wait mode epoch delay = case mode of + RISlow -> False <$ liftIO (threadDelay' delay) + RIFast -> waitEpoch getEpoch epoch delay + +-- the value sampled before the action is passed to wait, so that a change that happened during the +-- action resets the interval as one during the wait does +withRetryWait2 :: forall m e. Monad m => m e -> (RetryIntervalMode -> e -> Int64 -> m Bool) -> RetryInterval2 -> (RI2State -> (RetryIntervalMode -> m ()) -> m ()) -> m () +withRetryWait2 sample wait RetryInterval2 {riSlow, riFast} action = callAction (0, initialInterval riSlow) (0, initialInterval riFast) where callAction :: (Int64, Int64) -> (Int64, Int64) -> m () - callAction slow fast = action (RI2State (snd slow) (snd fast)) loop + callAction slow fast = sample >>= \e -> action (RI2State (snd slow) (snd fast)) (loop e) where - loop = \case - RISlow -> run slow riSlow (`callAction` fast) - RIFast -> run fast riFast (callAction slow) - run (elapsed, delay) ri call = do - wait delay - let elapsed' = elapsed + delay - delay' = nextRetryDelay elapsed' delay ri + loop e = \case + RISlow -> run RISlow e slow riSlow (`callAction` fast) + RIFast -> run RIFast e fast riFast (callAction slow) + run mode e (elapsed, delay) ri call = do + reset <- wait mode e delay + let (elapsed', delay') + | reset = (0, initialInterval ri) + | otherwise = (elapsed + delay, nextRetryDelay elapsed' delay ri) call (elapsed', delay') nextRetryDelay :: Int64 -> Int64 -> RetryInterval -> Int64 diff --git a/tests/AgentTests/FunctionalAPITests.hs b/tests/AgentTests/FunctionalAPITests.hs index 0cd91c446..88769674c 100644 --- a/tests/AgentTests/FunctionalAPITests.hs +++ b/tests/AgentTests/FunctionalAPITests.hs @@ -608,6 +608,7 @@ functionalAPITests ps = do it "should wait for user network" testWaitForUserNetwork it "should not reset online to offline if happens too quickly" testDoNotResetOnlineToOffline it "should reconnect to servers when network changes" $ testNetworkChangeReconnect ps + it "should retry delivery and subscription immediately when network changes" $ testNetworkChangeRetry ps it "should resume multiple threads" testResumeMultipleThreads describe "SMP queue info" $ do it "server should respond with queue and subscription information" $ @@ -4817,6 +4818,27 @@ testNetworkChangeReconnect ps = nGet a =##> \case ("", "", UP _ [c]) -> c == bId; _ -> False runRight_ $ exchangeGreetings a bId b aId +testNetworkChangeRetry :: HasCallStack => (ASrvTransport, AStoreType) -> IO () +testNetworkChangeRetry ps = + withAgentClientsCfg2 aCfg agentCfg $ \a b -> do + (aId, bId) <- withSmpServerStoreLogOn ps testPort $ \_ -> runRight $ makeConnection a b + nGet a =##> \case ("", "", DOWN _ [c]) -> c == bId; _ -> False + nGet b =##> \case ("", "", DOWN _ [c]) -> c == aId; _ -> False + -- the first delivery attempt fails, and the delivery worker waits for the configured 60 seconds + 2 <- runRight $ sendMessage a bId SMP.noMsgFlags "hello" + threadDelay 500000 + withSmpServerStoreLogOn ps testPort $ \_ -> do + -- without network change event both the message delivery and the subscription + -- would be retried in 60 seconds, the test would fail with timeout + setUserNetworkInfo a $ UserNetworkInfo UNCellular True + runRight_ $ do + withUP a bId $ \case ("", c, SENT 2) -> c == bId; _ -> False + withUP b aId $ \case ("", c, Msg "hello") -> c == aId; _ -> False + ackMessage b aId 2 Nothing + where + slowRI = RetryInterval {initialInterval = 60_000000, increaseAfter = 0, maxInterval = 60_000000} + aCfg = agentCfg {reconnectInterval = slowRI, messageRetryInterval = RetryInterval2 {riSlow = slowRI, riFast = slowRI}} + testDoNotResetOnlineToOffline :: IO () testDoNotResetOnlineToOffline = withAgent 1 aCfg initAgentServers testDB $ \a -> do noNetworkDelay a diff --git a/tests/CoreTests/RetryIntervalTests.hs b/tests/CoreTests/RetryIntervalTests.hs index d1ab97124..780866d7d 100644 --- a/tests/CoreTests/RetryIntervalTests.hs +++ b/tests/CoreTests/RetryIntervalTests.hs @@ -6,6 +6,7 @@ import Control.Concurrent (threadDelay) import Control.Concurrent.Async (concurrently_) import Control.Concurrent.STM import Control.Monad (when) +import Data.Int (Int64) import Data.Time.Clock (UTCTime, diffUTCTime, getCurrentTime, nominalDiffTimeToSeconds) import Simplex.Messaging.Agent.RetryInterval import Test.Hspec hiding (fit, it) @@ -16,13 +17,23 @@ retryIntervalTests = do describe "Retry interval with 2 modes and lock" $ do testRetryIntervalSameMode withRetryNewLock2 testRetryIntervalSwitchMode withRetryNewLock2 + testRetryIntervalEpochReset describe "Retry interval with 2 modes" $ do testRetryIntervalSameMode withRetryInterval2 testRetryIntervalSwitchMode withRetryInterval2 + describe "Retry interval with 2 modes and epoch" $ do + testRetryIntervalSameMode (`withRetryEpoch2` pure 0) + testRetryIntervalSwitchMode (`withRetryEpoch2` pure 0) describe "Foreground retry interval" $ do testRetryForeground testRetryToBackground testRetrySkipWhenForeground + testRetryForegroundEpoch + describe "Retry interval with epoch" $ do + testRetryEpochSameValue + testRetrySkipOnEpochChange + testRetryEpochChangedInAction + testRetryEpochKeepsCount testRI :: RetryInterval2 testRI = @@ -47,7 +58,7 @@ testFastRI = type WithRetry2 = RetryInterval2 -> (RI2State -> (RetryIntervalMode -> IO ()) -> IO ()) -> IO () withRetryNewLock2 :: WithRetry2 -withRetryNewLock2 ri action = newEmptyTMVarIO >>= \lock -> withRetryLock2 ri lock action +withRetryNewLock2 ri action = newEmptyTMVarIO >>= \lock -> withRetryLock2 ri (pure 0) lock action testRetryIntervalSameMode :: WithRetry2 -> Spec testRetryIntervalSameMode withRetry = @@ -97,6 +108,41 @@ testRetryIntervalSwitchMode withRetry = (40000, 40000) ] +testRetryIntervalEpochReset :: Spec +testRetryIntervalEpochReset = + it "should restart fast interval when epoch changes, but not slow interval" $ do + lock <- newEmptyTMVarIO + epoch <- newTVarIO (0 :: Int) + intervals <- newTVarIO [] + reportedIntervals <- newTVarIO [] + ts <- newTVarIO =<< getCurrentTime + concurrently_ + ( do + -- fast mode: waiting for 33750, restarts immediately at 65ms with the initial interval + threadDelay 65000 + atomically $ modifyTVar' epoch (+ 1) + -- slow mode: waiting for 20000 from 95ms, the change is ignored + threadDelay 40000 + atomically $ modifyTVar' epoch (+ 1) + ) + ( withRetryLock2 testRI (readTVar epoch) lock $ \(RI2State slow fast) loop -> do + ints <- addInterval intervals ts + atomically $ modifyTVar' reportedIntervals ((slow, fast) :) + when (length ints < 9) $ loop $ if length ints <= 6 then RIFast else RISlow + ) + (reverse <$> readTVarIO intervals) `shouldReturn` [0, 1, 1, 1, 2, 0, 1, 2, 2] + (reverse <$> readTVarIO reportedIntervals) + `shouldReturn` [ (20000, 10000), + (20000, 10000), + (20000, 15000), + (20000, 22500), + (20000, 33750), + (20000, 10000), + (20000, 10000), + (20000, 10000), + (30000, 10000) + ] + testRetryForeground :: Spec testRetryForeground = it "should increase elapased time and interval" $ do @@ -104,7 +150,7 @@ testRetryForeground = reportedIntervals <- newTVarIO [] ts <- newTVarIO =<< getCurrentTime let isForeground = pure True - withRetryForeground testFastRI isForeground (pure True) $ \delay loop -> do + withRetryForeground testFastRI isForeground (pure True) (pure 0) $ \delay loop -> do ints <- addInterval intervals ts atomically $ modifyTVar' reportedIntervals (delay :) when (length ints < 8) $ loop @@ -124,7 +170,7 @@ testRetryToBackground = threadDelay 50000 atomically $ writeTVar foreground False ) - ( withRetryForeground testFastRI (readTVar foreground) (pure True) $ \delay loop -> do + ( withRetryForeground testFastRI (readTVar foreground) (pure True) (pure 0) $ \delay loop -> do ints <- addInterval intervals ts atomically $ modifyTVar' reportedIntervals (delay :) when (length ints < 8) $ loop @@ -149,7 +195,89 @@ testRetrySkipWhenForeground = threadDelay 100000 atomically $ writeTVar foreground True ) - ( withRetryForeground testFastRI (readTVar foreground) (pure True) $ \delay loop -> do + ( withRetryForeground testFastRI (readTVar foreground) (pure True) (pure 0) $ \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] + +testRetryForegroundEpoch :: Spec +testRetryForegroundEpoch = + testEpochRestartsLoop "should repeat foreground loop as soon as epoch changes" $ \e -> + withRetryForeground testFastRI (pure True) (pure True) (readTVar e) + +testRetryEpochSameValue :: Spec +testRetryEpochSameValue = + it "should not change interval when epoch is written with the same value" $ do + intervals <- newTVarIO [] + reportedIntervals <- newTVarIO [] + ts <- newTVarIO =<< getCurrentTime + epoch <- newTVarIO (0 :: Int) + concurrently_ + ( do + threadDelay 50000 + atomically $ writeTVar epoch 0 + ) + ( withRetryEpoch testFastRI (readTVar epoch) $ \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] + +testRetryEpochChangedInAction :: Spec +testRetryEpochChangedInAction = + it "should restart interval when epoch changes while action runs" $ do + intervals <- newTVarIO [] + reportedIntervals <- newTVarIO [] + ts <- newTVarIO =<< getCurrentTime + epoch <- newTVarIO (0 :: Int) + withRetryEpoch testFastRI (readTVar epoch) $ \delay loop -> do + ints <- addInterval intervals ts + atomically $ modifyTVar' reportedIntervals (delay :) + when (length ints == 3) $ atomically $ modifyTVar' epoch (+ 1) + when (length ints < 6) $ loop + (reverse <$> readTVarIO intervals) `shouldReturn` [0, 1, 1, 0, 1, 1] + (reverse <$> readTVarIO reportedIntervals) + `shouldReturn` [10000, 10000, 15000, 10000, 10000, 15000] + +testRetryEpochKeepsCount :: Spec +testRetryEpochKeepsCount = + it "should not reset attempt count when epoch changes" $ do + counts <- newTVarIO ([] :: [Int]) + epoch <- newTVarIO (0 :: Int) + withRetryEpochCount testFastRI (readTVar epoch) $ \n _ loop -> do + ns <- atomically $ stateTVar counts $ \ns -> (n : ns, n : ns) + when (length ns == 2) $ atomically $ modifyTVar' epoch (+ 1) + when (length ns < 4) loop + (reverse <$> readTVarIO counts) `shouldReturn` [0, 1, 2, 3] + +testRetrySkipOnEpochChange :: Spec +testRetrySkipOnEpochChange = + testEpochRestartsLoop "should repeat loop as soon as epoch changes" $ \e -> + withRetryEpoch testFastRI (readTVar e) + +-- the epoch changes twice while the loop waits, so the interval restarts from the initial one twice +testEpochRestartsLoop :: String -> (TVar Int -> (Int64 -> IO () -> IO ()) -> IO ()) -> Spec +testEpochRestartsLoop name withRetry = + it name $ do + intervals <- newTVarIO [] + reportedIntervals <- newTVarIO [] + ts <- newTVarIO =<< getCurrentTime + epoch <- newTVarIO (0 :: Int) + concurrently_ + ( do + threadDelay 65000 + atomically $ modifyTVar' epoch (+ 1) + threadDelay 110000 + atomically $ modifyTVar' epoch (+ 1) + ) + ( withRetry epoch $ \delay loop -> do ints <- addInterval intervals ts atomically $ modifyTVar' reportedIntervals (delay :) when (length ints < 12) $ loop