YugabyteDB (2.13.0.0-b42, bfc6a6643e7399ac8a0e81d06a3ee6d6571b33ab)

// Licensed to the Apache Software Foundation (ASF) under one

// or more contributor license agreements.  See the NOTICE file

// distributed with this work for additional information

// regarding copyright ownership.  The ASF licenses this file

// to you under the Apache License, Version 2.0 (the

// "License"); you may not use this file except in compliance

// with the License.  You may obtain a copy of the License at

//

//   http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing,

// software distributed under the License is distributed on an

// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY

// KIND, either express or implied.  See the License for the

// specific language governing permissions and limitations

// under the License.

//

// The following only applies to changes made to this file as part of YugaByte development.

//

// Portions Copyright (c) YugaByte, Inc.

//

// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except

// in compliance with the License.  You may obtain a copy of the License at

//

// http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing, software distributed under the License

// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express

// or implied.  See the License for the specific language governing permissions and limitations

// under the License.

//

#include <gtest/gtest.h>

#include "yb/common/schema.h"

#include "yb/common/wire_protocol-test-util.h"

#include "yb/consensus/consensus-test-util.h"

#include "yb/consensus/log.h"

#include "yb/consensus/log_index.h"

#include "yb/consensus/log_reader.h"

#include "yb/consensus/log_util.h"

#include "yb/consensus/opid_util.h"

#include "yb/consensus/peer_manager.h"

#include "yb/consensus/quorum_util.h"

#include "yb/consensus/raft_consensus.h"

#include "yb/consensus/replica_state.h"

#include "yb/gutil/bind.h"

#include "yb/gutil/stl_util.h"

#include "yb/gutil/strings/strcat.h"

#include "yb/gutil/strings/substitute.h"

#include "yb/rpc/messenger.h"

#include "yb/server/logical_clock.h"

#include "yb/util/mem_tracker.h"

#include "yb/util/metrics.h"

#include "yb/util/status_log.h"

#include "yb/util/test_macros.h"

#include "yb/util/test_util.h"

#include "yb/util/threadpool.h"

DECLARE_int32(raft_heartbeat_interval_ms);

DECLARE_bool(enable_leader_failure_detection);

METRIC_DECLARE_entity(table);

METRIC_DECLARE_entity(tablet);

#define REPLICATE_SEQUENCE_OF_MESSAGES(...) \

  ASSERT_NO_FATALS(ReplicateSequenceOfMessages(__VA_ARGS__))

using std::shared_ptr;

using std::unique_ptr;

namespace yb {

namespace consensus {

using log::Log;

using log::LogEntryPB;

using log::LogOptions;

using log::LogReader;

using rpc::RpcContext;

using strings::Substitute;

using strings::SubstituteAndAppend;

const char* kTestTable = "TestTable";

const char* kTestTablet = "TestTablet";

void DoNothing(std::shared_ptr<consensus::StateChangeContext> context) {

}

// Test suite for tests that focus on multiple peer interaction, but

// without integrating with other components, such as transactions.

class RaftConsensusQuorumTest : public YBTest {

 public:

  RaftConsensusQuorumTest()

    : clock_(server::LogicalClock::CreateStartingAt(HybridTime(0))),

      table_metric_entity_(

          METRIC_ENTITY_table.Instantiate(&metric_registry_, "raft-test-table")),

      tablet_metric_entity_(

          METRIC_ENTITY_tablet.Instantiate(&metric_registry_, "raft-test-tablet")),

      schema_(GetSimpleTestSchema()) {

    options_.tablet_id = kTestTablet;

    FLAGS_enable_leader_failure_detection = false;

  }

  // Builds an initial configuration of 'num' elements.

  // All of the peers start as followers.

  void BuildInitialRaftConfigPB(int num) {

    config_ = BuildRaftConfigPBForTests(num);

    config_.set_opid_index(kInvalidOpIdIndex);

    peers_.reset(new TestPeerMapManager(config_));

  }

  Status BuildFsManagersAndLogs() {

    // Build the fsmanagers and logs

    for (int i = 0; i < config_.peers_size(); i++) {

      shared_ptr<MemTracker> parent_mem_tracker =

          MemTracker::CreateTracker(Substitute("peer-$0", i));

      parent_mem_trackers_.push_back(parent_mem_tracker);

      string test_path = GetTestPath(Substitute("peer-$0-root", i));

      FsManagerOpts opts;

      opts.parent_mem_tracker = parent_mem_tracker;

      opts.wal_paths = { test_path };

      opts.data_paths = { test_path };

      opts.server_type = "tserver_test";

      std::unique_ptr<FsManager> fs_manager(new FsManager(env_.get(), opts));

      RETURN_NOT_OK(fs_manager->CreateInitialFileSystemLayout());

      RETURN_NOT_OK(fs_manager->Open());

      scoped_refptr<Log> log;

      RETURN_NOT_OK(Log::Open(LogOptions(),

                              kTestTablet,

                              fs_manager->GetFirstTabletWalDirOrDie(kTestTable, kTestTablet),

                              fs_manager->uuid(),

                              schema_,

                              0, // schema_version

                              nullptr, // table_metric_entity

                              nullptr, // tablet_metric_entity

                              log_thread_pool_.get(),

                              log_thread_pool_.get(),

                              std::numeric_limits<int64_t>::max(), // cdc_min_replicated_index

                              &log));

      logs_.push_back(log.get());

      fs_managers_.push_back(fs_manager.release());

    }

    return Status::OK();

  }

