use super::super::observed::{ObservedTable, SourceId};
use super::super::{
WarpTable, WarpTableGetOutcome, WarpTableRemoveOutcome, WarpTableRow, WarpTableUpdateOutcome,
};
use super::{changes, *};
use crate::{StateMachine, StepInputOf};
use alloc::{collections::VecDeque, vec::Vec};
use core::num::NonZeroU64;
#[derive(WarpTableRow)]
#[index(primary = SessionId)]
struct Session {
sequence: u64,
}
#[derive(WarpTableRow)]
#[index(primary = OperationId)]
struct Retained<Row: PersistentRow> {
record: Record<Row, Row::Id>,
}
pub struct Authority<Row: PersistentRow> {
history: HistoryId,
epoch: u64,
clock: RevisionClock,
row_clock: RevisionClock,
next_session: u64,
next_attempt: u64,
limits: Limits,
replication: Replication,
accepted: ObservedTable<Row>,
staging: WarpTable<Row>,
local: ObservedTable<Row>,
remote: Option<ObservedTable<Row>>,
accepted_revision: Revision,
local_revision: Revision,
remote_revision: Revision,
restored_revision: Revision,
remote_ready: bool,
records: WarpTable<Retained<Row>>,
order: VecDeque<OperationId>,
sessions: WarpTable<Session>,
undo: Vec<(Row::Id, Row)>,
incarnation_scratch: Vec<(Row::Id, u64)>,
local_attempt: Option<Attempt>,
remote_attempt: Option<Attempt>,
poisoned: bool,
observation_incarnation_valid: bool,
}
impl<Row: PersistentRow> StateMachine for Authority<Row> {}
impl<Row: PersistentRow> Authority<Row> {
pub fn validate_changes(&mut self, proposed: &[Change<Row, Row::Id>]) -> Result<(), Error> {
if self.poisoned {
return Err(Error::Corrupt);
}
if proposed.len() > self.limits.changes {
return Err(Error::InvalidBounds);
}
let mut inverse = Vec::new();
inverse
.try_reserve_exact(proposed.len())
.map_err(|_| Error::Allocation)?;
for change in proposed {
match change {
Change::Insert { id, .. } => inverse.push(Change::Remove { id: *id }),
Change::Replace { id, .. } | Change::Remove { id } => {
let row = match self.accepted.table().get(*id) {
WarpTableGetOutcome::Present { row } => Row::own(row),
WarpTableGetOutcome::Absent { .. } => return Err(Error::Absent),
};
inverse.push(match change {
Change::Replace { .. } => Change::Replace { id: *id, row },
Change::Remove { .. } => Change::Insert { id: *id, row },
Change::Insert { .. } => return Err(Error::Corrupt),
});
}
}
}
if let Err(error) = changes::apply(&mut self.staging, proposed, &mut self.undo) {
if error == Error::Corrupt {
self.poisoned = true;
}
return Err(error);
}
if changes::apply(&mut self.staging, &inverse, &mut self.undo).is_err() {
self.poisoned = true;
return Err(Error::Corrupt);
}
Ok(())
}
pub fn recover(
recovery: Recovery<Row, Row::Id>,
reserved_through: u64,
limits: Limits,
replication: Replication,
) -> Result<Self, Error> {
if limits.changes == 0 || limits.row_bytes == 0 {
return Err(Error::InvalidBounds);
}
if recovery.revision.0 > recovery.reserved_through
|| (reserved_through <= recovery.reserved_through && reserved_through != u64::MAX)
{
return Err(Error::InvalidRecovery);
}
let epoch = recovery.reserved_through.saturating_add(1);
let create = || WarpTable::<Row>::try_new(limits.rows).map_err(|_| Error::Allocation);
let mut accepted = create()?;
let mut staging = create()?;
let mut local = create()?;
let mut remote = match replication {
Replication::Disabled => None,
Replication::Required => Some(create()?),
};
for (id, row) in recovery.rows {
if row.retained_bytes() > limits.row_bytes {
return Err(Error::InvalidBounds);
}
changes::insert(&mut accepted, id, row.clone())?;
changes::insert(&mut staging, id, row.clone())?;
changes::insert(&mut local, id, row.clone())?;
if let Some(remote) = &mut remote {
changes::insert(remote, id, row)?;
}
}
let mut records = WarpTable::<Retained<Row>>::try_new(limits.operations)
.map_err(|_| Error::Allocation)?;
let mut order = VecDeque::new();
order
.try_reserve_exact(limits.operations.value().get() as usize)
.map_err(|_| Error::Allocation)?;
let mut previous = Revision(0);
for record in recovery.outcomes {
super::replay::validate(&record, limits)?;
if record.revision <= previous
|| record.revision > recovery.revision
|| record.operation.history != recovery.history
|| (record.operation.epoch >= epoch && recovery.reserved_through != u64::MAX)
{
return Err(Error::InvalidRecovery);
}
previous = record.revision;
let operation = record.operation;
match records.insert_with_id(operation, Retained { record }) {
super::super::WarpTableInsertWithIdOutcome::Inserted { .. } => {
order.push_back(operation)
}
super::super::WarpTableInsertWithIdOutcome::AlreadyPresent { .. }
| super::super::WarpTableInsertWithIdOutcome::Full { .. }
| super::super::WarpTableInsertWithIdOutcome::UniqueConflict { .. } => {
return Err(Error::InvalidRecovery);
}
}
}
let mut undo = Vec::new();
undo.try_reserve_exact(limits.changes)
.map_err(|_| Error::Allocation)?;
let source = SourceId::persistent(
recovery.history.0,
NonZeroU64::new(epoch).ok_or(Error::InvalidRecovery)?,
);
let observe = |table: WarpTable<Row>| table.observe(source).map_err(|_| Error::Allocation);
let accepted = observe(accepted)?;
let local = observe(local)?;
let remote = remote.map(observe).transpose()?;
let mut incarnation_scratch = Vec::new();
incarnation_scratch
.try_reserve_exact(limits.changes)
.map_err(|_| Error::Allocation)?;
Ok(Self {
history: recovery.history,
epoch,
clock: RevisionClock::restart(recovery.reserved_through, reserved_through)?,
row_clock: RevisionClock::restart(recovery.reserved_through, reserved_through)?,
next_session: 0,
next_attempt: 0,
limits,
replication,
accepted,
staging,
local,
remote,
accepted_revision: recovery.revision,
local_revision: recovery.revision,
remote_revision: Revision(0),
restored_revision: recovery.revision,
remote_ready: false,
records,
order,
sessions: WarpTable::try_new(limits.sessions).map_err(|_| Error::Allocation)?,
undo,
incarnation_scratch,
local_attempt: None,
remote_attempt: None,
poisoned: false,
observation_incarnation_valid: recovery.reserved_through != u64::MAX,
})
}
pub const fn history(&self) -> HistoryId {
self.history
}
pub const fn revision(&self) -> Revision {
self.accepted_revision
}
pub const fn local_revision(&self) -> Revision {
self.local_revision
}
pub const fn remote_revision(&self) -> Revision {
self.remote_revision
}
pub const fn reserved_through(&self) -> u64 {
self.clock.reserved_through()
}
pub const fn limits(&self) -> Limits {
self.limits
}
pub fn reservation_required(&self) -> bool {
self.clock.available() == Err(Error::ReservationRequired)
|| self.row_clock.available() == Err(Error::ReservationRequired)
}
pub const fn epoch(&self) -> u64 {
self.epoch
}
pub fn open_session(&mut self) -> Result<SessionId, Error> {
if self.poisoned {
return Err(Error::Corrupt);
}
if self.clock.available() == Err(Error::Exhausted) {
return Err(Error::Exhausted);
}
let next = self.next_session.checked_add(1).ok_or(Error::Exhausted)?;
let id = SessionId(next);
match self.sessions.insert_with_id(id, Session { sequence: 0 }) {
super::super::WarpTableInsertWithIdOutcome::Inserted { .. } => {
self.next_session = next;
Ok(id)
}
super::super::WarpTableInsertWithIdOutcome::Full { .. } => Err(Error::Full),
super::super::WarpTableInsertWithIdOutcome::AlreadyPresent { .. }
| super::super::WarpTableInsertWithIdOutcome::UniqueConflict { .. } => {
Err(Error::Corrupt)
}
}
}
pub fn operation(&self, session: SessionId, sequence: u64) -> OperationId {
OperationId {
history: self.history,
epoch: self.epoch,
session,
sequence,
}
}
pub fn next_operation(&self, session: SessionId) -> Result<OperationId, Error> {
match self.sessions.get(session) {
WarpTableGetOutcome::Present { row } => Ok(self.operation(
session,
row.sequence.checked_add(1).ok_or(Error::Exhausted)?,
)),
WarpTableGetOutcome::Absent { .. } => Err(Error::UnknownSession),
}
}
pub fn close_session(&mut self, session: SessionId) -> Result<(), Error> {
match self.sessions.remove(session) {
WarpTableRemoveOutcome::Removed { .. } => Ok(()),
WarpTableRemoveOutcome::Absent { .. } => Err(Error::UnknownSession),
}
}
pub fn reserved(&mut self, through: u64) -> Result<(), Error> {
self.clock.reserve(through)?;
self.row_clock.reserve(through)?;
Ok(())
}
pub fn record(&self, operation: OperationId) -> Option<&Record<Row, Row::Id>> {
match self.records.get(operation) {
WarpTableGetOutcome::Present { row } => Some(row.record),
WarpTableGetOutcome::Absent { .. } => None,
}
}
pub fn inquire(&self, operation: OperationId, required: Guarantee) -> Confirmation {
if self.poisoned {
return Confirmation::Unavailable(Error::Corrupt);
}
if required == Guarantee::ReplicatedDurable && self.replication == Replication::Disabled {
return Confirmation::Unsupported;
}
let Some(record) = self.record(operation) else {
return Confirmation::Unknown;
};
let established = if self.remote_revision >= record.revision
&& self.replication == Replication::Required
{
Guarantee::ReplicatedDurable
} else if self.local_revision >= record.revision {
Guarantee::LocallyDurable
} else {
Guarantee::Accepted
};
let met = match required {
Guarantee::Accepted => true,
Guarantee::LocallyDurable => established != Guarantee::Accepted,
Guarantee::ReplicatedDurable => established == Guarantee::ReplicatedDurable,
};
if met {
Confirmation::Confirmed {
revision: record.revision,
established,
}
} else {
Confirmation::Pending {
revision: record.revision,
established,
}
}
}
pub fn retire(&mut self, operation: OperationId) -> Result<Record<Row, Row::Id>, Error> {
let required = match self.replication {
Replication::Disabled => Guarantee::LocallyDurable,
Replication::Required => Guarantee::ReplicatedDurable,
};
if !matches!(
self.inquire(operation, required),
Confirmation::Confirmed { .. }
) {
return Err(Error::Busy);
}
match self.records.remove(operation) {
WarpTableRemoveOutcome::Removed { removed, .. } => {
self.order.retain(|id| *id != operation);
Ok(removed.record)
}
WarpTableRemoveOutcome::Absent { .. } => Err(Error::ExpiredOperation),
}
}
pub fn prepare(&mut self, guarantee: Guarantee) -> Result<Option<Attempt>, Error> {
if self.poisoned {
return Err(Error::Corrupt);
}
let (current, through, busy) = match guarantee {
Guarantee::Accepted => return Err(Error::Unsupported),
Guarantee::LocallyDurable => (
self.local_revision,
self.accepted_revision,
self.local_attempt.is_some(),
),
Guarantee::ReplicatedDurable => {
if self.replication == Replication::Disabled {
return Err(Error::Unsupported);
}
(
self.remote_revision,
self.local_revision,
self.remote_attempt.is_some(),
)
}
};
if busy {
return Err(Error::Busy);
}
if current == through && (guarantee != Guarantee::ReplicatedDurable || self.remote_ready) {
return Ok(None);
}
let sequence = self.next_attempt.checked_add(1).ok_or(Error::Exhausted)?;
let attempt = Attempt {
history: self.history,
epoch: self.epoch,
sequence,
guarantee,
through,
};
self.next_attempt = sequence;
match guarantee {
Guarantee::LocallyDurable => self.local_attempt = Some(attempt),
Guarantee::ReplicatedDurable => self.remote_attempt = Some(attempt),
Guarantee::Accepted => return Err(Error::Unsupported),
}
Ok(Some(attempt))
}
pub fn batch(
&self,
attempt: Attempt,
) -> Result<impl Iterator<Item = &Record<Row, Row::Id>>, Error> {
let (pending, after) = match attempt.guarantee {
Guarantee::LocallyDurable => (self.local_attempt, self.local_revision),
Guarantee::ReplicatedDurable => (self.remote_attempt, self.remote_revision),
Guarantee::Accepted => return Err(Error::Unsupported),
};
if pending != Some(attempt) {
return Err(Error::StaleCompletion);
}
Ok(self
.order
.iter()
.filter_map(|id| self.record(*id))
.filter(move |record| record.revision > after && record.revision <= attempt.through))
}
pub fn checkpoint(&self) -> Result<Recovery<Row, Row::Id>, Error> {
if self.poisoned {
return Err(Error::Corrupt);
}
let mut rows = Vec::new();
rows.try_reserve_exact(self.local.table().len())
.map_err(|_| Error::Allocation)?;
for id in self.local.table().ids() {
match self.local.table().get(*id) {
WarpTableGetOutcome::Present { row } => rows.push((*id, Row::own(row))),
WarpTableGetOutcome::Absent { .. } => return Err(Error::Corrupt),
}
}
let mut outcomes = Vec::new();
outcomes
.try_reserve_exact(self.records.len())
.map_err(|_| Error::Allocation)?;
for id in &self.order {
let record = self.record(*id).ok_or(Error::Corrupt)?;
if record.revision <= self.local_revision {
outcomes.push(record.clone());
}
}
Ok(Recovery {
history: self.history,
revision: self.local_revision,
reserved_through: self.clock.reserved_through(),
rows,
outcomes,
})
}
}
pub struct Submit<Row: PersistentRow> {
pub receipt: Option<CommandReceipt>,
pub operation: OperationId,
pub expected: Revision,
pub guarantee: Guarantee,
pub changes: Vec<Change<Row, Row::Id>>,
}
impl<Row: PersistentRow> StepInputOf<Authority<Row>> for Submit<Row> {
type Outcome = Result<Confirmation, (Self, Error)>;
fn step(self, machine: &mut Authority<Row>) -> Self::Outcome {
match machine.submit(&self) {
Ok(Some(confirmation)) => Ok(confirmation),
Err(error) => Err((self, error)),
Ok(None) => {
let revision = machine.accepted_revision;
let operation = self.operation;
let guarantee = self.guarantee;
let record = Record {
receipt: self.receipt,
operation,
revision,
expected: self.expected,
changes: self.changes,
};
match machine
.records
.insert_with_id(operation, Retained { record })
{
super::super::WarpTableInsertWithIdOutcome::Inserted { .. } => {}
super::super::WarpTableInsertWithIdOutcome::AlreadyPresent {
rejected, ..
}
| super::super::WarpTableInsertWithIdOutcome::Full { rejected, .. }
| super::super::WarpTableInsertWithIdOutcome::UniqueConflict {
rejected, ..
} => {
machine.poisoned = true;
return Err((
Submit {
receipt: rejected.record.receipt,
operation,
expected: rejected.record.expected,
guarantee,
changes: rejected.record.changes,
},
Error::Corrupt,
));
}
}
machine.order.push_back(operation);
machine.poisoned = false;
Ok(machine.inquire(operation, guarantee))
}
}
}
}
impl<Row: PersistentRow> Authority<Row> {
pub fn admit_command(&self, operation: OperationId, guarantee: Guarantee) -> Result<(), Error> {
if self.poisoned {
return Err(Error::Corrupt);
}
if guarantee == Guarantee::ReplicatedDurable && self.replication == Replication::Disabled {
return Err(Error::Unsupported);
}
if operation.history != self.history {
return Err(Error::InvalidRecovery);
}
if operation.epoch != self.epoch {
return Err(Error::ExpiredOperation);
}
let sequence = match self.sessions.get(operation.session) {
WarpTableGetOutcome::Present { row } => *row.sequence,
WarpTableGetOutcome::Absent { .. } => return Err(Error::UnknownSession),
};
if operation.sequence <= sequence {
return Err(Error::ExpiredOperation);
}
if self.records.len() == self.limits.operations.value().get() as usize {
return Err(Error::Full);
}
self.clock.available()?;
Ok(())
}
fn submit(&mut self, request: &Submit<Row>) -> Result<Option<Confirmation>, Error> {
if self.poisoned {
return Err(Error::Corrupt);
}
if request.guarantee == Guarantee::ReplicatedDurable
&& self.replication == Replication::Disabled
{
return Err(Error::Unsupported);
}
if let Some(record) = self.record(request.operation) {
if record.expected != request.expected
|| record.changes != request.changes
|| record.receipt != request.receipt
{
return Err(Error::ConflictingRetry);
}
return Ok(Some(self.inquire(request.operation, request.guarantee)));
}
self.admit_command(request.operation, request.guarantee)?;
if request.expected != self.accepted_revision {
return Err(Error::StaleRevision);
}
if let Some(receipt) = &request.receipt
&& receipt.retained_bytes().ok_or(Error::InvalidBounds)? > self.limits.receipt_bytes
{
return Err(Error::InvalidBounds);
}
if (request.changes.is_empty() && request.receipt.is_none())
|| request.changes.capacity() > self.limits.changes
{
return Err(Error::InvalidBounds);
}
for change in &request.changes {
match change {
Change::Insert { row, .. } | Change::Replace { row, .. }
if row.retained_bytes() > self.limits.row_bytes =>
{
return Err(Error::InvalidBounds);
}
Change::Insert { .. } | Change::Replace { .. } | Change::Remove { .. } => {}
}
}
let revision = self.clock.available()?;
self.accepted
.sequence
.checked_add(u64::try_from(request.changes.len()).map_err(|_| Error::Exhausted)?)
.ok_or(Error::Exhausted)?;
self.poisoned = true;
if let Err(error) = changes::apply(&mut self.staging, &request.changes, &mut self.undo) {
if error != Error::Corrupt {
self.poisoned = false;
}
return Err(error);
}
if changes::apply_observed(
&mut self.accepted,
&request.changes,
&mut self.incarnation_scratch,
)
.is_err()
{
self.poisoned = true;
return Err(Error::Corrupt);
}
match self.sessions.update(request.operation.session, |mut row| {
*row.sequence_mut() = request.operation.sequence
}) {
WarpTableUpdateOutcome::Updated { .. } => {}
WarpTableUpdateOutcome::Absent { .. } => {
self.poisoned = true;
return Err(Error::Corrupt);
}
}
self.accepted_revision = Revision(revision);
let issued = self.clock.issue()?;
if issued != revision {
self.poisoned = true;
return Err(Error::Corrupt);
}
Ok(None)
}
}
pub enum ReadOutcome<'a, Row: PersistentRow> {
Ready {
history: HistoryId,
revision: Revision,
guarantee: Guarantee,
table: &'a WarpTable<Row>,
},
Pending,
Refused(Error),
}
impl<Row: PersistentRow> Authority<Row> {
pub fn read(&self, request: ReadRequest) -> ReadOutcome<'_, Row> {
if self.poisoned {
return ReadOutcome::Refused(Error::Corrupt);
}
let (revision, table) = match request.guarantee {
Guarantee::Accepted => (self.accepted_revision, self.accepted.table()),
Guarantee::LocallyDurable => (self.local_revision, self.local.table()),
Guarantee::ReplicatedDurable => {
if self.replication == Replication::Disabled {
return ReadOutcome::Refused(Error::Unsupported);
}
if !self.remote_ready {
return ReadOutcome::Pending;
}
let Some(remote) = &self.remote else {
return ReadOutcome::Refused(Error::Corrupt);
};
(self.remote_revision, remote.table())
}
};
ReadOutcome::Ready {
history: self.history,
revision,
guarantee: request.guarantee,
table,
}
}
pub fn observation(&self, guarantee: Guarantee) -> Result<&ObservedTable<Row>, Error> {
if !self.observation_incarnation_valid {
return Err(Error::Exhausted);
}
match self.read(ReadRequest { guarantee }) {
ReadOutcome::Ready { .. } => {}
ReadOutcome::Pending => return Err(Error::Busy),
ReadOutcome::Refused(error) => return Err(error),
}
match guarantee {
Guarantee::Accepted => Ok(&self.accepted),
Guarantee::LocallyDurable => Ok(&self.local),
Guarantee::ReplicatedDurable => self.remote.as_ref().ok_or(Error::Unsupported),
}
}
}
pub struct Complete {
pub attempt: Attempt,
pub result: PhysicalResult,
}
impl<Row: PersistentRow> StepInputOf<Authority<Row>> for Complete {
type Outcome = Result<(), Error>;
fn step(self, machine: &mut Authority<Row>) -> Self::Outcome {
if machine.poisoned {
return Err(Error::Corrupt);
}
let attempt = self.attempt;
let (pending, after) = match attempt.guarantee {
Guarantee::Accepted => return Err(Error::Unsupported),
Guarantee::LocallyDurable => (&mut machine.local_attempt, machine.local_revision),
Guarantee::ReplicatedDurable => (&mut machine.remote_attempt, machine.remote_revision),
};
if *pending != Some(attempt) {
return Err(Error::StaleCompletion);
}
*pending = None;
if self.result == PhysicalResult::Unresolved {
return Ok(());
}
machine.poisoned = true;
let table = match attempt.guarantee {
Guarantee::LocallyDurable => &mut machine.local,
Guarantee::ReplicatedDurable => machine.remote.as_mut().ok_or(Error::Unsupported)?,
Guarantee::Accepted => return Err(Error::Unsupported),
};
for id in &machine.order {
let record = match machine.records.get(*id) {
WarpTableGetOutcome::Present { row } => row.record,
WarpTableGetOutcome::Absent { .. } => {
machine.poisoned = true;
return Err(Error::Corrupt);
}
};
if record.revision > after
&& record.revision <= attempt.through
&& record.revision > machine.restored_revision
&& changes::apply_observed(table, &record.changes, &mut machine.incarnation_scratch)
.is_err()
{
machine.poisoned = true;
return Err(Error::Corrupt);
}
}
match attempt.guarantee {
Guarantee::LocallyDurable => machine.local_revision = attempt.through,
Guarantee::ReplicatedDurable => {
machine.remote_revision = attempt.through;
machine.remote_ready = true;
}
Guarantee::Accepted => return Err(Error::Unsupported),
}
machine.poisoned = false;
Ok(())
}
}
impl<Row: PersistentRow<Id = super::super::WarpId<Row>>> Authority<Row> {
pub fn allocate_id(&mut self) -> Result<super::super::WarpId<Row>, Error> {
if self.poisoned {
return Err(Error::Corrupt);
}
let value = self.row_clock.issue()?;
let value = core::num::NonZeroU64::new(value).ok_or(Error::Corrupt)?;
Ok(super::super::WarpId::new(value))
}
}