Update upstream source from tag 'upstream/6.0.0'master

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

27 files changed, 593 insertions, 1800 deletions

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