diff --git a/Changelog.md b/Changelog.md index f95ac5f..b9f79d3 100644 --- a/Changelog.md +++ b/Changelog.md @@ -1,6 +1,8 @@ # NEventStore.Persistence.MongoDB -## Unreleased +## Unreleased + +- Harden duplicate-write classification using driver error codes/categories and structured key patterns, with a narrowly matched legacy index-message fallback. Preserve checkpoint retries, duplicate commit detection and concurrency handling without matching index names inside user payloads. No public API or schema change. [#92](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/92) - Run the same acceptance and regression suite with pinned MongoDB.Driver 3.0.0 and 3.13.0 test dependencies in isolated CI jobs on MongoDB 7.0.45, retaining a net10.0/current-driver lane on MongoDB 9.0.4. Verify restored Driver/BSON versions and the packed dependency contract across supported frameworks. Production dependencies and the published `[3.0.0,4.0.0)` range are unchanged; no API, schema or migration change. [#93](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/93) diff --git a/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue92DuplicateClassification.cs b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue92DuplicateClassification.cs new file mode 100644 index 0000000..72f6937 --- /dev/null +++ b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue92DuplicateClassification.cs @@ -0,0 +1,49 @@ +using System.Reflection; +using FluentAssertions; +using MongoDB.Bson; +using MongoDB.Driver; +using NEventStore.Persistence.AcceptanceTests.BDD; +#if NUNIT +using NUnit.Framework; +namespace NEventStore.Persistence.MongoDB.Tests.AcceptanceTests.Issues; +[TestFixture(11000, "_id", "Checkpoint")] +[TestFixture(11000, "CommitId", "Commit")] +[TestFixture(11000, "StreamRevision", "Other")] +[TestFixture(121, "_id", "None")] +public class when_Issue92_structured_duplicate_details_are_available(int code, string key, string expected) : SpecificationBase +{ + private string? _actual; + protected override void Because() + { + var error = new MongoCommandException(new global::MongoDB.Driver.Core.Connections.ConnectionId(new global::MongoDB.Driver.Core.Servers.ServerId(new global::MongoDB.Driver.Core.Clusters.ClusterId(), new System.Net.DnsEndPoint("localhost", 27017))), "message without index information", new BsonDocument(), + new BsonDocument { { "code", code }, { "keyPattern", key == "CommitId" ? new BsonDocument { { "BucketId", 1 }, { "StreamId", 1 }, { "CommitId", 1 } } : new BsonDocument(key, 1) } }); + var classifier = typeof(MongoPersistenceEngine).Assembly.GetType("NEventStore.Persistence.MongoDB.DuplicateKeyClassifier"); + _actual = classifier?.GetMethod("Classify", BindingFlags.Static | BindingFlags.NonPublic)!.Invoke(null, new object[] { error })!.ToString(); + } + [Fact] public void Should_use_structured_error_identity_instead_of_message_text() => _actual.Should().Be(expected); +} +#endif + + +#if NUNIT +[TestFixture("E11000 duplicate key error collection: db.c index: _id_ dup key: { _id: 1 }", "Checkpoint")] +[TestFixture("E11000 duplicate key error index: db.c.$_id_ dup key: { : 1 }", "Checkpoint")] +[TestFixture("E11001 duplicate key error index: CommitId_Index dup key: { : 1 }", "Commit")] +[TestFixture("E11000 duplicate key error index: db.c.$CommitId_Index dup key: { : 1 }", "Commit")] +[TestFixture("E11000 duplicate key error index: StreamRevision_Index dup key: { value: '_id_ CommitId_Index' }", "Other")] +[TestFixture("E11000 duplicate key error index: custom_id_ dup key: { : 1 }", "Other")] +[TestFixture("E11000 duplicate key error index: CommitId_Index_extra dup key: { : 1 }", "Other")] +[TestFixture("E11000 duplicate key error without index", "Other")] +[TestFixture("validation rejected user text E11000 duplicate key error index: _id_ dup key: {}", "None")] +[TestFixture("E11002 duplicate key error index: _id_ dup key: {}", "None")] +public class when_Issue92_legacy_duplicate_messages_are_classified(string message, string expected) : SpecificationBase +{ + private string? _actual; + protected override void Because() + { + var classifier = typeof(MongoPersistenceEngine).Assembly.GetType("NEventStore.Persistence.MongoDB.DuplicateKeyClassifier"); + _actual = classifier?.GetMethod("Classify", BindingFlags.Static | BindingFlags.NonPublic)!.Invoke(null, new object[] { new MongoException(message) })!.ToString(); + } + [Fact] public void Should_only_use_the_duplicate_prefix_and_complete_index_token() => _actual.Should().Be(expected); +} +#endif diff --git a/src/NEventStore.Persistence.MongoDB/DuplicateKeyClassifier.cs b/src/NEventStore.Persistence.MongoDB/DuplicateKeyClassifier.cs new file mode 100644 index 0000000..04f97ae --- /dev/null +++ b/src/NEventStore.Persistence.MongoDB/DuplicateKeyClassifier.cs @@ -0,0 +1,52 @@ +using System.Text.RegularExpressions; +using MongoDB.Bson; +using MongoDB.Driver; + +namespace NEventStore.Persistence.MongoDB +{ + internal enum DuplicateKeyKind { None, Checkpoint, Commit, Other } + + internal static class DuplicateKeyClassifier + { + internal static DuplicateKeyKind Classify(MongoException exception) + { + BsonDocument? details = null; + string message = exception.Message; + bool duplicate; + if (exception is MongoWriteException write) + { + duplicate = write.WriteError.Category == ServerErrorCategory.DuplicateKey; + details = write.WriteError.Details; + message = write.WriteError.Message; + } + else if (exception is MongoCommandException command) + { + duplicate = command.Code == 11000 || command.Code == 11001 || command.Code == 12582; + details = command.Result; + } + else + { + // Legacy responses lack typed details. Only an actual duplicate-key prefix qualifies. + duplicate = Regex.IsMatch(message, @"^E1100[01]\s+duplicate key\b", RegexOptions.CultureInvariant); + } + if (!duplicate) return DuplicateKeyKind.None; + + if (details != null && details.TryGetValue("keyPattern", out var pattern) && pattern.IsBsonDocument) + { + var keys = pattern.AsBsonDocument; + if (keys.ElementCount == 1 && keys.Contains(MongoCommitFields.CheckpointNumber)) return DuplicateKeyKind.Checkpoint; + if (keys.ElementCount == 3 && keys.Contains(MongoCommitFields.BucketId) + && keys.Contains(MongoCommitFields.StreamId) && keys.Contains(MongoCommitFields.CommitId)) return DuplicateKeyKind.Commit; + return DuplicateKeyKind.Other; + } + + // Older servers expose the index only in errmsg; match its full token, not payload substrings. + var index = Regex.Match(message, @"\bindex:\s+(?\S+)\s+dup key\s*:", RegexOptions.CultureInvariant); + if (!index.Success) return DuplicateKeyKind.Other; + var name = index.Groups["name"].Value; + if (name == "_id_" || name.EndsWith(".$_id_", System.StringComparison.Ordinal)) return DuplicateKeyKind.Checkpoint; + if (name == MongoCommitIndexes.CommitId || name.EndsWith(".$" + MongoCommitIndexes.CommitId, System.StringComparison.Ordinal)) return DuplicateKeyKind.Commit; + return DuplicateKeyKind.Other; + } + } +} diff --git a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs index ceb1b27..d9c6031 100644 --- a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs +++ b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs @@ -253,15 +253,15 @@ public Task GetFromToAsync(long fromCheckpointToken, long toCheckpointToken, IAs } catch (MongoException e) { - if (!e.Message.Contains(ConcurrencyException)) + var duplicate = DuplicateKeyClassifier.Classify(e); + if (duplicate == DuplicateKeyKind.None) { MongoPersistenceEngineLogMessages.GenericPersistingError(Logger, attempt.CommitId, checkpointId, attempt.BucketId, attempt.StreamId, e.ToString(), e); throw; } // checkpoint index? - if (e.Message.Contains(MongoCommitIndexes.CheckpointNumberMMApV1) - || e.Message.Contains(MongoCommitIndexes.CheckpointNumberWiredTiger)) + if (duplicate == DuplicateKeyKind.Checkpoint) { MongoPersistenceEngineLogMessages.DuplicatedCheckpointTokenError(Logger, attempt.CommitId, checkpointId, attempt.BucketId, attempt.StreamId, e); await _checkpointGenerator.SignalDuplicateIdAsync(checkpointId, cancellationToken); @@ -275,7 +275,7 @@ public Task GetFromToAsync(long fromCheckpointToken, long toCheckpointToken, IAs await FillHoleAsync(attempt, checkpointId, cancellationToken).ConfigureAwait(false); } - if (e.Message.Contains(MongoCommitIndexes.CommitId)) + if (duplicate == DuplicateKeyKind.Commit) { var msg = string.Format(CultureInfo.InvariantCulture, MongoPersistenceEngineLogMessages.DuplicatedCommitErrorTemplate, attempt.CommitId, checkpointId, attempt.BucketId, attempt.StreamId); MongoPersistenceEngineLogMessages.Information(Logger, msg); diff --git a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs index 1a4f81c..23bb36d 100644 --- a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs +++ b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs @@ -16,7 +16,6 @@ namespace NEventStore.Persistence.MongoDB /// public partial class MongoPersistenceEngine : IPersistStreams { - private const string ConcurrencyException = "E1100"; private static readonly ILogger Logger = LogFactory.BuildLogger(typeof(MongoPersistenceEngine)); private readonly IDocumentSerializer _serializer; #if NET9_0_OR_GREATER @@ -498,15 +497,15 @@ public virtual IEnumerable GetFromTo(string bucketId, DateTime startDat } catch (MongoException e) { - if (!e.Message.Contains(ConcurrencyException)) + var duplicate = DuplicateKeyClassifier.Classify(e); + if (duplicate == DuplicateKeyKind.None) { MongoPersistenceEngineLogMessages.GenericPersistingError(Logger, attempt.CommitId, checkpointId, attempt.BucketId, attempt.StreamId, e.ToString(), e); throw; } // checkpoint index? - if (e.Message.Contains(MongoCommitIndexes.CheckpointNumberMMApV1) - || e.Message.Contains(MongoCommitIndexes.CheckpointNumberWiredTiger)) + if (duplicate == DuplicateKeyKind.Checkpoint) { MongoPersistenceEngineLogMessages.DuplicatedCheckpointTokenError(Logger, attempt.CommitId, checkpointId, attempt.BucketId, attempt.StreamId, e); _checkpointGenerator.SignalDuplicateId(checkpointId); @@ -520,7 +519,7 @@ public virtual IEnumerable GetFromTo(string bucketId, DateTime startDat FillHole(attempt, checkpointId); } - if (e.Message.Contains(MongoCommitIndexes.CommitId)) + if (duplicate == DuplicateKeyKind.Commit) { var msg = string.Format(CultureInfo.InvariantCulture, MongoPersistenceEngineLogMessages.DuplicatedCommitErrorTemplate, attempt.CommitId, checkpointId, attempt.BucketId, attempt.StreamId); MongoPersistenceEngineLogMessages.Information(Logger, msg);