Merge dbsp/zset-edge into dbsp/sched

# Conflicts:
#	context/test-evidence/zset.md

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,
-and applying it to another tracked shape would advance that shape past keys it never saw.
+**A notice from another epoch only marks the shape stale; `seen` moves only through an answer.** The
+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/code/zset.md

@@ -362,11 +362,17 @@ Nothing here spawns a thread; whoever owns each circuit calls `pump` and `poll`.
    (`a_new_epoch_frame_re_asks_once_after_a_lost_request`). A sender that reuses an epoch id at
    a lower seq cannot be told from a reordered old frame, so the receiver ignores it as a
    duplicate and is stranded until the sender restarts again: the caller must pass a fresh epoch.
-   A receiver that restarts (`EdgeSource::restart`) is stale again.
+   A receiver that restarts (`EdgeSource::restart`) is stale again and keeps its token counter.
+   A brand-new receiver (`EdgeSource::new`, a process that lost its state) seeds the counter from
+   a hash of a process-wide counter under `RandomState`: random across processes natively and
+   unique per process on wasm32, which has no entropy. Seeding at 0 let a dead receiver's late
+   reply match a new receiver's request and join an older incarnation
+   (`review_token_collision_after_new_source`).
 8. **A lost final message is invisible to the receiver.** `connection_lost()` is the host's
    timeout: it marks the receiver stale and lets it ask again. The host calls it on its own local
    monotonic timeout, never by comparing times across machines, and it must be longer than the
-   round trip, or every answer is one request behind. Requests carry the receiver's stamp and
+   round trip, or every answer is one request behind; a host that times out repeatedly must back
+   off, not retry at a fixed short interval. Requests carry the receiver's stamp and
    this sender does not read it.
 9. **The wire form is a hook.** `EdgeCodec<K>` turns a `Frame<K>` into `Wire` and back, with
    `CodecError` for what does not decode; `Direct` is the identity. `improvise-protocol` is to

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.

context/test-evidence/zset.md

@@ -726,10 +726,6 @@ After every action the receiver's applied state equals the sender's integral at
 claims and its epoch never goes down; after the network is reliable it catches up. Run at 20,000
 cases three times, all green. Rivals N4 and N7 (a foreign delta moves the receiver) fail it.
 
-### the_reviewers_late_reply_case_is_a_history
-
-The late-reply sequence as a fixed history through the same World.
-
 ### an_unreadable_frame_asks_for_a_correction
 
 A frame that does not decode sets `stale`. A delta before the join is ignored and a correction
@@ -778,3 +774,18 @@ clones its input piece first, as the reviewer injected, makes `N` more.
 `EdgeSink::restart` and `EdgeSource::restart` kept the frame counters at 0 (they rebuilt the
 struct), which could make a unit read "no progress" and lose a kick. Rival: either restart
 resetting its counter fails it.
+
+### review_token_collision_after_new_source
+
+The old receiver's request 2 is answered by E1 and the reply is delayed; E1 dies, E2 takes over;
+a brand-new receiver joins E2, times out, asks its request 2, and the dead reply matched it and
+took the receiver back to epoch 1. Tokens started at 0 for every `new`. Now they start at a
+random base. Rival T1 (base 0) fails it. The deep property found nothing at 256 cases; this
+history is the check.
+
+### the_reviewers_late_reply_case_is_a_history
+
+Now polls one frame at a time, so each late frame is judged alone (one `Poll(4)` hid it). Rival
+N2 (a foreign correction accepted) fails it. N1 (any reply while asked) is caught by
+`only_the_reply_to_the_latest_request_may_join_a_new_incarnation` and the collision history; the
+random property needs a deep run (20,000 cases) to find N1, and finds N2 at the default 256.

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 {
-                Some(seen) if seen.same_epoch(&end) => match notice.range.to <= seen.publication {
-                    true => continue,
-                    false => notice.range.from <= seen.publication,
-                },
-                _ => false,
+            read += 1;
+            let older = |s: Option<Stamp>| {
+                s.is_some_and(|s| s.host == end.host && end.generation < s.generation)
             };
+            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 { .. })));
+}

crates/zset/src/edge.rs

@@ -280,6 +280,16 @@ impl<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
     }
 }
 
