// 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.

///|
/// This server's soft state — its known leader and current role (etcd's
/// `raft.softState`). Reading it never disturbs the protocol.
pub fn RaftNode::soft_state(self : RaftNode) -> SoftState {
  { lead: self.leader(), state: self.role(), }
}

///|
/// The goroutine-free driver of a Raft server (etcd's `RawNode`). The core
/// `RaftNode` produces its output — messages to send, entries appended,
/// commits — as return values of `tick`/`step`/`propose`; `RawNode` batches
/// that output into the synchronous Ready/Advance cycle a real deployment runs:
/// take a `Ready`, persist its entries and hard state, send its messages, apply
/// its committed entries, then call `advance`.
///
/// It needs no language-level `async`: a `Ready` is a plain value and `advance`
/// a plain call. The unstable-vs-stable split that separates entries still to be
/// written (`entries`) from committed entries ready to apply
/// (`committed_entries`) is kept by a real `RaftLog`: its `storage` holds what
/// the application has persisted, its `unstable` tail holds what has not.
pub struct RawNode {
  raft : RaftNode
  // Mirror of the core node's durable state, kept current by `sync_log`. Rebuilt
  // when the core installs a snapshot (see `reflect_snapshot`), so it is mutable.
  mut log : RaftLog
  // The concrete store `log` reads through. Kept alongside so an operation only
  // the backend understands — a local compaction — can still be driven (etcd's
  // rafttest holds the `MemoryStorage` itself and calls `Compact` on it).
  mut storage : MemoryStorage
  // The per-Ready committed-entry byte budget, retained so the log can be rebuilt
  // with the same pagination after a snapshot install.
  max_committed : UInt64
  msgs : Array[Message]
  // Requests and responses that must be stepped only once the Ready that
  // produced them has been handled (etcd's `stepsOnAdvance`, in older versions
  // `msgsAfterAppend`): a follower's append acknowledgement is held until its
  // entries are durable, which is what puts the ack in the Ready *after* the
  // write. Captured by `accept_ready`, drained by `advance`.
  steps_on_advance : Array[Message]
  read_states : Array[ReadState]
  mut prev_soft : SoftState
  mut prev_hard : HardState
  // When set (etcd's AsyncStorageWrites), a Ready hands its entries and committed
  // entries to local storage threads as directives rather than expecting the
  // caller to persist/apply them inline before `advance`.
  mut async_storage : Bool
  // The snapshot baseline already reflected into `log`, and how many snapshots the
  // core has *installed from a leader*. A local compaction also advances the
  // core's `snapshot_index` (it moves the durable floor), but there is no snapshot
  // to hand the application — only an install produces `Ready.snapshot`.
  mut reflected_snap : UInt64
  mut seen_snap_installs : UInt64
}

///|
/// Build a driver over `raft` (etcd's `NewRawNode`). The previous soft and hard
/// states are seeded from the server as it stands, so the first `Ready` reports
/// a change only if one genuinely happens afterwards. The `RaftLog` is recovered
/// from the node's durable state: whatever is already in the node's log (and its
/// snapshot baseline) is loaded into stable storage, so only entries appended
/// afterwards land in the unstable tail and need writing.
pub fn RawNode::new(raft : RaftNode) -> RawNode {
  RawNode::new_sized(raft, no_limit)
}

///|
/// Build a driver that paginates each Ready's committed entries to at most
/// `max_committed_size` bytes (etcd's `MaxCommittedSizePerReady`). A burst of
/// commits is then handed to the application over several `Ready`s instead of
/// one unbounded batch.
pub fn RawNode::new_sized(
  raft : RaftNode,
  max_committed_size : UInt64,
) -> RawNode {
  // The driver knows when a Ready became durable, so responses that depend on
  // durability wait for `advance` rather than taking effect inside the call that
  // produced them.
  raft.enable_deferred_responses()
  let node = raft.node()
  let storage = MemoryStorage::new()
  if node.snapshot_index > 0 {
    // Seed the fresh storage's baseline directly: `apply_snapshot`'s out-of-date
    // guard is meaningless against an empty baseline, and its raising signature
    // would force a vacuous catch here.
    storage.seed_snapshot({
      last_index: node.snapshot_index,
      last_term: node.snapshot_term,
      data: b"",
      conf_state: ConfState::empty(),
    })
  }
  storage.append(node.entries_after(node.snapshot_index))
  let log = RaftLog::new_with_size(storage, max_committed_size)
  log.seed(node.commit_index, node.last_applied)
  {
    raft,
    log,
    storage,
    max_committed: max_committed_size,
    msgs: [],
    steps_on_advance: [],
    read_states: [],
    prev_soft: raft.soft_state(),
    prev_hard: node.hard_state(),
    async_storage: false,
    reflected_snap: node.snapshot_index,
    seen_snap_installs: node.snapshots_installed,
  }
}

