SERVER-44577 Ensure WiredTiger RecordStore and SortedDataInterface cursors have started a transaction before reading data

4 files changed, 239 insertions, 0 deletions

diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp
index e38da42c83e..137fb2cb52d 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp
@@ -1077,6 +1077,9 @@ protected:
 
     // Seeks to query. Returns true on exact match.
     bool seekWTCursor(const KeyString::Value& query) {
+        // Ensure an active transaction is open.
+        WiredTigerRecoveryUnit::get(_opCtx)->getSession();
+
         WT_CURSOR* c = _cursor->get();
 
         int cmp = -1;
@@ -1172,6 +1175,10 @@ protected:
         if (_eof) {
             return false;
         }
+
+        // Ensure an active transaction is open.
+        WiredTigerRecoveryUnit::get(_opCtx)->getSession();
+
         if (!_lastMoveSkippedKey) {
             advanceWTCursor();
         }
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.cpp
index dc5122b3f8c..eaea15ce763 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.cpp
@@ -2084,6 +2084,11 @@ boost::optional<Record> WiredTigerRecordStoreCursorBase::next() {
     if (_eof)
         return {};
 
+    // Ensure an active transaction is open. While WiredTiger supports using cursors on a session
+    // without an active transaction (i.e. an implicit transaction), that would bypass configuration
+    // options we pass when we explicitly start transactions in the RecoveryUnit.
+    WiredTigerRecoveryUnit::get(_opCtx)->getSession();
+
     WT_CURSOR* c = _cursor->get();
 
     RecordId id;
@@ -2143,6 +2148,11 @@ boost::optional<Record> WiredTigerRecordStoreCursorBase::seekExact(const RecordI
         return {};
     }
 
+    // Ensure an active transaction is open. While WiredTiger supports using cursors on a session
+    // without an active transaction (i.e. an implicit transaction), that would bypass configuration
+    // options we pass when we explicitly start transactions in the RecoveryUnit.
+    WiredTigerRecoveryUnit::get(_opCtx)->getSession();
+
     _skipNextAdvance = false;
     WT_CURSOR* c = _cursor->get();
     setKey(c, id);
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store_test.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store_test.cpp
index 9225fd8bc80..b32890b794b 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store_test.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store_test.cpp
@@ -915,5 +915,85 @@ TEST(WiredTigerRecordStoreTest, GetLatestOplogTest) {
     ASSERT_EQ(tsThree, wtrs->getLatestOplogTimestamp(op1.get()));
 }
 
+TEST(WiredTigerRecordStoreTest, CursorInActiveTxnAfterNext) {
+    unique_ptr<RecordStoreHarnessHelper> harnessHelper(newRecordStoreHarnessHelper());
+    unique_ptr<RecordStore> rs(harnessHelper->newNonCappedRecordStore());
+
+    RecordId rid1;
+    {
+        ServiceContext::UniqueOperationContext opCtx(harnessHelper->newOperationContext());
+
+        WriteUnitOfWork uow(opCtx.get());
+        StatusWith<RecordId> res = rs->insertRecord(opCtx.get(), "a", 2, Timestamp());
+        ASSERT_OK(res.getStatus());
+        rid1 = res.getValue();
+
+        res = rs->insertRecord(opCtx.get(), "b", 2, Timestamp());
+        ASSERT_OK(res.getStatus());
+
+        uow.commit();
+    }
+
+    // Cursors should always ensure they are in an active transaction when next() is called.
+    {
+        ServiceContext::UniqueOperationContext opCtx(harnessHelper->newOperationContext());
+        auto ru = WiredTigerRecoveryUnit::get(opCtx.get());
+
+        auto cursor = rs->getCursor(opCtx.get());
+        ASSERT(cursor->next());
+        ASSERT_TRUE(ru->inActiveTxn());
+
+        // Committing a WriteUnitOfWork will end the current transaction.
+        WriteUnitOfWork wuow(opCtx.get());
+        ASSERT_TRUE(ru->inActiveTxn());
+        wuow.commit();
+        ASSERT_FALSE(ru->inActiveTxn());
+
+        // If a cursor is used after a WUOW commits, it should implicitly start a new transaction.
+        ASSERT(cursor->next());
+        ASSERT_TRUE(ru->inActiveTxn());
+    }
+}
+
+TEST(WiredTigerRecordStoreTest, CursorInActiveTxnAfterSeek) {
+    unique_ptr<RecordStoreHarnessHelper> harnessHelper(newRecordStoreHarnessHelper());
+    unique_ptr<RecordStore> rs(harnessHelper->newNonCappedRecordStore());
+
+    RecordId rid1;
+    {
+        ServiceContext::UniqueOperationContext opCtx(harnessHelper->newOperationContext());
+
+        WriteUnitOfWork uow(opCtx.get());
+        StatusWith<RecordId> res = rs->insertRecord(opCtx.get(), "a", 2, Timestamp());
+        ASSERT_OK(res.getStatus());
+        rid1 = res.getValue();
+
+        res = rs->insertRecord(opCtx.get(), "b", 2, Timestamp());
+        ASSERT_OK(res.getStatus());
+
+        uow.commit();
+    }
+
+    // Cursors should always ensure they are in an active transaction when seekExact() is called.
+    {
+        ServiceContext::UniqueOperationContext opCtx(harnessHelper->newOperationContext());
+        auto ru = WiredTigerRecoveryUnit::get(opCtx.get());
+
+        auto cursor = rs->getCursor(opCtx.get());
+        ASSERT(cursor->seekExact(rid1));
+        ASSERT_TRUE(ru->inActiveTxn());
+
+        // Committing a WriteUnitOfWork will end the current transaction.
+        WriteUnitOfWork wuow(opCtx.get());
+        ASSERT_TRUE(ru->inActiveTxn());
+        wuow.commit();
+        ASSERT_FALSE(ru->inActiveTxn());
+
+        // If a cursor is used after a WUOW commits, it should implicitly start a new transaction.
+        ASSERT(cursor->seekExact(rid1));
+        ASSERT_TRUE(ru->inActiveTxn());
+    }
+}
+
 }  // namespace
 }  // namespace mongo
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_standard_index_test.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_standard_index_test.cpp
index 344c385c791..0d736c6a3b8 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_standard_index_test.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_standard_index_test.cpp
@@ -146,5 +146,147 @@ MONGO_INITIALIZER(RegisterSortedDataInterfaceHarnessFactory)(InitializerContext*
     mongo::registerSortedDataInterfaceHarnessHelperFactory(makeWTIndexHarnessHelper);
     return Status::OK();
 }
