tara.model

/*
 * Copyright (c) 2016 Diego Ongaro.
 * Licensed under the MIT license.
 * For full license text, see LICENSE.md file in the repo root or
 * https://opensource.org/licenses/MIT
 */

type MaybeIndex : 0..9;
type StrictIndex : 1..9;
type NextIndex : 1..10;

type MaybeTerm : 0..9;
type StrictTerm : 1..9;

type ServerId : 1..4; // Begins with triplet

type Entry : record {
  term: StrictTerm,
  // value omitted
};

type Log : Vector<Entry>[StrictIndex];

type CandidatePeer : record {
  voteResponded: Boolean,
  voteGranted: Boolean,
  requestVoteAt: Time,
};

type LeaderPeer : record {
  nextIndex: NextIndex,
  matchIndex: MaybeIndex,
  appendEntriesAt: Time,
};

type Server : record {
  // Persistent state
  currentTerm: StrictTerm,
  vote: either {
    Available,
    Granted { to: ServerId },
  },
  log: Log,

  // Volatile state
  state: either {
    Follower,
    Candidate {
      peers: Array<CandidatePeer>[ServerId],
    },
    Leader {
      peers: Array<LeaderPeer>[ServerId],
    },
    Offline,
  },
  commitIndex: MaybeIndex,
  timeoutAt: Time,
};

var servers: Array<Server>[ServerId];

type Message : record {
  from: ServerId,
  to: ServerId,
  sentAt: Time,
  deliverAt: Time,
  payload: either {
    RequestVoteRequest {
      term: StrictTerm,
      lastLogTerm: MaybeTerm,
      lastLogIndex: MaybeIndex,
    },
    RequestVoteResponse {
      term: StrictTerm,
      granted: Boolean,
    },
    AppendEntriesRequest {
      term: StrictTerm,
      prevLogIndex: MaybeIndex,
      prevLogTerm: MaybeTerm,
      entries: Log,
      commitIndex: MaybeIndex,
    },
    AppendEntriesResponse {
      term: StrictTerm,
      success: Boolean,
      matchIndex: MaybeIndex,
    },
  },
};

var network : MultiSet<Message>[0..20];

////////// History Variables //////////

// Used to check election safety property.
type ElectionResult : either {
  NoWinner,
  Winner { won: ServerId }
};
var electionHistory : Array<ElectionResult>[StrictTerm];

type Election: record {
  start: Time,
  stable: Time,
};
var elections: Output<Election>;
type CurrentElection : either {
  Started { start: Time },
  NotStarted,
};
var currentElection : CurrentElection;

// Used to check leader completeness property
type CommittedEntry : record {
  entry: Entry,
  termCommitted: StrictTerm,
};
var committedEntries : Vector<CommittedEntry>[StrictIndex];

var resetStableLeader : Boolean;

function makeElectionTimeout() -> Time {
  return later(urandomRange(100000, 200000));
}

function lastTerm(log: Log) -> MaybeTerm {
  if empty(log) {
    return 0;
  } else {
    return log[size(log)].term;
  }
}

function quorum(serverSet: Set<ServerId>[ServerId]) -> Boolean {
  return size(serverSet) * 2 > size(servers);
}

function sendMessage(message: Message) {
  message.sentAt = later(0);
  message.deliverAt = later(urandomRange(10000, 20000));
  push(network, message);
}

function init() {
  for server in servers {
    server.timeoutAt = makeElectionTimeout();
  }
}

// initialization
init();

/*
rule startOver {
  for server in servers {
    if server.currentTerm == 9 || full(server.log) {
      reset;
      init();
    }
  }
}
*/

function restart(serverId : ServerId) {
  servers[serverId].state = Follower;
  servers[serverId].commitIndex = 0;
  servers[serverId].timeoutAt = makeElectionTimeout();
}

external shutdown for server in servers {
  server.state = Offline;
}

external startup for serverId, server in servers {
  if server.state == Offline {
    restart(serverId);
  }
}

function stableLeader() -> Boolean {
  for serverId, server in servers {
    match server.state {
      Leader {
        var followers : Set<ServerId>[ServerId];
        push(followers, serverId);
        for server2Id, server2 in servers {
          if serverId == server2Id {
            continue;
          }
          match server2.state {
            Offline {}
            Follower {
              if server2.currentTerm == server.currentTerm {
                push(followers, server2Id);
              } else {
                return False;
              }
            }
            default {
              return False;
            }
          }
        }
        return quorum(followers);
      }
      default {}
    }
  }
  return False;
}

