From 6d7cd481f977ff6b6fc0f2446f7cd460b754597e Mon Sep 17 00:00:00 2001 From: "Claude (dbsp-threads)" Date: Sat, 10 Oct 2026 21:59:31 -0700 Subject: [PATCH] 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 Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze --- context/code/zset.md | 10 ++++-- context/test-evidence/zset.md | 19 ++++++++--- crates/zset/src/edge.rs | 12 ++++++- crates/zset/tests/edge.rs | 4 ++- crates/zset/tests/edge_net.rs | 59 +++++++++++++++++++++++++++++++---- 5 files changed, 90 insertions(+), 14 deletions(-) diff --git a/context/code/zset.md b/context/code/zset.md index 363a200c..ec838232 100644 --- a/context/code/zset.md +++ b/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` turns a `Frame` into `Wire` and back, with `CodecError` for what does not decode; `Direct` is the identity. `improvise-protocol` is to diff --git a/context/test-evidence/zset.md b/context/test-evidence/zset.md index 59d68e07..6361d626 100644 --- a/context/test-evidence/zset.md +++ b/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 @@ -767,3 +763,18 @@ Rival: `unsafe impl Sync for Circuit {}` fails the build of `tests/threads.rs`. A sink over `Counted` keys clones each key once (the insert into its sum). Rival: a sink that clones its input piece first, as the reviewer injected, makes `N` more. + +### 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. diff --git a/crates/zset/src/edge.rs b/crates/zset/src/edge.rs index 72f35872..8b23e41b 100644 --- a/crates/zset/src/edge.rs +++ b/crates/zset/src/edge.rs @@ -267,6 +267,16 @@ impl, P: Pipe> EdgeSink { } } +/// 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(z: &ZSet, d: &ZSet) { for (k, w) in d.iter() { @@ -306,7 +316,7 @@ impl, P: Pipe> EdgeSou held: ZSet::new(), stale: true, asked: false, - token: 0, + token: fresh_token_base(), marks: 0, ask_marks: 0, noted: None, diff --git a/crates/zset/tests/edge.rs b/crates/zset/tests/edge.rs index 46995044..5062ccde 100644 --- a/crates/zset/tests/edge.rs +++ b/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(); diff --git a/crates/zset/tests/edge_net.rs b/crates/zset/tests/edge_net.rs index 35ec8ef5..4a0eaabc 100644 --- a/crates/zset/tests/edge_net.rs +++ b/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, @@ -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::(); + 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::(); @@ -246,6 +257,7 @@ enum Act { Garble(usize, usize), RestartSink(u64), RestartSource, + NewSource, } fn act() -> impl Strategy { @@ -263,6 +275,7 @@ fn act() -> impl Strategy { 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); }