SERVER-28793 unit tests for ShardServerCatalogCacheLoader in primary mode

9 files changed, 771 insertions, 47 deletions

diff --git a/src/mongo/db/s/SConscript b/src/mongo/db/s/SConscript
index 1a9c825aee3..6481e53415e 100644
--- a/src/mongo/db/s/SConscript
+++ b/src/mongo/db/s/SConscript
@@ -233,9 +233,11 @@ env.CppUnitTest(
     target='shard_test',
     source=[
         'active_migrations_registry_test.cpp',
+        'catalog_cache_loader_mock.cpp',
         'migration_chunk_cloner_source_legacy_test.cpp',
         'namespace_metadata_change_notifications_test.cpp',
         'sharding_state_test.cpp',
+        'shard_server_catalog_cache_loader_test.cpp',
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/db/query/query_request',
diff --git a/src/mongo/db/s/catalog_cache_loader_mock.cpp b/src/mongo/db/s/catalog_cache_loader_mock.cpp
new file mode 100644
index 00000000000..d5dbe15c50e
--- /dev/null
+++ b/src/mongo/db/s/catalog_cache_loader_mock.cpp
@@ -0,0 +1,137 @@
+/**
+ *    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.
+ */
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/db/s/catalog_cache_loader_mock.h"
+
+#include "mongo/db/operation_context.h"
+#include "mongo/s/catalog/type_chunk.h"
+#include "mongo/s/catalog/type_collection.h"
+#include "mongo/stdx/thread.h"
+
+namespace mongo {
+
+using CollectionAndChangedChunks = CatalogCacheLoader::CollectionAndChangedChunks;
+
+namespace {
+
+/**
+ * Constructs the options for the loader thread pool.
+ */
+ThreadPool::Options makeDefaultThreadPoolOptions() {
+    ThreadPool::Options options;
+    options.poolName = "CatalogCacheLoaderMock";
+    options.minThreads = 0;
+    options.maxThreads = 1;
+
+    // Ensure all threads have a client.
+    options.onCreateThread = [](const std::string& threadName) {
+        Client::initThread(threadName.c_str());
+    };
+
+    return options;
+}
+
+}  // namespace
+
+CatalogCacheLoaderMock::CatalogCacheLoaderMock() : _threadPool(makeDefaultThreadPoolOptions()) {
+    _threadPool.startup();
+}
+
+CatalogCacheLoaderMock::~CatalogCacheLoaderMock() {
+    _threadPool.shutdown();
+    _threadPool.join();
+}
+
+void CatalogCacheLoaderMock::initializeReplicaSetRole(bool isPrimary) {
+    MONGO_UNREACHABLE;
+}
+
+void CatalogCacheLoaderMock::onStepDown() {
+    MONGO_UNREACHABLE;
+}
+
+void CatalogCacheLoaderMock::onStepUp() {
+    MONGO_UNREACHABLE;
+}
+
+void CatalogCacheLoaderMock::notifyOfCollectionVersionUpdate(const NamespaceString& nss) {
+    MONGO_UNREACHABLE;
+}
+
+Status CatalogCacheLoaderMock::waitForCollectionVersion(OperationContext* opCtx,
+                                                        const NamespaceString& nss,
+                                                        const ChunkVersion& version) {
+    MONGO_UNREACHABLE;
+}
+
+std::shared_ptr<Notification<void>> CatalogCacheLoaderMock::getChunksSince(
+    const NamespaceString& nss,
+    ChunkVersion version,
+    stdx::function<void(OperationContext*, StatusWith<CollectionAndChangedChunks>)> callbackFn) {
+
+    auto notify = std::make_shared<Notification<void>>();
+
+    uassertStatusOK(_threadPool.schedule([ this, notify, callbackFn ]() noexcept {
+        auto opCtx = Client::getCurrent()->makeOperationContext();
+
+        auto swCollAndChunks = [&]() -> StatusWith<CollectionAndChangedChunks> {
+            try {
+                uassertStatusOK(_swCollectionReturnValue);
+                uassertStatusOK(_swChunksReturnValue);
+
+                return CollectionAndChangedChunks{
+                    _swCollectionReturnValue.getValue().getEpoch(),
+                    _swCollectionReturnValue.getValue().getKeyPattern().toBSON(),
+                    _swCollectionReturnValue.getValue().getDefaultCollation(),
+                    _swCollectionReturnValue.getValue().getUnique(),
+                    _swChunksReturnValue.getValue()};
+            } catch (const DBException& ex) {
+                return ex.toStatus();
+            }
+        }();
+
+        callbackFn(opCtx.get(), std::move(swCollAndChunks));
+        notify->set();
+    }));
+
+    return notify;
+}
+
+void CatalogCacheLoaderMock::setCollectionRefreshReturnValue(
+    StatusWith<CollectionType> statusWithCollectionType) {
+    _swCollectionReturnValue = std::move(statusWithCollectionType);
+}
+
+void CatalogCacheLoaderMock::setChunkRefreshReturnValue(
+    StatusWith<std::vector<ChunkType>> statusWithChunks) {
+    _swChunksReturnValue = std::move(statusWithChunks);
+}
+
+}  // namespace mongo
diff --git a/src/mongo/db/s/catalog_cache_loader_mock.h b/src/mongo/db/s/catalog_cache_loader_mock.h
new file mode 100644
index 00000000000..0fb16efde8f
--- /dev/null
+++ b/src/mongo/db/s/catalog_cache_loader_mock.h
@@ -0,0 +1,88 @@
+/**
+ *    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 "mongo/s/catalog_cache_loader.h"
+#include "mongo/util/concurrency/thread_pool.h"
+
+namespace mongo {
+
+/**
+ * Mocks the metadata refresh results with settable return values. The purpose of this class is to
+ * facilitate testing of classes that use a CatalogCacheLoader.
+ */
+class CatalogCacheLoaderMock final : public CatalogCacheLoader {
+    MONGO_DISALLOW_COPYING(CatalogCacheLoaderMock);
+
+public:
+    CatalogCacheLoaderMock();
+    ~CatalogCacheLoaderMock();
+
+    /**
+     * These functions should never be called. They trigger invariants if called.
+     */
+    void initializeReplicaSetRole(bool isPrimary) override;
+    void onStepDown() override;
+    void onStepUp() override;
+    void notifyOfCollectionVersionUpdate(const NamespaceString& nss) override;
+    Status waitForCollectionVersion(OperationContext* opCtx,
+                                    const NamespaceString& nss,
+                                    const ChunkVersion& version) override;
+
+    std::shared_ptr<Notification<void>> getChunksSince(
+        const NamespaceString& nss,
+        ChunkVersion version,
+        stdx::function<void(OperationContext*, StatusWith<CollectionAndChangedChunks>)> callbackFn)
+        override;
+
+    /**
+     * Sets the mocked collection entry result that getChunksSince will use to construct its return
+     * value.
+     */
+    void setCollectionRefreshReturnValue(StatusWith<CollectionType> statusWithCollectionType);
+
+    /**
+     * Sets the mocked chunk results that getChunksSince will use to construct its return value.
+     */
+    void setChunkRefreshReturnValue(StatusWith<std::vector<ChunkType>> statusWithChunks);
+
+private:
+    // These variables hold the mocked chunks and collection entry results used to construct the
+    // return value of getChunksSince above.
+    StatusWith<CollectionType> _swCollectionReturnValue{Status(
+        ErrorCodes::InternalError, "config loader mock collection response is uninitialized")};
+
+    StatusWith<std::vector<ChunkType>> _swChunksReturnValue{
+        Status(ErrorCodes::InternalError, "config loader mock chunks response is uninitialized")};
+
+    // Thread pool on which to mock load chunk metadata.
+    ThreadPool _threadPool;
+};
+
+}  // namespace mongo
diff --git a/src/mongo/db/s/collection_sharding_state.cpp b/src/mongo/db/s/collection_sharding_state.cpp
index 23e6e2c5b6a..00576f2e00b 100644
--- a/src/mongo/db/s/collection_sharding_state.cpp
+++ b/src/mongo/db/s/collection_sharding_state.cpp
@@ -54,6 +54,7 @@
 #include "mongo/s/catalog/type_config_version.h"
 #include "mongo/s/catalog/type_shard.h"
 #include "mongo/s/catalog/type_shard_collection.h"
+#include "mongo/s/catalog_cache.h"
 #include "mongo/s/catalog_cache_loader.h"
 #include "mongo/s/chunk_version.h"
 #include "mongo/s/cluster_identity_loader.h"
@@ -94,14 +95,12 @@ private:
  */
 class CollectionVersionLogOpHandler final : public RecoveryUnit::Change {
 public:
-    CollectionVersionLogOpHandler(OperationContext* opCtx,
-                                  const NamespaceString& nss,
-                                  const ChunkVersion& updatedVersion)
-        : _opCtx(opCtx), _nss(nss), _updatedVersion(updatedVersion) {}
+    CollectionVersionLogOpHandler(OperationContext* opCtx, const NamespaceString& nss)
+        : _opCtx(opCtx), _nss(nss) {}
 
     void commit() override {
-        CatalogCacheLoader::get(_opCtx).notifyOfCollectionVersionUpdate(
-            _opCtx, _nss, _updatedVersion);
+        Grid::get(_opCtx)->catalogCache()->invalidateShardedCollection(_nss);
+        CatalogCacheLoader::get(_opCtx).notifyOfCollectionVersionUpdate(_nss);
     }
 
     void rollback() override {}
@@ -109,7 +108,6 @@ public:
 private:
     OperationContext* _opCtx;
     const NamespaceString _nss;
-    const ChunkVersion _updatedVersion;
 };
 
 }  // unnamed namespace
