From 3bba47dfdfd9e4d8b75fdffc6650e944e6ed0ac2 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?=
 <47506558+MegaRedHand@users.noreply.github.com>
Date: Fri, 30 Oct 2026 16:40:44 -0300
Subject: [PATCH 1/6] fix(rpc): end event streams when the node shuts down

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
channel never closes. Any subscriber left connected to /eth/v1/events or
/lean/v0/events therefore held SIGINT forever, and SIGKILL skips the state
flush. A permanent subscriber such as contributoor would make every restart
hang.

serve() now hands each router its shutdown token as a ShutdownSignal
extension, and sse_response ends the stream when it fires.
---
 crates/net/rpc/src/beacon/events.rs  |   7 +-
 crates/net/rpc/src/events.rs         |  55 ++++++++++--
 crates/net/rpc/src/lib.rs            |   5 ++
 crates/net/rpc/tests/http_servers.rs | 120 ++++++++++++++++++++-------
 docs/rpc.md                          |   5 +-
 5 files changed, 149 insertions(+), 43 deletions(-)

diff --git a/crates/net/rpc/src/beacon/events.rs b/crates/net/rpc/src/beacon/events.rs
index 5b7902079..e74eacb6a 100644
--- a/crates/net/rpc/src/beacon/events.rs
+++ b/crates/net/rpc/src/beacon/events.rs
@@ -29,13 +29,15 @@ use ethlambda_blockchain::{EventBus, Topic};
 use ethlambda_storage::Store;
 
 use super::ApiError;
+use crate::events::ShutdownSignal;
 
 async fn get_events(
     Extension(events): Extension<EventBus>,
+    Extension(shutdown): Extension<ShutdownSignal>,
     Query(params): Query<Vec<(String, String)>>,
 ) -> Response {
     match requested_topics(&params) {
-        Ok(topics) => crate::events::sse_response(&events, topics),
+        Ok(topics) => crate::events::sse_response(&events, topics, shutdown),
         Err(err) => err.into_response(),
     }
 }
@@ -96,7 +98,8 @@ mod tests {
         let fixture = beacon_fixture(64);
         let app = routes()
             .with_state(fixture.store)
-            .layer(Extension(events.clone()));
+            .layer(Extension(events.clone()))
+            .layer(Extension(ShutdownSignal::default()));
         app.oneshot(Request::builder().uri(uri).body(Body::empty()).unwrap())
             .await
             .unwrap()
diff --git a/crates/net/rpc/src/events.rs b/crates/net/rpc/src/events.rs
index ca9069343..6be6549b2 100644
--- a/crates/net/rpc/src/events.rs
+++ b/crates/net/rpc/src/events.rs
@@ -17,6 +17,10 @@
 //! The stream itself ([`sse_response`]) is shared with the Beacon API's
 //! `/eth/v1/events` (`crate::beacon::events`), which differs only in how it
 //! reads `topics` and words a refusal.
+//!
+//! Shutdown: a stream ends when the server's [`ShutdownSignal`] fires. Without
+//! that, a connected subscriber would keep graceful shutdown waiting forever,
+//! since the router's own [`EventBus`] clone keeps the channel open.
 
 use std::convert::Infallible;
 
@@ -29,9 +33,18 @@ use axum::{
 };
 use ethlambda_blockchain::{ChainEvent, EventBus, Topic};
 use ethlambda_storage::Store;
-use futures_util::{Stream, stream::unfold};
+use futures_util::{Stream, StreamExt as _, stream::unfold};
 use serde::Deserialize;
 use tokio::sync::broadcast::{self, error::RecvError};
+use tokio_util::sync::CancellationToken;
+
+/// The token that stops the HTTP server a request arrived on.
+///
+/// Every router the server serves carries it as an [`Extension`] (see
+/// `crate::serve`), so a response that would otherwise never finish can end
+/// itself when the server starts shutting down.
+#[derive(Clone, Default)]
+pub(crate) struct ShutdownSignal(pub(crate) CancellationToken);
 
 #[derive(Deserialize)]
 struct EventsParams {
@@ -40,6 +53,7 @@ struct EventsParams {
 
 async fn get_events(
     Extension(events): Extension<EventBus>,
+    Extension(shutdown): Extension<ShutdownSignal>,
     Query(params): Query<EventsParams>,
 ) -> Response {
     // `topics` is required, matching the Beacon API: a missing or empty value
@@ -65,18 +79,23 @@ async fn get_events(
         },
     };
 
-    sse_response(&events, topics)
+    sse_response(&events, topics, shutdown)
 }
 
 /// An SSE response streaming every event on `events` whose topic is in
-/// `topics`, from now on.
+/// `topics`, from now on, until `shutdown` fires.
 ///
 /// `X-Accel-Buffering: no` tells an nginx reverse proxy not to buffer the
 /// stream, which would otherwise hold events back until a buffer fills; axum's
 /// [`Sse`] already sets `content-type: text/event-stream` and
 /// `cache-control: no-cache`.
-pub(crate) fn sse_response(events: &EventBus, topics: Vec<Topic>) -> Response {
-    let mut response = Sse::new(event_stream(events.subscribe(), topics))
+pub(crate) fn sse_response(
+    events: &EventBus,
+    topics: Vec<Topic>,
+    shutdown: ShutdownSignal,
+) -> Response {
+    let stream = event_stream(events.subscribe(), topics).take_until(shutdown.0.cancelled_owned());
+    let mut response = Sse::new(stream)
         .keep_alive(axum::response::sse::KeepAlive::default())
         .into_response();
     response
@@ -125,7 +144,8 @@ fn event_stream(
                         Event::default().comment(format!("error - dropped {skipped} messages"));
                     return Some((Ok(frame), (rx, topics)));
                 }
-                // Publisher dropped: the node is shutting down.
+                // Every sender dropped. The router holds one, so in practice
+                // shutdown ends the stream through `ShutdownSignal` instead.
                 Err(RecvError::Closed) => return None,
             }
         }
@@ -143,12 +163,13 @@ mod tests {
         body::Body,
         http::{Request, StatusCode},
     };
-    use ethlambda_blockchain::{ChainEvent, EventBus};
+    use ethlambda_blockchain::{ChainEvent, EventBus, Topic};
     use ethlambda_storage::{Store, backend::InMemoryBackend};
     use ethlambda_types::constants::DEFAULT_MILLISECONDS_PER_SLOT;
     use futures_util::StreamExt;
     use http_body_util::BodyExt;
     use std::sync::Arc;
+    use std::time::Duration;
     use tower::ServiceExt;
 
     use crate::test_utils::create_test_state;
@@ -159,7 +180,9 @@ mod tests {
             create_test_state(),
             DEFAULT_MILLISECONDS_PER_SLOT,
         );
-        let app = crate::test_utils::test_api_router(store).layer(Extension(events.clone()));
+        let app = crate::test_utils::test_api_router(store)
+            .layer(Extension(events.clone()))
+            .layer(Extension(super::ShutdownSignal::default()));
         app.oneshot(Request::builder().uri(uri).body(Body::empty()).unwrap())
             .await
             .unwrap()
@@ -396,4 +419,20 @@ mod tests {
             assert_eq!(resp.status(), StatusCode::BAD_REQUEST, "uri: {uri}");
         }
     }
+
+    #[tokio::test]
+    async fn events_stream_ends_when_the_server_shuts_down() {
+        let events = EventBus::new(16);
+        let shutdown = super::ShutdownSignal::default();
+        let resp = super::sse_response(&events, vec![Topic::Head], shutdown.clone());
+
+        shutdown.0.cancel();
+
+        // The bus stays open (`events` is still alive), so only the signal
+        // can end the body; before it did, this waited forever.
+        tokio::time::timeout(Duration::from_secs(5), resp.into_body().collect())
+            .await
+            .expect("the stream ends once shutdown fires")
+            .expect("the body reads");
+    }
 }
