path: root/src/simtest/bench.rs

//! What passing changes on costs (`swarm::relay`): whole nodes on the simulated network,
//! a series of writes, and the messages, bytes and delays it took, per scenario. A
//! benchmark, not a test: `cargo benchmarks` (release, with the other `bench_*`) prints
//! a table per scenario to compare before and after a change; nothing is asserted but
//! that the nodes converge.
//!
//! Each scenario starts its swarm and waits until it has settled (every member known,
//! the same heads, then sync messages no fewer in one half of the quiet phase than in
//! the one before), those two halves being the quiet phase (no writes), then each
//! workload: 100 writes at 5 a second, and until every node has
//! them. What a workload sent is shown as it is, beside what the quiet phase sent in as
//! long: not taken off, since writes make keepalives unneeded (a pair that exchanges
//! anything needs none), so the quiet phase's count isn't part of a workload's.

use std::collections::HashMap;
use std::fmt::Write;
use std::time::Duration;

use serde_json::json;
use tokio::time::Instant;

use super::{SYNC_REPLY_TIMEOUT, Sim, write};
use crate::state::{AppState, Incidents, Timing};
use crate::swarm::bootstrap::Start;
use crate::swarm::members;
use crate::swarm::stats::{Propagation, Relay};
use crate::transport::sim::{DELIVERY_LIMIT, LinkParams};
use crate::transport::traffic::{ByKind, Counter};

/// Writes per workload, unless `BENCH_WRITES` says otherwise, and how many a second.
const WRITES: usize = 100;
const WRITE_EVERY: Duration = Duration::from_millis(200);
/// Longer than the tests' (`simtest::KEEPALIVE`), so that keepalives don't drown what
/// the writes cost; as long as the relay tests'.
const KEEPALIVE: Duration = Duration::from_secs(5);
/// The quiet phase: long enough to see the keepalives' rate (a STATUS every third, 15 s).
const QUIET: Duration = Duration::from_secs(20);
/// Writes per workload: `WRITES`, or `BENCH_WRITES` (longer, to see what grows with them).
fn writes() -> usize {
    std::env::var("BENCH_WRITES")
        .ok()
        .and_then(|value| value.parse().ok())
        .filter(|writes| *writes > 0)
        .unwrap_or(WRITES)
}

/// Only the scenarios whose names contain this, if set: `BENCH_SCENARIO=NAT`; or any of
/// several, comma-separated, run at once (as `BENCH_PARALLEL`): `BENCH_SCENARIO=Tor,Mycelium`.
const ONLY: &str = "BENCH_SCENARIO";
/// If set, all scenarios at once (`BENCH_PARALLEL=1`): several times faster on many
/// cores, but they share them, so delays (and what depends on timing) may come out
/// higher than alone; message counts barely.
const PARALLEL: &str = "BENCH_PARALLEL";
const SETTLE: Duration = Duration::from_secs(180);

/// How the swarm is made.
struct Scenario {
    name: &'static str,
    nodes: usize,
    /// The transports of node `i` (`sim`, `sim2`).
    on: fn(usize) -> &'static [&'static str],
    /// Before the nodes start: links cut, delays.
    net: fn(&Sim, &[String]),
    /// The node that writes in the one-writer workload.
    writer: usize,
    /// A node that goes away for the middle half of each workload (it is taken off the
    /// net, as if gone without a word) and comes back, if any: 50 writes at 5 a second,
    /// 10 s, not long enough to be held dormant (four keepalives, 20 s), so what a node
    /// that is gone costs before the others notice.
    away: Option<usize>,
    /// Nodes that join during each workload, one at a time, evenly spread (of
    /// `nodes`: they start with `nodes` less twice as many), and so what members joining
    /// cost while writes go on: member lists differ between nodes until the newcomer's
    /// registration reached them, and their trees with them. Their catching up counts
    /// in the messages, but not in the delays (`swarm::stats`: not until caught up).
    joining: usize,
    /// The first wait for a sync answer, if not `SYNC_REPLY_TIMEOUT`: longer where round
    /// trips take seconds, as Tor's transport asks for (`Transport::sync_reply_timeout`).
    reply_timeout: Option<Duration>,
}

fn sim_only(_: usize) -> &'static [&'static str] {
    &["sim"]
}

fn clean(_: &Sim, _: &[String]) {}

/// Node 1 reaches node 2 alone, as one behind NAT with a single link.
fn writer_behind_nat(sim: &Sim, ids: &[String]) {
    let cut = LinkParams {
        loss: 1.0,
        ..LinkParams::default()
    };
    for (i, id) in ids.iter().enumerate() {
        if i != 1 && i != 2 {
            sim.net.set_link(&ids[1], id, cut.clone());
            sim.net.set_link(id, &ids[1], cut.clone());
        }
    }
}

/// Nodes 0 to 2 on both nets (bridges), the first half of the rest on `sim`, the
/// other half on `sim2`.
fn two_transports(i: usize) -> &'static [&'static str] {
    match i {
        0..3 => &["sim", "sim2"],
        3..16 => &["sim"],
        _ => &["sim2"],
    }
}

/// As the soak test's: 2% lost, 1% twice, 50 to 500 ms.
fn lossy(sim: &Sim, _: &[String]) {
    sim.net.set_default_link(LinkParams {
        loss: 0.02,
        duplicate: 0.01,
        delay: (Duration::from_millis(50), Duration::from_millis(500)),
        ..LinkParams::default()
    });
}

/// As `lossy`, but each link with a latency of its own, as over Mycelium: round trips
/// from 20 to 250 ms, each message within a fifth of its link's. So what the uniform
/// 50–500 ms of `lossy` gives that real links don't: messages on one link overtake each
/// other only within the jitter, and the fast links to a node are fast every time.
pub(super) fn mycelium_like(sim: &Sim, ids: &[String]) {
    let mut rng = crate::testrand::Rng::new(7);
    for (i, from) in ids.iter().enumerate() {
        for to in &ids[i + 1..] {
            let rtt_ms = 20.0 + rng.next_f64() * 230.0;
            let half = Duration::from_secs_f64(rtt_ms / 2.0 / 1000.0);
            let link = LinkParams {
                loss: 0.02,
                duplicate: 0.01,
                delay: (half * 4 / 5, half * 6 / 5),
                ..LinkParams::default()
            };
            sim.net.set_link(from, to, link.clone());
            sim.net.set_link(to, from, link);
        }
    }
}

