delete 'shallow' fh eth blks on rpc+ingestor start (#4790)

* delete 'shallow' fh eth blks on rpc+ingestor start

when launching grpah-node with an ingestor on an eth chain with an RPC
provider (i.e. not firehose), this will delete the 'shallow' blocks in
chainX.blocks, so they do not cause issues to the RPC provider.

It only runs on ethereum chains where firehose is not used and when
ingestor is active.

* cleanup shallowblocks only within reorg threshold, remove nulldata indexes (too costly)

* also remove cursor on non-firehose ethereum chains

* delete_shallowblocks: check for overflow, use double reorg_threshold for safety

* add comment with github pr link regarding cleanup_shallow_blocks

* Only remove shallow blocks if cursor was removed

* minor style improvements

* more minor style refinement in block_store

---------

Co-authored-by: Stéphane Duchesneau <[email protected]>

Author: sduchesneau

node/src/main.rs

@@ -379,6 +379,18 @@ async fn main() {
             Some(&eth_firehose_only_networks)
         };
 
+        if !opt.disable_block_ingestor && eth_networks.networks.len() != 0 {
+            let eth_network_names = Vec::from_iter(eth_networks.networks.keys());
+            let fh_only = match eth_firehose_only_networks {
+                Some(firehose_only) => Some(Vec::from_iter(firehose_only.networks.keys())),
+                None => None,
+            };
+            network_store
+                .block_store()
+                .cleanup_ethereum_shallow_blocks(eth_network_names, fh_only)
+                .unwrap();
+        }
+
         let ethereum_chains = ethereum_networks_as_chains(
             &mut blockchain_map,
             &logger,

store/postgres/migrations/2023-06-22_050000_index_ethereum_nulldata/down.sql

@@ -7,7 +7,7 @@ use std::{
 use graph::{
     blockchain::ChainIdentifier,
     components::store::BlockStore as BlockStoreTrait,
-    prelude::{error, warn, BlockNumber, BlockPtr, Logger, ENV_VARS},
+    prelude::{error, info, warn, BlockNumber, BlockPtr, Logger, ENV_VARS},
 };
 use graph::{constraint_violation, prelude::CheapClone};
 use graph::{
@@ -442,6 +442,44 @@ impl BlockStore {
         Ok(())
     }
 
+    // cleanup_ethereum_shallow_blocks will delete cached blocks previously produced by firehose on
+    // an ethereum chain that is not currently configured to use firehose provider.
+    //
+    // This is to prevent an issue where firehose stores "shallow" blocks (with null data) in `chainX.blocks`
+    // table but RPC provider requires those blocks to be full.
+    //
+    // - This issue only affects ethereum chains.
+    // - This issue only happens when switching providers from firehose back to RPC. it is gated by
+    // the presence of a cursor in the public.ethereum_networks table for a chain configured without firehose.
+    // - Only the shallow blocks close to HEAD need to be deleted, the older blocks don't need data.
+    // - Deleting everything or creating an index on empty data would cause too much performance
+    // hit on graph-node startup.
+    //
+    // Discussed here: https://github.com/graphprotocol/graph-node/pull/4790
+    pub fn cleanup_ethereum_shallow_blocks(
+        &self,
+        ethereum_networks: Vec<&String>,
+        firehose_only_networks: Option<Vec<&String>>,
+    ) -> Result<(), StoreError> {
+        for store in self.stores.read().unwrap().values() {
+            if !ethereum_networks.contains(&&store.chain) {
+                continue;
+            };
+            if let Some(fh_nets) = firehose_only_networks.clone() {
+                if fh_nets.contains(&&store.chain) {
+                    continue;
+                };
+            }
+
+            if let Some(head_block) = store.remove_cursor(&&store.chain)? {
+                let lower_bound = head_block.saturating_sub(ENV_VARS.reorg_threshold * 2);
+                info!(&self.logger, "Removed cursor for non-firehose chain, now cleaning shallow blocks"; "network" => &store.chain, "lower_bound" => lower_bound);
+                store.cleanup_shallow_blocks(lower_bound)?;
+            }
+        }
+        Ok(())
+    }
+
     fn truncate_block_caches(&self) -> Result<(), StoreError> {
         for store in self.stores.read().unwrap().values() {
             store.truncate_block_cache()?

store/postgres/migrations/2023-06-22_050000_index_ethereum_nulldata/up.sql

@@ -418,6 +418,52 @@ mod data {
             Ok(())
         }
 
+        pub(super) fn cleanup_shallow_blocks(
+            &self,
+            conn: &PgConnection,
+            lowest_block: i32,
+        ) -> Result<(), StoreError> {
+            let table_name = match &self {
+                Storage::Shared => ETHEREUM_BLOCKS_TABLE_NAME,
+                Storage::Private(Schema { blocks, .. }) => &blocks.qname,
+            };
+            conn.batch_execute(&format!(
+                "delete from {} WHERE number >= {} AND data->'block'->'data' = 'null'::jsonb;",
+                table_name, lowest_block,
+            ))?;
+            Ok(())
+        }
+
+        pub(super) fn remove_cursor(
+            &self,
+            conn: &PgConnection,
+            chain: &str,
+        ) -> Result<Option<BlockNumber>, StoreError> {
+            use diesel::dsl::not;
+            use public::ethereum_networks::dsl::*;
+
+            match update(
+                ethereum_networks
+                    .filter(name.eq(chain))
+                    .filter(not(head_block_cursor.is_null())),
+            )
+            .set(head_block_cursor.eq(None as Option<String>))
+            .returning(head_block_number)
+            .get_result::<Option<i64>>(conn)
+            .optional()
+            {
+                Ok(res) => match res {
+                    Some(opt_num) => match opt_num {
+                        Some(num) => Ok(Some(num as i32)),
+                        None => Ok(None),
+                    },
+                    None => Ok(None),
+                },
+                Err(e) => Err(e),
+            }
+            .map_err(Into::into)
+        }
+
         /// 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
@@ -1553,6 +1599,18 @@ impl ChainStore {
             .delete_blocks_by_hash(&conn, &self.chain, block_hashes)
     }
 
+    pub fn cleanup_shallow_blocks(&self, lowest_block: i32) -> Result<(), StoreError> {
+        let conn = self.get_conn()?;
+        self.storage.cleanup_shallow_blocks(&conn, lowest_block)?;
+        Ok(())
+    }
+
+    // remove_cursor delete the chain_store cursor and return true if it was present
+    pub fn remove_cursor(&self, chain: &str) -> Result<Option<BlockNumber>, StoreError> {
+        let conn = self.get_conn()?;
+        self.storage.remove_cursor(&conn, chain)
+    }
+
     pub fn truncate_block_cache(&self) -> Result<(), StoreError> {
         let conn = self.get_conn()?;
         self.storage.truncate_block_cache(&conn)?;