tests: stall.rs gates only the pacing claim; the rates move to an ignored perf test

Review round 1, B1. The p99 sleep-overshoot bound measured the OS scheduler
and the published*2 bound failed whenever a starved debug publisher
coalesced, which it is designed to do, so the workspace gate went red on a
loaded box. The three gated tests keep what the slice claims: pacer lag 0,
no watch send held by a consumer (50 ms bound), no refused publication,
something reaches the bus, a bounded store, and a healthy client that reaches
the newest snapshot. The overshoot and throughput bounds are in
the_three_scenarios_keep_their_rates, #[ignore]d.

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

@@ -11,8 +11,16 @@
 //!   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
+//! The gated tests assert the claim, and only the claim: the pacer reports no lag, no watch send
-//! publishing without a refusal, and where there is a healthy client it stays current.
+//! waits on a consumer, no publication is refused, the store stays bounded, and a healthy client
+//! still reaches the newest snapshot. Those hold on a box at any load, because none of them is a
+//! rate.
+//!
+//! How fast the loop's sleeps come back and how many snapshots a debug-build publisher gets
+//! through measure the OS scheduler and the CPU left over, not the bus: a starved publisher
+//! coalesces by design. Those bounds are in `the_three_scenarios_keep_their_rates`, which is
+//! `#[ignore]`d; run it on a quiet box with
+//! `cargo test --release -p fly-edge --test stall -- --ignored --nocapture`.
 
 mod common;
 
@@ -93,7 +101,7 @@ fn run_loop(
     .unwrap()
 }
 
