/**
 *    Copyright 2014 MongoDB Inc.
 *
 *    This program is free software: you can redistribute it and/or  modify
 *    it under the terms of the GNU Affero General Public License, version 3,
 *    as published by the Free Software Foundation.
 *
 *    This program is distributed in the hope that it will be useful,
 *    but WITHOUT ANY WARRANTY; without even the implied warranty of
 *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *    GNU Affero General Public License for more details.
 *
 *    You should have received a copy of the GNU Affero General Public License
 *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *
 *    As a special exception, the copyright holders give permission to link the
 *    code of portions of this program with the OpenSSL library under certain
 *    conditions as described in each individual source file and distribute
 *    linked combinations including the program with the OpenSSL library. You
 *    must comply with the GNU Affero General Public License in all respects for
 *    all of the code used other than as permitted herein. If you modify file(s)
 *    with this exception, you may extend this exception to your version of the
 *    file(s), but you are not obligated to do so. If you do not wish to do so,
 *    delete this exception statement from your version. If you delete this
 *    exception statement from all source files in the program, then also delete
 *    it in the license file.
 */

#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kReplication

#include "mongo/platform/basic.h"

#include "mongo/db/repl/topology_coordinator_impl.h"

#include <limits>

#include "mongo/db/operation_context.h"
#include "mongo/db/repl/heartbeat_response_action.h"
#include "mongo/db/repl/is_master_response.h"
#include "mongo/db/repl/isself.h"
#include "mongo/db/repl/repl_set_heartbeat_args.h"
#include "mongo/db/repl/repl_set_heartbeat_response.h"
#include "mongo/db/repl/replication_executor.h"
#include "mongo/db/repl/rslog.h"
#include "mongo/db/server_parameters.h"
#include "mongo/util/hex.h"
#include "mongo/util/log.h"
#include "mongo/util/mongoutils/str.h"
#include "mongo/util/scopeguard.h"

namespace mongo {
namespace repl {

    const Seconds TopologyCoordinatorImpl::LastVote::leaseTime = Seconds(30);

namespace {

    template <typename T>
    int indexOfIterator(const std::vector<T>& vec,
                        typename std::vector<T>::const_iterator& it) {
        return static_cast<int>(it - vec.begin());
    }

    // Interval between the time the last heartbeat from a node was received successfully, or
    // the time when we gave up retrying, and when the next heartbeat should be sent to a target.
    const Milliseconds kHeartbeatInterval(Seconds(2).total_milliseconds());

    // Maximum number of retries for a failed heartbeat.
    const int kMaxHeartbeatRetries = 2;

    /**
     * Returns true if the only up heartbeats are auth errors.
     */
    bool _hasOnlyAuthErrorUpHeartbeats(const std::vector<MemberHeartbeatData>& hbdata,
                                       const int selfIndex) {
        bool foundAuthError = false;
        for (std::vector<MemberHeartbeatData>::const_iterator it = hbdata.begin();
             it != hbdata.end();
             ++it) {
            if (indexOfIterator(hbdata, it) == selfIndex) {
                continue;
            }

            if (it->up()) {
                return false;
            }

            if (it->hasAuthIssue()) {
                foundAuthError = true;
            }
        }

        return foundAuthError;
    }

}  // namespace

    PingStats::PingStats() :
        count(0),
        value(std::numeric_limits<unsigned int>::max()),
        _lastHeartbeatStartDate(0),
        _numFailuresSinceLastStart(std::numeric_limits<int>::max()) {
    }

    void PingStats::start(Date_t now) {
        _lastHeartbeatStartDate = now;
        _numFailuresSinceLastStart = 0;
    }

    void PingStats::hit(int millis) {
        _numFailuresSinceLastStart = std::numeric_limits<int>::max();
        ++count;
        value = value == std::numeric_limits<unsigned int>::max() ? millis :
            static_cast<unsigned long>((value * .8) + (millis * .2));
    }

    void PingStats::miss() {
        ++_numFailuresSinceLastStart;
    }

    TopologyCoordinatorImpl::TopologyCoordinatorImpl(Seconds maxSyncSourceLagSecs) :
        _role(Role::follower),
        _currentPrimaryIndex(-1),
        _forceSyncSourceIndex(-1),
        _maxSyncSourceLagSecs(maxSyncSourceLagSecs),
        _selfIndex(-1),
        _stepDownPending(false),
        _stepDownUntil(0),
        _electionSleepUntil(0),
        _maintenanceModeCalls(0),
        _followerMode(MemberState::RS_STARTUP2)
    {
        invariant(getMemberState() == MemberState::RS_STARTUP);
    }

    TopologyCoordinator::Role TopologyCoordinatorImpl::getRole() const {
        return _role;
    }

    void TopologyCoordinatorImpl::setForceSyncSourceIndex(int index) {
        invariant(_forceSyncSourceIndex < _currentConfig.getNumMembers());
        _forceSyncSourceIndex = index;
    }

    HostAndPort TopologyCoordinatorImpl::getSyncSourceAddress() const {
        return _syncSource;
    }

    HostAndPort TopologyCoordinatorImpl::chooseNewSyncSource(Date_t now, 
                                                             const OpTime& lastOpApplied) {

        // If we are not a member of the current replica set configuration, no sync source is valid.
        if (_selfIndex == -1) {
            LOG(2) << "Cannot sync from any members because we are not in the replica set config";
            return HostAndPort();
        }

        // if we have a target we've requested to sync from, use it
        if (_forceSyncSourceIndex != -1) {
            invariant(_forceSyncSourceIndex < _currentConfig.getNumMembers());
            _syncSource = _currentConfig.getMemberAt(_forceSyncSourceIndex).getHostAndPort();
            _forceSyncSourceIndex = -1;
            std::string msg(str::stream() << "syncing from: "
                                          << _syncSource.toString() << " by request");
            log() << msg << rsLog;
            setMyHeartbeatMessage(now, msg);
            return _syncSource;
        }

        // wait for 2N pings (not counting ourselves) before choosing a sync target
        int needMorePings = (_hbdata.size() - 1) * 2 - _getTotalPings();

        if (needMorePings > 0) {
            OCCASIONALLY log() << "waiting for " << needMorePings 
                               << " pings from other members before syncing";
            _syncSource = HostAndPort();
            return _syncSource;
        }

        // If we are only allowed to sync from the primary, set that
        if (!_currentConfig.isChainingAllowed()) {
            if (_currentPrimaryIndex == -1) {
                LOG(1) << "Cannot select sync source because chaining is"
                          " not allowed and primary is unknown/down";
                _syncSource = HostAndPort();
                return _syncSource;
            }
            else if (_memberIsBlacklisted(*_currentPrimaryMember(), now)) {
                LOG(1) << "Cannot select sync source because chaining is"
                    "not allowed and primary is not currently accepting our updates";
                _syncSource = HostAndPort();
                return _syncSource;
            }
            else {
                _syncSource = _currentConfig.getMemberAt(_currentPrimaryIndex).getHostAndPort();
                std::string msg(str::stream() << "syncing from primary: "
                                              << _syncSource.toString());
                log() << msg << rsLog;
                setMyHeartbeatMessage(now, msg);
                return _syncSource;
            }
        }

        // find the member with the lowest ping time that is ahead of me

        // Find primary's oplog time. Reject sync candidates that are more than
        // maxSyncSourceLagSecs seconds behind.
        OpTime primaryOpTime;
        if (_currentPrimaryIndex != -1) {
            primaryOpTime = _hbdata[_currentPrimaryIndex].getOpTime();
        }
        else {
            // choose a time that will exclude no candidates, since we don't see a primary
            primaryOpTime = OpTime(_maxSyncSourceLagSecs.total_seconds(), 0);
        }

        if (primaryOpTime.getSecs() < 
            static_cast<unsigned int>(_maxSyncSourceLagSecs.total_seconds())) {
            // erh - I think this means there was just a new election
            // and we don't yet know the new primary's optime
            primaryOpTime = OpTime(_maxSyncSourceLagSecs.total_seconds(), 0);
        }

        OpTime oldestSyncOpTime(primaryOpTime.getSecs() - _maxSyncSourceLagSecs.total_seconds(), 0);

        int closestIndex = -1;

        // Make two attempts.  The first attempt, we ignore those nodes with
        // slave delay higher than our own, hidden nodes, and nodes that are excessively lagged.
        // The second attempt includes such nodes, in case those are the only ones we can reach.
        // This loop attempts to set 'closestIndex'.
        for (int attempts = 0; attempts < 2; ++attempts) {
            for (std::vector<MemberHeartbeatData>::const_iterator it = _hbdata.begin(); 
                 it != _hbdata.end(); 
                 ++it) {
                const int itIndex = indexOfIterator(_hbdata, it);
                // Don't consider ourselves.
                if (itIndex == _selfIndex) {
                    continue;
                }
                // Candidate must be up to be considered.
                if (!it->up()) {
                    continue;
                }
                // Candidate must be PRIMARY or SECONDARY state to be considered.
                if (!it->getState().readable()) {
                    continue;
                }

                const MemberConfig& itMemberConfig(_currentConfig.getMemberAt(itIndex));

                // Candidate must build indexes if we build indexes, to be considered.
                if (_selfConfig().shouldBuildIndexes()) {
                    if (!itMemberConfig.shouldBuildIndexes()) {
                        continue;
                    }
                }

                // only consider candidates that are ahead of where we are
                if (it->getOpTime() <= lastOpApplied) {
                    continue;
                }

                // omit candidates that are excessively behind, on the first attempt at least.
                if (attempts == 0 &&
                    it->getOpTime() < oldestSyncOpTime) {
                    continue;
                }

                // omit nodes that are more latent than anything we've already considered
                if ((closestIndex != -1) &&
                    (_getPing(itMemberConfig.getHostAndPort())
                     > _getPing(_currentConfig.getMemberAt(closestIndex).getHostAndPort()))) {
                    continue;
                }

                if (attempts == 0) {
                    if (_selfConfig().getSlaveDelay() < itMemberConfig.getSlaveDelay()
                        || itMemberConfig.isHidden()) {
                        continue; // skip this one in the first attempt
                    }
                }

                if (_memberIsBlacklisted(itMemberConfig, now)) {
                    continue;
                }

                // This candidate has passed all tests; set 'closestIndex'
                closestIndex = itIndex;
            }
            if (closestIndex != -1) break; // no need for second attempt
        }

        if (closestIndex == -1) {
            // Did not find any members to sync from
            std::string msg("could not find member to sync from");
            // Only log when we had a valid sync source before
            if (!_syncSource.empty()) {
                log() << msg << rsLog;
            }
            setMyHeartbeatMessage(now, msg);

            _syncSource = HostAndPort();
            return _syncSource;
        }
        _syncSource = _currentConfig.getMemberAt(closestIndex).getHostAndPort();
        std::string msg(str::stream() << "syncing from: " << _syncSource.toString(), 0);
        log() << msg << rsLog;
        setMyHeartbeatMessage(now, msg);
        return _syncSource;
    }

