graphman command for clearing the call cache for a given chain (#4066)

docs/graphman.md

@@ -7,6 +7,7 @@
 - [Unused Remove](#unused-remove)
 - [Drop](#drop)
 - [Chain Check Blocks](#check-blocks)
+- [Chain Call Cache Remove](#chain-call-cache-remove)
 
 <a id="info"></a>
 # ⌘ Info
@@ -362,3 +363,54 @@ Inspect a block range, deleting any duplicated blocks:
 Inspect all blocks after block `13000000`:
 
     graphman --config config.toml chain check-blocks mainnet by-range --from 13000000
+
+<a id="chain-call-cache-remove"></a>
+# ⌘ Chain Call Cache Remove
+
+### SYNOPSIS
+
+Remove the call cache of the specified chain.
+
+If block numbers are not mentioned in `--from` and `--to`, then all the call cache will be removed.
+
+USAGE:
+    graphman chain call-cache <CHAIN_NAME> remove [OPTIONS]
+
+OPTIONS:
+    -f, --from <FROM>
+            Starting block number
+
+    -h, --help
+            Print help information
+
+    -t, --to <TO>
+            Ending block number
+
+### DESCRIPTION
+
+Remove the call cache of a specified chain.
+
+### OPTIONS
+
+The `from` and `to` options are used to decide the block range of the call cache that needs to be removed.
+
+#### `from`
+
+The `from` option is used to specify the starting block number of the block range. In the absence of `from` option,
+the first block number will be used as the starting block number.
+
+#### `to`
+
+The `to` option is used to specify the ending block number of the block range. In the absence of `to` option,
+the last block number will be used as the ending block number.
+
+### EXAMPLES
+
+Remove the call cache for all blocks numbered from 10 to 20:
+
+    graphman --config config.toml chain call-cache ethereum remove --from 10 --to 20
+
+Remove all the call cache of the specified chain:
+
+    graphman --config config.toml chain call-cache ethereum remove
+

graph/src/components/store/traits.rs

@@ -415,6 +415,9 @@ pub trait ChainStore: Send + Sync + 'static {
         &self,
         block_ptr: &H256,
     ) -> Result<Vec<transaction_receipt::LightTransactionReceipt>, StoreError>;
+
+    /// Clears call cache of the chain for the given `from` and `to` block number.
+    async fn clear_call_cache(&self, from: Option<i32>, to: Option<i32>) -> Result<(), Error>;
 }
 
 pub trait EthereumCallCache: Send + Sync + 'static {

node/src/bin/manager.rs

@@ -427,6 +427,31 @@ pub enum ChainCommand {
         #[clap(long, short)]
         force: bool,
     },
+
+    /// Execute operations on call cache.
+    CallCache {
+        #[clap(subcommand)]
+        method: CallCacheCommand,
+        /// Chain name (must be an existing chain, see 'chain list')
+        #[clap(empty_values = false)]
+        chain_name: String,
+    },
+}
+
+#[derive(Clone, Debug, Subcommand)]
+pub enum CallCacheCommand {
+    /// Remove the call cache of the specified chain.
+    ///
+    /// If block numbers are not mentioned in `--from` and `--to`, then all the call cache will be
+    /// removed.
+    Remove {
+        /// Starting block number
+        #[clap(long, short)]
+        from: Option<i32>,
+        /// Ending block number
+        #[clap(long, short)]
+        to: Option<i32>,
+    },
 }
 
 #[derive(Clone, Debug, Subcommand)]
@@ -1088,6 +1113,12 @@ async fn main() -> anyhow::Result<()> {
                     let chain_store = ctx.chain_store(&chain_name)?;
                     truncate(chain_store, force)
                 }
+                CallCache { method, chain_name } => match method {
+                    CallCacheCommand::Remove { from, to } => {
+                        let chain_store = ctx.chain_store(&chain_name)?;
+                        commands::chain::clear_call_cache(chain_store, from, to).await
+                    }
+                },
             }
         }
         Stats(cmd) => {

node/src/manager/commands/chain.rs

@@ -11,6 +11,7 @@ use graph::{
     components::store::BlockStore as _, prelude::anyhow::Error, prelude::serde_json as json,
 };
 use graph_store_postgres::BlockStore;
+use graph_store_postgres::ChainStore;
 use graph_store_postgres::{
     command_support::catalog::block_store, connection_pool::ConnectionPool,
 };
@@ -49,6 +50,16 @@ pub async fn list(primary: ConnectionPool, store: Arc<BlockStore>) -> Result<(),
     Ok(())
 }
 
