diff --git a/services/flysim/crates/flysim/src/feed.rs b/services/flysim/crates/flysim/src/feed.rs index dad5879..2e3d1c0 100644 --- a/services/flysim/crates/flysim/src/feed.rs +++ b/services/flysim/crates/flysim/src/feed.rs @@ -9,8 +9,14 @@ //! - 30 snapshots a second while running, 2 while paused or booting (header only); //! - drop-oldest, never queue: the sim publishes into a `watch` slot, so a slow client misses //! snapshots instead of slowing the loop down. Those misses are counted. +//! +//! The server only needs a [`FeedState`]: a watch slot of snapshots, the counters and the idle +//! cadence. flysim builds one from its own state when it serves the feed itself +//! (`FLY_FEED_VIA=direct`), and `fly-edge` builds one from the snapshots it takes off the bus +//! (`FLY_FEED_VIA=bus`), so both paths run this same code and write the same bytes. use std::sync::Arc; +use std::time::Duration; use axum::Router; use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade}; @@ -19,9 +25,22 @@ use axum::http::StatusCode; use axum::response::{IntoResponse, Response}; use axum::routing::any; use serde::Deserialize; +use tokio::sync::watch; +use crate::metrics::Metrics; use crate::snapshot::{AttachmentKind, FeedStatus, PROTOCOL, Snapshot, Wants}; -use crate::{AppState, metrics::Metrics}; + +/// Everything the feed server reads. +#[derive(Clone)] +pub struct FeedState { + /// The newest snapshot. Dropping its sender ends every client's stream. + pub snapshots: watch::Receiver>, + /// `frames_sent`, `feed_clients` and `feed_dropped` are the ones this module moves. + pub metrics: Arc, + /// The protocol's idle cadence: how long a paused or booting stream waits before it + /// repeats the current header (`config.publish_periods().1`). + pub idle_period: Duration, +} /// The one JSON text message a client sends on connect. #[derive(Debug, Clone, Deserialize)] @@ -37,7 +56,7 @@ pub struct ClientHello { /// Close code for a protocol violation, as the reference server uses. const CLOSE_PROTOCOL_ERROR: u16 = 1002; -pub fn router(state: AppState) -> Router { +pub fn router(state: FeedState) -> Router { Router::new() .route("/feed", any(upgrade)) .fallback(not_found) @@ -48,11 +67,11 @@ async fn not_found() -> Response { (StatusCode::NOT_FOUND, "not found").into_response() } -async fn upgrade(upgrade: WebSocketUpgrade, State(state): State) -> Response { +async fn upgrade(upgrade: WebSocketUpgrade, State(state): State) -> Response { upgrade.on_upgrade(move |socket| serve_client(socket, state)) } -async fn serve_client(mut socket: WebSocket, state: AppState) { +async fn serve_client(mut socket: WebSocket, state: FeedState) { let Some(hello) = read_hello(&mut socket).await else { return; }; @@ -64,9 +83,9 @@ async fn serve_client(mut socket: WebSocket, state: AppState) { spikes = wants.spikes, "feed client connected" ); - state.shared.metrics.client_joined(); + state.metrics.client_joined(); let result = pump(&mut socket, &state, wants).await; - state.shared.metrics.client_left(); + state.metrics.client_left(); match result { Ok(()) => tracing::info!("feed client disconnected"), Err(error) => tracing::info!(%error, "feed client dropped"), @@ -117,9 +136,9 @@ async fn read_hello(socket: &mut WebSocket) -> Option { None } -async fn pump(socket: &mut WebSocket, state: &AppState, wants: Wants) -> Result<(), axum::Error> { +async fn pump(socket: &mut WebSocket, state: &FeedState, wants: Wants) -> Result<(), axum::Error> { let mut receiver = state.snapshots.clone(); - let (_, idle_period) = state.shared.config.publish_periods(); + let idle_period = state.idle_period; let mut last_seq = 0u64; // The current snapshot first, so a client that connects while paused or booting sees the @@ -161,18 +180,18 @@ async fn pump(socket: &mut WebSocket, state: &AppState, wants: Wants) -> Result< async fn send( socket: &mut WebSocket, - state: &AppState, + state: &FeedState, snapshot: &Arc, wants: Wants, last_seq: &mut u64, ) -> Result<(), axum::Error> { let seq = snapshot.header.seq; if seq > *last_seq + 1 && *last_seq != 0 { - Metrics::add(&state.shared.metrics.feed_dropped, seq - *last_seq - 1); + Metrics::add(&state.metrics.feed_dropped, seq - *last_seq - 1); } *last_seq = seq; socket.send(Message::Binary(snapshot.encode(wants).into())).await?; - Metrics::incr(&state.shared.metrics.frames_sent); + Metrics::incr(&state.metrics.frames_sent); Ok(()) } diff --git a/services/flysim/crates/flysim/src/lib.rs b/services/flysim/crates/flysim/src/lib.rs index be481b0..a2811ec 100644 --- a/services/flysim/crates/flysim/src/lib.rs +++ b/services/flysim/crates/flysim/src/lib.rs @@ -59,6 +59,15 @@ impl AppState { pub fn snapshot(&self) -> Arc { Arc::clone(&self.snapshots.borrow()) } + + /// What the feed server needs, when flysim serves the feed itself. + pub fn feed(&self) -> feed::FeedState { + feed::FeedState { + snapshots: self.snapshots.clone(), + metrics: Arc::clone(&self.shared.metrics), + idle_period: self.shared.config.publish_periods().1, + } + } } /// Run the service until a signal or a fatal simulation error. @@ -107,7 +116,7 @@ pub fn run(config: Config) -> Result<()> { tracing::info!(feed = %feed_addr, control = %control_addr, metrics = ?metrics_addr, "listening"); { - let state = state.clone(); + let state = state.feed(); runtime.spawn(async move { if let Err(error) = axum::serve(feed_listener, feed::router(state)).await { tracing::error!(%error, "the feed listener stopped"); diff --git a/services/flysim/crates/flysim/src/metrics.rs b/services/flysim/crates/flysim/src/metrics.rs index 994e864..e4e4284 100644 --- a/services/flysim/crates/flysim/src/metrics.rs +++ b/services/flysim/crates/flysim/src/metrics.rs @@ -81,7 +81,7 @@ impl Metrics { } /// One metric line plus its help and type headers. -fn metric(out: &mut String, name: &str, kind: &str, help: &str, value: impl std::fmt::Display) { +pub fn metric(out: &mut String, name: &str, kind: &str, help: &str, value: impl std::fmt::Display) { use std::fmt::Write as _; let _ = writeln!(out, "# HELP {name} {help}"); let _ = writeln!(out, "# TYPE {name} {kind}"); diff --git a/services/flysim/crates/flysim/src/simloop.rs b/services/flysim/crates/flysim/src/simloop.rs index f5ecd55..0b24bcb 100644 --- a/services/flysim/crates/flysim/src/simloop.rs +++ b/services/flysim/crates/flysim/src/simloop.rs @@ -149,7 +149,7 @@ pub struct DecoderChannelStatus { #[derive(Debug)] pub struct Shared { pub config: Config, - pub metrics: Metrics, + pub metrics: Arc, pub events: EventRing, /// `Date.now()` at the top of the most recent loop iteration. `GET /healthz` is 200 while /// this is less than two seconds old, which is true while paused as well: a paused loop is @@ -173,7 +173,7 @@ impl Shared { pub fn new(config: Config, events: EventRing) -> Self { Self { config, - metrics: Metrics::default(), + metrics: Arc::default(), events, heartbeat_ms: AtomicU64::new(0), versions: OnceLock::new(),