diff --git a/services/flysim/Cargo.lock b/services/flysim/Cargo.lock index 7399620..7f84d26 100644 --- a/services/flysim/Cargo.lock +++ b/services/flysim/Cargo.lock @@ -484,6 +484,7 @@ dependencies = [ "fly-session-types", "flybrain-core", "flybrain-gb", + "flybus", "futures-util", "jsonschema", "serde", diff --git a/services/flysim/crates/flysim/Cargo.toml b/services/flysim/crates/flysim/Cargo.toml index 4d3109e..9f6c3db 100644 --- a/services/flysim/crates/flysim/Cargo.toml +++ b/services/flysim/crates/flysim/Cargo.toml @@ -30,6 +30,7 @@ cuda = ["flybrain-core/cuda"] [dependencies] flybrain-core = { path = "../flybrain-core" } flybrain-gb = { path = "../flybrain-gb" } +flybus = { path = "../flybus" } anyhow = "1.0" axum = { version = "0.8", features = ["ws"] } diff --git a/services/flysim/crates/flysim/src/config.rs b/services/flysim/crates/flysim/src/config.rs index 08d4674..e3ff354 100644 --- a/services/flysim/crates/flysim/src/config.rs +++ b/services/flysim/crates/flysim/src/config.rs @@ -9,7 +9,7 @@ //! - the `FLY_*` names the systemd units already set (`FLY_GAME`, `FLY_ROM`, `FLY_DATASET`, //! `FLY_STATE`, `FLY_STATE_HOT`, `FLY_FEED_BIND`, `FLY_CONTROL_BIND`, `FLY_METRICS_ADDR`, //! `FLY_ROM_SHA256`, `FLY_ROM_PLATFORMER_SHA256`, `FLY_CHAT_ENABLED`, `FLY_CHAT_DENY_LIST`, -//! `FLY_MACRO_MODE`, `RAYON_NUM_THREADS`); +//! `FLY_MACRO_MODE`, `FLY_FEED_VIA`, `FLY_BUS_DIR`, `RAYON_NUM_THREADS`); //! - `FLYSIM_
_` for everything, e.g. `FLYSIM_LOOP_SPEED=0`. //! //! Nothing here is secret (`docs/control-api.md`: "No secrets live in this service or its @@ -104,6 +104,35 @@ pub struct Feed { pub bind: SocketAddr, /// Audio attachment sample rate. The page wants Web Audio's native 48 kHz. pub audio_hz: u32, + /// Who serves `:7400/feed` (`docs/design/flybus.md`, "Feed over the bus"). + pub via: FeedVia, + /// The bus runtime directory in `bus` mode: the router's socket and its artifact store. + /// Belongs on tmpfs; a store here holds a few snapshots, never history. + pub bus_dir: PathBuf, +} + +/// Where the feed WebSocket is served from. +/// +/// `direct` is the default and is the behaviour that predates the bus, byte for byte: flysim +/// binds `feed.bind` itself. `bus` starts an embedded flybus router, publishes every snapshot +/// on it, and leaves `feed.bind` to the `fly-edge` process. The control API stays in flysim +/// either way. Nothing about the fly changes with this knob: it is outside the simulation loop +/// and outside the compatibility string. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum FeedVia { + #[default] + Direct, + Bus, +} + +impl FeedVia { + pub const fn as_str(self) -> &'static str { + match self { + Self::Direct => "direct", + Self::Bus => "bus", + } + } } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] @@ -197,6 +226,8 @@ impl Default for Feed { Self { bind: "127.0.0.1:7400".parse().expect("literal address"), audio_hz: 48_000, + via: FeedVia::Direct, + bus_dir: PathBuf::from("/run/fly/bus"), } } } @@ -255,6 +286,12 @@ impl Config { if let Some(value) = get("FLY_FEED_BIND") { self.feed.bind = parse_addr("FLY_FEED_BIND", value)?; } + if let Some(value) = get("FLY_FEED_VIA") { + self.feed.via = parse_feed_via("FLY_FEED_VIA", value)?; + } + if let Some(value) = get("FLY_BUS_DIR") { + self.feed.bus_dir = PathBuf::from(value); + } if let Some(value) = get("FLY_CONTROL_BIND") { self.control.bind = parse_addr("FLY_CONTROL_BIND", value)?; } @@ -330,6 +367,12 @@ impl Config { if let Some(value) = get("FLYSIM_FEED_AUDIO_HZ") { self.feed.audio_hz = parse("FLYSIM_FEED_AUDIO_HZ", value)?; } + if let Some(value) = get("FLYSIM_FEED_VIA") { + self.feed.via = parse_feed_via("FLYSIM_FEED_VIA", value)?; + } + if let Some(value) = get("FLYSIM_FEED_BUS_DIR") { + self.feed.bus_dir = PathBuf::from(value); + } if let Some(value) = get("FLYSIM_CONTROL_BIND") { self.control.bind = parse_addr("FLYSIM_CONTROL_BIND", value)?; } @@ -464,6 +507,14 @@ impl Config { } } +fn parse_feed_via(name: &str, value: &str) -> Result { + match value.to_ascii_lowercase().as_str() { + "direct" => Ok(FeedVia::Direct), + "bus" => Ok(FeedVia::Bus), + _ => bail!("{name}: {value:?} is not a feed path; expected \"direct\" or \"bus\""), + } +} + fn parse_addr(name: &str, value: &str) -> Result { value .parse() @@ -687,6 +738,29 @@ mod tests { assert_eq!(config, Config::default()); } + #[test] + fn the_feed_path_is_direct_unless_the_environment_says_bus() { + let config = Config::default(); + assert_eq!(config.feed.via, FeedVia::Direct); + assert_eq!(config.feed.bus_dir, PathBuf::from("/run/fly/bus")); + + let mut config = Config::default(); + config + .apply_env(&env(&[("FLY_FEED_VIA", "bus"), ("FLY_BUS_DIR", "/tmp/fly-bus")])) + .unwrap(); + assert_eq!(config.feed.via, FeedVia::Bus); + assert_eq!(config.feed.bus_dir, PathBuf::from("/tmp/fly-bus")); + + let mut config = Config::default(); + config.apply_env(&env(&[("FLYSIM_FEED_VIA", "DIRECT")])).unwrap(); + assert_eq!(config.feed.via, FeedVia::Direct); + + // A typo is a refusal, not a silent fallback to one of the two. + let error = Config::default().apply_env(&env(&[("FLY_FEED_VIA", "buss")])).unwrap_err(); + assert!(error.to_string().contains("FLY_FEED_VIA"), "{error}"); + assert_eq!(toml::from_str::("[feed]\nvia = \"bus\"\n").unwrap().feed.via, FeedVia::Bus); + } + #[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 new file mode 100644 index 0000000..bd97644 --- /dev/null +++ b/services/flysim/crates/flysim/src/feedbus.rs @@ -0,0 +1,347 @@ +//! The feed over flybus (`FLY_FEED_VIA=bus`, `docs/design/flybus.md` "Feed over the bus"). +//! +//! Both halves of the bus encoding live here, so the publisher in flysim and the subscriber in +//! `fly-edge` cannot drift apart: +//! +//! - [`publish`] turns one [`Snapshot`] into one publication on [`TOPIC`]: every attachment the +//! header lists as a sealed artifact named after its kind (`frame`, `audio`, `spikes`), and the +//! header itself as the envelope payload `{"header": {...}}`. A header too large for an +//! envelope travels as a `header` artifact instead, so no snapshot is ever unpublishable. +//! - [`receive`] turns that publication back into the same [`Snapshot`], which `fly-edge` hands to +//! [`crate::feed`] exactly as flysim does. The WebSocket bytes are therefore produced by the same +//! `Snapshot::encode` on both paths. +//! +//! The simulation thread never sees any of this. It publishes into its `watch` slot as it +//! always has; [`run_publisher`] is a task on the bus's own runtime that reads that slot and +//! skips whatever it was too slow to see, the same drop-oldest rule every feed client gets. +//! A stalled router, a full store or an absent edge can cost snapshots on the bus, never a +//! frame of the loop. + +use std::io::Write as _; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use flybus::{ + Artifact, BusError, Client, ClientConfig, ErrorCode, Grants, Limits, Message, Pattern, Policy, + PublishReceipt, Retained, Router, RouterConfig, UnixListenerHandle, +}; +use serde_json::{Map, Value}; +use tokio::sync::watch; + +use crate::metrics::Metrics; +use crate::snapshot::{AttachmentKind, FeedHeader, Snapshot}; + +/// The one topic: `latest` retention, so a subscriber that joins late starts from the newest +/// snapshot and one that falls behind is coalesced rather than queued. +pub const TOPIC: &str = "fly.feed.snapshots"; +/// The publisher's participant id (in-process, launcher-bound). +pub const PUBLISHER: &str = "flysim"; +/// The edge's participant id; the Unix socket is bound to it. +pub const EDGE: &str = "fly-edge"; +/// Socket file under `feed.bus_dir`, bound to [`EDGE`] only. +pub const SOCKET: &str = "edge.sock"; +/// Store root under `feed.bus_dir`; the router makes its per-incarnation directory inside it. +pub const STORE: &str = "store"; +/// Headers up to this many JSON bytes ride in the envelope; anything larger becomes an artifact. +/// Well under flybus's 65,536-byte envelope limit, leaving room for the attachment references +/// and the router's ids. A real header is 2 to 8 KB. +pub const HEADER_INLINE_MAX: usize = 48 * 1024; +/// The attachment name of an out-of-line header. +pub const HEADER_ARTIFACT: &str = "header"; + +/// `/edge.sock`. +pub fn socket_path(bus_dir: &Path) -> PathBuf { + bus_dir.join(SOCKET) +} + +/// `/store`. +pub fn store_root(bus_dir: &Path) -> PathBuf { + bus_dir.join(STORE) +} + +/// The router limits for the feed (`docs/design/flybus.md`, amendment "Feed sizing"). +/// +/// One snapshot with attachments is 122,367 bytes on the live fly: a 92,160-byte 160x144 RGBA +/// 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. +pub fn limits() -> Limits { + Limits { + max_clients: 8, + max_services: 8, + max_topics: 8, + max_subscriptions_per_client: 4, + max_subscriptions: 16, + max_latest_in_flight: 2, + max_owners_per_client: 64, + reserved_owners_per_client: 8, + // Audio accumulates while the loop is behind its publish deadline; 4 MiB is ten seconds + // of it, far past anything the pacer allows before it logs lag. + max_artifact_bytes: 4 << 20, + max_store_bytes: 32 << 20, + max_retained_bytes: 8 << 20, + ..Limits::default() + } +} + +/// flysim may declare and publish the feed topic; the edge may only subscribe to it. +pub fn policy() -> Policy { + Policy::closed() + .client( + PUBLISHER, + Grants { + publish: vec![Pattern::exact(TOPIC)], + manage_topics: vec![Pattern::exact(TOPIC)], + ..Grants::default() + }, + ) + .client( + EDGE, + Grants { + subscribe: vec![Pattern::exact(TOPIC)], + ..Grants::default() + }, + ) +} + +/// A running router and the edge's socket. Dropping it stops listening; the router stops with +/// the runtime it was started on. +pub struct BusFeed { + pub router: Router, + _listener: UnixListenerHandle, +} + +/// Start the embedded router under `bus_dir` and listen for the edge on `/edge.sock`. +/// +/// Must run inside a Tokio runtime. `Router::new` removes store directories a previous flysim +/// left behind (their `flock` is free once that process is gone); a stale socket file is removed +/// here, because a socket outlives its listener on disk. +pub async fn start_router(bus_dir: &Path) -> anyhow::Result { + use anyhow::Context as _; + use std::os::unix::fs::DirBuilderExt as _; + std::fs::DirBuilder::new() + .recursive(true) + .mode(0o700) + .create(bus_dir) + .with_context(|| format!("creating the bus directory {}", bus_dir.display()))?; + let root = store_root(bus_dir); + std::fs::DirBuilder::new() + .recursive(true) + .mode(0o700) + .create(&root) + .with_context(|| format!("creating the bus store root {}", root.display()))?; + let socket = socket_path(bus_dir); + match std::fs::remove_file(&socket) { + Ok(()) => tracing::info!(socket = %socket.display(), "removed a stale bus socket"), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => { + return Err(error).with_context(|| format!("removing {}", socket.display())); + } + } + let mut config = RouterConfig::new(root); + config.limits = limits(); + config.policy = policy(); + let router = Router::new(config).context("starting the flybus router")?; + let listener = router + .listen_unix_as(&socket, EDGE) + .await + .with_context(|| format!("listening on {}", socket.display()))?; + tracing::info!( + socket = %socket.display(), + store = %router.store_dir().display(), + router = router.router_id(), + "feed bus listening" + ); + Ok(BusFeed { + router, + _listener: listener, + }) +} + +fn attachment_name(kind: AttachmentKind) -> &'static str { + match kind { + AttachmentKind::Frame => "frame", + AttachmentKind::Audio => "audio", + AttachmentKind::Spikes => "spikes", + } +} + +fn content_type(kind: AttachmentKind) -> &'static str { + match kind { + AttachmentKind::Frame => "image/x-rgba", + AttachmentKind::Audio => "audio/x-f32le", + AttachmentKind::Spikes => "application/x-spike-bitset", + } +} + +fn bytes_of(snapshot: &Snapshot, kind: AttachmentKind) -> &[u8] { + match kind { + AttachmentKind::Frame => &snapshot.frame, + AttachmentKind::Audio => &snapshot.audio, + AttachmentKind::Spikes => &snapshot.spikes, + } +} + +async fn seal(client: &Client, bytes: &[u8], content_type: &str) -> Result { + let mut writer = client + .artifacts() + .allocate(bytes.len() as u64, content_type) + .await?; + writer + .write_all(bytes) + .map_err(|error| BusError::new(ErrorCode::StoreFailure, format!("staging: {error}")))?; + writer.seal().await +} + +/// Publish one snapshot: its attachments as artifacts, its header as the payload. +pub async fn publish(client: &Client, snapshot: &Snapshot) -> Result { + let header = &snapshot.header; + let json = serde_json::to_vec(header).expect("a FeedHeader always serializes"); + + let mut kinds: Vec = Vec::with_capacity(3); + for kind in header.attachments.iter().copied() { + if !kinds.contains(&kind) { + kinds.push(kind); + } + } + let mut artifacts: Vec<(&'static str, Artifact)> = Vec::with_capacity(4); + for kind in kinds { + let artifact = seal(client, bytes_of(snapshot, kind), content_type(kind)).await?; + artifacts.push((attachment_name(kind), artifact)); + } + + let mut payload = Map::new(); + if json.len() <= HEADER_INLINE_MAX { + let value: Value = serde_json::from_slice(&json).expect("a serialized header re-parses"); + payload.insert("header".into(), value); + } else { + artifacts.push(( + HEADER_ARTIFACT, + seal(client, &json, "application/json").await?, + )); + } + let attachments: Vec<(&str, &Artifact)> = artifacts + .iter() + .map(|(name, artifact)| (*name, artifact)) + .collect(); + client.publish(TOPIC, payload, &attachments).await +} + +/// Rebuild the snapshot one publication carries. The message's delivery is released when the +/// caller drops it; every byte has been copied out by then. +pub async fn receive(message: &Message) -> Result { + let invalid = |what: String| BusError::new(ErrorCode::InvalidEnvelope, what); + let header: FeedHeader = match message.payload().get("header") { + Some(value) => serde_json::from_value(value.clone()) + .map_err(|error| invalid(format!("feed header: {error}")))?, + None => { + let bytes = message.artifact(HEADER_ARTIFACT)?.read_all().await?; + serde_json::from_slice(&bytes) + .map_err(|error| invalid(format!("feed header artifact: {error}")))? + } + }; + let mut snapshot = Snapshot { + header, + frame: Arc::new(Vec::new()), + audio: Arc::new(Vec::new()), + spikes: Arc::new(Vec::new()), + }; + let kinds = snapshot.header.attachments.clone(); + for kind in kinds { + let bytes = Arc::new(message.artifact(attachment_name(kind))?.read_all().await?); + match kind { + AttachmentKind::Frame => snapshot.frame = bytes, + AttachmentKind::Audio => snapshot.audio = bytes, + AttachmentKind::Spikes => snapshot.spikes = bytes, + } + } + Ok(snapshot) +} + +/// How often a failing publisher repeats its warning. +const WARN_EVERY: Duration = Duration::from_secs(10); + +/// Publish every snapshot the sim puts in its watch slot until the sim is gone. +/// +/// Connects in process as [`PUBLISHER`], declares [`TOPIC`] with `latest` retention and +/// publishes the current snapshot first, so an edge that connects at once still sees the boot +/// state. A refused publication (a full store, say) is counted and skipped; a lost connection +/// is re-made after a second. Borrows of the watch slot end before any await, exactly as in +/// [`crate::feed`]: a held borrow is a lock the sim thread's next publish would wait on. +pub async fn run_publisher( + router: Router, + mut snapshots: watch::Receiver>, + metrics: Arc, +) { + let store_root = router.store_root().to_path_buf(); + let mut last_warning: Option = None; + let mut warn = |error: &BusError, what: &str| { + if last_warning.is_none_or(|at| at.elapsed() >= WARN_EVERY) { + tracing::warn!(%error, "feed bus: {what}"); + last_warning = Some(Instant::now()); + } + }; + loop { + let transport = router.connect_in_memory_as(PUBLISHER); + let client = + match Client::connect(transport, ClientConfig::new(PUBLISHER, &store_root)).await { + Ok(client) => client, + Err(error) => { + warn(&error, "the publisher could not connect"); + tokio::time::sleep(Duration::from_secs(1)).await; + continue; + } + }; + if let Err(error) = client.declare_topic(TOPIC, Retained::Latest).await { + warn(&error, "the feed topic could not be declared"); + tokio::time::sleep(Duration::from_secs(1)).await; + continue; + } + let mut current = snapshots.borrow_and_update().clone(); + loop { + match publish(&client, ¤t).await { + Ok(_) => Metrics::incr(&metrics.bus_published), + Err(error) => { + Metrics::incr(&metrics.bus_publish_failures); + warn(&error, "a snapshot was not published"); + if client.closed().is_some() { + break; + } + } + } + if snapshots.changed().await.is_err() { + // The sim thread is gone; so is the service. + client.close().await; + return; + } + current = snapshots.borrow_and_update().clone(); + } + tokio::time::sleep(Duration::from_secs(1)).await; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn the_limits_validate_and_hold_the_worst_case_with_room() { + let limits = limits(); + limits.validate().unwrap(); + // 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; + assert!( + pinned * snapshot_bytes as u64 * 10 <= limits.max_store_bytes, + "{pinned}" + ); + assert!(limits.max_artifact_bytes >= crate::snapshot::FRAME_BYTES as u64 * 40); + } +} diff --git a/services/flysim/crates/flysim/src/lib.rs b/services/flysim/crates/flysim/src/lib.rs index a2811ec..335f949 100644 --- a/services/flysim/crates/flysim/src/lib.rs +++ b/services/flysim/crates/flysim/src/lib.rs @@ -8,7 +8,8 @@ //! //! ```text //! +-- watch --> feed :7400/feed (axum + ws) -//! sim thread ---------+ +//! sim thread ---------+ \-> feedbus -> flybus -> fly-edge :7400/feed +//! | (FLY_FEED_VIA=bus instead of the line above) //! agent +-- Shared ------------> api :7401 (axum) //! emulator | /status /stimulate /reward /checkpoint //! adapter | /pause /resume /events /healthz /metrics @@ -25,6 +26,7 @@ pub mod chat; pub mod config; pub mod eventlog; pub mod feed; +pub mod feedbus; pub mod macros; pub mod metrics; pub mod pacing; @@ -41,7 +43,7 @@ use std::sync::Arc; use anyhow::{Context, Result}; use tokio::sync::{mpsc, watch}; -use crate::config::Config; +use crate::config::{Config, FeedVia}; use crate::eventlog::{EventRing, now_wall_ms}; use crate::simloop::{COMMAND_QUEUE, Command, Shared, Sim, booting_snapshot}; use crate::snapshot::Snapshot; @@ -95,10 +97,17 @@ pub fn run(config: Config) -> Result<()> { let feed_addr = config.feed.bind; let control_addr = config.control.bind; let metrics_addr = config.control.metrics_bind; + let via = config.feed.via; let listeners = runtime.block_on(async { - let feed = tokio::net::TcpListener::bind(feed_addr) - .await - .with_context(|| format!("binding the feed listener on {feed_addr}"))?; + // In bus mode the feed port belongs to `fly-edge`; binding it here would take it away. + let feed = match via { + FeedVia::Direct => Some( + tokio::net::TcpListener::bind(feed_addr) + .await + .with_context(|| format!("binding the feed listener on {feed_addr}"))?, + ), + FeedVia::Bus => None, + }; let control = tokio::net::TcpListener::bind(control_addr) .await .with_context(|| format!("binding the control listener on {control_addr}"))?; @@ -113,9 +122,15 @@ pub fn run(config: Config) -> Result<()> { Ok::<_, anyhow::Error>((feed, control, metrics)) })?; let (feed_listener, control_listener, metrics_listener) = listeners; - tracing::info!(feed = %feed_addr, control = %control_addr, metrics = ?metrics_addr, "listening"); + tracing::info!( + feed = %feed_addr, + feed_via = via.as_str(), + control = %control_addr, + metrics = ?metrics_addr, + "listening" + ); - { + if let Some(feed_listener) = feed_listener { let state = state.feed(); runtime.spawn(async move { if let Err(error) = axum::serve(feed_listener, feed::router(state)).await { @@ -123,6 +138,26 @@ pub fn run(config: Config) -> Result<()> { } }); } + // The bus gets a runtime of its own, so neither its router nor the artifact copies can take + // a worker from the control API; and it is fed from the watch slot, never from the sim thread. + let bus_runtime = match via { + FeedVia::Direct => None, + FeedVia::Bus => { + let bus_runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .thread_name("flysim-bus") + .enable_all() + .build() + .context("building the bus runtime")?; + let bus = bus_runtime.block_on(feedbus::start_router(&config.feed.bus_dir))?; + bus_runtime.spawn(feedbus::run_publisher( + bus.router.clone(), + state.snapshots.clone(), + Arc::clone(&state.shared.metrics), + )); + Some((bus_runtime, bus)) + } + }; { let state = state.clone(); runtime.spawn(async move { @@ -148,6 +183,11 @@ pub fn run(config: Config) -> Result<()> { let result = sim.run(¬ifier); notifier.notify("STOPPING=1\n"); drop(sim); + if let Some((bus_runtime, bus)) = bus_runtime { + bus.router.shutdown(); + drop(bus); + bus_runtime.shutdown_timeout(std::time::Duration::from_secs(1)); + } runtime.shutdown_timeout(std::time::Duration::from_secs(2)); result } diff --git a/services/flysim/crates/flysim/src/metrics.rs b/services/flysim/crates/flysim/src/metrics.rs index e4e4284..49d04a2 100644 --- a/services/flysim/crates/flysim/src/metrics.rs +++ b/services/flysim/crates/flysim/src/metrics.rs @@ -43,6 +43,10 @@ pub struct Metrics { pub lag_ms: AtomicU64, /// 1 when the restore fell back past the newest candidate. pub restore_fallback: AtomicU64, + /// Snapshots published on the feed bus (`FLY_FEED_VIA=bus`); 0 in direct mode. + pub bus_published: AtomicU64, + /// Snapshots the feed bus refused or could not take; each one is skipped, never retried. + pub bus_publish_failures: AtomicU64, } impl Metrics { @@ -114,6 +118,20 @@ pub fn render(metrics: &Metrics, snapshot: &Snapshot, now_wall_ms: u64) -> Strin "Snapshots superseded before a slow client could be sent them.", Metrics::get(&metrics.feed_dropped), ); + metric( + &mut out, + "fly_bus_published_total", + "counter", + "Snapshots published on the feed bus (FLY_FEED_VIA=bus).", + Metrics::get(&metrics.bus_published), + ); + metric( + &mut out, + "fly_bus_publish_failures_total", + "counter", + "Snapshots the feed bus did not take; skipped, like any superseded snapshot.", + Metrics::get(&metrics.bus_publish_failures), + ); metric( &mut out, "fly_snapshots_published_total",