use std::ops::{Bound, RangeBounds};
use serde::{Deserialize, Serialize};
use super::membership::MembershipEntry;
use super::{NodeID, Term};
use crate::error::Result;
use crate::storage;
pub type Index = u64;
const BINCODE: bincode::config::Configuration = bincode::config::standard();
fn encode_value<T: Serialize>(value: &T) -> Vec<u8> {
bincode::serde::encode_to_vec(value, BINCODE).expect("value must be serializable")
}
fn decode_value<'de, T: Deserialize<'de>>(bytes: &'de [u8]) -> Result<T> {
Ok(bincode::serde::borrow_decode_from_slice(bytes, BINCODE)?.0)
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Entry {
pub index: Index,
pub term: Term,
pub command: Option<Vec<u8>>,
#[serde(default)]
pub membership: Option<MembershipEntry>,
}
impl Entry {
fn encode(&self) -> Vec<u8> {
encode_value(self)
}
fn decode(bytes: &[u8]) -> Result<Self> {
decode_value(bytes)
}
}
#[derive(Clone, Debug, PartialEq)]
pub enum Key {
Entry(Index),
TermVote,
CommitIndex,
SnapshotMeta,
SnapshotData,
}
impl Key {
pub fn encode(&self) -> Vec<u8> {
match self {
Key::Entry(index) => {
let mut buf = Vec::with_capacity(1 + 8);
buf.push(0x00);
buf.extend_from_slice(&index.to_be_bytes());
buf
}
Key::TermVote => vec![0x01],
Key::CommitIndex => vec![0x02],
Key::SnapshotMeta => vec![0x03],
Key::SnapshotData => vec![0x04],
}
}
}
pub struct Log {
pub engine: Box<dyn storage::Engine>,
term: Term,
vote: Option<NodeID>,
first_index: Index,
snapshot_term: Term,
last_index: Index,
last_term: Term,
commit_index: Index,
commit_term: Term,
fsync: bool,
}
impl Log {
pub fn new(mut engine: Box<dyn storage::Engine>) -> Result<Self> {
let (term, vote) = engine
.get(&Key::TermVote.encode())?
.as_deref()
.map(decode_value)
.transpose()?
.unwrap_or((0, None));
let (mut last_index, mut last_term) = engine
.scan_dyn((
Bound::Included(Key::Entry(0).encode()),
Bound::Included(Key::Entry(u64::MAX).encode()),
))
.last()
.transpose()?
.map(|(_, v)| Entry::decode(&v))
.transpose()?
.map(|e| (e.index, e.term))
.unwrap_or((0, 0));
let (mut commit_index, mut commit_term) = engine
.get(&Key::CommitIndex.encode())?
.as_deref()
.map(decode_value)
.transpose()?
.unwrap_or((0, 0));
let (snap_index, snapshot_term) = engine
.get(&Key::SnapshotMeta.encode())?
.as_deref()
.map(decode_value)
.transpose()?
.unwrap_or((0, 0));
let mut first_index = if snap_index > 0 {
snap_index + 1
} else if last_index == 0 {
1
} else {
engine
.scan_dyn((
Bound::Included(Key::Entry(0).encode()),
Bound::Included(Key::Entry(u64::MAX).encode()),
))
.next()
.transpose()?
.map(|(_, v)| Entry::decode(&v))
.transpose()?
.map(|e| e.index)
.unwrap_or(1)
};
if snap_index > 0 {
if last_index < snap_index {
last_index = snap_index;
last_term = snapshot_term;
}
if commit_index < snap_index {
commit_index = snap_index;
commit_term = snapshot_term;
}
if first_index != snap_index + 1 {
first_index = snap_index + 1;
}
}
let fsync = true; Ok(Self {
engine,
term,
vote,
first_index,
snapshot_term,
last_index,
last_term,
commit_index,
commit_term,
fsync,
})
}
pub fn get_first_index(&self) -> Index {
self.first_index
}
pub fn get_snapshot_meta(&self) -> (Index, Term) {
if self.first_index <= 1 {
(0, 0)
} else {
(self.first_index - 1, self.snapshot_term)
}
}
pub fn compact_to(&mut self, last_included_index: Index, last_included_term: Term) -> Result<()> {
assert!(last_included_index <= self.commit_index, "compact beyond commit");
if last_included_index + 1 <= self.first_index && last_included_index > 0 {
return Ok(());
}
for i in self.first_index..=last_included_index {
self.engine.delete(&Key::Entry(i).encode())?;
}
self.engine.set(
&Key::SnapshotMeta.encode(),
encode_value(&(last_included_index, last_included_term)),
)?;
if self.fsync {
self.engine.flush()?;
}
self.first_index = last_included_index + 1;
self.snapshot_term = last_included_term;
if self.last_index < last_included_index {
self.last_index = last_included_index;
self.last_term = last_included_term;
}
Ok(())
}
pub fn reset_with_snapshot(
&mut self,
last_included_index: Index,
last_included_term: Term,
) -> Result<()> {
let to_delete: Vec<_> = self
.engine
.scan_dyn((
Bound::Included(Key::Entry(0).encode()),
Bound::Included(Key::Entry(u64::MAX).encode()),
))
.filter_map(|r| r.ok().map(|(k, _)| k))
.collect();
for k in to_delete {
self.engine.delete(&k)?;
}
self.engine.set(
&Key::SnapshotMeta.encode(),
encode_value(&(last_included_index, last_included_term)),
)?;
self.commit_index = last_included_index;
self.commit_term = last_included_term;
self.engine.set(
&Key::CommitIndex.encode(),
encode_value(&(self.commit_index, self.commit_term)),
)?;
if self.fsync {
self.engine.flush()?;
}
self.first_index = last_included_index + 1;
self.snapshot_term = last_included_term;
self.last_index = last_included_index;
self.last_term = last_included_term;
Ok(())
}
pub fn enable_fsync(&mut self, fsync: bool) {
self.fsync = fsync
}
pub fn get_commit_index(&self) -> (Index, Term) {
(self.commit_index, self.commit_term)
}
pub fn get_last_index(&self) -> (Index, Term) {
(self.last_index, self.last_term)
}
pub fn get_term_vote(&self) -> (Term, Option<NodeID>) {
(self.term, self.vote)
}
pub fn set_term_vote(&mut self, term: Term, vote: Option<NodeID>) -> Result<()> {
assert!(term > 0, "can't set term 0");
assert!(term >= self.term, "term regression {} → {}", self.term, term);
assert!(term > self.term || self.vote.is_none() || vote == self.vote, "can't change vote");
if term == self.term && vote == self.vote {
return Ok(());
}
self.engine.set(&Key::TermVote.encode(), encode_value(&(term, vote)))?;
self.engine.flush()?;
self.term = term;
self.vote = vote;
Ok(())
}
pub fn append(&mut self, command: Option<Vec<u8>>) -> Result<Index> {
self.append_entry(command, None)
}
pub fn append_membership(&mut self, membership: MembershipEntry) -> Result<Index> {
self.append_entry(None, Some(membership))
}
pub fn append_entry(
&mut self,
command: Option<Vec<u8>>,
membership: Option<MembershipEntry>,
) -> Result<Index> {
assert!(self.term > 0, "can't append entry in term 0");
assert!(
command.is_none() || membership.is_none(),
"command and membership are mutually exclusive"
);
let entry = Entry {
index: self.last_index + 1,
term: self.term,
command,
membership,
};
self.engine.set(&Key::Entry(entry.index).encode(), entry.encode())?;
if self.fsync {
self.engine.flush()?;
}
self.last_index = entry.index;
self.last_term = entry.term;
Ok(entry.index)
}
pub fn latest_membership(&mut self) -> Result<Option<(Index, MembershipEntry)>> {
let mut found = None;
for entry in self.scan(1..=self.last_index) {
let entry = entry?;
if let Some(m) = entry.membership {
found = Some((entry.index, m));
}
}
Ok(found)
}
pub fn commit(&mut self, index: Index) -> Result<Index> {
let term = match self.get(index)? {
Some(entry) if entry.index < self.commit_index => {
panic!("commit index regression {} → {}", self.commit_index, entry.index);
}
Some(entry) if entry.index == self.commit_index => return Ok(index),
Some(entry) => entry.term,
None => panic!("commit index {index} does not exist"),
};
self.engine.set(&Key::CommitIndex.encode(), encode_value(&(index, term)))?;
self.commit_index = index;
self.commit_term = term;
Ok(index)
}
pub fn get(&mut self, index: Index) -> Result<Option<Entry>> {
self.engine.get(&Key::Entry(index).encode())?.map(|v| Entry::decode(&v)).transpose()
}
pub fn has(&mut self, index: Index, term: Term) -> Result<bool> {
if index == 0 || index > self.last_index {
return Ok(false);
}
if index + 1 == self.first_index && term == self.snapshot_term && index > 0 {
return Ok(true);
}
if index < self.first_index {
return Ok(false);
}
if (index, term) == (self.last_index, self.last_term) {
return Ok(true);
}
Ok(self.get(index)?.map(|e| e.term == term).unwrap_or(false))
}
pub fn scan(&mut self, range: impl RangeBounds<Index>) -> Iterator<'_> {
let start_idx = match range.start_bound() {
Bound::Excluded(&i) => i.saturating_add(1),
Bound::Included(&i) => i,
Bound::Unbounded => 0,
};
let end_idx_inclusive = match range.end_bound() {
Bound::Excluded(&i) => i.saturating_sub(1),
Bound::Included(&i) => i,
Bound::Unbounded => Index::MAX,
};
if start_idx > end_idx_inclusive {
return Iterator::new(Box::new(std::iter::empty()));
}
let from = Bound::Included(Key::Entry(start_idx).encode());
let to = Bound::Included(Key::Entry(end_idx_inclusive).encode());
Iterator::new(self.engine.scan_dyn((from, to)))
}
pub fn scan_apply(&mut self, applied_index: Index) -> Iterator<'_> {
if applied_index >= self.commit_index {
return Iterator::new(Box::new(std::iter::empty()));
}
self.scan(applied_index + 1..=self.commit_index)
}
pub fn splice(&mut self, entries: Vec<Entry>) -> Result<Index> {
let (Some(first), Some(last)) = (entries.first(), entries.last()) else {
return Ok(self.last_index); };
assert!(first.index > 0 && first.term > 0, "spliced entry has index or term 0",);
assert!(
entries.windows(2).all(|w| w[0].index + 1 == w[1].index),
"spliced entries are not contiguous"
);
assert!(
entries.windows(2).all(|w| w[0].term <= w[1].term),
"spliced entries have term regression",
);
assert!(last.term <= self.term, "splice term {} beyond current {}", last.term, self.term);
match self.get(first.index - 1)? {
Some(base) if first.term < base.term => {
panic!("splice term regression {} → {}", base.term, first.term)
}
Some(_) => {}
None if first.index == 1 => {}
None => panic!("first index {} must touch existing log", first.index),
}
let mut entries = entries.as_slice();
let mut scan = self.scan(first.index..=last.index);
while let Some(entry) = scan.next().transpose()? {
assert!(entry.index == entries[0].index, "index mismatch at {entry:?}");
if entry.term != entries[0].term {
break;
}
assert!(
entry.command == entries[0].command && entry.membership == entries[0].membership,
"command/membership mismatch at {entry:?}"
);
entries = &entries[1..];
}
drop(scan);
let Some(first) = entries.first() else {
return Ok(self.last_index);
};
assert!(first.index > self.commit_index, "spliced entries below commit index");
for entry in entries {
self.engine.set(&Key::Entry(entry.index).encode(), entry.encode())?;
}
for index in last.index + 1..=self.last_index {
self.engine.delete(&Key::Entry(index).encode())?;
}
if self.fsync {
self.engine.flush()?;
}
self.last_index = last.index;
self.last_term = last.term;
Ok(self.last_index)
}
pub fn status(&mut self) -> Result<storage::Status> {
self.engine.status()
}
}
pub struct Iterator<'a> {
inner: Box<dyn storage::ScanIterator + 'a>,
}
impl<'a> Iterator<'a> {
fn new(inner: Box<dyn storage::ScanIterator + 'a>) -> Self {
Self { inner }
}
}
impl std::iter::Iterator for Iterator<'_> {
type Item = Result<Entry>;
fn next(&mut self) -> Option<Self::Item> {
self.inner.next().map(|r| r.and_then(|(_, v)| Entry::decode(&v)))
}
}