  void BuildPeers() {

    vector<LocalTestPeerProxyFactory*> proxy_factories;

    for (int i = 0; i < config_.peers_size(); i++) {

      auto proxy_factory = std::make_unique<LocalTestPeerProxyFactory>(peers_.get());

      proxy_factories.push_back(proxy_factory.get());

      auto operation_factory = new TestOperationFactory();

      string peer_uuid = Substitute("peer-$0", i);

      std::unique_ptr<ConsensusMetadata> cmeta;

      ASSERT_OK(ConsensusMetadata::Create(fs_managers_[i], kTestTablet, peer_uuid, config_,

                                         kMinimumTerm, &cmeta));

      RaftPeerPB local_peer_pb;

      ASSERT_OK(GetRaftConfigMember(config_, peer_uuid, &local_peer_pb));

      auto queue = std::make_unique<PeerMessageQueue>(

          tablet_metric_entity_,

          logs_[i],

          MemTracker::FindOrCreateTracker(peer_uuid),

          MemTracker::FindOrCreateTracker(peer_uuid),

          local_peer_pb,

          kTestTablet,

          clock_,

          nullptr /* consensus_context */,

          raft_pool_->NewToken(ThreadPool::ExecutionMode::SERIAL));

      unique_ptr<ThreadPoolToken> pool_token(

          raft_pool_->NewToken(ThreadPool::ExecutionMode::CONCURRENT));

      auto peer_manager = std::make_unique<PeerManager>(

          options_.tablet_id,

          config_.peers(i).permanent_uuid(),

          proxy_factory.get(),

          queue.get(),

          pool_token.get(),

          nullptr);

      shared_ptr<RaftConsensus> peer(new RaftConsensus(

          options_,

          std::move(cmeta),

          std::move(proxy_factory),

          std::move(queue),

          std::move(peer_manager),

          std::move(pool_token),

          table_metric_entity_,

          tablet_metric_entity_,

          config_.peers(i).permanent_uuid(),

          clock_,

          operation_factory,

          logs_[i],

          parent_mem_trackers_[i],

          Bind(&DoNothing),

          DEFAULT_TABLE_TYPE,

          nullptr /* retryable_requests */));

      operation_factory->SetConsensus(peer.get());

      operation_factories_.emplace_back(operation_factory);

      peers_->AddPeer(config_.peers(i).permanent_uuid(), peer);

    }

  }

  Status StartPeers() {

    ConsensusBootstrapInfo boot_info;

    TestPeerMap all_peers = peers_->GetPeerMapCopy();

    for (const TestPeerMap::value_type& entry : all_peers) {

      RETURN_NOT_OK(entry.second->Start(boot_info));

    }

    return Status::OK();

  }

  Status BuildConfig(int num) {

    RETURN_NOT_OK(ThreadPoolBuilder("raft").Build(&raft_pool_));

    RETURN_NOT_OK(ThreadPoolBuilder("log").Build(&log_thread_pool_));

    BuildInitialRaftConfigPB(num);

    RETURN_NOT_OK(BuildFsManagersAndLogs());

    BuildPeers();

    return Status::OK();

  }

  Status BuildAndStartConfig(int num) {

    RETURN_NOT_OK(BuildConfig(num));

    RETURN_NOT_OK(StartPeers());

    // Automatically elect the last node in the list.

    const int kLeaderIdx = num - 1;

    shared_ptr<RaftConsensus> leader;

    RETURN_NOT_OK(peers_->GetPeerByIdx(kLeaderIdx, &leader));

    RETURN_NOT_OK(leader->EmulateElection());

    RETURN_NOT_OK(leader->WaitUntilLeaderForTests(MonoDelta::FromSeconds(10)));

    return Status::OK();

  }

  LocalTestPeerProxy* GetLeaderProxyToPeer(int peer_idx, int leader_idx) {

    shared_ptr<RaftConsensus> follower;

    CHECK_OK(peers_->GetPeerByIdx(peer_idx, &follower));

    shared_ptr<RaftConsensus> leader;

    CHECK_OK(peers_->GetPeerByIdx(leader_idx, &leader));

    for (LocalTestPeerProxy* proxy : down_cast<LocalTestPeerProxyFactory*>(

        leader->peer_proxy_factory_.get())->GetProxies()) {

      if (proxy->GetTarget() == follower->peer_uuid()) {

        return proxy;

      }

    }

    CHECK(false) << "Proxy not found";

    return nullptr;

  }

  Status AppendDummyMessage(int peer_idx,

                            scoped_refptr<ConsensusRound>* round) {

    auto msg = std::make_shared<ReplicateMsg>();

    msg->set_op_type(NO_OP);

    msg->mutable_noop_request();

    msg->set_hybrid_time(clock_->Now().ToUint64());

    shared_ptr<RaftConsensus> peer;

    CHECK_OK(peers_->GetPeerByIdx(peer_idx, &peer));

    // Use a latch in place of a Transaction callback.

    auto sync = std::make_unique<Synchronizer>();

    *round = make_scoped_refptr<ConsensusRound>(peer.get(), std::move(msg));

    (**round).SetCallback(MakeNonTrackedRoundCallback(

        round->get(),

        [sync = sync.get()](const Status& status) {

      sync->StatusCB(status);

    }));

    (**round).BindToTerm(peer->LeaderTerm());

    InsertOrDie(&syncs_, round->get(), sync.release());

    RETURN_NOT_OK_PREPEND(peer->TEST_Replicate(round->get()),

                          Substitute("Unable to replicate to peer $0", peer_idx));

    return Status::OK();

  }

  Status WaitForReplicate(ConsensusRound* round) {

    return FindOrDie(syncs_, round)->Wait();

  }

  Status TimedWaitForReplicate(ConsensusRound* round, const MonoDelta& delta) {

    return FindOrDie(syncs_, round)->WaitFor(delta);

  }

