From e346a81e46529b0ae0ef530a5be4c6d259701aea Mon Sep 17 00:00:00 2001 From: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com> Date: Thu, 13 Apr 2023 17:46:43 +0400 Subject: [PATCH] xftp: delete snd files on servers (#722) --- src/Simplex/FileTransfer/Agent.hs | 88 +++++++++- src/Simplex/FileTransfer/Description.hs | 1 - src/Simplex/FileTransfer/Types.hs | 11 ++ src/Simplex/Messaging/Agent.hs | 15 +- src/Simplex/Messaging/Agent/Client.hs | 31 ++-- src/Simplex/Messaging/Agent/Env/SQLite.hs | 8 +- src/Simplex/Messaging/Agent/Store.hs | 2 + src/Simplex/Messaging/Agent/Store/SQLite.hs | 102 +++++++++++- .../SQLite/Migrations/M20230401_snd_files.hs | 19 +++ .../Store/SQLite/Migrations/agent_schema.sql | 17 ++ tests/XFTPAgent.hs | 155 ++++++++++-------- 11 files changed, 346 insertions(+), 103 deletions(-) diff --git a/src/Simplex/FileTransfer/Agent.hs b/src/Simplex/FileTransfer/Agent.hs index f032070f7..22e193848 100644 --- a/src/Simplex/FileTransfer/Agent.hs +++ b/src/Simplex/FileTransfer/Agent.hs @@ -21,6 +21,7 @@ module Simplex.FileTransfer.Agent sendFileExperimental, sendFile, deleteSndFileInternal, + deleteSndFileRemote, ) where @@ -65,7 +66,7 @@ import Simplex.Messaging.Encoding.String import Simplex.Messaging.Protocol (EntityId, XFTPServer, XFTPServerWithAuth) import Simplex.Messaging.TMap (TMap) import qualified Simplex.Messaging.TMap as TM -import Simplex.Messaging.Util (liftError, liftIOEither, tshow, whenM) +import Simplex.Messaging.Util (bshow, liftError, liftIOEither, tshow, whenM) import System.FilePath (takeFileName, ()) import UnliftIO import UnliftIO.Concurrent @@ -78,9 +79,10 @@ startWorkers c workDir = do atomically $ writeTVar wd workDir startRcvFiles startSndFiles + startDelFiles where startRcvFiles = do - rcvFilesTTL <- asks (rcvFilesTTL . config) + rcvFilesTTL <- asks $ rcvFilesTTL . config pendingRcvServers <- withStore' c (`getPendingRcvFilesServers` rcvFilesTTL) forM_ pendingRcvServers $ \s -> addXFTPRcvWorker c (Just s) -- start local worker for files pending decryption, @@ -88,11 +90,15 @@ startWorkers c workDir = do -- as the worker will check the store anyway addXFTPRcvWorker c Nothing startSndFiles = do - sndFilesTTL <- asks (sndFilesTTL . config) + sndFilesTTL <- asks $ sndFilesTTL . config -- start worker for files pending encryption/creation addXFTPSndWorker c Nothing pendingSndServers <- withStore' c (`getPendingSndFilesServers` sndFilesTTL) forM_ pendingSndServers $ \s -> addXFTPSndWorker c (Just s) + startDelFiles = do + rcvFilesTTL <- asks $ rcvFilesTTL . config + pendingDelServers <- withStore' c (`getPendingDelFilesServers` rcvFilesTTL) + forM_ pendingDelServers $ addXFTPDelWorker c closeXFTPAgent :: MonadUnliftIO m => XFTPAgent -> m () closeXFTPAgent XFTPAgent {xftpRcvWorkers, xftpSndWorkers} = do @@ -176,7 +182,7 @@ runXFTPRcvWorker c srv doWork = do case nextChunk of Nothing -> noWorkToDo Just RcvFileChunk {rcvFileId, rcvFileEntityId, fileTmpPath, replicas = []} -> rcvWorkerInternalError c rcvFileId rcvFileEntityId (Just fileTmpPath) "chunk has no replicas" - Just fc@RcvFileChunk {userId, rcvFileId, rcvFileEntityId, fileTmpPath, replicas = replica@RcvFileChunkReplica {rcvChunkReplicaId, server, replicaId, delay} : _} -> do + Just fc@RcvFileChunk {userId, rcvFileId, rcvFileEntityId, rcvChunkId, fileTmpPath, replicas = replica@RcvFileChunkReplica {rcvChunkReplicaId, server, delay} : _} -> do ri <- asks $ reconnectInterval . config let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay withRetryInterval ri' $ \delay' loop -> @@ -187,7 +193,7 @@ runXFTPRcvWorker c srv doWork = do flip catchError (\_ -> pure ()) $ do notifyOnRetry <- asks (xftpNotifyErrsOnRetry . config) when notifyOnRetry $ notify c rcvFileEntityId $ RFERR e - closeXFTPServerClient c userId server replicaId + closeXFTPServerClient c userId server $ bshow rcvChunkId withStore' c $ \db -> updateRcvChunkReplicaDelay db rcvChunkReplicaId replicaDelay -- TODO waitUntilNotSuspended atomically $ endAgentOperation c AORcvNetwork @@ -430,6 +436,8 @@ runXFTPSndPrepareWorker c doWork = do atomically $ beginAgentOperation c AOSndNetwork loop createWithNextSrv usedSrvs = do + deleted <- withStore' c $ \db -> getSndFileDeleted db sndFileId + when deleted $ throwError $ INTERNAL "file deleted, aborting chunk creation" withNextSrv c userId usedSrvs [] $ \srvAuth -> do replica <- agentXFTPNewChunk c ch numRecipients' srvAuth pure (replica, srvAuth) @@ -455,7 +463,7 @@ runXFTPSndWorker c srv doWork = do case nextChunk of Nothing -> noWorkToDo Just SndFileChunk {sndFileId, sndFileEntityId, filePrefixPath, replicas = []} -> sndWorkerInternalError c sndFileId sndFileEntityId (Just filePrefixPath) "chunk has no replicas" - Just fc@SndFileChunk {userId, sndFileId, sndFileEntityId, filePrefixPath, replicas = replica@SndFileChunkReplica {sndChunkReplicaId, server, replicaId, delay} : _} -> do + Just fc@SndFileChunk {userId, sndFileId, sndChunkId, sndFileEntityId, filePrefixPath, replicas = replica@SndFileChunkReplica {sndChunkReplicaId, server, delay} : _} -> do ri <- asks $ reconnectInterval . config let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay withRetryInterval ri' $ \delay' loop -> @@ -466,7 +474,7 @@ runXFTPSndWorker c srv doWork = do flip catchError (\_ -> pure ()) $ do notifyOnRetry <- asks (xftpNotifyErrsOnRetry . config) when notifyOnRetry $ notify c sndFileEntityId $ SFERR e - closeXFTPServerClient c userId server replicaId + closeXFTPServerClient c userId server $ bshow sndChunkId withStore' c $ \db -> updateSndChunkReplicaDelay db sndChunkReplicaId replicaDelay -- TODO waitUntilNotSuspended atomically $ endAgentOperation c AOSndNetwork @@ -578,3 +586,69 @@ deleteSndFileInternal c userId sndFileEntityId = do forM_ prefixPath $ removePath <=< toFSFilePath withStore' c (`deleteSndFile'` sndFileId) else withStore' c (`updateSndFileDeleted` sndFileId) + +deleteSndFileRemote :: forall m. AgentMonad m => AgentClient -> UserId -> SndFileId -> ValidFileDescription 'FSender -> m () +deleteSndFileRemote c userId sndFileEntityId (ValidFileDescription FileDescription {chunks}) = do + deleteSndFileInternal c userId sndFileEntityId `catchError` (notify c sndFileEntityId . SFERR) + forM_ chunks $ \ch -> deleteFileChunk ch `catchError` (notify c sndFileEntityId . SFERR) + where + deleteFileChunk :: FileChunk -> m () + deleteFileChunk FileChunk {replicas = replica@FileChunkReplica {server} : _} = do + withStore' c $ \db -> createDeletedSndChunkReplica db userId replica + addXFTPDelWorker c server + deleteFileChunk _ = pure () + +addXFTPDelWorker :: AgentMonad m => AgentClient -> XFTPServer -> m () +addXFTPDelWorker c srv = do + ws <- asks $ xftpDelWorkers . xftpAgent + atomically (TM.lookup srv ws) >>= \case + Nothing -> do + doWork <- newTMVarIO () + worker <- async $ runXFTPDelWorker c srv doWork `E.finally` atomically (TM.delete srv ws) + atomically $ TM.insert srv (doWork, worker) ws + Just (doWork, _) -> + void . atomically $ tryPutTMVar doWork () + +runXFTPDelWorker :: forall m. AgentMonad m => AgentClient -> XFTPServer -> TMVar () -> m () +runXFTPDelWorker c srv doWork = do + forever $ do + void . atomically $ readTMVar doWork + -- TODO waitUntilNotSuspended + agentOperationBracket c AOSndNetwork waitUntilActive runXFTPOperation + where + noWorkToDo = void . atomically $ tryTakeTMVar doWork + runXFTPOperation :: m () + runXFTPOperation = do + -- no point in deleting files older than rcv ttl, as they will be expired on server + rcvFilesTTL <- asks (rcvFilesTTL . config) + nextReplica <- withStore' c $ \db -> getNextDeletedSndChunkReplica db srv rcvFilesTTL + case nextReplica of + Nothing -> noWorkToDo + Just replica@DeletedSndChunkReplica {deletedSndChunkReplicaId, userId, server, replicaId = ChunkReplicaId replId, delay} -> do + ri <- asks $ reconnectInterval . config + let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay + withRetryInterval ri' $ \delay' loop -> + deleteChunkReplica replica + `catchError` \e -> retryOnError "XFTP del worker" (retryLoop loop e delay') (retryDone e) e + where + retryLoop loop e replicaDelay = do + flip catchError (\_ -> pure ()) $ do + notifyOnRetry <- asks (xftpNotifyErrsOnRetry . config) + when notifyOnRetry $ notify c "" $ SFERR e + closeXFTPServerClient c userId server replId + withStore' c $ \db -> updateDeletedSndChunkReplicaDelay db deletedSndChunkReplicaId replicaDelay + -- TODO waitUntilNotSuspended + atomically $ endAgentOperation c AOSndNetwork + atomically $ throwWhenInactive c + atomically $ beginAgentOperation c AOSndNetwork + loop + retryDone e = delWorkerInternalError c deletedSndChunkReplicaId e + deleteChunkReplica :: DeletedSndChunkReplica -> m () + deleteChunkReplica replica@DeletedSndChunkReplica {userId, deletedSndChunkReplicaId} = do + agentXFTPDeleteChunk c userId replica + withStore' c $ \db -> deleteDeletedSndChunkReplica db deletedSndChunkReplicaId + +delWorkerInternalError :: AgentMonad m => AgentClient -> Int64 -> AgentErrorType -> m () +delWorkerInternalError c deletedSndChunkReplicaId e = do + withStore' c $ \db -> deleteDeletedSndChunkReplica db deletedSndChunkReplicaId + notify c "" $ SFERR e diff --git a/src/Simplex/FileTransfer/Description.hs b/src/Simplex/FileTransfer/Description.hs index 581bf974f..316070f7f 100644 --- a/src/Simplex/FileTransfer/Description.hs +++ b/src/Simplex/FileTransfer/Description.hs @@ -4,7 +4,6 @@ {-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE GADTs #-} {-# LANGUAGE KindSignatures #-} -{-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE PatternSynonyms #-} diff --git a/src/Simplex/FileTransfer/Types.hs b/src/Simplex/FileTransfer/Types.hs index 479b2b9c0..04d471a7c 100644 --- a/src/Simplex/FileTransfer/Types.hs +++ b/src/Simplex/FileTransfer/Types.hs @@ -214,3 +214,14 @@ instance TextEncoding SndFileReplicaStatus where textEncode = \case SFRSCreated -> "created" SFRSUploaded -> "uploaded" + +data DeletedSndChunkReplica = DeletedSndChunkReplica + { deletedSndChunkReplicaId :: Int64, + userId :: Int64, + server :: XFTPServer, + replicaId :: ChunkReplicaId, + replicaKey :: C.APrivateSignKey, + delay :: Maybe Int64, + retries :: Int + } + deriving (Eq, Show) diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index 05a4d1357..1e9c0aa19 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -85,6 +85,7 @@ module Simplex.Messaging.Agent xftpDeleteRcvFile, xftpSendFile, xftpDeleteSndFileInternal, + xftpDeleteSndFileRemote, activateAgent, suspendAgent, execAgentStoreSQL, @@ -120,7 +121,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 (closeXFTPAgent, deleteRcvFile, deleteSndFileInternal, receiveFile, sendFile, startWorkers, toFSFilePath) +import Simplex.FileTransfer.Agent (closeXFTPAgent, deleteRcvFile, deleteSndFileInternal, deleteSndFileRemote, receiveFile, sendFile, startWorkers, toFSFilePath) import Simplex.FileTransfer.Description (ValidFileDescription) import Simplex.FileTransfer.Protocol (FileParty (..)) import Simplex.FileTransfer.Util (removePath) @@ -356,6 +357,10 @@ xftpSendFile c = withAgentEnv c .:. sendFile c xftpDeleteSndFileInternal :: AgentErrorMonad m => AgentClient -> UserId -> SndFileId -> m () xftpDeleteSndFileInternal c = withAgentEnv c .: deleteSndFileInternal c +-- | Delete XFTP snd file chunks on servers +xftpDeleteSndFileRemote :: AgentErrorMonad m => AgentClient -> UserId -> SndFileId -> ValidFileDescription 'FSender -> m () +xftpDeleteSndFileRemote c = withAgentEnv c .:. deleteSndFileRemote c + -- TODO rename setAgentForeground -- | Activate operations @@ -1605,6 +1610,7 @@ cleanupManager c@AgentClient {subQ} = do deleteSndFilesExpired `catchError` (notify "" . SFERR) deleteSndFilesDeleted `catchError` (notify "" . SFERR) deleteSndFilesPrefixPaths `catchError` (notify "" . SFERR) + deleteExpiredReplicasForDeletion `catchError` (notify "" . SFERR) liftIO $ threadDelay' int where deleteConns = @@ -1612,7 +1618,7 @@ cleanupManager c@AgentClient {subQ} = do void $ withStore' c getDeletedConnIds >>= deleteDeletedConns c withStore' c deleteUsersWithoutConns >>= mapM_ (notify "" . DEL_USER) deleteRcvFilesExpired = do - rcvFilesTTL <- asks (rcvFilesTTL . config) + rcvFilesTTL <- asks $ rcvFilesTTL . config rcvExpired <- withStore' c (`getRcvFilesExpired` rcvFilesTTL) forM_ rcvExpired $ \(dbId, entId, p) -> flip catchError (notify entId . RFERR) $ do removePath =<< toFSFilePath p @@ -1628,7 +1634,7 @@ cleanupManager c@AgentClient {subQ} = do removePath =<< toFSFilePath p withStore' c (`updateRcvFileNoTmpPath` dbId) deleteSndFilesExpired = do - sndFilesTTL <- asks (sndFilesTTL . config) + sndFilesTTL <- asks $ sndFilesTTL . config sndExpired <- withStore' c (`getSndFilesExpired` sndFilesTTL) forM_ sndExpired $ \(dbId, entId, p) -> flip catchError (notify entId . SFERR) $ do forM_ p $ removePath <=< toFSFilePath @@ -1643,6 +1649,9 @@ cleanupManager c@AgentClient {subQ} = do forM_ sndPrefixPaths $ \(dbId, entId, p) -> flip catchError (notify entId . SFERR) $ do removePath =<< toFSFilePath p withStore' c (`updateSndFileNoPrefixPath` dbId) + deleteExpiredReplicasForDeletion = do + rcvFilesTTL <- asks $ rcvFilesTTL . config + withStore' c (`deleteDeletedSndChunkReplicasExpired` rcvFilesTTL) notify :: forall e. AEntityI e => EntityId -> ACommand 'Agent e -> ExceptT AgentErrorType m () notify entId cmd = atomically $ writeTBQueue subQ ("", entId, APC (sAEntity @e) cmd) diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs index 3c178842c..38770f256 100644 --- a/src/Simplex/Messaging/Agent/Client.hs +++ b/src/Simplex/Messaging/Agent/Client.hs @@ -56,6 +56,7 @@ module Simplex.Messaging.Agent.Client agentXFTPNewChunk, agentXFTPUploadChunk, agentXFTPAddRecipients, + agentXFTPDeleteChunk, agentCbEncrypt, agentCbDecrypt, cryptoError, @@ -133,7 +134,7 @@ import qualified Simplex.FileTransfer.Client as X import Simplex.FileTransfer.Description (ChunkReplicaId (..), FileDigest (..), kb) import Simplex.FileTransfer.Protocol (FileInfo (..), FileResponse, XFTPErrorType (DIGEST)) import Simplex.FileTransfer.Transport (XFTPRcvChunkSpec (..)) -import Simplex.FileTransfer.Types (NewSndChunkReplica (..), RcvFileChunkReplica (..), SndFileChunk (..), SndFileChunkReplica (..)) +import Simplex.FileTransfer.Types (DeletedSndChunkReplica (..), NewSndChunkReplica (..), RcvFileChunkReplica (..), SndFileChunk (..), SndFileChunkReplica (..)) import Simplex.FileTransfer.Util (uniqueCombine) import Simplex.Messaging.Agent.Env.SQLite import Simplex.Messaging.Agent.Lock @@ -621,9 +622,9 @@ closeClient_ c cVar = do Just (Right client) -> closeProtocolServerClient client `catchAll_` pure () _ -> pure () -closeXFTPServerClient :: AgentMonad' m => AgentClient -> UserId -> XFTPServer -> ChunkReplicaId -> m () -closeXFTPServerClient c userId server (ChunkReplicaId fId) = - mkTransportSession c userId server fId >>= liftIO . closeClient c xftpClients +closeXFTPServerClient :: AgentMonad' m => AgentClient -> UserId -> XFTPServer -> ByteString -> m () +closeXFTPServerClient c userId server entityId = + mkTransportSession c userId server entityId >>= liftIO . closeClient c xftpClients cancelActions :: (Foldable f, Monoid (f (Async ()))) => TVar (f (Async ())) -> IO () cancelActions as = atomically (swapTVar as mempty) >>= mapM_ (forkIO . uninterruptibleCancel) @@ -678,13 +679,13 @@ withNtfClient c srv = withLogClient c (0, srv, Nothing) withXFTPClient :: (AgentMonad m, ProtocolServerClient err msg) => AgentClient -> - (UserId, ProtoServer msg, Int64) -> + (UserId, ProtoServer msg, EntityId) -> ByteString -> (Client msg -> ExceptT (ProtocolClientError err) IO b) -> m b -withXFTPClient c (userId, srv, chunkId) cmdStr action = do - tSess <- mkTransportSession c userId srv $ bshow chunkId - withLogClient c tSess (bshow chunkId) cmdStr action +withXFTPClient c (userId, srv, entityId) cmdStr action = do + tSess <- mkTransportSession c userId srv entityId + withLogClient c tSess entityId cmdStr action liftClient :: (AgentMonad m, Show err, Encoding err) => (err -> AgentErrorType) -> HostName -> ExceptT (ProtocolClientError err) IO a -> m a liftClient protocolError_ = liftError . protocolClientError protocolError_ @@ -1077,7 +1078,7 @@ agentNtfDeleteSubscription c subId NtfToken {ntfServer, ntfPrivKey} = agentXFTPDownloadChunk :: AgentMonad m => AgentClient -> UserId -> Int64 -> RcvFileChunkReplica -> XFTPRcvChunkSpec -> m () agentXFTPDownloadChunk c userId rcvChunkId RcvFileChunkReplica {server, replicaId = ChunkReplicaId fId, replicaKey} chunkSpec = - withXFTPClient c (userId, server, rcvChunkId) "FGET" $ \xftp -> X.downloadXFTPChunk xftp replicaKey fId chunkSpec + withXFTPClient c (userId, server, bshow rcvChunkId) "FGET" $ \xftp -> X.downloadXFTPChunk xftp replicaKey fId chunkSpec agentXFTPNewChunk :: AgentMonad m => AgentClient -> SndFileChunk -> Int -> XFTPServerWithAuth -> m NewSndChunkReplica agentXFTPNewChunk c SndFileChunk {userId, sndChunkId, chunkSpec = XFTPChunkSpec {chunkSize}, digest = FileDigest digest} n (ProtoServerWithAuth srv auth) = do @@ -1091,15 +1092,19 @@ agentXFTPNewChunk c SndFileChunk {userId, sndChunkId, chunkSpec = XFTPChunkSpec pure NewSndChunkReplica {server = srv, replicaId = ChunkReplicaId sndId, replicaKey, rcvIdsKeys = L.toList $ xftpRcvIdsKeys rIds rKeys} agentXFTPUploadChunk :: AgentMonad m => AgentClient -> UserId -> Int64 -> SndFileChunkReplica -> XFTPChunkSpec -> m () -agentXFTPUploadChunk c usedId sndChunkId SndFileChunkReplica {server, replicaId = ChunkReplicaId fId, replicaKey} chunkSpec = - withXFTPClient c (usedId, server, sndChunkId) "FPUT" $ \xftp -> X.uploadXFTPChunk xftp replicaKey fId chunkSpec +agentXFTPUploadChunk c userId sndChunkId SndFileChunkReplica {server, replicaId = ChunkReplicaId fId, replicaKey} chunkSpec = + withXFTPClient c (userId, server, bshow sndChunkId) "FPUT" $ \xftp -> X.uploadXFTPChunk xftp replicaKey fId chunkSpec agentXFTPAddRecipients :: AgentMonad m => AgentClient -> UserId -> Int64 -> SndFileChunkReplica -> Int -> m (NonEmpty (ChunkReplicaId, C.APrivateSignKey)) -agentXFTPAddRecipients c usedId sndChunkId SndFileChunkReplica {server, replicaId = ChunkReplicaId fId, replicaKey} n = do +agentXFTPAddRecipients c userId sndChunkId SndFileChunkReplica {server, replicaId = ChunkReplicaId fId, replicaKey} n = do rKeys <- xftpRcvKeys n - rIds <- withXFTPClient c (usedId, server, sndChunkId) "FADD" $ \xftp -> X.addXFTPRecipients xftp replicaKey fId (L.map fst rKeys) + rIds <- withXFTPClient c (userId, server, bshow sndChunkId) "FADD" $ \xftp -> X.addXFTPRecipients xftp replicaKey fId (L.map fst rKeys) pure $ xftpRcvIdsKeys rIds rKeys +agentXFTPDeleteChunk :: AgentMonad m => AgentClient -> UserId -> DeletedSndChunkReplica -> m () +agentXFTPDeleteChunk c userId DeletedSndChunkReplica {server, replicaId = ChunkReplicaId fId, replicaKey} = + withXFTPClient c (userId, server, fId) "FDEL" $ \xftp -> X.deleteXFTPChunk xftp replicaKey fId + xftpRcvKeys :: AgentMonad m => Int -> m (NonEmpty C.ASignatureKeyPair) xftpRcvKeys n = do rKeys <- liftIO $ replicateM n $ C.generateSignatureKeyPair C.SEd25519 diff --git a/src/Simplex/Messaging/Agent/Env/SQLite.hs b/src/Simplex/Messaging/Agent/Env/SQLite.hs index 25f2e7495..40772ee22 100644 --- a/src/Simplex/Messaging/Agent/Env/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Env/SQLite.hs @@ -210,9 +210,8 @@ data XFTPAgent = XFTPAgent { -- if set, XFTP file paths will be considered as relative to this directory xftpWorkDir :: TVar (Maybe FilePath), xftpRcvWorkers :: 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 ()), + xftpDelWorkers :: TMap XFTPServer (TMVar (), Async ()) } newXFTPAgent :: STM XFTPAgent @@ -220,4 +219,5 @@ newXFTPAgent = do xftpWorkDir <- newTVar Nothing xftpRcvWorkers <- TM.empty xftpSndWorkers <- TM.empty - pure XFTPAgent {xftpWorkDir, xftpRcvWorkers, xftpSndWorkers} + xftpDelWorkers <- TM.empty + pure XFTPAgent {xftpWorkDir, xftpRcvWorkers, xftpSndWorkers, xftpDelWorkers} diff --git a/src/Simplex/Messaging/Agent/Store.hs b/src/Simplex/Messaging/Agent/Store.hs index 248e52fc9..2db7264ba 100644 --- a/src/Simplex/Messaging/Agent/Store.hs +++ b/src/Simplex/Messaging/Agent/Store.hs @@ -539,4 +539,6 @@ data StoreError SEXFTPServerNotFound | -- | XFTP File not found. SEFileNotFound + | -- | XFTP Deleted snd chunk replica not found. + SEDeletedSndChunkReplicaNotFound deriving (Eq, Show, Exception) diff --git a/src/Simplex/Messaging/Agent/Store/SQLite.hs b/src/Simplex/Messaging/Agent/Store/SQLite.hs index b960c547d..9d8246991 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite.hs @@ -162,6 +162,7 @@ module Simplex.Messaging.Agent.Store.SQLite updateSndFileNoPrefixPath, updateSndFileDeleted, deleteSndFile', + getSndFileDeleted, createSndFileReplica, getNextSndChunkToUpload, updateSndChunkReplicaDelay, @@ -171,6 +172,12 @@ module Simplex.Messaging.Agent.Store.SQLite getCleanupSndFilesPrefixPaths, getCleanupSndFilesDeleted, getSndFilesExpired, + createDeletedSndChunkReplica, + getNextDeletedSndChunkReplica, + updateDeletedSndChunkReplicaDelay, + deleteDeletedSndChunkReplica, + getPendingDelFilesServers, + deleteDeletedSndChunkReplicasExpired, -- * utilities withConnection, @@ -2049,7 +2056,7 @@ getNextRcvFileToDecrypt db ttl = do getPendingRcvFilesServers :: DB.Connection -> NominalDiffTime -> IO [XFTPServer] getPendingRcvFilesServers db ttl = do cutoffTs <- addUTCTime (- ttl) <$> getCurrentTime - map toServer + map toXFTPServer <$> DB.query db [sql| @@ -2063,9 +2070,9 @@ getPendingRcvFilesServers db ttl = do AND f.status = ? AND f.deleted = 0 AND f.created_at >= ? |] (RFSReceiving, cutoffTs) - where - toServer :: (NonEmpty TransportHost, ServiceName, C.KeyHash) -> XFTPServer - toServer (host, port, keyHash) = XFTPServer host port keyHash + +toXFTPServer :: (NonEmpty TransportHost, ServiceName, C.KeyHash) -> XFTPServer +toXFTPServer (host, port, keyHash) = XFTPServer host port keyHash getCleanupRcvFilesTmpPaths :: DB.Connection -> IO [(DBRcvFileId, RcvFileId, FilePath)] getCleanupRcvFilesTmpPaths db = @@ -2248,6 +2255,11 @@ deleteSndFile' :: DB.Connection -> DBSndFileId -> IO () deleteSndFile' db sndFileId = DB.execute db "DELETE FROM snd_files WHERE snd_file_id = ?" (Only sndFileId) +getSndFileDeleted :: DB.Connection -> DBSndFileId -> IO Bool +getSndFileDeleted db sndFileId = + fromMaybe True + <$> maybeFirstRow fromOnly (DB.query db "SELECT deleted FROM snd_files WHERE snd_file_id = ?" (Only sndFileId)) + createSndFileReplica :: DB.Connection -> SndFileChunk -> NewSndChunkReplica -> IO () createSndFileReplica db SndFileChunk {sndChunkId} NewSndChunkReplica {server, replicaId, replicaKey, rcvIdsKeys} = do srvId <- createXFTPServer_ db server @@ -2342,7 +2354,7 @@ updateSndChunkReplicaStatus db replicaId status = do getPendingSndFilesServers :: DB.Connection -> NominalDiffTime -> IO [XFTPServer] getPendingSndFilesServers db ttl = do cutoffTs <- addUTCTime (- ttl) <$> getCurrentTime - map toServer + map toXFTPServer <$> DB.query db [sql| @@ -2356,9 +2368,6 @@ getPendingSndFilesServers db ttl = do AND (f.status = ? OR f.status = ?) AND f.deleted = 0 AND f.created_at >= ? |] (SFRSCreated, SFSEncrypted, SFSUploading, cutoffTs) - where - toServer :: (NonEmpty TransportHost, ServiceName, C.KeyHash) -> XFTPServer - toServer (host, port, keyHash) = XFTPServer host port keyHash getCleanupSndFilesPrefixPaths :: DB.Connection -> IO [(DBSndFileId, SndFileId, FilePath)] getCleanupSndFilesPrefixPaths db = @@ -2392,3 +2401,80 @@ getSndFilesExpired db ttl = do WHERE created_at < ? |] (Only cutoffTs) + +createDeletedSndChunkReplica :: DB.Connection -> UserId -> FileChunkReplica -> IO () +createDeletedSndChunkReplica db userId FileChunkReplica {server, replicaId, replicaKey} = do + srvId <- createXFTPServer_ db server + DB.execute + db + "INSERT INTO deleted_snd_chunk_replicas (user_id, xftp_server_id, replica_id, replica_key) VALUES (?,?,?,?)" + (userId, srvId, replicaId, replicaKey) + +getDeletedSndChunkReplica :: DB.Connection -> DBSndFileId -> IO (Either StoreError DeletedSndChunkReplica) +getDeletedSndChunkReplica db deletedSndChunkReplicaId = + firstRow toReplica SEDeletedSndChunkReplicaNotFound $ + DB.query + db + [sql| + SELECT + r.user_id, r.replica_id, r.replica_key, r.delay, r.retries, + s.xftp_host, s.xftp_port, s.xftp_key_hash + FROM deleted_snd_chunk_replicas r + JOIN xftp_servers s ON s.xftp_server_id = r.xftp_server_id + WHERE r.deleted_snd_chunk_replica_id = ? + |] + (Only deletedSndChunkReplicaId) + where + toReplica :: (UserId, ChunkReplicaId, C.APrivateSignKey, Maybe Int64, Int, NonEmpty TransportHost, ServiceName, C.KeyHash) -> DeletedSndChunkReplica + toReplica (userId, replicaId, replicaKey, delay, retries, host, port, keyHash) = + let server = XFTPServer host port keyHash + in DeletedSndChunkReplica {deletedSndChunkReplicaId, userId, server, replicaId, replicaKey, delay, retries} + +getNextDeletedSndChunkReplica :: DB.Connection -> XFTPServer -> NominalDiffTime -> IO (Maybe DeletedSndChunkReplica) +getNextDeletedSndChunkReplica db ProtocolServer {host, port, keyHash} ttl = do + cutoffTs <- addUTCTime (- ttl) <$> getCurrentTime + replicaId_ :: Maybe Int64 <- + maybeFirstRow fromOnly $ + DB.query + db + [sql| + SELECT r.deleted_snd_chunk_replica_id + FROM deleted_snd_chunk_replicas r + JOIN xftp_servers s ON s.xftp_server_id = r.xftp_server_id + WHERE s.xftp_host = ? AND s.xftp_port = ? AND s.xftp_key_hash = ? + AND r.created_at >= ? + ORDER BY r.created_at ASC LIMIT 1 + |] + (host, port, keyHash, cutoffTs) + case replicaId_ of + Nothing -> pure Nothing + Just replicaId -> eitherToMaybe <$> getDeletedSndChunkReplica db replicaId + +updateDeletedSndChunkReplicaDelay :: DB.Connection -> Int64 -> Int64 -> IO () +updateDeletedSndChunkReplicaDelay db deletedSndChunkReplicaId delay = do + updatedAt <- getCurrentTime + DB.execute db "UPDATE deleted_snd_chunk_replicas SET delay = ?, retries = retries + 1, updated_at = ? WHERE deleted_snd_chunk_replica_id = ?" (delay, updatedAt, deletedSndChunkReplicaId) + +deleteDeletedSndChunkReplica :: DB.Connection -> Int64 -> IO () +deleteDeletedSndChunkReplica db deletedSndChunkReplicaId = + DB.execute db "DELETE FROM deleted_snd_chunk_replicas WHERE deleted_snd_chunk_replica_id = ?" (Only deletedSndChunkReplicaId) + +getPendingDelFilesServers :: DB.Connection -> NominalDiffTime -> IO [XFTPServer] +getPendingDelFilesServers db ttl = do + cutoffTs <- addUTCTime (- ttl) <$> getCurrentTime + map toXFTPServer + <$> DB.query + db + [sql| + SELECT DISTINCT + s.xftp_host, s.xftp_port, s.xftp_key_hash + FROM deleted_snd_chunk_replicas r + JOIN xftp_servers s ON s.xftp_server_id = r.xftp_server_id + WHERE r.created_at >= ? + |] + (Only cutoffTs) + +deleteDeletedSndChunkReplicasExpired :: DB.Connection -> NominalDiffTime -> IO () +deleteDeletedSndChunkReplicasExpired db ttl = do + cutoffTs <- addUTCTime (- ttl) <$> getCurrentTime + DB.execute db "DELETE FROM deleted_snd_chunk_replicas WHERE created_at < ?" (Only cutoffTs) diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20230401_snd_files.hs b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20230401_snd_files.hs index 7db1a9f78..d75e21223 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20230401_snd_files.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20230401_snd_files.hs @@ -67,11 +67,30 @@ CREATE TABLE snd_file_chunk_replica_recipients ( ); CREATE INDEX idx_snd_file_chunk_replica_recipients_snd_file_chunk_replica_id ON snd_file_chunk_replica_recipients(snd_file_chunk_replica_id); + +CREATE TABLE deleted_snd_chunk_replicas ( + deleted_snd_chunk_replica_id INTEGER PRIMARY KEY, + user_id INTEGER NOT NULL REFERENCES users ON DELETE CASCADE, + xftp_server_id INTEGER NOT NULL REFERENCES xftp_servers ON DELETE CASCADE, + replica_id BLOB NOT NULL, + replica_key BLOB NOT NULL, + delay INTEGER, + retries INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL DEFAULT (datetime('now')), + updated_at TEXT NOT NULL DEFAULT (datetime('now')) +); + +CREATE INDEX idx_deleted_snd_chunk_replicas_user_id ON deleted_snd_chunk_replicas(user_id); +CREATE INDEX idx_deleted_snd_chunk_replicas_xftp_server_id ON deleted_snd_chunk_replicas(xftp_server_id); |] down_m20230401_snd_files :: Query down_m20230401_snd_files = [sql| +DROP INDEX idx_deleted_snd_chunk_replicas_xftp_server_id; +DROP INDEX idx_deleted_snd_chunk_replicas_user_id; +DROP TABLE deleted_snd_chunk_replicas; + DROP INDEX idx_snd_file_chunk_replica_recipients_snd_file_chunk_replica_id; DROP TABLE snd_file_chunk_replica_recipients; diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql index bfef4afd7..f2fa34218 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql +++ b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql @@ -403,3 +403,20 @@ CREATE TABLE snd_file_chunk_replica_recipients( CREATE INDEX idx_snd_file_chunk_replica_recipients_snd_file_chunk_replica_id ON snd_file_chunk_replica_recipients( snd_file_chunk_replica_id ); +CREATE TABLE deleted_snd_chunk_replicas( + deleted_snd_chunk_replica_id INTEGER PRIMARY KEY, + user_id INTEGER NOT NULL REFERENCES users ON DELETE CASCADE, + xftp_server_id INTEGER NOT NULL REFERENCES xftp_servers ON DELETE CASCADE, + replica_id BLOB NOT NULL, + replica_key BLOB NOT NULL, + delay INTEGER, + retries INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL DEFAULT(datetime('now')), + updated_at TEXT NOT NULL DEFAULT(datetime('now')) +); +CREATE INDEX idx_deleted_snd_chunk_replicas_user_id ON deleted_snd_chunk_replicas( + user_id +); +CREATE INDEX idx_deleted_snd_chunk_replicas_xftp_server_id ON deleted_snd_chunk_replicas( + xftp_server_id +); diff --git a/tests/XFTPAgent.hs b/tests/XFTPAgent.hs index 2f769d9da..aacd4656d 100644 --- a/tests/XFTPAgent.hs +++ b/tests/XFTPAgent.hs @@ -19,9 +19,9 @@ import SMPAgentClient (agentCfg, initAgentServers, testDB) import Simplex.FileTransfer.Description import Simplex.FileTransfer.Protocol (FileParty (..), XFTPErrorType (AUTH)) import Simplex.FileTransfer.Server.Env (XFTPServerConfig (..)) -import Simplex.Messaging.Agent (AgentClient, disconnectAgentClient, testProtocolServer, xftpDeleteRcvFile, xftpDeleteSndFileInternal, xftpReceiveFile, xftpSendFile, xftpStartWorkers) +import Simplex.Messaging.Agent (AgentClient, disconnectAgentClient, testProtocolServer, xftpDeleteRcvFile, xftpDeleteSndFileInternal, xftpDeleteSndFileRemote, xftpReceiveFile, xftpSendFile, xftpStartWorkers) import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestStep (..)) -import Simplex.Messaging.Agent.Protocol (ACommand (..), AgentErrorType (..), BrokerErrorType (..), noAuthSrv) +import Simplex.Messaging.Agent.Protocol (ACommand (..), AgentErrorType (..), BrokerErrorType (..), RcvFileId, SndFileId, noAuthSrv) import Simplex.Messaging.Encoding.String (StrEncoding (..)) import Simplex.Messaging.Protocol (BasicAuth, ProtoServerWithAuth (..), ProtocolServer (..), XFTPServerWithAuth) import System.Directory (doesDirectoryExist, doesFileExist, getFileSize, listDirectory) @@ -38,6 +38,7 @@ xftpAgentTests = around_ testBracket . describe "Functional API" $ do it "should cleanup rcv tmp path after permanent error" testXFTPAgentReceiveCleanup it "should resume sending file after restart" testXFTPAgentSendRestore it "should cleanup snd prefix path after permanent error" testXFTPAgentSendCleanup + it "should delete sent file on server" testXFTPAgentDelete describe "XFTP server test via agent API" $ do it "should pass without basic auth" $ testXFTPServerTest Nothing (noAuthSrv testXFTPServer2) `shouldReturn` Nothing let srv1 = testXFTPServer2 {keyHash = "1234"} @@ -76,39 +77,52 @@ checkProgress (prev, expected) (progress, total) loop testXFTPAgentSendReceive :: IO () testXFTPAgentSendReceive = withXFTPServer $ do - -- create random file using cli + filePath <- createRandomFile + + -- send file, delete snd file internally + sndr <- getSMPAgentClient' agentCfg initAgentServers testDB + (rfd1, rfd2) <- runRight $ do + (sfId, _, rfd1, rfd2) <- testSend sndr filePath + xftpDeleteSndFileInternal sndr 1 sfId + pure (rfd1, rfd2) + + -- receive file, delete rcv file + testReceiveDelete rfd1 filePath + testReceiveDelete rfd2 filePath + where + testReceiveDelete rfd originalFilePath = do + rcp <- getSMPAgentClient' agentCfg initAgentServers testDB + runRight_ $ do + rfId <- testReceive rcp rfd originalFilePath + xftpDeleteRcvFile rcp 1 rfId + +createRandomFile :: IO FilePath +createRandomFile = do let filePath = senderFiles "testfile" xftpCLI ["rand", filePath, "17mb"] `shouldReturn` ["File created: " <> filePath] getFileSize filePath `shouldReturn` mb 17 + pure filePath - -- send file - sndr <- getSMPAgentClient' agentCfg initAgentServers testDB - rfd <- runRight $ do - xftpStartWorkers sndr (Just senderFiles) - sfId <- xftpSendFile sndr 1 filePath 2 - sfProgress sndr $ mb 18 - ("", sfId', SFDONE _sndDescr [rfd1, _rfd2]) <- sfGet sndr - liftIO $ sfId' `shouldBe` sfId +testSend :: AgentClient -> FilePath -> ExceptT AgentErrorType IO (SndFileId, ValidFileDescription 'FSender, ValidFileDescription 'FRecipient, ValidFileDescription 'FRecipient) +testSend sndr filePath = do + xftpStartWorkers sndr (Just senderFiles) + sfId <- xftpSendFile sndr 1 filePath 2 + sfProgress sndr $ mb 18 + ("", sfId', SFDONE sndDescr [rfd1, rfd2]) <- sfGet sndr + liftIO $ sfId' `shouldBe` sfId + pure (sfId, sndDescr, rfd1, rfd2) - -- delete snd file internally - xftpDeleteSndFileInternal sndr 1 sfId - - pure rfd1 - - -- receive file - rcp <- getSMPAgentClient' agentCfg initAgentServers testDB - runRight_ $ do - xftpStartWorkers rcp (Just recipientFiles) - rfId <- xftpReceiveFile rcp 1 rfd - rfProgress rcp $ mb 18 - ("", rfId', RFDONE path) <- rfGet rcp - liftIO $ do - rfId' `shouldBe` rfId - file <- B.readFile filePath - B.readFile path `shouldReturn` file - - -- delete rcv file - xftpDeleteRcvFile rcp 1 rfId +testReceive :: AgentClient -> ValidFileDescription 'FRecipient -> FilePath -> ExceptT AgentErrorType IO RcvFileId +testReceive rcp rfd originalFilePath = do + xftpStartWorkers rcp (Just recipientFiles) + rfId <- xftpReceiveFile rcp 1 rfd + rfProgress rcp $ mb 18 + ("", rfId', RFDONE path) <- rfGet rcp + liftIO $ do + rfId' `shouldBe` rfId + file <- B.readFile originalFilePath + B.readFile path `shouldReturn` file + pure rfId getFileDescription :: FilePath -> ExceptT AgentErrorType IO (ValidFileDescription 'FRecipient) getFileDescription path = @@ -119,21 +133,14 @@ logCfgNoLogs = LogConfig {lc_file = Nothing, lc_stderr = False} testXFTPAgentReceiveRestore :: IO () testXFTPAgentReceiveRestore = withGlobalLogging logCfgNoLogs $ do - -- create random file using cli - let filePath = senderFiles "testfile" - xftpCLI ["rand", filePath, "17mb"] `shouldReturn` ["File created: " <> filePath] - getFileSize filePath `shouldReturn` mb 17 + filePath <- createRandomFile rfd <- withXFTPServerStoreLogOn $ \_ -> do -- send file sndr <- getSMPAgentClient' agentCfg initAgentServers testDB runRight $ do - xftpStartWorkers sndr (Just senderFiles) - sfId <- xftpSendFile sndr 1 filePath 2 - sfProgress sndr $ mb 18 - ("", sfId', SFDONE _sndDescr [rfd1, _rfd2]) <- sfGet sndr - liftIO $ sfId' `shouldBe` sfId - pure rfd1 + (_, _, rfd, _) <- testSend sndr filePath + pure rfd -- receive file - should not succeed with server down rcp <- getSMPAgentClient' agentCfg initAgentServers testDB @@ -174,21 +181,14 @@ testXFTPAgentReceiveRestore = withGlobalLogging logCfgNoLogs $ do testXFTPAgentReceiveCleanup :: IO () testXFTPAgentReceiveCleanup = withGlobalLogging logCfgNoLogs $ do - -- create random file using cli - let filePath = senderFiles "testfile" - xftpCLI ["rand", filePath, "17mb"] `shouldReturn` ["File created: " <> filePath] - getFileSize filePath `shouldReturn` mb 17 + filePath <- createRandomFile rfd <- withXFTPServerStoreLogOn $ \_ -> do -- send file sndr <- getSMPAgentClient' agentCfg initAgentServers testDB runRight $ do - xftpStartWorkers sndr (Just senderFiles) - sfId <- xftpSendFile sndr 1 filePath 2 - sfProgress sndr $ mb 18 - ("", sfId', SFDONE _sndDescr [rfd1, _rfd2]) <- sfGet sndr - liftIO $ sfId' `shouldBe` sfId - pure rfd1 + (_, _, rfd, _) <- testSend sndr filePath + pure rfd -- receive file - should not succeed with server down rcp <- getSMPAgentClient' agentCfg initAgentServers testDB @@ -215,10 +215,7 @@ testXFTPAgentReceiveCleanup = withGlobalLogging logCfgNoLogs $ do testXFTPAgentSendRestore :: IO () testXFTPAgentSendRestore = withGlobalLogging logCfgNoLogs $ do - -- create random file using cli - let filePath = senderFiles "testfile" - xftpCLI ["rand", filePath, "17mb"] `shouldReturn` ["File created: " <> filePath] - getFileSize filePath `shouldReturn` mb 17 + filePath <- createRandomFile -- send file - should not succeed with server down sndr <- getSMPAgentClient' agentCfg initAgentServers testDB @@ -260,22 +257,12 @@ testXFTPAgentSendRestore = withGlobalLogging logCfgNoLogs $ do -- receive file rcp <- getSMPAgentClient' agentCfg initAgentServers testDB - runRight_ $ do - xftpStartWorkers rcp (Just recipientFiles) - rfId <- xftpReceiveFile rcp 1 rfd1 - rfProgress rcp $ mb 18 - ("", rfId', RFDONE path) <- rfGet rcp - liftIO $ do - rfId' `shouldBe` rfId - file <- B.readFile filePath - B.readFile path `shouldReturn` file + runRight_ $ + void $ testReceive rcp rfd1 filePath testXFTPAgentSendCleanup :: IO () testXFTPAgentSendCleanup = withGlobalLogging logCfgNoLogs $ do - -- create random file using cli - let filePath = senderFiles "testfile" - xftpCLI ["rand", filePath, "17mb"] `shouldReturn` ["File created: " <> filePath] - getFileSize filePath `shouldReturn` mb 17 + filePath <- createRandomFile sfId <- withXFTPServerStoreLogOn $ \_ -> do -- send file @@ -309,6 +296,40 @@ testXFTPAgentSendCleanup = withGlobalLogging logCfgNoLogs $ do doesDirectoryExist prefixPath `shouldReturn` False doesFileExist encPath `shouldReturn` False +testXFTPAgentDelete :: IO () +testXFTPAgentDelete = withGlobalLogging logCfgNoLogs $ + withXFTPServer $ do + filePath <- createRandomFile + + -- send file + sndr <- getSMPAgentClient' agentCfg initAgentServers testDB + (sfId, sndDescr, rfd1, rfd2) <- runRight $ testSend sndr filePath + + -- receive file + rcp1 <- getSMPAgentClient' agentCfg initAgentServers testDB + runRight_ $ + void $ testReceive rcp1 rfd1 filePath + + length <$> listDirectory xftpServerFiles `shouldReturn` 6 + + -- delete file + runRight $ do + xftpStartWorkers sndr (Just senderFiles) + xftpDeleteSndFileRemote sndr 1 sfId sndDescr + Nothing <- liftIO $ 100000 `timeout` sfGet sndr + pure () + + threadDelay 1000000 + length <$> listDirectory xftpServerFiles `shouldReturn` 0 + + -- receive file - should fail with AUTH error + rcp2 <- getSMPAgentClient' agentCfg initAgentServers testDB + runRight $ do + xftpStartWorkers rcp2 (Just recipientFiles) + rfId <- xftpReceiveFile rcp2 1 rfd2 + ("", rfId', RFERR (INTERNAL "XFTP {xftpErr = AUTH}")) <- rfGet rcp2 + liftIO $ rfId' `shouldBe` rfId + testXFTPServerTest :: Maybe BasicAuth -> XFTPServerWithAuth -> IO (Maybe ProtocolTestFailure) testXFTPServerTest newFileBasicAuth srv = withXFTPServerCfg testXFTPServerConfig {newFileBasicAuth, xftpPort = xftpTestPort2} $ \_ -> do