// 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.
///|
/// The one way a `RaftLog` read can fail: the requested index predates the
/// snapshot baseline and has been compacted away. `slice` / `entries` /
/// `must_check_out_of_bounds` narrow the storage layer's four-variant
/// `StorageError` to this single mode at the boundary — a caller such as
/// `all_entries` then handles exactly the failure that can occur, and no wildcard
/// arm can silently swallow an `Unavailable` or contract-violating error (etcd
/// documents that `raftLog.slice` only ever returns `ErrCompacted`).
pub suberror LogCompacted
///|
fn u64_min(a : UInt64, b : UInt64) -> UInt64 {
if a < b {
a
} else {
b
}
}
///|
fn u64_max(a : UInt64, b : UInt64) -> UInt64 {
if a > b {
a
} else {
b
}
}
///|
/// The raft log (etcd's `raftLog`): a `MemoryStorage` of durable entries with an
/// `Unstable` in-memory tail layered on top, plus the commit/apply cursors.
///
/// `committed` is the highest index known committed on a quorum; `applying` and
/// `applied` track how far the state machine has been told to apply and has
/// finished applying. `applying_ents_size` / `max_applying_ents_size` bound the
/// bytes of committed-but-unapplied entries handed out at once, so a burst of
/// commits cannot force an unbounded apply batch (byte-level pagination).
pub struct RaftLog {
storage : &RaftStorage
unstable : Unstable
mut committed : UInt64
mut applying : UInt64
mut applied : UInt64
max_applying_ents_size : UInt64
mut applying_ents_size : UInt64
mut applying_ents_paused : Bool
}
///|
/// Recover a log from `storage`, positioned at the last compaction: committed,
/// applying and applied all start at `first_index - 1`, and the unstable tail is
/// empty just past `last_index`.
pub fn RaftLog::new(storage : MemoryStorage) -> RaftLog {
RaftLog::new_with_size(storage, no_limit)
}
///|
/// As `new`, but capping the byte size of entries returned per apply batch.
pub fn RaftLog::new_with_size(
storage : &RaftStorage,
max_applying_ents_size : UInt64,
) -> RaftLog {
let first = storage.first_index()
let last = storage.last_index()
{
storage,
unstable: Unstable::new(last + 1),
committed: first - 1,
applying: first - 1,
applied: first - 1,
max_applying_ents_size,
applying_ents_size: 0,
applying_ents_paused: false,
}
}
///|
/// A one-line description of the log's cursors for debugging (etcd's
/// `raftLog.String()`).
pub fn RaftLog::to_string(self : RaftLog) -> String {
"committed=\{self.committed}, applied=\{self.applied}, applying=\{self.applying}, unstable.offset=\{self.unstable.offset}, unstable.offsetInProgress=\{self.unstable.offset_in_progress}, len(unstable.Entries)=\{self.unstable.entries.length()}"
}
///|
/// Seed the commit and applied cursors from a recovered node on construction.
/// Commit only moves forward; applied and applying jump to the recovered point.
pub fn RaftLog::seed(
self : RaftLog,
committed : UInt64,
applied : UInt64,
) -> Unit {
if committed > self.committed {
self.commit_to(committed)
}
self.applied = applied
self.applying = u64_max(self.applying, applied)
}
///|
/// Move `entries` from the unstable tail into stable storage, then mark them
/// stable — the persist half of a driver's Ready/Advance cycle.
pub fn RaftLog::commit_stable(self : RaftLog, entries : Array[Entry]) -> Unit {
if entries.is_empty() {
return
}
self.storage.append(entries)
self.stable_to(entries[entries.length() - 1].id())
}
///|
/// Acknowledge an asynchronous append made durable up to `(index, log_term)`:
/// move the confirmed unstable prefix into storage and truncate the unstable
/// tail. This is a no-op — the ABA guard — unless the unstable log *still* holds
/// `(index, log_term)`; if a later term rewrote that index, the stale ack must
/// not be mistaken for a confirmation of the new entries.
pub fn RaftLog::async_stabilize(
self : RaftLog,
index : UInt64,
log_term : UInt64,
) -> Unit {
if index == 0 {
return
}
match self.unstable.maybe_term(index) {
Some(t) =>
if t == log_term {
if index >= self.unstable.offset {
self.storage.append(
self.unstable.slice(self.unstable.offset, index + 1),
)
}
self.stable_to({ term: log_term, index })
}
None => ()
}
}
///|
/// The committed index.
pub fn RaftLog::committed(self : RaftLog) -> UInt64 {
self.committed
}
///|
/// The first index still readable (one past the snapshot).
pub fn RaftLog::first_index(self : RaftLog) -> UInt64 {
match self.unstable.maybe_first_index() {
Some(i) => i
None => self.storage.first_index()
}
}
///|
/// The index of the last entry in the log.
pub fn RaftLog::last_index(self : RaftLog) -> UInt64 {
match self.unstable.maybe_last_index() {
Some(i) => i
None => self.storage.last_index()
}
}
///|
/// The term of entry `i`. `Compacted` if it predates the first index,
/// `Unavailable` if it is past the last. The term at `first_index-1` is retained
/// for log-matching even though the entry itself is gone.
pub fn RaftLog::term(self : RaftLog, i : UInt64) -> UInt64 raise StorageError {
// Consult the unstable tail first; a hit there is always in range.
match self.unstable.maybe_term(i) {
Some(t) => t
None => {
if i + 1 < self.first_index() {
raise Compacted
}
if i > self.last_index() {
raise Unavailable
}
self.storage.storage_term(i)
}
}
}
///|
/// The term at `i`, or 0 when the index is out of bounds (etcd's
/// `zeroTermOnOutOfBounds`), used where a missing term is simply "no match".
pub fn RaftLog::zero_term_on_out_of_bounds(
self : RaftLog,
i : UInt64,
) -> UInt64 {
self.term(i) catch {
Compacted => 0
Unavailable => 0
// etcd's zeroTermOnOutOfBounds panics on any other error; term only ever
// raises the two above, so this arm exists solely for exhaustiveness.
_ => abort("unexpected error getting term")
}
}
///|
/// Whether the log holds the identified entry with that exact term.
pub fn RaftLog::match_term(self : RaftLog, id : EntryId) -> Bool {
(Some(self.term(id.index)) catch { _ => None }) == Some(id.term)
}
///|
/// The identity of the last entry.
pub fn RaftLog::last_entry_id(self : RaftLog) -> EntryId {
let index = self.last_index()
let t = self.term(index) catch {
_ => abort("unexpected error getting last term")
}
{ term: t, index }
}
///|
/// Whether the log ending at `their` is at least as up-to-date as ours (§5.4.1).
pub fn RaftLog::is_up_to_date(self : RaftLog, their : EntryId) -> Bool {
let our = self.last_entry_id()
their.term > our.term || (their.term == our.term && their.index >= our.index)
}
///|
/// Try to append the slice `a` after verifying its `prev` matches our log. On a
/// match, splice in any genuinely new tail (a conflict with a committed entry
/// aborts), advance commit to `min(committed, last_new)`, and return the new
/// last index; otherwise return `None`.
pub fn RaftLog::maybe_append(
self : RaftLog,
a : LogSlice,
committed : UInt64,
) -> UInt64? {
if !self.match_term(a.prev) {
return None
}
let lastnewi = a.prev.index + a.entries.length().to_uint64()
let ci = self.find_conflict(a.entries[:])
if ci == 0 {
()
} else if ci <= self.committed {
abort("entry conflict with committed entry")
} else {
let offset = a.prev.index + 1
if ci - offset > a.entries.length().to_uint64() {
abort("index out of range")
}
self.append(a.entries[(ci - offset).to_int():].to_owned()) |> ignore
}
self.commit_to(u64_min(committed, lastnewi))
Some(lastnewi)
}
///|
/// Append `ents` to the unstable tail and return the new last index. Appending
/// at or before the commit index is a corruption and aborts.
pub fn RaftLog::append(self : RaftLog, ents : Array[Entry]) -> UInt64 {
if ents.is_empty() {
return self.last_index()
}
let after = ents[0].index - 1
if after < self.committed {
abort("append after is out of range [committed]")
}
self.unstable.truncate_and_append(ents)
self.last_index()
}
///|
/// The index of the first entry in `ents` that conflicts with our log (same
/// index, different term), or the first genuinely new index, or 0 if all match.
pub fn RaftLog::find_conflict(
self : RaftLog,
ents : ArrayView[Entry],
) -> UInt64 {
for e in ents {
let id = e.id()
if !self.match_term(id) {
return id.index
}
}
0
}
///|
/// A best guess at where our log stops matching another whose only known point
/// is `(index, term)`: the greatest `i <= index` with `term(i) <= term`, or with
/// an unknown (compacted/unstored) term. Returns `(i, term(i)-or-0)`.
pub fn RaftLog::find_conflict_by_term(
self : RaftLog,
index : UInt64,
term : UInt64,
) -> (UInt64, UInt64) {
let mut i = index
while i > 0 {
match (Some(self.term(i)) catch { _ => None }) {
None => return (i, 0)
Some(our) => if our <= term { return (i, our) }
}
i = i - 1
}
(0, 0)
}
///|
/// The committed entries whose term matches — advance commit to `at.index`.
pub fn RaftLog::maybe_commit(self : RaftLog, at : EntryId) -> Bool {
if at.term != 0 && at.index > self.committed && self.match_term(at) {
self.commit_to(at.index)
true
} else {
false
}
}
///|
/// Advance the commit index (never backwards). Committing past the last index is
/// a corruption and aborts.
pub fn RaftLog::commit_to(self : RaftLog, tocommit : UInt64) -> Unit {
if self.committed < tocommit {
if self.last_index() < tocommit {
abort("tocommit is out of range [lastIndex]")
}
self.committed = tocommit
}
}
///|
/// Record that the state machine has finished applying up to `i`, releasing
/// `size` bytes of the outstanding apply budget.
pub fn RaftLog::applied_to(self : RaftLog, i : UInt64, size : UInt64) -> Unit {
if self.committed < i || i < self.applied {
abort("applied is out of range [prevApplied, committed]")
}
self.applied = i
self.applying = u64_max(self.applying, i)
self.applying_ents_size = if self.applying_ents_size > size {
self.applying_ents_size - size
} else {
0
}
self.applying_ents_paused = self.applying_ents_size >=
self.max_applying_ents_size
}
///|
/// Record that the application has been handed entries up to `i` to apply,
/// charging `size` bytes against the budget and pausing when it is exhausted or
/// when the next entry would overshoot it.
pub fn RaftLog::accept_applying(
self : RaftLog,
i : UInt64,
size : UInt64,
allow_unstable : Bool,
) -> Unit {
if self.committed < i {
abort("applying is out of range [prevApplying, committed]")
}
self.applying = i
self.applying_ents_size = self.applying_ents_size + size
self.applying_ents_paused = self.applying_ents_size >=
self.max_applying_ents_size ||
i < self.max_appliable_index(allow_unstable)
}
///|
/// The highest index that may be applied: the commit index, capped at the last
/// stable index unless unstable entries are allowed.
fn RaftLog::max_appliable_index(
self : RaftLog,
allow_unstable : Bool,
) -> UInt64 {
let hi = self.committed
if !allow_unstable {
u64_min(hi, self.unstable.offset - 1)
} else {
hi
}
}
///|
/// Confirm the unstable tail entries up to `id` are durably stored.
pub fn RaftLog::stable_to(self : RaftLog, id : EntryId) -> Unit {
self.unstable.stable_to(id)
}
///|
/// Confirm the unstable snapshot at index `i` is durably stored.
pub fn RaftLog::stable_snap_to(self : RaftLog, i : UInt64) -> Unit {
self.unstable.stable_snap_to(i)
}
///|
/// The index of the snapshot still held in the unstable tail, if any (whether or
/// not its write is in progress). Used to acknowledge a snapshot made durable
/// under async storage writes, where the append response does not name it.
pub fn RaftLog::pending_snapshot_index(self : RaftLog) -> UInt64? {
self.unstable.snapshot.map(s => s.last_index)
}
///|
/// Mark the current unstable entries and snapshot as being written.
pub fn RaftLog::accept_unstable(self : RaftLog) -> Unit {
self.unstable.accept_in_progress()
}
///|
/// The unstable entries ready to be written and not already in progress.
pub fn RaftLog::next_unstable_ents(self : RaftLog) -> Array[Entry] {
self.unstable.next_entries()
}
///|
/// Whether any unstable entries are ready to be written.
pub fn RaftLog::has_next_unstable_ents(self : RaftLog) -> Bool {
!self.next_unstable_ents().is_empty()
}
///|
/// Whether there are any unstable entries, whether or not already in progress.
pub fn RaftLog::has_next_or_in_progress_unstable_ents(self : RaftLog) -> Bool {
!self.unstable.entries.is_empty()
}
///|
fn RaftLog::has_next_or_in_progress_snapshot(self : RaftLog) -> Bool {
self.unstable.snapshot is Some(_)
}
///|
/// The unstable snapshot ready to be applied and not already in progress.
pub fn RaftLog::next_unstable_snapshot(self : RaftLog) -> Snapshot? {
self.unstable.next_snapshot()
}
///|
/// Whether an unstable snapshot is ready to be applied.
pub fn RaftLog::has_next_unstable_snapshot(self : RaftLog) -> Bool {
self.unstable.next_snapshot() is Some(_)
}
///|
/// The committed-but-unapplied entries ready for execution, subject to the apply
/// pause, any pending snapshot, and the byte budget. `allow_unstable` lets
/// committed entries still in the unstable tail be applied.
pub fn RaftLog::next_committed_ents(
self : RaftLog,
allow_unstable : Bool,
) -> Array[Entry] {
if self.applying_ents_paused {
return []
}
if self.has_next_or_in_progress_snapshot() {
return []
}
let lo = self.applying + 1
let hi = self.max_appliable_index(allow_unstable) + 1
if lo >= hi {
return []
}
let max_size = self.max_applying_ents_size - self.applying_ents_size
self.slice(lo, hi, max_size) catch {
LogCompacted => abort("unexpected error getting unapplied entries")
}
}
///|
/// Whether any committed-but-unapplied entries are ready (a light check that
/// avoids the `slice` in `next_committed_ents`).
pub fn RaftLog::has_next_committed_ents(
self : RaftLog,
allow_unstable : Bool,
) -> Bool {
if self.applying_ents_paused {
return false
}
if self.has_next_or_in_progress_snapshot() {
return false
}
let lo = self.applying + 1
let hi = self.max_appliable_index(allow_unstable) + 1
lo < hi
}
///|
/// Restore the log to a snapshot baseline: commit follows the snapshot forward,
/// and the unstable tail is replaced by it.
pub fn RaftLog::restore(self : RaftLog, s : Snapshot) -> Unit {
self.committed = s.last_index
self.unstable.restore(s)
}
///|
/// The most recent snapshot: the unstable one if present, else storage's.
pub fn RaftLog::snapshot(self : RaftLog) -> Snapshot raise StorageError {
match self.unstable.snapshot {
Some(s) => s
None => self.storage.storage_snapshot()
}
}
///|
/// Entries starting at `i`, size-capped by `max_size`. Empty if `i` is past the
/// end; raises `Compacted` if `i` has been compacted.
pub fn RaftLog::entries(
self : RaftLog,
i : UInt64,
max_size : UInt64,
) -> Array[Entry] raise LogCompacted {
if i > self.last_index() {
return []
}
self.slice(i, self.last_index() + 1, max_size)
}
///|
/// Every entry currently in the log.
pub fn RaftLog::all_entries(self : RaftLog) -> Array[Entry] {
// `entries` narrows its failure to `LogCompacted`, so the retry on a racing
// compaction (etcd's `return l.allEntries()`) is the whole catch — there is no
// other error mode and thus no wildcard arm.
self.entries(self.first_index(), no_limit) catch {
LogCompacted => self.all_entries()
}
}
///|
/// Visit `[lo, hi)` in consecutive byte-bounded pages, passing each page to `v`.
/// `v` may raise to stop early. Each page holds at least one entry and at most
/// `page_size` bytes (unless a single entry exceeds it).
pub fn RaftLog::scan(
self : RaftLog,
lo : UInt64,
hi : UInt64,
page_size : UInt64,
v : (ArrayView[Entry]) -> Unit raise,
) -> Unit raise {
let mut lo = lo
while lo < hi {
let ents = self.slice(lo, hi, page_size)
// etcd's `scan` panics if `slice` hands back an empty page, which would leave
// `lo` unadvanced and loop forever; a `MemoryStorage` never does, but a broken
// backend could, so the guard stands.
if ents.is_empty() {
abort("scan got 0 entries")
}
v(ents[:])
lo = lo + ents.length().to_uint64()
}
}
///|
/// Entries with indices in `[lo, hi)`, size-capped by `max_size`, drawn from the
/// stable storage and the unstable tail and stitched together. `Compacted` if
/// `lo` predates the first index.
pub fn RaftLog::slice(
self : RaftLog,
lo : UInt64,
hi : UInt64,
max_size : UInt64,
) -> Array[Entry] raise LogCompacted {
self.must_check_out_of_bounds(lo, hi)
if lo == hi {
return []
}
if lo >= self.unstable.offset {
return limit_size(self.unstable.slice(lo, hi)[:], max_size)
}
let cut = u64_min(hi, self.unstable.offset)
// etcd's `raftLog.slice` dispatches on the storage error and only ever surfaces
// `ErrCompacted`: it propagates (a caller such as `all_entries` retries a racing
// compaction), while `ErrUnavailable` and any other error panic. Narrowing the
// storage layer's `StorageError` to `LogCompacted` here makes that fact a type,
// so no caller's catch can fold `Unavailable` or a contract-violating error into
// the compaction path.
let ents = self.storage.storage_entries(lo, cut, max_size) catch {
Compacted => raise LogCompacted
Unavailable => abort("entries are unavailable from storage")
_ => abort("unexpected error reading entries from storage")
}
if hi <= self.unstable.offset {
return ents
}
// The storage read may already have hit the size cap.
if ents.length().to_uint64() < cut - lo {
return ents
}
let size = ents_size(ents[:])
if size >= max_size {
return ents
}
let unstable_part = limit_size(
self.unstable.slice(self.unstable.offset, hi)[:],
max_size - size,
)
// A lone over-budget unstable entry is dropped rather than exceeding the cap.
if unstable_part.length() == 1 &&
size + ents_size(unstable_part[:]) > max_size {
return ents
}
let out : Array[Entry] = []
for e in ents {
out.push(e)
}
for e in unstable_part {
out.push(e)
}
out
}
///|
/// Guard: `first_index <= lo <= hi <= last_index + 1`. `Compacted` when `lo`
/// predates the first index; a high bound past the end aborts (etcd panics).
pub fn RaftLog::must_check_out_of_bounds(
self : RaftLog,
lo : UInt64,
hi : UInt64,
) -> Unit raise LogCompacted {
if lo > hi {
abort("invalid slice: lo > hi")
}
let fi = self.first_index()
if lo < fi {
raise LogCompacted
}
let length = self.last_index() + 1 - fi
if hi > fi + length {
abort("slice out of bound")
}
}