using System; using System.Collections.Generic; using System.Data; using System.Linq; using System.Runtime.CompilerServices; using System.Threading; using System.Threading.Tasks; using CoCoL; using Duplicati.Library.Interface; using Duplicati.Library.Main.Database; using Duplicati.Library.Main.Operation.Common; using Duplicati.Library.Main.Volumes; using Duplicati.Library.Utility; using Duplicati.StreamUtil; namespace Duplicati.Library.Main.Backend; #nullable enable /// /// The backend manager /// internal partial class BackendManager : IBackendManager { /// /// The log tag for the class /// private static readonly string LOGTAG = Logging.Log.LogTagFromType(); /// /// The channel for issuing and handling requests /// private readonly IChannel requestChannel = ChannelManager.CreateChannel(name: "BackendManager"); /// /// The queue runner task /// private readonly Task queueRunner; /// /// The execution context /// private readonly ExecuteContext context; /// /// Indicates whether the backend supports object locking /// private readonly Lazy supportsObjectLocking; /// /// Flag keeping track of whether the object has been disposed /// private bool isDisposed = false; /// /// Initializes a new instance of the class. /// /// The backend URL /// The options /// The backend writer /// The task reader public BackendManager(string backendUrl, Options options, IBackendWriter backendWriter, ITaskReader taskReader) { if (string.IsNullOrWhiteSpace(backendUrl)) throw new ArgumentNullException(nameof(backendUrl)); var isThrottleDisabled = options.DisableThrottle || options.ThrottleDisabledBackends.Contains(Library.Utility.Utility.GuessScheme(backendUrl) ?? string.Empty); // To avoid excessive parameter passing, the context is captured here context = new ExecuteContext( new ProgressHandler(backendWriter, taskReader), backendWriter ?? throw new ArgumentNullException(nameof(backendWriter)), new DatabaseCollector(), new ThrottleManager() { Limit = isThrottleDisabled ? 0 : options.MaxUploadPrSecond }, new ThrottleManager() { Limit = isThrottleDisabled ? 0 : options.MaxDownloadPrSecond }, taskReader ?? throw new ArgumentNullException(nameof(taskReader)), isThrottleDisabled, options ?? throw new ArgumentNullException(nameof(options)) ); supportsObjectLocking = new Lazy(() => { var backend = DynamicLoader.BackendLoader.GetBackend(backendUrl, context.Options.RawOptions); if (backend == null) return false; try { return backend is ILockingBackend; } finally { (backend as IDisposable)?.Dispose(); } }, isThreadSafe: true); // The BackendManager class is a wrapper that essentially sends // requests into a queue and processes them in order. // The Handler class is the one that actually processes the requests. queueRunner = Handler.RunHandlerAsync( requestChannel, backendUrl, context); } /// /// Gets a value indicating whether the configured backend supports object locking /// public bool SupportsObjectLocking => supportsObjectLocking.Value; /// /// Enters a task into the queue for processing. /// /// The operation to queue /// An awaitable task private async Task QueueTask(PendingOperationBase op) { if (queueRunner.IsCompleted) { if (queueRunner.IsFaulted) await queueRunner.ConfigureAwait(false); if (queueRunner.IsCanceled) throw new OperationCanceledException("Backend manager is stopped", queueRunner.Exception); throw new InvalidOperationException("Backend manager is stopped"); } try { await requestChannel.WriteAsync(op).ConfigureAwait(false); } catch (RetiredException ex) { // Try to get a better error message if (queueRunner.IsFaulted) await queueRunner.ConfigureAwait(false); throw new InvalidOperationException("Backend manager is stopped", ex); } } /// /// Calculates the hash of a file /// /// The filename /// The options /// The hash protected static string CalculateFileHash(string filename, Options options) { using (var fs = System.IO.File.OpenRead(filename)) using (var hasher = HashFactory.CreateHasher(options.FileHashAlgorithm)) return Convert.ToBase64String(hasher.ComputeHash(fs)); } /// /// Decrypts a file using the specified options /// /// The file to decrypt /// The name of the file. Used for detecting encryption algorithm if not specified in options or if it differs from the options /// The Duplicati options /// The decrypted file public TempFile DecryptFile(TempFile volume, string volume_name, Options options) { return GetOperation.DecryptFile(volume, volume_name, options); } /// /// Deletes a remote file /// /// The name of the remote file /// The size of the remote file, for statistics /// True if the operation should wait for the file to actually be deleted. If this argument is false, the task will complete once the operation is queued /// The cancellation token /// An awaitable task public async Task DeleteAsync(string remotename, long size, bool waitForComplete, CancellationToken cancelToken) { var op = new DeleteOperation(remotename, size, context, waitForComplete, cancelToken); await QueueTask(op).ConfigureAwait(false); await op.GetResult().ConfigureAwait(false); } /// /// Applies or updates an object lock on a remote file /// /// The name of the remote file /// The UTC time until which the lock should remain in effect /// The cancellation token /// An awaitable task public async Task SetObjectLockUntilAsync(string remotename, DateTime lockUntilUtc, CancellationToken cancelToken) { var op = new SetObjectLockOperation(remotename, lockUntilUtc, context, cancelToken); await QueueTask(op).ConfigureAwait(false); await op.GetResult().ConfigureAwait(false); } /// /// Gets the object lock expiration time for a remote file /// /// The name of the remote file /// The cancellation token /// The UTC time the lock expires, or null if no lock is set public async Task GetObjectLockUntilAsync(string remotename, CancellationToken cancelToken) { var op = new GetObjectLockOperation(remotename, context, cancelToken); await QueueTask(op).ConfigureAwait(false); return await op.GetResult().ConfigureAwait(false); } /// /// Gets a file from the remote location /// /// The name of the remote file /// The hash of the remote file, for verification /// The size of the remote file, for verification /// The cancellation token /// A temporary file with the contents of the remote file public async Task GetAsync(string remotename, string hash, long size, CancellationToken cancelToken) { var op = new GetOperation(remotename, size, context, cancelToken) { Hash = hash, Decrypt = true }; await QueueTask(op).ConfigureAwait(false); (var file, var _, var _) = await op.GetResult().ConfigureAwait(false); return file; } /// /// Gets a file from the remote location without decrypting it /// /// The name of the remote file /// The hash of the remote file, for verification /// The size of the remote file, for verification /// The cancellation token /// A temporary file with the contents of the remote file public async Task GetDirectAsync(string remotename, string hash, long size, CancellationToken cancelToken) { var op = new GetOperation(remotename, size, context, cancelToken) { Hash = hash, Decrypt = false }; await QueueTask(op).ConfigureAwait(false); (var file, var _, var _) = await op.GetResult().ConfigureAwait(false); return file; } /// /// Gets quota information from the backend /// /// The cancellation token /// The quota information public async Task GetQuotaInfoAsync(CancellationToken cancelToken) { var op = new QuotaInfoOperation(context, cancelToken); await QueueTask(op).ConfigureAwait(false); return await op.GetResult().ConfigureAwait(false); } /// /// Gets a file from the remote location, along with the hash and size of the file /// /// The name of the remote file /// The hash of the remote file, or null if not known /// The size of the remote file, or -1 if not known /// The cancellation token /// A tuple containing the temporary file, the hash of the file, and the size of the file public async Task<(TempFile File, string Hash, long Size)> GetWithInfoAsync(string remotename, string hash, long size, CancellationToken cancelToken) { var op = new GetOperation(remotename, size, context, cancelToken) { Hash = hash, Decrypt = true }; await QueueTask(op).ConfigureAwait(false); (var file, var downloadHash, var downloadSize) = await op.GetResult().ConfigureAwait(false); return (file, downloadHash, downloadSize); } /// /// Lists files on the remote destination /// /// The cancellation token /// The list of files public async Task> ListAsync(CancellationToken cancelToken) { var op = new ListOperation(context, cancelToken); await QueueTask(op).ConfigureAwait(false); return await op.GetResult().ConfigureAwait(false); } /// /// Uploads a volume to the remote location /// /// The volume to upload /// The index volume to upload, if any /// The callback to call when the index volume is finished /// True if the operation should wait for the file to actually be uploaded. If this argument is false, the task will complete once the operation is queued /// The callback to call when the database is updated /// The cancellation token /// An awaitable task public async Task PutAsync(VolumeWriterBase volume, IndexVolumeWriter? indexVolume, Func? indexVolumeFinished, bool waitForComplete, Func? onDbUpdate, CancellationToken cancelToken) { volume.Close(); var op = new PutOperation(volume.RemoteFilename, context, waitForComplete, cancelToken) { LocalTempfile = volume.TempFile, OriginalIndexFile = indexVolume, Unencrypted = false, TrackedInDb = true, IndexVolumeFinishedCallback = indexVolumeFinished, OnDbUpdate = onDbUpdate ?? PutOperation.OnDbUpdateDefault, }; // Prepare encryption op.StartEncryptionAndHashing(); await QueueTask(op).ConfigureAwait(false); await op.GetResult().ConfigureAwait(false); } /// /// Uploads a verification file to the remote location without encryption /// /// The name of the remote file /// The temporary file to upload /// The cancellation token /// An awaitable task public async Task PutVerificationFileAsync(string remotename, TempFile tempFile, CancellationToken cancelToken) { var op = new PutOperation(remotename, context, true, cancelToken) { LocalTempfile = tempFile, Unencrypted = true, // Avoid encrypting TrackedInDb = false, // Not tracked OriginalIndexFile = null, IndexVolumeFinishedCallback = null, OnDbUpdate = PutOperation.OnDbUpdateDefault, }; // Sets the task as already completed op.StartEncryptionAndHashing(); await QueueTask(op).ConfigureAwait(false); await op.GetResult().ConfigureAwait(false); } /// /// Flushes the database messages to the database. /// /// The database to write to. /// The cancellation token. /// A task that completes when the messages are flushed. public async Task FlushPendingMessagesAsync(LocalDatabase database, CancellationToken cancellationToken) { await context.Database.FlushPendingMessages(database, cancellationToken).ConfigureAwait(false); } /// /// Waits for the backend queue to be empty. /// /// The cancellation token. /// An awaitable task. public async Task WaitForEmptyAsync(CancellationToken cancellationToken) { var op = new WaitForEmptyOperation(context, cancellationToken); await QueueTask(op).ConfigureAwait(false); await op.GetResult().ConfigureAwait(false); } /// /// Waits for the backend queue to be empty and flushes the database messages /// /// The database to write to /// The transaction to use /// The cancellation token /// An awaitable task public async Task WaitForEmptyAsync(LocalDatabase database, CancellationToken cancellationToken) { await FlushPendingMessagesAsync(database, cancellationToken).ConfigureAwait(false); await WaitForEmptyAsync(cancellationToken).ConfigureAwait(false); await FlushPendingMessagesAsync(database, cancellationToken).ConfigureAwait(false); } /// /// Stops the backend manager and flushes any pending messages to the database. /// /// The database to write pending messages to. /// A task that completes when the runner is stopped and messages are flushed. public async Task StopRunnerAndFlushMessages(LocalDatabase database) { await requestChannel.RetireAsync().ConfigureAwait(false); await FlushPendingMessagesAsync(database, CancellationToken.None).ConfigureAwait(false); if (queueRunner.IsFaulted) Logging.Log.WriteWarningMessage(LOGTAG, "BackendManagerShutdown", queueRunner.Exception, "Backend manager queue runner crashed"); } /// /// Stops the backend manager and discards any pending messages /// public void StopRunnerAndDiscardMessages() { requestChannel.RetireAsync().Await(); if (queueRunner.IsFaulted) Logging.Log.WriteWarningMessage(LOGTAG, "BackendManagerShutdown", queueRunner.Exception, "Backend manager queue runner crashed"); context.Database.ClearPendingMessages(); } /// /// Performs a download of the files specified, with pre-fetch to overlap the download and processing /// /// The volumes to download /// The cancellation token /// The downloaded files and the volume they came from public async IAsyncEnumerable<(TempFile File, string Hash, long Size, string Name)> GetFilesOverlappedAsync(IEnumerable volumes, [EnumeratorCancellation] CancellationToken cancelToken) { var prevVolume = volumes.FirstOrDefault(); if (prevVolume == null) yield break; // Get the first volume, so we do not have pending parallel transfers var prevResult = await GetWithInfoAsync(prevVolume.Name, prevVolume.Hash, prevVolume.Size, cancelToken) .ConfigureAwait(false); foreach (var volume in volumes.Skip(1)) { // Prepare the next volume, while processing the previous one var nextTask = GetWithInfoAsync(volume.Name, volume.Hash, volume.Size, cancelToken); // Assuming we do not throw while yielding, otherwise we would need to dispose nextTask yield return (prevResult.File, prevResult.Hash, prevResult.Size, prevVolume.Name); prevResult.File.Dispose(); // Set up for next iteration prevVolume = volume; prevResult = await nextTask.ConfigureAwait(false); } // Return the last result yield return (prevResult.File, prevResult.Hash, prevResult.Size, prevVolume.Name); prevResult.File.Dispose(); } /// /// Updates the throttle values for upload and download /// /// The maximum upload speed in bytes per second /// The maximum download speed in bytes per second public void UpdateThrottleValues(long maxUploadPrSecond, long maxDownloadPrSecond) { if (context.IsThrottleDisabled) return; context.UploadThrottleManager.Limit = maxUploadPrSecond; context.DownloadThrottleManager.Limit = maxDownloadPrSecond; } /// /// Disposes the backend manager /// public void Dispose() { if (isDisposed) return; isDisposed = true; requestChannel.RetireAsync().Await(); context.Database.FlushMessagesToLog(); if (!queueRunner.IsCompleted) { Task.WhenAny(queueRunner, Task.Delay(1000)).Await(); if (!queueRunner.IsCompleted) Logging.Log.WriteWarningMessage(LOGTAG, "BackendManagerShutdown", null, "Backend manager queue runner did not stop"); if (queueRunner.IsFaulted) Logging.Log.WriteWarningMessage(LOGTAG, "BackendManagerShutdown", queueRunner.Exception, "Backend manager queue runner crashed"); } if (queueRunner.IsCompleted) queueRunner.Dispose(); } }