diff --git a/Changelog.md b/Changelog.md index 9f0e3cd..c7c8381 100644 --- a/Changelog.md +++ b/Changelog.md @@ -2,6 +2,8 @@ ## Unreleased +- Guard recycle-bin checkpoint reads and deletion together in synchronous and asynchronous cleanup. Disposed engines fail before collection access, failed reads never reach deletion, and acquired cursors are disposed. Preserve async cancellation, query shape and the strict checkpoint cutoff. No public API, stored-schema or MongoDB minimum-version change; no data migration required. [#107](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/107) + - Translate MongoDB driver failures from all nine lazy synchronous reads at cursor acquisition and subsequent batches, preserving streaming, cursor disposal, and serializer exception identity. Async reads, public signatures, stored data, CI, and the MongoDB 4.2 minimum are unchanged. [#91](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/91) - Make exact-ID explain assertions accept MongoDB IDHACK as well as named index plans, while verifying the built-in index, bounded examined work and absence of collection scans. Add plan-layout regression coverage and a forced-scan negative control. Test-only change; no API, schema or MongoDB minimum-version change and no migration required. [#94](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/94) @@ -24,6 +26,8 @@ ### Breaking Changes +- `EmptyRecycleBin()` and `EmptyRecycleBinAsync(...)` now translate checkpoint-read driver failures to `StorageException` or `StorageUnavailableException` (for `MongoConnectionException`), retaining the original inner exception. Consumers catching raw driver exceptions must handle the persistence exceptions instead. The existing protected virtual guard is still invoked once, but its callback now includes checkpoint acquisition and iteration before deletion. Non-driver exceptions and cancellation retain their identity. [#107](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/107) + - Synchronous reads reuse one cursor `MoveNext` callback per enumeration. Both protected virtual `TryMongo` overloads are still called for cursor acquisition and every batch attempt, including completion. Source and binary APIs are unchanged, but overrides that inspect callback reference identity now observe the same callback within an enumeration. [#91](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/91) - Synchronous enumeration now reports driver failures as `StorageException` or `StorageUnavailableException` (for `MongoConnectionException`), rather than leaking raw driver exceptions. Consumers that caught driver exceptions from iteration must handle the persistence exceptions instead. Keep the persistence engine alive until enumeration finishes; deferred driver calls now enforce its existing disposed-engine guard. Serializer errors remain unchanged. [#91](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/91) diff --git a/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue107CleanupCheckpointGuards.cs b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue107CleanupCheckpointGuards.cs new file mode 100644 index 0000000..70446d8 --- /dev/null +++ b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue107CleanupCheckpointGuards.cs @@ -0,0 +1,230 @@ +#if NUNIT +using System.Net; +using FakeItEasy; +using FluentAssertions; +using MongoDB.Bson; +using MongoDB.Bson.Serialization; +using MongoDB.Driver; +using MongoDB.Driver.Core.Clusters; +using MongoDB.Driver.Core.Connections; +using MongoDB.Driver.Core.Servers; +using NEventStore.Persistence.AcceptanceTests; +using NEventStore.Persistence.AcceptanceTests.BDD; +using NEventStore.Serialization; +using NUnit.Framework; + +namespace NEventStore.Persistence.MongoDB.Tests.AcceptanceTests.Issues +{ + [TestFixtureSource(nameof(Cases))] + public class when_Issue107_cleanup_reads_its_checkpoint(bool async, string stage, string outcome) : SpecificationBase + { + public static IEnumerable Cases + { + get + { + foreach (var async in new[] { false, true }) + { + foreach (var stage in new[] { "find", "cursor" }) + foreach (var failure in new[] { "mongo", "connection", "other" }) + yield return new object[] { async, stage, failure }; + foreach (var outcome in new[] { "disposed", "success", "empty" }) + yield return new object[] { async, "none", outcome }; + } + foreach (var stage in new[] { "find", "cursor" }) + foreach (var outcome in new[] { "canceled", "active cancellation" }) + yield return new object[] { true, stage, outcome }; + } + } + + private readonly IMongoCollection _collection = A.Fake>(); + private readonly IAsyncCursor _cursor = A.Fake>(); + private readonly CancellationTokenSource _cancellation = new(); + private CleanupEngine _engine = null!; + private Exception? _failure; + private Exception? _error; + private Task _operation = null!; + private FilterDefinition? _deleteFilter; + private FindOptions? _findOptions; + private int _findCalls; + private int _deleteCalls; + + protected override void Context() + { + _failure = outcome switch + { + "mongo" => new MongoException("checkpoint failed"), + "connection" => new MongoConnectionException( + new ConnectionId(new ServerId(new ClusterId(), new DnsEndPoint("localhost", 27017))), "connection failed"), + "other" => new InvalidOperationException("checkpoint failed"), + "canceled" or "active cancellation" => new OperationCanceledException(_cancellation.Token), + _ => null + }; + A.CallTo(() => _collection.FindSync(A>._, + A>._, A._)) + .ReturnsLazily((FilterDefinition filter, FindOptions options, CancellationToken token) => + { + async.Should().BeFalse("async cleanup must never perform synchronous checkpoint I/O"); + return Find(options); + }); + A.CallTo(() => _collection.FindAsync(A>._, + A>._, _cancellation.Token)) + .ReturnsLazily((FilterDefinition filter, FindOptions options, CancellationToken token) => + { + try { return Task.FromResult(Find(options)); } + catch (Exception error) { return Task.FromException>(error); } + }); + A.CallTo(() => _cursor.Current).Returns(outcome == "empty" ? Array.Empty() : + new[] { new BsonDocument(MongoCommitFields.CheckpointNumber, 42L) }); + A.CallTo(() => _cursor.MoveNext(A._)).ReturnsLazily(MoveNext); + A.CallTo(() => _cursor.MoveNextAsync(_cancellation.Token)).ReturnsLazily(() => + stage == "cursor" ? Task.FromException(_failure!) : Task.FromResult(outcome != "empty")); + A.CallTo(() => _collection.DeleteMany(A>._, A._)) + .ReturnsLazily((FilterDefinition filter, CancellationToken token) => + { + async.Should().BeFalse("async cleanup must never perform synchronous deletion"); + return Delete(filter); + }); + A.CallTo(() => _collection.DeleteManyAsync(A>._, _cancellation.Token)) + .ReturnsLazily((FilterDefinition filter, CancellationToken token) => Task.FromResult(Delete(filter))); + var database = A.Fake(); + A.CallTo(() => database.GetCollection(A._, A._)).Returns(_collection); + A.CallTo(() => _collection.WithWriteConcern(A._)).Returns(_collection); + _engine = new CleanupEngine(database); + if (outcome == "disposed") _engine.Dispose(); + if (outcome == "canceled") _cancellation.Cancel(); + } + + private IAsyncCursor Find(FindOptions options) + { + _findCalls++; + _findOptions = options; + if (stage == "find") throw _failure!; + return _cursor; + } + + private bool MoveNext() + { + if (stage == "cursor") throw _failure!; + return outcome != "empty"; + } + + private DeleteResult Delete(FilterDefinition filter) + { + _deleteCalls++; + _deleteFilter = filter; + return new DeleteResult.Acknowledged(0); + } + + protected override async Task BecauseAsync() + { + if (async) + { + _operation = _engine.EmptyRecycleBinAsync(_cancellation.Token); + _error = await Catch.ExceptionAsync(() => _operation).ConfigureAwait(false); + } + else _error = Catch.Exception(_engine.EmptyRecycleBin); + } + + [Fact] + public void Should_preserve_the_storage_exception_and_cancellation_contract() + { + if (outcome == "disposed") _error.Should().BeOfType(); + else if (outcome == "mongo" || outcome == "connection") + { + _error.Should().BeOfType(outcome == "connection" ? typeof(StorageUnavailableException) : typeof(StorageException)); + _error!.InnerException.Should().BeSameAs(_failure, "driver failures must be translated exactly once"); + } + else _error.Should().BeSameAs(_failure); + if (outcome == "canceled" || outcome == "active cancellation") + { + _operation.IsCanceled.Should().BeTrue(); + ((OperationCanceledException)_error!).CancellationToken.Should().Be(_cancellation.Token); + } + } + + [Fact] + public void Should_guard_the_entire_operation_once_before_any_collection_access() + { + _engine.GuardCalls.Should().Be(1, "retain the existing protected virtual guard hook without nesting"); + _engine.UnguardedAccesses.Should().Be(0); + if (outcome == "disposed") _engine.CollectionAccesses.Should().Be(0); + _findCalls.Should().Be(outcome == "disposed" ? 0 : 1); + } + + [Fact] + public void Should_never_delete_after_a_failed_read_and_preserve_the_cutoff_on_success() + { + var succeeds = outcome == "success" || outcome == "empty"; + _deleteCalls.Should().Be(succeeds ? 1 : 0); + if (!succeeds) return; + var registry = BsonSerializer.SerializerRegistry; + var args = new RenderArgs(registry.GetSerializer(), registry); + _findOptions!.Limit.Should().Be(1); + _findOptions.Sort.Render(args).Equals(new BsonDocument(MongoCommitFields.CheckpointNumber, -1)).Should().BeTrue(); + _findOptions.Projection.Render(args).Document[MongoCommitFields.CheckpointNumber].AsInt32.Should().Be(1); + _deleteFilter!.Render(args).Equals(new BsonDocument + { + { MongoCommitFields.BucketId, MongoSystemBuckets.RecycleBin }, + { MongoCommitFields.CheckpointNumber, new BsonDocument("$lt", outcome == "empty" ? 0L : 42L) } + }).Should().BeTrue(); + } + + [Fact] + public void Should_dispose_every_acquired_cursor_and_forward_the_request_token() + { + if (stage == "find" || outcome == "disposed") + A.CallTo(() => _cursor.Dispose()).MustNotHaveHappened(); + else + { + A.CallTo(() => _cursor.Dispose()).MustHaveHappenedOnceExactly(); + if (async) A.CallTo(() => _cursor.MoveNextAsync(_cancellation.Token)).MustHaveHappenedOnceExactly(); + else A.CallTo(() => _cursor.MoveNext(A._)).MustHaveHappenedOnceExactly(); + } + if (async && outcome != "disposed") + A.CallTo(() => _collection.FindAsync(A>._, + A>._, _cancellation.Token)).MustHaveHappenedOnceExactly(); + } + + protected override void Cleanup() + { + _engine?.Dispose(); + _cancellation.Dispose(); + } + + private sealed class CleanupEngine(IMongoDatabase database) + : MongoPersistenceEngine(database, new DocumentObjectSerializer(), new MongoPersistenceOptions()) + { + public int CollectionAccesses { get; private set; } + public int UnguardedAccesses { get; private set; } + public int GuardCalls { get; private set; } + private bool _guarded; + + protected override IMongoCollection PersistedCommits + { + get + { + CollectionAccesses++; + if (!_guarded) UnguardedAccesses++; + return base.PersistedCommits; + } + } + + protected override void TryMongo(Action callback) + { + GuardCalls++; + _guarded = true; + try { base.TryMongo(callback); } + finally { _guarded = false; } + } + + protected override async Task TryMongoAsync(Func callbackAsync, CancellationToken cancellationToken, IAsyncObserver? asyncObserver = null) + { + GuardCalls++; + _guarded = true; + try { await base.TryMongoAsync(callbackAsync, cancellationToken, asyncObserver).ConfigureAwait(false); } + finally { _guarded = false; } + } + } + } +} +#endif diff --git a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs index 00fb201..ceb1b27 100644 --- a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs +++ b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs @@ -665,13 +665,13 @@ private async Task GetLastCommittedCheckPointNumberAsync(CancellationToke /// public async Task EmptyRecycleBinAsync(CancellationToken cancellationToken) { - var lastCheckpointNumber = await GetLastCommittedCheckPointNumberAsync(cancellationToken).ConfigureAwait(false); - await TryMongoAsync(() => + await TryMongoAsync(async () => { - return PersistedCommits.DeleteManyAsync(Builders.Filter.And( + var lastCheckpointNumber = await GetLastCommittedCheckPointNumberAsync(cancellationToken).ConfigureAwait(false); + await PersistedCommits.DeleteManyAsync(Builders.Filter.And( Builders.Filter.Eq(MongoCommitFields.BucketId, MongoSystemBuckets.RecycleBin), Builders.Filter.Lt(MongoCommitFields.CheckpointNumber, lastCheckpointNumber) - ), cancellationToken); + ), cancellationToken).ConfigureAwait(false); }, cancellationToken, asyncObserver: null); diff --git a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs index 0933ff8..1a4f81c 100644 --- a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs +++ b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs @@ -877,9 +877,9 @@ private Int64 GetLastCommittedCheckPointNumber() /// public void EmptyRecycleBin() { - var lastCheckpointNumber = GetLastCommittedCheckPointNumber(); TryMongo(() => { + var lastCheckpointNumber = GetLastCommittedCheckPointNumber(); PersistedCommits.DeleteMany(Builders.Filter.And( Builders.Filter.Eq(MongoCommitFields.BucketId, MongoSystemBuckets.RecycleBin), Builders.Filter.Lt(MongoCommitFields.CheckpointNumber, lastCheckpointNumber)