Compare commits

...
Author SHA1 Message Date
spaced4ndy 85ef413e2e fix 2023-04-21 19:28:33 +04:00
spaced4ndy d6df82f7cc xftp: safer retries 2023-04-21 19:21:32 +04:00
2 changed files with 26 additions and 9 deletions
+12 -6
View File
@@ -197,7 +197,9 @@ runXFTPRcvWorker c srv doWork = do
withStore' c $ \db -> updateRcvChunkReplicaDelay db rcvChunkReplicaId replicaDelay withStore' c $ \db -> updateRcvChunkReplicaDelay db rcvChunkReplicaId replicaDelay
atomically $ assertAgentForeground c atomically $ assertAgentForeground c
loop loop
retryDone e = rcvWorkerInternalError c rcvFileId rcvFileEntityId (Just fileTmpPath) (show e) retryDone e = do
withStore' c (`deleteRcvFileReplica` rcvChunkReplicaId)
rcvWorkerInternalError c rcvFileId rcvFileEntityId (Just fileTmpPath) (show e)
downloadFileChunk :: RcvFileChunk -> RcvFileChunkReplica -> m () downloadFileChunk :: RcvFileChunk -> RcvFileChunkReplica -> m ()
downloadFileChunk RcvFileChunk {userId, rcvFileId, rcvFileEntityId, rcvChunkId, chunkNo, chunkSize, digest, fileTmpPath} replica = do downloadFileChunk RcvFileChunk {userId, rcvFileId, rcvFileEntityId, rcvChunkId, chunkNo, chunkSize, digest, fileTmpPath} replica = do
fsFileTmpPath <- toFSFilePath fileTmpPath fsFileTmpPath <- toFSFilePath fileTmpPath
@@ -415,7 +417,7 @@ runXFTPSndPrepareWorker c doWork = do
createChunk :: Int -> SndFileChunk -> m () createChunk :: Int -> SndFileChunk -> m ()
createChunk numRecipients' ch = do createChunk numRecipients' ch = do
atomically $ assertAgentForeground c atomically $ assertAgentForeground c
(replica, ProtoServerWithAuth srv _) <- agentOperationBracket c AOSndNetwork throwWhenInactive tryCreate (replica, ProtoServerWithAuth srv _) <- tryCreate
withStore' c $ \db -> createSndFileReplica db ch replica withStore' c $ \db -> createSndFileReplica db ch replica
addXFTPSndWorker c $ Just srv addXFTPSndWorker c $ Just srv
where where
@@ -445,7 +447,7 @@ runXFTPSndWorker c srv doWork = do
forever $ do forever $ do
void . atomically $ readTMVar doWork void . atomically $ readTMVar doWork
atomically $ assertAgentForeground c atomically $ assertAgentForeground c
agentOperationBracket c AOSndNetwork throwWhenInactive runXFTPOperation runXFTPOperation
where where
noWorkToDo = void . atomically $ tryTakeTMVar doWork noWorkToDo = void . atomically $ tryTakeTMVar doWork
runXFTPOperation :: m () runXFTPOperation :: m ()
@@ -470,7 +472,9 @@ runXFTPSndWorker c srv doWork = do
withStore' c $ \db -> updateSndChunkReplicaDelay db sndChunkReplicaId replicaDelay withStore' c $ \db -> updateSndChunkReplicaDelay db sndChunkReplicaId replicaDelay
atomically $ assertAgentForeground c atomically $ assertAgentForeground c
loop loop
retryDone e = sndWorkerInternalError c sndFileId sndFileEntityId (Just filePrefixPath) (show e) retryDone e = do
withStore' c (`deleteSndFileReplica` sndChunkReplicaId)
sndWorkerInternalError c sndFileId sndFileEntityId (Just filePrefixPath) (show e)
uploadFileChunk :: SndFileChunk -> SndFileChunkReplica -> m () uploadFileChunk :: SndFileChunk -> SndFileChunkReplica -> m ()
uploadFileChunk sndFileChunk@SndFileChunk {sndFileId, userId, chunkSpec = chunkSpec@XFTPChunkSpec {filePath}, digest = chunkDigest} replica = do uploadFileChunk sndFileChunk@SndFileChunk {sndFileId, userId, chunkSpec = chunkSpec@XFTPChunkSpec {filePath}, digest = chunkDigest} replica = do
replica'@SndFileChunkReplica {sndChunkReplicaId} <- addRecipients sndFileChunk replica replica'@SndFileChunkReplica {sndChunkReplicaId} <- addRecipients sndFileChunk replica
@@ -628,11 +632,13 @@ runXFTPDelWorker c srv doWork = do
withStore' c $ \db -> updateDeletedSndChunkReplicaDelay db deletedSndChunkReplicaId replicaDelay withStore' c $ \db -> updateDeletedSndChunkReplicaDelay db deletedSndChunkReplicaId replicaDelay
atomically $ assertAgentForeground c atomically $ assertAgentForeground c
loop loop
retryDone e = delWorkerInternalError c deletedSndChunkReplicaId e retryDone e = do
withStore' c (`deleteDeletedSndChunkReplica` deletedSndChunkReplicaId)
delWorkerInternalError c deletedSndChunkReplicaId e
deleteChunkReplica :: DeletedSndChunkReplica -> m () deleteChunkReplica :: DeletedSndChunkReplica -> m ()
deleteChunkReplica replica@DeletedSndChunkReplica {userId, deletedSndChunkReplicaId} = do deleteChunkReplica replica@DeletedSndChunkReplica {userId, deletedSndChunkReplicaId} = do
agentXFTPDeleteChunk c userId replica agentXFTPDeleteChunk c userId replica
withStore' c $ \db -> deleteDeletedSndChunkReplica db deletedSndChunkReplicaId withStore' c (`deleteDeletedSndChunkReplica` deletedSndChunkReplicaId)
delWorkerInternalError :: AgentMonad m => AgentClient -> Int64 -> AgentErrorType -> m () delWorkerInternalError :: AgentMonad m => AgentClient -> Int64 -> AgentErrorType -> m ()
delWorkerInternalError c deletedSndChunkReplicaId e = do delWorkerInternalError c deletedSndChunkReplicaId e = do
+14 -3
View File
@@ -144,6 +144,7 @@ module Simplex.Messaging.Agent.Store.SQLite
updateRcvFileNoTmpPath, updateRcvFileNoTmpPath,
updateRcvFileDeleted, updateRcvFileDeleted,
deleteRcvFile', deleteRcvFile',
deleteRcvFileReplica,
getNextRcvChunkToDownload, getNextRcvChunkToDownload,
getNextRcvFileToDecrypt, getNextRcvFileToDecrypt,
getPendingRcvFilesServers, getPendingRcvFilesServers,
@@ -162,6 +163,7 @@ module Simplex.Messaging.Agent.Store.SQLite
updateSndFileNoPrefixPath, updateSndFileNoPrefixPath,
updateSndFileDeleted, updateSndFileDeleted,
deleteSndFile', deleteSndFile',
deleteSndFileReplica,
getSndFileDeleted, getSndFileDeleted,
createSndFileReplica, createSndFileReplica,
getNextSndChunkToUpload, getNextSndChunkToUpload,
@@ -1998,6 +2000,10 @@ deleteRcvFile' :: DB.Connection -> DBRcvFileId -> IO ()
deleteRcvFile' db rcvFileId = deleteRcvFile' db rcvFileId =
DB.execute db "DELETE FROM rcv_files WHERE rcv_file_id = ?" (Only rcvFileId) DB.execute db "DELETE FROM rcv_files WHERE rcv_file_id = ?" (Only rcvFileId)
deleteRcvFileReplica :: DB.Connection -> Int64 -> IO ()
deleteRcvFileReplica db replicaId =
DB.execute db "DELETE FROM rcv_file_chunk_replicas WHERE rcv_file_chunk_replica_id = ?" (Only replicaId)
getNextRcvChunkToDownload :: DB.Connection -> XFTPServer -> NominalDiffTime -> IO (Maybe RcvFileChunk) getNextRcvChunkToDownload :: DB.Connection -> XFTPServer -> NominalDiffTime -> IO (Maybe RcvFileChunk)
getNextRcvChunkToDownload db server@ProtocolServer {host, port, keyHash} ttl = do getNextRcvChunkToDownload db server@ProtocolServer {host, port, keyHash} ttl = do
cutoffTs <- addUTCTime (- ttl) <$> getCurrentTime cutoffTs <- addUTCTime (- ttl) <$> getCurrentTime
@@ -2015,7 +2021,7 @@ getNextRcvChunkToDownload db server@ProtocolServer {host, port, keyHash} ttl = d
WHERE s.xftp_host = ? AND s.xftp_port = ? AND s.xftp_key_hash = ? WHERE s.xftp_host = ? AND s.xftp_port = ? AND s.xftp_key_hash = ?
AND r.received = 0 AND r.replica_number = 1 AND r.received = 0 AND r.replica_number = 1
AND f.status = ? AND f.deleted = 0 AND f.created_at >= ? AND f.status = ? AND f.deleted = 0 AND f.created_at >= ?
ORDER BY r.created_at ASC ORDER BY r.retries ASC, r.created_at ASC
LIMIT 1 LIMIT 1
|] |]
(host, port, keyHash, RFSReceiving, cutoffTs) (host, port, keyHash, RFSReceiving, cutoffTs)
@@ -2255,6 +2261,10 @@ deleteSndFile' :: DB.Connection -> DBSndFileId -> IO ()
deleteSndFile' db sndFileId = deleteSndFile' db sndFileId =
DB.execute db "DELETE FROM snd_files WHERE snd_file_id = ?" (Only sndFileId) DB.execute db "DELETE FROM snd_files WHERE snd_file_id = ?" (Only sndFileId)
deleteSndFileReplica :: DB.Connection -> Int64 -> IO ()
deleteSndFileReplica db replicaId =
DB.execute db "DELETE FROM snd_file_chunk_replicas WHERE snd_file_chunk_replica_id = ?" (Only replicaId)
getSndFileDeleted :: DB.Connection -> DBSndFileId -> IO Bool getSndFileDeleted :: DB.Connection -> DBSndFileId -> IO Bool
getSndFileDeleted db sndFileId = getSndFileDeleted db sndFileId =
fromMaybe True fromMaybe True
@@ -2301,7 +2311,7 @@ getNextSndChunkToUpload db server@ProtocolServer {host, port, keyHash} ttl = do
WHERE s.xftp_host = ? AND s.xftp_port = ? AND s.xftp_key_hash = ? WHERE s.xftp_host = ? AND s.xftp_port = ? AND s.xftp_key_hash = ?
AND r.replica_status = ? AND r.replica_number = 1 AND r.replica_status = ? AND r.replica_number = 1
AND (f.status = ? OR f.status = ?) AND f.deleted = 0 AND f.created_at >= ? AND (f.status = ? OR f.status = ?) AND f.deleted = 0 AND f.created_at >= ?
ORDER BY r.created_at ASC ORDER BY r.retries ASC, r.created_at ASC
LIMIT 1 LIMIT 1
|] |]
(host, port, keyHash, SFRSCreated, SFSEncrypted, SFSUploading, cutoffTs) (host, port, keyHash, SFRSCreated, SFSEncrypted, SFSUploading, cutoffTs)
@@ -2443,7 +2453,8 @@ getNextDeletedSndChunkReplica db ProtocolServer {host, port, keyHash} ttl = do
JOIN xftp_servers s ON s.xftp_server_id = r.xftp_server_id 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 = ? WHERE s.xftp_host = ? AND s.xftp_port = ? AND s.xftp_key_hash = ?
AND r.created_at >= ? AND r.created_at >= ?
ORDER BY r.created_at ASC LIMIT 1 ORDER BY r.retries ASC, r.created_at ASC
LIMIT 1
|] |]
(host, port, keyHash, cutoffTs) (host, port, keyHash, cutoffTs)
case replicaId_ of case replicaId_ of