Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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<TId, T>` and `DynamicCollection` are exclusively async/await — no blocking synchronous methods. Eliminates accidental blocking calls on thread-pool threads.
Expand Down
7 changes: 7 additions & 0 deletions src/BLite.Core/Storage/PageFile.cs
Original file line number Diff line number Diff line change
Expand Up @@ -132,6 +132,13 @@ public readonly struct PageFileConfig
/// </summary>
public bool AllowMultiProcessAccess { get; init; }

/// <summary>
/// When the write-ahead log is synced to disk. Defaults to <see cref="WalSyncMode.Full"/>: every
/// commit is on disk before it is acknowledged. <see cref="WalSyncMode.Normal"/> skips that sync
/// for lower commit latency and may lose the last commits on a power loss (#260).
/// </summary>
public WalSyncMode WalSync { get; init; }

/// <summary>
/// Returns a copy of this config where any zero-valued numeric fields are replaced
/// with the corresponding values from <see cref="Default"/>.
Expand Down
8 changes: 8 additions & 0 deletions src/BLite.Core/Storage/StorageEngine.GroupCommit.cs
Original file line number Diff line number Diff line change
Expand Up @@ -244,6 +244,14 @@ private async Task ProcessBatchAsync(List<PendingCommit> 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);

Expand Down
10 changes: 10 additions & 0 deletions src/BLite.Core/Storage/StorageEngine.Memory.cs
Original file line number Diff line number Diff line change
Expand Up @@ -204,6 +204,16 @@ private void FreePagesOfCommittedTransaction(ulong transactionId)
}
}

/// <summary>
/// With <see cref="WalSyncMode.Normal"/> 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 <see cref="WalSyncMode.Full"/> every commit is synced
/// already and this does nothing.
/// </summary>
private Task SyncDeferredWalAsync()
=> _wal is Transactions.WriteAheadLog fileWal && !fileWal.SyncsOnFlush ? fileWal.SyncAsync() : Task.CompletedTask;

/// <summary>Drops the frees queued by a transaction that rolled back: its pages stay in use.</summary>
private void DiscardFreesOnCommit(ulong transactionId) => _freeOnCommit.TryRemove(transactionId, out _);

Expand Down
3 changes: 3 additions & 0 deletions src/BLite.Core/Storage/StorageEngine.Recovery.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Expand Down
6 changes: 6 additions & 0 deletions src/BLite.Core/Storage/StorageEngine.Transactions.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand Down
3 changes: 2 additions & 1 deletion src/BLite.Core/Storage/StorageEngine.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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<ulong, ConcurrentDictionary<uint, byte[]>>();
_walIndex = new ConcurrentDictionary<uint, byte[]>();
Expand Down
25 changes: 25 additions & 0 deletions src/BLite.Core/Storage/WalSyncMode.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,25 @@
namespace BLite.Core.Storage;

/// <summary>
/// When the write-ahead log is synced to disk (<see cref="PageFileConfig.WalSync"/>), 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.
/// </summary>
public enum WalSyncMode
{
/// <summary>
/// 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
/// <c>synchronous=FULL</c>.
/// </summary>
Full = 0,

/// <summary>
/// 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
/// <c>synchronous=NORMAL</c> in WAL mode.
/// </summary>
Normal = 1,
}
61 changes: 50 additions & 11 deletions src/BLite.Core/Transactions/WriteAheadLog.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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;

/// <summary>
/// Whether this WAL was opened in multi-process mode
Expand Down Expand Up @@ -106,11 +107,17 @@ public sealed class WriteAheadLog : IWriteAheadLog
/// relaxes its own <see cref="FileShare"/> mode based on the flag. See
/// <c>roadmap/v5/MULTI_PROCESS_WAL.md</c> for the full design.
/// </param>
public WriteAheadLog(string walPath, ICryptoProvider? crypto, int writeTimeoutMs = 5_000, bool allowMultiProcessAccess = false)
/// <param name="syncOnFlush">
/// When <c>true</c> (the default), <see cref="FlushAsync"/> syncs the file to disk. When
/// <c>false</c> it only hands the buffered records to the OS, and the owner calls
/// <see cref="SyncAsync"/> where durability is required (<see cref="WalSyncMode.Normal"/>).
/// </param>
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.
Expand All @@ -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.
Expand Down Expand Up @@ -480,6 +487,17 @@ private void WriteDataRecordInternal(ulong transactionId, uint pageId, ReadOnlyS
}


/// <summary>
/// Writes the buffered records to the file and, unless the log was opened with
/// <c>syncOnFlush: false</c>, 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.
/// </summary>
/// <remarks>
/// <see cref="FileStream.FlushAsync(CancellationToken)"/> 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). <c>Flush(flushToDisk: true)</c> adds
/// the sync (fsync, <c>FlushFileBuffers</c> on Windows).
/// </remarks>
public async Task FlushAsync(CancellationToken ct = default)
{
if (!await _lock.WaitAsync(_writeTimeoutMs, ct))
Expand All @@ -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();
}
}

/// <summary>Whether <see cref="FlushAsync"/> syncs to disk (see the constructor's <c>syncOnFlush</c>).</summary>
public bool SyncsOnFlush => _syncOnFlush;

/// <summary>
/// Writes the buffered records to the file and syncs it to disk, whatever <see cref="SyncsOnFlush"/> says.
/// </summary>
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
Expand Down Expand Up @@ -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)
{
Expand Down
Loading
Loading