diff --git a/src/cli.rs b/src/cli.rs index 5c61eae4..3f7f73de 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -9,7 +9,8 @@ use rbuilder_utils::clickhouse::indexer::{ use crate::{ indexer::{BUNDLE_RECEIPTS_TABLE_NAME, BUNDLE_TABLE_NAME}, - SystemBundleDecoder, + ingress, + primitives::SystemBundleDecoder, }; /// The maximum request size in bytes (10 MiB). @@ -266,15 +267,25 @@ pub struct OrderflowIngressArgs { #[clap(long = "http.enable-gzip", default_value_t = false)] pub gzip_enabled: bool, - /// Maximum local ClickHouse backup disk size in MB above which user RPC (e.g. eth_sendBundle) - /// is rejected with disk full. Defaults to 1024 MB (1 GiB). + /// ClickHouse backup disk size in MB above which user RPC is rejected. Defaults to 1024 MB. #[clap( - long = "disk-max-size-to-accept-user-rpc-mb", + long = "disk-backup-size-reject-flow-threshold-mb", default_value_t = 1024, - env = "DISK_MAX_SIZE_TO_ACCEPT_USER_RPC", - id = "DISK_MAX_SIZE_TO_ACCEPT_USER_RPC" + env = "DISK_BACKUP_SIZE_REJECT_FLOW_THRESHOLD_MB", + id = "DISK_BACKUP_SIZE_REJECT_FLOW_THRESHOLD_MB" )] - pub disk_max_size_to_accept_user_rpc_mb: u64, + pub disk_backup_size_reject_flow_threshold_mb: u64, + + /// ClickHouse backup disk size in MB below which user RPC is accepted again after being + /// rejected. Must be less than or equal to disk-backup-size-reject-flow-threshold-mb. + /// Defaults to 512 MB. + #[clap( + long = "disk-backup-size-to-resume-flow-threshold-mb", + default_value_t = 512, + env = "DISK_BACKUP_SIZE_TO_RESUME_FLOW_THRESHOLD_MB", + id = "DISK_BACKUP_SIZE_TO_RESUME_FLOW_THRESHOLD_MB" + )] + pub disk_backup_size_resume_flow_threshold_mb: u64, /// The interval in seconds to update the peer list from BuilderHub. #[clap( @@ -344,7 +355,8 @@ impl Default for OrderflowIngressArgs { score_bucket_s: 4, log_json: false, gzip_enabled: false, - disk_max_size_to_accept_user_rpc_mb: 1024, + disk_backup_size_reject_flow_threshold_mb: 1024, + disk_backup_size_resume_flow_threshold_mb: 512, tcp_small_clients: NonZero::new(4).expect("non-zero"), tcp_big_clients: 0, io_threads: 4, @@ -361,6 +373,46 @@ impl Default for OrderflowIngressArgs { } impl OrderflowIngressArgs { + pub(crate) fn ingress_config(&self) -> eyre::Result { + use eyre::WrapErr as _; + use reqwest::Url; + + let local_builder_url = self + .builder_url + .as_ref() + .map(|url| Url::parse(url)) + .transpose() + .wrap_err("invalid builder URL")?; + let builder_ready_endpoint = self + .builder_ready_endpoint + .as_ref() + .map(|url| Url::parse(url)) + .transpose() + .wrap_err("invalid builder ready endpoint URL")?; + Ok(ingress::Config { + gzip_enabled: self.gzip_enabled, + rate_limiting_enabled: self.enable_rate_limiting, + rate_limit_lookback_s: self.rate_limit_lookback_s, + rate_limit_count: self.rate_limit_count, + score_lookback_s: self.score_lookback_s, + score_bucket_s: self.score_bucket_s, + max_txs_per_bundle: self.max_txs_per_bundle, + flashbots_signer: self.flashbots_signer, + local_builder_url, + builder_ready_endpoint, + disk_backup_size_reject_flow_threshold: self + .disk_backup_size_reject_flow_threshold_mb + .saturating_mul(1024 * 1024), + disk_backup_size_resume_flow_threshold: self + .disk_backup_size_resume_flow_threshold_mb + .saturating_mul(1024 * 1024), + order_cache_ttl: self.cache.order_cache_ttl, + order_cache_size: self.cache.order_cache_size, + signer_cache_ttl: self.cache.signer_cache_ttl, + signer_cache_size: self.cache.signer_cache_size, + }) + } + /// Set max request size. pub fn max_request_size(mut self, max: usize) -> Self { self.max_request_size = max; diff --git a/src/indexer/click/mod.rs b/src/indexer/click/mod.rs index 4e6a55db..c40572e6 100644 --- a/src/indexer/click/mod.rs +++ b/src/indexer/click/mod.rs @@ -628,19 +628,21 @@ pub(crate) mod tests { } } - /// Integration test: when backup DB exceeds disk_max_size_to_accept_user_rpc_mb, user RPC - /// returns DiskFull; after fixing ClickHouse and draining backup, RPC accepts again. + /// Integration test: when backup DB exceeds disk_backup_size_reject_flow_threshold_mb, + /// user RPC returns DiskFull; after fixing ClickHouse and draining backup below + /// disk_backup_size_resume_flow_threshold_mb, RPC accepts again. /// A little long func.. consider improving this. #[tokio::test(flavor = "multi_thread")] async fn disk_full_rpc_rejects_then_accepts_after_drain() { const BACKUP_DISK_MAX_SIZE_BYTES: u64 = 5; const BUNDLE_TX_COUNT: usize = 10; const BUNDLE_TX_INPUT_SIZE: usize = 1024; - const DISK_MAX_SIZE_TO_ACCEPT_USER_RPC_MB: u64 = 1; - // We need to fill DISK_MAX_SIZE_TO_ACCEPT_USER_RPC_MB with bundles. + const DISK_BACKUP_SIZE_REJECT_FLOW_THRESHOLD_MB: u64 = 1; + const DISK_BACKUP_SIZE_RESUME_FLOW_THRESHOLD_MB: u64 = 0; + // We need to fill DISK_BACKUP_SIZE_REJECT_FLOW_THRESHOLD_MB with bundles. // 2 is to play it safe. const BUNDLE_COUNT_TO_FILL_DISK: usize = 2 * - (DISK_MAX_SIZE_TO_ACCEPT_USER_RPC_MB * 1024 * 1024) as usize / + (DISK_BACKUP_SIZE_REJECT_FLOW_THRESHOLD_MB * 1024 * 1024) as usize / (BUNDLE_TX_COUNT * BUNDLE_TX_INPUT_SIZE); const FLOWPROXY_START_DELAY_MS: Duration = Duration::from_millis(800); @@ -673,7 +675,8 @@ pub(crate) mod tests { let mut args = OrderflowIngressArgs::default().gzip_enabled().disable_builder_hub(); args.peer_update_interval_s = 5; args.indexing = IndexerArgs { clickhouse: Some(clickhouse_args), parquet: None }; - args.disk_max_size_to_accept_user_rpc_mb = DISK_MAX_SIZE_TO_ACCEPT_USER_RPC_MB; // 1 MiB threshold + args.disk_backup_size_reject_flow_threshold_mb = DISK_BACKUP_SIZE_REJECT_FLOW_THRESHOLD_MB; + args.disk_backup_size_resume_flow_threshold_mb = DISK_BACKUP_SIZE_RESUME_FLOW_THRESHOLD_MB; args.builder_url = Some(builder_url.clone()); let user_listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); diff --git a/src/ingress/mod.rs b/src/ingress/mod.rs index 5fdb60d9..df99a529 100644 --- a/src/ingress/mod.rs +++ b/src/ingress/mod.rs @@ -9,7 +9,7 @@ use crate::{ forwarder::IngressForwarders, indexer::{click::CLICKHOUSE_LOCAL_BACKUP_DISK_SIZE, IndexerHandle, OrderIndexer as _}, jsonrpc::{JsonRpcError, JsonRpcRequest, JsonRpcResponse}, - metrics::{IngressMetrics, SYSTEM_METRICS}, + metrics::{IngressMetrics, CLICKHOUSE_METRICS, SYSTEM_METRICS}, primitives::{ decode_transaction, BundleHash as _, BundleReceipt, DecodedBundle, EthResponse, EthereumTransaction, RawBundleBitcode, Samplable, SystemBundle, SystemBundleDecoder, @@ -45,7 +45,10 @@ use std::{ io::Read as _, net::SocketAddr, str::FromStr as _, - sync::Arc, + sync::{ + atomic::{AtomicBool, Ordering}, + Arc, + }, time::{Duration, Instant}, }; use time::UtcDateTime; @@ -76,15 +79,85 @@ pub struct OrderflowIngress { pub local_builder_url: Option, pub builder_ready_endpoint: Option, pub indexer_handle: IndexerHandle, - /// Maximum local ClickHouse backup disk size in bytes above which user RPC is rejected. - pub disk_max_size_to_accept_user_rpc: u64, + /// backup disk size in bytes above which user RPC is rejected. + pub disk_backup_size_reject_flow_threshold: u64, + /// backup disk size in bytes below which user RPC is accepted again after rejection. + pub disk_backup_size_resume_flow_threshold: u64, + /// whether we are currently rejecting user RPCs due to disk size. + pub(crate) is_rejecting: AtomicBool, // Metrics pub(crate) user_metrics: IngressMetrics, pub(crate) system_metrics: IngressMetrics, } +pub(crate) struct Config { + pub(crate) gzip_enabled: bool, + pub(crate) rate_limiting_enabled: bool, + pub(crate) rate_limit_lookback_s: u64, + pub(crate) rate_limit_count: u64, + pub(crate) score_lookback_s: u64, + pub(crate) score_bucket_s: u64, + pub(crate) max_txs_per_bundle: usize, + pub(crate) flashbots_signer: Option
, + pub(crate) local_builder_url: Option, + pub(crate) builder_ready_endpoint: Option, + pub(crate) disk_backup_size_reject_flow_threshold: u64, + pub(crate) disk_backup_size_resume_flow_threshold: u64, + pub(crate) order_cache_ttl: u64, + pub(crate) order_cache_size: u64, + pub(crate) signer_cache_ttl: u64, + pub(crate) signer_cache_size: u64, +} + impl OrderflowIngress { + pub(crate) fn new( + config: Config, + pqueues: PriorityWorkers, + forwarders: IngressForwarders, + indexer_handle: IndexerHandle, + ) -> Self { + // Reject hysteresis state is not carried between restarts. We assume we are not rejecting + // and let next write update it. This is not a big problem since we don't restart + // that much. + let is_rejecting = false; + Self::update_is_rejecting_metric(is_rejecting); + CLICKHOUSE_METRICS + .disk_backup_size_reject_flow_threshold_bytes() + .set(config.disk_backup_size_reject_flow_threshold); + CLICKHOUSE_METRICS + .disk_backup_size_resume_flow_threshold_bytes() + .set(config.disk_backup_size_resume_flow_threshold); + let order_cache = OrderCache::new(config.order_cache_ttl, config.order_cache_size); + let signer_cache = SignerCache::new(config.signer_cache_ttl, config.signer_cache_size); + Self { + gzip_enabled: config.gzip_enabled, + rate_limiting_enabled: config.rate_limiting_enabled, + rate_limit_lookback_s: config.rate_limit_lookback_s, + rate_limit_count: config.rate_limit_count, + score_lookback_s: config.score_lookback_s, + score_bucket_s: config.score_bucket_s, + system_bundle_decoder: SystemBundleDecoder { + max_txs_per_bundle: config.max_txs_per_bundle, + }, + spam_thresholds: SpamThresholds::default(), + pqueues, + entities: DashMap::default(), + order_cache, + signer_cache, + forwarders, + flashbots_signer: config.flashbots_signer, + local_builder_url: config.local_builder_url, + builder_ready_endpoint: config.builder_ready_endpoint, + indexer_handle, + disk_backup_size_reject_flow_threshold: config.disk_backup_size_reject_flow_threshold, + disk_backup_size_resume_flow_threshold: config.disk_backup_size_resume_flow_threshold, + is_rejecting: AtomicBool::new(is_rejecting), + user_metrics: IngressMetrics::builder().with_label("handler", "user").build(), + system_metrics: IngressMetrics::builder().with_label("handler", "system").build(), + } + } + /// Return the score for the give entity. Unknown entities are not expected to be scored. /// /// # Panics @@ -144,8 +217,29 @@ impl OrderflowIngress { tracing::info!(entries = len_after, num_removed, "finished state maintenance"); } + fn update_is_rejecting_metric(is_rejecting: bool) { + CLICKHOUSE_METRICS.disk_backup_size_is_rejecting_flow().set(if is_rejecting { + 1 + } else { + 0 + }); + } + fn clickhouse_backup_disk_size_is_ok(&self) -> bool { - CLICKHOUSE_LOCAL_BACKUP_DISK_SIZE.disk_size() <= self.disk_max_size_to_accept_user_rpc + let size = CLICKHOUSE_LOCAL_BACKUP_DISK_SIZE.disk_size(); + let was_rejecting = self.is_rejecting.load(Ordering::Relaxed); + + let is_rejecting = if was_rejecting { + size > self.disk_backup_size_resume_flow_threshold + } else { + size > self.disk_backup_size_reject_flow_threshold + }; + + if was_rejecting != is_rejecting { + self.is_rejecting.store(is_rejecting, Ordering::Relaxed); + Self::update_is_rejecting_metric(is_rejecting); + } + !is_rejecting } #[tracing::instrument(skip_all, name = "ingress", diff --git a/src/lib.rs b/src/lib.rs index c011b455..a4a879a8 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -2,7 +2,6 @@ use crate::{ builderhub::{PeersUpdater, PeersUpdaterConfig}, - cache::SignerCache, forwarder::{ client::{default_http_builder, HttpClientPool}, http::spawn_http_forwarder, @@ -10,7 +9,6 @@ use crate::{ indexer::IndexerHandle, ingress::IngressSocket, metrics::IngressMetrics, - primitives::SystemBundleDecoder, priority::workers::PriorityWorkers, statics::LOCAL_PEER_STORE, }; @@ -23,17 +21,14 @@ use axum::{ Router, }; use dashmap::DashMap; -use entity::SpamThresholds; use forwarder::{IngressForwarders, PeerHandle}; use msg_socket::RepSocket; use msg_transport::tcp::Tcp; use prometric::exporter::ExporterBuilder; use rbuilder_utils::tasks::TaskExecutor; -use reqwest::Url; use std::{ net::SocketAddr, num::NonZero, - str::FromStr as _, sync::Arc, time::{Duration, Instant}, }; @@ -48,8 +43,6 @@ use cli::OrderflowIngressArgs; pub mod ingress; use ingress::OrderflowIngress; -use crate::cache::OrderCache; - pub mod builderhub; mod cache; pub mod consts; @@ -74,15 +67,19 @@ pub async fn run( ) -> eyre::Result<()> { fdlimit::raise_fd_limit()?; + eyre::ensure!( + args.disk_backup_size_resume_flow_threshold_mb <= args.disk_backup_size_reject_flow_threshold_mb, + "disk-backup-size-to-resume-flow-threshold-mb ({}) must be <= disk-backup-size-reject-flow-threshold-mb ({})", + args.disk_backup_size_resume_flow_threshold_mb, + args.disk_backup_size_reject_flow_threshold_mb, + ); + if let Some(ref metrics_addr) = args.metrics { ExporterBuilder::new().with_address(metrics_addr).with_namespace("flowproxy").install()?; metrics::spawn_process_collector().await?; // Set build info metric metrics::BUILD_INFO_METRICS.info(env!("CARGO_PKG_VERSION"), env!("GIT_HASH")).set(1); - metrics::CLICKHOUSE_METRICS - .disk_max_size_to_accept_user_rpc_bytes() - .set(args.disk_max_size_to_accept_user_rpc_mb.saturating_mul(1024 * 1024)); } let user_listener = TcpListener::bind(&args.user_listen_addr).await?; @@ -127,6 +124,8 @@ pub async fn run_with_listeners( let _ = registry.with(tracing_subscriber::fmt::layer()).try_init(); } + let config = args.ingress_config()?; + let orderflow_signer = match args.orderflow_signer { Some(signer) => { tracing::warn!( @@ -184,8 +183,7 @@ pub async fn run_with_listeners( let workers = PriorityWorkers::new_with_threads(args.compute_threads); // Spawn forwarders - let builder_url = args.builder_url.map(|url| Url::from_str(&url)).transpose()?; - let forwarders = if let Some(ref builder_url) = builder_url { + let forwarders = if let Some(ref builder_url) = config.local_builder_url { let local_sender = spawn_http_forwarder( String::from("local-builder"), builder_url.to_string(), @@ -201,34 +199,7 @@ pub async fn run_with_listeners( IngressForwarders::new(local_sender, peers, orderflow_signer, workers.clone()) }; - let builder_ready_endpoint = - args.builder_ready_endpoint.map(|url| Url::from_str(&url)).transpose()?; - - let order_cache = OrderCache::new(args.cache.order_cache_ttl, args.cache.order_cache_size); - let signer_cache = SignerCache::new(args.cache.signer_cache_ttl, args.cache.signer_cache_size); - - let ingress = Arc::new(OrderflowIngress { - gzip_enabled: args.gzip_enabled, - rate_limiting_enabled: args.enable_rate_limiting, - rate_limit_lookback_s: args.rate_limit_lookback_s, - rate_limit_count: args.rate_limit_count, - score_lookback_s: args.score_lookback_s, - score_bucket_s: args.score_bucket_s, - system_bundle_decoder: SystemBundleDecoder { max_txs_per_bundle: args.max_txs_per_bundle }, - spam_thresholds: SpamThresholds::default(), - flashbots_signer: args.flashbots_signer, - pqueues: workers, - entities: DashMap::default(), - order_cache, - signer_cache, - forwarders, - local_builder_url: builder_url, - builder_ready_endpoint, - indexer_handle, - disk_max_size_to_accept_user_rpc: args.disk_max_size_to_accept_user_rpc_mb * 1024 * 1024, - user_metrics: IngressMetrics::builder().with_label("handler", "user").build(), - system_metrics: IngressMetrics::builder().with_label("handler", "system").build(), - }); + let ingress = Arc::new(OrderflowIngress::new(config, workers, forwarders, indexer_handle)); // Spawn a state maintenance task. let cancellation_token_clone = cancellation_token.clone(); diff --git a/src/metrics.rs b/src/metrics.rs index d864a4b3..4bf9aded 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -175,23 +175,29 @@ pub struct ClickhouseMetrics { #[metric(labels = ["order", "backend"])] backup_size_batches: Gauge, /// Total number of bytes sent to Clickhouse backup. - #[metric(rename = "clickhouse_backup_data_bytes_total")] + #[metric(rename = "backup_data_bytes_total")] backup_data_bytes: Counter, /// Total number of rows sent to Clickhouse backup. - #[metric(rename = "clickhouse_backup_data_rows_total")] + #[metric(rename = "backup_data_rows_total")] backup_data_rows: Counter, /// Total number of bytes lost due to pressure on Clickhouse backup. - #[metric(rename = "clickhouse_backup_data_lost_bytes_total")] + #[metric(rename = "backup_data_lost_bytes_total")] backup_data_lost_bytes: Counter, /// Total number of rows lost due to pressure on Clickhouse backup. - #[metric(rename = "clickhouse_backup_data_lost_rows_total")] + #[metric(rename = "backup_data_lost_rows_total")] backup_data_lost_rows: Counter, /// Errors encountered during Clickhouse disk backup. #[metric(labels = ["order", "error"])] backup_disk_errors: Counter, - /// Configured max disk size (bytes) above which user RPC is rejected. - #[metric(rename = "disk_max_size_to_accept_user_rpc_bytes")] - disk_max_size_to_accept_user_rpc_bytes: Gauge, + /// Configured disk size (bytes) above which user RPC is rejected. + #[metric(rename = "disk_backup_size_reject_flow_threshold_bytes")] + disk_backup_size_reject_flow_threshold_bytes: Gauge, + /// Configured disk size (bytes) below which user RPC is accepted again after rejection. + #[metric(rename = "disk_backup_size_resume_flow_threshold_bytes")] + disk_backup_size_resume_flow_threshold_bytes: Gauge, + /// Whether user RPC is currently being rejected due to disk size (0 or 1). + #[metric(rename = "disk_backup_size_is_rejecting_flow")] + disk_backup_size_is_rejecting_flow: Gauge, } #[metrics(scope = "indexer_parquet")]