feat(rpc): add GET /lean/v0/events SSE stream

[MegaRedHand]

Adds GET /lean/v0/events, a Server-Sent Events stream of the chain events introduced in #516: head, block, justified_checkpoint, finalized_checkpoint. Every subscriber currently receives all events; server-side ?topics= filtering lands in the next PR of the series.

Frames carry the topic on the SSE event: line and a flat JSON payload in data: (built from ChainEvent::topic(), no hardcoded name mapping). A client that reads too slowly skips the events it missed (debug-logged, best-effort; re-sync via the blocks endpoints) rather than back-pressuring the actor. Keep-alive comments hold idle connections open. Documented in docs/rpc.md.

Stacked on #516.

Has handler tests and passes clippy with -D warnings.

[github-actions]

🤖 Kimi Code Review

The PR implements an SSE endpoint for chain events with proper best-effort semantics and clean separation between the blockchain actor (publisher) and RPC (subscriber). No critical issues found.

Minor observations:

1. Redundant import (crates/net/rpc/src/events.rs:84)
   StreamExt is imported at line 23 but used fully qualified at line 84. Consider using body.next().await instead of tokio_stream::StreamExt::next(&mut body) for consistency.

2. Workspace dependency consistency (crates/net/rpc/Cargo.toml)
   tokio-stream and futures-core are added as direct version dependencies while the crate typically uses workspace-inherited deps (.workspace = true). If these are defined in the workspace root Cargo.toml, prefer:

   tokio-stream.workspace = true
   futures-core.workspace = true

3. Test assertion style (crates/net/rpc/src/events.rs:97-99)
   The assertion text.contains("event:head") || text.contains("event: head") is permissive. Axum's SSE formatter uses event: head\n (with space), so the first condition is unnecessary but harmless.

Security/Performance notes:

• Backpressure handling: The Lagged branch at lines 35–38 correctly implements best-effort semantics by skipping events rather than blocking the actor. This prevents a slow HTTP client from stalling the consensus event bus.
• Memory safety: BroadcastStream + filter_map properly handles the bounded channel backpressure. No risk of unbounded memory growth from slow consumers.
• JSON failure handling: Line 43 silently drops events that fail serialization (after logging). This is acceptable for telemetry but ensure ChainEvent serialization is infallible in practice (no custom maps with non-string keys, etc.).

Architecture verification:

• The EventBus clone at main.rs:256 correctly shares the sender between the blockchain actor and RPC layer. The Extension layer at lib.rs:73 provides the bus to handlers.
• The Infallible error type in the stream signature (line 29) is correct since the broadcast channel adapter handles errors internally via the Lagged variant.

The implementation follows Rust async best practices and integrates cleanly with the existing axum router structure.

---

Automated review by Kimi (Moonshot AI) · kimi-k2.5 · custom prompt

[github-actions]

🤖 Codex Code Review

1. crates/net/rpc/src/events.rs and crates/net/rpc/src/lib.rs: the SSE stream never terminates on node shutdown. axum::serve(...).with_graceful_shutdown(...) stops accepting new connections, but it still waits for active responses to finish; /lean/v0/events is intentionally infinite, so one connected client can keep the RPC server alive indefinitely during shutdown/restart. The handler needs a shutdown-aware stream branch or request-scoped cancellation so open SSE responses end when shutdown is cancelled.

2. crates/net/rpc/src/events.rs: lagged subscribers silently lose events and have no way to detect the gap. On BroadcastStreamRecvError::Lagged(_) the code logs locally and drops the item, but the client receives neither a sentinel event nor a closed stream. That makes the documented “re-sync after a gap” behavior hard to implement correctly, and downstream consumers can keep operating on a stale event sequence without realizing they missed a block/head/checkpoint transition.

Otherwise the wiring looks clean, and the flat JSON framing test is a good regression check for the earlier double-tagging bug.

I couldn’t run cargo test in this environment because rustup failed creating temp files under the read-only /home/runner/.rustup.

---

Automated review by OpenAI Codex · gpt-5.4 · custom prompt

[github-actions]

🤖 Claude Code Review

Review: feat(rpc): add GET /lean/v0/events SSE stream (PR 517)

