xftp: wait for agent in foreground in XFTP workers without delaying suspension

src/Simplex/FileTransfer/Agent.hs

@@ -24,7 +24,7 @@
   )
 where
 
 import Control.Concurrent.STM (stateTVar)
 import Control.Logger.Simple (logError)
 import Control.Monad
 import Control.Monad.Except
@@ -165,8 +165,8 @@
 runXFTPRcvWorker c srv doWork = do
   forever $ do
     void . atomically $ readTMVar doWork
-    -- TODO waitUntilNotSuspended
-    agentOperationBracket c AORcvNetwork waitUntilActive runXFTPOperation
+    atomically $ checkAgentForeground c
+    runXFTPOperation
   where
     noWorkToDo = void . atomically $ tryTakeTMVar doWork
     runXFTPOperation :: m ()
@@ -189,10 +189,7 @@
                 when notifyOnRetry $ notify c rcvFileEntityId $ RFERR e
                 closeXFTPServerClient c userId server replicaId
                 withStore' c $ \db -> updateRcvChunkReplicaDelay db rcvChunkReplicaId replicaDelay
-              -- TODO waitUntilNotSuspended
-              atomically $ endAgentOperation c AORcvNetwork
-              atomically $ throwWhenInactive c
-              atomically $ beginAgentOperation c AORcvNetwork
+              atomically $ checkAgentForeground c
               loop
             retryDone e = rcvWorkerInternalError c rcvFileId rcvFileEntityId (Just fileTmpPath) (show e)
     downloadFileChunk :: RcvFileChunk -> RcvFileChunkReplica -> m ()
@@ -236,7 +233,7 @@
 runXFTPRcvLocalWorker c doWork = do
   forever $ do
     void . atomically $ readTMVar doWork
-    -- TODO waitUntilNotSuspended
+    atomically $ checkAgentForeground c
     runXFTPOperation
   where
     runXFTPOperation :: m ()
@@ -357,7 +354,7 @@
 runXFTPSndPrepareWorker c doWork = do
   forever $ do
     void . atomically $ readTMVar doWork
-    -- TODO waitUntilNotSuspended
+    atomically $ checkAgentForeground c
     runXFTPOperation
   where
     runXFTPOperation :: m ()
@@ -411,7 +408,7 @@
           any (\SndFileChunkReplica {replicaStatus} -> replicaStatus == SFRSCreated) replicas
         createChunk :: Int -> SndFileChunk -> m ()
         createChunk numRecipients' ch = do
-          -- TODO waitUntilNotSuspended
+          atomically $ checkAgentForeground c
           (replica, ProtoServerWithAuth srv _) <- agentOperationBracket c AOSndNetwork throwWhenInactive tryCreate
           withStore' c $ \db -> createSndFileReplica db ch replica
           addXFTPSndWorker c $ Just srv
@@ -423,12 +420,7 @@
                 createWithNextSrv usedSrvs
                   `catchError` \e -> retryOnError "XFTP prepare worker" (retryLoop loop) (throwError e) e
               where
-                retryLoop loop = do
-                  -- TODO waitUntilNotSuspended
-                  atomically $ endAgentOperation c AOSndNetwork
-                  atomically $ throwWhenInactive c
-                  atomically $ beginAgentOperation c AOSndNetwork
-                  loop
+                retryLoop loop = atomically (checkAgentForeground c) >> loop
             createWithNextSrv usedSrvs = do
               withNextSrv c userId usedSrvs [] $ \srvAuth -> do
                 replica <- agentXFTPNewChunk c ch numRecipients' srvAuth
@@ -444,7 +436,7 @@
 runXFTPSndWorker c srv doWork = do
   forever $ do
     void . atomically $ readTMVar doWork
-    -- TODO waitUntilNotSuspended
+    atomically $ checkAgentForeground c
     agentOperationBracket c AOSndNetwork throwWhenInactive runXFTPOperation
   where
     noWorkToDo = void . atomically $ tryTakeTMVar doWork
@@ -468,18 +460,15 @@
                 when notifyOnRetry $ notify c sndFileEntityId $ SFERR e
                 closeXFTPServerClient c userId server replicaId
                 withStore' c $ \db -> updateSndChunkReplicaDelay db sndChunkReplicaId replicaDelay
-              -- TODO waitUntilNotSuspended
-              atomically $ endAgentOperation c AOSndNetwork
-              atomically $ throwWhenInactive c
-              atomically $ beginAgentOperation c AOSndNetwork
+              atomically $ checkAgentForeground c
               loop
             retryDone e = sndWorkerInternalError c sndFileId sndFileEntityId (Just filePrefixPath) (show e)
     uploadFileChunk :: SndFileChunk -> SndFileChunkReplica -> m ()
     uploadFileChunk sndFileChunk@SndFileChunk {sndFileId, sndChunkId, userId, chunkSpec = chunkSpec@XFTPChunkSpec {filePath}} replica = do
       replica'@SndFileChunkReplica {sndChunkReplicaId} <- addRecipients sndFileChunk replica
       fsFilePath <- toFSFilePath filePath
       let chunkSpec' = chunkSpec {filePath = fsFilePath} :: XFTPChunkSpec
-      -- TODO waitUntilNotSuspended
+      atomically $ checkAgentForeground c
       agentXFTPUploadChunk c userId sndChunkId replica' chunkSpec'
       sf@SndFile {sndFileEntityId, prefixPath, chunks} <- withStore c $ \db -> do
         updateSndChunkReplicaStatus db sndChunkReplicaId SFRSUploaded

src/Simplex/FileTransfer/Client.hs

@@ -212,7 +212,5 @@ noFile HTTP2Body {bodyPart} a = case bodyPart of
   Just _ -> pure a -- throwError $ PCEResponseError HAS_FILE
   _ -> pure a
 
