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

@@ -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

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

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

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

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

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

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