SERVER-34572 Write commands in transactions must check for pending catalog changes

9 files changed, 108 insertions, 69 deletions

diff --git a/jstests/core/txns/abort_expired_transaction.js b/jstests/core/txns/abort_expired_transaction.js
index bcfe280834f..97eec8279c8 100644
--- a/jstests/core/txns/abort_expired_transaction.js
+++ b/jstests/core/txns/abort_expired_transaction.js
@@ -27,7 +27,7 @@
             db.adminCommand({setParameter: 1, transactionLifetimeLimitSeconds: 1}));
 
         jsTest.log("Create a collection '" + ns + "' outside of the transaction.");
-        assert.writeOK(testColl.insert({foo: "bar"}));
+        assert.writeOK(testColl.insert({foo: "bar"}, {writeConcern: {w: "majority"}}));
 
         jsTest.log("Set up the session.");
         const sessionOptions = {causalConsistency: false};
diff --git a/jstests/core/txns/list_collections_not_blocked_by_txn.js b/jstests/core/txns/list_collections_not_blocked_by_txn.js
index d30213fb5b0..82498110b3a 100644
--- a/jstests/core/txns/list_collections_not_blocked_by_txn.js
+++ b/jstests/core/txns/list_collections_not_blocked_by_txn.js
@@ -10,9 +10,9 @@
 
     mydb.foo.drop();
 
-    mydb.createCollection("foo");
+    assert.commandWorked(mydb.createCollection("foo", {writeConcern: {w: "majority"}}));
     session.startTransaction({readConcern: {level: "snapshot"}});
