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::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, 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);
/// Only the scenarios whose names contain this, if set: `BENCH_SCENARIO=NAT`.
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,
}

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()
    });
}

const SCENARIOS: &[Scenario] = &[
    Scenario {
        name: "full mesh, 13 nodes",
        nodes: 13,
        on: sim_only,
        net: clean,
        writer: 1,
        away: None,
        joining: 0,
    },
    Scenario {
        name: "writer behind NAT (one link), 13 nodes",
        nodes: 13,
        on: sim_only,
        net: writer_behind_nat,
        writer: 1,
        away: None,
        joining: 0,
    },
    Scenario {
        name: "two transports, 3 bridges, 30 nodes",
        nodes: 30,
        on: two_transports,
        net: clean,
        writer: 5,
        away: None,
        joining: 0,
    },
    Scenario {
        name: "partial mesh, 50 nodes",
        nodes: 50,
        on: sim_only,
        net: clean,
        writer: 1,
        away: None,
        joining: 0,
    },
    Scenario {
        name: "partial mesh, lossy and slow links, 50 nodes",
        nodes: 50,
        on: sim_only,
        net: lossy,
        writer: 1,
        away: None,
        joining: 0,
    },
    Scenario {
        name: "partial mesh, 100 nodes",
        nodes: 100,
        on: sim_only,
        net: clean,
        writer: 1,
        away: None,
        joining: 0,
    },
    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,
    },
    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,
    },
];

#[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 chosen: Vec<(u64, &'static Scenario)> = (70..)
        .zip(SCENARIOS)
        .filter(|(_, scenario)| scenario.name.contains(&only))
        .collect();
    if std::env::var_os(PARALLEL).is_some() {
        // 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)>,
}

/// 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(),
        };
        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.delivered += stats.relay.delivered;
            relay.delivered_bytes += stats.relay.delivered_bytes;
            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.carrying_bloom_bytes += stats.relay.carrying_bloom_bytes;
            relay.nothing_new_change_bytes += stats.relay.nothing_new_change_bytes;
            relay.nothing_new_bloom_bytes += stats.relay.nothing_new_bloom_bytes;
            relay.off_tree += stats.relay.off_tree;
            relay.off_tree_ms += stats.relay.off_tree_ms;
            for (_, peer_state) in node.peers_snapshot().await {
                totals.incidents.add(&peer_state.lock().await.incidents);
            }
            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; 9] = [
    "SYNC",
    "STATUS",
    "STATUS_REPLY",
    "BLOB",
    "HELLO",
    "HELLO_REPLY",
    "PING/PONG",
    "WHO/ME",
    "LINKS",
];

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

fn counters_mut(by: &mut ByKind) -> [&mut Counter; 9] {
    [
        &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,
    ]
}

/// 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 timing = |config: &mut crate::node::NodeConfig| {
        config.timing = Timing::new(KEEPALIVE, SYNC_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;
    // 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()
    );
    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,
        );
    }
    out
}

/// 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 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)",
        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,
    );
    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 and {:.0} of bloom filters; of those with nothing new, {:.0} and {:.0}",
        carrying as f64 / writes,
        each(
            r1.carrying_change_bytes - r0.carrying_change_bytes,
            carrying
        ),
        each(r1.carrying_bloom_bytes - r0.carrying_bloom_bytes, carrying),
        each(
            r1.nothing_new_change_bytes - r0.nothing_new_change_bytes,
            twice
        ),
        each(
            r1.nothing_new_bloom_bytes - r0.nothing_new_bloom_bytes,
            twice
        ),
    );
    let (i0, i1) = (during.0.incidents, during.1.incidents);
    let _ = writeln!(
        out,
        "    incidents: {} reply timeouts ({:.2} per write), {} heals, {} dormancies",
        i1.reply_timeouts - i0.reply_timeouts,
        (i1.reply_timeouts - i0.reply_timeouts) as f64 / writes,
        i1.heals - i0.heals,
        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 { "" }
        );
    }
}

/// 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 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 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 {}",
        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),
    );
}