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 { .. }))); +}