// Copyright 2015 The etcd Authors
// Copyright 2026 Leo Cheng
//
// 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.
// See the License for the specific language governing permissions and
// limitations under the License.

///|
/// A message-driven Raft server: the consensus core (`Node`) wrapped with
/// everything a real deployment needs to run it — the peer set and configuration,
/// per-follower replication progress, election and heartbeat timers with
/// randomized timeouts, pre-vote, and a deterministic PRNG so tests can replay a
/// run exactly. It communicates only through `Message`s: `tick` and `step`
/// return the messages to send, and a transport (real or simulated) delivers
/// them. This mirrors etcd's `raft.Node`/`Ready` split of logic from I/O.
pub struct RaftNode {
  core : Node
  id : String
  peers : Array[String]
  config : Membership
  progress : Map[String, Progress]
  votes : Map[String, Bool]
  // Whether this server runs the pre-vote probe before a real election (etcd's
  // Config.PreVote). Mutable so a mixed-version cluster can flip it mid-run, as a
  // rolling restart enables pre-vote replica by replica (etcd sets `r.preVote`
  // directly in newPreVoteMigrationCluster).
  mut pre_vote : Bool
  mut check_quorum : Bool
  mut in_pre_campaign : Bool
  mut leader_id : String?
  mut election_elapsed : Int32
  mut heartbeat_elapsed : Int32
  election_timeout : Int32
  heartbeat_timeout : Int32
  mut randomized_election_timeout : Int32
  // Byte cap on a single outbound AppendEntries batch (etcd's MaxSizePerMsg).
  max_msg_bytes : UInt64
  // Byte cap on the uncommitted log tail (etcd's MaxUncommittedEntriesSize); 0
  // disables the check. `uncommitted_size` is the running tally.
  max_uncommitted_size : UInt64
  mut uncommitted_size : UInt64
  // Per-follower replication flow-control window: at most `max_inflight`
  // outstanding AppendEntries and, when non-zero, at most `max_inflight_bytes`
  // outstanding payload bytes (etcd's MaxInflightMsgs / MaxInflightBytes). Both
  // are handed to each follower's `Progress` so its `Inflights` enforces them.
  max_inflight : Int32
  max_inflight_bytes : UInt64
  mut rng : UInt64
  // Highest log index whose configuration change has been folded into the live
  // configuration, so a committed ConfChange is applied exactly once.
  mut conf_applied : UInt64
  // The target of an in-flight leadership transfer, if any (Raft §3.10). While
  // set, the leader stops accepting proposals and hands off once the target is
  // caught up; it is cleared on timeout, step-down, or the target's removal.
  mut lead_transferee : String?
  // Linearizable read-index bookkeeping (Raft §6.4).
  read_only : ReadOnly
  // Read-index requests (origin id, opaque context) received before the leader
  // has committed an entry in its current term. etcd holds these in
  // `pendingReadIndexMessages` and releases them on the first in-term commit
  // rather than dropping the read.
  pending_read_index : Array[(String, Bytes)]
  // etcd's DisableProposalForwarding: when set, a follower drops client proposals
  // instead of forwarding them to the leader (for deployments that route writes
  // to the leader out-of-band). Off by default: a follower forwards.
  mut no_forward : Bool
  // The highest log index that carries a not-yet-applied configuration change
  // (etcd's `pendingConfIndex`). A leader refuses to propose a further change
  // until its applied index passes this, so at most one change is in flight.
  // Set conservatively to the last index on becoming leader, and reset on any
  // other state change.
  mut pending_conf_index : UInt64
  // etcd's DisableConfChangeValidation: turns off the propose-time checks below
  // (only-one-in-flight, joint-ordering). Off by default.
  no_conf_change_validation : Bool
  // etcd's StepDownOnRemoval: when set, a leader removed from the configuration
  // (or demoted to a learner) steps down. Off by default — matching etcd, whose
  // zero-valued Config leaves a removed leader in place until the new
  // configuration elects a replacement (etcd's `switchToConfig`).
  step_down_on_removal : Bool
  // Final appends owed to servers this leader is about to reconcile out of its
  // peer set: a committed removal drops a peer eagerly on apply, so before its
  // progress is discarded the leader stages the append carrying that commit here
  // and flushes it on the next broadcast, so the departing server learns of its
  // own removal instead of stalling on the pre-removal commit index. etcd keeps
  // the progress until the application applies the change, which has the same
  // effect.
  pending_farewell : Array[Message]
  // Self-addressed responses that must not take effect until the Ready that
  // produced them is durable (etcd's `msgsAfterAppend`): the candidate's own
  // vote and the leader's acknowledgement of its own append. They are stepped at
  // the advance boundary, which is what makes "became leader" land after the
  // Ready that persisted the vote rather than inside the campaign call.
  deferred : Array[Message]
  // Messages the protocol decided to send while folding a configuration change
  // (etcd's `switchToConfig` probes the new peers before returning, and reports
  // any further commit). They are collected here because the fold can happen
  // during `advance`, which returns nothing; `step` and the Ready driver drain
  // this buffer.
  pending : Array[Message]
  // Whether a configuration fold is already running: `maybe_commit` re-enters it
  // (committing a change can shrink the quorum and let more entries commit), and
  // the fold itself calls `maybe_commit`.
  mut applying_conf : Bool
  // Whether the most recent direct `propose` was refused (etcd's `Step`
  // returning `ErrProposalDropped`). `step_propose` clears it on entry and sets
  // it on every drop path, so a driver can report a dropped proposal instead of
  // silently losing it.
  mut last_propose_dropped : Bool
  // Whether those responses wait for the advance boundary. A caller that drives
  // the node through `RawNode` is saying when a Ready became durable, so the
  // responses wait (the faithful ordering). A caller that steps the node
  // directly has no such boundary — there is nothing to wait for — so the
  // candidate's self-vote and the leader's self-ack take effect at once.
  mut defer_responses : Bool
  // Diagnostics sink (etcd's `Config.Logger`); a no-op by default.
  logger : &Logger
  // State-transition tracer (etcd's `Config.TraceLogger`); a no-op by default.
  tracer : &Tracer
}