@@ -432,22 +430,8 @@ void CollectionShardingState::_onConfigRefreshCompleteInvalidateCachedMetadataAn
     // If 'lastRefreshedCollectionVersion' is present, then a refresh completed and the catalog
     // cache must be invalidated and the catalog cache loader notified of the new version.
     if (setField.hasField(ShardCollectionType::lastRefreshedCollectionVersion.name())) {
-        // Get the version epoch from the 'updatedDoc', since it didn't get updated and won't be
-        // in 'update'.
-        BSONElement oidElem;
-        fassert(40513,
-                bsonExtractTypedField(
-                    updatedDoc, ShardCollectionType::epoch(), BSONType::jstOID, &oidElem));
-
-        // Get the new collection version.
-        auto statusWithLastRefreshedChunkVersion = ChunkVersion::parseFromBSONWithFieldAndSetEpoch(
-            updatedDoc, ShardCollectionType::lastRefreshedCollectionVersion(), oidElem.OID());
-        fassert(40514, statusWithLastRefreshedChunkVersion.isOK());
-
         opCtx->recoveryUnit()->registerChange(
-            new CollectionVersionLogOpHandler(opCtx,
-                                              NamespaceString(refreshCollection),
-                                              statusWithLastRefreshedChunkVersion.getValue()));
+            new CollectionVersionLogOpHandler(opCtx, NamespaceString(refreshCollection)));
     }
 }
 
