fix: preserve async application exception origin (#108)

Changelog.md

@@ -28,6 +28,8 @@
 
 ### Breaking Changes
 
+- Async observer reads now preserve MongoDB exceptions thrown by application serializers or `OnNextAsync` callbacks instead of wrapping them as `StorageException` / `StorageUnavailableException`. Failures from the adapter's own MongoDB operations are still translated. The original application exception is delivered to `OnErrorAsync`; public method signatures are unchanged. [#108](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/108)
+
 - `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)
@@ -42,6 +44,8 @@
 
 ### Migration notes
 
+- For #108, update observer error handling to distinguish application exceptions from event-store failures. To retain the stable 12.1.0 wrapping behavior for async observer reads, set `MongoPersistenceOptions.WrapAsyncApplicationExceptions = true` when configuring the persistence engine. The default is `false`. This option does not change synchronous reads, non-observer operations, cancellation, or terminal callback behavior. Custom `TryMongoAsync` overrides that inspect callback exceptions now see an internal wrapper for application-origin Mongo exceptions; delegate to the base guard to unwrap it before notifying observers. Overrides that replace the base guard should enable `WrapAsyncApplicationExceptions` to retain the previous callback exception contract, or be updated for the new origin handling.
+
 - `Initialize()` now creates `GetFrom_RecycleBin_Index` on existing databases. No data backfill or old-index replacement is required, but deployments must budget index build time and disk/I/O costs and retain index-creation permission. Operators can pre-create the documented definition to move build work out of startup. See [recycle-bin rollout and indexing](docs/Recycle-Bin.md#recycle-lookup-index-and-deployment). [#87](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/87)
 
 ## 12.1.0

src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue108AsyncExceptionOrigin.cs

@@ -0,0 +1,209 @@
+#if NUNIT
+using FakeItEasy;
+using FluentAssertions;
+using MongoDB.Bson;
+using MongoDB.Driver;
+using NEventStore.Persistence.AcceptanceTests;
+using NEventStore.Persistence.AcceptanceTests.BDD;
+using NEventStore.Serialization;
+using NUnit.Framework;
+
+namespace NEventStore.Persistence.MongoDB.Tests.AcceptanceTests.Issues
+{
+    // Application failures must not be misidentified as event-store outages.
+    [TestFixtureSource(nameof(Cases))]
+    public class when_Issue108_an_async_read_fails(int path, string stage, string kind, bool legacy) : SpecificationBase
+    {
+        public static IEnumerable<object[]> Cases =>
+            from path in Enumerable.Range(0, 7)
+            from stage in new[] { "serializer", "observer throw", "observer task", "observer delayed", "early stop", "find", "later batch" }
+            where path != 5 || stage != "serializer"
+            from kind in new[] { "mongo", "connection", "other" }
+            where stage != "early stop" || kind == "other"
+            from legacy in new[] { false, true }
+            select new object[] { path, stage, kind, legacy };
+
+        private readonly CancellationTokenSource _cancellation = new();
+        private readonly IAsyncCursor<BsonDocument> _cursor = A.Fake<IAsyncCursor<BsonDocument>>();
+        private ProbeEngine _engine = null!;
+        private Exception _failure = null!;
+        private Exception? _reported;
+        private Exception? _escaped;
+        private readonly TaskCompletionSource<bool> _resume = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        private int _nextCalls;
+        private int _errors;
+        private int _completions;
+
+        protected override void Context()
+        {
+            _failure = EnumerationProbe.Failure(kind);
+            var database = A.Fake<IMongoDatabase>();
+            var collection = A.Fake<IMongoCollection<BsonDocument>>();
+            var serializer = A.Fake<IDocumentSerializer>();
+            A.CallTo(() => collection.Settings).Returns(new MongoCollectionSettings());
+            A.CallTo(() => collection.WithWriteConcern(A<WriteConcern>._)).Returns(collection);
+            A.CallTo(() => database.GetCollection<BsonDocument>(A<string>._, A<MongoCollectionSettings>._)).Returns(collection);
+            A.CallTo(() => serializer.Deserialize<EventMessage>(A<object>._)).Throws(_failure);
+            A.CallTo(() => collection.FindAsync(A<FilterDefinition<BsonDocument>>._,
+                A<FindOptions<BsonDocument, BsonDocument>>._, _cancellation.Token))
+                .ReturnsLazily(() => stage == "find"
+                    ? Task.FromException<IAsyncCursor<BsonDocument>>(_failure)
+                    : Task.FromResult(_cursor));
+            int batch = 0;
+            A.CallTo(() => _cursor.MoveNextAsync(_cancellation.Token)).ReturnsLazily(() =>
+            {
+                batch++;
+                return batch == 2 && stage == "later batch"
+                    ? Task.FromException<bool>(_failure) : Task.FromResult(batch == 1);
+            });
+            A.CallTo(() => _cursor.Current).Returns(new[] { Document(path == 5 ? 7 : 0, 1, stage == "serializer") });
+            _engine = new ProbeEngine(database, serializer, legacy);
+        }
+
+        private IAsyncObserver<T> Observer<T>()
+        {
+            var observer = A.Fake<IAsyncObserver<T>>();
+            A.CallTo(() => observer.OnNextAsync(A<T>._, _cancellation.Token)).ReturnsLazily(() =>
+            {
+                _nextCalls++;
+                if (stage == "observer throw") throw _failure;
+                if (stage == "observer delayed") return FailAfterResumeAsync();
+                if (stage == "early stop") return Task.FromResult(false);
+                return stage == "observer task" ? Task.FromException<bool>(_failure) : Task.FromResult(true);
+            });
+            A.CallTo(() => observer.OnErrorAsync(A<Exception>._, _cancellation.Token))
+                .ReturnsLazily((Exception error, CancellationToken token) =>
+                {
+                    _reported = error;
+                    _errors++;
+                    return Task.CompletedTask;
+                });
+            A.CallTo(() => observer.OnCompletedAsync(_cancellation.Token)).ReturnsLazily(() =>
+            {
+                _completions++;
+                return Task.CompletedTask;
+            });
+            return observer;
+        }
+
+        private async Task<bool> FailAfterResumeAsync()
+        {
+            await _resume.Task.ConfigureAwait(false);
+            throw _failure;
+        }
+
+        protected override async Task BecauseAsync()
+        {
+            var observer = Observer<ICommit>();
+            var operation = Catch.ExceptionAsync(() => path switch
+            {
+                0 => _engine.GetFromAsync("bucket", "stream", 0, int.MaxValue, observer, _cancellation.Token),
+                1 => _engine.GetFromAsync("bucket", 0L, observer, _cancellation.Token),
+                2 => _engine.GetFromToAsync("bucket", 0L, long.MaxValue, observer, _cancellation.Token),
+                3 => _engine.GetFromAsync(0L, observer, _cancellation.Token),
+                4 => _engine.GetFromToAsync(0L, long.MaxValue, observer, _cancellation.Token),
+                5 => _engine.GetStreamsToSnapshotAsync("bucket", 0, Observer<IStreamHead>(), _cancellation.Token),
+                _ => _engine.GetDeletedCommitsAsync(observer, _cancellation.Token)
+            });
+            if (stage == "observer delayed") operation.IsCompleted.Should().BeFalse();
+            _resume.TrySetResult(true);
+            _escaped = await operation.ConfigureAwait(false);
+        }
+
+        [Fact]
+        public void Should_translate_only_storage_failures_unless_legacy_behavior_is_requested()
+        {
+            _escaped.Should().BeNull();
+            if (stage == "early stop")
+            {
+                _reported.Should().BeNull();
+                return;
+            }
+            if (stage == "observer delayed")
+                _failure.StackTrace.Should().Contain(nameof(FailAfterResumeAsync));
+            if (kind == "other" || (!legacy && (stage == "serializer" || stage.StartsWith("observer")))) _reported.Should().BeSameAs(_failure);
+            else
+            {
+                _reported.Should().BeOfType(kind == "connection" ? typeof(StorageUnavailableException) : typeof(StorageException));
+                _reported!.InnerException.Should().BeSameAs(_failure);
+            }
+        }
+
+        [Fact]
+        public void Should_preserve_notification_and_resource_contracts()
+        {
+            // Custom guards observe the origin marker unless legacy behavior is enabled.
+            if (!legacy && kind != "other" && (stage == "serializer" || stage.StartsWith("observer")))
+                _engine.CallbackFailure!.InnerException.Should().BeSameAs(_failure);
+            else if (stage == "early stop")
+                _engine.CallbackFailure.Should().BeNull();
+            else
+                _engine.CallbackFailure.Should().BeSameAs(_failure);
+            _engine.WrapperCalls.Should().Be(1);
+            _engine.Token.Should().Be(_cancellation.Token);
+            _errors.Should().Be(stage == "early stop" ? 0 : 1);
+            _completions.Should().Be(stage == "early stop" ? 1 : 0);
+            _nextCalls.Should().Be(stage == "find" || stage == "serializer" ? 0 : 1);
+            if (stage == "find") A.CallTo(() => _cursor.Dispose()).MustNotHaveHappened();
+            else A.CallTo(() => _cursor.Dispose()).MustHaveHappenedOnceExactly();
+            if (stage == "early stop")
+                A.CallTo(() => _cursor.MoveNextAsync(_cancellation.Token)).MustHaveHappenedOnceExactly();
+        }
+
+        protected override void Cleanup()
+        {
+            _engine.Dispose();
+            _cancellation.Dispose();
+        }
+
+        private sealed class ProbeEngine(IMongoDatabase database, IDocumentSerializer serializer, bool legacy)
+            : MongoPersistenceEngine(database, serializer, new MongoPersistenceOptions { WrapAsyncApplicationExceptions = legacy })
+        {
+            public Exception? CallbackFailure { get; private set; }
+            public int WrapperCalls { get; private set; }
+            public CancellationToken Token { get; private set; }
+            protected override Task TryMongoAsync<T>(Func<Task> callbackAsync, CancellationToken cancellationToken, IAsyncObserver<T>? asyncObserver = null)
+            {
+                WrapperCalls++;
+                Token = cancellationToken;
+                return base.TryMongoAsync(async () =>
+                {
+                    try { await callbackAsync().ConfigureAwait(false); }
+                    catch (Exception error) { CallbackFailure = error; throw; }
+                }, cancellationToken, asyncObserver);
+            }
+        }
+        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()
+                }
+            };
+        }
+
+    }
+}
+#endif

src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue108ObserverFastPath.cs

@@ -0,0 +1,89 @@
+#if NUNIT
+using System.Reflection;
+using FakeItEasy;
+using FluentAssertions;
+using MongoDB.Bson;
+using MongoDB.Driver;
+using NEventStore.Persistence.AcceptanceTests.BDD;
+using NEventStore.Serialization;
+using NUnit.Framework;
+
+namespace NEventStore.Persistence.MongoDB.Tests.AcceptanceTests.Issues
+{
+    [TestFixtureSource(nameof(Cases))]
+    public class when_Issue108_an_observer_task_can_be_reused(bool streamHead, bool legacy, string state) : SpecificationBase
+    {
+        public static IEnumerable<object[]> Cases =>
+            from head in new[] { false, true }
+            from legacy in new[] { false, true }
+            from state in new[] { "true", "false", "pending", "faulted", "canceled", "delayed true", "delayed false" }
+            where legacy || state == "true" || state == "false" || state.StartsWith("delayed")
+            select new object[] { head, legacy, state };
+
+        private MongoPersistenceEngine _engine = null!;
+        private Task<bool> _source = null!;
+        private Task<bool> _actual = null!;
+        private TaskCompletionSource<bool> _completion = null!;
+        private bool _pendingBeforeResume;
+        private bool _result;
+        private object _observer = null!;
+        private BsonDocument _document = null!;
+
+        protected override void Context()
+        {
+            var database = A.Fake<IMongoDatabase>();
+            var collection = A.Fake<IMongoCollection<BsonDocument>>();
+            A.CallTo(() => collection.Settings).Returns(new MongoCollectionSettings());
+            A.CallTo(() => collection.WithWriteConcern(A<WriteConcern>._)).Returns(collection);
+            A.CallTo(() => database.GetCollection<BsonDocument>(A<string>._, A<MongoCollectionSettings>._)).Returns(collection);
+            _engine = new MongoPersistenceEngine(database, A.Fake<IDocumentSerializer>(),
+                new MongoPersistenceOptions { WrapAsyncApplicationExceptions = legacy });
+            var completion = _completion = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+            if (state == "true" || state == "false") completion.SetResult(state == "true");
+            if (state == "faulted") completion.SetException(new MongoException("application failure"));
+            if (state == "canceled") completion.SetCanceled();
+            _source = completion.Task;
+            var commitObserver = A.Fake<IAsyncObserver<ICommit>>();
+            var headObserver = A.Fake<IAsyncObserver<IStreamHead>>();
+            A.CallTo(() => commitObserver.OnNextAsync(A<ICommit>._, CancellationToken.None)).Returns(_source);
+            A.CallTo(() => headObserver.OnNextAsync(A<IStreamHead>._, CancellationToken.None)).Returns(_source);
+            _observer = streamHead ? (object)headObserver : commitObserver;
+            _document = (BsonDocument)typeof(when_Issue108_an_async_read_fails)
+                .GetMethod("Document", BindingFlags.NonPublic | BindingFlags.Static)!
+                .Invoke(null, new object[] { streamHead ? 7 : 0, 1, false })!;
+        }
+
+        protected override async Task BecauseAsync()
+        {
+            // The public read task also owns cursor lifetime; inspect the per-item boundary.
+            _actual = (Task<bool>)typeof(MongoPersistenceEngine)
+                .GetMethod(streamHead ? "ObserveStreamHeadAsync" : "ObserveCommitAsync", BindingFlags.NonPublic | BindingFlags.Instance)!
+                .Invoke(_engine, new object[] { _document, _observer, CancellationToken.None })!;
+            if (state.StartsWith("delayed"))
+            {
+                _pendingBeforeResume = !_actual.IsCompleted;
+                _completion.SetResult(state == "delayed true");
+                _result = await _actual.ConfigureAwait(false);
+            }
+        }
+
+        [Fact]
+        public void Should_reuse_tasks_when_possible_and_preserve_delayed_results()
+        {
+            if (!state.StartsWith("delayed") || legacy) _actual.Should().BeSameAs(_source);
+            if (state.StartsWith("delayed"))
+            {
+                _pendingBeforeResume.Should().BeTrue();
+                _result.Should().Be(state == "delayed true");
+            }
+        }
+
+        protected override void Cleanup()
+        {
+            _ = _source.Exception;
+            _ = _actual.Exception;
+            _engine.Dispose();
+        }
+    }
+}
+#endif