SERVER-26990 Unify tracking of secondary state between replication and topology coordinators

This removes the slaveInfo structure and consolidates secondary state into the topology coordinator
memberHeartbeatData.

Author: mtrussotto

src/mongo/db/repl/SConscript

@@ -504,6 +504,7 @@ env.CppUnitTest(
 env.Library('topology_coordinator',
             [
                 'heartbeat_response_action.cpp',
+                'member_heartbeat_data.cpp',
                 'topology_coordinator.cpp',
             ],
             LIBDEPS=[
@@ -513,7 +514,6 @@ env.Library('topology_coordinator',
 
 env.Library('topology_coordinator_impl',
             [
-                'member_heartbeat_data.cpp',
                 'topology_coordinator_impl.cpp',
             ],
             LIBDEPS=[

src/mongo/db/repl/heartbeat_response_action.cpp

@@ -75,5 +75,9 @@ void HeartbeatResponseAction::setNextHeartbeatStartDate(Date_t when) {
     _nextHeartbeatStartDate = when;
 }
 
+void HeartbeatResponseAction::setAdvancedOpTime(bool advanced) {
+    _advancedOpTime = advanced;
+}
+
 }  // namespace repl
 }  // namespace mongo

src/mongo/db/repl/heartbeat_response_action.h

@@ -100,6 +100,11 @@ class HeartbeatResponseAction {
      */
     void setNextHeartbeatStartDate(Date_t when);
 
+    /**
+     * Sets whether or not the heartbeat response advanced the member's opTime.
+     */
+    void setAdvancedOpTime(bool advanced);
+
     /**
      * Gets the action type of this action.
      */
@@ -123,10 +128,19 @@ class HeartbeatResponseAction {
         return _primaryIndex;
     }
 
+    /*
+     * Returns true if the heartbeat response resulting in our conception of the
+     * member's optime moving forward, so we need to recalculate lastCommittedOpTime.
+     */
+    bool getAdvancedOpTime() const {
+        return _advancedOpTime;
+    }
+
 private:
     Action _action;
     int _primaryIndex;
     Date_t _nextHeartbeatStartDate;
+    bool _advancedOpTime = false;
 };
 
 }  // namespace repl

src/mongo/db/repl/member_heartbeat_data.cpp

@@ -39,13 +39,14 @@
 namespace mongo {
 namespace repl {
 
-MemberHeartbeatData::MemberHeartbeatData() : _health(-1), _authIssue(false) {
+MemberHeartbeatData::MemberHeartbeatData()
+    : _health(-1), _authIssue(false), _configIndex(-1), _isSelf(false) {
     _lastResponse.setState(MemberState::RS_UNKNOWN);
     _lastResponse.setElectionTime(Timestamp());
     _lastResponse.setAppliedOpTime(OpTime());
 }
 
-void MemberHeartbeatData::setUpValues(Date_t now,
+bool MemberHeartbeatData::setUpValues(Date_t now,
                                       const HostAndPort& host,
                                       ReplSetHeartbeatResponse&& hbResponse) {
     _health = 1;
@@ -54,6 +55,8 @@ void MemberHeartbeatData::setUpValues(Date_t now,
     }
     _authIssue = false;
     _lastHeartbeat = now;
+    _lastUpdate = now;
+    _lastUpdateStale = false;
     _updatedSinceRestart = true;
 
     if (!hbResponse.hasState()) {
@@ -71,7 +74,11 @@ void MemberHeartbeatData::setUpValues(Date_t now,
               << hbResponse.getState().toString() << rsLog;
     }
 
+    bool opTimeAdvanced = advanceLastAppliedOpTime(hbResponse.getAppliedOpTime(), now);
+    auto durableOpTime = hbResponse.hasDurableOpTime() ? hbResponse.getDurableOpTime() : OpTime();
+    opTimeAdvanced = advanceLastDurableOpTime(durableOpTime, now) || opTimeAdvanced;
     _lastResponse = std::move(hbResponse);
+    return opTimeAdvanced;
 }
 
 void MemberHeartbeatData::setDownValues(Date_t now, const std::string& heartbeatMessage) {
@@ -87,6 +94,9 @@ void MemberHeartbeatData::setDownValues(Date_t now, const std::string& heartbeat
     _lastResponse.setAppliedOpTime(OpTime());
     _lastResponse.setHbMsg(heartbeatMessage);
     _lastResponse.setSyncingTo(HostAndPort());
+
+    // The _lastAppliedOpTime/_lastDurableOpTime fields don't get cleared merely by missing a
+    // heartbeat.
 }
 
 void MemberHeartbeatData::setAuthIssue(Date_t now) {
@@ -104,5 +114,47 @@ void MemberHeartbeatData::setAuthIssue(Date_t now) {
     _lastResponse.setSyncingTo(HostAndPort());
 }
 
+void MemberHeartbeatData::setLastAppliedOpTime(OpTime opTime, Date_t now) {
+    _lastUpdate = now;
+    _lastUpdateStale = false;
+    _lastAppliedOpTime = opTime;
+}
+
+void MemberHeartbeatData::setLastDurableOpTime(OpTime opTime, Date_t now) {
+    _lastUpdate = now;
+    _lastUpdateStale = false;
+    if (_lastAppliedOpTime < opTime) {
+        // TODO(russotto): We think this should never happen, rollback or no rollback.  Make this an
+        // invariant and see what happens.
+        log() << "Durable progress (" << opTime << ") is ahead of the applied progress ("
+              << _lastAppliedOpTime << ". This is likely due to a "
+                                       "rollback."
+              << " memberid: " << _memberId << " rid: " << _rid << " host "
+              << _hostAndPort.toString() << " previous durable progress: " << _lastDurableOpTime;
+    } else {
+        _lastDurableOpTime = opTime;
+    }
+}
+
+bool MemberHeartbeatData::advanceLastAppliedOpTime(OpTime opTime, Date_t now) {
+    _lastUpdate = now;
+    _lastUpdateStale = false;
+    if (_lastAppliedOpTime < opTime) {
+        setLastAppliedOpTime(opTime, now);
+        return true;
+    }
+    return false;
+}
+
+bool MemberHeartbeatData::advanceLastDurableOpTime(OpTime opTime, Date_t now) {
+    _lastUpdate = now;
+    _lastUpdateStale = false;
+    if (_lastDurableOpTime < opTime) {
+        setLastDurableOpTime(opTime, now);
+        return true;
+    }
+    return false;
+}
+
 }  // namespace repl
 }  // namespace mongo

src/mongo/db/repl/member_heartbeat_data.h

@@ -68,10 +68,10 @@ class MemberHeartbeatData {
     const HostAndPort& getSyncSource() const {
         return _lastResponse.getSyncingTo();
     }
-    OpTime getAppliedOpTime() const {
+    OpTime getHeartbeatAppliedOpTime() const {
         return _lastResponse.getAppliedOpTime();
     }
-    OpTime getDurableOpTime() const {
+    OpTime getHeartbeatDurableOpTime() const {
         return _lastResponse.hasDurableOpTime() ? _lastResponse.getDurableOpTime() : OpTime();
     }
     int getConfigVersion() const {
@@ -105,10 +105,49 @@ class MemberHeartbeatData {
         return _health != 0;
     }
 
+    OpTime getLastAppliedOpTime() const {
+        return _lastAppliedOpTime;
+    }
+
+    OpTime getLastDurableOpTime() const {
+        return _lastDurableOpTime;
+    }
+
+    // When was the last time this data was updated via any means?
+    Date_t getLastUpdate() const {
+        return _lastUpdate;
+    }
+    // Was the last update stale as of the last check?
+    bool lastUpdateStale() const {
+        return _lastUpdateStale;
+    }
+
+    // Index of this member in the replica set config member list.
+    int getConfigIndex() const {
+        return _configIndex;
+    }
+
+    int getMemberId() const {
+        return _memberId;
+    }
+
+    OID getRid() const {
+        return _rid;
+    }
+
+    bool isSelf() const {
+        return _isSelf;
+    }
+
+    HostAndPort getHostAndPort() const {
+        return _hostAndPort;
+    }
+
     /**
      * Sets values in this object from the results of a successful heartbeat command.
+     * Returns whether or not the optimes advanced as a result of this heartbeat response.
      */
-    void setUpValues(Date_t now, const HostAndPort& host, ReplSetHeartbeatResponse&& hbResponse);
+    bool setUpValues(Date_t now, const HostAndPort& host, ReplSetHeartbeatResponse&& hbResponse);
 
     /**
      * Sets values in this object from the results of a erroring/failed heartbeat command.
@@ -134,6 +173,66 @@ class MemberHeartbeatData {
         return _updatedSinceRestart;
     }
 
+    /**
+     * Sets the last applied op time (not the heartbeat applied op time) and updates the
+     * lastUpdate time.
+     */
+    void setLastAppliedOpTime(OpTime opTime, Date_t now);
+
+    /**
+     * Sets the last durable op time (not the heartbeat durable op time)
+     */
+    void setLastDurableOpTime(OpTime opTime, Date_t now);
+
+    /**
+     * Sets the last applied op time (not the heartbeat applied op time) iff the new optime is
+     * later than the current optime, and updates the lastUpdate time.  Returns true if the
+     * optime was advanced.
+     */
+    bool advanceLastAppliedOpTime(OpTime opTime, Date_t now);
+
+    /**
+     * Sets the last durable op time (not the heartbeat applied op time) iff the new optime is
+     * later than the current optime, and updates the lastUpdate time.  Returns true if the
+     * optime was advanced.
+     */
+    bool advanceLastDurableOpTime(OpTime opTime, Date_t now);
+
+    /*
+     * Indicates that this data is stale, based on _lastUpdateTime.
+     */
+    void markLastUpdateStale() {
+        _lastUpdateStale = true;
+    }
+
+    /*
+     * Updates the _lastUpdateTime and clears staleness without changing anything else.
+     */
+    void updateLiveness(Date_t now) {
+        _lastUpdate = now;
+        _lastUpdateStale = false;
+    }
+
+    void setConfigIndex(int configIndex) {
+        _configIndex = configIndex;
+    }
+
+    void setIsSelf(bool isSelf) {
+        _isSelf = isSelf;
+    }
+
+    void setHostAndPort(HostAndPort hostAndPort) {
+        _hostAndPort = hostAndPort;
+    }
+
+    void setMemberId(int memberId) {
+        _memberId = memberId;
+    }
+
+    void setRid(OID rid) {
+        _rid = rid;
+    }
+
 private:
     // -1 = not checked yet, 0 = member is down/unreachable, 1 = member is up
     int _health;
@@ -153,6 +252,38 @@ class MemberHeartbeatData {
 
     // Have we received heartbeats since the last restart?
     bool _updatedSinceRestart = false;
+
+    // Last time we got any information about this member, whether heartbeat
+    // or replSetUpdatePosition.
+    Date_t _lastUpdate;
+
+    // Set when lastUpdate time exceeds the election timeout.  Implies that the member is down
+    // on the primary, but not the secondaries.
+    bool _lastUpdateStale = false;
+
+    // Last known OpTime that the replica has applied and journaled to.
+    OpTime _lastDurableOpTime;
+
+    // Last known OpTime that the replica has applied, whether journaled or unjournaled.
+    OpTime _lastAppliedOpTime;
+
+    // TODO(russotto): Since memberHeartbeatData is kept in config order, _configIndex
+    // and _isSelf may not be necessary.
+    // Index of this member in the replica set configuration.
+    int _configIndex;
+
+    // Is this the data for this member?
+    bool _isSelf;
+
+    // This member's RID, used only in master/slave replication.
+    OID _rid;
+
+    // This member's member ID.  memberId and hostAndPort duplicate information in the
+    // configuration for replica sets, but are required to be here for master/slave replication.
+    int _memberId = -1;
+
+    // Client address of this member.
+    HostAndPort _hostAndPort;
 };
 
 }  // namespace repl

src/mongo/db/repl/repl_set_html_summary.cpp

@@ -187,7 +187,7 @@ const std::string ReplSetHtmlSummary::toHtmlString() const {
             // TODO(dannenberg): change timestamp to optime in V1
             memberTable << td(memberHB.getLastHeartbeat() == Date_t()
                                   ? "?"
-                                  : memberHB.getAppliedOpTime().toString());
+                                  : memberHB.getHeartbeatAppliedOpTime().toString());
         }
         memberTable << _tr();
     }
@@ -201,7 +201,7 @@ const std::string ReplSetHtmlSummary::toHtmlString() const {
     const MemberConfig& selfConfig = _config.getMemberAt(_selfIndex);
 
     if (_primaryIndex >= 0 && _primaryIndex != _selfIndex && !selfConfig.isArbiter()) {
-        int lag = _hbData[_primaryIndex].getAppliedOpTime().getTimestamp().getSecs() -
+        int lag = _hbData[_primaryIndex].getHeartbeatAppliedOpTime().getTimestamp().getSecs() -
             _selfOptime.getTimestamp().getSecs();
         s << tr("Lag: ", str::stream() << lag << " secs");
     }