SERVER-27721 Implement rate limiter check for advancing logical clocks

5 files changed, 94 insertions, 2 deletions

diff --git a/src/mongo/base/error_codes.err b/src/mongo/base/error_codes.err
index 947c21c0a50..eb78335dcaf 100644
--- a/src/mongo/base/error_codes.err
+++ b/src/mongo/base/error_codes.err
@@ -202,6 +202,7 @@ error_code("CannotBuildIndexKeys", 201)
 error_code("NetworkInterfaceExceededTimeLimit", 202)
 error_code("ShardingStateNotInitialized", 203)
 error_code("TimeProofMismatch", 204)
+error_code("ClusterTimeFailsRateLimiter", 205);
 
 # Error codes 4000-8999 are reserved.
 
diff --git a/src/mongo/db/SConscript b/src/mongo/db/SConscript
index 7ed8373af06..56792ae813d 100644
--- a/src/mongo/db/SConscript
+++ b/src/mongo/db/SConscript
@@ -1002,6 +1002,7 @@ env.Library(
         'logical_clock.cpp',
     ],
     LIBDEPS=[
+        'server_parameters',
         'service_context',
         'signed_logical_time',
         'time_proof_service',
diff --git a/src/mongo/db/logical_clock.cpp b/src/mongo/db/logical_clock.cpp
index 5e4150467a5..34a11fc163c 100644
--- a/src/mongo/db/logical_clock.cpp
+++ b/src/mongo/db/logical_clock.cpp
@@ -34,12 +34,39 @@
 
 #include "mongo/base/status.h"
 #include "mongo/db/operation_context.h"
+#include "mongo/db/server_parameters.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/time_proof_service.h"
 #include "mongo/util/log.h"
 
 namespace mongo {
 
+constexpr Seconds LogicalClock::kMaxAcceptableLogicalClockDrift;
+
+server_parameter_storage_type<long long, ServerParameterType::kStartupOnly>::value_type
+    maxAcceptableLogicalClockDrift(LogicalClock::kMaxAcceptableLogicalClockDrift.count());
+
+class MaxAcceptableLogicalClockDrift
+    : public ExportedServerParameter<long long, ServerParameterType::kStartupOnly> {
+public:
+    MaxAcceptableLogicalClockDrift()
+        : ExportedServerParameter<long long, ServerParameterType::kStartupOnly>(
+              ServerParameterSet::getGlobal(),
+              "maxAcceptableLogicalClockDrift",
+              &maxAcceptableLogicalClockDrift) {}
+
+    Status validate(const long long& potentialNewValue) {
+        if (potentialNewValue < 0) {
+            return Status(ErrorCodes::BadValue,
+                          str::stream() << "maxAcceptableLogicalClockDrift cannot be negative, but "
+                                           "attempted to set to: "
+                                        << potentialNewValue);
+        }
+
+        return Status::OK();
+    }
+} maxAcceptableLogicalClockDriftParameter;
+
 namespace {
 const auto getLogicalClock = ServiceContext::declareDecoration<std::unique_ptr<LogicalClock>>();
 }
@@ -83,8 +110,7 @@ Status LogicalClock::advanceClusterTime(const SignedLogicalTime& newTime) {
         return res;
     }
 
-    stdx::lock_guard<stdx::mutex> lock(_mutex);
-    return _advanceClusterTime_inlock(newTime);
+    return advanceClusterTimeFromTrustedSource(newTime);
 }
 
 Status LogicalClock::_advanceClusterTime_inlock(SignedLogicalTime newTime) {
@@ -97,6 +123,11 @@ Status LogicalClock::_advanceClusterTime_inlock(SignedLogicalTime newTime) {
 
 Status LogicalClock::advanceClusterTimeFromTrustedSource(SignedLogicalTime newTime) {
     stdx::lock_guard<stdx::mutex> lock(_mutex);
+    auto rateLimitStatus = _passesRateLimiter_inlock(newTime.getTime());
+    if (!rateLimitStatus.isOK()) {
+        return rateLimitStatus;
+    }
+
     return _advanceClusterTime_inlock(std::move(newTime));
 }
 
@@ -140,7 +171,27 @@ LogicalTime LogicalClock::reserveTicks(uint64_t ticks) {
 
 void LogicalClock::initClusterTimeFromTrustedSource(LogicalTime newTime) {
     invariant(_clusterTime.getTime() == LogicalTime::kUninitialized);
+    // Rate limit checks are skipped here so a server with no activity for longer than
+    // maxAcceptableLogicalClockDrift seconds can still have its cluster time initialized.
     _clusterTime = _makeSignedLogicalTime(newTime);
 }
 
+Status LogicalClock::_passesRateLimiter_inlock(LogicalTime newTime) {
+    const unsigned wallClockSecs =
+        durationCount<Seconds>(_service->getFastClockSource()->now().toDurationSinceEpoch());
+    auto maxAcceptableDrift = static_cast<const unsigned>(maxAcceptableLogicalClockDrift);
+    auto newTimeSecs = newTime.asTimestamp().getSecs();
+
+    // Both values are unsigned, so compare them first to avoid wrap-around.
+    if ((newTimeSecs > wallClockSecs) && (newTimeSecs - wallClockSecs) > maxAcceptableDrift) {
+        return Status(ErrorCodes::ClusterTimeFailsRateLimiter,
+                      str::stream() << "New cluster time, " << newTimeSecs
+                                    << ", is too far from this node's wall clock time, "
+                                    << wallClockSecs
+                                    << ".");
+    }
+
+    return Status::OK();
+}
+
 }  // namespace mongo
diff --git a/src/mongo/db/logical_clock.h b/src/mongo/db/logical_clock.h
index 0369eb9a641..aabe9a07c89 100644
--- a/src/mongo/db/logical_clock.h
+++ b/src/mongo/db/logical_clock.h
@@ -50,6 +50,9 @@ public:
     static LogicalClock* get(OperationContext* ctx);
     static void set(ServiceContext* service, std::unique_ptr<LogicalClock> logicalClock);
 
+    static constexpr Seconds kMaxAcceptableLogicalClockDrift =
+        Seconds(365 * 24 * 60 * 60);  // 1 year
+
     /**
      *  Creates an instance of LogicalClock. The TimeProofService must already be fully initialized.
      */
@@ -100,6 +103,12 @@ private:
 
     Status _advanceClusterTime_inlock(SignedLogicalTime newTime);
 
+    /**
+     * Rate limiter for advancing logical time. Rejects newTime if its seconds value is more than
+     * kMaxAcceptableLogicalClockDrift seconds ahead of this node's wall clock.
+     */
+    Status _passesRateLimiter_inlock(LogicalTime newTime);
+
     ServiceContext* const _service;
     std::unique_ptr<TimeProofService> _timeProofService;
 
diff --git a/src/mongo/db/logical_clock_test.cpp b/src/mongo/db/logical_clock_test.cpp
index bb8bdba9afc..a0b8dbaf30e 100644
--- a/src/mongo/db/logical_clock_test.cpp
+++ b/src/mongo/db/logical_clock_test.cpp
@@ -66,6 +66,11 @@ protected:
         return SignedLogicalTime(logicalTime, _timeProofService->getProof(logicalTime));
     }
 
+    const unsigned currentWallClockSecs() {
+        return durationCount<Seconds>(
+            _serviceContext->getFastClockSource()->now().toDurationSinceEpoch());
+    }
+
 private:
     TimeProofService* _timeProofService;
     std::unique_ptr<ServiceContextNoop> _serviceContext;
@@ -118,5 +123,30 @@ TEST_F(LogicalClockTestBase, advanceClusterTime) {
     ASSERT_TRUE(l1.getTime() == l2.getTime());
 }
 
+// Verify rate limiter rejects logical times whose seconds values are too far ahead.
+TEST_F(LogicalClockTestBase, RateLimiterRejectsLogicalTimesTooFarAhead) {
+    Timestamp tooFarAheadTimestamp(
+        currentWallClockSecs() +
+            durationCount<Seconds>(LogicalClock::kMaxAcceptableLogicalClockDrift) +
+            10,  // Add 10 seconds to ensure limit is exceeded.
+        1);
+    SignedLogicalTime l1 = makeSignedLogicalTime(LogicalTime(tooFarAheadTimestamp));
+
+    ASSERT_EQ(ErrorCodes::ClusterTimeFailsRateLimiter, getClock()->advanceClusterTime(l1));
+    ASSERT_EQ(ErrorCodes::ClusterTimeFailsRateLimiter,
+              getClock()->advanceClusterTimeFromTrustedSource(l1));
+}
+
+// Verify cluster time can be initialized to a very old time.
+TEST_F(LogicalClockTestBase, InitFromTrustedSourceCanAcceptVeryOldLogicalTime) {
+    Timestamp veryOldTimestamp(
+        currentWallClockSecs() -
+        (durationCount<Seconds>(LogicalClock::kMaxAcceptableLogicalClockDrift) * 5));
+    auto veryOldTime = LogicalTime(veryOldTimestamp);
+    getClock()->initClusterTimeFromTrustedSource(veryOldTime);
+
+    ASSERT_TRUE(getClock()->getClusterTime().getTime() == veryOldTime);
+}
+
 }  // unnamed namespace
 }  // namespace mongo