///|
/// Create a server with id `id`, the other voters `peers`, and a deterministic
/// seed. Election timeouts randomize in [election_timeout, 2*election_timeout),
/// the spread that keeps split votes rare (Raft §5.2). Heartbeats go out every
/// `heartbeat_timeout` ticks, which must be well below the election timeout.
///
/// `snapshot` starts the server on a compacted log: the snapshot's index and
/// term become the log's baseline, its commit index and applied watermark, and
/// its `ConfState` the configuration, exactly as etcd's `newRaft` reads them
/// back from `Storage.InitialState` (raft.go:439). A server restored that way
/// has no entries to replay, which is what makes a bootstrap at a non-zero index
/// reproducible in a test.
pub fn RaftNode::new(
  id : String,
  peers : Array[String],
  seed~ : UInt64 = 1,
  election_timeout~ : Int32 = 10,
  heartbeat_timeout~ : Int32 = 1,
  max_msg_bytes~ : UInt64 = 18446744073709551615UL,
  max_uncommitted_size~ : UInt64 = 0,
  max_inflight~ : Int32 = 256,
  max_inflight_bytes~ : UInt64 = 0,
  check_quorum~ : Bool = false,
  // etcd's Config.PreVote is opt-in (zero value false); default off to match.
  pre_vote~ : Bool = false,
  step_down_on_removal~ : Bool = false,
  disable_conf_change_validation~ : Bool = false,
  // etcd's zero value is ReadOnlySafe ("default and suggested"); a LeaseBased
  // default without check-quorum is a configuration etcd's validate() rejects.
  read_only_option~ : ReadOnlyOption = Safe,
  logger~ : &Logger = NopLogger::{ },
  tracer~ : &Tracer = NopTracer::{ },
  snapshot? : Snapshot,
  applied~ : UInt64 = 0,
) -> RaftNode {
  // Reject an unusable configuration at construction, as etcd's newRaft panics on
  // an invalid Config.validate() (raft.go:440). This convenience constructor
  // shares the exact checks with the explicit Config path.
  Config::new(
    id,
    peers,
    election_tick=election_timeout,
    heartbeat_tick=heartbeat_timeout,
    max_msg_bytes~,
    max_inflight~,
    max_inflight_bytes~,
    check_quorum~,
    read_only_option~,
  ).validate() catch {
    EmptyId => abort("invalid raft config: empty id")
    HeartbeatTickNotPositive =>
      abort("invalid raft config: heartbeat tick <= 0")
    ElectionTickNotGreater =>
      abort("invalid raft config: election tick <= heartbeat tick")
    MaxInflightNotPositive => abort("invalid raft config: max inflight <= 0")
    MaxInflightBytesTooSmall =>
      abort("invalid raft config: max inflight bytes < max message size")
    LeaseBasedNeedsCheckQuorum =>
      abort("invalid raft config: ReadOnlyLeaseBased requires CheckQuorum")
  }
  let all = [id]
  for p in peers {
    all.push(p)
  }
  let node = {
    core: Node::new(id),
    id,
    peers: peers.copy(),
    config: Membership::new(all),
    progress: Map([]),
    votes: Map([]),
    pre_vote,
    check_quorum,
    in_pre_campaign: false,
    leader_id: None,
    election_elapsed: 0,
    heartbeat_elapsed: 0,
    election_timeout,
    heartbeat_timeout,
    randomized_election_timeout: election_timeout,
    max_msg_bytes,
    max_uncommitted_size,
    uncommitted_size: 0,
    max_inflight,
    max_inflight_bytes,
    rng: seed,
    conf_applied: 0,
    lead_transferee: None,
    read_only: ReadOnly::new(),
    pending_read_index: [],
    no_forward: false,
    pending_conf_index: 0,
    no_conf_change_validation: disable_conf_change_validation,
    step_down_on_removal,
    pending_farewell: [],
    deferred: [],
    pending: [],
    applying_conf: false,
    last_propose_dropped: false,
    defer_responses: false,
    logger,
    tracer,
  }
  // Broadcasts and vote requests follow ascending id order upstream; keep the
  // peer set sorted from the start (a snapshot bootstrap re-sorts in
  // `reconcile_peers`).
  node.peers.sort()
  node.read_only.safe = read_only_option is Safe
  node.tracer.on_event(InitState(id))
  match snapshot {
    Some(snap) => {
      node.core.install_snapshot(snap)
      node.restore_conf_state(snap.conf_state)
    }
    None => ()
  }
  // A restart resumes with the watermark the application has already applied
  // (etcd's Config.Applied), set before the log position is announced.
  if applied > 0UL {
    node.advance_applied(applied)
  }
  node.reset_timeout()
  // etcd's `newRaft` announces the three facts a restart reconstructs: the
  // configuration it restored from storage, the state it settled into, and the
  // log position it resumes at (raft.go:439-497).
  node.logger.log(
    Info,
    "\{id} switched to configuration \{node.config.describe()}",
  )
  node.logger.log(
    Info,
    "\{id} became follower at term \{node.core.current_term()}",
  )
  node.log_new_raft()
  node
}

///|
/// The `newRaft` line: the peer list, term, commit, applied and log tail a
/// server starts from. One line, because a restart is a single event in a log.
fn RaftNode::log_new_raft(self : RaftNode) -> Unit {
  let voters = self.voter_nodes()
  self.logger.log(
    Info,
    "newRaft \{self.id} [peers: [\{voters.join(",")}], term: \{self.core.current_term()}, commit: \{self.core.commit_index}, applied: \{self.core.last_applied}, lastindex: \{self.core.last_log_index()}, lastterm: \{self.core.last_log_term()}]",
  )
}

///|
/// The voters of the current configuration, sorted (etcd's `VoterNodes`), for
/// status and diagnostics. Sorting makes the list deterministic regardless of
/// the order members were added.
pub fn RaftNode::voter_nodes(self : RaftNode) -> Array[String] {
  let out = self.config.voters()
  out.sort()
  out
}

///|
/// Whether this node may stand for election (etcd's `promotable`): it must be a
/// voter of the current configuration. A learner or a node that has been removed
/// is not promotable. (etcd also refuses while a snapshot is being applied; this
/// port installs snapshots synchronously, so there is no such in-progress state.)
pub fn RaftNode::promotable(self : RaftNode) -> Bool {
  self.config.contains(self.id)
}

///|
/// The next value of the node's deterministic PRNG (a 64-bit LCG). Randomizing
/// election timeouts from a per-node seed makes whole-cluster runs reproducible.
fn RaftNode::next_rand(self : RaftNode) -> UInt64 {
  self.rng = self.rng * 6364136223846793005UL + 1442695040888963407UL
  self.rng
}

///|
/// Reset both timers and pick a fresh randomized election timeout in
/// [election_timeout, 2*election_timeout). The spread is taken from the *high*
/// bits of the LCG: an LCG's low bits are notoriously non-random (the lowest
/// cycles with period 2), so `next_rand() % et` on the raw value never reaches
/// some residues — using the top bits gives the uniform spread §5.2 needs.
fn RaftNode::reset_timeout(self : RaftNode) -> Unit {
  self.election_elapsed = 0
  self.heartbeat_elapsed = 0
  let spread = ((self.next_rand() >> 33) % self.election_timeout.to_uint64()).to_int()
  self.randomized_election_timeout = self.election_timeout + spread
}

///|
/// This server's id.
pub fn RaftNode::id(self : RaftNode) -> String {
  self.id
}

///|
/// The current role (etcd's `SoftState.RaftState`). During the pre-vote probe
/// the core genuinely occupies the `PreCandidate` state — a first-class Raft
/// state, not a derived flag — having changed role without adopting the
/// hypothetical term.
pub fn RaftNode::role(self : RaftNode) -> Role {
  self.core.role()
}

///|
/// The randomized election timeout this server is currently counting toward, in
/// ticks. Exposed for tests that check the timeout stays within its band.
pub fn RaftNode::election_deadline(self : RaftNode) -> Int32 {
  self.randomized_election_timeout
}

///|
/// The current term.
pub fn RaftNode::term(self : RaftNode) -> UInt64 {
  self.core.current_term()
}

///|
/// Whether this server currently believes it is the leader.
pub fn RaftNode::is_leader(self : RaftNode) -> Bool {
  self.core.role() == Leader
}

///|
/// The leader this server last heard from, if any.
pub fn RaftNode::leader(self : RaftNode) -> String? {
  self.leader_id
}

///|
/// The underlying consensus core, for reading log/commit state in tests.
pub fn RaftNode::node(self : RaftNode) -> Node {
  self.core
}

///|
/// The commit index of the underlying core.
pub fn RaftNode::commit_index(self : RaftNode) -> UInt64 {
  self.core.commit_index
}

///|
/// The ids that have granted this server (pre-)votes so far, itself included.
fn RaftNode::granted(self : RaftNode) -> Array[String] {
  let out : Array[String] = []
  for id, ok in self.votes {
    if ok {
      out.push(id)
    }
  }
  out
}

///|
/// Advance the logical clock by one tick, returning the messages to send. A
/// leader beats every `heartbeat_timeout` ticks; a follower or candidate that
/// reaches its randomized election timeout starts a new campaign.
pub fn RaftNode::tick(self : RaftNode) -> Array[Message] {
  if self.core.role() == Leader {
    self.heartbeat_elapsed = self.heartbeat_elapsed + 1
    self.election_elapsed = self.election_elapsed + 1
    let out : Array[Message] = []
    // Once per election timeout, close the liveness window. With check-quorum
    // on, a leader that has not heard from a majority in that window steps down
    // so a partitioned-away leader cannot linger (etcd's CheckQuorum, §6.4).
    if self.election_elapsed >= self.election_timeout {
      self.election_elapsed = 0
      // A leadership transfer that has not completed within an election timeout
      // is abandoned, so proposals are not blocked indefinitely (§3.10).
      self.lead_transferee = None
      if self.check_quorum && !self.quorum_active() {
        self.logger.log(
          Warning,
          "\{self.id} stepped down to follower since quorum is not active",
        )
        self.step_down()
        return []
      }
      self.reset_recent_active()
    }
    if self.heartbeat_elapsed >= self.heartbeat_timeout {
      self.heartbeat_elapsed = 0
      for m in self.bcast_heartbeat() {
        out.push(m)
      }
    }
    return out
  }
  self.election_elapsed = self.election_elapsed + 1
  if self.election_elapsed >= self.randomized_election_timeout {
    return self.campaign()
  }
  []
}