rule outputOnStableLeader {
  match currentElection {
    Started(s) {
      if stableLeader() {
        push(elections, Election {
          start: s.start,
          stable: later(0),
        });
        if resetStableLeader {
          reset;
          resetStableLeader = True; // (reset cleared this)
          init();
        }
        currentElection = NotStarted;
      }
    }
    NotStarted {
      if !stableLeader() {
        currentElection = Started { start: later(0) };
      }
    }
  }
}

rule startNewElection for serverId, server in servers {
  var doTimeout : Boolean;
  match server.state {
    Offline   { doTimeout = False; }
    Follower  { doTimeout = True; }
    Candidate { doTimeout = True; }
    Leader    { doTimeout = False; }
  }
  if (past(server.timeoutAt) &&
      doTimeout &&
      server.currentTerm < 9 /* max term */) {
    server.currentTerm += 1;
    server.vote = Granted { to: serverId };
    var peers: Array<CandidatePeer>[ServerId];
    peers[serverId] = CandidatePeer {
      voteResponded: True,
      voteGranted: True,
    };
    server.state = Candidate { peers: peers };
    server.timeoutAt = makeElectionTimeout();
  }
}

rule sendRequestVote
for serverId, server in servers
for targetId, _target in servers {
  if !full(network) && serverId != targetId {
    match server.state {
      Candidate(cstate) {
        if !cstate.peers[targetId].voteResponded &&
           past(cstate.peers[targetId].requestVoteAt) {
          sendMessage(Message {
            from: serverId,
            to: targetId,
            payload: RequestVoteRequest {
              term: server.currentTerm,
              lastLogTerm: lastTerm(server.log),
              lastLogIndex: size(server.log),
            },
          });
          cstate.peers[targetId].requestVoteAt = later(50000);
          server.state = cstate;
        }
      }
      default { /* do nothing */ }
    }
  }
}

rule becomeLeader for serverId, server in servers {
  match server.state {
    Candidate(cstate) {
      var granted : Set<ServerId>[ServerId];
      for peerId, peer in cstate.peers {
        if peer.voteGranted {
          push(granted, peerId);
        }
      }
      if quorum(granted) {
        var peers : Array<LeaderPeer>[ServerId];
        peers[serverId].matchIndex = size(server.log);
        peers[serverId].nextIndex = size(server.log) + 1;
        server.state = Leader {
          peers: peers,
        };
        assert electionHistory[server.currentTerm] == NoWinner;
        electionHistory[server.currentTerm] = Winner { won: serverId };
      }
    }
    default { /* do nothing */ }
  }
}

rule sendAppendEntries
for serverId, server in servers
for targetId, _target in servers {
  if !full(network) && serverId != targetId {
    match server.state {
      Leader(lstate) {
        if past(lstate.peers[targetId].appendEntriesAt) {
          var prevLogIndex : MaybeIndex = lstate.peers[targetId].nextIndex - 1;
          var prevLogTerm : MaybeTerm = 0;
          if prevLogIndex > 0 {
            prevLogTerm = server.log[prevLogIndex].term;
          }
          var entries : Log;
          var index : NextIndex = prevLogIndex + 1;
          var limit : MaybeIndex = 1;
          while index <= size(server.log) && limit > 0 {
            push(entries, server.log[index]);
            index += 1;
            limit -= 1;
          }
          var commitIndex : MaybeIndex = server.commitIndex;
          if server.commitIndex > index {
            commitIndex = index;
          }
          sendMessage(Message {
            from: serverId,
            to: targetId,
            payload: AppendEntriesRequest {
              term: server.currentTerm,
              prevLogIndex: prevLogIndex,
              prevLogTerm: prevLogTerm,
              entries: entries,
              commitIndex: commitIndex,
            },
          });
          lstate.peers[targetId].appendEntriesAt = later(50000);
          server.state = lstate;
        }
      }
      default { /* do nothing */ }
    }
  }
}

external clientRequest for serverId, server in servers {
  if !full(server.log) {
    match server.state {
      Offline { /* do nothing */ },
      Follower { /* redirect client */ },
      Candidate { /* redirect client */ },
      Leader(lstate) {
        push(server.log, Entry {
          term: server.currentTerm,
          // value: to client-provided command
        });
        for peerId, peer in lstate.peers {
          if peerId == serverId {
            peer.nextIndex += 1;
            peer.matchIndex += 1;
          } else {
            peer.appendEntriesAt = 0;
          }
        }
        server.state = lstate;
      }
    }
  }
}