///|
/// View this server through the Ready/Advance contract, paginating committed
/// entries to `max_committed_size` bytes per `Ready`.
pub fn RaftNode::raw_with_max_committed_size(
  self : RaftNode,
  max_committed_size : UInt64,
) -> RawNode {
  RawNode::new_sized(self, max_committed_size)
}

///|
/// View this server under AsyncStorageWrites (etcd's `Config.AsyncStorageWrites`):
/// each `Ready` hands its entries and committed entries to local storage threads
/// as `StorageAppend` / `StorageApply` directives, and the caller returns the
/// paired responses through `step_append_resp` / `step_apply_resp` once the work
/// is durable/applied. `advance` becomes a no-op; the responses drive the
/// stable/applied cursors instead.
pub fn RaftNode::raw_async(self : RaftNode) -> RawNode {
  let rn = self.raw()
  rn.async_storage = true
  rn
}

///|
/// As `raw_async`, but also paginating each `StorageApply` batch to
/// `max_committed_size` bytes (etcd's `AsyncStorageWrites` +
/// `MaxCommittedSizePerReady`).
pub fn RaftNode::raw_async_sized(
  self : RaftNode,
  max_committed_size : UInt64,
) -> RawNode {
  let rn = RawNode::new_sized(self, max_committed_size)
  rn.async_storage = true
  rn
}

///|
/// Whether this driver is in AsyncStorageWrites mode.
pub fn RawNode::is_async(self : RawNode) -> Bool {
  self.async_storage
}

///|
/// Reconcile the log with the consensus core: fold any entries the core has
/// appended (or a divergent suffix it rewrote) into the unstable tail, and pull
/// the commit index forward. The core keeps the authoritative log on `Node`;
/// this is where that truth flows into the unstable/stable split the driver
/// reports. Committed entries never change, so the scan starts just past them.
fn RawNode::sync_log(self : RawNode) -> Unit {
  let node = self.raft.node()
  // A snapshot the core installed since the last sync (a follower digesting a
  // leader-sent snapshot) has discarded the core's log below its baseline. Rebuild
  // the mirror on that baseline and hand the snapshot to the unstable tail, so it
  // surfaces once through `Ready.snapshot` — etcd's `raftLog.restore`, which the
  // consumer persists and then acknowledges via `stable_snap_to`. A local
  // compaction moves the same `snapshot_index` but installs nothing, so the count
  // of installs — not the index — is what makes a snapshot real.
  if node.snapshots_installed != self.seen_snap_installs {
    self.seen_snap_installs = node.snapshots_installed
    self.reflect_snapshot({
      last_index: node.snapshot_index,
      last_term: node.snapshot_term,
      data: b"",
      conf_state: self.raft.conf_state(),
    })
  }
  let node_last = node.last_log_index()
  let mut i = self.log.committed() + 1
  let known_last = self.log.last_index()
  while i <= node_last && i <= known_last {
    if node.term_at(i) != self.log.zero_term_on_out_of_bounds(i) {
      break
    }
    i = i + 1
  }
  if i <= node_last {
    // Upstream's `raftLog.findConflict` names a real conflict; then
    // `unstable.truncateAndAppend` reports which branch the append takes, in its
    // exact order — a plain append (fromIndex == offset + len) says nothing.
    if i <= known_last {
      self.raft.logger.log(
        Info,
        "found conflict at index \{i} [existing term: \{self.log.zero_term_on_out_of_bounds(i)}, conflicting term: \{node.term_at(i)}]",
      )
    }
    let offset = self.log.unstable_offset()
    let end = offset + self.log.unstable_entries_len().to_uint64()
    if i == end {
      // pure append: no diagnostic
    } else if i <= offset {
      self.raft.logger.log(Info, "replace the unstable entries from index \{i}")
    } else {
      self.raft.logger.log(
        Info,
        "truncate the unstable entries before index \{i}",
      )
    }
    self.log.append(node.entries_after(i - 1)) |> ignore
  }
  let last = self.log.last_index()
  self.log.commit_to(
    if node.commit_index < last {
      node.commit_index
    } else {
      last
    },
  )
  // A read confirmed by the core (a lease read, or a safe read a heartbeat quorum
  // has now acknowledged) lives in the core's read-state buffer; fold it into the
  // driver's so it surfaces in the next Ready and is cleared on accept.
  for rs in self.raft.take_read_states() {
    self.read_states.push(rs)
  }
}

