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/cli/exercise/scenarios/wormhole.rs b/src/cli/exercise/scenarios/wormhole.rs index 2fcf482..f4b7fb6 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, log_verbose, 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,297 @@ 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); + + 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; + // 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 = + 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(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) { + return Ok((block_hash, ev)); + } + } + } + + 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 1dcba28..e8f3e42 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. @@ -215,14 +216,68 @@ 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: /// 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 @@ -238,9 +293,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"); @@ -257,9 +309,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![], @@ -267,30 +395,53 @@ pub async fn collect_rewards( }); } - // Step 2b: Connect to chain (needed for nullifier fallback 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 by checking nullifiers - // Tries Subsquid first, falls back to on-chain checking if Subsquid fails - progress.on_step("nullifiers", "Checking for already-spent nullifiers"); + // 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 = if let Some(subsquid_client) = subsquid_client { + filter_unspent_transfers( + &incoming_transfers, + &config.wormhole_secret_bytes, + subsquid_client, + &quantus_client, + progress, + ) + .await? + } else { + filter_unspent_transfers_onchain( + &incoming_transfers, + &config.wormhole_secret_bytes, + &quantus_client, + ) + .await? + }; - let unspent_transfers = filter_unspent_transfers_with_fallback( - &incoming_transfers, - &wormhole_secret_bytes, - &subsquid_client, - &quantus_client, - ) - .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)"); return Ok(CollectRewardsResult { - wormhole_address, + wormhole_address: config.wormhole_address, destination_address: config.destination_address, total_withdrawn: 0, batches: vec![], @@ -335,7 +486,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![], @@ -343,6 +494,23 @@ 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 +549,29 @@ 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, &config.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)) @@ -432,18 +615,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, @@ -469,6 +652,19 @@ 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: config.wormhole_address, + destination_address: config.destination_address, + total_withdrawn: 0, + batches: vec![], + transfers_processed: 0, + }); } // Step 5: Aggregate and submit in batches @@ -486,17 +682,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); @@ -509,11 +719,11 @@ pub async fn collect_rewards( } Ok(CollectRewardsResult { - wormhole_address, + wormhole_address: config.wormhole_address, destination_address: config.destination_address, total_withdrawn, batches: batch_results, - transfers_processed: selected_transfers.len(), + transfers_processed, }) } @@ -861,21 +1071,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, @@ -948,8 +1145,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 +1157,13 @@ 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 +1176,9 @@ 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 +1187,55 @@ 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") +} + +/// 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, + 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)) + })?; + + is_nullifier_bytes_spent_onchain(quantus_client, &nullifier).await +} + /// 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 +1246,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 +1255,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 +1269,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 +1287,54 @@ 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, + progress: &P, +) -> 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(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 } #[cfg(test)] @@ -1099,6 +1348,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) 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