diff --git a/crates/net/rpc/src/lib.rs b/crates/net/rpc/src/lib.rs
index 4870c2781..537aa49d4 100644
--- a/crates/net/rpc/src/lib.rs
+++ b/crates/net/rpc/src/lib.rs
@@ -136,6 +136,10 @@ pub async fn start_http_servers(
 }
 
 /// Bind one listener and serve `app` on it until `shutdown` is cancelled.
+///
+/// Graceful shutdown waits for every in-flight response, so `app` gets the
+/// token as a [`events::ShutdownSignal`] too: the event streams, which never
+/// finish on their own, end on it.
 async fn serve(
     address: IpAddr,
     port: u16,
@@ -145,6 +149,7 @@ async fn serve(
     let addr = SocketAddr::new(address, port);
     let listener = tokio::net::TcpListener::bind(addr).await?;
     tracing::info!(%addr, "HTTP server listening");
+    let app = app.layer(Extension(events::ShutdownSignal(shutdown.clone())));
     axum::serve(listener, app)
         .with_graceful_shutdown(async move {
             shutdown.cancelled().await;
diff --git a/crates/net/rpc/tests/http_servers.rs b/crates/net/rpc/tests/http_servers.rs
index 6171b1730..f64ec1809 100644
--- a/crates/net/rpc/tests/http_servers.rs
+++ b/crates/net/rpc/tests/http_servers.rs
@@ -146,12 +146,98 @@ async fn a_cancelled_token_stops_the_server() {
 #[tokio::test]
 async fn the_beacon_router_replaces_the_lean_one() {
     use axum::{body::Body, http::Request};
+    use tower::ServiceExt as _;
+
+    let router =
+        ethlambda_rpc::build_beacon_api_router(beacon_store(), "ethlambda/test", "peer".into())
+            .layer(axum::Extension(
+                ethlambda_blockchain::SyncStatusController::default(),
+            ));
+
+    let beacon = router
+        .clone()
+        .oneshot(
+            Request::builder()
+                .uri("/eth/v1/node/version")
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(beacon.status(), axum::http::StatusCode::OK);
+
+    let lean = router
+        .oneshot(
+            Request::builder()
+                .uri("/lean/v0/node/syncing")
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(
+        lean.status(),
+        axum::http::StatusCode::NOT_FOUND,
+        "a beacon node must not serve the lean surface off a beacon store"
+    );
+}
+
+/// A subscriber left connected to `/eth/v1/events` does not hold up shutdown.
+///
+/// Graceful shutdown waits for every in-flight response, and an event stream
+/// never finishes on its own: before the streams ended on the shutdown token, a
+/// node with any SSE client attached (a validator client, contributoor) hung on
+/// SIGINT until the client disconnected.
+#[tokio::test]
+async fn a_connected_event_stream_does_not_hold_up_shutdown() {
+    let port = free_port().await;
+    let shutdown = CancellationToken::new();
+    let config = RpcConfig {
+        http_address: LOCALHOST,
+        api_port: port,
+        metrics_port: port,
+        version: "ethlambda/test",
+    };
+    // Kept alive for the whole test, so the channel never closes and only the
+    // shutdown token can end the stream.
+    let events = ethlambda_blockchain::EventBus::new(16);
+    let router =
+        ethlambda_rpc::build_beacon_api_router(beacon_store(), "ethlambda/test", "peer".into())
+            .layer(axum::Extension(events.clone()));
+    let served = tokio::spawn(start_http_servers(config, Some(router), shutdown.clone()));
+
+    // Wait for the metrics route so the socket is accepting, then subscribe.
+    assert!(status_line(&get(port, "/metrics").await).contains("200"));
+    let mut stream = TcpStream::connect(SocketAddr::new(LOCALHOST, port))
+        .await
+        .expect("the server accepts");
+    stream
+        .write_all(b"GET /eth/v1/events?topics=head HTTP/1.1\r\nHost: localhost\r\n\r\n")
+        .await
+        .expect("the request writes");
+    let mut head = vec![0; 1024];
+    let read = stream.read(&mut head).await.expect("the headers read");
+    assert!(
+        String::from_utf8_lossy(&head[..read]).starts_with("HTTP/1.1 200"),
+        "the event stream opens"
+    );
+
+    shutdown.cancel();
+    tokio::time::timeout(Duration::from_secs(5), served)
+        .await
+        .expect("the server stops within five seconds despite the open stream")
+        .expect("the server task joins")
+        .expect("the server exits cleanly");
+}
+
+/// A beacon `Store` anchored at a phase0 block, enough for the beacon router
+/// to build.
+fn beacon_store() -> ethlambda_storage::Store {
     use ethlambda_types::beacon::{
         config::Config,
         containers::{SignedBeaconBlock, phase0},
         primitives::Root,
     };
-    use tower::ServiceExt as _;
 
     const GENESIS_TIME: u64 = 1_606_824_023;
     let slot = 64;
@@ -180,35 +266,5 @@ async fn the_beacon_router_replaces_the_lean_one() {
         .insert_signed_block(root, block)
         .expect("insert anchor block");
 
-    let router = ethlambda_rpc::build_beacon_api_router(store, "ethlambda/test", "peer".into())
-        .layer(axum::Extension(
-            ethlambda_blockchain::SyncStatusController::default(),
-        ));
-
-    let beacon = router
-        .clone()
-        .oneshot(
-            Request::builder()
-                .uri("/eth/v1/node/version")
-                .body(Body::empty())
-                .unwrap(),
-        )
-        .await
-        .unwrap();
-    assert_eq!(beacon.status(), axum::http::StatusCode::OK);
-
-    let lean = router
-        .oneshot(
-            Request::builder()
-                .uri("/lean/v0/node/syncing")
-                .body(Body::empty())
-                .unwrap(),
-        )
-        .await
-        .unwrap();
-    assert_eq!(
-        lean.status(),
-        axum::http::StatusCode::NOT_FOUND,
-        "a beacon node must not serve the lean surface off a beacon store"
-    );
+    store
 }
diff --git a/docs/rpc.md b/docs/rpc.md
index bdcc260ec..a878df0bb 100644
--- a/docs/rpc.md
+++ b/docs/rpc.md
@@ -133,6 +133,8 @@ Valid values are exactly the event names above: `head`, `block`, `justified_chec
 
 Events are fanned out over a single bounded broadcast channel shared by all topics. A client that reads too slowly skips past the events it missed: they are dropped for that subscriber rather than back-pressured onto the actor, so treat the stream as best-effort and re-sync via the blocks endpoints after a gap. A client that falls behind receives an SSE comment line `: error - dropped N messages` marking the gap (wire-compatible with Lighthouse) before the stream continues; re-sync via the blocks endpoints rather than trusting the skipped range. Keep-alive comments are sent periodically to hold idle connections open.
 
+The stream ends when the node shuts down, so a connected client never holds up a graceful shutdown; expect the connection to close and reconnect once the node is back.
+
 Because the ring buffer is shared, the high-rate `attestation` events (roughly one per validator per slot) dominate its occupancy: a subscriber's tolerable stall is `capacity / total_event_rate`, not per-topic, so filtering with `?topics=` narrows what you receive but does **not** widen the lag window against an attestation flood. Subscribers that only need low-rate topics (`head`, `finalized_checkpoint`, …) are still evicted at the aggregate rate. If real usage shows this biting, the fix is a per-topic channel split behind the event bus (the `subscribe(TopicSet)` API is unaffected).
 
 ### `GET /lean/v0/blocks/{block_id}` and `/header`
@@ -852,7 +854,8 @@ alone: the p2p specification counts it separately from the ENR's `seq`.
 
 The Beacon API eventstream: the same Server-Sent Events stream as
 [`/lean/v0/events`](#get-leanv0events) (topic on the `event:` line, payload on
-`data:`, one shared best-effort ring, `: error - dropped N messages` on a gap),
+`data:`, one shared best-effort ring, `: error - dropped N messages` on a gap,
+ended at shutdown),
 with the Beacon API's own topics and payloads. The response also sets
 `X-Accel-Buffering: no`, so a reverse proxy does not hold events back.
 

From d8f8b55c4b2d4ef35c17088fe300993042398971 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?=
 <47506558+MegaRedHand@users.noreply.github.com>
Date: Mon, 2 Nov 2026 17:36:01 -0300
Subject: [PATCH 2/6] fix(rpc): give handlers a child of the shutdown token,
 not the token

serve put the very token wait_for_shutdown cancels into request scope, so
any handler in the crate, or a future bug in one, could call cancel() and
take the whole node down. A child token is cancelled with its parent but
cannot cancel it, so a handler can now only wait for shutdown.
---
 crates/net/rpc/src/lib.rs | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/crates/net/rpc/src/lib.rs b/crates/net/rpc/src/lib.rs
index 537aa49d4..138d2868c 100644
--- a/crates/net/rpc/src/lib.rs
+++ b/crates/net/rpc/src/lib.rs
@@ -139,7 +139,9 @@ pub async fn start_http_servers(
 ///
 /// Graceful shutdown waits for every in-flight response, so `app` gets the
 /// token as a [`events::ShutdownSignal`] too: the event streams, which never
-/// finish on their own, end on it.
+/// finish on their own, end on it. It gets a child of the token, not a clone:
+/// a child is cancelled with its parent but cannot cancel it, so a handler can
+/// wait for shutdown and never trigger the node's.
 async fn serve(
     address: IpAddr,
     port: u16,
@@ -149,7 +151,7 @@ async fn serve(
     let addr = SocketAddr::new(address, port);
     let listener = tokio::net::TcpListener::bind(addr).await?;
     tracing::info!(%addr, "HTTP server listening");
-    let app = app.layer(Extension(events::ShutdownSignal(shutdown.clone())));
+    let app = app.layer(Extension(events::ShutdownSignal(shutdown.child_token())));
     axum::serve(listener, app)
         .with_graceful_shutdown(async move {
             shutdown.cancelled().await;

From cfbfff2b7778ad169240d1b4464f9ab72e4bc5aa Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?=
 <47506558+MegaRedHand@users.noreply.github.com>
Date: Mon, 2 Nov 2026 17:38:26 -0300
Subject: [PATCH 3/6] fix(rpc): let a ShutdownSignal only be waited on

The token sat in a pub(crate) field, so code holding the signal could
reach it and cancel, and Default built a signal that never fires, which
in production would bring back the very hang this branch removes. The
field is now private, the signal is built only through
ShutdownSignal::of, which takes the child token itself, it exposes only
a cancelled() future, and Default exists for tests alone.
---
 crates/net/rpc/src/events.rs | 38 +++++++++++++++++++++++++++---------
 crates/net/rpc/src/lib.rs    |  8 ++++----
 2 files changed, 33 insertions(+), 13 deletions(-)

diff --git a/crates/net/rpc/src/events.rs b/crates/net/rpc/src/events.rs
index 6be6549b2..f3d987374 100644
--- a/crates/net/rpc/src/events.rs
+++ b/crates/net/rpc/src/events.rs
@@ -36,15 +36,34 @@ use ethlambda_storage::Store;
 use futures_util::{Stream, StreamExt as _, stream::unfold};
 use serde::Deserialize;
 use tokio::sync::broadcast::{self, error::RecvError};
-use tokio_util::sync::CancellationToken;
+use tokio_util::sync::{CancellationToken, WaitForCancellationFutureOwned};
 
-/// The token that stops the HTTP server a request arrived on.
+/// What a response waits on to learn that the HTTP server it arrived on is
+/// shutting down.
 ///
 /// Every router the server serves carries it as an [`Extension`] (see
 /// `crate::serve`), so a response that would otherwise never finish can end
-/// itself when the server starts shutting down.
-#[derive(Clone, Default)]
-pub(crate) struct ShutdownSignal(pub(crate) CancellationToken);
+/// itself when the server starts shutting down. It holds a child of the
+/// server's token and exposes only [`ShutdownSignal::cancelled`], so a handler
+/// can wait for shutdown but never trigger it. `Default`, a signal that never
+/// fires, exists only for tests: in production it would bring back the hang
+/// this type removes.
+#[derive(Clone)]
+#[cfg_attr(test, derive(Default))]
+pub(crate) struct ShutdownSignal(CancellationToken);
+
+impl ShutdownSignal {
+    /// A signal that fires when `server` is cancelled. It holds a child token,
+    /// which cannot cancel `server`.
+    pub(crate) fn of(server: &CancellationToken) -> Self {
+        Self(server.child_token())
+    }
+
+    /// Resolves once the server starts shutting down.
+    pub(crate) fn cancelled(self) -> WaitForCancellationFutureOwned {
+        self.0.cancelled_owned()
+    }
+}
 
 #[derive(Deserialize)]
 struct EventsParams {
@@ -94,7 +113,7 @@ pub(crate) fn sse_response(
     topics: Vec<Topic>,
     shutdown: ShutdownSignal,
 ) -> Response {
-    let stream = event_stream(events.subscribe(), topics).take_until(shutdown.0.cancelled_owned());
+    let stream = event_stream(events.subscribe(), topics).take_until(shutdown.cancelled());
     let mut response = Sse::new(stream)
         .keep_alive(axum::response::sse::KeepAlive::default())
         .into_response();
@@ -423,10 +442,11 @@ mod tests {
     #[tokio::test]
     async fn events_stream_ends_when_the_server_shuts_down() {
         let events = EventBus::new(16);
-        let shutdown = super::ShutdownSignal::default();
-        let resp = super::sse_response(&events, vec![Topic::Head], shutdown.clone());
+        let server = tokio_util::sync::CancellationToken::new();
+        let shutdown = super::ShutdownSignal::of(&server);
+        let resp = super::sse_response(&events, vec![Topic::Head], shutdown);
 
-        shutdown.0.cancel();
+        server.cancel();
 
         // The bus stays open (`events` is still alive), so only the signal
         // can end the body; before it did, this waited forever.
diff --git a/crates/net/rpc/src/lib.rs b/crates/net/rpc/src/lib.rs
index 138d2868c..dfb8e294e 100644
--- a/crates/net/rpc/src/lib.rs
+++ b/crates/net/rpc/src/lib.rs
@@ -139,9 +139,9 @@ pub async fn start_http_servers(
 ///
 /// Graceful shutdown waits for every in-flight response, so `app` gets the
 /// token as a [`events::ShutdownSignal`] too: the event streams, which never
-/// finish on their own, end on it. It gets a child of the token, not a clone:
-/// a child is cancelled with its parent but cannot cancel it, so a handler can
-/// wait for shutdown and never trigger the node's.
+/// finish on their own, end on it. The signal holds a child of the token, not
+/// a clone: a child is cancelled with its parent but cannot cancel it, so a
+/// handler can wait for shutdown and never trigger the node's.
 async fn serve(
     address: IpAddr,
     port: u16,
@@ -151,7 +151,7 @@ async fn serve(
     let addr = SocketAddr::new(address, port);
     let listener = tokio::net::TcpListener::bind(addr).await?;
     tracing::info!(%addr, "HTTP server listening");
-    let app = app.layer(Extension(events::ShutdownSignal(shutdown.child_token())));
+    let app = app.layer(Extension(events::ShutdownSignal::of(&shutdown)));
     axum::serve(listener, app)
         .with_graceful_shutdown(async move {
             shutdown.cancelled().await;

From eb19261b0a7184916407e10724ce1da2ae273550 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?=
 <47506558+MegaRedHand@users.noreply.github.com>
Date: Mon, 2 Nov 2026 17:39:41 -0300
Subject: [PATCH 4/6] fix(rpc): answer the events routes without a
 ShutdownSignal layer

Both events handlers required Extension<ShutdownSignal>, which only
crate::serve adds, so a router served any other way (plain axum::serve
over build_beacon_api_router, as two test files already do) answered
GET /eth/v1/events with a 500 at runtime rather than failing to build.
The handlers now take it as an Option, and without one the stream ends
only when the client leaves, as every stream did before. The test-only
Default goes with it: the unit tests now run without the layer, which is
the case this covers.
---
 crates/net/rpc/src/beacon/events.rs |  8 ++++---
 crates/net/rpc/src/events.rs        | 35 ++++++++++++++++++-----------
 2 files changed, 27 insertions(+), 16 deletions(-)

diff --git a/crates/net/rpc/src/beacon/events.rs b/crates/net/rpc/src/beacon/events.rs
index e74eacb6a..1122b975a 100644
--- a/crates/net/rpc/src/beacon/events.rs
+++ b/crates/net/rpc/src/beacon/events.rs
@@ -33,9 +33,10 @@ use crate::events::ShutdownSignal;
 
 async fn get_events(
     Extension(events): Extension<EventBus>,
-    Extension(shutdown): Extension<ShutdownSignal>,
+    shutdown: Option<Extension<ShutdownSignal>>,
     Query(params): Query<Vec<(String, String)>>,
 ) -> Response {
+    let shutdown = shutdown.map(|Extension(signal)| signal);
     match requested_topics(&params) {
         Ok(topics) => crate::events::sse_response(&events, topics, shutdown),
         Err(err) => err.into_response(),
@@ -96,10 +97,11 @@ mod tests {
 
     async fn events_response(events: &EventBus, uri: &str) -> Response {
         let fixture = beacon_fixture(64);
+        // No `ShutdownSignal` layer: a router served outside `crate::serve`
+        // must still answer.
         let app = routes()
             .with_state(fixture.store)
-            .layer(Extension(events.clone()))
-            .layer(Extension(ShutdownSignal::default()));
+            .layer(Extension(events.clone()));
         app.oneshot(Request::builder().uri(uri).body(Body::empty()).unwrap())
             .await
             .unwrap()
diff --git a/crates/net/rpc/src/events.rs b/crates/net/rpc/src/events.rs
index f3d987374..74e0f59e4 100644
--- a/crates/net/rpc/src/events.rs
+++ b/crates/net/rpc/src/events.rs
@@ -45,11 +45,13 @@ use tokio_util::sync::{CancellationToken, WaitForCancellationFutureOwned};
 /// `crate::serve`), so a response that would otherwise never finish can end
 /// itself when the server starts shutting down. It holds a child of the
 /// server's token and exposes only [`ShutdownSignal::cancelled`], so a handler
-/// can wait for shutdown but never trigger it. `Default`, a signal that never
-/// fires, exists only for tests: in production it would bring back the hang
-/// this type removes.
+/// can wait for shutdown but never trigger it.
+///
+/// Handlers extract it as an `Option`: a router served some other way (a
+/// test, or a caller of `build_beacon_api_router` serving it with plain
+/// `axum::serve`) has no signal, and its streams then end only when the
+/// client leaves, rather than the request failing for a missing extension.
 #[derive(Clone)]
-#[cfg_attr(test, derive(Default))]
 pub(crate) struct ShutdownSignal(CancellationToken);
 
 impl ShutdownSignal {
@@ -72,7 +74,7 @@ struct EventsParams {
 
 async fn get_events(
     Extension(events): Extension<EventBus>,
-    Extension(shutdown): Extension<ShutdownSignal>,
+    shutdown: Option<Extension<ShutdownSignal>>,
     Query(params): Query<EventsParams>,
 ) -> Response {
     // `topics` is required, matching the Beacon API: a missing or empty value
@@ -98,11 +100,12 @@ async fn get_events(
         },
     };
 
-    sse_response(&events, topics, shutdown)
+    sse_response(&events, topics, shutdown.map(|Extension(signal)| signal))
 }
 
 /// An SSE response streaming every event on `events` whose topic is in
-/// `topics`, from now on, until `shutdown` fires.
+/// `topics`, from now on, until `shutdown` fires. With no signal, the stream
+/// ends only when the client leaves; see [`ShutdownSignal`].
 ///
 /// `X-Accel-Buffering: no` tells an nginx reverse proxy not to buffer the
 /// stream, which would otherwise hold events back until a buffer fills; axum's
@@ -111,9 +114,15 @@ async fn get_events(
 pub(crate) fn sse_response(
     events: &EventBus,
     topics: Vec<Topic>,
-    shutdown: ShutdownSignal,
+    shutdown: Option<ShutdownSignal>,
 ) -> Response {
-    let stream = event_stream(events.subscribe(), topics).take_until(shutdown.cancelled());
+    let shutdown = async move {
+        match shutdown {
+            Some(signal) => signal.cancelled().await,
+            None => std::future::pending().await,
+        }
+    };
+    let stream = event_stream(events.subscribe(), topics).take_until(shutdown);
     let mut response = Sse::new(stream)
         .keep_alive(axum::response::sse::KeepAlive::default())
         .into_response();
@@ -199,9 +208,9 @@ mod tests {
             create_test_state(),
             DEFAULT_MILLISECONDS_PER_SLOT,
         );
-        let app = crate::test_utils::test_api_router(store)
-            .layer(Extension(events.clone()))
-            .layer(Extension(super::ShutdownSignal::default()));
+        // No `ShutdownSignal` layer: a router served outside `crate::serve`
+        // must still answer.
+        let app = crate::test_utils::test_api_router(store).layer(Extension(events.clone()));
         app.oneshot(Request::builder().uri(uri).body(Body::empty()).unwrap())
             .await
             .unwrap()
@@ -444,7 +453,7 @@ mod tests {
         let events = EventBus::new(16);
         let server = tokio_util::sync::CancellationToken::new();
         let shutdown = super::ShutdownSignal::of(&server);
-        let resp = super::sse_response(&events, vec![Topic::Head], shutdown);
+        let resp = super::sse_response(&events, vec![Topic::Head], Some(shutdown));
 
         server.cancel();
 

From 7156834241c889380887bbd2f39671acb4caf539 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?=
 <47506558+MegaRedHand@users.noreply.github.com>
Date: Mon, 2 Nov 2026 17:42:18 -0300
Subject: [PATCH 5/6] fix(rpc): stop waiting on stalled responses after a drain
 timeout

Ending the event streams on the shutdown signal covered a client that
keeps reading, but axum's graceful shutdown waits with no limit for every
response, and a response stalls for as long as its client stops reading:
a validator client paused mid-stream, or a slow download of a whole state.
docs/rpc.md said a client never holds up shutdown, which was not true for
those. serve now stops waiting HTTP_DRAIN_TIMEOUT after shutdown starts;
the listener is already closed then, and the stalled connection tasks
end with the runtime.
---
 crates/net/rpc/src/lib.rs            | 36 +++++++++++++++++---
 crates/net/rpc/tests/http_servers.rs | 49 ++++++++++++++++++++++++++++
 docs/rpc.md                          |  2 +-
 3 files changed, 81 insertions(+), 6 deletions(-)

diff --git a/crates/net/rpc/src/lib.rs b/crates/net/rpc/src/lib.rs
index dfb8e294e..f20c26876 100644
--- a/crates/net/rpc/src/lib.rs
+++ b/crates/net/rpc/src/lib.rs
@@ -1,5 +1,7 @@
+use std::future::IntoFuture as _;
 use std::net::{IpAddr, SocketAddr};
 use std::sync::Arc;
+use std::time::Duration;
 
 use axum::{Extension, Router};
 use ethlambda_blockchain::{EventBus, SyncStatusController};
@@ -135,6 +137,14 @@ pub async fn start_http_servers(
     Ok(())
 }
 
+/// How long [`serve`] waits, once shutdown starts, for responses still in
+/// flight before it stops waiting on them.
+///
+/// axum's graceful shutdown waits with no limit, and a response can stall for
+/// as long as its client stops reading: a validator client paused mid-stream,
+/// or a slow download of a whole state.
+const HTTP_DRAIN_TIMEOUT: Duration = Duration::from_secs(5);
+
 /// Bind one listener and serve `app` on it until `shutdown` is cancelled.
 ///
 /// Graceful shutdown waits for every in-flight response, so `app` gets the
@@ -142,6 +152,10 @@ pub async fn start_http_servers(
 /// finish on their own, end on it. The signal holds a child of the token, not
 /// a clone: 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 wait is bounded by [`HTTP_DRAIN_TIMEOUT`]. Past it, this returns
+/// without the stalled responses: the listener is already closed by then, and
+/// their connection tasks end with the runtime.
 async fn serve(
     address: IpAddr,
     port: u16,
@@ -152,11 +166,23 @@ async fn serve(
     let listener = tokio::net::TcpListener::bind(addr).await?;
     tracing::info!(%addr, "HTTP server listening");
     let app = app.layer(Extension(events::ShutdownSignal::of(&shutdown)));
-    axum::serve(listener, app)
-        .with_graceful_shutdown(async move {
-            shutdown.cancelled().await;
-        })
-        .await
+    let server = axum::serve(listener, app)
+        .with_graceful_shutdown(shutdown.clone().cancelled_owned())
+        .into_future();
+    let drain_deadline = async move {
+        shutdown.cancelled().await;
+        tokio::time::sleep(HTTP_DRAIN_TIMEOUT).await;
+    };
+    tokio::select! {
+        served = server => served,
+        () = drain_deadline => {
+            tracing::warn!(
+                %addr,
+                "HTTP responses still in flight after the shutdown drain timeout; no longer waiting on them"
+            );
+            Ok(())
+        }
+    }
 }
 
 /// Build the API router with the given store, client version, and peer ID.
diff --git a/crates/net/rpc/tests/http_servers.rs b/crates/net/rpc/tests/http_servers.rs
index f64ec1809..96b01ac79 100644
--- a/crates/net/rpc/tests/http_servers.rs
+++ b/crates/net/rpc/tests/http_servers.rs
@@ -230,6 +230,55 @@ async fn a_connected_event_stream_does_not_hold_up_shutdown() {
         .expect("the server exits cleanly");
 }
 
+/// A response that never finishes does not hold up shutdown either.
+///
+/// The event streams end on the shutdown signal, but any response can stall
+/// for as long as its client stops reading, and axum's graceful shutdown waits
+/// for it with no limit. The server stops waiting after its drain timeout.
+#[tokio::test]
+async fn a_response_that_never_finishes_does_not_hold_up_shutdown() {
+    use axum::body::{Body, Bytes};
+    use std::convert::Infallible;
+
+    let port = free_port().await;
+    let shutdown = CancellationToken::new();
+    let config = RpcConfig {
+        http_address: LOCALHOST,
+        api_port: port,
+        metrics_port: port,
+        version: "ethlambda/test",
+    };
+    // A body that never yields and never ends, standing in for a client that
+    // stopped reading: nothing here knows about the shutdown signal.
+    let stalled = axum::routing::get(|| async {
+        Body::from_stream(futures_util::stream::pending::<Result<Bytes, Infallible>>())
+    });
+    let router = axum::Router::new().route("/stalled", stalled);
+    let served = tokio::spawn(start_http_servers(config, Some(router), shutdown.clone()));
+
+    assert!(status_line(&get(port, "/metrics").await).contains("200"));
+    let mut stream = TcpStream::connect(SocketAddr::new(LOCALHOST, port))
+        .await
+        .expect("the server accepts");
+    stream
+        .write_all(b"GET /stalled HTTP/1.1\r\nHost: localhost\r\n\r\n")
+        .await
+        .expect("the request writes");
+    let mut head = vec![0; 1024];
+    let read = stream.read(&mut head).await.expect("the headers read");
+    assert!(
+        String::from_utf8_lossy(&head[..read]).starts_with("HTTP/1.1 200"),
+        "the stalled response has started"
+    );
+
+    shutdown.cancel();
+    tokio::time::timeout(Duration::from_secs(30), served)
+        .await
+        .expect("the server stops waiting on the stalled response")
+        .expect("the server task joins")
+        .expect("the server exits cleanly");
+}
+
 /// A beacon `Store` anchored at a phase0 block, enough for the beacon router
 /// to build.
 fn beacon_store() -> ethlambda_storage::Store {
diff --git a/docs/rpc.md b/docs/rpc.md
index a878df0bb..7a1282fd4 100644
--- a/docs/rpc.md
+++ b/docs/rpc.md
@@ -133,7 +133,7 @@ Valid values are exactly the event names above: `head`, `block`, `justified_chec
 
 Events are fanned out over a single bounded broadcast channel shared by all topics. A client that reads too slowly skips past the events it missed: they are dropped for that subscriber rather than back-pressured onto the actor, so treat the stream as best-effort and re-sync via the blocks endpoints after a gap. A client that falls behind receives an SSE comment line `: error - dropped N messages` marking the gap (wire-compatible with Lighthouse) before the stream continues; re-sync via the blocks endpoints rather than trusting the skipped range. Keep-alive comments are sent periodically to hold idle connections open.
 
-The stream ends when the node shuts down, so a connected client never holds up a graceful shutdown; expect the connection to close and reconnect once the node is back.
+The stream ends when the node shuts down; expect the connection to close and reconnect once the node is back. A client that has stopped reading can still stall its response, as it can any other, so the HTTP servers wait at most `HTTP_DRAIN_TIMEOUT` (`crates/net/rpc/src/lib.rs`) for responses still in flight before they stop waiting.
 
 Because the ring buffer is shared, the high-rate `attestation` events (roughly one per validator per slot) dominate its occupancy: a subscriber's tolerable stall is `capacity / total_event_rate`, not per-topic, so filtering with `?topics=` narrows what you receive but does **not** widen the lag window against an attestation flood. Subscribers that only need low-rate topics (`head`, `finalized_checkpoint`, …) are still evicted at the aggregate rate. If real usage shows this biting, the fix is a per-topic channel split behind the event bus (the `subscribe(TopicSet)` API is unaffected).
 

From 16f0e05501f7714d1a7bf8be2078532772e8b785 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?=
 <47506558+MegaRedHand@users.noreply.github.com>
Date: Mon, 2 Nov 2026 17:43:06 -0300
Subject: [PATCH 6/6] test(rpc): check event streams end on shutdown with
 separate ports too

The shutdown test used one port for both routers, so it only covered the
merged listener. The default ports go through the other branch of
start_http_servers, two serve calls under try_join!, and a change that
layered the signal onto only one of them would have passed. The test now
runs both shapes. Its bound also drops well under the new drain timeout,
so it is the stream ending on the signal that lets the servers stop, not
the servers giving up on the stream.
---
 crates/net/rpc/tests/http_servers.rs | 40 ++++++++++++++++++++++------
 1 file changed, 32 insertions(+), 8 deletions(-)

diff --git a/crates/net/rpc/tests/http_servers.rs b/crates/net/rpc/tests/http_servers.rs
index 96b01ac79..4c09d1692 100644
--- a/crates/net/rpc/tests/http_servers.rs
+++ b/crates/net/rpc/tests/http_servers.rs
@@ -182,7 +182,8 @@ async fn the_beacon_router_replaces_the_lean_one() {
     );
 }
 
-/// A subscriber left connected to `/eth/v1/events` does not hold up shutdown.
+/// A subscriber left connected to `/eth/v1/events` does not hold up shutdown,
+/// with the API and metrics routers merged onto one listener.
 ///
 /// Graceful shutdown waits for every in-flight response, and an event stream
 /// never finishes on its own: before the streams ended on the shutdown token, a
@@ -191,11 +192,33 @@ async fn the_beacon_router_replaces_the_lean_one() {
 #[tokio::test]
 async fn a_connected_event_stream_does_not_hold_up_shutdown() {
     let port = free_port().await;
+    assert_an_event_stream_ends_on_shutdown(port, port).await;
+}
+
+/// The same with the API and metrics routers on two listeners, the default
+/// ports' shape, which starts two servers rather than one.
+#[tokio::test]
+async fn a_connected_event_stream_does_not_hold_up_shutdown_on_separate_ports() {
+    let api_port = free_port().await;
+    let metrics_port = free_port().await;
+    assert_ne!(
+        api_port, metrics_port,
+        "the probe handed out one port twice"
+    );
+    assert_an_event_stream_ends_on_shutdown(api_port, metrics_port).await;
+}
+
+/// Opens `/eth/v1/events` on `api_port`, cancels the shutdown token, and
+/// checks the servers stop.
+///
+/// The bound is well under the servers' drain timeout, so it is the stream
+/// ending on the signal that lets them stop, not the servers giving up on it.
+async fn assert_an_event_stream_ends_on_shutdown(api_port: u16, metrics_port: u16) {
     let shutdown = CancellationToken::new();
     let config = RpcConfig {
         http_address: LOCALHOST,
-        api_port: port,
-        metrics_port: port,
+        api_port,
+        metrics_port,
         version: "ethlambda/test",
     };
     // Kept alive for the whole test, so the channel never closes and only the
@@ -206,9 +229,10 @@ async fn a_connected_event_stream_does_not_hold_up_shutdown() {
             .layer(axum::Extension(events.clone()));
     let served = tokio::spawn(start_http_servers(config, Some(router), shutdown.clone()));
 
-    // Wait for the metrics route so the socket is accepting, then subscribe.
-    assert!(status_line(&get(port, "/metrics").await).contains("200"));
-    let mut stream = TcpStream::connect(SocketAddr::new(LOCALHOST, port))
+    // Wait for both listeners to accept, then subscribe.
+    assert!(status_line(&get(metrics_port, "/metrics").await).contains("200"));
+    assert!(status_line(&get(api_port, "/eth/v1/node/version").await).contains("200"));
+    let mut stream = TcpStream::connect(SocketAddr::new(LOCALHOST, api_port))
         .await
         .expect("the server accepts");
     stream
@@ -223,9 +247,9 @@ async fn a_connected_event_stream_does_not_hold_up_shutdown() {
     );
 
     shutdown.cancel();
-    tokio::time::timeout(Duration::from_secs(5), served)
+    tokio::time::timeout(Duration::from_secs(2), served)
         .await
-        .expect("the server stops within five seconds despite the open stream")
+        .expect("the servers stop promptly despite the open stream")
         .expect("the server task joins")
         .expect("the server exits cleanly");
 }