use super::*;
use crate::storage::warp_table::observed::{ViewAdmission, ViewApplyOutcome, ViewDefinition};
use alloc::{format, string::String, vec::Vec};
use proptest::prelude::*;
#[derive(Clone, Eq, PartialEq)]
struct StockView;
impl ViewDefinition<Stock> for StockView {
type Value = u32;
fn project(&self, row: StockRef<'_>) -> Option<u32> {
Some(*row.quantity)
}
}
fn debug(error: impl core::fmt::Debug) -> String {
format!("{error:?}")
}
fn check(steps: &[(bool, u8, bool, bool, bool)]) -> Result<(), String> {
let mut authority = fresh(Replication::Required).map_err(debug)?;
authority.reserved(1024).map_err(debug)?;
let session = authority.open_session().map_err(debug)?;
let seed = authority.operation(session, 1);
submit(
&mut authority,
seed,
vec![
Change::Insert {
id: 1,
row: Stock { quantity: 20 },
},
Change::Insert {
id: 2,
row: Stock { quantity: 0 },
},
],
)
.map_err(debug)?;
flush(&mut authority, Guarantee::LocallyDurable).map_err(debug)?;
flush(&mut authority, Guarantee::ReplicatedDurable).map_err(debug)?;
authority.retire(seed).map_err(debug)?;
let mut subscriptions = Subscriptions::new(SubscriptionLimits {
views: limits().rows,
subscribers: limits().rows,
retained_changes: Capacity::new(NonZeroU32::MIN),
})
.map_err(debug)?;
let mut views = Vec::new();
for guarantee in [
Guarantee::Accepted,
Guarantee::LocallyDurable,
Guarantee::ReplicatedDurable,
] {
let id = subscriptions
.subscribe(&authority, StockView, guarantee)
.map_err(debug)?;
views.push((
guarantee,
id,
ContinuousView::<u32, u32>::new(limits().rows),
));
}
let mut pending = Vec::new();
let mut accepted = (20_u32, 0_u32);
let mut local = accepted;
let mut remote = accepted;
for (index, (direction, quantity, flush_local, flush_remote, deliver)) in
steps.iter().copied().enumerate()
{
let amount = u32::from(quantity).min(if direction { accepted.0 } else { accepted.1 });
accepted = if direction {
(
accepted.0.checked_sub(amount).ok_or("underflow")?,
accepted.1.checked_add(amount).ok_or("overflow")?,
)
} else {
(
accepted.0.checked_add(amount).ok_or("overflow")?,
accepted.1.checked_sub(amount).ok_or("underflow")?,
)
};
let sequence = u64::try_from(index)
.map_err(debug)?
.checked_add(2)
.ok_or("sequence")?;
let operation = authority.operation(session, sequence);
submit(
&mut authority,
operation,
vec![
Change::Replace {
id: 1,
row: Stock {
quantity: accepted.0,
},
},
Change::Replace {
id: 2,
row: Stock {
quantity: accepted.1,
},
},
],
)
.map_err(debug)?;
pending.push(operation);
let release = pending.len() == 4;
if flush_local || release {
flush(&mut authority, Guarantee::LocallyDurable).map_err(debug)?;
local = accepted;
}
if flush_remote || release {
flush(&mut authority, Guarantee::ReplicatedDurable).map_err(debug)?;
remote = local;
}
if release {
for operation in pending.drain(..) {
authority.retire(operation).map_err(debug)?;
}
}
if deliver || index.checked_add(1).ok_or("index")? == steps.len() {
subscriptions.refresh(&authority).map_err(debug)?;
for (guarantee, id, copy) in &mut views {
let result = subscriptions
.transmit(*id, |delivery| {
if let Some(update) = ViewUpdate::capture(delivery) {
assert_eq!(
copy.receive(&update).map_err(debug)?,
ViewApplyOutcome::Applied
);
}
Ok::<_, String>(ViewAdmission::Admitted)
})
.map_err(debug)?;
if let Some(result) = result {
result.map_err(debug)?;
}
let expected = match guarantee {
Guarantee::Accepted => accepted,
Guarantee::LocallyDurable => local,
Guarantee::ReplicatedDurable => remote,
};
let copy = copy.copy().ok_or("baseline missing")?;
assert_eq!(
(
*copy.get(1).ok_or("kitchen missing")?.value,
*copy.get(2).ok_or("counter missing")?.value
),
expected
);
assert_eq!(expected.0.checked_add(expected.1), Some(20));
}
}
}
Ok(())
}
proptest! {
#[test]
fn delayed_publications_reconstruct_atomic_transfers_at_each_guarantee(steps in prop::collection::vec((any::<bool>(),any::<u8>(),any::<bool>(),any::<bool>(),any::<bool>()),1..96)) {
check(&steps).map_err(TestCaseError::fail)?;
}
}
#[test]
fn receiver_requires_a_baseline_and_preserves_newer_rows_on_stale_traffic() -> Result<(), String> {
use crate::storage::warp_table::observed::*;
use core::num::NonZeroU64;
let first = ViewRevision {
source: SourceId::new(NonZeroU64::MIN),
view: ViewId::new(NonZeroU64::MIN),
epoch: 1,
revision: 1,
};
let next = ViewRevision {
revision: 2,
..first
};
let baseline = ViewUpdate::Baseline {
through: first,
rows: vec![ViewChange::Upsert {
id: 1,
incarnation: 1,
value: 10,
}],
};
let delta = ViewUpdate::Changes {
base: first,
through: next,
changes: vec![ViewChange::Upsert {
id: 1,
incarnation: 1,
value: 12,
}],
};
let mut copy = ContinuousView::<u32, u32>::new(limits().rows);
assert_eq!(copy.rows().count(), 0);
assert_eq!(copy.get(1), None);
assert_eq!(copy.connection(), ViewConnection::Disconnected);
copy.set_connection(ViewConnection::Connected);
assert_eq!(
copy.receive(&delta).map_err(debug)?,
ViewApplyOutcome::BaselineRequired
);
assert_eq!(
copy.reconnect(&delta).map_err(debug)?,
ViewApplyOutcome::BaselineRequired
);
assert_eq!(copy.status(), ContinuousViewStatus::AwaitingBaseline);
assert_eq!(
copy.receive(&baseline).map_err(debug)?,
ViewApplyOutcome::Applied
);
assert_eq!(
copy.receive(&delta).map_err(debug)?,
ViewApplyOutcome::Applied
);
assert_eq!(
copy.receive(&baseline).map_err(debug)?,
ViewApplyOutcome::Stale
);
assert_eq!(
copy.rows()
.map(|(id, value)| (id, *value))
.collect::<Vec<_>>(),
[(1, 12)]
);
assert_eq!(
copy.receive(&delta).map_err(debug)?,
ViewApplyOutcome::Stale
);
assert_eq!(
copy.reconnect(&delta).map_err(debug)?,
ViewApplyOutcome::BaselineRequired
);
assert_eq!(copy.get(1), Some(&12));
assert!(ViewUpdate::<u32, u32>::capture(ViewDelivery::Current { through: next }).is_none());
let mut oversized = ContinuousView::<u32, u32>::new(Capacity::new(NonZeroU32::MAX));
assert!(matches!(
oversized.receive(&baseline),
Err(ContinuousViewError::Creation(
ViewCreationError::CapacityOverflow
))
));
Ok(())
}
#[test]
fn released_subscription_ids_stay_unknown_and_failed_transport_retains_its_cursor()
-> Result<(), String> {
use crate::storage::warp_table::observed::ViewTransmitError;
let authority = fresh(Replication::Required).map_err(debug)?;
let mut subscriptions = Subscriptions::new(SubscriptionLimits {
views: limits().rows,
subscribers: limits().rows,
retained_changes: limits().rows,
})
.map_err(debug)?;
let id = subscriptions
.subscribe(&authority, StockView, Guarantee::Accepted)
.map_err(debug)?;
assert_eq!(subscriptions.ids().collect::<Vec<_>>(), [id]);
subscriptions.refresh(&authority).map_err(debug)?;
assert!(matches!(
subscriptions
.transmit(id, |_| Err::<ViewAdmission, _>(17))
.map_err(debug)?,
Some(Err(ViewTransmitError::Transport(17)))
));
let mut baseline = false;
subscriptions
.transmit(id, |delivery| {
baseline = matches!(
ViewUpdate::capture(delivery),
Some(ViewUpdate::Baseline { .. })
);
Ok::<_, String>(ViewAdmission::Admitted)
})
.map_err(debug)?
.ok_or("missing materialization")?
.map_err(debug)?;
assert!(baseline);
subscriptions.unsubscribe(id).map_err(debug)?;
assert_eq!(subscriptions.ids().count(), 0);
assert!(matches!(
subscriptions.unsubscribe(id),
Err(SubscriptionError::Unknown)
));
assert!(matches!(
subscriptions.require_baseline(id),
Err(SubscriptionError::Unknown)
));
assert!(matches!(
subscriptions.transmit(id, |_| Ok::<_, String>(ViewAdmission::Admitted)),
Err(SubscriptionError::Unknown)
));
let later = subscriptions
.subscribe(&authority, StockView, Guarantee::Accepted)
.map_err(debug)?;
assert_ne!(id, later);
subscriptions
.subscribe(&authority, StockView, Guarantee::ReplicatedDurable)
.map_err(debug)?;
let unreplicated = fresh(Replication::Disabled).map_err(debug)?;
assert!(matches!(
subscriptions.refresh(&unreplicated),
Err(SubscriptionError::Authority(Error::Unsupported))
));
Ok(())
}