///|
/// Turn on check-quorum and lease reads for this server.
pub fn RaftNode::enable_check_quorum(self : RaftNode) -> Unit {
  self.check_quorum = true
}

///|
/// Enable or disable the pre-vote probe at runtime (etcd sets `r.preVote`
/// directly). A mixed-version rolling restart flips replicas one at a time, so a
/// cluster momentarily runs some nodes with pre-vote and some without.
pub fn RaftNode::set_pre_vote(self : RaftNode, on : Bool) -> Unit {
  self.pre_vote = on
}

///|
/// Pin this server's randomized election timeout, as etcd's test helper
/// `SetRandomizedElectionTimeout` does. A run that has to be reproducible needs
/// the timeout to be a number the test chose, not one the election PRNG drew;
/// the node draws a fresh one on its next state change, so a caller that wants
/// it held sets it immediately before the ticks it means to control.
pub fn RaftNode::set_randomized_election_timeout(
  self : RaftNode,
  ticks : Int32,
) -> Unit {
  self.randomized_election_timeout = ticks
}

///|
/// Stand down to follower in the current term, e.g. on losing quorum contact.
/// This is etcd's `becomeFollower`, so it announces the transition like every
/// other state change does.
fn RaftNode::step_down(self : RaftNode) -> Unit {
  self.become_follower(self.core.current_term())
}

///|
/// Start a campaign. With pre-vote enabled the node first runs a pre-vote round
/// under a hypothetical next term without touching its own term or vote, so a
/// partitioned node cannot force real elections and inflate terms (Raft §9.6,
/// pre-vote). Winning the pre-vote — or pre-vote being disabled — starts the
/// real election.
pub fn RaftNode::campaign(self : RaftNode) -> Array[Message] {
  // A leader ignores a fresh campaign (etcd's `hup`: "already leader"): it must
  // not step itself down to candidate and inflate the term.
  if self.core.role() == Leader {
    self.logger.log(Debug, "\{self.id} ignoring MsgHup because already leader")
    return []
  }
  // Only a promotable voter may stand for election; a learner or a node no
  // longer in the configuration never campaigns (§4.2.1, promotable).
  if !self.promotable() {
    self.logger.log(Warning, "\{self.id} is unpromotable and can not campaign")
    return []
  }
  // Do not campaign while a committed configuration change is still unapplied:
  // electing a leader under a stale view of the membership could split the
  // quorum (etcd's `hup`/`hasUnappliedConfChanges`, raft.go:983).
  if self.has_unapplied_conf_changes() {
    self.logger.log(
      Warning,
      "\{self.id} cannot campaign at term \{self.core.current_term()} since there are still pending configuration changes to apply",
    )
    return []
  }
  self.logger.log(
    Info,
    "\{self.id} is starting a new election at term \{self.core.current_term()}",
  )
  self.reset_timeout()
  if self.pre_vote {
    self.start_pre_campaign()
  } else {
    self.start_real_campaign()
  }
}

///|
fn RaftNode::start_pre_campaign(self : RaftNode) -> Array[Message] {
  // Genuinely enter the PreCandidate state (etcd's becomePreCandidate) — the
  // core changes role without adopting the hypothetical term. `in_pre_campaign`
  // stays as the pre-vote bookkeeping flag, kept in step with the role.
  self.core.become_pre_candidate()
  self.logger.log(
    Info,
    "\{self.id} became pre-candidate at term \{self.core.current_term()}",
  )
  self.leader_id = None
  self.in_pre_campaign = true
  self.votes.clear()
  if !self.defer_responses {
    self.votes[self.id] = true
  }
  let out : Array[Message] = []
  let args : RequestVoteArgs = {
    term: self.core.current_term() + 1,
    candidate_id: self.id,
    last_log_index: self.core.last_log_index(),
    last_log_term: self.core.last_log_term(),
  }
  for p in self.vote_peers() {
    self.logger.log(
      Info,
      "\{self.id} [logterm: \{args.last_log_term}, index: \{args.last_log_index}] sent MsgPreVote request to \{p} at term \{self.core.current_term()}",
    )
    out.push(Message::new(self.id, p, PreVote(args)))
  }
  if self.defer_responses {
    // The candidate votes for itself, but the vote only counts once it is
    // durable: the acknowledgement waits for the advance boundary (etcd's
    // msgsAfterAppend), so a single-node cluster becomes a candidate now and a
    // leader then.
    self.deferred.push(
      Message::new(
        self.id,
        self.id,
        PreVoteResp({ term: args.term, vote_granted: true, }),
      ),
    )
  } else if self.config.has_majority(self.granted()) {
    // No durability boundary to wait for: a single-node cluster wins its own
    // pre-vote inside the call.
    return self.start_real_campaign()
  }
  out
}

///|
/// The peers a campaign solicits: voters only (a learner never votes), excluding
/// ourselves (the self-vote is a response, not a request), in ascending id order
/// as etcd's sorted `trk.Voters.IDs()` produces.
fn RaftNode::vote_peers(self : RaftNode) -> Array[String] {
  let out : Array[String] = []
  for v in self.config.voters() {
    if v != self.id {
      out.push(v)
    }
  }
  out.sort()
  out
}

///|
fn RaftNode::start_real_campaign(
  self : RaftNode,
  force~ : Bool = false,
) -> Array[Message] {
  self.in_pre_campaign = false
  self.core.become_candidate()
  self.logger.log(
    Info,
    "\{self.id} became candidate at term \{self.core.current_term()}",
  )
  self.leader_id = None
  self.votes.clear()
  if !self.defer_responses {
    self.votes[self.id] = true
  }
  let out : Array[Message] = []
  // A leadership-transfer campaign marks its votes forced so a follower still
  // leased to the outgoing leader grants them, letting the handover complete
  // instead of stalling out a full election timeout (§3.10).
  let args = self.core.request_vote_args()
  for p in self.vote_peers() {
    self.logger.log(
      Info,
      "\{self.id} [logterm: \{args.last_log_term}, index: \{args.last_log_index}] sent MsgVote request to \{p} at term \{self.core.current_term()}",
    )
    out.push(Message::new(self.id, p, Vote(args), force~))
  }
  if self.defer_responses {
    self.deferred.push(
      Message::new(
        self.id,
        self.id,
        VoteResp({ term: self.core.current_term(), vote_granted: true, }),
      ),
    )
  } else if self.config.has_majority(self.granted()) {
    // No durability boundary to wait for: winning with only our own vote means
    // there are no peers, so the campaign returns the (empty) append round.
    self.become_leader()
    return self.bcast_append()
  }
  out
}

///|
/// Acknowledge the leader's own freshly-appended entries. The leader sends no
/// MsgApp to itself, so it acknowledges the append the way a follower would —
/// with a self-addressed MsgAppResp — and that acknowledgement waits for the
/// entries to be durable (etcd's `appendEntry`). A no-op on a node that does not
/// track itself (i.e. not a leader): the response would find no progress.
fn RaftNode::self_ack(self : RaftNode) -> Unit {
  if !self.defer_responses {
    if self.progress.get(self.id) is Some(p) {
      p.maybe_update(self.core.last_log_index()) |> ignore
    }
    return
  }
  if self.progress.contains(self.id) {
    self.deferred.push(
      Message::new(
        self.id,
        self.id,
        AppendResp({
          term: self.core.current_term(),
          success: true,
          match_index: self.core.last_log_index(),
          conflict_index: 0UL,
          conflict_term: 0UL,
          reject_index: 0UL,
        }),
      ),
    )
  }
}

///|
/// Make the responses that depend on durability wait for the advance boundary
/// (etcd's `msgsAfterAppend`, stepped by `Advance`). `RawNode` turns this on for
/// the server it drives.
pub fn RaftNode::enable_deferred_responses(self : RaftNode) -> Unit {
  self.defer_responses = true
}

