diff --git a/Changelog.md b/Changelog.md index 0d88b20..db10241 100644 --- a/Changelog.md +++ b/Changelog.md @@ -2,6 +2,8 @@ ## Unreleased +- 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) - Update CI and packaging actions to exact stable version tags and use MongoDB 9.0.4 in CI for the startup fix in [SERVER-131779](https://jira.mongodb.org/browse/SERVER-131779). This changes CI database coverage from MongoDB 8 to 9; the library, driver dependency, and supported-target declarations are unchanged. No library breaking changes are introduced by this workflow update. diff --git a/docs/Stream-Heads.md b/docs/Stream-Heads.md new file mode 100644 index 0000000..a89b334 --- /dev/null +++ b/docs/Stream-Heads.md @@ -0,0 +1,46 @@ +# Stream-head metadata updates + +Commits are authoritative. Stream heads are derived metadata, and their writes can +complete in a different order from accepted commits when background persistence +is enabled. Metadata writes retain their configured write concern. + +## Commit-side invariants (#89) + +Within one stream lifetime, each successfully applied commit metadata update: + +- Raises `HeadRevision` to the maximum of its current value and the commit revision. +- Increments `Unsnapshotted` by that commit's event count, even if a newer head + revision is already present. Filtering out old revisions would lose increments. +- Preserves `SnapshotRevision`, initializing it to zero for a missing head. + +These changes occur together in one MongoDB document update. An upsert initializes +all three counters. Duplicate commit retries fail before scheduling a second +metadata update. This does not make replaying a metadata increment idempotent or +guarantee delivery of failed/unacknowledged background metadata writes. + +## Snapshot and lifecycle boundaries + +Commit-side updates must preserve snapshot-side metadata. With a stable snapshot +and all commit metadata applied, `Unsnapshotted` equals `HeadRevision` minus +`SnapshotRevision`. A snapshot update must account for concurrent commits and +older snapshots atomically to retain this relationship; that separate work is +tracked in [#90](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/90). +In the existing snapshot path, its read/compute/set sequence can overwrite an +intervening increment, and an older snapshot can lower `SnapshotRevision`. +The commit-side `$max` change alone does not correct either snapshot-side race. + +Snapshot resynchronization also needs coordination with delayed commit metadata: +if the head reaches 6 while revision 4's two-event metadata update is pending, a +snapshot at 4 computes `Unsnapshotted = 2`; the delayed increment then makes it 4. +Computing the counter from the current head in a snapshot pipeline alone does not +resolve that interleaving. Combined commit/snapshot changes must review it before +claiming the counter is exact with pending background writes. + +After writes have drained and deletion removes the head, a recreated stream starts +with fresh counters. A delayed write from a deleted lifetime can still recreate or +modify that head; preventing cross-lifetime writes requires the lifecycle work in +[#96](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/96). + +This fix changes no public API or stored schema and requires no data migration. +It prevents future revision regression from reordered commit metadata writes; +it does not backfill already regressed heads. diff --git a/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue89StreamHeads.cs b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue89StreamHeads.cs new file mode 100644 index 0000000..12ebbd1 --- /dev/null +++ b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue89StreamHeads.cs @@ -0,0 +1,269 @@ +using System.Collections.Concurrent; +using FluentAssertions; +using global::MongoDB.Bson; +using global::MongoDB.Driver; +using global::MongoDB.Driver.Core.Events; +using NEventStore.Persistence.AcceptanceTests; +using NEventStore.Persistence.AcceptanceTests.BDD; +using NEventStore.Serialization; + +namespace NEventStore.Persistence.MongoDB.Tests.AcceptanceTests.Issues +{ + public abstract class Issue_89_stream_head_concern : SpecificationBase + { + protected const string Stream = "reordered-stream"; + protected MongoPersistenceEngine Engine = null!; + protected IMongoDatabase Database = null!; + protected HeadWriteGate Gate = null!; + protected MongoPersistenceOptions Options = null!; + protected virtual bool Async => false; + + protected override void Context() + { + _ = new AcceptanceTestMongoPersistenceFactory(); + Gate = new HeadWriteGate(); + Options = new AcknowledgedMetadataOptions + { + PersistStreamHeadsOnBackgroundThread = false, + ConfigureClientSettingsAction = settings => settings.ClusterConfigurator = cluster => + { + cluster.Subscribe(Gate.Started); + cluster.Subscribe(Gate.Succeeded); + cluster.Subscribe(Gate.Failed); + } + }; + var url = new MongoUrlBuilder(AcceptanceTestMongoPersistenceFactory.GetConnectionString()) + { + DatabaseName = $"issue89-{Guid.NewGuid():N}" + }; + Database = Options.ConnectToDatabase(url.ToString()); + Engine = new MongoPersistenceEngine(Database, new DocumentObjectSerializer(), Options); + Engine.Initialize(); + } + + protected Task Commit(CommitAttempt attempt) + { + if (Async) return Engine.CommitAsync(attempt, CancellationToken.None); + return Task.FromResult(Engine.Commit(attempt)); + } + + protected Task Snapshot(int revision) + { + var snapshot = new Snapshot(Bucket.Default, Stream, revision, "snapshot"); + if (Async) return Engine.AddSnapshotAsync(snapshot, CancellationToken.None); + return Task.FromResult(Engine.AddSnapshot(snapshot)); + } + + protected BsonDocument ReadHead() => Database.GetCollection("Streams") + .Find(new BsonDocument()).Single(); + + protected override void Cleanup() + { + Gate?.Release.TrySetResult(true); + Engine?.Dispose(); + if (Database != null) Database.Client.DropDatabase(Database.DatabaseNamespace.DatabaseName); + } + + protected static async Task WaitFor(Task task) + { + if (await Task.WhenAny(task, Task.Delay(TimeSpan.FromSeconds(30))).ConfigureAwait(false) != task) + throw new TimeoutException("A stream-head command did not reach its deterministic gate."); + await task.ConfigureAwait(false); + } + + private sealed class AcknowledgedMetadataOptions : MongoPersistenceOptions + { + public override MongoCollectionSettings GetStreamSettings() => new() { WriteConcern = WriteConcern.Acknowledged }; + public override MongoCollectionSettings GetSnapshotSettings() => new() { WriteConcern = WriteConcern.Acknowledged }; + } + + // Gate the real driver's wire command, before sending it to MongoDB. Success events + // come from acknowledged writes, so completion means the server applied the update. + protected sealed class HeadWriteGate + { + private readonly ConcurrentDictionary> _pending = new(); + private int _oldRevision; + private int _newRevision; + private int _writes; + public TaskCompletionSource Entered { get; } = NewGate(); + public TaskCompletionSource Release { get; } = NewGate(); + public TaskCompletionSource OlderCompleted { get; } = NewGate(); + public TaskCompletionSource NewerCompleted { get; } = NewGate(); + public int Writes => Volatile.Read(ref _writes); + + public void Arm(int oldRevision, int newRevision) + { + _oldRevision = oldRevision; + _newRevision = newRevision; + } + + public void Started(CommandStartedEvent command) + { + if (command.CommandName != "update" || command.Command["update"] != "Streams") return; + var value = command.Command["updates"][0]["u"]; + if (!value.IsBsonDocument) return; + var update = value.AsBsonDocument; + BsonValue revision; + // Accept the old $set and the fixed $max so the same test reproduces the bug. + if (!(update.TryGetValue("$set", out var fields) && fields.AsBsonDocument.TryGetValue(MongoStreamHeadFields.HeadRevision, out revision)) && + !(update.TryGetValue("$max", out fields) && fields.AsBsonDocument.TryGetValue(MongoStreamHeadFields.HeadRevision, out revision))) return; + Interlocked.Increment(ref _writes); + if (revision.AsInt32 != _oldRevision && revision.AsInt32 != _newRevision) return; + _pending[command.RequestId] = revision.AsInt32 == _oldRevision ? OlderCompleted : NewerCompleted; + if (revision.AsInt32 != _oldRevision) return; + Entered.TrySetResult(true); + if (!Release.Task.Wait(TimeSpan.FromSeconds(30))) + OlderCompleted.TrySetException(new TimeoutException("The older stream-head write was not released.")); + } + + public void Succeeded(CommandSucceededEvent command) + { + if (_pending.TryRemove(command.RequestId, out var completion)) completion.TrySetResult(true); + } + + public void Failed(CommandFailedEvent command) + { + if (_pending.TryRemove(command.RequestId, out var completion)) completion.TrySetException(command.Failure); + } + + private static TaskCompletionSource NewGate() => new(TaskCreationOptions.RunContinuationsAsynchronously); + } + } + + public class Issue_89_when_head_writes_complete_in_reverse_order : Issue_89_stream_head_concern + { + protected virtual bool ExistingHead => false; + protected virtual bool ExistingSnapshot => false; + private CommitAttempt _older = null!; + private CommitAttempt _newer = null!; + private BsonDocument _beforeRelease = null!; + private BsonDocument _afterRelease = null!; + private Exception? _retryError; + private int _writesBeforeRetry; + + protected override async Task ContextAsync() + { + _older = Stream.BuildAttempt(); + if (ExistingHead) + { + await Commit(_older); + if (ExistingSnapshot) (await Snapshot(_older.StreamRevision)).Should().BeTrue(); + _older = _older.BuildNextAttempt(); + } + _newer = _older.BuildNextAttempt(); + Gate.Arm(_older.StreamRevision, _newer.StreamRevision); + Options.PersistStreamHeadsOnBackgroundThread = true; + } + + protected override async Task BecauseAsync() + { + // Task.Run also keeps the async caller independent if the driver raises its + // command event synchronously before the first incomplete await. + var olderCommit = Task.Run(() => Commit(_older)); + try + { + await WaitFor(Gate.Entered.Task); + await Commit(_newer); + await WaitFor(Gate.NewerCompleted.Task); + _beforeRelease = ReadHead(); + } + finally + { + Gate.Release.TrySetResult(true); + await WaitFor(Gate.OlderCompleted.Task); + await WaitFor(olderCommit); + } + _afterRelease = ReadHead(); + _writesBeforeRetry = Gate.Writes; + Options.PersistStreamHeadsOnBackgroundThread = false; + try { await Commit(_older); } + catch (Exception error) { _retryError = error; } + } + + [Fact] + public void Older_completion_cannot_hide_a_newer_acknowledged_revision() + { + _beforeRelease[MongoStreamHeadFields.HeadRevision].AsInt32.Should().Be(_newer.StreamRevision); + _afterRelease[MongoStreamHeadFields.HeadRevision].AsInt32.Should().Be(_newer.StreamRevision); + } + + [Fact] + public void Both_writes_contribute_their_events_without_overwriting_snapshot_metadata() + { + var snapshotRevision = ExistingSnapshot ? 2 : 0; + _afterRelease[MongoStreamHeadFields.SnapshotRevision].AsInt32.Should().Be(snapshotRevision); + _afterRelease[MongoStreamHeadFields.Unsnapshotted].AsInt32.Should().Be(_newer.StreamRevision - snapshotRevision); + _afterRelease[MongoStreamHeadFields.Unsnapshotted].AsInt32.Should() + .Be(_beforeRelease[MongoStreamHeadFields.Unsnapshotted].AsInt32 + _older.Events.Count); + } + + [Fact] + public void Duplicate_commit_retry_does_not_increment_metadata_again() + { + _retryError.Should().BeOfType(); + Gate.Writes.Should().Be(_writesBeforeRetry); + ReadHead().Equals(_afterRelease).Should().BeTrue(); + Engine.GetFrom(Bucket.Default, Stream, 0, int.MaxValue).Should().HaveCount(ExistingHead ? 3 : 2); + } + } + + public class Issue_89_when_head_writes_complete_in_reverse_order_async : Issue_89_when_head_writes_complete_in_reverse_order + { + protected override bool Async => true; + } + public class Issue_89_when_existing_head_writes_complete_in_reverse_order : Issue_89_when_head_writes_complete_in_reverse_order + { + protected override bool ExistingHead => true; + } + public class Issue_89_when_existing_head_writes_complete_in_reverse_order_async : Issue_89_when_existing_head_writes_complete_in_reverse_order + { + protected override bool Async => true; + } + public class Issue_89_when_snapshotted_head_writes_complete_in_reverse_order : Issue_89_when_existing_head_writes_complete_in_reverse_order + { + protected override bool ExistingSnapshot => true; + } + public class Issue_89_when_snapshotted_head_writes_complete_in_reverse_order_async : Issue_89_when_snapshotted_head_writes_complete_in_reverse_order + { + protected override bool Async => true; + } + + public class Issue_89_when_inserting_a_stream_head : Issue_89_stream_head_concern + { + protected virtual bool Recreate => false; + protected override async Task ContextAsync() + { + if (!Recreate) return; + var initial = Stream.BuildAttempt(); + await Commit(initial); + await Commit(initial.BuildNextAttempt()); + (await Snapshot(2)).Should().BeTrue(); + // All metadata writes have drained. Concurrent deletion/recreation is #96. + if (Async) await Engine.DeleteStreamAsync(Bucket.Default, Stream, CancellationToken.None); + else Engine.DeleteStream(Bucket.Default, Stream); + } + + protected override async Task BecauseAsync() => await Commit(Stream.BuildAttempt()); + + [Fact] + public void Upsert_initializes_all_counters_for_this_stream_lifetime() + { + var head = ReadHead(); + head[MongoStreamHeadFields.HeadRevision].AsInt32.Should().Be(2); + head[MongoStreamHeadFields.SnapshotRevision].AsInt32.Should().Be(0); + head[MongoStreamHeadFields.Unsnapshotted].AsInt32.Should().Be(2); + } + } + public class Issue_89_when_inserting_a_stream_head_async : Issue_89_when_inserting_a_stream_head + { + protected override bool Async => true; + } + public class Issue_89_when_recreating_a_stream_head : Issue_89_when_inserting_a_stream_head + { + protected override bool Recreate => true; + } + public class Issue_89_when_recreating_a_stream_head_async : Issue_89_when_recreating_a_stream_head + { + protected override bool Async => true; + } +} diff --git a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs index 286b5ef..252a22b 100644 --- a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs +++ b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs @@ -525,7 +525,8 @@ private void UpdateStreamHeadInBackgroundThread(string bucketId, string streamId await PersistedStreamHeads.UpdateOneAsync( Builders.Filter.Eq(MongoStreamHeadFields.Id, streamHeadId), Builders.Update - .Set(MongoStreamHeadFields.HeadRevision, streamRevision) + // Completion order can differ from commit order; every write still adds its events. + .Max(MongoStreamHeadFields.HeadRevision, streamRevision) .Inc(MongoStreamHeadFields.SnapshotRevision, 0) .Inc(MongoStreamHeadFields.Unsnapshotted, eventsCount), UpsertUpdateOptions, diff --git a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs index ced168a..37f9fde 100644 --- a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs +++ b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs @@ -763,7 +763,8 @@ private void UpdateStreamHeadInBackgroundThread(string bucketId, string streamId PersistedStreamHeads.UpdateOne( Builders.Filter.Eq(MongoStreamHeadFields.Id, streamHeadId), Builders.Update - .Set(MongoStreamHeadFields.HeadRevision, streamRevision) + // Completion order can differ from commit order; every write still adds its events. + .Max(MongoStreamHeadFields.HeadRevision, streamRevision) .Inc(MongoStreamHeadFields.SnapshotRevision, 0) .Inc(MongoStreamHeadFields.Unsnapshotted, eventsCount), UpsertUpdateOptions