-    sessionDb.foo.insert({x: 1});
+    assert.commandWorked(sessionDb.foo.insert({x: 1}));
 
     for (let nameOnly of[false, true]) {
         // Check that both the nameOnly and full versions of listCollections don't block.
diff --git a/jstests/core/update_arrayFilters.js b/jstests/core/update_arrayFilters.js
index f5d837f18b0..a59b135e75d 100644
--- a/jstests/core/update_arrayFilters.js
+++ b/jstests/core/update_arrayFilters.js
@@ -6,8 +6,10 @@
 (function() {
     "use strict";
 
-    let coll = db.update_arrayFilters;
+    const collName = "update_arrayFilters";
+    let coll = db[collName];
     coll.drop();
+    assert.commandWorked(db.createCollection(collName));
     let res;
 
     //
@@ -512,9 +514,9 @@
 
         // arrayFilters respect the collection default collation.
         coll.drop();
-        assert.commandWorked(db.createCollection("update_arrayFilters",
-                                                 {collation: {locale: "en_US", strength: 2}}));
-        coll = db.update_arrayFilters;
+        assert.commandWorked(
+            db.createCollection(collName, {collation: {locale: "en_US", strength: 2}}));
+        coll = db[collName];
         assert.writeOK(coll.insert({_id: 0, a: ["foo", "FOO"]}));
         assert.writeOK(
             coll.update({_id: 0}, {$set: {"a.$[i]": "bar"}}, {arrayFilters: [{i: "foo"}]}));
diff --git a/jstests/noPassthrough/read_concern_snapshot_catalog_invalidation.js b/jstests/noPassthrough/read_concern_snapshot_catalog_invalidation.js
index 03f9a9adc50..d926e379fed 100644
--- a/jstests/noPassthrough/read_concern_snapshot_catalog_invalidation.js
+++ b/jstests/noPassthrough/read_concern_snapshot_catalog_invalidation.js
@@ -18,7 +18,6 @@
     }
     const adminDB = testDB.getSiblingDB("admin");
     const coll = testDB.getCollection(kCollName);
-    coll.drop();
 
     function waitForOp(curOpFilter) {
         assert.soon(
@@ -35,38 +34,74 @@
             });
     }
 
-    assert.commandWorked(coll.insert({x: 1}, {writeConcern: {w: "majority"}}));
+    function testCommand(cmd, curOpFilter) {
+        coll.drop();
+        assert.commandWorked(testDB.runCommand({
+            createIndexes: kCollName,
+            indexes:
+                [{key: {haystack: "geoHaystack", a: 1}, name: "haystack_geo", bucketSize: 1}]
+        }));
+        assert.commandWorked(coll.insert({x: 1}, {writeConcern: {w: "majority"}}));
 
-    // Start a snapshot find that hangs after establishing a storage engine transaction.
-    assert.commandWorked(testDB.adminCommand(
-        {configureFailPoint: "hangAfterPreallocateSnapshot", mode: "alwaysOn"}));
+        // Start a command with readConcern "snapshot" that hangs after establishing a storage
+        // engine transaction.
+        assert.commandWorked(testDB.adminCommand(
+            {configureFailPoint: "hangAfterPreallocateSnapshot", mode: "alwaysOn"}));
 
-    const awaitCommand = startParallelShell(function() {
-        const session = db.getMongo().startSession();
-        const sessionDb = session.getDatabase("test");
-        session.startTransaction({readConcern: {level: "snapshot"}});
-        const res = sessionDb.runCommand({find: "coll"});
-        assert.commandFailedWithCode(res, ErrorCodes.SnapshotUnavailable);
-        session.endSession();
-    }, rst.ports[0]);
+        const awaitCommand = startParallelShell(
+            "const session = db.getMongo().startSession();" +
+                "const sessionDb = session.getDatabase('test');" +
+                "session.startTransaction({readConcern: {level: 'snapshot'}});" +
+                "const res = sessionDb.runCommand(" + tojson(cmd) + ");" +
+                "assert.commandFailedWithCode(res, ErrorCodes.SnapshotUnavailable);" +
+                "session.endSession();",
+            rst.ports[0]);
 
-    waitForOp({"command.find": kCollName, "command.readConcern.level": "snapshot"});
+        waitForOp(curOpFilter);
 
-    // Create an index on the collection the find was executed against. This will move the
-    // collection's minimum visible timestamp to a point later than the point-in-time referenced
-    // by the find snapshot.
-    assert.commandWorked(testDB.runCommand({
-        createIndexes: kCollName,
-        indexes: [{key: {x: 1}, name: "x_1"}],
-        writeConcern: {w: "majority"}
-    }));
+        // Create an index on the collection the command was executed against. This will move the
+        // collection's minimum visible timestamp to a point later than the point-in-time referenced
+        // by the transaction snapshot.
+        assert.commandWorked(testDB.runCommand({
+            createIndexes: kCollName,
+            indexes: [{key: {x: 1}, name: "x_1"}],
+            writeConcern: {w: "majority"}
+        }));
 
-    // Disable the hang and check for parallel shell success. Success indicates that the find
-    // command failed due to collection metadata invalidation.
-    assert.commandWorked(
-        testDB.adminCommand({configureFailPoint: "hangAfterPreallocateSnapshot", mode: "off"}));
+        // Disable the hang and check for parallel shell success. Success indicates that the command
+        // failed due to collection metadata invalidation.
+        assert.commandWorked(
+            testDB.adminCommand({configureFailPoint: "hangAfterPreallocateSnapshot", mode: "off"}));
 
-    awaitCommand();
+        awaitCommand();
+    }
+
+    testCommand({aggregate: kCollName, pipeline: [], cursor: {}},
+                {"command.aggregate": kCollName, "command.readConcern.level": "snapshot"});
+    testCommand({count: kCollName, filter: {x: 1}},
+                {"command.count": kCollName, "command.readConcern.level": "snapshot"});
+    testCommand({delete: kCollName, deletes: [{q: {x: 1}, limit: 1}]},
+                {"command.delete": kCollName, "command.readConcern.level": "snapshot"});
+    testCommand({distinct: kCollName, key: "x"},
+                {"command.distinct": kCollName, "command.readConcern.level": "snapshot"});
+    testCommand({find: kCollName},
+                {"command.find": kCollName, "command.readConcern.level": "snapshot"});
+    testCommand({findAndModify: kCollName, query: {x: 1}, remove: true}, {
+        "command.findAndModify": kCollName,
+        "command.remove": true,
+        "command.readConcern.level": "snapshot"
+    });
+    testCommand({findAndModify: kCollName, query: {x: 1}, update: {$set: {x: 2}}}, {
+        "command.findAndModify": kCollName,
+        "command.update.$set": {x: 2},
+        "command.readConcern.level": "snapshot"
+    });
+    testCommand({geoSearch: kCollName, near: [0, 0], maxDistance: 1, search: {a: 1}},
+                {"command.geoSearch": kCollName, "command.readConcern.level": "snapshot"});
+    testCommand({insert: kCollName, documents: [{x: 1}]},
+                {"command.insert": kCollName, "command.readConcern.level": "snapshot"});
+    testCommand({update: kCollName, updates: [{q: {x: 1}, u: {$set: {x: 2}}}]},
+                {"command.update": kCollName, "command.readConcern.level": "snapshot"});
 
     rst.stopSet();
 })();
