fix(protocol): older epochs and dead hosts cannot regress a client, pulls are per shape, wire integers are canonical (improvise-z2vr)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze

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<Event> {
+        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<Event> {
         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<Stamp>| {
+            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);

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;

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,

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<WireElem> {
+    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<Reply>);
+
+impl Edge for Replies {
+    fn call(&mut self, _: Request) -> Result<Reply, EdgeError> {
+        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::<P>(text).is_err(), "{text}");
+    }
+    let good = r#"["9007199254740992","-9007199254740992"]"#;
+    assert_eq!(
+        serde_json::from_str::<P>(good).unwrap(),
+        P(1 << 53, -(1 << 53))
+    );
+    assert_eq!(
+        serde_json::from_str::<P>("[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::<AxisChange>(&json).unwrap(), change);
+}

crates/improvise-protocol/src/delivery/wire.rs

@@ -29,8 +29,17 @@ pub mod wide {
 
     pub fn deserialize<'de, T: TryFrom<u128>, D: Deserializer<'de>>(d: D) -> Result<T, D::Error> {
         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<i64, D::Error> {
         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")),
+                }
+            }
         }
     }
 }

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
     }
 }
 

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);
+}