store, node: Pass drop_schema flag into rebuild_storage

Rather than always running DROP SCHEMA IF EXISTS unconditionally,
the command layer passes the result of has_namespace as a drop_schema
bool down through BlockStore::rebuild_chain_storage into
ChainStore::rebuild_storage. When true, the existing
Storage::drop_storage is reused; when false, the drop is skipped
entirely.

Author: dimitrovmaksim

node/src/manager/commands/chain.rs

@@ -362,7 +362,8 @@ pub async fn rebuild_storage(
     let shard = &chain.shard;
     let ident = chain.network_identifier()?;
 
-    if store.has_namespace(&chain).await? {
+    let drop_schema = store.has_namespace(&chain).await?;
+    if drop_schema {
         let prompt = format!(
             "Storage {namespace} for chain {name} already exists on shard {shard}.\n\
              This will drop and rebuild chain storage. All cached blocks and call cache \
@@ -376,7 +377,7 @@ pub async fn rebuild_storage(
     }
 
     store
-        .rebuild_chain_storage(&name, &ident)
+        .rebuild_chain_storage(&name, &ident, drop_schema)
         .await
         .with_context(|| format!("Failed to rebuild storage {namespace} for chain {name}"))?;
 

store/postgres/src/block_store.rs

@@ -498,13 +498,14 @@ impl BlockStore {
         &self,
         chain: &str,
         ident: &ChainIdentifier,
+        drop_schema: bool,
     ) -> Result<(), StoreError> {
         let chain_store = self
             .store(chain)
             .await
             .ok_or_else(|| internal_error!("No chain store found for {}", chain))?;
 
-        Ok(chain_store.rebuild_storage(ident).await?)
+        Ok(chain_store.rebuild_storage(ident, drop_schema).await?)
     }
 
     // Helper to clone the list of chain stores to avoid holding the lock

store/postgres/src/chain_store.rs

@@ -3,7 +3,7 @@ use async_trait::async_trait;
 use diesel::sql_types::Text;
 use diesel::{insert_into, update, ExpressionMethods, OptionalExtension, QueryDsl};
 use diesel_async::AsyncConnection;
-use diesel_async::{scoped_futures::ScopedFutureExt, RunQueryDsl, SimpleAsyncConnection};
+use diesel_async::{scoped_futures::ScopedFutureExt, RunQueryDsl};
 
 use graph::components::store::ChainHeadStore;
 use graph::data::store::ethereum::call;
@@ -2483,7 +2483,11 @@ impl ChainStore {
     /// metadata in `ethereum_networks`, and rebuild the schema with
     /// empty tables. If the `ethereum_networks` row is missing, it is
     /// created from the provided `ident`.
-    pub(crate) async fn rebuild_storage(&self, ident: &ChainIdentifier) -> Result<(), Error> {
+    pub(crate) async fn rebuild_storage(
+        &self,
+        ident: &ChainIdentifier,
+        drop_schema: bool,
+    ) -> Result<(), Error> {
         use public::ethereum_networks as n;
 
         let nsp = self.storage.to_string();
@@ -2493,9 +2497,10 @@ impl ChainStore {
         let mut conn = self.pool.get_permitted().await?;
         conn.transaction(|conn| {
             async {
-                debug!(&self.logger, "Dropping existing schema if present"; "namespace" => &nsp);
-                conn.batch_execute(&format!("DROP SCHEMA IF EXISTS {nsp} CASCADE"))
-                    .await?;
+                if drop_schema {
+                    debug!(&self.logger, "Dropping existing schema"; "namespace" => &nsp);
+                    self.storage.drop_storage(conn, &self.chain).await?;
+                }
 
                 debug!(&self.logger, "Upserting ethereum_networks row"; "chain" => &self.chain);
                 insert_into(n::table)

store/test-store/tests/postgres/chain_head.rs

@@ -712,7 +712,7 @@ fn rebuild_storage_with_existing_namespace() {
 
         // Rebuild storage (should drop and recreate)
         block_store
-            .rebuild_chain_storage(NETWORK_NAME, &ident)
+            .rebuild_chain_storage(NETWORK_NAME, &ident, true)
             .await
             .expect("rebuild_chain_storage succeeds");
 
@@ -775,7 +775,7 @@ fn rebuild_storage_with_missing_namespace() {
 
         // Rebuild should recreate the missing namespace
         block_store
-            .rebuild_chain_storage(NETWORK_NAME, &ident)
+            .rebuild_chain_storage(NETWORK_NAME, &ident, false)
             .await
             .expect("rebuild_chain_storage succeeds on missing namespace");
 
@@ -829,7 +829,7 @@ fn rebuild_storage_repairs_ethereum_networks_row() {
         // Rebuild should recreate both the namespace and the
         // ethereum_networks row via upsert
         block_store
-            .rebuild_chain_storage(NETWORK_NAME, &ident)
+            .rebuild_chain_storage(NETWORK_NAME, &ident, false)
             .await
             .expect("rebuild_chain_storage succeeds with missing ethereum_networks row");
 
@@ -880,7 +880,7 @@ fn has_namespace_returns_false_for_missing_schema() {
         // Rebuild to leave things clean for other tests
         let ident = chain.network_identifier().unwrap();
         block_store
-            .rebuild_chain_storage(NETWORK_NAME, &ident)
+            .rebuild_chain_storage(NETWORK_NAME, &ident, false)
             .await
             .unwrap();
     });