Skip to content

fix: align async read disposal with synchronous reads #115

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
AGiorgetti wants to merge 1 commit into
base: develop
Choose a base branch
from
Open

fix: align async read disposal with synchronous reads #115

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions Changelog.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,8 @@

### Breaking Changes

- Active async observer reads now check engine disposal before every cursor batch advancement, matching synchronous enumeration. After disposal is observed, the owned cursor is disposed and `ObjectDisposedException` is sent once to `OnErrorAsync`, without completion notification; a successful error handler leaves the returned task successful, while a failing handler propagates its exception. Already-disposed invocations still fault directly without observer notification. In-flight acquisition/advancement and callbacks are not canceled, and documents in an already fetched batch can still be delivered. Early observer stop still completes normally. [#109](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/109)

- `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)
Expand All @@ -42,6 +44,8 @@

### Migration notes

- Keep the engine alive until all synchronous enumerations and async observer reads finish. For shutdown, cancel and await active async reads before disposing the shared engine; disposal is a batch-boundary guard, not immediate cancellation or atomic exclusion of concurrent reads. No data migration is required. [#109](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/109)

- `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
Expand Down
151 changes: 151 additions & 0 deletions src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue109AsyncDisposal.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,151 @@
using FakeItEasy;
using FluentAssertions;
using MongoDB.Bson;
using MongoDB.Driver;
using NEventStore.Persistence.AcceptanceTests;
using NEventStore.Persistence.AcceptanceTests.BDD;
using NEventStore.Serialization;
#if NUNIT
using NUnit.Framework;
namespace NEventStore.Persistence.MongoDB.Tests.AcceptanceTests.Issues.Async
{
[TestFixtureSource(nameof(Cases))]
public class when_Issue109_disposal_crosses_an_async_read_boundary(int path, string stage) : SpecificationBase
{
public static IEnumerable<object[]> Cases()
{
for (int path = 0; path < 7; path++)
foreach (var stage in new[] { "before", "acquisition", "advance", "observer", "stop", "normal", "error handler" })
yield return new object[] { path, stage };
}
private readonly IAsyncCursor<BsonDocument> _cursor = A.Fake<IAsyncCursor<BsonDocument>>();
private readonly TaskCompletionSource<bool> _entered = new(TaskCreationOptions.RunContinuationsAsynchronously);
private readonly TaskCompletionSource<bool> _resume = new(TaskCreationOptions.RunContinuationsAsynchronously);
private MongoPersistenceEngine _engine = null!;
private Exception? _reported, _escaped;
private readonly Exception _terminalFailure = new InvalidOperationException("Terminal callback failed.");
private int _finds, _moves, _next, _errors, _completed;
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);
A.CallTo(() => collection.FindAsync(A<FilterDefinition<BsonDocument>>._,
A<FindOptions<BsonDocument, BsonDocument>>._, CancellationToken.None)).ReturnsLazily(async () =>
{
_finds++;
if (stage == "acquisition") await Pause();
return _cursor;
});
A.CallTo(() => _cursor.MoveNextAsync(CancellationToken.None)).ReturnsLazily(async () =>
{
_moves++;
if (_moves == 1 && stage == "advance") await Pause();
return _moves <= 2;
});
// Two documents prove disposal is checked at batch boundaries, not per document.
A.CallTo(() => _cursor.Current).Returns(new[] { Document(path == 5 ? 7 : 0, 1, false), Document(path == 5 ? 7 : 0, 2, false) });
_engine = new MongoPersistenceEngine(database, A.Fake<IDocumentSerializer>(), new MongoPersistenceOptions());
}
private async Task Pause()
{
_entered.TrySetResult(true);
await _resume.Task.ConfigureAwait(false);
}
private IAsyncObserver<T> Observer<T>()
{
var observer = A.Fake<IAsyncObserver<T>>();
A.CallTo(() => observer.OnNextAsync(A<T>._, CancellationToken.None)).ReturnsLazily(async () =>
{
_next++;
if (_next == 1 && (stage == "observer" || stage == "stop" || stage == "error handler")) await Pause();
return stage != "stop";
});
A.CallTo(() => observer.OnErrorAsync(A<Exception>._, CancellationToken.None))
.ReturnsLazily((Exception error, CancellationToken _) => { _reported = error; _errors++; return stage == "error handler" ? Task.FromException(_terminalFailure) : Task.CompletedTask; });
A.CallTo(() => observer.OnCompletedAsync(CancellationToken.None))
.ReturnsLazily(() => { _completed++; return Task.CompletedTask; });
return observer;
}
protected override async Task BecauseAsync()
{
if (stage == "before") _engine.Dispose();
var observer = Observer<ICommit>();
var operation = Catch.ExceptionAsync(() => path switch
{
0 => _engine.GetFromAsync("bucket", "stream", 0, int.MaxValue, observer, CancellationToken.None),
1 => _engine.GetFromAsync("bucket", 0L, observer, CancellationToken.None),
2 => _engine.GetFromToAsync("bucket", 0L, long.MaxValue, observer, CancellationToken.None),
3 => _engine.GetFromAsync(0L, observer, CancellationToken.None),
4 => _engine.GetFromToAsync(0L, long.MaxValue, observer, CancellationToken.None),
5 => _engine.GetStreamsToSnapshotAsync("bucket", 0, Observer<IStreamHead>(), CancellationToken.None),
_ => _engine.GetDeletedCommitsAsync(observer, CancellationToken.None)
});
if (stage != "before" && stage != "normal")
{
var reached = await Task.WhenAny(_entered.Task, operation, Task.Delay(TimeSpan.FromSeconds(10)));
reached.Should().BeSameAs(_entered.Task);
operation.IsCompleted.Should().BeFalse();
_engine.Dispose();
_resume.TrySetResult(true);
}
_escaped = await operation.ConfigureAwait(false);
}
[Fact]
public void Should_preserve_terminal_notification_contract()
{
if (stage == "before") _escaped.Should().BeOfType<ObjectDisposedException>();
else if (stage == "error handler") _escaped.Should().BeSameAs(_terminalFailure);
else _escaped.Should().BeNull();
bool error = stage == "acquisition" || stage == "advance" || stage == "observer" || stage == "error handler";
if (error) _reported.Should().BeOfType<ObjectDisposedException>();
else _reported.Should().BeNull();
_errors.Should().Be(error ? 1 : 0);
_completed.Should().Be(stage == "normal" || stage == "stop" ? 1 : 0);
}
[Fact]
public void Should_stop_at_the_next_batch_boundary_and_release_the_owned_cursor()
{
_finds.Should().Be(stage == "before" ? 0 : 1);
_moves.Should().Be(stage == "before" || stage == "acquisition" ? 0 : stage == "normal" ? 3 : 1);
_next.Should().Be(stage == "before" || stage == "acquisition" ? 0 : stage == "normal" ? 4 : stage == "stop" ? 1 : 2);
if (stage == "before") A.CallTo(() => _cursor.Dispose()).MustNotHaveHappened();
else A.CallTo(() => _cursor.Dispose()).MustHaveHappenedOnceExactly();
}
protected override void Cleanup() { _resume.TrySetResult(true); _engine.Dispose(); }
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
25 changes: 18 additions & 7 deletions src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ public Task GetFromAsync(string bucketId, string streamId, int minRevision, int
.ToCursorAsync(cancellationToken)
.ConfigureAwait(false);

while (await cursor.MoveNextAsync(cancellationToken).ConfigureAwait(false))
while (await MoveNextWhileActiveAsync(cursor, cancellationToken).ConfigureAwait(false))
{
foreach (var doc in cursor.Current)
{
Expand Down Expand Up @@ -78,7 +78,7 @@ public Task GetFromAsync(string bucketId, long checkpointToken, IAsyncObserver<I
.Sort(SortByAscendingCheckpointNumber)
.ToCursorAsync(cancellationToken)
.ConfigureAwait(false);
while (await cursor.MoveNextAsync(cancellationToken).ConfigureAwait(false))
while (await MoveNextWhileActiveAsync(cursor, cancellationToken).ConfigureAwait(false))
{
foreach (var doc in cursor.Current)
{
Expand Down Expand Up @@ -123,7 +123,7 @@ public Task GetFromToAsync(string bucketId, long fromCheckpointToken, long toChe
.Sort(SortByAscendingCheckpointNumber)
.ToCursorAsync(cancellationToken)
.ConfigureAwait(false);
while (await cursor.MoveNextAsync(cancellationToken).ConfigureAwait(false))
while (await MoveNextWhileActiveAsync(cursor, cancellationToken).ConfigureAwait(false))
{
foreach (var doc in cursor.Current)
{
Expand Down Expand Up @@ -159,7 +159,7 @@ public Task GetFromAsync(long checkpointToken, IAsyncObserver<ICommit> asyncObse
.Sort(SortByAscendingCheckpointNumber)
.ToCursorAsync(cancellationToken)
.ConfigureAwait(false);
while (await cursor.MoveNextAsync(cancellationToken).ConfigureAwait(false))
while (await MoveNextWhileActiveAsync(cursor, cancellationToken).ConfigureAwait(false))
{
foreach (var doc in cursor.Current)
{
Expand Down Expand Up @@ -205,7 +205,7 @@ public Task GetFromToAsync(long fromCheckpointToken, long toCheckpointToken, IAs
.Sort(SortByAscendingCheckpointNumber)
.ToCursorAsync(cancellationToken)
.ConfigureAwait(false);
while (await cursor.MoveNextAsync(cancellationToken).ConfigureAwait(false))
while (await MoveNextWhileActiveAsync(cursor, cancellationToken).ConfigureAwait(false))
{
foreach (var doc in cursor.Current)
{
Expand Down Expand Up @@ -350,7 +350,7 @@ public Task GetStreamsToSnapshotAsync(string bucketId, int maxThreshold, IAsyncO
.ToCursorAsync(cancellationToken)
.ConfigureAwait(false);

while (await cursor.MoveNextAsync(cancellationToken).ConfigureAwait(false))
while (await MoveNextWhileActiveAsync(cursor, cancellationToken).ConfigureAwait(false))
{
foreach (var doc in cursor.Current)
{
Expand Down Expand Up @@ -623,6 +623,17 @@ protected virtual async Task TryMongoAsync<T>(Func<Task> callbackAsync, Cancella
}
}

// Match synchronous enumeration: check before advancing, not within a fetched batch.
// Disposal does not cancel a driver operation or observer callback already in flight.
private Task<bool> MoveNextWhileActiveAsync(IAsyncCursor<BsonDocument> cursor, CancellationToken cancellationToken)
{
if (IsDisposed)
{
throw new ObjectDisposedException("Attempt to use storage after it has been disposed.");
}
return cursor.MoveNextAsync(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.
Expand Down Expand Up @@ -689,7 +700,7 @@ public Task GetDeletedCommitsAsync(IAsyncObserver<ICommit> asyncObserver, Cancel
.Sort(SortByAscendingCheckpointNumber)
.ToCursorAsync(cancellationToken)
.ConfigureAwait(false);
while (await cursor.MoveNextAsync(cancellationToken).ConfigureAwait(false))
while (await MoveNextWhileActiveAsync(cursor, cancellationToken).ConfigureAwait(false))
{
foreach (var doc in cursor.Current)
{
Expand Down
Loading