SERVER-78122 Address bucket OID collisions with buckets pending compressionr6.3.2-rc1r6.3.2

1 files changed, 26 insertions, 7 deletions

diff --git a/src/mongo/db/timeseries/bucket_catalog/bucket_catalog.cpp b/src/mongo/db/timeseries/bucket_catalog/bucket_catalog.cpp
index d9c3e9e21b4..ee6f78b0c90 100644
--- a/src/mongo/db/timeseries/bucket_catalog/bucket_catalog.cpp
+++ b/src/mongo/db/timeseries/bucket_catalog/bucket_catalog.cpp
@@ -1613,13 +1613,32 @@ Bucket* BucketCatalog::_allocateBucket(Stripe* stripe,
     Bucket* bucket = it->second.get();
     stripe->openBuckets[info.key].emplace(bucket);
 
-    auto state = changeBucketState(
-        _bucketStateRegistry,
-        it->first,
-        [](boost::optional<BucketState> input, std::uint64_t) -> boost::optional<BucketState> {
-            invariant(!input.has_value());
-            return BucketState{};
-        });
+    bool conflicts = false;
+    auto state = changeBucketState(_bucketStateRegistry,
+                                   bucket->bucketId,
+                                   [&conflicts](boost::optional<BucketState> input,
+                                                std::uint64_t) -> boost::optional<BucketState> {
+                                       // If we have an existing state for a bucket not tracked in
+                                       // the catalog (which can happen if a bucket is undergoing
+                                       // compression) then we should avoid allocating the bucket
+                                       // and throw a WriteConflict.
+                                       if (input.has_value()) {
+                                           conflicts = true;
+                                           return input;
+                                       }
+
+                                       conflicts = false;
+                                       return BucketState{};
+                                   });
+
+    if (conflicts) {
+        // At this point, we have only added the bucket to the openBuckets and allBuckets maps so we
+        // can go ahead and simply remove them from those maps.
+        stripe->openBuckets[info.key].erase(bucket);
+        stripe->allBuckets.erase(it);
+        throwWriteConflictException("OID collision occurred when inserting document(s).");
+    }
+
     invariant(state == BucketState{});
     _numberOfActiveBuckets.fetchAndAdd(1);