///|
/// Hand over the responses that are waiting for durability and clear the queue
/// (etcd steps `msgsAfterAppend` on `Advance`).
pub fn RaftNode::take_deferred(self : RaftNode) -> Array[Message] {
  let out = self.deferred.copy()
  self.deferred.clear()
  out
}

///|
/// The responses waiting for durability, as a copy: what an async append
/// directive carries to the storage thread.
pub fn RaftNode::deferred_responses(self : RaftNode) -> Array[Message] {
  self.deferred.copy()
}

///|
/// Whether a response is waiting to be stepped once the current Ready is
/// durable, so the driver does not report the node idle while it is not.
pub fn RaftNode::has_deferred(self : RaftNode) -> Bool {
  !self.deferred.is_empty()
}

///|
/// Take leadership: initialise per-follower progress, then append a no-op entry
/// under the new term. Committing that entry commits everything before it,
/// which is how a new leader safely learns its true commit index (Raft §5.4.2).
fn RaftNode::become_leader(self : RaftNode) -> Unit {
  self.core.become_leader()
  self.leader_id = Some(self.id)
  self.lead_transferee = None
  // etcd's reset() zeroes the uncommitted-tail accumulator on every transition;
  // the no-op appended below is empty (payload 0), so it stays 0 for a new leader.
  self.uncommitted_size = 0
  self.progress.clear()
  let last = self.core.last_log_index()
  // Conservatively assume any inherited tail may hold a pending conf change:
  // block new ones until the applied index passes it (etcd sets pendingConfIndex
  // to lastIndex here, before appending the no-op below). For a fresh leader with
  // an empty log this is 0, so the first conf change is allowed at once.
  self.pending_conf_index = last
  // etcd's reset() seeds every follower's progress at match 0 / next last+1, but
  // the leader's own entry at its last index, already streaming (becomeLeader
  // then calls pr.BecomeReplicate on it). The leader tracks itself in the same
  // progress map because it must self-ack its own appends — it sends no MsgApp to
  // itself — so `maybe_commit` counts the leader's stored log like any voter's.
  let self_pr = Progress::new(
    last + 1,
    max_inflight=self.max_inflight,
    max_inflight_bytes=self.max_inflight_bytes,
  )
  self_pr.match_index = last
  self_pr.become_replicate()
  self_pr.recent_active = true
  self.progress[self.id] = self_pr
  for p in self.peers {
    self.progress[p] = Progress::new(
      last + 1,
      max_inflight=self.max_inflight,
      max_inflight_bytes=self.max_inflight_bytes,
    )
  }
  let _ = self.core.append(self.core.current_term(), b"")
  // The no-op is acknowledged through the deferred self-response, so it commits
  // only once it is durable — not inside this transition.
  self.self_ack()
  // Like etcd's reset() on every state change, re-randomize the election
  // timeout and clear the timers.
  self.reset_timeout()
  if !self.defer_responses {
    // A direct caller has no advance boundary, so a single-node leader commits
    // its no-op here, which is also what arms reads.
    self.maybe_commit() |> ignore
  }
  // A leader inheriting a committed auto-leave joint config (whose LeaveJoint
  // was never committed, e.g. truncated on the term change) appends the leave.
  self.maybe_append_auto_leave()
  self.tracer.on_event(StateChange(self.id, Leader, self.core.current_term()))
  self.logger.log(
    Info,
    "\{self.id} became leader at term \{self.core.current_term()}",
  )
}

///|
/// Transition to follower in `term`, re-randomizing the election timeout. Every
/// state change routes through the timer reset (Raft §5.2), which is what makes
/// the per-node timeout fresh on each transition and keeps split votes rare.
pub fn RaftNode::become_follower(self : RaftNode, term : UInt64) -> Unit {
  self.core.become_follower(term)
  self.leader_id = None
  self.in_pre_campaign = false
  self.pending_conf_index = 0
  self.uncommitted_size = 0
  self.lead_transferee = None
  self.reset_timeout()
  self.tracer.on_event(StateChange(self.id, Follower, term))
  self.logger.log(Info, "\{self.id} became follower at term \{term}")
}

///|
/// Transition to candidate (advancing the term and voting for self), also
/// re-randomizing the election timeout.
pub fn RaftNode::become_candidate(self : RaftNode) -> Unit {
  self.core.become_candidate()
  self.leader_id = None
  self.in_pre_campaign = false
  self.pending_conf_index = 0
  self.uncommitted_size = 0
  self.lead_transferee = None
  self.reset_timeout()
  self.tracer.on_event(
    StateChange(self.id, Candidate, self.core.current_term()),
  )
  self.logger.log(
    Info,
    "\{self.id} became candidate at term \{self.core.current_term()}",
  )
}

///|
/// Turn off proposal forwarding (etcd's DisableProposalForwarding): a follower
/// will drop client proposals instead of forwarding them to the leader.
pub fn RaftNode::disable_proposal_forwarding(self : RaftNode) -> Unit {
  self.no_forward = true
}

///|
/// Append a client command and replicate it. On a leader the command is appended
/// and the resulting AppendEntries returned; on a follower it is forwarded to the
/// leader (unless forwarding is disabled or no leader is known); on a candidate
/// it is dropped. This routes through the same `Propose` message path a remote
/// proposal takes, so local and forwarded proposals behave identically.
pub fn RaftNode::propose(self : RaftNode, command : Bytes) -> Array[Message] {
  // The entry is built the way etcd's `Propose` builds it: a payload with no
  // term or index attached. Those are assigned when the *leader* appends it
  // (`appendEntry`); a follower forwards the message untouched, so the forwarded
  // MsgProp still shows the unset `0/0`.
  let out = self.step_propose(self.id, [Entry::normal(0UL, 0, command)])
  // A proposal can commit a pending configuration change, whose effect (a probe
  // to a server that just joined) is decided while folding it.
  self.drain_pending(out)
}

///|
/// Whether the most recent `propose` was refused (etcd's `Step` returning
/// `ErrProposalDropped`). A driver surfaces this so a dropped proposal is not
/// silently lost.
pub fn RaftNode::propose_dropped(self : RaftNode) -> Bool {
  self.last_propose_dropped
}

///|
/// Append a configuration change to the leader's log and replicate it, so every
/// server folds the same membership change into its configuration at the same
/// log position once it commits (Raft §6). A no-op on a non-leader.
pub fn RaftNode::propose_conf(
  self : RaftNode,
  change : ConfChange,
) -> Array[Message] {
  // Route through the same MsgProp path a client proposal takes, so a conf change
  // is counted against the uncommitted-log quota (etcd's appendEntry always
  // calls increaseUncommittedSize), validated against `pending_conf_index`, and
  // — on a follower — forwarded to the leader rather than silently dropped.
  let out = self.step_propose(self.id, [Entry::conf(0UL, 0, change.encode())])
  // Committing this change folds it in, which probes the servers it added.
  self.drain_pending(out)
}

///|
/// Whether a configuration change carrying `is_leave` must be refused at propose
/// time (etcd's stepLeader checks, raft.go:1326): a change is refused while one
/// is already pending (applied has not caught up to `pending_conf_index`), while
/// a non-leave change is proposed in a joint config, or while a leave is proposed
/// outside one. `DisableConfChangeValidation` turns the checks off.
fn RaftNode::conf_change_refusal(self : RaftNode, is_leave : Bool) -> String {
  if self.no_conf_change_validation {
    return ""
  }
  // etcd compares `pendingConfIndex` against the applied index: a further change
  // is refused while the pending one has not been applied yet (raft.go:1338).
  // Under a Ready loop the watermark is the one the application reports, which is
  // what lets the first change through once the leader's no-op is applied. A
  // caller that steps the node directly has no such watermark, so the last folded
  // change stands in for it.
  let applied = if self.defer_responses {
    self.core.last_applied
  } else {
    self.conf_applied
  }
  let already_pending = self.pending_conf_index > applied
  let already_joint = self.config.is_joint()
  // Upstream reports the first failing check in this order, verbatim
  // (raft.go:1330), so the diagnostic names the real reason.
  if already_pending {
    "possible unapplied conf change at index \{self.pending_conf_index} (applied to \{applied})"
  } else if already_joint && !is_leave {
    "must transition out of joint config first"
  } else if !already_joint && is_leave {
    "not in joint state; refusing empty conf change"
  } else {
    ""
  }
}

