use super::super::{WarpTable, WarpTableGetOutcome, WarpTableInsertWithIdOutcome, WarpTableRow};
use super::*;
use alloc::vec::Vec;
#[derive(WarpTableRow)]
#[index(primary = OperationId)]
struct Saved<Row: PersistentRow> {
record: Record<Row, Row::Id>,
}
pub struct RecoveryBuilder<Row: PersistentRow> {
history: HistoryId,
revision: Revision,
reserved_through: u64,
table: WarpTable<Row>,
outcomes: WarpTable<Saved<Row>>,
order: Vec<OperationId>,
undo: Vec<(Row::Id, Row)>,
limits: Limits,
}
impl<Row: PersistentRow> RecoveryBuilder<Row> {
pub fn new(recovery: Recovery<Row, Row::Id>, limits: Limits) -> Result<Self, Error> {
if limits.changes == 0 || limits.row_bytes == 0 {
return Err(Error::InvalidBounds);
}
if recovery.revision.0 > recovery.reserved_through
|| recovery.outcomes.len() > limits.operations.value().get() as usize
{
return Err(Error::InvalidRecovery);
}
let mut table = WarpTable::try_new(limits.rows).map_err(|_| Error::Allocation)?;
for (id, row) in recovery.rows {
if row.retained_bytes() > limits.row_bytes {
return Err(Error::InvalidBounds);
}
changes::insert(&mut table, id, row)?;
}
let mut outcomes = WarpTable::try_new(limits.operations).map_err(|_| Error::Allocation)?;
let mut order = Vec::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 {
if record.operation.history != recovery.history
|| record.revision <= previous
|| record.revision > recovery.revision
{
return Err(Error::InvalidRecovery);
}
validate(&record, limits)?;
previous = record.revision;
let operation = record.operation;
match outcomes.insert_with_id(operation, Saved { record }) {
WarpTableInsertWithIdOutcome::Inserted { .. } => order.push(operation),
WarpTableInsertWithIdOutcome::AlreadyPresent { .. }
| WarpTableInsertWithIdOutcome::Full { .. }
| WarpTableInsertWithIdOutcome::UniqueConflict { .. } => {
return Err(Error::InvalidRecovery);
}
}
}
let mut undo = Vec::new();
undo.try_reserve_exact(limits.changes)
.map_err(|_| Error::Allocation)?;
Ok(Self {
history: recovery.history,
revision: recovery.revision,
reserved_through: recovery.reserved_through,
table,
outcomes,
order,
undo,
limits,
})
}
pub const fn history(&self) -> HistoryId {
self.history
}
pub const fn revision(&self) -> Revision {
self.revision
}
pub const fn reserved_through(&self) -> u64 {
self.reserved_through
}
pub fn reserve(&mut self, through: u64) -> Result<(), Error> {
if through < self.reserved_through {
return Err(Error::InvalidRecovery);
}
self.reserved_through = through;
Ok(())
}
pub fn rows_match(&self, rows: &[(Row::Id, Row)]) -> bool {
self.table.len() == rows.len()
&& rows.iter().all(|(id, value)| match self.table.get(*id) {
WarpTableGetOutcome::Present { row } => Row::own(row) == *value,
WarpTableGetOutcome::Absent { .. } => false,
})
}
pub fn apply(&mut self, record: Record<Row, Row::Id>) -> Result<(), Error> {
validate(&record, self.limits)?;
if record.operation.history != self.history {
return Err(Error::InvalidRecovery);
}
if let WarpTableGetOutcome::Present { row } = self.outcomes.get(record.operation) {
return if *row.record == record {
Ok(())
} else {
Err(Error::ConflictingRetry)
};
}
if record.expected != self.revision
|| record.revision <= self.revision
|| record.revision.0 > self.reserved_through
{
return Err(Error::InvalidRecovery);
}
if self.outcomes.len() == self.limits.operations.value().get() as usize {
return Err(Error::Full);
}
changes::apply(&mut self.table, &record.changes, &mut self.undo)?;
self.revision = record.revision;
let operation = record.operation;
match self.outcomes.insert_with_id(operation, Saved { record }) {
WarpTableInsertWithIdOutcome::Inserted { .. } => self.order.push(operation),
WarpTableInsertWithIdOutcome::AlreadyPresent { .. }
| WarpTableInsertWithIdOutcome::Full { .. }
| WarpTableInsertWithIdOutcome::UniqueConflict { .. } => return Err(Error::Corrupt),
}
Ok(())
}
pub fn snapshot(&self) -> Result<Recovery<Row, Row::Id>, Error> {
let mut rows = Vec::new();
rows.try_reserve_exact(self.table.len())
.map_err(|_| Error::Allocation)?;
for id in self.table.ids() {
match self.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.outcomes.len())
.map_err(|_| Error::Allocation)?;
for operation in &self.order {
match self.outcomes.get(*operation) {
WarpTableGetOutcome::Present { row } => outcomes.push(row.record.clone()),
WarpTableGetOutcome::Absent { .. } => return Err(Error::Corrupt),
}
}
Ok(Recovery {
history: self.history,
revision: self.revision,
reserved_through: self.reserved_through,
rows,
outcomes,
})
}
}
pub(super) fn validate<Row: PersistentRow>(
record: &Record<Row, Row::Id>,
limits: Limits,
) -> Result<(), Error> {
if record.operation.sequence == 0
|| record.operation.session.0 == 0
|| record.operation.epoch == 0
|| record.operation.epoch > record.revision.0
|| record.expected >= record.revision
|| (record.changes.is_empty() && record.receipt.is_none())
|| record.changes.len() > limits.changes
{
return Err(Error::InvalidRecovery);
}
if let Some(receipt) = &record.receipt
&& receipt.retained_bytes().ok_or(Error::InvalidBounds)? > limits.receipt_bytes
{
return Err(Error::InvalidBounds);
}
for (index, change) in record.changes.iter().enumerate() {
if record
.changes
.iter()
.take(index)
.any(|other| other.id() == change.id())
{
return Err(Error::DuplicateKey);
}
match change {
Change::Insert { row, .. } | Change::Replace { row, .. }
if row.retained_bytes() > limits.row_bytes =>
{
return Err(Error::InvalidBounds);
}
Change::Insert { .. } | Change::Replace { .. } | Change::Remove { .. } => {}
}
}
Ok(())
}