diff --git a/jstests/replsets/transaction_table_multi_statement_txn.js b/jstests/replsets/transaction_table_multi_statement_txn.js
index 097a400ef1a..21d95b0eb7a 100644
--- a/jstests/replsets/transaction_table_multi_statement_txn.js
+++ b/jstests/replsets/transaction_table_multi_statement_txn.js
@@ -20,7 +20,8 @@
     const coll = primaryDB.getCollection('coll');
 
     jsTestLog('Creating collection ' + coll.getFullName());
-    assert.commandWorked(primaryDB.createCollection(coll.getName()));
+    assert.commandWorked(
+        primaryDB.createCollection(coll.getName(), {writeConcern: {w: "majority"}}));
     replTest.awaitReplication();
 
     const sessionId = session.getSessionId();
diff --git a/src/mongo/db/catalog_raii.cpp b/src/mongo/db/catalog_raii.cpp
index 2763bad080c..116ca47dd6e 100644
--- a/src/mongo/db/catalog_raii.cpp
+++ b/src/mongo/db/catalog_raii.cpp
@@ -113,9 +113,29 @@ AutoGetCollection::AutoGetCollection(OperationContext* opCtx,
                             << " disappeared after successufully resolving "
                             << nsOrUUID.toString());
 
-    // If the collection exists, there is no need to check for views
-    if (_coll)
+    if (_coll) {
+        // Unlike read concern majority, read concern snapshot cannot yield and wait when there are
+        // pending catalog changes. Instead, we must return an error in such situations. We ignore
+        // this restriction for the oplog, since it never has pending catalog changes.
+        if (opCtx->recoveryUnit()->getReadConcernLevel() ==
+                repl::ReadConcernLevel::kSnapshotReadConcern &&
+            _resolvedNss != NamespaceString::kRsOplogNamespace) {
+            auto mySnapshot = opCtx->recoveryUnit()->getPointInTimeReadTimestamp();
+            invariant(mySnapshot);
+            auto minSnapshot = _coll->getMinimumVisibleSnapshot();
+            uassert(
+                ErrorCodes::SnapshotUnavailable,
+                str::stream() << "Unable to read from a snapshot due to pending collection catalog "
+                                 "changes; please retry the operation. Snapshot timestamp is "
+                              << mySnapshot->toString()
+                              << ". Collection minimum is "
+                              << minSnapshot->toString(),
+                !minSnapshot || *mySnapshot >= *minSnapshot);
+        }
+
+        // If the collection exists, there is no need to check for views.
         return;
+    }
 
     _view = db->getViewCatalog()->lookup(opCtx, _resolvedNss.ns());
     uassert(ErrorCodes::CommandNotSupportedOnView,
diff --git a/src/mongo/db/commands/find_and_modify.cpp b/src/mongo/db/commands/find_and_modify.cpp
index 3a36c306626..41c3e01c696 100644
--- a/src/mongo/db/commands/find_and_modify.cpp
+++ b/src/mongo/db/commands/find_and_modify.cpp
@@ -440,23 +440,25 @@ public:
                 ParsedUpdate parsedUpdate(opCtx, &request);
                 uassertStatusOK(parsedUpdate.parseRequest());
 
-                // These are boost::optionap, because if the database or collection does not exist,
+                // These are boost::optional, because if the database or collection does not exist,
                 // they will have to be reacquired in MODE_X
                 boost::optional<AutoGetOrCreateDb> autoDb;
-                boost::optional<Lock::CollectionLock> collLock;
+                boost::optional<AutoGetCollection> autoColl;
 
-                autoDb.emplace(opCtx, dbName, MODE_IX);
-                collLock.emplace(opCtx->lockState(), nsString.ns(), MODE_IX);
+                autoColl.emplace(opCtx, nsString, MODE_IX);
 
                 {
+                    boost::optional<int> dbProfilingLevel;
+                    if (autoColl->getDb())
+                        dbProfilingLevel = autoColl->getDb()->getProfilingLevel();
+
                     stdx::lock_guard<Client> lk(*opCtx->getClient());
-                    CurOp::get(opCtx)->enter_inlock(nsString.ns().c_str(),
-                                                    autoDb->getDb()->getProfilingLevel());
+                    CurOp::get(opCtx)->enter_inlock(nsString.ns().c_str(), dbProfilingLevel);
                 }
 
                 assertCanWrite(opCtx, nsString);
 
-                Collection* collection = autoDb->getDb()->getCollection(opCtx, nsString);
+                Collection* collection = autoColl->getCollection();
 
                 // Create the collection if it does not exist when performing an upsert because the
                 // update stage does not create its own collection
@@ -468,8 +470,7 @@ public:
 
                     // Release the collection lock and reacquire a lock on the database in exclusive
                     // mode in order to create the collection
-                    collLock.reset();
-                    autoDb.reset();
+                    autoColl.reset();
                     autoDb.emplace(opCtx, dbName, MODE_X);
 
                     assertCanWrite(opCtx, nsString);
@@ -490,13 +491,6 @@ public:
                     invariant(collection);
                 }
 
