diff --git a/Changelog.md b/Changelog.md index db10241..ded8ffd 100644 --- a/Changelog.md +++ b/Changelog.md @@ -2,6 +2,8 @@ ## Unreleased +- Atomically advance snapshot metadata without losing concurrent or delayed commit increments. Older snapshots cannot regress the recorded snapshot revision. Preserve configured write concerns; legacy heads without an unsnapshotted counter use the current head-minus-snapshot value as a fallback. Existing incorrect counters are not repaired. [#90](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/90) + - Keep stream-head revisions monotonic when synchronous or asynchronous commit metadata writes complete out of order, while retaining every write's unsnapshotted event increment and existing snapshot revision. No API/schema breaking changes or migration are required; existing regressed metadata is not backfilled. Snapshot-update races (#90) and deletion/recreation races (#96) remain separate. See [stream-head invariants](docs/Stream-Heads.md). [#89](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/89) - Fix async checkpoint reads to acquire and dispose cursors through `FindAsync`, preserving cancellation tokens, checkpoint values, duplicate-ID recovery, and recycle-bin cleanup boundaries. No public API or data-format changes; no migration is required. [#88](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/88) @@ -18,6 +20,8 @@ ### Breaking Changes +- **MongoDB 4.2 or later is now required** for aggregation-based snapshot metadata updates. Upgrade MongoDB before upgrading this adapter. No stored-schema migration is needed; MongoDB 4.2 is the feature minimum, not a recommended production release. This change does not repair historical metadata or solve stream lifecycle races. [#90](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/90) + - New recycled records use the synthetic stored `StreamId`. Older adapter binaries expose that value and cannot find these records by the original stream ID; raw readers that assume stored `StreamId` is original see the same change. Upgrade recycle-bin readers before deleting streams with the new format. Existing recycled records need no backfill. [#87](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/87) - 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) diff --git a/README.md b/README.md index e613b86..a333822 100644 --- a/README.md +++ b/README.md @@ -9,6 +9,20 @@ NEventStore.Persistence.MongoDB currently supports: - .net standard 2.1 - .net framework 4.7.2 +MongoDB server requirement +=== + +MongoDB **4.2 or later** is required. Snapshot metadata updates use aggregation update +pipelines introduced in 4.2. This is a breaking server requirement: upgrade MongoDB +before upgrading this adapter. Version 4.2 is the minimum feature level, not a +recommended production release; use a supported MongoDB release. + +No stored-schema migration is required. The fix preserves configured write concerns +and prevents new snapshot counter races; it does not repair previously incorrect +counters or provide transactional snapshot/commit or stream-lifecycle guarantees. +For legacy missing/null counters, the head-minus-snapshot fallback cannot reconstruct +commit increments that have not yet reached the head. + Build Status === diff --git a/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue90SnapshotMetadata.cs b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue90SnapshotMetadata.cs new file mode 100644 index 0000000..f7b37be --- /dev/null +++ b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue90SnapshotMetadata.cs @@ -0,0 +1,305 @@ +using FluentAssertions; +using MongoDB.Bson; +using MongoDB.Driver; +using MongoDB.Driver.Core.Events; +using NEventStore.Persistence.AcceptanceTests; +using NEventStore.Persistence.AcceptanceTests.BDD; +using NEventStore.Serialization; +#if NUNIT +using NUnit.Framework; + +namespace NEventStore.Persistence.MongoDB.Tests.AcceptanceTests.Issues +{ + [TestFixture(false)] + [TestFixture(true)] + public class Issue90SnapshotAccountingContracts + { + private readonly bool _async; + private IMongoDatabase _database = null!; + private MongoPersistenceEngine _engine = null!; + private readonly List _metadataCommands = new(); + private const string Bucket = "accounting"; + private const string Stream = "delayed-metadata"; + private IMongoCollection Heads => _database.GetCollection("Streams"); + + public Issue90SnapshotAccountingContracts(bool async) { _async = async; } + + [SetUp] + public void SetUp() + { + _ = new AcceptanceTestMongoPersistenceFactory(); + var url = new MongoUrl(AcceptanceTestMongoPersistenceFactory.GetConnectionString()); + _database = new MongoClient(url).GetDatabase($"issue90-contract-{Guid.NewGuid():N}"); + _engine = CreateEngine(WriteConcern.Acknowledged); + _engine.Initialize(); + _engine.CommitMany(2, Stream, Bucket).ToArray(); + _metadataCommands.Clear(); + } + + private MongoPersistenceEngine CreateEngine(WriteConcern concern) + { + var settings = MongoClientSettings.FromUrl(new MongoUrl(AcceptanceTestMongoPersistenceFactory.GetConnectionString())); + settings.ClusterConfigurator = cluster => cluster.Subscribe(command => + { + if (command.CommandName == "update" && command.Command["update"] == "Streams") + _metadataCommands.Add(command.Command.DeepClone().AsBsonDocument); + }); + return new MongoPersistenceEngine(new MongoClient(settings).GetDatabase(_database.DatabaseNamespace.DatabaseName), + new DocumentObjectSerializer(), new MetadataOptions(concern)); + } + + private Task Save(int revision, MongoPersistenceEngine? engine = null) + { + engine ??= _engine; + var snapshot = new Snapshot(Bucket, Stream, revision, "payload"); + return _async ? engine.AddSnapshotAsync(snapshot, CancellationToken.None) : Task.FromResult(engine.AddSnapshot(snapshot)); + } + + [Test] + public async Task Delayed_earlier_commit_metadata_is_not_counted_twice_after_a_snapshot() + { + // Two accepted two-event commits exist. Model revision 4's metadata arriving + // before revision 2's: only two event increments have reached the head. + Heads.UpdateOne(FilterDefinition.Empty, + Builders.Update.Set(MongoStreamHeadFields.Unsnapshotted, 2)); + (await Save(4)).Should().BeTrue(); + Heads.Find(FilterDefinition.Empty).Single()[MongoStreamHeadFields.Unsnapshotted] + .AsInt32.Should().Be(-2, "pending increments must not be clamped or reconstructed from HeadRevision"); + + // This is the delayed commit-side update from #89, not a second event commit. + Heads.UpdateOne(FilterDefinition.Empty, Builders.Update + .Max(MongoStreamHeadFields.HeadRevision, 2).Inc(MongoStreamHeadFields.Unsnapshotted, 2)); + var head = Heads.Find(FilterDefinition.Empty).Single(); + head[MongoStreamHeadFields.HeadRevision].AsInt32.Should().Be(4); + head[MongoStreamHeadFields.SnapshotRevision].AsInt32.Should().Be(4); + head[MongoStreamHeadFields.Unsnapshotted].AsInt32.Should().Be(0); + _engine.GetStreamsToSnapshot(Bucket, 1).Should().BeEmpty(); + } + + [TestCase(false)] + [TestCase(true)] + public async Task Legacy_missing_unsnapshotted_counter_is_initialized_without_a_null_result(bool nullValue) + { + (await Save(1)).Should().BeTrue(); + Heads.UpdateOne(FilterDefinition.Empty, + nullValue ? Builders.Update.Set(MongoStreamHeadFields.Unsnapshotted, BsonNull.Value) + : Builders.Update.Unset(MongoStreamHeadFields.Unsnapshotted)); + (await Save(2)).Should().BeTrue(); + var head = Heads.Find(FilterDefinition.Empty).Single(); + head[MongoStreamHeadFields.SnapshotRevision].AsInt32.Should().Be(2); + head[MongoStreamHeadFields.Unsnapshotted].AsInt32.Should().Be(2); + _engine.GetStreamsToSnapshot(Bucket, 2).Should().ContainSingle(); + _engine.GetStreamsToSnapshot(Bucket, 3).Should().BeEmpty(); + } + + [TestCase("head")] + [TestCase("revision")] + [TestCase("snapshot")] + public async Task Missing_head_or_required_revision_fields_still_return_false(string missing) + { + if (missing == "head") Heads.DeleteMany(FilterDefinition.Empty); + else Heads.UpdateOne(FilterDefinition.Empty, + Builders.Update.Unset(missing == "revision" + ? MongoStreamHeadFields.HeadRevision : MongoStreamHeadFields.SnapshotRevision)); + (await Save(2)).Should().BeFalse(); + _metadataCommands.Should().BeEmpty("validation must fail before a metadata update is sent"); + } + + [TestCase(false)] + [TestCase(true)] + public async Task Snapshot_metadata_retains_the_configured_write_concern(bool majority) + { + using var engine = CreateEngine(majority ? WriteConcern.WMajority : WriteConcern.Unacknowledged); + engine.Initialize(); + _metadataCommands.Clear(); + (await Save(2, engine)).Should().BeTrue(); + _metadataCommands.Should().ContainSingle(); + var concern = _metadataCommands.Single()["writeConcern"]["w"]; + if (majority) concern.AsString.Should().Be("majority"); + else concern.AsInt32.Should().Be(0); + } + + [Test] + public async Task Already_canceled_async_snapshot_does_not_mutate_metadata() + { + using var cancellation = new CancellationTokenSource(); + cancellation.Cancel(); + Func save = () => _engine.AddSnapshotAsync( + new Snapshot(Bucket, Stream, 2, "payload"), cancellation.Token); + await save.Should().ThrowAsync(); + _metadataCommands.Should().BeEmpty(); + Heads.Find(FilterDefinition.Empty).Single()[MongoStreamHeadFields.SnapshotRevision] + .AsInt32.Should().Be(0); + } + + [TearDown] + public void TearDown() + { + _engine?.Dispose(); + if (_database != null) _database.Client.DropDatabase(_database.DatabaseNamespace.DatabaseName); + } + + private sealed class MetadataOptions : MongoPersistenceOptions + { + private readonly WriteConcern _concern; + public MetadataOptions(WriteConcern concern) + { _concern = concern; PersistStreamHeadsOnBackgroundThread = false; } + public override MongoCollectionSettings GetStreamSettings() => new() { WriteConcern = _concern }; + public override MongoCollectionSettings GetSnapshotSettings() => new() { WriteConcern = WriteConcern.Acknowledged }; + } + } + + [TestFixture(false, 0)] + [TestFixture(true, 0)] + [TestFixture(false, 1)] + [TestFixture(true, 1)] + [TestFixture(false, 2)] + [TestFixture(true, 2)] + [TestFixture(false, 4)] + [TestFixture(true, 4)] + [TestFixture(false, 5)] + [TestFixture(true, 5)] + [TestFixture(false, 3)] + [TestFixture(true, 3)] + public class when_Issue_90_snapshot_metadata_overlaps_another_write : SpecificationBase + { + private readonly bool _async; + // Commit before metadata, newer snapshot, older snapshot, commit after metadata. + private readonly int _interleaving; + private readonly ManualResetEventSlim _entered = new(); + private readonly ManualResetEventSlim _release = new(); + private MongoPersistenceEngine _snapshotEngine = null!; + private MongoPersistenceEngine _otherEngine = null!; + private IMongoDatabase _database = null!; + private CommitAttempt _last = null!; + private Task? _snapshot; + private bool _saved; + private int _armed; + private const string Stream = "snapshot-race"; + private const string BucketId = "snapshot-bucket"; + private int ExpectedHead => _interleaving is 0 or 3 ? 6 : 4; + private int ExpectedSnapshot => _interleaving == 1 ? 4 : 2; + + public when_Issue_90_snapshot_metadata_overlaps_another_write(bool async, int interleaving) + { + _async = async; + _interleaving = interleaving; + } + + protected override void Context() + { + _ = new AcceptanceTestMongoPersistenceFactory(); + var url = new MongoUrl(AcceptanceTestMongoPersistenceFactory.GetConnectionString()); + var settings = MongoClientSettings.FromUrl(url); + settings.ClusterConfigurator = cluster => cluster.Subscribe(command => + { + if (command.CommandName != "update" || command.Command["update"] != "Streams") return; + var update = command.Command["updates"][0]["u"]; + var fields = update.IsBsonArray ? update[0]["$set"] : update.AsBsonDocument.GetValue("$set", new BsonDocument()); + if (!fields.AsBsonDocument.Contains(MongoStreamHeadFields.SnapshotRevision) + || Interlocked.Exchange(ref _armed, 0) != 1) return; + _entered.Set(); + if (!_release.Wait(TimeSpan.FromSeconds(20))) throw new TimeoutException("Snapshot metadata was not released."); + }); + var name = $"issue90-{Guid.NewGuid():N}"; + _database = new MongoClient(url).GetDatabase(name); + var options = new AcknowledgedMetadataOptions { PersistStreamHeadsOnBackgroundThread = false }; + _otherEngine = new MongoPersistenceEngine(_database, new DocumentObjectSerializer(), options); + _snapshotEngine = new MongoPersistenceEngine(new MongoClient(settings).GetDatabase(name), new DocumentObjectSerializer(), options); + _otherEngine.Initialize(); + _snapshotEngine.Initialize(); + _last = _otherEngine.CommitMany(2, Stream, BucketId).Last(); + } + + protected override async Task BecauseAsync() + { + Interlocked.Exchange(ref _armed, 1); + _snapshot = Task.Run(() => Save(_snapshotEngine, 2)); + try + { + _entered.Wait(TimeSpan.FromSeconds(20)).Should().BeTrue("the snapshot must pause immediately before its stream metadata write"); + if (_interleaving == 0) await CommitNext(); + if (_interleaving == 1) (await Save(_otherEngine, 4)).Should().BeTrue(); + if (_interleaving == 2) (await Save(_otherEngine, 1)).Should().BeTrue(); + } + finally { _release.Set(); } + _saved = await _snapshot; + if (_interleaving == 3) await CommitNext(); + if (_interleaving == 4) (await Save(_otherEngine, 1)).Should().BeTrue(); + if (_interleaving == 5) (await Save(_otherEngine, 2)).Should().BeTrue(); + } + + private async Task CommitNext() + { + var attempt = _last.BuildNextAttempt(); + if (_async) await _otherEngine.CommitAsync(attempt, CancellationToken.None); + else _otherEngine.Commit(attempt); + } + + private Task Save(MongoPersistenceEngine engine, int revision) + { + var snapshot = new Snapshot(BucketId, Stream, revision, "snapshot"); + return _async ? engine.AddSnapshotAsync(snapshot, CancellationToken.None) : Task.FromResult(engine.AddSnapshot(snapshot)); + } + + [Fact] + public void Metadata_preserves_commits_and_the_highest_snapshot_revision() + { + _saved.Should().BeTrue(); + var head = _database.GetCollection("Streams").Find(FilterDefinition.Empty).Single(); + head[MongoStreamHeadFields.HeadRevision].AsInt32.Should().Be(ExpectedHead); + head[MongoStreamHeadFields.SnapshotRevision].AsInt32.Should().Be(ExpectedSnapshot); + head[MongoStreamHeadFields.Unsnapshotted].AsInt32.Should().Be(ExpectedHead - ExpectedSnapshot); + } + + [Fact] + public async Task Public_candidate_queries_include_exactly_the_unsnapshotted_events() + { + var count = ExpectedHead - ExpectedSnapshot; + var candidates = await Candidates(Math.Max(1, count)); + if (count == 0) candidates.Should().BeEmpty(); + else + { + candidates.Should().ContainSingle(); + candidates[0].HeadRevision.Should().Be(ExpectedHead); + candidates[0].SnapshotRevision.Should().Be(ExpectedSnapshot); + } + (await Candidates(count + 1)).Should().BeEmpty(); + } + + private async Task> Candidates(int threshold) + { + if (!_async) return _snapshotEngine.GetStreamsToSnapshot(BucketId, threshold).ToList(); + var observer = new StreamHeadObserver(); + await _snapshotEngine.GetStreamsToSnapshotAsync(BucketId, threshold, observer, CancellationToken.None); + return observer.Heads; + } + + protected override void Cleanup() + { + _release.Set(); + _snapshot?.GetAwaiter().GetResult(); + _snapshotEngine?.Dispose(); + _otherEngine?.Dispose(); + if (_database != null) _database.Client.DropDatabase(_database.DatabaseNamespace.DatabaseName); + _entered.Dispose(); + _release.Dispose(); + } + + private sealed class StreamHeadObserver : IAsyncObserver + { + public IList Heads { get; } = new List(); + public Task OnNextAsync(IStreamHead value, CancellationToken cancellationToken = default) + { Heads.Add(value); return Task.FromResult(true); } + public Task OnErrorAsync(Exception ex, CancellationToken cancellationToken = default) => Task.FromException(ex); + public Task OnCompletedAsync(CancellationToken cancellationToken = default) => Task.CompletedTask; + } + + private sealed class AcknowledgedMetadataOptions : MongoPersistenceOptions + { + public override MongoCollectionSettings GetStreamSettings() => new() { WriteConcern = WriteConcern.Acknowledged }; + public override MongoCollectionSettings GetSnapshotSettings() => new() { WriteConcern = WriteConcern.Acknowledged }; + } + } +} +#endif diff --git a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs index 252a22b..00fb201 100644 --- a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs +++ b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs @@ -410,26 +410,20 @@ public async Task AddSnapshotAsync(ISnapshot snapshot, CancellationToken c // stream that needs to be snapshotted because the insert fails and the SnapshotRevision isn't being updated. await PersistedSnapshots.UpdateOneAsync(query, update, UpsertUpdateOptions, cancellationToken).ConfigureAwait(false); - // More commits could have been made between us deciding that a snapshot is required and writing it so just - // resetting the Un-snapshotted count may be a little off. Adding snapshots should be a separate process so - // this is a good chance to make sure the numbers are still in-sync - it only adds a 'read' after all ... + // Retain validation of missing/invalid heads, but calculate the update from current server-side values. BsonDocument streamHeadId = GetStreamHeadId(snapshot.BucketId, snapshot.StreamId); using var cursor = await PersistedStreamHeads .FindAsync(Builders.Filter.Eq(MongoStreamHeadFields.Id, streamHeadId), cancellationToken: cancellationToken) .ConfigureAwait(false); - StreamHead streamHead = (await cursor.FirstAsync(cancellationToken).ConfigureAwait(false)) + _ = (await cursor.FirstAsync(cancellationToken).ConfigureAwait(false)) .ToStreamHead(); - int unsnapshotted = streamHead.HeadRevision - snapshot.StreamRevision; - await PersistedStreamHeads.UpdateOneAsync( Builders.Filter .Eq(MongoStreamHeadFields.Id, streamHeadId), - Builders.Update - .Set(MongoStreamHeadFields.SnapshotRevision, snapshot.StreamRevision) - .Set(MongoStreamHeadFields.Unsnapshotted, unsnapshotted), + CreateSnapshotMetadataUpdate(snapshot.StreamRevision), cancellationToken: cancellationToken ); diff --git a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs index 37f9fde..d43e777 100644 --- a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs +++ b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs @@ -622,6 +622,30 @@ public virtual IEnumerable GetStreamsToSnapshot(string bucketId, in }); } + private static UpdateDefinition CreateSnapshotMetadataUpdate(int snapshotRevision) + { + var currentSnapshot = "$" + MongoStreamHeadFields.SnapshotRevision; + var nextSnapshot = new BsonDocument("$max", new BsonArray { currentSnapshot, snapshotRevision }); + // Subtract only the snapshot advance: commit increments may arrive before or after this update. + // Recomputing from HeadRevision would count delayed commit metadata twice. Do not clamp negatives. + // Legacy heads without Unsnapshotted retain the previous head-minus-snapshot fallback. + var currentCount = new BsonDocument("$ifNull", new BsonArray + { + "$" + MongoStreamHeadFields.Unsnapshotted, + new BsonDocument("$subtract", new BsonArray { "$" + MongoStreamHeadFields.HeadRevision, currentSnapshot }) + }); + var advance = new BsonDocument("$subtract", new BsonArray { nextSnapshot, currentSnapshot }); + PipelineDefinition pipeline = new[] + { + new BsonDocument("$set", new BsonDocument + { + { MongoStreamHeadFields.SnapshotRevision, nextSnapshot }, + { MongoStreamHeadFields.Unsnapshotted, new BsonDocument("$subtract", new BsonArray { currentCount, advance }) } + }) + }; + return Builders.Update.Pipeline(pipeline); + } + /// public virtual bool AddSnapshot(ISnapshot snapshot) { @@ -644,22 +668,16 @@ public virtual bool AddSnapshot(ISnapshot snapshot) // stream that needs to be snapshotted because the insert fails and the SnapshotRevision isn't being updated. PersistedSnapshots.UpdateOne(query, update, UpsertUpdateOptions); - // More commits could have been made between us deciding that a snapshot is required and writing it so just - // resetting the Un-snapshotted count may be a little off. Adding snapshots should be a separate process so - // this is a good chance to make sure the numbers are still in-sync - it only adds a 'read' after all ... + // Retain validation of missing/invalid heads, but calculate the update from current server-side values. BsonDocument streamHeadId = GetStreamHeadId(snapshot.BucketId, snapshot.StreamId); - StreamHead streamHead = PersistedStreamHeads.Find(Builders.Filter.Eq(MongoStreamHeadFields.Id, streamHeadId)) + _ = PersistedStreamHeads.Find(Builders.Filter.Eq(MongoStreamHeadFields.Id, streamHeadId)) .First() .ToStreamHead(); - int unsnapshotted = streamHead.HeadRevision - snapshot.StreamRevision; - PersistedStreamHeads.UpdateOne( Builders.Filter .Eq(MongoStreamHeadFields.Id, streamHeadId), - Builders.Update - .Set(MongoStreamHeadFields.SnapshotRevision, snapshot.StreamRevision) - .Set(MongoStreamHeadFields.Unsnapshotted, unsnapshotted) + CreateSnapshotMetadataUpdate(snapshot.StreamRevision) ); return true;