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}"