SERVER-22611 Sharding catalog cache refactor

46 files changed, 2732 insertions, 2427 deletions

diff --git a/src/mongo/s/SConscript b/src/mongo/s/SConscript
index f6ef30c691d..eede0298c6d 100644
--- a/src/mongo/s/SConscript
+++ b/src/mongo/s/SConscript
@@ -243,12 +243,10 @@ env.Library(
         'chunk.cpp',
         'chunk_manager.cpp',
         'cluster_identity_loader.cpp',
-        'config.cpp',
         'config_server_client.cpp',
         'grid.cpp',
         'shard_util.cpp',
         'sharding_egress_metadata_hook.cpp',
-        'sharding_raii.cpp',
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/db/audit',
@@ -265,8 +263,10 @@ env.Library(
 env.CppUnitTest(
     target='catalog_cache_test',
     source=[
-        'chunk_manager_test.cpp',
         'chunk_manager_index_bounds_test.cpp',
+        'chunk_manager_query_test.cpp',
+        'chunk_manager_refresh_test.cpp',
+        'chunk_manager_test_fixture.cpp',
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/s/catalog/sharding_catalog_test_fixture',
diff --git a/src/mongo/s/catalog/type_chunk.cpp b/src/mongo/s/catalog/type_chunk.cpp
index ff6df77d6f5..e5091f16954 100644
--- a/src/mongo/s/catalog/type_chunk.cpp
+++ b/src/mongo/s/catalog/type_chunk.cpp
@@ -132,6 +132,15 @@ bool ChunkRange::operator!=(const ChunkRange& other) const {
     return !(*this == other);
 }
 
+ChunkType::ChunkType() = default;
+
+ChunkType::ChunkType(NamespaceString nss, ChunkRange range, ChunkVersion version, ShardId shardId)
+    : _ns(nss.ns()),
+      _min(range.getMin()),
+      _max(range.getMax()),
+      _version(version),
+      _shard(std::move(shardId)) {}
+
 StatusWith<ChunkType> ChunkType::fromConfigBSON(const BSONObj& source) {
     ChunkType chunk;
 
diff --git a/src/mongo/s/catalog/type_chunk.h b/src/mongo/s/catalog/type_chunk.h
index 06a26db34be..6484f97b03b 100644
--- a/src/mongo/s/catalog/type_chunk.h
+++ b/src/mongo/s/catalog/type_chunk.h
@@ -32,6 +32,7 @@
 #include <string>
 
 #include "mongo/bson/bsonobj.h"
+#include "mongo/db/namespace_string.h"
 #include "mongo/s/chunk_version.h"
 #include "mongo/s/shard_id.h"
 
@@ -143,6 +144,9 @@ public:
     static const BSONField<Date_t> DEPRECATED_lastmod;
     static const BSONField<OID> DEPRECATED_epoch;
 
+    ChunkType();
+    ChunkType(NamespaceString nss, ChunkRange range, ChunkVersion version, ShardId shardId);
+
     /**
      * Constructs a new ChunkType object from BSON that has the config server's config.chunks
      * collection format.
diff --git a/src/mongo/s/catalog_cache.cpp b/src/mongo/s/catalog_cache.cpp
index d2c8eaf5504..c3607d6053f 100644
--- a/src/mongo/s/catalog_cache.cpp
+++ b/src/mongo/s/catalog_cache.cpp
@@ -26,64 +26,391 @@
  *    it in the license file.
  */
 
+#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kSharding
+
 #include "mongo/platform/basic.h"
 
 #include "mongo/s/catalog_cache.h"
 
+#include "mongo/base/status.h"
 #include "mongo/base/status_with.h"
+#include "mongo/db/query/collation/collator_factory_interface.h"
+#include "mongo/db/repl/optime_with.h"
 #include "mongo/s/catalog/sharding_catalog_client.h"
+#include "mongo/s/catalog/type_collection.h"
 #include "mongo/s/catalog/type_database.h"
-#include "mongo/s/config.h"
+#include "mongo/s/chunk_diff.h"
+#include "mongo/s/client/shard_registry.h"
 #include "mongo/s/grid.h"
+#include "mongo/stdx/memory.h"
+#include "mongo/util/log.h"
+#include "mongo/util/timer.h"
 
 namespace mongo {
+namespace {
 
-using std::shared_ptr;
-using std::string;
+// How many times to try refreshing the routing info if the set of chunks loaded from the config
+// server is found to be inconsistent.
+const int kMaxInconsistentRoutingInfoRefreshAttempts = 3;
 
-CatalogCache::CatalogCache() = default;
+/**
+ * This is an adapter so we can use config diffs - mongos and mongod do them slightly differently.
+ *
+ * The mongos adapter here tracks all shards, and stores ranges by (max, Chunk) in the map.
+ */
+class CMConfigDiffTracker : public ConfigDiffTracker<std::shared_ptr<Chunk>> {
+public:
+    CMConfigDiffTracker(const NamespaceString& nss,
+                        RangeMap* currMap,
+                        ChunkVersion* maxVersion,
+                        MaxChunkVersionMap* maxShardVersions)
+        : ConfigDiffTracker<std::shared_ptr<Chunk>>(
+              nss.ns(), currMap, maxVersion, maxShardVersions) {}
 
-CatalogCache::~CatalogCache() = default;
+    bool isTracked(const ChunkType& chunk) const final {
+        // Mongos tracks all shards
+        return true;
+    }
 
-StatusWith<std::shared_ptr<DBConfig>> CatalogCache::getDatabase(OperationContext* opCtx,
-                                                                StringData dbName) {
-    stdx::lock_guard<stdx::mutex> guard(_mutex);
+    bool isMinKeyIndexed() const final {
+        return false;
+    }
 
-    auto it = _databases.find(dbName);
-    if (it != _databases.end()) {
-        return it->second;
+    std::pair<BSONObj, std::shared_ptr<Chunk>> rangeFor(OperationContext* opCtx,
+                                                        const ChunkType& chunk) const final {
+        return std::make_pair(chunk.getMax(), std::make_shared<Chunk>(chunk));
     }
 
-    // Need to load from the store
-    auto status = Grid::get(opCtx)->catalogClient(opCtx)->getDatabase(opCtx, dbName.toString());
-    if (!status.isOK()) {
-        return status.getStatus();
+    ShardId shardFor(OperationContext* opCtx, const ShardId& shardId) const final {
+        const auto shard =
+            uassertStatusOK(Grid::get(opCtx)->shardRegistry()->getShard(opCtx, shardId));
+        return shard->getId();
     }
+};
+
+}  // namespace
+
+CatalogCache::CatalogCache() = default;
+
+CatalogCache::~CatalogCache() = default;
+
+StatusWith<CachedDatabaseInfo> CatalogCache::getDatabase(OperationContext* opCtx,
+                                                         StringData dbName) {
+    stdx::lock_guard<stdx::mutex> lg(_mutex);
 
-    const auto& dbOpTimePair = status.getValue();
-    auto db = std::make_shared<DBConfig>(dbOpTimePair.value, dbOpTimePair.opTime);
     try {
-        db->load(opCtx);
-        auto emplaceResult = _databases.try_emplace(dbName, std::move(db));
-        return emplaceResult.first->second;
+        return {CachedDatabaseInfo(_getDatabase_inlock(opCtx, dbName))};
     } catch (const DBException& ex) {
         return ex.toStatus();
     }
 }
 
-void CatalogCache::invalidate(StringData dbName) {
-    stdx::lock_guard<stdx::mutex> guard(_mutex);
+StatusWith<CachedCollectionRoutingInfo> CatalogCache::getCollectionRoutingInfo(
+    OperationContext* opCtx, const NamespaceString& nss) {
+    int numRefreshAttempts = 0;
 
-    ShardedDatabasesMap::iterator it = _databases.find(dbName);
-    if (it != _databases.end()) {
-        _databases.erase(it);
+    while (true) {
+        stdx::unique_lock<stdx::mutex> ul(_mutex);
+
+        std::shared_ptr<DatabaseInfoEntry> dbEntry;
+        try {
+            dbEntry = _getDatabase_inlock(opCtx, nss.db());
+        } catch (const DBException& ex) {
+            return ex.toStatus();
+        }
+
+        auto& collections = dbEntry->collections;
+
+        auto it = collections.find(nss.ns());
+        if (it == collections.end()) {
+            auto shardStatus =
+                Grid::get(opCtx)->shardRegistry()->getShard(opCtx, dbEntry->primaryShardId);
+            if (!shardStatus.isOK()) {
+                return {ErrorCodes::fromInt(40371),
+                        str::stream() << "The primary shard for collection " << nss.ns()
+                                      << " could not be loaded due to error "
+                                      << shardStatus.getStatus().toString()};
+            }
+
+            return {CachedCollectionRoutingInfo(
+                dbEntry->primaryShardId, nss, std::move(shardStatus.getValue()))};
+        }
+
+        auto& collEntry = it->second;
+
+        if (collEntry.needsRefresh) {
+            numRefreshAttempts++;
+
+            try {
+                auto newRoutingInfo =
+                    refreshCollectionRoutingInfo(opCtx, nss, std::move(collEntry.routingInfo));
+                if (newRoutingInfo == nullptr) {
+                    collections.erase(it);
+
+                    // Loop around so we can return an "unsharded" routing info
+                    continue;
+                }
+
+                collEntry.routingInfo = std::move(newRoutingInfo);
+                collEntry.needsRefresh = false;
+            } catch (const DBException& ex) {
+                // It is possible that the metadata is being changed concurrently, so retry the
+                // refresh with a wait
+                if (ex.getCode() == ErrorCodes::ConflictingOperationInProgress &&
+                    numRefreshAttempts < kMaxInconsistentRoutingInfoRefreshAttempts) {
+                    ul.unlock();
+
+                    log() << "Metadata refresh for " << nss.ns() << " failed and will be retried"
+                          << causedBy(redact(ex));
+
+                    // Do the sleep outside of the mutex
+                    sleepFor(Milliseconds(10) * numRefreshAttempts);
+                    continue;
+                }
+
+                return ex.toStatus();
+            }
+        }
+
+        return {CachedCollectionRoutingInfo(dbEntry->primaryShardId, collEntry.routingInfo)};
+    }
+}
+
+StatusWith<CachedCollectionRoutingInfo> CatalogCache::getCollectionRoutingInfo(
+    OperationContext* opCtx, StringData ns) {
+    return getCollectionRoutingInfo(opCtx, NamespaceString(ns));
+}
+
+StatusWith<CachedCollectionRoutingInfo> CatalogCache::getShardedCollectionRoutingInfoWithRefresh(
+    OperationContext* opCtx, const NamespaceString& nss) {
+    invalidateShardedCollection(nss);
+
+    auto routingInfoStatus = getCollectionRoutingInfo(opCtx, nss);
+    if (routingInfoStatus.isOK() && !routingInfoStatus.getValue().cm()) {
+        return {ErrorCodes::NamespaceNotSharded,
+                str::stream() << "Collection " << nss.ns() << " is not sharded."};
     }
+
+    return routingInfoStatus;
 }
 
-void CatalogCache::invalidateAll() {
-    stdx::lock_guard<stdx::mutex> guard(_mutex);
+StatusWith<CachedCollectionRoutingInfo> CatalogCache::getShardedCollectionRoutingInfoWithRefresh(
+    OperationContext* opCtx, StringData ns) {
+    return getShardedCollectionRoutingInfoWithRefresh(opCtx, NamespaceString(ns));
+}
+
+void CatalogCache::onStaleConfigError(CachedCollectionRoutingInfo&& ccrt) {
+    if (!ccrt._cm) {
+        // Here we received a stale config error for a collection which we previously thought was
+        // unsharded.
+        invalidateShardedCollection(ccrt._nss);
+        return;
+    }
 
+    // Here we received a stale config error for a collection which we previously though was sharded
+    stdx::lock_guard<stdx::mutex> lg(_mutex);
+
+    auto it = _databases.find(NamespaceString(ccrt._cm->getns()).db());
+    if (it == _databases.end()) {
+        // If the database does not exist, the collection must have been dropped so there is
+        // nothing to invalidate. The getCollectionRoutingInfo will handle the reload of the
+        // entire database and its collections.
+        return;
+    }
+
+    auto& collections = it->second->collections;
+
+    auto itColl = collections.find(ccrt._cm->getns());
+    if (itColl == collections.end()) {
+        // If the collection does not exist, this means it must have been dropped since the last
+        // time we retrieved a cache entry for it. Doing nothing in this case will cause the
+        // next call to getCollectionRoutingInfo to return an unsharded collection.
+        return;
+    } else if (itColl->second.routingInfo->getVersion() == ccrt._cm->getVersion()) {
+        // If the versions match, the last version of the routing information that we used is no
+        // longer valid, so trigger a refresh.
+        itColl->second.needsRefresh = true;
+    }
+}
+
+void CatalogCache::invalidateShardedCollection(const NamespaceString& nss) {
+    stdx::lock_guard<stdx::mutex> lg(_mutex);
+
+    auto it = _databases.find(nss.db());
+    if (it == _databases.end()) {
+        return;
+    }
+
+    it->second->collections[nss.ns()].needsRefresh = true;
+}
+
+void CatalogCache::invalidateShardedCollection(StringData ns) {
+    invalidateShardedCollection(NamespaceString(ns));
+}
+
+void CatalogCache::purgeDatabase(StringData dbName) {
+    stdx::lock_guard<stdx::mutex> lg(_mutex);
+
+    auto it = _databases.find(dbName);
+    if (it == _databases.end()) {
+        return;
+    }
+
+    _databases.erase(it);
+}
+
+void CatalogCache::purgeAllDatabases() {
+    stdx::lock_guard<stdx::mutex> lg(_mutex);
     _databases.clear();
 }
 
+std::shared_ptr<ChunkManager> CatalogCache::refreshCollectionRoutingInfo(
+    OperationContext* opCtx,
+    const NamespaceString& nss,
+    std::shared_ptr<ChunkManager> existingRoutingInfo) {
+    Timer t;
+
+    const auto catalogClient = Grid::get(opCtx)->catalogClient(opCtx);
+
+    // Decide whether to do a full or partial load based on the state of the collection
+    auto collStatus = catalogClient->getCollection(opCtx, nss.ns());
+    if (collStatus == ErrorCodes::NamespaceNotFound) {
+        return nullptr;
+    }
+
+    const auto coll = uassertStatusOK(std::move(collStatus)).value;
+    if (coll.getDropped()) {
+        return nullptr;
+    }
+
+    ChunkVersion startingCollectionVersion;
+    ChunkMap chunkMap =
+        SimpleBSONObjComparator::kInstance.makeBSONObjIndexedMap<std::shared_ptr<Chunk>>();
+
+    if (!existingRoutingInfo) {
+        // If we don't have a basis chunk manager, do a full refresh
+        startingCollectionVersion = ChunkVersion(0, 0, coll.getEpoch());
+    } else if (existingRoutingInfo->getVersion().epoch() != coll.getEpoch()) {
+        // If the collection's epoch has changed, do a full refresh
+        startingCollectionVersion = ChunkVersion(0, 0, coll.getEpoch());
+    } else {
+        // Otherwise do a partial refresh
+        startingCollectionVersion = existingRoutingInfo->getVersion();
+        chunkMap = existingRoutingInfo->chunkMap();
+    }
+
+    log() << "Refreshing chunks based on version " << startingCollectionVersion;
+
+    // Diff tracker should *always* find at least one chunk if collection exists
+    const auto diffQuery =
+        CMConfigDiffTracker::createConfigDiffQuery(nss, startingCollectionVersion);
+
+    // Query the chunks which have changed
+    std::vector<ChunkType> newChunks;
+    repl::OpTime opTime;
+    uassertStatusOK(Grid::get(opCtx)->catalogClient(opCtx)->getChunks(
+        opCtx,
+        diffQuery.query,
+        diffQuery.sort,
+        boost::none,
+        &newChunks,
+        &opTime,
+        repl::ReadConcernLevel::kMajorityReadConcern));
+
+    ChunkVersion collectionVersion = startingCollectionVersion;
+
+    ShardVersionMap unusedShardVersions;
+    CMConfigDiffTracker differ(nss, &chunkMap, &collectionVersion, &unusedShardVersions);
+
+    const int diffsApplied = differ.calculateConfigDiff(opCtx, newChunks);
+
+    if (diffsApplied < 1) {
+        log() << "Refresh took " << t.millis() << " ms and failed because the collection's "
+                                                  "sharding metadata either changed in between or "
+                                                  "became corrupted";
+
+        uasserted(ErrorCodes::ConflictingOperationInProgress,
+                  "Collection sharding status changed during refresh or became corrupted");
+    }
+
+    // If at least one diff was applied, the metadata is correct, but it might not have changed so
+    // in this case there is no need to recreate the chunk manager.
+    //
+    // NOTE: In addition to the above statement, it is also important that we return the same chunk
+    // manager object, because the write commands' code relies on changes of the chunk manager's
+    // sequence number to detect batch writes not making progress because of chunks moving across
+    // shards too frequently.
+    if (collectionVersion == startingCollectionVersion) {
+        log() << "Refresh took " << t.millis() << " ms and didn't find any metadata changes";
+
+        return existingRoutingInfo;
+    }
+
+    std::unique_ptr<CollatorInterface> defaultCollator;
+    if (!coll.getDefaultCollation().isEmpty()) {
+        // The collation should have been validated upon collection creation
+        defaultCollator = uassertStatusOK(CollatorFactoryInterface::get(opCtx->getServiceContext())
+                                              ->makeFromBSON(coll.getDefaultCollation()));
+    }
+
+    log() << "Refresh took " << t.millis() << " ms and found version " << collectionVersion;
+
+    return stdx::make_unique<ChunkManager>(nss,
+                                           coll.getKeyPattern(),
+                                           std::move(defaultCollator),
+                                           coll.getUnique(),
+                                           std::move(chunkMap),
+                                           collectionVersion);
+}
+
+std::shared_ptr<CatalogCache::DatabaseInfoEntry> CatalogCache::_getDatabase_inlock(
+    OperationContext* opCtx, StringData dbName) {
+    auto it = _databases.find(dbName);
+    if (it != _databases.end()) {
+        return it->second;
+    }
+
+    const auto catalogClient = Grid::get(opCtx)->catalogClient(opCtx);
+
+    const auto dbNameCopy = dbName.toString();
+
+    // Load the database entry
+    const auto opTimeWithDb = uassertStatusOK(catalogClient->getDatabase(opCtx, dbNameCopy));
+    const auto& dbDesc = opTimeWithDb.value;
+
+    // Load the sharded collections entries
+    std::vector<CollectionType> collections;
+    repl::OpTime collLoadConfigOptime;
+    uassertStatusOK(
+        catalogClient->getCollections(opCtx, &dbNameCopy, &collections, &collLoadConfigOptime));
+
+    StringMap<CollectionRoutingInfoEntry> collectionEntries;
+    for (const auto& coll : collections) {
+        collectionEntries[coll.getNs().ns()].needsRefresh = true;
+    }
+
+    return _databases[dbName] = std::shared_ptr<DatabaseInfoEntry>(new DatabaseInfoEntry{
+               dbDesc.getPrimary(), dbDesc.getSharded(), std::move(collectionEntries)});
+}
+
+CachedDatabaseInfo::CachedDatabaseInfo(std::shared_ptr<CatalogCache::DatabaseInfoEntry> db)
+    : _db(std::move(db)) {}
+
+const ShardId& CachedDatabaseInfo::primaryId() const {
+    return _db->primaryShardId;
+}
+
+bool CachedDatabaseInfo::shardingEnabled() const {
+    return _db->shardingEnabled;
+}
+
+CachedCollectionRoutingInfo::CachedCollectionRoutingInfo(ShardId primaryId,
+                                                         std::shared_ptr<ChunkManager> cm)
+    : _primaryId(std::move(primaryId)), _cm(std::move(cm)) {}
+
+CachedCollectionRoutingInfo::CachedCollectionRoutingInfo(ShardId primaryId,
+                                                         NamespaceString nss,
+                                                         std::shared_ptr<Shard> primary)
+    : _primaryId(std::move(primaryId)), _nss(std::move(nss)), _primary(std::move(primary)) {}
+
 }  // namespace mongo
diff --git a/src/mongo/s/catalog_cache.h b/src/mongo/s/catalog_cache.h
index 0e63f94b52a..528b2df4673 100644
--- a/src/mongo/s/catalog_cache.h
+++ b/src/mongo/s/catalog_cache.h
@@ -28,19 +28,20 @@
 
 #pragma once
 
-#include <memory>
-
 #include "mongo/base/disallow_copying.h"
 #include "mongo/base/string_data.h"
+#include "mongo/s/chunk_manager.h"
+#include "mongo/s/chunk_version.h"
+#include "mongo/s/client/shard.h"
 #include "mongo/stdx/mutex.h"
+#include "mongo/util/concurrency/notification.h"
 #include "mongo/util/string_map.h"
 
 namespace mongo {
 
-class DBConfig;
+class CachedDatabaseInfo;
+class CachedCollectionRoutingInfo;
 class OperationContext;
-template <typename T>
-class StatusWith;
 
 /**
  * This is the root of the "read-only" hierarchy of cached catalog metadata. It is read only
@@ -62,26 +63,184 @@ public:
      *
      * Returns the database cache entry if the database exists or a failed status otherwise.
      */
-    StatusWith<std::shared_ptr<DBConfig>> getDatabase(OperationContext* opCtx, StringData dbName);
+    StatusWith<CachedDatabaseInfo> getDatabase(OperationContext* opCtx, StringData dbName);
+
+    /**
+     * Blocking shortcut method to get a specific sharded collection from a given database using the
+     * complete namespace. If the collection is sharded returns a ScopedChunkManager initialized
+     * with ChunkManager. If the collection is not sharded, returns a ScopedChunkManager initialized
+     * with the primary shard for the specified database. If an error occurs loading the metadata
+     * returns a failed status.
+     */
+    StatusWith<CachedCollectionRoutingInfo> getCollectionRoutingInfo(OperationContext* opCtx,
+                                                                     const NamespaceString& nss);
+    StatusWith<CachedCollectionRoutingInfo> getCollectionRoutingInfo(OperationContext* opCtx,
+                                                                     StringData ns);
+
+    /**
+     * Same as getCollectionRoutingInfo above, but in addition causes the namespace to be refreshed
+     * and returns a NamespaceNotSharded error if the collection is not sharded.
+     */
+    StatusWith<CachedCollectionRoutingInfo> getShardedCollectionRoutingInfoWithRefresh(
+        OperationContext* opCtx, const NamespaceString& nss);
+    StatusWith<CachedCollectionRoutingInfo> getShardedCollectionRoutingInfoWithRefresh(
+        OperationContext* opCtx, StringData ns);
+
+    /**
+     * Non-blocking method to be called whenever using the specified routing table has encountered a
+     * stale config exception. Returns immediately and causes the routing table to be refreshed the
+     * next time getCollectionRoutingInfo is called. Does nothing if the routing table has been
+     * refreshed already.
+     */
+    void onStaleConfigError(CachedCollectionRoutingInfo&&);
+
+    /**
+     * Non-blocking method, which indiscriminately causes the routing table for the specified
+     * namespace to be refreshed the next time getCollectionRoutingInfo is called.
+     */
+    void invalidateShardedCollection(const NamespaceString& nss);
+    void invalidateShardedCollection(StringData ns);
+
+    /**
+     * Blocking method, which removes the entire specified database (including its collections) from
+     * the cache.
+     */
+    void purgeDatabase(StringData dbName);
 
     /**
-     * Removes the database information for the specified name from the cache, so that the
-     * next time getDatabase is called, it will be reloaded.
+     * Blocking method, which removes all databases (including their collections) from the cache.
      */
-    void invalidate(StringData dbName);
+    void purgeAllDatabases();
 
     /**
-     * Purges all cached database information, which will cause the data to be reloaded again.
+     * Blocking method, which refreshes the routing information for the specified collection. If
+     * 'existingRoutingInfo' has been specified uses this as a basis to perform an 'incremental'
+     * refresh, which only fetches the chunks which changed. Otherwise does a full refresh, fetching
+     * all the chunks for the collection.
+     *
+     * Returns the refreshed routing information if the collection is still sharded or nullptr if it
+     * is not. If refresh fails for any reason, throws a DBException.
+     *
+     * With the exception of ConflictingOperationInProgress, error codes thrown from this method are
+     * final in that there is nothing that can be done to remedy them other than pass the error to
+     * the user.
+     *
+     * ConflictingOperationInProgress indicates that the chunk metadata was found to be
+     * inconsistent. Since this may be transient, due to the collection being dropped or recreated,
+     * the caller must retry the reload up to some configurable number of attempts.
+     *
+     * NOTE: Should never be called directly and is exposed as public for testing purposes only.
      */
-    void invalidateAll();
+    static std::shared_ptr<ChunkManager> refreshCollectionRoutingInfo(
+        OperationContext* opCtx,
+        const NamespaceString& nss,
+        std::shared_ptr<ChunkManager> existingRoutingInfo);
 
 private:
-    using ShardedDatabasesMap = StringMap<std::shared_ptr<DBConfig>>;
+    // Make the cache entries friends so they can access the private classes below
+    friend class CachedDatabaseInfo;
+    friend class CachedCollectionRoutingInfo;
+
+    /**
+     * Cache entry describing a collection.
+     */
+    struct CollectionRoutingInfoEntry {
+        std::shared_ptr<ChunkManager> routingInfo;
+
+        bool needsRefresh{true};
+    };
+
+    /**
+     * Cache entry describing a database.
+     */
+    struct DatabaseInfoEntry {
+        ShardId primaryShardId;
+
+        bool shardingEnabled;
+
+        StringMap<CollectionRoutingInfoEntry> collections;
+    };
+
+    using DatabaseInfoMap = StringMap<std::shared_ptr<DatabaseInfoEntry>>;
+
+    /**
+     * Ensures that the specified database is in the cache, loading it if necessary. If the database
+     * was not in cache, all the sharded collections will be in the 'needsRefresh' state.
+     */
+    std::shared_ptr<DatabaseInfoEntry> _getDatabase_inlock(OperationContext* opCtx,
+                                                           StringData dbName);
 
     // Mutex to serialize access to the structures below
     stdx::mutex _mutex;
 
-    ShardedDatabasesMap _databases;
+    // Map from DB name to the info for that database
+    DatabaseInfoMap _databases;
+};
+
+/**
+ * Constructed exclusively by the CatalogCache, contains a reference to the cached information for
+ * the specified database.
+ */
+class CachedDatabaseInfo {
+public:
+    const ShardId& primaryId() const;
+
+    bool shardingEnabled() const;
+
+private:
+    friend class CatalogCache;
+
+    CachedDatabaseInfo(std::shared_ptr<CatalogCache::DatabaseInfoEntry> db);
+
+    std::shared_ptr<CatalogCache::DatabaseInfoEntry> _db;
+};
+
+/**
+ * Constructed exclusively by the CatalogCache contains a reference to the routing information for
+ * the specified collection.
+ */
+class CachedCollectionRoutingInfo {
+public:
+    /**
+     * Returns the ID of the primary shard for the database owining this collection, regardless of
+     * whether it is sharded or not.
+     */
+    const ShardId& primaryId() const {
+        return _primaryId;
+    }
+
+    /**
+     * If the collection is sharded, returns a chunk manager for it. Otherwise, nullptr.
+     */
+    std::shared_ptr<ChunkManager> cm() const {
+        return _cm;
+    }
+
+    /**
+     * If the collection is not sharded, returns its primary shard. Otherwise, nullptr.
+     */
+    std::shared_ptr<Shard> primary() const {
+        return _primary;
+    }
+
+private:
+    friend class CatalogCache;
+
+    CachedCollectionRoutingInfo(ShardId primaryId, std::shared_ptr<ChunkManager> cm);
+
+    CachedCollectionRoutingInfo(ShardId primaryId,
+                                NamespaceString nss,
+                                std::shared_ptr<Shard> primary);
+
+    // The id of the primary shard containing the database
+    ShardId _primaryId;
+
+    // Reference to the corresponding chunk manager (if sharded) or null
+    std::shared_ptr<ChunkManager> _cm;
+
+    // Reference to the primary of the database (if not sharded) or null
+    NamespaceString _nss;
+    std::shared_ptr<Shard> _primary;
 };
 
 }  // namespace mongo
diff --git a/src/mongo/s/chunk_diff.cpp b/src/mongo/s/chunk_diff.cpp
index f21555043ad..07d569a503d 100644
--- a/src/mongo/s/chunk_diff.cpp
+++ b/src/mongo/s/chunk_diff.cpp
@@ -105,7 +105,7 @@ int ConfigDiffTracker<ValType>::calculateConfigDiff(OperationContext* opCtx,
     // Store epoch now so it doesn't change when we change max
     OID currEpoch = _maxVersion->epoch();
 
-    _validDiffs = 0;
+    int validDiffs = 0;
 
     for (const ChunkType& chunk : chunks) {
         const ChunkVersion& chunkVersion = chunk.getVersion();
@@ -121,7 +121,7 @@ int ConfigDiffTracker<ValType>::calculateConfigDiff(OperationContext* opCtx,
             return -1;
         }
 
-        _validDiffs++;
+        validDiffs++;
 
         // Get max changed version and chunk version
         if (chunkVersion > *_maxVersion) {
@@ -151,7 +151,7 @@ int ConfigDiffTracker<ValType>::calculateConfigDiff(OperationContext* opCtx,
         }
     }
 
-    LOG(3) << "found " << _validDiffs << " new chunks for collection " << _ns << " (tracking "
+    LOG(3) << "found " << validDiffs << " new chunks for collection " << _ns << " (tracking "
            << newTracked.size() << "), new version is " << *_maxVersion;
 
     for (const ChunkType& chunk : newTracked) {
@@ -167,7 +167,7 @@ int ConfigDiffTracker<ValType>::calculateConfigDiff(OperationContext* opCtx,
         _currMap->insert(rangeFor(opCtx, chunk));
     }
 
-    return _validDiffs;
+    return validDiffs;
 }
 
 ConfigDiffTrackerBase::QueryAndSort ConfigDiffTrackerBase::createConfigDiffQuery(
diff --git a/src/mongo/s/chunk_diff.h b/src/mongo/s/chunk_diff.h
index 0cea9fa678a..5910cc93eed 100644
--- a/src/mongo/s/chunk_diff.h
+++ b/src/mongo/s/chunk_diff.h
@@ -93,12 +93,8 @@ public:
                       RangeMap* currMap,
                       ChunkVersion* maxVersion,
                       MaxChunkVersionMap* maxShardVersions);
-    virtual ~ConfigDiffTracker();
 
-    // Call after load for more information
-    int numValidDiffs() const {
-        return _validDiffs;
-    }
+    virtual ~ConfigDiffTracker();
 
     // Applies changes to the config data from a vector of chunks passed in. Also includes minor
     // version changes for particular major-version chunks if explicitly specified.
@@ -135,9 +131,6 @@ private:
     RangeMap* const _currMap;
     ChunkVersion* const _maxVersion;
     MaxChunkVersionMap* const _maxShardVersions;
-
-    // Store for later use
-    int _validDiffs{0};
 };
 
 }  // namespace mongo
diff --git a/src/mongo/s/chunk_manager.cpp b/src/mongo/s/chunk_manager.cpp
index 20cfd7e098f..047bfa1d696 100644
--- a/src/mongo/s/chunk_manager.cpp
+++ b/src/mongo/s/chunk_manager.cpp
@@ -32,296 +32,50 @@
 
 #include "mongo/s/chunk_manager.h"
 
-#include <boost/next_prior.hpp>
 #include <vector>
 
 #include "mongo/bson/simple_bsonobj_comparator.h"
-#include "mongo/client/read_preference.h"
 #include "mongo/db/matcher/extensions_callback_noop.h"
 #include "mongo/db/query/collation/collation_index_key.h"
 #include "mongo/db/query/index_bounds_builder.h"
 #include "mongo/db/query/query_planner.h"
 #include "mongo/db/query/query_planner_common.h"
-#include "mongo/s/catalog/sharding_catalog_client.h"
-#include "mongo/s/chunk_diff.h"
-#include "mongo/s/client/shard_registry.h"
 #include "mongo/s/grid.h"
 #include "mongo/util/log.h"
-#include "mongo/util/timer.h"
 
 namespace mongo {
-
-using std::map;
-using std::pair;
-using std::set;
-using std::shared_ptr;
-using std::string;
-using std::unique_ptr;
-
 namespace {
 
 // Used to generate sequence numbers to assign to each newly created ChunkManager
 AtomicUInt32 nextCMSequenceNumber(0);
 
-/**
- * This is an adapter so we can use config diffs - mongos and mongod do them slightly differently.
- *
- * The mongos adapter here tracks all shards, and stores ranges by (max, Chunk) in the map.
- */
-class CMConfigDiffTracker : public ConfigDiffTracker<shared_ptr<Chunk>> {
-public:
-    CMConfigDiffTracker(const std::string& ns,
-                        RangeMap* currMap,
-                        ChunkVersion* maxVersion,
-                        MaxChunkVersionMap* maxShardVersions,
-                        ChunkManager* manager)
-        : ConfigDiffTracker<shared_ptr<Chunk>>(ns, currMap, maxVersion, maxShardVersions),
-          _manager(manager) {}
-
-    bool isTracked(const ChunkType& chunk) const final {
-        // Mongos tracks all shards
-        return true;
-    }
-
-    bool isMinKeyIndexed() const final {
-        return false;
-    }
-
-    pair<BSONObj, shared_ptr<Chunk>> rangeFor(OperationContext* opCtx,
-                                              const ChunkType& chunk) const final {
-        return std::make_pair(chunk.getMax(), std::make_shared<Chunk>(chunk));
-    }
-
-    ShardId shardFor(OperationContext* opCtx, const ShardId& shardId) const final {
-        const auto shard =
-            uassertStatusOK(Grid::get(opCtx)->shardRegistry()->getShard(opCtx, shardId));
-        return shard->getId();
-    }
-
-private:
-    ChunkManager* const _manager;
-};
-
-bool allOfType(BSONType type, const BSONObj& o) {
-    BSONObjIterator it(o);
-    while (it.more()) {
-        if (it.next().type() != type) {
-            return false;
-        }
-    }
-    return true;
-}
-
-bool isChunkMapValid(const ChunkMap& chunkMap) {
-#define ENSURE(x)                                          \
-    do {                                                   \
-        if (!(x)) {                                        \
-            log() << "ChunkManager::_isValid failed: " #x; \
-            return false;                                  \
-        }                                                  \
-    } while (0)
-
-    if (chunkMap.empty()) {
-        return true;
-    }
-
-    // Check endpoints
-    ENSURE(allOfType(MinKey, chunkMap.begin()->second->getMin()));
-    ENSURE(allOfType(MaxKey, boost::prior(chunkMap.end())->second->getMax()));
-
-    // Make sure there are no gaps or overlaps
-    for (ChunkMap::const_iterator it = boost::next(chunkMap.begin()), end = chunkMap.end();
-         it != end;
-         ++it) {
-        ChunkMap::const_iterator last = boost::prior(it);
-
-        if (SimpleBSONObjComparator::kInstance.evaluate(it->second->getMin() !=
-                                                        last->second->getMax())) {
-            log() << last->second->toString();
-            log() << it->second->toString();
-            log() << it->second->getMin();
-            log() << last->second->getMax();
-        }
-
-        ENSURE(SimpleBSONObjComparator::kInstance.evaluate(it->second->getMin() ==
-                                                           last->second->getMax()));
+void checkAllElementsAreOfType(BSONType type, const BSONObj& o) {
+    for (const auto&& element : o) {
+        uassert(ErrorCodes::ConflictingOperationInProgress,
+                str::stream() << "Not all elements of " << o << " are of type " << typeName(type),
+                element.type() == type);
     }
-
-    return true;
-
-#undef ENSURE
 }
 
 }  // namespace
 
 ChunkManager::ChunkManager(NamespaceString nss,
-                           const OID& epoch,
-                           const ShardKeyPattern& shardKeyPattern,
+                           KeyPattern shardKeyPattern,
                            std::unique_ptr<CollatorInterface> defaultCollator,
-                           bool unique)
+                           bool unique,
+                           ChunkMap chunkMap,
+                           ChunkVersion collectionVersion)
     : _sequenceNumber(nextCMSequenceNumber.addAndFetch(1)),
       _nss(std::move(nss)),
-      _keyPattern(shardKeyPattern.getKeyPattern()),
+      _shardKeyPattern(shardKeyPattern),
       _defaultCollator(std::move(defaultCollator)),
       _unique(unique),
-      _chunkMap(SimpleBSONObjComparator::kInstance.makeBSONObjIndexedMap<std::shared_ptr<Chunk>>()),
-      _chunkRangeMap(
-          SimpleBSONObjComparator::kInstance.makeBSONObjIndexedMap<ShardAndChunkRange>()),
-      _version(0, 0, epoch) {}
+      _chunkMap(std::move(chunkMap)),
+      _chunkMapViews(_constructChunkMapViews(collectionVersion.epoch(), _chunkMap)),
+      _collectionVersion(collectionVersion) {}
 
 ChunkManager::~ChunkManager() = default;
 
-void ChunkManager::loadExistingRanges(OperationContext* opCtx, const ChunkManager* oldManager) {
-    invariant(!_version.isSet());
-
-    int tries = 3;
-
-    while (tries--) {
-        ChunkMap chunkMap =
-            SimpleBSONObjComparator::kInstance.makeBSONObjIndexedMap<std::shared_ptr<Chunk>>();
-        set<ShardId> shardIds;
-        ShardVersionMap shardVersions;
-
-        Timer t;
-
-        log() << "ChunkManager loading chunks for " << _nss
-              << " sequenceNumber: " << _sequenceNumber
-              << " based on: " << (oldManager ? oldManager->getVersion().toString() : "(empty)");
-
-        if (_load(opCtx, chunkMap, shardIds, &shardVersions, oldManager)) {
-            // TODO: Merge into diff code above, so we validate in one place
-            if (isChunkMapValid(chunkMap)) {
-                _chunkMap = std::move(chunkMap);
-                _shardVersions = std::move(shardVersions);
-                _chunkRangeMap = _constructRanges(_chunkMap);
-
-                log() << "ChunkManager load took " << t.millis() << " ms and found version "
-                      << _version;
-
-                return;
-            }
-        }
-
-        warning() << "ChunkManager load failed after " << t.millis()
-                  << " ms and will be retried up to " << tries << " more times";
-
-        sleepmillis(10 * (3 - tries));
-    }
-
-    // This will abort construction so we should never have a reference to an invalid config
-    msgasserted(13282,
-                str::stream() << "Couldn't load a valid config for " << _nss.ns()
-                              << " after 3 attempts. Please try again.");
-}
-
-bool ChunkManager::_load(OperationContext* opCtx,
-                         ChunkMap& chunkMap,
-                         set<ShardId>& shardIds,
-                         ShardVersionMap* shardVersions,
-                         const ChunkManager* oldManager) {
-    // Reset the max version, but not the epoch, when we aren't loading from the oldManager
-    _version = ChunkVersion(0, 0, _version.epoch());
-
-    // If we have a previous version of the ChunkManager to work from, use that info to reduce
-    // our config query
-    if (oldManager && oldManager->getVersion().isSet()) {
-        // Get the old max version
-        _version = oldManager->getVersion();
-
-        // Load a copy of the old versions
-        *shardVersions = oldManager->_shardVersions;
-
-        // Load a copy of the chunk map, replacing the chunk manager with our own
-        const ChunkMap& oldChunkMap = oldManager->getChunkMap();
-
-        for (const auto& oldChunkMapEntry : oldChunkMap) {
-            const auto& oldC = oldChunkMapEntry.second;
-            chunkMap.emplace(oldC->getMax(), std::make_shared<Chunk>(*oldC));
-        }
-
-        LOG(2) << "loading chunk manager for collection " << _nss
-               << " using old chunk manager w/ version " << _version.toString() << " and "
-               << oldChunkMap.size() << " chunks";
-    }
-
-    // Get the diff query required
-    const auto diffQuery = CMConfigDiffTracker::createConfigDiffQuery(_nss, _version);
-
-    // Attach a diff tracker for the versioned chunk data
-    CMConfigDiffTracker differ(_nss.ns(), &chunkMap, &_version, shardVersions, this);
-
-    // Diff tracker should *always* find at least one chunk if collection exists
-    repl::OpTime opTime;
-    std::vector<ChunkType> chunks;
-    uassertStatusOK(Grid::get(opCtx)->catalogClient(opCtx)->getChunks(
-        opCtx,
-        diffQuery.query,
-        diffQuery.sort,
-        boost::none,
-        &chunks,
-        &opTime,
-        repl::ReadConcernLevel::kMajorityReadConcern));
-
-    invariant(opTime >= _configOpTime);
-    _configOpTime = opTime;
-
-    int diffsApplied = differ.calculateConfigDiff(opCtx, chunks);
-    if (diffsApplied > 0) {
-        LOG(2) << "loaded " << diffsApplied << " chunks into new chunk manager for " << _nss
-               << " with version " << _version;
-
-        // Add all existing shards we find to the shards set
-        for (ShardVersionMap::iterator it = shardVersions->begin(); it != shardVersions->end();) {
-            auto shardStatus = Grid::get(opCtx)->shardRegistry()->getShard(opCtx, it->first);
-            if (shardStatus.isOK()) {
-                shardIds.insert(it->first);
-                ++it;
-            } else {
-                invariant(shardStatus == ErrorCodes::ShardNotFound);
-                shardVersions->erase(it++);
-            }
-        }
-
-        _configOpTime = opTime;
-
-        return true;
-    } else if (diffsApplied == 0) {
-        // No chunks were found for the ns
-        warning() << "no chunks found when reloading " << _nss << ", previous version was "
-                  << _version;
-
-        // Set all our data to empty
-        chunkMap.clear();
-        shardVersions->clear();
-
-        _version = ChunkVersion(0, 0, OID());
-        _configOpTime = opTime;
-
-        return true;
-    } else {  // diffsApplied < 0
-
-        bool allInconsistent = (differ.numValidDiffs() == 0);
-        if (allInconsistent) {
-            // All versions are different, this can be normal
-            warning() << "major change in chunk information found when reloading " << _nss
-                      << ", previous version was " << _version;
-        } else {
-            // Inconsistent load halfway through (due to yielding cursor during load)
-            // should be rare
-            warning() << "inconsistent chunks found when reloading " << _nss
-                      << ", previous version was " << _version << ", this should be rare";
-        }
-
-        // Set all our data to empty to be extra safe
-        chunkMap.clear();
-        shardVersions->clear();
-
-        _version = ChunkVersion(0, 0, OID());
-
-        return allInconsistent;
-    }
-}
-
 std::shared_ptr<Chunk> ChunkManager::findIntersectingChunk(const BSONObj& shardKey,
                                                            const BSONObj& collation) const {
     const bool hasSimpleCollation = (collation.isEmpty() && !_defaultCollator) ||
@@ -351,7 +105,7 @@ std::shared_ptr<Chunk> ChunkManager::findIntersectingChunkWithSimpleCollation(
 void ChunkManager::getShardIdsForQuery(OperationContext* opCtx,
                                        const BSONObj& query,
                                        const BSONObj& collation,
-                                       set<ShardId>* shardIds) const {
+                                       std::set<ShardId>* shardIds) const {
     auto qr = stdx::make_unique<QueryRequest>(_nss);
     qr->setFilter(query);
 
@@ -370,7 +124,7 @@ void ChunkManager::getShardIdsForQuery(OperationContext* opCtx,
     }
 
     // Fast path for targeting equalities on the shard key.
-    auto shardKeyToFind = _keyPattern.extractShardKeyFromQuery(*cq);
+    auto shardKeyToFind = _shardKeyPattern.extractShardKeyFromQuery(*cq);
     if (!shardKeyToFind.isEmpty()) {
         try {
             auto chunk = findIntersectingChunk(shardKeyToFind, collation);
@@ -387,20 +141,20 @@ void ChunkManager::getShardIdsForQuery(OperationContext* opCtx,
     //   Query { a : { $gte : 1, $lt : 2 },
     //            b : { $gte : 3, $lt : 4 } }
     //   => Bounds { a : [1, 2), b : [3, 4) }
-    IndexBounds bounds = getIndexBoundsForQuery(_keyPattern.toBSON(), *cq);
+    IndexBounds bounds = getIndexBoundsForQuery(_shardKeyPattern.toBSON(), *cq);
 
     // Transforms bounds for each shard key field into full shard key ranges
     // for example :
     //   Key { a : 1, b : 1 }
     //   Bounds { a : [1, 2), b : [3, 4) }
     //   => Ranges { a : 1, b : 3 } => { a : 2, b : 4 }
-    BoundList ranges = _keyPattern.flattenBounds(bounds);
+    BoundList ranges = _shardKeyPattern.flattenBounds(bounds);
 
     for (BoundList::const_iterator it = ranges.begin(); it != ranges.end(); ++it) {
         getShardIdsForRange(it->first /*min*/, it->second /*max*/, shardIds);
 
         // once we know we need to visit all shards no need to keep looping
-        if (shardIds->size() == _shardVersions.size()) {
+        if (shardIds->size() == _chunkMapViews.shardVersions.size()) {
             break;
         }
     }
@@ -409,38 +163,38 @@ void ChunkManager::getShardIdsForQuery(OperationContext* opCtx,
     // For now, we satisfy that assumption by adding a shard with no matches rather than returning
     // an empty set of shards.
     if (shardIds->empty()) {
-        shardIds->insert(_chunkRangeMap.begin()->second.getShardId());
+        shardIds->insert(_chunkMapViews.chunkRangeMap.begin()->second.shardId);
     }
 }
 
 void ChunkManager::getShardIdsForRange(const BSONObj& min,
                                        const BSONObj& max,
                                        std::set<ShardId>* shardIds) const {
-    auto it = _chunkRangeMap.upper_bound(min);
-    auto end = _chunkRangeMap.upper_bound(max);
+    auto it = _chunkMapViews.chunkRangeMap.upper_bound(min);
+    auto end = _chunkMapViews.chunkRangeMap.upper_bound(max);
 
     // The chunk range map must always cover the entire key space
-    invariant(it != _chunkRangeMap.end());
+    invariant(it != _chunkMapViews.chunkRangeMap.end());
 
     // We need to include the last chunk
-    if (end != _chunkRangeMap.cend()) {
+    if (end != _chunkMapViews.chunkRangeMap.cend()) {
         ++end;
     }
 
     for (; it != end; ++it) {
-        shardIds->insert(it->second.getShardId());
+        shardIds->insert(it->second.shardId);
 
         // No need to iterate through the rest of the ranges, because we already know we need to use
         // all shards.
-        if (shardIds->size() == _shardVersions.size()) {
+        if (shardIds->size() == _chunkMapViews.shardVersions.size()) {
             break;
         }
     }
 }
 
-void ChunkManager::getAllShardIds(set<ShardId>* all) const {
-    std::transform(_shardVersions.begin(),
-                   _shardVersions.end(),
+void ChunkManager::getAllShardIds(std::set<ShardId>* all) const {
+    std::transform(_chunkMapViews.shardVersions.begin(),
+                   _chunkMapViews.shardVersions.end(),
                    std::inserter(*all, all->begin()),
                    [](const ShardVersionMap::value_type& pair) { return pair.first; });
 }
@@ -457,7 +211,7 @@ IndexBounds ChunkManager::getIndexBoundsForQuery(const BSONObj& key,
     }
 
     // Consider shard key as an index
-    string accessMethod = IndexNames::findPluginName(key);
+    std::string accessMethod = IndexNames::findPluginName(key);
     dassert(accessMethod == IndexNames::BTREE || accessMethod == IndexNames::HASHED);
 
     // Use query framework to generate index bounds
@@ -564,19 +318,19 @@ bool ChunkManager::compatibleWith(const ChunkManager& other, const ShardId& shar
 }
 
 ChunkVersion ChunkManager::getVersion(const ShardId& shardName) const {
-    auto it = _shardVersions.find(shardName);
-    if (it == _shardVersions.end()) {
+    auto it = _chunkMapViews.shardVersions.find(shardName);
+    if (it == _chunkMapViews.shardVersions.end()) {
         // Shards without explicitly tracked shard versions (meaning they have no chunks) always
         // have a version of (0, 0, epoch)
-        return ChunkVersion(0, 0, _version.epoch());
+        return ChunkVersion(0, 0, _collectionVersion.epoch());
     }
 
     return it->second;
 }
 
-string ChunkManager::toString() const {
+std::string ChunkManager::toString() const {
     StringBuilder sb;
-    sb << "ChunkManager: " << _nss.ns() << " key:" << _keyPattern.toString() << '\n';
+    sb << "ChunkManager: " << _nss.ns() << " key:" << _shardKeyPattern.toString() << '\n';
 
     for (const auto& entry : _chunkMap) {
         sb << "\t" << entry.second->toString() << '\n';
@@ -585,47 +339,82 @@ string ChunkManager::toString() const {
     return sb.str();
 }
 
-ChunkManager::ChunkRangeMap ChunkManager::_constructRanges(const ChunkMap& chunkMap) {
+ChunkManager::ChunkMapViews ChunkManager::_constructChunkMapViews(const OID& epoch,
+                                                                  const ChunkMap& chunkMap) {
+    invariant(!chunkMap.empty());
+
     ChunkRangeMap chunkRangeMap =
         SimpleBSONObjComparator::kInstance.makeBSONObjIndexedMap<ShardAndChunkRange>();
 
-    if (chunkMap.empty()) {
-        return chunkRangeMap;
-    }
+    ShardVersionMap shardVersions;
 
     ChunkMap::const_iterator current = chunkMap.cbegin();
 
     while (current != chunkMap.cend()) {
-        const auto rangeFirst = current;
+        const auto& firstChunkInRange = current->second;
+
+        // Tracks the max shard version for the shard on which the current range will reside
+        auto shardVersionIt = shardVersions.find(firstChunkInRange->getShardId());
+        if (shardVersionIt == shardVersions.end()) {
+            shardVersionIt =
+                shardVersions.emplace(firstChunkInRange->getShardId(), ChunkVersion(0, 0, epoch))
+                    .first;
+        }
+
+        auto& maxShardVersion = shardVersionIt->second;
+
         current = std::find_if(
-            current, chunkMap.cend(), [&rangeFirst](const ChunkMap::value_type& chunkMapEntry) {
-                return chunkMapEntry.second->getShardId() != rangeFirst->second->getShardId();
+            current,
+            chunkMap.cend(),
+            [&firstChunkInRange, &maxShardVersion](const ChunkMap::value_type& chunkMapEntry) {
+                const auto& currentChunk = chunkMapEntry.second;
+
+                if (currentChunk->getShardId() != firstChunkInRange->getShardId())
+                    return true;
+
+                if (currentChunk->getLastmod() > maxShardVersion)
+                    maxShardVersion = currentChunk->getLastmod();
+
+                return false;
             });
+
         const auto rangeLast = std::prev(current);
 
-        const BSONObj rangeMin = rangeFirst->second->getMin();
+        const BSONObj rangeMin = firstChunkInRange->getMin();
         const BSONObj rangeMax = rangeLast->second->getMax();
 
-        auto insertResult = chunkRangeMap.insert(std::make_pair(
-            rangeMax, ShardAndChunkRange(rangeMin, rangeMax, rangeFirst->second->getShardId())));
-        invariant(insertResult.second);
-        if (insertResult.first != chunkRangeMap.begin()) {
+        const auto insertResult = chunkRangeMap.insert(std::make_pair(
+            rangeMax, ShardAndChunkRange{{rangeMin, rangeMax}, firstChunkInRange->getShardId()}));
+        uassert(ErrorCodes::ConflictingOperationInProgress,
+                str::stream() << "Metadata contains two chunks with the same max value "
+                              << rangeMax,
+                insertResult.second);
+
+        const auto& insertIterator = insertResult.first;
+
+        if (insertIterator != chunkRangeMap.begin()) {
             // Make sure there are no gaps in the ranges
-            insertResult.first--;
-            invariant(
-                SimpleBSONObjComparator::kInstance.evaluate(insertResult.first->first == rangeMin));
+            uassert(ErrorCodes::ConflictingOperationInProgress,
+                    str::stream() << "Gap or an overlap between ranges "
+                                  << insertIterator->second.range.toString()
+                                  << " and "
+                                  << std::prev(insertIterator)->second.range.toString(),
+                    SimpleBSONObjComparator::kInstance.evaluate(std::prev(insertIterator)->first ==
+                                                                rangeMin));
         }
+
+        // If a shard has chunks it must have a shard version, otherwise we have an invalid chunk
+        // somewhere, which should have been caught at chunk load time
+        invariant(maxShardVersion.isSet());
     }
 
     invariant(!chunkRangeMap.empty());
-    invariant(allOfType(MinKey, chunkRangeMap.begin()->second.getMin()));
-    invariant(allOfType(MaxKey, chunkRangeMap.rbegin()->first));
+    invariant(!shardVersions.empty());
 
-    return chunkRangeMap;
-}
+    checkAllElementsAreOfType(MinKey, chunkRangeMap.begin()->second.min());
+    checkAllElementsAreOfType(MaxKey, chunkRangeMap.rbegin()->first);
 
-repl::OpTime ChunkManager::getConfigOpTime() const {
-    return _configOpTime;
+    return {std::move(chunkRangeMap), std::move(shardVersions)};
 }
 
 }  // namespace mongo
diff --git a/src/mongo/s/chunk_manager.h b/src/mongo/s/chunk_manager.h
index 365d4d5df62..f1edeefc668 100644
--- a/src/mongo/s/chunk_manager.h
+++ b/src/mongo/s/chunk_manager.h
@@ -35,8 +35,6 @@
 #include "mongo/base/disallow_copying.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/query/collation/collator_interface.h"
-#include "mongo/db/repl/optime.h"
-#include "mongo/s/catalog/type_chunk.h"
 #include "mongo/s/chunk.h"
 #include "mongo/s/chunk_version.h"
 #include "mongo/s/client/shard.h"
@@ -60,10 +58,11 @@ class ChunkManager {
 
 public:
     ChunkManager(NamespaceString nss,
-                 const OID& epoch,
-                 const ShardKeyPattern& shardKeyPattern,
+                 KeyPattern shardKeyPattern,
                  std::unique_ptr<CollatorInterface> defaultCollator,
-                 bool unique);
+                 bool unique,
+                 ChunkMap chunkMap,
+                 ChunkVersion collectionVersion);
 
     ~ChunkManager();
 
@@ -79,7 +78,7 @@ public:
     }
 
     const ShardKeyPattern& getShardKeyPattern() const {
-        return _keyPattern;
+        return _shardKeyPattern;
     }
 
     const CollatorInterface* getDefaultCollator() const {
@@ -91,10 +90,12 @@ public:
     }
 
     ChunkVersion getVersion() const {
-        return _version;
+        return _collectionVersion;
     }
 
-    const ChunkMap& getChunkMap() const {
+    ChunkVersion getVersion(const ShardId& shardId) const;
+
+    const ChunkMap& chunkMap() const {
         return _chunkMap;
     }
 
@@ -102,12 +103,9 @@ public:
         return _chunkMap.size();
     }
 
-    // Loads existing ranges based on info in chunk manager
-    void loadExistingRanges(OperationContext* opCtx, const ChunkManager* oldManager);
-
-    //
-    // Methods to use once loaded / created
-    //
+    const ShardVersionMap& shardVersions() const {
+        return _chunkMapViews.shardVersions;
+    }
 
     /**
      * Given a shard key (or a prefix) that has been extracted from a document, returns the chunk
@@ -177,57 +175,46 @@ public:
 
     std::string toString() const;
 
-    ChunkVersion getVersion(const ShardId& shardName) const;
-
-    /**
-     * Returns the opTime of config server the last time chunks were loaded.
-     */
-    repl::OpTime getConfigOpTime() const;
-
 private:
+    friend class CollectionRoutingDataLoader;
+
     /**
      * Represents a range of chunk keys [getMin(), getMax()) and the id of the shard on which they
      * reside according to the metadata.
      */
-    class ShardAndChunkRange {
-    public:
-        ShardAndChunkRange(const BSONObj& min, const BSONObj& max, ShardId inShardId)
-            : _range(min, max), _shardId(std::move(inShardId)) {}
-
-        const BSONObj& getMin() const {
-            return _range.getMin();
-        }
-
-        const BSONObj& getMax() const {
-            return _range.getMax();
+    struct ShardAndChunkRange {
+        const BSONObj& min() const {
+            return range.getMin();
         }
 
-        const ShardId& getShardId() const {
-            return _shardId;
+        const BSONObj& max() const {
+            return range.getMax();
         }
 
-    private:
-        ChunkRange _range;
-        ShardId _shardId;
+        ChunkRange range;
+        ShardId shardId;
     };
 
     using ChunkRangeMap = BSONObjIndexedMap<ShardAndChunkRange>;
 
     /**
-     * If load was successful, returns true and it is guaranteed that the _chunkMap and
-     * _chunkRangeMap are consistent with each other. If false is returned, it is not safe to use
-     * the chunk manager anymore.
+     * Contains different transformations of the chunk map for efficient querying
      */
-    bool _load(OperationContext* opCtx,
-               ChunkMap& chunks,
-               std::set<ShardId>& shardIds,
-               ShardVersionMap* shardVersions,
-               const ChunkManager* oldManager);
+    struct ChunkMapViews {
+        // Transformation of the chunk map containing what range of keys reside on which shard. The
+        // index is the max key of the respective range and the union of all ranges in a such
+        // constructed map must cover the complete space from [MinKey, MaxKey).
+        const ChunkRangeMap chunkRangeMap;
+
+        // Map from shard id to the maximum chunk version for that shard. If a shard contains no
+        // chunks, it won't be present in this map.
+        const ShardVersionMap shardVersions;
+    };
 
     /**
-     * Merges consecutive chunks, which reside on the same shard into a single range.
+     * Does a single pass over the chunkMap and constructs the ChunkMapViews object.
      */
-    static ChunkRangeMap _constructRanges(const ChunkMap& chunkMap);
+    static ChunkMapViews _constructChunkMapViews(const OID& epoch, const ChunkMap& chunkMap);
 
     // The shard versioning mechanism hinges on keeping track of the number of times we reload
     // ChunkManagers.
@@ -237,7 +224,7 @@ private:
     const NamespaceString _nss;
 
     // The key pattern used to shard the collection
-    const ShardKeyPattern _keyPattern;
+    const ShardKeyPattern _shardKeyPattern;
 
     // Default collation to use for routing data queries for this collection
     const std::unique_ptr<CollatorInterface> _defaultCollator;
@@ -247,23 +234,15 @@ private:
 
     // Map from the max for each chunk to an entry describing the chunk. The union of all chunks'
     // ranges must cover the complete space from [MinKey, MaxKey).
-    ChunkMap _chunkMap;
-
-    // Transformation of the chunk map containing what range of keys reside on which shard. The
-    // index is the max key of the respective range and the union of all ranges in a such
-    // constructed map must cover the complete space from [MinKey, MaxKey).
-    ChunkRangeMap _chunkRangeMap;
+    const ChunkMap _chunkMap;
 
-    // Max known version per shard
-    ShardVersionMap _shardVersions;
+    // Different transformations of the chunk map for efficient querying
+    const ChunkMapViews _chunkMapViews;
 
     // Max version across all chunks
-    ChunkVersion _version;
+    const ChunkVersion _collectionVersion;
 
-    // OpTime of config server the last time chunks were loaded.
-    repl::OpTime _configOpTime;
-
-    // Auto-split throttling state
+    // Auto-split throttling state (state mutable by write commands)
     struct AutoSplitThrottle {
     public:
         AutoSplitThrottle() : _splitTickets(maxParallelSplits) {}
@@ -280,8 +259,6 @@ private:
                                                       ChunkManager*,
                                                       Chunk*,
                                                       long);
-
-    friend class TestableChunkManager;
 };
 
 }  // namespace mongo
diff --git a/src/mongo/s/chunk_manager_test.cpp b/src/mongo/s/chunk_manager_query_test.cpp
index 08f8357b776..013c4618d6c 100644
--- a/src/mongo/s/chunk_manager_test.cpp
+++ b/src/mongo/s/chunk_manager_query_test.cpp
@@ -32,143 +32,13 @@
 
 #include <set>
 
-#include "mongo/client/remote_command_targeter_mock.h"
-#include "mongo/db/client.h"
 #include "mongo/db/query/collation/collator_interface_mock.h"
-#include "mongo/s/catalog/sharding_catalog_test_fixture.h"
-#include "mongo/s/catalog/type_chunk.h"
-#include "mongo/s/catalog/type_collection.h"
-#include "mongo/s/catalog/type_shard.h"
 #include "mongo/s/chunk_manager.h"
-#include "mongo/stdx/memory.h"
-#include "mongo/util/scopeguard.h"
+#include "mongo/s/chunk_manager_test_fixture.h"
 
 namespace mongo {
 namespace {
 
-using executor::RemoteCommandResponse;
-using executor::RemoteCommandRequest;
-
-const NamespaceString kNss("TestDB", "TestColl");
-
-class ChunkManagerTestFixture : public ShardingCatalogTestFixture {
-protected:
-    void setUp() override {
-        ShardingCatalogTestFixture::setUp();
-        setRemote(HostAndPort("FakeRemoteClient:34567"));
-        configTargeter()->setFindHostReturnValue(HostAndPort{CONFIG_HOST_PORT});
-    }
-
-    /**
-     * Returns a chunk manager with chunks at the specified split points. Each individual chunk is
-     * placed on a separate shard with id ranging from "0" to the number of chunks.
-     */
-    std::unique_ptr<ChunkManager> makeChunkManager(
-        const ShardKeyPattern& shardKeyPattern,
-        std::unique_ptr<CollatorInterface> defaultCollator,
-        bool unique,
-        const std::vector<BSONObj>& splitPoints) {
-        ChunkVersion version(1, 0, OID::gen());
-
-        std::vector<BSONObj> shards;
-        std::vector<BSONObj> initialChunks;
-
-        auto splitPointsIncludingEnds(splitPoints);
-        splitPointsIncludingEnds.insert(splitPointsIncludingEnds.begin(),
-                                        shardKeyPattern.getKeyPattern().globalMin());
-        splitPointsIncludingEnds.push_back(shardKeyPattern.getKeyPattern().globalMax());
-
-        for (size_t i = 1; i < splitPointsIncludingEnds.size(); ++i) {
-            ShardType shard;
-            shard.setName(str::stream() << (i - 1));
-            shard.setHost(str::stream() << "Host" << (i - 1) << ":12345");
-
-            shards.push_back(shard.toBSON());
-
-            ChunkType chunk;
-            chunk.setNS(kNss.ns());
-            chunk.setMin(shardKeyPattern.getKeyPattern().extendRangeBound(
-                splitPointsIncludingEnds[i - 1], false));
-            chunk.setMax(shardKeyPattern.getKeyPattern().extendRangeBound(
-                splitPointsIncludingEnds[i], false));
-            chunk.setShard(shard.getName());
-            chunk.setVersion(version);
-
-            initialChunks.push_back(chunk.toConfigBSON());
-
-            version.incMajor();
-        }
-
-        // Load the initial manager
-        auto manager = stdx::make_unique<ChunkManager>(
-            kNss, version.epoch(), shardKeyPattern, std::move(defaultCollator), unique);
-
-        auto future = launchAsync([&manager] {
-            ON_BLOCK_EXIT([&] { Client::destroy(); });
-            Client::initThread("Test");
-            auto opCtx = cc().makeOperationContext();
-            manager->loadExistingRanges(opCtx.get(), nullptr);
-        });
-
-        expectFindOnConfigSendBSONObjVector(initialChunks);
-        expectFindOnConfigSendBSONObjVector(shards);
-
-        future.timed_get(kFutureTimeout);
-
-        return manager;
-    }
-};
-
-using ChunkManagerLoadTest = ChunkManagerTestFixture;
-
-TEST_F(ChunkManagerLoadTest, IncrementalLoadAfterSplit) {
-    const ShardKeyPattern shardKeyPattern(BSON("_id" << 1));
-
-    auto initialManager(makeChunkManager(shardKeyPattern, nullptr, true, {}));
-
-    ChunkVersion version = initialManager->getVersion();
-
-    CollectionType collType;
-    collType.setNs(kNss);
-    collType.setEpoch(version.epoch());
-    collType.setUpdatedAt(jsTime());
-    collType.setKeyPattern(shardKeyPattern.toBSON());
-    collType.setUnique(false);
-
-    ChunkManager manager(kNss, version.epoch(), shardKeyPattern, nullptr, true);
-
-    auto future =
-        launchAsync([&] { manager.loadExistingRanges(operationContext(), initialManager.get()); });
-
-    // Return set of chunks, which represent a split
-    expectFindOnConfigSendBSONObjVector([&]() {
-        version.incMajor();
-
-        ChunkType chunk1;
-        chunk1.setNS(kNss.ns());
-        chunk1.setMin(shardKeyPattern.getKeyPattern().globalMin());
-        chunk1.setMax(BSON("_id" << 0));
-        chunk1.setShard({"0"});
-        chunk1.setVersion(version);
-
-        version.incMinor();
-
-        ChunkType chunk2;
-        chunk2.setNS(kNss.ns());
-        chunk2.setMin(BSON("_id" << 0));
-        chunk2.setMax(shardKeyPattern.getKeyPattern().globalMax());
-        chunk2.setShard({"0"});
-        chunk2.setVersion(version);
-
-        return std::vector<BSONObj>{chunk1.toConfigBSON(), chunk2.toConfigBSON()};
-    }());
-
-    future.timed_get(kFutureTimeout);
-}
-
-/**
- * Fixture to be used as a shortcut for tests which exercise the getShardIdsForQuery routing logic
- */
 class ChunkManagerQueryTest : public ChunkManagerTestFixture {
 protected:
     void runQueryTest(const BSONObj& shardKey,
diff --git a/src/mongo/s/chunk_manager_refresh_test.cpp b/src/mongo/s/chunk_manager_refresh_test.cpp
new file mode 100644
index 00000000000..504893acf3c
--- /dev/null
+++ b/src/mongo/s/chunk_manager_refresh_test.cpp
@@ -0,0 +1,194 @@
+/**
+ *    Copyright (C) 2017 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kDefault
+
+#include "mongo/platform/basic.h"
+
+#include <set>
+
+#include "mongo/s/catalog/type_chunk.h"
+#include "mongo/s/catalog/type_collection.h"
+#include "mongo/s/catalog/type_shard.h"
+#include "mongo/s/catalog_cache.h"
+#include "mongo/s/chunk_manager_test_fixture.h"
+#include "mongo/util/scopeguard.h"
+
+namespace mongo {
+namespace {
+
+using ChunkManagerLoadTest = ChunkManagerTestFixture;
+
+TEST_F(ChunkManagerLoadTest, IncrementalLoadAfterSplit) {
+    const ShardKeyPattern shardKeyPattern(BSON("_id" << 1));
+
+    auto initialRoutingInfo(makeChunkManager(shardKeyPattern, nullptr, true, {}));
+    ASSERT_EQ(1, initialRoutingInfo->numChunks());
+
+    auto future = launchAsync([&] {
+        auto client = serviceContext()->makeClient("Test");
+        auto opCtx = client->makeOperationContext();
+        return CatalogCache::refreshCollectionRoutingInfo(opCtx.get(), kNss, initialRoutingInfo);
+    });
+
+    ChunkVersion version = initialRoutingInfo->getVersion();
+
+    expectFindOnConfigSendBSONObjVector([&]() {
+        CollectionType collType;
+        collType.setNs(kNss);
+        collType.setEpoch(version.epoch());
+        collType.setKeyPattern(shardKeyPattern.toBSON());
+        collType.setUnique(false);
+
+        return std::vector<BSONObj>{collType.toBSON()};
+    }());
+
+    // Return set of chunks, which represent a split
+    expectFindOnConfigSendBSONObjVector([&]() {
+        version.incMajor();
+        ChunkType chunk1(
+            kNss, {shardKeyPattern.getKeyPattern().globalMin(), BSON("_id" << 0)}, version, {"0"});
+
+        version.incMinor();
+        ChunkType chunk2(
+            kNss, {BSON("_id" << 0), shardKeyPattern.getKeyPattern().globalMax()}, version, {"0"});
+
+        return std::vector<BSONObj>{chunk1.toConfigBSON(), chunk2.toConfigBSON()};
+    }());
+
+    auto newRoutingInfo(future.timed_get(kFutureTimeout));
+    ASSERT_EQ(2, newRoutingInfo->numChunks());
+    ASSERT_EQ(version, newRoutingInfo->getVersion());
+    ASSERT_EQ(version, newRoutingInfo->getVersion({"0"}));
+    ASSERT_EQ(ChunkVersion(0, 0, version.epoch()), newRoutingInfo->getVersion({"1"}));
+}
+
+TEST_F(ChunkManagerLoadTest, IncrementalLoadAfterMove) {
+    const ShardKeyPattern shardKeyPattern(BSON("_id" << 1));
+
+    auto initialRoutingInfo(makeChunkManager(shardKeyPattern, nullptr, true, {BSON("_id" << 0)}));
+    ASSERT_EQ(2, initialRoutingInfo->numChunks());
+
+    auto future = launchAsync([&] {
+        auto client = serviceContext()->makeClient("Test");
+        auto opCtx = client->makeOperationContext();
+        return CatalogCache::refreshCollectionRoutingInfo(opCtx.get(), kNss, initialRoutingInfo);
+    });
+
+    ChunkVersion version = initialRoutingInfo->getVersion();
+
+    expectFindOnConfigSendBSONObjVector([&]() {
+        CollectionType collType;
+        collType.setNs(kNss);
+        collType.setEpoch(version.epoch());
+        collType.setKeyPattern(shardKeyPattern.toBSON());
+        collType.setUnique(false);
+
+        return std::vector<BSONObj>{collType.toBSON()};
+    }());
+
+    ChunkVersion expectedDestShardVersion;
+
+    // Return set of chunks, which represent a move
+    expectFindOnConfigSendBSONObjVector([&]() {
+        version.incMajor();
+        expectedDestShardVersion = version;
+        ChunkType chunk1(
+            kNss, {shardKeyPattern.getKeyPattern().globalMin(), BSON("_id" << 0)}, version, {"1"});
+
+        version.incMinor();
+        ChunkType chunk2(
+            kNss, {BSON("_id" << 0), shardKeyPattern.getKeyPattern().globalMax()}, version, {"0"});
+
+        return std::vector<BSONObj>{chunk1.toConfigBSON(), chunk2.toConfigBSON()};
+    }());
+
+    auto newRoutingInfo(future.timed_get(kFutureTimeout));
+    ASSERT_EQ(2, newRoutingInfo->numChunks());
+    ASSERT_EQ(version, newRoutingInfo->getVersion());
+    ASSERT_EQ(version, newRoutingInfo->getVersion({"0"}));
+    ASSERT_EQ(expectedDestShardVersion, newRoutingInfo->getVersion({"1"}));
+}
+
+TEST_F(ChunkManagerLoadTest, IncrementalLoadAfterMoveLastChunk) {
+    const ShardKeyPattern shardKeyPattern(BSON("_id" << 1));
+
+    auto initialRoutingInfo(makeChunkManager(shardKeyPattern, nullptr, true, {}));
+    ASSERT_EQ(1, initialRoutingInfo->numChunks());
+
+    auto future = launchAsync([&] {
+        auto client = serviceContext()->makeClient("Test");
+        auto opCtx = client->makeOperationContext();
+        return CatalogCache::refreshCollectionRoutingInfo(opCtx.get(), kNss, initialRoutingInfo);
+    });
+
+    ChunkVersion version = initialRoutingInfo->getVersion();
+
+    expectFindOnConfigSendBSONObjVector([&]() {
+        CollectionType collType;
+        collType.setNs(kNss);
+        collType.setEpoch(version.epoch());
+        collType.setKeyPattern(shardKeyPattern.toBSON());
+        collType.setUnique(false);
+
+        return std::vector<BSONObj>{collType.toBSON()};
+    }());
+
+    // Return set of chunks, which represent a move
+    expectFindOnConfigSendBSONObjVector([&]() {
+        version.incMajor();
+        ChunkType chunk1(kNss,
+                         {shardKeyPattern.getKeyPattern().globalMin(),
+                          shardKeyPattern.getKeyPattern().globalMax()},
+                         version,
+                         {"1"});
+
+        return std::vector<BSONObj>{chunk1.toConfigBSON()};
+    }());
+
+    expectFindOnConfigSendBSONObjVector([&]() {
+        ShardType shard1;
+        shard1.setName("0");
+        shard1.setHost(str::stream() << "Host0:12345");
+
+        ShardType shard2;
+        shard2.setName("1");
+        shard2.setHost(str::stream() << "Host1:12345");
+
+        return std::vector<BSONObj>{shard1.toBSON(), shard2.toBSON()};
+    }());
+
+    auto newRoutingInfo(future.timed_get(kFutureTimeout));
+    ASSERT_EQ(1, newRoutingInfo->numChunks());
+    ASSERT_EQ(version, newRoutingInfo->getVersion());
+    ASSERT_EQ(ChunkVersion(0, 0, version.epoch()), newRoutingInfo->getVersion({"0"}));
+    ASSERT_EQ(version, newRoutingInfo->getVersion({"1"}));
+}
+
+}  // namespace
+}  // namespace mongo
diff --git a/src/mongo/s/chunk_manager_test_fixture.cpp b/src/mongo/s/chunk_manager_test_fixture.cpp
new file mode 100644
index 00000000000..ada08673d70
--- /dev/null
+++ b/src/mongo/s/chunk_manager_test_fixture.cpp
@@ -0,0 +1,122 @@
+/**
+ *    Copyright (C) 2017 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kDefault
+
+#include "mongo/platform/basic.h"
+
+#include <set>
+#include <vector>
+
+#include "mongo/s/chunk_manager_test_fixture.h"
+
+#include "mongo/client/remote_command_targeter_mock.h"
+#include "mongo/db/client.h"
+#include "mongo/db/query/collation/collator_factory_mock.h"
+#include "mongo/s/catalog/type_chunk.h"
+#include "mongo/s/catalog/type_collection.h"
+#include "mongo/s/catalog/type_shard.h"
+#include "mongo/s/catalog_cache.h"
+#include "mongo/stdx/memory.h"
+#include "mongo/util/scopeguard.h"
+
+namespace mongo {
+
+const NamespaceString ChunkManagerTestFixture::kNss("TestDB", "TestColl");
+
+void ChunkManagerTestFixture::setUp() {
+    ShardingCatalogTestFixture::setUp();
+    setRemote(HostAndPort("FakeRemoteClient:34567"));
+    configTargeter()->setFindHostReturnValue(HostAndPort{CONFIG_HOST_PORT});
+
+    CollatorFactoryInterface::set(serviceContext(), stdx::make_unique<CollatorFactoryMock>());
+}
+
+std::shared_ptr<ChunkManager> ChunkManagerTestFixture::makeChunkManager(
+    const ShardKeyPattern& shardKeyPattern,
+    std::unique_ptr<CollatorInterface> defaultCollator,
+    bool unique,
+    const std::vector<BSONObj>& splitPoints) {
+    ChunkVersion version(1, 0, OID::gen());
+
+    const BSONObj collectionBSON = [&]() {
+        CollectionType coll;
+        coll.setNs(kNss);
+        coll.setEpoch(version.epoch());
+        coll.setKeyPattern(shardKeyPattern.getKeyPattern());
+        coll.setUnique(unique);
+
+        if (defaultCollator) {
+            coll.setDefaultCollation(defaultCollator->getSpec().toBSON());
+        }
+
+        return coll.toBSON();
+    }();
+
+    std::vector<BSONObj> shards;
+    std::vector<BSONObj> initialChunks;
+
+    auto splitPointsIncludingEnds(splitPoints);
+    splitPointsIncludingEnds.insert(splitPointsIncludingEnds.begin(),
+                                    shardKeyPattern.getKeyPattern().globalMin());
+    splitPointsIncludingEnds.push_back(shardKeyPattern.getKeyPattern().globalMax());
+
+    for (size_t i = 1; i < splitPointsIncludingEnds.size(); ++i) {
+        ShardType shard;
+        shard.setName(str::stream() << (i - 1));
+        shard.setHost(str::stream() << "Host" << (i - 1) << ":12345");
+
+        shards.push_back(shard.toBSON());
+
+        ChunkType chunk(
+            kNss,
+            {shardKeyPattern.getKeyPattern().extendRangeBound(splitPointsIncludingEnds[i - 1],
+                                                              false),
+             shardKeyPattern.getKeyPattern().extendRangeBound(splitPointsIncludingEnds[i], false)},
+            version,
+            shard.getName());
+
+        initialChunks.push_back(chunk.toConfigBSON());
+
+        version.incMajor();
+    }
+
+    auto future = launchAsync([&] {
+        auto client = serviceContext()->makeClient("Test");
+        auto opCtx = client->makeOperationContext();
+        return CatalogCache::refreshCollectionRoutingInfo(opCtx.get(), kNss, nullptr);
+    });
+
+    expectFindOnConfigSendBSONObjVector({collectionBSON});
+    expectFindOnConfigSendBSONObjVector(initialChunks);
+    expectFindOnConfigSendBSONObjVector(shards);
+
+    return future.timed_get(kFutureTimeout);
+}
+
+}  // namespace mongo
diff --git a/src/mongo/s/chunk_manager_test_fixture.h b/src/mongo/s/chunk_manager_test_fixture.h
new file mode 100644
index 00000000000..aaa059dd49d
--- /dev/null
+++ b/src/mongo/s/chunk_manager_test_fixture.h
@@ -0,0 +1,62 @@
+/**
+ *    Copyright (C) 2017 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include <vector>
+
+#include "mongo/db/namespace_string.h"
+#include "mongo/s/catalog/sharding_catalog_test_fixture.h"
+#include "mongo/stdx/memory.h"
+
+namespace mongo {
+
+class BSONObj;
+class ChunkManager;
+class CollatorInterface;
+class ShardKeyPattern;
+
+class ChunkManagerTestFixture : public ShardingCatalogTestFixture {
+protected:
+    void setUp() override;
+
+    /**
+     * Returns a chunk manager with chunks at the specified split points. Each individual chunk is
+     * placed on a separate shard with shard id being a single number ranging from "0" to the number
+     * of chunks.
+     */
+    std::shared_ptr<ChunkManager> makeChunkManager(
+        const ShardKeyPattern& shardKeyPattern,
+        std::unique_ptr<CollatorInterface> defaultCollator,
+        bool unique,
+        const std::vector<BSONObj>& splitPoints);
+
+    static const NamespaceString kNss;
+};
+
+}  // namespace mongo
diff --git a/src/mongo/s/client/parallel.cpp b/src/mongo/s/client/parallel.cpp
index 89d2836dc92..9e2aaaaf4e5 100644
--- a/src/mongo/s/client/parallel.cpp
+++ b/src/mongo/s/client/parallel.cpp
@@ -39,11 +39,9 @@
 #include "mongo/db/bson/dotted_path_support.h"
 #include "mongo/db/query/query_request.h"
 #include "mongo/s/catalog_cache.h"
-#include "mongo/s/chunk_manager.h"
 #include "mongo/s/client/shard_connection.h"
 #include "mongo/s/client/shard_registry.h"
 #include "mongo/s/grid.h"
-#include "mongo/s/sharding_raii.h"
 #include "mongo/s/stale_exception.h"
 #include "mongo/util/log.h"
 #include "mongo/util/net/socket_exception.h"
@@ -321,42 +319,20 @@ void ParallelSortClusteredCursor::fullInit(OperationContext* opCtx) {
     finishInit(opCtx);
 }
 
-void ParallelSortClusteredCursor::_markStaleNS(OperationContext* opCtx,
-                                               const NamespaceString& staleNS,
-                                               const StaleConfigException& e,
-                                               bool& forceReload) {
-    if (e.requiresFullReload()) {
-        Grid::get(opCtx)->catalogCache()->invalidate(staleNS.db());
-    }
-
-    if (_staleNSMap.find(staleNS.ns()) == _staleNSMap.end())
+void ParallelSortClusteredCursor::_markStaleNS(const NamespaceString& staleNS,
+                                               const StaleConfigException& e) {
+    if (_staleNSMap.find(staleNS.ns()) == _staleNSMap.end()) {
         _staleNSMap[staleNS.ns()] = 1;
+    }
 
-    int tries = ++_staleNSMap[staleNS.ns()];
+    const int tries = ++_staleNSMap[staleNS.ns()];
 
     if (tries >= 5) {
         throw SendStaleConfigException(staleNS.ns(),
-                                       str::stream() << "too many retries of stale version info",
+                                       "too many retries of stale version info",
                                        e.getVersionReceived(),
                                        e.getVersionWanted());
     }
-
-    forceReload = tries > 2;
-}
-
-void ParallelSortClusteredCursor::_handleStaleNS(OperationContext* opCtx,
-                                                 const NamespaceString& staleNS,
-                                                 bool forceReload) {
-    auto scopedCMStatus = ScopedChunkManager::get(opCtx, staleNS);
-    if (!scopedCMStatus.isOK()) {
-        log() << "cannot reload database info for stale namespace " << staleNS.ns();
-        return;
-    }
-
-    const auto& scopedCM = scopedCMStatus.getValue();
-
-    // Reload chunk manager, potentially forcing the namespace
-    scopedCM.db()->getChunkManagerIfExists(opCtx, staleNS.ns(), true, forceReload);
 }
 
 void ParallelSortClusteredCursor::setupVersionAndHandleSlaveOk(
@@ -459,12 +435,12 @@ void ParallelSortClusteredCursor::startInit(OperationContext* opCtx) {
     shared_ptr<Shard> primary;
 
     {
-        auto scopedCMStatus = ScopedChunkManager::get(opCtx, nss);
-        if (scopedCMStatus != ErrorCodes::NamespaceNotFound) {
-            uassertStatusOK(scopedCMStatus.getStatus());
-            const auto& scopedCM = scopedCMStatus.getValue();
-            manager = scopedCM.cm();
-            primary = scopedCM.primary();
+        auto routingInfoStatus =
+            Grid::get(opCtx)->catalogCache()->getCollectionRoutingInfo(opCtx, nss);
+        if (routingInfoStatus != ErrorCodes::NamespaceNotFound) {
+            auto routingInfo = uassertStatusOK(std::move(routingInfoStatus));
+            manager = routingInfo.cm();
+            primary = routingInfo.primary();
         }
     }
 
@@ -642,20 +618,17 @@ void ParallelSortClusteredCursor::startInit(OperationContext* opCtx) {
             if (staleNS.size() == 0)
                 staleNS = nss;  // ns is the *versioned* namespace, be careful of this
 
-            // Probably need to retry fully
-            bool forceReload;
-            _markStaleNS(opCtx, staleNS, e, forceReload);
+            _markStaleNS(staleNS, e);
+            Grid::get(opCtx)->catalogCache()->invalidateShardedCollection(staleNS);
 
-            LOG(1) << "stale config of ns " << staleNS
-                   << " during initialization, will retry with forced : " << forceReload
+            LOG(1) << "stale config of ns " << staleNS << " during initialization, will retry"
                    << causedBy(redact(e));
 
             // This is somewhat strange
-            if (staleNS != nss)
+            if (staleNS != nss) {
                 warning() << "versioned ns " << nss.ns() << " doesn't match stale config namespace "
                           << staleNS;
-
-            _handleStaleNS(opCtx, staleNS, forceReload);
+            }
 
             // Restart with new chunk manager
             startInit(opCtx);
@@ -860,26 +833,21 @@ void ParallelSortClusteredCursor::finishInit(OperationContext* opCtx) {
     if (retry) {
         // Refresh stale namespaces
         if (staleNSExceptions.size()) {
-            for (map<string, StaleConfigException>::iterator i = staleNSExceptions.begin(),
-                                                             end = staleNSExceptions.end();
-                 i != end;
-                 ++i) {
-                NamespaceString staleNS(i->first);
-                const StaleConfigException& exception = i->second;
+            for (const auto& exEntry : staleNSExceptions) {
+                const NamespaceString staleNS(exEntry.first);
+                const StaleConfigException& ex = exEntry.second;
 
-                bool forceReload;
-                _markStaleNS(opCtx, staleNS, exception, forceReload);
+                _markStaleNS(staleNS, ex);
+                Grid::get(opCtx)->catalogCache()->invalidateShardedCollection(staleNS);
 
-                LOG(1) << "stale config of ns " << staleNS
-                       << " on finishing query, will retry with forced : " << forceReload
-                       << causedBy(redact(exception));
+                LOG(1) << "stale config of ns " << staleNS << " on finishing query, will retry"
+                       << causedBy(redact(ex));
 
                 // This is somewhat strange
-                if (staleNS != ns)
+                if (staleNS != ns) {
                     warning() << "versioned ns " << ns << " doesn't match stale config namespace "
                               << staleNS;
-
-                _handleStaleNS(opCtx, staleNS, forceReload);
+                }
             }
         }
 
diff --git a/src/mongo/s/client/parallel.h b/src/mongo/s/client/parallel.h
index d375858bae0..d1680f1e74f 100644
--- a/src/mongo/s/client/parallel.h
+++ b/src/mongo/s/client/parallel.h
@@ -117,11 +117,7 @@ private:
 
     void _finishCons();
 
-    void _markStaleNS(OperationContext* opCtx,
-                      const NamespaceString& staleNS,
-                      const StaleConfigException& e,
-                      bool& forceReload);
-    void _handleStaleNS(OperationContext* opCtx, const NamespaceString& staleNS, bool forceReload);
+    void _markStaleNS(const NamespaceString& staleNS, const StaleConfigException& e);
 
     bool _didInit;
     bool _done;
diff --git a/src/mongo/s/client/version_manager.cpp b/src/mongo/s/client/version_manager.cpp
index 299a89f5941..7f97716f620 100644
--- a/src/mongo/s/client/version_manager.cpp
+++ b/src/mongo/s/client/version_manager.cpp
@@ -36,13 +36,13 @@
 #include "mongo/db/namespace_string.h"
 #include "mongo/s/catalog/sharding_catalog_client.h"
 #include "mongo/s/catalog_cache.h"
+#include "mongo/s/catalog_cache.h"
 #include "mongo/s/chunk_version.h"
 #include "mongo/s/client/shard_connection.h"
 #include "mongo/s/client/shard_registry.h"
 #include "mongo/s/grid.h"
 #include "mongo/s/is_mongos.h"
 #include "mongo/s/set_shard_version_request.h"
-#include "mongo/s/sharding_raii.h"
 #include "mongo/s/stale_exception.h"
 #include "mongo/util/log.h"
 
@@ -257,21 +257,21 @@ bool checkShardVersion(OperationContext* opCtx,
 
     const NamespaceString nss(ns);
 
+    auto const catalogCache = Grid::get(opCtx)->catalogCache();
+
     if (authoritative) {
-        ScopedChunkManager::refreshAndGet(opCtx, nss);
+        Grid::get(opCtx)->catalogCache()->invalidateShardedCollection(nss);
     }
 
-    auto scopedCMStatus = ScopedChunkManager::get(opCtx, nss);
-
-    if (!scopedCMStatus.isOK()) {
+    auto routingInfoStatus = catalogCache->getCollectionRoutingInfo(opCtx, nss);
+    if (!routingInfoStatus.isOK()) {
         return false;
     }
 
-    const auto& scopedCM = scopedCMStatus.getValue();
+    auto& routingInfo = routingInfoStatus.getValue();
 
-    auto conf = scopedCM.db();
-    const auto manager = scopedCM.cm();
-    const auto primary = scopedCM.primary();
+    const auto manager = routingInfo.cm();
+    const auto primary = routingInfo.primary();
 
     unsigned long long officialSequenceNumber = 0;
 
@@ -379,16 +379,7 @@ bool checkShardVersion(OperationContext* opCtx,
         return true;
     }
 
-    if (result["reloadConfig"].trueValue()) {
-        if (result["version"].timestampTime() == Date_t()) {
-            warning() << "reloading full configuration for " << conf->name()
-                      << ", connection state indicates significant version changes";
-
-            Grid::get(opCtx)->catalogCache()->invalidate(nss.db());
-        }
-
-        conf->getChunkManager(opCtx, nss.ns(), true);
-    }
+    Grid::get(opCtx)->catalogCache()->onStaleConfigError(std::move(routingInfo));
 
     const int maxNumTries = 7;
     if (tryNumber < maxNumTries) {
diff --git a/src/mongo/s/commands/chunk_manager_targeter.cpp b/src/mongo/s/commands/chunk_manager_targeter.cpp
index 609dee87d9e..42de7ba2903 100644
--- a/src/mongo/s/commands/chunk_manager_targeter.cpp
+++ b/src/mongo/s/commands/chunk_manager_targeter.cpp
@@ -32,31 +32,18 @@
 
 #include "mongo/s/commands/chunk_manager_targeter.h"
 
-#include <boost/thread/tss.hpp>
-
 #include "mongo/db/matcher/extensions_callback_noop.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/db/query/canonical_query.h"
 #include "mongo/db/query/collation/collation_index_key.h"
-#include "mongo/s/catalog_cache.h"
-#include "mongo/s/chunk.h"
 #include "mongo/s/client/shard_registry.h"
+#include "mongo/s/commands/cluster_commands_common.h"
 #include "mongo/s/grid.h"
 #include "mongo/s/shard_key_pattern.h"
-#include "mongo/s/sharding_raii.h"
-#include "mongo/stdx/memory.h"
 #include "mongo/util/log.h"
 #include "mongo/util/mongoutils/str.h"
 
 namespace mongo {
-
-using std::shared_ptr;
-using str::stream;
-using std::map;
-using std::set;
-using std::string;
-using std::vector;
-
 namespace {
 
 enum UpdateType { UpdateType_Replacement, UpdateType_OpStyle, UpdateType_Unknown };
@@ -65,11 +52,6 @@ enum CompareResult { CompareResult_Unknown, CompareResult_GTE, CompareResult_LT
 
 const ShardKeyPattern virtualIdShardKey(BSON("_id" << 1));
 
-// To match legacy reload behavior, we have to backoff on config reload per-thread
-// TODO: Centralize this behavior better by refactoring config reload in mongos
-boost::thread_specific_ptr<Backoff> perThreadBackoff;
-const int maxWaitMillis = 500;
-
 /**
  * There are two styles of update expressions:
  *
@@ -138,15 +120,6 @@ bool isExactIdQuery(OperationContext* opCtx, const CanonicalQuery& query, ChunkM
     return true;
 }
 
-void refreshBackoff() {
-    if (!perThreadBackoff.get()) {
-        perThreadBackoff.reset(new Backoff(maxWaitMillis, maxWaitMillis * 2));
-    }
-
-    perThreadBackoff.get()->nextSleepMillis();
-}
-
-
 //
 // Utilities to compare shard versions
 //
@@ -173,25 +146,19 @@ CompareResult compareShardVersions(const ChunkVersion& shardVersionA,
         return CompareResult_Unknown;
     }
 
-    if (shardVersionA < shardVersionB) {
+    if (shardVersionA < shardVersionB)
         return CompareResult_LT;
-    }
-
     else
         return CompareResult_GTE;
 }
 
-ChunkVersion getShardVersion(StringData shardName,
-                             const ChunkManager* manager,
-                             const Shard* primary) {
-    dassert(!(manager && primary));
-    dassert(manager || primary);
-
-    if (primary) {
-        return ChunkVersion::UNSHARDED();
+ChunkVersion getShardVersion(const CachedCollectionRoutingInfo& routingInfo,
+                             const ShardId& shardId) {
+    if (routingInfo.cm()) {
+        return routingInfo.cm()->getVersion(shardId);
     }
 
-    return manager->getVersion(shardName.toString());
+    return ChunkVersion::UNSHARDED();
 }
 
 /**
@@ -205,26 +172,21 @@ ChunkVersion getShardVersion(StringData shardName,
  * Note that the signature here is weird since our cached map of chunk versions is stored in a
  * ChunkManager or is implicit in the primary shard of the collection.
  */
-CompareResult compareAllShardVersions(const ChunkManager* cachedChunkManager,
-                                      const Shard* cachedPrimary,
-                                      const map<ShardId, ChunkVersion>& remoteShardVersions) {
+CompareResult compareAllShardVersions(const CachedCollectionRoutingInfo& routingInfo,
+                                      const ShardVersionMap& remoteShardVersions) {
     CompareResult finalResult = CompareResult_GTE;
 
-    for (map<ShardId, ChunkVersion>::const_iterator it = remoteShardVersions.begin();
-         it != remoteShardVersions.end();
-         ++it) {
-        // Get the remote and cached version for the next shard
-        const ShardId& shardName = it->first;
-        const ChunkVersion& remoteShardVersion = it->second;
+    for (const auto& shardVersionEntry : remoteShardVersions) {
+        const ShardId& shardId = shardVersionEntry.first;
+        const ChunkVersion& remoteShardVersion = shardVersionEntry.second;
 
         ChunkVersion cachedShardVersion;
 
         try {
             // Throws b/c shard constructor throws
-            cachedShardVersion =
-                getShardVersion(shardName.toString(), cachedChunkManager, cachedPrimary);
+            cachedShardVersion = getShardVersion(routingInfo, shardId);
         } catch (const DBException& ex) {
-            warning() << "could not lookup shard " << shardName
+            warning() << "could not lookup shard " << shardId
                       << " in local cache, shard metadata may have changed"
                       << " or be unavailable" << causedBy(ex);
 
@@ -236,6 +198,7 @@ CompareResult compareAllShardVersions(const ChunkManager* cachedChunkManager,
 
         if (result == CompareResult_Unknown)
             return result;
+
         if (result == CompareResult_LT)
             finalResult = CompareResult_LT;
 
@@ -248,10 +211,10 @@ CompareResult compareAllShardVersions(const ChunkManager* cachedChunkManager,
 /**
  * Whether or not the manager/primary pair is different from the other manager/primary pair.
  */
-bool isMetadataDifferent(const shared_ptr<ChunkManager>& managerA,
-                         const shared_ptr<Shard>& primaryA,
-                         const shared_ptr<ChunkManager>& managerB,
-                         const shared_ptr<Shard>& primaryB) {
+bool isMetadataDifferent(const std::shared_ptr<ChunkManager>& managerA,
+                         const std::shared_ptr<Shard>& primaryA,
+                         const std::shared_ptr<ChunkManager>& managerB,
+                         const std::shared_ptr<Shard>& primaryB) {
     if ((managerA && !managerB) || (!managerA && managerB) || (primaryA && !primaryB) ||
         (!primaryA && primaryB))
         return true;
@@ -268,10 +231,10 @@ bool isMetadataDifferent(const shared_ptr<ChunkManager>& managerA,
 * Whether or not the manager/primary pair was changed or refreshed from a previous version
 * of the metadata.
 */
-bool wasMetadataRefreshed(const shared_ptr<ChunkManager>& managerA,
-                          const shared_ptr<Shard>& primaryA,
-                          const shared_ptr<ChunkManager>& managerB,
-                          const shared_ptr<Shard>& primaryB) {
+bool wasMetadataRefreshed(const std::shared_ptr<ChunkManager>& managerA,
+                          const std::shared_ptr<Shard>& primaryA,
+                          const std::shared_ptr<ChunkManager>& managerB,
+                          const std::shared_ptr<Shard>& primaryB) {
     if (isMetadataDifferent(managerA, primaryA, managerB, primaryB))
         return true;
 
@@ -290,14 +253,18 @@ ChunkManagerTargeter::ChunkManagerTargeter(const NamespaceString& nss, TargeterS
 
 
 Status ChunkManagerTargeter::init(OperationContext* opCtx) {
-    auto scopedCMStatus = ScopedChunkManager::getOrCreate(opCtx, _nss);
-    if (!scopedCMStatus.isOK()) {
-        return scopedCMStatus.getStatus();
+    auto shardDbStatus = createShardDatabase(opCtx, _nss.db());
+    if (!shardDbStatus.isOK()) {
+        return shardDbStatus.getStatus();
     }
 
-    const auto& scopedCM = scopedCMStatus.getValue();
-    _manager = scopedCM.cm();
-    _primary = scopedCM.primary();
+    const auto routingInfoStatus =
+        Grid::get(opCtx)->catalogCache()->getCollectionRoutingInfo(opCtx, _nss);
+    if (!routingInfoStatus.isOK()) {
+        return routingInfoStatus.getStatus();
+    }
+
+    _routingInfo = std::move(routingInfoStatus.getValue());
 
     return Status::OK();
 }
@@ -311,21 +278,21 @@ Status ChunkManagerTargeter::targetInsert(OperationContext* opCtx,
                                           ShardEndpoint** endpoint) const {
     BSONObj shardKey;
 
-    if (_manager) {
+    if (_routingInfo->cm()) {
         //
         // Sharded collections have the following requirements for targeting:
         //
         // Inserts must contain the exact shard key.
         //
 
-        shardKey = _manager->getShardKeyPattern().extractShardKeyFromDoc(doc);
+        shardKey = _routingInfo->cm()->getShardKeyPattern().extractShardKeyFromDoc(doc);
 
         // Check shard key exists
         if (shardKey.isEmpty()) {
-            return Status(ErrorCodes::ShardKeyNotFound,
-                          stream() << "document " << doc
-                                   << " does not contain shard key for pattern "
-                                   << _manager->getShardKeyPattern().toString());
+            return {ErrorCodes::ShardKeyNotFound,
+                    str::stream() << "document " << doc
+                                  << " does not contain shard key for pattern "
+                                  << _routingInfo->cm()->getShardKeyPattern().toString()};
         }
 
         // Check shard key size on insert
@@ -338,13 +305,13 @@ Status ChunkManagerTargeter::targetInsert(OperationContext* opCtx,
     if (!shardKey.isEmpty()) {
         *endpoint = targetShardKey(shardKey, CollationSpec::kSimpleSpec, doc.objsize()).release();
     } else {
-        if (!_primary) {
+        if (!_routingInfo->primary()) {
             return Status(ErrorCodes::NamespaceNotFound,
                           str::stream() << "could not target insert in collection " << getNS().ns()
                                         << "; no metadata found");
         }
 
-        *endpoint = new ShardEndpoint(_primary->getId(), ChunkVersion::UNSHARDED());
+        *endpoint = new ShardEndpoint(_routingInfo->primary()->getId(), ChunkVersion::UNSHARDED());
     }
 
     return Status::OK();
@@ -376,14 +343,14 @@ Status ChunkManagerTargeter::targetUpdate(
     UpdateType updateType = getUpdateExprType(updateDoc.getUpdateExpr());
 
     if (updateType == UpdateType_Unknown) {
-        return Status(ErrorCodes::UnsupportedFormat,
-                      stream() << "update document " << updateExpr
-                               << " has mixed $operator and non-$operator style fields");
+        return {ErrorCodes::UnsupportedFormat,
+                str::stream() << "update document " << updateExpr
+                              << " has mixed $operator and non-$operator style fields"};
     }
 
     BSONObj shardKey;
 
-    if (_manager) {
+    if (_routingInfo->cm()) {
         //
         // Sharded collections have the following futher requirements for targeting:
         //
@@ -395,7 +362,7 @@ Status ChunkManagerTargeter::targetUpdate(
         if (updateType == UpdateType_OpStyle) {
             // Target using the query
             StatusWith<BSONObj> status =
-                _manager->getShardKeyPattern().extractShardKeyFromQuery(opCtx, query);
+                _routingInfo->cm()->getShardKeyPattern().extractShardKeyFromQuery(opCtx, query);
 
             // Bad query
             if (!status.isOK())
@@ -404,7 +371,7 @@ Status ChunkManagerTargeter::targetUpdate(
             shardKey = status.getValue();
         } else {
             // Target using the replacement document
-            shardKey = _manager->getShardKeyPattern().extractShardKeyFromDoc(updateExpr);
+            shardKey = _routingInfo->cm()->getShardKeyPattern().extractShardKeyFromDoc(updateExpr);
         }
 
         // Check shard key size on upsert.
@@ -431,13 +398,13 @@ Status ChunkManagerTargeter::targetUpdate(
     // We failed to target a single shard.
 
     // Upserts are required to target a single shard.
-    if (_manager && updateDoc.getUpsert()) {
+    if (_routingInfo->cm() && updateDoc.getUpsert()) {
         return Status(ErrorCodes::ShardKeyNotFound,
                       str::stream() << "An upsert on a sharded collection must contain the shard "
                                        "key and have the simple collation. Update request: "
                                     << updateDoc.toBSON()
                                     << ", shard key pattern: "
-                                    << _manager->getShardKeyPattern().toString());
+                                    << _routingInfo->cm()->getShardKeyPattern().toString());
     }
 
     // Parse update query.
@@ -454,8 +421,8 @@ Status ChunkManagerTargeter::targetUpdate(
     }
 
     // Single (non-multi) updates must target a single shard or be exact-ID.
-    if (_manager && !updateDoc.getMulti() &&
-        !isExactIdQuery(opCtx, *cq.getValue(), _manager.get())) {
+    if (_routingInfo->cm() && !updateDoc.getMulti() &&
+        !isExactIdQuery(opCtx, *cq.getValue(), _routingInfo->cm().get())) {
         return Status(ErrorCodes::ShardKeyNotFound,
                       str::stream()
                           << "A single update on a sharded collection must contain an exact "
@@ -464,7 +431,7 @@ Status ChunkManagerTargeter::targetUpdate(
                              "request: "
                           << updateDoc.toBSON()
                           << ", shard key pattern: "
-                          << _manager->getShardKeyPattern().toString());
+                          << _routingInfo->cm()->getShardKeyPattern().toString());
     }
 
     if (updateType == UpdateType_OpStyle) {
@@ -480,7 +447,7 @@ Status ChunkManagerTargeter::targetDelete(
     std::vector<std::unique_ptr<ShardEndpoint>>* endpoints) const {
     BSONObj shardKey;
 
-    if (_manager) {
+    if (_routingInfo->cm()) {
         //
         // Sharded collections have the following further requirements for targeting:
         //
@@ -489,7 +456,8 @@ Status ChunkManagerTargeter::targetDelete(
 
         // Get the shard key
         StatusWith<BSONObj> status =
-            _manager->getShardKeyPattern().extractShardKeyFromQuery(opCtx, deleteDoc.getQuery());
+            _routingInfo->cm()->getShardKeyPattern().extractShardKeyFromQuery(opCtx,
+                                                                              deleteDoc.getQuery());
 
         // Bad query
         if (!status.isOK())
@@ -527,8 +495,8 @@ Status ChunkManagerTargeter::targetDelete(
     }
 
     // Single deletes must target a single shard or be exact-ID.
-    if (_manager && deleteDoc.getLimit() == 1 &&
-        !isExactIdQuery(opCtx, *cq.getValue(), _manager.get())) {
+    if (_routingInfo->cm() && deleteDoc.getLimit() == 1 &&
+        !isExactIdQuery(opCtx, *cq.getValue(), _routingInfo->cm().get())) {
         return Status(ErrorCodes::ShardKeyNotFound,
                       str::stream()
                           << "A single delete on a sharded collection must contain an exact "
@@ -537,7 +505,7 @@ Status ChunkManagerTargeter::targetDelete(
                              "request: "
                           << deleteDoc.toBSON()
                           << ", shard key pattern: "
-                          << _manager->getShardKeyPattern().toString());
+                          << _routingInfo->cm()->getShardKeyPattern().toString());
     }
 
     return targetQuery(opCtx, deleteDoc.getQuery(), collation, endpoints);
@@ -558,26 +526,28 @@ Status ChunkManagerTargeter::targetQuery(
     const BSONObj& query,
     const BSONObj& collation,
     std::vector<std::unique_ptr<ShardEndpoint>>* endpoints) const {
-    if (!_primary && !_manager) {
-        return Status(ErrorCodes::NamespaceNotFound,
-                      stream() << "could not target query in " << getNS().ns()
-                               << "; no metadata found");
+    if (!_routingInfo->primary() && !_routingInfo->cm()) {
+        return {ErrorCodes::NamespaceNotFound,
+                str::stream() << "could not target query in " << getNS().ns()
+                              << "; no metadata found"};
     }
 
-    set<ShardId> shardIds;
-    if (_manager) {
+    std::set<ShardId> shardIds;
+    if (_routingInfo->cm()) {
         try {
-            _manager->getShardIdsForQuery(opCtx, query, collation, &shardIds);
+            _routingInfo->cm()->getShardIdsForQuery(opCtx, query, collation, &shardIds);
         } catch (const DBException& ex) {
             return ex.toStatus();
         }
     } else {
-        shardIds.insert(_primary->getId());
+        shardIds.insert(_routingInfo->primary()->getId());
     }
 
     for (const ShardId& shardId : shardIds) {
         endpoints->push_back(stdx::make_unique<ShardEndpoint>(
-            shardId, _manager ? _manager->getVersion(shardId) : ChunkVersion::UNSHARDED()));
+            shardId,
+            _routingInfo->cm() ? _routingInfo->cm()->getVersion(shardId)
+                               : ChunkVersion::UNSHARDED()));
     }
 
     return Status::OK();
@@ -586,7 +556,7 @@ Status ChunkManagerTargeter::targetQuery(
 std::unique_ptr<ShardEndpoint> ChunkManagerTargeter::targetShardKey(const BSONObj& shardKey,
                                                                     const BSONObj& collation,
                                                                     long long estDataSize) const {
-    auto chunk = _manager->findIntersectingChunk(shardKey, collation);
+    auto chunk = _routingInfo->cm()->findIntersectingChunk(shardKey, collation);
 
     // Track autosplit stats for sharded collections
     // Note: this is only best effort accounting and is not accurate.
@@ -595,27 +565,29 @@ std::unique_ptr<ShardEndpoint> ChunkManagerTargeter::targetShardKey(const BSONOb
     }
 
     return stdx::make_unique<ShardEndpoint>(chunk->getShardId(),
-                                            _manager->getVersion(chunk->getShardId()));
+                                            _routingInfo->cm()->getVersion(chunk->getShardId()));
 }
 
 Status ChunkManagerTargeter::targetCollection(
     std::vector<std::unique_ptr<ShardEndpoint>>* endpoints) const {
-    if (!_primary && !_manager) {
-        return Status(ErrorCodes::NamespaceNotFound,
-                      str::stream() << "could not target full range of " << getNS().ns()
-                                    << "; metadata not found");
+    if (!_routingInfo->primary() && !_routingInfo->cm()) {
+        return {ErrorCodes::NamespaceNotFound,
+                str::stream() << "could not target full range of " << getNS().ns()
+                              << "; metadata not found"};
     }
 
-    set<ShardId> shardIds;
-    if (_manager) {
-        _manager->getAllShardIds(&shardIds);
+    std::set<ShardId> shardIds;
+    if (_routingInfo->cm()) {
+        _routingInfo->cm()->getAllShardIds(&shardIds);
     } else {
-        shardIds.insert(_primary->getId());
+        shardIds.insert(_routingInfo->primary()->getId());
     }
 
     for (const ShardId& shardId : shardIds) {
         endpoints->push_back(stdx::make_unique<ShardEndpoint>(
-            shardId, _manager ? _manager->getVersion(shardId) : ChunkVersion::UNSHARDED()));
+            shardId,
+            _routingInfo->cm() ? _routingInfo->cm()->getVersion(shardId)
+                               : ChunkVersion::UNSHARDED()));
     }
 
     return Status::OK();
@@ -623,19 +595,20 @@ Status ChunkManagerTargeter::targetCollection(
 
 Status ChunkManagerTargeter::targetAllShards(
     std::vector<std::unique_ptr<ShardEndpoint>>* endpoints) const {
-    if (!_primary && !_manager) {
-        return Status(ErrorCodes::NamespaceNotFound,
-                      str::stream() << "could not target every shard with versions for "
-                                    << getNS().ns()
-                                    << "; metadata not found");
+    if (!_routingInfo->primary() && !_routingInfo->cm()) {
+        return {ErrorCodes::NamespaceNotFound,
+                str::stream() << "could not target every shard with versions for " << getNS().ns()
+                              << "; metadata not found"};
     }
 
-    vector<ShardId> shardIds;
+    std::vector<ShardId> shardIds;
     grid.shardRegistry()->getAllShardIds(&shardIds);
 
     for (const ShardId& shardId : shardIds) {
         endpoints->push_back(stdx::make_unique<ShardEndpoint>(
-            shardId, _manager ? _manager->getVersion(shardId) : ChunkVersion::UNSHARDED()));
+            shardId,
+            _routingInfo->cm() ? _routingInfo->cm()->getVersion(shardId)
+                               : ChunkVersion::UNSHARDED()));
     }
 
     return Status::OK();
@@ -649,8 +622,7 @@ void ChunkManagerTargeter::noteStaleResponse(const ShardEndpoint& endpoint,
     if (staleInfo["vWanted"].eoo()) {
         // If we don't have a vWanted sent, assume the version is higher than our current
         // version.
-        remoteShardVersion =
-            getShardVersion(endpoint.shardName.toString(), _manager.get(), _primary.get());
+        remoteShardVersion = getShardVersion(*_routingInfo, endpoint.shardName);
         remoteShardVersion.incMajor();
     } else {
         remoteShardVersion = ChunkVersion::fromBSON(staleInfo, "vWanted");
@@ -699,18 +671,14 @@ Status ChunkManagerTargeter::refreshIfNeeded(OperationContext* opCtx, bool* wasC
     // Get the latest metadata information from the cache if there were issues
     //
 
-    shared_ptr<ChunkManager> lastManager = _manager;
-    shared_ptr<Shard> lastPrimary = _primary;
+    auto lastManager = _routingInfo->cm();
+    auto lastPrimary = _routingInfo->primary();
 
-    auto scopedCMStatus = ScopedChunkManager::getOrCreate(opCtx, _nss);
-    if (!scopedCMStatus.isOK()) {
-        return scopedCMStatus.getStatus();
+    auto initStatus = init(opCtx);
+    if (!initStatus.isOK()) {
+        return initStatus;
     }
 
-    const auto& scopedCM = scopedCMStatus.getValue();
-    _manager = scopedCM.cm();
-    _primary = scopedCM.primary();
-
     // We now have the latest metadata from the cache.
 
     //
@@ -718,8 +686,6 @@ Status ChunkManagerTargeter::refreshIfNeeded(OperationContext* opCtx, bool* wasC
     // Either we couldn't target at all, or we have stale versions, but not both.
     //
 
-    dassert(!(_needsTargetingRefresh && !_remoteShardVersions.empty()));
-
     if (_needsTargetingRefresh) {
         // Reset the field
         _needsTargetingRefresh = false;
@@ -728,63 +694,44 @@ Status ChunkManagerTargeter::refreshIfNeeded(OperationContext* opCtx, bool* wasC
         // the
         // metadata since we last got it from the cache.
 
-        bool alreadyRefreshed = wasMetadataRefreshed(lastManager, lastPrimary, _manager, _primary);
+        bool alreadyRefreshed = wasMetadataRefreshed(
+            lastManager, lastPrimary, _routingInfo->cm(), _routingInfo->primary());
 
         // If didn't already refresh the targeting information, refresh it
         if (!alreadyRefreshed) {
             // To match previous behavior, we just need an incremental refresh here
-            return refreshNow(opCtx, RefreshType_RefreshChunkManager);
+            return refreshNow(opCtx);
         }
 
-        *wasChanged = isMetadataDifferent(lastManager, lastPrimary, _manager, _primary);
+        *wasChanged = isMetadataDifferent(
+            lastManager, lastPrimary, _routingInfo->cm(), _routingInfo->primary());
         return Status::OK();
     } else if (!_remoteShardVersions.empty()) {
         // If we got stale shard versions from remote shards, we may need to refresh
         // NOTE: Not sure yet if this can happen simultaneously with targeting issues
 
-        CompareResult result =
-            compareAllShardVersions(_manager.get(), _primary.get(), _remoteShardVersions);
+        CompareResult result = compareAllShardVersions(*_routingInfo, _remoteShardVersions);
+
         // Reset the versions
         _remoteShardVersions.clear();
 
-        if (result == CompareResult_Unknown) {
+        if (result == CompareResult_Unknown || result == CompareResult_LT) {
             // Our current shard versions aren't all comparable to the old versions, maybe drop
-            return refreshNow(opCtx, RefreshType_ReloadDatabase);
-        } else if (result == CompareResult_LT) {
-            // Our current shard versions are less than the remote versions, but no drop
-            return refreshNow(opCtx, RefreshType_RefreshChunkManager);
+            return refreshNow(opCtx);
         }
 
-        *wasChanged = isMetadataDifferent(lastManager, lastPrimary, _manager, _primary);
+        *wasChanged = isMetadataDifferent(
+            lastManager, lastPrimary, _routingInfo->cm(), _routingInfo->primary());
         return Status::OK();
     }
 
-    // unreachable
-    dassert(false);
-    return Status::OK();
+    MONGO_UNREACHABLE;
 }
 
-Status ChunkManagerTargeter::refreshNow(OperationContext* opCtx, RefreshType refreshType) {
-    if (refreshType == RefreshType_ReloadDatabase) {
-        Grid::get(opCtx)->catalogCache()->invalidate(_nss.db().toString());
-    }
-
-    // Try not to spam the configs
-    refreshBackoff();
+Status ChunkManagerTargeter::refreshNow(OperationContext* opCtx) {
+    Grid::get(opCtx)->catalogCache()->onStaleConfigError(std::move(*_routingInfo));
 
-    ScopedChunkManager::refreshAndGet(opCtx, _nss);
-
-    auto scopedCMStatus = ScopedChunkManager::get(opCtx, _nss);
-    if (!scopedCMStatus.isOK()) {
-        return scopedCMStatus.getStatus();
-    }
-
-    const auto& scopedCM = scopedCMStatus.getValue();
-
-    _manager = scopedCM.cm();
-    _primary = scopedCM.primary();
-
-    return Status::OK();
+    return init(opCtx);
 }
 
 }  // namespace mongo
diff --git a/src/mongo/s/commands/chunk_manager_targeter.cpp-8454ea5f b/src/mongo/s/commands/chunk_manager_targeter.cpp-8454ea5f
new file mode 100644
index 00000000000..42de7ba2903
--- /dev/null
+++ b/src/mongo/s/commands/chunk_manager_targeter.cpp-8454ea5f
@@ -0,0 +1,737 @@
+/**
+ *    Copyright (C) 2013 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kSharding
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/s/commands/chunk_manager_targeter.h"
+
+#include "mongo/db/matcher/extensions_callback_noop.h"
+#include "mongo/db/operation_context.h"
+#include "mongo/db/query/canonical_query.h"
+#include "mongo/db/query/collation/collation_index_key.h"
+#include "mongo/s/client/shard_registry.h"
+#include "mongo/s/commands/cluster_commands_common.h"
+#include "mongo/s/grid.h"
+#include "mongo/s/shard_key_pattern.h"
+#include "mongo/util/log.h"
+#include "mongo/util/mongoutils/str.h"
+
+namespace mongo {
+namespace {
+
+enum UpdateType { UpdateType_Replacement, UpdateType_OpStyle, UpdateType_Unknown };
+
+enum CompareResult { CompareResult_Unknown, CompareResult_GTE, CompareResult_LT };
+
+const ShardKeyPattern virtualIdShardKey(BSON("_id" << 1));
+
+/**
+ * There are two styles of update expressions:
+ *
+ * Replacement style: coll.update({ x : 1 }, { y : 2 })
+ * OpStyle: coll.update({ x : 1 }, { $set : { y : 2 } })
+ */
+UpdateType getUpdateExprType(const BSONObj& updateExpr) {
+    // Empty update is replacement-style, by default
+    if (updateExpr.isEmpty()) {
+        return UpdateType_Replacement;
+    }
+
+    UpdateType updateType = UpdateType_Unknown;
+
+    BSONObjIterator it(updateExpr);
+    while (it.more()) {
+        BSONElement next = it.next();
+
+        if (next.fieldName()[0] == '$') {
+            if (updateType == UpdateType_Unknown) {
+                updateType = UpdateType_OpStyle;
+            } else if (updateType == UpdateType_Replacement) {
+                return UpdateType_Unknown;
+            }
+        } else {
+            if (updateType == UpdateType_Unknown) {
+                updateType = UpdateType_Replacement;
+            } else if (updateType == UpdateType_OpStyle) {
+                return UpdateType_Unknown;
+            }
+        }
+    }
+
+    return updateType;
+}
+
+/**
+ * This returns "does the query have an _id field" and "is the _id field querying for a direct
+ * value like _id : 3 and not _id : { $gt : 3 }"
+ *
+ * If the query does not use the collection default collation, the _id field cannot contain strings,
+ * objects, or arrays.
+ *
+ * Ex: { _id : 1 } => true
+ *     { foo : <anything>, _id : 1 } => true
+ *     { _id : { $lt : 30 } } => false
+ *     { foo : <anything> } => false
+ */
+bool isExactIdQuery(OperationContext* opCtx, const CanonicalQuery& query, ChunkManager* manager) {
+    auto shardKey = virtualIdShardKey.extractShardKeyFromQuery(query);
+    BSONElement idElt = shardKey["_id"];
+
+    if (!idElt) {
+        return false;
+    }
+
+    if (CollationIndexKey::isCollatableType(idElt.type()) && manager &&
+        !query.getQueryRequest().getCollation().isEmpty() &&
+        !CollatorInterface::collatorsMatch(query.getCollator(), manager->getDefaultCollator())) {
+
+        // The collation applies to the _id field, but the user specified a collation which doesn't
+        // match the collection default.
+        return false;
+    }
+
+    return true;
+}
+
+//
+// Utilities to compare shard versions
+//
+
+/**
+ * Returns the relationship of two shard versions. Shard versions of a collection that has not
+ * been dropped and recreated and where there is at least one chunk on a shard are comparable,
+ * otherwise the result is ambiguous.
+ */
+CompareResult compareShardVersions(const ChunkVersion& shardVersionA,
+                                   const ChunkVersion& shardVersionB) {
+    // Collection may have been dropped
+    if (!shardVersionA.hasEqualEpoch(shardVersionB)) {
+        return CompareResult_Unknown;
+    }
+
+    // Zero shard versions are only comparable to themselves
+    if (!shardVersionA.isSet() || !shardVersionB.isSet()) {
+        // If both are zero...
+        if (!shardVersionA.isSet() && !shardVersionB.isSet()) {
+            return CompareResult_GTE;
+        }
+
+        return CompareResult_Unknown;
+    }
+
+    if (shardVersionA < shardVersionB)
+        return CompareResult_LT;
+    else
+        return CompareResult_GTE;
+}
+
+ChunkVersion getShardVersion(const CachedCollectionRoutingInfo& routingInfo,
+                             const ShardId& shardId) {
+    if (routingInfo.cm()) {
+        return routingInfo.cm()->getVersion(shardId);
+    }
+
+    return ChunkVersion::UNSHARDED();
+}
+
+/**
+ * Returns the relationship between two maps of shard versions. As above, these maps are often
+ * comparable when the collection has not been dropped and there is at least one chunk on the
+ * shards. If any versions in the maps are not comparable, the result is _Unknown.
+ *
+ * If any versions in the first map (cached) are _LT the versions in the second map (remote),
+ * the first (cached) versions are _LT the second (remote) versions.
+ *
+ * Note that the signature here is weird since our cached map of chunk versions is stored in a
+ * ChunkManager or is implicit in the primary shard of the collection.
+ */
+CompareResult compareAllShardVersions(const CachedCollectionRoutingInfo& routingInfo,
+                                      const ShardVersionMap& remoteShardVersions) {
+    CompareResult finalResult = CompareResult_GTE;
+
+    for (const auto& shardVersionEntry : remoteShardVersions) {
+        const ShardId& shardId = shardVersionEntry.first;
+        const ChunkVersion& remoteShardVersion = shardVersionEntry.second;
+
+        ChunkVersion cachedShardVersion;
+
+        try {
+            // Throws b/c shard constructor throws
+            cachedShardVersion = getShardVersion(routingInfo, shardId);
+        } catch (const DBException& ex) {
+            warning() << "could not lookup shard " << shardId
+                      << " in local cache, shard metadata may have changed"
+                      << " or be unavailable" << causedBy(ex);
+
+            return CompareResult_Unknown;
+        }
+
+        // Compare the remote and cached versions
+        CompareResult result = compareShardVersions(cachedShardVersion, remoteShardVersion);
+
+        if (result == CompareResult_Unknown)
+            return result;
+
+        if (result == CompareResult_LT)
+            finalResult = CompareResult_LT;
+
+        // Note that we keep going after _LT b/c there could be more _Unknowns.
+    }
+
+    return finalResult;
+}
+
+/**
+ * Whether or not the manager/primary pair is different from the other manager/primary pair.
+ */
+bool isMetadataDifferent(const std::shared_ptr<ChunkManager>& managerA,
+                         const std::shared_ptr<Shard>& primaryA,
+                         const std::shared_ptr<ChunkManager>& managerB,
+                         const std::shared_ptr<Shard>& primaryB) {
+    if ((managerA && !managerB) || (!managerA && managerB) || (primaryA && !primaryB) ||
+        (!primaryA && primaryB))
+        return true;
+
+    if (managerA) {
+        return !managerA->getVersion().isStrictlyEqualTo(managerB->getVersion());
+    }
+
+    dassert(NULL != primaryA.get());
+    return primaryA->getId() != primaryB->getId();
+}
+
+/**
+* Whether or not the manager/primary pair was changed or refreshed from a previous version
+* of the metadata.
+*/
+bool wasMetadataRefreshed(const std::shared_ptr<ChunkManager>& managerA,
+                          const std::shared_ptr<Shard>& primaryA,
+                          const std::shared_ptr<ChunkManager>& managerB,
+                          const std::shared_ptr<Shard>& primaryB) {
+    if (isMetadataDifferent(managerA, primaryA, managerB, primaryB))
+        return true;
+
+    if (managerA) {
+        dassert(managerB.get());  // otherwise metadata would be different
+        return managerA->getSequenceNumber() != managerB->getSequenceNumber();
+    }
+
+    return false;
+}
+
+}  // namespace
+
+ChunkManagerTargeter::ChunkManagerTargeter(const NamespaceString& nss, TargeterStats* stats)
+    : _nss(nss), _needsTargetingRefresh(false), _stats(stats) {}
+
+
+Status ChunkManagerTargeter::init(OperationContext* opCtx) {
+    auto shardDbStatus = createShardDatabase(opCtx, _nss.db());
+    if (!shardDbStatus.isOK()) {
+        return shardDbStatus.getStatus();
+    }
+
+    const auto routingInfoStatus =
+        Grid::get(opCtx)->catalogCache()->getCollectionRoutingInfo(opCtx, _nss);
+    if (!routingInfoStatus.isOK()) {
+        return routingInfoStatus.getStatus();
+    }
+
+    _routingInfo = std::move(routingInfoStatus.getValue());
+
+    return Status::OK();
+}
+
+const NamespaceString& ChunkManagerTargeter::getNS() const {
+    return _nss;
+}
+
+Status ChunkManagerTargeter::targetInsert(OperationContext* opCtx,
+                                          const BSONObj& doc,
+                                          ShardEndpoint** endpoint) const {
+    BSONObj shardKey;
+
+    if (_routingInfo->cm()) {
+        //
+        // Sharded collections have the following requirements for targeting:
+        //
+        // Inserts must contain the exact shard key.
+        //
+
+        shardKey = _routingInfo->cm()->getShardKeyPattern().extractShardKeyFromDoc(doc);
+
+        // Check shard key exists
+        if (shardKey.isEmpty()) {
+            return {ErrorCodes::ShardKeyNotFound,
+                    str::stream() << "document " << doc
+                                  << " does not contain shard key for pattern "
+                                  << _routingInfo->cm()->getShardKeyPattern().toString()};
+        }
+
+        // Check shard key size on insert
+        Status status = ShardKeyPattern::checkShardKeySize(shardKey);
+        if (!status.isOK())
+            return status;
+    }
+
+    // Target the shard key or database primary
+    if (!shardKey.isEmpty()) {
+        *endpoint = targetShardKey(shardKey, CollationSpec::kSimpleSpec, doc.objsize()).release();
+    } else {
+        if (!_routingInfo->primary()) {
+            return Status(ErrorCodes::NamespaceNotFound,
+                          str::stream() << "could not target insert in collection " << getNS().ns()
+                                        << "; no metadata found");
+        }
+
+        *endpoint = new ShardEndpoint(_routingInfo->primary()->getId(), ChunkVersion::UNSHARDED());
+    }
+
+    return Status::OK();
+}
+
+Status ChunkManagerTargeter::targetUpdate(
+    OperationContext* opCtx,
+    const BatchedUpdateDocument& updateDoc,
+    std::vector<std::unique_ptr<ShardEndpoint>>* endpoints) const {
+    //
+    // Update targeting may use either the query or the update.  This is to support save-style
+    // updates, of the form:
+    //
+    // coll.update({ _id : xxx }, { _id : xxx, shardKey : 1, foo : bar }, { upsert : true })
+    //
+    // Because drivers do not know the shard key, they can't pull the shard key automatically
+    // into the query doc, and to correctly support upsert we must target a single shard.
+    //
+    // The rule is simple - If the update is replacement style (no '$set'), we target using the
+    // update.  If the update is replacement style, we target using the query.
+    //
+    // If we have the exact shard key in either the query or replacement doc, we target using
+    // that extracted key.
+    //
+
+    BSONObj query = updateDoc.getQuery();
+    BSONObj updateExpr = updateDoc.getUpdateExpr();
+
+    UpdateType updateType = getUpdateExprType(updateDoc.getUpdateExpr());
+
+    if (updateType == UpdateType_Unknown) {
+        return {ErrorCodes::UnsupportedFormat,
+                str::stream() << "update document " << updateExpr
+                              << " has mixed $operator and non-$operator style fields"};
+    }
+
+    BSONObj shardKey;
+
+    if (_routingInfo->cm()) {
+        //
+        // Sharded collections have the following futher requirements for targeting:
+        //
+        // Upserts must be targeted exactly by shard key.
+        // Non-multi updates must be targeted exactly by shard key *or* exact _id.
+        //
+
+        // Get the shard key
+        if (updateType == UpdateType_OpStyle) {
+            // Target using the query
+            StatusWith<BSONObj> status =
+                _routingInfo->cm()->getShardKeyPattern().extractShardKeyFromQuery(opCtx, query);
+
+            // Bad query
+            if (!status.isOK())
+                return status.getStatus();
+
+            shardKey = status.getValue();
+        } else {
+            // Target using the replacement document
+            shardKey = _routingInfo->cm()->getShardKeyPattern().extractShardKeyFromDoc(updateExpr);
+        }
+
+        // Check shard key size on upsert.
+        if (updateDoc.getUpsert()) {
+            Status status = ShardKeyPattern::checkShardKeySize(shardKey);
+            if (!status.isOK())
+                return status;
+        }
+    }
+
+    const BSONObj collation = updateDoc.isCollationSet() ? updateDoc.getCollation() : BSONObj();
+
+    // Target the shard key, query, or replacement doc
+    if (!shardKey.isEmpty()) {
+        try {
+            endpoints->push_back(
+                targetShardKey(shardKey, collation, (query.objsize() + updateExpr.objsize())));
+            return Status::OK();
+        } catch (const DBException&) {
+            // This update is potentially not constrained to a single shard
+        }
+    }
+
+    // We failed to target a single shard.
+
+    // Upserts are required to target a single shard.
+    if (_routingInfo->cm() && updateDoc.getUpsert()) {
+        return Status(ErrorCodes::ShardKeyNotFound,
+                      str::stream() << "An upsert on a sharded collection must contain the shard "
+                                       "key and have the simple collation. Update request: "
+                                    << updateDoc.toBSON()
+                                    << ", shard key pattern: "
+                                    << _routingInfo->cm()->getShardKeyPattern().toString());
+    }
+
+    // Parse update query.
+    auto qr = stdx::make_unique<QueryRequest>(getNS());
+    qr->setFilter(updateDoc.getQuery());
+    if (!collation.isEmpty()) {
+        qr->setCollation(collation);
+    }
+    auto cq = CanonicalQuery::canonicalize(opCtx, std::move(qr), ExtensionsCallbackNoop());
+    if (!cq.isOK()) {
+        return Status(cq.getStatus().code(),
+                      str::stream() << "Could not parse update query " << updateDoc.getQuery()
+                                    << causedBy(cq.getStatus()));
+    }
+
+    // Single (non-multi) updates must target a single shard or be exact-ID.
+    if (_routingInfo->cm() && !updateDoc.getMulti() &&
+        !isExactIdQuery(opCtx, *cq.getValue(), _routingInfo->cm().get())) {
+        return Status(ErrorCodes::ShardKeyNotFound,
+                      str::stream()
+                          << "A single update on a sharded collection must contain an exact "
+                             "match on _id (and have the collection default collation) or "
+                             "contain the shard key (and have the simple collation). Update "
+                             "request: "
+                          << updateDoc.toBSON()
+                          << ", shard key pattern: "
+                          << _routingInfo->cm()->getShardKeyPattern().toString());
+    }
+
+    if (updateType == UpdateType_OpStyle) {
+        return targetQuery(opCtx, query, collation, endpoints);
+    } else {
+        return targetDoc(opCtx, updateExpr, collation, endpoints);
+    }
+}
+
+Status ChunkManagerTargeter::targetDelete(
+    OperationContext* opCtx,
+    const BatchedDeleteDocument& deleteDoc,
+    std::vector<std::unique_ptr<ShardEndpoint>>* endpoints) const {
+    BSONObj shardKey;
+
+    if (_routingInfo->cm()) {
+        //
+        // Sharded collections have the following further requirements for targeting:
+        //
+        // Limit-1 deletes must be targeted exactly by shard key *or* exact _id
+        //
+
+        // Get the shard key
+        StatusWith<BSONObj> status =
+            _routingInfo->cm()->getShardKeyPattern().extractShardKeyFromQuery(opCtx,
+                                                                              deleteDoc.getQuery());
+
+        // Bad query
+        if (!status.isOK())
+            return status.getStatus();
+
+        shardKey = status.getValue();
+    }
+
+    const BSONObj collation = deleteDoc.isCollationSet() ? deleteDoc.getCollation() : BSONObj();
+
+
+    // Target the shard key or delete query
+    if (!shardKey.isEmpty()) {
+        try {
+            endpoints->push_back(targetShardKey(shardKey, collation, 0));
+            return Status::OK();
+        } catch (const DBException&) {
+            // This delete is potentially not constrained to a single shard
+        }
+    }
+
+    // We failed to target a single shard.
+
+    // Parse delete query.
+    auto qr = stdx::make_unique<QueryRequest>(getNS());
+    qr->setFilter(deleteDoc.getQuery());
+    if (!collation.isEmpty()) {
+        qr->setCollation(collation);
+    }
+    auto cq = CanonicalQuery::canonicalize(opCtx, std::move(qr), ExtensionsCallbackNoop());
+    if (!cq.isOK()) {
+        return Status(cq.getStatus().code(),
+                      str::stream() << "Could not parse delete query " << deleteDoc.getQuery()
+                                    << causedBy(cq.getStatus()));
+    }
+
+    // Single deletes must target a single shard or be exact-ID.
+    if (_routingInfo->cm() && deleteDoc.getLimit() == 1 &&
+        !isExactIdQuery(opCtx, *cq.getValue(), _routingInfo->cm().get())) {
+        return Status(ErrorCodes::ShardKeyNotFound,
+                      str::stream()
+                          << "A single delete on a sharded collection must contain an exact "
+                             "match on _id (and have the collection default collation) or "
+                             "contain the shard key (and have the simple collation). Delete "
+                             "request: "
+                          << deleteDoc.toBSON()
+                          << ", shard key pattern: "
+                          << _routingInfo->cm()->getShardKeyPattern().toString());
+    }
+
+    return targetQuery(opCtx, deleteDoc.getQuery(), collation, endpoints);
+}
+
+Status ChunkManagerTargeter::targetDoc(
+    OperationContext* opCtx,
+    const BSONObj& doc,
+    const BSONObj& collation,
+    std::vector<std::unique_ptr<ShardEndpoint>>* endpoints) const {
+    // NOTE: This is weird and fragile, but it's the way our language works right now -
+    // documents are either A) invalid or B) valid equality queries over themselves.
+    return targetQuery(opCtx, doc, collation, endpoints);
+}
+
+Status ChunkManagerTargeter::targetQuery(
+    OperationContext* opCtx,
+    const BSONObj& query,
+    const BSONObj& collation,
+    std::vector<std::unique_ptr<ShardEndpoint>>* endpoints) const {
+    if (!_routingInfo->primary() && !_routingInfo->cm()) {
+        return {ErrorCodes::NamespaceNotFound,
+                str::stream() << "could not target query in " << getNS().ns()
+                              << "; no metadata found"};
+    }
+
+    std::set<ShardId> shardIds;
+    if (_routingInfo->cm()) {
+        try {
+            _routingInfo->cm()->getShardIdsForQuery(opCtx, query, collation, &shardIds);
+        } catch (const DBException& ex) {
+            return ex.toStatus();
+        }
+    } else {
+        shardIds.insert(_routingInfo->primary()->getId());
+    }
+
+    for (const ShardId& shardId : shardIds) {
+        endpoints->push_back(stdx::make_unique<ShardEndpoint>(
+            shardId,
+            _routingInfo->cm() ? _routingInfo->cm()->getVersion(shardId)
+                               : ChunkVersion::UNSHARDED()));
+    }
+
+    return Status::OK();
+}
+
+std::unique_ptr<ShardEndpoint> ChunkManagerTargeter::targetShardKey(const BSONObj& shardKey,
+                                                                    const BSONObj& collation,
+                                                                    long long estDataSize) const {
+    auto chunk = _routingInfo->cm()->findIntersectingChunk(shardKey, collation);
+
+    // Track autosplit stats for sharded collections
+    // Note: this is only best effort accounting and is not accurate.
+    if (estDataSize > 0) {
+        _stats->chunkSizeDelta[chunk->getMin()] += estDataSize;
+    }
+
+    return stdx::make_unique<ShardEndpoint>(chunk->getShardId(),
+                                            _routingInfo->cm()->getVersion(chunk->getShardId()));
+}
+
+Status ChunkManagerTargeter::targetCollection(
+    std::vector<std::unique_ptr<ShardEndpoint>>* endpoints) const {
+    if (!_routingInfo->primary() && !_routingInfo->cm()) {
+        return {ErrorCodes::NamespaceNotFound,
+                str::stream() << "could not target full range of " << getNS().ns()
+                              << "; metadata not found"};
+    }
+
+    std::set<ShardId> shardIds;
+    if (_routingInfo->cm()) {
+        _routingInfo->cm()->getAllShardIds(&shardIds);
+    } else {
+        shardIds.insert(_routingInfo->primary()->getId());
+    }
+
+    for (const ShardId& shardId : shardIds) {
+        endpoints->push_back(stdx::make_unique<ShardEndpoint>(
+            shardId,
+            _routingInfo->cm() ? _routingInfo->cm()->getVersion(shardId)
+                               : ChunkVersion::UNSHARDED()));
+    }
+
+    return Status::OK();
+}
+
+Status ChunkManagerTargeter::targetAllShards(
+    std::vector<std::unique_ptr<ShardEndpoint>>* endpoints) const {
+    if (!_routingInfo->primary() && !_routingInfo->cm()) {
+        return {ErrorCodes::NamespaceNotFound,
+                str::stream() << "could not target every shard with versions for " << getNS().ns()
+                              << "; metadata not found"};
+    }
+
+    std::vector<ShardId> shardIds;
+    grid.shardRegistry()->getAllShardIds(&shardIds);
+
+    for (const ShardId& shardId : shardIds) {
+        endpoints->push_back(stdx::make_unique<ShardEndpoint>(
+            shardId,
+            _routingInfo->cm() ? _routingInfo->cm()->getVersion(shardId)
+                               : ChunkVersion::UNSHARDED()));
+    }
+
+    return Status::OK();
+}
+
+void ChunkManagerTargeter::noteStaleResponse(const ShardEndpoint& endpoint,
+                                             const BSONObj& staleInfo) {
+    dassert(!_needsTargetingRefresh);
+
+    ChunkVersion remoteShardVersion;
+    if (staleInfo["vWanted"].eoo()) {
+        // If we don't have a vWanted sent, assume the version is higher than our current
+        // version.
+        remoteShardVersion = getShardVersion(*_routingInfo, endpoint.shardName);
+        remoteShardVersion.incMajor();
+    } else {
+        remoteShardVersion = ChunkVersion::fromBSON(staleInfo, "vWanted");
+    }
+
+    ShardVersionMap::iterator it = _remoteShardVersions.find(endpoint.shardName);
+    if (it == _remoteShardVersions.end()) {
+        _remoteShardVersions.insert(std::make_pair(endpoint.shardName, remoteShardVersion));
+    } else {
+        ChunkVersion& previouslyNotedVersion = it->second;
+        if (previouslyNotedVersion.hasEqualEpoch(remoteShardVersion)) {
+            if (previouslyNotedVersion.isOlderThan(remoteShardVersion)) {
+                previouslyNotedVersion = remoteShardVersion;
+            }
+        } else {
+            // Epoch changed midway while applying the batch so set the version to something
+            // unique
+            // and non-existent to force a reload when refreshIsNeeded is called.
+            previouslyNotedVersion = ChunkVersion::IGNORED();
+        }
+    }
+}
+
+void ChunkManagerTargeter::noteCouldNotTarget() {
+    dassert(_remoteShardVersions.empty());
+    _needsTargetingRefresh = true;
+}
+
+Status ChunkManagerTargeter::refreshIfNeeded(OperationContext* opCtx, bool* wasChanged) {
+    bool dummy;
+    if (!wasChanged) {
+        wasChanged = &dummy;
+    }
+
+    *wasChanged = false;
+
+    //
+    // Did we have any stale config or targeting errors at all?
+    //
+
+    if (!_needsTargetingRefresh && _remoteShardVersions.empty()) {
+        return Status::OK();
+    }
+
+    //
+    // Get the latest metadata information from the cache if there were issues
+    //
+
+    auto lastManager = _routingInfo->cm();
+    auto lastPrimary = _routingInfo->primary();
+
+    auto initStatus = init(opCtx);
+    if (!initStatus.isOK()) {
+        return initStatus;
+    }
+
+    // We now have the latest metadata from the cache.
+
+    //
+    // See if and how we need to do a remote refresh.
+    // Either we couldn't target at all, or we have stale versions, but not both.
+    //
+
+    if (_needsTargetingRefresh) {
+        // Reset the field
+        _needsTargetingRefresh = false;
+
+        // If we couldn't target, we might need to refresh if we haven't remotely refreshed
+        // the
+        // metadata since we last got it from the cache.
+
+        bool alreadyRefreshed = wasMetadataRefreshed(
+            lastManager, lastPrimary, _routingInfo->cm(), _routingInfo->primary());
+
+        // If didn't already refresh the targeting information, refresh it
+        if (!alreadyRefreshed) {
+            // To match previous behavior, we just need an incremental refresh here
+            return refreshNow(opCtx);
+        }
+
+        *wasChanged = isMetadataDifferent(
+            lastManager, lastPrimary, _routingInfo->cm(), _routingInfo->primary());
+        return Status::OK();
+    } else if (!_remoteShardVersions.empty()) {
+        // If we got stale shard versions from remote shards, we may need to refresh
+        // NOTE: Not sure yet if this can happen simultaneously with targeting issues
+
+        CompareResult result = compareAllShardVersions(*_routingInfo, _remoteShardVersions);
+
+        // Reset the versions
+        _remoteShardVersions.clear();
+
+        if (result == CompareResult_Unknown || result == CompareResult_LT) {
+            // Our current shard versions aren't all comparable to the old versions, maybe drop
+            return refreshNow(opCtx);
+        }
+
+        *wasChanged = isMetadataDifferent(
+            lastManager, lastPrimary, _routingInfo->cm(), _routingInfo->primary());
+        return Status::OK();
+    }
+
+    MONGO_UNREACHABLE;
+}
+
+Status ChunkManagerTargeter::refreshNow(OperationContext* opCtx) {
+    Grid::get(opCtx)->catalogCache()->onStaleConfigError(std::move(*_routingInfo));
+
+    return init(opCtx);
+}
+
+}  // namespace mongo
diff --git a/src/mongo/s/commands/chunk_manager_targeter.h b/src/mongo/s/commands/chunk_manager_targeter.h
index 36fe46a3fe5..97c0f4c1455 100644
--- a/src/mongo/s/commands/chunk_manager_targeter.h
+++ b/src/mongo/s/commands/chunk_manager_targeter.h
@@ -35,12 +35,12 @@
 #include "mongo/bson/bsonobj_comparator_interface.h"
 #include "mongo/bson/simple_bsonobj_comparator.h"
 #include "mongo/db/namespace_string.h"
+#include "mongo/s/catalog_cache.h"
 #include "mongo/s/ns_targeter.h"
 
 namespace mongo {
 
 class ChunkManager;
-class CollatorInterface;
 class OperationContext;
 class Shard;
 struct ChunkVersion;
@@ -109,21 +109,12 @@ public:
     Status refreshIfNeeded(OperationContext* opCtx, bool* wasChanged);
 
 private:
-    // Different ways we can refresh metadata
-    enum RefreshType {
-        // The version has gone up, but the collection hasn't been dropped
-        RefreshType_RefreshChunkManager,
-        // The collection may have been dropped, so we need to reload the db
-        RefreshType_ReloadDatabase
-    };
-
-    typedef std::map<ShardId, ChunkVersion> ShardVersionMap;
-
+    using ShardVersionMap = std::map<ShardId, ChunkVersion>;
 
     /**
      * Performs an actual refresh from the config server.
      */
-    Status refreshNow(OperationContext* opCtx, RefreshType refreshType);
+    Status refreshNow(OperationContext* opCtx);
 
     /**
      * Returns a vector of ShardEndpoints where a document might need to be placed.
@@ -170,10 +161,8 @@ private:
     // Represents only the view and not really part of the targeter state. This is not owned here.
     TargeterStats* _stats;
 
-    // Zero or one of these are filled at all times
-    // If sharded, _manager, if unsharded, _primary, on error, neither
-    std::shared_ptr<ChunkManager> _manager;
-    std::shared_ptr<Shard> _primary;
+    // The latest loaded routing cache entry
+    boost::optional<CachedCollectionRoutingInfo> _routingInfo;
 
     // Map of shard->remote shard version reported from stale errors
     ShardVersionMap _remoteShardVersions;
diff --git a/src/mongo/s/commands/cluster_aggregate.cpp b/src/mongo/s/commands/cluster_aggregate.cpp
index a6887ea0498..d9a182b2938 100644
--- a/src/mongo/s/commands/cluster_aggregate.cpp
+++ b/src/mongo/s/commands/cluster_aggregate.cpp
@@ -47,7 +47,7 @@
 #include "mongo/db/views/view.h"
 #include "mongo/executor/task_executor_pool.h"
 #include "mongo/rpc/get_status_from_command_result.h"
-#include "mongo/s/chunk_manager.h"
+#include "mongo/s/catalog_cache.h"
 #include "mongo/s/client/shard_connection.h"
 #include "mongo/s/client/shard_registry.h"
 #include "mongo/s/commands/cluster_commands_common.h"
@@ -55,7 +55,6 @@
 #include "mongo/s/grid.h"
 #include "mongo/s/query/cluster_query_knobs.h"
 #include "mongo/s/query/store_possible_cursor.h"
-#include "mongo/s/sharding_raii.h"
 #include "mongo/s/stale_exception.h"
 #include "mongo/util/log.h"
 
@@ -66,20 +65,22 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx,
                                       BSONObj cmdObj,
                                       int options,
                                       BSONObjBuilder* result) {
-    auto scopedShardDbStatus =
-        ScopedShardDatabase::getExisting(opCtx, namespaces.executionNss.db());
-    if (!scopedShardDbStatus.isOK()) {
-        appendEmptyResultSet(
-            *result, scopedShardDbStatus.getStatus(), namespaces.requestedNss.ns());
-        return Status::OK();
-    }
-
     auto request = AggregationRequest::parseFromBSON(namespaces.executionNss, cmdObj);
     if (!request.isOK()) {
         return request.getStatus();
     }
 
-    const auto conf = scopedShardDbStatus.getValue().db();
+    auto const catalogCache = Grid::get(opCtx)->catalogCache();
+
+    auto executionNsRoutingInfoStatus =
+        catalogCache->getCollectionRoutingInfo(opCtx, namespaces.executionNss);
+    if (!executionNsRoutingInfoStatus.isOK()) {
+        appendEmptyResultSet(
+            *result, executionNsRoutingInfoStatus.getStatus(), namespaces.requestedNss.ns());
+        return Status::OK();
+    }
+
+    const auto& executionNsRoutingInfo = executionNsRoutingInfoStatus.getValue();
 
     // Determine the appropriate collation and 'resolve' involved namespaces to make the
     // ExpressionContext.
@@ -91,16 +92,20 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx,
     // command on an unsharded collection.
     StringMap<ExpressionContext::ResolvedNamespace> resolvedNamespaces;
     LiteParsedPipeline liteParsedPipeline(request.getValue());
-    for (auto&& ns : liteParsedPipeline.getInvolvedNamespaces()) {
-        uassert(28769, str::stream() << ns.ns() << " cannot be sharded", !conf->isSharded(ns.ns()));
-        resolvedNamespaces[ns.coll()] = {ns, std::vector<BSONObj>{}};
+    for (auto&& nss : liteParsedPipeline.getInvolvedNamespaces()) {
+        const auto resolvedNsRoutingInfo =
+            uassertStatusOK(catalogCache->getCollectionRoutingInfo(opCtx, nss));
+        uassert(
+            28769, str::stream() << nss.ns() << " cannot be sharded", !resolvedNsRoutingInfo.cm());
+        resolvedNamespaces.try_emplace(nss.coll(), nss, std::vector<BSONObj>{});
     }
 
-    if (!conf->isSharded(namespaces.executionNss.ns())) {
-        return aggPassthrough(opCtx, namespaces, conf, cmdObj, result, options);
+    if (!executionNsRoutingInfo.cm()) {
+        return aggPassthrough(
+            opCtx, namespaces, executionNsRoutingInfo.primary()->getId(), cmdObj, result, options);
     }
 
-    auto chunkMgr = conf->getChunkManager(opCtx, namespaces.executionNss.ns());
+    const auto chunkMgr = executionNsRoutingInfo.cm();
 
     std::unique_ptr<CollatorInterface> collation;
     if (!request.getValue().getCollation().isEmpty()) {
@@ -260,9 +265,10 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx,
     // Run merging command on random shard, unless a stage needs the primary shard. Need to use
     // ShardConnection so that the merging mongod is sent the config servers on connection init.
     auto& prng = opCtx->getClient()->getPrng();
-    const auto& mergingShardId =
+    const auto mergingShardId =
         (needPrimaryShardMerger || internalQueryAlwaysMergeOnPrimaryShard.load())
-        ? conf->getPrimaryId()
+        ? uassertStatusOK(catalogCache->getDatabase(opCtx, namespaces.executionNss.db()))
+              .primaryId()
         : shardResults[prng.nextInt32(shardResults.size())].shardTargetId;
     const auto mergingShard =
         uassertStatusOK(Grid::get(opCtx)->shardRegistry()->getShard(opCtx, mergingShardId));
@@ -426,12 +432,12 @@ BSONObj ClusterAggregate::aggRunCommand(OperationContext* opCtx,
 
 Status ClusterAggregate::aggPassthrough(OperationContext* opCtx,
                                         const Namespaces& namespaces,
-                                        DBConfig* conf,
+                                        const ShardId& shardId,
                                         BSONObj cmdObj,
                                         BSONObjBuilder* out,
                                         int queryOptions) {
     // Temporary hack. See comment on declaration for details.
-    auto shardStatus = Grid::get(opCtx)->shardRegistry()->getShard(opCtx, conf->getPrimaryId());
+    auto shardStatus = Grid::get(opCtx)->shardRegistry()->getShard(opCtx, shardId);
     if (!shardStatus.isOK()) {
         return shardStatus.getStatus();
     }
diff --git a/src/mongo/s/commands/cluster_aggregate.h b/src/mongo/s/commands/cluster_aggregate.h
index b0fdd5d7375..d8e29744766 100644
--- a/src/mongo/s/commands/cluster_aggregate.h
+++ b/src/mongo/s/commands/cluster_aggregate.h
@@ -37,11 +37,11 @@
 #include "mongo/db/pipeline/document_source.h"
 #include "mongo/db/pipeline/document_source_merge_cursors.h"
 #include "mongo/s/commands/strategy.h"
-#include "mongo/s/config.h"
 
 namespace mongo {
 
 class OperationContext;
+class ShardId;
 
 /**
  * Methods for running aggregation across a sharded cluster.
@@ -90,7 +90,7 @@ private:
 
     static Status aggPassthrough(OperationContext* opCtx,
                                  const Namespaces& namespaces,
-                                 DBConfig* conf,
+                                 const ShardId& shardId,
                                  BSONObj cmd,
                                  BSONObjBuilder* result,
                                  int queryOptions);
diff --git a/src/mongo/s/commands/cluster_commands_common.cpp b/src/mongo/s/commands/cluster_commands_common.cpp
index 3ad8c373b9a..225cef2b6ef 100644
--- a/src/mongo/s/commands/cluster_commands_common.cpp
+++ b/src/mongo/s/commands/cluster_commands_common.cpp
@@ -40,7 +40,6 @@
 #include "mongo/s/client/shard_connection.h"
 #include "mongo/s/client/version_manager.h"
 #include "mongo/s/grid.h"
-#include "mongo/s/sharding_raii.h"
 #include "mongo/s/stale_exception.h"
 #include "mongo/util/log.h"
 
@@ -54,17 +53,21 @@ namespace {
 bool forceRemoteCheckShardVersionCB(OperationContext* opCtx, const string& ns) {
     const NamespaceString nss(ns);
 
+    if (!nss.isValid()) {
+        return false;
+    }
+
     // This will force the database catalog entry to be reloaded
-    Grid::get(opCtx)->catalogCache()->invalidate(nss.db());
+    Grid::get(opCtx)->catalogCache()->invalidateShardedCollection(nss);
 
-    auto scopedCMStatus = ScopedChunkManager::get(opCtx, nss);
-    if (!scopedCMStatus.isOK()) {
+    auto routingInfoStatus = Grid::get(opCtx)->catalogCache()->getCollectionRoutingInfo(opCtx, nss);
+    if (!routingInfoStatus.isOK()) {
         return false;
     }
 
-    const auto& scopedCM = scopedCMStatus.getValue();
+    auto& routingInfo = routingInfoStatus.getValue();
 
-    return scopedCM.cm() != nullptr;
+    return routingInfo.cm() != nullptr;
 }
 
 }  // namespace
@@ -261,4 +264,36 @@ std::vector<NamespaceString> getAllShardedCollectionsForDb(OperationContext* opC
     return collectionsToReturn;
 }
 
+CachedCollectionRoutingInfo getShardedCollection(OperationContext* opCtx,
+                                                 const NamespaceString& nss) {
+    auto routingInfo =
+        uassertStatusOK(Grid::get(opCtx)->catalogCache()->getCollectionRoutingInfo(opCtx, nss));
+    uassert(ErrorCodes::NamespaceNotSharded,
+            str::stream() << "Collection " << nss.ns() << " is not sharded.",
+            routingInfo.cm());
+
+    return routingInfo;
+}
+
+StatusWith<CachedDatabaseInfo> createShardDatabase(OperationContext* opCtx, StringData dbName) {
+    auto dbStatus = Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, dbName);
+    if (dbStatus == ErrorCodes::NamespaceNotFound) {
+        auto createDbStatus =
+            Grid::get(opCtx)->catalogClient(opCtx)->createDatabase(opCtx, dbName.toString());
+        if (createDbStatus.isOK() || createDbStatus == ErrorCodes::NamespaceExists) {
+            dbStatus = Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, dbName);
+        } else {
+            dbStatus = createDbStatus;
+        }
+    }
+
+    if (dbStatus.isOK()) {
+        return dbStatus;
+    }
+
+    return {dbStatus.getStatus().code(),
+            str::stream() << "Database " << dbName << " not found due to "
+                          << dbStatus.getStatus().reason()};
+}
+
 }  // namespace mongo
diff --git a/src/mongo/s/commands/cluster_commands_common.h b/src/mongo/s/commands/cluster_commands_common.h
index 7d6465bc400..960eb03e73b 100644
--- a/src/mongo/s/commands/cluster_commands_common.h
+++ b/src/mongo/s/commands/cluster_commands_common.h
@@ -39,6 +39,8 @@
 namespace mongo {
 
 class AScopedConnection;
+class CachedCollectionRoutingInfo;
+class CachedDatabaseInfo;
 class DBClientBase;
 class DBClientCursor;
 class OperationContext;
@@ -140,4 +142,17 @@ bool appendEmptyResultSet(BSONObjBuilder& result, Status status, const std::stri
 std::vector<NamespaceString> getAllShardedCollectionsForDb(OperationContext* opCtx,
                                                            StringData dbName);
 
+/**
+ * Abstracts the common pattern of refreshing a collection and checking if it is sharded used across
+ * multiple commands.
+ */
+CachedCollectionRoutingInfo getShardedCollection(OperationContext* opCtx,
+                                                 const NamespaceString& nss);
+
+/**
+ * If the specified database exists already, loads it in the cache (if not already there) and
+ * returns it. Otherwise, if it does not exist, this call will implicitly create it as non-sharded.
+ */
+StatusWith<CachedDatabaseInfo> createShardDatabase(OperationContext* opCtx, StringData dbName);
+
 }  // namespace mongo
diff --git a/src/mongo/s/commands/cluster_count_cmd.cpp b/src/mongo/s/commands/cluster_count_cmd.cpp
index 2fcf11086b9..45c46c26185 100644
--- a/src/mongo/s/commands/cluster_count_cmd.cpp
+++ b/src/mongo/s/commands/cluster_count_cmd.cpp
@@ -42,73 +42,42 @@
 #include "mongo/util/timer.h"
 
 namespace mongo {
-
-using std::string;
-using std::vector;
-
 namespace {
 
-long long applySkipLimit(long long num, const BSONObj& cmd) {
-    BSONElement s = cmd["skip"];
-    BSONElement l = cmd["limit"];
-
-    if (s.isNumber()) {
-        num = num - s.numberLong();
-        if (num < 0) {
-            num = 0;
-        }
-    }
-
-    if (l.isNumber()) {
-        long long limit = l.numberLong();
-        if (limit < 0) {
-            limit = -limit;
-        }
-
-        // 0 limit means no limit
-        if (limit < num && limit != 0) {
-            num = limit;
-        }
-    }
-
-    return num;
-}
-
-
 class ClusterCountCmd : public Command {
 public:
     ClusterCountCmd() : Command("count", false) {}
 
-    virtual bool slaveOk() const {
+    bool slaveOk() const override {
         return true;
     }
 
-    virtual bool adminOnly() const {
+    bool adminOnly() const override {
         return false;
     }
 
-
-    virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
+    bool supportsWriteConcern(const BSONObj& cmd) const override {
         return false;
     }
 
-    virtual void addRequiredPrivileges(const std::string& dbname,
-                                       const BSONObj& cmdObj,
-                                       std::vector<Privilege>* out) {
+    void addRequiredPrivileges(const std::string& dbname,
+                               const BSONObj& cmdObj,
+                               std::vector<Privilege>* out) override {
         ActionSet actions;
         actions.addAction(ActionType::find);
         out->push_back(Privilege(parseResourcePattern(dbname, cmdObj), actions));
     }
 
-    virtual bool run(OperationContext* opCtx,
-                     const std::string& dbname,
-                     BSONObj& cmdObj,
-                     int options,
-                     std::string& errmsg,
-                     BSONObjBuilder& result) {
+    bool run(OperationContext* opCtx,
+             const std::string& dbname,
+             BSONObj& cmdObj,
+             int options,
+             std::string& errmsg,
+             BSONObjBuilder& result) override {
         const NamespaceString nss(parseNs(dbname, cmdObj));
-        uassert(
-            ErrorCodes::InvalidNamespace, "count command requires valid namespace", nss.isValid());
+        uassert(ErrorCodes::InvalidNamespace,
+                str::stream() << "Invalid namespace specified '" << nss.ns() << "'",
+                nss.isValid());
 
         long long skip = 0;
 
@@ -167,7 +136,7 @@ public:
             }
         }
 
-        vector<Strategy::CommandResult> countResult;
+        std::vector<Strategy::CommandResult> countResult;
         Strategy::commandOp(opCtx,
                             dbname,
                             countCmdBuilder.done(),
@@ -214,20 +183,19 @@ public:
         long long total = 0;
         BSONObjBuilder shardSubTotal(result.subobjStart("shards"));
 
-        for (vector<Strategy::CommandResult>::const_iterator iter = countResult.begin();
-             iter != countResult.end();
-             ++iter) {
-            const ShardId& shardName = iter->shardTargetId;
+        for (const auto& resultEntry : countResult) {
+            const ShardId& shardName = resultEntry.shardTargetId;
+            const auto resultBSON = resultEntry.result;
 
-            if (iter->result["ok"].trueValue()) {
-                long long shardCount = iter->result["n"].numberLong();
+            if (resultBSON["ok"].trueValue()) {
+                long long shardCount = resultBSON["n"].numberLong();
 
                 shardSubTotal.appendNumber(shardName.toString(), shardCount);
                 total += shardCount;
             } else {
                 shardSubTotal.doneFast();
                 errmsg = "failed on : " + shardName.toString();
-                result.append("cause", iter->result);
+                result.append("cause", resultBSON);
 
                 // Add "code" to the top-level response, if the failure of the sharded command
                 // can be accounted to a single error
@@ -247,17 +215,16 @@ public:
         return true;
     }
 
-    virtual Status explain(OperationContext* opCtx,
-                           const std::string& dbname,
-                           const BSONObj& cmdObj,
-                           ExplainCommon::Verbosity verbosity,
-                           const rpc::ServerSelectionMetadata& serverSelectionMetadata,
-                           BSONObjBuilder* out) const {
+    Status explain(OperationContext* opCtx,
+                   const std::string& dbname,
+                   const BSONObj& cmdObj,
+                   ExplainCommon::Verbosity verbosity,
+                   const rpc::ServerSelectionMetadata& serverSelectionMetadata,
+                   BSONObjBuilder* out) const override {
         const NamespaceString nss(parseNs(dbname, cmdObj));
-        if (!nss.isValid()) {
-            return Status{ErrorCodes::InvalidNamespace,
-                          str::stream() << "Invalid collection name: " << nss.ns()};
-        }
+        uassert(ErrorCodes::InvalidNamespace,
+                str::stream() << "Invalid namespace specified '" << nss.ns() << "'",
+                nss.isValid());
 
         // Extract the targeting query.
         BSONObj targetingQuery;
@@ -284,7 +251,7 @@ public:
         // We will time how long it takes to run the commands on the shards
         Timer timer;
 
-        vector<Strategy::CommandResult> shardResults;
+        std::vector<Strategy::CommandResult> shardResults;
         Strategy::commandOp(opCtx,
                             dbname,
                             explainCmdBob.obj(),
@@ -329,6 +296,33 @@ public:
             opCtx, shardResults, mongosStageName, millisElapsed, out);
     }
 
+private:
+    static long long applySkipLimit(long long num, const BSONObj& cmd) {
+        BSONElement s = cmd["skip"];
+        BSONElement l = cmd["limit"];
+
+        if (s.isNumber()) {
+            num = num - s.numberLong();
+            if (num < 0) {
+                num = 0;
+            }
+        }
+
+        if (l.isNumber()) {
+            long long limit = l.numberLong();
+            if (limit < 0) {
+                limit = -limit;
+            }
+
+            // 0 limit means no limit
+            if (limit < num && limit != 0) {
+                num = limit;
+            }
+        }
+
+        return num;
+    }
+
 } clusterCountCmd;
 
 }  // namespace
diff --git a/src/mongo/s/commands/cluster_drop_cmd.cpp b/src/mongo/s/commands/cluster_drop_cmd.cpp
index 2c44a5a1dbc..583c8902cbf 100644
--- a/src/mongo/s/commands/cluster_drop_cmd.cpp
+++ b/src/mongo/s/commands/cluster_drop_cmd.cpp
@@ -41,7 +41,6 @@
 #include "mongo/s/commands/cluster_commands_common.h"
 #include "mongo/s/commands/sharded_command_processing.h"
 #include "mongo/s/grid.h"
-#include "mongo/s/sharding_raii.h"
 #include "mongo/s/stale_exception.h"
 #include "mongo/util/log.h"
 
@@ -80,20 +79,20 @@ public:
              BSONObjBuilder& result) override {
         const NamespaceString nss(parseNsCollectionRequired(dbname, cmdObj));
 
-        auto scopedDbStatus = ScopedShardDatabase::getExisting(opCtx, dbname);
-        if (scopedDbStatus == ErrorCodes::NamespaceNotFound) {
+        auto const catalogCache = Grid::get(opCtx)->catalogCache();
+
+        auto routingInfoStatus = catalogCache->getCollectionRoutingInfo(opCtx, nss);
+        if (routingInfoStatus == ErrorCodes::NamespaceNotFound) {
             return true;
         }
 
-        uassertStatusOK(scopedDbStatus.getStatus());
-
-        auto const db = scopedDbStatus.getValue().db();
+        auto routingInfo = uassertStatusOK(std::move(routingInfoStatus));
 
-        if (!db->isSharded(nss.ns())) {
-            _dropUnshardedCollectionFromShard(opCtx, db->getPrimaryId(), nss, &result);
+        if (!routingInfo.cm()) {
+            _dropUnshardedCollectionFromShard(opCtx, routingInfo.primaryId(), nss, &result);
         } else {
             uassertStatusOK(Grid::get(opCtx)->catalogClient(opCtx)->dropCollection(opCtx, nss));
-            db->markNSNotSharded(nss.ns());
+            catalogCache->invalidateShardedCollection(nss);
         }
 
         return true;
diff --git a/src/mongo/s/commands/cluster_drop_database_cmd.cpp b/src/mongo/s/commands/cluster_drop_database_cmd.cpp
index f86cf073273..178fc5f36bc 100644
--- a/src/mongo/s/commands/cluster_drop_database_cmd.cpp
+++ b/src/mongo/s/commands/cluster_drop_database_cmd.cpp
@@ -40,9 +40,7 @@
 #include "mongo/s/client/shard_registry.h"
 #include "mongo/s/commands/cluster_commands_common.h"
 #include "mongo/s/commands/sharded_command_processing.h"
-#include "mongo/s/config.h"
 #include "mongo/s/grid.h"
-#include "mongo/s/sharding_raii.h"
 #include "mongo/util/log.h"
 
 namespace mongo {
@@ -93,17 +91,19 @@ public:
         auto scopedDistLock = uassertStatusOK(catalogClient->getDistLockManager()->lock(
             opCtx, dbname, "dropDatabase", DistLockManager::kDefaultLockTimeout));
 
+        auto const catalogCache = Grid::get(opCtx)->catalogCache();
+
         // Refresh the database metadata so it kicks off a full reload
-        Grid::get(opCtx)->catalogCache()->invalidate(dbname);
+        catalogCache->purgeDatabase(dbname);
 
-        auto scopedDbStatus = ScopedShardDatabase::getExisting(opCtx, dbname);
+        auto dbInfoStatus = catalogCache->getDatabase(opCtx, dbname);
 
-        if (scopedDbStatus == ErrorCodes::NamespaceNotFound) {
+        if (dbInfoStatus == ErrorCodes::NamespaceNotFound) {
             result.append("info", "database does not exist");
             return true;
         }
 
-        uassertStatusOK(scopedDbStatus.getStatus());
+        uassertStatusOK(dbInfoStatus.getStatus());
 
         catalogClient->logChange(opCtx,
                                  "dropDatabase.start",
@@ -111,16 +111,15 @@ public:
                                  BSONObj(),
                                  ShardingCatalogClient::kMajorityWriteConcern);
 
-        auto const db = scopedDbStatus.getValue().db();
+        auto& dbInfo = dbInfoStatus.getValue();
 
         // Drop the database's collections from metadata
         for (const auto& nss : getAllShardedCollectionsForDb(opCtx, dbname)) {
             uassertStatusOK(catalogClient->dropCollection(opCtx, nss));
-            db->markNSNotSharded(nss.ns());
         }
 
         // Drop the database from the primary shard first
-        _dropDatabaseFromShard(opCtx, db->getPrimaryId(), dbname);
+        _dropDatabaseFromShard(opCtx, dbInfo.primaryId(), dbname);
 
         // Drop the database from each of the remaining shards
         {
@@ -146,7 +145,7 @@ public:
         }
 
         // Invalidate the database so the next access will do a full reload
-        Grid::get(opCtx)->catalogCache()->invalidate(dbname);
+        catalogCache->purgeDatabase(dbname);
 
         catalogClient->logChange(
             opCtx, "dropDatabase", dbname, BSONObj(), ShardingCatalogClient::kMajorityWriteConcern);
diff --git a/src/mongo/s/commands/cluster_enable_sharding_cmd.cpp b/src/mongo/s/commands/cluster_enable_sharding_cmd.cpp
index 1db7ea7ef03..64537920cb7 100644
--- a/src/mongo/s/commands/cluster_enable_sharding_cmd.cpp
+++ b/src/mongo/s/commands/cluster_enable_sharding_cmd.cpp
@@ -41,7 +41,6 @@
 #include "mongo/db/commands.h"
 #include "mongo/s/catalog/sharding_catalog_client.h"
 #include "mongo/s/catalog_cache.h"
-#include "mongo/s/config.h"
 #include "mongo/s/grid.h"
 #include "mongo/util/log.h"
 
@@ -109,7 +108,7 @@ public:
         audit::logEnableSharding(Client::getCurrent(), dbname);
 
         // Make sure to force update of any stale metadata
-        Grid::get(opCtx)->catalogCache()->invalidate(dbname);
+        Grid::get(opCtx)->catalogCache()->purgeDatabase(dbname);
 
         return true;
     }
diff --git a/src/mongo/s/commands/cluster_find_and_modify_cmd.cpp b/src/mongo/s/commands/cluster_find_and_modify_cmd.cpp
index 578968205af..feae1fab5e2 100644
--- a/src/mongo/s/commands/cluster_find_and_modify_cmd.cpp
+++ b/src/mongo/s/commands/cluster_find_and_modify_cmd.cpp
@@ -40,7 +40,6 @@
 #include "mongo/db/query/collation/collator_factory_interface.h"
 #include "mongo/s/balancer_configuration.h"
 #include "mongo/s/catalog_cache.h"
-#include "mongo/s/chunk_manager.h"
 #include "mongo/s/client/shard_connection.h"
 #include "mongo/s/client/shard_registry.h"
 #include "mongo/s/commands/cluster_explain.h"
@@ -48,7 +47,6 @@
 #include "mongo/s/commands/sharded_command_processing.h"
 #include "mongo/s/commands/strategy.h"
 #include "mongo/s/grid.h"
-#include "mongo/s/sharding_raii.h"
 #include "mongo/s/stale_exception.h"
 #include "mongo/util/timer.h"
 
@@ -63,48 +61,42 @@ class FindAndModifyCmd : public Command {
 public:
     FindAndModifyCmd() : Command("findAndModify", false, "findandmodify") {}
 
-    virtual bool slaveOk() const {
+    bool slaveOk() const override {
         return true;
     }
 
-    virtual bool adminOnly() const {
+    bool adminOnly() const override {
         return false;
     }
 
-
-    virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
+    bool supportsWriteConcern(const BSONObj& cmd) const override {
         return true;
     }
 
-    virtual void addRequiredPrivileges(const std::string& dbname,
-                                       const BSONObj& cmdObj,
-                                       std::vector<Privilege>* out) {
+    void addRequiredPrivileges(const std::string& dbname,
+                               const BSONObj& cmdObj,
+                               std::vector<Privilege>* out) override {
         find_and_modify::addPrivilegesRequiredForFindAndModify(this, dbname, cmdObj, out);
     }
 
-    virtual Status explain(OperationContext* opCtx,
-                           const std::string& dbName,
-                           const BSONObj& cmdObj,
-                           ExplainCommon::Verbosity verbosity,
-                           const rpc::ServerSelectionMetadata& serverSelectionMetadata,
-                           BSONObjBuilder* out) const {
-        const NamespaceString nss = parseNsCollectionRequired(dbName, cmdObj);
+    Status explain(OperationContext* opCtx,
+                   const std::string& dbName,
+                   const BSONObj& cmdObj,
+                   ExplainCommon::Verbosity verbosity,
+                   const rpc::ServerSelectionMetadata& serverSelectionMetadata,
+                   BSONObjBuilder* out) const override {
+        const NamespaceString nss(parseNsCollectionRequired(dbName, cmdObj));
 
-        auto scopedDB = uassertStatusOK(ScopedShardDatabase::getExisting(opCtx, dbName));
-        const auto conf = scopedDB.db();
+        auto routingInfo =
+            uassertStatusOK(Grid::get(opCtx)->catalogCache()->getCollectionRoutingInfo(opCtx, nss));
 
         shared_ptr<ChunkManager> chunkMgr;
         shared_ptr<Shard> shard;
 
-        if (!conf->isSharded(nss.ns())) {
-            auto shardStatus =
-                Grid::get(opCtx)->shardRegistry()->getShard(opCtx, conf->getPrimaryId());
-            if (!shardStatus.isOK()) {
-                return shardStatus.getStatus();
-            }
-            shard = shardStatus.getValue();
+        if (!routingInfo.cm()) {
+            shard = routingInfo.primary();
         } else {
-            chunkMgr = _getChunkManager(opCtx, conf, nss);
+            chunkMgr = routingInfo.cm();
 
             const BSONObj query = cmdObj.getObjectField("query");
 
@@ -118,7 +110,7 @@ public:
                 return collationElementStatus;
             }
 
-            StatusWith<BSONObj> status = _getShardKey(opCtx, chunkMgr, query);
+            StatusWith<BSONObj> status = _getShardKey(opCtx, *chunkMgr, query);
             if (!status.isOK()) {
                 return status.getStatus();
             }
@@ -131,6 +123,7 @@ public:
             if (!shardStatus.isOK()) {
                 return shardStatus.getStatus();
             }
+
             shard = shardStatus.getValue();
         }
 
@@ -143,7 +136,7 @@ public:
         Timer timer;
 
         BSONObjBuilder result;
-        bool ok = _runCommand(opCtx, conf, chunkMgr, shard->getId(), nss, explainCmd.obj(), result);
+        bool ok = _runCommand(opCtx, chunkMgr, shard->getId(), nss, explainCmd.obj(), result);
         long long millisElapsed = timer.millis();
 
         if (!ok) {
@@ -164,24 +157,23 @@ public:
             opCtx, shardResults, ClusterExplain::kSingleShard, millisElapsed, out);
     }
 
-    virtual bool run(OperationContext* opCtx,
-                     const std::string& dbName,
-                     BSONObj& cmdObj,
-                     int options,
-                     std::string& errmsg,
-                     BSONObjBuilder& result) {
+    bool run(OperationContext* opCtx,
+             const std::string& dbName,
+             BSONObj& cmdObj,
+             int options,
+             std::string& errmsg,
+             BSONObjBuilder& result) override {
         const NamespaceString nss = parseNsCollectionRequired(dbName, cmdObj);
 
         // findAndModify should only be creating database if upsert is true, but this would require
         // that the parsing be pulled into this function.
-        auto scopedDb = uassertStatusOK(ScopedShardDatabase::getOrCreate(opCtx, dbName));
-        const auto conf = scopedDb.db();
-
-        if (!conf->isSharded(nss.ns())) {
-            return _runCommand(opCtx, conf, nullptr, conf->getPrimaryId(), nss, cmdObj, result);
+        auto routingInfo =
+            uassertStatusOK(Grid::get(opCtx)->catalogCache()->getCollectionRoutingInfo(opCtx, nss));
+        if (!routingInfo.cm()) {
+            return _runCommand(opCtx, nullptr, routingInfo.primaryId(), nss, cmdObj, result);
         }
 
-        shared_ptr<ChunkManager> chunkMgr = _getChunkManager(opCtx, conf, nss);
+        const auto chunkMgr = routingInfo.cm();
 
         const BSONObj query = cmdObj.getObjectField("query");
 
@@ -195,17 +187,11 @@ public:
             return appendCommandStatus(result, collationElementStatus);
         }
 
-        StatusWith<BSONObj> status = _getShardKey(opCtx, chunkMgr, query);
-        if (!status.isOK()) {
-            // Bad query
-            return appendCommandStatus(result, status.getStatus());
-        }
+        BSONObj shardKey = uassertStatusOK(_getShardKey(opCtx, *chunkMgr, query));
 
-        BSONObj shardKey = status.getValue();
         auto chunk = chunkMgr->findIntersectingChunk(shardKey, collation);
 
-        const bool ok =
-            _runCommand(opCtx, conf, chunkMgr, chunk->getShardId(), nss, cmdObj, result);
+        const bool ok = _runCommand(opCtx, chunkMgr, chunk->getShardId(), nss, cmdObj, result);
         if (ok) {
             updateChunkWriteStatsAndSplitIfNeeded(
                 opCtx, chunkMgr.get(), chunk.get(), cmdObj.getObjectField("update").objsize());
@@ -215,21 +201,12 @@ public:
     }
 
 private:
-    shared_ptr<ChunkManager> _getChunkManager(OperationContext* opCtx,
-                                              DBConfig* conf,
-                                              const NamespaceString& nss) const {
-        shared_ptr<ChunkManager> chunkMgr = conf->getChunkManager(opCtx, nss.ns());
-        massert(13002, "shard internal error chunk manager should never be null", chunkMgr);
-
-        return chunkMgr;
-    }
-
-    StatusWith<BSONObj> _getShardKey(OperationContext* opCtx,
-                                     shared_ptr<ChunkManager> chunkMgr,
-                                     const BSONObj& query) const {
+    static StatusWith<BSONObj> _getShardKey(OperationContext* opCtx,
+                                            const ChunkManager& chunkMgr,
+                                            const BSONObj& query) {
         // Verify that the query has an equality predicate using the shard key
         StatusWith<BSONObj> status =
-            chunkMgr->getShardKeyPattern().extractShardKeyFromQuery(opCtx, query);
+            chunkMgr.getShardKeyPattern().extractShardKeyFromQuery(opCtx, query);
 
         if (!status.isOK()) {
             return status;
@@ -245,20 +222,19 @@ private:
         return shardKey;
     }
 
-    bool _runCommand(OperationContext* opCtx,
-                     DBConfig* conf,
-                     shared_ptr<ChunkManager> chunkManager,
-                     const ShardId& shardId,
-                     const NamespaceString& nss,
-                     const BSONObj& cmdObj,
-                     BSONObjBuilder& result) const {
+    static bool _runCommand(OperationContext* opCtx,
+                            shared_ptr<ChunkManager> chunkManager,
+                            const ShardId& shardId,
+                            const NamespaceString& nss,
+                            const BSONObj& cmdObj,
+                            BSONObjBuilder& result) {
         BSONObj res;
 
         const auto shard =
             uassertStatusOK(Grid::get(opCtx)->shardRegistry()->getShard(opCtx, shardId));
 
         ShardConnection conn(shard->getConnString(), nss.ns(), chunkManager);
-        bool ok = conn->runCommand(conf->name(), cmdObj, res);
+        bool ok = conn->runCommand(nss.db().toString(), cmdObj, res);
         conn.done();
 
         // ErrorCodes::RecvStaleConfig is the code for RecvStaleConfigException.
diff --git a/src/mongo/s/commands/cluster_flush_router_config_cmd.cpp b/src/mongo/s/commands/cluster_flush_router_config_cmd.cpp
index 35150ac3aca..8931166863c 100644
--- a/src/mongo/s/commands/cluster_flush_router_config_cmd.cpp
+++ b/src/mongo/s/commands/cluster_flush_router_config_cmd.cpp
@@ -70,7 +70,7 @@ public:
                      int options,
                      std::string& errmsg,
                      BSONObjBuilder& result) {
-        Grid::get(opCtx)->catalogCache()->invalidateAll();
+        Grid::get(opCtx)->catalogCache()->purgeAllDatabases();
 
         result.appendBool("flushed", true);
         return true;
diff --git a/src/mongo/s/commands/cluster_get_shard_version_cmd.cpp b/src/mongo/s/commands/cluster_get_shard_version_cmd.cpp
index 00e104c1e45..2bb23453bba 100644
--- a/src/mongo/s/commands/cluster_get_shard_version_cmd.cpp
+++ b/src/mongo/s/commands/cluster_get_shard_version_cmd.cpp
@@ -35,8 +35,9 @@
 #include "mongo/db/auth/authorization_manager.h"
 #include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/commands.h"
+#include "mongo/s/catalog_cache.h"
+#include "mongo/s/commands/cluster_commands_common.h"
 #include "mongo/s/grid.h"
-#include "mongo/s/sharding_raii.h"
 #include "mongo/util/log.h"
 
 namespace mongo {
@@ -86,13 +87,10 @@ public:
              BSONObjBuilder& result) override {
         const NamespaceString nss(parseNs(dbname, cmdObj));
 
-        auto scopedDb = uassertStatusOK(ScopedShardDatabase::getExisting(opCtx, nss.db()));
-        auto config = scopedDb.db();
+        auto routingInfo = getShardedCollection(opCtx, nss);
+        const auto cm = routingInfo.cm();
 
-        auto cm = config->getChunkManagerIfExists(opCtx, nss.ns());
-        uassert(ErrorCodes::NamespaceNotSharded, "ns [" + nss.ns() + " is not sharded.", cm);
-
-        for (const auto& cmEntry : cm->getChunkMap()) {
+        for (const auto& cmEntry : cm->chunkMap()) {
             log() << redact(cmEntry.second->toString());
         }
 
diff --git a/src/mongo/s/commands/cluster_map_reduce_cmd.cpp b/src/mongo/s/commands/cluster_map_reduce_cmd.cpp
index 088b8d6d4d1..b155d322b3e 100644
--- a/src/mongo/s/commands/cluster_map_reduce_cmd.cpp
+++ b/src/mongo/s/commands/cluster_map_reduce_cmd.cpp
@@ -45,27 +45,17 @@
 #include "mongo/s/catalog/dist_lock_manager.h"
 #include "mongo/s/catalog/sharding_catalog_client.h"
 #include "mongo/s/catalog_cache.h"
-#include "mongo/s/chunk_manager.h"
 #include "mongo/s/client/shard_connection.h"
 #include "mongo/s/client/shard_registry.h"
 #include "mongo/s/commands/cluster_commands_common.h"
 #include "mongo/s/commands/cluster_write.h"
 #include "mongo/s/commands/sharded_command_processing.h"
 #include "mongo/s/commands/strategy.h"
-#include "mongo/s/config.h"
 #include "mongo/s/grid.h"
-#include "mongo/s/sharding_raii.h"
 #include "mongo/stdx/chrono.h"
 #include "mongo/util/log.h"
 
 namespace mongo {
-
-using std::shared_ptr;
-using std::map;
-using std::set;
-using std::string;
-using std::vector;
-
 namespace {
 
 AtomicUInt32 JOB_NUMBER;
@@ -75,7 +65,7 @@ const Milliseconds kNoDistLockTimeout(-1);
 /**
  * Generates a unique name for the temporary M/R output collection.
  */
-string getTmpName(StringData coll) {
+std::string getTmpName(StringData coll) {
     return str::stream() << "tmp.mrs." << coll << "_" << time(0) << "_"
                          << JOB_NUMBER.fetchAndAdd(1);
 }
@@ -85,14 +75,14 @@ string getTmpName(StringData coll) {
  * be sent to the shards as part of the first phase of map/reduce.
  */
 BSONObj fixForShards(const BSONObj& orig,
-                     const string& output,
-                     string& badShardedField,
+                     const std::string& output,
+                     std::string& badShardedField,
                      int maxChunkSizeBytes) {
     BSONObjBuilder b;
     BSONObjIterator i(orig);
     while (i.more()) {
         BSONElement e = i.next();
-        const string fn = e.fieldName();
+        const std::string fn = e.fieldName();
 
         if (fn == bypassDocumentValidationCommandOption() || fn == "map" || fn == "mapreduce" ||
             fn == "mapReduce" || fn == "mapparams" || fn == "reduce" || fn == "query" ||
@@ -160,47 +150,49 @@ class MRCmd : public Command {
 public:
     MRCmd() : Command("mapReduce", false, "mapreduce") {}
 
-    virtual bool slaveOk() const {
+    bool slaveOk() const override {
         return true;
     }
 
-    virtual bool adminOnly() const {
+    bool adminOnly() const override {
         return false;
     }
 
-    virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
+    std::string parseNs(const std::string& dbname, const BSONObj& cmdObj) const override {
+        return parseNsCollectionRequired(dbname, cmdObj).ns();
+    }
+
+    bool supportsWriteConcern(const BSONObj& cmd) const override {
         return mr::mrSupportsWriteConcern(cmd);
     }
 
-    virtual void help(std::stringstream& help) const {
+    void help(std::stringstream& help) const override {
         help << "Runs the sharded map/reduce command";
     }
 
-    virtual void addRequiredPrivileges(const std::string& dbname,
-                                       const BSONObj& cmdObj,
-                                       std::vector<Privilege>* out) {
+    void addRequiredPrivileges(const std::string& dbname,
+                               const BSONObj& cmdObj,
+                               std::vector<Privilege>* out) override {
         mr::addPrivilegesRequiredForMapReduce(this, dbname, cmdObj, out);
     }
 
-    virtual bool run(OperationContext* opCtx,
-                     const std::string& dbname,
-                     BSONObj& cmdObj,
-                     int options,
-                     std::string& errmsg,
-                     BSONObjBuilder& result) {
+    bool run(OperationContext* opCtx,
+             const std::string& dbname,
+             BSONObj& cmdObj,
+             int options,
+             std::string& errmsg,
+             BSONObjBuilder& result) override {
         Timer t;
 
         const NamespaceString nss(parseNs(dbname, cmdObj));
-        uassert(ErrorCodes::InvalidNamespace, "Invalid namespace", nss.isValid());
-
-        const string shardResultCollection = getTmpName(nss.coll());
+        const std::string shardResultCollection = getTmpName(nss.coll());
 
         bool shardedOutput = false;
-        NamespaceString outputCollNss;
         bool customOutDB = false;
+        NamespaceString outputCollNss;
         bool inlineOutput = false;
 
-        string outDB = dbname;
+        std::string outDB = dbname;
 
         BSONElement outElmt = cmdObj.getField("out");
         if (outElmt.type() == Object) {
@@ -218,7 +210,8 @@ public:
                         !customOut.hasField("db"));
             } else {
                 // Mode must be 1st element
-                const string finalColShort = customOut.firstElement().str();
+                const std::string finalColShort = customOut.firstElement().str();
+
                 if (customOut.hasField("db")) {
                     customOutDB = true;
                     outDB = customOut.getField("db").str();
@@ -231,44 +224,27 @@ public:
             }
         }
 
-        // Ensure the input database exists
-        auto status = Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, dbname);
-        if (!status.isOK()) {
-            return appendCommandStatus(result, status.getStatus());
-        }
+        auto const catalogCache = Grid::get(opCtx)->catalogCache();
 
-        shared_ptr<DBConfig> confIn = status.getValue();
+        // Ensure the input database exists and set up the input collection
+        auto inputRoutingInfo = uassertStatusOK(catalogCache->getCollectionRoutingInfo(opCtx, nss));
 
-        shared_ptr<DBConfig> confOut;
-        if (customOutDB) {
-            // Create the output database implicitly, since we have a custom output requested
-            auto scopedDb = uassertStatusOK(ScopedShardDatabase::getOrCreate(opCtx, outDB));
-            confOut = scopedDb.getSharedDbReference();
-        } else {
-            confOut = confIn;
-        }
+        const bool shardedInput = inputRoutingInfo.cm() != nullptr;
 
-        if (confOut->getPrimaryId() == "config" && !inlineOutput) {
-            return appendCommandStatus(
-                result,
-                Status(ErrorCodes::CommandNotSupported,
-                       str::stream() << "Can not execute mapReduce with output database " << outDB
-                                     << " which lives on config servers"));
+        // Create the output database implicitly if we have a custom output requested
+        if (customOutDB) {
+            uassertStatusOK(createShardDatabase(opCtx, outDB));
         }
 
-        const bool shardedInput = confIn && confIn->isSharded(nss.ns());
-
-        if (!shardedOutput) {
-            uassert(15920,
-                    "Cannot output to a non-sharded collection because "
-                    "sharded collection exists already",
-                    !confOut->isSharded(outputCollNss.ns()));
-
-            // TODO: Should we also prevent going from non-sharded to sharded? During the
-            //       transition client may see partial data.
-        }
+        // Ensure that the output database doesn't reside on the config server
+        auto outputDbInfo = uassertStatusOK(catalogCache->getDatabase(opCtx, outDB));
+        uassert(ErrorCodes::CommandNotSupported,
+                str::stream() << "Can not execute mapReduce with output database " << outDB
+                              << " which lives on config servers",
+                inlineOutput || outputDbInfo.primaryId() != "config");
 
         int64_t maxChunkSizeBytes = 0;
+
         if (shardedOutput) {
             // Will need to figure out chunks, ask shards for points
             maxChunkSizeBytes = cmdObj["maxChunkSizeBytes"].numberLong();
@@ -279,29 +255,40 @@ public:
 
             // maxChunkSizeBytes is sent as int BSON field
             invariant(maxChunkSizeBytes < std::numeric_limits<int>::max());
+        } else if (outputCollNss.isValid()) {
+            auto outputRoutingInfo =
+                uassertStatusOK(catalogCache->getCollectionRoutingInfo(opCtx, outputCollNss));
+
+            uassert(15920,
+                    "Cannot output to a non-sharded collection because "
+                    "sharded collection exists already",
+                    !outputRoutingInfo.cm());
+
+            // TODO: Should we also prevent going from non-sharded to sharded? During the
+            //       transition client may see partial data.
         }
 
         const auto shardRegistry = Grid::get(opCtx)->shardRegistry();
 
         // modify command to run on shards with output to tmp collection
-        string badShardedField;
+        std::string badShardedField;
         BSONObj shardedCommand =
             fixForShards(cmdObj, shardResultCollection, badShardedField, maxChunkSizeBytes);
 
         if (!shardedInput && !shardedOutput && !customOutDB) {
             LOG(1) << "simple MR, just passthrough";
 
-            const auto shard =
-                uassertStatusOK(shardRegistry->getShard(opCtx, confIn->getPrimaryId()));
+            invariant(inputRoutingInfo.primary());
 
-            ShardConnection conn(shard->getConnString(), "");
+            ShardConnection conn(inputRoutingInfo.primary()->getConnString(), "");
 
             BSONObj res;
             bool ok = conn->runCommand(dbname, cmdObj, res);
             conn.done();
 
             if (auto wcErrorElem = res["writeConcernError"]) {
-                appendWriteConcernErrorToCmdResponse(shard->getId(), wcErrorElem, result);
+                appendWriteConcernErrorToCmdResponse(
+                    inputRoutingInfo.primary()->getId(), wcErrorElem, result);
             }
 
             result.appendElementsUnique(res);
@@ -323,12 +310,13 @@ public:
             collation = cmdObj["collation"].embeddedObjectUserCheck();
         }
 
-        set<string> servers;
-        vector<Strategy::CommandResult> mrCommandResults;
+        std::set<std::string> servers;
+        std::vector<Strategy::CommandResult> mrCommandResults;
 
         BSONObjBuilder shardResultsB;
         BSONObjBuilder shardCountsB;
-        map<string, int64_t> countsMap;
+        std::map<std::string, int64_t> countsMap;
+
         auto splitPts = SimpleBSONObjComparator::kInstance.makeBSONObjSet();
 
         {
@@ -349,12 +337,12 @@ public:
 
             for (const auto& mrResult : mrCommandResults) {
                 // Need to gather list of all servers even if an error happened
-                string server;
-                {
+                const auto server = [&]() {
                     const auto shard =
                         uassertStatusOK(shardRegistry->getShard(opCtx, mrResult.shardTargetId));
-                    server = shard->getConnString().toString();
-                }
+                    return shard->getConnString().toString();
+                }();
+
                 servers.insert(server);
 
                 if (!ok) {
@@ -386,15 +374,14 @@ public:
 
                 if (singleResult.hasField("splitKeys")) {
                     BSONElement splitKeys = singleResult.getField("splitKeys");
-                    vector<BSONElement> pts = splitKeys.Array();
-                    for (vector<BSONElement>::iterator it = pts.begin(); it != pts.end(); ++it) {
-                        splitPts.insert(it->Obj().getOwned());
+                    for (const auto& splitPt : splitKeys.Array()) {
+                        splitPts.insert(splitPt.Obj().getOwned());
                     }
                 }
             }
 
             if (!ok) {
-                _cleanUp(servers, dbname, shardResultCollection);
+                cleanUp(servers, dbname, shardResultCollection);
 
                 // Add "code" to the top-level response, if the failure of the sharded command
                 // can be accounted to a single error.
@@ -442,16 +429,15 @@ public:
 
         bool ok = true;
         BSONObj singleResult;
-        bool hasWCError = false;
 
         if (!shardedOutput) {
-            const auto shard =
-                uassertStatusOK(shardRegistry->getShard(opCtx, confOut->getPrimaryId()));
+            LOG(1) << "MR with single shard output, NS=" << outputCollNss
+                   << " primary=" << outputDbInfo.primaryId();
 
-            LOG(1) << "MR with single shard output, NS=" << outputCollNss.ns()
-                   << " primary=" << shard->toString();
+            const auto outputShard =
+                uassertStatusOK(shardRegistry->getShard(opCtx, outputDbInfo.primaryId()));
 
-            ShardConnection conn(shard->getConnString(), outputCollNss.ns());
+            ShardConnection conn(outputShard->getConnString(), outputCollNss.ns());
             ok = conn->runCommand(outDB, finalCmd.obj(), singleResult);
 
             BSONObj counts = singleResult.getObjectField("counts");
@@ -460,79 +446,19 @@ public:
             outputCount = counts.getIntField("output");
 
             conn.done();
-            if (!hasWCError) {
-                if (auto wcErrorElem = singleResult["writeConcernError"]) {
-                    appendWriteConcernErrorToCmdResponse(shard->getId(), wcErrorElem, result);
-                    hasWCError = true;
-                }
+
+            if (auto wcErrorElem = singleResult["writeConcernError"]) {
+                appendWriteConcernErrorToCmdResponse(outputShard->getId(), wcErrorElem, result);
             }
         } else {
             LOG(1) << "MR with sharded output, NS=" << outputCollNss.ns();
 
-            // Create the sharded collection if needed
-            if (!confOut->isSharded(outputCollNss.ns())) {
-                // Enable sharding on the output db
-                Status status = Grid::get(opCtx)->catalogClient(opCtx)->enableSharding(
-                    opCtx, outputCollNss.db().toString());
-
-                // If the database has sharding already enabled, we can ignore the error
-                if (status.isOK()) {
-                    // Invalidate the output database so it gets reloaded on the next fetch attempt
-                    Grid::get(opCtx)->catalogCache()->invalidate(outputCollNss.db());
-                } else if (status != ErrorCodes::AlreadyInitialized) {
-                    uassertStatusOK(status);
-                }
-
-                confOut.reset();
-                confOut = uassertStatusOK(Grid::get(opCtx)->catalogCache()->getDatabase(
-                    opCtx, outputCollNss.db().toString()));
+            auto outputRoutingInfo =
+                uassertStatusOK(catalogCache->getCollectionRoutingInfo(opCtx, outputCollNss));
 
-                // Shard collection according to split points
-                vector<BSONObj> sortedSplitPts;
-
-                // Points will be properly sorted using the set
-                for (const auto& splitPt : splitPts) {
-                    sortedSplitPts.push_back(splitPt);
-                }
-
-                // Pre-split the collection onto all the shards for this database. Note that
-                // it's not completely safe to pre-split onto non-primary shards using the
-                // shardcollection method (a conflict may result if multiple map-reduces are
-                // writing to the same output collection, for instance).
-                //
-                // TODO: pre-split mapReduce output in a safer way.
-
-                const std::set<ShardId> outShardIds = [&]() {
-                    std::vector<ShardId> shardIds;
-                    shardRegistry->getAllShardIds(&shardIds);
-                    uassert(ErrorCodes::ShardNotFound,
-                            str::stream()
-                                << "Unable to find shards on which to place output collection "
-                                << outputCollNss.ns(),
-                            !shardIds.empty());
-
-                    return std::set<ShardId>(shardIds.begin(), shardIds.end());
-                }();
-
-
-                BSONObj sortKey = BSON("_id" << 1);
-                ShardKeyPattern sortKeyPattern(sortKey);
-
-                // The collection default collation for the output collection. This is empty,
-                // representing the simple binary comparison collation.
-                BSONObj defaultCollation;
-
-                uassertStatusOK(
-                    Grid::get(opCtx)->catalogClient(opCtx)->shardCollection(opCtx,
-                                                                            outputCollNss.ns(),
-                                                                            sortKeyPattern,
-                                                                            defaultCollation,
-                                                                            true,
-                                                                            sortedSplitPts,
-                                                                            outShardIds));
-
-                // Make sure the cached metadata for the collection knows that we are now sharded
-                confOut->getChunkManager(opCtx, outputCollNss.ns(), true /* reload */);
+            // Create the sharded collection if needed
+            if (!outputRoutingInfo.cm()) {
+                outputRoutingInfo = createShardedOutputCollection(opCtx, outputCollNss, splitPts);
             }
 
             auto chunkSizes = SimpleBSONObjComparator::kInstance.makeBSONObjIndexedMap<int>();
@@ -567,14 +493,16 @@ public:
                     throw;
                 }
 
+                bool hasWCError = false;
+
                 for (const auto& mrResult : mrCommandResults) {
-                    string server;
-                    {
+                    const auto server = [&]() {
                         const auto shard =
                             uassertStatusOK(Grid::get(opCtx)->shardRegistry()->getShard(
                                 opCtx, mrResult.shardTargetId));
-                        server = shard->getConnString().toString();
-                    }
+                        return shard->getConnString().toString();
+                    }();
+
                     singleResult = mrResult.result;
                     if (!hasWCError) {
                         if (auto wcErrorElem = singleResult["writeConcernError"]) {
@@ -597,7 +525,8 @@ public:
                     // get the size inserted for each chunk
                     // split cannot be called here since we already have the distributed lock
                     if (singleResult.hasField("chunkSizes")) {
-                        vector<BSONElement> sizes = singleResult.getField("chunkSizes").Array();
+                        std::vector<BSONElement> sizes =
+                            singleResult.getField("chunkSizes").Array();
                         for (unsigned int i = 0; i < sizes.size(); i += 2) {
                             BSONObj key = sizes[i].Obj().getOwned();
                             const long long size = sizes[i + 1].numberLong();
@@ -610,34 +539,37 @@ public:
             }
 
             // Do the splitting round
-            shared_ptr<ChunkManager> cm =
-                confOut->getChunkManagerIfExists(opCtx, outputCollNss.ns());
+            catalogCache->onStaleConfigError(std::move(outputRoutingInfo));
+            outputRoutingInfo =
+                uassertStatusOK(catalogCache->getCollectionRoutingInfo(opCtx, outputCollNss));
             uassert(34359,
                     str::stream() << "Failed to write mapreduce output to " << outputCollNss.ns()
                                   << "; expected that collection to be sharded, but it was not",
-                    cm);
+                    outputRoutingInfo.cm());
+
+            const auto outputCM = outputRoutingInfo.cm();
 
             for (const auto& chunkSize : chunkSizes) {
                 BSONObj key = chunkSize.first;
                 const int size = chunkSize.second;
                 invariant(size < std::numeric_limits<int>::max());
 
-                // key reported should be the chunk's minimum
-                shared_ptr<Chunk> c = cm->findIntersectingChunkWithSimpleCollation(key);
+                // Key reported should be the chunk's minimum
+                auto c = outputCM->findIntersectingChunkWithSimpleCollation(key);
                 if (!c) {
                     warning() << "Mongod reported " << size << " bytes inserted for key " << key
                               << " but can't find chunk";
                 } else {
-                    updateChunkWriteStatsAndSplitIfNeeded(opCtx, cm.get(), c.get(), size);
+                    updateChunkWriteStatsAndSplitIfNeeded(opCtx, outputCM.get(), c.get(), size);
                 }
             }
         }
 
-        _cleanUp(servers, dbname, shardResultCollection);
+        cleanUp(servers, dbname, shardResultCollection);
 
         if (!ok) {
             errmsg = str::stream() << "MR post processing failed: " << singleResult.toString();
-            return 0;
+            return false;
         }
 
         // copy some elements from a single result
@@ -672,9 +604,69 @@ public:
 
 private:
     /**
+     * Creates and shards the collection for the output results.
+     */
+    static CachedCollectionRoutingInfo createShardedOutputCollection(OperationContext* opCtx,
+                                                                     const NamespaceString& nss,
+                                                                     const BSONObjSet& splitPts) {
+        auto const catalogClient = Grid::get(opCtx)->catalogClient(opCtx);
+        auto const catalogCache = Grid::get(opCtx)->catalogCache();
+        auto const shardRegistry = Grid::get(opCtx)->shardRegistry();
+
+        // Enable sharding on the output db
+        Status status = catalogClient->enableSharding(opCtx, nss.db().toString());
+
+        // If the database has sharding already enabled, we can ignore the error
+        if (status.isOK()) {
+            // Invalidate the output database so it gets reloaded on the next fetch attempt
+            catalogCache->purgeDatabase(nss.db());
+        } else if (status != ErrorCodes::AlreadyInitialized) {
+            uassertStatusOK(status);
+        }
+
+        // Points will be properly sorted using the set
+        const std::vector<BSONObj> sortedSplitPts(splitPts.begin(), splitPts.end());
+
+        // Pre-split the collection onto all the shards for this database. Note that
+        // it's not completely safe to pre-split onto non-primary shards using the
+        // shardcollection method (a conflict may result if multiple map-reduces are
+        // writing to the same output collection, for instance).
+        //
+        // TODO: pre-split mapReduce output in a safer way.
+
+        const std::set<ShardId> outShardIds = [&]() {
+            std::vector<ShardId> shardIds;
+            shardRegistry->getAllShardIds(&shardIds);
+            uassert(ErrorCodes::ShardNotFound,
+                    str::stream() << "Unable to find shards on which to place output collection "
+                                  << nss.ns(),
+                    !shardIds.empty());
+
+            return std::set<ShardId>(shardIds.begin(), shardIds.end());
+        }();
+
+
+        BSONObj sortKey = BSON("_id" << 1);
+        ShardKeyPattern sortKeyPattern(sortKey);
+
+        // The collection default collation for the output collection. This is empty,
+        // representing the simple binary comparison collation.
+        BSONObj defaultCollation;
+
+        uassertStatusOK(Grid::get(opCtx)->catalogClient(opCtx)->shardCollection(
+            opCtx, nss.ns(), sortKeyPattern, defaultCollation, true, sortedSplitPts, outShardIds));
+
+        // Make sure the cached metadata for the collection knows that we are now sharded
+        catalogCache->invalidateShardedCollection(nss);
+        return uassertStatusOK(catalogCache->getCollectionRoutingInfo(opCtx, nss));
+    }
+
+    /**
      * Drops the temporary results collections from each shard.
      */
-    void _cleanUp(const set<string>& servers, string dbName, string shardResultCollection) {
+    static void cleanUp(const std::set<std::string>& servers,
+                        const std::string& dbName,
+                        const std::string& shardResultCollection) {
         try {
             // drop collections with tmp results on each shard
             for (const auto& server : servers) {
diff --git a/src/mongo/s/commands/cluster_merge_chunks_cmd.cpp b/src/mongo/s/commands/cluster_merge_chunks_cmd.cpp
index 6b247823381..2a85b645df0 100644
--- a/src/mongo/s/commands/cluster_merge_chunks_cmd.cpp
+++ b/src/mongo/s/commands/cluster_merge_chunks_cmd.cpp
@@ -37,12 +37,10 @@
 #include "mongo/db/namespace_string.h"
 #include "mongo/s/catalog/sharding_catalog_client.h"
 #include "mongo/s/catalog_cache.h"
-#include "mongo/s/chunk_manager.h"
 #include "mongo/s/client/shard_connection.h"
 #include "mongo/s/client/shard_registry.h"
-#include "mongo/s/config.h"
+#include "mongo/s/commands/cluster_commands_common.h"
 #include "mongo/s/grid.h"
-#include "mongo/s/sharding_raii.h"
 
 namespace mongo {
 
@@ -60,14 +58,14 @@ class ClusterMergeChunksCommand : public Command {
 public:
     ClusterMergeChunksCommand() : Command("mergeChunks") {}
 
-    virtual void help(stringstream& h) const {
+    void help(stringstream& h) const override {
         h << "Merge Chunks command\n"
           << "usage: { mergeChunks : <ns>, bounds : [ <min key>, <max key> ] }";
     }
 
-    virtual Status checkAuthForCommand(Client* client,
-                                       const std::string& dbname,
-                                       const BSONObj& cmdObj) {
+    Status checkAuthForCommand(Client* client,
+                               const std::string& dbname,
+                               const BSONObj& cmdObj) override {
         if (!AuthorizationSession::get(client)->isAuthorizedForActionsOnResource(
                 ResourcePattern::forExactNamespace(NamespaceString(parseNs(dbname, cmdObj))),
                 ActionType::splitChunk)) {
@@ -76,17 +74,19 @@ public:
         return Status::OK();
     }
 
-    virtual std::string parseNs(const std::string& dbname, const BSONObj& cmdObj) const {
+    std::string parseNs(const std::string& dbname, const BSONObj& cmdObj) const override {
         return parseNsFullyQualified(dbname, cmdObj);
     }
 
-    virtual bool adminOnly() const {
+    bool adminOnly() const override {
         return true;
     }
-    virtual bool slaveOk() const {
+
+    bool slaveOk() const override {
         return false;
     }
-    virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
+
+    bool supportsWriteConcern(const BSONObj& cmd) const override {
         return false;
     }
 
@@ -104,10 +104,13 @@ public:
              BSONObj& cmdObj,
              int,
              string& errmsg,
-             BSONObjBuilder& result) {
+             BSONObjBuilder& result) override {
         const NamespaceString nss(parseNs(dbname, cmdObj));
 
-        auto scopedCM = uassertStatusOK(ScopedChunkManager::refreshAndGet(opCtx, nss));
+        auto routingInfo = uassertStatusOK(
+            Grid::get(opCtx)->catalogCache()->getShardedCollectionRoutingInfoWithRefresh(opCtx,
+                                                                                         nss));
+        const auto cm = routingInfo.cm();
 
         vector<BSONObj> bounds;
         if (!FieldParser::extract(cmdObj, boundsField, &bounds, &errmsg)) {
@@ -137,8 +140,6 @@ public:
             return false;
         }
 
-        auto const cm = scopedCM.cm();
-
         if (!cm->getShardKeyPattern().isShardKey(minKey) ||
             !cm->getShardKeyPattern().isShardKey(maxKey)) {
             errmsg = stream() << "shard key bounds "
@@ -179,6 +180,8 @@ public:
         bool ok = conn->runCommand("admin", remoteCmdObjB.obj(), remoteResult);
         conn.done();
 
+        Grid::get(opCtx)->catalogCache()->onStaleConfigError(std::move(routingInfo));
+
         result.appendElements(remoteResult);
         return ok;
     }
diff --git a/src/mongo/s/commands/cluster_move_chunk_cmd.cpp b/src/mongo/s/commands/cluster_move_chunk_cmd.cpp
index c3cb18ceb15..e597e80eeaa 100644
--- a/src/mongo/s/commands/cluster_move_chunk_cmd.cpp
+++ b/src/mongo/s/commands/cluster_move_chunk_cmd.cpp
@@ -40,12 +40,11 @@
 #include "mongo/db/write_concern_options.h"
 #include "mongo/s/balancer_configuration.h"
 #include "mongo/s/catalog_cache.h"
-#include "mongo/s/client/shard_connection.h"
 #include "mongo/s/client/shard_registry.h"
+#include "mongo/s/commands/cluster_commands_common.h"
 #include "mongo/s/config_server_client.h"
 #include "mongo/s/grid.h"
 #include "mongo/s/migration_secondary_throttle_options.h"
-#include "mongo/s/sharding_raii.h"
 #include "mongo/util/log.h"
 #include "mongo/util/timer.h"
 
@@ -60,19 +59,19 @@ class MoveChunkCmd : public Command {
 public:
     MoveChunkCmd() : Command("moveChunk", false, "movechunk") {}
 
-    virtual bool slaveOk() const {
+    bool slaveOk() const override {
         return true;
     }
 
-    virtual bool adminOnly() const {
+    bool adminOnly() const override {
         return true;
     }
 
-    virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
+    bool supportsWriteConcern(const BSONObj& cmd) const override {
         return true;
     }
 
-    virtual void help(std::stringstream& help) const {
+    void help(std::stringstream& help) const override {
         help << "Example: move chunk that contains the doc {num : 7} to shard001\n"
              << "  { movechunk : 'test.foo' , find : { num : 7 } , to : 'shard0001' }\n"
              << "Example: move chunk with lower bound 0 and upper bound 10 to shard001\n"
@@ -80,9 +79,9 @@ public:
              << " , to : 'shard001' }\n";
     }
 
-    virtual Status checkAuthForCommand(Client* client,
-                                       const std::string& dbname,
-                                       const BSONObj& cmdObj) {
+    Status checkAuthForCommand(Client* client,
+                               const std::string& dbname,
+                               const BSONObj& cmdObj) override {
         if (!AuthorizationSession::get(client)->isAuthorizedForActionsOnResource(
                 ResourcePattern::forExactNamespace(NamespaceString(parseNs(dbname, cmdObj))),
                 ActionType::moveChunk)) {
@@ -92,21 +91,24 @@ public:
         return Status::OK();
     }
 
-    virtual std::string parseNs(const std::string& dbname, const BSONObj& cmdObj) const {
+    std::string parseNs(const std::string& dbname, const BSONObj& cmdObj) const override {
         return parseNsFullyQualified(dbname, cmdObj);
     }
 
-    virtual bool run(OperationContext* opCtx,
-                     const std::string& dbname,
-                     BSONObj& cmdObj,
-                     int options,
-                     std::string& errmsg,
-                     BSONObjBuilder& result) {
+    bool run(OperationContext* opCtx,
+             const std::string& dbname,
+             BSONObj& cmdObj,
+             int options,
+             std::string& errmsg,
+             BSONObjBuilder& result) override {
         Timer t;
 
         const NamespaceString nss(parseNs(dbname, cmdObj));
 
-        auto scopedCM = uassertStatusOK(ScopedChunkManager::refreshAndGet(opCtx, nss));
+        auto routingInfo = uassertStatusOK(
+            Grid::get(opCtx)->catalogCache()->getShardedCollectionRoutingInfoWithRefresh(opCtx,
+                                                                                         nss));
+        const auto cm = routingInfo.cm();
 
         const auto toElt = cmdObj["to"];
         uassert(ErrorCodes::TypeMismatch,
@@ -145,8 +147,6 @@ public:
             return false;
         }
 
-        auto const cm = scopedCM.cm();
-
         shared_ptr<Chunk> chunk;
 
         if (!find.isEmpty()) {
@@ -199,9 +199,7 @@ public:
                                                     secondaryThrottle,
                                                     cmdObj["_waitForDelete"].trueValue()));
 
-        // Proactively refresh the chunk manager. Not strictly necessary, but this way it's
-        // immediately up-to-date the next time it's used.
-        scopedCM.db()->getChunkManagerIfExists(opCtx, nss.ns(), true);
+        Grid::get(opCtx)->catalogCache()->onStaleConfigError(std::move(routingInfo));
 
         result.append("millis", t.millis());
         return true;
diff --git a/src/mongo/s/commands/cluster_move_primary_cmd.cpp b/src/mongo/s/commands/cluster_move_primary_cmd.cpp
index dc192f1a6a6..e1746bebd72 100644
--- a/src/mongo/s/commands/cluster_move_primary_cmd.cpp
+++ b/src/mongo/s/commands/cluster_move_primary_cmd.cpp
@@ -49,7 +49,6 @@
 #include "mongo/s/client/shard_registry.h"
 #include "mongo/s/commands/cluster_commands_common.h"
 #include "mongo/s/commands/sharded_command_processing.h"
-#include "mongo/s/config.h"
 #include "mongo/s/grid.h"
 #include "mongo/util/log.h"
 
@@ -125,9 +124,9 @@ public:
         auto const shardRegistry = Grid::get(opCtx)->shardRegistry();
 
         // Flush all cached information. This can't be perfect, but it's better than nothing.
-        catalogCache->invalidate(dbname);
+        catalogCache->purgeDatabase(dbname);
 
-        auto config = uassertStatusOK(catalogCache->getDatabase(opCtx, dbname));
+        auto dbInfo = uassertStatusOK(catalogCache->getDatabase(opCtx, dbname));
 
         const auto toElt = cmdObj["to"];
         uassert(ErrorCodes::TypeMismatch,
@@ -139,8 +138,7 @@ public:
             return false;
         }
 
-        const auto fromShard =
-            uassertStatusOK(shardRegistry->getShard(opCtx, config->getPrimaryId()));
+        const auto fromShard = uassertStatusOK(shardRegistry->getShard(opCtx, dbInfo.primaryId()));
 
         const auto toShard = [&]() {
             auto toShardStatus = shardRegistry->getShard(opCtx, to);
@@ -223,7 +221,7 @@ public:
 
         // Ensure the next attempt to retrieve the database or any of its collections will do a full
         // reload
-        catalogCache->invalidate(dbname);
+        catalogCache->purgeDatabase(dbname);
 
         const string oldPrimary = fromShard->getConnString().toString();
 
diff --git a/src/mongo/s/commands/cluster_plan_cache_cmd.cpp b/src/mongo/s/commands/cluster_plan_cache_cmd.cpp
index b29eff8b0b2..65f149927d0 100644
--- a/src/mongo/s/commands/cluster_plan_cache_cmd.cpp
+++ b/src/mongo/s/commands/cluster_plan_cache_cmd.cpp
@@ -33,11 +33,11 @@
 #include "mongo/db/commands.h"
 #include "mongo/db/query/collation/collation_spec.h"
 #include "mongo/s/commands/strategy.h"
-#include "mongo/s/config.h"
 #include "mongo/s/grid.h"
 #include "mongo/s/stale_exception.h"
 
 namespace mongo {
+namespace {
 
 using std::string;
 using std::stringstream;
@@ -153,8 +153,6 @@ bool ClusterPlanCacheCmd::run(OperationContext* opCtx,
 // Register plan cache commands at startup
 //
 
-namespace {
-
 MONGO_INITIALIZER(RegisterPlanCacheCommands)(InitializerContext* context) {
     // Leaked intentionally: a Command registers itself when constructed.
 
@@ -174,5 +172,4 @@ MONGO_INITIALIZER(RegisterPlanCacheCommands)(InitializerContext* context) {
 }
 
 }  // namespace
-
 }  // namespace mongo
diff --git a/src/mongo/s/commands/cluster_shard_collection_cmd.cpp b/src/mongo/s/commands/cluster_shard_collection_cmd.cpp
index 7692e764e02..a1d99a608e6 100644
--- a/src/mongo/s/commands/cluster_shard_collection_cmd.cpp
+++ b/src/mongo/s/commands/cluster_shard_collection_cmd.cpp
@@ -53,15 +53,12 @@
 #include "mongo/s/balancer_configuration.h"
 #include "mongo/s/catalog/sharding_catalog_client.h"
 #include "mongo/s/catalog_cache.h"
-#include "mongo/s/chunk_manager.h"
 #include "mongo/s/client/shard_registry.h"
 #include "mongo/s/commands/cluster_write.h"
-#include "mongo/s/config.h"
 #include "mongo/s/config_server_client.h"
 #include "mongo/s/grid.h"
 #include "mongo/s/migration_secondary_throttle_options.h"
 #include "mongo/s/shard_util.h"
-#include "mongo/s/sharding_raii.h"
 #include "mongo/util/log.h"
 
 namespace mongo {
@@ -187,19 +184,21 @@ public:
 
         auto const catalogClient = Grid::get(opCtx)->catalogClient(opCtx);
         auto const shardRegistry = Grid::get(opCtx)->shardRegistry();
+        auto const catalogCache = Grid::get(opCtx)->catalogCache();
 
-        auto scopedShardedDb = uassertStatusOK(ScopedShardDatabase::getExisting(opCtx, nss.db()));
-        const auto config = scopedShardedDb.db();
+        auto dbInfo = uassertStatusOK(catalogCache->getDatabase(opCtx, nss.db()));
 
         // Ensure sharding is allowed on the database
         uassert(ErrorCodes::IllegalOperation,
                 str::stream() << "sharding not enabled for db " << nss.db(),
-                config->isShardingEnabled());
+                dbInfo.shardingEnabled());
+
+        auto routingInfo = uassertStatusOK(catalogCache->getCollectionRoutingInfo(opCtx, nss));
 
         // Ensure that the collection is not sharded already
         uassert(ErrorCodes::IllegalOperation,
                 str::stream() << "sharding already enabled for collection " << nss.ns(),
-                !config->isSharded(nss.ns()));
+                !routingInfo.cm());
 
         // NOTE: We *must* take ownership of the key here - otherwise the shared BSONObj becomes
         // corrupt as soon as the command ends.
@@ -279,13 +278,7 @@ public:
         }
 
         // The rest of the checks require a connection to the primary db
-        const ConnectionString shardConnString = [&]() {
-            const auto shard =
-                uassertStatusOK(shardRegistry->getShard(opCtx, config->getPrimaryId()));
-            return shard->getConnString();
-        }();
-
-        ScopedDbConnection conn(shardConnString);
+        ScopedDbConnection conn(routingInfo.primary()->getConnString());
 
         // Retrieve the collection metadata in order to verify that it is legal to shard this
         // collection.
@@ -590,17 +583,21 @@ public:
                                                        initSplits,
                                                        std::set<ShardId>{}));
 
-        // Make sure the cached metadata for the collection knows that we are now sharded
-        config->getChunkManager(opCtx, nss.ns(), true /* reload */);
-
         result << "collectionsharded" << nss.ns();
 
+        // Make sure the cached metadata for the collection knows that we are now sharded
+        catalogCache->invalidateShardedCollection(nss);
+
         // Only initially move chunks when using a hashed shard key
         if (isHashedShardKey && isEmpty) {
-            // Reload the new config info.  If we created more than one initial chunk, then
-            // we need to move them around to balance.
-            auto chunkManager = config->getChunkManager(opCtx, nss.ns(), true);
-            ChunkMap chunkMap = chunkManager->getChunkMap();
+            routingInfo = uassertStatusOK(catalogCache->getCollectionRoutingInfo(opCtx, nss));
+            uassert(ErrorCodes::ConflictingOperationInProgress,
+                    "Collection was successfully written as sharded but got dropped before it "
+                    "could be evenly distributed",
+                    routingInfo.cm());
+            auto chunkManager = routingInfo.cm();
+
+            const auto chunkMap = chunkManager->chunkMap();
 
             // 2. Move and commit each "big chunk" to a different shard.
             int i = 0;
@@ -646,7 +643,13 @@ public:
             }
 
             // Reload the config info, after all the migrations
-            chunkManager = config->getChunkManager(opCtx, nss.ns(), true);
+            catalogCache->invalidateShardedCollection(nss);
+            routingInfo = uassertStatusOK(catalogCache->getCollectionRoutingInfo(opCtx, nss));
+            uassert(ErrorCodes::ConflictingOperationInProgress,
+                    "Collection was successfully written as sharded but got dropped before it "
+                    "could be evenly distributed",
+                    routingInfo.cm());
+            chunkManager = routingInfo.cm();
 
             // 3. Subdivide the big chunks by splitting at each of the points in "allSplits"
             //    that we haven't already split by.
@@ -689,10 +692,6 @@ public:
                     subSplits.push_back(splitPoint);
                 }
             }
-
-            // Proactively refresh the chunk manager. Not really necessary, but this way it's
-            // immediately up-to-date the next time it's used.
-            config->getChunkManager(opCtx, nss.ns(), true);
         }
 
         return true;
diff --git a/src/mongo/s/commands/cluster_split_cmd.cpp b/src/mongo/s/commands/cluster_split_cmd.cpp
index b63da3b2ee7..80ed1526663 100644
--- a/src/mongo/s/commands/cluster_split_cmd.cpp
+++ b/src/mongo/s/commands/cluster_split_cmd.cpp
@@ -37,15 +37,13 @@
 #include "mongo/db/auth/action_type.h"
 #include "mongo/db/auth/authorization_manager.h"
 #include "mongo/db/auth/authorization_session.h"
-#include "mongo/db/client.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/field_parser.h"
 #include "mongo/s/catalog_cache.h"
-#include "mongo/s/chunk_manager.h"
 #include "mongo/s/client/shard_registry.h"
+#include "mongo/s/commands/cluster_commands_common.h"
 #include "mongo/s/grid.h"
 #include "mongo/s/shard_util.h"
-#include "mongo/s/sharding_raii.h"
 #include "mongo/util/log.h"
 
 namespace mongo {
@@ -90,20 +88,19 @@ class SplitCollectionCmd : public Command {
 public:
     SplitCollectionCmd() : Command("split", false, "split") {}
 
-    virtual bool slaveOk() const {
+    bool slaveOk() const override {
         return true;
     }
 
-    virtual bool adminOnly() const {
+    bool adminOnly() const override {
         return true;
     }
 
-
-    virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
+    bool supportsWriteConcern(const BSONObj& cmd) const override {
         return false;
     }
 
-    virtual void help(std::stringstream& help) const {
+    void help(std::stringstream& help) const override {
         help << " example: - split the shard that contains give key\n"
              << "   { split : 'alleyinsider.blog.posts' , find : { ts : 1 } }\n"
              << " example: - split the shard that contains the key with this as the middle\n"
@@ -111,9 +108,9 @@ public:
              << " NOTE: this does not move the chunks, it just creates a logical separation.";
     }
 
-    virtual Status checkAuthForCommand(Client* client,
-                                       const std::string& dbname,
-                                       const BSONObj& cmdObj) {
+    Status checkAuthForCommand(Client* client,
+                               const std::string& dbname,
+                               const BSONObj& cmdObj) override {
         if (!AuthorizationSession::get(client)->isAuthorizedForActionsOnResource(
                 ResourcePattern::forExactNamespace(NamespaceString(parseNs(dbname, cmdObj))),
                 ActionType::splitChunk)) {
@@ -122,19 +119,22 @@ public:
         return Status::OK();
     }
 
-    virtual std::string parseNs(const std::string& dbname, const BSONObj& cmdObj) const {
+    std::string parseNs(const std::string& dbname, const BSONObj& cmdObj) const override {
         return parseNsFullyQualified(dbname, cmdObj);
     }
 
-    virtual bool run(OperationContext* opCtx,
-                     const std::string& dbname,
-                     BSONObj& cmdObj,
-                     int options,
-                     std::string& errmsg,
-                     BSONObjBuilder& result) {
+    bool run(OperationContext* opCtx,
+             const std::string& dbname,
+             BSONObj& cmdObj,
+             int options,
+             std::string& errmsg,
+             BSONObjBuilder& result) override {
         const NamespaceString nss(parseNs(dbname, cmdObj));
 
-        auto scopedCM = uassertStatusOK(ScopedChunkManager::refreshAndGet(opCtx, nss));
+        auto routingInfo = uassertStatusOK(
+            Grid::get(opCtx)->catalogCache()->getShardedCollectionRoutingInfoWithRefresh(opCtx,
+                                                                                         nss));
+        const auto cm = routingInfo.cm();
 
         const BSONField<BSONObj> findField("find", BSONObj());
         const BSONField<BSONArray> boundsField("bounds", BSONArray());
@@ -190,8 +190,6 @@ public:
             return false;
         }
 
-        auto const cm = scopedCM.cm();
-
         std::shared_ptr<Chunk> chunk;
 
         if (!find.isEmpty()) {
@@ -275,9 +273,7 @@ public:
                                                   ChunkRange(chunk->getMin(), chunk->getMax()),
                                                   {splitPoint}));
 
-        // Proactively refresh the chunk manager. Not strictly necessary, but this way it's
-        // immediately up-to-date the next time it's used.
-        scopedCM.db()->getChunkManagerIfExists(opCtx, nss.ns(), true);
+        Grid::get(opCtx)->catalogCache()->onStaleConfigError(std::move(routingInfo));
 
         return true;
     }
diff --git a/src/mongo/s/commands/cluster_write.cpp b/src/mongo/s/commands/cluster_write.cpp
index 730d5e8a178..8b8a3f2e644 100644
--- a/src/mongo/s/commands/cluster_write.cpp
+++ b/src/mongo/s/commands/cluster_write.cpp
@@ -39,14 +39,13 @@
 #include "mongo/s/balancer_configuration.h"
 #include "mongo/s/catalog/sharding_catalog_client.h"
 #include "mongo/s/catalog/type_collection.h"
-#include "mongo/s/chunk.h"
+#include "mongo/s/catalog_cache.h"
 #include "mongo/s/client/shard_registry.h"
 #include "mongo/s/commands/chunk_manager_targeter.h"
 #include "mongo/s/commands/dbclient_multi_command.h"
 #include "mongo/s/config_server_client.h"
 #include "mongo/s/grid.h"
 #include "mongo/s/shard_util.h"
-#include "mongo/s/sharding_raii.h"
 #include "mongo/util/log.h"
 #include "mongo/util/mongoutils/str.h"
 
@@ -66,11 +65,6 @@ void toBatchError(const Status& status, BatchedCommandResponse* response) {
     dassert(response->isValid(NULL));
 }
 
-void reloadChunkManager(OperationContext* opCtx, const NamespaceString& nss) {
-    auto config = uassertStatusOK(ScopedShardDatabase::getExisting(opCtx, nss.db()));
-    config.db()->getChunkManagerIfExists(opCtx, nss.ns(), true);
-}
-
 /**
  * Given a maxChunkSize configuration and the number of chunks in a particular sharded collection,
  * returns an optimal chunk size to use in order to achieve a good ratio between number of chunks
@@ -176,30 +170,31 @@ BSONObj findExtremeKeyForShard(OperationContext* opCtx,
 void splitIfNeeded(OperationContext* opCtx,
                    const NamespaceString& nss,
                    const TargeterStats& stats) {
-    auto scopedCMStatus = ScopedChunkManager::get(opCtx, nss);
-    if (!scopedCMStatus.isOK()) {
-        warning() << "failed to get collection information for " << nss
-                  << " while checking for auto-split" << causedBy(scopedCMStatus.getStatus());
+    auto routingInfoStatus = Grid::get(opCtx)->catalogCache()->getCollectionRoutingInfo(opCtx, nss);
+    if (!routingInfoStatus.isOK()) {
+        log() << "failed to get collection information for " << nss
+              << " while checking for auto-split" << causedBy(routingInfoStatus.getStatus());
         return;
     }
 
-    const auto& scopedCM = scopedCMStatus.getValue();
+    auto& routingInfo = routingInfoStatus.getValue();
 
-    if (!scopedCM.cm()) {
+    if (!routingInfo.cm()) {
         return;
     }
 
     for (auto it = stats.chunkSizeDelta.cbegin(); it != stats.chunkSizeDelta.cend(); ++it) {
         std::shared_ptr<Chunk> chunk;
         try {
-            chunk = scopedCM.cm()->findIntersectingChunkWithSimpleCollation(it->first);
+            chunk = routingInfo.cm()->findIntersectingChunkWithSimpleCollation(it->first);
         } catch (const AssertionException& ex) {
             warning() << "could not find chunk while checking for auto-split: "
                       << causedBy(redact(ex));
             return;
         }
 
-        updateChunkWriteStatsAndSplitIfNeeded(opCtx, scopedCM.cm().get(), chunk.get(), it->second);
+        updateChunkWriteStatsAndSplitIfNeeded(
+            opCtx, routingInfo.cm().get(), chunk.get(), it->second);
     }
 }
 
@@ -472,21 +467,22 @@ void updateChunkWriteStatsAndSplitIfNeeded(OperationContext* opCtx,
               << (suggestedMigrateChunk ? "" : (std::string) " (migrate suggested" +
                           (shouldBalance ? ")" : ", but no migrations allowed)"));
 
+        // Reload the chunk manager after the split
+        auto routingInfo = uassertStatusOK(
+            Grid::get(opCtx)->catalogCache()->getShardedCollectionRoutingInfoWithRefresh(opCtx,
+                                                                                         nss));
+
         if (!shouldBalance || !suggestedMigrateChunk) {
-            reloadChunkManager(opCtx, nss);
             return;
         }
 
         // Top chunk optimization - try to move the top chunk out of this shard to prevent the hot
-        // spot
-        // from staying on a single shard. This is based on the assumption that succeeding inserts
-        // will
-        // fall on the top chunk.
+        // spot from staying on a single shard. This is based on the assumption that succeeding
+        // inserts will fall on the top chunk.
 
         // We need to use the latest chunk manager (after the split) in order to have the most
         // up-to-date view of the chunk we are about to move
-        auto scopedCM = uassertStatusOK(ScopedChunkManager::refreshAndGet(opCtx, nss));
-        auto suggestedChunk = scopedCM.cm()->findIntersectingChunkWithSimpleCollation(
+        auto suggestedChunk = routingInfo.cm()->findIntersectingChunkWithSimpleCollation(
             suggestedMigrateChunk->getMin());
 
         ChunkType chunkToMove;
@@ -498,7 +494,8 @@ void updateChunkWriteStatsAndSplitIfNeeded(OperationContext* opCtx,
 
         uassertStatusOK(configsvr_client::rebalanceChunk(opCtx, chunkToMove));
 
-        reloadChunkManager(opCtx, nss);
+        // Ensure the collection gets reloaded because of the move
+        Grid::get(opCtx)->catalogCache()->invalidateShardedCollection(nss);
     } catch (const DBException& ex) {
         chunk->randomizeBytesWritten();
 
diff --git a/src/mongo/s/commands/commands_public.cpp b/src/mongo/s/commands/commands_public.cpp
index 7d14499e9d8..e219d9396f9 100644
--- a/src/mongo/s/commands/commands_public.cpp
+++ b/src/mongo/s/commands/commands_public.cpp
@@ -54,17 +54,14 @@
 #include "mongo/rpc/get_status_from_command_result.h"
 #include "mongo/s/catalog/sharding_catalog_client.h"
 #include "mongo/s/catalog_cache.h"
-#include "mongo/s/chunk_manager.h"
 #include "mongo/s/client/shard_connection.h"
 #include "mongo/s/client/shard_registry.h"
 #include "mongo/s/commands/cluster_commands_common.h"
 #include "mongo/s/commands/cluster_explain.h"
 #include "mongo/s/commands/run_on_all_shards_cmd.h"
 #include "mongo/s/commands/sharded_command_processing.h"
-#include "mongo/s/config.h"
 #include "mongo/s/grid.h"
 #include "mongo/s/query/store_possible_cursor.h"
-#include "mongo/s/sharding_raii.h"
 #include "mongo/s/stale_exception.h"
 #include "mongo/scripting/engine.h"
 #include "mongo/util/log.h"
@@ -86,20 +83,19 @@ using std::vector;
 namespace {
 
 bool cursorCommandPassthrough(OperationContext* opCtx,
-                              shared_ptr<DBConfig> conf,
+                              StringData dbName,
+                              const ShardId& shardId,
                               const BSONObj& cmdObj,
                               const NamespaceString& nss,
                               int options,
                               BSONObjBuilder* out) {
-    const auto shardStatus =
-        Grid::get(opCtx)->shardRegistry()->getShard(opCtx, conf->getPrimaryId());
+    const auto shardStatus = Grid::get(opCtx)->shardRegistry()->getShard(opCtx, shardId);
     if (!shardStatus.isOK()) {
-        invariant(shardStatus.getStatus() == ErrorCodes::ShardNotFound);
         return Command::appendCommandStatus(*out, shardStatus.getStatus());
     }
     const auto shard = shardStatus.getValue();
     ScopedDbConnection conn(shard->getConnString());
-    auto cursor = conn->query(str::stream() << conf->name() << ".$cmd",
+    auto cursor = conn->query(str::stream() << dbName << ".$cmd",
                               cmdObj,
                               -1,    // nToReturn
                               0,     // nToSkip
@@ -155,11 +151,13 @@ StatusWith<BSONObj> getCollation(const BSONObj& cmdObj) {
 }
 
 class PublicGridCommand : public Command {
-public:
+protected:
     PublicGridCommand(const char* n, const char* oldname = NULL) : Command(n, false, oldname) {}
+
     virtual bool slaveOk() const {
         return true;
     }
+
     virtual bool adminOnly() const {
         return false;
     }
@@ -170,41 +168,29 @@ public:
         return false;
     }
 
-    // all grid commands are designed not to lock
-
-protected:
-    bool passthrough(OperationContext* opCtx,
-                     DBConfig* conf,
-                     const BSONObj& cmdObj,
-                     BSONObjBuilder& result) {
-        return _passthrough(opCtx, conf->name(), conf, cmdObj, 0, result);
-    }
-
     bool adminPassthrough(OperationContext* opCtx,
-                          DBConfig* conf,
+                          const ShardId& shardId,
                           const BSONObj& cmdObj,
                           BSONObjBuilder& result) {
-        return _passthrough(opCtx, "admin", conf, cmdObj, 0, result);
+        return passthrough(opCtx, "admin", shardId, cmdObj, result);
     }
 
     bool passthrough(OperationContext* opCtx,
-                     DBConfig* conf,
+                     const std::string& db,
+                     const ShardId& shardId,
                      const BSONObj& cmdObj,
-                     int options,
                      BSONObjBuilder& result) {
-        return _passthrough(opCtx, conf->name(), conf, cmdObj, options, result);
+        return passthrough(opCtx, db, shardId, cmdObj, 0, result);
     }
 
-private:
-    bool _passthrough(OperationContext* opCtx,
-                      const string& db,
-                      DBConfig* conf,
-                      const BSONObj& cmdObj,
-                      int options,
-                      BSONObjBuilder& result) {
-        const auto shardStatus =
-            Grid::get(opCtx)->shardRegistry()->getShard(opCtx, conf->getPrimaryId());
-        const auto shard = uassertStatusOK(shardStatus);
+    bool passthrough(OperationContext* opCtx,
+                     const std::string& db,
+                     const ShardId& shardId,
+                     const BSONObj& cmdObj,
+                     int options,
+                     BSONObjBuilder& result) {
+        const auto shard =
+            uassertStatusOK(Grid::get(opCtx)->shardRegistry()->getShard(opCtx, shardId));
 
         ShardConnection conn(shard->getConnString(), "");
 
@@ -223,53 +209,50 @@ private:
 };
 
 class AllShardsCollectionCommand : public RunOnAllShardsCommand {
-public:
+protected:
     AllShardsCollectionCommand(const char* n,
                                const char* oldname = NULL,
                                bool useShardConn = false,
                                bool implicitCreateDb = false)
         : RunOnAllShardsCommand(n, oldname, useShardConn, implicitCreateDb) {}
 
-    virtual void getShardIds(OperationContext* opCtx,
-                             const string& dbName,
-                             BSONObj& cmdObj,
-                             vector<ShardId>& shardIds) {
+    void getShardIds(OperationContext* opCtx,
+                     const string& dbName,
+                     BSONObj& cmdObj,
+                     vector<ShardId>& shardIds) override {
         const NamespaceString nss(parseNsCollectionRequired(dbName, cmdObj));
 
-        auto status = Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, dbName);
-        uassertStatusOK(status.getStatus());
-
-        shared_ptr<DBConfig> conf = status.getValue();
-
-        if (!conf->isSharded(nss.ns())) {
-            shardIds.push_back(conf->getPrimaryId());
-        } else {
+        const auto routingInfo =
+            uassertStatusOK(Grid::get(opCtx)->catalogCache()->getCollectionRoutingInfo(opCtx, nss));
+        if (routingInfo.cm()) {
+            // If it's a sharded collection, send it to all shards
             Grid::get(opCtx)->shardRegistry()->getAllShardIds(&shardIds);
+        } else {
+            // Otherwise just send it to the primary shard for the database
+            shardIds.push_back(routingInfo.primaryId());
         }
     }
 };
 
 class NotAllowedOnShardedCollectionCmd : public PublicGridCommand {
-public:
+protected:
     NotAllowedOnShardedCollectionCmd(const char* n) : PublicGridCommand(n) {}
 
-    virtual bool run(OperationContext* opCtx,
-                     const string& dbName,
-                     BSONObj& cmdObj,
-                     int options,
-                     string& errmsg,
-                     BSONObjBuilder& result) {
+    bool run(OperationContext* opCtx,
+             const string& dbName,
+             BSONObj& cmdObj,
+             int options,
+             string& errmsg,
+             BSONObjBuilder& result) override {
         const NamespaceString nss(parseNs(dbName, cmdObj));
 
-        auto conf = uassertStatusOK(Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, dbName));
-        if (!conf->isSharded(nss.ns())) {
-            return passthrough(opCtx, conf.get(), cmdObj, options, result);
-        }
+        const auto routingInfo =
+            uassertStatusOK(Grid::get(opCtx)->catalogCache()->getCollectionRoutingInfo(opCtx, nss));
+        uassert(ErrorCodes::IllegalOperation,
+                str::stream() << "can't do command: " << getName() << " on sharded collection",
+                !routingInfo.cm());
 
-        return appendCommandStatus(
-            result,
-            Status(ErrorCodes::IllegalOperation,
-                   str::stream() << "can't do command: " << getName() << " on sharded collection"));
+        return passthrough(opCtx, dbName, routingInfo.primaryId(), cmdObj, options, result);
     }
 };
 
@@ -407,6 +390,7 @@ public:
 class ReIndexCmd : public AllShardsCollectionCommand {
 public:
     ReIndexCmd() : AllShardsCollectionCommand("reIndex") {}
+
     virtual void addRequiredPrivileges(const std::string& dbname,
                                        const BSONObj& cmdObj,
                                        std::vector<Privilege>* out) {
@@ -418,6 +402,7 @@ public:
     virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
         return false;
     }
+
 } reIndexCmd;
 
 class CollectionModCmd : public AllShardsCollectionCommand {
@@ -434,12 +419,13 @@ public:
     virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
         return true;
     }
-} collectionModCmd;
 
+} collectionModCmd;
 
 class ValidateCmd : public PublicGridCommand {
 public:
     ValidateCmd() : PublicGridCommand("validate") {}
+
     virtual void addRequiredPrivileges(const std::string& dbname,
                                        const BSONObj& cmdObj,
                                        std::vector<Privilege>* out) {
@@ -460,13 +446,13 @@ public:
              BSONObjBuilder& output) {
         const NamespaceString nss(parseNsCollectionRequired(dbName, cmdObj));
 
-        auto conf = uassertStatusOK(Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, dbName));
-        if (!conf->isSharded(nss.ns())) {
-            return passthrough(opCtx, conf.get(), cmdObj, output);
+        auto routingInfo =
+            uassertStatusOK(Grid::get(opCtx)->catalogCache()->getCollectionRoutingInfo(opCtx, nss));
+        if (!routingInfo.cm()) {
+            return passthrough(opCtx, dbName, routingInfo.primaryId(), cmdObj, output);
         }
 
-        shared_ptr<ChunkManager> cm = conf->getChunkManager(opCtx, nss.ns());
-        massert(40051, "chunk manager should not be null", cm);
+        const auto cm = routingInfo.cm();
 
         vector<Strategy::CommandResult> results;
         const BSONObj query;
@@ -512,33 +498,35 @@ public:
         }
         return true;
     }
+
 } validateCmd;
 
 class CreateCmd : public PublicGridCommand {
 public:
     CreateCmd() : PublicGridCommand("create") {}
-    virtual Status checkAuthForCommand(Client* client,
-                                       const std::string& dbname,
-                                       const BSONObj& cmdObj) {
+
+    Status checkAuthForCommand(Client* client,
+                               const std::string& dbname,
+                               const BSONObj& cmdObj) override {
         const NamespaceString nss(parseNsCollectionRequired(dbname, cmdObj));
         return AuthorizationSession::get(client)->checkAuthForCreate(nss, cmdObj);
     }
-    virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
+
+    bool supportsWriteConcern(const BSONObj& cmd) const override {
         return true;
     }
+
     bool run(OperationContext* opCtx,
              const string& dbName,
              BSONObj& cmdObj,
              int,
              string& errmsg,
-             BSONObjBuilder& result) {
-        auto dbStatus = ScopedShardDatabase::getOrCreate(opCtx, dbName);
-        if (!dbStatus.isOK()) {
-            return appendCommandStatus(result, dbStatus.getStatus());
-        }
+             BSONObjBuilder& result) override {
+        uassertStatusOK(createShardDatabase(opCtx, dbName));
 
-        auto scopedDb = std::move(dbStatus.getValue());
-        return passthrough(opCtx, scopedDb.db(), cmdObj, result);
+        const auto dbInfo =
+            uassertStatusOK(Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, dbName));
+        return passthrough(opCtx, dbName, dbInfo.primaryId(), cmdObj, result);
     }
 
 } createCmd;
@@ -546,23 +534,27 @@ public:
 class RenameCollectionCmd : public PublicGridCommand {
 public:
     RenameCollectionCmd() : PublicGridCommand("renameCollection") {}
+
     virtual Status checkAuthForCommand(Client* client,
                                        const std::string& dbname,
                                        const BSONObj& cmdObj) {
         return rename_collection::checkAuthForRenameCollectionCommand(client, dbname, cmdObj);
     }
+
     virtual bool adminOnly() const {
         return true;
     }
+
     virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
         return true;
     }
+
     bool run(OperationContext* opCtx,
              const string& dbName,
              BSONObj& cmdObj,
-             int,
+             int options,
              string& errmsg,
-             BSONObjBuilder& result) {
+             BSONObjBuilder& result) override {
         const auto fullNsFromElt = cmdObj.firstElement();
         uassert(ErrorCodes::InvalidNamespace,
                 "'renameCollection' must be of type String",
@@ -571,10 +563,6 @@ public:
         uassert(ErrorCodes::InvalidNamespace,
                 str::stream() << "Invalid source namespace: " << fullnsFrom.ns(),
                 fullnsFrom.isValid());
-        const string dbNameFrom = fullnsFrom.db().toString();
-
-        auto confFrom =
-            uassertStatusOK(Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, dbNameFrom));
 
         const auto fullnsToElt = cmdObj["to"];
         uassert(ErrorCodes::InvalidNamespace,
@@ -584,24 +572,22 @@ public:
         uassert(ErrorCodes::InvalidNamespace,
                 str::stream() << "Invalid target namespace: " << fullnsTo.ns(),
                 fullnsTo.isValid());
-        const string dbNameTo = fullnsTo.db().toString();
-        auto confTo =
-            uassertStatusOK(Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, dbNameTo));
 
-        uassert(
-            13138, "You can't rename a sharded collection", !confFrom->isSharded(fullnsFrom.ns()));
-        uassert(
-            13139, "You can't rename to a sharded collection", !confTo->isSharded(fullnsTo.ns()));
+        const auto fromRoutingInfo = uassertStatusOK(
+            Grid::get(opCtx)->catalogCache()->getCollectionRoutingInfo(opCtx, fullnsFrom));
+        uassert(13138, "You can't rename a sharded collection", !fromRoutingInfo.cm());
 
-        auto shardTo = confTo->getPrimaryId();
-        auto shardFrom = confFrom->getPrimaryId();
+        const auto toRoutingInfo = uassertStatusOK(
+            Grid::get(opCtx)->catalogCache()->getCollectionRoutingInfo(opCtx, fullnsTo));
+        uassert(13139, "You can't rename to a sharded collection", !toRoutingInfo.cm());
 
         uassert(13137,
                 "Source and destination collections must be on same shard",
-                shardFrom == shardTo);
+                fromRoutingInfo.primaryId() == toRoutingInfo.primaryId());
 
-        return adminPassthrough(opCtx, confFrom.get(), cmdObj, result);
+        return adminPassthrough(opCtx, fromRoutingInfo.primaryId(), cmdObj, result);
     }
+
 } renameCollectionCmd;
 
 class CopyDBCmd : public PublicGridCommand {
@@ -637,14 +623,14 @@ public:
                 "Invalid todb argument",
                 NamespaceString::validDBName(todb, NamespaceString::DollarInDbNameBehavior::Allow));
 
-        auto scopedToDb = uassertStatusOK(ScopedShardDatabase::getOrCreate(opCtx, todb));
+        auto toDbInfo = uassertStatusOK(createShardDatabase(opCtx, todb));
         uassert(ErrorCodes::IllegalOperation,
                 "Cannot copy to a sharded database",
-                !scopedToDb.db()->isShardingEnabled());
+                !toDbInfo.shardingEnabled());
 
-        const string fromhost = cmdObj.getStringField("fromhost");
+        const std::string fromhost = cmdObj.getStringField("fromhost");
         if (!fromhost.empty()) {
-            return adminPassthrough(opCtx, scopedToDb.db(), cmdObj, result);
+            return adminPassthrough(opCtx, toDbInfo.primaryId(), cmdObj, result);
         }
 
         const auto fromDbElt = cmdObj["fromdb"];
@@ -657,10 +643,10 @@ public:
             "invalid fromdb argument",
             NamespaceString::validDBName(fromdb, NamespaceString::DollarInDbNameBehavior::Allow));
 
-        auto scopedFromDb = uassertStatusOK(ScopedShardDatabase::getExisting(opCtx, fromdb));
+        auto fromDbInfo = uassertStatusOK(createShardDatabase(opCtx, fromdb));
         uassert(ErrorCodes::IllegalOperation,
                 "Cannot copy from a sharded database",
-                !scopedFromDb.db()->isShardingEnabled());
+                !fromDbInfo.shardingEnabled());
 
         BSONObjBuilder b;
         BSONForEach(e, cmdObj) {
@@ -670,12 +656,12 @@ public:
         }
 
         {
-            const auto shard = uassertStatusOK(Grid::get(opCtx)->shardRegistry()->getShard(
-                opCtx, scopedFromDb.db()->getPrimaryId()));
+            const auto shard = uassertStatusOK(
+                Grid::get(opCtx)->shardRegistry()->getShard(opCtx, fromDbInfo.primaryId()));
             b.append("fromhost", shard->getConnString().toString());
         }
 
-        return adminPassthrough(opCtx, scopedToDb.db(), b.obj(), result);
+        return adminPassthrough(opCtx, toDbInfo.primaryId(), b.obj(), result);
     }
 
 } clusterCopyDBCmd;
@@ -683,38 +669,38 @@ public:
 class CollectionStats : public PublicGridCommand {
 public:
     CollectionStats() : PublicGridCommand("collStats", "collstats") {}
-    virtual void addRequiredPrivileges(const std::string& dbname,
-                                       const BSONObj& cmdObj,
-                                       std::vector<Privilege>* out) {
+
+    void addRequiredPrivileges(const std::string& dbname,
+                               const BSONObj& cmdObj,
+                               std::vector<Privilege>* out) override {
         ActionSet actions;
         actions.addAction(ActionType::collStats);
         out->push_back(Privilege(parseResourcePattern(dbname, cmdObj), actions));
     }
 
-    virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
+    bool supportsWriteConcern(const BSONObj& cmd) const override {
         return false;
     }
 
     bool run(OperationContext* opCtx,
              const string& dbName,
              BSONObj& cmdObj,
-             int,
+             int options,
              string& errmsg,
-             BSONObjBuilder& result) {
+             BSONObjBuilder& result) override {
         const NamespaceString nss(parseNsCollectionRequired(dbName, cmdObj));
 
-        auto conf = uassertStatusOK(Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, dbName));
-        if (!conf->isSharded(nss.ns())) {
+        auto routingInfo =
+            uassertStatusOK(Grid::get(opCtx)->catalogCache()->getCollectionRoutingInfo(opCtx, nss));
+        if (!routingInfo.cm()) {
             result.appendBool("sharded", false);
-            result.append("primary", conf->getPrimaryId().toString());
-
-            return passthrough(opCtx, conf.get(), cmdObj, result);
+            result.append("primary", routingInfo.primaryId().toString());
+            return passthrough(opCtx, dbName, routingInfo.primaryId(), cmdObj, result);
         }
 
-        result.appendBool("sharded", true);
+        const auto cm = routingInfo.cm();
 
-        shared_ptr<ChunkManager> cm = conf->getChunkManager(opCtx, nss.ns());
-        massert(12594, "how could chunk manager be null!", cm);
+        result.appendBool("sharded", true);
 
         BSONObjBuilder shardStats;
         map<string, long long> counts;
@@ -860,35 +846,38 @@ public:
 class DataSizeCmd : public PublicGridCommand {
 public:
     DataSizeCmd() : PublicGridCommand("dataSize", "datasize") {}
-    virtual string parseNs(const string& dbname, const BSONObj& cmdObj) const override {
+
+    std::string parseNs(const std::string& dbname, const BSONObj& cmdObj) const override {
         return parseNsFullyQualified(dbname, cmdObj);
     }
-    virtual void addRequiredPrivileges(const std::string& dbname,
-                                       const BSONObj& cmdObj,
-                                       std::vector<Privilege>* out) {
+
+    void addRequiredPrivileges(const std::string& dbname,
+                               const BSONObj& cmdObj,
+                               std::vector<Privilege>* out) override {
         ActionSet actions;
         actions.addAction(ActionType::find);
         out->push_back(Privilege(parseResourcePattern(dbname, cmdObj), actions));
     }
-    virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
+
+    bool supportsWriteConcern(const BSONObj& cmd) const override {
         return false;
     }
+
     bool run(OperationContext* opCtx,
              const string& dbName,
              BSONObj& cmdObj,
-             int,
+             int options,
              string& errmsg,
-             BSONObjBuilder& result) {
-        const string fullns = parseNs(dbName, cmdObj);
-        const string nsDBName = nsToDatabase(fullns);
+             BSONObjBuilder& result) override {
+        const NamespaceString nss(parseNs(dbName, cmdObj));
 
-        auto conf = uassertStatusOK(Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, nsDBName));
-        if (!conf->isSharded(fullns)) {
-            return passthrough(opCtx, conf.get(), cmdObj, result);
+        auto routingInfo =
+            uassertStatusOK(Grid::get(opCtx)->catalogCache()->getCollectionRoutingInfo(opCtx, nss));
+        if (!routingInfo.cm()) {
+            return passthrough(opCtx, dbName, routingInfo.primaryId(), cmdObj, result);
         }
 
-        shared_ptr<ChunkManager> cm = conf->getChunkManager(opCtx, fullns);
-        massert(13407, "how could chunk manager be null!", cm);
+        const auto cm = routingInfo.cm();
 
         BSONObj min = cmdObj.getObjectField("min");
         BSONObj max = cmdObj.getObjectField("max");
@@ -919,13 +908,12 @@ public:
         for (const ShardId& shardId : shardIds) {
             const auto shardStatus = Grid::get(opCtx)->shardRegistry()->getShard(opCtx, shardId);
             if (!shardStatus.isOK()) {
-                invariant(shardStatus.getStatus() == ErrorCodes::ShardNotFound);
                 continue;
             }
 
             ScopedDbConnection conn(shardStatus.getValue()->getConnString());
             BSONObj res;
-            bool ok = conn->runCommand(conf->name(), cmdObj, res);
+            bool ok = conn->runCommand(dbName, cmdObj, res);
             conn.done();
 
             if (!ok) {
@@ -949,19 +937,20 @@ public:
 class ConvertToCappedCmd : public NotAllowedOnShardedCollectionCmd {
 public:
     ConvertToCappedCmd() : NotAllowedOnShardedCollectionCmd("convertToCapped") {}
-    virtual void addRequiredPrivileges(const std::string& dbname,
-                                       const BSONObj& cmdObj,
-                                       std::vector<Privilege>* out) {
+
+    void addRequiredPrivileges(const std::string& dbname,
+                               const BSONObj& cmdObj,
+                               std::vector<Privilege>* out) override {
         ActionSet actions;
         actions.addAction(ActionType::convertToCapped);
         out->push_back(Privilege(parseResourcePattern(dbname, cmdObj), actions));
     }
 
-    virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
+    bool supportsWriteConcern(const BSONObj& cmd) const override {
         return true;
     }
 
-    virtual std::string parseNs(const std::string& dbname, const BSONObj& cmdObj) const {
+    std::string parseNs(const std::string& dbname, const BSONObj& cmdObj) const override {
         return parseNsCollectionRequired(dbname, cmdObj).ns();
     }
 
@@ -970,23 +959,24 @@ public:
 class GroupCmd : public NotAllowedOnShardedCollectionCmd {
 public:
     GroupCmd() : NotAllowedOnShardedCollectionCmd("group") {}
-    virtual void addRequiredPrivileges(const std::string& dbname,
-                                       const BSONObj& cmdObj,
-                                       std::vector<Privilege>* out) {
+
+    void addRequiredPrivileges(const std::string& dbname,
+                               const BSONObj& cmdObj,
+                               std::vector<Privilege>* out) override {
         ActionSet actions;
         actions.addAction(ActionType::find);
         out->push_back(Privilege(parseResourcePattern(dbname, cmdObj), actions));
     }
 
-    virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
+    bool supportsWriteConcern(const BSONObj& cmd) const override {
         return false;
     }
 
-    virtual bool passOptions() const {
+    bool passOptions() const override {
         return true;
     }
 
-    virtual std::string parseNs(const std::string& dbname, const BSONObj& cmdObj) const {
+    std::string parseNs(const std::string& dbname, const BSONObj& cmdObj) const override {
         const auto nsElt = cmdObj.firstElement().embeddedObjectUserCheck()["ns"];
         uassert(ErrorCodes::InvalidNamespace,
                 "'ns' must be of type String",
@@ -1003,7 +993,7 @@ public:
                    const BSONObj& cmdObj,
                    ExplainCommon::Verbosity verbosity,
                    const rpc::ServerSelectionMetadata& serverSelectionMetadata,
-                   BSONObjBuilder* out) const {
+                   BSONObjBuilder* out) const override {
         // We will time how long it takes to run the commands on the shards.
         Timer timer;
 
@@ -1019,36 +1009,17 @@ public:
 
         const NamespaceString nss(parseNs(dbname, cmdObj));
 
-        // Note that this implementation will not handle targeting retries and fails when the
-        // sharding metadata is too stale
-        auto status = Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, nss.db());
-        if (!status.isOK()) {
-            return Status(status.getStatus().code(),
-                          str::stream() << "Passthrough command failed: " << command.toString()
-                                        << " on ns "
-                                        << nss.ns()
-                                        << ". Caused by "
-                                        << causedBy(status.getStatus()));
-        }
-
-        shared_ptr<DBConfig> conf = status.getValue();
-        if (conf->isSharded(nss.ns())) {
-            return Status(ErrorCodes::IllegalOperation,
-                          str::stream() << "Passthrough command failed: " << command.toString()
-                                        << " on ns "
-                                        << nss.ns()
-                                        << ". Cannot run on sharded namespace.");
-        }
-
-        const auto primaryShardStatus =
-            Grid::get(opCtx)->shardRegistry()->getShard(opCtx, conf->getPrimaryId());
-        if (!primaryShardStatus.isOK()) {
-            return primaryShardStatus.getStatus();
-        }
+        auto routingInfo =
+            uassertStatusOK(Grid::get(opCtx)->catalogCache()->getCollectionRoutingInfo(opCtx, nss));
+        uassert(ErrorCodes::IllegalOperation,
+                str::stream() << "Passthrough command failed: " << command.toString() << " on ns "
+                              << nss.ns()
+                              << ". Cannot run on sharded namespace.",
+                !routingInfo.cm());
 
         BSONObj shardResult;
         try {
-            ShardConnection conn(primaryShardStatus.getValue()->getConnString(), "");
+            ShardConnection conn(routingInfo.primary()->getConnString(), "");
 
             // TODO: this can throw a stale config when mongos is not up-to-date -- fix.
             if (!conn->runCommand(nss.db().toString(), command, shardResult, options)) {
@@ -1060,6 +1031,7 @@ public:
                                             << "; result: "
                                             << shardResult);
             }
+
             conn.done();
         } catch (const DBException& ex) {
             return ex.toStatus();
@@ -1067,9 +1039,9 @@ public:
 
         // Fill out the command result.
         Strategy::CommandResult cmdResult;
-        cmdResult.shardTargetId = conf->getPrimaryId();
+        cmdResult.shardTargetId = routingInfo.primaryId();
         cmdResult.result = shardResult;
-        cmdResult.target = primaryShardStatus.getValue()->getConnString();
+        cmdResult.target = routingInfo.primary()->getConnString();
 
         return ClusterExplain::buildExplainResult(
             opCtx, {cmdResult}, ClusterExplain::kSingleShard, timer.millis(), out);
@@ -1080,15 +1052,18 @@ public:
 class SplitVectorCmd : public NotAllowedOnShardedCollectionCmd {
 public:
     SplitVectorCmd() : NotAllowedOnShardedCollectionCmd("splitVector") {}
-    virtual bool passOptions() const {
+
+    bool passOptions() const override {
         return true;
     }
-    virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
+
+    bool supportsWriteConcern(const BSONObj& cmd) const override {
         return false;
     }
-    virtual Status checkAuthForCommand(Client* client,
-                                       const std::string& dbname,
-                                       const BSONObj& cmdObj) {
+
+    Status checkAuthForCommand(Client* client,
+                               const std::string& dbname,
+                               const BSONObj& cmdObj) override {
         if (!AuthorizationSession::get(client)->isAuthorizedForActionsOnResource(
                 ResourcePattern::forExactNamespace(NamespaceString(parseNs(dbname, cmdObj))),
                 ActionType::splitVector)) {
@@ -1096,43 +1071,49 @@ public:
         }
         return Status::OK();
     }
-    virtual bool run(OperationContext* opCtx,
-                     const string& dbName,
-                     BSONObj& cmdObj,
-                     int options,
-                     string& errmsg,
-                     BSONObjBuilder& result) {
-        string x = parseNs(dbName, cmdObj);
-        if (!str::startsWith(x, dbName)) {
-            errmsg = str::stream() << "doing a splitVector across dbs isn't supported via mongos";
-            return false;
-        }
+
+    std::string parseNs(const string& dbname, const BSONObj& cmdObj) const override {
+        return parseNsFullyQualified(dbname, cmdObj);
+    }
+
+    bool run(OperationContext* opCtx,
+             const string& dbName,
+             BSONObj& cmdObj,
+             int options,
+             string& errmsg,
+             BSONObjBuilder& result) override {
+        const std::string ns = parseNs(dbName, cmdObj);
+        uassert(ErrorCodes::IllegalOperation,
+                "Performing splitVector across dbs isn't supported via mongos",
+                str::startsWith(ns, dbName));
+
         return NotAllowedOnShardedCollectionCmd::run(
             opCtx, dbName, cmdObj, options, errmsg, result);
     }
-    virtual std::string parseNs(const string& dbname, const BSONObj& cmdObj) const {
-        return parseNsFullyQualified(dbname, cmdObj);
-    }
 
 } splitVectorCmd;
 
 class DistinctCmd : public PublicGridCommand {
 public:
     DistinctCmd() : PublicGridCommand("distinct") {}
-    virtual void help(stringstream& help) const {
+
+    void help(stringstream& help) const override {
         help << "{ distinct : 'collection name' , key : 'a.b' , query : {} }";
     }
-    virtual bool passOptions() const {
+
+    bool passOptions() const override {
         return true;
     }
-    virtual void addRequiredPrivileges(const std::string& dbname,
-                                       const BSONObj& cmdObj,
-                                       std::vector<Privilege>* out) {
+
+    void addRequiredPrivileges(const std::string& dbname,
+                               const BSONObj& cmdObj,
+                               std::vector<Privilege>* out) override {
         ActionSet actions;
         actions.addAction(ActionType::find);
         out->push_back(Privilege(parseResourcePattern(dbname, cmdObj), actions));
     }
-    virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
+
+    bool supportsWriteConcern(const BSONObj& cmd) const override {
         return false;
     }
 
@@ -1141,18 +1122,13 @@ public:
              BSONObj& cmdObj,
              int options,
              string& errmsg,
-             BSONObjBuilder& result) {
+             BSONObjBuilder& result) override {
         const NamespaceString nss(parseNsCollectionRequired(dbName, cmdObj));
 
-        auto status = Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, dbName);
-        if (!status.isOK()) {
-            return appendEmptyResultSet(result, status.getStatus(), nss.ns());
-        }
-
-        shared_ptr<DBConfig> conf = status.getValue();
-        if (!conf->isSharded(nss.ns())) {
-
-            if (passthrough(opCtx, conf.get(), cmdObj, options, result)) {
+        auto routingInfo =
+            uassertStatusOK(Grid::get(opCtx)->catalogCache()->getCollectionRoutingInfo(opCtx, nss));
+        if (!routingInfo.cm()) {
+            if (passthrough(opCtx, dbName, routingInfo.primaryId(), cmdObj, options, result)) {
                 return true;
             }
 
@@ -1192,10 +1168,9 @@ public:
             return false;
         }
 
-        shared_ptr<ChunkManager> cm = conf->getChunkManager(opCtx, nss.ns());
-        massert(10420, "how could chunk manager be null!", cm);
+        const auto cm = routingInfo.cm();
 
-        BSONObj query = getQuery(cmdObj);
+        auto query = getQuery(cmdObj);
         auto queryCollation = getCollation(cmdObj);
         if (!queryCollation.isOK()) {
             return appendEmptyResultSet(result, queryCollation.getStatus(), nss.ns());
@@ -1230,7 +1205,7 @@ public:
 
             ShardConnection conn(shardStatus.getValue()->getConnString(), nss.ns());
             BSONObj res;
-            bool ok = conn->runCommand(conf->name(), cmdObj, res, options);
+            bool ok = conn->runCommand(nss.db().toString(), cmdObj, res, options);
             conn.done();
 
             if (!ok) {
@@ -1340,16 +1315,18 @@ public:
         return ClusterExplain::buildExplainResult(
             opCtx, shardResults, mongosStageName, millisElapsed, out);
     }
+
 } disinctCmd;
 
 class FileMD5Cmd : public PublicGridCommand {
 public:
     FileMD5Cmd() : PublicGridCommand("filemd5") {}
-    virtual void help(stringstream& help) const {
+
+    void help(stringstream& help) const override {
         help << " example: { filemd5 : ObjectId(aaaaaaa) , root : \"fs\" }";
     }
 
-    virtual std::string parseNs(const std::string& dbname, const BSONObj& cmdObj) const {
+    std::string parseNs(const std::string& dbname, const BSONObj& cmdObj) const override {
         std::string collectionName;
         if (const auto rootElt = cmdObj["root"]) {
             uassert(ErrorCodes::InvalidNamespace,
@@ -1363,31 +1340,32 @@ public:
         return NamespaceString(dbname, collectionName).ns();
     }
 
-    virtual void addRequiredPrivileges(const std::string& dbname,
-                                       const BSONObj& cmdObj,
-                                       std::vector<Privilege>* out) {
+    void addRequiredPrivileges(const std::string& dbname,
+                               const BSONObj& cmdObj,
+                               std::vector<Privilege>* out) override {
         out->push_back(Privilege(parseResourcePattern(dbname, cmdObj), ActionType::find));
     }
 
-    virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
+    bool supportsWriteConcern(const BSONObj& cmd) const override {
         return false;
     }
 
     bool run(OperationContext* opCtx,
              const string& dbName,
              BSONObj& cmdObj,
-             int,
+             int options,
              string& errmsg,
-             BSONObjBuilder& result) {
+             BSONObjBuilder& result) override {
         const NamespaceString nss(parseNs(dbName, cmdObj));
 
-        auto conf = uassertStatusOK(Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, dbName));
-        if (!conf->isSharded(nss.ns())) {
-            return passthrough(opCtx, conf.get(), cmdObj, result);
+        auto routingInfo =
+            uassertStatusOK(Grid::get(opCtx)->catalogCache()->getCollectionRoutingInfo(opCtx, nss));
+        if (!routingInfo.cm()) {
+            return passthrough(opCtx, dbName, routingInfo.primaryId(), cmdObj, result);
         }
 
-        shared_ptr<ChunkManager> cm = conf->getChunkManager(opCtx, nss.ns());
-        massert(13091, "how could chunk manager be null!", cm);
+        const auto cm = routingInfo.cm();
+
         if (SimpleBSONObjComparator::kInstance.evaluate(cm->getShardKeyPattern().toBSON() ==
                                                         BSON("files_id" << 1))) {
             BSONObj finder = BSON("files_id" << cmdObj.firstElement());
@@ -1461,13 +1439,15 @@ public:
 
                     errmsg =
                         string("sharded filemd5 failed because: ") + res["errmsg"].valuestrsafe();
+
                     return false;
                 }
 
-                uassert(16246,
-                        "Shard " + conf->name() +
-                            " is too old to support GridFS sharded by {files_id:1, n:1}",
-                        res.hasField("md5state"));
+                uassert(
+                    16246,
+                    str::stream() << "Shard for database " << nss.db()
+                                  << " is too old to support GridFS sharded by {files_id:1, n:1}",
+                    res.hasField("md5state"));
 
                 lastResult = res;
                 int nNext = res["numChunks"].numberInt();
@@ -1497,20 +1477,24 @@ public:
 class Geo2dFindNearCmd : public PublicGridCommand {
 public:
     Geo2dFindNearCmd() : PublicGridCommand("geoNear") {}
-    void help(stringstream& h) const {
+
+    void help(stringstream& h) const override {
         h << "http://dochub.mongodb.org/core/geo#GeospatialIndexing-geoNearCommand";
     }
-    virtual bool passOptions() const {
+
+    bool passOptions() const override {
         return true;
     }
-    virtual void addRequiredPrivileges(const std::string& dbname,
-                                       const BSONObj& cmdObj,
-                                       std::vector<Privilege>* out) {
+
+    void addRequiredPrivileges(const std::string& dbname,
+                               const BSONObj& cmdObj,
+                               std::vector<Privilege>* out) override {
         ActionSet actions;
         actions.addAction(ActionType::find);
         out->push_back(Privilege(parseResourcePattern(dbname, cmdObj), actions));
     }
-    virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
+
+    bool supportsWriteConcern(const BSONObj& cmd) const override {
         return false;
     }
 
@@ -1519,16 +1503,16 @@ public:
              BSONObj& cmdObj,
              int options,
              string& errmsg,
-             BSONObjBuilder& result) {
+             BSONObjBuilder& result) override {
         const NamespaceString nss(parseNsCollectionRequired(dbName, cmdObj));
 
-        auto conf = uassertStatusOK(Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, dbName));
-        if (!conf->isSharded(nss.ns())) {
-            return passthrough(opCtx, conf.get(), cmdObj, options, result);
+        auto routingInfo =
+            uassertStatusOK(Grid::get(opCtx)->catalogCache()->getCollectionRoutingInfo(opCtx, nss));
+        if (!routingInfo.cm()) {
+            return passthrough(opCtx, dbName, routingInfo.primaryId(), cmdObj, result);
         }
 
-        shared_ptr<ChunkManager> cm = conf->getChunkManager(opCtx, nss.ns());
-        massert(13500, "how could chunk manager be null!", cm);
+        const auto cm = routingInfo.cm();
 
         BSONObj query = getQuery(cmdObj);
         auto collation = getCollation(cmdObj);
@@ -1628,64 +1612,76 @@ public:
 
         return true;
     }
+
 } geo2dFindNearCmd;
 
-class CompactCmd : public PublicGridCommand {
+class CompactCmd : public Command {
 public:
-    CompactCmd() : PublicGridCommand("compact") {}
-    virtual void addRequiredPrivileges(const std::string& dbname,
-                                       const BSONObj& cmdObj,
-                                       std::vector<Privilege>* out) {
+    CompactCmd() : Command("compact") {}
+
+    bool slaveOk() const override {
+        return true;
+    }
+
+    bool adminOnly() const override {
+        return false;
+    }
+
+    void addRequiredPrivileges(const std::string& dbname,
+                               const BSONObj& cmdObj,
+                               std::vector<Privilege>* out) override {
         ActionSet actions;
         actions.addAction(ActionType::compact);
         out->push_back(Privilege(parseResourcePattern(dbname, cmdObj), actions));
     }
-    virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
+
+    bool supportsWriteConcern(const BSONObj& cmd) const override {
         return false;
     }
-    virtual bool run(OperationContext* opCtx,
-                     const string& dbName,
-                     BSONObj& cmdObj,
-                     int,
-                     string& errmsg,
-                     BSONObjBuilder& result) {
-        errmsg = "compact not allowed through mongos";
-        return false;
+
+    bool run(OperationContext* opCtx,
+             const string& dbName,
+             BSONObj& cmdObj,
+             int options,
+             string& errmsg,
+             BSONObjBuilder& result) override {
+        uasserted(ErrorCodes::CommandNotSupported, "compact not allowed through mongos");
     }
+
 } compactCmd;
 
 class EvalCmd : public PublicGridCommand {
 public:
     EvalCmd() : PublicGridCommand("eval", "$eval") {}
+
     virtual void addRequiredPrivileges(const std::string& dbname,
                                        const BSONObj& cmdObj,
                                        std::vector<Privilege>* out) {
         // $eval can do pretty much anything, so require all privileges.
         RoleGraph::generateUniversalPrivileges(out);
     }
-    virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
+
+    bool supportsWriteConcern(const BSONObj& cmd) const override {
         return false;
     }
-    virtual bool run(OperationContext* opCtx,
-                     const string& dbName,
-                     BSONObj& cmdObj,
-                     int,
-                     string& errmsg,
-                     BSONObjBuilder& result) {
+
+    bool run(OperationContext* opCtx,
+             const string& dbName,
+             BSONObj& cmdObj,
+             int options,
+             string& errmsg,
+             BSONObjBuilder& result) override {
         RARELY {
             warning() << "the eval command is deprecated" << startupWarningsLog;
         }
 
-        // $eval isn't allowed to access sharded collections, but we need to leave the
-        // shard to detect that.
-        auto status = Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, dbName);
-        if (!status.isOK()) {
-            return appendCommandStatus(result, status.getStatus());
-        }
-
-        shared_ptr<DBConfig> conf = status.getValue();
-        return passthrough(opCtx, conf.get(), cmdObj, result);
+        // $eval isn't allowed to access sharded collections, but we need to leave the shard to
+        // detect that
+        const auto dbInfo =
+            uassertStatusOK(Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, dbName));
+        return passthrough(opCtx, dbName, dbInfo.primaryId(), cmdObj, result);
     }
+
 } evalCmd;
 
 class CmdListCollections final : public PublicGridCommand {
@@ -1711,7 +1707,7 @@ public:
                       str::stream() << "Not authorized to create users on db: " << dbname);
     }
 
-    virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
+    bool supportsWriteConcern(const BSONObj& cmd) const override {
         return false;
     }
 
@@ -1723,18 +1719,23 @@ public:
              BSONObjBuilder& result) final {
         auto nss = NamespaceString::makeListCollectionsNSS(dbName);
 
-        auto conf = Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, dbName);
-        if (!conf.isOK()) {
-            return appendEmptyResultSet(result, conf.getStatus(), dbName + ".$cmd.listCollections");
+        auto dbInfoStatus = Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, dbName);
+        if (!dbInfoStatus.isOK()) {
+            return appendEmptyResultSet(result, dbInfoStatus.getStatus(), nss.ns());
         }
 
-        return cursorCommandPassthrough(opCtx, conf.getValue(), cmdObj, nss, options, &result);
+        const auto& dbInfo = dbInfoStatus.getValue();
+
+        return cursorCommandPassthrough(
+            opCtx, dbName, dbInfo.primaryId(), cmdObj, nss, options, &result);
     }
+
 } cmdListCollections;
 
 class CmdListIndexes final : public PublicGridCommand {
 public:
     CmdListIndexes() : PublicGridCommand("listIndexes") {}
+
     virtual Status checkAuthForCommand(Client* client,
                                        const std::string& dbname,
                                        const BSONObj& cmdObj) {
@@ -1757,7 +1758,7 @@ public:
                                     << ns.coll());
     }
 
-    virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
+    bool supportsWriteConcern(const BSONObj& cmd) const override {
         return false;
     }
 
@@ -1767,18 +1768,15 @@ public:
              int options,
              string& errmsg,
              BSONObjBuilder& result) final {
-        auto conf = Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, dbName);
-        if (!conf.isOK()) {
-            return appendCommandStatus(result, conf.getStatus());
-        }
+        const NamespaceString nss(parseNsCollectionRequired(dbName, cmdObj));
+
+        const auto routingInfo =
+            uassertStatusOK(Grid::get(opCtx)->catalogCache()->getCollectionRoutingInfo(opCtx, nss));
 
-        const NamespaceString targetNss(parseNsCollectionRequired(dbName, cmdObj));
-        const NamespaceString commandNss =
-            NamespaceString::makeListIndexesNSS(targetNss.db(), targetNss.coll());
-        dassert(targetNss == commandNss.getTargetNSForListIndexes());
+        const auto commandNss = NamespaceString::makeListIndexesNSS(nss.db(), nss.coll());
 
         return cursorCommandPassthrough(
-            opCtx, conf.getValue(), cmdObj, commandNss, options, &result);
+            opCtx, nss.db(), routingInfo.primaryId(), cmdObj, commandNss, options, &result);
     }
 
 } cmdListIndexes;
diff --git a/src/mongo/s/commands/run_on_all_shards_cmd.cpp b/src/mongo/s/commands/run_on_all_shards_cmd.cpp
index b534bf0628a..881b7d654ab 100644
--- a/src/mongo/s/commands/run_on_all_shards_cmd.cpp
+++ b/src/mongo/s/commands/run_on_all_shards_cmd.cpp
@@ -36,12 +36,12 @@
 #include <set>
 
 #include "mongo/db/jsobj.h"
+#include "mongo/s/catalog_cache.h"
 #include "mongo/s/client/shard.h"
 #include "mongo/s/client/shard_registry.h"
 #include "mongo/s/commands/cluster_commands_common.h"
 #include "mongo/s/commands/sharded_command_processing.h"
 #include "mongo/s/grid.h"
-#include "mongo/s/sharding_raii.h"
 #include "mongo/util/log.h"
 
 namespace mongo {
@@ -80,7 +80,7 @@ bool RunOnAllShardsCommand::run(OperationContext* opCtx,
     LOG(1) << "RunOnAllShardsCommand db: " << dbName << " cmd:" << redact(cmdObj);
 
     if (_implicitCreateDb) {
-        uassertStatusOK(ScopedShardDatabase::getOrCreate(opCtx, dbName));
+        uassertStatusOK(createShardDatabase(opCtx, dbName));
     }
 
     std::vector<ShardId> shardIds;
diff --git a/src/mongo/s/commands/strategy.cpp b/src/mongo/s/commands/strategy.cpp
index 0182a091ab7..647e6601dfa 100644
--- a/src/mongo/s/commands/strategy.cpp
+++ b/src/mongo/s/commands/strategy.cpp
@@ -54,8 +54,6 @@
 #include "mongo/rpc/metadata/server_selection_metadata.h"
 #include "mongo/rpc/metadata/tracking_metadata.h"
 #include "mongo/s/catalog_cache.h"
-#include "mongo/s/chunk_manager.h"
-#include "mongo/s/chunk_version.h"
 #include "mongo/s/client/parallel.h"
 #include "mongo/s/client/shard_connection.h"
 #include "mongo/s/client/shard_registry.h"
@@ -365,11 +363,10 @@ void Strategy::clientCommandOp(OperationContext* opCtx,
 
             ShardConnection::checkMyConnectionVersions(opCtx, staleNS);
             if (loops < 4) {
-                // This throws out the entire database cache entry in response to
-                // StaleConfigException instead of just the collection which encountered it. There
-                // is no good reason for it other than the lack of lower-granularity cache
-                // invalidation.
-                Grid::get(opCtx)->catalogCache()->invalidate(NamespaceString(staleNS).db());
+                const NamespaceString nss(staleNS);
+                if (nss.isValid()) {
+                    Grid::get(opCtx)->catalogCache()->invalidateShardedCollection(nss);
+                }
             }
         } catch (const DBException& e) {
             OpQueryReplyBuilder reply;
diff --git a/src/mongo/s/config.cpp b/src/mongo/s/config.cpp
deleted file mode 100644
index f5aec193923..00000000000
--- a/src/mongo/s/config.cpp
+++ /dev/null
@@ -1,366 +0,0 @@
-/**
- *    Copyright (C) 2008-2015 MongoDB Inc.
- *
- *    This program is free software: you can redistribute it and/or  modify
- *    it under the terms of the GNU Affero General Public License, version 3,
- *    as published by the Free Software Foundation.
- *
- *    This program is distributed in the hope that it will be useful,
- *    but WITHOUT ANY WARRANTY; without even the implied warranty of
- *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *    GNU Affero General Public License for more details.
- *
- *    You should have received a copy of the GNU Affero General Public License
- *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
- *
- *    As a special exception, the copyright holders give permission to link the
- *    code of portions of this program with the OpenSSL library under certain
- *    conditions as described in each individual source file and distribute
- *    linked combinations including the program with the OpenSSL library. You
- *    must comply with the GNU Affero General Public License in all respects for
- *    all of the code used other than as permitted herein. If you modify file(s)
- *    with this exception, you may extend this exception to your version of the
- *    file(s), but you are not obligated to do so. If you do not wish to do so,
- *    delete this exception statement from your version. If you delete this
- *    exception statement from all source files in the program, then also delete
- *    it in the license file.
- */
-
-#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kSharding
-
-#include "mongo/platform/basic.h"
-
-#include "mongo/s/config.h"
-
-#include <vector>
-
-#include "mongo/db/lasterror.h"
-#include "mongo/db/operation_context.h"
-#include "mongo/db/query/collation/collator_factory_interface.h"
-#include "mongo/s/catalog/sharding_catalog_client.h"
-#include "mongo/s/catalog/type_chunk.h"
-#include "mongo/s/catalog/type_collection.h"
-#include "mongo/s/catalog/type_database.h"
-#include "mongo/s/catalog_cache.h"
-#include "mongo/s/chunk_manager.h"
-#include "mongo/s/chunk_version.h"
-#include "mongo/s/grid.h"
-#include "mongo/stdx/memory.h"
-#include "mongo/util/log.h"
-
-namespace mongo {
-
-struct CollectionInfo {
-    // The config server opTime at which the chunk manager below was loaded
-    const repl::OpTime configOpTime;
-
-    // The chunk manager
-    const std::shared_ptr<ChunkManager> cm;
-};
-
-DBConfig::DBConfig(const DatabaseType& dbt, repl::OpTime configOpTime)
-    : _name(dbt.getName()),
-      _shardingEnabled(dbt.getSharded()),
-      _primaryId(dbt.getPrimary()),
-      _configOpTime(std::move(configOpTime)) {}
-
-DBConfig::~DBConfig() = default;
-
-bool DBConfig::isSharded(const std::string& ns) {
-    stdx::lock_guard<stdx::mutex> lk(_lock);
-
-    return _collections.count(ns) > 0;
-}
-
-void DBConfig::markNSNotSharded(const std::string& ns) {
-    stdx::lock_guard<stdx::mutex> lk(_lock);
-
-    CollectionInfoMap::iterator it = _collections.find(ns);
-    if (it != _collections.end()) {
-        _collections.erase(it);
-    }
-}
-
-std::shared_ptr<ChunkManager> DBConfig::getChunkManagerIfExists(OperationContext* opCtx,
-                                                                const std::string& ns,
-                                                                bool shouldReload,
-                                                                bool forceReload) {
-    // Don't report exceptions here as errors in GetLastError
-    LastError::Disabled ignoreForGLE(&LastError::get(cc()));
-
-    try {
-        return getChunkManager(opCtx, ns, shouldReload, forceReload);
-    } catch (const DBException&) {
-        return nullptr;
-    }
-}
-
-std::shared_ptr<ChunkManager> DBConfig::getChunkManager(OperationContext* opCtx,
-                                                        const std::string& ns,
-                                                        bool shouldReload,
-                                                        bool forceReload) {
-    ChunkVersion oldVersion;
-    std::shared_ptr<ChunkManager> oldManager;
-
-    {
-        stdx::lock_guard<stdx::mutex> lk(_lock);
-
-        auto it = _collections.find(ns);
-
-        const bool earlyReload = (it == _collections.end()) && (shouldReload || forceReload);
-        if (earlyReload) {
-            // This is to catch cases where there this is a new sharded collection.
-            // Note: read the _reloadCount inside the _lock mutex, so _loadIfNeeded will always
-            // be forced to perform a reload.
-            const auto currentReloadIteration = _reloadCount.load();
-            _loadIfNeeded(opCtx, currentReloadIteration);
-
-            it = _collections.find(ns);
-        }
-
-        uassert(ErrorCodes::NamespaceNotSharded,
-                str::stream() << "Collection is not sharded: " << ns,
-                it != _collections.end());
-
-        const auto& ci = it->second;
-
-        if (!(shouldReload || forceReload) || earlyReload) {
-            return ci.cm;
-        }
-
-        if (ci.cm) {
-            oldManager = ci.cm;
-            oldVersion = ci.cm->getVersion();
-        }
-    }
-
-    // TODO: We need to keep this first one-chunk check in until we have a more efficient way of
-    // creating/reusing a chunk manager, as doing so requires copying the full set of chunks
-    // currently
-    std::vector<ChunkType> newestChunk;
-    if (oldVersion.isSet() && !forceReload) {
-        uassertStatusOK(Grid::get(opCtx)->catalogClient(opCtx)->getChunks(
-            opCtx,
-            BSON(ChunkType::ns(ns)),
-            BSON(ChunkType::DEPRECATED_lastmod() << -1),
-            1,
-            &newestChunk,
-            nullptr,
-            repl::ReadConcernLevel::kMajorityReadConcern));
-
-        if (!newestChunk.empty()) {
-            invariant(newestChunk.size() == 1);
-            ChunkVersion v = newestChunk[0].getVersion();
-            if (v.equals(oldVersion)) {
-                stdx::lock_guard<stdx::mutex> lk(_lock);
-
-                auto it = _collections.find(ns);
-                uassert(15885,
-                        str::stream() << "not sharded after reloading from chunks : " << ns,
-                        it != _collections.end());
-
-                const auto& ci = it->second;
-                return ci.cm;
-            }
-        }
-    } else if (!oldVersion.isSet()) {
-        warning() << "version 0 found when " << (forceReload ? "reloading" : "checking")
-                  << " chunk manager; collection '" << ns << "' initially detected as sharded";
-    }
-
-    std::unique_ptr<ChunkManager> tempChunkManager;
-
-    {
-        stdx::lock_guard<stdx::mutex> lll(_hitConfigServerLock);
-
-        if (!newestChunk.empty() && !forceReload) {
-            // If we have a target we're going for see if we've hit already
-            stdx::lock_guard<stdx::mutex> lk(_lock);
-
-            auto it = _collections.find(ns);
-
-            if (it != _collections.end()) {
-                const auto& ci = it->second;
-
-                ChunkVersion currentVersion = newestChunk[0].getVersion();
-
-                // Only reload if the version we found is newer than our own in the same epoch
-                if (currentVersion <= ci.cm->getVersion() &&
-                    ci.cm->getVersion().hasEqualEpoch(currentVersion)) {
-                    return ci.cm;
-                }
-            }
-        }
-
-        // Reload the chunk manager outside of the DBConfig's mutex so as to not block operations
-        // for different collections on the same database
-        tempChunkManager.reset(new ChunkManager(
-            NamespaceString(oldManager->getns()),
-            oldManager->getVersion().epoch(),
-            oldManager->getShardKeyPattern(),
-            oldManager->getDefaultCollator() ? oldManager->getDefaultCollator()->clone() : nullptr,
-            oldManager->isUnique()));
-        tempChunkManager->loadExistingRanges(opCtx, oldManager.get());
-
-        if (!tempChunkManager->numChunks()) {
-            // Maybe we're not sharded any more, so do a full reload
-            const auto currentReloadIteration = _reloadCount.load();
-
-            const bool successful = [&]() {
-                stdx::lock_guard<stdx::mutex> lk(_lock);
-                return _loadIfNeeded(opCtx, currentReloadIteration);
-            }();
-
-            // If we aren't successful loading the database entry, we don't want to keep the stale
-            // object around which has invalid data.
-            if (!successful) {
-                Grid::get(opCtx)->catalogCache()->invalidate(_name);
-            }
-
-            return getChunkManager(opCtx, ns);
-        }
-    }
-
-    stdx::lock_guard<stdx::mutex> lk(_lock);
-
-    auto it = _collections.find(ns);
-    uassert(14822,
-            str::stream() << "Collection " << ns << " became unsharded in the middle.",
-            it != _collections.end());
-
-    const auto& ci = it->second;
-
-    // Reset if our versions aren't the same
-    bool shouldReset = !tempChunkManager->getVersion().equals(ci.cm->getVersion());
-
-    // Also reset if we're forced to do so
-    if (!shouldReset && forceReload) {
-        shouldReset = true;
-        warning() << "chunk manager reload forced for collection '" << ns << "', config version is "
-                  << tempChunkManager->getVersion();
-    }
-
-    //
-    // LEGACY BEHAVIOR
-    //
-    // It's possible to get into a state when dropping collections when our new version is
-    // less than our prev version. Behave identically to legacy mongos, for now, and warn to
-    // draw attention to the problem.
-    //
-    // TODO: Assert in next version, to allow smooth upgrades
-    //
-
-    if (shouldReset && tempChunkManager->getVersion() < ci.cm->getVersion()) {
-        shouldReset = false;
-
-        warning() << "not resetting chunk manager for collection '" << ns << "', config version is "
-                  << tempChunkManager->getVersion() << " and "
-                  << "old version is " << ci.cm->getVersion();
-    }
-
-    // end legacy behavior
-
-    if (shouldReset) {
-        const auto cmOpTime = tempChunkManager->getConfigOpTime();
-
-        // The existing ChunkManager could have been updated since we last checked, so replace the
-        // existing chunk manager only if it is strictly newer.
-        if (cmOpTime > ci.cm->getConfigOpTime()) {
-            _collections.erase(ns);
-            auto emplacedEntryIt =
-                _collections.emplace(ns, CollectionInfo{cmOpTime, std::move(tempChunkManager)})
-                    .first;
-            return emplacedEntryIt->second.cm;
-        }
-    }
-
-    return ci.cm;
-}
-
-bool DBConfig::load(OperationContext* opCtx) {
-    const auto currentReloadIteration = _reloadCount.load();
-    stdx::lock_guard<stdx::mutex> lk(_lock);
-    return _loadIfNeeded(opCtx, currentReloadIteration);
-}
-
-bool DBConfig::_loadIfNeeded(OperationContext* opCtx, Counter reloadIteration) {
-    if (reloadIteration != _reloadCount.load()) {
-        return true;
-    }
-
-    const auto catalogClient = Grid::get(opCtx)->catalogClient(opCtx);
-
-    auto status = catalogClient->getDatabase(opCtx, _name);
-    if (status == ErrorCodes::NamespaceNotFound) {
-        return false;
-    }
-
-    // All other errors are connectivity, etc so throw an exception.
-    uassertStatusOK(status.getStatus());
-
-    const auto& dbOpTimePair = status.getValue();
-    const auto& dbt = dbOpTimePair.value;
-    invariant(_name == dbt.getName());
-    _primaryId = dbt.getPrimary();
-
-    invariant(dbOpTimePair.opTime >= _configOpTime);
-    _configOpTime = dbOpTimePair.opTime;
-
-    // Load all collections
-    std::vector<CollectionType> collections;
-    repl::OpTime configOpTimeWhenLoadingColl;
-    uassertStatusOK(
-        catalogClient->getCollections(opCtx, &_name, &collections, &configOpTimeWhenLoadingColl));
-
-    invariant(configOpTimeWhenLoadingColl >= _configOpTime);
-
-    for (const auto& coll : collections) {
-        auto collIter = _collections.find(coll.getNs().ns());
-        if (collIter != _collections.end()) {
-            invariant(configOpTimeWhenLoadingColl >= collIter->second.configOpTime);
-        }
-
-        _collections.erase(coll.getNs().ns());
-
-        if (!coll.getDropped()) {
-            std::unique_ptr<CollatorInterface> defaultCollator;
-            if (!coll.getDefaultCollation().isEmpty()) {
-                auto statusWithCollator = CollatorFactoryInterface::get(opCtx->getServiceContext())
-                                              ->makeFromBSON(coll.getDefaultCollation());
-
-                // The collation was validated upon collection creation.
-                invariantOK(statusWithCollator.getStatus());
-
-                defaultCollator = std::move(statusWithCollator.getValue());
-            }
-
-            std::unique_ptr<ChunkManager> manager(
-                stdx::make_unique<ChunkManager>(coll.getNs(),
-                                                coll.getEpoch(),
-                                                ShardKeyPattern(coll.getKeyPattern()),
-                                                std::move(defaultCollator),
-                                                coll.getUnique()));
-
-            // Do the blocking collection load
-            manager->loadExistingRanges(opCtx, nullptr);
-
-            // Collections with no chunks are unsharded, no matter what the collections entry says
-            if (manager->numChunks()) {
-                _collections.emplace(
-                    coll.getNs().ns(),
-                    CollectionInfo{configOpTimeWhenLoadingColl, std::move(manager)});
-            }
-        }
-    }
-
-    _reloadCount.fetchAndAdd(1);
-
-    return true;
-}
-
-ShardId DBConfig::getPrimaryId() {
-    stdx::lock_guard<stdx::mutex> lk(_lock);
-    return _primaryId;
-}
-
-}  // namespace mongo
diff --git a/src/mongo/s/config.h b/src/mongo/s/config.h
deleted file mode 100644
index bbd63cf3b3b..00000000000
--- a/src/mongo/s/config.h
+++ /dev/null
@@ -1,146 +0,0 @@
-/**
- *    Copyright (C) 2008-2015 MongoDB Inc.
- *
- *    This program is free software: you can redistribute it and/or  modify
- *    it under the terms of the GNU Affero General Public License, version 3,
- *    as published by the Free Software Foundation.
- *
- *    This program is distributed in the hope that it will be useful,
- *    but WITHOUT ANY WARRANTY; without even the implied warranty of
- *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *    GNU Affero General Public License for more details.
- *
- *    You should have received a copy of the GNU Affero General Public License
- *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
- *
- *    As a special exception, the copyright holders give permission to link the
- *    code of portions of this program with the OpenSSL library under certain
- *    conditions as described in each individual source file and distribute
- *    linked combinations including the program with the OpenSSL library. You
- *    must comply with the GNU Affero General Public License in all respects for
- *    all of the code used other than as permitted herein. If you modify file(s)
- *    with this exception, you may extend this exception to your version of the
- *    file(s), but you are not obligated to do so. If you do not wish to do so,
- *    delete this exception statement from your version. If you delete this
- *    exception statement from all source files in the program, then also delete
- *    it in the license file.
- */
-
-#pragma once
-
-#include <string>
-
-#include "mongo/db/repl/optime.h"
-#include "mongo/platform/atomic_word.h"
-#include "mongo/s/shard_id.h"
-#include "mongo/stdx/mutex.h"
-
-namespace mongo {
-
-class ChunkManager;
-struct CollectionInfo;
-class DatabaseType;
-class OperationContext;
-
-/**
- * Represents the cache entry for a database.
- */
-class DBConfig {
-public:
-    DBConfig(const DatabaseType& dbt, repl::OpTime configOpTime);
-    ~DBConfig();
-
-    /**
-     * The name of the database which this entry caches.
-     */
-    const std::string& name() const {
-        return _name;
-    }
-
-    ShardId getPrimaryId();
-
-    /**
-     * Returns whether 'enableSharding' has been called for this database.
-     */
-    bool isShardingEnabled() const {
-        return _shardingEnabled;
-    }
-
-    /**
-     * Removes the specified namespace from the set of collections under this database entry so that
-     * from then onwards it will be treated as unsharded.
-     *
-     * Note that this method doesn't do any writes to the config metadata, but simply drops the
-     * specified namespace from the cache.
-     */
-    void markNSNotSharded(const std::string& ns);
-
-    /**
-     * @return whether or not the 'ns' collection is partitioned
-     */
-    bool isSharded(const std::string& ns);
-
-    std::shared_ptr<ChunkManager> getChunkManager(OperationContext* opCtx,
-                                                  const std::string& ns,
-                                                  bool reload = false,
-                                                  bool forceReload = false);
-    std::shared_ptr<ChunkManager> getChunkManagerIfExists(OperationContext* opCtx,
-                                                          const std::string& ns,
-                                                          bool reload = false,
-                                                          bool forceReload = false);
-
-    /**
-     * Returns true if it is successful at loading the DBConfig, false if the database is not found,
-     * and throws on all other errors.
-     */
-    bool load(OperationContext* opCtx);
-
-protected:
-    typedef std::map<std::string, CollectionInfo> CollectionInfoMap;
-    typedef AtomicUInt64::WordType Counter;
-
-    /**
-     * Returns true if it is successful at loading the DBConfig, false if the database is not found,
-     * and throws on all other errors.
-     * Also returns true without reloading if reloadIteration is not equal to the _reloadCount.
-     * This is to avoid multiple threads attempting to reload do duplicate work.
-     */
-    bool _loadIfNeeded(OperationContext* opCtx, Counter reloadIteration);
-
-    // All member variables are labeled with one of the following codes indicating the
-    // synchronization rules for accessing them.
-    //
-    // (L) Must hold _lock for access.
-    // (S) Self synchronizing, no explicit locking needed.
-    //
-    // Mutex lock order:
-    // _hitConfigServerLock -> _lock
-    //
-
-    // Name of the database which this entry caches
-    const std::string _name;
-
-    // Whether sharding is enabled for this database
-    const bool _shardingEnabled;
-
-    // Primary shard id
-    ShardId _primaryId;  // (L)
-
-    // Set of collections and lock to protect access
-    stdx::mutex _lock;
-    CollectionInfoMap _collections;  // (L)
-
-    // OpTime of config server when the database definition was loaded.
-    repl::OpTime _configOpTime;  // (L)
-
-    // Ensures that only one thread at a time loads collection configuration data from
-    // the config server
-    stdx::mutex _hitConfigServerLock;
-
-    // Increments every time this performs a full reload. Since a full reload can take a very
-    // long time for very large clusters, this can be used to minimize duplicate work when multiple
-    // threads tries to perform full rerload at roughly the same time.
-    AtomicUInt64 _reloadCount;  // (S)
-};
-
-}  // namespace mongo
diff --git a/src/mongo/s/query/cluster_find.cpp b/src/mongo/s/query/cluster_find.cpp
index d944954635a..3d4c384c506 100644
--- a/src/mongo/s/query/cluster_find.cpp
+++ b/src/mongo/s/query/cluster_find.cpp
@@ -52,7 +52,6 @@
 #include "mongo/s/query/cluster_client_cursor_impl.h"
 #include "mongo/s/query/cluster_cursor_manager.h"
 #include "mongo/s/query/store_possible_cursor.h"
-#include "mongo/s/sharding_raii.h"
 #include "mongo/s/stale_exception.h"
 #include "mongo/stdx/memory.h"
 #include "mongo/util/fail_point_service.h"
@@ -319,31 +318,34 @@ StatusWith<CursorId> ClusterFind::runQuery(OperationContext* opCtx,
                               << query.getQueryRequest().getProj()};
     }
 
+    auto const catalogCache = Grid::get(opCtx)->catalogCache();
+
     // Re-target and re-send the initial find command to the shards until we have established the
     // shard version.
     for (size_t retries = 1; retries <= kMaxStaleConfigRetries; ++retries) {
-        auto scopedCMStatus = ScopedChunkManager::get(opCtx, query.nss());
-        if (scopedCMStatus == ErrorCodes::NamespaceNotFound) {
+        auto routingInfoStatus = catalogCache->getCollectionRoutingInfo(opCtx, query.nss());
+        if (routingInfoStatus == ErrorCodes::NamespaceNotFound) {
             // If the database doesn't exist, we successfully return an empty result set without
             // creating a cursor.
             return CursorId(0);
-        } else if (!scopedCMStatus.isOK()) {
-            return scopedCMStatus.getStatus();
+        } else if (!routingInfoStatus.isOK()) {
+            return routingInfoStatus.getStatus();
         }
 
-        const auto& scopedCM = scopedCMStatus.getValue();
+        auto& routingInfo = routingInfoStatus.getValue();
 
         auto cursorId = runQueryWithoutRetrying(opCtx,
                                                 query,
                                                 readPref,
-                                                scopedCM.cm().get(),
-                                                scopedCM.primary(),
+                                                routingInfo.cm().get(),
+                                                routingInfo.primary(),
                                                 results,
                                                 viewDefinition);
         if (cursorId.isOK()) {
             return cursorId;
         }
-        auto status = std::move(cursorId.getStatus());
+
+        const auto& status = cursorId.getStatus();
 
         if (!ErrorCodes::isStaleShardingError(status.code()) &&
             status != ErrorCodes::ShardNotFound) {
@@ -357,11 +359,7 @@ StatusWith<CursorId> ClusterFind::runQuery(OperationContext* opCtx,
                << " on attempt " << retries << " of " << kMaxStaleConfigRetries << ": "
                << redact(status);
 
-        if (status == ErrorCodes::StaleEpoch) {
-            Grid::get(opCtx)->catalogCache()->invalidate(query.nss().db().toString());
-        } else {
-            scopedCM.db()->getChunkManagerIfExists(opCtx, query.nss().ns(), true);
-        }
+        catalogCache->onStaleConfigError(std::move(routingInfo));
     }
 
     return {ErrorCodes::StaleShardVersion,
diff --git a/src/mongo/s/sharding_raii.cpp b/src/mongo/s/sharding_raii.cpp
deleted file mode 100644
index b90f975ed35..00000000000
--- a/src/mongo/s/sharding_raii.cpp
+++ /dev/null
@@ -1,159 +0,0 @@
-/**
- *    Copyright (C) 2016 MongoDB Inc.
- *
- *    This program is free software: you can redistribute it and/or  modify
- *    it under the terms of the GNU Affero General Public License, version 3,
- *    as published by the Free Software Foundation.
- *
- *    This program is distributed in the hope that it will be useful,
- *    but WITHOUT ANY WARRANTY; without even the implied warranty of
- *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *    GNU Affero General Public License for more details.
- *
- *    You should have received a copy of the GNU Affero General Public License
- *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
- *
- *    As a special exception, the copyright holders give permission to link the
- *    code of portions of this program with the OpenSSL library under certain
- *    conditions as described in each individual source file and distribute
- *    linked combinations including the program with the OpenSSL library. You
- *    must comply with the GNU Affero General Public License in all respects for
- *    all of the code used other than as permitted herein. If you modify file(s)
- *    with this exception, you may extend this exception to your version of the
- *    file(s), but you are not obligated to do so. If you do not wish to do so,
- *    delete this exception statement from your version. If you delete this
- *    exception statement from all source files in the program, then also delete
- *    it in the license file.
- */
-
-#include "mongo/platform/basic.h"
-
-#include "mongo/s/sharding_raii.h"
-
-#include "mongo/base/status_with.h"
-#include "mongo/s/catalog/sharding_catalog_client.h"
-#include "mongo/s/catalog_cache.h"
-#include "mongo/s/chunk_manager.h"
-#include "mongo/s/client/shard_registry.h"
-#include "mongo/s/grid.h"
-
-namespace mongo {
-
-using std::shared_ptr;
-
-ScopedShardDatabase::ScopedShardDatabase(std::shared_ptr<DBConfig> db) : _db(db) {
-    invariant(_db);
-}
-
-ScopedShardDatabase::~ScopedShardDatabase() = default;
-
-StatusWith<ScopedShardDatabase> ScopedShardDatabase::getExisting(OperationContext* opCtx,
-                                                                 StringData dbName) {
-    auto dbStatus = Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, dbName.toString());
-    if (!dbStatus.isOK()) {
-        return {dbStatus.getStatus().code(),
-                str::stream() << "Database " << dbName << " was not found due to "
-                              << dbStatus.getStatus().toString()};
-    }
-
-    return {ScopedShardDatabase(std::move(dbStatus.getValue()))};
-}
-
-StatusWith<ScopedShardDatabase> ScopedShardDatabase::getOrCreate(OperationContext* opCtx,
-                                                                 StringData dbName) {
-    auto dbStatus = getExisting(opCtx, dbName);
-    if (dbStatus.isOK()) {
-        return dbStatus;
-    }
-
-    if (dbStatus == ErrorCodes::NamespaceNotFound) {
-        auto statusCreateDb =
-            Grid::get(opCtx)->catalogClient(opCtx)->createDatabase(opCtx, dbName.toString());
-        if (statusCreateDb.isOK() || statusCreateDb == ErrorCodes::NamespaceExists) {
-            return getExisting(opCtx, dbName);
-        }
-
-        return statusCreateDb;
-    }
-
-    return dbStatus.getStatus();
-}
-
-ScopedChunkManager::ScopedChunkManager(ScopedShardDatabase db, std::shared_ptr<ChunkManager> cm)
-    : _db(std::move(db)), _cm(std::move(cm)) {}
-
-ScopedChunkManager::ScopedChunkManager(ScopedShardDatabase db, std::shared_ptr<Shard> primary)
-    : _db(std::move(db)), _primary(std::move(primary)) {}
-
-ScopedChunkManager::~ScopedChunkManager() = default;
-
-StatusWith<ScopedChunkManager> ScopedChunkManager::get(OperationContext* opCtx,
-                                                       const NamespaceString& nss) {
-    auto scopedDbStatus = ScopedShardDatabase::getExisting(opCtx, nss.db());
-    if (!scopedDbStatus.isOK()) {
-        return scopedDbStatus.getStatus();
-    }
-
-    auto scopedDb = std::move(scopedDbStatus.getValue());
-
-    auto cm = scopedDb.db()->getChunkManagerIfExists(opCtx, nss.ns());
-    if (cm) {
-        return {ScopedChunkManager(std::move(scopedDb), std::move(cm))};
-    }
-
-    auto shardStatus =
-        Grid::get(opCtx)->shardRegistry()->getShard(opCtx, scopedDb.db()->getPrimaryId());
-    if (!shardStatus.isOK()) {
-        return {ErrorCodes::fromInt(40371),
-                str::stream() << "The primary shard for collection " << nss.ns()
-                              << " could not be loaded due to error "
-                              << shardStatus.getStatus().toString()};
-    }
-
-    return {ScopedChunkManager(std::move(scopedDb), std::move(shardStatus.getValue()))};
-}
-
-StatusWith<ScopedChunkManager> ScopedChunkManager::getOrCreate(OperationContext* opCtx,
-                                                               const NamespaceString& nss) {
-    auto scopedDbStatus = ScopedShardDatabase::getOrCreate(opCtx, nss.db());
-    if (!scopedDbStatus.isOK()) {
-        return scopedDbStatus.getStatus();
-    }
-
-    return ScopedChunkManager::get(opCtx, nss);
-}
-
-StatusWith<ScopedChunkManager> ScopedChunkManager::refreshAndGet(OperationContext* opCtx,
-                                                                 const NamespaceString& nss) {
-    auto scopedDbStatus = ScopedShardDatabase::getExisting(opCtx, nss.db());
-    if (!scopedDbStatus.isOK()) {
-        return scopedDbStatus.getStatus();
-    }
-
-    auto scopedDb = std::move(scopedDbStatus.getValue());
-
-    try {
-        std::shared_ptr<ChunkManager> cm =
-            scopedDb.db()->getChunkManager(opCtx, nss.ns(), true, false);
-
-        if (!cm) {
-            return {ErrorCodes::NamespaceNotSharded,
-                    str::stream() << "Collection " << nss.ns()
-                                  << " does not exist or is not sharded."};
-        }
-
-        if (cm->getChunkMap().empty()) {
-            return {ErrorCodes::NamespaceNotSharded,
-                    str::stream() << "Collection " << nss.ns()
-                                  << " is marked as sharded, but does not have any chunks. This "
-                                     "most likely indicates a corrupted metadata or "
-                                     "partially completed 'shardCollection' command."};
-        }
-
-        return {ScopedChunkManager(std::move(scopedDb), std::move(cm))};
-    } catch (const AssertionException& e) {
-        return e.toStatus();
-    }
-}
-
-}  // namespace mongo
diff --git a/src/mongo/s/sharding_raii.h b/src/mongo/s/sharding_raii.h
deleted file mode 100644
index 0c54f281985..00000000000
--- a/src/mongo/s/sharding_raii.h
+++ /dev/null
@@ -1,152 +0,0 @@
-/**
- *    Copyright (C) 2016 MongoDB Inc.
- *
- *    This program is free software: you can redistribute it and/or  modify
- *    it under the terms of the GNU Affero General Public License, version 3,
- *    as published by the Free Software Foundation.
- *
- *    This program is distributed in the hope that it will be useful,
- *    but WITHOUT ANY WARRANTY; without even the implied warranty of
- *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *    GNU Affero General Public License for more details.
- *
- *    You should have received a copy of the GNU Affero General Public License
- *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
- *
- *    As a special exception, the copyright holders give permission to link the
- *    code of portions of this program with the OpenSSL library under certain
- *    conditions as described in each individual source file and distribute
- *    linked combinations including the program with the OpenSSL library. You
- *    must comply with the GNU Affero General Public License in all respects for
- *    all of the code used other than as permitted herein. If you modify file(s)
- *    with this exception, you may extend this exception to your version of the
- *    file(s), but you are not obligated to do so. If you do not wish to do so,
- *    delete this exception statement from your version. If you delete this
- *    exception statement from all source files in the program, then also delete
- *    it in the license file.
- */
-
-#pragma once
-
-#include "mongo/base/disallow_copying.h"
-#include "mongo/s/chunk_manager.h"
-#include "mongo/s/config.h"
-
-namespace mongo {
-
-class OperationContext;
-
-class ScopedShardDatabase {
-    MONGO_DISALLOW_COPYING(ScopedShardDatabase);
-
-public:
-    ScopedShardDatabase(ScopedShardDatabase&&) = default;
-    ~ScopedShardDatabase();
-
-    /**
-     * Ensures that the specified database exists in the cache and if it does, returns it.
-     * Otherwise, either returns NamespaceNotFound if the database does not exist, or any other
-     * error code indicating why the database could not be loaded.
-     */
-    static StatusWith<ScopedShardDatabase> getExisting(OperationContext* opCtx, StringData dbName);
-
-    /**
-     * If the specified database exists already, loads it in the cache (if not already there) and
-     * returns it. Otherwise, if it does not exis, this call will implicitly create it as
-     * non-sharded.
-     */
-    static StatusWith<ScopedShardDatabase> getOrCreate(OperationContext* opCtx, StringData dbName);
-
-    /**
-     * Returns the underlying database cache entry.
-     */
-    DBConfig* db() const {
-        return _db.get();
-    }
-
-    /**
-     * This method is here only for compatibility with the legacy M/R code, which requires a shared
-     * reference to the underlying database. It should not be used in new code.
-     */
-    std::shared_ptr<DBConfig> getSharedDbReference() const {
-        return _db;
-    }
-
-private:
-    explicit ScopedShardDatabase(std::shared_ptr<DBConfig> db);
-
-    // Reference to the corresponding database. Never null.
-    std::shared_ptr<DBConfig> _db;
-};
-
-class ScopedChunkManager {
-    MONGO_DISALLOW_COPYING(ScopedChunkManager);
-
-public:
-    ScopedChunkManager(ScopedChunkManager&&) = default;
-    ~ScopedChunkManager();
-
-    /**
-     * If the specified namespace is sharded, returns a ScopedChunkManager initialized with that
-     * collection's routing information. If it is not, the object returned is initialized with the
-     * database primary node on which the unsharded collection must reside.
-     *
-     * Returns NamespaceNotFound if the database does not exist, or any other error indicating
-     * problem communicating with the config server.
-     */
-    static StatusWith<ScopedChunkManager> get(OperationContext* opCtx, const NamespaceString& nss);
-
-    /**
-     * If the database holding the specified namespace does not exist, creates it and then behaves
-     * like the 'get' method above.
-     */
-    static StatusWith<ScopedChunkManager> getOrCreate(OperationContext* opCtx,
-                                                      const NamespaceString& nss);
-
-    /**
-     * If the specified database and collection do not exist in the cache, tries to load them from
-     * the config server and returns a reference. If they are already in the cache, makes a call to
-     * the config server to check if there are any incremental updates to the collection chunk
-     * metadata and if so incorporates those. Otherwise, if it does not exist or any other error
-     * occurs, passes that error back.
-     */
-    static StatusWith<ScopedChunkManager> refreshAndGet(OperationContext* opCtx,
-                                                        const NamespaceString& nss);
-
-    /**
-     * Returns the underlying database for which we hold reference.
-     */
-    DBConfig* db() const {
-        return _db.db();
-    }
-
-    /**
-     * If the collection is sharded, returns a chunk manager for it. Otherwise, nullptr.
-     */
-    std::shared_ptr<ChunkManager> cm() const {
-        return _cm;
-    }
-
-    /**
-     * If the collection is not sharded, returns its primary shard. Otherwise, nullptr.
-     */
-    std::shared_ptr<Shard> primary() const {
-        return _primary;
-    }
-
-private:
-    ScopedChunkManager(ScopedShardDatabase db, std::shared_ptr<ChunkManager> cm);
-
-    ScopedChunkManager(ScopedShardDatabase db, std::shared_ptr<Shard> primary);
-
-    // Scoped reference to the owning database.
-    ScopedShardDatabase _db;
-
-    // Reference to the corresponding chunk manager (if sharded) or null
-    std::shared_ptr<ChunkManager> _cm;
-
-    // Reference to the primary of the database (if not sharded) or null
-    std::shared_ptr<Shard> _primary;
-};
-
-}  // namespace mongo