use super::super::{
Capacity, WarpTable, WarpTableGetOutcome, WarpTableInsertWithIdOutcome, WarpTableRow,
};
use super::*;
use alloc::vec::Vec;
use core::num::{NonZeroU32, NonZeroU64};
use proptest::prelude::*;
#[derive(Debug, WarpTableRow, Eq, PartialEq)]
#[index(primary = u32)]
struct Actor {
#[cfg_attr(feature = "secondary-indexes", index(secondary))]
room: u8,
position: u32,
private: u32,
}
#[derive(WarpTableRow)]
struct Generated {
value: u32,
}
struct InRoom(u8);
impl ViewDefinition<Actor> for InRoom {
type Value = u32;
fn project(&self, row: ActorRef<'_>) -> Option<u32> {
(*row.room == self.0).then_some(*row.position)
}
}
fn source(capacity: u32) -> Result<ObservedTable<Actor>, &'static str> {
let table = WarpTable::try_new(Capacity::new(
NonZeroU32::new(capacity).ok_or("zero bound")?,
))
.map_err(|_| "table unavailable")?;
table
.observe(SourceId::new(NonZeroU64::MIN))
.map_err(|_| "observation unavailable")
}
fn view(source: &ObservedTable<Actor>) -> Result<MaintainedView<Actor, InRoom>, &'static str> {
source
.maintain(ViewId::new(NonZeroU64::MIN), InRoom(1))
.map_err(|_| "view unavailable")
}
fn insert(
source: &mut ObservedTable<Actor>,
id: u32,
room: u8,
position: u32,
) -> Result<(), &'static str> {
match source
.insert_with_id(
id,
Actor {
room,
position,
private: 0,
},
)
.map_err(|_| "sequence exhausted")?
{
WarpTableInsertWithIdOutcome::Inserted { .. } => Ok(()),
WarpTableInsertWithIdOutcome::AlreadyPresent { .. } => Err("already present"),
WarpTableInsertWithIdOutcome::Full { .. } => Err("full"),
WarpTableInsertWithIdOutcome::UniqueConflict { conflict, .. } => match conflict {},
}
}
#[test]
fn coherent_publications_suppress_irrelevant_and_repeated_edits() -> Result<(), &'static str> {
let mut source = source(8)?;
insert(&mut source, 1, 1, 10)?;
let mut view = view(&source)?;
let initial = view.revision();
assert!(matches!(
source
.update(1, |mut row| *row.private_mut() = 77)
.map_err(|_| "edit refused")?,
super::super::WarpTableUpdateOutcome::Updated { .. }
));
assert_eq!(view.refresh(&source), Ok(ViewRefreshOutcome::Unchanged));
assert_eq!(view.revision(), initial);
assert!(matches!(
source
.update(1, |mut row| *row.position_mut() = 11)
.map_err(|_| "edit refused")?,
super::super::WarpTableUpdateOutcome::Updated { .. }
));
assert!(matches!(
source
.update(1, |mut row| *row.position_mut() = 12)
.map_err(|_| "edit refused")?,
super::super::WarpTableUpdateOutcome::Updated { .. }
));
assert_eq!(view.rows().map(|row| *row.value).collect::<Vec<_>>(), [10]);
assert_eq!(
view.refresh(&source),
Ok(ViewRefreshOutcome::Published {
rows: 1,
changes: 1
})
);
let ViewPublication::Changes {
base,
through,
changes,
} = view.publication()
else {
return Err("missing changes");
};
assert_eq!(base, initial);
assert_eq!(through.revision, 1);
assert_eq!(
changes,
[ViewChange::Upsert {
id: 1,
incarnation: 1,
value: 12
}]
);
Ok(())
}
#[test]
fn reinsertion_changes_incarnation_and_relocation_preserves_it() -> Result<(), &'static str> {
let mut source = source(8)?;
insert(&mut source, 1, 1, 10)?;
insert(&mut source, 2, 1, 20)?;
let mut view = view(&source)?;
assert!(matches!(
source.remove(1).map_err(|_| "remove refused")?,
super::super::WarpTableRemoveOutcome::Removed { .. }
));
insert(&mut source, 1, 1, 10)?;
assert_eq!(
view.refresh(&source),
Ok(ViewRefreshOutcome::Published {
rows: 2,
changes: 1
})
);
let row = view.rows().find(|row| row.id == 1).ok_or("row missing")?;
assert_eq!((row.incarnation, *row.value), (4, 10));
let other = view.rows().find(|row| row.id == 2).ok_or("other missing")?;
assert_eq!(other.incarnation, 2);
Ok(())
}
#[test]
fn insert_then_remove_before_publication_is_silent() -> Result<(), &'static str> {
let mut source = source(8)?;
let mut view = view(&source)?;
insert(&mut source, 1, 1, 10)?;
assert!(matches!(
source.remove(1).map_err(|_| "remove refused")?,
super::super::WarpTableRemoveOutcome::Removed { .. }
));
assert_eq!(view.refresh(&source), Ok(ViewRefreshOutcome::Unchanged));
assert!(view.is_empty());
Ok(())
}
#[test]
fn source_history_overflow_rebuilds_the_diff_without_forcing_remote_baseline()
-> Result<(), &'static str> {
let mut source = source(2)?;
insert(&mut source, 1, 1, 10)?;
let mut view = view(&source)?;
for position in 11..20 {
assert!(matches!(
source
.update(1, |mut row| *row.position_mut() = position)
.map_err(|_| "edit refused")?,
super::super::WarpTableUpdateOutcome::Updated { .. }
));
}
assert_eq!(source.retained_changes(), 2);
assert_eq!(
view.refresh(&source),
Ok(ViewRefreshOutcome::Published {
rows: 1,
changes: 1
})
);
assert_eq!(view.rows().map(|row| *row.value).collect::<Vec<_>>(), [19]);
Ok(())
}
#[test]
fn changing_parameters_replaces_membership_without_source_edits() -> Result<(), &'static str> {
let mut source = source(4)?;
insert(&mut source, 1, 1, 10)?;
insert(&mut source, 2, 2, 20)?;
let mut view = view(&source)?;
assert!(matches!(
view.replace_definition(&source, InRoom(2)),
Ok(ViewRefreshOutcome::Baseline { rows: 1 })
));
assert_eq!(view.revision().epoch, 2);
assert_eq!(
view.rows()
.map(|row| (row.id, *row.value))
.collect::<Vec<_>>(),
[(2, 20)]
);
Ok(())
}
#[cfg(feature = "secondary-indexes")]
#[test]
fn indexed_bulk_edits_are_observed() -> Result<(), &'static str> {
let mut source = source(4)?;
insert(&mut source, 1, 1, 10)?;
insert(&mut source, 2, 2, 20)?;
let mut view = view(&source)?;
assert_eq!(
source.update_selected(
|table| table.query().find_room(1).into(),
|mut row| {
*row.position_mut() = 50;
}
),
Ok(1)
);
assert_eq!(
view.refresh(&source),
Ok(ViewRefreshOutcome::Published {
rows: 1,
changes: 1
})
);
assert_eq!(view.rows().map(|row| *row.value).collect::<Vec<_>>(), [50]);
Ok(())
}
proptest! {
#[test]
fn maintained_projection_matches_fresh_evaluation(operations in prop::collection::vec((0_u8..5, 0_u32..12, any::<u32>()), 0..160)) {
let mut source = source(12).map_err(TestCaseError::fail)?;
let mut view = view(&source).map_err(TestCaseError::fail)?;
for (operation, id, value) in operations {
match operation {
0 => { let outcome = source.upsert(id, Actor { room: u8::from(value % 2 == 0), position: value, private: 0 }); prop_assert!(outcome.is_ok()); }
1 => { prop_assert!(source.remove(id).is_ok()); }
2 => { prop_assert!(source.update(id, |mut row| *row.position_mut() = value).is_ok()); }
3 => { prop_assert!(source.update(id, |mut row| *row.private_mut() = value).is_ok()); }
4 => {
prop_assert!(view.refresh(&source).is_ok());
let mut fresh: Vec<_> = source.table().rows().filter(|row| *row.room == 1).map(|row| (*row.id, *row.position)).collect();
let mut maintained: Vec<_> = view.rows().map(|row| (row.id, *row.value)).collect();
fresh.sort_unstable(); maintained.sort_unstable();
prop_assert_eq!(maintained, fresh);
}
_ => return Err(TestCaseError::fail("generator outside range")),
}
}
}
}
#[test]
fn rejected_insertion_preserves_owned_row_and_observed_state() -> Result<(), &'static str> {
let mut source = source(1)?;
insert(&mut source, 1, 1, 10)?;
let mut view = view(&source)?;
let rejected = Actor {
room: 2,
position: 20,
private: 30,
};
let outcome = source
.insert_with_id(2, rejected)
.map_err(|_| "sequence refused")?;
assert_eq!(
outcome,
WarpTableInsertWithIdOutcome::Full {
id: 2,
rejected: Actor {
room: 2,
position: 20,
private: 30
}
}
);
assert_eq!(view.refresh(&source), Ok(ViewRefreshOutcome::Unchanged));
assert!(matches!(
source.table().get(2),
WarpTableGetOutcome::Absent { id: 2 }
));
Ok(())
}
fn replica() -> Result<ViewReplica<u32, u32>, &'static str> {
ViewReplica::try_new(
SourceId::new(NonZeroU64::MIN),
ViewId::new(NonZeroU64::MIN),
Capacity::new(NonZeroU32::new(4).ok_or("zero")?),
)
.map_err(|_| "replica allocation")
}
fn stamp(epoch: u64, revision: u64) -> ViewRevision {
ViewRevision {
source: SourceId::new(NonZeroU64::MIN),
view: ViewId::new(NonZeroU64::MIN),
epoch,
revision,
}
}
fn baseline(
replica: &mut ViewReplica<u32, u32>,
revision: ViewRevision,
rows: &[ViewChange<u32, u32>],
) -> Result<(), &'static str> {
assert_eq!(
replica.begin_baseline(revision, rows.len()),
Ok(ViewApplyOutcome::BaselineRequired)
);
replica
.baseline_chunk(revision, rows)
.map_err(|_| "baseline chunk")?;
assert_eq!(
replica.finish_baseline(revision),
Ok(ViewApplyOutcome::Applied)
);
Ok(())
}
#[test]
fn incomplete_baseline_does_not_replace_last_coherent_view() -> Result<(), &'static str> {
let mut replica = replica()?;
assert_eq!(replica.status(), ViewReplicaStatus::AwaitingBaseline);
let first = [ViewChange::Upsert {
id: 1,
incarnation: 1,
value: 10,
}];
baseline(&mut replica, stamp(1, 0), &first)?;
assert_eq!(
replica.begin_baseline(stamp(2, 0), 2),
Ok(ViewApplyOutcome::BaselineRequired)
);
replica
.baseline_chunk(
stamp(2, 0),
&[ViewChange::Upsert {
id: 2,
incarnation: 2,
value: 20,
}],
)
.map_err(|_| "chunk")?;
assert_eq!(
replica.finish_baseline(stamp(2, 0)),
Err(ViewApplyError::BaselineIncomplete)
);
assert_eq!(
replica.rows().map(|row| *row.value).collect::<Vec<_>>(),
[10]
);
replica
.baseline_chunk(
stamp(2, 0),
&[ViewChange::Upsert {
id: 3,
incarnation: 3,
value: 30,
}],
)
.map_err(|_| "chunk")?;
assert_eq!(
replica.finish_baseline(stamp(2, 0)),
Ok(ViewApplyOutcome::Applied)
);
assert_eq!(
replica.rows().map(|row| *row.value).collect::<Vec<_>>(),
[20, 30]
);
Ok(())
}
#[test]
fn rejected_increment_is_atomic_and_requires_recovery() -> Result<(), &'static str> {
let mut replica = replica()?;
baseline(
&mut replica,
stamp(1, 0),
&[ViewChange::Upsert {
id: 1,
incarnation: 10,
value: 10,
}],
)?;
let changes = [
ViewChange::Upsert {
id: 1,
incarnation: 10,
value: 20,
},
ViewChange::Remove {
id: 2,
incarnation: 2,
},
];
assert_eq!(
replica.apply_changes(stamp(1, 0), stamp(1, 1), &changes),
Err(ViewApplyError::UnknownRemoval)
);
assert_eq!(
replica.rows().map(|row| *row.value).collect::<Vec<_>>(),
[10]
);
assert_eq!(
replica.status(),
ViewReplicaStatus::BaselineRequired {
through: stamp(1, 0),
required: stamp(1, 1)
}
);
assert_eq!(
replica.apply_changes(
stamp(1, 0),
stamp(1, 1),
changes.get(..1).ok_or("first change")?
),
Ok(ViewApplyOutcome::BaselineRequired)
);
Ok(())
}
#[test]
fn stale_incarnations_and_duplicate_identities_are_rejected() -> Result<(), &'static str> {
for changes in [
alloc::vec![ViewChange::Upsert {
id: 1,
incarnation: 9,
value: 20
}],
alloc::vec![ViewChange::Remove {
id: 1,
incarnation: 9
}],
alloc::vec![
ViewChange::Upsert {
id: 1,
incarnation: 10,
value: 20
},
ViewChange::Upsert {
id: 1,
incarnation: 11,
value: 30
}
],
] {
let mut replica = replica()?;
baseline(
&mut replica,
stamp(1, 0),
&[ViewChange::Upsert {
id: 1,
incarnation: 10,
value: 10,
}],
)?;
assert!(
replica
.apply_changes(stamp(1, 0), stamp(1, 1), &changes)
.is_err()
);
assert_eq!(
replica.rows().map(|row| *row.value).collect::<Vec<_>>(),
[10]
);
}
Ok(())
}
#[test]
fn view_continuity_is_independent_of_source_edit_sequence() -> Result<(), &'static str> {
let mut source = source(4)?;
insert(&mut source, 1, 1, 10)?;
let mut view = view(&source)?;
let mut replica = replica()?;
let ViewPublication::Baseline { through, rows } = view.publication() else {
return Err("baseline");
};
baseline(&mut replica, through, rows)?;
for value in 0..30 {
assert!(
source
.update(1, |mut row| *row.private_mut() = value)
.is_ok()
);
assert_eq!(view.refresh(&source), Ok(ViewRefreshOutcome::Unchanged));
}
assert!(source.update(1, |mut row| *row.position_mut() = 20).is_ok());
assert_eq!(
view.refresh(&source),
Ok(ViewRefreshOutcome::Published {
rows: 1,
changes: 1
})
);
let ViewPublication::Changes {
base,
through,
changes,
} = view.publication()
else {
return Err("changes");
};
assert_eq!(
replica.apply_changes(base, through, changes),
Ok(ViewApplyOutcome::Applied)
);
assert_eq!(
replica.apply_changes(base, through, changes),
Ok(ViewApplyOutcome::Stale)
);
assert_eq!(
replica.rows().map(|row| *row.value).collect::<Vec<_>>(),
[20]
);
assert_eq!(replica.connection(), ViewConnection::Disconnected);
replica.set_connection(ViewConnection::Connected);
assert_eq!(replica.connection(), ViewConnection::Connected);
Ok(())
}
#[test]
fn delayed_subscribers_use_retained_history_while_current_ones_share_latest_changes()
-> Result<(), &'static str> {
let mut source = source(4)?;
insert(&mut source, 1, 1, 10)?;
let mut view = view(&source)?;
let slow = view.revision();
let mut buffer = view.delivery_buffer().map_err(|_| "buffer")?;
assert!(source.update(1, |mut row| *row.position_mut() = 20).is_ok());
assert!(view.refresh(&source).is_ok());
let fast = view.revision();
assert!(source.update(1, |mut row| *row.position_mut() = 30).is_ok());
assert!(view.refresh(&source).is_ok());
assert!(matches!(
view.delivery(Some(slow), &mut buffer),
Ok(ViewDelivery::Changes { .. })
));
assert!(matches!(
view.delivery(Some(fast), &mut buffer),
Ok(ViewDelivery::Changes { .. })
));
assert_eq!(view.refresh(&source), Ok(ViewRefreshOutcome::Unchanged));
assert!(matches!(
view.delivery(Some(fast), &mut buffer),
Ok(ViewDelivery::Changes { .. })
));
assert!(matches!(
view.delivery(Some(view.revision()), &mut buffer),
Ok(ViewDelivery::Current { .. })
));
Ok(())
}
struct RoomCount(u8);
impl ViewDefinition<Actor> for RoomCount {
type Value = ();
fn project(&self, row: ActorRef<'_>) -> Option<()> {
(*row.room == self.0).then_some(())
}
}
#[test]
fn maintained_count_suppresses_membership_churn_without_cardinality_change()
-> Result<(), &'static str> {
let mut source = source(4)?;
insert(&mut source, 1, 1, 10)?;
let mut count = source
.maintain_count(ViewId::new(NonZeroU64::MIN), RoomCount(1))
.map_err(|_| "count")?;
assert_eq!(count.publication().count, 1);
assert!(source.remove(1).is_ok());
insert(&mut source, 2, 1, 20)?;
assert_eq!(count.refresh(&source), Ok(CountRefreshOutcome::Unchanged));
insert(&mut source, 3, 1, 30)?;
assert!(matches!(
count.refresh(&source),
Ok(CountRefreshOutcome::Published(CountPublication {
count: 2,
..
}))
));
assert_eq!(count.publication().through.revision, 1);
assert!(matches!(
count.replace_selection(&source, RoomCount(2)),
Ok(CountRefreshOutcome::Published(CountPublication {
count: 0,
..
}))
));
assert_eq!(count.publication().through.epoch, 2);
Ok(())
}
#[test]
fn one_dirty_identity_is_projected_once_without_rescanning_other_rows() -> Result<(), &'static str>
{
struct Counting<'a>(&'a core::cell::Cell<usize>);
impl ViewDefinition<Actor> for Counting<'_> {
type Value = u32;
fn project(&self, row: ActorRef<'_>) -> Option<u32> {
self.0.set(self.0.get().checked_add(1)?);
Some(*row.position)
}
}
let calls = core::cell::Cell::new(0);
let mut source = source(1024)?;
for id in 0..1024 {
insert(&mut source, id, 1, 0)?;
}
let mut view = source
.maintain(ViewId::new(NonZeroU64::MIN), Counting(&calls))
.map_err(|_| "view")?;
assert_eq!(calls.get(), 1024);
calls.set(0);
for value in 1..10 {
assert!(
source
.update(77, |mut row| *row.position_mut() = value)
.is_ok()
);
}
assert!(view.refresh(&source).is_ok());
assert_eq!(calls.get(), 1);
calls.set(0);
assert_eq!(view.refresh(&source), Ok(ViewRefreshOutcome::Unchanged));
assert_eq!(calls.get(), 0);
Ok(())
}
#[test]
fn filtered_private_views_never_publish_excluded_rows() -> Result<(), &'static str> {
struct Private(u32);
impl ViewDefinition<Actor> for Private {
type Value = u32;
fn project(&self, row: ActorRef<'_>) -> Option<u32> {
(*row.id == self.0).then_some(*row.private)
}
}
let mut source = source(4)?;
insert(&mut source, 1, 1, 10)?;
insert(&mut source, 2, 1, 20)?;
let mut view = source
.maintain(ViewId::new(NonZeroU64::MIN), Private(1))
.map_err(|_| "private view")?;
assert_eq!(view.rows().map(|row| row.id).collect::<Vec<_>>(), [1]);
assert!(source.update(2, |mut row| *row.private_mut() = 999).is_ok());
assert_eq!(view.refresh(&source), Ok(ViewRefreshOutcome::Unchanged));
assert!(source.update(1, |mut row| *row.private_mut() = 100).is_ok());
assert_eq!(
view.refresh(&source),
Ok(ViewRefreshOutcome::Published {
rows: 1,
changes: 1
})
);
let ViewPublication::Changes { changes, .. } = view.publication() else {
return Err("changes");
};
assert_eq!(
changes,
[ViewChange::Upsert {
id: 1,
incarnation: 1,
value: 100
}]
);
Ok(())
}
#[test]
fn exhausted_sequences_refuse_edits_without_losing_inputs() -> Result<(), &'static str> {
let mut source = source(2)?;
insert(&mut source, 1, 1, 10)?;
source.sequence = u64::MAX;
let row = Actor {
room: 1,
position: 20,
private: 0,
};
let Err(ObservationEditError::SequenceExhausted { rejected }) = source.insert_with_id(2, row)
else {
return Err("sequence accepted");
};
assert_eq!(rejected.position, 20);
let Err(ObservationEditError::SequenceExhausted { rejected: id }) = source.remove(1) else {
return Err("remove accepted");
};
assert_eq!(id, 1);
let calls = core::cell::Cell::new(0);
assert!(source.update(1, |_| calls.set(1)).is_err());
assert_eq!(calls.get(), 0);
assert_eq!(source.table().len(), 1);
Ok(())
}
#[test]
fn disconnected_readers_keep_the_last_coherent_projection() -> Result<(), &'static str> {
let mut replica = replica()?;
baseline(
&mut replica,
stamp(1, 0),
&[ViewChange::Upsert {
id: 1,
incarnation: 1,
value: 10,
}],
)?;
replica.set_connection(ViewConnection::Connected);
replica.set_connection(ViewConnection::Disconnected);
assert_eq!(
replica.rows().map(|row| *row.value).collect::<Vec<_>>(),
[10]
);
assert_eq!(
replica.status(),
ViewReplicaStatus::Current {
through: stamp(1, 0)
}
);
assert_eq!(replica.connection(), ViewConnection::Disconnected);
Ok(())
}
#[test]
fn remote_coalescing_is_relative_to_receiver_state_and_preserves_reincarnation()
-> Result<(), &'static str> {
let mut source = source(16)?;
insert(&mut source, 1, 1, 10)?;
let mut view = view(&source)?;
let mut buffer = view.delivery_buffer().map_err(|_| "buffer")?;
let base = view.revision();
let mut replica = replica()?;
let ViewPublication::Baseline { through, rows } = view.publication() else {
return Err("baseline");
};
baseline(&mut replica, through, rows)?;
for value in [20, 30] {
assert!(
source
.update(1, |mut row| *row.position_mut() = value)
.is_ok()
);
assert!(view.refresh(&source).is_ok());
}
insert(&mut source, 2, 1, 42)?;
assert!(view.refresh(&source).is_ok());
assert!(source.remove(2).is_ok());
assert!(view.refresh(&source).is_ok());
assert!(source.remove(1).is_ok());
assert!(view.refresh(&source).is_ok());
insert(&mut source, 1, 1, 30)?;
assert!(view.refresh(&source).is_ok());
let ViewDelivery::Changes {
base: actual_base,
through,
changes,
} = view
.delivery(Some(base), &mut buffer)
.map_err(|_| "delivery")?
else {
return Err("coalesced changes");
};
assert_eq!(actual_base, base);
assert_eq!(
changes,
[ViewChange::Upsert {
id: 1,
incarnation: 7,
value: 30
}]
);
assert_eq!(
replica.apply_changes(base, through, changes),
Ok(ViewApplyOutcome::Applied)
);
assert_eq!(
replica
.rows()
.map(|row| (row.id, row.incarnation, *row.value))
.collect::<Vec<_>>(),
[(1, 7, 30)]
);
Ok(())
}
#[test]
fn remote_coalescing_can_publish_an_empty_span_when_values_return_to_baseline()
-> Result<(), &'static str> {
let mut source = source(8)?;
insert(&mut source, 1, 1, 10)?;
let mut view = view(&source)?;
let base = view.revision();
let mut buffer = view.delivery_buffer().map_err(|_| "buffer")?;
for value in [20, 10] {
assert!(
source
.update(1, |mut row| *row.position_mut() = value)
.is_ok()
);
assert!(view.refresh(&source).is_ok());
}
let ViewDelivery::Changes {
changes, through, ..
} = view
.delivery(Some(base), &mut buffer)
.map_err(|_| "delivery")?
else {
return Err("span");
};
assert!(changes.is_empty());
assert_eq!(through.revision, 2);
Ok(())
}
#[test]
fn remote_retention_eviction_and_parameter_replacement_require_baselines()
-> Result<(), &'static str> {
let mut source = source(8)?;
insert(&mut source, 1, 1, 10)?;
let mut view = source
.maintain_with_retention(
ViewId::new(NonZeroU64::MIN),
InRoom(1),
Capacity::new(NonZeroU32::MIN),
)
.map_err(|_| "view")?;
let base = view.revision();
let mut buffer = view.delivery_buffer().map_err(|_| "buffer")?;
for value in [20, 30] {
assert!(
source
.update(1, |mut row| *row.position_mut() = value)
.is_ok()
);
assert!(view.refresh(&source).is_ok());
}
assert_eq!(view.retained_changes(), 1);
assert!(matches!(
view.delivery(Some(base), &mut buffer),
Ok(ViewDelivery::Baseline { .. })
));
let before_parameter_change = view.revision();
assert!(view.replace_definition(&source, InRoom(2)).is_ok());
assert!(matches!(
view.delivery(Some(before_parameter_change), &mut buffer),
Ok(ViewDelivery::Baseline { .. })
));
let mut future = view.revision();
future.revision = 99;
assert!(matches!(
view.delivery(Some(future), &mut buffer),
Err(ViewDeliveryError::FutureCursor)
));
Ok(())
}
proptest! {
#[test]
fn remote_reader_converges_after_arbitrary_delivery_delays(
operations in prop::collection::vec((0_u8..4, 0_u32..4, any::<u32>(), any::<bool>()), 0..160)
) {
let mut source = source(4).map_err(TestCaseError::fail)?;
let mut view = view(&source).map_err(TestCaseError::fail)?;
let mut buffer = view.delivery_buffer().map_err(|_| TestCaseError::fail("buffer"))?;
let mut replica = replica().map_err(TestCaseError::fail)?;
let mut cursor = None;
for (operation, id, value, deliver) in operations {
match operation {
0 => { let outcome = source.upsert(id, Actor { room: u8::from(value % 2 == 0), position: value, private: 0 }); prop_assert!(outcome.is_ok()); }
1 => { prop_assert!(source.remove(id).is_ok()); }
2 => { prop_assert!(source.update(id, |mut row| *row.position_mut() = value).is_ok()); }
3 => { prop_assert!(source.update(id, |mut row| *row.private_mut() = value).is_ok()); }
_ => return Err(TestCaseError::fail("generator outside range")),
}
prop_assert!(view.refresh(&source).is_ok());
if deliver {
match view.delivery(cursor, &mut buffer).map_err(|_| TestCaseError::fail("delivery"))? {
ViewDelivery::Current { through } => cursor = Some(through),
ViewDelivery::Changes { base, through, changes } => {
prop_assert_eq!(replica.apply_changes(base, through, changes), Ok(ViewApplyOutcome::Applied));
cursor = Some(through);
}
ViewDelivery::Baseline { through, rows } => {
let changes: Vec<_> = rows.map(|row| ViewChange::Upsert { id: row.id, incarnation: row.incarnation, value: *row.value }).collect();
baseline(&mut replica, through, &changes).map_err(TestCaseError::fail)?;
cursor = Some(through);
}
}
let mut expected: Vec<_> = view.rows().map(|row| (row.id, row.incarnation, *row.value)).collect();
let mut actual: Vec<_> = replica.rows().map(|row| (row.id, row.incarnation, *row.value)).collect();
expected.sort_unstable(); actual.sort_unstable();
prop_assert_eq!(actual, expected);
}
}
}
}
#[test]
fn subscriber_advances_only_after_owned_transport_admission() -> Result<(), &'static str> {
let mut source = source(8)?;
insert(&mut source, 1, 1, 10)?;
let mut view = view(&source)?;
let initial = view.revision();
let mut subscriber = view.subscribe();
let mut buffer = view.delivery_buffer().map_err(|_| "buffer")?;
assert_eq!(subscriber.admitted_through(), None);
assert_eq!(
view.transmit(&mut subscriber, &mut buffer, |delivery| {
assert!(matches!(delivery, ViewDelivery::Baseline { .. }));
Ok::<_, ()>(ViewAdmission::Backpressured)
}),
Ok(ViewTransmitOutcome::Backpressured { through: initial })
);
assert_eq!(subscriber.admitted_through(), None);
assert_eq!(
view.transmit(&mut subscriber, &mut buffer, |_| Err("closed")),
Err(ViewTransmitError::Transport("closed"))
);
assert_eq!(subscriber.admitted_through(), None);
assert_eq!(
view.transmit(&mut subscriber, &mut buffer, |_| Ok::<_, ()>(
ViewAdmission::Admitted
)),
Ok(ViewTransmitOutcome::Admitted { through: initial })
);
assert_eq!(subscriber.admitted_through(), Some(initial));
let mut called = false;
assert_eq!(
view.transmit(&mut subscriber, &mut buffer, |_| {
called = true;
Ok::<_, ()>(ViewAdmission::Admitted)
}),
Ok(ViewTransmitOutcome::Current { through: initial })
);
assert!(!called);
assert!(source.update(1, |mut row| *row.position_mut() = 20).is_ok());
assert!(view.refresh(&source).is_ok());
assert_eq!(
view.transmit(&mut subscriber, &mut buffer, |_| Ok::<_, ()>(
ViewAdmission::Backpressured
)),
Ok(ViewTransmitOutcome::Backpressured {
through: view.revision()
})
);
assert_eq!(subscriber.admitted_through(), Some(initial));
assert!(source.update(1, |mut row| *row.position_mut() = 30).is_ok());
assert!(view.refresh(&source).is_ok());
assert_eq!(
view.transmit(&mut subscriber, &mut buffer, |delivery| {
let ViewDelivery::Changes {
base,
through,
changes,
} = delivery
else {
return Err("missing coalesced span");
};
assert_eq!(base, initial);
assert_eq!(through, view.revision());
assert_eq!(
changes,
[ViewChange::Upsert {
id: 1,
incarnation: 1,
value: 30
}]
);
Ok(ViewAdmission::Admitted)
}),
Ok(ViewTransmitOutcome::Admitted {
through: view.revision()
})
);
assert_eq!(subscriber.admitted_through(), Some(view.revision()));
subscriber.require_baseline();
assert_eq!(subscriber.admitted_through(), None);
assert_eq!(
view.transmit(&mut subscriber, &mut buffer, |delivery| {
let ViewDelivery::Baseline { mut rows, .. } = delivery else {
return Err("missing recovery baseline");
};
assert_eq!(rows.next().map(|row| *row.value), Some(30));
Ok(ViewAdmission::Admitted)
}),
Ok(ViewTransmitOutcome::Admitted {
through: view.revision()
})
);
let other = source
.maintain(ViewId::new(NonZeroU64::new(2).ok_or("id")?), InRoom(1))
.map_err(|_| "other view")?;
assert_eq!(
other.transmit(&mut subscriber, &mut buffer, |_| Err::<ViewAdmission, _>(
"must not send"
)),
Err(ViewTransmitError::Preparation(ViewDeliveryError::WrongView))
);
assert_eq!(subscriber.admitted_through(), Some(view.revision()));
Ok(())
}
#[test]
fn replica_refuses_unrepresentable_bounds_before_allocating() {
assert!(matches!(
ViewReplica::<u32, u32>::try_new(
SourceId::new(NonZeroU64::MIN),
ViewId::new(NonZeroU64::MIN),
Capacity::new(NonZeroU32::MAX),
),
Err(ViewCreationError::CapacityOverflow)
));
}
#[test]
fn baseline_validation_keeps_visible_state_and_rejects_cross_view_input() -> Result<(), &'static str>
{
let mut replica = replica()?;
let old = [ViewChange::Upsert {
id: 1,
incarnation: 1,
value: 10,
}];
baseline(&mut replica, stamp(1, 2), &old)?;
assert!(!replica.is_empty());
assert_eq!(replica.len(), 1);
let row = replica.get(1).ok_or("replica row")?;
assert_eq!((row.id, row.incarnation, *row.value), (1, 1, 10));
assert_eq!(
replica.begin_baseline(stamp(1, 1), 0),
Ok(ViewApplyOutcome::Stale)
);
assert_eq!(
replica.begin_baseline(stamp(1, 3), 5),
Err(ViewApplyError::CapacityExceeded)
);
assert_eq!(
replica.baseline_chunk(stamp(1, 3), &[]),
Err(ViewApplyError::BaselineNotStarted)
);
assert_eq!(
replica.finish_baseline(stamp(1, 3)),
Err(ViewApplyError::BaselineNotStarted)
);
let mut foreign = stamp(1, 3);
foreign.source = SourceId::new(NonZeroU64::new(2).ok_or("source")?);
assert_eq!(
replica.begin_baseline(foreign, 0),
Err(ViewApplyError::WrongView)
);
foreign = stamp(1, 3);
foreign.view = ViewId::new(NonZeroU64::new(2).ok_or("view")?);
assert_eq!(
replica.begin_baseline(foreign, 0),
Err(ViewApplyError::WrongView)
);
assert_eq!(
replica.begin_baseline(stamp(0, 3), 0),
Err(ViewApplyError::InvalidRevision)
);
for (changes, error) in [
(
alloc::vec![ViewChange::Remove {
id: 1,
incarnation: 1
}],
ViewApplyError::BaselineContainsRemoval,
),
(
alloc::vec![ViewChange::Upsert {
id: 1,
incarnation: 0,
value: 20
}],
ViewApplyError::InvalidIncarnation,
),
(
alloc::vec![
old.first().ok_or("row")?.clone(),
old.first().ok_or("row")?.clone()
],
ViewApplyError::DuplicateIdentity,
),
] {
assert_eq!(
replica.begin_baseline(stamp(1, 3), 2),
Ok(ViewApplyOutcome::BaselineRequired)
);
assert_eq!(
replica.baseline_chunk(stamp(1, 4), &[]),
Err(ViewApplyError::InvalidRevision)
);
assert_eq!(
replica.finish_baseline(stamp(1, 4)),
Err(ViewApplyError::InvalidRevision)
);
assert_eq!(replica.baseline_chunk(stamp(1, 3), &changes), Err(error));
assert_eq!(
replica.finish_baseline(stamp(1, 3)),
Err(ViewApplyError::BaselineNotStarted)
);
assert_eq!(
replica.rows().map(|row| *row.value).collect::<Vec<_>>(),
[10]
);
}
assert_eq!(
replica.begin_baseline(stamp(1, 3), 0),
Ok(ViewApplyOutcome::BaselineRequired)
);
assert_eq!(
replica.baseline_chunk(stamp(1, 3), &old),
Err(ViewApplyError::CapacityExceeded)
);
baseline(&mut replica, stamp(1, 3), &[])?;
assert_eq!(replica.rows().count(), 0);
assert!(replica.is_empty());
assert_eq!(replica.len(), 0);
assert!(replica.get(1).is_none());
Ok(())
}
#[test]
fn incremental_validation_rejects_bad_revisions_and_overfull_final_state()
-> Result<(), &'static str> {
let mut replica = replica()?;
let initial = [ViewChange::Upsert {
id: 1,
incarnation: 1,
value: 10,
}];
baseline(&mut replica, stamp(1, 0), &initial)?;
for (base, through) in [
(stamp(1, 1), stamp(1, 1)),
(stamp(1, 2), stamp(1, 1)),
(stamp(1, 0), stamp(2, 1)),
] {
assert_eq!(
replica.apply_changes(base, through, &[]),
Err(ViewApplyError::InvalidRevision)
);
}
let mut overfull: Vec<_> = (2..6)
.map(|id| ViewChange::Upsert {
id,
incarnation: u64::from(id),
value: id,
})
.collect();
assert_eq!(
replica.apply_changes(stamp(1, 0), stamp(1, 1), &overfull),
Err(ViewApplyError::CapacityExceeded)
);
assert_eq!(
replica.rows().map(|row| *row.value).collect::<Vec<_>>(),
[10]
);
baseline(&mut replica, stamp(1, 1), &initial)?;
overfull.push(ViewChange::Remove {
id: 1,
incarnation: 1,
});
assert_eq!(
replica.apply_changes(stamp(1, 1), stamp(1, 2), &overfull),
Ok(ViewApplyOutcome::Applied)
);
assert_eq!(replica.rows().count(), 4);
assert_eq!(
replica.apply_changes(stamp(1, 1), stamp(1, 2), &overfull),
Ok(ViewApplyOutcome::Stale)
);
assert_eq!(
replica.apply_changes(stamp(1, 3), stamp(1, 4), &[]),
Ok(ViewApplyOutcome::BaselineRequired)
);
assert_eq!(
replica.apply_changes(stamp(1, 1), stamp(1, 2), &overfull),
Ok(ViewApplyOutcome::Stale)
);
baseline(&mut replica, stamp(1, 4), &initial)?;
assert_eq!(
replica.apply_changes(
stamp(1, 4),
stamp(1, 5),
&alloc::vec![initial.first().ok_or("row")?.clone(); 9]
),
Err(ViewApplyError::CapacityExceeded)
);
baseline(&mut replica, stamp(1, 5), &initial)?;
assert_eq!(
replica.apply_changes(
stamp(1, 5),
stamp(1, 6),
&[ViewChange::Remove {
id: 1,
incarnation: 0
}]
),
Err(ViewApplyError::InvalidIncarnation)
);
Ok(())
}
#[test]
fn prepared_publications_abandon_together_and_publish_explicit_source_cuts()
-> Result<(), &'static str> {
let mut players = source(4)?;
let mut npcs = source(4)?;
npcs.source = SourceId::new(NonZeroU64::new(2).ok_or("source")?);
insert(&mut players, 1, 1, 10)?;
insert(&mut npcs, 1, 1, 20)?;
let mut player_view = view(&players)?;
let mut npc_view = view(&npcs)?;
assert!(
players
.update(1, |mut row| *row.position_mut() = 11)
.is_ok()
);
assert!(npcs.update(1, |mut row| *row.position_mut() = 21).is_ok());
let prepared_players = player_view
.prepare_refresh(&players)
.map_err(|_| "players")?;
assert!(matches!(
npc_view.prepare_refresh(&players),
Err(ViewRefreshError::WrongSource)
));
drop(prepared_players);
assert_eq!(player_view.revision().revision, 0);
assert_eq!(npc_view.revision().revision, 0);
assert_eq!(
player_view.rows().map(|row| *row.value).collect::<Vec<_>>(),
[10]
);
assert_eq!(
npc_view.rows().map(|row| *row.value).collect::<Vec<_>>(),
[20]
);
let prepared_players = player_view
.prepare_refresh(&players)
.map_err(|_| "players")?;
let prepared_npcs = npc_view.prepare_refresh(&npcs).map_err(|_| "npcs")?;
assert!(
players
.update(1, |mut row| *row.position_mut() = 12)
.is_ok()
);
assert_eq!(
prepared_players.publish(),
ViewRefreshOutcome::Published {
rows: 1,
changes: 1
}
);
assert_eq!(
prepared_npcs.publish(),
ViewRefreshOutcome::Published {
rows: 1,
changes: 1
}
);
assert_eq!(
player_view.rows().map(|row| *row.value).collect::<Vec<_>>(),
[11]
);
assert_eq!(
npc_view.rows().map(|row| *row.value).collect::<Vec<_>>(),
[21]
);
assert_eq!(
player_view.refresh(&players),
Ok(ViewRefreshOutcome::Published {
rows: 1,
changes: 1
})
);
assert_eq!(
player_view.rows().map(|row| *row.value).collect::<Vec<_>>(),
[12]
);
Ok(())
}
#[test]
fn replacement_and_generated_ids_preserve_observation_and_exact_refusals()
-> Result<(), &'static str> {
use super::super::{
WarpTableInsertOutcome, WarpTableReplaceOutcome, WarpTableUpdateOutcome,
WarpTableUpsertOutcome,
};
let mut source = source(1)?;
assert_eq!(source.source(), SourceId::new(NonZeroU64::MIN));
insert(&mut source, 1, 1, 10)?;
let mut view = view(&source)?;
let before = source.retained_changes();
let absent = Actor {
room: 1,
position: 20,
private: 0,
};
assert!(matches!(
source.replace(2, absent),
Ok(WarpTableReplaceOutcome::Absent { id: 2, .. })
));
assert!(matches!(
source.update(2, |_| ()),
Ok(WarpTableUpdateOutcome::Absent { id: 2 })
));
assert_eq!(source.retained_changes(), before);
assert_eq!(view.refresh(&source), Ok(ViewRefreshOutcome::Unchanged));
assert!(matches!(
source.replace(
1,
Actor {
room: 1,
position: 20,
private: 0
}
),
Ok(WarpTableReplaceOutcome::Replaced { .. })
));
assert_eq!(
view.refresh(&source),
Ok(ViewRefreshOutcome::Published {
rows: 1,
changes: 1
})
);
let row = view.rows().next().ok_or("row")?;
assert_eq!((row.incarnation, *row.value), (1, 20));
assert!(matches!(
source.upsert(
2,
Actor {
room: 1,
position: 30,
private: 0
}
),
Ok(WarpTableUpsertOutcome::Full { .. })
));
source.sequence = u64::MAX;
assert!(matches!(
source.replace(
1,
Actor {
room: 1,
position: 40,
private: 0
}
),
Err(ObservationEditError::SequenceExhausted {
rejected: Actor { position: 40, .. }
})
));
assert!(matches!(
source.upsert(
1,
Actor {
room: 1,
position: 50,
private: 0
}
),
Err(ObservationEditError::SequenceExhausted {
rejected: Actor { position: 50, .. }
})
));
let mut table =
WarpTable::<Generated>::try_new(Capacity::new(NonZeroU32::new(2).ok_or("capacity")?))
.map_err(|_| "table")?;
let WarpTableInsertOutcome::Inserted { id: old_id, .. } = table.insert(Generated { value: 1 })
else {
return Err("old row");
};
let mut observed = table
.observe(SourceId::new(NonZeroU64::MIN))
.map_err(|_| "observe")?;
let Ok(WarpTableInsertOutcome::Inserted { id, .. }) = observed.insert(Generated { value: 2 })
else {
return Err("generated row");
};
assert_ne!(id, old_id);
assert_eq!(observed.incarnations, [1, 2]);
assert!(matches!(
observed.insert(Generated { value: 3 }),
Ok(WarpTableInsertOutcome::Full {
rejected: Generated { value: 3 }
})
));
observed.sequence = u64::MAX;
assert!(matches!(
observed.insert(Generated { value: 4 }),
Err(ObservationEditError::SequenceExhausted {
rejected: Generated { value: 4 }
})
));
Ok(())
}
#[test]
fn latest_publication_is_shared_even_when_its_history_exceeds_retention() -> Result<(), &'static str>
{
let mut source = source(4)?;
for id in 0..4 {
insert(&mut source, id, 1, id)?;
}
let mut view = view(&source)?;
let base = view.revision();
for id in 0..4 {
assert!(source.remove(id).is_ok());
}
for id in 4..8 {
insert(&mut source, id, 1, id)?;
}
assert_eq!(
view.refresh(&source),
Ok(ViewRefreshOutcome::Published {
rows: 4,
changes: 8
})
);
let mut first_buffer = view.delivery_buffer().map_err(|_| "buffer")?;
let mut second_buffer = view.delivery_buffer().map_err(|_| "buffer")?;
let ViewDelivery::Changes { changes: first, .. } = view
.delivery(Some(base), &mut first_buffer)
.map_err(|_| "first")?
else {
return Err("latest publication needs no old history");
};
let ViewDelivery::Changes {
changes: second, ..
} = view
.delivery(Some(base), &mut second_buffer)
.map_err(|_| "second")?
else {
return Err("shared publication");
};
assert_eq!(first.len(), 8);
assert!(core::ptr::eq(first, second));
let mut replica = replica()?;
let rows: Vec<_> = (0..4)
.map(|id| ViewChange::Upsert {
id,
incarnation: u64::from(id) + 1,
value: id,
})
.collect();
baseline(&mut replica, base, &rows)?;
assert_eq!(
replica.apply_changes(base, view.revision(), first),
Ok(ViewApplyOutcome::Applied)
);
let mut ids: Vec<_> = replica.rows().map(|row| row.id).collect();
ids.sort_unstable();
assert_eq!(ids, [4, 5, 6, 7]);
Ok(())
}
#[test]
fn grouped_receivers_validate_every_part_before_exposing_any_part() -> Result<(), &'static str> {
let mut players = replica()?;
let mut npcs = replica()?;
let initial = [ViewChange::Upsert {
id: 1,
incarnation: 1,
value: 10,
}];
baseline(&mut players, stamp(1, 0), &initial)?;
baseline(&mut npcs, stamp(1, 0), &initial)?;
let changed = [ViewChange::Upsert {
id: 1,
incarnation: 1,
value: 20,
}];
let ViewPreparation::Ready(prepared_players) = players
.prepare_changes(stamp(1, 0), stamp(1, 1), &changed)
.map_err(|_| "players")?
else {
return Err("players not prepared");
};
assert!(matches!(
npcs.prepare_changes(
stamp(1, 0),
stamp(1, 1),
&[ViewChange::Remove {
id: 2,
incarnation: 2
}]
),
Err(ViewApplyError::UnknownRemoval)
));
drop(prepared_players);
assert_eq!(players.get(1).map(|row| *row.value), Some(10));
assert_eq!(npcs.get(1).map(|row| *row.value), Some(10));
assert_eq!(
players.status(),
ViewReplicaStatus::Current {
through: stamp(1, 0)
}
);
for replica in [&mut players, &mut npcs] {
assert_eq!(
replica.begin_baseline(stamp(1, 1), 1),
Ok(ViewApplyOutcome::BaselineRequired)
);
replica
.baseline_chunk(stamp(1, 1), &changed)
.map_err(|_| "chunk")?;
}
let ViewBaselinePreparation::Ready(prepared_players) = players
.prepare_baseline(stamp(1, 1))
.map_err(|_| "player baseline")?
else {
return Err("player baseline stale");
};
let ViewBaselinePreparation::Ready(prepared_npcs) = npcs
.prepare_baseline(stamp(1, 1))
.map_err(|_| "npc baseline")?
else {
return Err("npc baseline stale");
};
assert_eq!(prepared_players.publish(), stamp(1, 1));
assert_eq!(prepared_npcs.publish(), stamp(1, 1));
assert_eq!(players.get(1).map(|row| *row.value), Some(20));
assert_eq!(npcs.get(1).map(|row| *row.value), Some(20));
let ViewPreparation::Ready(prepared_players) = players
.prepare_changes(stamp(1, 1), stamp(1, 2), &initial)
.map_err(|_| "players")?
else {
return Err("players not prepared");
};
let ViewPreparation::Ready(prepared_npcs) = npcs
.prepare_changes(stamp(1, 1), stamp(1, 2), &initial)
.map_err(|_| "npcs")?
else {
return Err("npcs not prepared");
};
assert_eq!(prepared_players.publish(), stamp(1, 2));
assert_eq!(prepared_npcs.publish(), stamp(1, 2));
assert_eq!(players.get(1).map(|row| *row.value), Some(10));
assert_eq!(npcs.get(1).map(|row| *row.value), Some(10));
Ok(())
}
#[test]
fn abandoning_prepared_baseline_retains_old_rows_and_requires_a_new_begin()
-> Result<(), &'static str> {
let mut replica = replica()?;
let initial = [ViewChange::Upsert {
id: 1,
incarnation: 1,
value: 10,
}];
baseline(&mut replica, stamp(1, 0), &initial)?;
assert_eq!(
replica.begin_baseline(stamp(1, 1), 0),
Ok(ViewApplyOutcome::BaselineRequired)
);
let ViewBaselinePreparation::Ready(prepared) = replica
.prepare_baseline(stamp(1, 1))
.map_err(|_| "prepare")?
else {
return Err("not prepared");
};
drop(prepared);
assert_eq!(replica.get(1).map(|row| *row.value), Some(10));
assert_eq!(
replica.finish_baseline(stamp(1, 1)),
Err(ViewApplyError::BaselineNotStarted)
);
assert_eq!(
replica.begin_baseline(stamp(1, 1), 0),
Ok(ViewApplyOutcome::BaselineRequired)
);
assert_eq!(
replica.apply_changes(stamp(1, 1), stamp(1, 2), &[]),
Ok(ViewApplyOutcome::BaselineRequired)
);
assert!(matches!(
replica.prepare_baseline(stamp(1, 1)),
Ok(ViewBaselinePreparation::Stale)
));
assert_eq!(replica.get(1).map(|row| *row.value), Some(10));
Ok(())
}
#[test]
fn foreign_sources_and_delivery_storage_are_refused_without_changing_the_view()
-> Result<(), &'static str> {
let mut source = source(8)?;
insert(&mut source, 1, 1, 10)?;
let mut view = view(&source)?;
let initial = view.revision();
let mut foreign = super::tests::source(1)?;
foreign.source = SourceId::new(NonZeroU64::new(2).ok_or("id")?);
let Err((returned, error)) = view.replace_definition(&foreign, InRoom(2)) else {
return Err("wrong source accepted");
};
assert_eq!(returned.0, 2);
assert_eq!(error, ViewRefreshError::WrongSource);
assert_eq!(view.revision(), initial);
assert_eq!(view.rows().map(|row| *row.value).collect::<Vec<_>>(), [10]);
let small = super::tests::view(&foreign)?;
let mut small_buffer = small.delivery_buffer().map_err(|_| "small buffer")?;
for position in [11, 12] {
assert!(
source
.update(1, |mut row| *row.position_mut() = position)
.is_ok()
);
assert!(view.refresh(&source).is_ok());
}
assert!(matches!(
view.delivery(Some(initial), &mut small_buffer),
Err(ViewDeliveryError::InsufficientBuffer)
));
let mut buffer = view.delivery_buffer().map_err(|_| "buffer")?;
let mut wrong = initial;
wrong.source = foreign.source();
assert!(matches!(
view.delivery(Some(wrong), &mut buffer),
Err(ViewDeliveryError::WrongView)
));
wrong = initial;
wrong.view = ViewId::new(NonZeroU64::new(2).ok_or("id")?);
assert!(matches!(
view.delivery(Some(wrong), &mut buffer),
Err(ViewDeliveryError::WrongView)
));
wrong = initial;
wrong.epoch = 2;
assert!(matches!(
view.delivery(Some(wrong), &mut buffer),
Err(ViewDeliveryError::FutureCursor)
));
assert!(matches!(
view.delivery(Some(initial), &mut buffer),
Ok(ViewDelivery::Changes { .. })
));
assert_eq!(view.rows().map(|row| *row.value).collect::<Vec<_>>(), [12]);
Ok(())
}
#[cfg(feature = "secondary-indexes")]
#[test]
fn bulk_edit_refuses_foreign_membership_and_preflights_sequence_space() -> Result<(), &'static str>
{
let mut source = source(4)?;
insert(&mut source, 1, 1, 10)?;
let mut foreign =
WarpTable::<Actor>::try_new(Capacity::new(NonZeroU32::MIN)).map_err(|_| "foreign")?;
assert!(matches!(
foreign.insert_with_id(
1,
Actor {
room: 1,
position: 50,
private: 0
}
),
WarpTableInsertWithIdOutcome::Inserted { .. }
));
let foreign: &'static WarpTable<Actor> =
alloc::boxed::Box::leak(alloc::boxed::Box::new(foreign));
let mut called = false;
assert_eq!(
source.update_selected(|_| foreign.query().find_room(1).into(), |_| called = true),
Err(ObservationSelectionError::ForeignTable)
);
assert!(!called);
source.sequence = u64::MAX;
assert_eq!(
source.update_selected(|table| table.query().find_room(1).into(), |_| called = true),
Err(ObservationSelectionError::SequenceExhausted)
);
assert!(!called);
assert!(
matches!(source.table().get(1), WarpTableGetOutcome::Present { row } if *row.position == 10)
);
Ok(())
}