SERVER-47914 Move clusterTime from LogicalClock to VectorClock

60 files changed, 1608 insertions, 1258 deletions

diff --git a/src/mongo/SConscript b/src/mongo/SConscript
index 99b0c2c5949..5b1aecdb10d 100644
--- a/src/mongo/SConscript
+++ b/src/mongo/SConscript
@@ -541,7 +541,6 @@ env.Library(
         'db/kill_sessions',
         'db/kill_sessions_local',
         'db/log_process_details',
-        'db/logical_clock',
         'db/mirror_maestro',
         'db/mongod_options',
         'db/op_observer',
@@ -570,6 +569,7 @@ env.Library(
         'db/storage/storage_engine_common',
         'db/system_index',
         'db/ttl_d',
+        'db/vector_clock',
         'mongod_initializers',
         's/grid',
         's/sessions_collection_sharded',
diff --git a/src/mongo/db/SConscript b/src/mongo/db/SConscript
index 5ec20fc2915..04b218c1af6 100644
--- a/src/mongo/db/SConscript
+++ b/src/mongo/db/SConscript
@@ -400,7 +400,7 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/util/caching',
         '$BUILD_DIR/mongo/util/concurrency/thread_pool',
-        'logical_clock',
+        'vector_clock',
     ],
 )
 
