[MegaRedHand]

Motivation

Graceful shutdown waits for every in-flight response, and an SSE stream never finishes on its own: the router keeps its own EventBus clone, so the broadcast channel never closes and RecvError::Closed never arrives. Any subscriber left connected to /eth/v1/events or /lean/v0/events therefore holds SIGINT forever (SIGKILL skips the state flush).

This blocks running a permanent subscriber next to a node, such as ethPandaOps' contributoor, which keeps one stream per topic open and reconnects on its own.

Ending the streams covers a client that keeps reading. axum's graceful shutdown also waits with no limit for any response whose client stopped reading, such as a paused validator client or a slow download of a whole state, so that needs a bound of its own.

Changes

• serve() hands each router a ShutdownSignal extension holding a child of its shutdown token. A child is cancelled with its parent but cannot cancel it, so a handler can wait for shutdown and never trigger the node's. The token is private: the signal is built only through ShutdownSignal::of and exposes only cancelled().
• sse_response ends the stream with take_until when the signal fires. Both the lean and the Beacon API handlers extract it as an Option: a router served without crate::serve (plain axum::serve over build_beacon_api_router) answers, and its streams end when the client leaves, instead of a 500 for a missing extension.
• serve() stops waiting on responses still in flight HTTP_DRAIN_TIMEOUT (5 s) after shutdown starts. The listener is already closed by then, and the stalled connection tasks end with the runtime.
• docs/rpc.md: streams end at shutdown, and the servers wait at most HTTP_DRAIN_TIMEOUT for stalled responses.

Tests

• events_stream_ends_when_the_server_shuts_down: the body completes once the server's token is cancelled while the bus stays open.
• a_connected_event_stream_does_not_hold_up_shutdown and ..._on_separate_ports (tests/http_servers.rs): bind real sockets, keep /eth/v1/events open, cancel, and expect the servers to stop within 2 s, well under the drain timeout, so the signal is what ends them. One merges the API and metrics routers onto one listener, the other runs the default shape, two servers. Verified to fail (time out) without the fix.
• a_response_that_never_finishes_does_not_hold_up_shutdown: a body that never ends and knows nothing of the signal; the server stops after the drain timeout. Verified to time out without it. Adds about 5 s to the suite.
• The existing events unit tests now run without the signal layer, the case that used to return 500.
• cargo test --profile release-fast -p ethlambda-rpc --lib: 336 passed. --test http_servers: 6 passed. cargo clippy --locked -p ethlambda-rpc --all-targets -- -D warnings: clean.

[github-actions]

🤖 Claude Code Review

Review of PR 745: end SSE streams on shutdown

The approach is correct and minimal. I found no blocking issues, and ran no tests or build.

• Root cause: axum::serve(...).with_graceful_shutdown waits for every in-flight response. The router's own EventBus clone keeps the broadcast channel open, so RecvError::Closed never fires and an SSE stream never ends. Ending the stream from the same CancellationToken that triggers graceful shutdown fixes this.
• Wiring (lib.rs): serve adds Extension(ShutdownSignal(shutdown.clone())) once per listener. Both the lean and beacon routers, and the merged single-port case, therefore get it with no per-router plumbing. Because the layer is applied after the routes are built, it wraps them and populates the request extensions before any handler runs.
• Stream (events.rs): take_until(shutdown.0.cancelled_owned()) ends the body cleanly after the current frame. Cancelling a token that is already cancelled resolves immediately, so a subscriber that connects during shutdown also ends at once.
• Tests: The unit test and the end-to-end test in http_servers.rs both keep the bus alive, so only the token can end the stream. The e2e test covers the real hang. Pulling beacon_store() out of the existing test is a clean refactor.

Minor, non-blocking points:

1. Missing extension: Extension<ShutdownSignal> is now a required extractor on both get_events handlers. Any caller that builds the router without going through serve and forgets the layer gets a 500 at request time, not a compile error. The updated unit tests handle this. A future test or embedding that builds build_api_router directly and only layers EventBus would hit it. Consider a doc note on the router builders, or have sse_response fall back to an Option. The current approach is acceptable.
2. Test read (http_servers.rs): stream.read is a single call. It assumes the HTTP/1.1 200 status line arrives in the first read. That is reliable on localhost, but a small read loop until \r\n would remove the theoretical flake.
3. ShutdownSignal visibility: pub(crate) struct ShutdownSignal(pub(crate) CancellationToken) with Default is only useful in tests. That's fine, but you could gate the Default derive with #[cfg(test)] to keep production code from building an unconnected signal by accident.

