diff --git a/src/Simplex/FileTransfer/Agent.hs b/src/Simplex/FileTransfer/Agent.hs index 9e091b9f1..055cb02a2 100644 --- a/src/Simplex/FileTransfer/Agent.hs +++ b/src/Simplex/FileTransfer/Agent.hs @@ -70,7 +70,8 @@ startWorkers c workDir = do startFiles where startFiles = do - pendingRcvServers <- withStore' c getPendingRcvFilesServers + rcvFilesTTL <- asks (rcvFilesTTL . config) + pendingRcvServers <- withStore' c (`getPendingRcvFilesServers` rcvFilesTTL) forM_ pendingRcvServers $ \s -> addXFTPWorker c (Just s) -- start local worker for files pending decryption, -- no need to make an extra query for the check @@ -140,7 +141,8 @@ runXFTPWorker c srv doWork = do noWorkToDo = void . atomically $ tryTakeTMVar doWork runXftpOperation :: m () runXftpOperation = do - nextChunk <- withStore' c (`getNextRcvChunkToDownload` srv) + rcvFilesTTL <- asks (rcvFilesTTL . config) + nextChunk <- withStore' c $ \db -> getNextRcvChunkToDownload db srv rcvFilesTTL case nextChunk of Nothing -> noWorkToDo Just RcvFileChunk {rcvFileId, rcvFileEntityId, fileTmpPath, replicas = []} -> workerInternalError c rcvFileId rcvFileEntityId (Just fileTmpPath) "chunk has no replicas" @@ -207,7 +209,8 @@ runXFTPLocalWorker c doWork = do where runXftpOperation :: m () runXftpOperation = do - nextFile <- withStore' c getNextRcvFileToDecrypt + rcvFilesTTL <- asks (rcvFilesTTL . config) + nextFile <- withStore' c (`getNextRcvFileToDecrypt` rcvFilesTTL) case nextFile of Nothing -> noWorkToDo Just f@RcvFile {rcvFileId, rcvFileEntityId, tmpPath} -> diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index bcf43650f..4231fc9d9 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -1617,7 +1617,8 @@ cleanupManager c@AgentClient {subQ} = do forever $ do void . runExceptT $ do deleteConns `catchError` (notify "" . ERR) - deleteRcvFiles `catchError` (notify "" . RFERR) + deleteRcvFilesExpired `catchError` (notify "" . RFERR) + deleteRcvFilesDeleted `catchError` (notify "" . RFERR) deleteRcvFilesTmpPaths `catchError` (notify "" . RFERR) threadDelay int where @@ -1625,7 +1626,13 @@ cleanupManager c@AgentClient {subQ} = do withLock (deleteLock c) "cleanupManager" $ do void $ withStore' c getDeletedConnIds >>= deleteDeletedConns c withStore' c deleteUsersWithoutConns >>= mapM_ (notify "" . DEL_USER) - deleteRcvFiles = do + deleteRcvFilesExpired = do + rcvFilesTTL <- asks (rcvFilesTTL . config) + rcvExpired <- withStore' c (`getRcvFilesExpired` rcvFilesTTL) + forM_ rcvExpired $ \(dbId, entId, p) -> flip catchError (notify entId . RFERR) $ do + removePath =<< toFSFilePath p + withStore' c (`deleteRcvFile'` dbId) + deleteRcvFilesDeleted = do rcvDeleted <- withStore' c getCleanupRcvFilesDeleted forM_ rcvDeleted $ \(dbId, entId, p) -> flip catchError (notify entId . RFERR) $ do removePath =<< toFSFilePath p diff --git a/src/Simplex/Messaging/Agent/Env/SQLite.hs b/src/Simplex/Messaging/Agent/Env/SQLite.hs index 701f36e0b..aae85d01e 100644 --- a/src/Simplex/Messaging/Agent/Env/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Env/SQLite.hs @@ -81,6 +81,7 @@ data AgentConfig = AgentConfig helloTimeout :: NominalDiffTime, initialCleanupDelay :: Int, cleanupInterval :: Int, + rcvFilesTTL :: NominalDiffTime, deleteErrorCount :: Int, ntfCron :: Word16, ntfWorkerDelay :: Int, @@ -140,6 +141,7 @@ defaultAgentConfig = helloTimeout = 2 * nominalDay, initialCleanupDelay = 30 * 1000000, -- 30 seconds cleanupInterval = 30 * 60 * 1000000, -- 30 minutes + rcvFilesTTL = 2 * nominalDay, deleteErrorCount = 10, ntfCron = 20, -- minutes ntfWorkerDelay = 100000, -- microseconds diff --git a/src/Simplex/Messaging/Agent/Store/SQLite.hs b/src/Simplex/Messaging/Agent/Store/SQLite.hs index 0b40aaf00..e6a4b8cd4 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite.hs @@ -146,6 +146,7 @@ module Simplex.Messaging.Agent.Store.SQLite getPendingRcvFilesServers, getCleanupRcvFilesTmpPaths, getCleanupRcvFilesDeleted, + getRcvFilesExpired, -- * utilities withConnection, @@ -180,7 +181,7 @@ import Data.Ord (Down (..)) import Data.Text (Text) import qualified Data.Text as T import Data.Text.Encoding (decodeLatin1, encodeUtf8) -import Data.Time.Clock (UTCTime, getCurrentTime) +import Data.Time.Clock (NominalDiffTime, UTCTime, addUTCTime, getCurrentTime) import Data.Word (Word32) import Database.SQLite.Simple (FromRow, NamedParam (..), Only (..), Query (..), SQLError, ToRow, field, (:.) (..)) import qualified Database.SQLite.Simple as DB @@ -1966,8 +1967,9 @@ deleteRcvFile' :: DB.Connection -> DBRcvFileId -> IO () deleteRcvFile' db rcvFileId = DB.execute db "DELETE FROM rcv_files WHERE rcv_file_id = ?" (Only rcvFileId) -getNextRcvChunkToDownload :: DB.Connection -> XFTPServer -> IO (Maybe RcvFileChunk) -getNextRcvChunkToDownload db server@ProtocolServer {host, port, keyHash} = do +getNextRcvChunkToDownload :: DB.Connection -> XFTPServer -> NominalDiffTime -> IO (Maybe RcvFileChunk) +getNextRcvChunkToDownload db server@ProtocolServer {host, port, keyHash} ttl = do + cutoffTs <- addUTCTime (- ttl) <$> getCurrentTime maybeFirstRow toChunk $ DB.query db @@ -1981,11 +1983,11 @@ getNextRcvChunkToDownload db server@ProtocolServer {host, port, keyHash} = do JOIN rcv_files f ON f.rcv_file_id = c.rcv_file_id WHERE s.xftp_host = ? AND s.xftp_port = ? AND s.xftp_key_hash = ? AND r.received = 0 AND r.replica_number = 1 - AND f.status = ? AND f.deleted = 0 + AND f.status = ? AND f.deleted = 0 AND f.created_at >= ? ORDER BY r.created_at ASC LIMIT 1 |] - (host, port, keyHash, RFSReceiving) + (host, port, keyHash, RFSReceiving, cutoffTs) where toChunk :: ((DBRcvFileId, RcvFileId, UserId, Int64, Int, FileSize Word32, FileDigest, FilePath, Maybe FilePath) :. (Int64, ChunkReplicaId, C.APrivateSignKey, Bool, Maybe Int, Int)) -> RcvFileChunk toChunk ((rcvFileId, rcvFileEntityId, userId, rcvChunkId, chunkNo, chunkSize, digest, fileTmpPath, chunkTmpPath) :. (rcvChunkReplicaId, replicaId, replicaKey, received, delay, retries)) = @@ -2002,8 +2004,9 @@ getNextRcvChunkToDownload db server@ProtocolServer {host, port, keyHash} = do replicas = [RcvFileChunkReplica {rcvChunkReplicaId, server, replicaId, replicaKey, received, delay, retries}] } -getNextRcvFileToDecrypt :: DB.Connection -> IO (Maybe RcvFile) -getNextRcvFileToDecrypt db = do +getNextRcvFileToDecrypt :: DB.Connection -> NominalDiffTime -> IO (Maybe RcvFile) +getNextRcvFileToDecrypt db ttl = do + cutoffTs <- addUTCTime (- ttl) <$> getCurrentTime fileId_ :: Maybe DBRcvFileId <- maybeFirstRow fromOnly $ DB.query @@ -2011,16 +2014,17 @@ getNextRcvFileToDecrypt db = do [sql| SELECT rcv_file_id FROM rcv_files - WHERE status IN (?,?) AND deleted = 0 + WHERE status IN (?,?) AND deleted = 0 AND created_at >= ? ORDER BY created_at ASC LIMIT 1 |] - (RFSReceived, RFSDecrypting) + (RFSReceived, RFSDecrypting, cutoffTs) case fileId_ of Nothing -> pure Nothing Just fileId -> eitherToMaybe <$> getRcvFile db fileId -getPendingRcvFilesServers :: DB.Connection -> IO [XFTPServer] -getPendingRcvFilesServers db = do +getPendingRcvFilesServers :: DB.Connection -> NominalDiffTime -> IO [XFTPServer] +getPendingRcvFilesServers db ttl = do + cutoffTs <- addUTCTime (- ttl) <$> getCurrentTime map toServer <$> DB.query db @@ -2032,9 +2036,9 @@ getPendingRcvFilesServers db = do JOIN rcv_file_chunks c ON c.rcv_file_chunk_id = r.rcv_file_chunk_id JOIN rcv_files f ON f.rcv_file_id = c.rcv_file_id WHERE r.received = 0 AND r.replica_number = 1 - AND f.status = ? AND f.deleted = 0 + AND f.status = ? AND f.deleted = 0 AND f.created_at >= ? |] - (Only RFSReceiving) + (RFSReceiving, cutoffTs) where toServer :: (NonEmpty TransportHost, ServiceName, C.KeyHash) -> XFTPServer toServer (host, port, keyHash) = XFTPServer host port keyHash @@ -2059,3 +2063,15 @@ getCleanupRcvFilesDeleted db = FROM rcv_files WHERE deleted = 1 |] + +getRcvFilesExpired :: DB.Connection -> NominalDiffTime -> IO [(DBRcvFileId, RcvFileId, FilePath)] +getRcvFilesExpired db ttl = do + cutoffTs <- addUTCTime (- ttl) <$> getCurrentTime + DB.query + db + [sql| + SELECT rcv_file_id, rcv_file_entity_id, prefix_path + FROM rcv_files + WHERE created_at < ? + |] + (Only cutoffTs)