/// Each link a round trip of its own, as over Tor: 250 to 400 ms, each message within a
/// fifth of its link's; one link in ten through a slow circuit, 1 to 3 s. Nothing lost
/// (Tor carries streams reliably). The simulated net can't make one link slow for a
/// while only, so the circuits that are slow at times are slow throughout here. With a
/// first wait for answers of 5 s, as Tor's transport waits longer (30 s in production).
/// In order, as a circuit carries a stream: a message never overtakes an earlier one.
fn tor_like(sim: &Sim, ids: &[String]) {
    let mut rng = crate::testrand::Rng::new(9);
    for (i, from) in ids.iter().enumerate() {
        for to in &ids[i + 1..] {
            let rtt_ms = if rng.chance(0.1) {
                1000.0 + rng.next_f64() * 2000.0
            } else {
                250.0 + rng.next_f64() * 150.0
            };
            let half = Duration::from_secs_f64(rtt_ms / 2.0 / 1000.0);
            let link = LinkParams {
                delay: (half * 4 / 5, half * 6 / 5),
                in_order: true,
                ..LinkParams::default()
            };
            sim.net.set_link(from, to, link.clone());
            sim.net.set_link(to, from, link);
        }
    }
}

const SCENARIOS: &[Scenario] = &[
    Scenario {
        name: "full mesh, 13 nodes",
        nodes: 13,
        on: sim_only,
        net: clean,
        writer: 1,
        away: None,
        joining: 0,
        reply_timeout: None,
    },
    Scenario {
        name: "writer behind NAT (one link), 13 nodes",
        nodes: 13,
        on: sim_only,
        net: writer_behind_nat,
        writer: 1,
        away: None,
        joining: 0,
        reply_timeout: None,
    },
    Scenario {
        name: "two transports, 3 bridges, 30 nodes",
        nodes: 30,
        on: two_transports,
        net: clean,
        writer: 5,
        away: None,
        joining: 0,
        reply_timeout: None,
    },
    Scenario {
        name: "partial mesh, 50 nodes",
        nodes: 50,
        on: sim_only,
        net: clean,
        writer: 1,
        away: None,
        joining: 0,
        reply_timeout: None,
    },
    Scenario {
        name: "partial mesh, lossy and slow links, 50 nodes",
        nodes: 50,
        on: sim_only,
        net: lossy,
        writer: 1,
        away: None,
        joining: 0,
        reply_timeout: None,
    },
    Scenario {
        name: "partial mesh, 100 nodes",
        nodes: 100,
        on: sim_only,
        net: clean,
        writer: 1,
        away: None,
        joining: 0,
        reply_timeout: None,
    },
    Scenario {
        name: "partial mesh, a member away for 10 s, 50 nodes",
        nodes: 50,
        on: sim_only,
        net: clean,
        writer: 1,
        away: Some(3),
        joining: 0,
        reply_timeout: None,
    },
    Scenario {
        name: "partial mesh, 5 members joining during each workload, 50 nodes",
        nodes: 50,
        on: sim_only,
        net: clean,
        writer: 1,
        away: None,
        joining: 5,
        reply_timeout: None,
    },
    // Last, so the others keep their seeds.
    Scenario {
        name: "partial mesh, lossy links with Mycelium's round trips, 50 nodes",
        nodes: 50,
        on: sim_only,
        net: mycelium_like,
        writer: 1,
        away: None,
        joining: 0,
        reply_timeout: None,
    },
    Scenario {
        name: "partial mesh, Tor's round trips, 50 nodes",
        nodes: 50,
        on: sim_only,
        net: tor_like,
        writer: 1,
        away: None,
        joining: 0,
        reply_timeout: Some(Duration::from_secs(5)),
    },
    Scenario {
        name: "writer behind NAT (one link), partial mesh, 50 nodes",
        nodes: 50,
        on: sim_only,
        net: writer_behind_nat,
        writer: 1,
        away: None,
        joining: 0,
        reply_timeout: None,
    },
];