The docs updates in docs/rpc.md are accurate and appropriately brief.

Verdict: Approve.

---

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

[github-actions]

🤖 Kimi Code Review

I'll review this PR which adds graceful shutdown support for SSE event streams. Let me analyze the changes across all files.

Overall Assessment

This is a well-structured PR that fixes a real operational issue: SSE streams blocking graceful shutdown. The approach using CancellationToken is idiomatic for Tokio. However, I've identified several issues ranging from minor to potentially serious.

---

crates/net/rpc/src/events.rs

Bug: take_until drops without yielding final events (Line 97)

let stream = event_stream(events.subscribe(), topics).take_until(shutdown.0.cancelled_owned());

Problem: StreamExt::take_until terminates the stream immediately when the future resolves, without yielding any item currently being processed. If a ChainEvent is in-flight when shutdown fires, it's silently dropped. For consensus events, this could mean a validator client misses the final head or finalized_checkpoint update before reconnect.

Suggestion: Consider take_until vs take_while behavior carefully. If event loss at shutdown matters, use a bounded channel or timeout-based drain. Document this behavior explicitly.

---

Issue: ShutdownSignal field visibility (Lines 37-38, 97)

#[derive(Clone, Default)]
pub(crate) struct ShutdownSignal(pub(crate) CancellationToken);

Problem: pub(crate) on both struct and field exposes internal implementation details unnecessarily. More critically, Default for CancellationToken creates a new, disconnected token—this is a footgun.

Line 97 usage:

.take_until(shutdown.0.cancelled_owned())

Direct field access to .0 is brittle. Suggest adding an accessor:

impl ShutdownSignal {
    /// Returns a future that resolves when shutdown is requested.
    pub(crate) fn cancelled_owned(self) -> impl Future<Output = ()> {
        self.0.cancelled_owned()
    }
}

And reconsider Default—it's only correct in tests where the token is never triggered. Consider #[cfg(test)] or a new() constructor that documents the intent.

---

Documentation/Logic Mismatch (Lines 128-131)

// Every sender dropped. The router holds one, so in practice
// shutdown ends the stream through `ShutdownSignal` instead.
Err(RecvError::Closed) => return None,

Problem: The comment claims shutdown ends the stream through ShutdownSignal, but RecvError::Closed is a different path. If ShutdownSignal is working correctly, this branch should be unreachable in normal shutdown. However, if there's a race where the last EventBus sender drops before the take_until resolves, the stream ends here without the client receiving a clean close.

Clarification needed: Is RecvError::Closed truly unreachable, or a fallback? The comment is confusing.

---

crates/net/rpc/src/lib.rs (Line 152)

let app = app.layer(Extension(events::ShutdownSignal(shutdown.clone())));

Problem: shutdown.clone() on a CancellationToken is cheap, but this layers the extension inside serve() per-connection. The Extension is cloned per-request anyway by axum, so this is fine functionally.

However, naming: events::ShutdownSignal wraps a token but the module is events. Consider if this belongs in a more general shutdown module as the pattern grows.

---

crates/net/rpc/tests/http_servers.rs

Test Bug: Port Collision (Lines 196-197)

