SERVER-57660 Fail time-series inserts if buckets collection does not exist

2 files changed, 86 insertions, 3 deletions

diff --git a/jstests/noPassthrough/timeseries_insert_no_buckets_collection.js b/jstests/noPassthrough/timeseries_insert_no_buckets_collection.js
new file mode 100644
index 00000000000..d3515e0a9ca
--- /dev/null
+++ b/jstests/noPassthrough/timeseries_insert_no_buckets_collection.js
@@ -0,0 +1,64 @@
+/**
+ * Tests that inserting into a time-series collection fails if the corresponding buckets collection
+ * does not exist.
+ */
+(function() {
+'use strict';
+
+load('jstests/libs/fail_point_util.js');
+load("jstests/libs/parallel_shell_helpers.js");
+
+const conn = MongoRunner.runMongod();
+const testDB = conn.getDB('test');
+
+const timeFieldName = 'time';
+
+let testCounter = 0;
+const runTest = function(ordered, insertBeforeDrop, dropBucketsColl) {
+    const coll = testDB[jsTestName() + '_' + testCounter++];
+
+    assert.commandWorked(
+        testDB.createCollection(coll.getName(), {timeseries: {timeField: timeFieldName}}));
+
+    if (insertBeforeDrop) {
+        assert.commandWorked(coll.insert({_id: 0, [timeFieldName]: ISODate()}));
+    }
+
+    const fp = configureFailPoint(conn, 'hangTimeseriesInsertBeforeWrite');
+
+    const awaitDrop = startParallelShell(
+        funWithArgs(
+            function(collName, fpName, fpTimesEntered) {
+                load("jstests/libs/fail_point_util.js");
+
+                assert.commandWorked(db.adminCommand({
+                    waitForFailPoint: fpName,
+                    timesEntered: fpTimesEntered + 1,
+                    maxTimeMS: kDefaultWaitForFailPointTimeout,
+                }));
+
+                assert(db[collName].drop());
+
+                assert.commandWorked(db.adminCommand({configureFailPoint: fpName, mode: 'off'}));
+            },
+            dropBucketsColl ? 'system.buckets.' + coll.getName() : coll.getName(),
+            fp.failPointName,
+            fp.timesEntered),
+        conn.port);
+
+    assert.commandFailedWithCode(
+        coll.insert({_id: 1, [timeFieldName]: ISODate()}, {ordered: ordered}),
+        ErrorCodes.NamespaceNotFound);
+
+    awaitDrop();
+};
+
+for (const dropBucketsColl of [false, true]) {
+    runTest(false /* ordered */, false /* insertBeforeDrop */, dropBucketsColl);
+    runTest(false /* ordered */, true /* insertBeforeDrop */, dropBucketsColl);
+    runTest(true /* ordered */, false /* insertBeforeDrop */, dropBucketsColl);
+    runTest(true /* ordered */, true /* insertBeforeDrop */, dropBucketsColl);
+}
+
+MongoRunner.stopMongod(conn);
+})();
diff --git a/src/mongo/db/ops/write_ops_exec.cpp b/src/mongo/db/ops/write_ops_exec.cpp
index a7ab1afa917..d4f6d4aa5c4 100644
--- a/src/mongo/db/ops/write_ops_exec.cpp
+++ b/src/mongo/db/ops/write_ops_exec.cpp
@@ -382,6 +382,12 @@ Status checkIfTransactionOnCappedColl(OperationContext* opCtx, const CollectionP
     return Status::OK();
 }
 
+void assertTimeseriesBucketsCollectionNotFound(const NamespaceString& ns) {
+    uasserted(ErrorCodes::NamespaceNotFound,
+              str::stream() << "Buckets collection not found for time-series collection "
+                            << ns.getTimeseriesViewNamespace());
+}
+
 /**
  * Returns true if caller should try to insert more documents. Does nothing else if batch is empty.
  */
@@ -421,9 +427,13 @@ bool insertBatchAndHandleErrors(OperationContext* opCtx,
                 opCtx,
                 wholeOp.getNamespace(),
                 fixLockModeForSystemDotViewsChanges(wholeOp.getNamespace(), MODE_IX));
-            if (collection->getCollection())
+            if (*collection)
                 break;
 
+            if (source == OperationSource::kTimeseries) {
+                assertTimeseriesBucketsCollectionNotFound(wholeOp.getNamespace());
+            }
+
             collection.reset();  // unlock.
             makeCollection(opCtx, wholeOp.getNamespace());
         }
@@ -723,10 +733,17 @@ static SingleWriteResult performSingleUpdateOp(OperationContext* opCtx,
     boost::optional<AutoGetCollection> collection;
     while (true) {
         collection.emplace(opCtx, ns, fixLockModeForSystemDotViewsChanges(ns, MODE_IX));
+        if (*collection) {
+            break;
+        }
+
+        if (source == OperationSource::kTimeseries) {
+            assertTimeseriesBucketsCollectionNotFound(ns);
+        }
 
         // If this is an upsert, which is an insert, we must have a collection.
         // An update on a non-existant collection is okay and handled later.
-        if (collection->getCollection() || !updateRequest.isUpsert())
+        if (!updateRequest.isUpsert())
             break;
 
         collection.reset();  // unlock.
@@ -1157,7 +1174,9 @@ Status performAtomicTimeseriesWrites(
     lastOpFixer.startingOp();
 
     AutoGetCollection coll{opCtx, ns, MODE_IX};
-    invariant(coll);
+    if (!coll) {
+        assertTimeseriesBucketsCollectionNotFound(ns);
+    }
 
     auto curOp = CurOp::get(opCtx);
     curOp->raiseDbProfileLevel(CollectionCatalog::get(opCtx)->getDatabaseProfileLevel(ns.db()));