    bool TopologyCoordinatorImpl::_memberIsBlacklisted(const MemberConfig& memberConfig, Date_t now)
    {
        std::map<HostAndPort,Date_t>::iterator blacklisted =
            _syncSourceBlacklist.find(memberConfig.getHostAndPort());
        if (blacklisted != _syncSourceBlacklist.end()) {
            if (blacklisted->second > now) {
                return true;
            }
            // Do some veto housekeeping; side-effect!
            _syncSourceBlacklist.erase(blacklisted);
        }
        return false;
    }

    void TopologyCoordinatorImpl::blacklistSyncSource(const HostAndPort& host, Date_t until) {
        _syncSourceBlacklist[host] = until;
    }

    void TopologyCoordinatorImpl::clearSyncSourceBlacklist() {
        _syncSourceBlacklist.clear();
    }

    void TopologyCoordinatorImpl::prepareSyncFromResponse(
            const ReplicationExecutor::CallbackData& data,
            const HostAndPort& target,
            const OpTime& lastOpApplied,
            BSONObjBuilder* response,
            Status* result) {
        if (data.status == ErrorCodes::CallbackCanceled) {
            *result = Status(ErrorCodes::ShutdownInProgress, "replication system is shutting down");
            return;
        }

        response->append("syncFromRequested", target.toString());

        if (_selfIndex == -1) {
            *result = Status(ErrorCodes::NotSecondary,
                             "Removed and uninitialized nodes do not sync");
            return;
        }

        const MemberConfig& selfConfig = _selfConfig();
        if (selfConfig.isArbiter()) {
            *result = Status(ErrorCodes::NotSecondary, "arbiters don't sync");
            return;
        }
        if (_selfIndex == _currentPrimaryIndex) {
            *result = Status(ErrorCodes::NotSecondary, "primaries don't sync");
            return;
        }

        ReplicaSetConfig::MemberIterator targetConfig = _currentConfig.membersEnd();
        int targetIndex = 0;
        for (ReplicaSetConfig::MemberIterator it = _currentConfig.membersBegin();
                it != _currentConfig.membersEnd(); ++it) {
            if (it->getHostAndPort() == target) {
                targetConfig = it;
                break;
            }
            ++targetIndex;
        }
        if (targetConfig == _currentConfig.membersEnd()) {
            *result = Status(ErrorCodes::NodeNotFound,
                             str::stream() << "Could not find member \"" << target.toString() <<
                                     "\" in replica set");
            return;
        }
        if (targetIndex == _selfIndex) {
            *result = Status(ErrorCodes::InvalidOptions, "I cannot sync from myself");
            return;
        }
        if (targetConfig->isArbiter()) {
            *result = Status(ErrorCodes::InvalidOptions,
                             str::stream() << "Cannot sync from \"" << target.toString() <<
                                     "\" because it is an arbiter");
            return;
        }
        if (!targetConfig->shouldBuildIndexes() && selfConfig.shouldBuildIndexes()) {
            *result = Status(ErrorCodes::InvalidOptions,
                             str::stream() << "Cannot sync from \"" << target.toString() <<
                                     "\" because it does not build indexes");
            return;
        }

        const MemberHeartbeatData& hbdata = _hbdata[targetIndex];
        if (hbdata.hasAuthIssue()) {
            *result = Status(ErrorCodes::Unauthorized,
                             str::stream() << "not authorized to communicate with " <<
                                     target.toString());
            return;
        }
        if (hbdata.getHealth() == 0) {
            *result = Status(ErrorCodes::HostUnreachable,
                             str::stream() << "I cannot reach the requested member: " <<
                                     target.toString());
            return;
        }
        if (hbdata.getOpTime().getSecs()+10 < lastOpApplied.getSecs()) {
            warning() << "attempting to sync from " << target
                      << ", but its latest opTime is " << hbdata.getOpTime().getSecs()
                      << " and ours is " << lastOpApplied.getSecs() << " so this may not work";
            response->append("warning",
                             str::stream() << "requested member \"" << target.toString() <<
                                     "\" is more than 10 seconds behind us");
            // not returning bad Status, just warning
        }

        HostAndPort prevSyncSource = getSyncSourceAddress();
        if (!prevSyncSource.empty()) {
            response->append("prevSyncTarget", prevSyncSource.toString());
        }

        setForceSyncSourceIndex(targetIndex);
        *result = Status::OK();
    }

    void TopologyCoordinatorImpl::prepareFreshResponse(
            const ReplicationCoordinator::ReplSetFreshArgs& args,
            const Date_t now,
            const OpTime lastOpApplied,
            BSONObjBuilder* response,
            Status* result) {

        if (_selfIndex == -1) {
            *result = Status(ErrorCodes::ReplicaSetNotFound,
                             "Cannot participate in elections because not initialized");
            return;
        }

        if (args.setName != _currentConfig.getReplSetName()) {
            *result = Status(ErrorCodes::ReplicaSetNotFound,
                             str::stream() << "Wrong repl set name. Expected: " <<
                                     _currentConfig.getReplSetName() <<
                                     ", received: " << args.setName);
            return;
        }

        if (args.id == static_cast<unsigned>(_selfConfig().getId())) {
            *result = Status(ErrorCodes::BadValue,
                             str::stream() << "Received replSetFresh command from member with the "
                                     "same member ID as ourself: " << args.id);
            return;
        }

        bool weAreFresher = false;
        if( _currentConfig.getConfigVersion() > args.cfgver ) {
            log() << "replSet member " << args.who << " is not yet aware its cfg version "
                  << args.cfgver << " is stale";
            response->append("info", "config version stale");
            weAreFresher = true;
        }
        // check not only our own optime, but any other member we can reach
        else if (args.opTime < _latestKnownOpTime(lastOpApplied)) {
            weAreFresher = true;
        }
        response->appendDate("opTime", lastOpApplied.asDate());
        response->append("fresher", weAreFresher);

        std::string errmsg;
        bool doVeto = _shouldVetoMember(args, now, lastOpApplied, &errmsg);
        response->append("veto", doVeto);
        if (doVeto) {
            response->append("errmsg", errmsg);
        }
        *result = Status::OK();
    }

    bool TopologyCoordinatorImpl::_shouldVetoMember(
            const ReplicationCoordinator::ReplSetFreshArgs& args,
            const Date_t& now,
            const OpTime& lastOpApplied,
            std::string* errmsg) const {

        if (_currentConfig.getConfigVersion() < args.cfgver) {
            // We are stale; do not veto.
            return false;
        }

        const unsigned int memberID = args.id;
        const int hopefulIndex = _getMemberIndex(memberID);
        invariant(hopefulIndex != _selfIndex);
        const int highestPriorityIndex = _getHighestPriorityElectableIndex(now, lastOpApplied);

        if (hopefulIndex == -1) {
            *errmsg = str::stream() << "replSet couldn't find member with id " << memberID;
            return true;
        }

        if (_iAmPrimary() && lastOpApplied >= _hbdata[hopefulIndex].getOpTime()) {
            // hbinfo is not updated for ourself, so if we are primary we have to check the
            // primary's last optime separately
            *errmsg = str::stream() << "I am already primary, " <<
                _currentConfig.getMemberAt(hopefulIndex).getHostAndPort().toString() << 
                " can try again once I've stepped down";
            return true;
        }

        if (_currentPrimaryIndex != -1 &&
                (hopefulIndex != _currentPrimaryIndex) &&
                (_hbdata[_currentPrimaryIndex].getOpTime() >=
                        _hbdata[hopefulIndex].getOpTime())) {
            // other members might be aware of more up-to-date nodes
            *errmsg = str::stream() <<
                _currentConfig.getMemberAt(hopefulIndex).getHostAndPort().toString() <<
                " is trying to elect itself but " << 
                _currentConfig.getMemberAt(_currentPrimaryIndex).getHostAndPort().toString() <<
                " is already primary and more up-to-date";
            return true;
        }

        if ((highestPriorityIndex != -1)) {
            const MemberConfig& hopefulMember = _currentConfig.getMemberAt(hopefulIndex);
            const MemberConfig& priorityMember = _currentConfig.getMemberAt(highestPriorityIndex);

            if (priorityMember.getPriority() > hopefulMember.getPriority()) {
                *errmsg = str::stream()
                            << hopefulMember.getHostAndPort().toString()
                            << " has lower priority of " << hopefulMember.getPriority() << " than "
                            << priorityMember.getHostAndPort().toString()
                            << " which has a priority of " << priorityMember.getPriority();
                return true;
            }
        }

        UnelectableReasonMask reason = _getUnelectableReason(hopefulIndex, lastOpApplied);
        reason &= ~RefusesToStand;
        if (reason) {
            *errmsg = str::stream()
                         << "I don't think "
                         << _currentConfig.getMemberAt(hopefulIndex).getHostAndPort().toString()
                         << " is electable because the " << _getUnelectableReasonString(reason);
            return true;
        }

        return false;
    }

