SERVER-56002 SERVER-56023 Store Collection metadata in the Collection and reply on the copy-on-write machinery to keep it in sync with the durable catalog.

All updates to the metadata needs to happen through the Collection, moved interfaces from the DurableCatalog to the Collection.
Removed back pointer to Collection in IndexCatalogEntryImpl, interfaces now correctly take a const or non-const Collection. This should make its iterface const-correct to avoid making bugs where the copy-on-write system for Collections are bypassed.

Multikey handle is special as it needs to happen without exclusive access to the Collection. Implemented isolation for the Collection metadata when multikey is changed. It handles multi-doc transactions and is only commited to the Collection instance after the write to the durable catalog successfully commits.

listCollections and listIndexes can now safetly read the metadata cache without needing to read from the durable catalog making them safe to do without Collection level locks.

1 files changed, 14 insertions, 23 deletions

diff --git a/src/mongo/db/repl/storage_interface_impl.cpp b/src/mongo/db/repl/storage_interface_impl.cpp
index 67e991b944c..352f2799ecf 100644
--- a/src/mongo/db/repl/storage_interface_impl.cpp
+++ b/src/mongo/db/repl/storage_interface_impl.cpp
@@ -78,7 +78,6 @@
 #include "mongo/db/storage/checkpointer.h"
 #include "mongo/db/storage/control/journal_flusher.h"
 #include "mongo/db/storage/control/storage_control.h"
-#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/oplog_cap_maintainer_thread.h"
 #include "mongo/logv2/log.h"
 #include "mongo/util/assert_util.h"
@@ -261,7 +260,8 @@ StorageInterfaceImpl::createCollectionForBulkLoading(
             if (!idIndexSpec.isEmpty()) {
                 auto status = autoColl->getWritableCollection()
                                   ->getIndexCatalog()
-                                  ->createIndexOnEmptyCollection(opCtx.get(), idIndexSpec);
+                                  ->createIndexOnEmptyCollection(
+                                      opCtx.get(), autoColl->getWritableCollection(), idIndexSpec);
                 if (!status.getStatus().isOK()) {
                     return status.getStatus();
                 }
@@ -269,7 +269,8 @@ StorageInterfaceImpl::createCollectionForBulkLoading(
             for (auto&& spec : secondaryIndexSpecs) {
                 auto status = autoColl->getWritableCollection()
                                   ->getIndexCatalog()
-                                  ->createIndexOnEmptyCollection(opCtx.get(), spec);
+                                  ->createIndexOnEmptyCollection(
+                                      opCtx.get(), autoColl->getWritableCollection(), spec);
                 if (!status.getStatus().isOK()) {
                     return status.getStatus();
                 }
@@ -454,26 +455,16 @@ Status StorageInterfaceImpl::createOplog(OperationContext* opCtx, const Namespac
 }
 
 StatusWith<size_t> StorageInterfaceImpl::getOplogMaxSize(OperationContext* opCtx) {
-    // This writeConflictRetry loop protects callers from WriteConflictExceptions thrown by the
-    // storage engine running out of cache space, despite this operation not performing any writes.
-    return writeConflictRetry(
-        opCtx,
-        "StorageInterfaceImpl::getOplogMaxSize",
-        NamespaceString::kRsOplogNamespace.ns(),
-        [&]() -> StatusWith<size_t> {
-            AutoGetOplog oplogRead(opCtx, OplogAccessMode::kRead);
-            const auto& oplog = oplogRead.getCollection();
-            if (!oplog) {
-                return {ErrorCodes::NamespaceNotFound, "Your oplog doesn't exist."};
-            }
-            const auto options =
-                DurableCatalog::get(opCtx)->getCollectionOptions(opCtx, oplog->getCatalogId());
-            if (!options.capped)
-                return {ErrorCodes::BadValue,
-                        str::stream()
-                            << NamespaceString::kRsOplogNamespace.ns() << " isn't capped"};
-            return options.cappedSize;
-        });
+    AutoGetOplog oplogRead(opCtx, OplogAccessMode::kRead);
+    const auto& oplog = oplogRead.getCollection();
+    if (!oplog) {
+        return {ErrorCodes::NamespaceNotFound, "Your oplog doesn't exist."};
+    }
+    const auto options = oplog->getCollectionOptions();
+    if (!options.capped)
+        return {ErrorCodes::BadValue,
+                str::stream() << NamespaceString::kRsOplogNamespace.ns() << " isn't capped"};
+    return options.cappedSize;
 }
 
 Status StorageInterfaceImpl::createCollection(OperationContext* opCtx,