Revert "Revert "SERVER-28562 Move LogicalTime HMAC computation outside collection lock""

This reverts commit b7c013aa097b2d999ad3f942cdfce130558ef40f.

22 files changed, 401 insertions, 273 deletions

diff --git a/src/mongo/db/SConscript b/src/mongo/db/SConscript
index 6aa7c0699d7..6cf5d02acca 100644
--- a/src/mongo/db/SConscript
+++ b/src/mongo/db/SConscript
@@ -973,11 +973,21 @@ env.Library(
         'logical_clock.cpp',
     ],
     LIBDEPS=[
+        'logical_time',
         'server_parameters',
         'service_context',
+    ],
+)
+
+env.Library(
+    target='logical_time_validator',
+    source=[
+        'logical_time_validator.cpp',
+    ],
+    LIBDEPS=[
+        'service_context',
         'signed_logical_time',
         'time_proof_service',
-        '$BUILD_DIR/mongo/util/clock_source_mock'
     ],
 )
 
@@ -1023,12 +1033,23 @@ env.CppUnitTest(
     ],
 )
 
+env.CppUnitTest(
+    target='logical_time_validator_test',
+    source=[
+        'logical_time_validator_test.cpp',
+    ],
+    LIBDEPS=[
+        'logical_time_validator',
+    ],
+)
+
 env.Library(
     target= 'logical_time_metadata_hook',
     source= [
         'logical_time_metadata_hook.cpp',
     ],
     LIBDEPS= [
+        'logical_time_validator',
         'signed_logical_time',
         '$BUILD_DIR/mongo/rpc/metadata',
     ],
@@ -1043,6 +1064,7 @@ env.Library(
         'logical_clock',
         'signed_logical_time',
         '$BUILD_DIR/mongo/s/sharding_mongod_test_fixture',
+        '$BUILD_DIR/mongo/util/clock_source_mock'
     ],
 )
 
diff --git a/src/mongo/db/commands/dbcommands.cpp b/src/mongo/db/commands/dbcommands.cpp
index da379905d84..537b8f1004e 100644
--- a/src/mongo/db/commands/dbcommands.cpp
+++ b/src/mongo/db/commands/dbcommands.cpp
@@ -76,6 +76,7 @@
 #include "mongo/db/keypattern.h"
 #include "mongo/db/lasterror.h"
 #include "mongo/db/logical_clock.h"
+#include "mongo/db/logical_time_validator.h"
 #include "mongo/db/matcher/extensions_callback_disallow_extensions.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/op_observer.h"
@@ -1322,9 +1323,12 @@ void appendReplyMetadata(OperationContext* opCtx,
                 .writeToMetadata(metadataBob);
         }
 
-        if (LogicalClock::get(opCtx)->canVerifyAndSign()) {
-            rpc::LogicalTimeMetadata logicalTimeMetadata(
-                LogicalClock::get(opCtx)->getClusterTime());
+        auto validator = LogicalTimeValidator::get(opCtx);
+        if (validator) {
+
+            auto currentTime =
+                validator->signLogicalTime(LogicalClock::get(opCtx)->getClusterTime());
+            rpc::LogicalTimeMetadata logicalTimeMetadata(currentTime);
             logicalTimeMetadata.writeToMetadata(metadataBob);
         }
     }
