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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze
This commit is contained in:
Claude (dbsp-delivery)
committed by
Edward Langley
Edward Langley
parent
f47efef5ae
commit
69fb5e625e
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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<Cell> {
|
||||
drawn_view(host, "Default", window)
|
||||
}
|
||||
|
||||
fn drawn_view(host: &Host, view: &str, window: &Window) -> ZSet<Cell> {
|
||||
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<Notice> = due.into_iter().map(|(_, n)| n).collect();
|
||||
let mut notices: Vec<Notice> = 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<Value = Op> {
|
||||
prop_oneof![
|
||||
4 => (any::<bool>(), 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<Value = (Window, Window)> {
|
||||
(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<Notice> = Vec::new();
|
||||
let mut old: Vec<Reply> = 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<Notice> = 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)));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<Event> {
|
||||
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)) {
|
||||
|
||||
@@ -157,6 +157,9 @@ pub struct Notice {
|
||||
pub flags: Flags,
|
||||
pub evicted: BTreeSet<ShapeKey>,
|
||||
pub moved: BTreeMap<ShapeKey, Moved>,
|
||||
/// Set when the notice was cut for one shape (a pull): it must reach no other.
|
||||
#[serde(default)]
|
||||
pub cut_for: Option<ShapeKey>,
|
||||
}
|
||||
|
||||
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()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 { .. })));
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user