From eddf84b283b4ec402e1dc8de5a0db590412b3746 Mon Sep 17 00:00:00 2001 From: avalonche Date: Mon, 6 Jul 2026 21:16:09 -0700 Subject: [PATCH] feat(authrpc): measure fcu lateness and missed flashblocks Add authrpc config for block time and flashblocks-per-block, and emit two metrics from the proxy postprocessor (off the client-response hot path): - fcu_lateness (ms): how late the first forkchoice-update with attributes lands into each block's building window, anchored on the payload's block timestamp - fcu_missed_flashblocks: flashblocks that no longer fit before the block's sealing deadline as a result (deadline-based attribution) Scoped to the first fcu-with-attributes per block via a monotonic block-timestamp high-water mark, with a clock-skew guard. Measurement is opt-in (--authrpc-flashblocks-per-block defaults to 0) and inert for the rpc proxy. block_time/flashblocks_per_block are exposed on the ConfigProxyHttp trait so the shared emit path can read them. Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/rproxy/src/server/metrics.rs | 33 ++++- .../rproxy/src/server/proxy/config/authrpc.rs | 47 ++++++ crates/rproxy/src/server/proxy/config/rpc.rs | 11 ++ crates/rproxy/src/server/proxy/http/config.rs | 2 + crates/rproxy/src/server/proxy/http/proxy.rs | 135 +++++++++++++----- 5 files changed, 193 insertions(+), 35 deletions(-) diff --git a/crates/rproxy/src/server/metrics.rs b/crates/rproxy/src/server/metrics.rs index 638846d..d8971fe 100644 --- a/crates/rproxy/src/server/metrics.rs +++ b/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, pub(crate) http_response_decompressed_size: Family, + pub(crate) fcu_arrival: Family, + pub(crate) fcu_reduced_flashblocks: Family, + + // 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", diff --git a/crates/rproxy/src/server/proxy/config/authrpc.rs b/crates/rproxy/src/server/proxy/config/authrpc.rs index abc0167..6c6cc00 100644 --- a/crates/rproxy/src/server/proxy/config/authrpc.rs +++ b/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::().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 }, diff --git a/crates/rproxy/src/server/proxy/config/rpc.rs b/crates/rproxy/src/server/proxy/config/rpc.rs index 3775bc5..585440d 100644 --- a/crates/rproxy/src/server/proxy/config/rpc.rs +++ b/crates/rproxy/src/server/proxy/config/rpc.rs @@ -384,6 +384,17 @@ impl ConfigProxyHttp for ConfigRpc { self.backend_url.parse::().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 diff --git a/crates/rproxy/src/server/proxy/http/config.rs b/crates/rproxy/src/server/proxy/http/config.rs index 704758a..2b0449a 100644 --- a/crates/rproxy/src/server/proxy/http/config.rs +++ b/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; diff --git a/crates/rproxy/src/server/proxy/http/proxy.rs b/crates/rproxy/src/server/proxy/http/proxy.rs index 7ae5839..c274ef3 100644 --- a/crates/rproxy/src/server/proxy/http/proxy.rs +++ b/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| { - // 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))