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

5 files changed, 876 insertions, 64 deletions

diff --git a/src/mongo/db/dbhelpers.h b/src/mongo/db/dbhelpers.h
index 0166a413041..75699cbafa8 100644
--- a/src/mongo/db/dbhelpers.h
+++ b/src/mongo/db/dbhelpers.h
@@ -166,6 +166,14 @@ struct Helpers {
          */
         Status goingToDelete(const BSONObj& o);
 
+        std::string directoryName() const {
+            return _root.generic_string();
+        }
+
+        std::string fileName() const {
+            return _file.generic_string();
+        }
+
     private:
         boost::filesystem::path _root;
         boost::filesystem::path _file;
diff --git a/src/mongo/db/op_observer.h b/src/mongo/db/op_observer.h
index c8ce8ab6759..4d5254c18ec 100644
--- a/src/mongo/db/op_observer.h
+++ b/src/mongo/db/op_observer.h
@@ -31,6 +31,7 @@
 #include <string>
 
 #include "mongo/base/disallow_copying.h"
+#include "mongo/bson/simple_bsonobj_comparator.h"
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/catalog/collection_options.h"
 #include "mongo/db/jsobj.h"
@@ -253,6 +254,13 @@ public:
         // Set of all session ids from ops being rolled back.
         std::set<UUID> rollbackSessionIds = {};
 
+        // Maps UUIDs to a set of BSONObjs containing the _ids of the documents that will be deleted
+        // from that collection due to rollback, and is used to populate rollback files.
+        // For simplicity, this BSONObj set uses the simple binary comparison, as it is never wrong
+        // to consider two _ids as distinct even if the collection default collation would put them
+        // in the same equivalence class.
+        stdx::unordered_map<UUID, SimpleBSONObjUnorderedSet, UUID::Hash> rollbackDeletedIdsMap;
+
         // True if the shard identity document was rolled back.
         bool shardIdentityRolledBack = false;
     };
diff --git a/src/mongo/db/repl/rollback_impl.cpp b/src/mongo/db/repl/rollback_impl.cpp
index 303c3146a07..7960232d10d 100644
--- a/src/mongo/db/repl/rollback_impl.cpp
+++ b/src/mongo/db/repl/rollback_impl.cpp
@@ -33,9 +33,11 @@
 #include "mongo/db/repl/rollback_impl.h"
 
 #include "mongo/db/background.h"
+#include "mongo/db/catalog/uuid_catalog.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/concurrency/d_concurrency.h"
 #include "mongo/db/db_raii.h"
+#include "mongo/db/dbhelpers.h"
 #include "mongo/db/logical_time_validator.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/db/repl/apply_ops.h"
@@ -60,6 +62,9 @@ constexpr bool createRollbackFilesDefault = true;
 MONGO_EXPORT_SERVER_PARAMETER(createRollbackDataFiles, bool, createRollbackFilesDefault);
 }  // namespace
 
+constexpr const char* RollbackImpl::kRollbackRemoveSaverType;
+constexpr const char* RollbackImpl::kRollbackRemoveSaverWhy;
+
 bool RollbackImpl::shouldCreateDataFiles() {
     return createRollbackDataFiles.load();
 }
@@ -70,12 +75,12 @@ RollbackImpl::RollbackImpl(OplogInterface* localOplog,
                            ReplicationProcess* replicationProcess,
                            ReplicationCoordinator* replicationCoordinator,
                            Listener* listener)
-    : _localOplog(localOplog),
+    : _listener(listener),
+      _localOplog(localOplog),
       _remoteOplog(remoteOplog),
       _storageInterface(storageInterface),
       _replicationProcess(replicationProcess),
