Skip to content

Harden duplicate-key classification (#92) #114

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
AGiorgetti wants to merge 1 commit into
base: develop
Choose a base branch
from
Draft

Harden duplicate-key classification (#92) #114

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion Changelog.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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)

Expand Down
49 changes: 49 additions & 0 deletions ...tStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue92DuplicateClassification.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
52 changes: 52 additions & 0 deletions src/NEventStore.Persistence.MongoDB/DuplicateKeyClassifier.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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+(?<name>\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;
}
}
}
8 changes: 4 additions & 4 deletions src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand All @@ -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);
Expand Down
9 changes: 4 additions & 5 deletions src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@ namespace NEventStore.Persistence.MongoDB
/// </summary>
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
Expand Down Expand Up @@ -498,15 +497,15 @@ public virtual IEnumerable<ICommit> 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);
Expand All @@ -520,7 +519,7 @@ public virtual IEnumerable<ICommit> 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);
Expand Down
Loading