#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct JournalGeneration(pub u128);
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, PartialOrd, Ord)]
pub struct OpNumber(pub u64);
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct WriteTicket {
generation: JournalGeneration,
first: OpNumber,
through: OpNumber,
}
impl WriteTicket {
pub const fn generation(self) -> JournalGeneration {
self.generation
}
pub const fn first(self) -> OpNumber {
self.first
}
pub const fn through(self) -> OpNumber {
self.through
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SyncTicket {
generation: JournalGeneration,
through: OpNumber,
}
impl SyncTicket {
pub const fn generation(self) -> JournalGeneration {
self.generation
}
pub const fn through(self) -> OpNumber {
self.through
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct JournalSnapshot {
pub generation: JournalGeneration,
pub accepted: OpNumber,
pub written: OpNumber,
pub durable: OpNumber,
pub faulted: bool,
}
#[derive(Debug)]
pub struct JournalProgress {
state: JournalSnapshot,
}
impl JournalProgress {
pub const fn recover(generation: JournalGeneration, through: OpNumber) -> Self {
Self {
state: JournalSnapshot {
generation,
accepted: through,
written: through,
durable: through,
faulted: false,
},
}
}
pub const fn snapshot(&self) -> JournalSnapshot {
self.state
}
pub fn admit(&mut self, operations: u64) -> Result<WriteTicket, ProgressError> {
self.require_active(self.state.generation)?;
if operations == 0 {
return Err(ProgressError::EmptyGroup);
}
let through = self
.state
.accepted
.0
.checked_add(operations)
.ok_or(ProgressError::OpExhausted)?;
let ticket = WriteTicket {
generation: self.state.generation,
first: OpNumber(self.state.accepted.0 + 1),
through: OpNumber(through),
};
self.state.accepted = ticket.through;
Ok(ticket)
}
pub fn complete_write(&mut self, ticket: WriteTicket) -> Result<(), ProgressError> {
self.require_active(ticket.generation)?;
if ticket.through > self.state.accepted {
return Err(ProgressError::BeyondAccepted);
}
if ticket.through <= self.state.written {
return Ok(());
}
if Some(ticket.first.0) != self.state.written.0.checked_add(1) {
return Err(ProgressError::WriteGap);
}
self.state.written = ticket.through;
Ok(())
}
pub fn begin_sync(&self) -> Result<SyncTicket, ProgressError> {
self.require_active(self.state.generation)?;
Ok(SyncTicket {
generation: self.state.generation,
through: self.state.written,
})
}
pub fn complete_sync(&mut self, ticket: SyncTicket) -> Result<(), ProgressError> {
self.require_active(ticket.generation)?;
if ticket.through > self.state.written {
return Err(ProgressError::BeyondWritten);
}
self.state.durable = self.state.durable.max(ticket.through);
Ok(())
}
pub fn complete_durable_write(&mut self, ticket: WriteTicket) -> Result<(), ProgressError> {
self.complete_write(ticket)?;
self.state.durable = self.state.durable.max(ticket.through);
Ok(())
}
pub fn fail(&mut self, generation: JournalGeneration) -> Result<(), ProgressError> {
if generation != self.state.generation {
return Err(ProgressError::StaleGeneration);
}
self.fault();
Ok(())
}
pub fn fault(&mut self) {
self.state.faulted = true;
}
fn require_active(&self, generation: JournalGeneration) -> Result<(), ProgressError> {
if generation != self.state.generation {
return Err(ProgressError::StaleGeneration);
}
if self.state.faulted {
return Err(ProgressError::Faulted);
}
Ok(())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
pub enum ProgressError {
#[error("journal generation is faulted")]
Faulted,
#[error("stale journal generation")]
StaleGeneration,
#[error("empty journal group")]
EmptyGroup,
#[error("journal operation number exhausted")]
OpExhausted,
#[error("write completion exceeds admitted prefix")]
BeyondAccepted,
#[error("write completion leaves a journal gap")]
WriteGap,
#[error("sync completion exceeds written prefix")]
BeyondWritten,
}
#[cfg(kani)]
mod proofs;