node, store: Add 'graphman chain ingest' command

Author: lutter

node/src/bin/manager.rs

@@ -8,7 +8,7 @@ use graph::components::network_provider::ChainName;
 use graph::endpoint::EndpointMetrics;
 use graph::env::ENV_VARS;
 use graph::log::logger_with_levels;
-use graph::prelude::{MetricsRegistry, BLOCK_NUMBER_MAX};
+use graph::prelude::{BlockNumber, MetricsRegistry, BLOCK_NUMBER_MAX};
 use graph::{data::graphql::load_manager::LoadManager, prelude::chrono, prometheus::Registry};
 use graph::{
     prelude::{
@@ -587,6 +587,19 @@ pub enum ChainCommand {
         #[clap(value_parser = clap::builder::NonEmptyStringValueParser::new())]
         chain_name: String,
     },
+
+    /// Ingest a block into the block cache.
+    ///
+    /// This will overwrite any blocks we may already have in the block
+    /// cache, and can therefore be used to get rid of duplicate blocks in
+    /// the block cache as well as making sure that a certain block is in
+    /// the cache
+    Ingest {
+        /// The name of the chain
+        name: String,
+        /// The block number to ingest
+        number: BlockNumber,
+    },
 }
 
 #[derive(Clone, Debug, Subcommand)]
@@ -1452,6 +1465,12 @@ async fn main() -> anyhow::Result<()> {
                         }
                     }
                 }
+                Ingest { name, number } => {
+                    let logger = ctx.logger.cheap_clone();
+                    let (chain_store, ethereum_adapter) =
+                        ctx.chain_store_and_adapter(&name).await?;
+                    commands::chain::ingest(&logger, chain_store, ethereum_adapter, number).await
+                }
             }
         }
         Stats(cmd) => {

node/src/manager/commands/chain.rs

@@ -10,11 +10,16 @@ use graph::cheap_clone::CheapClone;
 use graph::components::network_provider::ChainIdentifierStore;
 use graph::components::network_provider::ChainName;
 use graph::components::store::StoreError;
+use graph::futures03::compat::Future01CompatExt as _;
 use graph::prelude::BlockNumber;
 use graph::prelude::ChainStore as _;
+use graph::prelude::LightEthereumBlockExt;
 use graph::prelude::{anyhow, anyhow::bail};
 use graph::slog::Logger;
 use graph::{components::store::BlockStore as _, prelude::anyhow::Error};
+use graph_chain_ethereum::chain::BlockFinality;
+use graph_chain_ethereum::EthereumAdapter;
+use graph_chain_ethereum::EthereumAdapterTrait as _;
 use graph_store_postgres::add_chain;
 use graph_store_postgres::connection_pool::PoolCoordinator;
 use graph_store_postgres::find_chain;
@@ -261,3 +266,31 @@ pub fn change_block_cache_shard(
 
     Ok(())
 }
+
+pub async fn ingest(
+    logger: &Logger,
+    chain_store: Arc<ChainStore>,
+    ethereum_adapter: Arc<EthereumAdapter>,
+    number: BlockNumber,
+) -> Result<(), Error> {
+    let Some(block) = ethereum_adapter
+        .block_by_number(logger, number)
+        .compat()
+        .await
+        .map_err(|e| anyhow!("error getting block number {number}: {}", e))?
+    else {
+        bail!("block number {number} not found");
+    };
+    let ptr = block.block_ptr();
+    // For inserting the block, it doesn't matter whether the block is final or not.
+    let block = Arc::new(BlockFinality::Final(Arc::new(block)));
+    chain_store.upsert_block(block).await?;
+
+    let rows = chain_store.confirm_block_hash(ptr.number, &ptr.hash)?;
+
+    println!("Inserted block {}", ptr);
+    if rows > 0 {
+        println!("    (also deleted {rows} duplicate row(s) with that number)");
+    }
+    Ok(())
+}