From a32487a6b8db639f7c218f81137d5c1beae53411 Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Thu, 29 Oct 2026 11:10:47 +0100 Subject: [PATCH 1/9] test: reproduce async cancellation and observer termination regressions Co-authored by Alessandro Giorgetti and Codex (OpenAI). --- .../Issues/Issue86CancellationTests.Async.cs | 504 ++++++++++++++++++ 1 file changed, 504 insertions(+) create mode 100644 src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue86CancellationTests.Async.cs diff --git a/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue86CancellationTests.Async.cs b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue86CancellationTests.Async.cs new file mode 100644 index 0000000..0e68f45 --- /dev/null +++ b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue86CancellationTests.Async.cs @@ -0,0 +1,504 @@ +using FluentAssertions; +using MongoDB.Bson; +using MongoDB.Driver; +using NEventStore.Persistence.AcceptanceTests; +using NEventStore.Persistence.AcceptanceTests.BDD; +using NEventStore.Persistence.MongoDB.Support; +using NEventStore.Serialization; +#if NUNIT +using NUnit.Framework; + +namespace NEventStore.Persistence.MongoDB.Tests.AcceptanceTests.Issues.Async +{ + public abstract class Issue_86_concern : SpecificationBase + { + protected CancellationEngine Engine = null!; + protected CancellationTokenSource Cancellation = null!; + protected CommitAttempt Attempt = null!; + protected Task Operation = null!; + protected Exception? Error; + protected IMongoDatabase Database = null!; + protected readonly Checkpoints Generator = new(); + private MongoClient _client = null!; + + protected override void Context() + { + // Register the acceptance suite's serializers before using real MongoDB collections. + _ = new AcceptanceTestMongoPersistenceFactory(); + var url = new MongoUrl(AcceptanceTestMongoPersistenceFactory.GetConnectionString()); + _client = new MongoClient(url); + Database = _client.GetDatabase(url.DatabaseName + "_issue86_" + Guid.NewGuid().ToString("N")); + Engine = new CancellationEngine(Database, new DocumentObjectSerializer(), new SnapshotOptions + { + CheckpointGenerator = Generator, + PersistStreamHeadsOnBackgroundThread = false + }); + Engine.Initialize(); + Cancellation = new CancellationTokenSource(); + Attempt = Guid.NewGuid().ToString().BuildAttempt(); + } + + protected override void Cleanup() + { + Engine?.Dispose(); + Cancellation?.Dispose(); + if (Database != null) _client.DropDatabase(Database.DatabaseNamespace.DatabaseName); + } + + protected static async Task WaitFor(Task task) + { + if (await Task.WhenAny(task, Task.Delay(TimeSpan.FromSeconds(10))).ConfigureAwait(false) != task) + throw new TimeoutException("The deterministic cancellation gate was not reached."); + await task.ConfigureAwait(false); + } + + protected static TaskCompletionSource Gate() => new(TaskCreationOptions.RunContinuationsAsynchronously); + + protected class Checkpoints : ICheckpointGenerator + { + private long _next; + public Func>? GenerateAsync { get; set; } + public long Next() => Interlocked.Increment(ref _next); + public Task NextAsync(CancellationToken cancellationToken = default) => + GenerateAsync?.Invoke(cancellationToken) ?? Task.FromResult(Next()); + public void SignalDuplicateId(long id) => throw new InvalidOperationException("Unexpected checkpoint collision."); + public Task SignalDuplicateIdAsync(long id, CancellationToken cancellationToken = default) => + throw new InvalidOperationException("Unexpected checkpoint collision."); + } + + protected class CancellationEngine : MongoPersistenceEngine + { + public CancellationEngine(IMongoDatabase database, IDocumentSerializer serializer, MongoPersistenceOptions options) + : base(database, serializer, options) { } + public Task Execute(Func callback, CancellationToken token, IAsyncObserver? observer = null) => + TryMongoAsync(callback, token, observer); + public Task ExecuteValue(Func> callback) => TryMongoAsync(callback); + } + + private class SnapshotOptions : MongoPersistenceOptions + { + // Make test setup and postconditions deterministic; production write concerns are unchanged. + public override MongoCollectionSettings GetSnapshotSettings() => new() { WriteConcern = WriteConcern.Acknowledged }; + public override MongoCollectionSettings GetStreamSettings() => new() { WriteConcern = WriteConcern.Acknowledged }; + } + } + + [TestFixture(false, false)] + [TestFixture(false, true)] + [TestFixture(true, false)] + [TestFixture(true, true)] + public class when_Issue_86_commit_checkpoint_generation_is_canceled : Issue_86_concern + { + private readonly bool _inFlight; + private readonly bool _operationCanceledException; + private readonly TaskCompletionSource _entered = Gate(); + private readonly TaskCompletionSource _release = Gate(); + private long _persisted; + private bool _wasPending; + + public when_Issue_86_commit_checkpoint_generation_is_canceled(bool inFlight, bool operationCanceledException) + { + _inFlight = inFlight; + _operationCanceledException = operationCanceledException; + } + + protected override void Context() + { + base.Context(); + Generator.GenerateAsync = async token => + { + _entered.SetResult(true); + if (_inFlight) await _release.Task.ConfigureAwait(false); + if (_operationCanceledException) throw new OperationCanceledException(token); + return await Task.FromCanceled(token).ConfigureAwait(false); + }; + if (!_inFlight) Cancellation.Cancel(); + } + + protected override async Task BecauseAsync() + { + Operation = Engine.CommitAsync(Attempt, Cancellation.Token); + await WaitFor(_entered.Task).ConfigureAwait(false); + _wasPending = !Operation.IsCompleted; + Cancellation.Cancel(); + _release.SetResult(true); + Error = await Catch.ExceptionAsync(() => Operation).ConfigureAwait(false); + _persisted = Database.GetCollection("Commits").CountDocuments(FilterDefinition.Empty); + } + + [Fact] + public void Should_return_a_canceled_task_instead_of_a_successful_null_commit() + { + Operation.IsCanceled.Should().BeTrue(); + Error.Should().BeAssignableTo(); + ((OperationCanceledException)Error!).CancellationToken.Should().Be(Cancellation.Token); + } + + [Fact] + public void Should_not_write_the_canceled_commit() => _persisted.Should().Be(0); + + [Fact] + public void Should_cancel_after_the_checkpoint_operation_started() => _wasPending.Should().Be(_inFlight); + } + + [TestFixture(false, false, false)] + [TestFixture(false, false, true)] + [TestFixture(false, true, false)] + [TestFixture(false, true, true)] + [TestFixture(true, false, false)] + [TestFixture(true, false, true)] + [TestFixture(true, true, false)] + [TestFixture(true, true, true)] + public class when_Issue_86_a_wrapped_operation_is_canceled : Issue_86_concern + { + private readonly bool _hasResult; + private readonly bool _inFlight; + private readonly bool _operationCanceledException; + private readonly TaskCompletionSource _entered = Gate(); + private readonly TaskCompletionSource _release = Gate(); + private bool _wasPending; + + public when_Issue_86_a_wrapped_operation_is_canceled(bool hasResult, bool inFlight, bool operationCanceledException) + { + _hasResult = hasResult; + _inFlight = inFlight; + _operationCanceledException = operationCanceledException; + } + + protected override void Context() + { + base.Context(); + if (!_inFlight) Cancellation.Cancel(); + } + + private async Task CancelOperation() + { + _entered.SetResult(true); + if (_inFlight) await _release.Task.ConfigureAwait(false); + if (_operationCanceledException) throw new OperationCanceledException(Cancellation.Token); + return await Task.FromCanceled(Cancellation.Token).ConfigureAwait(false); + } + + protected override async Task BecauseAsync() + { + Operation = _hasResult ? Engine.ExecuteValue(CancelOperation) : Engine.Execute(async () => + { + await CancelOperation().ConfigureAwait(false); + }, Cancellation.Token); + await WaitFor(_entered.Task).ConfigureAwait(false); + _wasPending = !Operation.IsCompleted; + Cancellation.Cancel(); + _release.SetResult(true); + Error = await Catch.ExceptionAsync(() => Operation).ConfigureAwait(false); + } + + [Fact] + public void Should_preserve_the_canceled_task_and_token() + { + Operation.IsCanceled.Should().BeTrue(); + Error.Should().BeAssignableTo(); + ((OperationCanceledException)Error!).CancellationToken.Should().Be(Cancellation.Token); + } + + [Fact] + public void Should_exercise_cancellation_at_the_requested_boundary() => _wasPending.Should().Be(_inFlight); + } + + public class when_Issue_86_a_derived_engine_completes_its_own_callback : Issue_86_concern + { + private int _completed; + private IAsyncObserver _observer = null!; + + protected override void Context() + { + base.Context(); + _observer = new LambdaAsyncObserver((_, _) => Task.FromResult(true), onCompletedAsync: _ => + { + _completed++; + return Task.CompletedTask; + }); + } + + protected override Task BecauseAsync() => Engine.Execute(() => _observer.OnCompletedAsync(Cancellation.Token), Cancellation.Token, _observer); + + [Fact] + public void Should_not_add_completion_to_the_protected_wrapper_success_path() => _completed.Should().Be(1); + } + + [TestFixture("snapshot-read")] + [TestFixture("snapshot-write")] + [TestFixture("purge")] + [TestFixture("purge-bucket")] + [TestFixture("delete-stream")] + [TestFixture("empty-recycle-bin")] + public class when_Issue_86_an_ordinary_operation_is_already_canceled : Issue_86_concern + { + private readonly string _operation; + private long _commits; + private long _snapshots; + private ISnapshot _snapshot = null!; + + public when_Issue_86_an_ordinary_operation_is_already_canceled(string operation) => _operation = operation; + + protected override void Context() + { + base.Context(); + Engine.Commit(Attempt); + _snapshot = new Snapshot(Attempt.BucketId, Attempt.StreamId, 1, "snapshot"); + Cancellation.Cancel(); + } + + protected override async Task BecauseAsync() + { + Operation = _operation switch + { + "snapshot-read" => Engine.GetSnapshotAsync(Attempt.BucketId, Attempt.StreamId, 1, Cancellation.Token), + "snapshot-write" => Engine.AddSnapshotAsync(_snapshot, Cancellation.Token), + "purge" => Engine.PurgeAsync(Cancellation.Token), + "purge-bucket" => Engine.PurgeAsync(Attempt.BucketId, Cancellation.Token), + "delete-stream" => Engine.DeleteStreamAsync(Attempt.BucketId, Attempt.StreamId, Cancellation.Token), + _ => Engine.EmptyRecycleBinAsync(Cancellation.Token) + }; + Error = await Catch.ExceptionAsync(() => Operation).ConfigureAwait(false); + _commits = Database.GetCollection("Commits").CountDocuments(FilterDefinition.Empty); + _snapshots = Database.GetCollection("Snapshots").CountDocuments(FilterDefinition.Empty); + } + + [Fact] + public void Should_not_disguise_cancellation_as_success_or_missing_data() + { + Operation.IsCanceled.Should().BeTrue(); + Error.Should().BeAssignableTo(); + } + + [Fact] + public void Should_leave_persisted_data_unchanged() + { + _commits.Should().Be(1); + _snapshots.Should().Be(0); + } + } + + [TestFixture("stream", "canceled")] + [TestFixture("bucket", "canceled")] + [TestFixture("bucket-range", "canceled")] + [TestFixture("all", "canceled")] + [TestFixture("range", "canceled")] + [TestFixture("heads", "canceled")] + [TestFixture("deleted", "canceled")] + [TestFixture("stream", "stop")] + [TestFixture("bucket", "stop")] + [TestFixture("bucket-range", "stop")] + [TestFixture("all", "stop")] + [TestFixture("range", "stop")] + [TestFixture("heads", "stop")] + [TestFixture("deleted", "stop")] + [TestFixture("stream", "success")] + [TestFixture("bucket", "success")] + [TestFixture("bucket-range", "success")] + [TestFixture("all", "success")] + [TestFixture("range", "success")] + [TestFixture("heads", "success")] + [TestFixture("deleted", "success")] + [TestFixture("stream", "next-task-canceled")] + [TestFixture("stream", "next-operation-canceled")] + [TestFixture("stream", "next-error")] + [TestFixture("stream", "completion-task-canceled")] + [TestFixture("stream", "completion-operation-canceled")] + [TestFixture("stream", "completion-error")] + [TestFixture("stream", "error-callback-error")] + [TestFixture("stream", "error-callback-canceled")] + public class when_Issue_86_an_observer_receives_a_terminal_notification : Issue_86_concern + { + private readonly string _read; + private readonly string _mode; + private readonly TaskCompletionSource _terminalEntered = Gate(); + private readonly TaskCompletionSource _terminalRelease = Gate(); + private readonly TaskCompletionSource _nextEntered = Gate(); + private readonly TaskCompletionSource _nextRelease = Gate(); + private readonly Exception _failure = new InvalidOperationException("Observer failure."); + private int _completed; + private int _errors; + private int _next; + private bool _pendingAtTerminal; + private bool _pendingAtNext; + private CancellationToken _terminalToken; + private Exception? _reportedError; + private IAsyncObserver _commits = null!; + private IAsyncObserver _heads = null!; + + public when_Issue_86_an_observer_receives_a_terminal_notification(string read, string mode) + { + _read = read; + _mode = mode; + } + + protected override void Context() + { + base.Context(); + Engine.Commit(Attempt); + if (_read == "heads") Engine.CommitSingle(); + else Engine.CommitNext(Attempt); + if (_read == "deleted") Engine.DeleteStream(Attempt.BucketId, Attempt.StreamId); + if (_mode == "canceled") Cancellation.Cancel(); + _commits = new LambdaAsyncObserver((_, token) => OnNext(token), OnError, OnCompleted); + _heads = new LambdaAsyncObserver((_, token) => OnNext(token), OnError, OnCompleted); + } + + private async Task OnNext(CancellationToken token) + { + _next++; + if (_mode.StartsWith("next-", StringComparison.Ordinal)) + { + _nextEntered.TrySetResult(true); + await _nextRelease.Task.ConfigureAwait(false); + if (_mode == "next-task-canceled") return await Task.FromCanceled(token).ConfigureAwait(false); + if (_mode == "next-operation-canceled") throw new OperationCanceledException(token); + throw _failure; + } + if (_mode.StartsWith("error-callback-", StringComparison.Ordinal)) throw _failure; + return _mode != "stop"; + } + + private async Task OnCompleted(CancellationToken token) + { + _completed++; + _terminalToken = token; + _terminalEntered.TrySetResult(true); + await _terminalRelease.Task.ConfigureAwait(false); + if (_mode == "completion-task-canceled") await Task.FromCanceled(token).ConfigureAwait(false); + if (_mode == "completion-operation-canceled") throw new OperationCanceledException(token); + if (_mode == "completion-error") throw _failure; + } + + private async Task OnError(Exception error, CancellationToken token) + { + _errors++; + _reportedError = error; + _terminalToken = token; + _terminalEntered.TrySetResult(true); + await _terminalRelease.Task.ConfigureAwait(false); + if (_mode == "error-callback-error") throw _failure; + if (_mode == "error-callback-canceled") await Task.FromCanceled(token).ConfigureAwait(false); + } + + protected override async Task BecauseAsync() + { + Operation = _read switch + { + "stream" => Engine.GetFromAsync(Attempt.BucketId, Attempt.StreamId, 0, int.MaxValue, _commits, Cancellation.Token), + "bucket" => Engine.GetFromAsync(Attempt.BucketId, 0, _commits, Cancellation.Token), + "bucket-range" => Engine.GetFromToAsync(Attempt.BucketId, 0, long.MaxValue, _commits, Cancellation.Token), + "all" => Engine.GetFromAsync(0, _commits, Cancellation.Token), + "range" => Engine.GetFromToAsync(0, long.MaxValue, _commits, Cancellation.Token), + "heads" => Engine.GetStreamsToSnapshotAsync(Attempt.BucketId, 1, _heads, Cancellation.Token), + _ => Engine.GetDeletedCommitsAsync(_commits, Cancellation.Token) + }; + if (_mode.StartsWith("next-", StringComparison.Ordinal)) + { + await WaitFor(_nextEntered.Task).ConfigureAwait(false); + _pendingAtNext = !Operation.IsCompleted; + if (_mode != "next-error") Cancellation.Cancel(); + _nextRelease.SetResult(true); + } + await WaitFor(_terminalEntered.Task).ConfigureAwait(false); + _pendingAtTerminal = !Operation.IsCompleted; + if (_mode.StartsWith("completion-", StringComparison.Ordinal) || _mode == "error-callback-canceled") Cancellation.Cancel(); + _terminalRelease.SetResult(true); + Error = await Catch.ExceptionAsync(() => Operation).ConfigureAwait(false); + } + + [Fact] + public void Should_await_the_terminal_callback_and_preserve_the_original_token() + { + _pendingAtTerminal.Should().BeTrue(); + _terminalToken.Should().Be(Cancellation.Token); + } + + [Fact] + public void Should_deliver_exactly_one_terminal_notification() + { + bool dataError = _mode == "next-error" || _mode.StartsWith("error-callback-", StringComparison.Ordinal); + _completed.Should().Be(dataError ? 0 : 1); + _errors.Should().Be(dataError ? 1 : 0); + _reportedError.Should().BeSameAs(dataError ? _failure : null); + } + + [Fact] + public void Should_preserve_callback_failures_without_renotifying_the_observer() + { + if (_mode == "completion-task-canceled" || _mode == "completion-operation-canceled" || _mode == "error-callback-canceled") + { + Error.Should().BeAssignableTo(); + Operation.IsCanceled.Should().BeTrue(); + } + else if (_mode == "completion-error" || _mode == "error-callback-error") + Error.Should().BeSameAs(_failure); + else + Error.Should().BeNull("data cancellation is observer completion, not ordinary task cancellation"); + } + + [Fact] + public void Should_stop_reading_when_canceled_or_requested_by_the_observer() + { + _next.Should().Be(_mode == "canceled" ? 0 : _mode == "success" || _mode.StartsWith("completion-", StringComparison.Ordinal) ? 2 : 1); + _pendingAtNext.Should().Be(_mode.StartsWith("next-", StringComparison.Ordinal)); + } + } + + [TestFixture("commit", false)] + [TestFixture("commit", true)] + [TestFixture("snapshot", false)] + [TestFixture("snapshot", true)] + [TestFixture("missing-snapshot", false)] + [TestFixture("missing-snapshot", true)] + [TestFixture("invalid-snapshot", false)] + [TestFixture("invalid-snapshot", true)] + [TestFixture("null-snapshot", false)] + [TestFixture("null-snapshot", true)] + public class when_Issue_86_an_ordinary_operation_is_not_canceled : Issue_86_concern + { + private readonly string _operation; + private object? _result; + private readonly bool _synchronous; + private ISnapshot _snapshot = null!; + public when_Issue_86_an_ordinary_operation_is_not_canceled(string operation, bool synchronous) + { + _operation = operation; + _synchronous = synchronous; + } + + protected override void Context() + { + base.Context(); + if (_operation == "snapshot") Engine.Commit(Attempt); + _snapshot = _operation == "null-snapshot" ? null! : new Snapshot(Attempt.BucketId, Attempt.StreamId, 1, "snapshot"); + } + + protected override async Task BecauseAsync() + { + if (_operation == "commit") _result = _synchronous ? Engine.Commit(Attempt) : await Engine.CommitAsync(Attempt, CancellationToken.None).ConfigureAwait(false); + else if (_operation == "missing-snapshot") + { + _result = _synchronous ? Engine.GetSnapshot(Attempt.BucketId, Attempt.StreamId, 1) : + await Engine.GetSnapshotAsync(Attempt.BucketId, Attempt.StreamId, 1, CancellationToken.None).ConfigureAwait(false); + } + else + { + _result = _synchronous ? Engine.AddSnapshot(_snapshot) : + await Engine.AddSnapshotAsync(_snapshot, CancellationToken.None).ConfigureAwait(false); + } + } + + [Fact] + public void Should_preserve_success_and_non_cancellation_negative_results() + { + if (_operation == "commit") _result.Should().BeAssignableTo(); + else if (_operation == "snapshot") _result.Should().Be(true); + else if (_operation == "missing-snapshot") _result.Should().BeNull(); + else _result.Should().Be(false); + } + + } +} +#endif From d682aea89683c05dc43e1bbb0442e54e876a967f Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Thu, 29 Oct 2026 11:11:12 +0100 Subject: [PATCH 2/9] fix: preserve cancellation and await observer termination Co-authored by Alessandro Giorgetti and Codex (OpenAI). --- .../MongoPersistenceEngine.Async.cs | 61 +++++++++++-------- 1 file changed, 36 insertions(+), 25 deletions(-) diff --git a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs index b1ca6b9..e210688 100644 --- a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs +++ b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs @@ -15,7 +15,7 @@ public Task GetFromAsync(string bucketId, string streamId, int minRevision, int { MongoPersistenceEngineLogMessages.GettingAllCommitsBetween(Logger, streamId, bucketId, minRevision, maxRevision); - return TryMongoAsync(async () => + return ObserveMongoAsync(async () => { var filters = new List>() { @@ -50,12 +50,10 @@ public Task GetFromAsync(string bucketId, string streamId, int minRevision, int var res = await observer.OnNextAsync(doc.ToCommit(_serializer), cancellationToken).ConfigureAwait(false); if (!res) { - await observer.OnCompletedAsync(cancellationToken).ConfigureAwait(false); return; } } } - await observer.OnCompletedAsync(cancellationToken).ConfigureAwait(false); }, cancellationToken, observer); } @@ -65,7 +63,7 @@ public Task GetFromAsync(string bucketId, long checkpointToken, IAsyncObserver + return ObserveMongoAsync(async () => { var query = Builders.Filter.Eq(MongoCommitFields.BucketId, bucketId); if (checkpointToken > 0) @@ -87,12 +85,10 @@ public Task GetFromAsync(string bucketId, long checkpointToken, IAsyncObserver + return ObserveMongoAsync(async () => { var filters = new List>() { @@ -134,12 +130,10 @@ public Task GetFromToAsync(string bucketId, long fromCheckpointToken, long toChe var res = await asyncObserver.OnNextAsync(doc.ToCommit(_serializer), cancellationToken).ConfigureAwait(false); if (!res) { - await asyncObserver.OnCompletedAsync(cancellationToken).ConfigureAwait(false); return; } } } - await asyncObserver.OnCompletedAsync(cancellationToken).ConfigureAwait(false); }, cancellationToken, asyncObserver); @@ -150,7 +144,7 @@ public Task GetFromAsync(long checkpointToken, IAsyncObserver asyncObse { MongoPersistenceEngineLogMessages.GettingAllCommitsFromCheckpoint(Logger, checkpointToken); - return TryMongoAsync(async () => + return ObserveMongoAsync(async () => { var query = Builders.Filter.Ne(MongoCommitFields.BucketId, MongoSystemBuckets.RecycleBin); if (checkpointToken > 0) @@ -172,12 +166,10 @@ public Task GetFromAsync(long checkpointToken, IAsyncObserver asyncObse var res = await asyncObserver.OnNextAsync(doc.ToCommit(_serializer), cancellationToken).ConfigureAwait(false); if (!res) { - await asyncObserver.OnCompletedAsync(cancellationToken).ConfigureAwait(false); return; } } } - await asyncObserver.OnCompletedAsync(cancellationToken).ConfigureAwait(false); }, cancellationToken, asyncObserver); @@ -188,7 +180,7 @@ public Task GetFromToAsync(long fromCheckpointToken, long toCheckpointToken, IAs { MongoPersistenceEngineLogMessages.GettingCommitsFromToCheckpoint(Logger, fromCheckpointToken, toCheckpointToken); - return TryMongoAsync(async () => + return ObserveMongoAsync(async () => { var filters = new List>() { @@ -220,12 +212,10 @@ public Task GetFromToAsync(long fromCheckpointToken, long toCheckpointToken, IAs var res = await asyncObserver.OnNextAsync(doc.ToCommit(_serializer), cancellationToken).ConfigureAwait(false); if (!res) { - await asyncObserver.OnCompletedAsync(cancellationToken).ConfigureAwait(false); return; } } } - await asyncObserver.OnCompletedAsync(cancellationToken).ConfigureAwait(false); }, cancellationToken, asyncObserver); @@ -348,7 +338,7 @@ public Task GetStreamsToSnapshotAsync(string bucketId, int maxThreshold, IAsyncO MongoPersistenceEngineLogMessages.GettingStreamsToSnapshot(Logger); - return TryMongoAsync(async () => + return ObserveMongoAsync(async () => { var query = Builders.Filter.And( Builders.Filter.Eq(MongoStreamHeadFields.FullQualifiedBucketId, bucketId), @@ -367,12 +357,10 @@ public Task GetStreamsToSnapshotAsync(string bucketId, int maxThreshold, IAsyncO var res = await asyncObserver.OnNextAsync(doc.ToStreamHead(), cancellationToken).ConfigureAwait(false); if (!res) { - await asyncObserver.OnCompletedAsync(cancellationToken).ConfigureAwait(false); return; } } } - await asyncObserver.OnCompletedAsync(cancellationToken).ConfigureAwait(false); }, cancellationToken, asyncObserver); } @@ -447,6 +435,10 @@ await PersistedStreamHeads.UpdateOneAsync( return true; } + catch (OperationCanceledException) + { + throw; + } catch (Exception e) { MongoPersistenceEngineLogMessages.AddingSnapshotError(Logger, snapshot.StreamId, snapshot.BucketId, snapshot.StreamRevision, e); @@ -572,8 +564,9 @@ await TryMongoAsync( /// /// Executes the callback within a try/catch block to handle exceptions. /// If an observer is provided, it will be notified of any exceptions that occur during the operation. - /// If an observer in provided and the operation was cancelled, the observer will be notified of the cancellation calling OnCompleteAsync. - /// If no observer is provided, exceptions will be raised + /// If an observer is provided and the operation was cancelled, its OnCompletedAsync callback will be awaited. + /// If no observer is provided, exceptions (including cancellation) will be raised. + /// Exceptions from terminal observer callbacks propagate to the caller without another notification. /// /// Callback to execute /// Cancellation token @@ -605,10 +598,14 @@ protected virtual async Task TryMongoAsync(Func callbackAsync, Cancella MongoPersistenceEngineLogMessages.StorageThrewException(Logger, e); exception = ExceptionDispatchInfo.Capture(new StorageException(e.Message, e)); } - catch (TaskCanceledException ex) + catch (OperationCanceledException ex) { MongoPersistenceEngineLogMessages.TaskWasCancelled(Logger, ex); - asyncObserver?.OnCompletedAsync(cancellationToken); + if (asyncObserver == null) + { + throw; + } + await asyncObserver.OnCompletedAsync(cancellationToken).ConfigureAwait(false); } catch (Exception e) { @@ -628,6 +625,22 @@ protected virtual async Task TryMongoAsync(Func callbackAsync, Cancella } } + private async Task ObserveMongoAsync(Func callbackAsync, CancellationToken cancellationToken, IAsyncObserver asyncObserver) + { + bool completed = false; + await TryMongoAsync(async () => + { + await callbackAsync().ConfigureAwait(false); + completed = true; + }, cancellationToken, asyncObserver).ConfigureAwait(false); + + // Notify outside the operation's catch block: a terminal callback must not be retried or sent to OnErrorAsync. + if (completed) + { + await asyncObserver.OnCompletedAsync(cancellationToken).ConfigureAwait(false); + } + } + private async Task GetLastCommittedCheckPointNumberAsync(CancellationToken cancellationToken) { var filter = Builders.Filter.Empty; @@ -668,7 +681,7 @@ await TryMongoAsync(() => /// public Task GetDeletedCommitsAsync(IAsyncObserver asyncObserver, CancellationToken cancellationToken) { - return TryMongoAsync(async () => + return ObserveMongoAsync(async () => { using var cursor = await PersistedCommits .Find(Builders.Filter.Eq(MongoCommitFields.BucketId, MongoSystemBuckets.RecycleBin)) @@ -682,12 +695,10 @@ public Task GetDeletedCommitsAsync(IAsyncObserver asyncObserver, Cancel var res = await asyncObserver.OnNextAsync(doc.ToCommit(_serializer), cancellationToken).ConfigureAwait(false); if (!res) { - await asyncObserver.OnCompletedAsync(cancellationToken).ConfigureAwait(false); return; } } } - await asyncObserver.OnCompletedAsync(cancellationToken).ConfigureAwait(false); }, cancellationToken, asyncObserver); From 6ce7edc31c792b005c8e15f057d800120776a3ac Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Thu, 29 Oct 2026 11:11:31 +0100 Subject: [PATCH 3/9] docs: update changelog for #86 --- Changelog.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/Changelog.md b/Changelog.md index 9c56ec5..88dc51d 100644 --- a/Changelog.md +++ b/Changelog.md @@ -10,10 +10,14 @@ - Fix initialization races: serialize initialization with a lock and retain success only after all setup completes. A waiting or later caller retries after failure; successful calls remain idempotent. Same-thread recursive initialization fails explicitly. Use `System.Threading.Lock` on .NET 9+ and an object lock on older targets. [#85](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/85) +- Propagate cancellation from async operations, including snapshot writes. Observer reads still signal cancellation through `OnCompletedAsync` with the original token; terminal callbacks are now awaited, and callback outcomes reach the returned task. [#86](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/86) + ### Breaking Changes - 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) +- Non-observer async callers must now handle `OperationCanceledException` (including `TaskCanceledException`) from operations that previously swallowed cancellation or returned success/default values; for example, `AddSnapshotAsync` now propagates cancellation instead of returning `false`. Observer reads continue to invoke `OnCompletedAsync` with the original token on cancellation, but cancellation or failure from that callback now propagates after it is awaited, without a second terminal notification. [#86](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/86) + ### Migration notes - `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) From 9bac76a929c796385cd27cf23393562d0189693a Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Thu, 29 Oct 2026 11:29:09 +0100 Subject: [PATCH 4/9] test: classify observer cancellation by request token --- .../Issues/Issue86CancellationTests.Async.cs | 35 +++++++++++++++++-- 1 file changed, 32 insertions(+), 3 deletions(-) diff --git a/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue86CancellationTests.Async.cs b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue86CancellationTests.Async.cs index 0e68f45..216e0e7 100644 --- a/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue86CancellationTests.Async.cs +++ b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue86CancellationTests.Async.cs @@ -302,6 +302,8 @@ public void Should_leave_persisted_data_unchanged() [TestFixture("deleted", "success")] [TestFixture("stream", "next-task-canceled")] [TestFixture("stream", "next-operation-canceled")] + [TestFixture("stream", "next-unrelated-task-canceled")] + [TestFixture("stream", "next-unrelated-operation-canceled")] [TestFixture("stream", "next-error")] [TestFixture("stream", "completion-task-canceled")] [TestFixture("stream", "completion-operation-canceled")] @@ -324,6 +326,7 @@ public class when_Issue_86_an_observer_receives_a_terminal_notification : Issue_ private bool _pendingAtNext; private CancellationToken _terminalToken; private Exception? _reportedError; + private Exception? _unrelatedCancellation; private IAsyncObserver _commits = null!; private IAsyncObserver _heads = null!; @@ -354,6 +357,25 @@ private async Task OnNext(CancellationToken token) await _nextRelease.Task.ConfigureAwait(false); if (_mode == "next-task-canceled") return await Task.FromCanceled(token).ConfigureAwait(false); if (_mode == "next-operation-canceled") throw new OperationCanceledException(token); + if (_mode == "next-unrelated-task-canceled" || _mode == "next-unrelated-operation-canceled") + { + using var unrelatedCancellation = new CancellationTokenSource(); + unrelatedCancellation.Cancel(); + if (_mode == "next-unrelated-task-canceled") + { + try + { + return await Task.FromCanceled(unrelatedCancellation.Token).ConfigureAwait(false); + } + catch (OperationCanceledException exception) + { + _unrelatedCancellation = exception; + throw; + } + } + _unrelatedCancellation = new OperationCanceledException(unrelatedCancellation.Token); + throw _unrelatedCancellation; + } throw _failure; } if (_mode.StartsWith("error-callback-", StringComparison.Ordinal)) throw _failure; @@ -398,7 +420,7 @@ protected override async Task BecauseAsync() { await WaitFor(_nextEntered.Task).ConfigureAwait(false); _pendingAtNext = !Operation.IsCompleted; - if (_mode != "next-error") Cancellation.Cancel(); + if (_mode != "next-error" && !_mode.StartsWith("next-unrelated-", StringComparison.Ordinal)) Cancellation.Cancel(); _nextRelease.SetResult(true); } await WaitFor(_terminalEntered.Task).ConfigureAwait(false); @@ -418,10 +440,17 @@ public void Should_await_the_terminal_callback_and_preserve_the_original_token() [Fact] public void Should_deliver_exactly_one_terminal_notification() { - bool dataError = _mode == "next-error" || _mode.StartsWith("error-callback-", StringComparison.Ordinal); + bool unrelatedCancellation = _mode.StartsWith("next-unrelated-", StringComparison.Ordinal); + bool dataError = _mode == "next-error" || unrelatedCancellation || _mode.StartsWith("error-callback-", StringComparison.Ordinal); _completed.Should().Be(dataError ? 0 : 1); _errors.Should().Be(dataError ? 1 : 0); - _reportedError.Should().BeSameAs(dataError ? _failure : null); + _reportedError.Should().BeSameAs(unrelatedCancellation ? _unrelatedCancellation : dataError ? _failure : null); + if (unrelatedCancellation) + { + _reportedError.Should().BeOfType(_mode == "next-unrelated-task-canceled" ? typeof(TaskCanceledException) : typeof(OperationCanceledException)); + ((OperationCanceledException)_reportedError!).CancellationToken.Should().NotBe(Cancellation.Token); + Cancellation.IsCancellationRequested.Should().BeFalse(); + } } [Fact] From 46f14ab832a2f4362a5cc5a8b1bcc8108ec3ec3e Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Thu, 29 Oct 2026 11:29:16 +0100 Subject: [PATCH 5/9] fix: classify observer cancellation by request token --- Changelog.md | 5 +++-- .../MongoPersistenceEngine.Async.cs | 5 +++-- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/Changelog.md b/Changelog.md index 88dc51d..0a3ead2 100644 --- a/Changelog.md +++ b/Changelog.md @@ -10,13 +10,14 @@ - Fix initialization races: serialize initialization with a lock and retain success only after all setup completes. A waiting or later caller retries after failure; successful calls remain idempotent. Same-thread recursive initialization fails explicitly. Use `System.Threading.Lock` on .NET 9+ and an object lock on older targets. [#85](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/85) -- Propagate cancellation from async operations, including snapshot writes. Observer reads still signal cancellation through `OnCompletedAsync` with the original token; terminal callbacks are now awaited, and callback outcomes reach the returned task. [#86](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/86) + +- Propagate cancellation from async operations, including snapshot writes. Observer reads signal request cancellation through `OnCompletedAsync` with the original token; `OperationCanceledException` while the request token is active is reported through `OnErrorAsync`. Terminal callbacks are awaited, and callback outcomes reach the returned task. [#86](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/86) ### Breaking Changes - 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) -- Non-observer async callers must now handle `OperationCanceledException` (including `TaskCanceledException`) from operations that previously swallowed cancellation or returned success/default values; for example, `AddSnapshotAsync` now propagates cancellation instead of returning `false`. Observer reads continue to invoke `OnCompletedAsync` with the original token on cancellation, but cancellation or failure from that callback now propagates after it is awaited, without a second terminal notification. [#86](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/86) +- Non-observer async callers must now handle `OperationCanceledException` (including `TaskCanceledException`) from operations that previously swallowed cancellation or returned success/default values; for example, `AddSnapshotAsync` now propagates cancellation instead of returning `false`. Observer reads invoke `OnCompletedAsync` with the original token only when the request token is canceled; cancellation exceptions with an active request token go to `OnErrorAsync`. Cancellation or failure from a terminal callback propagates after it is awaited, without a second terminal notification. [#86](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/86) ### Migration notes diff --git a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs index e210688..0110e25 100644 --- a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs +++ b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs @@ -564,7 +564,8 @@ await TryMongoAsync( /// /// Executes the callback within a try/catch block to handle exceptions. /// If an observer is provided, it will be notified of any exceptions that occur during the operation. - /// If an observer is provided and the operation was cancelled, its OnCompletedAsync callback will be awaited. + /// If the supplied cancellation token is canceled, the observer's OnCompletedAsync callback will be awaited; + /// otherwise exceptions are delivered to OnErrorAsync. /// If no observer is provided, exceptions (including cancellation) will be raised. /// Exceptions from terminal observer callbacks propagate to the caller without another notification. /// @@ -598,7 +599,7 @@ protected virtual async Task TryMongoAsync(Func callbackAsync, Cancella MongoPersistenceEngineLogMessages.StorageThrewException(Logger, e); exception = ExceptionDispatchInfo.Capture(new StorageException(e.Message, e)); } - catch (OperationCanceledException ex) + catch (OperationCanceledException ex) when (cancellationToken.IsCancellationRequested) { MongoPersistenceEngineLogMessages.TaskWasCancelled(Logger, ex); if (asyncObserver == null) From 974bdd689662ea66cc340e3ce5b075fe59c29c62 Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Mon, 2 Nov 2026 22:29:27 +0100 Subject: [PATCH 6/9] test: preserve virtual guard dispatch and cancellation classification --- .../Issues/Issue86CancellationTests.Async.cs | 28 ++++++++++++++++--- 1 file changed, 24 insertions(+), 4 deletions(-) diff --git a/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue86CancellationTests.Async.cs b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue86CancellationTests.Async.cs index 216e0e7..85400c8 100644 --- a/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue86CancellationTests.Async.cs +++ b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue86CancellationTests.Async.cs @@ -68,8 +68,14 @@ public Task SignalDuplicateIdAsync(long id, CancellationToken cancellationToken protected class CancellationEngine : MongoPersistenceEngine { + public int WrapperCalls { get; private set; } public CancellationEngine(IMongoDatabase database, IDocumentSerializer serializer, MongoPersistenceOptions options) : base(database, serializer, options) { } + protected override Task TryMongoAsync(Func callbackAsync, CancellationToken cancellationToken, IAsyncObserver? asyncObserver = null) + { + WrapperCalls++; + return base.TryMongoAsync(callbackAsync, cancellationToken, asyncObserver); + } public Task Execute(Func callback, CancellationToken token, IAsyncObserver? observer = null) => TryMongoAsync(callback, token, observer); public Task ExecuteValue(Func> callback) => TryMongoAsync(callback); @@ -204,11 +210,16 @@ public void Should_preserve_the_canceled_task_and_token() public void Should_exercise_cancellation_at_the_requested_boundary() => _wasPending.Should().Be(_inFlight); } + [TestFixture(false)] + [TestFixture(true)] public class when_Issue_86_a_derived_engine_completes_its_own_callback : Issue_86_concern { + private readonly bool _completes; private int _completed; private IAsyncObserver _observer = null!; + public when_Issue_86_a_derived_engine_completes_its_own_callback(bool completes) => _completes = completes; + protected override void Context() { base.Context(); @@ -219,10 +230,10 @@ protected override void Context() }); } - protected override Task BecauseAsync() => Engine.Execute(() => _observer.OnCompletedAsync(Cancellation.Token), Cancellation.Token, _observer); + protected override Task BecauseAsync() => Engine.Execute(() => _completes ? _observer.OnCompletedAsync(Cancellation.Token) : Task.CompletedTask, Cancellation.Token, _observer); [Fact] - public void Should_not_add_completion_to_the_protected_wrapper_success_path() => _completed.Should().Be(1); + public void Should_not_add_completion_to_the_protected_wrapper_success_path() => _completed.Should().Be(_completes ? 1 : 0); } [TestFixture("snapshot-read")] @@ -304,6 +315,8 @@ public void Should_leave_persisted_data_unchanged() [TestFixture("stream", "next-operation-canceled")] [TestFixture("stream", "next-unrelated-task-canceled")] [TestFixture("stream", "next-unrelated-operation-canceled")] + [TestFixture("stream", "next-tokenless-operation-canceled")] + [TestFixture("stream", "next-canceled-request-unrelated-operation-canceled")] [TestFixture("stream", "next-error")] [TestFixture("stream", "completion-task-canceled")] [TestFixture("stream", "completion-operation-canceled")] @@ -357,6 +370,12 @@ private async Task OnNext(CancellationToken token) await _nextRelease.Task.ConfigureAwait(false); if (_mode == "next-task-canceled") return await Task.FromCanceled(token).ConfigureAwait(false); if (_mode == "next-operation-canceled") throw new OperationCanceledException(token); + if (_mode == "next-tokenless-operation-canceled") + { + _unrelatedCancellation = new OperationCanceledException(); + throw _unrelatedCancellation; + } + if (_mode == "next-canceled-request-unrelated-operation-canceled") throw new OperationCanceledException(); if (_mode == "next-unrelated-task-canceled" || _mode == "next-unrelated-operation-canceled") { using var unrelatedCancellation = new CancellationTokenSource(); @@ -420,7 +439,7 @@ protected override async Task BecauseAsync() { await WaitFor(_nextEntered.Task).ConfigureAwait(false); _pendingAtNext = !Operation.IsCompleted; - if (_mode != "next-error" && !_mode.StartsWith("next-unrelated-", StringComparison.Ordinal)) Cancellation.Cancel(); + if (_mode != "next-error" && _mode != "next-tokenless-operation-canceled" && !_mode.StartsWith("next-unrelated-", StringComparison.Ordinal)) Cancellation.Cancel(); _nextRelease.SetResult(true); } await WaitFor(_terminalEntered.Task).ConfigureAwait(false); @@ -440,7 +459,8 @@ public void Should_await_the_terminal_callback_and_preserve_the_original_token() [Fact] public void Should_deliver_exactly_one_terminal_notification() { - bool unrelatedCancellation = _mode.StartsWith("next-unrelated-", StringComparison.Ordinal); + Engine.WrapperCalls.Should().Be(1, "observer reads must still dispatch through the protected virtual wrapper"); + bool unrelatedCancellation = _mode.StartsWith("next-unrelated-", StringComparison.Ordinal) || _mode == "next-tokenless-operation-canceled"; bool dataError = _mode == "next-error" || unrelatedCancellation || _mode.StartsWith("error-callback-", StringComparison.Ordinal); _completed.Should().Be(dataError ? 0 : 1); _errors.Should().Be(dataError ? 1 : 0); From 15418b7cfc161cc59138788d670a76379e962d1b Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Mon, 2 Nov 2026 22:29:49 +0100 Subject: [PATCH 7/9] refactor: keep observer completion within TryMongoAsync overloads --- Changelog.md | 3 +- .../MongoPersistenceEngine.Async.cs | 41 ++++++++++--------- 2 files changed, 22 insertions(+), 22 deletions(-) diff --git a/Changelog.md b/Changelog.md index 0a3ead2..a1215d6 100644 --- a/Changelog.md +++ b/Changelog.md @@ -10,8 +10,7 @@ - Fix initialization races: serialize initialization with a lock and retain success only after all setup completes. A waiting or later caller retries after failure; successful calls remain idempotent. Same-thread recursive initialization fails explicitly. Use `System.Threading.Lock` on .NET 9+ and an object lock on older targets. [#85](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/85) - -- Propagate cancellation from async operations, including snapshot writes. Observer reads signal request cancellation through `OnCompletedAsync` with the original token; `OperationCanceledException` while the request token is active is reported through `OnErrorAsync`. Terminal callbacks are awaited, and callback outcomes reach the returned task. [#86](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/86) +- Propagate cancellation from async operations, including snapshot writes. Observer reads signal request cancellation through `OnCompletedAsync` with the original token; `OperationCanceledException` while the request token is active is reported through `OnErrorAsync`. Observer reads use a completion-aware `TryMongoAsync` overload, retaining the protected virtual guard and its existing success behavior for derived engines. Terminal callbacks are awaited, and callback outcomes reach the returned task. [#86](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/86) ### Breaking Changes diff --git a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs index 0110e25..e2ebba5 100644 --- a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs +++ b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs @@ -15,7 +15,7 @@ public Task GetFromAsync(string bucketId, string streamId, int minRevision, int { MongoPersistenceEngineLogMessages.GettingAllCommitsBetween(Logger, streamId, bucketId, minRevision, maxRevision); - return ObserveMongoAsync(async () => + return TryMongoAsync(async () => { var filters = new List>() { @@ -55,7 +55,7 @@ public Task GetFromAsync(string bucketId, string streamId, int minRevision, int } } }, - cancellationToken, observer); + observer, cancellationToken); } /// @@ -63,7 +63,7 @@ public Task GetFromAsync(string bucketId, long checkpointToken, IAsyncObserver + return TryMongoAsync(async () => { var query = Builders.Filter.Eq(MongoCommitFields.BucketId, bucketId); if (checkpointToken > 0) @@ -90,8 +90,8 @@ public Task GetFromAsync(string bucketId, long checkpointToken, IAsyncObserver @@ -99,7 +99,7 @@ public Task GetFromToAsync(string bucketId, long fromCheckpointToken, long toChe { MongoPersistenceEngineLogMessages.GettingCommitsFromBucketAndFromToCheckpoint(Logger, bucketId, fromCheckpointToken, toCheckpointToken); - return ObserveMongoAsync(async () => + return TryMongoAsync(async () => { var filters = new List>() { @@ -135,8 +135,8 @@ public Task GetFromToAsync(string bucketId, long fromCheckpointToken, long toChe } } }, - cancellationToken, - asyncObserver); + asyncObserver, + cancellationToken); } /// @@ -144,7 +144,7 @@ public Task GetFromAsync(long checkpointToken, IAsyncObserver asyncObse { MongoPersistenceEngineLogMessages.GettingAllCommitsFromCheckpoint(Logger, checkpointToken); - return ObserveMongoAsync(async () => + return TryMongoAsync(async () => { var query = Builders.Filter.Ne(MongoCommitFields.BucketId, MongoSystemBuckets.RecycleBin); if (checkpointToken > 0) @@ -171,8 +171,8 @@ public Task GetFromAsync(long checkpointToken, IAsyncObserver asyncObse } } }, - cancellationToken, - asyncObserver); + asyncObserver, + cancellationToken); } /// @@ -180,7 +180,7 @@ public Task GetFromToAsync(long fromCheckpointToken, long toCheckpointToken, IAs { MongoPersistenceEngineLogMessages.GettingCommitsFromToCheckpoint(Logger, fromCheckpointToken, toCheckpointToken); - return ObserveMongoAsync(async () => + return TryMongoAsync(async () => { var filters = new List>() { @@ -217,8 +217,8 @@ public Task GetFromToAsync(long fromCheckpointToken, long toCheckpointToken, IAs } } }, - cancellationToken, - asyncObserver); + asyncObserver, + cancellationToken); } /// @@ -338,7 +338,7 @@ public Task GetStreamsToSnapshotAsync(string bucketId, int maxThreshold, IAsyncO MongoPersistenceEngineLogMessages.GettingStreamsToSnapshot(Logger); - return ObserveMongoAsync(async () => + return TryMongoAsync(async () => { var query = Builders.Filter.And( Builders.Filter.Eq(MongoStreamHeadFields.FullQualifiedBucketId, bucketId), @@ -362,7 +362,7 @@ public Task GetStreamsToSnapshotAsync(string bucketId, int maxThreshold, IAsyncO } } }, - cancellationToken, asyncObserver); + asyncObserver, cancellationToken); } /// @@ -626,7 +626,8 @@ protected virtual async Task TryMongoAsync(Func callbackAsync, Cancella } } - private async Task ObserveMongoAsync(Func callbackAsync, CancellationToken cancellationToken, IAsyncObserver asyncObserver) + // Observer reads own successful completion; keep the virtual guard's success behavior for derived engines. + private async Task TryMongoAsync(Func callbackAsync, IAsyncObserver asyncObserver, CancellationToken cancellationToken) { bool completed = false; await TryMongoAsync(async () => @@ -682,7 +683,7 @@ await TryMongoAsync(() => /// public Task GetDeletedCommitsAsync(IAsyncObserver asyncObserver, CancellationToken cancellationToken) { - return ObserveMongoAsync(async () => + return TryMongoAsync(async () => { using var cursor = await PersistedCommits .Find(Builders.Filter.Eq(MongoCommitFields.BucketId, MongoSystemBuckets.RecycleBin)) @@ -701,8 +702,8 @@ public Task GetDeletedCommitsAsync(IAsyncObserver asyncObserver, Cancel } } }, - cancellationToken, - asyncObserver); + asyncObserver, + cancellationToken); } } } From 0ab329e23218a7b8b1bf8ee42bde8cadf5cad1c1 Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Mon, 2 Nov 2026 22:43:23 +0100 Subject: [PATCH 8/9] refactor: restore ObserveMongoAsync name and explain terminal isolation --- Changelog.md | 2 +- .../MongoPersistenceEngine.Async.cs | 27 ++++++++++--------- 2 files changed, 15 insertions(+), 14 deletions(-) diff --git a/Changelog.md b/Changelog.md index a1215d6..ea5816b 100644 --- a/Changelog.md +++ b/Changelog.md @@ -10,7 +10,7 @@ - Fix initialization races: serialize initialization with a lock and retain success only after all setup completes. A waiting or later caller retries after failure; successful calls remain idempotent. Same-thread recursive initialization fails explicitly. Use `System.Threading.Lock` on .NET 9+ and an object lock on older targets. [#85](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/85) -- Propagate cancellation from async operations, including snapshot writes. Observer reads signal request cancellation through `OnCompletedAsync` with the original token; `OperationCanceledException` while the request token is active is reported through `OnErrorAsync`. Observer reads use a completion-aware `TryMongoAsync` overload, retaining the protected virtual guard and its existing success behavior for derived engines. Terminal callbacks are awaited, and callback outcomes reach the returned task. [#86](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/86) +- Propagate cancellation from async operations, including snapshot writes. Observer reads signal request cancellation through `OnCompletedAsync` with the original token; `OperationCanceledException` while the request token is active is reported through `OnErrorAsync`. Observer reads use `ObserveMongoAsync` to complete successfully outside the guarded operation, retaining the protected virtual `TryMongoAsync` guard and its existing success behavior for derived engines. Terminal callbacks are awaited, and callback outcomes reach the returned task. [#86](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/86) ### Breaking Changes diff --git a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs index e2ebba5..28b85b8 100644 --- a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs +++ b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs @@ -15,7 +15,7 @@ public Task GetFromAsync(string bucketId, string streamId, int minRevision, int { MongoPersistenceEngineLogMessages.GettingAllCommitsBetween(Logger, streamId, bucketId, minRevision, maxRevision); - return TryMongoAsync(async () => + return ObserveMongoAsync(async () => { var filters = new List>() { @@ -63,7 +63,7 @@ public Task GetFromAsync(string bucketId, long checkpointToken, IAsyncObserver + return ObserveMongoAsync(async () => { var query = Builders.Filter.Eq(MongoCommitFields.BucketId, bucketId); if (checkpointToken > 0) @@ -99,7 +99,7 @@ public Task GetFromToAsync(string bucketId, long fromCheckpointToken, long toChe { MongoPersistenceEngineLogMessages.GettingCommitsFromBucketAndFromToCheckpoint(Logger, bucketId, fromCheckpointToken, toCheckpointToken); - return TryMongoAsync(async () => + return ObserveMongoAsync(async () => { var filters = new List>() { @@ -144,7 +144,7 @@ public Task GetFromAsync(long checkpointToken, IAsyncObserver asyncObse { MongoPersistenceEngineLogMessages.GettingAllCommitsFromCheckpoint(Logger, checkpointToken); - return TryMongoAsync(async () => + return ObserveMongoAsync(async () => { var query = Builders.Filter.Ne(MongoCommitFields.BucketId, MongoSystemBuckets.RecycleBin); if (checkpointToken > 0) @@ -180,7 +180,7 @@ public Task GetFromToAsync(long fromCheckpointToken, long toCheckpointToken, IAs { MongoPersistenceEngineLogMessages.GettingCommitsFromToCheckpoint(Logger, fromCheckpointToken, toCheckpointToken); - return TryMongoAsync(async () => + return ObserveMongoAsync(async () => { var filters = new List>() { @@ -338,7 +338,7 @@ public Task GetStreamsToSnapshotAsync(string bucketId, int maxThreshold, IAsyncO MongoPersistenceEngineLogMessages.GettingStreamsToSnapshot(Logger); - return TryMongoAsync(async () => + return ObserveMongoAsync(async () => { var query = Builders.Filter.And( Builders.Filter.Eq(MongoStreamHeadFields.FullQualifiedBucketId, bucketId), @@ -564,8 +564,8 @@ await TryMongoAsync( /// /// Executes the callback within a try/catch block to handle exceptions. /// If an observer is provided, it will be notified of any exceptions that occur during the operation. - /// If the supplied cancellation token is canceled, the observer's OnCompletedAsync callback will be awaited; - /// otherwise exceptions are delivered to OnErrorAsync. + /// OperationCanceledException completes the observer through an awaited OnCompletedAsync only when the request token is canceled. + /// Cancellation exceptions while the request token is active, and other operation errors, are delivered to OnErrorAsync. /// If no observer is provided, exceptions (including cancellation) will be raised. /// Exceptions from terminal observer callbacks propagate to the caller without another notification. /// @@ -626,8 +626,10 @@ protected virtual async Task TryMongoAsync(Func callbackAsync, Cancella } } - // Observer reads own successful completion; keep the virtual guard's success behavior for derived engines. - private async Task TryMongoAsync(Func callbackAsync, IAsyncObserver asyncObserver, CancellationToken cancellationToken) + // 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. + private async Task ObserveMongoAsync(Func callbackAsync, IAsyncObserver asyncObserver, CancellationToken cancellationToken) { bool completed = false; await TryMongoAsync(async () => @@ -636,7 +638,6 @@ await TryMongoAsync(async () => completed = true; }, cancellationToken, asyncObserver).ConfigureAwait(false); - // Notify outside the operation's catch block: a terminal callback must not be retried or sent to OnErrorAsync. if (completed) { await asyncObserver.OnCompletedAsync(cancellationToken).ConfigureAwait(false); @@ -683,7 +684,7 @@ await TryMongoAsync(() => /// public Task GetDeletedCommitsAsync(IAsyncObserver asyncObserver, CancellationToken cancellationToken) { - return TryMongoAsync(async () => + return ObserveMongoAsync(async () => { using var cursor = await PersistedCommits .Find(Builders.Filter.Eq(MongoCommitFields.BucketId, MongoSystemBuckets.RecycleBin)) @@ -709,4 +710,4 @@ public Task GetDeletedCommitsAsync(IAsyncObserver asyncObserver, Cancel } #pragma warning restore CA2254 // Template should be a static expression -#pragma warning restore IDE0079 // Remove unnecessary suppression \ No newline at end of file +#pragma warning restore IDE0079 // Remove unnecessary suppression From f62f4a51b7dcc3468618e53b00b1c5ae7084a250 Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Mon, 2 Nov 2026 23:03:21 +0100 Subject: [PATCH 9/9] docs: explain TaskCanceledException coverage in cancellation catch --- .../MongoPersistenceEngine.Async.cs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs index 28b85b8..eb0be44 100644 --- a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs +++ b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs @@ -601,6 +601,8 @@ protected virtual async Task TryMongoAsync(Func callbackAsync, Cancella } catch (OperationCanceledException ex) when (cancellationToken.IsCancellationRequested) { + // TaskCanceledException derives from OperationCanceledException, so no separate catch is needed. + // This filter handles both only when the request token is canceled; otherwise the general catch below handles them. MongoPersistenceEngineLogMessages.TaskWasCancelled(Logger, ex); if (asyncObserver == null) {