SERVER-43721 Make the AuthorizationManager use DistCache

The DistCache (to be later renamed to ReadThroughCache) was derived from
the same implementation under AuthorizationManager and this change
removes the code duplication.

In addition, it makes the following changes to InvalidatingLRUCache and
the DistCache:

* Simplifies and optimises the InvalidatingLRUCache:

The way it is implemented now, it performs up to 3 operations per
lookup, unvalidates entries unnecessarily and has overly complicated
logic, which is source of a crash. Instead of fixing the bug, this
change rewrites it in a simpler way, which introduces a ValueHandle
instead of bare shared_ptr for the return value, and only performs
additional work if entries fall off the underlying LRUCache.

* Moves the DistCache under src/util and adds unit tests:

This change pulls the DistCache (which is the main consumer of
InvalidatingLRUCache) into its own library and moves it to be
under src/util like the other caches and adds unit tests.

 delete mode 100644 jstests/auth/pinned_users.js
 create mode 100644 jstests/auth/pinned_users_clear_pinned_user_list.js
 create mode 100644 jstests/auth/pinned_users_exclusive_lock_on_admin.js
 create mode 100644 jstests/auth/pinned_users_remove_user_document_unpins_user.js
 create mode 100644 src/mongo/util/dist_cache.cpp
 rename src/mongo/{db => util}/dist_cache.h (56%)
 create mode 100644 src/mongo/util/dist_cache_test.cpp

17 files changed, 320 insertions, 695 deletions

diff --git a/src/mongo/db/auth/SConscript b/src/mongo/db/auth/SConscript
index 6ffcf7467ba..6c579a188eb 100644
--- a/src/mongo/db/auth/SConscript
+++ b/src/mongo/db/auth/SConscript
@@ -162,6 +162,7 @@ env.Library(
         '$BUILD_DIR/mongo/util/net/ssl_types',
     ],
     LIBDEPS_PRIVATE=[
+        '$BUILD_DIR/mongo/util/caching',
         '$BUILD_DIR/mongo/idl/server_parameter',
     ],
 )
diff --git a/src/mongo/db/auth/authorization_manager.cpp b/src/mongo/db/auth/authorization_manager.cpp
index b6c30368b2f..0a2eaf493f0 100644
--- a/src/mongo/db/auth/authorization_manager.cpp
+++ b/src/mongo/db/auth/authorization_manager.cpp
@@ -27,47 +27,17 @@
  *    it in the license file.
  */
 
-#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kAccessControl
-
 #include "mongo/platform/basic.h"
 
 #include "mongo/db/auth/authorization_manager.h"
 
-#include <memory>
-#include <string>
-#include <vector>
-
 #include "mongo/base/init.h"
 #include "mongo/base/shim.h"