///|
/// Rebuild the log mirror on a freshly installed snapshot baseline and place the
/// snapshot in the unstable tail (etcd's `raftLog.restore`). Storage is reseeded
/// to the baseline alone; any entries the core holds past it are folded back in by
/// the caller (`sync_log`). The snapshot then flows out through one `Ready` and is
/// cleared from the tail once the consumer persists it (`store` / `stable_snap_to`).
fn RawNode::reflect_snapshot(self : RawNode, snap : Snapshot) -> Unit {
  let storage = MemoryStorage::new()
  storage.seed_snapshot(snap)
  let log = RaftLog::new_with_size(storage, self.max_committed)
  log.restore(snap)
  self.log = log
  self.storage = storage
  self.reflected_snap = snap.last_index
}

///|
/// Drop the durable prefix below `new_first_index` after a local compaction
/// (etcd's `MemoryStorage.Compact`, which the rafttest `compact` command drives
/// on the node's storage). The core's own log has already been truncated by
/// `Node::compact`; this keeps the durable mirror in step so what a reader sees
/// is the post-compaction log. A request past the last durable index cannot be
/// honoured and is ignored, as the storage contract allows.
pub fn RawNode::compact_storage(
  self : RawNode,
  new_first_index : UInt64,
) -> Unit {
  let last = self.log.storage_last_index()
  if new_first_index == 0UL || new_first_index > last {
    // Nothing durable to compact to: the durable log does not reach that index
    // (the mirror only holds what the driver has been handed so far).
    return
  }
  self.storage.compact(new_first_index) catch {
    _ => ()
  }
}

///|
/// View this server through the Ready/Advance contract. The two are the same
/// server; the `RawNode` only batches the core's output into the cycle a real
/// deployment drives.
pub fn RaftNode::raw(self : RaftNode) -> RawNode {
  RawNode::new(self)
}

///|
/// The underlying core server, for reading protocol state in tests and drivers.
pub fn RawNode::node(self : RawNode) -> RaftNode {
  self.raft
}

///|
/// The snapshot available to be applied but not yet handed to a `Ready`, if any
/// (etcd's `raftLog.nextUnstableSnapshot`). Reconciles with the core first, so a
/// snapshot the core installed since the last poll is reflected into the mirror.
pub fn RawNode::next_unstable_snapshot(self : RawNode) -> Snapshot? {
  self.sync_log()
  self.log.next_unstable_snapshot()
}

///|
/// Buffer messages the core emitted so the next `Ready` carries them.
fn RawNode::record(self : RawNode, msgs : Array[Message]) -> Unit {
  for m in msgs {
    self.msgs.push(m)
  }
}

///|
/// Advance the logical clock by one tick, buffering any messages the tick emits
/// (a leader's heartbeats, or a follower's campaign) for the next `Ready`.
pub fn RawNode::tick(self : RawNode) -> Unit {
  for m in self.raft.tick() {
    self.msgs.push(m)
  }
}

///|
/// Start a campaign, buffering the vote (or pre-vote) requests it emits.
pub fn RawNode::campaign(self : RawNode) -> Unit {
  for m in self.raft.campaign() {
    self.msgs.push(m)
  }
}

