feat(indexer): make RPC response caches disableable

.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 |

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<RpcCacheKind>);
+
+/// 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<DisabledRpcCaches, String> {
+    value
+        .split(',')
+        .map(str::trim)
+        .filter(|name| !name.is_empty())
+        .map(|name| {
+            RpcCacheKind::from_str(name, false).map_err(|_| {
+                let valid: Vec<String> = 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::<Result<Vec<_>, _>>()
+        .map(DisabledRpcCaches)
+}
+
 /// Parses a share from 0.0 to 1.0. Rejects NaN.
 fn parse_sample_rate(value: &str) -> Result<f64, String> {
     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<RpcCacheKind>,
+    ) {
+        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(),

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

crates/tycho-indexer/src/services/cache.rs

@@ -47,9 +47,26 @@ impl<V: DeepSizeOf + Send + Sync> ValueWithSize<V> {
     }
 }
 
+/// 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<R, V> {
     name: String,
     cache: Cache<R, Arc<ValueWithSize<V>>>,
+    /// When `false`, every `get` runs the fallback and stores nothing.
+    enabled: bool,
 }
 
 impl<R, V> RpcCache<R, V>
@@ -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<Arc<V>, 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::<String, i32>::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));

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<G> {
     entity_cache: EntityCacheSetup<Arc<EntityCache>>,
     /// Share of state requests that shadow mode compares.
     shadow_sample_rate: f64,
+    /// RPC response caches that stay off.
+    disabled_rpc_caches: Vec<RpcCacheKind>,
 }
 
 /// 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<RpcCacheKind>) -> 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<Result<(), ExtractionError>>), 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),
         );
 

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