feat(tooling): add event-monitor arrival-time dashboard#538
Conversation
Standalone Rust/axum collector + vanilla-JS dashboard that dials the /lean/v0/events SSE stream of several ethlambda nodes, timestamps each event on one collector clock, and visualizes arrival offset within the slot (rolling per-node beeswarm) and inter-node propagation delay per block/aggregate/head (a second delta beeswarm, delay behind the first node to see each id). Own Cargo workspace with no ethlambda-crate deps; speaks only the documented SSE/HTTP wire shape frozen in CONTRACT.md. Rolling window is adjustable live from the header, a fresh page load backfills recent history via GET /api/history so it is never blank, and ?demo=1 runs the frontend fully offline on synthetic data.
🤖 Kimi Code ReviewOverall: This is a well-architected standalone monitoring tool with clear separation of concerns, good documentation ( 1. Critical: Invalid dependency versionFile: toml = "1.1.3"Issue: The Fix: toml = "0.8"2. Code correctness & robustnessFile: let url = format!(
"{}/lean/v0/events?topics={}",
node.url.trim_end_matches('/'),
topics.join(",")
);Issue: Topics are not URL-encoded. If a topic name contains reserved characters (e.g., Fix: Use let url = format!("{}/lean/v0/events", node.url.trim_end_matches('/'));
let url = reqwest::Url::parse_with_params(&url, &[("topics", topics.join(","))])?;File: axum::serve(listener, app).await?;Issue: No graceful shutdown handling. The server will not respond to SIGINT/SIGTERM cleanly, potentially dropping active SSE connections. Fix: Add signal handling: let server = axum::serve(listener, app);
let shutdown = tokio::signal::ctrl_c();
tokio::select! {
result = server => result?,
_ = shutdown => tracing::info!("shutting down"),
}File: Issue: No validation that Fix: Add validation after loading: if config.nodes.is_empty() {
return Err(anyhow::anyhow!("config must specify at least one node"));
}File: let serve_dir = ServeDir::new(static_dir).fallback(ServeFile::new(index_html));Issue: Fix: Resolve to absolute path: let static_dir = std::fs::canonicalize(static_dir)?;3. Security considerationsFile: Fix: Add use tower_http::cors::CorsLayer;
// ...
Router::new()
.layer(CorsLayer::permissive()) // or restricted to specific origins4. Performance & correctness observationsFile: let canonical = serde_json::to_string(&input).expect("aggregate id input is always valid JSON");Observation: The File: i64::try_from(duration.as_millis()).unwrap_or(i64::MAX)Good: Correctly handles the theoretical overflow case. File: const HISTORY_MAX_EVENTS: usize = 50_000;Good: Hard cap prevents unbounded memory growth during attestation floods. File: rate.textContent = `${status.events_per_sec.toFixed(1)}/s`;Issue: If the backend sends Fix: Add a guard: if (Number.isFinite(status.events_per_sec)) {
rate.textContent = `${status.events_per_sec.toFixed(1)}/s`;
} else {
rate.textContent = "—";
}5. TestingFile: Good: Integration test with a fake SSE server is well-written and covers the normalization pipeline end-to-end. Suggestion: Add a test for the pruning logic in Summary of required changes
The rest are suggestions for hardening. The architecture is sound and the separation between the Rust collector and JS frontend via Automated review by Kimi (Moonshot AI) · kimi-k2.5 · custom prompt |
🤖 Codex Code Review
No consensus-path code is touched here, so fork choice, attestation validation, STF, XMSS, and SSZ correctness are not directly affected by this PR. I couldn’t run Automated review by OpenAI Codex · gpt-5.4 · custom prompt |
🤖 Claude Code ReviewReview: PR 538 —
|
Greptile SummaryThis PR adds a standalone live event-arrival dashboard for ethlambda nodes. The main changes are:
Confidence Score: 4/5The default connection path and out-of-order history pruning need fixes before merging.
tooling/event-monitor/src/config.rs and tooling/event-monitor/src/hub.rs
|
| Filename | Overview |
|---|---|
| tooling/event-monitor/src/config.rs | Defines collector defaults, including a topic set that the current upstream endpoint rejects. |
| tooling/event-monitor/src/collector.rs | Adds per-node SSE collection, normalization, status reporting, and reconnect backoff. |
| tooling/event-monitor/src/hub.rs | Adds merged broadcasting and bounded history, but slot pruning assumes events arrive in slot order. |
| tooling/event-monitor/src/model.rs | Adds event payload normalization and stable aggregate identity generation. |
| tooling/event-monitor/src/timing.rs | Adds slot-geometry bootstrap and arrival-offset calculations. |
| tooling/event-monitor/src/server.rs | Adds metadata, history, live SSE, and static dashboard routes. |
| tooling/event-monitor/web/app.js | Adds stream-first startup, history backfill, overlap deduplication, and status rendering. |
| tooling/event-monitor/web/beeswarm.js | Adds the rolling per-node arrival-offset visualization. |
| tooling/event-monitor/web/propagation.js | Adds grouped per-node propagation-delta visualization. |
Sequence Diagram
sequenceDiagram
participant N as ethlambda nodes
participant C as Event collectors
participant H as Hub and history
participant A as Axum API
participant B as Browser
C->>N: "GET /lean/v0/events?topics=..."
N-->>C: SSE events
C->>C: Timestamp and normalize
C->>H: Store then broadcast
B->>A: Open /stream
A->>H: Subscribe
B->>A: GET /api/history
A->>H: Read snapshot
A-->>B: History response
A-->>B: Live chain and status events
B->>B: Deduplicate and render
Prompt To Fix All With AI
Fix the following 2 code review issues. Work through them one at a time, proposing concise fixes.
---
### Issue 1 of 2
tooling/event-monitor/src/config.rs:57-63
**Default Subscription Is Rejected**
The current event endpoint rejects the entire request when any topic is unknown. With these defaults, nodes on `main` return HTTP 400 for `attestation` and `aggregate`, so the collector also loses supported `block` events and remains reconnecting/down. The default must use currently supported topics or the collector must retry with unsupported topics removed.
### Issue 2 of 2
tooling/event-monitor/src/hub.rs:74-80
**Late Events Escape Slot Pruning**
This loop stops at the first recent front element, so an old event received out of slot order remains at the back of the deque even when it is outside `retain_slots`. Repeated delayed events can occupy the 50,000-event cap and force valid in-window events out of history, leaving a refreshed dashboard with missing recent data.
```suggestion
self.events.retain(|event| {
self.max_slot.saturating_sub(event.slot) < self.retain_slots
});
```
Reviews (1): Last reviewed commit: "feat(tooling): add event-monitor arrival..." | Re-trigger Greptile
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| fn default_topics() -> Vec<String> { | ||
| vec![ | ||
| "block".to_string(), | ||
| "attestation".to_string(), | ||
| "aggregate".to_string(), | ||
| ] | ||
| } |
There was a problem hiding this comment.
The current event endpoint rejects the entire request when any topic is unknown. With these defaults, nodes on main return HTTP 400 for attestation and aggregate, so the collector also loses supported block events and remains reconnecting/down. The default must use currently supported topics or the collector must retry with unsupported topics removed.
Prompt To Fix With AI
This is a comment left during a code review.
Path: tooling/event-monitor/src/config.rs
Line: 57-63
Comment:
**Default Subscription Is Rejected**
The current event endpoint rejects the entire request when any topic is unknown. With these defaults, nodes on `main` return HTTP 400 for `attestation` and `aggregate`, so the collector also loses supported `block` events and remains reconnecting/down. The default must use currently supported topics or the collector must retry with unsupported topics removed.
How can I resolve this? If you propose a fix, please make it concise.| while let Some(front) = self.events.front() { | ||
| if self.max_slot.saturating_sub(front.slot) >= self.retain_slots { | ||
| self.events.pop_front(); | ||
| } else { | ||
| break; | ||
| } | ||
| } |
There was a problem hiding this comment.
This loop stops at the first recent front element, so an old event received out of slot order remains at the back of the deque even when it is outside retain_slots. Repeated delayed events can occupy the 50,000-event cap and force valid in-window events out of history, leaving a refreshed dashboard with missing recent data.
| while let Some(front) = self.events.front() { | |
| if self.max_slot.saturating_sub(front.slot) >= self.retain_slots { | |
| self.events.pop_front(); | |
| } else { | |
| break; | |
| } | |
| } | |
| self.events.retain(|event| { | |
| self.max_slot.saturating_sub(event.slot) < self.retain_slots | |
| }); |
Prompt To Fix With AI
This is a comment left during a code review.
Path: tooling/event-monitor/src/hub.rs
Line: 74-80
Comment:
**Late Events Escape Slot Pruning**
This loop stops at the first recent front element, so an old event received out of slot order remains at the back of the deque even when it is outside `retain_slots`. Repeated delayed events can occupy the 50,000-event cap and force valid in-window events out of history, leaving a refreshed dashboard with missing recent data.
```suggestion
self.events.retain(|event| {
self.max_slot.saturating_sub(event.slot) < self.retain_slots
});
```
How can I resolve this? If you propose a fix, please make it concise.1 participant
Summary
Adds
tooling/event-monitor/, a standalone live arrival-time dashboard forlean-consensus (ethlambda) nodes. It dials the
GET /lean/v0/eventsSSE streamof several nodes, timestamps each event on a single collector clock, and serves
a browser dashboard visualizing:
The rolling window is adjustable live from the header, and a fresh page load
backfills recent history from the collector so it is never blank.
Design
tooling/event-monitor/(empty[workspace]table); it is not a member of the parentethlambdaworkspace and does not affect the main build.CONTRACT.mdis the frozen interface between the Rust/axum backend (src/) and the vanilla-JS frontend (web/, no build step).?demo=1runs the frontend fully offline with synthetic data.This PR contains only the tooling. The RPC / event-bus changes that make a
node emit these events are out of scope here and ship in the chain-events series
(see below).
Required ethlambda API functionality
The monitor consumes chain events over SSE plus two bootstrap endpoints for slot
geometry. Dependency status:
GET /lean/v0/eventsmain?topics=<csv>mainGET /lean/v0/genesis,GET /lean/v0/config/specmainblock,head,justified_checkpoint,finalized_checkpointmainattestation,aggregatesafe_target,chain_reorgblock_gossipRequired for the default dashboard view: #534. Both panels default to the
block/attestation/aggregatetopics.blockandheadalready workagainst
main, butattestationandaggregateneed #534 before the rollingbeeswarm and the aggregate propagation view populate.
Optional enrichment: #533 (
safe_target,chain_reorg) and #535(
block_gossip). The collector parses these topics if a node emits them, butthe dashboard does not require them.
Until #534 lands, the monitor is usable with
topics = ["block", "head"]for ablock-only slot-arrival and propagation view.
Testing
cargo test --releaseintooling/event-monitor/: 29 passed (27 lib + 2 integration), 0 failed.?demo=1offline synthetic mode renders both panels with no live nodes.Run