From 9f742a3ea25ab359160c018a793e2d147562049c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ricardo=20Gro=C3=9F?= Date: Sat, 17 Oct 2026 11:29:32 +0200 Subject: [PATCH] fix(wal): sync the WAL to disk before a commit is acknowledged Every commit path awaits WriteAheadLog.FlushAsync before it acknowledges a commit and frees the pages the commit stopped using. FlushAsync called FileStream.FlushAsync, which hands the 64 KB buffer to the OS and does not fsync (.NET 6 to 10), and the stream has no WriteThrough. Since the async-first rewrite removed the synchronous Flush(flushToDisk: true), no commit synced the WAL. A power loss therefore dropped acknowledged commits while the frees those commits made, written straight to the page file, survived, so documents could come back pointing at freed or reused overflow pages. TruncateAsync had the same gap: the pages a truncation releases are freed right after it returns. FlushAsync now follows the buffer hand-off with Flush(flushToDisk: true), once per group-commit batch, and TruncateAsync syncs the truncation. The lazy checkpoint is unchanged. The sync costs a lot here: on macOS/APFS a sequential single-document commit goes from about 0.13 ms to about 4.5 ms. So there is a switch, PageFileConfig.WalSync, defaulting to WalSyncMode.Full (the behaviour above). WalSyncMode.Normal acknowledges a commit after the hand-off alone and syncs the WAL before a checkpoint copies pages to the data file, before the pages a commit stopped using are freed, and on truncation. A power loss can then lose the most recent commits but leaves a committed state, as SQLite's synchronous=NORMAL does in WAL mode. README's durability line now says which mode guarantees what. Found-By: OT Defect-Class: WAL Co-Authored-By: Claude Opus 5.5 --- README.md | 2 +- src/BLite.Core/Storage/PageFile.cs | 7 + .../Storage/StorageEngine.GroupCommit.cs | 8 + .../Storage/StorageEngine.Memory.cs | 10 + .../Storage/StorageEngine.Recovery.cs | 3 + .../Storage/StorageEngine.Transactions.cs | 6 + src/BLite.Core/Storage/StorageEngine.cs | 3 +- src/BLite.Core/Storage/WalSyncMode.cs | 25 ++ src/BLite.Core/Transactions/WriteAheadLog.cs | 61 ++++- tests/BLite.Tests/WalCommitSyncTests.cs | 229 ++++++++++++++++++ 10 files changed, 341 insertions(+), 13 deletions(-) create mode 100644 src/BLite.Core/Storage/WalSyncMode.cs create mode 100644 tests/BLite.Tests/WalCommitSyncTests.cs diff --git a/README.md b/README.md index 1a868fde..305ceed3 100644 --- a/README.md +++ b/README.md @@ -391,7 +391,7 @@ using var dynSub = engine.GetOrCreateCollection("orders").Watch() ### 🛡️ Transactions & ACID - **Atomic**: Multi-document transactions. -- **Durable**: WAL ensures data safety even in power loss. +- **Durable**: a commit is synced to disk (once per group-commit batch) before it is acknowledged, so it survives a power loss. `PageFileConfig.WalSync = WalSyncMode.Normal` trades that for lower commit latency: the WAL is then synced only before a checkpoint, before the pages a commit stopped using are freed, and on truncation, and a power loss can lose the most recent commits (never corrupt older ones). - **Isolated**: Snapshot isolation allowing concurrent readers and writers. - **Thread-Safe**: Protected with `SemaphoreSlim` to prevent race conditions in concurrent scenarios. - **Async-Only**: All CRUD operations on `DocumentCollection` and `DynamicCollection` are exclusively async/await — no blocking synchronous methods. Eliminates accidental blocking calls on thread-pool threads. diff --git a/src/BLite.Core/Storage/PageFile.cs b/src/BLite.Core/Storage/PageFile.cs index 36b42b63..c368cd49 100644 --- a/src/BLite.Core/Storage/PageFile.cs +++ b/src/BLite.Core/Storage/PageFile.cs @@ -132,6 +132,13 @@ public readonly struct PageFileConfig /// public bool AllowMultiProcessAccess { get; init; } + /// + /// When the write-ahead log is synced to disk. Defaults to : every + /// commit is on disk before it is acknowledged. skips that sync + /// for lower commit latency and may lose the last commits on a power loss (#260). + /// + public WalSyncMode WalSync { get; init; } + /// /// Returns a copy of this config where any zero-valued numeric fields are replaced /// with the corresponding values from . diff --git a/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs b/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs index cc66c764..f56c84da 100644 --- a/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs +++ b/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs @@ -244,6 +244,14 @@ private async Task ProcessBatchAsync(List batch) // The batch is durable and published: release the pages its transactions stopped // using (#233). Last, so that a batch reported as failed never frees anything. + foreach (var commit in batch) + { + if (_freeOnCommit.ContainsKey(commit.TransactionId)) + { + await SyncDeferredWalAsync().ConfigureAwait(false); + break; + } + } foreach (var commit in batch) FreePagesOfCommittedTransaction(commit.TransactionId); diff --git a/src/BLite.Core/Storage/StorageEngine.Memory.cs b/src/BLite.Core/Storage/StorageEngine.Memory.cs index a70bee1f..e8963c5b 100644 --- a/src/BLite.Core/Storage/StorageEngine.Memory.cs +++ b/src/BLite.Core/Storage/StorageEngine.Memory.cs @@ -204,6 +204,16 @@ private void FreePagesOfCommittedTransaction(ulong transactionId) } } + /// + /// With a commit is acknowledged before its WAL records are on + /// disk. Syncs them before an action a power loss must not keep without them: freeing the pages + /// a commit stopped using (a free is written to the data file at once), or a checkpoint copying + /// committed pages to the data file. In every commit is synced + /// already and this does nothing. + /// + private Task SyncDeferredWalAsync() + => _wal is Transactions.WriteAheadLog fileWal && !fileWal.SyncsOnFlush ? fileWal.SyncAsync() : Task.CompletedTask; + /// Drops the frees queued by a transaction that rolled back: its pages stay in use. private void DiscardFreesOnCommit(ulong transactionId) => _freeOnCommit.TryRemove(transactionId, out _); diff --git a/src/BLite.Core/Storage/StorageEngine.Recovery.cs b/src/BLite.Core/Storage/StorageEngine.Recovery.cs index e247b421..b16b48dd 100644 --- a/src/BLite.Core/Storage/StorageEngine.Recovery.cs +++ b/src/BLite.Core/Storage/StorageEngine.Recovery.cs @@ -117,6 +117,9 @@ private async Task CheckpointCoreAsync(bool commitLockHeld, CancellationToken ct foreach (var kvp in toCheckpoint) _checkpointedWalPages.TryAdd(kvp.Key, 0); + // Normal mode: the commits being copied must be on disk before their pages are. + await SyncDeferredWalAsync().ConfigureAwait(false); + foreach (var kvp in toCheckpoint) GetPageFile(kvp.Key, out var physId).WritePage(physId, kvp.Value); } diff --git a/src/BLite.Core/Storage/StorageEngine.Transactions.cs b/src/BLite.Core/Storage/StorageEngine.Transactions.cs index 835e2f8d..261a79dd 100644 --- a/src/BLite.Core/Storage/StorageEngine.Transactions.cs +++ b/src/BLite.Core/Storage/StorageEngine.Transactions.cs @@ -427,6 +427,8 @@ public async Task CommitTransactionAsync(ulong transactionId) // No writes for this transaction, just write commit record await _wal.WriteCommitRecordAsync(transactionId); await _wal.FlushAsync(); + if (_freeOnCommit.ContainsKey(transactionId)) + await SyncDeferredWalAsync(); FreePagesOfCommittedTransaction(transactionId); completed = true; goto Completed; @@ -453,6 +455,8 @@ public async Task CommitTransactionAsync(ulong transactionId) _walCache.TryRemove(transactionId, out _); // The commit is durable: release the pages it stopped using (#233). + if (_freeOnCommit.ContainsKey(transactionId)) + await SyncDeferredWalAsync(); FreePagesOfCommittedTransaction(transactionId); // Check if checkpoint is needed, but defer it until after releasing the lock @@ -611,6 +615,8 @@ public async Task MarkTransactionCommittedAsync(ulong transactionId) } // The commit is durable: release the pages it stopped using (#233). + if (_freeOnCommit.ContainsKey(transactionId)) + await SyncDeferredWalAsync(); FreePagesOfCommittedTransaction(transactionId); // Check if checkpoint is needed, but defer it until after releasing the lock diff --git a/src/BLite.Core/Storage/StorageEngine.cs b/src/BLite.Core/Storage/StorageEngine.cs index 4c5d256b..caeb778a 100644 --- a/src/BLite.Core/Storage/StorageEngine.cs +++ b/src/BLite.Core/Storage/StorageEngine.cs @@ -318,7 +318,8 @@ public StorageEngine(string databasePath, PageFileConfig config) // provider so that WAL records are always encrypted alongside the rest of the database. walCryptoProvider = config.CryptoProvider?.CreateSiblingProvider(3, 0); - _wal = new WriteAheadLog(walPath, walCryptoProvider, config.LockTimeout.WriteTimeoutMs, config.AllowMultiProcessAccess); + _wal = new WriteAheadLog(walPath, walCryptoProvider, config.LockTimeout.WriteTimeoutMs, config.AllowMultiProcessAccess, + syncOnFlush: config.WalSync != WalSyncMode.Normal); walCryptoProvider = null; _walCache = new ConcurrentDictionary>(); _walIndex = new ConcurrentDictionary(); diff --git a/src/BLite.Core/Storage/WalSyncMode.cs b/src/BLite.Core/Storage/WalSyncMode.cs new file mode 100644 index 00000000..9ac97849 --- /dev/null +++ b/src/BLite.Core/Storage/WalSyncMode.cs @@ -0,0 +1,25 @@ +namespace BLite.Core.Storage; + +/// +/// When the write-ahead log is synced to disk (), the trade-off +/// between commit latency and what survives a power loss or an OS crash. A crash of the process +/// alone loses nothing in either mode: the OS keeps what the WAL handed it. +/// +public enum WalSyncMode +{ + /// + /// The default. Every commit (one per group-commit batch) syncs the WAL before it is + /// acknowledged: an acknowledged commit survives a power loss. Like SQLite's + /// synchronous=FULL. + /// + Full = 0, + + /// + /// A commit is acknowledged once the WAL has been handed to the OS. The WAL is synced only + /// before a checkpoint copies committed pages to the data file, before pages a commit stopped + /// using are freed, and when it is truncated. A power loss can lose the most recent + /// acknowledged commits, never more, and leaves a committed state. Like SQLite's + /// synchronous=NORMAL in WAL mode. + /// + Normal = 1, +} diff --git a/src/BLite.Core/Transactions/WriteAheadLog.cs b/src/BLite.Core/Transactions/WriteAheadLog.cs index 407233bd..c67fc3d9 100644 --- a/src/BLite.Core/Transactions/WriteAheadLog.cs +++ b/src/BLite.Core/Transactions/WriteAheadLog.cs @@ -55,6 +55,7 @@ public sealed class WriteAheadLog : IWriteAheadLog // allocation, the .wal-shm sidecar, and stale-PID recovery are owned by // StorageEngine + WalSharedMemory; see roadmap/v5/MULTI_PROCESS_WAL.md. private readonly bool _allowMultiProcessAccess; + private readonly bool _syncOnFlush; /// /// Whether this WAL was opened in multi-process mode @@ -106,11 +107,17 @@ public sealed class WriteAheadLog : IWriteAheadLog /// relaxes its own mode based on the flag. See /// roadmap/v5/MULTI_PROCESS_WAL.md for the full design. /// - public WriteAheadLog(string walPath, ICryptoProvider? crypto, int writeTimeoutMs = 5_000, bool allowMultiProcessAccess = false) + /// + /// When true (the default), syncs the file to disk. When + /// false it only hands the buffered records to the OS, and the owner calls + /// where durability is required (). + /// + public WriteAheadLog(string walPath, ICryptoProvider? crypto, int writeTimeoutMs = 5_000, bool allowMultiProcessAccess = false, bool syncOnFlush = true) { _walPath = walPath ?? throw new ArgumentNullException(nameof(walPath)); _writeTimeoutMs = writeTimeoutMs; _allowMultiProcessAccess = allowMultiProcessAccess; + _syncOnFlush = syncOnFlush; // Treat providers that add no overhead (NullCryptoProvider) as no encryption so // the on-disk format stays byte-for-byte identical to the pre-encryption format. @@ -122,8 +129,8 @@ public WriteAheadLog(string walPath, ICryptoProvider? crypto, int writeTimeoutMs FileAccess.ReadWrite, _allowMultiProcessAccess ? FileShare.ReadWrite : FileShare.None, bufferSize: 64 * 1024); - // REMOVED FileOptions.WriteThrough for SQLite-style lazy checkpointing - // Durability is ensured by explicit Flush() calls + // No FileOptions.WriteThrough, for SQLite-style lazy checkpointing: records are buffered + // and FlushAsync syncs them to disk once per commit (once per group commit batch). // If the WAL file already exists with content and encryption is enabled, // read the file header to derive the encryption key before any I/O. @@ -480,6 +487,17 @@ private void WriteDataRecordInternal(ulong transactionId, uint pageId, ReadOnlyS } + /// + /// Writes the buffered records to the file and, unless the log was opened with + /// syncOnFlush: false, syncs it to disk, so that every record written before the call + /// survives a power loss. Every commit path calls this before it acknowledges a commit. + /// + /// + /// alone only hands the buffer to the OS: + /// it does not sync on .NET 6 to 10, so a power loss dropped acknowledged commits while the + /// pages they freed stayed freed in the page file (#260). Flush(flushToDisk: true) adds + /// the sync (fsync, FlushFileBuffers on Windows). + /// public async Task FlushAsync(CancellationToken ct = default) { if (!await _lock.WaitAsync(_writeTimeoutMs, ct)) @@ -489,14 +507,32 @@ public async Task FlushAsync(CancellationToken ct = default) if (_walStream != null) { await _walStream.FlushAsync(ct); - // FlushAsync doesn't guarantee flushToDisk on all platforms/implementations in the same way as Flush(true) - // but FileStream in .NET 6+ handles this reasonable well. - // For strict durability, we might still want to invoke a sync flush or check platform specifics, - // but typically FlushAsync(ct) is sufficient for "Async" pattern. - // However, FileStream.FlushAsync() acts like flushToDisk=false by default in older .NET? - // Actually, FileStream.Flush() has flushToDisk arg, FlushAsync does not but implementation usually does buffer flush. - // To be safe for WAL, we might care about fsync. - // For now, just FlushAsync(); + if (_syncOnFlush) + _walStream.Flush(flushToDisk: true); + } + } + finally + { + _lock.Release(); + } + } + + /// Whether syncs to disk (see the constructor's syncOnFlush). + public bool SyncsOnFlush => _syncOnFlush; + + /// + /// Writes the buffered records to the file and syncs it to disk, whatever says. + /// + public async Task SyncAsync(CancellationToken ct = default) + { + if (!await _lock.WaitAsync(_writeTimeoutMs, ct)) + throw new TimeoutException("Timed out acquiring WAL write lock."); + try + { + if (_walStream != null) + { + await _walStream.FlushAsync(ct); + _walStream.Flush(flushToDisk: true); } } finally @@ -1035,6 +1071,9 @@ public async Task TruncateAsync(CancellationToken ct = default) _walStream.SetLength(0); _walStream.Position = 0; await _walStream.FlushAsync(ct); + // The caller frees the pages the truncation released right after this returns: a + // truncation lost to a power loss would let recovery replay old images over them. + _walStream.Flush(flushToDisk: true); // Reset so the crypto file header is re-written on the next record write. lock (_cryptoHeaderLock) { diff --git a/tests/BLite.Tests/WalCommitSyncTests.cs b/tests/BLite.Tests/WalCommitSyncTests.cs new file mode 100644 index 00000000..9983f93b --- /dev/null +++ b/tests/BLite.Tests/WalCommitSyncTests.cs @@ -0,0 +1,229 @@ +using System.Reflection; +using BLite.Bson; +using BLite.Core.Storage; +using BLite.Core.Transactions; +using BLite.Shared; +using Microsoft.Win32.SafeHandles; + +namespace BLite.Tests; + +/// +/// A commit is acknowledged only once its WAL records are synced to disk (#260). +/// , which the WAL used to call, hands the +/// buffer to the OS and does not sync, so a power loss dropped acknowledged commits while the pages +/// they freed stayed freed. A power loss cannot be simulated here; these tests record the calls the +/// WAL makes on its stream and check the order: every byte written for a commit is followed by a +/// sync (Flush(flushToDisk: true)) before the commit returns, and a truncation is synced +/// before the checkpoint returns. The WAL crash test (Oracles/WalCrashTests) checks what +/// survives each crash point. +/// +public sealed class WalCommitSyncTests : IDisposable +{ + private const BindingFlags Private = BindingFlags.NonPublic | BindingFlags.Instance; + private static readonly FieldInfo s_engineWal = typeof(StorageEngine).GetField("_wal", Private)!; + private static readonly FieldInfo s_walStream = typeof(WriteAheadLog).GetField("_walStream", Private)!; + + private readonly string _dir; + + public WalCommitSyncTests() + { + _dir = Path.Combine(Path.GetTempPath(), $"blite_walsync_{Guid.NewGuid():N}"); + Directory.CreateDirectory(_dir); + } + + public void Dispose() + { + try { Directory.Delete(_dir, recursive: true); } catch { } + } + + /// Opens a context whose WAL writes through a . + private (TestDbContext Db, RecordingWalStream Wal) OpenRecorded(WalSyncMode mode = WalSyncMode.Full) + { + var db = new TestDbContext(Path.Combine(_dir, "db.blite"), PageFileConfig.Default with { WalSync = mode }); + var wal = (WriteAheadLog)s_engineWal.GetValue(db.Storage)!; + var original = (FileStream)s_walStream.GetValue(wal)!; + original.Flush(); + var recording = new RecordingWalStream(original.SafeFileHandle, original.Position); + // Both streams share the handle: the WAL disposes the new one, the old one must not close it. + GC.SuppressFinalize(original); + s_walStream.SetValue(wal, recording); + return (db, recording); + } + + private static void AssertSyncedAfterLastWrite(RecordingWalStream wal, string what) + { + var events = wal.Events; + var lastWrite = events.FindLastIndex(e => e is "write" or "set-length"); + Assert.True(lastWrite >= 0, $"{what}: nothing was written to the WAL"); + var lastSync = events.FindLastIndex(e => e == "sync"); + Assert.True(lastSync > lastWrite, + $"{what} returned before the WAL was synced to disk; WAL calls since the open: {string.Join(", ", events)}"); + } + + [Fact] + public async Task AutoCommit_Insert_Returns_Only_After_The_Wal_Is_Synced() + { + var (db, wal) = OpenRecorded(); + using (db) + { + await db.Users.InsertAsync(new User { Id = ObjectId.NewObjectId(), Name = "a", Age = 1 }); + AssertSyncedAfterLastWrite(wal, "InsertAsync"); + } + } + + [Fact] + public async Task Transaction_CommitAsync_Returns_Only_After_The_Wal_Is_Synced() + { + var (db, wal) = OpenRecorded(); + using (db) + { + using var tx = await db.BeginTransactionAsync(); + await db.Users.InsertAsync(new User { Id = ObjectId.NewObjectId(), Name = "a", Age = 1 }, tx); + await db.Users.InsertAsync(new User { Id = ObjectId.NewObjectId(), Name = "b", Age = 2 }, tx); + wal.Events.Clear(); + + await tx.CommitAsync(); + + AssertSyncedAfterLastWrite(wal, "CommitAsync"); + } + } + + [Fact] + public async Task Checkpoint_Syncs_The_Wal_Truncation_Before_Returning() + { + var (db, wal) = OpenRecorded(); + using (db) + { + await db.Users.InsertAsync(new User { Id = ObjectId.NewObjectId(), Name = "a", Age = 1 }); + wal.Events.Clear(); + + await db.CheckpointAsync(); + + Assert.Contains("set-length", wal.Events); + AssertSyncedAfterLastWrite(wal, "CheckpointAsync"); + } + } + + [Fact] + public async Task Normal_Mode_Acknowledges_A_Commit_Without_Syncing() + { + var (db, wal) = OpenRecorded(WalSyncMode.Normal); + using (db) + { + await db.Users.InsertAsync(new User { Id = ObjectId.NewObjectId(), Name = "a", Age = 1 }); + + Assert.Contains("write", wal.Events); + Assert.DoesNotContain("sync", wal.Events); + } + } + + [Fact] + public async Task Normal_Mode_Syncs_Before_Freeing_The_Pages_A_Commit_Released() + { + var (db, wal) = OpenRecorded(WalSyncMode.Normal); + using (db) + { + // An overflow chain: deleting the document frees its pages when the delete commits. + var big = new User { Id = ObjectId.NewObjectId(), Name = new string('x', 60_000), Age = 1 }; + await db.Users.InsertAsync(big); + Assert.DoesNotContain("sync", wal.Events); + + await db.Users.DeleteAsync(big.Id); + + AssertSyncedAfterLastWrite(wal, "a delete that frees pages, in Normal mode,"); + } + } + + [Fact] + public async Task Normal_Mode_Syncs_Before_A_Checkpoint_Copies_Pages() + { + var (db, wal) = OpenRecorded(WalSyncMode.Normal); + using (db) + { + await db.Users.InsertAsync(new User { Id = ObjectId.NewObjectId(), Name = "a", Age = 1 }); + Assert.DoesNotContain("sync", wal.Events); + + await db.CheckpointAsync(); + + var firstSync = wal.Events.IndexOf("sync"); + var truncation = wal.Events.IndexOf("set-length"); + Assert.True(firstSync >= 0 && truncation > firstSync, + $"the WAL must be synced before the checkpoint writes its pages; WAL calls: {string.Join(", ", wal.Events)}"); + AssertSyncedAfterLastWrite(wal, "CheckpointAsync in Normal mode"); + } + } + + /// + /// An unbuffered stream over the WAL's handle that records, in order, each write it passes to + /// the OS, each truncation, each flush that only hands bytes to the OS (flush) and each + /// flush to disk (sync). + /// + private sealed class RecordingWalStream : FileStream + { + private readonly object _gate = new(); + private readonly List _events = new(); + + public RecordingWalStream(SafeFileHandle handle, long position) : base(handle, FileAccess.ReadWrite, bufferSize: 0) + { + Position = position; + } + + public List Events + { + get { lock (_gate) return _events; } + } + + private void Add(string e) + { + lock (_gate) _events.Add(e); + } + + // FileStream routes a derived stream's span writes through the array overload, so the + // array overload is the one that reaches the OS. + public override void Write(ReadOnlySpan buffer) + { + var copy = buffer.ToArray(); + Write(copy, 0, copy.Length); + } + + public override void Write(byte[] buffer, int offset, int count) + { + base.Write(buffer, offset, count); + Add("write"); + } + + public override void WriteByte(byte value) => Write(new[] { value }); + + public override ValueTask WriteAsync(ReadOnlyMemory buffer, CancellationToken cancellationToken = default) + { + Write(buffer.Span); + return ValueTask.CompletedTask; + } + + public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + { + Write(buffer, offset, count); + return Task.CompletedTask; + } + + public override void SetLength(long value) + { + base.SetLength(value); + Add("set-length"); + } + + public override void Flush() => Flush(false); + + public override void Flush(bool flushToDisk) + { + base.Flush(flushToDisk); + Add(flushToDisk ? "sync" : "flush"); + } + + public override Task FlushAsync(CancellationToken cancellationToken) + { + Add("flush"); + return base.FlushAsync(cancellationToken); + } + } +}