SERVER-33669 Stepdown and shutdown should abort all uncommitted transactions

16 files changed, 132 insertions, 53 deletions

diff --git a/jstests/noPassthrough/snapshot_cursor_shutdown_stepdown.js b/jstests/noPassthrough/snapshot_cursor_shutdown_stepdown.js
new file mode 100644
index 00000000000..aaae19766aa
--- /dev/null
+++ b/jstests/noPassthrough/snapshot_cursor_shutdown_stepdown.js
@@ -0,0 +1,84 @@
+// Tests that stashed transaction resources are destroyed at shutdown and stepdown.
+// @tags: [requires_replication]
+(function() {
+    "use strict";
+
+    const dbName = "test";
+    const collName = "coll";
+
+    //
+    // Test that stashed transaction resources are destroyed at shutdown.
+    //
+
+    let rst = new ReplSetTest({nodes: 1});
+    rst.startSet();
+    rst.initiate();
+
+    let primaryDB = rst.getPrimary().getDB(dbName);
+    if (!primaryDB.serverStatus().storageEngine.supportsSnapshotReadConcern) {
+        rst.stopSet();
+        return;
+    }
+
+    let session = primaryDB.getMongo().startSession();
+    let sessionDB = session.getDatabase(dbName);
+
+    for (let i = 0; i < 4; i++) {
+        assert.commandWorked(sessionDB.coll.insert({_id: i}, {writeConcern: {w: "majority"}}));
+    }
+
+    // Create a snapshot read cursor.
+    assert.commandWorked(sessionDB.runCommand({
+        find: collName,
+        batchSize: 2,
+        readConcern: {level: "snapshot"},
+        txnNumber: NumberLong(0)
+    }));
+
+    // It should be possible to shut down the server without hanging. We must skip collection
+    // validation, since this will hang.
+    const signal = true;  // Use default kill signal.
+    const forRestart = false;
+    rst.stopSet(signal, forRestart, {skipValidation: true});
+
+    //
+    // Test that stashed transaction resources are destroyed at stepdown.
+    //
+
+    rst = new ReplSetTest({nodes: 2});
+    rst.startSet();
+    rst.initiate();
+
+    const primary = rst.getPrimary();
+    primaryDB = primary.getDB(dbName);
+
+    session = primaryDB.getMongo().startSession();
+    sessionDB = session.getDatabase(dbName);
+
+    for (let i = 0; i < 4; i++) {
+        assert.commandWorked(sessionDB.coll.insert({_id: i}, {writeConcern: {w: "majority"}}));
+    }
+
+    // Create a snapshot read cursor.
+    const res = assert.commandWorked(sessionDB.runCommand({
+        find: collName,
+        batchSize: 2,
+        readConcern: {level: "snapshot"},
+        txnNumber: NumberLong(0)
+    }));
+    assert(res.hasOwnProperty("cursor"), tojson(res));
+    assert(res.cursor.hasOwnProperty("id"), tojson(res));
+
+    // It should be possible to step down the primary without hanging.
+    assert.throws(function() {
+        primary.adminCommand({replSetStepDown: 60, force: true});
+    });
+    rst.waitForState(primary, ReplSetTest.State.SECONDARY);
+
+    // TODO SERVER-33690: Destroying stashed transaction resources should kill the cursor, so this
+    // getMore should fail.
+    assert.commandWorked(sessionDB.runCommand(
+        {getMore: res.cursor.id, collection: collName, txnNumber: NumberLong(0)}));
+
+    rst.stopSet();
+})();
diff --git a/src/mongo/db/db.cpp b/src/mongo/db/db.cpp
index 2f9344562d3..cacb7f941a3 100644
--- a/src/mongo/db/db.cpp
+++ b/src/mongo/db/db.cpp
@@ -491,8 +491,6 @@ ExitCode _initAndListen(int listenPort) {
               << startupWarningsLog;
     }
 
