introduce eager links

1 files changed, 62 insertions, 1 deletions

diff --git a/src/simtest/relay.rs b/src/simtest/relay.rs
index dadeed3..88ece74 100644
--- a/src/simtest/relay.rs
+++ b/src/simtest/relay.rs
@@ -1,5 +1,6 @@
 //! Changes passed on in a full mesh (`swarm::relay`) to the members the writer can't
-//! reach: those on none of its transports.
+//! reach: those on none of its transports, or behind a link that fails; and the links
+//! a pull makes eager.
 
 use std::time::Duration;
 
@@ -117,3 +118,63 @@ async fn changes_reach_members_the_writer_fails_to_reach_at_once() {
         .await;
     }
 }
+
+/// Whether `node` asked `peer` for its changes eagerly, and whether `peer` heard so.
+async fn eager(node: &crate::state::AppState, peer: &crate::state::AppState) -> bool {
+    let asked = match node.get_peer_state(peer.local_id()).await {
+        Some(state) => state.lock().await.eager_from,
+        None => false,
+    };
+    let heard = match peer.get_peer_state(node.local_id()).await {
+        Some(state) => state.lock().await.eager_to,
+        None => false,
+    };
+    asked && heard
+}
+
+/// As above, x reaching b alone: c pulls x's first write from b, which makes the link
+/// from b to c eager; x's next write b forwards to c as it applies it, without
+/// announcing it. The link from x's side stays lazy: nothing was pulled over it.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn pulls_make_links_eager() {
+    let keepalive = Duration::from_secs(5);
+    let slow = move |config: &mut crate::node::NodeConfig| {
+        config.timing = Timing::new(keepalive, SYNC_REPLY_TIMEOUT, DELIVERY_LIMIT);
+    };
+    let sim = Sim::with_identity(65);
+    let cut = crate::transport::sim::LinkParams {
+        loss: 1.0,
+        ..Default::default()
+    };
+    sim.net.set_link("x", "c", cut.clone());
+    sim.net.set_link("c", "x", cut);
+    let b = sim.start_on_with("b", &["sim"], &[], Start::Genesis, slow);
+    let c = sim.start_on_with("c", &["sim"], &["b"], Start::Join, slow);
+    let x = sim.start_on_with("x", &["sim"], &["b"], Start::Join, slow);
+    sim.wait_ready(&[&b, &c, &x], SOON).await;
+    // c may have pulled already: what x writes when it registers comes that way too.
+    sim.wait_converged(&[&b, &c, &x], SOON).await;
+
+    write(&x, "y", json!(1)).await;
+    {
+        let (b, c) = (&b, &c);
+        sim.wait_until(
+            "c pulled x's write from b, and the link from b is eager",
+            SOON,
+            move || async move { read(c, "y").await == Some(json!(1)) && eager(c, b).await },
+        )
+        .await;
+    }
+    assert!(!eager(&b, &x).await, "nothing was pulled from x");
+
+    write(&x, "y", json!(2)).await;
+    {
+        let c = &c;
+        sim.wait_until(
+            "c has x's next write within a second",
+            Duration::from_secs(1),
+            move || async move { read(c, "y").await == Some(json!(2)) },
+        )
+        .await;
+    }
+}