feat(metrics): measure fcu lateness and missed flashblocks

crates/rproxy/src/server/metrics.rs

@@ -2,7 +2,12 @@ mod candlestick;
 
 // ---------------------------------------------------------------------
 
-use std::{borrow::Cow, net::TcpListener, sync::Arc, time::Duration};
+use std::{
+    borrow::Cow,
+    net::TcpListener,
+    sync::{Arc, atomic::AtomicI64},
+    time::Duration,
+};
 
 use actix_web::{
     App,
@@ -53,6 +58,14 @@ pub(crate) struct Metrics {
     pub(crate) http_request_decompressed_size: Family<LabelsProxyHttpJrpc, Candlestick>,
     pub(crate) http_response_decompressed_size: Family<LabelsProxyHttpJrpc, Candlestick>,
 
+    pub(crate) fcu_arrival: Family<LabelsProxy, Candlestick>,
+    pub(crate) fcu_reduced_flashblocks: Family<LabelsProxy, Counter>,
+
+    // Highest block timestamp (unix seconds) of the last fcu-with-attributes measured
+    // so we only measure fcu with attributes once per block timestamp. Assumes block
+    // timestamp requests increase monotonically and does not account for reorgs
+    pub(crate) fcu_block_timestamp_latest: AtomicI64,
+
     pub(crate) tls_certificate_valid_not_before: Gauge,
     pub(crate) tls_certificate_valid_not_after: Gauge,
 
@@ -97,6 +110,11 @@ impl Metrics {
             http_request_decompressed_size: Family::default(),
             http_response_decompressed_size: Family::default(),
 
+            fcu_arrival: Family::default(),
+            fcu_reduced_flashblocks: Family::default(),
+
+            fcu_block_timestamp_latest: AtomicI64::new(0),
+
             tls_certificate_valid_not_before: Gauge::default(),
             tls_certificate_valid_not_after: Gauge::default(),
 
@@ -219,6 +237,19 @@ impl Metrics {
             this.http_response_decompressed_size.clone(),
         );
 
+        this.registry.register_with_unit(
+            "fcu_arrival",
+            "how late the first forkchoice-update with attributes arrives into a block (relative to the block's building window start)",
+            Unit::Other(String::from("milliseconds")),
+            this.fcu_arrival.clone(),
+        );
+
+        this.registry.register(
+            "fcu_reduced_flashblocks",
+            "count of flashblocks missed as a result of late forkchoice-updates with attributes",
+            this.fcu_reduced_flashblocks.clone(),
+        );
+
         this.registry.register(
             "tls_certificate_valid_not_before",
             "tls certificate's not-valid-before timestamp",

crates/rproxy/src/server/proxy/config/authrpc.rs

@@ -52,6 +52,20 @@ pub(crate) struct ConfigAuthrpc {
     )]
     pub(crate) backend_timeout: Duration,
 
+    /// duration of a block; combined with --authrpc-flashblocks-per-block
+    /// to measure how late a forkchoice-update with attributes arrives
+    /// into a block
+    #[arg(
+        default_value = "1s",
+        env = "RPROXY_AUTHRPC_BLOCK_TIME",
+        help_heading = "authrpc",
+        long("authrpc-block-time"),
+        name("authrpc_block_time"),
+        value_name = "duration",
+        value_parser = humantime::parse_duration
+    )]
+    pub(crate) block_time: Duration,
+
     /// whether authrpc proxy should deduplicate incoming fcus w/o payload
     /// (mitigates fcu avalanche issue)
     #[arg(
@@ -71,6 +85,21 @@ pub(crate) struct ConfigAuthrpc {
     )]
     pub(crate) enabled: bool,
 
+    /// number of flashblocks in a block; when > 0 the authrpc proxy
+    /// measures how late the first forkchoice-update with attributes
+    /// arrives into each block (relative to --authrpc-block-time) and
+    /// emits the count of flashblocks missed as a result (0 disables
+    /// the measurement)
+    #[arg(
+        default_value = "0",
+        env = "RPROXY_AUTHRPC_FLASHBLOCKS_PER_BLOCK",
+        help_heading = "authrpc",
+        long("authrpc-flashblocks-per-block"),
+        name("authrpc_flashblocks_per_block"),
+        value_name = "count"
+    )]
+    pub(crate) flashblocks_per_block: u64,
+
     /// duration to keep idle authrpc connections open (0 means no
     /// keep-alive)
     #[arg(
@@ -282,6 +311,11 @@ impl ConfigAuthrpc {
             })
         });
 
