SERVER-89614 SERVER-91753 Add enum to track $out cleanup state to properly drop buckets collection during interrupt (#31968)

GitOrigin-RevId: f8582b723ddbdc1f93c7d193691d5790bfb87f76

Author: galon1

jstests/aggregation/bugs/server3253.js

@@ -8,10 +8,11 @@
 // server-3253 Unsharded support for $out
 load('jstests/aggregation/extras/utils.js');
 
-var input = db.server3253_in;
-var inputDoesntExist = db.server3253_doesnt_exist;
-var output = db.server3253_out;
-var cappedOutput = db.server3253_out_capped;
+const testDb = db.getSiblingDB("server3253");
+var input = testDb.server3253_in;
+var inputDoesntExist = testDb.server3253_doesnt_exist;
+var output = testDb.server3253_out;
+var cappedOutput = testDb.server3253_out_capped;
 
 input.drop();
 inputDoesntExist.drop();  // never created
@@ -45,8 +46,8 @@ function test(input, pipeline, expected) {
 }
 
 function listCollections(name) {
-    var collectionInfosCursor = db.runCommand("listCollections", {filter: {name: name}});
-    return new DBCommandCursor(db, collectionInfosCursor).toArray();
+    var collectionInfosCursor = testDb.runCommand("listCollections", {filter: {name: name}});
+    return new DBCommandCursor(testDb, collectionInfosCursor).toArray();
 }
 
 input.insert({_id: 1});
@@ -91,7 +92,7 @@ test(input, [{$project: {c: {$concat: ["hello there ", "_id"]}}}], [
 ]);
 
 cappedOutput.drop();
-db.createCollection(cappedOutput.getName(), {capped: true, size: 2});
+testDb.createCollection(cappedOutput.getName(), {capped: true, size: 2});
 assertErrorCode(input, {$out: cappedOutput.getName()}, 17152);
 
 // ensure everything works even if input doesn't exist.

jstests/concurrency/fsm_workloads/timeseries_agg_out.js

@@ -14,7 +14,8 @@
  *   does_not_support_transactions,
  *   does_not_support_stepdowns,
  *   requires_fcv_70,
- *   featureFlagAggOutTimeseries
+ *   # `convertToCapped` is not supported in serverless.
+ *   command_not_supported_in_serverless,
  * ]
  */
 load('jstests/concurrency/fsm_workloads/agg_out.js');  // for $super state functions

jstests/concurrency/fsm_workloads/timeseries_agg_out_interrupt_cleanup.js

@@ -11,7 +11,6 @@
  *   does_not_support_stepdowns,
  *   uses_curop_agg_stage,
  *   requires_fcv_70,
- *   featureFlagAggOutTimeseries
  * ]
  */
 load('jstests/concurrency/fsm_libs/extend_workload.js');  // for extendWorkload
@@ -58,7 +57,9 @@ var $config = extendWorkload($config, function($config, $super) {
             "command.drop": {
                 $exists: false
             }  // Exclude 'drop' command from the filter to make sure that we don't kill the the
-            // drop command which is responsible for dropping the temporary collection.
+            // drop command which is responsible for dropping the temporary collection in the
+            // destructor. This won't prevent any drop commands run internally (with the same
+            // operation context) by $out, such as in renameCollection.
         });
     };
 

jstests/core/timeseries/timeseries_out_non_sharded.js

@@ -9,7 +9,6 @@
  *   # We need a timeseries collection.
  *   requires_timeseries,
  *   requires_fcv_70,
