use super::*;
use alloc::vec;
use core::num::NonZeroU32;
#[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 {
4
}
}
fn fresh() -> Result<Authority<Stock>, Error> {
let capacity = super::super::super::Capacity::new(NonZeroU32::MIN.saturating_add(1));
Authority::recover(
Recovery {
history: HistoryId([3; 16]),
revision: Revision(0),
reserved_through: 0,
rows: vec![],
outcomes: vec![],
},
16,
Limits {
rows: capacity,
operations: capacity,
sessions: capacity,
changes: 2,
row_bytes: 4,
receipt_bytes: 16,
},
Replication::Required,
)
}
#[test]
fn retained_outcome_adapters_preserve_owned_receipts_changes_and_exact_refusals()
-> Result<(), Error> {
use crate::storage::warp_table::WarpTableInsertWithIdOutcome;
let operation = OperationId {
history: HistoryId([7; 16]),
epoch: 11,
session: SessionId(13),
sequence: 17,
};
let expected = Record {
receipt: Some(CommandReceipt {
schema: [3; 16],
version: 5,
request: vec![7, 11],
outcome: vec![13, 17],
}),
operation,
expected: Revision(19),
revision: Revision(23),
changes: vec![Change::Replace {
id: 29,
row: Stock { quantity: 31 },
}],
};
for occupied in [true, false] {
let record = expected.clone();
let changes_pointer = record.changes.as_ptr();
let receipt = record.receipt.as_ref().ok_or(Error::InvalidRecovery)?;
let request_pointer = receipt.request.as_ptr();
let outcome_pointer = receipt.outcome.as_ptr();
let rejected = Retained { record };
let outcome = if occupied {
WarpTableInsertWithIdOutcome::AlreadyPresent {
id: operation,
existing_row: 0,
rejected,
}
} else {
WarpTableInsertWithIdOutcome::Full {
id: operation,
rejected,
}
};
let Err(returned) = retention_result(outcome, Guarantee::ReplicatedDurable) else {
return Err(Error::InvalidRecovery);
};
assert_eq!(returned.operation, operation);
assert_eq!(returned.expected, expected.expected);
assert_eq!(returned.guarantee, Guarantee::ReplicatedDurable);
assert_eq!(returned.changes, expected.changes);
assert_eq!(returned.receipt, expected.receipt);
assert_eq!(returned.changes.as_ptr(), changes_pointer);
let receipt = returned.receipt.as_ref().ok_or(Error::InvalidRecovery)?;
assert_eq!(receipt.request.as_ptr(), request_pointer);
assert_eq!(receipt.outcome.as_ptr(), outcome_pointer);
}
Ok(())
}
#[test]
fn missing_retained_operation_or_publication_session_cannot_finish_authority_steps()
-> Result<(), Error> {
let mut authority = fresh()?;
let session = authority.open_session()?;
let operation = authority.next_operation(session)?;
assert!(matches!(
authority.retire_retained(operation),
Err(Error::ExpiredOperation)
));
assert!(authority.order.is_empty());
authority.close_session(session)?;
let clock = authority.clock;
let mut next_clock = clock;
let revision = next_clock.issue()?;
authority.poisoned = true;
assert_eq!(
authority.finish_submission(operation, revision, next_clock),
Err(Error::Corrupt)
);
assert!(authority.poisoned);
assert_eq!(authority.revision(), Revision(0));
assert_eq!(authority.clock, clock);
assert!(authority.records.is_empty());
Ok(())
}
#[test]
fn corrupted_retention_capacity_returns_the_whole_submission_and_poisons_reads() -> Result<(), Error>
{
let mut authority = fresh()?;
let session = authority.open_session()?;
let operation = authority.next_operation(session)?;
let changes = vec![Change::Insert {
id: 7,
row: Stock { quantity: 42 },
}];
authority.records.maximum_rows = 0;
let Err((rejected, error)) = authority.step(Submit {
receipt: None,
operation,
expected: Revision(0),
guarantee: Guarantee::Accepted,
changes: changes.clone(),
}) else {
return Err(Error::InvalidRecovery);
};
assert_eq!(error, Error::Corrupt);
assert_eq!(rejected.receipt, None);
assert_eq!(rejected.operation, operation);
assert_eq!(rejected.expected, Revision(0));
assert_eq!(rejected.guarantee, Guarantee::Accepted);
assert_eq!(rejected.changes, changes);
assert!(authority.records.is_empty());
assert!(authority.order.is_empty());
assert!(authority.poisoned);
assert!(matches!(
authority.read(ReadRequest {
guarantee: Guarantee::Accepted
}),
ReadOutcome::Refused(Error::Corrupt)
));
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 poisoned_authority_cannot_consume_generated_identifiers() -> Result<(), Error> {
let mut authority = Authority::<Generated>::recover(
Recovery {
history: HistoryId([3; 16]),
revision: Revision(0),
reserved_through: 0,
rows: vec![],
outcomes: vec![],
},
16,
fresh()?.limits(),
Replication::Disabled,
)?;
let before = authority.row_clock;
authority.poisoned = true;
assert_eq!(authority.allocate_id(), Err(Error::Corrupt));
assert_eq!(authority.row_clock, before);
Ok(())
}
#[test]
fn rejected_materialized_replacement_poisons_reads_and_returns_the_operation() -> Result<(), Error>
{
let mut authority = fresh()?;
let session = authority.open_session()?;
authority
.step(Submit {
receipt: None,
operation: authority.next_operation(session)?,
expected: Revision(0),
guarantee: Guarantee::Accepted,
changes: vec![Change::Insert {
id: 7,
row: Stock { quantity: 11 },
}],
})
.map_err(|(_, error)| error)?;
let operation = authority.next_operation(session)?;
let proposed = vec![Change::Replace {
id: 7,
row: Stock { quantity: 19 },
}];
authority.accepted.table.maximum_rows = 0;
let Err((rejected, error)) = authority.step(Submit {
receipt: None,
operation,
expected: Revision(1),
guarantee: Guarantee::Accepted,
changes: proposed.clone(),
}) else {
return Err(Error::InvalidRecovery);
};
assert_eq!(error, Error::Corrupt);
assert_eq!(rejected.operation, operation);
assert_eq!(rejected.changes, proposed);
assert_eq!(authority.revision(), Revision(1));
assert_eq!(authority.records.len(), 1);
assert!(authority.poisoned);
assert!(matches!(
authority.read(ReadRequest {
guarantee: Guarantee::Accepted
}),
ReadOutcome::Refused(Error::Corrupt)
));
Ok(())
}
#[test]
fn failed_validation_rollback_poisoning_preserves_the_published_rows() -> Result<(), Error> {
for change in [
Change::Remove { id: 7 },
Change::Replace {
id: 7,
row: Stock { quantity: 19 },
},
] {
let mut authority = fresh()?;
let session = authority.open_session()?;
authority
.step(Submit {
receipt: None,
operation: authority.next_operation(session)?,
expected: Revision(0),
guarantee: Guarantee::Accepted,
changes: vec![Change::Insert {
id: 7,
row: Stock { quantity: 11 },
}],
})
.map_err(|(_, error)| error)?;
authority.staging.maximum_rows = 0;
assert_eq!(authority.validate_changes(&[change]), Err(Error::Corrupt));
assert!(authority.poisoned);
assert_eq!(authority.revision(), Revision(1));
assert!(matches!(authority.accepted.table().get(7),
WarpTableGetOutcome::Present { row } if *row.quantity == 11));
assert!(matches!(
authority.read(ReadRequest {
guarantee: Guarantee::Accepted
}),
ReadOutcome::Refused(Error::Corrupt)
));
}
Ok(())
}
#[test]
fn poisoned_authority_preserves_ownership_and_refuses_further_publication() -> Result<(), Error> {
let mut authority = fresh()?;
let session = authority.open_session()?;
let operation = authority.next_operation(session)?;
authority.poisoned = true;
assert_eq!(authority.validate_changes(&[]), Err(Error::Corrupt));
assert_eq!(authority.open_session(), Err(Error::Corrupt));
assert_eq!(
authority.inquire(operation, Guarantee::Accepted),
Confirmation::Unavailable(Error::Corrupt)
);
assert_eq!(
authority.prepare(Guarantee::LocallyDurable),
Err(Error::Corrupt)
);
assert!(matches!(authority.checkpoint(), Err(Error::Corrupt)));
assert_eq!(
authority.admit_command(operation, Guarantee::Accepted),
Err(Error::Corrupt)
);
assert!(matches!(
authority.read(ReadRequest {
guarantee: Guarantee::Accepted
}),
ReadOutcome::Refused(Error::Corrupt)
));
assert!(matches!(
authority.observation(Guarantee::Accepted),
Err(Error::Corrupt)
));
let proposed = vec![Change::Insert {
id: 7,
row: Stock { quantity: 42 },
}];
let Err((rejected, error)) = authority.step(Submit {
receipt: None,
operation,
expected: Revision(0),
guarantee: Guarantee::Accepted,
changes: proposed.clone(),
}) else {
return Err(Error::InvalidRecovery);
};
assert_eq!(error, Error::Corrupt);
assert_eq!(rejected.operation, operation);
assert_eq!(rejected.changes, proposed);
let attempt = Attempt {
history: authority.history,
epoch: authority.epoch,
sequence: 1,
guarantee: Guarantee::LocallyDurable,
through: Revision(1),
};
assert_eq!(
authority.step(Complete {
attempt,
result: PhysicalResult::Durable
}),
Err(Error::Corrupt)
);
assert_eq!(authority.revision(), Revision(0));
assert_eq!(authority.local_revision(), Revision(0));
assert_eq!(authority.remote_revision(), Revision(0));
assert!(authority.accepted.table().is_empty());
assert!(authority.records.is_empty());
assert_eq!(authority.sessions.len(), 1);
Ok(())
}
#[test]
fn exhausted_session_sequences_cannot_wrap_or_consume_another_operation() -> Result<(), Error> {
let mut authority = fresh()?;
let session = authority.open_session()?;
assert!(matches!(
authority
.sessions
.replace(session, Session { sequence: u64::MAX }),
crate::storage::warp_table::WarpTableReplaceOutcome::Replaced { .. }
));
assert_eq!(authority.next_operation(session), Err(Error::Exhausted));
assert_eq!(authority.revision(), Revision(0));
assert!(authority.records.is_empty());
Ok(())
}
#[test]
fn validating_removal_and_replacement_restores_both_private_and_published_rows() -> Result<(), Error>
{
let mut authority = fresh()?;
let session = authority.open_session()?;
authority
.step(Submit {
receipt: None,
operation: authority.next_operation(session)?,
expected: Revision(0),
guarantee: Guarantee::Accepted,
changes: vec![
Change::Insert {
id: 1,
row: Stock { quantity: 10 },
},
Change::Insert {
id: 2,
row: Stock { quantity: 20 },
},
],
})
.map_err(|(_, error)| error)?;
authority.validate_changes(&[
Change::Replace {
id: 1,
row: Stock { quantity: 30 },
},
Change::Remove { id: 2 },
])?;
for (id, quantity) in [(1, 10), (2, 20)] {
for table in [authority.accepted.table(), &authority.staging] {
let WarpTableGetOutcome::Present { row } = table.get(id) else {
return Err(Error::Absent);
};
assert_eq!(*row.quantity, quantity);
}
}
assert_eq!(authority.revision(), Revision(1));
assert_eq!(authority.next_operation(session)?.sequence, 2);
assert!(!authority.poisoned);
Ok(())
}
#[test]
fn inconsistent_session_and_remote_storage_fail_closed() -> Result<(), Error> {
let mut authority = fresh()?;
let session = authority.open_session()?;
authority.next_session = 0;
assert_eq!(authority.open_session(), Err(Error::Corrupt));
assert_eq!(authority.next_operation(session)?.sequence, 1);
authority.remote_ready = true;
authority.remote = None;
assert!(matches!(
authority.read(ReadRequest {
guarantee: Guarantee::ReplicatedDurable
}),
ReadOutcome::Refused(Error::Corrupt)
));
assert!(matches!(
authority.observation(Guarantee::ReplicatedDurable),
Err(Error::Corrupt)
));
let attempt = Attempt {
history: authority.history,
epoch: authority.epoch,
sequence: 1,
guarantee: Guarantee::Accepted,
through: Revision(0),
};
assert!(matches!(authority.batch(attempt), Err(Error::Unsupported)));
assert_eq!(
authority.step(Complete {
attempt,
result: PhysicalResult::Durable
}),
Err(Error::Unsupported)
);
assert!(!authority.poisoned);
Ok(())
}
#[test]
fn a_missing_retained_record_poisons_completion_without_advancing_durability() -> Result<(), Error>
{
let mut authority = fresh()?;
let session = authority.open_session()?;
let operation = authority.next_operation(session)?;
authority
.step(Submit {
receipt: None,
operation,
expected: Revision(0),
guarantee: Guarantee::Accepted,
changes: vec![Change::Insert {
id: 1,
row: Stock { quantity: 10 },
}],
})
.map_err(|(_, error)| error)?;
let attempt = authority
.prepare(Guarantee::LocallyDurable)?
.ok_or(Error::Busy)?;
assert!(matches!(
authority.records.remove(operation),
WarpTableRemoveOutcome::Removed { .. }
));
assert_eq!(
authority.step(Complete {
attempt,
result: PhysicalResult::Durable
}),
Err(Error::Corrupt)
);
assert_eq!(authority.local_revision(), Revision(0));
assert!(authority.poisoned);
Ok(())
}
#[test]
fn missing_incarnations_poison_publication_without_advancing_its_revision() -> Result<(), Error> {
let mut authority = fresh()?;
let session = authority.open_session()?;
authority
.step(Submit {
receipt: None,
operation: authority.next_operation(session)?,
expected: Revision(0),
guarantee: Guarantee::Accepted,
changes: vec![Change::Insert {
id: 1,
row: Stock { quantity: 10 },
}],
})
.map_err(|(_, error)| error)?;
authority.accepted.incarnations.clear();
let operation = authority.next_operation(session)?;
let changes = vec![Change::Replace {
id: 1,
row: Stock { quantity: 20 },
}];
let Err((rejected, error)) = authority.step(Submit {
receipt: None,
operation,
expected: Revision(1),
guarantee: Guarantee::Accepted,
changes: changes.clone(),
}) else {
return Err(Error::InvalidRecovery);
};
assert_eq!(error, Error::Corrupt);
assert_eq!(rejected.operation, operation);
assert_eq!(rejected.changes, changes);
assert_eq!(authority.revision(), Revision(1));
assert_eq!(authority.local_revision(), Revision(0));
assert_eq!(authority.records.len(), 1);
assert!(authority.poisoned);
assert!(matches!(
authority.read(ReadRequest {
guarantee: Guarantee::Accepted
}),
ReadOutcome::Refused(Error::Corrupt)
));
Ok(())
}
#[test]
fn corrupt_durable_incarnations_refuse_completion_without_advancing_watermarks() -> Result<(), Error>
{
let mut authority = fresh()?;
let session = authority.open_session()?;
let operation = authority.next_operation(session)?;
authority
.step(Submit {
receipt: None,
operation,
expected: Revision(0),
guarantee: Guarantee::Accepted,
changes: vec![Change::Insert {
id: 1,
row: Stock { quantity: 10 },
}],
})
.map_err(|(_, error)| error)?;
let attempt = authority
.prepare(Guarantee::LocallyDurable)?
.ok_or(Error::Busy)?;
assert!(matches!(
authority
.local
.table
.insert_with_id(1, Stock { quantity: 99 }),
super::super::super::WarpTableInsertWithIdOutcome::Inserted { .. }
));
assert_eq!(
authority.step(Complete {
attempt,
result: PhysicalResult::Durable
}),
Err(Error::Corrupt)
);
assert!(authority.poisoned);
assert_eq!(authority.local_revision(), Revision(0));
assert_eq!(authority.remote_revision(), Revision(0));
assert!(matches!(authority.checkpoint(), Err(Error::Corrupt)));
Ok(())
}
#[test]
fn checkpoint_refuses_an_identity_missing_from_its_primary_index() -> Result<(), Error> {
let mut authority = fresh()?;
assert!(matches!(
authority
.local
.insert_with_id(1, Stock { quantity: 10 })
.map_err(|_| Error::Exhausted)?,
super::super::super::WarpTableInsertWithIdOutcome::Inserted { .. }
));
authority.local.table.primary_index =
super::super::super::dense_key_index::DenseKeyIndex::try_new(2)
.map_err(|_| Error::Allocation)?;
assert!(matches!(authority.checkpoint(), Err(Error::Corrupt)));
assert_eq!(authority.local.table().ids(), &[1]);
Ok(())
}