Merge gh/main (08bacbe2) into dbsp/zset-edge

context/code/delivery.md

@@ -127,8 +127,25 @@ for the reply to a request they just sent, takes an epoch off `left` before it a
 reply that arrives on its own, and keeps the guard. A stale reply can still move the client for one round;
 the next request it makes is answered by the live host and brings it back.
 
-`poll` hands a pulled notice only to the shape it asked about. A notice carries only that shape's keys,
+**A notice from another epoch only marks the shape stale; `seen` moves only through an answer.** The
-and applying it to another tracked shape would advance that shape past keys it never saw.
+first two livelocks (improvise-z2vr) were the same defect reached by different routes, which is why
+ignoring one kind of foreign notice after another was not enough. Epochs of different hosts are
+unordered, so a notice from an epoch other than `seen`'s cannot be placed. Reading it used to rewind
+`seen` into the foreign epoch, and then the live host's own old notices looked new, queued work and set
+`restructured`; the answers that followed were covered by the watermark, were duplicates, and never
+cleared it. Now such a notice sets `stale` and nothing else, the next request is answered by the live
+host, and that answer (`apply_answer`) is what moves `seen`. Two cheaper guards stay: a notice from an
+older generation of the host the shape is on, or from an epoch it has left, is ignored without the extra
+round; and a duplicate answer clears a `restructured` mark it covers. The last is defensive: no test
+reaches it now that `seen` cannot be rewound, and it is kept because a covered answer that cannot clear a mark
+it covers is the shape of both livelocks. Merging notices cut for different shapes, or a cut one with an
+uncut one, is refused. A notice the client asked for
+(`poll`) skips the left-epoch test, since the host that answers is live.
+
+`poll` hands a pulled notice only to the shape it asked about, and the notice says so: `for_shape` sets
+`Notice::cut_for`, and `accept` limits any notice that carries it to that shape, however it reaches the
+client (`on_notice` included). Without the mark, a pulled notice delivered as if pushed made the other
+shape skip keys it never saw.
 
 `connection_lost`, `request_failed` and `unreadable` set `stale`, which `wants` turns into
 `Ask::Correction` from the watermark the client held. A duplicate whose `through` equals the watermark

context/test-evidence/delivery.md

@@ -192,3 +192,58 @@ test is the pin. Rival: the left-epoch guard removed fails
 The property's invariant changed with it. "The watermark never returns to an epoch it left" was false by
 design once an unknown-epoch reply can move the client, so the property now checks the narrower, true
 rule: a replayed reply from an epoch the client was in and has left must be a duplicate.