@@ -1015,7 +1015,7 @@ env.Library(
     LIBDEPS=[
         '$BUILD_DIR/mongo/db/catalog/collection',
         'rw_concern_d',
-        'logical_clock',
+        'vector_clock',
     ],
     LIBDEPS_PRIVATE=[
         'index_builds_coordinator_interface',
@@ -1218,7 +1218,7 @@ env.Library(
     LIBDEPS=[
         '$BUILD_DIR/mongo/crypto/sha1_block',
         'keys_collection_document',
-        'logical_clock',
+        'vector_clock',
     ],
 )
 
@@ -1504,9 +1504,9 @@ env.Library(
         'keys_collection_client_direct.cpp',
     ],
     LIBDEPS=[
-        'logical_clock',
         'keys_collection_document',
         'logical_time',
+        '$BUILD_DIR/mongo/s/catalog/sharding_catalog_client',
         '$BUILD_DIR/mongo/s/client/rs_local_client',
         '$BUILD_DIR/mongo/s/client/shard_interface',
     ],
@@ -1518,7 +1518,6 @@ env.Library(
         'keys_collection_client_sharded.cpp',
     ],
     LIBDEPS=[
-        'logical_clock',
         'keys_collection_document',
         'logical_time',
         '$BUILD_DIR/mongo/s/catalog/sharding_catalog_client',
@@ -1526,45 +1525,31 @@ env.Library(
 )
 
 env.Library(
-    target='keys_collection_manager',
+    target='vector_clock',
     source=[
-        env.Idlc('keys_collection_manager.idl')[0],
-        'keys_collection_manager.cpp',
-        'keys_collection_cache.cpp',
         'key_generator.cpp',
-    ],
-    LIBDEPS=[
-        'logical_clock',
-        'keys_collection_document',
-        'logical_time',
-        'keys_collection_client_sharded',
-        'repl/repl_coordinator_interface',
-        '$BUILD_DIR/mongo/transport/transport_layer_common',
-    ],
-)
-
-env.Library(
-    target='logical_clock',
-    source=[
+        'keys_collection_cache.cpp',
+        'keys_collection_manager.cpp',
         'logical_clock.cpp',
+        'logical_time_validator.cpp',
+        'vector_clock.cpp',
+        env.Idlc('keys_collection_manager.idl')[0],
         env.Idlc('logical_clock.idl')[0],
     ],
     LIBDEPS=[
+        'auth/authprivilege',
         'global_settings',
+        'keys_collection_client_sharded',
+        'keys_collection_document',
         'logical_time',
+        'repl/repl_coordinator_interface',
         'service_context',
-        'vector_clock_mutable',
-    ],
-)
-
-env.Library(
-    target='vector_clock',
-    source=[
-        'vector_clock.cpp',
+        'signed_logical_time',
+        'time_proof_service',
+        '$BUILD_DIR/mongo/transport/transport_layer_common',
     ],
-    LIBDEPS=[
-        'logical_time',
-        'service_context',
+    LIBDEPS_PRIVATE=[
+        'server_options_core',
     ],
 )
 
@@ -1579,7 +1564,7 @@ env.Library(
 )
 
 env.Library(
-    target='vector_clock_d',
+    target='vector_clock_mongod',
     source=[
         'vector_clock_mongod.cpp',
     ],
@@ -1603,28 +1588,13 @@ env.Library(
 )
 
 env.Library(
-    target='logical_time_validator',
-    source=[
-        'logical_time_validator.cpp',
-    ],
-    LIBDEPS=[
-        'keys_collection_manager',
-        'service_context',
-        'signed_logical_time',
-        'time_proof_service',
-        '$BUILD_DIR/mongo/db/auth/auth',
-        '$BUILD_DIR/mongo/db/auth/authprivilege',
-    ],
-)
-
-env.Library(
     target= 'logical_time_metadata_hook',
     source= [
         'logical_time_metadata_hook.cpp',
     ],
     LIBDEPS= [
-        'logical_time_validator',
-        'signed_logical_time',
+        'logical_time',
+        'vector_clock',
         '$BUILD_DIR/mongo/rpc/metadata',
     ],
 )
@@ -1646,8 +1616,8 @@ env.Library(
         'logical_clock_test_fixture.cpp',
     ],
     LIBDEPS= [
-        'logical_clock',
         'signed_logical_time',
+        'vector_clock',
         '$BUILD_DIR/mongo/db/auth/authmocks',
         '$BUILD_DIR/mongo/s/sharding_mongod_test_fixture',
         '$BUILD_DIR/mongo/util/clock_source_mock'
@@ -1906,9 +1876,8 @@ envWithAsio.CppUnitTest(
         'dbmessage',
         'index_build_entry_helpers',
         'index_builds_coordinator_mongod',
+        'keys_collection_client_direct',
         'keys_collection_document',
-        'keys_collection_manager',
-        'logical_clock',
         'logical_clock_test_fixture',
         'logical_session_cache',
         'logical_session_cache_impl',
@@ -1946,6 +1915,7 @@ envWithAsio.CppUnitTest(
         'time_proof_service',
         'transaction',
         'update_index_data',
+        'vector_clock',
         'write_concern_options',
         'write_ops',
     ],
@@ -1966,8 +1936,7 @@ envWithAsio.CppUnitTest(
         '$BUILD_DIR/mongo/s/catalog/dist_lock_manager_mock',
         '$BUILD_DIR/mongo/s/config_server_test_fixture',
         'auth/authmocks',
-        'keys_collection_manager',
-        'logical_time_validator',
+        'vector_clock',
     ],
 )
 
diff --git a/src/mongo/db/catalog/SConscript b/src/mongo/db/catalog/SConscript
index 6aae6707afe..ac856017c24 100644
--- a/src/mongo/db/catalog/SConscript
+++ b/src/mongo/db/catalog/SConscript
@@ -187,9 +187,9 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception',
-        '$BUILD_DIR/mongo/db/logical_clock',
         '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface',
         '$BUILD_DIR/mongo/db/server_options_core',
+        '$BUILD_DIR/mongo/db/vector_clock',
     ]
 )
 
@@ -207,8 +207,8 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/index/index_build_interceptor',
-        '$BUILD_DIR/mongo/db/logical_clock',
         '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface',
+        '$BUILD_DIR/mongo/db/vector_clock',
     ]
 )
 
@@ -366,11 +366,11 @@ env.Library(
         '$BUILD_DIR/mongo/db/commands/server_status_core',
         '$BUILD_DIR/mongo/db/index/index_build_interceptor',
         '$BUILD_DIR/mongo/db/index/index_access_methods',
-        '$BUILD_DIR/mongo/db/logical_clock',
         '$BUILD_DIR/mongo/db/repl/repl_settings',
         '$BUILD_DIR/mongo/db/storage/storage_engine_common',
         '$BUILD_DIR/mongo/db/storage/storage_debug_util',
         '$BUILD_DIR/mongo/db/transaction',
+        '$BUILD_DIR/mongo/db/vector_clock',
         'index_build_block',
         'throttle_cursor',
         'validate_idl',
diff --git a/src/mongo/db/index/SConscript b/src/mongo/db/index/SConscript
index eaa3ebaaf40..84f2727d177 100644
--- a/src/mongo/db/index/SConscript
+++ b/src/mongo/db/index/SConscript
@@ -139,7 +139,7 @@ serveronlyEnv.Library(
     ],
     LIBDEPS_PRIVATE=[
         'skipped_record_tracker',
-        '$BUILD_DIR/mongo/db/logical_clock',
+        '$BUILD_DIR/mongo/db/vector_clock',
         '$BUILD_DIR/mongo/idl/server_parameter',
     ],
 )
diff --git a/src/mongo/db/key_generator_update_test.cpp b/src/mongo/db/key_generator_update_test.cpp
index 885d7dfe25b..511c9e493fc 100644
--- a/src/mongo/db/key_generator_update_test.cpp
+++ b/src/mongo/db/key_generator_update_test.cpp
@@ -37,7 +37,7 @@
 #include "mongo/db/key_generator.h"
 #include "mongo/db/keys_collection_client_sharded.h"
 #include "mongo/db/keys_collection_document.h"
-#include "mongo/db/logical_clock.h"
+#include "mongo/db/vector_clock_mutable.h"
 #include "mongo/s/catalog/dist_lock_manager_mock.h"
 #include "mongo/s/config_server_test_fixture.h"
 #include "mongo/unittest/unittest.h"
@@ -79,7 +79,8 @@ TEST_F(KeyGeneratorUpdateTest, ShouldCreate2KeysFromEmpty) {
     KeyGenerator generator("dummy", catalogClient(), Seconds(5));
 
     const LogicalTime currentTime(LogicalTime(Timestamp(100, 2)));
-    LogicalClock::get(operationContext())->setClusterTimeFromTrustedSource(currentTime);
+    VectorClockMutable::get(operationContext())
+        ->tickTo(VectorClock::Component::ClusterTime, currentTime);
 
     auto generateStatus = generator.generateNewKeysIfNeeded(operationContext());
     ASSERT_OK(generateStatus);
@@ -105,7 +106,8 @@ TEST_F(KeyGeneratorUpdateTest, ShouldPropagateWriteError) {
     KeyGenerator generator("dummy", catalogClient(), Seconds(5));
 
     const LogicalTime currentTime(LogicalTime(Timestamp(100, 2)));
-    LogicalClock::get(operationContext())->setClusterTimeFromTrustedSource(currentTime);
+    VectorClockMutable::get(operationContext())
+        ->tickTo(VectorClock::Component::ClusterTime, currentTime);
 
     FailPointEnableBlock failWriteBlock("failCollectionInserts");
 
@@ -116,8 +118,8 @@ TEST_F(KeyGeneratorUpdateTest, ShouldPropagateWriteError) {
 TEST_F(KeyGeneratorUpdateTest, ShouldCreateAnotherKeyIfOnlyOneKeyExists) {
     KeyGenerator generator("dummy", catalogClient(), Seconds(5));
 
-    LogicalClock::get(operationContext())
-        ->setClusterTimeFromTrustedSource(LogicalTime(Timestamp(100, 2)));
+    VectorClockMutable::get(operationContext())
+        ->tickTo(VectorClock::Component::ClusterTime, LogicalTime(Timestamp(100, 2)));
 
     KeysCollectionDocument origKey1(
         1, "dummy", TimeProofService::generateRandomKey(), LogicalTime(Timestamp(105, 0)));
@@ -135,7 +137,8 @@ TEST_F(KeyGeneratorUpdateTest, ShouldCreateAnotherKeyIfOnlyOneKeyExists) {
         ASSERT_EQ(Timestamp(105, 0), key1.getExpiresAt().asTimestamp());
     }
 
-    auto currentTime = LogicalClock::get(operationContext())->getClusterTime();
+    auto currentTime =
+        VectorClock::get(operationContext())->getTime()[VectorClock::Component::ClusterTime];
 
     auto generateStatus = generator.generateNewKeysIfNeeded(operationContext());
     ASSERT_OK(generateStatus);
@@ -163,8 +166,8 @@ TEST_F(KeyGeneratorUpdateTest, ShouldCreateAnotherKeyIfOnlyOneKeyExists) {
 TEST_F(KeyGeneratorUpdateTest, ShouldCreateAnotherKeyIfNoValidKeyAfterCurrent) {
     KeyGenerator generator("dummy", catalogClient(), Seconds(5));
 
-    LogicalClock::get(operationContext())
-        ->setClusterTimeFromTrustedSource(LogicalTime(Timestamp(108, 2)));
+    VectorClockMutable::get(operationContext())
+        ->tickTo(VectorClock::Component::ClusterTime, LogicalTime(Timestamp(108, 2)));
 
     KeysCollectionDocument origKey1(
         1, "dummy", TimeProofService::generateRandomKey(), LogicalTime(Timestamp(105, 0)));
@@ -192,7 +195,8 @@ TEST_F(KeyGeneratorUpdateTest, ShouldCreateAnotherKeyIfNoValidKeyAfterCurrent) {
         ASSERT_EQ(Timestamp(110, 0), key2.getExpiresAt().asTimestamp());
     }
 
-    auto currentTime = LogicalClock::get(operationContext())->getClusterTime();
+    auto currentTime =
+        VectorClock::get(operationContext())->getTime()[VectorClock::Component::ClusterTime];
 
     auto generateStatus = generator.generateNewKeysIfNeeded(operationContext());
     ASSERT_OK(generateStatus);
@@ -246,8 +250,8 @@ TEST_F(KeyGeneratorUpdateTest, ShouldCreateAnotherKeyIfNoValidKeyAfterCurrent) {
 TEST_F(KeyGeneratorUpdateTest, ShouldCreate2KeysIfAllKeysAreExpired) {
     KeyGenerator generator("dummy", catalogClient(), Seconds(5));
 
-    LogicalClock::get(operationContext())
-        ->setClusterTimeFromTrustedSource(LogicalTime(Timestamp(120, 2)));
+    VectorClockMutable::get(operationContext())
+        ->tickTo(VectorClock::Component::ClusterTime, LogicalTime(Timestamp(120, 2)));
 
     KeysCollectionDocument origKey1(
         1, "dummy", TimeProofService::generateRandomKey(), LogicalTime(Timestamp(105, 0)));
@@ -275,7 +279,8 @@ TEST_F(KeyGeneratorUpdateTest, ShouldCreate2KeysIfAllKeysAreExpired) {
         ASSERT_EQ(Timestamp(110, 0), key2.getExpiresAt().asTimestamp());
     }
 
-    auto currentTime = LogicalClock::get(operationContext())->getClusterTime();
+    auto currentTime =
+        VectorClock::get(operationContext())->getTime()[VectorClock::Component::ClusterTime];
 
     auto generateStatus = generator.generateNewKeysIfNeeded(operationContext());
     ASSERT_OK(generateStatus);
@@ -344,7 +349,8 @@ TEST_F(KeyGeneratorUpdateTest, ShouldNotCreateNewKeyIfThereAre2UnexpiredKeys) {
     KeyGenerator generator("dummy", catalogClient(), Seconds(5));
 
     const LogicalTime currentTime(LogicalTime(Timestamp(100, 2)));
-    LogicalClock::get(operationContext())->setClusterTimeFromTrustedSource(currentTime);
+    VectorClockMutable::get(operationContext())
+        ->tickTo(VectorClock::Component::ClusterTime, currentTime);
 
     KeysCollectionDocument origKey1(
         1, "dummy", TimeProofService::generateRandomKey(), LogicalTime(Timestamp(105, 0)));
@@ -403,7 +409,8 @@ TEST_F(KeyGeneratorUpdateTest, ShouldNotCreateKeysWithDisableKeyGenerationFailPo
     KeyGenerator generator("dummy", catalogClient(), Seconds(5));
 
     const LogicalTime currentTime(LogicalTime(Timestamp(100, 0)));
-    LogicalClock::get(operationContext())->setClusterTimeFromTrustedSource(currentTime);
+    VectorClockMutable::get(operationContext())
+        ->tickTo(VectorClock::Component::ClusterTime, currentTime);
 
     {
         FailPointEnableBlock failKeyGenerationBlock("disableKeyGeneration");
diff --git a/src/mongo/db/keys_collection_manager_sharding_test.cpp b/src/mongo/db/keys_collection_manager_sharding_test.cpp
index f6d4fc53a97..dd6e7b52f06 100644
--- a/src/mongo/db/keys_collection_manager_sharding_test.cpp
+++ b/src/mongo/db/keys_collection_manager_sharding_test.cpp
@@ -37,7 +37,7 @@
 #include "mongo/db/keys_collection_client_sharded.h"
 #include "mongo/db/keys_collection_document.h"
 #include "mongo/db/keys_collection_manager.h"
-#include "mongo/db/logical_clock.h"
+#include "mongo/db/vector_clock_mutable.h"
 #include "mongo/s/catalog/dist_lock_manager_mock.h"
 #include "mongo/s/config_server_test_fixture.h"
 #include "mongo/unittest/unittest.h"
@@ -261,7 +261,8 @@ TEST_F(KeysManagerShardedTest, ShouldCreateKeysIfKeyGeneratorEnabled) {
     keyManager()->startMonitoring(getServiceContext());
 
     const LogicalTime currentTime(LogicalTime(Timestamp(100, 0)));
-    LogicalClock::get(operationContext())->setClusterTimeFromTrustedSource(currentTime);
+    VectorClockMutable::get(operationContext())
+        ->tickTo(VectorClock::Component::ClusterTime, currentTime);
 
     keyManager()->enableKeyGenerator(operationContext(), true);
     keyManager()->refreshNow(operationContext());
@@ -277,7 +278,8 @@ TEST_F(KeysManagerShardedTest, EnableModeFlipFlopStressTest) {
     keyManager()->startMonitoring(getServiceContext());
 
     const LogicalTime currentTime(LogicalTime(Timestamp(100, 0)));
-    LogicalClock::get(operationContext())->setClusterTimeFromTrustedSource(currentTime);
+    VectorClockMutable::get(operationContext())
+        ->tickTo(VectorClock::Component::ClusterTime, currentTime);
 
     bool doEnable = true;
 
@@ -303,7 +305,8 @@ TEST_F(KeysManagerShardedTest, ShouldStillBeAbleToUpdateCacheEvenIfItCantCreateK
 
     // Set the time to be very ahead so the updater will be forced to create new keys.
     const LogicalTime fakeTime(Timestamp(20000, 0));
-    LogicalClock::get(operationContext())->setClusterTimeFromTrustedSource(fakeTime);
+    VectorClockMutable::get(operationContext())
+        ->tickTo(VectorClock::Component::ClusterTime, fakeTime);
 
     FailPointEnableBlock failWriteBlock("failCollectionInserts");
 
@@ -325,7 +328,8 @@ TEST_F(KeysManagerShardedTest, ShouldStillBeAbleToUpdateCacheEvenIfItCantCreateK
 
 TEST_F(KeysManagerShardedTest, ShouldNotCreateKeysWithDisableKeyGenerationFailPoint) {
     const LogicalTime currentTime(Timestamp(100, 0));
-    LogicalClock::get(operationContext())->setClusterTimeFromTrustedSource(currentTime);
+    VectorClockMutable::get(operationContext())
+        ->tickTo(VectorClock::Component::ClusterTime, currentTime);
 
     {
         FailPointEnableBlock failKeyGenerationBlock("disableKeyGeneration");
@@ -346,7 +350,8 @@ TEST_F(KeysManagerShardedTest, ShouldNotCreateKeysWithDisableKeyGenerationFailPo
 
 TEST_F(KeysManagerShardedTest, HasSeenKeysIsFalseUntilKeysAreFound) {
     const LogicalTime currentTime(Timestamp(100, 0));
-    LogicalClock::get(operationContext())->setClusterTimeFromTrustedSource(currentTime);
+    VectorClockMutable::get(operationContext())
+        ->tickTo(VectorClock::Component::ClusterTime, currentTime);
 
     ASSERT_EQ(false, keyManager()->hasSeenKeys());
 
diff --git a/src/mongo/db/logical_clock.cpp b/src/mongo/db/logical_clock.cpp
index 3cb7dc1b192..1f99fde99d9 100644
--- a/src/mongo/db/logical_clock.cpp
+++ b/src/mongo/db/logical_clock.cpp
@@ -32,25 +32,19 @@
 #include "mongo/platform/basic.h"
 
 #include "mongo/db/logical_clock.h"
-#include "mongo/db/logical_clock_gen.h"
 
 #include "mongo/base/status.h"
 #include "mongo/db/global_settings.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/time_proof_service.h"
-#include "mongo/db/vector_clock_mutable.h"
+#include "mongo/db/vector_clock.h"
 #include "mongo/logv2/log.h"
 
 namespace mongo {
 
 namespace {
 const auto getLogicalClock = ServiceContext::declareDecoration<std::unique_ptr<LogicalClock>>();
-
-bool lessThanOrEqualToMaxPossibleTime(LogicalTime time, uint64_t nTicks) {
-    return time.asTimestamp().getSecs() <= LogicalClock::kMaxSignedInt &&
-        time.asTimestamp().getInc() <= (LogicalClock::kMaxSignedInt - nTicks);
-}
 }  // namespace
 
 LogicalTime LogicalClock::getClusterTimeForReplicaSet(ServiceContext* svcCtx) {
@@ -81,116 +75,7 @@ void LogicalClock::set(ServiceContext* service, std::unique_ptr<LogicalClock> cl
 LogicalClock::LogicalClock(ServiceContext* service) : _service(service) {}
 
 LogicalTime LogicalClock::getClusterTime() {
-    stdx::lock_guard<Latch> lock(_mutex);
-    return _clusterTime;
-}
-
-Status LogicalClock::advanceClusterTime(const LogicalTime newTime) {
-    stdx::lock_guard<Latch> lock(_mutex);
-
-    auto rateLimitStatus = _passesRateLimiter_inlock(newTime);
-    if (!rateLimitStatus.isOK()) {
-        return rateLimitStatus;
-    }
-
-    if (newTime > _clusterTime) {
-        _clusterTime = newTime;
-    }
-
-    return Status::OK();
-}
-
-LogicalTime LogicalClock::reserveTicks(uint64_t nTicks) {
-    (void)VectorClockMutable::get(_service)->tick(VectorClock::Component::ClusterTime, nTicks);
-
-    invariant(nTicks > 0 && nTicks <= kMaxSignedInt);
-
-    stdx::lock_guard<Latch> lock(_mutex);
-
-    LogicalTime clusterTime = _clusterTime;
-
-    const unsigned wallClockSecs =
-        durationCount<Seconds>(_service->getFastClockSource()->now().toDurationSinceEpoch());
-    unsigned clusterTimeSecs = clusterTime.asTimestamp().getSecs();
-
-    // Synchronize clusterTime with wall clock time, if clusterTime was behind in seconds.
-    if (clusterTimeSecs < wallClockSecs) {
-        clusterTime = LogicalTime(Timestamp(wallClockSecs, 0));
-    }
-    // If reserving 'nTicks' would force the cluster timestamp's increment field to exceed (2^31-1),
-    // overflow by moving to the next second. We use the signed integer maximum as an overflow point
-    // in order to preserve compatibility with potentially signed or unsigned integral Timestamp
-    // increment types. It is also unlikely to apply more than 2^31 oplog entries in the span of one
-    // second.
-    else if (clusterTime.asTimestamp().getInc() > (kMaxSignedInt - nTicks)) {
-
-        LOGV2(20709,
-              "Exceeded maximum allowable increment value within one second. Moving clusterTime "
-              "forward to the next second.");
-
-        // Move time forward to the next second
-        clusterTime = LogicalTime(Timestamp(clusterTime.asTimestamp().getSecs() + 1, 0));
-    }
-
-    uassert(40482,
-            "cluster time cannot be advanced beyond its maximum value",
-            lessThanOrEqualToMaxPossibleTime(clusterTime, nTicks));
-
-    // Save the next cluster time.
-    clusterTime.addTicks(1);
-    _clusterTime = clusterTime;
-
-    // Add the rest of the requested ticks if needed.
-    if (nTicks > 1) {
-        _clusterTime.addTicks(nTicks - 1);
-    }
-
-    return clusterTime;
-}
-
-void LogicalClock::setClusterTimeFromTrustedSource(LogicalTime newTime) {
-    stdx::lock_guard<Latch> lock(_mutex);
-    // Rate limit checks are skipped here so a server with no activity for longer than
-    // maxAcceptableLogicalClockDriftSecs seconds can still have its cluster time initialized.
-
-    uassert(40483,
-            "cluster time cannot be advanced beyond its maximum value",
-            lessThanOrEqualToMaxPossibleTime(newTime, 0));
-
-    if (newTime > _clusterTime) {
-        _clusterTime = newTime;
-    }
-}
-
-Status LogicalClock::_passesRateLimiter_inlock(LogicalTime newTime) {
-    const unsigned wallClockSecs =
-        durationCount<Seconds>(_service->getFastClockSource()->now().toDurationSinceEpoch());
-    auto maxAcceptableDriftSecs = static_cast<const unsigned>(gMaxAcceptableLogicalClockDriftSecs);
-    auto newTimeSecs = newTime.asTimestamp().getSecs();
-
-    // Both values are unsigned, so compare them first to avoid wrap-around.
-    if ((newTimeSecs > wallClockSecs) && (newTimeSecs - wallClockSecs) > maxAcceptableDriftSecs) {
-        return Status(ErrorCodes::ClusterTimeFailsRateLimiter,
-                      str::stream() << "New cluster time, " << newTimeSecs
-                                    << ", is too far from this node's wall clock time, "
-                                    << wallClockSecs << ".");
-    }
-
-    uassert(40484,
-            "cluster time cannot be advanced beyond its maximum value",
-            lessThanOrEqualToMaxPossibleTime(newTime, 0));
-
-    return Status::OK();
-}
-
-bool LogicalClock::isEnabled() const {
-    stdx::lock_guard<Latch> lock(_mutex);
-    return _isEnabled;
-}
-
-void LogicalClock::disable() {
-    stdx::lock_guard<Latch> lock(_mutex);
-    _isEnabled = false;
+    return VectorClock::get(_service)->getTime()[VectorClock::Component::ClusterTime];
 }
 
 }  // namespace mongo
diff --git a/src/mongo/db/logical_clock.h b/src/mongo/db/logical_clock.h
index ec4bddc518a..059a9dcdfd6 100644
--- a/src/mongo/db/logical_clock.h
+++ b/src/mongo/db/logical_clock.h
@@ -36,9 +36,12 @@ namespace mongo {
 class ServiceContext;
 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 provides a legacy interface to the cluster time, which is now provided by the
+ * VectorClock class.
+ *
+ * TODO SERVER-48433: Remove this legacy LogicalClock interface.
  */
 class LogicalClock {
 public:
@@ -47,8 +50,6 @@ public:
     static LogicalClock* get(OperationContext* ctx);
     static void set(ServiceContext* service, std::unique_ptr<LogicalClock> logicalClock);
 
-    static const uint32_t kMaxSignedInt = ((1U << 31) - 1);
-
     /**
      * Returns the current cluster time if this is a replica set node, otherwise returns a null
      * logical time.
@@ -62,55 +63,12 @@ public:
     LogicalClock(ServiceContext*);
 
     /**
-     * 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 advanceClusterTime(const LogicalTime newTime);
-
-    /**
      * Returns the current clusterTime.
      */
     LogicalTime getClusterTime();
 
-    /**
-     * Returns the next clusterTime value and provides a guarantee that any future call to
-     * reserveTicks() will return a value at least 'nTicks' ticks in the future from the current
-     * clusterTime.
-     */
-    LogicalTime reserveTicks(uint64_t nTicks);
-
-    /**
-     * Resets current time to newTime. Should only be used for initializing this clock from an
-     * oplog timestamp.
-     */
-    void setClusterTimeFromTrustedSource(LogicalTime newTime);
-
-    /**
-     * Returns true if the clock is enabled and can be used. Defaults to true.
-     */
-    bool isEnabled() const;
-
-    /**
-     * Disables the logical clock. A disabled clock won't process logical times and can't be
-     * re-enabled.
-     */
-    void disable();
-
 private:
-    /**
-     * Rate limiter for advancing cluster time. Rejects newTime if its seconds value is more than
-     * kMaxAcceptableLogicalClockDriftSecs seconds ahead of this node's wall clock.
-     */
-    Status _passesRateLimiter_inlock(LogicalTime newTime);
-
     ServiceContext* const _service;
-
-    // The mutex protects _clusterTime and _isEnabled.
-    mutable Mutex _mutex = MONGO_MAKE_LATCH("LogicalClock::_mutex");
-    LogicalTime _clusterTime;
-    bool _isEnabled{true};
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/logical_clock.idl b/src/mongo/db/logical_clock.idl
index 18eb2e2b2ca..84a33d1cb3e 100644
--- a/src/mongo/db/logical_clock.idl
+++ b/src/mongo/db/logical_clock.idl
@@ -29,13 +29,14 @@
 global:
     cpp_namespace: "mongo"
 
+# TODO SERVER-48435: Rename this file to vector_clock.idl (but keep the parameter name the same).
 server_parameters:
     maxAcceptableLogicalClockDriftSecs:
-        description: "The value defines the maximum difference, in seconds, between the current clusterTime and a new clusterTime. Default value is 1 year."
+        description: "The value defines the maximum time, in seconds, that a new logical time (such as $clusterTime) may be ahead of the node's current wallclock time. Default value is 1 year."
         set_at: [ startup ]
         cpp_vartype: long long
         cpp_varname: gMaxAcceptableLogicalClockDriftSecs
         default: 
             expr: (365 * 24 * 60 * 60) # one year
         validator: 
-            gt: 0
\ No newline at end of file
+            gt: 0
diff --git a/src/mongo/db/logical_clock_test.cpp b/src/mongo/db/logical_clock_test.cpp
index fdb49a7390b..248d6609cc5 100644
--- a/src/mongo/db/logical_clock_test.cpp
+++ b/src/mongo/db/logical_clock_test.cpp
@@ -41,6 +41,7 @@
 #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/vector_clock_mutable.h"
 #include "mongo/unittest/unittest.h"
 #include "mongo/util/clock_source_mock.h"
 
@@ -51,6 +52,9 @@ const NamespaceString kDummyNamespaceString("test", "foo");
 
 using LogicalClockTest = LogicalClockTestFixture;
 
+constexpr auto ClusterTime = VectorClock::Component::ClusterTime;
+constexpr unsigned maxVal = std::numeric_limits<int32_t>::max();
+
 LogicalTime buildLogicalTime(unsigned secs, unsigned inc) {
     return LogicalTime(Timestamp(secs, inc));
 }
@@ -60,7 +64,7 @@ TEST_F(LogicalClockTest, roundtrip) {
     Timestamp tX(1);
     auto time = LogicalTime(tX);
 
-    getClock()->setClusterTimeFromTrustedSource(time);
+    VectorClockMutable::get(getServiceContext())->tickTo(ClusterTime, time);
     auto storedTime(getClock()->getClusterTime());
 
     ASSERT_TRUE(storedTime == time);
@@ -71,37 +75,37 @@ TEST_F(LogicalClockTest, reserveTicks) {
     // Set clock to a non-zero time, so we can verify wall clock synchronization.
     setMockClockSourceTime(Date_t::fromMillisSinceEpoch(10 * 1000));
 
-    auto t1 = getClock()->reserveTicks(1);
+    auto t1 = VectorClockMutable::get(getServiceContext())->tick(ClusterTime, 1);
     auto t2(getClock()->getClusterTime());
     ASSERT_TRUE(t1 == t2);
 
     // Make sure we synchronized with the wall clock.
     ASSERT_TRUE(t2.asTimestamp().getSecs() == 10);
 
-    auto t3 = getClock()->reserveTicks(1);
+    auto t3 = VectorClockMutable::get(getServiceContext())->tick(ClusterTime, 1);
     t1.addTicks(1);
     ASSERT_TRUE(t3 == t1);
 
-    t3 = getClock()->reserveTicks(100);
+    t3 = VectorClockMutable::get(getServiceContext())->tick(ClusterTime, 100);
     t1.addTicks(1);
     ASSERT_TRUE(t3 == t1);
 
-    t3 = getClock()->reserveTicks(1);
+    t3 = VectorClockMutable::get(getServiceContext())->tick(ClusterTime, 1);
     t1.addTicks(100);
     ASSERT_TRUE(t3 == t1);
 
     // Ensure overflow to a new second.
     auto initTimeSecs = getClock()->getClusterTime().asTimestamp().getSecs();
-    getClock()->reserveTicks((1U << 31) - 1);
+    VectorClockMutable::get(getServiceContext())->tick(ClusterTime, (1U << 31) - 1);
     auto newTimeSecs = getClock()->getClusterTime().asTimestamp().getSecs();
     ASSERT_TRUE(newTimeSecs == initTimeSecs + 1);
 }
 
 // Verify the advanceClusterTime functionality.
 TEST_F(LogicalClockTest, advanceClusterTime) {
-    auto t1 = getClock()->reserveTicks(1);
+    auto t1 = VectorClockMutable::get(getServiceContext())->tick(ClusterTime, 1);
     t1.addTicks(100);
-    ASSERT_OK(getClock()->advanceClusterTime(t1));
+    advanceClusterTime(t1);
     ASSERT_TRUE(t1 == getClock()->getClusterTime());
 }
 
@@ -116,7 +120,8 @@ TEST_F(LogicalClockTest, RateLimiterRejectsLogicalTimesTooFarAhead) {
         1);
     LogicalTime t1(tooFarAheadTimestamp);
 
-    ASSERT_EQ(ErrorCodes::ClusterTimeFailsRateLimiter, getClock()->advanceClusterTime(t1));
+    ASSERT_THROWS_CODE(
+        advanceClusterTime(t1), DBException, ErrorCodes::ClusterTimeFailsRateLimiter);
 }
 
 // Verify cluster time can be initialized to a very old time.
@@ -128,7 +133,7 @@ TEST_F(LogicalClockTest, InitFromTrustedSourceCanAcceptVeryOldLogicalTime) {
         durationCount<Seconds>(getMockClockSourceTime().toDurationSinceEpoch()) -
         (kMaxAcceptableLogicalClockDriftSecsDefault * 5));
     auto veryOldTime = LogicalTime(veryOldTimestamp);
-    getClock()->setClusterTimeFromTrustedSource(veryOldTime);
+    VectorClockMutable::get(getServiceContext())->tickTo(ClusterTime, veryOldTime);
 
     ASSERT_TRUE(getClock()->getClusterTime() == veryOldTime);
 }
@@ -138,7 +143,7 @@ TEST_F(LogicalClockTest, WritesToOplogAdvanceClusterTime) {
     Timestamp tX(1, 0);
     auto initialTime = LogicalTime(tX);
 
-    getClock()->setClusterTimeFromTrustedSource(initialTime);
+    VectorClockMutable::get(getServiceContext())->tickTo(ClusterTime, initialTime);
     ASSERT_TRUE(getClock()->getClusterTime() == initialTime);
 
     getDBClient()->insert(kDummyNamespaceString.ns(), BSON("x" << 1));
@@ -176,7 +181,7 @@ TEST_F(LogicalClockTest, WallClockSetTooFarInPast) {
 
     // Verify that maxAcceptableLogicalClockDriftSecs parameter does not need to be increased to
     // advance cluster time through metadata back to the current time.
-    ASSERT_OK(getClock()->advanceClusterTime(currentTime));
+    advanceClusterTime(currentTime);
     ASSERT_TRUE(getClock()->getClusterTime() == currentTime);
 }
 
@@ -210,117 +215,113 @@ TEST_F(LogicalClockTest, WallClockSetTooFarInFuture) {
     auto nextTime = getClock()->getClusterTime();
     nextTime.addTicks(1);  // The next lowest cluster time.
 
-    ASSERT_EQ(ErrorCodes::ClusterTimeFailsRateLimiter, getClock()->advanceClusterTime(nextTime));
+    ASSERT_THROWS_CODE(
+        advanceClusterTime(nextTime), DBException, ErrorCodes::ClusterTimeFailsRateLimiter);
 
     // Set wall clock to the current time + 1 day to simulate increasing the
     // maxAcceptableLogicalClockDriftSecs parameter, which can only be set at startup, and verify
     // time can be advanced through metadata again.
     setMockClockSourceTime(Date_t::fromDurationSinceEpoch(currentSecs + oneDay));
 
-    ASSERT_OK(getClock()->advanceClusterTime(nextTime));
+    advanceClusterTime(nextTime);
     ASSERT_TRUE(getClock()->getClusterTime() == nextTime);
 }
 
 // Verify the behavior of advancing cluster time around the max allowed values.
 TEST_F(LogicalClockTest, ReserveTicksBehaviorAroundMaxTime) {
-    unsigned maxVal = LogicalClock::kMaxSignedInt;
-
     // Verify clock can be advanced near the max values.
 
     // Can always advance to the max value for the inc field.
-    resetClock()->setClusterTimeFromTrustedSource(buildLogicalTime(maxVal - 1, maxVal - 1));
-    getClock()->reserveTicks(1);
+    resetClock()->tickTo(ClusterTime, buildLogicalTime(maxVal - 1, maxVal - 1));
+    VectorClockMutable::get(getServiceContext())->tick(ClusterTime, 1);
     ASSERT_EQ(getClock()->getClusterTime(), buildLogicalTime(maxVal - 1, maxVal));
 
-    resetClock()->setClusterTimeFromTrustedSource(buildLogicalTime(maxVal - 1, maxVal - 5));
-    getClock()->reserveTicks(5);
+    resetClock()->tickTo(ClusterTime, buildLogicalTime(maxVal - 1, maxVal - 5));
+    VectorClockMutable::get(getServiceContext())->tick(ClusterTime, 5);
     ASSERT_EQ(getClock()->getClusterTime(), buildLogicalTime(maxVal - 1, maxVal));
 
-    resetClock()->setClusterTimeFromTrustedSource(buildLogicalTime(0, maxVal - 1));
-    getClock()->reserveTicks(1);
+    resetClock()->tickTo(ClusterTime, buildLogicalTime(0, maxVal - 1));
+    VectorClockMutable::get(getServiceContext())->tick(ClusterTime, 1);
     ASSERT_EQ(getClock()->getClusterTime(), buildLogicalTime(0, maxVal));
 
     // Can overflow inc into seconds to reach max seconds value.
-    resetClock()->setClusterTimeFromTrustedSource(buildLogicalTime(maxVal - 1, maxVal));
-    getClock()->reserveTicks(1);
+    resetClock()->tickTo(ClusterTime, buildLogicalTime(maxVal - 1, maxVal));
+    VectorClockMutable::get(getServiceContext())->tick(ClusterTime, 1);
     ASSERT_EQ(getClock()->getClusterTime(), buildLogicalTime(maxVal, 1));
 
-    resetClock()->setClusterTimeFromTrustedSource(buildLogicalTime(maxVal - 1, maxVal - 5));
-    getClock()->reserveTicks(10);
+    resetClock()->tickTo(ClusterTime, buildLogicalTime(maxVal - 1, maxVal - 5));
+    VectorClockMutable::get(getServiceContext())->tick(ClusterTime, 10);
     ASSERT_EQ(getClock()->getClusterTime(), buildLogicalTime(maxVal, 10));
 
-    resetClock()->setClusterTimeFromTrustedSource(buildLogicalTime(maxVal - 1, 1));
-    getClock()->reserveTicks(maxVal);
+    resetClock()->tickTo(ClusterTime, buildLogicalTime(maxVal - 1, 1));
+    VectorClockMutable::get(getServiceContext())->tick(ClusterTime, maxVal);
     ASSERT_EQ(getClock()->getClusterTime(), buildLogicalTime(maxVal, maxVal));
 
     // Can advance inc field when seconds field is at the max value.
-    resetClock()->setClusterTimeFromTrustedSource(buildLogicalTime(maxVal, 1));
-    getClock()->reserveTicks(1);
+    resetClock()->tickTo(ClusterTime, buildLogicalTime(maxVal, 1));
+    VectorClockMutable::get(getServiceContext())->tick(ClusterTime, 1);
     ASSERT_EQ(getClock()->getClusterTime(), buildLogicalTime(maxVal, 2));
 
-    resetClock()->setClusterTimeFromTrustedSource(buildLogicalTime(maxVal, 1));
-    getClock()->reserveTicks(100);
+    resetClock()->tickTo(ClusterTime, buildLogicalTime(maxVal, 1));
+    VectorClockMutable::get(getServiceContext())->tick(ClusterTime, 100);
     ASSERT_EQ(getClock()->getClusterTime(), buildLogicalTime(maxVal, 101));
 
     // Can advance to the max value for both the inc and seconds fields.
-    resetClock()->setClusterTimeFromTrustedSource(buildLogicalTime(maxVal, maxVal - 1));
-    getClock()->reserveTicks(1);
+    resetClock()->tickTo(ClusterTime, buildLogicalTime(maxVal, maxVal - 1));
+    VectorClockMutable::get(getServiceContext())->tick(ClusterTime, 1);
     ASSERT_EQ(getClock()->getClusterTime(), buildLogicalTime(maxVal, maxVal));
 
-    resetClock()->setClusterTimeFromTrustedSource(buildLogicalTime(maxVal, maxVal - 5));
-    getClock()->reserveTicks(5);
+    resetClock()->tickTo(ClusterTime, buildLogicalTime(maxVal, maxVal - 5));
+    VectorClockMutable::get(getServiceContext())->tick(ClusterTime, 5);
     ASSERT_EQ(getClock()->getClusterTime(), buildLogicalTime(maxVal, maxVal));
 
     // Verify scenarios where the clock cannot be advanced.
 
     // Can't overflow inc into seconds when seconds field is at the max value.
-    resetClock()->setClusterTimeFromTrustedSource(buildLogicalTime(maxVal, maxVal));
-    ASSERT_THROWS(getClock()->reserveTicks(1), std::exception);
+    resetClock()->tickTo(ClusterTime, buildLogicalTime(maxVal, maxVal));
+    ASSERT_THROWS(VectorClockMutable::get(getServiceContext())->tick(ClusterTime, 1), DBException);
     ASSERT_EQ(getClock()->getClusterTime(), buildLogicalTime(maxVal, maxVal));
 
-    resetClock()->setClusterTimeFromTrustedSource(buildLogicalTime(maxVal, maxVal));
-    ASSERT_THROWS(getClock()->reserveTicks(5), std::exception);
+    resetClock()->tickTo(ClusterTime, buildLogicalTime(maxVal, maxVal));
+    ASSERT_THROWS(VectorClockMutable::get(getServiceContext())->tick(ClusterTime, 5), DBException);
     ASSERT_EQ(getClock()->getClusterTime(), buildLogicalTime(maxVal, maxVal));
 
-    resetClock()->setClusterTimeFromTrustedSource(buildLogicalTime(maxVal, maxVal - 1));
-    ASSERT_THROWS(getClock()->reserveTicks(2), std::exception);
+    resetClock()->tickTo(ClusterTime, buildLogicalTime(maxVal, maxVal - 1));
+    ASSERT_THROWS(VectorClockMutable::get(getServiceContext())->tick(ClusterTime, 2), DBException);
     ASSERT_EQ(getClock()->getClusterTime(), buildLogicalTime(maxVal, maxVal - 1));
 
-    resetClock()->setClusterTimeFromTrustedSource(buildLogicalTime(maxVal, maxVal - 11));
-    ASSERT_THROWS(getClock()->reserveTicks(12), std::exception);
+    resetClock()->tickTo(ClusterTime, buildLogicalTime(maxVal, maxVal - 11));
+    ASSERT_THROWS(VectorClockMutable::get(getServiceContext())->tick(ClusterTime, 12), DBException);
     ASSERT_EQ(getClock()->getClusterTime(), buildLogicalTime(maxVal, maxVal - 11));
 }
 
 // Verify behavior of advancing cluster time when the wall clock is near the max allowed value.
 TEST_F(LogicalClockTest, ReserveTicksBehaviorWhenWallClockNearMaxTime) {
-    unsigned maxVal = LogicalClock::kMaxSignedInt;
-
     // Can be set to the max possible time by catching up to the wall clock.
     setMockClockSourceTime(Date_t::fromDurationSinceEpoch(Seconds(maxVal)));
 
-    resetClock()->setClusterTimeFromTrustedSource(buildLogicalTime(1, 1));
-    getClock()->reserveTicks(1);
+    resetClock()->tickTo(ClusterTime, buildLogicalTime(1, 1));
+    VectorClockMutable::get(getServiceContext())->tick(ClusterTime, 1);
     ASSERT_EQ(getClock()->getClusterTime(), buildLogicalTime(maxVal, 1));
 
     // Should fail when wall clock would advance cluster time beyond the max allowed time.
     setMockClockSourceTime(Date_t::max());
 
-    resetClock()->setClusterTimeFromTrustedSource(buildLogicalTime(1, 1));
-    ASSERT_THROWS(getClock()->reserveTicks(1), std::exception);
+    resetClock()->tickTo(ClusterTime, buildLogicalTime(1, 1));
+    ASSERT_THROWS(VectorClockMutable::get(getServiceContext())->tick(ClusterTime, 1), DBException);
     ASSERT_EQ(getClock()->getClusterTime(), buildLogicalTime(1, 1));
 }
 
 // Verify the clock rejects cluster times greater than the max allowed time.
 TEST_F(LogicalClockTest, RejectsLogicalTimesGreaterThanMaxTime) {
-    unsigned maxVal = LogicalClock::kMaxSignedInt;
-
     // A cluster time can be greater than the maximum value allowed because the signed integer
     // maximum is used for legacy compatibility, but these fields are stored as unsigned integers.
     auto beyondMaxTime = buildLogicalTime(maxVal + 1, maxVal + 1);
 
     // The clock can't be initialized to a time greater than the max possible.
     resetClock();
-    ASSERT_THROWS(getClock()->setClusterTimeFromTrustedSource(beyondMaxTime), std::exception);
+    ASSERT_THROWS(VectorClockMutable::get(getServiceContext())->tickTo(ClusterTime, beyondMaxTime),
+                  DBException);
     ASSERT_TRUE(getClock()->getClusterTime() == LogicalTime());
 
     // The time can't be advanced through metadata to a time greater than the max possible.
@@ -328,7 +329,7 @@ TEST_F(LogicalClockTest, RejectsLogicalTimesGreaterThanMaxTime) {
     auto almostMaxSecs =
         Seconds(maxVal) - Seconds(kMaxAcceptableLogicalClockDriftSecsDefault) + Seconds(10);
     setMockClockSourceTime(Date_t::fromDurationSinceEpoch(almostMaxSecs));
-    ASSERT_THROWS(getClock()->advanceClusterTime(beyondMaxTime), std::exception);
+    ASSERT_THROWS(advanceClusterTime(beyondMaxTime), DBException);
     ASSERT_TRUE(getClock()->getClusterTime() == LogicalTime());
 }
 
diff --git a/src/mongo/db/logical_clock_test_fixture.cpp b/src/mongo/db/logical_clock_test_fixture.cpp
index 342f2679522..99d3ee0e7b9 100644
--- a/src/mongo/db/logical_clock_test_fixture.cpp
+++ b/src/mongo/db/logical_clock_test_fixture.cpp
@@ -40,6 +40,7 @@
 #include "mongo/db/service_context.h"
 #include "mongo/db/signed_logical_time.h"
 #include "mongo/db/time_proof_service.h"
+#include "mongo/db/vector_clock_mutable.h"
 #include "mongo/unittest/unittest.h"
 #include "mongo/util/clock_source_mock.h"
 
@@ -71,14 +72,14 @@ void LogicalClockTestFixture::tearDown() {
     ShardingMongodTestFixture::tearDown();
 }
 
-LogicalClock* LogicalClockTestFixture::resetClock() {
+VectorClockMutable* LogicalClockTestFixture::resetClock() {
     auto service = getServiceContext();
-    auto logicalClock = std::make_unique<LogicalClock>(service);
-
-    LogicalClock::set(service, std::move(logicalClock));
-    _clock = LogicalClock::get(service);
+    VectorClock::get(service)->resetVectorClock_forTest();
+    return VectorClockMutable::get(service);
+}
 
-    return _clock;
+void LogicalClockTestFixture::advanceClusterTime(LogicalTime newTime) {
+    VectorClock::get(getServiceContext())->advanceClusterTime_forTest(newTime);
 }
 
 LogicalClock* LogicalClockTestFixture::getClock() const {
diff --git a/src/mongo/db/logical_clock_test_fixture.h b/src/mongo/db/logical_clock_test_fixture.h
index 593670d5aaf..4b3322d406b 100644
--- a/src/mongo/db/logical_clock_test_fixture.h
+++ b/src/mongo/db/logical_clock_test_fixture.h
@@ -38,6 +38,7 @@ class ClockSourceMock;
 class DBDirectClient;
 class LogicalClock;
 class LogicalTime;
+class VectorClockMutable;
 
 /**
  * A test fixture that installs a LogicalClock instance with a TimeProofService onto a service
@@ -58,7 +59,9 @@ protected:
 
     void tearDown() override;
 
-    LogicalClock* resetClock();
+    VectorClockMutable* resetClock();
+
+    void advanceClusterTime(LogicalTime newTime);
 
     LogicalClock* getClock() const;
 
diff --git a/src/mongo/db/logical_time_metadata_hook.cpp b/src/mongo/db/logical_time_metadata_hook.cpp
index 36fda90c1ce..11a57a59885 100644
--- a/src/mongo/db/logical_time_metadata_hook.cpp
+++ b/src/mongo/db/logical_time_metadata_hook.cpp
@@ -33,10 +33,8 @@
 
 #include <memory>
 
-#include "mongo/db/logical_clock.h"
-#include "mongo/db/logical_time_validator.h"
 #include "mongo/db/operation_time_tracker.h"
-#include "mongo/rpc/metadata/logical_time_metadata.h"
+#include "mongo/db/vector_clock.h"
 
 namespace mongo {
 
@@ -45,48 +43,35 @@ namespace rpc {
 namespace {
 const char kOperationTimeFieldName[] = "operationTime";
 }
+
+// TODO SERVER-48434: Rename this class to VectorClockMetadataHook.
 LogicalTimeMetadataHook::LogicalTimeMetadataHook(ServiceContext* service) : _service(service) {}
 
 Status LogicalTimeMetadataHook::writeRequestMetadata(OperationContext* opCtx,
                                                      BSONObjBuilder* metadataBob) {
-    auto validator = LogicalTimeValidator::get(_service);
-    if (!validator || !LogicalClock::get(_service)->isEnabled()) {
-        return Status::OK();
-    }
-
-    auto newTime = LogicalClock::get(_service)->getClusterTime();
-    LogicalTimeMetadata metadata(validator->trySignLogicalTime(newTime));
-    metadata.writeToMetadata(metadataBob);
+    VectorClock::get(_service)->gossipOut(opCtx, metadataBob, transport::Session::kInternalClient);
     return Status::OK();
 }
 
 Status LogicalTimeMetadataHook::readReplyMetadata(OperationContext* opCtx,
                                                   StringData replySource,
                                                   const BSONObj& metadataObj) {
-    auto parseStatus = LogicalTimeMetadata::readFromMetadata(metadataObj);
-    if (!parseStatus.isOK()) {
-        return parseStatus.getStatus();
-    }
-
-    auto& signedTime = parseStatus.getValue().getSignedTime();
-
-    // LogicalTimeMetadata is default constructed if no cluster time metadata was sent, so a
-    // default constructed SignedLogicalTime should be ignored.
-    if (signedTime.getTime() == LogicalTime::kUninitialized ||
-        !LogicalClock::get(_service)->isEnabled()) {
+    if (!VectorClock::get(_service)->isEnabled()) {
         return Status::OK();
     }
 
     if (opCtx) {
         auto timeTracker = OperationTimeTracker::get(opCtx);
-
         auto operationTime = metadataObj[kOperationTimeFieldName];
         if (!operationTime.eoo()) {
             invariant(operationTime.type() == BSONType::bsonTimestamp);
             timeTracker->updateOperationTime(LogicalTime(operationTime.timestamp()));
         }
     }
-    return LogicalClock::get(_service)->advanceClusterTime(signedTime.getTime());
+
+    VectorClock::get(_service)->gossipIn(
+        opCtx, metadataObj, false /* couldBeUnauthorized */, transport::Session::kInternalClient);
+    return Status::OK();
 }
 
 }  // namespace rpc
diff --git a/src/mongo/db/logical_time_validator.cpp b/src/mongo/db/logical_time_validator.cpp
index 793772dffb8..3f0e4d9b33b 100644
--- a/src/mongo/db/logical_time_validator.cpp
+++ b/src/mongo/db/logical_time_validator.cpp
@@ -39,9 +39,9 @@
 #include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/auth/privilege.h"
 #include "mongo/db/keys_collection_manager.h"
-#include "mongo/db/logical_clock.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/db/service_context.h"
+#include "mongo/db/vector_clock.h"
 #include "mongo/logv2/log.h"
 #include "mongo/util/assert_util.h"
 #include "mongo/util/fail_point.h"
@@ -130,7 +130,7 @@ SignedLogicalTime LogicalTimeValidator::signLogicalTime(OperationContext* opCtx,
     auto keyStatusWith = keyManager->getKeyForSigning(nullptr, newTime);
     auto keyStatus = keyStatusWith.getStatus();
 
-    while (keyStatus == ErrorCodes::KeyNotFound && LogicalClock::get(opCtx)->isEnabled()) {
+    while (keyStatus == ErrorCodes::KeyNotFound && VectorClock::get(opCtx)->isEnabled()) {
         keyManager->refreshNow(opCtx);
 
         keyStatusWith = keyManager->getKeyForSigning(nullptr, newTime);
diff --git a/src/mongo/db/logical_time_validator_test.cpp b/src/mongo/db/logical_time_validator_test.cpp
index ce8fea3393e..a5ceeb0c135 100644
--- a/src/mongo/db/logical_time_validator_test.cpp
+++ b/src/mongo/db/logical_time_validator_test.cpp
@@ -34,12 +34,12 @@
 #include "mongo/bson/timestamp.h"
 #include "mongo/db/keys_collection_client_sharded.h"
 #include "mongo/db/keys_collection_manager.h"
-#include "mongo/db/logical_clock.h"
 #include "mongo/db/logical_time.h"
 #include "mongo/db/logical_time_validator.h"
 #include "mongo/db/server_options.h"
 #include "mongo/db/signed_logical_time.h"
 #include "mongo/db/time_proof_service.h"
+#include "mongo/db/vector_clock_mutable.h"
 #include "mongo/s/catalog/dist_lock_manager_mock.h"
 #include "mongo/s/config_server_test_fixture.h"
 #include "mongo/unittest/unittest.h"
@@ -64,7 +64,8 @@ protected:
             Grid::get(operationContext())->catalogClient());
 
         const LogicalTime currentTime(LogicalTime(Timestamp(1, 0)));
-        LogicalClock::get(operationContext())->setClusterTimeFromTrustedSource(currentTime);
+        VectorClockMutable::get(operationContext())
+            ->tickTo(VectorClock::Component::ClusterTime, currentTime);
 
         _keyManager = std::make_shared<KeysCollectionManager>(
             "dummy", std::move(catalogClient), Seconds(1000));
diff --git a/src/mongo/db/op_msg_fuzzer.cpp b/src/mongo/db/op_msg_fuzzer.cpp
index b1c2843e360..94e0cbd7efe 100644
--- a/src/mongo/db/op_msg_fuzzer.cpp
+++ b/src/mongo/db/op_msg_fuzzer.cpp
@@ -41,6 +41,7 @@
 #include "mongo/db/service_context.h"
 #include "mongo/db/service_entry_point_common.h"
 #include "mongo/db/service_entry_point_mongod.h"
+#include "mongo/db/vector_clock_mutable.h"
 #include "mongo/platform/basic.h"
 #include "mongo/transport/service_entry_point_impl.h"
 #include "mongo/transport/session.h"
@@ -98,8 +99,9 @@ extern "C" int LLVMFuzzerTestOneInput(const char* Data, size_t Size) {
     mongo::ServiceContext::UniqueOperationContext opCtx =
         serviceContext->makeOperationContext(client.get());
     auto logicalClock = std::make_unique<mongo::LogicalClock>(serviceContext);
-    logicalClock->setClusterTimeFromTrustedSource(kInMemoryLogicalTime);
     mongo::LogicalClock::set(serviceContext, std::move(logicalClock));
+    VectorClockMutable::get(getServiceContext())
+        ->tickTo(VectorClock::Component::ClusterTime, kInMemoryLogicalTime);
 
     int new_size = Size + sizeof(int);
     auto sb = mongo::SharedBuffer::allocate(new_size);
diff --git a/src/mongo/db/ops/insert.cpp b/src/mongo/db/ops/insert.cpp
index 22dd459e999..69d19f49958 100644
--- a/src/mongo/db/ops/insert.cpp
+++ b/src/mongo/db/ops/insert.cpp
@@ -34,8 +34,7 @@
 
 #include "mongo/bson/bson_depth.h"
 #include "mongo/db/commands/feature_compatibility_version_parser.h"
-#include "mongo/db/logical_clock.h"
-#include "mongo/db/logical_time.h"
+#include "mongo/db/vector_clock_mutable.h"
 #include "mongo/db/views/durable_view_catalog.h"
 #include "mongo/util/str.h"
 
@@ -163,7 +162,8 @@ StatusWith<BSONObj> fixDocumentForInsert(ServiceContext* service, const BSONObj&
         if (hadId && e.fieldNameStringData() == "_id") {
             // no-op
         } else if (e.type() == bsonTimestamp && e.timestampValue() == 0) {
-            auto nextTime = LogicalClock::get(service)->reserveTicks(1);
+            auto nextTime =
+                VectorClockMutable::get(service)->tick(VectorClock::Component::ClusterTime, 1);
             b.append(e.fieldName(), nextTime.asTimestamp());
         } else {
             b.append(e);
diff --git a/src/mongo/db/pipeline/expression_context_test.cpp b/src/mongo/db/pipeline/expression_context_test.cpp
index 1135f465bb4..a3dad86199d 100644
--- a/src/mongo/db/pipeline/expression_context_test.cpp
+++ b/src/mongo/db/pipeline/expression_context_test.cpp
@@ -37,6 +37,7 @@
 #include "mongo/db/pipeline/process_interface/stub_mongo_process_interface.h"
 #include "mongo/db/query/datetime/date_time_support.h"
 #include "mongo/db/service_context_test_fixture.h"
+#include "mongo/db/vector_clock_mutable.h"
 #include "mongo/unittest/unittest.h"
 
 #define ASSERT_DOES_NOT_THROW(EXPRESSION)                                          \
@@ -56,10 +57,12 @@ using ExpressionContextTest = ServiceContextTest;
 TEST_F(ExpressionContextTest, ExpressionContextSummonsMissingTimeValues) {
     auto opCtx = makeOperationContext();
     auto logicalClock = std::make_unique<LogicalClock>(opCtx->getServiceContext());
-    auto t1 = logicalClock->reserveTicks(1);
-    t1.addTicks(100);
-    ASSERT_OK(logicalClock->advanceClusterTime(t1));
     LogicalClock::set(opCtx->getServiceContext(), std::move(logicalClock));
+    auto t1 = VectorClockMutable::get(opCtx->getServiceContext())
+                  ->tick(VectorClock::Component::ClusterTime, 1);
+    t1.addTicks(100);
+    VectorClockMutable::get(opCtx->getServiceContext())
+        ->tickTo(VectorClock::Component::ClusterTime, t1);
     {
         const auto expCtx = ExpressionContext{opCtx.get(),
                                               {},     // explain
diff --git a/src/mongo/db/pipeline/process_interface/SConscript b/src/mongo/db/pipeline/process_interface/SConscript
index b24e45be165..8fc9a8e1fb3 100644
--- a/src/mongo/db/pipeline/process_interface/SConscript
+++ b/src/mongo/db/pipeline/process_interface/SConscript
@@ -109,6 +109,7 @@ env.CppUnitTest(
     LIBDEPS=[
         '$BUILD_DIR/mongo/db/query/query_test_service_context',
         '$BUILD_DIR/mongo/db/service_context_test_fixture',
+        '$BUILD_DIR/mongo/db/vector_clock_mongod',
         '$BUILD_DIR/mongo/s/catalog_cache_test_fixture',
         'mongos_process_interface',
         'shardsvr_process_interface',
diff --git a/src/mongo/db/read_write_concern_defaults_test.cpp b/src/mongo/db/read_write_concern_defaults_test.cpp
index cf9a9a2d9da..6e248abea38 100644
--- a/src/mongo/db/read_write_concern_defaults_test.cpp
+++ b/src/mongo/db/read_write_concern_defaults_test.cpp
@@ -35,6 +35,7 @@
 #include "mongo/db/read_write_concern_defaults_cache_lookup_mock.h"
 #include "mongo/db/repl/optime.h"
 #include "mongo/db/service_context_test_fixture.h"
+#include "mongo/db/vector_clock_mutable.h"
 #include "mongo/util/clock_source_mock.h"
 
 namespace mongo {
@@ -272,7 +273,7 @@ protected:
         _lookupMock.setLookupCallReturnValue(std::move(defaults));
         auto oldDefaults = _rwcd.getDefault(operationContext());
 
-        getClock()->reserveTicks(1);
+        VectorClockMutable::get(getServiceContext())->tick(VectorClock::Component::ClusterTime, 1);
         getMockClockSource()->advance(Milliseconds(1));
 
         return oldDefaults;
@@ -471,7 +472,7 @@ TEST_F(ReadWriteConcernDefaultsTestWithClusterTime, TestRefreshDefaultsWithDelet
     ASSERT_EQ(Timestamp(10, 20), *origCachedDefaults.getUpdateOpTime());
     ASSERT_EQ(Date_t::fromMillisSinceEpoch(1234), *origCachedDefaults.getUpdateWallClockTime());
 
-    getClock()->reserveTicks(1);
+    VectorClockMutable::get(getServiceContext())->tick(VectorClock::Component::ClusterTime, 1);
     getMockClockSource()->advance(Milliseconds(1));
 
     _lookupMock.setLookupCallReturnValue(RWConcernDefault());
diff --git a/src/mongo/db/repl/SConscript b/src/mongo/db/repl/SConscript
index b4900aa5c8e..1f4b55bbdf7 100644
--- a/src/mongo/db/repl/SConscript
+++ b/src/mongo/db/repl/SConscript
@@ -23,10 +23,9 @@ env.Library(
     LIBDEPS_PRIVATE=[
         'optime',
         'repl_coordinator_interface',
-        '$BUILD_DIR/mongo/db/logical_clock',
         '$BUILD_DIR/mongo/db/logical_time',
-        '$BUILD_DIR/mongo/db/vector_clock_mutable',
         '$BUILD_DIR/mongo/db/storage/flow_control',
+        '$BUILD_DIR/mongo/db/vector_clock_mutable',
     ],
 )
 
@@ -228,7 +227,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/catalog/database_holder',
         '$BUILD_DIR/mongo/db/storage/oplog_cap_maintainer_thread',
         '$BUILD_DIR/mongo/db/storage/storage_control',
-        '$BUILD_DIR/mongo/db/logical_clock',
+        '$BUILD_DIR/mongo/db/vector_clock',
     ],
 )
 
@@ -772,6 +771,7 @@ env.Library(
         'storage_interface',
         '$BUILD_DIR/mongo/db/concurrency/lock_manager',
         '$BUILD_DIR/mongo/db/service_context',
+        '$BUILD_DIR/mongo/db/replica_set_aware_service',
         '$BUILD_DIR/mongo/executor/network_interface_mock',
     ],
 )
@@ -1173,7 +1173,6 @@ env.Library(
         '$BUILD_DIR/mongo/db/free_mon/free_mon_mongod',
         '$BUILD_DIR/mongo/db/kill_sessions_local',
         '$BUILD_DIR/mongo/db/lasterror',
-        '$BUILD_DIR/mongo/db/logical_clock',
         '$BUILD_DIR/mongo/db/logical_time',
         '$BUILD_DIR/mongo/db/op_observer',
         '$BUILD_DIR/mongo/db/query_exec',
@@ -1184,6 +1183,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/service_context',
         '$BUILD_DIR/mongo/db/stats/counters',
         '$BUILD_DIR/mongo/db/system_index',
+        '$BUILD_DIR/mongo/db/vector_clock',
         '$BUILD_DIR/mongo/rpc/client_metadata',
         '$BUILD_DIR/mongo/util/fail_point',
         'bgsync',
diff --git a/src/mongo/db/repl/local_oplog_info.cpp b/src/mongo/db/repl/local_oplog_info.cpp
index f6573c6a5f8..a36dae6897f 100644
--- a/src/mongo/db/repl/local_oplog_info.cpp
+++ b/src/mongo/db/repl/local_oplog_info.cpp
@@ -33,12 +33,11 @@
 
 #include "mongo/db/repl/local_oplog_info.h"
 
-#include "mongo/db/logical_clock.h"
-#include "mongo/db/logical_time.h"
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/storage/flow_control.h"
 #include "mongo/db/storage/record_store.h"
 #include "mongo/db/storage/recovery_unit.h"
+#include "mongo/db/vector_clock_mutable.h"
 #include "mongo/util/assert_util.h"
 
 namespace mongo {
@@ -96,7 +95,8 @@ void LocalOplogInfo::resetCollection() {
 
 void LocalOplogInfo::setNewTimestamp(ServiceContext* service, const Timestamp& newTime) {
     stdx::lock_guard<Latch> lk(_newOpMutex);
-    LogicalClock::get(service)->setClusterTimeFromTrustedSource(LogicalTime(newTime));
+    VectorClockMutable::get(service)->tickTo(VectorClock::Component::ClusterTime,
+                                             LogicalTime(newTime));
 }
 
 std::vector<OplogSlot> LocalOplogInfo::getNextOpTimes(OperationContext* opCtx, std::size_t count) {
@@ -123,7 +123,9 @@ std::vector<OplogSlot> LocalOplogInfo::getNextOpTimes(OperationContext* opCtx, s
     {
         stdx::lock_guard<Latch> lk(_newOpMutex);
 
-        ts = LogicalClock::get(opCtx)->reserveTicks(count).asTimestamp();
+        ts = VectorClockMutable::get(opCtx)
+                 ->tick(VectorClock::Component::ClusterTime, count)
+                 .asTimestamp();
         const bool orderedCommit = false;
 
         // The local oplog collection pointer must already be established by this point.
diff --git a/src/mongo/db/repl/oplog_applier_impl_test_fixture.cpp b/src/mongo/db/repl/oplog_applier_impl_test_fixture.cpp
index fcc03c89ee8..45219111f34 100644
--- a/src/mongo/db/repl/oplog_applier_impl_test_fixture.cpp
+++ b/src/mongo/db/repl/oplog_applier_impl_test_fixture.cpp
@@ -35,7 +35,6 @@
 #include "mongo/db/curop.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/dbdirectclient.h"
-#include "mongo/db/logical_clock.h"
 #include "mongo/db/op_observer_registry.h"
 #include "mongo/db/query/internal_plans.h"
 #include "mongo/db/repl/drop_pending_collection_reaper.h"
@@ -46,6 +45,7 @@
 #include "mongo/db/repl/replication_recovery_mock.h"
 #include "mongo/db/repl/storage_interface.h"
 #include "mongo/db/repl/storage_interface_impl.h"
+#include "mongo/db/vector_clock_mutable.h"
 
 namespace mongo {
 namespace repl {
@@ -131,7 +131,8 @@ void OplogApplierImplTest::setUp() {
 
     // This is necessary to generate ghost timestamps for index builds that are not 0, since 0 is an
     // invalid timestamp.
-    ASSERT_OK(LogicalClock::get(_opCtx.get())->advanceClusterTime(LogicalTime(Timestamp(1, 0))));
+    VectorClockMutable::get(_opCtx.get())
+        ->tickTo(VectorClock::Component::ClusterTime, LogicalTime(Timestamp(1, 0)));
 }
 
 void OplogApplierImplTest::tearDown() {
diff --git a/src/mongo/db/repl/oplog_fetcher_test.cpp b/src/mongo/db/repl/oplog_fetcher_test.cpp
index bb22b7537fe..3987c4a6437 100644
--- a/src/mongo/db/repl/oplog_fetcher_test.cpp
+++ b/src/mongo/db/repl/oplog_fetcher_test.cpp
@@ -35,10 +35,10 @@
 #include "mongo/db/repl/repl_server_parameters_gen.h"
 #include "mongo/db/repl/task_executor_mock.h"
 #include "mongo/db/service_context_test_fixture.h"
+#include "mongo/db/signed_logical_time.h"
 #include "mongo/dbtests/mock/mock_dbclient_connection.h"
 #include "mongo/executor/thread_pool_task_executor_test_fixture.h"
 #include "mongo/rpc/metadata.h"
-#include "mongo/rpc/metadata/logical_time_metadata.h"
 #include "mongo/rpc/metadata/oplog_query_metadata.h"
 #include "mongo/rpc/metadata/repl_set_metadata.h"
 #include "mongo/unittest/death_test.h"
@@ -2208,13 +2208,22 @@ TEST_F(OplogFetcherTest, HandleLogicalTimeMetaDataAndAdvanceClusterTime) {
     auto oldClusterTime = LogicalClock::get(getGlobalServiceContext())->getClusterTime();
 
     auto logicalTime = LogicalTime(Timestamp(123456, 78));
-    auto logicalTimeMetadata =
-        rpc::LogicalTimeMetadata(SignedLogicalTime(logicalTime, TimeProofService::TimeProof(), 0));
+    auto signedTime = SignedLogicalTime(logicalTime, TimeProofService::TimeProof(), 0);
 
     BSONObjBuilder bob;
     ASSERT_OK(replSetMetadata.writeToMetadata(&bob));
     ASSERT_OK(oqMetadata.writeToMetadata(&bob));
-    logicalTimeMetadata.writeToMetadata(&bob);
+
+    BSONObjBuilder subObjBuilder(bob.subobjStart("$clusterTime"));
+    signedTime.getTime().asTimestamp().append(subObjBuilder.bb(), "clusterTime");
+
+    BSONObjBuilder signatureObjBuilder(subObjBuilder.subobjStart("signature"));
+    signedTime.getProof()->appendAsBinData(signatureObjBuilder, "hash");
+    signatureObjBuilder.append("keyId", signedTime.getKeyId());
+    signatureObjBuilder.doneFast();
+
+    subObjBuilder.doneFast();
+
     auto metadataObj = bob.obj();
 
     // Process one batch with the logical time metadata.
diff --git a/src/mongo/db/repl/replication_coordinator_external_state_mock.cpp b/src/mongo/db/repl/replication_coordinator_external_state_mock.cpp
index 9ff94e5af20..9f0053fb001 100644
--- a/src/mongo/db/repl/replication_coordinator_external_state_mock.cpp
+++ b/src/mongo/db/repl/replication_coordinator_external_state_mock.cpp
@@ -40,6 +40,7 @@
 #include "mongo/db/client.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/db/repl/oplog_buffer_blocking_queue.h"
+#include "mongo/db/replica_set_aware_service.h"
 #include "mongo/util/net/hostandport.h"
 #include "mongo/util/sequence_util.h"
 
@@ -235,7 +236,9 @@ void ReplicationCoordinatorExternalStateMock::closeConnections() {
 
 void ReplicationCoordinatorExternalStateMock::onStepDownHook() {}
 
-void ReplicationCoordinatorExternalStateMock::onBecomeArbiterHook() {}
+void ReplicationCoordinatorExternalStateMock::onBecomeArbiterHook() {
+    ReplicaSetAwareServiceRegistry::get(getGlobalServiceContext()).onBecomeArbiter();
+}
 
 void ReplicationCoordinatorExternalStateMock::signalApplierToChooseNewSyncSource() {}
 
diff --git a/src/mongo/db/repl/replication_coordinator_impl.cpp b/src/mongo/db/repl/replication_coordinator_impl.cpp
index 943266b43c9..531accbbeb6 100644
--- a/src/mongo/db/repl/replication_coordinator_impl.cpp
+++ b/src/mongo/db/repl/replication_coordinator_impl.cpp
@@ -62,9 +62,6 @@
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/index_builds_coordinator.h"
 #include "mongo/db/kill_sessions_local.h"
-#include "mongo/db/logical_clock.h"
-#include "mongo/db/logical_time.h"
-#include "mongo/db/logical_time_validator.h"
 #include "mongo/db/mongod_options_storage_gen.h"
 #include "mongo/db/prepare_conflict_tracker.h"
 #include "mongo/db/repl/check_quorum_for_config_change.h"
@@ -91,6 +88,8 @@
 #include "mongo/db/replica_set_aware_service.h"
 #include "mongo/db/server_options.h"
 #include "mongo/db/storage/storage_options.h"
+#include "mongo/db/vector_clock.h"
+#include "mongo/db/vector_clock_mutable.h"
 #include "mongo/db/write_concern.h"
 #include "mongo/db/write_concern_options.h"
 #include "mongo/executor/connection_pool_stats.h"
@@ -657,12 +656,6 @@ void ReplicationCoordinatorImpl::_finishLoadLocalConfig(
         }
     } else {
         _externalState->onBecomeArbiterHook();
-        // TODO SERVER-47914: move these actions into VectorClockMongoD::onBecomeArbiter().
-        // The node is an arbiter hence will not need logical clock for external operations.
-        LogicalClock::get(getServiceContext())->disable();
-        if (auto validator = LogicalTimeValidator::get(getServiceContext())) {
-            validator->stopKeyManager();
-        }
     }
 
     const auto lastOpTime = lastOpTimeAndWallTime.opTime;
@@ -1391,7 +1384,7 @@ void ReplicationCoordinatorImpl::_setMyLastAppliedOpTimeAndWallTime(
 
     // The last applied opTime should never advance beyond the global timestamp (i.e. the latest
     // cluster time). Not enforced if the logical clock is disabled, e.g. for arbiters.
-    dassert(!LogicalClock::get(getServiceContext())->isEnabled() ||
+    dassert(!VectorClock::get(getServiceContext())->isEnabled() ||
             _externalState->getGlobalTimestamp(getServiceContext()) >= opTime.getTimestamp());
 
     _topCoord->setMyLastAppliedOpTimeAndWallTime(
@@ -4193,7 +4186,9 @@ void ReplicationCoordinatorImpl::_performPostMemberStateUpdateAction(
 void ReplicationCoordinatorImpl::_postWonElectionUpdateMemberState(WithLock lk) {
     invariant(_topCoord->getTerm() != OpTime::kUninitializedTerm);
     _electionId = OID::fromTerm(_topCoord->getTerm());
-    auto ts = LogicalClock::get(getServiceContext())->reserveTicks(1).asTimestamp();
+    auto ts = VectorClockMutable::get(getServiceContext())
+                  ->tick(VectorClock::Component::ClusterTime, 1)
+                  .asTimestamp();
     _topCoord->processWinElection(_electionId, ts);
     const PostMemberStateUpdateAction nextAction = _updateMemberStateFromTopologyCoordinator(lk);
 
diff --git a/src/mongo/db/repl/replication_coordinator_impl_heartbeat.cpp b/src/mongo/db/repl/replication_coordinator_impl_heartbeat.cpp
index c8d1edc54a3..fc90826da9f 100644
--- a/src/mongo/db/repl/replication_coordinator_impl_heartbeat.cpp
+++ b/src/mongo/db/repl/replication_coordinator_impl_heartbeat.cpp
@@ -44,8 +44,6 @@
 #include "mongo/db/commands/test_commands_enabled.h"
 #include "mongo/db/index_builds_coordinator.h"
 #include "mongo/db/kill_sessions_local.h"
-#include "mongo/db/logical_clock.h"
-#include "mongo/db/logical_time_validator.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/db/repl/heartbeat_response_action.h"
 #include "mongo/db/repl/repl_server_parameters_gen.h"
@@ -690,11 +688,6 @@ void ReplicationCoordinatorImpl::_heartbeatReconfigStore(
 
         if (isArbiter) {
             _externalState->onBecomeArbiterHook();
-            // TODO SERVER-47914: move these actions into VectorClockMongoD::onBecomeArbiter().
-            LogicalClock::get(getGlobalServiceContext())->disable();
-            if (auto validator = LogicalTimeValidator::get(getGlobalServiceContext())) {
-                validator->stopKeyManager();
-            }
         }
 
         if (!isArbiter && isFirstConfig) {
diff --git a/src/mongo/db/s/SConscript b/src/mongo/db/s/SConscript
index 3cb6a57100b..fb555cb1171 100644
--- a/src/mongo/db/s/SConscript
+++ b/src/mongo/db/s/SConscript
@@ -99,7 +99,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/server_options_core',
         '$BUILD_DIR/mongo/db/storage/remove_saver',
         '$BUILD_DIR/mongo/db/transaction',
-        '$BUILD_DIR/mongo/db/vector_clock_d',
+        '$BUILD_DIR/mongo/db/vector_clock_mongod',
         '$BUILD_DIR/mongo/s/client/shard_local',
         '$BUILD_DIR/mongo/s/query/cluster_aggregate',
         '$BUILD_DIR/mongo/s/sharding_initialization',
@@ -165,6 +165,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/commands/txn_cmd_request',
         '$BUILD_DIR/mongo/db/dbdirectclient',
         '$BUILD_DIR/mongo/db/rw_concern_d',
+        '$BUILD_DIR/mongo/db/vector_clock_mutable',
         '$BUILD_DIR/mongo/executor/task_executor_pool',
         '$BUILD_DIR/mongo/s/grid',
         'sharding_api_d',
@@ -420,6 +421,7 @@ env.CppUnitTest(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/auth/authmocks',
+        '$BUILD_DIR/mongo/db/keys_collection_client_direct',
         '$BUILD_DIR/mongo/db/logical_session_cache_impl',
         '$BUILD_DIR/mongo/db/ops/write_ops_exec',
         '$BUILD_DIR/mongo/db/query/query_request',
diff --git a/src/mongo/db/s/transaction_coordinator.cpp b/src/mongo/db/s/transaction_coordinator.cpp
index 885960965be..0a9b900f853 100644
--- a/src/mongo/db/s/transaction_coordinator.cpp
+++ b/src/mongo/db/s/transaction_coordinator.cpp
@@ -33,10 +33,10 @@
 
 #include "mongo/db/s/transaction_coordinator.h"
 
-#include "mongo/db/logical_clock.h"
 #include "mongo/db/s/transaction_coordinator_metrics_observer.h"
 #include "mongo/db/s/wait_for_majority_service.h"
 #include "mongo/db/server_options.h"
+#include "mongo/db/vector_clock_mutable.h"
 #include "mongo/logv2/log.h"
 #include "mongo/s/grid.h"
 #include "mongo/util/fail_point.h"
@@ -209,9 +209,9 @@ TransactionCoordinator::TransactionCoordinator(OperationContext* operationContex
                                     "txnNumber"_attr = _txnNumber,
                                     "commitTimestamp"_attr = *_decision->getCommitTimestamp());
 
-                        uassertStatusOK(LogicalClock::get(_serviceContext)
-                                            ->advanceClusterTime(
-                                                LogicalTime(*_decision->getCommitTimestamp())));
+                        VectorClockMutable::get(_serviceContext)
+                            ->tickTo(VectorClock::Component::ClusterTime,
+                                     LogicalTime(*_decision->getCommitTimestamp()));
                     }
                 });
         })
diff --git a/src/mongo/db/s/vector_clock_config_server_test.cpp b/src/mongo/db/s/vector_clock_config_server_test.cpp
index 30761daea07..6c62f4834ac 100644
--- a/src/mongo/db/s/vector_clock_config_server_test.cpp
+++ b/src/mongo/db/s/vector_clock_config_server_test.cpp
@@ -29,22 +29,63 @@
 
 #include "mongo/platform/basic.h"
 
+#include "mongo/db/keys_collection_client_sharded.h"
+#include "mongo/db/keys_collection_manager.h"
+#include "mongo/db/logical_time_validator.h"
 #include "mongo/db/vector_clock_mutable.h"
 #include "mongo/s/config_server_test_fixture.h"
 #include "mongo/unittest/death_test.h"
 #include "mongo/unittest/unittest.h"
+#include "mongo/util/clock_source_mock.h"
 
 namespace mongo {
 namespace {
 
-using VectorClockConfigServerTest = ConfigServerTestFixture;
+class VectorClockConfigServerTest : public ConfigServerTestFixture {
+protected:
+    void setUp() override {
+        ConfigServerTestFixture::setUp();
+
+        auto clockSource = std::make_unique<ClockSourceMock>();
+        getServiceContext()->setFastClockSource(std::move(clockSource));
+
+        auto keysCollectionClient = std::make_unique<KeysCollectionClientSharded>(
+            Grid::get(operationContext())->catalogClient());
+
+        VectorClockMutable::get(getServiceContext())
+            ->tickTo(VectorClock::Component::ClusterTime, LogicalTime(Timestamp(1, 0)));
+
+        _keyManager = std::make_shared<KeysCollectionManager>(
+            "dummy", std::move(keysCollectionClient), Seconds(1000));
+        auto validator = std::make_unique<LogicalTimeValidator>(_keyManager);
+        validator->init(getServiceContext());
+        LogicalTimeValidator::set(getServiceContext(), std::move(validator));
+    }
+
+    void tearDown() override {
+        LogicalTimeValidator::get(getServiceContext())->shutDown();
+
+        ConfigServerTestFixture::tearDown();
+    }
+
+    /**
+     * Forces KeyManager to refresh cache and generate new keys.
+     */
+    void refreshKeyManager() {
+        _keyManager->refreshNow(operationContext());
+    }
+
+private:
+    std::shared_ptr<KeysCollectionManager> _keyManager;
+};
+
 
 TEST_F(VectorClockConfigServerTest, TickClusterTime) {
     auto sc = getGlobalServiceContext();
     auto vc = VectorClockMutable::get(sc);
 
     const auto t0 = vc->getTime();
-    ASSERT_EQ(LogicalTime(), t0[VectorClock::Component::ClusterTime]);
+    ASSERT_EQ(LogicalTime(Timestamp(1, 0)), t0[VectorClock::Component::ClusterTime]);
 
     const auto r1 = vc->tick(VectorClock::Component::ClusterTime, 1);
     const auto t1 = vc->getTime();
@@ -57,6 +98,32 @@ TEST_F(VectorClockConfigServerTest, TickClusterTime) {
     ASSERT_GT(t2[VectorClock::Component::ClusterTime], r1);
 }
 
+TEST_F(VectorClockConfigServerTest, TickToClusterTime) {
+    auto sc = getGlobalServiceContext();
+    auto vc = VectorClockMutable::get(sc);
+
+    const auto t0 = vc->getTime();
+    ASSERT_EQ(LogicalTime(Timestamp(1, 0)), t0[VectorClock::Component::ClusterTime]);
+
+    vc->tickTo(VectorClock::Component::ClusterTime, LogicalTime(Timestamp(1, 1)));
+    const auto t1 = vc->getTime();
+    ASSERT_EQ(LogicalTime(Timestamp(1, 1)), t1[VectorClock::Component::ClusterTime]);
+
+    vc->tickTo(VectorClock::Component::ClusterTime, LogicalTime(Timestamp(3, 3)));
+    const auto t2 = vc->getTime();
+    ASSERT_EQ(LogicalTime(Timestamp(3, 3)), t2[VectorClock::Component::ClusterTime]);
+
+    vc->tickTo(VectorClock::Component::ClusterTime, LogicalTime(Timestamp(2, 2)));
+    const auto t3 = vc->getTime();
+    ASSERT_EQ(LogicalTime(Timestamp(3, 3)), t3[VectorClock::Component::ClusterTime]);
+}
+
+DEATH_TEST_F(VectorClockConfigServerTest, CannotTickConfigTime, "Hit a MONGO_UNREACHABLE") {
+    auto sc = getGlobalServiceContext();
+    auto vc = VectorClockMutable::get(sc);
+    vc->tick(VectorClock::Component::ConfigTime, 1);
+}
+
 TEST_F(VectorClockConfigServerTest, TickToConfigTime) {
     auto sc = getGlobalServiceContext();
     auto vc = VectorClockMutable::get(sc);
@@ -77,38 +144,122 @@ TEST_F(VectorClockConfigServerTest, TickToConfigTime) {
     ASSERT_EQ(LogicalTime(Timestamp(3, 3)), t3[VectorClock::Component::ConfigTime]);
 }
 
-TEST_F(VectorClockConfigServerTest, GossipOutTest) {
-    // TODO SERVER-47914: after ClusterTime gossiping has been re-enabled: get the gossipOut
-    // internal and external, and for each check that $clusterTime and $configTime are there, with
-    // the right format, and right value.
+TEST_F(VectorClockConfigServerTest, GossipOutInternal) {
+    auto sc = getGlobalServiceContext();
+    auto vc = VectorClockMutable::get(sc);
+
+    LogicalTimeValidator::get(getServiceContext())->enableKeyGenerator(operationContext(), true);
+    refreshKeyManager();
 
-    // auto sc = getGlobalServiceContext();
-    // auto vc = VectorClockMutable::get(sc);
+    const auto clusterTime = vc->tick(VectorClock::Component::ClusterTime, 1);
+    const auto configTime = LogicalTime(Timestamp(1, 0));
+    vc->tickTo(VectorClock::Component::ConfigTime, configTime);
 
-    // const auto r1 = vc->tick(VectorClock::Component::ClusterTime, 1);
+    BSONObjBuilder bob;
+    vc->gossipOut(nullptr, &bob, transport::Session::kInternalClient);
+    auto obj = bob.obj();
+
+    // On config servers, gossip out to internal clients should have $clusterTime and $configTime.
+    ASSERT_TRUE(obj.hasField("$clusterTime"));
+    ASSERT_EQ(obj["$clusterTime"].Obj()["clusterTime"].timestamp(), clusterTime.asTimestamp());
+    ASSERT_TRUE(obj.hasField("$configTime"));
+    ASSERT_EQ(obj["$configTime"].timestamp(), configTime.asTimestamp());
 }
 
-TEST_F(VectorClockConfigServerTest, GossipInTest) {
-    // TODO SERVER-47914: after ClusterTime gossiping has been re-enabled: for each of gossipIn
-    // internal and external, give it BSON in the correct format, and then check that ClusterTime
-    // has been advanced (or not), ***and that ConfigTime has not***.
+TEST_F(VectorClockConfigServerTest, GossipOutExternal) {
+    auto sc = getGlobalServiceContext();
+    auto vc = VectorClockMutable::get(sc);
+
+    LogicalTimeValidator::get(getServiceContext())->enableKeyGenerator(operationContext(), true);
+    refreshKeyManager();
+
+    const auto clusterTime = vc->tick(VectorClock::Component::ClusterTime, 1);
+    const auto configTime = LogicalTime(Timestamp(1, 0));
+    vc->tickTo(VectorClock::Component::ConfigTime, configTime);
 
-    // auto sc = getGlobalServiceContext();
-    // auto vc = VectorClockMutable::get(sc);
+    BSONObjBuilder bob;
+    vc->gossipOut(nullptr, &bob);
+    auto obj = bob.obj();
 
-    // const auto r1 = vc->tick(VectorClock::Component::ClusterTime, 1);
+    // On config servers, gossip out to external clients should have $clusterTime, but not
+    // $configTime.
+    ASSERT_TRUE(obj.hasField("$clusterTime"));
+    ASSERT_EQ(obj["$clusterTime"].Obj()["clusterTime"].timestamp(), clusterTime.asTimestamp());
+    ASSERT_FALSE(obj.hasField("$configTime"));
 }
 
-DEATH_TEST_F(VectorClockConfigServerTest, CannotTickConfigTime, "Hit a MONGO_UNREACHABLE") {
+TEST_F(VectorClockConfigServerTest, GossipInInternal) {
     auto sc = getGlobalServiceContext();
     auto vc = VectorClockMutable::get(sc);
-    vc->tick(VectorClock::Component::ConfigTime, 1);
+
+    vc->tick(VectorClock::Component::ClusterTime, 1);
+    const auto configTime = LogicalTime(Timestamp(1, 0));
+    vc->tickTo(VectorClock::Component::ConfigTime, configTime);
+
+    auto dummySignature =
+        BSON("hash" << BSONBinData("\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1", 20, BinDataGeneral)
+                    << "keyId" << 0);
+    vc->gossipIn(nullptr,
+                 BSON("$clusterTime"
+                      << BSON("clusterTime" << Timestamp(2, 2) << "signature" << dummySignature)
+                      << "$configTime" << Timestamp(2, 2)),
+                 false,
+                 transport::Session::kInternalClient);
+
+    // On config servers, gossip in from internal clients should update $clusterTime, but not
+    // $configTime.
+    auto afterTime = vc->getTime();
+    ASSERT_EQ(afterTime[VectorClock::Component::ClusterTime].asTimestamp(), Timestamp(2, 2));
+    ASSERT_EQ(afterTime[VectorClock::Component::ConfigTime].asTimestamp(),
+              configTime.asTimestamp());
+
+    vc->gossipIn(nullptr,
+                 BSON("$clusterTime"
+                      << BSON("clusterTime" << Timestamp(1, 1) << "signature" << dummySignature)
+                      << "$configTime" << Timestamp(1, 1)),
+                 false,
+                 transport::Session::kInternalClient);
+
+    auto afterTime2 = vc->getTime();
+    ASSERT_EQ(afterTime2[VectorClock::Component::ClusterTime].asTimestamp(), Timestamp(2, 2));
+    ASSERT_EQ(afterTime2[VectorClock::Component::ConfigTime].asTimestamp(),
+              configTime.asTimestamp());
 }
 
-DEATH_TEST_F(VectorClockConfigServerTest, CannotTickToClusterTime, "Hit a MONGO_UNREACHABLE") {
+TEST_F(VectorClockConfigServerTest, GossipInExternal) {
     auto sc = getGlobalServiceContext();
     auto vc = VectorClockMutable::get(sc);
-    vc->tickTo(VectorClock::Component::ClusterTime, LogicalTime());
+
+    vc->tick(VectorClock::Component::ClusterTime, 1);
+    const auto configTime = LogicalTime(Timestamp(1, 0));
+    vc->tickTo(VectorClock::Component::ConfigTime, configTime);
+
+    auto dummySignature =
+        BSON("hash" << BSONBinData("\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1", 20, BinDataGeneral)
+                    << "keyId" << 0);
+    vc->gossipIn(nullptr,
+                 BSON("$clusterTime"
+                      << BSON("clusterTime" << Timestamp(2, 2) << "signature" << dummySignature)
+                      << "$configTime" << Timestamp(2, 2)),
+                 false);
+
+    // On config servers, gossip in from external clients should update $clusterTime, but not
+    // $configTime.
+    auto afterTime = vc->getTime();
+    ASSERT_EQ(afterTime[VectorClock::Component::ClusterTime].asTimestamp(), Timestamp(2, 2));
+    ASSERT_EQ(afterTime[VectorClock::Component::ConfigTime].asTimestamp(),
+              configTime.asTimestamp());
+
+    vc->gossipIn(nullptr,
+                 BSON("$clusterTime"
+                      << BSON("clusterTime" << Timestamp(1, 1) << "signature" << dummySignature)
+                      << "$configTime" << Timestamp(1, 1)),
+                 false);
+
+    auto afterTime2 = vc->getTime();
+    ASSERT_EQ(afterTime2[VectorClock::Component::ClusterTime].asTimestamp(), Timestamp(2, 2));
+    ASSERT_EQ(afterTime2[VectorClock::Component::ConfigTime].asTimestamp(),
+              configTime.asTimestamp());
 }
 
 }  // namespace
diff --git a/src/mongo/db/s/vector_clock_shard_server_test.cpp b/src/mongo/db/s/vector_clock_shard_server_test.cpp
index 594e7d86f59..66bae0d67a0 100644
--- a/src/mongo/db/s/vector_clock_shard_server_test.cpp
+++ b/src/mongo/db/s/vector_clock_shard_server_test.cpp
@@ -29,22 +29,62 @@
 
 #include "mongo/platform/basic.h"
 
+#include "mongo/db/keys_collection_client_direct.h"
+#include "mongo/db/keys_collection_manager.h"
+#include "mongo/db/logical_time_validator.h"
 #include "mongo/db/vector_clock_mutable.h"
 #include "mongo/s/shard_server_test_fixture.h"
 #include "mongo/unittest/death_test.h"
 #include "mongo/unittest/unittest.h"
+#include "mongo/util/clock_source_mock.h"
 
 namespace mongo {
 namespace {
 
-using VectorClockShardServerTest = ShardServerTestFixture;
+class VectorClockShardServerTest : public ShardServerTestFixture {
+protected:
+    void setUp() override {
+        ShardServerTestFixture::setUp();
+
+        auto clockSource = std::make_unique<ClockSourceMock>();
+        getServiceContext()->setFastClockSource(std::move(clockSource));
+
+        auto keysCollectionClient = std::make_unique<KeysCollectionClientDirect>();
+
+        VectorClockMutable::get(getServiceContext())
+            ->tickTo(VectorClock::Component::ClusterTime, LogicalTime(Timestamp(1, 0)));
+
+        _keyManager = std::make_shared<KeysCollectionManager>(
+            "dummy", std::move(keysCollectionClient), Seconds(1000));
+        auto validator = std::make_unique<LogicalTimeValidator>(_keyManager);
+        validator->init(getServiceContext());
+        LogicalTimeValidator::set(getServiceContext(), std::move(validator));
+    }
+
+    void tearDown() override {
+        LogicalTimeValidator::get(getServiceContext())->shutDown();
+
+        ShardServerTestFixture::tearDown();
+    }
+
+    /**
+     * Forces KeyManager to refresh cache and generate new keys.
+     */
+    void refreshKeyManager() {
+        _keyManager->refreshNow(operationContext());
+    }
+
+private:
+    std::shared_ptr<KeysCollectionManager> _keyManager;
+};
+
 
 TEST_F(VectorClockShardServerTest, TickClusterTime) {
     auto sc = getGlobalServiceContext();
     auto vc = VectorClockMutable::get(sc);
 
     const auto t0 = vc->getTime();
-    ASSERT_EQ(LogicalTime(), t0[VectorClock::Component::ClusterTime]);
+    ASSERT_EQ(LogicalTime(Timestamp(1, 0)), t0[VectorClock::Component::ClusterTime]);
 
     const auto r1 = vc->tick(VectorClock::Component::ClusterTime, 1);
     const auto t1 = vc->getTime();
@@ -57,26 +97,24 @@ TEST_F(VectorClockShardServerTest, TickClusterTime) {
     ASSERT_GT(t2[VectorClock::Component::ClusterTime], r1);
 }
 
-TEST_F(VectorClockShardServerTest, GossipOutTest) {
-    // TODO SERVER-47914: after ClusterTime gossiping has been re-enabled: get the gossipOut
-    // internal and external, and for each check that $clusterTime and $configTime are there, with
-    // the right format, and right value.
-
-    // auto sc = getGlobalServiceContext();
-    // auto vc = VectorClockMutable::get(sc);
+TEST_F(VectorClockShardServerTest, TickToClusterTime) {
+    auto sc = getGlobalServiceContext();
+    auto vc = VectorClockMutable::get(sc);
 
-    // const auto r1 = vc->tick(VectorClock::Component::ClusterTime, 1);
-}
+    const auto t0 = vc->getTime();
+    ASSERT_EQ(LogicalTime(Timestamp(1, 0)), t0[VectorClock::Component::ClusterTime]);
 
-TEST_F(VectorClockShardServerTest, GossipInTest) {
-    // TODO SERVER-47914: after ClusterTime gossiping has been re-enabled: for each of gossipIn
-    // internal and external, give it BSON in the correct format, and then check that ClusterTime
-    // and ConfigTime have been advanced (or not).
+    vc->tickTo(VectorClock::Component::ClusterTime, LogicalTime(Timestamp(1, 1)));
+    const auto t1 = vc->getTime();
+    ASSERT_EQ(LogicalTime(Timestamp(1, 1)), t1[VectorClock::Component::ClusterTime]);
 
-    // auto sc = getGlobalServiceContext();
-    // auto vc = VectorClockMutable::get(sc);
+    vc->tickTo(VectorClock::Component::ClusterTime, LogicalTime(Timestamp(3, 3)));
+    const auto t2 = vc->getTime();
+    ASSERT_EQ(LogicalTime(Timestamp(3, 3)), t2[VectorClock::Component::ClusterTime]);
 
-    // const auto r1 = vc->tick(VectorClock::Component::ClusterTime, 1);
+    vc->tickTo(VectorClock::Component::ClusterTime, LogicalTime(Timestamp(2, 2)));
+    const auto t3 = vc->getTime();
+    ASSERT_EQ(LogicalTime(Timestamp(3, 3)), t3[VectorClock::Component::ClusterTime]);
 }
 
 DEATH_TEST_F(VectorClockShardServerTest, CannotTickConfigTime, "Hit a MONGO_UNREACHABLE") {
@@ -85,16 +123,136 @@ DEATH_TEST_F(VectorClockShardServerTest, CannotTickConfigTime, "Hit a MONGO_UNRE
     vc->tick(VectorClock::Component::ConfigTime, 1);
 }
 
-DEATH_TEST_F(VectorClockShardServerTest, CannotTickToClusterTime, "Hit a MONGO_UNREACHABLE") {
+DEATH_TEST_F(VectorClockShardServerTest, CannotTickToConfigTime, "Hit a MONGO_UNREACHABLE") {
     auto sc = getGlobalServiceContext();
     auto vc = VectorClockMutable::get(sc);
-    vc->tickTo(VectorClock::Component::ClusterTime, LogicalTime());
+    vc->tickTo(VectorClock::Component::ConfigTime, LogicalTime());
 }
 
-DEATH_TEST_F(VectorClockShardServerTest, CannotTickToConfigTime, "Hit a MONGO_UNREACHABLE") {
+TEST_F(VectorClockShardServerTest, GossipOutInternal) {
     auto sc = getGlobalServiceContext();
     auto vc = VectorClockMutable::get(sc);
-    vc->tickTo(VectorClock::Component::ConfigTime, LogicalTime());
+
+    LogicalTimeValidator::get(getServiceContext())->enableKeyGenerator(operationContext(), true);
+    refreshKeyManager();
+
+    const auto clusterTime = vc->tick(VectorClock::Component::ClusterTime, 1);
+
+    BSONObjBuilder bob;
+    vc->gossipOut(nullptr, &bob, transport::Session::kInternalClient);
+    auto obj = bob.obj();
+
+    // On shard servers, gossip out to internal clients should have $clusterTime and $configTime.
+    ASSERT_TRUE(obj.hasField("$clusterTime"));
+    ASSERT_EQ(obj["$clusterTime"].Obj()["clusterTime"].timestamp(), clusterTime.asTimestamp());
+    ASSERT_TRUE(obj.hasField("$configTime"));
+    ASSERT_EQ(obj["$configTime"].timestamp(), Timestamp(0, 0));
+}
+
+TEST_F(VectorClockShardServerTest, GossipOutExternal) {
+    auto sc = getGlobalServiceContext();
+    auto vc = VectorClockMutable::get(sc);
+
+    LogicalTimeValidator::get(getServiceContext())->enableKeyGenerator(operationContext(), true);
+    refreshKeyManager();
+
+    const auto clusterTime = vc->tick(VectorClock::Component::ClusterTime, 1);
+
+    BSONObjBuilder bob;
+    vc->gossipOut(nullptr, &bob);
+    auto obj = bob.obj();
+
+    // On shard servers, gossip out to external clients should have $clusterTime, but not
+    // $configTime.
+    ASSERT_TRUE(obj.hasField("$clusterTime"));
+    ASSERT_EQ(obj["$clusterTime"].Obj()["clusterTime"].timestamp(), clusterTime.asTimestamp());
+    ASSERT_FALSE(obj.hasField("$configTime"));
+}
+
+TEST_F(VectorClockShardServerTest, GossipInInternal) {
+    auto sc = getGlobalServiceContext();
+    auto vc = VectorClockMutable::get(sc);
+
+    vc->tick(VectorClock::Component::ClusterTime, 1);
+
+    auto dummySignature =
+        BSON("hash" << BSONBinData("\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1", 20, BinDataGeneral)
+                    << "keyId" << 0);
+    vc->gossipIn(nullptr,
+                 BSON("$clusterTime"
+                      << BSON("clusterTime" << Timestamp(2, 2) << "signature" << dummySignature)
+                      << "$configTime" << Timestamp(2, 2)),
+                 false,
+                 transport::Session::kInternalClient);
+
+    // On shard servers, gossip in from internal clients should update $clusterTime and $configTime.
+    auto afterTime = vc->getTime();
+    ASSERT_EQ(afterTime[VectorClock::Component::ClusterTime].asTimestamp(), Timestamp(2, 2));
+    ASSERT_EQ(afterTime[VectorClock::Component::ConfigTime].asTimestamp(), Timestamp(2, 2));
+
+    vc->gossipIn(nullptr,
+                 BSON("$clusterTime"
+                      << BSON("clusterTime" << Timestamp(1, 1) << "signature" << dummySignature)
+                      << "$configTime" << Timestamp(1, 1)),
+                 false,
+                 transport::Session::kInternalClient);
+
+    auto afterTime2 = vc->getTime();
+    ASSERT_EQ(afterTime2[VectorClock::Component::ClusterTime].asTimestamp(), Timestamp(2, 2));
+    ASSERT_EQ(afterTime2[VectorClock::Component::ConfigTime].asTimestamp(), Timestamp(2, 2));
+
+    vc->gossipIn(nullptr,
+                 BSON("$clusterTime"
+                      << BSON("clusterTime" << Timestamp(3, 3) << "signature" << dummySignature)
+                      << "$configTime" << Timestamp(3, 3)),
+                 false,
+                 transport::Session::kInternalClient);
+
+    auto afterTime3 = vc->getTime();
+    ASSERT_EQ(afterTime3[VectorClock::Component::ClusterTime].asTimestamp(), Timestamp(3, 3));
+    ASSERT_EQ(afterTime3[VectorClock::Component::ConfigTime].asTimestamp(), Timestamp(3, 3));
+}
+
+TEST_F(VectorClockShardServerTest, GossipInExternal) {
+    auto sc = getGlobalServiceContext();
+    auto vc = VectorClockMutable::get(sc);
+
+    vc->tick(VectorClock::Component::ClusterTime, 1);
+
+    auto dummySignature =
+        BSON("hash" << BSONBinData("\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1", 20, BinDataGeneral)
+                    << "keyId" << 0);
+    vc->gossipIn(nullptr,
+                 BSON("$clusterTime"
+                      << BSON("clusterTime" << Timestamp(2, 2) << "signature" << dummySignature)
+                      << "$configTime" << Timestamp(2, 2)),
+                 false);
+
+    // On shard servers, gossip in from external clients should update $clusterTime, but not
+    // $configTime.
+    auto afterTime = vc->getTime();
+    ASSERT_EQ(afterTime[VectorClock::Component::ClusterTime].asTimestamp(), Timestamp(2, 2));
+    ASSERT_EQ(afterTime[VectorClock::Component::ConfigTime].asTimestamp(), Timestamp(0, 0));
+
+    vc->gossipIn(nullptr,
+                 BSON("$clusterTime"
+                      << BSON("clusterTime" << Timestamp(1, 1) << "signature" << dummySignature)
+                      << "$configTime" << Timestamp(1, 1)),
+                 false);
+
+    auto afterTime2 = vc->getTime();
+    ASSERT_EQ(afterTime2[VectorClock::Component::ClusterTime].asTimestamp(), Timestamp(2, 2));
+    ASSERT_EQ(afterTime2[VectorClock::Component::ConfigTime].asTimestamp(), Timestamp(0, 0));
+
+    vc->gossipIn(nullptr,
+                 BSON("$clusterTime"
+                      << BSON("clusterTime" << Timestamp(3, 3) << "signature" << dummySignature)
+                      << "$configTime" << Timestamp(3, 3)),
+                 false);
+
+    auto afterTime3 = vc->getTime();
+    ASSERT_EQ(afterTime3[VectorClock::Component::ClusterTime].asTimestamp(), Timestamp(3, 3));
+    ASSERT_EQ(afterTime3[VectorClock::Component::ConfigTime].asTimestamp(), Timestamp(0, 0));
 }
 
 }  // namespace
diff --git a/src/mongo/db/service_entry_point_common.cpp b/src/mongo/db/service_entry_point_common.cpp
index 8d7c6ce27e8..fa1e80051ac 100644
--- a/src/mongo/db/service_entry_point_common.cpp
+++ b/src/mongo/db/service_entry_point_common.cpp
@@ -87,7 +87,6 @@
 #include "mongo/rpc/get_status_from_command_result.h"
 #include "mongo/rpc/message.h"
 #include "mongo/rpc/metadata.h"
-#include "mongo/rpc/metadata/logical_time_metadata.h"
 #include "mongo/rpc/metadata/oplog_query_metadata.h"
 #include "mongo/rpc/metadata/repl_set_metadata.h"
 #include "mongo/rpc/metadata/tracking_metadata.h"
@@ -439,54 +438,20 @@ void appendClusterAndOperationTime(OperationContext* opCtx,
                                    LogicalTime startTime) {
     if (repl::ReplicationCoordinator::get(opCtx)->getReplicationMode() !=
             repl::ReplicationCoordinator::modeReplSet ||
-        !LogicalClock::get(opCtx)->isEnabled()) {
-        return;
-    }
-
-    VectorClock::get(opCtx)->gossipOut(metadataBob, opCtx->getClient()->getSessionTags());
-
-    // Authorized clients always receive operationTime and dummy signed $clusterTime.
-    if (LogicalTimeValidator::isAuthorizedToAdvanceClock(opCtx)) {
-        auto operationTime = computeOperationTime(opCtx, startTime);
-        auto signedTime = SignedLogicalTime(
-            LogicalClock::get(opCtx)->getClusterTime(), TimeProofService::TimeProof(), 0);
-
-        dassert(signedTime.getTime() >= operationTime);
-        rpc::LogicalTimeMetadata(signedTime).writeToMetadata(metadataBob);
-
-        LOGV2_DEBUG(
-            21957,
-            5,
-            "Appending operationTime to cmd response for authorized client: {operationTime}",
-            "operationTime"_attr = operationTime);
-        operationTime.appendAsOperationTime(commandBodyFieldsBob);
-
-        return;
-    }
-
-    // Servers without validators (e.g. a shard server not yet added to a cluster) do not return
-    // logical times to unauthorized clients.
-    auto validator = LogicalTimeValidator::get(opCtx);
-    if (!validator) {
+        !VectorClock::get(opCtx)->isEnabled()) {
         return;
     }
 
+    // The appended operationTime must always be <= the appended $clusterTime, so first compute the
+    // operationTime.
     auto operationTime = computeOperationTime(opCtx, startTime);
-    auto signedTime = validator->trySignLogicalTime(LogicalClock::get(opCtx)->getClusterTime());
 
-    // If there were no keys, do not return $clusterTime or operationTime to unauthorized clients.
-    if (signedTime.getKeyId() == 0) {
-        return;
-    }
-
-    dassert(signedTime.getTime() >= operationTime);
-    rpc::LogicalTimeMetadata(signedTime).writeToMetadata(metadataBob);
+    bool clusterTimeWasOutput = VectorClock::get(opCtx)->gossipOut(opCtx, metadataBob);
 
-    LOGV2_DEBUG(21958,
-                5,
-                "Appending operationTime to cmd response: {operationTime}",
-                "operationTime"_attr = operationTime);
-    operationTime.appendAsOperationTime(commandBodyFieldsBob);
+    // Ensure that either both operationTime and $clusterTime are output, or neither.
+    if (clusterTimeWasOutput) {
+        operationTime.appendAsOperationTime(metadataBob);
+    }
 }
 
 void appendErrorLabelsAndTopologyVersion(OperationContext* opCtx,
diff --git a/src/mongo/db/storage/SConscript b/src/mongo/db/storage/SConscript
index 7d2bc4aba98..740e673bd28 100644
--- a/src/mongo/db/storage/SConscript
+++ b/src/mongo/db/storage/SConscript
@@ -517,9 +517,9 @@ env.Library(
         '$BUILD_DIR/mongo/db/storage/storage_options',
     ],
     LIBDEPS_PRIVATE=[
-        '$BUILD_DIR/mongo/db/logical_clock',
         '$BUILD_DIR/mongo/db/storage/storage_repair_observer',
         '$BUILD_DIR/mongo/db/catalog/collection_catalog_helper',
+        '$BUILD_DIR/mongo/db/vector_clock',
         'two_phase_index_build_knobs_idl',
     ],
 )
diff --git a/src/mongo/db/update/SConscript b/src/mongo/db/update/SConscript
index 681e61d0944..4c9c00406c1 100644
--- a/src/mongo/db/update/SConscript
+++ b/src/mongo/db/update/SConscript
@@ -47,7 +47,6 @@ env.Library(
         'update_object_node.cpp',
     ],
     LIBDEPS=[
-        '$BUILD_DIR/mongo/db/logical_clock',
         '$BUILD_DIR/mongo/db/pipeline/pipeline',
         '$BUILD_DIR/mongo/db/update_index_data',
         '$BUILD_DIR/mongo/db/vector_clock_mutable',
@@ -101,7 +100,6 @@ env.CppUnitTest(
         '$BUILD_DIR/mongo/bson/mutable/mutable_bson',
         '$BUILD_DIR/mongo/bson/mutable/mutable_bson_test_utils',
         '$BUILD_DIR/mongo/db/exec/document_value/document_value_test_util',
-        '$BUILD_DIR/mongo/db/logical_clock',
         '$BUILD_DIR/mongo/db/matcher/expressions',
         '$BUILD_DIR/mongo/db/query/collation/collator_interface_mock',
         '$BUILD_DIR/mongo/db/query/query_planner',
diff --git a/src/mongo/db/update/current_date_node.cpp b/src/mongo/db/update/current_date_node.cpp
index 9df15695850..24ce4a95681 100644
--- a/src/mongo/db/update/current_date_node.cpp
+++ b/src/mongo/db/update/current_date_node.cpp
@@ -31,9 +31,8 @@
 
 #include "mongo/db/update/current_date_node.h"
 
-#include "mongo/db/logical_clock.h"
-#include "mongo/db/logical_time.h"
 #include "mongo/db/service_context.h"
+#include "mongo/db/vector_clock_mutable.h"
 
 namespace mongo {
 
@@ -42,12 +41,13 @@ constexpr StringData kType = "$type"_sd;
 constexpr StringData kDate = "date"_sd;
 constexpr StringData kTimestamp = "timestamp"_sd;
 
-void setValue(mutablebson::Element* element, bool typeIsDate) {
+void setValue(ServiceContext* service, mutablebson::Element* element, bool typeIsDate) {
     if (typeIsDate) {
         invariant(element->setValueDate(mongo::jsTime()));
     } else {
-        invariant(element->setValueTimestamp(
-            LogicalClock::get(getGlobalServiceContext())->reserveTicks(1).asTimestamp()));
+        invariant(element->setValueTimestamp(VectorClockMutable::get(service)
+                                                 ->tick(VectorClock::Component::ClusterTime, 1)
+                                                 .asTimestamp()));
     }
 }
 }  // namespace
@@ -92,17 +92,19 @@ Status CurrentDateNode::init(BSONElement modExpr,
                                        " or a $type expression ({$type: 'timestamp/date'}).");
     }
 
+    _service = expCtx->opCtx->getServiceContext();
+
     return Status::OK();
 }
 
 ModifierNode::ModifyResult CurrentDateNode::updateExistingElement(
     mutablebson::Element* element, std::shared_ptr<FieldRef> elementPath) const {
-    setValue(element, _typeIsDate);
+    setValue(_service, element, _typeIsDate);
     return ModifyResult::kNormalUpdate;
 }
 
 void CurrentDateNode::setValueForNewElement(mutablebson::Element* element) const {
-    setValue(element, _typeIsDate);
+    setValue(_service, element, _typeIsDate);
 }
 
 BSONObj CurrentDateNode::operatorValue() const {
diff --git a/src/mongo/db/update/current_date_node.h b/src/mongo/db/update/current_date_node.h
index a28de18ad36..2c3e4492f86 100644
--- a/src/mongo/db/update/current_date_node.h
+++ b/src/mongo/db/update/current_date_node.h
@@ -36,6 +36,8 @@
 
 namespace mongo {
 
+class ServiceContext;
+
 /**
  * Represents the application of a $currentDate to the value at the end of a path.
  */
@@ -71,6 +73,8 @@ private:
 
     // If true, the current date should be expressed as a Date. If false, a Timestamp.
     bool _typeIsDate;
+
+    ServiceContext* _service;
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/update/object_replace_executor.cpp b/src/mongo/db/update/object_replace_executor.cpp
index 31ea35df114..8f0251efc99 100644
--- a/src/mongo/db/update/object_replace_executor.cpp
+++ b/src/mongo/db/update/object_replace_executor.cpp
@@ -33,10 +33,9 @@
 
 #include "mongo/base/data_view.h"
 #include "mongo/db/bson/dotted_path_support.h"
-#include "mongo/db/logical_clock.h"
-#include "mongo/db/logical_time.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/update/storage_validation.h"
+#include "mongo/db/vector_clock_mutable.h"
 
 namespace mongo {
 
@@ -63,7 +62,9 @@ ObjectReplaceExecutor::ObjectReplaceExecutor(BSONObj replacement)
             unsigned long long timestamp = timestampView.read<unsigned long long>();
             if (timestamp == 0) {
                 ServiceContext* service = getGlobalServiceContext();
-                auto ts = LogicalClock::get(service)->reserveTicks(1).asTimestamp();
+                auto ts = VectorClockMutable::get(service)
+                              ->tick(VectorClock::Component::ClusterTime, 1)
+                              .asTimestamp();
                 timestampView.write(tagLittleEndian(ts.asULL()));
             }
         }
diff --git a/src/mongo/db/vector_clock.cpp b/src/mongo/db/vector_clock.cpp
index 5be53064537..24ce442a417 100644
--- a/src/mongo/db/vector_clock.cpp
+++ b/src/mongo/db/vector_clock.cpp
@@ -31,6 +31,11 @@
 
 #include "mongo/db/vector_clock.h"
 
+#include "mongo/bson/util/bson_extract.h"
+#include "mongo/db/auth/authorization_session.h"
+#include "mongo/db/logical_clock_gen.h"
+#include "mongo/db/logical_time_validator.h"
+
 namespace mongo {
 
 namespace {
@@ -65,8 +70,40 @@ VectorClock::VectorTime VectorClock::getTime() const {
     return VectorTime(_vectorTime);
 }
 
+bool VectorClock::_lessThanOrEqualToMaxPossibleTime(LogicalTime time, uint64_t nTicks) {
+    return time.asTimestamp().getSecs() <= kMaxValue &&
+        time.asTimestamp().getInc() <= (kMaxValue - nTicks);
+}
+
+void VectorClock::_ensurePassesRateLimiter(ServiceContext* service,
+                                           const LogicalTimeArray& newTime) {
+    const unsigned wallClockSecs =
+        durationCount<Seconds>(service->getFastClockSource()->now().toDurationSinceEpoch());
+    auto maxAcceptableDriftSecs = static_cast<const unsigned>(gMaxAcceptableLogicalClockDriftSecs);
+
+    for (auto newIt = newTime.begin(); newIt != newTime.end(); ++newIt) {
+        auto newTimeSecs = newIt->asTimestamp().getSecs();
+        auto name = _componentName(Component(newIt - newTime.begin()));
+
+        // Both values are unsigned, so compare them first to avoid wrap-around.
+        uassert(ErrorCodes::ClusterTimeFailsRateLimiter,
+                str::stream() << "New " << name << ", " << newTimeSecs
+                              << ", is too far from this node's wall clock time, " << wallClockSecs
+                              << ".",
+                ((newTimeSecs <= wallClockSecs) ||
+                 (newTimeSecs - wallClockSecs) <= maxAcceptableDriftSecs));
+
+        uassert(40484,
+                str::stream() << name << " cannot be advanced beyond its maximum value",
+                _lessThanOrEqualToMaxPossibleTime(*newIt, 0));
+    }
+}
+
 void VectorClock::_advanceTime(LogicalTimeArray&& newTime) {
+    _ensurePassesRateLimiter(_service, newTime);
+
     stdx::lock_guard<Latch> lock(_mutex);
+
     auto it = _vectorTime.begin();
     auto newIt = newTime.begin();
     for (; it != _vectorTime.end() && newIt != newTime.end(); ++it, ++newIt) {
@@ -80,14 +117,26 @@ class VectorClock::GossipFormat {
 public:
     class Plain;
     class Signed;
+    template <class ActualFormat>
+    class OnlyGossipOutOnNewFCV;
 
     static const ComponentArray<std::unique_ptr<GossipFormat>> _formatters;
 
     GossipFormat(std::string fieldName) : _fieldName(fieldName) {}
     virtual ~GossipFormat() = default;
 
-    virtual void out(BSONObjBuilder* out, LogicalTime time, Component component) const = 0;
-    virtual LogicalTime in(const BSONObj& in, Component component) const = 0;
+    // Returns true if the time was output, false otherwise.
+    virtual bool out(ServiceContext* service,
+                     OperationContext* opCtx,
+                     bool permitRefresh,
+                     BSONObjBuilder* out,
+                     LogicalTime time,
+                     Component component) const = 0;
+    virtual LogicalTime in(ServiceContext* service,
+                           OperationContext* opCtx,
+                           const BSONObj& in,
+                           bool couldBeUnauthenticated,
+                           Component component) const = 0;
 
     const std::string _fieldName;
 };
@@ -97,11 +146,21 @@ public:
     using GossipFormat::GossipFormat;
     virtual ~Plain() = default;
 
-    void out(BSONObjBuilder* out, LogicalTime time, Component component) const override {
+    bool out(ServiceContext* service,
+             OperationContext* opCtx,
+             bool permitRefresh,
+             BSONObjBuilder* out,
+             LogicalTime time,
+             Component component) const override {
         out->append(_fieldName, time.asTimestamp());
+        return true;
     }
 
-    LogicalTime in(const BSONObj& in, Component component) const override {
+    LogicalTime in(ServiceContext* service,
+                   OperationContext* opCtx,
+                   const BSONObj& in,
+                   bool couldBeUnauthenticated,
+                   Component component) const override {
         const auto componentElem(in[_fieldName]);
         if (componentElem.eoo()) {
             // Nothing to gossip in.
@@ -114,74 +173,225 @@ public:
     }
 };
 
+template <class ActualFormat>
+class VectorClock::GossipFormat::OnlyGossipOutOnNewFCV : public ActualFormat {
+public:
+    using ActualFormat::ActualFormat;
+    virtual ~OnlyGossipOutOnNewFCV() = default;
+
+    bool out(ServiceContext* service,
+             OperationContext* opCtx,
+             bool permitRefresh,
+             BSONObjBuilder* out,
+             LogicalTime time,
+             Component component) const override {
+        const auto& fcv = serverGlobalParams.featureCompatibility;
+        if (fcv.isVersionInitialized() &&
+            fcv.getVersion() ==
+                ServerGlobalParams::FeatureCompatibility::Version::kFullyUpgradedTo46) {
+            return ActualFormat::out(service, opCtx, permitRefresh, out, time, component);
+        }
+        return false;
+    }
+
+    LogicalTime in(ServiceContext* service,
+                   OperationContext* opCtx,
+                   const BSONObj& in,
+                   bool couldBeUnauthenticated,
+                   Component component) const override {
+        return ActualFormat::in(service, opCtx, in, couldBeUnauthenticated, component);
+    }
+};
+
 class VectorClock::GossipFormat::Signed : public VectorClock::GossipFormat {
 public:
     using GossipFormat::GossipFormat;
     virtual ~Signed() = default;
 
-    void out(BSONObjBuilder* out, LogicalTime time, Component component) const override {
-        // TODO SERVER-47914: make this do the actual proper signing
-        BSONObjBuilder bob;
-        bob.append("time", time.asTimestamp());
-        bob.append("signature", 0);
-        out->append(_fieldName, bob.done());
+    bool out(ServiceContext* service,
+             OperationContext* opCtx,
+             bool permitRefresh,
+             BSONObjBuilder* out,
+             LogicalTime time,
+             Component component) const override {
+        SignedLogicalTime signedTime;
+
+        if (opCtx && LogicalTimeValidator::isAuthorizedToAdvanceClock(opCtx)) {
+            // Authorized clients always receive a dummy-signed $clusterTime (and operationTime).
+            signedTime = SignedLogicalTime(time, TimeProofService::TimeProof(), 0);
+        } else {
+            // Servers without validators (e.g. a shard server not yet added to a cluster) do not
+            // return logical times to unauthorized clients.
+            auto validator = LogicalTimeValidator::get(service);
+            if (!validator) {
+                return false;
+            }
+
+            // There are some contexts where refreshing is not permitted.
+            if (permitRefresh && opCtx) {
+                signedTime = validator->signLogicalTime(opCtx, time);
+            } else {
+                signedTime = validator->trySignLogicalTime(time);
+            }
+
+            // If there were no keys, do not return $clusterTime (or operationTime) to unauthorized
+            // clients.
+            if (signedTime.getKeyId() == 0) {
+                return false;
+            }
+        }
+
+        // TODO SERVER-48432: use IDL to do this serialization.
+
+        BSONObjBuilder subObjBuilder(out->subobjStart(_fieldName));
+        signedTime.getTime().asTimestamp().append(subObjBuilder.bb(), kClusterTimeFieldName);
+
+        BSONObjBuilder signatureObjBuilder(subObjBuilder.subobjStart(kSignatureFieldName));
+        // Cluster time metadata is only written when the LogicalTimeValidator is set, which
+        // means the cluster time should always have a proof.
+        invariant(signedTime.getProof());
+        signedTime.getProof()->appendAsBinData(signatureObjBuilder, kSignatureHashFieldName);
+        signatureObjBuilder.append(kSignatureKeyIdFieldName, signedTime.getKeyId());
+        signatureObjBuilder.doneFast();
+
+        subObjBuilder.doneFast();
+
+        return true;
     }
 
-    LogicalTime in(const BSONObj& in, Component component) const override {
-        // TODO SERVER-47914: make this do the actual proper signing
-        const auto componentElem(in[_fieldName]);
-        if (componentElem.eoo()) {
+    LogicalTime in(ServiceContext* service,
+                   OperationContext* opCtx,
+                   const BSONObj& in,
+                   bool couldBeUnauthenticated,
+                   Component component) const override {
+        // TODO SERVER-48432: use IDL to do this deserialization.
+
+        const auto& metadataElem = in.getField(_fieldName);
+        if (metadataElem.eoo()) {
             // Nothing to gossip in.
             return LogicalTime();
         }
-        uassert(ErrorCodes::BadValue,
-                str::stream() << _fieldName << " is not a sub-object",
-                componentElem.isABSONObj());
-        const auto subobj = componentElem.embeddedObject();
-        const auto timeElem(subobj["time"]);
-        uassert(ErrorCodes::FailedToParse, "No time found", !timeElem.eoo());
-        uassert(ErrorCodes::BadValue,
-                str::stream() << "time is not a Timestamp",
-                timeElem.type() == bsonTimestamp);
-        return LogicalTime(timeElem.timestamp());
+
+        const auto& obj = metadataElem.Obj();
+
+        Timestamp ts;
+        uassertStatusOK(bsonExtractTimestampField(obj, kClusterTimeFieldName, &ts));
+
+        BSONElement signatureElem;
+        uassertStatusOK(bsonExtractTypedField(obj, kSignatureFieldName, Object, &signatureElem));
+
+        const auto& signatureObj = signatureElem.Obj();
+
+        // Extract BinData type signature hash and construct a SHA1Block instance from it.
+        BSONElement hashElem;
+        uassertStatusOK(
+            bsonExtractTypedField(signatureObj, kSignatureHashFieldName, BinData, &hashElem));
+
+        int hashLength = 0;
+        auto rawBinSignature = hashElem.binData(hashLength);
+        BSONBinData proofBinData(rawBinSignature, hashLength, hashElem.binDataType());
+        auto proofStatus = SHA1Block::fromBinData(proofBinData);
+        uassertStatusOK(proofStatus);
+
+        long long keyId;
+        uassertStatusOK(bsonExtractIntegerField(signatureObj, kSignatureKeyIdFieldName, &keyId));
+
+        auto signedTime =
+            SignedLogicalTime(LogicalTime(ts), std::move(proofStatus.getValue()), keyId);
+
+        if (!opCtx) {
+            // If there's no opCtx then this must be coming from a reply, which must be internal,
+            // and so doesn't require validation.
+            return signedTime.getTime();
+        }
+
+        // Validate the signature.
+        if (couldBeUnauthenticated && AuthorizationManager::get(service)->isAuthEnabled() &&
+            (!signedTime.getProof() || *signedTime.getProof() == TimeProofService::TimeProof())) {
+
+            AuthorizationSession* authSession = AuthorizationSession::get(opCtx->getClient());
+            // The client is not authenticated and is not using localhost auth bypass. Do not
+            // gossip.
+            if (authSession && !authSession->isAuthenticated() &&
+                !authSession->isUsingLocalhostBypass()) {
+                return {};
+            }
+        }
+
+        auto logicalTimeValidator = LogicalTimeValidator::get(service);
+        if (!LogicalTimeValidator::isAuthorizedToAdvanceClock(opCtx)) {
+            if (!logicalTimeValidator) {
+                uasserted(ErrorCodes::CannotVerifyAndSignLogicalTime,
+                          "Cannot accept logicalTime: " + signedTime.getTime().toString() +
+                              ". May not be a part of a sharded cluster");
+            } else {
+                uassertStatusOK(logicalTimeValidator->validate(opCtx, signedTime));
+            }
+        }
+
+        return signedTime.getTime();
     }
+
+private:
+    static constexpr char kClusterTimeFieldName[] = "clusterTime";
+    static constexpr char kSignatureFieldName[] = "signature";
+    static constexpr char kSignatureHashFieldName[] = "hash";
+    static constexpr char kSignatureKeyIdFieldName[] = "keyId";
 };
 
-// TODO SERVER-47914: update $clusterTimeNew to $clusterTime once LogicalClock is migrated into
-// VectorClock.
 const VectorClock::ComponentArray<std::unique_ptr<VectorClock::GossipFormat>>
     VectorClock::GossipFormat::_formatters{
-        std::make_unique<VectorClock::GossipFormat::Signed>("$clusterTimeNew"),
-        std::make_unique<VectorClock::GossipFormat::Plain>("$configTime")};
-
-void VectorClock::gossipOut(BSONObjBuilder* outMessage,
-                            const transport::Session::TagMask clientSessionTags) const {
+        std::make_unique<VectorClock::GossipFormat::Signed>(VectorClock::kClusterTimeFieldName),
+        std::make_unique<
+            VectorClock::GossipFormat::OnlyGossipOutOnNewFCV<VectorClock::GossipFormat::Plain>>(
+            VectorClock::kConfigTimeFieldName)};
+
+bool VectorClock::gossipOut(OperationContext* opCtx,
+                            BSONObjBuilder* outMessage,
+                            const transport::Session::TagMask defaultClientSessionTags) const {
+    auto clientSessionTags = defaultClientSessionTags;
+    if (opCtx && opCtx->getClient()) {
+        clientSessionTags = opCtx->getClient()->getSessionTags();
+    }
+    VectorTime now = getTime();
     if (clientSessionTags & transport::Session::kInternalClient) {
-        _gossipOutInternal(outMessage);
+        return _gossipOutInternal(opCtx, outMessage, now._time);
     } else {
-        _gossipOutExternal(outMessage);
+        return _gossipOutExternal(opCtx, outMessage, now._time);
     }
 }
 
-void VectorClock::gossipIn(const BSONObj& inMessage,
-                           const transport::Session::TagMask clientSessionTags) {
+void VectorClock::gossipIn(OperationContext* opCtx,
+                           const BSONObj& inMessage,
+                           bool couldBeUnauthenticated,
+                           const transport::Session::TagMask defaultClientSessionTags) {
+    auto clientSessionTags = defaultClientSessionTags;
+    if (opCtx && opCtx->getClient()) {
+        clientSessionTags = opCtx->getClient()->getSessionTags();
+    }
     if (clientSessionTags & transport::Session::kInternalClient) {
-        _advanceTime(_gossipInInternal(inMessage));
+        _advanceTime(_gossipInInternal(opCtx, inMessage, couldBeUnauthenticated));
     } else {
-        _advanceTime(_gossipInExternal(inMessage));
+        _advanceTime(_gossipInExternal(opCtx, inMessage, couldBeUnauthenticated));
     }
 }
 
-void VectorClock::_gossipOutComponent(BSONObjBuilder* out,
-                                      VectorTime time,
+bool VectorClock::_gossipOutComponent(OperationContext* opCtx,
+                                      BSONObjBuilder* out,
+                                      const LogicalTimeArray& time,
                                       Component component) const {
-    GossipFormat::_formatters[component]->out(out, time[component], component);
+    bool wasOutput = GossipFormat::_formatters[component]->out(
+        _service, opCtx, _permitRefreshDuringGossipOut(), out, time[component], component);
+    return (component == Component::ClusterTime) ? wasOutput : false;
 }
 
-void VectorClock::_gossipInComponent(const BSONObj& in,
+void VectorClock::_gossipInComponent(OperationContext* opCtx,
+                                     const BSONObj& in,
+                                     bool couldBeUnauthenticated,
                                      LogicalTimeArray* newTime,
                                      Component component) {
-    (*newTime)[component] = GossipFormat::_formatters[component]->in(in, component);
+    (*newTime)[component] = GossipFormat::_formatters[component]->in(
+        _service, opCtx, in, couldBeUnauthenticated, component);
 }
 
 std::string VectorClock::_componentName(Component component) {
@@ -193,9 +403,24 @@ bool VectorClock::isEnabled() const {
     return _isEnabled;
 }
 
-void VectorClock::disable() {
+void VectorClock::_disable() {
     stdx::lock_guard<Latch> lock(_mutex);
     _isEnabled = false;
 }
 
+void VectorClock::resetVectorClock_forTest() {
+    stdx::lock_guard<Latch> lock(_mutex);
+    auto it = _vectorTime.begin();
+    for (; it != _vectorTime.end(); ++it) {
+        *it = LogicalTime();
+    }
+    _isEnabled = true;
+}
+
+void VectorClock::advanceClusterTime_forTest(LogicalTime newClusterTime) {
+    LogicalTimeArray newTime;
+    newTime[Component::ClusterTime] = newClusterTime;
+    _advanceTime(std::move(newTime));
+}
+
 }  // namespace mongo
diff --git a/src/mongo/db/vector_clock.h b/src/mongo/db/vector_clock.h
index b7381652e9c..db2a74e6140 100644
--- a/src/mongo/db/vector_clock.h
+++ b/src/mongo/db/vector_clock.h
@@ -93,40 +93,154 @@ public:
         const LogicalTimeArray _time;
     };
 
+    static constexpr char kClusterTimeFieldName[] = "$clusterTime";
+    static constexpr char kConfigTimeFieldName[] = "$configTime";
+
     // Decorate ServiceContext with VectorClock* which points to the actual vector clock
     // implementation.
     static VectorClock* get(ServiceContext* service);
     static VectorClock* get(OperationContext* ctx);
-
     static void registerVectorClockOnServiceContext(ServiceContext* service,
                                                     VectorClock* vectorClock);
 
+    /**
+     * Returns an instantaneous snapshot of the current time of all components.
+     */
     VectorTime getTime() const;
 
-    // Gossipping
-    void gossipOut(BSONObjBuilder* outMessage,
-                   const transport::Session::TagMask clientSessionTags) const;
-    void gossipIn(const BSONObj& inMessage, const transport::Session::TagMask clientSessionTags);
+    /**
+     * Adds the necessary fields to outMessage to gossip the current time to another node, taking
+     * into account if the gossiping is to an internal or external client (based on the session
+     * tags).  Returns true if the ClusterTime was output into outMessage, or false otherwise.
+     */
+    bool gossipOut(OperationContext* opCtx,
+                   BSONObjBuilder* outMessage,
+                   const transport::Session::TagMask defaultClientSessionTags = 0) const;
+    /**
+     * Read the necessary fields from inMessage in order to update the current time, based on this
+     * message received from another node, taking into account if the gossiping is from an internal
+     * or external client (based on the session tags).
+     */
+    void gossipIn(OperationContext* opCtx,
+                  const BSONObj& inMessage,
+                  bool couldBeUnauthenticated,
+                  const transport::Session::TagMask defaultClientSessionTags = 0);
 
+    /**
+     * Returns true if the clock is enabled and can be used. Defaults to true.
+     */
     bool isEnabled() const;
-    void disable();
+
+    void resetVectorClock_forTest();
+    void advanceClusterTime_forTest(LogicalTime newClusterTime);
 
 protected:
     VectorClock();
     virtual ~VectorClock();
 
+    /**
+     * The maximum permissible value for each part of a LogicalTime's Timestamp (ie. "secs" and
+     * "inc").
+     */
+    static constexpr uint32_t kMaxValue = std::numeric_limits<int32_t>::max();
+
+    /**
+     * The "name" of the given component, for user-facing error messages. The name used is the
+     * field name used when gossiping.
+     */
     static std::string _componentName(Component component);
 
-    // Internal Gossipping API
-    virtual void _gossipOutInternal(BSONObjBuilder* out) const = 0;
-    virtual void _gossipOutExternal(BSONObjBuilder* out) const = 0;
-    virtual LogicalTimeArray _gossipInInternal(const BSONObj& in) = 0;
-    virtual LogicalTimeArray _gossipInExternal(const BSONObj& in) = 0;
+    /**
+     * Disables the clock. A disabled clock won't process logical times and can't be re-enabled.
+     */
+    void _disable();
+
+    /**
+     * "Rate limiter" for advancing logical times. Rejects newTime if any of its Components have a
+     * seconds value that's more than gMaxAcceptableLogicalClockDriftSecs ahead of this node's wall
+     * clock.
+     */
+    static void _ensurePassesRateLimiter(ServiceContext* service, const LogicalTimeArray& newTime);
+
+    /**
+     * Used to ensure that gossiped or ticked times never overflow the maximum possible LogicalTime.
+     */
+    static bool _lessThanOrEqualToMaxPossibleTime(LogicalTime time, uint64_t nTicks);
+
+
+    /**
+     * Adds the necessary fields to outMessage to gossip the given time to a node internal to the
+     * cluster.  Returns true if the ClusterTime was output into outMessage, or false otherwise.
+     */
+    virtual bool _gossipOutInternal(OperationContext* opCtx,
+                                    BSONObjBuilder* out,
+                                    const LogicalTimeArray& time) const = 0;
+
+    /**
+     * As for _gossipOutInternal, except for an outMessage to be sent to a client external to the
+     * cluster, eg. a driver or user client.
+     */
+    virtual bool _gossipOutExternal(OperationContext* opCtx,
+                                    BSONObjBuilder* out,
+                                    const LogicalTimeArray& time) const = 0;
+
+    /**
+     * Reads the necessary fields from the BSONObj, which has come from a node internal to the
+     * cluster, and returns an array of LogicalTimes for each component present in the BSONObj.
+     *
+     * This array is suitable for passing to _advanceTime(), in order to monotonically increase
+     * any Component times that are larger than the current time.  Since the times in
+     * LogicalTimeArray are default constructed (ie. to Timestamp(0, 0)), any fields not present
+     * in the input BSONObj won't be advanced.
+     *
+     * The couldBeUnauthenticated parameter is used to indicate if the source of the input BSONObj
+     * is an incoming request for a command that can be run by an unauthenticated client.
+     */
+    virtual LogicalTimeArray _gossipInInternal(OperationContext* opCtx,
+                                               const BSONObj& in,
+                                               bool couldBeUnauthenticated) = 0;
+
+    /**
+     * As for _gossipInInternal, except for an input BSONObj from a client external to the cluster,
+     * eg. a driver or user client.
+     */
+    virtual LogicalTimeArray _gossipInExternal(OperationContext* opCtx,
+                                               const BSONObj& in,
+                                               bool couldBeUnauthenticated) = 0;
+
+    /**
+     * Whether or not it's permissable to refresh external state (eg. updating gossip signing keys)
+     * during gossip out.
+     */
+    virtual bool _permitRefreshDuringGossipOut() const = 0;
+
+    /**
+     * Called by sub-classes in order to actually output a Component time to the output
+     * BSONObjBuilder, using the appropriate field name and representation for that Component.
+     *
+     * Returns true if the component is ClusterTime and it was output, or false otherwise.
+     */
+    bool _gossipOutComponent(OperationContext* opCtx,
+                             BSONObjBuilder* out,
+                             const LogicalTimeArray& time,
+                             Component component) const;
 
-    void _gossipOutComponent(BSONObjBuilder* out, VectorTime time, Component component) const;
-    void _gossipInComponent(const BSONObj& in, LogicalTimeArray* newTime, Component component);
+    /**
+     * Called by sub-classes in order to actually input a Component time into the given
+     * LogicalTimeArray from the given BSONObj, using the appropriate field name and representation
+     * for that Component.
+     */
+    void _gossipInComponent(OperationContext* opCtx,
+                            const BSONObj& in,
+                            bool couldBeUnauthenticated,
+                            LogicalTimeArray* newTime,
+                            Component component);
 
-    // Used to atomically advance the time of several components (eg. during gossip-in).
+    /**
+     * For each component in the LogicalTimeArray, sets the current time to newTime if the newTime >
+     * current time and it passes the rate check.  If any component fails the rate check, then this
+     * function uasserts on the first such component (without setting any current times).
+     */
     void _advanceTime(LogicalTimeArray&& newTime);
 
     ServiceContext* _service{nullptr};
diff --git a/src/mongo/db/vector_clock_mongod.cpp b/src/mongo/db/vector_clock_mongod.cpp
index 4f5248aed29..3a270314da0 100644
--- a/src/mongo/db/vector_clock_mongod.cpp
+++ b/src/mongo/db/vector_clock_mongod.cpp
@@ -54,26 +54,25 @@ public:
     void tickTo(Component component, LogicalTime newTime) override;
 
 protected:
-    void _gossipOutInternal(BSONObjBuilder* out) const override;
-    void _gossipOutExternal(BSONObjBuilder* out) const override;
-    LogicalTimeArray _gossipInInternal(const BSONObj& in) override;
-    LogicalTimeArray _gossipInExternal(const BSONObj& in) override;
+    bool _gossipOutInternal(OperationContext* opCtx,
+                            BSONObjBuilder* out,
+                            const LogicalTimeArray& time) const override;
+    bool _gossipOutExternal(OperationContext* opCtx,
+                            BSONObjBuilder* out,
+                            const LogicalTimeArray& time) const override;
+    LogicalTimeArray _gossipInInternal(OperationContext* opCtx,
+                                       const BSONObj& in,
+                                       bool couldBeUnauthenticated) override;
+    LogicalTimeArray _gossipInExternal(OperationContext* opCtx,
+                                       const BSONObj& in,
+                                       bool couldBeUnauthenticated) override;
+    bool _permitRefreshDuringGossipOut() const override;
 
 private:
-    enum class ReplState {
-        Unset,
-        StepUpBegin,
-        StepUpComplete,
-        StepDown,
-        Arbiter,
-    };
-
-    void onStepUpBegin(OperationContext* opCtx) override;
-    void onStepUpComplete(OperationContext* opCtx) override;
-    void onStepDown() override;
+    void onStepUpBegin(OperationContext* opCtx) override {}
+    void onStepUpComplete(OperationContext* opCtx) override {}
+    void onStepDown() override {}
     void onBecomeArbiter() override;
-
-    ReplState _replState{ReplState::Unset};
 };
 
 const auto vectorClockMongoDDecoration = ServiceContext::declareDecoration<VectorClockMongoD>();
@@ -98,62 +97,54 @@ VectorClockMongoD::VectorClockMongoD() = default;
 
 VectorClockMongoD::~VectorClockMongoD() = default;
 
-void VectorClockMongoD::onStepUpBegin(OperationContext* opCtx) {
-    _replState = ReplState::StepUpBegin;
-}
-
-void VectorClockMongoD::onStepUpComplete(OperationContext* opCtx) {
-    _replState = ReplState::StepUpComplete;
-}
-
-void VectorClockMongoD::onStepDown() {
-    _replState = ReplState::StepDown;
-}
-
 void VectorClockMongoD::onBecomeArbiter() {
-    _replState = ReplState::Arbiter;
-
     // The node has become an arbiter, hence will not need logical clock for external operations.
-    disable();
+    _disable();
+    if (auto validator = LogicalTimeValidator::get(_service)) {
+        validator->stopKeyManager();
+    }
 }
 
-void VectorClockMongoD::_gossipOutInternal(BSONObjBuilder* out) const {
-    VectorTime now = getTime();
-    // TODO SERVER-47914: re-enable gossipping of VectorClock's ClusterTime once LogicalClock has
-    // been migrated into VectorClock.
-    // _gossipOutComponent(out, now, Component::ClusterTime);
+bool VectorClockMongoD::_gossipOutInternal(OperationContext* opCtx,
+                                           BSONObjBuilder* out,
+                                           const LogicalTimeArray& time) const {
+    bool wasClusterTimeOutput = _gossipOutComponent(opCtx, out, time, Component::ClusterTime);
     if (serverGlobalParams.clusterRole == ClusterRole::ShardServer ||
         serverGlobalParams.clusterRole == ClusterRole::ConfigServer) {
-        _gossipOutComponent(out, now, Component::ConfigTime);
+        _gossipOutComponent(opCtx, out, time, Component::ConfigTime);
     }
+    return wasClusterTimeOutput;
 }
 
-void VectorClockMongoD::_gossipOutExternal(BSONObjBuilder* out) const {
-    // TODO SERVER-47914: re-enable gossipping of VectorClock's ClusterTime once LogicalClock has
-    // been migrated into VectorClock.
-    // VectorTime now = getTime();
-    // _gossipOutComponent(out, now, Component::ClusterTime);
+bool VectorClockMongoD::_gossipOutExternal(OperationContext* opCtx,
+                                           BSONObjBuilder* out,
+                                           const LogicalTimeArray& time) const {
+    return _gossipOutComponent(opCtx, out, time, Component::ClusterTime);
 }
 
-VectorClock::LogicalTimeArray VectorClockMongoD::_gossipInInternal(const BSONObj& in) {
+VectorClock::LogicalTimeArray VectorClockMongoD::_gossipInInternal(OperationContext* opCtx,
+                                                                   const BSONObj& in,
+                                                                   bool couldBeUnauthenticated) {
     LogicalTimeArray newTime;
-    // TODO SERVER-47914: re-enable gossipping of VectorClock's ClusterTime once LogicalClock has
-    // been migrated into VectorClock.
-    // _gossipInComponent(in, &newTime, Component::ClusterTime);
+    _gossipInComponent(opCtx, in, couldBeUnauthenticated, &newTime, Component::ClusterTime);
     if (serverGlobalParams.clusterRole == ClusterRole::ShardServer) {
-        _gossipInComponent(in, &newTime, Component::ConfigTime);
+        _gossipInComponent(opCtx, in, couldBeUnauthenticated, &newTime, Component::ConfigTime);
     }
     return newTime;
 }
 
-VectorClock::LogicalTimeArray VectorClockMongoD::_gossipInExternal(const BSONObj& in) {
+VectorClock::LogicalTimeArray VectorClockMongoD::_gossipInExternal(OperationContext* opCtx,
+                                                                   const BSONObj& in,
+                                                                   bool couldBeUnauthenticated) {
     LogicalTimeArray newTime;
-    // TODO SERVER-47914: re-enable gossipping of VectorClock's ClusterTime once LogicalClock has
-    // been migrated into VectorClock.
-    // _gossipInComponent(in, &newTime, Component::ClusterTime);
+    _gossipInComponent(opCtx, in, couldBeUnauthenticated, &newTime, Component::ClusterTime);
     return newTime;
 }
 
+bool VectorClockMongoD::_permitRefreshDuringGossipOut() const {
+    return false;
+}
+
 LogicalTime VectorClockMongoD::tick(Component component, uint64_t nTicks) {
     if (component == Component::ClusterTime) {
         // Although conceptually ClusterTime can only be ticked when a mongod is able to take writes
@@ -181,6 +172,13 @@ void VectorClockMongoD::tickTo(Component component, LogicalTime newTime) {
         return;
     }
 
+    if (component == Component::ClusterTime) {
+        // The ClusterTime is allowed to tickTo in certain very limited and trusted cases (eg.
+        // initializing based on oplog timestamps), so we have to allow it here.
+        _advanceComponentTimeTo(component, std::move(newTime));
+        return;
+    }
+
     // tickTo is not permitted in other circumstances.
     MONGO_UNREACHABLE;
 }
diff --git a/src/mongo/db/vector_clock_mongod_test.cpp b/src/mongo/db/vector_clock_mongod_test.cpp
index 1a443ca68ba..cbb34b41ccd 100644
--- a/src/mongo/db/vector_clock_mongod_test.cpp
+++ b/src/mongo/db/vector_clock_mongod_test.cpp
@@ -29,22 +29,67 @@
 
 #include "mongo/platform/basic.h"
 
-#include "mongo/db/service_context_test_fixture.h"
+#include "mongo/db/keys_collection_client_direct.h"
+#include "mongo/db/keys_collection_manager.h"
+#include "mongo/db/logical_time_validator.h"
 #include "mongo/db/vector_clock_mutable.h"
+#include "mongo/s/sharding_mongod_test_fixture.h"
 #include "mongo/unittest/death_test.h"
 #include "mongo/unittest/unittest.h"
+#include "mongo/util/clock_source_mock.h"
 
 namespace mongo {
 namespace {
 
-using VectorClockMongoDTest = ServiceContextTest;
+/**
+ * Even though these tests are to exercise logic for plain replica set members, it uses
+ * ShardingMongodTestFixture as a convenient way to get the necessary support infrastructure (such
+ * as a TaskExecutor with pool), while still being neither "config server" nor "shard server".
+ */
+class VectorClockMongoDTest : public ShardingMongodTestFixture {
+protected:
+    void setUp() override {
+        ShardingMongodTestFixture::setUp();
+
+        auto clockSource = std::make_unique<ClockSourceMock>();
+        getServiceContext()->setFastClockSource(std::move(clockSource));
+
+        auto keysCollectionClient = std::make_unique<KeysCollectionClientDirect>();
+
+        VectorClockMutable::get(getServiceContext())
+            ->tickTo(VectorClock::Component::ClusterTime, LogicalTime(Timestamp(1, 0)));
+
+        _keyManager = std::make_shared<KeysCollectionManager>(
+            "dummy", std::move(keysCollectionClient), Seconds(1000));
+        auto validator = std::make_unique<LogicalTimeValidator>(_keyManager);
+        validator->init(getServiceContext());
+        LogicalTimeValidator::set(getServiceContext(), std::move(validator));
+    }
+
+    void tearDown() override {
+        LogicalTimeValidator::get(getServiceContext())->shutDown();
+
+        ShardingMongodTestFixture::tearDown();
+    }
+
+    /**
+     * Forces KeyManager to refresh cache and generate new keys.
+     */
+    void refreshKeyManager() {
+        _keyManager->refreshNow(operationContext());
+    }
+
+private:
+    std::shared_ptr<KeysCollectionManager> _keyManager;
+};
+
 
 TEST_F(VectorClockMongoDTest, TickClusterTime) {
     auto sc = getGlobalServiceContext();
     auto vc = VectorClockMutable::get(sc);
 
     const auto t0 = vc->getTime();
-    ASSERT_EQ(LogicalTime(), t0[VectorClock::Component::ClusterTime]);
+    ASSERT_EQ(LogicalTime(Timestamp(1, 0)), t0[VectorClock::Component::ClusterTime]);
 
     const auto r1 = vc->tick(VectorClock::Component::ClusterTime, 1);
     const auto t1 = vc->getTime();
@@ -57,26 +102,24 @@ TEST_F(VectorClockMongoDTest, TickClusterTime) {
     ASSERT_GT(t2[VectorClock::Component::ClusterTime], r1);
 }
 
-TEST_F(VectorClockMongoDTest, GossipOutTest) {
-    // TODO SERVER-47914: after ClusterTime gossiping has been re-enabled: get the gossipOut
-    // internal and external, and for each check that $clusterTime is there, with the right format,
-    // and right value, and not configTime.
-
-    // auto sc = getGlobalServiceContext();
-    // auto vc = VectorClockMutable::get(sc);
+TEST_F(VectorClockMongoDTest, TickToClusterTime) {
+    auto sc = getGlobalServiceContext();
+    auto vc = VectorClockMutable::get(sc);
 
-    // const auto r1 = vc->tick(VectorClock::Component::ClusterTime, 1);
-}
+    const auto t0 = vc->getTime();
+    ASSERT_EQ(LogicalTime(Timestamp(1, 0)), t0[VectorClock::Component::ClusterTime]);
 
-TEST_F(VectorClockMongoDTest, GossipInTest) {
-    // TODO SERVER-47914: after ClusterTime gossiping has been re-enabled: for each of gossipIn
-    // internal and external, give it BSON in the correct format, and then check that ClusterTime
-    // has been advanced (or not), and that ConfigTime has not.
+    vc->tickTo(VectorClock::Component::ClusterTime, LogicalTime(Timestamp(1, 1)));
+    const auto t1 = vc->getTime();
+    ASSERT_EQ(LogicalTime(Timestamp(1, 1)), t1[VectorClock::Component::ClusterTime]);
 
-    // auto sc = getGlobalServiceContext();
-    // auto vc = VectorClockMutable::get(sc);
+    vc->tickTo(VectorClock::Component::ClusterTime, LogicalTime(Timestamp(3, 3)));
+    const auto t2 = vc->getTime();
+    ASSERT_EQ(LogicalTime(Timestamp(3, 3)), t2[VectorClock::Component::ClusterTime]);
 
-    // const auto r1 = vc->tick(VectorClock::Component::ClusterTime, 1);
+    vc->tickTo(VectorClock::Component::ClusterTime, LogicalTime(Timestamp(2, 2)));
+    const auto t3 = vc->getTime();
+    ASSERT_EQ(LogicalTime(Timestamp(3, 3)), t3[VectorClock::Component::ClusterTime]);
 }
 
 DEATH_TEST_F(VectorClockMongoDTest, CannotTickConfigTime, "Hit a MONGO_UNREACHABLE") {
@@ -85,16 +128,137 @@ DEATH_TEST_F(VectorClockMongoDTest, CannotTickConfigTime, "Hit a MONGO_UNREACHAB
     vc->tick(VectorClock::Component::ConfigTime, 1);
 }
 
-DEATH_TEST_F(VectorClockMongoDTest, CannotTickToClusterTime, "Hit a MONGO_UNREACHABLE") {
+DEATH_TEST_F(VectorClockMongoDTest, CannotTickToConfigTime, "Hit a MONGO_UNREACHABLE") {
     auto sc = getGlobalServiceContext();
     auto vc = VectorClockMutable::get(sc);
-    vc->tickTo(VectorClock::Component::ClusterTime, LogicalTime());
+    vc->tickTo(VectorClock::Component::ConfigTime, LogicalTime());
 }
 
-DEATH_TEST_F(VectorClockMongoDTest, CannotTickToConfigTime, "Hit a MONGO_UNREACHABLE") {
+TEST_F(VectorClockMongoDTest, GossipOutInternal) {
     auto sc = getGlobalServiceContext();
     auto vc = VectorClockMutable::get(sc);
-    vc->tickTo(VectorClock::Component::ConfigTime, LogicalTime());
+
+    LogicalTimeValidator::get(getServiceContext())->enableKeyGenerator(operationContext(), true);
+    refreshKeyManager();
+
+    const auto clusterTime = vc->tick(VectorClock::Component::ClusterTime, 1);
+
+    BSONObjBuilder bob;
+    vc->gossipOut(nullptr, &bob, transport::Session::kInternalClient);
+    auto obj = bob.obj();
+
+    // On plain replset servers, gossip out to internal clients should have $clusterTime, but not
+    // $configTime.
+    ASSERT_TRUE(obj.hasField("$clusterTime"));
+    ASSERT_EQ(obj["$clusterTime"].Obj()["clusterTime"].timestamp(), clusterTime.asTimestamp());
+    ASSERT_FALSE(obj.hasField("$configTime"));
+}
+
+TEST_F(VectorClockMongoDTest, GossipOutExternal) {
+    auto sc = getGlobalServiceContext();
+    auto vc = VectorClockMutable::get(sc);
+
+    LogicalTimeValidator::get(getServiceContext())->enableKeyGenerator(operationContext(), true);
+    refreshKeyManager();
+
+    const auto clusterTime = vc->tick(VectorClock::Component::ClusterTime, 1);
+
+    BSONObjBuilder bob;
+    vc->gossipOut(nullptr, &bob);
+    auto obj = bob.obj();
+
+    // On plain replset servers, gossip out to external clients should have $clusterTime, but not
+    // $configTime.
+    ASSERT_TRUE(obj.hasField("$clusterTime"));
+    ASSERT_EQ(obj["$clusterTime"].Obj()["clusterTime"].timestamp(), clusterTime.asTimestamp());
+    ASSERT_FALSE(obj.hasField("$configTime"));
+}
+
+TEST_F(VectorClockMongoDTest, GossipInInternal) {
+    auto sc = getGlobalServiceContext();
+    auto vc = VectorClockMutable::get(sc);
+
+    vc->tick(VectorClock::Component::ClusterTime, 1);
+
+    auto dummySignature =
+        BSON("hash" << BSONBinData("\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1", 20, BinDataGeneral)
+                    << "keyId" << 0);
+    vc->gossipIn(nullptr,
+                 BSON("$clusterTime"
+                      << BSON("clusterTime" << Timestamp(2, 2) << "signature" << dummySignature)
+                      << "$configTime" << Timestamp(2, 2)),
+                 false,
+                 transport::Session::kInternalClient);
+
+    // On plain replset servers, gossip in from internal clients should update $clusterTime, but not
+    // $configTime.
+    auto afterTime = vc->getTime();
+    ASSERT_EQ(afterTime[VectorClock::Component::ClusterTime].asTimestamp(), Timestamp(2, 2));
+    ASSERT_EQ(afterTime[VectorClock::Component::ConfigTime].asTimestamp(), Timestamp(0, 0));
+
+    vc->gossipIn(nullptr,
+                 BSON("$clusterTime"
+                      << BSON("clusterTime" << Timestamp(1, 1) << "signature" << dummySignature)
+                      << "$configTime" << Timestamp(1, 1)),
+                 false,
+                 transport::Session::kInternalClient);
+
+    auto afterTime2 = vc->getTime();
+    ASSERT_EQ(afterTime2[VectorClock::Component::ClusterTime].asTimestamp(), Timestamp(2, 2));
+    ASSERT_EQ(afterTime2[VectorClock::Component::ConfigTime].asTimestamp(), Timestamp(0, 0));
+
+    vc->gossipIn(nullptr,
+                 BSON("$clusterTime"
+                      << BSON("clusterTime" << Timestamp(3, 3) << "signature" << dummySignature)
+                      << "$configTime" << Timestamp(3, 3)),
+                 false,
+                 transport::Session::kInternalClient);
+
+    auto afterTime3 = vc->getTime();
+    ASSERT_EQ(afterTime3[VectorClock::Component::ClusterTime].asTimestamp(), Timestamp(3, 3));
+    ASSERT_EQ(afterTime3[VectorClock::Component::ConfigTime].asTimestamp(), Timestamp(0, 0));
+}
+
+TEST_F(VectorClockMongoDTest, GossipInExternal) {
+    auto sc = getGlobalServiceContext();
+    auto vc = VectorClockMutable::get(sc);
+
+    vc->tick(VectorClock::Component::ClusterTime, 1);
+
+    auto dummySignature =
+        BSON("hash" << BSONBinData("\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1\1", 20, BinDataGeneral)
+                    << "keyId" << 0);
+    vc->gossipIn(nullptr,
+                 BSON("$clusterTime"
+                      << BSON("clusterTime" << Timestamp(2, 2) << "signature" << dummySignature)
+                      << "$configTime" << Timestamp(2, 2)),
+                 false);
+
+    // On plain replset servers, gossip in from external clients should update $clusterTime, but not
+    // $configTime.
+    auto afterTime = vc->getTime();
+    ASSERT_EQ(afterTime[VectorClock::Component::ClusterTime].asTimestamp(), Timestamp(2, 2));
+    ASSERT_EQ(afterTime[VectorClock::Component::ConfigTime].asTimestamp(), Timestamp(0, 0));
+
+    vc->gossipIn(nullptr,
+                 BSON("$clusterTime"
+                      << BSON("clusterTime" << Timestamp(1, 1) << "signature" << dummySignature)
+                      << "$configTime" << Timestamp(1, 1)),
+                 false);
+
+    auto afterTime2 = vc->getTime();
+    ASSERT_EQ(afterTime2[VectorClock::Component::ClusterTime].asTimestamp(), Timestamp(2, 2));
+    ASSERT_EQ(afterTime2[VectorClock::Component::ConfigTime].asTimestamp(), Timestamp(0, 0));
+
+    vc->gossipIn(nullptr,
+                 BSON("$clusterTime"
+                      << BSON("clusterTime" << Timestamp(3, 3) << "signature" << dummySignature)
+                      << "$configTime" << Timestamp(3, 3)),
+                 false);
+
+    auto afterTime3 = vc->getTime();
+    ASSERT_EQ(afterTime3[VectorClock::Component::ClusterTime].asTimestamp(), Timestamp(3, 3));
+    ASSERT_EQ(afterTime3[VectorClock::Component::ConfigTime].asTimestamp(), Timestamp(0, 0));
 }
 
 }  // namespace
diff --git a/src/mongo/db/vector_clock_mutable.cpp b/src/mongo/db/vector_clock_mutable.cpp
index 6e18fbd028a..5bf3fda8d9e 100644
--- a/src/mongo/db/vector_clock_mutable.cpp
+++ b/src/mongo/db/vector_clock_mutable.cpp
@@ -29,8 +29,6 @@
 
 #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kDefault
 
-#include <limits>
-
 #include "mongo/platform/basic.h"
 
 #include "mongo/db/vector_clock_mutable.h"
@@ -65,13 +63,8 @@ void VectorClockMutable::registerVectorClockOnServiceContext(
     clock = std::move(vectorClockMutable);
 }
 
-bool VectorClockMutable::_lessThanOrEqualToMaxPossibleTime(LogicalTime time, uint64_t nTicks) {
-    return time.asTimestamp().getSecs() <= std::numeric_limits<uint32_t>::max() &&
-        time.asTimestamp().getInc() <= (std::numeric_limits<uint32_t>::max() - nTicks);
-}
-
 LogicalTime VectorClockMutable::_advanceComponentTimeByTicks(Component component, uint64_t nTicks) {
-    invariant(nTicks > 0 && nTicks <= std::numeric_limits<uint32_t>::max());
+    invariant(nTicks > 0 && nTicks <= kMaxValue);
 
     stdx::lock_guard<Latch> lock(_mutex);
 
@@ -90,10 +83,9 @@ LogicalTime VectorClockMutable::_advanceComponentTimeByTicks(Component component
     // in order to preserve compatibility with potentially signed or unsigned integral Timestamp
     // increment types. It is also unlikely to tick a clock by more than 2^31 in the span of one
     // second.
-    else if (time.asTimestamp().getInc() > (std::numeric_limits<uint32_t>::max() - nTicks)) {
+    else if (time.asTimestamp().getInc() > (kMaxValue - nTicks)) {
 
-        // TODO SERVER-47914: update this log id back to 20709.
-        LOGV2(4620000 /*20709*/,
+        LOGV2(20709,
               "Exceeded maximum allowable increment value within one second. Moving time forward "
               "to the next second.",
               "vectorClockComponent"_attr = _componentName(component));
@@ -102,10 +94,9 @@ LogicalTime VectorClockMutable::_advanceComponentTimeByTicks(Component component
         time = LogicalTime(Timestamp(time.asTimestamp().getSecs() + 1, 0));
     }
 
-    // TODO SERVER-47914: update this uassert id back to 40482.
-    uassert(4620001 /*40482*/,
+    uassert(40482,
             str::stream() << _componentName(component)
-                          << " cannot be advanced beyond the maximum cluster time value",
+                          << " cannot be advanced beyond the maximum logical time value",
             _lessThanOrEqualToMaxPossibleTime(time, nTicks));
 
     // Save the next time.
@@ -122,6 +113,14 @@ LogicalTime VectorClockMutable::_advanceComponentTimeByTicks(Component component
 
 void VectorClockMutable::_advanceComponentTimeTo(Component component, LogicalTime&& newTime) {
     stdx::lock_guard<Latch> lock(_mutex);
+
+    // Rate limit checks are skipped here so a server with no activity for longer than
+    // maxAcceptableLogicalClockDriftSecs seconds can still have its cluster time initialized.
+    uassert(40483,
+            str::stream() << _componentName(component)
+                          << " cannot be advanced beyond the maximum logical time value",
+            _lessThanOrEqualToMaxPossibleTime(newTime, 0));
+
     if (newTime > _vectorTime[component]) {
         _vectorTime[component] = std::move(newTime);
     }
diff --git a/src/mongo/db/vector_clock_mutable.h b/src/mongo/db/vector_clock_mutable.h
index 7e99aa32f2a..f8239b1fc4b 100644
--- a/src/mongo/db/vector_clock_mutable.h
+++ b/src/mongo/db/vector_clock_mutable.h
@@ -44,22 +44,41 @@ public:
     // clock implementation.
     static VectorClockMutable* get(ServiceContext* service);
     static VectorClockMutable* get(OperationContext* ctx);
-
     static void registerVectorClockOnServiceContext(ServiceContext* service,
                                                     VectorClockMutable* vectorClockMutable);
 
-    // Ticking
+    /**
+     * Returns the next time value for this Component, and provides a guarantee that any future call
+     * to tick() (for this Component) will return a value at least 'nTicks' ticks in the future from
+     * the current time.
+     */
     virtual LogicalTime tick(Component component, uint64_t nTicks) = 0;
+
+    /**
+     * Authoritatively ticks the current time of the Component to newTime.
+     *
+     * For ClusterTime, this should only be used for initializing from a trusted source, eg. from an
+     * oplog timestamp.
+     */
     virtual void tickTo(Component component, LogicalTime newTime) = 0;
 
 protected:
-    static bool _lessThanOrEqualToMaxPossibleTime(LogicalTime time, uint64_t nTicks);
-
     VectorClockMutable();
     virtual ~VectorClockMutable();
 
-    // Internal Ticking API
+    /**
+     * Called by sub-classes in order to actually tick a Component time, once they have determined
+     * that doing so is permissible.
+     *
+     * Returns as per tick(), ie. returns the next time value, and guarantees that future calls will
+     * return at least nTicks later.
+     */
     LogicalTime _advanceComponentTimeByTicks(Component component, uint64_t nTicks);
+
+    /**
+     * Called by sub-classes in order to actually tickTo a Component time, once they have determined
+     * that doing so is permissible.
+     */
     void _advanceComponentTimeTo(Component component, LogicalTime&& newTime);
 };
 
diff --git a/src/mongo/db/vector_clock_trivial.cpp b/src/mongo/db/vector_clock_trivial.cpp
index 88f00a512ea..96cde5788de 100644
--- a/src/mongo/db/vector_clock_trivial.cpp
+++ b/src/mongo/db/vector_clock_trivial.cpp
@@ -49,10 +49,19 @@ public:
     void tickTo(Component component, LogicalTime newTime) override;
 
 protected:
-    void _gossipOutInternal(BSONObjBuilder* out) const override;
-    void _gossipOutExternal(BSONObjBuilder* out) const override;
-    LogicalTimeArray _gossipInInternal(const BSONObj& in) override;
-    LogicalTimeArray _gossipInExternal(const BSONObj& in) override;
+    bool _gossipOutInternal(OperationContext* opCtx,
+                            BSONObjBuilder* out,
+                            const LogicalTimeArray& time) const override;
+    bool _gossipOutExternal(OperationContext* opCtx,
+                            BSONObjBuilder* out,
+                            const LogicalTimeArray& time) const override;
+    LogicalTimeArray _gossipInInternal(OperationContext* opCtx,
+                                       const BSONObj& in,
+                                       bool couldBeUnauthenticated) override;
+    LogicalTimeArray _gossipInExternal(OperationContext* opCtx,
+                                       const BSONObj& in,
+                                       bool couldBeUnauthenticated) override;
+    bool _permitRefreshDuringGossipOut() const override;
 };
 
 const auto vectorClockTrivialDecoration = ServiceContext::declareDecoration<VectorClockTrivial>();
@@ -70,24 +79,38 @@ VectorClockTrivial::VectorClockTrivial() = default;
 
 VectorClockTrivial::~VectorClockTrivial() = default;
 
-void VectorClockTrivial::_gossipOutInternal(BSONObjBuilder* out) const {
+bool VectorClockTrivial::_gossipOutInternal(OperationContext* opCtx,
+                                            BSONObjBuilder* out,
+                                            const LogicalTimeArray& time) const {
     // Clocks are not gossipped in trivial (non-distributed) environments.
+    return false;
 }
 
-void VectorClockTrivial::_gossipOutExternal(BSONObjBuilder* out) const {
+bool VectorClockTrivial::_gossipOutExternal(OperationContext* opCtx,
+                                            BSONObjBuilder* out,
+                                            const LogicalTimeArray& time) const {
     // Clocks are not gossipped in trivial (non-distributed) environments.
+    return false;
 }
 
-VectorClock::LogicalTimeArray VectorClockTrivial::_gossipInInternal(const BSONObj& in) {
+VectorClock::LogicalTimeArray VectorClockTrivial::_gossipInInternal(OperationContext* opCtx,
+                                                                    const BSONObj& in,
+                                                                    bool couldBeUnauthenticated) {
     // Clocks are not gossipped in trivial (non-distributed) environments.
     return {};
 }
 
-VectorClock::LogicalTimeArray VectorClockTrivial::_gossipInExternal(const BSONObj& in) {
+VectorClock::LogicalTimeArray VectorClockTrivial::_gossipInExternal(OperationContext* opCtx,
+                                                                    const BSONObj& in,
+                                                                    bool couldBeUnauthenticated) {
     // Clocks are not gossipped in trivial (non-distributed) environments.
     return {};
 }
 
+bool VectorClockTrivial::_permitRefreshDuringGossipOut() const {
+    return false;
+}
+
 LogicalTime VectorClockTrivial::tick(Component component, uint64_t nTicks) {
     return _advanceComponentTimeByTicks(component, nTicks);
 }
diff --git a/src/mongo/dbtests/SConscript b/src/mongo/dbtests/SConscript
index 3337f065559..02939bd9933 100644
--- a/src/mongo/dbtests/SConscript
+++ b/src/mongo/dbtests/SConscript
@@ -143,8 +143,6 @@ if not has_option('noshell') and usemozjs:
             "$BUILD_DIR/mongo/db/exec/document_value/document_value_test_util",
             "$BUILD_DIR/mongo/db/index/index_access_methods",
             "$BUILD_DIR/mongo/db/index/index_build_interceptor",
-            "$BUILD_DIR/mongo/db/logical_clock",
-            "$BUILD_DIR/mongo/db/logical_session_id_helpers",
             "$BUILD_DIR/mongo/db/logical_time_metadata_hook",
             "$BUILD_DIR/mongo/db/mongohasher",
             "$BUILD_DIR/mongo/db/query/collation/collator_interface_mock",
@@ -173,15 +171,8 @@ if not has_option('noshell') and usemozjs:
             "$BUILD_DIR/mongo/db/storage/wiredtiger/storage_wiredtiger" if wiredtiger else [],
             "$BUILD_DIR/mongo/db/traffic_reader",
             "$BUILD_DIR/mongo/db/transaction",
-            "$BUILD_DIR/mongo/executor/network_interface_factory",
-            "$BUILD_DIR/mongo/executor/network_interface_thread_pool",
-            "$BUILD_DIR/mongo/executor/thread_pool_task_executor",
-            "$BUILD_DIR/mongo/rpc/protocol",
-            "$BUILD_DIR/mongo/scripting/scripting",
-            "$BUILD_DIR/mongo/shell/benchrun",
-            "$BUILD_DIR/mongo/shell/mongojs",
+            "$BUILD_DIR/mongo/db/vector_clock",
             "$BUILD_DIR/mongo/shell/shell_utils",
-            "$BUILD_DIR/mongo/transport/message_compressor",
             "$BUILD_DIR/mongo/transport/message_compressor_options_server",
             "$BUILD_DIR/mongo/transport/transport_layer_manager",
             "$BUILD_DIR/mongo/util/clock_source_mock",
diff --git a/src/mongo/dbtests/storage_timestamp_tests.cpp b/src/mongo/dbtests/storage_timestamp_tests.cpp
index 3181b1cb0c0..19603ffdbad 100644
--- a/src/mongo/dbtests/storage_timestamp_tests.cpp
+++ b/src/mongo/dbtests/storage_timestamp_tests.cpp
@@ -52,7 +52,6 @@
 #include "mongo/db/index/index_build_interceptor.h"
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/index_builds_coordinator.h"
-#include "mongo/db/logical_clock.h"
 #include "mongo/db/multi_key_path_tracker.h"
 #include "mongo/db/op_observer_registry.h"
 #include "mongo/db/repl/apply_ops.h"
@@ -83,6 +82,7 @@
 #include "mongo/db/storage/storage_engine_impl.h"
 #include "mongo/db/transaction_participant.h"
 #include "mongo/db/transaction_participant_gen.h"
+#include "mongo/db/vector_clock_mutable.h"
 #include "mongo/dbtests/dbtests.h"
 #include "mongo/logv2/log.h"
 #include "mongo/stdx/future.h"
@@ -159,12 +159,13 @@ class StorageTimestampTest {
 public:
     ServiceContext::UniqueOperationContext _opCtxRaii = cc().makeOperationContext();
     OperationContext* _opCtx = _opCtxRaii.get();
-    LogicalClock* _clock = LogicalClock::get(_opCtx);
+    VectorClockMutable* _clock = VectorClockMutable::get(_opCtx);
 
     // Set up Timestamps in the past, present, and future.
-    const LogicalTime pastLt = _clock->reserveTicks(1);
+    static constexpr auto ClusterTime = VectorClock::Component::ClusterTime;
+    const LogicalTime pastLt = _clock->tick(ClusterTime, 1);
     const Timestamp pastTs = pastLt.asTimestamp();
-    const LogicalTime presentLt = _clock->reserveTicks(1);
+    const LogicalTime presentLt = _clock->tick(ClusterTime, 1);
     const Timestamp presentTs = presentLt.asTimestamp();
     const LogicalTime futureLt = presentLt.addTicks(1);
     const Timestamp futureTs = futureLt.asTimestamp();
@@ -211,7 +212,7 @@ public:
         repl::setOplogCollectionName(getGlobalServiceContext());
         repl::createOplog(_opCtx);
 
-        ASSERT_OK(_clock->advanceClusterTime(LogicalTime(Timestamp(1, 0))));
+        _clock->tickTo(ClusterTime, LogicalTime(Timestamp(1, 0)));
 
         ASSERT_EQUALS(presentTs, pastLt.addTicks(1).asTimestamp());
         setReplCoordAppliedOpTime(repl::OpTime(presentTs, presentTerm));
@@ -716,7 +717,7 @@ public:
         AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_IX);
 
         const std::int32_t docsToInsert = 10;
-        const LogicalTime firstInsertTime = _clock->reserveTicks(docsToInsert);
+        const LogicalTime firstInsertTime = _clock->tick(ClusterTime, docsToInsert);
         for (std::int32_t idx = 0; idx < docsToInsert; ++idx) {
             BSONObjBuilder result;
             ASSERT_OK(applyOps(
@@ -765,7 +766,7 @@ public:
         AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_IX);
 
         const std::int32_t docsToInsert = 10;
-        const LogicalTime firstInsertTime = _clock->reserveTicks(docsToInsert);
+        const LogicalTime firstInsertTime = _clock->tick(ClusterTime, docsToInsert);
 
         BSONObjBuilder oplogCommonBuilder;
         oplogCommonBuilder << "v" << 2 << "op"
@@ -824,7 +825,7 @@ public:
 
         // Insert some documents.
         const std::int32_t docsToInsert = 10;
-        const LogicalTime firstInsertTime = _clock->reserveTicks(docsToInsert);
+        const LogicalTime firstInsertTime = _clock->tick(ClusterTime, docsToInsert);
         const LogicalTime lastInsertTime = firstInsertTime.addTicks(docsToInsert - 1);
         WriteUnitOfWork wunit(_opCtx);
         for (std::int32_t num = 0; num < docsToInsert; ++num) {
@@ -837,7 +838,7 @@ public:
         ASSERT_EQ(docsToInsert, itCount(autoColl.getCollection()));
 
         // Delete all documents one at a time.
-        const LogicalTime startDeleteTime = _clock->reserveTicks(docsToInsert);
+        const LogicalTime startDeleteTime = _clock->tick(ClusterTime, docsToInsert);
         for (std::int32_t num = 0; num < docsToInsert; ++num) {
             ASSERT_OK(doNonAtomicApplyOps(
                           nss.db().toString(),
@@ -871,7 +872,7 @@ public:
         AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_IX);
 
         // Insert one document that will go through a series of updates.
-        const LogicalTime insertTime = _clock->reserveTicks(1);
+        const LogicalTime insertTime = _clock->tick(ClusterTime, 1);
         WriteUnitOfWork wunit(_opCtx);
         insertDocument(autoColl.getCollection(),
                        InsertStatement(BSON("_id" << 0), insertTime.asTimestamp(), 0LL));
@@ -896,7 +897,7 @@ public:
                                     << "theOtherSet")),
              BSON("_id" << 0 << "theMap" << BSON("val" << 1) << "theOtherSet" << BSONArray())}};
 
-        const LogicalTime firstUpdateTime = _clock->reserveTicks(updates.size());
+        const LogicalTime firstUpdateTime = _clock->tick(ClusterTime, updates.size());
         for (std::size_t idx = 0; idx < updates.size(); ++idx) {
             ASSERT_OK(doNonAtomicApplyOps(
                           nss.db().toString(),
@@ -933,7 +934,7 @@ public:
 
         AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_IX);
 
-        const LogicalTime insertTime = _clock->reserveTicks(2);
+        const LogicalTime insertTime = _clock->tick(ClusterTime, 2);
 
         // This applyOps runs into an insert of `{_id: 0, field: 0}` followed by a second insert
         // on the same collection with `{_id: 0}`. It's expected for this second insert to be
@@ -985,7 +986,7 @@ public:
         AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_IX);
 
         // Reserve a timestamp before the inserts should happen.
-        const LogicalTime preInsertTimestamp = _clock->reserveTicks(1);
+        const LogicalTime preInsertTimestamp = _clock->tick(ClusterTime, 1);
         auto swResult =
             doAtomicApplyOps(nss.db().toString(),
                              {BSON("op"
@@ -1037,7 +1038,7 @@ public:
 
         AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_IX);
 
-        const LogicalTime preInsertTimestamp = _clock->reserveTicks(1);
+        const LogicalTime preInsertTimestamp = _clock->tick(ClusterTime, 1);
         auto swResult =
             doAtomicApplyOps(nss.db().toString(),
                              {BSON("op"
@@ -1293,10 +1294,10 @@ public:
 
         _coordinatorMock->alwaysAllowWrites(false);
 
-        const LogicalTime pastTime = _clock->reserveTicks(1);
-        const LogicalTime insertTime0 = _clock->reserveTicks(1);
-        const LogicalTime insertTime1 = _clock->reserveTicks(1);
-        const LogicalTime insertTime2 = _clock->reserveTicks(1);
+        const LogicalTime pastTime = _clock->tick(ClusterTime, 1);
+        const LogicalTime insertTime0 = _clock->tick(ClusterTime, 1);
+        const LogicalTime insertTime1 = _clock->tick(ClusterTime, 1);
+        const LogicalTime insertTime2 = _clock->tick(ClusterTime, 1);
 
         BSONObj doc0 = BSON("_id" << 0 << "a" << 3);
         BSONObj doc1 = BSON("_id" << 1 << "a" << BSON_ARRAY(1 << 2));
@@ -1364,11 +1365,11 @@ public:
         _coordinatorMock->alwaysAllowWrites(false);
         ASSERT_OK(_coordinatorMock->setFollowerMode({repl::MemberState::MS::RS_STARTUP2}));
 
-        const LogicalTime pastTime = _clock->reserveTicks(1);
-        const LogicalTime insertTime0 = _clock->reserveTicks(1);
-        const LogicalTime indexBuildTime = _clock->reserveTicks(1);
-        const LogicalTime insertTime1 = _clock->reserveTicks(1);
-        const LogicalTime insertTime2 = _clock->reserveTicks(1);
+        const LogicalTime pastTime = _clock->tick(ClusterTime, 1);
+        const LogicalTime insertTime0 = _clock->tick(ClusterTime, 1);
+        const LogicalTime indexBuildTime = _clock->tick(ClusterTime, 1);
+        const LogicalTime insertTime1 = _clock->tick(ClusterTime, 1);
+        const LogicalTime insertTime2 = _clock->tick(ClusterTime, 1);
 
         BSONObj doc0 = BSON("_id" << 0 << "a" << 3);
         BSONObj doc1 = BSON("_id" << 1 << "a" << BSON_ARRAY(1 << 2));
@@ -1455,7 +1456,7 @@ public:
                                      << static_cast<int>(kIndexVersion));
         ASSERT_OK(dbtests::createIndexFromSpec(_opCtx, nss.ns(), indexSpec));
 
-        const LogicalTime pastTime = _clock->reserveTicks(1);
+        const LogicalTime pastTime = _clock->tick(ClusterTime, 1);
         const LogicalTime insertTime = pastTime.addTicks(1);
 
         BSONObj doc = BSON("_id" << 1 << "a" << BSON_ARRAY(1 << 2));
@@ -1484,7 +1485,7 @@ public:
                                      << static_cast<int>(kIndexVersion));
         ASSERT_OK(dbtests::createIndexFromSpec(_opCtx, nss.ns(), indexSpec));
 
-        const LogicalTime pastTime = _clock->reserveTicks(1);
+        const LogicalTime pastTime = _clock->tick(ClusterTime, 1);
         const LogicalTime insertTime = pastTime.addTicks(1);
 
         BSONObj doc = BSON("_id" << 1 << "a" << BSON_ARRAY(1 << 2));
@@ -1521,14 +1522,14 @@ public:
             ASSERT_OK(dbtests::createIndexFromSpec(_opCtx, nss.ns(), indexSpec));
         }
 
-        auto presentTs = _clock->getClusterTime().asTimestamp();
+        auto presentTs = _clock->getTime()[ClusterTime].asTimestamp();
 
         // This test does not run a real ReplicationCoordinator, so must advance the snapshot
         // manager manually.
         auto storageEngine = cc().getServiceContext()->getStorageEngine();
         storageEngine->getSnapshotManager()->setLastApplied(presentTs);
 
-        const auto beforeTxnTime = _clock->reserveTicks(1);
+        const auto beforeTxnTime = _clock->tick(ClusterTime, 1);
         auto beforeTxnTs = beforeTxnTime.asTimestamp();
         const auto multikeyNoopTime = beforeTxnTime.addTicks(1);
         auto multikeyNoopTs = multikeyNoopTime.asTimestamp();
@@ -1748,7 +1749,7 @@ public:
         auto durableCatalog = storageEngine->getCatalog();
 
         // Declare the database to be in a "synced" state, i.e: in steady-state replication.
-        Timestamp syncTime = _clock->reserveTicks(1).asTimestamp();
+        Timestamp syncTime = _clock->tick(ClusterTime, 1).asTimestamp();
         invariant(!syncTime.isNull());
         storageEngine->setInitialDataTimestamp(syncTime);
 
@@ -1777,7 +1778,7 @@ public:
             // side-effect of not timestamping the collection creation.
             repl::UnreplicatedWritesBlock notReplicated(_opCtx);
             if (nss.isReplicated()) {
-                TimestampBlock tsBlock(_opCtx, _clock->reserveTicks(1).asTimestamp());
+                TimestampBlock tsBlock(_opCtx, _clock->tick(ClusterTime, 1).asTimestamp());
                 reset(nss);
             } else {
                 reset(nss);
@@ -1801,7 +1802,7 @@ public:
 
         // Reserve a tick, this represents a time after the rename in which the `kvDropDatabase`
         // ident for `kvDropDatabase` still exists.
-        const Timestamp postRenameTime = _clock->reserveTicks(1).asTimestamp();
+        const Timestamp postRenameTime = _clock->tick(ClusterTime, 1).asTimestamp();
 
         // If the storage engine is managing drops internally, the ident should not be visible after
         // a drop.
@@ -1812,7 +1813,7 @@ public:
             assertIdentsExistAtTimestamp(durableCatalog, collIdent, indexIdent, postRenameTime);
         }
 
-        const Timestamp dropTime = _clock->reserveTicks(1).asTimestamp();
+        const Timestamp dropTime = _clock->tick(ClusterTime, 1).asTimestamp();
         if (SimulatePrimary) {
             ASSERT_OK(dropDatabaseForApplyOps(_opCtx, nss.db().toString()));
         } else {
@@ -1869,7 +1870,7 @@ public:
         AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_X);
         RecordId catalogId = autoColl.getCollection()->getCatalogId();
 
-        const LogicalTime insertTimestamp = _clock->reserveTicks(1);
+        const LogicalTime insertTimestamp = _clock->tick(ClusterTime, 1);
         {
             WriteUnitOfWork wuow(_opCtx);
             insertDocument(autoColl.getCollection(),
@@ -1890,7 +1891,7 @@ public:
             indexer.abortIndexBuild(
                 _opCtx, autoColl.getCollection(), MultiIndexBlock::kNoopOnCleanUpFn);
         });
-        const LogicalTime beforeIndexBuild = _clock->reserveTicks(2);
+        const LogicalTime beforeIndexBuild = _clock->tick(ClusterTime, 2);
         BSONObj indexInfoObj;
         {
             // Primaries do not have a wrapping `TimestampBlock`; secondaries do.
@@ -1916,7 +1917,7 @@ public:
             indexInfoObj = std::move(swIndexInfoObj.getValue()[0]);
         }
 
-        const LogicalTime afterIndexInit = _clock->reserveTicks(2);
+        const LogicalTime afterIndexInit = _clock->tick(ClusterTime, 2);
 
         // Inserting all the documents has the side-effect of setting internal state on the index
         // builder that the index is multikey.
@@ -1938,7 +1939,7 @@ public:
                             _opCtx, nss, autoColl.getCollection()->uuid(), indexSpec, false);
                     } else {
                         ASSERT_OK(_opCtx->recoveryUnit()->setTimestamp(
-                            _clock->getClusterTime().asTimestamp()));
+                            _clock->getTime()[ClusterTime].asTimestamp()));
                     }
                 },
                 MultiIndexBlock::kNoopOnCommitFn));
@@ -1946,7 +1947,7 @@ public:
         }
         abortOnExit.dismiss();
 
-        const Timestamp afterIndexBuild = _clock->reserveTicks(1).asTimestamp();
+        const Timestamp afterIndexBuild = _clock->tick(ClusterTime, 1).asTimestamp();
 
         const std::string indexIdent =
             getNewIndexIdentAtTime(durableCatalog, origIdents, Timestamp::min());
@@ -1981,6 +1982,160 @@ public:
     }
 };
 
+template <bool SimulatePrimary>
+class TimestampIndexBuildDrain : public StorageTimestampTest {
+public:
+    void run() {
+        const bool SimulateSecondary = !SimulatePrimary;
+        if (SimulateSecondary) {
+            // The MemberState is inspected during index builds to use a "ghost" write to timestamp
+            // index completion.
+            ASSERT_OK(_coordinatorMock->setFollowerMode({repl::MemberState::MS::RS_SECONDARY}));
+        }
+
+        NamespaceString nss("unittests.timestampIndexBuildDrain");
+        reset(nss);
+
+        AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_X);
+
+        // Build an index on `{a: 1}`.
+        MultiIndexBlock indexer;
+        auto abortOnExit = makeGuard([&] {
+            indexer.abortIndexBuild(
+                _opCtx, autoColl.getCollection(), MultiIndexBlock::kNoopOnCleanUpFn);
+        });
+        const LogicalTime beforeIndexBuild = _clock->tick(ClusterTime, 2);
+        BSONObj indexInfoObj;
+        {
+            // Primaries do not have a wrapping `TimestampBlock`; secondaries do.
+            const Timestamp commitTimestamp =
+                SimulatePrimary ? Timestamp::min() : beforeIndexBuild.addTicks(1).asTimestamp();
+            TimestampBlock tsBlock(_opCtx, commitTimestamp);
+
+            // Secondaries will also be in an `UnreplicatedWritesBlock` that prevents the `logOp`
+            // from making creating an entry.
+            boost::optional<repl::UnreplicatedWritesBlock> unreplicated;
+            if (SimulateSecondary) {
+                unreplicated.emplace(_opCtx);
+            }
+
+            auto swIndexInfoObj = indexer.init(
+                _opCtx,
+                autoColl.getCollection(),
+                {BSON("v" << 2 << "unique" << true << "name"
+                          << "a_1"
+                          << "ns" << nss.ns() << "key" << BSON("a" << 1))},
+                MultiIndexBlock::makeTimestampedIndexOnInitFn(_opCtx, autoColl.getCollection()));
+            ASSERT_OK(swIndexInfoObj.getStatus());
+            indexInfoObj = std::move(swIndexInfoObj.getValue()[0]);
+        }
+
+        const LogicalTime afterIndexInit = _clock->tick(ClusterTime, 1);
+
+        // Insert a document that will be intercepted and need to be drained. This timestamp will
+        // become the lastApplied time.
+        const LogicalTime firstInsert = _clock->tick(ClusterTime, 1);
+        {
+            WriteUnitOfWork wuow(_opCtx);
+            insertDocument(autoColl.getCollection(),
+                           InsertStatement(BSON("_id" << 0 << "a" << 1),
+                                           firstInsert.asTimestamp(),
+                                           presentTerm));
+            wuow.commit();
+            ASSERT_EQ(1, itCount(autoColl.getCollection()));
+        }
+
+        // Index build drain will timestamp writes from the side table into the index with the
+        // lastApplied timestamp. This is because these writes are not associated with any specific
+        // oplog entry.
+        ASSERT_EQ(repl::ReplicationCoordinator::get(getGlobalServiceContext())
+                      ->getMyLastAppliedOpTime()
+                      .getTimestamp(),
+                  firstInsert.asTimestamp());
+
+        ASSERT_OK(indexer.drainBackgroundWrites(_opCtx,
+                                                RecoveryUnit::ReadSource::kUnset,
+                                                IndexBuildInterceptor::DrainYieldPolicy::kNoYield));
+
+        auto indexCatalog = autoColl.getCollection()->getIndexCatalog();
+        const IndexCatalogEntry* buildingIndex = indexCatalog->getEntry(
+            indexCatalog->findIndexByName(_opCtx, "a_1", /* includeUnfinished */ true));
+        ASSERT(buildingIndex);
+
+        {
+            // Before the drain, there are no writes to apply.
+            OneOffRead oor(_opCtx, afterIndexInit.asTimestamp());
+            ASSERT_TRUE(buildingIndex->indexBuildInterceptor()->areAllWritesApplied(_opCtx));
+        }
+
+        // Note: In this case, we can't observe a state where all writes are not applied, because
+        // the index build drain effectively rewrites history by retroactively committing the drain
+        // at the same time as the first insert, meaning there is no point-in-time with undrained
+        // writes. This is fine, as long as the drain does not commit at a time before this insert.
+
+        {
+            // At time of the first insert, all writes are applied.
+            OneOffRead oor(_opCtx, firstInsert.asTimestamp());
+            ASSERT_TRUE(buildingIndex->indexBuildInterceptor()->areAllWritesApplied(_opCtx));
+        }
+
+        // Insert a second document that will be intercepted and need to be drained.
+        const LogicalTime secondInsert = _clock->tick(ClusterTime, 1);
+        {
+            WriteUnitOfWork wuow(_opCtx);
+            insertDocument(autoColl.getCollection(),
+                           InsertStatement(BSON("_id" << 1 << "a" << 2),
+                                           secondInsert.asTimestamp(),
+                                           presentTerm));
+            wuow.commit();
+            ASSERT_EQ(2, itCount(autoColl.getCollection()));
+        }
+
+        // Advance the lastApplied optime to observe a point before the drain where there are
+        // un-drained writes.
+        const LogicalTime afterSecondInsert = _clock->tick(ClusterTime, 1);
+        setReplCoordAppliedOpTime(repl::OpTime(afterSecondInsert.asTimestamp(), presentTerm));
+
+        ASSERT_OK(indexer.drainBackgroundWrites(_opCtx,
+                                                RecoveryUnit::ReadSource::kUnset,
+                                                IndexBuildInterceptor::DrainYieldPolicy::kNoYield));
+
+        {
+            // At time of the second insert, there are un-drained writes.
+            OneOffRead oor(_opCtx, secondInsert.asTimestamp());
+            ASSERT_FALSE(buildingIndex->indexBuildInterceptor()->areAllWritesApplied(_opCtx));
+        }
+
+        {
+            // After the second insert, also the lastApplied time, all writes are applied.
+            OneOffRead oor(_opCtx, afterSecondInsert.asTimestamp());
+            ASSERT_TRUE(buildingIndex->indexBuildInterceptor()->areAllWritesApplied(_opCtx));
+        }
+
+        ASSERT_OK(indexer.checkConstraints(_opCtx));
+
+        {
+            WriteUnitOfWork wuow(_opCtx);
+            ASSERT_OK(indexer.commit(
+                _opCtx,
+                autoColl.getCollection(),
+                [&](const BSONObj& indexSpec) {
+                    if (SimulatePrimary) {
+                        // The timestamping responsibility for each index is placed on the caller.
+                        _opCtx->getServiceContext()->getOpObserver()->onCreateIndex(
+                            _opCtx, nss, autoColl.getCollection()->uuid(), indexSpec, false);
+                    } else {
+                        ASSERT_OK(_opCtx->recoveryUnit()->setTimestamp(
+                            _clock->getTime()[ClusterTime].asTimestamp()));
+                    }
+                },
+                MultiIndexBlock::kNoopOnCommitFn));
+            wuow.commit();
+        }
+        abortOnExit.dismiss();
+    }
+};
+
 class TimestampMultiIndexBuilds : public StorageTimestampTest {
 public:
     void run() {
@@ -2010,7 +2165,7 @@ public:
         {
             AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_X);
 
-            const LogicalTime insertTimestamp = _clock->reserveTicks(1);
+            const LogicalTime insertTimestamp = _clock->tick(ClusterTime, 1);
 
             WriteUnitOfWork wuow(_opCtx);
             insertDocument(autoColl.getCollection(),
@@ -2117,7 +2272,7 @@ public:
         {
             AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_X);
 
-            const LogicalTime insertTimestamp = _clock->reserveTicks(1);
+            const LogicalTime insertTimestamp = _clock->tick(ClusterTime, 1);
 
             WriteUnitOfWork wuow(_opCtx);
             insertDocument(autoColl.getCollection(),
@@ -2235,8 +2390,8 @@ public:
         {
             AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_X);
 
-            auto insertTimestamp1 = _clock->reserveTicks(1);
-            auto insertTimestamp2 = _clock->reserveTicks(1);
+            auto insertTimestamp1 = _clock->tick(ClusterTime, 1);
+            auto insertTimestamp2 = _clock->tick(ClusterTime, 1);
 
             // Insert two documents with the same value for field 'a' so that
             // we will fail to create a unique index.
@@ -2334,7 +2489,7 @@ public:
 
         AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_X);
 
-        const LogicalTime insertTimestamp = _clock->reserveTicks(1);
+        const LogicalTime insertTimestamp = _clock->tick(ClusterTime, 1);
         {
             WriteUnitOfWork wuow(_opCtx);
             insertDocument(autoColl.getCollection(),
@@ -2346,7 +2501,7 @@ public:
         }
 
 
-        const Timestamp beforeIndexBuild = _clock->reserveTicks(1).asTimestamp();
+        const Timestamp beforeIndexBuild = _clock->tick(ClusterTime, 1).asTimestamp();
 
         // Save the pre-state idents so we can capture the specific ident related to index
         // creation.
@@ -2359,7 +2514,7 @@ public:
             createIndex(autoColl.getCollection(), str::stream() << key << "_1", BSON(key << 1));
 
             // Timestamps at the completion of each index build.
-            afterCreateTimestamps.push_back(_clock->reserveTicks(1).asTimestamp());
+            afterCreateTimestamps.push_back(_clock->tick(ClusterTime, 1).asTimestamp());
 
             // Add the new ident to the vector and reset the current idents.
             indexIdents.push_back(
@@ -2375,7 +2530,7 @@ public:
                 durableCatalog, "", indexIdents[i], afterCreateTimestamps[i]);
         }
 
-        const LogicalTime beforeDropTs = _clock->getClusterTime();
+        const LogicalTime beforeDropTs = _clock->getTime()[ClusterTime];
 
         // Drop all of the indexes.
         BSONObjBuilder result;
@@ -2490,7 +2645,7 @@ public:
             BSON("_id" << 2 << "a" << BSON_ARRAY(4 << 5) << "b" << BSON_ARRAY(4 << 5));
 
         // NOTE: This test does not test any timestamp reads.
-        const LogicalTime insert1 = _clock->reserveTicks(1);
+        const LogicalTime insert1 = _clock->tick(ClusterTime, 1);
         {
             LOGV2(22505, "inserting {badDoc1}", "badDoc1"_attr = badDoc1);
             WriteUnitOfWork wuow(_opCtx);
@@ -2499,7 +2654,7 @@ public:
             wuow.commit();
         }
 
-        const LogicalTime insert2 = _clock->reserveTicks(1);
+        const LogicalTime insert2 = _clock->tick(ClusterTime, 1);
         {
             LOGV2(22506, "inserting {badDoc2}", "badDoc2"_attr = badDoc2);
             WriteUnitOfWork wuow(_opCtx);
@@ -2518,7 +2673,7 @@ public:
         const auto buildUUID = UUID::gen();
         indexer.setTwoPhaseBuildUUID(buildUUID);
 
-        const LogicalTime indexInit = _clock->reserveTicks(3);
+        const LogicalTime indexInit = _clock->tick(ClusterTime, 3);
 
         // First, simulate being a secondary. Indexing errors are ignored.
         {
@@ -2717,7 +2872,7 @@ public:
         NamespaceString nss(dbName, "oplogApplicationOnPrimary");
         BSONObj doc = BSON("_id" << 1 << "field" << 1);
 
-        const LogicalTime setupStart = _clock->reserveTicks(1);
+        const LogicalTime setupStart = _clock->tick(ClusterTime, 1);
 
         UUID collUUID = UUID::gen();
         {
@@ -2738,7 +2893,7 @@ public:
             auto coll = autoColl.getCollection();
             ASSERT(coll);
 
-            const auto presentTs = _clock->getClusterTime().asTimestamp();
+            const auto presentTs = _clock->getTime()[ClusterTime].asTimestamp();
             assertDocumentAtTimestamp(coll, presentTs, doc);
         }
 
@@ -2747,7 +2902,7 @@ public:
         // the applyOps command no longer allows createIndexes (see SERVER-41554).
         _coordinatorMock->alwaysAllowWrites(false);
         {
-            const auto beforeBuildTime = _clock->reserveTicks(2);
+            const auto beforeBuildTime = _clock->tick(ClusterTime, 2);
             const auto startBuildTs = beforeBuildTime.addTicks(1).asTimestamp();
 
             // Grab the existing idents to identify the ident created by the index build.
@@ -2965,12 +3120,12 @@ public:
             ui = coll->uuid();
         }
 
-        presentTs = _clock->getClusterTime().asTimestamp();
+        presentTs = _clock->getTime()[ClusterTime].asTimestamp();
         // This test does not run a real ReplicationCoordinator, so must advance the snapshot
         // manager manually.
         auto storageEngine = cc().getServiceContext()->getStorageEngine();
         storageEngine->getSnapshotManager()->setLastApplied(presentTs);
-        const auto beforeTxnTime = _clock->reserveTicks(1);
+        const auto beforeTxnTime = _clock->tick(ClusterTime, 1);
         beforeTxnTs = beforeTxnTime.asTimestamp();
         commitEntryTs = beforeTxnTime.addTicks(1).asTimestamp();
 
@@ -3096,7 +3251,7 @@ private:
 class MultiOplogEntryTransaction : public MultiDocumentTransactionTest {
 public:
     MultiOplogEntryTransaction() : MultiDocumentTransactionTest("multiOplogEntryTransaction") {
-        const auto currentTime = _clock->getClusterTime();
+        const auto currentTime = _clock->getTime()[ClusterTime];
         firstOplogEntryTs = currentTime.addTicks(1).asTimestamp();
         commitEntryTs = currentTime.addTicks(2).asTimestamp();
     }
@@ -3200,7 +3355,7 @@ class CommitPreparedMultiOplogEntryTransaction : public MultiDocumentTransaction
 public:
     CommitPreparedMultiOplogEntryTransaction()
         : MultiDocumentTransactionTest("preparedMultiOplogEntryTransaction") {
-        const auto currentTime = _clock->getClusterTime();
+        const auto currentTime = _clock->getTime()[ClusterTime];
         firstOplogEntryTs = currentTime.addTicks(1).asTimestamp();
         prepareEntryTs = currentTime.addTicks(2).asTimestamp();
         commitEntryTs = currentTime.addTicks(3).asTimestamp();
@@ -3395,7 +3550,7 @@ class AbortPreparedMultiOplogEntryTransaction : public MultiDocumentTransactionT
 public:
     AbortPreparedMultiOplogEntryTransaction()
         : MultiDocumentTransactionTest("preparedMultiOplogEntryTransaction") {
-        const auto currentTime = _clock->getClusterTime();
+        const auto currentTime = _clock->getTime()[ClusterTime];
         prepareEntryTs = currentTime.addTicks(1).asTimestamp();
         abortEntryTs = currentTime.addTicks(2).asTimestamp();
     }
@@ -3504,7 +3659,7 @@ public:
         auto txnParticipant = TransactionParticipant::get(_opCtx);
         ASSERT(txnParticipant);
 
-        const auto currentTime = _clock->getClusterTime();
+        const auto currentTime = _clock->getTime()[ClusterTime];
         const auto prepareTs = currentTime.addTicks(1).asTimestamp();
         commitEntryTs = currentTime.addTicks(2).asTimestamp();
         LOGV2(22514, "Prepare TS: {prepareTs}", "prepareTs"_attr = prepareTs);
@@ -3605,7 +3760,7 @@ public:
         auto txnParticipant = TransactionParticipant::get(_opCtx);
         ASSERT(txnParticipant);
 
-        const auto currentTime = _clock->getClusterTime();
+        const auto currentTime = _clock->getTime()[ClusterTime];
         const auto prepareTs = currentTime.addTicks(1).asTimestamp();
         const auto abortEntryTs = currentTime.addTicks(2).asTimestamp();
         LOGV2(22515, "Prepare TS: {prepareTs}", "prepareTs"_attr = prepareTs);
diff --git a/src/mongo/rpc/SConscript b/src/mongo/rpc/SConscript
index d9814e5345b..997326b96b5 100644
--- a/src/mongo/rpc/SConscript
+++ b/src/mongo/rpc/SConscript
@@ -95,7 +95,6 @@ env.Library(
         'metadata.cpp',
         'metadata/config_server_metadata.cpp',
         'metadata/egress_metadata_hook_list.cpp',
-        'metadata/logical_time_metadata.cpp',
         'metadata/sharding_metadata.cpp',
         'metadata/repl_set_metadata.cpp',
         'metadata/oplog_query_metadata.cpp',
@@ -109,9 +108,9 @@ env.Library(
         '$BUILD_DIR/mongo/bson/util/bson_extract',
         '$BUILD_DIR/mongo/client/read_preference',
         '$BUILD_DIR/mongo/db/commands/test_commands_enabled',
-        '$BUILD_DIR/mongo/db/logical_time_validator',
         '$BUILD_DIR/mongo/db/repl/optime',
         '$BUILD_DIR/mongo/db/signed_logical_time',
+        '$BUILD_DIR/mongo/db/vector_clock',
     ],
 )
 
@@ -159,7 +158,6 @@ env.CppUnitTest(
         'metadata/client_metadata_test.cpp',
         'metadata/config_server_metadata_test.cpp',
         'metadata/egress_metadata_hook_list_test.cpp',
-        'metadata/logical_time_metadata_test.cpp',
         'metadata/oplog_query_metadata_test.cpp',
         'metadata/repl_set_metadata_test.cpp',
         'metadata/sharding_metadata_test.cpp',
diff --git a/src/mongo/rpc/metadata.cpp b/src/mongo/rpc/metadata.cpp
index 9a546b02572..2220e31755e 100644
--- a/src/mongo/rpc/metadata.cpp
+++ b/src/mongo/rpc/metadata.cpp
@@ -42,7 +42,6 @@
 #include "mongo/rpc/metadata/client_metadata_ismaster.h"
 #include "mongo/rpc/metadata/config_server_metadata.h"
 #include "mongo/rpc/metadata/impersonated_user_metadata.h"
-#include "mongo/rpc/metadata/logical_time_metadata.h"
 #include "mongo/rpc/metadata/sharding_metadata.h"
 #include "mongo/rpc/metadata/tracking_metadata.h"
 #include "mongo/util/string_map.h"
@@ -54,12 +53,13 @@ BSONObj makeEmptyMetadata() {
     return BSONObj();
 }
 
-void readRequestMetadata(OperationContext* opCtx, const BSONObj& metadataObj, bool requiresAuth) {
+void readRequestMetadata(OperationContext* opCtx,
+                         const BSONObj& metadataObj,
+                         bool cmdRequiresAuth) {
     BSONElement readPreferenceElem;
     BSONElement configSvrElem;
     BSONElement trackingElem;
     BSONElement clientElem;
-    BSONElement logicalTimeElem;
     BSONElement impersonationElem;
     BSONElement clientOperationKeyElem;
 
@@ -73,8 +73,6 @@ void readRequestMetadata(OperationContext* opCtx, const BSONObj& metadataObj, bo
             clientElem = metadataElem;
         } else if (fieldName == TrackingMetadata::fieldName()) {
             trackingElem = metadataElem;
-        } else if (fieldName == LogicalTimeMetadata::fieldName()) {
-            logicalTimeElem = metadataElem;
         } else if (fieldName == kImpersonationMetadataSectionName) {
             impersonationElem = metadataElem;
         } else if (fieldName == "clientOperationKey"_sd) {
@@ -107,43 +105,7 @@ void readRequestMetadata(OperationContext* opCtx, const BSONObj& metadataObj, bo
     TrackingMetadata::get(opCtx) =
         uassertStatusOK(TrackingMetadata::readFromMetadata(trackingElem));
 
-    VectorClock::get(opCtx)->gossipIn(metadataObj, opCtx->getClient()->getSessionTags());
-
-    auto logicalClock = LogicalClock::get(opCtx);
-    if (logicalClock && logicalClock->isEnabled()) {
-        auto logicalTimeMetadata =
-            uassertStatusOK(rpc::LogicalTimeMetadata::readFromMetadata(logicalTimeElem));
-
-        auto& signedTime = logicalTimeMetadata.getSignedTime();
-
-        if (!requiresAuth &&
-            AuthorizationManager::get(opCtx->getServiceContext())->isAuthEnabled() &&
-            (!signedTime.getProof() || *signedTime.getProof() == TimeProofService::TimeProof())) {
-
-            // The client is not authenticated and is not using localhost auth bypass.
-            if (authSession && !authSession->isAuthenticated() &&
-                !authSession->isUsingLocalhostBypass()) {
-                return;
-            }
-        }
-
-        // LogicalTimeMetadata is default constructed if no cluster time metadata was sent, so a
-        // default constructed SignedLogicalTime should be ignored.
-        if (signedTime.getTime() != LogicalTime::kUninitialized) {
-            auto logicalTimeValidator = LogicalTimeValidator::get(opCtx);
-            if (!LogicalTimeValidator::isAuthorizedToAdvanceClock(opCtx)) {
-                if (!logicalTimeValidator) {
-                    uasserted(ErrorCodes::CannotVerifyAndSignLogicalTime,
-                              "Cannot accept logicalTime: " + signedTime.getTime().toString() +
-                                  ". May not be a part of a sharded cluster");
-                } else {
-                    uassertStatusOK(logicalTimeValidator->validate(opCtx, signedTime));
-                }
-            }
-
-            uassertStatusOK(logicalClock->advanceClusterTime(signedTime.getTime()));
-        }
-    }
+    VectorClock::get(opCtx)->gossipIn(opCtx, metadataObj, !cmdRequiresAuth);
 }
 
 namespace {
diff --git a/src/mongo/rpc/metadata/logical_time_metadata.cpp b/src/mongo/rpc/metadata/logical_time_metadata.cpp
deleted file mode 100644
index f10e479aa4a..00000000000
--- a/src/mongo/rpc/metadata/logical_time_metadata.cpp
+++ /dev/null
@@ -1,124 +0,0 @@
-/**
- *    Copyright (C) 2018-present MongoDB, Inc.
- *
- *    This program is free software: you can redistribute it and/or modify
- *    it under the terms of the Server Side Public License, version 1,
- *    as published by MongoDB, Inc.
- *
- *    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
- *    Server Side Public License for more details.
- *
- *    You should have received a copy of the Server Side Public License
- *    along with this program. If not, see
- *    <http://www.mongodb.com/licensing/server-side-public-license>.
- *
- *    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 Server Side 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/rpc/metadata/logical_time_metadata.h"
-
-#include "mongo/bson/bsonobjbuilder.h"
-#include "mongo/bson/util/bson_extract.h"
-
-namespace mongo {
-namespace rpc {
-
-namespace {
-
-const char kClusterTimeFieldName[] = "clusterTime";
-const char kSignatureFieldName[] = "signature";
-const char kSignatureHashFieldName[] = "hash";
-const char kSignatureKeyIdFieldName[] = "keyId";
-
-}  // unnamed namespace
-
-
-LogicalTimeMetadata::LogicalTimeMetadata(SignedLogicalTime time) : _clusterTime(std::move(time)) {}
-
-StatusWith<LogicalTimeMetadata> LogicalTimeMetadata::readFromMetadata(const BSONObj& metadata) {
-    return readFromMetadata(metadata.getField(fieldName()));
-}
-
-StatusWith<LogicalTimeMetadata> LogicalTimeMetadata::readFromMetadata(
-    const BSONElement& metadataElem) {
-    if (metadataElem.eoo()) {
-        return LogicalTimeMetadata();
-    }
-
-    const auto& obj = metadataElem.Obj();
-
-    Timestamp ts;
-    Status status = bsonExtractTimestampField(obj, kClusterTimeFieldName, &ts);
-    if (!status.isOK()) {
-        return status;
-    }
-
-    BSONElement signatureElem;
-    status = bsonExtractTypedField(obj, kSignatureFieldName, Object, &signatureElem);
-    if (!status.isOK()) {
-        return status;
-    }
-
-    const auto& signatureObj = signatureElem.Obj();
-
-    // Extract BinData type signature hash and construct a SHA1Block instance from it.
-    BSONElement hashElem;
-    status = bsonExtractTypedField(signatureObj, kSignatureHashFieldName, BinData, &hashElem);
-    if (!status.isOK()) {
-        return status;
-    }
-
-    int hashLength = 0;
-    auto rawBinSignature = hashElem.binData(hashLength);
-    BSONBinData proofBinData(rawBinSignature, hashLength, hashElem.binDataType());
-    auto proofStatus = SHA1Block::fromBinData(proofBinData);
-
-    if (!proofStatus.isOK()) {
-        return proofStatus.getStatus();
-    }
-
-    long long keyId;
-    status = bsonExtractIntegerField(signatureObj, kSignatureKeyIdFieldName, &keyId);
-    if (!status.isOK()) {
-        return status;
-    }
-
-    return LogicalTimeMetadata(
-        SignedLogicalTime(LogicalTime(ts), std::move(proofStatus.getValue()), keyId));
-}
-
-void LogicalTimeMetadata::writeToMetadata(BSONObjBuilder* metadataBuilder) const {
-    BSONObjBuilder subObjBuilder(metadataBuilder->subobjStart(fieldName()));
-    _clusterTime.getTime().asTimestamp().append(subObjBuilder.bb(), kClusterTimeFieldName);
-
-    BSONObjBuilder signatureObjBuilder(subObjBuilder.subobjStart(kSignatureFieldName));
-    // Cluster 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);
-    signatureObjBuilder.append(kSignatureKeyIdFieldName, _clusterTime.getKeyId());
-    signatureObjBuilder.doneFast();
-
-    subObjBuilder.doneFast();
-}
-
-const SignedLogicalTime& LogicalTimeMetadata::getSignedTime() const {
-    return _clusterTime;
-}
-
-}  // namespace rpc
-}  // namespace mongo
diff --git a/src/mongo/rpc/metadata/logical_time_metadata.h b/src/mongo/rpc/metadata/logical_time_metadata.h
deleted file mode 100644
index 1b8c81e3286..00000000000
--- a/src/mongo/rpc/metadata/logical_time_metadata.h
+++ /dev/null
@@ -1,77 +0,0 @@
-/**
- *    Copyright (C) 2018-present MongoDB, Inc.
- *
- *    This program is free software: you can redistribute it and/or modify
- *    it under the terms of the Server Side Public License, version 1,
- *    as published by MongoDB, Inc.
- *
- *    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
- *    Server Side Public License for more details.
- *
- *    You should have received a copy of the Server Side Public License
- *    along with this program. If not, see
- *    <http://www.mongodb.com/licensing/server-side-public-license>.
- *
- *    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 Server Side 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 "mongo/base/status_with.h"
-#include "mongo/db/signed_logical_time.h"
-
-namespace mongo {
-
-class BSONElement;
-class BSONObjBuilder;
-
-namespace rpc {
-
-/**
- * Format:
- * logicalTime: {
- *     clusterTime: <Timestamp>,
- *     signature: {
- *         hash: <SHA1 hash of clusterTime as BinData>,
- *         keyId: <long long>
- *     }
- * }
- */
-class LogicalTimeMetadata {
-public:
-    LogicalTimeMetadata() = default;
-    explicit LogicalTimeMetadata(SignedLogicalTime time);
-
-    /**
-     * Parses the metadata from BSON. Returns an empty LogicalTimeMetadata If the metadata is not
-     * present.
-     */
-    static StatusWith<LogicalTimeMetadata> readFromMetadata(const BSONObj& metadata);
-    static StatusWith<LogicalTimeMetadata> readFromMetadata(const BSONElement& metadataElem);
-
-    void writeToMetadata(BSONObjBuilder* metadataBuilder) const;
-
-    const SignedLogicalTime& getSignedTime() const;
-
-    static StringData fieldName() {
-        return "$clusterTime";
-    }
-
-private:
-    SignedLogicalTime _clusterTime;
-};
-
-}  // namespace rpc
-}  // namespace mongo
diff --git a/src/mongo/rpc/metadata/logical_time_metadata_test.cpp b/src/mongo/rpc/metadata/logical_time_metadata_test.cpp
deleted file mode 100644
index a922b176446..00000000000
--- a/src/mongo/rpc/metadata/logical_time_metadata_test.cpp
+++ /dev/null
@@ -1,195 +0,0 @@
-/**
- *    Copyright (C) 2018-present MongoDB, Inc.
- *
- *    This program is free software: you can redistribute it and/or modify
- *    it under the terms of the Server Side Public License, version 1,
- *    as published by MongoDB, Inc.
- *
- *    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
- *    Server Side Public License for more details.
- *
- *    You should have received a copy of the Server Side Public License
- *    along with this program. If not, see
- *    <http://www.mongodb.com/licensing/server-side-public-license>.
- *
- *    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 Server Side 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_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kTest
-
-#include "mongo/platform/basic.h"
-
-#include "mongo/db/jsobj.h"
-#include "mongo/rpc/metadata.h"
-#include "mongo/rpc/metadata/logical_time_metadata.h"
-#include "mongo/unittest/unittest.h"
-
-
-namespace mongo {
-namespace rpc {
-namespace {
-
-TEST(LogicalTimeMetadataTest, Roundtrip) {
-    const auto ts = LogicalTime(Timestamp(100, 200));
-
-    SHA1Block::HashType proof;
-    proof.fill(0);
-    proof[19] = 6;
-    proof[0] = 12;
-
-    long long keyId = 1;
-
-    SignedLogicalTime signedTs(LogicalTime(ts), proof, keyId);
-
-    LogicalTimeMetadata origMetadata(signedTs);
-    BSONObjBuilder builder;
-    origMetadata.writeToMetadata(&builder);
-
-    auto serializedObj = builder.done();
-    auto parseStatus = LogicalTimeMetadata::readFromMetadata(serializedObj);
-    ASSERT_OK(parseStatus.getStatus());
-
-    const auto& parsedMetadata = parseStatus.getValue();
-    const auto& parsedTs = parsedMetadata.getSignedTime();
-    ASSERT_EQ(ts.asTimestamp(), parsedTs.getTime().asTimestamp());
-    ASSERT_TRUE(SHA1Block(proof) == parsedTs.getProof());
-    ASSERT_TRUE(keyId == parsedTs.getKeyId());
-}
-
-TEST(LogicalTimeMetadataTest, MissingClusterTimeShouldFailToParse) {
-    std::array<uint8_t, 20> proof;
-    proof.fill(0);
-
-    long long keyId = 1;
-
-    BSONObjBuilder builder;
-    BSONObjBuilder subObjBuilder(builder.subobjStart("$clusterTime"));
-    BSONObjBuilder signatureObjBuilder(subObjBuilder.subobjStart("signature"));
-    signatureObjBuilder.append("hash", BSONBinData(proof.data(), proof.size(), BinDataGeneral));
-    signatureObjBuilder.append("keyId", keyId);
-    signatureObjBuilder.doneFast();
-    subObjBuilder.doneFast();
-
-    auto serializedObj = builder.done();
-    auto status = LogicalTimeMetadata::readFromMetadata(serializedObj).getStatus();
-    ASSERT_EQ(ErrorCodes::NoSuchKey, status);
-}
-
-TEST(LogicalTimeMetadataTest, MissingSignatureShouldFailToParse) {
-    const auto ts = Timestamp(100, 200);
-
-    BSONObjBuilder builder;
-    BSONObjBuilder subObjBuilder(builder.subobjStart("$clusterTime"));
-    ts.append(subObjBuilder.bb(), "clusterTime");
-    subObjBuilder.doneFast();
-
-    auto serializedObj = builder.done();
-    auto status = LogicalTimeMetadata::readFromMetadata(serializedObj).getStatus();
-    ASSERT_EQ(ErrorCodes::NoSuchKey, status);
-}
-
-TEST(LogicalTimeMetadataTest, MissingHashShouldFailToParse) {
-    const auto ts = Timestamp(100, 200);
-
-    long long keyId = 1;
-
-    BSONObjBuilder builder;
-    BSONObjBuilder subObjBuilder(builder.subobjStart("$clusterTime"));
-    ts.append(subObjBuilder.bb(), "clusterTime");
-    BSONObjBuilder signatureObjBuilder(subObjBuilder.subobjStart("signature"));
-    signatureObjBuilder.append("keyId", keyId);
-    signatureObjBuilder.doneFast();
-    subObjBuilder.doneFast();
-
-    auto serializedObj = builder.done();
-    auto status = LogicalTimeMetadata::readFromMetadata(serializedObj).getStatus();
-    ASSERT_EQ(ErrorCodes::NoSuchKey, status);
-}
-
-TEST(LogicalTimeMetadataTest, MissingKeyIdShouldFailToParse) {
-    const auto ts = Timestamp(100, 200);
-
-    std::array<uint8_t, 20> proof;
-    proof.fill(0);
-
-    BSONObjBuilder builder;
-    BSONObjBuilder subObjBuilder(builder.subobjStart("$clusterTime"));
-    ts.append(subObjBuilder.bb(), "clusterTime");
-    BSONObjBuilder signatureObjBuilder(subObjBuilder.subobjStart("signature"));
-    signatureObjBuilder.append("hash", BSONBinData(proof.data(), proof.size(), BinDataGeneral));
-    signatureObjBuilder.doneFast();
-    subObjBuilder.doneFast();
-
-    auto serializedObj = builder.done();
-    auto status = LogicalTimeMetadata::readFromMetadata(serializedObj).getStatus();
-    ASSERT_EQ(ErrorCodes::NoSuchKey, status);
-}
-
-TEST(LogicalTimeMetadataTest, ProofWithWrongLengthShouldFailToParse) {
-    const auto ts = Timestamp(100, 200);
-
-    std::array<uint8_t, 10> proof;
-    proof.fill(0);
-
-    long long keyId = 1;
-
-    BSONObjBuilder builder;
-    BSONObjBuilder subObjBuilder(builder.subobjStart("$clusterTime"));
-    ts.append(subObjBuilder.bb(), "clusterTime");
-    BSONObjBuilder signatureObjBuilder(subObjBuilder.subobjStart("signature"));
-    signatureObjBuilder.append("hash", BSONBinData(proof.data(), proof.size(), BinDataGeneral));
-    signatureObjBuilder.append("keyId", keyId);
-    signatureObjBuilder.doneFast();
-    subObjBuilder.doneFast();
-
-    auto serializedObj = builder.done();
-    auto status = LogicalTimeMetadata::readFromMetadata(serializedObj).getStatus();
-    ASSERT_EQ(ErrorCodes::UnsupportedFormat, status);
-}
-
-TEST(LogicalTimeMetadataTest, UpconvertPass) {
-
-    const auto ts = Timestamp(100, 200);
-
-    std::array<uint8_t, 20> proof;
-    proof.fill(0);
-
-    long long keyId = 1;
-
-    BSONObjBuilder builder;
-    builder.append("aaa", 1);
-    builder.append("bbb", 1);
-    BSONObjBuilder subObjBuilder(builder.subobjStart("$clusterTime"));
-    ts.append(subObjBuilder.bb(), "clusterTime");
-    BSONObjBuilder signatureObjBuilder(subObjBuilder.subobjStart("signature"));
-    signatureObjBuilder.append("hash", BSONBinData(proof.data(), proof.size(), BinDataGeneral));
-    signatureObjBuilder.append("keyId", keyId);
-    signatureObjBuilder.doneFast();
-    auto logicalTimeMetadata = subObjBuilder.asTempObj();
-    subObjBuilder.doneFast();
-
-    auto commandObj = builder.done();
-    BSONObjBuilder metadataBob;
-    BSONObjBuilder commandBob;
-    auto converted = upconvertRequest("db", commandObj, 0);
-    ASSERT_BSONOBJ_EQ(BSON("aaa" << 1 << "bbb" << 1 << "$clusterTime" << logicalTimeMetadata
-                                 << "$db"
-                                 << "db"),
-                      converted.body);
-}
-
-}  // namespace
-}  // namespace rpc
-}  // namespace mongo
diff --git a/src/mongo/s/SConscript b/src/mongo/s/SConscript
index 8d3ed551a04..e5b3ec08eb6 100644
--- a/src/mongo/s/SConscript
+++ b/src/mongo/s/SConscript
@@ -449,6 +449,16 @@ env.Library(
     ],
 )
 
+env.Library(
+    target='vector_clock_mongos',
+    source=[
+        'vector_clock_mongos.cpp',
+    ],
+    LIBDEPS=[
+        '$BUILD_DIR/mongo/db/vector_clock',
+    ],
+)
+
 env.CppUnitTest(
     target='s_test',
     source=[
@@ -504,11 +514,11 @@ env.CppUnitTest(
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/db/auth/authmocks',
-        '$BUILD_DIR/mongo/db/logical_clock',
         '$BUILD_DIR/mongo/db/ops/write_ops_parsers_test_helpers',
         '$BUILD_DIR/mongo/db/query/query_test_service_context',
         '$BUILD_DIR/mongo/db/s/sharding_runtime_d',
         '$BUILD_DIR/mongo/db/service_context_test_fixture',
+        '$BUILD_DIR/mongo/db/vector_clock',
         '$BUILD_DIR/mongo/dbtests/mocklib',
         '$BUILD_DIR/mongo/util/net/network',
         'catalog_cache_test_fixture',
diff --git a/src/mongo/s/client/SConscript b/src/mongo/s/client/SConscript
index c1a4f9e2afa..c488f2793d8 100644
--- a/src/mongo/s/client/SConscript
+++ b/src/mongo/s/client/SConscript
@@ -81,6 +81,7 @@ env.CppUnitTest(
         '$BUILD_DIR/mongo/s/coreshard',
         '$BUILD_DIR/mongo/s/query/async_results_merger',
         '$BUILD_DIR/mongo/s/sharding_router_test_fixture',
+        '$BUILD_DIR/mongo/s/vector_clock_mongos',
         'sharding_client',
     ],
 )
diff --git a/src/mongo/s/commands/SConscript b/src/mongo/s/commands/SConscript
index f59f09d6aa6..9f7c310ffef 100644
--- a/src/mongo/s/commands/SConscript
+++ b/src/mongo/s/commands/SConscript
@@ -96,7 +96,6 @@ env.Library(
         'kill_sessions_remote.cpp',
         's_read_write_concern_defaults_server_status.cpp',
         'strategy.cpp',
-        'vector_clock_mongos.cpp',
         env.Idlc('cluster_multicast.idl')[0],
         env.Idlc('kill_sessions_remote.idl')[0],
     ],
@@ -140,6 +139,7 @@ env.Library(
         '$BUILD_DIR/mongo/s/query/cluster_client_cursor',
         '$BUILD_DIR/mongo/s/sharding_api',
         '$BUILD_DIR/mongo/s/sharding_router_api',
+        '$BUILD_DIR/mongo/s/vector_clock_mongos',
         '$BUILD_DIR/mongo/transport/message_compressor',
         '$BUILD_DIR/mongo/transport/transport_layer_common',
         'shared_cluster_commands',
diff --git a/src/mongo/s/commands/cluster_command_test_fixture.cpp b/src/mongo/s/commands/cluster_command_test_fixture.cpp
index 60575f966c9..0d1a88bf578 100644
--- a/src/mongo/s/commands/cluster_command_test_fixture.cpp
+++ b/src/mongo/s/commands/cluster_command_test_fixture.cpp
@@ -41,6 +41,7 @@
 #include "mongo/db/logical_clock.h"
 #include "mongo/db/logical_session_cache_noop.h"
 #include "mongo/db/logical_time_validator.h"
+#include "mongo/db/vector_clock.h"
 #include "mongo/s/cluster_last_error_info.h"
 #include "mongo/util/fail_point.h"
 #include "mongo/util/options_parser/startup_option_init.h"
@@ -54,8 +55,8 @@ void ClusterCommandTestFixture::setUp() {
 
     // Set up a logical clock with an initial time.
     auto logicalClock = std::make_unique<LogicalClock>(getServiceContext());
-    logicalClock->setClusterTimeFromTrustedSource(kInMemoryLogicalTime);
     LogicalClock::set(getServiceContext(), std::move(logicalClock));
+    VectorClock::get(getServiceContext())->advanceClusterTime_forTest(kInMemoryLogicalTime);
 
     auto keysCollectionClient = std::make_unique<KeysCollectionClientSharded>(
         Grid::get(operationContext())->catalogClient());
diff --git a/src/mongo/s/commands/strategy.cpp b/src/mongo/s/commands/strategy.cpp
index e4cd71330d3..39763246f92 100644
--- a/src/mongo/s/commands/strategy.cpp
+++ b/src/mongo/s/commands/strategy.cpp
@@ -68,7 +68,6 @@
 #include "mongo/logv2/log.h"
 #include "mongo/rpc/factory.h"
 #include "mongo/rpc/get_status_from_command_result.h"
-#include "mongo/rpc/metadata/logical_time_metadata.h"
 #include "mongo/rpc/metadata/tracking_metadata.h"
 #include "mongo/rpc/op_msg.h"
 #include "mongo/rpc/op_msg_rpc_impls.h"
@@ -99,42 +98,6 @@ namespace {
 const auto kOperationTime = "operationTime"_sd;
 
 /**
- * Extract and process metadata from the command request body.
- */
-Status processCommandMetadata(OperationContext* opCtx, const BSONObj& cmdObj) {
-    ReadPreferenceSetting::get(opCtx) =
-        uassertStatusOK(ReadPreferenceSetting::fromContainingBSON(cmdObj));
-
-    VectorClock::get(opCtx)->gossipIn(cmdObj, opCtx->getClient()->getSessionTags());
-
-    auto logicalClock = LogicalClock::get(opCtx);
-    invariant(logicalClock);
-
-    auto logicalTimeMetadata = rpc::LogicalTimeMetadata::readFromMetadata(cmdObj);
-    if (!logicalTimeMetadata.isOK()) {
-        return logicalTimeMetadata.getStatus();
-    }
-
-    auto logicalTimeValidator = LogicalTimeValidator::get(opCtx);
-    const auto& signedTime = logicalTimeMetadata.getValue().getSignedTime();
-
-    // No need to check proof is no time is given.
-    if (signedTime.getTime() == LogicalTime::kUninitialized) {
-        return Status::OK();
-    }
-
-    if (!LogicalTimeValidator::isAuthorizedToAdvanceClock(opCtx)) {
-        auto advanceClockStatus = logicalTimeValidator->validate(opCtx, signedTime);
-
-        if (!advanceClockStatus.isOK()) {
-            return advanceClockStatus;
-        }
-    }
-
-    return logicalClock->advanceClusterTime(signedTime.getTime());
-}
-
-/**
  * Invoking `shouldGossipLogicalTime()` is expected to always return "true" during normal execution.
  * SERVER-48013 uses this property to avoid the cost of calling this function during normal
  * execution. However, it might be desired to do the validation for test purposes (e.g.,
@@ -148,8 +111,6 @@ MONGO_FAIL_POINT_DEFINE(allowSkippingAppendRequiredFieldsToResponse);
  * Append required fields to command response.
  */
 void appendRequiredFieldsToResponse(OperationContext* opCtx, BSONObjBuilder* responseBuilder) {
-    VectorClock::get(opCtx)->gossipOut(responseBuilder, opCtx->getClient()->getSessionTags());
-
     // TODO SERVER-48142 should remove the following block.
     if (MONGO_unlikely(allowSkippingAppendRequiredFieldsToResponse.shouldFail())) {
         auto validator = LogicalTimeValidator::get(opCtx);
@@ -159,35 +120,34 @@ void appendRequiredFieldsToResponse(OperationContext* opCtx, BSONObjBuilder* res
         }
     }
 
-    auto now = LogicalClock::get(opCtx)->getClusterTime();
-
-    // Add operationTime.
-    auto operationTime = OperationTimeTracker::get(opCtx)->getMaxOperationTime();
-    if (operationTime != LogicalTime::kUninitialized) {
-        LOGV2_DEBUG(22764,
-                    5,
-                    "Appending operationTime: {operationTime}",
-                    "Appending operationTime",
-                    "operationTime"_attr = operationTime.asTimestamp());
-        responseBuilder->append(kOperationTime, operationTime.asTimestamp());
-    } else if (now != LogicalTime::kUninitialized) {
-        // If we don't know the actual operation time, use the cluster time instead. This is
-        // safe but not optimal because we can always return a later operation time than actual.
-        LOGV2_DEBUG(22765,
-                    5,
-                    "Appending clusterTime as operationTime {clusterTime}",
-                    "Appending clusterTime as operationTime",
-                    "clusterTime"_attr = now.asTimestamp());
-        responseBuilder->append(kOperationTime, now.asTimestamp());
-    }
-
-    // Add $clusterTime.
-    if (LogicalTimeValidator::isAuthorizedToAdvanceClock(opCtx)) {
-        SignedLogicalTime dummySignedTime(now, TimeProofService::TimeProof(), 0);
-        rpc::LogicalTimeMetadata(dummySignedTime).writeToMetadata(responseBuilder);
-    } else {
-        auto currentTime = LogicalTimeValidator::get(opCtx)->signLogicalTime(opCtx, now);
-        rpc::LogicalTimeMetadata(currentTime).writeToMetadata(responseBuilder);
+    // The appended operationTime must always be <= the appended $clusterTime, so in case we need to
+    // use $clusterTime as the operationTime below, take a $clusterTime value which is guaranteed to
+    // be <= the value output by gossipOut().
+    auto clusterTime = VectorClock::get(opCtx)->getTime()[VectorClock::Component::ClusterTime];
+
+    bool clusterTimeWasOutput = VectorClock::get(opCtx)->gossipOut(opCtx, responseBuilder);
+
+    // Ensure that either both operationTime and $clusterTime are output, or neither.
+    if (clusterTimeWasOutput) {
+        auto operationTime = OperationTimeTracker::get(opCtx)->getMaxOperationTime();
+        if (operationTime != LogicalTime::kUninitialized) {
+            LOGV2_DEBUG(22764,
+                        5,
+                        "Appending operationTime: {operationTime}",
+                        "Appending operationTime",
+                        "operationTime"_attr = operationTime.asTimestamp());
+            operationTime.appendAsOperationTime(responseBuilder);
+        } else if (clusterTime != LogicalTime::kUninitialized) {
+            // If we don't know the actual operation time, use the cluster time instead. This is
+            // safe but not optimal because we can always return a later operation time than
+            // actual.
+            LOGV2_DEBUG(22765,
+                        5,
+                        "Appending clusterTime as operationTime {clusterTime}",
+                        "Appending clusterTime as operationTime",
+                        "clusterTime"_attr = clusterTime.asTimestamp());
+            clusterTime.appendAsOperationTime(responseBuilder);
+        }
     }
 }
 
@@ -278,12 +238,10 @@ void execCommandClient(OperationContext* opCtx,
         trackingMetadata.initWithOperName(c->getName());
         rpc::TrackingMetadata::get(opCtx) = trackingMetadata;
 
-        auto metadataStatus = processCommandMetadata(opCtx, request.body);
-        if (!metadataStatus.isOK()) {
-            auto body = result->getBodyBuilder();
-            CommandHelpers::appendCommandStatusNoThrow(body, metadataStatus);
-            return;
-        }
+        // Extract and process metadata from the command request body.
+        ReadPreferenceSetting::get(opCtx) =
+            uassertStatusOK(ReadPreferenceSetting::fromContainingBSON(request.body));
+        VectorClock::get(opCtx)->gossipIn(opCtx, request.body, !c->requiresAuth());
 
         auto txnRouter = TransactionRouter::get(opCtx);
         if (txnRouter) {
diff --git a/src/mongo/s/transaction_router_test.cpp b/src/mongo/s/transaction_router_test.cpp
index d7ef7a1f019..385f02a781d 100644
--- a/src/mongo/s/transaction_router_test.cpp
+++ b/src/mongo/s/transaction_router_test.cpp
@@ -40,6 +40,7 @@
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/repl/read_concern_args.h"
 #include "mongo/db/session_catalog.h"
+#include "mongo/db/vector_clock_mutable.h"
 #include "mongo/logger/logger.h"
 #include "mongo/rpc/get_status_from_command_result.h"
 #include "mongo/s/catalog/type_shard.h"
@@ -125,8 +126,9 @@ protected:
 
         // Set up a logical clock with an initial time.
         auto logicalClock = std::make_unique<LogicalClock>(getServiceContext());
-        logicalClock->setClusterTimeFromTrustedSource(kInMemoryLogicalTime);
         LogicalClock::set(getServiceContext(), std::move(logicalClock));
+        VectorClockMutable::get(getServiceContext())
+            ->tickTo(VectorClock::Component::ClusterTime, kInMemoryLogicalTime);
 
         // Set up a tick source for transaction metrics.
         auto tickSource = std::make_unique<TickSourceMock<Microseconds>>();
@@ -1428,7 +1430,8 @@ TEST_F(TransactionRouterTestWithDefaultSession, SnapshotErrorsResetAtClusterTime
     // Advance the latest time in the logical clock so the retry attempt will pick a later time.
     LogicalTime laterTime(Timestamp(1000, 1));
     ASSERT_GT(laterTime, kInMemoryLogicalTime);
-    LogicalClock::get(operationContext())->setClusterTimeFromTrustedSource(laterTime);
+    VectorClockMutable::get(operationContext())
+        ->tickTo(VectorClock::Component::ClusterTime, laterTime);
 
     // Simulate a snapshot error.
 
@@ -1477,7 +1480,8 @@ TEST_F(TransactionRouterTestWithDefaultSession,
 
     LogicalTime laterTimeSameStmt(Timestamp(100, 1));
     ASSERT_GT(laterTimeSameStmt, kInMemoryLogicalTime);
-    LogicalClock::get(operationContext())->setClusterTimeFromTrustedSource(laterTimeSameStmt);
+    VectorClockMutable::get(operationContext())
+        ->tickTo(VectorClock::Component::ClusterTime, laterTimeSameStmt);
 
     txnRouter.setDefaultAtClusterTime(operationContext());
 
@@ -1501,7 +1505,8 @@ TEST_F(TransactionRouterTestWithDefaultSession,
 
     LogicalTime laterTimeNewStmt(Timestamp(1000, 1));
     ASSERT_GT(laterTimeNewStmt, laterTimeSameStmt);
-    LogicalClock::get(operationContext())->setClusterTimeFromTrustedSource(laterTimeNewStmt);
+    VectorClockMutable::get(operationContext())
+        ->tickTo(VectorClock::Component::ClusterTime, laterTimeNewStmt);
 
     txnRouter.setDefaultAtClusterTime(operationContext());
 
@@ -1749,7 +1754,8 @@ TEST_F(TransactionRouterTestWithDefaultSession,
 
     LogicalTime laterTime(Timestamp(1000, 1));
     ASSERT_GT(laterTime, kInMemoryLogicalTime);
-    LogicalClock::get(operationContext())->setClusterTimeFromTrustedSource(laterTime);
+    VectorClockMutable::get(operationContext())
+        ->tickTo(VectorClock::Component::ClusterTime, laterTime);
 
     ASSERT(txnRouter.canContinueOnStaleShardOrDbError("find", kDummyStatus));
     txnRouter.onStaleShardOrDbError(operationContext(), "find", kDummyStatus);
@@ -2315,7 +2321,8 @@ TEST_F(TransactionRouterTestWithDefaultSession,
 
     LogicalTime laterTimeSameStmt(Timestamp(100, 1));
     ASSERT_GT(laterTimeSameStmt, kInMemoryLogicalTime);
-    LogicalClock::get(operationContext())->setClusterTimeFromTrustedSource(laterTimeSameStmt);
+    VectorClockMutable::get(operationContext())
+        ->tickTo(VectorClock::Component::ClusterTime, laterTimeSameStmt);
 
     txnRouter.setDefaultAtClusterTime(operationContext());
 
diff --git a/src/mongo/s/commands/vector_clock_mongos.cpp b/src/mongo/s/vector_clock_mongos.cpp
index a8d0b0dbc81..320993f6d92 100644
--- a/src/mongo/s/commands/vector_clock_mongos.cpp
+++ b/src/mongo/s/vector_clock_mongos.cpp
@@ -46,10 +46,19 @@ public:
     virtual ~VectorClockMongoS();
 
 protected:
-    void _gossipOutInternal(BSONObjBuilder* out) const override;
-    void _gossipOutExternal(BSONObjBuilder* out) const override;
-    LogicalTimeArray _gossipInInternal(const BSONObj& in) override;
-    LogicalTimeArray _gossipInExternal(const BSONObj& in) override;
+    bool _gossipOutInternal(OperationContext* opCtx,
+                            BSONObjBuilder* out,
+                            const LogicalTimeArray& time) const override;
+    bool _gossipOutExternal(OperationContext* opCtx,
+                            BSONObjBuilder* out,
+                            const LogicalTimeArray& time) const override;
+    LogicalTimeArray _gossipInInternal(OperationContext* opCtx,
+                                       const BSONObj& in,
+                                       bool couldBeUnauthenticated) override;
+    LogicalTimeArray _gossipInExternal(OperationContext* opCtx,
+                                       const BSONObj& in,
+                                       bool couldBeUnauthenticated) override;
+    bool _permitRefreshDuringGossipOut() const override;
 };
 
 const auto vectorClockMongoSDecoration = ServiceContext::declareDecoration<VectorClockMongoS>();
@@ -67,37 +76,40 @@ VectorClockMongoS::VectorClockMongoS() = default;
 
 VectorClockMongoS::~VectorClockMongoS() = default;
 
-void VectorClockMongoS::_gossipOutInternal(BSONObjBuilder* out) const {
-    VectorTime now = getTime();
-    // TODO SERVER-47914: re-enable gossipping of VectorClock's ClusterTime once LogicalClock has
-    // been migrated into VectorClock.
-    // _gossipOutComponent(out, now, Component::ClusterTime);
-    _gossipOutComponent(out, now, Component::ConfigTime);
+bool VectorClockMongoS::_gossipOutInternal(OperationContext* opCtx,
+                                           BSONObjBuilder* out,
+                                           const LogicalTimeArray& time) const {
+    bool wasClusterTimeOutput = _gossipOutComponent(opCtx, out, time, Component::ClusterTime);
+    _gossipOutComponent(opCtx, out, time, Component::ConfigTime);
+    return wasClusterTimeOutput;
 }
 
-void VectorClockMongoS::_gossipOutExternal(BSONObjBuilder* out) const {
-    // TODO SERVER-47914: re-enable gossipping of VectorClock's ClusterTime once LogicalClock has
-    // been migrated into VectorClock.
-    // VectorTime now = getTime();
-    // _gossipOutComponent(out, now, Component::ClusterTime);
+bool VectorClockMongoS::_gossipOutExternal(OperationContext* opCtx,
+                                           BSONObjBuilder* out,
+                                           const LogicalTimeArray& time) const {
+    return _gossipOutComponent(opCtx, out, time, Component::ClusterTime);
 }
 
-VectorClock::LogicalTimeArray VectorClockMongoS::_gossipInInternal(const BSONObj& in) {
+VectorClock::LogicalTimeArray VectorClockMongoS::_gossipInInternal(OperationContext* opCtx,
+                                                                   const BSONObj& in,
+                                                                   bool couldBeUnauthenticated) {
     LogicalTimeArray newTime;
-    // TODO SERVER-47914: re-enable gossipping of VectorClock's ClusterTime once LogicalClock has
-    // been migrated into VectorClock.
-    // _gossipInComponent(in, &newTime, Component::ClusterTime);
-    _gossipInComponent(in, &newTime, Component::ConfigTime);
+    _gossipInComponent(opCtx, in, couldBeUnauthenticated, &newTime, Component::ClusterTime);
+    _gossipInComponent(opCtx, in, couldBeUnauthenticated, &newTime, Component::ConfigTime);
     return newTime;
 }
 
-VectorClock::LogicalTimeArray VectorClockMongoS::_gossipInExternal(const BSONObj& in) {
+VectorClock::LogicalTimeArray VectorClockMongoS::_gossipInExternal(OperationContext* opCtx,
+                                                                   const BSONObj& in,
+                                                                   bool couldBeUnauthenticated) {
     LogicalTimeArray newTime;
-    // TODO SERVER-47914: re-enable gossipping of VectorClock's ClusterTime once LogicalClock has
-    // been migrated into VectorClock.
-    // _gossipInComponent(in, &newTime, Component::ClusterTime);
+    _gossipInComponent(opCtx, in, couldBeUnauthenticated, &newTime, Component::ClusterTime);
     return newTime;
 }
 
+bool VectorClockMongoS::_permitRefreshDuringGossipOut() const {
+    return true;
+}
+
 }  // namespace
 }  // namespace mongo
diff --git a/src/mongo/s/write_ops/SConscript b/src/mongo/s/write_ops/SConscript
index 62d2ed4a9db..65eb38704c4 100644
--- a/src/mongo/s/write_ops/SConscript
+++ b/src/mongo/s/write_ops/SConscript
@@ -16,10 +16,10 @@ env.Library(
         '$BUILD_DIR/mongo/base',
         '$BUILD_DIR/mongo/db/common',
         '$BUILD_DIR/mongo/db/error_labels',
-        '$BUILD_DIR/mongo/db/logical_clock',
         '$BUILD_DIR/mongo/db/ops/write_ops_parsers',
         '$BUILD_DIR/mongo/db/repl/optime',
         '$BUILD_DIR/mongo/db/commands',
+        '$BUILD_DIR/mongo/db/vector_clock',
         '$BUILD_DIR/mongo/rpc/command_status',
         '$BUILD_DIR/mongo/s/common_s',
     ],
diff --git a/src/mongo/s/write_ops/batch_write_exec_test.cpp b/src/mongo/s/write_ops/batch_write_exec_test.cpp
index 4e230942e68..4be04e06b8d 100644
--- a/src/mongo/s/write_ops/batch_write_exec_test.cpp
+++ b/src/mongo/s/write_ops/batch_write_exec_test.cpp
@@ -35,6 +35,7 @@
 #include "mongo/db/commands.h"
 #include "mongo/db/logical_clock.h"
 #include "mongo/db/logical_session_id.h"
+#include "mongo/db/vector_clock_mutable.h"
 #include "mongo/s/catalog/type_shard.h"
 #include "mongo/s/client/shard_registry.h"
 #include "mongo/s/session_catalog_router.h"
@@ -1594,8 +1595,9 @@ public:
             repl::ReadConcernArgs(repl::ReadConcernLevel::kSnapshotReadConcern);
 
         auto logicalClock = std::make_unique<LogicalClock>(getServiceContext());
-        logicalClock->setClusterTimeFromTrustedSource(kInMemoryLogicalTime);
         LogicalClock::set(getServiceContext(), std::move(logicalClock));
+        VectorClockMutable::get(getServiceContext())
+            ->tickTo(VectorClock::Component::ClusterTime, kInMemoryLogicalTime);
 
         _scopedSession.emplace(operationContext());