///|
/// Append a batch configuration change (joint consensus, Raft §4.3) and
/// replicate it. Entering joint with a non-empty batch moves the cluster to
/// C(old,new); once committed, `auto_leave` has the leader append the matching
/// leave automatically. A no-op on a non-leader or mid-transfer.
pub fn RaftNode::propose_conf_v2(
  self : RaftNode,
  change : ConfChangeV2,
) -> Array[Message] {
  self.step_propose(self.id, [Entry::conf(0UL, 0, change.encode())])
}

///|
/// Begin transferring leadership to `target` (Raft §3.10). If the target is
/// already caught up it is sent a TimeoutNow so it campaigns immediately;
/// otherwise it is first sent the entries it lacks and the caller retries the
/// transfer once it has caught up. A no-op on a non-leader or an unknown target.
pub fn RaftNode::transfer_leadership(
  self : RaftNode,
  target : String,
) -> Array[Message] {
  if self.core.role() != Leader {
    return []
  }
  // A learner can never be leader, so a transfer to one is ignored outright,
  // without disturbing any transfer already under way (etcd checks pr.IsLearner
  // first).
  if self.config.is_learner(target) {
    self.logger.log(
      Debug,
      "\{self.id} is learner. Ignored transferring leadership",
    )
    return []
  }
  // A transfer already in progress (§3.10): a repeat to the same target is
  // ignored (the handoff continues); a request naming a different target aborts
  // the old one before the new is considered (etcd's stepLeader).
  if self.lead_transferee is Some(last) {
    if last == target {
      self.logger.log(
        Info,
        "\{self.id} [term \{self.core.current_term()}] transfer leadership to \{target} is in progress, ignores request to same node \{target}",
      )
      return []
    }
    self.lead_transferee = None
    self.logger.log(
      Info,
      "\{self.id} [term \{self.core.current_term()}] abort previous transferring leadership to \{last}",
    )
  }
  // Transferring to ourselves is a no-op — but note it still aborts a pending
  // transfer above, which is how a leader cancels a handoff (etcd's
  // "already leader" branch runs after abortLeaderTransfer).
  if target == self.id {
    self.logger.log(
      Debug,
      "\{self.id} is already leader. Ignored transferring leadership to self",
    )
    return []
  }
  // An unknown / non-member target is a no-op.
  if !self.config.contains(target) {
    return []
  }
  self.logger.log(
    Info,
    "\{self.id} [term \{self.core.current_term()}] starts to transfer leadership to \{target}",
  )
  self.lead_transferee = Some(target)
  self.election_elapsed = 0
  // A voting member on a leader always has a progress entry (reconcile_peers
  // keeps the two in step), so index it directly.
  let p = self.progress[target]
  if p.match_index == self.core.last_log_index() {
    self.logger.log(
      Info,
      "\{self.id} sends MsgTimeoutNow to \{target} immediately as \{target} already has up-to-date log",
    )
    [Message::new(self.id, target, TimeoutNow(self.core.current_term()))]
  } else if self.send_to(target) is Some(m) {
    [m]
  } else {
    // The target is behind the compacted snapshot and has not been heard from,
    // so `send_to` withholds even the snapshot (D-snap-recentactive). Nothing is
    // sent; the transfer waits for the target to become active.
    []
  }
}

///|
/// Request that leadership move to `target` (etcd's TransferLeader). On a leader
/// this begins the handoff; on a follower the request is forwarded to the leader,
/// so a transfer may be initiated from any server. Routes through the same
/// `TransferLeader` message path a forwarded request takes.
pub fn RaftNode::request_transfer_leader(
  self : RaftNode,
  target : String,
) -> Array[Message] {
  self.step_transfer_leader(self.id, target)
}

///|
/// Forget the currently-recognised leader (etcd's MsgForgetLeader): a follower
/// clears its leader so it may grant (pre)votes at once instead of waiting out a
/// check-quorum lease, without moving its term or resetting its election timer.
/// Ignored under lease-based reads; a no-op on a candidate or leader.
pub fn RaftNode::forget_leader(self : RaftNode) -> Unit {
  self.step(Message::new(self.id, self.id, ForgetLeader)) |> ignore
}

///|
/// Build the AppendEntries or InstallSnapshot message for one follower, based on
/// its progress. If the entry the follower needs has been compacted away, a
/// snapshot is sent instead (Raft §7). Returns `None` when nothing should be
/// sent — currently only when a snapshot would be needed but the follower has
/// not been heard from this liveness window (etcd's `maybeSendSnapshot` refuses
/// to ship a snapshot to a `!RecentActive` follower, since it is probably gone
/// and the snapshot would be wasted bandwidth).
fn RaftNode::send_to(self : RaftNode, peer : String) -> Message? {
  self.send_to_if(peer, true)
}

///|
/// The workhorse behind `send_to`: `send_if_empty` mirrors etcd's
/// `maybeSendAppend(to, sendIfEmpty)`. A batch pump passes `false` so it never
/// emits a content-free MsgApp on top of the commit bump it already sent.
fn RaftNode::send_to_if(
  self : RaftNode,
  peer : String,
  send_if_empty : Bool,
) -> Message? {
  // A leader never sends an append to itself: it acknowledges its own entries
  // through `self_ack`. Reaching here with our own id means a response travelled
  // the self-addressed path, and sending would produce a message no transport can
  // deliver (etcd panics rather than send it).
  if peer == self.id {
    return None
  }
  let p = self.progress[peer]
  // A snapshot is already in flight: etcd's `IsPaused` is always true for
  // StateSnapshot, so no append (and no second snapshot) is sent until its ack
  // reconnects the follower.
  if p.state == Snapshot {
    return None
  }
  let prev = p.next_index - 1
  if prev < self.core.snapshot_index {
    if !p.is_active() {
      self.logger.log(
        Debug,
        "ignore sending snapshot to \{peer} since it is not recently active",
      )
      return None
    }
    // The snapshot that ships is the state the application has applied, which is
    // at or ahead of the compaction point: a server that has applied past its
    // last compaction still holds a snapshot of that state (etcd keeps the two
    // apart — `raftLog.snapshot()` returns the storage's, whose index is the
    // applied one, while `firstIndex()` follows the compaction).
    let snap_index = if self.core.last_applied > self.core.snapshot_index {
      self.core.last_applied
    } else {
      self.core.snapshot_index
    }
    let snap_term = self.core.term_at(snap_index)
    // Report the snapshot before the progress moves into `StateSnapshot`, then
    // report the pause it causes (etcd logs both, in that order).
    self.logger.log(
      Debug,
      "\{self.id} [firstindex: \{self.core.snapshot_index + 1}, commit: \{self.core.commit_index}] sent snapshot[index: \{snap_index}, term: \{snap_term}] to \{peer} [\{progress_text(p)}]",
    )
    p.become_snapshot(snap_index)
    self.logger.log(
      Debug,
      "\{self.id} paused sending replication messages to \{peer} [\{progress_text(p)}]",
    )
    let snap : Snapshot = {
      last_index: snap_index,
      last_term: snap_term,
      data: b"",
      // Stamp the current membership so the follower rebuilds it on install.
      conf_state: self.conf_state(),
    }
    return Some(
      Message::new(
        self.id,
        peer,
        Snapshot(
          InstallSnapshotArgs::whole(self.core.current_term(), self.id, snap),
        ),
      ),
    )
  }
  // A streaming follower whose in-flight window is full is sent a content-free
  // probe (no new entries, no window slot consumed), rather than more entries it
  // has no room to track — matching etcd's ShouldSendMsgApp (§5.3 flow control).
  let entries = if p.state == Replicate && p.inflights.full() {
    []
  } else {
    // Cap the batch by encoded byte size alone (etcd's MaxSizePerMsg), matching
    // `sendAppend`'s `entries(pr.Next, r.maxMsgSize)` (raft.go:640); limit_size
    // always keeps at least one entry.
    limit_size(self.core.entries_after(prev)[:], self.max_msg_bytes)
  }
  let has = !entries.is_empty()
  // etcd's `maybeSendAppend`: an empty batch is dropped unless the caller asked
  // for a content-free message to carry the commit index.
  if !has && !send_if_empty {
    return None
  }
  let last = if has { entries[entries.length() - 1].index } else { prev }
  // Total payload bytes in this batch, for the in-flight byte budget
  // (etcd's payloadsSize). Inert unless MaxInflightBytes is set.
  let mut ents_size = 0UL
  for e in entries {
    ents_size = ents_size + payload_size(e)
  }
  // Advance the flow controller: stream (consume a window slot) or pause a probe.
  p.sent_entries(last, has, bytes=ents_size)
  // Record the commit index just put in flight to this follower, so a later
  // eager commit-only send can be skipped when it would tell the follower
  // nothing new (etcd's SentCommit / CanBumpCommit optimization).
  p.note_commit_sent(self.core.commit_index)
  Some(
    Message::new(
      self.id,
      peer,
      Append({
        term: self.core.current_term(),
        leader_id: self.id,
        prev_log_index: prev,
        prev_log_term: self.core.term_at(prev),
        entries,
        leader_commit: self.core.commit_index,
      }),
    ),
  )
}

