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);