fix(zset): a new edge receiver seeds its request tokens from a fresh base; the late-reply history polls one frame at a time

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

crates/zset/src/edge.rs

@@ -267,6 +267,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() {
@@ -306,7 +316,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

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