SERVER-44942 Don't allow drop of out temp collection before index copying

(cherry picked from commit 817a57416b3069ae4738e5bd472d443a6b12f215)
This commit also backports SERVER-45636 Tag out_drop_temp_collection.js as requiring replication and sharding

3 files changed, 85 insertions, 0 deletions

diff --git a/jstests/noPassthrough/out_drop_temp_collection.js b/jstests/noPassthrough/out_drop_temp_collection.js
new file mode 100644
index 00000000000..ce312853c90
--- /dev/null
+++ b/jstests/noPassthrough/out_drop_temp_collection.js
@@ -0,0 +1,71 @@
+/**
+ * Test that the temp collection created by $out is not dropped even if the database containing it
+ * is dropped during the operation.
+ *
+ * @tags: [
+ *   assumes_unsharded_collection,
+ *   do_not_wrap_aggregations_in_facets,
+ *   assumes_read_concern_unchanged,
+ *   requires_replication,
+ *   requires_sharding,
+ * ]
+ */
+
+(function() {
+"use strict";
+
+load("jstests/libs/curop_helpers.js");    // for waitForCurOpByFilter.
+load("jstests/libs/fixture_helpers.js");  // For FixtureHelpers.
+
+function setFailPointOnShards(st, failpointName, mode, numShards) {
+    for (let i = 0; i < numShards; i++) {
+        const shardConn = st["rs" + i].getPrimary();
+        assert.commandWorked(
+            shardConn.adminCommand({configureFailPoint: failpointName, mode: mode}));
+    }
+}
+
+function runTest(st, testDb, portNum) {
+    const failpointName = "outWaitAfterTempCollectionCreation";
+    const coll = testDb.out_source_coll;
+    coll.drop();
+
+    const targetColl = testDb.out_target_coll;
+    targetColl.drop();
+
+    assert.commandWorked(coll.insert({val: 0}));
+    assert.commandWorked(coll.createIndex({val: 1}));
+
+    let res = FixtureHelpers.runCommandOnEachPrimary({
+        db: testDb.getSiblingDB("admin"),
+        cmdObj: {
+            configureFailPoint: failpointName,
+            mode: "alwaysOn",
+        }
+    });
+
+    const aggDone = startParallelShell(() => {
+        const targetColl = db.getSiblingDB("out_drop_temp").out_target_coll;
+        const pipeline = [{$out: "out_target_coll"}];
+        targetColl.aggregate(pipeline);
+    }, portNum);
+
+    waitForCurOpByFilter(testDb, {"msg": failpointName});
+    testDb.adminCommand({dropDatabase: testDb.getName()});
+    // The $out should complete once the failpoint is disabled, not fail on index creation.
+    FixtureHelpers.runCommandOnEachPrimary({
+        db: testDb.getSiblingDB("admin"),
+        cmdObj: {
+            configureFailPoint: failpointName,
+            mode: "off",
+        }
+    });
+    aggDone();
+}
+const conn = MongoRunner.runMongod({});
+runTest(null, conn.getDB("out_drop_temp"), conn.port);
+MongoRunner.stopMongod(conn);
+const st = new ShardingTest({shards: 2, mongos: 1, config: 1});
+runTest(st, st.s.getDB("out_drop_temp"), st.s.port);
+st.stop();
+})();
diff --git a/src/mongo/db/pipeline/SConscript b/src/mongo/db/pipeline/SConscript
index 2f80afdddba..22825423f05 100644
--- a/src/mongo/db/pipeline/SConscript
+++ b/src/mongo/db/pipeline/SConscript
@@ -458,6 +458,7 @@ pipelineeEnv.Library(
         'parsed_aggregation_projection',
     ],
     LIBDEPS_PRIVATE=[
+        '$BUILD_DIR/mongo/db/background',
         '$BUILD_DIR/mongo/db/commands/test_commands_enabled',
         '$BUILD_DIR/mongo/rpc/command_status',
     ]
diff --git a/src/mongo/db/pipeline/document_source_out.cpp b/src/mongo/db/pipeline/document_source_out.cpp
index f8478395328..ba44634df87 100644
--- a/src/mongo/db/pipeline/document_source_out.cpp
+++ b/src/mongo/db/pipeline/document_source_out.cpp
@@ -35,11 +35,13 @@
 
 #include <fmt/format.h>
 
+#include "mongo/db/background.h"
 #include "mongo/db/curop_failpoint_helpers.h"
 #include "mongo/db/ops/write_ops.h"
 #include "mongo/db/pipeline/document_path_support.h"
 #include "mongo/rpc/get_status_from_command_result.h"
 #include "mongo/util/destructor_guard.h"
+#include "mongo/util/fail_point.h"
 #include "mongo/util/log.h"
 
 namespace mongo {
@@ -48,6 +50,7 @@ using namespace fmt::literals;
 static AtomicWord<unsigned> aggOutCounter;
 
 MONGO_FAIL_POINT_DEFINE(hangWhileBuildingDocumentSourceOutBatch);
+MONGO_FAIL_POINT_DEFINE(outWaitAfterTempCollectionCreation);
 REGISTER_DOCUMENT_SOURCE(out,
                          DocumentSourceOut::LiteParsed::parse,
                          DocumentSourceOut::createFromBson);
@@ -140,6 +143,16 @@ void DocumentSourceOut::initialize() {
                 conn->runCommand(outputNs.db().toString(), cmd.done(), info));
     }
 
+    // Disallows drops and renames on this namespace.
+    BackgroundOperation backgroundOp(_tempNs.ns());
+    CurOpFailpointHelpers::waitWhileFailPointEnabled(
+        &outWaitAfterTempCollectionCreation,
+        pExpCtx->opCtx,
+        "outWaitAfterTempCollectionCreation",
+        []() {
+            log() << "Hanging aggregation due to 'outWaitAfterTempCollectionCreation' "
+                  << "failpoint";
+        });
     if (_originalIndexes.empty()) {
         return;
     }