tests: the edge writes the direct feed's bytes, and a stalled consumer never lags the loop

parity.rs replays the four committed stage fixtures whose headers the Rust
producer can read (macros, shop, center, bigpad; 400 snapshots each, all of
them with FLY_EDGE_PARITY_ALL=1) into one watch slot served both ways at once,
recorded by a stage, a bridge and a frame-only client per path: headers equal
without wall times, attachments byte-equal and equal to the fixture's, and
the whole messages byte-equal. FLY_EDGE_PARITY_OUT writes the recordings as
.flyfeed files. Also: a header past the envelope limit, and the edge dropping
its clients, unbinding and coming back across a router restart.

stall.rs runs flysim's Pacer at 60 Hz publishing full-size snapshots at 30 Hz
against three stages that stopped reading, a bus subscriber that hoards every
delivery, and no subscriber at all: pacer lag 0, no slow watch send, no
refused publication, a bounded store, and a healthy client that stays current.

services/flysim/crates/fly-edge/tests/common/mod.rs

@@ -0,0 +1,246 @@
+#![allow(dead_code)]
+
+//! Shared pieces of the edge tests: the committed `.flyfeed` fixtures as snapshots, a feed
+//! client, the two serving paths side by side, and a `.flyfeed` writer.
+
+use std::io::Read as _;
+use std::net::SocketAddr;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use std::time::Duration;
+
+use fly_edge::{EdgeConfig, EdgeMetrics};
+use flysim::feed::FeedState;
+use flysim::feedbus;
+use flysim::metrics::Metrics;
+use flysim::snapshot::{AttachmentKind, FeedHeader, Snapshot};
+use futures_util::{SinkExt as _, StreamExt as _};
+use tokio::sync::watch;
+use tokio_tungstenite::tungstenite::Message as WsMessage;
+
+pub fn repo_root() -> PathBuf {
+    PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+        .join("../../../..")
+        .canonicalize()
+        .expect("the repository root is above services/flysim/crates/fly-edge")
+}
+
+/// `u32 LE headerLength | header JSON | (u32 LE length | bytes)*`, split.
+pub fn split(message: &[u8]) -> (&[u8], Vec<&[u8]>) {
+    let read = |at: usize| u32::from_le_bytes(message[at..at + 4].try_into().unwrap()) as usize;
+    let header_len = read(0);
+    let header = &message[4..4 + header_len];
+    let mut at = 4 + header_len;
+    let mut attachments = Vec::new();
+    while at < message.len() {
+        let len = read(at);
+        attachments.push(&message[at + 4..at + 4 + len]);
+        at += 4 + len;
+    }
+    (header, attachments)
+}
+
+/// A wire message back into the snapshot that produced it, or `None` when its header predates
+/// fields the Rust producer always writes (the three oldest fixtures lack `game.scene`).
+pub fn snapshot_of(message: &[u8]) -> Option<Snapshot> {
+    let (header, attachments) = split(message);
+    let header: FeedHeader = serde_json::from_slice(header).ok()?;
+    let mut snapshot = Snapshot {
+        header,
+        frame: Arc::new(Vec::new()),
+        audio: Arc::new(Vec::new()),
+        spikes: Arc::new(Vec::new()),
+    };
+    for (kind, bytes) in snapshot
+        .header
+        .attachments
+        .clone()
+        .into_iter()
+        .zip(attachments)
+    {
+        let bytes = Arc::new(bytes.to_vec());
+        match kind {
+            AttachmentKind::Frame => snapshot.frame = bytes,
+            AttachmentKind::Audio => snapshot.audio = bytes,
+            AttachmentKind::Spikes => snapshot.spikes = bytes,
+        }
+    }
+    Some(snapshot)
+}
+
+/// Every record of `apps/stage/public/fixtures/<name>.flyfeed.gz`, as wire messages.
+pub fn fixture_messages(name: &str) -> Vec<Vec<u8>> {
+    let path = repo_root().join(format!("apps/stage/public/fixtures/{name}.flyfeed.gz"));
+    let gz = std::fs::read(&path).unwrap_or_else(|error| panic!("{}: {error}", path.display()));
+    let mut bytes = Vec::new();
+    flate2::read::GzDecoder::new(gz.as_slice())
+        .read_to_end(&mut bytes)
+        .unwrap();
+    assert_eq!(&bytes[..8], b"FLYFEED\0", "{name}");
+    let read = |at: usize| u32::from_le_bytes(bytes[at..at + 4].try_into().unwrap()) as usize;
+    assert_eq!(read(8), 1, "{name}: container version");
+    let mut at = 16 + read(12);
+    let mut out = Vec::new();
+    while at < bytes.len() {
+        let len = read(at);
+        out.push(bytes[at + 4..at + 4 + len].to_vec());
+        at += 4 + len;
+    }
+    out
+}
+
+/// A `.flyfeed` file of `messages` (`packages/feed/src/fixture.ts`).
+pub fn encode_flyfeed(name: &str, source: &str, messages: &[Vec<u8>]) -> Vec<u8> {
+    let wall = |message: &Vec<u8>| -> u64 {
+        let header: serde_json::Value = serde_json::from_slice(split(message).0).unwrap();
+        header["wallMs"].as_u64().unwrap_or(0)
+    };
+    let duration = match (messages.first(), messages.last()) {
+        (Some(first), Some(last)) => wall(last).saturating_sub(wall(first)),
+        _ => 0,
+    };
+    let manifest = serde_json::json!({
+        "name": name,
+        "protocol": 1,
+        "recordedAt": "1970-01-01T00:00:00.000Z",
+        "source": source,
+        "snapshotCount": messages.len(),
+        "durationMs": duration,
+        "hz": 30,
+        "attachmentPolicy": {
+            "frame": { "stride": 1 },
+            "audio": { "stride": 1 },
+            "spikes": { "stride": 1 }
+        },
+    });
+    let manifest = serde_json::to_vec(&manifest).unwrap();
+    let mut out = b"FLYFEED\0".to_vec();
+    out.extend_from_slice(&1u32.to_le_bytes());
+    out.extend_from_slice(&(manifest.len() as u32).to_le_bytes());
+    out.extend_from_slice(&manifest);
+    for message in messages {
+        out.extend_from_slice(&(message.len() as u32).to_le_bytes());
+        out.extend_from_slice(message);
+    }
+    out
+}
+
+pub fn free_port() -> SocketAddr {
+    std::net::TcpListener::bind("127.0.0.1:0")
+        .unwrap()
+        .local_addr()
+        .unwrap()
+}
+
+pub type Ws =
+    tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>;
+
+/// Connect and say `hello`, retrying while the port is not bound yet (the edge binds only once
+/// its first snapshot has arrived).
+pub async fn connect(addr: SocketAddr, wants: &[&str]) -> Ws {
+    let url = format!("ws://{addr}/feed");
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(20);
+    loop {
+        match tokio_tungstenite::connect_async(&url).await {
+            Ok((mut ws, _)) => {
+                let hello = serde_json::json!({ "protocol": 1, "client": "test", "wants": wants });
+                ws.send(WsMessage::Text(hello.to_string().into()))
+                    .await
+                    .unwrap();
+                return ws;
+            }
+            Err(error) => {
+                assert!(tokio::time::Instant::now() < deadline, "{url}: {error}");
+                tokio::time::sleep(Duration::from_millis(50)).await;
+            }
+        }
+    }
+}
+
+/// The next binary message, within `within`.
+pub async fn next_binary(ws: &mut Ws, within: Duration) -> Vec<u8> {
+    let deadline = tokio::time::Instant::now() + within;
+    loop {
+        let message = tokio::time::timeout_at(deadline, ws.next())
+            .await
+            .expect("a snapshot in time")
+            .expect("the feed stays open")
+            .expect("a well-formed frame");
+        if let WsMessage::Binary(bytes) = message {
+            return bytes.to_vec();
+        }
+    }
+}
+
+pub fn seq_of(message: &[u8]) -> u64 {
+    let header: serde_json::Value = serde_json::from_slice(split(message).0).unwrap();
+    header["seq"].as_u64().unwrap()
+}
+
+/// The same watch slot served both ways at once: flysim's direct server on `direct`, and the
+/// bus (router, publisher, edge) on `edge`. Owns its runtime-side tasks through the handles.
+pub struct Paths {
+    pub snapshots: watch::Sender<Arc<Snapshot>>,
+    pub direct: SocketAddr,
+    pub edge: SocketAddr,
+    pub publisher_metrics: Arc<Metrics>,
+    pub edge_metrics: Arc<EdgeMetrics>,
+    pub bus_dir: tempfile::TempDir,
+    pub bus: feedbus::BusFeed,
+}
+
+/// Idle cadence long enough that no test sees a header repeated for idleness.
+pub const NO_IDLE: Duration = Duration::from_secs(3_600);
+
+/// Start both paths over `first`. `edge` false leaves the edge out (a test then plays its part).
+pub async fn start(first: Snapshot, with_edge: bool) -> Paths {
+    let bus_dir = tempfile::tempdir().unwrap();
+    let (snapshots, receiver) = watch::channel(Arc::new(first));
+    let bus = feedbus::start_router(bus_dir.path()).await.unwrap();
+    let publisher_metrics = Arc::new(Metrics::default());
+    tokio::spawn(feedbus::run_publisher(
+        bus.router.clone(),
+        receiver.clone(),
+        Arc::clone(&publisher_metrics),
+    ));
+
+    let direct = free_port();
+    let listener = tokio::net::TcpListener::bind(direct).await.unwrap();
+    let state = FeedState {
+        snapshots: receiver,
+        metrics: Arc::new(Metrics::default()),
+        idle_period: NO_IDLE,
+    };
+    tokio::spawn(async move { axum::serve(listener, flysim::feed::router(state)).await });
+
+    let edge = free_port();
+    let edge_metrics = Arc::new(EdgeMetrics::default());
+    if with_edge {
+        let config = EdgeConfig {
+            bus_dir: bus_dir.path().to_path_buf(),
+            feed_bind: edge,
+            idle_period: NO_IDLE,
+            metrics_bind: None,
+            retry: Duration::from_millis(50),
+        };
+        tokio::spawn(fly_edge::run(config, Arc::clone(&edge_metrics)));
+    }
+    Paths {
+        snapshots,
+        direct,
+        edge,
+        publisher_metrics,
+        edge_metrics,
+        bus_dir,
+        bus,
+    }
+}
+
+pub fn out_dir() -> Option<PathBuf> {
+    std::env::var_os("FLY_EDGE_PARITY_OUT").map(PathBuf::from)
+}
+
+pub fn write(path: &Path, bytes: &[u8]) {
+    std::fs::create_dir_all(path.parent().unwrap()).unwrap();
+    std::fs::write(path, bytes).unwrap();
+}

