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(())
}