diff --git a/crates/improvise-protocol/src/delivery/client.rs b/crates/improvise-protocol/src/delivery/client.rs index 67ce2800..d9d47f42 100644 --- a/crates/improvise-protocol/src/delivery/client.rs +++ b/crates/improvise-protocol/src/delivery/client.rs @@ -156,8 +156,16 @@ impl EdgeClient { /// A pushed notice. Duplicates are ignored, a gap or a new epoch makes every shape ask. pub fn on_notice(&mut self, notice: &Notice) -> Vec { + self.accept(notice, None) + } + + /// `only` limits the notice to one shape: a pulled notice carries no other shape's keys. + fn accept(&mut self, notice: &Notice, only: Option<&ShapeKey>) -> Vec { let end = notice.stamp(); for (shape, t) in &mut self.shapes { + if only.is_some_and(|o| o != shape) { + continue; + } let contiguous = match t.seen { Some(seen) if seen.same_epoch(&end) => match notice.range.to <= seen.publication { true => continue, @@ -170,7 +178,7 @@ impl EdgeClient { t.unapplied.push((end, Moved::Correct)); continue; } - if !notice.axis.is_empty() || !notice.relabels.is_empty() { + if !notice.axis.is_empty() || !notice.relabels.is_empty() || notice.flags.views { t.restructured = Some(end); } match notice.moved.get(shape) { @@ -235,6 +243,12 @@ impl EdgeClient { return (Outcome::Untracked, Vec::new()); }; let target = decode(&c.target); + let older = |s: Option| { + s.is_some_and(|s| s.host == c.through.host && c.through.generation < s.generation) + }; + if older(t.watermark) || older(t.seen) { + return (Outcome::Duplicate, Vec::new()); + } let covered = t .watermark .is_some_and(|w| w.same_epoch(&c.through) && w.publication >= c.through.publication); @@ -245,6 +259,15 @@ impl EdgeClient { let delta = match (&c.body, covered) { (_, true) => { t.stale &= t.watermark != Some(c.through); + let live = c.through; + t.unapplied + .retain(|(at, _)| at.same_epoch(&live) && at.publication > live.publication); + if t.restructured.is_some_and(|r| !r.same_epoch(&live)) { + t.restructured = None; + } + if t.seen.is_some_and(|s| !s.same_epoch(&live)) { + t.seen = Some(live); + } return (Outcome::Duplicate, Vec::new()); } (CorrectionBody::Window, false) => &target - &t.held, @@ -312,7 +335,8 @@ impl EdgeClient { match edge.call(request) { Ok(Reply::Correction(c)) => events.extend(self.apply(&c).1), Ok(Reply::Notices(list)) => { - list.iter().for_each(|n| events.extend(self.on_notice(n))); + list.iter() + .for_each(|n| events.extend(self.accept(n, Some(shape)))); } Err(EdgeError::Failed) => { self.request_failed(shape); @@ -346,7 +370,9 @@ impl EdgeClient { }; let mut events = Vec::new(); match edge.call(request) { - Ok(Reply::Notices(list)) => list.iter().for_each(|n| events.extend(self.on_notice(n))), + Ok(Reply::Notices(list)) => list + .iter() + .for_each(|n| events.extend(self.accept(n, Some(shape)))), Ok(Reply::Correction(c)) => events.extend(self.apply(&c).1), Err(EdgeError::Failed) => { self.request_failed(shape); diff --git a/crates/improvise-protocol/src/delivery/mod.rs b/crates/improvise-protocol/src/delivery/mod.rs index 6d85bee0..bb30c512 100644 --- a/crates/improvise-protocol/src/delivery/mod.rs +++ b/crates/improvise-protocol/src/delivery/mod.rs @@ -15,3 +15,6 @@ pub use wire::{Cell, WireElem, WireKey, WireNum, WireValue}; #[cfg(test)] mod client_tests; + +#[cfg(test)] +mod review_tests; diff --git a/crates/improvise-protocol/src/delivery/notice.rs b/crates/improvise-protocol/src/delivery/notice.rs index 616bf04d..476c27fb 100644 --- a/crates/improvise-protocol/src/delivery/notice.rs +++ b/crates/improvise-protocol/src/delivery/notice.rs @@ -85,7 +85,8 @@ pub enum AxisChange { ItemInserted { category: String, item: String, - at: usize, + #[serde(with = "wide")] + at: u128, }, ItemRemoved { category: String, diff --git a/crates/improvise-protocol/src/delivery/review_tests.rs b/crates/improvise-protocol/src/delivery/review_tests.rs new file mode 100644 index 00000000..3074d0d2 --- /dev/null +++ b/crates/improvise-protocol/src/delivery/review_tests.rs @@ -0,0 +1,183 @@ +//! Findings from the review of #348: each is a hand-built history that failed before its fix. + +use serde::{Deserialize, Serialize}; + +use super::client::*; +use super::exchange::*; +use super::notice::*; +use super::wire::{self, Cell, WireElem, WireKey, WireValue, encode}; + +fn at(host: u64, generation: u64, publication: u64) -> Stamp { + Stamp { + host, + generation, + publication, + } +} + +fn shape(name: &str) -> ShapeKey { + ShapeKey(name.into()) +} + +fn window() -> Window { + Window { + top: 0, + left: 0, + rows: 10, + cols: 10, + } +} + +fn key(n: u8) -> WireKey { + WireKey(vec![("C".into(), n.to_string())]) +} + +fn cell(k: u8, v: &str) -> Cell { + (key(k), WireValue::Text(v.into())) +} + +fn target(cells: &[Cell]) -> Vec { + encode(&cells.iter().cloned().map(|c| (c, 1)).collect()) +} + +fn window_correction(name: &str, through: Stamp, cells: &[Cell]) -> Correction { + Correction { + shape: shape(name), + at: through, + from: None, + through, + body: CorrectionBody::Window, + target: target(cells), + carried: None, + } +} + +fn synced(name: &str, through: Stamp, cells: &[Cell]) -> EdgeClient { + let mut c = EdgeClient::default(); + c.track(shape(name), window()); + c.apply(&window_correction(name, through, cells)); + c +} + +#[test] +fn a_late_window_reply_from_an_older_generation_does_not_regress() { + let mut c = synced("s", at(1, 2, 5), &[cell(1, "new")]); + let late = window_correction("s", at(1, 1, 9), &[cell(1, "old")]); + assert_eq!(c.apply(&late).0, Outcome::Duplicate); + assert_eq!(c.watermark(&shape("s")), Some(at(1, 2, 5))); + let kept = wire::decode(&target(&[cell(1, "new")])); + assert_eq!(c.held(&shape("s")), Some(&kept)); +} + +#[test] +fn a_late_notice_from_a_dead_host_settles_against_the_live_one() { + let mut c = synced("s", at(2, 1, 4), &[cell(1, "a")]); + c.on_notice(&Notice::empty(PubRange::one(at(1, 1, 9)))); + assert!(matches!( + c.wants(&shape("s")), + Some(Want::Correction { .. }) + )); + let live = window_correction("s", at(2, 1, 4), &[cell(1, "a")]); + assert_eq!(c.apply(&live).0, Outcome::Duplicate); + assert_eq!( + c.wants(&shape("s")), + None, + "the dead host's notice is dropped" + ); + c.on_notice(&Notice::empty(PubRange::one(at(2, 1, 5)))); + assert_eq!( + c.wants(&shape("s")), + None, + "and the live host's next one is contiguous" + ); +} + +struct Replies(Vec); + +impl Edge for Replies { + fn call(&mut self, _: Request) -> Result { + match self.0.is_empty() { + true => Err(EdgeError::Failed), + false => Ok(self.0.remove(0)), + } + } +} + +#[test] +fn a_poll_for_one_shape_leaves_the_other_shapes_alone() { + let (a, b) = (shape("a"), shape("b")); + let mut c = EdgeClient::default(); + for s in [&a, &b] { + c.track(s.clone(), window()); + c.apply(&window_correction(&s.0, at(1, 1, 4), &[])); + } + let mut pulled = Notice::empty(PubRange::one(at(1, 1, 5))); + pulled.moved.insert(a.clone(), Moved::Keys([key(1)].into())); + let mut edge = Replies(vec![Reply::Notices(vec![pulled])]); + c.poll(&mut edge, &a); + assert_eq!(c.watermark(&b), Some(at(1, 1, 4)), "b was not asked about"); + let mut next = Notice::empty(PubRange::one(at(1, 1, 6))); + next.moved.insert(b.clone(), Moved::Keys([key(2)].into())); + c.on_notice(&next); + assert!( + matches!(c.wants(&b), Some(Want::Correction { .. })), + "b missed publication 5, and must see the gap: {:?}", + c.wants(&b) + ); +} + +#[test] +fn a_views_flag_makes_the_client_fetch_its_window() { + let mut c = synced("s", at(1, 1, 4), &[cell(1, "a")]); + let mut n = Notice::empty(PubRange::one(at(1, 1, 5))); + n.flags.views = true; + c.on_notice(&n); + assert!(matches!( + c.wants(&shape("s")), + Some(Want::Fetch { window: true, .. }) + )); +} + +#[test] +fn wide_integers_are_canonical_on_input() { + #[derive(Debug, PartialEq, Serialize, Deserialize)] + struct P( + #[serde(with = "wire::wide")] u64, + #[serde(with = "wire::wide_weight")] i64, + ); + let bad = [ + r#"[9007199254740992,1]"#, + r#"["5",1]"#, + r#"["+9007199254740992",1]"#, + r#"["09007199254740992",1]"#, + r#"[1,9007199254740992]"#, + r#"[1,"5"]"#, + r#"[1,"-5"]"#, + r#"[1,"+9007199254740992"]"#, + r#"[1,"-9007199254740992 "]"#, + ]; + for text in bad { + assert!(serde_json::from_str::

(text).is_err(), "{text}"); + } + let good = r#"["9007199254740992","-9007199254740992"]"#; + assert_eq!( + serde_json::from_str::

