fix(protocol): a reply from an epoch the client has left changes nothing (improvise-z2vr)

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

crates/improvise-command/src/ui/delivery_tests.rs

@@ -1059,6 +1059,33 @@ fn a_fetch_covers_only_the_notices_it_named_when_a_later_commit_is_already_in_th
     held_equals_drawn(&client, &host);
 }
 
+#[test]
+fn a_reply_held_across_a_host_restart_cannot_pull_the_client_back() {
+    let mut first = host_instance(1);
+    let id = connect(&mut first);
+    let mut client = settled(&mut first);
+    send(&mut first, id, vec![set("North", 99.0)]);
+    let mut lost = Flaky::scripted(&mut first, vec![Some(Fault::DropReply)]);
+    client.connection_lost();
+    client.sync(&mut lost, &shape(), 6);
+    let in_flight = lost.replies.remove(0);
+
+    let mut second = host_instance(2);
+    client.connection_lost();
+    client.sync(&mut Flaky::new(&mut second), &shape(), 6);
+    held_equals_drawn(&client, &second);
+    let at = client.watermark(&shape()).unwrap();
+
+    let Reply::Correction(late) = in_flight else {
+        panic!("the lost reply was a correction")
+    };
+    assert_eq!(client.apply(&late).0, Outcome::Duplicate);
+    assert_eq!(client.watermark(&shape()), Some(at));
+    held_equals_drawn(&client, &second);
+    let (state, _) = client.sync(&mut Flaky::new(&mut second), &shape(), 6);
+    assert_eq!(state, Synced::Settled);
+}
+
 // ── the whole thing, at random ─────────────────────────────────────────────
 
 mod random {
@@ -1253,6 +1280,7 @@ mod random {
             host.take_notices();
             let mut held_back: Vec<(usize, Notice)> = Vec::new();
             let mut old: Vec<Reply> = Vec::new();
+            let mut epochs: Vec<(u64, u64)> = Vec::new();
 
             for (step, (op, weather)) in steps.iter().enumerate() {
                 match (commands(op), op) {
@@ -1288,6 +1316,21 @@ mod random {
                 let mut edge = Flaky::scripted(&mut host, script);
                 client.sync(&mut edge, &shape(), 6);
                 old.append(&mut edge.replies);
+                let moved_host = matches!(op, Op::Restart | Op::Swap);
+                if let (true, Some(Reply::Correction(c))) = (moved_host, old.get(step % old.len().max(1))) {
+                    client.apply(c);
+                }
+                if let Some(w) = client.watermark(&shape()) {
+                    let now = (w.host, w.generation);
+                    let seen_at = epochs.iter().position(|e| *e == now);
+                    prop_assert!(
+                        seen_at.is_none_or(|i| i + 1 == epochs.len()),
+                        "the client went back to a superseded epoch {now:?} after {epochs:?}"
+                    );
+                    if seen_at.is_none() {
+                        epochs.push(now);
+                    }
+                }
             }
 
             let rest: Vec<Notice> = held_back.drain(..).map(|(_, n)| n).collect();

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

@@ -53,6 +53,9 @@ pub enum Synced {
     Exhausted,
 }
 
+/// How many abandoned epochs a shape remembers.
+const LEFT_EPOCHS: usize = 64;
+
 #[derive(Debug, Clone)]
 struct Tracked {
     window: Window,
@@ -63,6 +66,8 @@ struct Tracked {
     unapplied: Vec<(Stamp, Moved)>,
     /// The last notice that moved an axis: until a window is read past it, positions are unknown.
     restructured: Option<Stamp>,
+    /// Epochs the watermark has moved out of: nothing from them applies again.
+    left: Vec<(u64, u64)>,
     stale: bool,
 }
 
@@ -92,6 +97,7 @@ impl EdgeClient {
                 seen: None,
                 unapplied: Vec::new(),
                 restructured: None,
+                left: Vec::new(),
                 stale: true,
             },
         );
@@ -246,7 +252,10 @@ impl EdgeClient {
         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) {
+        if older(t.watermark)
+            || older(t.seen)
+            || t.left.contains(&(c.through.host, c.through.generation))
+        {
             return (Outcome::Duplicate, Vec::new());
         }
         let covered = t
@@ -285,6 +294,11 @@ impl EdgeClient {
             }
         };
         t.held += &delta;
+        if let Some(w) = t.watermark.filter(|w| !w.same_epoch(&c.through)) {
+            t.left.push((w.host, w.generation));
+            let excess = t.left.len().saturating_sub(LEFT_EPOCHS);
+            t.left.drain(..excess);
+        }
         t.watermark = Some(c.through);
         let window = matches!(c.body, CorrectionBody::Window);
         t.stale &= !window;

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

@@ -181,3 +181,23 @@ fn an_item_insert_position_crosses_as_a_wide_integer() {
     assert!(json.contains(r#""at":"1152921504606846976""#), "{json}");
     assert_eq!(serde_json::from_str::<AxisChange>(&json).unwrap(), change);
 }
+
+#[test]
+fn a_late_window_reply_from_a_dead_host_does_not_regress() {
+    let mut c = synced("s", at(1, 1, 9), &[cell(1, "old")]);
+    c.apply(&window_correction("s", at(2, 1, 3), &[cell(1, "new")]));
+    let late = window_correction("s", at(1, 1, 12), &[cell(1, "dead")]);
+    assert_eq!(c.apply(&late).0, Outcome::Duplicate);
+    assert_eq!(c.watermark(&shape("s")), Some(at(2, 1, 3)));
+    let kept = wire::decode(&target(&[cell(1, "new")]));
+    assert_eq!(c.held(&shape("s")), Some(&kept));
+}
+
+#[test]
+fn a_reply_from_a_generation_the_client_left_does_not_return() {
+    let mut c = synced("s", at(1, 1, 9), &[cell(1, "old")]);
+    c.apply(&window_correction("s", at(1, 2, 1), &[cell(1, "new")]));
+    c.apply(&window_correction("s", at(1, 3, 1), &[cell(1, "newer")]));
+    let late = window_correction("s", at(1, 2, 7), &[cell(1, "gen2")]);
+    assert_eq!(c.apply(&late).0, Outcome::Duplicate);
+}