From f47efef5aea5960abc5c344b7f23778db4426df4 Mon Sep 17 00:00:00 2001 From: "Claude (dbsp-delivery)" Date: Sat, 10 Oct 2026 19:35:49 -0700 Subject: [PATCH 1/4] fix(protocol): a notice from an older generation or a left epoch is ignored (improvise-z2vr) Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze --- context/code/delivery.md | 7 +++ context/test-evidence/delivery.md | 14 +++++ .../src/ui/delivery_tests.rs | 60 +++++++++++++++++++ .../improvise-protocol/src/delivery/client.rs | 23 +++++-- 4 files changed, 99 insertions(+), 5 deletions(-) diff --git a/context/code/delivery.md b/context/code/delivery.md index 17db04d6..45f29e67 100644 --- a/context/code/delivery.md +++ b/context/code/delivery.md @@ -127,6 +127,13 @@ 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. +**A notice from an older generation of the host the client is on, or from an epoch it has left, is +ignored.** Reading one rewinds `seen` into the old epoch and queues a correction entry, and the +correction that answers it is then covered by the watermark and a duplicate, so a `restructured` mark +set from the same notice is never cleared and `sync` returns `Exhausted`. A notice the client asked for +(`poll`) skips the left-epoch test, since the host that answers is live (see the answer rule above); it +still skips an older generation. When every shape ignores a notice, its events are dropped too. + `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. diff --git a/context/test-evidence/delivery.md b/context/test-evidence/delivery.md index fdb92e52..24a80264 100644 --- a/context/test-evidence/delivery.md +++ b/context/test-evidence/delivery.md @@ -192,3 +192,17 @@ 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: the notice-ignore test in `accept` disabled; this test +and the random property fail. 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. diff --git a/crates/improvise-command/src/ui/delivery_tests.rs b/crates/improvise-command/src/ui/delivery_tests.rs index dcad00d6..0e528ef0 100644 --- a/crates/improvise-command/src/ui/delivery_tests.rs +++ b/crates/improvise-command/src/ui/delivery_tests.rs @@ -1337,6 +1337,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 = 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( diff --git a/crates/improvise-protocol/src/delivery/client.rs b/crates/improvise-protocol/src/delivery/client.rs index 87a20a77..83f3305c 100644 --- a/crates/improvise-protocol/src/delivery/client.rs +++ b/crates/improvise-protocol/src/delivery/client.rs @@ -162,16 +162,26 @@ 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 { - 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 { + fn accept(&mut self, notice: &Notice, only: Option<&ShapeKey>, answer: bool) -> Vec { 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; } + read += 1; + let older = |s: Option| { + 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 contiguous = match t.seen { Some(seen) if seen.same_epoch(&end) => match notice.range.to <= seen.publication { true => continue, @@ -197,7 +207,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. @@ -365,7 +378,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 +414,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); From 69fb5e625ea30208834b5e1dda6fa5a155286f58 Mon Sep 17 00:00:00 2001 From: "Claude (dbsp-delivery)" Date: Sat, 10 Oct 2026 20:08:34 -0700 Subject: [PATCH 2/4] fix(protocol): a notice from another epoch only marks the shape stale, and a pulled notice stays with its shape (improvise-z2vr) Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze --- context/code/delivery.md | 23 ++-- context/test-evidence/delivery.md | 32 +++++ .../src/ui/delivery_tests.rs | 113 +++++++++++++++++- .../improvise-protocol/src/delivery/client.rs | 19 +-- .../improvise-protocol/src/delivery/notice.rs | 6 + .../src/delivery/review_tests.rs | 106 ++++++++++++++++ 6 files changed, 281 insertions(+), 18 deletions(-) diff --git a/context/code/delivery.md b/context/code/delivery.md index 45f29e67..8f8e10c8 100644 --- a/context/code/delivery.md +++ b/context/code/delivery.md @@ -127,15 +127,22 @@ 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. -**A notice from an older generation of the host the client is on, or from an epoch it has left, is -ignored.** Reading one rewinds `seen` into the old epoch and queues a correction entry, and the -correction that answers it is then covered by the watermark and a duplicate, so a `restructured` mark -set from the same notice is never cleared and `sync` returns `Exhausted`. A notice the client asked for -(`poll`) skips the left-epoch test, since the host that answers is live (see the answer rule above); it -still skips an older generation. When every shape ignores a notice, its events are dropped too. +**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. 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. A notice carries only that shape's keys, -and applying it to another tracked shape would advance that shape past keys it never saw. +`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 diff --git a/context/test-evidence/delivery.md b/context/test-evidence/delivery.md index 24a80264..c8e7cf3d 100644 --- a/context/test-evidence/delivery.md +++ b/context/test-evidence/delivery.md @@ -206,3 +206,35 @@ property's own weather and a Flaky edge. Rival: the notice-ignore test in `accep and the random property fail. 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_notice_from_another_epoch_marks_the_shape_stale_and_moves_nothing_else + +The new rule itself: a foreign notice sets `stale`; the home host's next notice is still contiguous. + +### 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. diff --git a/crates/improvise-command/src/ui/delivery_tests.rs b/crates/improvise-command/src/ui/delivery_tests.rs index 0e528ef0..981a08f4 100644 --- a/crates/improvise-command/src/ui/delivery_tests.rs +++ b/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 { + drawn_view(host, "Default", window) +} + +fn drawn_view(host: &Host, view: &str, window: &Window) -> ZSet { 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 = due.into_iter().map(|(_, n)| n).collect(); + let mut notices: Vec = due.into_iter().map(|(_, n)| n).collect(); + if step % 2 == 1 { + notices.reverse(); + } deliver_all(client, ¬ices); } @@ -1474,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 { + prop_oneof![ + 4 => (any::(), 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 { + (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 = Vec::new(); + let mut old: Vec = 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 = 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))); + } + } + } } diff --git a/crates/improvise-protocol/src/delivery/client.rs b/crates/improvise-protocol/src/delivery/client.rs index 83f3305c..1045e21d 100644 --- a/crates/improvise-protocol/src/delivery/client.rs +++ b/crates/improvise-protocol/src/delivery/client.rs @@ -167,6 +167,7 @@ impl EdgeClient { /// `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>, answer: bool) -> Vec { + 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 { @@ -182,15 +183,15 @@ impl EdgeClient { ignored += 1; 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, + 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; } @@ -299,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)) { diff --git a/crates/improvise-protocol/src/delivery/notice.rs b/crates/improvise-protocol/src/delivery/notice.rs index 476c27fb..2530f960 100644 --- a/crates/improvise-protocol/src/delivery/notice.rs +++ b/crates/improvise-protocol/src/delivery/notice.rs @@ -157,6 +157,9 @@ pub struct Notice { pub flags: Flags, pub evicted: BTreeSet, pub moved: BTreeMap, + /// Set when the notice was cut for one shape (a pull): it must reach no other. + #[serde(default)] + pub cut_for: Option, } impl Notice { @@ -168,6 +171,7 @@ impl Notice { flags: Flags::default(), evicted: BTreeSet::new(), moved: BTreeMap::new(), + cut_for: None, } } @@ -183,6 +187,7 @@ impl Notice { false => Err(Box::new((self, next))), true => { self.range.to = next.range.to; + self.cut_for = self.cut_for.take().or(next.cut_for); self.axis.extend(next.axis); self.relabels.extend(next.relabels); self.flags = self.flags.union(next.flags); @@ -207,6 +212,7 @@ impl Notice { .get(shape) .map(|m| BTreeMap::from([(shape.clone(), m.clone())])) .unwrap_or_default(), + cut_for: Some(shape.clone()), ..self.clone() } } diff --git a/crates/improvise-protocol/src/delivery/review_tests.rs b/crates/improvise-protocol/src/delivery/review_tests.rs index 04ac8a88..1d32c299 100644 --- a/crates/improvise-protocol/src/delivery/review_tests.rs +++ b/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_notice_from_another_epoch_marks_the_shape_stale_and_moves_nothing_else() { + 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 { .. }))); +} From 3378d3d48acda44b0cf8c51e330679a407d00b34 Mon Sep 17 00:00:00 2001 From: "Claude (dbsp-delivery)" Date: Sat, 10 Oct 2026 20:25:23 -0700 Subject: [PATCH 3/4] fix(protocol): refuse merging notices cut for different shapes; correct the evidence (improvise-z2vr) Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze --- context/code/delivery.md | 5 ++++- context/test-evidence/delivery.md | 17 +++++++++++++---- .../improvise-protocol/src/delivery/notice.rs | 13 +++++++++++-- .../src/delivery/review_tests.rs | 2 +- 4 files changed, 29 insertions(+), 8 deletions(-) diff --git a/context/code/delivery.md b/context/code/delivery.md index 8f8e10c8..a9330ba4 100644 --- a/context/code/delivery.md +++ b/context/code/delivery.md @@ -136,7 +136,10 @@ unordered, so a notice from an epoch other than `seen`'s cannot be placed. Readi 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. A notice the client asked for +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 diff --git a/context/test-evidence/delivery.md b/context/test-evidence/delivery.md index c8e7cf3d..13733493 100644 --- a/context/test-evidence/delivery.md +++ b/context/test-evidence/delivery.md @@ -202,8 +202,11 @@ swap's notice delivered three steps late, an old reply replayed). The client had 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: the notice-ignore test in `accept` disabled; this test -and the random property fail. The seed file is not committed (`proptest-regressions/` is ignored). +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. @@ -221,9 +224,15 @@ The same rewind, then an unsolicited covered correction: the duplicate branch se below a queued entry, and each fetch for that `through` came back a duplicate. Failed on the head before the change. -### a_notice_from_another_epoch_marks_the_shape_stale_and_moves_nothing_else +### a_foreign_notice_marks_the_shape_stale_and_the_home_hosts_answer_settles_it -The new rule itself: a foreign notice sets `stale`; the home host's next notice is still contiguous. +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 diff --git a/crates/improvise-protocol/src/delivery/notice.rs b/crates/improvise-protocol/src/delivery/notice.rs index 2530f960..d56e824e 100644 --- a/crates/improvise-protocol/src/delivery/notice.rs +++ b/crates/improvise-protocol/src/delivery/notice.rs @@ -182,12 +182,11 @@ 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> { - 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 => { self.range.to = next.range.to; - self.cut_for = self.cut_for.take().or(next.cut_for); self.axis.extend(next.axis); self.relabels.extend(next.relabels); self.flags = self.flags.union(next.flags); @@ -275,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(); diff --git a/crates/improvise-protocol/src/delivery/review_tests.rs b/crates/improvise-protocol/src/delivery/review_tests.rs index 1d32c299..e53dbe6a 100644 --- a/crates/improvise-protocol/src/delivery/review_tests.rs +++ b/crates/improvise-protocol/src/delivery/review_tests.rs @@ -275,7 +275,7 @@ fn route_b_a_covered_reply_cannot_set_seen_below_a_queued_entry() { } #[test] -fn a_notice_from_another_epoch_marks_the_shape_stale_and_moves_nothing_else() { +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!( From 6d7cd481f977ff6b6fc0f2446f7cd460b754597e Mon Sep 17 00:00:00 2001 From: "Claude (dbsp-threads)" Date: Sat, 10 Oct 2026 21:59:31 -0700 Subject: [PATCH 4/4] 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); }