node: Fix duplicate key constraint when reverting chain shard changes

- Implement robust backup naming system to avoid conflicts
- Add support for reusing existing backups when appropriate
- Preserve previous backups with unique names
- Add comprehensive logging for backup operations

Fixes #6196

Author: erayack

node/src/manager/commands/chain.rs

@@ -22,6 +22,7 @@ use graph::{
 use graph_chain_ethereum::EthereumAdapter;
 use graph_chain_ethereum::EthereumAdapterTrait as _;
 use graph_chain_ethereum::chain::BlockFinality;
+use graph_store_postgres::AsyncPgConnection;
 use graph_store_postgres::BlockStore;
 use graph_store_postgres::ChainStore;
 use graph_store_postgres::PoolCoordinator;
@@ -236,6 +237,39 @@ pub async fn update_chain_genesis(
     Ok(())
 }
 
+struct ChainSwapOutcome {
+    latest_backup_name: String,
+    previous_backup_final_name: Option<String>,
+    reused_previous_backup: bool,
+    allocated_chain: bool,
+}
+
+fn backup_name_for_suffix(chain: &str, base: &str, suffix: usize) -> String {
+    if suffix == 0 {
+        format!("{chain}-{base}")
+    } else {
+        format!("{chain}-{base}-{suffix}")
+    }
+}
+
+async fn next_backup_name(
+    conn: &mut AsyncPgConnection,
+    chain: &str,
+    base: &str,
+) -> Result<String, StoreError> {
+    let mut suffix = 0usize;
+
+    loop {
+        let candidate = backup_name_for_suffix(chain, base, suffix);
+
+        if find_chain(conn, &candidate).await?.is_none() {
+            return Ok(candidate);
+        }
+
+        suffix += 1;
+    }
+}
+
 pub async fn change_block_cache_shard(
     primary_store: ConnectionPool,
     store: BlockStore,
@@ -250,51 +284,117 @@ pub async fn change_block_cache_shard(
         .await?
         .ok_or_else(|| anyhow!("unknown chain: {}", chain_name))?;
     let old_shard = chain.shard;
+    let canonical_backup_name = format!("{chain_name}-old");
+
+    let existing_backup = find_chain(&mut conn, &canonical_backup_name).await?;
+    let had_existing_backup = existing_backup.is_some();
+    let existing_backup_store = store.chain_store(&canonical_backup_name).await;
 
     println!("Current shard: {}", old_shard);
 
     let chain_store = store
         .chain_store(&chain_name)
         .await
         .ok_or_else(|| anyhow!("unknown chain: {}", &chain_name))?;
-    let new_name = format!("{}-old", &chain_name);
     let ident = chain_store.chain_identifier().await?;
+    let target_shard = Shard::new(shard.clone())?;
+    let reuse_existing_backup = existing_backup
+        .as_ref()
+        .map(|backup| backup.shard.as_str() == target_shard.as_str())
+        .unwrap_or(false);
+
+    let allocated_chain = if reuse_existing_backup {
+        None
+    } else {
+        let chain =
+            BlockStore::allocate_chain(&mut conn, &chain_name, &target_shard, &ident).await?;
+        store.add_chain_store(&chain, true).await?;
+        Some(chain)
+    };
 
-    conn.transaction::<(), StoreError, _>(|conn|  {
+    let outcome = conn.transaction::<ChainSwapOutcome, StoreError, _>(|conn| {
         async {
-            let shard = Shard::new(shard.to_string())?;
-
-            let chain = BlockStore::allocate_chain(conn, &chain_name, &shard, &ident).await?;
-
-            store.add_chain_store(&chain, true).await?;
-
-            // Drop the foreign key constraint on deployment_schemas
             sql_query(
                 "alter table deployment_schemas drop constraint deployment_schemas_network_fkey;",
             )
             .execute(conn).await?;
 
-            // Update the current chain name to chain-old
-            update_chain_name(conn, &chain_name, &new_name).await?;
+            let mut previous_backup_final_name: Option<String> = None;
+
+            let temp_backup_name = if let Some(backup) = existing_backup.as_ref() {
+                let temp_name = next_backup_name(conn, &chain_name, "old").await?;
+                update_chain_name(conn, &backup.name, &temp_name).await?;
+                previous_backup_final_name = Some(temp_name.clone());
+                Some(temp_name)
+            } else {
+                None
+            };
+
+            let latest_backup_name = next_backup_name(conn, &chain_name, "old").await?;
+            update_chain_name(conn, &chain_name, &latest_backup_name).await?;
 
-            // Create a new chain with the name in the destination shard
-            let _ = add_chain(conn, &chain_name, &shard, ident).await?;
+            if reuse_existing_backup {
+                if let Some(temp_name) = temp_backup_name.as_ref() {
+                    update_chain_name(conn, temp_name, &chain_name).await?;
+                    previous_backup_final_name = Some(chain_name.clone());
+                }
+            } else {
+                add_chain(conn, &chain_name, &target_shard, ident.clone()).await?;
+            }
 
-            // Re-add the foreign key constraint
             sql_query(
                 "alter table deployment_schemas add constraint deployment_schemas_network_fkey foreign key (network) references chains(name);",
             )
             .execute(conn).await?;
-            Ok(())
+
+            Ok(ChainSwapOutcome {
+                latest_backup_name,
+                previous_backup_final_name,
+                reused_previous_backup: reuse_existing_backup,
+                allocated_chain: allocated_chain.is_some(),
+            })
         }.scope_boxed()
     }).await?;
 
-    chain_store.update_name(&new_name).await?;
+    let ChainSwapOutcome {
+        latest_backup_name,
+        previous_backup_final_name,
+        reused_previous_backup,
+        allocated_chain,
+    } = outcome;
+
+    chain_store.update_name(&latest_backup_name).await?;
+
+    if let (Some(backup_store), Some(final_name)) = (
+        existing_backup_store.as_ref(),
+        previous_backup_final_name.as_ref(),
+    ) {
+        backup_store.update_name(final_name).await?;
+    }
 
     println!(
         "Changed block cache shard for {} from {} to {}",
         chain_name, old_shard, shard
     );
+    println!("Latest backup recorded as `{}`", latest_backup_name);
+
+    if reused_previous_backup {
+        println!(
+            "Reused existing backup `{}` as the active `{}` chain",
+            canonical_backup_name, chain_name
+        );
+    } else if let Some(ref preserved) = previous_backup_final_name {
+        if had_existing_backup {
+            println!("Preserved earlier backup as `{}`", preserved);
+        }
+    }
+
+    if allocated_chain {
+        println!(
+            "Allocated new chain state for `{}` on shard {}",
+            chain_name, shard
+        );
+    }
 
     Ok(())
 }
@@ -329,3 +429,22 @@ pub async fn ingest(
     }
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::backup_name_for_suffix;
+
+    #[test]
+    fn backup_name_for_suffix_uses_plain_name_first() {
+        assert_eq!(backup_name_for_suffix("mainnet", "old", 0), "mainnet-old");
+    }
+
+    #[test]
+    fn backup_name_for_suffix_adds_numeric_suffixes() {
+        assert_eq!(backup_name_for_suffix("mainnet", "old", 1), "mainnet-old-1");
+        assert_eq!(
+            backup_name_for_suffix("mainnet", "old", 42),
+            "mainnet-old-42"
+        );
+    }
+}