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; + } +}