-#include "mongo/base/status.h"
-#include "mongo/bson/mutable/document.h"
-#include "mongo/bson/mutable/element.h"
-#include "mongo/bson/util/bson_extract.h"
-#include "mongo/crypto/mechanism_scram.h"
-#include "mongo/db/auth/action_set.h"
-#include "mongo/db/auth/address_restriction.h"
-#include "mongo/db/auth/authorization_session.h"
-#include "mongo/db/auth/authz_manager_external_state.h"
-#include "mongo/db/auth/privilege.h"
-#include "mongo/db/auth/privilege_parser.h"
-#include "mongo/db/auth/role_graph.h"
-#include "mongo/db/auth/sasl_options.h"
-#include "mongo/db/auth/user.h"
-#include "mongo/db/auth/user_document_parser.h"
-#include "mongo/db/auth/user_name.h"
-#include "mongo/db/global_settings.h"
-#include "mongo/db/jsobj.h"
-#include "mongo/platform/compiler.h"
-#include "mongo/platform/mutex.h"
-#include "mongo/stdx/unordered_map.h"
-#include "mongo/util/assert_util.h"
-#include "mongo/util/log.h"
-#include "mongo/util/str.h"
-
-mongo::AuthInfo mongo::internalSecurity;
 
 namespace mongo {
 
+AuthInfo internalSecurity;
+
 constexpr StringData AuthorizationManager::USERID_FIELD_NAME;
 constexpr StringData AuthorizationManager::USER_NAME_FIELD_NAME;
 constexpr StringData AuthorizationManager::USER_DB_FIELD_NAME;
@@ -101,7 +71,6 @@ const int AuthorizationManager::schemaVersion26Upgrade;
 const int AuthorizationManager::schemaVersion26Final;
 const int AuthorizationManager::schemaVersion28SCRAM;
 
-
 std::unique_ptr<AuthorizationManager> AuthorizationManager::create() {
     static auto w = MONGO_WEAK_FUNCTION_DEFINITION(AuthorizationManager::create);
     return w();
diff --git a/src/mongo/db/auth/authorization_manager.h b/src/mongo/db/auth/authorization_manager.h
index 7f11e10739c..3ac6c7cc7ce 100644
--- a/src/mongo/db/auth/authorization_manager.h
+++ b/src/mongo/db/auth/authorization_manager.h
@@ -29,28 +29,18 @@
 
 #pragma once
 
-#include <functional>
-#include <memory>
-#include <string>
-
 #include <boost/optional.hpp>
+#include <memory>
 
-#include "mongo/base/secure_allocator.h"
 #include "mongo/base/status.h"
-#include "mongo/bson/mutable/element.h"
 #include "mongo/bson/oid.h"
 #include "mongo/db/auth/action_set.h"
 #include "mongo/db/auth/privilege_format.h"
 #include "mongo/db/auth/resource_pattern.h"
 #include "mongo/db/auth/role_graph.h"
 #include "mongo/db/auth/user.h"
-#include "mongo/db/auth/user_name.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/db/namespace_string.h"
-#include "mongo/db/server_options.h"
-#include "mongo/platform/mutex.h"
-#include "mongo/stdx/condition_variable.h"
-#include "mongo/stdx/unordered_map.h"
 
 namespace mongo {
 
@@ -58,7 +48,6 @@ class AuthorizationSession;
 class AuthzManagerExternalState;
 class OperationContext;
 class ServiceContext;
-class UserDocumentParser;
 
 /**
  * Internal secret key info.
@@ -91,11 +80,11 @@ public:
     static AuthorizationManager* get(ServiceContext& service);
     static void set(ServiceContext* service, std::unique_ptr<AuthorizationManager> authzManager);
 
-    virtual ~AuthorizationManager() = default;
+    static std::unique_ptr<AuthorizationManager> create();
 
     AuthorizationManager() = default;
 
-    static std::unique_ptr<AuthorizationManager> create();
+    virtual ~AuthorizationManager() = default;
 
     static constexpr StringData USERID_FIELD_NAME = "userId"_sd;
     static constexpr StringData USER_NAME_FIELD_NAME = "user"_sd;
@@ -115,7 +104,6 @@ public:
     static const NamespaceString defaultTempUsersCollectionNamespace;  // for mongorestore
     static const NamespaceString defaultTempRolesCollectionNamespace;  // for mongorestore
 
-
     /**
      * Status to be returned when authentication fails. Being consistent about our returned Status
      * prevents information leakage.
@@ -302,14 +290,6 @@ public:
     virtual void updatePinnedUsersList(std::vector<UserName> names) = 0;
 
     /**
-     * Parses privDoc and fully initializes the user object (credentials, roles, and privileges)
-     * with the information extracted from the privilege document.
-     * This should never be called from outside the AuthorizationManager - the only reason it's
-     * public instead of private is so it can be unit tested.
-     */
-    virtual Status _initializeUserFromPrivilegeDocument(User* user, const BSONObj& privDoc) = 0;
-
-    /**
      * Hook called by replication code to let the AuthorizationManager observe changes
      * to relevant collections.
      */
diff --git a/src/mongo/db/auth/authorization_manager_impl.cpp b/src/mongo/db/auth/authorization_manager_impl.cpp
index b9ef57eef94..03ba2a98884 100644
--- a/src/mongo/db/auth/authorization_manager_impl.cpp
+++ b/src/mongo/db/auth/authorization_manager_impl.cpp
@@ -40,31 +40,17 @@
 #include "mongo/base/init.h"
 #include "mongo/base/shim.h"
 #include "mongo/base/status.h"
-#include "mongo/bson/mutable/document.h"
-#include "mongo/bson/mutable/element.h"
 #include "mongo/bson/util/bson_extract.h"
 #include "mongo/crypto/mechanism_scram.h"
-#include "mongo/db/auth/action_set.h"
 #include "mongo/db/auth/address_restriction.h"
 #include "mongo/db/auth/authorization_manager_impl_parameters_gen.h"
-#include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/auth/authorization_session_impl.h"
 #include "mongo/db/auth/authz_manager_external_state.h"
-#include "mongo/db/auth/privilege.h"
-#include "mongo/db/auth/privilege_parser.h"
-#include "mongo/db/auth/role_graph.h"
 #include "mongo/db/auth/sasl_options.h"
-#include "mongo/db/auth/user.h"
 #include "mongo/db/auth/user_document_parser.h"
 #include "mongo/db/auth/user_management_commands_parser.h"
-#include "mongo/db/auth/user_name.h"
-#include "mongo/db/concurrency/d_concurrency.h"
 #include "mongo/db/global_settings.h"
-#include "mongo/db/jsobj.h"
 #include "mongo/db/mongod_options.h"
-#include "mongo/platform/compiler.h"
-#include "mongo/platform/mutex.h"
-#include "mongo/stdx/unordered_map.h"
 #include "mongo/util/assert_util.h"
 #include "mongo/util/log.h"
 #include "mongo/util/str.h"
@@ -72,18 +58,11 @@
 namespace mongo {
 namespace {
 
-using std::back_inserter;
-using std::begin;
-using std::end;
-using std::endl;
-using std::string;
-using std::vector;
-
 MONGO_INITIALIZER_GENERAL(SetupInternalSecurityUser,
                           ("EndStartupOptionStorage"),
                           ("CreateAuthorizationManager"))
 (InitializerContext* const context) try {
-    UserHandle user = std::make_shared<User>(UserName("__system", "local"));
+    UserHandle user(User(UserName("__system", "local")));
 
     ActionSet allActions;
     allActions.addAllActions();
@@ -104,7 +83,6 @@ MONGO_INITIALIZER_GENERAL(SetupInternalSecurityUser,
         user->setRestrictions(std::move(clusterWhiteList));
     }
 
-
     internalSecurity.user = user;
 
     return Status::OK();
@@ -114,7 +92,7 @@ MONGO_INITIALIZER_GENERAL(SetupInternalSecurityUser,
 
 class PinnedUserSetParameter {
 public:
-    void append(OperationContext* opCtx, BSONObjBuilder& b, const std::string& name) const {
+    void append(BSONObjBuilder& b, const std::string& name) const {
         BSONArrayBuilder sub(b.subarrayStart(name));
         stdx::lock_guard<Latch> lk(_mutex);
         for (const auto& username : _pinnedUsersList) {
@@ -199,156 +177,137 @@ private:
         }
         return Status::OK();
     }
-    AuthorizationManager* _authzManager = nullptr;
 
     mutable Mutex _mutex = MONGO_MAKE_LATCH("PinnedUserSetParameter::_mutex");
-    std::vector<UserName> _pinnedUsersList;
-} authorizationManagerPinnedUsers;
 
-}  // namespace
-
-int authorizationManagerCacheSize;
+    AuthorizationManager* _authzManager = nullptr;
 
-void AuthorizationManagerPinnedUsersServerParameter::append(OperationContext* opCtx,
-                                                            BSONObjBuilder& out,
-                                                            const std::string& name) {
-    return authorizationManagerPinnedUsers.append(opCtx, out, name);
-}
+    std::vector<UserName> _pinnedUsersList;
 
-Status AuthorizationManagerPinnedUsersServerParameter::set(const BSONElement& newValue) {
-    return authorizationManagerPinnedUsers.set(newValue);
-}
+} authorizationManagerPinnedUsers;
 
-Status AuthorizationManagerPinnedUsersServerParameter::setFromString(const std::string& str) {
-    return authorizationManagerPinnedUsers.setFromString(str);
+bool isAuthzNamespace(const NamespaceString& nss) {
+    return (nss == AuthorizationManager::rolesCollectionNamespace ||
+            nss == AuthorizationManager::usersCollectionNamespace ||
+            nss == AuthorizationManager::versionCollectionNamespace);
 }
 
-namespace {
-
-std::unique_ptr<AuthorizationManager> authorizationManagerCreateImpl() {
-    return std::make_unique<AuthorizationManagerImpl>();
+bool isAuthzCollection(StringData coll) {
+    return (coll == AuthorizationManager::rolesCollectionNamespace.coll() ||
+            coll == AuthorizationManager::usersCollectionNamespace.coll() ||
+            coll == AuthorizationManager::versionCollectionNamespace.coll());
 }
 
-auto authorizationManagerCreateRegistration =
-    MONGO_WEAK_FUNCTION_REGISTRATION(AuthorizationManager::create, authorizationManagerCreateImpl);
+bool loggedCommandOperatesOnAuthzData(const NamespaceString& nss, const BSONObj& cmdObj) {
+    if (nss != AuthorizationManager::adminCommandNamespace)
+        return false;
 
-}  // namespace
+    const StringData cmdName(cmdObj.firstElement().fieldNameStringData());
 
-/**
- * Guard object for synchronizing accesses to data cached in AuthorizationManager instances.
- * This guard allows one thread to access the cache at a time, and provides an exception-safe
- * mechanism for a thread to release the cache mutex while performing network or disk operations
- * while allowing other readers to proceed.
- *
- * There are two ways to use this guard.  One may simply instantiate the guard like a
- * std::lock_guard, and perform reads or writes of the cache.
- *
- * Alternatively, one may instantiate the guard, examine the cache, and then enter into an
- * update mode by first wait()ing until otherUpdateInFetchPhase() is false, and then
- * calling beginFetchPhase().  At this point, other threads may acquire the guard in the simple
- * manner and do reads, but other threads may not enter into a fetch phase.  During the fetch
- * phase, the thread should perform required network or disk activity to determine what update
- * it will make to the cache.  Then, it should call endFetchPhase(), to reacquire the user cache
- * mutex.  At that point, the thread can make its modifications to the cache and let the guard
- * go out of scope.
- *
- * All updates by guards using a fetch-phase are totally ordered with respect to one another,
- * and all guards using no fetch phase are totally ordered with respect to one another, but
- * there is not a total ordering among all guard objects.
- *
- * The cached data has an associated counter, called the cache generation.  If the cache
- * generation changes while a guard is in fetch phase, the fetched data should not be stored
- * into the cache, because some invalidation event occurred during the fetch phase.
- */
-class AuthorizationManagerImpl::CacheGuard {
-    CacheGuard(const CacheGuard&) = delete;
-    CacheGuard& operator=(const CacheGuard&) = delete;
+    if (cmdName == "drop") {
+        return isAuthzCollection(cmdObj.firstElement().valueStringData());
+    } else if (cmdName == "dropDatabase") {
+        return true;
+    } else if (cmdName == "renameCollection") {
+        const NamespaceString fromNamespace(cmdObj.firstElement().valueStringDataSafe());
+        const NamespaceString toNamespace(cmdObj["to"].valueStringDataSafe());
 
-public:
-    /**
-     * Constructs a cache guard, locking the mutex that synchronizes user cache accesses.
-     */
-    explicit CacheGuard(OperationContext* opCtx, AuthorizationManagerImpl* authzManager)
-        : _isThisGuardInFetchPhase(false),
-          _authzManager(authzManager),
-          _cacheLock(authzManager->_cacheWriteMutex) {}
-
-    /**
-     * Releases the mutex that synchronizes user cache access, if held, and notifies
-     * any threads waiting for their own opportunity to update the user cache.
-     */
-    ~CacheGuard() {
-        if (!_cacheLock.owns_lock()) {
-            _cacheLock.lock();
-        }
-        if (_isThisGuardInFetchPhase) {
-            fassert(17190, otherUpdateInFetchPhase());
-            _authzManager->_isFetchPhaseBusy = false;
-            _authzManager->_fetchPhaseIsReady.notify_all();
+        if (fromNamespace.isAdminDB() || toNamespace.isAdminDB()) {
+            return isAuthzCollection(fromNamespace.coll()) || isAuthzCollection(toNamespace.coll());
+        } else {
+            return false;
         }
+    } else if (cmdName == "dropIndexes" || cmdName == "deleteIndexes") {
+        return false;
+    } else if (cmdName == "create") {
+        return false;
+    } else {
+        return true;
     }
+}
 
-    /**
-     * Returns true of the authzManager reports that it is in fetch phase.
-     */
-    bool otherUpdateInFetchPhase() const {
-        return _authzManager->_isFetchPhaseBusy;
+bool appliesToAuthzData(const char* op, const NamespaceString& nss, const BSONObj& o) {
+    switch (*op) {
+        case 'i':
+        case 'u':
+        case 'd':
+            if (op[1] != '\0')
+                return false;  // "db" op type
+            return isAuthzNamespace(nss);
+        case 'c':
+            return loggedCommandOperatesOnAuthzData(nss, o);
+            break;
+        case 'n':
+            return false;
+        default:
+            return true;
     }
+}
 
-    /**
-     * Waits on the _authzManager->_fetchPhaseIsReady condition.
-     */
-    void wait() {
-        fassert(17222, !_isThisGuardInFetchPhase);
-        _authzManager->_fetchPhaseIsReady.wait(_cacheLock,
-                                               [&] { return !otherUpdateInFetchPhase(); });
+/**
+ * Parses privDoc and fully initializes the user object (credentials, roles, and privileges) with
+ * the information extracted from the privilege document.
+ */
+Status initializeUserFromPrivilegeDocument(User* user, const BSONObj& privDoc) {
+    V2UserDocumentParser parser;
+    std::string userName = parser.extractUserNameFromUserDocument(privDoc);
+    if (userName != user->getName().getUser()) {
+        return Status(ErrorCodes::BadValue,
+                      str::stream() << "User name from privilege document \"" << userName
+                                    << "\" doesn't match name of provided User \""
+                                    << user->getName().getUser() << "\"");
     }
 
-    /**
-     * Enters fetch phase, releasing the _authzManager->_cacheMutex after recording the current
-     * cache generation.
-     */
-    void beginFetchPhase() {
-        fassert(17191, !otherUpdateInFetchPhase());
-        _isThisGuardInFetchPhase = true;
-        _authzManager->_isFetchPhaseBusy = true;
-        _startGeneration = _authzManager->_fetchGeneration;
-        _cacheLock.unlock();
-    }
+    user->setID(parser.extractUserIDFromUserDocument(privDoc));
 
-    /**
-     * Exits the fetch phase, reacquiring the _authzManager->_cacheMutex.
-     */
-    void endFetchPhase() {
-        _cacheLock.lock();
-        // We do not clear _authzManager->_isFetchPhaseBusy or notify waiters until
-        // ~CacheGuard(), for two reasons.  First, there's no value to notifying the waiters
-        // before you're ready to release the mutex, because they'll just go to sleep on the
-        // mutex.  Second, in order to meaningfully check the preconditions of
-        // isSameCacheGeneration(), we need a state that means "fetch phase was entered and now
-        // has been exited."  That state is _isThisGuardInFetchPhase == true and
-        // _lock.owns_lock() == true.
+    Status status = parser.initializeUserCredentialsFromUserDocument(user, privDoc);
+    if (!status.isOK()) {
+        return status;
     }
-
-    /**
-     * Returns true if _authzManager->_fetchGeneration remained the same while this guard was
-     * in fetch phase.  Behavior is undefined if this guard never entered fetch phase.
-     *
-     * If this returns true, do not update the cached data with this
-     */
-    bool isSameCacheGeneration() const {
-        fassert(17223, _isThisGuardInFetchPhase);
-        fassert(17231, _cacheLock.owns_lock());
-        return _startGeneration == _authzManager->_fetchGeneration;
+    status = parser.initializeUserRolesFromUserDocument(privDoc, user);
+    if (!status.isOK()) {
+        return status;
+    }
+    status = parser.initializeUserIndirectRolesFromUserDocument(privDoc, user);
+    if (!status.isOK()) {
+        return status;
+    }
+    status = parser.initializeUserPrivilegesFromUserDocument(privDoc, user);
+    if (!status.isOK()) {
+        return status;
+    }
+    status = parser.initializeAuthenticationRestrictionsFromUserDocument(privDoc, user);
+    if (!status.isOK()) {
+        return status;
     }
 
-private:
-    OID _startGeneration;
-    bool _isThisGuardInFetchPhase;
-    AuthorizationManagerImpl* _authzManager;
+    return Status::OK();
+}
 
-    stdx::unique_lock<Latch> _cacheLock;
-};
+std::unique_ptr<AuthorizationManager> authorizationManagerCreateImpl() {
+    return std::make_unique<AuthorizationManagerImpl>();
+}
+
+auto authorizationManagerCreateRegistration =
+    MONGO_WEAK_FUNCTION_REGISTRATION(AuthorizationManager::create, authorizationManagerCreateImpl);
+
+}  // namespace
+
+int authorizationManagerCacheSize;
+
+void AuthorizationManagerPinnedUsersServerParameter::append(OperationContext* opCtx,
+                                                            BSONObjBuilder& out,
+                                                            const std::string& name) {
+    return authorizationManagerPinnedUsers.append(out, name);
+}
+
+Status AuthorizationManagerPinnedUsersServerParameter::set(const BSONElement& newValue) {
+    return authorizationManagerPinnedUsers.set(newValue);
+}
+
+Status AuthorizationManagerPinnedUsersServerParameter::setFromString(const std::string& str) {
+    return authorizationManagerPinnedUsers.setFromString(str);
+}
 
 AuthorizationManagerImpl::AuthorizationManagerImpl()
     : AuthorizationManagerImpl(AuthzManagerExternalState::create(),
@@ -356,12 +315,9 @@ AuthorizationManagerImpl::AuthorizationManagerImpl()
 
 AuthorizationManagerImpl::AuthorizationManagerImpl(
     std::unique_ptr<AuthzManagerExternalState> externalState, InstallMockForTestingOrAuthImpl)
-    : _authEnabled(false),
-      _privilegeDocsExist(false),
-      _externalState(std::move(externalState)),
-      _version(schemaVersionInvalid),
-      _userCache(authorizationManagerCacheSize, UserCacheInvalidator()),
-      _fetchGeneration(OID::gen()) {}
+    : _externalState(std::move(externalState)),
+      _authSchemaVersionCache(_externalState.get()),
+      _userCache(&_authSchemaVersionCache, _externalState.get(), authorizationManagerCacheSize) {}
 
 AuthorizationManagerImpl::~AuthorizationManagerImpl() {}
 
@@ -379,33 +335,16 @@ bool AuthorizationManagerImpl::shouldValidateAuthSchemaOnStartup() {
     return _startupAuthSchemaValidation;
 }
 
-Status AuthorizationManagerImpl::getAuthorizationVersion(OperationContext* opCtx, int* version) {
-    CacheGuard guard(opCtx, this);
-    int newVersion = _version;
-    if (schemaVersionInvalid == newVersion) {
-        while (guard.otherUpdateInFetchPhase())
-            guard.wait();
-        guard.beginFetchPhase();
-        Status status = _externalState->getStoredAuthorizationVersion(opCtx, &newVersion);
-        guard.endFetchPhase();
-        if (!status.isOK()) {
-            warning() << "Problem fetching the stored schema version of authorization data: "
-                      << redact(status);
-            *version = schemaVersionInvalid;
-            return status;
-        }
-
-        if (guard.isSameCacheGeneration()) {
-            _version = newVersion;
-        }
-    }
-    *version = newVersion;
+Status AuthorizationManagerImpl::getAuthorizationVersion(OperationContext* opCtx,
+                                                         int* version) try {
+    *version = *_authSchemaVersionCache.acquire(opCtx, 0);
     return Status::OK();
+} catch (const DBException& ex) {
+    return ex.toStatus();
 }
 
 OID AuthorizationManagerImpl::getCacheGeneration() {
-    stdx::lock_guard<Latch> lk(_cacheWriteMutex);
-    return _fetchGeneration;
+    return _userCache.getCacheGeneration();
 }
 
 void AuthorizationManagerImpl::setAuthEnabled(bool enabled) {
@@ -431,43 +370,6 @@ bool AuthorizationManagerImpl::hasAnyPrivilegeDocuments(OperationContext* opCtx)
     return _privilegeDocsExist.load();
 }
 
-Status AuthorizationManagerImpl::_initializeUserFromPrivilegeDocument(User* user,
-                                                                      const BSONObj& privDoc) {
-    V2UserDocumentParser parser;
-    std::string userName = parser.extractUserNameFromUserDocument(privDoc);
-    if (userName != user->getName().getUser()) {
-        return Status(ErrorCodes::BadValue,
-                      str::stream() << "User name from privilege document \"" << userName
-                                    << "\" doesn't match name of provided User \""
-                                    << user->getName().getUser() << "\"");
-    }
-
-    user->setID(parser.extractUserIDFromUserDocument(privDoc));
-
-    Status status = parser.initializeUserCredentialsFromUserDocument(user, privDoc);
-    if (!status.isOK()) {
-        return status;
-    }
-    status = parser.initializeUserRolesFromUserDocument(privDoc, user);
-    if (!status.isOK()) {
-        return status;
-    }
-    status = parser.initializeUserIndirectRolesFromUserDocument(privDoc, user);
-    if (!status.isOK()) {
-        return status;
-    }
-    status = parser.initializeUserPrivilegesFromUserDocument(privDoc, user);
-    if (!status.isOK()) {
-        return status;
-    }
-    status = parser.initializeAuthenticationRestrictionsFromUserDocument(privDoc, user);
-    if (!status.isOK()) {
-        return status;
-    }
-
-    return Status::OK();
-}
-
 Status AuthorizationManagerImpl::getUserDescription(OperationContext* opCtx,
                                                     const UserName& userName,
                                                     BSONObj* result) {
@@ -497,117 +399,24 @@ Status AuthorizationManagerImpl::getRoleDescriptionsForDB(
     PrivilegeFormat privileges,
     AuthenticationRestrictionsFormat restrictions,
     bool showBuiltinRoles,
-    vector<BSONObj>* result) {
+    std::vector<BSONObj>* result) {
     return _externalState->getRoleDescriptionsForDB(
         opCtx, dbname, privileges, restrictions, showBuiltinRoles, result);
 }
 
-void AuthorizationManagerImpl::UserCacheInvalidator::operator()(User* user) {
-    LOG(1) << "Invalidating user " << user->getName().toString();
-    user->_invalidate();
-}
-
 StatusWith<UserHandle> AuthorizationManagerImpl::acquireUser(OperationContext* opCtx,
-                                                             const UserName& userName) {
+                                                             const UserName& userName) try {
     if (userName == internalSecurity.user->getName()) {
         return internalSecurity.user;
     }
 
-    boost::optional<UserHandle> cachedUser = _userCache.get(userName);
-    auto returnUser = [&](boost::optional<UserHandle> cachedUser) {
-        auto ret = *cachedUser;
-        fassert(16914, ret.get());
-
-        LOG(1) << "Returning user " << userName << " from cache";
-        return ret;
-    };
-
-    // The userCache is thread-safe, so if we can get a user out of the cache we don't need to
-    // take any locks!
-    if (cachedUser) {
-        return returnUser(cachedUser);
-    }
-
-    // Otherwise make sure we have the locks we need and check whether and wait on another
-    // thread is fetching into the cache
-    CacheGuard guard(opCtx, this);
-
-    while ((boost::none == (cachedUser = _userCache.get(userName))) &&
-           guard.otherUpdateInFetchPhase()) {
-        guard.wait();
-    }
-
-    if (cachedUser != boost::none) {
-        return returnUser(cachedUser);
-    }
-
-    guard.beginFetchPhase();
-    // If there's still no user in the cache, then we need to go to disk. Take the slow path.
-    LOG(1) << "Getting user " << userName << " from disk";
-
-    int authzVersion = _version;
-
-    // Number of times to retry a user document that fetches due to transient
-    // AuthSchemaIncompatible errors.  These errors should only ever occur during and shortly
-    // after schema upgrades.
-    static const int maxAcquireRetries = 2;
-    Status status = Status::OK();
-    std::unique_ptr<User> user;
-    for (int i = 0; i < maxAcquireRetries; ++i) {
-        if (authzVersion == schemaVersionInvalid) {
-            Status status = _externalState->getStoredAuthorizationVersion(opCtx, &authzVersion);
-            if (!status.isOK())
-                return status;
-        }
-
-        switch (authzVersion) {
-            default:
-                status =
-                    Status(ErrorCodes::BadValue,
-                           str::stream() << "Illegal value for authorization data schema version, "
-                                         << authzVersion);
-                break;
-            case schemaVersion28SCRAM:
-            case schemaVersion26Final:
-            case schemaVersion26Upgrade:
-                status = _fetchUserV2(opCtx, userName, &user);
-                break;
-            case schemaVersion24:
-                status =
-                    Status(ErrorCodes::AuthSchemaIncompatible,
-                           str::stream() << "Authorization data schema version " << schemaVersion24
-                                         << " not supported after MongoDB version 2.6.");
-                break;
-        }
-        if (status.isOK())
-            break;
-        if (status != ErrorCodes::AuthSchemaIncompatible)
-            return status;
-
-        authzVersion = schemaVersionInvalid;
-    }
-
-    if (!status.isOK())
-        return status;
-
-    // All this does is re-acquire the _cacheWriteMutex if we don't hold it already - a caller
-    // may also call endFetchPhase() after this returns.
-    guard.endFetchPhase();
-
-    UserHandle ret;
-    if (guard.isSameCacheGeneration()) {
-        if (_version == schemaVersionInvalid)
-            _version = authzVersion;
-        ret = _userCache.insertOrAssignAndGet(userName, std::move(user));
-    } else {
-        // If the cache generation changed while this thread was in fetch mode, the data
-        // associated with the user may now be invalid, so we must mark it as such.  The caller
-        // may still opt to use the information for a short while, but not indefinitely.
-        user->_invalidate();
-        ret = UserHandle(std::move(user));
-    }
+    auto cachedUser = _userCache.acquire(opCtx, userName);
+    invariant(cachedUser);
 
-    return ret;
+    LOG(1) << "Returning user " << userName << " from cache";
+    return cachedUser;
+} catch (const DBException& ex) {
+    return ex.toStatus();
 }
 
 StatusWith<UserHandle> AuthorizationManagerImpl::acquireUserForSessionRefresh(
@@ -627,26 +436,6 @@ StatusWith<UserHandle> AuthorizationManagerImpl::acquireUserForSessionRefresh(
     return ret;
 }
 
-Status AuthorizationManagerImpl::_fetchUserV2(OperationContext* opCtx,
-                                              const UserName& userName,
-                                              std::unique_ptr<User>* out) {
-    BSONObj userObj;
-    Status status = getUserDescription(opCtx, userName, &userObj);
-    if (!status.isOK()) {
-        return status;
-    }
-
-    auto user = std::make_unique<User>(userName);
-
-    status = _initializeUserFromPrivilegeDocument(user.get(), userObj);
-    if (!status.isOK()) {
-        return status;
-    }
-
-    std::swap(*out, user);
-    return Status::OK();
-}
-
 void AuthorizationManagerImpl::updatePinnedUsersList(std::vector<UserName> names) {
     stdx::unique_lock<Latch> lk(_pinnedUsersMutex);
     _usersToPin = std::move(names);
@@ -694,7 +483,7 @@ void AuthorizationManagerImpl::_pinnedUsersThreadRoutine() noexcept try {
                     return (user->getName() == userName);
                 });
 
-            if (!user->isValid() || !shouldPin) {
+            if (!user.isValid() || !shouldPin) {
                 if (!shouldPin) {
                     LOG(2) << "Unpinning user " << user->getName();
                 } else {
@@ -739,32 +528,22 @@ void AuthorizationManagerImpl::_pinnedUsersThreadRoutine() noexcept try {
 
 void AuthorizationManagerImpl::invalidateUserByName(OperationContext* opCtx,
                                                     const UserName& userName) {
-    CacheGuard guard(opCtx, this);
-    _updateCacheGeneration_inlock(guard);
     LOG(2) << "Invalidating user " << userName;
+    _authSchemaVersionCache.invalidateAll();
     _userCache.invalidate(userName);
 }
 
 void AuthorizationManagerImpl::invalidateUsersFromDB(OperationContext* opCtx, StringData dbname) {
-    CacheGuard guard(opCtx, this);
-    _updateCacheGeneration_inlock(guard);
     LOG(2) << "Invalidating all users from database " << dbname;
+    _authSchemaVersionCache.invalidateAll();
     _userCache.invalidateIf(
         [&](const UserName& user, const User*) { return user.getDB() == dbname; });
 }
 
 void AuthorizationManagerImpl::invalidateUserCache(OperationContext* opCtx) {
-    CacheGuard guard(opCtx, this);
     LOG(2) << "Invalidating user cache";
-    _invalidateUserCache_inlock(guard);
-}
-
-void AuthorizationManagerImpl::_invalidateUserCache_inlock(const CacheGuard& guard) {
-    _updateCacheGeneration_inlock(guard);
-    _userCache.invalidateIf([](const UserName& a, const User*) { return true; });
-
-    // Reread the schema version before acquiring the next user.
-    _version = schemaVersionInvalid;
+    _authSchemaVersionCache.invalidateAll();
+    _userCache.invalidateAll();
 }
 
 Status AuthorizationManagerImpl::initialize(OperationContext* opCtx) {
@@ -777,70 +556,6 @@ Status AuthorizationManagerImpl::initialize(OperationContext* opCtx) {
     return Status::OK();
 }
 
-namespace {
-bool isAuthzNamespace(const NamespaceString& nss) {
-    return (nss == AuthorizationManager::rolesCollectionNamespace ||
-            nss == AuthorizationManager::usersCollectionNamespace ||
-            nss == AuthorizationManager::versionCollectionNamespace);
-}
-
-bool isAuthzCollection(StringData coll) {
-    return (coll == AuthorizationManager::rolesCollectionNamespace.coll() ||
-            coll == AuthorizationManager::usersCollectionNamespace.coll() ||
-            coll == AuthorizationManager::versionCollectionNamespace.coll());
-}
-
-bool loggedCommandOperatesOnAuthzData(const NamespaceString& nss, const BSONObj& cmdObj) {
-    if (nss != AuthorizationManager::adminCommandNamespace)
-        return false;
-    const StringData cmdName(cmdObj.firstElement().fieldNameStringData());
-    if (cmdName == "drop") {
-        return isAuthzCollection(cmdObj.firstElement().valueStringData());
-    } else if (cmdName == "dropDatabase") {
-        return true;
-    } else if (cmdName == "renameCollection") {
-        const NamespaceString fromNamespace(cmdObj.firstElement().str());
-        const NamespaceString toNamespace(cmdObj["to"].str());
-        if (fromNamespace.db() == "admin" || toNamespace.db() == "admin") {
-            return isAuthzCollection(fromNamespace.coll().toString()) ||
-                isAuthzCollection(toNamespace.coll().toString());
-        } else {
-            return false;
-        }
-
-    } else if (cmdName == "dropIndexes" || cmdName == "deleteIndexes") {
-        return false;
-    } else if (cmdName == "create") {
-        return false;
-    } else {
-        return true;
-    }
-}
-
-bool appliesToAuthzData(const char* op, const NamespaceString& nss, const BSONObj& o) {
-    switch (*op) {
-        case 'i':
-        case 'u':
-        case 'd':
-            if (op[1] != '\0')
-                return false;  // "db" op type
-            return isAuthzNamespace(nss);
-        case 'c':
-            return loggedCommandOperatesOnAuthzData(nss, o);
-            break;
-        case 'n':
-            return false;
-        default:
-            return true;
-    }
-}
-
-}  // namespace
-
-void AuthorizationManagerImpl::_updateCacheGeneration_inlock(const CacheGuard&) {
-    _fetchGeneration = OID::gen();
-}
-
 void AuthorizationManagerImpl::logOp(OperationContext* opCtx,
                                      const char* op,
                                      const NamespaceString& nss,
@@ -858,10 +573,77 @@ std::vector<AuthorizationManager::CachedUserInfo> AuthorizationManagerImpl::getU
     ret.reserve(cacheData.size());
     std::transform(
         cacheData.begin(), cacheData.end(), std::back_inserter(ret), [](const auto& info) {
-            return AuthorizationManager::CachedUserInfo{info.key, info.active};
+            return AuthorizationManager::CachedUserInfo{info.key, info.useCount > 0};
         });
 
     return ret;
 }
 
+AuthorizationManagerImpl::AuthSchemaVersionDistCache::AuthSchemaVersionDistCache(
+    AuthzManagerExternalState* externalState)
+    : DistCache(1, _mutex), _externalState(externalState) {}
+
+boost::optional<int> AuthorizationManagerImpl::AuthSchemaVersionDistCache::lookup(
+    OperationContext* opCtx, const int& unusedKey) {
+    invariant(unusedKey == 0);
+
+    int authzVersion;
+    uassertStatusOK(_externalState->getStoredAuthorizationVersion(opCtx, &authzVersion));
+
+    return authzVersion;
+}
+
+AuthorizationManagerImpl::UserDistCacheImpl::UserDistCacheImpl(
+    AuthSchemaVersionDistCache* authSchemaVersionCache,
+    AuthzManagerExternalState* externalState,
+    int cacheSize)
+    : UserDistCache(cacheSize, _mutex),
+      _authSchemaVersionCache(authSchemaVersionCache),
+      _externalState(externalState) {}
+
+boost::optional<User> AuthorizationManagerImpl::UserDistCacheImpl::lookup(
+    OperationContext* opCtx, const UserName& userName) {
+    LOG(1) << "Getting user " << userName << " from disk";
+
+    // Number of times to retry a user document that fetches due to transient AuthSchemaIncompatible
+    // errors. These errors should only ever occur during and shortly after schema upgrades.
+    int acquireAttemptsLeft = 2;
+
+    while (true) {
+        const int authzVersion = [&] {
+            auto authSchemaVersionHandle = _authSchemaVersionCache->acquire(opCtx, 0);
+            invariant(authSchemaVersionHandle);
+            return *authSchemaVersionHandle;
+        }();
+
+        switch (authzVersion) {
+            case schemaVersion28SCRAM:
+            case schemaVersion26Final:
+            case schemaVersion26Upgrade: {
+                BSONObj userObj;
+                uassertStatusOK(_externalState->getUserDescription(opCtx, userName, &userObj));
+
+                User user(userName);
+                uassertStatusOK(initializeUserFromPrivilegeDocument(&user, userObj));
+                return user;
+            }
+            case schemaVersion24:
+                _authSchemaVersionCache->invalidateAll();
+
+                uassert(ErrorCodes::AuthSchemaIncompatible,
+                        str::stream() << "Authorization data schema version " << schemaVersion24
+                                      << " not supported after MongoDB version 2.6.",
+                        --acquireAttemptsLeft);
+                break;
+            default:
+                uasserted(ErrorCodes::BadValue,
+                          str::stream() << "Illegal value for authorization data schema version, "
+                                        << authzVersion);
+                break;
+        }
+    }
+
+    MONGO_UNREACHABLE;
+}
+
 }  // namespace mongo
diff --git a/src/mongo/db/auth/authorization_manager_impl.h b/src/mongo/db/auth/authorization_manager_impl.h
index d80d0b668f6..ecf7a9b1ace 100644
--- a/src/mongo/db/auth/authorization_manager_impl.h
+++ b/src/mongo/db/auth/authorization_manager_impl.h
@@ -30,37 +30,12 @@
 #pragma once
 
 #include "mongo/db/auth/authorization_manager.h"
-
-#include <functional>
-#include <memory>
-#include <mutex>
-#include <string>
-
-#include "mongo/base/secure_allocator.h"
-#include "mongo/base/status.h"
-#include "mongo/bson/mutable/element.h"
-#include "mongo/bson/oid.h"
-#include "mongo/db/auth/action_set.h"
-#include "mongo/db/auth/privilege_format.h"
-#include "mongo/db/auth/resource_pattern.h"
-#include "mongo/db/auth/role_graph.h"
-#include "mongo/db/auth/user.h"
-#include "mongo/db/auth/user_name.h"
-#include "mongo/db/jsobj.h"
-#include "mongo/db/namespace_string.h"
-#include "mongo/db/server_options.h"
 #include "mongo/platform/atomic_word.h"
 #include "mongo/platform/mutex.h"
 #include "mongo/stdx/condition_variable.h"
 #include "mongo/stdx/unordered_map.h"
-#include "mongo/util/invalidating_lru_cache.h"
 
 namespace mongo {
-class AuthorizationSession;
-class AuthzManagerExternalState;
-class OperationContext;
-class ServiceContext;
-class UserDocumentParser;
 
 /**
  * Contains server/cluster-wide information about Authorization.
@@ -138,8 +113,6 @@ public:
 
     void updatePinnedUsersList(std::vector<UserName> names) override;
 
-    Status _initializeUserFromPrivilegeDocument(User* user, const BSONObj& privDoc) override;
-
     void logOp(OperationContext* opCtx,
                const char* opstr,
                const NamespaceString& nss,
@@ -150,18 +123,6 @@ public:
 
 private:
     /**
-     * Type used to guard accesses and updates to the user cache.
-     */
-    class CacheGuard;
-    friend class AuthorizationManagerImpl::CacheGuard;
-
-    /**
-     * Invalidates all User objects in the cache and removes them from the cache.
-     * Should only be called when already holding _cacheMutex.
-     */
-    void _invalidateUserCache_inlock(const CacheGuard&);
-
-    /**
      * Given the objects describing an oplog entry that affects authorization data, invalidates
      * the portion of the user cache that is affected by that operation.  Should only be called
      * with oplog entries that have been pre-verified to actually affect authorization data.
@@ -172,96 +133,75 @@ private:
                                       const BSONObj& o,
                                       const BSONObj* o2);
 
-    /**
-     * Updates _cacheGeneration to a new OID
-     */
-    void _updateCacheGeneration_inlock(const CacheGuard&);
-
     void _pinnedUsersThreadRoutine() noexcept;
 
-    /**
-     * Fetches user information from a v2-schema user document for the named user,
-     * and stores a pointer to a new user object into *acquiredUser on success.
-     */
-    Status _fetchUserV2(OperationContext* opCtx,
-                        const UserName& userName,
-                        std::unique_ptr<User>* acquiredUser);
+    std::unique_ptr<AuthzManagerExternalState> _externalState;
 
     /**
      * True if AuthSchema startup checks should be applied in this AuthorizationManager.
      *
-     * Defaults to true.  Changes to its value are not synchronized, so it should only be set
-     * at initalization-time.
+     * Changes to its value are not synchronized, so it should only be set at initalization-time.
      */
-    bool _startupAuthSchemaValidation;
+    bool _startupAuthSchemaValidation{true};
 
     /**
      * True if access control enforcement is enabled in this AuthorizationManager.
      *
-     * Defaults to false.  Changes to its value are not synchronized, so it should only be set
-     * at initalization-time.
+     * Changes to its value are not synchronized, so it should only be set at initalization-time.
      */
-    bool _authEnabled;
+    bool _authEnabled{false};
 
     /**
      * A cache of whether there are any users set up for the cluster.
      */
-    AtomicWord<bool> _privilegeDocsExist;
-
-    std::unique_ptr<AuthzManagerExternalState> _externalState;
+    AtomicWord<bool> _privilegeDocsExist{false};
 
     /**
-     * Cached value of the authorization schema version.
-     *
-     * May be set by acquireUser() and getAuthorizationVersion().  Invalidated by
-     * invalidateUserCache().
-     *
-     * Reads and writes guarded by CacheGuard.
+     * Cache which contains at most a single entry (which has key 0), whose value is the version of
+     * the auth schema.
      */
-    int _version;
+    class AuthSchemaVersionDistCache : public DistCache<int, int> {
+    public:
+        AuthSchemaVersionDistCache(AuthzManagerExternalState* externalState);
 
-    /**
-     * Caches User objects with information about user privileges, to avoid the need to
-     * go to disk to read user privilege documents whenever possible.  Every User object
-     * has a reference count - the AuthorizationManager must not delete a User object in the
-     * cache unless its reference count is zero.
-     */
-    struct UserCacheInvalidator {
-        void operator()(User* user);
-    };
+        // Even though the dist cache permits for lookup to return boost::none for non-existent
+        // values, the contract of the authorization manager is that it should throw an exception if
+        // the value can not be loaded, so if it returns, the value will always be set.
+        boost::optional<int> lookup(OperationContext* opCtx, const int& unusedKey) override;
 
-    InvalidatingLRUCache<UserName, User, UserCacheInvalidator> _userCache;
+    private:
+        Mutex _mutex =
+            MONGO_MAKE_LATCH("AuthorizationManagerImpl::AuthSchemaVersionDistCache::_mutex");
 
-    Mutex _pinnedUsersMutex = MONGO_MAKE_LATCH("AuthorizationManagerImpl::_pinnedUsersMutex");
-    stdx::condition_variable _pinnedUsersCond;
-    std::once_flag _pinnedThreadTrackerStarted;
-    boost::optional<std::vector<UserName>> _usersToPin;
+        AuthzManagerExternalState* const _externalState;
+    } _authSchemaVersionCache;
 
     /**
-     * Protects _cacheGeneration, _version and _isFetchPhaseBusy.  Manipulated
-     * via CacheGuard.
+     * Cache of the users known to the authentication subsystem.
      */
-    Mutex _cacheWriteMutex = MONGO_MAKE_LATCH("AuthorizationManagerImpl::_cacheWriteMutex");
+    class UserDistCacheImpl : public UserDistCache {
+    public:
+        UserDistCacheImpl(AuthSchemaVersionDistCache* authSchemaVersionCache,
+                          AuthzManagerExternalState* externalState,
+                          int cacheSize);
 
-    /**
-     * Current generation of cached data.  Updated every time part of the cache gets
-     * invalidated.  Protected by CacheGuard.
-     */
-    OID _fetchGeneration;
+        // Even though the dist cache permits for lookup to return boost::none for non-existent
+        // values, the contract of the authorization manager is that it should throw an exception if
+        // the value can not be loaded, so if it returns, the value will always be set.
+        boost::optional<User> lookup(OperationContext* opCtx, const UserName& userName) override;
 
-    /**
-     * True if there is an update to the _userCache in progress, and that update is currently in
-     * the "fetch phase", during which it does not hold the _cacheMutex.
-     *
-     * Manipulated via CacheGuard.
-     */
-    bool _isFetchPhaseBusy = false;
+    private:
+        Mutex _mutex = MONGO_MAKE_LATCH("AuthorizationManagerImpl::UserDistCacheImpl::_mutex");
 
-    /**
-     * Condition used to signal that it is OK for another CacheGuard to enter a fetch phase.
-     * Manipulated via CacheGuard.
-     */
-    stdx::condition_variable _fetchPhaseIsReady;
+        AuthSchemaVersionDistCache* const _authSchemaVersionCache;
+
+        AuthzManagerExternalState* const _externalState;
+    } _userCache;
+
+    Mutex _pinnedUsersMutex = MONGO_MAKE_LATCH("AuthorizationManagerImpl::_pinnedUsersMutex");
+    stdx::condition_variable _pinnedUsersCond;
+    std::once_flag _pinnedThreadTrackerStarted;
+    boost::optional<std::vector<UserName>> _usersToPin;
 };
 
 extern int authorizationManagerCacheSize;
diff --git a/src/mongo/db/auth/authorization_manager_test.cpp b/src/mongo/db/auth/authorization_manager_test.cpp
index 54f99db1aa7..30a9c19a766 100644
--- a/src/mongo/db/auth/authorization_manager_test.cpp
+++ b/src/mongo/db/auth/authorization_manager_test.cpp
@@ -31,9 +31,6 @@
 
 #include <memory>
 
-/**
- * Unit tests of the AuthorizationManager type.
- */
 #include "mongo/base/status.h"
 #include "mongo/bson/mutable/document.h"
 #include "mongo/config.h"
@@ -151,7 +148,7 @@ TEST_F(AuthorizationManagerTest, testAcquireV2User) {
     ASSERT_OK(swu.getStatus());
     auto v2read = std::move(swu.getValue());
     ASSERT_EQUALS(UserName("v2read", "test"), v2read->getName());
-    ASSERT(v2read->isValid());
+    ASSERT(v2read.isValid());
     RoleNameIterator roles = v2read->getRoles();
     ASSERT_EQUALS(RoleName("read", "test"), roles.next());
     ASSERT_FALSE(roles.more());
@@ -164,7 +161,7 @@ TEST_F(AuthorizationManagerTest, testAcquireV2User) {
     ASSERT_OK(swu.getStatus());
     auto v2cluster = std::move(swu.getValue());
     ASSERT_EQUALS(UserName("v2cluster", "admin"), v2cluster->getName());
-    ASSERT(v2cluster->isValid());
+    ASSERT(v2cluster.isValid());
     RoleNameIterator clusterRoles = v2cluster->getRoles();
     ASSERT_EQUALS(RoleName("clusterAdmin", "admin"), clusterRoles.next());
     ASSERT_FALSE(clusterRoles.more());
@@ -184,7 +181,7 @@ TEST_F(AuthorizationManagerTest, testLocalX509Authorization) {
     auto swu = authzManager->acquireUser(opCtx.get(), UserName("CN=mongodb.com", "$external"));
     ASSERT_OK(swu.getStatus());
     auto x509User = std::move(swu.getValue());
-    ASSERT(x509User->isValid());
+    ASSERT(x509User.isValid());
 
     stdx::unordered_set<RoleName> expectedRoles{RoleName("read", "test"),
                                                 RoleName("readWrite", "test")};
@@ -311,7 +308,7 @@ TEST_F(AuthorizationManagerTest, testAcquireV2UserWithUnrecognizedActions) {
     ASSERT_OK(swu.getStatus());
     auto myUser = std::move(swu.getValue());
     ASSERT_EQUALS(UserName("myUser", "test"), myUser->getName());
-    ASSERT(myUser->isValid());
+    ASSERT(myUser.isValid());
     RoleNameIterator roles = myUser->getRoles();
     ASSERT_EQUALS(RoleName("myRole", "test"), roles.next());
     ASSERT_FALSE(roles.more());
diff --git a/src/mongo/db/auth/authorization_session_for_test.cpp b/src/mongo/db/auth/authorization_session_for_test.cpp
index 2bc20288c39..ac993b51b1e 100644
--- a/src/mongo/db/auth/authorization_session_for_test.cpp
+++ b/src/mongo/db/auth/authorization_session_for_test.cpp
@@ -52,11 +52,11 @@ void AuthorizationSessionForTest::assumePrivilegesForDB(Privilege privilege, Str
 
 void AuthorizationSessionForTest::assumePrivilegesForDB(PrivilegeVector privileges,
                                                         StringData dbName) {
-    auto user = std::make_shared<User>(UserName("authorizationSessionForTestUser", dbName));
-    user->addPrivileges(privileges);
+    UserHandle userHandle(User(UserName("authorizationSessionForTestUser", dbName)));
+    userHandle->addPrivileges(privileges);
 
-    _authenticatedUsers.add(user);
-    _testUsers.emplace_back(std::move(user));
+    _authenticatedUsers.add(userHandle);
+    _testUsers.emplace_back(std::move(userHandle));
     _buildAuthenticatedRolesVector();
 }
 
diff --git a/src/mongo/db/auth/authorization_session_impl.cpp b/src/mongo/db/auth/authorization_session_impl.cpp
index 73c2feb31b2..0da6ac46e7b 100644
--- a/src/mongo/db/auth/authorization_session_impl.cpp
+++ b/src/mongo/db/auth/authorization_session_impl.cpp
@@ -155,7 +155,8 @@ Status AuthorizationSessionImpl::addAndAuthorizeUser(OperationContext* opCtx,
 }
 
 User* AuthorizationSessionImpl::lookupUser(const UserName& name) {
-    return _authenticatedUsers.lookup(name).get();
+    auto user = _authenticatedUsers.lookup(name);
+    return user ? user.get() : nullptr;
 }
 
 User* AuthorizationSessionImpl::getSingleUser() {
@@ -723,7 +724,7 @@ void AuthorizationSessionImpl::_refreshUserInfoAsNeeded(OperationContext* opCtx)
 
     while (it != _authenticatedUsers.end()) {
         auto& user = *it;
-        if (!user->isValid()) {
+        if (!user.isValid()) {
             // Make a good faith effort to acquire an up-to-date user object, since the one
             // we've cached is marked "out-of-date."
             UserName name = user->getName();
diff --git a/src/mongo/db/auth/authorization_session_test.cpp b/src/mongo/db/auth/authorization_session_test.cpp
index a8b51dc3137..f7df2444e5e 100644
--- a/src/mongo/db/auth/authorization_session_test.cpp
+++ b/src/mongo/db/auth/authorization_session_test.cpp
@@ -31,9 +31,6 @@
 
 #include <memory>
 
-/**
- * Unit tests of the AuthorizationSession type.
- */
 #include "mongo/base/status.h"
 #include "mongo/bson/bson_depth.h"
 #include "mongo/crypto/mechanism_scram.h"
@@ -56,25 +53,22 @@
 #include "mongo/unittest/unittest.h"
 #include "mongo/util/map_util.h"
 
-#define ASSERT_NULL(EXPR) ASSERT_FALSE(EXPR)
-#define ASSERT_NON_NULL(EXPR) ASSERT_TRUE(EXPR)
-
 namespace mongo {
 namespace {
 
 class FailureCapableAuthzManagerExternalStateMock : public AuthzManagerExternalStateMock {
 public:
-    FailureCapableAuthzManagerExternalStateMock() : _findsShouldFail(false) {}
-    virtual ~FailureCapableAuthzManagerExternalStateMock() {}
+    FailureCapableAuthzManagerExternalStateMock() = default;
+    ~FailureCapableAuthzManagerExternalStateMock() = default;
 
     void setFindsShouldFail(bool enable) {
         _findsShouldFail = enable;
     }
 
-    virtual Status findOne(OperationContext* opCtx,
-                           const NamespaceString& collectionName,
-                           const BSONObj& query,
-                           BSONObj* result) {
+    Status findOne(OperationContext* opCtx,
+                   const NamespaceString& collectionName,
+                   const BSONObj& query,
+                   BSONObj* result) override {
         if (_findsShouldFail && collectionName == AuthorizationManager::usersCollectionNamespace) {
             return Status(ErrorCodes::UnknownError,
                           "findOne on admin.system.users set to fail in mock.");
@@ -83,7 +77,7 @@ public:
     }
 
 private:
-    bool _findsShouldFail;
+    bool _findsShouldFail{false};
 };
 
 class AuthorizationSessionTest : public ::mongo::unittest::Test {
@@ -413,17 +407,14 @@ TEST_F(AuthorizationSessionTest, InvalidateUser) {
         authzSession->isAuthorizedForActionsOnResource(testFooCollResource, ActionType::insert));
 
     User* user = authzSession->lookupUser(UserName("spencer", "test"));
-    ASSERT(user->isValid());
 
     // Change the user to be read-only
     int ignored;
-    managerState
-        ->remove(_opCtx.get(),
-                 AuthorizationManager::usersCollectionNamespace,
-                 BSONObj(),
-                 BSONObj(),
-                 &ignored)
-        .transitional_ignore();
+    ASSERT_OK(managerState->remove(_opCtx.get(),
+                                   AuthorizationManager::usersCollectionNamespace,
+                                   BSONObj(),
+                                   BSONObj(),
+                                   &ignored));
     ASSERT_OK(managerState->insertPrivilegeDocument(_opCtx.get(),
                                                     BSON("user"
                                                          << "spencer"
@@ -445,16 +436,13 @@ TEST_F(AuthorizationSessionTest, InvalidateUser) {
         authzSession->isAuthorizedForActionsOnResource(testFooCollResource, ActionType::insert));
 
     user = authzSession->lookupUser(UserName("spencer", "test"));
-    ASSERT(user->isValid());
 
     // Delete the user.
-    managerState
-        ->remove(_opCtx.get(),
-                 AuthorizationManager::usersCollectionNamespace,
-                 BSONObj(),
-                 BSONObj(),
-                 &ignored)
-        .transitional_ignore();
+    ASSERT_OK(managerState->remove(_opCtx.get(),
+                                   AuthorizationManager::usersCollectionNamespace,
+                                   BSONObj(),
+                                   BSONObj(),
+                                   &ignored));
     // Make sure that invalidating the user causes the session to reload its privileges.
     authzManager->invalidateUserByName(_opCtx.get(), user->getName());
     authzSession->startRequest(_opCtx.get());  // Refreshes cached data for invalid users
@@ -486,18 +474,15 @@ TEST_F(AuthorizationSessionTest, UseOldUserInfoInFaceOfConnectivityProblems) {
         authzSession->isAuthorizedForActionsOnResource(testFooCollResource, ActionType::insert));
 
     User* user = authzSession->lookupUser(UserName("spencer", "test"));
-    ASSERT(user->isValid());
 
     // Change the user to be read-only
     int ignored;
     managerState->setFindsShouldFail(true);
-    managerState
-        ->remove(_opCtx.get(),
-                 AuthorizationManager::usersCollectionNamespace,
-                 BSONObj(),
-                 BSONObj(),
-                 &ignored)
-        .transitional_ignore();
+    ASSERT_OK(managerState->remove(_opCtx.get(),
+                                   AuthorizationManager::usersCollectionNamespace,
+                                   BSONObj(),
+                                   BSONObj(),
+                                   &ignored));
     ASSERT_OK(managerState->insertPrivilegeDocument(_opCtx.get(),
                                                     BSON("user"
                                                          << "spencer"
diff --git a/src/mongo/db/auth/sasl_mechanism_registry_test.cpp b/src/mongo/db/auth/sasl_mechanism_registry_test.cpp
index 2b81bb86bc7..a969c00b47a 100644
--- a/src/mongo/db/auth/sasl_mechanism_registry_test.cpp
+++ b/src/mongo/db/auth/sasl_mechanism_registry_test.cpp
@@ -39,7 +39,6 @@
 namespace mongo {
 namespace {
 
-
 TEST(SecurityProperty, emptyHasEmptyProperties) {
     SecurityPropertySet set(SecurityPropertySet{});
     ASSERT_TRUE(set.hasAllProperties(set));
@@ -226,7 +225,7 @@ public:
                                                         << "roles" << BSONArray()),
                                                    BSONObj()));
 
-        internalSecurity.user = std::make_shared<User>(UserName("__system", "local"));
+        internalSecurity.user = UserHandle(User(UserName("__system", "local")));
     }
 
     BSONObj getMechsFor(const UserName user) {
diff --git a/src/mongo/db/auth/security_key_test.cpp b/src/mongo/db/auth/security_key_test.cpp
index 851b5a46fd2..fc1787b5962 100644
--- a/src/mongo/db/auth/security_key_test.cpp
+++ b/src/mongo/db/auth/security_key_test.cpp
@@ -29,7 +29,7 @@
 
 #include "mongo/platform/basic.h"
 
-#include "boost/filesystem.hpp"
+#include <boost/filesystem.hpp>
 
 #include "mongo/base/string_data.h"
 #include "mongo/db/auth/authorization_manager.h"
@@ -38,6 +38,7 @@
 #include "mongo/unittest/unittest.h"
 
 namespace mongo {
+namespace {
 
 class TestFile {
     TestFile(TestFile&) = delete;
@@ -157,7 +158,7 @@ TEST(SecurityFile, Test) {
 }
 
 TEST(SecurityKey, Test) {
-    internalSecurity.user = std::make_shared<User>(UserName("__system", "local"));
+    internalSecurity.user = UserHandle(User(UserName("__system", "local")));
 
     for (const auto& testCase : testCases) {
         TestFile file(testCase.fileContents, testCase.mode != TestCase::FailureMode::Permissions);
@@ -170,4 +171,5 @@ TEST(SecurityKey, Test) {
     }
 }
 
+}  // namespace
 }  // namespace mongo
diff --git a/src/mongo/db/auth/user.cpp b/src/mongo/db/auth/user.cpp
index 96d1251c316..2392bfb0ee0 100644
--- a/src/mongo/db/auth/user.cpp
+++ b/src/mongo/db/auth/user.cpp
@@ -89,11 +89,6 @@ const User::CredentialData& User::getCredentials() const {
     return _credentials;
 }
 
-bool User::isValid() const {
-    return _isValid.loadRelaxed();
-}
-
-
 const ActionSet User::getActionsForResource(const ResourcePattern& resource) const {
     stdx::unordered_map<ResourcePattern, Privilege>::const_iterator it = _privileges.find(resource);
     if (it == _privileges.end()) {
@@ -102,7 +97,6 @@ const ActionSet User::getActionsForResource(const ResourcePattern& resource) con
     return it->second.getActions();
 }
 
-
 bool User::hasActionsForResource(const ResourcePattern& resource) const {
     return !getActionsForResource(resource).empty();
 }
@@ -164,8 +158,4 @@ void User::setRestrictions(RestrictionDocuments restrictions) & {
     _restrictions = std::move(restrictions);
 }
 
-void User::_invalidate() {
-    _isValid.store(false);
-}
-
 }  // namespace mongo
diff --git a/src/mongo/db/auth/user.h b/src/mongo/db/auth/user.h
index f38f90bd084..c5114d19694 100644
--- a/src/mongo/db/auth/user.h
+++ b/src/mongo/db/auth/user.h
@@ -42,6 +42,7 @@
 #include "mongo/platform/atomic_word.h"
 #include "mongo/stdx/unordered_map.h"
 #include "mongo/stdx/unordered_set.h"
+#include "mongo/util/dist_cache.h"
 
 namespace mongo {
 
@@ -64,6 +65,8 @@ class User {
     User& operator=(const User&) = delete;
 
 public:
+    using UserId = std::vector<std::uint8_t>;
+
     template <typename HashBlock>
     struct SCRAMCredentials {
         SCRAMCredentials() : iterationCount(0), salt(""), serverKey(""), storedKey("") {}
@@ -87,6 +90,7 @@ public:
             return !iterationCount && salt.empty() && serverKey.empty() && storedKey.empty();
         }
     };
+
     struct CredentialData {
         CredentialData() : scram_sha1(), scram_sha256(), isExternal(false) {}
 
@@ -104,11 +108,11 @@ public:
         const SCRAMCredentials<HashBlock>& scram() const;
     };
 
-    typedef stdx::unordered_map<ResourcePattern, Privilege> ResourcePrivilegeMap;
+    using ResourcePrivilegeMap = stdx::unordered_map<ResourcePattern, Privilege>;
 
     explicit User(const UserName& name);
+    User(User&&) = default;
 
-    using UserId = std::vector<std::uint8_t>;
     const UserId& getID() const {
         return _id;
     }
@@ -131,7 +135,6 @@ public:
         return _digest;
     }
 
-
     /**
      * Returns an iterator over the names of the user's direct roles
      */
@@ -169,13 +172,6 @@ public:
      */
     bool hasActionsForResource(const ResourcePattern& resource) const;
 
-    /**
-     * Returns true if this copy of information about this user is still valid. If this returns
-     * false, this object should no longer be used and should be returned to the
-     * AuthorizationManager and a new User object for this user should be requested.
-     */
-    bool isValid() const;
-
     // Mutators below.  Mutation functions should *only* be called by the AuthorizationManager
 
     /**
@@ -232,21 +228,11 @@ public:
     }
     void getRestrictions() && = delete;
 
-protected:
-    friend class AuthorizationManagerImpl;
-    /**
-     * Marks this instance of the User object as invalid, most likely because information about
-     * the user has been updated and needs to be reloaded from the AuthorizationManager.
-     *
-     * This method should *only* be called by the AuthorizationManager.
-     */
-    void _invalidate();
-
 private:
-    // Unique ID (often UUID) for this user.
-    // May be empty for legacy users.
+    // Unique ID (often UUID) for this user. May be empty for legacy users.
     UserId _id;
 
+    // The full user name (as specified by the administrator)
     UserName _name;
 
     // Digest of the full username
@@ -266,11 +252,9 @@ private:
 
     // Restrictions which must be met by a Client in order to authenticate as this user.
     RestrictionDocuments _restrictions;
-
-    // Indicates whether the user has been marked as invalid by the AuthorizationManager.
-    AtomicWord<bool> _isValid{true};
 };
 
-using UserHandle = std::shared_ptr<User>;
+using UserDistCache = DistCache<UserName, User>;
+using UserHandle = UserDistCache::ValueHandle;
 
 }  // namespace mongo
diff --git a/src/mongo/db/auth/user_name.cpp b/src/mongo/db/auth/user_name.cpp
index 6aa4446abc2..65a08c3fa59 100644
--- a/src/mongo/db/auth/user_name.cpp
+++ b/src/mongo/db/auth/user_name.cpp
@@ -33,7 +33,6 @@
 #include <iostream>
 #include <string>
 
-#include "mongo/base/string_data.h"
 #include "mongo/db/auth/authorization_manager.h"
 #include "mongo/util/assert_util.h"
 
@@ -50,7 +49,6 @@ UserName::UserName(StringData user, StringData dbname) {
     _splitPoint = user.size();
 }
 
-
 StatusWith<UserName> UserName::parse(StringData userNameStr) {
     size_t splitPoint = userNameStr.find('.');
 
diff --git a/src/mongo/db/auth/user_set.cpp b/src/mongo/db/auth/user_set.cpp
index 52fa1bf164a..cdc19034fbc 100644
--- a/src/mongo/db/auth/user_set.cpp
+++ b/src/mongo/db/auth/user_set.cpp
@@ -27,19 +27,15 @@
  *    it in the license file.
  */
 
+#include "mongo/platform/basic.h"
+
 #include "mongo/db/auth/user_set.h"
 
 #include <algorithm>
-#include <iostream>
-#include <string>
-#include <vector>
-
-#include "mongo/db/auth/authorization_manager.h"
-#include "mongo/db/auth/user.h"
 
 namespace mongo {
-
 namespace {
+
 class UserSetNameIteratorImpl : public UserNameIterator::Impl {
     UserSetNameIteratorImpl(const UserSetNameIteratorImpl&) = delete;
     UserSetNameIteratorImpl& operator=(const UserSetNameIteratorImpl&) = delete;
@@ -48,17 +44,22 @@ public:
     UserSetNameIteratorImpl(const UserSet::const_iterator& begin,
                             const UserSet::const_iterator& end)
         : _curr(begin), _end(end) {}
-    virtual ~UserSetNameIteratorImpl() {}
-    virtual bool more() const {
+
+    ~UserSetNameIteratorImpl() = default;
+
+    bool more() const override {
         return _curr != _end;
     }
-    virtual const UserName& next() {
+
+    const UserName& next() override {
         return (*(_curr++))->getName();
     }
-    virtual const UserName& get() const {
+
+    const UserName& get() const override {
         return (*_curr)->getName();
     }
-    virtual UserNameIterator::Impl* doClone() const {
+
+    UserNameIterator::Impl* doClone() const override {
         return new UserSetNameIteratorImpl(_curr, _end);
     }
 
@@ -66,8 +67,11 @@ private:
     UserSet::const_iterator _curr;
     UserSet::const_iterator _end;
 };
+
 }  // namespace
 
+UserSet::UserSet() = default;
+
 void UserSet::add(UserHandle user) {
     auto it = std::find_if(_users.begin(), _users.end(), [&](const auto& storedUser) {
         return user->getName().getDB() == storedUser->getName().getDB();
@@ -97,24 +101,21 @@ void UserSet::removeAt(iterator it) {
 }
 
 UserHandle UserSet::lookup(const UserName& name) const {
-    auto it = std::find_if(_users.begin(), _users.end(), [&](const auto& user) {
-        invariant(user);
-        return user->getName() == name;
-    });
-
-    return (it != _users.end()) ? *it : nullptr;
+    auto it = std::find_if(
+        _users.begin(), _users.end(), [&](const auto& user) { return user->getName() == name; });
+    return (it != _users.end()) ? *it : UserHandle();
 }
 
 UserHandle UserSet::lookupByDBName(StringData dbname) const {
     auto it = std::find_if(_users.begin(), _users.end(), [&](const auto& user) {
-        invariant(user);
         return user->getName().getDB() == dbname;
     });
-    return (it != _users.end()) ? *it : nullptr;
+    return (it != _users.end()) ? *it : UserHandle();
 }
 
 UserNameIterator UserSet::getNames() const {
     return UserNameIterator(
         std::make_unique<UserSetNameIteratorImpl>(_users.cbegin(), _users.cend()));
 }
+
 }  // namespace mongo
diff --git a/src/mongo/db/auth/user_set.h b/src/mongo/db/auth/user_set.h
index 25ecd841d89..3f03c0c0967 100644
--- a/src/mongo/db/auth/user_set.h
+++ b/src/mongo/db/auth/user_set.h
@@ -30,13 +30,8 @@
 #pragma once
 
 #include <list>
-#include <string>
-#include <vector>
 
-#include "mongo/base/string_data.h"
-#include "mongo/db/auth/authorization_manager.h"
 #include "mongo/db/auth/user.h"
-#include "mongo/db/auth/user_name.h"
 
 namespace mongo {
 
@@ -53,7 +48,7 @@ public:
     using iterator = std::list<UserHandle>::iterator;
     using const_iterator = std::list<UserHandle>::const_iterator;
 
-    UserSet() = default;
+    UserSet();
 
     /**
      * Adds a User to the UserSet.
@@ -105,6 +100,7 @@ public:
     iterator begin() {
         return _users.begin();
     }
+
     iterator end() {
         return _users.end();
     }
diff --git a/src/mongo/db/auth/user_set_test.cpp b/src/mongo/db/auth/user_set_test.cpp
index e0b7df62a74..e2a87b12b4c 100644
--- a/src/mongo/db/auth/user_set_test.cpp
+++ b/src/mongo/db/auth/user_set_test.cpp
@@ -44,9 +44,9 @@ namespace {
 TEST(UserSetTest, BasicTest) {
     UserSet set;
 
-    UserHandle p1 = std::make_shared<User>(UserName("Bob", "test"));
-    UserHandle p2 = std::make_shared<User>(UserName("George", "test"));
-    UserHandle p3 = std::make_shared<User>(UserName("Bob", "test2"));
+    UserHandle p1(User(UserName("Bob", "test")));
+    UserHandle p2(User(UserName("George", "test")));
+    UserHandle p3(User(UserName("Bob", "test2")));
 
     ASSERT_NULL(set.lookup(UserName("Bob", "test")));
     ASSERT_NULL(set.lookup(UserName("George", "test")));
@@ -98,7 +98,7 @@ TEST(UserSetTest, IterateNames) {
     UserNameIterator iter = pset.getNames();
     ASSERT(!iter.more());
 
-    UserHandle user = std::make_shared<User>(UserName("bob", "test"));
+    UserHandle user(User(UserName("bob", "test")));
     pset.add(std::move(user));
 
     iter = pset.getNames();