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

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/src/block_store.rs

@@ -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,40 @@ impl BlockStore {
         Ok(())
     }
 
+    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;
+                };
+            }
+            match store.chain_head_block(&&store.chain).unwrap_or(None) {
+                Some(head) => {
+                    let lower_bound = head - ENV_VARS.reorg_threshold;
+
+                    info!(&self.logger, "Cleaning shallow blocks and cursor on non-firehose chain"; "network" => &store.chain, "lower_bound" => lower_bound);
+                    let ret = store.remove_cursor();
+                    if ret.is_err() {
+                        return ret;
+                    }
+                    store.cleanup_shallow_blocks(lower_bound)?
+                }
+                None => {
+                    info!(&self.logger, "Cleaning any cursor on non-firehose chain"; "network" => &store.chain);
+                    store.remove_cursor()?
+                }
+            }
+        }
+        Ok(())
+    }
+
     fn truncate_block_caches(&self) -> Result<(), StoreError> {
         for store in self.stores.read().unwrap().values() {
             store.truncate_block_cache()?

store/postgres/src/chain_store.rs

@@ -418,6 +418,31 @@ 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) -> Result<(), StoreError> {
+            use public::ethereum_networks::dsl::*;
+
+            update(ethereum_networks)
+                .set(head_block_cursor.eq(None as Option<String>))
+                .execute(conn)?;
+            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
@@ -1553,6 +1578,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(())
+    }
+
+    pub fn remove_cursor(&self) -> Result<(), StoreError> {
+        let conn = self.get_conn()?;
+        self.storage.remove_cursor(&conn)?;
+        Ok(())
+    }
+
     pub fn truncate_block_cache(&self) -> Result<(), StoreError> {
         let conn = self.get_conn()?;
         self.storage.truncate_block_cache(&conn)?;