From f0eb67d8df1e5c8ef6a4331748564f9554af49db Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ricardo=20Gro=C3=9F?= Date: Fri, 16 Oct 2026 22:43:20 +0200 Subject: [PATCH] fix(storage): keep prepared transactions' WAL records until they commit or roll back (#188) - PrepareTransactionAsync appends its Begin/Write records and flushes under _commitLock, like a commit, so they no longer interleave with a group-commit batch and no truncation runs while they are written. Returns false if the lock cannot be taken within WriteTimeoutMs. - A prepared transaction is tracked until it commits or rolls back (CompleteTransaction). Meanwhile checkpoints still write pages but skip the WAL truncation, and RecoverAsync skips it too, so MarkTransactionCommittedAsync, which appends only the commit record, finds the prepared Write records in the WAL and the commit survives a crash. - Document the behaviour, including that an unresolved prepare holds back truncation and that recovery rolls back a transaction still prepared. - Drop comments that said prepare writes without _commitLock. Co-Authored-By: Claude Opus 5.5 --- .../Storage/StorageEngine.GroupCommit.cs | 7 +- .../Storage/StorageEngine.Recovery.cs | 24 +- .../Storage/StorageEngine.Transactions.cs | 76 +++++-- src/BLite.Core/Transactions/WriteAheadLog.cs | 4 +- .../PreparedTransactionWalTests.cs | 208 ++++++++++++++++++ 5 files changed, 293 insertions(+), 26 deletions(-) create mode 100644 tests/BLite.Tests/PreparedTransactionWalTests.cs diff --git a/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs b/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs index cc66c764..7be47a53 100644 --- a/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs +++ b/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs @@ -153,8 +153,7 @@ private async Task ProcessBatchAsync(List batch) // In multi-process mode we record the WAL byte offset at the start of each // Write record so the SHM hash table can map pageId → offset. We use // WriteDataRecordAndGetOffsetAsync which atomically captures the position - // inside the WAL write lock, ensuring PrepareTransactionAsync callers that - // also write to the WAL (without _commitLock) cannot shift the position + // inside the WAL write lock, so no other WAL writer can shift the position // between the snapshot and the actual write. var wal = (_shm != null) ? (_wal as Transactions.WriteAheadLog) : null; var pageOffsets = wal != null ? new List<(uint pageId, long walOffset)>() : null; @@ -185,8 +184,8 @@ private async Task ProcessBatchAsync(List batch) { // When multi-process mode is active, use the offset-returning variant // so the record start offset is captured atomically inside the WAL - // write lock, preventing concurrent PrepareTransactionAsync writers - // from shifting the stream position between snapshot and write. + // write lock, preventing other WAL writers from shifting the stream + // position between snapshot and write. // In single-process mode, fall back to the standard write. if (wal != null) pageOffsets!.Add((pageId, await wal.WriteDataRecordAndGetOffsetAsync( diff --git a/src/BLite.Core/Storage/StorageEngine.Recovery.cs b/src/BLite.Core/Storage/StorageEngine.Recovery.cs index e247b421..2f971752 100644 --- a/src/BLite.Core/Storage/StorageEngine.Recovery.cs +++ b/src/BLite.Core/Storage/StorageEngine.Recovery.cs @@ -218,7 +218,7 @@ private async Task CheckpointCoreAsync(bool commitLockHeld, CancellationToken ct /// /// Truncates the WAL if every committed record has been checkpointed, then releases the /// pages queued by . Skipped when a commit keeps the lock for longer - /// than a short wait. + /// than a short wait, and while a prepared transaction has neither committed nor rolled back. /// private Task TryTruncateWalAsync(CancellationToken ct) => TryTruncateWalCoreAsync(commitLockHeld: false, ct); @@ -267,7 +267,9 @@ private async Task TryTruncateWalCoreAsync(bool commitLockHeld, CancellationToke } // Double-check: new commits may have promoted entries // between our IsEmpty check and acquiring the lock. - if (_walIndex.IsEmpty) + // A prepared transaction's records stay until it commits or rolls back: committing it + // through MarkTransactionCommittedAsync appends only the commit record (#188). + if (_walIndex.IsEmpty && _preparedTransactions.IsEmpty) { await _wal.TruncateAsync(ct); if (_shm != null) @@ -378,7 +380,9 @@ internal async Task BackupDetailedAsync(string destinationDb /// /// Recovers from crash by replaying WAL. - /// Applies all committed transactions to PageFile, then truncates WAL. + /// Applies all committed transactions to PageFile, then truncates WAL, unless a prepared + /// transaction has neither committed nor rolled back yet (see ). + /// A transaction that never committed, prepared or not, is not applied. /// public async Task RecoverAsync(CancellationToken ct = default) { @@ -472,6 +476,20 @@ public async Task RecoverAsync(CancellationToken ct = default) return; } + // A prepared transaction's records stay until it commits or rolls back (#188), and with + // them the committed images just written, which FreePage must still see as in the WAL, + // as after a checkpoint that does not truncate. + if (!_preparedTransactions.IsEmpty) + { + foreach (var txnId in committedTxns) + { + if (txnWrites.TryGetValue(txnId, out var writes)) + foreach (var (pageId, _) in writes) + _checkpointedWalPages.TryAdd(pageId, 0); + } + return; + } + // 5. Truncate WAL (all changes now in PageFile) await _wal.TruncateAsync(ct); if (_shm != null) diff --git a/src/BLite.Core/Storage/StorageEngine.Transactions.cs b/src/BLite.Core/Storage/StorageEngine.Transactions.cs index 835e2f8d..200b71c2 100644 --- a/src/BLite.Core/Storage/StorageEngine.Transactions.cs +++ b/src/BLite.Core/Storage/StorageEngine.Transactions.cs @@ -1,3 +1,4 @@ +using System.Collections.Concurrent; using System.Diagnostics; using BLite.Core.Transactions; @@ -306,11 +307,24 @@ private void DiscardSharedWalState(long generation) return pageData; } + // Transactions whose records PrepareTransactionAsync has written to the WAL and that have not + // committed or rolled back since. Their records must stay in the WAL until then: + // MarkTransactionCommittedAsync appends only the commit record, so recovery needs the prepared + // Write records (#188). Added and checked under _commitLock, so no truncation runs between a + // prepare and its registration; removed by CompleteTransaction. + private readonly ConcurrentDictionary _preparedTransactions = new(); + /// /// Forgets a transaction once it has committed or rolled back through the ulong /// overloads, which callers use directly without going through the ones. + /// A prepared transaction no longer holds back WAL truncation: a commit has made its pages + /// durable (in the WAL until a checkpoint writes them), and a rolled-back one needs nothing. /// - private void CompleteTransaction(ulong transactionId) => _activeTransactions.TryRemove(transactionId, out _); + private void CompleteTransaction(ulong transactionId) + { + _activeTransactions.TryRemove(transactionId, out _); + _preparedTransactions.TryRemove(transactionId, out _); + } public async Task CommitTransactionAsync(Transaction transaction, CancellationToken ct = default) { @@ -342,24 +356,29 @@ public async Task RollbackTransactionAsync(Transaction transaction) /// /// Prepares a transaction: writes all changes to WAL but doesn't commit yet. /// Part of 2-Phase Commit protocol. + /// + /// The records are appended under the commit lock, like a commit's, so they never interleave + /// with a commit batch and no WAL truncation runs while they are written. Until the + /// transaction commits or rolls back, checkpoints still write committed pages to the page + /// files but do not truncate the WAL, so finds the + /// prepared records there (#188). A prepared transaction that is never resolved therefore + /// keeps the WAL growing; and roll back + /// the transactions they own. Recovery applies only committed transactions: a transaction + /// still prepared when the engine closes or the process dies is rolled back on reopen. + /// /// /// Transaction ID - /// All writes to record in WAL - /// True if preparation succeeded + /// + /// True if preparation succeeded; false if it failed, including when the commit lock could not + /// be taken within . + /// /// Multi-process mode (see ). public async Task PrepareTransactionAsync(ulong transactionId) { ThrowIfMultiProcessAccess(nameof(PrepareTransactionAsync)); try { - await _wal.WriteBeginRecordAsync(transactionId); - - foreach (var walEntry in _walCache[transactionId]) - { - await _wal.WriteDataRecordAsync(transactionId, walEntry.Key, walEntry.Value); - } - - await _wal.FlushAsync(); // Ensure WAL is persisted + await PrepareCoreAsync(transactionId, _walCache[transactionId], CancellationToken.None); return true; } catch (Exception ex) @@ -374,23 +393,44 @@ public async Task PrepareTransactionAsync(ulong transactionId, Cancellatio { ThrowIfMultiProcessAccess(nameof(PrepareTransactionAsync)); try + { + _walCache.TryGetValue(transactionId, out var changes); + await PrepareCoreAsync(transactionId, changes, ct); + return true; + } + catch + { + return false; + } + } + + /// + /// Appends the Begin and Write records of a prepared transaction and flushes them, under + /// _commitLock, then registers the transaction so that no truncation drops the records + /// before it commits or rolls back (see ). + /// + private async Task PrepareCoreAsync(ulong transactionId, IEnumerable>? pages, CancellationToken ct) + { + if (!await _commitLock.WaitAsync(_config.LockTimeout.WriteTimeoutMs, ct)) + throw new TimeoutException("Timed out acquiring commit lock (PrepareTransaction)."); + try { await _wal.WriteBeginRecordAsync(transactionId, ct); - if (_walCache.TryGetValue(transactionId, out var changes)) + if (pages != null) { - foreach (var walEntry in changes) + foreach (var walEntry in pages) { await _wal.WriteDataRecordAsync(transactionId, walEntry.Key, walEntry.Value, ct); } } - + await _wal.FlushAsync(ct); // Ensure WAL is persisted - return true; + _preparedTransactions[transactionId] = 0; } - catch + finally { - return false; + _commitLock.Release(); } } @@ -585,6 +625,8 @@ public async Task CommitTransactionAsync(ulong transactionId, CancellationToken /// /// Marks a transaction as committed after WAL writes. /// Used for 2PC: after PrepareAsync() writes to WAL, this finalizes the commit. + /// Appends only the commit record: the Write records are the ones the prepare wrote, which no + /// checkpoint truncates before this call (#188). /// /// Transaction to mark committed /// Multi-process mode (see ). diff --git a/src/BLite.Core/Transactions/WriteAheadLog.cs b/src/BLite.Core/Transactions/WriteAheadLog.cs index 407233bd..7e4c1470 100644 --- a/src/BLite.Core/Transactions/WriteAheadLog.cs +++ b/src/BLite.Core/Transactions/WriteAheadLog.cs @@ -591,8 +591,8 @@ internal void SeekToSharedEnd(long sharedEnd) /// /// Writes a WAL data record and returns the byte offset at which the record /// starts in the WAL file. The offset is captured atomically inside the WAL - /// write lock — so no concurrent WAL writer (e.g. PrepareTransactionAsync) - /// can shift the stream position between the snapshot and the actual write. + /// write lock — so no concurrent WAL writer can shift the stream position between + /// the snapshot and the actual write. /// Used by the group-commit path (Phase 4) to build the SHM page index. /// internal async ValueTask WriteDataRecordAndGetOffsetAsync( diff --git a/tests/BLite.Tests/PreparedTransactionWalTests.cs b/tests/BLite.Tests/PreparedTransactionWalTests.cs new file mode 100644 index 00000000..39245ee6 --- /dev/null +++ b/tests/BLite.Tests/PreparedTransactionWalTests.cs @@ -0,0 +1,208 @@ +using BLite.Core.Storage; +using BLite.Core.Transactions; + +namespace BLite.Tests; + +/// +/// Two-phase commit in single-process mode: the records +/// writes are appended under the commit lock and stay in the WAL until the transaction commits or +/// rolls back. appends only the commit +/// record, so a truncation in between left a commit that recovery could not apply (#188). +/// Disposing the engine does not checkpoint, so reopening it replays the WAL as after a crash. +/// +public class PreparedTransactionWalTests : IDisposable +{ + private readonly string _dir; + private readonly string _dbPath; + + public PreparedTransactionWalTests() + { + _dir = Path.Combine(Path.GetTempPath(), $"blite_prepared_wal_{Guid.NewGuid():N}"); + Directory.CreateDirectory(_dir); + _dbPath = Path.Combine(_dir, "prepared.db"); + } + + public void Dispose() + { + try { Directory.Delete(_dir, recursive: true); } catch { /* best-effort */ } + } + + private long WalLength => new FileInfo(Path.ChangeExtension(_dbPath, ".wal")).Length; + + [Fact] + public async Task PreparedRecords_SurviveACheckpoint_SoTheMarkedCommitIsRecovered() + { + uint other, prepared; + using (var engine = new StorageEngine(_dbPath, PageFileConfig.Default)) + { + other = engine.AllocatePage(); + prepared = engine.AllocatePage(); + await WriteAndCommitAsync(engine, other, 0x11); // gives the checkpoint a page to write + + using (var tx = engine.BeginTransaction()) + { + engine.WritePage(prepared, tx.TransactionId, Filled(engine.PageSize, 0x22)); + Assert.True(await engine.PrepareTransactionAsync(tx.TransactionId)); + + await engine.CheckpointAsync(); + Assert.NotEqual(0, WalLength); // the prepared records are still there + + await engine.MarkTransactionCommittedAsync(tx.TransactionId); + } + AssertPage(engine, prepared, 0x22); + } + + using var reopened = new StorageEngine(_dbPath, PageFileConfig.Default); + AssertPage(reopened, other, 0x11); + AssertPage(reopened, prepared, 0x22); + } + + [Fact] + public async Task PreparedTransaction_HoldsBackTruncation_UntilItCommits() + { + uint other, prepared; + using (var engine = new StorageEngine(_dbPath, PageFileConfig.Default)) + { + other = engine.AllocatePage(); + prepared = engine.AllocatePage(); + await WriteAndCommitAsync(engine, other, 0x11); + + using (var tx = engine.BeginTransaction()) + { + engine.WritePage(prepared, tx.TransactionId, Filled(engine.PageSize, 0x22)); + Assert.True(await tx.PrepareAsync()); + + await engine.CheckpointAsync(); + Assert.NotEqual(0, WalLength); + + await tx.CommitAsync(); + } + + // Committed: the next checkpoint writes its pages and truncates the WAL again. + await engine.CheckpointAsync(); + Assert.Equal(0, WalLength); + } + + using var reopened = new StorageEngine(_dbPath, PageFileConfig.Default); + AssertPage(reopened, other, 0x11); + AssertPage(reopened, prepared, 0x22); + } + + [Fact] + public async Task PreparedTransaction_HoldsBackTruncation_UntilItRollsBack() + { + uint other, prepared; + using (var engine = new StorageEngine(_dbPath, PageFileConfig.Default)) + { + other = engine.AllocatePage(); + prepared = engine.AllocatePage(); + await WriteAndCommitAsync(engine, other, 0x11); + + using (var tx = engine.BeginTransaction()) + { + engine.WritePage(prepared, tx.TransactionId, Filled(engine.PageSize, 0x22)); + Assert.True(await tx.PrepareAsync()); + + await engine.CheckpointAsync(); + Assert.NotEqual(0, WalLength); + + await tx.RollbackAsync(); + } + + // Rolled back: it no longer holds back the truncation, and recovery never applies it. + await WriteAndCommitAsync(engine, other, 0x33); + await engine.CheckpointAsync(); + Assert.Equal(0, WalLength); + } + + using var reopened = new StorageEngine(_dbPath, PageFileConfig.Default); + AssertPage(reopened, other, 0x33); + AssertNotPage(reopened, prepared, 0x22); + } + + [Fact] + public async Task Prepare_WaitsForTheCommitLock_AndWritesNothingWithoutIt() + { + // A prepare appends under the commit lock, as a commit batch or a truncation does, so its + // records never interleave with theirs. Unable to take the lock, it fails without writing. + uint prepared; + using (var engine = new StorageEngine(_dbPath, PageFileConfig.Default)) + { + prepared = engine.AllocatePage(); + using (var tx = engine.BeginTransaction()) + { + engine.WritePage(prepared, tx.TransactionId, Filled(engine.PageSize, 0x22)); + + long walBefore = WalLength; + using (engine.HoldCommitLockForTests()) + { + Assert.False(await engine.PrepareTransactionAsync(tx.TransactionId)); + Assert.False(await engine.PrepareTransactionAsync(tx.TransactionId, CancellationToken.None)); + Assert.Equal(walBefore, WalLength); + } + + Assert.True(await engine.PrepareTransactionAsync(tx.TransactionId)); + Assert.NotEqual(walBefore, WalLength); + await engine.MarkTransactionCommittedAsync(tx.TransactionId); + } + } + + using var reopened = new StorageEngine(_dbPath, PageFileConfig.Default); + AssertPage(reopened, prepared, 0x22); + } + + [Fact] + public async Task Recovery_KeepsThePreparedRecords() + { + uint other, prepared; + using (var engine = new StorageEngine(_dbPath, PageFileConfig.Default)) + { + other = engine.AllocatePage(); + prepared = engine.AllocatePage(); + await WriteAndCommitAsync(engine, other, 0x11); + + using (var tx = engine.BeginTransaction()) + { + engine.WritePage(prepared, tx.TransactionId, Filled(engine.PageSize, 0x22)); + Assert.True(await engine.PrepareTransactionAsync(tx.TransactionId)); + + await engine.RecoverAsync(); + Assert.NotEqual(0, WalLength); + + await engine.MarkTransactionCommittedAsync(tx.TransactionId); + } + } + + using var reopened = new StorageEngine(_dbPath, PageFileConfig.Default); + AssertPage(reopened, other, 0x11); + AssertPage(reopened, prepared, 0x22); + } + + private static void AssertPage(StorageEngine storage, uint page, byte fill) + { + var read = new byte[storage.PageSize]; + storage.ReadPage(page, null, read); + Assert.Equal(Filled(storage.PageSize, fill), read); + } + + private static void AssertNotPage(StorageEngine storage, uint page, byte fill) + { + var read = new byte[storage.PageSize]; + storage.ReadPage(page, null, read); + Assert.NotEqual(Filled(storage.PageSize, fill), read); + } + + private static async Task WriteAndCommitAsync(StorageEngine storage, uint page, byte fill) + { + using var txn = storage.BeginTransaction(); + storage.WritePage(page, txn.TransactionId, Filled(storage.PageSize, fill)); + await txn.CommitAsync(); + } + + private static byte[] Filled(int size, byte value) + { + var buffer = new byte[size]; + buffer.AsSpan(32).Fill(value); // keep a zeroed page header + return buffer; + } +}