///|
/// Propose a client command. On a leader it is appended and the resulting
/// AppendEntries are buffered; on a follower it is forwarded to the known leader
/// (etcd's MsgProp forwarding), or dropped if forwarding is disabled or no leader
/// is known; a candidate drops it.
pub fn RawNode::propose(self : RawNode, data : Bytes) -> Unit {
  for m in self.raft.propose(data) {
    self.msgs.push(m)
  }
}

///|
/// As `propose`, but reports whether the proposal was refused (etcd's `Step`
/// returning `ErrProposalDropped`). The datadriven harness surfaces a refusal as
/// "raft proposal dropped".
pub fn RawNode::propose_checked(self : RawNode, data : Bytes) -> Bool {
  for m in self.raft.propose(data) {
    self.msgs.push(m)
  }
  self.raft.propose_dropped()
}

///|
/// Propose a configuration change, appended and replicated like any entry so
/// every server folds it in at the same log position once committed (§6).
pub fn RawNode::propose_conf(self : RawNode, cc : ConfChange) -> Unit {
  for m in self.raft.propose_conf(cc) {
    self.msgs.push(m)
  }
}

///|
/// Propose a batch (joint) configuration change, appended and replicated like
/// any entry so every server folds the same change in at the same log position
/// once it commits (etcd's `ProposeConfChange` with a `ConfChangeV2`).
pub fn RawNode::propose_conf_v2(self : RawNode, cc : ConfChangeV2) -> Unit {
  for m in self.raft.propose_conf_v2(cc) {
    self.msgs.push(m)
  }
}

///|
/// The current membership as a `ConfState` (etcd's `ApplyConfChange` return /
/// `ConfState()`): the voters, learners, the outgoing half while joint, the
/// staged-demotion `learners_next`, and whether an auto-leave is pending.
pub fn RawNode::conf_state(self : RawNode) -> ConfState {
  self.raft.conf_state()
}

///|
/// Feed one received message to the core, buffering the replies it produces.
pub fn RawNode::step(self : RawNode, msg : Message) -> Unit {
  self.step_checked(msg) |> ignore
}

///|
/// The synthetic sender of a `StorageAppend` acknowledgement (etcd's
/// `LocalAppendThread`): a local thread, not a network peer.
pub let append_thread_id : String = "AppendThread"

///|
/// The synthetic sender of a `StorageApply` acknowledgement (etcd's
/// `LocalApplyThread`).
pub let apply_thread_id : String = "ApplyThread"

///|
/// As `step`, but reports whether the message was accepted. A reply from a peer
/// this node no longer tracks is refused rather than stepped — upstream's
/// `RawNode.Step` returns `ErrStepPeerNotFound`, which the datadriven harness
/// prints as `raft: cannot step as peer not found` (a removed server keeps
/// sending acknowledgements for a while, and there is no progress to fold them
/// into). Returns true when the message was handled.
pub fn RawNode::step_checked(self : RawNode, msg : Message) -> Bool {
  // A storage thread's reply is not a protocol message: it releases the cursors
  // that only this driver can move (etcd handles MsgStorageAppendResp and
  // MsgStorageApplyResp in `raft.Step` for the same reason).
  match msg.payload {
    StorageAppendResp(resp) => {
      self.step_append_resp(resp)
      return true
    }
    StorageApplyResp(resp) => {
      self.step_apply_resp(resp)
      return true
    }
    _ => ()
  }
  // etcd's `IsResponseMsg` gate: only the four protocol replies are checked, so
  // a proposal forwarded by a server that has since been removed still reaches
  // the leader, while its acknowledgements do not.
  let is_response = match msg.payload {
    AppendResp(_) | VoteResp(_) | PreVoteResp(_) | HeartbeatResp(_) => true
    _ => false
  }
  let from_is_local = msg.from == append_thread_id ||
    msg.from == apply_thread_id ||
    msg.from == msg.to
  // A peer is tracked when it is either in the replication progress (a leader) or
  // still part of the configuration (any other role, where this port builds no
  // progress). Only a sender that is neither has been removed outright.
  let from_is_tracked = self.raft.progress.contains(msg.from) ||
    self.raft.config.contains(msg.from)
  if is_response && !from_is_local && !from_is_tracked {
    return false
  }
  self.record(self.raft.step(msg))
  // Reconcile the mirror now so an append that rewrote a divergent suffix is
  // reported from within this step (upstream prints it from `raftLog.append`),
  // not deferred to the next Ready poll.
  self.sync_log()
  true
}