@@ -461,8 +445,8 @@ void CollectionShardingState::_onConfigDeleteInvalidateCachedMetadataAndNotify(
     fassertStatusOK(
         40479, bsonExtractStringField(query, ShardCollectionType::uuid.name(), &deletedCollection));
 
-    opCtx->recoveryUnit()->registerChange(new CollectionVersionLogOpHandler(
-        opCtx, NamespaceString(deletedCollection), ChunkVersion::UNSHARDED()));
+    opCtx->recoveryUnit()->registerChange(
+        new CollectionVersionLogOpHandler(opCtx, NamespaceString(deletedCollection)));
 }
 
 bool CollectionShardingState::_checkShardVersionOk(OperationContext* opCtx,
diff --git a/src/mongo/db/s/read_only_catalog_cache_loader.cpp b/src/mongo/db/s/read_only_catalog_cache_loader.cpp
index 007a6203489..46cd792ddb7 100644
--- a/src/mongo/db/s/read_only_catalog_cache_loader.cpp
+++ b/src/mongo/db/s/read_only_catalog_cache_loader.cpp
@@ -48,9 +48,7 @@ void ReadOnlyCatalogCacheLoader::onStepUp() {
     return;
 }
 
-void ReadOnlyCatalogCacheLoader::notifyOfCollectionVersionUpdate(OperationContext* opCtx,
-                                                                 const NamespaceString& nss,
-                                                                 const ChunkVersion& version) {
+void ReadOnlyCatalogCacheLoader::notifyOfCollectionVersionUpdate(const NamespaceString& nss) {
     return;
 }
 
diff --git a/src/mongo/db/s/read_only_catalog_cache_loader.h b/src/mongo/db/s/read_only_catalog_cache_loader.h
index 1615ba55c16..6485538c127 100644
--- a/src/mongo/db/s/read_only_catalog_cache_loader.h
+++ b/src/mongo/db/s/read_only_catalog_cache_loader.h
@@ -46,9 +46,7 @@ public:
     void initializeReplicaSetRole(bool isPrimary) override;
     void onStepDown() override;
     void onStepUp() override;
-    void notifyOfCollectionVersionUpdate(OperationContext* opCtx,
-                                         const NamespaceString& nss,
-                                         const ChunkVersion& version) override;
+    void notifyOfCollectionVersionUpdate(const NamespaceString& nss) override;
     Status waitForCollectionVersion(OperationContext* opCtx,
                                     const NamespaceString& nss,
                                     const ChunkVersion& version) override;
diff --git a/src/mongo/db/s/shard_server_catalog_cache_loader.cpp b/src/mongo/db/s/shard_server_catalog_cache_loader.cpp
index 59ee1da26c5..83596a1da3c 100644
--- a/src/mongo/db/s/shard_server_catalog_cache_loader.cpp
+++ b/src/mongo/db/s/shard_server_catalog_cache_loader.cpp
@@ -38,7 +38,6 @@
 #include "mongo/db/s/shard_metadata_util.h"
 #include "mongo/db/s/sharding_state.h"
 #include "mongo/s/catalog/type_shard_collection.h"
-#include "mongo/s/catalog_cache.h"
 #include "mongo/s/client/shard_registry.h"
 #include "mongo/s/grid.h"
 #include "mongo/stdx/memory.h"
@@ -286,11 +285,11 @@ ShardServerCatalogCacheLoader::~ShardServerCatalogCacheLoader() {
     invariant(_contexts.isEmpty());
 }
 
-void ShardServerCatalogCacheLoader::notifyOfCollectionVersionUpdate(OperationContext* opCtx,
-                                                                    const NamespaceString& nss,
-                                                                    const ChunkVersion& version) {
-    Grid::get(opCtx)->catalogCache()->invalidateShardedCollection(nss);
+void ShardServerCatalogCacheLoader::setForTesting() {
+    _testing = true;
+}
 
+void ShardServerCatalogCacheLoader::notifyOfCollectionVersionUpdate(const NamespaceString& nss) {
     _namespaceNotifications.notifyChange(nss);
 }
 
@@ -505,7 +504,6 @@ void ShardServerCatalogCacheLoader::_schedulePrimaryGetChunksSince(
                 }
             }
 
-
             if (swCollectionAndChangedChunks.isOK()) {
                 log() << "Cache loader remotely refreshed for collection " << nss
                       << " from collection version " << maxLoaderVersion
@@ -581,15 +579,15 @@ StatusWith<CollectionAndChangedChunks> ShardServerCatalogCacheLoader::_getLoader
     if (!tasksAreEnqueued) {
         // There are no tasks in the queue. Return the persisted metadata.
         return persisted;
-    } else if (enqueued.changedChunks.empty() || enqueued.epoch != persisted.epoch) {
-        // There is a task queue and either:
-        // - nothing was returned, which means the last task enqueued is a drop task.
+    } else if (persisted.changedChunks.empty() || enqueued.changedChunks.empty() ||
+               enqueued.epoch != persisted.epoch) {
+        // There is a task queue and:
+        // - nothing is persisted.
+        // - nothing was returned from enqueued, which means the last task enqueued is a drop task.
         // - the epoch changed in the enqueued metadata, which means there's a drop operation
         //   enqueued somewhere.
-        // Either way, the persisted metadata is out-dated. Return enqueued results.
-        return enqueued;
-    } else if (persisted.changedChunks.empty()) {
-        // Nothing is persisted. Return enqueued results.
+        // Whichever the cause, the persisted metadata is out-dated/non-existent. Return enqueued
+        // results.
         return enqueued;
     } else {
         // There can be overlap between persisted and enqueued metadata because enqueued work can
@@ -699,6 +697,10 @@ void ShardServerCatalogCacheLoader::_runTasks(const NamespaceString& nss) {
         _taskLists[nss].removeActiveTask();
     }
 
+    if (_testing) {
+        notifyOfCollectionVersionUpdate(nss);
+    }
+
     // Schedule more work if there is any
     if (!_taskLists[nss].empty()) {
         Status status = _threadPool.schedule([this, nss]() { _runTasks(nss); });
@@ -912,9 +914,22 @@ CollectionAndChangedChunks ShardServerCatalogCacheLoader::TaskList::getEnqueuedM
                 // the chunks vector. This will be either reset by the next task with a total reload
                 // with a new epoch, or cause the original getChunksSince caller to throw out the
                 // results and refresh again.
+
+                // Make sure we do not append a duplicate chunk. The diff query is GTE, so there can
+                // be duplicates of the same exact versioned chunk across tasks. This is no problem
+                // for our diff application algorithms, but it can return unpredictable numbers of
+                // chunks for testing purposes. Eliminate unpredicatable duplicates for testing
+                // stability.
+                auto taskCollectionAndChangedChunksIt =
+                    task.collectionAndChangedChunks->changedChunks.begin();
+                if (collAndChunks.changedChunks.back().getVersion() ==
+                    task.collectionAndChangedChunks->changedChunks.front().getVersion()) {
+                    ++taskCollectionAndChangedChunksIt;
+                }
+
                 collAndChunks.changedChunks.insert(
                     collAndChunks.changedChunks.end(),
-                    task.collectionAndChangedChunks->changedChunks.begin(),
+                    taskCollectionAndChangedChunksIt,
                     task.collectionAndChangedChunks->changedChunks.end());
             }
         }
diff --git a/src/mongo/db/s/shard_server_catalog_cache_loader.h b/src/mongo/db/s/shard_server_catalog_cache_loader.h
index 8eb863329c2..257791be7d0 100644
--- a/src/mongo/db/s/shard_server_catalog_cache_loader.h
+++ b/src/mongo/db/s/shard_server_catalog_cache_loader.h
@@ -51,6 +51,15 @@ public:
     ~ShardServerCatalogCacheLoader();
 
     /**
+     * For testing use only.
+     *
+     * Currently this only sets a boolean such that after metadata updates the notification system
+     * is signaled internally, rather than depending on the OpObservers which are not connectted for
+     * unit testing.
+     */
+    void setForTesting();
+
+    /**
      * Initializes internal state so that the loader behaves as a primary or secondary. This can
      * only be called once, when the sharding state is initialized.
      */
@@ -70,9 +79,7 @@ public:
      * Sets any notifications waiting for this version to arrive and invalidates the catalog cache's
      * chunk metadata for collection 'nss' so that the next caller provokes a refresh.
      */
-    void notifyOfCollectionVersionUpdate(OperationContext* opCtx,
-                                         const NamespaceString& nss,
-                                         const ChunkVersion& version) override;
+    void notifyOfCollectionVersionUpdate(const NamespaceString& nss) override;
 
     /**
      * Waits for the persisted collection version to be gte to 'version', or an epoch change. Only
@@ -355,6 +362,9 @@ private:
 
     // The collection of operation contexts in use by all threads.
     OperationContextGroup _contexts;
+
+    // Gates additional actions needed when testing.
+    bool _testing{false};
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/s/shard_server_catalog_cache_loader_test.cpp b/src/mongo/db/s/shard_server_catalog_cache_loader_test.cpp
new file mode 100644
index 00000000000..9ff8e837f71
--- /dev/null
+++ b/src/mongo/db/s/shard_server_catalog_cache_loader_test.cpp
@@ -0,0 +1,492 @@
+/**
+ * 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.
+ */
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/db/s/shard_server_catalog_cache_loader.h"
+
+#include "mongo/db/s/catalog_cache_loader_mock.h"
+#include "mongo/s/catalog/type_chunk.h"
+#include "mongo/s/catalog/type_collection.h"
+#include "mongo/s/shard_server_test_fixture.h"
+#include "mongo/unittest/unittest.h"
+
+namespace mongo {
+namespace {
+
+using executor::RemoteCommandRequest;
+using executor::RemoteCommandResponse;
+using std::vector;
+using unittest::assertGet;
+using CollectionAndChangedChunks = CatalogCacheLoader::CollectionAndChangedChunks;
+
+const NamespaceString kNss = NamespaceString("foo.bar");
+const std::string kPattern = "_id";
+const ShardId kShardId = ShardId("shard0");
+
+class ShardServerCatalogCacheLoaderTest : public ShardServerTestFixture {
+public:
+    /**
+     * Returns five chunks using collVersion as a starting version.
+     */
+    vector<ChunkType> makeFiveChunks(const ChunkVersion& collectionVersion);
+
+    /**
+     * Returns a chunk update diff GTE 'collVersion' for the chunks created by makeFiveChunks above.
+     */
+    vector<ChunkType> makeThreeUpdatedChunksDiff(const ChunkVersion& collectionVersion);
+
+    /**
+     * Returns a routing table applying 'threeUpdatedChunks' (the result of
+     * makeThreeUpdatedChunksDiff) to 'originalFiveChunks' (the result of makeFiveChunks).
+     */
+    vector<ChunkType> makeCombinedOriginalFiveChunksAndThreeNewChunksDiff(
+        const vector<ChunkType>& originalFiveChunks, const vector<ChunkType>& threeUpdatedChunks);
+
+    /**
+     * This helper makes a CollectionType with the current _maxCollVersion.
+     */
+    CollectionType makeCollectionType(const ChunkVersion& collVersion);
+
+    /**
+     * Sets up the _shardLoader with the results of makeFiveChunks().
+     */
+    vector<ChunkType> setUpChunkLoaderWithFiveChunks();
+
+    const KeyPattern kKeyPattern = KeyPattern(BSON(kPattern << 1));
+    const stdx::function<void(OperationContext*, StatusWith<CollectionAndChangedChunks>)>
+        kDoNothingCallbackFn = [](
+            OperationContext * opCtx,
+            StatusWith<CatalogCacheLoader::CollectionAndChangedChunks> swCollAndChunks) noexcept {};
+
+    CatalogCacheLoaderMock* _remoteLoaderMock;
+    std::unique_ptr<ShardServerCatalogCacheLoader> _shardLoader;
+
+private:
+    void setUp() override;
+    void tearDown() override;
+};
+
+void ShardServerCatalogCacheLoaderTest::setUp() {
+    ShardServerTestFixture::setUp();
+
+    // Create mock remote and real shard loader, retaining a pointer to the mock remote loader so
+    // that unit tests can manipulate it to return certain responses.
+    std::unique_ptr<CatalogCacheLoaderMock> mockLoader =
+        stdx::make_unique<CatalogCacheLoaderMock>();
+    _remoteLoaderMock = mockLoader.get();
+    _shardLoader = stdx::make_unique<ShardServerCatalogCacheLoader>(std::move(mockLoader));
+
+    // Set the shard loader to primary mode, and set it for testing.
+    _shardLoader->initializeReplicaSetRole(true);
+    _shardLoader->setForTesting();
+}
+
+void ShardServerCatalogCacheLoaderTest::tearDown() {
+    _shardLoader.reset();
+    ShardServerTestFixture::tearDown();
+}
+
+vector<ChunkType> ShardServerCatalogCacheLoaderTest::makeFiveChunks(
+    const ChunkVersion& collectionVersion) {
+    ChunkVersion collVersion(collectionVersion);
+    vector<ChunkType> chunks;
+
+    BSONObj mins[] = {
+        BSON("a" << MINKEY), BSON("a" << 10), BSON("a" << 50), BSON("a" << 100), BSON("a" << 200)};
+    BSONObj maxs[] = {
+        BSON("a" << 10), BSON("a" << 50), BSON("a" << 100), BSON("a" << 200), BSON("a" << MAXKEY)};
+
+    for (int i = 0; i < 5; ++i) {
+        collVersion.incMajor();
+
+        ChunkType chunk;
+        chunk.setNS(kNss.ns());
+        chunk.setMin(mins[i]);
+        chunk.setMax(maxs[i]);
+        chunk.setShard(kShardId);
+        chunk.setVersion(collVersion);
+
+        chunks.push_back(chunk);
+    }
+
+    return chunks;
+}
+
+vector<ChunkType> ShardServerCatalogCacheLoaderTest::makeThreeUpdatedChunksDiff(
+    const ChunkVersion& collectionVersion) {
+    ChunkVersion collVersion(collectionVersion);
+    vector<ChunkType> chunks;
+
+    // The diff query is for GTE a known version, so prepend the previous newest chunk, which is
+    // unmodified by this change and so should be found. Note: it is important for testing that the
+    // previous highest versioned chunk is unmodified. Otherwise the shard loader's results are
+    // dependent on a race between persistence and retrieving data because it combines enqueued and
+    // persisted results without applying modifications.
+    ChunkType oldChunk;
+    oldChunk.setNS(kNss.ns());
+    oldChunk.setMin(BSON("a" << 200));
+    oldChunk.setMax(BSON("a" << MAXKEY));
+    oldChunk.setShard(kShardId);
+    oldChunk.setVersion(collVersion);
+    chunks.push_back(oldChunk);
+
+
+    // Make chunk updates
+    BSONObj mins[] = {BSON("a" << MINKEY), BSON("a" << 5), BSON("a" << 10)};
+    BSONObj maxs[] = {BSON("a" << 5), BSON("a" << 10), BSON("a" << 100)};
+
+    for (int i = 0; i < 3; ++i) {
+        collVersion.incMinor();
+
+        ChunkType chunk;
+        chunk.setNS(kNss.ns());
+        chunk.setMin(mins[i]);
+        chunk.setMax(maxs[i]);
+        chunk.setShard(kShardId);
+        chunk.setVersion(collVersion);
+
+        chunks.push_back(chunk);
+    }
+
+    return chunks;
+}
+
+vector<ChunkType>
+ShardServerCatalogCacheLoaderTest::makeCombinedOriginalFiveChunksAndThreeNewChunksDiff(
+    const vector<ChunkType>& originalFiveChunks, const vector<ChunkType>& threeUpdatedChunksDiff) {
+    vector<ChunkType> chunks;
+
+    // Sorted by ascending chunk version, not range! Note, threeUpdatedChunksDiff already includes
+    // the last originalFiveChunks chunk because the diff query is GTE.
+    chunks.push_back(originalFiveChunks[3]);
+    chunks.insert(chunks.end(), threeUpdatedChunksDiff.begin(), threeUpdatedChunksDiff.end());
+
+    return chunks;
+}
+
+CollectionType ShardServerCatalogCacheLoaderTest::makeCollectionType(
+    const ChunkVersion& collVersion) {
+    CollectionType coll;
+    coll.setNs(kNss);
+    coll.setEpoch(collVersion.epoch());
+    coll.setUpdatedAt(Date_t::fromMillisSinceEpoch(collVersion.toLong()));
+    coll.setKeyPattern(kKeyPattern);
+    coll.setUnique(false);
+    return coll;
+}
+
+vector<ChunkType> ShardServerCatalogCacheLoaderTest::setUpChunkLoaderWithFiveChunks() {
+    ChunkVersion collectionVersion(1, 0, OID::gen());
+
+    CollectionType collectionType = makeCollectionType(collectionVersion);
+    vector<ChunkType> chunks = makeFiveChunks(collectionVersion);
+    _remoteLoaderMock->setCollectionRefreshReturnValue(collectionType);
+    _remoteLoaderMock->setChunkRefreshReturnValue(chunks);
+
+    StatusWith<CatalogCacheLoader::CollectionAndChangedChunks> results{
+        Status(ErrorCodes::InternalError, "")};
+    const auto refreshCallbackFn = [&results](
+        OperationContext * opCtx,
+        StatusWith<CatalogCacheLoader::CollectionAndChangedChunks> swCollAndChunks) noexcept {
+        results = std::move(swCollAndChunks);
+    };
+
+    auto notification =
+        _shardLoader->getChunksSince(kNss, ChunkVersion::UNSHARDED(), refreshCallbackFn);
+    notification->get();
+
+    // Check refreshCallbackFn thread results where we can safely throw.
+    ASSERT_OK(results.getStatus());
+    auto collAndChunkRes = results.getValue();
+    ASSERT_EQUALS(collAndChunkRes.epoch, collectionType.getEpoch());
+    ASSERT_EQUALS(collAndChunkRes.changedChunks.size(), 5UL);
+    for (unsigned int i = 0; i < collAndChunkRes.changedChunks.size(); ++i) {
+        ASSERT_BSONOBJ_EQ(collAndChunkRes.changedChunks[i].toConfigBSON(),
+                          chunks[i].toConfigBSON());
+    }
+
+    return chunks;
+}
+
+TEST_F(ShardServerCatalogCacheLoaderTest, PrimaryLoadFromUnshardedToUnsharded) {
+    // Return a NamespaceNotFound error that means the collection doesn't exist.
+
+    Status errorStatus = Status(ErrorCodes::NamespaceNotFound, "collection not found");
+    _remoteLoaderMock->setCollectionRefreshReturnValue(errorStatus);
+
+    StatusWith<CatalogCacheLoader::CollectionAndChangedChunks> results{
+        Status(ErrorCodes::InternalError, "")};
+    const auto refreshCallbackFn = [&results](
+        OperationContext * opCtx,
+        StatusWith<CatalogCacheLoader::CollectionAndChangedChunks> swCollAndChunks) noexcept {
+        results = std::move(swCollAndChunks);
+    };
+
+    auto notification =
+        _shardLoader->getChunksSince(kNss, ChunkVersion::UNSHARDED(), refreshCallbackFn);
+    notification->get();
+
+    ASSERT_EQUALS(results.getStatus(), errorStatus);
+}
+
+TEST_F(ShardServerCatalogCacheLoaderTest, PrimaryLoadFromShardedToUnsharded) {
+    // First set up the shard chunk loader as sharded.
+
+    auto chunks = setUpChunkLoaderWithFiveChunks();
+
+    // Then return a NamespaceNotFound error, which means the collection must have been dropped,
+    // clearing the chunk metadata.
+
+    Status errorStatus = Status(ErrorCodes::NamespaceNotFound, "collection not found");
+    _remoteLoaderMock->setCollectionRefreshReturnValue(errorStatus);
+
+    StatusWith<CatalogCacheLoader::CollectionAndChangedChunks> results{
+        Status(ErrorCodes::InternalError, "")};
+    const auto nextRefreshCallbackFn = [&results](
+        OperationContext * opCtx,
+        StatusWith<CatalogCacheLoader::CollectionAndChangedChunks> swCollAndChunks) noexcept {
+        results = std::move(swCollAndChunks);
+    };
+
+    auto notification =
+        _shardLoader->getChunksSince(kNss, chunks.back().getVersion(), nextRefreshCallbackFn);
+    notification->get();
+
+    ASSERT_EQUALS(results.getStatus(), errorStatus);
+}
+
+TEST_F(ShardServerCatalogCacheLoaderTest, PrimaryLoadFromShardedAndFindNoDiff) {
+    // First set up the shard chunk loader as sharded.
+
+    vector<ChunkType> chunks = setUpChunkLoaderWithFiveChunks();
+
+    // Then set up the remote loader to return a single document we've already seen -- indicates
+    // there's nothing new.
+
+    vector<ChunkType> lastChunk;
+    lastChunk.push_back(chunks.back());
+    _remoteLoaderMock->setChunkRefreshReturnValue(lastChunk);
+
+    StatusWith<CatalogCacheLoader::CollectionAndChangedChunks> results{
+        Status(ErrorCodes::InternalError, "")};
+    const auto refreshCallbackFn = [&results](
+        OperationContext * opCtx,
+        StatusWith<CatalogCacheLoader::CollectionAndChangedChunks> swCollAndChunks) noexcept {
+        results = std::move(swCollAndChunks);
+    };
+
+    auto notification =
+        _shardLoader->getChunksSince(kNss, chunks.back().getVersion(), refreshCallbackFn);
+    notification->get();
+
+    // Check that refreshing from the latest version returned a single document matching that
+    // version.
+    ASSERT_OK(results.getStatus());
+    auto collAndChunksRes = results.getValue();
+    ASSERT_EQUALS(collAndChunksRes.epoch, chunks.back().getVersion().epoch());
+    ASSERT_EQUALS(collAndChunksRes.changedChunks.size(), 1UL);
+    ASSERT_BSONOBJ_EQ(collAndChunksRes.changedChunks.back().toConfigBSON(),
+                      chunks.back().toConfigBSON());
+}
+
+// Same as the above unit test, PrimaryLoadFromShardedAndFindNoDiff, but caller requests complete
+// routing table, rather than diff from a known version.
+TEST_F(ShardServerCatalogCacheLoaderTest, PrimaryLoadFromShardedAndFindNoDiffRequestAll) {
+    // First set up the shard chunk loader as sharded.
+
+    vector<ChunkType> chunks = setUpChunkLoaderWithFiveChunks();
+
+    // Then set up the remote loader to return a single document we've already seen -- indicates
+    // there's nothing new.
+
+    vector<ChunkType> lastChunk;
+    lastChunk.push_back(chunks.back());
+    _remoteLoaderMock->setChunkRefreshReturnValue(lastChunk);
+
+    StatusWith<CatalogCacheLoader::CollectionAndChangedChunks> results{
+        Status(ErrorCodes::InternalError, "")};
+    const auto completeRefreshCallbackFn = [&results](
+        OperationContext * opCtx,
+        StatusWith<CatalogCacheLoader::CollectionAndChangedChunks> swCollAndChunks) noexcept {
+        results = std::move(swCollAndChunks);
+    };
+
+    auto notification =
+        _shardLoader->getChunksSince(kNss, ChunkVersion::UNSHARDED(), completeRefreshCallbackFn);
+    notification->get();
+
+    // Check that the complete routing table was returned successfully.
+    ASSERT_OK(results.getStatus());
+    auto collAndChunksRes = results.getValue();
+    ASSERT_EQUALS(collAndChunksRes.epoch, chunks.back().getVersion().epoch());
+    ASSERT_EQUALS(collAndChunksRes.changedChunks.size(), 5UL);
+    for (unsigned int i = 0; i < collAndChunksRes.changedChunks.size(); ++i) {
+        ASSERT_BSONOBJ_EQ(collAndChunksRes.changedChunks[i].getMin(), chunks[i].getMin());
+        ASSERT_BSONOBJ_EQ(collAndChunksRes.changedChunks[i].getMax(), chunks[i].getMax());
+        ASSERT_EQUALS(collAndChunksRes.changedChunks[i].getVersion(), chunks[i].getVersion());
+    }
+}
+
+TEST_F(ShardServerCatalogCacheLoaderTest, PrimaryLoadFromShardedAndFindDiff) {
+    // First set up the shard chunk loader as sharded.
+
+    vector<ChunkType> chunks = setUpChunkLoaderWithFiveChunks();
+
+    // Then refresh again and find updated chunks.
+
+    ChunkVersion collVersion = chunks.back().getVersion();
+    vector<ChunkType> updatedChunksDiff = makeThreeUpdatedChunksDiff(collVersion);
+    _remoteLoaderMock->setChunkRefreshReturnValue(updatedChunksDiff);
+
+    StatusWith<CatalogCacheLoader::CollectionAndChangedChunks> results{
+        Status(ErrorCodes::InternalError, "")};
+    const auto refreshCallbackFn = [&results](
+        OperationContext * opCtx,
+        StatusWith<CatalogCacheLoader::CollectionAndChangedChunks> swCollAndChunks) noexcept {
+        results = std::move(swCollAndChunks);
+    };
+
+    auto notification =
+        _shardLoader->getChunksSince(kNss, chunks.back().getVersion(), refreshCallbackFn);
+    notification->get();
+
+    // Check that the diff was returned successfull.
+    ASSERT_OK(results.getStatus());
+    auto collAndChunksRes = results.getValue();
+    ASSERT_EQUALS(collAndChunksRes.epoch, updatedChunksDiff.front().getVersion().epoch());
+    ASSERT_EQUALS(collAndChunksRes.changedChunks.size(), 4UL);
+    for (unsigned int i = 0; i < collAndChunksRes.changedChunks.size(); ++i) {
+        ASSERT_BSONOBJ_EQ(collAndChunksRes.changedChunks[i].getMin(),
+                          updatedChunksDiff[i].getMin());
+        ASSERT_BSONOBJ_EQ(collAndChunksRes.changedChunks[i].getMax(),
+                          updatedChunksDiff[i].getMax());
+        ASSERT_EQUALS(collAndChunksRes.changedChunks[i].getVersion(),
+                      updatedChunksDiff[i].getVersion());
+    }
+}
+
+TEST_F(ShardServerCatalogCacheLoaderTest, PrimaryLoadFromShardedAndFindDiffRequestAll) {
+    // First set up the shard chunk loader as sharded.
+
+    vector<ChunkType> chunks = setUpChunkLoaderWithFiveChunks();
+
+    // First cause a remote refresh to find the updated chunks. Then wait for persistence, so that
+    // we ensure that nothing is enqueued and the next getChunksSince call will return a predictable
+    // number of chunk documents: the result of applying the enqueued update diff.
+
+    vector<ChunkType> updatedChunksDiff = makeThreeUpdatedChunksDiff(chunks.back().getVersion());
+    _remoteLoaderMock->setChunkRefreshReturnValue(updatedChunksDiff);
+
+    auto notification =
+        _shardLoader->getChunksSince(kNss, chunks.back().getVersion(), kDoNothingCallbackFn);
+    notification->get();
+
+    // Wait for persistence of update
+    ASSERT_OK(_shardLoader->waitForCollectionVersion(
+        operationContext(), kNss, updatedChunksDiff.back().getVersion()));
+
+    // Set up the remote loader to return a single document we've already seen, indicating no change
+    // occurred.
+    vector<ChunkType> lastChunk;
+    lastChunk.push_back(updatedChunksDiff.back());
+    _remoteLoaderMock->setChunkRefreshReturnValue(lastChunk);
+
+    vector<ChunkType> completeRoutingTableWithDiffApplied =
+        makeCombinedOriginalFiveChunksAndThreeNewChunksDiff(chunks, updatedChunksDiff);
+
+    StatusWith<CatalogCacheLoader::CollectionAndChangedChunks> results{
+        Status(ErrorCodes::InternalError, "")};
+    const auto refreshCallbackFn = [&results](
+        OperationContext * opCtx,
+        StatusWith<CatalogCacheLoader::CollectionAndChangedChunks> swCollAndChunks) noexcept {
+        results = std::move(swCollAndChunks);
+    };
+
+    auto nextNotification =
+        _shardLoader->getChunksSince(kNss, ChunkVersion::UNSHARDED(), refreshCallbackFn);
+    nextNotification->get();
+
+    // Check that the complete routing table, with diff applied, was returned.
+    ASSERT_OK(results.getStatus());
+    auto collAndChunksRes = results.getValue();
+    ASSERT_EQUALS(collAndChunksRes.epoch,
+                  completeRoutingTableWithDiffApplied.front().getVersion().epoch());
+    ASSERT_EQUALS(collAndChunksRes.changedChunks.size(), 5UL);
+    for (unsigned int i = 0; i < collAndChunksRes.changedChunks.size(); ++i) {
+        ASSERT_BSONOBJ_EQ(collAndChunksRes.changedChunks[i].getMin(),
+                          completeRoutingTableWithDiffApplied[i].getMin());
+        ASSERT_BSONOBJ_EQ(collAndChunksRes.changedChunks[i].getMax(),
+                          completeRoutingTableWithDiffApplied[i].getMax());
+        ASSERT_EQUALS(collAndChunksRes.changedChunks[i].getVersion(),
+                      completeRoutingTableWithDiffApplied[i].getVersion());
+    }
+}
+
+TEST_F(ShardServerCatalogCacheLoaderTest, PrimaryLoadFromShardedAndFindNewEpoch) {
+    // First set up the shard chunk loader as sharded.
+
+    vector<ChunkType> chunks = setUpChunkLoaderWithFiveChunks();
+
+    // Then refresh again and find that the collection has been dropped and recreated.
+
+    ChunkVersion collVersionWithNewEpoch(1, 0, OID::gen());
+    CollectionType collectionTypeWithNewEpoch = makeCollectionType(collVersionWithNewEpoch);
+    vector<ChunkType> chunksWithNewEpoch = makeFiveChunks(collVersionWithNewEpoch);
+    _remoteLoaderMock->setCollectionRefreshReturnValue(collectionTypeWithNewEpoch);
+    _remoteLoaderMock->setChunkRefreshReturnValue(chunksWithNewEpoch);
+
+    StatusWith<CatalogCacheLoader::CollectionAndChangedChunks> results{
+        Status(ErrorCodes::InternalError, "")};
+    const auto refreshCallbackFn = [&results](
+        OperationContext * opCtx,
+        StatusWith<CatalogCacheLoader::CollectionAndChangedChunks> swCollAndChunks) noexcept {
+        results = std::move(swCollAndChunks);
+    };
+
+    auto notification =
+        _shardLoader->getChunksSince(kNss, chunks.back().getVersion(), refreshCallbackFn);
+    notification->get();
+
+    // Check that the complete routing table for the new epoch was returned.
+    ASSERT_OK(results.getStatus());
+    auto collAndChunksRes = results.getValue();
+    ASSERT_EQUALS(collAndChunksRes.epoch, collectionTypeWithNewEpoch.getEpoch());
+    ASSERT_EQUALS(collAndChunksRes.changedChunks.size(), 5UL);
+    for (unsigned int i = 0; i < collAndChunksRes.changedChunks.size(); ++i) {
+        ASSERT_BSONOBJ_EQ(collAndChunksRes.changedChunks[i].getMin(),
+                          chunksWithNewEpoch[i].getMin());
+        ASSERT_BSONOBJ_EQ(collAndChunksRes.changedChunks[i].getMax(),
+                          chunksWithNewEpoch[i].getMax());
+        ASSERT_EQUALS(collAndChunksRes.changedChunks[i].getVersion(),
+                      chunksWithNewEpoch[i].getVersion());
+    }
+}
+
+}  // namespace
+}  // namespace mongo