+/// A token base no other receiver incarnation shares (`context/code/zset.md`, "Edges", item 7):
+/// a hash of a process-wide counter under `RandomState`.
+fn fresh_token_base() -> u64 {
+    use std::hash::BuildHasher;
+    use std::sync::atomic::{AtomicU64, Ordering};
+    static NEXT: AtomicU64 = AtomicU64::new(1);
+    let n = NEXT.fetch_add(1, Ordering::Relaxed);
+    std::collections::hash_map::RandomState::new().hash_one(n) >> 1
+}
+
 /// Panics, before anything changes, if adding `d` to `z` would overflow a weight.
 fn preflight<K: Ord>(z: &ZSet<K>, d: &ZSet<K>) {
     for (k, w) in d.iter() {
@@ -320,7 +330,7 @@ impl<K: Ord + Clone + Send + 'static, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSou
             held: ZSet::new(),
             stale: true,
             asked: false,
-            token: 0,
+            token: fresh_token_base(),
             marks: 0,
             ask_marks: 0,
             noted: None,

crates/zset/tests/edge.rs

@@ -782,7 +782,9 @@ fn an_unreadable_frame_asks_for_a_correction() {
         src.stale(),
         "[initial] a new receiver asks for a correction"
     );
-    b.try_send("C 1 1 1 1:1".to_string()).unwrap();
+    let request = b.try_recv().unwrap();
+    let token = request.split(' ').nth(1).unwrap();
+    b.try_send(format!("C 1 1 {token} 1:1")).unwrap();
     src.poll(&mut c, 10);
     assert!(!src.stale());
     b.try_send("garbage".to_string()).unwrap();

crates/zset/tests/edge_net.rs

@@ -67,10 +67,6 @@ fn net(cap: usize) -> (Shared, End, End) {
     (shared, sink_end, src_end)
 }
 
-fn z(entries: &[(u8, i64)]) -> Z {
-    entries.iter().copied().collect()
-}
-
 struct World {
     net: Shared,
     sink: EdgeSink<u8, Text, End>,
@@ -134,6 +130,21 @@ impl World {
         self.history.insert((self.epoch, seq), self.total.clone());
     }
 
+    /// The consumer process lost its state: a brand-new receiver, with no token to carry over.
+    fn new_source(&mut self) {
+        let mut cons = Circuit::new();
+        let (i, s) = cons.input::<u8>();
+        self.out = cons.output(&s);
+        let end = End {
+            net: self.net.clone(),
+            out: 1,
+        };
+        self.src = EdgeSource::new(Text, end, i);
+        self.cons = cons;
+        self.got = Z::new();
+        self.joined_max = 0;
+    }
+
     fn restart_source(&mut self) {
         let mut cons = Circuit::new();
         let (i, s) = cons.input::<u8>();
@@ -246,6 +257,7 @@ enum Act {
     Garble(usize, usize),
     RestartSink(u64),
     RestartSource,
+    NewSource,
 }
 
 fn act() -> impl Strategy<Value = Act> {
@@ -263,6 +275,7 @@ fn act() -> impl Strategy<Value = Act> {
         1 => (lane, at).prop_map(|(l, i)| Act::Garble(l, i)),
         1 => (0u64..8).prop_map(Act::RestartSink),
         1 => Just(Act::RestartSource),
+        1 => Just(Act::NewSource),
     ]
 }
 
@@ -293,6 +306,7 @@ fn run(cap: usize, acts: &[Act]) {
             Act::Garble(d, i) => w.lane(*d, *i, |l, at| l.flight[at] = "garbage".to_string()),
             Act::RestartSink(seq) => w.restart_sink(*seq),
             Act::RestartSource => w.restart_source(),
+            Act::NewSource => w.new_source(),
         }
         w.check(&format!("after action {n} ({a:?})"));
     }
@@ -313,6 +327,7 @@ proptest! {
 #[test]
 fn the_reviewers_late_reply_case_is_a_history() {
     // Epoch 1 reads a request, its reply is delayed, the sender restarts, and the late reply comes.
+    // One frame per poll, so each is judged on its own.
     let acts = [
         Act::Poll(1),
         Act::Pump,
@@ -325,9 +340,41 @@ fn the_reviewers_late_reply_case_is_a_history() {
         Act::Deliver(1, 0),
         Act::Pump,
         Act::Deliver(0, 1),
+        Act::Poll(1),
+        Act::Deliver(0, 0),
+        Act::Poll(1),
+        Act::Poll(1),
+    ];
+    run(2, &acts);
+}
+
+#[test]
+fn review_token_collision_after_new_source() {
+    // The old receiver's request 2 is answered by E1 and the reply is delayed. E1 dies, E2 takes
+    // over, a new receiver joins E2 and times out, and the dead reply arrives for its request 2.
+    let acts = [
+        Act::Poll(1),
+        Act::Deliver(1, 0),
+        Act::Pump,
+        Act::Deliver(0, 0),
+        Act::Poll(4),
+        Act::Commit(vec![(1, 1)]),
+        Act::Timeout,
+        Act::Poll(1),
+        Act::Deliver(1, 0),
+        Act::Pump,
+        Act::RestartSink(5),
+        Act::Commit(vec![(2, 1)]),
+        Act::NewSource,
+        Act::Poll(1),
+        Act::Deliver(1, 0),
+        Act::Pump,
+        Act::Deliver(0, 1),
+        Act::Poll(4),
+        Act::Timeout,
+        Act::Poll(1),
         Act::Deliver(0, 0),
         Act::Poll(4),
     ];
-    run(2, &acts);
-    let _ = z(&[]);
+    run(4, &acts);
 }