  void WaitForReplicateIfNotAlreadyPresent(const OpIdPB& to_wait_for, int peer_idx) {

    shared_ptr<RaftConsensus> peer;

    ASSERT_OK(peers_->GetPeerByIdx(peer_idx, &peer));

    ReplicaState* state = peer->GetReplicaStateForTests();

    while (true) {

      {

        auto lock = state->LockForRead();

        if (state->GetLastReceivedOpIdUnlocked().index >= to_wait_for.index()) {

          return;

        }

      }

      SleepFor(MonoDelta::FromMilliseconds(1));

    }

  }

  // Waits for an operation to be (database) committed in the replica at index

  // 'peer_idx'. If the operation was already committed this returns immediately.

  void WaitForCommitIfNotAlreadyPresent(const OpIdPB& to_wait_for,

                                        int peer_idx,

                                        int leader_idx) {

    MonoDelta timeout(MonoDelta::FromSeconds(10));

    MonoTime start(MonoTime::Now());

    shared_ptr<RaftConsensus> peer;

    ASSERT_OK(peers_->GetPeerByIdx(peer_idx, &peer));

    ReplicaState* state = peer->GetReplicaStateForTests();

    int backoff_exp = 0;

    const int kMaxBackoffExp = 8;

    OpIdPB committed_op_id;

    while (true) {

      {

        auto lock = state->LockForRead();

        state->GetCommittedOpIdUnlocked().ToPB(&committed_op_id);

        if (OpIdCompare(committed_op_id, to_wait_for) >= 0) {

          return;

        }

      }

      MonoDelta elapsed = MonoTime::Now().GetDeltaSince(start);

      if (elapsed.MoreThan(timeout)) {

        break;

      }

      SleepFor(MonoDelta::FromMilliseconds(1 << backoff_exp));

      backoff_exp = std::min(backoff_exp + 1, kMaxBackoffExp);

    }

    LOG(ERROR) << "Max timeout reached (" << timeout.ToString() << ") while waiting for commit of "

               << "op " << to_wait_for << " on replica. Last committed op on replica: "

               << committed_op_id << ". Dumping state and quitting.";

    vector<string> lines;

    shared_ptr<RaftConsensus> leader;

    ASSERT_OK(peers_->GetPeerByIdx(leader_idx, &leader));

    for (const string& line : lines) {

      LOG(ERROR) << line;

    }

    // Gather the replica and leader operations for printing

    log::LogEntries replica_ops = GatherLogEntries(peer_idx, logs_[peer_idx]);

    log::LogEntries leader_ops = GatherLogEntries(leader_idx, logs_[leader_idx]);

    SCOPED_TRACE(PrintOnError(replica_ops, Substitute("local peer ($0)", peer->peer_uuid())));

    SCOPED_TRACE(PrintOnError(leader_ops, Substitute("leader (peer-$0)", leader_idx)));

    FAIL() << "Replica did not commit.";

  }

  // Used in ReplicateSequenceOfMessages() to specify whether

  // we should wait for all replicas to have replicated the

  // sequence or just a majority.

  enum ReplicateWaitMode {

    WAIT_FOR_ALL_REPLICAS,

    WAIT_FOR_MAJORITY

  };

  // Used in ReplicateSequenceOfMessages() to specify whether

  // we should also commit the messages in the sequence

  enum CommitMode {

    DONT_COMMIT,

    COMMIT_ONE_BY_ONE

  };

  // Replicates a sequence of messages to the peer passed as leader.

  // Optionally waits for the messages to be replicated to followers.

  // 'last_op_id' is set to the id of the last replicated operation.

  // The operations are only committed if 'commit_one_by_one' is true.

  void ReplicateSequenceOfMessages(int seq_size,

                                   int leader_idx,

                                   ReplicateWaitMode wait_mode,

                                   CommitMode commit_mode,

                                   OpIdPB* last_op_id,

                                   vector<scoped_refptr<ConsensusRound> >* rounds) {

    for (int i = 0; i < seq_size; i++) {

      scoped_refptr<ConsensusRound> round;

      ASSERT_OK(AppendDummyMessage(leader_idx, &round));

      ASSERT_OK(WaitForReplicate(round.get()));

      round->id().ToPB(last_op_id);

      rounds->push_back(round);

    }

    if (wait_mode == WAIT_FOR_ALL_REPLICAS) {

      shared_ptr<RaftConsensus> leader;

      ASSERT_OK(peers_->GetPeerByIdx(leader_idx, &leader));

      TestPeerMap all_peers = peers_->GetPeerMapCopy();

      int i = 0;

      for (const TestPeerMap::value_type& entry : all_peers) {

        if (entry.second->peer_uuid() != leader->peer_uuid()) {

          WaitForReplicateIfNotAlreadyPresent(*last_op_id, i);

        }

        i++;

      }

    }

  }

  log::LogEntries GatherLogEntries(int idx, const scoped_refptr<Log>& log) {

    EXPECT_OK(log->WaitUntilAllFlushed());

    EXPECT_OK(log->Close());

    std::unique_ptr<LogReader> log_reader;

    EXPECT_OK(log::LogReader::Open(fs_managers_[idx]->env(),

                                   scoped_refptr<log::LogIndex>(),

                                   "Log reader: ",

                                   fs_managers_[idx]->GetFirstTabletWalDirOrDie(kTestTable,

                                                                                kTestTablet),

                                   table_metric_entity_.get(),

                                   tablet_metric_entity_.get(),

                                   &log_reader));

    log::LogEntries ret;

    log::SegmentSequence segments;

    EXPECT_OK(log_reader->GetSegmentsSnapshot(&segments));

    for (const log::SegmentSequence::value_type& entry : segments) {

      auto result = entry->ReadEntries();

      EXPECT_OK(result.status);

      for (auto& e : result.entries) {

        ret.push_back(std::move(e));

      }

    }

    return ret;

  }

