From 267f8cb32406379d90de848f606f9dcd61328a20 Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Fri, 6 Nov 2026 12:55:46 +0100 Subject: [PATCH 1/3] test: cover deferred MongoDB read failures (#91) Co-authored-by: Alessandro Giorgetti Co-authored-by: Codex (OpenAI) --- .../DeferredEnumerationTests.cs | 326 ++++++++++++++++++ 1 file changed, 326 insertions(+) create mode 100644 src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/DeferredEnumerationTests.cs diff --git a/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/DeferredEnumerationTests.cs b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/DeferredEnumerationTests.cs new file mode 100644 index 0000000..8efd940 --- /dev/null +++ b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/DeferredEnumerationTests.cs @@ -0,0 +1,326 @@ +#if NUNIT +using System.Net; +using FakeItEasy; +using FluentAssertions; +using MongoDB.Bson; +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 +{ + [TestFixtureSource(nameof(Cases))] + public class when_a_synchronous_read_fails : SpecificationBase + { + public static IEnumerable Cases => + from path in Enumerable.Range(0, EnumerationProbe.ReadCount) + from stage in new[] { "immediate", "find", "first batch", "later batch" } + from kind in new[] { "mongo", "connection", "other" } + select new object[] { path, stage, kind }; + + private readonly int _path; + private readonly string _stage; + private readonly string _kind; + private EnumerationProbe _probe = null!; + private Exception _failure = null!; + private Exception? _creationError; + private Exception? _enumerationError; + private int _findCallsBeforeEnumeration; + private int _itemsRead; + + public when_a_synchronous_read_fails(int path, string stage, string kind) + { + _path = path; + _stage = stage; + _kind = kind; + } + + protected override void Context() + { + _failure = EnumerationProbe.Failure(_kind); + _probe = new EnumerationProbe(_path, _stage, _failure); + } + + protected override void Because() + { + IEnumerable? result = null; + _creationError = Catch.Exception(() => result = _probe.Read()); + _findCallsBeforeEnumeration = _probe.FindCalls; + if (result != null) + _enumerationError = Catch.Exception(() => + { + foreach (var item in result) + _itemsRead++; + }); + } + + [Fact] + public void Should_translate_only_driver_failures_at_the_original_boundary() + { + var error = _stage == "immediate" ? _creationError : _enumerationError; + if (_kind == "other") + error.Should().BeSameAs(_failure); + else + { + error.Should().BeOfType(_kind == "connection" + ? typeof(StorageUnavailableException) : typeof(StorageException)); + error!.InnerException.Should().BeSameAs(_failure); + } + if (_stage != "immediate") + _creationError.Should().BeNull(); + } + + [Fact] + public void Should_remain_lazy_and_stream_prior_batches_before_a_later_failure() + { + _findCallsBeforeEnumeration.Should().Be(0); + _itemsRead.Should().Be(_stage == "later batch" ? 1 : 0); + } + + [Fact] + public void Should_dispose_every_acquired_cursor_on_failure() + { + _probe.Cursors.Count.Should().Be(_stage == "first batch" || _stage == "later batch" ? 1 : 0); + foreach (var cursor in _probe.Cursors) + A.CallTo(() => cursor.Dispose()).MustHaveHappenedOnceExactly(); + } + + protected override void Cleanup() => _probe.Engine.Dispose(); + } + + [TestFixtureSource(nameof(Cases))] + public class when_a_synchronous_read_is_enumerated : SpecificationBase + { + public static IEnumerable Cases => + from path in Enumerable.Range(0, EnumerationProbe.ReadCount) + from mode in new[] { "complete", "early", "consumer failure", "repeat" } + select new object[] { path, mode }; + + private readonly int _path; + private readonly string _mode; + private EnumerationProbe _probe = null!; + private int _itemsRead; + private int _findCallsBeforeEnumeration; + private Exception? _error; + private readonly Exception _consumerFailure = new InvalidOperationException("consumer"); + + public when_a_synchronous_read_is_enumerated(int path, string mode) + { + _path = path; + _mode = mode; + } + + protected override void Context() => _probe = new EnumerationProbe(_path); + + protected override void Because() + { + var result = _probe.Read(); + _findCallsBeforeEnumeration = _probe.FindCalls; + _error = Catch.Exception(() => + { + foreach (var item in result) + { + _itemsRead++; + if (_mode == "early") + break; + if (_mode == "consumer failure") + throw _consumerFailure; + } + if (_mode == "repeat") + _itemsRead += result.Count(); + }); + } + + [Fact] + public void Should_keep_streaming_and_allow_repeated_enumeration() + { + _findCallsBeforeEnumeration.Should().Be(0); + _itemsRead.Should().Be(_mode == "repeat" ? 4 : _mode == "complete" ? 2 : 1); + _probe.FindCalls.Should().Be(_mode == "repeat" ? 2 : 1); + // Early termination must not fetch a second batch, even if it is empty. + _probe.MoveNextCalls.Should().Be(_mode == "repeat" ? 8 : _mode == "complete" ? 4 : 1); + _error.Should().BeSameAs(_mode == "consumer failure" ? _consumerFailure : null); + } + + [Fact] + public void Should_dispose_cursors_on_completion_and_early_termination() + { + foreach (var cursor in _probe.Cursors) + A.CallTo(() => cursor.Dispose()).MustHaveHappenedOnceExactly(); + } + + protected override void Cleanup() => _probe.Engine.Dispose(); + } + + [TestFixtureSource(nameof(Cases))] + public class when_a_synchronous_commit_serializer_fails : SpecificationBase + { + public static IEnumerable Cases => + from path in Enumerable.Range(0, EnumerationProbe.ReadCount).Where(path => path != 7) + from kind in new[] { "mongo", "connection", "other" } + select new object[] { path, kind }; + + private readonly int _path; + private readonly string _kind; + private EnumerationProbe _probe = null!; + private Exception _failure = null!; + private Exception? _error; + + public when_a_synchronous_commit_serializer_fails(int path, string kind) + { + _path = path; + _kind = kind; + } + + protected override void Context() + { + _failure = EnumerationProbe.Failure(_kind); + _probe = new EnumerationProbe(_path, "serializer", _failure); + } + + protected override void Because() => _error = Catch.Exception(() => _probe.Read().ToArray()); + + [Fact] + public void Should_preserve_serializer_errors_even_when_they_use_driver_exception_types() + { + _error.Should().BeSameAs(_failure); + _probe.FindCalls.Should().Be(1); + _probe.MoveNextCalls.Should().Be(1); + A.CallTo(() => _probe.Cursors.Single().Dispose()).MustHaveHappenedOnceExactly(); + } + + protected override void Cleanup() => _probe.Engine.Dispose(); + } + + internal sealed class EnumerationProbe + { + public const int ReadCount = 9; + private readonly int _path; + public MongoPersistenceEngine Engine { get; } + public int FindCalls { get; private set; } + public int MoveNextCalls { get; private set; } + public List> Cursors { get; } = []; + + public EnumerationProbe(int path, string? failureStage = null, Exception? failure = null) + { + _path = path; + var database = A.Fake(); + var collection = A.Fake>(); + var serializer = A.Fake(); + A.CallTo(() => collection.Settings).Returns(new MongoCollectionSettings()); + A.CallTo(() => collection.WithWriteConcern(A._)).Returns(collection); + A.CallTo(() => database.GetCollection(A._, A._)) + .Returns(collection); + A.CallTo(() => serializer.Deserialize(A._)).Throws(failure ?? new Exception("unexpected serializer call")); + A.CallTo(() => collection.FindSync( + A>._, + A>._, + A._)).ReturnsLazily(() => + { + FindCalls++; + if (failureStage == "find") + throw failure!; + var cursor = A.Fake>(); + Cursors.Add(cursor); + int batch = 0; + A.CallTo(() => cursor.MoveNext(A._)).ReturnsLazily(() => + { + MoveNextCalls++; + batch++; + if (failureStage == "first batch" || (failureStage == "later batch" && batch == 2)) + throw failure!; + return batch <= 3; + }); + A.CallTo(() => cursor.Current).ReturnsLazily(() => batch == 2 + ? Array.Empty() + : new[] { Document(path, batch, failureStage == "serializer") }); + return cursor; + }); + Engine = new ProbeEngine(database, serializer, collection, failureStage == "immediate" ? failure : null); + } + + public IEnumerable Read() + { +#pragma warning disable CS0618 // Date-based public reads are still supported. + return _path switch + { + 0 => Engine.GetFrom("bucket", "stream", 0, int.MaxValue), + 1 => Engine.GetFrom("bucket", DateTime.MinValue), + 2 => Engine.GetFrom("bucket", 0L), + 3 => Engine.GetFromTo("bucket", 0L, long.MaxValue), + 4 => Engine.GetFrom(0L), + 5 => Engine.GetFromTo(0L, long.MaxValue), + 6 => Engine.GetFromTo("bucket", DateTime.MinValue, DateTime.MaxValue), + 7 => Engine.GetStreamsToSnapshot("bucket", 0), + 8 => Engine.GetDeletedCommits(), + _ => throw new ArgumentOutOfRangeException(nameof(_path)) + }; +#pragma warning restore CS0618 + } + + public static Exception Failure(string kind) => kind switch + { + "connection" => new MongoConnectionException( + new ConnectionId(new ServerId(new ClusterId(), new DnsEndPoint("localhost", 27017))), "connection failed"), + "mongo" => new MongoException("driver failed"), + _ => new FormatException("not a driver failure") + }; + + private static BsonDocument Document(int path, int batch, bool serializedPayload) + { + if (path == 7) + return new BsonDocument + { + { MongoStreamHeadFields.Id, new BsonDocument + { + { MongoStreamHeadFields.BucketId, "bucket" }, + { MongoStreamHeadFields.StreamId, "stream" } + } + }, + { MongoStreamHeadFields.HeadRevision, batch }, + { MongoStreamHeadFields.SnapshotRevision, 0 } + }; + return new BsonDocument + { + { MongoCommitFields.CheckpointNumber, (long)batch }, + { MongoCommitFields.BucketId, "bucket" }, + { MongoCommitFields.StreamId, "stream" }, + { MongoCommitFields.StreamRevisionFrom, batch }, + { MongoCommitFields.StreamRevisionTo, batch }, + { MongoCommitFields.CommitSequence, batch }, + { MongoCommitFields.CommitStamp, new BsonDateTime(DateTime.UtcNow) }, + { MongoCommitFields.Headers, new BsonArray() }, + { MongoCommitFields.Events, serializedPayload + ? new BsonArray { new BsonDocument(MongoCommitFields.Payload, new BsonBinaryData(new byte[] { 1 })) } + : new BsonArray() + } + }; + } + + private sealed class ProbeEngine : MongoPersistenceEngine + { + private readonly IMongoCollection _collection; + private readonly Exception? _failure; + + public ProbeEngine(IMongoDatabase database, IDocumentSerializer serializer, + IMongoCollection collection, Exception? failure) + : base(database, serializer, new MongoPersistenceOptions()) + { + _collection = collection; + _failure = failure; + } + + protected override IMongoCollection PersistedCommits => + _failure == null ? _collection : throw _failure; + + protected override IMongoCollection PersistedStreamHeads => + _failure == null ? _collection : throw _failure; + } + } +} +#endif From 6e896aba8ba894011dbdcbb9b71b64bc22360af5 Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Fri, 6 Nov 2026 12:55:50 +0100 Subject: [PATCH 2/3] fix: translate lazy MongoDB cursor failures (#91) Guard cursor acquisition and batches while retaining streaming and serializer error identity. Cover disposal guards and document exception compatibility. Co-authored-by: Alessandro Giorgetti Co-authored-by: Codex (OpenAI) --- Changelog.md | 4 ++ .../DeferredEnumerationTests.cs | 53 +++++++++++++++++ .../MongoPersistenceEngine.cs | 57 ++++++++++--------- 3 files changed, 87 insertions(+), 27 deletions(-) diff --git a/Changelog.md b/Changelog.md index 1091598..7d140ae 100644 --- a/Changelog.md +++ b/Changelog.md @@ -2,6 +2,8 @@ ## Unreleased +- 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) - Atomically advance snapshot metadata without losing concurrent or delayed commit increments. Older snapshots cannot regress the recorded snapshot revision. Preserve configured write concerns; legacy heads without an unsnapshotted counter use the current head-minus-snapshot value as a fallback. Existing incorrect counters are not repaired. [#90](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/90) @@ -22,6 +24,8 @@ ### Breaking Changes +- 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) + - **MongoDB 4.2 or later is now required** for aggregation-based snapshot metadata updates. Upgrade MongoDB before upgrading this adapter. No stored-schema migration is needed; MongoDB 4.2 is the feature minimum, not a recommended production release. This change does not repair historical metadata or solve stream lifecycle races. [#90](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/90) - New recycled records use the synthetic stored `StreamId`. Older adapter binaries expose that value and cannot find these records by the original stream ID; raw readers that assume stored `StreamId` is original see the same change. Upgrade recycle-bin readers before deleting streams with the new format. Existing recycled records need no backfill. [#87](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/87) diff --git a/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/DeferredEnumerationTests.cs b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/DeferredEnumerationTests.cs index 8efd940..07232c4 100644 --- a/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/DeferredEnumerationTests.cs +++ b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/DeferredEnumerationTests.cs @@ -105,6 +105,7 @@ from path in Enumerable.Range(0, EnumerationProbe.ReadCount) private readonly string _mode; private EnumerationProbe _probe = null!; private int _itemsRead; + private readonly List _positions = []; private int _findCallsBeforeEnumeration; private Exception? _error; private readonly Exception _consumerFailure = new InvalidOperationException("consumer"); @@ -126,6 +127,7 @@ protected override void Because() foreach (var item in result) { _itemsRead++; + _positions.Add(item is ICommit commit ? commit.CheckpointToken : ((IStreamHead)item).HeadRevision); if (_mode == "early") break; if (_mode == "consumer failure") @@ -144,6 +146,7 @@ public void Should_keep_streaming_and_allow_repeated_enumeration() _probe.FindCalls.Should().Be(_mode == "repeat" ? 2 : 1); // Early termination must not fetch a second batch, even if it is empty. _probe.MoveNextCalls.Should().Be(_mode == "repeat" ? 8 : _mode == "complete" ? 4 : 1); + _positions.Should().Equal(_mode == "early" || _mode == "consumer failure" ? new long[] { 1 } : new long[] { 1, 3 }); _error.Should().BeSameAs(_mode == "consumer failure" ? _consumerFailure : null); } @@ -197,6 +200,56 @@ public void Should_preserve_serializer_errors_even_when_they_use_driver_exceptio protected override void Cleanup() => _probe.Engine.Dispose(); } + [TestFixtureSource(nameof(Cases))] + public class when_the_engine_is_disposed_during_a_synchronous_read : SpecificationBase + { + public static IEnumerable Cases => + from path in Enumerable.Range(0, EnumerationProbe.ReadCount) + from stage in new[] { "before call", "before iteration", "between batches" } + select new object[] { path, stage }; + + private readonly int _path; + private readonly string _stage; + private EnumerationProbe _probe = null!; + private Exception? _error; + + public when_the_engine_is_disposed_during_a_synchronous_read(int path, string stage) + { + _path = path; + _stage = stage; + } + + protected override void Context() => _probe = new EnumerationProbe(_path); + + protected override void Because() + { + _error = Catch.Exception(() => + { + if (_stage == "before call") + _probe.Engine.Dispose(); + var result = _probe.Read(); + using var iterator = result.GetEnumerator(); + if (_stage == "between batches") + iterator.MoveNext(); + _probe.Engine.Dispose(); + iterator.MoveNext(); + }); + } + + [Fact] + public void Should_guard_driver_calls_and_still_release_acquired_cursors() + { + _error.Should().BeOfType(); + _probe.FindCalls.Should().Be(_stage == "between batches" ? 1 : 0); + _probe.MoveNextCalls.Should().Be(_stage == "between batches" ? 1 : 0); + _probe.Cursors.Count.Should().Be(_stage == "between batches" ? 1 : 0); + foreach (var cursor in _probe.Cursors) + A.CallTo(() => cursor.Dispose()).MustHaveHappenedOnceExactly(); + } + + protected override void Cleanup() => _probe.Engine.Dispose(); + } + internal sealed class EnumerationProbe { public const int ReadCount = 9; diff --git a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs index d43e777..170cd6c 100644 --- a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs +++ b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs @@ -290,10 +290,9 @@ public virtual IEnumerable GetFrom(string bucketId, string streamId, in var query = Builders.Filter.And(filters); - return PersistedCommits + return EnumerateMongo(PersistedCommits .Find(query) - .Sort(SortByAscendingStreamRevisionFrom) - .ToEnumerable() + .Sort(SortByAscendingStreamRevisionFrom)) .Select(mc => mc.ToCommit(_serializer)); }); } @@ -315,10 +314,9 @@ public virtual IEnumerable GetFrom(string bucketId, DateTime startDate) ); } - return PersistedCommits + return EnumerateMongo(PersistedCommits .Find(query) - .Sort(SortByAscendingCheckpointNumber) - .ToEnumerable() + .Sort(SortByAscendingCheckpointNumber)) .Select(x => x.ToCommit(_serializer)); }); } @@ -339,10 +337,9 @@ public virtual IEnumerable GetFrom(string bucketId, Int64 checkpointTok ); } - return PersistedCommits + return EnumerateMongo(PersistedCommits .Find(query) - .Sort(SortByAscendingCheckpointNumber) - .ToEnumerable() + .Sort(SortByAscendingCheckpointNumber)) .Select(x => x.ToCommit(_serializer)); }); } @@ -373,10 +370,9 @@ public virtual IEnumerable GetFromTo(string bucketId, long fromCheckpoi var query = Builders.Filter.And(filters); - return PersistedCommits + return EnumerateMongo(PersistedCommits .Find(query) - .Sort(SortByAscendingCheckpointNumber) - .ToEnumerable() + .Sort(SortByAscendingCheckpointNumber)) .Select(x => x.ToCommit(_serializer)); }); } @@ -397,10 +393,9 @@ public virtual IEnumerable GetFrom(Int64 checkpointToken) ); } - return PersistedCommits + return EnumerateMongo(PersistedCommits .Find(query) - .Sort(SortByAscendingCheckpointNumber) - .ToEnumerable() + .Sort(SortByAscendingCheckpointNumber)) .Select(x => x.ToCommit(_serializer)); }); } @@ -431,10 +426,9 @@ public virtual IEnumerable GetFromTo(long fromCheckpointToken, long toC var query = Builders.Filter.And(filters); - return PersistedCommits + return EnumerateMongo(PersistedCommits .Find(query) - .Sort(SortByAscendingCheckpointNumber) - .ToEnumerable() + .Sort(SortByAscendingCheckpointNumber)) .Select(x => x.ToCommit(_serializer)); }); } @@ -466,10 +460,9 @@ public virtual IEnumerable GetFromTo(string bucketId, DateTime startDat var query = Builders.Filter.And(filters); - return PersistedCommits + return EnumerateMongo(PersistedCommits .Find(query) - .Sort(SortByAscendingCheckpointNumber) - .ToEnumerable() + .Sort(SortByAscendingCheckpointNumber)) .Select(x => x.ToCommit(_serializer)); }); } @@ -593,10 +586,9 @@ public virtual IEnumerable GetStreamsToSnapshot(string bucketId, in Builders.Filter.Eq(MongoStreamHeadFields.FullQualifiedBucketId, bucketId), Builders.Filter.Gte(MongoStreamHeadFields.Unsnapshotted, maxThreshold) ); - return PersistedStreamHeads + return EnumerateMongo(PersistedStreamHeads .Find(query) - .Sort(Builders.Sort.Descending(MongoStreamHeadFields.Unsnapshotted)) - .ToEnumerable() + .Sort(Builders.Sort.Descending(MongoStreamHeadFields.Unsnapshotted))) .Select(x => x.ToStreamHead()); }); return result ?? []; @@ -801,6 +793,18 @@ private void UpdateStreamHeadInBackgroundThread(string bucketId, string streamId }); } + private IEnumerable EnumerateMongo(IAsyncCursorSource source) + { + // Keep only driver work inside TryMongo. Callers project documents outside + // this iterator so serializer errors retain their original exception types. + using var cursor = TryMongo(() => source.ToCursor()); + while (TryMongo(() => cursor.MoveNext())) + { + foreach (var document in cursor.Current) + yield return document; + } + } + /// /// Executes the callback within a try/catch block to handle exceptions. /// @@ -887,10 +891,9 @@ public void EmptyRecycleBin() /// public IEnumerable GetDeletedCommits() { - return TryMongo(() => PersistedCommits + return TryMongo(() => EnumerateMongo(PersistedCommits .Find(Builders.Filter.Eq(MongoCommitFields.BucketId, MongoSystemBuckets.RecycleBin)) - .Sort(SortByAscendingCheckpointNumber) - .ToEnumerable() + .Sort(SortByAscendingCheckpointNumber)) .Select(mc => mc.ToCommit(_serializer))); } From 240f563bf52b6c723552e7c26fdd1d42aafc7604 Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Fri, 6 Nov 2026 12:55:54 +0100 Subject: [PATCH 3/3] perf: reuse synchronous cursor callback per enumeration (#91) Hoist the MoveNext delegate while retaining both virtual TryMongo dispatches. Cover dispatch order and callback identity across all nine read paths, and document the observable identity change for derived engines. Co-authored-by: Alessandro Giorgetti Co-authored-by: Codex (OpenAI) --- Changelog.md | 2 + .../DeferredEnumerationTests.cs | 74 +++++++++++++++++++ .../MongoPersistenceEngine.cs | 3 +- 3 files changed, 78 insertions(+), 1 deletion(-) diff --git a/Changelog.md b/Changelog.md index 7d140ae..9f0e3cd 100644 --- a/Changelog.md +++ b/Changelog.md @@ -24,6 +24,8 @@ ### Breaking Changes +- 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) - **MongoDB 4.2 or later is now required** for aggregation-based snapshot metadata updates. Upgrade MongoDB before upgrading this adapter. No stored-schema migration is needed; MongoDB 4.2 is the feature minimum, not a recommended production release. This change does not repair historical metadata or solve stream lifecycle races. [#90](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/90) diff --git a/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/DeferredEnumerationTests.cs b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/DeferredEnumerationTests.cs index 07232c4..1cb5bee 100644 --- a/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/DeferredEnumerationTests.cs +++ b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/DeferredEnumerationTests.cs @@ -250,6 +250,62 @@ public void Should_guard_driver_calls_and_still_release_acquired_cursors() protected override void Cleanup() => _probe.Engine.Dispose(); } + [TestFixtureSource(nameof(Cases))] + public class when_synchronous_reads_use_virtual_guards : SpecificationBase + { + public static IEnumerable Cases => + Enumerable.Range(0, EnumerationProbe.ReadCount).Select(path => new object[] { path }); + + private readonly int _path; + private EnumerationProbe _probe = null!; + private int _callsBeforeEnumeration; + + public when_synchronous_reads_use_virtual_guards(int path) => _path = path; + + protected override void Context() => _probe = new EnumerationProbe(_path); + + protected override void Because() + { + var result = _probe.Read(); + _callsBeforeEnumeration = _probe.GuardCalls.Count; + _probe.GuardCalls.Clear(); + result.ToArray(); + result.ToArray(); + } + + [Fact] + public void Should_keep_both_virtual_dispatches_for_acquisition_and_every_batch_attempt() + { + // Only the public read's guard runs eagerly. Each enumeration then guards + // acquisition, both populated batches, the empty batch, and completion. + _callsBeforeEnumeration.Should().Be(2); + var expected = new Type?[] + { + typeof(IAsyncCursor), null, + typeof(bool), null, typeof(bool), null, + typeof(bool), null, typeof(bool), null + }; + _probe.GuardCalls.Should().Equal(expected.Concat(expected)); + _probe.FindCalls.Should().Be(2); + _probe.MoveNextCalls.Should().Be(8); + } + + [Fact] + public void Should_reuse_the_callback_within_but_not_across_enumerations() + { + _probe.MoveNextCallbacks.Should().HaveCount(8); + var first = _probe.MoveNextCallbacks[0]; + var second = _probe.MoveNextCallbacks[4]; + first.Should().NotBeSameAs(second); + foreach (var callback in _probe.MoveNextCallbacks.Take(4)) + callback.Should().BeSameAs(first); + foreach (var callback in _probe.MoveNextCallbacks.Skip(4)) + callback.Should().BeSameAs(second); + } + + protected override void Cleanup() => _probe.Engine.Dispose(); + } + internal sealed class EnumerationProbe { public const int ReadCount = 9; @@ -258,6 +314,8 @@ internal sealed class EnumerationProbe public int FindCalls { get; private set; } public int MoveNextCalls { get; private set; } public List> Cursors { get; } = []; + public List GuardCalls => ((ProbeEngine)Engine).GuardCalls; + public List MoveNextCallbacks => ((ProbeEngine)Engine).MoveNextCallbacks; public EnumerationProbe(int path, string? failureStage = null, Exception? failure = null) { @@ -359,6 +417,22 @@ private sealed class ProbeEngine : MongoPersistenceEngine { private readonly IMongoCollection _collection; private readonly Exception? _failure; + public List GuardCalls { get; } = []; + public List MoveNextCallbacks { get; } = []; + + protected override T TryMongo(Func callback) + { + GuardCalls.Add(typeof(T)); + if (typeof(T) == typeof(bool)) + MoveNextCallbacks.Add(callback); + return base.TryMongo(callback); + } + + protected override void TryMongo(Action callback) + { + GuardCalls.Add(null); + base.TryMongo(callback); + } public ProbeEngine(IMongoDatabase database, IDocumentSerializer serializer, IMongoCollection collection, Exception? failure) diff --git a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs index 170cd6c..0933ff8 100644 --- a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs +++ b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs @@ -798,7 +798,8 @@ private IEnumerable EnumerateMongo(IAsyncCursorSource source.ToCursor()); - while (TryMongo(() => cursor.MoveNext())) + Func moveNext = () => cursor.MoveNext(); + while (TryMongo(moveNext)) { foreach (var document in cursor.Current) yield return document;