Update upstream source from tag 'upstream/6.0.0'master

Update to upstream version '6.0.0'
with Debian dir 5604a80ec1c96ca76f25f40d78e6ef855abec322

70 files changed, 930 insertions, 2704 deletions

diff --git a/src/mongo/db/storage/SConscript b/src/mongo/db/storage/SConscript
index 01b8e741b81..45931873ba3 100644
--- a/src/mongo/db/storage/SConscript
+++ b/src/mongo/db/storage/SConscript
@@ -90,7 +90,6 @@ env.Library(
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/base',
-        '$BUILD_DIR/mongo/bson/bson_validate',
     ],
 )
 
@@ -333,7 +332,6 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/concurrency/lock_manager',
-        '$BUILD_DIR/mongo/db/exec/scoped_timer',
         '$BUILD_DIR/mongo/db/service_context',
         'storage_change_lock',
         'storage_control',
@@ -390,7 +388,6 @@ env.Library(
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/base',
-        '$BUILD_DIR/mongo/bson/bson_validate',
         '$BUILD_DIR/mongo/db/bson/dotted_path_support',
         '$BUILD_DIR/mongo/db/server_options_core',
     ],
@@ -543,13 +540,14 @@ env.Library(
         '$BUILD_DIR/mongo/bson/util/bson_extract',
         '$BUILD_DIR/mongo/db/catalog/collection_catalog',
         '$BUILD_DIR/mongo/db/concurrency/lock_manager',
-        '$BUILD_DIR/mongo/db/index/index_access_method',
+        '$BUILD_DIR/mongo/db/index/index_descriptor',
         '$BUILD_DIR/mongo/db/index_names',
         '$BUILD_DIR/mongo/db/namespace_string',
         '$BUILD_DIR/mongo/db/storage/bson_collection_catalog_entry',
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/catalog/collection',
+        '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception',
         '$BUILD_DIR/mongo/db/multitenancy',
         '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface',
         '$BUILD_DIR/mongo/db/server_options_core',
@@ -625,7 +623,6 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/audit',
         '$BUILD_DIR/mongo/db/catalog/catalog_helpers',
-        '$BUILD_DIR/mongo/db/catalog/clustered_collection_options',
         '$BUILD_DIR/mongo/db/catalog/index_catalog',
         '$BUILD_DIR/mongo/db/multitenancy',
         '$BUILD_DIR/mongo/db/resumable_index_builds_idl',
@@ -647,7 +644,6 @@ env.CppLibfuzzerTest(
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/base',
-        '$BUILD_DIR/mongo/bson/bson_validate',
         'key_string',
     ],
 )
diff --git a/src/mongo/db/storage/bson_collection_catalog_entry.h b/src/mongo/db/storage/bson_collection_catalog_entry.h
index f188ffbb34e..9b3d196ec0f 100644
--- a/src/mongo/db/storage/bson_collection_catalog_entry.h
+++ b/src/mongo/db/storage/bson_collection_catalog_entry.h
@@ -120,7 +120,6 @@ public:
         mutable Mutex multikeyMutex;
         mutable bool multikey = false;
         mutable MultikeyPaths multikeyPaths;
