From c705c7428960d287f190cad6bfbf57c193da31df Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Mon, 24 Aug 2026 19:22:33 +0400 Subject: [PATCH] perf(runtime): bound telemetry and hot-path work Cap and fairly distribute live snapshots, serialize dashboard refreshes, reduce telemetry fanout copies, and use a faster codec registry map. --- Cargo.lock | 1 + crates/dashboard/README.md | 2 +- crates/dashboard/src/control_plane_page.html | 31 +++++--- crates/dashboard/src/live_explorer.rs | 82 ++++++++++++++++++-- crates/dashboard/src/live_explorer_page.html | 13 ++++ crates/telemetry/src/endpoint.rs | 62 +++++++++------ crates/telemetry/src/store.rs | 3 + crates/transport/Cargo.toml | 1 + crates/transport/src/codec.rs | 5 +- 9 files changed, 157 insertions(+), 43 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1e3e6e3..cc272c1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4349,6 +4349,7 @@ dependencies = [ "ed25519-dalek 2.2.0", "proptest", "rand_core 0.6.4", + "rustc-hash", "serde", "serde_json", "swactor", diff --git a/crates/dashboard/README.md b/crates/dashboard/README.md index 882fe28..f5c5ce5 100644 --- a/crates/dashboard/README.md +++ b/crates/dashboard/README.md @@ -20,7 +20,7 @@ is hard-capped at 50. - `GET /api/frames` — recent raw frame window - `GET /api/views` — registered view metadata - `GET /view/telemetry/live` — generic live explorer over retained and incoming telemetry frames -- `GET /api/view/telemetry/live` — 2,000 frames per stream/channel, newest lifetime per node, all live streams plus 50 stale streams +- `GET /api/view/telemetry/live` — bounded bootstrap snapshot: up to 500 recent frames and 256 KiB of raw payload, distributed across channels; retention remains 500 frames per stream/channel for live inspection - `GET /view/fleet` — fused control-plane page (machine + actors per node) - `GET /api/view/fleet` — live/stale pools with per-node machine and roster snapshot - `GET /api/view/fleet/detail?stream=&actor=` — bounded per-actor dossier detail (diet, history, sampled receipts) diff --git a/crates/dashboard/src/control_plane_page.html b/crates/dashboard/src/control_plane_page.html index 317f284..8778c16 100644 --- a/crates/dashboard/src/control_plane_page.html +++ b/crates/dashboard/src/control_plane_page.html @@ -173,6 +173,7 @@ let rosterSort = { key: 'address', direction: 'asc' }; let hardwareSource = 'node'; let lastSnapshot = null; let detailTimer = null; +let refreshInFlight = false; // A1: last-rendered HTML per region. A poll that yields identical markup // must not swap innerHTML — the swap destroyed hover/selection/presses and // restarted animations every second even in a fully converged steady state. @@ -256,13 +257,18 @@ function pushUrl() { } async function refresh() { - let data; + if (refreshInFlight) return; + refreshInFlight = true; try { const response = await fetch('/api/view/fleet'); - data = await response.json(); - } catch { return; } - lastSnapshot = data; - render(); + const data = await response.json(); + lastSnapshot = data; + render(); + } catch { + // Keep the last good snapshot while the next scheduled refresh retries. + } finally { + refreshInFlight = false; + } } function render() { @@ -681,15 +687,20 @@ function stopDetailPolling() { function startDetailPolling() { stopDetailPolling(); + let pollInFlight = false; const poll = async () => { - if (!selectedStream || !selectedActor) return; - let detail; + if (!selectedStream || !selectedActor || pollInFlight) return; + pollInFlight = true; try { const response = await fetch(`/api/view/fleet/detail?stream=${encodeURIComponent(selectedStream)}&actor=${encodeURIComponent(selectedActor)}`); if (!response.ok) { stopDetailPolling(); return; } - detail = await response.json(); - } catch { return; } - renderDossier(detail); + const detail = await response.json(); + renderDossier(detail); + } catch { + // Keep the current dossier while the next scheduled refresh retries. + } finally { + pollInFlight = false; + } }; poll(); detailTimer = setInterval(poll, POLL_MS); diff --git a/crates/dashboard/src/live_explorer.rs b/crates/dashboard/src/live_explorer.rs index 05bfc25..775f3ec 100644 --- a/crates/dashboard/src/live_explorer.rs +++ b/crates/dashboard/src/live_explorer.rs @@ -9,7 +9,9 @@ use crate::FrameEvent; use crate::view::DashboardView; const LIVE_EXPLORER_HTML: &str = include_str!("live_explorer_page.html"); -const FRAME_HISTORY_CAP: usize = 2_000; +const FRAME_HISTORY_CAP: usize = 500; +const SNAPSHOT_FRAME_CAP: usize = 500; +const SNAPSHOT_PAYLOAD_BYTE_CAP: usize = 256 * 1024; const LIVE_TTL: Duration = Duration::from_secs(8); const STALE_STREAM_CAP: usize = 50; @@ -65,12 +67,39 @@ impl LiveTelemetryExplorer { fn snapshot_at(&self, now: Instant) -> Value { let mut state = self.state.write(); prune_stale(&mut state.streams, now); - let mut frames = state + + // Take recent frames round-robin across channels so a busy channel + // cannot crowd quiet channels out of the bounded bootstrap snapshot. + let mut channels = state .streams .values() .flat_map(|stream| stream.channels.values()) - .flat_map(|frames| frames.iter()) + .map(|frames| frames.iter().rev()) .collect::>(); + let mut frames = Vec::with_capacity(SNAPSHOT_FRAME_CAP.min(channels.len())); + let mut payload_bytes = 0; + 'snapshot: loop { + let mut found_frame = false; + for channel in &mut channels { + let Some(frame) = channel.next() else { + continue; + }; + found_frame = true; + if frame.payload.len() > SNAPSHOT_PAYLOAD_BYTE_CAP - payload_bytes { + continue; + } + payload_bytes += frame.payload.len(); + frames.push(frame.clone()); + if frames.len() == SNAPSHOT_FRAME_CAP { + break 'snapshot; + } + } + if !found_frame || payload_bytes == SNAPSHOT_PAYLOAD_BYTE_CAP { + break; + } + } + drop(state); + frames.sort_by(|left, right| { left.stream .node @@ -138,6 +167,16 @@ mod tests { for position in 0..=FRAME_HISTORY_CAP as u64 { ingest(&view, &stream, "busy", position); } + { + let state = view.state.read(); + let busy = &state.streams.values().next().expect("stream").channels["busy"]; + assert_eq!(busy.len(), FRAME_HISTORY_CAP); + assert_eq!(busy.front().map(|frame| frame.position), Some(1)); + assert_eq!( + busy.back().map(|frame| frame.position), + Some(FRAME_HISTORY_CAP as u64) + ); + } let snapshot = view.snapshot_json(); let frames = snapshot["frames"].as_array().expect("frames array"); @@ -151,11 +190,7 @@ mod tests { .collect::>(); assert_eq!(quiet.len(), 1, "a quiet channel remains discoverable"); - assert_eq!(busy.len(), FRAME_HISTORY_CAP); - assert_eq!( - busy.first().and_then(|frame| frame["position"].as_u64()), - Some(1) - ); + assert_eq!(busy.len(), SNAPSHOT_FRAME_CAP - quiet.len()); assert_eq!( busy.last().and_then(|frame| frame["position"].as_u64()), Some(FRAME_HISTORY_CAP as u64) @@ -201,6 +236,37 @@ mod tests { assert!(streams.contains("node-59")); } + #[test] + fn snapshot_payload_is_bounded() { + let view = LiveTelemetryExplorer::default(); + let stream = StreamId::new(NodeId::new("node-a"), Lifetime(1)); + let payload = vec![7; SNAPSHOT_PAYLOAD_BYTE_CAP / 2]; + for (position, channel) in ["first", "second", "third"].into_iter().enumerate() { + let event = FrameEvent { + stream: crate::StreamEvent { + node: stream.node.as_str().to_owned(), + life: stream.life.0, + origin: None, + label: None, + }, + channel: channel.to_owned(), + position: position as u64, + payload: payload.clone(), + }; + view.ingest_at(&event, Instant::now()); + } + + let snapshot = view.snapshot_json(); + let frames = snapshot["frames"].as_array().expect("frames array"); + let payload_bytes = frames + .iter() + .map(|frame| frame["payload"].as_array().expect("payload").len()) + .sum::(); + assert!(frames.len() <= SNAPSHOT_FRAME_CAP); + assert!(payload_bytes <= SNAPSHOT_PAYLOAD_BYTE_CAP); + assert_eq!(frames.len(), 2, "the payload byte cap bounds the snapshot"); + } + fn test_frame(position: u64) -> Frame { Frame::new( ChannelId(position as u32 + 1), diff --git a/crates/dashboard/src/live_explorer_page.html b/crates/dashboard/src/live_explorer_page.html index 079adc7..4347073 100644 --- a/crates/dashboard/src/live_explorer_page.html +++ b/crates/dashboard/src/live_explorer_page.html @@ -147,6 +147,8 @@ RISK: terminal cosplay if decoration creeps past data; held by the palette law. seq: 0, paintQueued: false, paintHandle: 0, + reconciling: false, + reconcilePending: false, }; const dirty = { @@ -523,6 +525,11 @@ RISK: terminal cosplay if decoration creeps past data; held by the palette law. state.messageVersion++; updateMetrics(); } + if (state.reconciling) { + state.reconcilePending = true; + return; + } + state.reconciling = true; const messageVersion = state.messageVersion; const startedAt = performance.now(); try { @@ -554,6 +561,12 @@ RISK: terminal cosplay if decoration creeps past data; held by the palette law. updateMetrics(); renderDetail(); } + } finally { + state.reconciling = false; + if (state.reconcilePending) { + state.reconcilePending = false; + reconcileSnapshot({ notice: 'Catching up after concurrent telemetry updates.' }); + } } } diff --git a/crates/telemetry/src/endpoint.rs b/crates/telemetry/src/endpoint.rs index 98124f5..539d70b 100644 --- a/crates/telemetry/src/endpoint.rs +++ b/crates/telemetry/src/endpoint.rs @@ -112,7 +112,7 @@ impl TelemetrySubscription { } pub fn drain_available(&self) -> Vec { - let mut out = Vec::new(); + let mut out = Vec::with_capacity(self.rx.len()); while let Ok(event) = self.rx.try_recv() { out.push(event); } @@ -137,6 +137,31 @@ struct FanoutReport { disconnected: bool, } +fn publish_to_target( + target: FanoutTarget, + events: impl IntoIterator, +) -> (usize, Option) { + let mut delivered = 0; + let mut dropped = 0; + let mut disconnected = false; + for event in events { + match target.tx.try_send(event) { + Ok(()) => delivered += 1, + Err(TrySendError::Full(_)) => dropped += 1, + Err(TrySendError::Disconnected(_)) => { + dropped += 1; + disconnected = true; + } + } + } + let report = (dropped > 0 || disconnected).then_some(FanoutReport { + id: target.id, + dropped, + disconnected, + }); + (delivered, report) +} + struct FanoutState { next_id: u64, subscribers: BTreeMap, @@ -236,11 +261,12 @@ impl DeliveryFanout { if events.is_empty() { return EndpointTick::default(); } + let drained = events.len(); // Snapshot sender handles while holding the subscriber map lock, then // deliver outside the lock so large batches or slow subscribers do not // block subscribe/snapshot control-plane operations. - let (targets, subscribers) = { + let (mut targets, subscribers) = { let state = self.state.lock().expect("telemetry fanout poisoned"); let subscribers = state.subscribers.len(); let targets = state @@ -256,7 +282,7 @@ impl DeliveryFanout { if targets.is_empty() { return EndpointTick { - drained: events.len(), + drained, subscribers: 0, ..EndpointTick::default() }; @@ -264,27 +290,19 @@ impl DeliveryFanout { let mut delivered = 0; let mut reports = Vec::new(); + let last_target = targets.pop().expect("nonempty fanout targets"); for target in targets { - let mut dropped = 0; - let mut disconnected = false; - for event in &events { - match target.tx.try_send(event.clone()) { - Ok(()) => delivered += 1, - Err(TrySendError::Full(_)) => dropped += 1, - Err(TrySendError::Disconnected(_)) => { - dropped += 1; - disconnected = true; - } - } - } - if dropped > 0 || disconnected { - reports.push(FanoutReport { - id: target.id, - dropped, - disconnected, - }); + let (target_delivered, report) = publish_to_target(target, events.iter().cloned()); + delivered += target_delivered; + if let Some(report) = report { + reports.push(report); } } + let (target_delivered, report) = publish_to_target(last_target, events); + delivered += target_delivered; + if let Some(report) = report { + reports.push(report); + } let dropped_for_subscribers = reports.iter().map(|report| report.dropped).sum::(); if !reports.is_empty() { @@ -302,7 +320,7 @@ impl DeliveryFanout { } EndpointTick { - drained: events.len(), + drained, delivered, dropped_for_subscribers: usize::try_from(dropped_for_subscribers).unwrap_or(usize::MAX), subscribers, diff --git a/crates/telemetry/src/store.rs b/crates/telemetry/src/store.rs index aa85292..a3cc9b9 100644 --- a/crates/telemetry/src/store.rs +++ b/crates/telemetry/src/store.rs @@ -143,6 +143,9 @@ impl Store { /// The stored stream for a node's life, creating an empty one if needed. pub fn stream_mut(&mut self, id: &StreamId) -> &mut StoredStream { + if self.streams.contains_key(id) { + return self.streams.get_mut(id).expect("stream checked as present"); + } self.streams.entry(id.clone()).or_default() } diff --git a/crates/transport/Cargo.toml b/crates/transport/Cargo.toml index 56972bb..22ed541 100644 --- a/crates/transport/Cargo.toml +++ b/crates/transport/Cargo.toml @@ -14,6 +14,7 @@ ed25519-dalek = { version = "2", features = ["std", "rand_core", "serde"] } rand_core = { version = "0.6", features = ["getrandom"] } serde = { version = "1", features = ["derive"] } serde_json = "1" +rustc-hash = "2" [dev-dependencies] proptest = "1" diff --git a/crates/transport/src/codec.rs b/crates/transport/src/codec.rs index 25c63cb..4849856 100644 --- a/crates/transport/src/codec.rs +++ b/crates/transport/src/codec.rs @@ -5,6 +5,7 @@ //! per-type encoders/decoders so a type-erased message can be put on the wire //! and a [`WireEnvelope`] can be turned back into a concrete message. +use rustc_hash::FxHashMap; use std::any::{Any, TypeId}; use std::collections::HashMap; use std::sync::Arc; @@ -150,7 +151,7 @@ type DecodeFn = Box Result, Error> + Send + /// [`register_decoder`](Self::register_decoder) (variant-multiplexing encode, /// fan-in decode), then shared read-only via `Arc`. pub struct CodecRegistry { - encoders: HashMap, + encoders: FxHashMap, decoders: HashMap, } @@ -163,7 +164,7 @@ impl Default for CodecRegistry { impl CodecRegistry { pub fn new() -> Self { Self { - encoders: HashMap::new(), + encoders: FxHashMap::default(), decoders: HashMap::new(), } }