From d20f57bf03930e6374f1d9293358dcaf6a8c4330 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ricardo=20Gro=C3=9F?= Date: Sat, 17 Oct 2026 15:42:57 +0200 Subject: [PATCH 1/5] fix(storage): detect write-write conflicts at commit (#304) Two transactions open at the same time that changed the same page both committed, and the second one's page copies replaced the first one's changes. Every page now carries the sequence number of the commit that last published it; a transaction records the version of each page it first read (or first wrote), and group commit rejects, before writing any WAL record, a commit whose pages moved on since or that an earlier commit of the same batch also writes. The first committer wins; the other gets BLiteTransactionConflictException and is rolled back. Every object that writes to a collection now shares one write lock (StorageEngine.GetCollectionWriteLock): before, each DynamicCollection had its own, so concurrent auto-commit inserts through different sessions ran on the same pages and most of them were lost. README no longer claims snapshot isolation. Co-Authored-By: Claude Opus 5.5 --- README.md | 8 +- .../BLiteTransactionConflictException.cs | 44 +++ .../Collections/DocumentCollection.cs | 2 +- src/BLite.Core/DynamicCollection.cs | 6 +- .../Storage/StorageEngine.Conflicts.cs | 149 +++++++++ .../Storage/StorageEngine.GroupCommit.cs | 37 ++- src/BLite.Core/Storage/StorageEngine.Pages.cs | 17 +- .../Storage/StorageEngine.Transactions.cs | 28 +- src/BLite.Core/Storage/StorageEngine.cs | 22 +- src/BLite.Core/Transactions/Transaction.cs | 13 +- tests/BLite.Tests/TransactionConflictTests.cs | 282 ++++++++++++++++++ 11 files changed, 587 insertions(+), 21 deletions(-) create mode 100644 src/BLite.Core/BLiteTransactionConflictException.cs create mode 100644 src/BLite.Core/Storage/StorageEngine.Conflicts.cs create mode 100644 tests/BLite.Tests/TransactionConflictTests.cs diff --git a/README.md b/README.md index 1a868fde..27828dec 100644 --- a/README.md +++ b/README.md @@ -33,7 +33,7 @@ Most embedded databases for .NET are either wrappers around C libraries (SQLite, - **Zero Allocation**: I/O and interaction paths use `Span` and `stackalloc`. No heap allocations for reads/writes. - **Type-Safe**: No reflection. All serialization code is generated at compile-time. - **Developer Experience**: Full LINQ provider (`IQueryable`) that feels like Entity Framework but runs on bare metal. -- **Reliable**: Full ACID transactions with Write-Ahead Logging (WAL) and Snapshot Isolation. +- **Reliable**: ACID transactions with Write-Ahead Logging (WAL). Concurrent transactions never overwrite each other's commits: the first to commit wins. --- @@ -392,7 +392,7 @@ using var dynSub = engine.GetOrCreateCollection("orders").Watch() ### 🛡️ Transactions & ACID - **Atomic**: Multi-document transactions. - **Durable**: WAL ensures data safety even in power loss. -- **Isolated**: Snapshot isolation allowing concurrent readers and writers. +- **Isolated**: Read committed: a transaction sees its own writes and what other transactions have committed. Two transactions that change the same page cannot both commit: the first one wins, the second one's `CommitAsync` throws `BLiteTransactionConflictException` and rolls it back, so run it again in a new transaction. - **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. - **Implicit Transactions**: Use `SaveChangesAsync()` for automatic transaction management. @@ -908,7 +908,7 @@ if (doc is not null) ## 🔌 BLiteSession — Per-Connection Isolation *(v3.8.0)* -When a single `BLiteEngine` is shared across multiple concurrent clients (e.g. inside a custom server layer), `BLiteSession` provides **per-connection isolated transaction contexts**. Each session carries its own transaction state so independent callers cannot interfere with each other. +When a single `BLiteEngine` is shared across multiple concurrent clients (e.g. inside a custom server layer), `BLiteSession` provides **per-connection isolated transaction contexts**. Each session carries its own transaction state. Sessions are not serialized against each other: when two sessions' open transactions change the same page, the first to commit wins and the other's `CommitAsync` throws `BLiteTransactionConflictException` (see Transactions & ACID above). Open a session with `engine.OpenSession()`. Disposing the session automatically rolls back any uncommitted transaction. @@ -1164,7 +1164,7 @@ We are actively building the core. Here is where we stand: - ✅ **CDC Watch on `DynamicCollection` (v3.6.0)**: `DynamicCollection.Watch()` adds real-time change streams to the schema-less API — previously only available on typed `DocumentCollection`. - ✅ **HNSW Vector Search Correctness (v3.6.2)**: full correctness pass — fixes `AllocateNode` overflow, neighbor link integrity, `SelectNeighbors` heuristic, random level distribution (`mL = 1/ln(M)`), and index persistence across close/reopen. 12 dedicated edge-case tests added. - ✅ **OLAP GroupBy Push-down (v3.7.0)**: aggregate terminal operators (`Count`, `Sum`, `Min`, `Max`, `Average`) are pushed down to the storage layer via `BTreeQueryProvider.TryBsonAggregate`, eliminating unnecessary document materialization for large scans. -- ✅ **BLiteSession — Per-Connection Isolation (v3.8.0)**: `BLiteEngine.OpenSession()` returns a `BLiteSession` with its own isolated transaction context. Multiple sessions on the same engine run independent concurrent transactions. Disposing a session rolls back any uncommitted transaction automatically. +- ✅ **BLiteSession — Per-Connection Isolation (v3.8.0)**: `BLiteEngine.OpenSession()` returns a `BLiteSession` with its own isolated transaction context. Multiple sessions on the same engine run independent concurrent transactions; when two change the same page, the first to commit wins and the other fails with `BLiteTransactionConflictException`. Disposing a session rolls back any uncommitted transaction automatically. - ✅ **Multi-File Storage Layout (v3.8.0)**: `PageFileConfig.Server(dbPath)` configures separate files for WAL, index data, and per-collection data. `BLiteMigration.ToMultiFile()` / `ToSingleFile()` migrate existing databases between layouts, preserving all documents, KV entries (including TTL), and index definitions. - ✅ **Non-Blocking Checkpoints (v3.8.0)**: checkpoint and metadata writes are deferred to avoid blocking the hot path. `PageFile` uses `Lazy` for collection file initialization and `ReaderWriterLockSlim` to fix concurrent read/write races. - ✅ **Async-Only CRUD API (v4.0.0 — Breaking Change)**: Synchronous data methods (`Insert`, `Update`, `Delete`, `FindById`, `FindAll`, `Find`, and all bulk variants) have been **removed** from `DocumentCollection` and `DynamicCollection`. All data operations are now exclusively async — this eliminates accidental blocking calls on thread-pool threads, simplifies the internal code paths, and enforces correct async usage throughout the entire stack. diff --git a/src/BLite.Core/BLiteTransactionConflictException.cs b/src/BLite.Core/BLiteTransactionConflictException.cs new file mode 100644 index 00000000..e834ae09 --- /dev/null +++ b/src/BLite.Core/BLiteTransactionConflictException.cs @@ -0,0 +1,44 @@ +using System; + +namespace BLite.Core; + +/// +/// Thrown by a commit when another transaction has committed a change to a page this transaction +/// also changed, after this transaction read that page. The first transaction to commit wins; this +/// one is rolled back ( becomes +/// ) so that neither commit silently overwrites +/// the other (#304). +/// +/// +/// +/// Nothing of the rejected transaction is written. Run its work again in a new transaction to retry: +/// it then reads the other transaction's committed changes first. +/// +/// +/// Conflicts are detected per page, not per document: two transactions that change different +/// documents stored on the same data page, or whose index entries share a B-Tree leaf, conflict too. +/// Writes made without an explicit transaction commit on their own and retry nothing; they can only +/// make a caller's open transaction fail, never be lost to it. +/// +/// +/// Derives from , so code that already catches that type +/// for commit failures keeps working. +/// +/// +public class BLiteTransactionConflictException : InvalidOperationException +{ + /// Creates the exception for and the first conflicting page. + public BLiteTransactionConflictException(ulong transactionId, uint pageId) + : base($"Transaction {transactionId} conflicts with a transaction that committed first: page {pageId} " + + "was changed by both. The transaction was rolled back; run it again to retry.") + { + TransactionId = transactionId; + PageId = pageId; + } + + /// The transaction that was rolled back. + public ulong TransactionId { get; } + + /// The first page found that both transactions changed. + public uint PageId { get; } +} diff --git a/src/BLite.Core/Collections/DocumentCollection.cs b/src/BLite.Core/Collections/DocumentCollection.cs index 95189124..fef57039 100644 --- a/src/BLite.Core/Collections/DocumentCollection.cs +++ b/src/BLite.Core/Collections/DocumentCollection.cs @@ -848,7 +848,7 @@ internal DocumentCollection(StorageEngine storage, ITransactionHolder transactio _transactionHolder = transactionHolder ?? throw new ArgumentNullException(nameof(transactionHolder)); _mapper = mapper ?? throw new ArgumentNullException(nameof(mapper)); _collectionName = collectionName ?? _mapper.CollectionName; - _collectionLock = _storage.CreateCollectionWriteLock(); + _collectionLock = _storage.GetCollectionWriteLock(_collectionName); // Multi-process mode: create or open the catalog entry, schema and index roots of the // collection under the cross-process open lock (#216). No-op in single-process mode. diff --git a/src/BLite.Core/DynamicCollection.cs b/src/BLite.Core/DynamicCollection.cs index 4e144373..4762fb83 100644 --- a/src/BLite.Core/DynamicCollection.cs +++ b/src/BLite.Core/DynamicCollection.cs @@ -46,7 +46,8 @@ public sealed class DynamicCollection : IDisposable private readonly BTreeIndex _primaryIndex; private readonly string _collectionName; private readonly BsonIdType _idType; - private readonly SemaphoreSlim _collectionLock = new(1, 1); + // Shared with every other object that writes to this collection (#304), see StorageEngine.GetCollectionWriteLock. + private readonly SemaphoreSlim _collectionLock; private int WriteLockTimeoutMs => _storage.LockTimeout.WriteTimeoutMs; private readonly FreeSpaceIndex _fsi; // Cached delegate — avoids per-call closure allocation when passed to _fsi.FindPage. @@ -116,6 +117,7 @@ internal DynamicCollection(StorageEngine storage, ITransactionHolder transaction _storage = storage ?? throw new ArgumentNullException(nameof(storage)); _transactionHolder = transactionHolder ?? throw new ArgumentNullException(nameof(transactionHolder)); _collectionName = collectionName ?? throw new ArgumentNullException(nameof(collectionName)); + _collectionLock = _storage.GetCollectionWriteLock(_collectionName); _idType = idType; _maxDocumentSizeForSinglePage = _storage.PageSize - 128; _fsi = freeSpaceIndex ?? new FreeSpaceIndex(_storage.PageSize); @@ -2870,6 +2872,6 @@ public void Dispose() { _retentionTimer?.Dispose(); _retentionTimer = null; - _collectionLock.Dispose(); + // The write lock belongs to the engine and is shared with other instances: not disposed here. } } diff --git a/src/BLite.Core/Storage/StorageEngine.Conflicts.cs b/src/BLite.Core/Storage/StorageEngine.Conflicts.cs new file mode 100644 index 00000000..86a7dc81 --- /dev/null +++ b/src/BLite.Core/Storage/StorageEngine.Conflicts.cs @@ -0,0 +1,149 @@ +using System.Collections.Concurrent; + +namespace BLite.Core.Storage; + +public sealed partial class StorageEngine +{ + // ------------------------------------------------------------------------- + // Write-write conflict detection (#304) + // + // Each transaction writes private copies of the pages it changes and a commit publishes them + // whole. Without a check, two transactions that change the same page both commit and the + // second one's copy, made from the page as it was before the first commit, silently replaces + // the first one's changes. + // + // Every page has a version: the sequence number of the last commit that published it. A + // transaction remembers, for each page, the version it first saw (at its first read from the + // committed state, or at its first write if it never read the page). At commit, inside the + // group-commit batch and before any WAL record is written, a page whose version has moved on + // since, or that an earlier commit of the same batch also writes, makes the commit fail with + // BLiteTransactionConflictException. The first committer wins. + // + // Orderings that keep the check from missing a conflict (it may report a spurious one, which + // the caller retries, but never let a lost write through): + // • A read captures the version before it copies the page; a commit publishes the image + // before it raises the version. A reader never pairs a new version with old data. + // • Pruning and a truncation by another process raise the floor every page version is + // compared against, so a version can only grow. + // ------------------------------------------------------------------------- + + /// Sequence number of the last commit (or replay, or floor raise) that changed a page version. + private long _commitSequence; + + /// Version of every page that has no entry in . + private long _pageVersionFloor; + + /// Last commit sequence that published each page, above . + private readonly ConcurrentDictionary _pageVersions = new(); + + /// Per active transaction: the version of each page it first read or wrote. + private readonly ConcurrentDictionary> _baseVersions = new(); + + /// Number of page versions above which a commit tries to forget them. + private const int PageVersionPruneThreshold = 4096; + + private long GetPageVersion(uint pageId) + { + long floor = Volatile.Read(ref _pageVersionFloor); + return _pageVersions.TryGetValue(pageId, out long version) && version > floor ? version : floor; + } + + /// + /// Records , captured before the page was copied, as the version + /// based its view of on, unless it + /// saw the page before. Reads without a transaction, or by one that is no longer active, record nothing. + /// + private void RecordRead(ulong? transactionId, uint pageId, long version) + { + if (!transactionId.HasValue || transactionId.Value == 0) + return; + + if (!_baseVersions.TryGetValue(transactionId.Value, out var bases)) + { + if (!_activeTransactions.ContainsKey(transactionId.Value)) + return; + bases = _baseVersions.GetOrAdd(transactionId.Value, _ => new ConcurrentDictionary()); + } + bases.TryAdd(pageId, version); + } + + /// Records the version of a page a transaction writes without having read it first. + private void RecordFirstWrite(ulong transactionId, uint pageId) + { + var bases = _baseVersions.GetOrAdd(transactionId, _ => new ConcurrentDictionary()); + if (!bases.ContainsKey(pageId)) + bases.TryAdd(pageId, GetPageVersion(pageId)); + } + + /// + /// Returns the first page of that another commit has published since + /// first saw it, or that + /// holds; null if there is none. Called under _commitLock. + /// + private uint? FindConflict(ulong transactionId, ConcurrentDictionary pages, HashSet? writtenEarlierInBatch) + { + _baseVersions.TryGetValue(transactionId, out var bases); + foreach (var pageId in pages.Keys) + { + if (writtenEarlierInBatch != null && writtenEarlierInBatch.Contains(pageId)) + return pageId; + if (bases != null && bases.TryGetValue(pageId, out long baseVersion) && GetPageVersion(pageId) != baseVersion) + return pageId; + } + return null; + } + + /// + /// Raises the version of every page in to a new commit sequence. + /// Call after their new images are published. Called under _commitLock. + /// + private void PublishPageVersions(IEnumerable pageIds) + { + long sequence = Interlocked.Increment(ref _commitSequence); + foreach (var pageId in pageIds) + _pageVersions[pageId] = sequence; + } + + /// + /// Makes every page version newer than any a transaction may have recorded, e.g. after another + /// process truncated the shared WAL and this engine discarded what it held from it. + /// + private void RaisePageVersionFloor() + { + Volatile.Write(ref _pageVersionFloor, Interlocked.Increment(ref _commitSequence)); + _pageVersions.Clear(); + } + + /// + /// Forgets the page versions once no active transaction has recorded any, so that the map does + /// not grow with every page ever written. Called under _commitLock. + /// + private void TryPrunePageVersions() + { + if (_pageVersions.Count <= PageVersionPruneThreshold) + return; + + foreach (var transactionId in _baseVersions.Keys) + { + if (!_activeTransactions.ContainsKey(transactionId)) + _baseVersions.TryRemove(transactionId, out _); + } + + if (_baseVersions.IsEmpty) + RaisePageVersionFloor(); + } + + /// + /// Throws if + /// conflicts with a commit published since it first saw one of its pages. For the direct commit + /// paths, which run under _commitLock outside group commit. + /// + private void ThrowIfConflicting(ulong transactionId, ConcurrentDictionary? pages) + { + if (pages != null && FindConflict(transactionId, pages, null) is uint pageId) + throw new BLiteTransactionConflictException(transactionId, pageId); + } + + /// Number of active transactions with recorded page versions (diagnostics and tests). + internal int TrackedTransactionCount => _baseVersions.Count; +} diff --git a/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs b/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs index cc66c764..8683f5b7 100644 --- a/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs +++ b/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs @@ -115,6 +115,8 @@ private async Task ProcessBatchAsync(List batch) { bool needsCheckpoint = false; Exception? failure = null; + // Commits rejected because a commit before them changed one of their pages (#304). + Dictionary? conflicts = null; // Acquire the commit lock once for the entire batch. // If this times out, signal all waiters with the exception so they don't hang. @@ -171,8 +173,28 @@ private async Task ProcessBatchAsync(List batch) wal.SeekToSharedEnd(_shm!.ReadWalEndOffset()); } + // First committer wins (#304): a commit whose pages another commit has published since + // it read them, or that an earlier commit of this batch also writes, is rejected before + // any of its records reach the WAL. In batch order, so each commit sees the ones before it. + HashSet? writtenInBatch = null; foreach (var commit in batch) { + if (commit.Pages == null || commit.Pages.IsEmpty) + continue; + if (FindConflict(commit.TransactionId, commit.Pages, writtenInBatch) is uint conflictPage) + { + (conflicts ??= new()).Add(commit, new BLiteTransactionConflictException(commit.TransactionId, conflictPage)); + continue; + } + writtenInBatch ??= new HashSet(); + foreach (var pageId in commit.Pages.Keys) + writtenInBatch.Add(pageId); + } + + foreach (var commit in batch) + { + if (conflicts != null && conflicts.ContainsKey(commit)) + continue; if (commit.Pages == null || commit.Pages.IsEmpty) { // Read-only or empty transaction: just a commit tombstone. @@ -204,12 +226,16 @@ private async Task ProcessBatchAsync(List batch) // Promote to WAL index (makes pages visible to readers). foreach (var commit in batch) { + if (conflicts != null && conflicts.ContainsKey(commit)) + continue; if (commit.Pages != null) { // Index first, then drop the cache: a page must never be in neither // (FreePage relies on finding its WAL image in one of them). foreach (var kvp in commit.Pages) _walIndex[kvp.Key] = kvp.Value; + // Images first, then versions: a reader never pairs a new version with old data (#304). + PublishPageVersions(commit.Pages.Keys); _walCache.TryRemove(commit.TransactionId, out _); } } @@ -245,7 +271,12 @@ 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) - FreePagesOfCommittedTransaction(commit.TransactionId); + { + if (conflicts == null || !conflicts.ContainsKey(commit)) + FreePagesOfCommittedTransaction(commit.TransactionId); + } + + TryPrunePageVersions(); // Check if checkpoint is needed, but defer until after releasing the lock needsCheckpoint = newWalEnd > MaxWalSize; @@ -267,7 +298,9 @@ private async Task ProcessBatchAsync(List batch) // Signal all waiters outside the lock so their continuations run freely. foreach (var commit in batch) { - if (failure != null) + if (conflicts != null && conflicts.TryGetValue(commit, out var conflict)) + commit.Completion.TrySetException(conflict); + else if (failure != null) commit.Completion.TrySetException(failure); else commit.Completion.TrySetResult(true); diff --git a/src/BLite.Core/Storage/StorageEngine.Pages.cs b/src/BLite.Core/Storage/StorageEngine.Pages.cs index 546e1913..f16db6d5 100644 --- a/src/BLite.Core/Storage/StorageEngine.Pages.cs +++ b/src/BLite.Core/Storage/StorageEngine.Pages.cs @@ -26,7 +26,11 @@ public void ReadPage(uint pageId, ulong? transactionId, Span destination) uncommittedData.AsSpan(0, length).CopyTo(destination); return; } - + + // From here on the page comes from the committed state: remember which version this + // transaction bases its view on (#304), captured before the copy. + RecordRead(transactionId, pageId, GetPageVersion(pageId)); + // 2. Check WAL index (committed but not checkpointed). In multi-process mode the image // is served only if another process has not rewritten the page since (#241); the // check comes first because catching up may replace the image. @@ -47,7 +51,7 @@ public void ReadPage(uint pageId, ulong? transactionId, Span destination) pageData.AsSpan(0, length).CopyTo(destination); return; } - + // 3. Read committed baseline from PageFile GetPageFile(pageId, out var physId).ReadPage(physId, destination); } @@ -77,6 +81,9 @@ public void ReadPageHeader(uint pageId, ulong? transactionId, Span destina return; } + // Committed state from here on: record the version before the copy (#304), see ReadPage. + RecordRead(transactionId, pageId, GetPageVersion(pageId)); + // Multi-process: the image is served only if it is still current (#241), see ReadPage. if (LocalImageIsCurrent(pageId) && _walIndex.TryGetValue(pageId, out var committedData)) { @@ -122,6 +129,9 @@ public void WritePage(uint pageId, ulong transactionId, ReadOnlySpan data) } else { + // A page written without a read first (e.g. a newly allocated one) is based on the + // version current now; a page read first keeps the version of that read (#304). + RecordFirstWrite(transactionId, pageId); txnPages[pageId] = data.ToArray(); } } @@ -169,6 +179,9 @@ public ValueTask ReadPageAsync(uint pageId, ulong? transactionId, Memory d #endif } + // Committed state from here on: record the version before the copy (#304), see ReadPage. + RecordRead(transactionId, pageId, GetPageVersion(pageId)); + // 2. WAL index — committed but not yet checkpointed (synchronous, no I/O). Multi-process: // the image is served only if it is still current (#241), see ReadPage. if (LocalImageIsCurrent(pageId) && _walIndex.TryGetValue(pageId, out var committedData)) diff --git a/src/BLite.Core/Storage/StorageEngine.Transactions.cs b/src/BLite.Core/Storage/StorageEngine.Transactions.cs index 835e2f8d..f6069028 100644 --- a/src/BLite.Core/Storage/StorageEngine.Transactions.cs +++ b/src/BLite.Core/Storage/StorageEngine.Transactions.cs @@ -133,6 +133,11 @@ private bool TryCatchUpWithSharedWal(Transactions.WriteAheadLog wal) // GetMinReaderOffset() safe boundary for this page (Phase 6). _walOffsets?[pid] = walOffset; } + + // Other processes' commits move these pages on: a local transaction that read them + // before now conflicts at its commit (#304). Images first, then versions. + if (newPages.Count > 0) + PublishPageVersions(newPages.Select(p => p.pageId)); } Volatile.Write(ref _lastKnownWalEndOffset, shmEnd); @@ -265,6 +270,10 @@ private void DiscardSharedWalState(long generation) Volatile.Write(ref _knownWalGeneration, generation); Volatile.Write(ref _lastKnownWalEndOffset, 0); } + + // Commits other processes made in the old WAL may never be replayed here: every page + // version moves on, so a local transaction that read a page before conflicts (#304). + RaisePageVersionFloor(); } /// @@ -310,7 +319,11 @@ private void DiscardSharedWalState(long generation) /// Forgets a transaction once it has committed or rolled back through the ulong /// overloads, which callers use directly without going through the ones. /// - private void CompleteTransaction(ulong transactionId) => _activeTransactions.TryRemove(transactionId, out _); + private void CompleteTransaction(ulong transactionId) + { + _activeTransactions.TryRemove(transactionId, out _); + _baseVersions.TryRemove(transactionId, out _); + } public async Task CommitTransactionAsync(Transaction transaction, CancellationToken ct = default) { @@ -432,6 +445,9 @@ public async Task CommitTransactionAsync(ulong transactionId) goto Completed; } + // First committer wins (#304): nothing is written if another commit changed a page first. + ThrowIfConflicting(transactionId, pages); + // 1. Write all changes to WAL (from cache, not writeSet!) await _wal.WriteBeginRecordAsync(transactionId); @@ -450,6 +466,7 @@ public async Task CommitTransactionAsync(ulong transactionId) { _walIndex[kvp.Key] = kvp.Value; } + PublishPageVersions(pages.Keys); _walCache.TryRemove(transactionId, out _); // The commit is durable: release the pages it stopped using (#233). @@ -596,17 +613,22 @@ public async Task MarkTransactionCommittedAsync(ulong transactionId) _commitLock.Wait(); try { + // First committer wins (#304): without a commit record the prepared records are never applied. + _walCache.TryGetValue(transactionId, out var pages); + ThrowIfConflicting(transactionId, pages); + await _wal.WriteCommitRecordAsync(transactionId); await _wal.FlushAsync(); - + // Move from cache to WAL index. Index first, then drop the cache: a page must // never be in neither (FreePage relies on finding its WAL image). - if (_walCache.TryGetValue(transactionId, out var pages)) + if (pages != null) { foreach (var kvp in pages) { _walIndex[kvp.Key] = kvp.Value; } + PublishPageVersions(pages.Keys); _walCache.TryRemove(transactionId, out _); } diff --git a/src/BLite.Core/Storage/StorageEngine.cs b/src/BLite.Core/Storage/StorageEngine.cs index 4c5d256b..d23938a8 100644 --- a/src/BLite.Core/Storage/StorageEngine.cs +++ b/src/BLite.Core/Storage/StorageEngine.cs @@ -150,13 +150,23 @@ public sealed partial class StorageEngine : IDisposable internal bool IsReadOnly => _config.Access == System.IO.MemoryMappedFiles.MemoryMappedFileAccess.Read; /// - /// Returns the semaphore a must hold for - /// write operations. In single-file mode every collection of this engine gets the same - /// instance, because they share the free-space index and the page allocator - /// (see ). In collection-per-file mode each collection gets its - /// own, preserving write concurrency between collections. + /// Returns the semaphore a collection must hold for write operations. In single-file mode + /// every collection of this engine gets the same instance, because they share the free-space + /// index and the page allocator (see ). In collection-per-file + /// mode each collection name gets its own, preserving write concurrency between collections. + /// + /// Every object that writes to the same collection gets the same instance: a typed + /// , the engine's + /// and a session's own for + /// that name. Before, each had a lock of its own, so writes + /// through a session and through the engine (or another session) ran concurrently on the same + /// pages, and all but one of them were lost (#304). + /// /// - internal SemaphoreSlim CreateCollectionWriteLock() => _dataWriteLock ?? new SemaphoreSlim(1, 1); + internal SemaphoreSlim GetCollectionWriteLock(string collectionName) + => _dataWriteLock ?? _collectionWriteLocks.GetOrAdd(collectionName, _ => new SemaphoreSlim(1, 1)); + + private readonly ConcurrentDictionary _collectionWriteLocks = new(StringComparer.OrdinalIgnoreCase); /// /// Multi-process WAL coordination sidecar; null when diff --git a/src/BLite.Core/Transactions/Transaction.cs b/src/BLite.Core/Transactions/Transaction.cs index 41d6b722..bc105d0d 100644 --- a/src/BLite.Core/Transactions/Transaction.cs +++ b/src/BLite.Core/Transactions/Transaction.cs @@ -84,7 +84,18 @@ public async ValueTask CommitAsync(CancellationToken ct = default) throw new InvalidOperationException($"Cannot commit transaction in state {_state}"); // StorageEngine handles WAL writes and buffer management - await _storage.CommitTransactionAsync(_transactionId, ct); + try + { + await _storage.CommitTransactionAsync(_transactionId, ct); + } + catch (BLiteTransactionConflictException) + { + // Another transaction committed a change to one of this transaction's pages first + // (#304). Nothing was written: roll back so that the caches, the free-space index and + // the state are back in step, and let the caller retry in a new transaction. + await RollbackAsync().ConfigureAwait(false); + throw; + } _state = TransactionState.Committed; diff --git a/tests/BLite.Tests/TransactionConflictTests.cs b/tests/BLite.Tests/TransactionConflictTests.cs new file mode 100644 index 00000000..a6a62665 --- /dev/null +++ b/tests/BLite.Tests/TransactionConflictTests.cs @@ -0,0 +1,282 @@ +using BLite.Bson; +using BLite.Core; +using BLite.Core.Transactions; +using BLite.Shared; + +namespace BLite.Tests; + +/// +/// Two transactions that are open at the same time and write the same page must never both +/// commit with one of them silently lost (#304). The first committer wins; the second commit +/// either succeeds with both writes intact or throws +/// and is rolled back. +/// +public class TransactionConflictTests : IDisposable +{ + private readonly string _dir = Path.Combine(Path.GetTempPath(), $"blite_conflict_{Guid.NewGuid():N}"); + + public TransactionConflictTests() => Directory.CreateDirectory(_dir); + + public void Dispose() + { + try { Directory.Delete(_dir, recursive: true); } catch { } + } + + private string DbPath(string name) => Path.Combine(_dir, name + ".db"); + + /// + /// Commits and reports whether it was rejected as a conflict. + /// A rejected transaction must be rolled back. + /// + private static async Task CommitSecondAsync(ITransaction second) + { + try + { + await second.CommitAsync(); + return false; + } + catch (BLiteTransactionConflictException) + { + Assert.Equal(TransactionState.Aborted, second.State); + return true; + } + } + + [Fact] + public async Task Typed_two_open_transactions_on_one_collection_never_lose_a_commit() + { + var path = DbPath("typed"); + ObjectId a, b; + bool conflict; + using (var db = new TestDbContext(path)) + { + await db.Users.InsertAsync(new User { Name = "seed" }); + await db.SaveChangesAsync(); + + var t1 = db.BeginTransaction(); + var t2 = db.BeginTransaction(); + var first = new User { Name = "from-t1" }; + var second = new User { Name = "from-t2" }; + await db.Users.InsertAsync(first, t1); + await db.Users.InsertAsync(second, t2); + a = first.Id; b = second.Id; + + await t1.CommitAsync(); + conflict = await CommitSecondAsync(t2); + + Assert.NotNull(await db.Users.FindByIdAsync(a)); + Assert.Equal(!conflict, await db.Users.FindByIdAsync(b) is not null); + Assert.Equal(conflict ? 2 : 3, await db.Users.CountAsync()); + t1.Dispose(); t2.Dispose(); + } + + using (var db = new TestDbContext(path)) + { + Assert.NotNull(await db.Users.FindByIdAsync(a)); + Assert.Equal(conflict ? 2 : 3, await db.Users.CountAsync()); + } + } + + [Fact] + public async Task Dynamic_two_open_transactions_on_one_collection_never_lose_a_commit() + { + var path = DbPath("dynamic"); + BsonId a, b; + bool conflict; + using (var engine = new BLiteEngine(path)) + { + var col = engine.GetOrCreateCollection("people"); + await col.InsertAsync(col.CreateDocument(["name"], x => x.AddString("name", "seed"))); + + var t1 = engine.BeginTransaction(); + var t2 = engine.BeginTransaction(); + a = await col.InsertAsync(col.CreateDocument(["name"], x => x.AddString("name", "from-t1")), t1); + b = await col.InsertAsync(col.CreateDocument(["name"], x => x.AddString("name", "from-t2")), t2); + + await t1.CommitAsync(); + conflict = await CommitSecondAsync(t2); + + Assert.NotNull(await col.FindByIdAsync(a)); + Assert.Equal(!conflict, await col.FindByIdAsync(b) is not null); + Assert.Equal(conflict ? 2 : 3, await col.CountAsync()); + t1.Dispose(); t2.Dispose(); + } + + using (var engine = new BLiteEngine(path)) + { + var col = engine.GetOrCreateCollection("people"); + Assert.NotNull(await col.FindByIdAsync(a)); + Assert.Equal(conflict ? 2 : 3, await col.CountAsync()); + } + } + + [Fact] + public async Task Two_sessions_with_open_transactions_never_lose_a_commit() + { + var path = DbPath("sessions"); + BsonId a, b; + bool conflict; + using (var engine = new BLiteEngine(path)) + { + var col = engine.GetOrCreateCollection("orders"); + await col.InsertAsync(col.CreateDocument(["n"], x => x.AddInt32("n", 0))); + + using var s1 = engine.OpenSession(); + using var s2 = engine.OpenSession(); + var tx1 = s1.BeginTransaction(); + var tx2 = s2.BeginTransaction(); + a = await s1.InsertAsync("orders", col.CreateDocument(["n"], x => x.AddInt32("n", 1))); + b = await s2.InsertAsync("orders", col.CreateDocument(["n"], x => x.AddInt32("n", 2))); + + await s1.CommitAsync(); + try + { + await s2.CommitAsync(); + conflict = false; + } + catch (BLiteTransactionConflictException) + { + conflict = true; + Assert.Equal(TransactionState.Aborted, tx2.State); + } + + Assert.NotNull(await col.FindByIdAsync(a)); + Assert.Equal(!conflict, await col.FindByIdAsync(b) is not null); + Assert.Equal(conflict ? 2 : 3, await col.CountAsync()); + } + + using (var engine = new BLiteEngine(path)) + Assert.Equal(conflict ? 2 : 3, await engine.GetOrCreateCollection("orders").CountAsync()); + } + + [Fact] + public async Task An_open_transaction_never_overwrites_an_auto_commit_write_to_the_same_page() + { + var path = DbPath("autocommit"); + using var db = new TestDbContext(path); + await db.Users.InsertAsync(new User { Name = "seed" }); + await db.SaveChangesAsync(); + + using var t1 = db.BeginTransaction(); + var inTxn = new User { Name = "in-transaction" }; + await db.Users.InsertAsync(inTxn, t1); + + var auto = new User { Name = "auto-commit" }; + await db.Users.InsertAsync(auto); // commits on its own, same data page and leaf + + bool conflict = await CommitSecondAsync(t1); + + Assert.NotNull(await db.Users.FindByIdAsync(auto.Id)); + Assert.Equal(!conflict, await db.Users.FindByIdAsync(inTxn.Id) is not null); + Assert.Equal(conflict ? 2 : 3, await db.Users.CountAsync()); + } + + [Fact] + public async Task Transactions_on_different_collections_both_commit() + { + var path = DbPath("disjoint"); + using var db = new TestDbContext(path); + await db.Users.InsertAsync(new User { Name = "seed" }); + await db.TestDocuments.InsertAsync(new TestDocument { Category = "seed" }); + await db.SaveChangesAsync(); + + using var t1 = db.BeginTransaction(); + using var t2 = db.BeginTransaction(); + var user = new User { Name = "u" }; + var doc = new TestDocument { Category = "d" }; + await db.Users.InsertAsync(user, t1); + await db.TestDocuments.InsertAsync(doc, t2); + + await t1.CommitAsync(); + await t2.CommitAsync(); + + Assert.NotNull(await db.Users.FindByIdAsync(user.Id)); + Assert.NotNull(await db.TestDocuments.FindByIdAsync(doc.Id)); + } + + [Fact] + public async Task A_rejected_transaction_leaves_no_trace_and_can_be_retried() + { + var path = DbPath("retry"); + using (var db = new TestDbContext(path)) + { + await db.Users.InsertAsync(new User { Name = "seed" }); + await db.SaveChangesAsync(); + + var t1 = db.BeginTransaction(); + var t2 = db.BeginTransaction(); + await db.Users.InsertAsync(new User { Name = "from-t1" }, t1); + var loser = new User { Name = "from-t2" }; + await db.Users.InsertAsync(loser, t2); + await t1.CommitAsync(); + + // On this shape (same data page, same primary leaf) the second commit must be rejected. + await Assert.ThrowsAsync(async () => await t2.CommitAsync()); + Assert.Equal(TransactionState.Aborted, t2.State); + Assert.Null(await db.Users.FindByIdAsync(loser.Id)); + t1.Dispose(); t2.Dispose(); + + // The free-space index and caches are back in step: a retry and a plain insert work. + using (var retry = db.BeginTransaction()) + { + await db.Users.InsertAsync(new User { Name = "from-t2-retry" }, retry); + await retry.CommitAsync(); + } + await db.Users.InsertAsync(new User { Name = "after" }); + await db.SaveChangesAsync(); + Assert.Equal(4, await db.Users.CountAsync()); + } + + using (var db = new TestDbContext(path)) + Assert.Equal(4, await db.Users.CountAsync()); + } + + [Fact] + public async Task Sequential_transactions_on_one_collection_never_conflict() + { + var path = DbPath("sequential"); + using var db = new TestDbContext(path); + for (int i = 0; i < 50; i++) + { + using var t = db.BeginTransaction(); + await db.Users.InsertAsync(new User { Name = $"u{i}" }, t); + await t.CommitAsync(); + } + Assert.Equal(50, await db.Users.CountAsync()); + } + + [Fact] + public async Task Concurrent_sessions_inserting_without_a_transaction_lose_nothing() + { + // Each session builds its own DynamicCollection for "concurrent". With a lock per instance + // their auto-commit inserts ran concurrently on the same pages and most were lost, while + // every InsertAsync returned an id. + var path = DbPath("sessions-auto"); + using var engine = new BLiteEngine(path); + engine.RegisterKeys(["n"]); + for (int run = 0; run < 5; run++) + { + var ids = await Task.WhenAll(Enumerable.Range(0, 10).Select(async i => + { + using var session = engine.OpenSession(); + return await session.InsertAsync("concurrent", engine.CreateDocument(["n"], b => b.AddInt32("n", i))); + })); + Assert.Equal(10, ids.Distinct().Count()); + } + Assert.Equal(50, await engine.GetOrCreateCollection("concurrent").CountAsync()); + } + + [Fact] + public async Task Concurrent_auto_commit_writers_never_conflict() + { + var path = DbPath("concurrent-auto"); + using var db = new TestDbContext(path); + var writers = Enumerable.Range(0, 4).Select(w => Task.Run(async () => + { + for (int i = 0; i < 25; i++) + await db.Users.InsertAsync(new User { Name = $"w{w}-{i}" }); + })).ToArray(); + await Task.WhenAll(writers); + Assert.Equal(100, await db.Users.CountAsync()); + } +} From bc6bed6360b6cfc9ec844f4ea445774c7275c655 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ricardo=20Gro=C3=9F?= Date: Mon, 19 Oct 2026 06:42:36 +0200 Subject: [PATCH 2/5] fix(storage): retry an auto-commit write whose commit was rejected (#304) A caller-owned transaction commits without taking the collection lock, so its commit can land between the reads and the commit of a write made without a transaction. That write was rejected with BLiteTransactionConflictException, although its caller never opened a transaction: 118 of 300 auto-commit inserts threw while racing 300 explicit transactions. The 17 write methods that open their own transaction (typed and dynamic insert, update, upsert, delete and their bulk variants, typed TruncateAsync) now run in a loop inside the collection lock. When the commit of their own transaction is rejected, CommitAsync has rolled it back and nothing was written, so they run again in a new transaction, up to 10 attempts (WriteFailure.ShouldRetry). The bulk methods read their input once. The exception's documentation and the README say which operations retry and which can still throw it. Co-Authored-By: Claude Opus 5.5 --- README.md | 2 +- .../BLiteTransactionConflictException.cs | 9 +- .../Collections/DocumentCollection.cs | 364 +++++++++------ src/BLite.Core/DynamicCollection.cs | 415 +++++++++++------- src/BLite.Core/Transactions/WriteFailure.cs | 17 + tests/BLite.Tests/TransactionConflictTests.cs | 78 ++++ 6 files changed, 581 insertions(+), 304 deletions(-) diff --git a/README.md b/README.md index 9305837e..b524bbd0 100644 --- a/README.md +++ b/README.md @@ -393,7 +393,7 @@ using var dynSub = engine.GetOrCreateCollection("orders").Watch() ### 🛡️ Transactions & ACID - **Atomic**: Multi-document transactions. - **Durable**: WAL ensures data safety even in power loss. -- **Isolated**: Read committed: a transaction sees its own writes and what other transactions have committed. Two transactions that change the same page cannot both commit: the first one wins, the second one's `CommitAsync` throws `BLiteTransactionConflictException` and rolls it back, so run it again in a new transaction. +- **Isolated**: Read committed: a transaction sees its own writes and what other transactions have committed. Two transactions that change the same page cannot both commit: the first one wins, the second one's `CommitAsync` throws `BLiteTransactionConflictException` and rolls it back, so run it again in a new transaction. A write made without a transaction (`InsertAsync(doc)`, `UpdateAsync`, `DeleteAsync`, …) commits on its own and, when a transaction committed one of its pages first, runs again by itself; it throws `BLiteTransactionConflictException` only after 10 rejected attempts. Index creation, `VacuumAsync` and retention commit without retrying and can throw it too. - **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. - **Implicit Transactions**: Use `SaveChangesAsync()` for automatic transaction management. diff --git a/src/BLite.Core/BLiteTransactionConflictException.cs b/src/BLite.Core/BLiteTransactionConflictException.cs index e834ae09..7fce95ff 100644 --- a/src/BLite.Core/BLiteTransactionConflictException.cs +++ b/src/BLite.Core/BLiteTransactionConflictException.cs @@ -17,8 +17,13 @@ namespace BLite.Core; /// /// Conflicts are detected per page, not per document: two transactions that change different /// documents stored on the same data page, or whose index entries share a B-Tree leaf, conflict too. -/// Writes made without an explicit transaction commit on their own and retry nothing; they can only -/// make a caller's open transaction fail, never be lost to it. +/// A write method called without a transaction (InsertAsync(document), UpdateAsync, +/// DeleteAsync, their bulk and upsert variants, TruncateAsync of a typed collection, and +/// the BLiteSession methods) commits on its own. When a transaction that committed first changed +/// one of its pages, it runs again in a new transaction while it still holds the collection lock, and +/// throws this exception only if all of its attempts (10) were rejected. Other operations that commit a +/// transaction of their own (index creation, VacuumAsync, retention and time-series pruning, +/// DynamicCollection.TruncateAsync) do not retry and can throw it; run them again. /// /// /// Derives from , so code that already catches that type diff --git a/src/BLite.Core/Collections/DocumentCollection.cs b/src/BLite.Core/Collections/DocumentCollection.cs index c18ffa85..90134747 100644 --- a/src/BLite.Core/Collections/DocumentCollection.cs +++ b/src/BLite.Core/Collections/DocumentCollection.cs @@ -792,37 +792,47 @@ public async Task TruncateAsync(ITransaction? transaction, CancellationToke if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct).ConfigureAwait(false)) throw new TimeoutException("Timed out acquiring collection lock (TruncateAsync)."); - transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); try { - try + for (int attempt = 1; ; attempt++) { - // Collect all keys first to avoid modifying the index while iterating. - // Use the caller's transaction ID so that uncommitted inserts/updates - // visible to this transaction are included in the truncation. - var txnId = transaction.TransactionId; - var keys = new List(); - foreach (var entry in _primaryIndex.Range(IndexKey.MinKey, IndexKey.MaxKey, IndexDirection.Forward, txnId)) + transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); + try { - ct.ThrowIfCancellationRequested(); - keys.Add(entry.Key); - } + // Collect all keys first to avoid modifying the index while iterating. + // Use the caller's transaction ID so that uncommitted inserts/updates + // visible to this transaction are included in the truncation. + var txnId = transaction.TransactionId; + var keys = new List(); + foreach (var entry in _primaryIndex.Range(IndexKey.MinKey, IndexKey.MaxKey, IndexDirection.Forward, txnId)) + { + ct.ThrowIfCancellationRequested(); + keys.Add(entry.Key); + } + + foreach (var key in keys) + { + ct.ThrowIfCancellationRequested(); + var id = _mapper.FromIndexKey(key); + if (await DeleteCore(id, transaction, notifyCdc: false).ConfigureAwait(false)) + deleted++; + } - foreach (var key in keys) + if (autoCommit) + await transaction.CommitAsync(ct).ConfigureAwait(false); + break; + } + catch (Exception ex) when (WriteFailure.ShouldRetry(transaction, autoCommit, ex, attempt)) { - ct.ThrowIfCancellationRequested(); - var id = _mapper.FromIndexKey(key); - if (await DeleteCore(id, transaction, notifyCdc: false).ConfigureAwait(false)) - deleted++; + // Rolled back and nothing written: run again in a new transaction (#304). + transaction = null; + deleted = 0; + } + catch (Exception ex) + { + await WriteFailure.HandleAsync(transaction, autoCommit, ex); + throw; } - - if (autoCommit) - await transaction.CommitAsync(ct).ConfigureAwait(false); - } - catch (Exception ex) - { - await WriteFailure.HandleAsync(transaction, autoCommit, ex); - throw; } } finally @@ -2274,32 +2284,40 @@ public async ValueTask InsertAsync(T entity, ITransaction? transaction, Can if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (Insert)."); - transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); try { - try + for (int attempt = 1; ; attempt++) { - var id = await InsertCore(entity, transaction); - if (autoCommit) + transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); + try { - await transaction.CommitAsync(ct); - // ── OnInsert retention trigger — runs AFTER commit within the lock ── - if (_retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0) - await ApplyRetentionPolicyCoreAsync(ct); + var id = await InsertCore(entity, transaction); + if (autoCommit) + { + await transaction.CommitAsync(ct); + // ── OnInsert retention trigger — runs AFTER commit within the lock ── + if (_retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0) + await ApplyRetentionPolicyCoreAsync(ct); + } + else if (_retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0 + && transaction is Transaction concreteTx) + { + // For caller-managed transactions, fire retention in the background after commit. + concreteTx.OnCommit += () => _ = RunScheduledRetentionAsync(); + } + success = true; + return id; } - else if (_retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0 - && transaction is Transaction concreteTx) + catch (Exception ex) when (WriteFailure.ShouldRetry(transaction, autoCommit, ex, attempt)) { - // For caller-managed transactions, fire retention in the background after commit. - concreteTx.OnCommit += () => _ = RunScheduledRetentionAsync(); + // Rolled back and nothing written: run again in a new transaction (#304). + transaction = null; + } + catch (Exception ex) + { + await WriteFailure.HandleAsync(transaction, autoCommit, ex); + throw; } - success = true; - return id; - } - catch (Exception ex) - { - await WriteFailure.HandleAsync(transaction, autoCommit, ex); - throw; } } finally @@ -2341,31 +2359,40 @@ public async ValueTask> InsertBulkAsync(IEnumerable entities, ITran throw new TimeoutException("Timed out acquiring collection lock (InsertBulk)."); bool autoCommit = transaction == null; - transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); try { - try + for (int attempt = 1; ; attempt++) { - await InsertBulkInternal(entityList, ids, transaction); - if (autoCommit) + transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); + try { - await transaction.CommitAsync(ct); - // ── OnInsert retention trigger — runs AFTER commit within the lock ── - if (_retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0) - await ApplyRetentionPolicyCoreAsync(ct); + await InsertBulkInternal(entityList, ids, transaction); + if (autoCommit) + { + await transaction.CommitAsync(ct); + // ── OnInsert retention trigger — runs AFTER commit within the lock ── + if (_retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0) + await ApplyRetentionPolicyCoreAsync(ct); + } + else if (_retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0 + && transaction is Transaction concreteTx) + { + // For caller-managed transactions, fire retention in the background after commit. + concreteTx.OnCommit += () => _ = RunScheduledRetentionAsync(); + } + return ids; } - else if (_retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0 - && transaction is Transaction concreteTx) + catch (Exception ex) when (WriteFailure.ShouldRetry(transaction, autoCommit, ex, attempt)) { - // For caller-managed transactions, fire retention in the background after commit. - concreteTx.OnCommit += () => _ = RunScheduledRetentionAsync(); + // Rolled back and nothing written: run again in a new transaction (#304). + transaction = null; + ids.Clear(); + } + catch (Exception ex) + { + await WriteFailure.HandleAsync(transaction, autoCommit, ex, callHasWritten: ids.Count > 0); + throw; } - return ids; - } - catch (Exception ex) - { - await WriteFailure.HandleAsync(transaction, autoCommit, ex, callHasWritten: ids.Count > 0); - throw; } } finally @@ -2936,20 +2963,28 @@ public async ValueTask UpdateAsync(T entity, ITransaction? transaction, Ca if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (Update)."); - transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); try { - try - { - var result = await UpdateCore(entity, transaction); - if (autoCommit && result) await transaction.CommitAsync(ct); - success = result; - return result; - } - catch (Exception ex) + for (int attempt = 1; ; attempt++) { - await WriteFailure.HandleAsync(transaction, autoCommit, ex); - throw; + transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); + try + { + var result = await UpdateCore(entity, transaction); + if (autoCommit && result) await transaction.CommitAsync(ct); + success = result; + return result; + } + catch (Exception ex) when (WriteFailure.ShouldRetry(transaction, autoCommit, ex, attempt)) + { + // Rolled back and nothing written: run again in a new transaction (#304). + transaction = null; + } + catch (Exception ex) + { + await WriteFailure.HandleAsync(transaction, autoCommit, ex); + throw; + } } } finally @@ -2993,19 +3028,27 @@ public async ValueTask UpdateBulkAsync(IEnumerable entities, ITransactio if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (UpdateBulk)."); - transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); try { - try + for (int attempt = 1; ; attempt++) { - updateCount = await UpdateBulkInternal(entityList, transaction, progress); - if (autoCommit) await transaction.CommitAsync(ct); - return updateCount; - } - catch (Exception ex) - { - await WriteFailure.HandleAsync(transaction, autoCommit, ex, callHasWritten: progress.Value > 0); - throw; + transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); + try + { + updateCount = await UpdateBulkInternal(entityList, transaction, progress); + if (autoCommit) await transaction.CommitAsync(ct); + return updateCount; + } + catch (Exception ex) when (WriteFailure.ShouldRetry(transaction, autoCommit, ex, attempt)) + { + // Rolled back and nothing written: run again in a new transaction (#304). + transaction = null; + } + catch (Exception ex) + { + await WriteFailure.HandleAsync(transaction, autoCommit, ex, callHasWritten: progress.Value > 0); + throw; + } } } finally @@ -3205,32 +3248,40 @@ public async ValueTask> UpsertAsync(T entity, ITransaction? tr if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (Upsert)."); - transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); try { - try + for (int attempt = 1; ; attempt++) { - var result = await UpsertCore(entity, transaction); - inserted = result.Inserted; - if (autoCommit) + transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); + try { - await transaction.CommitAsync(ct); - // ── OnInsert retention trigger — only fires when this call actually inserted ── - if (inserted && _retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0) - await ApplyRetentionPolicyCoreAsync(ct); + var result = await UpsertCore(entity, transaction); + inserted = result.Inserted; + if (autoCommit) + { + await transaction.CommitAsync(ct); + // ── OnInsert retention trigger — only fires when this call actually inserted ── + if (inserted && _retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0) + await ApplyRetentionPolicyCoreAsync(ct); + } + else if (inserted && _retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0 + && transaction is Transaction concreteTx) + { + concreteTx.OnCommit += () => _ = RunScheduledRetentionAsync(); + } + success = true; + return result; } - else if (inserted && _retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0 - && transaction is Transaction concreteTx) + catch (Exception ex) when (WriteFailure.ShouldRetry(transaction, autoCommit, ex, attempt)) { - concreteTx.OnCommit += () => _ = RunScheduledRetentionAsync(); + // Rolled back and nothing written: run again in a new transaction (#304). + transaction = null; + } + catch (Exception ex) + { + await WriteFailure.HandleAsync(transaction, autoCommit, ex); + throw; } - success = true; - return result; - } - catch (Exception ex) - { - await WriteFailure.HandleAsync(transaction, autoCommit, ex); - throw; } } finally @@ -3277,36 +3328,45 @@ public async ValueTask>> UpsertBulkAsync(IEnumerable e if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (UpsertBulk)."); - transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); try { - try + for (int attempt = 1; ; attempt++) { - bool anyInserted = false; - foreach (var entity in entityList) + transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); + try { - var result = await UpsertCore(entity, transaction); - anyInserted |= result.Inserted; - results.Add(result); - } + bool anyInserted = false; + foreach (var entity in entityList) + { + var result = await UpsertCore(entity, transaction); + anyInserted |= result.Inserted; + results.Add(result); + } - if (autoCommit) + if (autoCommit) + { + await transaction.CommitAsync(ct); + if (anyInserted && _retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0) + await ApplyRetentionPolicyCoreAsync(ct); + } + else if (anyInserted && _retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0 + && transaction is Transaction concreteTx) + { + concreteTx.OnCommit += () => _ = RunScheduledRetentionAsync(); + } + return results; + } + catch (Exception ex) when (WriteFailure.ShouldRetry(transaction, autoCommit, ex, attempt)) { - await transaction.CommitAsync(ct); - if (anyInserted && _retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0) - await ApplyRetentionPolicyCoreAsync(ct); + // Rolled back and nothing written: run again in a new transaction (#304). + transaction = null; + results.Clear(); } - else if (anyInserted && _retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0 - && transaction is Transaction concreteTx) + catch (Exception ex) { - concreteTx.OnCommit += () => _ = RunScheduledRetentionAsync(); + await WriteFailure.HandleAsync(transaction, autoCommit, ex, callHasWritten: results.Count > 0); + throw; } - return results; - } - catch (Exception ex) - { - await WriteFailure.HandleAsync(transaction, autoCommit, ex, callHasWritten: results.Count > 0); - throw; } } finally @@ -3372,20 +3432,28 @@ public async ValueTask DeleteAsync(TId id, ITransaction? transaction, Canc if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (Delete)."); - transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); try { - try + for (int attempt = 1; ; attempt++) { - var result = await DeleteCore(id, transaction); - if (autoCommit && result) await transaction.CommitAsync(ct); - success = result; - return result; - } - catch (Exception ex) - { - await WriteFailure.HandleAsync(transaction, autoCommit, ex); - throw; + transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); + try + { + var result = await DeleteCore(id, transaction); + if (autoCommit && result) await transaction.CommitAsync(ct); + success = result; + return result; + } + catch (Exception ex) when (WriteFailure.ShouldRetry(transaction, autoCommit, ex, attempt)) + { + // Rolled back and nothing written: run again in a new transaction (#304). + transaction = null; + } + catch (Exception ex) + { + await WriteFailure.HandleAsync(transaction, autoCommit, ex); + throw; + } } } finally @@ -3414,6 +3482,8 @@ public async ValueTask DeleteAsync(TId id, ITransaction? transaction, Canc public async ValueTask DeleteBulkAsync(IEnumerable ids, ITransaction? transaction, CancellationToken ct = default) { if (ids == null) throw new ArgumentNullException(nameof(ids)); + // Read once: a write retried after a conflict (#304) runs over the same items. + var idList = ids.ToList(); int deleteCount = 0; bool autoCommit = transaction == null; @@ -3423,19 +3493,27 @@ public async ValueTask DeleteBulkAsync(IEnumerable ids, ITransaction? if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (DeleteBulk)."); - transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); try { - try + for (int attempt = 1; ; attempt++) { - deleteCount = await DeleteBulkInternal(ids, transaction); - if (autoCommit) await transaction.CommitAsync(ct); - return deleteCount; - } - catch (Exception ex) - { - await WriteFailure.HandleAsync(transaction, autoCommit, ex); - throw; + transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); + try + { + deleteCount = await DeleteBulkInternal(idList, transaction); + if (autoCommit) await transaction.CommitAsync(ct); + return deleteCount; + } + catch (Exception ex) when (WriteFailure.ShouldRetry(transaction, autoCommit, ex, attempt)) + { + // Rolled back and nothing written: run again in a new transaction (#304). + transaction = null; + } + catch (Exception ex) + { + await WriteFailure.HandleAsync(transaction, autoCommit, ex); + throw; + } } } finally diff --git a/src/BLite.Core/DynamicCollection.cs b/src/BLite.Core/DynamicCollection.cs index 2fcb65ad..f66033f4 100644 --- a/src/BLite.Core/DynamicCollection.cs +++ b/src/BLite.Core/DynamicCollection.cs @@ -864,30 +864,41 @@ public async ValueTask InsertAsync(BsonDocument document, ITransaction? if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (Insert)."); - transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); try { - var result = await InsertCore(document, transaction); - if (autoCommit) + for (int attempt = 1; ; attempt++) { - await transaction.CommitAsync(ct); - // ── OnInsert retention trigger — runs AFTER commit within the lock ── - if (_retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0) - await ApplyRetentionPolicyCoreAsync(ct); - } - else if (_retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0 - && transaction is Transaction concreteTx) - { - // For caller-managed transactions, fire retention in the background after commit. - concreteTx.OnCommit += () => _ = RunScheduledRetentionAsync(); + transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); + try + { + var result = await InsertCore(document, transaction); + if (autoCommit) + { + await transaction.CommitAsync(ct); + // ── OnInsert retention trigger — runs AFTER commit within the lock ── + if (_retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0) + await ApplyRetentionPolicyCoreAsync(ct); + } + else if (_retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0 + && transaction is Transaction concreteTx) + { + // For caller-managed transactions, fire retention in the background after commit. + concreteTx.OnCommit += () => _ = RunScheduledRetentionAsync(); + } + success = true; + return result; + } + catch (Exception ex) when (WriteFailure.ShouldRetry(transaction, autoCommit, ex, attempt)) + { + // Rolled back and nothing written: run again in a new transaction (#304). + transaction = null; + } + catch (Exception ex) + { + await WriteFailure.HandleAsync(transaction, autoCommit, ex); + throw; + } } - success = true; - return result; - } - catch (Exception ex) - { - await WriteFailure.HandleAsync(transaction, autoCommit, ex); - throw; } finally { @@ -954,6 +965,8 @@ public ValueTask> InsertBulkAsync(IEnumerable documen public async ValueTask> InsertBulkAsync(IEnumerable documents, ITransaction? transaction, CancellationToken ct = default) { if (documents == null) throw new ArgumentNullException(nameof(documents)); + // Read once: a write retried after a conflict (#304) runs over the same items. + var documentList = documents.ToList(); bool autoCommit = transaction == null; WriteFailure.EnsureWritable(transaction); @@ -961,34 +974,46 @@ public async ValueTask> InsertBulkAsync(IEnumerable d if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (InsertBulk)."); - transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); var ids = new List(); try { - foreach (var doc in documents) - { - ct.ThrowIfCancellationRequested(); - ids.Add(await InsertCore(doc, transaction)); - } - if (autoCommit) + for (int attempt = 1; ; attempt++) { - await transaction.CommitAsync(ct); - // ── OnInsert retention trigger — runs AFTER commit within the lock ── - if (_retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0) - await ApplyRetentionPolicyCoreAsync(ct); - } - else if (_retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0 - && transaction is Transaction concreteTx) - { - // For caller-managed transactions, fire retention in the background after commit. - concreteTx.OnCommit += () => _ = RunScheduledRetentionAsync(); + transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); + try + { + foreach (var doc in documentList) + { + ct.ThrowIfCancellationRequested(); + ids.Add(await InsertCore(doc, transaction)); + } + if (autoCommit) + { + await transaction.CommitAsync(ct); + // ── OnInsert retention trigger — runs AFTER commit within the lock ── + if (_retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0) + await ApplyRetentionPolicyCoreAsync(ct); + } + else if (_retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0 + && transaction is Transaction concreteTx) + { + // For caller-managed transactions, fire retention in the background after commit. + concreteTx.OnCommit += () => _ = RunScheduledRetentionAsync(); + } + return ids; + } + catch (Exception ex) when (WriteFailure.ShouldRetry(transaction, autoCommit, ex, attempt)) + { + // Rolled back and nothing written: run again in a new transaction (#304). + transaction = null; + ids.Clear(); + } + catch (Exception ex) + { + await WriteFailure.HandleAsync(transaction, autoCommit, ex, callHasWritten: ids.Count > 0); + throw; + } } - return ids; - } - catch (Exception ex) - { - await WriteFailure.HandleAsync(transaction, autoCommit, ex, callHasWritten: ids.Count > 0); - throw; } finally { @@ -1237,23 +1262,34 @@ public async ValueTask UpdateAsync(BsonId id, BsonDocument newDocument, IT if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (Update)."); - transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); try { - var key = new IndexKey(id.ToBytes()); - if (!_primaryIndex.TryFind(key, out var oldLocation, transaction.TransactionId)) - return false; + for (int attempt = 1; ; attempt++) + { + transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); + try + { + var key = new IndexKey(id.ToBytes()); + if (!_primaryIndex.TryFind(key, out var oldLocation, transaction.TransactionId)) + return false; - await UpdateAtLocationCore(id, key, newDocument, oldLocation, transaction); + await UpdateAtLocationCore(id, key, newDocument, oldLocation, transaction); - if (autoCommit) await transaction.CommitAsync(ct); - success = true; - return true; - } - catch (Exception ex) - { - await WriteFailure.HandleAsync(transaction, autoCommit, ex); - throw; + if (autoCommit) await transaction.CommitAsync(ct); + success = true; + return true; + } + catch (Exception ex) when (WriteFailure.ShouldRetry(transaction, autoCommit, ex, attempt)) + { + // Rolled back and nothing written: run again in a new transaction (#304). + transaction = null; + } + catch (Exception ex) + { + await WriteFailure.HandleAsync(transaction, autoCommit, ex); + throw; + } + } } finally { @@ -1351,29 +1387,40 @@ public async ValueTask> UpsertAsync(BsonDocument document, if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (Upsert)."); - transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); try { - var result = await UpsertCore(document, transaction); - inserted = result.Inserted; - if (autoCommit) - { - await transaction.CommitAsync(ct); - if (inserted && _retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0) - await ApplyRetentionPolicyCoreAsync(ct); - } - else if (inserted && _retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0 - && transaction is Transaction concreteTx) + for (int attempt = 1; ; attempt++) { - concreteTx.OnCommit += () => _ = RunScheduledRetentionAsync(); + transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); + try + { + var result = await UpsertCore(document, transaction); + inserted = result.Inserted; + if (autoCommit) + { + await transaction.CommitAsync(ct); + if (inserted && _retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0) + await ApplyRetentionPolicyCoreAsync(ct); + } + else if (inserted && _retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0 + && transaction is Transaction concreteTx) + { + concreteTx.OnCommit += () => _ = RunScheduledRetentionAsync(); + } + success = true; + return result; + } + catch (Exception ex) when (WriteFailure.ShouldRetry(transaction, autoCommit, ex, attempt)) + { + // Rolled back and nothing written: run again in a new transaction (#304). + transaction = null; + } + catch (Exception ex) + { + await WriteFailure.HandleAsync(transaction, autoCommit, ex); + throw; + } } - success = true; - return result; - } - catch (Exception ex) - { - await WriteFailure.HandleAsync(transaction, autoCommit, ex); - throw; } finally { @@ -1437,6 +1484,8 @@ public ValueTask>> UpsertBulkAsync(IEnumerable>> UpsertBulkAsync(IEnumerable documents, ITransaction? transaction, CancellationToken ct = default) { if (documents == null) throw new ArgumentNullException(nameof(documents)); + // Read once: a write retried after a conflict (#304) runs over the same items. + var documentList = documents.ToList(); bool autoCommit = transaction == null; WriteFailure.EnsureWritable(transaction); @@ -1444,36 +1493,48 @@ public async ValueTask>> UpsertBulkAsync(IEnumerable>(); try { - bool anyInserted = false; - foreach (var doc in documents) + for (int attempt = 1; ; attempt++) { - ct.ThrowIfCancellationRequested(); - var result = await UpsertCore(doc, transaction); - anyInserted |= result.Inserted; - results.Add(result); - } + transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); + try + { + bool anyInserted = false; + foreach (var doc in documentList) + { + ct.ThrowIfCancellationRequested(); + var result = await UpsertCore(doc, transaction); + anyInserted |= result.Inserted; + results.Add(result); + } - if (autoCommit) - { - await transaction.CommitAsync(ct); - if (anyInserted && _retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0) - await ApplyRetentionPolicyCoreAsync(ct); - } - else if (anyInserted && _retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0 - && transaction is Transaction concreteTx) - { - concreteTx.OnCommit += () => _ = RunScheduledRetentionAsync(); + if (autoCommit) + { + await transaction.CommitAsync(ct); + if (anyInserted && _retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0) + await ApplyRetentionPolicyCoreAsync(ct); + } + else if (anyInserted && _retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0 + && transaction is Transaction concreteTx) + { + concreteTx.OnCommit += () => _ = RunScheduledRetentionAsync(); + } + return results; + } + catch (Exception ex) when (WriteFailure.ShouldRetry(transaction, autoCommit, ex, attempt)) + { + // Rolled back and nothing written: run again in a new transaction (#304). + transaction = null; + results.Clear(); + } + catch (Exception ex) + { + await WriteFailure.HandleAsync(transaction, autoCommit, ex, callHasWritten: results.Count > 0); + throw; + } } - return results; - } - catch (Exception ex) - { - await WriteFailure.HandleAsync(transaction, autoCommit, ex, callHasWritten: results.Count > 0); - throw; } finally { @@ -1526,6 +1587,8 @@ public ValueTask UpdateBulkAsync(IEnumerable<(BsonId Id, BsonDocument Docum public async ValueTask UpdateBulkAsync(IEnumerable<(BsonId Id, BsonDocument Document)> updates, ITransaction? transaction, CancellationToken ct = default) { if (updates == null) throw new ArgumentNullException(nameof(updates)); + // Read once: a write retried after a conflict (#304) runs over the same items. + var updateList = updates.ToList(); bool autoCommit = transaction == null; WriteFailure.EnsureWritable(transaction); @@ -1533,27 +1596,39 @@ public async ValueTask UpdateBulkAsync(IEnumerable<(BsonId Id, BsonDocument if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (UpdateBulk)."); - transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); var count = 0; try { - foreach (var (id, doc) in updates) + for (int attempt = 1; ; attempt++) { - ct.ThrowIfCancellationRequested(); - var key = new IndexKey(id.ToBytes()); - if (!_primaryIndex.TryFind(key, out var oldLocation, transaction.TransactionId)) - continue; + transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); + try + { + foreach (var (id, doc) in updateList) + { + ct.ThrowIfCancellationRequested(); + var key = new IndexKey(id.ToBytes()); + if (!_primaryIndex.TryFind(key, out var oldLocation, transaction.TransactionId)) + continue; - await UpdateAtLocationCore(id, key, doc, oldLocation, transaction); - count++; + await UpdateAtLocationCore(id, key, doc, oldLocation, transaction); + count++; + } + if (autoCommit) await transaction.CommitAsync(ct); + return count; + } + catch (Exception ex) when (WriteFailure.ShouldRetry(transaction, autoCommit, ex, attempt)) + { + // Rolled back and nothing written: run again in a new transaction (#304). + transaction = null; + count = 0; + } + catch (Exception ex) + { + await WriteFailure.HandleAsync(transaction, autoCommit, ex, callHasWritten: count > 0); + throw; + } } - if (autoCommit) await transaction.CommitAsync(ct); - return count; - } - catch (Exception ex) - { - await WriteFailure.HandleAsync(transaction, autoCommit, ex, callHasWritten: count > 0); - throw; } finally { @@ -1589,32 +1664,43 @@ public async ValueTask DeleteAsync(BsonId id, ITransaction? transaction, C if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (Delete)."); - transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); try { - var key = new IndexKey(id.ToBytes()); - if (!_primaryIndex.TryFind(key, out var location, transaction.TransactionId)) - return false; + for (int attempt = 1; ; attempt++) + { + transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); + try + { + var key = new IndexKey(id.ToBytes()); + if (!_primaryIndex.TryFind(key, out var location, transaction.TransactionId)) + return false; - var doc = ReadDocumentAt(location, transaction.TransactionId); - _primaryIndex.Delete(key, location, transaction.TransactionId); + var doc = ReadDocumentAt(location, transaction.TransactionId); + _primaryIndex.Delete(key, location, transaction.TransactionId); - if (doc != null) - { - foreach (var (_, idx) in _secondaryIndexes) - IndexDelete(idx, doc, location, transaction); - } + if (doc != null) + { + foreach (var (_, idx) in _secondaryIndexes) + IndexDelete(idx, doc, location, transaction); + } - DeleteSlot(location, transaction); - await NotifyCdcAsync(OperationType.Delete, id, transaction); - if (autoCommit) await transaction.CommitAsync(ct); - success = true; - return true; - } - catch (Exception ex) - { - await WriteFailure.HandleAsync(transaction, autoCommit, ex); - throw; + DeleteSlot(location, transaction); + await NotifyCdcAsync(OperationType.Delete, id, transaction); + if (autoCommit) await transaction.CommitAsync(ct); + success = true; + return true; + } + catch (Exception ex) when (WriteFailure.ShouldRetry(transaction, autoCommit, ex, attempt)) + { + // Rolled back and nothing written: run again in a new transaction (#304). + transaction = null; + } + catch (Exception ex) + { + await WriteFailure.HandleAsync(transaction, autoCommit, ex); + throw; + } + } } finally { @@ -1647,6 +1733,8 @@ public async ValueTask DeleteBulkAsync(IEnumerable ids, Cancellatio public async ValueTask DeleteBulkAsync(IEnumerable ids, ITransaction? transaction, CancellationToken ct = default) { if (ids == null) throw new ArgumentNullException(nameof(ids)); + // Read once: a write retried after a conflict (#304) runs over the same items. + var idList = ids.ToList(); bool autoCommit = transaction == null; WriteFailure.EnsureWritable(transaction); @@ -1654,37 +1742,48 @@ public async ValueTask DeleteBulkAsync(IEnumerable ids, ITransactio if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (DeleteBulk)."); - transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); try { - var count = 0; - foreach (var id in ids) + for (int attempt = 1; ; attempt++) { - ct.ThrowIfCancellationRequested(); - var key = new IndexKey(id.ToBytes()); - if (!_primaryIndex.TryFind(key, out var location, transaction.TransactionId)) - continue; + transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); + try + { + var count = 0; + foreach (var id in idList) + { + ct.ThrowIfCancellationRequested(); + var key = new IndexKey(id.ToBytes()); + if (!_primaryIndex.TryFind(key, out var location, transaction.TransactionId)) + continue; - var doc = ReadDocumentAt(location, transaction.TransactionId); - _primaryIndex.Delete(key, location, transaction.TransactionId); + var doc = ReadDocumentAt(location, transaction.TransactionId); + _primaryIndex.Delete(key, location, transaction.TransactionId); - if (doc != null) + if (doc != null) + { + foreach (var (_, idx) in _secondaryIndexes) + IndexDelete(idx, doc, location, transaction); + } + + DeleteSlot(location, transaction); + await NotifyCdcAsync(OperationType.Delete, id, transaction); + count++; + } + if (autoCommit) await transaction.CommitAsync(ct); + return count; + } + catch (Exception ex) when (WriteFailure.ShouldRetry(transaction, autoCommit, ex, attempt)) { - foreach (var (_, idx) in _secondaryIndexes) - IndexDelete(idx, doc, location, transaction); + // Rolled back and nothing written: run again in a new transaction (#304). + transaction = null; + } + catch (Exception ex) + { + await WriteFailure.HandleAsync(transaction, autoCommit, ex); + throw; } - - DeleteSlot(location, transaction); - await NotifyCdcAsync(OperationType.Delete, id, transaction); - count++; } - if (autoCommit) await transaction.CommitAsync(ct); - return count; - } - catch (Exception ex) - { - await WriteFailure.HandleAsync(transaction, autoCommit, ex); - throw; } finally { diff --git a/src/BLite.Core/Transactions/WriteFailure.cs b/src/BLite.Core/Transactions/WriteFailure.cs index cd46a7cc..fc9e5b40 100644 --- a/src/BLite.Core/Transactions/WriteFailure.cs +++ b/src/BLite.Core/Transactions/WriteFailure.cs @@ -44,4 +44,21 @@ internal static async ValueTask HandleAsync(ITransaction transaction, bool ownsT await transaction.RollbackAsync().ConfigureAwait(false); } + + /// Attempts a write without a caller-owned transaction makes before a conflict reaches its caller. + internal const int AutoCommitAttempts = 10; + + /// + /// True if a write method that opened itself must run again after + /// : its commit was rejected because a caller-owned transaction committed one + /// of its pages first (#304). Such commits do not take the collection lock, so they can land between + /// the write's reads and its commit. has rolled the transaction + /// back and nothing was written; the write method still holds the collection lock, so running it again + /// in a new transaction reads that commit's changes and cannot conflict with another such write. + /// + internal static bool ShouldRetry(ITransaction transaction, bool ownsTransaction, Exception error, int attempt) + => ownsTransaction + && attempt < AutoCommitAttempts + && error is BLiteTransactionConflictException conflict + && conflict.TransactionId == transaction.TransactionId; } diff --git a/tests/BLite.Tests/TransactionConflictTests.cs b/tests/BLite.Tests/TransactionConflictTests.cs index a6a62665..109831c8 100644 --- a/tests/BLite.Tests/TransactionConflictTests.cs +++ b/tests/BLite.Tests/TransactionConflictTests.cs @@ -1,5 +1,6 @@ using BLite.Bson; using BLite.Core; +using BLite.Core.Storage; using BLite.Core.Transactions; using BLite.Shared; @@ -279,4 +280,81 @@ public async Task Concurrent_auto_commit_writers_never_conflict() await Task.WhenAll(writers); Assert.Equal(100, await db.Users.CountAsync()); } + + // ── Review of #310 ───────────────────────────────────────────────────────────────────── + + /// A long lock timeout: these tests race writers on purpose and must not fail on #303. + private static readonly PageFileConfig LoadTolerant = PageFileConfig.Default with { LockTimeout = LockTimeout.From(30_000, 30_000) }; + + [Fact] + public async Task Typed_auto_commit_inserts_racing_explicit_transactions_retry_instead_of_throwing() + { + // An explicit commit does not take the collection lock, so it can land between an + // auto-commit insert's reads and its commit. That insert is rejected and must run again, + // not throw BLiteTransactionConflictException to a caller who never opened a transaction. + using var db = new TestDbContext(DbPath("typed-auto-vs-explicit"), LoadTolerant); + await db.Users.InsertAsync(new User { Name = "seed" }); + + var autoIds = new System.Collections.Concurrent.ConcurrentBag(); + var explicitIds = new System.Collections.Concurrent.ConcurrentBag(); + var autoWriters = Enumerable.Range(0, 3).Select(w => Task.Run(async () => + { + for (int i = 0; i < 100; i++) + { + var user = new User { Name = $"auto-{w}-{i}" }; + await db.Users.InsertAsync(user); + autoIds.Add(user.Id); + } + })); + var explicitWriters = Enumerable.Range(0, 3).Select(w => Task.Run(async () => + { + for (int i = 0; i < 100; i++) + { + using var t = db.BeginTransaction(); + var user = new User { Name = $"explicit-{w}-{i}" }; + await db.Users.InsertAsync(user, t); + if (!await CommitSecondAsync(t)) + explicitIds.Add(user.Id); + } + })); + await Task.WhenAll(autoWriters.Concat(explicitWriters)); + + Assert.Equal(300, autoIds.Count); + foreach (var id in autoIds.Concat(explicitIds)) + Assert.NotNull(await db.Users.FindByIdAsync(id)); + Assert.Equal(1 + 300 + explicitIds.Count, await db.Users.CountAsync()); + } + + [Fact] + public async Task Dynamic_auto_commit_inserts_racing_explicit_transactions_retry_instead_of_throwing() + { + using var engine = new BLiteEngine(DbPath("dynamic-auto-vs-explicit"), LoadTolerant); + engine.RegisterKeys(["n"]); + var col = engine.GetOrCreateCollection("items"); + await col.InsertAsync(engine.CreateDocument(["n"], b => b.AddInt32("n", -1))); + + var autoIds = new System.Collections.Concurrent.ConcurrentBag(); + var explicitIds = new System.Collections.Concurrent.ConcurrentBag(); + var autoWriters = Enumerable.Range(0, 3).Select(w => Task.Run(async () => + { + for (int i = 0; i < 100; i++) + autoIds.Add(await col.InsertAsync(engine.CreateDocument(["n"], b => b.AddInt32("n", i)))); + })); + var explicitWriters = Enumerable.Range(0, 3).Select(w => Task.Run(async () => + { + for (int i = 0; i < 100; i++) + { + using var t = engine.BeginTransaction(); + var id = await col.InsertAsync(engine.CreateDocument(["n"], b => b.AddInt32("n", i)), t); + if (!await CommitSecondAsync(t)) + explicitIds.Add(id); + } + })); + await Task.WhenAll(autoWriters.Concat(explicitWriters)); + + Assert.Equal(300, autoIds.Count); + foreach (var id in autoIds.Concat(explicitIds)) + Assert.NotNull(await col.FindByIdAsync(id)); + Assert.Equal(1 + 300 + explicitIds.Count, await col.CountAsync()); + } } From 973358accc71db2ddf3d251c08b4ea449a8db3bc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ricardo=20Gro=C3=9F?= Date: Mon, 19 Oct 2026 06:42:58 +0200 Subject: [PATCH 3/5] fix(storage): prune page versions after the batch's own commits (#304) TryPrunePageVersions runs under the commit lock before the committing transactions' waiters reach CompleteTransaction, so the batch's own base versions kept the map in use and the prune never ran: after 60k sequential inserts the map held 5,142 entries against a threshold of 4,096. Group commit now drops the base versions of each commit it published before pruning. Co-Authored-By: Claude Opus 5.5 --- .../Storage/StorageEngine.Conflicts.cs | 5 ++- .../Storage/StorageEngine.GroupCommit.cs | 5 +++ tests/BLite.Tests/TransactionConflictTests.cs | 31 +++++++++++++++++++ 3 files changed, 40 insertions(+), 1 deletion(-) diff --git a/src/BLite.Core/Storage/StorageEngine.Conflicts.cs b/src/BLite.Core/Storage/StorageEngine.Conflicts.cs index 86a7dc81..19730e65 100644 --- a/src/BLite.Core/Storage/StorageEngine.Conflicts.cs +++ b/src/BLite.Core/Storage/StorageEngine.Conflicts.cs @@ -40,7 +40,7 @@ public sealed partial class StorageEngine private readonly ConcurrentDictionary> _baseVersions = new(); /// Number of page versions above which a commit tries to forget them. - private const int PageVersionPruneThreshold = 4096; + internal const int PageVersionPruneThreshold = 4096; private long GetPageVersion(uint pageId) { @@ -146,4 +146,7 @@ private void ThrowIfConflicting(ulong transactionId, ConcurrentDictionaryNumber of active transactions with recorded page versions (diagnostics and tests). internal int TrackedTransactionCount => _baseVersions.Count; + + /// Number of pages with a version above the floor (diagnostics and tests). + internal int TrackedPageVersionCount => _pageVersions.Count; } diff --git a/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs b/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs index 8683f5b7..9b22da03 100644 --- a/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs +++ b/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs @@ -273,7 +273,12 @@ private async Task ProcessBatchAsync(List batch) foreach (var commit in batch) { if (conflicts == null || !conflicts.ContainsKey(commit)) + { FreePagesOfCommittedTransaction(commit.TransactionId); + // Its base versions are no longer needed. Left until its waiter reaches + // CompleteTransaction, they would keep the prune below from ever running (#304). + _baseVersions.TryRemove(commit.TransactionId, out _); + } } TryPrunePageVersions(); diff --git a/tests/BLite.Tests/TransactionConflictTests.cs b/tests/BLite.Tests/TransactionConflictTests.cs index 109831c8..9b74e38a 100644 --- a/tests/BLite.Tests/TransactionConflictTests.cs +++ b/tests/BLite.Tests/TransactionConflictTests.cs @@ -357,4 +357,35 @@ public async Task Dynamic_auto_commit_inserts_racing_explicit_transactions_retry Assert.NotNull(await col.FindByIdAsync(id)); Assert.Equal(1 + 300 + explicitIds.Count, await col.CountAsync()); } + + [Fact] + public async Task Page_versions_are_pruned_once_no_open_transaction_needs_them() + { + // Before, the prune ran while the committing transactions were still registered, so it + // never found the map unused: after enough commits it only grew. + using var storage = new StorageEngine(DbPath("prune"), PageFileConfig.Small); + var pages = Enumerable.Range(0, StorageEngine.PageVersionPruneThreshold + 1000).Select(_ => storage.AllocatePage()).ToArray(); + var data = new byte[storage.PageSize]; + foreach (var chunk in pages.Chunk(500)) + { + using var t = storage.BeginTransaction(); + foreach (var page in chunk) + storage.WritePage(page, t.TransactionId, data); + await t.CommitAsync(); + } + + Assert.InRange(storage.TrackedPageVersionCount, 1, StorageEngine.PageVersionPruneThreshold); + Assert.Equal(0, storage.TrackedTransactionCount); + + // Conflicts are still detected on the pages whose versions were forgotten. + var buffer = new byte[storage.PageSize]; + using var t1 = storage.BeginTransaction(); + using var t2 = storage.BeginTransaction(); + storage.ReadPage(pages[0], t1.TransactionId, buffer); + storage.ReadPage(pages[0], t2.TransactionId, buffer); + storage.WritePage(pages[0], t1.TransactionId, data); + storage.WritePage(pages[0], t2.TransactionId, data); + await t1.CommitAsync(); + await Assert.ThrowsAsync(async () => await t2.CommitAsync()); + } } From b41031ef41242e178b82adc2714efec532c214b7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ricardo=20Gro=C3=9F?= Date: Mon, 19 Oct 2026 06:43:19 +0200 Subject: [PATCH 4/5] fix(storage): close the remaining gaps of the conflict check (#304) - PrepareTransactionAsync (both overloads) votes no and writes nothing for a transaction whose commit would be rejected. MarkTransaction- CommittedAsync still checks again under the commit lock. - The exception's constructor is internal. - CommitTransactionAsync(Transaction) rolls the transaction back on a conflict, as Transaction.CommitAsync does, so that its uncommitted pages and base versions are not left behind. - Multi-process: when the catch-up with the shared WAL fails, the page versions do not include the other processes' commits, so group commit rejects every commit that writes pages instead of validating it against stale versions. Two MultiProcessWalTruncationTests committed from the engine whose catch-up failed; they now commit its page before the other engine's records exist and assert that the commit after the corruption is rejected and writes nothing. Co-Authored-By: Claude Opus 5.5 --- .../BLiteTransactionConflictException.cs | 27 ++++++++++-- .../Storage/StorageEngine.Conflicts.cs | 11 ++++- .../Storage/StorageEngine.GroupCommit.cs | 7 +++ .../Storage/StorageEngine.Transactions.cs | 15 +++++++ .../MultiProcessWalTruncationTests.cs | 26 +++++++---- tests/BLite.Tests/TransactionConflictTests.cs | 44 +++++++++++++++++++ 6 files changed, 117 insertions(+), 13 deletions(-) diff --git a/src/BLite.Core/BLiteTransactionConflictException.cs b/src/BLite.Core/BLiteTransactionConflictException.cs index 7fce95ff..dbb99f20 100644 --- a/src/BLite.Core/BLiteTransactionConflictException.cs +++ b/src/BLite.Core/BLiteTransactionConflictException.cs @@ -26,6 +26,11 @@ namespace BLite.Core; /// DynamicCollection.TruncateAsync) do not retry and can throw it; run them again. /// /// +/// In multi-process mode (PageFileConfig.AllowMultiProcessAccess) a commit that writes pages is +/// also rejected with this exception when the commits of other processes in the shared WAL could not +/// be read, so that it cannot be checked against them. +/// +/// /// Derives from , so code that already catches that type /// for commit failures keeps working. /// @@ -33,17 +38,33 @@ namespace BLite.Core; public class BLiteTransactionConflictException : InvalidOperationException { /// Creates the exception for and the first conflicting page. - public BLiteTransactionConflictException(ulong transactionId, uint pageId) - : base($"Transaction {transactionId} conflicts with a transaction that committed first: page {pageId} " + + internal BLiteTransactionConflictException(ulong transactionId, uint pageId) + : this(transactionId, pageId, + $"Transaction {transactionId} conflicts with a transaction that committed first: page {pageId} " + "was changed by both. The transaction was rolled back; run it again to retry.") + { + } + + private BLiteTransactionConflictException(ulong transactionId, uint pageId, string message) + : base(message) { TransactionId = transactionId; PageId = pageId; } + /// + /// Creates the exception for a commit that writes (its first page) while + /// this engine could not replay what other processes committed to the shared WAL. + /// + internal static BLiteTransactionConflictException SharedWalNotReplayed(ulong transactionId, uint pageId) + => new(transactionId, pageId, + $"Transaction {transactionId} was not committed: the commits of other processes in the shared WAL " + + $"could not be read, so its pages (first: page {pageId}) cannot be checked against them. " + + "The transaction was rolled back; run it again to retry."); + /// The transaction that was rolled back. public ulong TransactionId { get; } - /// The first page found that both transactions changed. + /// The first page found that both transactions changed, or the first page of a commit that could not be checked. public uint PageId { get; } } diff --git a/src/BLite.Core/Storage/StorageEngine.Conflicts.cs b/src/BLite.Core/Storage/StorageEngine.Conflicts.cs index 19730e65..e75dacc8 100644 --- a/src/BLite.Core/Storage/StorageEngine.Conflicts.cs +++ b/src/BLite.Core/Storage/StorageEngine.Conflicts.cs @@ -78,7 +78,8 @@ private void RecordFirstWrite(ulong transactionId, uint pageId) /// /// Returns the first page of that another commit has published since /// first saw it, or that - /// holds; null if there is none. Called under _commitLock. + /// holds; null if there is none. Called under _commitLock, except for the vote of + /// . /// private uint? FindConflict(ulong transactionId, ConcurrentDictionary pages, HashSet? writtenEarlierInBatch) { @@ -144,6 +145,14 @@ private void ThrowIfConflicting(ulong transactionId, ConcurrentDictionary + /// True if a commit of would be rejected now. Without + /// _commitLock, so only a vote for two-phase commit: a commit can still land before the + /// commit record, and checks again under the lock. + /// + private bool WouldConflict(ulong transactionId) + => _walCache.TryGetValue(transactionId, out var pages) && FindConflict(transactionId, pages, null) != null; + /// Number of active transactions with recorded page versions (diagnostics and tests). internal int TrackedTransactionCount => _baseVersions.Count; diff --git a/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs b/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs index 9b22da03..153e5189 100644 --- a/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs +++ b/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs @@ -181,6 +181,13 @@ private async Task ProcessBatchAsync(List batch) { if (commit.Pages == null || commit.Pages.IsEmpty) continue; + // Other processes' commits could not be replayed: the page versions do not include + // them, so no commit that writes pages can be checked. Rejected, to be retried. + if (wal != null && !caughtUp) + { + (conflicts ??= new()).Add(commit, BLiteTransactionConflictException.SharedWalNotReplayed(commit.TransactionId, commit.Pages.Keys.First())); + continue; + } if (FindConflict(commit.TransactionId, commit.Pages, writtenInBatch) is uint conflictPage) { (conflicts ??= new()).Add(commit, new BLiteTransactionConflictException(commit.TransactionId, conflictPage)); diff --git a/src/BLite.Core/Storage/StorageEngine.Transactions.cs b/src/BLite.Core/Storage/StorageEngine.Transactions.cs index f6069028..0ec09369 100644 --- a/src/BLite.Core/Storage/StorageEngine.Transactions.cs +++ b/src/BLite.Core/Storage/StorageEngine.Transactions.cs @@ -334,6 +334,13 @@ public async Task CommitTransactionAsync(Transaction transaction, CancellationTo { await CommitTransactionAsync(transaction.TransactionId, ct); } + catch (BLiteTransactionConflictException) + { + // Nothing was written (#304): roll back, as Transaction.CommitAsync does, so that its + // pages, base versions and the free-space index changes are not left behind. + await transaction.RollbackAsync().ConfigureAwait(false); + throw; + } finally { // Always clean up, even if the commit throws (timeout, etc.). @@ -365,6 +372,10 @@ public async Task PrepareTransactionAsync(ulong transactionId) ThrowIfMultiProcessAccess(nameof(PrepareTransactionAsync)); try { + // A commit that would be rejected as a conflict (#304) votes no and writes nothing. + if (WouldConflict(transactionId)) + return false; + await _wal.WriteBeginRecordAsync(transactionId); foreach (var walEntry in _walCache[transactionId]) @@ -388,6 +399,10 @@ public async Task PrepareTransactionAsync(ulong transactionId, Cancellatio ThrowIfMultiProcessAccess(nameof(PrepareTransactionAsync)); try { + // A commit that would be rejected as a conflict (#304) votes no and writes nothing. + if (WouldConflict(transactionId)) + return false; + await _wal.WriteBeginRecordAsync(transactionId, ct); if (_walCache.TryGetValue(transactionId, out var changes)) diff --git a/tests/BLite.Tests/MultiProcessWalTruncationTests.cs b/tests/BLite.Tests/MultiProcessWalTruncationTests.cs index 451739e4..4cc78144 100644 --- a/tests/BLite.Tests/MultiProcessWalTruncationTests.cs +++ b/tests/BLite.Tests/MultiProcessWalTruncationTests.cs @@ -1,4 +1,5 @@ using System.Diagnostics; +using BLite.Core; using BLite.Core.Storage; using BLite.Core.Transactions; @@ -289,19 +290,25 @@ public async Task UnreadableRecordInsideTheWal_StopsTheReplay_WithoutLettingTheR var p1 = a.AllocatePage(); var p2 = a.AllocatePage(); var p3 = a.AllocatePage(); + var p4 = a.AllocatePage(); - using (b.BeginTransaction()) { } // B's bookmark: the empty WAL + await WriteAndCommitAsync(b, p3, 0x33); // B's own page; B's bookmark: right after it + long firstRecordOfA = WalLength; await WriteAndCommitAsync(a, p1, 0x11); long secondRecord = WalLength; await WriteAndCommitAsync(a, p2, 0x22); // The first of A's records becomes unreadable for B. - CorruptWalByte(0); + CorruptWalByte(firstRecordOfA); - // B's replay stops at it, so it must not count A's records as replayed: committing and - // checkpointing its own page must leave the WAL, and with it A's records, in place. - await WriteAndCommitAsync(b, p3, 0x33); + // B cannot check a commit against records it cannot replay, so it rejects it and writes + // nothing (#304). long walBeforeCheckpoint = WalLength; + await Assert.ThrowsAsync(() => WriteAndCommitAsync(b, p4, 0x44)); + Assert.Equal(walBeforeCheckpoint, WalLength); + + // B's replay stops at it, so it must not count A's records as replayed: checkpointing its + // own page must leave the WAL, and with it A's records, in place. await b.CheckpointAsync(); Assert.Equal(walBeforeCheckpoint, WalLength); Assert.True(walBeforeCheckpoint > secondRecord); @@ -357,14 +364,15 @@ public async Task UnreadableRecordInsideThePublishedRange_IsReportedAsAStall() var p2 = a.AllocatePage(); var p3 = a.AllocatePage(); + long firstRecordOfA; var warnings = new WarningCollector(Path.ChangeExtension(_dbPath, ".wal")); Trace.Listeners.Add(warnings); try { - using (b.BeginTransaction()) { } // B's bookmark: the empty WAL + await WriteAndCommitAsync(b, p3, 0x33); // something for B to checkpoint; B's bookmark: right after it + firstRecordOfA = WalLength; await WriteAndCommitAsync(a, p1, 0x11); - CorruptWalByte(0); // A's record: unreadable for B, whose bookmark lies before it - await WriteAndCommitAsync(b, p3, 0x33); // something for B to checkpoint + CorruptWalByte(firstRecordOfA); // A's record: unreadable for B, whose bookmark lies before it await WriteAndCommitAsync(a, p2, 0x22); for (int i = 0; i < 2 * StorageEngine.CatchUpStallThreshold; i++) @@ -380,7 +388,7 @@ public async Task UnreadableRecordInsideThePublishedRange_IsReportedAsAStall() var messages = warnings.Messages; Assert.NotEmpty(messages); Assert.InRange(messages.Count, 2, 7); - Assert.All(messages, m => Assert.Contains("the record at offset 0 is unreadable", m)); + Assert.All(messages, m => Assert.Contains($"the record at offset {firstRecordOfA} is unreadable", m)); Assert.Contains(messages, m => m.Contains("stalled")); Assert.DoesNotContain("stalled", messages[0]); diff --git a/tests/BLite.Tests/TransactionConflictTests.cs b/tests/BLite.Tests/TransactionConflictTests.cs index 9b74e38a..99bf1e4a 100644 --- a/tests/BLite.Tests/TransactionConflictTests.cs +++ b/tests/BLite.Tests/TransactionConflictTests.cs @@ -388,4 +388,48 @@ public async Task Page_versions_are_pruned_once_no_open_transaction_needs_them() await t1.CommitAsync(); await Assert.ThrowsAsync(async () => await t2.CommitAsync()); } + + [Fact] + public async Task Prepare_votes_no_for_a_transaction_whose_commit_would_be_rejected() + { + using var db = new TestDbContext(DbPath("prepare")); + await db.Users.InsertAsync(new User { Name = "seed" }); + + var t1 = db.BeginTransaction(); + var t2 = db.BeginTransaction(); + await db.Users.InsertAsync(new User { Name = "from-t1" }, t1); + await db.Users.InsertAsync(new User { Name = "from-t2" }, t2); + await t1.CommitAsync(); + + // Same data page and primary leaf as in A_rejected_transaction_leaves_no_trace_and_can_be_retried. + Assert.False(await t2.PrepareAsync()); + await t2.RollbackAsync(); + Assert.Equal(2, await db.Users.CountAsync()); + t1.Dispose(); t2.Dispose(); + } + + [Fact] + public async Task A_conflict_in_the_engine_commit_overload_leaves_nothing_behind() + { + using var db = new TestDbContext(DbPath("engine-commit")); + await db.Users.InsertAsync(new User { Name = "seed" }); + var storage = db.Storage; + + // Committed through StorageEngine.CommitTransactionAsync(Transaction), not Transaction.CommitAsync. + var t1 = (Transaction)db.BeginTransaction(); + var t2 = (Transaction)db.BeginTransaction(); + await db.Users.InsertAsync(new User { Name = "from-t1" }, t1); + var loser = new User { Name = "from-t2" }; + await db.Users.InsertAsync(loser, t2); + await storage.CommitTransactionAsync(t1); + + await Assert.ThrowsAsync(() => storage.CommitTransactionAsync(t2)); + Assert.Equal(TransactionState.Aborted, t2.State); + Assert.Equal(0, storage.TrackedTransactionCount); // no base versions left + Assert.Equal(0, storage.ActiveTransactionCount); // no uncommitted pages left + Assert.Null(await db.Users.FindByIdAsync(loser.Id)); + + await db.Users.InsertAsync(new User { Name = "after" }); + Assert.Equal(3, await db.Users.CountAsync()); + } } From 3718934d2779e6122611ce09cdffbda7e7898f8e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ricardo=20Gro=C3=9F?= Date: Mon, 19 Oct 2026 06:43:30 +0200 Subject: [PATCH 5/5] test(storage): conflicts across processes and in collection-per-file mode (#304) - Two engines with AllowMultiProcessAccess on the same files: A's open transaction reads a page, B commits it, A writes it; A's commit is rejected, B's value stays, and A's retry goes through. - PageFileConfig.Server: GetCollectionWriteLock returns one instance per collection name, sessions' auto-commit inserts lose nothing, and two open transactions on one collection never both commit over each other. Co-Authored-By: Claude Opus 5.5 --- tests/BLite.Tests/TransactionConflictTests.cs | 88 +++++++++++++++++++ 1 file changed, 88 insertions(+) diff --git a/tests/BLite.Tests/TransactionConflictTests.cs b/tests/BLite.Tests/TransactionConflictTests.cs index 99bf1e4a..fab96b32 100644 --- a/tests/BLite.Tests/TransactionConflictTests.cs +++ b/tests/BLite.Tests/TransactionConflictTests.cs @@ -389,6 +389,73 @@ public async Task Page_versions_are_pruned_once_no_open_transaction_needs_them() await Assert.ThrowsAsync(async () => await t2.CommitAsync()); } + [Fact] + public async Task Multi_process_an_open_transaction_conflicts_with_a_commit_of_the_other_process() + { + // Two engines on the same files stand in for two processes. + var path = DbPath("multi-process"); + var config = PageFileConfig.Default with { AllowMultiProcessAccess = true }; + using var a = new StorageEngine(path, config); + using var b = new StorageEngine(path, config); + var page = a.AllocatePage(); + await CommitPageAsync(a, page, 0x01); + + using var ta = a.BeginTransaction(); + Assert.Equal(0x01, ReadPage(a, page, ta.TransactionId)); + await CommitPageAsync(b, page, 0x02); // B commits the page A's transaction read + a.WritePage(page, ta.TransactionId, Filled(a.PageSize, 0x03)); + + await Assert.ThrowsAsync(async () => await ta.CommitAsync()); + Assert.Equal(TransactionState.Aborted, ta.State); + using (var check = b.BeginTransaction()) + Assert.Equal(0x02, ReadPage(b, page, check.TransactionId)); + + // Run again: it reads B's commit first and goes through. + using (var retry = a.BeginTransaction()) + { + Assert.Equal(0x02, ReadPage(a, page, retry.TransactionId)); + a.WritePage(page, retry.TransactionId, Filled(a.PageSize, 0x03)); + await retry.CommitAsync(); + } + using (var check = b.BeginTransaction()) + Assert.Equal(0x03, ReadPage(b, page, check.TransactionId)); + } + + [Fact] + public async Task Collection_per_file_mode_shares_one_write_lock_per_collection() + { + var path = DbPath("per-file"); + using var engine = new BLiteEngine(path, PageFileConfig.Server(path) with { LockTimeout = LockTimeout.From(30_000, 30_000) }); + Assert.Same(engine.Storage.GetCollectionWriteLock("orders"), engine.Storage.GetCollectionWriteLock("ORDERS")); + Assert.NotSame(engine.Storage.GetCollectionWriteLock("orders"), engine.Storage.GetCollectionWriteLock("events")); + + // Every session builds its own DynamicCollection: with a lock per instance their + // auto-commit inserts ran concurrently on the same pages and were lost. + engine.RegisterKeys(["n"]); + for (int run = 0; run < 5; run++) + { + await Task.WhenAll(Enumerable.Range(0, 10).Select(async i => + { + using var session = engine.OpenSession(); + await session.InsertAsync("orders", engine.CreateDocument(["n"], x => x.AddInt32("n", i))); + })); + } + Assert.Equal(50, await engine.GetOrCreateCollection("orders").CountAsync()); + + // Open transactions on one collection still never both commit over each other. + var col = engine.GetOrCreateCollection("orders"); + var t1 = engine.BeginTransaction(); + var t2 = engine.BeginTransaction(); + var a = await col.InsertAsync(engine.CreateDocument(["n"], x => x.AddInt32("n", 100)), t1); + var b = await col.InsertAsync(engine.CreateDocument(["n"], x => x.AddInt32("n", 200)), t2); + await t1.CommitAsync(); + bool conflict = await CommitSecondAsync(t2); + Assert.NotNull(await col.FindByIdAsync(a)); + Assert.Equal(!conflict, await col.FindByIdAsync(b) is not null); + Assert.Equal(conflict ? 51 : 52, await col.CountAsync()); + t1.Dispose(); t2.Dispose(); + } + [Fact] public async Task Prepare_votes_no_for_a_transaction_whose_commit_would_be_rejected() { @@ -432,4 +499,25 @@ public async Task A_conflict_in_the_engine_commit_overload_leaves_nothing_behind await db.Users.InsertAsync(new User { Name = "after" }); Assert.Equal(3, await db.Users.CountAsync()); } + + private static async Task CommitPageAsync(StorageEngine engine, uint pageId, byte value) + { + using var tx = engine.BeginTransaction(); + engine.WritePage(pageId, tx.TransactionId, Filled(engine.PageSize, value)); + await tx.CommitAsync(); + } + + private static byte ReadPage(StorageEngine engine, uint pageId, ulong transactionId) + { + var buffer = new byte[engine.PageSize]; + engine.ReadPage(pageId, transactionId, buffer); + return buffer[0]; + } + + private static byte[] Filled(int size, byte value) + { + var data = new byte[size]; + Array.Fill(data, value); + return data; + } }