- *  featureFlagAggOutTimeseries
  * ]
  */
 (function() {

jstests/noPassthrough/out_timeseries_cleans_up_bucket_collections.js

@@ -0,0 +1,157 @@
+/*
+ * Tests that $out cleans up the buckets collections if interrupted after the rename, but before the
+ * view is created.
+ *
+ * @tags: [
+ *   # We need a timeseries collection.
+ *   requires_timeseries,
+ *   requires_persistence,
+ * ]
+ */
+(function() {
+"use strict";
+
+load("jstests/libs/fail_point_util.js");
+load("jstests/libs/parallel_shell_helpers.js");
+
+const st = new ShardingTest({shards: 2});
+
+const dbName = jsTestName();
+const outCollName = 'out';
+const sourceCollName = 'foo';
+const outBucketsCollName = 'system.buckets.out';
+const testDB = st.s.getDB(dbName);
+const sourceDocument = {
+    x: 1,
+    t: ISODate()
+};
+const primary = st.shard0.shardName;
+assert.commandWorked(st.s.adminCommand({enableSharding: dbName, primaryShard: primary}));
+
+function listCollections(collName) {
+    return testDB.getCollectionNames().filter(coll => coll === collName);
+}
+
+function resetCollections() {
+    if (testDB[sourceCollName]) {
+        assert(testDB[sourceCollName].drop());
+    }
+    if (testDB[outCollName]) {
+        assert(testDB[outCollName].drop());
+    }
+}
+
+function killOp() {
+    const adminDB = st.s.getDB("admin");
+    const curOps = adminDB
+                       .aggregate([
+                           {$currentOp: {allUsers: true}},
+                           {
+                               $match: {
+                                   "command.comment": "testComment",
+                                   // The create coordinator issues fire and forget refreshes after
+                                   // creating a collection. We filter these out to ensure we are
+                                   // killing the correct operation.
+                                   "command._flushRoutingTableCacheUpdates": {$exists: false}
+                               }
+                           }
+                       ])
+                       .toArray();
+    assert.eq(1, curOps.length, curOps);
+    assert.commandWorked(adminDB.killOp(curOps[0].opid));
+}
+
+function runOut(dbName, sourceCollName, targetCollName) {
+    const cmdRes = db.getSiblingDB(dbName).runCommand({
+        aggregate: sourceCollName,
+        pipeline: [{$out: {db: dbName, coll: targetCollName, timeseries: {timeField: 't'}}}],
+        cursor: {},
+        comment: "testComment",
+    });
+    assert.commandFailed(cmdRes);
+}
+
+function runOutAndInterrupt() {
+    const fp = configureFailPoint(st.shard0.rs.getPrimary(),
+                                  'outWaitAfterTempCollectionRenameBeforeView',
+                                  {shouldCheckForInterrupt: true});
+
+    let outShell =
+        startParallelShell(funWithArgs(runOut, dbName, sourceCollName, outCollName), st.s.port);
+
+    fp.wait();
+
+    // Verify that the buckets collection is created.
+    let bucketCollections = listCollections(outBucketsCollName);
+    assert.eq(1, bucketCollections.length, bucketCollections);
+
+    // Provoke failure.
+    killOp();
+    outShell();
+
+    // Assert that the temporary collections has been garbage collected.
+    const tempCollections =
+        testDB.getCollectionNames().filter(coll => coll.startsWith('tmp.agg_out'));
+    const garbageCollectionEntries = st.s.getDB('config')['agg_temp_collections'].count();
+    assert(tempCollections.length === 0 && garbageCollectionEntries === 0);
+}
+
+// Validates $out should clean up the buckets collection if the command is interrupted before the
+// view is created.
+function testCreatingNewCollection() {
+    resetCollections();
+
+    // Create source collection.
+    assert.commandWorked(testDB.runCommand({create: sourceCollName}));
+    assert.commandWorked(testDB[sourceCollName].insert(sourceDocument));
+
+    let bucketCollections = listCollections(outBucketsCollName);
+    assert.eq(0, bucketCollections, bucketCollections);
+
+    // The output collection is a time-series collection and needs a view so $out will run on the
+    // primary.
+    runOutAndInterrupt();
+
+    // There should neither be a buckets collection nor a view.
+    bucketCollections = listCollections(outBucketsCollName);
+    assert.eq(0, bucketCollections.length, bucketCollections);
+    let view = listCollections(outCollName);
+    assert.eq(0, view.length, view);
+}
+
+testCreatingNewCollection();
+
+// Validates $out should not clean up the buckets collection if the command is interrupted when the
+// view exists.
+function testReplacingExistingCollection() {
+    resetCollections();
+
+    // Create source collection.
+    assert.commandWorked(testDB.runCommand({create: sourceCollName}));
+    assert.commandWorked(testDB[sourceCollName].insert(sourceDocument));
+
+    // Create the time-series collection $out will replace.
+    assert.commandWorked(testDB.runCommand({create: outCollName, timeseries: {timeField: "t"}}));
+    assert.commandWorked(testDB[outCollName].insert({a: 1, t: ISODate()}));
+
+    let bucketCollections = listCollections(outBucketsCollName);
+    assert.eq(1, bucketCollections.length, bucketCollections);
+
+    // The output collection is a time-series collection and needs a view so $out will run on the
+    // primary.
+    runOutAndInterrupt();
+
+    // There should be a buckets collection and a view.
+    bucketCollections = listCollections(outBucketsCollName);
+    assert.eq(1, bucketCollections.length, bucketCollections);
+    let view = listCollections(outCollName);
+    assert.eq(1, view.length, view);
+
+    // $out should replace the document inside the collection.
+    assert.sameMembers(testDB[outCollName].find({}, {_id: 0}).toArray(), [sourceDocument]);
+}
+
+testReplacingExistingCollection();
+
+st.stop();
+}());

jstests/sharding/timeseries_out_sharded.js

@@ -10,7 +10,6 @@
  *   # We need a timeseries collection.
  *   requires_timeseries,
  *   requires_fcv_70,
- *   featureFlagAggOutTimeseries
  * ]
  */
 (function() {