Reject flow on disk backup size hysteresis

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<ingress::Config> {
+        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;

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();

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<Url>,
     pub builder_ready_endpoint: Option<Url>,
     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<Address>,
+    pub(crate) local_builder_url: Option<Url>,
+    pub(crate) builder_ready_endpoint: Option<Url>,
+    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",

src/lib.rs

@@ -2,15 +2,13 @@
 
 use crate::{
     builderhub::{PeersUpdater, PeersUpdaterConfig},
-    cache::SignerCache,
     forwarder::{
         client::{default_http_builder, HttpClientPool},
         http::spawn_http_forwarder,
     },
     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();