use super::*;
#[derive(Clone, Copy, Debug)]
struct RevisionWithProvenance {
number: u8,
provenance: u8,
}
impl PartialEq for RevisionWithProvenance {
fn eq(&self, other: &Self) -> bool {
self.number == other.number
}
}
impl Eq for RevisionWithProvenance {}
impl Ord for RevisionWithProvenance {
fn cmp(&self, other: &Self) -> core::cmp::Ordering {
self.number.cmp(&other.number)
}
}
impl PartialOrd for RevisionWithProvenance {
fn partial_cmp(&self, other: &Self) -> Option<core::cmp::Ordering> {
Some(self.cmp(other))
}
}
impl SynchronizedViewRevision for RevisionWithProvenance {
type ChangeId = Self;
fn changed_through(change_id: Self::ChangeId) -> Self {
change_id
}
fn next_change_id(self) -> Option<Self::ChangeId> {
self.number.checked_add(1).map(|number| Self {
number,
provenance: self.provenance,
})
}
}
#[test]
fn equal_revisions_preserve_the_first_observed_provenance() -> Result<(), &'static str> {
use core::cell::Cell;
let stored = RevisionWithProvenance {
number: 4,
provenance: 10,
};
let equal = RevisionWithProvenance {
number: 4,
provenance: 20,
};
let synchronized = RevisionWithProvenance {
number: 2,
provenance: 2,
};
assert_eq!(stored, equal);
assert_eq!(stored.cmp(&equal), core::cmp::Ordering::Equal);
assert_ne!(stored.provenance, equal.provenance);
for observing in [false, true] {
let mut view = SynchronizedView::new(RevisionWithProvenance {
number: 0,
provenance: 0,
});
if observing {
assert!(matches!(
view.apply_snapshot(synchronized, 7_u8),
ApplySynchronizedViewSnapshotOutcome::Synchronized { .. }
));
}
view.require_snapshot_through(stored);
view.require_snapshot_through(equal);
assert_eq!(view.required_snapshot_through().provenance, 10);
let calls = Cell::new(0_u8);
let ApplySynchronizedViewRevisionSpanOutcome::SnapshotRequired {
observed_through,
reason,
} = view.apply_revision_span(
synchronized,
equal,
|_| {
calls.set(calls.get() | 1);
Ok::<_, ()>(9)
},
|_, _| {
calls.set(calls.get() | 2);
Ok(())
},
|projection, next| {
calls.set(calls.get() | 4);
*projection = next;
},
)
else {
return Err("a revision span bypassed snapshot recovery");
};
assert_eq!(observed_through.provenance, 10);
assert_eq!(view.required_snapshot_through().provenance, 10);
if observing {
assert!(matches!(
reason,
SynchronizedViewSnapshotRequiredReason::AlreadyRequired
));
} else {
assert!(matches!(
reason,
SynchronizedViewSnapshotRequiredReason::NoSnapshotApplied
));
}
let ApplySynchronizedViewChangeOutcome::SnapshotRequired {
observed_through,
reason,
} = view.apply_change(
equal,
|_| {
calls.set(calls.get() | 1);
Ok::<_, ()>(9)
},
|_, _| {
calls.set(calls.get() | 2);
Ok(())
},
|projection, next| {
calls.set(calls.get() | 4);
*projection = next;
},
)
else {
return Err("a change bypassed snapshot recovery");
};
assert_eq!(observed_through.provenance, 10);
assert_eq!(view.required_snapshot_through().provenance, 10);
assert_eq!(calls.get(), 0);
if observing {
assert!(matches!(
reason,
SynchronizedViewSnapshotRequiredReason::AlreadyRequired
));
assert!(matches!(
view.read(),
SynchronizedViewRead::SynchronizationRequired { projection: 7, .. }
));
} else {
assert!(matches!(
reason,
SynchronizedViewSnapshotRequiredReason::NoSnapshotApplied
));
assert!(matches!(
view.read(),
SynchronizedViewRead::AwaitingSnapshot { .. }
));
}
}
Ok(())
}
#[derive(Clone, Copy, Eq, Ord, PartialEq, PartialOrd)]
struct Revision(u8);
impl SynchronizedViewRevision for Revision {
type ChangeId = u8;
fn changed_through(change_id: Self::ChangeId) -> Self {
Self(change_id)
}
fn next_change_id(self) -> Option<Self::ChangeId> {
self.0.checked_add(1)
}
}
#[test]
fn a_gap_retains_the_last_coherent_projection_until_a_covering_snapshot() {
let mut view = SynchronizedView::new(Revision(0));
assert!(matches!(
view.apply_snapshot(Revision(2), 20),
ApplySynchronizedViewSnapshotOutcome::Synchronized {
through: Revision(2)
}
));
assert!(matches!(
view.apply_change(
4,
|_| Ok::<_, ()>(40),
|_, _| Ok(()),
|projection, value| {
*projection = value;
}
),
ApplySynchronizedViewChangeOutcome::SnapshotRequired {
observed_through: Revision(4),
reason: SynchronizedViewSnapshotRequiredReason::ChangeGap {
synchronized_through: Revision(2)
}
}
));
assert!(matches!(
view.read(),
SynchronizedViewRead::SynchronizationRequired {
synchronized_through: Revision(2),
observed_through: Revision(4),
projection: 20,
}
));
assert!(matches!(
view.apply_snapshot(Revision(3), 30),
ApplySynchronizedViewSnapshotOutcome::Stale {
snapshot_through: Revision(3),
required_through: Revision(4),
}
));
assert!(matches!(
view.apply_snapshot(Revision(4), 40),
ApplySynchronizedViewSnapshotOutcome::Synchronized {
through: Revision(4)
}
));
assert!(matches!(
view.read(),
SynchronizedViewRead::Synchronized {
through: Revision(4),
projection: 40,
}
));
}
#[test]
fn only_the_exact_next_change_reaches_the_projection_reducer() {
let mut view = SynchronizedView::new(Revision(0));
assert!(matches!(
view.apply_snapshot(Revision(1), 10),
ApplySynchronizedViewSnapshotOutcome::Synchronized { .. }
));
let mut stale_called = false;
assert!(matches!(
view.apply_change(
1,
|_| {
stale_called = true;
Ok::<_, ()>(())
},
|_, _| Ok(()),
|_, ()| {},
),
ApplySynchronizedViewChangeOutcome::Stale { .. }
));
assert!(!stale_called);
assert!(matches!(
view.apply_change(
2,
|_| Ok::<_, ()>(20),
|_, _| Ok(()),
|projection, value| {
*projection = value;
}
),
ApplySynchronizedViewChangeOutcome::Applied {
through: Revision(2)
}
));
assert!(matches!(
view.read(),
SynchronizedViewRead::Synchronized {
through: Revision(2),
projection: 20,
}
));
}
#[test]
fn a_rejected_next_change_requires_recovery_without_advancing_the_revision() {
let mut view = SynchronizedView::new(Revision(0));
assert!(matches!(
view.apply_snapshot(Revision(1), 10),
ApplySynchronizedViewSnapshotOutcome::Synchronized { .. }
));
assert!(matches!(
view.apply_change(2, |_| Err("invalid"), |_, &()| Ok(()), |_, ()| {}),
ApplySynchronizedViewChangeOutcome::SnapshotRequired {
observed_through: Revision(2),
reason: SynchronizedViewSnapshotRequiredReason::ChangeRejected("invalid"),
}
));
assert!(matches!(
view.read(),
SynchronizedViewRead::SynchronizationRequired {
synchronized_through: Revision(1),
observed_through: Revision(2),
projection: 10,
}
));
}
#[test]
fn changes_seen_before_hydration_raise_the_minimum_acceptable_snapshot() {
let mut view = SynchronizedView::new(Revision(0));
assert!(matches!(
view.apply_change(2, |_| Ok::<_, ()>(()), |_, _| Ok(()), |_, ()| {}),
ApplySynchronizedViewChangeOutcome::SnapshotRequired {
observed_through: Revision(2),
reason: SynchronizedViewSnapshotRequiredReason::NoSnapshotApplied,
}
));
assert!(matches!(
view.apply_change(4, |_| Ok::<_, ()>(()), |_, _| Ok(()), |_, ()| {}),
ApplySynchronizedViewChangeOutcome::SnapshotRequired {
observed_through: Revision(4),
reason: SynchronizedViewSnapshotRequiredReason::NoSnapshotApplied,
}
));
assert!(matches!(
view.apply_snapshot(Revision(3), 30),
ApplySynchronizedViewSnapshotOutcome::Stale {
required_through: Revision(4),
..
}
));
}
#[test]
fn an_explicit_requirement_accepts_an_equal_revision_snapshot() {
let mut view = SynchronizedView::new(Revision(0));
assert!(matches!(
view.apply_snapshot(Revision(2), 20),
ApplySynchronizedViewSnapshotOutcome::Synchronized { .. }
));
view.require_snapshot_through(Revision(2));
assert!(matches!(
view.read(),
SynchronizedViewRead::SynchronizationRequired {
synchronized_through: Revision(2),
observed_through: Revision(2),
projection: 20,
}
));
assert!(matches!(
view.apply_snapshot(Revision(2), 21),
ApplySynchronizedViewSnapshotOutcome::Synchronized {
through: Revision(2)
}
));
}
#[test]
fn preparation_and_reservation_failures_do_not_commit() {
let mut view = SynchronizedView::new(Revision(0));
assert!(matches!(
view.apply_snapshot(Revision(1), 10),
ApplySynchronizedViewSnapshotOutcome::Synchronized { .. }
));
let mut committed = false;
assert!(matches!(
view.apply_change(
2,
|_| Ok::<_, &str>(20),
|_, _| Err("unavailable"),
|projection, value| {
committed = true;
*projection = value;
},
),
ApplySynchronizedViewChangeOutcome::SnapshotRequired {
reason: SynchronizedViewSnapshotRequiredReason::ChangeRejected("unavailable"),
..
}
));
assert!(!committed);
assert!(matches!(
view.read(),
SynchronizedViewRead::SynchronizationRequired { projection: 10, .. }
));
}
#[test]
fn revision_spans_require_hydration_and_never_regress_observed_progress() {
use core::cell::Cell;
let calls = Cell::new(0_u8);
let mut view = SynchronizedView::<u8, u8>::new(2);
for (through, observed) in [(1, 2), (4, 4), (3, 4), (4, 4)] {
assert!(
matches!(view.apply_revision_span(0, through, |_| { calls.set(calls.get() | 1); Ok::<_, ()>(9) }, |_, _| { calls.set(calls.get() | 2); Ok(()) }, |value, next| { calls.set(calls.get() | 4); *value = next; }), ApplySynchronizedViewRevisionSpanOutcome::SnapshotRequired { observed_through, reason: SynchronizedViewSnapshotRequiredReason::NoSnapshotApplied } if observed_through == observed)
);
assert_eq!(view.required_snapshot_through(), observed);
assert!(
matches!(view.read(), SynchronizedViewRead::AwaitingSnapshot { observed_through } if observed_through == observed)
);
}
assert_eq!(calls.get(), 0);
assert!(matches!(
view.apply_snapshot(3, 30),
ApplySynchronizedViewSnapshotOutcome::Stale {
snapshot_through: 3,
required_through: 4
}
));
assert!(matches!(
view.apply_snapshot(4, 40),
ApplySynchronizedViewSnapshotOutcome::Synchronized { through: 4 }
));
assert!(matches!(
view.apply_revision_span(
4,
6,
|value| {
assert_eq!(*value, 40);
calls.set(1);
Ok::<_, ()>(60)
},
|value, prepared| {
assert_eq!((*value, *prepared), (40, 60));
calls.set(calls.get() | 2);
Ok(())
},
|value, next| {
calls.set(calls.get() | 4);
*value = next;
}
),
ApplySynchronizedViewRevisionSpanOutcome::Applied { through: 6 }
));
assert_eq!(calls.get(), 7);
assert_eq!(view.required_snapshot_through(), 6);
assert!(matches!(
view.read(),
SynchronizedViewRead::Synchronized {
through: 6,
projection: 60
}
));
}
#[test]
fn stale_gapped_and_already_required_spans_do_not_run_callbacks() {
use core::cell::Cell;
let calls = Cell::new(0_u8);
let mut view = SynchronizedView::new(0_u8);
assert!(matches!(
view.apply_snapshot(2, 20),
ApplySynchronizedViewSnapshotOutcome::Synchronized { through: 2 }
));
for through in [1, 2] {
assert!(
matches!(view.apply_revision_span(2, through, |_| { calls.set(1); Ok::<_, ()>(99) }, |_, _| { calls.set(2); Ok(()) }, |_, _| { calls.set(3); }), ApplySynchronizedViewRevisionSpanOutcome::Stale { span_through, synchronized_through: 2 } if span_through == through)
);
}
assert!(matches!(
view.apply_revision_span(
3,
4,
|_| {
calls.set(1);
Ok::<_, ()>(99)
},
|_, _| {
calls.set(2);
Ok(())
},
|_, _| {
calls.set(3);
}
),
ApplySynchronizedViewRevisionSpanOutcome::SnapshotRequired {
observed_through: 4,
reason: SynchronizedViewSnapshotRequiredReason::ChangeGap {
synchronized_through: 2
}
}
));
for (through, observed) in [(3, 4), (4, 4), (7, 7)] {
assert!(
matches!(view.apply_revision_span(2, through, |_| { calls.set(1); Ok::<_, ()>(99) }, |_, _| { calls.set(2); Ok(()) }, |_, _| { calls.set(3); }), ApplySynchronizedViewRevisionSpanOutcome::SnapshotRequired { observed_through, reason: SynchronizedViewSnapshotRequiredReason::AlreadyRequired } if observed_through == observed)
);
}
assert_eq!(calls.get(), 0);
assert!(matches!(
view.read(),
SynchronizedViewRead::SynchronizationRequired {
synchronized_through: 2,
observed_through: 7,
projection: 20
}
));
}
#[test]
fn rejected_span_preparation_and_reservation_preserve_the_last_coherent_view() {
use core::cell::Cell;
for fail_prepare in [true, false] {
let calls = Cell::new(0_u8);
let mut view = SynchronizedView::new(0_u8);
assert!(matches!(
view.apply_snapshot(2, 20),
ApplySynchronizedViewSnapshotOutcome::Synchronized { through: 2 }
));
let result = view.apply_revision_span(
2,
5,
|_| {
calls.set(1);
if fail_prepare { Err("prepare") } else { Ok(99) }
},
|_, _| {
calls.set(calls.get() | 2);
Err("reserve")
},
|value, next| {
calls.set(calls.get() | 4);
*value = next;
},
);
assert!(
matches!(result, ApplySynchronizedViewRevisionSpanOutcome::SnapshotRequired { observed_through: 5, reason: SynchronizedViewSnapshotRequiredReason::ChangeRejected(reason) } if reason == if fail_prepare { "prepare" } else { "reserve" })
);
assert_eq!(calls.get(), if fail_prepare { 1 } else { 3 });
assert!(matches!(
view.read(),
SynchronizedViewRead::SynchronizationRequired {
synchronized_through: 2,
observed_through: 5,
projection: 20
}
));
}
}
#[test]
fn a_required_single_change_never_reenters_the_reducer() {
use core::cell::Cell;
let calls = Cell::new(0_u8);
let mut view = SynchronizedView::new(Revision(0));
assert!(matches!(
view.apply_snapshot(Revision(2), 20),
ApplySynchronizedViewSnapshotOutcome::Synchronized {
through: Revision(2)
}
));
view.require_snapshot_through(Revision(4));
for (change, observed) in [(3, 4), (4, 4), (6, 6)] {
assert!(
matches!(view.apply_change(change, |_| { calls.set(1); Ok::<_, ()>(99) }, |_, _| { calls.set(2); Ok(()) }, |_, _| { calls.set(3); }), ApplySynchronizedViewChangeOutcome::SnapshotRequired { observed_through: Revision(actual), reason: SynchronizedViewSnapshotRequiredReason::AlreadyRequired } if actual == observed)
);
}
assert_eq!(calls.get(), 0);
assert!(matches!(
view.read(),
SynchronizedViewRead::SynchronizationRequired {
synchronized_through: Revision(2),
observed_through: Revision(6),
projection: 20
}
));
}