From a6467ea056febf5814e756d7a304050fc85d2e98 Mon Sep 17 00:00:00 2001 From: illuzen Date: Wed, 29 Jul 2026 16:56:04 +0800 Subject: [PATCH 1/6] fix(collect-rewards): treat on-chain UsedNullifiers as authoritative Stale Subsquid nullifier data could leave spent transfers in the candidate list, wasting proof generation and failing with NullifierAlreadyUsed. Always verify against chain storage, surface ExtrinsicFailed, and skip spent batches instead of aborting the whole collection. Co-authored-by: Cursor --- src/cli/common.rs | 2 +- src/collect_rewards_lib.rs | 261 ++++++++++++++++++++++++++++--------- 2 files changed, 199 insertions(+), 64 deletions(-) diff --git a/src/cli/common.rs b/src/cli/common.rs index d8f6fcd..efbb116 100644 --- a/src/cli/common.rs +++ b/src/cli/common.rs @@ -731,7 +731,7 @@ async fn wait_tx_inclusion( } } -fn format_dispatch_error( +pub(crate) fn format_dispatch_error( error: &crate::chain::quantus_subxt::api::runtime_types::sp_runtime::DispatchError, metadata: &subxt::Metadata, ) -> String { diff --git a/src/collect_rewards_lib.rs b/src/collect_rewards_lib.rs index 1dcba28..16a193b 100644 --- a/src/collect_rewards_lib.rs +++ b/src/collect_rewards_lib.rs @@ -5,9 +5,10 @@ //! //! The main function `collect_rewards` handles the entire flow: //! 1. Query Subsquid for pending transfers -//! 2. Generate ZK proofs for each transfer -//! 3. Aggregate proofs into batches -//! 4. Submit withdrawal transactions to chain +//! 2. Filter spent nullifiers (Subsquid pre-filter + authoritative on-chain UsedNullifiers) +//! 3. Generate ZK proofs for each unspent transfer +//! 4. Aggregate proofs into batches +//! 5. Submit withdrawal transactions to chain //! //! This is designed to be called from the SDK without needing CLI-specific features. @@ -220,9 +221,10 @@ pub struct CollectRewardsConfig { /// This is the main entry point for the SDK to collect rewards. It handles the entire flow: /// 1. Derives wormhole address from mnemonic /// 2. Queries Subsquid for pending transfers to that address -/// 3. Generates ZK proofs for selected transfers -/// 4. Aggregates proofs into batches (size determined by circuit config) -/// 5. Submits withdrawal transactions to chain +/// 3. Filters spent nullifiers via on-chain `UsedNullifiers` (Subsquid is a pre-filter only) +/// 4. Generates ZK proofs for selected unspent transfers +/// 5. Aggregates proofs into batches (size determined by circuit config) +/// 6. Submits withdrawal transactions to chain (skips batches that race on spent nullifiers) /// /// # Arguments /// * `config` - Configuration for the operation @@ -267,18 +269,19 @@ pub async fn collect_rewards( }); } - // Step 2b: Connect to chain (needed for nullifier fallback and proof submission) + // Step 2b: Connect to chain (needed for on-chain nullifier checks and proof submission) progress.on_step("connect", "Connecting to chain"); let quantus_client = QuantusClient::new(&config.node_url) .await .map_err(|e| CollectRewardsError::from(format!("Failed to connect to node: {}", e)))?; - // Step 2c: Filter out already-spent transfers by checking nullifiers - // Tries Subsquid first, falls back to on-chain checking if Subsquid fails - progress.on_step("nullifiers", "Checking for already-spent nullifiers"); + // Step 2c: Filter out already-spent transfers. + // Subsquid is a best-effort pre-filter; on-chain UsedNullifiers is authoritative. + progress.on_step("nullifiers", "Checking for already-spent nullifiers (on-chain)"); - let unspent_transfers = filter_unspent_transfers_with_fallback( + let incoming_count = incoming_transfers.len(); + let unspent_transfers = filter_unspent_transfers( &incoming_transfers, &wormhole_secret_bytes, &subsquid_client, @@ -286,6 +289,18 @@ pub async fn collect_rewards( ) .await?; + let spent_filtered = incoming_count.saturating_sub(unspent_transfers.len()); + if spent_filtered > 0 { + progress.on_step( + "nullifiers", + &format!( + "Filtered {} already-spent transfer(s) via on-chain UsedNullifiers ({} remaining)", + spent_filtered, + unspent_transfers.len() + ), + ); + } + if unspent_transfers.is_empty() { progress .on_step("complete", "All transfers have already been withdrawn (nullifiers spent)"); @@ -343,6 +358,22 @@ pub async fn collect_rewards( }); } + // Refuse to generate proofs against an unsupported runtime + let (spec_version, transaction_version) = quantus_client.get_runtime_version().await.map_err( + |e| CollectRewardsError::from(format!("Failed to get runtime version: {}", e)), + )?; + if !crate::config::is_runtime_compatible(spec_version, transaction_version) { + let supported = crate::config::COMPATIBLE_RUNTIMES + .iter() + .map(|r| format!("spec {} / tx {}", r.spec_version, r.transaction_version)) + .collect::>() + .join(", "); + return Err(CollectRewardsError::from(format!( + "CLI is incompatible with connected runtime (spec {}, tx {}). Supported: {}. Refusing to generate proofs.", + spec_version, transaction_version, supported + ))); + } + // Get block for proofs - either specific block or latest let proof_block = if let Some(block_num) = config.at_block { // Fetch block hash for the specified block number @@ -381,14 +412,27 @@ pub async fn collect_rewards( let proof_block_hash = proof_block.hash(); // Step 4: Generate proofs - progress.on_step("proofs", &format!("Generating {} proofs", selected_transfers.len())); + progress.on_step("proofs", &format!("Generating up to {} proofs", selected_transfers.len())); let bins_dir = Path::new(&config.bins_dir); let num_transfers = selected_transfers.len(); let mut proof_bytes_list: Vec> = Vec::new(); for (i, transfer) in selected_transfers.iter().enumerate() { - progress.on_proof_generated(i + 1, num_transfers); + // Re-check on-chain immediately before proving — nullifiers can be spent + // while earlier proofs in this run are still being generated. + if is_nullifier_spent_onchain(&quantus_client, &wormhole_secret_bytes, transfer).await? { + progress.on_step( + "nullifiers", + &format!( + "Skipping transfer {} ({}/{}) — nullifier already spent on-chain", + transfer.id, + i + 1, + num_transfers + ), + ); + continue; + } let leaf_index: u64 = transfer.leaf_index.parse().map_err(|_| { CollectRewardsError::from(format!("Invalid leaf_index: {}", transfer.leaf_index)) @@ -469,6 +513,21 @@ pub async fn collect_rewards( .map_err(|e| CollectRewardsError::from(e.message))?; proof_bytes_list.push(result.proof_bytes); + progress.on_proof_generated(proof_bytes_list.len(), num_transfers); + } + + if proof_bytes_list.is_empty() { + progress.on_step( + "complete", + "No unspent transfers remained after on-chain nullifier checks", + ); + return Ok(CollectRewardsResult { + wormhole_address, + destination_address: config.destination_address, + total_withdrawn: 0, + batches: vec![], + transfers_processed: 0, + }); } // Step 5: Aggregate and submit in batches @@ -486,17 +545,31 @@ pub async fn collect_rewards( let mut total_withdrawn: u128 = 0; let mut batch_results: Vec = Vec::new(); + let mut transfers_processed = 0usize; for (batch_idx, batch_proofs) in batches.iter().enumerate() { // Aggregate proofs let aggregated_proof = aggregate_proof_bytes(batch_proofs, bins_dir)?; - // Submit to chain + // Submit to chain — skip batches that race with already-spent nullifiers let (block_hash, tx_hash, transfer_events) = - submit_and_get_events(&quantus_client, aggregated_proof, bins_dir).await?; + match submit_and_get_events(&quantus_client, aggregated_proof, bins_dir).await { + Ok(result) => result, + Err(e) if is_spent_nullifier_error(&e) => { + progress.on_error(&format!( + "Batch {}/{} skipped (spent nullifier): {} — continuing with remaining batches", + batch_idx + 1, + batches.len(), + e.message + )); + continue; + }, + Err(e) => return Err(e), + }; let batch_amount: u128 = transfer_events.iter().map(|e| e.amount).sum(); total_withdrawn += batch_amount; + transfers_processed += batch_proofs.len(); progress.on_batch_submitted(batch_idx + 1, batches.len(), batch_amount); @@ -513,7 +586,7 @@ pub async fn collect_rewards( destination_address: config.destination_address, total_withdrawn, batches: batch_results, - transfers_processed: selected_transfers.len(), + transfers_processed, }) } @@ -948,8 +1021,11 @@ async fn submit_and_get_events( CollectRewardsError::from("Could not find extrinsic in block".to_string()) })?; + use crate::chain::quantus_subxt::api::system::events::ExtrinsicFailed; + let mut transfer_events = Vec::new(); let mut found_proof_verified = false; + let mut dispatch_error_msg: Option = None; for event_result in events.iter() { let event = event_result @@ -957,6 +1033,15 @@ async fn submit_and_get_events( if let subxt::events::Phase::ApplyExtrinsic(ext_idx) = event.phase() { if ext_idx == our_ext_idx { + if let Ok(Some(ExtrinsicFailed { dispatch_error, .. })) = + event.as_event::() + { + let metadata = quantus_client.client().metadata(); + dispatch_error_msg = Some(crate::cli::common::format_dispatch_error( + &dispatch_error, + &metadata, + )); + } if let Ok(Some(_)) = event.as_event::() { found_proof_verified = true; } @@ -969,6 +1054,12 @@ async fn submit_and_get_events( } if !found_proof_verified { + if let Some(msg) = dispatch_error_msg { + return Err(CollectRewardsError::from(format!( + "Proof verification failed: {}", + msg + ))); + } return Err(CollectRewardsError::from( "Proof verification failed - no ProofVerified event".to_string(), )); @@ -977,11 +1068,49 @@ async fn submit_and_get_events( Ok((block_hash, tx_hash, transfer_events)) } +/// Whether an error indicates a nullifier was already spent (safe to skip a batch). +fn is_spent_nullifier_error(err: &CollectRewardsError) -> bool { + err.message.contains("NullifierAlreadyUsed") +} + +/// Compute the nullifier for a transfer and check on-chain `UsedNullifiers`. +async fn is_nullifier_spent_onchain( + quantus_client: &QuantusClient, + secret_bytes: &[u8; 32], + transfer: &Transfer, +) -> Result { + let ctx = format!("transfer {}", transfer.id); + let transfer_count = parse_transfer_count(&transfer.transfer_count, &ctx)?; + + let nullifier = + wormhole_lib::compute_nullifier(secret_bytes, transfer_count).map_err(|e| { + CollectRewardsError::from(format!("Failed to compute nullifier: {}", e.message)) + })?; + + let storage_key = quantus_node::api::storage().wormhole().used_nullifiers(nullifier); + let is_used = quantus_client + .client() + .storage() + .at_latest() + .await + .map_err(|e| CollectRewardsError::from(format!("Failed to get storage: {}", e)))? + .fetch(&storage_key) + .await + .map_err(|e| { + CollectRewardsError::from(format!( + "Failed to query nullifier for transfer_count {}: {}", + transfer_count, e + )) + })?; + + Ok(is_used.is_some()) +} + /// Filter out transfers whose nullifiers have already been spent (on-chain check). /// /// For each transfer, computes the nullifier from (secret, transfer_count) /// and checks on-chain storage to see if it's been consumed by a previous withdrawal. -/// This is more reliable than Subsquid as it queries the chain directly. +/// This is authoritative relative to Subsquid, which can lag. async fn filter_unspent_transfers_onchain( transfers: &[Transfer], secret_bytes: &[u8; 32], @@ -992,32 +1121,8 @@ async fn filter_unspent_transfers_onchain( } let mut unspent = Vec::new(); - let storage = quantus_client - .client() - .storage() - .at_latest() - .await - .map_err(|e| CollectRewardsError::from(format!("Failed to get storage: {}", e)))?; - for transfer in transfers { - let ctx = format!("transfer {}", transfer.id); - let transfer_count = parse_transfer_count(&transfer.transfer_count, &ctx)?; - - let nullifier = - wormhole_lib::compute_nullifier(secret_bytes, transfer_count).map_err(|e| { - CollectRewardsError::from(format!("Failed to compute nullifier: {}", e.message)) - })?; - - // Query on-chain UsedNullifiers storage - let storage_key = quantus_node::api::storage().wormhole().used_nullifiers(nullifier); - let is_used = storage.fetch(&storage_key).await.map_err(|e| { - CollectRewardsError::from(format!( - "Failed to query nullifier for transfer_count {}: {}", - transfer_count, e - )) - })?; - - if is_used.is_none() { + if !is_nullifier_spent_onchain(quantus_client, secret_bytes, transfer).await? { unspent.push(transfer.clone()); } } @@ -1025,16 +1130,13 @@ async fn filter_unspent_transfers_onchain( Ok(unspent) } -/// Filter out transfers whose nullifiers have already been spent. +/// Best-effort Subsquid pre-filter for spent nullifiers. /// -/// For each transfer, computes the nullifier from (secret, transfer_count) -/// and checks Subsquid to see if it's been consumed by a previous withdrawal. -/// If Subsquid query fails, falls back to on-chain checking. -async fn filter_unspent_transfers_with_fallback( +/// A successful response may still be stale; callers must verify with on-chain storage. +async fn filter_unspent_transfers_subsquid( transfers: &[Transfer], secret_bytes: &[u8; 32], subsquid_client: &SubsquidClient, - quantus_client: &QuantusClient, ) -> Result> { use std::collections::HashSet; @@ -1042,7 +1144,6 @@ async fn filter_unspent_transfers_with_fallback( return Ok(vec![]); } - // Compute nullifiers for all transfers // Map: nullifier_hex -> (nullifier_hash, transfer) let mut nullifier_map: std::collections::HashMap = std::collections::HashMap::new(); @@ -1061,31 +1162,50 @@ async fn filter_unspent_transfers_with_fallback( nullifier_map.insert(nullifier_hex, (nullifier_hash, transfer)); } - // Build list for Subsquid query let nullifier_pairs: Vec<(String, String)> = nullifier_map .iter() .map(|(nul_hex, (nul_hash, _))| (nul_hex.clone(), nul_hash.clone())) .collect(); - // Try Subsquid first, fall back to on-chain if it fails let spent_nullifiers: HashSet = - match subsquid_client.check_nullifiers_spent(&nullifier_pairs, 8).await { - Ok(spent) => spent, - Err(_) => { - // Fall back to on-chain checking - return filter_unspent_transfers_onchain(transfers, secret_bytes, quantus_client) - .await; - }, - }; + subsquid_client.check_nullifiers_spent(&nullifier_pairs, 8).await?; - // Filter to only unspent transfers - let unspent: Vec = nullifier_map + Ok(nullifier_map .into_iter() .filter(|(nul_hex, _)| !spent_nullifiers.contains(nul_hex)) .map(|(_, (_, transfer))| transfer.clone()) - .collect(); + .collect()) +} - Ok(unspent) +/// Filter out transfers whose nullifiers have already been spent. +/// +/// Uses Subsquid as a best-effort pre-filter when available, then always verifies +/// remaining candidates against on-chain `UsedNullifiers`. On-chain is authoritative — +/// a successful but stale Subsquid response must not be trusted alone. +async fn filter_unspent_transfers( + transfers: &[Transfer], + secret_bytes: &[u8; 32], + subsquid_client: &SubsquidClient, + quantus_client: &QuantusClient, +) -> Result> { + if transfers.is_empty() { + return Ok(vec![]); + } + + // Optional fast path: shrink the candidate set via Subsquid when it works. + // On failure or staleness we still run the on-chain check below. + let candidates = match filter_unspent_transfers_subsquid( + transfers, + secret_bytes, + subsquid_client, + ) + .await + { + Ok(filtered) => filtered, + Err(_) => transfers.to_vec(), + }; + + filter_unspent_transfers_onchain(&candidates, secret_bytes, quantus_client).await } #[cfg(test)] @@ -1099,6 +1219,21 @@ mod tests { assert_eq!(format!("{}", err), "test error"); } + #[test] + fn test_is_spent_nullifier_error_matches_prevalidation_and_dispatch() { + assert!(is_spent_nullifier_error(&CollectRewardsError::from( + "Pre-validation failed: NullifierAlreadyUsed - Nullifier 0 (0xab) has already been spent" + .to_string() + ))); + assert!(is_spent_nullifier_error(&CollectRewardsError::from( + "Proof verification failed: Wormhole::NullifierAlreadyUsed - Nullifier already used" + .to_string() + ))); + assert!(!is_spent_nullifier_error(&CollectRewardsError::from( + "Proof verification failed: Wormhole::InvalidProof".to_string() + ))); + } + #[test] fn test_parse_ss58_address() { // Valid Quantus address (SS58 prefix 189) From 7289aee338592253df6d8487e22adceaa7565c9d Mon Sep 17 00:00:00 2001 From: illuzen Date: Wed, 29 Jul 2026 18:57:41 +0800 Subject: [PATCH 2/6] test(wormhole): cover collect-rewards and spent-nullifier filtering Add public-batch multiround plus collect-rewards exercise paths that fund known deposits without Subsquid, and read UsedNullifiers at best block so PoW finality lag cannot hide freshly spent nullifiers. Co-authored-by: Cursor --- src/cli/exercise/scenarios/wormhole.rs | 330 ++++++++++++++++++++++++- src/collect_rewards_lib.rs | 242 +++++++++++++----- src/lib.rs | 7 +- 3 files changed, 517 insertions(+), 62 deletions(-) diff --git a/src/cli/exercise/scenarios/wormhole.rs b/src/cli/exercise/scenarios/wormhole.rs index 2fcf482..30cc2bc 100644 --- a/src/cli/exercise/scenarios/wormhole.rs +++ b/src/cli/exercise/scenarios/wormhole.rs @@ -1,13 +1,44 @@ -//! Wormhole multiround smoke test (CPU-heavy; use `--skip wormhole` on debug builds). +//! Wormhole smoke tests (CPU-heavy; use `--skip wormhole` on debug builds). +//! +//! Covers: +//! - private-batch multiround +//! - public-batch multiround +//! - collect-rewards via known deposits (no Subsquid) +//! - spent-nullifier filtering / retry after collect use crate::{ - cli::exercise::{report::Report, runner::ExerciseCtx}, - error::Result, - exercise_step, + chain::quantus_subxt::api::wormhole as wormhole_api, + cli::{ + address_format::bytes_to_quantus_ss58, + exercise::{report::Report, runner::ExerciseCtx}, + send::transfer, + wormhole::{decode_full_leaf_data, get_zk_merkle_proof}, + }, + collect_rewards_lib::{ + collect_rewards_from_known, CollectKnownTransfersConfig, KnownTransfer, NoOpProgress, + WormholeCredential, + }, + error::{QuantusError, Result}, + exercise_step, wormhole_lib, }; +use rand::RngCore; +use std::{collections::HashSet, time::Duration}; +use tokio::time::sleep; + +/// How many recent blocks to scan for a deposit's `NativeTransferred` event. +const INCLUSION_SCAN_BLOCKS: u32 = 60; +const INCLUSION_POLL_INTERVAL_MS: u64 = 500; pub async fn run(ctx: &mut ExerciseCtx, report: &mut Report, phase: &str) -> Result<()> { exercise_step!(report, phase, "multiround", multiround(ctx)); + exercise_step!(report, phase, "multiround_public", multiround_public(ctx)); + exercise_step!(report, phase, "collect_rewards", collect_rewards_happy_path(ctx)); + exercise_step!( + report, + phase, + "collect_rewards_spent_filter", + collect_rewards_spent_filter(ctx) + ); Ok(()) } @@ -27,3 +58,294 @@ async fn multiround(ctx: &mut ExerciseCtx) -> Result { crate::cli::wormhole::handle_wormhole_command(command, &ctx.node_url).await?; Ok("wormhole multiround (5 rounds, 5 proofs each) completed".to_string()) } + +async fn multiround_public(ctx: &mut ExerciseCtx) -> Result { + let command = crate::cli::wormhole::WormholeCommands::Multiround { + num_proofs: 2, + rounds: 2, + amount: 20.0, + wallet: "crystal_alice".to_string(), + password: Some(String::new()), + password_file: None, + keep_files: false, + output_dir: "/tmp/wormhole_exercise_public".to_string(), + dry_run: false, + public: true, + }; + crate::cli::wormhole::handle_wormhole_command(command, &ctx.node_url).await?; + Ok("wormhole public-batch multiround (2 rounds, 2 proofs each) completed".to_string()) +} + +/// Fund two deposits, collect both, assert withdrawal succeeded. +async fn collect_rewards_happy_path(ctx: &mut ExerciseCtx) -> Result { + let secret = random_secret(ctx); + let mut seen_leaves = HashSet::new(); + let deposits = + fund_wormhole_deposits(ctx, &secret, 2, ctx.unit, &mut seen_leaves).await?; + let dest = ctx.eph[0].to_account_id_ss58check(); + let before = ctx.free_balance(&dest).await?; + + let result = run_collect_known(ctx, &secret, &dest, deposits.clone(), None).await?; + if result.total_withdrawn == 0 || result.batches.is_empty() { + return Err(QuantusError::Generic(format!( + "collect-rewards withdrew nothing (transfers_processed={}, batches={})", + result.transfers_processed, + result.batches.len() + ))); + } + + let after = ctx.free_balance(&dest).await?; + if after <= before { + return Err(QuantusError::Generic(format!( + "destination balance did not increase after collect ({before} -> {after})" + ))); + } + + Ok(format!( + "collected {} planck from {} deposits into {} batch(es)", + result.total_withdrawn, + deposits.len(), + result.batches.len() + )) +} + +/// Collect, then retry with spent nullifiers still in the candidate list. +/// +/// 1. Fund 2 deposits and withdraw them. +/// 2. Fund 2 more to the same wormhole address. +/// 3. Dry-run with all 4 known transfers — spent ones must be filtered (2 remain). +/// 4. Collect the 2 fresh deposits. +/// 5. Collect all 4 again — nothing left. +async fn collect_rewards_spent_filter(ctx: &mut ExerciseCtx) -> Result { + let secret = random_secret(ctx); + let dest = ctx.eph[1].to_account_id_ss58check(); + let mut seen_leaves = HashSet::new(); + + let initial = + fund_wormhole_deposits(ctx, &secret, 2, ctx.unit, &mut seen_leaves).await?; + let first = run_collect_known(ctx, &secret, &dest, initial.clone(), None) + .await + .map_err(|e| QuantusError::Generic(format!("initial collect failed: {e}")))?; + if first.total_withdrawn == 0 || first.transfers_processed != 2 { + return Err(QuantusError::Generic(format!( + "initial collect expected 2 transfers processed, got withdrawn={} transfers_processed={}", + first.total_withdrawn, first.transfers_processed + ))); + } + + // Prove UsedNullifiers is observable via the same key encoding collect-rewards uses. + assert_nullifiers_spent(&ctx.client, &secret, &initial).await?; + + let more = fund_wormhole_deposits(ctx, &secret, 2, ctx.unit, &mut seen_leaves).await?; + let mut all = initial; + all.extend(more.clone()); + + // Dry-run exercises on-chain UsedNullifiers filtering without submitting. + let filtered = run_collect_known(ctx, &secret, &dest, all.clone(), Some(true)) + .await + .map_err(|e| QuantusError::Generic(format!("mixed spent/unspent dry-run failed: {e}")))?; + if filtered.transfers_processed != 2 { + return Err(QuantusError::Generic(format!( + "dry-run over mixed spent/unspent list expected 2 remaining transfers, got {}", + filtered.transfers_processed + ))); + } + + let second = run_collect_known(ctx, &secret, &dest, more, None) + .await + .map_err(|e| QuantusError::Generic(format!("fresh-deposit collect failed: {e}")))?; + if second.total_withdrawn == 0 || second.transfers_processed != 2 { + return Err(QuantusError::Generic(format!( + "fresh-deposit collect expected 2 transfers processed, got \ + withdrawn={} transfers_processed={}", + second.total_withdrawn, second.transfers_processed + ))); + } + + let third = run_collect_known(ctx, &secret, &dest, all, None) + .await + .map_err(|e| QuantusError::Generic(format!("final spent retry failed: {e}")))?; + if third.total_withdrawn != 0 || third.transfers_processed != 0 || !third.batches.is_empty() { + return Err(QuantusError::Generic(format!( + "expected empty collect after all nullifiers spent, got withdrawn={} \ + transfers_processed={} batches={}", + third.total_withdrawn, + third.transfers_processed, + third.batches.len() + ))); + } + + Ok(format!( + "spent-nullifier filter ok: initial withdrew {}, dry-run kept 2 unspent, \ + fresh collect withdrew {}, final retry collected nothing", + first.total_withdrawn, second.total_withdrawn + )) +} + +fn random_secret(ctx: &mut ExerciseCtx) -> [u8; 32] { + let mut secret = [0u8; 32]; + ctx.rng.fill_bytes(&mut secret); + secret +} + +async fn run_collect_known( + ctx: &ExerciseCtx, + secret: &[u8; 32], + destination: &str, + transfers: Vec, + dry_run: Option, +) -> Result { + let bins_dir = crate::bins::ensure_bins_dir() + .map_err(|e| QuantusError::Generic(format!("Failed to resolve bins dir: {e}")))? + .to_string_lossy() + .into_owned(); + + let config = CollectKnownTransfersConfig { + credential: WormholeCredential::Secret { hex: hex::encode(secret) }, + destination_address: destination.to_string(), + node_url: ctx.node_url.clone(), + bins_dir, + amount: None, + dry_run: dry_run.unwrap_or(false), + at_block: None, + }; + + collect_rewards_from_known(config, transfers, &NoOpProgress) + .await + .map_err(|e| QuantusError::Generic(e.message)) +} + +/// Deposit `count` transfers of `amount_each` to the wormhole address for `secret`. +/// +/// `seen_leaves` is shared across calls so later deposits to the same address are not +/// confused with earlier `NativeTransferred` events still in the recent block window. +async fn fund_wormhole_deposits( + ctx: &ExerciseCtx, + secret: &[u8; 32], + count: usize, + amount_each: u128, + seen_leaves: &mut HashSet, +) -> Result> { + let wh_addr = wormhole_lib::compute_wormhole_address(secret) + .map_err(|e| QuantusError::Generic(e.message))?; + let wh_ss58 = bytes_to_quantus_ss58(&wh_addr); + + let mut known = Vec::with_capacity(count); + for _ in 0..count { + transfer(&ctx.client, &ctx.alice, &wh_ss58, amount_each, None, ctx.wait_mode()).await?; + + let (block_hash, event) = + wait_for_native_transferred(&ctx.client, &wh_addr, seen_leaves).await?; + seen_leaves.insert(event.leaf_index); + + // Nullifier checks must use the leaf's transfer_count (what the circuit burns). + // Prefer that over the event field so UsedNullifiers lookups stay consistent. + let merkle = get_zk_merkle_proof(&ctx.client, event.leaf_index, block_hash).await?; + let (_to, leaf_transfer_count, _asset_id, leaf_amount) = + decode_full_leaf_data(&merkle.leaf_data)?; + + known.push(KnownTransfer { + leaf_index: event.leaf_index, + transfer_count: leaf_transfer_count, + amount: leaf_amount, + }); + } + Ok(known) +} + +/// Fail if spent deposits are not visible in on-chain `UsedNullifiers` (best block). +async fn assert_nullifiers_spent( + client: &crate::chain::client::QuantusClient, + secret: &[u8; 32], + spent: &[KnownTransfer], +) -> Result<()> { + use crate::chain::quantus_subxt; + + let best = client.get_latest_block().await?; + let storage = client.client().storage().at(best); + + let mut missing = Vec::new(); + for t in spent { + let nullifier = wormhole_lib::compute_nullifier(secret, t.transfer_count) + .map_err(|e| QuantusError::Generic(e.message))?; + let key = quantus_subxt::api::storage().wormhole().used_nullifiers(nullifier); + let used = storage + .fetch(&key) + .await + .map_err(|e| QuantusError::NetworkError(format!("UsedNullifiers fetch: {e}")))?; + if !matches!(used, Some(true)) { + missing.push(format!( + "leaf={} tc={} nullifier=0x{} used={:?}", + t.leaf_index, + t.transfer_count, + hex::encode(nullifier), + used + )); + } + } + + if !missing.is_empty() { + return Err(QuantusError::Generic(format!( + "after successful collect, UsedNullifiers missing: {}", + missing.join("; ") + ))); + } + + Ok(()) +} + +/// Poll recent blocks until a new `NativeTransferred` to `wh_addr` appears. +/// Returns `(inclusion_block_hash, event)`. +async fn wait_for_native_transferred( + client: &crate::chain::client::QuantusClient, + wh_addr: &[u8; 32], + seen_leaves: &HashSet, +) -> Result<(subxt::utils::H256, wormhole_api::events::NativeTransferred)> { + use jsonrpsee::core::client::ClientT; + use subxt::utils::H256; + + let start = std::time::Instant::now(); + let timeout = Duration::from_secs(60); + let mut last_seen_best: u32 = 0; + + loop { + let best = client.get_latest_block().await?; + let best_number = client + .client() + .blocks() + .at(best) + .await + .map_err(|e| QuantusError::NetworkError(format!("blocks().at(best): {e:?}")))? + .header() + .number; + let lower = best_number.saturating_sub(INCLUSION_SCAN_BLOCKS); + let scan_from = last_seen_best.max(lower); + + for n in scan_from..=best_number { + let hash: Option = + client.rpc_client().request("chain_getBlockHash", [n]).await.map_err(|e| { + QuantusError::NetworkError(format!("chain_getBlockHash({n}): {e:?}")) + })?; + let Some(block_hash) = hash else { continue }; + let events = match client.client().events().at(block_hash).await { + Ok(e) => e, + Err(_) => continue, + }; + for ev in events.find::().flatten() { + if &ev.to.0 == wh_addr && !seen_leaves.contains(&ev.leaf_index) { + return Ok((block_hash, ev)); + } + } + } + last_seen_best = best_number.saturating_add(1); + + if start.elapsed() > timeout { + return Err(QuantusError::Generic(format!( + "Timed out waiting for NativeTransferred to 0x{} after {:.1}s", + hex::encode(wh_addr), + start.elapsed().as_secs_f64() + ))); + } + sleep(Duration::from_millis(INCLUSION_POLL_INTERVAL_MS)).await; + } +} diff --git a/src/collect_rewards_lib.rs b/src/collect_rewards_lib.rs index 16a193b..6094a09 100644 --- a/src/collect_rewards_lib.rs +++ b/src/collect_rewards_lib.rs @@ -216,6 +216,58 @@ pub struct CollectRewardsConfig { pub at_block: Option, } +/// A transfer discovered outside Subsquid (e.g. captured from `NativeTransferred` at deposit time). +/// +/// Used by [`collect_rewards_from_known`] so tests and local tooling can withdraw without an indexer. +#[derive(Debug, Clone)] +pub struct KnownTransfer { + /// Leaf index in the ZK tree + pub leaf_index: u64, + /// Transfer count (for nullifier computation) + pub transfer_count: u64, + /// Transfer amount in planck + pub amount: u128, +} + +/// Configuration for [`collect_rewards_from_known`] (no Subsquid URL). +#[derive(Debug, Clone)] +pub struct CollectKnownTransfersConfig { + /// Wormhole credential - either mnemonic or direct secret + pub credential: WormholeCredential, + /// Destination address (SS58) to receive withdrawn funds + pub destination_address: String, + /// Chain RPC node URL + pub node_url: String, + /// Path to circuit binary directory (verifier/common + private/public-batch bins). + pub bins_dir: String, + /// Optional: specific amount to withdraw (None = withdraw all) + pub amount: Option, + /// If true, only query and return info without submitting transactions + pub dry_run: bool, + /// Optional: specific block number to use for proofs (None = use latest) + pub at_block: Option, +} + +impl KnownTransfer { + fn to_transfer(&self, index: usize) -> Transfer { + Transfer { + id: format!("known-{index}"), + block_id: String::new(), + block_height: 0, + timestamp: String::new(), + extrinsic_hash: None, + from_id: String::new(), + to_id: String::new(), + amount: self.amount.to_string(), + fee: "0".to_string(), + from_hash: String::new(), + to_hash: String::new(), + leaf_index: self.leaf_index.to_string(), + transfer_count: self.transfer_count.to_string(), + } + } +} + /// Collect miner rewards by querying Subsquid, generating proofs, and submitting withdrawals. /// /// This is the main entry point for the SDK to collect rewards. It handles the entire flow: @@ -240,9 +292,6 @@ pub async fn collect_rewards( resolve_credential(&config.credential)?; progress.on_step("derive", &format!("Derived wormhole address: {}", wormhole_address)); - // Parse destination address - let destination_bytes = parse_ss58_address(&config.destination_address)?; - // Step 2: Query Subsquid for pending transfers progress.on_step("query", "Querying Subsquid for pending transfers"); @@ -259,9 +308,85 @@ pub async fn collect_rewards( let incoming_transfers: Vec<_> = transfers.into_iter().filter(|t| t.to_hash == address_hash).collect(); + collect_rewards_pipeline( + CollectPipelineArgs { + wormhole_address, + wormhole_address_bytes, + wormhole_secret_bytes, + destination_address: config.destination_address, + node_url: config.node_url, + bins_dir: config.bins_dir, + amount: config.amount, + dry_run: config.dry_run, + at_block: config.at_block, + }, + incoming_transfers, + Some(&subsquid_client), + progress, + ) + .await +} + +/// Collect rewards for transfers that are already known (no Subsquid). +/// +/// Filters spent nullifiers against on-chain `UsedNullifiers`, then proves and submits +/// the same way as [`collect_rewards`]. Intended for tests, local exercise, and callers +/// that captured deposit events themselves. +pub async fn collect_rewards_from_known( + config: CollectKnownTransfersConfig, + transfers: Vec, + progress: &P, +) -> Result { + let (wormhole_address, wormhole_address_bytes, wormhole_secret_bytes) = + resolve_credential(&config.credential)?; + progress.on_step("derive", &format!("Derived wormhole address: {}", wormhole_address)); + + let incoming_transfers: Vec = + transfers.iter().enumerate().map(|(i, t)| t.to_transfer(i)).collect(); + + collect_rewards_pipeline( + CollectPipelineArgs { + wormhole_address, + wormhole_address_bytes, + wormhole_secret_bytes, + destination_address: config.destination_address, + node_url: config.node_url, + bins_dir: config.bins_dir, + amount: config.amount, + dry_run: config.dry_run, + at_block: config.at_block, + }, + incoming_transfers, + None, + progress, + ) + .await +} + +struct CollectPipelineArgs { + wormhole_address: String, + wormhole_address_bytes: [u8; 32], + wormhole_secret_bytes: [u8; 32], + destination_address: String, + node_url: String, + bins_dir: String, + amount: Option, + dry_run: bool, + at_block: Option, +} + +/// Shared prove/submit pipeline after transfer discovery. +async fn collect_rewards_pipeline( + config: CollectPipelineArgs, + incoming_transfers: Vec, + subsquid_client: Option<&SubsquidClient>, + progress: &P, +) -> Result { + let destination_bytes = parse_ss58_address(&config.destination_address)?; + if incoming_transfers.is_empty() { return Ok(CollectRewardsResult { - wormhole_address, + wormhole_address: config.wormhole_address, destination_address: config.destination_address, total_withdrawn: 0, batches: vec![], @@ -269,25 +394,34 @@ pub async fn collect_rewards( }); } - // Step 2b: Connect to chain (needed for on-chain nullifier checks and proof submission) + // Connect to chain (needed for on-chain nullifier checks and proof submission) progress.on_step("connect", "Connecting to chain"); let quantus_client = QuantusClient::new(&config.node_url) .await .map_err(|e| CollectRewardsError::from(format!("Failed to connect to node: {}", e)))?; - // Step 2c: Filter out already-spent transfers. - // Subsquid is a best-effort pre-filter; on-chain UsedNullifiers is authoritative. + // Filter out already-spent transfers. + // Subsquid (when provided) is a best-effort pre-filter; on-chain is authoritative. progress.on_step("nullifiers", "Checking for already-spent nullifiers (on-chain)"); let incoming_count = incoming_transfers.len(); - let unspent_transfers = filter_unspent_transfers( - &incoming_transfers, - &wormhole_secret_bytes, - &subsquid_client, - &quantus_client, - ) - .await?; + let unspent_transfers = if let Some(subsquid_client) = subsquid_client { + filter_unspent_transfers( + &incoming_transfers, + &config.wormhole_secret_bytes, + subsquid_client, + &quantus_client, + ) + .await? + } else { + filter_unspent_transfers_onchain( + &incoming_transfers, + &config.wormhole_secret_bytes, + &quantus_client, + ) + .await? + }; let spent_filtered = incoming_count.saturating_sub(unspent_transfers.len()); if spent_filtered > 0 { @@ -305,7 +439,7 @@ pub async fn collect_rewards( progress .on_step("complete", "All transfers have already been withdrawn (nullifiers spent)"); return Ok(CollectRewardsResult { - wormhole_address, + wormhole_address: config.wormhole_address, destination_address: config.destination_address, total_withdrawn: 0, batches: vec![], @@ -350,7 +484,7 @@ pub async fn collect_rewards( if config.dry_run { return Ok(CollectRewardsResult { - wormhole_address, + wormhole_address: config.wormhole_address, destination_address: config.destination_address, total_withdrawn: 0, batches: vec![], @@ -421,7 +555,9 @@ pub async fn collect_rewards( for (i, transfer) in selected_transfers.iter().enumerate() { // Re-check on-chain immediately before proving — nullifiers can be spent // while earlier proofs in this run are still being generated. - if is_nullifier_spent_onchain(&quantus_client, &wormhole_secret_bytes, transfer).await? { + if is_nullifier_spent_onchain(&quantus_client, &config.wormhole_secret_bytes, transfer) + .await? + { progress.on_step( "nullifiers", &format!( @@ -476,18 +612,18 @@ pub async fn collect_rewards( let digest = header.digest.encode(); let block_number = header.number; - if leaf_to_account != wormhole_address_bytes { + if leaf_to_account != config.wormhole_address_bytes { return Err(CollectRewardsError::from(format!( "Leaf to_account mismatch: expected 0x{}, got 0x{}", - hex::encode(wormhole_address_bytes), + hex::encode(config.wormhole_address_bytes), hex::encode(leaf_to_account) ))); } let input = wormhole_lib::ProofGenerationInput { - secret: wormhole_secret_bytes, + secret: config.wormhole_secret_bytes, transfer_count, - wormhole_address: wormhole_address_bytes, + wormhole_address: config.wormhole_address_bytes, input_amount, block_hash: proof_block_hash.0, block_number, @@ -522,7 +658,7 @@ pub async fn collect_rewards( "No unspent transfers remained after on-chain nullifier checks", ); return Ok(CollectRewardsResult { - wormhole_address, + wormhole_address: config.wormhole_address, destination_address: config.destination_address, total_withdrawn: 0, batches: vec![], @@ -582,7 +718,7 @@ pub async fn collect_rewards( } Ok(CollectRewardsResult { - wormhole_address, + wormhole_address: config.wormhole_address, destination_address: config.destination_address, total_withdrawn, batches: batch_results, @@ -934,21 +1070,8 @@ async fn submit_and_get_events( CollectRewardsError::from(format!("Failed to convert nullifier {} to bytes", i)) })?; - // Query UsedNullifiers storage - let storage_key = quantus_node::api::storage().wormhole().used_nullifiers(nullifier_bytes); - let is_used = quantus_client - .client() - .storage() - .at_latest() - .await - .map_err(|e| CollectRewardsError::from(format!("Failed to get storage: {}", e)))? - .fetch(&storage_key) - .await - .map_err(|e| { - CollectRewardsError::from(format!("Failed to query nullifier {}: {}", i, e)) - })?; - - if is_used.is_some() { + // Query UsedNullifiers at best (not finalized) — PoW finality can lag inclusion. + if is_nullifier_bytes_spent_onchain(quantus_client, &nullifier_bytes).await? { return Err(CollectRewardsError::from(format!( "Pre-validation failed: NullifierAlreadyUsed - Nullifier {} (0x{}) has already been spent", i, @@ -1073,6 +1196,31 @@ fn is_spent_nullifier_error(err: &CollectRewardsError) -> bool { err.message.contains("NullifierAlreadyUsed") } +/// Whether `nullifier` is marked spent in on-chain `UsedNullifiers`. +/// +/// Reads at the **best** block (via RPC), not Subxt's `at_latest()` which is +/// finalized — on PoW, finality can lag best inclusion, so post-collect filters +/// would otherwise miss freshly spent nullifiers. +async fn is_nullifier_bytes_spent_onchain( + quantus_client: &QuantusClient, + nullifier: &[u8; 32], +) -> Result { + let best = quantus_client.get_latest_block().await.map_err(|e| { + CollectRewardsError::from(format!("Failed to get best block for UsedNullifiers: {}", e)) + })?; + let storage_key = quantus_node::api::storage().wormhole().used_nullifiers(*nullifier); + let is_used = quantus_client + .client() + .storage() + .at(best) + .fetch(&storage_key) + .await + .map_err(|e| { + CollectRewardsError::from(format!("Failed to query UsedNullifiers: {}", e)) + })?; + Ok(matches!(is_used, Some(true))) +} + /// Compute the nullifier for a transfer and check on-chain `UsedNullifiers`. async fn is_nullifier_spent_onchain( quantus_client: &QuantusClient, @@ -1087,23 +1235,7 @@ async fn is_nullifier_spent_onchain( CollectRewardsError::from(format!("Failed to compute nullifier: {}", e.message)) })?; - let storage_key = quantus_node::api::storage().wormhole().used_nullifiers(nullifier); - let is_used = quantus_client - .client() - .storage() - .at_latest() - .await - .map_err(|e| CollectRewardsError::from(format!("Failed to get storage: {}", e)))? - .fetch(&storage_key) - .await - .map_err(|e| { - CollectRewardsError::from(format!( - "Failed to query nullifier for transfer_count {}: {}", - transfer_count, e - )) - })?; - - Ok(is_used.is_some()) + is_nullifier_bytes_spent_onchain(quantus_client, &nullifier).await } /// Filter out transfers whose nullifiers have already been spent (on-chain check). diff --git a/src/lib.rs b/src/lib.rs index e71fd2d..9bb53ec 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -67,9 +67,10 @@ pub use cli::wormhole::{ // Re-export collect rewards library for SDK usage pub use collect_rewards_lib::{ - collect_rewards, query_pending_transfers, query_pending_transfers_for_address, - CollectRewardsConfig, CollectRewardsError, CollectRewardsResult, NoOpProgress, PendingTransfer, - ProgressCallback, QueryPendingTransfersResult, WithdrawalBatch, + collect_rewards, collect_rewards_from_known, query_pending_transfers, + query_pending_transfers_for_address, CollectKnownTransfersConfig, CollectRewardsConfig, + CollectRewardsError, CollectRewardsResult, KnownTransfer, NoOpProgress, PendingTransfer, + ProgressCallback, QueryPendingTransfersResult, WithdrawalBatch, WormholeCredential, }; /// Library version From 0c4beea92b96fe9c4f8477f3584ce41f6aea298d Mon Sep 17 00:00:00 2001 From: illuzen Date: Wed, 29 Jul 2026 19:36:38 +0800 Subject: [PATCH 3/6] fix(exercise): always re-scan tip for NativeTransferred A watermark advanced past best_number could skip the inclusion block when the deposit landed in that tip after an earlier empty poll. Co-authored-by: Cursor --- src/cli/exercise/scenarios/wormhole.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/cli/exercise/scenarios/wormhole.rs b/src/cli/exercise/scenarios/wormhole.rs index 30cc2bc..98102f4 100644 --- a/src/cli/exercise/scenarios/wormhole.rs +++ b/src/cli/exercise/scenarios/wormhole.rs @@ -306,7 +306,6 @@ async fn wait_for_native_transferred( let start = std::time::Instant::now(); let timeout = Duration::from_secs(60); - let mut last_seen_best: u32 = 0; loop { let best = client.get_latest_block().await?; @@ -318,8 +317,10 @@ async fn wait_for_native_transferred( .map_err(|e| QuantusError::NetworkError(format!("blocks().at(best): {e:?}")))? .header() .number; - let lower = best_number.saturating_sub(INCLUSION_SCAN_BLOCKS); - let scan_from = last_seen_best.max(lower); + // Always re-scan a trailing window (including the current tip). Advancing a + // watermark past `best_number` can skip the inclusion block when the transfer + // lands in that block after an earlier empty pass. + let scan_from = best_number.saturating_sub(INCLUSION_SCAN_BLOCKS); for n in scan_from..=best_number { let hash: Option = @@ -337,7 +338,6 @@ async fn wait_for_native_transferred( } } } - last_seen_best = best_number.saturating_add(1); if start.elapsed() > timeout { return Err(QuantusError::Generic(format!( From d68f39b48c3f417487b18eab7a926dcba596a8d7 Mon Sep 17 00:00:00 2001 From: illuzen Date: Wed, 29 Jul 2026 19:42:57 +0800 Subject: [PATCH 4/6] fix(collect-rewards): surface Subsquid nullifier pre-filter errors Keep the on-chain fallback, but log and report via progress when the indexer pre-filter fails so failures are not silently discarded. Co-authored-by: Cursor --- src/collect_rewards_lib.rs | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/src/collect_rewards_lib.rs b/src/collect_rewards_lib.rs index 6094a09..c0bdada 100644 --- a/src/collect_rewards_lib.rs +++ b/src/collect_rewards_lib.rs @@ -412,6 +412,7 @@ async fn collect_rewards_pipeline( &config.wormhole_secret_bytes, subsquid_client, &quantus_client, + progress, ) .await? } else { @@ -1314,11 +1315,12 @@ async fn filter_unspent_transfers_subsquid( /// Uses Subsquid as a best-effort pre-filter when available, then always verifies /// remaining candidates against on-chain `UsedNullifiers`. On-chain is authoritative — /// a successful but stale Subsquid response must not be trusted alone. -async fn filter_unspent_transfers( +async fn filter_unspent_transfers( transfers: &[Transfer], secret_bytes: &[u8; 32], subsquid_client: &SubsquidClient, quantus_client: &QuantusClient, + progress: &P, ) -> Result> { if transfers.is_empty() { return Ok(vec![]); @@ -1334,7 +1336,15 @@ async fn filter_unspent_transfers( .await { Ok(filtered) => filtered, - Err(_) => transfers.to_vec(), + Err(e) => { + let msg = format!( + "Subsquid nullifier pre-filter failed ({}); falling back to full on-chain check", + e.message + ); + crate::log_error!("{}", msg); + progress.on_step("nullifiers", &msg); + transfers.to_vec() + }, }; filter_unspent_transfers_onchain(&candidates, secret_bytes, quantus_client).await From 065b6a8609136a70459b77b5f18873bfa5dee226 Mon Sep 17 00:00:00 2001 From: illuzen Date: Wed, 29 Jul 2026 19:44:15 +0800 Subject: [PATCH 5/6] fix(exercise): log block events fetch failures while polling deposits Keep skipping unreadable blocks, but emit log_verbose details so deposit inclusion timeouts are easier to diagnose with -v. Co-authored-by: Cursor --- src/cli/exercise/scenarios/wormhole.rs | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/src/cli/exercise/scenarios/wormhole.rs b/src/cli/exercise/scenarios/wormhole.rs index 98102f4..672bd16 100644 --- a/src/cli/exercise/scenarios/wormhole.rs +++ b/src/cli/exercise/scenarios/wormhole.rs @@ -19,7 +19,7 @@ use crate::{ WormholeCredential, }, error::{QuantusError, Result}, - exercise_step, wormhole_lib, + exercise_step, log_verbose, wormhole_lib, }; use rand::RngCore; use std::{collections::HashSet, time::Duration}; @@ -330,7 +330,12 @@ async fn wait_for_native_transferred( let Some(block_hash) = hash else { continue }; let events = match client.client().events().at(block_hash).await { Ok(e) => e, - Err(_) => continue, + Err(e) => { + log_verbose!( + "wait_for_native_transferred: events().at(#{n} {block_hash:?}) failed: {e:?}" + ); + continue; + }, }; for ev in events.find::().flatten() { if &ev.to.0 == wh_addr && !seen_leaves.contains(&ev.leaf_index) { From 5f08e98e818ee5eeef99b32a8afed6d69bec8a2f Mon Sep 17 00:00:00 2001 From: illuzen Date: Wed, 29 Jul 2026 19:45:30 +0800 Subject: [PATCH 6/6] fmt --- src/cli/exercise/scenarios/wormhole.rs | 6 +-- src/collect_rewards_lib.rs | 63 ++++++++++---------------- 2 files changed, 27 insertions(+), 42 deletions(-) diff --git a/src/cli/exercise/scenarios/wormhole.rs b/src/cli/exercise/scenarios/wormhole.rs index 672bd16..f4b7fb6 100644 --- a/src/cli/exercise/scenarios/wormhole.rs +++ b/src/cli/exercise/scenarios/wormhole.rs @@ -80,8 +80,7 @@ async fn multiround_public(ctx: &mut ExerciseCtx) -> Result { async fn collect_rewards_happy_path(ctx: &mut ExerciseCtx) -> Result { let secret = random_secret(ctx); let mut seen_leaves = HashSet::new(); - let deposits = - fund_wormhole_deposits(ctx, &secret, 2, ctx.unit, &mut seen_leaves).await?; + let deposits = fund_wormhole_deposits(ctx, &secret, 2, ctx.unit, &mut seen_leaves).await?; let dest = ctx.eph[0].to_account_id_ss58check(); let before = ctx.free_balance(&dest).await?; @@ -121,8 +120,7 @@ async fn collect_rewards_spent_filter(ctx: &mut ExerciseCtx) -> Result { let dest = ctx.eph[1].to_account_id_ss58check(); let mut seen_leaves = HashSet::new(); - let initial = - fund_wormhole_deposits(ctx, &secret, 2, ctx.unit, &mut seen_leaves).await?; + let initial = fund_wormhole_deposits(ctx, &secret, 2, ctx.unit, &mut seen_leaves).await?; let first = run_collect_known(ctx, &secret, &dest, initial.clone(), None) .await .map_err(|e| QuantusError::Generic(format!("initial collect failed: {e}")))?; diff --git a/src/collect_rewards_lib.rs b/src/collect_rewards_lib.rs index c0bdada..e8f3e42 100644 --- a/src/collect_rewards_lib.rs +++ b/src/collect_rewards_lib.rs @@ -218,7 +218,8 @@ pub struct CollectRewardsConfig { /// A transfer discovered outside Subsquid (e.g. captured from `NativeTransferred` at deposit time). /// -/// Used by [`collect_rewards_from_known`] so tests and local tooling can withdraw without an indexer. +/// Used by [`collect_rewards_from_known`] so tests and local tooling can withdraw without an +/// indexer. #[derive(Debug, Clone)] pub struct KnownTransfer { /// Leaf index in the ZK tree @@ -494,9 +495,10 @@ async fn collect_rewards_pipeline( } // Refuse to generate proofs against an unsupported runtime - let (spec_version, transaction_version) = quantus_client.get_runtime_version().await.map_err( - |e| CollectRewardsError::from(format!("Failed to get runtime version: {}", e)), - )?; + let (spec_version, transaction_version) = quantus_client + .get_runtime_version() + .await + .map_err(|e| CollectRewardsError::from(format!("Failed to get runtime version: {}", e)))?; if !crate::config::is_runtime_compatible(spec_version, transaction_version) { let supported = crate::config::COMPATIBLE_RUNTIMES .iter() @@ -654,10 +656,8 @@ async fn collect_rewards_pipeline( } if proof_bytes_list.is_empty() { - progress.on_step( - "complete", - "No unspent transfers remained after on-chain nullifier checks", - ); + progress + .on_step("complete", "No unspent transfers remained after on-chain nullifier checks"); return Ok(CollectRewardsResult { wormhole_address: config.wormhole_address, destination_address: config.destination_address, @@ -1161,10 +1161,8 @@ async fn submit_and_get_events( event.as_event::() { let metadata = quantus_client.client().metadata(); - dispatch_error_msg = Some(crate::cli::common::format_dispatch_error( - &dispatch_error, - &metadata, - )); + dispatch_error_msg = + Some(crate::cli::common::format_dispatch_error(&dispatch_error, &metadata)); } if let Ok(Some(_)) = event.as_event::() { found_proof_verified = true; @@ -1179,10 +1177,7 @@ async fn submit_and_get_events( if !found_proof_verified { if let Some(msg) = dispatch_error_msg { - return Err(CollectRewardsError::from(format!( - "Proof verification failed: {}", - msg - ))); + return Err(CollectRewardsError::from(format!("Proof verification failed: {}", msg))); } return Err(CollectRewardsError::from( "Proof verification failed - no ProofVerified event".to_string(), @@ -1216,9 +1211,7 @@ async fn is_nullifier_bytes_spent_onchain( .at(best) .fetch(&storage_key) .await - .map_err(|e| { - CollectRewardsError::from(format!("Failed to query UsedNullifiers: {}", e)) - })?; + .map_err(|e| CollectRewardsError::from(format!("Failed to query UsedNullifiers: {}", e)))?; Ok(matches!(is_used, Some(true))) } @@ -1231,10 +1224,9 @@ async fn is_nullifier_spent_onchain( let ctx = format!("transfer {}", transfer.id); let transfer_count = parse_transfer_count(&transfer.transfer_count, &ctx)?; - let nullifier = - wormhole_lib::compute_nullifier(secret_bytes, transfer_count).map_err(|e| { - CollectRewardsError::from(format!("Failed to compute nullifier: {}", e.message)) - })?; + let nullifier = wormhole_lib::compute_nullifier(secret_bytes, transfer_count).map_err(|e| { + CollectRewardsError::from(format!("Failed to compute nullifier: {}", e.message)) + })?; is_nullifier_bytes_spent_onchain(quantus_client, &nullifier).await } @@ -1328,24 +1320,19 @@ async fn filter_unspent_transfers( // Optional fast path: shrink the candidate set via Subsquid when it works. // On failure or staleness we still run the on-chain check below. - let candidates = match filter_unspent_transfers_subsquid( - transfers, - secret_bytes, - subsquid_client, - ) - .await - { - Ok(filtered) => filtered, - Err(e) => { - let msg = format!( + let candidates = + match filter_unspent_transfers_subsquid(transfers, secret_bytes, subsquid_client).await { + Ok(filtered) => filtered, + Err(e) => { + let msg = format!( "Subsquid nullifier pre-filter failed ({}); falling back to full on-chain check", e.message ); - crate::log_error!("{}", msg); - progress.on_step("nullifiers", &msg); - transfers.to_vec() - }, - }; + crate::log_error!("{}", msg); + progress.on_step("nullifiers", &msg); + transfers.to_vec() + }, + }; filter_unspent_transfers_onchain(&candidates, secret_bytes, quantus_client).await }