///|
/// Request a linearizable read (etcd's `ReadIndex`, which steps a `MsgReadIndex`).
/// Under `ReadOnlySafe` the read is not answered from the leader's lease but
/// confirmed by a fresh heartbeat quorum, so the confirming heartbeats are
/// buffered for the next `Ready`; the read surfaces once a quorum acknowledges the
/// position (drained from the core in `sync_log`). The synchronous lease accessor
/// `RaftNode::read_index` was wrong here: it returns an index without emitting the
/// confirmation round, so under `ReadOnlySafe` the read was never confirmed.
pub fn RawNode::read_index(self : RawNode, rctx : Bytes) -> Unit {
  for m in self.raft.request_read_index(rctx) {
    self.msgs.push(m)
  }
}

///|
/// Apply a committed configuration change to the local configuration and report
/// the resulting voter set (etcd's `ApplyConfChange`).
pub fn RawNode::apply_conf_change(
  self : RawNode,
  cc : ConfChange,
) -> Array[String] {
  cc.apply_to(self.raft.config)
  self.raft.config.voters()
}

///|
/// Begin transferring leadership to `target` (etcd's `TransferLeader`).
pub fn RawNode::transfer_leader(self : RawNode, target : String) -> Unit {
  for m in self.raft.transfer_leadership(target) {
    self.msgs.push(m)
  }
}

///|
/// Voluntarily forget the current leader so this node can start an election
/// without waiting out the election timeout (etcd's `ForgetLeader`).
pub fn RawNode::forget_leader(self : RawNode) -> Unit {
  self.raft.forget_leader()
}

///|
/// Report that a message to `id` could not be delivered (etcd's
/// `ReportUnreachable`): the leader stops streaming to that follower until it
/// responds again.
pub fn RawNode::report_unreachable(self : RawNode, id : String) -> Unit {
  self.raft.report_unreachable(id)
}

///|
/// Report the outcome of a snapshot sent to `id` (etcd's `ReportSnapshot`):
/// `reject` true means the follower could not apply it, so the leader retries.
pub fn RawNode::report_snapshot(
  self : RawNode,
  id : String,
  reject : Bool,
) -> Unit {
  self.raft.report_snapshot(id, reject)
}

///|
/// This server's basic status (etcd's `BasicStatus`): id, term, vote, commit,
/// leader and role, without the per-follower progress map.
pub fn RawNode::basic_status(self : RawNode) -> RaftStatus {
  self.raft.status()
}

///|
/// This server's full status (etcd's `Status`): basic status plus the
/// per-follower progress view and the current configuration.
pub fn RawNode::full_status(self : RawNode) -> FullStatus {
  self.raft.full_status()
}

///|
/// Visit each follower's progress (etcd's `WithProgress`).
pub fn RawNode::with_progress(
  self : RawNode,
  visit : (String, ProgressStatus) -> Unit,
) -> Unit {
  self.raft.with_progress(visit)
}

///|
/// The entries appended but not yet persisted by the application (etcd's
/// `nextUnstableEnts`): the unstable tail not already being written.
fn RawNode::next_unstable(self : RawNode) -> Array[Entry] {
  self.log.next_unstable_ents()
}

///|
/// The committed-but-unapplied entries ready to hand to the state machine
/// (etcd's `nextCommittedEnts`). In the synchronous contract the caller persists
/// `entries` before applying these, so committed entries need not be capped at
/// the last stable index (etcd's `applyUnstableEntries` == `!asyncStorageWrites`):
/// a freshly committed entry surfaces as both `entries` and `committed_entries` in
/// the same `Ready`. Under async writes application waits for the append ack, so
/// they are capped at the stable index.
fn RawNode::next_committed(self : RawNode) -> Array[Entry] {
  self.log.next_committed_ents(!self.async_storage)
}

