Fix recycle-bin identity collisions and metadata cleanup ordering (#87)

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.

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<BsonDocument> _original = null!;
+        private List<BsonDocument> _afterFailure = null!;
+        private List<BsonDocument> _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<BsonDocument>("Commits").UpdateOne(
+                new BsonDocument(MongoCommitFields.CommitSequence, 1),
+                Builders<BsonDocument>.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<BsonDocument>(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<BsonDocument>(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<StorageException>().Which.InnerException;
+            cause.Should().BeOfType<MongoWriteException>().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<BsonDocument> ReadDocuments() => Database.GetCollection<BsonDocument>("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;
+    }
+}

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<BsonDocument>(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;
+    }
+}