    // produce a reply to a received electCmd
    void TopologyCoordinatorImpl::prepareElectResponse(
            const ReplicationCoordinator::ReplSetElectArgs& args,
            const Date_t now,
            const OpTime lastOpApplied,
            BSONObjBuilder* response,
            Status* result) {

        if (_selfIndex == -1) {
            *result = Status(ErrorCodes::ReplicaSetNotFound,
                             "Cannot participate in election because not initialized");
            return;
        }

        const long long myver = _currentConfig.getConfigVersion();
        const int highestPriorityIndex = _getHighestPriorityElectableIndex(now, lastOpApplied);

        const MemberConfig* primary = _currentPrimaryMember();
        const MemberConfig* hopeful = _currentConfig.findMemberByID(args.whoid);
        const MemberConfig* highestPriority = highestPriorityIndex == -1 ? NULL :
                &_currentConfig.getMemberAt(highestPriorityIndex);

        int vote = 0;
        if (args.set != _currentConfig.getReplSetName()) {
            log() << "replSet error received an elect request for '" << args.set
                  << "' but our set name is '" <<
                _currentConfig.getReplSetName() << "'";
        }
        else if ( myver < args.cfgver ) {
            // we are stale.  don't vote
            log() << "replSetElect not voting because our config version is stale. Our version: " <<
                    myver << ", their version: " << args.cfgver;
        }
        else if ( myver > args.cfgver ) {
            // they are stale!
            log() << "replSetElect command received stale config version # during election. "
                    "Our version: " << myver << ", their version: " << args.cfgver;
            vote = -10000;
        }
        else if (!hopeful) {
            log() << "replSetElect couldn't find member with id " << args.whoid;
            vote = -10000;
        }
        else if (_iAmPrimary()) {
            log() << "I am already primary, "  << hopeful->getHostAndPort().toString()
                  << " can try again once I've stepped down";
            vote = -10000;
        }
        else if (primary) {
            log() << hopeful->getHostAndPort().toString() << " is trying to elect itself but "
                  << primary->getHostAndPort().toString() << " is already primary";
            vote = -10000;
        }
        else if (highestPriority && highestPriority->getPriority() > hopeful->getPriority()) {
            // TODO(spencer): What if the lower-priority member is more up-to-date?
            log() << hopeful->getHostAndPort().toString() << " has lower priority than "
                  << highestPriority->getHostAndPort().toString();
            vote = -10000;
        }
        else if (_lastVote.when.millis + LastVote::leaseTime.total_milliseconds() >= now.millis &&
                 _lastVote.whoId != args.whoid) {
            log() << "replSet voting no for "
                  <<  hopeful->getHostAndPort().toString()
                  << "; voted for " << _lastVote.whoHostAndPort.toString() << ' '
                  << (now.millis - _lastVote.when.millis) / 1000 << " secs ago";
        }
        else {
            _lastVote.when = now;
            _lastVote.whoId = args.whoid;
            _lastVote.whoHostAndPort = hopeful->getHostAndPort();
            vote = _selfConfig().getNumVotes();
            invariant(hopeful->getId() == args.whoid);
            if (vote > 0) {
                log() << "replSetElect voting yea for " << hopeful->getHostAndPort().toString()
                      << " (" << args.whoid << ')';
            }
        }

        response->append("vote", vote);
        response->append("round", args.round);
        *result = Status::OK();
    }

    // produce a reply to a heartbeat
    Status TopologyCoordinatorImpl::prepareHeartbeatResponse(
            Date_t now,
            const ReplSetHeartbeatArgs& args,
            const std::string& ourSetName,
            const OpTime& lastOpApplied,
            ReplSetHeartbeatResponse* response) {

        if (args.getProtocolVersion() != 1) {
            return Status(ErrorCodes::BadValue,
                          str::stream() << "replset: incompatible replset protocol version: "
                          << args.getProtocolVersion());
        }

        // Verify that replica set names match
        const std::string rshb = args.getSetName();
        if (ourSetName != rshb) {
            log() << "replSet set names do not match, ours: " << ourSetName <<
                "; remote node's: " << rshb;
            response->noteMismatched();
            return Status(ErrorCodes::InconsistentReplicaSetNames, str::stream() <<
                          "Our set name of " << ourSetName << " does not match name " << rshb <<
                          " reported by remote node");
        }

        const MemberState myState = getMemberState();
        if (_selfIndex == -1) {
            if (myState.removed()) {
                return Status(ErrorCodes::InvalidReplicaSetConfig,
                              "Our replica set configuration is invalid or does not include us");
            }
        }
        else {
            invariant(_currentConfig.getReplSetName() == args.getSetName());
            if (args.getSenderId() == _selfConfig().getId()) {
                return Status(ErrorCodes::BadValue,
                              str::stream() << "Received heartbeat from member with the same "
                                      "member ID as ourself: " << args.getSenderId());
            }
        }

        // This is a replica set
        response->noteReplSet();

        // For 2.6 compatibility
        if (_currentConfig.isInitialized()) {
            response->setSetName(ourSetName);
        }
        response->setState(myState.s);
        if (myState.primary()) {
            response->setElectionTime(_electionTime);
        }

        // Are we electable
        response->setElectable(!_getMyUnelectableReason(now, lastOpApplied));

        // Heartbeat status message
        response->setHbMsg(_getHbmsg(now));
        response->setTime(Seconds(Milliseconds(now.asInt64()).total_seconds()));
        response->setOpTime(lastOpApplied.asDate());

        if (!_syncSource.empty()) {
            response->setSyncingTo(_syncSource.toString());
        }

        if (!_currentConfig.isInitialized()) {
            response->setVersion(-2);
            return Status::OK();
        }

        const long long v = _currentConfig.getConfigVersion();
        response->setVersion(v);
        // Deliver new config if caller's version is older than ours
        if (v > args.getConfigVersion()) {
            response->setConfig(_currentConfig);
        }

        // Resolve the caller's id in our Member list
        int from = -1;
        if (v == args.getConfigVersion() && args.getSenderId() != -1) {
            from = _getMemberIndex(args.getSenderId());
        }
        if (from == -1) {
            // Can't find the member, so we leave out the stateDisagreement field
            return Status::OK();
        }
        invariant(from != _selfIndex);

        // if we thought that this node is down, let it know
        if (!_hbdata[from].up()) {
            response->noteStateDisagreement();
        }

        // note that we got a heartbeat from this node
        _hbdata[from].setLastHeartbeatRecv(now);
        return Status::OK();
    }


    int TopologyCoordinatorImpl::_getMemberIndex(int id) const {
        int index = 0;
        for (ReplicaSetConfig::MemberIterator it = _currentConfig.membersBegin();
             it != _currentConfig.membersEnd();
             ++it, ++index) {
            if (it->getId() == id) {
                return index;
            }
        }
        return -1;
    }

    std::pair<ReplSetHeartbeatArgs, Milliseconds> TopologyCoordinatorImpl::prepareHeartbeatRequest(
                Date_t now,
                const std::string& ourSetName,
                const HostAndPort& target) {

        PingStats& hbStats = _pings[target];
        Milliseconds alreadyElapsed(now.asInt64() - hbStats.getLastHeartbeatStartDate().asInt64());
        if (!_currentConfig.isInitialized() ||
            (hbStats.getNumFailuresSinceLastStart() > kMaxHeartbeatRetries) ||
            (alreadyElapsed >= _currentConfig.getHeartbeatTimeoutPeriodMillis())) {

            // This is either the first request ever for "target", or the heartbeat timeout has
            // passed, so we're starting a "new" heartbeat.
            hbStats.start(now);
            alreadyElapsed = Milliseconds(0);
        }
        ReplSetHeartbeatArgs hbArgs;
        hbArgs.setProtocolVersion(1);
        hbArgs.setCheckEmpty(false);
        if (_currentConfig.isInitialized()) {
            hbArgs.setSetName(_currentConfig.getReplSetName());
            hbArgs.setConfigVersion(_currentConfig.getConfigVersion());
            if (_selfIndex >= 0) {
                const MemberConfig& me = _selfConfig();
                hbArgs.setSenderHost(me.getHostAndPort());
                hbArgs.setSenderId(me.getId());
            }
        }
        else {
            hbArgs.setSetName(ourSetName);
            hbArgs.setConfigVersion(-2);
        }

        const Milliseconds timeoutPeriod(
                _currentConfig.isInitialized() ?
                _currentConfig.getHeartbeatTimeoutPeriodMillis() :
                Milliseconds(
                        ReplicaSetConfig::kDefaultHeartbeatTimeoutPeriod.total_milliseconds()));
        const Milliseconds timeout(
                timeoutPeriod.total_milliseconds() - alreadyElapsed.total_milliseconds());
        return std::make_pair(hbArgs, timeout);
    }