///|
/// Assemble a `Ready` without committing to handle it (etcd's
/// `readyWithoutAccept`): a pure read that leaves the buffered messages, read
/// states, and cursors untouched, so a caller may inspect pending work and
/// decide not to consume it.
pub fn RawNode::ready_without_accept(self : RawNode) -> Ready {
  self.sync_log()
  let node = self.raft.node()
  let entries = self.next_unstable()
  let committed = self.next_committed()
  let snapshot = self.log.next_unstable_snapshot()
  let soft = self.raft.soft_state()
  let hard = node.hard_state()
  let hard_changed = hard != self.prev_hard
  let mut storage_append : StorageAppend? = None
  let mut storage_apply : StorageApply? = None
  if self.async_storage {
    // Hand the append (entries + any hard-state change) to the local append
    // thread. The response attests the current last (index, term) — not the last
    // of `entries` — so that a response arriving after a term change is dropped
    // (see StorageAppendResp / newStorageAppendResp).
    // A response waiting for durability needs an append to ride on, even when
    // nothing else is being written (etcd's `needStorageAppendMsg`).
    if !entries.is_empty() ||
      hard_changed ||
      snapshot is Some(_) ||
      self.raft.has_deferred() {
      let resp : StorageAppendResp = if self.log.has_next_or_in_progress_unstable_ents() {
        let last = self.log.last_entry_id()
        { index: last.index, log_term: last.term, term: node.current_term(), }
      } else {
        { index: 0, log_term: 0, term: node.current_term(), }
      }
      storage_append = Some({
        entries,
        hard_state: if hard_changed {
          Some(hard)
        } else {
          None
        },
        snapshot,
        resp,
        responses: self.raft.deferred_responses(),
      })
    }
    if !committed.is_empty() {
      storage_apply = Some({
        entries: committed,
        resp: { entries: committed, },
      })
    }
  }
  // Outbound work first, then the acknowledgements that were waiting for the
  // state they attest (etcd appends `msgsAfterAppend` to `Messages` in the same
  // order). Under async writes they travel inside the append directive instead.
  let messages = self.msgs.copy()
  if !self.async_storage {
    for m in self.raft.deferred_responses() {
      if m.to != self.raft.id {
        messages.push(m)
      }
    }
  }
  {
    soft_state: if soft == self.prev_soft {
      None
    } else {
      Some(soft)
    },
    hard_state: if hard_changed {
      Some(hard)
    } else {
      None
    },
    read_states: self.read_states.copy(),
    entries,
    snapshot,
    committed_entries: committed,
    messages,
    must_sync: must_sync(hard, self.prev_hard, entries.length()),
    storage_append,
    storage_apply,
  }
}

///|
/// Record that the caller has taken the given `Ready` and will handle it
/// (etcd's `acceptReady`): adopt the reported soft/hard states as the new
/// baseline, drop the served read states, and clear the drained message buffer.
/// The stable/applied watermarks move in `advance`, not here.
fn RawNode::accept_ready(self : RawNode, rd : Ready) -> Unit {
  if rd.soft_state is Some(s) {
    self.prev_soft = s
  }
  if rd.hard_state is Some(h) {
    self.prev_hard = h
  }
  // Mark the reported entries and committed entries as being written / applied,
  // so the next Ready does not re-offer them. `accept_unstable` is called
  // unconditionally, as in etcd's `acceptReady` (a no-op when the unstable tail is
  // empty or already in progress).
  self.log.accept_unstable()
  let c = rd.committed_entries.length()
  if c > 0 {
    self.log.accept_applying(
      rd.committed_entries[c - 1].index,
      ents_size(rd.committed_entries[:]),
      !self.async_storage,
    )
  }
  if !rd.read_states.is_empty() {
    self.read_states.clear()
  }
  self.msgs.clear()
  if self.async_storage {
    // The deferred responses travelled with the append directive, so the node
    // no longer holds them: the storage thread steps them when the write is
    // durable.
    self.raft.take_deferred() |> ignore
  } else {
    // A response addressed to ourselves is stepped at the advance boundary —
    // that is what makes a persisted vote turn into a leader and a persisted
    // append turn into a commit *after* the Ready that carried them (etcd's
    // `stepsOnAdvance`). The ones addressed to peers already left in `Messages`.
    for m in self.raft.take_deferred() {
      if m.to == self.raft.id {
        self.steps_on_advance.push(m)
      }
    }
  }
}

