Merge pull request #359 from fiddlerwoaroof/fix/delivery-left-epoch-livelock

fix(delivery): the live host's answer is never refused as a left epoch (improvise-z2vr)

context/code/delivery.md

@@ -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
 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,
 and applying it to another tracked shape would advance that shape past keys it never saw.
 
@@ -182,7 +190,9 @@ counts comparisons on a counting item type rather than reading a clock.
 **The host instance** is a random id, never a time (`Host::new` takes one from `RandomState`).
 `RandomState` has no entropy on `wasm32-unknown-unknown`, so a worker must pass a real one through
 `Host::with_instance`. A restarted host with the same id would accept a stale watermark as current.
-`Worker` still calls `Host::new` (improvise-bw4p.7).
+`Worker` still calls `Host::new` (improvise-bw4p.7). Such a restart, with the same instance id and
+generation 0, is not recognised as a restart: the client's `older` guard refuses its replies as coming from
+an older generation of the host it is on (improvise-bw4p.7).
 
 **Eviction** (`evict_idle`) takes a local count and an idle length in the same unit, never a wall clock
 compared across machines. An evicted key is announced in the next notice and forgotten; naming it again
@@ -207,6 +217,13 @@ The token in a request is carried and echoed by the client and never read by the
 
 ## 8. Edge adapters
 
+**Transport contract.** A reply must reach the call that made the request. `Request` and `Correction`
+carry no id, so the client cannot tell a late reply from an answer on its own; `sync` and `poll` treat
+whatever `Edge::call` returns as the answer (`apply_answer`), and only a reply handed to `apply` is
+treated as late. A transport that reorders or duplicates replies must therefore correlate them itself
+or drop the strays (improvise-bw4p.10 adds the id to the frames).
+
+
 `Edge` is one end of a stream: `call(Request) -> Result<Reply, EdgeError>`. `InProcess` is the
 function call; `WireEdge` passes request and reply through JSON text, the adapter at a wire boundary,
 so the lossless properties are exercised by every test that uses it. Neither adds a `ClientFrame` or

context/test-evidence/delivery.md

@@ -172,3 +172,23 @@ 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
 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.
+
+## 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) fails this test. **The random
+property does not catch it**: it passed 11 of 11 runs under that fault, and again after the late reply was
+made an explicit event (a stashed reply from a host the client never joined, replayed before a lost
+connection), because the property rarely produces a reply from a host the client never joined. This
+test is the pin. 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.

crates/improvise-command/src/ui/delivery_tests.rs

@@ -1086,6 +1086,45 @@ fn a_reply_held_across_a_host_restart_cannot_pull_the_client_back() {
     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 ─────────────────────────────────────────────
 
 mod random {
@@ -1265,6 +1304,39 @@ mod random {
         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! {
         #[test]
         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()) {
                     match &old[i % old.len()] {
-                        Reply::Correction(c) => {
-                            client.apply(c);
-                        }
+                        Reply::Correction(c) => replay(&mut client, c, &mut epochs)?,
                         Reply::Notices(list) => deliver_all(&mut client, list),
                     }
                 }
@@ -1316,20 +1386,10 @@ mod random {
                 let mut edge = Flaky::scripted(&mut host, script);
                 client.sync(&mut edge, &shape(), 6);
                 old.append(&mut edge.replies);
+                note_epoch(&client, &mut epochs);
                 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))) {
-                    client.apply(c);
-                }
-                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);
-                    }
+                    replay(&mut client, c, &mut epochs)?;
                 }
             }
 

crates/improvise-protocol/src/delivery/client.rs

@@ -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>) {
+        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 {
             return (Outcome::Untracked, Vec::new());
         };
+        if answer {
+            t.left
+                .retain(|e| *e != (c.through.host, c.through.generation));
+        }
         let target = decode(&c.target);
         let older = |s: Option<Stamp>| {
             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);
             };
             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)) => {
                     list.iter()
                         .for_each(|n| events.extend(self.accept(n, Some(shape))));
@@ -387,7 +402,7 @@ impl EdgeClient {
             Ok(Reply::Notices(list)) => list
                 .iter()
                 .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) => {
                 self.request_failed(shape);
                 return (Synced::Failed, events);