#[tokio::test(flavor = "multi_thread", worker_threads = 16)]
#[ignore = "a benchmark: cargo benchmarks"]
async fn bench_propagation() {
    // Past the test harness's `test … ...`, so each report is on a line of its own.
    println!();
    let only = std::env::var(ONLY).unwrap_or_default();
    let parts: Vec<&str> = only.split(',').map(str::trim).collect();
    let chosen: Vec<(u64, &'static Scenario)> = (70..)
        .zip(SCENARIOS)
        .filter(|(_, scenario)| parts.iter().any(|part| scenario.name.contains(part)))
        .collect();
    if std::env::var_os(PARALLEL).is_some() || parts.len() > 1 {
        // All at once, each reported when all are done, in order.
        let running: Vec<_> = chosen
            .into_iter()
            .map(|(seed, scenario)| tokio::spawn(run(seed, scenario)))
            .collect();
        for report in running {
            print!("{}", report.await.expect("scenario panicked"));
        }
    } else {
        for (seed, scenario) in chosen {
            print!("{}", run(seed, scenario).await);
        }
    }
}

/// What all nodes have sent, counted, and their relay counters and propagation delays
/// summed: at one moment.
#[derive(Clone)]
struct Totals {
    at: Instant,
    sent: ByKind,
    relay: Relay,
    /// What went wrong with peers (`PeerState::incidents`), summed over every node's.
    incidents: Incidents,
    propagation: Propagation,
    /// Per node, by name: its arrivals, their delays summed, and those later than
    /// `LATE_MS`.
    by_node: Vec<(String, u64, u64, u64)>,
    /// Per node, by name: the changes it asked for that the asked had counted as its
    /// (`PeerState::asked_known`), summed over those it asked.
    asked_by: HashMap<String, u64>,
}

/// Arrivals later than this are late (`report`): the propagation buckets above it.
const LATE_MS: u64 = 500;

impl Totals {
    async fn of(nodes: &[AppState]) -> Self {
        let mut totals = Self {
            at: Instant::now(),
            sent: ByKind::default(),
            relay: Relay::default(),
            incidents: Incidents::default(),
            propagation: Propagation::default(),
            by_node: Vec::new(),
            asked_by: HashMap::new(),
        };
        let names: HashMap<String, String> = nodes
            .iter()
            .map(|node| {
                let id = node.local_id().to_owned();
                (id.clone(), node.alias.clone().unwrap_or(id))
            })
            .collect();
        for node in nodes {
            let sent = node.traffic.total().sent;
            for (total, theirs) in counters_mut(&mut totals.sent)
                .into_iter()
                .zip(counters(&sent))
            {
                total.messages += theirs.messages;
                total.bytes += theirs.bytes;
            }
            let stats = node.stats.snapshot();
            let relay = &mut totals.relay;
            relay.announce_rounds += stats.relay.announce_rounds;
            relay.treeless_rounds += stats.relay.treeless_rounds;
            relay.slow_rounds += stats.relay.slow_rounds;
            relay.pulls += stats.relay.pulls;
            relay.pulls_queued += stats.relay.pulls_queued;
            relay.tree_forwards += stats.relay.tree_forwards;
            relay.tree_from_others += stats.relay.tree_from_others;
            relay.tree_missed += stats.relay.tree_missed;
            relay.delivered += stats.relay.delivered;
            relay.delivered_bytes += stats.relay.delivered_bytes;
            relay.framed += stats.relay.framed;
            relay.framing_bytes += stats.relay.framing_bytes;
            relay.deflated += stats.relay.deflated;
            relay.deflated_from += stats.relay.deflated_from;
            relay.deflated_to += stats.relay.deflated_to;
            relay.nothing_new += stats.relay.nothing_new;
            relay.nothing_new_bytes += stats.relay.nothing_new_bytes;
            relay.nothing_new_along_trees += stats.relay.nothing_new_along_trees;
            relay.nothing_new_delivered += stats.relay.nothing_new_delivered;
            relay.carrying += stats.relay.carrying;
            relay.carrying_change_bytes += stats.relay.carrying_change_bytes;
            relay.new_changes += stats.relay.new_changes;
            relay.asked_known_lately += stats.relay.asked_known_lately;
            relay.asked_known_later += stats.relay.asked_known_later;
            relay.answering_behind += stats.relay.answering_behind;
            relay.nothing_new_change_bytes += stats.relay.nothing_new_change_bytes;
            relay.waiting += stats.relay.waiting;
            relay.off_tree += stats.relay.off_tree;
            relay.off_tree_ms += stats.relay.off_tree_ms;
            for (peer, peer_state) in node.peers_snapshot().await {
                let peer_state = peer_state.lock().await;
                totals.incidents.add(&peer_state.incidents);
                let name = names.get(&peer).cloned().unwrap_or(peer);
                *totals.asked_by.entry(name).or_default() += peer_state.asked_known;
            }
            let propagation = &mut totals.propagation;
            for (sum, count) in propagation
                .buckets
                .iter_mut()
                .zip(stats.propagation.buckets)
            {
                *sum += count;
            }
            propagation.count += stats.propagation.count;
            propagation.sum_ms += stats.propagation.sum_ms;
            let late: u64 = stats
                .propagation
                .buckets()
                .filter(|(bound, _)| bound.is_none_or(|bound| bound > LATE_MS))
                .map(|(_, count)| count)
                .sum();
            totals.by_node.push((
                node.alias
                    .clone()
                    .unwrap_or_else(|| node.local_id().to_owned()),
                stats.propagation.count,
                stats.propagation.sum_ms,
                late,
            ));
        }
        totals
    }
}

/// The kinds, in the order of the report.
const KINDS: [&str; 11] = [
    "SYNC",
    "STATUS",
    "STATUS_REPLY",
    "BLOB",
    "HELLO",
    "HELLO_REPLY",
    "PING/PONG",
    "WHO/ME",
    "LINKS",
    "PROBE",
    "PRUNE/GRAFT",
];

fn counters(by: &ByKind) -> [Counter; 11] {
    [
        by.sync,
        by.status,
        by.status_reply,
        by.blob,
        by.hello,
        by.hello_reply,
        by.paths,
        by.identity,
        by.links,
        by.probe,
        by.plumtree,
    ]
}

fn counters_mut(by: &mut ByKind) -> [&mut Counter; 11] {
    [
        &mut by.sync,
        &mut by.status,
        &mut by.status_reply,
        &mut by.blob,
        &mut by.hello,
        &mut by.hello_reply,
        &mut by.paths,
        &mut by.identity,
        &mut by.links,
        &mut by.probe,
        &mut by.plumtree,
    ]
}

/// Runs `scenario`; returns its report.
async fn run(seed: u64, scenario: &'static Scenario) -> String {
    let mut out = String::new();
    let sim = Sim::with_identity(seed);
    let ids: Vec<String> = (0..scenario.nodes).map(|i| format!("n{i:03}")).collect();
    (scenario.net)(&sim, &ids);
    let reply_timeout = scenario.reply_timeout.unwrap_or(SYNC_REPLY_TIMEOUT);
    let timing = move |config: &mut crate::node::NodeConfig| {
        config.timing = Timing::new(KEEPALIVE, reply_timeout, DELIVERY_LIMIT);
        // As in production (`--neighbours`).
        config.neighbours_k = 12;
    };
    let mut nodes = vec![sim.start_on_with(&ids[0], (scenario.on)(0), &[], Start::Genesis, timing)];
    // Those there from the start; the others join during the workloads.
    let initial = scenario.nodes - 2 * scenario.joining;
    for (i, id) in ids.iter().enumerate().take(initial).skip(1) {
        // The writer behind NAT reaches node 2 alone.
        let seed = if i == 1 {
            ids[2].as_str()
        } else {
            ids[0].as_str()
        };
        nodes.push(sim.start_on_with(id, (scenario.on)(i), &[seed], Start::Join, timing));
    }
    let all: Vec<&AppState> = nodes.iter().collect();
    let all = all.as_slice();
    sim.wait_ready(all, SETTLE).await;
    let count = initial;
    sim.wait_until(
        "every node knows every member",
        SETTLE,
        move || async move {
            for node in all {
                if members::members(&node.doc.read().await.doc).len() < count {
                    return false;
                }
            }
            true
        },
    )
    .await;
    sim.wait_converged(all, SETTLE).await;
    // The links that don't work recorded (`swarm::working_links`, after their hold), so
    // the trees are built over the links that do from the first write: every node
    // checked since, with nothing awaiting its hold. Lossy links may keep some waiting:
    // not for longer than settling.
    let checks: Vec<u64> = all
        .iter()
        .map(|node| crate::swarm::working_links::settled(node).0)
        .collect();
    let recording = Instant::now();
    while recording.elapsed() < SETTLE
        && !all.iter().zip(&checks).all(|(node, before)| {
            let (checks, settled) = crate::swarm::working_links::settled(node);
            checks > *before && settled
        })
    {
        tokio::time::sleep(Duration::from_secs(1)).await;
    }
    // Neighbours chosen, handshakes done, what joining set off over: sync messages no
    // fewer in a half of the quiet phase than in the one before (some go on for good:
    // alive stamps, registrations). Those two halves are the quiet phase.
    let settling = Instant::now();
    let mut halves = vec![Totals::of(&nodes).await];
    loop {
        tokio::time::sleep(QUIET / 2).await;
        halves.push(Totals::of(&nodes).await);
        let count = |i: usize| halves[i + 1].sent.sync.messages - halves[i].sent.sync.messages;
        let n = halves.len();
        if (n >= 3 && count(n - 2) >= count(n - 3)) || settling.elapsed() > SETTLE {
            break;
        }
    }
    let n = halves.len();
    let quiet = (halves[n.saturating_sub(3)].clone(), halves[n - 1].clone());

    let _ = writeln!(
        out,
        "\n{} ({} writes at {} a second each; {})",
        scenario.name,
        writes(),
        1000 / WRITE_EVERY.as_millis(),
        all[0].dissemination.name()
    );
    let mut joined = initial;
    for (w, (workload, writer)) in [
        ("one writer", Some(scenario.writer)),
        ("writers in turn", None),
    ]
    .into_iter()
    .enumerate()
    {
        let start = Totals::of(&nodes).await;
        // Before which writes the members joining in this workload start.
        let joins: Vec<usize> = (1..=scenario.joining)
            .map(|k| writes() * k / (scenario.joining + 1))
            .collect();
        for i in 0..writes() {
            if joins.contains(&i) {
                let id = &ids[joined];
                nodes.push(sim.start_on_with(
                    id,
                    (scenario.on)(joined),
                    &[ids[0].as_str()],
                    Start::Join,
                    timing,
                ));
                joined += 1;
            }
            // The middle half, the node that goes away is gone (writing nothing).
            if let Some(away) = scenario.away {
                if i == writes() / 4 {
                    sim.net.take_down(&ids[away]);
                } else if i == writes() * 3 / 4 {
                    sim.net.bring_up(&ids[away]);
                }
            }
            let gone = scenario.away.is_some_and(|away| {
                (writes() / 4..writes() * 3 / 4).contains(&i) && away == i % initial
            });
            if writer.is_none() && gone {
                tokio::time::sleep(WRITE_EVERY).await;
                continue;
            }
            // In turn, those there from the start: one that just joined may not be
            // ready yet.
            let node = &nodes[writer.unwrap_or(i % initial)];
            // A path of each workload's own: writing what is there already changes
            // nothing, and nothing would pass.
            write(node, &format!("bench/{w}/{i}"), json!(i)).await;
            tokio::time::sleep(WRITE_EVERY).await;
        }
        let written = Instant::now();
        let all: Vec<&AppState> = nodes.iter().collect();
        sim.wait_converged(&all, SETTLE).await;
        let converged = written.elapsed();
        report(
            &mut out,
            workload,
            quiet.clone(),
            (start, Totals::of(&nodes).await),
            converged,
        );
        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;
        }
    }
    out
}

