2 files changed, 75 insertions, 5 deletions

diff --git a/src/simtest/bench.rs b/src/simtest/bench.rs
index c3996a1..d797115 100644
--- a/src/simtest/bench.rs
+++ b/src/simtest/bench.rs
@@ -424,7 +424,7 @@ impl Totals {
 }
 
 /// The kinds, in the order of the report.
-const KINDS: [&str; 10] = [
+const KINDS: [&str; 11] = [
     "SYNC",
     "STATUS",
     "STATUS_REPLY",
@@ -435,9 +435,10 @@ const KINDS: [&str; 10] = [
     "WHO/ME",
     "LINKS",
     "PROBE",
+    "PRUNE/GRAFT",
 ];
 
-fn counters(by: &ByKind) -> [Counter; 10] {
+fn counters(by: &ByKind) -> [Counter; 11] {
     [
         by.sync,
         by.status,
@@ -449,10 +450,11 @@ fn counters(by: &ByKind) -> [Counter; 10] {
         by.identity,
         by.links,
         by.probe,
+        by.plumtree,
     ]
 }
 
-fn counters_mut(by: &mut ByKind) -> [&mut Counter; 10] {
+fn counters_mut(by: &mut ByKind) -> [&mut Counter; 11] {
     [
         &mut by.sync,
         &mut by.status,
@@ -464,6 +466,7 @@ fn counters_mut(by: &mut ByKind) -> [&mut Counter; 10] {
         &mut by.identity,
         &mut by.links,
         &mut by.probe,
+        &mut by.plumtree,
     ]
 }
 
@@ -545,10 +548,11 @@ async fn run(seed: u64, scenario: &'static Scenario) -> String {
 
     let _ = writeln!(
         out,
-        "\n{} ({} writes at {} a second each)",
+        "\n{} ({} writes at {} a second each; {})",
         scenario.name,
         writes(),
-        1000 / WRITE_EVERY.as_millis()
+        1000 / WRITE_EVERY.as_millis(),
+        all[0].dissemination.name()
     );
     let mut joined = initial;
     for (w, (workload, writer)) in [
@@ -611,6 +615,7 @@ async fn run(seed: u64, scenario: &'static Scenario) -> String {
         );
         trees(&mut out, &nodes, &sim.net).await;
         sent_kept(&mut out, &nodes).await;
+        eager_links(&mut out, &nodes);
         if let Some(writer) = writer {
             slowest(&mut out, &nodes, &nodes[writer], &sim.net).await;
         }
@@ -727,6 +732,66 @@ pub(super) async fn sent_kept(out: &mut String, nodes: &[AppState]) {
     );
 }
 
+/// PlumTree's links after the workload (`swarm::plumtree`): how many neighbours each
+/// node pushes to at once (eager), how many push to it, the nodes none pushes to (they
+/// get every change by announcement and pull), and the links eager at one end only (a
+/// PRUNE or GRAFT lost, or crossing). Nothing for a way that keeps no such links.
+fn eager_links(out: &mut String, nodes: &[AppState]) {
+    let mut pushes: HashMap<String, Vec<String>> = HashMap::new();
+    for node in nodes {
+        let Some(lazy) = node.dissemination.lazy() else {
+            return;
+        };
+        let eager = crate::swarm::topology::our_neighbours(node)
+            .into_iter()
+            .filter(|peer| !lazy.contains(peer))
+            .collect();
+        pushes.insert(node.local_id().to_owned(), eager);
+    }
+    let name = |id: &str| {
+        nodes
+            .iter()
+            .find(|node| node.local_id() == id)
+            .and_then(|node| node.alias.clone())
+            .unwrap_or_else(|| id.to_owned())
+    };
+    let mut pushed_by: HashMap<&str, usize> = HashMap::new();
+    let mut one_ended = 0;
+    for (node, eager) in &pushes {
+        for peer in eager {
+            *pushed_by.entry(peer.as_str()).or_default() += 1;
+            if !pushes.get(peer).is_some_and(|theirs| theirs.contains(node)) {
+                one_ended += 1;
+            }
+        }
+    }
+    let count = pushes.len().max(1) as f64;
+    let out_sum: usize = pushes.values().map(Vec::len).sum();
+    let in_least = pushes
+        .keys()
+        .map(|node| pushed_by.get(node.as_str()).copied().unwrap_or(0))
+        .min()
+        .unwrap_or(0);
+    let mut unpushed: Vec<String> = pushes
+        .keys()
+        .filter(|node| !pushed_by.contains_key(node.as_str()))
+        .map(|node| name(node))
+        .collect();
+    unpushed.sort();
+    let _ = writeln!(
+        out,
+        "    PlumTree's links at the end: each node pushes to {:.1} neighbours on average; \
+         pushed to by {in_least} at least; {} pushed to by none{}; {one_ended} links eager at one end only",
+        out_sum as f64 / count,
+        unpushed.len(),
+        if unpushed.is_empty() {
+            String::new()
+        } else {
+            format!(" ({})", unpushed.join(", "))
+        },
+    );
+}
+
 /// Whether the nodes' views of the writers' trees agree: each member's parent as the
 /// member chose it (`swarm::tree::own_parent`) against what that parent and the
 /// others plan (`swarm::tree::plan`: their children, the neighbours they count as
diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs
index e9d8c15..c3b2368 100644
--- a/src/simtest/mod.rs
+++ b/src/simtest/mod.rs
@@ -193,6 +193,11 @@ impl Sim {
             clock_skew_warn: Duration::from_secs(2),
             stream_heartbeat: Duration::from_secs(30),
             superseded_writes: Default::default(),
+            // `SIM_DISSEMINATION=plumtree`: every simulated node passes changes on so.
+            dissemination: match std::env::var("SIM_DISSEMINATION").as_deref() {
+                Ok("plumtree") => crate::swarm::dissemination::Kind::Plumtree,
+                _ => Default::default(),
+            },
             blob_chunk_bytes: 64 * 1024,
             swarm_key: None,
             protocol_writers: HashSet::new(),