(good).unwrap(), + P(1 << 53, -(1 << 53)) + ); + assert_eq!( + serde_json::from_str::

("[9007199254740991,-9007199254740991]").unwrap(), + P((1 << 53) - 1, -((1 << 53) - 1)) + ); +} + +#[test] +fn an_item_insert_position_crosses_as_a_wide_integer() { + let change = AxisChange::ItemInserted { + category: "C".into(), + item: "i".into(), + at: 1 << 60, + }; + let json = serde_json::to_string(&change).unwrap(); + assert!(json.contains(r#""at":"1152921504606846976""#), "{json}"); + assert_eq!(serde_json::from_str::(&json).unwrap(), change); +} diff --git a/crates/improvise-protocol/src/delivery/wire.rs b/crates/improvise-protocol/src/delivery/wire.rs index 7686f805..a66fbf33 100644 --- a/crates/improvise-protocol/src/delivery/wire.rs +++ b/crates/improvise-protocol/src/delivery/wire.rs @@ -29,8 +29,17 @@ pub mod wide { pub fn deserialize<'de, T: TryFrom, D: Deserializer<'de>>(d: D) -> Result { let n: u128 = match Repr::deserialize(d)? { - Repr::Small(n) => n.into(), - Repr::Wide(text) => text.parse().map_err(serde::de::Error::custom)?, + Repr::Small(n) if u128::from(n) < EXACT_JSON => n.into(), + Repr::Small(_) => { + return Err(serde::de::Error::custom("a wide integer must be a string")); + } + Repr::Wide(text) => { + let n: u128 = text.parse().map_err(serde::de::Error::custom)?; + match n >= EXACT_JSON && n.to_string() == text { + true => n, + false => return Err(serde::de::Error::custom("not a canonical wide integer")), + } + } }; T::try_from(n).map_err(|_| serde::de::Error::custom("integer out of range")) } @@ -56,8 +65,16 @@ pub mod wide_weight { pub fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result { match Repr::deserialize(d)? { - Repr::Small(n) => Ok(n), - Repr::Wide(text) => text.parse().map_err(serde::de::Error::custom), + Repr::Small(n) if u128::from(n.unsigned_abs()) < EXACT_JSON => Ok(n), + Repr::Small(_) => Err(serde::de::Error::custom("a wide weight must be a string")), + Repr::Wide(text) => { + let n: i64 = text.parse().map_err(serde::de::Error::custom)?; + let wide = u128::from(n.unsigned_abs()) >= EXACT_JSON; + match wide && n.to_string() == text { + true => Ok(n), + false => Err(serde::de::Error::custom("not a canonical wide weight")), + } + } } } } diff --git a/crates/numtower/src/sum.rs b/crates/numtower/src/sum.rs index ce882009..c7d2c307 100644 --- a/crates/numtower/src/sum.rs +++ b/crates/numtower/src/sum.rs @@ -40,7 +40,9 @@ impl PartialEq for ExactSum { propagate(&mut mine); propagate(&mut theirs); let counts = |s: &ExactSum| (s.nan, s.pos_inf, s.neg_inf); - mine == theirs && counts(self) == counts(o) && self.negative_zero() == o.negative_zero() + let zero = mine.iter().all(|l| *l == 0); + let same_zero = !zero || self.negative_zero() == o.negative_zero(); + mine == theirs && counts(self) == counts(o) && same_zero } } diff --git a/crates/numtower/tests/zero_sign.rs b/crates/numtower/tests/zero_sign.rs index ae7cabbf..215821ba 100644 --- a/crates/numtower/tests/zero_sign.rs +++ b/crates/numtower/tests/zero_sign.rs @@ -87,3 +87,16 @@ fn a_sum_that_is_not_zero_never_reads_as_negative_zero() { sum.add(7.0, -1); assert_eq!(sum.value(), -2.0); } + +#[test] +fn equal_values_compare_equal_whatever_the_zero_counts_say() { + let mut with = ExactSum::new(); + with.add(-0.0, 1); + with.add(5.0, 1); + with.add(7.0, -1); + let mut without = ExactSum::new(); + without.add(5.0, 1); + without.add(7.0, -1); + assert_eq!(with.value(), without.value()); + assert_eq!(with, without); +}