fly-edge, flysim: a taken feed port is named as such; FLY_BUS_DIR must be absolute; worst case restated

Review round 1. A bind failure after subscribing was logged once as
"waiting for the feed bus". session() now ends as Unreachable, BindFailed or
BusLost, each logged as what it is (once per streak), and
fly_edge_bind_failures_total counts the second; a new parity test holds the
port, sees the edge fail to bind without claiming to serve, frees it and gets
served.

feed.bus_dir (FLY_BUS_DIR) must be absolute and non-empty, checked in either
mode, since flysim and the edge each resolve it and a relative path would
let them disagree.

The sizing worst case was 7 stuck subscribers; that seat does not exist,
since edge.sock admits one client. It is that client with all 4
subscriptions it may open: 4*3+1+2 = 15 snapshots, about 1.8 MB. feedbus's
comment and unit test say so, and stall.rs's hoarding scenario now takes all
four subscriptions, checks a fifth and a second connection are refused, and
bounds the store at 15 snapshots.

services/flysim/crates/fly-edge/src/lib.rs

@@ -73,6 +73,8 @@ pub struct EdgeMetrics {
     pub bus_lost: AtomicU64,
     /// Publications that could not be turned back into a snapshot.
     pub decode_failures: AtomicU64,
+    /// Sessions that reached the bus but could not bind the feed port.
+    pub bind_failures: AtomicU64,
 }
 
 impl EdgeMetrics {
@@ -128,6 +130,13 @@ impl EdgeMetrics {
             "Feed bus publications that did not decode to a snapshot.",
             self.decode_failures.load(Ordering::Relaxed),
         );
+        metric(
+            &mut out,
+            "fly_edge_bind_failures_total",
+            "counter",
+            "Times the bus was reachable but the feed port could not be bound.",
+            self.bind_failures.load(Ordering::Relaxed),
+        );
         out
     }
 }
@@ -149,25 +158,55 @@ pub async fn run(config: EdgeConfig, metrics: Arc<EdgeMetrics>) -> Result<()> {
             }
         });
     }