///|
/// Take the outstanding work and commit to handling it (etcd's `Ready`): the
/// same batch as `ready_without_accept`, but the messages are now drained and
/// the soft/hard baseline advanced, so the next `Ready` reports only new work.
/// The returned batch *must* be handled and passed back to `advance`.
pub fn RawNode::ready(self : RawNode) -> Ready {
  let rd = self.ready_without_accept()
  self.accept_ready(rd)
  rd
}

///|
/// Whether any work is outstanding (etcd's `HasReady`): a soft- or hard-state
/// change, buffered messages, entries to persist or apply, or pending read
/// states. Lets a driver skip building a `Ready` when the node is idle.
pub fn RawNode::has_ready(self : RawNode) -> Bool {
  self.sync_log()
  if self.raft.soft_state() != self.prev_soft {
    return true
  }
  let hard = self.raft.node().hard_state()
  if !hard.is_empty() && hard != self.prev_hard {
    return true
  }
  if self.log.has_next_unstable_snapshot() {
    return true
  }
  if !self.msgs.is_empty() || self.raft.has_deferred() {
    return true
  }
  if !self.next_unstable().is_empty() {
    return true
  }
  if !self.next_committed().is_empty() {
    return true
  }
  if !self.read_states.is_empty() {
    return true
  }
  false
}

///|
/// Persist a `Ready`'s entries to stable storage (etcd's `storage.Append(rd.
/// Entries)`, which the caller runs *before* applying `committed_entries`). This
/// is the first half of handling a synchronous `Ready`: the caller `store`s, then
/// applies `committed_entries`, then calls `advance`. A no-op under async writes,
/// where the append is instead driven by `StorageAppend` + `step_append_resp`.
pub fn RawNode::store(self : RawNode, rd : Ready) -> Unit {
  if self.async_storage {
    return
  }
  if !rd.entries.is_empty() {
    self.log.commit_stable(rd.entries)
  }
  // Persisting the snapshot to stable storage is acknowledged by dropping it from
  // the unstable tail (etcd's `appliedSnap` -> `stableSnapTo`); the mirror's
  // storage already carries the baseline from `reflect_snapshot`, so the tail is
  // simply cleared and the applied cursor caught up to the snapshot index.
  if rd.snapshot is Some(s) {
    self.log.stable_snap_to(s.last_index)
    self.log.applied_to(s.last_index, 0)
  }
}

///|
/// Notify the driver that the last `Ready` has been handled (etcd's `Advance`):
/// its entries were persisted (by `store`) and its committed entries applied, so
/// move the applied cursor past them. The stable watermark already moved in
/// `store`. Under async writes the responses drive the cursors instead, and the
/// deferred responses travel with the append directive.
///
/// The self-addressed responses that waited for durability are stepped here (etcd
/// keeps them in `stepsOnAdvance`), which is what turns a just-persisted vote into
/// a leader or a just-persisted append into a commit.
pub fn RawNode::advance(self : RawNode, rd : Ready) -> Unit {
  if self.async_storage {
    return
  }
  let c = rd.committed_entries.length()
  if c != 0 {
    let last = rd.committed_entries[c - 1].index
    self.log.applied_to(last, ents_size(rd.committed_entries[:]))
    // Applying the committed entries releases their uncommitted-tail quota
    // (etcd releases at apply time, not commit time), advances the core's applied
    // watermark, and is where a committed configuration change takes effect — so
    // the messages it emits (a probe to a new peer) are collected here.
    //
    // This runs *before* the durability responses below because that is the
    // caller's side of the contract: an application applies `CommittedEntries`
    // and only then calls `Advance` (etcd's `ApplyConfChange` therefore sees the
    // pre-advance commit index). A config change that is applied first probes the
    // newly added peers with the commit index in force at that moment.
    self.raft.advance_applied(last)
    for m in self.raft.take_pending() {
      self.msgs.push(m)
    }
  }
  // Then release the responses that waited for durability: stepping them can
  // commit the entries that just became durable and broadcast the new commit.
  for m in self.steps_on_advance {
    self.record(self.raft.step(m))
  }
  self.steps_on_advance.clear()
}