+
+TEST(WiredTigerStandardIndexText, CursorInActiveTxnAfterNext) {
+    auto harnessHelper = makeWTIndexHarnessHelper();
+    bool unique = false;
+    bool partial = false;
+    auto sdi = harnessHelper->newSortedDataInterface(unique, partial);
+
+    // Populate data.
+    {
+        ServiceContext::UniqueOperationContext opCtx(harnessHelper->newOperationContext());
+
+        WriteUnitOfWork uow(opCtx.get());
+        auto ks = makeKeyString(sdi.get(), BSON("" << 1), RecordId(1));
+        auto res = sdi->insert(opCtx.get(), ks, true);
+        ASSERT_OK(res);
+
+        ks = makeKeyString(sdi.get(), BSON("" << 2), RecordId(2));
+        res = sdi->insert(opCtx.get(), ks, true);
+        ASSERT_OK(res);
+
+        uow.commit();
+    }
+
+    // Cursors should always ensure they are in an active transaction when next() is called.
+    {
+        ServiceContext::UniqueOperationContext opCtx(harnessHelper->newOperationContext());
+        auto ru = WiredTigerRecoveryUnit::get(opCtx.get());
+
+        auto cursor = sdi->newCursor(opCtx.get());
+        auto res = cursor->seek(makeKeyStringForSeek(sdi.get(), BSONObj(), true, true));
+        ASSERT(res);
+
+        ASSERT_TRUE(ru->inActiveTxn());
+
+        // Committing a WriteUnitOfWork will end the current transaction.
+        WriteUnitOfWork wuow(opCtx.get());
+        ASSERT_TRUE(ru->inActiveTxn());
+        wuow.commit();
+        ASSERT_FALSE(ru->inActiveTxn());
+
+        // If a cursor is used after a WUOW commits, it should implicitly start a new transaction.
+        ASSERT(cursor->next());
+        ASSERT_TRUE(ru->inActiveTxn());
+    }
+}
+
+TEST(WiredTigerStandardIndexText, CursorInActiveTxnAfterSeek) {
+    auto harnessHelper = makeWTIndexHarnessHelper();
+    bool unique = false;
+    bool partial = false;
+    auto sdi = harnessHelper->newSortedDataInterface(unique, partial);
+
+    // Populate data.
+    {
+        ServiceContext::UniqueOperationContext opCtx(harnessHelper->newOperationContext());
+
+        WriteUnitOfWork uow(opCtx.get());
+        auto ks = makeKeyString(sdi.get(), BSON("" << 1), RecordId(1));
+        auto res = sdi->insert(opCtx.get(), ks, true);
+        ASSERT_OK(res);
+
+        ks = makeKeyString(sdi.get(), BSON("" << 2), RecordId(2));
+        res = sdi->insert(opCtx.get(), ks, true);
+        ASSERT_OK(res);
+
+        uow.commit();
+    }
+
+    // Cursors should always ensure they are in an active transaction when seek() is called.
+    {
+        ServiceContext::UniqueOperationContext opCtx(harnessHelper->newOperationContext());
+        auto ru = WiredTigerRecoveryUnit::get(opCtx.get());
+
+        auto cursor = sdi->newCursor(opCtx.get());
+
+        bool forward = true;
+        bool inclusive = true;
+        auto seekKs = makeKeyStringForSeek(sdi.get(), BSON("" << 1), forward, inclusive);
+        ASSERT(cursor->seek(seekKs));
+        ASSERT_TRUE(ru->inActiveTxn());
+
+        // Committing a WriteUnitOfWork will end the current transaction.
+        WriteUnitOfWork wuow(opCtx.get());
+        ASSERT_TRUE(ru->inActiveTxn());
+        wuow.commit();
+        ASSERT_FALSE(ru->inActiveTxn());
+
+        // If a cursor is used after a WUOW commits, it should implicitly start a new
+        // transaction.
+        ASSERT(cursor->seek(seekKs));
+        ASSERT_TRUE(ru->inActiveTxn());
+    }
+}
+
+TEST(WiredTigerStandardIndexText, CursorInActiveTxnAfterSeekExact) {
+    auto harnessHelper = makeWTIndexHarnessHelper();
+    bool unique = false;
+    bool partial = false;
+    auto sdi = harnessHelper->newSortedDataInterface(unique, partial);
+
+    // Populate data.
+    {
+        ServiceContext::UniqueOperationContext opCtx(harnessHelper->newOperationContext());
+
+        WriteUnitOfWork uow(opCtx.get());
+        auto ks = makeKeyString(sdi.get(), BSON("" << 1), RecordId(1));
+        auto res = sdi->insert(opCtx.get(), ks, true);
+        ASSERT_OK(res);
+
+        ks = makeKeyString(sdi.get(), BSON("" << 2), RecordId(2));
+        res = sdi->insert(opCtx.get(), ks, true);
+        ASSERT_OK(res);
+
+        uow.commit();
+    }
+
+    // Cursors should always ensure they are in an active transaction when seekExact() is
+    // called.
+    {
+        ServiceContext::UniqueOperationContext opCtx(harnessHelper->newOperationContext());
+        auto ru = WiredTigerRecoveryUnit::get(opCtx.get());
+
+        auto cursor = sdi->newCursor(opCtx.get());
+
+        auto seekKs = makeKeyString(sdi.get(), BSON("" << 1));
+
+        ASSERT(cursor->seekExact(seekKs));
+        ASSERT_TRUE(ru->inActiveTxn());
+
+        // Committing a WriteUnitOfWork will end the current transaction.
+        WriteUnitOfWork wuow(opCtx.get());
+        ASSERT_TRUE(ru->inActiveTxn());
+        wuow.commit();
+        ASSERT_FALSE(ru->inActiveTxn());
+
+        // If a cursor is used after a WUOW commits, it should implicitly start a new
+        // transaction.
+        ASSERT(cursor->seekExact(seekKs));
+        ASSERT_TRUE(ru->inActiveTxn());
+    }
+}
+
 }  // namespace
 }  // namespace mongo