/// Every node's own writer's tree (`swarm::tree::own_shape`, `/admin/stats`), as it
/// computes it at the end: the hops its writes take, and the members it leaves out;
/// and how long they take by the links' latencies alone, against parents chosen by
/// latency and the fastest paths (`swarm::tree::own_latencies`).
async fn trees(out: &mut String, nodes: &[AppState], net: &crate::transport::sim::SimNet) {
    let mut shapes = Vec::new();
    for node in nodes {
        shapes.extend(crate::swarm::tree::own_shape(node).await);
    }
    if shapes.is_empty() {
        return;
    }
    let count = shapes.len() as f64;
    let members: usize = shapes.iter().map(|shape| shape.members).sum();
    let hops: u64 = shapes.iter().map(|shape| shape.hops).sum();
    let deepest = shapes.iter().map(|shape| shape.depth).max().unwrap_or(0);
    let depth: f64 = shapes
        .iter()
        .map(|shape| f64::from(shape.depth))
        .sum::<f64>()
        / count;
    let off: usize = shapes.iter().map(|shape| shape.off).sum();
    let _ = writeln!(
        out,
        "    own trees ({} nodes): depth {depth:.1} on average, {deepest} at most; {:.2} hops to a member on average; {:.1} members off each",
        shapes.len(),
        if members > 0 {
            hops as f64 / members as f64
        } else {
            0.0
        },
        off as f64 / count,
    );

    // Node IDs to the net's names (each node's alias).
    let names: std::collections::HashMap<String, String> = nodes
        .iter()
        .filter_map(|node| Some((node.local_id().to_owned(), node.alias.clone()?)))
        .collect();
    let latency = |from: &str, to: &str| match (names.get(from), names.get(to)) {
        (Some(from), Some(to)) => net.mean_delay(from, to).as_secs_f64() * 1000.0,
        _ => f64::NAN,
    };
    let mut latencies = Vec::new();
    let mut views = Views::default();
    for node in nodes {
        // Each member's parent in this node's tree, as the member chose it.
        let mut parents = std::collections::HashMap::new();
        for member in nodes {
            if member.local_id() != node.local_id() {
                let chosen = crate::swarm::tree::own_parent(member, node.local_id()).await;
                parents.insert(member.local_id().to_owned(), chosen);
            }
        }
        views.add(nodes, node, &parents).await;
        latencies.extend(crate::swarm::tree::own_latencies(node, &parents, &latency).await);
    }
    views.report(out);
    let members: usize = latencies.iter().map(|l| l.members).sum();
    if members == 0 {
        return;
    }
    let mean = |of: fn(&crate::swarm::tree::Latencies) -> f64| {
        latencies
            .iter()
            .map(|l| of(l) * l.members as f64)
            .sum::<f64>()
            / members as f64
    };
    let _ = writeln!(
        out,
        "    by link latency, to a member: {:.0} ms along own trees; {:.0} ms with parents chosen by latency; {:.0} ms by the fastest paths",
        mean(|l| l.tree),
        mean(|l| l.by_latency),
        mean(|l| l.fastest),
    );
}

/// How many changes each session counts as sent to its peer (`SyncSession::sent_hashes_len`),
/// after the workload: the most, and on average. Kept until the peer's next message
/// says it has them; a session whose peer never sends one (along a tree, with one
/// writer) keeps all. And of those, how many because another parent brings them
/// (`SyncSession::known_to_peer`): a session with a neighbour we count as covered.
pub(super) async fn sent_kept(out: &mut String, nodes: &[AppState]) {
    let (mut most, mut sum) = (0, 0);
    let (mut most_elsewhere, mut sum_elsewhere) = (0, 0);
    let mut sessions = 0;
    for node in nodes {
        for (_, peer_state) in node.peers_snapshot().await {
            let peer_state = peer_state.lock().await;
            let kept = peer_state.sync.sent_hashes_len();
            let elsewhere = peer_state.sync.sent_known_elsewhere_len();
            most = most.max(kept);
            sum += kept;
            most_elsewhere = most_elsewhere.max(elsewhere);
            sum_elsewhere += elsewhere;
            sessions += 1;
        }
    }
    let sessions = f64::from(sessions.max(1));
    let _ = writeln!(
        out,
        "    changes counted as sent, per session: {most} at most, {:.1} on average; \
         of those, because another parent brings them: {most_elsewhere} at most, {:.1} on average",
        sum as f64 / sessions,
        sum_elsewhere as f64 / sessions,
    );
}

