SERVER-29051 create data files during rollback via recover to timestamp

1 files changed, 696 insertions, 49 deletions

diff --git a/src/mongo/db/repl/rollback_impl_test.cpp b/src/mongo/db/repl/rollback_impl_test.cpp
index 7e3d0528f5a..2ad96e7be70 100644
--- a/src/mongo/db/repl/rollback_impl_test.cpp
+++ b/src/mongo/db/repl/rollback_impl_test.cpp
@@ -25,9 +25,16 @@
  *    exception statement from all source files in the program, then also delete
  *    it in the license file.
  */
+#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kReplicationRollback
 
 #include "mongo/platform/basic.h"
 
+#include <boost/optional.hpp>
+
+#include "mongo/db/catalog/collection_mock.h"
+#include "mongo/db/catalog/drop_collection.h"
+#include "mongo/db/catalog/uuid_catalog.h"
+#include "mongo/db/repl/drop_pending_collection_reaper.h"
 #include "mongo/db/repl/oplog_entry.h"
 #include "mongo/db/repl/oplog_interface_local.h"
 #include "mongo/db/repl/oplog_interface_mock.h"
@@ -37,8 +44,8 @@
 #include "mongo/db/s/type_shard_identity.h"
 #include "mongo/unittest/death_test.h"
 #include "mongo/util/assert_util.h"
+#include "mongo/util/log.h"
 #include "mongo/util/uuid.h"
-#include <boost/optional.hpp>
 
 namespace {
 
@@ -47,6 +54,87 @@ using namespace mongo::repl;
 
 NamespaceString nss("local.oplog.rs");
 
+BSONObj makeInsertOplogEntry(long long time, BSONObj obj, StringData ns, UUID uuid) {
+    return BSON("ts" << Timestamp(time, time) << "h" << time << "t" << time << "op"
+                     << "i"
+                     << "o"
+                     << obj
+                     << "ns"
+                     << ns
+                     << "ui"
+                     << uuid);
+}
+
+BSONObj makeUpdateOplogEntry(
+    long long time, BSONObj query, BSONObj update, StringData ns, UUID uuid) {
+    return BSON("ts" << Timestamp(time, time) << "h" << time << "t" << time << "op"
+                     << "u"
+                     << "ns"
+                     << ns
+                     << "ui"
+                     << uuid
+                     << "o2"
+                     << query
+                     << "o"
+                     << BSON("$set" << update));
+}
+
+BSONObj makeDeleteOplogEntry(long long time, BSONObj id, StringData ns, UUID uuid) {
+    return BSON("ts" << Timestamp(time, time) << "h" << time << "t" << time << "op"
+                     << "d"
+                     << "ns"
+                     << ns
+                     << "ui"
+                     << uuid
+                     << "o"
+                     << id);
+}
+
+class RollbackImplForTest final : public RollbackImpl {
+public:
+    using RollbackImpl::RollbackImpl;
+
+    static const std::vector<BSONObj> kEmptyVector;
+
+    /**
+     * Returns a reference to a vector containing all of the BSONObjs deleted from the namespace
+     * represented by 'uuid', or kEmptyVector if that namespace wasn't found in '_uuidToObjsMap'.
+     */
+    const std::vector<BSONObj>& docsDeletedForNamespace_forTest(UUID uuid) const& final {
+        auto iter = _uuidToObjsMap.find(uuid);
+        if (iter == _uuidToObjsMap.end()) {
+            return kEmptyVector;
+        }
+        return iter->second;
+    }
+
+protected:
+    /**
+     * Saves documents that would be deleted in '_uuidToObjsMap', rather than writing them out to a
+     * file.
+     */
+    void _writeRollbackFileForNamespace(OperationContext* opCtx,
+                                        UUID uuid,
+                                        NamespaceString nss,
+                                        const SimpleBSONObjUnorderedSet& idSet) final {
+        log() << "Simulating writing a rollback file for namespace " << nss.ns() << " with uuid "
+              << uuid;
+        for (auto&& id : idSet) {
+            log() << "Looking up " << id.jsonString();
+            auto document = _findDocumentById(opCtx, uuid, nss, id.firstElement());
+            if (document) {
+                _uuidToObjsMap[uuid].push_back(*document);
+            }
+        }
+        _listener->onRollbackFileWrittenForNamespace(std::move(uuid), std::move(nss));
+    }
+
+private:
+    stdx::unordered_map<UUID, std::vector<BSONObj>, UUID::Hash> _uuidToObjsMap;
+};
+
+const std::vector<BSONObj> RollbackImplForTest::kEmptyVector;
+
 /**
  * Unit test for rollback implementation introduced in 3.6.
  */
@@ -64,9 +152,76 @@ private:
     friend class RollbackImplTest::Listener;
 
 protected:
+    /**
+     * Creates a new mock collection with name 'nss' via the StorageInterface and associates 'uuid'
+     * with the new collection in the UUIDCatalog. There must not already exist a collection with
+     * name 'nss'.
+     */
+    std::unique_ptr<Collection> _initializeCollection(OperationContext* opCtx,
+                                                      UUID uuid,
+                                                      const NamespaceString& nss) {
+        // Create a new collection in the storage interface.
+        CollectionOptions options;
+        options.uuid = uuid;
+        ASSERT_OK(_storageInterface->createCollection(opCtx, nss, options));
+
+        // Initialize a mock collection.
+        std::unique_ptr<Collection> coll =
+            std::make_unique<Collection>(std::make_unique<CollectionMock>(nss));
+
+        // Register the UUID to that collection in the UUIDCatalog.
+        UUIDCatalog::get(opCtx).registerUUIDCatalogEntry(uuid, coll.get());
+        return coll;
+    }
+
+    /**
+     * Creates an oplog entry that represents the insertion of 'doc' into the namespace 'nss' with
+     * UUID 'uuid', and inserts the document into the storage interface.
+     *
+     * Unless 'time' is explicitly given, this insert is timestamped with an arbitrary time that
+     * monotonically increases with each successive call to this function.
+     */
+    void _insertDocAndGenerateOplogEntry(BSONObj doc,
+                                         UUID uuid,
+                                         NamespaceString nss,
+                                         boost::optional<long> time = boost::none) {
+        const auto optime = time.value_or(_counter++);
+        ASSERT_OK(_insertOplogEntry(makeInsertOplogEntry(optime, doc, nss.ns(), uuid)));
+        ASSERT_OK(_storageInterface->insertDocument(
+            _opCtx.get(), nss, {doc, Timestamp(optime, optime)}, optime));
+    }
+
+    /**
+     * Creates an oplog entry that represents updating an object matched by 'query' to be 'newDoc'
+     * in the namespace 'nss, with UUID 'uuid'. It also inserts 'newDoc' into the storage interface.
+     * This update is timestamped with an arbitrary time that monotonically increases with each
+     * successive call.
+     */
+    void _updateDocAndGenerateOplogEntry(BSONObj query,
+                                         BSONObj newDoc,
+                                         UUID uuid,
+                                         NamespaceString nss) {
+        const auto time = _counter++;
+        ASSERT_OK(_insertOplogEntry(makeUpdateOplogEntry(time, query, newDoc, nss.ns(), uuid)));
+        ASSERT_OK(_storageInterface->insertDocument(
+            _opCtx.get(), nss, {newDoc, Timestamp(time, time)}, time));
+    }
+
+    /**
+     * Creates an oplog entry that represents deleting an object with _id 'id' in the namespace
+     * 'nss' with UUID 'uuid'. This function also removes that document from the storage interface.
+     * This update is timestamped with an arbitrary time that monotonically increases with each
+     * successive call.
+     */
+    void _deleteDocAndGenerateOplogEntry(BSONElement id, UUID uuid, NamespaceString nss) {
+        const auto time = _counter++;
+        ASSERT_OK(_insertOplogEntry(makeDeleteOplogEntry(time, id.wrap(), nss.ns(), uuid)));
+        ASSERT_OK(_storageInterface->deleteById(_opCtx.get(), nss, id));
+    }
+
     std::unique_ptr<OplogInterfaceLocal> _localOplog;
     std::unique_ptr<OplogInterfaceMock> _remoteOplog;
-    std::unique_ptr<RollbackImpl> _rollback;
+    std::unique_ptr<RollbackImplForTest> _rollback;
 
     bool _transitionedToRollback = false;
     stdx::function<void()> _onTransitionToRollbackFn = [this]() { _transitionedToRollback = true; };
@@ -94,7 +249,13 @@ protected:
     stdx::function<void(const OpObserver::RollbackObserverInfo& rbInfo)> _onRollbackOpObserverFn =
         [this](const OpObserver::RollbackObserverInfo& rbInfo) {};
 
+    stdx::function<void(UUID, NamespaceString)> _onRollbackFileWrittenForNamespaceFn =
+        [this](UUID, NamespaceString) {};
+
     std::unique_ptr<Listener> _listener;
+
+private:
+    long _counter = 100;
 };
 
 void RollbackImplTest::setUp() {
@@ -104,12 +265,12 @@ void RollbackImplTest::setUp() {
                                                          NamespaceString::kRsOplogNamespace.ns());
     _remoteOplog = stdx::make_unique<OplogInterfaceMock>();
     _listener = stdx::make_unique<Listener>(this);
-    _rollback = stdx::make_unique<RollbackImpl>(_localOplog.get(),
-                                                _remoteOplog.get(),
-                                                _storageInterface,
-                                                _replicationProcess.get(),
-                                                _coordinator,
-                                                _listener.get());
+    _rollback = stdx::make_unique<RollbackImplForTest>(_localOplog.get(),
+                                                       _remoteOplog.get(),
+                                                       _storageInterface,
+                                                       _replicationProcess.get(),
+                                                       _coordinator,
+                                                       _listener.get());
 
     createOplog(_opCtx.get());
 }
