diff --git a/crates/net/rpc/src/beacon/events.rs b/crates/net/rpc/src/beacon/events.rs index 5b7902079..1122b975a 100644 --- a/crates/net/rpc/src/beacon/events.rs +++ b/crates/net/rpc/src/beacon/events.rs @@ -29,13 +29,16 @@ use ethlambda_blockchain::{EventBus, Topic}; use ethlambda_storage::Store; use super::ApiError; +use crate::events::ShutdownSignal; async fn get_events( Extension(events): Extension, + shutdown: Option>, Query(params): Query>, ) -> Response { + let shutdown = shutdown.map(|Extension(signal)| signal); match requested_topics(¶ms) { - Ok(topics) => crate::events::sse_response(&events, topics), + Ok(topics) => crate::events::sse_response(&events, topics, shutdown), Err(err) => err.into_response(), } } @@ -94,6 +97,8 @@ 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())); diff --git a/crates/net/rpc/src/events.rs b/crates/net/rpc/src/events.rs index ca9069343..74e0f59e4 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,39 @@ 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, WaitForCancellationFutureOwned}; + +/// 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. 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. +/// +/// 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)] +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 { @@ -40,6 +74,7 @@ struct EventsParams { async fn get_events( Extension(events): Extension, + shutdown: Option>, Query(params): Query, ) -> Response { // `topics` is required, matching the Beacon API: a missing or empty value @@ -65,18 +100,30 @@ async fn get_events( }, }; - sse_response(&events, topics) + 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. +/// `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 /// [`Sse`] already sets `content-type: text/event-stream` and /// `cache-control: no-cache`. -pub(crate) fn sse_response(events: &EventBus, topics: Vec) -> Response { - let mut response = Sse::new(event_stream(events.subscribe(), topics)) +pub(crate) fn sse_response( + events: &EventBus, + topics: Vec, + shutdown: Option, +) -> Response { + 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(); response @@ -125,7 +172,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 +191,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,6 +208,8 @@ mod tests { create_test_state(), DEFAULT_MILLISECONDS_PER_SLOT, ); + // 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 @@ -396,4 +447,21 @@ 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 server = tokio_util::sync::CancellationToken::new(); + let shutdown = super::ShutdownSignal::of(&server); + let resp = super::sse_response(&events, vec![Topic::Head], Some(shutdown)); + + server.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..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,7 +137,25 @@ 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 +/// token as a [`events::ShutdownSignal`] too: the event streams, which never +/// 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, @@ -145,11 +165,24 @@ async fn serve( let addr = SocketAddr::new(address, port); let listener = tokio::net::TcpListener::bind(addr).await?; tracing::info!(%addr, "HTTP server listening"); - axum::serve(listener, app) - .with_graceful_shutdown(async move { - shutdown.cancelled().await; - }) - .await + let app = app.layer(Extension(events::ShutdownSignal::of(&shutdown))); + 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 6171b1730..4c09d1692 100644 --- a/crates/net/rpc/tests/http_servers.rs +++ b/crates/net/rpc/tests/http_servers.rs @@ -146,12 +146,171 @@ 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, +/// 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 +/// 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; + 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, + metrics_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 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 + .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(2), served) + .await + .expect("the servers stop promptly despite the open stream") + .expect("the server task joins") + .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::>()) + }); + 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 { 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 +339,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..7a1282fd4 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; 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). ### `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.