    HeartbeatResponseAction TopologyCoordinatorImpl::processHeartbeatResponse(
            Date_t now,
            Milliseconds networkRoundTripTime,
            const HostAndPort& target,
            const StatusWith<ReplSetHeartbeatResponse>& hbResponse,
            OpTime myLastOpApplied) {

        PingStats& hbStats = _pings[target];
        invariant(hbStats.getLastHeartbeatStartDate() != Date_t(0));
        if (!hbResponse.isOK()) {
            hbStats.miss();
        }
        else {
            hbStats.hit(networkRoundTripTime.total_milliseconds());
            // Log diagnostics.
            if (hbResponse.getValue().isStateDisagreement()) {
                LOG(1) << target <<
                    " thinks that we are down because they cannot send us heartbeats.";
            }
        }

        const bool isUnauthorized =
                            (hbResponse.getStatus().code() == ErrorCodes::Unauthorized) ||
                            (hbResponse.getStatus().code() == ErrorCodes::AuthenticationFailed);

        Milliseconds alreadyElapsed(now.asInt64() - hbStats.getLastHeartbeatStartDate().asInt64());
        Date_t nextHeartbeatStartDate;
        if (_currentConfig.isInitialized() &&
            (hbStats.getNumFailuresSinceLastStart() <= kMaxHeartbeatRetries) &&
            (alreadyElapsed < _currentConfig.getHeartbeatTimeoutPeriodMillis())) {

            if (!hbResponse.isOK() && !isUnauthorized) {
                LOG(1) << "Bad heartbeat response from " << target <<
                    "; trying again; Retries left: " <<
                    (kMaxHeartbeatRetries - hbStats.getNumFailuresSinceLastStart()) <<
                    "; " << alreadyElapsed.total_milliseconds() << "ms have already elapsed";
            }
            if (isUnauthorized) {
                nextHeartbeatStartDate = now + kHeartbeatInterval.total_milliseconds();
            } else {
                nextHeartbeatStartDate = now;
            }
        }
        else {
            nextHeartbeatStartDate = now + kHeartbeatInterval.total_milliseconds();
        }

        if (hbResponse.isOK() && hbResponse.getValue().hasConfig()) {
            const long long currentConfigVersion =
                _currentConfig.isInitialized() ? _currentConfig.getConfigVersion() : -2;
            const ReplicaSetConfig& newConfig = hbResponse.getValue().getConfig();
            if (newConfig.getConfigVersion() > currentConfigVersion) {
                HeartbeatResponseAction nextAction = HeartbeatResponseAction::makeReconfigAction();
                nextAction.setNextHeartbeatStartDate(nextHeartbeatStartDate);
                return nextAction;
            }
            else {
                // Could be we got the newer version before we got the response, or the
                // target erroneously sent us one, even through it isn't newer.
                if (newConfig.getConfigVersion() < currentConfigVersion) {
                    LOG(1) << "Config version from heartbeat was older than ours.";
                }
                else {
                    LOG(2) << "Config from heartbeat response was same as ours.";
                }
                if (logger::globalLogDomain()->shouldLog(
                            MongoLogDefaultComponent_component,
                            ::mongo::LogstreamBuilder::severityCast(2))) {
                    LogstreamBuilder lsb = log();
                    if (_currentConfig.isInitialized()) {
                        lsb << "Current config: " << _currentConfig.toBSON() << "; ";
                    }
                    lsb << "Config in heartbeat: " << newConfig.toBSON();
                }
            }
        }

        // Check if the heartbeat target is in our config.  If it isn't, there's nothing left to do,
        // so return early.
        if (!_currentConfig.isInitialized()) {
            HeartbeatResponseAction nextAction = HeartbeatResponseAction::makeNoAction();
            nextAction.setNextHeartbeatStartDate(nextHeartbeatStartDate);
            return nextAction;
        }
        const int memberIndex = _currentConfig.findMemberIndexByHostAndPort(target);
        if (memberIndex == -1) {
            LOG(1) << "replset: Could not find " << target  << " in current config so ignoring --"
                " current config: " << _currentConfig.toBSON();
            HeartbeatResponseAction nextAction = HeartbeatResponseAction::makeNoAction();
            nextAction.setNextHeartbeatStartDate(nextHeartbeatStartDate);
            return nextAction;
        }
        invariant(memberIndex != _selfIndex);

        MemberHeartbeatData& hbData = _hbdata[memberIndex];
        const MemberConfig member = _currentConfig.getMemberAt(memberIndex);
        if (!hbResponse.isOK()) {
            if (isUnauthorized) {
                LOG(3) << "setAuthIssue: heartbeat response failed due to authentication"
                    " issue for member _id:" << member.getId();
                hbData.setAuthIssue(now);
            } else {
                LOG(3) << "setDownValues: heartbeat response failed for member _id:"
                       << member.getId() << ", msg:  "
                       << hbResponse.getStatus().reason();

                hbData.setDownValues(now, hbResponse.getStatus().reason());
            }
        }
        else {
            ReplSetHeartbeatResponse hbr = hbResponse.getValue();
            LOG(3) << "setUpValues: heartbeat response good for member _id:"
                   << member.getId() << ", msg:  "
                   << hbr.getHbMsg();
            hbData.setUpValues(now, member.getHostAndPort(), hbr);
        }
        HeartbeatResponseAction nextAction = _updateHeartbeatDataImpl(
                memberIndex,
                now,
                myLastOpApplied);

        nextAction.setNextHeartbeatStartDate(nextHeartbeatStartDate);
        return nextAction;
    }