@@ -1717,7 +1721,7 @@ void mongo::execCommandDatabase(OperationContext* opCtx,
         //
         // TODO: SERVER-28445 change this to use computeOperationTime once the exception handling
         // path is moved into Command::run()
-        auto operationTime = LogicalClock::get(opCtx)->getClusterTime().getTime();
+        auto operationTime = LogicalClock::get(opCtx)->getClusterTime();
 
         // An uninitialized operation time means the cluster time is not propagated, so the
         // operation time should not be attached to the error response.
diff --git a/src/mongo/db/keys_collection_cache_reader_and_updater.cpp b/src/mongo/db/keys_collection_cache_reader_and_updater.cpp
index 7c959c5f6c0..965482ebf8e 100644
--- a/src/mongo/db/keys_collection_cache_reader_and_updater.cpp
+++ b/src/mongo/db/keys_collection_cache_reader_and_updater.cpp
@@ -78,7 +78,7 @@ KeysCollectionCacheReaderAndUpdater::KeysCollectionCacheReaderAndUpdater(
 StatusWith<KeysCollectionDocument> KeysCollectionCacheReaderAndUpdater::refresh(
     OperationContext* opCtx) {
 
-    auto currentTime = LogicalClock::get(opCtx)->getClusterTime().getTime();
+    auto currentTime = LogicalClock::get(opCtx)->getClusterTime();
     auto keyStatus = Grid::get(opCtx)->catalogClient(opCtx)->getNewKeys(
         opCtx, _purpose, currentTime, repl::ReadConcernLevel::kLocalReadConcern);
 
diff --git a/src/mongo/db/keys_collection_cache_reader_and_updater_test.cpp b/src/mongo/db/keys_collection_cache_reader_and_updater_test.cpp
index 540b4af8898..2639af95ced 100644
--- a/src/mongo/db/keys_collection_cache_reader_and_updater_test.cpp
+++ b/src/mongo/db/keys_collection_cache_reader_and_updater_test.cpp
@@ -66,7 +66,7 @@ TEST_F(CacheUpdaterTest, ShouldCreate2KeysFromEmpty) {
     KeysCollectionCacheReaderAndUpdater updater("dummy", Seconds(5));
 
     const LogicalTime currentTime(LogicalTime(Timestamp(100, 2)));
-    LogicalClock::get(operationContext())->initClusterTimeFromTrustedSource(currentTime);
+    LogicalClock::get(operationContext())->setClusterTimeFromTrustedSource(currentTime);
 
     {
         auto keyStatus = updater.refresh(operationContext());
@@ -99,7 +99,7 @@ TEST_F(CacheUpdaterTest, ShouldPropagateWriteError) {
     KeysCollectionCacheReaderAndUpdater updater("dummy", Seconds(5));
 
     const LogicalTime currentTime(LogicalTime(Timestamp(100, 2)));
-    LogicalClock::get(operationContext())->initClusterTimeFromTrustedSource(currentTime);
+    LogicalClock::get(operationContext())->setClusterTimeFromTrustedSource(currentTime);
 
     FailPointEnableBlock failWriteBlock("failCollectionInserts");
 
@@ -111,7 +111,7 @@ TEST_F(CacheUpdaterTest, ShouldCreateAnotherKeyIfOnlyOneKeyExists) {
     KeysCollectionCacheReaderAndUpdater updater("dummy", Seconds(5));
 
     LogicalClock::get(operationContext())
-        ->initClusterTimeFromTrustedSource(LogicalTime(Timestamp(100, 2)));
+        ->setClusterTimeFromTrustedSource(LogicalTime(Timestamp(100, 2)));
 
     KeysCollectionDocument origKey1(
         1, "dummy", TimeProofService::generateRandomKey(), LogicalTime(Timestamp(105, 0)));
@@ -129,7 +129,7 @@ TEST_F(CacheUpdaterTest, ShouldCreateAnotherKeyIfOnlyOneKeyExists) {
         ASSERT_EQ(Timestamp(105, 0), key1.getExpiresAt().asTimestamp());
     }
 
-    auto currentTime = LogicalClock::get(operationContext())->getClusterTime().getTime();
+    auto currentTime = LogicalClock::get(operationContext())->getClusterTime();
 
     {
         auto keyStatus = updater.refresh(operationContext());
@@ -165,7 +165,7 @@ TEST_F(CacheUpdaterTest, ShouldCreateAnotherKeyIfNoValidKeyAfterCurrent) {
     KeysCollectionCacheReaderAndUpdater updater("dummy", Seconds(5));
 
     LogicalClock::get(operationContext())
-        ->initClusterTimeFromTrustedSource(LogicalTime(Timestamp(108, 2)));
+        ->setClusterTimeFromTrustedSource(LogicalTime(Timestamp(108, 2)));
 
     KeysCollectionDocument origKey1(
         1, "dummy", TimeProofService::generateRandomKey(), LogicalTime(Timestamp(105, 0)));
@@ -193,7 +193,7 @@ TEST_F(CacheUpdaterTest, ShouldCreateAnotherKeyIfNoValidKeyAfterCurrent) {
         ASSERT_EQ(Timestamp(110, 0), key2.getExpiresAt().asTimestamp());
     }
 
-    auto currentTime = LogicalClock::get(operationContext())->getClusterTime().getTime();
+    auto currentTime = LogicalClock::get(operationContext())->getClusterTime();
 
     {
         auto keyStatus = updater.refresh(operationContext());
@@ -256,7 +256,7 @@ TEST_F(CacheUpdaterTest, ShouldCreate2KeysIfAllKeysAreExpired) {
     KeysCollectionCacheReaderAndUpdater updater("dummy", Seconds(5));
 
     LogicalClock::get(operationContext())
-        ->initClusterTimeFromTrustedSource(LogicalTime(Timestamp(120, 2)));
+        ->setClusterTimeFromTrustedSource(LogicalTime(Timestamp(120, 2)));
 
     KeysCollectionDocument origKey1(
         1, "dummy", TimeProofService::generateRandomKey(), LogicalTime(Timestamp(105, 0)));
@@ -284,7 +284,7 @@ TEST_F(CacheUpdaterTest, ShouldCreate2KeysIfAllKeysAreExpired) {
         ASSERT_EQ(Timestamp(110, 0), key2.getExpiresAt().asTimestamp());
     }
 
-    auto currentTime = LogicalClock::get(operationContext())->getClusterTime().getTime();
+    auto currentTime = LogicalClock::get(operationContext())->getClusterTime();
 
     {
         auto keyStatus = updater.refresh(operationContext());
@@ -360,7 +360,7 @@ TEST_F(CacheUpdaterTest, ShouldNotCreateNewKeyIfThereAre2UnexpiredKeys) {
     KeysCollectionCacheReaderAndUpdater updater("dummy", Seconds(5));
 
     LogicalClock::get(operationContext())
-        ->initClusterTimeFromTrustedSource(LogicalTime(Timestamp(100, 2)));
+        ->setClusterTimeFromTrustedSource(LogicalTime(Timestamp(100, 2)));
 
     KeysCollectionDocument origKey1(
         1, "dummy", TimeProofService::generateRandomKey(), LogicalTime(Timestamp(105, 0)));
diff --git a/src/mongo/db/keys_collection_manager.cpp b/src/mongo/db/keys_collection_manager.cpp
index f65e1163e8f..61cd43f73f1 100644
--- a/src/mongo/db/keys_collection_manager.cpp
+++ b/src/mongo/db/keys_collection_manager.cpp
@@ -243,7 +243,7 @@ void KeysCollectionManager::PeriodicRunner::_doPeriodicRefresh(ServiceContext* s
         auto latestKeyStatusWith = (*doRefresh)(opCtx.get());
         if (latestKeyStatusWith.getStatus().isOK()) {
             const auto& latestKey = latestKeyStatusWith.getValue();
-            auto currentTime = LogicalClock::get(service)->getClusterTime().getTime();
+            auto currentTime = LogicalClock::get(service)->getClusterTime();
 
             nextWakeup =
                 howMuchSleepNeedFor(currentTime, latestKey.getExpiresAt(), refreshInterval);
diff --git a/src/mongo/db/keys_collection_manager_test.cpp b/src/mongo/db/keys_collection_manager_test.cpp
index f2c19f6a6c5..28b6a3fc786 100644
--- a/src/mongo/db/keys_collection_manager_test.cpp
+++ b/src/mongo/db/keys_collection_manager_test.cpp
@@ -234,7 +234,7 @@ TEST_F(KeysManagerTest, ShouldCreateKeysIfKeyGeneratorEnabled) {
     keyManager()->startMonitoring(getServiceContext());
 
     const LogicalTime currentTime(LogicalTime(Timestamp(100, 0)));
-    LogicalClock::get(operationContext())->initClusterTimeFromTrustedSource(currentTime);
+    LogicalClock::get(operationContext())->setClusterTimeFromTrustedSource(currentTime);
 
     keyManager()->enableKeyGenerator(operationContext(), true);
 
@@ -250,7 +250,7 @@ TEST_F(KeysManagerTest, EnableModeFlipFlopStressTest) {
     keyManager()->startMonitoring(getServiceContext());
 
     const LogicalTime currentTime(LogicalTime(Timestamp(100, 0)));
-    LogicalClock::get(operationContext())->initClusterTimeFromTrustedSource(currentTime);
+    LogicalClock::get(operationContext())->setClusterTimeFromTrustedSource(currentTime);
 
     bool doEnable = true;
 
@@ -275,9 +275,8 @@ TEST_F(KeysManagerTest, ShouldStillBeAbleToUpdateCacheEvenIfItCantCreateKeys) {
         operationContext(), NamespaceString(KeysCollectionDocument::ConfigNS), origKey1.toBSON()));
 
     // Set the time to be very ahead so the updater will be forced to create new keys.
-    const LogicalTime currentTime(LogicalTime(Timestamp(20000, 0)));
-    const SignedLogicalTime fakeTime(currentTime, 2);
-    ASSERT_OK(LogicalClock::get(operationContext())->advanceClusterTimeFromTrustedSource(fakeTime));
+    const LogicalTime fakeTime(Timestamp(20000, 0));
+    LogicalClock::get(operationContext())->setClusterTimeFromTrustedSource(fakeTime);
 
     FailPointEnableBlock failWriteBlock("failCollectionInserts");
 
diff --git a/src/mongo/db/logical_clock.cpp b/src/mongo/db/logical_clock.cpp
index 6a4df1e6c60..c40324f7120 100644
--- a/src/mongo/db/logical_clock.cpp
+++ b/src/mongo/db/logical_clock.cpp
@@ -86,107 +86,35 @@ void LogicalClock::set(ServiceContext* service, std::unique_ptr<LogicalClock> cl
 
 LogicalClock::LogicalClock(ServiceContext* service) : _service(service) {}
 
-SignedLogicalTime LogicalClock::getClusterTime() {
+LogicalTime LogicalClock::getClusterTime() {
     stdx::lock_guard<stdx::mutex> lock(_mutex);
     return _clusterTime;
 }
 
-void LogicalClock::setTimeProofService(std::unique_ptr<TimeProofService> tps) {
+Status LogicalClock::advanceClusterTime(const LogicalTime newTime) {
     stdx::lock_guard<stdx::mutex> lock(_mutex);
-    _timeProofService = std::move(tps);
-
-    // Ensure a clock with a time proof service cannot have a cluster time without a proof to
-    // simplify reasoning about signed logical times.
-    if (!_clusterTime.getProof()) {
-        _clusterTime = _makeSignedLogicalTime_inlock(_clusterTime.getTime());
-    }
-}
-
-bool LogicalClock::canVerifyAndSign() {
-    stdx::lock_guard<stdx::mutex> lock(_mutex);
-    return !!_timeProofService;
-}
-
-SignedLogicalTime LogicalClock::_makeSignedLogicalTime_inlock(LogicalTime logicalTime) {
-    if (_timeProofService) {
-        // TODO: SERVER-28436 Implement KeysCollectionManager
-        // Replace dummy keyId with real id from key manager.
-        return SignedLogicalTime(
-            logicalTime, _timeProofService->getProof(logicalTime, _tempKey), 0);
-    }
-    return SignedLogicalTime(logicalTime, 0);
-}
-
-Status LogicalClock::advanceClusterTime(const SignedLogicalTime& newTime) {
-    stdx::lock_guard<stdx::mutex> lock(_mutex);
-    if (!_timeProofService) {
-        return Status(ErrorCodes::CannotVerifyAndSignLogicalTime,
-                      "Cannot accept logicalTime: " + newTime.getTime().toString() +
-                          ". May not be a part of a sharded cluster");
-    }
-
-    const auto& newLogicalTime = newTime.getTime();
-
-    // No need to check proof if no time was given.
-    if (newLogicalTime == LogicalTime::kUninitialized) {
-        return Status::OK();
-    }
-
-    const auto newProof = newTime.getProof();
-    // Logical time is only sent if a server's clock can verify and sign logical times, so any
-    // received logical times should have proofs.
-    invariant(newProof);
-
-    auto res = _timeProofService->checkProof(newLogicalTime, newProof.get(), _tempKey);
-    if (res != Status::OK()) {
-        return res;
-    }
 
     // The rate limiter check cannot be moved into _advanceClusterTime_inlock to avoid code
     // repetition because it shouldn't be called on direct oplog operations.
-    auto rateLimitStatus = _passesRateLimiter_inlock(newTime.getTime());
+    auto rateLimitStatus = _passesRateLimiter_inlock(newTime);
     if (!rateLimitStatus.isOK()) {
         return rateLimitStatus;
     }
 
-    return _advanceClusterTime_inlock(std::move(newTime));
-}
-
-Status LogicalClock::_advanceClusterTime_inlock(SignedLogicalTime newTime) {
-    if (newTime.getTime() > _clusterTime.getTime()) {
+    if (newTime > _clusterTime) {
         _clusterTime = newTime;
     }
 
     return Status::OK();
 }
 
-Status LogicalClock::advanceClusterTimeFromTrustedSource(SignedLogicalTime newTime) {
-    stdx::lock_guard<stdx::mutex> lock(_mutex);
-    // The rate limiter check cannot be moved into _advanceClusterTime_inlock to avoid code
-    // repetition because it shouldn't be called on direct oplog operations.
-    auto rateLimitStatus = _passesRateLimiter_inlock(newTime.getTime());
-    if (!rateLimitStatus.isOK()) {
-        return rateLimitStatus;
-    }
-
-    return _advanceClusterTime_inlock(std::move(newTime));
-}
-
-Status LogicalClock::signAndAdvanceClusterTime(LogicalTime newTime) {
-    stdx::lock_guard<stdx::mutex> lock(_mutex);
-    auto newSignedTime = _makeSignedLogicalTime_inlock(newTime);
-
-    return _advanceClusterTime_inlock(std::move(newSignedTime));
-}
-
 LogicalTime LogicalClock::reserveTicks(uint64_t nTicks) {
 
     invariant(nTicks > 0 && nTicks < (1U << 31));
 
     stdx::lock_guard<stdx::mutex> lock(_mutex);
 
-    LogicalTime clusterTime = _clusterTime.getTime();
-    LogicalTime nextClusterTime;
+    LogicalTime clusterTime = _clusterTime;
 
     const unsigned wallClockSecs =
         durationCount<Seconds>(_service->getFastClockSource()->now().toDurationSinceEpoch());
@@ -212,23 +140,25 @@ LogicalTime LogicalClock::reserveTicks(uint64_t nTicks) {
 
     // Save the next cluster time.
     clusterTime.addTicks(1);
-    nextClusterTime = clusterTime;
+    _clusterTime = clusterTime;
 
     // Add the rest of the requested ticks if needed.
     if (nTicks > 1) {
-        clusterTime.addTicks(nTicks - 1);
+        _clusterTime.addTicks(nTicks - 1);
     }
 
-    _clusterTime = _makeSignedLogicalTime_inlock(clusterTime);
-    return nextClusterTime;
+    return clusterTime;
 }
 
-void LogicalClock::initClusterTimeFromTrustedSource(LogicalTime newTime) {
+void LogicalClock::setClusterTimeFromTrustedSource(LogicalTime newTime) {
     stdx::lock_guard<stdx::mutex> lock(_mutex);
-    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_inlock(newTime);
+
+    if (newTime > _clusterTime) {
+        _clusterTime = newTime;
+    }
 }
 
 Status LogicalClock::_passesRateLimiter_inlock(LogicalTime newTime) {
diff --git a/src/mongo/db/logical_clock.h b/src/mongo/db/logical_clock.h
index 001d91eb9e6..6d0b73385e5 100644
--- a/src/mongo/db/logical_clock.h
+++ b/src/mongo/db/logical_clock.h
@@ -38,10 +38,7 @@ class OperationContext;
 
 /**
  * LogicalClock maintain the clusterTime for a clusterNode. Every cluster node in a replica set has
- * an instance of the LogicalClock installed as a ServiceContext decoration. LogicalClock owns the
- * TimeProofService that allows it to generate proofs to sign LogicalTime values and to validate the
- * proofs of SignedLogicalTime values.LogicalClock instance must be created before the instance
- * starts up.
+ * an instance of the LogicalClock installed as a ServiceContext decoration.
  */
 class LogicalClock {
 public:
@@ -59,40 +56,17 @@ public:
     LogicalClock(ServiceContext*);
 
     /**
-     * Attach a pointer to a TimeProofService to the logical clock. Will overwrite an existing
-     * pointer if a TimeProofService has already been attached.
-     */
-    void setTimeProofService(std::unique_ptr<TimeProofService>);
-
-    /**
-     * Returns true if a TimeProofService has been attached to the logical clock.
-     */
-    bool canVerifyAndSign();
-
-    /**
-     * The method sets clusterTime to the newTime if the newTime > _clusterTime and the newTime
-     * passes the rate check and proof validation.
-     * Returns an error if the newTime does not pass the rate check or proof validation,
-     * OK otherwise.
-     */
-    Status advanceClusterTime(const SignedLogicalTime&);
-
-    /**
-     * Similar to advanceClusterTime, but only does rate checking and not proof validation.
-     */
-    Status advanceClusterTimeFromTrustedSource(SignedLogicalTime newTime);
-
-    /**
-     * Similar to advanceClusterTimeFromTrustedSource, but also signs the new time. Note that this
-     * should only be used on trusted LogicalTime (for example, LogicalTime extracted from local
-     * oplog entry).
+     * The method sets current time to newTime if the newTime > current time and it passes the rate
+     * check.
+     *
+     * Returns an error if the newTime does not pass the rate check.
      */
-    Status signAndAdvanceClusterTime(LogicalTime newTime);
+    Status advanceClusterTime(const LogicalTime newTime);
 
     /**
      * Returns the current clusterTime.
      */
-    SignedLogicalTime getClusterTime();
+    LogicalTime getClusterTime();
 
     /**
      * Returns the next clusterTime value and provides a guarantee that any future call to
@@ -102,20 +76,13 @@ public:
     LogicalTime reserveTicks(uint64_t nTicks);
 
     /**
-     * Resets _clusterTime to the signed time created from newTime. Should be used at the
-     * initialization after reading the oplog. Must not be called on already initialized clock.
+     * Resets current time to newTime. Should only be used for initializing this clock from an
+     * oplog timestamp.
      */
-    void initClusterTimeFromTrustedSource(LogicalTime newTime);
+    void setClusterTimeFromTrustedSource(LogicalTime newTime);
 
 private:
     /**
-     * Utility to create valid SignedLogicalTime from LogicalTime.
-     */
-    SignedLogicalTime _makeSignedLogicalTime_inlock(LogicalTime);
-
-    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.
      */
@@ -123,18 +90,9 @@ private:
 
     ServiceContext* const _service;
 
-    // The mutex protects _clusterTime and _timeProofService.
+    // The mutex protects _clusterTime.
     stdx::mutex _mutex;
-    SignedLogicalTime _clusterTime;
-    std::unique_ptr<TimeProofService> _timeProofService;
-
-    /**
-     * Temporary key only used for unit tests.
-     *
-     * TODO: SERVER-28436 Implement KeysCollectionManager
-     * Remove _tempKey and its uses from logical clock, and pass actual key from key manager.
-     */
-    TimeProofService::Key _tempKey = {};
+    LogicalTime _clusterTime;
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/logical_clock_test.cpp b/src/mongo/db/logical_clock_test.cpp
index 7a9b2fc4a9b..dd259b7d7e2 100644
--- a/src/mongo/db/logical_clock_test.cpp
+++ b/src/mongo/db/logical_clock_test.cpp
@@ -37,8 +37,6 @@
 #include "mongo/db/logical_clock_test_fixture.h"
 #include "mongo/db/logical_time.h"
 #include "mongo/db/repl/replication_coordinator_mock.h"
-#include "mongo/db/signed_logical_time.h"
-#include "mongo/db/time_proof_service.h"
 #include "mongo/stdx/memory.h"
 #include "mongo/unittest/unittest.h"
 #include "mongo/util/clock_source_mock.h"
@@ -56,10 +54,10 @@ TEST_F(LogicalClockTest, roundtrip) {
     Timestamp tX(1);
     auto time = LogicalTime(tX);
 
-    getClock()->initClusterTimeFromTrustedSource(time);
+    getClock()->setClusterTimeFromTrustedSource(time);
     auto storedTime(getClock()->getClusterTime());
 
-    ASSERT_TRUE(storedTime.getTime() == time);
+    ASSERT_TRUE(storedTime == time);
 }
 
 // Verify the reserve ticks functionality.
@@ -69,10 +67,10 @@ TEST_F(LogicalClockTest, reserveTicks) {
 
     auto t1 = getClock()->reserveTicks(1);
     auto t2(getClock()->getClusterTime());
-    ASSERT_TRUE(t1 == t2.getTime());
+    ASSERT_TRUE(t1 == t2);
 
     // Make sure we synchronized with the wall clock.
-    ASSERT_TRUE(t2.getTime().asTimestamp().getSecs() == 10);
+    ASSERT_TRUE(t2.asTimestamp().getSecs() == 10);
 
     auto t3 = getClock()->reserveTicks(1);
     t1.addTicks(1);
@@ -87,9 +85,9 @@ TEST_F(LogicalClockTest, reserveTicks) {
     ASSERT_TRUE(t3 == t1);
 
     // Ensure overflow to a new second.
-    auto initTimeSecs = getClock()->getClusterTime().getTime().asTimestamp().getSecs();
+    auto initTimeSecs = getClock()->getClusterTime().asTimestamp().getSecs();
     getClock()->reserveTicks((1U << 31) - 1);
-    auto newTimeSecs = getClock()->getClusterTime().getTime().asTimestamp().getSecs();
+    auto newTimeSecs = getClock()->getClusterTime().asTimestamp().getSecs();
     ASSERT_TRUE(newTimeSecs == initTimeSecs + 1);
 }
 
@@ -97,10 +95,8 @@ TEST_F(LogicalClockTest, reserveTicks) {
 TEST_F(LogicalClockTest, advanceClusterTime) {
     auto t1 = getClock()->reserveTicks(1);
     t1.addTicks(100);
-    SignedLogicalTime l1 = makeSignedLogicalTime(t1);
-    ASSERT_OK(getClock()->advanceClusterTimeFromTrustedSource(l1));
-    auto l2(getClock()->getClusterTime());
-    ASSERT_TRUE(l1.getTime() == l2.getTime());
+    ASSERT_OK(getClock()->advanceClusterTime(t1));
+    ASSERT_TRUE(t1 == getClock()->getClusterTime());
 }
 
 // Verify rate limiter rejects logical times whose seconds values are too far ahead.
@@ -112,11 +108,9 @@ TEST_F(LogicalClockTest, RateLimiterRejectsLogicalTimesTooFarAhead) {
             durationCount<Seconds>(LogicalClock::kMaxAcceptableLogicalClockDrift) +
             10,  // Add 10 seconds to ensure limit is exceeded.
         1);
-    SignedLogicalTime l1 = makeSignedLogicalTime(LogicalTime(tooFarAheadTimestamp));
+    LogicalTime t1(tooFarAheadTimestamp);
 
-    ASSERT_EQ(ErrorCodes::ClusterTimeFailsRateLimiter, getClock()->advanceClusterTime(l1));
-    ASSERT_EQ(ErrorCodes::ClusterTimeFailsRateLimiter,
-              getClock()->advanceClusterTimeFromTrustedSource(l1));
+    ASSERT_EQ(ErrorCodes::ClusterTimeFailsRateLimiter, getClock()->advanceClusterTime(t1));
 }
 
 // Verify cluster time can be initialized to a very old time.
@@ -128,47 +122,9 @@ TEST_F(LogicalClockTest, InitFromTrustedSourceCanAcceptVeryOldLogicalTime) {
         durationCount<Seconds>(getMockClockSourceTime().toDurationSinceEpoch()) -
         (durationCount<Seconds>(LogicalClock::kMaxAcceptableLogicalClockDrift) * 5));
     auto veryOldTime = LogicalTime(veryOldTimestamp);
-    getClock()->initClusterTimeFromTrustedSource(veryOldTime);
+    getClock()->setClusterTimeFromTrustedSource(veryOldTime);
 
-    ASSERT_TRUE(getClock()->getClusterTime().getTime() == veryOldTime);
-}
-
-// A clock with no TimeProofService should reject new times in advanceClusterTime.
-TEST_F(LogicalClockTest, AdvanceClusterTimeFailsWithoutTimeProofService) {
-    LogicalTime initialTime(Timestamp(10));
-    getClock()->initClusterTimeFromTrustedSource(initialTime);
-
-    unsetTimeProofService();
-
-    SignedLogicalTime l1 = makeSignedLogicalTime(LogicalTime(Timestamp(100)));
-    ASSERT_EQ(ErrorCodes::CannotVerifyAndSignLogicalTime, getClock()->advanceClusterTime(l1));
-    ASSERT_TRUE(getClock()->getClusterTime().getTime() == initialTime);
-
-    resetTimeProofService();
-
-    SignedLogicalTime l2 = makeSignedLogicalTime(LogicalTime(Timestamp(200)));
-    ASSERT_OK(getClock()->advanceClusterTime(l2));
-    ASSERT_TRUE(getClock()->getClusterTime().getTime() == l2.getTime());
-}
-
-// A clock with no TimeProofService can still advance its time through certain methods.
-TEST_F(LogicalClockTest, CertainMethodsCanAdvanceClockWithoutTimeProofService) {
-    unsetTimeProofService();
-
-    LogicalTime t1(Timestamp(100));
-    getClock()->initClusterTimeFromTrustedSource(t1);
-    ASSERT_TRUE(getClock()->getClusterTime().getTime() == t1);
-
-    auto t2 = getClock()->reserveTicks(1);
-    ASSERT_TRUE(getClock()->getClusterTime().getTime() == t2);
-
-    LogicalTime t3(Timestamp(300));
-    ASSERT_OK(getClock()->signAndAdvanceClusterTime(t3));
-    ASSERT_TRUE(getClock()->getClusterTime().getTime() == t3);
-
-    SignedLogicalTime l4 = makeSignedLogicalTime(LogicalTime(Timestamp(400)));
-    ASSERT_OK(getClock()->advanceClusterTimeFromTrustedSource(l4));
-    ASSERT_TRUE(getClock()->getClusterTime().getTime() == l4.getTime());
+    ASSERT_TRUE(getClock()->getClusterTime() == veryOldTime);
 }
 
 // Verify writes to the oplog advance cluster time.
@@ -176,12 +132,12 @@ TEST_F(LogicalClockTest, WritesToOplogAdvanceClusterTime) {
     Timestamp tX(1);
     auto initialTime = LogicalTime(tX);
 
-    getClock()->initClusterTimeFromTrustedSource(initialTime);
-    ASSERT_TRUE(getClock()->getClusterTime().getTime() == initialTime);
+    getClock()->setClusterTimeFromTrustedSource(initialTime);
+    ASSERT_TRUE(getClock()->getClusterTime() == initialTime);
 
     getDBClient()->insert(kDummyNamespaceString, BSON("x" << 1));
-    ASSERT_TRUE(getClock()->getClusterTime().getTime() > initialTime);
-    ASSERT_EQ(getClock()->getClusterTime().getTime().asTimestamp(),
+    ASSERT_TRUE(getClock()->getClusterTime() > initialTime);
+    ASSERT_EQ(getClock()->getClusterTime().asTimestamp(),
               replicationCoordinator()->getMyLastAppliedOpTime().getTimestamp());
 }
 
diff --git a/src/mongo/db/logical_clock_test_fixture.cpp b/src/mongo/db/logical_clock_test_fixture.cpp
index d7f6e8036b6..588ff7dd766 100644
--- a/src/mongo/db/logical_clock_test_fixture.cpp
+++ b/src/mongo/db/logical_clock_test_fixture.cpp
@@ -56,10 +56,6 @@ void LogicalClockTestFixture::setUp() {
     LogicalClock::set(service, std::move(logicalClock));
     _clock = LogicalClock::get(service);
 
-    auto pTps = stdx::make_unique<TimeProofService>();
-    _timeProofService = pTps.get();
-    _clock->setTimeProofService(std::move(pTps));
-
     service->setFastClockSource(stdx::make_unique<SharedClockSourceAdapter>(_mockClockSource));
     service->setPreciseClockSource(stdx::make_unique<SharedClockSourceAdapter>(_mockClockSource));
 
@@ -92,21 +88,6 @@ Date_t LogicalClockTestFixture::getMockClockSourceTime() const {
     return _mockClockSource->now();
 }
 
-SignedLogicalTime LogicalClockTestFixture::makeSignedLogicalTime(LogicalTime logicalTime) const {
-    TimeProofService::Key key = {};
-    return SignedLogicalTime(logicalTime, _timeProofService->getProof(logicalTime, key), 0);
-}
-
-void LogicalClockTestFixture::resetTimeProofService() {
-    auto pTps = stdx::make_unique<TimeProofService>();
-    _timeProofService = pTps.get();
-    _clock->setTimeProofService(std::move(pTps));
-}
-
-void LogicalClockTestFixture::unsetTimeProofService() const {
-    _clock->setTimeProofService(std::unique_ptr<TimeProofService>());
-}
-
 DBDirectClient* LogicalClockTestFixture::getDBClient() const {
     return _dbDirectClient.get();
 }
diff --git a/src/mongo/db/logical_clock_test_fixture.h b/src/mongo/db/logical_clock_test_fixture.h
index 4c11f31e704..9a3115b79d3 100644
--- a/src/mongo/db/logical_clock_test_fixture.h
+++ b/src/mongo/db/logical_clock_test_fixture.h
@@ -37,8 +37,6 @@ class ClockSourceMock;
 class DBDirectClient;
 class LogicalClock;
 class LogicalTime;
-class SignedLogicalTime;
-class TimeProofService;
 
 /**
  * A test fixture that installs a LogicalClock instance with a TimeProofService onto a service
@@ -67,17 +65,10 @@ protected:
 
     Date_t getMockClockSourceTime() const;
 
-    SignedLogicalTime makeSignedLogicalTime(LogicalTime logicalTime) const;
-
-    void resetTimeProofService();
-
-    void unsetTimeProofService() const;
-
     DBDirectClient* getDBClient() const;
 
 private:
     LogicalClock* _clock;
-    TimeProofService* _timeProofService;
     std::shared_ptr<ClockSourceMock> _mockClockSource = std::make_shared<ClockSourceMock>();
     std::unique_ptr<DBDirectClient> _dbDirectClient;
 };
diff --git a/src/mongo/db/logical_time_metadata_hook.cpp b/src/mongo/db/logical_time_metadata_hook.cpp
index bb4124d611b..29949e36c82 100644
--- a/src/mongo/db/logical_time_metadata_hook.cpp
+++ b/src/mongo/db/logical_time_metadata_hook.cpp
@@ -31,6 +31,7 @@
 #include "mongo/db/logical_time_metadata_hook.h"
 
 #include "mongo/db/logical_clock.h"
+#include "mongo/db/logical_time_validator.h"
 #include "mongo/rpc/metadata/logical_time_metadata.h"
 #include "mongo/stdx/memory.h"
 
@@ -42,11 +43,13 @@ LogicalTimeMetadataHook::LogicalTimeMetadataHook(ServiceContext* service) : _ser
 
 Status LogicalTimeMetadataHook::writeRequestMetadata(OperationContext* opCtx,
                                                      BSONObjBuilder* metadataBob) {
-    if (!LogicalClock::get(_service)->canVerifyAndSign()) {
+    auto validator = LogicalTimeValidator::get(_service);
+    if (!validator) {
         return Status::OK();
     }
 
-    LogicalTimeMetadata metadata(LogicalClock::get(_service)->getClusterTime());
+    auto newTime = LogicalClock::get(_service)->getClusterTime();
+    LogicalTimeMetadata metadata(validator->signLogicalTime(newTime));
     metadata.writeToMetadata(metadataBob);
     return Status::OK();
 }
@@ -57,14 +60,21 @@ Status LogicalTimeMetadataHook::readReplyMetadata(StringData replySource,
     if (!parseStatus.isOK()) {
         return parseStatus.getStatus();
     }
+
     auto& signedTime = parseStatus.getValue().getSignedTime();
+
     // LogicalTimeMetadata is default constructed if no logical time metadata was sent, so a
     // default constructed SignedLogicalTime should be ignored.
     if (signedTime.getTime() == LogicalTime::kUninitialized) {
         return Status::OK();
     }
 
-    return LogicalClock::get(_service)->advanceClusterTimeFromTrustedSource(signedTime);
+    auto validator = LogicalTimeValidator::get(_service);
+    if (validator) {
+        validator->updateCacheTrustedSource(signedTime);
+    }
+
+    return LogicalClock::get(_service)->advanceClusterTime(signedTime.getTime());
 }
 
 }  // namespace rpc
diff --git a/src/mongo/db/logical_time_validator.cpp b/src/mongo/db/logical_time_validator.cpp
new file mode 100644
index 00000000000..3e6e4466742
--- /dev/null
+++ b/src/mongo/db/logical_time_validator.cpp
@@ -0,0 +1,110 @@
+/**
+ *    Copyright (C) 2017 MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/db/logical_time_validator.h"
+
+#include "mongo/db/operation_context.h"
+#include "mongo/db/service_context.h"
+
+namespace mongo {
+
+namespace {
+const auto getLogicalClockValidator =
+    ServiceContext::declareDecoration<std::unique_ptr<LogicalTimeValidator>>();
+stdx::mutex validatorMutex;
+
+// TODO: SERVER-28127 Implement KeysCollectionManager
+// Remove _tempKey and its uses from logical clock, and pass actual key from key manager.
+TimeProofService::Key tempKey = {};
+}
+
+LogicalTimeValidator* LogicalTimeValidator::get(ServiceContext* service) {
+    stdx::lock_guard<stdx::mutex> lk(validatorMutex);
+    return getLogicalClockValidator(service).get();
+}
+
+LogicalTimeValidator* LogicalTimeValidator::get(OperationContext* ctx) {
+    return get(ctx->getClient()->getServiceContext());
+}
+
+void LogicalTimeValidator::set(ServiceContext* service,
+                               std::unique_ptr<LogicalTimeValidator> newValidator) {
+    stdx::lock_guard<stdx::mutex> lk(validatorMutex);
+    auto& validator = getLogicalClockValidator(service);
+    validator = std::move(newValidator);
+}
+
+SignedLogicalTime LogicalTimeValidator::signLogicalTime(const LogicalTime& newTime) {
+    // Compare and calculate HMAC inside mutex to prevent multiple threads computing HMAC for the
+    // same logical time.
+    stdx::lock_guard<stdx::mutex> lk(_mutex);
+    if (newTime == _lastSeenValidTime.getTime()) {
+        return _lastSeenValidTime;
+    }
+
+    auto signature = _timeProofService.getProof(newTime, tempKey);
+    SignedLogicalTime newSignedTime(newTime, std::move(signature), 0);
+
+    if (newTime > _lastSeenValidTime.getTime()) {
+        _lastSeenValidTime = newSignedTime;
+    }
+
+    return newSignedTime;
+}
+
+Status LogicalTimeValidator::validate(const SignedLogicalTime& newTime) {
+    {
+        stdx::lock_guard<stdx::mutex> lk(_mutex);
+        if (newTime.getTime() == _lastSeenValidTime.getTime()) {
+            return Status::OK();
+        }
+    }
+
+    const auto newProof = newTime.getProof();
+    // Logical time is only sent if a server's clock can verify and sign logical times, so any
+    // received logical times should have proofs.
+    invariant(newProof);
+
+    auto res = _timeProofService.checkProof(newTime.getTime(), newProof.get(), tempKey);
+    if (res != Status::OK()) {
+        return res;
+    }
+
+    return Status::OK();
+}
+
+void LogicalTimeValidator::updateCacheTrustedSource(const SignedLogicalTime& newTime) {
+    stdx::lock_guard<stdx::mutex> lk(_mutex);
+    if (newTime.getTime() > _lastSeenValidTime.getTime()) {
+        _lastSeenValidTime = newTime;
+    }
+}
+
+}  // namespace mongo
diff --git a/src/mongo/db/logical_time_validator.h b/src/mongo/db/logical_time_validator.h
new file mode 100644
index 00000000000..74bf93077bf
--- /dev/null
+++ b/src/mongo/db/logical_time_validator.h
@@ -0,0 +1,75 @@
+/**
+ *    Copyright (C) 2017 MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include <memory>
+
+#include "mongo/db/signed_logical_time.h"
+#include "mongo/db/time_proof_service.h"
+#include "mongo/stdx/mutex.h"
+
+namespace mongo {
+
+class OperationContext;
+class ServiceContext;
+
+/**
+ * This is responsible for signing logical times that can be used to sent to other servers and
+ * verifying signatures of signed logical times.
+ */
+class LogicalTimeValidator {
+public:
+    // Decorate ServiceContext with LogicalTimeValidator instance.
+    static LogicalTimeValidator* get(ServiceContext* service);
+    static LogicalTimeValidator* get(OperationContext* ctx);
+    static void set(ServiceContext* service, std::unique_ptr<LogicalTimeValidator> validator);
+
+    /**
+     * Returns the newTime with a valid signature.
+     */
+    SignedLogicalTime signLogicalTime(const LogicalTime& newTime);
+
+    /**
+     * Returns true if the signature of newTime is valid.
+     */
+    Status validate(const SignedLogicalTime& newTime);
+
+    /**
+     * Saves the newTime if it is newer than the last seen valid LogicalTime without performing
+     * validation.
+     */
+    void updateCacheTrustedSource(const SignedLogicalTime& newTime);
+
+private:
+    stdx::mutex _mutex;
+    SignedLogicalTime _lastSeenValidTime;
+    TimeProofService _timeProofService;
+};
+
+}  // namespace mongo
diff --git a/src/mongo/db/logical_time_validator_test.cpp b/src/mongo/db/logical_time_validator_test.cpp
new file mode 100644
index 00000000000..3eb328c7044
--- /dev/null
+++ b/src/mongo/db/logical_time_validator_test.cpp
@@ -0,0 +1,85 @@
+/**
+ *    Copyright (C) 2017 MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/bson/timestamp.h"
+#include "mongo/db/logical_clock.h"
+#include "mongo/db/logical_time.h"
+#include "mongo/db/logical_time_validator.h"
+#include "mongo/db/service_context_noop.h"
+#include "mongo/db/signed_logical_time.h"
+#include "mongo/db/time_proof_service.h"
+#include "mongo/platform/basic.h"
+#include "mongo/stdx/memory.h"
+#include "mongo/unittest/unittest.h"
+#include "mongo/util/clock_source_mock.h"
+
+namespace mongo {
+namespace {
+
+TEST(LogicalTimeValidator, GetTimeWithIncreasingTimes) {
+    LogicalTimeValidator validator;
+
+    LogicalTime t1(Timestamp(10, 0));
+    auto newTime = validator.signLogicalTime(t1);
+
+    ASSERT_EQ(t1.asTimestamp(), newTime.getTime().asTimestamp());
+    ASSERT_TRUE(newTime.getProof());
+
+    LogicalTime t2(Timestamp(20, 0));
+    auto newTime2 = validator.signLogicalTime(t2);
+
+    ASSERT_EQ(t2.asTimestamp(), newTime2.getTime().asTimestamp());
+    ASSERT_TRUE(newTime2.getProof());
+}
+
+TEST(LogicalTimeValidator, ValidateReturnsOkForValidSignature) {
+    LogicalTimeValidator validator;
+
+    LogicalTime t1(Timestamp(20, 0));
+    auto newTime = validator.signLogicalTime(t1);
+
+    ASSERT_OK(validator.validate(newTime));
+}
+
+TEST(LogicalTimeValidator, ValidateErrorsOnInvalidTime) {
+    LogicalTimeValidator validator;
+
+    LogicalTime t1(Timestamp(20, 0));
+    auto newTime = validator.signLogicalTime(t1);
+
+    TimeProofService::TimeProof invalidProof = {{1, 2, 3}};
+    SignedLogicalTime invalidTime(LogicalTime(Timestamp(30, 0)), invalidProof, 0);
+
+    auto status = validator.validate(invalidTime);
+    ASSERT_EQ(ErrorCodes::TimeProofMismatch, status);
+}
+
+}  // unnamed namespace
+}  // namespace mongo
diff --git a/src/mongo/db/read_concern.cpp b/src/mongo/db/read_concern.cpp
index 2b3a0c7c184..03b278b22be 100644
--- a/src/mongo/db/read_concern.cpp
+++ b/src/mongo/db/read_concern.cpp
@@ -126,7 +126,7 @@ Status waitForReadConcern(OperationContext* opCtx, const repl::ReadConcernArgs&
 
     auto afterClusterTime = readConcernArgs.getArgsClusterTime();
     if (afterClusterTime) {
-        auto currentTime = LogicalClock::get(opCtx)->getClusterTime().getTime();
+        auto currentTime = LogicalClock::get(opCtx)->getClusterTime();
         if (currentTime < *afterClusterTime) {
             return {ErrorCodes::InvalidOptions,
                     "readConcern afterClusterTime must not be greater than clusterTime value"};
diff --git a/src/mongo/db/repl/oplog.cpp b/src/mongo/db/repl/oplog.cpp
index 0e9c293ab7c..8b8c14a7c32 100644
--- a/src/mongo/db/repl/oplog.cpp
+++ b/src/mongo/db/repl/oplog.cpp
@@ -1127,7 +1127,7 @@ Status applyCommand_inlock(OperationContext* opCtx,
 
 void setNewTimestamp(ServiceContext* service, const Timestamp& newTime) {
     stdx::lock_guard<stdx::mutex> lk(newOpMutex);
-    LogicalClock::get(service)->signAndAdvanceClusterTime(LogicalTime(newTime));
+    LogicalClock::get(service)->setClusterTimeFromTrustedSource(LogicalTime(newTime));
     lastSetTimestamp = newTime;
     newTimestampNotifier.notify_all();
 }
diff --git a/src/mongo/rpc/SConscript b/src/mongo/rpc/SConscript
index 19635a49756..3d7982e6a34 100644
--- a/src/mongo/rpc/SConscript
+++ b/src/mongo/rpc/SConscript
@@ -146,8 +146,8 @@ env.Clone().InjectModule("enterprise").Library(
         '$BUILD_DIR/mongo/bson/util/bson_extract',
         '$BUILD_DIR/mongo/client/read_preference',
         '$BUILD_DIR/mongo/db/auth/authcore',
+        '$BUILD_DIR/mongo/db/logical_time_validator',
         '$BUILD_DIR/mongo/db/repl/optime',
-        '$BUILD_DIR/mongo/db/logical_clock',
         '$BUILD_DIR/mongo/db/signed_logical_time',
         '$BUILD_DIR/mongo/util/decorable',
     ],
diff --git a/src/mongo/rpc/metadata.cpp b/src/mongo/rpc/metadata.cpp
index d0e837302c4..07cdfef9fca 100644
--- a/src/mongo/rpc/metadata.cpp
+++ b/src/mongo/rpc/metadata.cpp
@@ -38,6 +38,7 @@
 #include "mongo/db/auth/privilege.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/db/logical_clock.h"
+#include "mongo/db/logical_time_validator.h"
 #include "mongo/rpc/metadata/audit_metadata.h"
 #include "mongo/rpc/metadata/client_metadata_ismaster.h"
 #include "mongo/rpc/metadata/config_server_metadata.h"
@@ -131,7 +132,7 @@ Status readRequestMetadata(OperationContext* opCtx, const BSONObj& metadataObj)
 
     auto logicalClock = LogicalClock::get(opCtx);
     if (logicalClock) {
-        auto logicalTimeMetadata = LogicalTimeMetadata::readFromMetadata(logicalTimeElem);
+        auto logicalTimeMetadata = rpc::LogicalTimeMetadata::readFromMetadata(logicalTimeElem);
         if (!logicalTimeMetadata.isOK()) {
             return logicalTimeMetadata.getStatus();
         }
@@ -143,19 +144,24 @@ Status readRequestMetadata(OperationContext* opCtx, const BSONObj& metadataObj)
             return Status::OK();
         }
 
+        auto logicalTimeValidator = LogicalTimeValidator::get(opCtx);
         if (isAuthorizedToAdvanceClock(opCtx)) {
-            auto advanceClockStatus = logicalClock->advanceClusterTimeFromTrustedSource(signedTime);
-
-            if (!advanceClockStatus.isOK()) {
-                return advanceClockStatus;
+            if (logicalTimeValidator) {
+                logicalTimeValidator->updateCacheTrustedSource(signedTime);
             }
+        } else if (!logicalTimeValidator) {
+            return Status(ErrorCodes::CannotVerifyAndSignLogicalTime,
+                          "Cannot accept logicalTime: " + signedTime.getTime().toString() +
+                              ". May not be a part of a sharded cluster");
         } else {
-            auto advanceClockStatus = logicalClock->advanceClusterTime(signedTime);
+            auto advanceClockStatus = logicalTimeValidator->validate(signedTime);
 
             if (!advanceClockStatus.isOK()) {
                 return advanceClockStatus;
             }
         }
+
+        logicalClock->advanceClusterTime(signedTime.getTime());
     }
 
     return Status::OK();
diff --git a/src/mongo/rpc/metadata/logical_time_metadata.cpp b/src/mongo/rpc/metadata/logical_time_metadata.cpp
index 235fc0d99bb..d7ad869c9e8 100644
--- a/src/mongo/rpc/metadata/logical_time_metadata.cpp
+++ b/src/mongo/rpc/metadata/logical_time_metadata.cpp
@@ -105,7 +105,7 @@ void LogicalTimeMetadata::writeToMetadata(BSONObjBuilder* metadataBuilder) const
     _clusterTime.getTime().asTimestamp().append(subObjBuilder.bb(), kClusterTimeFieldName);
 
     BSONObjBuilder signatureObjBuilder(subObjBuilder.subobjStart(kSignatureFieldName));
-    // Logical time metadata is only written when LogicalClock::canVerifyAndSign returns true, which
+    // Logical time metadata is only written when the LogicalTimeValidator is set, which
     // means the cluster time should always have a proof.
     invariant(_clusterTime.getProof());
     _clusterTime.getProof()->appendAsBinData(signatureObjBuilder, kSignatureHashFieldName);
diff --git a/src/mongo/s/commands/strategy.cpp b/src/mongo/s/commands/strategy.cpp
index 25ac876dfd2..5c24d16bdc1 100644
--- a/src/mongo/s/commands/strategy.cpp
+++ b/src/mongo/s/commands/strategy.cpp
@@ -44,6 +44,7 @@
 #include "mongo/db/commands.h"
 #include "mongo/db/lasterror.h"
 #include "mongo/db/logical_clock.h"
+#include "mongo/db/logical_time_validator.h"
 #include "mongo/db/matcher/extensions_callback_noop.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/operation_time_tracker.h"
@@ -149,30 +150,29 @@ Status processCommandMetadata(OperationContext* opCtx, const BSONObj& cmdObj) {
     }
 
     auto authSession = AuthorizationSession::get(opCtx->getClient());
+    auto logicalTimeValidator = LogicalTimeValidator::get(opCtx);
+    const auto& signedTime = logicalTimeMetadata.getValue().getSignedTime();
+
     if (authSession->getAuthorizationManager().isAuthEnabled()) {
-        auto advanceClockStatus =
-            logicalClock->advanceClusterTime(logicalTimeMetadata.getValue().getSignedTime());
+        auto advanceClockStatus = logicalTimeValidator->validate(signedTime);
 
         if (!advanceClockStatus.isOK()) {
             return advanceClockStatus;
         }
     } else {
-        auto advanceClockStatus = logicalClock->advanceClusterTimeFromTrustedSource(
-            logicalTimeMetadata.getValue().getSignedTime());
-
-        if (!advanceClockStatus.isOK()) {
-            return advanceClockStatus;
-        }
+        logicalTimeValidator->updateCacheTrustedSource(signedTime);
     }
 
-    return Status::OK();
+    return logicalClock->advanceClusterTime(signedTime.getTime());
 }
 
 /**
  * Append required fields to command response.
  */
 void appendRequiredFieldsToResponse(OperationContext* opCtx, BSONObjBuilder* responseBuilder) {
-    rpc::LogicalTimeMetadata logicalTimeMetadata(LogicalClock::get(opCtx)->getClusterTime());
+    auto validator = LogicalTimeValidator::get(opCtx);
+    auto currentTime = validator->signLogicalTime(LogicalClock::get(opCtx)->getClusterTime());
+    rpc::LogicalTimeMetadata logicalTimeMetadata(currentTime);
     logicalTimeMetadata.writeToMetadata(responseBuilder);
     auto tracker = OperationTimeTracker::get(opCtx);
     if (tracker) {
diff --git a/src/mongo/s/sharding_initialization.cpp b/src/mongo/s/sharding_initialization.cpp
index 30403eb725e..9220176f29e 100644
--- a/src/mongo/s/sharding_initialization.cpp
+++ b/src/mongo/s/sharding_initialization.cpp
@@ -38,6 +38,7 @@
 #include "mongo/client/remote_command_targeter_factory_impl.h"
 #include "mongo/db/audit.h"
 #include "mongo/db/logical_clock.h"
+#include "mongo/db/logical_time_validator.h"
 #include "mongo/db/s/sharding_task_executor.h"
 #include "mongo/db/server_options.h"
 #include "mongo/db/server_parameters.h"
@@ -208,9 +209,6 @@ Status initializeGlobalShardingState(OperationContext* opCtx,
         std::move(executorPool),
         networkPtr);
 
-    auto timeProofService = stdx::make_unique<TimeProofService>();
-    LogicalClock::get(opCtx)->setTimeProofService(std::move(timeProofService));
-
     // must be started once the grid is initialized
     grid.shardRegistry()->startup(opCtx);
 
@@ -227,6 +225,9 @@ Status initializeGlobalShardingState(OperationContext* opCtx,
         }
     }
 
+    LogicalTimeValidator::set(opCtx->getServiceContext(),
+                              stdx::make_unique<LogicalTimeValidator>());
+
     return Status::OK();
 }