From 80ce9935260029b222a9d4fa80a0bf10ac934605 Mon Sep 17 00:00:00 2001 From: Kenneth Skovhede Date: Wed, 16 Apr 2025 19:05:51 +0200 Subject: [PATCH 1/7] Simple fix to prevent uploading dlist on failures --- Duplicati/Library/Main/Operation/BackupHandler.cs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/Duplicati/Library/Main/Operation/BackupHandler.cs b/Duplicati/Library/Main/Operation/BackupHandler.cs index 226b265b8..03f8af50b 100644 --- a/Duplicati/Library/Main/Operation/BackupHandler.cs +++ b/Duplicati/Library/Main/Operation/BackupHandler.cs @@ -709,6 +709,8 @@ namespace Duplicati.Library.Main.Operation using (new Logging.Timer(LOGTAG, "VerifyConsistency", "VerifyConsistency")) await db.VerifyConsistencyAsync(m_options.Blocksize, m_options.BlockhashSize, false); + await FlushBackend(m_database, null, m_result, backendManager).ConfigureAwait(false); + // Send the actual filelist await Backup.UploadRealFilelist.Run(m_result, db, backendManager, m_options, filesetvolume, filesetid, m_result.TaskControl, lastTempVolumeIncomplete); From 7be96780802adcd6866fe91463200b532af14eb2 Mon Sep 17 00:00:00 2001 From: Kenneth Skovhede Date: Wed, 23 Apr 2025 11:48:35 +0200 Subject: [PATCH 2/7] Fixed handling promotion of pending upload to actual filelist --- Duplicati/Library/Main/Operation/BackupHandler.cs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/Duplicati/Library/Main/Operation/BackupHandler.cs b/Duplicati/Library/Main/Operation/BackupHandler.cs index 03f8af50b..636472463 100644 --- a/Duplicati/Library/Main/Operation/BackupHandler.cs +++ b/Duplicati/Library/Main/Operation/BackupHandler.cs @@ -375,7 +375,8 @@ namespace Duplicati.Library.Main.Operation throw; } - return new PreBackupVerifyResult(database, lastTempFilelist); + // Return the updated version of the fileset, in case it was actually complete + return new PreBackupVerifyResult(database, database.GetLastIncompleteFilesetVolume(null)); } catch { From 3ed195d8218a06dea8d16e40be8d14e5f0a2215c Mon Sep 17 00:00:00 2001 From: Kenneth Skovhede Date: Wed, 23 Apr 2025 11:49:06 +0200 Subject: [PATCH 3/7] Updated tests --- Duplicati/UnitTest/DisruptionTests.cs | 45 +++++++++++++++++++++++---- 1 file changed, 39 insertions(+), 6 deletions(-) diff --git a/Duplicati/UnitTest/DisruptionTests.cs b/Duplicati/UnitTest/DisruptionTests.cs index 2f10b2371..2e106e24a 100644 --- a/Duplicati/UnitTest/DisruptionTests.cs +++ b/Duplicati/UnitTest/DisruptionTests.cs @@ -898,12 +898,15 @@ namespace Duplicati.UnitTest using (var c = new Controller("file://" + TARGETFOLDER, testopts.Expand(new { full_remote_verification = true }), null)) TestUtils.AssertResults(c.Test(long.MaxValue)); + // If we fail on the dlist after the upload, it is promoted to a regular backup + var expectedFilesets = before || failOnLastDblock ? 2 : 1; + // Make a new backup, should continue as normal using (var c = new Controller("file://" + TARGETFOLDER, testopts, null)) { var backupResults = c.Backup(new string[] { DATAFOLDER }); TestUtils.AssertResults(backupResults); - Assert.AreEqual(2, c.List().Filesets.Count()); + Assert.AreEqual(expectedFilesets, c.List().Filesets.Count()); } // Test that we can recreate @@ -916,12 +919,12 @@ namespace Duplicati.UnitTest using (var c = new Controller("file://" + TARGETFOLDER, testopts, null)) TestUtils.AssertResults(c.Repair()); - // Check that we have 2 versions + // Check that we have the correct versions using (var c = new Controller("file://" + TARGETFOLDER, testopts, null)) { var listResults = c.List(); TestUtils.AssertResults(listResults); - Assert.AreEqual(2, listResults.Filesets.Count()); + Assert.AreEqual(expectedFilesets, listResults.Filesets.Count()); } } @@ -984,12 +987,15 @@ namespace Duplicati.UnitTest // Make spacing for the next backup Thread.Sleep(3000); + // If we fail on the dlist after the upload, it is promoted to a regular backup + var expectedFilesets = before || failOnLastDblock ? 2 : 1; + // Make a new backup, should continue as normal using (var c = new Controller("file://" + TARGETFOLDER, testopts, null)) { IBackupResults backupResults = c.Backup(new string[] { DATAFOLDER }); TestUtils.AssertResults(backupResults); - Assert.AreEqual(2, c.List().Filesets.Count()); + Assert.AreEqual(expectedFilesets, c.List().Filesets.Count()); } // Verify that all is in order @@ -1011,7 +1017,7 @@ namespace Duplicati.UnitTest { var listResults = c.List(); TestUtils.AssertResults(listResults); - Assert.AreEqual(2, listResults.Filesets.Count()); + Assert.AreEqual(expectedFilesets, listResults.Filesets.Count()); } } @@ -1066,12 +1072,39 @@ namespace Duplicati.UnitTest [TestCase(false, true, 3, false)] [TestCase(true, true, 3, true)] [TestCase(false, true, 3, true)] + [TestCase(false, false, 3, true)] + [TestCase(true, false, 3, true)] public void TestMultiExceptionOnDlist(bool before, bool modifyInBetween, int runs, bool withBase) { var testopts = TestOptions; testopts["number-of-retries"] = "0"; testopts["dblock-size"] = "10mb"; - var expectedFilesets = withBase ? 3 : 2; + // We always have at least 1 backup at the end + var expectedFilesets = 1; + // If the files are modified, this will create multiple versions + if (modifyInBetween) + { + // If we prevent the file from being uploaded, + // it will be a temporary file, and a new synthetic file + // will be attempted, but that fails as well + // If it is after, then the upload completes, + // and it will be promoted to a regular uploaded backup + expectedFilesets += before ? 1 : runs; + + // If we modify with a base, then that creates a version as well + if (withBase) + expectedFilesets += 1; + } + else if (before && !withBase) + { + // Because an incomplete fileset is treated as a partial fileset, + // the final non-interrupted backup will see the syntheticly + // created fileset as a partial fileset, and create a new non-partial + // fileset, even though there are no changes between the two + // This does not happen with a base, because it simply does not + // attempt to upload the failing filesets + expectedFilesets += 1; + } if (withBase) { From e6ba6098d7dc24b8bab2c93751268d74288859d5 Mon Sep 17 00:00:00 2001 From: Kenneth Skovhede Date: Wed, 23 Apr 2025 12:11:16 +0200 Subject: [PATCH 4/7] Implemented immediate transaction commits for almost all uploads. This ensures that files cannot appear on the remote destination, despite terminating the host program. This fixes #4485 There is still work needed to handle the repair and compact operations so they use transactions and can commit changes during transfer progress. --- .../Backend/BackendManager.PutOperation.cs | 26 ++++++++++++++-- .../Library/Main/Backend/BackendManager.cs | 31 +++++++++++++++---- Duplicati/Library/Main/IBackendManager.cs | 12 ++++++- .../Operation/Backup/DataBlockProcessor.cs | 2 +- .../Backup/RecreateMissingIndexFiles.cs | 2 +- .../Operation/Backup/SpillCollectorProcess.cs | 2 +- .../Operation/Backup/UploadRealFilelist.cs | 2 +- .../Backup/UploadSyntheticFilelist.cs | 3 +- .../Main/Operation/Common/DatabaseCommon.cs | 10 ++++++ .../Library/Main/Operation/CompactHandler.cs | 16 +++++++--- .../Main/Operation/PurgeFilesHandler.cs | 6 +++- .../Library/Main/Operation/RepairHandler.cs | 12 ++++--- 12 files changed, 100 insertions(+), 24 deletions(-) diff --git a/Duplicati/Library/Main/Backend/BackendManager.PutOperation.cs b/Duplicati/Library/Main/Backend/BackendManager.PutOperation.cs index a63d5f07a..7b8c3a240 100644 --- a/Duplicati/Library/Main/Backend/BackendManager.PutOperation.cs +++ b/Duplicati/Library/Main/Backend/BackendManager.PutOperation.cs @@ -75,6 +75,10 @@ partial class BackendManager /// public required bool TrackedInDb { get; init; } /// + /// A callback that is invoked when the database needs to be updated + /// + public required Func OnDbUpdate { get; init; } + /// /// The encryption and hashing task, the return value indicates if the hash and size was updated /// private Task<(string Hash, long Size)>? encryptionAndHashingTask; @@ -94,6 +98,10 @@ partial class BackendManager /// A callback that is invoked when the index volume is finished /// public required Action? IndexVolumeFinishedCallback { get; init; } + /// + /// The default callback for database updates (no-op) + /// + public static Func OnDbUpdateDefault = () => Task.CompletedTask; /// /// Creates a new put operation @@ -201,12 +209,16 @@ partial class BackendManager // On first upload attempt, calculate the hash and size if (operationState == OperationState.None && TrackedInDb) + { Context.Database.LogRemoteVolumeUpdated(RemoteFilename, RemoteVolumeState.Uploading, size, hash); + await OnDbUpdate().ConfigureAwait(false); + } // First attempt here, finish the index file now that all information is known if (OriginalIndexFile != null && indexOperation == null) { Context.Database.LogRemoteVolumeUpdated(OriginalIndexFile.RemoteFilename, RemoteVolumeState.Uploading, -1, null); + await OnDbUpdate().ConfigureAwait(false); // Prepare an upload operation for the index file var req2 = new PutOperation(OriginalIndexFile.RemoteFilename, Context, false, cancelToken) @@ -215,9 +227,11 @@ partial class BackendManager TrackedInDb = TrackedInDb, Unencrypted = Unencrypted, IndexVolumeFinishedCallback = null, - OriginalIndexFile = null + OriginalIndexFile = null, + OnDbUpdate = OnDbUpdate }; + OriginalIndexFile.FinishVolume(hash, size); IndexVolumeFinishedCallback?.Invoke(); OriginalIndexFile.Close(); @@ -226,8 +240,11 @@ partial class BackendManager } // If we have previously attempted to upload, we need to rename the file - if (operationState == OperationState.Uploading) + if (operationState == OperationState.Uploading && TrackedInDb) + { RenameFileAfterError(size); + await OnDbUpdate().ConfigureAwait(false); + } // Flag for next attempt, if any operationState = OperationState.Uploading; @@ -263,6 +280,8 @@ partial class BackendManager { Context.Database.LogRemoteOperation("put", RemoteFilename, JsonConvert.SerializeObject(new { Size = Size, Hash = Hash })); Context.Statwriter.SendEvent(BackendActionType.Put, BackendEventType.Started, RemoteFilename, Size); + if (TrackedInDb) + await OnDbUpdate().ConfigureAwait(false); var begin = DateTime.Now; @@ -280,7 +299,10 @@ partial class BackendManager Logging.Log.WriteProfilingMessage(LOGTAG, "UploadSpeed", "Uploaded {0} in {1}, {2}/s", Library.Utility.Utility.FormatSizeString(Size), duration, Library.Utility.Utility.FormatSizeString((long)(Size / duration.TotalSeconds))); if (TrackedInDb) + { Context.Database.LogRemoteVolumeUpdated(RemoteFilename, RemoteVolumeState.Uploaded, Size, Hash); + await OnDbUpdate().ConfigureAwait(false); + } Context.Statwriter.SendEvent(BackendActionType.Put, BackendEventType.Completed, RemoteFilename, Size); diff --git a/Duplicati/Library/Main/Backend/BackendManager.cs b/Duplicati/Library/Main/Backend/BackendManager.cs index d6647e48a..0f9c40b15 100644 --- a/Duplicati/Library/Main/Backend/BackendManager.cs +++ b/Duplicati/Library/Main/Backend/BackendManager.cs @@ -249,9 +249,10 @@ internal partial class BackendManager : IBackendManager /// 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, Action? indexVolumeFinished, bool waitForComplete, CancellationToken cancelToken) + public async Task PutAsync(VolumeWriterBase volume, IndexVolumeWriter? indexVolume, Action? indexVolumeFinished, bool waitForComplete, Func? onDbUpdate, CancellationToken cancelToken) { volume.Close(); @@ -261,7 +262,8 @@ internal partial class BackendManager : IBackendManager OriginalIndexFile = indexVolume, Unencrypted = false, TrackedInDb = true, - IndexVolumeFinishedCallback = indexVolumeFinished + IndexVolumeFinishedCallback = indexVolumeFinished, + OnDbUpdate = onDbUpdate ?? PutOperation.OnDbUpdateDefault, }; // Prepare encryption @@ -285,7 +287,8 @@ internal partial class BackendManager : IBackendManager Unencrypted = true, // Avoid encrypting TrackedInDb = false, // Not tracked OriginalIndexFile = null, - IndexVolumeFinishedCallback = null + IndexVolumeFinishedCallback = null, + OnDbUpdate = PutOperation.OnDbUpdateDefault, }; // Sets the task as already completed @@ -294,20 +297,35 @@ internal partial class BackendManager : IBackendManager await op.GetResult().ConfigureAwait(false); } + /// + /// Flushes the database messages to the database + /// + /// The database to write to + /// The transaction to use + /// The cancellation token + /// + public Task FlushPendingMessagesAsync(LocalDatabase database, IDbTransaction? transaction, CancellationToken cancellationToken) + { + context.Database.FlushPendingMessages(database, transaction); + return Task.CompletedTask; + } + /// /// 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, IDbTransaction? transaction, CancellationToken cancellationToken) { - context.Database.FlushPendingMessages(database, transaction); + await FlushPendingMessagesAsync(database, transaction, cancellationToken).ConfigureAwait(false); var op = new WaitForEmptyOperation(context, cancellationToken); await QueueTask(op).ConfigureAwait(false); await op.GetResult().ConfigureAwait(false); - context.Database.FlushPendingMessages(database, transaction); + await FlushPendingMessagesAsync(database, transaction, cancellationToken).ConfigureAwait(false); } /// @@ -318,7 +336,8 @@ internal partial class BackendManager : IBackendManager public async Task StopRunnerAndFlushMessages(LocalDatabase database, IDbTransaction? transaction) { await requestChannel.RetireAsync().ConfigureAwait(false); - context.Database.FlushPendingMessages(database, transaction); + await FlushPendingMessagesAsync(database, transaction, CancellationToken.None).ConfigureAwait(false); + if (queueRunner.IsFaulted) Logging.Log.WriteWarningMessage(LOGTAG, "BackendManagerShutdown", queueRunner.Exception, "Backend manager queue runner crashed"); } diff --git a/Duplicati/Library/Main/IBackendManager.cs b/Duplicati/Library/Main/IBackendManager.cs index c9cd0372c..13e589d35 100644 --- a/Duplicati/Library/Main/IBackendManager.cs +++ b/Duplicati/Library/Main/IBackendManager.cs @@ -25,9 +25,10 @@ internal interface IBackendManager : IDisposable /// The index volume to upload, if any /// The action to call when the index volume is finished /// Whether to wait for the upload to complete + /// The action to call when the database should be updated /// The cancellation token /// An awaitable task - Task PutAsync(VolumeWriterBase blockVolume, IndexVolumeWriter? indexVolume, Action? indexVolumeFinished, bool waitForComplete, CancellationToken cancelToken); + Task PutAsync(VolumeWriterBase blockVolume, IndexVolumeWriter? indexVolume, Action? indexVolumeFinished, bool waitForComplete, Func? onDbUpdate, CancellationToken cancelToken); /// /// Uploads a file to the backend without encryption @@ -128,4 +129,13 @@ internal interface IBackendManager : IDisposable /// long LastWriteSize { get; } + /// + /// Flushes the database messages to the database + /// + /// The database to write to + /// The transaction to use + /// The cancellation token + /// + Task FlushPendingMessagesAsync(LocalDatabase database, IDbTransaction? transaction, CancellationToken cancellationToken); + } diff --git a/Duplicati/Library/Main/Operation/Backup/DataBlockProcessor.cs b/Duplicati/Library/Main/Operation/Backup/DataBlockProcessor.cs index 670f3a809..b6ebd49a5 100644 --- a/Duplicati/Library/Main/Operation/Backup/DataBlockProcessor.cs +++ b/Duplicati/Library/Main/Operation/Backup/DataBlockProcessor.cs @@ -143,7 +143,7 @@ namespace Duplicati.Library.Main.Operation.Backup blockvolume = null; indexvolume = null; - await backendManager.PutAsync(blockVolumeCopy, indexVolumeCopy, null, false, taskreader.ProgressToken); + await backendManager.PutAsync(blockVolumeCopy, indexVolumeCopy, null, false, () => database.FlushBackendMessagesAndCommitAsync(backendManager), taskreader.ProgressToken); } diff --git a/Duplicati/Library/Main/Operation/Backup/RecreateMissingIndexFiles.cs b/Duplicati/Library/Main/Operation/Backup/RecreateMissingIndexFiles.cs index 46e80a80a..88be0f140 100644 --- a/Duplicati/Library/Main/Operation/Backup/RecreateMissingIndexFiles.cs +++ b/Duplicati/Library/Main/Operation/Backup/RecreateMissingIndexFiles.cs @@ -47,7 +47,7 @@ internal static class RecreateMissingIndexFiles return; await database.UpdateRemoteVolumeAsync(w.RemoteFilename, RemoteVolumeState.Uploading, -1, null).ConfigureAwait(false); - await backendManager.PutAsync(w, null, null, false, taskreader.ProgressToken).ConfigureAwait(false); + await backendManager.PutAsync(w, null, null, false, () => database.FlushBackendMessagesAndCommitAsync(backendManager), taskreader.ProgressToken).ConfigureAwait(false); } } } diff --git a/Duplicati/Library/Main/Operation/Backup/SpillCollectorProcess.cs b/Duplicati/Library/Main/Operation/Backup/SpillCollectorProcess.cs index c391f7b21..06b6326cf 100644 --- a/Duplicati/Library/Main/Operation/Backup/SpillCollectorProcess.cs +++ b/Duplicati/Library/Main/Operation/Backup/SpillCollectorProcess.cs @@ -172,7 +172,7 @@ namespace Duplicati.Library.Main.Operation.Backup await database.AddIndexBlockLinkAsync(indexVolumeCopy.VolumeID, target.BlockVolume.VolumeID).ConfigureAwait(false); } - await backendManager.PutAsync(target.BlockVolume, indexVolumeCopy, null, false, taskreader.ProgressToken).ConfigureAwait(false); + await backendManager.PutAsync(target.BlockVolume, indexVolumeCopy, null, false, () => database.FlushBackendMessagesAndCommitAsync(backendManager), taskreader.ProgressToken).ConfigureAwait(false); } } } diff --git a/Duplicati/Library/Main/Operation/Backup/UploadRealFilelist.cs b/Duplicati/Library/Main/Operation/Backup/UploadRealFilelist.cs index 3d8431f62..aa8a85d9c 100644 --- a/Duplicati/Library/Main/Operation/Backup/UploadRealFilelist.cs +++ b/Duplicati/Library/Main/Operation/Backup/UploadRealFilelist.cs @@ -67,7 +67,7 @@ internal static class UploadRealFilelist await db.UpdateRemoteVolumeAsync(filesetvolume.RemoteFilename, RemoteVolumeState.Uploading, -1, null).ConfigureAwait(false); await db.CommitTransactionAsync("CommitUpdateRemoteVolume").ConfigureAwait(false); - await backendManager.PutAsync(filesetvolume, null, null, false, taskreader.ProgressToken).ConfigureAwait(false); + await backendManager.PutAsync(filesetvolume, null, null, false, () => db.FlushBackendMessagesAndCommitAsync(backendManager), taskreader.ProgressToken).ConfigureAwait(false); } } else diff --git a/Duplicati/Library/Main/Operation/Backup/UploadSyntheticFilelist.cs b/Duplicati/Library/Main/Operation/Backup/UploadSyntheticFilelist.cs index 7efa1b13e..76b80f852 100644 --- a/Duplicati/Library/Main/Operation/Backup/UploadSyntheticFilelist.cs +++ b/Duplicati/Library/Main/Operation/Backup/UploadSyntheticFilelist.cs @@ -115,8 +115,7 @@ namespace Duplicati.Library.Main.Operation.Backup if (!string.IsNullOrWhiteSpace(lastTempFilelist.Name) && (lastTempFilelist.State == RemoteVolumeState.Uploading || lastTempFilelist.State == RemoteVolumeState.Temporary)) await database.UpdateRemoteVolumeAsync(lastTempFilelist.Name, RemoteVolumeState.Deleting, -1, null); await database.CommitTransactionAsync("CommitUpdateFilelistVolume"); - - await backendManager.PutAsync(fsw, null, null, false, taskreader.ProgressToken); + await backendManager.PutAsync(fsw, null, null, true, () => database.FlushBackendMessagesAndCommitAsync(backendManager), taskreader.ProgressToken); } catch { diff --git a/Duplicati/Library/Main/Operation/Common/DatabaseCommon.cs b/Duplicati/Library/Main/Operation/Common/DatabaseCommon.cs index d7acefa5a..7a3a20a9a 100644 --- a/Duplicati/Library/Main/Operation/Common/DatabaseCommon.cs +++ b/Duplicati/Library/Main/Operation/Common/DatabaseCommon.cs @@ -21,6 +21,7 @@ using System; using System.Linq; +using System.Threading; using System.Threading.Tasks; using Duplicati.Library.Main.Database; @@ -66,6 +67,15 @@ namespace Duplicati.Library.Main.Operation.Common return RunOnMain(() => m_db.UpdateRemoteVolume(name, state, size, hash, suppressCleanup, deleteGraceTime, GetTransaction())); } + public async Task FlushBackendMessagesAndCommitAsync(IBackendManager backendManager) + { + await FlushPendingMessagesAsync(backendManager).ConfigureAwait(false); + await CommitTransactionAsync("FlushBackendMessagesAndCommitAsync").ConfigureAwait(false); + } + + public Task FlushPendingMessagesAsync(IBackendManager backendManager) + => RunOnMain(() => backendManager.FlushPendingMessagesAsync(m_db, GetTransaction(), CancellationToken.None).ConfigureAwait(false)); + public Task CommitTransactionAsync(string message, bool restart = true) { return RunOnMain(() => diff --git a/Duplicati/Library/Main/Operation/CompactHandler.cs b/Duplicati/Library/Main/Operation/CompactHandler.cs index 6a05d3de5..f975c3eb4 100644 --- a/Duplicati/Library/Main/Operation/CompactHandler.cs +++ b/Duplicati/Library/Main/Operation/CompactHandler.cs @@ -183,7 +183,7 @@ namespace Duplicati.Library.Main.Operation if (newvol.Filesize > (m_options.VolumeSize - m_options.Blocksize)) { - await FinishVolumeAndUpload(db, backendManager, newvol, newvolindex, uploadedVolumes); + await FinishVolumeAndUpload(db, backendManager, newvol, newvolindex, uploadedVolumes, rtr); newvol = new BlockVolumeWriter(m_options); newvol.VolumeID = db.RegisterRemoteVolume(newvol.RemoteFilename, RemoteVolumeType.Blocks, RemoteVolumeState.Temporary, rtr.Transaction); @@ -215,7 +215,7 @@ namespace Duplicati.Library.Main.Operation if (blocksInVolume > 0) { - await FinishVolumeAndUpload(db, backendManager, newvol, newvolindex, uploadedVolumes).ConfigureAwait(false); + await FinishVolumeAndUpload(db, backendManager, newvol, newvolindex, uploadedVolumes, rtr).ConfigureAwait(false); } else { @@ -323,7 +323,7 @@ namespace Duplicati.Library.Main.Operation yield return d; } - private async Task FinishVolumeAndUpload(LocalDeleteDatabase db, IBackendManager backendManager, BlockVolumeWriter newvol, IndexVolumeWriter newvolindex, List> uploadedVolumes) + private async Task FinishVolumeAndUpload(LocalDeleteDatabase db, IBackendManager backendManager, BlockVolumeWriter newvol, IndexVolumeWriter newvolindex, List> uploadedVolumes, ReusableTransaction rtr) { Action indexVolumeFinished = () => { @@ -339,8 +339,16 @@ namespace Duplicati.Library.Main.Operation uploadedVolumes.Add(new KeyValuePair(newvol.RemoteFilename, newvol.Filesize)); if (newvolindex != null) uploadedVolumes.Add(new KeyValuePair(newvolindex.RemoteFilename, newvolindex.Filesize)); + + // TODO: The upload here does not flush the database messages, + // and this can leave the database in a state where it does not know of the remote file + // To fix it, we need thread-safe access to the daabase and transaction + // Once fixed, we can perhaps let he backend manager simply call the database directly if (!m_options.Dryrun) - await backendManager.PutAsync(newvol, newvolindex, indexVolumeFinished, false, m_result.TaskControl.ProgressToken).ConfigureAwait(false); + { + rtr.Commit("CommitUpload"); + await backendManager.PutAsync(newvol, newvolindex, indexVolumeFinished, false, null, m_result.TaskControl.ProgressToken).ConfigureAwait(false); + } else Logging.Log.WriteDryrunMessage(LOGTAG, "WouldUploadGeneratedBlockset", "Would upload generated blockset of size {0}", Library.Utility.Utility.FormatSizeString(newvol.Filesize)); } diff --git a/Duplicati/Library/Main/Operation/PurgeFilesHandler.cs b/Duplicati/Library/Main/Operation/PurgeFilesHandler.cs index 0748482b8..cc873661e 100644 --- a/Duplicati/Library/Main/Operation/PurgeFilesHandler.cs +++ b/Duplicati/Library/Main/Operation/PurgeFilesHandler.cs @@ -213,7 +213,11 @@ namespace Duplicati.Library.Main.Operation db.UpdateRemoteVolume(f.Key, RemoteVolumeState.Deleting, f.Value, null, tr); tr.Commit(); - await backendManager.PutAsync(vol, null, null, true, m_result.TaskControl.ProgressToken).ConfigureAwait(false); + + await backendManager.PutAsync(vol, null, null, true, async () => + { + await backendManager.FlushPendingMessagesAsync(db, null, m_result.TaskControl.ProgressToken).ConfigureAwait(false); + }, m_result.TaskControl.ProgressToken).ConfigureAwait(false); await backendManager.DeleteAsync(prevfilename, -1, true, m_result.TaskControl.ProgressToken).ConfigureAwait(false); await backendManager.WaitForEmptyAsync(db, null, m_result.TaskControl.ProgressToken).ConfigureAwait(false); } diff --git a/Duplicati/Library/Main/Operation/RepairHandler.cs b/Duplicati/Library/Main/Operation/RepairHandler.cs index d28cbd44e..3a967e007 100644 --- a/Duplicati/Library/Main/Operation/RepairHandler.cs +++ b/Duplicati/Library/Main/Operation/RepairHandler.cs @@ -350,7 +350,8 @@ namespace Duplicati.Library.Main.Operation fsw.VolumeID = db.RegisterRemoteVolume(fsw.RemoteFilename, RemoteVolumeType.Files, -1, RemoteVolumeState.Temporary); db.LinkFilesetToVolume(filesetId, fsw.VolumeID, null); - await backendManager.PutAsync(fsw, null, null, false, cancellationToken).ConfigureAwait(false); + // TODO: Rewrite to use transactions and flush db messages as needed + await backendManager.PutAsync(fsw, null, null, false, null, cancellationToken).ConfigureAwait(false); } if (anyDlistUploads) @@ -415,7 +416,8 @@ namespace Duplicati.Library.Main.Operation else { db.UpdateRemoteVolume(volumeWriter.RemoteFilename, RemoteVolumeState.Uploading, -1, null, null); - await backendManager.PutAsync(volumeWriter, null, null, false, cancellationToken).ConfigureAwait(false); + // TODO: Rewrite to use transactions and flush db messages as needed + await backendManager.PutAsync(volumeWriter, null, null, false, null, cancellationToken).ConfigureAwait(false); } } else if (n.Type == RemoteVolumeType.Index) @@ -456,7 +458,8 @@ namespace Duplicati.Library.Main.Operation else { db.UpdateRemoteVolume(w.RemoteFilename, RemoteVolumeState.Uploading, -1, null, null); - await backendManager.PutAsync(w, null, null, false, cancellationToken).ConfigureAwait(false); + // TODO: Rewrite to use transactions and flush db messages as needed + await backendManager.PutAsync(w, null, null, false, null, cancellationToken).ConfigureAwait(false); } } else if (n.Type == RemoteVolumeType.Blocks) @@ -552,7 +555,8 @@ namespace Duplicati.Library.Main.Operation else { db.UpdateRemoteVolume(w.RemoteFilename, RemoteVolumeState.Uploading, -1, null, null); - await backendManager.PutAsync(w, null, null, false, cancellationToken).ConfigureAwait(false); + // TODO: Rewrite to use transactions and flush db messages as needed + await backendManager.PutAsync(w, null, null, false, null, cancellationToken).ConfigureAwait(false); } } } From e8f85961d0666719c66bcab9532afed8dd5fa1bf Mon Sep 17 00:00:00 2001 From: Kenneth Skovhede Date: Wed, 23 Apr 2025 12:11:33 +0200 Subject: [PATCH 5/7] Include more changes before comitting the transaction --- Duplicati/Library/Main/Operation/Backup/DataBlockProcessor.cs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/Duplicati/Library/Main/Operation/Backup/DataBlockProcessor.cs b/Duplicati/Library/Main/Operation/Backup/DataBlockProcessor.cs index b6ebd49a5..01e4f123a 100644 --- a/Duplicati/Library/Main/Operation/Backup/DataBlockProcessor.cs +++ b/Duplicati/Library/Main/Operation/Backup/DataBlockProcessor.cs @@ -127,8 +127,6 @@ namespace Duplicati.Library.Main.Operation.Backup blockvolume.Close(); - await database.CommitTransactionAsync("CommitAddBlockToOutputFlush"); - IndexVolumeWriter indexVolumeCopy = null; if (indexvolume != null) { @@ -143,6 +141,7 @@ namespace Duplicati.Library.Main.Operation.Backup blockvolume = null; indexvolume = null; + await database.CommitTransactionAsync("CommitAddBlockToOutputFlush"); await backendManager.PutAsync(blockVolumeCopy, indexVolumeCopy, null, false, () => database.FlushBackendMessagesAndCommitAsync(backendManager), taskreader.ProgressToken); } From d19e7f443032a637cab2ebda97cf45ed93dd81cb Mon Sep 17 00:00:00 2001 From: Kenneth Skovhede Date: Wed, 23 Apr 2025 12:12:38 +0200 Subject: [PATCH 6/7] More safeguards against getting an unknown remote file --- .../Library/Main/Operation/Backup/RecreateMissingIndexFiles.cs | 1 + Duplicati/Library/Main/Operation/Backup/SpillCollectorProcess.cs | 1 + 2 files changed, 2 insertions(+) diff --git a/Duplicati/Library/Main/Operation/Backup/RecreateMissingIndexFiles.cs b/Duplicati/Library/Main/Operation/Backup/RecreateMissingIndexFiles.cs index 88be0f140..608ff7b7e 100644 --- a/Duplicati/Library/Main/Operation/Backup/RecreateMissingIndexFiles.cs +++ b/Duplicati/Library/Main/Operation/Backup/RecreateMissingIndexFiles.cs @@ -47,6 +47,7 @@ internal static class RecreateMissingIndexFiles return; await database.UpdateRemoteVolumeAsync(w.RemoteFilename, RemoteVolumeState.Uploading, -1, null).ConfigureAwait(false); + await database.CommitTransactionAsync("RecreateMissingIndexFile").ConfigureAwait(false); await backendManager.PutAsync(w, null, null, false, () => database.FlushBackendMessagesAndCommitAsync(backendManager), taskreader.ProgressToken).ConfigureAwait(false); } } diff --git a/Duplicati/Library/Main/Operation/Backup/SpillCollectorProcess.cs b/Duplicati/Library/Main/Operation/Backup/SpillCollectorProcess.cs index 06b6326cf..46d7fb02c 100644 --- a/Duplicati/Library/Main/Operation/Backup/SpillCollectorProcess.cs +++ b/Duplicati/Library/Main/Operation/Backup/SpillCollectorProcess.cs @@ -172,6 +172,7 @@ namespace Duplicati.Library.Main.Operation.Backup await database.AddIndexBlockLinkAsync(indexVolumeCopy.VolumeID, target.BlockVolume.VolumeID).ConfigureAwait(false); } + await database.CommitTransactionAsync("UploadSpillVolume").ConfigureAwait(false); await backendManager.PutAsync(target.BlockVolume, indexVolumeCopy, null, false, () => database.FlushBackendMessagesAndCommitAsync(backendManager), taskreader.ProgressToken).ConfigureAwait(false); } } From 66862043b662c387b152e9c66f122c16c5d88507 Mon Sep 17 00:00:00 2001 From: Kenneth Skovhede Date: Wed, 23 Apr 2025 13:44:46 +0200 Subject: [PATCH 7/7] Fixed a case where the transactions were not used, despite being active --- .../Library/Main/Backend/BackendManager.cs | 18 +++++++++++++----- Duplicati/Library/Main/IBackendManager.cs | 7 +++++++ .../Library/Main/Operation/BackupHandler.cs | 10 ++++++---- .../Main/Operation/Common/DatabaseCommon.cs | 4 ++-- 4 files changed, 28 insertions(+), 11 deletions(-) diff --git a/Duplicati/Library/Main/Backend/BackendManager.cs b/Duplicati/Library/Main/Backend/BackendManager.cs index 0f9c40b15..20ded2a5f 100644 --- a/Duplicati/Library/Main/Backend/BackendManager.cs +++ b/Duplicati/Library/Main/Backend/BackendManager.cs @@ -310,6 +310,18 @@ internal partial class BackendManager : IBackendManager return Task.CompletedTask; } + /// + /// 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 /// @@ -320,11 +332,7 @@ internal partial class BackendManager : IBackendManager public async Task WaitForEmptyAsync(LocalDatabase database, IDbTransaction? transaction, CancellationToken cancellationToken) { await FlushPendingMessagesAsync(database, transaction, cancellationToken).ConfigureAwait(false); - - var op = new WaitForEmptyOperation(context, cancellationToken); - await QueueTask(op).ConfigureAwait(false); - await op.GetResult().ConfigureAwait(false); - + await WaitForEmptyAsync(cancellationToken).ConfigureAwait(false); await FlushPendingMessagesAsync(database, transaction, cancellationToken).ConfigureAwait(false); } diff --git a/Duplicati/Library/Main/IBackendManager.cs b/Duplicati/Library/Main/IBackendManager.cs index 13e589d35..99c4104ba 100644 --- a/Duplicati/Library/Main/IBackendManager.cs +++ b/Duplicati/Library/Main/IBackendManager.cs @@ -39,6 +39,13 @@ internal interface IBackendManager : IDisposable /// An awaitable task Task PutVerificationFileAsync(string remotename, TempFile tempFile, CancellationToken cancelToken); + /// + /// Waits for the backend queue to be empty + /// + /// The cancellation token + /// An awaitable task + Task WaitForEmptyAsync(CancellationToken cancellationToken); + /// /// Waits for the backend queue to be empty and flushes any pending messages to the database /// diff --git a/Duplicati/Library/Main/Operation/BackupHandler.cs b/Duplicati/Library/Main/Operation/BackupHandler.cs index 636472463..322e0f1a2 100644 --- a/Duplicati/Library/Main/Operation/BackupHandler.cs +++ b/Duplicati/Library/Main/Operation/BackupHandler.cs @@ -595,14 +595,16 @@ namespace Duplicati.Library.Main.Operation return new AggregateException(ex.First().Message, ex); } - private static async Task FlushBackend(LocalDatabase database, IDbTransaction transaction, BackupResults result, IBackendManager backendManager) + private static async Task FlushBackend(Backup.BackupDatabase database, BackupResults result, IBackendManager backendManager) { // Wait for upload completion result.OperationProgressUpdater.UpdatePhase(OperationPhase.Backup_WaitForUpload); try { - await backendManager.WaitForEmptyAsync(database, transaction, result.TaskControl.ProgressToken).ConfigureAwait(false); + await database.FlushBackendMessagesAndCommitAsync(backendManager).ConfigureAwait(false); + await backendManager.WaitForEmptyAsync(result.TaskControl.ProgressToken).ConfigureAwait(false); + await database.FlushBackendMessagesAndCommitAsync(backendManager).ConfigureAwait(false); // Grab the size of the last uploaded volume return backendManager.LastWriteSize; } @@ -710,14 +712,14 @@ namespace Duplicati.Library.Main.Operation using (new Logging.Timer(LOGTAG, "VerifyConsistency", "VerifyConsistency")) await db.VerifyConsistencyAsync(m_options.Blocksize, m_options.BlockhashSize, false); - await FlushBackend(m_database, null, m_result, backendManager).ConfigureAwait(false); + await FlushBackend(db, m_result, backendManager).ConfigureAwait(false); // Send the actual filelist await Backup.UploadRealFilelist.Run(m_result, db, backendManager, m_options, filesetvolume, filesetid, m_result.TaskControl, lastTempVolumeIncomplete); // Wait for upload completion m_result.OperationProgressUpdater.UpdatePhase(OperationPhase.Backup_WaitForUpload); - var lastVolumeSize = await FlushBackend(m_database, null, m_result, backendManager).ConfigureAwait(false); + var lastVolumeSize = await FlushBackend(db, m_result, backendManager).ConfigureAwait(false); if (!m_options.Dryrun) database.TerminatedWithActiveUploads = false; diff --git a/Duplicati/Library/Main/Operation/Common/DatabaseCommon.cs b/Duplicati/Library/Main/Operation/Common/DatabaseCommon.cs index 7a3a20a9a..0d9d8d8e3 100644 --- a/Duplicati/Library/Main/Operation/Common/DatabaseCommon.cs +++ b/Duplicati/Library/Main/Operation/Common/DatabaseCommon.cs @@ -69,11 +69,11 @@ namespace Duplicati.Library.Main.Operation.Common public async Task FlushBackendMessagesAndCommitAsync(IBackendManager backendManager) { - await FlushPendingMessagesAsync(backendManager).ConfigureAwait(false); + await FlushPendingBackendMessagesAsync(backendManager).ConfigureAwait(false); await CommitTransactionAsync("FlushBackendMessagesAndCommitAsync").ConfigureAwait(false); } - public Task FlushPendingMessagesAsync(IBackendManager backendManager) + private Task FlushPendingBackendMessagesAsync(IBackendManager backendManager) => RunOnMain(() => backendManager.FlushPendingMessagesAsync(m_db, GetTransaction(), CancellationToken.None).ConfigureAwait(false)); public Task CommitTransactionAsync(string message, bool restart = true)