-    SessionCatalog::create(serviceContext);
-
     // This function may take the global lock.
     auto shardingInitialized =
         uassertStatusOK(ShardingState::get(startupOpCtx.get())
@@ -864,6 +862,11 @@ void shutdownTask() {
         repl::ReplicationCoordinator::get(serviceContext)->shutdown(opCtx);
 
         ShardingState::get(serviceContext)->shutDown(opCtx);
+
+        // Destroy all stashed transaction resources, in order to release locks.
+        SessionKiller::Matcher matcherAllSessions(
+            KillAllSessionsByPatternSet{makeKillAllSessionsByPattern(opCtx)});
+        killSessionsLocalKillTransactions(opCtx, matcherAllSessions);
     }
 
     serviceContext->setKillAllOperations();
diff --git a/src/mongo/db/kill_sessions_local.cpp b/src/mongo/db/kill_sessions_local.cpp
index fd206cca8fa..5f4a08b02a9 100644
--- a/src/mongo/db/kill_sessions_local.cpp
+++ b/src/mongo/db/kill_sessions_local.cpp
@@ -48,6 +48,7 @@ void killSessionsLocalKillCursors(OperationContext* opCtx, const SessionKiller::
     auto res = CursorManager::killCursorsWithMatchingSessions(opCtx, matcher);
     uassertStatusOK(res.first);
 }
+}  // namespace
 
 void killSessionsLocalKillTransactions(OperationContext* opCtx,
                                        const SessionKiller::Matcher& matcher) {
@@ -56,7 +57,6 @@ void killSessionsLocalKillTransactions(OperationContext* opCtx,
             session->abortTransaction();
         });
 }
-}  // namespace
 
 SessionKiller::Result killSessionsLocal(OperationContext* opCtx,
                                         const SessionKiller::Matcher& matcher,
diff --git a/src/mongo/db/kill_sessions_local.h b/src/mongo/db/kill_sessions_local.h
index 7177fad6b0d..c9d813f4915 100644
--- a/src/mongo/db/kill_sessions_local.h
+++ b/src/mongo/db/kill_sessions_local.h
@@ -39,4 +39,10 @@ SessionKiller::Result killSessionsLocal(OperationContext* opCtx,
                                         const SessionKiller::Matcher& matcher,
                                         SessionKiller::UniformRandomBitGenerator* urbg);
 
+/**
+ * Kills all transactions on mongod for sessions matching 'matcher'.
+ */
+void killSessionsLocalKillTransactions(OperationContext* opCtx,
+                                       const SessionKiller::Matcher& matcher);
+
 }  // namespace mongo
diff --git a/src/mongo/db/op_observer_impl_test.cpp b/src/mongo/db/op_observer_impl_test.cpp
index cc62f5dfbe1..91937db07b5 100644
--- a/src/mongo/db/op_observer_impl_test.cpp
+++ b/src/mongo/db/op_observer_impl_test.cpp
@@ -255,9 +255,8 @@ public:
     void setUp() override {
         OpObserverTest::setUp();
         auto opCtx = cc().makeOperationContext();
-        SessionCatalog::reset_forTest(getServiceContext());
-        SessionCatalog::create(getServiceContext());
         auto sessionCatalog = SessionCatalog::get(getServiceContext());
+        sessionCatalog->reset_forTest();
         sessionCatalog->onStepUp(opCtx.get());
     }
 
diff --git a/src/mongo/db/repl/SConscript b/src/mongo/db/repl/SConscript
index 68fb340da4b..78c7ebef39d 100644
--- a/src/mongo/db/repl/SConscript
+++ b/src/mongo/db/repl/SConscript
@@ -1542,6 +1542,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/concurrency/lock_manager',
         '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/index_d',
+        '$BUILD_DIR/mongo/db/kill_sessions_local',
         '$BUILD_DIR/mongo/db/lasterror',
         '$BUILD_DIR/mongo/db/logical_clock',
         '$BUILD_DIR/mongo/db/logical_time',
