From acf7c2ebb88c2e84db1a3de4f8bfafd0f2b8911f Mon Sep 17 00:00:00 2001 From: acamilo Date: Wed, 23 Sep 2026 08:16:31 +0000 Subject: [PATCH] flysim: the feed server takes a FeedState, so another process can run it feed::router needed the whole AppState for three things: the watch slot, the feed counters and the idle cadence. It now takes exactly those, and Shared's Metrics sits behind an Arc so the counters can be handed over. Nothing about what the feed writes changes. --- services/flysim/crates/flysim/src/feed.rs | 41 ++++++++++++++------ services/flysim/crates/flysim/src/lib.rs | 11 +++++- services/flysim/crates/flysim/src/metrics.rs | 2 +- services/flysim/crates/flysim/src/simloop.rs | 4 +- 4 files changed, 43 insertions(+), 15 deletions(-) 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(),