///|
/// Return a `StorageAppend` acknowledgement: the entries have been made durable.
/// The confirmed unstable prefix moves into stable storage and the unstable tail
/// is truncated — but only if the response is not stale: a response whose term is
/// below the node's current term (a later term has taken over) is ignored, and
/// the ABA guard in `async_stabilize` further requires the unstable log to still
/// hold that `(index, log_term)`.
pub fn RawNode::step_append_resp(
  self : RawNode,
  resp : StorageAppendResp,
) -> Unit {
  let term = self.raft.node().current_term()
  if resp.term < term {
    // The write completed, but the entries it attests belong to a log this node
    // has since replaced: confirming them would mark the *new* entries at those
    // indices durable. etcd reports the dropped confirmation and still honours
    // any snapshot the write carried, because snapshot contents are committed
    // and therefore term-independent.
    if resp.index != 0 {
      self.raft.logger.log(
        Info,
        "\{self.raft.id} [term: \{term}] ignored entry appends from a MsgStorageAppendResp message with lower term [term: \{resp.term}]",
      )
    }
    return
  }
  match self.log.async_stabilize_checked(resp.index, resp.log_term) {
    Stabilized => ()
    Missing =>
      self.raft.logger.log(
        Info,
        "entry at index \{resp.index} missing from unstable log; ignoring",
      )
    MatchedSnapshot =>
      self.raft.logger.log(
        Info,
        "entry at index \{resp.index} matched unstable snapshot; ignoring",
      )
    TermMismatch(existing_term~) =>
      self.raft.logger.log(
        Info,
        "entry at (index,term)=(\{resp.index},\{resp.log_term}) mismatched with entry at (\{resp.index},\{existing_term}) in unstable log; ignoring",
      )
  }
  // etcd carries the snapshot on the MsgStorageAppendResp and stabilizes it there;
  // our response does not name it, so a pending snapshot (guarded by the same
  // non-stale term check above) is acknowledged on the append that carried it.
  if self.log.pending_snapshot_index() is Some(idx) {
    self.log.stable_snap_to(idx)
    self.log.applied_to(idx, 0)
  }
}

///|
/// Write an append directive's entries to this driver's durable storage without
/// moving the stable watermark — what the local append thread of an async-storage
/// deployment does on its own schedule (etcd's rafttest `processAppend`). The
/// write is unconditional: an append that lost a race against a newer term still
/// lands, and the acknowledgement it produces is then ignored.
pub fn RawNode::append_stable(self : RawNode, entries : Array[Entry]) -> Unit {
  self.log.append_stable(entries)
}

///|
/// Return a `StorageApply` acknowledgement: the committed entries have been
/// applied. Advance the applied cursor and release the applied entries' quota.
/// Committed entries are term-independent, so there is no staleness check.
pub fn RawNode::step_apply_resp(
  self : RawNode,
  resp : StorageApplyResp,
) -> Unit {
  let n = resp.entries.length()
  if n == 0 {
    return
  }
  let last = resp.entries[n - 1].index
  self.log.applied_to(last, ents_size(resp.entries[:]))
  self.raft.advance_applied(last)
}

///|
/// This server's status snapshot (etcd's `Status`/`BasicStatus`).
pub fn RawNode::status(self : RawNode) -> RaftStatus {
  self.raft.status()
}

///|
/// Drive the node to quiescence for a caller that owns message delivery: keep
/// taking a `Ready`, treat its entries as persisted and its committed entries as
/// applied via `advance`, and collect every outbound message, until no work
/// remains. Returns the messages a transport would deliver. This is the plain,
/// synchronous read of the Ready/Advance loop that single-node drivers and the
/// browser demo run in place of a goroutine.
pub fn RawNode::stabilize(self : RawNode) -> Array[Message] {
  let out : Array[Message] = []
  while self.has_ready() {
    let rd = self.ready()
    for m in rd.messages {
      out.push(m)
    }
    // The full synchronous cycle in one place: persist, (implicitly apply), then
    // advance. A caller with its own state machine runs the same three steps.
    self.store(rd)
    self.advance(rd)
  }
  out
}