refactor: remove 'waterlevel'

crates/swactor-gossip/tests/bug_hunt_gossip.rs

@@ -0,0 +1,139 @@
+//! Bug-hunt tests for swactor-gossip.
+//!
+//! Each test asserts correct protocol behaviour and **fails** when
+//! the bug is present.
+
+use swactor::runtime::{Inbox, Runtime, RuntimeConfig};
+use swactor_gossip::{GossipActor, GossipMessage, GossipQueryResponse, GossipState};
+
+// ═════════════════════════════════════════════════════════════════════════
+// Bug 1 — GossipState::merge drops remote value on version ties
+//
+// `merge` uses `remote.version > local.version`.  When two nodes
+// independently `set` the same key, both create version 1.  Neither
+// value dominates the other, so the nodes never converge — permanent
+// split-brain.
+//
+// A correct LWW register needs a tiebreaker (e.g. node-id comparison
+// or a Lamport timestamp) so that one value always wins.
+// ═════════════════════════════════════════════════════════════════════════
+
+/// Direct GossipState unit test — no runtime needed.
+#[test]
+fn merge_equal_version_causes_permanent_divergence() {
+    let mut state_a = GossipState::new();
+    let mut state_b = GossipState::new();
+
+    // Both independently set "x" → both get version 1.
+    state_a.set("x".into(), b"alpha".to_vec());
+    state_b.set("x".into(), b"beta".to_vec());
+
+    assert_eq!(state_a.get("x").unwrap().version, 1);
+    assert_eq!(state_b.get("x").unwrap().version, 1);
+
+    // Simulate a full gossip exchange (both directions).
+    state_a.merge(&state_b);
+    state_b.merge(&state_a);
+
+    // Extra rounds — shouldn't matter, but proves the deadlock is permanent.
+    for _ in 0..10 {
+        state_a.merge(&state_b);
+        state_b.merge(&state_a);
+    }
+
+    let val_a = &state_a.get("x").unwrap().value;
+    let val_b = &state_b.get("x").unwrap().value;
+
+    assert_eq!(
+        val_a, val_b,
+        "BUG: permanent divergence — A holds {:?}, B holds {:?}, \
+         both at version 1.  merge() rejects the remote value when \
+         versions are equal, so neither node ever adopts the other's value.",
+        String::from_utf8_lossy(val_a),
+        String::from_utf8_lossy(val_b),
+    );
+}
+
+/// Full actor-level test: two gossip nodes with concurrent writes.
+#[test]
+fn concurrent_writes_diverge_through_gossip_actors() {
+    let rt = Runtime::new(RuntimeConfig::default());
+
+    let node_a = rt.spawn(GossipActor::new()).unwrap();
+    let node_b = rt.spawn(GossipActor::new()).unwrap();
+
+    // Wire bidirectional peering.
+    rt.send_to(node_a, GossipMessage::AddPeer(node_b)).unwrap();
+    rt.send_to(node_b, GossipMessage::AddPeer(node_a)).unwrap();
+    for _ in 0..5 {
+        rt.tick();
+    }
+
+    // Both set the same key to different values → version 1 on each.
+    rt.send_to(
+        node_a,
+        GossipMessage::Set {
+            key: "x".into(),
+            value: b"alpha".to_vec(),
+        },
+    )
+    .unwrap();
+    rt.send_to(
+        node_b,
+        GossipMessage::Set {
+            key: "x".into(),
+            value: b"beta".to_vec(),
+        },
+    )
+    .unwrap();
+    for _ in 0..5 {
+        rt.tick();
+    }
+
+    // Run many gossip rounds.
+    for _ in 0..50 {
+        rt.send_to(node_a, GossipMessage::DoGossipRound).unwrap();
+        rt.send_to(node_b, GossipMessage::DoGossipRound).unwrap();
+        for _ in 0..10 {
+            rt.tick();
+        }
+    }
+
+    // Query both nodes.
+    let inbox_a: Inbox<GossipQueryResponse> = rt.new_inbox().unwrap();
+    let inbox_b: Inbox<GossipQueryResponse> = rt.new_inbox().unwrap();
+
+    rt.send_to(
+        node_a,
+        GossipMessage::Query {
+            key: "x".into(),
+            reply_to: *inbox_a.addr(),
+        },
+    )
+    .unwrap();
+    rt.send_to(
+        node_b,
+        GossipMessage::Query {
+            key: "x".into(),
+            reply_to: *inbox_b.addr(),
+        },
+    )
+    .unwrap();
+    for _ in 0..10 {
+        rt.tick();
+    }
+
+    let resp_a = inbox_a.try_recv().expect("response from A");
+    let resp_b = inbox_b.try_recv().expect("response from B");
+
+    assert_eq!(
+        resp_a.value, resp_b.value,
+        "BUG: permanent split-brain — node A has {:?}, node B has {:?} \
+         (versions {:?} vs {:?}).  merge() rejects the remote value on \
+         equal versions, so the two nodes can never agree.",
+        resp_a.value.as_ref().map(|v| String::from_utf8_lossy(v).to_string()),
+        resp_b.value.as_ref().map(|v| String::from_utf8_lossy(v).to_string()),
+        resp_a.version,
+        resp_b.version,
+    );
+}