use super::super::{
WarpId, WarpTable, WarpTableInsertOutcome, WarpTableInsertWithIdOutcome,
WarpTableRemoveOutcome, WarpTableReplaceOutcome, WarpTableRow, WarpTableUpdateOutcome,
};
use alloc::{
collections::{TryReserveError, VecDeque},
vec::Vec,
};
use core::num::NonZeroU64;
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub enum SourceId {
Memory(NonZeroU64),
Persistent {
history: [u8; 16],
incarnation: NonZeroU64,
},
}
impl SourceId {
pub const fn new(identity: NonZeroU64) -> Self {
Self::Memory(identity)
}
pub const fn persistent(history: [u8; 16], incarnation: NonZeroU64) -> Self {
Self::Persistent {
history,
incarnation,
}
}
}
#[derive(Debug)]
pub enum ObservationCreationError {
StorageUnavailable(TryReserveError),
SequenceExhausted,
}
#[derive(Debug)]
pub enum ObservationEditError<Input> {
SequenceExhausted { rejected: Input },
}
#[derive(Clone, Copy)]
pub(super) struct Change<Id> {
pub sequence: u64,
pub id: Id,
}
pub struct ObservedTable<Row: WarpTableRow, Brand = Row> {
pub(crate) table: WarpTable<Row, Brand>,
pub(super) source: SourceId,
pub(crate) sequence: u64,
pub(crate) incarnations: Vec<u64>,
pub(super) history: VecDeque<Change<Row::Id>>,
pub(super) history_capacity: usize,
#[cfg(feature = "secondary-indexes")]
selection_ids: Vec<Row::Id>,
pub(super) discarded_through: u64,
}
impl<Row: WarpTableRow, Brand> WarpTable<Row, Brand> {
pub fn observe(
self,
source: SourceId,
) -> Result<ObservedTable<Row, Brand>, (Self, ObservationCreationError)> {
let mut incarnations = Vec::new();
if let Err(error) = incarnations.try_reserve_exact(self.maximum_rows) {
return Err((self, ObservationCreationError::StorageUnavailable(error)));
}
let mut history = VecDeque::new();
if let Err(error) = history.try_reserve_exact(self.maximum_rows) {
return Err((self, ObservationCreationError::StorageUnavailable(error)));
}
#[cfg(feature = "secondary-indexes")]
let mut selection_ids = Vec::new();
#[cfg(feature = "secondary-indexes")]
if let Err(error) = selection_ids.try_reserve_exact(self.maximum_rows) {
return Err((self, ObservationCreationError::StorageUnavailable(error)));
}
let mut sequence = 0_u64;
for (next, _) in (1_u64..).zip(self.ids()) {
sequence = next;
incarnations.push(sequence);
}
let history_capacity = self.maximum_rows;
Ok(ObservedTable {
table: self,
source,
sequence,
incarnations,
history,
history_capacity,
discarded_through: sequence,
#[cfg(feature = "secondary-indexes")]
selection_ids,
})
}
}
impl<Row: WarpTableRow, Brand> ObservedTable<Row, Brand> {
pub const fn table(&self) -> &WarpTable<Row, Brand> {
&self.table
}
pub const fn source(&self) -> SourceId {
self.source
}
pub fn retained_changes(&self) -> usize {
self.history.len()
}
pub(crate) fn record(&mut self, id: Row::Id, sequence: u64) {
if self.history.len() == self.history_capacity
&& let Some(discarded) = self.history.pop_front()
{
self.discarded_through = discarded.sequence;
}
self.sequence = sequence;
self.history.push_back(Change { sequence, id });
}
pub fn insert_with_id(
&mut self,
id: Row::Id,
row: Row,
) -> Result<WarpTableInsertWithIdOutcome<Row::Id, Row>, ObservationEditError<Row>> {
let Some(sequence) = self.sequence.checked_add(1) else {
return Err(ObservationEditError::SequenceExhausted { rejected: row });
};
let outcome = self.table.insert_with_id(id, row);
match &outcome {
WarpTableInsertWithIdOutcome::Inserted { .. } => {
self.incarnations.push(sequence);
self.record(id, sequence);
}
WarpTableInsertWithIdOutcome::AlreadyPresent { .. }
| WarpTableInsertWithIdOutcome::Full { .. }
| WarpTableInsertWithIdOutcome::UniqueConflict { .. } => {}
}
Ok(outcome)
}
pub fn update<Output, Update>(
&mut self,
id: Row::Id,
update: Update,
) -> Result<WarpTableUpdateOutcome<Row::Id, Output>, ObservationEditError<Update>>
where
Update: for<'row> FnOnce(Row::Mut<'row>) -> Output,
{
let Some(sequence) = self.sequence.checked_add(1) else {
return Err(ObservationEditError::SequenceExhausted { rejected: update });
};
if let super::super::WarpTableGetOutcome::Absent { .. } = self.table.get(id) {
return Ok(WarpTableUpdateOutcome::Absent { id });
}
self.record(id, sequence);
Ok(self.table.update(id, update))
}
pub fn replace(
&mut self,
id: Row::Id,
row: Row,
) -> Result<WarpTableReplaceOutcome<Row::Id, Row>, ObservationEditError<Row>> {
let Some(sequence) = self.sequence.checked_add(1) else {
return Err(ObservationEditError::SequenceExhausted { rejected: row });
};
let outcome = self.table.replace(id, row);
match &outcome {
WarpTableReplaceOutcome::Replaced { .. } => self.record(id, sequence),
WarpTableReplaceOutcome::Absent { .. }
| WarpTableReplaceOutcome::UniqueConflict { .. } => {}
}
Ok(outcome)
}
pub fn upsert(
&mut self,
id: Row::Id,
row: Row,
) -> Result<super::super::WarpTableUpsertOutcome<Row::Id, Row>, ObservationEditError<Row>> {
let Some(sequence) = self.sequence.checked_add(1) else {
return Err(ObservationEditError::SequenceExhausted { rejected: row });
};
let outcome = self.table.upsert(id, row);
match &outcome {
super::super::WarpTableUpsertOutcome::Inserted { .. } => {
self.incarnations.push(sequence);
self.record(id, sequence);
}
super::super::WarpTableUpsertOutcome::Replaced { .. } => self.record(id, sequence),
super::super::WarpTableUpsertOutcome::Full { .. }
| super::super::WarpTableUpsertOutcome::UniqueConflict { .. } => {}
}
Ok(outcome)
}
#[cfg(feature = "secondary-indexes")]
pub fn update_selected<Select, Update>(
&mut self,
select: Select,
mut update: Update,
) -> Result<usize, ObservationSelectionError>
where
Select: for<'table> FnOnce(
&'table WarpTable<Row, Brand>,
) -> super::super::Selection<'table, Row, Brand>,
Update: for<'row> FnMut(Row::Mut<'row>),
Row::Indexes: super::super::secondary_index::StableSlotResolver,
{
self.selection_ids.clear();
let selection = select(&self.table);
if !selection.belongs_to(&self.table) {
return Err(ObservationSelectionError::ForeignTable);
}
self.selection_ids.extend(selection.ids());
let count = self.selection_ids.len();
let increments =
u64::try_from(count).map_err(|_| ObservationSelectionError::SequenceExhausted)?;
let through = self
.sequence
.checked_add(increments)
.ok_or(ObservationSelectionError::SequenceExhausted)?;
for (index, sequence) in (self.sequence..through).enumerate() {
let id = *super::super::get::stored_id(&self.selection_ids, index);
self.record(id, sequence.saturating_add(1));
match self.table.update(id, &mut update) {
WarpTableUpdateOutcome::Updated { .. } => {}
WarpTableUpdateOutcome::Absent { .. } => {
super::super::inconsistency::invariant_violation("selected row disappeared")
}
}
}
self.selection_ids.clear();
Ok(count)
}
pub fn remove(
&mut self,
id: Row::Id,
) -> Result<WarpTableRemoveOutcome<Row::Id, Row>, ObservationEditError<Row::Id>> {
let Some(sequence) = self.sequence.checked_add(1) else {
return Err(ObservationEditError::SequenceExhausted { rejected: id });
};
let dense_row = match self.table.primary_index.find(id, &self.table.ids) {
Ok(super::super::dense_key_index::FindOutcome::Present { location, .. }) => {
location.row
}
Ok(super::super::dense_key_index::FindOutcome::Absent) => {
return Ok(WarpTableRemoveOutcome::Absent { id });
}
Err(reason) => super::super::inconsistency::invariant_violation(reason),
};
let outcome = self.table.remove(id);
self.incarnations.swap_remove(dense_row);
self.record(id, sequence);
Ok(outcome)
}
}
impl<Row, IdRow, Brand> ObservedTable<Row, Brand>
where
Row: WarpTableRow<Id = WarpId<IdRow>>,
{
pub fn insert(
&mut self,
row: Row,
) -> Result<WarpTableInsertOutcome<WarpId<IdRow>, Row>, ObservationEditError<Row>> {
let Some(sequence) = self.sequence.checked_add(1) else {
return Err(ObservationEditError::SequenceExhausted { rejected: row });
};
let outcome = self.table.insert(row);
match &outcome {
WarpTableInsertOutcome::Inserted { id, .. } => {
self.incarnations.push(sequence);
self.record(*id, sequence);
}
WarpTableInsertOutcome::Full { .. }
| WarpTableInsertOutcome::IdentifiersExhausted { .. }
| WarpTableInsertOutcome::UniqueConflict { .. } => {}
}
Ok(outcome)
}
}
#[cfg(feature = "secondary-indexes")]
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ObservationSelectionError {
ForeignTable,
SequenceExhausted,
}