-      _replicationCoordinator(replicationCoordinator),
-      _listener(listener) {
+      _replicationCoordinator(replicationCoordinator) {
 
     invariant(localOplog);
     invariant(remoteOplog);
@@ -133,6 +138,13 @@ Status RollbackImpl::runRollback(OperationContext* opCtx) {
         return status;
     }
 
+    // Write a rollback file for each namespace that has documents that would be deleted by
+    // rollback.
+    status = _writeRollbackFiles(opCtx);
+    if (!status.isOK()) {
+        return status;
+    }
+
     // Recover to the stable timestamp.
     auto stableTimestampSW = _recoverToStableTimestamp(opCtx);
     if (!stableTimestampSW.isOK()) {
@@ -366,8 +378,22 @@ Status RollbackImpl::_processRollbackOp(const OplogEntry& oplogEntry) {
         _observerInfo.rollbackSessionIds.insert(sessionId->getId());
     }
 
-    // Check if the creation of the shard identity document is being rolled back.
+    // Keep track of the _ids of inserted and updated documents, as we may need to write them out to
+    // a rollback file.
+    if (opType == OpTypeEnum::kInsert || opType == OpTypeEnum::kUpdate) {
+        const auto uuid = oplogEntry.getUuid();
+        dassert(uuid);
+        const auto idElem = oplogEntry.getIdElement();
+        if (!idElem.eoo()) {
+            // We call BSONElement::wrap() on each _id element to create a new BSONObj with an owned
+            // buffer, as the underlying storage may be gone when we access this map to write
+            // rollback files.
+            _observerInfo.rollbackDeletedIdsMap[uuid.get()].insert(idElem.wrap());
+        }
+    }
+
     if (opType == OpTypeEnum::kInsert) {
+        // Check if the creation of the shard identity document is being rolled back.
         auto idVal = oplogEntry.getObject().getStringField("_id");
         if (serverGlobalParams.clusterRole == ClusterRole::ShardServer &&
             opNss == NamespaceString::kServerConfigurationNamespace &&
@@ -460,6 +486,68 @@ Timestamp RollbackImpl::_findTruncateTimestamp(
     return truncatePointTime.getValue().getTimestamp();
 }
 
+boost::optional<BSONObj> RollbackImpl::_findDocumentById(OperationContext* opCtx,
+                                                         UUID uuid,
+                                                         NamespaceString nss,
+                                                         BSONElement id) {
+    auto document = _storageInterface->findById(opCtx, {nss.db().toString(), uuid}, id);
+    if (document.isOK()) {
+        return document.getValue();
+    } else if (document.getStatus().code() == ErrorCodes::NoSuchKey) {
+        return boost::none;
+    } else {
+        severe() << "Rollback failed to read document with " << redact(id) << " in namespace "
+                 << nss.ns() << " with uuid " << uuid.toString() << causedBy(document.getStatus());
+        fassert(50751, document.getStatus());
+    }
+
+    MONGO_UNREACHABLE;
+}
+
+Status RollbackImpl::_writeRollbackFiles(OperationContext* opCtx) {
+    const auto& uuidCatalog = UUIDCatalog::get(opCtx);
+    for (auto&& entry : _observerInfo.rollbackDeletedIdsMap) {
+        const auto& uuid = entry.first;
+        const auto nss = uuidCatalog.lookupNSSByUUID(uuid);
+        invariant(!nss.isEmpty(),
+                  str::stream() << "The collection with UUID " << uuid
+                                << " is unexpectedly missing in the UUIDCatalog");
+
+        if (_isInShutdown()) {
+            log() << "Rollback shutting down; not writing rollback file for namespace " << nss.ns()
+                  << " with uuid " << uuid;
+            continue;
+        }
+
+        _writeRollbackFileForNamespace(opCtx, uuid, nss, entry.second);
+    }
+
+    if (_isInShutdown()) {
+        return {ErrorCodes::ShutdownInProgress, "rollback shutting down"};
+    }
+
+    return Status::OK();
+}
+
+void RollbackImpl::_writeRollbackFileForNamespace(OperationContext* opCtx,
+                                                  UUID uuid,
+                                                  NamespaceString nss,
+                                                  const SimpleBSONObjUnorderedSet& idSet) {
+    Helpers::RemoveSaver removeSaver(kRollbackRemoveSaverType, nss.ns(), kRollbackRemoveSaverWhy);
+    log() << "Preparing to write deleted documents to a rollback file for collection " << nss.ns()
+          << " with uuid " << uuid.toString() << " to " << removeSaver.fileName();
+    for (auto&& id : idSet) {
+        // StorageInterface::findById() does not respect the collation, but because we are using
+        // exact _id fields recorded in the oplog, we can get away with binary string
+        // comparisons.
+        auto document = _findDocumentById(opCtx, uuid, nss, id.firstElement());
+        if (document) {
+            fassert(50750, removeSaver.goingToDelete(*document));
+        }
+    }
+    _listener->onRollbackFileWrittenForNamespace(std::move(uuid), std::move(nss));
+}
+
 StatusWith<Timestamp> RollbackImpl::_recoverToStableTimestamp(OperationContext* opCtx) {
     if (_isInShutdown()) {
         return Status(ErrorCodes::ShutdownInProgress, "rollback shutting down");
diff --git a/src/mongo/db/repl/rollback_impl.h b/src/mongo/db/repl/rollback_impl.h
index d19da54cac6..6331e69d34d 100644
--- a/src/mongo/db/repl/rollback_impl.h
+++ b/src/mongo/db/repl/rollback_impl.h
@@ -90,6 +90,12 @@ class ReplicationProcess;
 class RollbackImpl : public Rollback {
 public:
     /**
+     * Used to indicate that the files we create with deleted documents are from rollback.
+     */
+    static constexpr auto kRollbackRemoveSaverType = "rollback";
+    static constexpr auto kRollbackRemoveSaverWhy = "removed";
+
+    /**
      * A class with functions that get called throughout rollback. These can be overridden to
      * instrument this class for diagnostics and testing.
      */
@@ -113,6 +119,12 @@ public:
         virtual void onCommonPointFound(Timestamp commonPoint) noexcept {}
 
         /**
+         * Function called after a rollback file has been written for each namespace with inserts or
+         * updates that are being rolled back.
+         */
+        virtual void onRollbackFileWrittenForNamespace(UUID, NamespaceString) noexcept {}
+
+        /**
          * Function called after we recover to the stable timestamp.
          */
         virtual void onRecoverToStableTimestamp(Timestamp stableTimestamp) noexcept {}
@@ -184,6 +196,53 @@ public:
      */
     static bool shouldCreateDataFiles();
 
+    /**
+     * Returns a structure containing all of the documents that would have been written to a
+     * rollback data file for the namespace represented by 'uuid'.
+     *
+     * Only exposed for testing. It is invalid to call this function on a real RollbackImpl.
+     */
+    virtual const std::vector<BSONObj>& docsDeletedForNamespace_forTest(UUID uuid) const& {
+        MONGO_UNREACHABLE;
+    }
+    void docsDeletedForNamespace_forTest(UUID)&& = delete;
+
+protected:
+    /**
+     * Returns the document with _id 'id' in the namespace 'nss', or boost::none if that document
+     * no longer exists in 'nss'. This function is used to write documents to rollback data files,
+     * and this function will terminate the server if an unexpected error is returned by the storage
+     * interface.
+     *
+     * This function is protected so that subclasses can access this method for test purposes.
+     */
+    boost::optional<BSONObj> _findDocumentById(OperationContext* opCtx,
+                                               UUID uuid,
+                                               NamespaceString nss,
+                                               BSONElement id);
+
+    /**
+     * Writes a rollback file for the namespace 'nss' containing all of the documents whose _ids are
+     * listed in 'idSet'.
+     *
+     * This function is protected so that subclasses can override it for test purposes.
+     */
+    virtual void _writeRollbackFileForNamespace(OperationContext* opCtx,
+                                                UUID uuid,
+                                                NamespaceString nss,
+                                                const SimpleBSONObjUnorderedSet& idSet);
+
+    // All member variables are labeled with one of the following codes indicating the
+    // synchronization rules for accessing them.
+    //
+    // (R)  Read-only in concurrent operation; no synchronization required.
+    // (S)  Self-synchronizing; access in any way from any context.
+    // (M)  Reads and writes guarded by _mutex.
+    // (N)  Should only ever be accessed by a single thread; no synchronization required.
+
+    // A listener that's called at various points throughout rollback.
+    Listener* _listener;  // (R)
+
 private:
     /**
      * Returns if shutdown was called on this rollback process.
@@ -236,7 +295,8 @@ private:
 
     /**
      * Process a single oplog entry that is getting rolled back and update the necessary rollback
-     * info structures.
+     * info structures. This function assumes that oplog entries are processed in descending
+     * timestamp order (that is, starting from the newest oplog entry, going backwards).
      */
     Status _processRollbackOp(const OplogEntry& oplogEntry);
 
@@ -262,13 +322,17 @@ private:
      */
     StatusWith<std::set<NamespaceString>> _namespacesForOp(const OplogEntry& oplogEntry);
 
-    // All member variables are labeled with one of the following codes indicating the
-    // synchronization rules for accessing them.
-    //
-    // (R)  Read-only in concurrent operation; no synchronization required.
-    // (S)  Self-synchronizing; access in any way from any context.
-    // (M)  Reads and writes guarded by _mutex.
-    // (N)  Should only ever be accessed by a single thread; no synchronization required.
+    /**
+     * Persists rollback files to disk for each namespace that contains documents inserted or
+     * updated after the common point, as these changes will be gone after rollback completes.
+     * Before each namespace is examined, we check for interrupt and return a non-OK status if
+     * shutdown is in progress.
+     *
+     * This function causes the server to terminate if an error occurs while fetching documents from
+     * disk or while writing documents to the rollback file. It must be called before marking the
+     * oplog truncate point, and before the storage engine recovers to the stable timestamp.
+     */
+    Status _writeRollbackFiles(OperationContext* opCtx);
 
     // Guards access to member variables.
     mutable stdx::mutex _mutex;  // (S)
@@ -294,9 +358,6 @@ private:
     // - update transition member states;
     ReplicationCoordinator* const _replicationCoordinator;  // (R)
 
-    // A listener that's called at various points throughout rollback.
-    Listener* _listener;  // (R)
-
     // Contains information about the rollback that will be passed along to the rollback OpObserver
     // method.
     OpObserver::RollbackObserverInfo _observerInfo = {};  // (N)
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