-fn assert_unharmed(report: &LoopReport, publisher: &Metrics, what: &str) {
+fn print_report(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,
@@ -105,27 +113,40 @@ fn assert_unharmed(report: &LoopReport, publisher: &Metrics, what: &str) {
         Metrics::get(&publisher.bus_published),
         Metrics::get(&publisher.bus_publish_failures),
     );
+}
+
+/// The claim: the loop is never held by the bus, whatever the load.
+fn assert_unharmed(report: &LoopReport, publisher: &Metrics, what: &str) {
     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 watch send is a lock and a swap. A send that waited on a consumer would be a whole
-    // a consumer would be one whole stall, seconds.
+    // stall, seconds; 50 ms leaves room for a preempted thread on a loaded box.
     assert!(
-        report.worst_send < Duration::from_millis(20),
+        report.worst_send < Duration::from_millis(50),
         "{what}: a send took {:?}",
         report.worst_send
     );
-    // Sleep overshoot is absorbed by the next frame; staying under one frame at p99 means the
+    // A slow or absent consumer is never a reason to refuse a latest publication.
-    // 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.
+    // Coalescing is allowed, stopping is not.
+    assert!(
+        Metrics::get(&publisher.bus_published) >= 1,
+        "{what}: nothing reached the bus"
+    );
+}
+
+/// Rates: meaningful only on a quiet box (see the module comment).
+fn assert_rates(report: &LoopReport, publisher: &Metrics, what: &str) {
+    // Sleep overshoot is absorbed by the next frame; 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
+    );
     let published = Metrics::get(&publisher.bus_published);
     assert!(
         published * 2 >= report.published,
@@ -136,8 +157,15 @@ fn assert_unharmed(report: &LoopReport, publisher: &Metrics, what: &str) {
 
 const SECONDS: f64 = 6.0;
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+/// What a scenario leaves for the rate checks.
-async fn clients_of_the_edge_that_never_read_do_not_lag_the_loop_or_the_healthy_client() {
+struct Outcome {
+    report: LoopReport,
+    publisher: Arc<Metrics>,
+    /// Snapshots the healthy client received, where there is one.
+    healthy_received: Option<u64>,
+}
+
+async fn stalled_clients() -> Outcome {
     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.
@@ -153,7 +181,7 @@ async fn clients_of_the_edge_that_never_read_do_not_lag_the_loop_or_the_healthy_
         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;
+                let message = next_binary(&mut healthy, Duration::from_secs(120)).await;
                 newest.store(seq_of(&message), Ordering::Relaxed);
                 received.fetch_add(1, Ordering::Relaxed);
             }
@@ -164,11 +192,13 @@ async fn clients_of_the_edge_that_never_read_do_not_lag_the_loop_or_the_healthy_
     let report = tokio::task::spawn_blocking(move || run_loop(snapshots, template, SECONDS))
         .await
         .unwrap();
+    print_report(&report, &paths.publisher_metrics, "stalled clients");
     assert_unharmed(&report, &paths.publisher_metrics, "stalled clients");
 
-    // The healthy client is current: within a few snapshots of the last one published.
+    // The healthy client reaches the last snapshot published: the newest one always gets
+    // through, however many in between were coalesced.
     let last = paths.snapshots.borrow().header.seq;
-    let deadline = Instant::now() + Duration::from_secs(10);
+    let deadline = Instant::now() + Duration::from_secs(60);
     while newest.load(Ordering::Relaxed) < last {
         assert!(
             Instant::now() < deadline,
@@ -177,20 +207,18 @@ async fn clients_of_the_edge_that_never_read_do_not_lag_the_loop_or_the_healthy_
         );
         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);
+    Outcome {
+        report,
+        publisher: Arc::clone(&paths.publisher_metrics),
+        healthy_received: Some(received.load(Ordering::Relaxed)),
+    }
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn hoarding_subscriber() -> Outcome {
-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.
@@ -219,6 +247,7 @@ async fn a_bus_subscriber_that_never_releases_does_not_lag_the_loop_or_fill_the_
     let report = tokio::task::spawn_blocking(move || run_loop(snapshots, template, SECONDS))
         .await
         .unwrap();
+    print_report(&report, &paths.publisher_metrics, "hoarding subscriber");
     assert_unharmed(&report, &paths.publisher_metrics, "hoarding subscriber");
     let stats = paths.bus.router.stats();
     eprintln!(
@@ -233,16 +262,21 @@ async fn a_bus_subscriber_that_never_releases_does_not_lag_the_loop_or_fill_the_
         stats.store_bytes
     );
     hoard.abort();
+    Outcome {
+        report,
+        publisher: Arc::clone(&paths.publisher_metrics),
+        healthy_received: None,
+    }
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn absent_edge() -> Outcome {
-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();
+    print_report(&report, &paths.publisher_metrics, "absent edge");
     assert_unharmed(&report, &paths.publisher_metrics, "absent edge");
     let stats = paths.bus.router.stats();
     assert!(
@@ -250,4 +284,45 @@ async fn an_absent_edge_costs_the_loop_nothing() {
         "store holds {} bytes",
         stats.store_bytes
     );
+    Outcome {
+        report,
+        publisher: Arc::clone(&paths.publisher_metrics),
+        healthy_received: None,
+    }
+}
+
+#[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() {
+    stalled_clients().await;
+}
+
+#[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() {
+    hoarding_subscriber().await;
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn an_absent_edge_costs_the_loop_nothing() {
+    absent_edge().await;
+}
+
+/// The same three scenarios, plus the rates. A measurement of the box as much as of the bus,
+/// so not part of the workspace gate.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[ignore = "timing: needs a quiet box; run with --release -- --ignored"]
+async fn the_three_scenarios_keep_their_rates() {
+    for (what, outcome) in [
+        ("stalled clients", stalled_clients().await),
+        ("hoarding subscriber", hoarding_subscriber().await),
+        ("absent edge", absent_edge().await),
+    ] {
+        assert_rates(&outcome.report, &outcome.publisher, what);
+        if let Some(got) = outcome.healthy_received {
+            assert!(
+                got * 2 >= outcome.report.published,
+                "{what}: the healthy client got only {got} of {}",
+                outcome.report.published
+            );
+        }
+    }
 }