From b64f7f1a17d705f634795e20e60edbfe311fec77 Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Tue, 3 Nov 2026 13:15:54 +0100 Subject: [PATCH 1/4] test: reproduce snapshot metadata races (#90) --- .../Issues/Issue90SnapshotMetadata.cs | 166 ++++++++++++++++++ 1 file changed, 166 insertions(+) create mode 100644 src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue90SnapshotMetadata.cs 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..a838c45 --- /dev/null +++ b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue90SnapshotMetadata.cs @@ -0,0 +1,166 @@ +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, 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 From 230343dbd4ea55e7abf7cfd18390d34bde98c92f Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Tue, 3 Nov 2026 13:16:02 +0100 Subject: [PATCH 2/4] Fix atomic snapshot accounting and require MongoDB 4.2 (#90) --- .github/workflows/ci.yml | 62 +++++++- Changelog.md | 4 + README.md | 14 ++ .../Issues/Issue90SnapshotMetadata.cs | 139 ++++++++++++++++++ .../MongoPersistenceEngine.Async.cs | 12 +- .../MongoPersistenceEngine.cs | 36 +++-- 6 files changed, 248 insertions(+), 19 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b7f5319..0384c50 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -104,7 +104,7 @@ jobs: steps: - name: Checkout - uses: actions/checkout@v7.0.1 + uses: actions/checkout@v7.0.1 with: fetch-depth: 0 fetch-tags: true @@ -139,3 +139,63 @@ jobs: path: "**/test-results-${{ matrix.tfm }}.trx" if-no-files-found: error retention-days: 14 + + # Verify the aggregation-update feature minimum without replacing current-server coverage. + test-minimum-mongodb: + name: Test (MongoDB 4.2, net8.0) + needs: build + runs-on: ubuntu-latest + services: + # The test project reads NEventStore.MongoDB from the environment and expects + # a real MongoDB server. This sidecar provides that dependency for each matrix leg. + mongodb: + image: mongo:4.2.24 + ports: + - 27017:27017 + options: >- + --health-cmd "mongo --eval \"db.adminCommand('ping')\"" + --health-interval 20s + --health-timeout 10s + --health-retries 10 + + env: + # Match the connection-string convention used by the acceptance tests. + NEventStore.MongoDB: mongodb://127.0.0.1:27017/NEventStore + + steps: + - name: Checkout + uses: actions/checkout@v7.0.1 + with: + fetch-depth: 0 + fetch-tags: true + submodules: recursive + + - name: Setup .NET SDKs + uses: actions/setup-dotnet@v6.0.0 + with: + dotnet-version: | + 6.0.x + 8.0.x + 9.0.x + 10.0.x + + - name: Cache NuGet packages + uses: actions/cache@v6.1.0 + with: + path: ~/.nuget/packages + key: nuget-${{ runner.os }}-${{ hashFiles('**/*.csproj', '**/*.props', '**/*.targets', '.config/dotnet-tools.json') }} + restore-keys: | + nuget-${{ runner.os }}- + + - name: Run tests for net8.0 + # Only the test project is executed here. It pulls in the production project and + # linked acceptance coverage from the NEventStore submodule. + run: dotnet test ./src/NEventStore.Persistence.MongoDB.Tests/NEventStore.Persistence.MongoDB.Core.Tests.csproj -c Release -f net8.0 --logger "trx;LogFileName=test-results-mongo42-net8.0.trx" + + - name: Upload test results + uses: actions/upload-artifact@v7.0.2 + with: + name: test-results-mongo42-net8.0 + path: "**/test-results-mongo42-net8.0.trx" + if-no-files-found: error + retention-days: 14 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 index a838c45..f7b37be 100644 --- a/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue90SnapshotMetadata.cs +++ b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue90SnapshotMetadata.cs @@ -10,6 +10,145 @@ 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)] 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; From f766a1400307daa0150921a8864b2bddef402a8c Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Tue, 3 Nov 2026 13:16:10 +0100 Subject: [PATCH 3/4] test: verify combined commit and snapshot metadata in CI --- .github/workflows/ci.yml | 68 +++++++++++++++++++++++++++++++++++++ ci/CombinedMetadataTests.cs | 57 +++++++++++++++++++++++++++++++ 2 files changed, 125 insertions(+) create mode 100644 ci/CombinedMetadataTests.cs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0384c50..f006902 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -199,3 +199,71 @@ jobs: path: "**/test-results-mongo42-net8.0.trx" if-no-files-found: error retention-days: 14 + + # Integration-only workspace overlay; issue 89 stays in its own PR. + test-combined-metadata: + name: Test (combined metadata, net8.0) + needs: build + runs-on: ubuntu-latest + services: + # The test project reads NEventStore.MongoDB from the environment and expects + # a real MongoDB server. This sidecar provides that dependency for each matrix leg. + mongodb: + image: mongo:9.0.4 + ports: + - 27017:27017 + options: >- + --health-cmd "mongosh --eval \"db.adminCommand('ping')\"" + --health-interval 20s + --health-timeout 10s + --health-retries 10 + + env: + # Match the connection-string convention used by the acceptance tests. + NEventStore.MongoDB: mongodb://127.0.0.1:27017/NEventStore + + steps: + - name: Checkout + uses: actions/checkout@v7.0.1 + with: + fetch-depth: 0 + fetch-tags: true + submodules: recursive + + - name: Apply pinned issue 89 implementation for integration verification + shell: bash + run: | + git fetch origin d12d48ad04da78112e885cf35b68704a48e98254 + git diff d12d48ad04da78112e885cf35b68704a48e98254^ d12d48ad04da78112e885cf35b68704a48e98254 -- src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs | git apply + git show d12d48ad04da78112e885cf35b68704a48e98254:src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue89StreamHeads.cs > src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue89StreamHeads.cs + cp ci/CombinedMetadataTests.cs src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/CombinedMetadataTests.cs + + - name: Setup .NET SDKs + uses: actions/setup-dotnet@v6.0.0 + with: + dotnet-version: | + 6.0.x + 8.0.x + 9.0.x + 10.0.x + + - name: Cache NuGet packages + uses: actions/cache@v6.1.0 + with: + path: ~/.nuget/packages + key: nuget-${{ runner.os }}-${{ hashFiles('**/*.csproj', '**/*.props', '**/*.targets', '.config/dotnet-tools.json') }} + restore-keys: | + nuget-${{ runner.os }}- + + - name: Run tests for net8.0 + # Only the test project is executed here. It pulls in the production project and + # linked acceptance coverage from the NEventStore submodule. + run: dotnet test ./src/NEventStore.Persistence.MongoDB.Tests/NEventStore.Persistence.MongoDB.Core.Tests.csproj -c Release -f net8.0 --logger "trx;LogFileName=test-results-combined-net8.0.trx" + + - name: Upload test results + uses: actions/upload-artifact@v7.0.2 + with: + name: test-results-combined-net8.0 + path: "**/test-results-combined-net8.0.trx" + if-no-files-found: error + retention-days: 14 diff --git a/ci/CombinedMetadataTests.cs b/ci/CombinedMetadataTests.cs new file mode 100644 index 0000000..53e17d9 --- /dev/null +++ b/ci/CombinedMetadataTests.cs @@ -0,0 +1,57 @@ +using FluentAssertions; +using NEventStore.Persistence.AcceptanceTests; +using NEventStore.Persistence.AcceptanceTests.BDD; + +namespace NEventStore.Persistence.MongoDB.Tests.AcceptanceTests.Issues +{ + // Compiled only by the integration CI job, with the pinned #89 implementation and fixtures. + public class Combined_metadata_when_an_older_commit_finishes_after_a_snapshot : Issue_89_stream_head_concern + { + private CommitAttempt _older = null!; + private CommitAttempt _newer = null!; + + protected override Task ContextAsync() + { + _older = Stream.BuildAttempt(); + _newer = _older.BuildNextAttempt(); + Gate.Arm(_older.StreamRevision, _newer.StreamRevision); + Options.PersistStreamHeadsOnBackgroundThread = true; + return Task.CompletedTask; + } + + protected override async Task BecauseAsync() + { + var olderCommit = Task.Run(() => Commit(_older)); + try + { + await WaitFor(Gate.Entered.Task); + await Commit(_newer); + await WaitFor(Gate.NewerCompleted.Task); + (await Snapshot(_newer.StreamRevision)).Should().BeTrue(); + ReadHead()[MongoStreamHeadFields.Unsnapshotted].AsInt32.Should().Be(-_older.Events.Count); + } + finally + { + Gate.Release.TrySetResult(true); + await WaitFor(Gate.OlderCompleted.Task); + await WaitFor(olderCommit); + } + } + + [Fact] + public void Completed_commit_and_snapshot_updates_agree_on_the_public_candidate_state() + { + var head = ReadHead(); + head[MongoStreamHeadFields.HeadRevision].AsInt32.Should().Be(_newer.StreamRevision); + head[MongoStreamHeadFields.SnapshotRevision].AsInt32.Should().Be(_newer.StreamRevision); + head[MongoStreamHeadFields.Unsnapshotted].AsInt32.Should().Be(0); + Engine.GetStreamsToSnapshot(Bucket.Default, 1).Should().BeEmpty(); + } + } + + public class Combined_metadata_when_an_older_commit_finishes_after_a_snapshot_async + : Combined_metadata_when_an_older_commit_finishes_after_a_snapshot + { + protected override bool Async => true; + } +} From fa6f1ecc398ea3086a2b5ddc64f51011b1d53da4 Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Tue, 3 Nov 2026 13:16:22 +0100 Subject: [PATCH 4/4] Restore base CI workflow and remove one-time integration fixture --- .github/workflows/ci.yml | 130 +----------------------------------- ci/CombinedMetadataTests.cs | 57 ---------------- 2 files changed, 1 insertion(+), 186 deletions(-) delete mode 100644 ci/CombinedMetadataTests.cs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f006902..b7f5319 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -104,7 +104,7 @@ jobs: steps: - name: Checkout - uses: actions/checkout@v7.0.1 + uses: actions/checkout@v7.0.1 with: fetch-depth: 0 fetch-tags: true @@ -139,131 +139,3 @@ jobs: path: "**/test-results-${{ matrix.tfm }}.trx" if-no-files-found: error retention-days: 14 - - # Verify the aggregation-update feature minimum without replacing current-server coverage. - test-minimum-mongodb: - name: Test (MongoDB 4.2, net8.0) - needs: build - runs-on: ubuntu-latest - services: - # The test project reads NEventStore.MongoDB from the environment and expects - # a real MongoDB server. This sidecar provides that dependency for each matrix leg. - mongodb: - image: mongo:4.2.24 - ports: - - 27017:27017 - options: >- - --health-cmd "mongo --eval \"db.adminCommand('ping')\"" - --health-interval 20s - --health-timeout 10s - --health-retries 10 - - env: - # Match the connection-string convention used by the acceptance tests. - NEventStore.MongoDB: mongodb://127.0.0.1:27017/NEventStore - - steps: - - name: Checkout - uses: actions/checkout@v7.0.1 - with: - fetch-depth: 0 - fetch-tags: true - submodules: recursive - - - name: Setup .NET SDKs - uses: actions/setup-dotnet@v6.0.0 - with: - dotnet-version: | - 6.0.x - 8.0.x - 9.0.x - 10.0.x - - - name: Cache NuGet packages - uses: actions/cache@v6.1.0 - with: - path: ~/.nuget/packages - key: nuget-${{ runner.os }}-${{ hashFiles('**/*.csproj', '**/*.props', '**/*.targets', '.config/dotnet-tools.json') }} - restore-keys: | - nuget-${{ runner.os }}- - - - name: Run tests for net8.0 - # Only the test project is executed here. It pulls in the production project and - # linked acceptance coverage from the NEventStore submodule. - run: dotnet test ./src/NEventStore.Persistence.MongoDB.Tests/NEventStore.Persistence.MongoDB.Core.Tests.csproj -c Release -f net8.0 --logger "trx;LogFileName=test-results-mongo42-net8.0.trx" - - - name: Upload test results - uses: actions/upload-artifact@v7.0.2 - with: - name: test-results-mongo42-net8.0 - path: "**/test-results-mongo42-net8.0.trx" - if-no-files-found: error - retention-days: 14 - - # Integration-only workspace overlay; issue 89 stays in its own PR. - test-combined-metadata: - name: Test (combined metadata, net8.0) - needs: build - runs-on: ubuntu-latest - services: - # The test project reads NEventStore.MongoDB from the environment and expects - # a real MongoDB server. This sidecar provides that dependency for each matrix leg. - mongodb: - image: mongo:9.0.4 - ports: - - 27017:27017 - options: >- - --health-cmd "mongosh --eval \"db.adminCommand('ping')\"" - --health-interval 20s - --health-timeout 10s - --health-retries 10 - - env: - # Match the connection-string convention used by the acceptance tests. - NEventStore.MongoDB: mongodb://127.0.0.1:27017/NEventStore - - steps: - - name: Checkout - uses: actions/checkout@v7.0.1 - with: - fetch-depth: 0 - fetch-tags: true - submodules: recursive - - - name: Apply pinned issue 89 implementation for integration verification - shell: bash - run: | - git fetch origin d12d48ad04da78112e885cf35b68704a48e98254 - git diff d12d48ad04da78112e885cf35b68704a48e98254^ d12d48ad04da78112e885cf35b68704a48e98254 -- src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs | git apply - git show d12d48ad04da78112e885cf35b68704a48e98254:src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue89StreamHeads.cs > src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue89StreamHeads.cs - cp ci/CombinedMetadataTests.cs src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/CombinedMetadataTests.cs - - - name: Setup .NET SDKs - uses: actions/setup-dotnet@v6.0.0 - with: - dotnet-version: | - 6.0.x - 8.0.x - 9.0.x - 10.0.x - - - name: Cache NuGet packages - uses: actions/cache@v6.1.0 - with: - path: ~/.nuget/packages - key: nuget-${{ runner.os }}-${{ hashFiles('**/*.csproj', '**/*.props', '**/*.targets', '.config/dotnet-tools.json') }} - restore-keys: | - nuget-${{ runner.os }}- - - - name: Run tests for net8.0 - # Only the test project is executed here. It pulls in the production project and - # linked acceptance coverage from the NEventStore submodule. - run: dotnet test ./src/NEventStore.Persistence.MongoDB.Tests/NEventStore.Persistence.MongoDB.Core.Tests.csproj -c Release -f net8.0 --logger "trx;LogFileName=test-results-combined-net8.0.trx" - - - name: Upload test results - uses: actions/upload-artifact@v7.0.2 - with: - name: test-results-combined-net8.0 - path: "**/test-results-combined-net8.0.trx" - if-no-files-found: error - retention-days: 14 diff --git a/ci/CombinedMetadataTests.cs b/ci/CombinedMetadataTests.cs deleted file mode 100644 index 53e17d9..0000000 --- a/ci/CombinedMetadataTests.cs +++ /dev/null @@ -1,57 +0,0 @@ -using FluentAssertions; -using NEventStore.Persistence.AcceptanceTests; -using NEventStore.Persistence.AcceptanceTests.BDD; - -namespace NEventStore.Persistence.MongoDB.Tests.AcceptanceTests.Issues -{ - // Compiled only by the integration CI job, with the pinned #89 implementation and fixtures. - public class Combined_metadata_when_an_older_commit_finishes_after_a_snapshot : Issue_89_stream_head_concern - { - private CommitAttempt _older = null!; - private CommitAttempt _newer = null!; - - protected override Task ContextAsync() - { - _older = Stream.BuildAttempt(); - _newer = _older.BuildNextAttempt(); - Gate.Arm(_older.StreamRevision, _newer.StreamRevision); - Options.PersistStreamHeadsOnBackgroundThread = true; - return Task.CompletedTask; - } - - protected override async Task BecauseAsync() - { - var olderCommit = Task.Run(() => Commit(_older)); - try - { - await WaitFor(Gate.Entered.Task); - await Commit(_newer); - await WaitFor(Gate.NewerCompleted.Task); - (await Snapshot(_newer.StreamRevision)).Should().BeTrue(); - ReadHead()[MongoStreamHeadFields.Unsnapshotted].AsInt32.Should().Be(-_older.Events.Count); - } - finally - { - Gate.Release.TrySetResult(true); - await WaitFor(Gate.OlderCompleted.Task); - await WaitFor(olderCommit); - } - } - - [Fact] - public void Completed_commit_and_snapshot_updates_agree_on_the_public_candidate_state() - { - var head = ReadHead(); - head[MongoStreamHeadFields.HeadRevision].AsInt32.Should().Be(_newer.StreamRevision); - head[MongoStreamHeadFields.SnapshotRevision].AsInt32.Should().Be(_newer.StreamRevision); - head[MongoStreamHeadFields.Unsnapshotted].AsInt32.Should().Be(0); - Engine.GetStreamsToSnapshot(Bucket.Default, 1).Should().BeEmpty(); - } - } - - public class Combined_metadata_when_an_older_commit_finishes_after_a_snapshot_async - : Combined_metadata_when_an_older_commit_finishes_after_a_snapshot - { - protected override bool Async => true; - } -}