use super::*;
fn image() -> Recovery<Stock, u32> {
Recovery {
history: HistoryId([7; 16]),
revision: Revision(2),
reserved_through: 16,
rows: vec![(1, Stock { quantity: 20 })],
outcomes: vec![Record {
receipt: None,
operation: OperationId {
history: HistoryId([7; 16]),
epoch: 1,
session: SessionId(1),
sequence: 2,
},
revision: Revision(2),
expected: Revision(1),
changes: vec![Change::Replace {
id: 1,
row: Stock { quantity: 20 },
}],
}],
}
}
#[test]
fn every_malformed_retained_identity_or_revision_is_refused_by_both_recovery_paths()
-> Result<(), Error> {
for case in 0..12 {
let mut recovery = image();
let record = recovery.outcomes.first_mut().ok_or(Error::Absent)?;
match case {
0 => record.operation.sequence = 0,
1 => record.operation.session = SessionId(0),
2 => record.operation.epoch = 0,
3 => record.operation.epoch = 3,
4 => record.expected = record.revision,
5 => record.expected = Revision(3),
6 => record.changes.clear(),
7 => record.changes.push(Change::Remove { id: 1 }),
8 => record.operation.history = HistoryId([8; 16]),
9 => record.revision = Revision(3),
10 => recovery
.outcomes
.push(recovery.outcomes.first().ok_or(Error::Absent)?.clone()),
11 => recovery.reserved_through = 1,
_ => return Err(Error::Corrupt),
}
assert!(
Authority::recover(recovery.clone(), 32, limits(), Replication::Required).is_err(),
"case {case}"
);
assert!(
RecoveryBuilder::new(recovery, limits()).is_err(),
"case {case}"
);
}
Ok(())
}
#[test]
fn recovery_capacity_and_owned_bytes_are_enforced_before_publication() -> Result<(), Error> {
for case in 0..5 {
let mut bounds = limits();
let mut recovery = image();
match case {
0 => bounds.changes = 0,
1 => bounds.row_bytes = 0,
2 => bounds.row_bytes = 3,
3 => recovery.rows.push((1, Stock { quantity: 30 })),
4 => {
recovery.rows = (1..=5).map(|id| (id, Stock { quantity: id })).collect();
}
_ => return Err(Error::Corrupt),
}
assert!(Authority::recover(recovery.clone(), 32, bounds, Replication::Disabled).is_err());
assert!(RecoveryBuilder::new(recovery, bounds).is_err());
}
Ok(())
}
#[test]
fn ordered_replay_rejects_gaps_conflicts_and_excess_retention() -> Result<(), Error> {
let original = image();
let mut builder = RecoveryBuilder::new(original.clone(), limits())?;
assert!(builder.rows_match(&original.rows));
assert!(!builder.rows_match(&[]));
assert!(!builder.rows_match(&[(2, Stock { quantity: 20 })]));
assert!(!builder.rows_match(&[(1, Stock { quantity: 21 })]));
let saved = original.outcomes.first().ok_or(Error::Absent)?.clone();
builder.apply(saved.clone())?;
let mut conflicting = saved.clone();
conflicting.changes = vec![Change::Remove { id: 1 }];
assert_eq!(builder.apply(conflicting), Err(Error::ConflictingRetry));
let next = Record {
receipt: None,
operation: OperationId {
sequence: 3,
..saved.operation
},
revision: Revision(3),
expected: Revision(2),
changes: vec![Change::Remove { id: 1 }],
};
for case in 0..4 {
let mut invalid = next.clone();
match case {
0 => invalid.expected = Revision(1),
1 => invalid.revision = Revision(2),
2 => invalid.revision = Revision(17),
3 => invalid.operation.history = HistoryId([8; 16]),
_ => return Err(Error::Corrupt),
}
assert_eq!(builder.apply(invalid), Err(Error::InvalidRecovery));
assert!(builder.rows_match(&original.rows));
}
assert_eq!(builder.reserve(15), Err(Error::InvalidRecovery));
builder.reserve(16)?;
builder.reserve(32)?;
assert_eq!(builder.reserved_through(), 32);
builder.apply(next)?;
assert!(builder.rows_match(&[]));
let snapshot = builder.snapshot()?;
assert_eq!(snapshot.revision, Revision(3));
assert_eq!(snapshot.outcomes.len(), 2);
let mut bounds = limits();
bounds.operations = Capacity::new(NonZeroU32::MIN);
let mut full = RecoveryBuilder::new(original, bounds)?;
let next = snapshot.outcomes.last().ok_or(Error::Absent)?.clone();
assert_eq!(full.apply(next), Err(Error::Full));
assert_eq!(full.revision(), Revision(2));
Ok(())
}
#[test]
fn recovered_outcomes_answer_before_remote_confirmation_and_never_reapply() -> Result<(), Error> {
let original = image();
let saved = original.outcomes.first().ok_or(Error::Absent)?.clone();
let mut authority = Authority::recover(original, 32, limits(), Replication::Required)?;
assert_eq!(authority.local_revision(), Revision(2));
assert_eq!(authority.remote_revision(), Revision(0));
assert_eq!(authority.epoch(), 17);
assert_eq!(authority.limits().changes, 4);
assert_eq!(
authority.inquire(saved.operation, Guarantee::LocallyDurable),
Confirmation::Confirmed {
revision: Revision(2),
established: Guarantee::LocallyDurable
}
);
assert_eq!(
authority.inquire(saved.operation, Guarantee::ReplicatedDurable),
Confirmation::Pending {
revision: Revision(2),
established: Guarantee::LocallyDurable
}
);
assert!(matches!(
authority.read(ReadRequest {
guarantee: Guarantee::ReplicatedDurable,
}),
ReadOutcome::Pending
));
let attempt = authority
.prepare(Guarantee::ReplicatedDurable)?
.ok_or(Error::Busy)?;
assert_eq!(attempt.history(), authority.history());
assert_eq!(attempt.through(), Revision(2));
authority.step(Complete {
attempt,
result: PhysicalResult::Unresolved,
})?;
assert!(authority.batch(attempt).is_err());
let next = authority
.prepare(Guarantee::ReplicatedDurable)?
.ok_or(Error::Busy)?;
assert_ne!(next, attempt);
assert_eq!(
authority.step(Complete {
attempt,
result: PhysicalResult::Durable
}),
Err(Error::StaleCompletion)
);
authority.step(Complete {
attempt: next,
result: PhysicalResult::Durable,
})?;
assert_eq!(quantity(&authority, Guarantee::ReplicatedDurable, 1)?, 20);
assert_eq!(authority.prepare(Guarantee::ReplicatedDurable)?, None);
let replay = authority
.step(Submit {
receipt: None,
operation: saved.operation,
expected: saved.expected,
guarantee: Guarantee::ReplicatedDurable,
changes: saved.changes,
})
.map_err(|(_, error)| error)?;
assert_eq!(
replay,
Confirmation::Confirmed {
revision: Revision(2),
established: Guarantee::ReplicatedDurable
}
);
assert_eq!(authority.revision(), Revision(2));
authority.retire(saved.operation)?;
assert_eq!(
authority.inquire(saved.operation, Guarantee::Accepted),
Confirmation::Unknown
);
Ok(())
}
proptest::proptest! {
#[test]
fn checkpoint_and_incremental_replay_match_an_independent_map(values in proptest::collection::vec((0u8..4,proptest::num::u32::ANY),1..100)) {
fn run(values:&[(u8,u32)]) -> Result<(),Error> {
let mut authority=fresh(Replication::Required)?;
let mut builder=RecoveryBuilder::new(authority.checkpoint()?,limits())?;
let session=authority.open_session()?;
let mut model=alloc::collections::BTreeMap::new();
for (index,(id,quantity)) in values.iter().copied().enumerate() {
if authority.reservation_required() {
let end=authority.reserved_through().checked_add(16).ok_or(Error::Exhausted)?;
builder.reserve(end)?;authority.reserved(end)?;
}
let id=u32::from(id);
let change=if model.contains_key(&id) {Change::Replace{id,row:Stock{quantity}}} else {Change::Insert{id,row:Stock{quantity}}};
let operation=authority.operation(session,(index as u64).checked_add(1).ok_or(Error::Exhausted)?);
submit(&mut authority,operation,vec![change])?;
builder.apply(authority.record(operation).ok_or(Error::Absent)?.clone())?;
model.insert(id,Stock{quantity});
let snapshot=builder.snapshot()?;
let actual:alloc::collections::BTreeMap<_,_>=snapshot.rows.iter().cloned().collect();
assert_eq!(actual,model);
flush(&mut authority,Guarantee::LocallyDurable)?;
flush(&mut authority,Guarantee::ReplicatedDurable)?;
assert_eq!(authority.local_revision(),snapshot.revision);
let _retired=authority.retire(operation)?;
builder=RecoveryBuilder::new(authority.checkpoint()?,limits())?;
assert_eq!(builder.snapshot()?.rows.len(),model.len());
}
Ok(())
}
proptest::prop_assert_eq!(run(&values),Ok(()));
}
}
#[test]
fn empty_initialization_validates_bounds_and_loaded_rows_independently_of_outcomes()
-> Result<(), Error> {
for (changes, row_bytes) in [(0, 4), (4, 0)] {
let mut bounds = limits();
bounds.changes = changes;
bounds.row_bytes = row_bytes;
let mut recovery = image();
recovery.rows.clear();
recovery.outcomes.clear();
assert!(matches!(
Authority::recover(recovery, 32, bounds, Replication::Disabled),
Err(Error::InvalidBounds)
));
}
let mut recovery = image();
recovery.outcomes.clear();
let mut bounds = limits();
bounds.row_bytes = 3;
assert!(matches!(
Authority::recover(recovery.clone(), 32, bounds, Replication::Required),
Err(Error::InvalidBounds)
));
bounds.row_bytes = 8;
let table = Authority::recover(recovery, 32, bounds, Replication::Required)?;
assert_eq!(quantity(&table, Guarantee::Accepted, 1)?, 20);
assert!(!table.reservation_required());
let mut exhausted = image();
exhausted.revision = Revision(u64::MAX);
exhausted.reserved_through = u64::MAX;
let record = exhausted.outcomes.first_mut().ok_or(Error::Absent)?;
record.revision = Revision(u64::MAX);
record.expected = Revision(u64::MAX - 1);
record.operation.epoch = u64::MAX;
record.changes = vec![Change::Insert {
id: 1,
row: Stock { quantity: 20 },
}];
let saved = record.operation;
let mut table = Authority::recover(exhausted, u64::MAX, limits(), Replication::Required)?;
assert!(!table.reservation_required());
flush(&mut table, Guarantee::ReplicatedDurable)?;
assert_eq!(quantity(&table, Guarantee::ReplicatedDurable, 1)?, 20);
assert_eq!(
table.inquire(saved, Guarantee::ReplicatedDurable),
Confirmation::Confirmed {
revision: Revision(u64::MAX),
established: Guarantee::ReplicatedDurable
}
);
assert_eq!(
table.inquire(saved, Guarantee::LocallyDurable),
Confirmation::Confirmed {
revision: Revision(u64::MAX),
established: Guarantee::ReplicatedDurable
}
);
Ok(())
}
#[test]
fn an_outstanding_batch_never_includes_newer_or_already_confirmed_records() -> Result<(), Error> {
let mut table = fresh(Replication::Required)?;
let session = table.open_session()?;
let first = table.operation(session, 1);
submit(
&mut table,
first,
vec![Change::Insert {
id: 1,
row: Stock { quantity: 10 },
}],
)?;
let local_first = table
.prepare(Guarantee::LocallyDurable)?
.ok_or(Error::Busy)?;
let second = table.operation(session, 2);
submit(
&mut table,
second,
vec![Change::Replace {
id: 1,
row: Stock { quantity: 20 },
}],
)?;
assert_eq!(
table
.batch(local_first)?
.map(|record| record.operation)
.collect::<alloc::vec::Vec<_>>(),
vec![first]
);
table.step(Complete {
attempt: local_first,
result: PhysicalResult::Durable,
})?;
assert_eq!(quantity(&table, Guarantee::LocallyDurable, 1)?, 10);
assert_eq!(quantity(&table, Guarantee::Accepted, 1)?, 20);
let remote_first = table
.prepare(Guarantee::ReplicatedDurable)?
.ok_or(Error::Busy)?;
let local_second = table
.prepare(Guarantee::LocallyDurable)?
.ok_or(Error::Busy)?;
assert_eq!(
table
.batch(local_second)?
.map(|record| record.operation)
.collect::<alloc::vec::Vec<_>>(),
vec![second]
);
table.step(Complete {
attempt: local_second,
result: PhysicalResult::Durable,
})?;
assert_eq!(
table
.batch(remote_first)?
.map(|record| record.operation)
.collect::<alloc::vec::Vec<_>>(),
vec![first]
);
table.step(Complete {
attempt: remote_first,
result: PhysicalResult::Durable,
})?;
assert_eq!(quantity(&table, Guarantee::ReplicatedDurable, 1)?, 10);
let remote_second = table
.prepare(Guarantee::ReplicatedDurable)?
.ok_or(Error::Busy)?;
assert_eq!(
table
.batch(remote_second)?
.map(|record| record.operation)
.collect::<alloc::vec::Vec<_>>(),
vec![second]
);
table.step(Complete {
attempt: remote_second,
result: PhysicalResult::Durable,
})?;
assert_eq!(quantity(&table, Guarantee::ReplicatedDurable, 1)?, 20);
Ok(())
}
#[test]
fn exact_admission_bounds_are_usable_and_row_size_is_a_separate_bound() -> Result<(), Error> {
for row_bytes in [3, 4, 8] {
let mut bounds = limits();
bounds.row_bytes = row_bytes;
let mut table = Authority::recover(
Recovery {
history: HistoryId([7; 16]),
revision: Revision(0),
reserved_through: 0,
rows: vec![],
outcomes: vec![],
},
16,
bounds,
Replication::Required,
)?;
let session = table.open_session()?;
let operation = table.operation(session, 1);
let changes = (1..=4)
.map(|id| Change::Insert {
id,
row: Stock { quantity: id },
})
.collect::<alloc::vec::Vec<_>>();
assert_eq!(changes.capacity(), 4);
let result = submit(&mut table, operation, changes);
if row_bytes == 3 {
assert_eq!(result, Err(Error::InvalidBounds));
assert_eq!(table.revision(), Revision(0));
} else {
assert!(result.is_ok());
assert_eq!(quantity(&table, Guarantee::Accepted, 4)?, 4);
}
}
Ok(())
}
#[test]
fn insert_never_silently_replaces_an_existing_identity() -> Result<(), Error> {
let mut table = fresh(Replication::Required)?;
let session = table.open_session()?;
let first = table.operation(session, 1);
submit(
&mut table,
first,
vec![Change::Insert {
id: 1,
row: Stock { quantity: 10 },
}],
)?;
let operation = table.operation(session, 2);
assert_eq!(
submit(
&mut table,
operation,
vec![Change::Insert {
id: 1,
row: Stock { quantity: 99 }
}]
),
Err(Error::Present)
);
assert_eq!(table.revision(), Revision(1));
assert_eq!(quantity(&table, Guarantee::Accepted, 1)?, 10);
submit(
&mut table,
operation,
vec![Change::Replace {
id: 1,
row: Stock { quantity: 20 },
}],
)?;
Ok(())
}
#[test]
fn replay_validates_record_payload_bounds_even_when_snapshot_rows_are_empty() -> Result<(), Error> {
for row_bytes in [3, 4, 8] {
for count in [4, 5] {
let mut bounds = limits();
bounds.row_bytes = row_bytes;
let record = Record {
receipt: None,
operation: OperationId {
history: HistoryId([7; 16]),
epoch: 1,
session: SessionId(1),
sequence: 1,
},
revision: Revision(1),
expected: Revision(0),
changes: (0..count)
.map(|id| Change::Insert {
id,
row: Stock { quantity: 20 },
})
.collect(),
};
let snapshot = Recovery {
history: record.operation.history,
revision: Revision(1),
reserved_through: 16,
rows: vec![],
outcomes: vec![record],
};
let result = RecoveryBuilder::new(snapshot, bounds);
if count == 5 {
assert!(matches!(result, Err(Error::InvalidRecovery)));
} else if row_bytes == 3 {
assert!(matches!(result, Err(Error::InvalidBounds)));
} else {
assert_eq!(result?.snapshot()?.outcomes.len(), 1);
}
}
}
Ok(())
}
#[test]
fn every_identity_component_retains_bit_influence_in_the_primary_index() -> Result<(), Error> {
use crate::storage::warp_table::IndexKey;
for background in [1, u64::MAX] {
let base = OperationId {
history: HistoryId([7; 16]),
epoch: background,
session: SessionId(background),
sequence: background,
};
for bit in 0..64 {
let mask = 1u64.checked_shl(bit).ok_or(Error::Exhausted)?;
for varied in [
OperationId {
epoch: base.epoch ^ mask,
..base
},
OperationId {
session: SessionId(base.session.0 ^ mask),
..base
},
OperationId {
sequence: base.sequence ^ mask,
..base
},
] {
assert_ne!(varied.index_word(), base.index_word());
}
assert_ne!(
SessionId(background ^ mask).index_word(),
SessionId(background).index_word()
);
}
}
Ok(())
}
#[test]
fn increasing_revisions_cannot_retain_the_same_operation_twice() -> Result<(), Error> {
let mut original = image();
let mut earlier = original.outcomes.first().ok_or(Error::Absent)?.clone();
earlier.revision = Revision(1);
earlier.expected = Revision(0);
original.outcomes.insert(0, earlier);
assert!(matches!(
Authority::recover(original.clone(), 32, limits(), Replication::Required),
Err(Error::InvalidRecovery)
));
assert!(matches!(
RecoveryBuilder::new(original, limits()),
Err(Error::InvalidRecovery)
));
let builder = RecoveryBuilder::new(image(), limits())?;
assert_eq!(builder.history(), HistoryId([7; 16]));
assert_eq!(builder.revision(), Revision(2));
Ok(())
}