///|
/// A follower's progress in etcd's `Progress.String` shape:
/// `StateReplicate match=11 next=15 [learner] [paused] [pendingSnap=n]
/// [inactive] [inflight=n[full]]`.
fn progress_text(p : Progress) -> String {
  let buf = StringBuilder()
  let state = match p.state {
    Probe => "StateProbe"
    Replicate => "StateReplicate"
    Snapshot => "StateSnapshot"
  }
  buf.write_string("\{state} match=\{p.match_index} next=\{p.next_index}")
  if p.is_learner {
    buf.write_string(" learner")
  }
  // `is_paused` follows etcd's `MsgAppFlowPaused`; also surface a full in-flight
  // window, which reads as paused to an operator even before the next send.
  if p.is_paused() || p.inflights.full() {
    buf.write_string(" paused")
  }
  if p.pending_snapshot > 0UL {
    buf.write_string(" pendingSnap=\{p.pending_snapshot}")
  }
  if !p.is_active() {
    buf.write_string(" inactive")
  }
  let n = p.inflights.count()
  if n > 0 {
    buf.write_string(" inflight=\{n}")
    if p.inflights.full() {
      buf.write_string("[full]")
    }
  }
  buf.to_string()
}

///|
/// Send an AppendEntries (or snapshot) to every follower.
fn RaftNode::bcast_append(self : RaftNode) -> Array[Message] {
  let out : Array[Message] = []
  for peer in self.peers {
    // Skip a throttled follower: a streaming one whose in-flight window is full,
    // or one still digesting a snapshot. A prober is sent its one probe.
    let p = self.progress[peer]
    if p.is_paused() {
      continue
    }
    if self.send_to(peer) is Some(m) {
      out.push(m)
    }
  }
  // Flush any final appends owed to servers just reconciled out of the peer set
  // (a committed removal), so the departing server hears the removing commit.
  for m in self.pending_farewell {
    out.push(m)
  }
  self.pending_farewell.clear()
  out
}

///|
/// Send a heartbeat to every follower. The `commit` carried is capped at what
/// the follower is known to hold, so it never learns of a commit index past its
/// own log (Raft §5.3). Every heartbeat carries the read-confirmation position
/// (etcd's `bcastHeartbeat` → `heartbeatCtx`); it is empty when no read is
/// pending, and a follower echoes it back so a quorum confirms pending reads.
fn RaftNode::bcast_heartbeat(self : RaftNode) -> Array[Message] {
  let context = self.read_only.heartbeat_ctx()
  let out : Array[Message] = []
  for peer in self.peers {
    let p = self.progress[peer]
    let commit = if p.match_index < self.core.commit_index {
      p.match_index
    } else {
      self.core.commit_index
    }
    // A heartbeat carries a commit index too, so it also advances what the
    // follower is known to have been told (etcd's sendHeartbeat SentCommit).
    p.note_commit_sent(commit)
    out.push(
      Message::new(
        self.id,
        peer,
        Heartbeat({
          term: self.core.current_term(),
          leader_id: self.id,
          commit,
          context,
        }),
      ),
    )
  }
  out
}

///|
/// Recompute the commit index from replication progress. An index is committed
/// once it is stored on a majority of voters *and* belongs to the current term;
/// a leader never commits an entry from an earlier term by counting replicas
/// alone (Raft §5.4.2). Self counts as having stored up to its last index.
fn RaftNode::maybe_commit(self : RaftNode) -> Bool {
  let mut any = false
  // Committing a configuration change can shrink the quorum, which may in turn
  // let a further entry commit, so re-evaluate until the commit index is stable.
  // A leader that removed itself steps down inside apply, ending the loop.
  while self.core.role() == Leader {
    // The highest index a quorum of the configuration has stored (the leader
    // itself included, at its last index). In a joint configuration this is the
    // smaller of the two halves' agreed indices.
    let acked : Map[String, UInt64] = Map([])
    for peer, p in self.progress {
      acked[peer] = p.match_index
    }
    // The leader's own vote in the quorum is what it has made durable, which is
    // exactly what its progress records: the self-addressed MsgAppResp that
    // advances it is stepped on Advance, so an entry that is only in the unstable
    // tail does not count toward the commit index yet (etcd's tracker does the
    // same for `Progress[r.id]`).
    let mci = self.config.committed_index(acked)
    // §5.4.2: a leader only commits an index whose entry is from its own term;
    // since terms are monotonic in the log, no lower current-term index can be
    // committed if this one is not.
    if mci > self.core.commit_index &&
      self.core.term_at(mci) == self.core.current_term() {
      self.core.advance_commit(mci)
      self.tracer.on_event(Commit(self.id, mci))
      self.apply_committed_conf()
      any = true
    } else {
      break
    }
  }
  any
}

///|
/// Account for `size` bytes of newly-proposed payload against the uncommitted
/// quota (etcd's increaseUncommittedSize). Returns false — meaning the proposal
/// must be dropped — only when the tail is already non-empty and this would push
/// it over `max_uncommitted_size`; an empty tail always accepts at least one
/// proposal, however large, and empty (zero-byte) entries always pass.
fn RaftNode::increase_uncommitted_size(self : RaftNode, size : UInt64) -> Bool {
  if self.max_uncommitted_size > 0 &&
    self.uncommitted_size > 0 &&
    size > 0 &&
    self.uncommitted_size + size > self.max_uncommitted_size {
    return false
  }
  self.uncommitted_size = self.uncommitted_size + size
  true
}

///|
/// Release `size` bytes from the uncommitted tally as entries commit or apply
/// (etcd's reduceUncommittedSize). Never goes below zero.
pub fn RaftNode::reduce_uncommitted_size(
  self : RaftNode,
  size : UInt64,
) -> Unit {
  self.uncommitted_size = if size > self.uncommitted_size {
    0
  } else {
    self.uncommitted_size - size
  }
}

///|
/// The current size of the uncommitted log tail, in payload bytes.
pub fn RaftNode::uncommitted_size(self : RaftNode) -> UInt64 {
  self.uncommitted_size
}

///|
/// Advance the applied watermark to `index`, releasing the uncommitted-tail
/// bytes of the entries now applied (etcd releases the quota at *apply* time,
/// on `MsgStorageApplyResp`, not at commit time). Deferring the release to apply
/// is what bounds the un-applied tail even for a single-voter leader — which
/// commits its own appends instantly, so a commit-time release could never form
/// a bounded backlog. The application (or the RawNode/Advance layer) calls this
/// once it has applied committed entries.
pub fn RaftNode::advance_applied(self : RaftNode, index : UInt64) -> Unit {
  if index <= self.core.last_applied {
    return
  }
  let mut freed = 0UL
  let mut i = self.core.last_applied + 1
  while i <= index {
    if self.core.entry_at(i) is Some(e) {
      freed = freed + payload_size(e)
    }
    i = i + 1
  }
  self.core.mark_applied(index)
  self.reduce_uncommitted_size(freed)
  // Applying the entries is also when a committed configuration change takes
  // effect: the application has now seen the entry, so every server switches at
  // the same point in its own apply loop (etcd's `ApplyConfChange`, which the
  // driver calls once it has applied the entry).
  self.apply_committed_conf()
}

