SERVER-24563 Fix race in check for DB names that differ in case only

(cherry picked from commit dceaf6bf28fb879eb23f3c022647ee3e8f15c370)

Modifications for backport:
        src/mongo/db/catalog/database_holder.h
        src/mongo/db/catalog/database_holder.cpp
        jstests/concurrency/fsm_all_sharded_replication.js
        jstests/concurrency/fsm_all_sharded_replication_with_balancer.js
        jstests/concurrency/fsm_workloads/create_database.js

8 files changed, 220 insertions, 84 deletions

diff --git a/jstests/concurrency/fsm_all_sharded_replication.js b/jstests/concurrency/fsm_all_sharded_replication.js
index d4068148d47..435277a2a47 100644
--- a/jstests/concurrency/fsm_all_sharded_replication.js
+++ b/jstests/concurrency/fsm_all_sharded_replication.js
@@ -9,6 +9,7 @@ var blacklist = [
     'distinct.js',             // SERVER-13116 distinct isn't sharding aware
     'distinct_noindex.js',     // SERVER-13116 distinct isn't sharding aware
     'distinct_projection.js',  // SERVER-13116 distinct isn't sharding aware
+    'create_database.js',      // SERVER-17397 Drops of sharded namespaces may not fully succeed
     'drop_database.js',        // SERVER-17397 Drops of sharded namespaces may not fully succeed
 
     // Disabled due to SERVER-3645, '.count() can be wrong on sharded collections'.
diff --git a/jstests/concurrency/fsm_all_sharded_replication_with_balancer.js b/jstests/concurrency/fsm_all_sharded_replication_with_balancer.js
index 444f7eab3cb..59d6e0f4a61 100644
--- a/jstests/concurrency/fsm_all_sharded_replication_with_balancer.js
+++ b/jstests/concurrency/fsm_all_sharded_replication_with_balancer.js
@@ -9,6 +9,7 @@ var blacklist = [
     'distinct.js',             // SERVER-13116 distinct isn't sharding aware
     'distinct_noindex.js',     // SERVER-13116 distinct isn't sharding aware
     'distinct_projection.js',  // SERVER-13116 distinct isn't sharding aware
+    'create_database.js',      // SERVER-17397 Drops of sharded namespaces may not fully succeed
     'drop_database.js',        // SERVER-17397 Drops of sharded namespaces may not fully succeed
     'remove_where.js',  // SERVER-14669 Multi-removes that use $where miscount removed documents
 
diff --git a/jstests/concurrency/fsm_workloads/create_database.js b/jstests/concurrency/fsm_workloads/create_database.js
new file mode 100644
index 00000000000..10b619454b4
--- /dev/null
+++ b/jstests/concurrency/fsm_workloads/create_database.js
@@ -0,0 +1,140 @@
+'use strict';
+
+/**
+ * create_database.js
+ *
+ * Repeatedly creates and drops a database, with the focus on creation using different name casing.
+ * Create using all different methods, implicitly by inserting, creating views/indexes etc.
+ *
+ * Each thread uses its own database, though sometimes threads may try to create databases with
+ * names that only differ in case, expecting the appriopriate error code.
+ */
+var $config = (function() {
+
+    var data = {
+        checkCommandResult: function checkCommandResult(mayFailWithDatabaseDifferCase, res) {
+            if (mayFailWithDatabaseDifferCase && !res.ok)
+                assertAlways.commandFailedWithCode(res, ErrorCodes.DatabaseDifferCase);
+            else
+                assertAlways.commandWorked(res);
+            return res;
+        },
+
+        checkWriteResult: function checkWriteResult(mayFailWithDatabaseDifferCase, res) {
+            if (mayFailWithDatabaseDifferCase && res.hasWriteError()) {
+                assertAlways.eq(res.getWriteError().code, ErrorCodes.DatabaseDifferCase);
+            } else
+                assertAlways.writeOK(res);
+            return res;
+        }
+    };
+
+    var states = (function() {
+        function init(db, collName) {
+            var uniqueNr = this.tid;
+            var semiUniqueNr = Math.floor(uniqueNr / 2);
+
+            // The semiUniqueDBName may clash and result in a DatabaseDifferCas error on
+            // creation,
+            // while the uniqueDBName does not clash. The unique and created variables track
+            // this.
+            this.semiUniqueDBName =
+                (this.tid % 2 ? 'create_database' : 'CREATE_DATABASE') + semiUniqueNr;
+            this.uniqueDBName = 'CreateDatabase' + uniqueNr;
+            this.myDB = db.getSiblingDB(this.uniqueDBName);
+            this.created = false;
+            this.unique = true;
+        }
+
+        function useSemiUniqueDBName(db, collName) {
+            this.myDB = db.getSiblingDB(this.semiUniqueDBName);
+            this.unique = false;
+        }
+
+        function createCollection(db, collName) {
+            this.created =
+                this.checkCommandResult(!this.unique, this.myDB.createCollection(collName)).ok;
+        }
+
+        function createIndex(db, collName) {
+            var background = Math.random > 0.5;
+            var res = this.myDB.getCollection(collName).createIndex({x: 1}, {background});
+            this.created |=
+                this.checkCommandResult(!this.unique, res).createdCollectionAutomatically;
+        }
+
+        function insert(db, collName) {
+            this.created |= this.checkWriteResult(!this.created && !this.unique,
+                                                  this.myDB.getCollection(collName).insert({x: 1}))
+                                .nInserted == 1;
+        }
+
+        function upsert(db, collName) {
+            this.created |= this.checkWriteResult(!this.created && !this.unique,
+                                                  this.myDB.getCollection(collName).update(
+                                                      {x: 1}, {x: 2}, {upsert: 1})).nUpserted == 1;
+        }
+
+        function drop(db, collName) {
+            if (this.created)
+                assertAlways(this.myDB.getCollection(collName).drop());
+        }
+
+        function dropDatabase(db, collName) {
+            if (this.created)
+                assertAlways.commandWorked(this.myDB.dropDatabase());
+        }
+
+        function listDatabases(db, collName) {
+            for (var database of db.adminCommand({listDatabases: 1}).databases) {
+                var res = db.getSiblingDB(database.name).runCommand({listCollections: 1});
+                assertAlways.commandWorked(res);
+                assertAlways.neq(database.name, this.myDB.toString(), "this DB shouldn't exist");
+            }
+        }
+
+        return {
+            init: init,
+            useSemiUniqueDBName: useSemiUniqueDBName,
+            createCollection: createCollection,
+            createIndex: createIndex,
+            insert: insert,
+            upsert: upsert,
+            drop: drop,
+            dropDatabase: dropDatabase,
+            listDatabases: listDatabases,
+        };
+    })();
+
+    var transitions = {
+        init: {
+            useSemiUniqueDBName: 0.25,
+            createCollection: 0.375,
+            createIndex: 0.125,
+            insert: 0.125,
+            upsert: 0.125
+        },
+        useSemiUniqueDBName: {createCollection: 1.00},
+        createCollection: {dropDatabase: 0.25, createIndex: 0.25, insert: 0.25, upsert: 0.25},
+        createIndex: {insert: 0.25, upsert: 0.25, dropDatabase: 0.5},
+        insert: {dropDatabase: 0.2, drop: 0.05, insert: 0.5, upsert: 0.25},
+        upsert: {dropDatabase: 0.2, drop: 0.05, insert: 0.25, upsert: 0.5},
+        drop: {dropDatabase: 0.75, init: 0.25},  // OK to leave the empty database behind sometimes
+        dropDatabase: {init: 0.75, listDatabases: 0.25},
+        listDatabases: {init: 0.75, listDatabases: 0.25},
+    };
+
+    return {
+        data: data,
+        // We only run a few iterations to reduce the amount of data cumulatively
+        // written to disk by mmapv1. For example, setting 10 threads and 180
+        // iterations (with an expected 6 transitions per create/drop roundtrip)
+        // causes this workload to write at least 32MB (.ns and .0 files) * 10 threads
+        // * 30 iterations worth of data to disk, or about 10GB, which can be slow on
+        // test hosts.
+        threadCount: 10,
+        iterations: 180,
+        states: states,
+        transitions: transitions,
+    };
+})();
diff --git a/src/mongo/db/catalog/database.cpp b/src/mongo/db/catalog/database.cpp
index a54a5e4798c..9b68549f295 100644
--- a/src/mongo/db/catalog/database.cpp
+++ b/src/mongo/db/catalog/database.cpp
@@ -40,16 +40,14 @@
 #include "mongo/db/audit.h"
 #include "mongo/db/auth/auth_index_d.h"
 #include "mongo/db/background.h"
-#include "mongo/db/clientcursor.h"
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/collection_options.h"
 #include "mongo/db/catalog/database_catalog_entry.h"
 #include "mongo/db/catalog/database_holder.h"
+#include "mongo/db/clientcursor.h"
 #include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/dbhelpers.h"
-#include "mongo/db/service_context.h"
-#include "mongo/db/service_context_d.h"
 #include "mongo/db/index/index_access_method.h"
 #include "mongo/db/instance.h"
 #include "mongo/db/introspect.h"
@@ -57,10 +55,12 @@
 #include "mongo/db/repl/oplog.h"
 #include "mongo/db/repl/replication_coordinator_global.h"
 #include "mongo/db/server_parameters.h"
+#include "mongo/db/service_context.h"
+#include "mongo/db/service_context_d.h"
 #include "mongo/db/stats/top.h"
-#include "mongo/db/storage/storage_options.h"
-#include "mongo/db/storage/storage_engine.h"
 #include "mongo/db/storage/recovery_unit.h"
+#include "mongo/db/storage/storage_engine.h"
+#include "mongo/db/storage/storage_options.h"
 #include "mongo/util/log.h"
 
 namespace mongo {
@@ -215,35 +215,6 @@ Database::Database(OperationContext* txn, StringData name, DatabaseCatalogEntry*
     }
 }
 
-
-/*static*/
-string Database::duplicateUncasedName(const string& name, set<string>* duplicates) {
-    if (duplicates) {
-        duplicates->clear();
-    }
-
-    set<string> allShortNames;
-    dbHolder().getAllShortNames(allShortNames);
-
-    for (const auto& dbname : allShortNames) {
-        if (strcasecmp(dbname.c_str(), name.c_str()))
-            continue;
-
-        if (strcmp(dbname.c_str(), name.c_str()) == 0)
-            continue;
-
-        if (duplicates) {
-            duplicates->insert(dbname);
-        } else {
-            return dbname;
-        }
-    }
-    if (duplicates) {
-        return duplicates->empty() ? "" : *duplicates->begin();
-    }
-    return "";
-}
-
 void Database::clearTmpCollections(OperationContext* txn) {
     invariant(txn->lockState()->isDbLockedForMode(name(), MODE_X));
 
diff --git a/src/mongo/db/catalog/database.h b/src/mongo/db/catalog/database.h
index 53c3cb5712d..2fa20bc9358 100644
--- a/src/mongo/db/catalog/database.h
+++ b/src/mongo/db/catalog/database.h
@@ -167,15 +167,6 @@ public:
                             StringData toNS,
                             bool stayTemp);
 
-    /**
-     * @return name of an existing database with same text name but different
-     * casing, if one exists.  Otherwise the empty std::string is returned.  If
-     * 'duplicates' is specified, it is filled with all duplicate names.
-     // TODO move???
-     */
-    static std::string duplicateUncasedName(const std::string& name,
-                                            std::set<std::string>* duplicates = 0);
-
     static Status validateDBName(StringData dbname);
 
     const std::string& getSystemIndexesName() const {
diff --git a/src/mongo/db/catalog/database_holder.cpp b/src/mongo/db/catalog/database_holder.cpp
index f4a2cf62970..b6633f5c328 100644
--- a/src/mongo/db/catalog/database_holder.cpp
+++ b/src/mongo/db/catalog/database_holder.cpp
@@ -35,14 +35,15 @@
 #include "mongo/db/audit.h"
 #include "mongo/db/auth/auth_index_d.h"
 #include "mongo/db/background.h"
-#include "mongo/db/client.h"
-#include "mongo/db/clientcursor.h"
 #include "mongo/db/catalog/database.h"
 #include "mongo/db/catalog/database_catalog_entry.h"
-#include "mongo/db/service_context.h"
+#include "mongo/db/client.h"
+#include "mongo/db/clientcursor.h"
 #include "mongo/db/operation_context.h"
+#include "mongo/db/service_context.h"
 #include "mongo/db/storage/storage_engine.h"
 #include "mongo/util/log.h"
+#include "mongo/util/scopeguard.h"
 
 namespace mongo {
 
@@ -91,56 +92,81 @@ Database* DatabaseHolder::get(OperationContext* txn, StringData ns) const {
     return NULL;
 }
 
+std::set<std::string> DatabaseHolder::_getNamesWithConflictingCasing_inlock(StringData name) {
+    std::set<std::string> duplicates;
+
+    for (const auto& nameAndPointer : _dbs) {
+        // A name that's equal with case-insensitive match must be identical, or it's a duplicate.
+        if (name.equalCaseInsensitive(nameAndPointer.first) && name != nameAndPointer.first)
+            duplicates.insert(nameAndPointer.first);
+    }
+    return duplicates;
+}
+
+std::set<std::string> DatabaseHolder::getNamesWithConflictingCasing(StringData name) {
+    stdx::lock_guard<SimpleMutex> lk(_m);
+    return _getNamesWithConflictingCasing_inlock(name);
+}
+
 Database* DatabaseHolder::openDb(OperationContext* txn, StringData ns, bool* justCreated) {
     const StringData dbname = _todb(ns);
     invariant(txn->lockState()->isDbLockedForMode(dbname, MODE_X));
 
-    Database* db = get(txn, ns);
-    if (db) {
-        if (justCreated) {
-            *justCreated = false;
-        }
+    if (justCreated)
+        *justCreated = false;  // Until proven otherwise.
 
-        return db;
-    }
+    stdx::unique_lock<SimpleMutex> lk(_m);
 
-    // Check casing
-    const string duplicate = Database::duplicateUncasedName(dbname.toString());
-    if (!duplicate.empty()) {
-        stringstream ss;
-        ss << "db already exists with different case already have: [" << duplicate
-           << "] trying to create [" << dbname.toString() << "]";
-        uasserted(ErrorCodes::DatabaseDifferCase, ss.str());
-    }
+    // The following will insert a nullptr for dbname, which will treated the same as a non-
+    // existant database by the get method, yet still counts in getNamesWithConflictingCasing.
+    if (auto db = _dbs[dbname])
+        return db;
 
+    // We've inserted a nullptr entry for dbname: make sure to remove it on unsuccessful exit.
+    auto removeDbGuard = MakeGuard([this, &lk, dbname] {
+        if (!lk.owns_lock())
+            lk.lock();
+        _dbs.erase(dbname);
+    });
+
+    // Check casing in lock to avoid transient duplicates.
+    auto duplicates = _getNamesWithConflictingCasing_inlock(dbname);
+    uassert(ErrorCodes::DatabaseDifferCase,
+            str::stream() << "db already exists with different case already have: ["
+                          << *duplicates.cbegin() << "] trying to create [" << dbname.toString()
+                          << "]",
+            duplicates.empty());
+
+
+    // Do the catalog lookup and database creation outside of the scoped lock, because these may
+    // block. Only one thread can be inside this method for the same DB name, because of the
+    // requirement for X-lock on the database when we enter. So there is no way we can insert two
+    // different databases for the same name.
+    lk.unlock();
     StorageEngine* storageEngine = getGlobalServiceContext()->getGlobalStorageEngine();
-    invariant(storageEngine);
-
     DatabaseCatalogEntry* entry = storageEngine->getDatabaseCatalogEntry(txn, dbname);
-    invariant(entry);
-    const bool exists = entry->exists();
-    if (!exists) {
-        audit::logCreateDatabase(&cc(), dbname);
-    }
 
-    if (justCreated) {
-        *justCreated = !exists;
+    if (!entry->exists()) {
+        audit::logCreateDatabase(&cc(), dbname);
+        if (justCreated)
+            *justCreated = true;
     }
 
-    // Do this outside of the scoped lock, because database creation does transactional
-    // operations which may block. Only one thread can be inside this method for the same DB
-    // name, because of the requirement for X-lock on the database when we enter. So there is
-    // no way we can insert two different databases for the same name.
-    db = new Database(txn, dbname, entry);
+    auto newDb = stdx::make_unique<Database>(txn, dbname, entry);
 
-    stdx::lock_guard<SimpleMutex> lk(_m);
-    _dbs[dbname] = db;
+    // Finally replace our nullptr entry with the new Database pointer.
+    removeDbGuard.Dismiss();
+    lk.lock();
+    auto it = _dbs.find(dbname);
+    invariant(it != _dbs.end() && it->second == nullptr);
+    Database* newDbPointer = newDb.release();
+    _dbs[dbname] = newDbPointer;
+    invariant(_getNamesWithConflictingCasing_inlock(dbname.toString()).empty());
 
-    return db;
+    return newDbPointer;
 }
 
 void DatabaseHolder::close(OperationContext* txn, StringData ns) {
-    // TODO: This should be fine if only a DB X-lock
     invariant(txn->lockState()->isW());
 
     const StringData dbName = _todb(ns);
diff --git a/src/mongo/db/catalog/database_holder.h b/src/mongo/db/catalog/database_holder.h
index ce99747937b..91850197771 100644
--- a/src/mongo/db/catalog/database_holder.h
+++ b/src/mongo/db/catalog/database_holder.h
@@ -29,6 +29,7 @@
 #pragma once
 
 #include <set>
+#include <string>
 
 #include "mongo/base/string_data.h"
 #include "mongo/db/namespace_string.h"
@@ -87,9 +88,15 @@ public:
         }
     }
 
+    /**
+     * Returns the set of existing database names that differ only in casing.
+     */
+    std::set<std::string> getNamesWithConflictingCasing(StringData name);
+
 private:
-    typedef StringMap<Database*> DBs;
+    std::set<std::string> _getNamesWithConflictingCasing_inlock(StringData name);
 
+    typedef StringMap<Database*> DBs;
     mutable SimpleMutex _m;
     DBs _dbs;
 };
diff --git a/src/mongo/db/repl/master_slave.cpp b/src/mongo/db/repl/master_slave.cpp
index 67a156cbce7..4ba4193e462 100644
--- a/src/mongo/db/repl/master_slave.cpp
+++ b/src/mongo/db/repl/master_slave.cpp
@@ -537,7 +537,7 @@ bool ReplSource::handleDuplicateDbName(OperationContext* txn,
         // missing from master after optime "ts".
         return false;
     }
-    if (Database::duplicateUncasedName(db).empty()) {
+    if (dbHolder().getNamesWithConflictingCasing(db).empty()) {
         // No duplicate database names are present.
         return true;
     }
@@ -594,8 +594,7 @@ bool ReplSource::handleDuplicateDbName(OperationContext* txn,
     }
 
     // Check for duplicates again, since we released the lock above.
-    set<string> duplicates;
-    Database::duplicateUncasedName(db, &duplicates);
+    auto duplicates = dbHolder().getNamesWithConflictingCasing(db);
 
     // The database is present on the master and no conflicting databases
     // are present on the master.  Drop any local conflicts.
@@ -610,7 +609,7 @@ bool ReplSource::handleDuplicateDbName(OperationContext* txn,
 
     massert(14034,
             "Duplicate database names present after attempting to delete duplicates",
-            Database::duplicateUncasedName(db).empty());
+            dbHolder().getNamesWithConflictingCasing(db).empty());
     return true;
 }