haeaeae/duplicati
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #6457 from carljohnsen/bugfix/rsync-reinstantiate-backend

Browse Source 
Bugfix/Remote synchronization tool reinstantiate backend
This commit is contained in:
Carl-Johannes Johnsen
2025-08-19 15:02:06 +02:00
committed by Kenneth Skovhede
parent d74df06796
commit d74fc3fd05
3 changed files with 576 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Tools/RemoteSynchronization/Program.cs
+102 -182
View File
@@ -49,6 +49,10 @@ namespace RemoteSynchronization
string Dst,
// Options
bool AutoCreateFolders,
int BackendRetries,
int BackendRetryDelay,
bool BackendRetryWithExponentialBackoff,
bool Confirm,
bool DryRun,
List<string> DstOptions,
@@ -107,6 +111,10 @@ destination will be verified before being overwritten (if they seemingly match).
arg_src,
arg_dst,
new Option<bool>(aliases: ["--auto-create-folders"], description: "Automatically create folders in the destination backend if they do not exist", getDefaultValue: () => true),
new Option<int>(aliases: ["--backend-retries"], description: "Number of times to recreate a backend on backend errors", getDefaultValue: () => 3) { Arity = ArgumentArity.ExactlyOne },
new Option<int>(aliases: ["--backend-retry-delay"], description: "Delay in milliseconds between backend retries", getDefaultValue: () => 1000) { Arity = ArgumentArity.ExactlyOne },
new Option<bool>(aliases: ["--backend-retry-with-exponential-backoff"], description: "Use exponential backoff for backend retries, multiplying the delay by two for each failure.", getDefaultValue: () => true),
new Option<bool>(aliases: ["--confirm", "--yes", "-y"], description: "Automatically confirm the operation", getDefaultValue: () => false),
new Option<bool>(aliases: ["--dry-run", "-d"], description: "Do not actually write or delete files. If not set here, the global options will be checked", getDefaultValue: () => false),
OptionWithMultipleTokens(aliases: ["--dst-options"], description: "Options for the destination backend. Each option is a key-value pair separated by an equals sign, e.g. --dst-options key1=value1 key2=value2 [default: empty]", getDefaultValue: () => []),
@@ -123,24 +131,23 @@ destination will be verified before being overwritten (if they seemingly match).
new Option<bool>(aliases: ["--verify-get-after-put"], description: "Verify the files after uploading them to ensure that they were uploaded correctly", getDefaultValue: () => false),
};
root_cmd.Handler = CommandHandler.Create((string backend_src, string backend_dst, Config config) =>
root_cmd.Handler = CommandHandler.Create((string backend_src, string backend_dst, Config config, CancellationToken token) =>
{
var config_with_args = config with { Dst = backend_dst, Src = backend_src };
return Run(config_with_args);
return Run(config_with_args, token);
});
return await root_cmd.InvokeAsync(args);
return await root_cmd.InvokeAsync(args).ConfigureAwait(false);
}
/// <summary>
/// The main logic of the tool.
/// </summary>
/// <param name="src">The connection string for the source backend.</param>
/// <param name="dst">The connection string for the destination backend.</param>
/// <param name="options">Various options for the tool</param>
/// <param name="config">The parsed configuration for the tool.</param>
/// <param name="token">The cancellation token to use for the asynchronous operations.</param>
/// <returns>The return code for the main entry; 0 on success.</returns>
private static async Task<int> Run(Config config)
private static async Task<int> Run(Config config, CancellationToken token)
{
// Unpack and parse the multi token options
var global_options = ParseOptions(config.GlobalOptions);
@@ -182,27 +189,15 @@ destination will be verified before being overwritten (if they seemingly match).
// Check if we only had to parse the arguments
if (config.ParseArgumentsOnly)
{
Duplicati.Library.Logging.Log.WriteInformationMessage(LOGTAG, "rsync", "Arguments parsed successfully; exiting.");
Duplicati.Library.Logging.Log.WriteInformationMessage(LOGTAG, "rsync", "Arguments parsed successfully; {0}; exiting.", config);
return 0;
}
// Load the backends
static Exception throw_message(string target)
{
var message = $"The {target} backend does not support streaming operations.";
var ex = new Exception(message);
Duplicati.Library.Logging.Log.WriteErrorMessage(LOGTAG, "rsync", ex, message);
return ex;
}
using var b1 = Duplicati.Library.DynamicLoader.BackendLoader.GetBackend(config.Src, src_opts);
var b1s = b1 as IStreamingBackend ?? throw throw_message("source");
using var b2 = Duplicati.Library.DynamicLoader.BackendLoader.GetBackend(config.Dst, dst_opts);
var b2s = b2 as IStreamingBackend ?? throw throw_message("destination");
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);
// Prepare the operations
var (to_copy, to_delete, to_verify) = await PrepareFileLists(b1s, b2s, config, CancellationToken.None);
var (to_copy, to_delete, to_verify) = await PrepareFileLists(b1m, b2m, config, token).ConfigureAwait(false);
// Verify the files if requested. If the files are not verified, they will be deleted and copied again.
long verified = 0, failed_verify = 0;
@@ -220,7 +215,7 @@ destination will be verified before being overwritten (if they seemingly match).
}
}
var not_verified = await VerifyAsync(b1s, b2s, to_verify, config);
var not_verified = await VerifyAsync(b1m, b2m, to_verify, config, token).ConfigureAwait(false);
failed_verify = not_verified.Count();
verified = to_verify.Count() - failed_verify;
@@ -237,7 +232,7 @@ destination will be verified before being overwritten (if they seemingly match).
Duplicati.Library.Logging.Log.WriteInformationMessage(LOGTAG, "rsync",
"The remote synchronization plan is to {0} {1} files from {2}, then copy {3} files from {4} to {2}.",
config.Retention ? "rename" : "delete",
to_delete.Count(), b2s.DisplayName, to_copy.Count(), b1s.DisplayName);
to_delete.Count(), b2m.DisplayName, to_copy.Count(), b1m.DisplayName);
// As this is a potentially destructive operation, ask for confirmation
if (!config.Confirm)
@@ -261,21 +256,21 @@ destination will be verified before being overwritten (if they seemingly match).
long renamed = 0, deleted = 0;
if (config.Retention)
{
renamed = await RenameAsync(b2, to_delete, config);
renamed = await RenameAsync(b2m, to_delete, config, token).ConfigureAwait(false);
Duplicati.Library.Logging.Log.WriteInformationMessage(LOGTAG, "rsync",
"Renamed {0} files in {1}", renamed, b2s.DisplayName);
"Renamed {0} files in {1}", renamed, b2m.DisplayName);
}
else
{
deleted = await DeleteAsync(b2s, to_delete, config);
deleted = await DeleteAsync(b2m, to_delete, config, token).ConfigureAwait(false);
Duplicati.Library.Logging.Log.WriteInformationMessage(LOGTAG, "rsync",
"Deleted {0} files from {1}", deleted, b2s.DisplayName);
"Deleted {0} files from {1}", deleted, b2m.DisplayName);
}
// Copy the files
var (copied, copy_errors) = await CopyAsync(b1s, b2s, to_copy, config);
var (copied, copy_errors) = await CopyAsync(b1m, b2m, to_copy, config, token).ConfigureAwait(false);
Duplicati.Library.Logging.Log.WriteInformationMessage(LOGTAG, "rsync",
"Copied {0} files from {1} to {2}", copied, b1s, b2s);
"Copied {0} files from {1} to {2}", copied, b1m.DisplayName, b2m.DisplayName);
// If there are still errors, retry a few times
if (copy_errors.Any())
@@ -289,10 +284,10 @@ destination will be verified before being overwritten (if they seemingly match).
config.Retry, copy_errors.Count());
for (int i = 0; i < config.Retry; i++)
{
await Task.Delay(5000); // Wait 5 seconds before retrying
(copied, copy_errors) = await CopyAsync(b1s, b2s, copy_errors, config);
await Task.Delay(5000).ConfigureAwait(false); // Wait 5 seconds before retrying
(copied, copy_errors) = await CopyAsync(b1m, b2m, copy_errors, config, token).ConfigureAwait(false);
Duplicati.Library.Logging.Log.WriteInformationMessage(LOGTAG, "rsync",
"Copied {0} files from {1} to {2}", copied, b1s, b2s);
"Copied {0} files from {1} to {2}", copied, b1m.DisplayName, b2m.DisplayName);
if (!copy_errors.Any())
break;
}
@@ -310,20 +305,20 @@ destination will be verified before being overwritten (if they seemingly match).
if (verified > 0)
Duplicati.Library.Logging.Log.WriteInformationMessage(LOGTAG, "rsync",
"Verified {0} files in {1} that didn't need to be copied",
verified, b2s.DisplayName);
verified, b2m.DisplayName);
if (failed_verify > 0)
Duplicati.Library.Logging.Log.WriteInformationMessage(LOGTAG, "rsync",
"Failed to verify {0} files in {1}, which were then attempted to be copied",
failed_verify, b2s.DisplayName);
failed_verify, b2m.DisplayName);
if (copied > 0)
Duplicati.Library.Logging.Log.WriteInformationMessage(LOGTAG, "rsync",
"Copied {0} files from {1} to {2}", copied, b1s.DisplayName, b2s.DisplayName);
"Copied {0} files from {1} to {2}", copied, b1m.DisplayName, b2m.DisplayName);
if (deleted > 0)
Duplicati.Library.Logging.Log.WriteInformationMessage(LOGTAG, "rsync",
"Deleted {0} files from {1}", deleted, b2s.DisplayName);
"Deleted {0} files from {1}", deleted, b2m.DisplayName);
if (renamed > 0)
Duplicati.Library.Logging.Log.WriteInformationMessage(LOGTAG, "rsync",
"Renamed {0} files in {1}", renamed, b2s.DisplayName);
"Renamed {0} files in {1}", renamed, b2m.DisplayName);
Duplicati.Library.Logging.Log.WriteInformationMessage(LOGTAG, "rsync",
"Remote synchronization completed successfully");
@@ -345,8 +340,9 @@ destination will be verified before being overwritten (if they seemingly match).
/// <param name="b_dst">The destination backend.</param>
/// <param name="files">The files that will be copied.</param>
/// <param name="config">The parsed configuration for the tool.</param>
/// <param name="token">The cancellation token to use for the asynchronous operations.</param>
/// <returns>A tuple holding the number of succesful copies and a List of the files that failed.</returns>
private static async Task<(long, IEnumerable<IFileEntry>)> CopyAsync(IStreamingBackend b_src, IStreamingBackend b_dst, IEnumerable<IFileEntry> files, Config config)
private static async Task<(long, IEnumerable<IFileEntry>)> CopyAsync(LightWeightBackendManager b_src, LightWeightBackendManager b_dst, IEnumerable<IFileEntry> files, Config config, CancellationToken token)
{
long successful_copies = 0;
List<IFileEntry> errors = [];
@@ -369,7 +365,7 @@ destination will be verified before being overwritten (if they seemingly match).
try
{
sw_get_src.Start();
await b_src.GetAsync(f.Name, s_src, CancellationToken.None);
await b_src.GetAsync(f.Name, s_src, token).ConfigureAwait(false);
s_src.Position = 0;
sw_get_src.Stop();
if (config.DryRun)
@@ -382,7 +378,7 @@ destination will be verified before being overwritten (if they seemingly match).
else
{
sw_put_dst.Start();
await b_dst.PutAsync(f.Name, s_src, CancellationToken.None);
await b_dst.PutAsync(f.Name, s_src, token).ConfigureAwait(false);
s_src.Position = 0;
sw_put_dst.Stop();
if (config.VerifyGetAfterPut)
@@ -397,7 +393,7 @@ destination will be verified before being overwritten (if they seemingly match).
using var s_dst = Duplicati.Library.Utility.TempFileStream.Create();
sw_get_dst.Start();
await b_dst.GetAsync(f.Name, s_dst, CancellationToken.None);
await b_dst.GetAsync(f.Name, s_dst, token).ConfigureAwait(false);
s_dst.Position = 0;
sw_get_dst.Stop();
@@ -411,7 +407,7 @@ destination will be verified before being overwritten (if they seemingly match).
using var hasher = HashFactory.CreateHasher("SHA256");
var dsthash = Convert.ToBase64String(hasher.ComputeHash(s_dst));
if (await srchashtask != dsthash)
if (await srchashtask.ConfigureAwait(false) != dsthash)
{
err_string = (err_string is null ? "" : err_string + " ") + "The contents of the files do not match.";
}
@@ -463,8 +459,9 @@ destination will be verified before being overwritten (if they seemingly match).
/// <param name="b">The backend to delete the files from.</param>
/// <param name="files">The files to delete.</param>
/// <param name="config">The parsed configuration for the tool.</param>
/// <param name="token">The cancellation token to use for the asynchronous operations.</param>
/// <returns>The number of successful deletions.</returns>
private static async Task<long> DeleteAsync(IStreamingBackend b, IEnumerable<IFileEntry> files, Config config)
private static async Task<long> DeleteAsync(LightWeightBackendManager b, IEnumerable<IFileEntry> files, Config config, CancellationToken token)
{
long successful_deletes = 0;
long i = 0, n = files.Count();
@@ -489,7 +486,7 @@ destination will be verified before being overwritten (if they seemingly match).
}
else
{
await b.DeleteAsync(f.Name, CancellationToken.None);
await b.DeleteAsync(f.Name, token).ConfigureAwait(false);
}
successful_deletes++;
}
@@ -556,19 +553,20 @@ destination will be verified before being overwritten (if they seemingly match).
/// The files to delete are the files that are found in the destination but not found in the source.
/// The files to verify are the files that are found in both the source and the destination, and that have the same size and modification date.
/// </summary>
/// <param name="b_src">The source backend.</param>
/// <param name="b_dst">The destination backend.</param>
/// <param name="b_src">The source lightweight backend manager.</param>
/// <param name="b_dst">The destination lightweight backend manager.</param>
/// <param name="config">The parsed configuration for the tool.</param>
/// <param name="token">The cancellation token to use for the asynchronous operations.</param>
/// <returns>A tuple of Lists each holding the files to copy, delete and verify.</returns>
private static async Task<(IEnumerable<IFileEntry>, IEnumerable<IFileEntry>, IEnumerable<IFileEntry>)> PrepareFileLists(IStreamingBackend b_src, IStreamingBackend b_dst, Config config, CancellationToken cancelToken)
private static async Task<(IEnumerable<IFileEntry>, IEnumerable<IFileEntry>, IEnumerable<IFileEntry>)> PrepareFileLists(LightWeightBackendManager b_src, LightWeightBackendManager b_dst, Config config, CancellationToken token)
{
IEnumerable<IFileEntry> files_src, files_dst;
using (new Duplicati.Library.Logging.Timer(LOGTAG, "rsync", "Prepare | List source"))
files_src = await b_src.ListAsync(cancelToken).ToListAsync(cancelToken).ConfigureAwait(false);
files_src = await b_src.ListAsync(token).ConfigureAwait(false);
using (new Duplicati.Library.Logging.Timer(LOGTAG, "rsync", "Prepare | List destination"))
files_dst = await b_dst.ListAsync(cancelToken).ToListAsync(cancelToken).ConfigureAwait(false);
files_dst = await b_dst.ListAsync(token).ConfigureAwait(false);
// Shortcut for force
if (config.Force)
@@ -642,158 +640,80 @@ destination will be verified before being overwritten (if they seemingly match).
/// Renames the files in a backend.
/// The renaming is done by deleting the file and re-uploading it with a new name.
/// </summary>
/// <param name="b">The backend to rename the files in.</param>
/// <param name="bm">The lightweight backend manager to issue rename operations to.</param>
/// <param name="files">The files to rename.</param>
/// <param name="config">The parsed configuration for the tool.</param>
/// <param name="token">The cancellation token to use for the asynchronous operations.</param>
/// <returns>The number of successful renames.</returns>
private static async Task<long> RenameAsync(IBackend b, IEnumerable<IFileEntry> files, Config config)
private static async Task<long> RenameAsync(LightWeightBackendManager bm, IEnumerable<IFileEntry> files, Config config, CancellationToken token)
{
long successful_renames = 0;
string prefix = $"{System.DateTime.UtcNow:yyyyMMddHHmmss}.old";
using var downloaded = new MemoryStream();
long i = 0, n = files.Count();
switch (b)
var sw = new System.Diagnostics.Stopwatch();
foreach (var f in files)
{
case IStreamingBackend sb:
if (config.Progress)
Console.Write($"\rRenaming: {i}/{n}");
Duplicati.Library.Logging.Log.WriteVerboseMessage(LOGTAG, "rsync",
"Renaming {0} to {1}.{0} by calling Rename on {2}",
f.Name, prefix, bm.DisplayName);
try
{
if (config.DryRun)
{
var sw_get_src = new System.Diagnostics.Stopwatch();
var sw_put_dst = new System.Diagnostics.Stopwatch();
var sw_del_src = new System.Diagnostics.Stopwatch();
foreach (var f in files)
{
if (config.Progress)
Console.Write($"\rRenaming: {i}/{n}");
Duplicati.Library.Logging.Log.WriteVerboseMessage(LOGTAG, "rsync",
"Renaming {0} to {1}.{0} by deleting and re-uploading {2} bytes to {3}",
f.Name, prefix,
Duplicati.Library.Utility.Utility.FormatSizeString(downloaded.Length),
sb.DisplayName);
try
{
sw_get_src.Start();
await sb.GetAsync(f.Name, downloaded, CancellationToken.None);
sw_get_src.Stop();
if (config.DryRun)
{
Duplicati.Library.Logging.Log.WriteDryrunMessage(LOGTAG, "rsync",
"Would rename {0} to {1}.{0} by deleting and re-uploading {2} bytes to {3}",
f.Name, prefix,
Duplicati.Library.Utility.Utility.FormatSizeString(downloaded.Length),
sb.DisplayName);
}
else
{
sw_put_dst.Start();
await sb.PutAsync($"{prefix}.{f.Name}", downloaded, CancellationToken.None);
sw_put_dst.Stop();
sw_del_src.Start();
await sb.DeleteAsync(f.Name, CancellationToken.None);
sw_del_src.Stop();
}
successful_renames++;
}
catch (Exception e)
{
Duplicati.Library.Logging.Log.WriteErrorMessage(LOGTAG, "rsync", e,
"Error renaming {0}: {1}", f.Name, e.Message);
}
finally
{
// Reset the stream
downloaded.SetLength(0);
// Stop any running timers
sw_get_src.Stop();
sw_put_dst.Stop();
sw_del_src.Stop();
}
i++;
}
Duplicati.Library.Logging.Log.WriteProfilingMessage(LOGTAG, "rsync",
"Rename | Get source: {0} ms, Put destination: {1} ms, Delete source: {2} ms",
TimeSpan.FromMilliseconds(sw_get_src.ElapsedMilliseconds),
TimeSpan.FromMilliseconds(sw_put_dst.ElapsedMilliseconds),
TimeSpan.FromMilliseconds(sw_del_src.ElapsedMilliseconds));
if (config.Progress)
Console.WriteLine($"\rRenaming: {n}/{n}");
return successful_renames;
Duplicati.Library.Logging.Log.WriteDryrunMessage(LOGTAG, "rsync",
"Would rename {0} to {1}.{0} by calling Rename on {2}",
f.Name, prefix, bm.DisplayName);
}
case IRenameEnabledBackend rb:
else
{
var sw = new System.Diagnostics.Stopwatch();
foreach (var f in files)
{
if (config.Progress)
Console.Write($"\rRenaming: {i}/{n}");
Duplicati.Library.Logging.Log.WriteVerboseMessage(LOGTAG, "rsync",
"Renaming {0} to {1}.{0} by calling Rename on {2}",
f.Name, prefix, rb.DisplayName);
try
{
if (config.DryRun)
{
Duplicati.Library.Logging.Log.WriteDryrunMessage(LOGTAG, "rsync",
"Would rename {0} to {1}.{0} by calling Rename on {2}",
f.Name, prefix, rb.DisplayName);
}
else
{
sw.Start();
await rb.RenameAsync(f.Name, $"{prefix}.{f.Name}", CancellationToken.None);
sw.Stop();
}
successful_renames++;
}
catch (Exception e)
{
Duplicati.Library.Logging.Log.WriteErrorMessage(LOGTAG, "rsync", e,
"Error renaming {0}: {1}", f.Name, e.Message);
}
finally
{
// Ensure the timer is stopped
sw.Stop();
}
i++;
}
Duplicati.Library.Logging.Log.WriteProfilingMessage(LOGTAG, "rsync",
"Rename: {0} ms",
TimeSpan.FromMilliseconds(sw.ElapsedMilliseconds));
if (config.Progress)
Console.WriteLine($"\rRenaming: {n}/{n}");
return successful_renames;
sw.Start();
await bm.RenameAsync(f.Name, $"{prefix}.{f.Name}", token).ConfigureAwait(false);
sw.Stop();
}
default:
throw new NotSupportedException("The backend does not support renaming");
successful_renames++;
}
catch (Exception e)
{
Duplicati.Library.Logging.Log.WriteErrorMessage(LOGTAG, "rsync", e,
"Error renaming {0}: {1}", f.Name, e.Message);
}
finally
{
// Ensure the timer is stopped
sw.Stop();
}
i++;
}
Duplicati.Library.Logging.Log.WriteProfilingMessage(LOGTAG, "rsync",
"Rename: {0} ms",
TimeSpan.FromMilliseconds(sw.ElapsedMilliseconds));
if (config.Progress)
Console.WriteLine($"\rRenaming: {n}/{n}");
return successful_renames;
}
/// <summary>
/// Verifies the files in the destination backend.
/// The verification is done by downloading the files from the destination backend and comparing them to the source files.
/// </summary>
/// <param name="b_src">The source backend.</param>
/// <param name="b_dst">The destination backend.</param>
/// <param name="b_src">The source lightweight backend manager.</param>
/// <param name="b_dst">The destination lightweight backend manager.</param>
/// <param name="files">The files to verify.</param>
/// <param name="config">The parsed configuration for the tool.</param>
/// <param name="token">The cancellation token to use for the asynchronous operations.</param>
/// <returns>A list of the files that failed verification.</returns>
private static async Task<IEnumerable<IFileEntry>> VerifyAsync(IStreamingBackend b_src, IStreamingBackend b_dst, IEnumerable<IFileEntry> files, Config config)
private static async Task<IEnumerable<IFileEntry>> VerifyAsync(LightWeightBackendManager b_src, LightWeightBackendManager b_dst, IEnumerable<IFileEntry> files, Config config, CancellationToken token)
{
var errors = new List<IFileEntry>();
using var s_src = new MemoryStream();
@@ -817,9 +737,9 @@ destination will be verified before being overwritten (if they seemingly match).
{
// Get both files
sw_get.Start();
var fs = b_src.GetAsync(f.Name, s_src, CancellationToken.None);
var ds = b_dst.GetAsync(f.Name, s_dst, CancellationToken.None);
await Task.WhenAll(fs, ds);
var fs = b_src.GetAsync(f.Name, s_src, token);
var ds = b_dst.GetAsync(f.Name, s_dst, token);
await Task.WhenAll(fs, ds).ConfigureAwait(false);
sw_get.Stop();
// Compare the contents