rule advanceCommitIndex for serverId, server in servers {
  match server.state {
    Leader(lstate) {
      var quorumMatchIndex : MaybeIndex = 0;
      for peerId, peer in lstate.peers {
        // The set of servers with match index >= peer.matchIndex.
        var geq : Set<ServerId>[ServerId];
        for p2Id, p2 in lstate.peers {
          if p2.matchIndex >= peer.matchIndex {
            push(geq, p2Id);
          }
        }
        if quorum(geq) && peer.matchIndex > quorumMatchIndex {
          quorumMatchIndex = peer.matchIndex;
        }
      }
      if (quorumMatchIndex > 0 &&
          server.log[quorumMatchIndex].term == server.currentTerm) {
        // save new commit index
        server.commitIndex = quorumMatchIndex;

        // update committedEntries history variable
        for index, entry in server.log {
          if index > server.commitIndex {
            break;
          }
          if index == size(committedEntries) + 1 {
            push(committedEntries, CommittedEntry {
              entry: entry,
              termCommitted: server.currentTerm,
            });
          } else {
            assert committedEntries[index].entry == entry;
            if committedEntries[index].termCommitted > server.currentTerm {
              committedEntries[index].termCommitted = server.currentTerm;
            }
          }
        }
      }
    } // Leader
    default { /* do nothing */}
  } // match
}

rule handleMessage for message in network {
  if past(message.deliverAt) {
    var from : ServerId = message.from;
    var to : ServerId = message.to;
    if servers[message.to].state == Offline {
      remove(network, message);
    } else {
      match message.payload {
        RequestVoteRequest(request) {
          if request.term > servers[to].currentTerm {
            servers[to].currentTerm = request.term;
            servers[to].state = Follower;
            servers[to].vote = Available;
            // leave request in network for further processing
          } else {
            // check that vote is available and if so, record vote
            var granted : Boolean = False;
            if (request.term == servers[to].currentTerm &&
                (request.lastLogTerm > lastTerm(servers[to].log) ||
                 (request.lastLogTerm == lastTerm(servers[to].log) &&
                  request.lastLogIndex >= size(servers[to].log)))) {
              match servers[to].vote {
                Available {
                  granted = True;
                  servers[to].vote = Granted { to: from };
                }
                Granted(g) {
                  granted = (g.to == from);
                }
              }
            }
            if granted {
              servers[to].timeoutAt = makeElectionTimeout();
            }
            remove(network, message);
            sendMessage(Message {
              from: to,
              to: from,
              payload: RequestVoteResponse {
                term: servers[to].currentTerm,
                granted: granted,
              },
            });
          }
        } // RequestVoteRequest

        RequestVoteResponse(reply) {
          if reply.term == servers[to].currentTerm {
            match servers[to].state {
              Offline { assert False; }
              Follower {
                // Must have restarted to get back into Follower state after
                // requesting votes. Just ignore it.
              }
              Candidate(cstate) {
                cstate.peers[from].voteResponded = True;
                cstate.peers[from].voteGranted = reply.granted;
                servers[to].state = cstate;
              }
              Leader { /* ignore */ }
            }
          } else if reply.term > servers[to].currentTerm {
              servers[to].currentTerm = reply.term;
              servers[to].state = Follower;
              servers[to].vote = Available;
          } else {
            // stale term: ignore
          }
          remove(network, message);
        } // RequestVoteResponse

        AppendEntriesRequest(request) {
          if request.term > servers[to].currentTerm {
            servers[to].currentTerm = request.term;
            servers[to].state = Follower;
            servers[to].vote = Available;
            // leave request in network for further processing
          } else {
            var rejectReply : Message = Message {
              from: to,
              to: from,
              payload: AppendEntriesResponse {
                term: servers[to].currentTerm,
                success: False,
                matchIndex: 0,
              },
            };
            if request.term < servers[to].currentTerm {
              remove(network, message);
              sendMessage(rejectReply);
            } else { // request.term == servers[to].currentTerm
              servers[to].timeoutAt = makeElectionTimeout();
              match servers[to].state {
                Offline { assert False; },
                Follower {
                  var logOk : Boolean = (
                    request.prevLogIndex == 0 ||
                      (request.prevLogIndex <= size(servers[to].log) &&
                       request.prevLogTerm ==
                        servers[to].log[request.prevLogIndex].term));
                  if logOk {
                    var done : Boolean = True;
                    var index : MaybeIndex = request.prevLogIndex;
                    for entry in request.entries {
                      index += 1;
                      if index <= size(servers[to].log) {
                        if servers[to].log[index].term == entry.term {
                          // already have this entry
                          continue;
                        } else {
                          // terms conflict at 'index': remove last entry in
                          // server's log
                          var lastIndex : StrictIndex = size(servers[to].log);
                          remove(servers[to].log, servers[to].log[lastIndex]);
                          // leave request in network for further processing
                          done = False;
                          break;
                        }
                      } else { // new entry
                        push(servers[to].log, entry);
                        assert index == size(servers[to].log);
                        // leave request in network for further processing
                        done = False;
                        break;
                      }
                    }
                    if done {
                      servers[to].commitIndex = request.commitIndex;
                      remove(network, message);
                      sendMessage(Message {
                        from: to,
                        to: from,
                        payload: AppendEntriesResponse {
                          term: servers[to].currentTerm,
                          success: True,
                          matchIndex: request.prevLogIndex + size(request.entries),
                        },
                      });
                    } else {
                      // leave request in network for further processing
                    }
                  } else {
                    remove(network, message);
                    sendMessage(rejectReply);
                  }
                }
                Candidate {
                  servers[to].state = Follower;
                  // leave request in network for further processing
                }
                Leader {
                  assert False;
                }
              }
            }
          }
        } // AppendEntriesRequest

        AppendEntriesResponse(reply) {
          if reply.term == servers[to].currentTerm {
            match servers[to].state {
              Offline { assert False; }
              Follower {
                // Must have restarted to get back into Follower state after
                // AppendEntries. Just ignore it.
              }
              Candidate {
                // Must have restarted to get back into Candidate state after
                // AppendEntries. Just ignore it.
              }
              Leader(lstate) {
                if reply.success {
                  lstate.peers[from].nextIndex = reply.matchIndex + 1;
                  lstate.peers[from].matchIndex = reply.matchIndex;
                  servers[to].state = lstate;
                } else {
                  if lstate.peers[from].nextIndex > 1 {
                    lstate.peers[from].nextIndex -= 1;
                  }
                }
              }
            }
          } else if reply.term > servers[to].currentTerm {
            servers[to].currentTerm = reply.term;
            servers[to].state = Follower;
            servers[to].vote = Available;
          } else {
            // stale term: ignore
          }
          remove(network, message);
        } // AppendEntriesResponse
      } // match message payload
    } // if online
  } // if past delivery time
}

