use super::*;
use crate::storage::warp_table::WarpTable;
struct Commands;
#[derive(Clone, Copy)]
enum Command {
Insert(u32),
Refuse,
ExceedBound,
}
impl TableProtocol<Stock> for Commands {
const SCHEMA: [u8; 16] = [3; 16];
const VERSION: u32 = 1;
type Query = u32;
type Value = u32;
type Mutation = Command;
type Outcome = u32;
type Refusal = Error;
fn query(id: &u32, table: &WarpTable<Stock>) -> Result<u32, Error> {
match table.get(*id) {
WarpTableGetOutcome::Present { row } => Ok(*row.quantity),
WarpTableGetOutcome::Absent { .. } => Err(Error::Absent),
}
}
fn prepare(
command: &Command,
table: &WarpTable<Stock>,
changes: &mut ChangeSet<Stock>,
) -> Result<u32, Error> {
assert!(table.is_empty());
match command {
Command::Insert(quantity) => {
changes
.push(Change::Insert {
id: 1,
row: Stock {
quantity: *quantity,
},
})
.map_err(|_| Error::Full)?;
Ok(*quantity)
}
Command::Refuse => {
changes
.push(Change::Insert {
id: 1,
row: Stock { quantity: 99 },
})
.map_err(|_| Error::Full)?;
Err(Error::UniqueConflict)
}
Command::ExceedBound => {
for id in 1..=5 {
let _refused = changes.push(Change::Remove { id });
}
Ok(0)
}
}
}
}
#[test]
fn preparation_retains_exact_outcomes_without_publishing_or_consuming_operations()
-> Result<(), Error> {
let mut authority = fresh(Replication::Disabled)?;
let session = authority.open_session()?;
let operation = authority.next_operation(session)?;
assert_eq!(Receipt::from_operation(operation).operation(), operation);
let prepared = authority.step(PrepareCommand::<Stock, Commands> {
operation,
guarantee: Guarantee::Accepted,
command: &Command::Insert(7),
})?;
assert_eq!(prepared.expected, Revision(0));
assert_eq!(prepared.outcome, Ok(7));
assert!(matches!(
prepared.changes.as_slice(),
[Change::Insert {
id: 1,
row: Stock { quantity: 7 }
}]
));
assert_eq!(
quantity(&authority, Guarantee::Accepted, 1),
Err(Error::Absent)
);
assert_eq!(authority.next_operation(session)?, operation);
let refused = authority.step(PrepareCommand::<Stock, Commands> {
operation,
guarantee: Guarantee::Accepted,
command: &Command::Refuse,
})?;
assert_eq!(refused.expected, Revision(0));
assert_eq!(refused.outcome, Err(Error::UniqueConflict));
assert!(refused.changes.is_empty());
assert!(matches!(
authority.step(PrepareCommand::<Stock, Commands> {
operation,
guarantee: Guarantee::Accepted,
command: &Command::ExceedBound
}),
Err(Error::Full)
));
assert_eq!(authority.revision(), Revision(0));
assert_eq!(authority.next_operation(session)?, operation);
assert!(matches!(
authority.step(PrepareCommand::<Stock, Commands> {
operation,
guarantee: Guarantee::ReplicatedDurable,
command: &Command::Insert(7)
}),
Err(Error::Unsupported)
));
let mut wrong = operation;
wrong.history = HistoryId([99; 16]);
assert!(matches!(
authority.step(PrepareCommand::<Stock, Commands> {
operation: wrong,
guarantee: Guarantee::Accepted,
command: &Command::Insert(7)
}),
Err(Error::InvalidRecovery)
));
Ok(())
}
#[test]
fn sessions_and_attempts_preserve_identity_and_refuse_unsupported_guarantees() -> Result<(), Error>
{
let mut authority = fresh(Replication::Disabled)?;
assert_eq!(
authority.next_operation(SessionId(0)),
Err(Error::UnknownSession)
);
let session = authority.open_session()?;
let operation = authority.next_operation(session)?;
assert_eq!(operation.sequence, 1);
let _second = authority.open_session()?;
assert_eq!(authority.open_session(), Err(Error::Full));
assert_eq!(
authority.inquire(operation, Guarantee::ReplicatedDurable),
Confirmation::Unsupported
);
assert_eq!(
authority.prepare(Guarantee::Accepted),
Err(Error::Unsupported)
);
authority
.step(Submit {
operation,
expected: Revision(0),
guarantee: Guarantee::Accepted,
receipt: None,
changes: vec![Change::Insert {
id: 1,
row: Stock { quantity: 7 },
}],
})
.map_err(|(_, error)| error)?;
assert_eq!(authority.next_operation(session)?.sequence, 2);
let attempt = authority
.prepare(Guarantee::LocallyDurable)?
.ok_or(Error::Busy)?;
assert_eq!(attempt.guarantee(), Guarantee::LocallyDurable);
assert_eq!(attempt.history(), authority.history());
assert_eq!(attempt.through(), Revision(1));
authority.step(Complete {
attempt,
result: PhysicalResult::Durable,
})?;
authority.close_session(session)?;
assert_eq!(
authority.next_operation(session),
Err(Error::UnknownSession)
);
let replacement = authority.open_session()?;
assert_ne!(replacement, session);
assert_eq!(authority.next_operation(replacement)?.sequence, 1);
Ok(())
}
#[test]
fn validation_restores_insertions_and_removals_and_rejects_excess_changes() -> Result<(), Error> {
let mut authority = fresh(Replication::Disabled)?;
authority.validate_changes(&[Change::Insert {
id: 1,
row: Stock { quantity: 7 },
}])?;
assert_eq!(
quantity(&authority, Guarantee::Accepted, 1),
Err(Error::Absent)
);
assert_eq!(authority.revision(), Revision(0));
assert_eq!(
authority.validate_changes(&[Change::Remove { id: 1 }]),
Err(Error::Absent)
);
assert_eq!(
authority.validate_changes(
&(1..=5)
.map(|id| Change::Remove { id })
.collect::<alloc::vec::Vec<_>>()
),
Err(Error::InvalidBounds)
);
let session = authority.open_session()?;
let operation = authority.next_operation(session)?;
authority
.step(Submit {
operation,
expected: Revision(0),
guarantee: Guarantee::Accepted,
receipt: None,
changes: vec![Change::Insert {
id: 1,
row: Stock { quantity: 7 },
}],
})
.map_err(|(_, error)| error)?;
authority.validate_changes(&[Change::Remove { id: 1 }])?;
assert_eq!(quantity(&authority, Guarantee::Accepted, 1)?, 7);
assert_eq!(authority.revision(), Revision(1));
Ok(())
}
#[test]
fn receipt_capacity_is_checked_before_admission_and_recovery() -> Result<(), Error> {
let mut authority = fresh(Replication::Disabled)?;
let session = authority.open_session()?;
let operation = authority.next_operation(session)?;
let receipt = CommandReceipt {
schema: [4; 16],
version: 2,
request: vec![7; 1025],
outcome: vec![8; 3],
};
assert_eq!(
receipt.retained_bytes(),
Some(
receipt
.request
.capacity()
.saturating_add(receipt.outcome.capacity())
)
);
let submitted = Submit {
operation,
expected: Revision(0),
guarantee: Guarantee::Accepted,
receipt: Some(receipt.clone()),
changes: vec![],
};
let Err((rejected, error)) = authority.step(submitted) else {
return Err(Error::Corrupt);
};
assert_eq!(error, Error::InvalidBounds);
assert_eq!(rejected.operation, operation);
assert_eq!(rejected.receipt, Some(receipt.clone()));
assert_eq!(authority.revision(), Revision(0));
assert_eq!(authority.next_operation(session)?, operation);
let record = Record::<Stock, u32> {
receipt: Some(receipt),
operation,
expected: Revision(0),
revision: Revision(1),
changes: vec![],
};
let image = Recovery {
history: authority.history(),
revision: Revision(1),
reserved_through: 16,
rows: vec![],
outcomes: vec![record],
};
assert!(matches!(
Authority::recover(image.clone(), 32, limits(), Replication::Disabled),
Err(Error::InvalidBounds)
));
assert!(matches!(
RecoveryBuilder::new(image, limits()),
Err(Error::InvalidBounds)
));
let exact = CommandReceipt {
schema: [4; 16],
version: 2,
request: vec![7; limits().receipt_bytes],
outcome: vec![],
};
assert_eq!(exact.retained_bytes(), Some(limits().receipt_bytes));
let mut replay = RecoveryBuilder::new(authority.checkpoint()?, limits())?;
assert_eq!(
authority
.step(Submit {
operation,
expected: Revision(0),
guarantee: Guarantee::Accepted,
receipt: Some(exact),
changes: vec![],
})
.map_err(|(_, error)| error)?,
Confirmation::Confirmed {
revision: Revision(1),
established: Guarantee::Accepted,
}
);
assert_eq!(authority.revision(), Revision(1));
flush(&mut authority, Guarantee::LocallyDurable)?;
let checkpoint = authority.checkpoint()?;
replay.apply(checkpoint.outcomes.first().ok_or(Error::Absent)?.clone())?;
assert_eq!(replay.snapshot()?.outcomes, checkpoint.outcomes);
assert_eq!(
RecoveryBuilder::new(checkpoint.clone(), limits())?
.snapshot()?
.outcomes,
checkpoint.outcomes
);
let recovered = Authority::recover(checkpoint, 32, limits(), Replication::Disabled)?;
assert_eq!(
recovered.inquire(operation, Guarantee::LocallyDurable),
Confirmation::Confirmed {
revision: Revision(1),
established: Guarantee::LocallyDurable,
}
);
Ok(())
}