+pub async fn clear_call_cache(
+    chain_store: Arc<ChainStore>,
+    from: Option<i32>,
+    to: Option<i32>,
+) -> Result<(), Error> {
+    chain_store.clear_call_cache(from, to).await?;
+    println!("The call cache has cleared");
+    Ok(())
+}
+
 pub async fn info(
     primary: ConnectionPool,
     store: Arc<BlockStore>,

store/postgres/src/chain_store.rs

@@ -77,6 +77,8 @@ mod data {
 
     pub(crate) const ETHEREUM_BLOCKS_TABLE_NAME: &'static str = "public.ethereum_blocks";
 
+    pub(crate) const ETHEREUM_CALL_CACHE_TABLE_NAME: &'static str = "public.eth_call_cache";
+
     mod public {
         pub(super) use super::super::public::ethereum_networks;
 
@@ -404,6 +406,15 @@ mod data {
             Ok(())
         }
 
+        fn truncate_call_cache(&self, conn: &PgConnection) -> Result<(), StoreError> {
+            let table_name = match &self {
+                Storage::Shared => ETHEREUM_CALL_CACHE_TABLE_NAME,
+                Storage::Private(Schema { call_cache, .. }) => &call_cache.qname,
+            };
+            conn.batch_execute(&format!("truncate table {} restart identity", table_name))?;
+            Ok(())
+        }
+
         /// Insert a block. If the table already contains a block with the
         /// same hash, then overwrite that block since it may be adding
         /// transaction receipts. If `overwrite` is `true`, overwrite a
@@ -1028,6 +1039,72 @@ mod data {
                 .collect())
         }
 
+        pub(super) fn clear_call_cache(
+            &self,
+            conn: &PgConnection,
+            from: Option<i32>,
+            to: Option<i32>,
+        ) -> Result<(), Error> {
+            if from.is_none() && to.is_none() {
+                // If both `from` and `to` arguments are equal to `None`, then truncation should be
+                // preferred over deletion as it is a faster operation.
+                self.truncate_call_cache(conn)?;
+                return Ok(());
+            }
+            match self {
+                Storage::Shared => {
+                    use public::eth_call_cache as cache;
+                    let mut delete_stmt = diesel::delete(cache::table).into_boxed();
+                    if let Some(from) = from {
+                        delete_stmt = delete_stmt.filter(cache::block_number.ge(from));
+                    }
+                    if let Some(to) = to {
+                        delete_stmt = delete_stmt.filter(cache::block_number.le(to))
+                    }
+                    delete_stmt.execute(conn).map_err(Error::from)?;
+                    Ok(())
+                }
+                Storage::Private(Schema { call_cache, .. }) => match (from, to) {
+                    // Because they are dynamically defined, our private call cache tables can't
+                    // implement all the required traits for deletion. This means we can't use Diesel
+                    // DSL with them and must rely on the `sql_query` function instead.
+                    (Some(from), None) => {
+                        let query =
+                            format!("delete from {} where block_number >= $1", call_cache.qname);
+                        sql_query(query)
+                            .bind::<Integer, _>(from)
+                            .execute(conn)
+                            .map_err(Error::from)?;
+                        Ok(())
+                    }
+                    (None, Some(to)) => {
+                        let query =
+                            format!("delete from {} where block_number <= $1", call_cache.qname);
+                        sql_query(query)
+                            .bind::<Integer, _>(to)
+                            .execute(conn)
+                            .map_err(Error::from)?;
+                        Ok(())
+                    }
+                    (Some(from), Some(to)) => {
+                        let query = format!(
+                            "delete from {} where block_number >= $1 and block_number <= $2",
+                            call_cache.qname
+                        );
+                        sql_query(query)
+                            .bind::<Integer, _>(from)
+                            .bind::<Integer, _>(to)
+                            .execute(conn)
+                            .map_err(Error::from)?;
+                        Ok(())
+                    }
+                    (None, None) => {
+                        unreachable!("truncation was handled at the beginning of this function");
+                    }
+                },
+            }
+        }
+
         pub(super) fn update_accessed_at(
             &self,
             conn: &PgConnection,
@@ -1715,6 +1792,11 @@ impl ChainStoreTrait for ChainStore {
             .await
     }
 
+    async fn clear_call_cache(&self, from: Option<i32>, to: Option<i32>) -> Result<(), Error> {
+        let conn = self.get_conn()?;
+        self.storage.clear_call_cache(&conn, from, to)
+    }
+
     async fn transaction_receipts_in_block(
         &self,
         block_hash: &H256,