    HeartbeatResponseAction TopologyCoordinatorImpl::_updateHeartbeatDataImpl(
            int updatedConfigIndex,
            Date_t now,
            const OpTime& lastOpApplied) {

        // This method has two interrelated responsibilities, performed in two phases.
        //
        // First, it updates the local notion of which remote node, if any is primary.  In the
        // process, it may request a remote primary to step down because there is a higher priority
        // node waiting, or because the local node thinks it is primary and that it has a more
        // recent electionTime.  It may instead decide that the local node should step down itself,
        // because a remote has a more recent election time.
        //
        // Second, if there is no remote primary, and the local node is not primary, it considers
        // whether or not to stand for election.
        invariant(updatedConfigIndex != _selfIndex);

        // We are missing from the config, so do not participate in primary maintenance or election.
        if (_selfIndex == -1) {
            return HeartbeatResponseAction::makeNoAction();
        }

        ////////////////////
        // Phase 1
        ////////////////////

        // If we believe the node whose data was just updated is primary, confirm that
        // the updated data supports that notion.  If not, erase our notion of who is primary.
        if (updatedConfigIndex == _currentPrimaryIndex) {
            const MemberHeartbeatData& updatedHBData = _hbdata[updatedConfigIndex];
            if (!updatedHBData.up() || !updatedHBData.getState().primary()) {
                _currentPrimaryIndex = -1;
            }
        }

        // If the current primary is not highest priority and up to date (within 10s),
        // have them/me stepdown.
        if (_currentPrimaryIndex != -1) {
            // check if we should ask the primary (possibly ourselves) to step down
            const int highestPriorityIndex = _getHighestPriorityElectableIndex(now, lastOpApplied);
            if (highestPriorityIndex != -1) {
                const MemberConfig& currentPrimaryMember =
                    _currentConfig.getMemberAt(_currentPrimaryIndex);
                const MemberConfig& highestPriorityMember =
                    _currentConfig.getMemberAt(highestPriorityIndex);
                const OpTime highestPriorityMemberOptime = highestPriorityIndex == _selfIndex ?
                        lastOpApplied : _hbdata[highestPriorityIndex].getOpTime();

                if ((highestPriorityMember.getPriority() > currentPrimaryMember.getPriority()) &&
                        _isOpTimeCloseEnoughToLatestToElect(highestPriorityMemberOptime,
                                                            lastOpApplied)) {
                    const OpTime latestOpTime = _latestKnownOpTime(lastOpApplied);

                    if (_iAmPrimary()) {
                        if (_stepDownPending) {
                            return HeartbeatResponseAction::makeNoAction();
                        }
                        _stepDownPending = true;
                        log() << "Stepping down self (priority "
                              << currentPrimaryMember.getPriority() << ") because "
                              << highestPriorityMember.getHostAndPort() << " has higher priority "
                              << highestPriorityMember.getPriority() << " and is only "
                              << (latestOpTime.getSecs() - highestPriorityMemberOptime.getSecs())
                              << " seconds behind me";
                        const Date_t until = now +
                            LastVote::leaseTime.total_milliseconds() +
                            kHeartbeatInterval.total_milliseconds();
                        if (_electionSleepUntil < until) {
                            _electionSleepUntil = until;
                        }
                        return HeartbeatResponseAction::makeStepDownSelfAction(_selfIndex);
                    }
                    else if ((highestPriorityMemberOptime == _selfIndex) &&
                             (_electionSleepUntil <= now)) {
                        // If this node is the highest priority node, and it is not in
                        // an inter-election sleep period, ask the current primary to step down.
                        // This is an optimization, because the remote primary will almost certainly
                        // notice this node's electability promptly, via its own heartbeat process.
                        log() << "Requesting that " << currentPrimaryMember.getHostAndPort()
                              << " (priority " << currentPrimaryMember.getPriority()
                              << ") step down because I have higher priority "
                              << highestPriorityMember.getPriority() << " and am only "
                              << (latestOpTime.getSecs() - highestPriorityMemberOptime.getSecs())
                              << " seconds behind it";
                        int primaryIndex = _currentPrimaryIndex;
                        _currentPrimaryIndex = -1;
                        return HeartbeatResponseAction::makeStepDownRemoteAction(primaryIndex);
                    }
                }
            }
        }

        // Scan the member list's heartbeat data for who is primary, and update
        // _currentPrimaryIndex and _role, or request a remote to step down, as necessary.
        {
            int remotePrimaryIndex = -1;
            for (std::vector<MemberHeartbeatData>::const_iterator it = _hbdata.begin();
                 it != _hbdata.end();
                 ++it) {
                const int itIndex = indexOfIterator(_hbdata, it);
                if (itIndex == _selfIndex) {
                    continue;
                }

                if( it->getState().primary() && it->up() ) {
                    if (remotePrimaryIndex != -1) {
                        // two other nodes think they are primary (asynchronously polled) 
                        // -- wait for things to settle down.
                        log() << "replSet info two remote primaries (transiently)";
                        return HeartbeatResponseAction::makeNoAction();
                    }
                    remotePrimaryIndex = itIndex;
                }
            }

            if (remotePrimaryIndex != -1) {
                // If it's the same as last time, don't do anything further.
                if (_currentPrimaryIndex == remotePrimaryIndex) {
                    return HeartbeatResponseAction::makeNoAction();
                }
                // Clear last heartbeat message on ourselves (why?)
                setMyHeartbeatMessage(now, "");

                // If we are also primary, this is a problem.  Determine who should step down.
                if (_iAmPrimary()) {
                    OpTime remoteElectionTime = _hbdata[remotePrimaryIndex].getElectionTime();
                    log() << "replset: another primary seen with election time " 
                          << remoteElectionTime << " my elction time is " << _electionTime;

                    // Step down whomever has the older election time.
                    if (remoteElectionTime > _electionTime) {
                        if (_stepDownPending) {
                            return HeartbeatResponseAction::makeNoAction();
                        }
                        _stepDownPending = true;
                        log() << "stepping down; another primary was elected more recently";
                        return HeartbeatResponseAction::makeStepDownSelfAction(_selfIndex);
                    }
                    else {
                        log() << "another PRIMARY detected and it should step down"
                                 " since it was elected earlier than me";
                        return HeartbeatResponseAction::makeStepDownRemoteAction(
                                remotePrimaryIndex);
                    }
                }

                _currentPrimaryIndex = remotePrimaryIndex;
                return HeartbeatResponseAction::makeNoAction();
            }
        }

        ////////////////////
        // Phase 2
        ////////////////////

        // We do not believe any remote to be primary.

        // If we are primary, check if we can still see majority of the set;
        // stepdown if we can't.
        if (_iAmPrimary()) {
            if (CannotSeeMajority & _getMyUnelectableReason(now, lastOpApplied)) {
                if (_stepDownPending) {
                    return HeartbeatResponseAction::makeNoAction();
                }
                _stepDownPending = true;
                log() << "can't see a majority of the set, relinquishing primary";
                return HeartbeatResponseAction::makeStepDownSelfAction(_selfIndex);
            }

            LOG(2) << "Choosing to remain primary";
            return HeartbeatResponseAction::makeNoAction();
        }

        fassert(18505, _currentPrimaryIndex == -1);

        // At this point, there is no primary anywhere.  Check to see if we should become a
        // candidate.
        if (!checkShouldStandForElection(now, lastOpApplied)) {
            return HeartbeatResponseAction::makeNoAction();
        }
        return HeartbeatResponseAction::makeElectAction();
    }

    bool TopologyCoordinatorImpl::checkShouldStandForElection(
            Date_t now, const OpTime& lastOpApplied) {
        if (_currentPrimaryIndex != -1) {
            return false;
        }
        invariant (_role != Role::leader);

        if (_role == Role::candidate) {
            LOG(2) << "Not standing for election again; already candidate";
            return false;
        }

        const UnelectableReasonMask unelectableReason = _getMyUnelectableReason(now, lastOpApplied);
        if (NotCloseEnoughToLatestOptime & unelectableReason) {
            LOG(2) << "Not standing for election because " <<
                _getUnelectableReasonString(unelectableReason) << "; my last optime is " <<
                lastOpApplied << " and the newest is " << _latestKnownOpTime(lastOpApplied);
            return false;
        }
        if (unelectableReason) {
            LOG(2) << "Not standing for election because " <<
                _getUnelectableReasonString(unelectableReason);
            return false;
        }
        if (_electionSleepUntil > now) {
            LOG(2) << "Not standing for election before " <<
                dateToISOStringLocal(_electionSleepUntil) << " because I stood too recently";
            return false;
        }
        // All checks passed, become a candidate and start election proceedings.
        _role = Role::candidate;
        return true;
    }

    bool TopologyCoordinatorImpl::_aMajoritySeemsToBeUp() const {
        int vUp = 0;
        for (std::vector<MemberHeartbeatData>::const_iterator it = _hbdata.begin(); 
             it != _hbdata.end(); 
             ++it) {
            const int itIndex = indexOfIterator(_hbdata, it);
            if (itIndex == _selfIndex || it->up()) {
                vUp += _currentConfig.getMemberAt(itIndex).getNumVotes();
            }
        }

        return vUp * 2 > _currentConfig.getTotalVotingMembers();
    }

    bool TopologyCoordinatorImpl::_isOpTimeCloseEnoughToLatestToElect(
            const OpTime& otherOpTime, const OpTime& ourLastOpApplied) const {
        const OpTime latestKnownOpTime = _latestKnownOpTime(ourLastOpApplied);
        return otherOpTime.getSecs() >= (latestKnownOpTime.getSecs() - 10);
    }

    bool TopologyCoordinatorImpl::_iAmPrimary() const {
        if (_role == Role::leader) {
            invariant(_currentPrimaryIndex == _selfIndex);
            return true;
        }
        return false;
    }

    OpTime TopologyCoordinatorImpl::_latestKnownOpTime(OpTime ourLastOpApplied) const {
        OpTime latest = ourLastOpApplied;

        for (std::vector<MemberHeartbeatData>::const_iterator it = _hbdata.begin();
             it != _hbdata.end(); 
             ++it) {

            if (indexOfIterator(_hbdata, it) == _selfIndex) {
                continue;
            }
            if (!it->up()) {
                continue;
            }

            OpTime optime = it->getOpTime();

            if (optime > latest) {
                latest = optime;
            }
        }

        return latest;
    }

    bool TopologyCoordinatorImpl::_isMemberHigherPriority(int memberOneIndex,
                                                          int memberTwoIndex) const {
        if (memberOneIndex == -1)
            return false;

        if (memberTwoIndex == -1)
            return true;

        return _currentConfig.getMemberAt(memberOneIndex).getPriority() >
                _currentConfig.getMemberAt(memberTwoIndex).getPriority();
    }

    int TopologyCoordinatorImpl::_getHighestPriorityElectableIndex(
            Date_t now, OpTime lastOpApplied) const {
        int maxIndex = -1;
        for (int currentIndex = 0; currentIndex < _currentConfig.getNumMembers(); currentIndex++) {
            UnelectableReasonMask reason = currentIndex == _selfIndex ?
                    _getMyUnelectableReason(now, lastOpApplied) :
                    _getUnelectableReason(currentIndex, lastOpApplied);
            if (None == reason && _isMemberHigherPriority(currentIndex, maxIndex)) {
                maxIndex = currentIndex;
            }
        }

        return maxIndex;
    }

    void TopologyCoordinatorImpl::changeMemberState_forTest(const MemberState& newMemberState,
                                                            OpTime electionTime) {
        invariant(_selfIndex != -1);
        if (newMemberState == getMemberState())
            return;
        switch(newMemberState.s) {
        case MemberState::RS_PRIMARY:
            _role = Role::candidate;
            processWinElection(OID(), electionTime);
            invariant(_role == Role::leader);
            break;
        case MemberState::RS_SECONDARY:
        case MemberState::RS_ROLLBACK:
        case MemberState::RS_RECOVERING:
        case MemberState::RS_STARTUP2:
            _role = Role::follower;
            _followerMode = newMemberState.s;
            if (_currentPrimaryIndex == _selfIndex) {
                _currentPrimaryIndex = -1;
                _stepDownPending = false;
            }
            break;
        case MemberState::RS_STARTUP:
            updateConfig(
                    ReplicaSetConfig(),
                    -1,
                    Date_t(),
                    OpTime());
            break;
        default:
            severe() << "Cannot switch to state " << newMemberState;
            invariant(false);
        }
        if (getMemberState() != newMemberState.s) {
            severe() << "Expected to enter state " << newMemberState << " but am now in " <<
                getMemberState();
            invariant(false);
        }
        log() << "replSet " << newMemberState;
    }

