xftp: relative file paths, worker management (#694)

src/Simplex/FileTransfer/Agent.hs

@@ -7,12 +7,15 @@
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TupleSections #-}
 {-# LANGUAGE TypeApplications #-}
 
 module Simplex.FileTransfer.Agent
-  ( -- Receiving files
+  ( startWorkers,
+    closeXFTPAgent,
+    toFSFilePath,
+    -- Receiving files
     receiveFile,
-    addXFTPWorker,
     deleteRcvFile,
     -- Sending files
     sendFileExperimental,
@@ -33,6 +36,7 @@ import qualified Data.ByteString.Char8 as B
 import Data.List (isSuffixOf, partition)
 import Data.List.NonEmpty (nonEmpty)
 import qualified Data.List.NonEmpty as L
+import qualified Data.Map.Strict as M
 import Data.Time.Clock (getCurrentTime)
 import Data.Time.Format (defaultTimeLocale, formatTime)
 import Simplex.FileTransfer.Client.Main (CLIError, SendOptions (..), cliSendFile)
@@ -57,19 +61,39 @@ import UnliftIO.Concurrent
 import UnliftIO.Directory
 import qualified UnliftIO.Exception as E
 
-receiveFile :: AgentMonad m => AgentClient -> UserId -> ValidFileDescription 'FRecipient -> Maybe FilePath -> m RcvFileId
-receiveFile c userId (ValidFileDescription fd@FileDescription {chunks}) xftpWorkPath = do
+startWorkers :: AgentMonad m => AgentClient -> Maybe FilePath -> m ()
+startWorkers c workDir = do
+  wd <- asks $ xftpWorkDir . xftpAgent
+  atomically $ writeTVar wd workDir
+  startFiles
+  where
+    startFiles = do
+      pendingRcvServers <- withStore' c getPendingRcvFilesServers
+      forM_ pendingRcvServers $ \s -> addXFTPWorker c (Just s)
+      -- start local worker for files pending decryption,
+      -- no need to make an extra query for the check
+      -- as the worker will check the store anyway
+      addXFTPWorker c Nothing
+
+closeXFTPAgent :: MonadUnliftIO m => XFTPAgent -> m ()
+closeXFTPAgent XFTPAgent {xftpWorkers} = do
+  ws <- atomically $ stateTVar xftpWorkers (,M.empty)
+  mapM_ (uninterruptibleCancel . snd) ws
+
+receiveFile :: AgentMonad m => AgentClient -> UserId -> ValidFileDescription 'FRecipient -> m RcvFileId
+receiveFile c userId (ValidFileDescription fd@FileDescription {chunks}) = do
   g <- asks idsDrg
-  workPath <- maybe getTemporaryDirectory pure xftpWorkPath
+  workPath <- getWorkPath
   ts <- liftIO getCurrentTime
   let isoTime = formatTime defaultTimeLocale "%Y%m%d_%H%M%S_%6q" ts
   prefixPath <- uniqueCombine workPath (isoTime <> "_rcv.xftp")
   createDirectory prefixPath
-  let tmpPath = prefixPath </> "xftp.encrypted"
-  createDirectory tmpPath
-  let savePath = prefixPath </> "xftp.decrypted"
-  createEmptyFile savePath
-  fId <- withStore c $ \db -> createRcvFile db g userId fd prefixPath tmpPath savePath
+  let relPrefixPath = takeFileName prefixPath
+      relTmpPath = relPrefixPath </> "xftp.encrypted"
+      relSavePath = relPrefixPath </> "xftp.decrypted"
+  createDirectory =<< toFSFilePath relTmpPath
+  createEmptyFile =<< toFSFilePath relSavePath
+  fId <- withStore c $ \db -> createRcvFile db g userId fd relPrefixPath relTmpPath relSavePath
   forM_ chunks downloadChunk
   pure fId
   where
@@ -78,6 +102,14 @@ receiveFile c userId (ValidFileDescription fd@FileDescription {chunks}) xftpWork
       addXFTPWorker c (Just server)
     downloadChunk _ = throwError $ INTERNAL "no replicas"
 
+getWorkPath :: AgentMonad m => m FilePath
+getWorkPath = do
+  workDir <- readTVarIO =<< asks (xftpWorkDir . xftpAgent)
+  maybe getTemporaryDirectory pure workDir
+
+toFSFilePath :: AgentMonad m => FilePath -> m FilePath
+toFSFilePath f = (</> f) <$> getWorkPath
+
 createEmptyFile :: AgentMonad m => FilePath -> m ()
 createEmptyFile fPath = do
   h <- openFile fPath AppendMode
@@ -101,7 +133,7 @@ runXFTPWorker :: forall m. AgentMonad m => AgentClient -> XFTPServer -> TMVar ()
 runXFTPWorker c srv doWork = do
   forever $ do
     void . atomically $ readTMVar doWork
-    agentOperationBracket c AORcvNetwork throwWhenInactive runXftpOperation
+    agentOperationBracket c AORcvNetwork waitUntilActive runXftpOperation
   where
     noWorkToDo = void . atomically $ tryTakeTMVar doWork
     runXftpOperation :: m ()
@@ -133,12 +165,14 @@ runXFTPWorker c srv doWork = do
                   loop
     downloadFileChunk :: RcvFileChunk -> RcvFileChunkReplica -> m ()
     downloadFileChunk RcvFileChunk {userId, rcvFileId, rcvChunkId, chunkNo, chunkSize, digest, fileTmpPath} replica = do
-      chunkPath <- uniqueCombine fileTmpPath $ show chunkNo
+      fsFileTmpPath <- toFSFilePath fileTmpPath
+      chunkPath <- uniqueCombine fsFileTmpPath $ show chunkNo
       let chunkSpec = XFTPRcvChunkSpec chunkPath (unFileSize chunkSize) (unFileDigest digest)
+          relChunkPath = fileTmpPath </> takeFileName chunkPath
       agentXFTPDownloadChunk c userId replica chunkSpec
       fileReceived <- withStore c $ \db -> runExceptT $ do
         -- both actions can be done in a single store method
-        f <- ExceptT $ updateRcvFileChunkReceived db (rcvChunkReplicaId replica) rcvChunkId rcvFileId chunkPath
+        f <- ExceptT $ updateRcvFileChunkReceived db (rcvChunkReplicaId replica) rcvChunkId rcvFileId relChunkPath
         let fileReceived = allChunksReceived f
         when fileReceived $
           liftIO $ updateRcvFileStatus db rcvFileId RFSReceived
@@ -151,7 +185,7 @@ runXFTPWorker c srv doWork = do
 
 workerInternalError :: AgentMonad m => AgentClient -> DBRcvFileId -> RcvFileId -> Maybe FilePath -> String -> m ()
 workerInternalError c rcvFileId rcvFileEntityId tmpPath internalErrStr = do
-  forM_ tmpPath removePath
+  forM_ tmpPath (removePath <=< toFSFilePath)
   withStore' c $ \db -> updateRcvFileError db rcvFileId internalErrStr
   notifyInternalError c rcvFileEntityId internalErrStr
 
@@ -162,6 +196,7 @@ runXFTPLocalWorker :: forall m. AgentMonad m => AgentClient -> TMVar () -> m ()
 runXFTPLocalWorker c@AgentClient {subQ} doWork = do
   forever $ do
     void . atomically $ readTMVar doWork
+    -- TODO agentOperationBracket?
     runXftpOperation
   where
     runXftpOperation :: m ()
@@ -174,16 +209,17 @@ runXFTPLocalWorker c@AgentClient {subQ} doWork = do
     noWorkToDo = void . atomically $ tryTakeTMVar doWork
     decryptFile :: RcvFile -> m ()
     decryptFile RcvFile {rcvFileId, rcvFileEntityId, key, nonce, tmpPath, savePath, chunks} = do
+      fsSavePath <- toFSFilePath savePath
       -- TODO test; recreate file if it's in status RFSDecrypting
       -- when (status == RFSDecrypting) $
-      --   whenM (doesFileExist savePath) (removeFile savePath >> createEmptyFile savePath)
+      --   whenM (doesFileExist fsSavePath) (removeFile fsSavePath >> createEmptyFile fsSavePath)
       withStore' c $ \db -> updateRcvFileStatus db rcvFileId RFSDecrypting
       chunkPaths <- getChunkPaths chunks
       encSize <- liftIO $ foldM (\s path -> (s +) . fromIntegral <$> getFileSize path) 0 chunkPaths
-      void $ liftError (INTERNAL . show) $ decryptChunks encSize chunkPaths key nonce $ \_ -> pure savePath
-      forM_ tmpPath removePath
+      void $ liftError (INTERNAL . show) $ decryptChunks encSize chunkPaths key nonce $ \_ -> pure fsSavePath
+      forM_ tmpPath (removePath <=< toFSFilePath)
       withStore' c (`updateRcvFileComplete` rcvFileId)
-      notify $ RFDONE savePath
+      notify $ RFDONE fsSavePath
       where
         notify :: forall e. AEntityI e => ACommand 'Agent e -> m ()
         notify cmd = atomically $ writeTBQueue subQ ("", rcvFileEntityId, APC (sAEntity @e) cmd)
@@ -191,7 +227,8 @@ runXFTPLocalWorker c@AgentClient {subQ} doWork = do
         getChunkPaths [] = pure []
         getChunkPaths (RcvFileChunk {chunkTmpPath = Just path} : cs) = do
           ps <- getChunkPaths cs
-          pure $ path : ps
+          fsPath <- toFSFilePath path
+          pure $ fsPath : ps
         getChunkPaths (RcvFileChunk {chunkTmpPath = Nothing} : _cs) =
           throwError $ INTERNAL "no chunk path"
 
@@ -204,8 +241,8 @@ deleteRcvFile c userId rcvFileEntityId = do
       withStore' c (`deleteRcvFile'` rcvFileId)
     else withStore' c (`updateRcvFileDeleted` rcvFileId)
 
-sendFileExperimental :: forall m. AgentMonad m => AgentClient -> UserId -> FilePath -> Int -> Maybe FilePath -> m SndFileId
-sendFileExperimental AgentClient {subQ, xftpServers} userId filePath numRecipients xftpWorkPath = do
+sendFileExperimental :: forall m. AgentMonad m => AgentClient -> UserId -> FilePath -> Int -> m SndFileId
+sendFileExperimental AgentClient {subQ, xftpServers} userId filePath numRecipients = do
   g <- asks idsDrg
   sndFileId <- liftIO $ randomId g 12
   xftpSrvs <- atomically $ TM.lookup userId xftpServers
@@ -217,7 +254,7 @@ sendFileExperimental AgentClient {subQ, xftpServers} userId filePath numRecipien
     sendCLI :: SndFileId -> [XFTPServerWithAuth] -> m ()
     sendCLI sndFileId xftpSrvs = do
       let fileName = takeFileName filePath
-      workPath <- maybe getTemporaryDirectory pure xftpWorkPath
+      workPath <- getWorkPath
       outputDir <- uniqueCombine workPath $ fileName <> ".descr"
       createDirectory outputDir
       let tempPath = workPath </> "snd"
@@ -234,6 +271,8 @@ sendFileExperimental AgentClient {subQ, xftpServers} userId filePath numRecipien
               }
       liftCLI $ cliSendFile sendOptions
       (sndDescr, rcvDescrs) <- readDescrs outputDir fileName
+      removePath tempPath
+      removePath outputDir
       notify sndFileId $ SFDONE sndDescr rcvDescrs
     liftCLI :: ExceptT CLIError IO () -> m ()
     liftCLI = either (throwError . INTERNAL . show) pure <=< liftIO . runExceptT
@@ -250,9 +289,9 @@ sendFileExperimental AgentClient {subQ, xftpServers} userId filePath numRecipien
     notify :: forall e. AEntityI e => SndFileId -> ACommand 'Agent e -> m ()
     notify sndFileId cmd = atomically $ writeTBQueue subQ ("", sndFileId, APC (sAEntity @e) cmd)
 
--- _sendFile :: AgentMonad m => AgentClient -> UserId -> Int -> FilePath -> FilePath -> m SndFileId
-_sendFile :: AgentClient -> UserId -> Int -> FilePath -> FilePath -> m SndFileId
-_sendFile _c _userId _numRecipients _xftpPath _filePath = do
+-- _sendFile :: AgentMonad m => AgentClient -> UserId -> FilePath -> Int -> m SndFileId
+_sendFile :: AgentClient -> UserId -> FilePath -> Int -> m SndFileId
+_sendFile _c _userId _filePath _numRecipients = do
   -- db: create file in status New without chunks
   -- add local snd worker for encryption
   -- return file id to client

src/Simplex/Messaging/Agent.hs

@@ -80,6 +80,7 @@ module Simplex.Messaging.Agent
     getNtfToken,
     getNtfTokenData,
     toggleConnectionNtfs,
+    xftpStartWorkers,
     xftpReceiveFile,
     xftpDeleteRcvFile,
     xftpSendFile,
@@ -116,7 +117,7 @@ import qualified Data.Text as T
 import Data.Time.Clock
 import Data.Time.Clock.System (systemToUTCTime)
 import qualified Database.SQLite.Simple as DB
-import Simplex.FileTransfer.Agent (addXFTPWorker, deleteRcvFile, receiveFile, sendFileExperimental)
+import Simplex.FileTransfer.Agent (closeXFTPAgent, deleteRcvFile, receiveFile, sendFileExperimental, startWorkers, toFSFilePath)
 import Simplex.FileTransfer.Description (ValidFileDescription)
 import Simplex.FileTransfer.Protocol (FileParty (..))
 import Simplex.FileTransfer.Util (removePath)
@@ -157,22 +158,13 @@ getSMPAgentClient cfg initServers = newSMPAgentEnv cfg >>= runReaderT runAgent
     runAgent = do
       c <- getAgentClient initServers
       void $ raceAny_ [subscriber c, runNtfSupervisor c, cleanupManager c] `forkFinally` const (disconnectAgentClient c)
-      runExceptT (startFiles c) >>= \case
-        Left e -> liftIO $ print e
-        Right _ -> pure ()
       pure c
-    startFiles c = do
-      pendingRcvServers <- withStore' c getPendingRcvFilesServers
-      forM_ pendingRcvServers $ \s -> addXFTPWorker c (Just s)
-      -- start local worker for files pending decryption,
-      -- no need to make an extra query for the check
-      -- as the worker will check the store anyway
-      addXFTPWorker c Nothing
 
 disconnectAgentClient :: MonadUnliftIO m => AgentClient -> m ()
-disconnectAgentClient c@AgentClient {agentEnv = Env {ntfSupervisor = ns}} = do
+disconnectAgentClient c@AgentClient {agentEnv = Env {ntfSupervisor = ns, xftpAgent = xa}} = do
   closeAgentClient c
-  liftIO $ closeNtfSupervisor ns
+  closeNtfSupervisor ns
+  closeXFTPAgent xa
   logConnection c False
 
 resumeAgentClient :: MonadIO m => AgentClient -> m ()
@@ -339,17 +331,20 @@ getNtfTokenData c = withAgentEnv c $ getNtfTokenData' c
 toggleConnectionNtfs :: AgentErrorMonad m => AgentClient -> ConnId -> Bool -> m ()
 toggleConnectionNtfs c = withAgentEnv c .: toggleConnectionNtfs' c
 
+xftpStartWorkers :: AgentErrorMonad m => AgentClient -> Maybe FilePath -> m ()
+xftpStartWorkers c = withAgentEnv c . startWorkers c
+
 -- | Receive XFTP file
-xftpReceiveFile :: AgentErrorMonad m => AgentClient -> UserId -> ValidFileDescription 'FRecipient -> Maybe FilePath -> m RcvFileId
-xftpReceiveFile c = withAgentEnv c .:. receiveFile c
+xftpReceiveFile :: AgentErrorMonad m => AgentClient -> UserId -> ValidFileDescription 'FRecipient -> m RcvFileId
+xftpReceiveFile c = withAgentEnv c .: receiveFile c
 
 -- | Delete XFTP rcv file (deletes work files from file system and db records)
 xftpDeleteRcvFile :: AgentErrorMonad m => AgentClient -> UserId -> RcvFileId -> m ()
 xftpDeleteRcvFile c = withAgentEnv c .: deleteRcvFile c
 
 -- | Send XFTP file
-xftpSendFile :: AgentErrorMonad m => AgentClient -> UserId -> FilePath -> Int -> Maybe FilePath -> m SndFileId
-xftpSendFile c = withAgentEnv c .:: sendFileExperimental c
+xftpSendFile :: AgentErrorMonad m => AgentClient -> UserId -> FilePath -> Int -> m SndFileId
+xftpSendFile c = withAgentEnv c .:. sendFileExperimental c
 
 -- | Activate operations
 activateAgent :: MonadUnliftIO m => AgentClient -> m ()
@@ -1625,12 +1620,12 @@ cleanupManager c = do
       -- cleanup rcv files marked for deletion
       rcvDeleted <- withStore' c getCleanupRcvFilesDeleted
       forM_ rcvDeleted $ \(fId, p) -> do
-        removePath p
+        removePath =<< toFSFilePath p
         withStore' c (`deleteRcvFile'` fId)
       -- cleanup rcv tmp paths
       rcvTmpPaths <- withStore' c getCleanupRcvFilesTmpPaths
       forM_ rcvTmpPaths $ \(fId, p) -> do
-        removePath p
+        removePath =<< toFSFilePath p
         withStore' c (`updateRcvFileNoTmpPath` fId)
 
 processSMPTransmission :: forall m. AgentMonad m => AgentClient -> ServerTransmission BrokerMsg -> m ()

src/Simplex/Messaging/Agent/Env/SQLite.hs

@@ -35,14 +35,11 @@ import Control.Monad.Reader
 import Crypto.Random
 import Data.List.NonEmpty (NonEmpty)
 import Data.Map (Map)
-import Data.Set (Set)
-import qualified Data.Set as S
 import Data.Time.Clock (NominalDiffTime, nominalDay)
 import Data.Word (Word16)
 import Network.Socket
 import Numeric.Natural
 import Simplex.FileTransfer.Client (XFTPClientConfig (..), defaultXFTPClientConfig)
-import Simplex.FileTransfer.Types (DBSndFileId)
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval
 import Simplex.Messaging.Agent.Store.SQLite
@@ -220,18 +217,19 @@ newNtfSubSupervisor qSize = do
   pure NtfSupervisor {ntfTkn, ntfSubQ, ntfWorkers, ntfSMPWorkers}
 
 data XFTPAgent = XFTPAgent
-  { xftpWorkers :: TMap (Maybe XFTPServer) (TMVar (), Async ()),
+  { -- if set, XFTP file paths will be considered as relative to this directory
+    xftpWorkDir :: TVar (Maybe FilePath),
+    xftpWorkers :: TMap (Maybe XFTPServer) (TMVar (), Async ())
     -- separate send workers for unhindered concurrency between download and upload,
     -- clients can also be separate by passing direction to withXFTPClient, and differentiating by it
-    xftpSndWorkers :: TMap (Maybe XFTPServer) (TMVar (), Async ()),
+    -- xftpSndWorkers :: TMap (Maybe XFTPServer) (TMVar (), Async ()),
     -- files currently in upload - to throttle upload of other files' chunks,
     -- this optimization can be dropped for the MVP
-    xftpSndFiles :: TVar (Set DBSndFileId)
+    -- xftpSndFiles :: TVar (Set DBSndFileId)
   }
 
 newXFTPAgent :: STM XFTPAgent
 newXFTPAgent = do
+  xftpWorkDir <- newTVar Nothing
   xftpWorkers <- TM.empty
-  xftpSndWorkers <- TM.empty
-  xftpSndFiles <- newTVar S.empty
-  pure XFTPAgent {xftpWorkers, xftpSndWorkers, xftpSndFiles}
+  pure XFTPAgent {xftpWorkDir, xftpWorkers}

src/Simplex/Messaging/Agent/NtfSubSupervisor.hs

@@ -17,12 +17,10 @@ module Simplex.Messaging.Agent.NtfSubSupervisor
   )
 where
 
-import Control.Concurrent.Async (Async, uninterruptibleCancel)
 import Control.Concurrent.STM (stateTVar)
 import Control.Logger.Simple (logError, logInfo)
 import Control.Monad
 import Control.Monad.Except
-import Control.Monad.IO.Unlift (MonadUnliftIO)
 import Control.Monad.Reader
 import Data.Bifunctor (first)
 import Data.Fixed (Fixed (MkFixed), Pico)
@@ -44,10 +42,9 @@ import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Util (tshow, unlessM)
 import System.Random (randomR)
-import UnliftIO (async)
+import UnliftIO
 import UnliftIO.Concurrent (forkIO, threadDelay)
 import qualified UnliftIO.Exception as E
-import UnliftIO.STM
 
 runNtfSupervisor :: forall m. AgentMonad' m => AgentClient -> m ()
 runNtfSupervisor c = do
@@ -349,12 +346,12 @@ instantNotifications = \case
   Just NtfToken {ntfTknStatus = NTActive, ntfMode = NMInstant} -> True
   _ -> False
 
-closeNtfSupervisor :: NtfSupervisor -> IO ()
+closeNtfSupervisor :: MonadUnliftIO m => NtfSupervisor -> m ()
 closeNtfSupervisor ns = do
   cancelNtfWorkers_ $ ntfWorkers ns
   cancelNtfWorkers_ $ ntfSMPWorkers ns
 
-cancelNtfWorkers_ :: TMap (ProtocolServer s) (TMVar (), Async ()) -> IO ()
+cancelNtfWorkers_ :: MonadUnliftIO m => TMap (ProtocolServer s) (TMVar (), Async ()) -> m ()
 cancelNtfWorkers_ wsVar = do
   ws <- atomically $ stateTVar wsVar (,M.empty)
   mapM_ (uninterruptibleCancel . snd) ws

tests/XFTPAgent.hs

@@ -13,7 +13,7 @@ import qualified Data.ByteString.Char8 as B
 import SMPAgentClient (agentCfg, initAgentServers)
 import Simplex.FileTransfer.Description
 import Simplex.FileTransfer.Protocol (FileParty (..))
-import Simplex.Messaging.Agent (disconnectAgentClient, getSMPAgentClient, xftpDeleteRcvFile, xftpReceiveFile, xftpSendFile)
+import Simplex.Messaging.Agent (disconnectAgentClient, getSMPAgentClient, xftpDeleteRcvFile, xftpReceiveFile, xftpSendFile, xftpStartWorkers)
 import Simplex.Messaging.Agent.Protocol (ACommand (..), AgentErrorType (..))
 import Simplex.Messaging.Encoding.String (StrEncoding (..))
 import System.Directory (doesDirectoryExist, getFileSize, listDirectory)
@@ -49,8 +49,9 @@ testXFTPAgentReceive = withXFTPServer $ do
   -- receive file using agent
   rcp <- getSMPAgentClient agentCfg initAgentServers
   runRight_ $ do
+    xftpStartWorkers rcp (Just recipientFiles)
     fd :: ValidFileDescription 'FRecipient <- getFileDescription fdRcv
-    fId <- xftpReceiveFile rcp 1 fd (Just recipientFiles)
+    fId <- xftpReceiveFile rcp 1 fd
     ("", fId', RFDONE path) <- rfGet rcp
     liftIO $ do
       fId' `shouldBe` fId
@@ -87,8 +88,9 @@ testXFTPAgentReceiveRestore = withGlobalLogging logCfgNoLogs $ do
   -- receive file using agent - should not succeed due to server being down
   rcp <- getSMPAgentClient agentCfg initAgentServers
   fId <- runRight $ do
+    xftpStartWorkers rcp (Just recipientFiles)
     fd :: ValidFileDescription 'FRecipient <- getFileDescription fdRcv
-    fId <- xftpReceiveFile rcp 1 fd (Just recipientFiles)
+    fId <- xftpReceiveFile rcp 1 fd
     liftIO $ timeout 300000 (get rcp) `shouldReturn` Nothing -- wait for worker attempt
     pure fId
   disconnectAgentClient rcp
@@ -97,9 +99,10 @@ testXFTPAgentReceiveRestore = withGlobalLogging logCfgNoLogs $ do
   let tmpPath = recipientFiles </> prefixDir </> "xftp.encrypted"
   doesDirectoryExist tmpPath `shouldReturn` True
 
-  rcp' <- getSMPAgentClient agentCfg initAgentServers
   withXFTPServerStoreLogOn $ \_ -> do
     -- receive file using agent - should succeed with server up
+    rcp' <- getSMPAgentClient agentCfg initAgentServers
+    runRight_ $ xftpStartWorkers rcp' (Just recipientFiles)
     ("", fId', RFDONE path) <- rfGet rcp'
     liftIO $ do
       fId' `shouldBe` fId
@@ -130,8 +133,9 @@ testXFTPAgentReceiveCleanup = withGlobalLogging logCfgNoLogs $ do
   -- receive file using agent - should not succeed due to server being down
   rcp <- getSMPAgentClient agentCfg initAgentServers
   fId <- runRight $ do
+    xftpStartWorkers rcp (Just recipientFiles)
     fd :: ValidFileDescription 'FRecipient <- getFileDescription fdRcv
-    fId <- xftpReceiveFile rcp 1 fd (Just recipientFiles)
+    fId <- xftpReceiveFile rcp 1 fd
     liftIO $ timeout 300000 (get rcp) `shouldReturn` Nothing -- wait for worker attempt
     pure fId
   disconnectAgentClient rcp
@@ -140,9 +144,10 @@ testXFTPAgentReceiveCleanup = withGlobalLogging logCfgNoLogs $ do
   let tmpPath = recipientFiles </> prefixDir </> "xftp.encrypted"
   doesDirectoryExist tmpPath `shouldReturn` True
 
-  -- receive file using agent - should fail with AUTH error
-  rcp' <- getSMPAgentClient agentCfg initAgentServers
   withXFTPServerThreadOn $ \_ -> do
+    -- receive file using agent - should fail with AUTH error
+    rcp' <- getSMPAgentClient agentCfg initAgentServers
+    runRight_ $ xftpStartWorkers rcp' (Just recipientFiles)
     ("", fId', RFERR (INTERNAL "XFTP {xftpErr = AUTH}")) <- rfGet rcp'
     fId' `shouldBe` fId
 
@@ -160,20 +165,17 @@ testXFTPAgentSendExperimental = withXFTPServer $ do
   -- send file using experimental agent API
   sndr <- getSMPAgentClient agentCfg initAgentServers
   rfd <- runRight $ do
-    sfId <- xftpSendFile sndr 1 filePath 2 $ Just senderFiles
-    ("", sfId', SFDONE sndDescr rcvDescrs) <- sfGet sndr
-    liftIO $ do
-      sfId' `shouldBe` sfId
-      strDecode <$> B.readFile (senderFiles </> "testfile.descr/testfile.xftp/snd.xftp.private") `shouldReturn` Right sndDescr
-      Right rfd1 <- strDecode <$> B.readFile (senderFiles </> "testfile.descr/testfile.xftp/rcv1.xftp")
-      Right rfd2 <- strDecode <$> B.readFile (senderFiles </> "testfile.descr/testfile.xftp/rcv2.xftp")
-      rcvDescrs `shouldMatchList` [rfd1, rfd2]
-      pure rfd1
+    xftpStartWorkers sndr (Just senderFiles)
+    sfId <- xftpSendFile sndr 1 filePath 2
+    ("", sfId', SFDONE _sndDescr [rfd1, _rfd2]) <- sfGet sndr
+    liftIO $ sfId' `shouldBe` sfId
+    pure rfd1
 
   -- receive file using agent
   rcp <- getSMPAgentClient agentCfg initAgentServers
   runRight_ $ do
-    rfId <- xftpReceiveFile rcp 1 rfd (Just recipientFiles)
+    xftpStartWorkers rcp (Just recipientFiles)
+    rfId <- xftpReceiveFile rcp 1 rfd
     ("", rfId', RFDONE path) <- rfGet rcp
     liftIO $ do
       rfId' `shouldBe` rfId