SERVER-39796 Make the transaction history iterator virtualizable, and make it available from

             the OplogInterface used in rollback.

9 files changed, 88 insertions, 0 deletions

diff --git a/src/mongo/db/repl/SConscript b/src/mongo/db/repl/SConscript
index e4a39ceeddc..420539e791a 100644
--- a/src/mongo/db/repl/SConscript
+++ b/src/mongo/db/repl/SConscript
@@ -513,6 +513,7 @@ env.Library(
     LIBDEPS=[
         '$BUILD_DIR/mongo/db/db_raii',
         '$BUILD_DIR/mongo/db/query_exec',
+        '$BUILD_DIR/mongo/db/transaction',
     ],
 )
 
@@ -522,6 +523,7 @@ env.Library(
         'oplog_interface_mock.cpp',
     ],
     LIBDEPS=[
+        'oplog_entry',
         '$BUILD_DIR/mongo/base',
         '$BUILD_DIR/mongo/util/net/network',
     ],
diff --git a/src/mongo/db/repl/oplog_interface.h b/src/mongo/db/repl/oplog_interface.h
index 710e1b6de6b..5b0831e0c28 100644
--- a/src/mongo/db/repl/oplog_interface.h
+++ b/src/mongo/db/repl/oplog_interface.h
@@ -36,6 +36,7 @@
 #include "mongo/base/status_with.h"
 #include "mongo/bson/bsonobj.h"
 #include "mongo/db/record_id.h"
+#include "mongo/db/transaction_history_iterator.h"
 #include "mongo/util/net/hostandport.h"
 
 namespace mongo {
@@ -61,6 +62,13 @@ public:
     virtual std::unique_ptr<Iterator> makeIterator() const = 0;
 
     /**
+     * Produces an iterator that returns operations within a transaction.  Valid only for local
+     * oplogs.
+     */
+    virtual std::unique_ptr<TransactionHistoryIteratorBase> makeTransactionHistoryIterator(
+        const OpTime& startingOpTime) const = 0;
+
+    /**
      * The host and port of the server.
      */
     virtual HostAndPort hostAndPort() const = 0;
diff --git a/src/mongo/db/repl/oplog_interface_local.cpp b/src/mongo/db/repl/oplog_interface_local.cpp
index b44baf7d1c8..1d79763f37d 100644
--- a/src/mongo/db/repl/oplog_interface_local.cpp
+++ b/src/mongo/db/repl/oplog_interface_local.cpp
@@ -100,6 +100,11 @@ std::unique_ptr<OplogInterface::Iterator> OplogInterfaceLocal::makeIterator() co
     return std::unique_ptr<OplogInterface::Iterator>(new OplogIteratorLocal(_opCtx));
 }
 
+std::unique_ptr<TransactionHistoryIteratorBase> OplogInterfaceLocal::makeTransactionHistoryIterator(
+    const OpTime& startingOpTime) const {
+    return std::make_unique<TransactionHistoryIterator>(startingOpTime);
+}
+
 HostAndPort OplogInterfaceLocal::hostAndPort() const {
     return {getHostNameCached(), serverGlobalParams.port};
 }
diff --git a/src/mongo/db/repl/oplog_interface_local.h b/src/mongo/db/repl/oplog_interface_local.h
index a7b7134a36f..d8941550a3f 100644
--- a/src/mongo/db/repl/oplog_interface_local.h
+++ b/src/mongo/db/repl/oplog_interface_local.h
@@ -46,6 +46,8 @@ public:
     OplogInterfaceLocal(OperationContext* opCtx);
     std::string toString() const override;
     std::unique_ptr<OplogInterface::Iterator> makeIterator() const override;
+    std::unique_ptr<TransactionHistoryIteratorBase> makeTransactionHistoryIterator(
+        const OpTime& startingOpTime) const override;
     HostAndPort hostAndPort() const override;
 
 private:
diff --git a/src/mongo/db/repl/oplog_interface_mock.cpp b/src/mongo/db/repl/oplog_interface_mock.cpp
index 1dd623a07ea..b3b8de86c90 100644
--- a/src/mongo/db/repl/oplog_interface_mock.cpp
+++ b/src/mongo/db/repl/oplog_interface_mock.cpp
@@ -30,6 +30,7 @@
 #include "mongo/platform/basic.h"
 
 #include "mongo/db/repl/oplog_interface_mock.h"
