diff --git a/src/BLite.Core/BLiteValidationException.cs b/src/BLite.Core/BLiteValidationException.cs new file mode 100644 index 00000000..98b39067 --- /dev/null +++ b/src/BLite.Core/BLiteValidationException.cs @@ -0,0 +1,41 @@ +using System; + +namespace BLite.Core; + +/// +/// Thrown by a collection write method when it rejects its input before writing anything: +/// a duplicate primary key, a duplicate value for a unique index, a vector whose length does not +/// match its vector index, or a document or index key that is too large to be stored. +/// +/// +/// +/// This is the only exception after which a write that ran in a caller-owned transaction leaves +/// that transaction usable. A single-document write (insert, update, upsert) that throws it wrote +/// nothing, so the caller can catch it, carry on with the same transaction and commit. A bulk write +/// that throws it on its first document wrote nothing either; one that throws it after earlier +/// documents of the batch were written rolls the transaction back like any other failure. +/// +/// +/// Any other exception thrown while a write method is modifying pages rolls the caller's +/// transaction back ( becomes +/// ), because the failed write may already have +/// updated the data page and some of the indexes but not the others, and committing that would +/// leave the indexes out of step with the data. +/// +/// +/// Derives from , the type these errors had before, except +/// that a vector dimension mismatch used to surface as an . +/// +/// +public class BLiteValidationException : InvalidOperationException +{ + /// Creates the exception with a message describing the rejected input. + public BLiteValidationException(string message) : base(message) + { + } + + /// Creates the exception with a message and the error that caused it. + public BLiteValidationException(string message, Exception innerException) : base(message, innerException) + { + } +} diff --git a/src/BLite.Core/Collections/DocumentCollection.cs b/src/BLite.Core/Collections/DocumentCollection.cs index 2ab1a32b..2c5223cf 100644 --- a/src/BLite.Core/Collections/DocumentCollection.cs +++ b/src/BLite.Core/Collections/DocumentCollection.cs @@ -752,11 +752,18 @@ public Task TruncateAsync(CancellationToken ct = default) => TruncateAsync(null, ct); /// + /// + /// With a caller-owned , a failure while writing rolls the + /// transaction back ( becomes ), + /// so that a half-applied delete can never be committed. + /// public async Task TruncateAsync(ITransaction? transaction, CancellationToken ct = default) { int deleted = 0; bool autoCommit = transaction == null; + WriteFailure.EnsureWritable(transaction); + if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct).ConfigureAwait(false)) throw new TimeoutException("Timed out acquiring collection lock (TruncateAsync)."); @@ -787,10 +794,9 @@ public async Task TruncateAsync(ITransaction? transaction, CancellationToke if (autoCommit) await transaction.CommitAsync(ct).ConfigureAwait(false); } - catch + catch (Exception ex) { - if (autoCommit) - await transaction.RollbackAsync().ConfigureAwait(false); + await WriteFailure.HandleAsync(transaction, autoCommit, ex); throw; } } @@ -2059,6 +2065,14 @@ private uint AllocateOverflowPage(ReadOnlySpan data, uint nextOverflowPage /// /// Asynchronously inserts a new document into the collection /// + /// + /// With a caller-owned : if the call throws + /// (duplicate key, vector dimension mismatch, document or + /// key too large), it wrote nothing and the transaction stays active, so the caller can catch it, + /// carry on and commit. Any other failure while writing rolls the transaction back + /// ( becomes ), so that a + /// half-applied write, with the data and its indexes out of step, can never be committed. + /// public async ValueTask InsertAsync(T entity, ITransaction? transaction, CancellationToken ct = default) { if (entity == null) throw new ArgumentNullException(nameof(entity)); @@ -2067,6 +2081,8 @@ public async ValueTask InsertAsync(T entity, ITransaction? transaction, Can bool success = false; bool autoCommit = transaction == null; + WriteFailure.EnsureWritable(transaction); + if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (Insert)."); @@ -2092,9 +2108,9 @@ public async ValueTask InsertAsync(T entity, ITransaction? transaction, Can success = true; return id; } - catch + catch (Exception ex) { - await transaction.RollbackAsync(); + await WriteFailure.HandleAsync(transaction, autoCommit, ex); throw; } } @@ -2116,6 +2132,14 @@ public async ValueTask InsertAsync(T entity, ITransaction? transaction, Can /// /// Asynchronously inserts multiple documents in a single transaction. /// + /// + /// With a caller-owned the call is all or nothing: if the first + /// document is rejected with (duplicate key, vector + /// dimension mismatch, document or key too large), nothing was written and the transaction stays + /// active. Any other failure, including a rejected document after earlier documents of the batch + /// were written, rolls the transaction back ( becomes + /// ). + /// public async ValueTask> InsertBulkAsync(IEnumerable entities, ITransaction? transaction, CancellationToken ct = default) { if (entities == null) throw new ArgumentNullException(nameof(entities)); @@ -2123,6 +2147,8 @@ public async ValueTask> InsertBulkAsync(IEnumerable entities, ITran var entityList = entities.ToList(); var ids = new List(entityList.Count); + WriteFailure.EnsureWritable(transaction); + if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (InsertBulk)."); @@ -2148,9 +2174,9 @@ public async ValueTask> InsertBulkAsync(IEnumerable entities, ITran } return ids; } - catch + catch (Exception ex) { - await transaction.RollbackAsync(); + await WriteFailure.HandleAsync(transaction, autoCommit, ex, callHasWritten: ids.Count > 0); throw; } } @@ -2261,8 +2287,12 @@ private async Task InsertDataCore(TId id, T entity, byte[] docData, ITransaction : default; // ──────────────────────────────────────────────────────────────────── - // Unique secondary indexes are checked before any page is written. + // Everything that can reject the document is checked before any page is written, so a + // rejected insert leaves nothing behind (BLiteValidationException keeps the transaction usable). + var key = _mapper.ToIndexKey(id); + ValidatePrimaryKeyForInsert(id, key, transaction); _indexManager.EnsureUniqueInAll(entity, null, transaction); + _indexManager.ValidateKeysInAll(entity); DocumentLocation location; if (_isTimeSeries) @@ -2286,7 +2316,6 @@ private async Task InsertDataCore(TId id, T entity, byte[] docData, ITransaction location = new DocumentLocation(pageId, slotIndex); } - var key = _mapper.ToIndexKey(id); _primaryIndex.Insert(key, location, transaction.TransactionId); _indexManager.InsertIntoAll(entity, location, transaction); @@ -2322,6 +2351,21 @@ private async Task InsertDataCore(TId id, T entity, byte[] docData, ITransaction await NotifyCdc(OperationType.Insert, id, transaction, docData); } + /// + /// Throws if cannot be inserted + /// into the primary index: it is already taken, or it is larger than a node page. + /// + private void ValidatePrimaryKeyForInsert(TId id, IndexKey key, ITransaction transaction) + { + if (key.Data.Length > _primaryIndex.MaxKeySize) + throw new BLiteValidationException( + $"Id of {key.Data.Length} bytes is too large for the primary index of collection '{_collectionName}' (maximum {_primaryIndex.MaxKeySize} bytes)."); + + if (_primaryIndex.TryFind(key, out _, transaction.TransactionId)) + throw new BLiteValidationException( + $"Duplicate key violation: a document with id '{id}' already exists in collection '{_collectionName}'."); + } + /// /// Compacts the data area of a slotted page by reclaiming bytes occupied by deleted slots. /// Live documents are packed contiguously toward the top of the page and each slot's Offset @@ -2683,6 +2727,14 @@ public async IAsyncEnumerable FindAllAsync(ITransaction? transaction, [System /// /// Asynchronously updates an existing document in the collection /// + /// + /// With a caller-owned : if the call throws + /// (duplicate key, vector dimension mismatch, document or + /// key too large), it wrote nothing and the transaction stays active, so the caller can catch it, + /// carry on and commit. Any other failure while writing rolls the transaction back + /// ( becomes ), so that a + /// half-applied write, with the data and its indexes out of step, can never be committed. + /// public async ValueTask UpdateAsync(T entity, ITransaction? transaction, CancellationToken ct = default) { if (entity == null) throw new ArgumentNullException(nameof(entity)); @@ -2691,6 +2743,8 @@ public async ValueTask UpdateAsync(T entity, ITransaction? transaction, Ca bool success = false; bool autoCommit = transaction == null; + WriteFailure.EnsureWritable(transaction); + if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (Update)."); @@ -2704,9 +2758,9 @@ public async ValueTask UpdateAsync(T entity, ITransaction? transaction, Ca success = result; return result; } - catch + catch (Exception ex) { - if (autoCommit) await transaction.RollbackAsync(); + await WriteFailure.HandleAsync(transaction, autoCommit, ex); throw; } } @@ -2728,6 +2782,14 @@ public async ValueTask UpdateAsync(T entity, ITransaction? transaction, Ca /// /// Asynchronously updates multiple documents in a single transaction. /// + /// + /// With a caller-owned the call is all or nothing: if the first + /// document is rejected with (duplicate key, vector + /// dimension mismatch, document or key too large), nothing was written and the transaction stays + /// active. Any other failure, including a rejected document after earlier documents of the batch + /// were written, rolls the transaction back ( becomes + /// ). + /// public async ValueTask UpdateBulkAsync(IEnumerable entities, ITransaction? transaction, CancellationToken ct = default) { if (entities == null) throw new ArgumentNullException(nameof(entities)); @@ -2735,6 +2797,10 @@ public async ValueTask UpdateBulkAsync(IEnumerable entities, ITransactio var entityList = entities.ToList(); int updateCount = 0; bool autoCommit = transaction == null; + // Documents of this call written so far: a failure after the first one must roll back. + var progress = new StrongBox(); + + WriteFailure.EnsureWritable(transaction); if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (UpdateBulk)."); @@ -2744,13 +2810,13 @@ public async ValueTask UpdateBulkAsync(IEnumerable entities, ITransactio { try { - updateCount = await UpdateBulkInternal(entityList, transaction); + updateCount = await UpdateBulkInternal(entityList, transaction, progress); if (autoCommit) await transaction.CommitAsync(ct); return updateCount; } - catch + catch (Exception ex) { - if (autoCommit) await transaction.RollbackAsync(); + await WriteFailure.HandleAsync(transaction, autoCommit, ex, callHasWritten: progress.Value > 0); throw; } } @@ -2760,7 +2826,7 @@ public async ValueTask UpdateBulkAsync(IEnumerable entities, ITransactio } } - private async Task UpdateBulkInternal(List entityList, ITransaction transaction) + private async Task UpdateBulkInternal(List entityList, ITransaction transaction, StrongBox progress) { int updateCount = 0; const int BATCH_SIZE = 50; @@ -2802,7 +2868,7 @@ private async Task UpdateBulkInternal(List entityList, ITransaction tran try { if (await UpdateDataCore(id, entity, docData, transaction, length)) - updateCount++; + progress.Value = ++updateCount; } finally { @@ -2852,9 +2918,11 @@ private async Task UpdateAtLocationCore(TId id, T entity, byte[] docData, var oldEntity = await FindByLocation(oldLocation, transaction); if (oldEntity == null) return false; - // Unique secondary indexes are checked before any page is written; only those whose - // key changed, so documents that already share a value can still be updated. + // Everything that can reject the document is checked before any page is written. + // Unique indexes only where the key changed, so documents that already share a value + // can still be updated. _indexManager.EnsureUniqueInAll(entity, oldEntity, transaction); + _indexManager.ValidateKeysInAll(entity); // Read old page var pageBuffer = ArrayPool.Shared.Rent(_storage.PageSize); @@ -2924,6 +2992,14 @@ public ValueTask> UpsertAsync(T entity, CancellationToken ct = => UpsertAsync(entity, null, ct); /// + /// + /// With a caller-owned : if the call throws + /// (duplicate key, vector dimension mismatch, document or + /// key too large), it wrote nothing and the transaction stays active, so the caller can catch it, + /// carry on and commit. Any other failure while writing rolls the transaction back + /// ( becomes ), so that a + /// half-applied write, with the data and its indexes out of step, can never be committed. + /// public async ValueTask> UpsertAsync(T entity, ITransaction? transaction, CancellationToken ct = default) { if (entity == null) throw new ArgumentNullException(nameof(entity)); @@ -2933,6 +3009,8 @@ public async ValueTask> UpsertAsync(T entity, ITransaction? tr bool inserted = false; bool autoCommit = transaction == null; + WriteFailure.EnsureWritable(transaction); + if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (Upsert)."); @@ -2958,9 +3036,9 @@ public async ValueTask> UpsertAsync(T entity, ITransaction? tr success = true; return result; } - catch + catch (Exception ex) { - await transaction.RollbackAsync(); + await WriteFailure.HandleAsync(transaction, autoCommit, ex); throw; } } @@ -2987,6 +3065,14 @@ public ValueTask>> UpsertBulkAsync(IEnumerable entitie => UpsertBulkAsync(entities, null, ct); /// + /// + /// With a caller-owned the call is all or nothing: if the first + /// document is rejected with (duplicate key, vector + /// dimension mismatch, document or key too large), nothing was written and the transaction stays + /// active. Any other failure, including a rejected document after earlier documents of the batch + /// were written, rolls the transaction back ( becomes + /// ). + /// public async ValueTask>> UpsertBulkAsync(IEnumerable entities, ITransaction? transaction, CancellationToken ct = default) { if (entities == null) throw new ArgumentNullException(nameof(entities)); @@ -2995,6 +3081,8 @@ public async ValueTask>> UpsertBulkAsync(IEnumerable e var results = new List>(entityList.Count); bool autoCommit = transaction == null; + WriteFailure.EnsureWritable(transaction); + if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (UpsertBulk)."); @@ -3024,9 +3112,9 @@ public async ValueTask>> UpsertBulkAsync(IEnumerable e } return results; } - catch + catch (Exception ex) { - await transaction.RollbackAsync(); + await WriteFailure.HandleAsync(transaction, autoCommit, ex, callHasWritten: results.Count > 0); throw; } } @@ -3077,12 +3165,19 @@ private async Task> UpsertCore(T entity, ITransaction transact /// /// Asynchronously deletes a document by its primary key. /// + /// + /// With a caller-owned , a failure while writing rolls the + /// transaction back ( becomes ), + /// so that a half-applied delete can never be committed. + /// public async ValueTask DeleteAsync(TId id, ITransaction? transaction, CancellationToken ct = default) { var sw = _storage.MetricsDispatcher != null ? ValueStopwatch.StartNew() : default; bool success = false; bool autoCommit = transaction == null; + WriteFailure.EnsureWritable(transaction); + if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (Delete)."); @@ -3096,9 +3191,9 @@ public async ValueTask DeleteAsync(TId id, ITransaction? transaction, Canc success = result; return result; } - catch + catch (Exception ex) { - if (autoCommit) await transaction.RollbackAsync(); + await WriteFailure.HandleAsync(transaction, autoCommit, ex); throw; } } @@ -3120,6 +3215,11 @@ public async ValueTask DeleteAsync(TId id, ITransaction? transaction, Canc /// /// Asynchronously deletes multiple documents in a single transaction. /// + /// + /// With a caller-owned , a failure while writing rolls the + /// transaction back ( becomes ), + /// so that a half-applied delete can never be committed. + /// public async ValueTask DeleteBulkAsync(IEnumerable ids, ITransaction? transaction, CancellationToken ct = default) { if (ids == null) throw new ArgumentNullException(nameof(ids)); @@ -3127,6 +3227,8 @@ public async ValueTask DeleteBulkAsync(IEnumerable ids, ITransaction? int deleteCount = 0; bool autoCommit = transaction == null; + WriteFailure.EnsureWritable(transaction); + if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (DeleteBulk)."); @@ -3139,9 +3241,9 @@ public async ValueTask DeleteBulkAsync(IEnumerable ids, ITransaction? if (autoCommit) await transaction.CommitAsync(ct); return deleteCount; } - catch + catch (Exception ex) { - if (autoCommit) await transaction.RollbackAsync(); + await WriteFailure.HandleAsync(transaction, autoCommit, ex); throw; } } @@ -3766,7 +3868,7 @@ private int SerializeWithRetry(T entity, out byte[] rentedBuffer) } rentedBuffer = null!; // specific compiler satisfaction, though we throw - throw new InvalidOperationException($"Document too large. Maximum size allowed is 16MB."); + throw new BLiteValidationException($"Document too large. Maximum size allowed is 16MB."); } /// diff --git a/src/BLite.Core/Collections/IDocumentCollection.cs b/src/BLite.Core/Collections/IDocumentCollection.cs index ad615cfb..77719c7f 100644 --- a/src/BLite.Core/Collections/IDocumentCollection.cs +++ b/src/BLite.Core/Collections/IDocumentCollection.cs @@ -31,9 +31,25 @@ public interface IDocumentCollection where T : class // ── Insert ──────────────────────────────────────────────────────────────── ValueTask InsertAsync(T entity, CancellationToken ct = default); + /// + /// With a caller-owned : if the call throws + /// (duplicate key, vector dimension mismatch, document or + /// key too large), it wrote nothing and the transaction stays active, so the caller can catch it, + /// carry on and commit. Any other failure while writing rolls the transaction back + /// ( becomes ), so that a + /// half-applied write, with the data and its indexes out of step, can never be committed. + /// ValueTask InsertAsync(T entity, ITransaction? transaction, CancellationToken ct = default); ValueTask> InsertBulkAsync(IEnumerable entities, CancellationToken ct = default); + /// + /// With a caller-owned the call is all or nothing: if the first + /// document is rejected with (duplicate key, vector + /// dimension mismatch, document or key too large), nothing was written and the transaction stays + /// active. Any other failure, including a rejected document after earlier documents of the batch + /// were written, rolls the transaction back ( becomes + /// ). + /// ValueTask> InsertBulkAsync(IEnumerable entities, ITransaction? transaction, CancellationToken ct = default); // ── Read ────────────────────────────────────────────────────────────────── @@ -72,9 +88,25 @@ public interface IDocumentCollection where T : class // ── Update ──────────────────────────────────────────────────────────────── ValueTask UpdateAsync(T entity, CancellationToken ct = default); + /// + /// With a caller-owned : if the call throws + /// (duplicate key, vector dimension mismatch, document or + /// key too large), it wrote nothing and the transaction stays active, so the caller can catch it, + /// carry on and commit. Any other failure while writing rolls the transaction back + /// ( becomes ), so that a + /// half-applied write, with the data and its indexes out of step, can never be committed. + /// ValueTask UpdateAsync(T entity, ITransaction? transaction, CancellationToken ct = default); ValueTask UpdateBulkAsync(IEnumerable entities, CancellationToken ct = default); + /// + /// With a caller-owned the call is all or nothing: if the first + /// document is rejected with (duplicate key, vector + /// dimension mismatch, document or key too large), nothing was written and the transaction stays + /// active. Any other failure, including a rejected document after earlier documents of the batch + /// were written, rolls the transaction back ( becomes + /// ). + /// ValueTask UpdateBulkAsync(IEnumerable entities, ITransaction? transaction, CancellationToken ct = default); // ── Upsert ──────────────────────────────────────────────────────────────── @@ -83,17 +115,43 @@ public interface IDocumentCollection where T : class // per document — not a Find followed by a separate Insert/Update call. ValueTask> UpsertAsync(T entity, CancellationToken ct = default); + /// + /// With a caller-owned : if the call throws + /// (duplicate key, vector dimension mismatch, document or + /// key too large), it wrote nothing and the transaction stays active, so the caller can catch it, + /// carry on and commit. Any other failure while writing rolls the transaction back + /// ( becomes ), so that a + /// half-applied write, with the data and its indexes out of step, can never be committed. + /// ValueTask> UpsertAsync(T entity, ITransaction? transaction, CancellationToken ct = default); ValueTask>> UpsertBulkAsync(IEnumerable entities, CancellationToken ct = default); + /// + /// With a caller-owned the call is all or nothing: if the first + /// document is rejected with (duplicate key, vector + /// dimension mismatch, document or key too large), nothing was written and the transaction stays + /// active. Any other failure, including a rejected document after earlier documents of the batch + /// were written, rolls the transaction back ( becomes + /// ). + /// ValueTask>> UpsertBulkAsync(IEnumerable entities, ITransaction? transaction, CancellationToken ct = default); // ── Delete ──────────────────────────────────────────────────────────────── ValueTask DeleteAsync(TId id, CancellationToken ct = default); + /// + /// With a caller-owned , a failure while writing rolls the + /// transaction back ( becomes ), + /// so that a half-applied delete can never be committed. + /// ValueTask DeleteAsync(TId id, ITransaction? transaction, CancellationToken ct = default); ValueTask DeleteBulkAsync(IEnumerable ids, CancellationToken ct = default); + /// + /// With a caller-owned , a failure while writing rolls the + /// transaction back ( becomes ), + /// so that a half-applied delete can never be committed. + /// ValueTask DeleteBulkAsync(IEnumerable ids, ITransaction? transaction, CancellationToken ct = default); // ── Index management ────────────────────────────────────────────────────── @@ -185,6 +243,11 @@ IAsyncEnumerable ParallelScanAsync( Task TruncateAsync(CancellationToken ct = default); /// + /// + /// With a caller-owned , a failure while writing rolls the + /// transaction back ( becomes ), + /// so that a half-applied delete can never be committed. + /// Task TruncateAsync(ITransaction? transaction, CancellationToken ct = default); // ── TimeSeries (local engine only; remote throws NotSupportedException) ─── diff --git a/src/BLite.Core/DynamicCollection.cs b/src/BLite.Core/DynamicCollection.cs index 67966dd6..a25711be 100644 --- a/src/BLite.Core/DynamicCollection.cs +++ b/src/BLite.Core/DynamicCollection.cs @@ -599,6 +599,11 @@ private async Task InsertCore(BsonDocument document, ITransaction transa document = PrependId(document, id); } + // Everything that can reject the document is checked before any page is written, so a + // rejected insert leaves nothing behind (BLiteValidationException keeps the transaction usable). + var key = new IndexKey(id.ToBytes()); + ValidateWrite(id, key, document, oldDocument: null, isInsert: true, transaction); + // Write raw BSON to storage var docData = document.RawData; DocumentLocation location = default; @@ -629,11 +634,11 @@ private async Task InsertCore(BsonDocument document, ITransaction transa } else { - throw new InvalidOperationException($"Document size {docData.Length} exceeds maximum single page size. Overflow not yet supported in DynamicCollection."); + // Unreachable: ValidateWrite rejects documents that do not fit in a page. + throw new BLiteValidationException($"Document size {docData.Length} exceeds maximum single page size. Overflow not yet supported in DynamicCollection."); } // Index the _id - var key = new IndexKey(id.ToBytes()); _primaryIndex.Insert(key, location, transaction.TransactionId); // UpdateAsync secondary indexes @@ -688,6 +693,14 @@ public ValueTask InsertAsync(BsonDocument document, CancellationToken ct /// If the document has no _id field, one is auto-generated. /// Returns the BsonId of the inserted document. /// + /// + /// With a caller-owned : if the call throws + /// (duplicate key, vector dimension mismatch, document or + /// key too large), it wrote nothing and the transaction stays active, so the caller can catch it, + /// carry on and commit. Any other failure while writing rolls the transaction back + /// ( becomes ), so that a + /// half-applied write, with the data and its indexes out of step, can never be committed. + /// public async ValueTask InsertAsync(BsonDocument document, ITransaction? transaction, CancellationToken ct = default) { if (document == null) throw new ArgumentNullException(nameof(document)); @@ -702,6 +715,8 @@ public async ValueTask InsertAsync(BsonDocument document, ITransaction? bool success = false; bool autoCommit = transaction == null; + WriteFailure.EnsureWritable(transaction); + if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (Insert)."); @@ -725,9 +740,9 @@ public async ValueTask InsertAsync(BsonDocument document, ITransaction? success = true; return result; } - catch + catch (Exception ex) { - await transaction.RollbackAsync(); + await WriteFailure.HandleAsync(transaction, autoCommit, ex); throw; } finally @@ -784,18 +799,28 @@ public ValueTask> InsertBulkAsync(IEnumerable documen /// Inserts multiple BsonDocuments asynchronously in a single transaction. /// Returns the list of generated/existing BsonIds in insertion order. /// + /// + /// With a caller-owned the call is all or nothing: if the first + /// document is rejected with (duplicate key, vector + /// dimension mismatch, document or key too large), nothing was written and the transaction stays + /// active. Any other failure, including a rejected document after earlier documents of the batch + /// were written, rolls the transaction back ( becomes + /// ). + /// public async ValueTask> InsertBulkAsync(IEnumerable documents, ITransaction? transaction, CancellationToken ct = default) { if (documents == null) throw new ArgumentNullException(nameof(documents)); bool autoCommit = transaction == null; + WriteFailure.EnsureWritable(transaction); + 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 { - var ids = new List(); foreach (var doc in documents) { ct.ThrowIfCancellationRequested(); @@ -816,9 +841,9 @@ public async ValueTask> InsertBulkAsync(IEnumerable d } return ids; } - catch + catch (Exception ex) { - await transaction.RollbackAsync(); + await WriteFailure.HandleAsync(transaction, autoCommit, ex, callHasWritten: ids.Count > 0); throw; } finally @@ -1008,6 +1033,14 @@ public ValueTask UpdateAsync(BsonId id, BsonDocument newDocument, Cancella /// /// Updates a document by its BsonId asynchronously. Replaces the entire document. /// + /// + /// With a caller-owned : if the call throws + /// (duplicate key, vector dimension mismatch, document or + /// key too large), it wrote nothing and the transaction stays active, so the caller can catch it, + /// carry on and commit. Any other failure while writing rolls the transaction back + /// ( becomes ), so that a + /// half-applied write, with the data and its indexes out of step, can never be committed. + /// public async ValueTask UpdateAsync(BsonId id, BsonDocument newDocument, ITransaction? transaction, CancellationToken ct = default) { if (newDocument == null) throw new ArgumentNullException(nameof(newDocument)); @@ -1016,6 +1049,8 @@ public async ValueTask UpdateAsync(BsonId id, BsonDocument newDocument, IT bool success = false; bool autoCommit = transaction == null; + WriteFailure.EnsureWritable(transaction); + if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (Update)."); @@ -1032,9 +1067,9 @@ public async ValueTask UpdateAsync(BsonId id, BsonDocument newDocument, IT success = true; return true; } - catch + catch (Exception ex) { - await transaction.RollbackAsync(); + await WriteFailure.HandleAsync(transaction, autoCommit, ex); throw; } finally @@ -1061,24 +1096,20 @@ public async ValueTask UpdateAsync(BsonId id, BsonDocument newDocument, IT private async Task UpdateAtLocationCore(BsonId id, IndexKey key, BsonDocument newDocument, DocumentLocation oldLocation, ITransaction transaction) { var oldDoc = ReadDocumentAt(oldLocation, transaction.TransactionId); - DeleteSlot(oldLocation, transaction); if (!newDocument.TryGetId(out _)) newDocument = PrependId(newDocument, id); + // Everything that can reject the document is checked before any page is written. + ValidateWrite(id, key, newDocument, oldDoc, isInsert: false, transaction); + + DeleteSlot(oldLocation, transaction); + var docData = newDocument.RawData; - DocumentLocation newLocation; - if (docData.Length + SlotEntry.Size <= _maxDocumentSizeForSinglePage) - { - var pageId = FindPageWithSpace(docData.Length + SlotEntry.Size, transaction.TransactionId); - if (pageId == 0) pageId = AllocateNewDataPage(transaction); - var slotIndex = InsertIntoPage(pageId, docData, transaction); - newLocation = new DocumentLocation(pageId, slotIndex); - } - else - { - throw new InvalidOperationException("Document too large for single page. Overflow not yet supported in DynamicCollection."); - } + var pageId = FindPageWithSpace(docData.Length + SlotEntry.Size, transaction.TransactionId); + if (pageId == 0) pageId = AllocateNewDataPage(transaction); + var slotIndex = InsertIntoPage(pageId, docData, transaction); + var newLocation = new DocumentLocation(pageId, slotIndex); _primaryIndex.Delete(key, oldLocation, transaction.TransactionId); _primaryIndex.Insert(key, newLocation, transaction.TransactionId); @@ -1099,6 +1130,14 @@ public ValueTask> UpsertAsync(BsonDocument document, Cancel => UpsertAsync(document, null, ct); /// + /// + /// With a caller-owned : if the call throws + /// (duplicate key, vector dimension mismatch, document or + /// key too large), it wrote nothing and the transaction stays active, so the caller can catch it, + /// carry on and commit. Any other failure while writing rolls the transaction back + /// ( becomes ), so that a + /// half-applied write, with the data and its indexes out of step, can never be committed. + /// public async ValueTask> UpsertAsync(BsonDocument document, ITransaction? transaction, CancellationToken ct = default) { if (document == null) throw new ArgumentNullException(nameof(document)); @@ -1112,6 +1151,8 @@ public async ValueTask> UpsertAsync(BsonDocument document, bool inserted = false; bool autoCommit = transaction == null; + WriteFailure.EnsureWritable(transaction); + if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (Upsert)."); @@ -1134,9 +1175,9 @@ public async ValueTask> UpsertAsync(BsonDocument document, success = true; return result; } - catch + catch (Exception ex) { - await transaction.RollbackAsync(); + await WriteFailure.HandleAsync(transaction, autoCommit, ex); throw; } finally @@ -1190,18 +1231,28 @@ public ValueTask>> UpsertBulkAsync(IEnumerable UpsertBulkAsync(documents, null, ct); /// + /// + /// With a caller-owned the call is all or nothing: if the first + /// document is rejected with (duplicate key, vector + /// dimension mismatch, document or key too large), nothing was written and the transaction stays + /// active. Any other failure, including a rejected document after earlier documents of the batch + /// were written, rolls the transaction back ( becomes + /// ). + /// public async ValueTask>> UpsertBulkAsync(IEnumerable documents, ITransaction? transaction, CancellationToken ct = default) { if (documents == null) throw new ArgumentNullException(nameof(documents)); bool autoCommit = transaction == null; + WriteFailure.EnsureWritable(transaction); + if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (UpsertBulk)."); transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); + var results = new List>(); try { - var results = new List>(); bool anyInserted = false; foreach (var doc in documents) { @@ -1224,9 +1275,9 @@ public async ValueTask>> UpsertBulkAsync(IEnumerable 0); throw; } finally @@ -1269,18 +1320,28 @@ public ValueTask UpdateBulkAsync(IEnumerable<(BsonId Id, BsonDocument Docum /// Updates multiple documents asynchronously in a single transaction. /// Returns the number of documents successfully updated. /// + /// + /// With a caller-owned the call is all or nothing: if the first + /// document is rejected with (duplicate key, vector + /// dimension mismatch, document or key too large), nothing was written and the transaction stays + /// active. Any other failure, including a rejected document after earlier documents of the batch + /// were written, rolls the transaction back ( becomes + /// ). + /// public async ValueTask UpdateBulkAsync(IEnumerable<(BsonId Id, BsonDocument Document)> updates, ITransaction? transaction, CancellationToken ct = default) { if (updates == null) throw new ArgumentNullException(nameof(updates)); bool autoCommit = transaction == null; + WriteFailure.EnsureWritable(transaction); + if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (UpdateBulk)."); transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted); + var count = 0; try { - var count = 0; foreach (var (id, doc) in updates) { ct.ThrowIfCancellationRequested(); @@ -1288,37 +1349,15 @@ public async ValueTask UpdateBulkAsync(IEnumerable<(BsonId Id, BsonDocument if (!_primaryIndex.TryFind(key, out var oldLocation, transaction.TransactionId)) continue; - var oldDoc = ReadDocumentAt(oldLocation, transaction.TransactionId); - DeleteSlot(oldLocation, transaction); - - var newDoc = doc; - if (!newDoc.TryGetId(out _)) - newDoc = PrependId(newDoc, id); - - var docData = newDoc.RawData; - if (docData.Length + SlotEntry.Size > _maxDocumentSizeForSinglePage) - throw new InvalidOperationException("Document too large for single page."); - - var pageId = FindPageWithSpace(docData.Length + SlotEntry.Size, transaction.TransactionId); - if (pageId == 0) pageId = AllocateNewDataPage(transaction); - var slotIndex = InsertIntoPage(pageId, docData, transaction); - var newLocation = new DocumentLocation(pageId, slotIndex); - - _primaryIndex.Delete(key, oldLocation, transaction.TransactionId); - _primaryIndex.Insert(key, newLocation, transaction.TransactionId); - - foreach (var (_, idx) in _secondaryIndexes) - IndexUpdate(idx, oldDoc, newDoc, oldLocation, newLocation, transaction); - - await NotifyCdcAsync(OperationType.Update, id, transaction, newDoc.RawData); + await UpdateAtLocationCore(id, key, doc, oldLocation, transaction); count++; } if (autoCommit) await transaction.CommitAsync(ct); return count; } - catch + catch (Exception ex) { - await transaction.RollbackAsync(); + await WriteFailure.HandleAsync(transaction, autoCommit, ex, callHasWritten: count > 0); throw; } finally @@ -1339,12 +1378,19 @@ public ValueTask DeleteAsync(BsonId id, CancellationToken ct = default) /// /// Deletes a document by its BsonId asynchronously. /// + /// + /// With a caller-owned , a failure while writing rolls the + /// transaction back ( becomes ), + /// so that a half-applied delete can never be committed. + /// public async ValueTask DeleteAsync(BsonId id, ITransaction? transaction, CancellationToken ct = default) { var sw = _storage.MetricsDispatcher != null ? Metrics.ValueStopwatch.StartNew() : default; bool success = false; bool autoCommit = transaction == null; + WriteFailure.EnsureWritable(transaction); + if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (Delete)."); @@ -1370,9 +1416,9 @@ public async ValueTask DeleteAsync(BsonId id, ITransaction? transaction, C success = true; return true; } - catch + catch (Exception ex) { - await transaction.RollbackAsync(); + await WriteFailure.HandleAsync(transaction, autoCommit, ex); throw; } finally @@ -1398,11 +1444,18 @@ public async ValueTask DeleteBulkAsync(IEnumerable ids, Cancellatio /// Deletes multiple documents asynchronously in a single transaction. /// Returns the number of documents successfully deleted. /// + /// + /// With a caller-owned , a failure while writing rolls the + /// transaction back ( becomes ), + /// so that a half-applied delete can never be committed. + /// public async ValueTask DeleteBulkAsync(IEnumerable ids, ITransaction? transaction, CancellationToken ct = default) { if (ids == null) throw new ArgumentNullException(nameof(ids)); bool autoCommit = transaction == null; + WriteFailure.EnsureWritable(transaction); + if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct)) throw new TimeoutException("Timed out acquiring collection lock (DeleteBulk)."); @@ -1433,9 +1486,9 @@ public async ValueTask DeleteBulkAsync(IEnumerable ids, ITransactio if (autoCommit) await transaction.CommitAsync(ct); return count; } - catch + catch (Exception ex) { - await transaction.RollbackAsync(); + await WriteFailure.HandleAsync(transaction, autoCommit, ex); throw; } finally @@ -1848,6 +1901,76 @@ internal void PersistIndexMetadata() // ── Index dispatch helpers ──────────────────────────────────────────────── + /// + /// Throws if cannot be written: + /// it is too large for a page, its id is already taken (insert) or too large, a unique index + /// already holds its value, a vector does not match its index dimensions, or a B-Tree key is + /// larger than a node page. Called before any page is written, so a rejected write leaves the + /// data and every index untouched instead of failing halfway through the indexes. + /// + /// Stored version of the document on update; null on insert. + private void ValidateWrite(BsonId id, IndexKey key, BsonDocument document, BsonDocument? oldDocument, bool isInsert, ITransaction transaction) + { + var size = document.RawData.Length; + if (!_isTimeSeries && size + SlotEntry.Size > _maxDocumentSizeForSinglePage) + throw new BLiteValidationException( + $"Document size {size} exceeds maximum single page size. Overflow not yet supported in DynamicCollection."); + + if (key.Data.Length > _primaryIndex.MaxKeySize) + throw new BLiteValidationException( + $"Id of {key.Data.Length} bytes is too large for the primary index of collection '{_collectionName}' (maximum {_primaryIndex.MaxKeySize} bytes)."); + + if (isInsert && _primaryIndex.TryFind(key, out _, transaction.TransactionId)) + throw new BLiteValidationException( + $"Duplicate key violation: a document with _id '{id}' already exists in collection '{_collectionName}'."); + + foreach (var (name, idx) in _secondaryIndexes) + { + if (!TryGetNestedValue(document, idx.FieldPath, out var val)) continue; + + switch (idx.Kind) + { + case DynamicIndexKind.BTree: + { + if (BsonValueToIndexKey(val) is not { } indexKey) break; + + if (indexKey.Data.Length > idx.BTree!.MaxKeySize) + throw new BLiteValidationException( + $"Key of {indexKey.Data.Length} bytes is too large for index '{name}' (maximum {idx.BTree.MaxKeySize} bytes)."); + + if (!idx.Options.Unique) break; + + // On update an unchanged key keeps its entry (IndexUpdate deletes it first), + // so documents that already share a value can still be updated. + if (oldDocument != null + && TryGetNestedValue(oldDocument, idx.FieldPath, out var oldVal) + && BsonValueToIndexKey(oldVal) is { } oldKey + && oldKey.Data.SequenceEqual(indexKey.Data)) + break; + + if (idx.BTree.TryFind(indexKey, out _, transaction.TransactionId)) + throw new BLiteValidationException($"Duplicate key violation for unique index '{name}'"); + break; + } + case DynamicIndexKind.Vector: + { + float[]? vector; + try { vector = ExtractFloatVector(val); } + catch (InvalidOperationException ex) + { + throw new BLiteValidationException( + $"Field '{idx.FieldPath}' of vector index '{name}' is not a numeric array.", ex); + } + + if (vector != null && vector.Length != idx.Options.Dimensions) + throw new BLiteValidationException( + $"Vector dimension mismatch for index '{name}'. Expected {idx.Options.Dimensions}, got {vector.Length}"); + break; + } + } + } + } + private void IndexInsert(DynamicSecondaryIndex idx, BsonDocument document, DocumentLocation location, ITransaction transaction) { // Support nested paths with dot-notation (e.g., "address.city.name") diff --git a/src/BLite.Core/Indexing/BTreeIndex.cs b/src/BLite.Core/Indexing/BTreeIndex.cs index d7ac1e29..def4b6d0 100644 --- a/src/BLite.Core/Indexing/BTreeIndex.cs +++ b/src/BLite.Core/Indexing/BTreeIndex.cs @@ -77,6 +77,15 @@ public BTreeIndex(StorageEngine storage, /// The root page of this index. It is fixed for the lifetime of the index. public uint RootPageId => _rootPageId; + /// + /// The largest key that fits in a node page on its own: page header (32) + node header (20) + /// + the leftmost child pointer of an internal node (4) + one entry (4-byte length, key, + /// 4-byte child pointer). A larger key can never be inserted, so the collections reject it + /// before writing anything. Nodes split by entry count, not by size, so many keys below + /// this limit can still overflow a node; that is caught as a failure during the write. + /// + internal int MaxKeySize => _storage.PageSize - 32 - 20 - 4 - 4 - 4; + /// /// Reads a page using StorageEngine for transaction isolation. /// Implements "Read Your Own Writes" isolation. diff --git a/src/BLite.Core/Indexing/CollectionIndexManager.cs b/src/BLite.Core/Indexing/CollectionIndexManager.cs index 7dddfb0f..fd576583 100644 --- a/src/BLite.Core/Indexing/CollectionIndexManager.cs +++ b/src/BLite.Core/Indexing/CollectionIndexManager.cs @@ -545,6 +545,25 @@ public void EnsureUniqueInAll(T document, T? oldDocument, ITransaction transacti } } + /// + /// Throws if 's key could not + /// be written to one of the indexes (vector dimension mismatch, key too large). + /// Call before any page is written, together with . + /// + public void ValidateKeysInAll(T document) + { + if (document == null) + throw new ArgumentNullException(nameof(document)); + + lock (_lock) + { + foreach (var index in _indexes.Values) + { + index.ValidateKey(document); + } + } + } + /// /// Updates a document in all indexes /// diff --git a/src/BLite.Core/Indexing/CollectionSecondaryIndex.cs b/src/BLite.Core/Indexing/CollectionSecondaryIndex.cs index 2c7a65ce..ec59904d 100644 --- a/src/BLite.Core/Indexing/CollectionSecondaryIndex.cs +++ b/src/BLite.Core/Indexing/CollectionSecondaryIndex.cs @@ -536,11 +536,48 @@ internal void EnsureUnique(T document, T? oldDocument, ITransaction? transaction if (s_fixedSizeId && idPart.Length != ownIdKey.Data.Length) continue; - throw new InvalidOperationException( + throw new BLiteValidationException( $"Duplicate key violation for unique index '{_definition.Name}'"); } } + /// + /// Throws if 's key could not + /// be written to this index: a vector whose length differs from the index dimensions, or a + /// B-Tree key larger than a node page. Called before any page is written, so a rejected write + /// leaves nothing behind instead of failing halfway through the indexes. + /// + internal void ValidateKey(T document) + { + if (!_definition.TryGetKey(document, out var keyValue) || keyValue == null) + return; // null keys are not indexed + + if (_vectorIndex != null) + { + if (keyValue is float[] singleVector) + ValidateVector(singleVector); + else if (keyValue is IEnumerable vectors) + foreach (var v in vectors) ValidateVector(v); + } + else if (_btreeIndex != null && !keyValue.GetType().IsValueType) + { + // Value-type keys (numbers, dates, Guid, ObjectId, enums) have a small fixed size; + // only strings, byte arrays and the ToString fallback can grow past a page. + var keyLength = ConvertToIndexKey(keyValue).Data.Length + + _mapper.ToIndexKey(_mapper.GetId(document)).Data.Length; + if (keyLength > _btreeIndex.MaxKeySize) + throw new BLiteValidationException( + $"Key of {keyLength} bytes is too large for index '{_definition.Name}' (maximum {_btreeIndex.MaxKeySize} bytes)."); + } + } + + private void ValidateVector(float[] vector) + { + if (vector.Length != _definition.Dimensions) + throw new BLiteValidationException( + $"Vector dimension mismatch for index '{_definition.Name}'. Expected {_definition.Dimensions}, got {vector.Length}"); + } + // [user key][one byte]: a range bound between the entries of user key and those of longer values. private static IndexKey CreateIdPrefixBoundary(IndexKey userKey, byte idPrefixBound) { diff --git a/src/BLite.Core/Transactions/ITransaction.cs b/src/BLite.Core/Transactions/ITransaction.cs index a6bf8c44..6503e566 100644 --- a/src/BLite.Core/Transactions/ITransaction.cs +++ b/src/BLite.Core/Transactions/ITransaction.cs @@ -16,6 +16,30 @@ namespace BLite.Core.Transactions; /// txn.Commit(); /// } /// +/// +/// +/// When a collection write method that was passed this transaction fails, what the caller can +/// rely on depends on the exception: +/// +/// +/// +/// (duplicate primary or unique key, vector dimension +/// mismatch, document or index key too large): the input was rejected before anything was +/// written. The transaction is still ; earlier writes in it +/// are intact and can be committed. For a bulk method this holds only if the first document +/// was rejected; otherwise the transaction is rolled back as below. +/// +/// +/// Any other exception thrown while the method was writing: the write may have been applied +/// to the data page and some indexes but not others, so the method rolls the transaction back +/// ( becomes ). Later writes with it +/// throw , and so does . +/// +/// +/// +/// There are no per-operation savepoints: a write that succeeded cannot be undone on its own. +/// +/// public interface ITransaction : IDisposable { /// diff --git a/src/BLite.Core/Transactions/WriteFailure.cs b/src/BLite.Core/Transactions/WriteFailure.cs new file mode 100644 index 00000000..cd46a7cc --- /dev/null +++ b/src/BLite.Core/Transactions/WriteFailure.cs @@ -0,0 +1,47 @@ +using System; +using System.Threading.Tasks; + +namespace BLite.Core.Transactions; + +/// +/// Shared failure handling for the collection write methods (typed and dynamic). +/// +internal static class WriteFailure +{ + /// + /// Throws if a caller-owned can no longer take writes, e.g. because + /// an earlier failed write rolled it back. Writing into it anyway would not fail at the storage + /// level: the pages would land in a cache entry that is never committed or released. + /// + internal static void EnsureWritable(ITransaction? transaction) + { + if (transaction != null && transaction.State != TransactionState.Active) + throw new InvalidOperationException( + $"Cannot write in transaction {transaction.TransactionId}: its state is {transaction.State}. " + + "A failed write rolls the transaction back unless it threw BLiteValidationException."); + } + + /// + /// Called from the catch block of a write method. Rolls back unless + /// the failure left it clean, which is only the case for a + /// in a caller-owned transaction when the call had not written anything yet. A write that failed + /// after its first page write may have updated the data and some indexes but not the others, so + /// committing the transaction afterwards would leave the indexes out of step with the data. + /// A transaction that is no longer active is left alone: it was committed (a step after the + /// commit failed, e.g. OnInsert retention) or has already been rolled back. + /// + /// The transaction the write ran in. + /// True if the write method opened the transaction itself. + /// The exception being handled. + /// True if an earlier document of a bulk call was already written. + internal static async ValueTask HandleAsync(ITransaction transaction, bool ownsTransaction, Exception error, bool callHasWritten = false) + { + if (transaction.State != TransactionState.Active && transaction.State != TransactionState.Preparing) + return; + + if (!ownsTransaction && !callHasWritten && error is BLiteValidationException) + return; + + await transaction.RollbackAsync().ConfigureAwait(false); + } +} diff --git a/tests/BLite.Tests/BLiteEngineTests.cs b/tests/BLite.Tests/BLiteEngineTests.cs index 65fff6ce..2661d8af 100644 --- a/tests/BLite.Tests/BLiteEngineTests.cs +++ b/tests/BLite.Tests/BLiteEngineTests.cs @@ -101,7 +101,7 @@ await col.InsertAsync(col.CreateDocument(["_id", "name"], b => b .AddString("name", "Alice"))); await _engine.CommitAsync(); - var ex = await Assert.ThrowsAsync(async () => + var ex = await Assert.ThrowsAsync(async () => await col.InsertAsync(col.CreateDocument(["_id", "name"], b => b .AddId((BsonId)1) .AddString("name", "Bob")))); diff --git a/tests/BLite.Tests/DocumentCollectionTests.cs b/tests/BLite.Tests/DocumentCollectionTests.cs index 1bacc587..a84b3047 100644 --- a/tests/BLite.Tests/DocumentCollectionTests.cs +++ b/tests/BLite.Tests/DocumentCollectionTests.cs @@ -43,7 +43,7 @@ public async Task Insert_With_Duplicate_Id_Throws() await _db.Users.InsertAsync(new User { Id = id, Name = "Alice", Age = 30 }); await _db.SaveChangesAsync(); - var ex = await Assert.ThrowsAsync(async () => + var ex = await Assert.ThrowsAsync(async () => await _db.Users.InsertAsync(new User { Id = id, Name = "Bob", Age = 31 })); Assert.Contains("Duplicate key violation", ex.Message); Assert.Equal(1, await _db.Users.CountAsync()); diff --git a/tests/BLite.Tests/ExplicitTransactionFailureTests.cs b/tests/BLite.Tests/ExplicitTransactionFailureTests.cs new file mode 100644 index 00000000..8899536b --- /dev/null +++ b/tests/BLite.Tests/ExplicitTransactionFailureTests.cs @@ -0,0 +1,524 @@ +using BLite.Bson; +using BLite.Core; +using BLite.Core.Storage; +using BLite.Core.Transactions; +using BLite.Shared; + +namespace BLite.Tests; + +/// +/// A write that fails inside a caller-owned transaction (#185). +/// +/// Input rejected up front () leaves nothing behind and +/// the transaction usable: the caller can carry on and commit the other writes. +/// Any other failure while writing rolls the transaction back, so a half-applied write can +/// never be committed. +/// +/// After every commit the tests check that the data and every index agree: the primary count, +/// each index's entry count and lookups, and the live slots on the data pages (a failed insert must +/// not leave an orphaned slot behind). +/// +public class ExplicitTransactionFailureTests : IDisposable +{ + private readonly string _dbPath; + + public ExplicitTransactionFailureTests() + { + _dbPath = Path.Combine(Path.GetTempPath(), $"blite_tx_failure_{Guid.NewGuid():N}.db"); + } + + public void Dispose() + { + foreach (var p in new[] { _dbPath, Path.ChangeExtension(_dbPath, ".wal"), _dbPath + "-shm" }) + if (File.Exists(p)) File.Delete(p); + } + + // ── Typed collections: validation failures keep the transaction usable ─── + + [Fact] + public async Task InsertAsync_Duplicate_Id_Keeps_Transaction_And_Leaves_No_Orphan_Slot() + { + using var db = new TestDbContext(_dbPath); + using var tx = await db.BeginTransactionAsync(); + + await db.AsyncDocs.InsertAsync(new AsyncDoc { Id = 1, Name = "first" }, tx); + + var ex = await Assert.ThrowsAsync(async () => + await db.AsyncDocs.InsertAsync(new AsyncDoc { Id = 1, Name = "duplicate" }, tx)); + Assert.Contains("Duplicate key violation", ex.Message); + Assert.Equal(TransactionState.Active, tx.State); + + await db.AsyncDocs.InsertAsync(new AsyncDoc { Id = 2, Name = "second" }, tx); + await tx.CommitAsync(); + + Assert.Equal("first", (await db.AsyncDocs.FindByIdAsync(1))?.Name); + Assert.Equal("second", (await db.AsyncDocs.FindByIdAsync(2))?.Name); + Assert.Equal(2, await db.AsyncDocs.CountAsync()); + Assert.Equal(2, CountLiveSlots(db.Storage, db.AsyncDocs.CollectionName)); + } + + [Fact] + public async Task InsertBulkAsync_Rejected_First_Document_Keeps_Transaction() + { + using var db = new TestDbContext(_dbPath); + using var tx = await db.BeginTransactionAsync(); + + await db.AsyncDocs.InsertAsync(new AsyncDoc { Id = 1, Name = "first" }, tx); + + await Assert.ThrowsAsync(async () => + await db.AsyncDocs.InsertBulkAsync([new AsyncDoc { Id = 1, Name = "duplicate" }], tx)); + Assert.Equal(TransactionState.Active, tx.State); + + await tx.CommitAsync(); + + Assert.Equal("first", (await db.AsyncDocs.FindByIdAsync(1))?.Name); + Assert.Equal(1, await db.AsyncDocs.CountAsync()); + Assert.Equal(1, CountLiveSlots(db.Storage, db.AsyncDocs.CollectionName)); + } + + [Fact] + public async Task InsertBulkAsync_Rejected_Later_Document_Rolls_Back_Transaction() + { + using var db = new TestDbContext(_dbPath); + await db.AsyncDocs.InsertAsync(new AsyncDoc { Id = 1, Name = "committed" }); + + using var tx = await db.BeginTransactionAsync(); + await db.AsyncDocs.InsertAsync(new AsyncDoc { Id = 2, Name = "earlier in tx" }, tx); + + // Document 3 is written before document 1 is rejected: the call cannot be undone on + // its own, so the whole transaction is rolled back. + await Assert.ThrowsAsync(async () => + await db.AsyncDocs.InsertBulkAsync( + [new AsyncDoc { Id = 3, Name = "written" }, new AsyncDoc { Id = 1, Name = "duplicate" }], tx)); + Assert.Equal(TransactionState.Aborted, tx.State); + await Assert.ThrowsAsync(async () => await tx.CommitAsync()); + + Assert.Equal(1, await db.AsyncDocs.CountAsync()); + Assert.Equal("committed", (await db.AsyncDocs.FindByIdAsync(1))?.Name); + Assert.Null(await db.AsyncDocs.FindByIdAsync(2)); + Assert.Null(await db.AsyncDocs.FindByIdAsync(3)); + } + + [Fact] + public async Task UpsertAsync_Bad_Embedding_Leaves_Every_Index_Consistent() + { + using var db = new TestDbContext(_dbPath); + // idx_vector (3 dimensions) is declared on the model; idx_title comes after it. + await db.VectorItems.CreateIndexAsync(x => x.Title, "idx_title"); + using var tx = await db.BeginTransactionAsync(); + + var good = new VectorEntity { Title = "good", Embedding = [1f, 2f, 3f] }; + await db.VectorItems.UpsertAsync(good, tx); + + await Assert.ThrowsAsync(async () => + await db.VectorItems.UpsertAsync(new VectorEntity { Title = "bad", Embedding = [1f, 2f] }, tx)); + Assert.Equal(TransactionState.Active, tx.State); + + await tx.CommitAsync(); + + Assert.Equal("good", (await db.VectorItems.FindByIdAsync(good.Id))?.Title); + await AssertTypedVectorItemsConsistent(db, expectedCount: 1); + } + + [Fact] + public async Task UpsertBulkAsync_Bad_Embedding_First_Keeps_Transaction() + { + using var db = new TestDbContext(_dbPath); + await db.VectorItems.CreateIndexAsync(x => x.Title, "idx_title"); + using var tx = await db.BeginTransactionAsync(); + + var good = new VectorEntity { Title = "good", Embedding = [1f, 2f, 3f] }; + await db.VectorItems.UpsertAsync(good, tx); + + await Assert.ThrowsAsync(async () => + await db.VectorItems.UpsertBulkAsync([new VectorEntity { Title = "bad", Embedding = [1f, 2f] }], tx)); + + await tx.CommitAsync(); + + Assert.Equal("good", (await db.VectorItems.FindByIdAsync(good.Id))?.Title); + await AssertTypedVectorItemsConsistent(db, expectedCount: 1); + } + + [Fact] + public async Task UpdateAsync_Bad_Embedding_Leaves_Every_Index_Consistent() + { + using var db = new TestDbContext(_dbPath); + await db.VectorItems.CreateIndexAsync(x => x.Title, "idx_title"); + var a = new VectorEntity { Title = "a", Embedding = [1f, 2f, 3f] }; + var b = new VectorEntity { Title = "b", Embedding = [4f, 5f, 6f] }; + await db.VectorItems.InsertAsync(a); + await db.VectorItems.InsertAsync(b); + + using var tx = await db.BeginTransactionAsync(); + a.Title = "a2"; + Assert.True(await db.VectorItems.UpdateAsync(a, tx)); + + await Assert.ThrowsAsync(async () => + await db.VectorItems.UpdateAsync(new VectorEntity { Id = b.Id, Title = "z", Embedding = [1f, 2f] }, tx)); + Assert.Equal(TransactionState.Active, tx.State); + + await tx.CommitAsync(); + + Assert.Equal("a2", (await db.VectorItems.FindByIdAsync(a.Id))?.Title); + Assert.Equal("b", (await db.VectorItems.FindByIdAsync(b.Id))?.Title); + Assert.Empty(await db.VectorItems.QueryIndexAsync("idx_title", "z", "z").ToListAsync()); + await AssertTypedVectorItemsConsistent(db, expectedCount: 2); + } + + [Fact] + public async Task UniqueIndex_Violation_Keeps_Transaction() + { + using var db = new TestDbContext(_dbPath); + await db.PeopleV2.CreateIndexAsync(p => p.Name, "idx_name", unique: true); + using var tx = await db.BeginTransactionAsync(); + + await db.PeopleV2.InsertAsync(new PersonV2 { Name = "dup", Age = 1 }, tx); + await Assert.ThrowsAsync(async () => + await db.PeopleV2.InsertAsync(new PersonV2 { Name = "dup", Age = 2 }, tx)); + await db.PeopleV2.InsertAsync(new PersonV2 { Name = "other", Age = 3 }, tx); + await tx.CommitAsync(); + + Assert.Equal(2, await db.PeopleV2.CountAsync()); + Assert.Equal(2, (await db.PeopleV2.QueryIndexAsync("idx_name", null, null).ToListAsync()).Count); + // PersonV2 is declared with an index on Age. + Assert.Equal(2, (await db.PeopleV2.QueryIndexAsync("idx_Age", null, null).ToListAsync()).Count); + Assert.Equal(2, CountLiveSlots(db.Storage, db.PeopleV2.CollectionName)); + } + + // ── Typed collections: other failures roll the transaction back ────────── + + [Fact] + public async Task Failure_After_The_First_Page_Write_Rolls_Back_Callers_Transaction() + { + using var db = new TestDbContext(_dbPath); + await db.PeopleV2.CreateIndexAsync(p => p.Name, "idx_name"); + await db.PeopleV2.InsertAsync(new PersonV2 { Name = "committed", Age = 1 }); + + using var tx = await db.BeginTransactionAsync(); + var ex = await InsertUntilBTreeNodeOverflows(i => + db.PeopleV2.InsertAsync(new PersonV2 { Name = LongKey(i), Age = i }, tx).AsTask()); + + // The failing insert had already written its data page and the primary index. + Assert.IsNotType(ex); + Assert.Equal(TransactionState.Aborted, tx.State); + await Assert.ThrowsAsync(async () => + await db.PeopleV2.InsertAsync(new PersonV2 { Name = "after", Age = 2 }, tx)); + await Assert.ThrowsAsync(async () => await tx.CommitAsync()); + + Assert.Equal(1, await db.PeopleV2.CountAsync()); + Assert.Single(await db.PeopleV2.QueryIndexAsync("idx_name", null, null).ToListAsync()); + Assert.Equal(1, CountLiveSlots(db.Storage, db.PeopleV2.CollectionName)); + } + + [Fact] + public async Task Implicit_Transaction_Is_Still_Rolled_Back_On_Failure() + { + using var db = new TestDbContext(_dbPath); + + await db.AsyncDocs.InsertAsync(new AsyncDoc { Id = 1, Name = "first" }); + await Assert.ThrowsAsync(async () => + await db.AsyncDocs.InsertAsync(new AsyncDoc { Id = 1, Name = "duplicate" })); + + // A later auto-committed write must not be blocked by a dangling transaction. + await db.AsyncDocs.InsertAsync(new AsyncDoc { Id = 2, Name = "second" }); + + Assert.Equal("first", (await db.AsyncDocs.FindByIdAsync(1))?.Name); + Assert.Equal("second", (await db.AsyncDocs.FindByIdAsync(2))?.Name); + Assert.Equal(2, CountLiveSlots(db.Storage, db.AsyncDocs.CollectionName)); + } + + // ── Dynamic collections: validation failures keep the transaction usable ─ + + [Fact] + public async Task Dynamic_InsertAsync_Duplicate_Id_Keeps_Transaction_And_Leaves_No_Orphan_Slot() + { + using var engine = new BLiteEngine(_dbPath); + var col = engine.GetOrCreateCollection("docs", BsonIdType.Int32); + var tx = await engine.BeginTransactionAsync(); + + await col.InsertAsync(Doc(col, 1, "first"), tx); + + var ex = await Assert.ThrowsAsync(async () => + await col.InsertAsync(Doc(col, 1, "duplicate"), tx)); + Assert.Contains("Duplicate key violation", ex.Message); + Assert.Equal(TransactionState.Active, tx.State); + + await col.InsertAsync(Doc(col, 2, "second"), tx); + await tx.CommitAsync(); + + Assert.Equal("first", await NameOf(col, 1)); + Assert.Equal("second", await NameOf(col, 2)); + Assert.Equal(2, await col.CountAsync()); + Assert.Equal(2, CountLiveSlots(engine.Storage, col.Name)); + } + + [Fact] + public async Task Dynamic_InsertBulkAsync_Rejected_First_Document_Keeps_Transaction() + { + using var engine = new BLiteEngine(_dbPath); + var col = engine.GetOrCreateCollection("docs", BsonIdType.Int32); + var tx = await engine.BeginTransactionAsync(); + + await col.InsertAsync(Doc(col, 1, "first"), tx); + + await Assert.ThrowsAsync(async () => + await col.InsertBulkAsync([Doc(col, 1, "duplicate")], tx)); + + await tx.CommitAsync(); + + Assert.Equal("first", await NameOf(col, 1)); + Assert.Equal(1, await col.CountAsync()); + Assert.Equal(1, CountLiveSlots(engine.Storage, col.Name)); + } + + [Fact] + public async Task Dynamic_UniqueViolation_After_Another_Index_Leaves_Every_Index_Consistent() + { + using var engine = new BLiteEngine(_dbPath); + var col = engine.GetOrCreateCollection("people", BsonIdType.Int32); + // idx_email is maintained after idx_tag, so a late duplicate check would leave the + // document in the primary index and in idx_tag. + await col.CreateIndexAsync("tag", "idx_tag"); + await col.CreateIndexAsync("email", "idx_email", unique: true); + var tx = await engine.BeginTransactionAsync(); + + await col.InsertAsync(Person(col, 1, "a@x", "t"), tx); + var ex = await Assert.ThrowsAsync(async () => + await col.InsertAsync(Person(col, 2, "a@x", "t"), tx)); + Assert.Contains("Duplicate key violation", ex.Message); + await col.InsertAsync(Person(col, 3, "b@x", "t"), tx); + await tx.CommitAsync(); + + Assert.Null(await col.FindByIdAsync((BsonId)2)); + await AssertDynamicBTreeIndexConsistent(col, "idx_tag", "tag"); + await AssertDynamicBTreeIndexConsistent(col, "idx_email", "email"); + Assert.Equal(2, await col.CountAsync()); + Assert.Equal(2, CountLiveSlots(engine.Storage, col.Name)); + } + + [Fact] + public async Task Dynamic_Vector_Dimension_Mismatch_After_BTree_Index_Leaves_Every_Index_Consistent() + { + using var engine = new BLiteEngine(_dbPath); + var col = engine.GetOrCreateCollection("vecs", BsonIdType.Int32); + await col.CreateIndexAsync("tag", "idx_tag"); + await col.CreateVectorIndexAsync("emb", dimensions: 3, name: "idx_emb"); + var tx = await engine.BeginTransactionAsync(); + + await col.InsertAsync(VecDoc(col, 1, [1f, 2f, 3f]), tx); + await Assert.ThrowsAsync(async () => + await col.InsertAsync(VecDoc(col, 2, [1f, 2f]), tx)); + await tx.CommitAsync(); + + Assert.Null(await col.FindByIdAsync((BsonId)2)); + await AssertDynamicVectorItemsConsistent(engine, col, expectedCount: 1); + } + + [Fact] + public async Task Dynamic_UpsertAsync_Bad_Vector_Leaves_Every_Index_Consistent() + { + using var engine = new BLiteEngine(_dbPath); + var col = engine.GetOrCreateCollection("vecs", BsonIdType.Int32); + await col.CreateIndexAsync("tag", "idx_tag"); + await col.CreateVectorIndexAsync("emb", dimensions: 3, name: "idx_emb"); + var tx = await engine.BeginTransactionAsync(); + + await col.UpsertAsync(VecDoc(col, 1, [1f, 2f, 3f]), tx); + await Assert.ThrowsAsync(async () => + await col.UpsertAsync(VecDoc(col, 2, [1f, 2f]), tx)); + Assert.Equal(TransactionState.Active, tx.State); + + await tx.CommitAsync(); + + Assert.NotNull(await col.FindByIdAsync((BsonId)1)); + await AssertDynamicVectorItemsConsistent(engine, col, expectedCount: 1); + } + + [Fact] + public async Task Dynamic_UpdateAsync_Bad_Vector_Leaves_Every_Index_Consistent() + { + using var engine = new BLiteEngine(_dbPath); + var col = engine.GetOrCreateCollection("vecs", BsonIdType.Int32); + await col.CreateIndexAsync("tag", "idx_tag"); + await col.CreateVectorIndexAsync("emb", dimensions: 3, name: "idx_emb"); + await col.InsertAsync(VecDoc(col, 1, [1f, 2f, 3f], "x")); + await col.InsertAsync(VecDoc(col, 2, [4f, 5f, 6f], "x")); + + var tx = await engine.BeginTransactionAsync(); + Assert.True(await col.UpdateAsync((BsonId)1, VecDoc(col, 1, [7f, 8f, 9f], "updated"), tx)); + + await Assert.ThrowsAsync(async () => + await col.UpdateAsync((BsonId)2, VecDoc(col, 2, [1f, 2f], "z"), tx)); + Assert.Equal(TransactionState.Active, tx.State); + + await tx.CommitAsync(); + + Assert.Equal("updated", await TagOf(col, 1)); + Assert.Equal("x", await TagOf(col, 2)); + Assert.Empty(await col.QueryIndexAsync("idx_tag", "z", "z").ToListAsync()); + Assert.Equal([2], await IdsInIndex(col, "idx_tag", "x")); + await AssertDynamicVectorItemsConsistent(engine, col, expectedCount: 2); + } + + [Fact] + public async Task Dynamic_UpdateBulkAsync_Rejected_Later_Document_Rolls_Back_Transaction() + { + using var engine = new BLiteEngine(_dbPath); + var col = engine.GetOrCreateCollection("vecs", BsonIdType.Int32); + await col.CreateIndexAsync("tag", "idx_tag"); + await col.CreateVectorIndexAsync("emb", dimensions: 3, name: "idx_emb"); + await col.InsertAsync(VecDoc(col, 1, [1f, 2f, 3f], "x")); + await col.InsertAsync(VecDoc(col, 2, [4f, 5f, 6f], "x")); + + var tx = await engine.BeginTransactionAsync(); + await Assert.ThrowsAsync(async () => + await col.UpdateBulkAsync( + [((BsonId)1, VecDoc(col, 1, [7f, 8f, 9f], "y")), ((BsonId)2, VecDoc(col, 2, [1f, 2f], "z"))], tx)); + Assert.Equal(TransactionState.Aborted, tx.State); + + Assert.Equal("x", await TagOf(col, 1)); + Assert.Equal("x", await TagOf(col, 2)); + await AssertDynamicVectorItemsConsistent(engine, col, expectedCount: 2); + } + + // ── Dynamic collections: other failures roll the transaction back ──────── + + [Fact] + public async Task Dynamic_Failure_After_The_First_Page_Write_Rolls_Back_Callers_Transaction() + { + using var engine = new BLiteEngine(_dbPath); + var col = engine.GetOrCreateCollection("docs", BsonIdType.Int32); + await col.CreateIndexAsync("name", "idx_name"); + await col.InsertAsync(Doc(col, 0, "committed")); + + var tx = await engine.BeginTransactionAsync(); + var ex = await InsertUntilBTreeNodeOverflows(i => + col.InsertAsync(Doc(col, i, LongKey(i)), tx).AsTask()); + + Assert.IsNotType(ex); + Assert.Equal(TransactionState.Aborted, tx.State); + await Assert.ThrowsAsync(async () => + await col.InsertAsync(Doc(col, 10_000, "after"), tx)); + await Assert.ThrowsAsync(async () => await tx.CommitAsync()); + + Assert.Equal(1, await col.CountAsync()); + await AssertDynamicBTreeIndexConsistent(col, "idx_name", "name"); + Assert.Equal(1, CountLiveSlots(engine.Storage, col.Name)); + } + + // ── Helpers ────────────────────────────────────────────────────────────── + + /// + /// B-Tree nodes split by entry count, not by size, so a run of long keys overflows a leaf + /// page while the key is being inserted, after the document's data page and primary index + /// entry were written: a real failure in the middle of a write. + /// + private static async Task InsertUntilBTreeNodeOverflows(Func insert) + { + for (int i = 1; i <= 200; i++) + { + try { await insert(i); } + catch (Exception ex) { return ex; } + } + throw new Xunit.Sdk.XunitException("Expected a B-Tree node to overflow."); + } + + private static string LongKey(int i) => i.ToString("D5") + new string('k', 1000); + + private static async Task AssertTypedVectorItemsConsistent(TestDbContext db, int expectedCount) + { + var docs = await db.VectorItems.FindAllAsync().ToListAsync(); + Assert.Equal(expectedCount, docs.Count); + Assert.Equal(expectedCount, await db.VectorItems.CountAsync()); + Assert.Equal(expectedCount, CountLiveSlots(db.Storage, db.VectorItems.CollectionName)); + + var byTitle = await db.VectorItems.QueryIndexAsync("idx_title", null, null).ToListAsync(); + Assert.Equal(docs.Select(d => d.Id).OrderBy(i => i), byTitle.Select(d => d.Id).OrderBy(i => i)); + foreach (var doc in docs) + Assert.Contains(await db.VectorItems.QueryIndexAsync("idx_title", doc.Title, doc.Title).ToListAsync(), + d => d.Id == doc.Id); + + var byVector = await db.VectorItems.VectorSearchAsync("idx_vector", [0f, 0f, 0f], k: 100).ToListAsync(); + Assert.Equal(docs.Select(d => d.Id).OrderBy(i => i), byVector.Select(d => d.Id).OrderBy(i => i)); + } + + private static async Task AssertDynamicVectorItemsConsistent(BLiteEngine engine, DynamicCollection col, int expectedCount) + { + var docs = await col.FindAllAsync().ToListAsync(); + Assert.Equal(expectedCount, docs.Count); + Assert.Equal(expectedCount, await col.CountAsync()); + Assert.Equal(expectedCount, CountLiveSlots(engine.Storage, col.Name)); + + await AssertDynamicBTreeIndexConsistent(col, "idx_tag", "tag"); + + var byVector = await col.VectorSearchAsync("idx_emb", [0f, 0f, 0f], k: 100).ToListAsync(); + Assert.Equal(docs.Select(IdOf).OrderBy(i => i), byVector.Select(IdOf).OrderBy(i => i)); + } + + /// + /// The index holds exactly one entry per document, and looking up each document's value + /// finds that document. + /// + private static async Task AssertDynamicBTreeIndexConsistent(DynamicCollection col, string indexName, string field) + { + var docs = await col.FindAllAsync().ToListAsync(); + var indexed = await col.QueryIndexAsync(indexName, null, null).ToListAsync(); + Assert.Equal(docs.Select(IdOf).OrderBy(i => i), indexed.Select(IdOf).OrderBy(i => i)); + + foreach (var doc in docs) + { + Assert.True(doc.TryGetString(field, out var value)); + Assert.Contains(IdOf(doc), await IdsInIndex(col, indexName, value)); + } + } + + private static async Task> IdsInIndex(DynamicCollection col, string indexName, string value) => + (await col.QueryIndexAsync(indexName, value, value).ToListAsync()).Select(IdOf).ToList(); + + /// Live (not deleted) slots on the collection's data pages. + private static int CountLiveSlots(StorageEngine storage, string collectionName) + { + var buffer = new byte[storage.PageSize]; + int live = 0; + foreach (var pageId in storage.GetCollectionPageIds(collectionName)) + { + storage.ReadPage(pageId, null, buffer); + var header = SlottedPageHeader.ReadFrom(buffer); + if (header.PageType != PageType.Data) continue; + + for (int i = 0; i < header.SlotCount; i++) + { + var slot = SlotEntry.ReadFrom(buffer.AsSpan(SlottedPageHeader.Size + i * SlotEntry.Size)); + if ((slot.Flags & SlotFlags.Deleted) == 0) live++; + } + } + return live; + } + + private static int IdOf(BsonDocument doc) + { + Assert.True(doc.TryGetId(out var id)); + return id.AsInt32(); + } + + private static BsonDocument Doc(DynamicCollection col, int id, string name) => + col.CreateDocument(["_id", "name"], b => b.AddId((BsonId)id).AddString("name", name)); + + private static BsonDocument Person(DynamicCollection col, int id, string email, string tag) => + col.CreateDocument(["_id", "email", "tag"], b => b.AddId((BsonId)id).AddString("email", email).AddString("tag", tag)); + + private static BsonDocument VecDoc(DynamicCollection col, int id, float[] emb, string tag = "orig") => + col.CreateDocument(["_id", "emb", "tag"], b => b.AddId((BsonId)id).AddFloatArray("emb", emb).AddString("tag", tag)); + + private static async Task NameOf(DynamicCollection col, int id) + { + var doc = await col.FindByIdAsync((BsonId)id); + return doc != null && doc.TryGetString("name", out var name) ? name : null; + } + + private static async Task TagOf(DynamicCollection col, int id) + { + var doc = await col.FindByIdAsync((BsonId)id); + return doc != null && doc.TryGetString("tag", out var tag) ? tag : null; + } +} diff --git a/tests/BLite.Tests/StringIndexPrefixMatchTests.cs b/tests/BLite.Tests/StringIndexPrefixMatchTests.cs index 9b868107..05a23cd1 100644 --- a/tests/BLite.Tests/StringIndexPrefixMatchTests.cs +++ b/tests/BLite.Tests/StringIndexPrefixMatchTests.cs @@ -319,8 +319,8 @@ public async Task Unique_Index_Accepts_A_Prefix_Of_An_Existing_Value_And_Rejects await _db.PeopleV2.InsertAsync(new PersonV2 { Name = "p70" }); await _db.PeopleV2.InsertAsync(new PersonV2 { Name = "p7" }); - await Assert.ThrowsAsync(() => _db.PeopleV2.InsertAsync(new PersonV2 { Name = "p7" }).AsTask()); - await Assert.ThrowsAsync(() => _db.PeopleV2.InsertAsync(new PersonV2 { Name = "p70" }).AsTask()); + await Assert.ThrowsAsync(() => _db.PeopleV2.InsertAsync(new PersonV2 { Name = "p7" }).AsTask()); + await Assert.ThrowsAsync(() => _db.PeopleV2.InsertAsync(new PersonV2 { Name = "p70" }).AsTask()); Assert.Single(_db.PeopleV2.AsQueryable().Where(p => p.Name == "p7").ToList()); } diff --git a/tests/BLite.Tests/UniqueIndexTests.cs b/tests/BLite.Tests/UniqueIndexTests.cs index c468c367..26182cce 100644 --- a/tests/BLite.Tests/UniqueIndexTests.cs +++ b/tests/BLite.Tests/UniqueIndexTests.cs @@ -40,7 +40,7 @@ public async Task Typed_Unique_Index_Rejects_Duplicates() await _db.PeopleV2.CreateIndexAsync(p => p.Name, "idx_name", unique: true); await _db.PeopleV2.InsertAsync(new PersonV2 { Name = "dup", Age = 1 }); - await Assert.ThrowsAsync( + await Assert.ThrowsAsync( () => _db.PeopleV2.InsertAsync(new PersonV2 { Name = "dup", Age = 2 }).AsTask()); // The rejected insert leaves nothing behind. @@ -58,7 +58,7 @@ public async Task Dynamic_Unique_Index_Rejects_Duplicates() await col.CreateIndexAsync("name", unique: true); await col.InsertAsync(col.CreateDocument(["name"], b => b.AddString("name", "dup"))); - await Assert.ThrowsAsync( + await Assert.ThrowsAsync( () => col.InsertAsync(col.CreateDocument(["name"], b => b.AddString("name", "dup"))).AsTask()); } @@ -72,7 +72,7 @@ public async Task Dynamic_Unique_Index_Rejects_A_Second_Explicit_Null() await col.CreateIndexAsync("name", unique: true); await col.InsertAsync(col.CreateDocument(["name"], b => b.AddNull("name"))); - await Assert.ThrowsAsync( + await Assert.ThrowsAsync( () => col.InsertAsync(col.CreateDocument(["name"], b => b.AddNull("name"))).AsTask()); } @@ -86,7 +86,7 @@ public async Task Typed_Unique_Index_Rejects_Update_Onto_Existing_Value() var bob = (await _db.PeopleV2.FindByIdAsync(bobId))!; bob.Name = "alice"; - await Assert.ThrowsAsync( + await Assert.ThrowsAsync( () => _db.PeopleV2.UpdateAsync(bob).AsTask()); // The rejected update leaves the document and its index entry unchanged. @@ -137,7 +137,7 @@ public async Task Typed_Unique_Index_Allows_Updates_Of_Documents_That_Already_Sh // Changing the key onto a value another document holds is still rejected. second.Name = "carol"; - await Assert.ThrowsAsync( + await Assert.ThrowsAsync( () => _db.PeopleV2.UpdateAsync(second).AsTask()); Assert.Equal("dup", (await _db.PeopleV2.FindByIdAsync(secondId))!.Name); @@ -156,7 +156,7 @@ public async Task Typed_Unique_Index_Separates_Values_Followed_By_Low_Control_Ch foreach (var name in names) { - await Assert.ThrowsAsync( + await Assert.ThrowsAsync( () => _db.PeopleV2.InsertAsync(new PersonV2 { Name = name, Age = 2 }).AsTask()); } @@ -193,7 +193,7 @@ public async Task Typed_Unique_Index_With_String_Ids_Rejects_Duplicates_But_Not_ await _db.StringEntities.InsertAsync(new StringEntity { Id = "1", Value = "abc" }); await _db.StringEntities.InsertAsync(new StringEntity { Id = "2", Value = "ab" }); - await Assert.ThrowsAsync( + await Assert.ThrowsAsync( () => _db.StringEntities.InsertAsync(new StringEntity { Id = "3", Value = "ab" }).AsTask()); Assert.Equal(2, await _db.StringEntities.CountAsync()); @@ -214,7 +214,7 @@ public async Task Typed_Unique_Index_Rejects_Duplicates_Within_A_Bulk_Insert() { await _db.PeopleV2.CreateIndexAsync(p => p.Name, "idx_name", unique: true); - await Assert.ThrowsAsync(() => _db.PeopleV2.InsertBulkAsync( + await Assert.ThrowsAsync(() => _db.PeopleV2.InsertBulkAsync( [ new PersonV2 { Name = "x", Age = 1 }, new PersonV2 { Name = "dup", Age = 2 }, diff --git a/tests/BLite.Tests/VectorSearchTests.cs b/tests/BLite.Tests/VectorSearchTests.cs index 30cd92a0..6437c035 100644 --- a/tests/BLite.Tests/VectorSearchTests.cs +++ b/tests/BLite.Tests/VectorSearchTests.cs @@ -326,8 +326,8 @@ public async Task Test_VectorSearch_DimensionMismatch_Throws() using (var db = new TestDbContext(dbPath)) { - // The index expects dim=3; passing a 5-element vector must throw - await Assert.ThrowsAsync(async () => + // The index expects dim=3; a 5-element vector is rejected before anything is written + await Assert.ThrowsAsync(async () => await db.VectorItems.InsertAsync(new VectorEntity { Title = "Bad", Embedding = [1.0f, 2.0f, 3.0f, 4.0f, 5.0f] })); }