Solid, minimal implementation that follows the actor → bus → RPC one-directional flow established in the stacked PR (#516). No correctness or security-blocking issues found. A few observations below.

Correctness / Design

• Subscribe-before-response race is handled correctly. get_events calls events.subscribe() synchronously before constructing the Sse response, and the test in events.rs:66-100 relies on that ordering (subscribe happens during oneshot(), before emit is called). This is correct given emit's receiver-count guard (crates/blockchain/src/events.rs:106) — a subscriber that connects after emit would simply miss that event, which is by design.
• Lagged-skip path is untested. The single test (events.rs:66-100) only covers the happy path (one event, no lag). The PR description and docs put a lot of weight on the "lagged client skips events" behavior (events.rs:33-40) — worth a follow-up test that fills the bounded channel past capacity and asserts the stream keeps running past a Lagged gap instead of ending, since that's the one subtle behavior a regression could silently break (e.g., if a future refactor mishandles BroadcastStreamRecvError and ends the stream instead of skipping).

Minor / Style

• crates/net/rpc/src/events.rs:34-42: the Some(Ok(Event::default()...inspect_err(...).ok()?)) construction is dense — mixing an early-return ? for None (serialization failure) inside a Some(Ok(...)) wrapper for the success case. Consider splitting into an explicit match or intermediate let event = ...; per the project's own "extract long expressions for clarity" convention (CLAUDE.md's Formatting Patterns section) — e.g.:

  let Ok(event) = Event::default().event(ev.topic().as_str()).json_data(&ev) else {
      tracing::warn!("Failed to serialize SSE chain event");
      return None;
  };
  Some(Ok(event))

• No metric for lagged/dropped SSE clients or active-subscriber count. The codebase leans heavily on Prometheus counters/gauges for exactly this kind of "best-effort, may silently drop" behavior (see the metrics conventions in CLAUDE.md). A lean_rpc_sse_events_lagged_total counter alongside the tracing::debug! at events.rs:37 would give operators visibility into how often clients are actually losing events, rather than relying on debug logs.
• /lean/v0/events has no cap on concurrent connections/subscribers — a client can open unbounded SSE connections, each holding a broadcast receiver alive indefinitely. This is consistent with the rest of the unauthenticated API surface (default bind 127.0.0.1), so not a new risk introduced by this PR, but worth keeping in mind if this port is ever exposed beyond localhost.

Everything else checks out

• EventBus cloning/threading in main.rs is correct: the actor gets one clone (publisher), the RPC layer gets another (each SSE connection subscribes its own receiver from it).
• Extension<EventBus> layering matches the existing pattern for aggregator/sync_status.
• Flat/untagged JSON payload with topic only on the event: line matches the documented contract and is exercised by the test's "not a data/event wrapper" assertion.
• Cargo.toml additions (tokio-stream with sync feature, futures-core) are minimal and appropriately scoped.
• Docs (docs/rpc.md) accurately describe the endpoint, payload shapes, and best-effort/lagged semantics.

---

Automated review by Claude (Anthropic) · sonnet · custom prompt

[greptile-apps]

Greptile Summary

This PR adds a live SSE stream for blockchain events. The main changes are:

• Adds GET /lean/v0/events with event names and flat JSON payloads.
• Connects the RPC server to the blockchain event bus.
• Adds keep-alives and best-effort handling for slow clients.
• Adds handler coverage and RPC documentation.

Confidence Score: 5/5

This looks safe to merge.

• No additional blocking issue was found in the updated code.

Important Files Changed

Filename	Overview
crates/net/rpc/src/events.rs	Adds the SSE handler, event framing, keep-alives, and handler coverage.
crates/net/rpc/src/lib.rs	Registers the events route and provides the event bus through an extension.
bin/ethlambda/src/main.rs	Shares the blockchain event bus with the RPC server.
crates/net/rpc/Cargo.toml	Adds the stream dependencies required by the SSE endpoint.
docs/rpc.md	Documents the endpoint, event payloads, delivery behavior, and content type.

_{Reviews (3): Last reviewed commit: "feat(rpc): add GET /lean/v0/events SSE s..." | Re-trigger Greptile}

[github-actions]

🤖 Kimi Code Review

Overall Assessment: Clean, idiomatic implementation of SSE streaming for consensus events. Correctly handles backpressure via bounded broadcast channel semantics and implements best-effort delivery as specified.

Minor Issues & Suggestions:

1. Rust Version Compatibility (Line 47, crates/net/rpc/src/events.rs)

• Result::inspect_err requires Rust 1.76+. Verify rust-toolchain.toml or CI supports this; otherwise use map_err with logging.

// Alternative for older Rust versions:
.json_data(&ev)
.map_err(|err| tracing::warn!(%err, "Failed to serialize SSE chain event"))
.ok()?

