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 keep-alive frames.
• Sends event topics separately from flat JSON payloads.
• Shares the event bus between the blockchain actor and RPC server.
• Adds handler coverage and RPC documentation.

Confidence Score: 5/5

This looks safe to merge.

• No new blocking issues were found in the updated code.
• The event bus, router extension, and serialization paths are wired consistently.

Important Files Changed

Filename	Overview
crates/net/rpc/src/events.rs	Adds the SSE handler, event framing, keep-alives, and payload test.
crates/net/rpc/src/lib.rs	Registers the events route and provides the shared event bus.
bin/ethlambda/src/main.rs	Connects one event bus to the blockchain publisher and RPC server.
crates/blockchain/src/events.rs	Adds direct serialization of chain events as flat JSON objects.

_{Reviews (4): Last reviewed commit: "Merge branch 'main' into feat/events-sse..." | 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

[github-actions]

🤖 Kimi Code Review

Overall Assessment: Clean, well-documented implementation of the SSE endpoint. The use of #[serde(untagged)] for flat JSON payloads is appropriate given the out-of-band topic discriminator, and the lagged-client handling aligns with the best-effort contract.

Minor Issues & Suggestions:

1. Silent event dropping on serialization failure
File: crates/net/rpc/src/events.rs, lines 43-45
If json_data fails (e.g., H256 serialization error), the event is silently dropped for that client after logging. Consider if this is acceptable or if you should emit an SSE comment: line or close the stream to signal data loss. Current behavior is reasonable for SSE, but document explicitly if clients should handle gaps via re-sync.

2. H256 serialization assumption
File: crates/net/rpc/src/events.rs (tests), lines 97-110
The test asserts "slot":3 and hex-encoded H256 values. Ensure ethlambda_types::H256 implements Serialize producing 0x-prefixed compact hex (not an array of bytes). If using ethereum-types, enable the serialize feature; if custom, verify the impl. The test covers this, but a unit test on H256 serialization would catch feature flag issues earlier.

3. Untagged enum ambiguity (acceptable but note)
File: crates/blockchain/src/events.rs, lines 58-71
Head, JustifiedCheckpoint, and FinalizedCheckpoint serialize to identical JSON shapes ({"slot":..., "block":..., "state":...}). This is safe because the SSE event: line discriminates them, but ensure no downstream consumer deserializes these without the topic context (the comment correctly notes Deserialize is intentionally omitted).

4. Channel capacity visibility
File: crates/net/rpc/src/events.rs, line 35
The broadcast channel capacity (16 in tests) isn't visible in this diff. If the production capacity is too small (e.g., <128), high event rates during sync could cause frequent lag. Verify the EventBus constructor uses an appropriate buffer size for mainnet event volumes.

5. Clone semantics of EventBus
File: bin/ethlambda/src/main.rs, line 258
events.clone() implies EventBus is Clone (likely Arc-wrapped). Confirm this is shallow cloning the sender, not the channel state. If EventBus wraps broadcast::Sender, this is correct (tokio's sender is refcounted).

6. Test fragility
File: crates/net/rpc/src/events.rs, lines 103-106
The assertion text.contains("event:head") || text.contains("event: head") is permissive but could mask formatting changes. Axum consistently uses event: head\n, so prefer an exact match on the Event struct's output if possible, or use assert!(text.contains("event: ")) to be robust.

Security/DoS: No rate limiting on the SSE endpoint is visible. Consider adding a connection limit or maximum subscriber count to the EventBus to prevent DoS via slowloris SSE clients exhausting the broadcast channel buffer (each subscriber gets a slot in the channel).

Correctness: The lagged-client handling (skipping gaps) correctly implements the documented best-effort contract. The BroadcastStreamRecvError::Lagged branch is appropriately logged at debug level.

---

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

[github-actions]

🤖 Codex Code Review

Findings

1. bin/ethlambda/src/main.rs:256-305 with crates/blockchain/src/events.rs:99-123 keeps an extra broadcast::Sender alive inside the RPC layer. That means if the blockchain actor dies or drops its publisher unexpectedly, /lean/v0/events receivers will not see EOF; they will just sit on keep-alives forever because the router still owns a sender clone. It also weakens the “blockchain actor is the sole publisher” invariant at the type level. I’d split this into a publish-only handle for the actor and a subscribe-only handle/factory for RPC.

2. crates/blockchain/src/events.rs:47-71 and docs/rpc.md:91-98 describe the stream as beacon-compatible, but the checkpoint topics still serialize {slot, block, state} instead of the beacon API’s checkpoint shape centered on epoch (and they also omit other standard fields). That will break consumers that treat this as a beacon eventstream. Either make the wire format actually beacon-compatible before exposing it, or document it as an ethlambda-specific SSE contract.

3. crates/net/rpc/src/events.rs:91-96 has a brittle test: it assumes the first body chunk contains a whole SSE frame. Hyper/axum do not guarantee frame-to-chunk alignment, so this can flap across runtime/library changes. Reading until the \n\n frame terminator would make the test deterministic.

No fork-choice, attestation, justification/finalization, STF, XMSS, or SSZ logic changed in this diff, so I did not find a consensus-path regression beyond the new RPC/event-surface concerns above.

I couldn’t run cargo check or cargo test in this environment because the workspace depends on the git-sourced leansig crate and network access is blocked here.

---

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)

Summary

This adds the SSE transport (crates/net/rpc/src/events.rs) on top of the chain-event bus from PR #516, plus wiring in main.rs/lib.rs and docs. The implementation is clean: strictly one-directional (actor → bus → SSE), correct handling of BroadcastStreamRecvError::Lagged (skip-and-log rather than terminate the stream), and the Serialize-only/untagged design on ChainEvent is well-reasoned (no Deserialize ambiguity risk since serialization never needs to disambiguate). The one included test correctly avoids the subscribe-before-emit race by issuing the request before calling events.emit(...).

Findings

1. Unbounded SSE fan-out cost sits directly in the actor's synchronous hot path
EventBus::emit (from #516) calls tx.send() synchronously inside BlockChain's tick/block-import path, and crates/net/rpc/src/events.rs places no cap on how many clients can call GET /lean/v0/events concurrently. tokio::sync::broadcast::Sender::send wakes every live receiver, so a large number of open SSE connections (accidental or adversarial, since the endpoint has no auth) directly increases the per-tick cost paid by the consensus actor described in the tick-based validator duties section of CLAUDE.md. Worth at least a connection cap or a comment acknowledging the tradeoff, since this is unlike the rest of the request/response RPC surface where slow clients only cost a request-handler task, not actor time.

2. No Cache-Control: no-cache header on the SSE response
axum::response::Sse sets Content-Type: text/event-stream but does not add Cache-Control: no-cache by default. Without it, intermediary proxies/CDNs in front of the node could buffer or cache the stream, breaking live delivery for clients behind such infrastructure. Common practice for SSE endpoints is to set this explicitly (crates/net/rpc/src/events.rs:37, the get_events return).

3. Minor: dependency convention inconsistency
crates/net/rpc/Cargo.toml adds tokio-stream = { version = "0.1", features = ["sync"] } and futures-core = "0.3" as direct version pins, while every other dependency in this crate (and the codebase generally) is declared once in the root [workspace.dependencies] and referenced via .workspace = true (e.g. hex.workspace = true, tracing.workspace = true). Since these two are newly introduced by this PR, moving them to the workspace table would keep the convention consistent for any future crate that needs them.

4. Test coverage is single-path
Only events_streams_head_with_flat_payload is present, covering the happy path. There's no RPC-layer test for the lagged-client skip path (BroadcastStreamRecvError::Lagged) or for the json_data serialization-failure branch (.inspect_err(...).ok()?), though the underlying gating/diff logic is well covered at the blockchain-crate level from #516. Not blocking, but worth a follow-up test given this is the code path that decides whether a client silently loses events.

Non-issues (verified, no action needed)

• H256 serializes as 0x-hex and slot: u64 serializes as a plain JSON number — this matches the existing convention already used by Checkpoint (crates/common/types/src/checkpoint.rs:26, which only customizes deserialize to accept decimal strings), so it's not a new inconsistency despite earlier commit messages mentioning "decimal-string ints."
• The Some(Ok(...ok()?)) construct in the filter_map closure (crates/net/rpc/src/events.rs) is valid: the ? short-circuits the closure via Option's Try impl, correctly turning a serialization failure into a dropped (but logged) event rather than terminating the stream.
• EventBus: Clone wiring in main.rs (actor keeps the original, RPC gets a clone) is correct — broadcast::Sender is a cheap Arc-backed clone.

---

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