+        // fcu-lateness measurement
+        if self.flashblocks_per_block > 0 && self.block_time.is_zero() {
+            errs.push(ConfigAuthrpcError::BlockTimeZero);
+        }
+
         // mirroring_peer_urls
         for peer_url in self.mirroring_peer_urls.iter() {
             match Url::parse(peer_url) {
@@ -383,6 +417,16 @@ impl ConfigProxyHttp for ConfigAuthrpc {
         self.backend_url.parse::<Url>().expect(ALREADY_VALIDATED)
     }
 
+    #[inline]
+    fn block_time(&self) -> Duration {
+        self.block_time
+    }
+
+    #[inline]
+    fn flashblocks_per_block(&self) -> u64 {
+        self.flashblocks_per_block
+    }
+
     #[inline]
     fn idle_connection_timeout(&self) -> Duration {
         self.idle_connection_timeout
@@ -474,6 +518,9 @@ pub(crate) enum ConfigAuthrpcError {
     #[error("invalid authrpc backend url '{url}': {err}")]
     BackendUrlInvalid { url: String, err: url::ParseError },
 
+    #[error("authrpc block time must be greater than zero when flashblocks-per-block is set")]
+    BlockTimeZero,
+
     #[error("invalid authrpc backend url '{url}': host is missing")]
     BackendUrlMissesHost { url: String },
 

crates/rproxy/src/server/proxy/config/rpc.rs

@@ -384,6 +384,17 @@ impl ConfigProxyHttp for ConfigRpc {
         self.backend_url.parse::<Url>().expect(ALREADY_VALIDATED)
     }
 
+    // fcu-lateness measurement is authrpc-only
+    #[inline]
+    fn block_time(&self) -> Duration {
+        Duration::ZERO
+    }
+
+    #[inline]
+    fn flashblocks_per_block(&self) -> u64 {
+        0
+    }
+
     #[inline]
     fn idle_connection_timeout(&self) -> Duration {
         self.idle_connection_timeout

crates/rproxy/src/server/proxy/http/config.rs

@@ -8,6 +8,8 @@ pub(crate) trait ConfigProxyHttp: Clone + Send + Unpin + 'static {
     fn backend_max_concurrent_requests(&self) -> usize;
     fn backend_timeout(&self) -> Duration;
     fn backend_url(&self) -> Url;
+    fn block_time(&self) -> Duration;
+    fn flashblocks_per_block(&self) -> u64;
     fn idle_connection_timeout(&self) -> Duration;
     fn keepalive_interval(&self) -> Duration;
     fn keepalive_retries(&self) -> u32;

crates/rproxy/src/server/proxy/http/proxy.rs

@@ -97,18 +97,12 @@ impl ProxyHttpMetrics {
         let labels = LabelsProxy { proxy: proxy_name };
         Self {
             proxy_name,
-            in_flight_client: metrics
-                .http_in_flight_requests_client
-                .get_or_create(&labels)
-                .clone(),
+            in_flight_client: metrics.http_in_flight_requests_client.get_or_create(&labels).clone(),
             in_flight_backend: metrics
                 .http_in_flight_requests_backend
                 .get_or_create(&labels)
                 .clone(),
-            proxy_failure_count: metrics
-                .http_proxy_failure_count
-                .get_or_create(&labels)
-                .clone(),
+            proxy_failure_count: metrics.http_proxy_failure_count.get_or_create(&labels).clone(),
             client_info_family: metrics.client_info.clone(),
             client_info_cache: HashMap::default(),
         }
@@ -238,14 +232,7 @@ where
 
         let metrics = Arc::new(ProxyHttpMetrics::new(P::name(), &shared.metrics));
 
-        Self {
-            id,
-            shared,
-            backend,
-            requests: HashMap::default(),
-            postprocessor,
-            metrics,
-        }
+        Self { id, shared, backend, requests: HashMap::default(), postprocessor, metrics }
     }
 
     pub(crate) async fn run(
@@ -881,7 +868,13 @@ where
                     worker_id,
                 );
 
-                Self::emit_metrics_on_proxy_success(&jrpc, &clnt_req, &bknd_res, metrics.clone());
+                Self::emit_metrics_on_proxy_success(
+                    &inner,
+                    &jrpc,
+                    &clnt_req,
+                    &bknd_res,
+                    metrics.clone(),
+                );
             }
 
             Err(err) => {
@@ -1153,6 +1146,7 @@ where
     }
 
     fn emit_metrics_on_proxy_success(
+        inner: &P,
         jrpc: &JrpcRequestMetaMaybeBatch,
         req: &ProxiedHttpRequest,
         res: &ProxiedHttpResponse,
@@ -1223,6 +1217,75 @@ where
             .http_response_decompressed_size
             .get_or_create_owned(&metric_labels_jrpc)
             .record(res.decompressed_size as i64);
+
+        // measure how late the first fcu-with-attributes lands in each block and
+        // how many flashblocks that costs
+        Self::maybe_emit_fcu_metrics(
+            inner.config().flashblocks_per_block(),
+            inner.config().block_time(),
+            jrpc,
+            req,
+            &metrics,
+        );
+    }
+
+    fn maybe_emit_fcu_metrics(
+        flashblocks_per_block: u64,
+        block_time: Duration,
+        jrpc: &JrpcRequestMetaMaybeBatch,
+        req: &ProxiedHttpRequest,
+        metrics: &Metrics,
+    ) {
+        if flashblocks_per_block == 0 || block_time.is_zero() {
+            return;
+        }
+
+        // only a single fcu carrying payload attributes has a block timestamp
+        let JrpcRequestMetaMaybeBatch::Single(jrpc) = jrpc else {
+            return;
+        };
+        if !jrpc.method_enriched().ends_with("withPayload") {
+            return;
+        }
+
+        let params = jrpc.params();
+        if params.len() < 2 {
+            return;
+        }
+        let Some(block_ts_secs) = params[1]
+            .as_object()
+            .and_then(|attrs| attrs.get("timestamp"))
+            .and_then(|ts| ts.as_str())
+            .and_then(|s| {
+                let hex = s.strip_prefix("0x").or_else(|| s.strip_prefix("0X")).unwrap_or(s);
+                i64::from_str_radix(hex, 16).ok()
+            })
+        else {
+            return;
+        };
+
+        // scope to the first fcu-with-attributes per block: only proceed when we
+        // advance the (monotonic) higher timestamp
+        let prev = metrics.fcu_block_timestamp_latest.fetch_max(block_ts_secs, Ordering::Relaxed);
+        if block_ts_secs <= prev {
+            return;
+        }
+
+        let arrival_ms = (req.start().unix_timestamp_nanos() / 1_000_000) as i64;
+        let block_ts_ms = block_ts_secs * 1_000;
+        let block_time_ms = block_time.as_millis() as i64;
+
+        let ms_into_slot = arrival_ms - (block_ts_ms - block_time_ms);
+        let interval_ms = block_time_ms / flashblocks_per_block as i64;
+        let remaining_flashblocks = ((block_ts_ms - arrival_ms) / interval_ms)
+            .clamp(0, flashblocks_per_block as i64) as u64;
+        let reduced_flashblocks = flashblocks_per_block - remaining_flashblocks;
+
+        let labels = LabelsProxy { proxy: P::name() };
+        metrics.fcu_arrival.get_or_create(&labels).record(ms_into_slot);
+        if reduced_flashblocks > 0 {
+            metrics.fcu_reduced_flashblocks.get_or_create(&labels).inc_by(reduced_flashblocks);
+        }
     }
 }
 
@@ -1399,24 +1462,28 @@ where
 
         // Build the inner TCP connector ourselves so we can flip
         // TCP_NODELAY on after each connect
-        let tcp_nodelay = actix_tls::connect::Connector::new(
-            actix_tls::connect::Resolver::default(),
-        )
-        .service()
-        .map(|conn: actix_tls::connect::Connection<awc::http::Uri, tokio::net::TcpStream>| {
-            // Mirror the client leg (ConnectionGuard::on_connect): fail open on
-            // a set_nodelay error, but log it. A silent failure here would
-            // quietly reintroduce the Nagle/DELACK tail with no signal — most
-            // relevant if the backend ever moves off loopback.
-            if let Err(err) = conn.io_ref().set_nodelay(true) {
-                debug!(
-                    proxy = P::name(),
-                    error = ?err,
-                    "Failed to set TCP_NODELAY on backend connection",
+        let tcp_nodelay =
+            actix_tls::connect::Connector::new(actix_tls::connect::Resolver::default())
+                .service()
+                .map(
+                    |conn: actix_tls::connect::Connection<
+                        awc::http::Uri,
+                        tokio::net::TcpStream,
+                    >| {
+                        // Mirror the client leg (ConnectionGuard::on_connect): fail open on
+                        // a set_nodelay error, but log it. A silent failure here would
+                        // quietly reintroduce the Nagle/DELACK tail with no signal — most
+                        // relevant if the backend ever moves off loopback.
+                        if let Err(err) = conn.io_ref().set_nodelay(true) {
+                            debug!(
+                                proxy = P::name(),
+                                error = ?err,
+                                "Failed to set TCP_NODELAY on backend connection",
+                            );
+                        }
+                        conn
+                    },
                 );
-            }
-            conn
-        });
 
         let client = Client::builder()
             .add_default_header((header::HOST, host))