    void TopologyCoordinatorImpl::_setCurrentPrimaryForTest(int primaryIndex) {
        if (primaryIndex == _selfIndex) {
            changeMemberState_forTest(MemberState::RS_PRIMARY);
        }
        else {
            if (_iAmPrimary()) {
                changeMemberState_forTest(MemberState::RS_SECONDARY);
            }
            if (primaryIndex != -1) {
                ReplSetHeartbeatResponse hbResponse;
                hbResponse.setState(MemberState::RS_PRIMARY);
                hbResponse.setElectionTime(OpTime());
                hbResponse.setOpTime(_hbdata[primaryIndex].getOpTime());
                hbResponse.setSyncingTo("");
                hbResponse.setHbMsg("");
                _hbdata[primaryIndex].setUpValues(
                        _hbdata[primaryIndex].getLastHeartbeat(),
                        _currentConfig.getMemberAt(primaryIndex).getHostAndPort(),
                        hbResponse);
            }
            _currentPrimaryIndex = primaryIndex;
        }
    }

    const MemberConfig* TopologyCoordinatorImpl::_currentPrimaryMember() const {
        if (_currentPrimaryIndex == -1)
            return NULL;

        return &(_currentConfig.getMemberAt(_currentPrimaryIndex));
    }

    void TopologyCoordinatorImpl::prepareStatusResponse(
            const ReplicationExecutor::CallbackData& data,
            Date_t now,
            unsigned selfUptime,
            const OpTime& lastOpApplied,
            BSONObjBuilder* response,
            Status* result) {
        if (data.status == ErrorCodes::CallbackCanceled) {
            *result = Status(ErrorCodes::ShutdownInProgress, "replication system is shutting down");
            return;
        }

        // output for each member
        vector<BSONObj> membersOut;
        const MemberState myState = getMemberState();

        if (_selfIndex == -1) {
            // We're REMOVED or have an invalid config
            response->append("state", static_cast<int>(myState.s));
            response->append("stateStr", myState.toString());
            response->append("uptime", selfUptime);
            response->append("optime", lastOpApplied);
            response->appendDate("optimeDate", Date_t(lastOpApplied.getSecs() * 1000ULL));
            if (_maintenanceModeCalls) {
                response->append("maintenanceMode", _maintenanceModeCalls);
            }
            std::string s = _getHbmsg(now);
            if( !s.empty() )
                response->append("infoMessage", s);
            *result = Status(ErrorCodes::InvalidReplicaSetConfig,
                             "Our replica set config is invalid or we are not a member of it");
            return;
        }

        for (std::vector<MemberHeartbeatData>::const_iterator it = _hbdata.begin(); 
             it != _hbdata.end(); 
             ++it) {
            const int itIndex = indexOfIterator(_hbdata, it);
            if (itIndex == _selfIndex) {
                // add self
                BSONObjBuilder bb;
                bb.append("_id", _selfConfig().getId());
                bb.append("name", _selfConfig().getHostAndPort().toString());
                bb.append("health", 1.0);
                bb.append("state", static_cast<int>(myState.s));
                bb.append("stateStr", myState.toString());
                bb.append("uptime", selfUptime);
                if (!_selfConfig().isArbiter()) {
                    bb.append("optime", lastOpApplied);
                    bb.appendDate("optimeDate", Date_t(lastOpApplied.getSecs() * 1000ULL));
                }

                if (!_syncSource.empty()) {
                    bb.append("syncingTo", _syncSource.toString());
                }

                if (_maintenanceModeCalls) {
                    bb.append("maintenanceMode", _maintenanceModeCalls);
                }

                std::string s = _getHbmsg(now);
                if( !s.empty() )
                    bb.append("infoMessage", s);

                if (myState.primary()) {
                    bb.append("electionTime", _electionTime);
                    bb.appendDate("electionDate", Date_t(_electionTime.getSecs() * 1000ULL));
                }
                bb.appendIntOrLL("configVersion", _currentConfig.getConfigVersion());
                bb.append("self", true);
                membersOut.push_back(bb.obj());
            }
            else {
                // add non-self member
                const MemberConfig& itConfig = _currentConfig.getMemberAt(itIndex);
                BSONObjBuilder bb;
                bb.append("_id", itConfig.getId());
                bb.append("name", itConfig.getHostAndPort().toString());
                double h = it->getHealth();
                bb.append("health", h);
                const MemberState state = it->getState();
                bb.append("state", static_cast<int>(state.s));
                if( h == 0 ) {
                    // if we can't connect the state info is from the past
                    // and could be confusing to show
                    bb.append("stateStr", "(not reachable/healthy)");
                }
                else {
                    bb.append("stateStr", it->getState().toString());
                }

                const unsigned int uptime = static_cast<unsigned int> ((it->getUpSince() ?
                        (now - it->getUpSince()) / 1000 /* convert millis to secs */ : 0));
                bb.append("uptime", uptime);
                if (!itConfig.isArbiter()) {
                    bb.append("optime", it->getOpTime());
                    bb.appendDate("optimeDate", Date_t(it->getOpTime().getSecs() * 1000ULL));
                }
                bb.appendDate("lastHeartbeat", it->getLastHeartbeat());
                bb.appendDate("lastHeartbeatRecv", it->getLastHeartbeatRecv());
                const int ping = _getPing(itConfig.getHostAndPort());
                if (ping != -1) {
                    bb.append("pingMs", ping);
                    std::string s = it->getLastHeartbeatMsg();
                    if( !s.empty() )
                        bb.append("lastHeartbeatMessage", s);
                }
                if (it->hasAuthIssue()) {
                    bb.append("authenticated", false);
                }
                const std::string syncSource = it->getSyncSource();
                if (!syncSource.empty()) {
                    bb.append("syncingTo", syncSource);
                }

                if (state == MemberState::RS_PRIMARY) {
                    bb.append("electionTime", it->getElectionTime());
                    bb.appendDate("electionDate",
                                  Date_t(it->getElectionTime().getSecs() * 1000ULL));
                }
                bb.appendIntOrLL("configVersion", it->getConfigVersion());
                membersOut.push_back(bb.obj());
            }
        }

        // sort members bson
        sort(membersOut.begin(), membersOut.end());

        response->append("set",
                         _currentConfig.isInitialized() ? _currentConfig.getReplSetName() : "");
        response->append("date", now);
        response->append("myState", myState.s);

        // Add sync source info
        if (!_syncSource.empty() && !myState.primary() && !myState.removed()) {
            response->append("syncingTo", _syncSource.toString());
        }

        response->append("members", membersOut);
        *result = Status::OK();
    }

    void TopologyCoordinatorImpl::fillIsMasterForReplSet(IsMasterResponse* response) {

        const MemberState myState = getMemberState();
        if (!_currentConfig.isInitialized() || myState.removed()) {
            response->markAsNoConfig();
            return;
        }

        response->setReplSetName(_currentConfig.getReplSetName());
        response->setReplSetVersion(_currentConfig.getConfigVersion());
        response->setIsMaster(myState.primary());
        response->setIsSecondary(myState.secondary());

        {
            for (ReplicaSetConfig::MemberIterator it = _currentConfig.membersBegin();
                    it != _currentConfig.membersEnd(); ++it) {
                if (it->isHidden() || it->getSlaveDelay().total_seconds() > 0) {
                    continue;
                }

                if (it->isElectable()) {
                    response->addHost(it->getHostAndPort());
                }
                else if (it->isArbiter()) {
                    response->addArbiter(it->getHostAndPort());
                }
                else {
                    response->addPassive(it->getHostAndPort());
                }
            }
        }

        const MemberConfig* curPrimary = _currentPrimaryMember();
        if (curPrimary) {
            response->setPrimary(curPrimary->getHostAndPort());
        }

        const MemberConfig& selfConfig = _currentConfig.getMemberAt(_selfIndex);
        if (selfConfig.isArbiter()) {
            response->setIsArbiterOnly(true);
        }
        else if (selfConfig.getPriority() == 0) {
            response->setIsPassive(true);
        }
        if (selfConfig.getSlaveDelay().total_seconds()) {
            response->setSlaveDelay(selfConfig.getSlaveDelay());
        }
        if (selfConfig.isHidden()) {
            response->setIsHidden(true);
        }
        if (!selfConfig.shouldBuildIndexes()) {
            response->setShouldBuildIndexes(false);
        }
        const ReplicaSetTagConfig tagConfig = _currentConfig.getTagConfig();
        if (selfConfig.hasTags(tagConfig)) {
            for (MemberConfig::TagIterator tag = selfConfig.tagsBegin();
                    tag != selfConfig.tagsEnd(); ++tag) {
                std::string tagKey = tagConfig.getTagKey(*tag);
                if (tagKey[0] == '$') {
                    // Filter out internal tags
                    continue;
                }
                response->addTag(tagKey, tagConfig.getTagValue(*tag));
            }
        }
        response->setMe(selfConfig.getHostAndPort());
        if (_iAmPrimary()) {
            response->setElectionId(_electionId);
        }
    }

    void TopologyCoordinatorImpl::prepareFreezeResponse(
            Date_t now, int secs, BSONObjBuilder* response) {

        if (secs == 0) {
            _stepDownUntil = now;
            log() << "replSet info 'unfreezing'";
            response->append("info", "unfreezing");

            if (_followerMode == MemberState::RS_SECONDARY &&
                    _currentConfig.getNumMembers() == 1 &&
                    _selfIndex == 0 &&
                    _currentConfig.getMemberAt(_selfIndex).isElectable()) {
                // If we are a one-node replica set, we're the one member,
                // we're electable, and we are currently in followerMode SECONDARY,
                // we must transition to candidate now that our stepdown period
                // is no longer active, in leiu of heartbeats.
                _role = Role::candidate;
            }
        }
        else {
            if ( secs == 1 )
                response->append("warning", "you really want to freeze for only 1 second?");

            if (!_iAmPrimary()) {
                _stepDownUntil = std::max(_stepDownUntil, Date_t(now + (secs * 1000)));
                log() << "replSet info 'freezing' for " << secs << " seconds";
            }
            else {
                log() << "replSet info received freeze command but we are primary";
            }
        }
    }