-        mutable AtomicWord<int32_t> concurrentWriters;
     };
 
     struct MetaData {
@@ -152,7 +151,6 @@ public:
 
         std::string ns;
         CollectionOptions options;
-        // May include empty instances which represent indexes already dropped.
         std::vector<IndexMetaData> indexes;
 
         // Time-series collections created in versions 5.1 and earlier are allowed to contain
diff --git a/src/mongo/db/storage/control/storage_control.cpp b/src/mongo/db/storage/control/storage_control.cpp
index 3f5af49c729..11222448eba 100644
--- a/src/mongo/db/storage/control/storage_control.cpp
+++ b/src/mongo/db/storage/control/storage_control.cpp
@@ -79,8 +79,8 @@ void startStorageControls(ServiceContext* serviceContext, bool forTestOnly) {
         std::unique_ptr<JournalFlusher> journalFlusher = std::make_unique<JournalFlusher>(
             /*disablePeriodicFlushes*/ forTestOnly ||
             (!storageEngine->isDurable() && !storageEngine->isEphemeral()));
+        journalFlusher->go();
         JournalFlusher::set(serviceContext, std::move(journalFlusher));
-        JournalFlusher::get(serviceContext)->go();
     }
 
     if (!storageEngine->isEphemeral() && !storageGlobalParams.readOnly) {
diff --git a/src/mongo/db/storage/deferred_drop_record_store.cpp b/src/mongo/db/storage/deferred_drop_record_store.cpp
index bb9df1f062f..2fd14d88f37 100644
--- a/src/mongo/db/storage/deferred_drop_record_store.cpp
+++ b/src/mongo/db/storage/deferred_drop_record_store.cpp
@@ -41,9 +41,6 @@ DeferredDropRecordStore::~DeferredDropRecordStore() {
         return;
     }
     try {
-        // TODO SERVER-81373: Multi-document transactions will cause the log to be spammed with
-        // EBUSY errors. This should ideally be replaced with a recoveryUnit onCommit handler if
-        // within a WUOW.
         _storageEngine->addDropPendingIdent(
             Timestamp::min(), std::make_shared<Ident>(_rs->getIdent()), nullptr);
     } catch (...) {
diff --git a/src/mongo/db/storage/devnull/devnull_kv_engine.cpp b/src/mongo/db/storage/devnull/devnull_kv_engine.cpp
index 69f3a0ece5f..86ea5ece9c4 100644
--- a/src/mongo/db/storage/devnull/devnull_kv_engine.cpp
+++ b/src/mongo/db/storage/devnull/devnull_kv_engine.cpp
@@ -133,9 +133,7 @@ public:
         MONGO_UNREACHABLE;
     }
 
-    virtual void printRecordMetadata(OperationContext* opCtx,
-                                     const RecordId& recordId,
-                                     std::set<Timestamp>* recordTimestamps) const {
+    virtual void printRecordMetadata(OperationContext* opCtx, const RecordId& recordId) const {
         MONGO_UNREACHABLE;
     }
 
@@ -238,9 +236,6 @@ public:
         return true;
     }
 
-    virtual void printIndexEntryMetadata(OperationContext* opCtx,
-                                         const KeyString::Value& keyString) const {}
-
     virtual std::unique_ptr<SortedDataInterface::Cursor> newCursor(OperationContext* opCtx,
                                                                    bool isForward) const {
         return {};
diff --git a/src/mongo/db/storage/devnull/ephemeral_catalog_record_store.h b/src/mongo/db/storage/devnull/ephemeral_catalog_record_store.h
index 67c786f0b23..c3b30e45ef0 100644
--- a/src/mongo/db/storage/devnull/ephemeral_catalog_record_store.h
+++ b/src/mongo/db/storage/devnull/ephemeral_catalog_record_store.h
@@ -83,9 +83,7 @@ public:
                                                const char* damageSource,
                                                const mutablebson::DamageVector& damages) override;
 
-    virtual void printRecordMetadata(OperationContext* opCtx,
-                                     const RecordId& recordId,
-                                     std::set<Timestamp>* recordTimestamps) const {}
+    virtual void printRecordMetadata(OperationContext* opCtx, const RecordId& recordId) const {}
 
     std::unique_ptr<SeekableRecordCursor> getCursor(OperationContext* opCtx,
                                                     bool forward) const final;
diff --git a/src/mongo/db/storage/durable_catalog.h b/src/mongo/db/storage/durable_catalog.h
index 9a157e0ece9..49db4546918 100644
--- a/src/mongo/db/storage/durable_catalog.h
+++ b/src/mongo/db/storage/durable_catalog.h
@@ -115,12 +115,7 @@ public:
 
     virtual bool isInternalIdent(StringData ident) const = 0;
 
-    static bool isCollectionIdent(StringData ident) {
-        // Internal idents prefixed "internal-" should not be considered collections, because
-        // they are not eligible for orphan recovery through repair.
-        return ident.find("collection-") != std::string::npos ||
-            ident.find("collection/") != std::string::npos;
-    }
+    virtual bool isCollectionIdent(StringData ident) const = 0;
 
 
     virtual RecordStore* getRecordStore() = 0;
@@ -128,12 +123,11 @@ public:
     /**
      * Create an entry in the catalog for an orphaned collection found in the
      * storage engine. Return the generated ns of the collection.
-     * Note that this function does not recreate the _id index on the for non-clustered collections
-     * because it does not have access to index catalog.
+     * Note that this function does not recreate the _id index on the collection because it does not
+     * have access to index catalog.
      */
     virtual StatusWith<std::string> newOrphanedIdent(OperationContext* opCtx,
-                                                     std::string ident,
-                                                     const CollectionOptions& optionsWithUUID) = 0;
+                                                     std::string ident) = 0;
 
     virtual std::string getFilesystemPathForDb(const std::string& dbName) const = 0;
 
diff --git a/src/mongo/db/storage/durable_catalog_impl.cpp b/src/mongo/db/storage/durable_catalog_impl.cpp
index 10015081c93..13b86f9af07 100644
--- a/src/mongo/db/storage/durable_catalog_impl.cpp
+++ b/src/mongo/db/storage/durable_catalog_impl.cpp
@@ -547,8 +547,15 @@ bool DurableCatalogImpl::isResumableIndexBuildIdent(StringData ident) const {
     return ident.find(kResumableIndexBuildIdentStem) != std::string::npos;
 }
 
-StatusWith<std::string> DurableCatalogImpl::newOrphanedIdent(
-    OperationContext* opCtx, std::string ident, const CollectionOptions& optionsWithUUID) {
+bool DurableCatalogImpl::isCollectionIdent(StringData ident) const {
+    // Internal idents prefixed "internal-" should not be considered collections, because
+    // they are not eligible for orphan recovery through repair.
+    return ident.find("collection-") != std::string::npos ||
+        ident.find("collection/") != std::string::npos;
+}
+
+StatusWith<std::string> DurableCatalogImpl::newOrphanedIdent(OperationContext* opCtx,
+                                                             std::string ident) {
     // The collection will be named local.orphan.xxxxx.
     std::string identNs = ident;
     std::replace(identNs.begin(), identNs.end(), '-', '_');
@@ -556,6 +563,9 @@ StatusWith<std::string> DurableCatalogImpl::newOrphanedIdent(
     NamespaceString nss(NamespaceString(NamespaceString::kOrphanCollectionDb,
                                         NamespaceString::kOrphanCollectionPrefix + identNs));
 
+    // Generate a new UUID for the orphaned collection.
+    CollectionOptions optionsWithUUID;
+    optionsWithUUID.uuid.emplace(UUID::gen());
     BSONObj obj;
     {
         BSONObjBuilder b;
diff --git a/src/mongo/db/storage/durable_catalog_impl.h b/src/mongo/db/storage/durable_catalog_impl.h
index 011ca72758b..873733d58b4 100644
--- a/src/mongo/db/storage/durable_catalog_impl.h
+++ b/src/mongo/db/storage/durable_catalog_impl.h
@@ -92,13 +92,13 @@ public:
 
     bool isResumableIndexBuildIdent(StringData ident) const;
 
+    bool isCollectionIdent(StringData ident) const;
+
     RecordStore* getRecordStore() {
         return _rs;
     }
 
-    StatusWith<std::string> newOrphanedIdent(OperationContext* opCtx,
-                                             std::string ident,
-                                             const CollectionOptions& optionsWithUUID);
+    StatusWith<std::string> newOrphanedIdent(OperationContext* opCtx, std::string ident);
 
     std::string getFilesystemPathForDb(const std::string& dbName) const;
 
diff --git a/src/mongo/db/storage/ephemeral_for_test/SConscript b/src/mongo/db/storage/ephemeral_for_test/SConscript
index 0195d42ce8a..335f6ca4bf5 100644
--- a/src/mongo/db/storage/ephemeral_for_test/SConscript
+++ b/src/mongo/db/storage/ephemeral_for_test/SConscript
@@ -15,6 +15,7 @@ env.Library(
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/base',
+        '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception',
         '$BUILD_DIR/mongo/db/storage/index_entry_comparison',
         '$BUILD_DIR/mongo/db/storage/record_store_base',
         '$BUILD_DIR/mongo/db/storage/recovery_unit_base',
@@ -62,7 +63,7 @@ env.CppUnitTest(
     LIBDEPS=[
         '$BUILD_DIR/mongo/db/auth/authmocks',
         '$BUILD_DIR/mongo/db/common',
-        '$BUILD_DIR/mongo/db/index/index_access_method',
+        '$BUILD_DIR/mongo/db/index/index_descriptor',
         '$BUILD_DIR/mongo/db/multitenancy',
         '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface',
         '$BUILD_DIR/mongo/db/repl/replmocks',
diff --git a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_record_store.h b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_record_store.h
index cf80ed48894..b6b913c57ce 100644
--- a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_record_store.h
+++ b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_record_store.h
@@ -88,9 +88,7 @@ public:
                                                const char* damageSource,
                                                const mutablebson::DamageVector& damages) final;
 
-    virtual void printRecordMetadata(OperationContext* opCtx,
-                                     const RecordId& recordId,
-                                     std::set<Timestamp>* recordTimestamps) const {}
+    virtual void printRecordMetadata(OperationContext* opCtx, const RecordId& recordId) const {}
 
     std::unique_ptr<SeekableRecordCursor> getCursor(OperationContext* opCtx,
                                                     bool forward) const final;
diff --git a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_sorted_impl.cpp b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_sorted_impl.cpp
index 7446dbdd645..519226630c6 100644
--- a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_sorted_impl.cpp
+++ b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_sorted_impl.cpp
@@ -1460,9 +1460,7 @@ void SortedDataInterfaceUnique::fullValidate(OperationContext* opCtx,
         numKeys += UniqueIndexData(it->second, _rsKeyFormat).size();
         ++it;
     }
-    if (numKeysOut) {
-        *numKeysOut = numKeys;
-    }
+    *numKeysOut = numKeys;
 }
 
 bool SortedDataInterfaceBase::appendCustomStats(OperationContext* opCtx,
@@ -1644,9 +1642,7 @@ void SortedDataInterfaceStandard::fullValidate(OperationContext* opCtx,
         ++numKeys;
         ++it;
     }
-    if (numKeysOut) {
-        *numKeysOut = numKeys;
-    }
+    *numKeysOut = numKeys;
 }
 
 std::unique_ptr<mongo::SortedDataInterface::Cursor> SortedDataInterfaceStandard::newCursor(
diff --git a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_sorted_impl.h b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_sorted_impl.h
index 91210ddef9c..3f93c83ece0 100644
--- a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_sorted_impl.h
+++ b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_sorted_impl.h
@@ -99,8 +99,6 @@ public:
                                            RecordId rid) override {
         MONGO_UNREACHABLE;
     }
-    void printIndexEntryMetadata(OperationContext* opCtx, const KeyString::Value& keyString) const {
-    }
 
 protected:
     // These two are the same as before.
diff --git a/src/mongo/db/storage/key_string.cpp b/src/mongo/db/storage/key_string.cpp
index 1284e5e4e44..908bd51908a 100644
--- a/src/mongo/db/storage/key_string.cpp
+++ b/src/mongo/db/storage/key_string.cpp
@@ -281,7 +281,7 @@ StringData readCStringWithNuls(BufReader* reader, std::string* scratch) {
         return initial;  // Don't alloc or copy for simple case with no NUL bytes.
 
     scratch->append(initial.rawData(), initial.size());
-    while (reader->remaining() && reader->peek<unsigned char>() == 0xFF) {
+    while (reader->peek<unsigned char>() == 0xFF) {
         // Each time we enter this loop it means we hit a NUL byte encoded as "\x00\xFF".
         *scratch += '\0';
         reader->skip(1);
@@ -2598,15 +2598,10 @@ BSONObj toBson(StringData data, Ordering ord, const TypeBits& typeBits) {
 
 RecordId decodeRecordIdLongAtEnd(const void* bufferRaw, size_t bufSize) {
     const unsigned char* buffer = static_cast<const unsigned char*>(bufferRaw);
-    keyStringAssert(8273006,
-                    fmt::format("Input too short to encode RecordId. bufSize: {}", bufSize),
-                    bufSize >= 2);  // smallest possible encoding of a RecordId.
+    invariant(bufSize >= 2);  // smallest possible encoding of a RecordId.
     const unsigned char lastByte = *(buffer + bufSize - 1);
     const size_t ridSize = 2 + (lastByte & 0x7);  // stored in low 3 bits.
-    keyStringAssert(
-        8273001,
-        fmt::format("Encoded RecordId size is too big. bufSize: {}, ridSize: {}", bufSize, ridSize),
-        bufSize >= ridSize);
+    invariant(bufSize >= ridSize);
     const unsigned char* firstBytePtr = buffer + bufSize - ridSize;
     BufReader reader(firstBytePtr, ridSize);
     return decodeRecordIdLong(&reader);
@@ -2661,12 +2656,7 @@ RecordId decodeRecordIdLong(BufReader* reader) {
     }
 
     const uint8_t lastByte = readType<uint8_t>(reader, false);
-    keyStringAssert(8273000,
-                    fmt::format("Number of extra bytes for RecordId is not encoded correctly. Low "
-                                "3 bits of lastByte: {}, high 3 bits of firstByte: {}",
-                                lastByte & 0x7,
-                                numExtraBytes),
-                    (lastByte & 0x7) == numExtraBytes);
+    invariant((lastByte & 0x7) == numExtraBytes);
     repr = (repr << 5) | (lastByte >> 3);  // fold in high 5 bits of last byte
     return RecordId(repr);
 }
@@ -2686,21 +2676,12 @@ RecordId decodeRecordIdStrAtEnd(const void* bufferRaw, size_t bufSize) {
     // Continuation bytes
     size_t sizeByteId = 0;
     for (; buffer[bufSize - 1 - sizeByteId] & 0x80; sizeByteId++) {
-        keyStringAssert(
-            8273002,
-            fmt::format("size bytes too long. bufSize: {}, sizeByteId: {}", bufSize, sizeByteId),
-            bufSize > sizeByteId + 1 /* this is cont, so next byte must be within buffer */);
-        keyStringAssert(
-            8273003,
-            fmt::format("size bytes longer than maximum allowed bytes. sizeByteId: {}", sizeByteId),
-            sizeByteId < kRecordIdStrEncodedSizeMaxBytes);
+        invariant(bufSize >= sizeByteId + 1 /* non-cont byte */);
+        invariant(sizeByteId < kRecordIdStrEncodedSizeMaxBytes);
         sizes[sizeByteId] = buffer[bufSize - 1 - sizeByteId] & 0x7F;
     }
     // Last (non-continuation) byte
-    keyStringAssert(
-        8273004,
-        fmt::format("size bytes longer than maximum allowed bytes. sizeByteId: {}", sizeByteId),
-        sizeByteId < kRecordIdStrEncodedSizeMaxBytes);
+    invariant(sizeByteId < kRecordIdStrEncodedSizeMaxBytes);
     sizes[sizeByteId] = buffer[bufSize - 1 - sizeByteId];
 
     const size_t numSegments = sizeByteId + 1;
@@ -2710,12 +2691,7 @@ RecordId decodeRecordIdStrAtEnd(const void* bufferRaw, size_t bufSize) {
     }
     ridSize += static_cast<size_t>(sizes[sizeByteId]) << ((numSegments - sizeByteId - 1) * 7);
 
-    keyStringAssert(8273005,
-                    fmt::format("RecordId too long. bufSize: {}, ridSize: {}, numSegments: {}",
-                                bufSize,
-                                ridSize,
-                                numSegments),
-                    bufSize >= ridSize + numSegments);
+    invariant(bufSize >= ridSize + numSegments);
 
     return RecordId(reinterpret_cast<const char*>(buffer) + (bufSize - ridSize - numSegments),
                     ridSize);
diff --git a/src/mongo/db/storage/key_string.h b/src/mongo/db/storage/key_string.h
index 30a3a1cbe8e..e1daa8881e0 100644
--- a/src/mongo/db/storage/key_string.h
+++ b/src/mongo/db/storage/key_string.h
@@ -432,14 +432,16 @@ public:
         return deserialize(buf, settings.keyStringVersion);
     }
 
-    // It is illegal to call this function on a value that is backed by a buffer that is shared
-    // elsewhere. The SharedBufferFragment cannot accurately report memory usage per individual
-    // Value, so we require the sorter to look at the SharedBufferFragmentBuilder's memory usage in
-    // aggregate and free unused memory periodically.
     int memUsageForSorter() const {
-        invariant(!_buffer.isShared(),
-                  "Cannot obtain memory usage from shared buffer on KeyString::Value");
-        return sizeof(Value) + _buffer.underlyingCapacity();
+        // Ideally we want to always use the buffer capacity as a more accurate measure of memory
+        // usage here. But when built using the PooledBuilder we cannot do that as the buffer is
+        // shared between many instances we have to use the size() as an approximation of memory
+        // use. There might be a chunk at the end of the buffer that's not used by any KeyString and
+        // that memory will be unaccounted for unfortunately.
+        // When the PooledBuilder is used this buffer will always be shared as the
+        // SharedBufferFragmentBuilder will keep a reference. If it is not shared we've used either
+        // the Heap or Static builder and want to report the whole memory allocation in the buffer.
+        return sizeof(Value) + (_buffer.isShared() ? _buffer.size() : _buffer.underlyingCapacity());
     }
 
     Value getOwned() const {
diff --git a/src/mongo/db/storage/key_string_test.cpp b/src/mongo/db/storage/key_string_test.cpp
index e1a29dcd8a1..ef9f7e744a5 100644
--- a/src/mongo/db/storage/key_string_test.cpp
+++ b/src/mongo/db/storage/key_string_test.cpp
@@ -283,18 +283,6 @@ TEST_F(KeyStringBuilderTest, MaxElementsInCompoundKey) {
     KeyString::getKeySize(ks.getBuffer(), ks.getSize(), ALL_ASCENDING, ks.getTypeBits());
 }
 
-TEST_F(KeyStringBuilderTest, EmbeddedNullString) {
-    // Construct a KeyString where \x3c defines the type kStringLike then embedded with null
-    // characters and followed by \x00.
-    const char* data = "\x3c\x00\xff\x00";
-    const size_t size = 4;
-    KeyString::TypeBits typeBits(KeyString::Version::kLatestVersion);
-
-    // No exceptions should be thrown.
-    ASSERT_BSONOBJ_EQ(KeyString::toBson(data, size, ALL_ASCENDING, typeBits),
-                      BSON("" << StringData("\x00", 1)));
-};
-
 TEST_F(KeyStringBuilderTest, ExceededBSONDepth) {
     KeyString::Builder ks(KeyString::Version::V1);
 
@@ -758,7 +746,7 @@ TEST_F(KeyStringBuilderTest, InvalidInfinityDecimalV0) {
 
 TEST_F(KeyStringBuilderTest, ReasonableSize) {
     // Tests that KeyString::Builders do not use an excessive amount of memory for small key
-    // generation. These upper bounds were the calculated sizes of each type at the time this
+    // generation. These upper bounds were the calculate sizes of each type at the time this
     // test was written.
     KeyString::Builder stackBuilder(KeyString::Version::kLatestVersion, BSONObj(), ALL_ASCENDING);
     static_assert(sizeof(stackBuilder) <= 624);
@@ -767,13 +755,8 @@ TEST_F(KeyStringBuilderTest, ReasonableSize) {
         KeyString::Version::kLatestVersion, BSONObj(), ALL_ASCENDING);
     static_assert(sizeof(heapBuilder) <= 104);
 
-    // Use a small block size to ensure we do not use more. Additionally, the minimum allocation
-    // size is 64.
-    const auto minSize = 64;
-    SharedBufferFragmentBuilder fragmentBuilder(
-        minSize,
-        SharedBufferFragmentBuilder::DoubleGrowStrategy(
-            SharedBufferFragmentBuilder::kDefaultMaxBlockSize));
+    // Use large 1KB blocks and verify that we use way less
+    SharedBufferFragmentBuilder fragmentBuilder(1024);
     KeyString::PooledBuilder pooledBuilder(
         fragmentBuilder, KeyString::Version::kLatestVersion, BSONObj(), ALL_ASCENDING);
     static_assert(sizeof(pooledBuilder) <= 104);
@@ -793,16 +776,11 @@ TEST_F(KeyStringBuilderTest, ReasonableSize) {
 
     KeyString::Value value4 = pooledBuilder.getValueCopy();
     ASSERT_LTE(sizeof(value4), 32);
-    // This is safe because we are operating on a copy of the value and it is not shared elsewhere.
     ASSERT_LTE(value4.memUsageForSorter(), 34);
-    // We should still be using the initially-allocated size.
-    ASSERT_LTE(fragmentBuilder.memUsage(), 64);
 
-    // For values created with the pooledBuilder, it is invalid to call memUsageForSorter(). Instead
-    // we look at the mem usage of the builder itself.
     KeyString::Value value5 = pooledBuilder.release();
     ASSERT_LTE(sizeof(value5), 32);
-    ASSERT_LTE(fragmentBuilder.memUsage(), 64);
+    ASSERT_LTE(value5.memUsageForSorter(), 34);
 }
 
 TEST_F(KeyStringBuilderTest, DiscardIfNotReleased) {
diff --git a/src/mongo/db/storage/key_string_to_bson_fuzzer.cpp b/src/mongo/db/storage/key_string_to_bson_fuzzer.cpp
index 67dcc3c1f2d..3ed29c66d5b 100644
--- a/src/mongo/db/storage/key_string_to_bson_fuzzer.cpp
+++ b/src/mongo/db/storage/key_string_to_bson_fuzzer.cpp
@@ -121,17 +121,5 @@ extern "C" int LLVMFuzzerTestOneInput(const char* Data, size_t Size) {
         // We need to catch exceptions caused by invalid inputs
     }
 
-    try {
-        mongo::KeyString::decodeRecordIdLongAtEnd(&Data[2 + len], Size - (2 + len));
-    } catch (const mongo::AssertionException&) {
-        // We need to catch exceptions caused by invalid inputs
-    }
-
-    try {
-        mongo::KeyString::decodeRecordIdStrAtEnd(&Data[2 + len], Size - (2 + len));
-    } catch (const mongo::AssertionException&) {
-        // We need to catch exceptions caused by invalid inputs
-    }
-
     return 0;
 }
diff --git a/src/mongo/db/storage/kv/SConscript b/src/mongo/db/storage/kv/SConscript
index 8e53d62525f..31310e89a59 100644
--- a/src/mongo/db/storage/kv/SConscript
+++ b/src/mongo/db/storage/kv/SConscript
@@ -9,9 +9,9 @@ env.Library(
     target='kv_drop_pending_ident_reaper',
     source=['kv_drop_pending_ident_reaper.cpp'],
     LIBDEPS=[
-        '$BUILD_DIR/mongo/db/concurrency/exception_util',
         '$BUILD_DIR/mongo/db/concurrency/lock_manager',
-        '$BUILD_DIR/mongo/db/query/query_stats/query_stats',
+        '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception',
+        '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/storage/write_unit_of_work',
     ],
 )
diff --git a/src/mongo/db/storage/kv/durable_catalog_test.cpp b/src/mongo/db/storage/kv/durable_catalog_test.cpp
index 28167e375a0..14994f957e8 100644
--- a/src/mongo/db/storage/kv/durable_catalog_test.cpp
+++ b/src/mongo/db/storage/kv/durable_catalog_test.cpp
@@ -112,7 +112,8 @@ public:
             getCatalog()->getMetaData(operationContext(), catalogId),
             std::move(coll.second));
         CollectionCatalog::write(operationContext(), [&](CollectionCatalog& catalog) {
-            catalog.registerCollection(operationContext(), std::move(collection));
+            catalog.registerCollection(
+                operationContext(), options.uuid.get(), std::move(collection));
         });
 
         wuow.commit();
@@ -478,134 +479,6 @@ TEST_F(DurableCatalogTest, NoOpWhenEntireIndexAlreadySetAsMultikey) {
     }
 }
 
-class ConcurrentMultikeyTest : public DurableCatalogTest {
-public:
-    void testConcurrentMultikey(BSONObj keyPattern,
-                                const MultikeyPaths& first,
-                                const MultikeyPaths& second,
-                                const MultikeyPaths& expected) {
-        /*
-         * This test verifies that we can set multikey on two threads concurrently with the
-         * following interleaving that do not cause a WCE from the storage engine:
-         *
-         * T1: open storage snapshot
-         *
-         * T1: set multikey paths to {first}
-         *
-         * T1: commit
-         *
-         * T2: open storage snapshot
-         *
-         * T2: set multikey paths to {second}
-         *
-         * T2: commit
-         *
-         * T1: onCommit handler
-         *
-         * T2: onCommit handler
-         *
-         */
-
-        auto indexEntry = createIndex(keyPattern);
-        auto collection = getCollection();
-
-        mongo::Mutex mutex;
-        stdx::condition_variable cv;
-        int numMultikeyCalls = 0;
-
-        // Start a thread that will set multikey paths to 'first'. It will commit the change to the
-        // storage engine but block before running the onCommit handler that updates the in-memory
-        // state in the Collection instance.
-        stdx::thread t([svcCtx = getServiceContext(),
-                        &collection,
-                        &indexEntry,
-                        &first,
-                        &mutex,
-                        &cv,
-                        &numMultikeyCalls] {
-            ThreadClient client(svcCtx);
-            auto opCtx = client->makeOperationContext();
-
-            Lock::GlobalLock globalLock{opCtx.get(), MODE_IX};
-            WriteUnitOfWork wuow(opCtx.get());
-
-            // Register a onCommit that will block until the main thread has committed its multikey
-            // write. This onCommit handler is registered before any writes and will thus be
-            // performed first, blocking all other onCommit handlers.
-            opCtx->recoveryUnit()->onCommit(
-                [&mutex, &cv, &numMultikeyCalls](boost::optional<Timestamp> commitTime) {
-                    stdx::unique_lock lock(mutex);
-
-                    // Let the main thread now we have committed to the storage engine
-                    numMultikeyCalls = 1;
-                    cv.notify_all();
-
-                    // Wait until the main thread has committed its multikey write
-                    cv.wait(lock, [&numMultikeyCalls]() { return numMultikeyCalls == 2; });
-                });
-
-            // Set the index to multikey with 'first' as paths.
-            collection->setIndexIsMultikey(
-                opCtx.get(), indexEntry->descriptor()->indexName(), first);
-            wuow.commit();
-        });
-
-        // Wait for the thread above to commit its multikey write to the storage engine
-        {
-            stdx::unique_lock lock(mutex);
-            cv.wait(lock, [&numMultikeyCalls]() { return numMultikeyCalls == 1; });
-        }
-
-        // Set the index to multikey with 'second' as paths. This will not cause a WCE as the write
-        // in the thread is fully committed to the storage engine.
-        {
-            Lock::GlobalLock globalLock{operationContext(), MODE_IX};
-            // First confirm that we can observe the multikey write set by the other thread.
-            MultikeyPaths paths;
-            ASSERT_TRUE(collection->isIndexMultikey(
-                operationContext(), indexEntry->descriptor()->indexName(), &paths));
-            assertMultikeyPathsAreEqual(paths, first);
-
-            // Then perform our own multikey write.
-            WriteUnitOfWork wuow(operationContext());
-            collection->setIndexIsMultikey(
-                operationContext(), indexEntry->descriptor()->indexName(), second);
-            wuow.commit();
-        }
-
-        // Notify the thread that our multikey write is committed
-        {
-            stdx::unique_lock lock(mutex);
-            numMultikeyCalls = 2;
-            cv.notify_all();
-        }
-        t.join();
-
-        // Verify that our Collection instance has 'expected' as multikey paths for this index
-        {
-            MultikeyPaths multikeyPaths;
-            ASSERT(collection->isIndexMultikey(
-                operationContext(), indexEntry->descriptor()->indexName(), &multikeyPaths));
-            assertMultikeyPathsAreEqual(multikeyPaths, expected);
-        }
-
-        // Verify that the durable catalog has 'expected' as multikey paths for this index
-        Lock::GlobalLock globalLock{operationContext(), MODE_IS};
-        auto md = getCatalog()->getMetaData(operationContext(), collection->getCatalogId());
-
-        auto indexOffset = md->findIndexOffset(indexEntry->descriptor()->indexName());
-        assertMultikeyPathsAreEqual(md->indexes[indexOffset].multikeyPaths, expected);
-    }
-};
-
-TEST_F(ConcurrentMultikeyTest, MultikeyPathsConcurrentSecondSubset) {
-    testConcurrentMultikey(BSON("a.b" << 1), {{0U, 1U}}, {{0U}}, {{0U, 1U}});
-}
-
-TEST_F(ConcurrentMultikeyTest, MultikeyPathsConcurrentDistinct) {
-    testConcurrentMultikey(BSON("a.b" << 1), {{0U}}, {{1U}}, {{0U, 1U}});
-}
-
 TEST_F(DurableCatalogTest, SinglePhaseIndexBuild) {
     auto indexEntry = createIndex(BSON("a" << 1));
     auto collection = getCollection();
diff --git a/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper.cpp b/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper.cpp
index 24709d135ca..cd08960a53e 100644
--- a/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper.cpp
+++ b/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper.cpp
@@ -36,39 +36,19 @@
 #include <algorithm>
 
 #include "mongo/db/concurrency/d_concurrency.h"
-#include "mongo/db/concurrency/exception_util.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/storage/ident.h"
 #include "mongo/db/storage/write_unit_of_work.h"
 #include "mongo/logv2/log.h"
 
 namespace mongo {
 
-bool KVDropPendingIdentReaper::IdentInfo::isExpired(const KVEngine* engine,
-                                                    const Timestamp& ts) const {
-    return dropToken.expired() &&
-        stdx::visit(
-               visit_helper::Overloaded{[&](const Timestamp& dropTs) {
-                                            return dropTs < ts || dropTs == Timestamp::min();
-                                        },
-                                        [&](const StorageEngine::CheckpointIteration& iteration) {
-                                            return engine->hasDataBeenCheckpointed(iteration);
-                                        }},
-               dropTime);
-}
-
 KVDropPendingIdentReaper::KVDropPendingIdentReaper(KVEngine* engine) : _engine(engine) {}
 
-void KVDropPendingIdentReaper::addDropPendingIdent(
-    const stdx::variant<Timestamp, StorageEngine::CheckpointIteration>& dropTime,
-    std::shared_ptr<Ident> ident,
-    StorageEngine::DropIdentCallback&& onDrop) {
+void KVDropPendingIdentReaper::addDropPendingIdent(const Timestamp& dropTimestamp,
+                                                   std::shared_ptr<Ident> ident,
+                                                   StorageEngine::DropIdentCallback&& onDrop) {
     stdx::lock_guard<Latch> lock(_mutex);
-    auto dropTimestamp =
-        stdx::visit(visit_helper::Overloaded{[](const Timestamp& ts) { return ts; },
-                                             [](const StorageEngine::CheckpointIteration&) {
-                                                 return Timestamp::min();
-                                             }},
-                    dropTime);
     const auto equalRange = _dropPendingIdents.equal_range(dropTimestamp);
     const auto& lowerBound = equalRange.first;
     const auto& upperBound = equalRange.second;
@@ -77,25 +57,14 @@ void KVDropPendingIdentReaper::addDropPendingIdent(
         IdentInfo info;
         info.identName = ident->getIdent();
         info.dropToken = ident;
-        info.dropTime = dropTime;
         info.onDrop = std::move(onDrop);
         _dropPendingIdents.insert(std::make_pair(dropTimestamp, info));
     } else {
-        LOGV2_WARNING(8097403,
-                      "Ignoring duplicate ident drop with same drop time",
-                      "ident"_attr = ident->getIdent(),
-                      "dropTimestamp"_attr = dropTimestamp);
-    }
-}
-
-bool KVDropPendingIdentReaper::hasExpiredIdents(const Timestamp& ts) const {
-    stdx::lock_guard<Latch> lock(_mutex);
-    auto [it, end] = _dropPendingIdents.equal_range(Timestamp::min());
-    if (end != _dropPendingIdents.end()) {
-        // Include the earliest timestamped write as well.
-        end++;
+        LOGV2_FATAL_NOTRACE(51023,
+                            "Failed to add drop-pending ident, duplicate timestamp and ident pair",
+                            "ident"_attr = ident->getIdent(),
+                            "dropTimestamp"_attr = dropTimestamp);
     }
-    return std::any_of(it, end, [&](const auto& kv) { return kv.second.isExpired(_engine, ts); });
 }
 
 boost::optional<Timestamp> KVDropPendingIdentReaper::getEarliestDropTimestamp() const {
@@ -128,8 +97,7 @@ void KVDropPendingIdentReaper::dropIdentsOlderThan(OperationContext* opCtx, cons
             // This collection/index satisfies the 'ts' requirement to be safe to drop, but we must
             // also check that there are no active operations remaining that still retain a
             // reference by which to access the collection/index data.
-            const auto& info = it->second;
-            if (info.isExpired(_engine, ts)) {
+            if (it->second.dropToken.expired()) {
                 toDrop.insert(*it);
             }
         }
@@ -163,10 +131,6 @@ void KVDropPendingIdentReaper::dropIdentsOlderThan(OperationContext* opCtx, cons
                                     "error"_attr = status);
             }
             wuow.commit();
-            LOGV2(6776600,
-                  "The ident was successfully dropped",
-                  "ident"_attr = identName,
-                  "dropTimestamp"_attr = dropTimestamp);
         });
     }
 
@@ -195,12 +159,7 @@ void KVDropPendingIdentReaper::dropIdentsOlderThan(OperationContext* opCtx, cons
 
 void KVDropPendingIdentReaper::clearDropPendingState() {
     stdx::lock_guard<Latch> lock(_mutex);
-    // We only delete the timestamped drops. Non-timestamped drops cannot be rolled back, and the
-    // drops should still go through.
-    auto firstElem = std::find_if_not(_dropPendingIdents.begin(),
-                                      _dropPendingIdents.end(),
-                                      [](const auto& kv) { return kv.first == Timestamp::min(); });
-    _dropPendingIdents.erase(firstElem, _dropPendingIdents.end());
+    _dropPendingIdents.clear();
 }
 
 }  // namespace mongo
diff --git a/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper.h b/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper.h
index b9f814fe688..eda67957e9a 100644
--- a/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper.h
+++ b/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper.h
@@ -72,20 +72,16 @@ public:
     virtual ~KVDropPendingIdentReaper() = default;
 
     /**
-     * Adds a new drop-pending ident, with its drop time and namespace, to be managed by this
+     * Adds a new drop-pending ident, with its drop timestamp and namespace, to be managed by this
      * class.
      *
-     * When the drop time is old enough and no remaining operations reference the 'ident', then the
+     * When the timestamp is old enough -- the op cannot be rolled back nor new users access the
+     * record store data -- and no remaining operations reference the 'ident', then the
      * index/collection data will be safe to drop unversioned.
-     *
-     * A drop time is considered old enough when:
-     * - (Timestamp) The op cannot be rolled back nor new users access the record store data.
-     * - (CheckpointIteration) The catalog has made its changes durable.
      */
-    void addDropPendingIdent(
-        const stdx::variant<Timestamp, StorageEngine::CheckpointIteration>& dropTime,
-        std::shared_ptr<Ident> ident,
-        StorageEngine::DropIdentCallback&& onDrop = nullptr);
+    void addDropPendingIdent(const Timestamp& dropTimestamp,
+                             std::shared_ptr<Ident> ident,
+                             StorageEngine::DropIdentCallback&& onDrop = nullptr);
 
     /**
      * Returns earliest drop timestamp in '_dropPendingIdents'.
@@ -93,9 +89,6 @@ public:
      */
     boost::optional<Timestamp> getEarliestDropTimestamp() const;
 
-    // Returns whether there are expired idents
-    bool hasExpiredIdents(const Timestamp& ts) const;
-
     /**
      * Returns drop-pending idents in a sorted set.
      * Used by the storage engine during catalog reconciliation.
@@ -105,34 +98,29 @@ public:
     /**
      * Notifies this class that the storage engine has advanced its oldest timestamp.
      * Drops all unreferenced drop-pending idents with drop timestamps before 'ts', as well as all
-     * unreferenced idents with Timestamp::min() drop timestamps (untimestamped on standalones) as
-     * long as the changes have been checkpointed.
+     * unreferenced idents with Timestamp::min() drop timestamps (untimestamped on standalones).
      */
     void dropIdentsOlderThan(OperationContext* opCtx, const Timestamp& ts);
 
     /**
-     * Clears maps of drop pending idents for timestamped writes but does not drop idents in storage
-     * engine. Used by rollback before recovering to a stable timestamp.
+     * Clears maps of drop pending idents but does not drop idents in storage engine.
+     * Used by rollback after recovering to a stable timestamp.
      */
     void clearDropPendingState();
 
 private:
     // Contains information identifying what collection/index data to drop as well as determining
     // when to do so.
-    struct IdentInfo {
+    using IdentInfo = struct {
         // Identifier for the storage to drop the associated collection or index data.
         std::string identName;
 
         // The collection or index data can be safely dropped when no references to this token
-        // remain and the catalog has checkpointed the changes. The latter is mostly useful for
-        // untimestamped writes.
-        stdx::variant<Timestamp, StorageEngine::CheckpointIteration> dropTime;
-        std::weak_ptr<Ident> dropToken;
+        // remain.
+        std::weak_ptr<void> dropToken;
 
         // Callback to run once the ident has been dropped.
         StorageEngine::DropIdentCallback onDrop;
-
-        bool isExpired(const KVEngine* engine, const Timestamp& ts) const;
     };
 
     // Container type for drop-pending namespaces. We use a multimap so that we can order the
diff --git a/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper_test.cpp b/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper_test.cpp
index 1e84381c63b..4c2280c3fda 100644
--- a/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper_test.cpp
+++ b/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper_test.cpp
@@ -257,6 +257,20 @@ TEST_F(KVDropPendingIdentReaperTest,
     ASSERT_EQUALS(identName2, engine->droppedIdents.back());
 }
 
+DEATH_TEST_F(KVDropPendingIdentReaperTest,
+             AddDropPendingIdentTerminatesOnDuplicateDropTimestampAndIdent,
+             "Failed to add drop-pending ident") {
+    Timestamp dropTimestamp{Seconds(100), 0};
+
+    KVDropPendingIdentReaper reaper(getEngine());
+
+    {
+        std::shared_ptr<Ident> ident = std::make_shared<Ident>("myident");
+        reaper.addDropPendingIdent(dropTimestamp, ident);
+        reaper.addDropPendingIdent(dropTimestamp, ident);
+    }
+}
+
 TEST_F(KVDropPendingIdentReaperTest,
        DropIdentsOlderThanDropsIdentsWithDropTimestampsBeforeOldestTimestamp) {
     auto opCtx = makeOpCtx();
diff --git a/src/mongo/db/storage/kv/kv_engine.h b/src/mongo/db/storage/kv/kv_engine.h
index a68aa39fd4c..25f52f0e40f 100644
--- a/src/mongo/db/storage/kv/kv_engine.h
+++ b/src/mongo/db/storage/kv/kv_engine.h
@@ -232,15 +232,6 @@ public:
 
     virtual bool isDurable() const = 0;
 
-    virtual StorageEngine::CheckpointIteration getCheckpointIteration() const {
-        return StorageEngine::CheckpointIteration{0};
-    }
-
-    virtual bool hasDataBeenCheckpointed(
-        StorageEngine::CheckpointIteration checkpointIteration) const {
-        MONGO_UNREACHABLE;
-    }
-
     /**
      * Returns true if the KVEngine is ephemeral -- that is, it is NOT persistent and all data is
      * lost after shutdown. Otherwise, returns false.
@@ -431,23 +422,6 @@ public:
     }
 
     /**
-     * Returns the 'KeyFormat' tied to 'ident'.
-     */
-    virtual KeyFormat getKeyFormat(OperationContext* opCtx, StringData ident) const {
-        MONGO_UNREACHABLE;
-    }
-
-    /**
-     * Returns the input storage engine options, sanitized to remove options that may not apply to
-     * this node, such as encryption. Might be called for both collection and index options. See
-     * SERVER-68122.
-     */
-    virtual BSONObj getSanitizedStorageOptionsForSecondaryReplication(
-        const BSONObj& options) const {
-        return options;
-    }
-
-    /**
      * The destructor will never be called from mongod, but may be called from tests.
      * Engines may assume that this will only be called in the case of clean shutdown, even if
      * cleanShutdown() hasn't been called.
diff --git a/src/mongo/db/storage/kv/kv_engine_test_harness.cpp b/src/mongo/db/storage/kv/kv_engine_test_harness.cpp
index 2861ee689e4..de86431f25c 100644
--- a/src/mongo/db/storage/kv/kv_engine_test_harness.cpp
+++ b/src/mongo/db/storage/kv/kv_engine_test_harness.cpp
@@ -323,9 +323,6 @@ TEST_F(KVEngineTestHarness, TemporaryRecordStoreSimple) {
         ASSERT_EQUALS(1U, all.size());
         ASSERT_EQUALS(ident, all[0]);
 
-        // Dropping a collection might fail if we haven't checkpointed the data
-        engine->checkpoint();
-
         WriteUnitOfWork wuow(opCtx.get());
         ASSERT_OK(engine->dropIdent(opCtx->recoveryUnit(), ident));
         wuow.commit();
diff --git a/src/mongo/db/storage/kv/storage_engine_test.cpp b/src/mongo/db/storage/kv/storage_engine_test.cpp
index 0a7fb957003..1fbdc65188d 100644
--- a/src/mongo/db/storage/kv/storage_engine_test.cpp
+++ b/src/mongo/db/storage/kv/storage_engine_test.cpp
@@ -190,9 +190,8 @@ public:
         using TimestampType = StorageEngineImpl::TimestampMonitor::TimestampType;
         using TimestampListener = StorageEngineImpl::TimestampMonitor::TimestampListener;
         auto pf = makePromiseFuture<void>();
-        auto listener = TimestampListener(
-            TimestampType::kOldest,
-            [promise = &pf.promise](OperationContext* opCtx, Timestamp t) mutable {
+        auto listener =
+            TimestampListener(TimestampType::kOldest, [promise = &pf.promise](Timestamp t) mutable {
                 promise->emplaceValue();
             });
         timestampMonitor->addListener(&listener);
@@ -388,7 +387,6 @@ TEST_F(StorageEngineTest, ReconcileTwoPhaseIndexBuilds) {
     ASSERT_EQUALS(0UL, reconcileResult.indexBuildsToResume.size());
 }
 
-#ifndef _WIN32  // WiredTiger does not support orphan file recovery on Windows.
 TEST_F(StorageEngineRepairTest, LoadCatalogRecoversOrphans) {
     auto opCtx = cc().makeOperationContext();
 
@@ -396,8 +394,7 @@ TEST_F(StorageEngineRepairTest, LoadCatalogRecoversOrphans) {
     auto swCollInfo = createCollection(opCtx.get(), collNs);
     ASSERT_OK(swCollInfo.getStatus());
 
-    // Drop the ident from the storage engine but keep the underlying files.
-    _storageEngine->getEngine()->dropIdentForImport(opCtx.get(), swCollInfo.getValue().ident);
+    ASSERT_OK(dropIdent(opCtx.get()->recoveryUnit(), swCollInfo.getValue().ident));
     ASSERT(collectionExists(opCtx.get(), collNs));
 
     // After the catalog is reloaded, we expect that the ident has been recovered because the
@@ -413,7 +410,6 @@ TEST_F(StorageEngineRepairTest, LoadCatalogRecoversOrphans) {
     StorageRepairObserver::get(getGlobalServiceContext())->onRepairDone(opCtx.get());
     ASSERT_EQ(1U, StorageRepairObserver::get(getGlobalServiceContext())->getModifications().size());
 }
-#endif
 
 TEST_F(StorageEngineRepairTest, ReconcileSucceeds) {
     auto opCtx = cc().makeOperationContext();
@@ -588,9 +584,9 @@ TEST_F(TimestampKVEngineTest, TimestampMonitorRunning) {
 }
 
 TEST_F(TimestampKVEngineTest, TimestampListeners) {
-    TimestampListener first(stable, [](OperationContext* opCtx, Timestamp timestamp) {});
-    TimestampListener second(oldest, [](OperationContext* opCtx, Timestamp timestamp) {});
-    TimestampListener third(stable, [](OperationContext* opCtx, Timestamp timestamp) {});
+    TimestampListener first(stable, [](Timestamp timestamp) {});
+    TimestampListener second(oldest, [](Timestamp timestamp) {});
+    TimestampListener third(stable, [](Timestamp timestamp) {});
 
     // Can only register the listener once.
     _storageEngine->getTimestampMonitor()->addListener(&first);
@@ -611,7 +607,7 @@ TEST_F(TimestampKVEngineTest, TimestampMonitorNotifiesListeners) {
 
     bool changes[4] = {false, false, false, false};
 
-    TimestampListener first(checkpoint, [&](OperationContext* opCtx, Timestamp timestamp) {
+    TimestampListener first(checkpoint, [&](Timestamp timestamp) {
         stdx::lock_guard<Latch> lock(mutex);
         if (!changes[0]) {
             changes[0] = true;
@@ -619,7 +615,7 @@ TEST_F(TimestampKVEngineTest, TimestampMonitorNotifiesListeners) {
         }
     });
 
-    TimestampListener second(oldest, [&](OperationContext* opCtx, Timestamp timestamp) {
+    TimestampListener second(oldest, [&](Timestamp timestamp) {
         stdx::lock_guard<Latch> lock(mutex);
         if (!changes[1]) {
             changes[1] = true;
@@ -627,7 +623,7 @@ TEST_F(TimestampKVEngineTest, TimestampMonitorNotifiesListeners) {
         }
     });
 
-    TimestampListener third(stable, [&](OperationContext* opCtx, Timestamp timestamp) {
+    TimestampListener third(stable, [&](Timestamp timestamp) {
         stdx::lock_guard<Latch> lock(mutex);
         if (!changes[2]) {
             changes[2] = true;
@@ -635,7 +631,7 @@ TEST_F(TimestampKVEngineTest, TimestampMonitorNotifiesListeners) {
         }
     });
 
-    TimestampListener fourth(stable, [&](OperationContext* opCtx, Timestamp timestamp) {
+    TimestampListener fourth(stable, [&](Timestamp timestamp) {
         stdx::lock_guard<Latch> lock(mutex);
         if (!changes[3]) {
             changes[3] = true;
@@ -649,17 +645,15 @@ TEST_F(TimestampKVEngineTest, TimestampMonitorNotifiesListeners) {
     _storageEngine->getTimestampMonitor()->addListener(&fourth);
 
     // Wait until all 4 listeners get notified at least once.
-    {
-        stdx::unique_lock<Latch> lk(mutex);
-        cv.wait(lk, [&] {
-            for (auto const& change : changes) {
-                if (!change) {
-                    return false;
-                }
+    stdx::unique_lock<Latch> lk(mutex);
+    cv.wait(lk, [&] {
+        for (auto const& change : changes) {
+            if (!change) {
+                return false;
             }
-            return true;
-        });
-    };
+        }
+        return true;
+    });
 
     _storageEngine->getTimestampMonitor()->clearListeners();
 }
@@ -668,7 +662,7 @@ TEST_F(TimestampKVEngineTest, TimestampAdvancesOnNotification) {
     Timestamp previous = Timestamp();
     AtomicWord<int> timesNotified{0};
 
-    TimestampListener listener(stable, [&](OperationContext* opCtx, Timestamp timestamp) {
+    TimestampListener listener(stable, [&](Timestamp timestamp) {
         ASSERT_TRUE(previous < timestamp);
         previous = timestamp;
         timesNotified.fetchAndAdd(1);
@@ -684,7 +678,7 @@ TEST_F(TimestampKVEngineTest, TimestampAdvancesOnNotification) {
     _storageEngine->getTimestampMonitor()->clearListeners();
 }
 
-TEST_F(StorageEngineTestNotEphemeral, UseAlternateStorageLocation) {
+TEST_F(StorageEngineTest, UseAlternateStorageLocation) {
     auto opCtx = cc().makeOperationContext();
 
     const NamespaceString coll1Ns("db.coll1");
diff --git a/src/mongo/db/storage/record_store.h b/src/mongo/db/storage/record_store.h
index 1742cddba5e..1473cf51547 100644
--- a/src/mongo/db/storage/record_store.h
+++ b/src/mongo/db/storage/record_store.h
@@ -30,8 +30,6 @@
 #pragma once
 
 #include <boost/optional.hpp>
-#include <functional>
-#include <set>
 
 #include "mongo/bson/mutable/damage_vector.h"
 #include "mongo/db/exec/collection_scan_common.h"
@@ -417,13 +415,8 @@ public:
 
     /**
      * Prints any storage engine provided metadata for the record with 'recordId'.
-     *
-     * If provided, saves any valid timestamps (startTs, startDurableTs, stopTs, stopDurableTs)
-     * related to this record in 'recordTimestamps'.
      */
-    virtual void printRecordMetadata(OperationContext* opCtx,
-                                     const RecordId& recordId,
-                                     std::set<Timestamp>* recordTimestamps) const = 0;
+    virtual void printRecordMetadata(OperationContext* opCtx, const RecordId& recordId) const = 0;
 
     /**
      * Returns a new cursor over this record store.
diff --git a/src/mongo/db/storage/record_store_test_oplog.cpp b/src/mongo/db/storage/record_store_test_oplog.cpp
index 833477b5d5a..6c61f93ee76 100644
--- a/src/mongo/db/storage/record_store_test_oplog.cpp
+++ b/src/mongo/db/storage/record_store_test_oplog.cpp
@@ -424,7 +424,6 @@ TEST(RecordStoreTestHarness, OplogOrder) {
         auto client2 = harnessHelper->serviceContext()->makeClient("c2");
         auto opCtx = harnessHelper->newOperationContext(client2.get());
         rs->cappedTruncateAfter(opCtx.get(), id1, /*inclusive*/ false);
-        harnessHelper->getEngine()->setStableTimestamp(Timestamp(id1.getLong()), true /* force */);
     }
 
     rs->waitForAllEarlierOplogWritesToBeVisible(harnessHelper->newOperationContext().get());
diff --git a/src/mongo/db/storage/recovery_unit.h b/src/mongo/db/storage/recovery_unit.h
index 7b867906535..a67d1f090f5 100644
--- a/src/mongo/db/storage/recovery_unit.h
+++ b/src/mongo/db/storage/recovery_unit.h
@@ -37,7 +37,6 @@
 #include "mongo/bson/timestamp.h"
 #include "mongo/db/repl/read_concern_level.h"
 #include "mongo/db/storage/snapshot.h"
-#include "mongo/db/storage/storage_stats.h"
 #include "mongo/util/decorable.h"
 
 namespace mongo {
@@ -77,21 +76,36 @@ enum class PrepareConflictBehavior {
 };
 
 /**
- * DataCorruptionDetectionMode determines how we handle the discovery of evidence of data
- * corruption.
+ * Storage statistics management class, with interfaces to provide the statistics in the BSON format
+ * and an operator to add the statistics values.
  */
-enum class DataCorruptionDetectionMode {
+class StorageStats {
+    StorageStats(const StorageStats&) = delete;
+    StorageStats& operator=(const StorageStats&) = delete;
+
+public:
+    StorageStats() = default;
+
+    virtual ~StorageStats(){};
+
     /**
-     * Always throw a DataCorruptionDetected error when evidence of data corruption is detected.
+     * Provides the storage statistics in the form of a BSONObj.
      */
-    kThrow,
+    virtual BSONObj toBSON() = 0;
+
     /**
-     * When evidence of data corruption is decected, log an entry to the health log and the server
-     * logs, but do not throw an error. Continue attempting to return results.
+     * Add the statistics values.
      */
-    kLogAndContinue,
+    virtual StorageStats& operator+=(const StorageStats&) = 0;
+
+    /**
+     * Provides the ability to create an instance of this class outside of the storage integration
+     * layer.
+     */
+    virtual std::shared_ptr<StorageStats> getCopy() = 0;
 };
 
+
 /**
  * A RecoveryUnit is responsible for ensuring that data is persisted.
  * All on-disk information must be mutated through this interface.
@@ -392,10 +406,9 @@ public:
     }
 
     /**
-     * Computes the storage level statistics accrued since the last call to this function, or
-     * since the recovery unit was instantiated. Should be called at the end of each operation.
+     * Fetches the storage level statistics.
      */
-    virtual std::unique_ptr<StorageStats> computeOperationStatisticsSinceLastCall() {
+    virtual std::shared_ptr<StorageStats> getOperationStatistics() const {
         return (nullptr);
     }
 
@@ -727,14 +740,6 @@ public:
         return _noEvictionAfterRollback;
     }
 
-    void setDataCorruptionDetectionMode(DataCorruptionDetectionMode mode) {
-        _dataCorruptionDetectionMode = mode;
-    }
-
-    DataCorruptionDetectionMode getDataCorruptionDetectionMode() const {
-        return _dataCorruptionDetectionMode;
-    }
-
     /**
      * Returns true if this is an instance of RecoveryUnitNoop.
      */
@@ -742,13 +747,6 @@ public:
         return false;
     }
 
-    /**
-     * Sets a maximum timeout that the storage engine will block an operation when the cache is
-     * under pressure.
-     * If not set (default 0) then the storage engine will block indefinitely.
-     */
-    virtual void setCacheMaxWaitTimeout(Milliseconds) {}
-
 protected:
     RecoveryUnit();
 
@@ -801,8 +799,6 @@ protected:
 
     AbandonSnapshotMode _abandonSnapshotMode = AbandonSnapshotMode::kAbort;
 
-    DataCorruptionDetectionMode _dataCorruptionDetectionMode = DataCorruptionDetectionMode::kThrow;
-
 private:
     // Sets the snapshot associated with this RecoveryUnit to a new globally unique id number.
     void assignNextSnapshotId();
diff --git a/src/mongo/db/storage/snapshot_helper.cpp b/src/mongo/db/storage/snapshot_helper.cpp
index 5a173673158..1245ff9f2ad 100644
--- a/src/mongo/db/storage/snapshot_helper.cpp
+++ b/src/mongo/db/storage/snapshot_helper.cpp
@@ -117,136 +117,72 @@ bool shouldReadAtLastApplied(OperationContext* opCtx,
     }
 
     // Linearizable read concern should never be read at lastApplied, they must always read from
-    // latest and are only allowed on primaries. We are either a primary not accepting writes or
-    // secondary at this point, neither which can satisfy the noop write after the read. However, if
-    // we manage to transition to a writable primary when we do the noop write we may have read data
-    // during oplog application with kNoTimestamp which should be an error. In both cases it is OK
-    // to error with NotWritablePrimary here and we do not need to do any further checks after
-    // acquiring the snapshot because state transitions causes the repl term to increment and we
-    // can't transition directly from primary to primary catchup without a repl term increase
-    // happening.
+    // latest and are only allowed on primaries.
     if (repl::ReadConcernArgs::get(opCtx).getLevel() ==
         repl::ReadConcernLevel::kLinearizableReadConcern) {
-        uasserted(ErrorCodes::NotWritablePrimary,
-                  "cannot satisfy linearizable read concern on non-primary node");
+        if (reason) {
+            *reason = "linearizable read concern";
+        }
+        return false;
     }
 
     return true;
 }
-
 }  // namespace
 
 namespace SnapshotHelper {
 
-bool changeReadSourceIfNeeded(OperationContext* opCtx, const NamespaceString& nss) {
+ReadSourceChange shouldChangeReadSource(OperationContext* opCtx, const NamespaceString& nss) {
     std::string reason;
-    // Write to the reason string if debug logging is enabled. This avoids writing this string every
-    // time we check if we should read at last applied. This string itself is only used in logging
-    // with the same debug level as this check.
-    std::string* reasonWriter =
-        shouldLog(MONGO_LOGV2_DEFAULT_COMPONENT, logv2::LogSeverity::Debug(2)) ? &reason : nullptr;
-    bool readAtLastApplied = shouldReadAtLastApplied(opCtx, nss, reasonWriter);
+    const bool readAtLastApplied = shouldReadAtLastApplied(opCtx, nss, &reason);
 
     if (!canReadAtLastApplied(opCtx)) {
-        return readAtLastApplied;
+        return {boost::none, readAtLastApplied};
     }
 
-    const auto originalReadSource = opCtx->recoveryUnit()->getTimestampReadSource();
+    const auto existing = opCtx->recoveryUnit()->getTimestampReadSource();
     if (opCtx->recoveryUnit()->isReadSourcePinned()) {
         LOGV2_DEBUG(5863601,
                     2,
                     "Not changing readSource as it is pinned",
-                    "current"_attr = RecoveryUnit::toString(originalReadSource),
+                    "current"_attr = RecoveryUnit::toString(existing),
                     "rejected"_attr = readAtLastApplied
                         ? RecoveryUnit::toString(RecoveryUnit::ReadSource::kLastApplied)
                         : RecoveryUnit::toString(RecoveryUnit::ReadSource::kNoTimestamp));
-        return false;
-    }
-
-    // We may only change to kLastApplied if we were reading without a timestamp (or if kLastApplied
-    // is already set)
-    if (originalReadSource != RecoveryUnit::ReadSource::kNoTimestamp &&
-        originalReadSource != RecoveryUnit::ReadSource::kLastApplied) {
-        return readAtLastApplied;
+        return {boost::none, false};
     }
 
-    // Helper to set read source to the recovery unit and remember our current setting
-    auto currentReadSource = originalReadSource;
-    auto setReadSource = [&](RecoveryUnit::ReadSource readSource) {
-        opCtx->recoveryUnit()->setTimestampReadSource(readSource);
-        currentReadSource = readSource;
-    };
-
-    // Set read source based on current setting and readAtLastApplied decision.
-    if (originalReadSource == RecoveryUnit::ReadSource::kLastApplied && !readAtLastApplied) {
-        setReadSource(RecoveryUnit::ReadSource::kNoTimestamp);
-    } else if (readAtLastApplied) {
-        // Shifting from reading without a timestamp to reading with a timestamp can be
-        // dangerous because writes will appear to vanish.
-        //
-        // If a query recovers from a yield and the node is no longer primary, it must start
-        // reading at the lastApplied point because reading without a timestamp is not safe.
-        //
-        // An operation that yields a timestamped snapshot must restore a snapshot with at least
-        // as large of a timestamp, or with proper consideration of rollback scenarios, no
-        // timestamp. Given readers do not survive rollbacks, it's okay to go from reading with
-        // a timestamp to reading without one. More writes will become visible.
-        //
-        // If we already had kLastApplied as our read source then this call will refresh the
-        // timestamp.
-        setReadSource(RecoveryUnit::ReadSource::kLastApplied);
+    if (existing == RecoveryUnit::ReadSource::kNoTimestamp) {
+        // Shifting from reading without a timestamp to reading with a timestamp can be dangerous
+        // because writes will appear to vanish. This case is intended for new reads on secondaries
+        // and query yield recovery after state transitions from primary to secondary.
 
-        // We need to make sure the decision if we need to read at last applied is not changing
-        // concurrently with setting the read source with its read timestamp to the recovery unit.
-        //
-        // When the timestamp is being selected we might have transitioned into PRIMARY that is
-        // accepting writes. The lastApplied timestamp can have oplog holes behind it, in PRIMARY
-        // mode, making it unsafe as a read timestamp as concurrent writes could commit at earlier
-        // timestamps.
-        //
-        // This is handled by re-verifying the conditions if we need to read at last applied after
-        // determining the timestamp but before opening the storage snapshot. If the conditions do
-        // not match what we recorded at the beginning of the operation, we set the read source back
-        // to kNoTimestamp and read without a timestamp.
-        //
-        // The above mainly applies for Lock-free reads that is not holding the RSTL which protects
-        // against state changes.
-        reason.clear();
-        if (!shouldReadAtLastApplied(opCtx, nss, reasonWriter)) {
-            // State changed concurrently with setting the read source and we should no longer read
-            // at lastApplied.
-            setReadSource(RecoveryUnit::ReadSource::kNoTimestamp);
-            readAtLastApplied = false;
+        // If a query recovers from a yield and the node is no longer primary, it must start reading
+        // at the lastApplied point because reading without a timestamp is not safe.
+        if (readAtLastApplied) {
+            LOGV2_DEBUG(4452901, 2, "Changing ReadSource to kLastApplied", logAttrs(nss));
+            return {RecoveryUnit::ReadSource::kLastApplied, readAtLastApplied};
+        }
+    } else if (existing == RecoveryUnit::ReadSource::kLastApplied) {
+        // For some reason, we can no longer read at lastApplied.
+        // An operation that yields a timestamped snapshot must restore a snapshot with at least as
+        // large of a timestamp, or with proper consideration of rollback scenarios, no timestamp.
+        // Given readers do not survive rollbacks, it's okay to go from reading with a timestamp to
+        // reading without one. More writes will become visible.
+        if (!readAtLastApplied) {
+            LOGV2_DEBUG(4452902,
+                        2,
+                        "Changing ReadSource to kNoTimestamp",
+                        logAttrs(nss),
+                        "reason"_attr = reason);
+            // This shift to kNoTimestamp assumes that callers will not make future attempts to
+            // manipulate their ReadSources after performing reads at an un-timetamped snapshot. The
+            // only exception is callers of this function that may need to change from kNoTimestamp
+            // to kLastApplied in the event of a catalog conflict or query yield.
+            return {RecoveryUnit::ReadSource::kNoTimestamp, readAtLastApplied};
         }
     }
-
-    // All done, log if we made a change to the read source
-    if (originalReadSource == RecoveryUnit::ReadSource::kNoTimestamp &&
-        currentReadSource == RecoveryUnit::ReadSource::kLastApplied) {
-        LOGV2_DEBUG(4452901,
-                    2,
-                    "Changed ReadSource to kLastApplied",
-                    logAttrs(nss),
-                    "ts"_attr = opCtx->recoveryUnit()->getPointInTimeReadTimestamp(opCtx));
-    } else if (originalReadSource == RecoveryUnit::ReadSource::kLastApplied &&
-               currentReadSource == RecoveryUnit::ReadSource::kLastApplied) {
-        LOGV2_DEBUG(6730500,
-                    2,
-                    "ReadSource kLastApplied updated timestamp",
-                    logAttrs(nss),
-                    "ts"_attr = opCtx->recoveryUnit()->getPointInTimeReadTimestamp(opCtx));
-    } else if (originalReadSource == RecoveryUnit::ReadSource::kLastApplied &&
-               currentReadSource == RecoveryUnit::ReadSource::kNoTimestamp) {
-        LOGV2_DEBUG(4452902,
-                    2,
-                    "Changed ReadSource to kNoTimestamp",
-                    logAttrs(nss),
-                    "reason"_attr = reason);
-    }
-
-    // Return if we need to read at last applied to the caller in case further checks need to be
-    // performed.
-    return readAtLastApplied;
+    return {boost::none, readAtLastApplied};
 }
 
 bool collectionChangesConflictWithRead(boost::optional<Timestamp> collectionMin,
diff --git a/src/mongo/db/storage/snapshot_helper.h b/src/mongo/db/storage/snapshot_helper.h
index 19b383017d3..b87aebff351 100644
--- a/src/mongo/db/storage/snapshot_helper.h
+++ b/src/mongo/db/storage/snapshot_helper.h
@@ -33,15 +33,20 @@
 
 namespace mongo {
 namespace SnapshotHelper {
+struct ReadSourceChange {
+    boost::optional<RecoveryUnit::ReadSource> newReadSource;
+    bool shouldReadAtLastApplied;
+};
 
 /**
- * Changes the read source in the recovery unit if needed depending on server state. If reading at
- * last applied is needed and we already have that read source set, refresh the lastApplied
- * timestamp.
+ * Returns a ReadSourceChange containing data necessary to decide if we need to change read source.
  *
- * Returns true if the state is such that we should read at last applied, false otherwise.
+ * For Lock-Free Reads, the decisions made within this function based on replication state may
+ * become invalid after it returns and multiple calls may yield different answers. Higher level code
+ * must validate the relevance of the outcome based on replication state before and after calling
+ * this function.
  */
-bool changeReadSourceIfNeeded(OperationContext* opCtx, const NamespaceString& nss);
+ReadSourceChange shouldChangeReadSource(OperationContext* opCtx, const NamespaceString& nss);
 
 /**
  * Returns true if 'collectionMin' is not compatible with 'readTimestamp'. They are incompatible
diff --git a/src/mongo/db/storage/sorted_data_interface.h b/src/mongo/db/storage/sorted_data_interface.h
index ffe71378957..ae24d486761 100644
--- a/src/mongo/db/storage/sorted_data_interface.h
+++ b/src/mongo/db/storage/sorted_data_interface.h
@@ -169,12 +169,6 @@ public:
     virtual bool isEmpty(OperationContext* opCtx) = 0;
 
     /**
-     * Prints any storage engine provided metadata for the index entry with key 'keyString'.
-     */
-    virtual void printIndexEntryMetadata(OperationContext* opCtx,
-                                         const KeyString::Value& keyString) const = 0;
-
-    /**
      * Return the number of entries in 'this' index.
      *
      * The default implementation should be overridden with a more
diff --git a/src/mongo/db/storage/storage_engine.h b/src/mongo/db/storage/storage_engine.h
index 58a6d50f275..bfdb5745ac6 100644
--- a/src/mongo/db/storage/storage_engine.h
+++ b/src/mongo/db/storage/storage_engine.h
@@ -33,8 +33,6 @@
 #include <string>
 #include <vector>
 
-#include <boost/serialization/strong_typedef.hpp>
-
 #include "mongo/base/status.h"
 #include "mongo/bson/bsonobj.h"
 #include "mongo/bson/timestamp.h"
@@ -451,19 +449,12 @@ public:
      */
     virtual void clearDropPendingState() = 0;
 
-    BOOST_STRONG_TYPEDEF(uint64_t, CheckpointIteration);
-
     /**
      * Adds 'ident' to a list of indexes/collections whose data will be dropped when:
-     * - the 'dropTime' is sufficiently old to ensure no future data accesses
+     * - the dropTimestamp' is sufficiently old to ensure no future data accesses
      * - and no holders of 'ident' remain (the index/collection is no longer in active use)
-     *
-     * 'dropTime' can be either a CheckpointIteration or a Timestamp. In the case of a Timestamp the
-     * ident will be dropped when we can guarantee that no other operation can access the ident.
-     * CheckpointIteration should be chosen when performing untimestamped drops as they
-     * will make the ident wait for a catalog checkpoint before proceeding with the ident drop.
      */
-    virtual void addDropPendingIdent(const stdx::variant<Timestamp, CheckpointIteration>& dropTime,
+    virtual void addDropPendingIdent(const Timestamp& dropTimestamp,
                                      std::shared_ptr<Ident> ident,
                                      DropIdentCallback&& onDrop = nullptr) = 0;
 
@@ -480,23 +471,6 @@ public:
     virtual void checkpoint() = 0;
 
     /**
-     * Returns the checkpoint iteration the committed write will be part of.
-     *
-     * This token is only meaningful if obtained after a WriteUnitOfWork commit. You can use the
-     * number with StorageEngine::hasDataBeenCheckpointed(CheckpointIteration) in order to check
-     * whether the write has been checkpointed or not.
-     *
-     * Mostly of use for writes that are untimestamped. Timestamped writes should use the commit
-     * time used and the durable timestamp.
-     */
-    virtual CheckpointIteration getCheckpointIteration() const = 0;
-
-    /**
-     * Returns whether the given checkpoint iteration has been durably flushed to disk.
-     */
-    virtual bool hasDataBeenCheckpointed(CheckpointIteration checkpointIteration) const = 0;
-
-    /**
      * Recovers the storage engine state to the last stable timestamp. "Stable" in this case
      * refers to a timestamp that is guaranteed to never be rolled back. The stable timestamp
      * used should be one provided by StorageEngine::setStableTimestamp().
@@ -698,13 +672,6 @@ public:
     virtual void setPinnedOplogTimestamp(const Timestamp& pinnedTimestamp) = 0;
 
     /**
-     * Returns the input storage engine options, sanitized to remove options that may not apply to
-     * this node, such as encryption. Might be called for both collection and index options. See
-     * SERVER-68122.
-     */
-    virtual BSONObj getSanitizedStorageOptionsForSecondaryReplication(
-        const BSONObj& options) const = 0;
-    /**
      * Instructs the storage engine to dump its internal state.
      */
     virtual void dump() const = 0;
diff --git a/src/mongo/db/storage/storage_engine_impl.cpp b/src/mongo/db/storage/storage_engine_impl.cpp
index 21f04d4445a..5f5e0de5a9d 100644
--- a/src/mongo/db/storage/storage_engine_impl.cpp
+++ b/src/mongo/db/storage/storage_engine_impl.cpp
@@ -35,13 +35,11 @@
 
 #include "mongo/db/audit.h"
 #include "mongo/db/catalog/catalog_control.h"
-#include "mongo/db/catalog/clustered_collection_util.h"
 #include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/collection_catalog_helper.h"
 #include "mongo/db/catalog_raii.h"
 #include "mongo/db/client.h"
 #include "mongo/db/concurrency/d_concurrency.h"
-#include "mongo/db/concurrency/lock_state.h"
 #include "mongo/db/index_builds_coordinator.h"
 #include "mongo/db/multitenancy.h"
 #include "mongo/db/operation_context.h"
@@ -86,13 +84,10 @@ StorageEngineImpl::StorageEngineImpl(OperationContext* opCtx,
       _dropPendingIdentReaper(_engine.get()),
       _minOfCheckpointAndOldestTimestampListener(
           TimestampMonitor::TimestampType::kMinOfCheckpointAndOldest,
-          [this](OperationContext* opCtx, Timestamp timestamp) {
-              _onMinOfCheckpointAndOldestTimestampChanged(opCtx, timestamp);
-          }),
+          [this](Timestamp timestamp) { _onMinOfCheckpointAndOldestTimestampChanged(timestamp); }),
       _historicalIdentTimestampListener(
           TimestampMonitor::TimestampType::kCheckpoint,
-          [serviceContext = opCtx->getServiceContext()](OperationContext* opCtx,
-                                                        Timestamp timestamp) {
+          [serviceContext = opCtx->getServiceContext()](Timestamp timestamp) {
               HistoricalIdentTracker::get(serviceContext).removeEntriesOlderThan(timestamp);
           }),
       _supportsCappedCollections(_engine->supportsCappedCollections()) {
@@ -197,7 +192,7 @@ void StorageEngineImpl::loadCatalog(OperationContext* opCtx, LastShutdownState l
         // 'local.orphan.xxxxx' for it. However, in a nonrepair context, the orphaned idents
         // will be dropped in reconcileCatalogAndIdents().
         for (const auto& ident : identsKnownToStorageEngine) {
-            if (DurableCatalog::isCollectionIdent(ident)) {
+            if (_catalog->isCollectionIdent(ident)) {
                 bool isOrphan = !std::any_of(
                     catalogEntries.begin(),
                     catalogEntries.end(),
@@ -206,36 +201,19 @@ void StorageEngineImpl::loadCatalog(OperationContext* opCtx, LastShutdownState l
                     // If the catalog does not have information about this
                     // collection, we create an new entry for it.
                     WriteUnitOfWork wuow(opCtx);
-
-                    auto keyFormat = _engine->getKeyFormat(opCtx, ident);
-                    bool isClustered = keyFormat == KeyFormat::String;
-                    CollectionOptions optionsWithUUID;
-                    optionsWithUUID.uuid.emplace(UUID::gen());
-                    if (isClustered) {
-                        optionsWithUUID.clusteredIndex =
-                            clustered_util::makeDefaultClusteredIdIndex();
-                    }
-
-                    StatusWith<std::string> statusWithNs =
-                        _catalog->newOrphanedIdent(opCtx, ident, optionsWithUUID);
-
+                    StatusWith<std::string> statusWithNs = _catalog->newOrphanedIdent(opCtx, ident);
                     if (statusWithNs.isOK()) {
                         wuow.commit();
                         auto orphanCollNs = statusWithNs.getValue();
                         LOGV2(22247,
                               "Successfully created an entry in the catalog for orphaned "
                               "collection",
-                              "namespace"_attr = orphanCollNs,
-                              "options"_attr = optionsWithUUID);
-
-                        if (!isClustered) {
-                            // The _id index is already implicitly created on collections clustered
-                            // by _id.
-                            LOGV2_WARNING(22265,
-                                          "Collection does not have an _id index. Please manually "
-                                          "build the index",
-                                          "namespace"_attr = orphanCollNs);
-                        }
+                              "namespace"_attr = orphanCollNs);
+                        LOGV2_WARNING(22265,
+                                      "Collection does not have an _id index. Please manually "
+                                      "build the index",
+                                      "namespace"_attr = orphanCollNs);
+
                         StorageRepairObserver::get(getGlobalServiceContext())
                             ->benignModification(str::stream() << "Orphan collection created: "
                                                                << statusWithNs.getValue());
@@ -407,7 +385,7 @@ void StorageEngineImpl::_initCollection(OperationContext* opCtx,
     collection->setMinimumVisibleSnapshot(minVisibleTs);
 
     CollectionCatalog::write(opCtx, [&](CollectionCatalog& catalog) {
-        catalog.registerCollection(opCtx, std::move(collection));
+        catalog.registerCollection(opCtx, md->options.uuid.get(), std::move(collection));
     });
 }
 
@@ -614,7 +592,7 @@ StatusWith<StorageEngine::ReconcileResult> StorageEngineImpl::reconcileCatalogAn
 
         // In repair context, any orphaned collection idents from the engine should already be
         // recovered in the catalog in loadCatalog().
-        invariant(!(DurableCatalog::isCollectionIdent(it) && _options.forRepair));
+        invariant(!(_catalog->isCollectionIdent(it) && _options.forRepair));
 
         // Leave drop-pending idents alone.
         // These idents have to be retained as long as the corresponding drops are not part of a
@@ -687,19 +665,21 @@ StatusWith<StorageEngine::ReconcileResult> StorageEngineImpl::reconcileCatalogAn
                     logAttrs(nss));
             }
 
-            if (!engineIdents.count(indexIdent)) {
-                // There are cetain cases where the catalog entry may reference an index ident which
-                // is no longer present. One example of this is when an unclean shutdown occurs
-                // before a checkpoint is taken during startup recovery. Since we drop the index
-                // ident without a timestamp when restarting the index build for startup recovery,
-                // the subsequent startup recovery can see the now-dropped ident referenced by the
-                // old index catalog entry.
-                LOGV2(6386500,
-                      "Index catalog entry ident not found",
-                      "ident"_attr = indexIdent,
-                      "entry"_attr = indexMetaData.spec,
-                      logAttrs(nss));
-            }
+            // Two-phase index drop ensures that the underlying data table for an index in the
+            // catalog is not dropped until the index removal from the catalog has been majority
+            // committed and become part of the latest checkpoint. Therefore, there should almost
+            // never be a case where the index catalog entry remains but the index table (identified
+            // by ident) has been removed.
+            //
+            // There is an exception to this due to the fact that we drop the index ident without a
+            // timestamp when restarting an index build for startup recovery. Then, if we experience
+            // an unclean shutdown before a checkpoint is taken, the subsequent startup recovery can
+            // see the now-dropped ident referenced by the old index catalog entry.
+            invariant(engineIdents.find(indexIdent) != engineIdents.end() ||
+                          lastShutdownState == LastShutdownState::kUnclean,
+                      str::stream() << "Failed to find an index data table matching " << indexIdent
+                                    << " for durable index catalog entry " << indexMetaData.spec
+                                    << " in collection " << nss.ns());
 
             // Any index build with a UUID is an unfinished two-phase build and must be restarted.
             // There are no special cases to handle on primaries or secondaries. An index build may
@@ -797,16 +777,19 @@ std::string StorageEngineImpl::getFilesystemPathForDb(
 }
 
 void StorageEngineImpl::cleanShutdown() {
-    _timestampMonitor.reset();
+    if (_timestampMonitor) {
+        _timestampMonitor->clearListeners();
+    }
 
     CollectionCatalog::write(getGlobalServiceContext(), [](CollectionCatalog& catalog) {
-        catalog.onCloseCatalog();
         catalog.deregisterAllCollectionsAndViews();
     });
 
     _catalog.reset();
     _catalogRecordStore.reset();
 
+    _timestampMonitor.reset();
+
     _engine->cleanShutdown();
     // intentionally not deleting _engine
 }
@@ -901,10 +884,8 @@ Status StorageEngineImpl::_dropCollectionsNoTimestamp(OperationContext* opCtx,
 
         // No need to remove the indexes from the IndexCatalog because eliminating the Collection
         // will have the same effect.
-        auto ii = coll->getIndexCatalog()->getIndexIterator(
-            opCtx,
-            IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished |
-                IndexCatalog::InclusionPolicy::kFrozen);
+        auto ii =
+            coll->getIndexCatalog()->getIndexIterator(opCtx, true /* includeUnfinishedIndexes */);
         while (ii->more()) {
             const IndexCatalogEntry* ice = ii->next();
 
@@ -1172,37 +1153,28 @@ void StorageEngineImpl::_dumpCatalog(OperationContext* opCtx) {
     opCtx->recoveryUnit()->abandonSnapshot();
 }
 
-void StorageEngineImpl::addDropPendingIdent(
-    const stdx::variant<Timestamp, StorageEngine::CheckpointIteration>& dropTime,
-    std::shared_ptr<Ident> ident,
-    DropIdentCallback&& onDrop) {
-    _dropPendingIdentReaper.addDropPendingIdent(dropTime, ident, std::move(onDrop));
+void StorageEngineImpl::addDropPendingIdent(const Timestamp& dropTimestamp,
+                                            std::shared_ptr<Ident> ident,
+                                            DropIdentCallback&& onDrop) {
+    _dropPendingIdentReaper.addDropPendingIdent(dropTimestamp, ident, std::move(onDrop));
 }
 
 void StorageEngineImpl::checkpoint() {
     _engine->checkpoint();
 }
 
-StorageEngine::CheckpointIteration StorageEngineImpl::getCheckpointIteration() const {
-    return _engine->getCheckpointIteration();
-}
-
-bool StorageEngineImpl::hasDataBeenCheckpointed(
-    StorageEngine::CheckpointIteration checkpointIteration) const {
-    return _engine->hasDataBeenCheckpointed(checkpointIteration);
-}
+void StorageEngineImpl::_onMinOfCheckpointAndOldestTimestampChanged(const Timestamp& timestamp) {
+    // No drop-pending idents present if getEarliestDropTimestamp() returns boost::none.
+    if (auto earliestDropTimestamp = _dropPendingIdentReaper.getEarliestDropTimestamp()) {
+        if (timestamp >= *earliestDropTimestamp) {
+            LOGV2(22260,
+                  "Removing drop-pending idents with drop timestamps before timestamp",
+                  "timestamp"_attr = timestamp);
+            auto opCtx = cc().getOperationContext();
+            invariant(opCtx);
 
-void StorageEngineImpl::_onMinOfCheckpointAndOldestTimestampChanged(OperationContext* opCtx,
-                                                                    const Timestamp& timestamp) {
-    if (_dropPendingIdentReaper.hasExpiredIdents(timestamp)) {
-        LOGV2(22260,
-              "Removing drop-pending idents with drop timestamps before timestamp",
-              "timestamp"_attr = timestamp);
-
-        _dropPendingIdentReaper.dropIdentsOlderThan(opCtx, timestamp);
-    } else {
-        LOGV2_DEBUG(
-            8097401, 1, "No drop-pending idents have expired", "timestamp"_attr = timestamp);
+            _dropPendingIdentReaper.dropIdentsOlderThan(opCtx, timestamp);
+        }
     }
 }
 
@@ -1213,6 +1185,8 @@ StorageEngineImpl::TimestampMonitor::TimestampMonitor(KVEngine* engine, Periodic
 
 StorageEngineImpl::TimestampMonitor::~TimestampMonitor() {
     LOGV2(22261, "Timestamp monitor shutting down");
+    stdx::lock_guard<Latch> lock(_monitorMutex);
+    invariant(_listeners.empty());
 }
 
 void StorageEngineImpl::TimestampMonitor::_startup() {
@@ -1236,12 +1210,12 @@ void StorageEngineImpl::TimestampMonitor::_startup() {
             }
 
             try {
-                auto uniqueOpCtx = client->makeOperationContext();
-                auto opCtx = uniqueOpCtx.get();
-
-                // The TimestampMonitor is an important background cleanup task for the storage
-                // engine and needs to be able to make progress to free up resources.
-                SkipTicketAcquisitionForLock skipTicketAcquisition(opCtx);
+                auto opCtx = client->getOperationContext();
+                mongo::ServiceContext::UniqueOperationContext uOpCtx;
+                if (!opCtx) {
+                    uOpCtx = client->makeOperationContext();
+                    opCtx = uOpCtx.get();
+                }
 
                 Timestamp checkpoint;
                 Timestamp oldest;
@@ -1268,19 +1242,19 @@ void StorageEngineImpl::TimestampMonitor::_startup() {
                     stdx::lock_guard<Latch> lock(_monitorMutex);
                     for (const auto& listener : _listeners) {
                         if (listener->getType() == TimestampType::kCheckpoint) {
-                            listener->notify(opCtx, checkpoint);
+                            listener->notify(checkpoint);
                         } else if (listener->getType() == TimestampType::kOldest) {
-                            listener->notify(opCtx, oldest);
+                            listener->notify(oldest);
                         } else if (listener->getType() == TimestampType::kStable) {
-                            listener->notify(opCtx, stable);
+                            listener->notify(stable);
                         } else if (listener->getType() ==
                                    TimestampType::kMinOfCheckpointAndOldest) {
-                            listener->notify(opCtx, minOfCheckpointAndOldest);
+                            listener->notify(minOfCheckpointAndOldest);
                         } else if (stable == Timestamp::min()) {
                             // Special case notification of all listeners when writes do not have
                             // timestamps. This handles standalone mode and storage engines that
                             // don't support timestamps.
-                            listener->notify(opCtx, Timestamp::min());
+                            listener->notify(Timestamp::min());
                         }
                     }
                 }
@@ -1338,9 +1312,7 @@ int64_t StorageEngineImpl::sizeOnDiskForDb(OperationContext* opCtx,
     auto perCollectionWork = [&](const CollectionPtr& collection) {
         size += collection->getRecordStore()->storageSize(opCtx);
 
-        auto it = collection->getIndexCatalog()->getIndexIterator(
-            opCtx,
-            IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished);
+        auto it = collection->getIndexCatalog()->getIndexIterator(opCtx, true);
         while (it->more()) {
             size += _engine->getIdentSize(opCtx, it->next()->getIdent());
         }
@@ -1350,8 +1322,10 @@ int64_t StorageEngineImpl::sizeOnDiskForDb(OperationContext* opCtx,
 
     if (opCtx->isLockFreeReadsOp()) {
         auto collectionCatalog = CollectionCatalog::get(opCtx);
-        for (auto&& coll : collectionCatalog->range(tenantDbName)) {
-            perCollectionWork(coll);
+        for (auto it = collectionCatalog->begin(opCtx, tenantDbName);
+             it != collectionCatalog->end(opCtx);
+             ++it) {
+            perCollectionWork(*it);
         }
     } else {
         catalog::forEachCollectionFromDb(opCtx, tenantDbName, MODE_IS, perCollectionWork);
@@ -1385,11 +1359,6 @@ const DurableCatalog* StorageEngineImpl::getCatalog() const {
     return _catalog.get();
 }
 
-BSONObj StorageEngineImpl::getSanitizedStorageOptionsForSecondaryReplication(
-    const BSONObj& options) const {
-    return _engine->getSanitizedStorageOptionsForSecondaryReplication(options);
-}
-
 void StorageEngineImpl::dump() const {
     _engine->dump();
 }
diff --git a/src/mongo/db/storage/storage_engine_impl.h b/src/mongo/db/storage/storage_engine_impl.h
index 3af0ac8a1db..a8e36a1af88 100644
--- a/src/mongo/db/storage/storage_engine_impl.h
+++ b/src/mongo/db/storage/storage_engine_impl.h
@@ -199,7 +199,7 @@ public:
         class TimestampListener {
         public:
             // Caller must ensure that the lifetime of the variables used in the callback are valid.
-            using Callback = std::function<void(OperationContext* opCtx, Timestamp timestamp)>;
+            using Callback = std::function<void(Timestamp timestamp)>;
 
             /**
              * A TimestampListener saves a 'callback' that will be executed whenever the specified
@@ -213,15 +213,15 @@ public:
              * Executes the appropriate function with the callback of the listener with the new
              * timestamp.
              */
-            void notify(OperationContext* opCtx, Timestamp newTimestamp) {
+            void notify(Timestamp newTimestamp) {
                 if (_type == TimestampType::kCheckpoint)
-                    _onCheckpointTimestampChanged(opCtx, newTimestamp);
+                    _onCheckpointTimestampChanged(newTimestamp);
                 else if (_type == TimestampType::kOldest)
-                    _onOldestTimestampChanged(opCtx, newTimestamp);
+                    _onOldestTimestampChanged(newTimestamp);
                 else if (_type == TimestampType::kStable)
-                    _onStableTimestampChanged(opCtx, newTimestamp);
+                    _onStableTimestampChanged(newTimestamp);
                 else if (_type == TimestampType::kMinOfCheckpointAndOldest)
-                    _onMinOfCheckpointAndOldestTimestampChanged(opCtx, newTimestamp);
+                    _onMinOfCheckpointAndOldestTimestampChanged(newTimestamp);
             }
 
             TimestampType getType() const {
@@ -229,21 +229,20 @@ public:
             }
 
         private:
-            void _onCheckpointTimestampChanged(OperationContext* opCtx, Timestamp newTimestamp) {
-                _callback(opCtx, newTimestamp);
+            void _onCheckpointTimestampChanged(Timestamp newTimestamp) {
+                _callback(newTimestamp);
             }
 
-            void _onOldestTimestampChanged(OperationContext* opCtx, Timestamp newTimestamp) {
-                _callback(opCtx, newTimestamp);
+            void _onOldestTimestampChanged(Timestamp newTimestamp) {
+                _callback(newTimestamp);
             }
 
-            void _onStableTimestampChanged(OperationContext* opCtx, Timestamp newTimestamp) {
-                _callback(opCtx, newTimestamp);
+            void _onStableTimestampChanged(Timestamp newTimestamp) {
+                _callback(newTimestamp);
             }
 
-            void _onMinOfCheckpointAndOldestTimestampChanged(OperationContext* opCtx,
-                                                             Timestamp newTimestamp) {
-                _callback(opCtx, newTimestamp);
+            void _onMinOfCheckpointAndOldestTimestampChanged(Timestamp newTimestamp) {
+                _callback(newTimestamp);
             }
 
             // Timestamp type this listener monitors.
@@ -316,20 +315,14 @@ public:
         return _engine.get();
     }
 
-    void addDropPendingIdent(
-        const stdx::variant<Timestamp, StorageEngine::CheckpointIteration>& dropTime,
-        std::shared_ptr<Ident> ident,
-        DropIdentCallback&& onDrop) override;
+    void addDropPendingIdent(const Timestamp& dropTimestamp,
+                             std::shared_ptr<Ident> ident,
+                             DropIdentCallback&& onDrop) override;
 
     void startTimestampMonitor() override;
 
     void checkpoint() override;
 
-    StorageEngine::CheckpointIteration getCheckpointIteration() const override;
-
-    virtual bool hasDataBeenCheckpointed(
-        StorageEngine::CheckpointIteration checkpointIteration) const override;
-
     StatusWith<ReconcileResult> reconcileCatalogAndIdents(
         OperationContext* opCtx, LastShutdownState lastShutdownState) override;
 
@@ -374,9 +367,6 @@ public:
 
     void setPinnedOplogTimestamp(const Timestamp& pinnedTimestamp) override;
 
-    BSONObj getSanitizedStorageOptionsForSecondaryReplication(
-        const BSONObj& options) const override;
-
     void dump() const override;
 
 private:
@@ -419,8 +409,7 @@ private:
      * Called when the min of checkpoint timestamp (if exists) and oldest timestamp advances in the
      * KVEngine.
      */
-    void _onMinOfCheckpointAndOldestTimestampChanged(OperationContext* opCtx,
-                                                     const Timestamp& timestamp);
+    void _onMinOfCheckpointAndOldestTimestampChanged(const Timestamp& timestamp);
 
     /**
      * Returns whether the given ident is an internal ident and if it should be dropped or used to
diff --git a/src/mongo/db/storage/storage_engine_init.cpp b/src/mongo/db/storage/storage_engine_init.cpp
index f16cb0e3c2d..7c345ddcfe6 100644
--- a/src/mongo/db/storage/storage_engine_init.cpp
+++ b/src/mongo/db/storage/storage_engine_init.cpp
@@ -39,7 +39,6 @@
 #include "mongo/base/init.h"
 #include "mongo/bson/bsonobjbuilder.h"
 #include "mongo/db/concurrency/lock_state.h"
-#include "mongo/db/exec/scoped_timer.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/db/storage/control/storage_control.h"
 #include "mongo/db/storage/recovery_unit_noop.h"
@@ -66,10 +65,8 @@ namespace {
 void createLockFile(ServiceContext* service);
 }  // namespace
 
-StorageEngine::LastShutdownState initializeStorageEngine(
-    OperationContext* opCtx,
-    const StorageEngineInitFlags initFlags,
-    BSONObjBuilder* startupTimeElapsedBuilder) {
+StorageEngine::LastShutdownState initializeStorageEngine(OperationContext* opCtx,
+                                                         const StorageEngineInitFlags initFlags) {
     ServiceContext* service = opCtx->getServiceContext();
 
     if (storageGlobalParams.restore) {
@@ -83,10 +80,6 @@ StorageEngine::LastShutdownState initializeStorageEngine(
         invariant(!service->getStorageEngine());
 
     if ((initFlags & StorageEngineInitFlags::kAllowNoLockFile) == StorageEngineInitFlags{}) {
-        auto scopedTimer = createTimeElapsedBuilderScopedTimer(
-            service->getFastClockSource(),
-            "Create storage engine lock file in the data directory",
-            startupTimeElapsedBuilder);
         createLockFile(service);
     }
 
@@ -151,10 +144,6 @@ StorageEngine::LastShutdownState initializeStorageEngine(
 
     std::unique_ptr<StorageEngineMetadata> metadata;
     if ((initFlags & StorageEngineInitFlags::kSkipMetadataFile) == StorageEngineInitFlags{}) {
-        auto scopedTimer =
-            createTimeElapsedBuilderScopedTimer(service->getFastClockSource(),
-                                                "Get metadata describing storage engine",
-                                                startupTimeElapsedBuilder);
         metadata = StorageEngineMetadata::forPath(dbpath);
     }
 
@@ -167,10 +156,6 @@ StorageEngine::LastShutdownState initializeStorageEngine(
 
     // Validate options in metadata against current startup options.
     if (metadata.get()) {
-        auto scopedTimer = createTimeElapsedBuilderScopedTimer(
-            service->getFastClockSource(),
-            "Validate options in metadata against current startup options",
-            startupTimeElapsedBuilder);
         uassertStatusOK(factory->validateMetadata(*metadata, storageGlobalParams));
     }
 
@@ -216,26 +201,20 @@ StorageEngine::LastShutdownState initializeStorageEngine(
     });
 
     auto& lockFile = StorageEngineLockFile::get(service);
-    {
-        auto scopedTimer = createTimeElapsedBuilderScopedTimer(
-            service->getFastClockSource(), "Create storage engine", startupTimeElapsedBuilder);
-        if ((initFlags & StorageEngineInitFlags::kForRestart) == StorageEngineInitFlags{}) {
-            auto storageEngine = std::unique_ptr<StorageEngine>(
-                factory->create(opCtx, storageGlobalParams, lockFile ? &*lockFile : nullptr));
-            service->setStorageEngine(std::move(storageEngine));
-        } else {
-            auto storageEngineChangeContext = StorageEngineChangeContext::get(service);
-            auto token = storageEngineChangeContext->killOpsForStorageEngineChange(service);
-            auto storageEngine = std::unique_ptr<StorageEngine>(
-                factory->create(opCtx, storageGlobalParams, lockFile ? &*lockFile : nullptr));
-            storageEngineChangeContext->changeStorageEngine(
-                service, std::move(token), std::move(storageEngine));
-        }
+    if ((initFlags & StorageEngineInitFlags::kForRestart) == StorageEngineInitFlags{}) {
+        auto storageEngine = std::unique_ptr<StorageEngine>(
+            factory->create(opCtx, storageGlobalParams, lockFile ? &*lockFile : nullptr));
+        service->setStorageEngine(std::move(storageEngine));
+    } else {
+        auto storageEngineChangeContext = StorageEngineChangeContext::get(service);
+        auto token = storageEngineChangeContext->killOpsForStorageEngineChange(service);
+        auto storageEngine = std::unique_ptr<StorageEngine>(
+            factory->create(opCtx, storageGlobalParams, lockFile ? &*lockFile : nullptr));
+        storageEngineChangeContext->changeStorageEngine(
+            service, std::move(token), std::move(storageEngine));
     }
 
     if (lockFile) {
-        auto scopedTimer = createTimeElapsedBuilderScopedTimer(
-            service->getFastClockSource(), "Write current PID to file", startupTimeElapsedBuilder);
         uassertStatusOK(lockFile->writePid());
     }
 
@@ -243,10 +222,6 @@ StorageEngine::LastShutdownState initializeStorageEngine(
     if (!metadata.get() &&
         (initFlags & StorageEngineInitFlags::kSkipMetadataFile) == StorageEngineInitFlags{}) {
         invariant(!storageGlobalParams.readOnly);
-        auto scopedTimer =
-            createTimeElapsedBuilderScopedTimer(service->getFastClockSource(),
-                                                "Write a new metadata for storage engine",
-                                                startupTimeElapsedBuilder);
         metadata.reset(new StorageEngineMetadata(storageGlobalParams.dbpath));
         metadata->setStorageEngine(factory->getCanonicalName().toString());
         metadata->setStorageEngineOptions(factory->createMetadataOptions(storageGlobalParams));
diff --git a/src/mongo/db/storage/storage_engine_init.h b/src/mongo/db/storage/storage_engine_init.h
index b993d47ffc0..395e0446a6a 100644
--- a/src/mongo/db/storage/storage_engine_init.h
+++ b/src/mongo/db/storage/storage_engine_init.h
@@ -59,14 +59,9 @@ constexpr StorageEngineInitFlags operator|(StorageEngineInitFlags a,
 
 /**
  * Initializes the storage engine on "service".
- * The optional parameter `startupTimeElapsedBuilder` is for adding time elapsed of tasks done in
- * this function into one single builder that records the time elapsed during startup. Its default
- * value is nullptr because we only want to time this function when it is called during startup.
  */
-StorageEngine::LastShutdownState initializeStorageEngine(
-    OperationContext* opCtx,
-    StorageEngineInitFlags initFlags,
-    BSONObjBuilder* startupTimeElapsedBuilder = nullptr);
+StorageEngine::LastShutdownState initializeStorageEngine(OperationContext* opCtx,
+                                                         StorageEngineInitFlags initFlags);
 
 /**
  * Shuts down storage engine cleanly and releases any locks on mongod.lock.
diff --git a/src/mongo/db/storage/storage_engine_mock.h b/src/mongo/db/storage/storage_engine_mock.h
index 6198424e70d..d39b8b248f3 100644
--- a/src/mongo/db/storage/storage_engine_mock.h
+++ b/src/mongo/db/storage/storage_engine_mock.h
@@ -168,22 +168,13 @@ public:
     std::set<std::string> getDropPendingIdents() const final {
         return {};
     }
-    void addDropPendingIdent(
-        const stdx::variant<Timestamp, StorageEngine::CheckpointIteration>& dropTime,
-        std::shared_ptr<Ident> ident,
-        DropIdentCallback&& onDrop) final {}
+    void addDropPendingIdent(const Timestamp& dropTimestamp,
+                             std::shared_ptr<Ident> ident,
+                             DropIdentCallback&& onDrop) final {}
     void startTimestampMonitor() final {}
 
     void checkpoint() final {}
 
-    StorageEngine::CheckpointIteration getCheckpointIteration() const final {
-        return StorageEngine::CheckpointIteration{0};
-    }
-
-    bool hasDataBeenCheckpointed(StorageEngine::CheckpointIteration checkpointIteration) const {
-        return false;
-    }
-
     int64_t sizeOnDiskForDb(OperationContext* opCtx, const TenantDatabaseName& tenantDbName) final {
         return 0;
     }
@@ -217,10 +208,6 @@ public:
 
     void setPinnedOplogTimestamp(const Timestamp& pinnedTimestamp) final {}
 
-    BSONObj getSanitizedStorageOptionsForSecondaryReplication(const BSONObj& options) const final {
-        return options;
-    }
-
     void dump() const final {}
 };
 
diff --git a/src/mongo/db/storage/storage_engine_parameters.idl b/src/mongo/db/storage/storage_engine_parameters.idl
index 0a342866db0..dcd72e5d6db 100644
--- a/src/mongo/db/storage/storage_engine_parameters.idl
+++ b/src/mongo/db/storage/storage_engine_parameters.idl
@@ -52,7 +52,7 @@ server_parameters:
     # Default value being 0 means we're allowing the underlying storage engines to use their default values.
     default: 0
     validator:
-      gte: 0
+      gt: 0
 
   storageEngineConcurrentReadTransactions:
     description: "Storage Engine Concurrent Read Transactions"
@@ -65,7 +65,7 @@ server_parameters:
     # Default value being 0 means we're allowing the underlying storage engines to use their default values.
     default: 0
     validator:
-      gte: 0
+      gt: 0
 
 feature_flags:
   featureFlagEnableExecutionControl:
diff --git a/src/mongo/db/storage/storage_engine_test_fixture.h b/src/mongo/db/storage/storage_engine_test_fixture.h
index cfb329992eb..db8f499f42b 100644
--- a/src/mongo/db/storage/storage_engine_test_fixture.h
+++ b/src/mongo/db/storage/storage_engine_test_fixture.h
@@ -76,7 +76,7 @@ public:
             _storageEngine->getCatalog()->getMetaData(opCtx, catalogId),
             std::move(rs));
         CollectionCatalog::write(opCtx, [&](CollectionCatalog& catalog) {
-            catalog.registerCollection(opCtx, std::move(coll));
+            catalog.registerCollection(opCtx, options.uuid.get(), std::move(coll));
         });
 
         return {{_storageEngine->getCatalog()->getEntry(catalogId)}};
@@ -183,10 +183,7 @@ public:
         Collection* collection =
             CollectionCatalog::get(opCtx)->lookupCollectionByNamespaceForMetadataWrite(opCtx,
                                                                                        collNs);
-        auto descriptor = collection->getIndexCatalog()->findIndexByName(
-            opCtx,
-            key,
-            IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished);
+        auto descriptor = collection->getIndexCatalog()->findIndexByName(opCtx, key, true);
         collection->indexBuildSuccess(opCtx, descriptor->getEntry());
     }
 
@@ -202,8 +199,9 @@ public:
 
 class StorageEngineRepairTest : public StorageEngineTest {
 public:
+    // TODO (SERVER-65191): Use wiredTiger.
     StorageEngineRepairTest()
-        : StorageEngineTest(Options{}.repair(RepairAction::kRepair).ephemeral(false)) {}
+        : StorageEngineTest(Options{}.engine("ephemeralForTest").repair(RepairAction::kRepair)) {}
 
     void tearDown() {
         auto repairObserver = StorageRepairObserver::get(getGlobalServiceContext());
@@ -221,9 +219,4 @@ public:
     }
 };
 
-class StorageEngineTestNotEphemeral : public StorageEngineTest {
-public:
-    StorageEngineTestNotEphemeral() : StorageEngineTest(Options{}.ephemeral(false)){};
-};
-
 }  // namespace mongo
diff --git a/src/mongo/db/storage/storage_options.h b/src/mongo/db/storage/storage_options.h
index 5b254748247..8e95c4c934f 100644
--- a/src/mongo/db/storage/storage_options.h
+++ b/src/mongo/db/storage/storage_options.h
@@ -82,9 +82,6 @@ struct StorageGlobalParams {
 
     bool dur;  // --dur durability (now --journal)
 
-    // Whether the Storage Engine selected should be ephemeral in nature or not.
-    bool ephemeral = false;
-
     // --journalCommitInterval
     static constexpr int kMaxJournalCommitIntervalMs = 500;
     AtomicWord<int> journalCommitIntervalMs;
diff --git a/src/mongo/db/storage/storage_parameters.idl b/src/mongo/db/storage/storage_parameters.idl
index 678f5365ccc..c48fd47a7fa 100644
--- a/src/mongo/db/storage/storage_parameters.idl
+++ b/src/mongo/db/storage/storage_parameters.idl
@@ -114,14 +114,6 @@ server_parameters:
         cpp_vartype: bool
         default: false
 
-    skipDroppingHashedShardKeyIndex:
-        description: >-
-            Skips dropping hashed shard key supporting indexes when calling dropIndexes with the '*' parameter. Used for testing purposes.
-        set_at: [ startup, runtime ]
-        cpp_vartype: bool
-        cpp_varname: skipDroppingHashedShardKeyIndex
-        default: false
-
 feature_flags:
     featureFlagClusteredIndexes:
         description: "When enabled, support non time-series collections with clustered indexes"
diff --git a/src/mongo/db/storage/storage_stats.h b/src/mongo/db/storage/storage_stats.h
deleted file mode 100644
index 0c7ea1ca179..00000000000
--- a/src/mongo/db/storage/storage_stats.h
+++ /dev/null
@@ -1,61 +0,0 @@
-/**
- *    Copyright (C) 2022-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/bson/bsonobj.h"
-
-namespace mongo {
-
-/**
- * Manages statistics from the storage engine, allowing addition of statistics and serialization to
- * BSON.
- */
-class StorageStats {
-public:
-    // This is a pure virtual class, so the constructors will never be called directly, and slicing
-    // should not be an issue.
-    StorageStats() = default;
-    StorageStats(const StorageStats&) = default;
-    StorageStats(StorageStats&&) = default;
-
-    StorageStats& operator=(const StorageStats&) = delete;
-    StorageStats& operator=(StorageStats&&) = delete;
-
-    virtual ~StorageStats() = default;
-
-    virtual BSONObj toBSON() const = 0;
-
-    virtual std::unique_ptr<StorageStats> clone() const = 0;
-
-    virtual StorageStats& operator+=(const StorageStats&) = 0;
-    virtual StorageStats& operator-=(const StorageStats&) = 0;
-};
-
-}  // namespace mongo
diff --git a/src/mongo/db/storage/storage_util.cpp b/src/mongo/db/storage/storage_util.cpp
index 30623a36aa0..0dace14bfd2 100644
--- a/src/mongo/db/storage/storage_util.cpp
+++ b/src/mongo/db/storage/storage_util.cpp
@@ -47,49 +47,40 @@
 namespace mongo {
 namespace catalog {
 namespace {
-auto removeEmptyDirectory =
-    [](ServiceContext* svcCtx, StorageEngine* storageEngine, const NamespaceString& ns) {
-        // Nothing to do if not using directoryperdb or there are still collections in the database.
-        // If we don't support supportsPendingDrops then this is executing before the collection is
-        // removed from the catalog. In that case, just blindly attempt to delete the directory, it
-        // will only succeed if it is empty which is the behavior we want.
-        auto collectionCatalog = CollectionCatalog::get(svcCtx);
-        const TenantDatabaseName tenantDbName(boost::none, ns.db());
-        if (!storageEngine->isUsingDirectoryPerDb() ||
-            (storageEngine->supportsPendingDrops() &&
-             !collectionCatalog->range(tenantDbName).empty())) {
-            return;
-        }
+auto removeEmptyDirectory = [](ServiceContext* svcCtx,
+                               StorageEngine* storageEngine,
+                               const NamespaceString& ns) {
+    // Nothing to do if not using directoryperdb or there are still collections in the database.
+    // If we don't support supportsPendingDrops then this is executing before the collection is
+    // removed from the catalog. In that case, just blindly attempt to delete the directory, it
+    // will only succeed if it is empty which is the behavior we want.
+    auto collectionCatalog = CollectionCatalog::get(svcCtx);
+    const TenantDatabaseName tenantDbName(boost::none, ns.db());
+    if (!storageEngine->isUsingDirectoryPerDb() ||
+        (storageEngine->supportsPendingDrops() &&
+         collectionCatalog->begin(nullptr, tenantDbName) != collectionCatalog->end(nullptr))) {
+        return;
+    }
 
-        boost::system::error_code ec;
-        boost::filesystem::remove(storageEngine->getFilesystemPathForDb(tenantDbName), ec);
-
-        if (!ec) {
-            LOGV2(4888200, "Removed empty database directory", "db"_attr = tenantDbName.dbName());
-        } else if (collectionCatalog->range(tenantDbName).empty()) {
-            // It is possible for a new collection to be created in the database between when we
-            // check whether the database is empty and actually attempting to remove the directory.
-            // In this case, don't log that the removal failed because it is expected. However,
-            // since we attempt to remove the directory for both the collection and index ident
-            // drops, once the database is empty it will be still logged until the final of these
-            // ident drops occurs.
-            LOGV2_DEBUG(4888201,
-                        1,
-                        "Failed to remove database directory",
-                        "db"_attr = tenantDbName.dbName(),
-                        "error"_attr = ec.message());
-        }
-    };
-
-BSONObj toBSON(const stdx::variant<Timestamp, StorageEngine::CheckpointIteration>& x) {
-    return stdx::visit(visit_helper::Overloaded{[](const Timestamp& ts) { return ts.toBSON(); },
-                                                [](const StorageEngine::CheckpointIteration& iter) {
-                                                    auto underlyingValue = uint64_t{iter};
-                                                    return BSON("checkpointIteration"
-                                                                << std::to_string(underlyingValue));
-                                                }},
-                       x);
-}
+    boost::system::error_code ec;
+    boost::filesystem::remove(storageEngine->getFilesystemPathForDb(tenantDbName), ec);
+
+    if (!ec) {
+        LOGV2(4888200, "Removed empty database directory", "db"_attr = tenantDbName.dbName());
+    } else if (collectionCatalog->begin(nullptr, tenantDbName) == collectionCatalog->end(nullptr)) {
+        // It is possible for a new collection to be created in the database between when we
+        // check whether the database is empty and actually attempting to remove the directory.
+        // In this case, don't log that the removal failed because it is expected. However,
+        // since we attempt to remove the directory for both the collection and index ident
+        // drops, once the database is empty it will be still logged until the final of these
+        // ident drops occurs.
+        LOGV2_DEBUG(4888201,
+                    1,
+                    "Failed to remove database directory",
+                    "db"_attr = tenantDbName.dbName(),
+                    "error"_attr = ec.message());
+    }
+};
 }  // namespace
 
 void removeIndex(OperationContext* opCtx,
@@ -132,13 +123,9 @@ void removeIndex(OperationContext* opCtx,
         };
 
         if (storageEngine->supportsPendingDrops()) {
-            stdx::variant<Timestamp, StorageEngine::CheckpointIteration> dropTime;
             if (!commitTimestamp) {
-                // Standalone mode and unreplicated drops will not provide a timestamp. Use the
-                // checkpoint iteration instead.
-                dropTime = storageEngine->getEngine()->getCheckpointIteration();
-            } else {
-                dropTime = *commitTimestamp;
+                // Standalone mode will not provide a timestamp.
+                commitTimestamp = Timestamp::min();
             }
             LOGV2(22206,
                   "Deferring table drop for index",
@@ -146,8 +133,8 @@ void removeIndex(OperationContext* opCtx,
                   logAttrs(nss),
                   "uuid"_attr = uuid,
                   "ident"_attr = ident->getIdent(),
-                  "dropTime"_attr = toBSON(dropTime));
-            storageEngine->addDropPendingIdent(dropTime, ident, std::move(onDrop));
+                  "commitTimestamp"_attr = commitTimestamp);
+            storageEngine->addDropPendingIdent(*commitTimestamp, ident, std::move(onDrop));
         } else {
             // Intentionally ignoring failure here. Since we've removed the metadata pointing to
             // the collection, we should never see it again anyway.
@@ -190,20 +177,16 @@ Status dropCollection(OperationContext* opCtx,
             };
 
             if (storageEngine->supportsPendingDrops()) {
-                stdx::variant<Timestamp, StorageEngine::CheckpointIteration> dropTime;
                 if (!commitTimestamp) {
-                    // Standalone mode and unreplicated drops will not provide a timestamp. Use the
-                    // checkpoint iteration instead.
-                    dropTime = storageEngine->getEngine()->getCheckpointIteration();
-                } else {
-                    dropTime = *commitTimestamp;
+                    // Standalone mode will not provide a timestamp.
+                    commitTimestamp = Timestamp::min();
                 }
                 LOGV2(22214,
                       "Deferring table drop for collection",
                       logAttrs(nss),
                       "ident"_attr = ident->getIdent(),
-                      "dropTime"_attr = toBSON(dropTime));
-                storageEngine->addDropPendingIdent(dropTime, ident, std::move(onDrop));
+                      "commitTimestamp"_attr = commitTimestamp);
+                storageEngine->addDropPendingIdent(*commitTimestamp, ident, std::move(onDrop));
             } else {
                 // Intentionally ignoring failure here. Since we've removed the metadata pointing to
                 // the collection, we should never see it again anyway.
diff --git a/src/mongo/db/storage/wiredtiger/SConscript b/src/mongo/db/storage/wiredtiger/SConscript
index 1b7790e61c6..9d9f5ca18f4 100644
--- a/src/mongo/db/storage/wiredtiger/SConscript
+++ b/src/mongo/db/storage/wiredtiger/SConscript
@@ -35,7 +35,6 @@ wtEnv.Library(
         'wiredtiger_global_options.cpp',
         'wiredtiger_index.cpp',
         'wiredtiger_kv_engine.cpp',
-        'wiredtiger_stats.cpp',
         'wiredtiger_oplog_manager.cpp',
         'wiredtiger_parameters.cpp',
         'wiredtiger_prepare_conflict.cpp',
@@ -53,11 +52,12 @@ wtEnv.Library(
         '$BUILD_DIR/mongo/db/catalog/collection',
         '$BUILD_DIR/mongo/db/catalog/collection_options',
         '$BUILD_DIR/mongo/db/concurrency/lock_manager',
+        '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception',
+        '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/global_settings',
-        '$BUILD_DIR/mongo/db/index/index_access_method',
+        '$BUILD_DIR/mongo/db/index/index_descriptor',
         '$BUILD_DIR/mongo/db/namespace_string',
         '$BUILD_DIR/mongo/db/prepare_conflict_tracker',
-        '$BUILD_DIR/mongo/db/query/query_stats/query_stats',
         '$BUILD_DIR/mongo/db/record_id_helpers',
         '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface',
         '$BUILD_DIR/mongo/db/repl/repl_settings',
@@ -80,9 +80,7 @@ wtEnv.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/catalog/database_holder',
-        '$BUILD_DIR/mongo/db/catalog/health_log_interface',
         '$BUILD_DIR/mongo/db/commands/server_status',
-        '$BUILD_DIR/mongo/db/concurrency/exception_util',
         '$BUILD_DIR/mongo/db/db_raii',
         '$BUILD_DIR/mongo/db/mongod_options',
         '$BUILD_DIR/mongo/db/multitenancy',
@@ -137,15 +135,13 @@ wtEnv.CppUnitTest(
     source=[
         'wiredtiger_init_test.cpp',
         'wiredtiger_kv_engine_test.cpp',
-        'wiredtiger_stats_test.cpp',
         'wiredtiger_recovery_unit_test.cpp',
         'wiredtiger_session_cache_test.cpp',
-        'wiredtiger_size_storer_test.cpp',
         'wiredtiger_util_test.cpp',
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/db/auth/authmocks',
-        '$BUILD_DIR/mongo/db/index/index_access_method',
+        '$BUILD_DIR/mongo/db/index/index_access_methods',
         '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface',
         '$BUILD_DIR/mongo/db/repl/replmocks',
         '$BUILD_DIR/mongo/db/service_context',
diff --git a/src/mongo/db/storage/wiredtiger/oplog_stone_parameters.idl b/src/mongo/db/storage/wiredtiger/oplog_stone_parameters.idl
index 17af7e126d9..54afd12971c 100644
--- a/src/mongo/db/storage/wiredtiger/oplog_stone_parameters.idl
+++ b/src/mongo/db/storage/wiredtiger/oplog_stone_parameters.idl
@@ -64,10 +64,3 @@ server_parameters:
         cpp_varname: gOplogSamplingLogIntervalSeconds
         default: 10
         validator: { gte: 0 }
-    oplogTruncationCheckPeriodSeconds:
-        description: 'The number of seconds the oplog truncation thread wakes up periodically to check and truncate oplog.'
-        set_at: [ startup ]
-        cpp_vartype: 'int'
-        cpp_varname: gOplogTruncationCheckPeriodSeconds
-        default: 300
-        validator: { gt: 300 }
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp
index 7aa20bf0960..cd9a25c2930 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp
@@ -38,16 +38,12 @@
 #include <set>
 
 #include "mongo/base/checked_cast.h"
-#include "mongo/base/string_data.h"
-#include "mongo/db/catalog/health_log.h"
-#include "mongo/db/catalog/health_log_gen.h"
 #include "mongo/db/catalog/index_catalog_entry.h"
 #include "mongo/db/catalog/validate_results.h"
 #include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/global_settings.h"
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/json.h"
-#include "mongo/db/namespace_string.h"
 #include "mongo/db/repl/repl_settings.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/stats/resource_consumption_metrics.h"
@@ -63,7 +59,6 @@
 #include "mongo/util/assert_util.h"
 #include "mongo/util/fail_point.h"
 #include "mongo/util/hex.h"
-#include "mongo/util/stacktrace.h"
 #include "mongo/util/str.h"
 #include "mongo/util/testing_proctor.h"
 
@@ -88,54 +83,8 @@ namespace {
 
 MONGO_FAIL_POINT_DEFINE(WTCompactIndexEBUSY);
 MONGO_FAIL_POINT_DEFINE(WTIndexPauseAfterSearchNear);
-MONGO_FAIL_POINT_DEFINE(WTValidateIndexStructuralDamage);
-MONGO_FAIL_POINT_DEFINE(WTIndexUassertDuplicateRecordForKeyOnIdUnindex);
 
 static const WiredTigerItem emptyItem(nullptr, 0);
-
-/**
- * Add a data corruption entry to the health log.
- */
-void addDataCorruptionEntryToHealthLog(OperationContext* opCtx,
-                                       const NamespaceString& nss,
-                                       StringData operation,
-                                       StringData message,
-                                       const BSONObj& key,
-                                       StringData indexName,
-                                       StringData uri) {
-    HealthLogEntry entry;
-    entry.setNss(nss);
-    entry.setTimestamp(Date_t::now());
-    entry.setSeverity(SeverityEnum::Error);
-    entry.setScope(ScopeEnum::Index);
-    entry.setOperation(operation);
-    entry.setMsg(message);
-
-    BSONObjBuilder bob;
-    bob.append("key", key);
-    bob.append("indexName", indexName);
-    bob.append("uri", uri);
-    bob.appendElements(getStackTrace().getBSONRepresentation());
-    entry.setData(bob.obj());
-
-    HealthLog::get(opCtx)->log(entry);
-}
-
-/**
- * Returns the logv2::LogOptions controlling the behaviour after logging a data corruption
- * error. When the TestingProctor is enabled we will fatally assert. When the testing proctor is
- * disabled or when 'forceUassert' is specified (for instance because a failpoint is enabled),
- * we should log and throw DataCorruptionDetected.
- */
-logv2::LogOptions getLogOptionsForDataCorruption(RecoveryUnit& ru, bool forceUassert = false) {
-    if (ru.getDataCorruptionDetectionMode() == DataCorruptionDetectionMode::kThrow ||
-        MONGO_unlikely(forceUassert)) {
-        return logv2::LogOptions{logv2::UserAssertAfterLog(ErrorCodes::DataCorruptionDetected)};
-    } else {
-        return logv2::LogOptions(logv2::LogComponent::kAutomaticDetermination);
-    }
-}
-
 }  // namespace
 
 void WiredTigerIndex::setKey(WT_CURSOR* cursor, const WT_ITEM* item) {
@@ -153,7 +102,7 @@ void WiredTigerIndex::getKey(OperationContext* opCtx, WT_CURSOR* cursor, WT_ITEM
 StatusWith<std::string> WiredTigerIndex::parseIndexOptions(const BSONObj& options) {
     StringBuilder ss;
     BSONForEach(elem, options) {
-        if (elem.fieldNameStringData() == WiredTigerUtil::kConfigStringField) {
+        if (elem.fieldNameStringData() == "configString") {
             Status status = WiredTigerUtil::checkTableCreationOptions(elem);
             if (!status.isOK()) {
                 return status;
@@ -380,15 +329,6 @@ void WiredTigerIndex::fullValidate(OperationContext* opCtx,
                                    IndexValidateResults* fullResults) const {
     dassert(opCtx->lockState()->isReadLocked());
     if (fullResults && !WiredTigerRecoveryUnit::get(opCtx)->getSessionCache()->isEphemeral()) {
-        if (WTValidateIndexStructuralDamage.shouldFail()) {
-            std::string msg = str::stream() << "verify() returned an error. "
-                                            << "This indicates structural damage. "
-                                            << "Not examining individual index entries.";
-            fullResults->errors.push_back(msg);
-            fullResults->valid = false;
-            return;
-        }
-
         int err = WiredTigerUtil::verifyTable(opCtx, _uri, &(fullResults->errors));
         if (err == EBUSY) {
             std::string msg = str::stream()
@@ -416,25 +356,25 @@ void WiredTigerIndex::fullValidate(OperationContext* opCtx,
         }
     }
 
+    auto cursor = newCursor(opCtx);
+    long long count = 0;
+    LOGV2_TRACE_INDEX(20094, "fullValidate");
+
+    const auto requestedInfo = TRACING_ENABLED ? Cursor::kKeyAndLoc : Cursor::kJustExistance;
+
+    KeyString::Value keyStringForSeek =
+        IndexEntryComparison::makeKeyStringFromBSONKeyForSeek(BSONObj(),
+                                                              getKeyStringVersion(),
+                                                              getOrdering(),
+                                                              true, /* forward */
+                                                              true  /* inclusive */
+        );
+
+    for (auto kv = cursor->seek(keyStringForSeek, requestedInfo); kv; kv = cursor->next()) {
+        LOGV2_TRACE_INDEX(20095, "fullValidate {kv}", "kv"_attr = kv);
+        count++;
+    }
     if (numKeysOut) {
-        auto cursor = newCursor(opCtx);
-        long long count = 0;
-        LOGV2_TRACE_INDEX(20094, "fullValidate");
-
-        const auto requestedInfo = TRACING_ENABLED ? Cursor::kKeyAndLoc : Cursor::kJustExistance;
-
-        KeyString::Value keyStringForSeek =
-            IndexEntryComparison::makeKeyStringFromBSONKeyForSeek(BSONObj(),
-                                                                  getKeyStringVersion(),
-                                                                  getOrdering(),
-                                                                  true, /* forward */
-                                                                  true  /* inclusive */
-            );
-
-        for (auto kv = cursor->seek(keyStringForSeek, requestedInfo); kv; kv = cursor->next()) {
-            LOGV2_TRACE_INDEX(20095, "fullValidate {kv}", "kv"_attr = kv);
-            count++;
-        }
         *numKeysOut = count;
     }
 }
@@ -505,74 +445,6 @@ bool WiredTigerIndex::isEmpty(OperationContext* opCtx) {
     return false;
 }
 
-void WiredTigerIndex::printIndexEntryMetadata(OperationContext* opCtx,
-                                              const KeyString::Value& keyString) const {
-    // Printing the index entry metadata requires a new session. We cannot open other cursors when
-    // there are open history store cursors in the session. We also need to make sure that the
-    // existing session has not written data to avoid potential deadlocks.
-    invariant(!opCtx->lockState()->inAWriteUnitOfWork());
-    WiredTigerSession session(WiredTigerRecoveryUnit::get(opCtx)->getSessionCache()->conn());
-
-    // Per the version cursor API:
-    // - A version cursor can only be called with the read timestamp as the oldest timestamp.
-    // - If there is no oldest timestamp, the version cursor can only be called with a read
-    //   timestamp of 1.
-    // - If there is an oldest timestamp, reading at timestamp 1 will get rounded up.
-    const std::string config = "read_timestamp=1,roundup_timestamps=(read=true)";
-    WiredTigerBeginTxnBlock beginTxn(session.getSession(), config.c_str());
-
-    // Open a version cursor. This is a debug cursor that enables iteration through the history of
-    // values for a given index entry.
-    WT_CURSOR* cursor = session.getNewCursor(_uri, "debug=(dump_version=true)");
-
-    const WiredTigerItem searchKey(keyString.getBuffer(), keyString.getSize());
-    cursor->set_key(cursor, searchKey.Get());
-
-    int ret = cursor->search(cursor);
-    while (ret != WT_NOTFOUND) {
-        invariantWTOK(ret, cursor->session);
-
-        uint64_t startTs = 0, startDurableTs = 0, stopTs = 0, stopDurableTs = 0;
-        uint64_t startTxnId = 0, stopTxnId = 0;
-        uint8_t flags = 0, location = 0, prepare = 0, type = 0;
-        WT_ITEM value;
-
-        invariantWTOK(cursor->get_value(cursor,
-                                        &startTxnId,
-                                        &startTs,
-                                        &startDurableTs,
-                                        &stopTxnId,
-                                        &stopTs,
-                                        &stopDurableTs,
-                                        &type,
-                                        &prepare,
-                                        &flags,
-                                        &location,
-                                        &value),
-                      cursor->session);
-
-        auto indexKey = KeyString::toBson(
-            keyString.getBuffer(), keyString.getSize(), _ordering, keyString.getTypeBits());
-
-        LOGV2(6601200,
-              "WiredTiger index entry metadata",
-              "keyString"_attr = keyString,
-              "indexKey"_attr = indexKey,
-              "startTxnId"_attr = startTxnId,
-              "startTs"_attr = Timestamp(startTs),
-              "startDurableTs"_attr = Timestamp(startDurableTs),
-              "stopTxnId"_attr = stopTxnId,
-              "stopTs"_attr = Timestamp(stopTs),
-              "stopDurableTs"_attr = Timestamp(stopDurableTs),
-              "type"_attr = type,
-              "prepare"_attr = prepare,
-              "flags"_attr = flags,
-              "location"_attr = location);
-
-        ret = cursor->next(cursor);
-    }
-}
-
 long long WiredTigerIndex::getSpaceUsedBytes(OperationContext* opCtx) const {
     dassert(opCtx->lockState()->isReadLocked());
     auto ru = WiredTigerRecoveryUnit::get(opCtx);
@@ -1320,12 +1192,16 @@ protected:
 
             LOGV2_TRACE_CURSOR(5683900, "cmp after advance: {cmp}", "cmp"_attr = cmp);
 
+            // We do not expect any exact matches or matches of prefixes by comparing keys of
+            // different lengths. Callers either seek using keys with discriminators that always
+            // compare unequally, or in the case of restoring a cursor, perform exact searches. In
+            // the case of an exact search, we will have returned earlier.
+            dassert(cmp);
+
             if (enforcingPrepareConflicts) {
                 // If we are enforcing prepare conflicts, calling next() or prev() must always give
                 // us a key that compares, respectively, greater than or less than our search key.
-                // An exact match is also possible in the case of _id indexes, because the recordid
-                // is not a part of the key.
-                dassert(_forward ? cmp >= 0 : cmp <= 0);
+                dassert(_forward ? cmp > 0 : cmp < 0);
             }
         }
 
@@ -1560,24 +1436,14 @@ private:
         _typeBits.resetFromBuffer(&br);
 
         if (!br.atEof()) {
-            const auto bsonKey = redact(curr(kWantKey)->key);
-            const auto collectionNamespace = _idx.getCollectionNamespace(_opCtx);
-            addDataCorruptionEntryToHealthLog(
-                _opCtx,
-                collectionNamespace,
-                "WiredTigerIndexUniqueCursor::_updateIdAndTypeBitsFromValue",
-                "Unique index cursor seeing multiple records for key in index",
-                bsonKey,
-                _idx.indexName(),
-                _idx.uri());
-
-            LOGV2_ERROR_OPTIONS(7623202,
-                                getLogOptionsForDataCorruption(*_opCtx->recoveryUnit()),
-                                "Unique index cursor seeing multiple records for key in index",
-                                "key"_attr = bsonKey,
-                                "index"_attr = _idx.indexName(),
-                                "uri"_attr = _idx.uri(),
-                                logAttrs(collectionNamespace));
+            LOGV2_FATAL(28608,
+                        "Unique index cursor seeing multiple records for key {key} in index "
+                        "{index} ({uri}) belonging to collection {collection}",
+                        "Unique index cursor seeing multiple records for key in index",
+                        "key"_attr = redact(curr(kWantKey)->key),
+                        "index"_attr = _idx.indexName(),
+                        "uri"_attr = _idx.uri(),
+                        "collection"_attr = _idx.getCollectionNamespace(_opCtx));
         }
     }
 };
@@ -1605,25 +1471,12 @@ public:
         _typeBits.resetFromBuffer(&br);
 
         if (!br.atEof()) {
-            const auto bsonKey = redact(curr(kWantKey)->key);
-            const auto collectionNamespace = _idx.getCollectionNamespace(_opCtx);
-
-            addDataCorruptionEntryToHealthLog(
-                _opCtx,
-                collectionNamespace,
-                "WiredTigerIdIndexCursor::updateIdAndTypeBits",
-                "Index cursor seeing multiple records for key in _id index",
-                bsonKey,
-                _idx.indexName(),
-                _idx.uri());
-
-            LOGV2_ERROR_OPTIONS(5176200,
-                                getLogOptionsForDataCorruption(*_opCtx->recoveryUnit()),
-                                "Index cursor seeing multiple records for key in _id index",
-                                "key"_attr = bsonKey,
-                                "index"_attr = _idx.indexName(),
-                                "uri"_attr = _idx.uri(),
-                                logAttrs(collectionNamespace));
+            LOGV2_FATAL(5176200,
+                        "Index cursor seeing multiple records for key in _id index",
+                        "key"_attr = redact(curr(kWantKey)->key),
+                        "index"_attr = _idx.indexName(),
+                        "uri"_attr = _idx.uri(),
+                        "collection"_attr = _idx.getCollectionNamespace(_opCtx));
         }
     }
 };
@@ -1846,11 +1699,9 @@ void WiredTigerIdIndex::_unindex(OperationContext* opCtx,
     WiredTigerItem keyItem(keyString.getBuffer(), sizeWithoutRecordId);
     setKey(c, keyItem.Get());
 
-    const auto failWithDataCorruptionForTest =
-        WTIndexUassertDuplicateRecordForKeyOnIdUnindex.shouldFail();
     // On the _id index, the RecordId is stored in the value of the index entry. If the dupsAllowed
     // flag is not set, we blindly delete using only the key without checking the RecordId.
-    if (!dupsAllowed && MONGO_likely(!failWithDataCorruptionForTest)) {
+    if (!dupsAllowed) {
         int ret = WT_OP_CHECK(wiredTigerCursorRemove(opCtx, c));
         if (ret == WT_NOTFOUND) {
             return;
@@ -1882,26 +1733,14 @@ void WiredTigerIdIndex::_unindex(OperationContext* opCtx,
 
     RecordId idInIndex = KeyString::decodeRecordIdLong(&br);
     KeyString::TypeBits typeBits = KeyString::TypeBits::fromBuffer(getKeyStringVersion(), &br);
-    if (!br.atEof() || MONGO_unlikely(failWithDataCorruptionForTest)) {
+    if (!br.atEof()) {
         auto bsonKey = KeyString::toBson(keyString, _ordering);
-        const auto collectionNamespace = getCollectionNamespace(opCtx);
-
-        addDataCorruptionEntryToHealthLog(opCtx,
-                                          collectionNamespace,
-                                          "WiredTigerIdIndex::_unindex",
-                                          "Un-index seeing multiple records for key",
-                                          bsonKey,
-                                          _indexName,
-                                          _uri);
-
-        LOGV2_ERROR_OPTIONS(
-            5176201,
-            getLogOptionsForDataCorruption(*opCtx->recoveryUnit(), failWithDataCorruptionForTest),
-            "Un-index seeing multiple records for key",
-            "key"_attr = bsonKey,
-            "index"_attr = _indexName,
-            "uri"_attr = _uri,
-            logAttrs(collectionNamespace));
+        LOGV2_FATAL(5176201,
+                    "Un-index seeing multiple records for key",
+                    "key"_attr = bsonKey,
+                    "index"_attr = _desc->indexName(),
+                    "uri"_attr = _uri,
+                    "collection"_attr = getCollectionNamespace(opCtx));
     }
 
     // The RecordId matches, so remove the entry.
@@ -1947,74 +1786,16 @@ void WiredTigerIndexUnique::_unindex(OperationContext* opCtx,
         return;
     }
 
-    // WT_NOTFOUND is possible if index key is in old (v4.0) format. Retry removal of key using old
-    // format.
-    _unindexTimestampUnsafe(opCtx, c, keyString, dupsAllowed);
-}
-
-void WiredTigerIndexUnique::_unindexTimestampUnsafe(OperationContext* opCtx,
-                                                    WT_CURSOR* c,
-                                                    const KeyString::Value& keyString,
-                                                    bool dupsAllowed) {
-    // The old unique index format had a key-value of indexKey-RecordId. This means that the
-    // RecordId in an index entry might not match the indexKey+RecordId keyString passed into this
-    // function: an index on a field where multiple collection documents have the same field value
-    // but only one passes the partial index filter.
-    //
-    // The dupsAllowed flag is no longer relevant for the old unique index format. No new index
-    // entries are written in the old format, let alone during temporary phases of the server when
-    // duplicates are allowed.
-
-    const RecordId id =
-        KeyString::decodeRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize());
-    invariant(id.isValid());
-
+    // After a rolling upgrade an index can have keys from both timestamp unsafe (old) and
+    // timestamp safe (new) unique indexes. Old format keys just had the index key while new
+    // format key has index key + Record id. WT_NOTFOUND is possible if index key is in old format.
+    // Retry removal of key using old format.
     auto sizeWithoutRecordId =
         KeyString::sizeWithoutRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize());
     WiredTigerItem keyItem(keyString.getBuffer(), sizeWithoutRecordId);
     setKey(c, keyItem.Get());
 
-    if (_partial) {
-        int ret = wiredTigerPrepareConflictRetry(opCtx, [&] { return c->search(c); });
-        if (ret == WT_NOTFOUND) {
-            return;
-        }
-        invariantWTOK(ret, c->session);
-
-        WT_ITEM value;
-        invariantWTOK(c->get_value(c, &value), c->session);
-        BufReader br(value.data, value.size);
-        fassert(40416, br.remaining());
-
-        // Check that the record id matches. We may be called to unindex records that are not
-        // present in the index due to the partial filter expression.
-        bool foundRecord = [&]() {
-            if (KeyString::decodeRecordIdLong(&br) != id) {
-                return false;
-            }
-            return true;
-        }();
-
-        // Ensure the index entry value is not a list of RecordIds, which should only be possible
-        // temporarily in v4.0 when dupsAllowed is true, not ever across upgrades or in upgraded
-        // versions.
-        KeyString::TypeBits::fromBuffer(getKeyStringVersion(), &br);
-        if (br.remaining()) {
-            LOGV2_FATAL_NOTRACE(
-                7592201,
-                "An index entry was found that contains an unexpected old format that should no "
-                "longer exist. The index should be dropped and rebuilt.",
-                "indexName"_attr = _indexName,
-                "uri"_attr = uri(),
-                "collection"_attr = getCollectionNamespace(opCtx));
-        }
-
-        if (!foundRecord) {
-            return;
-        }
-    }
-
-    int ret = WT_OP_CHECK(wiredTigerCursorRemove(opCtx, c));
+    ret = WT_OP_CHECK(wiredTigerCursorRemove(opCtx, c));
     if (ret == WT_NOTFOUND) {
         return;
     }
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_index.h b/src/mongo/db/storage/wiredtiger/wiredtiger_index.h
index 8fccc7d8c24..5a94980c22c 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_index.h
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_index.h
@@ -156,9 +156,6 @@ public:
 
     virtual Status initAsEmpty(OperationContext* opCtx);
 
-    virtual void printIndexEntryMetadata(OperationContext* opCtx,
-                                         const KeyString::Value& keyString) const;
-
     Status compact(OperationContext* opCtx) override;
 
     const std::string& uri() const {
@@ -298,16 +295,6 @@ protected:
                   const KeyString::Value& keyString,
                   bool dupsAllowed) override;
 
-    /**
-     * This function continues to exist in order to support v4.0 unique partial index format: the
-     * format changed in v4.2 and onward. _unindex will call this if an index entry in the new
-     * format cannot be found, and this function will check for the old format.
-     */
-    void _unindexTimestampUnsafe(OperationContext* opCtx,
-                                 WT_CURSOR* c,
-                                 const KeyString::Value& keyString,
-                                 bool dupsAllowed);
-
 private:
     bool _partial;
 };
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_init.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_init.cpp
index a093fb70661..da82969596e 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_init.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_init.cpp
@@ -116,6 +116,7 @@ public:
                               "RAM. See http://dochub.mongodb.org/core/faq-memory-diagnostics-wt");
             }
         }
+        const bool ephemeral = false;
         auto kv =
             std::make_unique<WiredTigerKVEngine>(getCanonicalName().toString(),
                                                  params.dbpath,
@@ -124,7 +125,7 @@ public:
                                                  cacheMB,
                                                  wiredTigerGlobalOptions.getMaxHistoryFileSizeMB(),
                                                  params.dur,
-                                                 params.ephemeral,
+                                                 ephemeral,
                                                  params.repair,
                                                  params.readOnly);
         kv->setRecordStoreExtraOptions(wiredTigerGlobalOptions.collectionConfig);
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
index ed1ec49e90f..2c5a6ed5559 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
@@ -472,12 +472,6 @@ WiredTigerKVEngine::WiredTigerKVEngine(const std::string& canonicalName,
         ss << WiredTigerUtil::generateRestoreConfig() << ",";
     }
 
-    // If we've requested an ephemeral instance we store everything into memory instead of backing
-    // it onto disk. Logging is not supported in this instance, thus we also have to disable it.
-    if (_ephemeral) {
-        ss << "in_memory=true,log=(enabled=false),";
-    }
-
     string config = ss.str();
     LOGV2(22315, "Opening WiredTiger", "config"_attr = config);
     auto startTime = Date_t::now();
@@ -725,6 +719,7 @@ void WiredTigerKVEngine::_openWiredTiger(const std::string& path, const std::str
 
 void WiredTigerKVEngine::cleanShutdown() {
     LOGV2(22317, "WiredTigerKVEngine shutting down");
+    WiredTigerUtil::resetTableLoggingInfo();
 
     if (!_conn) {
         return;
@@ -1064,30 +1059,6 @@ std::deque<std::string> getUniqueFiles(const std::vector<std::string>& files,
     return result;
 }
 
-/**
- * Normalizes ident names with and without 'directoryPerDb' and 'wiredTigerDirectoryForIndexes'
- * mode.
- *
- * The durable catalog can return idents in four forms:
- *  - <db_name>/<collection|index>/<ident_identifier>
- *    - directoryPerDb + wiredTigerDirectoryForIndexes
- *  - <db_name>/<ident_name>
- *    - directoryPerDb
- *  - <collection|index>/<ident_identifier>
- *    - wiredTigerDirectoryForIndexes
- *  - <ident_name>
- *    - default, no options enabled
- *
- * ident_identifier: <counter>-<random number>
- * ident_name: <collection|index>-<ident_identifier>
- *
- * This function trims the leading directory names leaving only the ident's unique identifier.
- */
-inline std::string getIdentStem(const std::string& ident) {
-    boost::filesystem::path identPath(ident);
-    return identPath.stem().string();
-}
-
 class StreamingCursorImpl : public StorageEngine::StreamingCursor {
 public:
     StreamingCursorImpl() = delete;
@@ -1197,22 +1168,11 @@ private:
         int wtRet;
         bool fileUnchangedFlag = false;
         if (!_wtBackup->dupCursor) {
-            size_t attempt = 0;
-            do {
-                wtRet = _session->open_cursor(
-                    _session, nullptr, _wtBackup->cursor, config.c_str(), &_wtBackup->dupCursor);
-
-                if (wtRet == EBUSY) {
-                    logAndBackoff(8927900,
-                                  ::mongo::logv2::LogComponent::kStorage,
-                                  logv2::LogSeverity::Debug(1),
-                                  ++attempt,
-                                  "Opening duplicate backup cursor returned EBUSY, retrying",
-                                  "config"_attr = config);
-                } else if (wtRet != 0) {
-                    return wtRCToStatus(wtRet, _session);
-                }
-            } while (wtRet == EBUSY);
+            wtRet = (_session)->open_cursor(
+                _session, nullptr, _wtBackup->cursor, config.c_str(), &_wtBackup->dupCursor);
+            if (wtRet != 0) {
+                return wtRCToStatus(wtRet, _session);
+            }
             fileUnchangedFlag = true;
         }
 
@@ -1345,14 +1305,11 @@ WiredTigerKVEngine::beginNonBlockingBackup(OperationContext* opCtx,
         for (const DurableCatalog::Entry& e : catalogEntries) {
             // Populate the collection ident with its namespace and UUID.
             UUID uuid = catalog->getMetaData(opCtx, e.catalogId)->options.uuid.get();
-            std::string collectionIdent = getIdentStem(e.ident);
-            _wtBackup.identToNamespaceAndUUIDMap.emplace(collectionIdent,
-                                                         std::make_pair(e.nss, uuid));
+            _wtBackup.identToNamespaceAndUUIDMap.emplace(e.ident, std::make_pair(e.nss, uuid));
 
             // Populate the collection's index idents with the collection's namespace and UUID.
             std::vector<std::string> idxIdents = catalog->getIndexIdents(opCtx, e.catalogId);
-            for (const std::string& idxIdentFull : idxIdents) {
-                std::string idxIdent = getIdentStem(idxIdentFull);
+            for (const std::string& idxIdent : idxIdents) {
                 _wtBackup.identToNamespaceAndUUIDMap.emplace(idxIdent, std::make_pair(e.nss, uuid));
             }
         }
@@ -1441,15 +1398,10 @@ void WiredTigerKVEngine::syncSizeInfo(bool sync) const {
     if (!_sizeStorer)
         return;
 
-    while (true) {
-        try {
-            return _sizeStorer->flush(sync);
-        } catch (const WriteConflictException&) {
-            if (!sync) {
-                // ignore, we'll try again later.
-                return;
-            }
-        }
+    try {
+        _sizeStorer->flush(sync);
+    } catch (const WriteConflictException&) {
+        // ignore, we'll try again later.
     }
 }
 
@@ -1897,8 +1849,8 @@ Status WiredTigerKVEngine::dropIdent(RecoveryUnit* ru,
 
     WiredTigerSession session(_conn);
 
-    int ret =
-        session.getSession()->drop(session.getSession(), uri.c_str(), "checkpoint_wait=false");
+    int ret = session.getSession()->drop(
+        session.getSession(), uri.c_str(), "force,checkpoint_wait=false");
     LOGV2_DEBUG(22338, 1, "WT drop", "uri"_attr = uri, "ret"_attr = ret);
 
     if (ret == EBUSY) {
@@ -1911,16 +1863,11 @@ Status WiredTigerKVEngine::dropIdent(RecoveryUnit* ru,
         return Status::OK();
     }
 
-    if (DurableCatalog::isCollectionIdent(ident)) {
-        _sizeStorer->remove(uri);
-    }
-
     if (onDrop) {
         onDrop();
     }
 
     if (ret == ENOENT) {
-        // Ident doesn't exist, it is effectively dropped.
         return Status::OK();
     }
 
@@ -1940,7 +1887,7 @@ void WiredTigerKVEngine::dropIdentForImport(OperationContext* opCtx, StringData
     // cursor is open. In short, using "checkpoint_wait=false" and "lock_wait=true" means that we
     // can potentially be waiting for a short period of time for WT_SESSION::drop() to run, but
     // would rather get EBUSY than wait a long time for a checkpoint to complete.
-    const std::string config = "checkpoint_wait=false,lock_wait=true,remove_files=false";
+    const std::string config = "force=true,checkpoint_wait=false,lock_wait=true,remove_files=false";
     int ret = 0;
     size_t attempt = 0;
     do {
@@ -1961,10 +1908,6 @@ void WiredTigerKVEngine::dropIdentForImport(OperationContext* opCtx, StringData
                       "config"_attr = config,
                       "ret"_attr = ret);
     } while (ret == EBUSY);
-    if (ret == ENOENT) {
-        // If the ident doesn't exist then it has already been dropped.
-        return;
-    }
     invariantWTOK(ret, session.getSession());
 }
 
@@ -2041,17 +1984,14 @@ void WiredTigerKVEngine::dropSomeQueuedIdents() {
             _identToDrop.pop_front();
         }
         int ret = session.getSession()->drop(
-            session.getSession(), identToDrop.uri.c_str(), "checkpoint_wait=false");
+            session.getSession(), identToDrop.uri.c_str(), "force,checkpoint_wait=false");
         LOGV2_DEBUG(22340, 1, "WT queued drop", "uri"_attr = identToDrop.uri, "ret"_attr = ret);
 
         if (ret == EBUSY) {
             stdx::lock_guard<Latch> lk(_identToDropMutex);
             _identToDrop.push_back(std::move(identToDrop));
         } else {
-            if (ret != ENOENT) {
-                // Ident doesn't exist, it is effectively dropped. The error is safe to ignore.
-                invariantWTOK(ret, session.getSession());
-            }
+            invariantWTOK(ret, session.getSession());
             if (identToDrop.callback) {
                 identToDrop.callback();
             }
@@ -2063,26 +2003,7 @@ bool WiredTigerKVEngine::supportsDirectoryPerDB() const {
     return true;
 }
 
-void WiredTigerKVEngine::_checkpoint(WT_SESSION* session, bool useTimestamp) {
-    _currentCheckpointIteration.fetchAndAdd(1);
-    if (useTimestamp) {
-        invariantWTOK(session->checkpoint(session, "use_timestamp=true"), session);
-    } else {
-        invariantWTOK(session->checkpoint(session, "use_timestamp=false"), session);
-    }
-    auto checkpointedIteration = _finishedCheckpointIteration.fetchAndAdd(1);
-    LOGV2_FOR_RECOVERY(8097402,
-                       2,
-                       "Finished checkpoint, updated iteration counter",
-                       "checkpointIteration"_attr = checkpointedIteration);
-}
-
 void WiredTigerKVEngine::_checkpoint(WT_SESSION* session) {
-    // Ephemeral WiredTiger instances cannot do a checkpoint to disk as there is no disk backing
-    // the data.
-    if (_ephemeral) {
-        return;
-    }
     // TODO: SERVER-64507: Investigate whether we can smartly rely on one checkpointer if two or
     // more threads checkpoint at the same time.
     stdx::lock_guard lk(_checkpointMutex);
@@ -2118,7 +2039,7 @@ void WiredTigerKVEngine::_checkpoint(WT_SESSION* session) {
         // Third, stableTimestamp >= initialDataTimestamp: Take stable checkpoint. Steady state
         // case.
         if (initialDataTimestamp.asULL() <= 1) {
-            _checkpoint(session, /*useTimestamp=*/false);
+            invariantWTOK(session->checkpoint(session, "use_timestamp=false"), session);
             LOGV2_FOR_RECOVERY(5576602,
                                2,
                                "Completed unstable checkpoint.",
@@ -2139,7 +2060,7 @@ void WiredTigerKVEngine::_checkpoint(WT_SESSION* session) {
                                "stableTimestamp"_attr = stableTimestamp,
                                "oplogNeededForRollback"_attr = toString(oplogNeededForRollback));
 
-            _checkpoint(session, /*useTimestamp=*/true);
+            invariantWTOK(session->checkpoint(session, "use_timestamp=true"), session);
 
             if (oplogNeededForRollback.isOK()) {
                 // Now that the checkpoint is durable, publish the oplog needed to recover from it.
@@ -2159,12 +2080,6 @@ void WiredTigerKVEngine::checkpoint() {
     return _checkpoint(s);
 }
 
-void WiredTigerKVEngine::forceCheckpoint(bool useStableTimestamp) {
-    UniqueWiredTigerSession session = _sessionCache->getSession();
-    WT_SESSION* s = session->getSession();
-    return _checkpoint(s, useStableTimestamp);
-}
-
 bool WiredTigerKVEngine::hasIdent(OperationContext* opCtx, StringData ident) const {
     return _hasUri(WiredTigerRecoveryUnit::get(opCtx)->getSession()->getSession(), _uri(ident));
 }
@@ -2509,9 +2424,6 @@ StatusWith<Timestamp> WiredTigerKVEngine::recoverToStableTimestamp(OperationCont
                        "initialDataTimestamp"_attr = initialDataTimestamp);
     int ret = 0;
 
-    // Shut down the cache before rollback and restart afterwards.
-    _sessionCache->shuttingDown();
-
     // The rollback_to_stable operation requires all open cursors to be closed or reset before the
     // call, otherwise EBUSY will be returned. Occasionally, there could be an operation that hasn't
     // been killed yet, such as the CappedInsertNotifier for a yielded oplog getMore. We will retry
@@ -2545,9 +2457,6 @@ StatusWith<Timestamp> WiredTigerKVEngine::recoverToStableTimestamp(OperationCont
 
     _sizeStorer = std::make_unique<WiredTigerSizeStorer>(_conn, _sizeStorerUri, _readOnly);
 
-    // SERVER-85167: restart the cache after resetting the size storer.
-    _sessionCache->restart();
-
     return {stableTimestamp};
 }
 
@@ -2837,22 +2746,4 @@ Status WiredTigerKVEngine::reconfigureLogging() {
     return wtRCToStatus(_conn->reconfigure(_conn, verboseConfig.c_str()), nullptr);
 }
 
-KeyFormat WiredTigerKVEngine::getKeyFormat(OperationContext* opCtx, StringData ident) const {
-
-    const std::string wtTableConfig =
-        uassertStatusOK(WiredTigerUtil::getMetadataCreate(opCtx, "table:{}"_format(ident)));
-    return wtTableConfig.find("key_format=u") != string::npos ? KeyFormat::String : KeyFormat::Long;
-}
-
-BSONObj WiredTigerKVEngine::getSanitizedStorageOptionsForSecondaryReplication(
-    const BSONObj& options) const {
-
-    // Skip inMemory storage engine, encryption at rest only applies to storage backed engine.
-    if (_ephemeral) {
-        return options;
-    }
-
-    return WiredTigerUtil::getSanitizedStorageOptionsForSecondaryReplication(options);
-}
-
 }  // namespace mongo
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h
index 24af566c560..688db855b74 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h
@@ -127,18 +127,6 @@ public:
         return _durable;
     }
 
-    // Force a WT checkpoint, this will not update internal timestamps.
-    void forceCheckpoint(bool useStableTimestamp);
-
-    StorageEngine::CheckpointIteration getCheckpointIteration() const override {
-        return StorageEngine::CheckpointIteration{_currentCheckpointIteration.load()};
-    }
-
-    bool hasDataBeenCheckpointed(
-        StorageEngine::CheckpointIteration checkpointIteration) const override {
-        return _ephemeral || _finishedCheckpointIteration.load() > checkpointIteration;
-    }
-
     bool isEphemeral() const override {
         return _ephemeral;
     }
@@ -395,11 +383,6 @@ public:
 
     Status reconfigureLogging() override;
 
-    KeyFormat getKeyFormat(OperationContext* opCtx, StringData ident) const override;
-
-    BSONObj getSanitizedStorageOptionsForSecondaryReplication(
-        const BSONObj& options) const override;
-
 private:
     class WiredTigerSessionSweeper;
 
@@ -410,8 +393,6 @@ private:
 
     void _checkpoint(WT_SESSION* session);
 
-    void _checkpoint(WT_SESSION* session, bool useTimestamp);
-
     /**
      * Opens a connection on the WiredTiger database 'path' with the configuration 'wtOpenConfig'.
      * Only returns when successful. Intializes both '_conn' and '_fileVersion'.
@@ -544,16 +525,5 @@ private:
     // checkpoint. WT has a mutex of its own to only have one checkpoint active at all times so this
     // is only to protect our internal updates.
     Mutex _checkpointMutex = MONGO_MAKE_LATCH("WiredTigerKVEngine::_checkpointMutex");
-
-    // Counters used for computing whether a checkpointIteration has lapsed or not.
-    //
-    // We use two counters because one isn't sufficient to prove correctness. With two counters we
-    // first increase the first one in order to inform later operations that they will be part of
-    // the next checkpoint. The second one is there to inform waiters on whether they've
-    // successfully been checkpointed or not.
-    //
-    // This is valid because durability is a state all operations will converge to eventually.
-    AtomicWord<std::uint64_t> _currentCheckpointIteration{0};
-    AtomicWord<std::uint64_t> _finishedCheckpointIteration{0};
 };
 }  // namespace mongo
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine_test.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine_test.cpp
index 4a56d8ece58..9917d95e0f2 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine_test.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine_test.cpp
@@ -170,10 +170,6 @@ TEST_F(WiredTigerKVEngineRepairTest, OrphanedDataFilesCanBeRecovered) {
         _engine->recoverOrphanedIdent(opCtxPtr.get(), nss, ident, defaultCollectionOptions);
     ASSERT_EQ(ErrorCodes::CommandNotSupported, status.code());
 #else
-
-    // Dropping a collection might fail if we haven't checkpointed the data.
-    _engine->checkpoint();
-
     // Move the data file out of the way so the ident can be dropped. This not permitted on Windows
     // because the file cannot be moved while it is open. The implementation for orphan recovery is
     // also not implemented on Windows for this reason.
@@ -223,9 +219,6 @@ TEST_F(WiredTigerKVEngineRepairTest, UnrecoverableOrphanedDataFilesAreRebuilt) {
 
     ASSERT(boost::filesystem::exists(*dataFilePath));
 
-    // Dropping a collection might fail if we haven't checkpointed the data
-    _engine->checkpoint();
-
     ASSERT_OK(_engine->dropIdent(opCtxPtr.get()->recoveryUnit(), ident));
 
 #ifdef _WIN32
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_oplog_manager.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_oplog_manager.cpp
index 1be18b4c8ff..fb3bc211faa 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_oplog_manager.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_oplog_manager.cpp
@@ -237,6 +237,8 @@ void WiredTigerOplogManager::_updateOplogVisibilityLoop(WiredTigerSessionCache*
         invariant(_triggerOplogVisibilityUpdate);
         _triggerOplogVisibilityUpdate = false;
 
+        lk.unlock();
+
         // Fetch the all_durable timestamp from the storage engine, which is guaranteed not to have
         // any holes behind it in-memory.
         const uint64_t newTimestamp = sessionCache->getKVEngine()->getAllDurableTimestamp().asULL();
@@ -252,6 +254,7 @@ void WiredTigerOplogManager::_updateOplogVisibilityLoop(WiredTigerSessionCache*
             continue;
         }
 
+        lk.lock();
         // Publish the new timestamp value. Avoid going backward.
         auto currentVisibleTimestamp = getOplogReadTimestamp();
         if (newTimestamp > currentVisibleTimestamp) {
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.cpp
index 9fcde57485c..ace8dd90cf8 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.cpp
@@ -44,8 +44,8 @@
 #include "mongo/base/static_assert.h"
 #include "mongo/bson/util/builder.h"
 #include "mongo/db/catalog/validate_results.h"
-#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/concurrency/locker.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/global_settings.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/operation_context.h"
@@ -134,7 +134,6 @@ std::size_t computeRecordIdSize(const RecordId& id) {
 }  // namespace
 
 MONGO_FAIL_POINT_DEFINE(WTCompactRecordStoreEBUSY);
-MONGO_FAIL_POINT_DEFINE(WTRecordStoreUassertOutOfOrder);
 MONGO_FAIL_POINT_DEFINE(WTWriteConflictException);
 MONGO_FAIL_POINT_DEFINE(WTWriteConflictExceptionForReads);
 MONGO_FAIL_POINT_DEFINE(slowOplogSamplingReads);
@@ -265,7 +264,7 @@ void WiredTigerRecordStore::OplogStones::awaitHasExcessStonesOrDead() {
                 }
             }
         }
-        _oplogReclaimCv.wait_for(lock, stdx::chrono::seconds{gOplogTruncationCheckPeriodSeconds});
+        _oplogReclaimCv.wait(lock);
     }
 }
 
@@ -649,7 +648,7 @@ void WiredTigerRecordStore::OplogStones::adjust(int64_t maxSize) {
 StatusWith<std::string> WiredTigerRecordStore::parseOptionsField(const BSONObj options) {
     StringBuilder ss;
     BSONForEach(elem, options) {
-        if (elem.fieldNameStringData() == WiredTigerUtil::kConfigStringField) {
+        if (elem.fieldNameStringData() == "configString") {
             Status status = WiredTigerUtil::checkTableCreationOptions(elem);
             if (!status.isOK()) {
                 return status;
@@ -677,15 +676,6 @@ public:
             // On destruction, we must always handle freeing the underlying raw WT_CURSOR pointer.
             _saveStorageCursorOnDetachFromOperationContext = false;
 
-            // Shutdown does not wait for any threads running queries to be interrupted and exit.
-            // In addition, the RandomCursor destructor doesn't hold any global lock so we need to
-            // check if the server is shutting down to avoid calling into the storage engine, whose
-            // connection may have already been closed.
-            Status interruptStatus = _opCtx->checkForInterruptNoAssert();
-            if (interruptStatus.code() == ErrorCodes::InterruptedAtShutdown) {
-                return;
-            }
-
             detachFromOperationContext();
         }
     }
@@ -811,11 +801,7 @@ StatusWith<std::string> WiredTigerRecordStore::generateCreateString(
             ident.startsWith("internal-") ||
             // TODO (SERVER-60753): Remove special handling for index build during recovery. This
             // includes the following _mdb_catalog ident.
-            nss == NamespaceString::kIndexBuildEntryNamespace ||
-            // SERVER-68330: Reconstructing config.transactions after a rollback does a mixed-mode
-            // write.
-            nss == NamespaceString::kSessionTransactionsTableNamespace ||
-            ident.startsWith("_mdb_catalog")) {
+            nss == NamespaceString::kIndexBuildEntryNamespace || ident.startsWith("_mdb_catalog")) {
             ss << "write_timestamp_usage=mixed_mode,";
         } else {
             ss << "write_timestamp_usage=ordered,";
@@ -959,7 +945,7 @@ WiredTigerRecordStore::WiredTigerRecordStore(WiredTigerKVEngine* kvEngine,
     // case for temporary RecordStores (those not associated with any collection) and in unit
     // tests. Persistent size information is not required in either case. If a RecordStore needs
     // persistent size information, we require it to use a SizeStorer.
-    _sizeInfo = _sizeStorer ? _sizeStorer->load(_uri)
+    _sizeInfo = _sizeStorer ? _sizeStorer->load(ctx, _uri)
                             : std::make_shared<WiredTigerSizeStorer::SizeInfo>(0, 0);
 }
 
@@ -1060,8 +1046,7 @@ bool WiredTigerRecordStore::inShutdown() const {
 }
 
 long long WiredTigerRecordStore::dataSize(OperationContext* opCtx) const {
-    auto dataSize = _sizeInfo->dataSize.load();
-    return dataSize > 0 ? dataSize : 0;
+    return _sizeInfo->dataSize.load();
 }
 
 long long WiredTigerRecordStore::numRecords(OperationContext* opCtx) const {
@@ -1172,7 +1157,8 @@ void WiredTigerRecordStore::doDeleteRecord(OperationContext* opCtx, const Record
     auto keyLength = computeRecordIdSize(id);
     metricsCollector.incrementOneDocWritten(old_length + keyLength);
 
-    _changeNumRecordsAndDataSize(opCtx, -1, -old_length);
+    _changeNumRecords(opCtx, -1);
+    _increaseDataSize(opCtx, -old_length);
 }
 
 Timestamp WiredTigerRecordStore::getPinnedOplog() const {
@@ -1189,7 +1175,8 @@ bool WiredTigerRecordStore::yieldAndAwaitOplogDeletionRequest(OperationContext*
 
     // Release any locks before waiting on the condition variable. It is illegal to access any
     // methods or members of this record store after this line because it could be deleted.
-    locker->saveLockStateAndUnlock(&snapshot);
+    bool releasedAnyLocks = locker->saveLockStateAndUnlock(&snapshot);
+    invariant(releasedAnyLocks);
 
     // The top-level locks were freed, so also release any potential low-level (storage engine)
     // locks that might be held.
@@ -1296,7 +1283,8 @@ void WiredTigerRecordStore::reclaimOplog(OperationContext* opCtx, Timestamp mayT
             invariantWTOK(cursor->reset(cursor), cursor->session);
             setKey(cursor, &truncateUpToKey);
             invariantWTOK(session->truncate(session, nullptr, nullptr, cursor, nullptr), session);
-            _changeNumRecordsAndDataSize(opCtx, -stone->records, -stone->bytes);
+            _changeNumRecords(opCtx, -stone->records);
+            _increaseDataSize(opCtx, -stone->bytes);
 
             wuow.commit();
 
@@ -1436,7 +1424,9 @@ Status WiredTigerRecordStore::_insertRecords(OperationContext* opCtx,
             metricsCollector.incrementOneDocWritten(value.size + keyLength);
         }
     }
-    _changeNumRecordsAndDataSize(opCtx, nRecords, totalLength);
+
+    _changeNumRecords(opCtx, nRecords);
+    _increaseDataSize(opCtx, totalLength);
 
     if (_oplogStones) {
         _oplogStones->updateCurrentStoneAfterInsertOnCommit(
@@ -1611,7 +1601,7 @@ Status WiredTigerRecordStore::doUpdateRecord(OperationContext* opCtx,
     }
     invariantWTOK(ret, c->session);
 
-    _changeNumRecordsAndDataSize(opCtx, 0, len - old_length);
+    _increaseDataSize(opCtx, len - old_length);
     return Status::OK();
 }
 
@@ -1668,8 +1658,9 @@ StatusWith<RecordData> WiredTigerRecordStore::doUpdateWithDamages(
 }
 
 void WiredTigerRecordStore::printRecordMetadata(OperationContext* opCtx,
-                                                const RecordId& recordId,
-                                                std::set<Timestamp>* recordTimestamps) const {
+                                                const RecordId& recordId) const {
+    LOGV2(6120300, "Printing record metadata", "recordId"_attr = recordId);
+
     // Printing the record metadata requires a new session. We cannot open other cursors when there
     // are open history store cursors in the session.
     WiredTigerSession session(_kvEngine->getConnection());
@@ -1714,7 +1705,7 @@ void WiredTigerRecordStore::printRecordMetadata(OperationContext* opCtx,
                       cursor->session);
 
         RecordData recordData(static_cast<const char*>(value.data), value.size);
-        LOGV2(6120300,
+        LOGV2(6120301,
               "WiredTiger record metadata",
               "recordId"_attr = recordId,
               "startTxnId"_attr = startTxnId,
@@ -1729,20 +1720,6 @@ void WiredTigerRecordStore::printRecordMetadata(OperationContext* opCtx,
               "location"_attr = location,
               "value"_attr = redact(recordData.toBson()));
 
-        // Save all relevant timestamps that we just printed.
-        if (recordTimestamps) {
-            auto saveRecordTimestampIfValid = [recordTimestamps](Timestamp ts) {
-                if (ts.isNull() || ts == Timestamp::max() || ts == Timestamp::min()) {
-                    return;
-                }
-                (void)recordTimestamps->emplace(ts);
-            };
-            saveRecordTimestampIfValid(Timestamp(startTs));
-            saveRecordTimestampIfValid(Timestamp(startDurableTs));
-            saveRecordTimestampIfValid(Timestamp(stopTs));
-            saveRecordTimestampIfValid(Timestamp(stopDurableTs));
-        }
-
         ret = cursor->next(cursor);
     }
 }
@@ -1765,7 +1742,8 @@ Status WiredTigerRecordStore::doTruncate(OperationContext* opCtx) {
     WT_SESSION* session = WiredTigerRecoveryUnit::get(opCtx)->getSession()->getSession();
     invariantWTOK(WT_OP_CHECK(session->truncate(session, nullptr, start, nullptr, nullptr)),
                   session);
-    _changeNumRecordsAndDataSize(opCtx, -numRecords(opCtx), -dataSize(opCtx));
+    _changeNumRecords(opCtx, -numRecords(opCtx));
+    _increaseDataSize(opCtx, -dataSize(opCtx));
 
     if (_oplogStones) {
         _oplogStones->clearStonesOnCommit(opCtx);
@@ -1996,9 +1974,7 @@ RecordId WiredTigerRecordStore::_nextId(OperationContext* opCtx) {
     return out;
 }
 
-void WiredTigerRecordStore::_changeNumRecordsAndDataSize(OperationContext* opCtx,
-                                                         int64_t numRecordDiff,
-                                                         int64_t dataSizeDiff) {
+void WiredTigerRecordStore::_changeNumRecords(OperationContext* opCtx, int64_t diff) {
     if (!_tracksSizeAdjustments) {
         return;
     }
@@ -2007,23 +1983,32 @@ void WiredTigerRecordStore::_changeNumRecordsAndDataSize(OperationContext* opCtx
         return;
     }
 
-    const auto updateAndStoreSizeInfo = [this](int64_t numRecordDiff, int64_t dataSizeDiff) {
-        _sizeInfo->numRecords.addAndFetch(numRecordDiff);
-        _sizeInfo->dataSize.addAndFetch(dataSizeDiff);
+    opCtx->recoveryUnit()->onRollback([this, diff]() {
+        LOGV2_DEBUG(
+            22404, 3, "WiredTigerRecordStore: rolling back NumRecordsChange", "diff"_attr = -diff);
+        _sizeInfo->numRecords.addAndFetch(-diff);
+    });
+    _sizeInfo->numRecords.addAndFetch(diff);
+}
 
-        if (_sizeStorer)
-            _sizeStorer->store(_uri, _sizeInfo);
-    };
+void WiredTigerRecordStore::_increaseDataSize(OperationContext* opCtx, int64_t amount) {
+    if (!_tracksSizeAdjustments) {
+        return;
+    }
 
-    opCtx->recoveryUnit()->onRollback([updateAndStoreSizeInfo, numRecordDiff, dataSizeDiff]() {
-        LOGV2_DEBUG(7105300,
-                    3,
-                    "WiredTigerRecordStore: rolling back change to numRecords and dataSize",
-                    "numRecordDiff"_attr = -numRecordDiff,
-                    "dataSizeDiff"_attr = -dataSizeDiff);
-        updateAndStoreSizeInfo(-numRecordDiff, -dataSizeDiff);
-    });
-    updateAndStoreSizeInfo(numRecordDiff, dataSizeDiff);
+    if (!sizeRecoveryState(getGlobalServiceContext()).collectionNeedsSizeAdjustment(getIdent())) {
+        return;
+    }
+
+    if (opCtx)
+        opCtx->recoveryUnit()->onRollback(
+            [this, amount]() { _increaseDataSize(nullptr, -amount); });
+
+    if (_sizeInfo->dataSize.fetchAndAdd(amount) < 0)
+        _sizeInfo->dataSize.store(std::max(amount, int64_t(0)));
+
+    if (_sizeStorer)
+        _sizeStorer->store(_uri, _sizeInfo);
 }
 
 void WiredTigerRecordStore::setNumRecords(long long numRecords) {
@@ -2107,7 +2092,8 @@ void WiredTigerRecordStore::doCappedTruncateAfter(OperationContext* opCtx,
     WT_SESSION* session = WiredTigerRecoveryUnit::get(opCtx)->getSession()->getSession();
     invariantWTOK(session->truncate(session, nullptr, start, nullptr, nullptr), session);
 
-    _changeNumRecordsAndDataSize(opCtx, -recordsRemoved, -bytesRemoved);
+    _changeNumRecords(opCtx, -recordsRemoved);
+    _increaseDataSize(opCtx, -bytesRemoved);
 
     wuow.commit();
 
@@ -2233,32 +2219,20 @@ boost::optional<Record> WiredTigerRecordStoreCursorBase::next() {
         return {};
     }
 
-    const bool failWithOutOfOrderForTest = WTRecordStoreUassertOutOfOrder.shouldFail();
-    if ((_forward && _lastReturnedId >= id) || MONGO_unlikely(failWithOutOfOrderForTest)) {
-        if (!failWithOutOfOrderForTest) {
-            // Crash when testing diagnostics are enabled and not explicitly uasserting on
-            // out-of-order keys.
-            invariant(!TestingProctor::instance().isEnabled(), "cursor returned out-of-order keys");
-        }
+    if (_forward && _lastReturnedId >= id) {
+        LOGV2_ERROR(22406,
+                    "WTCursor::next -- c->next_key ( {next}) was not greater than _lastReturnedId "
+                    "({last}) which is a bug.",
+                    "WTCursor::next -- next was not greater than last which is a bug",
+                    "next"_attr = id,
+                    "last"_attr = _lastReturnedId);
 
-        auto options = [&] {
-            if (_opCtx->recoveryUnit()->getDataCorruptionDetectionMode() ==
-                DataCorruptionDetectionMode::kThrow) {
-                // uassert with 'DataCorruptionDetected' after logging.
-                return logv2::LogOptions{
-                    logv2::UserAssertAfterLog(ErrorCodes::DataCorruptionDetected)};
-            } else {
-                return logv2::LogOptions(logv2::LogComponent::kAutomaticDetermination);
-            }
-        }();
-        LOGV2_ERROR_OPTIONS(22406,
-                            options,
-                            "WT_Cursor::next -- returned out-of-order keys",
-                            "forward"_attr = _forward,
-                            "next"_attr = id,
-                            "last"_attr = _lastReturnedId,
-                            "ident"_attr = _rs._ident,
-                            "ns"_attr = _rs.ns());
+        // Crash when testing diagnostics are enabled.
+        invariant(!TestingProctor::instance().isEnabled(), "next was not greater than last");
+
+        // Force a retry of the operation from our last known position by acting as-if
+        // we received a WT_ROLLBACK error.
+        throw WriteConflictException();
     }
 
     WT_ITEM value;
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.h b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.h
index 119b907f7a7..7630cc0900b 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.h
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.h
@@ -169,9 +169,7 @@ public:
                                                const char* damageSource,
                                                const mutablebson::DamageVector& damages) final;
 
-    virtual void printRecordMetadata(OperationContext* opCtx,
-                                     const RecordId& recordId,
-                                     std::set<Timestamp>* recordTimestamps) const;
+    virtual void printRecordMetadata(OperationContext* opCtx, const RecordId& recordId) const;
 
     virtual std::unique_ptr<SeekableRecordCursor> getCursor(OperationContext* opCtx,
                                                             bool forward) const = 0;
@@ -308,9 +306,9 @@ private:
     void _initNextIdIfNeeded(OperationContext* opCtx);
 
     /**
-     * Adjusts the record count and data size metadata for this record store. The function consults
-     * the SizeRecoveryState to determine whether or not to actually change the size metadata if the
-     * server is undergoing recovery.
+     * Adjusts the record count and data size metadata for this record store, respectively. These
+     * functions consult the SizeRecoveryState to determine whether or not to actually change the
+     * size metadata if the server is undergoing recovery.
      *
      * For most record stores, we will not update the size metadata during recovery, as we trust
      * that the values in the SizeStorer are accurate with respect to the end state of recovery.
@@ -324,9 +322,8 @@ private:
      *      are pending writes to this ident as part of the recovery process, and so we must
      *      always adjust size metadata for these idents.
      */
-    void _changeNumRecordsAndDataSize(OperationContext* opCtx,
-                                      int64_t numRecordDiff,
-                                      int64_t dataSizeDiff);
+    void _changeNumRecords(OperationContext* opCtx, int64_t diff);
+    void _increaseDataSize(OperationContext* opCtx, int64_t amount);
 
     const std::string _uri;
     const uint64_t _tableId;  // not persisted
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store_test.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store_test.cpp
index f4909583ce3..a2d9c5f7e99 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store_test.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store_test.cpp
@@ -1150,9 +1150,9 @@ TEST(WiredTigerRecordStoreTest, ClusteredRecordStore) {
     ASSERT_EQ(0, memcmp(dataUpdated, rd.data(), strlen(dataUpdated)));
 }
 
-// Make sure numRecords and dataSize are accurate after a delete rolls back and some other
-// transaction deletes the same rows before we have a chance of patching up the metadata.
-TEST(WiredTigerRecordStoreTest, SizeInfoAccurateAfterRollbackWithDelete) {
+// Make sure numRecords is accurate after a delete rolls back and some other transaction deletes the
+// same rows before we have a chance of patching up the metadata.
+TEST(WiredTigerRecordStoreTest, NumRecordsAccurateAfterRollbackWithDelete) {
     const auto harnessHelper(newRecordStoreHarnessHelper());
     unique_ptr<RecordStore> rs(harnessHelper->newRecordStore());
 
@@ -1166,7 +1166,6 @@ TEST(WiredTigerRecordStoreTest, SizeInfoAccurateAfterRollbackWithDelete) {
     }
 
     ASSERT_EQ(1, rs->numRecords(ctx.get()));
-    ASSERT_EQ(2, rs->dataSize(ctx.get()));
 
     WriteUnitOfWork uow(ctx.get());
 
@@ -1198,7 +1197,6 @@ TEST(WiredTigerRecordStoreTest, SizeInfoAccurateAfterRollbackWithDelete) {
 
     abortedThread.join();
     ASSERT_EQ(0, rs->numRecords(ctx.get()));
-    ASSERT_EQ(0, rs->dataSize(ctx.get()));
 }
 
 }  // namespace
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_recovery_unit.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_recovery_unit.cpp
index 3d0849c140c..ca2e5d8c1b4 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_recovery_unit.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_recovery_unit.cpp
@@ -39,7 +39,6 @@
 #include "mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h"
 #include "mongo/db/storage/wiredtiger/wiredtiger_prepare_conflict.h"
 #include "mongo/db/storage/wiredtiger/wiredtiger_session_cache.h"
-#include "mongo/db/storage/wiredtiger/wiredtiger_stats.h"
 #include "mongo/db/storage/wiredtiger/wiredtiger_util.h"
 #include "mongo/logv2/log.h"
 #include "mongo/util/hex.h"
@@ -48,7 +47,6 @@
 
 #include <fmt/compile.h>
 #include <fmt/format.h>
-#include <memory>
 
 namespace mongo {
 namespace {
@@ -81,6 +79,104 @@ void handleWriteContextForDebugging(WiredTigerRecoveryUnit& ru, Timestamp& ts) {
 
 AtomicWord<std::int64_t> snapshotTooOldErrorCount{0};
 
+using Section = WiredTigerOperationStats::Section;
+
+std::map<int, std::pair<StringData, Section>> WiredTigerOperationStats::_statNameMap = {
+    {WT_STAT_SESSION_BYTES_READ, std::make_pair("bytesRead"_sd, Section::DATA)},
+    {WT_STAT_SESSION_BYTES_WRITE, std::make_pair("bytesWritten"_sd, Section::DATA)},
+    {WT_STAT_SESSION_LOCK_DHANDLE_WAIT, std::make_pair("handleLock"_sd, Section::WAIT)},
+    {WT_STAT_SESSION_READ_TIME, std::make_pair("timeReadingMicros"_sd, Section::DATA)},
+    {WT_STAT_SESSION_WRITE_TIME, std::make_pair("timeWritingMicros"_sd, Section::DATA)},
+    {WT_STAT_SESSION_LOCK_SCHEMA_WAIT, std::make_pair("schemaLock"_sd, Section::WAIT)},
+    {WT_STAT_SESSION_CACHE_TIME, std::make_pair("cache"_sd, Section::WAIT)}};
+
+std::shared_ptr<StorageStats> WiredTigerOperationStats::getCopy() {
+    std::shared_ptr<WiredTigerOperationStats> copy = std::make_shared<WiredTigerOperationStats>();
+    *copy += *this;
+    return copy;
+}
+
+void WiredTigerOperationStats::fetchStats(WT_SESSION* session,
+                                          const std::string& uri,
+                                          const std::string& config) {
+    invariant(session);
+
+    WT_CURSOR* c = nullptr;
+    const char* cursorConfig = config.empty() ? nullptr : config.c_str();
+    int ret = session->open_cursor(session, uri.c_str(), nullptr, cursorConfig, &c);
+    uassert(ErrorCodes::CursorNotFound, "Unable to open statistics cursor", ret == 0);
+
+    invariant(c);
+    ON_BLOCK_EXIT([&] { c->close(c); });
+
+    const char* desc;
+    uint64_t value;
+    int32_t key;
+    while (c->next(c) == 0 && c->get_key(c, &key) == 0) {
+        fassert(51035, c->get_value(c, &desc, nullptr, &value) == 0);
+        _stats[key] = WiredTigerUtil::castStatisticsValue<long long>(value);
+    }
+
+    // Reset the statistics so that the next fetch gives the recent values.
+    invariantWTOK(c->reset(c), c->session);
+}
+
+BSONObj WiredTigerOperationStats::toBSON() {
+    BSONObjBuilder bob;
+    std::unique_ptr<BSONObjBuilder> dataSection;
+    std::unique_ptr<BSONObjBuilder> waitSection;
+
+    for (auto const& stat : _stats) {
+        // Find the user consumable name for this statistic.
+        auto statIt = _statNameMap.find(stat.first);
+        invariant(statIt != _statNameMap.end());
+
+        auto statName = statIt->second.first;
+        Section subs = statIt->second.second;
+        long long val = stat.second;
+        // Add this statistic only if higher than zero.
+        if (val > 0) {
+            // Gather the statistic into its own subsection in the BSONObj.
+            switch (subs) {
+                case Section::DATA:
+                    if (!dataSection)
+                        dataSection = std::make_unique<BSONObjBuilder>();
+
+                    dataSection->append(statName, val);
+                    break;
+                case Section::WAIT:
+                    if (!waitSection)
+                        waitSection = std::make_unique<BSONObjBuilder>();
+
+                    waitSection->append(statName, val);
+                    break;
+                default:
+                    MONGO_UNREACHABLE;
+            }
+        }
+    }
+
+    if (dataSection)
+        bob.append("data", dataSection->obj());
+    if (waitSection)
+        bob.append("timeWaitingMicros", waitSection->obj());
+
+    return bob.obj();
+}
+
+WiredTigerOperationStats& WiredTigerOperationStats::operator+=(
+    const WiredTigerOperationStats& other) {
+    for (auto const& otherStat : other._stats) {
+        _stats[otherStat.first] += otherStat.second;
+    }
+    return (*this);
+}
+
+StorageStats& WiredTigerOperationStats::operator+=(const StorageStats& other) {
+    *this += checked_cast<const WiredTigerOperationStats&>(other);
+    return (*this);
+}
+
 WiredTigerRecoveryUnit::WiredTigerRecoveryUnit(WiredTigerSessionCache* sc)
     : WiredTigerRecoveryUnit(sc, sc->getKVEngine()->getOplogManager()) {}
 
@@ -91,13 +187,6 @@ WiredTigerRecoveryUnit::WiredTigerRecoveryUnit(WiredTigerSessionCache* sc,
 WiredTigerRecoveryUnit::~WiredTigerRecoveryUnit() {
     invariant(!_inUnitOfWork(), toString(_getState()));
     _abort();
-
-    // If the session has non zero timeout then reset it back to 0 before returning the session back
-    // to the cache.
-    if (durationCount<Milliseconds>(_cacheMaxWaitTimeout)) {
-        auto wtSession = getSessionNoTxn()->getSession();
-        invariantWTOK(wtSession->reconfigure(wtSession, "cache_max_wait_ms=0"), wtSession);
-    }
 }
 
 void WiredTigerRecoveryUnit::_commit() {
@@ -361,22 +450,36 @@ void WiredTigerRecoveryUnit::_txnClose(bool commit) {
 
     int wtRet;
     if (commit) {
+        // Avoid heap allocation in favour of a stack allocation for the commit string.
+        static constexpr auto commitTimestampFmtString = "commit_timestamp={:X},";
+        static constexpr auto durableTimestampFmtString = "durable_timestamp={:X}";
+        static constexpr auto bytesRequired =
+            std::char_traits<char>::length(commitTimestampFmtString) +
+            (sizeof(decltype(_commitTimestamp.asULL())) * 2) +
+            std::char_traits<char>::length(durableTimestampFmtString) +
+            (sizeof(decltype(_durableTimestamp.asULL())) * 2) + 1;
+        std::array<char, bytesRequired> conf;
+        auto end = conf.begin();
         if (!_commitTimestamp.isNull()) {
             // There is currently no scenario where it is intentional to commit before the current
             // read timestamp.
             invariant(_readAtTimestamp.isNull() || _commitTimestamp >= _readAtTimestamp);
 
             if (MONGO_likely(!doUntimestampedWritesForIdempotencyTests.shouldFail())) {
-                s->timestamp_transaction_uint(s, WT_TS_TXN_TYPE_COMMIT, _commitTimestamp.asULL());
+                end = fmt::format_to(
+                    end, FMT_STRING(commitTimestampFmtString), _commitTimestamp.asULL());
             }
             _isTimestamped = true;
         }
 
         if (!_durableTimestamp.isNull()) {
-            s->timestamp_transaction_uint(s, WT_TS_TXN_TYPE_DURABLE, _durableTimestamp.asULL());
+            end = fmt::format_to(
+                end, FMT_STRING(durableTimestampFmtString), _durableTimestamp.asULL());
         }
 
-        wtRet = s->commit_transaction(s, nullptr);
+        *end = '\0';
+
+        wtRet = s->commit_transaction(s, conf.data());
 
         LOGV2_DEBUG(
             22412, 3, "WT commit_transaction", "snapshotId"_attr = getSnapshotId().toNumber());
@@ -428,12 +531,6 @@ void WiredTigerRecoveryUnit::_txnClose(bool commit) {
     _isOplogReader = false;
     _oplogVisibleTs = boost::none;
     _orderedCommit = true;  // Default value is true; we assume all writes are ordered.
-    // Reset the kLastApplied read source back to the default of kNoTimestamp. Any reader requiring
-    // kLastApplied will set the read source again before reading. Resetting this read source
-    // simplifies the handling when stepup happens concurrently with read operations.
-    if (_timestampReadSource == ReadSource::kLastApplied) {
-        _timestampReadSource = ReadSource::kNoTimestamp;
-    }
 }
 
 Status WiredTigerRecoveryUnit::majorityCommittedSnapshotAvailable() const {
@@ -460,16 +557,11 @@ boost::optional<Timestamp> WiredTigerRecoveryUnit::getPointInTimeReadTimestamp(
             // The read timestamp is set by the user and does not require a transaction to be open.
             invariant(!_readAtTimestamp.isNull());
             return _readAtTimestamp;
-        case ReadSource::kLastApplied:
-            // The lastApplied timestamp is not always available if the system has not accepted
-            // writes, so it is not possible to invariant that it exists.
-            if (_readAtTimestamp.isNull()) {
-                return boost::none;
-            }
-            return _readAtTimestamp;
+
         // The following ReadSources can only establish a read timestamp when a transaction is
         // opened.
         case ReadSource::kNoOverlap:
+        case ReadSource::kLastApplied:
         case ReadSource::kAllDurableSnapshot:
         case ReadSource::kMajorityCommitted:
             break;
@@ -530,7 +622,7 @@ void WiredTigerRecoveryUnit::_txnOpen() {
             break;
         }
         case ReadSource::kLastApplied: {
-            _beginTransactionAtLastAppliedTimestamp(session);
+            _readAtTimestamp = _beginTransactionAtLastAppliedTimestamp(session);
             break;
         }
         case ReadSource::kNoOverlap: {
@@ -587,8 +679,9 @@ Timestamp WiredTigerRecoveryUnit::_beginTransactionAtAllDurableTimestamp(WT_SESS
     return readTimestamp;
 }
 
-void WiredTigerRecoveryUnit::_beginTransactionAtLastAppliedTimestamp(WT_SESSION* session) {
-    if (_readAtTimestamp.isNull()) {
+Timestamp WiredTigerRecoveryUnit::_beginTransactionAtLastAppliedTimestamp(WT_SESSION* session) {
+    auto lastApplied = _sessionCache->snapshotManager().getLastApplied();
+    if (!lastApplied) {
         // When there is not a lastApplied timestamp available, read without a timestamp. Do not
         // round up the read timestamp to the oldest timestamp.
 
@@ -602,20 +695,20 @@ void WiredTigerRecoveryUnit::_beginTransactionAtLastAppliedTimestamp(WT_SESSION*
             session, _prepareConflictBehavior, _roundUpPreparedTimestamps);
         LOGV2_DEBUG(4847500, 2, "no read timestamp available for kLastApplied");
         txnOpen.done();
-        return;
+        return Timestamp();
     }
 
     WiredTigerBeginTxnBlock txnOpen(session,
                                     _prepareConflictBehavior,
                                     _roundUpPreparedTimestamps,
                                     RoundUpReadTimestamp::kRound);
-    auto status = txnOpen.setReadSnapshot(_readAtTimestamp);
+    auto status = txnOpen.setReadSnapshot(*lastApplied);
     fassert(4847501, status);
     txnOpen.done();
 
     // We might have rounded to oldest between calling getLastApplied and setReadSnapshot. We
     // need to get the actual read timestamp we used.
-    _readAtTimestamp = _getTransactionReadTimestamp(session);
+    return _getTransactionReadTimestamp(session);
 }
 
 Timestamp WiredTigerRecoveryUnit::_beginTransactionAtNoOverlapTimestamp(WT_SESSION* session) {
@@ -857,16 +950,7 @@ void WiredTigerRecoveryUnit::setTimestampReadSource(ReadSource readSource,
     invariant(!(provided && provided->isNull()));
 
     _timestampReadSource = readSource;
-    if (readSource == kLastApplied) {
-        // The lastApplied timestamp is not always available if the system has not accepted writes.
-        if (auto lastApplied = _sessionCache->snapshotManager().getLastApplied()) {
-            _readAtTimestamp = *lastApplied;
-        } else {
-            _readAtTimestamp = Timestamp();
-        }
-    } else {
-        _readAtTimestamp = (provided) ? *provided : Timestamp();
-    }
+    _readAtTimestamp = (provided) ? *provided : Timestamp();
 }
 
 RecoveryUnit::ReadSource WiredTigerRecoveryUnit::getTimestampReadSource() const {
@@ -894,21 +978,19 @@ void WiredTigerRecoveryUnit::beginIdle() {
     }
 }
 
-std::unique_ptr<StorageStats> WiredTigerRecoveryUnit::computeOperationStatisticsSinceLastCall() {
-    if (!_session)
-        return nullptr;
+std::shared_ptr<StorageStats> WiredTigerRecoveryUnit::getOperationStatistics() const {
+    std::shared_ptr<WiredTigerOperationStats> statsPtr(nullptr);
 
-    // We compute operation statistics as the difference between the current session statistics and
-    // the session statistics of the last time the method was called, which should correspond to the
-    // end of one operation.
-    WiredTigerStats currentSessionStats{_session->getSession()};
+    if (!_session)
+        return statsPtr;
 
-    auto operationStats =
-        std::make_unique<WiredTigerStats>(currentSessionStats - _sessionStatsAfterLastOperation);
+    WT_SESSION* s = _session->getSession();
+    invariant(s);
 
-    _sessionStatsAfterLastOperation = std::move(currentSessionStats);
+    statsPtr = std::make_shared<WiredTigerOperationStats>();
+    statsPtr->fetchStats(s, "statistics:session", "statistics=(fast)");
 
-    return operationStats;
+    return statsPtr;
 }
 
 void WiredTigerRecoveryUnit::setCatalogConflictingTimestamp(Timestamp timestamp) {
@@ -938,15 +1020,4 @@ void WiredTigerRecoveryUnit::storeWriteContextForDebugging(const BSONObj& info)
     _writeContextForDebugging.push_back(info);
 }
 
-void WiredTigerRecoveryUnit::setCacheMaxWaitTimeout(Milliseconds timeout) {
-    _cacheMaxWaitTimeout = timeout;
-
-    auto wtSession = getSessionNoTxn()->getSession();
-    invariantWTOK(
-        wtSession->reconfigure(
-            wtSession,
-            fmt::format("cache_max_wait_ms={}", durationCount<Milliseconds>(_cacheMaxWaitTimeout))
-                .c_str()),
-        wtSession);
-}
 }  // namespace mongo
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_recovery_unit.h b/src/mongo/db/storage/wiredtiger/wiredtiger_recovery_unit.h
index f1f692043b4..2d75b4e54f1 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_recovery_unit.h
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_recovery_unit.h
@@ -45,8 +45,8 @@
 #include "mongo/db/storage/recovery_unit.h"
 #include "mongo/db/storage/wiredtiger/wiredtiger_begin_transaction_block.h"
 #include "mongo/db/storage/wiredtiger/wiredtiger_session_cache.h"
-#include "mongo/db/storage/wiredtiger/wiredtiger_stats.h"
 #include "mongo/util/timer.h"
+
 namespace mongo {
 
 using RoundUpPreparedTimestamps = WiredTigerBeginTxnBlock::RoundUpPreparedTimestamps;
@@ -56,6 +56,41 @@ extern AtomicWord<std::int64_t> snapshotTooOldErrorCount;
 
 class BSONObjBuilder;
 
+class WiredTigerOperationStats final : public StorageStats {
+public:
+    /**
+     *  There are two types of statistics provided by WiredTiger engine - data and wait.
+     */
+    enum class Section { DATA, WAIT };
+
+    BSONObj toBSON() final;
+
+    StorageStats& operator+=(const StorageStats&) final;
+
+    WiredTigerOperationStats& operator+=(const WiredTigerOperationStats&);
+
+    /**
+     * Fetches an operation's storage statistics from WiredTiger engine.
+     */
+    void fetchStats(WT_SESSION*, const std::string&, const std::string&);
+
+    std::shared_ptr<StorageStats> getCopy() final;
+
+private:
+    /**
+     * Each statistic in WiredTiger has an integer key, which this map associates with a section
+     * (either DATA or WAIT) and user-readable name.
+     */
+    static std::map<int, std::pair<StringData, Section>> _statNameMap;
+
+    /**
+     * Stores the value for each statistic returned by a WiredTiger cursor. Each statistic is
+     * associated with an integer key, which can be mapped to a name and section using the
+     * '_statNameMap'.
+     */
+    std::map<int, long long> _stats;
+};
+
 class WiredTigerRecoveryUnit final : public RecoveryUnit {
 public:
     WiredTigerRecoveryUnit(WiredTigerSessionCache* sc);
@@ -139,7 +174,7 @@ public:
         return _readOnce;
     };
 
-    std::unique_ptr<StorageStats> computeOperationStatisticsSinceLastCall() override;
+    std::shared_ptr<StorageStats> getOperationStatistics() const override;
 
     void refreshSnapshot() override;
 
@@ -147,8 +182,6 @@ public:
         _multiTimestampConstraintTracker.ignoreAllMultiTimestampConstraints = true;
     }
 
-    void setCacheMaxWaitTimeout(Milliseconds) override;
-
     // ---- WT STUFF
 
     WiredTigerSession* getSession();
@@ -224,11 +257,10 @@ private:
     Timestamp _beginTransactionAtNoOverlapTimestamp(WT_SESSION* session);
 
     /**
-     * Starts a transaction at the lastApplied timestamp stored in '_readAtTimestamp'. Sets
-     * '_readAtTimestamp' to the actual timestamp used by the storage engine in case rounding
-     * occured.
+     * Starts a transaction at the lastApplied timestamp. Returns the timestamp at which the
+     * transaction was started.
      */
-    void _beginTransactionAtLastAppliedTimestamp(WT_SESSION* session);
+    Timestamp _beginTransactionAtLastAppliedTimestamp(WT_SESSION* session);
 
     /**
      * Returns the timestamp at which the current transaction is reading.
@@ -283,10 +315,6 @@ private:
     boost::optional<int64_t> _oplogVisibleTs = boost::none;
     bool _gatherWriteContextForDebugging = false;
     std::vector<BSONObj> _writeContextForDebugging;
-
-    WiredTigerStats _sessionStatsAfterLastOperation;
-
-    Milliseconds _cacheMaxWaitTimeout{0};
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_session_cache.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_session_cache.cpp
index 26e9824cc57..3ed1d4e985b 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_session_cache.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_session_cache.cpp
@@ -36,6 +36,7 @@
 #include <memory>
 
 #include "mongo/base/error_codes.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/global_settings.h"
 #include "mongo/db/repl/repl_settings.h"
 #include "mongo/db/storage/journal_listener.h"
@@ -241,10 +242,6 @@ void WiredTigerSessionCache::shuttingDown() {
     closeAll();
 }
 
-void WiredTigerSessionCache::restart() {
-    _shuttingDown.fetchAndBitAnd(~kShuttingDownMask);
-}
-
 bool WiredTigerSessionCache::isShuttingDown() {
     return _shuttingDown.load() & kShuttingDownMask;
 }
@@ -291,26 +288,32 @@ void WiredTigerSessionCache::waitUntilDurable(OperationContext* opCtx,
     // waiters, as a log flush is much cheaper than a full checkpoint.
     if ((syncType == Fsync::kCheckpointStableTimestamp || syncType == Fsync::kCheckpointAll) &&
         _engine->isDurable()) {
-        auto journalListener = [&]() -> JournalListener* {
-            // The JournalListener may not be set immediately, so we must check under a mutex so
-            // as not to access the variable while setting a JournalListener. A JournalListener
-            // is only allowed to be set once, so using the pointer outside of a mutex is safe.
-            stdx::unique_lock<Latch> lk(_journalListenerMutex);
-            return _journalListener;
-        }();
-        boost::optional<JournalListener::Token> token;
-        if (journalListener && useListener == UseJournalListener::kUpdate) {
-            // Update a persisted value with the latest write timestamp that is safe across
-            // startup recovery in the repl layer. Then report that timestamp as durable to the
-            // repl layer below after we have flushed in-memory data to disk.
-            // Note: only does a write if primary, otherwise just fetches the timestamp.
-            token = journalListener->getToken(opCtx);
-        }
+        UniqueWiredTigerSession session = getSession();
+        WT_SESSION* s = session->getSession();
+        {
+            auto journalListener = [&]() -> JournalListener* {
+                // The JournalListener may not be set immediately, so we must check under a mutex so
+                // as not to access the variable while setting a JournalListener. A JournalListener
+                // is only allowed to be set once, so using the pointer outside of a mutex is safe.
+                stdx::unique_lock<Latch> lk(_journalListenerMutex);
+                return _journalListener;
+            }();
+            boost::optional<JournalListener::Token> token;
+            if (journalListener && useListener == UseJournalListener::kUpdate) {
+                // Update a persisted value with the latest write timestamp that is safe across
+                // startup recovery in the repl layer. Then report that timestamp as durable to the
+                // repl layer below after we have flushed in-memory data to disk.
+                // Note: only does a write if primary, otherwise just fetches the timestamp.
+                token = journalListener->getToken(opCtx);
+            }
 
-        getKVEngine()->forceCheckpoint(syncType == Fsync::kCheckpointStableTimestamp);
+            auto config = syncType == Fsync::kCheckpointStableTimestamp ? "use_timestamp=true"
+                                                                        : "use_timestamp=false";
+            invariantWTOK(s->checkpoint(s, config), s);
 
-        if (token) {
-            journalListener->onDurable(token.get());
+            if (token) {
+                journalListener->onDurable(token.get());
+            }
         }
         LOGV2_DEBUG(22418, 4, "created checkpoint (forced)");
         return;
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_session_cache.h b/src/mongo/db/storage/wiredtiger/wiredtiger_session_cache.h
index 99550d2e749..18da44e3246 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_session_cache.h
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_session_cache.h
@@ -311,11 +311,6 @@ public:
      */
     bool isShuttingDown();
 
-    /**
-     * Restart a previously shut down cache.
-     */
-    void restart();
-
     bool isEphemeral();
 
     /**
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_size_storer.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_size_storer.cpp
index 18695aa6561..59a4e92e6a0 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_size_storer.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_size_storer.cpp
@@ -35,7 +35,6 @@
 
 #include "mongo/bson/bsonobj.h"
 #include "mongo/bson/bsonobjbuilder.h"
-#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/storage/wiredtiger/wiredtiger_begin_transaction_block.h"
 #include "mongo/db/storage/wiredtiger/wiredtiger_customization_hooks.h"
@@ -91,29 +90,29 @@ void WiredTigerSizeStorer::store(StringData uri, std::shared_ptr<SizeInfo> sizeI
                 "entryUseCount"_attr = entry.use_count());
 }
 
-std::shared_ptr<WiredTigerSizeStorer::SizeInfo> WiredTigerSizeStorer::load(StringData uri) const {
+std::shared_ptr<WiredTigerSizeStorer::SizeInfo> WiredTigerSizeStorer::load(OperationContext* opCtx,
+                                                                           StringData uri) const {
     {
         // Check if we can satisfy the read from the buffer.
         stdx::lock_guard<Latch> bufferLock(_bufferMutex);
         Buffer::const_iterator it = _buffer.find(uri);
         if (it != _buffer.end())
-            return it->second ? it->second : std::make_shared<SizeInfo>();
+            return it->second;
     }
 
-    WiredTigerSession session{_conn};
-    auto cursor = session.getNewCursor(_storageUri);
+    WiredTigerCursor cursor(_storageUri, _tableId, /*allowOverwrite=*/false, opCtx);
 
     {
         WT_ITEM key = {uri.rawData(), uri.size()};
-        cursor->set_key(cursor, &key);
-        int ret = cursor->search(cursor);
+        cursor->set_key(cursor.get(), &key);
+        int ret = cursor->search(cursor.get());
         if (ret == WT_NOTFOUND)
             return std::make_shared<SizeInfo>();
         invariantWTOK(ret, cursor->session);
     }
 
     WT_ITEM value;
-    invariantWTOK(cursor->get_value(cursor, &value), cursor->session);
+    invariantWTOK(cursor->get_value(cursor.get(), &value), cursor->session);
     BSONObj data(reinterpret_cast<const char*>(value.data));
 
     LOGV2_DEBUG(
@@ -122,17 +121,6 @@ std::shared_ptr<WiredTigerSizeStorer::SizeInfo> WiredTigerSizeStorer::load(Strin
                                       data["dataSize"].safeNumberLong());
 }
 
-void WiredTigerSizeStorer::remove(StringData uri) {
-    stdx::lock_guard<Latch> bufferLock{_bufferMutex};
-
-    // Insert a new nullptr entry into the buffer, or set the existing one to nullptr if there
-    // already is one.
-    if (auto& sizeInfo = _buffer[uri]) {
-        sizeInfo->_dirty.store(false);
-        sizeInfo.reset();
-    }
-}
-
 void WiredTigerSizeStorer::flush(bool syncToDisk) {
     Buffer buffer;
     {
@@ -172,45 +160,33 @@ void WiredTigerSizeStorer::flush(bool syncToDisk) {
         }
         WiredTigerBeginTxnBlock txnOpen(session.getSession(), txnConfig.c_str());
 
-        for (auto&& [uri, sizeInfo] : buffer) {
+        for (auto it = buffer.begin(); it != buffer.end(); ++it) {
+
+            // Ordering is important here: when the store method checks if the SizeInfo
+            // is dirty and it returns true, the current values of numRecords and dataSize must
+            // still be written back. So, the required order is to clear the dirty flag first.
+            SizeInfo& sizeInfo = *it->second;
+            sizeInfo._dirty.store(false);
+            BSONObj data = BSON("numRecords" << sizeInfo.numRecords.load() << "dataSize"
+                                             << sizeInfo.dataSize.load());
+
+            auto& uri = it->first;
+            LOGV2_DEBUG(22425,
+                        2,
+                        "WiredTigerSizeStorer::flush",
+                        "uri"_attr = uri,
+                        "data"_attr = redact(data));
             WiredTigerItem key(uri.c_str(), uri.size());
+            WiredTigerItem value(data.objdata(), data.objsize());
             cursor->set_key(cursor, key.Get());
-
-            int ret = 0;
-            if (!sizeInfo) {
-                LOGV2_DEBUG(
-                    3349400, 2, "WiredTigerSizeStorer::flush removing entry", "uri"_attr = uri);
-
-                ret = cursor->remove(cursor);
-                if (ret == WT_NOTFOUND) {
-                    ret = 0;
-                }
-            } else {
-                // Ordering is important here: when the store method checks if the SizeInfo
-                // is dirty and it returns true, the current values of numRecords and dataSize must
-                // still be written back. So, the required order is to clear the dirty flag first.
-                sizeInfo->_dirty.store(false);
-                auto data = BSON("numRecords" << sizeInfo->numRecords.load() << "dataSize"
-                                              << sizeInfo->dataSize.load());
-
-                LOGV2_DEBUG(22425,
-                            2,
-                            "WiredTigerSizeStorer::flush inserting/updating entry",
-                            "uri"_attr = uri,
-                            "data"_attr = redact(data));
-
-                WiredTigerItem value(data.objdata(), data.objsize());
-                cursor->set_value(cursor, value.Get());
-                ret = cursor->insert(cursor);
-            }
-
+            cursor->set_value(cursor, value.Get());
+            int ret = cursor->insert(cursor);
             if (ret == WT_ROLLBACK) {
                 // One of the code paths calling this function is when a session is checked back
                 // into the session cache. This could involve read-only operations which don't
                 // except write conflicts. If WiredTiger returns WT_ROLLBACK during the flush, we
-                // return an exception here and let the caller decide whether to ignore it or retry
-                // flushing.
-                throw WriteConflictException("Size storer flush received a rollback.");
+                // skip flushing.
+                return;
             }
             invariantWTOK(ret, cursor->session);
         }
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_size_storer.h b/src/mongo/db/storage/wiredtiger/wiredtiger_size_storer.h
index 349a800b5d6..50eb23324c0 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_size_storer.h
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_size_storer.h
@@ -85,17 +85,7 @@ public:
      */
     void store(StringData uri, std::shared_ptr<SizeInfo> sizeInfo);
 
-    /**
-     * Returns the size info for the given URI. Creates a default-initialized SizeInfo if there is
-     * no existing size info for the given URI. Never returns nullptr.
-     */
-    std::shared_ptr<SizeInfo> load(StringData uri) const;
-
-    /**
-     * Informs the size storer that the size information about the given ident should be removed
-     * upon the next flush.
-     */
-    void remove(StringData uri);
+    std::shared_ptr<SizeInfo> load(OperationContext* opCtx, StringData uri) const;
 
     /**
      * Writes all changes to the underlying table.
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_size_storer_test.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_size_storer_test.cpp
deleted file mode 100644
index c7207bd2686..00000000000
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_size_storer_test.cpp
+++ /dev/null
@@ -1,184 +0,0 @@
-/**
- *    Copyright (C) 2023-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 <wiredtiger.h>
-
-#include "mongo/db/service_context_test_fixture.h"
-#include "mongo/db/storage/wiredtiger/wiredtiger_size_storer.h"
-#include "mongo/db/storage/wiredtiger/wiredtiger_util.h"
-#include "mongo/unittest/temp_dir.h"
-#include "mongo/unittest/unittest.h"
-
-namespace mongo {
-namespace {
-
-class WiredTigerSizeStorerTest : public ServiceContextTest {
-protected:
-    WiredTigerSizeStorerTest() {
-        ASSERT_OK(wtRCToStatus(wiredtiger_open(_tempDir.path().c_str(), nullptr, "create", &_conn),
-                               nullptr));
-    }
-
-    WiredTigerSizeStorer makeSizeStorer() const {
-        return {_conn, "table:sizeStorer"};
-    }
-
-private:
-    unittest::TempDir _tempDir{"WiredTigerSizeStorerTest"};
-    WT_CONNECTION* _conn;
-};
-
-TEST_F(WiredTigerSizeStorerTest, Store) {
-    auto sizeStorer1 = makeSizeStorer();
-    auto sizeStorer2 = makeSizeStorer();
-    auto sizeInfo = std::make_shared<WiredTigerSizeStorer::SizeInfo>(1, 10);
-    StringData uri{"uri1"};
-
-    sizeStorer1.store(uri, sizeInfo);
-
-    auto loaded = sizeStorer1.load(uri);
-    ASSERT(loaded);
-    ASSERT_EQ(loaded, sizeInfo);
-    ASSERT_EQ(loaded->numRecords.load(), sizeInfo->numRecords.load());
-    ASSERT_EQ(loaded->dataSize.load(), sizeInfo->dataSize.load());
-
-    loaded = sizeStorer2.load(uri);
-    ASSERT(loaded);
-    ASSERT_NE(loaded, sizeInfo);
-    ASSERT_EQ(loaded->numRecords.load(), 0);
-    ASSERT_EQ(loaded->dataSize.load(), 0);
-
-    sizeStorer1.flush(false);
-
-    loaded = sizeStorer1.load(uri);
-    ASSERT(loaded);
-    ASSERT_NE(loaded, sizeInfo);
-    ASSERT_EQ(loaded->numRecords.load(), sizeInfo->numRecords.load());
-    ASSERT_EQ(loaded->dataSize.load(), sizeInfo->dataSize.load());
-
-    loaded = sizeStorer2.load(uri);
-    ASSERT(loaded);
-    ASSERT_NE(loaded, sizeInfo);
-    ASSERT_EQ(loaded->numRecords.load(), sizeInfo->numRecords.load());
-    ASSERT_EQ(loaded->dataSize.load(), sizeInfo->dataSize.load());
-}
-
-TEST_F(WiredTigerSizeStorerTest, RemoveBeforeFlush) {
-    auto sizeStorer = makeSizeStorer();
-    auto sizeInfo = std::make_shared<WiredTigerSizeStorer::SizeInfo>(1, 10);
-    StringData uri{"uri1"};
-
-    sizeStorer.store(uri, sizeInfo);
-
-    auto loaded = sizeStorer.load(uri);
-    ASSERT(loaded);
-    ASSERT_EQ(loaded, sizeInfo);
-    ASSERT_EQ(loaded->numRecords.load(), sizeInfo->numRecords.load());
-    ASSERT_EQ(loaded->dataSize.load(), sizeInfo->dataSize.load());
-
-    sizeStorer.remove(uri);
-
-    loaded = sizeStorer.load(uri);
-    ASSERT(loaded);
-    ASSERT_NE(loaded, sizeInfo);
-    ASSERT_EQ(loaded->numRecords.load(), 0);
-    ASSERT_EQ(loaded->dataSize.load(), 0);
-
-    sizeStorer.flush(false);
-
-    loaded = sizeStorer.load(uri);
-    ASSERT(loaded);
-    ASSERT_NE(loaded, sizeInfo);
-    ASSERT_EQ(loaded->numRecords.load(), 0);
-    ASSERT_EQ(loaded->dataSize.load(), 0);
-}
-
-TEST_F(WiredTigerSizeStorerTest, RemoveAfterFlush) {
-    auto sizeStorer = makeSizeStorer();
-    auto sizeInfo = std::make_shared<WiredTigerSizeStorer::SizeInfo>(1, 10);
-    StringData uri{"uri1"};
-
-    sizeStorer.store(uri, sizeInfo);
-    sizeStorer.flush(false);
-
-    auto loaded = sizeStorer.load(uri);
-    ASSERT(loaded);
-    ASSERT_NE(loaded, sizeInfo);
-    ASSERT_EQ(loaded->numRecords.load(), sizeInfo->numRecords.load());
-    ASSERT_EQ(loaded->dataSize.load(), sizeInfo->dataSize.load());
-
-    sizeStorer.remove(uri);
-
-    loaded = sizeStorer.load(uri);
-    ASSERT(loaded);
-    ASSERT_NE(loaded, sizeInfo);
-    ASSERT_EQ(loaded->numRecords.load(), 0);
-    ASSERT_EQ(loaded->dataSize.load(), 0);
-
-    sizeStorer.flush(false);
-
-    loaded = sizeStorer.load(uri);
-    ASSERT(loaded);
-    ASSERT_NE(loaded, sizeInfo);
-    ASSERT_EQ(loaded->numRecords.load(), 0);
-    ASSERT_EQ(loaded->dataSize.load(), 0);
-}
-
-TEST_F(WiredTigerSizeStorerTest, RemoveNonexistent) {
-    auto sizeStorer = makeSizeStorer();
-    auto sizeInfo = std::make_shared<WiredTigerSizeStorer::SizeInfo>(1, 10);
-    StringData uri{"uri1"};
-
-    sizeStorer.remove(uri);
-
-    auto loaded = sizeStorer.load(uri);
-    ASSERT(loaded);
-    ASSERT_NE(loaded, sizeInfo);
-    ASSERT_EQ(loaded->numRecords.load(), 0);
-    ASSERT_EQ(loaded->dataSize.load(), 0);
-
-    sizeStorer.store(uri, sizeInfo);
-
-    loaded = sizeStorer.load(uri);
-    ASSERT(loaded);
-    ASSERT_EQ(loaded, sizeInfo);
-    ASSERT_EQ(loaded->numRecords.load(), sizeInfo->numRecords.load());
-    ASSERT_EQ(loaded->dataSize.load(), sizeInfo->dataSize.load());
-
-    sizeStorer.flush(false);
-
-    loaded = sizeStorer.load(uri);
-    ASSERT(loaded);
-    ASSERT_NE(loaded, sizeInfo);
-    ASSERT_EQ(loaded->numRecords.load(), sizeInfo->numRecords.load());
-    ASSERT_EQ(loaded->dataSize.load(), sizeInfo->dataSize.load());
-}
-
-}  // namespace
-}  // namespace mongo
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_standard_index_test.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_standard_index_test.cpp
index 0101172b81b..f4cdc403149 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_standard_index_test.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_standard_index_test.cpp
@@ -66,11 +66,15 @@ public:
 
         _fastClockSource = std::make_unique<SystemClockSource>();
         _sessionCache = new WiredTigerSessionCache(_conn, _fastClockSource.get());
+
+        WiredTigerUtil::notifyStartupComplete();
     }
 
     ~WiredTigerIndexHarnessHelper() final {
         delete _sessionCache;
         _conn->close(_conn, nullptr);
+
+        WiredTigerUtil::resetTableLoggingInfo();
     }
 
     std::unique_ptr<SortedDataInterface> newIdIndexSortedDataInterface() final {
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_standard_record_store_test.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_standard_record_store_test.cpp
index 8255a880d00..1b74a9ead4e 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_standard_record_store_test.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_standard_record_store_test.cpp
@@ -38,6 +38,7 @@
 #include "mongo/base/init.h"
 #include "mongo/base/string_data.h"
 #include "mongo/bson/bsonobjbuilder.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/json.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h"
@@ -97,7 +98,8 @@ TEST(WiredTigerRecordStoreTest, SizeStorer1) {
     rs.reset(nullptr);
 
     {
-        auto& info = *ss.load(uri);
+        ServiceContext::UniqueOperationContext opCtx(harnessHelper->newOperationContext());
+        auto& info = *ss.load(opCtx.get(), uri);
         ASSERT_EQUALS(N, info.numRecords.load());
     }
 
@@ -143,9 +145,10 @@ TEST(WiredTigerRecordStoreTest, SizeStorer1) {
     }
 
     {
+        ServiceContext::UniqueOperationContext opCtx(harnessHelper->newOperationContext());
         const bool enableWtLogging = false;
         WiredTigerSizeStorer ss2(harnessHelper->conn(), indexUri, enableWtLogging);
-        auto info = ss2.load(uri);
+        auto info = ss2.load(opCtx.get(), uri);
         ASSERT_EQUALS(N, info->numRecords.load());
     }
 
@@ -175,12 +178,12 @@ private:
     }
 
 protected:
-    long long getNumRecords() const {
-        return sizeStorer->load(uri)->numRecords.load();
+    long long getNumRecords(OperationContext* opCtx) const {
+        return sizeStorer->load(opCtx, uri)->numRecords.load();
     }
 
-    long long getDataSize() const {
-        return sizeStorer->load(uri)->dataSize.load();
+    long long getDataSize(OperationContext* opCtx) const {
+        return sizeStorer->load(opCtx, uri)->dataSize.load();
     }
 
     std::unique_ptr<WiredTigerHarnessHelper> harnessHelper;
@@ -195,49 +198,9 @@ TEST_F(SizeStorerUpdateTest, Basic) {
     ServiceContext::UniqueOperationContext opCtx(harnessHelper->newOperationContext());
     long long val = 5;
     rs->updateStatsAfterRepair(opCtx.get(), val, val);
-    ASSERT_EQUALS(getNumRecords(), val);
-    ASSERT_EQUALS(getDataSize(), val);
+    ASSERT_EQUALS(getNumRecords(opCtx.get()), val);
+    ASSERT_EQUALS(getDataSize(opCtx.get()), val);
 }
 
-// Verify that the size storer contains accurate data after a transaction rollback just before a
-// flush (simulating a shutdown). That is, that the rollback marks the size info as dirty, and is
-// properly flushed to disk.
-TEST_F(SizeStorerUpdateTest, ReloadAfterRollbackAndFlush) {
-    ServiceContext::UniqueOperationContext opCtx(harnessHelper->newOperationContext());
-    // Do an op for which the sizeInfo is persisted, for safety so we don't check against 0.
-    {
-        WriteUnitOfWork uow(opCtx.get());
-        auto rId = rs->insertRecord(opCtx.get(), "12345", 5, Timestamp{1});
-        ASSERT_TRUE(rId.isOK());
-
-        uow.commit();
-    }
-
-    // An operation to rollback, with a flush between the original modification and the rollback.
-    {
-        WriteUnitOfWork uow(opCtx.get());
-        auto rId = rs->insertRecord(opCtx.get(), "12345", 5, Timestamp{2});
-        ASSERT_TRUE(rId.isOK());
-
-        ASSERT_EQ(getNumRecords(), 2);
-        ASSERT_EQ(getDataSize(), 10);
-        // Mark size info as clean, before rollback is done.
-        sizeStorer->flush(false);
-    }
-
-    // Simulate a shutdown and restart, which loads the size storer from disk.
-    sizeStorer->flush(true);
-    sizeStorer.reset(new WiredTigerSizeStorer(harnessHelper->conn(),
-                                              WiredTigerKVEngine::kTableUriPrefix + "sizeStorer"));
-    WiredTigerRecordStore* wtrs = checked_cast<WiredTigerRecordStore*>(rs.get());
-    wtrs->setSizeStorer(sizeStorer.get());
-
-    // As the operation was rolled back, numRecords and dataSize should be for the first op only. If
-    // rollback does not properly mark the sizeInfo as dirty, on load sizeInfo will account for the
-    // two operations, as the rollback sizeInfo update has not been flushed.
-    ASSERT_EQ(getNumRecords(), 1);
-    ASSERT_EQ(getDataSize(), 5);
-};
-
 }  // namespace
 }  // namespace mongo
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_stats.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_stats.cpp
deleted file mode 100644
index da3d2f461e5..00000000000
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_stats.cpp
+++ /dev/null
@@ -1,150 +0,0 @@
-/**
- *    Copyright (C) 2022-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/db/storage/wiredtiger/wiredtiger_stats.h"
-
-#include "mongo/bson/bsonobjbuilder.h"
-#include "mongo/db/storage/wiredtiger/wiredtiger_util.h"
-
-namespace mongo {
-namespace {
-
-enum class StatType { kData, kWait };
-
-struct StatInfo {
-    StringData name;
-    StatType type;
-};
-
-const stdx::unordered_map<int, StatInfo> kWiredTigerStatCodeToStatInfo = {
-    {WT_STAT_SESSION_BYTES_READ, {"bytesRead"_sd, StatType::kData}},
-    {WT_STAT_SESSION_BYTES_WRITE, {"bytesWritten"_sd, StatType::kData}},
-    {WT_STAT_SESSION_LOCK_DHANDLE_WAIT, {"handleLock"_sd, StatType::kWait}},
-    {WT_STAT_SESSION_READ_TIME, {"timeReadingMicros"_sd, StatType::kData}},
-    {WT_STAT_SESSION_WRITE_TIME, {"timeWritingMicros"_sd, StatType::kData}},
-    {WT_STAT_SESSION_LOCK_SCHEMA_WAIT, {"schemaLock"_sd, StatType::kWait}},
-    {WT_STAT_SESSION_CACHE_TIME, {"cache"_sd, StatType::kWait}}};
-
-}  // namespace
-
-WiredTigerStats::WiredTigerStats(WT_SESSION* session) {
-    invariant(session);
-
-    WT_CURSOR* c;
-    uassert(ErrorCodes::CursorNotFound,
-            "Unable to open statistics cursor",
-            !session->open_cursor(session, "statistics:session", nullptr, "statistics=(fast)", &c));
-
-    ScopeGuard guard{[c] { c->close(c); }};
-
-    int32_t key;
-    uint64_t value;
-    while (c->next(c) == 0 && c->get_key(c, &key) == 0) {
-        fassert(51035, c->get_value(c, nullptr, nullptr, &value) == 0);
-        _stats[key] = WiredTigerUtil::castStatisticsValue<long long>(value);
-    }
-}
-
-BSONObj WiredTigerStats::toBSON() const {
-    boost::optional<BSONObjBuilder> dataSection;
-    boost::optional<BSONObjBuilder> waitSection;
-
-    for (auto&& [stat, value] : _stats) {
-        if (value == 0) {
-            continue;
-        }
-
-        auto it = kWiredTigerStatCodeToStatInfo.find(stat);
-        if (it == kWiredTigerStatCodeToStatInfo.end()) {
-            continue;
-        }
-        auto&& [name, type] = it->second;
-
-        auto appendToSection = [name = name,
-                                value = value](boost::optional<BSONObjBuilder>& section) {
-            if (!section) {
-                section.emplace();
-            }
-            section->append(name, value);
-        };
-
-        switch (type) {
-            case StatType::kData:
-                appendToSection(dataSection);
-                break;
-            case StatType::kWait:
-                appendToSection(waitSection);
-                break;
-        }
-    }
-
-    BSONObjBuilder builder;
-    if (dataSection) {
-        builder.append("data", dataSection->obj());
-    }
-    if (waitSection) {
-        builder.append("timeWaitingMicros", waitSection->obj());
-    }
-
-    return builder.obj();
-}
-
-std::unique_ptr<StorageStats> WiredTigerStats::clone() const {
-    return std::make_unique<WiredTigerStats>(*this);
-}
-
-WiredTigerStats& WiredTigerStats::operator=(WiredTigerStats&& other) {
-    _stats = std::move(other._stats);
-    return *this;
-}
-
-WiredTigerStats& WiredTigerStats::operator+=(const WiredTigerStats& other) {
-    for (auto&& [stat, value] : other._stats) {
-        _stats[stat] += value;
-    }
-    return *this;
-}
-
-StorageStats& WiredTigerStats::operator+=(const StorageStats& other) {
-    return *this += checked_cast<const WiredTigerStats&>(other);
-}
-
-WiredTigerStats& WiredTigerStats::operator-=(const WiredTigerStats& other) {
-    for (auto const& otherStat : other._stats) {
-        _stats[otherStat.first] -= otherStat.second;
-    }
-    return (*this);
-}
-
-StorageStats& WiredTigerStats::operator-=(const StorageStats& other) {
-    *this -= checked_cast<const WiredTigerStats&>(other);
-    return (*this);
-}
-
-}  // namespace mongo
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_stats.h b/src/mongo/db/storage/wiredtiger/wiredtiger_stats.h
deleted file mode 100644
index d35a582cd34..00000000000
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_stats.h
+++ /dev/null
@@ -1,72 +0,0 @@
-/**
- *    Copyright (C) 2022-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 <wiredtiger.h>
-
-#include "mongo/db/storage/storage_stats.h"
-
-namespace mongo {
-
-class WiredTigerStats final : public StorageStats {
-public:
-    /**
-     * Construct a new WiredTigerStats object with the statistics of the specified session.
-     */
-    WiredTigerStats(WT_SESSION*);
-
-    WiredTigerStats() = default;
-    WiredTigerStats(const WiredTigerStats&) = default;
-    WiredTigerStats(WiredTigerStats&&) = default;
-
-    BSONObj toBSON() const final;
-
-    std::unique_ptr<StorageStats> clone() const final;
-
-    WiredTigerStats& operator=(WiredTigerStats&&);
-
-    StorageStats& operator+=(const StorageStats&) final;
-
-    WiredTigerStats& operator+=(const WiredTigerStats&);
-
-    StorageStats& operator-=(const StorageStats&) final;
-
-    WiredTigerStats& operator-=(const WiredTigerStats&);
-
-protected:
-    std::map<int, long long> _stats;
-};
-
-inline WiredTigerStats operator-(WiredTigerStats lhs, const WiredTigerStats& rhs) {
-    lhs -= rhs;
-    return lhs;
-}
-
-}  // namespace mongo
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_stats_test.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_stats_test.cpp
deleted file mode 100644
index de55e539056..00000000000
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_stats_test.cpp
+++ /dev/null
@@ -1,345 +0,0 @@
-/**
- *    Copyright (C) 2022-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::kWiredTiger
-
-#include "mongo/db/storage/wiredtiger/wiredtiger_stats.h"
-#include "mongo/db/storage/wiredtiger/wiredtiger_util.h"
-#include "mongo/logv2/log.h"
-#include "mongo/unittest/log_test.h"
-#include "mongo/unittest/temp_dir.h"
-#include "mongo/unittest/unittest.h"
-#include <memory>
-
-namespace mongo {
-namespace {
-
-#define ASSERT_WT_OK(result) ASSERT_EQ(result, 0) << wiredtiger_strerror(result)
-
-class WiredTigerStatsTest : public unittest::Test {
-protected:
-    void setUp() override {
-        openConnectionAndCreateSession();
-        // Prepare data to be read by tests. Reading data written within the same transaction does
-        // not count towards bytes read into cache.
-
-        // Tests in the fixture do up to _maxReads reads.
-        for (int64_t i = 0; i < _kMaxReads; ++i) {
-            // Make the write big enough to span different pages.
-            writeAtKey(std::string(20000, 'a'), i);
-        }
-
-        // Closing the connection will ensure that tests actually have to read into cache.
-        closeConnection();
-        openConnectionAndCreateSession();
-    }
-
-    void tearDown() override {
-        closeConnection();
-    }
-
-    void openConnectionAndCreateSession() {
-        ASSERT_WT_OK(
-            wiredtiger_open(_path.path().c_str(), nullptr, "create,statistics=(fast),", &_conn));
-        ASSERT_WT_OK(_conn->open_session(_conn, nullptr, "isolation=snapshot", &_session));
-        ASSERT_WT_OK(_session->create(
-            _session, _uri.c_str(), "type=file,key_format=q,value_format=u,log=(enabled=false)"));
-    }
-
-    void closeConnection() {
-        ASSERT_EQ(_conn->close(_conn, nullptr), 0);
-    }
-
-    /**
-     * Writes some data using WT. Causes the bytesWritten stat to be incremented, and may also
-     * increment timeWritingMicros.
-     */
-    void write() {
-        writeAtKey(std::string(_writeKey + 1, 'a'), _writeKey);
-        ++_writeKey;
-    }
-
-    /**
-     * Writes the specified data using WT. Causes the bytesWritten stat to be incremented, and may
-     * also increment timeWritingMicros.
-     */
-    void write(const std::string& data) {
-        writeAtKey(data, _writeKey);
-        ++_writeKey;
-    }
-
-    /**
-     * Writes at the specified key to WT.
-     */
-    void writeAtKey(const std::string& data, int64_t key) {
-        ASSERT_WT_OK(_session->begin_transaction(_session, nullptr));
-
-        WT_CURSOR* cursor;
-        ASSERT_WT_OK(_session->open_cursor(_session, _uri.c_str(), nullptr, nullptr, &cursor));
-
-        cursor->set_key(cursor, key);
-
-        WT_ITEM item{data.data(), data.size()};
-        cursor->set_value(cursor, &item);
-
-        ASSERT_WT_OK(cursor->insert(cursor));
-        ASSERT_WT_OK(cursor->close(cursor));
-        ASSERT_WT_OK(_session->commit_transaction(_session, nullptr));
-
-        // Without a checkpoint, an operation is not guaranteed to write to disk.
-        ASSERT_WT_OK(_session->checkpoint(_session, nullptr));
-    }
-
-    /**
-     * Reads at the specified key from WT.
-     */
-    void readAtKey(int64_t key) {
-        ASSERT_WT_OK(_session->begin_transaction(_session, nullptr));
-
-        WT_CURSOR* cursor;
-        ASSERT_WT_OK(_session->open_cursor(_session, _uri.c_str(), nullptr, nullptr, &cursor));
-
-        cursor->set_key(cursor, key);
-        ASSERT_WT_OK(cursor->search(cursor));
-
-        WT_ITEM value;
-        ASSERT_WT_OK(cursor->get_value(cursor, &value));
-
-        ASSERT_WT_OK(cursor->close(cursor));
-        ASSERT_WT_OK(_session->commit_transaction(_session, nullptr));
-    }
-
-    /**
-     * Reads fixture data from WT. Causes the bytesRead stat to be incremented. May also cause
-     * timeReadingMicros to be incremented, but not always. This function can only be called up to
-     * _kMaxReads times within a test.
-     */
-    void read() {
-        ASSERT_LT(_readKey, _kMaxReads);
-        readAtKey(_readKey++);
-    }
-
-    /**
-     * Reads data written by the test from WT. Causes the bytesRead stat to be incremented. May
-     * also cause timeReadingMicros to be incremented, but not always.
-     */
-    void readTestWrites() {
-        for (int64_t i = _kMaxReads; i < _writeKey; i++) {
-            readAtKey(i);
-        }
-    }
-
-    unittest::TempDir _path{"wiredtiger_operation_stats_test"};
-    std::string _uri{"table:wiredtiger_operation_stats_test"};
-    WT_CONNECTION* _conn;
-    WT_SESSION* _session;
-    /* Number of reads the fixture will prepare in setUp(), consequently max amount of times read()
-     * can be called in a test.  */
-    static constexpr int64_t _kMaxReads = 2;
-    /* Next key to be used by read(), must be initialized at 0. */
-    int64_t _readKey = 0;
-    /* Next key to be used by write(), must be initialized >= _kMaxReads. */
-    int64_t _writeKey = _kMaxReads;
-};
-
-TEST_F(WiredTigerStatsTest, EmptySession) {
-    // Increase log component verbosity for WiredTiger
-    auto verbosityGuard = unittest::MinimumLoggedSeverityGuard{logv2::LogComponent::kWiredTiger,
-                                                               logv2::LogSeverity::Debug(5)};
-    auto verboseConfig = WiredTigerUtil::generateWTVerboseConfiguration();
-    ASSERT_OK(wtRCToStatus(_conn->reconfigure(_conn, verboseConfig.c_str()), nullptr));
-
-    // Read and write statistics should be empty. Check "data" field does not exist. "wait" fields
-    // such as the schemaLock might have some value.
-    auto statsBson = WiredTigerStats{_session}.toBSON();
-
-    {
-        BSONObjBuilder bob;
-        ASSERT_OK(WiredTigerUtil::exportTableToBSON(_session, "statistics:", "", &bob));
-        LOGV2(9032000, "Connection statistics", "stats"_attr = bob.obj());
-    }
-
-    ASSERT_FALSE(statsBson.hasField("data")) << statsBson;
-}
-
-TEST_F(WiredTigerStatsTest, SessionWithWrite) {
-    write();
-
-    auto statsObj = WiredTigerStats{_session}.toBSON();
-    auto dataSection = statsObj["data"];
-    ASSERT_EQ(dataSection.type(), BSONType::Object) << statsObj;
-
-    ASSERT(dataSection["bytesWritten"]) << statsObj;
-    for (auto&& [name, value] : dataSection.Obj()) {
-        ASSERT_EQ(value.type(), BSONType::NumberLong) << statsObj;
-        ASSERT_GT(value.numberLong(), 0) << statsObj;
-    }
-}
-
-TEST_F(WiredTigerStatsTest, SessionWithRead) {
-    read();
-
-    auto statsObj = WiredTigerStats{_session}.toBSON();
-
-    auto dataSection = statsObj["data"];
-    ASSERT_EQ(dataSection.type(), BSONType::Object) << statsObj;
-
-    ASSERT(dataSection["bytesRead"]) << statsObj;
-    for (auto&& [name, value] : dataSection.Obj()) {
-        ASSERT_EQ(value.type(), BSONType::NumberLong) << statsObj;
-        ASSERT_GT(value.numberLong(), 0) << statsObj;
-    }
-}
-
-TEST_F(WiredTigerStatsTest, SessionWithLargeWriteAndLargeRead) {
-    auto remaining = static_cast<int64_t>(std::numeric_limits<uint32_t>::max()) + 1;
-    while (remaining > 0) {
-        std::string data(1024 * 1024, 'a');
-        remaining -= data.size();
-        write(data);
-    }
-
-    auto statsObj = WiredTigerStats{_session}.toBSON();
-    ASSERT_GT(statsObj["data"]["bytesWritten"].numberLong(), std::numeric_limits<uint32_t>::max())
-        << statsObj;
-
-    // Closing the connection will ensure that tests actually have to read into cache.
-    closeConnection();
-    openConnectionAndCreateSession();
-
-    readTestWrites();
-
-    statsObj = WiredTigerStats{_session}.toBSON();
-    ASSERT_GT(statsObj["data"]["bytesRead"].numberLong(), std::numeric_limits<uint32_t>::max())
-        << statsObj;
-}
-
-TEST_F(WiredTigerStatsTest, OperationsAddToSessionStats) {
-    std::vector<std::unique_ptr<WiredTigerStats>> operationStats;
-
-    write();
-    WiredTigerStats firstWrite(_session);
-    operationStats.push_back(std::make_unique<WiredTigerStats>(firstWrite - WiredTigerStats{}));
-    read();
-    WiredTigerStats firstRead(_session);
-    operationStats.push_back(std::make_unique<WiredTigerStats>(firstRead - firstWrite));
-    write();
-    WiredTigerStats secondWrite(_session);
-    operationStats.push_back(std::make_unique<WiredTigerStats>(secondWrite - firstRead));
-    read();
-    WiredTigerStats secondRead(_session);
-    operationStats.push_back(std::make_unique<WiredTigerStats>(secondRead - secondWrite));
-
-    const WiredTigerStats& fetchedSessionStats = secondRead;
-
-    long long bytesWritten = 0;
-    long long timeWritingMicros = 0;
-    long long bytesRead = 0;
-    long long timeReadingMicros = 0;
-
-    WiredTigerStats addedSessionStats;
-
-    for (auto&& op : operationStats) {
-        auto statsObj = op->toBSON();
-
-        bytesWritten += statsObj["data"]["bytesWritten"].numberLong();
-        timeWritingMicros += statsObj["data"]["timeWritingMicros"].numberLong();
-        bytesRead += statsObj["data"]["bytesRead"].numberLong();
-        timeReadingMicros += statsObj["data"]["timeReadingMicros"].numberLong();
-
-        addedSessionStats += *op;
-    }
-
-    auto addedObj = addedSessionStats.toBSON();
-    auto dataSection = addedObj["data"];
-    ASSERT_EQ(dataSection.type(), BSONType::Object) << addedObj;
-    ASSERT_EQ(dataSection["bytesWritten"].numberLong(), bytesWritten) << addedObj;
-    ASSERT_EQ(dataSection["timeWritingMicros"].numberLong(), timeWritingMicros) << addedObj;
-    ASSERT_EQ(dataSection["bytesRead"].numberLong(), bytesRead) << addedObj;
-    ASSERT_EQ(dataSection["timeReadingMicros"].numberLong(), timeReadingMicros) << addedObj;
-
-    auto fetchedObj = fetchedSessionStats.toBSON();
-    auto fetchedDataSection = fetchedObj["data"];
-    ASSERT_EQ(fetchedDataSection.type(), BSONType::Object) << fetchedObj;
-    ASSERT_EQ(fetchedDataSection["bytesWritten"].numberLong(), bytesWritten) << fetchedObj;
-    ASSERT_EQ(fetchedDataSection["timeWritingMicros"].numberLong(), timeWritingMicros)
-        << fetchedObj;
-    ASSERT_EQ(fetchedDataSection["bytesRead"].numberLong(), bytesRead) << fetchedObj;
-    ASSERT_EQ(fetchedDataSection["timeReadingMicros"].numberLong(), timeReadingMicros)
-        << fetchedObj;
-}
-
-TEST_F(WiredTigerStatsTest, OperationsSubtractToZero) {
-    std::vector<std::unique_ptr<WiredTigerStats>> operationStats;
-
-    write();
-    WiredTigerStats firstWrite(_session);
-    operationStats.push_back(std::make_unique<WiredTigerStats>(firstWrite - WiredTigerStats{}));
-    read();
-    WiredTigerStats firstRead(_session);
-    operationStats.push_back(std::make_unique<WiredTigerStats>(firstRead - firstWrite));
-    write();
-    WiredTigerStats secondWrite(_session);
-    operationStats.push_back(std::make_unique<WiredTigerStats>(secondWrite - firstRead));
-    read();
-    WiredTigerStats secondRead(_session);
-    operationStats.push_back(std::make_unique<WiredTigerStats>(secondRead - secondWrite));
-
-    WiredTigerStats& fetchedSessionStats = secondRead;
-
-    // Assert fetchedSessionStats was not zero before checking subtract results in it being zero.
-    // We ignore the time statistics as those might still be 0 from time to time.
-    auto preSubtractObj = fetchedSessionStats.toBSON();
-    auto preSubtract = preSubtractObj["data"];
-    ASSERT_EQ(preSubtract.type(), BSONType::Object) << preSubtractObj;
-    ASSERT_GT(preSubtract["bytesWritten"].numberLong(), 0) << preSubtractObj;
-    ASSERT_GT(preSubtract["bytesRead"].numberLong(), 0) << preSubtractObj;
-
-    for (auto&& op : operationStats) {
-        fetchedSessionStats -= *op;
-    }
-
-    auto subtractedObj = fetchedSessionStats.toBSON();
-    ASSERT_BSONOBJ_EQ(subtractedObj, BSONObj{});
-}
-
-TEST_F(WiredTigerStatsTest, Clone) {
-    write();
-
-    WiredTigerStats stats{_session};
-    auto clone = stats.clone();
-
-    ASSERT_BSONOBJ_EQ(stats.toBSON(), clone->toBSON());
-
-    stats += *clone;
-    ASSERT_BSONOBJ_NE(stats.toBSON(), clone->toBSON());
-}
-
-}  // namespace
-}  // namespace mongo
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_util.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_util.cpp
index 1afdb79babc..de31ec10751 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_util.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_util.cpp
@@ -37,15 +37,12 @@
 
 #include <boost/filesystem.hpp>
 #include <boost/filesystem/path.hpp>
-#include <pcrecpp.h>
 
 #include "mongo/base/simple_string_data_comparator.h"
 #include "mongo/bson/bsonobjbuilder.h"
-#include "mongo/bson/json.h"
-#include "mongo/db/concurrency/exception_util.h"
-#include "mongo/db/concurrency/exception_util_gen.h"
 #include "mongo/db/concurrency/temporarily_unavailable_exception.h"
 #include "mongo/db/concurrency/write_conflict_exception.h"
+#include "mongo/db/concurrency/write_conflict_exception_gen.h"
 #include "mongo/db/global_settings.h"
 #include "mongo/db/server_options_general_gen.h"
 #include "mongo/db/snapshot_window_options_gen.h"
@@ -57,6 +54,7 @@
 #include "mongo/db/storage/wiredtiger/wiredtiger_session_cache.h"
 #include "mongo/logv2/log.h"
 #include "mongo/util/assert_util.h"
+#include "mongo/util/fail_point.h"
 #include "mongo/util/processinfo.h"
 #include "mongo/util/scopeguard.h"
 #include "mongo/util/static_immortal.h"
@@ -64,13 +62,12 @@
 #include "mongo/util/testing_proctor.h"
 
 // From src/third_party/wiredtiger/src/include/txn.h
-#define WT_TXN_ROLLBACK_REASON_OLDEST_FOR_EVICTION \
-    "oldest pinned transaction ID rolled back for eviction"
+#define WT_TXN_ROLLBACK_REASON_CACHE "oldest pinned transaction ID rolled back for eviction"
 
-#define WT_TXN_ROLLBACK_REASON_TOO_LARGE_FOR_CACHE \
-    "transaction is too large and will not fit in the storage engine cache"
 namespace mongo {
 
+MONGO_FAIL_POINT_DEFINE(crashAfterUpdatingFirstTableLoggingSettings);
+
 namespace {
 
 const std::string kTableChecksFileName = "_wt_table_checks";
@@ -78,6 +75,41 @@ const std::string kTableExtension = ".wt";
 const std::string kWiredTigerBackupFile = "WiredTiger.backup";
 
 /**
+ * Returns true if the 'kTableChecksFileName' file exists in the dbpath.
+ *
+ * Must be called before createTableChecksFile() or removeTableChecksFile() to get accurate results.
+ */
+bool hasPreviouslyIncompleteTableChecks() {
+    auto path = boost::filesystem::path(storageGlobalParams.dbpath) /
+        boost::filesystem::path(kTableChecksFileName);
+
+    return boost::filesystem::exists(path);
+}
+
+/**
+ * Creates the 'kTableChecksFileName' file in the dbpath.
+ */
+void createTableChecksFile() {
+    auto path = boost::filesystem::path(storageGlobalParams.dbpath) /
+        boost::filesystem::path(kTableChecksFileName);
+
+    boost::filesystem::ofstream fileStream(path);
+    fileStream << "This file indicates that a WiredTiger table check operation is in progress or "
+                  "incomplete."
+               << std::endl;
+    if (fileStream.fail()) {
+        LOGV2_FATAL_NOTRACE(4366400,
+                            "Failed to write to file",
+                            "file"_attr = path.generic_string(),
+                            "error"_attr = errnoWithDescription());
+    }
+    fileStream.close();
+
+    fassertNoTrace(4366401, fsyncFile(path));
+    fassertNoTrace(4366402, fsyncParentDirectory(path));
+}
+
+/**
  * Removes the 'kTableChecksFileName' file in the dbpath, if it exists.
  */
 void removeTableChecksFile() {
@@ -127,87 +159,34 @@ void setTableWriteTimestampAssertion(WiredTigerSessionCache* sessionCache,
 
 using std::string;
 
+Mutex WiredTigerUtil::_tableLoggingInfoMutex =
+    MONGO_MAKE_LATCH("WiredTigerUtil::_tableLoggingInfoMutex");
+WiredTigerUtil::TableLoggingInfo WiredTigerUtil::_tableLoggingInfo;
+
 bool wasRollbackReasonCachePressure(WT_SESSION* session) {
     if (session) {
         const auto reason = session->get_rollback_reason(session);
         if (reason) {
-            return strncmp(WT_TXN_ROLLBACK_REASON_OLDEST_FOR_EVICTION,
+            return strncmp(WT_TXN_ROLLBACK_REASON_CACHE,
                            reason,
-                           sizeof(WT_TXN_ROLLBACK_REASON_OLDEST_FOR_EVICTION)) == 0;
+                           sizeof(WT_TXN_ROLLBACK_REASON_CACHE)) == 0;
         }
     }
     return false;
 }
 
-/**
- * Configured WT cache is deemed insufficient for a transaction when its dirty bytes in cache
- * exceed a certain threshold on the proportion of total cache which is used by transaction.
- *
- * For instance, if the transaction uses 80% of WT cache and the threshold is set to 75%, the
- * transaction is considered too large.
- */
-bool isCacheInsufficientForTransaction(WT_SESSION* session, double threshold) {
-    StatusWith<int64_t> txnDirtyBytes = WiredTigerUtil::getStatisticsValue(
-        session, "statistics:session", "", WT_STAT_SESSION_TXN_BYTES_DIRTY);
-    if (!txnDirtyBytes.isOK()) {
-        tasserted(6190900,
-                  str::stream() << "unable to gather the WT session's txn dirty bytes: "
-                                << txnDirtyBytes.getStatus());
-    }
-
-    StatusWith<int64_t> cacheDirtyBytes = WiredTigerUtil::getStatisticsValue(
-        session, "statistics:", "", WT_STAT_CONN_CACHE_BYTES_DIRTY);
-    if (!cacheDirtyBytes.isOK()) {
-        tasserted(6190901,
-                  str::stream() << "unable to gather the WT connection's cache dirty bytes: "
-                                << txnDirtyBytes.getStatus());
-    }
-
-
-    double txnBytesDirtyOverCacheBytesDirty =
-        static_cast<double>(txnDirtyBytes.getValue()) / cacheDirtyBytes.getValue();
-
-    LOGV2_DEBUG(6190902,
-                2,
-                "Checking if transaction can eventually succeed",
-                "txnDirtyBytes"_attr = txnDirtyBytes.getValue(),
-                "cacheDirtyBytes"_attr = cacheDirtyBytes.getValue(),
-                "txnBytesDirtyOverCacheBytesDirty"_attr = txnBytesDirtyOverCacheBytesDirty,
-                "threshold"_attr = threshold);
-
-    return txnBytesDirtyOverCacheBytesDirty > threshold;
-}
-
 Status wtRCToStatus_slow(int retCode, WT_SESSION* session, StringData prefix) {
     if (retCode == 0)
         return Status::OK();
 
-    const auto generateContextStrStream = [&](StringData reason) {
-        str::stream contextStrStream;
-        if (!prefix.empty())
-            contextStrStream << prefix << " ";
-        contextStrStream << retCode << ": " << reason;
-
-        return contextStrStream;
-    };
-
     if (retCode == WT_ROLLBACK) {
-        double cacheThreshold = gTransactionTooLargeForCacheThreshold.load();
-        bool txnTooLargeEnabled = cacheThreshold < 1.0;
-        bool temporarilyUnavailableEnabled = gEnableTemporarilyUnavailableExceptions.load();
-        bool reasonWasCachePressure = (txnTooLargeEnabled || temporarilyUnavailableEnabled) &&
-            wasRollbackReasonCachePressure(session);
-
-        if (reasonWasCachePressure) {
-            if (txnTooLargeEnabled && isCacheInsufficientForTransaction(session, cacheThreshold)) {
-                auto s = generateContextStrStream(WT_TXN_ROLLBACK_REASON_TOO_LARGE_FOR_CACHE);
-                throwTransactionTooLargeForCache(s);
-            }
-
-            if (temporarilyUnavailableEnabled) {
-                auto s = generateContextStrStream(WT_TXN_ROLLBACK_REASON_OLDEST_FOR_EVICTION);
-                throw TemporarilyUnavailableException(s);
-            }
+        if (gEnableTemporarilyUnavailableExceptions.load() &&
+            wasRollbackReasonCachePressure(session)) {
+            str::stream s;
+            if (!prefix.empty())
+                s << prefix << " ";
+            s << retCode << ": " << WT_TXN_ROLLBACK_REASON_CACHE;
+            throw TemporarilyUnavailableException(s);
         }
 
         throw WriteConflictException(prefix);
@@ -216,7 +195,10 @@ Status wtRCToStatus_slow(int retCode, WT_SESSION* session, StringData prefix) {
     // Don't abort on WT_PANIC when repairing, as the error will be handled at a higher layer.
     fassert(28559, retCode != WT_PANIC || storageGlobalParams.repair);
 
-    auto s = generateContextStrStream(wiredtiger_strerror(retCode));
+    str::stream s;
+    if (!prefix.empty())
+        s << prefix << " ";
+    s << retCode << ": " << wiredtiger_strerror(retCode);
 
     if (retCode == EINVAL) {
         return Status(ErrorCodes::BadValue, s);
@@ -465,7 +447,7 @@ StatusWith<int64_t> WiredTigerUtil::checkApplicationMetadataFormatVersion(Operat
 
 // static
 Status WiredTigerUtil::checkTableCreationOptions(const BSONElement& configElem) {
-    invariant(configElem.fieldNameStringData() == WiredTigerUtil::kConfigStringField);
+    invariant(configElem.fieldNameStringData() == "configString");
 
     if (configElem.type() != String) {
         return {ErrorCodes::TypeMismatch, "'configString' must be a string."};
@@ -602,12 +584,9 @@ logv2::LogSeverity getWTLOGV2SeverityLevel(const BSONObj& obj) {
             return logv2::LogSeverity::Info();
         case WT_VERBOSE_INFO:
             return logv2::LogSeverity::Log();
+        case WT_VERBOSE_DEBUG:
+            return logv2::LogSeverity::Debug(1);
         default:
-            // MongoDB enables some WT debug compnonents by default. If performed a 1:1
-            // translation from WT log severity levels, MongoDB would not log anything
-            // below default level Log, even if a Debug message came through the message
-            // handler.  To solve this, we upgrade all Debug messages to the Log level
-            // to ensure they are seen.
             return logv2::LogSeverity::Log();
     }
 }
@@ -847,7 +826,20 @@ int WiredTigerUtil::verifyTable(OperationContext* opCtx,
 }
 
 void WiredTigerUtil::notifyStartupComplete() {
-    removeTableChecksFile();
+    {
+        stdx::lock_guard<Latch> lk(_tableLoggingInfoMutex);
+        invariant(_tableLoggingInfo.isInitializing);
+        _tableLoggingInfo.isInitializing = false;
+    }
+
+    if (!storageGlobalParams.readOnly) {
+        removeTableChecksFile();
+    }
+}
+
+void WiredTigerUtil::resetTableLoggingInfo() {
+    stdx::lock_guard<Latch> lk(_tableLoggingInfoMutex);
+    _tableLoggingInfo = TableLoggingInfo();
 }
 
 bool WiredTigerUtil::useTableLogging(const NamespaceString& nss) {
@@ -881,16 +873,120 @@ bool WiredTigerUtil::useTableLogging(const NamespaceString& nss) {
 }
 
 Status WiredTigerUtil::setTableLogging(OperationContext* opCtx, const std::string& uri, bool on) {
+    // Try to close as much as possible to avoid EBUSY errors.
+    WiredTigerRecoveryUnit::get(opCtx)->getSession()->closeAllCursors(uri);
+    WiredTigerSessionCache* sessionCache = WiredTigerRecoveryUnit::get(opCtx)->getSessionCache();
+    sessionCache->closeAllCursors(uri);
+
+    invariant(!storageGlobalParams.readOnly);
+    stdx::lock_guard<Latch> lk(_tableLoggingInfoMutex);
+
+    // Update the table logging settings regardless if we're no longer starting up the process.
+    if (!_tableLoggingInfo.isInitializing) {
+        return _setTableLogging(sessionCache, uri, on);
+    }
+
+    // During the start up process, the table logging settings are checked for each table to verify
+    // that they are set appropriately. We can speed this process up by assuming that the logging
+    // setting is identical for each table.
+    // We cross reference the logging settings for the first table and if it isn't correctly set, we
+    // change the logging settings for all tables during start up.
+    // In the event that the server wasn't shutdown cleanly, the logging settings will be modified
+    // for all tables as a safety precaution, or if repair mode is running.
+    if (_tableLoggingInfo.isFirstTable && hasPreviouslyIncompleteTableChecks()) {
+        _tableLoggingInfo.hasPreviouslyIncompleteTableChecks = true;
+    }
+
     if (gWiredTigerSkipTableLoggingChecksOnStartup) {
+        if (_tableLoggingInfo.hasPreviouslyIncompleteTableChecks) {
+            LOGV2_FATAL_NOTRACE(
+                5548300,
+                "Cannot use the 'wiredTigerSkipTableLoggingChecksOnStartup' startup parameter when "
+                "there are previously incomplete table checks");
+        }
+
+        // Only log this warning once.
+        if (_tableLoggingInfo.isFirstTable) {
+            _tableLoggingInfo.isFirstTable = false;
+            LOGV2_WARNING_OPTIONS(
+                5548301,
+                {logv2::LogTag::kStartupWarnings},
+                "Skipping table logging checks for all existing WiredTiger tables on startup",
+                "wiredTigerSkipTableLoggingChecksOnStartup"_attr =
+                    gWiredTigerSkipTableLoggingChecksOnStartup);
+        }
+
         LOGV2_DEBUG(5548302, 1, "Skipping table logging check", "uri"_attr = uri);
         return Status::OK();
     }
 
-    // Try to close as much as possible to avoid EBUSY errors.
-    WiredTigerRecoveryUnit::get(opCtx)->getSession()->closeAllCursors(uri);
-    WiredTigerSessionCache* sessionCache = WiredTigerRecoveryUnit::get(opCtx)->getSessionCache();
-    sessionCache->closeAllCursors(uri);
+    if (storageGlobalParams.repair || _tableLoggingInfo.hasPreviouslyIncompleteTableChecks) {
+        if (_tableLoggingInfo.isFirstTable) {
+            _tableLoggingInfo.isFirstTable = false;
+            if (!_tableLoggingInfo.hasPreviouslyIncompleteTableChecks) {
+                createTableChecksFile();
+            }
 
+            LOGV2(4366405,
+                  "Modifying the table logging settings for all existing WiredTiger tables",
+                  "loggingEnabled"_attr = on,
+                  "repair"_attr = storageGlobalParams.repair,
+                  "hasPreviouslyIncompleteTableChecks"_attr =
+                      _tableLoggingInfo.hasPreviouslyIncompleteTableChecks);
+        }
+
+        return _setTableLogging(sessionCache, uri, on);
+    }
+
+    if (!_tableLoggingInfo.isFirstTable) {
+        if (_tableLoggingInfo.changeTableLogging) {
+            return _setTableLogging(sessionCache, uri, on);
+        }
+
+        // The table logging settings do not need to be modified.
+        return Status::OK();
+    }
+
+    invariant(_tableLoggingInfo.isFirstTable);
+    invariant(!_tableLoggingInfo.hasPreviouslyIncompleteTableChecks);
+
+    // When repair or a forced modification to the table logging settings isn't running, check that
+    // the first table is the catalog.
+    invariant(uri == "table:_mdb_catalog", str::stream() << "First table checked was: " << uri);
+    _tableLoggingInfo.isFirstTable = false;
+
+    // Check if the first tables logging settings need to be modified.
+    const std::string setting = on ? "log=(enabled=true)" : "log=(enabled=false)";
+    const std::string existingMetadata = getMetadataCreate(opCtx, uri).getValue();
+    if (existingMetadata.find(setting) != std::string::npos) {
+        // The table is running with the expected logging settings.
+        LOGV2(4366408,
+              "No table logging settings modifications are required for existing WiredTiger tables",
+              "loggingEnabled"_attr = on);
+        return Status::OK();
+    }
+
+    // The first table is running with the incorrect logging settings. All tables will need to have
+    // their logging settings modified.
+    _tableLoggingInfo.changeTableLogging = true;
+    createTableChecksFile();
+
+    LOGV2(4366406,
+          "Modifying the table logging settings for all existing WiredTiger tables",
+          "loggingEnabled"_attr = on);
+
+    Status status = _setTableLogging(sessionCache, uri, on);
+
+    if (MONGO_unlikely(crashAfterUpdatingFirstTableLoggingSettings.shouldFail())) {
+        LOGV2_FATAL_NOTRACE(
+            4366407, "Crashing due to 'crashAfterUpdatingFirstTableLoggingSettings' fail point");
+    }
+    return status;
+}
+
+Status WiredTigerUtil::_setTableLogging(WiredTigerSessionCache* sessionCache,
+                                        const std::string& uri,
+                                        bool on) {
     const std::string setting = on ? "log=(enabled=true)" : "log=(enabled=false)";
 
     // This method does some "weak" parsing to see if the table is in the expected logging
@@ -1194,44 +1290,5 @@ std::string WiredTigerUtil::generateWTVerboseConfiguration() {
     return cfg;
 }
 
-// static
-boost::optional<std::string> WiredTigerUtil::getConfigStringFromStorageOptions(
-    const BSONObj& options) {
-    if (auto wtElem = options[kWiredTigerEngineName]) {
-        BSONObj wtObj = wtElem.Obj();
-        if (auto configStringElem = wtObj.getField(kConfigStringField)) {
-            return configStringElem.String();
-        }
-    }
-
-    return boost::none;
-}
-
-// static
-BSONObj WiredTigerUtil::setConfigStringToStorageOptions(const BSONObj& options,
-                                                        const std::string& configString) {
-    // Storage options may contain settings for non-WiredTiger storage engines (e.g. inMemory).
-    // We should leave these settings intact.
-    auto wtElem = options[kWiredTigerEngineName];
-    auto wtObj = wtElem ? wtElem.Obj() : BSONObj();
-    return options.addFields(
-        BSON(kWiredTigerEngineName << wtObj.addFields(BSON(kConfigStringField << configString))));
-}
-
-void WiredTigerUtil::removeEncryptionFromConfigString(std::string* configString) {
-    static const StaticImmortal<pcrecpp::RE> encryptionOptsRegex(R"re(encryption=\([^\)]*\),?)re");
-    encryptionOptsRegex->GlobalReplace("", configString);
-}
-
-// static
-BSONObj WiredTigerUtil::getSanitizedStorageOptionsForSecondaryReplication(const BSONObj& options) {
-    auto configString = getConfigStringFromStorageOptions(options);
-    if (!configString) {
-        return options;
-    }
-
-    removeEncryptionFromConfigString(configString.get_ptr());
-    return setConfigStringToStorageOptions(options, *configString);
-}
 
 }  // namespace mongo
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_util.h b/src/mongo/db/storage/wiredtiger/wiredtiger_util.h
index a70c050b2ea..83bcd4e0293 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_util.h
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_util.h
@@ -158,8 +158,6 @@ private:
     WiredTigerUtil();
 
 public:
-    static constexpr StringData kConfigStringField = "configString"_sd;
-
     /**
      * Fetch the type and source fields out of the colgroup metadata.  'tableUri' must be a
      * valid table: uri.
@@ -317,6 +315,8 @@ public:
 
     static void notifyStartupComplete();
 
+    static void resetTableLoggingInfo();
+
     static bool useTableLogging(const NamespaceString& nss);
 
     static Status setTableLogging(OperationContext* opCtx, const std::string& uri, bool on);
@@ -334,32 +334,6 @@ public:
     template <typename T>
     static T castStatisticsValue(uint64_t statisticsValue);
 
-    /**
-     * Gets the WiredTiger configuration string from storage engine collection options.
-     */
-    static boost::optional<std::string> getConfigStringFromStorageOptions(const BSONObj& options);
-
-    /**
-     * Sets the WiredTiger configuration string to storage engine collection options.
-     */
-    static BSONObj setConfigStringToStorageOptions(const BSONObj& options,
-                                                   const std::string& configString);
-
-    /**
-     * Removes encryption configuration from a config string. Should only be applied on custom
-     * config strings on secondaries. Fixes an issue where encryption configuration might be
-     * replicated to non-encrypted nodes, or nodes with different encryption options, causing
-     * initial sync or replication to fail. See SERVER-68122.
-     */
-    static void removeEncryptionFromConfigString(std::string* configString);
-
-    /**
-     * Removes encryption configuration from storage engine collection options.
-     * See CollectionOptions.storageEngine and WiredTigerUtil::removeEncryptionFromConfigString().
-     * TODO(SERVER-81069): Remove this since it's intrinsically tied to encryption options only.
-     */
-    static BSONObj getSanitizedStorageOptionsForSecondaryReplication(const BSONObj& options);
-
 private:
     /**
      * Casts unsigned 64-bit statistics value to T.
@@ -367,6 +341,20 @@ private:
      */
     template <typename T>
     static T _castStatisticsValue(uint64_t statisticsValue, T maximumResultType);
+
+    static Status _setTableLogging(WiredTigerSessionCache* sessionCache,
+                                   const std::string& uri,
+                                   bool on);
+
+    // Used to keep track of the table logging setting modifications during start up. The mutex must
+    // be held prior to accessing any of the member variables in the struct.
+    static Mutex _tableLoggingInfoMutex;
+    static struct TableLoggingInfo {
+        bool isInitializing = true;
+        bool isFirstTable = true;
+        bool changeTableLogging = false;
+        bool hasPreviouslyIncompleteTableChecks = false;
+    } _tableLoggingInfo;
 };
 
 class WiredTigerConfigParser {
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_util_test.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_util_test.cpp
index c735b58bf50..088fb820474 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_util_test.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_util_test.cpp
@@ -452,119 +452,4 @@ TEST(WiredTigerUtilTest, GenerateVerboseConfiguration) {
     }
 }
 
-TEST(WiredTigerUtilTest, RemoveEncryptionFromConfigString) {
-    {  // Found at the middle.
-        std::string input{
-            "debug_mode=(table_logging=true,checkpoint_retention=4),encryption=(name=AES256-CBC,"
-            "keyid="
-            "\".system\"),extensions=[local={entry=mongo_addWiredTigerEncryptors,early_load=true},,"
-            "],"};
-        const std::string expectedOutput{
-            "debug_mode=(table_logging=true,checkpoint_retention=4),extensions=[local={entry=mongo_"
-            "addWiredTigerEncryptors,early_load=true},,],"};
-        WiredTigerUtil::removeEncryptionFromConfigString(&input);
-        ASSERT_EQUALS(input, expectedOutput);
-    }
-    {  // Found at start.
-        std::string input{
-            "encryption=(name=AES256-CBC,keyid=\".system\"),extensions=[local={entry=mongo_"
-            "addWiredTigerEncryptors,early_load=true},,],"};
-        const std::string expectedOutput{
-            "extensions=[local={entry=mongo_addWiredTigerEncryptors,early_load=true},,],"};
-        WiredTigerUtil::removeEncryptionFromConfigString(&input);
-        ASSERT_EQUALS(input, expectedOutput);
-    }
-    {  // Found at the end.
-        std::string input{
-            "debug_mode=(table_logging=true,checkpoint_retention=4),encryption=(name=AES256-CBC,"
-            "keyid=\".system\")"};
-        const std::string expectedOutput{"debug_mode=(table_logging=true,checkpoint_retention=4),"};
-        WiredTigerUtil::removeEncryptionFromConfigString(&input);
-        ASSERT_EQUALS(input, expectedOutput);
-    }
-    {  // Matches full configString.
-        std::string input{"encryption=(name=AES256-CBC,keyid=\".system\")"};
-        const std::string expectedOutput{""};
-        WiredTigerUtil::removeEncryptionFromConfigString(&input);
-        ASSERT_EQUALS(input, expectedOutput);
-    }
-    {  // Matches full configString, trailing comma.
-        std::string input{"encryption=(name=AES256-CBC,keyid=\".system\"),"};
-        const std::string expectedOutput{""};
-        WiredTigerUtil::removeEncryptionFromConfigString(&input);
-        ASSERT_EQUALS(input, expectedOutput);
-    }
-    {  // No match.
-        std::string input{"debug_mode=(table_logging=true,checkpoint_retention=4)"};
-        const std::string expectedOutput{"debug_mode=(table_logging=true,checkpoint_retention=4)"};
-        WiredTigerUtil::removeEncryptionFromConfigString(&input);
-        ASSERT_EQUALS(input, expectedOutput);
-    }
-    {  // No match, empty.
-        std::string input{""};
-        const std::string expectedOutput{""};
-        WiredTigerUtil::removeEncryptionFromConfigString(&input);
-        ASSERT_EQUALS(input, expectedOutput);
-    }
-    {  // Removes multiple instances.
-        std::string input{
-            "encryption=(name=AES256-CBC,keyid=\".system\"),debug_mode=(table_logging=true,"
-            "checkpoint_retention=4),encryption=(name=AES256-CBC,keyid=\".system\")"};
-        const std::string expectedOutput{"debug_mode=(table_logging=true,checkpoint_retention=4),"};
-        WiredTigerUtil::removeEncryptionFromConfigString(&input);
-        ASSERT_EQUALS(input, expectedOutput);
-    }
-}
-
-TEST(WiredTigerUtilTest, GetSanitizedStorageOptionsForSecondaryReplication) {
-    {  // Empty storage options.
-        auto input = BSONObj();
-        auto expectedOutput = input;
-        auto output = WiredTigerUtil::getSanitizedStorageOptionsForSecondaryReplication(input);
-        ASSERT_BSONOBJ_EQ(output, expectedOutput);
-    }
-    {
-        // Preserve WT config string without encryption options.
-        auto input = BSON("wiredTiger" << BSON("configString"
-                                               << "split_pct=88"));
-        auto expectedOutput = input;
-        auto output = WiredTigerUtil::getSanitizedStorageOptionsForSecondaryReplication(input);
-        ASSERT_BSONOBJ_EQ(output, expectedOutput);
-    }
-    {
-        // Remove encryption options from WT config string in results.
-        auto input = BSON(
-            "wiredTiger" << BSON("configString"
-                                 << "encryption=(name=AES256-CBC,keyid=\".system\"),split_pct=88"));
-        auto expectedOutput = BSON("wiredTiger" << BSON("configString"
-                                                        << "split_pct=88"));
-        auto output = WiredTigerUtil::getSanitizedStorageOptionsForSecondaryReplication(input);
-        ASSERT_BSONOBJ_EQ(output, expectedOutput);
-    }
-    {
-        // Leave non-WT settings intact.
-        auto input = BSON("inMemory" << BSON("configString"
-                                             << "split_pct=66"));
-        auto expectedOutput = input;
-        auto output = WiredTigerUtil::getSanitizedStorageOptionsForSecondaryReplication(input);
-        ASSERT_BSONOBJ_EQ(output, expectedOutput);
-    }
-    {
-        // Change only WT settings in storage options containing a mix of WT and non-WT settings.
-        auto input = BSON(
-            "inMemory" << BSON("configString"
-                               << "split_pct=66")
-                       << "wiredTiger"
-                       << BSON("configString"
-                               << "encryption=(name=AES256-CBC,keyid=\".system\"),split_pct=88"));
-        auto expectedOutput = BSON("inMemory" << BSON("configString"
-                                                      << "split_pct=66")
-                                              << "wiredTiger"
-                                              << BSON("configString"
-                                                      << "split_pct=88"));
-        auto output = WiredTigerUtil::getSanitizedStorageOptionsForSecondaryReplication(input);
-        ASSERT_BSONOBJ_EQ(output, expectedOutput);
-    }
-}
-
 }  // namespace mongo