rawnode_test.cpp

/**
 * Copyright 2025 AntGroup CO., Ltd.
 * Copyright 2015 The etcd Authors
 *
 * 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.
 */

// written by botu.wzy, inspired by etcd raft

#include "rawnode.h"

#include <gtest/gtest.h>

namespace eraft {
using namespace detail;
// TestRawNodeStep ensures that RawNode.Step ignore local message.
extern Config newTestConfig(uint64_t id, int election, int heartbeat,
                            std::shared_ptr<Storage> storage);
using testMemoryStorageOptions =
    std::function<void(std::shared_ptr<MemoryStorage>&)>;
extern testMemoryStorageOptions withPeers(const std::vector<uint64_t>& peers);
extern std::shared_ptr<MemoryStorage> newTestMemoryStorage(
    const std::vector<testMemoryStorageOptions>& opts);

// newTestRawNode sets up a RawNode with the given peers. The configuration will
// not be reflected in the Storage.
std::shared_ptr<RawNode> newTestRawNode(uint64_t id, int election,
                                        int heartbeat,
                                        std::shared_ptr<Storage> storage) {
  auto cfg = newTestConfig(id, election, heartbeat, std::move(storage));
  Error err;
  std::shared_ptr<RawNode> rn;
  std::tie(rn, err) = NewRawNode(cfg);
  EXPECT_EQ(err, nullptr);
  return rn;
}

TEST(raft, TestRawNodeStep) {
  for (int i = static_cast<int>(raftpb::MessageType_MIN);
       i <= static_cast<int>(raftpb::MessageType_MAX); i++) {
    auto s = NewMemoryStorage();
    raftpb::HardState hs;
    hs.set_term(1);
    hs.set_commit(1);
    s->SetHardState(hs);
    s->Append({EntryHelper{.index = 1, .term = 1}.Done()});
    raftpb::Snapshot sp;
    std::vector<uint64_t> voters = {1};
    *sp.mutable_metadata()->mutable_conf_state()->mutable_voters() = {
        voters.begin(), voters.end()};
    sp.mutable_metadata()->set_index(1);
    sp.mutable_metadata()->set_term(1);
    auto err = s->ApplySnapshot(sp);
    EXPECT_EQ(err, nullptr);
    // Append an empty entry to make sure the non-local messages (like
    // vote requests) are ignored and don't trigger assertions.
    std::shared_ptr<RawNode> rawNode;
    std::tie(rawNode, err) = NewRawNode(newTestConfig(1, 10, 1, s));
    EXPECT_EQ(err, nullptr);
    raftpb::Message msgt;
    msgt.set_type(static_cast<raftpb::MessageType>(i));
    err = rawNode->Step(msgt);
    // LocalMsg should be ignored.
    if (IsLocalMsg(static_cast<raftpb::MessageType>(i))) {
      EXPECT_EQ(err, ErrStepLocalMsg);
    }
  }
}

// TestNodeStepUnblock from node_test.go has no equivalent in rawNode because
// there is no goroutine in RawNode.

// TestRawNodeProposeAndConfChange tests the configuration change mechanism.
// Each test case sends a configuration change which is either simple or joint,
// verifies that it applies and that the resulting ConfState matches
// expectations, and for joint configurations makes sure that they are exited
// successfully.
std::shared_ptr<raftpb::ConfChangeWrap> NewConfChange(
    raftpb::ConfChangeType type, int node_id) {
  raftpb::ConfChange cc;
  cc.set_type(type);
  cc.set_node_id(node_id);
  return std::make_shared<raftpb::ConfChangeWrap>(cc);
}

raftpb::ConfChangeSingle NewConfChangeSingle(raftpb::ConfChangeType type,
                                             int node_id) {
  raftpb::ConfChangeSingle ccs;
  ccs.set_type(type);
  ccs.set_node_id(node_id);
  return ccs;
}

raftpb::ConfState NewConfState(const std::vector<uint64_t>& voters,
                               const std::vector<uint64_t>& learners,
                               const std::vector<uint64_t>& voters_outgoing,
                               const std::vector<uint64_t>& learners_next,
                               bool auto_leave) {
  raftpb::ConfState cs;
  *cs.mutable_voters() = {voters.begin(), voters.end()};
  *cs.mutable_learners() = {learners.begin(), learners.end()};
  *cs.mutable_voters_outgoing() = {voters_outgoing.begin(),
                                   voters_outgoing.end()};
  *cs.mutable_learners_next() = {learners_next.begin(), learners_next.end()};
  cs.set_auto_leave(auto_leave);
  return cs;
}

std::shared_ptr<raftpb::ConfChangeV2Wrap> NewConfChangeV2(
    const std::vector<raftpb::ConfChangeSingle>& ccss) {
  raftpb::ConfChangeV2 ccv2;
  *ccv2.mutable_changes() = {ccss.begin(), ccss.end()};
  return std::make_shared<raftpb::ConfChangeV2Wrap>(ccv2);
}

std::shared_ptr<raftpb::ConfChangeV2Wrap> NewConfChangeV2(
    const std::vector<raftpb::ConfChangeSingle>& ccss,
    raftpb::ConfChangeTransition cct) {
  raftpb::ConfChangeV2 ccv2;
  *ccv2.mutable_changes() = {ccss.begin(), ccss.end()};
  ccv2.set_transition(cct);
  return std::make_shared<raftpb::ConfChangeV2Wrap>(ccv2);
}

// TestNodeStepUnblock from node_test.go has no equivalent in rawNode because
// there is no goroutine in RawNode.

// TestRawNodeProposeAndConfChange tests the configuration change mechanism.
// Each test case sends a configuration change which is either simple or joint,
// verifies that it applies and that the resulting ConfState matches
// expectations, and for joint configurations makes sure that they are exited
// successfully.
TEST(raft, TestRawNodeProposeAndConfChange) {
  struct Test {
    std::shared_ptr<raftpb::ConfChangeI> cc;
    raftpb::ConfState exp;
    std::shared_ptr<raftpb::ConfState> exp2;
  };
  std::vector<Test> testCases = {
      // V1 config change.
      {
          NewConfChange(raftpb::ConfChangeAddNode, 2),
          NewConfState({1, 2}, {}, {}, {}, false),
          nullptr,
      },
      // Proposing the same as a V2 change works just the same, without entering
      // a joint config.
      {
          NewConfChangeV2({NewConfChangeSingle(raftpb::ConfChangeAddNode, 2)}),
          NewConfState({1, 2}, {}, {}, {}, false),
          nullptr,
      },
      // Ditto if we add it as a learner instead.
      {
          NewConfChangeV2(
              {NewConfChangeSingle(raftpb::ConfChangeAddLearnerNode, 2)}),
          NewConfState({1}, {2}, {}, {}, false),
          nullptr,
      },
      // We can ask explicitly for joint consensus if we want it.
      {
          NewConfChangeV2(
              {NewConfChangeSingle(raftpb::ConfChangeAddLearnerNode, 2)},
              raftpb::ConfChangeTransitionJointExplicit),
          NewConfState({1}, {1}, {2}, {}, false),
          std::make_shared<raftpb::ConfState>(
              NewConfState({1}, {2}, {}, {}, false)),
      },
      // Ditto, but with implicit transition (the harness checks this).
      {
          NewConfChangeV2(
              {NewConfChangeSingle(raftpb::ConfChangeAddLearnerNode, 2)},
              raftpb::ConfChangeTransitionJointImplicit),
          NewConfState({1}, {1}, {2}, {}, true),
          std::make_shared<raftpb::ConfState>(
              NewConfState({1}, {2}, {}, {}, false)),
      },
      // Add a new node and demote n1. This exercises the interesting case in
      // which we really need joint config changes and also need LearnersNext.
      {
          NewConfChangeV2(
              {NewConfChangeSingle(raftpb::ConfChangeAddLearnerNode, 2),
               NewConfChangeSingle(raftpb::ConfChangeAddLearnerNode, 1),
               NewConfChangeSingle(raftpb::ConfChangeAddLearnerNode, 3)}),
          NewConfState({2}, {1}, {3}, {1}, true),
          std::make_shared<raftpb::ConfState>(
              NewConfState({2}, {1, 3}, {}, {}, false)),
      },
      // Ditto explicit.
      {
          NewConfChangeV2(
              {NewConfChangeSingle(raftpb::ConfChangeAddNode, 2),
               NewConfChangeSingle(raftpb::ConfChangeAddLearnerNode, 1),
               NewConfChangeSingle(raftpb::ConfChangeAddLearnerNode, 3)},
              raftpb::ConfChangeTransitionJointExplicit),
          NewConfState({2}, {1}, {3}, {1}, false),
          std::make_shared<raftpb::ConfState>(
              NewConfState({2}, {1, 3}, {}, {}, false)),
      },
      // Ditto implicit.
      {
          NewConfChangeV2(
              {NewConfChangeSingle(raftpb::ConfChangeAddNode, 2),
               NewConfChangeSingle(raftpb::ConfChangeAddLearnerNode, 1),
               NewConfChangeSingle(raftpb::ConfChangeAddLearnerNode, 3)},
              raftpb::ConfChangeTransitionJointImplicit),
          NewConfState({2}, {1}, {3}, {1}, true),
          std::make_shared<raftpb::ConfState>(
              NewConfState({2}, {1, 3}, {}, {}, false)),
      }};
  for (auto& tc : testCases) {
    auto s = newTestMemoryStorage({withPeers({1})});
    std::shared_ptr<RawNode> rawNode;
    Error err;
    std::tie(rawNode, err) = NewRawNode(newTestConfig(1, 10, 1, s));
    EXPECT_EQ(err, nullptr);
    rawNode->Campaign();
    auto proposed = false;
    uint64_t lastIndex;
    std::string ccdata;
    // Propose the ConfChange, wait until it applies, save the resulting
    // ConfState.
    std::shared_ptr<raftpb::ConfState> cs;
    while (cs == nullptr) {
      auto rd = rawNode->GetReady();
      s->Append(rd.entries_);
      for (auto& ent : rd.committedEntries_) {
        std::unique_ptr<raftpb::ConfChangeI> cc;
        if (ent.type() == raftpb::EntryConfChange) {
          raftpb::ConfChange ccc;
          EXPECT_TRUE(ccc.ParseFromString(ent.data()));
          cc.reset(new raftpb::ConfChangeWrap(ccc));
        } else if (ent.type() == raftpb::EntryConfChangeV2) {
          raftpb::ConfChangeV2 ccc;
          EXPECT_TRUE(ccc.ParseFromString(ent.data()));
          cc.reset(new raftpb::ConfChangeV2Wrap(ccc));
        }
        if (cc != nullptr) {
          cs = rawNode->ApplyConfChange(*cc);
        }
      }
      rawNode->Advance(rd);
      // Once we are the leader, propose a command and a ConfChange.
      if (!proposed && rd.softState_->lead_ == rawNode->raft_->id_) {
        err = rawNode->Propose("somedata");
        EXPECT_EQ(err, nullptr);
        raftpb::ConfChange ccv1;
        bool ok;
        std::tie(ccv1, ok) = tc.cc->AsV1();
        if (ok) {
          EXPECT_TRUE(ccv1.SerializeToString(&ccdata));
          auto ccw = raftpb::ConfChangeWrap(ccv1);
          rawNode->ProposeConfChange(&ccw);
        } else {
          auto ccv2 = tc.cc->AsV2();
          EXPECT_TRUE(ccv2.SerializeToString(&ccdata));
          auto ccv2w = raftpb::ConfChangeV2Wrap(ccv2);
          rawNode->ProposeConfChange(&ccv2w);
        }
        proposed = true;
      }
    }

    // Check that the last index is exactly the conf change we put in,
    // down to the bits. Note that this comes from the Storage, which
    // will not reflect any unstable entries that we'll only be presented
    // with in the next Ready.
    std::tie(lastIndex, err) = s->LastIndex();
    EXPECT_EQ(err, nullptr);
    std::vector<raftpb::Entry> entries;
    std::tie(entries, err) = s->Entries(lastIndex - 1, lastIndex + 1, noLimit);
    EXPECT_EQ(err, nullptr);
    EXPECT_EQ(entries.size(), 2);
    EXPECT_EQ(entries[0].data(), "somedata");
    auto typ = raftpb::EntryConfChange;
    bool ok;
    std::tie(std::ignore, ok) = tc.cc->AsV1();
    if (!ok) {
      typ = raftpb::EntryConfChangeV2;
    }
    EXPECT_EQ(entries[1].type(), typ);
    EXPECT_EQ(entries[1].data(), ccdata);
    EXPECT_TRUE(
        google::protobuf::util::MessageDifferencer::Equals(tc.exp, *cs));

    uint64_t maybePlusOne = 0;
    bool autoLeave = false;
    std::tie(autoLeave, ok) =
        raftpb::ConfChangeV2Wrap(tc.cc->AsV2()).EnterJoint();
    if (ok && autoLeave) {
      // If this is an auto-leaving joint conf change, it will have
      // appended the entry that auto-leaves, so add one to the last
      // index that forms the basis of our expectations on
      // pendingConfIndex. (Recall that lastIndex was taken from stable
      // storage, but this auto-leaving entry isn't on stable storage
      // yet).
      maybePlusOne = 1;
    }
    EXPECT_EQ(lastIndex + maybePlusOne, rawNode->raft_->pendingConfIndex_);

    // Move the RawNode along. If the ConfChange was simple, nothing else
    // should happen. Otherwise, we're in a joint state, which is either
    // left automatically or not. If not, we add the proposal that leaves
    // it manually.
    auto rd = rawNode->GetReady();
    std::string context;
    if (!tc.exp.auto_leave()) {
      EXPECT_TRUE(rd.entries_.empty());
      rawNode->Advance(rd);
      if (tc.exp2 == nullptr) {
        return;
      }
      context = "manual";
      raftpb::ConfChangeV2 ccv2;
      ccv2.set_context(context);
      auto ccv2Wrap = raftpb::ConfChangeV2Wrap(ccv2);
      err = rawNode->ProposeConfChange(&ccv2Wrap);
      EXPECT_EQ(err, nullptr);
      rd = rawNode->GetReady();
    }

    // Check that the right ConfChange comes out.
    EXPECT_FALSE(rd.entries_.size() != 1 ||
                 rd.entries_[0].type() != raftpb::EntryConfChangeV2);
    raftpb::ConfChangeV2 cc;
    EXPECT_TRUE(cc.ParseFromString(rd.entries_[0].data()));
    raftpb::ConfChangeV2 ccv2;
    ccv2.set_context(context);
    EXPECT_TRUE(google::protobuf::util::MessageDifferencer::Equals(cc, ccv2));
    // Lie and pretend the ConfChange applied. It won't do so because now
    // we require the joint quorum and we're only running one node.
    cs = rawNode->ApplyConfChange(raftpb::ConfChangeV2Wrap(cc));
    EXPECT_TRUE(
        google::protobuf::util::MessageDifferencer::Equals(*tc.exp2, *cs));
    rawNode->Advance(rd);
  }
}

// TestRawNodeJointAutoLeave tests the configuration change auto leave even
// leader lost leadership.
TEST(raft, TestRawNodeJointAutoLeave) {
  raftpb::ConfChangeV2 testCc;
  testCc.set_transition(raftpb::ConfChangeTransitionJointImplicit);
  {
    auto ccs = testCc.mutable_changes()->Add();
    ccs->set_node_id(2);
    ccs->set_type(raftpb::ConfChangeAddLearnerNode);
  }
  raftpb::ConfState expCs;
  expCs.set_auto_leave(true);
  std::vector<uint64_t> voters = {1};
  std::vector<uint64_t> votersOutgoing = {1};
  std::vector<uint64_t> learners = {2};
  *expCs.mutable_voters() = {voters.begin(), voters.end()};
  *expCs.mutable_voters_outgoing() = {votersOutgoing.begin(),
                                      votersOutgoing.end()};
  *expCs.mutable_learners() = {learners.begin(), learners.end()};
  raftpb::ConfState exp2Cs;
  *exp2Cs.mutable_voters() = {voters.begin(), voters.end()};
  *exp2Cs.mutable_learners() = {learners.begin(), learners.end()};

  auto s = newTestMemoryStorage({withPeers({1})});
  std::shared_ptr<RawNode> rawNode;
  Error err;
  std::tie(rawNode, err) = NewRawNode(newTestConfig(1, 10, 1, s));
  EXPECT_EQ(err, nullptr);

  rawNode->Campaign();
  bool proposed = false;
  uint64_t lastIndex = 0;
  std::string ccdata;
  // Propose the ConfChange, wait until it applies, save the resulting
  // ConfState.
  std::shared_ptr<raftpb::ConfState> cs;
  while (cs == nullptr) {
    auto rd = rawNode->GetReady();
    s->Append(rd.entries_);
    for (auto& ent : rd.committedEntries_) {
      std::unique_ptr<raftpb::ConfChangeI> cc;
      if (ent.type() == raftpb::EntryConfChangeV2) {
        raftpb::ConfChangeV2 ccc;
        EXPECT_TRUE(ccc.ParseFromString(ent.data()));
        cc.reset(new raftpb::ConfChangeV2Wrap(ccc));
      }
      if (cc != nullptr) {
        // Force it step down.
        raftpb::Message m;
        m.set_type(raftpb::MsgHeartbeatResp);
        m.set_from(1);
        m.set_term(rawNode->raft_->term_ + 1);
        rawNode->Step(m);
        cs = rawNode->ApplyConfChange(*cc);
      }
    }
    rawNode->Advance(rd);
    // Once we are the leader, propose a command and a ConfChange.
    if (!proposed && rd.softState_->lead_ == rawNode->raft_->id_) {
      EXPECT_EQ(rawNode->Propose("somedata"), nullptr);
      ccdata = testCc.SerializeAsString();
      raftpb::ConfChangeV2Wrap testCcWrap(testCc);
      rawNode->ProposeConfChange(&testCcWrap);
      proposed = true;
    }
  }

  // Check that the last index is exactly the conf change we put in,
  // down to the bits. Note that this comes from the Storage, which
  // will not reflect any unstable entries that we'll only be presented
  // with in the next Ready.
  std::tie(lastIndex, err) = s->LastIndex();
  EXPECT_EQ(err, nullptr);
  std::vector<raftpb::Entry> entries;
  std::tie(entries, err) = s->Entries(lastIndex - 1, lastIndex + 1, noLimit);
  EXPECT_EQ(err, nullptr);
  EXPECT_EQ(entries.size(), 2);
  EXPECT_EQ(entries[0].data(), "somedata");
  EXPECT_EQ(entries[1].type(), raftpb::EntryConfChangeV2);
  EXPECT_EQ(entries[1].data(), ccdata);
  EXPECT_TRUE(google::protobuf::util::MessageDifferencer::Equals(expCs, *cs));
  EXPECT_EQ(rawNode->raft_->pendingConfIndex_, 0);

  // Move the RawNode along. It should not leave joint because it's follower.
  auto rd = rawNode->readyWithoutAccept();
  // Check that the right ConfChange comes out.
  EXPECT_EQ(rd.entries_.size(), 0);

  // Make it leader again. It should leave joint automatically after moving
  // apply index.
  rawNode->Campaign();
  rd = rawNode->GetReady();
  s->Append(rd.entries_);
  rawNode->Advance(rd);
  rd = rawNode->GetReady();
  s->Append(rd.entries_);
  rawNode->Advance(rd);
  rd = rawNode->GetReady();
  s->Append(rd.entries_);
  rawNode->Advance(rd);
  rd = rawNode->GetReady();
  s->Append(rd.entries_);
  // Check that the right ConfChange comes out.
  EXPECT_FALSE(rd.entries_.size() != 1 ||
               rd.entries_[0].type() != raftpb::EntryConfChangeV2);
  raftpb::ConfChangeV2 cc;
  EXPECT_TRUE(cc.ParseFromString(rd.entries_[0].data()));
  EXPECT_TRUE(google::protobuf::util::MessageDifferencer::Equals(
      cc, raftpb::ConfChangeV2()));
  // Lie and pretend the ConfChange applied. It won't do so because now
  // we require the joint quorum and we're only running one node.
  cs = rawNode->ApplyConfChange(raftpb::ConfChangeV2Wrap(cc));
  EXPECT_TRUE(google::protobuf::util::MessageDifferencer::Equals(exp2Cs, *cs));
}

// TestRawNodeProposeAddDuplicateNode ensures that two proposes to add the same
// node should not affect the later propose to add new node.
TEST(raft, TestRawNodeProposeAddDuplicateNode) {
  auto s = newTestMemoryStorage({withPeers({1})});
  std::shared_ptr<RawNode> rawNode;
  Error err;
  std::tie(rawNode, err) = NewRawNode(newTestConfig(1, 10, 1, s));
  EXPECT_EQ(err, nullptr);
  auto rd = rawNode->GetReady();
  s->Append(rd.entries_);
  rawNode->Advance(rd);

  rawNode->Campaign();
  while (true) {
    rd = rawNode->GetReady();
    s->Append(rd.entries_);
    if (rd.softState_->lead_ == rawNode->raft_->id_) {
      rawNode->Advance(rd);
      break;
    }
    rawNode->Advance(rd);
  }

  auto proposeConfChangeAndApply = [&](const raftpb::ConfChange& cc) {
    auto ccw = raftpb::ConfChangeWrap(cc);
    rawNode->ProposeConfChange(&ccw);
    rd = rawNode->GetReady();
    s->Append(rd.entries_);
    for (auto& entry : rd.committedEntries_) {
      if (entry.type() == raftpb::EntryConfChange) {
        raftpb::ConfChange cc1;
        cc1.ParseFromString(entry.data());
        rawNode->ApplyConfChange(raftpb::ConfChangeWrap(cc1));
      }
    }
    rawNode->Advance(rd);
  };

  raftpb::ConfChange cc1;
  cc1.set_node_id(1);
  cc1.set_type(raftpb::ConfChangeAddNode);
  auto ccdata1 = cc1.SerializeAsString();
  proposeConfChangeAndApply(cc1);

  // try to add the same node again
  proposeConfChangeAndApply(cc1);

  // the new node join should be ok
  raftpb::ConfChange cc2;
  cc2.set_node_id(2);
  cc2.set_type(raftpb::ConfChangeAddNode);
  auto ccdata2 = cc2.SerializeAsString();
  proposeConfChangeAndApply(cc2);

  uint64_t lastIndex;
  std::tie(lastIndex, err) = s->LastIndex();
  EXPECT_EQ(err, nullptr);

  // the last three entries should be: ConfChange cc1, cc1, cc2
  std::vector<raftpb::Entry> entries;
  std::tie(entries, err) = s->Entries(lastIndex - 2, lastIndex + 1, noLimit);
  EXPECT_EQ(err, nullptr);
  EXPECT_EQ(entries.size(), 3);
  EXPECT_EQ(entries[0].data(), ccdata1);
  EXPECT_EQ(entries[2].data(), ccdata2);
}

// TestRawNodeReadIndex ensures that Rawnode.ReadIndex sends the MsgReadIndex
// message to the underlying raft. It also ensures that ReadState can be read
// out.
TEST(raft, TestRawNodeReadIndex) {
  std::vector<raftpb::Message> msgs;
  auto appendStep = [&](raft&, raftpb::Message m) -> Error {
    msgs.emplace_back(std::move(m));
    return {};
  };
  auto equal = [](const std::vector<ReadState>& a,
                  const std::vector<ReadState>& b) {
    if (a.size() != b.size()) {
      return false;
    }
    for (size_t i = 0; i < a.size(); i++) {
      if (a[i].index_ != b[i].index_ || a[i].requestCtx_ != b[i].requestCtx_) {
        return false;
      }
    }
    return true;
  };
  std::vector<ReadState> wrs = {{1, "somedata"}};

  auto s = newTestMemoryStorage({withPeers({1})});
  auto c = newTestConfig(1, 10, 1, s);
  std::shared_ptr<RawNode> rawNode;
  Error err;
  std::tie(rawNode, err) = NewRawNode(c);
  EXPECT_EQ(err, nullptr);

  rawNode->raft_->readStates_ = wrs;
  // ensure the ReadStates can be read out
  auto hasReady = rawNode->HasReady();
  EXPECT_TRUE(hasReady);
  auto rd = rawNode->GetReady();
  EXPECT_TRUE(equal(rd.readStates_, wrs));
  s->Append(rd.entries_);
  rawNode->Advance(rd);
  // ensure raft.readStates is reset after advance
  EXPECT_TRUE(rawNode->raft_->readStates_.empty());

  std::string wrequestCtx("somedata2");
  rawNode->Campaign();
  while (true) {
    rd = rawNode->GetReady();
    s->Append(rd.entries_);

    if (rd.softState_->lead_ == rawNode->raft_->id_) {
      rawNode->Advance(rd);

      // Once we are the leader, issue a ReadIndex request
      rawNode->raft_->step_ = appendStep;
      rawNode->ReadIndex(wrequestCtx);
      break;
    }
    rawNode->Advance(rd);
  }
  // ensure that MsgReadIndex message is sent to the underlying raft
  EXPECT_EQ(msgs.size(), 1);
  EXPECT_EQ(msgs[0].type(), raftpb::MsgReadIndex);
  EXPECT_EQ(msgs[0].entries(0).data(), wrequestCtx);
}

// TestBlockProposal from node_test.go has no equivalent in rawNode because
// there is no leader check in RawNode.

// TestNodeTick from node_test.go has no equivalent in rawNode because
// it reaches into the raft object which is not exposed.

// TestNodeStop from node_test.go has no equivalent in rawNode because there is
// no goroutine in RawNode.

// TestRawNodeStart ensures that a node can be started correctly. Note that
// RawNode requires the application to bootstrap the state, i.e. it does not
// accept peers and will not create faux configuration change entries.
TEST(raft, TestRawNodeStart) {
  std::vector<raftpb::Entry> entries = {
      {EntryHelper{.index = 2, .term = 1}.Done(),
       EntryHelper{.index = 3, .term = 1, .data = "foo"}.Done()}};
  Ready want;
  want.softState_ = std::make_shared<SoftState>();
  want.softState_->lead_ = 1;
  want.softState_->raftState_ = StateLeader;

  want.hardState_.set_term(1);
  want.hardState_.set_commit(3);
  want.hardState_.set_vote(1);

  want.committedEntries_ = entries;
  want.mustSync_ = false;

  auto storage = NewMemoryStorage();
  storage->ents_[0].set_index(1);

  // TODO(tbg): this is a first prototype of what bootstrapping could look
  // like (without the annoying faux ConfChanges). We want to persist a
  // ConfState at some index and make sure that this index can't be reached
  // from log position 1, so that followers are forced to pick up the
  // ConfState in order to move away from log position 1 (unless they got
  // bootstrapped in the same way already). Failing to do so would mean that
  // followers diverge from the bootstrapped nodes and don't learn about the
  // initial config.
  //
  // NB: this is exactly what CockroachDB does. The Raft log really begins at
  // index 10, so empty followers (at index 1) always need a snapshot first.
  auto bootstrap = [](const std::shared_ptr<MemoryStorage>& storage,
                      const raftpb::ConfState& cs) -> Error {
    if (cs.voters().empty()) {
      return Error("no voters specified");
    }
    uint64_t fi;
    Error err;
    std::tie(fi, err) = storage->FirstIndex();
    if (err != nullptr) {
      return err;
    }
    if (fi < 2) {
      return Error("FirstIndex >= 2 is prerequisite for bootstrap");
    }
    std::tie(std::ignore, err) =
        storage->Entries(fi, fi, std::numeric_limits<uint64_t>::max());
    if (err == nullptr) {
      // TODO(tbg): match exact error
      return Error("should not have been able to load first index");
    }
    uint64_t li;
    std::tie(li, err) = storage->LastIndex();
    if (err != nullptr) {
      return err;
    }
    std::tie(std::ignore, err) =
        storage->Entries(li, li, std::numeric_limits<uint64_t>::max());
    if (err == nullptr) {
      return Error("should not have been able to load last index");
    };
    raftpb::HardState hs;
    raftpb::ConfState ics;
    std::tie(hs, ics, err) = storage->InitialState();
    if (err != nullptr) {
      return err;
    }
    if (!IsEmptyHardState(hs)) {
      return Error("HardState not empty");
    }
    if (!ics.voters().empty()) {
      return Error("ConfState not empty");
    }
    raftpb::SnapshotMetadata meta;
    meta.set_index(1);
    meta.set_term(0);
    *meta.mutable_conf_state() = cs;
    raftpb::Snapshot snap;
    *snap.mutable_metadata() = meta;
    return storage->ApplySnapshot(snap);
  };
  std::vector<uint64_t> voters = {1};
  raftpb::ConfState cs;
  *cs.mutable_voters() = {voters.begin(), voters.end()};
  EXPECT_EQ(bootstrap(storage, cs), nullptr);
  std::shared_ptr<RawNode> rawNode;
  Error err;
  std::tie(rawNode, err) = NewRawNode(newTestConfig(1, 10, 1, storage));
  EXPECT_EQ(err, nullptr);
  EXPECT_FALSE(rawNode->HasReady());
  rawNode->Campaign();
  auto rd = rawNode->GetReady();
  storage->Append(rd.entries_);
  rawNode->Advance(rd);
  rawNode->Propose("foo");
  EXPECT_TRUE(rawNode->HasReady());
  rd = rawNode->GetReady();
  EXPECT_TRUE(VectorEquals(entries, rd.entries_));
  storage->Append(rd.entries_);
  rawNode->Advance(rd);

  EXPECT_TRUE(rawNode->HasReady());
  rd = rawNode->GetReady();
  EXPECT_EQ(rd.entries_.size(), 0);
  EXPECT_FALSE(rd.mustSync_);
  rawNode->Advance(rd);

  rd.softState_ = nullptr;
  want.softState_ = nullptr;
  EXPECT_EQ(rd, want);
  EXPECT_FALSE(rawNode->HasReady());
}

TEST(raft, TestRawNodeRestart) {
  std::vector<raftpb::Entry> entries = {
      EntryHelper{.index = 1, .term = 1}.Done(),
      EntryHelper{.index = 2, .term = 1, .data = "foo"}.Done()};
  raftpb::HardState st;
  st.set_term(1);
  st.set_commit(1);

  Ready want;
  want.committedEntries_ = {entries.begin(), entries.begin() + st.commit()};
  want.mustSync_ = false;

  auto storage = newTestMemoryStorage({withPeers({1})});
  storage->SetHardState(st);
  storage->Append(entries);
  std::shared_ptr<RawNode> rawNode;
  Error err;
  std::tie(rawNode, err) = NewRawNode(newTestConfig(1, 10, 1, storage));
  EXPECT_EQ(err, nullptr);
  auto rd = rawNode->GetReady();
  EXPECT_EQ(rd, want);
  rawNode->Advance(rd);
  EXPECT_FALSE(rawNode->HasReady());
}

TEST(raft, TestRawNodeRestartFromSnapshot) {
  raftpb::Snapshot snap;
  snap.mutable_metadata()->set_index(2);
  snap.mutable_metadata()->set_term(1);
  raftpb::ConfState cs;
  std::vector<uint64_t> voters = {1, 2};
  *cs.mutable_voters() = {voters.begin(), voters.end()};
  *snap.mutable_metadata()->mutable_conf_state() = cs;

  std::vector<raftpb::Entry> entries = {
      EntryHelper{.index = 3, .term = 1, .data = "foo"}.Done()};
  raftpb::HardState st;
  st.set_term(1);
  st.set_commit(3);

  Ready want;
  want.committedEntries_ = entries;
  want.mustSync_ = false;

  auto s = NewMemoryStorage();
  s->SetHardState(st);
  s->ApplySnapshot(snap);
  s->Append(entries);
  std::shared_ptr<RawNode> rawNode;
  Error err;
  std::tie(rawNode, err) = NewRawNode(newTestConfig(1, 10, 1, s));
  EXPECT_EQ(err, nullptr);
  auto rd = rawNode->GetReady();
  EXPECT_EQ(rd, want);
  rawNode->Advance(rd);
  EXPECT_FALSE(rawNode->HasReady());
}

// TestNodeAdvance from node_test.go has no equivalent in rawNode because there
// is no dependency check between Ready() and Advance()
TEST(raft, TestRawNodeStatus) {
  auto s = newTestMemoryStorage({withPeers({1})});
  std::shared_ptr<RawNode> rn;
  Error err;
  std::tie(rn, err) = NewRawNode(newTestConfig(1, 10, 1, s));
  EXPECT_EQ(err, nullptr);
  auto status = rn->GetStatus();
  EXPECT_TRUE(status.progress_.empty());
  err = rn->Campaign();
  EXPECT_EQ(err, nullptr);
  auto rd = rn->GetReady();
  s->Append(rd.entries_);
  rn->Advance(rd);
  status = rn->GetStatus();
  EXPECT_EQ(status.basicStatus_.softState_.lead_, 1);
  EXPECT_EQ(status.basicStatus_.softState_.raftState_, StateLeader);
  const auto& exp = *rn->raft_->trk_.progress_[1];
  auto act = status.progress_.at(1);
  EXPECT_EQ(exp, act);
  tracker::Config expCfg;
  expCfg.voters_.data_[0].data_ = {1};
  EXPECT_EQ(expCfg, status.config_);
}

// TestRawNodeCommitPaginationAfterRestart is the RawNode version of
// TestNodeCommitPaginationAfterRestart. The anomaly here was even worse as the
// Raft group would forget to apply entries:
//
//   - node learns that index 11 is committed
//   - nextCommittedEnts returns index 1..10 in CommittedEntries (but index 10
//     already exceeds maxBytes), which isn't noticed internally by Raft
//   - Commit index gets bumped to 10
//   - the node persists the HardState, but crashes before applying the entries
//   - upon restart, the storage returns the same entries, but `slice` takes a
//     different code path and removes the last entry.
//   - Raft does not emit a HardState, but when the app calls Advance(), it
//   bumps
//     its internal applied index cursor to 10 (when it should be 9)
//   - the next Ready asks the app to apply index 11 (omitting index 10), losing
//   a
//     write.
TEST(raft, TestRawNodeCommitPaginationAfterRestart) {
  auto s = newTestMemoryStorage({withPeers({1})});
  raftpb::HardState persistedHardState;
  persistedHardState.set_term(1);
  persistedHardState.set_vote(1);
  persistedHardState.set_commit(10);
  s->hardState_ = persistedHardState;
  s->ents_.resize(10);
  uint64_t size;
  for (size_t i = 0; i < s->ents_.size(); i++) {
    s->ents_[i].set_term(1);
    s->ents_[i].set_index(i + 1);
    s->ents_[i].set_type(raftpb::EntryNormal);
    s->ents_[i].set_data("a");
    size += s->ents_[i].ByteSizeLong();
  }

  auto cfg = newTestConfig(1, 10, 1, s);
  // Set a MaxSizePerMsg that would suggest to Raft that the last committed
  // entry should not be included in the initial rd.CommittedEntries. However,
  // our storage will ignore this and *will* return it (which is how the Commit
  // index ended up being 10 initially).
  cfg.maxSizePerMsg_ = size - s->ents_[s->ents_.size() - 1].ByteSizeLong() - 1;

  s->ents_.emplace_back();
  s->ents_.back().set_term(1);
  s->ents_.back().set_index(11);
  s->ents_.back().set_type(raftpb::EntryNormal);
  s->ents_.back().set_data("boom");

  std::shared_ptr<RawNode> rawNode;
  Error err;
  std::tie(rawNode, err) = NewRawNode(cfg);
  EXPECT_EQ(err, nullptr);

  for (size_t highestApplied = 0; highestApplied != 11;) {
    auto rd = rawNode->GetReady();
    auto n = rd.committedEntries_.size();
    EXPECT_TRUE(n != 0);
    if (highestApplied != 0 &&
        highestApplied + 1 != rd.committedEntries_[0].index()) {
      EXPECT_FALSE(true);
    }
    highestApplied = rd.committedEntries_[n - 1].index();
    rawNode->Advance(rd);
    raftpb::Message msg;
    msg.set_type(raftpb::MsgHeartbeat);
    msg.set_to(1);
    msg.set_from(2);  // illegal, but we get away with it
    msg.set_term(1);
    msg.set_commit(11);
    rawNode->Step(msg);
  }
}

// TestRawNodeBoundedLogGrowthWithPartition tests a scenario where a leader is
// partitioned from a quorum of nodes. It verifies that the leader's log is
// protected from unbounded growth even as new entries continue to be proposed.
// This protection is provided by the MaxUncommittedEntriesSize configuration.
TEST(raft, TestRawNodeBoundedLogGrowthWithPartition) {
  const size_t maxEntries = 16;
  std::string data = "testdata";
  raftpb::Entry testEntry = EntryHelper{.data = data}.Done();
  auto maxEntrySize = maxEntries * payloadSize(testEntry);

  auto s = newTestMemoryStorage({withPeers({1})});
  auto cfg = newTestConfig(1, 10, 1, s);
  cfg.maxUncommittedEntriesSize_ = maxEntrySize;
  std::shared_ptr<RawNode> rawNode;
  Error err;
  std::tie(rawNode, err) = NewRawNode(cfg);
  EXPECT_EQ(err, nullptr);

  // Become the leader and apply empty entry.
  rawNode->Campaign();
  while (true) {
    auto rd = rawNode->GetReady();
    s->Append(rd.entries_);
    rawNode->Advance(rd);
    if (!rd.committedEntries_.empty()) {
      break;
    }
  }

  // Simulate a network partition while we make our proposals by never
  // committing anything. These proposals should not cause the leader's
  // log to grow indefinitely.
  for (auto i = 0; i < 1024; i++) {
    rawNode->Propose(data);
  }

  // Check the size of leader's uncommitted log tail. It should not exceed the
  // MaxUncommittedEntriesSize limit.
  auto checkUncommitted = [&rawNode](uint64_t exp) {
    EXPECT_EQ(rawNode->raft_->uncommittedSize_, exp);
  };
  checkUncommitted(maxEntrySize);

  // Recover from the partition. The uncommitted tail of the Raft log should
  // disappear as entries are committed.
  auto rd = rawNode->GetReady();
  EXPECT_EQ(rd.entries_.size(), maxEntries);
  s->Append(rd.entries_);
  rawNode->Advance(rd);

  // Entries are appended, but not applied.
  checkUncommitted(maxEntrySize);

  rd = rawNode->GetReady();
  EXPECT_TRUE(rd.entries_.empty());
  EXPECT_EQ(rd.committedEntries_.size(), maxEntries);
  rawNode->Advance(rd);

  checkUncommitted(0);
}

TEST(raft, TestRawNodeConsumeReady) {
  // Check that readyWithoutAccept() does not call acceptReady (which resets
  // the messages) but Ready() does.
  auto s = newTestMemoryStorage({withPeers({1})});
  auto rn = newTestRawNode(1, 3, 1, s);
  raftpb::Message m1, m2;
  m1.set_context("foo");
  m2.set_context("bar");

  // Inject first message, make sure it's visible via readyWithoutAccept.
  rn->raft_->msgs_.push_back(m1);
  auto rd = rn->readyWithoutAccept();
  EXPECT_FALSE(
      rd.messages_.size() != 1 ||
      !google::protobuf::util::MessageDifferencer::Equals(rd.messages_[0], m1));
  // Now call Ready() which should move the message into the Ready (as opposed
  // to leaving it in both places).
  rd = rn->GetReady();
  EXPECT_TRUE(rn->raft_->msgs_.empty());
  EXPECT_FALSE(
      rd.messages_.size() != 1 ||
      !google::protobuf::util::MessageDifferencer::Equals(rd.messages_[0], m1));
  // Add a message to raft to make sure that Advance() doesn't drop it.
  rn->raft_->msgs_.push_back(m2);
  rn->Advance(rd);
  EXPECT_FALSE(rn->raft_->msgs_.size() != 1 ||
               !google::protobuf::util::MessageDifferencer::Equals(
                   rn->raft_->msgs_[0], m2));
}

}  // namespace eraft