SERVER-47475 Make PeriodicJob aware of cancelation at global shutdown

(cherry picked from commit e7165a3e49beb6b6904ba5047301a142cccea646)

4 files changed, 54 insertions, 8 deletions

diff --git a/src/mongo/base/error_codes.err b/src/mongo/base/error_codes.err
index e003f808812..fddae0b50e7 100644
--- a/src/mongo/base/error_codes.err
+++ b/src/mongo/base/error_codes.err
@@ -293,6 +293,7 @@ error_code("WaitForMajorityServiceEarlierOpTimeAvailable", 289)
 # 290 is TransactionExceededLifetimeLimitSeconds, but not used.
 error_code("NoQueryExecutionPlans", 291)
 error_code("HierarchicalAcquisitionLevelViolation", 297)
+error_code("PeriodicJobIsStopped", 310)
 
 # Error codes 4000-8999 are reserved.
 
@@ -352,7 +353,7 @@ error_class("ShutdownError", ["ShutdownInProgress", "InterruptedAtShutdown"])
 
 # isCancelationError() includes all codes that, when passed to a function as its parameter,
 # indicates that it cannot be executed as normal and must abort its intended work.
-error_class("CancelationError", ["ShutdownInProgress", "InterruptedAtShutdown", "CallbackCanceled"])
+error_class("CancelationError", ["ShutdownInProgress", "InterruptedAtShutdown", "CallbackCanceled", "PeriodicJobIsStopped"])
 
 error_class("ConnectionFatalMessageParseError", ["IllegalOpMsgFlag",
                                                  "TooManyDocumentSequences",
diff --git a/src/mongo/db/db.cpp b/src/mongo/db/db.cpp
index 788fda46b59..008cec9d986 100644
--- a/src/mongo/db/db.cpp
+++ b/src/mongo/db/db.cpp
@@ -639,11 +639,19 @@ ExitCode _initAndListen(int listenPort) {
     // Only do this on storage engines supporting snapshot reads, which hold resources we wish to
     // release periodically in order to avoid storage cache pressure build up.
     if (storageEngine->supportsReadConcernSnapshot()) {
-        PeriodicThreadToAbortExpiredTransactions::get(serviceContext)->start();
-        // The inMemory engine is not yet used for replica or sharded transactions in production so
-        // it does not currently maintain snapshot history. It is live in testing, however.
-        if (!storageEngine->isEphemeral() || getTestCommandsEnabled()) {
-            PeriodicThreadToDecreaseSnapshotHistoryIfNotNeeded::get(serviceContext)->start();
+        try {
+            PeriodicThreadToAbortExpiredTransactions::get(serviceContext)->start();
+            // The inMemory engine is not yet used for replica or sharded transactions in production
+            // so it does not currently maintain snapshot history. It is live in testing, however.
+            if (!storageEngine->isEphemeral() || getTestCommandsEnabled()) {
+                PeriodicThreadToDecreaseSnapshotHistoryIfNotNeeded::get(serviceContext)->start();
+            }
+        } catch (ExceptionFor<ErrorCodes::PeriodicJobIsStopped>&) {
+            log() << "Not starting periodic jobs as shutdown is in progress";
+            // Shutdown has already started before initialization is complete. Wait for the
+            // shutdown task to complete and return.
+            MONGO_IDLE_THREAD_BLOCK;
+            return waitForShutdown();
         }
     }
 
diff --git a/src/mongo/util/periodic_runner_impl.cpp b/src/mongo/util/periodic_runner_impl.cpp
index dc9f091505d..88e83782957 100644
--- a/src/mongo/util/periodic_runner_impl.cpp
+++ b/src/mongo/util/periodic_runner_impl.cpp
@@ -33,8 +33,10 @@
 
 #include "mongo/util/periodic_runner_impl.h"
 
+#include "mongo/base/error_codes.h"
 #include "mongo/db/client.h"
 #include "mongo/db/service_context.h"
+#include "mongo/util/assert_util.h"
 #include "mongo/util/clock_source.h"
 #include "mongo/util/log.h"
 #include "mongo/util/scopeguard.h"
@@ -57,13 +59,15 @@ PeriodicRunnerImpl::PeriodicJobImpl::PeriodicJobImpl(PeriodicJob job,
     : _job(std::move(job)), _clockSource(source), _serviceContext(svc) {}
 
 void PeriodicRunnerImpl::PeriodicJobImpl::_run() {
-    auto [startPromise, startFuture] = makePromiseFuture<void>();
-
     {
         stdx::lock_guard lk(_mutex);
+        if (MONGO_unlikely(_execStatus == ExecutionStatus::CANCELED))
+            uasserted(ErrorCodes::PeriodicJobIsStopped,
+                      "Attempted to start an already stopped job");
         invariant(_execStatus == ExecutionStatus::NOT_SCHEDULED);
     }
 
+    auto [startPromise, startFuture] = makePromiseFuture<void>();
 
     _thread = stdx::thread([this, startPromise = std::move(startPromise)]() mutable {
         auto guard = makeGuard([this] { _stopPromise.emplaceValue(); });
@@ -121,6 +125,8 @@ void PeriodicRunnerImpl::PeriodicJobImpl::start() {
 
 void PeriodicRunnerImpl::PeriodicJobImpl::pause() {
     stdx::lock_guard<Latch> lk(_mutex);
+    if (MONGO_unlikely(_execStatus == ExecutionStatus::CANCELED))
+        uasserted(ErrorCodes::PeriodicJobIsStopped, "Attempted to pause an already stopped job");
     invariant(_execStatus == PeriodicJobImpl::ExecutionStatus::RUNNING);
     _execStatus = PeriodicJobImpl::ExecutionStatus::PAUSED;
 }
@@ -128,6 +134,9 @@ void PeriodicRunnerImpl::PeriodicJobImpl::pause() {
 void PeriodicRunnerImpl::PeriodicJobImpl::resume() {
     {
         stdx::lock_guard<Latch> lk(_mutex);
+        if (MONGO_unlikely(_execStatus == ExecutionStatus::CANCELED))
+            uasserted(ErrorCodes::PeriodicJobIsStopped,
+                      "Attempted to resume an already stopped job");
         invariant(_execStatus == PeriodicJobImpl::ExecutionStatus::PAUSED);
         _execStatus = PeriodicJobImpl::ExecutionStatus::RUNNING;
     }
diff --git a/src/mongo/util/periodic_runner_impl_test.cpp b/src/mongo/util/periodic_runner_impl_test.cpp
index 9dce8c05c87..4f3f8ee44b6 100644
--- a/src/mongo/util/periodic_runner_impl_test.cpp
+++ b/src/mongo/util/periodic_runner_impl_test.cpp
@@ -27,6 +27,9 @@
  *    it in the license file.
  */
 
+#include <boost/optional.hpp>
+
+#include "mongo/base/error_codes.h"
 #include "mongo/platform/basic.h"
 
 #include "mongo/unittest/unittest.h"
@@ -69,6 +72,14 @@ public:
     void setUp() override {
         PeriodicRunnerImplTestNoSetup::setUp();
     }
+
+    auto makeStoppedJob() {
+        PeriodicRunner::PeriodicJob job("job", [](Client* client) {}, Seconds{1});
+        auto jobAnchor = runner().makeJob(std::move(job));
+        jobAnchor.start();
+        jobAnchor.stop();
+        return jobAnchor;
+    }
 };
 
 TEST_F(PeriodicRunnerImplTest, OneJobTest) {
@@ -437,5 +448,22 @@ TEST_F(PeriodicRunnerImplTest, ChangingIntervalWorks) {
     tearDown();
 }
 
+TEST_F(PeriodicRunnerImplTest, ThrowsErrorOnceStopped) {
+    auto jobAnchor = makeStoppedJob();
+    ASSERT_THROWS_CODE_AND_WHAT(jobAnchor.start(),
+                                AssertionException,
+                                ErrorCodes::PeriodicJobIsStopped,
+                                "Attempted to start an already stopped job");
+    ASSERT_THROWS_CODE_AND_WHAT(jobAnchor.pause(),
+                                AssertionException,
+                                ErrorCodes::PeriodicJobIsStopped,
+                                "Attempted to pause an already stopped job");
+    ASSERT_THROWS_CODE_AND_WHAT(jobAnchor.resume(),
+                                AssertionException,
+                                ErrorCodes::PeriodicJobIsStopped,
+                                "Attempted to resume an already stopped job");
+    jobAnchor.stop();
+}
+
 }  // namespace
 }  // namespace mongo