From 34c7a56b256c47d6e3bf58ee816b44bd596ea921 Mon Sep 17 00:00:00 2001 From: acamilo Date: Wed, 23 Sep 2026 08:17:13 +0000 Subject: [PATCH] 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. --- .../crates/fly-edge/tests/common/mod.rs | 246 +++++++++++++++++ .../flysim/crates/fly-edge/tests/parity.rs | 242 +++++++++++++++++ .../flysim/crates/fly-edge/tests/stall.rs | 253 ++++++++++++++++++ 3 files changed, 741 insertions(+) create mode 100644 services/flysim/crates/fly-edge/tests/common/mod.rs create mode 100644 services/flysim/crates/fly-edge/tests/parity.rs create mode 100644 services/flysim/crates/fly-edge/tests/stall.rs diff --git a/services/flysim/crates/fly-edge/tests/common/mod.rs b/services/flysim/crates/fly-edge/tests/common/mod.rs new file mode 100644 index 0000000..e35c3ed --- /dev/null +++ b/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 { + 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/.flyfeed.gz`, as wire messages. +pub fn fixture_messages(name: &str) -> Vec> { + 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]) -> Vec { + let wall = |message: &Vec| -> 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>; + +/// 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 { + 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>, + pub direct: SocketAddr, + pub edge: SocketAddr, + pub publisher_metrics: Arc, + pub edge_metrics: Arc, + 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 { + 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(); +} diff --git a/services/flysim/crates/fly-edge/tests/parity.rs b/services/flysim/crates/fly-edge/tests/parity.rs new file mode 100644 index 0000000..d983d97 --- /dev/null +++ b/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=` 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); +} diff --git a/services/flysim/crates/fly-edge/tests/stall.rs b/services/flysim/crates/fly-edge/tests/stall.rs new file mode 100644 index 0000000..714d800 --- /dev/null +++ b/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>, + 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 + ); +}