fix(protocol): the live host's answer is never refused as a left epoch (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)
2026-10-10 18:22:14 -07:00
committed by Edward Langley
parent cae6f26ee8
commit bf02a4b589
4 changed files with 118 additions and 18 deletions
+8
View File
@@ -119,6 +119,14 @@ dead host's replies are refused. A duplicate answer also prunes every queued ent
and moves `seen` onto the live epoch, so a late notice from a dead host costs one extra round and cannot and moves `seen` onto the live epoch, so a late notice from a dead host costs one extra round and cannot
livelock `sync`. livelock `sync`.
**The host that answers a request is live, whatever the client believed.** `left` is a guess, because
epochs of different hosts are unordered: a late reply from a host the client never joined moves it onto
that host, and writes off the live one. Refusing the live host's answers as "left" then strands the client
(`sync` returned `Exhausted`, found by the property test). So `apply_answer`, which `sync` and `poll` use
for the reply to a request they just sent, takes an epoch off `left` before it applies. `apply` is for 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, `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. and applying it to another tracked shape would advance that shape past keys it never saw.
+17
View File
@@ -172,3 +172,20 @@ host, and the old reply then arrives. Rival: the `left` check removed.
every restart and swap. Rival: the `left` check removed fails it. The older-generation guard on its own is every restart and swap. Rival: the `left` check removed fails it. The older-generation guard on its own is
not separable from `left` in the histories the property reaches, so it is held by not separable from `left` in the histories the property reaches, so it is held by
`a_late_window_reply_from_an_older_generation_does_not_regress` alone. `a_late_window_reply_from_an_older_generation_does_not_regress` alone.
## The left-epoch livelock
### a_late_reply_from_a_host_the_client_never_joined_cannot_strand_it
The property test failed with `Exhausted` on a shrunk history of two restarts and four writes. Cause: the
client was on host 3, never having joined host 2; a late host-2 reply was applied (an unknown epoch is
accepted), which wrote host 3 into `left`, and every later answer from the live host 3 was then refused as
a duplicate. The test replays that history by hand: three hosts, the host-2 reply lost in flight, the client
settling on host 3, the late reply applied, the connection lost, and a sync that must settle on host 3's
grid. Rival: `apply_answer` treated like `apply` (the `left.retain` removed); this test and the random
property fail. Rival: the left-epoch guard removed fails `a_reply_held_across_a_host_restart_cannot_pull_the_client_back`
and the property, as before.
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.
@@ -1086,6 +1086,45 @@ fn a_reply_held_across_a_host_restart_cannot_pull_the_client_back() {
assert_eq!(state, Synced::Settled); assert_eq!(state, Synced::Settled);
} }
#[test]
fn a_late_reply_from_a_host_the_client_never_joined_cannot_strand_it() {
let window = Window {
top: 0,
left: 0,
rows: 1,
cols: 1,
};
let mut first = host_instance(1);
let mut client = EdgeClient::new(session());
client.track(shape(), window);
client.sync(&mut Flaky::new(&mut first), &shape(), 6);
let mut second = host_instance(2);
let id = connect(&mut second);
send(&mut second, id, vec![set("South", 1.0)]);
let mut lost = Flaky::scripted(&mut second, vec![Some(Fault::DropReply)]);
client.connection_lost();
client.sync(&mut lost, &shape(), 6);
let late = lost.replies.remove(0);
let mut third = host_instance(3);
client.connection_lost();
client.sync(&mut Flaky::new(&mut third), &shape(), 6);
let Reply::Correction(late) = late else {
panic!("the lost reply was a correction")
};
client.apply(&late);
client.connection_lost();
let (state, _) = client.sync(&mut Flaky::new(&mut third), &shape(), 6);
assert_eq!(
state,
Synced::Settled,
"the live host's answers must be taken"
);
assert_eq!(client.held(&shape()), Some(&drawn(&third, &window)));
}
// ── the whole thing, at random ───────────────────────────────────────────── // ── the whole thing, at random ─────────────────────────────────────────────
mod random { mod random {
@@ -1265,6 +1304,39 @@ mod random {
deliver_all(client, &notices); deliver_all(client, &notices);
} }
/// Epochs the client has applied a correction from, oldest first.
fn note_epoch(client: &EdgeClient, epochs: &mut Vec<(u64, u64)>) {
if let Some(w) = client.watermark(&shape()) {
let now = (w.host, w.generation);
if !epochs.contains(&now) {
epochs.push(now);
}
}
}
/// A late reply arrives out of band. One from an epoch the client was in and has left is refused;
/// one from an epoch it never joined may legitimately move it, and the next answer brings it back.
fn replay(
client: &mut EdgeClient,
c: &improvise_protocol::delivery::Correction,
epochs: &mut Vec<(u64, u64)>,
) -> Result<(), TestCaseError> {
let theirs = (c.through.host, c.through.generation);
let current = client.watermark(&shape()).map(|w| (w.host, w.generation));
let left = epochs.contains(&theirs) && current != Some(theirs);
let (outcome, _) = client.apply(c);
if left {
prop_assert_eq!(
outcome,
Outcome::Duplicate,
"a reply from the left epoch {:?}",
theirs
);
}
note_epoch(client, epochs);
Ok(())
}
proptest! { proptest! {
#[test] #[test]
fn a_corrected_state_is_the_target_and_correcting_it_again_is_the_zero_zset( fn a_corrected_state_is_the_target_and_correcting_it_again_is_the_zero_zset(
@@ -1306,9 +1378,7 @@ mod random {
} }
if let Some(i) = weather.replay.filter(|_| !old.is_empty()) { if let Some(i) = weather.replay.filter(|_| !old.is_empty()) {
match &old[i % old.len()] { match &old[i % old.len()] {
Reply::Correction(c) => { Reply::Correction(c) => replay(&mut client, c, &mut epochs)?,
client.apply(c);
}
Reply::Notices(list) => deliver_all(&mut client, list), Reply::Notices(list) => deliver_all(&mut client, list),
} }
} }
@@ -1316,20 +1386,10 @@ mod random {
let mut edge = Flaky::scripted(&mut host, script); let mut edge = Flaky::scripted(&mut host, script);
client.sync(&mut edge, &shape(), 6); client.sync(&mut edge, &shape(), 6);
old.append(&mut edge.replies); old.append(&mut edge.replies);
note_epoch(&client, &mut epochs);
let moved_host = matches!(op, Op::Restart | Op::Swap); let moved_host = matches!(op, Op::Restart | Op::Swap);
if let (true, Some(Reply::Correction(c))) = (moved_host, old.get(step % old.len().max(1))) { if let (true, Some(Reply::Correction(c))) = (moved_host, old.get(step % old.len().max(1))) {
client.apply(c); replay(&mut client, c, &mut epochs)?;
}
if let Some(w) = client.watermark(&shape()) {
let now = (w.host, w.generation);
let seen_at = epochs.iter().position(|e| *e == now);
prop_assert!(
seen_at.is_none_or(|i| i + 1 == epochs.len()),
"the client went back to a superseded epoch {now:?} after {epochs:?}"
);
if seen_at.is_none() {
epochs.push(now);
}
} }
} }
@@ -243,11 +243,26 @@ impl EdgeClient {
} }
} }
/// A correction. It applies only at the watermark it starts from: `held + (target - held)`. /// A correction that arrives on its own: it applies only at the watermark it starts from,
/// as `held + (target - held)`, and never from an epoch the client has left.
pub fn apply(&mut self, c: &Correction) -> (Outcome, Vec<Event>) { pub fn apply(&mut self, c: &Correction) -> (Outcome, Vec<Event>) {
self.apply_with(c, false)
}
/// The reply to a request this client just made. The host that answers is the live one, so
/// an epoch the client had written off is live again (`context/code/delivery.md` section 4).
pub fn apply_answer(&mut self, c: &Correction) -> (Outcome, Vec<Event>) {
self.apply_with(c, true)
}
fn apply_with(&mut self, c: &Correction, answer: bool) -> (Outcome, Vec<Event>) {
let Some(t) = self.shapes.get_mut(&c.shape) else { let Some(t) = self.shapes.get_mut(&c.shape) else {
return (Outcome::Untracked, Vec::new()); return (Outcome::Untracked, Vec::new());
}; };
if answer {
t.left
.retain(|e| *e != (c.through.host, c.through.generation));
}
let target = decode(&c.target); let target = decode(&c.target);
let older = |s: Option<Stamp>| { let older = |s: Option<Stamp>| {
s.is_some_and(|s| s.host == c.through.host && c.through.generation < s.generation) s.is_some_and(|s| s.host == c.through.host && c.through.generation < s.generation)
@@ -347,7 +362,7 @@ impl EdgeClient {
return (Synced::Settled, events); return (Synced::Settled, events);
}; };
match edge.call(request) { match edge.call(request) {
Ok(Reply::Correction(c)) => events.extend(self.apply(&c).1), Ok(Reply::Correction(c)) => events.extend(self.apply_answer(&c).1),
Ok(Reply::Notices(list)) => { Ok(Reply::Notices(list)) => {
list.iter() list.iter()
.for_each(|n| events.extend(self.accept(n, Some(shape)))); .for_each(|n| events.extend(self.accept(n, Some(shape))));
@@ -387,7 +402,7 @@ impl EdgeClient {
Ok(Reply::Notices(list)) => list Ok(Reply::Notices(list)) => list
.iter() .iter()
.for_each(|n| events.extend(self.accept(n, Some(shape)))), .for_each(|n| events.extend(self.accept(n, Some(shape)))),
Ok(Reply::Correction(c)) => events.extend(self.apply(&c).1), Ok(Reply::Correction(c)) => events.extend(self.apply_answer(&c).1),
Err(EdgeError::Failed) => { Err(EdgeError::Failed) => {
self.request_failed(shape); self.request_failed(shape);
return (Synced::Failed, events); return (Synced::Failed, events);