SERVER-36704 Logical session support for embedded, behaves as standalone mongod

18 files changed, 224 insertions, 82 deletions

diff --git a/src/mongo/db/commands/SConscript b/src/mongo/db/commands/SConscript
index aaea5a7e389..1d99c2f9cb0 100644
--- a/src/mongo/db/commands/SConscript
+++ b/src/mongo/db/commands/SConscript
@@ -91,12 +91,19 @@ env.Library(
 env.Library(
     target="core",
     source=[
+        "end_sessions_command.cpp",
         "fail_point_cmd.cpp",
         "find_and_modify_common.cpp",
         "generic.cpp",
         "hashcmd.cpp",
+        "kill_all_sessions_by_pattern_command.cpp",
+        "kill_all_sessions_command.cpp",
+        "kill_sessions_command.cpp",
         "parameters.cpp",
+        "refresh_logical_session_cache_now.cpp",
+        "refresh_sessions_command.cpp",
         "rename_collection_common.cpp",
+        "start_session_command.cpp",
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/bson/mutable/mutable_bson',
@@ -105,6 +112,11 @@ env.Library(
         '$BUILD_DIR/mongo/db/commands',
         '$BUILD_DIR/mongo/db/commands/test_commands_enabled',
         '$BUILD_DIR/mongo/db/common',
+        '$BUILD_DIR/mongo/db/kill_sessions',
+        '$BUILD_DIR/mongo/db/logical_session_cache',
+        '$BUILD_DIR/mongo/db/logical_session_cache_impl',
+        '$BUILD_DIR/mongo/db/logical_session_id',
+        '$BUILD_DIR/mongo/db/logical_session_id_helpers',
         '$BUILD_DIR/mongo/db/mongohasher',
         '$BUILD_DIR/mongo/db/server_options_core',
         '$BUILD_DIR/mongo/logger/parse_log_component_settings',
@@ -120,29 +132,21 @@ env.Library(
         "conn_pool_stats.cpp",
         "conn_pool_sync.cpp",
         "connection_status.cpp",
-        "end_sessions_command.cpp",
         "generic_servers.cpp",
         "isself.cpp",
-        "kill_all_sessions_by_pattern_command.cpp",
-        "kill_all_sessions_command.cpp",
-        "kill_sessions_command.cpp",
         "mr_common.cpp",
         "reap_logical_session_cache_now.cpp",
-        "refresh_logical_session_cache_now.cpp",
-        "refresh_sessions_command.cpp",
         "refresh_sessions_command_internal.cpp",
-        "start_session_command.cpp",
         "user_management_commands_common.cpp",
     ],
     LIBDEPS_PRIVATE=[
+        '$BUILD_DIR/mongo/client/clientdriver_minimal',
         '$BUILD_DIR/mongo/db/audit',
         '$BUILD_DIR/mongo/db/auth/sasl_options',
         '$BUILD_DIR/mongo/db/auth/user_document_parser',
         '$BUILD_DIR/mongo/db/commands',
         '$BUILD_DIR/mongo/db/commands/test_commands_enabled',
         '$BUILD_DIR/mongo/db/common',
-        '$BUILD_DIR/mongo/client/clientdriver_minimal',
-        '$BUILD_DIR/mongo/db/kill_sessions',
         '$BUILD_DIR/mongo/db/log_process_details',
         '$BUILD_DIR/mongo/db/logical_session_cache',
         '$BUILD_DIR/mongo/db/logical_session_cache_impl',
diff --git a/src/mongo/db/db.cpp b/src/mongo/db/db.cpp
index abe28452d2e..7e531cd2cf7 100644
--- a/src/mongo/db/db.cpp
+++ b/src/mongo/db/db.cpp
@@ -631,7 +631,7 @@ ExitCode _initAndListen(int listenPort) {
         kind = LogicalSessionCacheServer::kReplicaSet;
     }
 
-    auto sessionCache = makeLogicalSessionCacheD(serviceContext, kind);
+    auto sessionCache = makeLogicalSessionCacheD(kind);
     LogicalSessionCache::set(serviceContext, std::move(sessionCache));
 
     // MessageServer::run will return when exit code closes its socket and we don't need the
diff --git a/src/mongo/db/logical_session_cache_factory_mongod.cpp b/src/mongo/db/logical_session_cache_factory_mongod.cpp
index d8d1e84c548..9c952ab4835 100644
--- a/src/mongo/db/logical_session_cache_factory_mongod.cpp
+++ b/src/mongo/db/logical_session_cache_factory_mongod.cpp
@@ -65,8 +65,7 @@ std::shared_ptr<SessionsCollection> makeSessionsCollection(LogicalSessionCacheSe
 
 }  // namespace
 
-std::unique_ptr<LogicalSessionCache> makeLogicalSessionCacheD(ServiceContext* svc,
-                                                              LogicalSessionCacheServer state) {
+std::unique_ptr<LogicalSessionCache> makeLogicalSessionCacheD(LogicalSessionCacheServer state) {
     auto liaison = stdx::make_unique<ServiceLiaisonMongod>();
 
     // Set up the logical session cache
diff --git a/src/mongo/db/logical_session_cache_factory_mongod.h b/src/mongo/db/logical_session_cache_factory_mongod.h
index 7cf8e803036..495c5c4c5c1 100644
--- a/src/mongo/db/logical_session_cache_factory_mongod.h
+++ b/src/mongo/db/logical_session_cache_factory_mongod.h
@@ -37,9 +37,6 @@ namespace mongo {
 
 enum class LogicalSessionCacheServer { kSharded, kConfigServer, kReplicaSet, kStandalone };
 
-class ServiceContext;
-
-std::unique_ptr<LogicalSessionCache> makeLogicalSessionCacheD(ServiceContext* svc,
-                                                              LogicalSessionCacheServer state);
+std::unique_ptr<LogicalSessionCache> makeLogicalSessionCacheD(LogicalSessionCacheServer state);
 
 }  // namespace mongo
diff --git a/src/mongo/db/service_liaison_mongod.cpp b/src/mongo/db/service_liaison_mongod.cpp
index a6e91477f6c..0d5b5cb6970 100644
--- a/src/mongo/db/service_liaison_mongod.cpp
+++ b/src/mongo/db/service_liaison_mongod.cpp
@@ -36,7 +36,6 @@
 #include "mongo/db/service_context.h"
 #include "mongo/stdx/mutex.h"
 #include "mongo/util/clock_source.h"
-#include "mongo/util/periodic_runner.h"
 
 namespace mongo {
 
@@ -86,12 +85,17 @@ LogicalSessionIdSet ServiceLiaisonMongod::getOpenCursorSessions() const {
 
 void ServiceLiaisonMongod::scheduleJob(PeriodicRunner::PeriodicJob job) {
     invariant(hasGlobalServiceContext());
-    getGlobalServiceContext()->getPeriodicRunner()->scheduleJob(std::move(job));
+    auto jobHandle = getGlobalServiceContext()->getPeriodicRunner()->makeJob(std::move(job));
+    jobHandle->start();
+    _jobs.push_back(std::move(jobHandle));
 }
 
 void ServiceLiaisonMongod::join() {
     invariant(hasGlobalServiceContext());
-    getGlobalServiceContext()->getPeriodicRunner()->shutdown();
+    for (auto&& jobHandle : _jobs) {
+        jobHandle->stop();
+    }
+    _jobs.clear();
 }
 
 Date_t ServiceLiaisonMongod::now() const {
diff --git a/src/mongo/db/service_liaison_mongod.h b/src/mongo/db/service_liaison_mongod.h
index ef152b270e0..3eb4704516f 100644
--- a/src/mongo/db/service_liaison_mongod.h
+++ b/src/mongo/db/service_liaison_mongod.h
@@ -67,6 +67,7 @@ protected:
      * Returns the service context.
      */
     ServiceContext* _context() override;
+    std::vector<std::unique_ptr<PeriodicRunner::PeriodicJobHandle>> _jobs;
 };
 
 }  // namespace mongo
diff --git a/src/mongo/embedded/SConscript b/src/mongo/embedded/SConscript
index 1319c183876..ece5e89334d 100644
--- a/src/mongo/embedded/SConscript
+++ b/src/mongo/embedded/SConscript
@@ -32,6 +32,7 @@ env.Library(
         'embedded_options.cpp',
         'embedded_options_init.cpp',
         'embedded_options_parser_init.cpp',
+        'logical_session_cache_factory_embedded.cpp',
         'periodic_runner_embedded.cpp',
         'replication_coordinator_embedded.cpp',
         'service_entry_point_embedded.cpp',
@@ -48,6 +49,8 @@ env.Library(
         '$BUILD_DIR/mongo/db/commands/mongod_fcv',
         '$BUILD_DIR/mongo/db/commands/standalone',
         '$BUILD_DIR/mongo/db/concurrency/lock_manager',
+        '$BUILD_DIR/mongo/db/logical_session_cache',
+        '$BUILD_DIR/mongo/db/logical_session_cache_impl',
         '$BUILD_DIR/mongo/db/op_observer_d',
         '$BUILD_DIR/mongo/db/repair_database_and_check_version',
         '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface',
@@ -59,6 +62,8 @@ env.Library(
         '$BUILD_DIR/mongo/db/server_options',
         '$BUILD_DIR/mongo/db/service_context',
         '$BUILD_DIR/mongo/db/service_entry_point_common',
+        '$BUILD_DIR/mongo/db/service_liaison_mongod',
+        '$BUILD_DIR/mongo/db/sessions_collection_standalone',
         '$BUILD_DIR/mongo/db/storage/mobile/storage_mobile',
         '$BUILD_DIR/mongo/db/storage/storage_engine_common',
         '$BUILD_DIR/mongo/db/storage/storage_engine_lock_file',
diff --git a/src/mongo/embedded/capi_test.cpp b/src/mongo/embedded/capi_test.cpp
index 93f84878d8f..97845e14bf2 100644
--- a/src/mongo/embedded/capi_test.cpp
+++ b/src/mongo/embedded/capi_test.cpp
@@ -547,59 +547,68 @@ TEST_F(MongodbCAPITest, InsertAndUpdate) {
 TEST_F(MongodbCAPITest, RunListCommands) {
     auto client = createClient();
 
-    std::vector<std::string> whitelist = {"_hashBSONElement",
-                                          "aggregate",
-                                          "buildInfo",
-                                          "collMod",
-                                          "collStats",
-                                          "configureFailPoint",
-                                          "count",
-                                          "create",
-                                          "createIndexes",
-                                          "currentOp",
-                                          "dataSize",
-                                          "dbStats",
-                                          "delete",
-                                          "distinct",
-                                          "drop",
-                                          "dropDatabase",
-                                          "dropIndexes",
-                                          "echo",
-                                          "explain",
-                                          "find",
-                                          "findAndModify",
-                                          "getLastError",
-                                          "getMore",
-                                          "getParameter",
-                                          "getPrevError",
-                                          "insert",
-                                          "isMaster",
-                                          "killCursors",
-                                          "killOp",
-                                          "listCollections",
-                                          "listCommands",
-                                          "listDatabases",
-                                          "listIndexes",
-                                          "lockInfo",
-                                          "ping",
-                                          "planCacheClear",
-                                          "planCacheClearFilters",
-                                          "planCacheListFilters",
-                                          "planCacheListPlans",
-                                          "planCacheListQueryShapes",
-                                          "planCacheSetFilter",
-                                          "reIndex",
-                                          "renameCollection",
-                                          "repairCursor",
-                                          "repairDatabase",
-                                          "resetError",
-                                          "serverStatus",
-                                          "setBatteryLevel",
-                                          "setParameter",
-                                          "sleep",
-                                          "trimMemory",
-                                          "update",
-                                          "validate"};
+    std::vector<std::string> whitelist = {
+        "_hashBSONElement",
+        "aggregate",
+        "buildInfo",
+        "collMod",
+        "collStats",
+        "configureFailPoint",
+        "count",
+        "create",
+        "createIndexes",
+        "currentOp",
+        "dataSize",
+        "dbStats",
+        "delete",
+        "distinct",
+        "drop",
+        "dropDatabase",
+        "dropIndexes",
+        "echo",
+        "endSessions",
+        "explain",
+        "find",
+        "findAndModify",
+        "getLastError",
+        "getMore",
+        "getParameter",
+        "getPrevError",
+        "insert",
+        "isMaster",
+        "killCursors",
+        "killOp",
+        "killSessions",
+        "killAllSessions",
+        "killAllSessionsByPattern",
+        "listCollections",
+        "listCommands",
+        "listDatabases",
+        "listIndexes",
+        "lockInfo",
+        "ping",
+        "planCacheClear",
+        "planCacheClearFilters",
+        "planCacheListFilters",
+        "planCacheListPlans",
+        "planCacheListQueryShapes",
+        "planCacheSetFilter",
+        "reIndex",
+        "refreshLogicalSessionCacheNow",
+        "refreshSessions",
+        "renameCollection",
+        "repairCursor",
+        "repairDatabase",
+        "resetError",
+        "serverStatus",
+        "setBatteryLevel",
+        "setParameter",
+        "sleep",
+        "startSession",
+        "trimMemory",
+        "update",
+        "validate",
+    };
     std::sort(whitelist.begin(), whitelist.end());
 
     mongo::BSONObj listCommandsObj = mongo::fromjson("{ listCommands: 1 }");
diff --git a/src/mongo/embedded/embedded.cpp b/src/mongo/embedded/embedded.cpp
index fa520e22ac3..bc0fdfd9f04 100644
--- a/src/mongo/embedded/embedded.cpp
+++ b/src/mongo/embedded/embedded.cpp
@@ -56,6 +56,7 @@
 #include "mongo/db/storage/encryption_hooks.h"
 #include "mongo/db/storage/storage_engine_init.h"
 #include "mongo/db/ttl.h"
+#include "mongo/embedded/logical_session_cache_factory_embedded.h"
 #include "mongo/embedded/periodic_runner_embedded.h"
 #include "mongo/embedded/replication_coordinator_embedded.h"
 #include "mongo/embedded/service_entry_point_embedded.h"
@@ -151,6 +152,8 @@ void shutdown(ServiceContext* srvContext) {
         Lock::GlobalLock lk(shutdownOpCtx.get(), MODE_X);
         DatabaseHolder::getDatabaseHolder().closeAll(shutdownOpCtx.get(), "shutdown");
 
+        LogicalSessionCache::set(serviceContext, nullptr);
+
         // Shut down the background periodic task runner
         if (auto runner = serviceContext->getPeriodicRunner()) {
             runner->shutdown();
@@ -306,6 +309,10 @@ ServiceContext* initialize(const char* yaml_config) {
     periodicRunner->startup();
     serviceContext->setPeriodicRunner(std::move(periodicRunner));
 
+    // Set up the logical session cache
+    auto sessionCache = makeLogicalSessionCacheEmbedded();
+    LogicalSessionCache::set(serviceContext, std::move(sessionCache));
+
     // MessageServer::run will return when exit code closes its socket and we don't need the
     // operation context anymore
     startupOpCtx.reset();
diff --git a/src/mongo/embedded/embedded_ismaster.cpp b/src/mongo/embedded/embedded_ismaster.cpp
index e0836077552..526bfde3f4d 100644
--- a/src/mongo/embedded/embedded_ismaster.cpp
+++ b/src/mongo/embedded/embedded_ismaster.cpp
@@ -97,6 +97,7 @@ public:
         result.appendNumber("maxMessageSizeBytes", MaxMessageSizeBytes);
         result.appendNumber("maxWriteBatchSize", write_ops::kMaxWriteBatchSize);
         result.appendDate("localTime", jsTime());
+        result.append("logicalSessionTimeoutMinutes", localLogicalSessionTimeoutMinutes);
         result.appendNumber("connectionId", opCtx->getClient()->getConnectionId());
 
         result.append("minWireVersion", WireSpec::instance().incomingExternalClient.minWireVersion);
diff --git a/src/mongo/embedded/logical_session_cache_factory_embedded.cpp b/src/mongo/embedded/logical_session_cache_factory_embedded.cpp
new file mode 100644
index 00000000000..106970bc636
--- /dev/null
+++ b/src/mongo/embedded/logical_session_cache_factory_embedded.cpp
@@ -0,0 +1,55 @@
+/**
+ *    Copyright (C) 2018 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kControl
+
+#include "mongo/platform/basic.h"
+
+#include <memory>
+
+#include "mongo/embedded/logical_session_cache_factory_embedded.h"
+
+#include "mongo/db/logical_session_cache_impl.h"
+#include "mongo/db/service_liaison_mongod.h"
+#include "mongo/db/sessions_collection_standalone.h"
+#include "mongo/stdx/memory.h"
+#include "mongo/util/log.h"
+
+namespace mongo {
+
+std::unique_ptr<LogicalSessionCache> makeLogicalSessionCacheEmbedded() {
+    auto liaison = std::make_unique<ServiceLiaisonMongod>();
+
+    // Set up the logical session cache
+    auto sessionsColl = std::make_shared<SessionsCollectionStandalone>();
+
+    return stdx::make_unique<LogicalSessionCacheImpl>(
+        std::move(liaison), std::move(sessionsColl), nullptr, LogicalSessionCacheImpl::Options{});
+}
+
+}  // namespace mongo
diff --git a/src/mongo/embedded/logical_session_cache_factory_embedded.h b/src/mongo/embedded/logical_session_cache_factory_embedded.h
new file mode 100644
index 00000000000..3a104487baf
--- /dev/null
+++ b/src/mongo/embedded/logical_session_cache_factory_embedded.h
@@ -0,0 +1,40 @@
+/**
+ *    Copyright (C) 2018 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 <memory>
+
+#include "mongo/db/logical_session_cache.h"
+#include "mongo/db/service_liaison.h"
+
+namespace mongo {
+
+std::unique_ptr<LogicalSessionCache> makeLogicalSessionCacheEmbedded();
+
+}  // namespace mongo
diff --git a/src/mongo/embedded/periodic_runner_embedded.cpp b/src/mongo/embedded/periodic_runner_embedded.cpp
index e65f9afb8c3..254dd1b9c3c 100644
--- a/src/mongo/embedded/periodic_runner_embedded.cpp
+++ b/src/mongo/embedded/periodic_runner_embedded.cpp
@@ -69,7 +69,7 @@ PeriodicRunnerEmbedded::~PeriodicRunnerEmbedded() {
 
 std::shared_ptr<PeriodicRunnerEmbedded::PeriodicJobImpl> PeriodicRunnerEmbedded::createAndAddJob(
     PeriodicJob job, bool shouldStart) {
-    auto impl = std::make_shared<PeriodicJobImpl>(std::move(job), this->_clockSource, this->_svc);
+    auto impl = std::make_shared<PeriodicJobImpl>(std::move(job), this->_clockSource, this);
 
     stdx::lock_guard<stdx::mutex> lk(_mutex);
     _jobs.push_back(impl);
@@ -197,8 +197,8 @@ bool PeriodicRunnerEmbedded::tryPump() {
 
 PeriodicRunnerEmbedded::PeriodicJobImpl::PeriodicJobImpl(PeriodicJob job,
                                                          ClockSource* source,
-                                                         ServiceContext* svc)
-    : _job(std::move(job)), _clockSource(source), _serviceContext(svc) {}
+                                                         PeriodicRunnerEmbedded* runner)
+    : _job(std::move(job)), _clockSource(source), _periodicRunner(runner) {}
 
 void PeriodicRunnerEmbedded::PeriodicJobImpl::start() {
     stdx::lock_guard<stdx::mutex> lk(_mutex);
@@ -219,6 +219,9 @@ void PeriodicRunnerEmbedded::PeriodicJobImpl::resume() {
 }
 
 void PeriodicRunnerEmbedded::PeriodicJobImpl::stop() {
+    // Also take the master lock, the job lock is not held while executing the job and we must make
+    // sure the user can invalidate it after this call.
+    stdx::lock_guard<stdx::mutex> masterLock(_periodicRunner->_mutex);
     stdx::lock_guard<stdx::mutex> lk(_mutex);
     invariant(isAlive(lk));
 
@@ -234,6 +237,11 @@ void PeriodicRunnerEmbedded::PeriodicJobHandleImpl::start() {
     job->start();
 }
 
+void PeriodicRunnerEmbedded::PeriodicJobHandleImpl::stop() {
+    auto job = lockAndAssertExists(_jobWeak, kPeriodicJobHandleLifetimeErrMsg);
+    job->stop();
+}
+
 void PeriodicRunnerEmbedded::PeriodicJobHandleImpl::pause() {
     auto job = lockAndAssertExists(_jobWeak, kPeriodicJobHandleLifetimeErrMsg);
     job->pause();
diff --git a/src/mongo/embedded/periodic_runner_embedded.h b/src/mongo/embedded/periodic_runner_embedded.h
index ecbe1a20f77..3654ef9c72a 100644
--- a/src/mongo/embedded/periodic_runner_embedded.h
+++ b/src/mongo/embedded/periodic_runner_embedded.h
@@ -65,7 +65,7 @@ private:
 
     public:
         friend class PeriodicRunnerEmbedded;
-        PeriodicJobImpl(PeriodicJob job, ClockSource* source, ServiceContext* svc);
+        PeriodicJobImpl(PeriodicJob job, ClockSource* source, PeriodicRunnerEmbedded* runner);
 
         void start();
         void pause();
@@ -83,7 +83,7 @@ private:
     private:
         PeriodicJob _job;
         ClockSource* _clockSource;
-        ServiceContext* _serviceContext;
+        PeriodicRunnerEmbedded* _periodicRunner;
         Date_t _lastRun{};
 
         // The mutex is protecting _execStatus, the variable that can be accessed from other
@@ -103,6 +103,7 @@ private:
         explicit PeriodicJobHandleImpl(std::weak_ptr<PeriodicJobImpl> jobImpl)
             : _jobWeak(jobImpl) {}
         void start() override;
+        void stop() override;
         void pause() override;
         void resume() override;
 
@@ -110,7 +111,6 @@ private:
         std::weak_ptr<PeriodicJobImpl> _jobWeak;
     };
 
-
     ServiceContext* _svc;
     ClockSource* _clockSource;
 
diff --git a/src/mongo/util/mock_periodic_runner_impl.h b/src/mongo/util/mock_periodic_runner_impl.h
index b1204ef510a..bbc90b8b40c 100644
--- a/src/mongo/util/mock_periodic_runner_impl.h
+++ b/src/mongo/util/mock_periodic_runner_impl.h
@@ -43,9 +43,10 @@ public:
     public:
         ~MockPeriodicJobHandleImpl() = default;
 
-        virtual void start(){};
-        virtual void pause(){};
-        virtual void resume(){};
+        void start() override{};
+        void stop() override{};
+        void pause() override{};
+        void resume() override{};
     };
 
     ~MockPeriodicRunnerImpl() = default;
diff --git a/src/mongo/util/periodic_runner.h b/src/mongo/util/periodic_runner.h
index c6d7fb01ac4..2da18b9a611 100644
--- a/src/mongo/util/periodic_runner.h
+++ b/src/mongo/util/periodic_runner.h
@@ -89,6 +89,11 @@ public:
          * Resumes a paused job so that it continues executing each interval
          */
         virtual void resume() = 0;
+        /**
+         * Stops the job, this function blocks until the job is stopped
+         * Safe to invalidate the job callable after calling this.
+         */
+        virtual void stop() = 0;
     };
 
     virtual ~PeriodicRunner();
diff --git a/src/mongo/util/periodic_runner_impl.cpp b/src/mongo/util/periodic_runner_impl.cpp
index 0be6fba57a7..2d0c80ddeb5 100644
--- a/src/mongo/util/periodic_runner_impl.cpp
+++ b/src/mongo/util/periodic_runner_impl.cpp
@@ -159,6 +159,7 @@ void PeriodicRunnerImpl::PeriodicJobImpl::stop() {
 
         _execStatus = PeriodicJobImpl::ExecutionStatus::CANCELED;
     }
+    invariant(_thread.joinable());
     _condvar.notify_one();
     _thread.join();
 }
@@ -167,7 +168,6 @@ bool PeriodicRunnerImpl::PeriodicJobImpl::isAlive() {
     return _execStatus == ExecutionStatus::RUNNING || _execStatus == ExecutionStatus::PAUSED;
 }
 
-
 namespace {
 
 template <typename T>
@@ -189,6 +189,11 @@ void PeriodicRunnerImpl::PeriodicJobHandleImpl::start() {
     job->start();
 }
 
+void PeriodicRunnerImpl::PeriodicJobHandleImpl::stop() {
+    auto job = lockAndAssertExists(_jobWeak, kPeriodicJobHandleLifetimeErrMsg);
+    job->stop();
+}
+
 void PeriodicRunnerImpl::PeriodicJobHandleImpl::pause() {
     auto job = lockAndAssertExists(_jobWeak, kPeriodicJobHandleLifetimeErrMsg);
     job->pause();
diff --git a/src/mongo/util/periodic_runner_impl.h b/src/mongo/util/periodic_runner_impl.h
index 49f7040e361..890c909fabb 100644
--- a/src/mongo/util/periodic_runner_impl.h
+++ b/src/mongo/util/periodic_runner_impl.h
@@ -97,6 +97,7 @@ private:
         explicit PeriodicJobHandleImpl(std::weak_ptr<PeriodicJobImpl> jobImpl)
             : _jobWeak(jobImpl) {}
         void start() override;
+        void stop() override;
         void pause() override;
         void resume() override;