merge queue: embarking release-v8.1 (c25a975) and #7892 together

beacon_node/http_api/src/lib.rs

@@ -3226,7 +3226,19 @@ pub fn serve<T: BeaconChainTypes>(
 
                     let s = futures::stream::select_all(receivers);
 
-                    Ok(warp::sse::reply(warp::sse::keep_alive().stream(s)))
+                    let response = warp::sse::reply(warp::sse::keep_alive().stream(s));
+
+                    // Set headers to bypass nginx caching and buffering, which breaks realtime
+                    // delivery.
+                    let response = warp::reply::with_header(response, "X-Accel-Buffering", "no");
+                    let response = warp::reply::with_header(response, "X-Accel-Expires", "0");
+                    let response = warp::reply::with_header(
+                        response,
+                        "Cache-Control",
+                        "no-cache, no-store, must-revalidate",
+                    );
+
+                    Ok(response)
                 })
             },
         );

book/src/help_vc.md

@@ -185,6 +185,12 @@ Flags:
           If present, do not attempt to discover new validators in the
           validators-dir. Validators will need to be manually added to the
           validator_definitions.yml file.
+      --disable-beacon-head-monitor
+          Disable the beacon head monitor which tries to attest as soon as any
+          of the configured beacon nodes sends a head event. Leaving the service
+          enabled is recommended, but disabling it can lead to reduced bandwidth
+          and more predictable usage of the primary beacon node (rather than the
+          fastest BN).
       --disable-latency-measurement-service
           Disables the service that periodically attempts to measure latency to
           BNs.

common/eth2/src/error.rs

@@ -17,6 +17,8 @@ pub enum Error {
     #[cfg(feature = "events")]
     /// The `reqwest_eventsource` client raised an error.
     SseClient(Box<reqwest_eventsource::Error>),
+    #[cfg(feature = "events")]
+    SseEventSource(reqwest_eventsource::CannotCloneRequestError),
     /// The server returned an error message where the body was able to be parsed.
     ServerMessage(ErrorMessage),
     /// The server returned an error message with an array of errors.
@@ -100,6 +102,7 @@ impl Error {
                     None
                 }
             }
+            Error::SseEventSource(_) => None,
             Error::ServerMessage(msg) => StatusCode::try_from(msg.code).ok(),
             Error::ServerIndexedMessage(msg) => StatusCode::try_from(msg.code).ok(),
             Error::StatusCode(status) => Some(*status),

common/eth2/src/lib.rs

@@ -40,7 +40,7 @@ use reqwest::{
     header::{HeaderMap, HeaderValue},
 };
 #[cfg(feature = "events")]
-use reqwest_eventsource::{Event, EventSource};
+use reqwest_eventsource::{Event, RequestBuilderExt};
 use serde::{Serialize, de::DeserializeOwned};
 use ssz::Encode;
 use std::fmt;
@@ -76,6 +76,8 @@ const HTTP_GET_BEACON_BLOCK_SSZ_TIMEOUT_QUOTIENT: u32 = 4;
 const HTTP_GET_DEBUG_BEACON_STATE_QUOTIENT: u32 = 4;
 const HTTP_GET_DEPOSIT_SNAPSHOT_QUOTIENT: u32 = 4;
 const HTTP_GET_VALIDATOR_BLOCK_TIMEOUT_QUOTIENT: u32 = 4;
+// Generally the timeout for events should be longer than a slot.
+const HTTP_GET_EVENTS_TIMEOUT_MULTIPLIER: u32 = 50;
 const HTTP_DEFAULT_TIMEOUT_QUOTIENT: u32 = 4;
 
 /// A struct to define a variety of different timeouts for different validator tasks to ensure
@@ -96,6 +98,7 @@ pub struct Timeouts {
     pub get_debug_beacon_states: Duration,
     pub get_deposit_snapshot: Duration,
     pub get_validator_block: Duration,
+    pub events: Duration,
     pub default: Duration,
 }
 
@@ -116,6 +119,7 @@ impl Timeouts {
             get_debug_beacon_states: timeout,
             get_deposit_snapshot: timeout,
             get_validator_block: timeout,
+            events: HTTP_GET_EVENTS_TIMEOUT_MULTIPLIER * timeout,
             default: timeout,
         }
     }
@@ -138,6 +142,7 @@ impl Timeouts {
             get_debug_beacon_states: base_timeout / HTTP_GET_DEBUG_BEACON_STATE_QUOTIENT,
             get_deposit_snapshot: base_timeout / HTTP_GET_DEPOSIT_SNAPSHOT_QUOTIENT,
             get_validator_block: base_timeout / HTTP_GET_VALIDATOR_BLOCK_TIMEOUT_QUOTIENT,
+            events: HTTP_GET_EVENTS_TIMEOUT_MULTIPLIER * base_timeout,
             default: base_timeout / HTTP_DEFAULT_TIMEOUT_QUOTIENT,
         }
     }
@@ -2800,7 +2805,12 @@ impl BeaconNodeHttpClient {
             .join(",");
         path.query_pairs_mut().append_pair("topics", &topic_string);
 
-        let mut es = EventSource::get(path);
+        let mut es = self
+            .client
+            .get(path)
+            .timeout(self.timeouts.events)
+            .eventsource()
+            .map_err(Error::SseEventSource)?;
         // If we don't await `Event::Open` here, then the consumer
         // will not get any Message events until they start awaiting the stream.
         // This is a way to register the stream with the sse server before

lighthouse/tests/validator_client.rs

@@ -758,3 +758,21 @@ fn validator_proposer_nodes() {
             );
         });
 }
+
+// Head monitor is enabled by default.
+#[test]
+fn head_monitor_default() {
+    CommandLineTest::new().run().with_config(|config| {
+        assert!(config.enable_beacon_head_monitor);
+    });
+}
+
+#[test]
+fn head_monitor_disabled() {
+    CommandLineTest::new()
+        .flag("disable-beacon-head-monitor", None)
+        .run()
+        .with_config(|config| {
+            assert!(!config.enable_beacon_head_monitor);
+        });
+}

validator_client/beacon_node_fallback/Cargo.toml

@@ -11,7 +11,7 @@ path = "src/lib.rs"
 [dependencies]
 bls = { workspace = true }
 clap = { workspace = true }
-eth2 = { workspace = true }
+eth2 = { workspace = true, features = ["events"] }
 futures = { workspace = true }
 itertools = { workspace = true }
 sensitive_url = { workspace = true }