/// A link with a longer round trip is slow (`eager_links`): over Tor, a slow circuit
/// (`tor_like`).
const SLOW_LINK_MS: u64 = 1000;

/// The mean of `values`, rounded; 0 for none.
fn mean(values: &[u64]) -> u64 {
    if values.is_empty() {
        0
    } else {
        values.iter().sum::<u64>() / values.len() as u64
    }
}

/// 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), the links eager at one end only (a
/// PRUNE or GRAFT lost, or crossing), and the round trips of eager links against lazy
/// ones (does the tree take slow links?). 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();
    // The round trips of eager and of lazy links, as each node measured them.
    let (mut eager_rtts, mut lazy_rtts) = (Vec::new(), Vec::new());
    for node in nodes {
        let Some(lazy) = node.dissemination.lazy() else {
            return;
        };
        let (eager, others): (Vec<String>, Vec<String>) =
            crate::swarm::topology::our_neighbours(node)
                .into_iter()
                .partition(|peer| !lazy.contains(peer));
        {
            let rtt = node.lock_rtt();
            let now = tokio::time::Instant::now();
            let fresh = node.timing.rtt_fresh_for();
            for (peers, rtts) in [(&eager, &mut eager_rtts), (&others, &mut lazy_rtts)] {
                rtts.extend(
                    peers
                        .iter()
                        .filter_map(|peer| rtt.effective(peer, now, fresh))
                        .map(|rtt| rtt.as_millis() as u64),
                );
            }
        }
        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; \
         round trips of eager links {} ms on average, {} of them over {SLOW_LINK_MS} ms; of lazy links {} ms",
        out_sum as f64 / count,
        unpushed.len(),
        if unpushed.is_empty() {
            String::new()
        } else {
            format!(" ({})", unpushed.join(", "))
        },
        mean(&eager_rtts),
        eager_rtts.iter().filter(|rtt| **rtt > SLOW_LINK_MS).count(),
        mean(&lazy_rtts),
    );
}

/// 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
/// covered). Summed over every node's tree.
#[derive(Default)]
struct Views {
    /// Child links planned (a node sends to its children).
    planned: usize,
    /// Of those, the child's own choice of parent.
    confirmed: usize,
    /// Members whose chosen parent doesn't plan them as a child, so that nothing comes
    /// along the tree: the parent counts them as covered (as having the changes from
    /// another), or neither (it doesn't reach them, or ranks them before itself).
    orphans_covered: usize,
    orphans_neither: usize,
    /// Members two nodes or more plan as a child (all but one send in vain).
    doubled: usize,
    /// Members (not writers) that chose no parent: announcements bring them all.
    parentless: usize,
}

impl Views {
    async fn add(
        &mut self,
        nodes: &[AppState],
        writer: &AppState,
        parents: &HashMap<String, Option<String>>,
    ) {
        let key = crate::swarm::topology::pair_key(writer.local_id());
        let mut plans = HashMap::new();
        for node in nodes {
            if let Some(plan) = crate::swarm::tree::plan(node, key).await {
                plans.insert(node.local_id().to_owned(), plan);
            }
        }
        let mut planned_by: HashMap<&str, usize> = HashMap::new();
        for (node, plan) in &plans {
            for child in &plan.children {
                self.planned += 1;
                *planned_by.entry(child.as_str()).or_default() += 1;
                if parents
                    .get(child)
                    .is_some_and(|parent| parent.as_ref() == Some(node))
                {
                    self.confirmed += 1;
                }
            }
        }
        self.doubled += planned_by.values().filter(|count| **count > 1).count();
        for (member, parent) in parents {
            let Some(parent) = parent else {
                self.parentless += 1;
                continue;
            };
            let Some(plan) = plans.get(parent) else {
                continue;
            };
            if plan.children.contains(member) {
                continue;
            }
            if plan.covered.contains(member) {
                self.orphans_covered += 1;
            } else {
                self.orphans_neither += 1;
            }
        }
    }

    fn report(&self, out: &mut String) {
        let _ = writeln!(
            out,
            "    views of the trees: {} of {} child links the child's own choice; orphans (the parent it chose doesn't send): {} counted covered, {} neither; {} planned as a child twice or more; {} members without a parent",
            self.confirmed,
            self.planned,
            self.orphans_covered,
            self.orphans_neither,
            self.doubled,
            self.parentless,
        );
    }
}

/// The members `writer`'s changes take longest to reach along its tree, by the links'
/// latencies alone (the parents as each member chose them, `swarm::tree::own_place`),
/// those later than `LATE_MS`: how far each is from the writer, its parent and the
/// latency to it, and the members it could take instead (they come before it) and the
/// fastest link to one. One way, in milliseconds.
async fn slowest(
    out: &mut String,
    nodes: &[AppState],
    writer: &AppState,
    net: &crate::transport::sim::SimNet,
) {
    let names: HashMap<String, String> = nodes
        .iter()
        .filter_map(|node| Some((node.local_id().to_owned(), node.alias.clone()?)))
        .collect();
    let name = |id: &str| names.get(id).cloned().unwrap_or_else(|| id.to_owned());
    let latency =
        |from: &str, to: &str| net.mean_delay(&name(from), &name(to)).as_secs_f64() * 1000.0;
    let mut places = HashMap::new();
    for node in nodes {
        if node.local_id() != writer.local_id()
            && let Some(place) = crate::swarm::tree::own_place(node, writer.local_id()).await
        {
            places.insert(node.local_id().to_owned(), place);
        }
    }
    // Arrival along parents; None where the chain ends without the writer.
    fn arrival(
        member: &str,
        writer: &str,
        places: &HashMap<String, crate::swarm::tree::Place>,
        latency: &dyn Fn(&str, &str) -> f64,
        depth: usize,
    ) -> Option<f64> {
        if member == writer {
            return Some(0.0);
        }
        let parent = places.get(member)?.parent.as_deref()?;
        if depth > places.len() {
            return None;
        }
        Some(arrival(parent, writer, places, latency, depth + 1)? + latency(parent, member))
    }
    let mut late: Vec<(f64, &String)> = places
        .keys()
        .filter_map(|member| {
            arrival(member, writer.local_id(), &places, &latency, 0).map(|at| (at, member))
        })
        .filter(|(at, _)| *at > LATE_MS as f64)
        .collect();
    if late.is_empty() {
        return;
    }
    late.sort_by(|a, b| b.0.total_cmp(&a.0));
    let listed: Vec<String> = late
        .iter()
        .take(5)
        .map(|(at, member)| {
            let place = &places[member.as_str()];
            let parent = place.parent.as_deref().unwrap_or_default();
            let fastest = place
                .before
                .iter()
                .map(|other| (latency(other, member), other))
                .min_by(|a, b| a.0.total_cmp(&b.0));
            format!(
                "{} {at:.0} ms ({} hops; from {} over {:.0} ms; of {} it could take, {})",
                name(member),
                place.distance,
                name(parent),
                latency(parent, member),
                place.before.len(),
                fastest.map_or_else(String::new, |(ms, other)| format!(
                    "{} over {ms:.0} ms",
                    name(other)
                )),
            )
        })
        .collect();
    let _ = writeln!(
        out,
        "    later than {LATE_MS} ms by the links along the writer's tree: {}{}",
        listed.join("; "),
        if late.len() > 5 { "; …" } else { "" }
    );
}

