xftp: delete snd files on servers (#722)

This commit is contained in:
spaced4ndy authored and GitHub committed 2023-04-13 17:46:43 +04:00
1 parent e6e17ec637
commit e346a81e46
11 files changed
+346 -103

No files matched your search

+81 -7
View File
@@ -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
-1
View File
@@ -4,7 +4,6 @@
{-# LANGUAGE DuplicateRecordFields #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE KindSignatures #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE PatternSynonyms #-}
+11
View File
@@ -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)
+12 -3
View File
@@ -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)
+18 -13
View File
@@ -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
+4 -4
View File
@@ -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}
+2
View File
@@ -539,4 +539,6 @@ data StoreError
SEXFTPServerNotFound
| -- | XFTP File not found.
SEFileNotFound
| -- | XFTP Deleted snd chunk replica not found.
SEDeletedSndChunkReplicaNotFound
deriving (Eq, Show, Exception)
+94 -8
View File
@@ -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)
@@ -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;
@@ -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
);
+88 -67
View File
@@ -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