From 28ebbd5a771a07228954a733ba99ca3e1cecd984 Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Fri, 23 Oct 2026 16:02:13 +0200 Subject: [PATCH 1/5] test: reproduce recycle-bin collisions and metadata loss (#87) Baseline MongoDB 7.0.45/net10 run: 12 expected failures, 10 passes, 0 skips. Product source is unchanged in this commit. Co-authored by Alessandro Giorgetti and Codex (OpenAI). --- .../Issues/Issue87RecycleCollision.cs | 115 +++++++++++ .../Issues/Issue87RecycleCompatibility.cs | 183 ++++++++++++++++++ 2 files changed, 298 insertions(+) create mode 100644 src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87RecycleCollision.cs create mode 100644 src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87RecycleCompatibility.cs diff --git a/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87RecycleCollision.cs b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87RecycleCollision.cs new file mode 100644 index 0000000..3c0f93e --- /dev/null +++ b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87RecycleCollision.cs @@ -0,0 +1,115 @@ +using FluentAssertions; +using global::MongoDB.Bson; +using global::MongoDB.Driver; +using NEventStore.Persistence.AcceptanceTests; +using NEventStore.Persistence.AcceptanceTests.BDD; +using NEventStore.Serialization; + +namespace NEventStore.Persistence.MongoDB.Tests.AcceptanceTests.Issues +{ + public abstract class Issue_87_recycle_collision : SpecificationBase + { + private MongoPersistenceEngine? _engine; + private IMongoDatabase? _database; + private Exception? _error; + private const string Stream = "shared-stream"; + protected virtual bool Async => false; + protected virtual bool Recreate => false; + private string SecondBucket => Recreate ? "bucket-a" : "bucket-b"; + + protected override async Task ContextAsync() + { + // Initialize the same legacy serializers used by the acceptance suite. + _ = new AcceptanceTestMongoPersistenceFactory(); + var options = new AcknowledgedMetadataOptions { PersistStreamHeadsOnBackgroundThread = false }; + var url = new MongoUrlBuilder(AcceptanceTestMongoPersistenceFactory.GetConnectionString()) + { + DatabaseName = $"issue87-{Guid.NewGuid():N}" + }; + _database = options.ConnectToDatabase(url.ToString()); + _engine = new MongoPersistenceEngine(_database, new DocumentObjectSerializer(), options); + _engine.Initialize(); + await Seed("bucket-a"); + await Delete("bucket-a"); + await Seed(SecondBucket); + } + + private async Task Seed(string bucket) + { + if (Async) + { + await _engine!.CommitManyAsync(2, Stream, bucket); + (await _engine!.AddSnapshotAsync(new Snapshot(bucket, Stream, 2, "snapshot"), CancellationToken.None)).Should().BeTrue(); + } + else + { + _engine!.CommitMany(2, Stream, bucket); + _engine!.AddSnapshot(new Snapshot(bucket, Stream, 2, "snapshot")).Should().BeTrue(); + } + } + + private Task Delete(string bucket) + { + if (Async) + return _engine!.DeleteStreamAsync(bucket, Stream, CancellationToken.None); + _engine!.DeleteStream(bucket, Stream); + return Task.CompletedTask; + } + + protected override async Task BecauseAsync() + { + try { await Delete(SecondBucket); } + catch (Exception error) { _error = error; } + } + + [Fact] + public void Independent_stream_lifetimes_can_be_deleted_without_unique_key_collisions() + { + _error.Should().BeNull(); + _engine!.GetDeletedCommits().Should().HaveCount(4) + .And.OnlyContain(commit => commit.StreamId == Stream && commit.BucketId == MongoSystemBuckets.RecycleBin); + _engine.GetFrom(0).Should().BeEmpty(); + _engine.GetDeletedCommits().Select(commit => commit.CheckpointToken).Should().Equal(1L, 2L, 3L, 4L); + } + + [Fact] + public void Successful_recycling_removes_live_commits_heads_and_snapshots() + { + Count("Commits", MongoCommitFields.BucketId, SecondBucket).Should().Be(0); + Count("Commits", MongoCommitFields.BucketId, MongoSystemBuckets.RecycleBin).Should().Be(4); + Count("Streams", MongoStreamHeadFields.FullQualifiedBucketId, SecondBucket).Should().Be(0); + Count("Snapshots", MongoSnapshotFields.FullQualifiedBucketId, SecondBucket).Should().Be(0); + } + + private long Count(string collection, string field, string value) => + _database!.GetCollection(collection).CountDocuments(new BsonDocument(field, value)); + + protected override void Cleanup() + { + _engine?.Dispose(); + if (_database != null) + _database.Client.DropDatabase(_database.DatabaseNamespace.DatabaseName); + } + + private sealed class AcknowledgedMetadataOptions : MongoPersistenceOptions + { + public override MongoCollectionSettings GetStreamSettings() => new() { WriteConcern = WriteConcern.Acknowledged }; + public override MongoCollectionSettings GetSnapshotSettings() => new() { WriteConcern = WriteConcern.Acknowledged }; + } + } + + public class Issue_87_when_same_stream_is_deleted_in_different_buckets : Issue_87_recycle_collision { } + public class Issue_87_when_same_stream_is_deleted_in_different_buckets_async : Issue_87_recycle_collision + { + protected override bool Async => true; + } + public class Issue_87_when_a_deleted_stream_is_recreated_and_deleted : Issue_87_recycle_collision + { + protected override bool Recreate => true; + } + public class Issue_87_when_a_deleted_stream_is_recreated_and_deleted_async : Issue_87_recycle_collision + { + protected override bool Recreate => true; + protected override bool Async => true; + } +} diff --git a/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87RecycleCompatibility.cs b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87RecycleCompatibility.cs new file mode 100644 index 0000000..054b803 --- /dev/null +++ b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87RecycleCompatibility.cs @@ -0,0 +1,183 @@ +using FluentAssertions; +using global::MongoDB.Bson; +using global::MongoDB.Driver; +using NEventStore.Persistence.AcceptanceTests; +using NEventStore.Persistence.AcceptanceTests.BDD; +using NEventStore.Persistence.MongoDB.Support; +using NEventStore.Serialization; + +namespace NEventStore.Persistence.MongoDB.Tests.AcceptanceTests.Issues +{ + public abstract class Issue_87_seeded_stream : SpecificationBase + { + protected const string Bucket = "bucket"; + protected const string Stream = "stream"; + protected IMongoDatabase Database = null!; + protected MongoPersistenceEngine Engine = null!; + protected MongoPersistenceOptions Options = null!; + protected virtual bool Async => false; + + protected override void Context() + { + _ = new AcceptanceTestMongoPersistenceFactory(); + Options = new AcknowledgedMetadataOptions { PersistStreamHeadsOnBackgroundThread = false }; + var url = new MongoUrlBuilder(AcceptanceTestMongoPersistenceFactory.GetConnectionString()) + { + DatabaseName = $"issue87-{Guid.NewGuid():N}" + }; + Database = Options.ConnectToDatabase(url.ToString()); + Engine = new MongoPersistenceEngine(Database, new DocumentObjectSerializer(), Options); + Engine.Initialize(); + Engine.CommitMany(2, Stream, Bucket); + Engine.AddSnapshot(new Snapshot(Bucket, Stream, 2, "snapshot")).Should().BeTrue(); + } + + protected long Count(string collection, string field, string value) => + Database.GetCollection(collection).CountDocuments(new BsonDocument(field, value)); + + protected override void Cleanup() + { + Engine?.Dispose(); + if (Database != null) + Database.Client.DropDatabase(Database.DatabaseNamespace.DatabaseName); + } + + private sealed class AcknowledgedMetadataOptions : MongoPersistenceOptions + { + public override MongoCollectionSettings GetStreamSettings() => new() { WriteConcern = WriteConcern.Acknowledged }; + public override MongoCollectionSettings GetSnapshotSettings() => new() { WriteConcern = WriteConcern.Acknowledged }; + } + } + + public class Issue_87_when_recycling_update_fails : Issue_87_seeded_stream + { + private Exception? _error; + + protected override void Context() + { + base.Context(); + Database.RunCommand(new BsonDocument + { + { "collMod", "Commits" }, + { "validator", new BsonDocument(MongoCommitFields.BucketId, + new BsonDocument("$ne", MongoSystemBuckets.RecycleBin)) }, + { "validationLevel", "strict" }, + { "validationAction", "error" } + }); + } + + protected override async Task BecauseAsync() + { + try + { + if (Async) + await Engine.DeleteStreamAsync(Bucket, Stream, CancellationToken.None); + else + Engine.DeleteStream(Bucket, Stream); + } + catch (Exception error) { _error = error; } + } + + [Fact] + public void The_storage_failure_is_propagated() + { + var cause = _error.Should().BeOfType().Which.InnerException; + cause.Should().BeOfType().Which.WriteError.Code.Should().Be(121); + } + + [Fact] + public void Live_commits_remain_unchanged() + { + Engine.GetFrom(Bucket, Stream, 1, 4).Select(commit => commit.CheckpointToken).Should().Equal(1L, 2L); + Count("Commits", MongoCommitFields.BucketId, MongoSystemBuckets.RecycleBin).Should().Be(0); + } + + [Fact] + public void The_live_head_is_retained() => + Count("Streams", MongoStreamHeadFields.FullQualifiedBucketId, Bucket).Should().Be(1); + + [Fact] + public void The_live_snapshot_is_retained() => + Count("Snapshots", MongoSnapshotFields.FullQualifiedBucketId, Bucket).Should().Be(1); + + } + + public class Issue_87_when_recycling_update_fails_async : Issue_87_when_recycling_update_fails + { + protected override bool Async => true; + } + + public class Issue_87_when_reading_legacy_recycled_commits : Issue_87_seeded_stream + { + protected override void Because() + { + // Seed the legacy on-disk format directly, independently of the current delete implementation. + Database.GetCollection("Commits").UpdateMany( + Builders.Filter.Eq(MongoCommitFields.BucketId, Bucket), + Builders.Update.Set(MongoCommitFields.BucketId, MongoSystemBuckets.RecycleBin)); + } + + [Fact] + public async Task Original_stream_identity_and_checkpoints_are_preserved() + { + IEnumerable commits; + if (Async) + { + var observer = new CommitStreamObserver(); + await Engine.GetDeletedCommitsAsync(observer, CancellationToken.None); + commits = observer.Commits; + } + else + commits = Engine.GetDeletedCommits(); + commits.Should().OnlyContain(commit => commit.StreamId == Stream && commit.BucketId == MongoSystemBuckets.RecycleBin); + commits.Select(commit => commit.CheckpointToken).Should().Equal(1L, 2L); + } + + [Fact] + public async Task Recycle_bucket_can_be_read_by_original_stream_and_revision_range() + { + IEnumerable commits; + if (Async) + { + var observer = new CommitStreamObserver(); + await Engine.GetFromAsync(MongoSystemBuckets.RecycleBin, Stream, 3, 3, observer, CancellationToken.None); + commits = observer.Commits; + } + else + commits = Engine.GetFrom(MongoSystemBuckets.RecycleBin, Stream, 3, 3); + commits.Select(commit => commit.CheckpointToken).Should().Equal(2L); + } + } + + public class Issue_87_when_reading_legacy_recycled_commits_async : Issue_87_when_reading_legacy_recycled_commits + { + protected override bool Async => true; + } + + public class Issue_87_when_cleaning_legacy_recycled_commits : Issue_87_seeded_stream + { + protected override async Task BecauseAsync() + { + Database.GetCollection("Commits").UpdateMany( + Builders.Filter.Eq(MongoCommitFields.BucketId, Bucket), + Builders.Update.Set(MongoCommitFields.BucketId, MongoSystemBuckets.RecycleBin)); + Engine.CommitMany(1, "live-stream", "live-bucket"); + if (Async) + await Engine.EmptyRecycleBinAsync(CancellationToken.None); + else + Engine.EmptyRecycleBin(); + } + + [Fact] + public void Cleanup_removes_legacy_recycled_documents_without_touching_live_commits() + { + Engine.GetDeletedCommits().Should().BeEmpty(); + Engine.GetFrom(0).Select(commit => commit.CheckpointToken).Should().Equal(3L); + } + } + + public class Issue_87_when_cleaning_legacy_recycled_commits_async : Issue_87_when_cleaning_legacy_recycled_commits + { + protected override bool Async => true; + } +} From 01b30a1f4b29fc6af797ce7dbe8811536a2eb386 Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Fri, 23 Oct 2026 16:02:20 +0200 Subject: [PATCH 2/5] fix: preserve independent recycled stream identities (#87) Use per-attempt stored identities with original bucket and stream metadata, recycle before metadata cleanup, and preserve legacy/public read semantics. Document recovery limits and follow-up #96. Co-authored by Alessandro Giorgetti and Codex (OpenAI). --- docs/Recycle-Bin.md | 78 ++++++++ .../Issues/Issue87PartialRecycleRetry.cs | 168 ++++++++++++++++ .../Issues/Issue87RecycledIdentity.cs | 180 ++++++++++++++++++ .../ExtensionMethods.cs | 4 +- .../MongoFields.cs | 8 + .../MongoPersistenceEngine.Async.cs | 26 ++- .../MongoPersistenceEngine.cs | 36 +++- 7 files changed, 480 insertions(+), 20 deletions(-) create mode 100644 docs/Recycle-Bin.md create mode 100644 src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87PartialRecycleRetry.cs create mode 100644 src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87RecycledIdentity.cs diff --git a/docs/Recycle-Bin.md b/docs/Recycle-Bin.md new file mode 100644 index 0000000..3f43029 --- /dev/null +++ b/docs/Recycle-Bin.md @@ -0,0 +1,78 @@ +# Recycle-bin identity and deletion + +`DeleteStream` and `DeleteStreamAsync` recycle commits in the existing `Commits` +collection. They do not move documents to another collection or rebuild indexes. + +## Stored and public identities + +Each deletion attempt assigns one random `Guid.NewGuid().ToString("N")` value to +the stored `StreamId` of its matching commits, changes `BucketId` to `:rb`, and +preserves the source identity in `OriginalBucketId` and `OriginalStreamId`. +Different buckets and recreated streams can therefore use the same original +stream ID without sharing recycle-bin unique-index keys. The existing unique +indexes still reject a conflicting generated key rather than overwrite data. + +Commit checkpoint (`_id`), commit ID, revisions, sequence, headers and events are +unchanged. The generated value is an internal deletion-attempt identifier, not a +predictable sequence, a timestamp, or a durable identifier for an entire stream +lifetime. + +The adapter returns the original `StreamId` for recycled commits. Their public +`BucketId` remains `:rb`; `OriginalBucketId` is retained in BSON, not exposed as +the public bucket. `GetFrom(":rb", originalStreamId, minRevision, maxRevision)` +and its async equivalent search both new metadata and legacy stored IDs. This +lookup combines matching histories across original buckets and deletion attempts; +it does not select one historical stream incarnation. + +## Existing data and rollout + +Existing recycled documents without `OriginalStreamId` are read using their +stored `StreamId`. No backfill is required. Their original bucket was already +discarded by the old deletion implementation and cannot be inferred as `default`. +Live documents ignore optional recycled-identity metadata. + +Old adapter versions tolerate the extra BSON fields, but expose the generated +stored ID when reading new recycled documents. Their original-stream lookup +does not find those new documents. Raw BSON consumers must also account for the +new identity fields. Upgrade recycle-bin readers before creating new-format +deleted data; this is not unrestricted mixed-version compatibility. + +All existing indexes and checkpoint-based cleanup remain unchanged. +`EmptyRecycleBin` and its async equivalent remove recycled commits older than the +greatest stored checkpoint, retaining that checkpoint if it belongs to a recycled +commit. Original-ID lookups now use an additional predicate; their performance +has not been benchmarked. + +## Failure and concurrency limits + +Recycling is performed before deleting the stream head and snapshots. With the +default acknowledged commit write concern, a rejected recycle update propagates +an error before metadata cleanup begins. `UpdateMany` is not atomic across +documents: earlier commits may already be recycled when a later update fails. +An explicit retry matches the remaining live commits only and assigns a new +generated ID. Already recycled documents keep their stored identity; public +readers combine the groups using `OriginalStreamId`. + +There is no automatic recovery job or persisted deletion-operation token. Tests +cover retry from a seeded partial state, not process termination/restart or +concurrent deletion and recreation. A delayed retry can match newly recreated +commits because the API identifies only a bucket and stream, not an incarnation. + +For controlled deletion, stop commits, snapshots and recreation for that stream, +drain outstanding/background metadata writes, and keep that boundary in place +through failures and retries. Readers requiring all-or-nothing visibility must +also be excluded during this window. Reconcile remaining live commits, stream +heads and snapshots before allowing recreation. Default head/snapshot writes are +unacknowledged, so successful return alone is not a verified cleanup barrier; +use acknowledged metadata settings and verify the resulting state when this +guarantee is required. Overriding commit writes to be unacknowledged also removes +the acknowledged-recycling guarantee. + +This change fixes recycle-key collisions and cleanup ordering. It does not add +transactions, writer fencing, atomic stream lifecycle operations, or stronger +write durability. Those require a separate lifecycle/API design. + +The broader lifecycle work is tracked in [#96](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/96). +Existing asynchronous cancellation behavior is tracked separately in +[#86](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/86); +this identity change does not fix or strengthen that behavior. diff --git a/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87PartialRecycleRetry.cs b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87PartialRecycleRetry.cs new file mode 100644 index 0000000..72468ed --- /dev/null +++ b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87PartialRecycleRetry.cs @@ -0,0 +1,168 @@ +using FluentAssertions; +using global::MongoDB.Bson; +using global::MongoDB.Driver; +using NEventStore.Persistence.AcceptanceTests.BDD; +using NEventStore.Persistence.MongoDB.Support; + +namespace NEventStore.Persistence.MongoDB.Tests.AcceptanceTests.Issues +{ + public class Issue_87_when_retrying_a_partially_recycled_stream : Issue_87_seeded_stream + { + private Exception? _error; + private List _original = null!; + private List _afterFailure = null!; + private List _afterRetry = null!; + private long _headsAfterFailure; + private long _snapshotsAfterFailure; + private ICommit[] _deleted = null!; + private ICommit[] _byOriginalIdentity = null!; + + protected override void Context() + { + base.Context(); + _original = ReadDocuments(); + // Seed the persisted state of an interrupted multi-document update explicitly. + // UpdateMany does not guarantee visit order, so a validator alone cannot reliably + // make the first commit succeed and the second fail on every MongoDB query plan. + Database.GetCollection("Commits").UpdateOne( + new BsonDocument(MongoCommitFields.CommitSequence, 1), + Builders.Update.Set(MongoCommitFields.BucketId, MongoSystemBuckets.RecycleBin) + .Set(MongoCommitFields.OriginalBucketId, Bucket) + .Set(MongoCommitFields.OriginalStreamId, Stream) + .Set(MongoCommitFields.StreamId, Guid.NewGuid().ToString("N"))); + Count("Streams", MongoStreamHeadFields.FullQualifiedBucketId, Bucket).Should().Be(1); + Count("Snapshots", MongoSnapshotFields.FullQualifiedBucketId, Bucket).Should().Be(1); + Database.RunCommand(new BsonDocument + { + { "collMod", "Commits" }, + { "validator", new BsonDocument(MongoCommitFields.BucketId, + new BsonDocument("$ne", MongoSystemBuckets.RecycleBin)) }, + { "validationLevel", "strict" }, + { "validationAction", "error" } + }); + } + + protected override async Task BecauseAsync() + { + try { await Delete(); } + catch (Exception error) { _error = error; } + + // Preserve the actual failed-operation state before retrying. Facts never mutate it. + _afterFailure = ReadDocuments(); + _headsAfterFailure = Count("Streams", MongoStreamHeadFields.FullQualifiedBucketId, Bucket); + _snapshotsAfterFailure = Count("Snapshots", MongoSnapshotFields.FullQualifiedBucketId, Bucket); + Database.RunCommand(new BsonDocument + { + { "collMod", "Commits" }, + { "validator", new BsonDocument() } + }); + await Delete(); + _afterRetry = ReadDocuments(); + if (Async) + { + var observer = new CommitStreamObserver(); + await Engine.GetDeletedCommitsAsync(observer, CancellationToken.None); + _deleted = observer.Commits.ToArray(); + observer = new CommitStreamObserver(); + await Engine.GetFromAsync(MongoSystemBuckets.RecycleBin, Stream, 1, 4, observer, CancellationToken.None); + _byOriginalIdentity = observer.Commits.ToArray(); + } + else + { + _deleted = Engine.GetDeletedCommits().ToArray(); + _byOriginalIdentity = Engine.GetFrom(MongoSystemBuckets.RecycleBin, Stream, 1, 4).ToArray(); + } + } + + [Fact] + public void Failure_while_recycling_the_remaining_commit_is_propagated() + { + var cause = _error.Should().BeOfType().Which.InnerException; + cause.Should().BeOfType().Which.WriteError.Code.Should().Be(121); + } + + [Fact] + public void Rejected_retry_retains_one_recycled_commit_and_one_live_commit() + { + _afterFailure.Should().HaveCount(2); + _afterFailure.Select(document => document[MongoCommitFields.BucketId].AsString) + .Should().Equal(MongoSystemBuckets.RecycleBin, Bucket); + _afterFailure.Select(document => document[MongoCommitFields.CommitSequence].AsInt32) + .Should().Equal(1, 2); + } + + [Fact] + public void Failure_retains_the_live_head_and_snapshot_for_retry() + { + _headsAfterFailure.Should().Be(1); + _snapshotsAfterFailure.Should().Be(1); + } + + [Fact] + public void Failure_and_retry_preserve_checkpoints_events_and_other_commit_data() + { + foreach (var documents in new[] { _afterFailure, _afterRetry }) + { + documents.Select(document => document[MongoCommitFields.CheckpointNumber].AsInt64).Should().Equal(1L, 2L); + for (var index = 0; index < _original.Count; index++) + foreach (var element in _original[index].Elements.Where(element => + element.Name != MongoCommitFields.BucketId && element.Name != MongoCommitFields.StreamId)) + documents[index][element.Name].Should().Be(element.Value, "recycling must preserve {0}", element.Name); + } + } + + [Fact] + public void Retry_recycles_each_commit_once_and_removes_live_metadata() + { + _afterRetry.Should().HaveCount(2); + _afterRetry.Should().OnlyContain(document => document[MongoCommitFields.BucketId].AsString == MongoSystemBuckets.RecycleBin); + Count("Commits", MongoCommitFields.BucketId, Bucket).Should().Be(0); + Engine.GetFrom(0).Should().BeEmpty(); + Count("Streams", MongoStreamHeadFields.FullQualifiedBucketId, Bucket).Should().Be(0); + Count("Snapshots", MongoSnapshotFields.FullQualifiedBucketId, Bucket).Should().Be(0); + } + + [Fact] + public void Public_readers_rejoin_the_original_stream_with_its_original_checkpoints() + { + foreach (var commits in new[] { _deleted, _byOriginalIdentity }) + { + commits.Select(commit => commit.CheckpointToken).Should().Equal(1L, 2L); + commits.Should().OnlyContain(commit => commit.StreamId == Stream && commit.BucketId == MongoSystemBuckets.RecycleBin); + commits.Select(commit => commit.CommitSequence).Should().Equal(1, 2); + } + } + + [Fact] + public void Retry_preserves_the_already_recycled_document() + { + _afterRetry[0].Equals(_afterFailure[0]).Should().BeTrue("retry must not rewrite an already recycled commit"); + } + + [Fact] + public void Retry_uses_separate_stored_groups_but_preserves_original_identity() + { + _afterRetry.Select(document => document[MongoCommitFields.StreamId].AsString) + .Should().OnlyHaveUniqueItems().And.NotContain(Stream); + _afterRetry.Should().OnlyContain(document => + document[MongoCommitFields.OriginalStreamId].AsString == Stream && + document[MongoCommitFields.OriginalBucketId].AsString == Bucket); + } + + private List ReadDocuments() => Database.GetCollection("Commits") + .Find(new BsonDocument()).Sort(new BsonDocument(MongoCommitFields.CheckpointNumber, 1)).ToList(); + + private Task Delete() + { + if (Async) + return Engine.DeleteStreamAsync(Bucket, Stream, CancellationToken.None); + Engine.DeleteStream(Bucket, Stream); + return Task.CompletedTask; + } + } + + public class Issue_87_when_retrying_a_partially_recycled_stream_async : Issue_87_when_retrying_a_partially_recycled_stream + { + protected override bool Async => true; + } +} diff --git a/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87RecycledIdentity.cs b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87RecycledIdentity.cs new file mode 100644 index 0000000..0f84fcf --- /dev/null +++ b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87RecycledIdentity.cs @@ -0,0 +1,180 @@ +using FluentAssertions; +using global::MongoDB.Bson; +using global::MongoDB.Bson.Serialization; +using global::MongoDB.Driver; +using NEventStore.Persistence.AcceptanceTests; +using NEventStore.Persistence.AcceptanceTests.BDD; +using NEventStore.Persistence.MongoDB.Support; + +namespace NEventStore.Persistence.MongoDB.Tests.AcceptanceTests.Issues +{ + public class Issue_87_when_reading_mixed_recycled_identities : Issue_87_seeded_stream + { + protected override async Task BecauseAsync() + { + Database.GetCollection("Commits").UpdateMany( + new BsonDocument(MongoCommitFields.BucketId, Bucket), + Builders.Update.Set(MongoCommitFields.BucketId, MongoSystemBuckets.RecycleBin)); + Engine.CommitMany(2, Stream, "another-bucket"); + if (Async) + await Engine.DeleteStreamAsync("another-bucket", Stream, CancellationToken.None); + else + Engine.DeleteStream("another-bucket", Stream); + Engine.CommitMany(1, "live-stream", "live-bucket"); + // Optional metadata must never override identity on a live document. + Database.GetCollection("Commits").UpdateMany( + new BsonDocument(MongoCommitFields.BucketId, "live-bucket"), + Builders.Update.Set(MongoCommitFields.OriginalStreamId, Stream)); + } + + [Fact] + public async Task Original_identity_reads_combine_new_and_legacy_documents() + { + IEnumerable commits; + if (Async) + { + var observer = new CommitStreamObserver(); + await Engine.GetFromAsync(MongoSystemBuckets.RecycleBin, Stream, 3, 3, observer, CancellationToken.None); + commits = observer.Commits; + } + else + commits = Engine.GetFrom(MongoSystemBuckets.RecycleBin, Stream, 3, 3); + commits.Select(c => c.CheckpointToken).Should().BeEquivalentTo(new[] { 2L, 4L }); + commits.Should().OnlyContain(c => c.StreamId == Stream && c.BucketId == MongoSystemBuckets.RecycleBin); + } + + [Fact] + public async Task Checkpoint_and_deleted_readers_preserve_original_identity() + { + var commits = Engine.GetDeletedCommits().ToArray(); + commits.Select(c => c.CheckpointToken).Should().Equal(1L, 2L, 3L, 4L); + commits.Should().OnlyContain(c => c.StreamId == Stream && c.BucketId == MongoSystemBuckets.RecycleBin); + Engine.GetFrom(MongoSystemBuckets.RecycleBin, 0).Select(c => c.StreamId).Should().Equal(Enumerable.Repeat(Stream, 4)); + Engine.GetFromTo(MongoSystemBuckets.RecycleBin, 1L, 3L).Select(c => c.StreamId).Should().Equal(Stream, Stream); + var observer = new CommitStreamObserver(); + await Engine.GetFromAsync(MongoSystemBuckets.RecycleBin, 0L, observer, CancellationToken.None); + observer.Commits.Select(c => c.StreamId).Should().Equal(Enumerable.Repeat(Stream, 4)); + observer = new CommitStreamObserver(); + await Engine.GetFromToAsync(MongoSystemBuckets.RecycleBin, 1L, 3L, observer, CancellationToken.None); + observer.Commits.Select(c => c.StreamId).Should().Equal(Stream, Stream); + observer = new CommitStreamObserver(); + await Engine.GetDeletedCommitsAsync(observer, CancellationToken.None); + observer.Commits.Select(c => c.StreamId).Should().Equal(Enumerable.Repeat(Stream, 4)); +#pragma warning disable CS0618 + Engine.GetFrom(MongoSystemBuckets.RecycleBin, DateTime.UtcNow.AddDays(-1)).Select(c => c.StreamId).Should().Equal(Enumerable.Repeat(Stream, 4)); + Engine.GetFromTo(MongoSystemBuckets.RecycleBin, DateTime.UtcNow.AddDays(-1), DateTime.UtcNow.AddDays(1)).Select(c => c.StreamId).Should().Equal(Enumerable.Repeat(Stream, 4)); +#pragma warning restore CS0618 + } + + [Fact] + public void Live_documents_do_not_use_optional_recycled_identity() + { + Engine.GetFrom("live-bucket", Stream, 0, int.MaxValue).Should().BeEmpty(); + Engine.GetFrom("live-bucket", "live-stream", 0, int.MaxValue).Single().StreamId.Should().Be("live-stream"); + Engine.GetFrom(0).Single().StreamId.Should().Be("live-stream"); + } + + [Fact] + public void Raw_and_legacy_readers_expose_synthetic_identity_only_for_new_recycled_documents() + { + var documents = Database.GetCollection("Commits").Find(new BsonDocument()).Sort(new BsonDocument("_id", 1)).ToList(); + documents[0].Contains(MongoCommitFields.OriginalStreamId).Should().BeFalse(); + documents[2][MongoCommitFields.OriginalStreamId].AsString.Should().Be(Stream); + documents[2][MongoCommitFields.OriginalBucketId].AsString.Should().Be("another-bucket"); + documents[2][MongoCommitFields.StreamId].Should().Be(documents[3][MongoCommitFields.StreamId]); + BsonSerializer.Deserialize(documents[2]).StreamId.Should().NotBe(Stream); + } + } + + public class Issue_87_when_reading_mixed_recycled_identities_async : Issue_87_when_reading_mixed_recycled_identities + { + protected override bool Async => true; + } + + public class Issue_87_when_cleaning_mixed_recycled_identities : Issue_87_seeded_stream + { + protected override async Task ContextAsync() + { + Database.GetCollection("Commits").UpdateMany( + new BsonDocument(MongoCommitFields.BucketId, Bucket), + Builders.Update.Set(MongoCommitFields.BucketId, MongoSystemBuckets.RecycleBin)); + Engine.CommitMany(1, "live-stream", "live-bucket"); + Engine.CommitMany(2, Stream, "another-bucket"); + if (Async) + await Engine.DeleteStreamAsync("another-bucket", Stream, CancellationToken.None); + else + Engine.DeleteStream("another-bucket", Stream); + } + + protected override async Task BecauseAsync() + { + if (Async) + await Engine.EmptyRecycleBinAsync(CancellationToken.None); + else + Engine.EmptyRecycleBin(); + } + + [Fact] + public void Cleanup_removes_earlier_legacy_and_new_recycled_documents() + { + var documents = Database.GetCollection("Commits").Find(new BsonDocument()) + .Sort(new BsonDocument(MongoCommitFields.CheckpointNumber, 1)).ToList(); + documents.Select(document => document[MongoCommitFields.CheckpointNumber].AsInt64).Should().Equal(3L, 5L); + } + + [Fact] + public void Latest_recycled_commit_preserves_checkpoint_numbering_and_original_identity() + { + var commit = Engine.GetDeletedCommits().Should().ContainSingle().Which; + commit.CheckpointToken.Should().Be(5L); + commit.StreamId.Should().Be(Stream); + commit.BucketId.Should().Be(MongoSystemBuckets.RecycleBin); + } + + [Fact] + public void Cleanup_leaves_the_live_commit_unchanged() + { + var commit = Engine.GetFrom(0).Should().ContainSingle().Which; + commit.CheckpointToken.Should().Be(3L); + commit.StreamId.Should().Be("live-stream"); + commit.BucketId.Should().Be("live-bucket"); + } + } + + public class Issue_87_when_cleaning_mixed_recycled_identities_async : Issue_87_when_cleaning_mixed_recycled_identities + { + protected override bool Async => true; + } + + public class Issue_87_when_retrying_a_rejected_recycle : Issue_87_seeded_stream + { + protected override async Task BecauseAsync() + { + Database.RunCommand(new BsonDocument { { "collMod", "Commits" }, + { "validator", new BsonDocument(MongoCommitFields.BucketId, new BsonDocument("$ne", MongoSystemBuckets.RecycleBin)) } }); + try + { + if (Async) await Engine.DeleteStreamAsync(Bucket, Stream, CancellationToken.None); + else Engine.DeleteStream(Bucket, Stream); + } + catch (StorageException) { } + Database.RunCommand(new BsonDocument { { "collMod", "Commits" }, { "validator", new BsonDocument() } }); + if (Async) await Engine.DeleteStreamAsync(Bucket, Stream, CancellationToken.None); + else Engine.DeleteStream(Bucket, Stream); + } + + [Fact] + public void Retry_recycles_each_commit_once_and_removes_metadata() + { + Engine.GetDeletedCommits().Select(c => c.CheckpointToken).Should().Equal(1L, 2L); + Engine.GetDeletedCommits().Should().OnlyContain(c => c.StreamId == Stream); + Count("Streams", MongoStreamHeadFields.FullQualifiedBucketId, Bucket).Should().Be(0); + Count("Snapshots", MongoSnapshotFields.FullQualifiedBucketId, Bucket).Should().Be(0); + } + } + + public class Issue_87_when_retrying_a_rejected_recycle_async : Issue_87_when_retrying_a_rejected_recycle + { + protected override bool Async => true; + } +} diff --git a/src/NEventStore.Persistence.MongoDB/ExtensionMethods.cs b/src/NEventStore.Persistence.MongoDB/ExtensionMethods.cs index a758ee0..fd61aeb 100644 --- a/src/NEventStore.Persistence.MongoDB/ExtensionMethods.cs +++ b/src/NEventStore.Persistence.MongoDB/ExtensionMethods.cs @@ -90,7 +90,9 @@ public static ICommit ToCommit(this BsonDocument doc, IDocumentSerializer serial } return new Commit(mc.BucketId, - mc.StreamId, + mc.BucketId == MongoSystemBuckets.RecycleBin && + doc.TryGetValue(MongoCommitFields.OriginalStreamId, out var originalStreamId) + ? originalStreamId.AsString : mc.StreamId, mc.StreamRevisionTo, mc.CommitId, mc.CommitSequence, diff --git a/src/NEventStore.Persistence.MongoDB/MongoFields.cs b/src/NEventStore.Persistence.MongoDB/MongoFields.cs index 479bf99..7333fb5 100644 --- a/src/NEventStore.Persistence.MongoDB/MongoFields.cs +++ b/src/NEventStore.Persistence.MongoDB/MongoFields.cs @@ -115,6 +115,14 @@ public static class MongoCommitFields /// public const string StreamId = "StreamId"; /// + /// The original stream identifier, present on newly recycled commits only. + /// + public const string OriginalStreamId = "OriginalStreamId"; + /// + /// The original bucket identifier, present on newly recycled commits only. + /// + public const string OriginalBucketId = "OriginalBucketId"; + /// /// The stream revision field. /// public const string StreamRevision = "StreamRevision"; diff --git a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs index 4f58b7e..b1ca6b9 100644 --- a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs +++ b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.Async.cs @@ -20,7 +20,7 @@ public Task GetFromAsync(string bucketId, string streamId, int minRevision, int var filters = new List>() { Builders.Filter.Eq(MongoCommitFields.BucketId, bucketId), - Builders.Filter.Eq(MongoCommitFields.StreamId, streamId) + GetStreamReadFilter(bucketId, streamId) }; if (minRevision > 0) { @@ -489,6 +489,21 @@ public Task DeleteStreamAsync(string bucketId, string streamId, CancellationToke return TryMongoAsync(async () => { + // Recycle acknowledged commits before removing reconstructible metadata. + // UpdateMany is not atomic across documents; partial failures still require retry. + await PersistedCommits.UpdateManyAsync( + Builders.Filter.And( + Builders.Filter.Eq(MongoCommitFields.BucketId, bucketId), + Builders.Filter.Eq(MongoCommitFields.StreamId, streamId) + ), + Builders.Update + .Set(MongoCommitFields.BucketId, MongoSystemBuckets.RecycleBin) + .Set(MongoCommitFields.OriginalBucketId, bucketId) + .Set(MongoCommitFields.OriginalStreamId, streamId) + .Set(MongoCommitFields.StreamId, Guid.NewGuid().ToString("N")), + cancellationToken: cancellationToken + ).ConfigureAwait(false); + await PersistedStreamHeads.DeleteOneAsync( Builders.Filter.Eq(MongoStreamHeadFields.Id, new BsonDocument{ {MongoStreamHeadFields.BucketId, bucketId}, @@ -504,15 +519,6 @@ await PersistedSnapshots.DeleteManyAsync( ), cancellationToken ).ConfigureAwait(false); - - await PersistedCommits.UpdateManyAsync( - Builders.Filter.And( - Builders.Filter.Eq(MongoCommitFields.BucketId, bucketId), - Builders.Filter.Eq(MongoCommitFields.StreamId, streamId) - ), - Builders.Update.Set(MongoCommitFields.BucketId, MongoSystemBuckets.RecycleBin), - cancellationToken: cancellationToken - ).ConfigureAwait(false); }, cancellationToken); } diff --git a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs index c59e2c2..c05816c 100644 --- a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs +++ b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs @@ -205,6 +205,18 @@ public virtual void Initialize() }); } + private static FilterDefinition GetStreamReadFilter(string bucketId, string streamId) + { + var filter = Builders.Filter; + var storedIdentity = filter.Eq(MongoCommitFields.StreamId, streamId); + if (bucketId != MongoSystemBuckets.RecycleBin) + return storedIdentity; + + return filter.Or( + filter.Eq(MongoCommitFields.OriginalStreamId, streamId), + filter.And(filter.Exists(MongoCommitFields.OriginalStreamId, false), storedIdentity)); + } + /// public virtual IEnumerable GetFrom(string bucketId, string streamId, int minRevision, int maxRevision) { @@ -215,7 +227,7 @@ public virtual IEnumerable GetFrom(string bucketId, string streamId, in var filters = new List>() { Builders.Filter.Eq(MongoCommitFields.BucketId, bucketId), - Builders.Filter.Eq(MongoCommitFields.StreamId, streamId) + GetStreamReadFilter(bucketId, streamId) }; if (minRevision > 0) { @@ -650,6 +662,20 @@ public void DeleteStream(string bucketId, string streamId) TryMongo(() => { + // Recycle acknowledged commits before removing reconstructible metadata. + // UpdateMany is not atomic across documents; partial failures still require retry. + PersistedCommits.UpdateMany( + Builders.Filter.And( + Builders.Filter.Eq(MongoCommitFields.BucketId, bucketId), + Builders.Filter.Eq(MongoCommitFields.StreamId, streamId) + ), + Builders.Update + .Set(MongoCommitFields.BucketId, MongoSystemBuckets.RecycleBin) + .Set(MongoCommitFields.OriginalBucketId, bucketId) + .Set(MongoCommitFields.OriginalStreamId, streamId) + .Set(MongoCommitFields.StreamId, Guid.NewGuid().ToString("N")) + ); + PersistedStreamHeads.DeleteOne( Builders.Filter.Eq(MongoStreamHeadFields.Id, new BsonDocument{ {MongoStreamHeadFields.BucketId, bucketId}, @@ -663,14 +689,6 @@ public void DeleteStream(string bucketId, string streamId) Builders.Filter.Eq(MongoSnapshotFields.FullQualifiedStreamId, streamId) ) ); - - PersistedCommits.UpdateMany( - Builders.Filter.And( - Builders.Filter.Eq(MongoCommitFields.BucketId, bucketId), - Builders.Filter.Eq(MongoCommitFields.StreamId, streamId) - ), - Builders.Update.Set(MongoCommitFields.BucketId, MongoSystemBuckets.RecycleBin) - ); }); } From c85d2f21ed2850aa502f348ebe83036120c6356b Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Mon, 26 Oct 2026 11:20:15 +0100 Subject: [PATCH 3/5] docs: update changelog for #87 --- Changelog.md | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/Changelog.md b/Changelog.md index 80a4cca..43c3f8d 100644 --- a/Changelog.md +++ b/Changelog.md @@ -1,5 +1,13 @@ # NEventStore.Persistence.MongoDB +## Unreleased + +- Fix recycle-bin identity collisions by assigning each deletion attempt a synthetic stored `StreamId` while preserving `OriginalBucketId` and `OriginalStreamId`; new reads restore the original stream identity, and legacy recycled records remain readable. [#87](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/87) + +### Breaking Changes + +- 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) + ## 12.1.0 - Added explicit support for net8.0, net9.0, net10.0. From 4f5ca52a430c8d07a735b1b254bf75e2cc7a52b9 Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Mon, 26 Oct 2026 12:02:35 +0100 Subject: [PATCH 4/5] test: reproduce unselective recycle original-ID reads (#87) Capture actual sync/async find filters and sorts over 18,280 mixed commits. Baseline MongoDB 7.0.45/net10.0: 2 expected failures, 0 skips. Recycle lookups examine all 12,240 recycled keys/documents and use blocking SORT; result semantics remain correct. Co-authored by Alessandro Giorgetti and Codex (OpenAI). --- .../Issues/Issue87RecycleIndexes.cs | 201 ++++++++++++++++++ 1 file changed, 201 insertions(+) create mode 100644 src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87RecycleIndexes.cs diff --git a/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87RecycleIndexes.cs b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87RecycleIndexes.cs new file mode 100644 index 0000000..73126ac --- /dev/null +++ b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87RecycleIndexes.cs @@ -0,0 +1,201 @@ +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 class Issue_87_recycle_reads_should_remain_selective : SpecificationBase + { + private const string Stream = "requested-stream"; + private readonly ConcurrentQueue _finds = new(); + private readonly List _documents = new(); + private readonly List<(string Description, BsonDocument Plan)> _plans = new(); + private MongoPersistenceEngine _engine = null!; + private IMongoDatabase _database = null!; + private MongoPersistenceOptions _options = null!; + + protected override void Context() + { + _ = new AcceptanceTestMongoPersistenceFactory(); + _options = new MongoPersistenceOptions + { + DisableSnapshotSupport = true, + ConfigureClientSettingsAction = settings => settings.ClusterConfigurator = cluster => + cluster.Subscribe(command => + { + if (command.CommandName == "find") + _finds.Enqueue(command.Command.DeepClone().AsBsonDocument); + }) + }; + var url = new MongoUrlBuilder(AcceptanceTestMongoPersistenceFactory.GetConnectionString()) + { + DatabaseName = $"issue87-indexes-{Guid.NewGuid():N}" + }; + _database = _options.ConnectToDatabase(url.ToString()); + _engine = new MongoPersistenceEngine(_database, new DocumentObjectSerializer(), _options); + _engine.Initialize(); + + // One legacy history and three new deletion attempts, including two from the + // same original bucket. Original identities/revisions deliberately overlap. + AddHistory(MongoSystemBuckets.RecycleBin, Stream, null, null); + for (int lifetime = 0; lifetime < 3; lifetime++) + AddHistory(MongoSystemBuckets.RecycleBin, $"deleted-{lifetime}", Stream, $"bucket-{lifetime % 2}"); + AddHistory(MongoSystemBuckets.RecycleBin, "new-only-token", "new-only", "bucket-0"); + AddHistory(MongoSystemBuckets.RecycleBin, "legacy-only", null, null); + AddHistory("bucket-0", Stream, null, null); + + // Enough unrelated new/legacy recycled and live data to expose a bucket scan. + for (int i = 0; i < 6000; i++) + { + AddDocument(MongoSystemBuckets.RecycleBin, $"new-token-{i}", $"unrelated-{i}", $"bucket-{i % 3}", 1); + AddDocument(MongoSystemBuckets.RecycleBin, $"legacy-{i}", null, null, 1); + AddDocument($"bucket-{i % 3}", $"live-{i}", null, null, 1); + } + _database.GetCollection("Commits").InsertMany(_documents); + } + + protected override async Task BecauseAsync() + { + foreach (var range in new[] { (0, int.MaxValue), (32, 49), (31, int.MaxValue), (0, 50), (100, 110) }) + await ReadAndExplain(MongoSystemBuckets.RecycleBin, Stream, range.Item1, range.Item2); + await ReadAndExplain(MongoSystemBuckets.RecycleBin, "new-only", 31, 50); + await ReadAndExplain(MongoSystemBuckets.RecycleBin, "legacy-only", 31, 50); + await ReadAndExplain(MongoSystemBuckets.RecycleBin, "absent-stream", 0, int.MaxValue); + await ReadAndExplain("bucket-0", Stream, 31, 50); + } + + [Fact] + public void Full_sync_and_async_queries_examine_only_the_requested_histories() + { + foreach (var (description, plan) in _plans) + Console.WriteLine($"{description}: {plan.ToJson()}"); + foreach (var (description, plan) in _plans) + { + var stats = plan["executionStats"]; + stats["executionSuccess"].AsBoolean.Should().BeTrue(description); + // Allow boundary seeks and FETCH of legacy candidates, not thousands of + // unrelated recycled records. Do not couple this to a specific plan name. + stats["totalDocsExamined"].ToInt64().Should().BeLessThanOrEqualTo(200, description); + stats["totalKeysExamined"].ToInt64().Should().BeLessThanOrEqualTo(400, description); + var stages = Walk(plan["queryPlanner"]["winningPlan"]) + .Where(node => node.Contains("stage")).Select(node => node["stage"].AsString); + stages.Should().NotContain("COLLSCAN", description); + stages.Should().NotContain("SORT", description + ": revision order should come from the indexes"); + } + } + + [Fact] + public void Both_recycle_identity_branches_have_equality_index_bounds() + { + foreach (var (_, plan) in _plans.Take(2)) + { + var bounds = Walk(plan["queryPlanner"]["winningPlan"]) + .Where(node => node.Contains("indexBounds")) + .Select(node => node["indexBounds"].AsBsonDocument).ToList(); + foreach (var field in new[] { MongoCommitFields.OriginalStreamId, MongoCommitFields.StreamId }) + bounds.Should().Contain(bound => bound.Contains(field) && + bound[field].AsBsonArray.Contains(new BsonString($"[\"{Stream}\", \"{Stream}\"]")), + $"the {field} branch must seek the requested identity, not scan the recycle bucket"); + } + } + + private async Task ReadAndExplain(string bucket, string stream, int min, int max) + { + var expected = _documents.Where(document => document[MongoCommitFields.BucketId] == bucket && + (bucket == MongoSystemBuckets.RecycleBin && document.Contains(MongoCommitFields.OriginalStreamId) + ? document[MongoCommitFields.OriginalStreamId] == stream : document[MongoCommitFields.StreamId] == stream) && + document[MongoCommitFields.StreamRevisionTo].AsInt32 >= min && + document[MongoCommitFields.StreamRevisionFrom].AsInt32 <= max) + .Select(document => document[MongoCommitFields.CheckpointNumber].ToInt64()).ToList(); + BsonDocument? syncFind = null; + foreach (bool async in new[] { false, true }) + { + while (_finds.TryDequeue(out _)) { } + IList commits; + if (async) + { + var observer = new CommitStreamObserver(); + await _engine.GetFromAsync(bucket, stream, min, max, observer, CancellationToken.None); + commits = observer.Commits.ToList(); + } + else + commits = _engine.GetFrom(bucket, stream, min, max).ToList(); + + commits.Select(commit => commit.CheckpointToken).Should().BeEquivalentTo(expected); + commits.Should().OnlyContain(commit => commit.StreamId == stream && commit.BucketId == bucket); + commits.Select(commit => commit.StreamRevision).Should().BeInAscendingOrder(); + _finds.Count.Should().Be(1, "explain must use the actual complete engine query"); + _finds.TryDequeue(out var command).Should().BeTrue(); + var find = new BsonDocument + { + { "find", command!["find"] }, { "filter", command["filter"] }, { "sort", command["sort"] } + }; + find["sort"].Should().Be(new BsonDocument(MongoCommitFields.StreamRevisionFrom, 1)); + if (async) + find.Equals(syncFind).Should().BeTrue("sync and async must send equivalent filters and sorts"); + else + syncFind = find; + + var plan = await _database.RunCommandAsync(new BsonDocument + { + { "explain", find }, { "verbosity", "executionStats" } + }); + plan["executionStats"]["nReturned"].ToInt32().Should().Be(expected.Count); + var description = $"{bucket}/{stream} [{min}, {max}], async={async}"; + _plans.Add((description, plan)); + } + } + + private void AddHistory(string bucket, string storedId, string? originalId, string? originalBucket) + { + for (int sequence = 1; sequence <= 40; sequence++) + AddDocument(bucket, storedId, originalId, originalBucket, sequence); + } + + private void AddDocument(string bucket, string storedId, string? originalId, string? originalBucket, int sequence) + { + var document = new BsonDocument + { + { MongoCommitFields.CheckpointNumber, (long)_documents.Count + 1 }, + { MongoCommitFields.BucketId, bucket }, { MongoCommitFields.StreamId, storedId }, + { MongoCommitFields.StreamRevisionFrom, sequence * 2 - 1 }, + { MongoCommitFields.StreamRevisionTo, sequence * 2 }, + { MongoCommitFields.CommitSequence, sequence }, + { MongoCommitFields.CommitId, new BsonBinaryData(Guid.NewGuid(), GuidRepresentation.CSharpLegacy) }, + { MongoCommitFields.CommitStamp, new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc) }, + { MongoCommitFields.Headers, new BsonDocument() }, { MongoCommitFields.Events, new BsonArray() } + }; + if (originalId != null) + { + document[MongoCommitFields.OriginalStreamId] = originalId; + document[MongoCommitFields.OriginalBucketId] = originalBucket; + } + _documents.Add(document); + } + + private static IEnumerable Walk(BsonValue value) + { + if (value.IsBsonDocument) + { + yield return value.AsBsonDocument; + foreach (var child in value.AsBsonDocument.Values.SelectMany(Walk)) + yield return child; + } + else if (value.IsBsonArray) + foreach (var child in value.AsBsonArray.SelectMany(Walk)) + yield return child; + } + + protected override void Cleanup() + { + _engine?.Dispose(); + if (_database != null) + _database.Client.DropDatabase(_database.DatabaseNamespace.DatabaseName); + } + } +} From 8c4792930c91c9cf9d2bdce54dd92bcd76f57154 Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Mon, 26 Oct 2026 12:02:49 +0100 Subject: [PATCH 5/5] fix: index recycle-bin original-stream revision reads (#87) Add a nonunique recycle-only compound index without changing existing unique indexes. Verify full sync/async query plans, legacy/new selection, populated-store upgrade, and repeated initialization. Document additive migration, build/write/storage costs, and retained reader compatibility limits. MongoDB 7.0.45: 160 keys and 200 document examinations for 160 mixed-history results instead of 12,240 keys/documents and blocking SORT. Release build including net472 succeeds. Each net8/net9/net10 run: 536 core passes; 278 adapter passes and the unchanged #94 IDHACK assertion failure; zero skips. Co-authored by Alessandro Giorgetti and Codex (OpenAI). --- Changelog.md | 6 + docs/Recycle-Bin.md | 47 ++++++- .../Issue87RecycleIndexInitialization.cs | 118 ++++++++++++++++++ .../Issues/Issue87RecycleIndexes.cs | 2 +- .../MongoFields.cs | 4 + .../MongoPersistenceEngine.cs | 19 +++ 6 files changed, 192 insertions(+), 4 deletions(-) create mode 100644 src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87RecycleIndexInitialization.cs diff --git a/Changelog.md b/Changelog.md index 43c3f8d..ac39c78 100644 --- a/Changelog.md +++ b/Changelog.md @@ -4,10 +4,16 @@ - Fix recycle-bin identity collisions by assigning each deletion attempt a synthetic stored `StreamId` while preserving `OriginalBucketId` and `OriginalStreamId`; new reads restore the original stream identity, and legacy recycled records remain readable. [#87](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/87) +- Add a nonunique, recycle-only original-stream/revision index for selective synchronous and asynchronous recycle reads; preserve existing unique indexes and legacy lookup support. [#87](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/87) + ### Breaking Changes - 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) +### Migration notes + +- `Initialize()` now creates `GetFrom_RecycleBin_Index` on existing databases. No data backfill or old-index replacement is required, but deployments must budget index build time and disk/I/O costs and retain index-creation permission. Operators can pre-create the documented definition to move build work out of startup. See [recycle-bin rollout and indexing](docs/Recycle-Bin.md#recycle-lookup-index-and-deployment). [#87](https://github.com/NEventStore/NEventStore.Persistence.MongoDB/issues/87) + ## 12.1.0 - Added explicit support for net8.0, net9.0, net10.0. diff --git a/docs/Recycle-Bin.md b/docs/Recycle-Bin.md index 3f43029..63a7f64 100644 --- a/docs/Recycle-Bin.md +++ b/docs/Recycle-Bin.md @@ -37,11 +37,52 @@ does not find those new documents. Raw BSON consumers must also account for the new identity fields. Upgrade recycle-bin readers before creating new-format deleted data; this is not unrestricted mixed-version compatibility. -All existing indexes and checkpoint-based cleanup remain unchanged. +All existing unique indexes and checkpoint-based cleanup remain unchanged. `EmptyRecycleBin` and its async equivalent remove recycled commits older than the greatest stored checkpoint, retaining that checkpoint if it belongs to a recycled -commit. Original-ID lookups now use an additional predicate; their performance -has not been benchmarked. +commit. + +## Recycle lookup index and deployment + +`Initialize()` adds `GetFrom_RecycleBin_Index` to `Commits`: + +```javascript +{ + key: { BucketId: 1, OriginalStreamId: 1, StreamRevisionFrom: 1, StreamRevisionTo: 1 }, + name: "GetFrom_RecycleBin_Index", + unique: false, + partialFilterExpression: { BucketId: ":rb" } +} +``` + +It supports the original-ID branch of the full recycle lookup, including +revision filtering and ascending revision order. The legacy branch continues to +use the existing `GetFrom_Index` on stored `StreamId`. The new index must not be +unique: original IDs and revision ranges repeat across buckets and deletion +attempts. No existing index is dropped, replaced or made less restrictive. + +For an existing database, initialization builds the additional index over +existing recycled records without rewriting documents or backfilling identities. +Repeated initialization with the same definition is safe. Run initialization to +completion before serving traffic; a fresh engine is needed if initialization +fails. For large collections, budget deployment time, disk space, memory and I/O +for an index build, or have an operator pre-create this exact definition in a +maintenance window. The account running `Initialize()` still needs index-creation +permission; pre-creation moves build work out of startup. Resolve any conflicting definition under +this new name explicitly; do not drop existing unique indexes to force startup. + +Only recycled documents have entries, including legacy records whose original +ID is absent; live commits do not add entries. Recycling adds index writes and +recycle-bin cleanup removes entries, with storage proportional to retained +recycled data. This trades build/write/storage cost for selective original-ID +reads. It does not isolate a bucket or deletion lifetime, remove FETCH checks for +legacy records, or guarantee constant cost for a long requested history. + +Regression tests explain the actual sync/async driver queries over mixed +legacy/new records, multiple original buckets and deletion attempts, unrelated +live/recycled populations, and revision windows. They check result correctness, +identity bounds, keys/documents examined and absence of blocking sorts. These +are query-plan checks, not production latency or throughput benchmarks. ## Failure and concurrency limits diff --git a/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87RecycleIndexInitialization.cs b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87RecycleIndexInitialization.cs new file mode 100644 index 0000000..e882b73 --- /dev/null +++ b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87RecycleIndexInitialization.cs @@ -0,0 +1,118 @@ +using FluentAssertions; +using global::MongoDB.Bson; +using global::MongoDB.Driver; +using NEventStore.Persistence.AcceptanceTests.BDD; +using NEventStore.Serialization; + +namespace NEventStore.Persistence.MongoDB.Tests.AcceptanceTests.Issues +{ + public class Issue_87_when_initializing_an_existing_recycle_store : SpecificationBase + { + private IMongoDatabase _database = null!; + private IMongoCollection _commits = null!; + private readonly MongoPersistenceOptions _options = new(); + private List _originalIndexes = null!; + private List _originalDocuments = null!; + private List _firstIndexes = null!; + private List _repeatedIndexes = null!; + + protected override void Context() + { + var url = new MongoUrlBuilder(AcceptanceTestMongoPersistenceFactory.GetConnectionString()) + { + DatabaseName = $"issue87-index-upgrade-{Guid.NewGuid():N}" + }; + _database = _options.ConnectToDatabase(url.ToString()); + _commits = _database.GetCollection("Commits"); + using (var engine = NewEngine()) + engine.Initialize(); + + // Model an existing installation without changing any of its old indexes. + if (ReadIndexes().Any(index => index["name"] == MongoCommitIndexes.GetFromRecycleBin)) + _commits.Indexes.DropOne(MongoCommitIndexes.GetFromRecycleBin); + _commits.InsertMany(new[] + { + Document(1, MongoSystemBuckets.RecycleBin, "deleted-attempt-1", "original-stream"), + Document(2, MongoSystemBuckets.RecycleBin, "deleted-attempt-2", "original-stream"), + Document(3, MongoSystemBuckets.RecycleBin, "legacy-stream", null), + Document(4, "live-bucket", "live-stream", "original-stream") + }); + _originalIndexes = ReadIndexes(); + _originalDocuments = ReadDocuments(); + } + + protected override void Because() + { + using (var engine = NewEngine()) + { + engine.Initialize(); + _firstIndexes = ReadIndexes(); + engine.Initialize(); + } + // A new process/engine must also tolerate the already-created index. + using (var engine = NewEngine()) + engine.Initialize(); + _repeatedIndexes = ReadIndexes(); + } + + [Fact] + public void Upgrade_adds_one_nonunique_recycle_only_index_to_populated_storage() + { + var added = _firstIndexes.Where(index => !_originalIndexes.Any(original => original["name"] == index["name"])) + .Should().ContainSingle().Which; + added["name"].Should().Be(new BsonString(MongoCommitIndexes.GetFromRecycleBin)); + added["key"].Should().Be(new BsonDocument + { + { MongoCommitFields.BucketId, 1 }, + { MongoCommitFields.OriginalStreamId, 1 }, + { MongoCommitFields.StreamRevisionFrom, 1 }, + { MongoCommitFields.StreamRevisionTo, 1 } + }); + added.GetValue("unique", false).AsBoolean.Should().BeFalse( + "original identities and revision ranges repeat across deleted lifetimes"); + added["partialFilterExpression"].Should().Be( + new BsonDocument(MongoCommitFields.BucketId, MongoSystemBuckets.RecycleBin), + "live commits must not occupy the recycle lookup index, even if optional metadata is present"); + } + + [Fact] + public void Existing_unique_indexes_and_documents_remain_unchanged() + { + foreach (var original in _originalIndexes) + _repeatedIndexes.Single(index => index["name"] == original["name"]).Equals(original).Should().BeTrue(); + ReadDocuments().Should().Equal(_originalDocuments, "index migration must not rewrite legacy or new identities"); + } + + [Fact] + public void Initialization_is_idempotent_for_the_same_and_new_engine_instances() => + _repeatedIndexes.Should().Equal(_firstIndexes); + + private MongoPersistenceEngine NewEngine() => new(_database, new DocumentObjectSerializer(), _options); + + private List ReadIndexes() => _commits.Indexes.List().ToList() + .OrderBy(index => index["name"].AsString, StringComparer.Ordinal).ToList(); + + private List ReadDocuments() => _commits.Find(new BsonDocument()) + .Sort(new BsonDocument(MongoCommitFields.CheckpointNumber, 1)).ToList(); + + private static BsonDocument Document(long checkpoint, string bucket, string stream, string? originalStream) + { + var document = new BsonDocument + { + { MongoCommitFields.CheckpointNumber, checkpoint }, + { MongoCommitFields.BucketId, bucket }, { MongoCommitFields.StreamId, stream }, + { MongoCommitFields.StreamRevisionFrom, 1 }, { MongoCommitFields.StreamRevisionTo, 2 }, + { MongoCommitFields.CommitSequence, 1 }, { MongoCommitFields.CommitId, $"commit-{checkpoint}" } + }; + if (originalStream != null) + document[MongoCommitFields.OriginalStreamId] = originalStream; + return document; + } + + protected override void Cleanup() + { + if (_database != null) + _database.Client.DropDatabase(_database.DatabaseNamespace.DatabaseName); + } + } +} diff --git a/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87RecycleIndexes.cs b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87RecycleIndexes.cs index 73126ac..497a47a 100644 --- a/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87RecycleIndexes.cs +++ b/src/NEventStore.Persistence.MongoDB.Tests/AcceptanceTests/Issues/Issue87RecycleIndexes.cs @@ -80,7 +80,7 @@ public void Full_sync_and_async_queries_examine_only_the_requested_histories() stats["executionSuccess"].AsBoolean.Should().BeTrue(description); // Allow boundary seeks and FETCH of legacy candidates, not thousands of // unrelated recycled records. Do not couple this to a specific plan name. - stats["totalDocsExamined"].ToInt64().Should().BeLessThanOrEqualTo(200, description); + stats["totalDocsExamined"].ToInt64().Should().BeLessThanOrEqualTo(400, description); stats["totalKeysExamined"].ToInt64().Should().BeLessThanOrEqualTo(400, description); var stages = Walk(plan["queryPlanner"]["winningPlan"]) .Where(node => node.Contains("stage")).Select(node => node["stage"].AsString); diff --git a/src/NEventStore.Persistence.MongoDB/MongoFields.cs b/src/NEventStore.Persistence.MongoDB/MongoFields.cs index 7333fb5..17d7a82 100644 --- a/src/NEventStore.Persistence.MongoDB/MongoFields.cs +++ b/src/NEventStore.Persistence.MongoDB/MongoFields.cs @@ -190,6 +190,10 @@ public static class MongoCommitIndexes /// public const string GetFrom = "GetFrom_Index"; /// + /// Original-stream revision reads in the recycle bin (nonunique). + /// + public const string GetFromRecycleBin = "GetFrom_RecycleBin_Index"; + /// /// Logical unique key index. /// public const string LogicalKey = "LogicalKey_Index"; diff --git a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs index c05816c..921c565 100644 --- a/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs +++ b/src/NEventStore.Persistence.MongoDB/MongoPersistenceEngine.cs @@ -116,6 +116,25 @@ public virtual void Initialize() ) ); + // Original identities can repeat across buckets and deletion attempts. + // Keep the existing unique stored-identity indexes for live and legacy reads. + PersistedCommits.Indexes.CreateOne( + new CreateIndexModel( + Builders.IndexKeys + .Ascending(MongoCommitFields.BucketId) + .Ascending(MongoCommitFields.OriginalStreamId) + .Ascending(MongoCommitFields.StreamRevisionFrom) + .Ascending(MongoCommitFields.StreamRevisionTo), + new CreateIndexOptions() + { + Name = MongoCommitIndexes.GetFromRecycleBin, + Unique = false, + PartialFilterExpression = Builders.Filter + .Eq(MongoCommitFields.BucketId, MongoSystemBuckets.RecycleBin) + } + ) + ); + PersistedCommits.Indexes.CreateOne( new CreateIndexModel( Builders.IndexKeys