2. Test Race Condition Risk (Lines 67-97, crates/net/rpc/src/events.rs)

• The test assumes the subscription is established before events.emit() is called. While oneshot drives the handler to completion before returning, the BroadcastStream subscription happens when the handler runs. This is correct, but fragile if the handler structure changes. Consider adding a small yield or explicit synchronization if tests become flaky.

3. Connection Resource Limits

• No per-IP connection limits or global SSE connection caps are implemented. A malicious client could exhaust memory by opening thousands of SSE connections (each holds a broadcast::Receiver). Consider adding Tower middleware (e.g., tower::limit::ConcurrencyLimit) in build_api_router or documenting this as a known limitation for operators to handle via reverse proxy (nginx/haproxy).

4. Serialization Failure Handling (Line 47)

• Failed JSON serialization silently drops the event (returns None). This is acceptable for best-effort delivery, but ensure ChainEvent types never contain non-serializable fields (e.g., raw bytes without hex encoding). The current Head/Block/Checkpoint events appear to use SSZ-compatible types, so this is low risk.

5. Documentation Accuracy (Line 27, docs/rpc.md)

• The SSE event: line format is checked in tests for both "event:head" and "event: head" (with space). The SSE spec requires a space after the colon (event: head\n). The test's defensive check is fine, but the docs show event: head which is correct.

Security & Consensus Considerations:

• Data exposure: Events contain only public consensus data (heads, checkpoints, blocks). No validator private keys or sensitive P2P metadata is exposed. Safe.
• Fork choice safety: This is a read-only endpoint; no mutation of fork choice or attestation pool. Safe.

Performance:

• Memory: Each SSE client holds a broadcast::Receiver slot in the channel buffer. With default capacity (16) and slow clients, memory is bounded correctly.
• CPU: JSON serialization per event is acceptable for consensus-layer event frequency (slots every 12s).

Acknowledgments:

