From 0cafe9228dedc391769c0b3ae36aadb778511da3 Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Thu, 12 Nov 2026 12:29:18 +0100 Subject: [PATCH 1/2] fix: preserve async application exception origin (#108) Default async observer reads to preserve application Mongo exceptions. Add a legacy opt-out and document the breaking change. Co-authored-by: Alessandro Giorgetti Co-authored-by: Codex (OpenAI) --- Changelog.md | 4 + .../Issues/Issue108AsyncExceptionOrigin.cs | 209 ++++++++++++++++++ .../MongoPersistenceEngine.Async.cs | 47 +++- .../MongoPersistenceOptions.cs | 8 + 4 files changed, 261 insertions(+), 7 deletions(-) create mode 100644 src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue108AsyncExceptionOrigin.cs diff --git a/Changelog.md b/Changelog.md index f95ac5f..4a6bfd6 100644 --- a/Changelog.md +++ b/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 diff --git a/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue108AsyncExceptionOrigin.cs b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue108AsyncExceptionOrigin.cs new file mode 100644 index 0000000..e108dea --- /dev/null +++ b/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 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 _cursor = A.Fake>(); + private ProbeEngine _engine = null!; + private Exception _failure = null!; + private Exception? _reported; + private Exception? _escaped; + private readonly TaskCompletionSource _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(); + 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); + A.CallTo(() => collection.FindAsync(A>._, + A>._, _cancellation.Token)) + .ReturnsLazily(() => stage == "find" + ? Task.FromException>(_failure) + : Task.FromResult(_cursor)); + int batch = 0; + A.CallTo(() => _cursor.MoveNextAsync(_cancellation.Token)).ReturnsLazily(() => + { + batch++; + return batch == 2 && stage == "later batch" + ? Task.FromException(_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 Observer() + { + var observer = A.Fake>(); + A.CallTo(() => observer.OnNextAsync(A._, _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(_failure) : Task.FromResult(true); + }); + A.CallTo(() => observer.OnErrorAsync(A._, _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 FailAfterResumeAsync() + { + await _resume.Task.ConfigureAwait(false); + throw _failure; + } + + protected override async Task BecauseAsync() + { + var observer = Observer(); + 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(), _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(Func callbackAsync, CancellationToken cancellationToken, IAsyncObserver? 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 diff --git a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs index ceb1b27..5919fd4 100644 --- a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs +++ b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs @@ -47,7 +47,7 @@ public Task GetFromAsync(string bucketId, string streamId, int minRevision, int { foreach (var doc in cursor.Current) { - var res = await observer.OnNextAsync(doc.ToCommit(_serializer), cancellationToken).ConfigureAwait(false); + var res = await ObserveCommitAsync(doc, observer, cancellationToken).ConfigureAwait(false); if (!res) { return; @@ -82,7 +82,7 @@ public Task GetFromAsync(string bucketId, long checkpointToken, IAsyncObserver asyncObse { foreach (var doc in cursor.Current) { - var res = await asyncObserver.OnNextAsync(doc.ToCommit(_serializer), cancellationToken).ConfigureAwait(false); + var res = await ObserveCommitAsync(doc, asyncObserver, cancellationToken).ConfigureAwait(false); if (!res) { return; @@ -209,7 +209,7 @@ public Task GetFromToAsync(long fromCheckpointToken, long toCheckpointToken, IAs { foreach (var doc in cursor.Current) { - var res = await asyncObserver.OnNextAsync(doc.ToCommit(_serializer), cancellationToken).ConfigureAwait(false); + var res = await ObserveCommitAsync(doc, asyncObserver, cancellationToken).ConfigureAwait(false); if (!res) { return; @@ -354,7 +354,7 @@ public Task GetStreamsToSnapshotAsync(string bucketId, int maxThreshold, IAsyncO { foreach (var doc in cursor.Current) { - var res = await asyncObserver.OnNextAsync(doc.ToStreamHead(), cancellationToken).ConfigureAwait(false); + var res = await ObserveStreamHeadAsync(doc, asyncObserver, cancellationToken).ConfigureAwait(false); if (!res) { return; @@ -584,6 +584,11 @@ protected virtual async Task TryMongoAsync(Func callbackAsync, Cancella { await callbackAsync().ConfigureAwait(false); } + catch (AsyncApplicationException e) + { + // Preserve the original application exception and stack, not the transport marker. + exception = ExceptionDispatchInfo.Capture(e.InnerException!); + } catch (MongoConnectionException e) { MongoPersistenceEngineLogMessages.StorageUnavailable(Logger, e); @@ -623,6 +628,34 @@ protected virtual async Task TryMongoAsync(Func callbackAsync, Cancella } } + // Mark only application-origin Mongo exceptions. Cursor acquisition, iteration and + // disposal remain inside the normal storage guard. No shared per-operation state. + private sealed class AsyncApplicationException(Exception innerException) : Exception(null, innerException); + + private async Task ObserveCommitAsync(BsonDocument document, IAsyncObserver observer, CancellationToken cancellationToken) + { + try + { + return await observer.OnNextAsync(document.ToCommit(_serializer), cancellationToken).ConfigureAwait(false); + } + catch (MongoException e) when (!_options.WrapAsyncApplicationExceptions) + { + throw new AsyncApplicationException(e); + } + } + + private async Task ObserveStreamHeadAsync(BsonDocument document, IAsyncObserver observer, CancellationToken cancellationToken) + { + try + { + return await observer.OnNextAsync(document.ToStreamHead(), cancellationToken).ConfigureAwait(false); + } + catch (MongoException e) when (!_options.WrapAsyncApplicationExceptions) + { + throw new AsyncApplicationException(e); + } + } + // Successful observer completion must run outside TryMongoAsync's guarded operation. // Otherwise an OnCompletedAsync failure would be caught and sent to OnErrorAsync, // notifying the observer a second time. Keep the protected virtual guard's success behavior unchanged. @@ -693,7 +726,7 @@ public Task GetDeletedCommitsAsync(IAsyncObserver asyncObserver, Cancel { foreach (var doc in cursor.Current) { - var res = await asyncObserver.OnNextAsync(doc.ToCommit(_serializer), cancellationToken).ConfigureAwait(false); + var res = await ObserveCommitAsync(doc, asyncObserver, cancellationToken).ConfigureAwait(false); if (!res) { return; diff --git a/src/NEventStore.Persistence.MongoDB/MongoPersistenceOptions.cs b/src/NEventStore.Persistence.MongoDB/MongoPersistenceOptions.cs index f816a37..38fe4ad 100644 --- a/src/NEventStore.Persistence.MongoDB/MongoPersistenceOptions.cs +++ b/src/NEventStore.Persistence.MongoDB/MongoPersistenceOptions.cs @@ -170,6 +170,14 @@ public virtual IMongoDatabase ConnectToDatabase(string connectionString) /// public Boolean DisableSnapshotSupport { get; set; } + /// + /// Opts into legacy async observer-read behavior: MongoDB exceptions raised by + /// application serializers or OnNextAsync callbacks are wrapped as storage exceptions. + /// Defaults to false, preserving those application exceptions unchanged. + /// Driver failures from the persistence adapter are always translated. + /// + public Boolean WrapAsyncApplicationExceptions { get; set; } + /// /// The default behavior when using snapshot is to persist the stream heads in /// a background threads, but this way it can be hard to test if the heads and snapshots From caba27b359447f1b244881307e096fefe8a154c9 Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Thu, 12 Nov 2026 16:38:23 +0100 Subject: [PATCH 2/2] perf: avoid redundant async observer task wrappers (#108) Return completed observer tasks directly and share the async exception-marking slow path. Preserve legacy opt-out task forwarding. Co-authored-by: Alessandro Giorgetti Co-authored-by: Codex (OpenAI) --- .../Issues/Issue108ObserverFastPath.cs | 89 +++++++++++++++++++ .../MongoPersistenceEngine.Async.cs | 27 +++++- 2 files changed, 112 insertions(+), 4 deletions(-) create mode 100644 src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue108ObserverFastPath.cs diff --git a/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue108ObserverFastPath.cs b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue108ObserverFastPath.cs new file mode 100644 index 0000000..a790dd9 --- /dev/null +++ b/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 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 _source = null!; + private Task _actual = null!; + private TaskCompletionSource _completion = null!; + private bool _pendingBeforeResume; + private bool _result; + private object _observer = null!; + private BsonDocument _document = null!; + + protected override void Context() + { + var database = A.Fake(); + var collection = A.Fake>(); + A.CallTo(() => collection.Settings).Returns(new MongoCollectionSettings()); + A.CallTo(() => collection.WithWriteConcern(A._)).Returns(collection); + A.CallTo(() => database.GetCollection(A._, A._)).Returns(collection); + _engine = new MongoPersistenceEngine(database, A.Fake(), + new MongoPersistenceOptions { WrapAsyncApplicationExceptions = legacy }); + var completion = _completion = new TaskCompletionSource(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>(); + var headObserver = A.Fake>(); + A.CallTo(() => commitObserver.OnNextAsync(A._, CancellationToken.None)).Returns(_source); + A.CallTo(() => headObserver.OnNextAsync(A._, 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)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 diff --git a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs index 5919fd4..92d09f1 100644 --- a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs +++ b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs @@ -632,11 +632,11 @@ protected virtual async Task TryMongoAsync(Func callbackAsync, Cancella // disposal remain inside the normal storage guard. No shared per-operation state. private sealed class AsyncApplicationException(Exception innerException) : Exception(null, innerException); - private async Task ObserveCommitAsync(BsonDocument document, IAsyncObserver observer, CancellationToken cancellationToken) + private Task ObserveCommitAsync(BsonDocument document, IAsyncObserver observer, CancellationToken cancellationToken) { try { - return await observer.OnNextAsync(document.ToCommit(_serializer), cancellationToken).ConfigureAwait(false); + return ObserveApplicationTask(observer.OnNextAsync(document.ToCommit(_serializer), cancellationToken)); } catch (MongoException e) when (!_options.WrapAsyncApplicationExceptions) { @@ -644,11 +644,30 @@ private async Task ObserveCommitAsync(BsonDocument document, IAsyncObserve } } - private async Task ObserveStreamHeadAsync(BsonDocument document, IAsyncObserver observer, CancellationToken cancellationToken) + private Task ObserveStreamHeadAsync(BsonDocument document, IAsyncObserver observer, CancellationToken cancellationToken) { try { - return await observer.OnNextAsync(document.ToStreamHead(), cancellationToken).ConfigureAwait(false); + return ObserveApplicationTask(observer.OnNextAsync(document.ToStreamHead(), cancellationToken)); + } + catch (MongoException e) when (!_options.WrapAsyncApplicationExceptions) + { + throw new AsyncApplicationException(e); + } + } + + private Task ObserveApplicationTask(Task task) + { + // Completed callbacks and legacy mode need no additional async state machine. + return task.Status == TaskStatus.RanToCompletion || _options.WrapAsyncApplicationExceptions + ? task : AwaitApplicationTaskAsync(task); + } + + private async Task AwaitApplicationTaskAsync(Task task) + { + try + { + return await task.ConfigureAwait(false); } catch (MongoException e) when (!_options.WrapAsyncApplicationExceptions) {