    bool TopologyCoordinatorImpl::becomeCandidateIfStepdownPeriodOverAndSingleNodeSet(Date_t now) {
        if (_stepDownUntil > now) {
            return false;
        }

        if (_followerMode == MemberState::RS_SECONDARY &&
                _currentConfig.getNumMembers() == 1 &&
                _selfIndex == 0 &&
                _currentConfig.getMemberAt(_selfIndex).isElectable()) {
            // If the new config describes a one-node replica set, we're the one member,
            // we're electable, and we are currently in followerMode SECONDARY,
            // we must transition to candidate, in leiu of heartbeats.
            _role = Role::candidate;
            return true;
        }
        return false;
    }

    void TopologyCoordinatorImpl::setElectionSleepUntil(Date_t newTime) {
        if (_electionSleepUntil < newTime) {
            _electionSleepUntil = newTime;
        }
    }

    OpTime TopologyCoordinatorImpl::getElectionTime() const {
        return _electionTime;
    }

    OID TopologyCoordinatorImpl::getElectionId() const {
        return _electionId;
    }

    int TopologyCoordinatorImpl::getCurrentPrimaryIndex() const {
        return _currentPrimaryIndex;
    }

    Date_t TopologyCoordinatorImpl::getStepDownTime() const {
        return _stepDownUntil;
    }

    void TopologyCoordinatorImpl::_updateHeartbeatDataForReconfig(const ReplicaSetConfig& newConfig,
                                                                  int selfIndex,
                                                                  Date_t now) {
        std::vector<MemberHeartbeatData> oldHeartbeats;
        _hbdata.swap(oldHeartbeats);

        int index = 0;
        for (ReplicaSetConfig::MemberIterator it = newConfig.membersBegin();
                it != newConfig.membersEnd();
                ++it, ++index) {
            const MemberConfig& newMemberConfig = *it;
            // TODO: C++11: use emplace_back()
            if (index == selfIndex) {
                // Insert placeholder for ourself, though we will never consult it.
                _hbdata.push_back(MemberHeartbeatData());
            }
            else {
                MemberHeartbeatData newHeartbeatData;
                for (int oldIndex = 0; oldIndex < _currentConfig.getNumMembers(); ++oldIndex) {
                    const MemberConfig& oldMemberConfig = _currentConfig.getMemberAt(oldIndex);
                    if (oldMemberConfig.getId() == newMemberConfig.getId() &&
                            oldMemberConfig.getHostAndPort() == newMemberConfig.getHostAndPort()) {
                        // This member existed in the old config with the same member ID and
                        // HostAndPort, so copy its heartbeat data over.
                        newHeartbeatData = oldHeartbeats[oldIndex];
                        break;
                    }
                }
                _hbdata.push_back(newHeartbeatData);
            }
        }
    }

    // This function installs a new config object and recreates MemberHeartbeatData objects
    // that reflect the new config.
    void TopologyCoordinatorImpl::updateConfig(const ReplicaSetConfig& newConfig,
                                               int selfIndex,
                                               Date_t now,
                                               OpTime lastOpApplied) {
        invariant(_role != Role::candidate);
        invariant(selfIndex < newConfig.getNumMembers());

        _updateHeartbeatDataForReconfig(newConfig, selfIndex, now);
        _currentConfig = newConfig;
        _selfIndex = selfIndex;
        _forceSyncSourceIndex = -1;

        if (_role == Role::leader) {
            if (_selfIndex == -1) {
                log() << "Could not remain primary because no longer a member of the replica set";
            }
            else if (!_selfConfig().isElectable()) {
                log() <<" Could not remain primary because no longer electable";
            }
            else {
                // Don't stepdown if you don't have to.
                _currentPrimaryIndex = _selfIndex;
                return;
            }
            _role = Role::follower;
        }

        // By this point we know we are in Role::follower
        _currentPrimaryIndex = -1; // force secondaries to re-detect who the primary is
        _stepDownPending = false;

        if (_followerMode == MemberState::RS_SECONDARY &&
                _currentConfig.getNumMembers() == 1 &&
                _selfIndex == 0 &&
                _currentConfig.getMemberAt(_selfIndex).isElectable()) {
            // If the new config describes a one-node replica set, we're the one member,
            // we're electable, and we are currently in followerMode SECONDARY, 
            // we must transition to candidate, in leiu of heartbeats.
            _role = Role::candidate;
        }
    }
    std::string TopologyCoordinatorImpl::_getHbmsg(Date_t now) const {
        // ignore messages over 2 minutes old
        if ((now - _hbmsgTime) > 120) {
            return "";
        }
        return _hbmsg;
    }

    void TopologyCoordinatorImpl::setMyHeartbeatMessage(const Date_t now,
                                                        const std::string& message) {
        _hbmsgTime = now;
        _hbmsg = message;
    }

    const MemberConfig& TopologyCoordinatorImpl::_selfConfig() const {
        return _currentConfig.getMemberAt(_selfIndex);
    }

    TopologyCoordinatorImpl::UnelectableReasonMask TopologyCoordinatorImpl::_getUnelectableReason(
            int index,
            const OpTime& lastOpApplied) const {
        invariant(index != _selfIndex);
        const MemberConfig& memberConfig = _currentConfig.getMemberAt(index);
        const MemberHeartbeatData& hbData = _hbdata[index];
        UnelectableReasonMask result = None;
        if (memberConfig.isArbiter()) {
            result |= ArbiterIAm;
        }
        if (memberConfig.getPriority() <= 0) {
            result |= NoPriority;
        }
        if (hbData.getState() != MemberState::RS_SECONDARY) {
            result |= NotSecondary;
        }
        if (!_isOpTimeCloseEnoughToLatestToElect(hbData.getOpTime(), lastOpApplied)) {
            result |= NotCloseEnoughToLatestOptime;
        }
        if (hbData.up() && hbData.isUnelectable()) {
            result |= RefusesToStand;
        }
        invariant(result || memberConfig.isElectable());
        return result;
    }

    TopologyCoordinatorImpl::UnelectableReasonMask TopologyCoordinatorImpl::_getMyUnelectableReason(
            const Date_t now,
            const OpTime lastApplied) const {

        UnelectableReasonMask result = None;
        if (lastApplied.isNull()) {
            result |= NoData;
        }
        if (!_aMajoritySeemsToBeUp()) {
            result |= CannotSeeMajority;
        }
        if (_selfIndex == -1) {
            result |= NotInitialized;
            return result;
        }
        if (_selfConfig().isArbiter()) {
            result |= ArbiterIAm;
        }
        if (_selfConfig().getPriority() <= 0) {
            result |= NoPriority;
        }
        if (_stepDownUntil > now) {
            result |= StepDownPeriodActive;
        }
        if (_lastVote.whoId != -1 &&
                _lastVote.whoId !=_currentConfig.getMemberAt(_selfIndex).getId() &&
                _lastVote.when.millis + LastVote::leaseTime.total_milliseconds() >= now.millis) {
            result |= VotedTooRecently;
        }

        // Cannot be electable unless secondary or already primary
        if (!getMemberState().secondary() && !_iAmPrimary()) {
            result |= NotSecondary;
        }
        if (!_isOpTimeCloseEnoughToLatestToElect(lastApplied, lastApplied)) {
            result |= NotCloseEnoughToLatestOptime;
        }
        return result;
    }

    std::string TopologyCoordinatorImpl::_getUnelectableReasonString(
            const UnelectableReasonMask ur) const {
        invariant(ur);
        str::stream ss;
        bool hasWrittenToStream = false;
        if (ur & NoData) {
            ss << "node has no applied oplog entries";
            hasWrittenToStream = true;
        }
        if (ur & VotedTooRecently) {
            if (hasWrittenToStream) {
                ss << "; ";
            }
            hasWrittenToStream = true;
            ss << "I recently voted for " << _lastVote.whoHostAndPort.toString();
        }
        if (ur & CannotSeeMajority) {
            if (hasWrittenToStream) {
                ss << "; ";
            }
            hasWrittenToStream = true;
            ss << "I cannot see a majority";
        }
        if (ur & ArbiterIAm) {
            if (hasWrittenToStream) {
                ss << "; ";
            }
            hasWrittenToStream = true;
            ss << "member is an arbiter";
        }
        if (ur & NoPriority) {
            if (hasWrittenToStream) {
                ss << "; ";
            }
            hasWrittenToStream = true;
            ss <<  "member has zero priority";
        }
        if (ur & StepDownPeriodActive) {
            if (hasWrittenToStream) {
                ss << "; ";
            }
            hasWrittenToStream = true;
            ss << "I am still waiting for stepdown period to end at " <<
                dateToISOStringLocal(_stepDownUntil);
        }
        if (ur & NotSecondary) {
            if (hasWrittenToStream) {
                ss << "; ";
            }
            hasWrittenToStream = true;
            ss << "member is not currently a secondary";
        }
        if (ur & NotCloseEnoughToLatestOptime) {
            if (hasWrittenToStream) {
                ss << "; ";
            }
            hasWrittenToStream = true;
            ss << "member is more than 10 seconds behind the most up-to-date member";
        }
        if (ur & NotInitialized) {
            if (hasWrittenToStream) {
                ss << "; ";
            }
            hasWrittenToStream = true;
            ss <<  "node is not a member of a valid replica set configuration";
        }
        if (ur & RefusesToStand) {
            if (hasWrittenToStream) {
                ss << "; ";
            }
            hasWrittenToStream = true;
            ss << "most recent heartbeat indicates node will not stand for election";
        }
        if (!hasWrittenToStream) {
            severe() << "Invalid UnelectableReasonMask value 0x" << integerToHex(ur);
            fassertFailed(26011);
        }
        ss << " (mask 0x" << integerToHex(ur) << ")";
        return ss;
    }