@@ -134,6 +295,10 @@ public:
         _test->_onCommonPointFoundFn(commonPoint);
     }
 
+    void onRollbackFileWrittenForNamespace(UUID uuid, NamespaceString nss) noexcept final {
+        _test->_onRollbackFileWrittenForNamespaceFn(std::move(uuid), std::move(nss));
+    }
+
     void onRecoverToStableTimestamp(Timestamp stableTimestamp) noexcept override {
         _test->_onRecoverToStableTimestampFn(stableTimestamp);
     }
@@ -256,6 +421,8 @@ TEST_F(RollbackImplTest, RollbackPersistsDocumentAfterCommonPointToOplogTruncate
 
     auto nextTime = 3;
     ASSERT_OK(_insertOplogEntry(makeOp(nextTime)));
+    const auto nss = NamespaceString("test.coll");
+    auto coll = _initializeCollection(_opCtx.get(), UUID::gen(), nss);
 
     ASSERT_OK(_rollback->runRollback(_opCtx.get()));
     ASSERT_EQUALS(_truncatePoint, Timestamp(3, 3));
@@ -587,6 +754,366 @@ TEST_F(RollbackImplTest, RollbackSkipsTriggerOpObserverWhenShutDownEarly) {
     ASSERT_EQUALS(_coordinator->getMemberState(), MemberState::RS_SECONDARY);
 }
 
+TEST_F(RollbackImplTest, RollbackDoesNotWriteRollbackFilesIfNoInsertsOrUpdatesAfterCommonPoint) {
+    const auto commonOp = makeOpAndRecordId(1);
+    _remoteOplog->setOperations({commonOp});
+    ASSERT_OK(_insertOplogEntry(commonOp.first));
+    _storageInterface->setStableTimestamp(nullptr, Timestamp(1, 1));
+
+    const auto uuid = UUID::gen();
+    const auto nss = NamespaceString("db.coll");
+    const auto coll = _initializeCollection(_opCtx.get(), uuid, nss);
+    const auto oplogEntry = BSON("ts" << Timestamp(3, 3) << "h" << 3L << "t" << 3L << "op"
+                                      << "c"
+                                      << "o"
+                                      << BSON("create" << nss.coll())
+                                      << "ns"
+                                      << nss.ns()
+                                      << "ui"
+                                      << uuid);
+    ASSERT_OK(_insertOplogEntry(oplogEntry));
+
+    ASSERT_OK(_rollback->runRollback(_opCtx.get()));
+    ASSERT(_rollback->docsDeletedForNamespace_forTest(uuid).empty());
+}
+
+TEST_F(RollbackImplTest, RollbackSavesInsertedDocumentToFile) {
+    const auto commonOp = makeOpAndRecordId(1);
+    _remoteOplog->setOperations({commonOp});
+    ASSERT_OK(_insertOplogEntry(commonOp.first));
+    _storageInterface->setStableTimestamp(nullptr, Timestamp(1, 1));
+
+    const auto nss = NamespaceString("db.people");
+    const auto uuid = UUID::gen();
+    const auto coll = _initializeCollection(_opCtx.get(), uuid, nss);
+
+    const auto obj = BSON("_id" << 0 << "name"
+                                << "kyle");
+    _insertDocAndGenerateOplogEntry(obj, uuid, nss);
+
+    ASSERT_OK(_rollback->runRollback(_opCtx.get()));
+
+    const auto& deletedObjs = _rollback->docsDeletedForNamespace_forTest(uuid);
+    ASSERT_EQ(deletedObjs.size(), 1UL);
+    ASSERT_BSONOBJ_EQ(deletedObjs.front(), obj);
+}
+
+TEST_F(RollbackImplTest, RollbackSavesLatestVersionOfDocumentWhenThereAreMultipleInserts) {
+    const auto commonOp = makeOpAndRecordId(1);
+    _remoteOplog->setOperations({commonOp});
+    ASSERT_OK(_insertOplogEntry(commonOp.first));
+    _storageInterface->setStableTimestamp(nullptr, Timestamp(1, 1));
+
+    const auto nss = NamespaceString("db.people");
+    const auto uuid = UUID::gen();
+    const auto coll = _initializeCollection(_opCtx.get(), uuid, nss);
+
+    const auto oldObj = BSON("_id" << 0 << "name"
+                                   << "kyle");
+    const auto newObj = BSON("_id" << 0 << "name"
+                                   << "jungsoo");
+    _insertDocAndGenerateOplogEntry(oldObj, uuid, nss);
+    _deleteDocAndGenerateOplogEntry(oldObj["_id"], uuid, nss);
+    _insertDocAndGenerateOplogEntry(newObj, uuid, nss);
+
+    ASSERT_OK(_rollback->runRollback(_opCtx.get()));
+
+    const auto& deletedObjs = _rollback->docsDeletedForNamespace_forTest(uuid);
+    ASSERT_EQ(deletedObjs.size(), 1UL);
+    ASSERT_BSONOBJ_EQ(deletedObjs.front(), newObj);
+}
+
+TEST_F(RollbackImplTest, RollbackSavesUpdatedDocumentToFile) {
+    const auto commonOp = makeOpAndRecordId(1);
+    _remoteOplog->setOperations({commonOp});
+    ASSERT_OK(_insertOplogEntry(commonOp.first));
+    _storageInterface->setStableTimestamp(nullptr, Timestamp(1, 1));
+
+    const auto nss = NamespaceString("db.people");
+    const auto uuid = UUID::gen();
+    const auto coll = _initializeCollection(_opCtx.get(), uuid, nss);
+
+    const auto query = BSON("_id" << 0);
+    const auto obj = BSON("_id" << 0 << "name"
+                                << "kyle");
+    _updateDocAndGenerateOplogEntry(query, obj, uuid, nss);
+
+    ASSERT_OK(_rollback->runRollback(_opCtx.get()));
+
+    const auto& deletedObjs = _rollback->docsDeletedForNamespace_forTest(uuid);
+    ASSERT_EQ(deletedObjs.size(), 1UL);
+    ASSERT_BSONOBJ_EQ(deletedObjs.front(), obj);
+}
+
+TEST_F(RollbackImplTest, RollbackSavesLatestVersionOfDocumentWhenThereAreMultipleUpdates) {
+    const auto commonOp = makeOpAndRecordId(1);
+    _remoteOplog->setOperations({commonOp});
+    ASSERT_OK(_insertOplogEntry(commonOp.first));
+    _storageInterface->setStableTimestamp(nullptr, Timestamp(1, 1));
+
+    const auto nss = NamespaceString("db.people");
+    const auto uuid = UUID::gen();
+    const auto coll = _initializeCollection(_opCtx.get(), uuid, nss);
+
+    const auto query = BSON("_id" << 3.14);
+    const auto oldObj = BSON("_id" << 3.14 << "name"
+                                   << "kyle");
+    const auto newObj = BSON("_id" << 3.14 << "name"
+                                   << "jungsoo");
+    _updateDocAndGenerateOplogEntry(query, oldObj, uuid, nss);
+    _deleteDocAndGenerateOplogEntry(oldObj["_id"], uuid, nss);
+    _updateDocAndGenerateOplogEntry(query, newObj, uuid, nss);
+
+    ASSERT_OK(_rollback->runRollback(_opCtx.get()));
+
+    const auto& deletedObjs = _rollback->docsDeletedForNamespace_forTest(uuid);
+    ASSERT_EQ(deletedObjs.size(), 1UL);
+    ASSERT_BSONOBJ_EQ(deletedObjs.front(), newObj);
+}
+
+TEST_F(RollbackImplTest, RollbackDoesNotWriteDocumentToFileIfInsertIsRevertedByDelete) {
+    const auto commonOp = makeOpAndRecordId(1);
+    _remoteOplog->setOperations({commonOp});
+    ASSERT_OK(_insertOplogEntry(commonOp.first));
+    _storageInterface->setStableTimestamp(nullptr, Timestamp(1, 1));
+
+    const auto nss = NamespaceString("db.numbers");
+    const auto uuid = UUID::gen();
+    const auto coll = _initializeCollection(_opCtx.get(), uuid, nss);
+
+    const auto objToKeep = BSON("_id" << 6);
+    const auto objToDelete = BSON("_id" << 7);
+    _insertDocAndGenerateOplogEntry(objToDelete, uuid, nss);
+    _insertDocAndGenerateOplogEntry(objToKeep, uuid, nss);
+    _deleteDocAndGenerateOplogEntry(objToDelete["_id"], uuid, nss);
+
+    ASSERT_OK(_rollback->runRollback(_opCtx.get()));
+
+    const auto& deletedObjs = _rollback->docsDeletedForNamespace_forTest(uuid);
+    ASSERT_EQ(deletedObjs.size(), 1UL);
+    ASSERT_BSONOBJ_EQ(deletedObjs.front(), objToKeep);
+}
+
+TEST_F(RollbackImplTest, RollbackDoesNotWriteDocumentToFileIfUpdateIsFollowedByDelete) {
+    const auto commonOp = makeOpAndRecordId(1);
+    _remoteOplog->setOperations({commonOp});
+    ASSERT_OK(_insertOplogEntry(commonOp.first));
+    _storageInterface->setStableTimestamp(nullptr, Timestamp(1, 1));
+
+    const auto nss = NamespaceString("db.numbers");
+    const auto uuid = UUID::gen();
+    const auto coll = _initializeCollection(_opCtx.get(), uuid, nss);
+
+    const auto objToKeep = BSON("_id" << 6);
+    const auto objToDelete = BSON("_id" << 7);
+    _updateDocAndGenerateOplogEntry(objToDelete, objToDelete, uuid, nss);
+    _updateDocAndGenerateOplogEntry(objToKeep, objToKeep, uuid, nss);
+    _deleteDocAndGenerateOplogEntry(objToDelete["_id"], uuid, nss);
+
+    ASSERT_OK(_rollback->runRollback(_opCtx.get()));
+
+    const auto& deletedObjs = _rollback->docsDeletedForNamespace_forTest(uuid);
+    ASSERT_EQ(deletedObjs.size(), 1UL);
+    ASSERT_BSONOBJ_EQ(deletedObjs.front(), objToKeep);
+}
+
+TEST_F(RollbackImplTest, RollbackProperlySavesFilesWhenCollectionIsRenamed) {
+    const auto commonOp = makeOpAndRecordId(1);
+    _remoteOplog->setOperations({commonOp});
+    ASSERT_OK(_insertOplogEntry(commonOp.first));
+    _storageInterface->setStableTimestamp(nullptr, Timestamp(1, 1));
+
+    const auto nssBeforeRename = NamespaceString("db.firstColl");
+    const auto uuidBeforeRename = UUID::gen();
+    const auto collBeforeRename =
+        _initializeCollection(_opCtx.get(), uuidBeforeRename, nssBeforeRename);
+
+    // Insert a document into the collection.
+    const auto objInRenamedCollection = BSON("_id"
+                                             << "kyle");
+    _insertDocAndGenerateOplogEntry(objInRenamedCollection, uuidBeforeRename, nssBeforeRename, 2);
+
+    // Rename the original collection.
+    const auto nssAfterRename = NamespaceString("db.secondColl");
+    auto renameCmdObj =
+        BSON("renameCollection" << nssBeforeRename.ns() << "to" << nssAfterRename.ns());
+    auto renameCmdOp =
+        makeCommandOp(Timestamp(3, 3), uuidBeforeRename, nssBeforeRename.ns(), renameCmdObj, 3);
+    ASSERT_OK(_insertOplogEntry(renameCmdOp.first));
+    ASSERT_OK(
+        _storageInterface->renameCollection(_opCtx.get(), nssBeforeRename, nssAfterRename, true));
+
+    // Create a new collection with the old name.
+    const auto uuidAfterRename = UUID::gen();
+    const auto collAfterRename =
+        _initializeCollection(_opCtx.get(), uuidAfterRename, nssBeforeRename);
+
+    // Insert a different document into the new collection.
+    const auto objInNewCollection = BSON("_id"
+                                         << "jungsoo");
+    _insertDocAndGenerateOplogEntry(objInNewCollection, uuidAfterRename, nssBeforeRename, 4);
+
+    ASSERT_OK(_rollback->runRollback(_opCtx.get()));
+
+    const auto& deletedObjsRenamedColl =
+        _rollback->docsDeletedForNamespace_forTest(uuidBeforeRename);
+    ASSERT_EQ(deletedObjsRenamedColl.size(), 1UL);
+    ASSERT_BSONOBJ_EQ(deletedObjsRenamedColl.front(), objInRenamedCollection);
+
+    const auto& deletedObjsNewColl = _rollback->docsDeletedForNamespace_forTest(uuidAfterRename);
+    ASSERT_EQ(deletedObjsNewColl.size(), 1UL);
+    ASSERT_BSONOBJ_EQ(deletedObjsNewColl.front(), objInNewCollection);
+}
+
+TEST_F(RollbackImplTest, RollbackProperlySavesFilesWhenInsertsAndDropOfCollectionAreRolledBack) {
+    const auto commonOp = makeOpAndRecordId(1);
+    _remoteOplog->setOperations({commonOp});
+    ASSERT_OK(_insertOplogEntry(commonOp.first));
+    _storageInterface->setStableTimestamp(nullptr, Timestamp(1, 1));
+
+    // Create the collection, but as a drop-pending collection.
+    const auto dropOpTime = OpTime(Timestamp(200, 200), 200L);
+    const auto nss = NamespaceString("db.people").makeDropPendingNamespace(dropOpTime);
+    const auto uuid = UUID::gen();
+    const auto coll = _initializeCollection(_opCtx.get(), uuid, nss);
+    DropPendingCollectionReaper::get(_opCtx.get())->addDropPendingNamespace(dropOpTime, nss);
+
+    // Insert documents into the collection. We'll write them out even though the collection is
+    // later dropped.
+    const auto obj1 = BSON("_id"
+                           << "kyle");
+    const auto obj2 = BSON("_id"
+                           << "glenn");
+    _insertDocAndGenerateOplogEntry(obj1, uuid, nss);
+    _insertDocAndGenerateOplogEntry(obj2, uuid, nss);
+
+    // Create an oplog entry for the collection drop.
+    const auto oplogEntry =
+        BSON("ts" << dropOpTime.getTimestamp() << "h" << 200L << "t" << dropOpTime.getTerm() << "op"
+                  << "c"
+                  << "o"
+                  << BSON("drop" << nss.coll())
+                  << "ns"
+                  << nss.ns()
+                  << "ui"
+                  << uuid);
+    ASSERT_OK(_insertOplogEntry(oplogEntry));
+
+    ASSERT_OK(_rollback->runRollback(_opCtx.get()));
+
+    const auto& deletedObjs = _rollback->docsDeletedForNamespace_forTest(uuid);
+    ASSERT_EQ(deletedObjs.size(), 2UL);
+    std::vector<BSONObj> expectedObjs({obj1, obj2});
+    ASSERT(std::is_permutation(deletedObjs.begin(),
+                               deletedObjs.end(),
+                               expectedObjs.begin(),
+                               expectedObjs.end(),
+                               SimpleBSONObjComparator::kInstance.makeEqualTo()));
+}
+
+TEST_F(RollbackImplTest, RollbackProperlySavesFilesWhenCreateCollAndInsertsAreRolledBack) {
+    const auto commonOp = makeOpAndRecordId(1);
+    _remoteOplog->setOperations({commonOp});
+    ASSERT_OK(_insertOplogEntry(commonOp.first));
+    _storageInterface->setStableTimestamp(nullptr, Timestamp(1, 1));
+
+    // Create the collection and make an oplog entry for the creation event.
+    const auto nss = NamespaceString("db.people");
+    const auto uuid = UUID::gen();
+    const auto coll = _initializeCollection(_opCtx.get(), uuid, nss);
+    const auto oplogEntry = BSON("ts" << Timestamp(3, 3) << "h" << 3L << "t" << 3L << "op"
+                                      << "c"
+                                      << "o"
+                                      << BSON("create" << nss.coll())
+                                      << "ns"
+                                      << nss.ns()
+                                      << "ui"
+                                      << uuid);
+    ASSERT_OK(_insertOplogEntry(oplogEntry));
+
+    // Insert documents into the collection.
+    const std::vector<BSONObj> objs({BSON("_id"
+                                          << "kyle"),
+                                     BSON("_id"
+                                          << "jungsoo"),
+                                     BSON("_id"
+                                          << "erjon")});
+    for (auto&& obj : objs) {
+        _insertDocAndGenerateOplogEntry(obj, uuid, nss);
+    }
+
+    ASSERT_OK(_rollback->runRollback(_opCtx.get()));
+
+    const auto& deletedObjs = _rollback->docsDeletedForNamespace_forTest(uuid);
+    ASSERT_EQ(deletedObjs.size(), objs.size());
+    ASSERT(std::is_permutation(deletedObjs.begin(),
+                               deletedObjs.end(),
+                               objs.begin(),
+                               objs.end(),
+                               SimpleBSONObjComparator::kInstance.makeEqualTo()));
+}
+
+TEST_F(RollbackImplTest, RollbackStopsWritingRollbackFilesWhenShutdownIsInProgress) {
+    const auto commonOp = makeOpAndRecordId(1);
+    _remoteOplog->setOperations({commonOp});
+    ASSERT_OK(_insertOplogEntry(commonOp.first));
+    _storageInterface->setStableTimestamp(nullptr, Timestamp(1, 1));
+
+    const auto nss1 = NamespaceString("db.people");
+    const auto uuid1 = UUID::gen();
+    const auto coll1 = _initializeCollection(_opCtx.get(), uuid1, nss1);
+    const auto obj1 = BSON("_id" << 0 << "name"
+                                 << "kyle");
+    _insertDocAndGenerateOplogEntry(obj1, uuid1, nss1);
+
+    const auto nss2 = NamespaceString("db.persons");
+    const auto uuid2 = UUID::gen();
+    const auto coll2 = _initializeCollection(_opCtx.get(), uuid2, nss2);
+    const auto obj2 = BSON("_id" << 0 << "name"
+                                 << "jungsoo");
+    _insertDocAndGenerateOplogEntry(obj2, uuid2, nss2);
+
+    // Register a listener that sends rollback into shutdown.
+    std::vector<UUID> collsWithSuccessfullyWrittenDataFiles;
+    _onRollbackFileWrittenForNamespaceFn =
+        [this, &collsWithSuccessfullyWrittenDataFiles](UUID uuid, NamespaceString nss) {
+            collsWithSuccessfullyWrittenDataFiles.emplace_back(std::move(uuid));
+            _rollback->shutdown();
+        };
+
+    ASSERT_EQ(_rollback->runRollback(_opCtx.get()), ErrorCodes::ShutdownInProgress);
+
+    ASSERT_EQ(collsWithSuccessfullyWrittenDataFiles.size(), 1UL);
+    const auto& uuid = collsWithSuccessfullyWrittenDataFiles.front();
+    ASSERT(uuid == uuid1 || uuid == uuid2) << "wrote out a data file for unknown uuid " << uuid
+                                           << "; expected it to be either " << uuid1 << " or "
+                                           << uuid2;
+}
+
+DEATH_TEST_F(RollbackImplTest,
+             InvariantFailureIfNamespaceIsMissingWhenWritingRollbackFiles,
+             "unexpectedly missing in the UUIDCatalog") {
+    const auto commonOp = makeOpAndRecordId(1);
+    _remoteOplog->setOperations({commonOp});
+    ASSERT_OK(_insertOplogEntry(commonOp.first));
+    _storageInterface->setStableTimestamp(nullptr, Timestamp(1, 1));
+
+    const auto nss = NamespaceString("db.people");
+    const auto uuid = UUID::gen();
+    const auto coll = _initializeCollection(_opCtx.get(), uuid, nss);
+
+    const auto obj = BSON("_id" << 0 << "name"
+                                << "kyle");
+    _insertDocAndGenerateOplogEntry(obj, uuid, nss);
+
+    // Drop the collection (immediately; not a two-phase drop), so that the namespace can no longer
+    // be found.
+    ASSERT_OK(_storageInterface->dropCollection(_opCtx.get(), nss));
+
+    auto status = _rollback->runRollback(_opCtx.get());
+    unittest::log() << "mongod did not crash when expected; status: " << status;
+}
+
 /**
  * Fixture to help test that rollback records the correct information in its RollbackObserverInfo
  * struct.
@@ -614,13 +1141,12 @@ public:
         return _rollback->runRollback(_opCtx.get());
     }
 
-    BSONObj makeSessionOp(NamespaceString nss, UUID sessionId, TxnNumber txnNum) {
-        auto uuid = UUID::gen();
+    BSONObj makeSessionOp(UUID collId, NamespaceString nss, UUID sessionId, TxnNumber txnNum) {
         BSONObjBuilder bob;
         bob.append("ts", Timestamp(2, 1));
         bob.append("h", 1LL);
         bob.append("op", "i");
-        uuid.appendToBuilder(&bob, "ui");
+        collId.appendToBuilder(&bob, "ui");
         bob.append("ns", nss.ns());
         bob.append("o", BSON("_id" << 1));
         bob.append("lsid",
@@ -630,6 +1156,20 @@ public:
         return bob.obj();
     }
 
+    void assertRollbackInfoContainsObjectForUUID(UUID uuid, BSONObj bson) {
+        const auto& uuidToIdMap = _rbInfo.rollbackDeletedIdsMap;
+        auto search = uuidToIdMap.find(uuid);
+        ASSERT(search != uuidToIdMap.end()) << "map is unexpectedly missing an entry for uuid "
+                                            << uuid.toString() << " containing object "
+                                            << bson.jsonString();
+        const auto& idObjSet = search->second;
+        const auto iter = idObjSet.find(bson);
+        ASSERT(iter != idObjSet.end()) << "_id object set is unexpectedly missing object "
+                                       << bson.jsonString() << " in namespace with uuid "
+                                       << uuid.toString();
+    }
+
+
 protected:
     OpObserver::RollbackObserverInfo _rbInfo;
 };
@@ -869,37 +1409,37 @@ TEST_F(RollbackImplObserverInfoTest, RollbackFailsOnMalformedApplyOpsOplogEntry)
 }
 
 TEST_F(RollbackImplObserverInfoTest, RollbackRecordsNamespaceOfSingleOplogEntry) {
-    auto nss = NamespaceString("test", "coll");
-    auto insertOp = makeCRUDOp(OpTypeEnum::kInsert,
-                               Timestamp(2, 2),
-                               UUID::gen(),
-                               nss.ns(),
-                               BSON("_id" << 1),
-                               boost::none,
-                               2);
+    const auto nss = NamespaceString("test", "coll");
+    const auto uuid = UUID::gen();
+    const auto coll = _initializeCollection(_opCtx.get(), uuid, nss);
+    const auto insertOp = makeCRUDOp(
+        OpTypeEnum::kInsert, Timestamp(2, 2), uuid, nss.ns(), BSON("_id" << 1), boost::none, 2);
     ASSERT_OK(rollbackOps({insertOp}));
     std::set<NamespaceString> expectedNamespaces = {nss};
     ASSERT(expectedNamespaces == _rbInfo.rollbackNamespaces);
 }
 
 TEST_F(RollbackImplObserverInfoTest, RollbackRecordsMultipleNamespacesOfOplogEntries) {
-    auto makeInsertOp = [&](NamespaceString nss, Timestamp ts, int recordId) {
-        return makeCRUDOp(OpTypeEnum::kInsert,
-                          ts,
-                          UUID::gen(),
-                          nss.ns(),
-                          BSON("_id" << 1),
-                          boost::none,
-                          recordId);
+    const auto makeInsertOp = [&](UUID uuid, NamespaceString nss, Timestamp ts, int recordId) {
+        return makeCRUDOp(
+            OpTypeEnum::kInsert, ts, uuid, nss.ns(), BSON("_id" << 1), boost::none, recordId);
     };
 
-    auto nss1 = NamespaceString("test", "coll1");
-    auto nss2 = NamespaceString("test", "coll2");
-    auto nss3 = NamespaceString("test", "coll3");
+    const auto nss1 = NamespaceString("test", "coll1");
+    const auto nss2 = NamespaceString("test", "coll2");
+    const auto nss3 = NamespaceString("test", "coll3");
 
-    auto insertOp1 = makeInsertOp(nss1, Timestamp(2, 1), 2);
-    auto insertOp2 = makeInsertOp(nss2, Timestamp(3, 1), 3);
-    auto insertOp3 = makeInsertOp(nss3, Timestamp(4, 1), 4);
+    const auto uuid1 = UUID::gen();
+    const auto uuid2 = UUID::gen();
+    const auto uuid3 = UUID::gen();
+
+    const auto coll1 = _initializeCollection(_opCtx.get(), uuid1, nss1);
+    const auto coll2 = _initializeCollection(_opCtx.get(), uuid2, nss2);
+    const auto coll3 = _initializeCollection(_opCtx.get(), uuid3, nss3);
+
+    auto insertOp1 = makeInsertOp(uuid1, nss1, Timestamp(2, 1), 2);
+    auto insertOp2 = makeInsertOp(uuid2, nss2, Timestamp(3, 1), 3);
+    auto insertOp3 = makeInsertOp(uuid3, nss3, Timestamp(4, 1), 4);
 
     ASSERT_OK(rollbackOps({insertOp3, insertOp2, insertOp1}));
     std::set<NamespaceString> expectedNamespaces = {nss1, nss2, nss3};
@@ -924,10 +1464,11 @@ DEATH_TEST_F(RollbackImplObserverInfoTest,
 }
 
 TEST_F(RollbackImplObserverInfoTest, RollbackRecordsSessionIdFromOplogEntry) {
-
     NamespaceString nss("test.coll");
+    const auto collId = UUID::gen();
+    const auto coll = _initializeCollection(_opCtx.get(), collId, nss);
     auto sessionId = UUID::gen();
-    auto sessionOpObj = makeSessionOp(nss, sessionId, 1L);
+    auto sessionOpObj = makeSessionOp(collId, nss, sessionId, 1L);
     auto sessionOp = std::make_pair(sessionOpObj, RecordId(recordId));
 
     // Run the rollback and make sure the correct session id was recorded.
@@ -938,23 +1479,21 @@ TEST_F(RollbackImplObserverInfoTest, RollbackRecordsSessionIdFromOplogEntry) {
 
 TEST_F(RollbackImplObserverInfoTest,
        RollbackDoesntRecordSessionIdFromOplogEntryWithoutSessionInfo) {
-    auto nss = NamespaceString("test", "coll");
-    auto insertOp = makeCRUDOp(OpTypeEnum::kInsert,
-                               Timestamp(2, 2),
-                               UUID::gen(),
-                               nss.ns(),
-                               BSON("_id" << 1),
-                               boost::none,
-                               2);
+    const auto nss = NamespaceString("test", "coll");
+    const auto uuid = UUID::gen();
+    const auto coll = _initializeCollection(_opCtx.get(), uuid, nss);
+    auto insertOp = makeCRUDOp(
+        OpTypeEnum::kInsert, Timestamp(2, 2), uuid, nss.ns(), BSON("_id" << 1), boost::none, 2);
     ASSERT_OK(rollbackOps({insertOp}));
     ASSERT(_rbInfo.rollbackSessionIds.empty());
 }
 
 TEST_F(RollbackImplObserverInfoTest, RollbackRecordsSessionIdFromApplyOpsSubOp) {
-
     NamespaceString nss("test.coll");
+    const auto collId = UUID::gen();
+    const auto coll = _initializeCollection(_opCtx.get(), collId, nss);
     auto sessionId = UUID::gen();
-    auto sessionOpObj = makeSessionOp(nss, sessionId, 1L);
+    auto sessionOpObj = makeSessionOp(collId, nss, sessionId, 1L);
 
     // Create the applyOps command object.
     BSONArrayBuilder subops;
@@ -970,10 +1509,12 @@ TEST_F(RollbackImplObserverInfoTest, RollbackRecordsSessionIdFromApplyOpsSubOp)
 
 TEST_F(RollbackImplObserverInfoTest, RollbackRecordsShardIdentityRollback) {
     serverGlobalParams.clusterRole = ClusterRole::ShardServer;
-    auto nss = NamespaceString::kServerConfigurationNamespace;
+    const auto uuid = UUID::gen();
+    const auto nss = NamespaceString::kServerConfigurationNamespace;
+    const auto coll = _initializeCollection(_opCtx.get(), uuid, nss);
     auto insertShardIdOp = makeCRUDOp(OpTypeEnum::kInsert,
                                       Timestamp(2, 2),
-                                      UUID::gen(),
+                                      uuid,
                                       nss.ns(),
                                       BSON("_id" << ShardIdentityType::IdName),
                                       boost::none,
@@ -986,10 +1527,12 @@ TEST_F(RollbackImplObserverInfoTest, RollbackRecordsShardIdentityRollback) {
 
 TEST_F(RollbackImplObserverInfoTest, RollbackDoesntRecordShardIdentityRollbackForNormalDocument) {
     serverGlobalParams.clusterRole = ClusterRole::ShardServer;
-    auto nss = NamespaceString::kServerConfigurationNamespace;
+    const auto nss = NamespaceString::kServerConfigurationNamespace;
+    const auto uuid = UUID::gen();
+    const auto coll = _initializeCollection(_opCtx.get(), uuid, nss);
     auto deleteOp = makeCRUDOp(OpTypeEnum::kDelete,
                                Timestamp(2, 2),
-                               UUID::gen(),
+                               uuid,
                                nss.ns(),
                                BSON("_id"
                                     << "not_the_shard_id_document"),
@@ -998,4 +1541,108 @@ TEST_F(RollbackImplObserverInfoTest, RollbackDoesntRecordShardIdentityRollbackFo
     ASSERT_OK(rollbackOps({deleteOp}));
     ASSERT_FALSE(_rbInfo.shardIdentityRolledBack);
 }
+
+TEST_F(RollbackImplObserverInfoTest, RollbackRecordsInsertOpsInUUIDToIdMap) {
+    const auto nss1 = NamespaceString("db.people");
+    const auto uuid1 = UUID::gen();
+    const auto coll1 = _initializeCollection(_opCtx.get(), uuid1, nss1);
+    const auto obj1 = BSON("_id"
+                           << "kyle");
+    const auto insertOp1 =
+        makeCRUDOp(OpTypeEnum::kInsert, Timestamp(2, 2), uuid1, nss.ns(), obj1, boost::none, 2);
+
+    const auto nss2 = NamespaceString("db.persons");
+    const auto uuid2 = UUID::gen();
+    const auto coll2 = _initializeCollection(_opCtx.get(), uuid2, nss2);
+    const auto obj2 = BSON("_id"
+                           << "jungsoo");
+    const auto insertOp2 =
+        makeCRUDOp(OpTypeEnum::kInsert, Timestamp(3, 3), uuid2, nss2.ns(), obj2, boost::none, 3);
+
+    ASSERT_OK(rollbackOps({insertOp2, insertOp1}));
+    ASSERT_EQ(_rbInfo.rollbackDeletedIdsMap.size(), 2UL);
+
+    assertRollbackInfoContainsObjectForUUID(uuid1, obj1);
+    assertRollbackInfoContainsObjectForUUID(uuid2, obj2);
+}
+
+TEST_F(RollbackImplObserverInfoTest, RollbackRecordsUpdateOpsInUUIDToIdMap) {
+    const auto nss1 = NamespaceString("db.coll1");
+    const auto uuid1 = UUID::gen();
+    const auto coll1 = _initializeCollection(_opCtx.get(), uuid1, nss1);
+    const auto id1 = BSON("_id" << 1);
+    const auto updateOp1 = makeCRUDOp(OpTypeEnum::kUpdate,
+                                      Timestamp(2, 2),
+                                      uuid1,
+                                      nss1.ns(),
+                                      BSON("$set" << BSON("foo" << 1)),
+                                      id1,
+                                      2);
+
+    const auto nss2 = NamespaceString("db.coll2");
+    const auto uuid2 = UUID::gen();
+    const auto coll2 = _initializeCollection(_opCtx.get(), uuid2, nss2);
+    const auto id2 = BSON("_id" << 2);
+    const auto updateOp2 = makeCRUDOp(OpTypeEnum::kUpdate,
+                                      Timestamp(3, 3),
+                                      uuid2,
+                                      nss2.ns(),
+                                      BSON("$set" << BSON("foo" << 1)),
+                                      id2,
+                                      3);
+
+    ASSERT_OK(rollbackOps({updateOp2, updateOp1}));
+    ASSERT_EQ(_rbInfo.rollbackDeletedIdsMap.size(), 2UL);
+
+    assertRollbackInfoContainsObjectForUUID(uuid1, id1);
+    assertRollbackInfoContainsObjectForUUID(uuid2, id2);
+}
+
+TEST_F(RollbackImplObserverInfoTest, RollbackRecordsMultipleInsertOpsForSameNamespace) {
+    const auto nss = NamespaceString("db.coll");
+    const auto uuid = UUID::gen();
+    const auto coll = _initializeCollection(_opCtx.get(), uuid, nss);
+    const auto obj1 = BSON("_id" << 1);
+    const auto insertOp1 =
+        makeCRUDOp(OpTypeEnum::kInsert, Timestamp(2, 2), uuid, nss.ns(), obj1, boost::none, 2);
+
+    const auto obj2 = BSON("_id" << 2);
+    const auto insertOp2 =
+        makeCRUDOp(OpTypeEnum::kInsert, Timestamp(3, 3), uuid, nss.ns(), obj2, boost::none, 3);
+
+    ASSERT_OK(rollbackOps({insertOp2, insertOp1}));
+    ASSERT_EQ(_rbInfo.rollbackDeletedIdsMap.size(), 1UL);
+
+    assertRollbackInfoContainsObjectForUUID(uuid, obj1);
+    assertRollbackInfoContainsObjectForUUID(uuid, obj2);
+}
+
+TEST_F(RollbackImplObserverInfoTest, RollbackRecordsMultipleUpdateOpsForSameNamespace) {
+    const auto nss = NamespaceString("db.coll");
+    const auto uuid = UUID::gen();
+    const auto coll = _initializeCollection(_opCtx.get(), uuid, nss);
+    const auto obj1 = BSON("_id" << 1);
+    const auto updateOp1 = makeCRUDOp(OpTypeEnum::kUpdate,
+                                      Timestamp(2, 2),
+                                      uuid,
+                                      nss.ns(),
+                                      BSON("$set" << BSON("foo" << 1)),
+                                      obj1,
+                                      2);
+
+    const auto obj2 = BSON("_id" << 2);
+    const auto updateOp2 = makeCRUDOp(OpTypeEnum::kUpdate,
+                                      Timestamp(3, 3),
+                                      uuid,
+                                      nss.ns(),
+                                      BSON("$set" << BSON("bar" << 2)),
+                                      obj2,
+                                      3);
+
+    ASSERT_OK(rollbackOps({updateOp2, updateOp1}));
+    ASSERT_EQ(_rbInfo.rollbackDeletedIdsMap.size(), 1UL);
+
+    assertRollbackInfoContainsObjectForUUID(uuid, obj1);
+    assertRollbackInfoContainsObjectForUUID(uuid, obj2);
+}
 }  // namespace