-    let mut quiet = false;
+    // One line per outage of each kind, not one per retry.
+    let mut last: Option<&'static str> = None;
     loop {
-        match session(&config, &metrics).await {
+        let end = session(&config, &metrics).await;
-            Ok(()) => {
+        match &end {
+            SessionEnd::BusLost => {
                 tracing::warn!("the feed bus went away; clients dropped, reconnecting");
-                quiet = false;
             }
-            Err(error) => {
+            SessionEnd::Unreachable(error) if last != Some(end.kind()) => {
-                // One line per outage, not one per retry.
-                if !quiet {
                 tracing::info!(error = format!("{error:#}"), "waiting for the feed bus");
-                    quiet = true;
             }
+            SessionEnd::BindFailed(error) if last != Some(end.kind()) => {
+                // The bus is fine; the port is not ours. Most likely flysim is still in direct
+                // mode and holds it (FLY_FEED_VIA is not bus), or another process does.
+                tracing::warn!(
+                    error = format!("{error:#}"),
+                    "the bus is up but the feed port cannot be bound; retrying"
+                );
             }
+            _ => {}
         }
+        last = match end {
+            SessionEnd::BusLost => None,
+            other => Some(other.kind()),
+        };
         tokio::time::sleep(config.retry).await;
     }
 }
 
+/// Why a [`session`] ended.
+enum SessionEnd {
+    /// No router answered, or it closed before the first snapshot. Nothing was served.
+    Unreachable(anyhow::Error),
+    /// Subscribed and holding a snapshot, but the feed port could not be bound.
+    BindFailed(anyhow::Error),
+    /// A session that served has ended because the bus went away.
+    BusLost,
+}
+
+impl SessionEnd {
+    fn kind(&self) -> &'static str {
+        match self {
+            Self::Unreachable(_) => "unreachable",
+            Self::BindFailed(_) => "bind",
+            Self::BusLost => "lost",
+        }
+    }
+}
+
 async fn prometheus(State(metrics): State<Arc<EdgeMetrics>>) -> impl IntoResponse {
     (
         [(
@@ -186,9 +225,31 @@ async fn healthz(State(metrics): State<Arc<EdgeMetrics>>) -> impl IntoResponse {
     }
 }
 
-/// One subscription's lifetime. `Err` before serving began (nothing to reach yet); `Ok` once a
+/// One subscription's lifetime.
-/// session that did serve has ended because the bus went away.
+async fn session(config: &EdgeConfig, metrics: &EdgeMetrics) -> SessionEnd {
-async fn session(config: &EdgeConfig, metrics: &EdgeMetrics) -> Result<()> {
+    let (subscription, client, first) = match subscribe(config, metrics).await {
+        Ok(subscribed) => subscribed,
+        Err(error) => return SessionEnd::Unreachable(error),
+    };
+    let listener = match tokio::net::TcpListener::bind(config.feed_bind).await {
+        Ok(listener) => listener,
+        Err(error) => {
+            metrics.bind_failures.fetch_add(1, Ordering::Relaxed);
+            return SessionEnd::BindFailed(
+                anyhow::Error::new(error)
+                    .context(format!("binding the feed listener on {}", config.feed_bind)),
+            );
+        }
+    };
+    serve(config, metrics, client, subscription, first, listener).await;
+    SessionEnd::BusLost
+}
+
+/// Connect, subscribe and wait for the first snapshot that decodes.
+async fn subscribe(
+    config: &EdgeConfig,
+    metrics: &EdgeMetrics,
+) -> Result<(flybus::Subscription, Client, flysim::snapshot::Snapshot)> {
     let client = Client::connect_unix(
         feedbus::socket_path(&config.bus_dir),
         ClientConfig::new(feedbus::EDGE, feedbus::store_root(&config.bus_dir)),
@@ -217,12 +278,20 @@ async fn session(config: &EdgeConfig, metrics: &EdgeMetrics) -> Result<()> {
             }
         }
     };
+    Ok((subscription, client, first))
+}
+
+/// Serve `listener` from `subscription` until the bus goes away.
+async fn serve(
+    config: &EdgeConfig,
+    metrics: &EdgeMetrics,
+    client: Client,
+    mut subscription: flybus::Subscription,
+    first: flysim::snapshot::Snapshot,
+    listener: tokio::net::TcpListener,
+) {
     let (snapshots, receiver) = watch::channel(Arc::new(first));
     metrics.snapshots.fetch_add(1, Ordering::Relaxed);
-
-    let listener = tokio::net::TcpListener::bind(config.feed_bind)
-        .await
-        .with_context(|| format!("binding the feed listener on {}", config.feed_bind))?;
     tracing::info!(feed = %config.feed_bind, bus = %config.bus_dir.display(), "serving the feed from the bus");
     metrics.connected.store(1, Ordering::Relaxed);
 
@@ -272,5 +341,4 @@ async fn session(config: &EdgeConfig, metrics: &EdgeMetrics) -> Result<()> {
     {
         tracing::warn!("the feed listener took more than 5 s to stop");
     }
-    Ok(())
 }

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

@@ -240,3 +240,62 @@ async fn the_edge_drops_its_clients_and_unbinds_when_the_bus_goes_away_then_come
     );
     drop(snapshots);
 }
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn an_edge_whose_port_is_taken_keeps_retrying_and_serves_once_it_is_free() {
+    let snapshot = snapshot_of(&fixture_messages("center")[7]).unwrap();
+    // Someone else (flysim still in direct mode, say) holds the feed port before the edge starts.
+    let squatter = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+    let port = squatter.local_addr().unwrap();
+    let bus_dir = tempfile::tempdir().unwrap();
+    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 metrics = std::sync::Arc::new(fly_edge::EdgeMetrics::default());
+    tokio::spawn(fly_edge::run(
+        fly_edge::EdgeConfig {
+            bus_dir: bus_dir.path().to_path_buf(),
+            feed_bind: port,
+            idle_period: NO_IDLE,
+            metrics_bind: None,
+            retry: Duration::from_millis(50),
+        },
+        std::sync::Arc::clone(&metrics),
+    ));
+    // It reaches the bus, fails to bind, and says so rather than claiming to wait for the bus.
+    let deadline = std::time::Instant::now() + Duration::from_secs(10);
+    while metrics
+        .bind_failures
+        .load(std::sync::atomic::Ordering::Relaxed)
+        < 3
+    {
+        assert!(
+            std::time::Instant::now() < deadline,
+            "the edge never reached the bus"
+        );
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+    assert_eq!(
+        metrics.connected.load(std::sync::atomic::Ordering::Relaxed),
+        0
+    );
+    assert_eq!(
+        metrics.bus_lost.load(std::sync::atomic::Ordering::Relaxed),
+        0
+    );
+
+    drop(squatter);
+    let mut edge = connect(port, &[]).await;
+    let message = next_binary(&mut edge, Duration::from_secs(20)).await;
+    assert_eq!(seq_of(&message), snapshot.header.seq);
+    assert_eq!(
+        metrics.connected.load(std::sync::atomic::Ordering::Relaxed),
+        1
+    );
+    drop(snapshots);
+    drop(bus);
+}

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

@@ -7,8 +7,8 @@
 //! 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
+//! - the edge's place on the bus is held by a client that opens every subscription it may
-//!   releases them (the "slow edge");
+//!   (4) and never releases a delivery on any of them (the "slow edge", at its worst);
 //! - nobody is subscribed at all (the "absent edge").
 //!
 //! The gated tests assert the claim, and only the claim: the pacer reports no lag, no watch send
@@ -228,6 +228,11 @@ async fn hoarding_subscriber() -> Outcome {
     )
     .await
     .unwrap();
+    // Every subscription the seat may open, each keeping every delivery at the in-flight cap:
+    // the worst case the store has to hold (`feedbus::limits`, flybus.md "Feed sizing").
+    let seats = feedbus::limits().max_subscriptions_per_client;
+    let mut hoards = Vec::new();
+    for _ in 0..seats {
         let mut subscription = client
             .subscribe(
                 feedbus::TOPIC,
@@ -235,13 +240,31 @@ async fn hoarding_subscriber() -> Outcome {
             )
             .await
             .unwrap();
-    let hoard = tokio::spawn(async move {
+        hoards.push(tokio::spawn(async move {
             let mut kept = Vec::new();
             while let Some(message) = subscription.next().await {
                 kept.push(message);
             }
             kept.len()
-    });
+        }));
+    }
+    assert!(
+        client
+            .subscribe(feedbus::TOPIC, SubscriptionConfig::latest())
+            .await
+            .is_err(),
+        "a subscription past max_subscriptions_per_client was admitted"
+    );
+    // And the seat is the only one: a second connection as the edge is refused.
+    assert!(
+        Client::connect_unix(
+            feedbus::socket_path(paths.bus_dir.path()),
+            ClientConfig::new(feedbus::EDGE, feedbus::store_root(paths.bus_dir.path())),
+        )
+        .await
+        .is_err(),
+        "a second client was admitted on edge.sock"
+    );
 
     let snapshots = paths.snapshots.clone();
     let report = tokio::task::spawn_blocking(move || run_loop(snapshots, template, SECONDS))
@@ -254,14 +277,17 @@ async fn hoarding_subscriber() -> Outcome {
         "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
+    // Held: per subscription two in flight and one queued, plus one retained and whatever is
-    // growing with the number published.
+    // mid-seal: 4 * 3 + 1 + 2 = 15 snapshots at most. Bounded, not growing with the number
+    // published.
     assert!(
-        stats.store_bytes <= 8 * 122_367,
+        stats.store_bytes <= 15 * 122_367,
         "store holds {} bytes",
         stats.store_bytes
     );
+    for hoard in hoards {
         hoard.abort();
+    }
     Outcome {
         report,
         publisher: Arc::clone(&paths.publisher_metrics),

services/flysim/crates/flysim/src/config.rs

@@ -459,6 +459,16 @@ impl Config {
         if self.control.sugar_per_minute == 0 {
             bail!("control.sugar_per_minute must be at least 1");
         }
+        // The router's socket and store, and the edge's way to them. A relative path would
+        // resolve against whichever working directory each process happens to have, so the two
+        // could silently disagree; an empty one is a typo. Checked in either mode, so a bad
+        // value is found before the day a box is switched to the bus.
+        if self.feed.bus_dir.as_os_str().is_empty() || !self.feed.bus_dir.is_absolute() {
+            bail!(
+                "feed.bus_dir (FLY_BUS_DIR) must be an absolute path, got {:?}",
+                self.feed.bus_dir
+            );
+        }
         if self.feed.bind == self.control.bind {
             bail!("feed.bind and control.bind must differ (7400 and 7401)");
         }
@@ -761,6 +771,21 @@ mod tests {
         assert_eq!(toml::from_str::<Config>("[feed]\nvia = \"bus\"\n").unwrap().feed.via, FeedVia::Bus);
     }
 
+    #[test]
+    fn the_bus_dir_must_be_absolute_and_not_empty() {
+        Config::default().validate().unwrap();
+        for bad in ["", "run/fly/bus", "./bus"] {
+            let mut config = Config::default();
+            config.feed.bus_dir = PathBuf::from(bad);
+            let error = config.validate().unwrap_err();
+            assert!(error.to_string().contains("FLY_BUS_DIR"), "{bad:?}: {error}");
+        }
+        // Through the environment too.
+        let mut config = Config::default();
+        config.apply_env(&env(&[("FLY_BUS_DIR", "relative/bus")])).unwrap();
+        assert!(config.validate().is_err());
+    }
+
     #[test]
     fn the_example_file_parses_and_validates() {
         let path = concat!(env!("CARGO_MANIFEST_DIR"), "/../../flysim.toml.example");

services/flysim/crates/flysim/src/feedbus.rs

@@ -66,10 +66,15 @@ pub fn store_root(bus_dir: &Path) -> PathBuf {
 /// frame, a 17,407-byte spike bitset (139,255 neurons) and about 12,800 bytes of audio (1,600
 /// stereo f32 frames at 48 kHz per 30 Hz snapshot). A `latest` subscriber pins at most its one
 /// queued slot plus its in-flight credits, the topic pins one retained value, and the publisher
-/// holds one snapshot of staging plus the sealed copy while it seals. With [`Limits::max_clients`]
+/// holds one snapshot of staging plus the sealed copy while it seals.
-/// at 8 and in-flight credits capped at 2, the worst case is 7 subscribers that never consume:
+///
-/// `7 * 3 + 1 + 2 = 24` snapshots, about 3 MB. The store cap is ten times that so a burst of
+/// Only one client can subscribe at all: the publisher is in process, and the one socket is
-/// catch-up audio after a stall still fits, and it is RAM (tmpfs), so it is kept small on purpose.
+/// launcher-bound to [`EDGE`], which the router admits once at a time. So the worst case is
+/// that client holding every subscription it may open ([`Limits::max_subscriptions_per_client`],
+/// 4), each never consuming with in-flight credits at the cap of 2: `4 * 3 + 1 + 2 = 15`
+/// snapshots, about 1.8 MB. `max_clients` bounds connections, pending handshakes included, not
+/// subscribers. The store cap is well over ten times that so a burst of catch-up audio after a
+/// stall still fits, and it is RAM (tmpfs), so it is kept small on purpose.
 pub fn limits() -> Limits {
     Limits {
         max_clients: 8,
@@ -336,8 +341,10 @@ mod tests {
         // A full snapshot on the live fly (see `limits`).
         let snapshot_bytes = crate::snapshot::FRAME_BYTES + 139_255usize.div_ceil(8) + 12_800;
         assert_eq!(snapshot_bytes, 122_367);
-        let subscribers = limits.max_clients as u64 - 1;
+        // One subscribing client (the socket's), every subscription it may open, none consuming.
-        let pinned = subscribers * (1 + limits.max_latest_in_flight) + 1 + 2;
+        let subscriptions = limits.max_subscriptions_per_client as u64;
+        let pinned = subscriptions * (1 + limits.max_latest_in_flight) + 1 + 2;
+        assert_eq!(pinned, 15);
         assert!(
             pinned * snapshot_bytes as u64 * 10 <= limits.max_store_bytes,
             "{pinned}"