mod exit_view;
mod installation;
mod validation;
pub use installation::ActivationTicket;
pub use validation::ValidationTicket;
use std::time::Duration;
use ozzy_proto::NodeId;
use crate::wire::Control;
use crate::{
Admission, Configuration, Digest, InstallingView, JournalGeneration, LogSource, NormalReplica,
OpNumber, Prefix, PreparedOperation, PromiseTicket, ReplicaSnapshot, ReplicationError, Scope,
SelectedView, StartView, Status, SyncTicket, ViewChange, ViewChangeError, WriteTicket,
};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Timing {
pub heartbeat: Duration,
pub primary_timeout: Duration,
pub retransmit: Duration,
pub election_timeout: Duration,
pub max_election_timeout: Duration,
}
impl Timing {
fn validate(self, now: Duration) -> Result<(), DriverError> {
if self.heartbeat.is_zero()
|| self.retransmit.is_zero()
|| self.heartbeat >= self.primary_timeout
|| self.retransmit >= self.election_timeout
|| self.election_timeout > self.max_election_timeout
{
return Err(DriverError::InvalidTiming);
}
if now
.checked_add(self.max_election_timeout.max(self.primary_timeout))
.is_none()
{
return Err(DriverError::TimeExhausted);
}
Ok(())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Action {
Broadcast(Control),
Send {
to: NodeId,
message: Control,
},
PersistPromise(PromiseTicket),
}
#[derive(Debug)]
#[expect(
clippy::large_enum_variant,
reason = "move preallocated roles without boxing"
)]
pub enum Role {
Normal(NormalReplica),
ViewChanging(ViewChange),
Installing(InstallingView),
}
#[derive(Debug)]
pub struct ReplicaDriver {
role: Option<Role>,
configuration: Configuration,
local: NodeId,
timing: Timing,
now: Duration,
contact_deadline: Duration,
progress_deadline: Duration,
persistence_deadline: Duration,
election_deadline: Duration,
election_delay: Duration,
retry_at: Duration,
start_pending: bool,
report_pending: bool,
commit_pending: bool,
activation_pending: u8,
exit_votes: u8,
exit_retry_at: Option<Duration>,
exit_echo: Option<(Scope, u8)>,
exit_echo_at: [Duration; 3],
}
impl ReplicaDriver {
pub fn from_normal(
normal: NormalReplica,
now: Duration,
timing: Timing,
) -> Result<Self, DriverError> {
let (configuration, local) = normal.driver_identity();
Self::new(Role::Normal(normal), configuration, local, now, timing)
}
pub fn from_view_change(
changing: ViewChange,
now: Duration,
timing: Timing,
) -> Result<Self, DriverError> {
let configuration = changing.configuration;
let local = configuration.voters()[changing.local];
Self::new(
Role::ViewChanging(changing),
configuration,
local,
now,
timing,
)
}
fn new(
role: Role,
configuration: Configuration,
local: NodeId,
now: Duration,
timing: Timing,
) -> Result<Self, DriverError> {
timing.validate(now)?;
let changing = matches!(role, Role::ViewChanging(_));
let activating =
matches!(&role, Role::Normal(normal) if !normal.snapshot().ready_for_appends);
Ok(Self {
role: Some(role),
configuration,
local,
timing,
now,
contact_deadline: now + timing.primary_timeout,
persistence_deadline: now + timing.primary_timeout,
progress_deadline: now
+ if activating {
timing.election_timeout
} else {
timing.primary_timeout
},
election_deadline: now + timing.election_timeout,
election_delay: timing.election_timeout,
retry_at: now
+ if changing {
timing.retransmit
} else {
timing.heartbeat
},
start_pending: changing,
report_pending: changing,
commit_pending: false,
activation_pending: 0,
exit_votes: 0,
exit_retry_at: None,
exit_echo: None,
exit_echo_at: [Duration::ZERO; 3],
})
}
pub fn normal(&self) -> Option<&NormalReplica> {
match self.role.as_ref().expect("owned role") {
Role::Normal(normal) => Some(normal),
Role::ViewChanging(_) | Role::Installing(_) => None,
}
}
pub fn election_needs_sync(&self) -> bool {
let Some(Role::ViewChanging(changing)) = &self.role else {
return false;
};
let journal = changing.normal_snapshot().journal;
journal.written == journal.accepted && journal.written > journal.durable
}
pub fn scope(&self) -> Scope {
match self.role.as_ref().expect("owned role") {
Role::Normal(normal) => normal.snapshot().scope,
Role::ViewChanging(changing) => changing.scope(),
Role::Installing(installing) => installing.requested_scope(),
}
}
pub fn into_role(self) -> Role {
self.role.expect("owned role")
}
pub fn poll(&mut self, now: Duration) -> Result<Option<Action>, DriverError> {
self.observe_time(now)?;
if let Some(Role::Installing(installing)) = &self.role {
installing.require_pending()?;
if let Some(action) = self.exit_echo_action() {
return Ok(Some(action));
}
if now >= self.election_deadline {
self.request_exit()?;
}
return Ok(self.exit_action());
}
if let Some(Role::ViewChanging(changing)) = &self.role
&& changing.normal_snapshot().status == Status::Faulted
{
return Err(ViewChangeError::Faulted.into());
}
if let Some(action) = self.exit_echo_action() {
return Ok(Some(action));
}
if let Some(normal) = self.normal() {
let snapshot = normal.snapshot();
if snapshot.status != Status::Normal {
return Err(ReplicationError::NotNormal.into());
}
let primary = self.configuration.primary(snapshot.scope.view) == self.local;
let outstanding =
snapshot.accepted != snapshot.committed || !snapshot.ready_for_appends;
if self.normal_timed_out(&snapshot) {
self.request_exit()?;
if self.normal().is_some() {
return Ok(self.exit_action());
}
} else {
self.withdraw_exit();
if !outstanding {
self.progress_deadline = now + self.timing.primary_timeout;
}
if primary && self.commit_pending {
self.commit_pending = false;
let message = self.normal().expect("normal role").announcement()?;
return Ok(Some(Action::Broadcast(Control::Commit(message))));
}
if primary && let Some(action) = self.activation_action()? {
return Ok(Some(action));
}
if primary && now >= self.retry_at {
self.retry_at = now + self.timing.heartbeat;
if let Some(start) = self.normal().expect("normal role").start_view()? {
self.commit_pending = true;
return Ok(Some(Action::Broadcast(Control::StartView(start))));
}
let message = self.normal().expect("normal role").announcement()?;
return Ok(Some(Action::Broadcast(Control::Commit(message))));
}
return Ok(None);
}
} else if now >= self.election_deadline {
self.request_exit()?;
}
if let Some(action) = self.exit_action() {
return Ok(Some(action));
}
if now >= self.retry_at {
self.start_pending = true;
self.report_pending = true;
self.retry_at = now + self.timing.retransmit;
}
let Role::ViewChanging(changing) = self.role.as_mut().expect("owned role") else {
unreachable!("normal path returned or entered election");
};
if changing.promised_view() != changing.scope().view {
return match changing.begin_promise() {
Ok(ticket) => Ok(Some(Action::PersistPromise(ticket))),
Err(ViewChangeError::StoragePending | ViewChangeError::PromisePending) => Ok(None),
Err(error) => Err(error.into()),
};
}
if self.start_pending {
let message = changing.start_message()?;
self.start_pending = false;
return Ok(Some(Action::Broadcast(Control::StartViewChange(message))));
}
if self.report_pending {
match changing.report() {
Ok(report) => {
self.report_pending = false;
let to = self.configuration.primary(report.scope.view);
if to != self.local {
return Ok(Some(Action::Send {
to,
message: Control::DoViewChange(report),
}));
}
}
Err(ViewChangeError::StartQuorumMissing) => {}
Err(error) => return Err(error.into()),
}
}
Ok(None)
}
pub fn receive(
&mut self,
from: NodeId,
message: Control,
now: Duration,
) -> Result<Option<StartView>, DriverError> {
self.observe_time(now)?;
let voter = self.configuration.voter_index(from)?;
if from == self.local {
return Err(ReplicationError::WrongRole.into());
}
let scope = message.scope();
if scope
!= (Scope {
view: scope.view,
..self.configuration.scope()
})
{
return Err(ReplicationError::ScopeMismatch.into());
}
if let Control::ExitView(_) = message {
self.receive_exit(voter, scope)?;
return Ok(None);
}
if scope.view < self.scope().view {
return Ok(None);
}
let primary = self.configuration.primary(scope.view);
match message {
Control::Commit(_) | Control::StartView(_) if from != primary => {
return Err(ReplicationError::WrongRole.into());
}
Control::PrepareOk { .. }
| Control::PrepareRetained { .. }
| Control::DoViewChange(_)
if self.local != primary =>
{
return Err(ReplicationError::WrongRole.into());
}
_ => {}
}
if let Control::StartViewChange(_) = message {
match self.role.as_ref().expect("owned role") {
Role::Normal(normal) if normal.snapshot().status == Status::Faulted => {
return Err(ViewChangeError::Faulted.into());
}
Role::ViewChanging(changing)
if changing.normal_snapshot().status == Status::Faulted =>
{
return Err(ViewChangeError::Faulted.into());
}
Role::Installing(installing) => installing.require_pending()?,
Role::Normal(_) | Role::ViewChanging(_) => {}
}
if scope.view > self.scope().view {
self.change_view(scope.view)?;
}
if scope.view != self.scope().view {
return Ok(None);
}
}
if scope.view > self.scope().view {
self.change_view(scope.view)?;
}
match self.role.as_mut().expect("owned role") {
Role::Normal(normal) => {
let committed = normal.snapshot().committed;
match message {
Control::Commit(commit) => {
normal.receive_commit(from, commit)?;
self.contact_deadline = now + self.timing.primary_timeout;
}
Control::PrepareOk { ack, .. } => normal.receive_ack(from, ack)?,
Control::PrepareRetained { ack, .. } => {
normal.receive_retained_ack(from, ack)?;
}
Control::StartView(start) => {
normal.observe_installed_start(from, start)?;
self.contact_deadline = now + self.timing.primary_timeout;
}
_ => return Ok(None),
}
if let Some(prefix) = message.acknowledged()
&& let Some(start) = normal.start_view()?
&& prefix == start.accepted
&& start.accepted.op < normal.snapshot().committed.op
{
self.activation_pending |= 1 << voter;
}
if normal.snapshot().committed != committed {
self.progress_deadline = now + self.progress_timeout();
}
}
Role::ViewChanging(changing) => match message {
Control::StartViewChange(start) => changing.receive_start(from, start)?,
Control::DoViewChange(report) => changing.receive_report(from, report)?,
Control::StartView(start) => return Ok(Some(start)),
Control::Commit(_)
| Control::PrepareOk { .. }
| Control::PrepareRetained { .. }
| Control::ExitView(_) => {}
},
Role::Installing(installing) => installing.observe_view(from, scope)?,
}
Ok(None)
}
pub fn complete_promise(&mut self, ticket: PromiseTicket) -> Result<(), DriverError> {
self.changing()?.complete_promise(ticket)?;
Ok(())
}
pub fn fail_promise(&mut self, ticket: PromiseTicket) -> Result<(), DriverError> {
self.changing()?.fail_promise(ticket)?;
Ok(())
}
pub fn prepare(
&mut self,
from: NodeId,
scope: Scope,
operations: &[PreparedOperation],
now: Duration,
) -> Result<Admission, DriverError> {
self.observe_time(now)?;
self.configuration.voter_index(from)?;
if scope
!= (Scope {
view: scope.view,
..self.configuration.scope()
})
{
return Err(ReplicationError::ScopeMismatch.into());
}
if from != self.configuration.primary(scope.view) {
return Err(ReplicationError::WrongRole.into());
}
if scope.view > self.scope().view {
self.change_view(scope.view)?;
}
let Role::Normal(normal) = self.role.as_mut().expect("owned role") else {
return Err(ReplicationError::NotNormal.into());
};
let previous = normal.snapshot();
let admitted = normal.prepare(from, scope, operations)?;
if normal.snapshot().accepted != previous.accepted
&& previous.accepted == previous.committed
&& previous.ready_for_appends
{
self.progress_deadline = now + self.timing.primary_timeout;
}
if normal.snapshot().accepted != previous.accepted
&& previous.accepted.op == self.storage_progress(&previous)
{
self.persistence_deadline = now + self.timing.primary_timeout;
}
self.contact_deadline = now + self.timing.primary_timeout;
Ok(admitted)
}
pub fn complete_write(&mut self, ticket: WriteTicket) -> Result<(), DriverError> {
match self.role.as_mut().expect("owned role") {
Role::Normal(normal) => normal.complete_write(ticket)?,
Role::ViewChanging(changing) => changing.complete_write(ticket)?,
Role::Installing(_) => return Err(ReplicationError::WrongRole.into()),
}
Ok(())
}
pub fn complete_buffered_write(
&mut self,
ticket: WriteTicket,
now: Duration,
) -> Result<(), DriverError> {
if self.configuration.policy() != crate::QuorumPolicy::Replicated {
return Err(ReplicationError::PolicyMismatch.into());
}
self.observe_time(now)?;
let previous = self.normal().map(NormalReplica::snapshot);
self.complete_write(ticket)?;
if let Some(previous) = previous {
self.observe_normal_progress(&previous, now);
}
Ok(())
}
pub fn complete_durable_write(
&mut self,
ticket: WriteTicket,
now: Duration,
) -> Result<(), DriverError> {
self.observe_time(now)?;
match self.role.as_mut().expect("owned role") {
Role::Normal(normal) => {
let previous = normal.snapshot();
normal.complete_durable_write(ticket)?;
self.observe_normal_progress(&previous, now);
}
Role::ViewChanging(changing) => changing.normal.complete_durable_write(ticket)?,
Role::Installing(_) => return Err(ReplicationError::WrongRole.into()),
}
Ok(())
}
pub fn apply_through(&mut self, through: Prefix) -> Result<(), DriverError> {
let Role::Normal(normal) = self.role.as_mut().expect("owned role") else {
return Err(ReplicationError::NotNormal.into());
};
let was_ready = normal.snapshot().ready_for_appends;
normal.apply_through(through)?;
if !was_ready && normal.snapshot().ready_for_appends {
self.progress_deadline = self.now + self.timing.primary_timeout;
self.election_delay = self.timing.election_timeout;
}
Ok(())
}
pub fn fail_io(&mut self, generation: JournalGeneration) -> Result<(), DriverError> {
match self.role.as_mut().expect("owned role") {
Role::Normal(normal) => normal.fail_io(generation)?,
Role::ViewChanging(changing) => changing.fail_io(generation)?,
Role::Installing(installing) => {
let ticket = installing.ticket();
if generation != ticket.generation() {
return Err(ViewChangeError::StaleInstallation.into());
}
installing.fail(ticket)?;
}
}
Ok(())
}
pub fn begin_sync(&self) -> Result<SyncTicket, DriverError> {
match self.role.as_ref().expect("owned role") {
Role::Normal(normal) => Ok(normal.begin_sync()?),
Role::ViewChanging(changing) => Ok(changing.begin_sync()?),
Role::Installing(_) => Err(ReplicationError::WrongRole.into()),
}
}
pub fn complete_sync(&mut self, ticket: SyncTicket, now: Duration) -> Result<(), DriverError> {
self.observe_time(now)?;
match self.role.as_mut().expect("owned role") {
Role::Normal(normal) => {
let previous = normal.snapshot();
normal.complete_sync(ticket)?;
self.observe_normal_progress(&previous, now);
}
Role::ViewChanging(changing) => changing.complete_sync(ticket)?,
Role::Installing(_) => return Err(ReplicationError::WrongRole.into()),
}
Ok(())
}
pub fn select(
&mut self,
lookup: impl FnMut(LogSource, OpNumber) -> Option<Digest>,
) -> Result<SelectedView, DriverError> {
Ok(self.changing()?.select(lookup)?)
}
pub fn history_sources(&self) -> [Option<LogSource>; 3] {
match self.role.as_ref().expect("owned role") {
Role::ViewChanging(changing) => changing.history_sources(),
Role::Normal(_) | Role::Installing(_) => [None; 3],
}
}
fn changing(&mut self) -> Result<&mut ViewChange, DriverError> {
match self.role.as_mut().expect("owned role") {
Role::ViewChanging(changing) => Ok(changing),
Role::Normal(_) | Role::Installing(_) => Err(ReplicationError::WrongRole.into()),
}
}
fn observe_time(&mut self, now: Duration) -> Result<(), DriverError> {
if now < self.now {
return Err(DriverError::ClockRegressed);
}
self.timing.validate(now)?;
self.now = now;
Ok(())
}
fn progress_timeout(&self) -> Duration {
if self
.normal()
.is_some_and(|normal| normal.snapshot().ready_for_appends)
{
self.timing.primary_timeout
} else {
self.election_delay
}
}
fn observe_normal_progress(&mut self, previous: &ReplicaSnapshot, now: Duration) {
let current = self.normal().expect("normal completion").snapshot();
if current.committed != previous.committed {
self.progress_deadline = now + self.progress_timeout();
}
if self.storage_progress(¤t) > self.storage_progress(previous) {
self.persistence_deadline = now + self.timing.primary_timeout;
}
}
fn normal_timed_out(&self, snapshot: &ReplicaSnapshot) -> bool {
let primary = self.configuration.primary(snapshot.scope.view) == self.local;
let outstanding = snapshot.accepted != snapshot.committed || !snapshot.ready_for_appends;
(!primary && self.now >= self.contact_deadline)
|| (outstanding && self.now >= self.progress_deadline)
|| (snapshot.accepted.op > self.storage_progress(snapshot)
&& self.now >= self.persistence_deadline)
}
fn storage_progress(&self, snapshot: &ReplicaSnapshot) -> OpNumber {
match self.configuration.policy() {
crate::QuorumPolicy::Durable => snapshot.journal.durable,
crate::QuorumPolicy::Replicated => snapshot.journal.written,
}
}
fn next_view(&mut self) -> Result<(), DriverError> {
let view = self
.scope()
.view
.checked_add(1)
.ok_or(ViewChangeError::ViewExhausted)?;
self.change_view(view)
}
fn change_view(&mut self, view: u64) -> Result<(), DriverError> {
if let Some(Role::ViewChanging(changing)) = self.role.as_mut() {
changing.advance_view(view)?;
} else if let Some(Role::Installing(installing)) = self.role.as_mut() {
installing.request_view(view)?;
} else {
if self.normal().expect("normal role").snapshot().status == Status::Faulted {
return Err(ViewChangeError::Faulted.into());
}
let Role::Normal(normal) = self.role.take().expect("owned role") else {
unreachable!("normal branch");
};
self.role = Some(Role::ViewChanging(
normal
.into_view_change(view)
.expect("healthy role and higher view"),
));
}
self.exit_votes = 0;
self.exit_retry_at = None;
self.election_deadline = self.now + self.election_delay;
self.retry_at = self.now + self.timing.retransmit;
self.start_pending = true;
self.report_pending = true;
self.commit_pending = false;
self.activation_pending = 0;
Ok(())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
pub enum DriverError {
#[error("invalid replica timing intervals")]
InvalidTiming,
#[error("replica clock regressed")]
ClockRegressed,
#[error("replica clock exhausted")]
TimeExhausted,
#[error("stale replica application validation")]
StaleValidation,
#[error(transparent)]
Replication(#[from] ReplicationError),
#[error(transparent)]
ViewChange(#[from] ViewChangeError),
}