Changed to tempfile rather than tempstream in remote synchronization as backends and wrapping streams close the underlying streams on dispose, leading to temporary data being deleted occassionally and on errors.
This commit is contained in:
@@ -26,7 +26,9 @@ namespace Duplicati.Library.Main.Backend
|
||||
/// <param name="retryDelay">The delay between retries, in milliseconds.</param>
|
||||
/// <param name="autoCreateFolders">Whether to automatically create folders if they do not exist.</param>
|
||||
/// <param name="retryWithExponentialBackoff">Whether to use exponential backoff for retries.</param>
|
||||
public class LightWeightBackendManager(string backendUrl, Dictionary<string, string> options, int maxRetries = 3, int retryDelay = 1000, bool autoCreateFolders = false, bool retryWithExponentialBackoff = false) : IDisposable
|
||||
/// <param name="progressUpdater">An optional progress updater for file operations.</param>
|
||||
/// <param name="backendProgressUpdater">An optional progress updater for backend operations.</param>
|
||||
public class LightWeightBackendManager(string backendUrl, Dictionary<string, string> options, int maxRetries = 3, int retryDelay = 1000, bool autoCreateFolders = false, bool retryWithExponentialBackoff = false, IOperationProgressUpdater? progressUpdater = null, IBackendProgressUpdater? backendProgressUpdater = null) : IDisposable
|
||||
{
|
||||
private static readonly string LOGTAG = Logging.Log.LogTagFromType<LightWeightBackendManager>();
|
||||
|
||||
@@ -115,6 +117,46 @@ namespace Duplicati.Library.Main.Backend
|
||||
);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gets a file from the remote backend and saves it to a temporary file.
|
||||
/// This method retries the operation with a delay if it fails.
|
||||
/// </summary>
|
||||
/// <param name="remotename">The name of the remote file to get.</param>
|
||||
/// <param name="token">A cancellation token to cancel the operation.</param>
|
||||
/// <returns>A task representing the asynchronous get operation. The task result is the path to the temporary file containing the downloaded data.</returns>
|
||||
public async Task<TempFile> GetAsync(string remotename, CancellationToken token)
|
||||
{
|
||||
var temp = new TempFile();
|
||||
|
||||
try
|
||||
{
|
||||
await RetryWithDelay(
|
||||
$"Get {remotename} to TempFile",
|
||||
async () =>
|
||||
{
|
||||
using var fileStream = System.IO.File.OpenWrite(temp);
|
||||
using var progressStream = new ProgressReportingStream(fileStream, pg =>
|
||||
{
|
||||
backendProgressUpdater?.UpdateProgress(remotename, pg);
|
||||
progressUpdater?.UpdateFileProgress(pg);
|
||||
});
|
||||
await _streamingBackend!.GetAsync(remotename, progressStream, token).ConfigureAwait(false);
|
||||
_anyDownloaded = true;
|
||||
},
|
||||
null,
|
||||
false,
|
||||
token
|
||||
).ConfigureAwait(false);
|
||||
|
||||
return temp;
|
||||
}
|
||||
catch
|
||||
{
|
||||
temp.Dispose();
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Instantiates the backend if it has not been instantiated yet.
|
||||
/// If the backend is already instantiated, it simply returns.
|
||||
@@ -187,6 +229,35 @@ namespace Duplicati.Library.Main.Backend
|
||||
);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Puts a file to the remote backend from a temporary file.
|
||||
/// This method retries the operation with a delay if it fails.
|
||||
/// </summary>
|
||||
/// <param name="remotename">The name of the remote file to put.</param>
|
||||
/// <param name="temp">The temporary file containing the data to upload.</param>
|
||||
/// <param name="token">A cancellation token to cancel the operation.</param>
|
||||
/// <returns>A task representing the asynchronous put operation.</returns>
|
||||
public Task PutAsync(string remotename, TempFile temp, CancellationToken token)
|
||||
{
|
||||
return RetryWithDelay(
|
||||
$"Put {remotename}",
|
||||
async () =>
|
||||
{
|
||||
using var fileStream = System.IO.File.OpenRead(temp);
|
||||
using var progressStream = new ProgressReportingStream(fileStream, pg =>
|
||||
{
|
||||
backendProgressUpdater?.UpdateProgress(remotename, pg);
|
||||
progressUpdater?.UpdateFileProgress(pg);
|
||||
});
|
||||
await _streamingBackend!.PutAsync(remotename, progressStream, token).ConfigureAwait(false);
|
||||
_anyUploaded = true;
|
||||
},
|
||||
null,
|
||||
false,
|
||||
token
|
||||
);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Renames a file in the remote backend.
|
||||
/// If the backend supports renaming, it uses the RenameAsync method.
|
||||
@@ -237,6 +308,11 @@ namespace Duplicati.Library.Main.Backend
|
||||
};
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Resets the state of the backend manager.
|
||||
/// This method resets the download and upload flags and resets the retry delay to its initial value.
|
||||
/// </summary>
|
||||
/// <returns>A task representing the asynchronous reset operation.</returns>
|
||||
public async Task Reset()
|
||||
{
|
||||
_anyDownloaded = false;
|
||||
|
||||
@@ -203,8 +203,8 @@ namespace RemoteSynchronization
|
||||
return 0;
|
||||
}
|
||||
|
||||
using var b1m = new LightWeightBackendManager(config.Src, src_opts, config.BackendRetries, config.BackendRetryDelay, config.BackendRetryWithExponentialBackoff);
|
||||
using var b2m = new LightWeightBackendManager(config.Dst, dst_opts, config.BackendRetries, config.BackendRetryDelay, config.BackendRetryWithExponentialBackoff);
|
||||
using var b1m = new LightWeightBackendManager(config.Src, src_opts, config.BackendRetries, config.BackendRetryDelay, config.BackendRetryWithExponentialBackoff, progressUpdater: progressUpdater, backendProgressUpdater: backendProgressUpdater);
|
||||
using var b2m = new LightWeightBackendManager(config.Dst, dst_opts, config.BackendRetries, config.BackendRetryDelay, config.BackendRetryWithExponentialBackoff, progressUpdater: progressUpdater, backendProgressUpdater: backendProgressUpdater);
|
||||
|
||||
// Prepare the operations
|
||||
var (to_copy, to_delete, to_verify) = await PrepareFileLists(b1m, b2m, config, token).ConfigureAwait(false);
|
||||
@@ -388,31 +388,24 @@ namespace RemoteSynchronization
|
||||
|
||||
try
|
||||
{
|
||||
var s_src_length = 0L;
|
||||
using var s_src = Duplicati.Library.Utility.TempFileStream.Create();
|
||||
|
||||
// Report downloading current file with real-time progress
|
||||
var fileSize = Math.Max(f.Size, 0);
|
||||
progressUpdater?.StartFile($"Downloading {f.Name}", fileSize);
|
||||
backendProgressUpdater?.StartAction(BackendActionType.Get, f.Name, fileSize);
|
||||
using var s_src_download_progress = new ProgressReportingStream(s_src, pg =>
|
||||
{
|
||||
backendProgressUpdater?.UpdateProgress(f.Name, pg);
|
||||
progressUpdater?.UpdateFileProgress(pg);
|
||||
});
|
||||
|
||||
sw_get_src.Start();
|
||||
await b_src.GetAsync(f.Name, s_src_download_progress, token).ConfigureAwait(false);
|
||||
s_src.Position = 0;
|
||||
using var tmp_src = await b_src.GetAsync(f.Name, token).ConfigureAwait(false);
|
||||
sw_get_src.Stop();
|
||||
|
||||
backendProgressUpdater?.EndAction(BackendActionType.Get, f.Name);
|
||||
|
||||
var s_src_length = SystemIO.IO_OS.FileLength(tmp_src);
|
||||
|
||||
if (config.DryRun)
|
||||
{
|
||||
Duplicati.Library.Logging.Log.WriteDryrunMessage(LOGTAG, "rsync",
|
||||
"Would write {0} bytes of {1} to {2}",
|
||||
Duplicati.Library.Utility.Utility.FormatSizeString(s_src.Length),
|
||||
Duplicati.Library.Utility.Utility.FormatSizeString(s_src_length),
|
||||
f.Name, b_dst.DisplayName);
|
||||
}
|
||||
else
|
||||
@@ -422,23 +415,16 @@ namespace RemoteSynchronization
|
||||
if (config.VerifyGetAfterPut)
|
||||
{
|
||||
using var hasher = HashFactory.CreateHasher("SHA256");
|
||||
sourceHash = hasher.ComputeHash(s_src);
|
||||
using var s = SystemIO.IO_OS.FileOpenRead(tmp_src);
|
||||
sourceHash = hasher.ComputeHash(s);
|
||||
}
|
||||
|
||||
// Store the size BEFORE uploading, as we cannot access it after PutAsync
|
||||
s_src_length = s_src.Length;
|
||||
|
||||
// Report uploading current file with real-time progress
|
||||
progressUpdater?.StartFile($"Uploading {f.Name}", s_src_length);
|
||||
backendProgressUpdater?.StartAction(BackendActionType.Put, f.Name, s_src_length);
|
||||
using var s_src_upload_progress = new ProgressReportingStream(s_src, pg =>
|
||||
{
|
||||
backendProgressUpdater?.UpdateProgress(f.Name, pg);
|
||||
progressUpdater?.UpdateFileProgress(pg);
|
||||
});
|
||||
|
||||
sw_put_dst.Start();
|
||||
await b_dst.PutAsync(f.Name, s_src_upload_progress, token).ConfigureAwait(false);
|
||||
await b_dst.PutAsync(f.Name, tmp_src, token).ConfigureAwait(false);
|
||||
sw_put_dst.Stop();
|
||||
|
||||
backendProgressUpdater?.EndAction(BackendActionType.Put, f.Name);
|
||||
|
||||
Reference in New Issue
Block a user