external dropMessage for message in network {
  remove(network, message);
}

external duplicateMessage for message in network {
  if !full(network) {
    var messageCopy : Message = message;
    messageCopy.deliverAt = later(urandomRange(10000, 20000));
    push(network, messageCopy);
  }
}

invariant VotesConsistent {
  for serverId, server in servers {
    match server.state {
      Candidate(cstate) {
        for peerId, peer in cstate.peers {
          if peerId == serverId {
            assert peer.voteGranted;
          }
          if peer.voteGranted {
            assert peer.voteResponded;
            if servers[peerId].currentTerm == server.currentTerm {
              match servers[peerId].vote {
                Available { assert False; }
                Granted(g) { assert g.to == serverId; }
              }
            }
          }
        }
      }
      default {}
    }
  }
}

invariant LeaderPeersConsistent {
  for serverId, server in servers {
    match server.state {
      Leader(lstate) {
        for peerId, peer in lstate.peers {
          assert 1 <= peer.nextIndex;
          assert peer.nextIndex <= size(server.log) + 1;
          assert 0 <= peer.matchIndex;
          assert peer.matchIndex <= size(server.log);
          if servers[peerId].currentTerm == server.currentTerm {
            assert peer.matchIndex <= size(servers[peerId].log);
          }
          if peerId == serverId {
            assert peer.nextIndex == size(server.log) + 1;
            assert peer.matchIndex == size(server.log);
          }
        }
      }
      default {}
    }
  }
}

invariant CommitIndexConsistent {
  for server in servers {
    assert server.commitIndex <= size(server.log);
  }
}

invariant LogTermsMonotonic {
  for server in servers {
    var term : MaybeTerm = 0;
    for entry in server.log {
      assert entry.term >= term;
      term = entry.term;
    }
  }
}

invariant LogTermLeqCurrentTerm {
  for server in servers {
    assert lastTerm(server.log) <= server.currentTerm;
  }
}

invariant LeaderWonElection {
  for serverId, server in servers {
    match server.state {
      Leader {
        assert (electionHistory[server.currentTerm] ==
          Winner { won: serverId });
      }
      default {}
    }
  }
}

invariant LogMatching {
  for server1 in servers {
    for server2 in servers {
      var index : MaybeIndex = size(server1.log);
      if size(server2.log) < index {
        index = size(server2.log);
      }
      while index > 0 {
        if server1.log[index].term == server2.log[index].term {
          while index > 0 {
            assert server1.log[index] == server2.log[index];
            index -= 1;
          }
          break;
        } else {
          index -= 1;
        }
      }
    }
  }
}

invariant LeaderCompleteness {
  for server in servers {
    match server.state {
      Leader {
        for index, centry in committedEntries {
          if centry.termCommitted <= server.currentTerm {
            assert server.log[index] == centry.entry;
          }
        }
      }
      default {}
    }
  }
}

invariant CommittedEntriesAgreement {
  for server in servers {
    for index, centry in committedEntries {
      if index > server.commitIndex {
        break;
      }
      assert server.log[index] == centry.entry;
    }
  }
}