SERVER-44813 Implement streamable isMaster for standalones

3 files changed, 54 insertions, 38 deletions

diff --git a/jstests/noPassthrough/awaitable_ismaster.js b/jstests/noPassthrough/awaitable_ismaster.js
index 3899f7a8f4c..c0d25590771 100644
--- a/jstests/noPassthrough/awaitable_ismaster.js
+++ b/jstests/noPassthrough/awaitable_ismaster.js
@@ -62,7 +62,7 @@ function runTest(db) {
         },
         maxAwaitTimeMS: 0
     }),
-                                 [31382, 51761]);
+                                 [31382, 51761, 51764]);
 
     // Check that passing a topologyVersion not of type object fails.
     assert.commandFailedWithCode(
@@ -148,6 +148,11 @@ function runTest(db) {
                                  [31373, 51759]);
 }
 
+const conn = MongoRunner.runMongod({});
+assert.neq(null, conn, "mongod was unable to start up");
+runTest(conn.getDB("admin"));
+MongoRunner.stopMongod(conn);
+
 const replTest = new ReplSetTest({nodes: 1});
 replTest.startSet();
 replTest.initiate();
diff --git a/src/mongo/db/repl/replication_info.cpp b/src/mongo/db/repl/replication_info.cpp
index e058a7442aa..82e493e73d4 100644
--- a/src/mongo/db/repl/replication_info.cpp
+++ b/src/mongo/db/repl/replication_info.cpp
@@ -78,26 +78,47 @@ namespace {
 /**
  * Appends replication-related fields to the isMaster response. Returns the topology version that
  * was included in the response.
- * TODO SERVER-44813: Always return a topology version, including on standalones.
  */
-boost::optional<TopologyVersion> appendReplicationInfo(
-    OperationContext* opCtx,
-    BSONObjBuilder& result,
-    int level,
-    boost::optional<TopologyVersion> clientTopologyVersion,
-    boost::optional<Date_t> deadline) {
+TopologyVersion appendReplicationInfo(OperationContext* opCtx,
+                                      BSONObjBuilder& result,
+                                      int level,
+                                      boost::optional<TopologyVersion> clientTopologyVersion,
+                                      boost::optional<long long> maxAwaitTimeMS) {
     TopologyVersion topologyVersion;
     ReplicationCoordinator* replCoord = ReplicationCoordinator::get(opCtx);
     if (replCoord->getSettings().usingReplSets()) {
         const auto& horizonParams = SplitHorizon::getParameters(opCtx->getClient());
 
+        boost::optional<Date_t> deadline;
+        if (maxAwaitTimeMS) {
+            deadline = opCtx->getServiceContext()->getPreciseClockSource()->now() +
+                Milliseconds(*maxAwaitTimeMS);
+        }
         auto isMasterResponse =
             replCoord->awaitIsMasterResponse(opCtx, horizonParams, clientTopologyVersion, deadline);
         result.appendElements(isMasterResponse->toBSON());
         if (level) {
             replCoord->appendSlaveInfoData(&result);
         }
-        return isMasterResponse->getTopologyVersion();
+        invariant(isMasterResponse->getTopologyVersion());
+        return isMasterResponse->getTopologyVersion().get();
+    }
+
+    auto currentTopologyVersion = replCoord->getTopologyVersion();
+
+    if (clientTopologyVersion &&
+        clientTopologyVersion->getProcessId() == currentTopologyVersion.getProcessId()) {
+        uassert(51764,
+                str::stream() << "Received a topology version with counter: "
+                              << clientTopologyVersion->getCounter()
+                              << " which is greater than the server topology version counter: "
+                              << currentTopologyVersion.getCounter(),
+                clientTopologyVersion->getCounter() == currentTopologyVersion.getCounter());
+
+        // The topologyVersion never changes on a running standalone process, so just sleep for
+        // maxAwaitTimeMS.
+        invariant(maxAwaitTimeMS);
+        opCtx->sleepFor(Milliseconds(*maxAwaitTimeMS));
     }
 
     result.appendBool("ismaster",
@@ -167,7 +188,10 @@ boost::optional<TopologyVersion> appendReplicationInfo(
         replCoord->appendSlaveInfoData(&result);
     }
 
-    return boost::none;
+    BSONObjBuilder topologyVersionBuilder(result.subobjStart("topologyVersion"));
+    currentTopologyVersion.serialize(&topologyVersionBuilder);
+
+    return currentTopologyVersion;
 }
 
 class ReplicationInfoServerStatus : public ServerStatusSection {
@@ -191,7 +215,7 @@ public:
                               result,
                               level,
                               boost::none /* clientTopologyVersion */,
-                              boost::none /* deadline */);
+                              boost::none /* maxAwaitTimeMS */);
 
         auto rbid = ReplicationProcess::get(opCtx)->getRollbackID();
         if (ReplicationProcess::kUninitializedRollbackId != rbid) {
@@ -384,8 +408,7 @@ public:
         auto topologyVersionElement = cmdObj["topologyVersion"];
         auto maxAwaitTimeMSField = cmdObj["maxAwaitTimeMS"];
         boost::optional<TopologyVersion> clientTopologyVersion;
-        boost::optional<Date_t> deadline;
-        long long maxAwaitTimeMS;
+        boost::optional<long long> maxAwaitTimeMS;
         if (topologyVersionElement && maxAwaitTimeMSField) {
             clientTopologyVersion = TopologyVersion::parse(IDLParserErrorContext("TopologyVersion"),
                                                            topologyVersionElement.Obj());
@@ -393,12 +416,16 @@ public:
                     "topologyVersion must have a non-negative counter",
                     clientTopologyVersion->getCounter() >= 0);
 
-            uassertStatusOK(bsonExtractIntegerField(cmdObj, "maxAwaitTimeMS", &maxAwaitTimeMS));
-            uassert(31373, "maxAwaitTimeMS must be a non-negative integer", maxAwaitTimeMS >= 0);
+            {
+                long long parsedMaxAwaitTimeMS;
+                uassertStatusOK(
+                    bsonExtractIntegerField(cmdObj, "maxAwaitTimeMS", &parsedMaxAwaitTimeMS));
+                maxAwaitTimeMS = parsedMaxAwaitTimeMS;
+            }
+
+            uassert(31373, "maxAwaitTimeMS must be a non-negative integer", *maxAwaitTimeMS >= 0);
 
             LOG(3) << "Using maxAwaitTimeMS for awaitable isMaster protocol.";
-            deadline = opCtx->getServiceContext()->getPreciseClockSource()->now() +
-                Milliseconds(maxAwaitTimeMS);
         } else {
             uassert(31368,
                     (topologyVersionElement
@@ -409,7 +436,7 @@ public:
 
         auto result = replyBuilder->getBodyBuilder();
         auto currentTopologyVersion =
-            appendReplicationInfo(opCtx, result, 0, clientTopologyVersion, deadline);
+            appendReplicationInfo(opCtx, result, 0, clientTopologyVersion, maxAwaitTimeMS);
 
         if (serverGlobalParams.clusterRole == ClusterRole::ConfigServer) {
             const int configServerModeNumber = 2;
@@ -451,8 +478,7 @@ public:
         auto& saslMechanismRegistry = SASLServerMechanismRegistry::get(opCtx->getServiceContext());
         saslMechanismRegistry.advertiseMechanismNamesForUser(opCtx, cmdObj, &result);
 
-        // TODO SERVER-44813: currentTopologyVersion should always be set.
-        if (opCtx->isExhaust() && currentTopologyVersion) {
+        if (opCtx->isExhaust()) {
             LOG(3) << "Using exhaust for isMaster protocol";
 
             uassert(51756,
@@ -460,8 +486,8 @@ public:
                     maxAwaitTimeMSField);
             invariant(clientTopologyVersion);
 
-            if (clientTopologyVersion->getProcessId() == currentTopologyVersion->getProcessId() &&
-                clientTopologyVersion->getCounter() == currentTopologyVersion->getCounter()) {
+            if (clientTopologyVersion->getProcessId() == currentTopologyVersion.getProcessId() &&
+                clientTopologyVersion->getCounter() == currentTopologyVersion.getCounter()) {
                 // Indicate that an exhaust message should be generated and the previous BSONObj
                 // command parameters should be reused as the next BSONObj command parameters.
                 replyBuilder->setNextInvocation(boost::none);
@@ -471,7 +497,7 @@ public:
                     if (elt.fieldNameStringData() == "topologyVersion"_sd) {
                         BSONObjBuilder topologyVersionBuilder(
                             nextInvocationBuilder.subobjStart("topologyVersion"));
-                        currentTopologyVersion->serialize(&topologyVersionBuilder);
+                        currentTopologyVersion.serialize(&topologyVersionBuilder);
                     } else {
                         nextInvocationBuilder.append(elt);
                     }
diff --git a/src/mongo/rpc/op_msg_integration_test.cpp b/src/mongo/rpc/op_msg_integration_test.cpp
index 2e5cdad314b..3e9125c00a6 100644
--- a/src/mongo/rpc/op_msg_integration_test.cpp
+++ b/src/mongo/rpc/op_msg_integration_test.cpp
@@ -528,11 +528,6 @@ TEST(OpMsg, ServerHandlesExhaustIsMasterCorrectly) {
     uassert(ErrorCodes::SocketException, errMsg, fixtureConn);
     DBClientBase* conn = fixtureConn.get();
 
-    // TODO SERVER-44813: Run this test on standalone.
-    if (!fixtureConn->isReplicaSetMember() && !fixtureConn->isMongos()) {
-        return;
-    }
-
     if (fixtureConn->isReplicaSetMember()) {
         // Connect directly to the primary.
         conn = &static_cast<DBClientReplicaSet*>(fixtureConn.get())->masterConn();
@@ -596,11 +591,6 @@ TEST(OpMsg, ServerHandlesExhaustIsMasterWithTopologyChange) {
     uassert(ErrorCodes::SocketException, errMsg, fixtureConn);
     DBClientBase* conn = fixtureConn.get();
 
-    // TODO SERVER-44813: Run this test on standalone.
-    if (!fixtureConn->isReplicaSetMember() && !fixtureConn->isMongos()) {
-        return;
-    }
-
     if (fixtureConn->isReplicaSetMember()) {
         // Connect directly to the primary.
         conn = &static_cast<DBClientReplicaSet*>(fixtureConn.get())->masterConn();
@@ -667,11 +657,6 @@ TEST(OpMsg, ServerRejectsExhaustIsMasterWithoutMaxAwaitTimeMS) {
     uassert(ErrorCodes::SocketException, errMsg, fixtureConn);
     DBClientBase* conn = fixtureConn.get();
 
-    // TODO SERVER-44813: Run this test on standalone.
-    if (!fixtureConn->isReplicaSetMember() && !fixtureConn->isMongos()) {
-        return;
-    }
-
     if (fixtureConn->isReplicaSetMember()) {
         // Connect directly to the primary.
         conn = &static_cast<DBClientReplicaSet*>(fixtureConn.get())->masterConn();