/// The workload's counts (`during`), and the quiet phase's for as long.
fn report(
    out: &mut String,
    workload: &str,
    quiet: (Totals, Totals),
    during: (Totals, Totals),
    converged: Duration,
) {
    let took = during.1.at - during.0.at;
    let scale = took.as_secs_f64() / (quiet.1.at - quiet.0.at).as_secs_f64();
    let writes = writes() as f64;
    let _ = writeln!(
        out,
        "  {workload}: {:.1} s, converged {} ms after the last write",
        took.as_secs_f64(),
        converged.as_millis()
    );
    let _ = writeln!(
        out,
        "    {:<13} {:>9} {:>9} {:>11}   (quiet phase, as long: msgs)",
        "kind", "msgs", "/write", "bytes/write"
    );
    let rows = KINDS
        .iter()
        .zip(
            counters(&quiet.0.sent)
                .into_iter()
                .zip(counters(&quiet.1.sent)),
        )
        .zip(
            counters(&during.0.sent)
                .into_iter()
                .zip(counters(&during.1.sent)),
        );
    let (mut all_messages, mut all_bytes) = (0.0, 0.0);
    for ((kind, (q0, q1)), (d0, d1)) in rows {
        let quiet_messages = (q1.messages - q0.messages) as f64 * scale;
        let messages = (d1.messages - d0.messages) as f64;
        let bytes = (d1.bytes - d0.bytes) as f64;
        all_messages += messages;
        all_bytes += bytes;
        let _ = writeln!(
            out,
            "    {kind:<13} {messages:>9.0} {:>9.1} {:>11.0}   ({quiet_messages:.0})",
            messages / writes,
            bytes / writes
        );
    }
    let _ = writeln!(
        out,
        "    {:<13} {all_messages:>9.0} {:>9.1} {:>11.0}",
        "all",
        all_messages / writes,
        all_bytes / writes
    );
    let (r0, r1) = (during.0.relay, during.1.relay);
    let _ = writeln!(
        out,
        "    relay per write: {:.1} announcement rounds ({:.1} without a tree, {:.1} slow), {:.1} pulls ({:.1} queued), {:.1} along writers' trees",
        (r1.announce_rounds - r0.announce_rounds) as f64 / writes,
        (r1.treeless_rounds - r0.treeless_rounds) as f64 / writes,
        (r1.slow_rounds - r0.slow_rounds) as f64 / writes,
        (r1.pulls - r0.pulls) as f64 / writes,
        (r1.pulls_queued - r0.pulls_queued) as f64 / writes,
        (r1.tree_forwards - r0.tree_forwards) as f64 / writes,
    );
    let _ = writeln!(
        out,
        "    trees as the nodes see them (`/admin/stats`), per write: {:.2} sync messages along a tree from another than the parent, {:.2} changes the trees didn't bring",
        (r1.tree_from_others - r0.tree_from_others) as f64 / writes,
        (r1.tree_missed - r0.tree_missed) as f64 / writes,
    );
    let per = |count: u64, bytes: u64| {
        let count = count as f64;
        (
            count / writes,
            if count > 0.0 {
                bytes as f64 / count
            } else {
                0.0
            },
        )
    };
    let delivered = per(
        r1.delivered - r0.delivered,
        r1.delivered_bytes - r0.delivered_bytes,
    );
    let nothing_new = per(
        r1.nothing_new - r0.nothing_new,
        r1.nothing_new_bytes - r0.nothing_new_bytes,
    );
    let _ = writeln!(
        out,
        "    sync per write: {:.1} taken as delivered ({:.0} bytes each), {:.1} with only changes the receiver had ({:.0} bytes each; {:.1} along trees, {:.1} taken as delivered), {:.1} with changes waiting for predecessors",
        delivered.0,
        delivered.1,
        nothing_new.0,
        nothing_new.1,
        (r1.nothing_new_along_trees - r0.nothing_new_along_trees) as f64 / writes,
        (r1.nothing_new_delivered - r0.nothing_new_delivered) as f64 / writes,
        (r1.waiting - r0.waiting) as f64 / writes,
    );
    let each = |bytes: u64, count: u64| {
        if count > 0 {
            bytes as f64 / count as f64
        } else {
            0.0
        }
    };
    let carrying = r1.carrying - r0.carrying;
    let twice = r1.nothing_new - r0.nothing_new;
    let _ = writeln!(
        out,
        "    received with changes: {:.1} per write, each {:.0} bytes of changes; of those with nothing new, {:.0}",
        carrying as f64 / writes,
        each(
            r1.carrying_change_bytes - r0.carrying_change_bytes,
            carrying
        ),
        each(
            r1.nothing_new_change_bytes - r0.nothing_new_change_bytes,
            twice
        ),
    );
    // Bytes of changes per new change in the messages that brought some: several
    // changes per message, or larger changes. Not those whose changes wait for
    // predecessors: they count as new with the message that brings those.
    let with_news = carrying - twice - (r1.waiting - r0.waiting);
    let news = r1.new_changes - r0.new_changes;
    let _ = writeln!(
        out,
        "    of those with something new: {:.1} new changes each, {:.0} bytes of changes per new change",
        each(news, with_news),
        each(
            (r1.carrying_change_bytes - r0.carrying_change_bytes)
                - (r1.nothing_new_change_bytes - r0.nothing_new_change_bytes),
            news
        ),
    );
    // What our envelope adds to each Automerge message, beside the message itself.
    let framed = r1.framed - r0.framed;
    let _ = writeln!(
        out,
        "    sync envelopes: {:.0} bytes each beside the Automerge message",
        each(r1.framing_bytes - r0.framing_bytes, framed),
    );
    let deflated = r1.deflated - r0.deflated;
    let _ = writeln!(
        out,
        "    deflated: {:.1} sync messages per write, {:.0} bytes each to {:.0}",
        deflated as f64 / writes,
        each(r1.deflated_from - r0.deflated_from, deflated),
        each(r1.deflated_to - r0.deflated_to, deflated),
    );
    let (lately, later) = (
        r1.asked_known_lately - r0.asked_known_lately,
        r1.asked_known_later - r0.asked_known_later,
    );
    let _ = writeln!(
        out,
        "    asked for, though counted as the asker's (another parent brings them): {:.2} per write ({lately} within 2 s, {later} later); {:.2} per write taken as delivered by a sender behind the receiver, answered",
        (lately + later) as f64 / writes,
        (r1.answering_behind - r0.answering_behind) as f64 / writes,
    );
    let (i0, i1) = (during.0.incidents, during.1.incidents);
    let _ = writeln!(
        out,
        "    incidents: {} reply timeouts ({:.2} per write), {} heals, {} asked again ({:.1} per write), {} dormancies",
        i1.reply_timeouts - i0.reply_timeouts,
        (i1.reply_timeouts - i0.reply_timeouts) as f64 / writes,
        i1.heals - i0.heals,
        i1.asked_again - i0.asked_again,
        (i1.asked_again - i0.asked_again) as f64 / writes,
        i1.dormancies - i0.dormancies,
    );
    let mut propagation = Propagation::default();
    let (p0, p1) = (during.0.propagation, during.1.propagation);
    for ((sum, a), b) in propagation
        .buckets
        .iter_mut()
        .zip(p0.buckets)
        .zip(p1.buckets)
    {
        *sum = b - a;
    }
    propagation.count = p1.count - p0.count;
    propagation.sum_ms = p1.sum_ms - p0.sum_ms;
    let within = |quantile| {
        super::soak::percentile_bound(&propagation, quantile)
            .map_or("more than an hour".to_owned(), |ms| format!("{ms} ms"))
    };
    let _ = writeln!(
        out,
        "    propagation: {} arrivals, mean {} ms, half within {}, 99% within {}",
        propagation.count,
        propagation.sum_ms / propagation.count.max(1),
        within(0.5),
        within(0.99),
    );
    let off_tree = r1.off_tree - r0.off_tree;
    let along = propagation.count.saturating_sub(off_tree);
    let _ = writeln!(
        out,
        "    of those, along writers' trees: {along} (mean {} ms); not: {off_tree} (mean {} ms)",
        propagation
            .sum_ms
            .saturating_sub(r1.off_tree_ms - r0.off_tree_ms)
            / along.max(1),
        (r1.off_tree_ms - r0.off_tree_ms) / off_tree.max(1),
    );
    // Where the late arrivals were: the nodes that had any, the most first.
    let mut late: Vec<(&str, u64, u64)> = during
        .0
        .by_node
        .iter()
        .zip(&during.1.by_node)
        .map(|((name, c0, s0, l0), (_, c1, s1, l1))| {
            (name.as_str(), l1 - l0, (s1 - s0) / (c1 - c0).max(1))
        })
        .filter(|(_, late, _)| *late > 0)
        .collect();
    late.sort_by_key(|(_, late, _)| std::cmp::Reverse(*late));
    if !late.is_empty() {
        let listed: Vec<String> = late
            .iter()
            .take(5)
            .map(|(name, late, mean)| format!("{name} {late} (mean {mean} ms)"))
            .collect();
        let _ = writeln!(
            out,
            "    later than {LATE_MS} ms: {}{}",
            listed.join(", "),
            if late.len() > 5 { ", …" } else { "" }
        );
    }
    // Who asked for what it was counted as having: the most first, with how many of the
    // writes reached it late; and the share of the nodes most writes reached late.
    let late_of: HashMap<&str, u64> = late.iter().map(|(name, late, _)| (*name, *late)).collect();
    let mut askers: Vec<(&str, u64)> = during
        .1
        .asked_by
        .iter()
        .map(|(name, asked)| {
            let before = during.0.asked_by.get(name).copied().unwrap_or(0);
            (name.as_str(), asked - before)
        })
        .filter(|(_, asked)| *asked > 0)
        .collect();
    askers.sort_by_key(|(name, asked)| (std::cmp::Reverse(*asked), *name));
    let asked: u64 = askers.iter().map(|(_, asked)| asked).sum();
    if asked > 0 {
        let listed: Vec<String> = askers
            .iter()
            .take(5)
            .map(|(name, asked)| {
                format!(
                    "{name} {asked} ({} late)",
                    late_of.get(name).copied().unwrap_or(0)
                )
            })
            .collect();
        let writes_seen = |name: &str| {
            during
                .0
                .by_node
                .iter()
                .zip(&during.1.by_node)
                .find(|((n, ..), _)| n == name)
                .map_or(0, |((_, c0, ..), (_, c1, ..))| c1 - c0)
        };
        let mostly_late: Vec<&str> = late
            .iter()
            .filter(|(name, late, _)| 2 * late > writes_seen(name))
            .map(|(name, ..)| *name)
            .collect();
        let by_mostly_late: u64 = askers
            .iter()
            .filter(|(name, _)| mostly_late.contains(name))
            .map(|(_, asked)| asked)
            .sum();
        let _ = writeln!(
            out,
            "    asked for, though counted as theirs, by: {}{}; {:.0}% by the {} nodes most writes reached later than {LATE_MS} ms",
            listed.join(", "),
            if askers.len() > 5 { ", …" } else { "" },
            100.0 * by_mostly_late as f64 / asked as f64,
            mostly_late.len(),
        );
    }
}

