use super::*;
use crate::storage::warp_table::persistent::{HistoryId, Limits, Recovery, Replication, Revision};
use alloc::{format, string::String, vec};
use core::num::NonZeroU32;
use proptest::prelude::*;
#[derive(Clone, 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
}
}
#[derive(Clone, Eq, PartialEq)]
struct AtLeast(u32);
impl ViewDefinition<Stock> for AtLeast {
type Value = u32;
fn project(&self, row: StockRef<'_>) -> Option<u32> {
(*row.quantity >= self.0).then_some(*row.quantity)
}
}
fn debug(error: impl core::fmt::Debug) -> String {
format!("{error:?}")
}
fn owner() -> Result<Authority<Stock>, String> {
let one = Capacity::new(NonZeroU32::MIN);
Authority::recover(
Recovery {
history: HistoryId([3; 16]),
revision: Revision(0),
reserved_through: 0,
rows: vec![(1, Stock { quantity: 10 })],
outcomes: vec![],
},
16,
Limits {
rows: one,
operations: one,
sessions: one,
changes: 1,
row_bytes: 4,
receipt_bytes: 0,
},
Replication::Disabled,
)
.map_err(debug)
}
fn subscriptions() -> Result<Subscriptions<Stock, AtLeast>, String> {
Subscriptions::new(SubscriptionLimits {
views: Capacity::new(NonZeroU32::MIN.saturating_add(1)),
subscribers: Capacity::new(NonZeroU32::MIN.saturating_add(2)),
retained_changes: Capacity::new(NonZeroU32::MIN),
})
.map_err(debug)
}
fn invariant(owner: &Subscriptions<Stock, AtLeast>) {
assert!(owner.views.len() <= 2);
assert!(owner.subscribers.len() <= 3);
for row in owner.views.rows() {
assert!(*row.id > 0 && *row.id <= owner.next);
let actual = owner
.subscribers
.rows()
.filter(|subscriber| subscriber.cursor.view == *row.id)
.count();
assert!(actual > 0);
assert_eq!(row.projection.subscribers, actual);
}
for subscriber in owner.subscribers.rows() {
assert!(*subscriber.id > 0 && *subscriber.id <= owner.next);
assert_eq!(
owner
.views
.rows()
.filter(|row| *row.id == subscriber.cursor.view)
.count(),
1
);
}
}
fn lifecycle(steps: &[(u8, u8, bool)]) -> Result<(), String> {
let authority = owner()?;
let mut owner = subscriptions()?;
let mut handles = Vec::new();
invariant(&owner);
for &(action, selector, admit) in steps {
match action % 4 {
0 => {
let old = owner.ids().collect::<Vec<_>>();
match owner.subscribe(
&authority,
AtLeast(u32::from(selector % 3)),
Guarantee::Accepted,
) {
Ok(id) => {
assert!(!handles.contains(&id));
handles.push(id);
}
Err(SubscriptionError::Full) => {
assert_eq!(owner.ids().collect::<Vec<_>>(), old)
}
Err(other) => return Err(debug(other)),
}
}
operation @ 1..=3 => {
if let Some(id) = handles
.get(
usize::from(selector)
.checked_rem(handles.len().max(1))
.ok_or("handle bound")?,
)
.copied()
{
let exists = owner.ids().any(|actual| actual == id);
match operation {
1 => {
let result = owner.unsubscribe(id);
if exists {
result.map_err(debug)?;
} else {
assert!(matches!(result, Err(SubscriptionError::Unknown)));
}
}
2 => {
let result = owner.require_baseline(id);
if exists {
result.map_err(debug)?;
} else {
assert!(matches!(result, Err(SubscriptionError::Unknown)));
}
}
3 => {
owner.refresh(&authority).map_err(debug)?;
let result = owner.transmit(id, |_| {
Ok::<_, String>(if admit {
ViewAdmission::Admitted
} else {
ViewAdmission::Backpressured
})
});
if exists {
result
.map_err(debug)?
.ok_or("missing view")?
.map_err(debug)?;
} else {
assert!(matches!(result, Err(SubscriptionError::Unknown)));
}
}
_ => return Err("invalid test operation".into()),
}
}
}
_ => return Err("invalid test action".into()),
}
invariant(&owner);
}
for id in owner.ids().collect::<Vec<_>>() {
owner.unsubscribe(id).map_err(debug)?;
invariant(&owner);
}
assert!(owner.views.is_empty());
Ok(())
}
proptest! {
#[test]
fn initialization_and_each_subscription_step_preserve_bounds_and_exact_ownership(steps in prop::collection::vec((any::<u8>(),any::<u8>(),any::<bool>()),0..96)) {
lifecycle(&steps).map_err(TestCaseError::fail)?;
}
}
#[test]
fn identifier_exhaustion_never_reuses_a_released_identity() -> Result<(), String> {
let authority = owner()?;
let mut owner = subscriptions()?;
owner.next = u64::MAX.checked_sub(1).ok_or("boundary")?;
let id = owner
.subscribe(&authority, AtLeast(0), Guarantee::Accepted)
.map_err(debug)?;
assert_eq!(id, SubscriptionId(u64::MAX));
owner.unsubscribe(id).map_err(debug)?;
assert!(matches!(
owner.subscribe(&authority, AtLeast(0), Guarantee::Accepted),
Err(SubscriptionError::Exhausted)
));
invariant(&owner);
Ok(())
}
#[test]
fn missing_view_index_bindings_refuse_subscription_and_refresh() -> Result<(), String> {
let authority = owner()?;
let mut subscriptions = subscriptions()?;
let id = subscriptions
.subscribe(&authority, AtLeast(0), Guarantee::Accepted)
.map_err(debug)?;
subscriptions.views.primary_index =
crate::storage::warp_table::dense_key_index::DenseKeyIndex::try_new(2)
.map_err(|_| "index storage unavailable")?;
assert!(matches!(
subscriptions.subscribe(&authority, AtLeast(0), Guarantee::Accepted),
Err(SubscriptionError::Corrupt)
));
assert!(matches!(
subscriptions.refresh(&authority),
Err(SubscriptionError::Corrupt)
));
assert_eq!(subscriptions.ids().collect::<Vec<_>>(), [id]);
assert_eq!(subscriptions.next, id.0);
assert!(matches!(
subscriptions.remove_unused_view(id.0),
Err(SubscriptionError::Corrupt)
));
assert_eq!(subscriptions.ids().collect::<Vec<_>>(), [id]);
Ok(())
}
#[test]
fn corrupted_identity_and_storage_bounds_cannot_admit_a_new_subscriber() -> Result<(), String> {
let authority = owner()?;
let mut subscriptions = subscriptions()?;
let id = subscriptions
.subscribe(&authority, AtLeast(0), Guarantee::Accepted)
.map_err(debug)?;
subscriptions.next = 0;
assert!(matches!(
subscriptions.subscribe(&authority, AtLeast(1), Guarantee::Accepted),
Err(SubscriptionError::Corrupt)
));
assert_eq!(subscriptions.ids().collect::<Vec<_>>(), [id]);
assert_eq!(subscriptions.views.len(), 1);
subscriptions.next = id.0;
subscriptions.subscribers.maximum_rows = 1;
assert!(matches!(
subscriptions.subscribe(&authority, AtLeast(1), Guarantee::Accepted),
Err(SubscriptionError::Corrupt)
));
assert_eq!(subscriptions.ids().collect::<Vec<_>>(), [id]);
assert_eq!(subscriptions.next, id.0);
Ok(())
}
#[test]
fn pending_replication_defers_materialization_without_sending() -> Result<(), String> {
let authority = owner()?;
let replicated = Authority::recover(
authority.checkpoint().map_err(debug)?,
32,
authority.limits(),
Replication::Required,
)
.map_err(debug)?;
let mut subscriptions = subscriptions()?;
let id = subscriptions
.subscribe(&replicated, AtLeast(0), Guarantee::ReplicatedDurable)
.map_err(debug)?;
subscriptions.refresh(&replicated).map_err(debug)?;
let mut calls = 0;
assert!(
subscriptions
.transmit(id, |_| {
calls += 1;
Ok::<_, ()>(ViewAdmission::Admitted)
})
.map_err(debug)?
.is_none()
);
assert_eq!(calls, 0);
invariant(&subscriptions);
Ok(())
}
#[test]
fn unsupported_durability_is_refused_on_subscription_and_refresh() -> Result<(), String> {
let authority = owner()?;
let mut subscriptions = subscriptions()?;
assert!(matches!(
subscriptions.subscribe(&authority, AtLeast(0), Guarantee::ReplicatedDurable),
Err(SubscriptionError::Authority(Error::Unsupported))
));
assert!(subscriptions.ids().next().is_none());
assert!(subscriptions.views.is_empty());
let replicated = Authority::recover(
authority.checkpoint().map_err(debug)?,
32,
authority.limits(),
Replication::Required,
)
.map_err(debug)?;
let id = subscriptions
.subscribe(&replicated, AtLeast(0), Guarantee::ReplicatedDurable)
.map_err(debug)?;
assert!(matches!(
subscriptions.refresh(&authority),
Err(SubscriptionError::Authority(Error::Unsupported))
));
assert_eq!(subscriptions.ids().collect::<Vec<_>>(), [id]);
assert_eq!(subscriptions.capacity().value().get(), 3);
assert!(matches!(
subscriptions.transmit(id, |_| Err::<ViewAdmission, _>("unexpected send")),
Ok(None)
));
subscriptions.unsubscribe(id).map_err(debug)?;
invariant(&subscriptions);
Ok(())
}
#[test]
fn missing_view_is_reported_without_sending_a_partial_delivery() -> Result<(), String> {
let authority = owner()?;
let mut subscriptions = subscriptions()?;
let id = subscriptions
.subscribe(&authority, AtLeast(0), Guarantee::Accepted)
.map_err(debug)?;
assert!(matches!(
subscriptions.views.remove(id.0),
WarpTableRemoveOutcome::Removed { .. }
));
assert!(matches!(
subscriptions.transmit(id, |_| Err::<ViewAdmission, _>("unexpected send")),
Err(SubscriptionError::Corrupt)
));
assert_eq!(subscriptions.ids().collect::<Vec<_>>(), [id]);
assert!(matches!(
subscriptions.unsubscribe(id),
Err(SubscriptionError::Corrupt)
));
assert!(subscriptions.ids().next().is_none());
Ok(())
}