--- FADD :: NonEmpty RcvPublicVerifyKey -> FileCommand Sender
--- FDEL :: FileCommand Sender
 -- FACK :: FileCommand Recipient
 -- PING :: FileCommand Recipient

src/Simplex/Messaging/Agent.hs

@@ -85,7 +85,7 @@
     xftpDeleteRcvFile,
     xftpSendFile,
     xftpDeleteSndFileInternal,
-    activateAgent,
+    foregroundAgent,
     suspendAgent,
     execAgentStoreSQL,
     getAgentMigrations,
@@ -96,9 +96,8 @@
   )
 where
 
 import Control.Concurrent.STM (stateTVar)
 import Control.Logger.Simple (logError, logInfo, showText)
-import Control.Monad ((<=<))
 import Control.Monad.Except
 import Control.Monad.IO.Unlift (MonadUnliftIO)
 import Control.Monad.Reader
@@ -356,11 +355,9 @@
 xftpDeleteSndFileInternal :: AgentErrorMonad m => AgentClient -> UserId -> SndFileId -> m ()
 xftpDeleteSndFileInternal c = withAgentEnv c .: deleteSndFileInternal c
 
--- TODO rename setAgentForeground
-
 -- | Activate operations
-activateAgent :: MonadUnliftIO m => AgentClient -> m ()
-activateAgent c = withAgentEnv c $ activateAgent' c
+foregroundAgent :: MonadUnliftIO m => AgentClient -> m ()
+foregroundAgent c = withAgentEnv c $ foregroundAgent' c
 
 -- | Suspend operations with max delay to deliver pending messages
 suspendAgent :: MonadUnliftIO m => AgentClient -> Int -> m ()
@@ -1536,9 +1533,9 @@
 setNtfServers' :: AgentMonad' m => AgentClient -> [NtfServer] -> m ()
 setNtfServers' c = atomically . writeTVar (ntfServers c)
 
-activateAgent' :: AgentMonad' m => AgentClient -> m ()
-activateAgent' c = do
-  atomically $ writeTVar (agentState c) ASActive
+foregroundAgent' :: AgentMonad' m => AgentClient -> m ()
+foregroundAgent' c = do
+  atomically $ writeTVar (agentState c) ASForeground
   mapM_ activate $ reverse agentOperations
   where
     activate opSel = atomically $ modifyTVar' (opSel c) $ \s -> s {opSuspended = False}

src/Simplex/Messaging/Agent/Client.hs

@@ -80,6 +80,8 @@
     throwWhenNoDelivery,
     beginAgentOperation,
     endAgentOperation,
+    waitUntilForeground,
+    checkAgentForeground,
     suspendSendingAndDatabase,
     suspendOperation,
     notifySuspended,
@@ -98,7 +100,7 @@
 import Control.Applicative ((<|>))
 import Control.Concurrent (forkIO, threadDelay)
 import Control.Concurrent.Async (Async, uninterruptibleCancel)
 import Control.Concurrent.STM (retry, stateTVar, throwSTM)
 import Control.Exception (AsyncException (..))
 import Control.Logger.Simple
 import Control.Monad.Except
@@ -265,7 +267,7 @@
 
 data AgentOpState = AgentOpState {opSuspended :: Bool, opsInProgress :: Int}
 
-data AgentState = ASActive | ASSuspending | ASSuspended -- TODO rename ASActive -> ASForeground
+data AgentState = ASForeground | ASSuspending | ASSuspended
   deriving (Eq, Show)
 
 data AgentLocks = AgentLocks {connLocks :: Map String String, srvLocks :: Map String String, delLock :: Maybe String}
@@ -310,7 +312,7 @@
   msgDeliveryOp <- newTVar $ AgentOpState False 0
   sndNetworkOp <- newTVar $ AgentOpState False 0
   databaseOp <- newTVar $ AgentOpState False 0
-  agentState <- newTVar ASActive
+  agentState <- newTVar ASForeground
   getMsgLocks <- TM.empty
   connLocks <- TM.empty
   deleteLock <- createLock
@@ -1207,6 +1209,14 @@
     (\_ -> atomically $ endAgentOperation c op)
     (const action)
 
+waitUntilForeground :: AgentClient -> STM ()
+waitUntilForeground c = unlessM ((ASForeground ==) <$> readTVar (agentState c)) retry
+
+checkAgentForeground :: AgentClient -> STM ()
+checkAgentForeground c = do
+  throwWhenInactive c
+  waitUntilForeground c
+
 withStore' :: AgentMonad m => AgentClient -> (DB.Connection -> IO a) -> m a
 withStore' c action = withStore c $ fmap Right . action
 
@@ -1255,7 +1265,7 @@
   where
     statsKey = AgentStatsKey {userId, host = strEncode $ clientTransportHost pc, clientTs = strEncode $ clientSessionTs pc, cmd, res}
 
 userServers :: forall p. (ProtocolTypeI p, UserProtocol p) => AgentClient -> TMap UserId (NonEmpty (ProtoServerWithAuth p))
 userServers c = case protocolTypeI @p of
   SPSMP -> smpServers c
   SPXFTP -> xftpServers c

tests/AgentTests/FunctionalAPITests.hs

@@ -556,7 +556,7 @@ testSuspendingAgent = do
     5 <- sendMessage a bId SMP.noMsgFlags "hello 2"
     get a ##> ("", bId, SENT 5)
     Nothing <- 100000 `timeout` get b
-    activateAgent b
+    foregroundAgent b
     get b =##> \case ("", c, Msg "hello 2") -> c == aId; _ -> False
 
 testSuspendingAgentCompleteSending :: ATransport -> IO ()