node: Prompt before reusing change-shard backup

Author: erayack

node/src/manager/commands/chain.rs

@@ -33,6 +33,7 @@ use graph_store_postgres::find_chain;
 use graph_store_postgres::update_chain_name;
 use graph_store_postgres::{ConnectionPool, command_support::catalog::block_store};
 
+use crate::manager::prompt::prompt_for_confirmation;
 use crate::network_setup::Networks;
 
 pub async fn list(primary: ConnectionPool, store: BlockStore) -> Result<(), Error> {
@@ -238,12 +239,17 @@ pub async fn update_chain_genesis(
 }
 
 struct ChainSwapOutcome {
-    latest_backup_name: String,
-    previous_backup_final_name: Option<String>,
     reused_previous_backup: bool,
     allocated_chain: bool,
 }
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum ExistingBackupDisposition<'a> {
+    ProceedFresh,
+    PromptReuse,
+    AbortWrongShard(&'a str),
+}
+
 fn backup_name(chain: &str, base: &str) -> String {
     format!("{chain}-{base}")
 }
@@ -252,7 +258,18 @@ fn suffixed_backup_name(backup_name: &str, suffix: usize) -> String {
     format!("{backup_name}-{suffix}")
 }
 
-async fn next_backup_name(
+fn existing_backup_disposition<'a>(
+    existing_backup_shard: Option<&'a str>,
+    target_shard: &str,
+) -> ExistingBackupDisposition<'a> {
+    match existing_backup_shard {
+        None => ExistingBackupDisposition::ProceedFresh,
+        Some(shard) if shard == target_shard => ExistingBackupDisposition::PromptReuse,
+        Some(shard) => ExistingBackupDisposition::AbortWrongShard(shard),
+    }
+}
+
+async fn next_temporary_backup_name(
     conn: &mut AsyncPgConnection,
     chain: &str,
     base: &str,
@@ -289,9 +306,7 @@ pub async fn change_block_cache_shard(
         .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 existing_backup_store = store.chain_store(&canonical_backup_name).await;
 
     println!("Current shard: {}", old_shard);
 
@@ -301,10 +316,47 @@ pub async fn change_block_cache_shard(
         .ok_or_else(|| anyhow!("unknown chain: {}", &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 existing_backup_disposition = existing_backup_disposition(
+        existing_backup.as_ref().map(|backup| backup.shard.as_str()),
+        target_shard.as_str(),
+    );
+    let reuse_existing_backup = matches!(
+        existing_backup_disposition,
+        ExistingBackupDisposition::PromptReuse
+    );
+
+    match existing_backup_disposition {
+        ExistingBackupDisposition::ProceedFresh => {}
+        ExistingBackupDisposition::AbortWrongShard(backup_shard) => {
+            bail!(
+                "`{}` already exists on shard `{}`. Remove it with `graphman chain remove {}` before changing `{}` to shard `{}`",
+                canonical_backup_name,
+                backup_shard,
+                canonical_backup_name,
+                chain_name,
+                target_shard,
+            );
+        }
+        ExistingBackupDisposition::PromptReuse => {
+            let prompt = format!(
+                "`{}` already exists on shard `{}` and will be reused as the active `{}` chain.\nProceed?",
+                canonical_backup_name, target_shard, chain_name
+            );
+            if !prompt_for_confirmation(&prompt)? {
+                println!(
+                    "Aborting. Remove `{}` with `graphman chain remove {}` if you want to create a fresh cache on shard `{}`.",
+                    canonical_backup_name, canonical_backup_name, target_shard
+                );
+                return Ok(());
+            }
+        }
+    }
+
+    let existing_backup_store = if reuse_existing_backup {
+        store.chain_store(&canonical_backup_name).await
+    } else {
+        None
+    };
 
     let allocated_chain = if reuse_existing_backup {
         None
@@ -322,24 +374,20 @@ pub async fn change_block_cache_shard(
             )
             .execute(conn).await?;
 
-            let previous_backup_final_name = if let Some(backup) = existing_backup.as_ref() {
-                let temp_name = next_backup_name(conn, &chain_name, "old").await?;
+            let temp_backup_name = if let Some(backup) = existing_backup.as_ref() {
+                let temp_name = next_temporary_backup_name(conn, &chain_name, "old").await?;
                 update_chain_name(conn, &backup.name, &temp_name).await?;
-
-                if reuse_existing_backup {
-                    update_chain_name(conn, &temp_name, &chain_name).await?;
-                    Some(chain_name.clone())
-                } else {
-                    Some(temp_name)
-                }
+                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?;
+            update_chain_name(conn, &chain_name, &canonical_backup_name).await?;
 
-            if !reuse_existing_backup {
+            if reuse_existing_backup {
+                debug_assert!(temp_backup_name.is_some());
+                update_chain_name(conn, temp_backup_name.as_ref().unwrap(), &chain_name).await?;
+            } else {
                 add_chain(conn, &chain_name, &target_shard, ident.clone()).await?;
             }
 
@@ -349,43 +397,34 @@ pub async fn change_block_cache_shard(
             .execute(conn).await?;
 
             Ok(ChainSwapOutcome {
-                latest_backup_name,
-                previous_backup_final_name,
                 reused_previous_backup: reuse_existing_backup,
                 allocated_chain: allocated_chain.is_some(),
             })
         }.scope_boxed()
     }).await?;
 
     let ChainSwapOutcome {
-        latest_backup_name,
-        previous_backup_final_name,
         reused_previous_backup,
         allocated_chain,
     } = outcome;
 
-    chain_store.update_name(&latest_backup_name).await?;
+    chain_store.update_name(&canonical_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?;
+    if reused_previous_backup && let Some(backup_store) = existing_backup_store.as_ref() {
+        backup_store.update_name(&chain_name).await?;
     }
 
     println!(
         "Changed block cache shard for {} from {} to {}",
         chain_name, old_shard, shard
     );
-    println!("Latest backup recorded as `{}`", latest_backup_name);
+    println!("Latest backup recorded as `{}`", canonical_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 {
-        println!("Preserved earlier backup as `{}`", preserved);
     }
 
     if allocated_chain {
@@ -431,7 +470,9 @@ pub async fn ingest(
 
 #[cfg(test)]
 mod tests {
-    use super::{backup_name, suffixed_backup_name};
+    use super::{
+        ExistingBackupDisposition, backup_name, existing_backup_disposition, suffixed_backup_name,
+    };
 
     #[test]
     fn backup_name_uses_plain_name_first() {
@@ -443,4 +484,28 @@ mod tests {
         assert_eq!(suffixed_backup_name("mainnet-old", 1), "mainnet-old-1");
         assert_eq!(suffixed_backup_name("mainnet-old", 42), "mainnet-old-42");
     }
+
+    #[test]
+    fn existing_backup_disposition_proceeds_when_backup_missing() {
+        assert_eq!(
+            existing_backup_disposition(None, "shard_b"),
+            ExistingBackupDisposition::ProceedFresh
+        );
+    }
+
+    #[test]
+    fn existing_backup_disposition_prompts_when_backup_matches_target_shard() {
+        assert_eq!(
+            existing_backup_disposition(Some("shard_b"), "shard_b"),
+            ExistingBackupDisposition::PromptReuse
+        );
+    }
+
+    #[test]
+    fn existing_backup_disposition_aborts_when_backup_is_on_another_shard() {
+        assert_eq!(
+            existing_backup_disposition(Some("shard_a"), "shard_b"),
+            ExistingBackupDisposition::AbortWrongShard("shard_a")
+        );
+    }
 }