diff --git a/README.md b/README.md index f884d0e2..b524bbd0 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. --- @@ -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**: 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. 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. @@ -909,7 +909,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. @@ -1165,7 +1165,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..dbb99f20 --- /dev/null +++ b/src/BLite.Core/BLiteTransactionConflictException.cs @@ -0,0 +1,70 @@ +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. +/// 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. +/// +/// +/// 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. +/// +/// +public class BLiteTransactionConflictException : InvalidOperationException +{ + /// Creates the exception for and the first conflicting page. + 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, or the first page of a commit that could not be checked. + public uint PageId { get; } +} diff --git a/src/BLite.Core/Collections/DocumentCollection.cs b/src/BLite.Core/Collections/DocumentCollection.cs index f387bf12..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 @@ -848,7 +858,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. @@ -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 1715a719..f66033f4 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); @@ -862,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 { @@ -952,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); @@ -959,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 { @@ -1235,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 { @@ -1349,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 { @@ -1435,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); @@ -1442,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 { @@ -1524,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); @@ -1531,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 { @@ -1587,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 { @@ -1645,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); @@ -1652,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 { @@ -2899,6 +3000,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..e75dacc8 --- /dev/null +++ b/src/BLite.Core/Storage/StorageEngine.Conflicts.cs @@ -0,0 +1,161 @@ +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. + internal 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, except for the vote of + /// . + /// + 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); + } + + /// + /// 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; + + /// 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 cc66c764..153e5189 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,35 @@ 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; + // 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)); + 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 +233,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 +278,17 @@ 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); + // 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(); // Check if checkpoint is needed, but defer until after releasing the lock needsCheckpoint = newWalEnd > MaxWalSize; @@ -267,7 +310,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..0ec09369 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) { @@ -321,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.). @@ -352,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]) @@ -375,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)) @@ -432,6 +460,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 +481,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 +628,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 68c1b8dc..60d68742 100644 --- a/src/BLite.Core/Storage/StorageEngine.cs +++ b/src/BLite.Core/Storage/StorageEngine.cs @@ -152,13 +152,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/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/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 new file mode 100644 index 00000000..fab96b32 --- /dev/null +++ b/tests/BLite.Tests/TransactionConflictTests.cs @@ -0,0 +1,523 @@ +using BLite.Bson; +using BLite.Core; +using BLite.Core.Storage; +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()); + } + + // ── 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()); + } + + [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()); + } + + [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() + { + 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()); + } + + 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; + } +}