diff --git a/services/flysim/Cargo.lock b/services/flysim/Cargo.lock index 7f84d26..322d886 100644 --- a/services/flysim/Cargo.lock +++ b/services/flysim/Cargo.lock @@ -416,6 +416,25 @@ dependencies = [ "serde", ] +[[package]] +name = "fly-edge" +version = "0.1.1" +dependencies = [ + "anyhow", + "axum", + "clap", + "flate2", + "flybus", + "flysim", + "futures-util", + "serde_json", + "tempfile", + "tokio", + "tokio-tungstenite", + "tracing", + "tracing-subscriber", +] + [[package]] name = "fly-session" version = "0.1.1" diff --git a/services/flysim/Cargo.toml b/services/flysim/Cargo.toml index dec6a38..2def2d3 100644 --- a/services/flysim/Cargo.toml +++ b/services/flysim/Cargo.toml @@ -1,6 +1,7 @@ [workspace] resolver = "3" members = [ + "crates/fly-edge", "crates/fly-session", "crates/fly-session-types", "crates/flybrain-core", diff --git a/services/flysim/crates/fly-edge/Cargo.toml b/services/flysim/crates/fly-edge/Cargo.toml new file mode 100644 index 0000000..90ba894 --- /dev/null +++ b/services/flysim/crates/fly-edge/Cargo.toml @@ -0,0 +1,36 @@ +[package] +name = "fly-edge" +version.workspace = true +edition = "2024" +rust-version.workspace = true +license.workspace = true +publish = false +description = "The feed WebSocket (:7400) served from flysim's feed bus (FLY_FEED_VIA=bus)." + +[lib] +name = "fly_edge" +path = "src/lib.rs" + +[[bin]] +name = "fly-edge" +path = "src/main.rs" + +[dependencies] +# The feed server and the bus encoding are flysim's own modules (`feed`, `feedbus`, +# `snapshot`), so the edge writes the WebSocket bytes with the code flysim uses in direct mode. +flysim = { path = "../flysim" } +flybus = { path = "../flybus" } + +anyhow = "1.0" +axum = { version = "0.8", features = ["ws"] } +clap = { version = "4.5", features = ["derive"] } +tokio = { version = "1", features = ["rt-multi-thread", "net", "sync", "time", "signal", "macros"] } +tracing = "0.1" +tracing-subscriber = { version = "0.3", features = ["env-filter"] } + +[dev-dependencies] +flate2 = { workspace = true } +futures-util = "0.3" +serde_json = { workspace = true } +tempfile = "3" +tokio-tungstenite = "0.29" diff --git a/services/flysim/crates/fly-edge/src/lib.rs b/services/flysim/crates/fly-edge/src/lib.rs new file mode 100644 index 0000000..996ad7c --- /dev/null +++ b/services/flysim/crates/fly-edge/src/lib.rs @@ -0,0 +1,276 @@ +//! `fly-edge`: the feed WebSocket, served from flysim's feed bus. +//! +//! With `FLY_FEED_VIA=bus` flysim does not bind the feed port. It publishes every snapshot on an +//! embedded flybus router (`flysim::feedbus`), and this process subscribes and serves +//! `ws:///feed` to the stage, the bridge and tests. The contract is still +//! `docs/feed-protocol.md`, byte for byte: the snapshots come off the bus as the same +//! `flysim::snapshot::Snapshot` values and are written by the same `flysim::feed` server, so the +//! per-client `hello`, `wants`, drop-oldest and idle cadence are flysim's own code. +//! +//! Lifecycle (`docs/design/flybus.md`, amendment "Feed store lifecycle"): +//! +//! - the feed port is bound only once the first snapshot has arrived, so before that a client +//! is refused exactly as it would be by a flysim that has not started; +//! - when the bus goes away (flysim stopped or restarted) the edge drops every client and unbinds +//! the port, again exactly what a stopped flysim looks like to the stage, then reconnects every +//! `retry` until a router answers. It never serves a stale snapshot as if it were live. + +use std::net::SocketAddr; +use std::path::PathBuf; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; + +use anyhow::{Context, Result, anyhow}; +use axum::extract::State; +use axum::http::StatusCode; +use axum::response::IntoResponse; +use axum::routing::get; +use flybus::{Client, ClientConfig, SubscriptionConfig}; +use flysim::feed::{self, FeedState}; +use flysim::feedbus; +use flysim::metrics::{Metrics, metric}; +use tokio::sync::{oneshot, watch}; + +/// What the edge needs to know. Built from flysim's own configuration, so both processes read +/// one environment file and cannot disagree about the port, the bus directory or the cadence. +#[derive(Debug, Clone)] +pub struct EdgeConfig { + /// `feed.bus_dir`: the router's socket and store root. + pub bus_dir: PathBuf, + /// `feed.bind`, the port flysim leaves alone in bus mode. + pub feed_bind: SocketAddr, + /// `1 / loop.idle_snapshot_hz`, the protocol's header-only cadence. + pub idle_period: Duration, + /// `FLY_EDGE_METRICS_ADDR`: `/metrics` and `/healthz` for the watchdog, when set. + pub metrics_bind: Option, + /// Delay between attempts to reach the bus. + pub retry: Duration, +} + +impl EdgeConfig { + pub fn from_flysim(config: &flysim::config::Config, metrics_bind: Option) -> Self { + Self { + bus_dir: config.feed.bus_dir.clone(), + feed_bind: config.feed.bind, + idle_period: config.publish_periods().1, + metrics_bind, + retry: Duration::from_millis(500), + } + } +} + +/// The edge's counters. `feed` is the same `Metrics` type flysim uses, so +/// `fly_frames_sent_total` and `fly_feed_clients` mean exactly what they mean there. +#[derive(Debug, Default)] +pub struct EdgeMetrics { + pub feed: Arc, + /// Snapshots taken off the bus and handed to the feed server. + pub snapshots: AtomicU64, + /// 1 while subscribed and serving. + pub connected: AtomicU64, + /// Times a serving session ended because the bus went away. + pub bus_lost: AtomicU64, + /// Publications that could not be turned back into a snapshot. + pub decode_failures: AtomicU64, +} + +impl EdgeMetrics { + pub fn render(&self) -> String { + let mut out = String::with_capacity(1_024); + let feed = &self.feed; + metric( + &mut out, + "fly_frames_sent_total", + "counter", + "Feed snapshots written to a client socket.", + Metrics::get(&feed.frames_sent), + ); + metric( + &mut out, + "fly_feed_clients", + "gauge", + "Feed clients currently subscribed.", + feed.clients(), + ); + metric( + &mut out, + "fly_feed_dropped_total", + "counter", + "Snapshots superseded before a slow client could be sent them.", + Metrics::get(&feed.feed_dropped), + ); + metric( + &mut out, + "fly_edge_snapshots_total", + "counter", + "Snapshots taken off the feed bus.", + self.snapshots.load(Ordering::Relaxed), + ); + metric( + &mut out, + "fly_edge_bus_connected", + "gauge", + "1 while the edge is subscribed to the feed bus and serving.", + self.connected.load(Ordering::Relaxed), + ); + metric( + &mut out, + "fly_edge_bus_lost_total", + "counter", + "Serving sessions ended by the feed bus going away.", + self.bus_lost.load(Ordering::Relaxed), + ); + metric( + &mut out, + "fly_edge_decode_failures_total", + "counter", + "Feed bus publications that did not decode to a snapshot.", + self.decode_failures.load(Ordering::Relaxed), + ); + out + } +} + +/// Serve until the process is stopped. Only a metrics listener that cannot bind is fatal; +/// everything about the bus is retried. +pub async fn run(config: EdgeConfig, metrics: Arc) -> Result<()> { + if let Some(addr) = config.metrics_bind { + let listener = tokio::net::TcpListener::bind(addr) + .await + .with_context(|| format!("binding the edge metrics listener on {addr}"))?; + let app = axum::Router::new() + .route("/metrics", get(prometheus)) + .route("/healthz", get(healthz)) + .with_state(Arc::clone(&metrics)); + tokio::spawn(async move { + if let Err(error) = axum::serve(listener, app).await { + tracing::error!(%error, "the edge metrics listener stopped"); + } + }); + } + let mut quiet = false; + loop { + match session(&config, &metrics).await { + Ok(()) => { + 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; + } + } + } + tokio::time::sleep(config.retry).await; + } +} + +async fn prometheus(State(metrics): State>) -> impl IntoResponse { + ( + [( + axum::http::header::CONTENT_TYPE, + "text/plain; version=0.0.4", + )], + metrics.render(), + ) +} + +async fn healthz(State(metrics): State>) -> impl IntoResponse { + if metrics.connected.load(Ordering::Relaxed) == 1 { + (StatusCode::OK, "ok") + } else { + (StatusCode::SERVICE_UNAVAILABLE, "waiting for the feed bus") + } +} + +/// 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<()> { + let client = Client::connect_unix( + feedbus::socket_path(&config.bus_dir), + ClientConfig::new(feedbus::EDGE, feedbus::store_root(&config.bus_dir)), + ) + .await + .map_err(|error| anyhow!("connecting to the feed bus: {error}"))?; + // One in flight: while a snapshot is being copied out, the next one waits in the single + // latest slot and anything newer replaces it. The edge is never more than one behind. + let mut subscription = client + .subscribe( + feedbus::TOPIC, + SubscriptionConfig::latest().in_flight(1).replay(true), + ) + .await + .map_err(|error| anyhow!("subscribing to {}: {error}", feedbus::TOPIC))?; + let first = loop { + let message = subscription + .next() + .await + .ok_or_else(|| anyhow!("the feed bus closed before the first snapshot"))?; + match feedbus::receive(&message).await { + Ok(snapshot) => break snapshot, + Err(error) => { + metrics.decode_failures.fetch_add(1, Ordering::Relaxed); + tracing::warn!(%error, "a feed bus publication did not decode"); + } + } + }; + 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); + + let state = FeedState { + snapshots: receiver, + metrics: Arc::clone(&metrics.feed), + idle_period: config.idle_period, + }; + let (stop, stopped) = oneshot::channel::<()>(); + let server = tokio::spawn(async move { + let result = axum::serve(listener, feed::router(state)) + .with_graceful_shutdown(async move { + let _ = stopped.await; + }) + .await; + if let Err(error) = result { + tracing::error!(%error, "the feed listener stopped"); + } + }); + + while let Some(message) = subscription.next().await { + match feedbus::receive(&message).await { + Ok(snapshot) => { + drop(message); + snapshots.send_replace(Arc::new(snapshot)); + metrics.snapshots.fetch_add(1, Ordering::Relaxed); + } + Err(error) => { + metrics.decode_failures.fetch_add(1, Ordering::Relaxed); + tracing::warn!(%error, "a feed bus publication did not decode"); + if client.closed().is_some() { + break; + } + } + } + } + + // The bus is gone. Dropping the sender ends every client's pump (a closed stream, as when + // flysim itself stops), and the graceful shutdown unbinds the port. + metrics.connected.store(0, Ordering::Relaxed); + metrics.bus_lost.fetch_add(1, Ordering::Relaxed); + drop(snapshots); + let _ = stop.send(()); + if tokio::time::timeout(Duration::from_secs(5), server) + .await + .is_err() + { + tracing::warn!("the feed listener took more than 5 s to stop"); + } + Ok(()) +} diff --git a/services/flysim/crates/fly-edge/src/main.rs b/services/flysim/crates/fly-edge/src/main.rs new file mode 100644 index 0000000..067637b --- /dev/null +++ b/services/flysim/crates/fly-edge/src/main.rs @@ -0,0 +1,81 @@ +//! `fly-edge`: serve the feed WebSocket from flysim's feed bus. +//! +//! ```sh +//! FLY_FEED_VIA=bus flysim & +//! fly-edge +//! ``` +//! +//! Configured through the same environment as flysim (`FLY_FEED_BIND`, `FLY_BUS_DIR`, +//! `FLYSIM_LOOP_IDLE_SNAPSHOT_HZ`, or `--config flysim.toml`), plus `FLY_EDGE_METRICS_ADDR` for +//! its own `/metrics` and `/healthz`. `infra/units/flyedge.service` runs it with no arguments. + +use std::sync::Arc; + +use anyhow::{Context, Result}; +use clap::Parser; +use fly_edge::{EdgeConfig, EdgeMetrics}; + +#[derive(Debug, Parser)] +#[command( + name = "fly-edge", + about = "The feed WebSocket, served from flysim's feed bus.", + version +)] +struct Args { + /// Path to `flysim.toml`, read for `[feed]` and `[loop]`. Environment overrides apply as + /// they do for flysim. + #[arg(long, value_name = "PATH")] + config: Option, +} + +fn main() -> Result<()> { + let args = Args::parse(); + init_tracing(); + let config = flysim::config::Config::load(args.config.as_deref())?; + let metrics_bind = match std::env::var("FLY_EDGE_METRICS_ADDR") { + Ok(value) if !value.is_empty() => Some(value.parse().with_context(|| { + format!("FLY_EDGE_METRICS_ADDR: {value:?} is not a host:port address") + })?), + _ => None, + }; + let edge = EdgeConfig::from_flysim(&config, metrics_bind); + if config.feed.via != flysim::config::FeedVia::Bus { + tracing::warn!( + "FLY_FEED_VIA is not \"bus\": flysim serves the feed itself and binds {}; \ + this edge will wait for a bus that is not there", + edge.feed_bind + ); + } + tracing::info!(config = ?edge, "fly-edge starting"); + + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .thread_name("fly-edge") + .enable_all() + .build() + .context("building the tokio runtime")?; + runtime.block_on(async move { + let metrics = Arc::new(EdgeMetrics::default()); + let mut terminate = + tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) + .context("installing the SIGTERM handler")?; + tokio::select! { + result = fly_edge::run(edge, metrics) => result, + _ = tokio::signal::ctrl_c() => { tracing::info!("SIGINT: shutting down"); Ok(()) } + _ = terminate.recv() => { tracing::info!("SIGTERM: shutting down"); Ok(()) } + } + }) +} + +/// Logs to stderr, like flysim, under `FLY_EDGE_LOG` (or `RUST_LOG`). +fn init_tracing() { + use tracing_subscriber::EnvFilter; + let filter = EnvFilter::try_from_env("FLY_EDGE_LOG") + .or_else(|_| EnvFilter::try_from_default_env()) + .unwrap_or_else(|_| EnvFilter::new("info")); + tracing_subscriber::fmt() + .with_env_filter(filter) + .with_writer(std::io::stderr) + .with_target(false) + .init(); +}