From 167bd46a8a5876f38a91c00295d12ea2c99c051b Mon Sep 17 00:00:00 2001 From: Carl Johnsen Date: Tue, 5 Aug 2025 11:07:19 +0200 Subject: [PATCH] Added a lightweight backend manager to the remote synchronization tool --- .../LightWeightBackendManager.cs | 166 +++++++++++++ Tools/RemoteSynchronization/Program.cs | 224 ++++++------------ 2 files changed, 232 insertions(+), 158 deletions(-) create mode 100644 Tools/RemoteSynchronization/LightWeightBackendManager.cs diff --git a/Tools/RemoteSynchronization/LightWeightBackendManager.cs b/Tools/RemoteSynchronization/LightWeightBackendManager.cs new file mode 100644 index 000000000..233f48961 --- /dev/null +++ b/Tools/RemoteSynchronization/LightWeightBackendManager.cs @@ -0,0 +1,166 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Threading; +using System.Threading.Tasks; +using Duplicati.Library.Interface; +using Duplicati.Library.Utility; + +namespace RemoteSynchronization +{ + public class LightWeightBackendManager(string backendUrl, Dictionary options, int maxRetries = 3, int retryDelay = 1000) : IDisposable + { + private static readonly string LOGTAG = Duplicati.Library.Logging.Log.LogTagFromType(); + + public IBackend? _backend = null; + private readonly string _backendUrl = backendUrl; + private int _instantiations = 0; + private readonly int _maxRetries = maxRetries; + private Dictionary _options = options; + private readonly int _retryDelay = retryDelay; + private IStreamingBackend? _streamingBackend = null; + + public Task DeleteAsync(string remotename, CancellationToken token) + { + return RetryWithDelay( + $"Delete {remotename}", + () => _streamingBackend.DeleteAsync(remotename, token) + ); + } + + public string DisplayName + { + get + { + if (_streamingBackend == null) + { + Instantiate(); + } + + return _streamingBackend.DisplayName; + } + } + + public void Dispose() + { + try + { + _streamingBackend?.Dispose(); + _streamingBackend = null; + _backend?.Dispose(); + _backend = null; + } + catch (Exception ex) + { + Duplicati.Library.Logging.Log.WriteErrorMessage(LOGTAG, "rsync", ex, "Error during Dispose", null); + } + } + + public Task GetAsync(string remotename, Stream stream, CancellationToken token) + { + return RetryWithDelay( + $"Get {remotename}", + () => _streamingBackend.GetAsync(remotename, stream, token) + ); + } + + private void Instantiate() + { + if (_instantiations < _maxRetries) + { + _backend = Duplicati.Library.DynamicLoader.BackendLoader.GetBackend(_backendUrl, _options); + _streamingBackend = _backend as IStreamingBackend ?? throw new InvalidOperationException("Backend does not support streaming operations."); + _instantiations++; + } + else + { + throw new InvalidOperationException("Maximum number of backend instantiations reached."); + } + } + + public IAsyncEnumerable ListAsync(CancellationToken token) + { + while (true) + { + if (_streamingBackend == null) + { + Instantiate(); + } + + try + { + return _streamingBackend.ListAsync(token); + } + catch (Exception ex) + { + Duplicati.Library.Logging.Log.WriteErrorMessage(LOGTAG, "rsync", ex, "Error during operation: List", null); + Dispose(); // Dispose current backend and streaming backend + Task.Delay(_retryDelay, token).Await(); + } + } + + } + + public Task PutAsync(string remotename, Stream stream, CancellationToken token) + { + return RetryWithDelay( + $"Put {remotename}", + () => _streamingBackend.PutAsync(remotename, stream, token) + ); + } + + public Task RenameAsync(string oldname, string newname, CancellationToken token) + { + if (_streamingBackend == null) + { + Instantiate(); + } + + return _backend switch + { + IStreamingBackend sb => + RetryWithDelay( + $"Rename {oldname} to {newname}", + async () => + { + // Download the file, rename it, and delete the old one + using var downloaded = new MemoryStream(); + await sb.GetAsync(oldname, downloaded, token); + await sb.PutAsync(newname, downloaded, token); + await sb.DeleteAsync(oldname, token); + } + ), + IRenameEnabledBackend ireb => + RetryWithDelay( + $"Rename {oldname} to {newname}", + () => ireb.RenameAsync(oldname, newname, token) + ), + _ => throw new InvalidOperationException("Backend does not support renaming."), + }; + } + + private async Task RetryWithDelay(string operationName, Func action) + { + while (true) + { + if (_streamingBackend == null) + { + Instantiate(); + } + + try + { + await action(); + return; + } + catch (Exception ex) + { + Duplicati.Library.Logging.Log.WriteErrorMessage(LOGTAG, "rsync", ex, "Error during operation: {0}", operationName); + Dispose(); // Dispose current backend and streaming backend + await Task.Delay(_retryDelay); + } + } + } + + } +} diff --git a/Tools/RemoteSynchronization/Program.cs b/Tools/RemoteSynchronization/Program.cs index c713e1443..9450766ef 100644 --- a/Tools/RemoteSynchronization/Program.cs +++ b/Tools/RemoteSynchronization/Program.cs @@ -186,23 +186,11 @@ destination will be verified before being overwritten (if they seemingly match). 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); + using var b2m = new LightWeightBackendManager(config.Dst, dst_opts); // 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, CancellationToken.None); // 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 +208,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); failed_verify = not_verified.Count(); verified = to_verify.Count() - failed_verify; @@ -237,7 +225,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 +249,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); 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); 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); 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()) @@ -290,9 +278,9 @@ destination will be verified before being overwritten (if they seemingly match). 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); + (copied, copy_errors) = await CopyAsync(b1m, b2m, copy_errors, config); 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 +298,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"); @@ -346,7 +334,7 @@ destination will be verified before being overwritten (if they seemingly match). /// The files that will be copied. /// The parsed configuration for the tool. /// A tuple holding the number of succesful copies and a List of the files that failed. - private static async Task<(long, IEnumerable)> CopyAsync(IStreamingBackend b_src, IStreamingBackend b_dst, IEnumerable files, Config config) + private static async Task<(long, IEnumerable)> CopyAsync(LightWeightBackendManager b_src, LightWeightBackendManager b_dst, IEnumerable files, Config config) { long successful_copies = 0; List errors = []; @@ -464,7 +452,7 @@ destination will be verified before being overwritten (if they seemingly match). /// The files to delete. /// The parsed configuration for the tool. /// The number of successful deletions. - private static async Task DeleteAsync(IStreamingBackend b, IEnumerable files, Config config) + private static async Task DeleteAsync(LightWeightBackendManager b, IEnumerable files, Config config) { long successful_deletes = 0; long i = 0, n = files.Count(); @@ -560,7 +548,7 @@ destination will be verified before being overwritten (if they seemingly match). /// The destination backend. /// The parsed configuration for the tool. /// A tuple of Lists each holding the files to copy, delete and verify. - private static async Task<(IEnumerable, IEnumerable, IEnumerable)> PrepareFileLists(IStreamingBackend b_src, IStreamingBackend b_dst, Config config, CancellationToken cancelToken) + private static async Task<(IEnumerable, IEnumerable, IEnumerable)> PrepareFileLists(LightWeightBackendManager b_src, LightWeightBackendManager b_dst, Config config, CancellationToken cancelToken) { IEnumerable files_src, files_dst; @@ -646,142 +634,62 @@ destination will be verified before being overwritten (if they seemingly match). /// The files to rename. /// The parsed configuration for the tool. /// The number of successful renames. - private static async Task RenameAsync(IBackend b, IEnumerable files, Config config) + private static async Task RenameAsync(LightWeightBackendManager bm, IEnumerable files, Config config) { 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}", CancellationToken.None); + 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; } /// @@ -793,7 +701,7 @@ destination will be verified before being overwritten (if they seemingly match). /// The files to verify. /// The parsed configuration for the tool. /// A list of the files that failed verification. - private static async Task> VerifyAsync(IStreamingBackend b_src, IStreamingBackend b_dst, IEnumerable files, Config config) + private static async Task> VerifyAsync(LightWeightBackendManager b_src, LightWeightBackendManager b_dst, IEnumerable files, Config config) { var errors = new List(); using var s_src = new MemoryStream();