/// Values written in `bench_bridge`, unless `BENCH_BRIDGE_WRITES` says otherwise.
const BRIDGE_WRITES: usize = 200;
/// Their size: random, so nothing compresses them.
const BRIDGE_VALUE: usize = 256 * 1024;
/// A hundred a second.
const BRIDGE_EVERY: Duration = Duration::from_millis(10);
/// The fast transport and the slow one, unless `BENCH_BRIDGE_FAST_RATE`,
/// `BENCH_BRIDGE_SLOW_RATE` (MiB/s) and `BENCH_BRIDGE_FAST_RTT`, `BENCH_BRIDGE_SLOW_RTT`
/// (ms) say otherwise.
const BRIDGE_FAST_RATE_MIB: u64 = 1024;
const BRIDGE_FAST_RTT_MS: u64 = 5;
const BRIDGE_SLOW_RATE_MIB: u64 = 50;
const BRIDGE_SLOW_RTT_MS: u64 = 500;

/// A writer on a fast transport (5 ms round trips, 1 GiB/s, by default), a bridge on it
/// and a slow one (500 ms round trips, 50 MiB/s), and a node on the slow one alone: a writes 256
/// KiB values at 100 a second (25 MiB/s, a quarter of the fast link, half of the slow
/// one), b passes them on to z. What z gets, as sync messages and as blobs (deltas),
/// against what was written, and how late.
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
#[ignore = "a benchmark: cargo benchmarks"]
async fn bench_bridge() {
    println!();
    let env = |name: &str, default| {
        std::env::var(name)
            .ok()
            .and_then(|value| value.parse().ok())
            .unwrap_or(default)
    };
    let writes = env("BENCH_BRIDGE_WRITES", BRIDGE_WRITES as u64) as usize;
    let fast_rate = env("BENCH_BRIDGE_FAST_RATE", BRIDGE_FAST_RATE_MIB);
    let fast_rtt = env("BENCH_BRIDGE_FAST_RTT", BRIDGE_FAST_RTT_MS);
    let slow_rate = env("BENCH_BRIDGE_SLOW_RATE", BRIDGE_SLOW_RATE_MIB);
    let slow_rtt = env("BENCH_BRIDGE_SLOW_RTT", BRIDGE_SLOW_RTT_MS);
    let sim = Sim::with_identity(71);
    // Half the round trip each way, give or take a fifth.
    let link = |rate_mib: u64, rtt_ms: u64| {
        let half = Duration::from_micros(rtt_ms * 500);
        LinkParams {
            delay: (half - half / 5, half + half / 5),
            rate: Some(rate_mib << 20),
            ..LinkParams::default()
        }
    };
    sim.net.set_default_link(link(fast_rate, fast_rtt));
    sim.net_for("sim2")
        .set_default_link(link(slow_rate, slow_rtt));
    let timing = |config: &mut crate::node::NodeConfig| {
        config.timing = Timing::new(KEEPALIVE, SYNC_REPLY_TIMEOUT, DELIVERY_LIMIT);
        config.neighbours_k = 12;
    };
    let a = sim.start_on_with("a", &["sim"], &[], Start::Genesis, timing);
    let b = sim.start_on_with("b", &["sim", "sim2"], &["a"], Start::Join, timing);
    let z = sim.start_on_with("z", &["sim2"], &["b"], Start::Join, timing);
    let all = [&a, &b, &z];
    sim.wait_ready(&all, SETTLE).await;
    sim.wait_until("every node knows every member", SETTLE, || async {
        for node in all {
            if members::members(&node.doc.read().await.doc).len() < 3 {
                return false;
            }
        }
        true
    })
    .await;
    sim.wait_converged(&all, SETTLE).await;

    let received = |node: &AppState| node.traffic.total().received;
    let (z0, b0) = (received(&z), b.traffic.total().sent);
    let p0 = z.stats.snapshot().propagation;
    let served0 = b.stats.snapshot().relay;
    let fetched0 = z.stats.snapshot().relay;
    let mut rng = crate::testrand::Rng::new(72);
    let started = Instant::now();
    for i in 0..writes {
        let value: String = (0..BRIDGE_VALUE / 16)
            .map(|_| format!("{:016x}", rng.next_u64()))
            .collect();
        write(&a, &format!("bridge/{i}"), json!(value)).await;
        // At the pace, however long a write took.
        let due = started + BRIDGE_EVERY * (i as u32 + 1);
        tokio::time::sleep_until(due).await;
    }
    let written_in = started.elapsed();
    let written = Instant::now();
    sim.wait_converged(&all, SETTLE).await;
    let converged = written.elapsed();

    let (z1, b1) = (received(&z), b.traffic.total().sent);
    let p1 = z.stats.snapshot().propagation;
    let served1 = b.stats.snapshot().relay;
    let fetched1 = z.stats.snapshot().relay;
    let diff = |c1: Counter, c0: Counter| (c1.messages - c0.messages, c1.bytes - c0.bytes);
    let mib = |bytes: u64| bytes as f64 / f64::from(1 << 20);
    let total = (writes * BRIDGE_VALUE) as u64;
    let (z_sync, z_blob) = (diff(z1.sync, z0.sync), diff(z1.blob, z0.blob));
    let (b_sync, b_blob) = (diff(b1.sync, b0.sync), diff(b1.blob, b0.blob));
    let mut propagation = Propagation::default();
    for ((sum, before), after) in propagation
        .buckets
        .iter_mut()
        .zip(p0.buckets)
        .zip(p1.buckets)
    {
        *sum = after - before;
    }
    propagation.count = p1.count - p0.count;
    propagation.sum_ms = p1.sum_ms - p0.sum_ms;
    let within = |quantile| {
        propagation
            .within(quantile)
            .map_or("beyond the last bucket".to_owned(), |ms| format!("{ms} ms"))
    };
    println!(
        "\nfast writer, bridge, slow node ({writes} values of {} KiB, {} a second; fast link {fast_rate} MiB/s, {fast_rtt} ms round trips; slow link {slow_rate} MiB/s, {slow_rtt} ms round trips)\n  \
         written: {:.1} MiB in {:.1} s; z converged {} ms after the last write\n  \
         b sent (to a and z): SYNC {} messages, {:.1} MiB; BLOB {} messages, {:.1} MiB\n  \
         z received: SYNC {} messages, {:.1} MiB; BLOB {} messages, {:.1} MiB: {:.2} times what was written\n  \
         z's delays: {} changes, mean {} ms, half within {}, 99% within {}\n  \
         b's blob chunks: {} served, {} of them deflated; {:.1} MiB as {:.1} MiB\n  \
         z's chunks: {} timed out, {} asked of a second source, {} came twice, {} from a source dropped",
        BRIDGE_VALUE / 1024,
        1000 / BRIDGE_EVERY.as_millis(),
        mib(total),
        written_in.as_secs_f64(),
        converged.as_millis(),
        b_sync.0,
        mib(b_sync.1),
        b_blob.0,
        mib(b_blob.1),
        z_sync.0,
        mib(z_sync.1),
        z_blob.0,
        mib(z_blob.1),
        (z_sync.1 + z_blob.1) as f64 / total as f64,
        propagation.count,
        propagation.sum_ms / propagation.count.max(1),
        within(0.5),
        within(0.99),
        served1.chunks_served - served0.chunks_served,
        served1.chunks_deflated - served0.chunks_deflated,
        mib(served1.chunk_bytes - served0.chunk_bytes),
        mib(served1.chunk_bytes_sent - served0.chunk_bytes_sent),
        fetched1.chunks_timed_out - fetched0.chunks_timed_out,
        fetched1.chunks_doubled - fetched0.chunks_doubled,
        fetched1.chunks_twice - fetched0.chunks_twice,
        fetched1.chunks_from_dropped - fetched0.chunks_from_dropped,
    );
}