SERVER-52557 Publishing a writable Collection to the catalog is done with a special commit handler that runs after all regular commit handlers.

It is now safe to write to a writable Collection in a commit handler but not in a rollback handler.
We should not need to be able to write to a writable Collection that rolls back as the cloned instance will just be discarded.

6 files changed, 118 insertions, 62 deletions

diff --git a/jstests/concurrency/fsm_workloads/secondary_reads_with_catalog_changes.js b/jstests/concurrency/fsm_workloads/secondary_reads_with_catalog_changes.js
index 1705e185c56..c67b4793c01 100644
--- a/jstests/concurrency/fsm_workloads/secondary_reads_with_catalog_changes.js
+++ b/jstests/concurrency/fsm_workloads/secondary_reads_with_catalog_changes.js
@@ -25,7 +25,6 @@ load('jstests/concurrency/fsm_workloads/secondary_reads.js');  // for $config
  *   creates_background_indexes,
  *   requires_replication,
  *   uses_write_concern,
- *   incompatible_with_lockfreereads,
  * ]
  */
 var $config = extendWorkload($config, function($config, $super) {
diff --git a/src/mongo/db/catalog/collection_catalog.cpp b/src/mongo/db/catalog/collection_catalog.cpp
index 7f63f218eb4..7ca30a9d778 100644
--- a/src/mongo/db/catalog/collection_catalog.cpp
+++ b/src/mongo/db/catalog/collection_catalog.cpp
@@ -187,6 +187,43 @@ private:
 
 }  // namespace
 
+/**
+ * Publishes a pending writable Collection to the catalog. It needs to be registered with
+ * registerChangeForCatalogVisibility so other commit handlers can still write to this Collection.
+ * It also cleans up the collection from the UncommittedWritableCollections decoration.
+ */
+class CollectionCatalog::PublishWritableCollection final : public RecoveryUnit::Change {
+public:
+    PublishWritableCollection(OperationContext* opCtx,
+                              UncommittedWritableCollections& uncommittedWritableCollections,
+                              Collection* collection)
+        : _opCtx(opCtx),
+          _uncommittedWritableCollections(uncommittedWritableCollections),
+          _collection(collection) {}
+
+    void commit(boost::optional<Timestamp> commitTime) override {
+        auto [collection, commitHandlers] = _uncommittedWritableCollections.remove(_collection);
+        if (collection) {
+            CollectionCatalog::write(
+                _opCtx,
+                [collection = std::move(collection), &commitTime, commitHandlers = &commitHandlers](
+                    CollectionCatalog& catalog) {
+                    catalog._commitWritableClone(
+                        std::move(collection), commitTime, *commitHandlers);
+                });
+        }
+    }
+
+    void rollback() override {
+        _uncommittedWritableCollections.remove(_collection);
+    }
+
+private:
+    OperationContext* _opCtx;
+    UncommittedWritableCollections& _uncommittedWritableCollections;
+    Collection* _collection;
+};
+
 CollectionCatalog::iterator::iterator(OperationContext* opCtx,
                                       StringData dbName,
                                       const CollectionCatalog& catalog)
@@ -529,25 +566,9 @@ Collection* CollectionCatalog::lookupCollectionByUUIDForMetadataWrite(OperationC
     auto cloned = coll->clone();
     uncommittedWritableCollections.insert(cloned);
 
-    opCtx->recoveryUnit()->onCommit(
-        [opCtx, &uncommittedWritableCollections, clonedPtr = cloned.get()](
-            boost::optional<Timestamp> commitTime) {
-            auto [collection, commitHandlers] = uncommittedWritableCollections.remove(clonedPtr);
-            if (collection) {
-                CollectionCatalog::write(
-                    opCtx,
-                    [collection = std::move(collection),
-                     &commitTime,
-                     commitHandlers = &commitHandlers](CollectionCatalog& catalog) {
-                        catalog._commitWritableClone(
-                            std::move(collection), commitTime, *commitHandlers);
-                    });
-            }
-        });
-
-    opCtx->recoveryUnit()->onRollback([&uncommittedWritableCollections, cloned]() {
-        uncommittedWritableCollections.remove(cloned.get());
-    });
+    opCtx->recoveryUnit()->registerChangeForCatalogVisibility(
+        std::make_unique<PublishWritableCollection>(
+            opCtx, uncommittedWritableCollections, cloned.get()));
 
     return cloned.get();
 }
@@ -617,25 +638,9 @@ Collection* CollectionCatalog::lookupCollectionByNamespaceForMetadataWrite(
     auto cloned = coll->clone();
     uncommittedWritableCollections.insert(cloned);
 
-    opCtx->recoveryUnit()->onCommit(
-        [opCtx, &uncommittedWritableCollections, clonedPtr = cloned.get()](
-            boost::optional<Timestamp> commitTime) {
-            auto [collection, commitHandlers] = uncommittedWritableCollections.remove(clonedPtr);
-            if (collection) {
-                CollectionCatalog::write(
-                    opCtx,
-                    [collection = std::move(collection),
-                     &commitTime,
-                     commitHandlers = &commitHandlers](CollectionCatalog& catalog) {
-                        catalog._commitWritableClone(
-                            std::move(collection), commitTime, *commitHandlers);
-                    });
-            }
-        });
-
-    opCtx->recoveryUnit()->onRollback([&uncommittedWritableCollections, cloned]() {
-        uncommittedWritableCollections.remove(cloned.get());
-    });
+    opCtx->recoveryUnit()->registerChangeForCatalogVisibility(
+        std::make_unique<PublishWritableCollection>(
+            opCtx, uncommittedWritableCollections, cloned.get()));
 
     return cloned.get();
 }
diff --git a/src/mongo/db/catalog/collection_catalog.h b/src/mongo/db/catalog/collection_catalog.h
index ee1e7e12318..35d777263b4 100644
--- a/src/mongo/db/catalog/collection_catalog.h
+++ b/src/mongo/db/catalog/collection_catalog.h
@@ -355,6 +355,7 @@ public:
 
 private:
     friend class CollectionCatalog::iterator;
+    class PublishWritableCollection;
 
     std::shared_ptr<Collection> _lookupCollectionByUUID(CollectionUUID uuid) const;
 
diff --git a/src/mongo/db/storage/recovery_unit.cpp b/src/mongo/db/storage/recovery_unit.cpp
index bd2f0b70e52..9aeddf2d6ad 100644
--- a/src/mongo/db/storage/recovery_unit.cpp
+++ b/src/mongo/db/storage/recovery_unit.cpp
@@ -66,10 +66,15 @@ void RecoveryUnit::runPreCommitHooks(OperationContext* opCtx) {
 }
 
 void RecoveryUnit::registerChange(std::unique_ptr<Change> change) {
-    invariant(_inUnitOfWork(), toString(_getState()));
+    validateInUnitOfWork();
     _changes.push_back(std::move(change));
 }
 
+void RecoveryUnit::registerChangeForCatalogVisibility(std::unique_ptr<Change> change) {
+    validateInUnitOfWork();
+    _changesForCatalogVisibility.push_back(std::move(change));
+}
+
 void RecoveryUnit::commitRegisteredChanges(boost::optional<Timestamp> commitTimestamp) {
     // Getting to this method implies `runPreCommitHooks` completed successfully, resulting in
     // having its contents cleared.
@@ -77,6 +82,10 @@ void RecoveryUnit::commitRegisteredChanges(boost::optional<Timestamp> commitTime
     if (MONGO_unlikely(widenWUOWChangesWindow.shouldFail())) {
         sleepmillis(1000);
     }
+    _executeCommitHandlers(commitTimestamp);
+}
+
+void RecoveryUnit::_executeCommitHandlers(boost::optional<Timestamp> commitTimestamp) {
     for (auto& change : _changes) {
         try {
             // Log at higher level because commits occur far more frequently than rollbacks.
@@ -89,7 +98,20 @@ void RecoveryUnit::commitRegisteredChanges(boost::optional<Timestamp> commitTime
             std::terminate();
         }
     }
+    for (auto& change : _changesForCatalogVisibility) {
+        try {
+            // Log at higher level because commits occur far more frequently than rollbacks.
+            LOGV2_DEBUG(5255701,
+                        2,
+                        "CUSTOM COMMIT {demangleName_typeid_change}",
+                        "demangleName_typeid_change"_attr = redact(demangleName(typeid(*change))));
+            change->commit(commitTimestamp);
+        } catch (...) {
+            std::terminate();
+        }
+    }
     _changes.clear();
+    _changesForCatalogVisibility.clear();
 }
 
 void RecoveryUnit::abortRegisteredChanges() {
@@ -97,7 +119,21 @@ void RecoveryUnit::abortRegisteredChanges() {
     if (MONGO_unlikely(widenWUOWChangesWindow.shouldFail())) {
         sleepmillis(1000);
     }
+    _executeRollbackHandlers();
+}
+void RecoveryUnit::_executeRollbackHandlers() {
     try {
+        for (Changes::const_reverse_iterator it = _changesForCatalogVisibility.rbegin(),
+                                             end = _changesForCatalogVisibility.rend();
+             it != end;
+             ++it) {
+            Change* change = it->get();
+            LOGV2_DEBUG(5255702,
+                        2,
+                        "CUSTOM ROLLBACK {demangleName_typeid_change}",
+                        "demangleName_typeid_change"_attr = redact(demangleName(typeid(*change))));
+            change->rollback();
+        }
         for (Changes::const_reverse_iterator it = _changes.rbegin(), end = _changes.rend();
              it != end;
              ++it) {
@@ -108,9 +144,15 @@ void RecoveryUnit::abortRegisteredChanges() {
                         "demangleName_typeid_change"_attr = redact(demangleName(typeid(*change))));
             change->rollback();
         }
+        _changesForCatalogVisibility.clear();
         _changes.clear();
     } catch (...) {
         std::terminate();
     }
 }
+
+void RecoveryUnit::validateInUnitOfWork() const {
+    invariant(_inUnitOfWork(), toString(_getState()));
+}
+
 }  // namespace mongo
diff --git a/src/mongo/db/storage/recovery_unit.h b/src/mongo/db/storage/recovery_unit.h
index d35af5cf258..6f7c0752ed8 100644
--- a/src/mongo/db/storage/recovery_unit.h
+++ b/src/mongo/db/storage/recovery_unit.h
@@ -535,7 +535,19 @@ public:
      * The registerChange() method may only be called when a WriteUnitOfWork is active, and
      * may not be called during commit or rollback.
      */
-    virtual void registerChange(std::unique_ptr<Change> change);
+    void registerChange(std::unique_ptr<Change> change);
+
+    /**
+     * Like registerChange() above but should only be used to make new state visible in the
+     * in-memory catalog. Changes registered with this function will commit after the commit changes
+     * registered with registerChange and rollback will run before the rollback changes registered
+     * with registerChange.
+     *
+     * This separation ensures that regular Changes that can modify state are run before the Change
+     * to install the new state in the in-memory catalog, after which there should be no further
+     * changes.
+     */
+    void registerChangeForCatalogVisibility(std::unique_ptr<Change> change);
 
     /**
      * Registers a callback to be called if the current WriteUnitOfWork rolls back.
@@ -704,6 +716,16 @@ protected:
         return State::kCommitting == _state || State::kAborting == _state;
     }
 
+    /**
+     * Executes all registered commit handlers and clears all registered changes
+     */
+    void _executeCommitHandlers(boost::optional<Timestamp> commitTimestamp);
+
+    /**
+     * Executes all registered rollback handlers and clears all registered changes
+     */
+    void _executeRollbackHandlers();
+
     bool _mustBeTimestamped = false;
 
     bool _noEvictionAfterRollback = false;
@@ -716,10 +738,13 @@ private:
     virtual void doCommitUnitOfWork() = 0;
     virtual void doAbortUnitOfWork() = 0;
 
+    virtual void validateInUnitOfWork() const;
+
     std::vector<std::function<void(OperationContext*)>> _preCommitHooks;
 
     typedef std::vector<std::unique_ptr<Change>> Changes;
     Changes _changes;
+    Changes _changesForCatalogVisibility;
     State _state = State::kInactive;
     uint64_t _mySnapshotId;
 };
diff --git a/src/mongo/db/storage/recovery_unit_noop.h b/src/mongo/db/storage/recovery_unit_noop.h
index d5e3afeb161..4c06eeb62b2 100644
--- a/src/mongo/db/storage/recovery_unit_noop.h
+++ b/src/mongo/db/storage/recovery_unit_noop.h
@@ -42,15 +42,13 @@ class RecoveryUnitNoop : public RecoveryUnit {
 public:
     void beginUnitOfWork(OperationContext* opCtx) final {}
 
-    virtual bool waitUntilDurable(OperationContext* opCtx) {
+    bool waitUntilDurable(OperationContext* opCtx) override {
         return true;
     }
 
-    virtual void registerChange(std::unique_ptr<Change> change) {
-        _changes.push_back(std::move(change));
-    }
+    void setOrderedCommit(bool orderedCommit) override {}
 
-    virtual void setOrderedCommit(bool orderedCommit) {}
+    void validateInUnitOfWork() const override {}
 
     bool inActiveTxn() const {
         return false;
@@ -62,25 +60,11 @@ public:
 
 private:
     void doCommitUnitOfWork() final {
-        for (auto& change : _changes) {
-            try {
-                change->commit(boost::none);
-            } catch (...) {
-                std::terminate();
-            }
-        }
-        _changes.clear();
+        _executeCommitHandlers(boost::none);
     }
 
     void doAbortUnitOfWork() final {
-        for (auto it = _changes.rbegin(); it != _changes.rend(); ++it) {
-            try {
-                (*it)->rollback();
-            } catch (...) {
-                std::terminate();
-            }
-        }
-        _changes.clear();
+        _executeRollbackHandlers();
     }
 
     virtual void doAbandonSnapshot() {}