///|
/// Fold every committed-but-unapplied `ConfChange` entry into the live
/// configuration, replication progress and peer set, in log order (Raft §6).
/// This is what actually makes an added server a voter, stops a removed server
/// counting toward quorum, and steps down a leader that removed itself.
fn RaftNode::apply_committed_conf(self : RaftNode) -> Unit {
  if self.applying_conf {
    return
  }
  self.applying_conf = true
  // How far the configuration may have advanced. A driver tells us what it has
  // applied (`last_applied`), and a change takes effect there; a caller without a
  // Ready loop has no such watermark, so commitment stands in for it.
  let upto = if self.defer_responses {
    self.core.last_applied
  } else {
    self.core.commit_index
  }
  while self.conf_applied < upto {
    let idx = self.conf_applied + 1
    if self.core.entry_at(idx) is Some(e) && e.is_conf_change() {
      // A batch (ConfChangeV2) is tagged 'V'; anything else is a single change.
      match ConfChangeV2::decode(e.command) {
        Some(v2) => self.apply_conf_change_v2(v2)
        None =>
          if ConfChange::decode(e.command) is Some(cc) {
            cc.apply_to(self.config)
          }
      }
      // etcd's `switchToConfig` announces the new configuration before acting on
      // it — stepping down a removed leader, probing new peers — so the
      // "switched" line precedes any removal-driven "became follower".
      self.logger.log(
        Info,
        "\{self.id} switched to configuration \{self.config.describe()}",
      )
      self.reconcile_peers()
      self.report_after_config()
    }
    self.conf_applied = idx
  }
  // Auto-leave: a leader in an auto-leave joint config appends the matching
  // LeaveJoint (§4.3). This is driven by the *durable* config flag, not a
  // one-shot marker, and re-checked whenever the committed prefix advances, so
  // if the LeaveJoint is truncated before committing the leader appends it
  // again — the flag is only cleared when the LeaveJoint itself commits.
  self.maybe_append_auto_leave()
  self.applying_conf = false
}

///|
/// What a leader does right after a change takes effect (etcd's
/// `switchToConfig`): if the smaller quorum let more entries commit, broadcast
/// that; otherwise probe the peers, including the one that just joined, so it
/// does not have to wait out a heartbeat interval or the next proposal.
fn RaftNode::report_after_config(self : RaftNode) -> Unit {
  if self.core.role() != Leader || self.config.voters().is_empty() {
    return
  }
  if self.maybe_commit() {
    for m in self.bcast_append() {
      self.pending.push(m)
    }
  } else {
    for peer in self.peers.copy() {
      // `maybeSendAppend(id, sendIfEmpty=false)` bails out on a paused follower
      // (full in-flight window or pending snapshot), just like bcastAppend.
      if self.progress[peer].is_paused() {
        continue
      }
      // A follower with nothing to catch up on is left alone, while an append
      // that carries entries — or the snapshot a far-behind follower needs — goes
      // out now.
      let msg = self.send_to(peer)
      if msg is Some(m) && m.kind() != "Probe" {
        self.pending.push(m)
      }
    }
  }
}

///|
/// Append the messages a configuration fold produced to `out`, so an entry point
/// that folded a change returns them with the rest of its output.
fn RaftNode::drain_pending(
  self : RaftNode,
  out : Array[Message],
) -> Array[Message] {
  for m in self.take_pending() {
    out.push(m)
  }
  out
}

///|
/// Hand over the messages a configuration fold produced, and clear the buffer.
pub fn RaftNode::take_pending(self : RaftNode) -> Array[Message] {
  let out = self.pending.copy()
  self.pending.clear()
  out
}

///|
/// If we are a leader in an auto-leave joint config and no LeaveJoint entry is
/// already in flight, append one. Robust to truncation: `has_pending_leave`
/// scans the uncommitted tail, so a truncated LeaveJoint is simply re-appended.
fn RaftNode::maybe_append_auto_leave(self : RaftNode) -> Unit {
  // A LeaveJoint is already staged if one sits at or past the applied watermark
  // (`has_pending_leave` scans from there). That covers a leave that has been
  // committed but not yet applied — which would otherwise look absent and be
  // appended a second time — while a leave lost to a truncation is, correctly,
  // no longer found and gets re-appended.
  if self.core.role() == Leader &&
    self.config.is_joint() &&
    self.config.auto_leave &&
    !self.has_pending_leave() {
    // The leader announces the transition it is starting, so a reader can see
    // which entry the leave came from (etcd's `maybeAutoLeave`).
    self.logger.log(
      Info,
      "initiating automatic transition out of joint configuration \{self.config.describe()}",
    )
    let _ = self.core.append_conf(
      self.core.current_term(),
      ConfChangeV2::leave_joint().encode(),
    )
    // Self-ack the auto-appended leave, as with any leader self-append.
    self.self_ack()
    // Upstream proposes the leave by stepping a MsgProp, which appends *and*
    // broadcasts in one go, so the peers see the leave in the same turn the
    // leader logs that it is starting the transition.
    for m in self.bcast_append() {
      self.pending.push(m)
    }
    // The auto-appended leave is itself now the pending change, so no other
    // conf change is proposed until it applies (etcd tracks pendingConfIndex for
    // the automatic transition too).
    self.pending_conf_index = self.core.last_log_index()
  }
}

///|
/// Whether any committed-but-unapplied entry is a configuration change (etcd's
/// `hasUnappliedConfChanges`). A server refuses to campaign while one is pending,
/// so it does not become leader under a membership it has not yet folded in.
fn RaftNode::has_unapplied_conf_changes(self : RaftNode) -> Bool {
  if self.core.last_applied >= self.core.commit_index {
    return false
  }
  let mut i = self.core.last_applied + 1
  while i <= self.core.commit_index {
    if self.core.entry_at(i) is Some(e) && e.is_conf_change() {
      return true
    }
    i = i + 1
  }
  false
}

///|
/// Whether a LeaveJoint conf-change entry is already present in the uncommitted
/// tail (so we should not append another).
fn RaftNode::has_pending_leave(self : RaftNode) -> Bool {
  // Start at the applied watermark, not the commit index: a LeaveJoint that is
  // committed but not yet applied is still "pending" and must not be re-appended.
  let mut i = self.conf_applied + 1
  while i <= self.core.last_log_index() {
    if self.core.entry_at(i) is Some(e) &&
      e.is_conf_change() &&
      ConfChangeV2::decode(e.command) is Some(v2) &&
      v2.is_leave() {
      return true
    }
    i = i + 1
  }
  false
}

///|
/// The current membership as a `ConfState`, to stamp into a snapshot so a
/// follower restoring from it rebuilds the same voter/learner sets (§7).
pub fn RaftNode::conf_state(self : RaftNode) -> ConfState {
  {
    voters: self.config.members.copy(),
    learners: self.config.learners.copy(),
    voters_outgoing: if self.config.joint {
      self.config.outgoing.copy()
    } else {
      []
    },
    learners_next: self.config.learners_next.copy(),
    auto_leave: self.config.auto_leave,
  }
}

///|
/// A read-only view of one follower's replication progress, surfaced for status
/// and monitoring (etcd's `tracker.Progress` as exposed through `Status` and
/// `WithProgress`). It is a copied value, so reading it never disturbs the
/// running protocol.
#derive(Eq)
pub(all) struct ProgressStatus {
  id : String
  match_index : UInt64
  next_index : UInt64
  state : ProgressState
  paused : Bool
  pending_snapshot : UInt64
  is_learner : Bool
  // Whether the follower has been heard from within the last election timeout
  // (etcd's `RecentActive`); an inactive follower is not sent entries.
  recent_active : Bool
  // Outstanding AppendEntries not yet acknowledged, and whether that window is
  // full (etcd's `Inflights.Count()` / `Full()`).
  inflight : Int32
  inflight_full : Bool
}