diff --git a/src/mongo/db/repl/do_txn_test.cpp b/src/mongo/db/repl/do_txn_test.cpp
index c9d940ee485..ebc0e58500f 100644
--- a/src/mongo/db/repl/do_txn_test.cpp
+++ b/src/mongo/db/repl/do_txn_test.cpp
@@ -105,8 +105,7 @@ void DoTxnTest::setUp() {
     ASSERT_OK(replCoord->setFollowerMode(MemberState::RS_PRIMARY));
 
     // Set up session catalog
-    SessionCatalog::reset_forTest(service);
-    SessionCatalog::create(service);
+    SessionCatalog::get(service)->reset_forTest();
     SessionCatalog::get(service)->onStepUp(_opCtx.get());
 
     // Need the OpObserverImpl in the registry in order for doTxn to work.
diff --git a/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp b/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp
index bb15c789cf2..6e5d8e608ca 100644
--- a/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp
+++ b/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp
@@ -49,6 +49,7 @@
 #include "mongo/db/dbdirectclient.h"
 #include "mongo/db/dbhelpers.h"
 #include "mongo/db/jsobj.h"
+#include "mongo/db/kill_sessions_local.h"
 #include "mongo/db/logical_time_metadata_hook.h"
 #include "mongo/db/logical_time_validator.h"
 #include "mongo/db/op_observer.h"
@@ -641,6 +642,11 @@ void ReplicationCoordinatorExternalStateImpl::closeConnections() {
 void ReplicationCoordinatorExternalStateImpl::killAllUserOperations(OperationContext* opCtx) {
     ServiceContext* environment = opCtx->getServiceContext();
     environment->killAllUserOperations(opCtx, ErrorCodes::InterruptedDueToReplStateChange);
+
+    // Destroy all stashed transaction resources, in order to release locks.
+    SessionKiller::Matcher matcherAllSessions(
+        KillAllSessionsByPatternSet{makeKillAllSessionsByPattern(opCtx)});
+    killSessionsLocalKillTransactions(opCtx, matcherAllSessions);
 }
 
 void ReplicationCoordinatorExternalStateImpl::shardingOnStepDownHook() {
diff --git a/src/mongo/db/repl/rollback_test_fixture.cpp b/src/mongo/db/repl/rollback_test_fixture.cpp
index 31ccc100913..1ed99e3bb8f 100644
--- a/src/mongo/db/repl/rollback_test_fixture.cpp
+++ b/src/mongo/db/repl/rollback_test_fixture.cpp
@@ -85,8 +85,6 @@ void RollbackTest::setUp() {
                                 std::unique_ptr<ReplicationCoordinator>(_coordinator));
     setOplogCollectionName(serviceContext);
 
-    SessionCatalog::create(serviceContext);
-
     _opCtx = cc().makeOperationContext();
     _replicationProcess->getConsistencyMarkers()->clearAppliedThrough(_opCtx.get(), {});
     _replicationProcess->getConsistencyMarkers()->setMinValid(_opCtx.get(), OpTime{});
@@ -101,7 +99,7 @@ void RollbackTest::tearDown() {
     _coordinator = nullptr;
     _opCtx.reset();
 
-    SessionCatalog::reset_forTest(_serviceContextMongoDTest.getServiceContext());
+    SessionCatalog::get(_serviceContextMongoDTest.getServiceContext())->reset_forTest();
 
     // We cannot unset the global replication coordinator because ServiceContextMongoD::tearDown()
     // calls dropAllDatabasesExceptLocal() which requires the replication coordinator to clear all
diff --git a/src/mongo/db/repl/sync_tail_test.cpp b/src/mongo/db/repl/sync_tail_test.cpp
index 1e101a42676..69a40a55b9c 100644
--- a/src/mongo/db/repl/sync_tail_test.cpp
+++ b/src/mongo/db/repl/sync_tail_test.cpp
@@ -1539,7 +1539,6 @@ public:
     void setUp() override {
         SyncTailTest::setUp();
 
-        SessionCatalog::create(_opCtx->getServiceContext());
         SessionCatalog::get(_opCtx->getServiceContext())->onStepUp(_opCtx.get());
 
         DBDirectClient client(_opCtx.get());
@@ -1547,7 +1546,7 @@ public:
         ASSERT(client.runCommand(kNs.db().toString(), BSON("create" << kNs.coll()), result));
     }
     void tearDown() override {
-        SessionCatalog::reset_forTest(_opCtx->getServiceContext());
+        SessionCatalog::get(_opCtx->getServiceContext())->reset_forTest();
         SyncTailTest::tearDown();
     }
 
diff --git a/src/mongo/db/s/session_catalog_migration_destination_test.cpp b/src/mongo/db/s/session_catalog_migration_destination_test.cpp
index ff1529fd6cd..b37e7056c53 100644
--- a/src/mongo/db/s/session_catalog_migration_destination_test.cpp
+++ b/src/mongo/db/s/session_catalog_migration_destination_test.cpp
@@ -129,13 +129,12 @@ public:
                 ->setFindHostReturnValue(kDonorConnStr.getServers()[0]);
         }
 
-        SessionCatalog::create(getServiceContext());
         SessionCatalog::get(getServiceContext())->onStepUp(operationContext());
         LogicalSessionCache::set(getServiceContext(), stdx::make_unique<LogicalSessionCacheNoop>());
     }
 
     void tearDown() override {
-        SessionCatalog::reset_forTest(getServiceContext());
+        SessionCatalog::get(getServiceContext())->reset_forTest();
         ShardServerTestFixture::tearDown();
     }
 
diff --git a/src/mongo/db/session_catalog.cpp b/src/mongo/db/session_catalog.cpp
index fe78ad94dbe..209fe31fc72 100644
--- a/src/mongo/db/session_catalog.cpp
+++ b/src/mongo/db/session_catalog.cpp
@@ -58,16 +58,13 @@ struct CheckedOutSession {
     int checkOutNestingLevel = 0;
 };
 
-const auto sessionTransactionTableDecoration =
-    ServiceContext::declareDecoration<boost::optional<SessionCatalog>>();
+const auto sessionTransactionTableDecoration = ServiceContext::declareDecoration<SessionCatalog>();
 
 const auto operationSessionDecoration =
     OperationContext::declareDecoration<boost::optional<CheckedOutSession>>();
 
 }  // namespace
 
-SessionCatalog::SessionCatalog(ServiceContext* serviceContext) : _serviceContext(serviceContext) {}
-
 SessionCatalog::~SessionCatalog() {
     stdx::lock_guard<stdx::mutex> lg(_mutex);
     for (const auto& entry : _txnTable) {
@@ -76,16 +73,9 @@ SessionCatalog::~SessionCatalog() {
     }
 }
 
-void SessionCatalog::create(ServiceContext* service) {
-    auto& sessionTransactionTable = sessionTransactionTableDecoration(service);
-    invariant(!sessionTransactionTable);
-
-    sessionTransactionTable.emplace(service);
-}
-
-void SessionCatalog::reset_forTest(ServiceContext* service) {
-    auto& sessionTransactionTable = sessionTransactionTableDecoration(service);
-    sessionTransactionTable.reset();
+void SessionCatalog::reset_forTest() {
+    stdx::lock_guard<stdx::mutex> lg(_mutex);
+    _txnTable.clear();
 }
 
 SessionCatalog* SessionCatalog::get(OperationContext* opCtx) {
@@ -94,9 +84,7 @@ SessionCatalog* SessionCatalog::get(OperationContext* opCtx) {
 
 SessionCatalog* SessionCatalog::get(ServiceContext* service) {
     auto& sessionTransactionTable = sessionTransactionTableDecoration(service);
-    invariant(sessionTransactionTable);
-
-    return sessionTransactionTable.get_ptr();
+    return &sessionTransactionTable;
 }
 
 boost::optional<UUID> SessionCatalog::getTransactionTableUUID(OperationContext* opCtx) {
diff --git a/src/mongo/db/session_catalog.h b/src/mongo/db/session_catalog.h
index d3b37c2183c..e5b45432285 100644
--- a/src/mongo/db/session_catalog.h
+++ b/src/mongo/db/session_catalog.h
@@ -53,24 +53,11 @@ class SessionCatalog {
     friend class ScopedCheckedOutSession;
 
 public:
-    explicit SessionCatalog(ServiceContext* serviceContext);
+    SessionCatalog() = default;
     ~SessionCatalog();
 
     /**
-     * Instantiates a transaction table on the specified service context. Must be called only once
-     * and is not thread-safe.
-     */
-    static void create(ServiceContext* service);
-
-    /**
-     * Resets the transaction table on the specified service context to an uninitialized state.
-     * Meant only for testing.
-     */
-    static void reset_forTest(ServiceContext* service);
-
-    /**
      * Retrieves the session transaction table associated with the service or operation context.
-     * Must only be called after 'create' has been called.
      */
     static SessionCatalog* get(OperationContext* opCtx);
     static SessionCatalog* get(ServiceContext* service);
@@ -82,6 +69,12 @@ public:
     static boost::optional<UUID> getTransactionTableUUID(OperationContext* opCtx);
 
     /**
+     * Resets the transaction table to an uninitialized state.
+     * Meant only for testing.
+     */
+    void reset_forTest();
+
+    /**
      * Invoked when the node enters the primary state. Ensures that the transactions collection is
      * created. Throws on severe exceptions due to which it is not safe to continue the step-up
      * process.
@@ -189,8 +182,6 @@ private:
      */
     void _releaseSession(const LogicalSessionId& lsid);
 
-    ServiceContext* const _serviceContext;
-
     stdx::mutex _mutex;
     SessionRuntimeInfoMap _txnTable;
 };
diff --git a/src/mongo/db/session_catalog_test.cpp b/src/mongo/db/session_catalog_test.cpp
index 42fb7d5b321..7ad97852248 100644
--- a/src/mongo/db/session_catalog_test.cpp
+++ b/src/mongo/db/session_catalog_test.cpp
@@ -49,8 +49,7 @@ protected:
         MockReplCoordServerFixture::setUp();
 
         auto service = opCtx()->getServiceContext();
-        SessionCatalog::reset_forTest(service);
-        SessionCatalog::create(service);
+        SessionCatalog::get(service)->reset_forTest();
         SessionCatalog::get(service)->onStepUp(opCtx());
     }
 
diff --git a/src/mongo/db/session_test.cpp b/src/mongo/db/session_test.cpp
index 0b0f3c728be..3c248646206 100644
--- a/src/mongo/db/session_test.cpp
+++ b/src/mongo/db/session_test.cpp
@@ -83,8 +83,7 @@ protected:
         MockReplCoordServerFixture::setUp();
 
         auto service = opCtx()->getServiceContext();
-        SessionCatalog::reset_forTest(service);
-        SessionCatalog::create(service);
+        SessionCatalog::get(service)->reset_forTest();
         SessionCatalog::get(service)->onStepUp(opCtx());
     }
 
diff --git a/src/mongo/shell/servers.js b/src/mongo/shell/servers.js
index ebe96ab57bd..4c868a34e0a 100644
--- a/src/mongo/shell/servers.js
+++ b/src/mongo/shell/servers.js
@@ -889,7 +889,9 @@ var MongoRunner, _startMongod, startMongoProgram, runMongoProgram, startMongoPro
      *      auth: {
      *        user {string}: admin user name
      *        pwd {string}: admin password
-     *      }
+     *      },
+     *      skipValidation: <bool>,
+     *      allowedExitCode: <int>
      *    }
      *
      * Note: The auth option is required in a authenticated mongod running in Windows since
@@ -928,7 +930,13 @@ var MongoRunner, _startMongod, startMongoProgram, runMongoProgram, startMongoPro
             // Invoke callback to validate collections and indexes before shutting down mongod.
             // We skip calling the callback function when the expected return code of
             // the mongod process is non-zero since it's likely the process has already exited.
-            if (allowedExitCode === MongoRunner.EXIT_CLEAN) {
+
+            var skipValidation = false;
+            if (opts.skipValidation) {
+                skipValidation = true;
+            }
+
+            if (allowedExitCode === MongoRunner.EXIT_CLEAN && !skipValidation) {
                 MongoRunner.validateCollectionsCallback(port);
             }