gnd: Convert command handlers to async to fix nested runtime panic

lutter · claude · lutter · commit 34b6e8478cab · 2026-01-19T16:07:53.000-08:00
The main() function uses #[tokio::main] which creates a tokio runtime,
but init, add, and build commands were creating their own runtimes and
calling block_on(), causing "Cannot start a runtime from within a
runtime" errors.

Changes:
- Convert run_init, run_add, run_build to async functions
- Convert helper functions (run_interactive, init_from_contract,
  get_contract_info, build_subgraph, watch_and_build) to async
- Replace std::sync::mpsc with tokio::sync::mpsc in file watcher
- Replace std::thread::sleep with tokio::time::sleep
- Update main.rs to await the async command handlers
- Convert affected tests to #[tokio::test]

Co-Authored-By: Claude Opus 4.5 &lt;noreply@anthropic.com&gt;
diff --git a/gnd/src/commands/add.rs b/gnd/src/commands/add.rs
@@ -49,7 +49,7 @@ pub struct AddOpt {
 }
 
 /// Run the add command.
-pub fn run_add(opt: AddOpt) -> Result<()> {
+pub async fn run_add(opt: AddOpt) -> Result<()> {
     // Validate address format first
     if !opt.address.starts_with("0x") || opt.address.len() != 42 {
         return Err(anyhow!(
@@ -94,7 +94,7 @@ pub fn run_add(opt: AddOpt) -> Result<()> {
         .unwrap_or_else(|| "mainnet".to_string());
 
     // Fetch or load ABI
-    let (abi, contract_name, start_block) = get_contract_info(&opt, &network)?;
+    let (abi, contract_name, start_block) = get_contract_info(&opt, &network).await?;
 
     // Get project directory
     let project_dir = opt.manifest.parent().unwrap_or(Path::new("."));
@@ -143,7 +143,10 @@ pub fn run_add(opt: AddOpt) -> Result<()> {
 }
 
 /// Get contract info (ABI, name, start block) from local file or network.
-fn get_contract_info(opt: &AddOpt, network: &str) -> Result<(JsonValue, String, Option<u64>)> {
+async fn get_contract_info(
+    opt: &AddOpt,
+    network: &str,
+) -> Result<(JsonValue, String, Option<u64>)> {
     if let Some(abi_path) = &opt.abi {
         // Load ABI from file
         step(
@@ -175,18 +178,14 @@ fn get_contract_info(opt: &AddOpt, network: &str) -> Result<(JsonValue, String,
             &format!("Fetching ABI from {} network", network),
         );
 
-        let runtime = tokio::runtime::Runtime::new().context("Failed to create async runtime")?;
-
-        let contract_info = runtime.block_on(async {
-            let service = ContractService::load()
-                .await
-                .context("Failed to load contract service")?;
+        let service = ContractService::load()
+            .await
+            .context("Failed to load contract service")?;
 
-            service
-                .get_contract_info(network, &opt.address)
-                .await
-                .context("Failed to fetch contract info")
-        })?;
+        let contract_info = service
+            .get_contract_info(network, &opt.address)
+            .await
+            .context("Failed to fetch contract info")?;
 
         let contract_name = opt.contract_name.clone().unwrap_or(contract_info.name);
 
@@ -601,8 +600,8 @@ mod tests {
     use super::*;
     use crate::scaffold::manifest::{EventInfo, EventInput};
 
-    #[test]
-    fn test_invalid_address() {
+    #[tokio::test]
+    async fn test_invalid_address() {
         let opt = AddOpt {
             address: "invalid".to_string(),
             manifest: PathBuf::from("subgraph.yaml"),
@@ -613,7 +612,7 @@ mod tests {
             start_block: None,
         };
 
-        let result = run_add(opt);
+        let result = run_add(opt).await;
         assert!(result.is_err());
         assert!(result
             .unwrap_err()
@@ -754,8 +753,8 @@ mod tests {
         assert!(!mapping.contains("export function handle"));
     }
 
-    #[test]
-    fn test_missing_manifest() {
+    #[tokio::test]
+    async fn test_missing_manifest() {
         let opt = AddOpt {
             address: "0x1234567890123456789012345678901234567890".to_string(),
             manifest: PathBuf::from("nonexistent.yaml"),
@@ -766,7 +765,7 @@ mod tests {
             start_block: None,
         };
 
-        let result = run_add(opt);
+        let result = run_add(opt).await;
         assert!(result.is_err());
         assert!(result.unwrap_err().to_string().contains("Manifest file"));
     }
diff --git a/gnd/src/commands/build.rs b/gnd/src/commands/build.rs
@@ -7,9 +7,10 @@
 use std::collections::{HashMap, HashSet};
 use std::fs;
 use std::path::{Path, PathBuf};
-use std::sync::mpsc;
 use std::time::Duration;
 
+use tokio::sync::mpsc;
+
 use anyhow::{anyhow, Context, Result};
 use clap::Parser;
 use notify::{recommended_watcher, RecursiveMode, Watcher};
@@ -72,7 +73,7 @@ pub struct BuildResult {
 }
 
 /// Run the build command.
-pub fn run_build(opt: BuildOpt) -> Result<Option<String>> {
+pub async fn run_build(opt: BuildOpt) -> Result<Option<String>> {
     // Validate output format
     if opt.output_format != "wasm" && opt.output_format != "wast" {
         return Err(anyhow!(
@@ -82,18 +83,18 @@ pub fn run_build(opt: BuildOpt) -> Result<Option<String>> {
     }
 
     if opt.watch {
-        watch_and_build(&opt)?;
+        watch_and_build(&opt).await?;
         Ok(None)
     } else {
-        let result = build_subgraph(&opt)?;
+        let result = build_subgraph(&opt).await?;
         Ok(result.ipfs_hash)
     }
 }
 
 /// Watch subgraph files and rebuild on changes.
-fn watch_and_build(opt: &BuildOpt) -> Result<()> {
+async fn watch_and_build(opt: &BuildOpt) -> Result<()> {
     // Do initial build
-    if let Err(e) = build_subgraph(opt) {
+    if let Err(e) = build_subgraph(opt).await {
         eprintln!("Error during initial build: {}", e);
     }
 
@@ -104,12 +105,13 @@ fn watch_and_build(opt: &BuildOpt) -> Result<()> {
     println!("\nWatching subgraph files for changes...");
     println!("Press Ctrl+C to stop.\n");
 
-    // Set up file watcher
-    let (tx, rx) = mpsc::channel();
+    // Set up file watcher with tokio channel
+    let (tx, mut rx) = mpsc::channel(100);
 
     let mut watcher = recommended_watcher(move |res| {
         if let Ok(event) = res {
-            let _ = tx.send(event);
+            // Use blocking_send since we're in a sync callback
+            let _ = tx.blocking_send(event);
         }
     })
     .map_err(|e| anyhow!("Failed to create file watcher: {}", e))?;
@@ -128,8 +130,8 @@ fn watch_and_build(opt: &BuildOpt) -> Result<()> {
 
     // Event loop
     loop {
-        match rx.recv() {
-            Ok(event) => {
+        match rx.recv().await {
+            Some(event) => {
                 // Check if the event is for a file we care about
                 let relevant = event.paths.iter().any(|p| {
                     files_to_watch.iter().any(|f| {
@@ -140,7 +142,7 @@ fn watch_and_build(opt: &BuildOpt) -> Result<()> {
 
                 if relevant {
                     // Debounce: wait a bit and drain any pending events
-                    std::thread::sleep(WATCH_DEBOUNCE);
+                    tokio::time::sleep(WATCH_DEBOUNCE).await;
                     while rx.try_recv().is_ok() {}
 
                     // Get the changed file for logging
@@ -152,12 +154,12 @@ fn watch_and_build(opt: &BuildOpt) -> Result<()> {
 
                     println!("\nFile change detected: {}\n", changed);
 
-                    if let Err(e) = build_subgraph(opt) {
+                    if let Err(e) = build_subgraph(opt).await {
                         eprintln!("Error during rebuild: {}", e);
                     }
                 }
             }
-            Err(_) => {
+            None => {
                 return Err(anyhow!("File watcher channel closed"));
             }
         }
@@ -194,7 +196,7 @@ fn get_files_to_watch(manifest_path: &Path, manifest: &Manifest) -> Vec<PathBuf>
 }
 
 /// Build the subgraph.
-fn build_subgraph(opt: &BuildOpt) -> Result<BuildResult> {
+async fn build_subgraph(opt: &BuildOpt) -> Result<BuildResult> {
     // Apply migrations unless skipped
     if !opt.skip_migrations {
         migrations::apply_migrations(&opt.manifest)?;
@@ -287,9 +289,7 @@ fn build_subgraph(opt: &BuildOpt) -> Result<BuildResult> {
 
     // Upload to IPFS if requested
     let ipfs_hash = if let Some(ipfs_url) = &opt.ipfs {
-        let runtime = tokio::runtime::Runtime::new()
-            .context("Failed to create async runtime for IPFS upload")?;
-        let hash = runtime.block_on(upload_to_ipfs(ipfs_url, &opt.output_dir, &manifest))?;
+        let hash = upload_to_ipfs(ipfs_url, &opt.output_dir, &manifest).await?;
         step(Step::Done, &format!("Build completed: {}", hash));
         Some(hash)
     } else {
diff --git a/gnd/src/commands/deploy.rs b/gnd/src/commands/deploy.rs
@@ -137,7 +137,7 @@ async fn build_and_upload(opt: &DeployOpt) -> Result<String> {
         network_file: opt.network_file.clone(),
     };
 
-    match run_build(build_opt)? {
+    match run_build(build_opt).await? {
         Some(ipfs_hash) => Ok(ipfs_hash),
         None => Err(anyhow!(
             "Build succeeded but no IPFS hash was returned. This is unexpected."
diff --git a/gnd/src/commands/init.rs b/gnd/src/commands/init.rs
@@ -112,7 +112,7 @@ pub struct InitOpt {
 }
 
 /// Run the init command.
-pub fn run_init(opt: InitOpt) -> Result<()> {
+pub async fn run_init(opt: InitOpt) -> Result<()> {
     // Check if we need interactive mode
     let needs_interactive = should_run_interactive(&opt);
 
@@ -127,7 +127,7 @@ pub fn run_init(opt: InitOpt) -> Result<()> {
             ));
         }
 
-        return run_interactive(opt);
+        return run_interactive(opt).await;
     }
 
     // Non-interactive mode - determine the scaffold source
@@ -143,7 +143,7 @@ pub fn run_init(opt: InitOpt) -> Result<()> {
     };
 
     match source {
-        ScaffoldSource::Contract => init_from_contract(&opt),
+        ScaffoldSource::Contract => init_from_contract(&opt).await,
         ScaffoldSource::Example => init_from_example(&opt),
         ScaffoldSource::Subgraph => init_from_subgraph(&opt),
     }
@@ -176,12 +176,11 @@ fn should_run_interactive(opt: &InitOpt) -> bool {
 }
 
 /// Run in interactive mode.
-fn run_interactive(opt: InitOpt) -> Result<()> {
+async fn run_interactive(opt: InitOpt) -> Result<()> {
     println!("Creating a new subgraph...\n");
 
     // Load the networks registry
-    let runtime = tokio::runtime::Runtime::new().context("Failed to create async runtime")?;
-    let registry = runtime.block_on(async { NetworksRegistry::load().await })?;
+    let registry = NetworksRegistry::load().await?;
 
     // Parse start block if provided
     let start_block = opt.start_block.as_ref().and_then(|s| s.parse::<u64>().ok());
@@ -229,7 +228,7 @@ fn run_interactive(opt: InitOpt) -> Result<()> {
                 skip_git: opt.skip_git,
                 ..Default::default()
             };
-            init_from_contract(&contract_opt)
+            init_from_contract(&contract_opt).await
         }
     }
 }
@@ -241,7 +240,7 @@ enum ScaffoldSource {
 }
 
 /// Initialize a subgraph from a contract address.
-fn init_from_contract(opt: &InitOpt) -> Result<()> {
+async fn init_from_contract(opt: &InitOpt) -> Result<()> {
     let address = opt
         .from_contract
         .as_ref()
@@ -262,48 +261,44 @@ fn init_from_contract(opt: &InitOpt) -> Result<()> {
         &format!("Fetching contract info from {} on {}", address, network),
     );
 
-    // Create async runtime to fetch contract info
-    let runtime = tokio::runtime::Runtime::new().context("Failed to create async runtime")?;
-
-    let contract_info = runtime.block_on(async {
+    // Fetch contract info
+    let contract_info = if let Some(abi_path) = &opt.abi {
         // Load ABI from file if provided
-        if let Some(abi_path) = &opt.abi {
-            let abi_str = fs::read_to_string(abi_path)
-                .with_context(|| format!("Failed to read ABI file: {}", abi_path.display()))?;
-            let abi: serde_json::Value = serde_json::from_str(&abi_str)
-                .with_context(|| format!("Failed to parse ABI file: {}", abi_path.display()))?;
-
-            // Try to get start block from API if not provided
-            let start_block = if let Some(block) = &opt.start_block {
-                block.parse::<u64>().ok()
-            } else {
-                // Try to fetch from API
-                match ContractService::load().await {
-                    Ok(service) => service.get_start_block(network, address).await.ok(),
-                    Err(_) => None,
-                }
-            };
-
-            Ok::<_, anyhow::Error>(crate::services::ContractInfo {
-                abi,
-                name: opt
-                    .contract_name
-                    .clone()
-                    .unwrap_or_else(|| "Contract".to_string()),
-                start_block,
-            })
+        let abi_str = fs::read_to_string(abi_path)
+            .with_context(|| format!("Failed to read ABI file: {}", abi_path.display()))?;
+        let abi: serde_json::Value = serde_json::from_str(&abi_str)
+            .with_context(|| format!("Failed to parse ABI file: {}", abi_path.display()))?;
+
+        // Try to get start block from API if not provided
+        let start_block = if let Some(block) = &opt.start_block {
+            block.parse::<u64>().ok()
         } else {
-            // Fetch ABI from Etherscan/Sourcify
-            let service = ContractService::load()
-                .await
-                .context("Failed to load contract service")?;
-
-            service
-                .get_contract_info(network, address)
-                .await
-                .context("Failed to fetch contract info")
+            // Try to fetch from API
+            match ContractService::load().await {
+                Ok(service) => service.get_start_block(network, address).await.ok(),
+                Err(_) => None,
+            }
+        };
+
+        crate::services::ContractInfo {
+            abi,
+            name: opt
+                .contract_name
+                .clone()
+                .unwrap_or_else(|| "Contract".to_string()),
+            start_block,
         }
-    })?;
+    } else {
+        // Fetch ABI from Etherscan/Sourcify
+        let service = ContractService::load()
+            .await
+            .context("Failed to load contract service")?;
+
+        service
+            .get_contract_info(network, address)
+            .await
+            .context("Failed to fetch contract info")?
+    };
 
     step(
         Step::Done,
diff --git a/gnd/src/main.rs b/gnd/src/main.rs
@@ -162,7 +162,7 @@ async fn main() -> Result<()> {
             }
         }
         Commands::Build(build_opt) => {
-            if let Err(e) = run_build(build_opt) {
+            if let Err(e) = run_build(build_opt).await {
                 eprintln!("Error: {}", e);
                 std::process::exit(1);
             }
@@ -174,13 +174,13 @@ async fn main() -> Result<()> {
             }
         }
         Commands::Init(init_opt) => {
-            if let Err(e) = run_init(init_opt) {
+            if let Err(e) = run_init(init_opt).await {
                 eprintln!("Error: {}", e);
                 std::process::exit(1);
             }
         }
         Commands::Add(add_opt) => {
-            if let Err(e) = run_add(add_opt) {
+            if let Err(e) = run_add(add_opt).await {
                 eprintln!("Error: {}", e);
                 std::process::exit(1);
             }

Original file line number	Diff line number	Diff line change
`@@ -162,7 +162,7 @@ async fn main() -> Result<()> {`
`162`	`162`	`}`
`163`	`163`	`}`
`164`	`164`	`Commands::Build(build_opt) => {`
`165`		`- if let Err(e) = run_build(build_opt) {`
	`165`	`+ if let Err(e) = run_build(build_opt).await {`
`166`	`166`	`eprintln!("Error: {}", e);`
`167`	`167`	`std::process::exit(1);`
`168`	`168`	`}`
`@@ -174,13 +174,13 @@ async fn main() -> Result<()> {`
`174`	`174`	`}`
`175`	`175`	`}`
`176`	`176`	`Commands::Init(init_opt) => {`
`177`		`- if let Err(e) = run_init(init_opt) {`
	`177`	`+ if let Err(e) = run_init(init_opt).await {`
`178`	`178`	`eprintln!("Error: {}", e);`
`179`	`179`	`std::process::exit(1);`
`180`	`180`	`}`
`181`	`181`	`}`
`182`	`182`	`Commands::Add(add_opt) => {`
`183`		`- if let Err(e) = run_add(add_opt) {`
	`183`	`+ if let Err(e) = run_add(add_opt).await {`
`184`	`184`	`eprintln!("Error: {}", e);`
`185`	`185`	`std::process::exit(1);`
`186`	`186`	`}`