+
+## The old-notice livelock
+
+### a_late_notice_from_an_older_generation_cannot_rewind_the_client
+
+The property failed with `Exhausted` on a shrunk history (two swaps, a transpose, a limit change, the first
+swap's notice delivered three steps late, an old reply replayed). The client had settled on generation 2 when
+the generation-1 notice arrived: `seen` was rewound into generation 1, a notice from the same history set
+`restructured`, and the window correction that answered was covered by the watermark, so it never cleared
+the mark and `poll` retried until `Exhausted`. The test replays the history step for step, with the
+property's own weather and a Flaky edge. Rival at the time: the notice-ignore test in `accept` disabled;
+this test and the random property failed. **Since the round-three change (a foreign notice only marks the
+shape stale) nothing fails with that shortcut disabled**: it is an optimisation that saves one round, not a
+correctness rule. The test still replays the seed. The seed file is not committed
+(`proptest-regressions/` is ignored).
+
+Run after the fix, the property passed 4 x 4000 cases and 2 x 20000 cases, with no third failure.
+
+## Round three: the same livelock by other routes
+
+### route_a_a_dead_hosts_notice_cannot_make_the_live_hosts_old_notices_look_new
+
+A client goes host 1 to host 3 and never joins host 2. A late host-2 notice rewinds `seen`, then host 3's
+own old notices look new and set `restructured` at the watermark, and the answers are duplicates that
+never clear it. Failed on the head before the change (`wants` stayed `Fetch { window: true }`).
+
+### route_b_a_covered_reply_cannot_set_seen_below_a_queued_entry
+
+The same rewind, then an unsolicited covered correction: the duplicate branch set `seen` to its `through`,
+below a queued entry, and each fetch for that `through` came back a duplicate. Failed on the head before
+the change.
+
+### a_foreign_notice_marks_the_shape_stale_and_the_home_hosts_answer_settles_it
+
+A foreign notice sets `stale`, and the home host's answer then settles the shape. It does not see `seen`
+stay put (an answer heals it either way); route A and route B are what pin that.
+
+### notices_cut_for_different_shapes_or_not_cut_do_not_merge
+
+A merge of notices cut for different shapes, or a cut one with an uncut one, would keep one shape's
+mark and make the other shape see a gap. `merge` refuses it. Rival: the `cut_for` comparison removed.
+
+### a_pulled_notice_pushed_through_on_notice_still_reaches_only_its_shape
+
+A notice cut for shape `a` by `for_shape`, delivered through `on_notice`, must leave shape `b` alone.
+Failed before `Notice::cut_for` existed.
+
+### two_shapes_survive_late_repeated_and_dead_messages
+
+The reviewer's generator, kept: two shapes, notices late, repeated and reordered (the due set is reversed
+on odd steps), dead hosts' notices kept across restarts, stored notices and stored replies replayed on
+their own, hosts that restart often. At the default 256 cases it did not find the old head's livelock, and
+at 20,000 it did within ten seconds (an `Exhausted`), so the default run is a smoke test and the deep run
+is the check. After the change it passed three runs of 20,000 cases, and the older property passed
+2 x 20,000 with it.

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

