text: section \<open>Implementation\<close>
  
  text \<open>This section presents the implementation of the algorithm.\<close>
  
  theory Implementation
    imports Preliminaries
  begin
  
  subsection \<open>Protocol messages\<close>
  
  text \<open>The
  proven-safe core of the protocol works by sending messages as described here. The remainder of the
  protocol may send other messages too, and may drop, reorder or duplicate any of these messages, but
  must not send these messages itself to ensure safety. Another way of thinking of these messages is
  to consider them as ``fire-and-forget'' RPC invocations that, on receipt, call some local method, maybe
  update the receiving node's state, and maybe yield some further messages. The @{type nat} parameter to each
  message refers to a slot number.\<close>
  
  datatype TermOption = NO_TERM | SomeTerm Term
  
  instantiation TermOption :: linorder
  begin
  
  fun less_TermOption :: "TermOption \<Rightarrow> TermOption \<Rightarrow> bool"
    where "t < NO_TERM = False"
    | "NO_TERM < SomeTerm t = True"
    | "SomeTerm t\<^sub>1 < SomeTerm t\<^sub>2 = (t\<^sub>1 < t\<^sub>2)"
  
  definition less_eq_TermOption :: "TermOption \<Rightarrow> TermOption \<Rightarrow> bool"
    where "(t\<^sub>1 :: TermOption) \<le> t\<^sub>2 \<equiv> t\<^sub>1 = t\<^sub>2 \<or> t\<^sub>1 < t\<^sub>2"
  
  instance proof
    fix x y z :: TermOption
    show "(x < y) = (x \<le> y \<and> \<not> y \<le> x)" unfolding less_eq_TermOption_def apply auto
      using less_TermOption.elims apply fastforce
      by (metis less_TermOption.elims(2) less_TermOption.simps(3) less_not_sym)
  
    show "x \<le> x" by (simp add: less_eq_TermOption_def)
  
    show "x \<le> y \<Longrightarrow> y \<le> z \<Longrightarrow> x \<le> z" unfolding less_eq_TermOption_def apply auto
      by (metis TermOption.distinct(1) TermOption.inject dual_order.strict_trans less_TermOption.elims(2) less_TermOption.elims(3))
  
    show "x \<le> y \<Longrightarrow> y \<le> x \<Longrightarrow> x = y" unfolding less_eq_TermOption_def apply auto
      using \<open>(x < y) = (x \<le> y \<and> \<not> y \<le> x)\<close> less_eq_TermOption_def by blast
  
    show "x \<le> y \<or> y \<le> x" unfolding less_eq_TermOption_def apply auto
      by (metis TermOption.distinct(1) TermOption.inject less_TermOption.elims(3) neqE)
  qed
  
  end
  
  lemma NO_TERM_le [simp]: "NO_TERM \<le> t" by (cases t, simp_all add: less_eq_TermOption_def)
  lemma le_NO_TERM [simp]: "(t \<le> NO_TERM) = (t = NO_TERM)" by (cases t, simp_all add: less_eq_TermOption_def)
  lemma le_SomeTerm [simp]: "(SomeTerm t\<^sub>1 \<le> SomeTerm t\<^sub>2) = (t\<^sub>1 \<le> t\<^sub>2)" by (auto simp add: less_eq_TermOption_def)
  
  datatype Message
    = StartJoin Term
    | Vote Slot Term TermOption
    | ClientValue Value
    | PublishRequest Slot Term Value
    | PublishResponse Slot Term
    | ApplyCommit Slot Term
    | CatchUpRequest
    | CatchUpResponse Slot "Node set" ClusterState
    | DiscardJoinVotes
    | Reboot
  
  text \<open>Some prose descriptions of these messages follows, in order to give a bit more of an
  intuitive understanding of their purposes.\<close>
  
  text \<open>The message @{term "StartJoin t"} may be sent by any node to attempt to start a master
  election in the given term @{term t}.\<close>
  
  text \<open>The message @{term "Vote i t a"} may be sent by a node in response
  to a @{term StartJoin} message. It indicates that the sender knows all committed values for slots
  strictly below @{term i}, and that the sender will no longer vote (i.e. send an @{term
  PublishResponse}) in any term prior to @{term t}. The field @{term a} is either @{term
  None} or @{term "Some t'"}. In the former case this indicates that
  the node has not yet sent any @{term PublishResponse} message in slot @{term i}, and in the latter
  case it indicates that the largest term in which it has previously sent an @{term PublishResponse}
  message is @{term t'}.  All
  nodes must avoid sending a @{term Vote} message to two different masters in the same term.\<close>
  
  text \<open>The message @{term "ClientValue x"} may be sent by any node and indicates an attempt to
  reach consensus on the value @{term x}.\<close>
  
  text \<open>The message @{term "PublishRequest i t v"} may be sent by the elected master of term
  @{term t} to request the other master-eligible nodes to vote for value @{term v} to be committed in
  slot @{term i}.\<close>
  
  text \<open>The message @{term "PublishResponse i t"} may be sent by node in response to
  the corresponding @{term PublishRequest} message, indicating that the sender votes for the value
  proposed by the master of term @{term t} to be committed in slot @{term i}.\<close>
  
  text \<open>The message @{term "ApplyCommit i t"} indicates that the value proposed by the master of
  term @{term t} in slot @{term i} received a quorum of votes and is therefore committed.\<close>
  
  text \<open>The message @{term Reboot} may be sent by any node to represent the restart of a node, which
  loses any ephemeral state.\<close>
  
  text \<open>The abstract model of Zen keeps track of the set of all messages that have ever been
  sent, and asserts that this set obeys certain invariants, listed below. Further below, it will be
  shown that these invariants imply that each slot obeys the @{term oneSlot} invariants above and
  hence that each slot cannot see inconsistent committed values.\<close>
  
  datatype Destination = Broadcast | OneNode Node
  
  record RoutedMessage =
    sender :: Node
    destination :: Destination
    payload :: Message
  
  text \<open>It will be useful to be able to choose the optional term with the greater term,
  so here is a function that does that.\<close>
  
  subsection \<open>Node implementation\<close>
  
  text \<open>Each node holds the following local data.\<close>
  
  record TermValue =
    tvTerm :: Term
    tvValue :: Value
  
  record NodeData =
    currentNode :: Node
    currentTerm :: Term
    (* committed state *)
    firstUncommittedSlot :: Slot
    currentVotingNodes :: "Node set"
    currentClusterState :: ClusterState
    (* accepted state *)
    lastAcceptedData :: "TermValue option"
    (* election state *)
    joinVotes :: "Node set"
    electionWon :: bool
    (* publish state *)
    publishPermitted :: bool
    publishVotes :: "Node set"
  
  definition lastAcceptedValue :: "NodeData \<Rightarrow> Value"
    where "lastAcceptedValue nd \<equiv> tvValue (THE lad. lastAcceptedData nd = Some lad)"
  
  definition lastAcceptedTerm :: "NodeData \<Rightarrow> TermOption"
    where "lastAcceptedTerm nd \<equiv> case lastAcceptedData nd of None \<Rightarrow> NO_TERM | Some lad \<Rightarrow> SomeTerm (tvTerm lad)"
  
  definition isQuorum :: "NodeData \<Rightarrow> Node set \<Rightarrow> bool"
    where "isQuorum nd q \<equiv> q \<in> majorities (currentVotingNodes nd)"
  
  lemma lastAcceptedValue_joinVotes_update[simp]: "lastAcceptedValue (joinVotes_update f nd) = lastAcceptedValue nd" by (simp add: lastAcceptedValue_def)
  lemma lastAcceptedTerm_joinVotes_update[simp]: "lastAcceptedTerm (joinVotes_update f nd) = lastAcceptedTerm nd" by (simp add: lastAcceptedTerm_def)
  
  lemma lastAcceptedValue_electionWon_update[simp]: "lastAcceptedValue (electionWon_update f nd) = lastAcceptedValue nd" by (simp add: lastAcceptedValue_def)
  lemma lastAcceptedTerm_electionWon_update[simp]: "lastAcceptedTerm (electionWon_update f nd) = lastAcceptedTerm nd" by (simp add: lastAcceptedTerm_def)
  
  text \<open>This method publishes a value via a @{term PublishRequest} message.\<close>
  
  definition publishValue :: "Value \<Rightarrow> NodeData \<Rightarrow> (NodeData * Message option)"
    where
      "publishValue x nd \<equiv>
          if electionWon nd \<and> publishPermitted nd
                then ( nd \<lparr> publishPermitted := False \<rparr>
                     , Some (PublishRequest
                               (firstUncommittedSlot nd)
                               (currentTerm nd) x) )
                else (nd, None)"
  
  text \<open>This method updates the node's current term (if necessary) and discards any data associated
  with the previous term.\<close>
  
  definition ensureCurrentTerm :: "Term \<Rightarrow> NodeData \<Rightarrow> NodeData"
    where
      "ensureCurrentTerm t nd \<equiv>
          if t \<le> currentTerm nd
              then nd
              else nd
                \<lparr> joinVotes := {}
                , currentTerm := t
                , electionWon := False
                , publishPermitted := True
                , publishVotes := {} \<rparr>"
  
  text \<open>This method updates the node's state on receipt of a vote (a @{term Vote}) in an election.\<close>
  
  definition addElectionVote :: "Node \<Rightarrow> Slot => TermOption \<Rightarrow> NodeData \<Rightarrow> NodeData"
    where
      "addElectionVote s i a nd \<equiv> let newVotes = insert s (joinVotes nd)
        in nd \<lparr> joinVotes := newVotes
              , electionWon := isQuorum nd newVotes \<rparr>"
  
  text \<open>Clients request the cluster to achieve consensus on certain values using the @{term ClientValue}
  message which is handled as follows.\<close>
  
  definition handleClientValue :: "Value \<Rightarrow> NodeData \<Rightarrow> (NodeData * Message option)"
    where
      "handleClientValue x nd \<equiv> if lastAcceptedTerm nd = NO_TERM then publishValue x nd else (nd, None)"
  
  text \<open>A @{term StartJoin} message is checked for acceptability and then handled by updating the
  node's term and yielding a @{term Vote} message as follows.\<close>
  
  definition handleStartJoin :: "Term \<Rightarrow> NodeData \<Rightarrow> (NodeData * Message option)"
    where
      "handleStartJoin t nd \<equiv>
          if currentTerm nd < t
            then ( ensureCurrentTerm t nd
                 , Some (Vote (firstUncommittedSlot nd)
                                       t
                                      (lastAcceptedTerm nd)))
            else (nd, None)"
  
  text \<open>A @{term Vote} message is checked for acceptability and then handled as follows, perhaps
  yielding a @{term PublishRequest} message.\<close>
  
  definition handleVote :: "Node \<Rightarrow> Slot \<Rightarrow> Term \<Rightarrow> TermOption \<Rightarrow> NodeData \<Rightarrow> (NodeData * Message option)"
    where
      "handleVote s i t a nd \<equiv>
           if t = currentTerm nd
               \<and> (i < firstUncommittedSlot nd
                  \<or> (i = firstUncommittedSlot nd \<and> a \<le> lastAcceptedTerm nd))
            then let nd1 = addElectionVote s i a nd
                 in (if lastAcceptedTerm nd = NO_TERM then (nd1, None) else publishValue (lastAcceptedValue nd1) nd1)
            else (nd, None)"
  
  text \<open>A @{term PublishRequest} message is checked for acceptability and then handled as follows,
  yielding a @{term PublishResponse} message.\<close>
  
  definition handlePublishRequest :: "Slot \<Rightarrow> Term \<Rightarrow> Value \<Rightarrow> NodeData \<Rightarrow> (NodeData * Message option)"
    where
      "handlePublishRequest i t x nd \<equiv>
            if i = firstUncommittedSlot nd
                  \<and> t = currentTerm nd
            then ( nd \<lparr> lastAcceptedData := Some \<lparr> tvTerm = t, tvValue = x \<rparr> \<rparr>
                 , Some (PublishResponse i t))
            else (nd, None)"
  
  text \<open>This method sends an @{term ApplyCommit} message if a quorum of votes has been received.\<close>
  
  definition commitIfQuorate :: "NodeData \<Rightarrow> (NodeData * Message option)"
    where
      "commitIfQuorate nd = (nd, if isQuorum nd (publishVotes nd)
                                    then Some (ApplyCommit (firstUncommittedSlot nd) (currentTerm nd)) else None)"
  
  text \<open>A @{term PublishResponse} message is checked for acceptability and handled as follows. If
  this message, together with the previously-received messages, forms a quorum of votes then the
  value is committed, yielding an @{term ApplyCommit} message.\<close>
  
  definition handlePublishResponse :: "Node \<Rightarrow> Slot \<Rightarrow> Term \<Rightarrow> NodeData \<Rightarrow> (NodeData * Message option)"
    where
      "handlePublishResponse s i t nd \<equiv>
          if i = firstUncommittedSlot nd \<and> t = currentTerm nd
          then commitIfQuorate (nd \<lparr> publishVotes := insert s (publishVotes nd) \<rparr>)
          else (nd, None)"
  
  text \<open>This method updates the node's state when a value is committed.\<close>
  
  definition applyAcceptedValue :: "NodeData \<Rightarrow> NodeData"
    where
      "applyAcceptedValue nd \<equiv> case lastAcceptedValue nd of
          NoOp \<Rightarrow> nd
        | Reconfigure votingNodes \<Rightarrow> nd
            \<lparr> currentVotingNodes := set votingNodes
            , electionWon := joinVotes nd \<in> majorities (set votingNodes) \<rparr>
        | ClusterStateDiff diff \<Rightarrow> nd \<lparr> currentClusterState := diff (currentClusterState nd) \<rparr>"
  
  text \<open>An @{term ApplyCommit} message is applied to the current node's state, updating its configuration
  and \texttt{ClusterState} via the @{term applyValue} method. It yields no messages.\<close>
  
  definition handleApplyCommit :: "Slot \<Rightarrow> Term \<Rightarrow> NodeData \<Rightarrow> NodeData"
    where
      "handleApplyCommit i t nd \<equiv>
          if i = firstUncommittedSlot nd \<and> lastAcceptedTerm nd = SomeTerm t
            then (applyAcceptedValue nd)
                       \<lparr> firstUncommittedSlot := i + 1
                       , lastAcceptedData := None
                       , publishPermitted := True
                       , publishVotes := {} \<rparr>
            else nd"
  
  definition handleCatchUpRequest :: "NodeData \<Rightarrow> (NodeData * Message option)"
    where
      "handleCatchUpRequest nd = (nd, Some (CatchUpResponse (firstUncommittedSlot nd)
                                                (currentVotingNodes nd) (currentClusterState nd)))"
  
  definition handleCatchUpResponse :: "Slot \<Rightarrow> Node set \<Rightarrow> ClusterState \<Rightarrow> NodeData \<Rightarrow> NodeData"
    where
      "handleCatchUpResponse i conf cs nd \<equiv>
        if firstUncommittedSlot nd < i
          then nd \<lparr> firstUncommittedSlot := i
                  , publishPermitted := True
                  , publishVotes := {}
                  , currentVotingNodes := conf
                  , currentClusterState := cs
                  , lastAcceptedData := None
                  , joinVotes := {}
                  , electionWon := False \<rparr>
          else nd"
  
  text \<open>A @{term Reboot} message simulates the effect of a reboot, discarding any ephemeral state but
  preserving the persistent state. It yields no messages.\<close>
  
  definition handleReboot :: "NodeData \<Rightarrow> NodeData"
    where
      "handleReboot nd \<equiv>
        \<lparr> currentNode = currentNode nd
        , currentTerm = currentTerm nd
        , firstUncommittedSlot = firstUncommittedSlot nd
        , currentVotingNodes = currentVotingNodes nd
        , currentClusterState = currentClusterState nd
        , lastAcceptedData = lastAcceptedData nd
        , joinVotes = {}
        , electionWon = False
        , publishPermitted = False
        , publishVotes = {} \<rparr>"
  
  text \<open>A @{term DiscardJoinVotes} message discards the votes received by a node. It yields
  no messages.\<close>
  
  definition handleDiscardJoinVotes :: "NodeData \<Rightarrow> NodeData"
    where
    "handleDiscardJoinVotes nd \<equiv> nd \<lparr> electionWon := False, joinVotes := {} \<rparr>"
  
  text \<open>This function dispatches incoming messages to the appropriate handler method, and
  routes any responses to the appropriate places. In particular, @{term Vote} messages
  (sent by the @{term handleStartJoin} method) and
  @{term PublishResponse} messages (sent by the @{term handlePublishRequest} method) are
  only sent to a single node, whereas all other responses are broadcast to all nodes.\<close>
  
  definition ProcessMessage :: "NodeData \<Rightarrow> RoutedMessage \<Rightarrow> (NodeData * RoutedMessage option)"
    where
      "ProcessMessage nd msg \<equiv>
        let respondTo =
            (\<lambda> d (nd, mmsg). case mmsg of
                 None \<Rightarrow> (nd, None)
               | Some msg \<Rightarrow> (nd,
                   Some \<lparr> sender = currentNode nd, destination = d,
                               payload = msg \<rparr>));
            respondToSender = respondTo (OneNode (sender msg));
            respondToAll    = respondTo Broadcast
        in
          if destination msg \<in> { Broadcast, OneNode (currentNode nd) }
          then case payload msg of
            StartJoin t
                \<Rightarrow> respondToSender (handleStartJoin t nd)
            | Vote i t a
                \<Rightarrow> respondToAll (handleVote (sender msg) i t a nd)
            | ClientValue x
                \<Rightarrow> respondToAll (handleClientValue x nd)
            | PublishRequest i t x
                \<Rightarrow> respondToSender (handlePublishRequest i t x nd)
            | PublishResponse i t
                \<Rightarrow> respondToAll (handlePublishResponse (sender msg) i t nd)
            | ApplyCommit i t
                \<Rightarrow> (handleApplyCommit i t nd, None)
            | CatchUpRequest
                \<Rightarrow> respondToSender (handleCatchUpRequest nd)
            | CatchUpResponse i conf cs
                \<Rightarrow> (handleCatchUpResponse i conf cs nd, None)
            | DiscardJoinVotes
                \<Rightarrow> (handleDiscardJoinVotes nd, None)
            | Reboot
                \<Rightarrow> (handleReboot nd, None)
          else (nd, None)"
  
  text \<open>Nodes are initialised to this state. The data required is the initial configuration, @{term Q\<^sub>0}
  and the initial \texttt{ClusterState}, here shown as @{term "ClusterState 0"}.\<close>
  
  definition initialNodeState :: "Node \<Rightarrow> NodeData"
    where "initialNodeState n =
        \<lparr> currentNode = n
        , currentTerm = 0
        , firstUncommittedSlot = 0
        , currentVotingNodes = V\<^sub>0
        , currentClusterState = CS\<^sub>0
        , lastAcceptedData = None
        , joinVotes = {}
        , electionWon = False
        , publishPermitted = False
        , publishVotes = {} \<rparr>"
  (* Note: publishPermitted could be True initially, but in the actual implementation we call the
  same constructor whether we're starting up from afresh or recovering from a reboot, and the value
  is really unimportant as we need to run an election in a new term before becoming master anyway,
  so it's hard to justify putting any effort into calculating different values for these two cases.
  Instead just set it to False initially.*)
  
  end
  