diff --git a/context/code/delivery.md b/context/code/delivery.md index 16897206..44359565 100644 --- a/context/code/delivery.md +++ b/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 diff --git a/context/test-evidence/delivery.md b/context/test-evidence/delivery.md index fdb92e52..13733493 100644 --- a/context/test-evidence/delivery.md +++ b/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. diff --git a/crates/improvise-command/src/ui/delivery_tests.rs b/crates/improvise-command/src/ui/delivery_tests.rs index dcad00d6..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); } @@ -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 = 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 { + 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 87a20a77..1045e21d 100644 --- a/crates/improvise-protocol/src/delivery/client.rs +++ b/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 { - 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 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| { + 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); diff --git a/crates/improvise-protocol/src/delivery/notice.rs b/crates/improvise-protocol/src/delivery/notice.rs index 476c27fb..d56e824e 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, } } @@ -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> { - 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(); diff --git a/crates/improvise-protocol/src/delivery/review_tests.rs b/crates/improvise-protocol/src/delivery/review_tests.rs index 04ac8a88..e53dbe6a 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_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 { .. }))); +}