let config = RpcConfig {
    http_address: LOCALHOST,
    api_port: port,
    metrics_port: port,  // Same port!

Problem: api_port and metrics_port are the same! The test passes because start_http_servers likely binds metrics first, then API, or vice versa—this is a race condition. If binding order changes or ports are checked separately, this will fail with "address already in use."

Line 200:

let router = ethlambda_rpc::build_beacon_api_router(...)
    .layer(axum::Extension(events.clone()));

The router lacks SyncStatusController extension that the refactored the_beacon_router_replaces_the_lean_one test adds. Is this intentional? The test only hits /events, so it may not matter, but it's inconsistent.

---

Resource Leak: TCP Stream Not Shut Down (Lines 210-220)

let mut stream = TcpStream::connect(...).await...;
// ... write, read ...
shutdown.cancel();
tokio::time::timeout(Duration::from_secs(5), served).await...

Problem: The stream is never explicitly closed. While OS handles it on process exit, in a test this could leave the socket in TIME_WAIT or cause the spawned server task to hang if it tries to send more data. Add drop(stream) before shutdown.cancel() or use a scope.

---

Fragile Test: Hardcoded Buffer Size (Line 216)

let mut head = vec![0; 1024];
let read = stream.read(&mut head).await.expect("the headers read");

Problem: If HTTP headers exceed 1024 bytes (unlikely but possible with many extensions), this silently truncates. Use a loop or tokio::io::AsyncBufReadExt for robust header reading.

---

crates/net/rpc/src/beacon/events.rs

Correctness: No issues

The beacon events wrapper properly threads through ShutdownSignal. The test update at lines 101-102 correctly adds the extension.

---

General Observations

1. Dependency: New tokio-util dependency is pulled in for CancellationToken. Verify this is already in the workspace or add it to Cargo.toml (not shown in diff—check before merge).

2. Consistency: ShutdownSignal is pub(crate) but used across crate boundaries in tests (tests/http_servers.rs is an integration test, so it has extern crate ethlambda_rpc; access). Verify visibility is correct for test compilation.

3. Documentation: The rpc.md update is accurate and helpful.

---

Summary Table

Priority	Location	Issue
Medium	events.rs:97	take_until may drop in-flight events at shutdown
Medium	tests/http_servers.rs:197	api_port == metrics_port is fragile
Low	events.rs:37-38	Default on ShutdownSignal is a footgun; accessor missing
Low	tests/http_servers.rs:210-220	TCP stream not explicitly closed
Low	events.rs:128-131	Comment contradicts code path

The core fix is correct and the tests validate the behavior. Address the port collision and consider the event-dropping semantics before merge.

---

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

[github-actions]

🤖 Codex Code Review

⚠️ The review job did not complete (result: failure). See the job log

---

Automated review by OpenAI Codex

[pablodeymo]

Went through the Kimi review with a security lens. One item is worth acting
on, and it connects to something the bots did not say.

The handlers get the node-wide shutdown token. serve wraps
shutdown.clone(), which is the same token wait_for_shutdown cancels, and
ShutdownSignal exposes it as a pub(crate) field. Any handler in this crate,
or a future bug in one, can call cancel() and take the whole node down, and
Default builds a token that never fires, which is exactly the hang this PR
removes. Nothing exposes either over the wire today, but both are loaded
capabilities sitting in request scope. Two cheap changes close it:

1. In serve, wrap shutdown.child_token() instead of a clone. A child is
   cancelled with its parent but cannot cancel it, so a handler can only
   wait on shutdown, never trigger it.
2. Make the field private, gate Default with #[cfg(test)], and expose
   only a cancelled() future on ShutdownSignal.

The rest of Kimi's items are not security issues on inspection:

• take_until dropping queued events at shutdown is real but low: the
  actors are stopped before the cancel, the stream is documented best-effort
  with re-sync, and SSE has no end marker either way. Optional to drain the
  receiver after the cancel.
• api_port == metrics_port in the test is the supported merged mode, used
  by two existing tests in the same file.
• Closing the TCP stream before the cancel would make the test pass without
  the fix; the open stream is the premise.
• tokio-util is already a dependency of the crate, and the integration
  test never names ShutdownSignal.

[MegaRedHand]

@pablodeymo both are in:

1. d8f8b55c4: serve wraps shutdown.child_token().
2. cfbfff2b7: the field is private, the signal is built only through ShutdownSignal::of (which takes the child token itself), and it exposes only cancelled(). Default went behind #[cfg(test)] there, then went away entirely in eb19261b0. In that commit the handlers take Option<Extension<ShutdownSignal>>, so a router served with plain axum::serve answers instead of returning a 500, and no test needs a never-firing signal any more.

Two more from a second review pass:

• 715683424: take_until covers a client that keeps reading. But axum's graceful shutdown waits with no limit for every response, so a client that stops reading (a paused validator client, a slow state download) still held SIGINT, and the new line in docs/rpc.md promised more than the code did. serve now stops waiting HTTP_DRAIN_TIMEOUT (5 s) after shutdown starts. a_response_that_never_finishes_does_not_hold_up_shutdown times out without it.
• 16f0e0550: the shutdown test also runs with separate API and metrics ports, the try_join! branch the defaults go through. Its bound is now 2 s, well under the drain timeout, so it's the signal that ends the stream, not the timeout.

Agreed on the rest of your notes; nothing changed for them.