graphman: Add simpler chain rewind command to delete blocks

node/src/bin/manager.rs

@@ -583,6 +583,21 @@ pub enum ChainCommand {
         #[clap(value_parser = clap::builder::NonEmptyStringValueParser::new())]
         chain_name: String,
     },
+    /// Rewind all blocks on a specific chain
+    ///
+    /// Note that this doesn't rewind the deployments, only deletes blocks past the target and
+    /// updates the chain head.
+    Rewind {
+        /// Chain name (must be an existing chain, see 'chain list')
+        #[clap(required = true, value_parser = clap::builder::NonEmptyStringValueParser::new())]
+        chain_name: String,
+        /// The block hash of the target block
+        #[clap(long, short = 'H')]
+        block_hash: String,
+        /// The block number of the target block
+        #[clap(long, short = 'n')]
+        block_number: i32,
+    },
 }
 
 #[derive(Clone, Debug, Subcommand)]
@@ -1442,6 +1457,17 @@ async fn main() -> anyhow::Result<()> {
                         }
                     }
                 }
+                Rewind {
+                    chain_name,
+                    block_hash,
+                    block_number,
+                } => {
+                    let (block_store, _) = ctx.block_store_and_primary_pool();
+                    let block_hash =
+                        BlockHash::from_str(&block_hash).context("invalid block hash")?;
+
+                    commands::chain::rewind(block_store, chain_name, block_hash, block_number)
+                }
             }
         }
         Stats(cmd) => {

node/src/manager/commands/chain.rs

@@ -261,3 +261,24 @@ pub fn change_block_cache_shard(
 
     Ok(())
 }
+
+pub fn rewind(
+    store: Arc<BlockStore>,
+    chain_name: String,
+    block_hash: BlockHash,
+    block_number: i32,
+) -> Result<(), Error> {
+    let block_ptr_to = BlockPtr::new(block_hash.clone(), block_number);
+    let chain_store = store
+        .chain_store(&chain_name)
+        .ok_or_else(|| anyhow!("Chain store not found for {}", chain_name))?;
+
+    chain_store.rewind_chain(block_ptr_to)?;
+
+    println!(
+        "Successfully rewound {} to block {} ({})",
+        chain_name, block_number, block_hash
+    );
+
+    Ok(())
+}

store/postgres/src/chain_store.rs

@@ -1123,6 +1123,32 @@ mod data {
             }
         }
 
+        pub(super) fn delete_blocks_after(
+            &self,
+            conn: &mut PgConnection,
+            chain: &str,
+            block: i64,
+        ) -> Result<usize, Error> {
+            match self {
+                Storage::Shared => {
+                    use public::ethereum_blocks as b;
+
+                    diesel::delete(b::table)
+                        .filter(b::network_name.eq(chain))
+                        .filter(b::number.gt(block))
+                        .execute(conn)
+                        .map_err(Error::from)
+                }
+                Storage::Private(Schema { blocks, .. }) => {
+                    let query = format!("delete from {} where number > $1", blocks.qname);
+                    sql_query(query)
+                        .bind::<BigInt, _>(block)
+                        .execute(conn)
+                        .map_err(Error::from)
+                }
+            }
+        }
+
         pub(super) fn delete_blocks_by_hash(
             &self,
             conn: &mut PgConnection,
@@ -1870,6 +1896,29 @@ impl ChainStore {
             .await?;
         Ok(values)
     }
+
+    pub fn rewind_chain(&self, block_ptr_to: BlockPtr) -> Result<(), StoreError> {
+        let mut conn = self.pool.get()?;
+
+        conn.transaction(|conn| {
+            use public::ethereum_networks;
+
+            self.storage
+                .delete_blocks_after(conn, &self.chain, block_ptr_to.number as i64)?;
+
+            // Update the chain head
+            diesel::update(
+                ethereum_networks::table.filter(ethereum_networks::name.eq(&self.chain)),
+            )
+            .set((
+                ethereum_networks::head_block_number.eq(block_ptr_to.number as i64),
+                ethereum_networks::head_block_hash.eq(block_ptr_to.hash_hex()),
+            ))
+            .execute(conn)?;
+
+            Ok(())
+        })
+    }
 }
 
 #[async_trait]