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