Fix atomic snapshot metadata accounting (#90)

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)

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
 ===
 

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<BsonDocument> _metadataCommands = new();
+        private const string Bucket = "accounting";
+        private const string Stream = "delayed-metadata";
+        private IMongoCollection<BsonDocument> Heads => _database.GetCollection<BsonDocument>("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<CommandStartedEvent>(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<bool> 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<BsonDocument>.Empty,
+                Builders<BsonDocument>.Update.Set(MongoStreamHeadFields.Unsnapshotted, 2));
+            (await Save(4)).Should().BeTrue();
+            Heads.Find(FilterDefinition<BsonDocument>.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<BsonDocument>.Empty, Builders<BsonDocument>.Update
+                .Max(MongoStreamHeadFields.HeadRevision, 2).Inc(MongoStreamHeadFields.Unsnapshotted, 2));
+            var head = Heads.Find(FilterDefinition<BsonDocument>.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<BsonDocument>.Empty,
+                nullValue ? Builders<BsonDocument>.Update.Set(MongoStreamHeadFields.Unsnapshotted, BsonNull.Value)
+                    : Builders<BsonDocument>.Update.Unset(MongoStreamHeadFields.Unsnapshotted));
+            (await Save(2)).Should().BeTrue();
+            var head = Heads.Find(FilterDefinition<BsonDocument>.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<BsonDocument>.Empty);
+            else Heads.UpdateOne(FilterDefinition<BsonDocument>.Empty,
+                Builders<BsonDocument>.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<Task> save = () => _engine.AddSnapshotAsync(
+                new Snapshot(Bucket, Stream, 2, "payload"), cancellation.Token);
+            await save.Should().ThrowAsync<OperationCanceledException>();
+            _metadataCommands.Should().BeEmpty();
+            Heads.Find(FilterDefinition<BsonDocument>.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<bool>? _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<CommandStartedEvent>(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<bool> 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<BsonDocument>("Streams").Find(FilterDefinition<BsonDocument>.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<IList<IStreamHead>> 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<IStreamHead>
+        {
+            public IList<IStreamHead> Heads { get; } = new List<IStreamHead>();
+            public Task<bool> 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