Revert "SERVER-92565 Remove NITL implementation structures used to facilitate hedging (#27021)" (#27151)

GitOrigin-RevId: b7faf4d14cb9c690918aaf5d62ac29c1ea7157c4

2 files changed, 339 insertions, 189 deletions

diff --git a/src/mongo/executor/network_interface_tl.cpp b/src/mongo/executor/network_interface_tl.cpp
index 974619aa5fa..b4085be759a 100644
--- a/src/mongo/executor/network_interface_tl.cpp
+++ b/src/mongo/executor/network_interface_tl.cpp
@@ -428,14 +428,12 @@ NetworkInterfaceTL::CommandStateBase::CommandStateBase(
     RemoteCommandRequestOnAny request_,
     const TaskExecutor::CallbackHandle& cbHandle_)
     : interface(interface_),
-      request(RemoteCommandRequest(std::move(request_), 0)),
-      requestToSend(request),
+      requestOnAny(std::move(request_)),
       cbHandle(cbHandle_),
       timer(interface->_reactor->makeTimer()),
-      operationKey(request.operationKey) {}
+      operationKey(requestOnAny.operationKey) {}
 
 NetworkInterfaceTL::CommandStateBase::~CommandStateBase() {
-    invariant(!conn);
     interface->_unregisterCommand(cbHandle);
 }
 
@@ -450,6 +448,7 @@ auto NetworkInterfaceTL::CommandState::make(NetworkInterfaceTL* interface,
     auto state = std::make_shared<CommandState>(interface, std::move(request), cbHandle);
     auto [promise, future] = makePromiseFuture<RemoteCommandOnAnyResponse>();
     state->promise = std::move(promise);
+    state->requestManager = std::make_unique<RequestManager>(state.get());
 
     interface->_registerCommand(cbHandle, state);
 
@@ -473,8 +472,7 @@ auto NetworkInterfaceTL::CommandState::make(NetworkInterfaceTL* interface,
     return std::pair(state, std::move(future));
 }
 
-AsyncDBClient* NetworkInterfaceTL::CommandStateBase::getClient(
-    const ConnectionHandle& conn) noexcept {
+AsyncDBClient* NetworkInterfaceTL::RequestState::getClient(const ConnectionHandle& conn) noexcept {
     if (!conn) {
         return nullptr;
     }
@@ -482,42 +480,44 @@ AsyncDBClient* NetworkInterfaceTL::CommandStateBase::getClient(
     return checked_cast<connection_pool_tl::TLConnection*>(conn.get())->client();
 }
 
-void NetworkInterfaceTL::CommandStateBase::setTimer() {
+void NetworkInterfaceTL::CommandStateBase::setTimer(
+    const std::shared_ptr<RequestState>& requestState) {
     auto nowVal = interface->now();
 
     triggerSendRequestNetworkTimeout.executeIf(
         [&](const BSONObj& data) {
             LOGV2(6496503,
                   "triggerSendRequestNetworkTimeout failpoint enabled, timing out request",
-                  "request"_attr = request.cmdObj.toString());
+                  "request"_attr = requestOnAny.cmdObj.toString());
             // Sleep to make sure the elapsed wait time for connection timeout is > 1 millisecond.
             sleepmillis(100);
             deadline = nowVal;
         },
         [&](const BSONObj& data) {
             return data["collectionNS"].valueStringData() ==
-                request.cmdObj.firstElement().valueStringData();
+                requestOnAny.cmdObj.firstElement().valueStringData();
         });
 
-    if (deadline == kNoExpirationDate || !request.enforceLocalTimeout) {
+    if (deadline == kNoExpirationDate || !requestOnAny.enforceLocalTimeout) {
         return;
     }
 
     const auto timeoutCode =
-        request.timeoutCode.get_value_or(ErrorCodes::NetworkInterfaceExceededTimeLimit);
+        requestOnAny.timeoutCode.get_value_or(ErrorCodes::NetworkInterfaceExceededTimeLimit);
 
-    // We don't need to capture an anchor for the CommandStateBase (i.e. this). If the request gets
-    // fulfilled and misses cancelling the timer (i.e. we can't lock the weak_ptr), we just want to
-    // return. Ideally we'd ensure that cancellation could never miss timers, but since they will
-    // eventually fire anyways it's not a huge deal that we don't.
+    // We don't need to capture an anchor for the CommandStateBase (i.e. this) since requestState
+    // owns a full shared_ptr to it. If the request gets fulfilled and misses cancelling the
+    // timer (i.e. we can't lock the weak_ptr), we just want to return. Ideally we'd ensure that
+    // cancellation could never miss timers, but since they will eventually fire anyways it's not a
+    // huge deal that we don't.
     timer->waitUntil(deadline, baton)
-        .getAsync([this, weakState = weak_from_this(), timeoutCode](Status status) {
+        .getAsync([this, timeoutCode, weakReq = std::weak_ptr(requestState)](Status status) {
             if (!status.isOK()) {
                 return;
             }
 
-            auto cmdState = weakState.lock();
-            if (!cmdState) {
+            auto requestState = weakReq.lock();
+            if (!requestState) {
                 return;
             }
 
@@ -525,22 +525,23 @@ void NetworkInterfaceTL::CommandStateBase::setTimer() {
                 return;
             }
 
-            const std::string message = str::stream() << "Request " << request.id << " timed out"
-                                                      << ", deadline was " << deadline.toString()
-                                                      << ", op was " << redact(request.toString());
+            const std::string message = str::stream()
+                << "Request " << requestOnAny.id << " timed out"
+                << ", deadline was " << deadline.toString() << ", op was "
+                << redact(requestOnAny.toString());
 
             LOGV2_DEBUG(22595,
                         2,
                         "Request timed out",
-                        "requestId"_attr = request.id,
+                        "requestId"_attr = requestOnAny.id,
                         "deadline"_attr = deadline,
-                        "request"_attr = request);
+                        "request"_attr = requestOnAny);
             fulfillFinalPromise(StatusWith<RemoteCommandOnAnyResponse>(RemoteCommandOnAnyResponse(
-                request.target, Status(timeoutCode, message), stopwatch.elapsed())));
+                requestState->host, Status(timeoutCode, message), stopwatch.elapsed())));
         });
 }
 
-void NetworkInterfaceTL::CommandStateBase::returnConnection(Status status) noexcept {
+void NetworkInterfaceTL::RequestState::returnConnection(Status status) noexcept {
     invariant(conn);
 
     auto connToReturn = std::exchange(conn, {});
@@ -558,7 +559,7 @@ void NetworkInterfaceTL::CommandStateBase::tryFinish(Status status) noexcept {
     invariant(promiseFulfilling.load());
 
     LOGV2_DEBUG(
-        4646302, 2, "Finished request", "requestId"_attr = request.id, "status"_attr = status);
+        4646302, 2, "Finished request", "requestId"_attr = requestOnAny.id, "status"_attr = status);
 
     // The command has resolved one way or another.
     timer->cancel(baton);
@@ -568,14 +569,17 @@ void NetworkInterfaceTL::CommandStateBase::tryFinish(Status status) noexcept {
         interface->_counters->recordResult(status);
     }
 
-    if (operationKey) {
+    invariant(requestManager);
+    if (operationKey &&
+        !MONGO_unlikely(networkInterfaceShouldNotKillPendingRequests.shouldFail())) {
         // Kill operations for the request that we didn't use to fulfill the promise.
-        killOperation();
+        requestManager->killOperationsForPendingRequests();
     }
 
     if (!status.isOK()) {
-        // Cancel after we issue _killOperations
-        cancel();
+        // We cancel after we issue _killOperations because, if we cancel before, existing
+        // RequestStates may finish and destruct to quickly.
+        requestManager->cancelRequests();
     }
 
     networkInterfaceCommandsFailedWithErrorCode.shouldFail([&](const BSONObj& data) {
@@ -584,7 +588,7 @@ void NetworkInterfaceTL::CommandStateBase::tryFinish(Status status) noexcept {
             return false;
         }
 
-        const std::string requestCmdName = request.cmdObj.firstElement().fieldName();
+        const std::string requestCmdName = requestOnAny.cmdObj.firstElement().fieldName();
         for (auto&& cmdName : data.getObjectField("cmdNames")) {
             if (cmdName.type() == String && cmdName.valueStringData() == requestCmdName) {
                 return true;
@@ -606,16 +610,18 @@ void NetworkInterfaceTL::_unregisterCommand(const TaskExecutor::CallbackHandle&
     }
 }
 
-void NetworkInterfaceTL::CommandStateBase::cancel() noexcept {
-    LOGV2_DEBUG(4646301, 2, "Cancelling request", "requestId"_attr = request.id);
-
+void NetworkInterfaceTL::RequestState::cancel() noexcept {
     auto connToCancel = weakConn.lock();
     if (auto clientPtr = getClient(connToCancel)) {
         // If we have a client, cancel it
-        clientPtr->cancel(baton);
+        clientPtr->cancel(cmdState->baton);
     }
 }
 
+NetworkInterfaceTL::RequestState::~RequestState() {
+    invariant(!conn);
+}
+
 Status NetworkInterfaceTL::startCommand(const TaskExecutor::CallbackHandle& cbHandle,
                                         RemoteCommandRequestOnAny& request,
                                         RemoteCommandCompletionFn&& onFinish,
@@ -636,8 +642,8 @@ Status NetworkInterfaceTL::startCommand(const TaskExecutor::CallbackHandle& cbHa
     auto targetNode = request.target.front();
 
     auto [cmdState, future] = CommandState::make(this, request, cbHandle);
-    if (cmdState->request.timeout != cmdState->request.kNoTimeout) {
-        cmdState->deadline = cmdState->stopwatch.start() + cmdState->request.timeout;
+    if (cmdState->requestOnAny.timeout != cmdState->requestOnAny.kNoTimeout) {
+        cmdState->deadline = cmdState->stopwatch.start() + cmdState->requestOnAny.timeout;
     }
     cmdState->baton = baton;
 
@@ -664,14 +670,15 @@ Status NetworkInterfaceTL::startCommand(const TaskExecutor::CallbackHandle& cbHa
                 numConnectionNetworkTimeouts.increment(1);
                 timeSpentWaitingBeforeConnectionTimeoutMillis.increment(
                     durationCount<Milliseconds>(cmdState->connTimeoutWaitTime));
-                auto timeoutCode = cmdState->request.timeoutCode;
-                if (timeoutCode && cmdState->connTimeoutWaitTime >= cmdState->request.timeout) {
+                auto timeoutCode = cmdState->requestOnAny.timeoutCode;
+                if (timeoutCode &&
+                    cmdState->connTimeoutWaitTime >= cmdState->requestOnAny.timeout) {
                     rs.status = Status(*timeoutCode, rs.status.reason());
                 }
                 if (gEnableDetailedConnectionHealthMetricLogLines.load()) {
                     LOGV2(6496500,
                           "Operation timed out while waiting to acquire connection",
-                          "requestId"_attr = cmdState->request.id,
+                          "requestId"_attr = cmdState->requestOnAny.id,
                           "duration"_attr = cmdState->connTimeoutWaitTime);
                 }
             }
@@ -679,7 +686,7 @@ Status NetworkInterfaceTL::startCommand(const TaskExecutor::CallbackHandle& cbHa
             LOGV2_DEBUG(22597,
                         2,
                         "Request finished with response",
-                        "requestId"_attr = cmdState->request.id,
+                        "requestId"_attr = cmdState->requestOnAny.id,
                         "isOK"_attr = rs.isOK(),
                         "response"_attr =
                             redact(rs.isOK() ? rs.data.toString() : rs.status.toString()));
@@ -695,11 +702,11 @@ Status NetworkInterfaceTL::startCommand(const TaskExecutor::CallbackHandle& cbHa
     auto connFuture = _pool->get(targetNode, request.sslMode, request.timeout);
 
     if (connFuture.isReady()) {
-        cmdState->trySend(std::move(connFuture).getNoThrow());
+        cmdState->requestManager->trySend(std::move(connFuture).getNoThrow());
     } else {
         // Otherwise, schedule the request.
         std::move(connFuture).thenRunOn(_reactor).getAsync([cmdState = cmdState](auto swConn) {
-            cmdState->trySend(std::move(swConn));
+            cmdState->requestManager->trySend(std::move(swConn));
         });
     }
 
@@ -720,17 +727,19 @@ void NetworkInterfaceTL::testEgress(const HostAndPort& hostAndPort,
     }
 }
 
-Future<RemoteCommandResponse> NetworkInterfaceTL::CommandState::sendRequest() {
-    return makeReadyFutureWith([this] {
-               setTimer();
+Future<RemoteCommandResponse> NetworkInterfaceTL::CommandState::sendRequest(
+    std::shared_ptr<RequestState> requestState) {
+    return makeReadyFutureWith([this, requestState] {
+               setTimer(requestState);
                const auto connAcquiredTimer =
-                   checked_cast<connection_pool_tl::TLConnection*>(conn.get())
+                   checked_cast<connection_pool_tl::TLConnection*>(requestState->conn.get())
                        ->getConnAcquiredTimer();
-               return getClient(conn)->runCommandRequest(
-                   requestToSend, baton, std::move(connAcquiredTimer));
+               return RequestState::getClient(requestState->conn)
+                   ->runCommandRequest(*requestState->request, baton, std::move(connAcquiredTimer));
            })
-        .then([this](RemoteCommandResponse response) {
-            uassertStatusOK(doMetadataHook(RemoteCommandOnAnyResponse(request.target, response)));
+        .then([this, requestState](RemoteCommandResponse response) {
+            uassertStatusOK(
+                doMetadataHook(RemoteCommandOnAnyResponse(requestState->host, response)));
             return response;
         });
 }
@@ -750,58 +759,107 @@ void NetworkInterfaceTL::CommandState::fulfillFinalPromise(
     promiseFulfilled.set();
 }
 
+NetworkInterfaceTL::RequestManager::RequestManager(CommandStateBase* cmdState_)
+    : cmdState{cmdState_} {}
 
-void NetworkInterfaceTL::CommandStateBase::killOperation() {
+void NetworkInterfaceTL::RequestManager::cancelRequests() {
+    std::shared_ptr<RequestState> requestToCancel;
     {
         stdx::lock_guard<Latch> lk(mutex);
-        if (!conn) {
+
+        // Once we've set isLocked to true, no more requests will be created for this manager.
+        // Thus, only those that have been sent already need to be cancelled.
+        isLocked = true;
+        if (!isSent) {
+            return;
+        }
+        requestToCancel = request.request.lock();
+        if (!requestToCancel) {
             return;
         }
     }
 
-    if (auto status = interface->_killOperation(this); !status.isOK()) {
+    LOGV2_DEBUG(4646301, 2, "Cancelling request", "requestId"_attr = cmdState->requestOnAny.id);
+    requestToCancel->cancel();
+    requestToCancel.reset();
+}
+
+void NetworkInterfaceTL::RequestManager::killOperationsForPendingRequests() {
+    // Send `_killOperation` out of band to the target with the initialized request (which
+    // acquired a connection), regardless of its state so long as it's not used to fulfill the
+    // operation.
+    {
+        stdx::lock_guard<Latch> lk(mutex);
+        isLocked = true;
+        if (!isSent) {
+            return;
+        }
+        auto& context = request;
+        invariant(context.initialized);
+        if (auto requestState = context.request.lock();
+            requestState && requestState->fulfilledPromise) {
+            return;
+        }
+    }
+
+    if (auto status = cmdState->interface->_killOperation(cmdState); !status.isOK()) {
         LOGV2_DEBUG(4664810, 2, "Failed to send remote _killOperations", "error"_attr = status);
     }
 }
 
-void NetworkInterfaceTL::CommandStateBase::trySend(
+void NetworkInterfaceTL::RequestManager::trySend(
     StatusWith<ConnectionPool::ConnectionHandle> swConn) noexcept {
     forceConnectionNetworkTimeout.executeIf(
         [&](const BSONObj& data) {
             LOGV2(6496502,
                   "forceConnectionNetworkTimeout failpoint enabled, timing out request",
-                  "request"_attr = request.cmdObj.toString());
+                  "request"_attr = cmdState->requestOnAny.cmdObj.toString());
             swConn =
                 Status(ErrorCodes::PooledConnectionAcquisitionExceededTimeLimit,
                        "PooledConnectionAcquisitionExceededTimeLimit triggered via fail point.");
         },
         [&](const BSONObj& data) {
             return data["collectionNS"].valueStringData() ==
-                request.cmdObj.firstElement().valueStringData();
+                cmdState->requestOnAny.cmdObj.firstElement().valueStringData();
         });
 
     // Our connection wasn't any good
     if (!swConn.isOK()) {
         {
             stdx::lock_guard<Latch> lk(mutex);
-            invariant(!conn);
+
+            auto currentConnsResolved = ++connsResolved;
+            if (currentConnsResolved < cmdState->maxPossibleConns()) {
+                // If we still have connections outstanding, we don't need to fail the promise.
+                return;
+            }
+
+            if (isSent) {
+                // If a request has been sent, we shouldn't fail the promise.
+                return;
+            }
+
+            if (isLocked) {
+                // If we've finished, obviously we don't need to fail the promise.
+                return;
+            }
         }
 
         // We're the last one, set the promise if it hasn't already been set via cancel or timeout
-        if (!promiseFulfilling.swap(true)) {
+        if (!cmdState->promiseFulfilling.swap(true)) {
             if (swConn.getStatus() == ErrorCodes::PooledConnectionAcquisitionExceededTimeLimit) {
-                connTimeoutWaitTime = stopwatch.elapsed();
+                cmdState->connTimeoutWaitTime = cmdState->stopwatch.elapsed();
             }
 
-            auto& reactor = interface->_reactor;
-            boost::optional<HostAndPort> target = request.target;
+            auto& reactor = cmdState->interface->_reactor;
+            boost::optional<HostAndPort> target = cmdState->requestOnAny.target.front();
             if (reactor->onReactorThread()) {
-                fulfillFinalPromise(StatusWith<RemoteCommandOnAnyResponse>(
+                cmdState->fulfillFinalPromise(StatusWith<RemoteCommandOnAnyResponse>(
                     RemoteCommandOnAnyResponse(target, std::move(swConn.getStatus()))));
             } else {
                 ExecutorFuture<void>(reactor, swConn.getStatus())
-                    .getAsync([this, anchor = shared_from_this(), target](Status status) {
-                        fulfillFinalPromise(StatusWith<RemoteCommandOnAnyResponse>(
+                    .getAsync([this, anchor = cmdState->shared_from_this(), target](Status status) {
+                        cmdState->fulfillFinalPromise(StatusWith<RemoteCommandOnAnyResponse>(
                             RemoteCommandOnAnyResponse(target, std::move(status))));
                     });
             }
@@ -811,111 +869,137 @@ void NetworkInterfaceTL::CommandStateBase::trySend(
 
     checked_cast<connection_pool_tl::TLConnection*>(swConn.getValue().get())
         ->startConnAcquiredTimer();
+    std::shared_ptr<RequestState> requestState;
 
     bool logSetMaxTimeMS = false;
+    RemoteCommandRequestImpl<HostAndPort>* requestImpl;
     {
         stdx::lock_guard<Latch> lk(mutex);
 
-        invariant(!conn);
+        // Increment the number of conns we were able to resolve.
+        ++connsResolved;
+
+        if (isSent || isLocked) {
+            // Our command has already been satisfied or we have already sent out
+            // the request.
+            swConn.getValue()->indicateSuccess();
+            return;
+        }
+
+        isSent = true;
+
+        requestState = std::make_shared<RequestState>(this, cmdState->shared_from_this());
 
         // Set conn/weakConn+request under the lock so they will always be observed during cancel.
-        conn = std::move(swConn.getValue());
-        weakConn = conn;
+        requestState->conn = std::move(swConn.getValue());
+        requestState->weakConn = requestState->conn;
+
+        requestState->request = RemoteCommandRequest(cmdState->requestOnAny, 0);
+        requestState->host = requestState->request->target;
 
-        if (interface->_svcCtx && request.timeout != RemoteCommandRequest::kNoTimeout &&
-            WireSpec::getWireSpec(interface->_svcCtx).get()->isInternalClient) {
+        requestImpl = &requestState->request.value();
+
+        if (cmdState->interface->_svcCtx &&
+            requestImpl->timeout != RemoteCommandRequest::kNoTimeout &&
+            WireSpec::getWireSpec(cmdState->interface->_svcCtx).get()->isInternalClient) {
             logSetMaxTimeMS = true;
             BSONObjBuilder updatedCmdBuilder;
-            updatedCmdBuilder.appendElements(requestToSend.cmdObj);
-            updatedCmdBuilder.append("maxTimeMSOpOnly", requestToSend.timeout.count());
-            requestToSend.cmdObj = updatedCmdBuilder.obj();
+            updatedCmdBuilder.appendElements(requestImpl->cmdObj);
+            updatedCmdBuilder.append("maxTimeMSOpOnly", requestImpl->timeout.count());
+            requestImpl->cmdObj = updatedCmdBuilder.obj();
         }
+
+        auto& context = request;
+        context.initialized = true;
+        context.request = requestState;
     }
 
     LOGV2_DEBUG(4646300,
                 2,
                 "Sending request",
-                "requestId"_attr = request.id,
-                "target"_attr = request.target);
+                "requestId"_attr = cmdState->requestOnAny.id,
+                "target"_attr = cmdState->requestOnAny.target.front());
 
     if (logSetMaxTimeMS) {
         LOGV2_DEBUG(4924402,
                     2,
                     "Set maxTimeMSOpOnly for request",
-                    "maxTimeMSOpOnly"_attr = request.timeout,
-                    "requestId"_attr = request.id,
-                    "target"_attr = request.target);
+                    "maxTimeMSOpOnly"_attr = requestImpl->timeout,
+                    "requestId"_attr = cmdState->requestOnAny.id,
+                    "target"_attr = cmdState->requestOnAny.target.front());
     }
 
     LOGV2_DEBUG(4630601,
                 2,
                 "Request acquired a connection",
-                "requestId"_attr = request.id,
-                "target"_attr = request.target);
+                "requestId"_attr = requestState->request->id,
+                "target"_attr = requestState->request->target);
 
     networkInterfaceHangCommandsAfterAcquireConn.pauseWhileSet();
 
     // An attempt to avoid sending a request after its command has been canceled or already executed
     // using another connection. Just a best effort to mitigate unnecessary resource consumption if
     // possible, and allow deterministic cancellation of requests in testing.
-    if (promiseFulfilling.load()) {
+    if (cmdState->promiseFulfilling.load()) {
         LOGV2_DEBUG(5813901,
                     2,
                     "Skipping request as it has already been fulfilled or canceled",
-                    "requestId"_attr = request.id,
-                    "target"_attr = request.target);
-        returnConnection(Status::OK());
+                    "requestId"_attr = requestState->request->id,
+                    "target"_attr = requestState->request->target);
+        requestState->returnConnection(Status::OK());
         return;
     }
 
-    if (auto counters = interface->_counters) {
+    if (auto counters = cmdState->interface->_counters) {
         counters->recordSent();
     }
 
     if (waitForShutdownBeforeSendRequest.shouldFail()) {
-        invariant(!interface->onNetworkThread());
-        promiseFulfilled.get();
+        invariant(!cmdState->interface->onNetworkThread());
+        cmdState->promiseFulfilled.get();
     }
 
-    resolve(sendRequest());
+    requestState->resolve(cmdState->sendRequest(requestState));
 }
 
-void NetworkInterfaceTL::CommandStateBase::resolve(Future<RemoteCommandResponse> future) noexcept {
+void NetworkInterfaceTL::RequestState::resolve(Future<RemoteCommandResponse> future) noexcept {
+    auto& reactor = interface()->_reactor;
+    auto& baton = cmdState->baton;
+
     // Convert the RemoteCommandResponse to a RemoteCommandOnAnyResponse and wrap any error
-    auto anyFuture = std::move(future)
-                         .then([this, anchor = shared_from_this()](RemoteCommandResponse response) {
-                             // The RCRq ran successfully, wrap the result with the host in question
-                             return RemoteCommandOnAnyResponse(request.target, std::move(response));
-                         })
-                         .onError([this, anchor = shared_from_this()](Status error) {
-                             // The RCRq failed, wrap the error into a RCRsp with the host and
-                             // duration
-                             return RemoteCommandOnAnyResponse(
-                                 request.target, std::move(error), stopwatch.elapsed());
-                         });
-
-    std::move(anyFuture)
-        .thenRunOn(
-            makeGuaranteedExecutor(baton, interface->_reactor))  // Switch to the baton/reactor.
+    auto anyFuture =
+        std::move(future)
+            .then([this, anchor = shared_from_this()](RemoteCommandResponse response) {
+                // The RCRq ran successfully, wrap the result with the host in question
+                return RemoteCommandOnAnyResponse(host, std::move(response));
+            })
+            .onError([this, anchor = shared_from_this()](Status error) {
+                // The RCRq failed, wrap the error into a RCRsp with the host and duration
+                return RemoteCommandOnAnyResponse(host, std::move(error), stopwatch.elapsed());
+            });
+
+    std::move(anyFuture)                                    //
+        .thenRunOn(makeGuaranteedExecutor(baton, reactor))  // Switch to the baton/reactor.
         .getAsync([this, anchor = shared_from_this()](auto swr) noexcept {
             auto response = uassertStatusOK(swr);
             auto status = response.status;
 
             returnConnection(status);
 
-            if (promiseFulfilling.swap(true)) {
+            if (cmdState->promiseFulfilling.swap(true)) {
                 LOGV2_DEBUG(4754301,
                             2,
-                            "Skipping the response because the operation was cancelled",
-                            "requestId"_attr = request.id,
-                            "target"_attr = request.target,
+                            "Skipping the response because it was already received from other node",
+                            "requestId"_attr = request->id,
+                            "target"_attr = request->target,
                             "status"_attr = response.status,
                             "response"_attr = redact(response.data));
 
                 return;
             }
 
-            fulfillFinalPromise(std::move(response));
+            fulfilledPromise = true;
+            cmdState->fulfillFinalPromise(std::move(response));
         });
 }
 
@@ -935,6 +1019,7 @@ auto NetworkInterfaceTL::ExhaustCommandState::make(NetworkInterfaceTL* interface
     auto state = std::make_shared<ExhaustCommandState>(
         interface, std::move(request), cbHandle, std::move(onReply));
     auto [promise, future] = makePromiseFuture<void>();
+    state->requestManager = std::make_unique<RequestManager>(state.get());
     state->promise = std::move(promise);
 
     interface->_registerCommand(cbHandle, state);
@@ -956,16 +1041,17 @@ auto NetworkInterfaceTL::ExhaustCommandState::make(NetworkInterfaceTL* interface
     return state;
 }
 
-Future<RemoteCommandResponse> NetworkInterfaceTL::ExhaustCommandState::sendRequest() try {
+Future<RemoteCommandResponse> NetworkInterfaceTL::ExhaustCommandState::sendRequest(
+    std::shared_ptr<RequestState> requestState) try {
     auto [promise, future] = makePromiseFuture<RemoteCommandResponse>();
     finalResponsePromise = std::move(promise);
 
-    setTimer();
-    getClient(conn)
-        ->beginExhaustCommandRequest(request, baton)
-        .thenRunOn(interface->_reactor)
-        .getAsync([this](StatusWith<RemoteCommandResponse> swResponse) mutable {
-            continueExhaustRequest(swResponse);
+    setTimer(requestState);
+    requestState->getClient(requestState->conn)
+        ->beginExhaustCommandRequest(*requestState->request, baton)
+        .thenRunOn(requestState->interface()->_reactor)
+        .getAsync([this, requestState](StatusWith<RemoteCommandResponse> swResponse) mutable {
+            continueExhaustRequest(std::move(requestState), swResponse);
         });
     return std::move(future).then([this](const auto& finalResponse) { return finalResponse; });
 } catch (const DBException& ex) {
@@ -986,7 +1072,7 @@ void NetworkInterfaceTL::ExhaustCommandState::fulfillFinalPromise(
 }
 
 void NetworkInterfaceTL::ExhaustCommandState::continueExhaustRequest(
-    StatusWith<RemoteCommandResponse> swResponse) {
+    std::shared_ptr<RequestState> requestState, StatusWith<RemoteCommandResponse> swResponse) {
     RemoteCommandResponse response;
     if (!swResponse.isOK()) {
         response = RemoteCommandResponse(std::move(swResponse.getStatus()));
@@ -994,12 +1080,13 @@ void NetworkInterfaceTL::ExhaustCommandState::continueExhaustRequest(
         response = std::move(swResponse.getValue());
     }
 
-    if (interface->inShutdown() || ErrorCodes::isCancellationError(response.status)) {
+    if (requestState->interface()->inShutdown() ||
+        ErrorCodes::isCancellationError(response.status)) {
         finalResponsePromise.emplaceValue(response);
         return;
     }
 
-    auto onAnyResponse = RemoteCommandOnAnyResponse(request.target, response);
+    auto onAnyResponse = RemoteCommandOnAnyResponse(requestState->host, response);
     if (Status metadataHookStatus = doMetadataHook(onAnyResponse); !metadataHookStatus.isOK()) {
         finalResponsePromise.setError(metadataHookStatus);
         return;
@@ -1024,16 +1111,16 @@ void NetworkInterfaceTL::ExhaustCommandState::continueExhaustRequest(
         stopwatch.restart();
     }
     if (deadline != kNoExpirationDate) {
-        deadline = stopwatch.start() + request.timeout;
+        deadline = stopwatch.start() + requestOnAny.timeout;
     }
 
-    setTimer();
+    setTimer(requestState);
 
-    getClient(conn)
+    requestState->getClient(requestState->conn)
         ->awaitExhaustCommand(baton)
-        .thenRunOn(interface->_reactor)
-        .getAsync([this](StatusWith<RemoteCommandResponse> swResponse) mutable {
-            continueExhaustRequest(swResponse);
+        .thenRunOn(requestState->interface()->_reactor)
+        .getAsync([this, requestState](StatusWith<RemoteCommandResponse> swResponse) mutable {
+            continueExhaustRequest(std::move(requestState), swResponse);
         });
 }
 
@@ -1054,20 +1141,21 @@ Status NetworkInterfaceTL::startExhaustCommand(const TaskExecutor::CallbackHandl
     }
 
     auto cmdState = ExhaustCommandState::make(this, request, cbHandle, std::move(onReply), baton);
-    if (cmdState->request.timeout != cmdState->request.kNoTimeout) {
-        cmdState->deadline = cmdState->stopwatch.start() + cmdState->request.timeout;
+    if (cmdState->requestOnAny.timeout != cmdState->requestOnAny.kNoTimeout) {
+        cmdState->deadline = cmdState->stopwatch.start() + cmdState->requestOnAny.timeout;
     }
     cmdState->baton = baton;
+    cmdState->requestManager = std::make_unique<RequestManager>(cmdState.get());
 
     // Attempt to get a connection to the target host
     auto connFuture = _pool->get(request.target.front(), request.sslMode, request.timeout);
 
     if (connFuture.isReady()) {
-        cmdState->trySend(std::move(connFuture).getNoThrow());
+        cmdState->requestManager->trySend(std::move(connFuture).getNoThrow());
     } else {
         // For every connection future we didn't have immediately ready, schedule
         std::move(connFuture).thenRunOn(_reactor).getAsync([cmdState](auto swConn) {
-            cmdState->trySend(std::move(swConn));
+            cmdState->requestManager->trySend(std::move(swConn));
         });
     }
 
@@ -1095,18 +1183,18 @@ void NetworkInterfaceTL::cancelCommand(const TaskExecutor::CallbackHandle& cbHan
         LOGV2_DEBUG(22599,
                     2,
                     "Canceling operation for request",
-                    "request"_attr = redact(cmdStateToCancel->request.cmdObj));
-        cmdStateToCancel->fulfillFinalPromise({ErrorCodes::CallbackCanceled,
-                                               str::stream()
-                                                   << "Command canceled; original request was: "
-                                                   << redact(cmdStateToCancel->request.cmdObj)});
+                    "request"_attr = redact(cmdStateToCancel->requestOnAny.toString()));
+        cmdStateToCancel->fulfillFinalPromise(
+            {ErrorCodes::CallbackCanceled,
+             str::stream() << "Command canceled; original request was: "
+                           << redact(cmdStateToCancel->requestOnAny.toString())});
     }
 }
 
 Status NetworkInterfaceTL::_killOperation(CommandStateBase* cmdStateToKill) try {
     auto [target, sslMode] = [&] {
-        const auto& request = cmdStateToKill->request;
-        return std::make_pair(request.target, request.sslMode);
+        const auto& request = cmdStateToKill->requestOnAny;
+        return std::make_pair(request.target.front(), request.sslMode);
     }();
 
     auto operationKey = cmdStateToKill->operationKey.value();
@@ -1146,7 +1234,7 @@ Status NetworkInterfaceTL::_killOperation(CommandStateBase* cmdStateToKill) try
     std::move(connFuture)
         .thenRunOn(_reactor)
         .getAsync([this, killOpCmdState = killOpCmdState](auto swConn) {
-            killOpCmdState->trySend(std::move(swConn));
+            killOpCmdState->requestManager->trySend(std::move(swConn));
         });
     return Status::OK();
 } catch (const DBException& ex) {
diff --git a/src/mongo/executor/network_interface_tl.h b/src/mongo/executor/network_interface_tl.h
index 942c581a8fd..88d22debaa5 100644
--- a/src/mongo/executor/network_interface_tl.h
+++ b/src/mongo/executor/network_interface_tl.h
@@ -163,10 +163,16 @@ public:
         Milliseconds timeout) override;
 
 private:
+    struct RequestState;
+    struct RequestManager;
+
     /**
-     * For an RPC, an instance of `CommandState` is created to capture the state of the
-     * remote command. As part of running a remote command, `NITL` sends out a request
-     * to the specified target.
+     * For each logical RPC, an instance of `CommandState` is created to capture the state of the
+     * remote command. As part of running a remote command, `NITL` sends out one or more requests
+     * to the specified targets, and `RequestState` represents the state of each request.
+     * `CommandState` owns a `RequestManager` that tracks individual requests. For each request sent
+     * over the wire, `RequestManager` creates a `Context` that holds a weak pointer to the
+     * `Request`, as well as the index of the target.
      */
 
     struct CommandStateBase : public std::enable_shared_from_this<CommandStateBase> {
@@ -175,10 +181,16 @@ private:
                          const TaskExecutor::CallbackHandle& cbHandle_);
         virtual ~CommandStateBase();
 
-        using ConnectionHandle = std::shared_ptr<ConnectionPool::ConnectionHandle::element_type>;
-        using WeakConnectionHandle = std::weak_ptr<ConnectionPool::ConnectionHandle::element_type>;
+        /**
+         * Use the current RequestState to send out a command request.
+         */
+        virtual Future<RemoteCommandResponse> sendRequest(
+            std::shared_ptr<RequestState> requestState) = 0;
 
-        virtual Future<RemoteCommandResponse> sendRequest() = 0;
+        /**
+         * Set a timer to fulfill the promise with a timeout error.
+         */
+        void setTimer(const std::shared_ptr<RequestState>& requestState);
 
         /**
          * Fulfill the promise with the response.
@@ -195,50 +207,21 @@ private:
         void tryFinish(Status status) noexcept;
 
         /**
-         * Return the current connection to the pool and unset it locally.
-         *
-         * This must be called from the networking thread (i.e. the reactor).
-         */
-        void returnConnection(Status status) noexcept;
-
-        void trySend(StatusWith<ConnectionPool::ConnectionHandle> swConn) noexcept;
-
-        void killOperation();
-
-        /**
-         * Set a timer to fulfill the promise with a timeout error.
-         */
-        virtual void setTimer();
-
-        /**
-         * Resolve an eventual response
-         */
-        void resolve(Future<RemoteCommandResponse> future) noexcept;
-
-        /**
-         * Return the client for a given connection
-         */
-        static AsyncDBClient* getClient(const ConnectionHandle& conn) noexcept;
-
-        /**
-         * Cancel the current client operation or do nothing if there is no client.
-         */
-        void cancel() noexcept;
-
-        /**
          * Run the NetworkInterface's MetadataHook on a given request if this Command isn't already
          * finished.
          */
         Status doMetadataHook(const RemoteCommandOnAnyResponse& response);
 
-        NetworkInterfaceTL* interface;
-
-        // Original request as received from the caller.
-        const RemoteCommandRequest request;
+        /**
+         * Return the most connections we expect to be able to acquire.
+         */
+        size_t maxPossibleConns() const noexcept {
+            return requestOnAny.target.size();
+        }
 
-        // Modified request to emit on the wire.
-        RemoteCommandRequest requestToSend;
+        NetworkInterfaceTL* interface;
 
+        RemoteCommandRequestOnAny requestOnAny;
         TaskExecutor::CallbackHandle cbHandle;
         Date_t deadline = kNoExpirationDate;
 
@@ -247,6 +230,8 @@ private:
         BatonHandle baton;
         std::unique_ptr<transport::ReactorTimer> timer;
 
+        std::unique_ptr<RequestManager> requestManager;
+
         // The thread that sets this bit must subsequently call fulfillFinalPromise() exactly once.
         // Once it is set, no other thread may call fulfillFinalPromise().
         Atomic<bool> promiseFulfilling{false};
@@ -258,12 +243,6 @@ private:
 
         // Total time spent waiting for connections that eventually time out.
         Milliseconds connTimeoutWaitTime{0};
-
-        ConnectionHandle conn;
-        WeakConnectionHandle weakConn;
-
-        // Synchronizes requestToSend, conn, and weakConn.
-        Mutex mutex = MONGO_MAKE_LATCH("NetworkInterfaceTL::CommandStateBase::mutex");
     };
 
     struct CommandState final : public CommandStateBase {
@@ -278,7 +257,8 @@ private:
                          RemoteCommandRequestOnAny request,
                          const TaskExecutor::CallbackHandle& cbHandle);
 
-        Future<RemoteCommandResponse> sendRequest() override;
+        Future<RemoteCommandResponse> sendRequest(
+            std::shared_ptr<RequestState> requestState) override;
 
         void fulfillFinalPromise(StatusWith<RemoteCommandOnAnyResponse> response) override;
 
@@ -300,11 +280,13 @@ private:
                          RemoteCommandOnReplyFn&& onReply,
                          const BatonHandle& baton);
 
-        Future<RemoteCommandResponse> sendRequest() override;
+        Future<RemoteCommandResponse> sendRequest(
+            std::shared_ptr<RequestState> requestState) override;
 
         void fulfillFinalPromise(StatusWith<RemoteCommandOnAnyResponse> response) override;
 
-        void continueExhaustRequest(StatusWith<RemoteCommandResponse> swResponse);
+        void continueExhaustRequest(std::shared_ptr<RequestState> requestState,
+                                    StatusWith<RemoteCommandResponse> swResponse);
 
         // Protects against race between reactor thread restarting stopwatch during exhaust
         // request and main thread reading stopwatch elapsed time during shutdown.
@@ -315,6 +297,86 @@ private:
         RemoteCommandOnReplyFn onReplyFn;
     };
 
+    struct RequestManager {
+        RequestManager(CommandStateBase* cmdState);
+
+        void trySend(StatusWith<ConnectionPool::ConnectionHandle> swConn) noexcept;
+        void cancelRequests();
+        void killOperationsForPendingRequests();
+
+        CommandStateBase* cmdState;
+
+        /**
+         * Holds context for individual requests, and is only valid if initialized.
+         */
+        struct Context {
+            bool initialized = false;
+            std::weak_ptr<RequestState> request;
+        };
+        Context request;
+
+        Mutex mutex = MONGO_MAKE_LATCH("NetworkInterfaceTL::RequestManager::mutex");
+
+        // Number of connections we've resolved.
+        size_t connsResolved{0};
+
+        // Set to true after we have sent the request.
+        bool isSent{false};
+
+        // Set to true when the command finishes or is canceled to block remaining requests.
+        bool isLocked{false};
+    };
+
+    struct RequestState final : public std::enable_shared_from_this<RequestState> {
+        using ConnectionHandle = std::shared_ptr<ConnectionPool::ConnectionHandle::element_type>;
+        using WeakConnectionHandle = std::weak_ptr<ConnectionPool::ConnectionHandle::element_type>;
+        RequestState(RequestManager* mgr, std::shared_ptr<CommandStateBase> cmdState_)
+            : cmdState{std::move(cmdState_)}, requestManager(mgr) {}
+
+        ~RequestState();
+
+        /**
+         * Return the client for a given connection
+         */
+        static AsyncDBClient* getClient(const ConnectionHandle& conn) noexcept;
+
+        /**
+         * Cancel the current client operation or do nothing if there is no client.
+         */
+        void cancel() noexcept;
+
+        /**
+         * Return the current connection to the pool and unset it locally.
+         *
+         * This must be called from the networking thread (i.e. the reactor).
+         */
+        void returnConnection(Status status) noexcept;
+
+        /**
+         * Resolve an eventual response
+         */
+        void resolve(Future<RemoteCommandResponse> future) noexcept;
+
+        NetworkInterfaceTL* interface() noexcept {
+            return cmdState->interface;
+        }
+
+        std::shared_ptr<CommandStateBase> cmdState;
+
+        ClockSource::StopWatch stopwatch;
+
+        RequestManager* const requestManager{nullptr};
+
+        boost::optional<RemoteCommandRequest> request;
+        HostAndPort host;
+        ConnectionHandle conn;
+        WeakConnectionHandle weakConn;
+
+        // Set to true if the response to the request is used to fulfill the command's
+        // promise.
+        bool fulfilledPromise{false};
+    };
+
     struct AlarmState {
         AlarmState(Date_t when_,
                    TaskExecutor::CallbackHandle cbHandle_,