services/flysim/crates/fly-edge/tests/parity.rs

@@ -0,0 +1,242 @@
+//! Parity: a feed served through the bus and `fly-edge` is the feed flysim serves directly.
+//!
+//! The committed stage fixtures (`apps/stage/public/fixtures/*.flyfeed.gz`, the recordings the
+//! stage's e2e suite plays) are fed snapshot by snapshot into one watch slot, served both ways at
+//! once, and recorded by one client per path and per `wants` flavour: the stage's (everything),
+//! the bridge's (nothing) and a frame-only one. The two recordings must match: headers equal
+//! apart from wall-time fields and attachments byte-equal -- and in fact the whole messages are
+//! byte-equal, because both are written by `Snapshot::encode` from equal snapshots. The edge's
+//! attachments must also equal the fixture's own.
+//!
+//! `FLY_EDGE_PARITY_OUT=<dir>` also writes each pair of recordings as `.flyfeed` files, which
+//! `packages/feed`'s `decodeFlyfeed` reads. `FLY_EDGE_PARITY_ALL=1` replays whole fixtures
+//! instead of their first 400 snapshots.
+
+mod common;
+
+use std::time::Duration;
+
+use common::*;
+use serde_json::Value;
+
+/// The fixtures whose headers carry every field the Rust producer writes. The three older ones
+/// (`cold-open`, `steady`, `big-moment`) predate `game.scene` and cannot be a Rust `Snapshot`.
+const FIXTURES: [&str; 4] = ["macros", "shop", "center", "bigpad"];
+const WANTS: [(&str, &[&str]); 3] = [
+    ("all", &["frame", "audio", "spikes"]),
+    ("none", &[]),
+    ("frame", &["frame"]),
+];
+
+/// The header with every `wallMs` removed, at any depth.
+fn without_wall_time(header: &[u8]) -> Value {
+    fn strip(value: &mut Value) {
+        match value {
+            Value::Object(map) => {
+                map.remove("wallMs");
+                map.values_mut().for_each(strip);
+            }
+            Value::Array(items) => items.iter_mut().for_each(strip),
+            _ => {}
+        }
+    }
+    let mut value: Value = serde_json::from_slice(header).unwrap();
+    strip(&mut value);
+    value
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn the_edge_writes_the_bytes_the_direct_feed_writes_for_every_committed_fixture() {
+    let limit = if std::env::var_os("FLY_EDGE_PARITY_ALL").is_some() {
+        usize::MAX
+    } else {
+        400
+    };
+    for name in FIXTURES {
+        let snapshots: Vec<_> = fixture_messages(name)
+            .iter()
+            .take(limit)
+            .map(|message| {
+                (
+                    message.clone(),
+                    snapshot_of(message).unwrap_or_else(|| panic!("{name}")),
+                )
+            })
+            .collect();
+        assert!(
+            snapshots.len() >= 100,
+            "{name}: {} snapshots",
+            snapshots.len()
+        );
+
+        let paths = start(snapshots[0].1.clone(), true).await;
+        let mut clients = Vec::new();
+        for (flavour, wants) in WANTS {
+            let direct = connect(paths.direct, wants).await;
+            let edge = connect(paths.edge, wants).await;
+            clients.push((flavour, direct, edge, Vec::new(), Vec::new()));
+        }
+
+        // Lockstep: publish one snapshot, wait until every client has it. Nothing is superseded,
+        // so both recordings are complete and comparable message by message.
+        for (index, (_, snapshot)) in snapshots.iter().enumerate() {
+            if index > 0 {
+                paths
+                    .snapshots
+                    .send_replace(std::sync::Arc::new(snapshot.clone()));
+            }
+            for (_, direct, edge, direct_log, edge_log) in &mut clients {
+                for (ws, log) in [
+                    (&mut *direct, &mut *direct_log),
+                    (&mut *edge, &mut *edge_log),
+                ] {
+                    let message = next_binary(ws, Duration::from_secs(20)).await;
+                    assert_eq!(seq_of(&message), snapshot.header.seq, "{name} #{index}");
+                    log.push(message);
+                }
+            }
+        }
+
+        for (flavour, _, _, direct_log, edge_log) in &clients {
+            assert_eq!(direct_log.len(), snapshots.len());
+            assert_eq!(edge_log.len(), direct_log.len());
+            for (index, (direct, edge)) in direct_log.iter().zip(edge_log).enumerate() {
+                let (direct_header, direct_attachments) = split(direct);
+                let (edge_header, edge_attachments) = split(edge);
+                assert_eq!(
+                    without_wall_time(direct_header),
+                    without_wall_time(edge_header),
+                    "{name}/{flavour} #{index}: headers"
+                );
+                assert_eq!(
+                    direct_attachments, edge_attachments,
+                    "{name}/{flavour} #{index}: attachments"
+                );
+                // The stronger fact: the whole message, wall times included, is the same bytes.
+                assert!(direct == edge, "{name}/{flavour} #{index}: messages differ");
+                if *flavour == "all" {
+                    let (_, fixture_attachments) = split(&snapshots[index].0);
+                    assert_eq!(
+                        edge_attachments, fixture_attachments,
+                        "{name} #{index}: vs the fixture"
+                    );
+                }
+            }
+            if let Some(dir) = out_dir() {
+                write(
+                    &dir.join(format!("{name}-{flavour}-direct.flyfeed")),
+                    &encode_flyfeed(name, "flysim direct", direct_log),
+                );
+                write(
+                    &dir.join(format!("{name}-{flavour}-edge.flyfeed")),
+                    &encode_flyfeed(name, "flysim bus + fly-edge", edge_log),
+                );
+            }
+        }
+        let published = flysim::metrics::Metrics::get(&paths.publisher_metrics.bus_published);
+        assert!(
+            published >= snapshots.len() as u64,
+            "{name}: {published} published"
+        );
+        assert_eq!(
+            flysim::metrics::Metrics::get(&paths.publisher_metrics.bus_publish_failures),
+            0
+        );
+        eprintln!(
+            "{name}: {} snapshots x {} flavours identical on both paths",
+            snapshots.len(),
+            WANTS.len()
+        );
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn a_header_too_large_for_an_envelope_travels_as_an_artifact_and_arrives_intact() {
+    let mut snapshot = snapshot_of(&fixture_messages("macros")[10]).unwrap();
+    // Far past flybus's 65,536-byte envelope: 400 events of 200 characters.
+    for id in 0..400u64 {
+        snapshot.header.events.push(flysim::snapshot::FeedEvent {
+            id: 10_000 + id,
+            wall_ms: 1_757_000_000_000 + id,
+            brain_ms: 5.0,
+            kind: flysim::snapshot::FeedEventKind::System,
+            label: "x".repeat(200),
+            value: None,
+            reward_kind: None,
+            by: None,
+        });
+    }
+    assert!(serde_json::to_vec(&snapshot.header).unwrap().len() > 65_536);
+    let paths = start(snapshot.clone(), true).await;
+    let mut direct = connect(paths.direct, &["frame", "audio", "spikes"]).await;
+    let mut edge = connect(paths.edge, &["frame", "audio", "spikes"]).await;
+    let direct = next_binary(&mut direct, Duration::from_secs(20)).await;
+    let edge = next_binary(&mut edge, Duration::from_secs(20)).await;
+    assert!(direct == edge);
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn the_edge_drops_its_clients_and_unbinds_when_the_bus_goes_away_then_comes_back() {
+    let snapshot = snapshot_of(&fixture_messages("shop")[5]).unwrap();
+    let paths = start(snapshot.clone(), true).await;
+    let mut edge = connect(paths.edge, &[]).await;
+    next_binary(&mut edge, Duration::from_secs(20)).await;
+
+    // flysim stopping is its router stopping.
+    let Paths {
+        snapshots,
+        edge: edge_addr,
+        edge_metrics,
+        bus_dir,
+        bus,
+        ..
+    } = paths;
+    bus.router.shutdown();
+    drop(bus);
+    drop(snapshots);
+    let closed = tokio::time::timeout(Duration::from_secs(10), async {
+        use futures_util::StreamExt as _;
+        loop {
+            match edge.next().await {
+                None | Some(Err(_)) => break,
+                Some(Ok(tokio_tungstenite::tungstenite::Message::Close(_))) => break,
+                Some(Ok(_)) => continue,
+            }
+        }
+    })
+    .await;
+    assert!(closed.is_ok(), "the client was not dropped");
+    let deadline = std::time::Instant::now() + Duration::from_secs(10);
+    while tokio::net::TcpStream::connect(edge_addr).await.is_ok() {
+        assert!(
+            std::time::Instant::now() < deadline,
+            "the feed port stayed bound"
+        );
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    }
+    assert_eq!(
+        edge_metrics
+            .connected
+            .load(std::sync::atomic::Ordering::Relaxed),
+        0
+    );
+
+    // A new flysim on the same directory: the edge finds it and serves again.
+    let (snapshots, receiver) = tokio::sync::watch::channel(std::sync::Arc::new(snapshot.clone()));
+    let bus = flysim::feedbus::start_router(bus_dir.path()).await.unwrap();
+    tokio::spawn(flysim::feedbus::run_publisher(
+        bus.router.clone(),
+        receiver,
+        std::sync::Arc::new(flysim::metrics::Metrics::default()),
+    ));
+    let mut edge = connect(edge_addr, &[]).await;
+    let message = next_binary(&mut edge, Duration::from_secs(20)).await;
+    assert_eq!(seq_of(&message), snapshot.header.seq);
+    assert_eq!(
+        edge_metrics
+            .bus_lost
+            .load(std::sync::atomic::Ordering::Relaxed),
+        1
+    );
+    drop(snapshots);
+}

services/flysim/crates/fly-edge/tests/stall.rs

@@ -0,0 +1,253 @@
+//! A slow or absent edge never slows the loop.
+//!
+//! A thread stands in for the sim loop: flysim's own `Pacer` at realtime speed and 60 Hz Game
+//! Boy frames, publishing full-size snapshots (a real 92,160-byte frame, a 17,407-byte spike
+//! bitset for 139,255 neurons, 12,800 bytes of audio) into the watch slot every second frame,
+//! with `watch::Sender::send`, exactly as `Sim::publish` does. Around it, three kinds of bad
+//! consumer:
+//!
+//! - the edge is up but three of its WebSocket clients never read, so their sockets fill;
+//! - the edge's place on the bus is held by a subscriber that takes deliveries and never
+//!   releases them (the "slow edge");
+//! - nobody is subscribed at all (the "absent edge").
+//!
+//! In every case the pacer reports no lag, no watch send is slow, the publisher keeps
+//! publishing without a refusal, and where there is a healthy client it stays current.
+
+mod common;
+
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::{Duration, Instant};
+
+use common::*;
+use flybus::{Client, ClientConfig, SubscriptionConfig};
+use flysim::feedbus;
+use flysim::metrics::Metrics;
+use flysim::pacing::Pacer;
+use flysim::snapshot::{AttachmentKind, FeedStatus, Snapshot};
+
+/// A full-size running snapshot from a real fixture frame.
+fn full_snapshot() -> Snapshot {
+    let mut snapshot = snapshot_of(&fixture_messages("macros")[3]).unwrap();
+    assert_eq!(snapshot.frame.len(), flysim::snapshot::FRAME_BYTES);
+    snapshot.header.status = FeedStatus::Running;
+    snapshot.header.attachments = AttachmentKind::ALL.to_vec();
+    snapshot.spikes = Arc::new(vec![0b1010_0101; 139_255usize.div_ceil(8)]);
+    snapshot.audio = Arc::new(vec![7; 12_800]);
+    snapshot
+}
+
+struct LoopReport {
+    frames: u64,
+    published: u64,
+    lag_seconds: f64,
+    worst_send: Duration,
+    worst_shortfall: f64,
+    p99_shortfall: f64,
+}
+
+/// Run the stand-in loop for `seconds` on its own thread.
+fn run_loop(
+    snapshots: tokio::sync::watch::Sender<Arc<Snapshot>>,
+    template: Snapshot,
+    seconds: f64,
+) -> LoopReport {
+    std::thread::spawn(move || {
+        let frame_ms = flysim::config::GAMEBOY_MS_PER_FRAME;
+        let mut pacer = Pacer::new(frame_ms, 1.0, Instant::now());
+        let frames = (seconds * 1000.0 / frame_ms) as u64;
+        let mut worst_send = Duration::ZERO;
+        let mut shortfalls = Vec::with_capacity(frames as usize);
+        let mut seq = template.header.seq;
+        let mut published = 0;
+        for frame in 0..frames {
+            if frame % 2 == 0 {
+                let mut snapshot = template.clone();
+                seq += 1;
+                snapshot.header.seq = seq;
+                snapshot.header.frame = frame;
+                let started = Instant::now();
+                let _ = snapshots.send(Arc::new(snapshot));
+                worst_send = worst_send.max(started.elapsed());
+                published += 1;
+            }
+            let now = Instant::now();
+            shortfalls.push(pacer.shortfall_seconds(now));
+            let sleep = pacer.next_sleep(now);
+            if !sleep.is_zero() {
+                std::thread::sleep(sleep);
+            }
+        }
+        shortfalls.sort_by(f64::total_cmp);
+        LoopReport {
+            frames,
+            published,
+            lag_seconds: pacer.lag_seconds(),
+            worst_send,
+            worst_shortfall: *shortfalls.last().unwrap(),
+            p99_shortfall: shortfalls[shortfalls.len() * 99 / 100],
+        }
+    })
+    .join()
+    .unwrap()
+}
+
+fn assert_unharmed(report: &LoopReport, publisher: &Metrics, what: &str) {
+    eprintln!(
+        "{what}: {} frames, {} snapshots, lag {:.3} s, worst send {:?}, shortfall p99 {:.2} ms worst {:.2} ms, bus published {} failed {}",
+        report.frames,
+        report.published,
+        report.lag_seconds,
+        report.worst_send,
+        report.p99_shortfall * 1e3,
+        report.worst_shortfall * 1e3,
+        Metrics::get(&publisher.bus_published),
+        Metrics::get(&publisher.bus_publish_failures),
+    );
+    assert_eq!(report.lag_seconds, 0.0, "{what}: the pacer fell behind");
+    // A watch send is a lock and a swap. Generous for a loaded 4-core box; a send that waited on
+    // a consumer would be one whole stall, seconds.
+    assert!(
+        report.worst_send < Duration::from_millis(20),
+        "{what}: a send took {:?}",
+        report.worst_send
+    );
+    // Sleep overshoot is absorbed by the next frame; staying under one frame at p99 means the
+    // loop kept its absolute deadlines.
+    assert!(
+        report.p99_shortfall < 0.016,
+        "{what}: p99 shortfall {:.2} ms",
+        report.p99_shortfall * 1e3
+    );
+    assert_eq!(
+        Metrics::get(&publisher.bus_publish_failures),
+        0,
+        "{what}: a publication was refused"
+    );
+    // The publisher is allowed to coalesce, never to stop.
+    let published = Metrics::get(&publisher.bus_published);
+    assert!(
+        published * 2 >= report.published,
+        "{what}: only {published} of {} reached the bus",
+        report.published
+    );
+}
+
+const SECONDS: f64 = 6.0;
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn clients_of_the_edge_that_never_read_do_not_lag_the_loop_or_the_healthy_client() {
+    let template = full_snapshot();
+    let paths = start(template.clone(), true).await;
+    // Three stages that said hello and then stopped reading: their sockets fill and stay full.
+    let mut stalled = Vec::new();
+    for _ in 0..3 {
+        stalled.push(connect(paths.edge, &["frame", "audio", "spikes"]).await);
+    }
+    // One healthy stage, read continuously.
+    let mut healthy = connect(paths.edge, &["frame", "audio", "spikes"]).await;
+    let newest = Arc::new(AtomicU64::new(0));
+    let received = Arc::new(AtomicU64::new(0));
+    let reader = {
+        let (newest, received) = (Arc::clone(&newest), Arc::clone(&received));
+        tokio::spawn(async move {
+            loop {
+                let message = next_binary(&mut healthy, Duration::from_secs(30)).await;
+                newest.store(seq_of(&message), Ordering::Relaxed);
+                received.fetch_add(1, Ordering::Relaxed);
+            }
+        })
+    };
+
+    let snapshots = paths.snapshots.clone();
+    let report = tokio::task::spawn_blocking(move || run_loop(snapshots, template, SECONDS))
+        .await
+        .unwrap();
+    assert_unharmed(&report, &paths.publisher_metrics, "stalled clients");
+
+    // The healthy client is current: within a few snapshots of the last one published.
+    let last = paths.snapshots.borrow().header.seq;
+    let deadline = Instant::now() + Duration::from_secs(10);
+    while newest.load(Ordering::Relaxed) < last {
+        assert!(
+            Instant::now() < deadline,
+            "healthy client stuck at {} of {last}",
+            newest.load(Ordering::Relaxed)
+        );
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+    let got = received.load(Ordering::Relaxed);
+    assert!(
+        got * 2 >= report.published,
+        "the healthy client got only {got} of {}",
+        report.published
+    );
+    // The stalled ones are still connected, not dropped for being slow.
+    assert_eq!(paths.edge_metrics.feed.clients(), 4);
+    reader.abort();
+    drop(stalled);
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn a_bus_subscriber_that_never_releases_does_not_lag_the_loop_or_fill_the_store() {
+    let template = full_snapshot();
+    let paths = start(template.clone(), false).await;
+    // The edge's seat, taken by a subscriber that keeps every delivery it gets.
+    let client = Client::connect_unix(
+        feedbus::socket_path(paths.bus_dir.path()),
+        ClientConfig::new(feedbus::EDGE, feedbus::store_root(paths.bus_dir.path())),
+    )
+    .await
+    .unwrap();
+    let mut subscription = client
+        .subscribe(
+            feedbus::TOPIC,
+            SubscriptionConfig::latest().in_flight(2).replay(true),
+        )
+        .await
+        .unwrap();
+    let hoard = tokio::spawn(async move {
+        let mut kept = Vec::new();
+        while let Some(message) = subscription.next().await {
+            kept.push(message);
+        }
+        kept.len()
+    });
+
+    let snapshots = paths.snapshots.clone();
+    let report = tokio::task::spawn_blocking(move || run_loop(snapshots, template, SECONDS))
+        .await
+        .unwrap();
+    assert_unharmed(&report, &paths.publisher_metrics, "hoarding subscriber");
+    let stats = paths.bus.router.stats();
+    eprintln!(
+        "hoarding subscriber: store {} bytes, retained {} bytes",
+        stats.store_bytes, stats.retained_bytes
+    );
+    // Held: two in flight, one queued, one retained, and whatever is mid-seal. Bounded, not
+    // growing with the number published.
+    assert!(
+        stats.store_bytes <= 8 * 122_367,
+        "store holds {} bytes",
+        stats.store_bytes
+    );
+    hoard.abort();
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn an_absent_edge_costs_the_loop_nothing() {
+    let template = full_snapshot();
+    let paths = start(template.clone(), false).await;
+    let snapshots = paths.snapshots.clone();
+    let report = tokio::task::spawn_blocking(move || run_loop(snapshots, template, SECONDS))
+        .await
+        .unwrap();
+    assert_unharmed(&report, &paths.publisher_metrics, "absent edge");
+    let stats = paths.bus.router.stats();
+    assert!(
+        stats.store_bytes <= 3 * 122_367,
+        "store holds {} bytes",
+        stats.store_bytes
+    );
+}