Consensus

func (nd *Node) resetTimer(now int64) {
  span := nd.t.ElectionMax - nd.t.ElectionMin
  jitter := int64(0)
  if span > 0 {
    jitter = nd.rng.Int64N(span + 1)
  }
  nd.deadline = now + nd.t.ElectionMin + jitter
}

func (nd *Node) campaign(now int64) []Message {
  nd.Role = Candidate
  nd.Term++
  nd.VotedFor = nd.ID
  nd.LeaderID = -1
  nd.votes = map[int]bool{nd.ID: true}
  nd.resetTimer(now) // a split vote times out again, at a new random time
  if nd.n == 1 {
    return nd.becomeLeader(now)
  }
  var out []Message
  for p := range nd.n {
    if p != nd.ID {
      out = append(out, Message{Kind: VoteReq, From: nd.ID, To: p, Term: nd.Term, LastIndex: nd.LastIndex(), LastTerm: nd.lastTerm()})
    }
  }
  return out
}

func (nd *Node) handleVoteReq(now int64, m Message) Message {
  upToDate := m.LastTerm > nd.lastTerm() || (m.LastTerm == nd.lastTerm() && m.LastIndex >= nd.LastIndex())
  grant := m.Term == nd.Term &&
    (nd.VotedFor == -1 || nd.VotedFor == m.From) &&
    (upToDate || !nd.check) // the election restriction
  if grant {
    nd.VotedFor = m.From
    nd.resetTimer(now)
  }
  return Message{Kind: VoteResp, From: nd.ID, To: m.From, Term: nd.Term, OK: grant}
}

func (nd *Node) handleAppendReq(now int64, m Message) Message {
  reply := Message{Kind: AppendResp, From: nd.ID, To: m.From, Term: nd.Term}
  if m.Term < nd.Term {
    return reply // a stale leader: refuse, and tell it the newer term
  }
  nd.Role = Follower
  nd.LeaderID = m.From
  nd.resetTimer(now)
  if m.PrevIndex > nd.LastIndex() || nd.termAt(m.PrevIndex) != m.PrevTerm {
    reply.Match = min(nd.LastIndex(), m.PrevIndex-1) // the consistency check failed
    return reply
  }
  for i, e := range m.Entries {
    at := m.PrevIndex + 1 + i
    if at <= nd.LastIndex() && nd.Log[at-1].Term != e.Term {
      nd.Log = nd.Log[:at-1] // a conflict: drop this entry and all after it
    }
    if at > nd.LastIndex() {
      nd.Log = append(nd.Log, e)
    }
  }
  last := m.PrevIndex + len(m.Entries)
  if m.LeaderCommit > nd.Commit {
    nd.Commit = min(m.LeaderCommit, last)
  }
  reply.OK, reply.Match = true, last
  return reply
}

func (nd *Node) handleAppendResp(m Message) []Message {
  if nd.Role != Leader || m.Term != nd.Term {
    return nil
  }
  if !m.OK {
    nd.next[m.From] = max(1, min(nd.next[m.From]-1, m.Match+1)) // step back, then retry
    return []Message{nd.appendTo(m.From)}
  }
  nd.match[m.From] = max(nd.match[m.From], m.Match)
  nd.next[m.From] = nd.match[m.From] + 1
  nd.advanceCommit()
  return nil
}

func (nd *Node) advanceCommit() {
  for i := nd.LastIndex(); i > nd.Commit; i-- {
    if nd.Log[i-1].Term != nd.Term {
      break // count replicas only for entries of the current term
    }
    copies := 0
    for _, mi := range nd.match {
      if mi >= i {
        copies++
      }
    }
    if copies >= nd.majority() {
      nd.Commit = i
      return
    }
  }
}

// check tests the Raft safety properties on node i after it changed.
func (c *Cluster) check(i int) {
  nd := c.Nodes[i]
  if nd.Role == Leader {
    if prev, ok := c.leaders[nd.Term]; !ok {
      c.leaders[nd.Term] = i
      // Leader completeness: a new leader holds every committed entry.
      for k, e := range c.committed {
        if k >= len(nd.Log) || nd.Log[k] != e {
          c.Incomplete = true
          c.fail("t=%d: S%d leads term %d without committed entry %d", c.Now, i+1, nd.Term, k+1)
          break
        }
      }
    } else if prev != i {
      c.TwoLeaders = true
      c.fail("t=%d: two leaders in term %d: S%d and S%d", c.Now, nd.Term, prev+1, i+1)
    }
  }
  if nd.Commit > len(nd.Log) {
    c.Replaced = true
    c.fail("t=%d: S%d dropped committed entries: commit %d, log %d", c.Now, i+1, nd.Commit, len(nd.Log))
    return
  }
  // State machine safety: every node commits the same entry at each index.
  for k := c.checked[i] + 1; k <= nd.Commit; k++ {
    e := nd.Log[k-1]
    switch {
    case k <= len(c.committed) && c.committed[k-1] != e:
      c.Replaced = true
      c.fail("t=%d: S%d committed %v at %d, but %v was committed there", c.Now, i+1, e, k, c.committed[k-1])
    case k == len(c.committed)+1:
      c.committed = append(c.committed, e)
    }
  }
  c.checked[i] = max(c.checked[i], nd.Commit)
}