SERVER-75725 Control TaskExecutorCursor's connection-pinning behavior with a server parameter and add appropriate testing. In this PR, we:

    - Remove the now-obsolete getCallbackHandle interface from TEC.
    - Hook up the TEC default pinConnection-parameter to the relevant server parameter
    - Add a new suite running all search tests with that server parameter enabled
    - Fix a small bug to ensure we use the SSL mode provided by the request in PCTE
    - Fix a small bug to appropriately handle the case where we can't acquire a stream from the transport layer

10 files changed, 90 insertions, 30 deletions

diff --git a/buildscripts/resmokeconfig/suites/search_pinned_connections_auth.yml b/buildscripts/resmokeconfig/suites/search_pinned_connections_auth.yml
new file mode 100644
index 00000000000..0fd19d6e3f1
--- /dev/null
+++ b/buildscripts/resmokeconfig/suites/search_pinned_connections_auth.yml
@@ -0,0 +1,23 @@
+config_variables:
+- &keyFile jstests/libs/authTestsKey
+- &keyFileData Thiskeyisonlyforrunningthesuitewithauthenticationdontuseitinanytestsdirectly
+
+test_kind: js_test
+
+selector:
+  roots:
+  - src/mongo/db/modules/*/jstests/search/*.js
+
+executor:
+  config:
+    shell_options:
+      global_vars:
+        TestData:
+          auth: true
+          authMechanism: SCRAM-SHA-256
+          keyFile: *keyFile
+          keyFileData: *keyFileData
+          roleGraphInvalidationIsFatal: true
+          setParameters:
+            pinTaskExecCursorConns: true
+      nodb: ''
diff --git a/etc/evergreen.yml b/etc/evergreen.yml
index b5e9c0707ae..28313f81151 100644
--- a/etc/evergreen.yml
+++ b/etc/evergreen.yml
@@ -271,6 +271,7 @@ variables:
   - name: sasl
   - name: search
   - name: search_auth
+  - name: search_pinned_connections_auth
   - name: search_ssl
   - name: sharding_auth_audit_gen
   - name: .stitch
@@ -470,6 +471,7 @@ buildvariants:
   - name: sasl
   - name: search
   - name: search_auth
+  - name: search_pinned_connections_auth
   - name: search_ssl
   - name: session_jscore_passthrough
   - name: .sharding .jscore !.wo_snapshot !.multi_stmt !.multiversion
@@ -700,6 +702,7 @@ buildvariants:
     - name: sasl
     - name: search
     - name: search_auth
+    - name: search_pinned_connections_auth
     - name: search_ssl
     - name: session_jscore_passthrough
     - name: .sharding .jscore !.wo_snapshot !.multi_stmt !.multiversion
@@ -1334,6 +1337,7 @@ buildvariants:
   - name: sasl
   - name: search
   - name: search_auth
+  - name: search_pinned_connections_auth
   - name: search_ssl
   - name: session_jscore_passthrough
   - name: .sharding .jscore !.wo_snapshot !.multi_stmt
@@ -1439,6 +1443,7 @@ buildvariants:
   - name: sasl
   - name: search
   - name: search_auth
+  - name: search_pinned_connections_auth
   - name: search_ssl
   - name: session_jscore_passthrough
   - name: .sharding .jscore !.wo_snapshot !.multi_stmt !.no_debug_mode
@@ -1575,6 +1580,7 @@ buildvariants:
   - name: sasl
   - name: search
   - name: search_auth
+  - name: search_pinned_connections_auth
   - name: search_ssl
   - name: session_jscore_passthrough
   - name: .sharding .jscore !.wo_snapshot !.multi_stmt
@@ -1691,6 +1697,7 @@ buildvariants:
   - name: sasl
   - name: search
   - name: search_auth
+  - name: search_pinned_connections_auth
   - name: search_ssl
   - name: secondary_reads_passthrough_gen
   - name: session_jscore_passthrough
@@ -1875,6 +1882,7 @@ buildvariants:
   - name: sasl
   - name: search
   - name: search_auth
+  - name: search_pinned_connections_auth
   - name: search_ssl
   - name: sharding_auth_audit_gen
   - name: .stitch
@@ -2182,6 +2190,7 @@ buildvariants:
   - name: sasl
   - name: search
   - name: search_auth
+  - name: search_pinned_connections_auth
   - name: search_ssl
   - name: session_jscore_passthrough
   - name: .sharding .jscore !.wo_snapshot !.multi_stmt
@@ -2643,6 +2652,7 @@ buildvariants:
   - name: sasl
   - name: search
   - name: search_auth
+  - name: search_pinned_connections_auth
   - name: search_ssl
   - name: session_jscore_passthrough
   # - name: .sharding .jscore !.wo_snapshot !.multi_stmt # Not passing
diff --git a/etc/evergreen_yml_components/definitions.yml b/etc/evergreen_yml_components/definitions.yml
index 812ab0483f8..981fa411cac 100644
--- a/etc/evergreen_yml_components/definitions.yml
+++ b/etc/evergreen_yml_components/definitions.yml
@@ -7835,6 +7835,15 @@ tasks:
       resmoke_jobs_max: 1
 
 - <<: *task_template
+  name: search_pinned_connections_auth
+  tags: []
+  commands:
+  - func: "do setup"
+  - func: "run tests"
+    vars:
+      resmoke_jobs_max: 1
+
+- <<: *task_template
   name: search_ssl
   tags: []
   commands:
diff --git a/etc/evergreen_yml_components/variants/atlas.yml b/etc/evergreen_yml_components/variants/atlas.yml
index 18876f1e7a9..bb1b3b05cc1 100644
--- a/etc/evergreen_yml_components/variants/atlas.yml
+++ b/etc/evergreen_yml_components/variants/atlas.yml
@@ -62,6 +62,7 @@ buildvariants:
   - name: sasl
   - name: search
   - name: search_auth
+  - name: search_pinned_connections_auth
   - name: search_ssl
   - name: sharding_auth_audit_gen
   - name: .stitch
diff --git a/etc/evergreen_yml_components/variants/misc_release.yml b/etc/evergreen_yml_components/variants/misc_release.yml
index 923791e3b78..d4f7e386362 100644
--- a/etc/evergreen_yml_components/variants/misc_release.yml
+++ b/etc/evergreen_yml_components/variants/misc_release.yml
@@ -278,6 +278,7 @@ buildvariants:
   - name: sasl
   - name: search
   - name: search_auth
+  - name: search_pinned_connections_auth
   - name: search_ssl
   - name: session_jscore_passthrough
   - name: .sharding .jscore !.wo_snapshot !.multi_stmt
@@ -433,6 +434,7 @@ buildvariants:
   - name: sasl
   - name: search
   - name: search_auth
+  - name: search_pinned_connections_auth
   - name: search_ssl
   - name: session_jscore_passthrough
   - name: .sharding .jscore !.wo_snapshot !.multi_stmt
@@ -592,6 +594,7 @@ buildvariants:
   - name: sasl
   - name: search
   - name: search_auth
+  - name: search_pinned_connections_auth
   - name: search_ssl
   - name: session_jscore_passthrough
   - name: .sharding .jscore !.wo_snapshot !.multi_stmt
@@ -1122,6 +1125,7 @@ buildvariants:
   - name: sasl
   - name: search
   - name: search_auth
+  - name: search_pinned_connections_auth
   - name: search_ssl
   - name: session_jscore_passthrough
   - name: .sharding .jscore !.wo_snapshot !.multi_stmt
@@ -1259,6 +1263,7 @@ buildvariants:
   - name: sasl
   - name: search
   - name: search_auth
+  - name: search_pinned_connections_auth
   - name: search_ssl
   - name: sharding_auth_audit_gen
   - name: .stitch
@@ -1383,6 +1388,7 @@ buildvariants:
   - name: sasl
   - name: search
   - name: search_auth
+  - name: search_pinned_connections_auth
   - name: search_ssl
   - name: sharding_auth_audit_gen
   - name: .stitch
@@ -1509,6 +1515,7 @@ buildvariants:
   - name: sasl
   - name: search
   - name: search_auth
+  - name: search_pinned_connections_auth
   - name: search_ssl
   - name: sharding_auth_audit_gen
   - name: sharding_auth_gen
diff --git a/src/mongo/executor/network_interface_tl.h b/src/mongo/executor/network_interface_tl.h
index 52bbe07273a..880a8f46e21 100644
--- a/src/mongo/executor/network_interface_tl.h
+++ b/src/mongo/executor/network_interface_tl.h
@@ -395,8 +395,10 @@ private:
     std::unique_ptr<transport::TransportLayer> _ownedTransportLayer;
     transport::ReactorHandle _reactor;
 
-    mutable Mutex _mutex =
-        MONGO_MAKE_LATCH(HierarchicalAcquisitionLevel(3), "NetworkInterfaceTL::_mutex");
+    // TODO SERVER-75830: This Mutex used to be at hierarcichal acquisition level 3. We temporary
+    // removed the level because it is sometimes acquired as part of task-scheduling when
+    // lower-level mutexes (like the ConnectionPool's) are held.
+    mutable Mutex _mutex = MONGO_MAKE_LATCH("NetworkInterfaceTL::_mutex");
     const ConnectionPool::Options _connPoolOpts;
     std::unique_ptr<NetworkConnectionHook> _onConnectHook;
     std::shared_ptr<ConnectionPool> _pool;
diff --git a/src/mongo/executor/pinned_connection_task_executor.cpp b/src/mongo/executor/pinned_connection_task_executor.cpp
index 975a3cbfcdb..eb3083b81d7 100644
--- a/src/mongo/executor/pinned_connection_task_executor.cpp
+++ b/src/mongo/executor/pinned_connection_task_executor.cpp
@@ -80,12 +80,12 @@ public:
         callback({exec, cbHandle, rcb.first, errorResponse});
     }
 
-    // Run callback with the provided success result.
-    static void runCallbackSuccess(stdx::unique_lock<Latch>& lk,
-                                   RequestAndCallback rcb,
-                                   TaskExecutor* exec,
-                                   const StatusWith<RemoteCommandResponse>& result,
-                                   const HostAndPort& targetUsed) {
+    // Run callback with the provided result.
+    static void runCallbackFinished(stdx::unique_lock<Latch>& lk,
+                                    RequestAndCallback rcb,
+                                    TaskExecutor* exec,
+                                    const StatusWith<RemoteCommandResponse>& result,
+                                    boost::optional<HostAndPort> targetUsed) {
         // Convert the result into a RemoteCommandResponse unconditionally.
         RemoteCommandResponse asRcr =
             result.isOK() ? result.getValue() : RemoteCommandResponse(result.getStatus());
@@ -192,8 +192,10 @@ void PinnedConnectionTaskExecutor::_cancel(WithLock, CallbackState* cbState) {
         case CallbackState::State::kRunning: {
             // Cancel the ongoing operation.
             cbState->state = CallbackState::State::kCanceled;
-            auto client = _stream->getClient();
-            client->cancel(cbState->baton);
+            if (_stream) {
+                auto client = _stream->getClient();
+                client->cancel(cbState->baton);
+            }
             break;
         }
         case CallbackState::State::kCanceled:
@@ -216,11 +218,10 @@ void PinnedConnectionTaskExecutor::cancel(const CallbackHandle& cbHandle) {
     return _cancel(std::move(lk), cbState);
 }
 
-ExecutorFuture<void> PinnedConnectionTaskExecutor::_ensureStream(WithLock,
-                                                                 HostAndPort target,
-                                                                 Milliseconds timeout) {
+ExecutorFuture<void> PinnedConnectionTaskExecutor::_ensureStream(
+    WithLock, HostAndPort target, Milliseconds timeout, transport::ConnectSSLMode sslMode) {
     if (!_stream) {
-        auto streamFuture = _net->leaseStream(target, transport::kGlobalSSLMode, timeout);
+        auto streamFuture = _net->leaseStream(target, sslMode, timeout);
         // If the stream is ready, send the RPC immediately by continuing inline.
         if (streamFuture.isReady()) {
             auto stream = std::move(streamFuture).getNoThrow();
@@ -292,7 +293,7 @@ void PinnedConnectionTaskExecutor::_doNetworking(stdx::unique_lock<Latch>&& lk)
     // Set req state to running
     invariant(req.second->state == CallbackState::State::kWaiting);
     req.second->state = CallbackState::State::kRunning;
-    auto streamFut = _ensureStream(lk, req.first.target, req.first.timeout);
+    auto streamFut = _ensureStream(lk, req.first.target, req.first.timeout, req.first.sslMode);
     // Stash the in-progress operation before releasing the lock so we can
     // access it if we're shutdown while it's in-progress.
     _inProgressRequest = req.second;
@@ -308,10 +309,18 @@ void PinnedConnectionTaskExecutor::_doNetworking(stdx::unique_lock<Latch>&& lk)
                 CallbackState::runCallbackCanceled(lk, req, this);
             } else {
                 invariant(state == CallbackState::State::kRunning);
-                invariant(req.second->startedNetworking);
+                // Three possibilities here: we either finished the RPC
+                // successfully, got a local error from the stream after
+                // attempting to start networking, or never were able to acquire a
+                // stream. In any case, we first complete the current request
+                // by invoking it's callback:
                 state = CallbackState::State::kDone;
-                auto target = _stream->getClient()->remote();
-                CallbackState::runCallbackSuccess(lk, req, this, result, target);
+                // Get the target if we successfully acquired a stream.
+                boost::optional<HostAndPort> target = boost::none;
+                if (_stream) {
+                    target = _stream->getClient()->remote();
+                }
+                CallbackState::runCallbackFinished(lk, req, this, result, target);
             }
             // If we used the _stream, update it accordingly.
             if (req.second->startedNetworking) {
@@ -326,6 +335,10 @@ void PinnedConnectionTaskExecutor::_doNetworking(stdx::unique_lock<Latch>&& lk)
                     _shutdown(lk);
                 }
             }
+            // If we weren't able to acquire a stream, shut-down.
+            if (!_stream) {
+                _shutdown(lk);
+            }
             _isDoingNetworking = false;
             if (!_requestQueue.empty()) {
                 return _doNetworking(std::move(lk));
diff --git a/src/mongo/executor/pinned_connection_task_executor.h b/src/mongo/executor/pinned_connection_task_executor.h
index cf960fb0121..9d423931327 100644
--- a/src/mongo/executor/pinned_connection_task_executor.h
+++ b/src/mongo/executor/pinned_connection_task_executor.h
@@ -135,7 +135,10 @@ private:
     // If we already have a _stream when this function is called, ensures the
     // remote is `target` and returns a ready-future. Otherwise asynchronously
     // initailizes _stream and returns a future that resolves once _stream is ready.
-    ExecutorFuture<void> _ensureStream(WithLock, HostAndPort target, Milliseconds timeout);
+    ExecutorFuture<void> _ensureStream(WithLock,
+                                       HostAndPort target,
+                                       Milliseconds timeout,
+                                       transport::ConnectSSLMode sslMode);
 
     // Start processing pending/queued RPCs.
     void _doNetworking(stdx::unique_lock<Latch>&&);
diff --git a/src/mongo/executor/task_executor_cursor.h b/src/mongo/executor/task_executor_cursor.h
index 5da3a68bae5..458e010f976 100644
--- a/src/mongo/executor/task_executor_cursor.h
+++ b/src/mongo/executor/task_executor_cursor.h
@@ -41,6 +41,7 @@
 #include "mongo/db/session/logical_session_id.h"
 #include "mongo/executor/remote_command_request.h"
 #include "mongo/executor/task_executor.h"
+#include "mongo/executor/task_executor_cursor_parameters_gen.h"
 #include "mongo/util/duration.h"
 #include "mongo/util/future.h"
 
@@ -68,7 +69,7 @@ public:
 
     struct Options {
         boost::optional<int64_t> batchSize;
-        bool pinConnection{false};
+        bool pinConnection{gPinTaskExecCursorConns.load()};
         Options() {}
     };
 
@@ -163,16 +164,6 @@ public:
         return _additionalCursors.size();
     }
 
-    /**
-     * Return the callback that this cursor is waiting on. Can be used to block on getting a
-     * response to this request. Can be boost::none.
-     */
-    boost::optional<TaskExecutor::CallbackHandle> getCallbackHandle() {
-        if (_cmdState)
-            return _cmdState->cbHandle;
-        return {};
-    }
-
 private:
     /**
      * Runs a remote command and pipes the output back to this object
diff --git a/src/mongo/executor/task_executor_cursor_test.cpp b/src/mongo/executor/task_executor_cursor_test.cpp
index 02db9ed2967..eaece63fd6c 100644
--- a/src/mongo/executor/task_executor_cursor_test.cpp
+++ b/src/mongo/executor/task_executor_cursor_test.cpp
@@ -568,6 +568,7 @@ public:
 
     TaskExecutorCursor makeTec(RemoteCommandRequest rcr,
                                TaskExecutorCursor::Options&& options = {}) {
+        options.pinConnection = false;
         return TaskExecutorCursor(getExecutorPtr(), rcr, std::move(options));
     }
 };