///|
fn progress_status_of(id : String, p : Progress) -> ProgressStatus {
  {
    id,
    match_index: p.match_index,
    next_index: p.next_index,
    state: p.state,
    paused: p.is_paused(),
    pending_snapshot: p.pending_snapshot,
    is_learner: p.is_learner,
    recent_active: p.is_active(),
    inflight: p.inflights.count(),
    inflight_full: p.inflights.full(),
  }
}

///|
/// The full status of the server (etcd's `Status`): the basic status, a snapshot
/// of every tracked follower's replication progress — including the leader's own
/// entry, fully caught up — and the current configuration as a `ConfState`.
pub(all) struct FullStatus {
  basic : RaftStatus
  progress : Array[ProgressStatus]
  config : ConfState
}

///|
/// A snapshot of every tracked follower's progress. On a leader this includes
/// the leader's own entry (caught up to its last index); a non-leader tracks no
/// progress and returns an empty view.
pub fn RaftNode::progress_status(self : RaftNode) -> Array[ProgressStatus] {
  let out : Array[ProgressStatus] = []
  // Every tracked peer, the leader itself included: its own progress advances
  // only when the self-addressed acknowledgement becomes durable, so reporting
  // it from the log tail would claim entries the leader has not persisted (etcd
  // reports `trk.Progress` verbatim).
  for id, p in self.progress {
    out.push(progress_status_of(id, p))
  }
  out
}

///|
/// One member's progress view, or `None` if it is not tracked. A leader also
/// answers for itself (etcd tracks the leader in its own progress map).
pub fn RaftNode::progress_of(self : RaftNode, id : String) -> ProgressStatus? {
  if id == self.id && self.core.role() == Leader {
    return Some({
      id,
      match_index: self.core.last_log_index(),
      next_index: self.core.last_log_index() + 1,
      state: Replicate,
      paused: false,
      pending_snapshot: 0,
      is_learner: false,
      recent_active: true,
      inflight: 0,
      inflight_full: false,
    })
  }
  self.progress.get(id).map(p => progress_status_of(id, p))
}

///|
/// Visit every tracked follower's progress (etcd's `WithProgress`). The visitor
/// must not retain the `Progress` beyond the call; it is handed a copied view.
pub fn RaftNode::with_progress(
  self : RaftNode,
  visit : (String, ProgressStatus) -> Unit,
) -> Unit {
  for entry in self.progress_status() {
    visit(entry.id, entry)
  }
}

///|
/// The full status of this server (etcd's `Status`): basic status + per-follower
/// progress + the current configuration.
pub fn RaftNode::full_status(self : RaftNode) -> FullStatus {
  {
    basic: self.status(),
    progress: self.progress_status(),
    config: self.conf_state(),
  }
}

///|
/// Rebuild the live configuration from a snapshot's `ConfState` — the voters,
/// the outgoing half (so the cluster resumes mid-joint if the snapshot was taken
/// then), and the learners — then reconcile peers/progress. This is what stops a
/// snapshot restore from silently losing the membership.
fn RaftNode::restore_conf_state(self : RaftNode, cs : ConfState) -> Unit {
  self.config.members.clear()
  for v in cs.voters {
    self.config.members.push(v)
  }
  self.config.outgoing.clear()
  for v in cs.voters_outgoing {
    self.config.outgoing.push(v)
  }
  self.config.joint = !cs.voters_outgoing.is_empty()
  // The auto-leave intent is durable config state carried in the snapshot, so a
  // follower that installs it keeps the leader's pending transition.
  self.config.auto_leave = cs.auto_leave
  self.config.learners.clear()
  for l in cs.learners {
    self.config.learners.push(l)
  }
  self.config.learners_next.clear()
  for l in cs.learners_next {
    self.config.learners_next.push(l)
  }
  self.reconcile_peers()
}

///|
/// Apply one committed batch configuration change (joint enter/leave).
fn RaftNode::apply_conf_change_v2(self : RaftNode, v2 : ConfChangeV2) -> Unit {
  if v2.is_leave() {
    self.config.leave_joint()
  } else {
    let (auto, joint) = v2.enters_joint()
    if joint {
      // Enter joint consensus: snapshot the current voters as the outgoing half
      // *before* applying the batch, so a voter demoted here is seen to still be
      // an outgoing voter and is staged in learners_next rather than losing its
      // vote mid-transition. Decisions then need a majority of *both* halves
      // until the matching leave commits (§4.3).
      self.config.begin_joint(self.config.members.copy())
      for c in v2.changes {
        c.apply_to(self.config)
      }
      // Record auto-leave as *durable* config state (from this committed entry),
      // not a transient flag, so a truncated LeaveJoint cannot lose the intent.
      self.config.auto_leave = auto
    } else {
      // Applied simply (an `Auto` transition changing at most one voter): no
      // joint transition, matching etcd's ConfChangeV2 `Auto` path.
      for c in v2.changes {
        c.apply_to(self.config)
      }
    }
  }
}

///|
/// Bring the peer set and replication progress into line with the current
/// configuration after a change: add newcomers (with progress, if leading),
/// drop departed servers, abort a transfer to a removed target, and step down if
/// we ourselves are no longer part of the configuration (Raft §6).
fn RaftNode::reconcile_peers(self : RaftNode) -> Unit {
  let desired = self.config.nodes()
  self.peers.retain(fn(p) { desired.contains(p) })
  let stale : Array[String] = []
  for peer, _ in self.progress {
    if !desired.contains(peer) {
      stale.push(peer)
    }
  }
  for peer in stale {
    // A leader tells a departing server about the commit that removes it before
    // discarding its progress, so it does not stall on the pre-removal commit
    // index. Withheld only when `send_to` itself withholds (an inactive peer
    // behind the snapshot), matching etcd, which would not ship it a snapshot.
    if self.core.role() == Leader {
      if self.send_to(peer) is Some(m) {
        self.pending_farewell.push(m)
      }
    }
    self.progress.remove(peer)
  }
  for id in desired {
    if id != self.id {
      if !self.peers.contains(id) {
        self.peers.push(id)
      }
      if self.core.role() == Leader && !self.progress.contains(id) {
        // A server that just joined starts *optimistically* at the last index,
        // not past it: the leader assumes it has everything but the final entry
        // and probes with that one, so an empty follower rejects and is caught up
        // by the ordinary (snapshot) recovery path (etcd's `Changer`, which uses
        // `LastIndex` here).
        let p = Progress::new(
          self.core.last_log_index(),
          max_inflight=self.max_inflight,
          max_inflight_bytes=self.max_inflight_bytes,
        )
        // etcd's initProgress marks a newcomer recently active, so a check-quorum
        // leader that adds a voter is not immediately stepped down by the very
        // next quorum check for a member it has not yet had a chance to hear from.
        p.recent_active = true
        self.progress[id] = p
      }
      // Keep each progress's learner flag in step with the configuration.
      if self.progress.get(id) is Some(p) {
        p.is_learner = self.config.is_learner(id)
      }
    }
  }
  // Upstream iterates voters in ascending id order (`ProgressMap.Visit` sorts
  // before visiting), which fixes the order of broadcasts and vote requests.
  // Keeping `peers` sorted makes every loop over it reproduce that order.
  self.peers.sort()
  if self.lead_transferee is Some(t) && !self.config.contains(t) {
    self.lead_transferee = None
  }
  // etcd's `switchToConfig`: a leader that is removed (or demoted to a learner)
  // steps down only when `StepDownOnRemoval` is set. By default it stays leader
  // over a configuration it is no longer part of, until the survivors elect a
  // replacement — matching etcd, whose zero-valued Config does not step down.
  if self.core.role() == Leader &&
    self.step_down_on_removal &&
    (!self.config.contains(self.id) || self.config.is_learner(self.id)) {
    self.step_down()
  }
}

///|
pub extend ProgressStatus with Eq::{not_equal, equal}