diff --git a/.claude/CODEBASE.md b/.claude/CODEBASE.md index 84599a1d18..30085c9487 100644 --- a/.claude/CODEBASE.md +++ b/.claude/CODEBASE.md @@ -203,6 +203,7 @@ error rather than a silent custom chain (`Chain::builtin_from_str` skips the reg | `DELTA_WINDOW_DEPTH` | Blocks each extractor's RPC-side window retains (default 128) | | `DELTA_WINDOW_FOLD_BATCH` | Evictable blocks required before a fold runs (default 1) | | `ENTITY_CACHE_MODE` | `off` (default), `shadow`, or `serve`; anything but `off` loads the entity cache at startup and folds the windows into it | +| `DISABLED_RPC_CACHES` | Comma-separated RPC response caches to turn off: `contract_storage`, `protocol_state`, `protocol_components`, `traced_entry_points` (default: all on) | | `ENTITY_CACHE_SHADOW_SAMPLE_RATE` | Share of state requests that `shadow` compares with the entity cache, 0.0 to 1.0 (default 0.0) | | `TYCHO_S3_BUCKET` | S3 bucket the Substreams spkg packages are fetched from | | `OTLP_EXPORTER_ENDPOINT` | OpenTelemetry trace exporter | diff --git a/crates/tycho-indexer/src/cli.rs b/crates/tycho-indexer/src/cli.rs index 53facc4492..7cc1a2060d 100644 --- a/crates/tycho-indexer/src/cli.rs +++ b/crates/tycho-indexer/src/cli.rs @@ -1,4 +1,4 @@ -use clap::{builder::TypedValueParser as _, Args, Parser, Subcommand}; +use clap::{builder::TypedValueParser as _, Args, Parser, Subcommand, ValueEnum}; use tycho_common::{models::Chain, Bytes}; use tycho_ethereum::rpc::{ config::{RPCBatchingConfig, RPCRetryConfig}, @@ -7,7 +7,7 @@ use tycho_ethereum::rpc::{ use crate::{ extractor::ExtractionError, - services::{EntityCacheMode, WindowConfig}, + services::{EntityCacheMode, RpcCacheKind, WindowConfig}, }; /// Tycho Indexer using Substreams @@ -93,6 +93,19 @@ pub struct GlobalArgs { #[clap(long, env = "ENTITY_CACHE_SHADOW_SAMPLE_RATE", default_value_t = 0.0, value_parser = parse_sample_rate)] pub entity_cache_shadow_sample_rate: f64, + /// RPC response caches to turn off, comma-separated: `contract_storage`, `protocol_state`, + /// `protocol_components`, `traced_entry_points`. A request to an endpoint whose cache is off + /// always runs the full path. Blank entries are ignored, so an empty value keeps every + /// cache on (the default). + #[clap( + long, + env = "DISABLED_RPC_CACHES", + default_value = "", + hide_default_value = true, + value_parser = parse_disabled_rpc_caches + )] + pub disabled_rpc_caches: DisabledRpcCaches, + /// Count and page `/protocol_components` requests by protocol system from an in-memory index /// of component ids and TVLs, instead of SQL. Postgres then loads only the components on the /// requested page. On by default; `false` serves these requests entirely from SQL. @@ -125,6 +138,36 @@ pub struct GlobalArgs { pub rpc: RPCArgs, } +/// The caches named by `--disabled-rpc-caches`. +#[derive(Clone, Debug, Default, PartialEq)] +pub struct DisabledRpcCaches(pub Vec); + +/// Parses a comma-separated list of cache names. Blank entries are ignored, so an empty value +/// names no cache. +fn parse_disabled_rpc_caches(value: &str) -> Result { + value + .split(',') + .map(str::trim) + .filter(|name| !name.is_empty()) + .map(|name| { + RpcCacheKind::from_str(name, false).map_err(|_| { + let valid: Vec = RpcCacheKind::value_variants() + .iter() + .filter_map(|kind| { + Some( + kind.to_possible_value()? + .get_name() + .to_string(), + ) + }) + .collect(); + format!("unknown RPC cache '{name}', expected one of: {}", valid.join(", ")) + }) + }) + .collect::, _>>() + .map(DisabledRpcCaches) +} + /// Parses a share from 0.0 to 1.0. Rejects NaN. fn parse_sample_rate(value: &str) -> Result { let rate: f64 = value @@ -399,6 +442,7 @@ mod cli_tests { delta_window_fold_batch: 1, entity_cache_mode: EntityCacheMode::Off, entity_cache_shadow_sample_rate: 0.0, + disabled_rpc_caches: DisabledRpcCaches::default(), component_index: true, s3_bucket: Some("repo.propellerheads-propellerheads".to_string()), server_ip: "0.0.0.0".to_string(), @@ -496,6 +540,34 @@ mod cli_tests { assert_eq!(cli.global_args.entity_cache_mode, expected); } + #[rstest] + #[case::empty("", vec![])] + #[case::one("protocol_state", vec![RpcCacheKind::ProtocolState])] + #[case::two( + "contract_storage,traced_entry_points", + vec![RpcCacheKind::ContractStorage, RpcCacheKind::TracedEntryPoints] + )] + #[case::blank_entries_and_spaces(" protocol_components , ,", vec![RpcCacheKind::ProtocolComponents])] + fn test_arg_parsing_disabled_rpc_caches( + #[case] value: &'static str, + #[case] expected: Vec, + ) { + let mut args = args_with_delta_window("128", "1"); + args.splice(1..1, ["--disabled-rpc-caches", value]); + + let cli = Cli::try_parse_from(args).expect("parse errored"); + + assert_eq!(cli.global_args.disabled_rpc_caches, DisabledRpcCaches(expected)); + } + + #[test] + fn test_arg_parsing_rejects_unknown_rpc_cache() { + let mut args = args_with_delta_window("128", "1"); + args.splice(1..1, ["--disabled-rpc-caches", "tokens"]); + + assert!(Cli::try_parse_from(args).is_err()); + } + #[test] fn test_arg_parsing_component_index_is_on_by_default() { let default = @@ -590,6 +662,7 @@ mod cli_tests { delta_window_fold_batch: 1, entity_cache_mode: EntityCacheMode::Off, entity_cache_shadow_sample_rate: 0.0, + disabled_rpc_caches: DisabledRpcCaches::default(), component_index: true, s3_bucket: Some("repo.propellerheads-propellerheads".to_string()), server_ip: "0.0.0.0".to_string(), diff --git a/crates/tycho-indexer/src/main.rs b/crates/tycho-indexer/src/main.rs index 8a9e914781..a4a051110d 100644 --- a/crates/tycho-indexer/src/main.rs +++ b/crates/tycho-indexer/src/main.rs @@ -458,6 +458,12 @@ async fn run_rpc(global_args: GlobalArgs, rpc_args: RpcServerArgs) -> Result<(), .bind(&global_args.server_ip) .port(global_args.server_port) .plans_config(plans_config) + .disabled_rpc_caches( + global_args + .disabled_rpc_caches + .0 + .clone(), + ) .run()?; info!(server_url, "Http and Ws server started"); let shutdown_task = tokio::spawn(shutdown_handler(server_handle, vec![], None)); @@ -558,6 +564,12 @@ async fn create_indexing_tasks( }) .entity_cache(entity_cache) .shadow_sample_rate(global_args.entity_cache_shadow_sample_rate) + .disabled_rpc_caches( + global_args + .disabled_rpc_caches + .0 + .clone(), + ) .run()?; info!(server_url, "Http and Ws server started"); diff --git a/crates/tycho-indexer/src/services/cache.rs b/crates/tycho-indexer/src/services/cache.rs index 7f21f63e35..9e4a5b0d45 100644 --- a/crates/tycho-indexer/src/services/cache.rs +++ b/crates/tycho-indexer/src/services/cache.rs @@ -47,9 +47,26 @@ impl ValueWithSize { } } +/// The response caches of the RPC handler, one per endpoint. Set the ones to turn off per +/// deployment with `DISABLED_RPC_CACHES`. +#[derive(Clone, Copy, Debug, PartialEq, Eq, clap::ValueEnum)] +#[clap(rename_all = "snake_case")] +pub enum RpcCacheKind { + /// `/contract_state` responses. + ContractStorage, + /// `/protocol_state` responses. + ProtocolState, + /// `/protocol_components` responses. + ProtocolComponents, + /// `/traced_entry_points` responses. + TracedEntryPoints, +} + pub struct RpcCache { name: String, cache: Cache>>, + /// When `false`, every `get` runs the fallback and stores nothing. + enabled: bool, } impl RpcCache @@ -71,7 +88,12 @@ where .weigher(weigher) .build(); - Self { name: name.to_string(), cache } + Self { name: name.to_string(), cache, enabled: true } + } + + /// Turns the cache off: every `get` runs the fallback and stores nothing. + pub fn disable(&mut self) { + self.enabled = false; } #[instrument( @@ -90,6 +112,11 @@ where request: R, fallback: F, ) -> Result, E> { + if !self.enabled { + trace!("CacheDisabled"); + let (response, _) = (fallback)(request).await?; + return Ok(Arc::new(response)); + } tracing::Span::current().record("size", self.cache.entry_count()); // Check the cache for a cached response if let Some(inflight_val) = self.cache.get(&request) { @@ -191,6 +218,25 @@ mod test { Ok((1, true)) } + #[test_log::test(tokio::test)] + async fn test_disabled_cache_runs_fallback_every_time() { + let access_counter = Arc::new(Mutex::new(0)); + let mut cache = RpcCache::::new("test", 100, 3600); + cache.disable(); + + for _ in 0..2 { + cache + .get("k0".to_string(), |_| async { + increment_counter(access_counter.clone()).await + }) + .await + .unwrap(); + } + + assert_eq!(*access_counter.lock().await, 2); + assert_eq!(cache.cache.entry_count(), 0); + } + #[test_log::test(tokio::test)] async fn test_parallel_access() { let access_counter = Arc::new(Mutex::new(0)); diff --git a/crates/tycho-indexer/src/services/mod.rs b/crates/tycho-indexer/src/services/mod.rs index e7e5bbd09f..7939824380 100644 --- a/crates/tycho-indexer/src/services/mod.rs +++ b/crates/tycho-indexer/src/services/mod.rs @@ -47,6 +47,7 @@ mod rpc; mod state; mod ws; +pub use cache::RpcCacheKind; pub use middleware::PlansConfig; pub use state::{ cache::EntityCache, service::EntityCacheSetup, window::WindowConfig, EntityCacheMode, @@ -73,6 +74,8 @@ pub struct ServicesBuilder { entity_cache: EntityCacheSetup>, /// Share of state requests that shadow mode compares. shadow_sample_rate: f64, + /// RPC response caches that stay off. + disabled_rpc_caches: Vec, } /// Resolves with the first error either service task produces, or with `Ok` once both end @@ -110,9 +113,17 @@ where window_config: WindowConfig::default(), entity_cache: EntityCacheSetup::Off, shadow_sample_rate: 0.0, + disabled_rpc_caches: Vec::new(), } } + /// Turns the given RPC response caches off. Requests to those endpoints always run the + /// full path. + pub fn disabled_rpc_caches(mut self, kinds: Vec) -> Self { + self.disabled_rpc_caches = kinds; + self + } + /// Sets the retention depth and fold batch of every extractor's `DeltaWindow`. pub fn window_config(mut self, v: WindowConfig) -> Self { self.window_config = v; @@ -292,6 +303,9 @@ where ) -> Result<(ServerHandle, JoinHandle>), ExtractionError> { let tracer = EVMEntrypointService::new(&self.rpc); + if !self.disabled_rpc_caches.is_empty() { + info!(disabled = ?self.disabled_rpc_caches, "RPC response caches turned off"); + } let rpc_data = web::Data::new( rpc::RpcHandler::new( self.db_gateway, @@ -301,6 +315,7 @@ where self.dci_protocols, self.protocol_systems, ) + .with_disabled_caches(&self.disabled_rpc_caches) .with_state_service(state_service), ); diff --git a/crates/tycho-indexer/src/services/rpc.rs b/crates/tycho-indexer/src/services/rpc.rs index a019783439..cb0d4921ed 100644 --- a/crates/tycho-indexer/src/services/rpc.rs +++ b/crates/tycho-indexer/src/services/rpc.rs @@ -28,7 +28,7 @@ use tycho_common::{ use crate::{ extractor::reorg_buffer::{BlockNumberOrTimestamp, CommitStatus}, services::{ - cache::RpcCache, + cache::{RpcCache, RpcCacheKind}, deltas_buffer::{PendingDeltasBuffer, PendingDeltasError}, middleware::{ PlanRestrictions, PlansConfig, RequestPaginationValidation, ValidateRestrictions, @@ -185,6 +185,20 @@ where } } + /// Turns the given response caches off. A request to an endpoint whose cache is off always + /// runs the full path and stores nothing. + pub(crate) fn with_disabled_caches(mut self, kinds: &[RpcCacheKind]) -> Self { + for kind in kinds { + match kind { + RpcCacheKind::ContractStorage => self.contract_storage_cache.disable(), + RpcCacheKind::ProtocolState => self.protocol_state_cache.disable(), + RpcCacheKind::ProtocolComponents => self.component_cache.disable(), + RpcCacheKind::TracedEntryPoints => self.traced_entry_point_cache.disable(), + } + } + self + } + /// Sets which path answers the state endpoints. pub(crate) fn with_state_service( mut self, @@ -3339,6 +3353,44 @@ mod tests { assert_eq!(tokens.tokens[1].symbol, "WETH"); } + #[tokio::test] + async fn test_disabled_cache_sends_every_request_to_the_gateway() { + let mut gw = MockGateway::new(); + // The component cache is off: both identical requests reach the gateway. + gw.expect_get_protocol_components() + .times(2) + .returning(|_, _, _, _, _| { + Box::pin(async { Ok(WithTotal { entity: vec![], total: Some(0) }) }) + }); + let req_handler = RpcHandler::new( + gw, + None, + MockEntryPointTracer::new(), + PlansConfig::default(), + vec![], + vec![], + ) + .with_disabled_caches(&[RpcCacheKind::ProtocolComponents]); + + let request = dto::ProtocolComponentsRequestBody { + protocol_system: "uniswap_v2".to_string(), + component_ids: None, + tvl_gt: None, + chain: dto::Chain::Ethereum, + pagination: dto::PaginationParams { page: 0, page_size: 10 }, + }; + + for _ in 0..2 { + let components = req_handler + .get_protocol_components(&request) + .await + .unwrap(); + assert!(components + .protocol_components + .is_empty()); + } + } + #[tokio::test] async fn test_get_protocol_state() { let mut gw = MockGateway::new();