use super::*;
use alloc::vec::Vec;
type Client = SubscribedView<u32, u32>;
type Buffer = AppliedViewChanges<u32>;
fn ready() -> Result<(Client, SubscriptionSession, SubscriptionId, Buffer), String> {
let (subscription, _) = subscription_ids()?;
let mut client = Client::new(ContinuousView::new(limits().rows));
let session = start(&mut client)?;
let mut buffer = Buffer::try_new(limits().rows).map_err(debug)?;
assert_eq!(
client
.step(
ReceiveSubscription {
session,
reply: &Reply::Subscribed {
id: REQUEST,
subscription
},
}
.with_changes(&mut buffer)
)
.map_err(debug)?,
SubscriptionChanges::Subscribed,
);
Ok((client, session, subscription, buffer))
}
fn row(id: u32, incarnation: u64, value: u32) -> ViewChange<u32, u32> {
ViewChange::Upsert {
id,
incarnation,
value,
}
}
fn snapshot(subscription: SubscriptionId, revision: u64, rows: Vec<ViewChange<u32, u32>>) -> Reply {
Reply::View {
subscription,
update: ViewUpdate::Baseline {
through: stamp(revision),
rows,
},
}
}
fn changes(
subscription: SubscriptionId,
base: u64,
through: u64,
changes: Vec<ViewChange<u32, u32>>,
) -> Reply {
Reply::View {
subscription,
update: ViewUpdate::Changes {
base: stamp(base),
through: stamp(through),
changes,
},
}
}
fn receive<'a>(
client: &mut Client,
session: SubscriptionSession,
reply: &Reply,
buffer: &'a mut Buffer,
) -> Result<SubscriptionChanges<'a, u32>, SubscriptionReceiveError> {
client.step(ReceiveSubscription { session, reply }.with_changes(buffer))
}
#[test]
fn reports_describe_installed_membership_values_and_incarnations() -> Result<(), String> {
use AppliedViewChange::*;
let (mut client, session, subscription, mut buffer) = ready()?;
let initial = snapshot(
subscription,
1,
vec![row(1, 1, 10), row(2, 1, 20), row(3, 1, 30)],
);
assert_eq!(
receive(&mut client, session, &initial, &mut buffer).map_err(debug)?,
SubscriptionChanges::Applied {
changes: &[Entered { id: 1 }, Entered { id: 2 }, Entered { id: 3 }]
}
);
let repeated = snapshot(
subscription,
2,
vec![row(3, 1, 30), row(1, 1, 10), row(2, 1, 20)],
);
assert_eq!(
receive(&mut client, session, &repeated, &mut buffer).map_err(debug)?,
SubscriptionChanges::Applied { changes: &[] }
);
let update = changes(
subscription,
2,
3,
vec![
row(1, 1, 11),
row(2, 2, 20),
ViewChange::Remove {
id: 3,
incarnation: 1,
},
row(4, 1, 40),
],
);
assert_eq!(
receive(&mut client, session, &update, &mut buffer).map_err(debug)?,
SubscriptionChanges::Applied {
changes: &[
Updated { id: 1 },
Replaced { id: 2 },
Left { id: 3 },
Entered { id: 4 }
]
}
);
assert_eq!(client.view().get(1), Some(&11));
assert_eq!(client.view().status(), ContinuousViewStatus::Current);
assert_eq!(client.view().get(3), None);
assert_eq!(client.view().get(4), Some(&40));
assert_eq!(
receive(&mut client, session, &update, &mut buffer).map_err(debug)?,
SubscriptionChanges::Stale
);
let invalid = changes(
subscription,
3,
4,
vec![
row(1, 1, 99),
ViewChange::Remove {
id: 99,
incarnation: 1,
},
],
);
assert!(matches!(
receive(&mut client, session, &invalid, &mut buffer),
Err(SubscriptionReceiveError::Apply(ContinuousViewError::Apply(
ViewApplyError::UnknownRemoval
)))
));
assert_eq!(client.view().get(1), Some(&11));
Ok(())
}
#[test]
fn report_capacity_refuses_before_baseline_or_delta_becomes_visible() -> Result<(), String> {
let (mut client, session, subscription, mut buffer) = ready()?;
let initial = snapshot(
subscription,
1,
vec![row(1, 1, 10), row(2, 1, 20), row(3, 1, 30)],
);
let mut small = Buffer::try_new(Capacity::new(NonZeroU32::MIN)).map_err(debug)?;
assert!(matches!(
receive(&mut client, session, &initial, &mut small),
Err(SubscriptionReceiveError::Apply(ContinuousViewError::Apply(
ViewApplyError::ReportCapacityExceeded
)))
));
assert_eq!(client.view().get(1), None);
let _applied = receive(&mut client, session, &initial, &mut buffer).map_err(debug)?;
let delta = changes(
subscription,
1,
2,
vec![row(1, 1, 11), row(2, 1, 21), row(3, 1, 31)],
);
assert!(matches!(
receive(&mut client, session, &delta, &mut small),
Err(SubscriptionReceiveError::Apply(ContinuousViewError::Apply(
ViewApplyError::ReportCapacityExceeded
)))
));
assert_eq!(client.view().get(1), Some(&10));
assert_eq!(client.view().get(2), Some(&20));
let _applied = receive(&mut client, session, &delta, &mut buffer).map_err(debug)?;
let replacement = snapshot(
subscription,
3,
vec![row(4, 1, 40), row(5, 1, 50), row(6, 1, 60)],
);
assert!(matches!(
receive(&mut client, session, &replacement, &mut small),
Err(SubscriptionReceiveError::Apply(ContinuousViewError::Apply(
ViewApplyError::ReportCapacityExceeded
)))
));
assert_eq!(client.view().get(1), Some(&11));
assert_eq!(client.view().get(4), None);
let SubscriptionChanges::Applied { changes } =
receive(&mut client, session, &replacement, &mut buffer).map_err(debug)?
else {
return Err("report".into());
};
assert_eq!(changes.len(), 6);
assert_eq!(client.view().get(1), None);
assert_eq!(client.view().get(4), Some(&40));
Ok(())
}
#[test]
fn recovery_reports_net_changes_and_reconnect_fences_old_traffic() -> Result<(), String> {
use AppliedViewChange::*;
let (mut client, session, subscription, mut buffer) = ready()?;
let _applied = receive(
&mut client,
session,
&snapshot(subscription, 1, vec![row(1, 1, 10), row(2, 1, 20)]),
&mut buffer,
)
.map_err(debug)?;
assert_eq!(
receive(
&mut client,
session,
&changes(subscription, 2, 3, vec![]),
&mut buffer
)
.map_err(debug)?,
SubscriptionChanges::BaselineRequired
);
assert_eq!(client.view().get(2), Some(&20));
assert_eq!(
client.view().status(),
ContinuousViewStatus::BaselineRequired
);
let _request: Request = client.resynchronize(NEXT).map_err(debug)?;
let replacement = snapshot(subscription, 3, vec![row(1, 1, 10), row(3, 1, 30)]);
assert!(matches!(
receive(&mut client, session, &replacement, &mut buffer),
Err(SubscriptionReceiveError::Unexpected)
));
assert_eq!(
receive(
&mut client,
session,
&Reply::Resynchronizing {
id: NEXT,
subscription
},
&mut buffer
)
.map_err(debug)?,
SubscriptionChanges::Resynchronizing
);
assert_eq!(
receive(&mut client, session, &replacement, &mut buffer).map_err(debug)?,
SubscriptionChanges::Applied {
changes: &[Left { id: 2 }, Entered { id: 3 }]
}
);
client.disconnect(session).map_err(debug)?;
let next = start(&mut client)?;
let _subscribed = receive(
&mut client,
next,
&Reply::Subscribed {
id: REQUEST,
subscription,
},
&mut buffer,
)
.map_err(debug)?;
assert!(matches!(
receive(&mut client, session, &replacement, &mut buffer),
Err(SubscriptionReceiveError::WrongSession)
));
let mut restarted = replacement;
if let Reply::View {
update: ViewUpdate::Baseline { through, .. },
..
} = &mut restarted
{
through.epoch = 2;
}
let SubscriptionChanges::Applied { changes } =
receive(&mut client, next, &restarted, &mut buffer).map_err(debug)?
else {
return Err("report".into());
};
assert_eq!(changes.len(), 2);
assert!(changes.contains(&Replaced { id: 1 }));
assert!(changes.contains(&Replaced { id: 3 }));
Ok(())
}
proptest! {
#[test]
fn replacement_reports_exact_set_and_value_differences(
before in prop::collection::vec(prop::option::of(0_u32..4), 4),
after in prop::collection::vec(prop::option::of(0_u32..4), 4),
) {
let (mut client, session, subscription, mut buffer) = ready().map_err(TestCaseError::fail)?;
let make = |values: &[Option<u32>]| values.iter().zip(0_u32..).filter_map(|(value,id)| value.map(|value| row(id,1,value))).collect();
let _applied = receive(&mut client, session, &snapshot(subscription,1,make(&before)), &mut buffer).map_err(|e| TestCaseError::fail(debug(e)))?;
let reply = snapshot(subscription,2,make(&after));
let outcome = receive(&mut client, session, &reply, &mut buffer).map_err(|e| TestCaseError::fail(debug(e)))?;
let SubscriptionChanges::Applied { changes } = outcome else { return Err(TestCaseError::fail("report")); };
let mut expected = Vec::new();
for ((before, after),id) in before.iter().zip(&after).zip(0_u32..) {
match (before,after) {
(None,Some(_)) => expected.push(AppliedViewChange::Entered { id }),
(Some(_),None) => expected.push(AppliedViewChange::Left { id }),
(Some(a),Some(b)) if a != b => expected.push(AppliedViewChange::Updated { id }),
(None,None) | (Some(_),Some(_)) => {},
}
prop_assert_eq!(client.view().get(id),after.as_ref());
}
prop_assert_eq!(changes.len(),expected.len());
for change in expected { prop_assert!(changes.contains(&change)); }
}
}
#[test]
fn reconnect_to_same_authority_reports_value_changes_removals_and_new_members() -> Result<(), String>
{
use AppliedViewChange::*;
let (mut client, session, subscription, mut buffer) = ready()?;
let _applied = receive(
&mut client,
session,
&snapshot(subscription, 1, vec![row(1, 1, 10), row(2, 1, 20)]),
&mut buffer,
)
.map_err(debug)?;
client.disconnect(session).map_err(debug)?;
let next = start(&mut client)?;
let _subscribed = receive(
&mut client,
next,
&Reply::Subscribed {
id: REQUEST,
subscription,
},
&mut buffer,
)
.map_err(debug)?;
assert_eq!(
receive(
&mut client,
next,
&snapshot(subscription, 2, vec![row(1, 1, 11), row(3, 1, 30)]),
&mut buffer
)
.map_err(debug)?,
SubscriptionChanges::Applied {
changes: &[Updated { id: 1 }, Left { id: 2 }, Entered { id: 3 }]
}
);
let mut epoch = snapshot(subscription, 3, vec![row(1, 1, 11), row(3, 1, 30)]);
if let Reply::View {
update: ViewUpdate::Baseline { through, .. },
..
} = &mut epoch
{
through.epoch = 2;
}
assert_eq!(
receive(&mut client, next, &epoch, &mut buffer).map_err(debug)?,
SubscriptionChanges::Applied {
changes: &[Replaced { id: 1 }, Replaced { id: 3 }]
}
);
let mut unchanged = snapshot(subscription, 4, vec![row(1, 1, 11), row(3, 1, 30)]);
if let Reply::View {
update: ViewUpdate::Baseline { through, .. },
..
} = &mut unchanged
{
through.epoch = 2;
}
assert_eq!(
receive(&mut client, next, &unchanged, &mut buffer).map_err(debug)?,
SubscriptionChanges::Applied { changes: &[] }
);
client.disconnect(next).map_err(debug)?;
let next = start(&mut client)?;
assert_eq!(
receive(
&mut client,
next,
&Reply::SubscriptionRefused {
id: REQUEST,
reason: SubscriptionRefusal::Full
},
&mut buffer
)
.map_err(debug)?,
SubscriptionChanges::Refused(SubscriptionRefusal::Full)
);
assert_eq!(client.view().get(1), Some(&11));
Ok(())
}
#[test]
fn reconnect_report_refusal_preserves_the_old_copy_and_authority_changes_are_replacements()
-> Result<(), String> {
let (mut client, session, subscription, mut buffer) = ready()?;
let _applied = receive(
&mut client,
session,
&snapshot(
subscription,
1,
vec![row(1, 1, 10), row(2, 1, 20), row(3, 1, 30)],
),
&mut buffer,
)
.map_err(debug)?;
assert_eq!(
receive(
&mut client,
session,
&changes(subscription, 2, 3, vec![]),
&mut buffer
)
.map_err(debug)?,
SubscriptionChanges::BaselineRequired
);
client.disconnect(session).map_err(debug)?;
let mut session = start(&mut client)?;
let _subscribed = receive(
&mut client,
session,
&Reply::Subscribed {
id: REQUEST,
subscription,
},
&mut buffer,
)
.map_err(debug)?;
let update = snapshot(
subscription,
3,
vec![row(1, 1, 11), row(2, 1, 21), row(3, 1, 31)],
);
let mut small = Buffer::try_new(Capacity::new(NonZeroU32::MIN)).map_err(debug)?;
assert!(matches!(
receive(&mut client, session, &update, &mut small),
Err(SubscriptionReceiveError::Apply(ContinuousViewError::Apply(
ViewApplyError::ReportCapacityExceeded
)))
));
assert_eq!(small.as_slice().len(), 2);
assert_eq!(client.view().get(1), Some(&10));
let _applied = receive(&mut client, session, &update, &mut buffer).map_err(debug)?;
for change_source in [true, false] {
client.disconnect(session).map_err(debug)?;
session = start(&mut client)?;
let _subscribed = receive(
&mut client,
session,
&Reply::Subscribed {
id: REQUEST,
subscription,
},
&mut buffer,
)
.map_err(debug)?;
let mut update = snapshot(subscription, 4, vec![row(1, 1, 11)]);
if let Reply::View {
update: ViewUpdate::Baseline { through, .. },
..
} = &mut update
{
through.source = SourceId::new(NonZeroU64::MIN.saturating_add(1));
if !change_source {
through.view = ViewId::new(NonZeroU64::MIN.saturating_add(1));
}
}
let SubscriptionChanges::Applied { changes } =
receive(&mut client, session, &update, &mut buffer).map_err(debug)?
else {
return Err("report".into());
};
assert!(changes.contains(&AppliedViewChange::Replaced { id: 1 }));
}
Ok(())
}
#[test]
fn empty_local_copy_reports_its_initial_baseline() -> Result<(), String> {
let mut view = ContinuousView::<u32, u32>::new(limits().rows);
let mut report = Buffer::try_new(limits().rows).map_err(debug)?;
assert_eq!(
view.receive_reporting(
&ViewUpdate::Baseline {
through: stamp(1),
rows: vec![row(7, 1, 70)]
},
Some(&mut report)
)
.map_err(debug)?,
ViewApplyOutcome::Applied
);
assert_eq!(report.as_slice(), &[AppliedViewChange::Entered { id: 7 }]);
assert_eq!(view.get(7), Some(&70));
Ok(())
}