  // Verifies that the replica's log match the leader's. This deletes the

  // peers (so we're sure that no further writes occur) and closes the logs

  // so it must be the very last thing to run, in a test.

  void VerifyLogs(int leader_idx, int first_replica_idx, int last_replica_idx) {

    // Wait for in-flight transactions to be done. We're destroying the

    // peers next and leader transactions won't be able to commit anymore.

    for (const auto& factory : operation_factories_) {

      factory->WaitDone();

    }

    // Shut down all the peers.

    TestPeerMap all_peers = peers_->GetPeerMapCopy();

    for (const TestPeerMap::value_type& entry : all_peers) {

      entry.second->Shutdown();

    }

    log::LogEntries leader_entries = GatherLogEntries(leader_idx, logs_[leader_idx]);

    shared_ptr<RaftConsensus> leader;

    ASSERT_OK(peers_->GetPeerByIdx(leader_idx, &leader));

    for (int replica_idx = first_replica_idx; replica_idx < last_replica_idx; replica_idx++) {

      log::LogEntries replica_entries = GatherLogEntries(replica_idx, logs_[replica_idx]);

      shared_ptr<RaftConsensus> replica;

      ASSERT_OK(peers_->GetPeerByIdx(replica_idx, &replica));

      VerifyReplica(leader_entries,

                    replica_entries,

                    leader->peer_uuid(),

                    replica->peer_uuid());

    }

  }

  std::vector<OpIdPB> ExtractReplicateIds(const log::LogEntries& entries) {

    std::vector<OpIdPB> result;

    result.reserve(entries.size() / 2);

    for (const auto& entry : entries) {

      if (entry->has_replicate()) {

        result.push_back(entry->replicate().id());

      }

    }

    return result;

  }

  void VerifyReplicateOrderMatches(const log::LogEntries& leader_entries,

                                   const log::LogEntries& replica_entries) {

    auto leader_ids = ExtractReplicateIds(leader_entries);

    auto replica_ids = ExtractReplicateIds(replica_entries);

    ASSERT_EQ(leader_ids.size(), replica_ids.size());

    for (size_t i = 0; i < leader_ids.size(); i++) {

      ASSERT_EQ(leader_ids[i].ShortDebugString(),

                replica_ids[i].ShortDebugString());

    }

  }

  void VerifyNoCommitsBeforeReplicates(const log::LogEntries& entries) {

    std::unordered_set<OpIdPB, OpIdHashFunctor, OpIdEqualsFunctor> replication_ops;

    for (const auto& entry : entries) {

      if (entry->has_replicate()) {

        ASSERT_TRUE(InsertIfNotPresent(&replication_ops, entry->replicate().id()))

          << "REPLICATE op id showed up twice: " << entry->ShortDebugString();

      }

    }

  }

  void VerifyReplica(const log::LogEntries& leader_entries,

                     const log::LogEntries& replica_entries,

                     const string& leader_name,

                     const string& replica_name) {

    SCOPED_TRACE(PrintOnError(leader_entries, Substitute("Leader: $0", leader_name)));

    SCOPED_TRACE(PrintOnError(replica_entries, Substitute("Replica: $0", replica_name)));

    // Check that the REPLICATE messages come in the same order on both nodes.

    VerifyReplicateOrderMatches(leader_entries, replica_entries);

    // Check that no COMMIT precedes its related REPLICATE on both the replica

    // and leader.

    VerifyNoCommitsBeforeReplicates(replica_entries);

    VerifyNoCommitsBeforeReplicates(leader_entries);

  }

  string PrintOnError(const log::LogEntries& replica_entries,

                      const string& replica_id) {

    string ret = "";

    SubstituteAndAppend(&ret, "$1 log entries for replica $0:\n",

                        replica_id, replica_entries.size());

    for (const auto& replica_entry : replica_entries) {

      StrAppend(&ret, "Replica log entry: ", replica_entry->ShortDebugString(), "\n");

    }

    return ret;

  }

  // Read the ConsensusMetadata for the given peer from disk.

  std::unique_ptr<ConsensusMetadata> ReadConsensusMetadataFromDisk(int peer_index) {

    string peer_uuid = Substitute("peer-$0", peer_index);

    std::unique_ptr<ConsensusMetadata> cmeta;

    CHECK_OK(ConsensusMetadata::Load(fs_managers_[peer_index], kTestTablet, peer_uuid, &cmeta));

    return cmeta;

  }

  // Assert that the durable term == term and that the peer that got the vote == voted_for.

  void AssertDurableTermAndVote(int peer_index, int64_t term, const std::string& voted_for) {

    auto cmeta = ReadConsensusMetadataFromDisk(peer_index);

    ASSERT_EQ(term, cmeta->current_term());

    ASSERT_EQ(voted_for, cmeta->voted_for());

  }

  // Assert that the durable term == term and that the peer has not yet voted.

  void AssertDurableTermWithoutVote(int peer_index, int64_t term) {

    auto cmeta = ReadConsensusMetadataFromDisk(peer_index);

    ASSERT_EQ(term, cmeta->current_term());

    ASSERT_FALSE(cmeta->has_voted_for());

  }

  ~RaftConsensusQuorumTest() {

    peers_->Clear();

    operation_factories_.clear();

    // We need to clear the logs before deleting the fs_managers_ or we'll

    // get a SIGSEGV when closing the logs.

    logs_.clear();

    STLDeleteElements(&fs_managers_);

    STLDeleteValues(&syncs_);

  }

 protected:

  ConsensusOptions options_;

  RaftConfigPB config_;