+#include "mongo/db/transaction_history_iterator.h"
 
 namespace mongo {
 namespace repl {
@@ -58,6 +59,55 @@ StatusWith<OplogInterface::Iterator::Value> OplogIteratorMock::next() {
     return *(_iterator++);
 }
 
+class TransactionHistoryIteratorMock : public TransactionHistoryIteratorBase {
+public:
+    TransactionHistoryIteratorMock(const OpTime& startOpTime,
+                                   std::unique_ptr<OplogInterface::Iterator> iter)
+        : _nextOpTime(startOpTime), _iter(std::move(iter)) {}
+
+    repl::OplogEntry next(OperationContext*) override {
+        invariant(hasNext());
+        auto operation = _iter->next();
+        while (operation.isOK()) {
+            auto& oplogBSON = operation.getValue().first;
+            auto oplogEntry = uassertStatusOK(repl::OplogEntry::parse(oplogBSON));
+            if (oplogEntry.getOpTime() == _nextOpTime) {
+                const auto& oplogPrevTsOption = oplogEntry.getPrevWriteOpTimeInTransaction();
+                uassert(
+                    ErrorCodes::FailedToParse,
+                    str::stream()
+                        << "Missing prevTs field on oplog entry of previous write in transaction: "
+                        << oplogBSON,
+                    oplogPrevTsOption);
+
+                _nextOpTime = oplogPrevTsOption.value();
+                return oplogEntry;
+            }
+            operation = _iter->next();
+        }
+        if (operation.getStatus().code() == ErrorCodes::NoSuchKey) {
+            uasserted(ErrorCodes::IncompleteTransactionHistory,
+                      str::stream()
+                          << "oplog no longer contains the complete write history of this "
+                             "transaction, log with opTime "
+                          << _nextOpTime.toBSON()
+                          << " cannot be found");
+        }
+        // We shouldn't get any other error.
+        MONGO_UNREACHABLE;
+    }
+
+    virtual ~TransactionHistoryIteratorMock() {}
+
+    bool hasNext() const override {
+        return !_nextOpTime.isNull();
+    }
+
+private:
+    repl::OpTime _nextOpTime;
+    std::unique_ptr<OplogInterface::Iterator> _iter;
+};
+
 }  // namespace
 
 OplogInterfaceMock::OplogInterfaceMock(std::initializer_list<Operation> operations)
@@ -78,6 +128,11 @@ std::unique_ptr<OplogInterface::Iterator> OplogInterfaceMock::makeIterator() con
         new OplogIteratorMock(_operations.begin(), _operations.end()));
 }
 
+std::unique_ptr<TransactionHistoryIteratorBase> OplogInterfaceMock::makeTransactionHistoryIterator(
+    const OpTime& startOpTime) const {
+    return std::make_unique<TransactionHistoryIteratorMock>(startOpTime, makeIterator());
+}
+
 HostAndPort OplogInterfaceMock::hostAndPort() const {
     // Returns a default-constructed HostAndPort, which has an empty hostname and an invalid port.
     return {};
diff --git a/src/mongo/db/repl/oplog_interface_mock.h b/src/mongo/db/repl/oplog_interface_mock.h
index 678a0b28c8c..d6d6c66617c 100644
--- a/src/mongo/db/repl/oplog_interface_mock.h
+++ b/src/mongo/db/repl/oplog_interface_mock.h
@@ -33,6 +33,8 @@
 #include <initializer_list>
 
 namespace mongo {
+class TransactionHistoryIteratorBase;
+
 namespace repl {
 
 /**
@@ -51,6 +53,8 @@ public:
     void setOperations(const Operations& operations);
     std::string toString() const override;
     std::unique_ptr<OplogInterface::Iterator> makeIterator() const override;
+    std::unique_ptr<TransactionHistoryIteratorBase> makeTransactionHistoryIterator(
+        const OpTime& startOpTime) const override;
     HostAndPort hostAndPort() const override;
 
 private:
diff --git a/src/mongo/db/repl/oplog_interface_remote.cpp b/src/mongo/db/repl/oplog_interface_remote.cpp
index 4c0332cdfa6..eefac22ae76 100644
--- a/src/mongo/db/repl/oplog_interface_remote.cpp
+++ b/src/mongo/db/repl/oplog_interface_remote.cpp
@@ -87,6 +87,12 @@ std::unique_ptr<OplogInterface::Iterator> OplogInterfaceRemote::makeIterator() c
             NamespaceString(_collectionName), query, 0, 0, &fields, 0, _batchSize)));
 }
 
+std::unique_ptr<TransactionHistoryIteratorBase>
+OplogInterfaceRemote::makeTransactionHistoryIterator(const OpTime&) const {
+    // Should never ask for remote transaction history.
+    MONGO_UNREACHABLE;
+}
+
 HostAndPort OplogInterfaceRemote::hostAndPort() const {
     return _hostAndPort;
 }
diff --git a/src/mongo/db/repl/oplog_interface_remote.h b/src/mongo/db/repl/oplog_interface_remote.h
index 9403319e14e..4b688ca5898 100644
--- a/src/mongo/db/repl/oplog_interface_remote.h
+++ b/src/mongo/db/repl/oplog_interface_remote.h
@@ -55,6 +55,8 @@ public:
                          int batchSize);
     std::string toString() const override;
     std::unique_ptr<OplogInterface::Iterator> makeIterator() const override;
+    std::unique_ptr<TransactionHistoryIteratorBase> makeTransactionHistoryIterator(
+        const OpTime& startingOpTime) const override;
     HostAndPort hostAndPort() const override;
 
 private:
diff --git a/src/mongo/db/repl/roll_back_local_operations_test.cpp b/src/mongo/db/repl/roll_back_local_operations_test.cpp
index 4965f6cfa9c..0293b16a860 100644
--- a/src/mongo/db/repl/roll_back_local_operations_test.cpp
+++ b/src/mongo/db/repl/roll_back_local_operations_test.cpp
@@ -72,6 +72,10 @@ TEST(RollBackLocalOperationsTest, InvalidLocalOplogIterator) {
         std::unique_ptr<Iterator> makeIterator() const override {
             return std::unique_ptr<Iterator>();
         }
+        std::unique_ptr<TransactionHistoryIteratorBase> makeTransactionHistoryIterator(
+            const OpTime& startingOpTime) const override {
+            MONGO_UNREACHABLE;
+        };
         HostAndPort hostAndPort() const override {
             return {};
         }