agent: restart the retry loops when the network changes

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