  OpIdPB initial_id_;

  vector<shared_ptr<MemTracker> > parent_mem_trackers_;

  vector<FsManager*> fs_managers_;

  vector<scoped_refptr<Log> > logs_;

  unique_ptr<ThreadPool> raft_pool_;

  unique_ptr<ThreadPool> log_thread_pool_;

  std::unique_ptr<TestPeerMapManager> peers_;

  std::vector<std::unique_ptr<TestOperationFactory>> operation_factories_;

  scoped_refptr<server::Clock> clock_;

  MetricRegistry metric_registry_;

  scoped_refptr<MetricEntity> table_metric_entity_;

  scoped_refptr<MetricEntity> tablet_metric_entity_;

  const Schema schema_;

  std::unordered_map<ConsensusRound*, Synchronizer*> syncs_;

};

TEST_F(RaftConsensusQuorumTest, TestConsensusContinuesIfAMinorityFallsBehind) {

  // Constants with the indexes of peers with certain roles,

  // since peers don't change roles in this test.

  const int kFollower0Idx = 0;

  const int kFollower1Idx = 1;

  const int kLeaderIdx = 2;

  ASSERT_OK(BuildAndStartConfig(3));

  OpIdPB last_replicate;

  vector<scoped_refptr<ConsensusRound> > rounds;

  {

    // lock one of the replicas down by obtaining the state lock

    // and never letting it go.

    shared_ptr<RaftConsensus> follower0;

    ASSERT_OK(peers_->GetPeerByIdx(kFollower0Idx, &follower0));

    ReplicaState* follower0_rs = follower0->GetReplicaStateForTests();

    auto lock = follower0_rs->LockForRead();

    // If the locked replica would stop consensus we would hang here

    // as we wait for operations to be replicated to a majority.

    ASSERT_NO_FATALS(ReplicateSequenceOfMessages(

                              10,

                              kLeaderIdx,

                              WAIT_FOR_MAJORITY,

                              COMMIT_ONE_BY_ONE,

                              &last_replicate,

                              &rounds));

    // Follower 1 should be fine (Were we to wait for follower0's replicate

    // this would hang here). We know he must have replicated but make sure

    // by calling Wait().

    WaitForReplicateIfNotAlreadyPresent(last_replicate, kFollower1Idx);

    WaitForCommitIfNotAlreadyPresent(last_replicate, kFollower1Idx, kLeaderIdx);

  }

  // After we let the lock go the remaining follower should get up-to-date

  WaitForReplicateIfNotAlreadyPresent(last_replicate, kFollower0Idx);

  WaitForCommitIfNotAlreadyPresent(last_replicate, kFollower0Idx, kLeaderIdx);

  VerifyLogs(2, 0, 1);

}

TEST_F(RaftConsensusQuorumTest, TestConsensusStopsIfAMajorityFallsBehind) {

  // Constants with the indexes of peers with certain roles,

  // since peers don't change roles in this test.

  const int kFollower0Idx = 0;

  const int kFollower1Idx = 1;

  const int kLeaderIdx = 2;

  ASSERT_OK(BuildAndStartConfig(3));

  OpIdPB last_op_id;

  scoped_refptr<ConsensusRound> round;

  {

    // lock two of the replicas down by obtaining the state locks

    // and never letting them go.

    shared_ptr<RaftConsensus> follower0;

    ASSERT_OK(peers_->GetPeerByIdx(kFollower0Idx, &follower0));

    ReplicaState* follower0_rs = follower0->GetReplicaStateForTests();

    auto lock0 = follower0_rs->LockForRead();

    shared_ptr<RaftConsensus> follower1;

    ASSERT_OK(peers_->GetPeerByIdx(kFollower1Idx, &follower1));

    ReplicaState* follower1_rs = follower1->GetReplicaStateForTests();

    auto lock1 = follower1_rs->LockForRead();

    // Append a single message to the queue

    ASSERT_OK(AppendDummyMessage(kLeaderIdx, &round));

    round->id().ToPB(&last_op_id);

    // This should timeout.

    Status status = TimedWaitForReplicate(round.get(), MonoDelta::FromMilliseconds(500));

    ASSERT_TRUE(status.IsTimedOut());

  }

  // After we release the locks the operation should replicate to all replicas

  // and we commit.

  ASSERT_OK(WaitForReplicate(round.get()));

  // Assert that everything was ok

  WaitForReplicateIfNotAlreadyPresent(last_op_id, kFollower0Idx);

  WaitForReplicateIfNotAlreadyPresent(last_op_id, kFollower1Idx);

  WaitForCommitIfNotAlreadyPresent(last_op_id, kFollower0Idx, kLeaderIdx);

  WaitForCommitIfNotAlreadyPresent(last_op_id, kFollower1Idx, kLeaderIdx);

  VerifyLogs(2, 0, 1);

}

// If some communication error happens the leader will resend the request to the

// peers. This tests that the peers handle repeated requests.

TEST_F(RaftConsensusQuorumTest, TestReplicasHandleCommunicationErrors) {

  // Constants with the indexes of peers with certain roles,

  // since peers don't change roles in this test.

  const int kFollower0Idx = 0;

  const int kFollower1Idx = 1;

  const int kLeaderIdx = 2;

  ASSERT_OK(BuildAndStartConfig(3));

  OpIdPB last_op_id;

  // Append a dummy message, with faults injected on the first attempt

  // to send the message.

  scoped_refptr<ConsensusRound> round;

  GetLeaderProxyToPeer(kFollower0Idx, kLeaderIdx)->InjectCommFaultLeaderSide();

  GetLeaderProxyToPeer(kFollower1Idx, kLeaderIdx)->InjectCommFaultLeaderSide();

  ASSERT_OK(AppendDummyMessage(kLeaderIdx, &round));

  // We should successfully replicate it due to retries.

  ASSERT_OK(WaitForReplicate(round.get()));

  GetLeaderProxyToPeer(kFollower0Idx, kLeaderIdx)->InjectCommFaultLeaderSide();

  GetLeaderProxyToPeer(kFollower1Idx, kLeaderIdx)->InjectCommFaultLeaderSide();

  // The commit should eventually reach both followers as well.

  round->id().ToPB(&last_op_id);

  WaitForCommitIfNotAlreadyPresent(last_op_id, kFollower0Idx, kLeaderIdx);

  WaitForCommitIfNotAlreadyPresent(last_op_id, kFollower1Idx, kLeaderIdx);

  // Append a sequence of messages, and keep injecting errors into the

  // replica proxies.

  vector<scoped_refptr<ConsensusRound> > rounds;

  for (int i = 0; i < 100; i++) {

    scoped_refptr<ConsensusRound> round;

    ASSERT_OK(AppendDummyMessage(kLeaderIdx, &round));

    ConsensusRound* round_ptr = round.get();

    round->id().ToPB(&last_op_id);

    rounds.push_back(round);

    // inject comm faults

    if (i % 2 == 0) {

      GetLeaderProxyToPeer(kFollower0Idx, kLeaderIdx)->InjectCommFaultLeaderSide();

    } else {

      GetLeaderProxyToPeer(kFollower1Idx, kLeaderIdx)->InjectCommFaultLeaderSide();

    }

    ASSERT_OK(WaitForReplicate(round_ptr));

  }

  // Assert last operation was correctly replicated and committed.

  WaitForReplicateIfNotAlreadyPresent(last_op_id, kFollower0Idx);

  WaitForReplicateIfNotAlreadyPresent(last_op_id, kFollower1Idx);

  // See comment at the end of TestFollowersReplicateAndCommitMessage

  // for an explanation on this waiting sequence.

  WaitForCommitIfNotAlreadyPresent(last_op_id, kFollower0Idx, kLeaderIdx);

  WaitForCommitIfNotAlreadyPresent(last_op_id, kFollower1Idx, kLeaderIdx);

  VerifyLogs(2, 0, 1);

}

// In this test we test the ability of the leader to send heartbeats

// to replicas by simply pushing nothing after the configuration round

// and still expecting for the replicas Update() hooks to be called.

TEST_F(RaftConsensusQuorumTest, TestLeaderHeartbeats) {

  // Constants with the indexes of peers with certain roles,

  // since peers don't change roles in this test.

  const int kFollower0Idx = 0;

  const int kFollower1Idx = 1;

  const int kLeaderIdx = 2;

  ASSERT_OK(BuildConfig(3));

  shared_ptr<RaftConsensus> follower0;

  ASSERT_OK(peers_->GetPeerByIdx(kFollower0Idx, &follower0));

  shared_ptr<RaftConsensus> follower1;

  ASSERT_OK(peers_->GetPeerByIdx(kFollower1Idx, &follower1));

  shared_ptr<CounterHooks> counter_hook_rpl0(

      new CounterHooks(follower0->GetFaultHooks()));

  shared_ptr<CounterHooks> counter_hook_rpl1(

      new CounterHooks(follower1->GetFaultHooks()));

  // Replace the default fault hooks on the replicas with counter hooks

  // before we start the configuration.

  follower0->SetFaultHooks(counter_hook_rpl0);

  follower1->SetFaultHooks(counter_hook_rpl1);

  ASSERT_OK(StartPeers());

  shared_ptr<RaftConsensus> leader;

  ASSERT_OK(peers_->GetPeerByIdx(kLeaderIdx, &leader));

  ASSERT_OK(leader->EmulateElection());

  // Wait for the config round to get committed and count the number

  // of update calls, calls after that will be heartbeats.

  OpIdPB config_round;

  config_round.set_term(1);

  config_round.set_index(1);

  WaitForCommitIfNotAlreadyPresent(config_round, kFollower0Idx, kLeaderIdx);

  WaitForCommitIfNotAlreadyPresent(config_round, kFollower1Idx, kLeaderIdx);

  int repl0_init_count = counter_hook_rpl0->num_pre_update_calls();

  int repl1_init_count = counter_hook_rpl1->num_pre_update_calls();

  // Now wait for about 4 times the heartbeat period the counters

  // should have increased between 3 to 8 times.

  //

  // Why the variance? Heartbeat timing is jittered such that the period

  // between heartbeats can be anywhere from half the interval to the full interval.

  SleepFor(MonoDelta::FromMilliseconds(FLAGS_raft_heartbeat_interval_ms * 4));

  int repl0_final_count = counter_hook_rpl0->num_pre_update_calls();

  int repl1_final_count = counter_hook_rpl1->num_pre_update_calls();

  ASSERT_GE(repl0_final_count - repl0_init_count, 3);

  ASSERT_LE(repl0_final_count - repl0_init_count, 8);

  ASSERT_GE(repl1_final_count - repl1_init_count, 3);

  ASSERT_LE(repl1_final_count - repl1_init_count, 8);

  VerifyLogs(2, 0, 1);

}

// After creating the initial configuration, this test writes a small sequence

// of messages to the initial leader. It then shuts down the current

// leader, makes another peer become leader and writes a sequence of

// messages to it. The new leader and the follower should agree on the

// sequence of messages.

TEST_F(RaftConsensusQuorumTest, TestLeaderElectionWithQuiescedQuorum) {

  const int kInitialNumPeers = 5;

  ASSERT_OK(BuildAndStartConfig(kInitialNumPeers));

  OpIdPB last_op_id;

  vector<scoped_refptr<ConsensusRound> > rounds;

  // Loop twice, successively shutting down the previous leader.

  for (int current_config_size = kInitialNumPeers;

       current_config_size >= kInitialNumPeers - 1;

       current_config_size--) {

    REPLICATE_SEQUENCE_OF_MESSAGES(10,

                                   current_config_size - 1, // The index of the leader.

                                   WAIT_FOR_ALL_REPLICAS,

                                   COMMIT_ONE_BY_ONE,

                                   &last_op_id,

                                   &rounds);

    // Make sure the last operation is committed everywhere

    for (int i = 0; i < current_config_size - 1; i++) {

      WaitForCommitIfNotAlreadyPresent(last_op_id, i, current_config_size - 1);

    }

    // Now shutdown the current leader.

    LOG(INFO) << "Shutting down current leader with index " << (current_config_size - 1);

    shared_ptr<RaftConsensus> current_leader;

    ASSERT_OK(peers_->GetPeerByIdx(current_config_size - 1, &current_leader));

    current_leader->Shutdown();

    peers_->RemovePeer(current_leader->peer_uuid());

    // ... and make the peer before it become leader.

    shared_ptr<RaftConsensus> new_leader;

    ASSERT_OK(peers_->GetPeerByIdx(current_config_size - 2, &new_leader));

    // This will force an election in which we expect to make the last

    // non-shutdown peer in the list become leader.

    LOG(INFO) << "Running election for future leader with index " << (current_config_size - 1);

    ASSERT_OK(new_leader->StartElection({ElectionMode::ELECT_EVEN_IF_LEADER_IS_ALIVE}));

    ASSERT_OK(new_leader->WaitUntilLeaderForTests(MonoDelta::FromSeconds(15)));

    LOG(INFO) << "Election won";

    // ... replicating a set of messages to the new leader should now be possible.

    REPLICATE_SEQUENCE_OF_MESSAGES(10,

                                   current_config_size - 2, // The index of the new leader.

                                   WAIT_FOR_MAJORITY,

                                   COMMIT_ONE_BY_ONE,

                                   &last_op_id,

                                   &rounds);

    // Make sure the last operation is committed everywhere

    for (int i = 0; i < current_config_size - 2; i++) {

      WaitForCommitIfNotAlreadyPresent(last_op_id, i, current_config_size - 2);

    }

  }

  // We can only verify the logs of the peers that were not killed, due to the

  // old leaders being out-of-date now.

  VerifyLogs(2, 0, 1);

}

TEST_F(RaftConsensusQuorumTest, TestReplicasEnforceTheLogMatchingProperty) {

  ASSERT_OK(BuildAndStartConfig(3));

  OpIdPB last_op_id;

  vector<scoped_refptr<ConsensusRound> > rounds;

  REPLICATE_SEQUENCE_OF_MESSAGES(10,

                                 2, // The index of the initial leader.

                                 WAIT_FOR_ALL_REPLICAS,

                                 COMMIT_ONE_BY_ONE,

                                 &last_op_id,

                                 &rounds);

  // Make sure the last operation is committed everywhere

  WaitForCommitIfNotAlreadyPresent(last_op_id, 0, 2);

  WaitForCommitIfNotAlreadyPresent(last_op_id, 1, 2);

  // Now replicas should only accept operations with

  // 'last_id' as the preceding id.

  ConsensusRequestPB req;

  ConsensusResponsePB resp;

  shared_ptr<RaftConsensus> leader;

  ASSERT_OK(peers_->GetPeerByIdx(2, &leader));

  shared_ptr<RaftConsensus> follower;

  ASSERT_OK(peers_->GetPeerByIdx(0, &follower));

  req.set_caller_uuid(leader->peer_uuid());

  req.set_caller_term(last_op_id.term());

  req.mutable_preceding_id()->CopyFrom(last_op_id);

  req.mutable_committed_op_id()->CopyFrom(last_op_id);

  ReplicateMsg* replicate = req.add_ops();

  replicate->set_hybrid_time(clock_->Now().ToUint64());

  OpIdPB* id = replicate->mutable_id();

  id->set_term(last_op_id.term());

  id->set_index(last_op_id.index() + 1);

  // Make a copy of the OpId to be TSAN friendly.

  auto req_copy = req;

  auto id_copy = req_copy.mutable_ops(0)->mutable_id();

  replicate->set_op_type(NO_OP);

  // Appending this message to peer0 should work and update

  // its 'last_received' to 'id'.

  ASSERT_OK(follower->Update(&req, &resp, CoarseBigDeadline()));

  ASSERT_TRUE(OpIdEquals(resp.status().last_received(), *id));

  // Now skip one message in the same term. The replica should

  // complain with the right error message.

  req_copy.mutable_preceding_id()->set_index(id_copy->index() + 1);

  id_copy->set_index(id_copy->index() + 2);

  // Appending this message to peer0 should return a Status::OK

  // but should contain an error referring to the log matching property.

  ASSERT_OK(follower->Update(&req_copy, &resp, CoarseBigDeadline()));

  ASSERT_TRUE(resp.has_status());

  ASSERT_TRUE(resp.status().has_error());

  ASSERT_EQ(resp.status().error().code(), ConsensusErrorPB::PRECEDING_ENTRY_DIDNT_MATCH);

  ASSERT_STR_CONTAINS(resp.status().error().status().message(),

                      "Log matching property violated");

}

// Test that RequestVote performs according to "spec".

TEST_F(RaftConsensusQuorumTest, TestRequestVote) {

  ASSERT_OK(BuildAndStartConfig(3));

  OpIdPB last_op_id;

  vector<scoped_refptr<ConsensusRound> > rounds;

  REPLICATE_SEQUENCE_OF_MESSAGES(10,

                                 2, // The index of the initial leader.

                                 WAIT_FOR_ALL_REPLICAS,

                                 COMMIT_ONE_BY_ONE,

                                 &last_op_id,

                                 &rounds);

  // Make sure the last operation is committed everywhere

  WaitForCommitIfNotAlreadyPresent(last_op_id, 0, 2);

  WaitForCommitIfNotAlreadyPresent(last_op_id, 1, 2);

  // Ensure last-logged OpId is > (0,0).

  ASSERT_TRUE(OpIdLessThan(MinimumOpId(), last_op_id));

  const int kPeerIndex = 1;

  shared_ptr<RaftConsensus> peer;

  ASSERT_OK(peers_->GetPeerByIdx(kPeerIndex, &peer));

  VoteRequestPB request;

  request.set_tablet_id(kTestTablet);

  request.mutable_candidate_status()->mutable_last_received()->CopyFrom(last_op_id);

  // Test that the replica won't vote since it has recently heard from

  // a valid leader.

  VoteResponsePB response;

  request.set_candidate_uuid("peer-0");

  request.set_candidate_term(last_op_id.term() + 1);

  ASSERT_OK(peer->RequestVote(&request, &response));

  ASSERT_FALSE(response.vote_granted());

  ASSERT_EQ(ConsensusErrorPB::LEADER_IS_ALIVE, response.consensus_error().code());

  // Test that replicas only vote yes for a single peer per term.

  // Indicate that replicas should vote even if they think another leader is alive.

  // This will allow the rest of the requests in the test to go through.

  request.set_ignore_live_leader(true);

  ASSERT_OK(peer->RequestVote(&request, &response));

  ASSERT_TRUE(response.vote_granted());

  ASSERT_EQ(last_op_id.term() + 1, response.responder_term());

  ASSERT_NO_FATALS(AssertDurableTermAndVote(kPeerIndex, last_op_id.term() + 1, "peer-0"));

  // Ensure we get same response for same term and same UUID.

  response.Clear();

  ASSERT_OK(peer->RequestVote(&request, &response));

  ASSERT_TRUE(response.vote_granted());

  // Ensure we get a "no" for a different candidate UUID for that term.

  response.Clear();

  request.set_candidate_uuid("peer-2");

  ASSERT_OK(peer->RequestVote(&request, &response));

  ASSERT_FALSE(response.vote_granted());

  ASSERT_TRUE(response.has_consensus_error());

  ASSERT_EQ(ConsensusErrorPB::ALREADY_VOTED, response.consensus_error().code());

  ASSERT_EQ(last_op_id.term() + 1, response.responder_term());

  ASSERT_NO_FATALS(AssertDurableTermAndVote(kPeerIndex, last_op_id.term() + 1, "peer-0"));

  //

  // Test that replicas refuse votes for an old term.

  //

  // Increase the term of our candidate, which will cause the voter replica to

  // increase its own term to match.

  request.set_candidate_uuid("peer-0");

  request.set_candidate_term(last_op_id.term() + 2);

  response.Clear();

  ASSERT_OK(peer->RequestVote(&request, &response));

  ASSERT_TRUE(response.vote_granted());

  ASSERT_EQ(last_op_id.term() + 2, response.responder_term());

  ASSERT_NO_FATALS(AssertDurableTermAndVote(kPeerIndex, last_op_id.term() + 2, "peer-0"));

  // Now try the old term.

  // Note: Use the peer who "won" the election on the previous term (peer-0),

  // although in practice the impl does not store historical vote data.

  request.set_candidate_term(last_op_id.term() + 1);

  response.Clear();

  ASSERT_OK(peer->RequestVote(&request, &response));

  ASSERT_FALSE(response.vote_granted());

  ASSERT_TRUE(response.has_consensus_error());

  ASSERT_EQ(ConsensusErrorPB::INVALID_TERM, response.consensus_error().code());

  ASSERT_EQ(last_op_id.term() + 2, response.responder_term());

  ASSERT_NO_FATALS(AssertDurableTermAndVote(kPeerIndex, last_op_id.term() + 2, "peer-0"));

  //

  // Ensure replicas vote no for an old op index.

  //

  request.set_candidate_uuid("peer-0");

  request.set_candidate_term(last_op_id.term() + 3);

  request.mutable_candidate_status()->mutable_last_received()->CopyFrom(MinimumOpId());

  response.Clear();

  ASSERT_OK(peer->RequestVote(&request, &response));

  ASSERT_FALSE(response.vote_granted());

  ASSERT_TRUE(response.has_consensus_error());

  ASSERT_EQ(ConsensusErrorPB::LAST_OPID_TOO_OLD, response.consensus_error().code());

  ASSERT_EQ(last_op_id.term() + 3, response.responder_term());

  ASSERT_NO_FATALS(AssertDurableTermWithoutVote(kPeerIndex, last_op_id.term() + 3));

  // Send a "heartbeat" to the peer. It should be rejected.

  ConsensusRequestPB req;

  req.set_caller_term(last_op_id.term());

  req.set_caller_uuid("peer-0");

  req.mutable_committed_op_id()->CopyFrom(last_op_id);

  ConsensusResponsePB res;

  Status s = peer->Update(&req, &res, CoarseBigDeadline());

  ASSERT_EQ(last_op_id.term() + 3, res.responder_term());

  ASSERT_TRUE(res.status().has_error());

  ASSERT_EQ(ConsensusErrorPB::INVALID_TERM, res.status().error().code());

  LOG(INFO) << "Follower rejected old heartbeat, as expected: " << res.ShortDebugString();

}

}  // namespace consensus

}  // namespace yb