use super::super::{Capacity, WarpTableGetOutcome, WarpTableRow};
use super::*;
use crate::StateMachine;
use alloc::vec;
use core::num::NonZeroU32;
mod consumer;
mod continuous;
#[test]
fn mutation_request_widening_preserves_receipt_guarantee_and_owned_command() {
let receipt = Receipt::from_operation(OperationId {
history: HistoryId([3; 16]),
epoch: 7,
session: SessionId(11),
sequence: 19,
});
for guarantee in [
Guarantee::Accepted,
Guarantee::LocallyDurable,
Guarantee::ReplicatedDurable,
] {
let command = alloc::boxed::Box::new(23);
let original = core::ptr::from_ref(command.as_ref());
let request = TableMutation {
receipt,
command,
guarantee,
}
.into_request::<()>();
match request {
TableRequest::Mutate {
receipt: returned,
command,
guarantee: requested,
} => {
assert_eq!(returned, receipt);
assert_eq!(requested, guarantee);
assert_eq!(*command, 23);
assert_eq!(core::ptr::from_ref(command.as_ref()), original);
}
TableRequest::Read { .. }
| TableRequest::Inquire { .. }
| TableRequest::Retire { .. } => {
assert!(matches!(request, TableRequest::Mutate { .. }));
}
}
}
}
#[derive(Clone, Debug, Eq, PartialEq, WarpTableRow)]
#[index(primary = u32)]
struct Stock {
quantity: u32,
}
impl PersistentRow for Stock {
fn own(row: StockRef<'_>) -> Self {
Self {
quantity: *row.quantity,
}
}
fn retained_bytes(&self) -> usize {
core::mem::size_of::<Self>()
}
}
fn limits() -> Limits {
Limits {
receipt_bytes: 1024,
rows: Capacity::new(NonZeroU32::MIN.saturating_add(3)),
operations: Capacity::new(NonZeroU32::MIN.saturating_add(7)),
sessions: Capacity::new(NonZeroU32::MIN.saturating_add(1)),
changes: 4,
row_bytes: 4,
}
}
fn fresh(replication: Replication) -> Result<Authority<Stock>, Error> {
Authority::recover(
Recovery {
history: HistoryId([7; 16]),
revision: Revision(0),
reserved_through: 0,
rows: vec![],
outcomes: vec![],
},
16,
limits(),
replication,
)
}
#[test]
fn exhausted_recovery_remains_readable_but_cannot_reuse_a_view_incarnation() -> Result<(), Error> {
let authority = Authority::recover(
Recovery {
history: HistoryId([1; 16]),
revision: Revision(u64::MAX),
reserved_through: u64::MAX,
rows: vec![(1, Stock { quantity: 7 })],
outcomes: vec![],
},
u64::MAX,
limits(),
Replication::Disabled,
)?;
assert_eq!(quantity(&authority, Guarantee::LocallyDurable, 1)?, 7);
assert!(matches!(
authority.observation(Guarantee::LocallyDurable),
Err(Error::Exhausted)
));
Ok(())
}
fn submit(
machine: &mut Authority<Stock>,
op: OperationId,
changes: alloc::vec::Vec<Change<Stock, u32>>,
) -> Result<Confirmation, Error> {
machine
.step(Submit {
receipt: None,
operation: op,
expected: machine.revision(),
guarantee: Guarantee::ReplicatedDurable,
changes,
})
.map_err(|(_, e)| e)
}
fn flush(machine: &mut Authority<Stock>, guarantee: Guarantee) -> Result<(), Error> {
if let Some(attempt) = machine.prepare(guarantee)? {
machine.step(Complete {
attempt,
result: PhysicalResult::Durable,
})?;
}
Ok(())
}
fn quantity(machine: &Authority<Stock>, guarantee: Guarantee, id: u32) -> Result<u32, Error> {
match machine.read(ReadRequest { guarantee }) {
ReadOutcome::Ready { table, .. } => match table.get(id) {
WarpTableGetOutcome::Present { row } => Ok(*row.quantity),
WarpTableGetOutcome::Absent { .. } => Err(Error::Absent),
},
ReadOutcome::Pending => Err(Error::Busy),
ReadOutcome::Refused(e) => Err(e),
}
}
#[test]
fn removal_advances_each_durability_view_and_preserves_the_relocated_incarnation()
-> Result<(), Error> {
let mut authority = fresh(Replication::Required)?;
let session = authority.open_session()?;
let inserted = authority.next_operation(session)?;
submit(
&mut authority,
inserted,
vec![
Change::Insert {
id: 7,
row: Stock { quantity: 11 },
},
Change::Insert {
id: 9,
row: Stock { quantity: 13 },
},
],
)?;
flush(&mut authority, Guarantee::LocallyDurable)?;
flush(&mut authority, Guarantee::ReplicatedDurable)?;
let operation = authority.next_operation(session)?;
assert!(matches!(
submit(&mut authority, operation, vec![Change::Remove { id: 7 }])?,
Confirmation::Pending {
revision: Revision(2),
established: Guarantee::Accepted
}
));
assert_eq!(
quantity(&authority, Guarantee::Accepted, 7),
Err(Error::Absent)
);
assert_eq!(quantity(&authority, Guarantee::LocallyDurable, 7)?, 11);
assert_eq!(quantity(&authority, Guarantee::ReplicatedDurable, 7)?, 11);
for guarantee in [
Guarantee::Accepted,
Guarantee::LocallyDurable,
Guarantee::ReplicatedDurable,
] {
if guarantee != Guarantee::Accepted {
flush(&mut authority, guarantee)?;
}
assert_eq!(quantity(&authority, guarantee, 7), Err(Error::Absent));
assert_eq!(quantity(&authority, guarantee, 9)?, 13);
let observed = authority.observation(guarantee)?;
assert_eq!(observed.table().ids(), &[9]);
assert_eq!(observed.incarnations, [2]);
}
assert_eq!(
authority.inquire(operation, Guarantee::ReplicatedDurable),
Confirmation::Confirmed {
revision: Revision(2),
established: Guarantee::ReplicatedDurable
}
);
assert_eq!(
authority.retire(operation)?.changes,
[Change::Remove { id: 7 }]
);
Ok(())
}
#[test]
fn missing_observed_removal_cannot_consume_an_incarnation_or_revision() -> Result<(), Error> {
let table =
super::super::WarpTable::<Stock>::try_new(limits().rows).map_err(|_| Error::Allocation)?;
let mut observed = table
.observe(super::super::observed::SourceId::new(
core::num::NonZeroU64::MIN,
))
.map_err(|_| Error::Allocation)?;
assert_eq!(
changes::remove_observed_row(&mut observed, 7, 3),
Err(Error::Corrupt)
);
assert!(observed.table().is_empty());
assert!(observed.incarnations.is_empty());
assert_eq!(observed.sequence, 0);
assert_eq!(observed.retained_changes(), 0);
Ok(())
}
#[test]
fn visibility_and_confirmation_progress_independently() -> Result<(), Error> {
let mut machine = fresh(Replication::Required)?;
let session = machine.open_session()?;
let first = machine.operation(session, 1);
let changes = vec![
Change::Insert {
id: 1,
row: Stock { quantity: 10 },
},
Change::Insert {
id: 2,
row: Stock { quantity: 0 },
},
];
assert_eq!(
submit(&mut machine, first, changes)?,
Confirmation::Pending {
revision: Revision(1),
established: Guarantee::Accepted
}
);
assert_eq!(quantity(&machine, Guarantee::Accepted, 1)?, 10);
assert_eq!(
quantity(&machine, Guarantee::LocallyDurable, 1),
Err(Error::Absent)
);
assert_eq!(
quantity(&machine, Guarantee::ReplicatedDurable, 1),
Err(Error::Busy)
);
flush(&mut machine, Guarantee::LocallyDurable)?;
flush(&mut machine, Guarantee::ReplicatedDurable)?;
let second = machine.operation(session, 2);
submit(
&mut machine,
second,
vec![
Change::Replace {
id: 1,
row: Stock { quantity: 8 },
},
Change::Replace {
id: 2,
row: Stock { quantity: 2 },
},
],
)?;
assert_eq!(quantity(&machine, Guarantee::Accepted, 1)?, 8);
assert_eq!(quantity(&machine, Guarantee::Accepted, 2)?, 2);
assert_eq!(quantity(&machine, Guarantee::LocallyDurable, 1)?, 10);
flush(&mut machine, Guarantee::LocallyDurable)?;
assert_eq!(quantity(&machine, Guarantee::LocallyDurable, 2)?, 2);
assert_eq!(quantity(&machine, Guarantee::ReplicatedDurable, 2)?, 0);
flush(&mut machine, Guarantee::ReplicatedDurable)?;
assert_eq!(
machine.inquire(second, Guarantee::ReplicatedDurable),
Confirmation::Confirmed {
revision: Revision(2),
established: Guarantee::ReplicatedDurable
}
);
Ok(())
}
#[test]
fn refused_multirow_change_preserves_every_row() -> Result<(), Error> {
let mut machine = fresh(Replication::Required)?;
let session = machine.open_session()?;
let op = machine.operation(session, 1);
submit(
&mut machine,
op,
vec![Change::Insert {
id: 1,
row: Stock { quantity: 10 },
}],
)?;
let next = machine.operation(session, 2);
assert_eq!(
submit(
&mut machine,
next,
vec![
Change::Replace {
id: 1,
row: Stock { quantity: 8 }
},
Change::Replace {
id: 2,
row: Stock { quantity: 2 }
}
]
),
Err(Error::Absent)
);
assert_eq!(machine.revision(), Revision(1));
assert_eq!(quantity(&machine, Guarantee::Accepted, 1)?, 10);
Ok(())
}
#[test]
fn restart_skips_reserved_numbers_and_never_reexecutes_lost_operations() -> Result<(), Error> {
let mut machine = fresh(Replication::Required)?;
let session = machine.open_session()?;
let saved = machine.operation(session, 1);
submit(
&mut machine,
saved,
vec![Change::Insert {
id: 1,
row: Stock { quantity: 10 },
}],
)?;
flush(&mut machine, Guarantee::LocallyDurable)?;
let lost = machine.operation(session, 2);
submit(
&mut machine,
lost,
vec![Change::Replace {
id: 1,
row: Stock { quantity: 8 },
}],
)?;
let checkpoint = machine.checkpoint()?;
let mut recovered = Authority::recover(checkpoint, 32, limits(), Replication::Required)?;
assert_eq!(quantity(&recovered, Guarantee::Accepted, 1)?, 10);
assert_eq!(
recovered.inquire(lost, Guarantee::LocallyDurable),
Confirmation::Unknown
);
assert_eq!(
submit(
&mut recovered,
lost,
vec![Change::Replace {
id: 1,
row: Stock { quantity: 8 }
}]
),
Err(Error::ExpiredOperation)
);
let session = recovered.open_session()?;
let next = recovered.operation(session, 1);
submit(
&mut recovered,
next,
vec![Change::Replace {
id: 1,
row: Stock { quantity: 9 },
}],
)?;
assert_eq!(recovered.revision(), Revision(17));
flush(&mut recovered, Guarantee::LocallyDurable)?;
assert_eq!(
recovered.inquire(lost, Guarantee::LocallyDurable),
Confirmation::Unknown
);
flush(&mut recovered, Guarantee::ReplicatedDurable)?;
assert_eq!(quantity(&recovered, Guarantee::ReplicatedDurable, 1)?, 9);
assert_eq!(
recovered
.observation(Guarantee::ReplicatedDurable)?
.retained_changes(),
1
);
Ok(())
}
#[test]
fn retries_preserve_payload_and_stale_physical_completions_are_rejected() -> Result<(), Error> {
let mut machine = fresh(Replication::Required)?;
let session = machine.open_session()?;
let op = machine.operation(session, 1);
let changes = vec![Change::Insert {
id: 1,
row: Stock { quantity: 10 },
}];
submit(&mut machine, op, changes.clone())?;
let first = machine
.prepare(Guarantee::LocallyDurable)?
.ok_or(Error::Busy)?;
machine.step(Complete {
attempt: first,
result: PhysicalResult::Unresolved,
})?;
let retry = machine
.prepare(Guarantee::LocallyDurable)?
.ok_or(Error::Busy)?;
assert_eq!(
machine.step(Complete {
attempt: first,
result: PhysicalResult::Durable
}),
Err(Error::StaleCompletion)
);
assert_eq!(machine.batch(retry)?.count(), 1);
machine.step(Complete {
attempt: retry,
result: PhysicalResult::Durable,
})?;
assert!(matches!(
machine.step(Submit {
receipt: None,
operation: op,
expected: Revision(0),
guarantee: Guarantee::LocallyDurable,
changes: changes.clone()
}),
Ok(Confirmation::Confirmed {
revision: Revision(1),
..
})
));
assert_eq!(machine.revision(), Revision(1));
flush(&mut machine, Guarantee::ReplicatedDurable)?;
machine.retire(op)?;
let Err((_, error)) = machine.step(Submit {
receipt: None,
operation: op,
expected: Revision(0),
guarantee: Guarantee::Accepted,
changes,
}) else {
return Err(Error::Corrupt);
};
assert_eq!(error, Error::ExpiredOperation);
Ok(())
}
#[derive(Clone, Debug, Eq, PartialEq, WarpTableRow)]
#[index(primary=u32)]
struct Indexed {
#[index(unique)]
code: u32,
}
impl PersistentRow for Indexed {
fn own(row: IndexedRef<'_>) -> Self {
Self { code: *row.code }
}
fn retained_bytes(&self) -> usize {
4
}
}
#[test]
fn unique_value_swaps_are_atomic_and_conflicts_restore_staging() -> Result<(), Error> {
let mut table = Authority::<Indexed>::recover(
Recovery {
history: HistoryId([9; 16]),
revision: Revision(0),
reserved_through: 0,
rows: vec![(1, Indexed { code: 10 }), (2, Indexed { code: 20 })],
outcomes: vec![],
},
16,
limits(),
Replication::Disabled,
)?;
let session = table.open_session()?;
let op = table.operation(session, 1);
let rejected = table.step(Submit {
receipt: None,
operation: op,
expected: Revision(0),
guarantee: Guarantee::Accepted,
changes: vec![Change::Replace {
id: 1,
row: Indexed { code: 20 },
}],
});
assert!(matches!(rejected, Err((_, Error::UniqueConflict))));
let changes = vec![
Change::Replace {
id: 1,
row: Indexed { code: 20 },
},
Change::Replace {
id: 2,
row: Indexed { code: 10 },
},
];
assert!(matches!(
table.step(Submit {
receipt: None,
operation: op,
expected: Revision(0),
guarantee: Guarantee::Accepted,
changes
}),
Ok(Confirmation::Confirmed {
revision: Revision(1),
..
})
));
let ReadOutcome::Ready { table, .. } = table.read(ReadRequest {
guarantee: Guarantee::Accepted,
}) else {
return Err(Error::Corrupt);
};
let WarpTableGetOutcome::Present { row } = table.get(1) else {
return Err(Error::Absent);
};
assert_eq!(*row.code, 20);
let WarpTableGetOutcome::Present { row } = table.get(2) else {
return Err(Error::Absent);
};
assert_eq!(*row.code, 10);
Ok(())
}
#[test]
fn bounded_history_requires_explicit_retirement_and_reservation_never_wraps() -> Result<(), Error> {
let mut bounds = limits();
bounds.operations = Capacity::new(NonZeroU32::MIN);
let mut table = Authority::<Stock>::recover(
Recovery {
history: HistoryId([9; 16]),
revision: Revision(0),
reserved_through: 0,
rows: vec![],
outcomes: vec![],
},
1,
bounds,
Replication::Disabled,
)?;
let session = table.open_session()?;
let op = table.operation(session, 1);
let edit = |operation, expected, quantity| Submit {
receipt: None,
operation,
expected,
guarantee: Guarantee::Accepted,
changes: vec![Change::Insert {
id: quantity,
row: Stock { quantity },
}],
};
table.step(edit(op, Revision(0), 1)).map_err(|(_, e)| e)?;
let next = table.operation(session, 2);
assert!(matches!(
table.step(edit(next, Revision(1), 2)),
Err((_, Error::Full))
));
assert_eq!(table.retire(op), Err(Error::Busy));
flush(&mut table, Guarantee::LocallyDurable)?;
let _record = table.retire(op)?;
assert!(matches!(
table.step(edit(next, Revision(1), 2)),
Err((_, Error::ReservationRequired))
));
table.reserved(2)?;
table.step(edit(next, Revision(1), 2)).map_err(|(_, e)| e)?;
let mut clock =
RevisionClock::restart(u64::MAX.checked_sub(1).ok_or(Error::Exhausted)?, u64::MAX)?;
assert_eq!(clock.issue(), Ok(u64::MAX));
assert_eq!(clock.issue(), Err(Error::Exhausted));
assert_eq!(clock.reserve(u64::MAX), Err(Error::InvalidRecovery));
Ok(())
}
proptest::proptest! {
#[test]
fn reservation_sequences_are_unique_and_restart_abandons_every_previous_number(
previous in 0u64..u64::MAX, width in 1u64..1024, steps in 0usize..128,
) {
if let Some(end)=previous.checked_add(width) {
let mut clock=RevisionClock::restart(previous,end).map_err(|_|proptest::test_runner::TestCaseError::fail("clock"))?;
let mut last=previous;
for _ in 0..steps {
match clock.issue() {
Ok(next)=>{proptest::prop_assert!(next>last && next<=end);last=next;},
Err(error)=>proptest::prop_assert!(error==Error::Exhausted || error==Error::ReservationRequired),
}
}
if let Some(next_end)=end.checked_add(width) {
let mut restarted=RevisionClock::restart(end,next_end).map_err(|_|proptest::test_runner::TestCaseError::fail("restart"))?;
let next=restarted.issue().map_err(|_|proptest::test_runner::TestCaseError::fail("issue"))?;
proptest::prop_assert!(next>last && next>end);
}
}
}
}
#[test]
fn exhausted_history_reopens_read_only_without_consumer_revision_requirements() -> Result<(), Error>
{
let mut table = Authority::<Stock>::recover(
Recovery {
history: HistoryId([9; 16]),
revision: Revision(u64::MAX),
reserved_through: u64::MAX,
rows: vec![(1, Stock { quantity: 10 })],
outcomes: vec![],
},
u64::MAX,
limits(),
Replication::Disabled,
)?;
assert_eq!(quantity(&table, Guarantee::LocallyDurable, 1)?, 10);
assert_eq!(table.open_session(), Err(Error::Exhausted));
assert!(matches!(
table.read(ReadRequest {
guarantee: Guarantee::LocallyDurable,
}),
ReadOutcome::Ready {
revision: Revision(u64::MAX),
..
}
));
Ok(())
}
#[test]
fn clock_refusals_leave_the_reservation_and_next_number_unchanged() -> Result<(), Error> {
assert_eq!(RevisionClock::restart(5, 5), Err(Error::InvalidRecovery));
assert_eq!(RevisionClock::restart(5, 4), Err(Error::InvalidRecovery));
assert_eq!(
RevisionClock::restart(u64::MAX, 4),
Err(Error::InvalidRecovery)
);
let mut clock = RevisionClock::restart(5, 6)?;
assert_eq!(clock.reserved_through(), 6);
let before = clock;
assert_eq!(clock.reserve(6), Err(Error::InvalidRecovery));
assert_eq!(clock.reserve(5), Err(Error::InvalidRecovery));
assert_eq!(clock, before);
assert_eq!(clock.issue(), Ok(6));
assert_eq!(clock.available(), Err(Error::ReservationRequired));
clock.reserve(9)?;
assert_eq!(clock.reserved_through(), 9);
assert_eq!(clock.issue(), Ok(7));
Ok(())
}
#[derive(Clone, Eq, PartialEq, WarpTableRow)]
struct Generated {
value: u32,
}
impl PersistentRow for Generated {
fn own(row: GeneratedRef<'_>) -> Self {
Self { value: *row.value }
}
fn retained_bytes(&self) -> usize {
4
}
}
#[test]
fn generated_row_id_reservations_survive_abandoned_allocations_and_restart() -> Result<(), Error> {
let mut table = Authority::<Generated>::recover(
Recovery {
history: HistoryId([9; 16]),
revision: Revision(0),
reserved_through: 0,
rows: vec![],
outcomes: vec![],
},
16,
limits(),
Replication::Disabled,
)?;
let session = table.open_session()?;
let saved = table.allocate_id()?;
let op = table.operation(session, 1);
table
.step(Submit {
receipt: None,
operation: op,
expected: Revision(0),
guarantee: Guarantee::LocallyDurable,
changes: vec![Change::Insert {
id: saved,
row: Generated { value: 10 },
}],
})
.map_err(|(_, error)| error)?;
let attempt = table
.prepare(Guarantee::LocallyDurable)?
.ok_or(Error::Busy)?;
table.step(Complete {
attempt,
result: PhysicalResult::Durable,
})?;
let abandoned = table.allocate_id()?;
let mut recovered =
Authority::recover(table.checkpoint()?, 32, limits(), Replication::Disabled)?;
let next = recovered.allocate_id()?;
assert_eq!(saved.value().get(), 1);
assert_eq!(abandoned.value().get(), 2);
assert_eq!(next.value().get(), 17);
Ok(())
}
#[test]
fn admission_refuses_excess_retained_buffer_capacity_before_publication() -> Result<(), Error> {
let mut table = fresh(Replication::Disabled)?;
let session = table.open_session()?;
let op = table.operation(session, 1);
let mut changes = alloc::vec::Vec::with_capacity(100);
changes.push(Change::Insert {
id: 1,
row: Stock { quantity: 10 },
});
let result = table.step(Submit {
receipt: None,
operation: op,
expected: Revision(0),
guarantee: Guarantee::Accepted,
changes,
});
assert!(matches!(result, Err((_, Error::InvalidBounds))));
assert_eq!(table.revision(), Revision(0));
Ok(())
}
fn model_run(steps: &[(u8, u8)]) -> Result<(), Error> {
let mut table = Authority::<Stock>::recover(
Recovery {
history: HistoryId([9; 16]),
revision: Revision(0),
reserved_through: 0,
rows: vec![(1, Stock { quantity: 0 }), (2, Stock { quantity: 0 })],
outcomes: vec![],
},
1024,
limits(),
Replication::Required,
)?;
let session = table.open_session()?;
let mut sequence = 1u64;
let mut operations = alloc::vec::Vec::new();
let mut accepted = 0u32;
let mut local = 0u32;
let mut remote = None;
for (action, value) in steps {
match action {
0 => {
let op = table.operation(session, sequence);
let value = u32::from(*value);
let result = submit(
&mut table,
op,
vec![
Change::Replace {
id: 1,
row: Stock { quantity: value },
},
Change::Replace {
id: 2,
row: Stock { quantity: value },
},
],
);
match result {
Ok(_confirmation) => {
sequence = sequence.checked_add(1).ok_or(Error::Exhausted)?;
operations.push(op);
accepted = value;
}
Err(Error::Full) => {}
Err(error) => return Err(error),
}
}
1 => {
flush(&mut table, Guarantee::LocallyDurable)?;
local = accepted;
}
2 => {
flush(&mut table, Guarantee::ReplicatedDurable)?;
remote = Some(local);
}
3 => {
for op in &operations {
if let Some(record) = table.record(*op) {
let before = table.revision();
let record = record.clone();
table
.step(Submit {
receipt: None,
operation: *op,
expected: record.expected,
guarantee: Guarantee::Accepted,
changes: record.changes,
})
.map_err(|(_, error)| error)?;
assert_eq!(table.revision(), before);
}
}
}
4..=u8::MAX => {
operations.retain(|op| {
if matches!(
table.inquire(*op, Guarantee::ReplicatedDurable),
Confirmation::Confirmed { .. }
) {
assert!(table.retire(*op).is_ok());
false
} else {
true
}
});
}
}
for id in [1, 2] {
assert_eq!(quantity(&table, Guarantee::Accepted, id)?, accepted);
assert_eq!(quantity(&table, Guarantee::LocallyDurable, id)?, local);
match remote {
Some(value) => {
assert_eq!(quantity(&table, Guarantee::ReplicatedDurable, id)?, value)
}
None => assert_eq!(
quantity(&table, Guarantee::ReplicatedDurable, id),
Err(Error::Busy)
),
}
}
}
Ok(())
}
proptest::proptest! {
#[test]
fn arbitrary_admission_flush_retry_and_retirement_preserve_three_coherent_views(
steps in proptest::collection::vec((0u8..5,proptest::num::u8::ANY),0..128)
) {
proptest::prop_assert_eq!(model_run(&steps),Ok(()));
}
}
#[test]
fn admission_errors_preserve_request_identity_and_state() -> Result<(), Error> {
let mut table = fresh(Replication::Disabled)?;
let session = table.open_session()?;
let op = table.operation(session, 1);
let request = |operation, expected, guarantee, changes| Submit {
receipt: None,
operation,
expected,
guarantee,
changes,
};
let insert = || {
vec![Change::Insert {
id: 1,
row: Stock { quantity: 10 },
}]
};
assert!(matches!(
table.step(request(
op,
Revision(0),
Guarantee::ReplicatedDurable,
insert()
)),
Err((_, Error::Unsupported))
));
let ahead = table.operation(session, 2);
assert!(matches!(
table.step(request(ahead, Revision(99), Guarantee::Accepted, insert())),
Err((_, Error::StaleRevision))
));
assert!(matches!(
table.step(request(op, Revision(99), Guarantee::Accepted, insert())),
Err((_, Error::StaleRevision))
));
assert!(matches!(
table.step(request(op, Revision(0), Guarantee::Accepted, vec![])),
Err((_, Error::InvalidBounds))
));
assert!(matches!(
table.step(request(
op,
Revision(0),
Guarantee::Accepted,
vec![
Change::Insert {
id: 1,
row: Stock { quantity: 10 }
},
Change::Insert {
id: 1,
row: Stock { quantity: 20 }
}
]
)),
Err((_, Error::DuplicateKey))
));
table
.step(request(op, Revision(0), Guarantee::Accepted, insert()))
.map_err(|(_, error)| error)?;
assert!(matches!(
table.step(request(
op,
Revision(0),
Guarantee::Accepted,
vec![Change::Insert {
id: 1,
row: Stock { quantity: 11 }
}]
)),
Err((_, Error::ConflictingRetry))
));
assert_eq!(table.prepare(Guarantee::Accepted), Err(Error::Unsupported));
assert_eq!(
table.prepare(Guarantee::ReplicatedDurable),
Err(Error::Unsupported)
);
let attempt = table
.prepare(Guarantee::LocallyDurable)?
.ok_or(Error::Busy)?;
assert_eq!(table.prepare(Guarantee::LocallyDurable), Err(Error::Busy));
table.step(Complete {
attempt,
result: PhysicalResult::Durable,
})?;
assert_eq!(
table.step(Complete {
attempt,
result: PhysicalResult::Durable
}),
Err(Error::StaleCompletion)
);
table.close_session(session)?;
assert_eq!(table.close_session(session), Err(Error::UnknownSession));
assert!(matches!(
table.step(request(ahead, Revision(1), Guarantee::Accepted, insert())),
Err((_, Error::UnknownSession))
));
assert_eq!(quantity(&table, Guarantee::Accepted, 1)?, 10);
Ok(())
}
#[test]
fn partially_inserted_full_batch_rolls_back_before_publication() -> Result<(), Error> {
let mut bounds = limits();
bounds.rows = Capacity::new(NonZeroU32::MIN);
let mut table = Authority::<Stock>::recover(
Recovery {
history: HistoryId([9; 16]),
revision: Revision(0),
reserved_through: 0,
rows: vec![],
outcomes: vec![],
},
16,
bounds,
Replication::Disabled,
)?;
let session = table.open_session()?;
let operation = table.operation(session, 1);
let changes = vec![
Change::Insert {
id: 1,
row: Stock { quantity: 10 },
},
Change::Insert {
id: 2,
row: Stock { quantity: 20 },
},
];
assert!(matches!(
table.step(Submit {
receipt: None,
operation,
expected: Revision(0),
guarantee: Guarantee::Accepted,
changes
}),
Err((_, Error::Full))
));
assert_eq!(quantity(&table, Guarantee::Accepted, 1), Err(Error::Absent));
assert_eq!(table.revision(), Revision(0));
table
.step(Submit {
receipt: None,
operation,
expected: Revision(0),
guarantee: Guarantee::Accepted,
changes: vec![Change::Insert {
id: 2,
row: Stock { quantity: 20 },
}],
})
.map_err(|(_, error)| error)?;
assert_eq!(quantity(&table, Guarantee::Accepted, 2)?, 20);
Ok(())
}
mod commands;
#[cfg(feature = "secondary-indexes")]
mod located;
mod recovery;