From 4b1559b45a93eaa4d5930ab9f83d43a6f3146588 Mon Sep 17 00:00:00 2001 From: acamilo Date: Wed, 23 Sep 2026 11:59:26 +0000 Subject: [PATCH] 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 | 104 +++++++++++++++--- .../flysim/crates/fly-edge/tests/parity.rs | 59 ++++++++++ .../flysim/crates/fly-edge/tests/stall.rs | 62 ++++++++--- services/flysim/crates/flysim/src/config.rs | 25 +++++ services/flysim/crates/flysim/src/feedbus.rs | 19 +++- 5 files changed, 227 insertions(+), 42 deletions(-) diff --git a/services/flysim/crates/fly-edge/src/lib.rs b/services/flysim/crates/fly-edge/src/lib.rs index 996ad7c..cdf9375 100644 --- a/services/flysim/crates/fly-edge/src/lib.rs +++ b/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) -> 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 { - Ok(()) => { + let end = session(&config, &metrics).await; + match &end { + SessionEnd::BusLost => { tracing::warn!("the feed bus went away; clients dropped, reconnecting"); - quiet = false; } - Err(error) => { - // One line per outage, not one per retry. - if !quiet { - tracing::info!(error = format!("{error:#}"), "waiting for the feed bus"); - quiet = true; - } + SessionEnd::Unreachable(error) if last != Some(end.kind()) => { + tracing::info!(error = format!("{error:#}"), "waiting for the feed bus"); } + 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>) -> impl IntoResponse { ( [( @@ -186,9 +225,31 @@ async fn healthz(State(metrics): State>) -> impl IntoResponse { } } -/// One subscription's lifetime. `Err` before serving began (nothing to reach yet); `Ok` once a -/// session that did serve has ended because the bus went away. -async fn session(config: &EdgeConfig, metrics: &EdgeMetrics) -> Result<()> { +/// One subscription's lifetime. +async fn session(config: &EdgeConfig, metrics: &EdgeMetrics) -> SessionEnd { + 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(()) } diff --git a/services/flysim/crates/fly-edge/tests/parity.rs b/services/flysim/crates/fly-edge/tests/parity.rs index d983d97..a041500 100644 --- a/services/flysim/crates/fly-edge/tests/parity.rs +++ b/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); +} diff --git a/services/flysim/crates/fly-edge/tests/stall.rs b/services/flysim/crates/fly-edge/tests/stall.rs index c095f62..50176f0 100644 --- a/services/flysim/crates/fly-edge/tests/stall.rs +++ b/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 -//! releases them (the "slow edge"); +//! - the edge's place on the bus is held by a client that opens every subscription it may +//! (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,20 +228,43 @@ async fn hoarding_subscriber() -> Outcome { ) .await .unwrap(); - let mut subscription = client - .subscribe( - feedbus::TOPIC, - SubscriptionConfig::latest().in_flight(2).replay(true), + // 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, + SubscriptionConfig::latest().in_flight(2).replay(true), + ) + .await + .unwrap(); + 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 - .unwrap(); - let hoard = tokio::spawn(async move { - let mut kept = Vec::new(); - while let Some(message) = subscription.next().await { - kept.push(message); - } - kept.len() - }); + .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 - // growing with the number published. + // Held: per subscription two in flight and one queued, plus one retained and whatever is + // 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 ); - hoard.abort(); + for hoard in hoards { + hoard.abort(); + } Outcome { report, publisher: Arc::clone(&paths.publisher_metrics), diff --git a/services/flysim/crates/flysim/src/config.rs b/services/flysim/crates/flysim/src/config.rs index e3ff354..3580217 100644 --- a/services/flysim/crates/flysim/src/config.rs +++ b/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::("[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"); diff --git a/services/flysim/crates/flysim/src/feedbus.rs b/services/flysim/crates/flysim/src/feedbus.rs index bd97644..bf0cbf0 100644 --- a/services/flysim/crates/flysim/src/feedbus.rs +++ b/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`] -/// 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 -/// catch-up audio after a stall still fits, and it is RAM (tmpfs), so it is kept small on purpose. +/// holds one snapshot of staging plus the sealed copy while it seals. +/// +/// Only one client can subscribe at all: the publisher is in process, and the one socket is +/// 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; - let pinned = subscribers * (1 + limits.max_latest_in_flight) + 1 + 2; + // One subscribing client (the socket's), every subscription it may open, none consuming. + 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}"