@@ -122,8 +122,12 @@ fn refused(answer: &[ServerFrame]) -> bool {
 
 /// What the grid draws, derived without the delivery module's own reader.
 fn drawn(host: &Host, window: &Window) -> ZSet<Cell> {
+    drawn_view(host, "Default", window)
+}
+
+fn drawn_view(host: &Host, view: &str, window: &Window) -> ZSet<Cell> {
     let wb = &host.app().model_state.workbook;
-    let layout = GridLayout::with_frozen_records(&wb.model, &wb.views["Default"], None);
+    let layout = GridLayout::with_frozen_records(&wb.model, &wb.views[view], None);
     let mut cells = ZSet::new();
     for row in 0..layout.row_count() {
         for col in 0..layout.col_count() {
@@ -1271,8 +1275,17 @@ mod random {
         other.add_item("_Measure", "Units").unwrap();
         other.active_view_mut().set_axis("Product", Axis::Row);
         other.active_view_mut().set_axis("_Measure", Axis::Column);
+        let mut second = other.active_view().clone();
+        second.name = "Second".into();
+        other.views.insert("Second".into(), second);
         crate::persistence::save_to(&storage, &other, std::path::Path::new("other.improv"))
             .unwrap();
+        let mut wb = wb;
+        if !wb.views.contains_key("Second") {
+            let mut second = wb.active_view().clone();
+            second.name = "Second".into();
+            wb.views.insert("Second".into(), second);
+        }
         let mut app = App::new(wb, None);
         app.storage = Box::new(storage);
         app.rebuild_layout();
@@ -1300,7 +1313,10 @@ mod random {
         }
         let (due, later): (Vec<_>, Vec<_>) = held_back.drain(..).partition(|(at, _)| *at <= step);
         *held_back = later;
-        let notices: Vec<Notice> = due.into_iter().map(|(_, n)| n).collect();
+        let mut notices: Vec<Notice> = due.into_iter().map(|(_, n)| n).collect();
+        if step % 2 == 1 {
+            notices.reverse();
+        }
         deliver_all(client, &notices);
     }
 
@@ -1337,6 +1353,66 @@ mod random {
         Ok(())
     }
 
+    /// The shrunk history of the second livelock: two swaps, a transpose and a limit change, with
+    /// the first swap's notice delivered three steps late and an old reply replayed at the end.
+    #[test]
+    fn a_late_notice_from_an_older_generation_cannot_rewind_the_client() {
+        let window = Window {
+            top: 0,
+            left: 0,
+            rows: 1,
+            cols: 1,
+        };
+        let mut host = host_over(workbook(), 1);
+        let id = connect(&mut host);
+        let mut client = EdgeClient::new(session());
+        client.track(shape(), window);
+        client.sync(&mut Flaky::new(&mut host), &shape(), 6);
+        host.take_notices();
+        let mut held_back: Vec<(usize, Notice)> = Vec::new();
+        let mut old: Vec<Reply> = Vec::new();
+        let plan = [
+            (Some(0), Arrival::Late(3), vec![Some(0)], None),
+            (Some(0), Arrival::Dropped, vec![], None),
+            (Some(1), Arrival::Now, vec![], None),
+            (None, Arrival::Dropped, vec![], Some(6)),
+        ];
+        for (step, (op, arrival, sync, replay)) in plan.into_iter().enumerate() {
+            match op {
+                Some(0) => {
+                    send(
+                        &mut host,
+                        id,
+                        vec![ModelCommand::LoadModel("other.improv".into())],
+                    );
+                }
+                Some(_) => {
+                    send(&mut host, id, vec![transpose()]);
+                }
+                None => host.limit_delivery(2, 1),
+            }
+            let w = Weather {
+                arrival,
+                lose_connection: true,
+                sync: sync.clone(),
+                replay,
+            };
+            let fresh = host.take_notices();
+            arrive(&mut client, &mut held_back, step, &w, fresh);
+            client.connection_lost();
+            if let Some(Reply::Correction(c)) = replay.and_then(|i| old.get(i % old.len().max(1))) {
+                client.apply(c);
+            }
+            let script = sync.iter().map(|f| f.map(fault)).collect();
+            let mut edge = Flaky::scripted(&mut host, script);
+            client.sync(&mut edge, &shape(), 6);
+            old.append(&mut edge.replies);
+        }
+        let (state, _) = client.poll(&mut Flaky::new(&mut host), &shape());
+        assert_eq!(state, Synced::Settled, "an old notice must not queue work");
+        assert_eq!(client.held(&shape()), Some(&drawn(&host, &window)));
+    }
+
     proptest! {
         #[test]
         fn a_corrected_state_is_the_target_and_correcting_it_again_is_the_zero_zset(
@@ -1414,4 +1490,97 @@ mod random {
             prop_assert_eq!(client.held(&shape()).cloned(), before);
         }
     }
+
+    /// Hosts come and go often here, so a client can skip one and meet its notices later.
+    fn churn() -> impl Strategy<Value = Op> {
+        prop_oneof![
+            4 => (any::<bool>(), 0usize..8).prop_map(|(r, v)| Op::Set(r, v)),
+            4 => Just(Op::Restart),
+            2 => Just(Op::Transpose),
+            2 => Just(Op::Swap),
+            1 => (0usize..3).prop_map(Op::Add),
+            1 => Just(Op::TightLimits),
+        ]
+    }
+
+    fn window_pair() -> impl Strategy<Value = (Window, Window)> {
+        (windows(), windows())
+    }
+
+    proptest! {
+        /// Two shapes, notices late, repeated and reordered, a dead host's notices kept across
+        /// restarts, and stored notices and replies replayed on their own.
+        #[test]
+        fn two_shapes_survive_late_repeated_and_dead_messages(
+            steps in proptest::collection::vec(
+                (churn(), weather(), proptest::option::of(0usize..32)),
+                1..12,
+            ),
+            (wa, wb) in window_pair(),
+        ) {
+            let second = shape_of_view("Second");
+            let mut instance = 1;
+            let mut host = host_over(workbook(), instance);
+            let mut id = connect(&mut host);
+            let mut client = EdgeClient::new(session());
+            client.track(shape(), wa);
+            client.track(second.clone(), wb);
+            client.sync(&mut Flaky::new(&mut host), &shape(), 6);
+            client.sync(&mut Flaky::new(&mut host), &second, 6);
+            host.take_notices();
+            let mut held_back: Vec<(usize, Notice)> = Vec::new();
+            let mut pool: Vec<Notice> = Vec::new();
+            let mut old: Vec<Reply> = Vec::new();
+
+            for (step, (op, weather, ghost)) in steps.iter().enumerate() {
+                match (commands(op), op) {
+                    (cmds, _) if !cmds.is_empty() => {
+                        send(&mut host, id, cmds);
+                    }
+                    (_, Op::TightLimits) => host.limit_delivery(2, 1),
+                    (_, Op::Evict) => {
+                        host.evict_idle(10, 0);
+                    }
+                    _ => {
+                        instance += 1;
+                        let wb = host.app().model_state.workbook.clone();
+                        host = host_over(wb, instance);
+                        id = connect(&mut host);
+                    }
+                }
+                let fresh = host.take_notices();
+                pool.extend(fresh.iter().cloned());
+                arrive(&mut client, &mut held_back, step, weather, fresh);
+                if let Some(i) = ghost.filter(|_| !pool.is_empty()) {
+                    let n = &pool[i % pool.len()];
+                    client.on_notice(n);
+                    client.on_notice(&n.for_shape(&shape()));
+                }
+                if let Some(i) = weather.replay.filter(|_| !old.is_empty()) {
+                    match &old[i % old.len()] {
+                        Reply::Correction(c) => {
+                            client.apply(c);
+                        }
+                        Reply::Notices(list) => deliver_all(&mut client, list),
+                    }
+                }
+                if weather.lose_connection {
+                    client.connection_lost();
+                }
+                let script = weather.sync.iter().map(|f| f.map(fault)).collect();
+                let mut edge = Flaky::scripted(&mut host, script);
+                client.sync(&mut edge, &shape(), 6);
+                client.sync(&mut edge, &second, 6);
+                old.append(&mut edge.replies);
+            }
+
+            let rest: Vec<Notice> = held_back.drain(..).map(|(_, n)| n).collect();
+            deliver_all(&mut client, &rest);
+            for (name, window, view) in [(shape(), wa, "Default"), (second, wb, "Second")] {
+                let (state, _) = client.poll(&mut WireEdge(Flaky::new(&mut host)), &name);
+                prop_assert_eq!(state, Synced::Settled);
+                prop_assert_eq!(client.held(&name), Some(&drawn_view(&host, view, &window)));
+            }
+        }
+    }
 }

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

@@ -162,25 +162,36 @@ 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)
+        self.accept(notice, None, false)
     }
 
     /// `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> {
+    fn accept(&mut self, notice: &Notice, only: Option<&ShapeKey>, answer: bool) -> Vec<Event> {
+        let only = only.or(notice.cut_for.as_ref());
         let end = notice.stamp();
+        let (mut read, mut ignored) = (0, 0);
         for (shape, t) in &mut self.shapes {
             if only.is_some_and(|o| o != shape) {
                 continue;
             }
-            let contiguous = match t.seen {
+            read += 1;
-                Some(seen) if seen.same_epoch(&end) => match notice.range.to <= seen.publication {
+            let older = |s: Option<Stamp>| {
-                    true => continue,
+                s.is_some_and(|s| s.host == end.host && end.generation < s.generation)
-                    false => notice.range.from <= seen.publication,
-                },
-                _ => false,
             };
+            let left = !answer && t.left.contains(&(end.host, end.generation));
+            if older(t.watermark) || older(t.seen) || left {
+                ignored += 1;
+                continue;
+            }
+            let Some(seen) = t.seen.filter(|s| s.same_epoch(&end)) else {
+                t.stale = true;
+                continue;
+            };
+            if notice.range.to <= seen.publication {
+                continue;
+            }
             t.seen = Some(end);
-            if !contiguous || notice.evicted.contains(shape) {
+            if notice.range.from > seen.publication || notice.evicted.contains(shape) {
                 t.unapplied.push((end, Moved::Correct));
                 continue;
             }
@@ -197,7 +208,10 @@ impl EdgeClient {
                 t.watermark = Some(end);
             }
         }
-        self.events_of(notice)
+        match read > 0 && ignored == read {
+            true => Vec::new(),
+            false => self.events_of(notice),
+        }
     }
 
     /// The connection dropped: every shape asks for a correction from its watermark.
@@ -286,7 +300,9 @@ impl EdgeClient {
                 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)) {
+                let covered_mark =
+                    |r: Stamp| !r.same_epoch(&live) || r.publication <= live.publication;
+                if t.restructured.is_some_and(covered_mark) {
                     t.restructured = None;
                 }
                 if t.seen.is_some_and(|s| !s.same_epoch(&live)) {
@@ -365,7 +381,7 @@ impl EdgeClient {
                 Ok(Reply::Correction(c)) => events.extend(self.apply_answer(&c).1),
                 Ok(Reply::Notices(list)) => {
                     list.iter()
-                        .for_each(|n| events.extend(self.accept(n, Some(shape))));
+                        .for_each(|n| events.extend(self.accept(n, Some(shape), true)));
                 }
                 Err(EdgeError::Failed) => {
                     self.request_failed(shape);
@@ -401,7 +417,7 @@ impl EdgeClient {
         match edge.call(request) {
             Ok(Reply::Notices(list)) => list
                 .iter()
-                .for_each(|n| events.extend(self.accept(n, Some(shape)))),
+                .for_each(|n| events.extend(self.accept(n, Some(shape), true))),
             Ok(Reply::Correction(c)) => events.extend(self.apply_answer(&c).1),
             Err(EdgeError::Failed) => {
                 self.request_failed(shape);

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

@@ -157,6 +157,9 @@ pub struct Notice {
     pub flags: Flags,
     pub evicted: BTreeSet<ShapeKey>,
     pub moved: BTreeMap<ShapeKey, Moved>,
+    /// Set when the notice was cut for one shape (a pull): it must reach no other.
+    #[serde(default)]
+    pub cut_for: Option<ShapeKey>,
 }
 
 impl Notice {
@@ -168,6 +171,7 @@ impl Notice {
             flags: Flags::default(),
             evicted: BTreeSet::new(),
             moved: BTreeMap::new(),
+            cut_for: None,
         }
     }
 
@@ -178,7 +182,7 @@ impl Notice {
     /// `self` followed by `next`, if `next` starts where `self` ends. Anything else
     /// has a gap or another epoch in it, and merging would hide a loss.
     pub fn merge(mut self, next: Notice) -> Result<Notice, Box<(Notice, Notice)>> {
-        let adjacent = self.range.end() == next.range.start();
+        let adjacent = self.range.end() == next.range.start() && self.cut_for == next.cut_for;
         match adjacent {
             false => Err(Box::new((self, next))),
             true => {
@@ -207,6 +211,7 @@ impl Notice {
                 .get(shape)
                 .map(|m| BTreeMap::from([(shape.clone(), m.clone())]))
                 .unwrap_or_default(),
+            cut_for: Some(shape.clone()),
             ..self.clone()
         }
     }
@@ -269,6 +274,16 @@ mod tests {
         assert!(merged.moved.contains_key(&ShapeKey("b".into())));
     }
 
+    #[test]
+    fn notices_cut_for_different_shapes_or_not_cut_do_not_merge() {
+        let (a, b) = (ShapeKey("a".into()), ShapeKey("b".into()));
+        let cut = |p: u64, s: &ShapeKey| notice(p, &[]).for_shape(s);
+        assert!(cut(1, &a).merge(cut(2, &a)).is_ok());
+        assert!(cut(1, &a).merge(cut(2, &b)).is_err());
+        assert!(cut(1, &a).merge(notice(2, &[])).is_err());
+        assert!(notice(1, &[]).merge(cut(2, &a)).is_err());
+    }
+
     #[test]
     fn a_gap_does_not_merge() {
         let (a, b) = *notice(3, &[]).merge(notice(5, &[])).unwrap_err();

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

@@ -201,3 +201,109 @@ fn a_reply_from_a_generation_the_client_left_does_not_return() {
     let late = window_correction("s", at(1, 2, 7), &[cell(1, "gen2")]);
     assert_eq!(c.apply(&late).0, Outcome::Duplicate);
 }
+
+fn keyed(from: Stamp, name: &str, keys: &[u8]) -> Notice {
+    let mut n = Notice::empty(PubRange::one(from));
+    n.moved.insert(
+        shape(name),
+        Moved::Keys(keys.iter().map(|k| key(*k)).collect()),
+    );
+    n
+}
+
+/// A client that went host 1 -> host 3 and never joined host 2, at publication 2 of host 3.
+fn on_host_three() -> EdgeClient {
+    let mut c = synced("s", at(1, 0, 0), &[cell(1, "a")]);
+    c.apply_answer(&window_correction("s", at(3, 0, 2), &[cell(1, "a")]));
+    c
+}
+
+#[test]
+fn route_a_a_dead_hosts_notice_cannot_make_the_live_hosts_old_notices_look_new() {
+    let mut c = on_host_three();
+    c.on_notice(&Notice::empty(PubRange::one(at(2, 0, 1))));
+    c.on_notice(&Notice::empty(PubRange::one(at(3, 0, 1))));
+    let mut axis = Notice::empty(PubRange::one(at(3, 0, 2)));
+    axis.axis.push(AxisChange::CategoryAdded {
+        category: "X".into(),
+    });
+    c.on_notice(&axis);
+    for _ in 0..2 {
+        if c.wants(&shape("s")).is_some() {
+            c.apply_answer(&window_correction("s", at(3, 0, 2), &[cell(1, "a")]));
+        }
+    }
+    assert_eq!(
+        c.wants(&shape("s")),
+        None,
+        "the live host's answer settles it"
+    );
+}
+
+#[test]
+fn route_b_a_covered_reply_cannot_set_seen_below_a_queued_entry() {
+    let mut c = on_host_three();
+    c.on_notice(&keyed(at(3, 0, 3), "s", &[1]));
+    c.on_notice(&Notice::empty(PubRange::one(at(2, 0, 4))));
+    c.apply(&window_correction("s", at(3, 0, 1), &[cell(1, "a")]));
+    for _ in 0..4 {
+        match c.wants(&shape("s")) {
+            Some(Want::Fetch {
+                watermark,
+                keys,
+                through,
+                ..
+            }) => {
+                let moved = Correction {
+                    shape: shape("s"),
+                    at: at(3, 0, 3),
+                    from: watermark,
+                    through,
+                    body: CorrectionBody::Moved { keys },
+                    target: target(&[cell(1, "b")]),
+                    carried: None,
+                };
+                c.apply_answer(&moved);
+            }
+            Some(Want::Correction { .. }) => {
+                c.apply_answer(&window_correction("s", at(3, 0, 3), &[cell(1, "b")]));
+            }
+            None => {}
+        }
+    }
+    assert_eq!(c.wants(&shape("s")), None);
+}
+
+#[test]
+fn a_foreign_notice_marks_the_shape_stale_and_the_home_hosts_answer_settles_it() {
+    let mut c = synced("s", at(1, 0, 4), &[cell(1, "a")]);
+    c.on_notice(&Notice::empty(PubRange::one(at(2, 0, 9))));
+    assert!(matches!(
+        c.wants(&shape("s")),
+        Some(Want::Correction { .. })
+    ));
+    c.on_notice(&Notice::empty(PubRange::one(at(1, 0, 5))));
+    c.apply_answer(&window_correction("s", at(1, 0, 5), &[cell(1, "a")]));
+    assert_eq!(
+        c.wants(&shape("s")),
+        None,
+        "the home host's next notice was contiguous"
+    );
+}
+
+#[test]
+fn a_pulled_notice_pushed_through_on_notice_still_reaches_only_its_shape() {
+    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, 0, 4), &[]));
+    }
+    let mut both = Notice::empty(PubRange::one(at(1, 0, 5)));
+    both.moved.insert(a.clone(), Moved::Keys([key(1)].into()));
+    both.moved.insert(b.clone(), Moved::Keys([key(2)].into()));
+    c.on_notice(&both.for_shape(&a));
+    assert_eq!(c.watermark(&b), Some(at(1, 0, 4)));
+    assert_eq!(c.wants(&b), None, "b was not told anything");
+    assert!(matches!(c.wants(&a), Some(Want::Fetch { .. })));
+}