SERVER-52877 Unify how writable Collections instances are handled

CollectionCatalog::LifetimeMode has been removed. Catalog writes now
require that we are in an active WUOW.

Make it allowed to use WriteUnitOfWork when the server is in readOnly
mode. It does not open storage sessions, just allows registration of
RecoveryUnit callbacks that are executed when calling commit(). This
allows for the unification of code where we need to initialize Collection
instances even in readOnly mode.

Handling of enforcing readOnly has been pushed down to the RecordStore.
All interfaces that perform write now check if we are in readOnly mode
and throw if we are.

Catalog updates using the BatchedCollectionCatalogWriter class bypass
the Collection cloning if the batched CollectionCatalog instance already
has a uniquely owned copy (a previous write to this collection has been
requested). It is also not required to be in an active WUOW when the
BatchedCollectionCatalogWriter is used.

1 files changed, 19 insertions, 12 deletions

diff --git a/src/mongo/db/catalog/catalog_control.cpp b/src/mongo/db/catalog/catalog_control.cpp
index 510f45537a9..dadfed598c0 100644
--- a/src/mongo/db/catalog/catalog_control.cpp
+++ b/src/mongo/db/catalog/catalog_control.cpp
@@ -51,14 +51,19 @@ namespace catalog {
 namespace {
 void reopenAllDatabasesAndReloadCollectionCatalog(
     OperationContext* opCtx,
-    std::shared_ptr<const CollectionCatalog> catalog,
     StorageEngine* storageEngine,
     const MinVisibleTimestampMap& minVisibleTimestampMap,
     Timestamp stableTimestamp) {
+
     // Open all databases and repopulate the CollectionCatalog.
     LOGV2(20276, "openCatalog: reopening all databases");
-    boost::optional<BatchedCollectionCatalogWriter> catalogBatchWriter;
-    catalogBatchWriter.emplace(opCtx);
+
+    // Applies all Collection writes to the in-memory catalog in a single copy-on-write to the
+    // catalog. This avoids quadratic behavior where we iterate over every collection and perform
+    // writes where the catalog would be copied every time. boost::optional is used to be able to
+    // finish the write batch when encountering the oplog as other systems except immediate
+    // visibility for the oplog.
+    boost::optional<BatchedCollectionCatalogWriter> catalogWriter(opCtx);
 
     auto databaseHolder = DatabaseHolder::get(opCtx);
     std::vector<TenantDatabaseName> databasesToOpen = storageEngine->listDatabases();
@@ -67,10 +72,10 @@ void reopenAllDatabasesAndReloadCollectionCatalog(
             23992, 1, "openCatalog: dbholder reopening database", "db"_attr = tenantDbName);
         auto db = databaseHolder->openDb(opCtx, tenantDbName);
         invariant(db, str::stream() << "failed to reopen database " << tenantDbName.toString());
-        for (auto&& collNss : catalog->getAllCollectionNamesFromDb(opCtx, tenantDbName)) {
+        for (auto&& collNss :
+             catalogWriter.get()->getAllCollectionNamesFromDb(opCtx, tenantDbName)) {
             // Note that the collection name already includes the database component.
-            auto collection = catalog->lookupCollectionByNamespaceForMetadataWrite(
-                opCtx, CollectionCatalog::LifetimeMode::kInplace, collNss);
+            auto collection = catalogWriter.get()->lookupCollectionByNamespace(opCtx, collNss);
             invariant(collection,
                       str::stream()
                           << "failed to get valid collection pointer for namespace " << collNss);
@@ -87,7 +92,10 @@ void reopenAllDatabasesAndReloadCollectionCatalog(
                 // cost/effort.
                 auto minVisible = std::min(stableTimestamp,
                                            minVisibleTimestampMap.find(collection->uuid())->second);
-                collection->setMinimumVisibleSnapshot(minVisible);
+                auto writableCollection =
+                    catalogWriter.get()->lookupCollectionByUUIDForMetadataWrite(opCtx,
+                                                                                collection->uuid());
+                writableCollection->setMinimumVisibleSnapshot(minVisible);
             }
 
             // If this is the oplog collection, re-establish the replication system's cached pointer
@@ -97,9 +105,9 @@ void reopenAllDatabasesAndReloadCollectionCatalog(
 
                 // The oplog collection must be visible when establishing for repl. Finish our
                 // batched catalog write and continue on a new batch afterwards.
-                catalogBatchWriter.reset();
+                catalogWriter.reset();
                 collection->establishOplogCollectionForLogging(opCtx);
-                catalogBatchWriter.emplace(opCtx);
+                catalogWriter.emplace(opCtx);
             }
         }
     }
@@ -246,15 +254,14 @@ void openCatalog(OperationContext* opCtx,
         opCtx, reconcileResult.indexBuildsToRestart, reconcileResult.indexBuildsToResume);
 
     reopenAllDatabasesAndReloadCollectionCatalog(
-        opCtx, catalog, storageEngine, minVisibleTimestampMap, stableTimestamp);
+        opCtx, storageEngine, minVisibleTimestampMap, stableTimestamp);
 }
 
 
 void openCatalogAfterStorageChange(OperationContext* opCtx) {
     invariant(opCtx->lockState()->isW());
     auto storageEngine = opCtx->getServiceContext()->getStorageEngine();
-    auto catalog = CollectionCatalog::get(opCtx);
-    reopenAllDatabasesAndReloadCollectionCatalog(opCtx, catalog, storageEngine, {}, {});
+    reopenAllDatabasesAndReloadCollectionCatalog(opCtx, storageEngine, {}, {});
 }
 
 }  // namespace catalog