diff --git a/src/mongo/client/dbclientcursor.cpp b/src/mongo/client/dbclientcursor.cpp
index 23a9b12a79..c96fb18d2e 100644
--- a/src/mongo/client/dbclientcursor.cpp
+++ b/src/mongo/client/dbclientcursor.cpp
@@ -117,7 +117,7 @@ Message DBClientCursor::_assembleInit() {
                                                 nToSkip,
                                                 nextBatchSize(),
                                                 opts);
-        if (qr.isOK() && !qr.getValue()->isExplain() && !qr.getValue()->isExhaust()) {
+        if (qr.isOK() && !qr.getValue()->isExplain()) {
             BSONObj cmd = qr.getValue()->asFindCommand();
             if (auto readPref = query["$readPreference"]) {
                 // QueryRequest doesn't handle $readPreference.
@@ -144,7 +144,11 @@ Message DBClientCursor::_assembleGetMore() {
                                   boost::none,   // awaitDataTimeout
                                   boost::none,   // term
                                   boost::none);  // lastKnownCommittedOptime
-        return assembleCommandRequest(_client, ns.db(), opts, gmr.toBSON());
+        auto msg = assembleCommandRequest(_client, ns.db(), opts, gmr.toBSON());
+        if (opts & QueryOption_Exhaust) {  // TODO version check
+            OpMsg::setFlag(&msg, OpMsg::kExhaustSupported);
+        }
+        return msg;
     } else {
         // Assemble a legacy getMore request.
         return makeGetMoreMessage(ns.ns(), cursorId, nextBatchSize(), opts);
@@ -204,7 +208,7 @@ bool DBClientCursor::initLazyFinish(bool& retry) {
 }
 
 void DBClientCursor::requestMore() {
-    if (opts & QueryOption_Exhaust) {
+    if ((opts & QueryOption_Exhaust) && (!_useFindCommand || _connectionHasPendingReplies)) {
         return exhaustReceiveMore();
     }
 
@@ -255,6 +259,10 @@ BSONObj DBClientCursor::commandDataReceived(const Message& reply) {
     int op = reply.operation();
     invariant(op == opReply || op == dbCommandReply || op == dbMsg);
 
+    _connectionHasPendingReplies = (op == dbMsg && OpMsg::isFlagSet(reply, OpMsg::kMoreToCome));
+    if (_connectionHasPendingReplies)
+        _lastRequestId = reply.header().getId();
+
     auto commandReply = _client->parseCommandReplyMessage(_client->getServerAddress(), reply);
     auto commandStatus = getStatusFromCommandResult(commandReply->getCommandReply());
 
diff --git a/src/mongo/rpc/op_msg.h b/src/mongo/rpc/op_msg.h
index ed7f1993d8..a12b375fbb 100644
--- a/src/mongo/rpc/op_msg.h
+++ b/src/mongo/rpc/op_msg.h
@@ -46,6 +46,7 @@ struct OpMsg {
 
     static constexpr uint32_t kChecksumPresent = 1 << 0;
     static constexpr uint32_t kMoreToCome = 1 << 1;
+    static constexpr uint32_t kExhaustSupported = 1 << 16;
 
     /**
      * Returns the unvalidated flags for the given message if it is an OP_MSG message.
diff --git a/src/mongo/rpc/op_msg_integration_test.cpp b/src/mongo/rpc/op_msg_integration_test.cpp
index 6dd9fec7f2..81583dd30e 100644
--- a/src/mongo/rpc/op_msg_integration_test.cpp
+++ b/src/mongo/rpc/op_msg_integration_test.cpp
@@ -26,6 +26,8 @@
  *    it in the license file.
  */
 
+#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kNetwork
+
 #include "mongo/platform/basic.h"
 
 #include "mongo/client/dbclientinterface.h"
@@ -33,6 +35,7 @@
 #include "mongo/rpc/op_msg.h"
 #include "mongo/unittest/integration_test.h"
 #include "mongo/unittest/unittest.h"
+#include "mongo/util/log.h"
 #include "mongo/util/scopeguard.h"
 
 namespace mongo {
@@ -165,4 +168,47 @@ TEST(OpMsg, CloseConnectionOnFireAndForgetNotMasterError) {
     ASSERT(foundSecondary);
 }
 
+TEST(OpMsg, Perf) {
+    std::string errMsg;
+    auto conn = std::unique_ptr<DBClientBase>(
+        unittest::getFixtureConnectionString().connect("integration_test", errMsg));
+    uassert(ErrorCodes::SocketException, errMsg, conn);
+
+    PseudoRandom prng(0xdeadbeef);
+
+    const auto ns = std::string("test.foo");
+    constexpr auto count = 1000'000;
+
+    log() << "setup start";
+    conn->dropCollection(ns);
+    for (auto i = 0; i < 1000; i++) {
+        // Random fill the string to defeat compression.
+        std::vector<BSONObj> objs;
+        objs.reserve(1000);
+        for (auto j = 0; j < count / 1000; j++) {
+            auto filler = std::string(1000, 'x');
+            std::generate(filler.begin(), filler.end(), [&] { return 'a' + prng.nextInt32(25); });
+            objs.push_back(BSON(GENOID << "s" << filler));
+        }
+        conn->insert(ns, objs);
+    }
+    ASSERT_EQ(conn->count(ns), size_t(count));  // also blocks until inserts are finished.
+    log() << "setup done";
+
+    for (auto i = 0; i < 5; i++) {
+        {
+            Timer timer;
+            ASSERT_EQ(conn->query(ns, {}, 0, 0, nullptr, 0)->itcount(), count);
+            log() << "plain:   " << timer.micros() / 1000. << "ms";
+        }
+
+        {
+            Timer timer;
+            ASSERT_EQ(conn->query(ns, {}, 0, 0, nullptr, QueryOption_Exhaust)->itcount(), count);
+            log() << "exhaust: " << timer.micros() / 1000. << "ms";
+        }
+        log() << "----\n";
+    }
+}
+
 }  // namespace mongo
diff --git a/src/mongo/transport/service_state_machine.cpp b/src/mongo/transport/service_state_machine.cpp
index d2b423d6c3..4940b1d636 100644
--- a/src/mongo/transport/service_state_machine.cpp
+++ b/src/mongo/transport/service_state_machine.cpp
@@ -37,6 +37,7 @@
 #include "mongo/db/dbmessage.h"
 #include "mongo/db/stats/counters.h"
 #include "mongo/rpc/message.h"
+#include "mongo/rpc/op_msg.h"
 #include "mongo/stdx/memory.h"
 #include "mongo/transport/message_compressor_manager.h"
 #include "mongo/transport/service_entry_point.h"
@@ -54,37 +55,68 @@
 
 namespace mongo {
 namespace {
-// Set up proper headers for formatting an exhaust request, if we need to
-bool setExhaustMessage(Message* m, const DbResponse& dbresponse) {
+// Returns a request that should be used for this exhaust-style reply, if we need to. Returns empty
+// message if reply shouldn't be handled as exhaust-style.
+Message makeLegacyExhaustMessage(const Message& request, const DbResponse& dbresponse) {
+    if (dbresponse.exhaustNS.empty())
+        return Message();
+
     MsgData::View header = dbresponse.response.header();
     QueryResult::View qr = header.view2ptr();
     long long cursorid = qr.getCursorId();
 
-    if (!cursorid) {
-        return false;
-    }
-
-    invariant(dbresponse.exhaustNS.size() && dbresponse.exhaustNS[0]);
-
-    auto ns = dbresponse.exhaustNS;  // m->reset() will free this so we must cache a copy
-
-    m->reset();
+    if (!cursorid)
+        return Message();
 
-    // Rebuild out the response.
+    // Make a fake request that will produce the next reply.
     BufBuilder b(512);
-    b.appendNum(static_cast<int>(0) /* size set later in setLen() */);
+    b.appendNum(static_cast<int>(0));          // size set later in setLen()
     b.appendNum(header.getId());               // message id
     b.appendNum(header.getResponseToMsgId());  // in response to
     b.appendNum(static_cast<int>(dbGetMore));  // opCode is OP_GET_MORE
     b.appendNum(static_cast<int>(0));          // Must be ZERO (reserved)
-    b.appendStr(ns);                           // Namespace
+    b.appendStr(dbresponse.exhaustNS);         // Namespace
     b.appendNum(static_cast<int>(0));          // ntoreturn
     b.appendNum(cursorid);                     // cursor id from the OP_REPLY
 
     MsgData::View(b.buf()).setLen(b.len());
-    m->setData(b.release());
 
-    return true;
+    return Message(b.release());
+}
+
+Message makeExhaustMessage(Message requestMsg, DbResponse* dbresponse) {
+    if (requestMsg.operation() != dbMsg)
+        return makeLegacyExhaustMessage(requestMsg, *dbresponse);
+
+    if (!OpMsg::isFlagSet(requestMsg, OpMsg::kExhaustSupported))
+        return Message();
+
+    auto& replyMsg = dbresponse->response;
+
+    // TODO don't reparse here. Propagate information more cleanly.
+    auto request = OpMsgRequest::parse(requestMsg);
+    if (request.getCommandName() != "getMore"_sd) {
+        // For now only supported on getMore requests so we can just reuse the original request.
+        // TODO consider supporting other commands and generate a synthetic getMore.
+        return Message();
+    }
+    auto reply = OpMsg::parse(replyMsg);
+    if (!reply.body["ok"].trueValue()) {
+        return Message();
+    }
+
+    // TODO don't assume reply is valid. (Related to above)
+    auto id = reply.body["cursor"]["id"].Long();
+    auto ns = reply.body["cursor"]["ns"].str();
+    if (id == 0 || ns.empty()) {
+        return Message();
+    }
+
+    // OK, keep doing the exhaust thing.
+    OpMsg::setFlag(&replyMsg, OpMsg::kMoreToCome);
+    requestMsg.header().setId(replyMsg.header().getId());
+    requestMsg.header().setResponseToMsgId(replyMsg.header().getResponseToMsgId());
+    return requestMsg;
 }
 
 }  // namespace
@@ -389,12 +421,8 @@ void ServiceStateMachine::_processMessage(ThreadGuard guard) {
         toSink.header().setResponseToMsgId(_inMessage.header().getId());
 
         // If this is an exhaust cursor, don't source more Messages
-        if (dbresponse.exhaustNS.size() > 0 && setExhaustMessage(&_inMessage, dbresponse)) {
-            _inExhaust = true;
-        } else {
-            _inExhaust = false;
-            _inMessage.reset();
-        }
+        _inMessage = makeExhaustMessage(std::move(_inMessage), &dbresponse);
+        _inExhaust = !_inMessage.empty();
 
         networkCounter.hitLogicalOut(toSink.size());