agent: retry loop that resumes subscriptions as soon as agent is moved to foreground, suspend retry loops while agent is suspended (#1249)

* agent: retry loop that resumes subscriptions as soon as agent is moved to foreground, suspend retry loops while agent is suspended

* reset retry enterval when moving to foreground

* account for network state too

* simplify

* typo

* simplify

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

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

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' #-}

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

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)

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 =

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

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