-                // Perform an explicit check for "not a view" because the update path doesn't use
-                // AutoGetCollection
-                uassert(ErrorCodes::CommandNotSupportedOnView,
-                        "findAndModify not supported on a view",
-                        collection ||
-                            !autoDb->getDb()->getViewCatalog()->lookup(opCtx, nsString.ns()));
-
                 const auto exec =
                     uassertStatusOK(getExecutorUpdate(opCtx, opDebug, collection, &parsedUpdate));
 
diff --git a/src/mongo/db/db_raii.cpp b/src/mongo/db/db_raii.cpp
index 59d49c5943b..3ff4649584a 100644
--- a/src/mongo/db/db_raii.cpp
+++ b/src/mongo/db/db_raii.cpp
@@ -134,8 +134,7 @@ AutoGetCollectionForRead::AutoGetCollectionForRead(OperationContext* opCtx,
 
         // Return if there are no conflicting catalog changes on the collection.
         auto minSnapshot = coll->getMinimumVisibleSnapshot();
-        if (!_conflictingCatalogChanges(
-                opCtx, readConcernLevel, minSnapshot, lastAppliedTimestamp)) {
+        if (!_conflictingCatalogChanges(opCtx, minSnapshot, lastAppliedTimestamp)) {
             return;
         }
 
@@ -217,7 +216,6 @@ bool AutoGetCollectionForRead::_shouldReadAtLastAppliedTimestamp(
 
 bool AutoGetCollectionForRead::_conflictingCatalogChanges(
     OperationContext* opCtx,
-    repl::ReadConcernLevel readConcernLevel,
     boost::optional<Timestamp> minSnapshot,
     boost::optional<Timestamp> lastAppliedTimestamp) const {
     // This is the timestamp of the most recent catalog changes to this collection. If this is
@@ -246,16 +244,6 @@ bool AutoGetCollectionForRead::_conflictingCatalogChanges(
         return false;
     }
 
-    // Snapshot readConcern can't yield its locks when there are catalog changes.
-    if (readConcernLevel == repl::ReadConcernLevel::kSnapshotReadConcern) {
-        uasserted(
-            ErrorCodes::SnapshotUnavailable,
-            str::stream() << "Unable to read from a snapshot due to pending collection catalog "
-                             "changes; please retry the operation. Snapshot timestamp is "
-                          << mySnapshot->toString()
-                          << ". Collection minimum is "
-                          << minSnapshot->toString());
-    }
     return true;
 }
 
diff --git a/src/mongo/db/db_raii.h b/src/mongo/db/db_raii.h
index 847813a8e02..8f27a8a962a 100644
--- a/src/mongo/db/db_raii.h
+++ b/src/mongo/db/db_raii.h
@@ -122,10 +122,9 @@ private:
                                            const NamespaceString& nss,
                                            repl::ReadConcernLevel readConcernLevel) const;
 
-    // Returns true if the minSnapshot causes conflicting catalog changes for the provided read
-    // concern level or lastAppliedTimestamp.
+    // Returns true if the minSnapshot causes conflicting catalog changes for either the provided
+    // lastAppliedTimestamp or the point-in-time snapshot of the RecoveryUnit on 'opCtx'.
     bool _conflictingCatalogChanges(OperationContext* opCtx,
-                                    repl::ReadConcernLevel readConcernLevel,
                                     boost::optional<Timestamp> minSnapshot,
                                     boost::optional<Timestamp> lastAppliedTimestamp) const;
 };