• Proper use of axum::response::Sse with KeepAlive.
• Correct handling of BroadcastStreamRecvError::Lagged (Item 35-42) — this is critical for preventing slow clients from back-pressuring the blockchain actor.
• Good test coverage for the "flat payload" regression (double-tag bug feat(rpc): add GET /lean/v0/events SSE stream (head/block/finalized_checkpoint) #460).

Verdict: LGTM. Address the Rust version check for inspect_err if necessary, and consider documenting the need for reverse-proxy connection limits for production deployments.

---

Automated review by Kimi (Moonshot AI) · kimi-k2.5 · custom prompt

[github-actions]

🤖 Codex Code Review

1. Medium: the new SSE route introduces an unbounded long-lived public connection surface with no app-level throttling or cap. crates/net/rpc/src/events.rs creates a dedicated broadcast::Receiver per client, and crates/net/rpc/src/lib.rs wires it into the normal RPC server without any concurrency limit or opt-in gate. In practice, an attacker can hold many idle /lean/v0/events connections open and turn routine chain activity into O(subscribers) wakeups/serialization work, or just exhaust file descriptors. For a consensus client, I’d want a per-route connection limit, an explicit feature flag, or both.

I didn’t find other correctness issues in the diff. This PR doesn’t alter fork choice, attestation validation, finalization, state transition, XMSS, or SSZ logic; it only exposes existing ChainEvent emissions over RPC, and the ordering in the existing publisher path still looks consistent.

I couldn’t run cargo validation here because the environment blocks writes under ~/.cargo / ~/.rustup, so this is a manual review only.

---

Automated review by OpenAI Codex · gpt-5.4 · custom prompt

[github-actions]

🤖 Claude Code Review

Consistent with existing conventions. I have enough context to write the review now.

Review: feat(rpc): add GET /lean/v0/events SSE stream

Clean, well-scoped addition on top of the #516 event bus groundwork. The handler is a straightforward one-directional subscribe-and-forward, the flat-payload framing matches the bus's untagged design, the receiver-count no-op guard is correctly preserved (the RPC-held EventBus clone is a Sender, not a Receiver, so it doesn't itself count toward receiver_count()), and the test correctly avoids the subscribe-race by subscribing synchronously inside the handler before the oneshot response is observed. Docs are thorough and match the implementation.

A few things worth a look:

1. SSE connections aren't wired into graceful shutdown (crates/net/rpc/src/events.rs:28-49, bin/ethlambda/src/main.rs:337-339)

axum::serve(...).with_graceful_shutdown(...) stops accepting new connections on cancellation but doesn't forcibly terminate in-flight ones — it waits for each connection to finish naturally. The /lean/v0/events stream never ends on its own (the BroadcastStream only terminates when the EventBus's Sender is fully dropped, which won't happen while the RPC layer holds its extension clone alive), so any client that keeps a long-lived SSE connection open will make rpc_handle.await (main.rs:339) block past the intended graceful-shutdown point. In practice the operator has to hit ctrl-C three times to force std::process::exit(1) (main.rs:326-328) to actually restart the node. Worth confirming this is the accepted behavior, or having the handler select! against the shutdown token (or an axum::extract::Request cancellation) so streams end on shutdown rather than relying on the force-exit fallback.

2. No cap on concurrent SSE subscribers

Each connection holds its own broadcast::Receiver plus a live task for the lifetime of the connection, with no limit on how many clients can attach to /lean/v0/events at once. Since the endpoint is unauthenticated (matching the rest of the RPC API), an attacker could open many parallel long-lived connections to exhaust file descriptors/memory. Given ethlambda's minimalist scope this may be an accepted tradeoff consistent with the rest of the API, but SSE amplifies it relative to typical short-lived request handlers, so it's worth a deliberate call rather than an implicit one.

3. Minor readability nit — the filter_map closure (events.rs:43-47)

Some(Ok(Event::default()
    .event(ev.topic().as_str())
    .json_data(&ev)
    .inspect_err(|err| tracing::warn!(%err, "Failed to serialize SSE chain event"))
    .ok()?))

The ? here relies on the closure's Option<Result<_, Infallible>> return type to short-circuit through Some(Ok(...)) on serialization failure — correct, but easy to misread at a glance (looks like it could early-return Err, but Infallible makes that impossible, and the actual short-circuit is a filtered-out None). A match or let Ok(event) = ... else { return None } would make the control flow more obvious. Low priority since ChainEvent's fields (u64/H256) can't realistically fail json_data serialization anyway.

4. New deps skip workspace management (crates/net/rpc/Cargo.toml:29-30)

tokio-stream = { version = "0.1", features = ["sync"] }
futures-core = "0.3"

Added as crate-local pins rather than .workspace = true like the rest of the block above them. The crate already has a couple of non-workspace deps (axum, dev-dep tower/http-body-util), so this isn't a new pattern, but if tokio-stream/futures-core are likely to be reused by another crate later, centralizing now avoids version drift. Not blocking.

Nice-to-have (not required)

Test coverage is a single happy-path case (head event, flat payload). No test exercises the Lagged skip-and-continue branch or multiple concurrent subscribers each getting independent streams — both are core to the "best-effort, per-subscriber" design called out in the docs/comments, so a regression there wouldn't be caught by CI.

---

Automated review by Claude (Anthropic) · sonnet · custom prompt

[github-actions]

🤖 Codex Code Review

No findings.

The diff stays out of consensus-critical execution paths: it only wires the existing EventBus into RPC and exposes it as a best-effort SSE stream. I don’t see a correctness regression in fork choice, attestation handling, justification/finalization, state transition, XMSS verification, or SSZ handling from these changes.

Residual risk: I couldn’t run cargo test here because rustup tried to write under /home/runner/.rustup and hit a read-only filesystem. The added unit test covers SSE framing, but runtime behavior under lagged subscribers and disconnect/reconnect remains validated only by inspection, not by an executable test in this environment.

---

Automated review by OpenAI Codex · gpt-5.4 · custom prompt

[github-actions]

🤖 Kimi Code Review

The PR adds a clean, well-tested SSE endpoint for chain events. Implementation follows Axum best practices and correctly handles slow clients via lag detection.

Minor observations:

1. Serialization failure handling (crates/net/rpc/src/events.rs:38-42):
   Events that fail JSON serialization are silently dropped for that subscriber (logged at warn). While acceptable for best-effort streaming, consider if a synthetic error event should be sent instead to signal data loss to the client.

2. Dependency granularity (crates/net/rpc/Cargo.toml):
   futures-core is added solely for the Stream trait. You could instead use tokio_stream::Stream (re-exported from futures-core) to avoid the extra explicit dependency, though the current approach is explicit about trait requirements.

3. Test robustness (crates/net/rpc/src/events.rs:86-88):
   The assertion text.contains("event:head") || text.contains("event: head") is permissive. Axum’s Event::event always inserts a space after the colon per SSE spec, so you could tighten this to just "event: head" unless you're intentionally supporting non-compliant clients.

Security & Performance:

• DoS vector: Each SSE connection creates a BroadcastStream receiver. With no limit on concurrent connections, memory grows linearly with receivers (each tracks its own lag cursor). Consider adding a connection limit or timeout middleware in a follow-up if this endpoint is exposed to untrusted clients.
• Clone cost: Confirmed EventBus is cloned between the blockchain actor and RPC router. Ensure EventBus is a cheap handle (wrapping Arc<broadcast::Sender<_>>), which appears to be the case.

Consensus considerations:

• Event ordering relies on the broadcast channel’s FIFO guarantees, which is correct for single-publisher (blockchain actor) semantics.
• The “flat payload” test correctly prevents regression of the double-tag bug (feat(rpc): add GET /lean/v0/events SSE stream (head/block/finalized_checkpoint) #460) where events were previously nested under data/event keys.

Verdict: LGTM. The lag-handling logic is correct for a best-effort consensus event stream.

---

Automated review by Kimi (Moonshot AI) · kimi-k2.5 · custom prompt

[github-actions]

🤖 Claude Code Review

Review: feat(rpc): add GET /lean/v0/events SSE stream (PR 517)

Overall this is a clean, well-scoped addition — the flat-JSON/out-of-band-topic design correctly avoids the double-tagging bug referenced in the tests (#460), the actor→bus→SSE flow stays one-directional as documented, and the lagged-client handling matches the bus's documented best-effort contract. A few things worth considering before merge:

1. Unbounded concurrent SSE connections (crates/net/rpc/src/events.rs:26-45)

Every request to /lean/v0/events subscribes a new broadcast receiver and holds the connection open indefinitely (only closed by client disconnect). There's no cap on the number of simultaneous subscribers. This is the first long-lived endpoint in the API — all the others are short request/response cycles — so it introduces a new resource-exhaustion surface: an unauthenticated client (or script) can open many connections and hold file descriptors/memory indefinitely. Given this is a consensus node exposed on a configurable port, worth at least documenting the exposure or considering a connection-count guard/timeout, even if full auth is out of scope for this PR.

2. Graceful shutdown can hang on a connected SSE client

bin/ethlambda/src/main.rs:339 awaits rpc_handle before logging "Shutdown complete", and start_rpc_server uses axum::serve(...).with_graceful_shutdown(...) (crates/net/rpc/src/lib.rs:86-104), which waits for in-flight connections to finish. Since the new stream is intentionally infinite (keep_alive forever until the client disconnects), a single open /lean/v0/events subscriber will block node shutdown until the operator presses ctrl+C three times to force std::process::exit(1) (main.rs:316-329). Previously every endpoint completed quickly, so shutdown was effectively instant; this PR changes that assumption. Worth a comment near the shutdown path (or a shutdown-aware stream) noting the new behavior, since it's easy to blame a future shutdown hang on something else.

3. Dense control flow in the event-mapping closure (crates/net/rpc/src/events.rs:36-42)

Some(Ok(Event::default()
    .event(ev.topic().as_str())
    .json_data(&ev)
    .inspect_err(|err| tracing::warn!(%err, "Failed to serialize SSE chain event"))
    .ok()?))

This is correct (the ? early-returns None from the filter_map closure on serialization failure) but relies on a ? inside a Some(Ok(...)) wrapper, which is easy to misread at a glance. A small match would read more clearly:

let ev = match Event::default().event(ev.topic().as_str()).json_data(&ev) {
    Ok(ev) => ev,
    Err(err) => {
        tracing::warn!(%err, "Failed to serialize SSE chain event");
        return None;
    }
};
Some(Ok(ev))

4. Test coverage gap

Only the happy path (single head event, read immediately after subscribing) is tested. The Lagged skip-and-continue branch and the json_data serialization-failure branch — both called out explicitly in the code's own comments as intentional, important behavior — aren't exercised by any test. Consider adding a test that overflows the broadcast channel to assert the stream continues past a Lagged error rather than terminating.

Minor

• Cargo.toml additions (tokio-stream, futures-core) use ad hoc versions rather than workspace deps, but this matches existing crate-local style for axum/tower/http-body-util in this same file, so it's consistent, not a regression.
• Docs (docs/rpc.md) accurately reflect the event payload shapes and framing — good.

No correctness issues in the fork-choice/state-transition/consensus paths — this PR only wires a read-only observer onto an already-existing event bus and doesn't touch consensus logic.

---

Automated review by Claude (Anthropic) · sonnet · custom prompt