type Storage interface {
  // InitialState returns the saved HardState and ConfState information.
  InitialState() (pb.HardState, pb.ConfState, error)
  // Entries returns a slice of log entries in the range [lo,hi).
  // MaxSize limits the total size of the log entries returned, but
  // Entries returns at least one entry if any.
  Entries(lo, hi, maxSize uint64) ([]pb.Entry, error)
  // Term returns the term of entry i, which must be in the range
  // [FirstIndex()-1, LastIndex()]. The term of the entry before
  // FirstIndex is retained for matching purposes even though the
  // rest of that entry may not be available.
  Term(i uint64) (uint64, error)
  // LastIndex returns the index of the last entry in the log.
  LastIndex() (uint64, error)
  // FirstIndex returns the index of the first log entry that is
  // possibly available via Entries (older entries have been incorporated
  // into the latest Snapshot; if storage only contains the dummy entry the
  // first log entry is not available).
  FirstIndex() (uint64, error)
  // Snapshot returns the most recent snapshot.
  // If snapshot is temporarily unavailable, it should return 
  // ErrSnapshotTemporarilyUnavailable,
  // so raft state machine could know that Storage needs some time to prepare
  // snapshot and call Snapshot later.
  Snapshot() (pb.Snapshot, error)
}	

// Append the new entries to storage.
// TODO (xiangli): ensure the entries are continuous and
// entries[0].Index > ms.entries[0].Index
func (ms *MemoryStorage) Append(entries []pb.Entry) error {
  if len(entries) == 0 {
    return nil
  }

  ms.Lock()
  defer ms.Unlock()

  first := ms.firstIndex()
  last := entries[0].Index + uint64(len(entries)) - 1

  // shortcut if there is no new entry.
  if last < first {
    return nil
  }
  // truncate compacted entries
  if first > entries[0].Index {
    entries = entries[first-entries[0].Index:]
  }

  offset := entries[0].Index - ms.ents[0].Index
  switch {
  case uint64(len(ms.ents)) > offset:
    ms.ents = append([]pb.Entry{}, ms.ents[:offset]...)
    ms.ents = append(ms.ents, entries...)
  case uint64(len(ms.ents)) == offset:
    ms.ents = append(ms.ents, entries...)
  default:
    raftLogger.Panicf("missing log entry [last: %d, append at: %d]",
      ms.lastIndex(), entries[0].Index)
  }
  return nil

// unstable.entries[i] has raft log position i+unstable.offset.
// Note that unstable.offset may be less than the highest log
// position in storage; this means that the next write to storage
// might need to truncate the log before persisting unstable.entries.
type unstable struct {
  // the incoming unstable snapshot, if any.
  snapshot *pb.Snapshot
  // all entries that have not yet been written to storage.
  entries []pb.Entry
  offset  uint64
  // record error msg
  logger Logger
}

func (u *unstable) truncateAndAppend(ents []pb.Entry) {
  after := ents[0].Index
  switch {
  case after == u.offset+uint64(len(u.entries)):
    // after is the next index in the u.entries
    // directly append
    u.entries = append(u.entries, ents...)
  case after <= u.offset:
    u.logger.Infof("replace the unstable entries from index %d", after)
    // The log is being truncated to before our current offset
    // portion, so set the offset and replace the entries
    u.offset = after
    u.entries = ents
  default:
    // truncate to after and copy to u.entries
    // then append
    u.logger.Infof("truncate the unstable entries before index %d", after)
    u.entries = append([]pb.Entry{}, u.slice(u.offset, after)...)
    u.entries = append(u.entries, ents...)
  }
}

type raftLog struct {
  // storage contains all stable entries since the last snapshot.
  storage Storage

  // unstable contains all unstable entries and snapshot.
  // they will be saved into storage.
  unstable unstable

  // committed is the highest log position that is known to be in
  // stable storage on a quorum of nodes.
  committed uint64
  // applied is the highest log position that the application has
  // been instructed to apply to its state machine.
  // Invariant: applied <= committed
  applied uint64

  logger Logger

  // maxNextEntsSize is the maximum number aggregate
  // byte size of the messages
  // returned from calls to nextEnts.
  maxNextEntsSize uint64
}

// maybeAppend returns (0, false) if the entries cannot be appended.
// Otherwise, it returns (last index of new entries, true).
func (l *raftLog)
    maybeAppend(index, logTerm, committed uint64, ents ...pb.Entry)
    (lastnewi uint64, ok bool) {
  if l.matchTerm(index, logTerm) {
    lastnewi = index + uint64(len(ents))
    ci := l.findConflict(ents)
    switch {
    case ci == 0:
    case ci <= l.committed:
      l.logger.Panicf("entry %d conflict with committed entry
          [committed(%d)]", ci, l.committed)
    default:
      offset := index + 1
      l.append(ents[ci-offset:]...)
    }
    // AZ: follower behavior
    l.commitTo(min(committed, lastnewi))
    return lastnewi, true
  }
  return 0, false
}

func (l *raftLog) append(ents ...pb.Entry) uint64 {
  if len(ents) == 0 {
    return l.lastIndex()
  }
  if after := ents[0].Index - 1; after < l.committed {
    l.logger.Panicf("after(%d) is out of range [committed(%d)]",
        after, l.committed)
  }
  // AZ: append in unstable only
  l.unstable.truncateAndAppend(ents)
  return l.lastIndex()
}

// findConflict finds the index of the conflict.
// It returns the first pair of conflicting entries between the existing
// entries and the given entries, if there are any.
// If there is no conflicting entries, and the existing entries contains
// all the given entries, zero will be returned.
// If there is no conflicting entries, but the given entries contains new
// entries, the index of the first new entry will be returned.
// An entry is considered to be conflicting if it has the same index but
// a different term.
// The first entry MUST have an index equal to the argument 'from'.
// The index of the given entries MUST be continuously increasing.
func (l *raftLog) findConflict(ents []pb.Entry) uint64 {
  // AZ: check every ents in raftLog, if theie term match
  for _, ne := range ents {
    if !l.matchTerm(ne.Index, ne.Term) {
      if ne.Index <= l.lastIndex() {
        l.logger.Infof("found conflict")
      }
      return ne.Index
    }
  }
  return 0
}

type raft struct {
  id uint64
  Term uint64
  Vote uint64
  prs tracker.ProgressTracker
  state StateType
  // AZ: step 是一个函数指针，具体的实现由其角色决定
  //（stepLeader stepCandidate stepFollower）
  step stepFunc
}

func (r *raft) Step(m pb.Message) error {
  // Handle the message term
  // which may result in our stepping down to a follower.
  switch {
  case m.Term == 0:
    // local message
  case m.Term > r.Term:
    // If a server receives a RequestVote request within the minimum
    // election timeout of hearing from a current leader, it does not
    // update its term or grant its vote
    switch {
    // Process Prevote Logic
    ...
    default:
      if m.Type == pb.MsgApp || m.Type == pb.MsgHeartbeat
        || m.Type == pb.MsgSnap {
        r.becomeFollower(m.Term, m.From)
      } else {
        r.becomeFollower(m.Term, None)
      }
    }
  case m.Term < r.Term:
    if (r.checkQuorum) && (m.Type == pb.MsgHeartbeat ||m.Type == pb.MsgApp) {
      r.send(pb.Message{To: m.From, Type: pb.MsgAppResp}) // AZ: with term 0
    } else {
      // ignore other cases
    }
    return nil
  }
  switch m.Type{
    // process pb.MsgHup and pb.MsgVote, pb.MsgPreVote
    default:
      err := r.step(r, m)
      if err != nil {
        return err
      }
    }
  }
}

func stepLeader(r *raft, m pb.Message) error {
switch m.Type {
  case pb.MsgProp:
  // AZ: check if there is config change pending. If there is one pending,
  // drop this proposal.
  // AZ: go over m.Entries, to check is there is config change request.
  // Merge config changes to be one config change request.
  r.appendEntry(m.Entries...)
  r.bcastAppend()
}

func (r *raft) appendEntry(es ...pb.Entry) (accepted bool) {
  li := r.raftLog.lastIndex()
  for i := range es {
    es[i].Term = r.Term
    es[i].Index = li + 1 + uint64(i)
  }
  // Track the size of this uncommitted proposal.
  if !r.increaseUncommittedSize(es) {
    // Drop the proposal.
    return false
  }
  // use latest "last" index after truncate/append
  li = r.raftLog.append(es...)
  r.prs.Progress[r.id].MaybeUpdate(li)
  // Regardless of maybeCommit's return, our caller will call bcastAppend.
  r.maybeCommit()
  return true
}

// bcastAppend sends RPC, with entries to all peers that are not up-to-date
// according to the progress recorded in r.prs.
func (r *raft) bcastAppend() {
  r.prs.Visit(func(id uint64, _ *tracker.Progress) {
    if id == r.id {
      return
    }
    r.sendAppend(id)
  })
}

// sendAppend sends an append RPC with new entries (if any) and the
// current commit index to the given peer.
func (r *raft) sendAppend(to uint64) {
  term, errt := r.raftLog.term(pr.Next - 1)
  ents, erre := r.raftLog.entries(pr.Next, r.maxMsgSize)
  // send snapshot if we failed to get term or entries
  if errt != nil || erre != nil {
    pr.BecomeSnapshot(sindex)
  } else {
    m.Type = pb.MsgApp
    m.Index = pr.Next - 1 // AZ: last_index
    m.LogTerm = term // AZ: last_term
    m.Entries = ents
    m.Commit = r.raftLog.committed
  }
  r.send(m)
}

func stepFollower(r *raft, m pb.Message) error {
  switch m.Type {
  ...
  case pb.MsgProp:
    if r.lead == None {
      r.logger.Infof("%x no leader at term %d; dropping proposal")
      return ErrProposalDropped
    }
    m.To = r.lead
    r.send(m)
  case pb.MsgApp:
    r.electionElapsed = 0
    r.lead = m.From
    r.handleAppendEntries(m)
  ...
}

func (r *raft) handleAppendEntries(m pb.Message) {
  if m.Index < r.raftLog.committed {
    r.send(pb.Message
      {To: m.From, Type: pb.MsgAppResp, Index: r.raftLog.committed})
    return
  }
  
  if mlastIndex, ok := r.raftLog.maybeAppend
    (m.Index, m.LogTerm, m.Commit, m.Entries...); ok {
    r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: mlastIndex})
  } else {
    // AZ：maybeAppend failed, last_index last_term not match
    r.send(pb.Message
      {To: m.From, Type: pb.MsgAppResp, Index: m.Index,
      Reject: true, RejectHint: r.raftLog.lastIndex()})
  }
}

func stepLeader(r *raft, m pb.Message) error {
  switch m.Type {
  case pb.MsgAppResp:
    pr.RecentActive = true
    if m.Reject {
      if pr.MaybeDecrTo(m.Index, m.RejectHint) {
        if pr.State == tracker.StateReplicate {
          pr.BecomeProbe()
        }
        r.sendAppend(m.From)
      }
    } else {
      // AZ: progress update match and next
      if pr.MaybeUpdate(m.Index) {
        switch {
        case pr.State == tracker.StateProbe:
          pr.BecomeReplicate()
        case pr.State == tracker.StateSnapshot
          && pr.Match >= pr.PendingSnapshot:
          // Transition back to replicating state via probing state
          // (which takes the snapshot into account). If we didn't
          // move to replicating state, that would only happen with
          // the next round of appends (but there may not be a next
          // round for a while, exposing an inconsistent RaftStatus).
          pr.BecomeProbe()
          pr.BecomeReplicate()
        case pr.State == tracker.StateReplicate:
          pr.Inflights.FreeLE(m.Index)
        }
        // AZ: Update leader committed id
        if r.maybeCommit() {
          // AZ: if update leader commit success, broadcast to peers
          r.bcastAppend()
        }
        // AZ: send next round of MsgApp
        r.maybeSendAppend(m.From, false)
      }
    }
  }
}


func (pr *Progress) MaybeDecrTo(rejected, last uint64) bool {
  if pr.State == StateReplicate {
    // The rejection must be stale if the progress has matched and "rejected"
    // is smaller than "match".
    if rejected <= pr.Match {
      return false
    }
    // Directly decrease next to match + 1.
    pr.Next = pr.Match + 1
    return true
  }
  return true
}