    int TopologyCoordinatorImpl::_getPing(const HostAndPort& host) {
        return _pings[host].getMillis();
    }

    void TopologyCoordinatorImpl::_setElectionTime(const OpTime& newElectionTime) {
        _electionTime = newElectionTime;
    }

    int TopologyCoordinatorImpl::_getTotalPings() {
        PingMap::iterator it = _pings.begin();
        PingMap::iterator end = _pings.end();
        int totalPings = 0;
        while (it != end) {
            totalPings += it->second.getCount();
            it++;
        }
        return totalPings;
    }

    std::vector<HostAndPort> TopologyCoordinatorImpl::getMaybeUpHostAndPorts() const {
        std::vector<HostAndPort> upHosts;
        for (std::vector<MemberHeartbeatData>::const_iterator it = _hbdata.begin(); 
             it != _hbdata.end(); 
             ++it) {
            const int itIndex = indexOfIterator(_hbdata, it);
            if (itIndex == _selfIndex) {
                continue;    // skip ourselves
            }
            if (!it->maybeUp()) {
                continue;    // skip DOWN nodes
            }

            upHosts.push_back(_currentConfig.getMemberAt(itIndex).getHostAndPort());
        }
        return upHosts;
    }

    bool TopologyCoordinatorImpl::voteForMyself(Date_t now) {
        if (_role != Role::candidate) {
            return false;
        }
        int selfId = _selfConfig().getId();
        if ((_lastVote.when + LastVote::leaseTime.total_milliseconds() >= now) 
            && (_lastVote.whoId != selfId)) {
            log() << "replSet not voting yea for " << selfId <<
                " voted for " << _lastVote.whoHostAndPort.toString() << ' ' << 
                (now - _lastVote.when) / 1000 << " secs ago";
            return false;
        }
        _lastVote.when = now;
        _lastVote.whoId = selfId;
        _lastVote.whoHostAndPort = _selfConfig().getHostAndPort();
        return true;
    }

    MemberState TopologyCoordinatorImpl::getMemberState() const {
        if (_selfIndex == -1) {
            if (_currentConfig.isInitialized()) {
                return MemberState::RS_REMOVED;
            }
            return MemberState::RS_STARTUP;
        }
        if (_role == Role::leader) {
            invariant(_currentPrimaryIndex == _selfIndex);
            return MemberState::RS_PRIMARY;
        }
        const MemberConfig& myConfig = _selfConfig();
        if (myConfig.isArbiter()) {
            return MemberState::RS_ARBITER;
        }
        if (((_maintenanceModeCalls > 0) || (_hasOnlyAuthErrorUpHeartbeats(_hbdata, _selfIndex)))
            && (_followerMode == MemberState::RS_SECONDARY)) {
            return MemberState::RS_RECOVERING;
        }
        return _followerMode;
    }

    void TopologyCoordinatorImpl::processWinElection(
            OID electionId,
            OpTime electionOpTime) {
        invariant(_role == Role::candidate);
        _electionTime = electionOpTime;
        _electionId = electionId;
        _role = Role::leader;
        _currentPrimaryIndex = _selfIndex;
    }

    void TopologyCoordinatorImpl::processLoseElection() {
        invariant(_role == Role::candidate);
        const HostAndPort syncSourceAddress = getSyncSourceAddress();
        _electionTime = OpTime(0, 0);
        _electionId = OID();
        _role = Role::follower;

        // Clear lastVote time, if we voted for ourselves in this election.
        // This will allow us to vote for others.
        if (_lastVote.whoId == _selfConfig().getId()) {
            _lastVote.when = 0;
        }
    }

    bool TopologyCoordinatorImpl::stepDown(Date_t until, bool force, OpTime lastOpApplied) {
        bool canStepDown = force;
        for (int i = 0; !canStepDown && i < _currentConfig.getNumMembers(); ++i) {
            if (i == _selfIndex) {
                continue;
            }
            UnelectableReasonMask reason = _getUnelectableReason(i, lastOpApplied);
            if (!reason && _hbdata[i].getOpTime() >= lastOpApplied) {
                canStepDown = true;
            }
        }

        if (!canStepDown) {
            return false;
        }
        _stepDownUntil = until;
        _stepDownSelfAndReplaceWith(-1);
        return true;
    }

    void TopologyCoordinatorImpl::setFollowerMode(MemberState::MS newMode) {
        invariant(_role == Role::follower);
        switch (newMode) {
        case MemberState::RS_RECOVERING:
        case MemberState::RS_ROLLBACK:
        case MemberState::RS_SECONDARY:
        case MemberState::RS_STARTUP2:
            _followerMode = newMode;
            break;
        default:
            invariant(false);
        }

        if (_followerMode != MemberState::RS_SECONDARY) {
            return;
        }

        // When a single node replica set transitions to SECONDARY, we must check if we should
        // be a candidate here.  This is necessary because a single node replica set has no
        // heartbeats that would normally change the role to candidate.

        if (_currentConfig.getNumMembers() == 1 &&
            _selfIndex == 0 &&
            _currentConfig.getMemberAt(_selfIndex).isElectable()) {
            _role = Role::candidate;
        }
    }

    bool TopologyCoordinatorImpl::stepDownIfPending() {
        if (!_stepDownPending) {
            return false;
        }

        int remotePrimaryIndex = -1;
        for (std::vector<MemberHeartbeatData>::const_iterator it = _hbdata.begin();
                it != _hbdata.end(); ++it) {
            const int itIndex = indexOfIterator(_hbdata, it);
            if (itIndex == _selfIndex) {
                continue;
            }

            if (it->getState().primary() && it->up()) {
                if (remotePrimaryIndex != -1) {
                    // two other nodes think they are primary (asynchronously polled)
                    // -- wait for things to settle down.
                    remotePrimaryIndex = -1;
                    log() << "replSet info two remote primaries (transiently)";
                    break;
                }
                remotePrimaryIndex = itIndex;
            }
        }
        _stepDownSelfAndReplaceWith(remotePrimaryIndex);
        return true;
    }

    void TopologyCoordinatorImpl::_stepDownSelfAndReplaceWith(int newPrimary) {
        invariant(_role == Role::leader);
        invariant(_selfIndex != -1);
        invariant(_selfIndex != newPrimary);
        invariant(_selfIndex == _currentPrimaryIndex);
        _currentPrimaryIndex = newPrimary;
        _role = Role::follower;
        _stepDownPending = false;
    }

    void TopologyCoordinatorImpl::adjustMaintenanceCountBy(int inc) {
        invariant(_role == Role::follower);
        _maintenanceModeCalls += inc;
        invariant(_maintenanceModeCalls >= 0);
    }

    int TopologyCoordinatorImpl::getMaintenanceCount() const {
        return _maintenanceModeCalls;
    }

    bool TopologyCoordinatorImpl::shouldChangeSyncSource(const HostAndPort& currentSource) const
    {
        // Methodology:
        // If there exists a viable sync source member other than currentSource, whose oplog has
        // reached an optime greater than _maxSyncSourceLagSecs later than currentSource's, return
        // true.

        // If the user requested a sync source change, return true.
        if (_forceSyncSourceIndex != -1) {
            return true;
        }

        const int currentMemberIndex = _currentConfig.findMemberIndexByHostAndPort(currentSource);
        if (currentMemberIndex == -1) {
            return true;
        }
        invariant(currentMemberIndex != _selfIndex);

        OpTime currentOpTime = _hbdata[currentMemberIndex].getOpTime();
        if (currentOpTime.isNull()) {
            // Haven't received a heartbeat from the sync source yet, so can't tell if we should
            // change.
            return false;
        }
        unsigned int currentSecs = currentOpTime.getSecs();
        unsigned int goalSecs = currentSecs + _maxSyncSourceLagSecs.total_seconds();

        for (std::vector<MemberHeartbeatData>::const_iterator it = _hbdata.begin();
             it != _hbdata.end();
             ++it) {
            const int itIndex = indexOfIterator(_hbdata, it);
            const MemberConfig& candidateConfig = _currentConfig.getMemberAt(itIndex);
            if (it->up() &&
                (candidateConfig.shouldBuildIndexes() || !_selfConfig().shouldBuildIndexes()) &&
                it->getState().readable() &&
                goalSecs < it->getOpTime().getSecs()) {
                log() << "changing sync target because current sync target's most recent OpTime is "
                      << currentOpTime.toStringLong() << " which is more than "
                      << _maxSyncSourceLagSecs.total_seconds() << " seconds behind member " 
                      <<  candidateConfig.getHostAndPort().toString()
                      << " whose most recent OpTime is " << it->getOpTime().toStringLong();
                invariant(itIndex != _selfIndex);
                return true;
            }
        }
        return false;
    }


} // namespace repl
} // namespace mongo