From 609043288aa5a6c75b225e0891aade764b143b5c Mon Sep 17 00:00:00 2001 From: Nikolaus Heger Date: Tue, 22 Sep 2026 17:27:16 +0800 Subject: [PATCH] Add airdrop payout commands: pull, review, pay, mark-paid Operator side of the airdrop. `pull` turns the claim server's unpaid recorded claims into a manifest (one transfer per payout account), `review` re-checks it against the server and the chain and approves it, `pay` submits it as a single utility.batch_all from a hot or cold wallet and waits for finalization, and `mark-paid` marks every reward in it paid on the server. The manifest carries the state between steps. A new pull is refused while another manifest in the directory is not yet marked, pay re-verifies the server right before signing and checks the approval hash, and `pay --recover` resolves an interrupted payment by scanning the finalized blocks the batch could have landed in for the exact signer and call bytes. Also default the claim server to https://airdrop-claim.quantus.com, share one JSON GET between /snapshot and /unpaid, and add QuantusClient::get_block_hash in place of the inline chain_getBlockHash calls. --- README.md | 38 ++ src/chain/client.rs | 11 + src/cli/airdrop.rs | 109 ++++- src/cli/airdrop/payout.rs | 843 +++++++++++++++++++++++++++++++++++++ src/cli/block.rs | 10 +- src/cli/mod.rs | 5 +- src/collect_rewards_lib.rs | 16 +- 7 files changed, 989 insertions(+), 43 deletions(-) create mode 100644 src/cli/airdrop/payout.rs diff --git a/README.md b/README.md index 65b6c3d..d0a4bd3 100644 --- a/README.md +++ b/README.md @@ -439,6 +439,44 @@ For each match the CLI builds the appropriate proof: The command exits non-zero if any claim fails, and prints a summary of recorded, skipped, and failed claims. +#### Paying out claims (operators) + +Four commands turn recorded claims into one on-chain payment each. Every step +is separate so it can be checked by hand, and a manifest file carries the state +between them (`pulled` → `approved` → `paying` → `paid` → `marked`). + +```bash +# 1. Pull unpaid recorded claims into a manifest (one transfer per payout account) +quantus airdrop pull --out payout.json + +# 2. Show it, re-check it against the server and the chain, approve it +quantus airdrop review --manifest payout.json --approve + +# 3. Pay it as ONE utility.batch_all — every transfer lands or none does. +# A cold wallet signs over QR; the command waits for finalization. +quantus airdrop pay --manifest payout.json --from treasury_cold + +# 4. Mark every reward in the manifest paid on the claim server +quantus airdrop mark-paid --manifest payout.json --admin-token-file ./admin-token +``` + +- `pull` skips unclaimed rows, aggregates rewards by `claim_account`, and refuses + to run while another manifest in the same directory is not yet `marked`, so + no reward can be pulled into two payments. `--limit N` keeps only the first + N destinations when a batch must be split (a cold-wallet QR payload holds + about 180 transfers). +- `review` fails if any reward was paid, re-claimed or changed on the server + since the pull, reports the encoded batch size, and `--approve` pins the + transfers with a hash that `pay` verifies. +- `pay` re-checks the server right before signing, records the signer, nonce + and anchor block in the manifest, then submits. If it is interrupted, run it + again with `--recover`: it scans the finalized blocks the batch could have + landed in and either records the payment or, once the transaction can no + longer be included, returns the manifest to `approved`. +- `mark-paid` is idempotent and re-runnable; a `409 already marked paid` counts + as done. The admin token comes from `--admin-token-file` (owner-only file) or + `AIRDROP_ADMIN_TOKEN`, never from argv. + --- ### Developer Tools diff --git a/src/chain/client.rs b/src/chain/client.rs index d671f31..adaa123 100644 --- a/src/chain/client.rs +++ b/src/chain/client.rs @@ -238,6 +238,17 @@ impl QuantusClient { Ok(latest_hash) } + /// Hash of the block at `number`; an error if the chain has no such block. + pub async fn get_block_hash(&self, number: u64) -> crate::error::Result { + let hash: Option = + self.rpc_client.request("chain_getBlockHash", [number]).await.map_err(|e| { + QuantusError::NetworkError(format!( + "Failed to get block hash for block {number}: {e:?}" + )) + })?; + hash.ok_or_else(|| QuantusError::NetworkError(format!("Block {number} not found"))) + } + /// Interpret a System::Account nonce lookup without collapsing absence into a silent zero. /// /// Returns `(nonce, account_exists)`. Missing accounts use nonce `0` (correct for the first diff --git a/src/cli/airdrop.rs b/src/cli/airdrop.rs index 5f62a9d..b4cb440 100644 --- a/src/cli/airdrop.rs +++ b/src/cli/airdrop.rs @@ -21,9 +21,11 @@ use serde::{Deserialize, Serialize}; use sp_core::crypto::{AccountId32, Ss58Codec}; use std::{collections::HashMap, path::PathBuf, time::Duration}; +mod payout; + const CLAIM_CONTEXT: &[u8] = b"qp-airdrop-claim-v1"; const CLAIM_TTL_SECS: i64 = 10 * 60; -const DEFAULT_SERVER: &str = "http://127.0.0.1:8080"; +pub(super) const DEFAULT_SERVER: &str = "https://airdrop-claim.quantus.com"; const HD_WORMHOLE_INDEXES: std::ops::RangeInclusive = 0..=16; const CLAIMABLE_WORMHOLE_SCHEME: &str = "wormhole-rate8-compact"; /// BIP44 coin type for Dilithium keys (the wormhole coin type is 189189189'). @@ -130,9 +132,68 @@ pub enum AirdropCommands { #[arg(long)] dry_run: bool, }, + /// Operator: pull unpaid recorded claims into a payout manifest + Pull { + /// Claim server base URL + #[arg(long, default_value = DEFAULT_SERVER)] + server: String, + + /// Manifest file to write (default: airdrop-payout-.json) + #[arg(long)] + out: Option, + + /// Only include the first N payout destinations (one batch per manifest) + #[arg(long)] + limit: Option, + }, + + /// Operator: show a manifest, check it against the server and chain, optionally approve it + Review { + /// Payout manifest written by `airdrop pull` + #[arg(long)] + manifest: PathBuf, + + /// Approve the manifest for payment + #[arg(long)] + approve: bool, + }, + + /// Operator: pay an approved manifest as one atomic batch (all transfers or none) + Pay { + /// Approved payout manifest + #[arg(long)] + manifest: PathBuf, + + /// Wallet that pays (a cold wallet signs over QR) + #[arg(long, short)] + from: String, + + /// Password for the wallet (unsupported on argv; use --password-file or prompt) + #[arg(short, long, hide = true)] + password: Option, + + /// Read password from file (for scripting) + #[arg(long)] + password_file: Option, + + /// After an interrupted payment: find out whether the batch landed and update the manifest + #[arg(long)] + recover: bool, + }, + + /// Operator: mark every reward in a paid manifest as paid on the claim server + MarkPaid { + /// Paid payout manifest + #[arg(long)] + manifest: PathBuf, + + /// File holding the server admin token (chmod 600); otherwise AIRDROP_ADMIN_TOKEN + #[arg(long)] + admin_token_file: Option, + }, } -pub async fn handle_airdrop_command(command: AirdropCommands) -> Result<()> { +pub async fn handle_airdrop_command(command: AirdropCommands, node_url: &str) -> Result<()> { match command { AirdropCommands::Check { server, @@ -180,6 +241,14 @@ pub async fn handle_airdrop_command(command: AirdropCommands) -> Result<()> { dry_run, ) .await, + AirdropCommands::Pull { server, out, limit } => + payout::handle_pull(server, out, limit).await, + AirdropCommands::Review { manifest, approve } => + payout::handle_review(node_url, manifest, approve).await, + AirdropCommands::Pay { manifest, from, password, password_file, recover } => + payout::handle_pay(node_url, manifest, from, password, password_file, recover).await, + AirdropCommands::MarkPaid { manifest, admin_token_file } => + payout::handle_mark_paid(manifest, admin_token_file).await, } } @@ -1032,7 +1101,20 @@ struct ErrorBody { } async fn fetch_snapshot(server: &str) -> Result { - let url = format!("{}/snapshot", server.trim_end_matches('/')); + let wire: SnapshotWire = get_json(server, "snapshot").await?; + let mut by_account = HashMap::new(); + for row in &wire.rows { + let account = parse_account_id(&row.account).or_else(|_| parse_account_id(&row.address))?; + by_account.insert(account, row.clone()); + } + Ok(SnapshotFile { version: wire.version, sha256: wire.sha256, rows: wire.rows, by_account }) +} + +pub(super) async fn get_json( + server: &str, + path: &str, +) -> Result { + let url = format!("{}/{path}", server.trim_end_matches('/')); log_verbose!("GET {url}"); let response = http_client()?.get(&url).send().await.map_err(http_err)?; let status = response.status(); @@ -1040,17 +1122,10 @@ async fn fetch_snapshot(server: &str) -> Result { if !status.is_success() { return Err(QuantusError::Generic(format_server_error(status, &text))); } - let wire: SnapshotWire = serde_json::from_str(&text) - .map_err(|e| QuantusError::Generic(format!("snapshot JSON: {e}")))?; - let mut by_account = HashMap::new(); - for row in &wire.rows { - let account = parse_account_id(&row.account).or_else(|_| parse_account_id(&row.address))?; - by_account.insert(account, row.clone()); - } - Ok(SnapshotFile { version: wire.version, sha256: wire.sha256, rows: wire.rows, by_account }) + serde_json::from_str(&text).map_err(|e| QuantusError::Generic(format!("{path} JSON: {e}"))) } -fn parse_account_id(s: &str) -> Result<[u8; 32]> { +pub(super) fn parse_account_id(s: &str) -> Result<[u8; 32]> { let s = s.trim(); if s.starts_with("qz") { let (account, _) = AccountId32::from_ss58check_with_version(s) @@ -1065,18 +1140,18 @@ fn parse_account_id(s: &str) -> Result<[u8; 32]> { }) } -fn http_client() -> Result { +pub(super) fn http_client() -> Result { reqwest::Client::builder() .timeout(Duration::from_secs(120)) .build() .map_err(|e| QuantusError::Generic(format!("HTTP client: {e}"))) } -fn http_err(e: reqwest::Error) -> QuantusError { +pub(super) fn http_err(e: reqwest::Error) -> QuantusError { QuantusError::NetworkError(e.to_string()) } -fn format_server_error(status: reqwest::StatusCode, body: &str) -> String { +pub(super) fn format_server_error(status: reqwest::StatusCode, body: &str) -> String { if let Ok(err) = serde_json::from_str::(body) { format!("server {status}: {}", err.error) } else { @@ -1084,11 +1159,11 @@ fn format_server_error(status: reqwest::StatusCode, body: &str) -> String { } } -fn format_hundredths(amount: u64) -> String { +pub(super) fn format_hundredths(amount: u64) -> String { format!("{}.{:02}", amount / 100, amount % 100) } -fn now_unix() -> Result { +pub(super) fn now_unix() -> Result { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map(|d| d.as_secs() as i64) diff --git a/src/cli/airdrop/payout.rs b/src/cli/airdrop/payout.rs new file mode 100644 index 0000000..90b0f32 --- /dev/null +++ b/src/cli/airdrop/payout.rs @@ -0,0 +1,843 @@ +//! Operator side of the airdrop: pull unpaid recorded claims into a manifest, +//! review and approve it, pay it as one atomic `utility.batch_all`, then mark +//! every reward paid on the claim server. Each step is its own command and the +//! manifest file carries the state between them. + +use super::{ + format_hundredths, format_server_error, get_json, http_client, http_err, now_unix, + parse_account_id, +}; +use crate::{ + chain::client::QuantusClient, + cli::{ + cold_signing::{MAX_COLD_PAYLOAD, MORTALITY_BLOCKS}, + common::{self, ExecutionMode}, + send, wormhole, + }, + error::{QuantusError, Result}, + log_error, log_print, log_success, log_verbose, + wallet::{self, password, WalletSigner}, +}; +use colored::Colorize; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use sp_core::crypto::{AccountId32, Ss58Codec}; +use std::{ + collections::{BTreeMap, HashMap}, + path::{Path, PathBuf}, +}; +use subxt::tx::Payload; + +const MANIFEST_VERSION: u32 = 1; +const ADMIN_TOKEN_ENV: &str = "AIRDROP_ADMIN_TOKEN"; +const RECORDED: &str = "recorded"; +/// Transaction extensions added to the call bytes in a cold-wallet payload. +const COLD_PAYLOAD_OVERHEAD: usize = 128; +/// The submit path anchors its mortality a few blocks after `Payment::anchor_block` +/// is recorded; recovery only declares a batch expired past this extra margin. +const ANCHOR_SLACK_BLOCKS: u64 = 16; + +#[derive(Serialize, Deserialize, Clone, Copy, PartialEq, Eq, Debug)] +#[serde(rename_all = "snake_case")] +enum State { + Pulled, + Approved, + Paying, + Paid, + Marked, +} + +impl State { + fn label(self) -> &'static str { + match self { + Self::Pulled => "pulled", + Self::Approved => "approved", + Self::Paying => "paying", + Self::Paid => "paid", + Self::Marked => "marked", + } + } + + fn next_step(self) -> &'static str { + match self { + Self::Pulled => "approve it with `quantus airdrop review --approve`", + Self::Approved => "pay it with `quantus airdrop pay`", + Self::Paying => "run `quantus airdrop pay --recover` to learn whether the batch landed", + Self::Paid => "run `quantus airdrop mark-paid`", + Self::Marked => "it is finished", + } + } +} + +#[derive(Serialize, Deserialize, Clone, PartialEq, Eq, Debug)] +struct Reward { + address: String, + amount_hundredths: u64, +} + +/// One `transfer_allow_death` in the batch: every reward claimed to the same account. +#[derive(Serialize, Deserialize, Clone, PartialEq, Eq, Debug)] +struct Transfer { + to: String, + amount_hundredths: u64, + rewards: Vec, +} + +#[derive(Serialize, Deserialize, Clone, Debug)] +struct Payment { + signer: String, + nonce: u64, + anchor_block: u64, + #[serde(default, skip_serializing_if = "Option::is_none")] + tx_hash: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + block_hash: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + paid_at: Option, +} + +#[derive(Serialize, Deserialize, Debug)] +struct Manifest { + version: u32, + server: String, + pulled_at: i64, + state: State, + total_hundredths: u64, + transfers: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + approved_sha256: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + payment: Option, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + marked: Vec, +} + +impl Manifest { + fn load(path: &Path) -> Result { + let text = std::fs::read_to_string(path) + .map_err(|e| QuantusError::Generic(format!("read manifest {}: {e}", path.display())))?; + let manifest: Self = serde_json::from_str(&text).map_err(|e| { + QuantusError::Generic(format!("parse manifest {}: {e}", path.display())) + })?; + if manifest.version != MANIFEST_VERSION { + return Err(QuantusError::Generic(format!( + "manifest {} is version {}, this CLI writes version {MANIFEST_VERSION}", + path.display(), + manifest.version + ))); + } + Ok(manifest) + } + + /// Temp write + rename: a crash never leaves a truncated manifest behind. + fn save(&self, path: &Path) -> Result<()> { + let json = serde_json::to_string_pretty(self) + .map_err(|e| QuantusError::Generic(format!("serialize manifest: {e}")))?; + let tmp = path.with_extension("json.tmp"); + std::fs::write(&tmp, json)?; + std::fs::rename(&tmp, path)?; + Ok(()) + } + + fn addresses(&self) -> impl Iterator { + self.transfers.iter().flat_map(|t| t.rewards.iter().map(|r| r.address.as_str())) + } + + fn ensure_state(&self, expected: State) -> Result<()> { + if self.state == expected { + return Ok(()); + } + Err(QuantusError::Generic(format!( + "manifest is '{}', not '{}'; {}", + self.state.label(), + expected.label(), + self.state.next_step() + ))) + } + + fn ensure_approved(&self) -> Result<()> { + self.ensure_state(State::Approved)?; + if self.approved_sha256.as_deref() == Some(transfers_sha256(&self.transfers)?.as_str()) { + return Ok(()); + } + Err(QuantusError::Generic( + "manifest transfers changed after approval; re-run `quantus airdrop review --approve`" + .into(), + )) + } + + fn payment(&self) -> Result<&Payment> { + self.payment.as_ref().ok_or_else(|| { + QuantusError::Generic(format!( + "manifest is '{}' but has no payment record", + self.state.label() + )) + }) + } +} + +#[derive(Deserialize)] +struct UnpaidResponse { + rows: Vec, +} + +#[derive(Clone, Deserialize)] +struct UnpaidRow { + address: String, + claim_account: Option, + amount_hundredths: u64, + status: String, +} + +async fn fetch_unpaid(server: &str) -> Result> { + let wire: UnpaidResponse = get_json(server, "unpaid").await?; + Ok(wire.rows) +} + +/// Group recorded claims by payout account, deterministically ordered. +/// Unclaimed rows have no destination and are skipped. +fn aggregate(rows: &[UnpaidRow], limit: Option) -> Result> { + let mut by_account: BTreeMap<&str, Vec> = BTreeMap::new(); + for row in rows.iter().filter(|r| r.status == RECORDED) { + let to = row.claim_account.as_deref().ok_or_else(|| { + QuantusError::Generic(format!("recorded claim {} has no claim_account", row.address)) + })?; + by_account.entry(to).or_default().push(Reward { + address: row.address.clone(), + amount_hundredths: row.amount_hundredths, + }); + } + let mut transfers = Vec::with_capacity(by_account.len()); + for (to, mut rewards) in by_account { + rewards.sort_by(|a, b| a.address.cmp(&b.address)); + let amount_hundredths = sum_hundredths(rewards.iter().map(|r| r.amount_hundredths))?; + transfers.push(Transfer { to: to.to_string(), amount_hundredths, rewards }); + } + if let Some(limit) = limit { + transfers.truncate(limit); + } + if transfers.is_empty() { + return Err(QuantusError::Generic("no recorded unpaid claims to pay".into())); + } + Ok(transfers) +} + +fn sum_hundredths(mut amounts: impl Iterator) -> Result { + amounts + .try_fold(0u64, |acc, a| acc.checked_add(a)) + .ok_or_else(|| QuantusError::Generic("payout total overflows".into())) +} + +/// Every manifest reward must still be an unpaid recorded claim with the same +/// destination and amount; anything else means the server moved on since pull. +fn drift_problems(transfers: &[Transfer], rows: &[UnpaidRow]) -> Vec { + let live: HashMap<&str, &UnpaidRow> = rows + .iter() + .filter(|r| r.status == RECORDED) + .map(|r| (r.address.as_str(), r)) + .collect(); + let mut problems = Vec::new(); + for transfer in transfers { + for reward in &transfer.rewards { + match live.get(reward.address.as_str()) { + None => + problems.push(format!("{}: no longer an unpaid recorded claim", reward.address)), + Some(row) if row.claim_account.as_deref() != Some(transfer.to.as_str()) => problems + .push(format!( + "{}: destination changed to {}", + reward.address, + row.claim_account.as_deref().unwrap_or("none") + )), + Some(row) if row.amount_hundredths != reward.amount_hundredths => + problems.push(format!( + "{}: amount changed {} -> {} QUAN", + reward.address, + format_hundredths(reward.amount_hundredths), + format_hundredths(row.amount_hundredths) + )), + Some(_) => {}, + } + } + } + problems +} + +async fn ensure_no_drift(manifest: &Manifest) -> Result<()> { + let rows = fetch_unpaid(&manifest.server).await?; + let problems = drift_problems(&manifest.transfers, &rows); + if problems.is_empty() { + log_success!("Server still lists every manifest reward as unpaid"); + return Ok(()); + } + for problem in &problems { + log_error!("{problem}"); + } + Err(QuantusError::Generic(format!( + "{} reward(s) changed on the server since this manifest was pulled; pull a fresh one", + problems.len() + ))) +} + +fn transfers_sha256(transfers: &[Transfer]) -> Result { + let bytes = serde_json::to_vec(transfers) + .map_err(|e| QuantusError::Generic(format!("serialize transfers: {e}")))?; + Ok(hex::encode(Sha256::digest(bytes))) +} + +/// Two live manifests could pay the same rewards twice, so a new pull waits +/// until every manifest in the directory is fully marked. +fn ensure_no_open_manifest(dir: &Path) -> Result<()> { + for entry in std::fs::read_dir(dir)? { + let path = entry?.path(); + if path.extension().and_then(|e| e.to_str()) != Some("json") { + continue; + } + let manifest = match Manifest::load(&path) { + Ok(manifest) => manifest, + Err(e) => { + log_verbose!("Ignoring {}: {e}", path.display()); + continue; + }, + }; + if manifest.state != State::Marked { + return Err(QuantusError::Generic(format!( + "{} is still '{}'; {} before pulling again", + path.display(), + manifest.state.label(), + manifest.state.next_step() + ))); + } + } + Ok(()) +} + +fn print_manifest(manifest: &Manifest) { + let pulled = chrono::DateTime::from_timestamp(manifest.pulled_at, 0) + .map(|t| t.to_rfc3339()) + .unwrap_or_else(|| manifest.pulled_at.to_string()); + log_print!( + "Manifest state {} server {} pulled {pulled}", + manifest.state.label().bright_yellow(), + manifest.server + ); + for transfer in &manifest.transfers { + log_print!( + " {} {} QUAN ({} reward{})", + transfer.to.bright_cyan(), + format_hundredths(transfer.amount_hundredths), + transfer.rewards.len(), + if transfer.rewards.len() == 1 { "" } else { "s" } + ); + } + log_print!( + "{} transfer(s) covering {} snapshot reward(s), {} QUAN total", + manifest.transfers.len(), + manifest.addresses().count(), + format_hundredths(manifest.total_hundredths).bright_green() + ); + if let Some(payment) = &manifest.payment { + log_print!( + "Payment: signer {} nonce {} anchor block #{} tx {} block {}", + payment.signer, + payment.nonce, + payment.anchor_block, + payment.tx_hash.as_deref().unwrap_or("pending"), + payment.block_hash.as_deref().unwrap_or("pending") + ); + } + if !manifest.marked.is_empty() { + log_print!( + "Marked paid on the server: {} of {}", + manifest.marked.len(), + manifest.addresses().count() + ); + } +} + +pub(super) async fn handle_pull( + server: String, + out: Option, + limit: Option, +) -> Result<()> { + let pulled_at = now_unix()?; + let out = out.unwrap_or_else(|| PathBuf::from(format!("airdrop-payout-{pulled_at}.json"))); + if out.exists() { + return Err(QuantusError::Generic(format!( + "{} already exists; manifests are never overwritten", + out.display() + ))); + } + let dir = out.parent().filter(|p| !p.as_os_str().is_empty()).unwrap_or(Path::new(".")); + ensure_no_open_manifest(dir)?; + + let rows = fetch_unpaid(&server).await?; + let recorded = rows.iter().filter(|r| r.status == RECORDED).count(); + let transfers = aggregate(&rows, limit)?; + let total_hundredths = sum_hundredths(transfers.iter().map(|t| t.amount_hundredths))?; + let manifest = Manifest { + version: MANIFEST_VERSION, + server, + pulled_at, + state: State::Pulled, + total_hundredths, + transfers, + approved_sha256: None, + payment: None, + marked: Vec::new(), + }; + manifest.save(&out)?; + print_manifest(&manifest); + log_print!( + "Server: {recorded} recorded claim(s) unpaid, {} unclaimed row(s) skipped", + rows.len() - recorded + ); + log_success!("Wrote {}", out.display()); + log_print!("Next: quantus airdrop review --manifest {} --approve", out.display()); + Ok(()) +} + +async fn chain_transfers( + client: &QuantusClient, + manifest: &Manifest, +) -> Result> { + let (_, decimals) = send::get_chain_properties(client).await?; + manifest + .transfers + .iter() + .map(|t| { + let amount = send::parse_amount_with_decimals( + &format_hundredths(t.amount_hundredths), + decimals, + )?; + Ok((t.to.clone(), amount)) + }) + .collect() +} + +fn encode_call(client: &QuantusClient, call: &impl Payload) -> Result> { + call.encode_call_data(&client.client().metadata()) + .map_err(|e| QuantusError::Generic(format!("encode batch call: {e}"))) +} + +fn fits_cold_payload(call_len: usize) -> bool { + call_len + COLD_PAYLOAD_OVERHEAD <= MAX_COLD_PAYLOAD +} + +pub(super) async fn handle_review( + node_url: &str, + manifest_path: PathBuf, + approve: bool, +) -> Result<()> { + let mut manifest = Manifest::load(&manifest_path)?; + print_manifest(&manifest); + if !matches!(manifest.state, State::Pulled | State::Approved) { + log_print!("Next: {}", manifest.state.next_step()); + return Ok(()); + } + ensure_no_drift(&manifest).await?; + + let client = QuantusClient::new(node_url).await?; + let transfers = chain_transfers(&client, &manifest).await?; + let call_len = encode_call(&client, &send::build_batch_transfer_call(&transfers)?)?.len(); + log_print!( + "Batch call: {} transfer(s), {call_len} bytes, {}", + transfers.len(), + if fits_cold_payload(call_len) { + "fits a cold-wallet QR payload".to_string() + } else { + format!("too large for a cold wallet ({MAX_COLD_PAYLOAD} bytes); pull with --limit") + } + ); + + if approve { + manifest.approved_sha256 = Some(transfers_sha256(&manifest.transfers)?); + manifest.state = State::Approved; + manifest.save(&manifest_path)?; + log_success!("Approved {}", manifest_path.display()); + log_print!( + "Next: quantus airdrop pay --manifest {} --from ", + manifest_path.display() + ); + } else if manifest.state == State::Approved { + manifest.ensure_approved()?; + log_print!("Manifest is approved. Next: {}", manifest.state.next_step()); + } else { + log_print!("Re-run with --approve to approve this manifest for payment."); + } + Ok(()) +} + +async fn block_number(client: &QuantusClient, hash: subxt::utils::H256) -> Result { + Ok(client.client().blocks().at(hash).await?.header().number as u64) +} + +pub(super) async fn handle_pay( + node_url: &str, + manifest_path: PathBuf, + from: String, + password: Option, + password_file: Option, + recover: bool, +) -> Result<()> { + let mut manifest = Manifest::load(&manifest_path)?; + let client = QuantusClient::new(node_url).await?; + if recover { + return recover_payment(&client, &mut manifest, &manifest_path).await; + } + manifest.ensure_approved()?; + ensure_no_drift(&manifest).await?; + + let signer = wallet::load_signer_from_wallet(&from, password, password_file)?; + let from_ss58 = signer.try_account_id_ss58check()?; + let transfers = chain_transfers(&client, &manifest).await?; + send::validate_batch_transfer_request(&client, &signer, &transfers).await?; + let call = send::build_batch_transfer_call(&transfers)?; + let call_len = encode_call(&client, &call)?.len(); + if matches!(signer, WalletSigner::Cold { .. }) && !fits_cold_payload(call_len) { + return Err(QuantusError::Generic(format!( + "batch call is {call_len} bytes, too large for cold-wallet signing ({MAX_COLD_PAYLOAD} max); pull with --limit" + ))); + } + let total = transfers + .iter() + .try_fold(0u128, |acc, (_, amount)| send::checked_add(acc, *amount, "payout total"))?; + let balance = send::get_balance(&client, &from_ss58).await?; + send::ensure_balance_covers_call( + &client, + &signer, + &call, + balance, + total, + None, + "airdrop payout", + ) + .await?; + + let account = AccountId32::from_ss58check_with_version(&from_ss58) + .map_err(|e| QuantusError::Generic(format!("invalid signer address {from_ss58}: {e:?}")))? + .0; + let nonce = client.get_account_nonce_from_best_block(&account).await?; + let anchor_block = block_number(&client, client.get_latest_block().await?).await?; + manifest.payment = Some(Payment { + signer: from_ss58.clone(), + nonce, + anchor_block, + tx_hash: None, + block_hash: None, + paid_at: None, + }); + manifest.state = State::Paying; + manifest.save(&manifest_path)?; + log_print!( + "Paying {} transfer(s), {} QUAN, from {} in one atomic batch…", + transfers.len(), + format_hundredths(manifest.total_hundredths).bright_green(), + from_ss58.bright_cyan() + ); + + let mode = ExecutionMode { finalized: true, wait_for_transaction: true }; + match common::submit_transaction_with_inclusion_block(&client, &signer, call, None, mode).await + { + Ok((tx_hash, included_in)) => + record_paid(&mut manifest, &manifest_path, tx_hash, included_in), + Err(e) => { + log_error!("Payment did not complete: {e}"); + log_print!( + "Manifest left in state 'paying'. Before anything else run: quantus airdrop pay --manifest {} --from {from} --recover", + manifest_path.display() + ); + Err(e) + }, + } +} + +fn record_paid( + manifest: &mut Manifest, + path: &Path, + tx_hash: subxt::utils::H256, + block_hash: Option, +) -> Result<()> { + let paid_at = now_unix()?; + let payment = manifest + .payment + .as_mut() + .ok_or_else(|| QuantusError::Generic("payment finished without a payment record".into()))?; + payment.tx_hash = Some(format!("{tx_hash:#x}")); + payment.block_hash = block_hash.map(|h| format!("{h:#x}")); + payment.paid_at = Some(paid_at); + manifest.state = State::Paid; + manifest.save(path)?; + log_success!("Batch finalized: {tx_hash:#x}"); + log_print!("Next: quantus airdrop mark-paid --manifest {}", path.display()); + Ok(()) +} + +fn reset_to_approved(manifest: &mut Manifest, path: &Path) -> Result<()> { + manifest.payment = None; + manifest.state = State::Approved; + manifest.save(path)?; + log_print!("Nothing was paid. Manifest is back to 'approved'; re-run `quantus airdrop pay`."); + Ok(()) +} + +/// After an interrupted `pay`: look for the exact batch (same signer, same +/// call bytes) in the finalized blocks it could have landed in. +async fn recover_payment( + client: &QuantusClient, + manifest: &mut Manifest, + path: &Path, +) -> Result<()> { + manifest.ensure_state(State::Paying)?; + let payment = manifest.payment()?.clone(); + let transfers = chain_transfers(client, manifest).await?; + let expected_call = encode_call(client, &send::build_batch_transfer_call(&transfers)?)?; + // SCALE `MultiAddress::Id(account)`, as `ExtrinsicDetails::address_bytes` returns it. + let mut expected_signer = vec![0u8]; + expected_signer.extend_from_slice(&parse_account_id(&payment.signer)?); + + let finalized = wormhole::at_finalized_block(client).await?; + let finalized_number = finalized.header().number as u64; + let expiry = payment.anchor_block + MORTALITY_BLOCKS + ANCHOR_SLACK_BLOCKS; + let last = finalized_number.min(expiry); + log_print!( + "Scanning finalized blocks #{}..=#{last} for the batch signed by {}", + payment.anchor_block, + payment.signer + ); + for number in payment.anchor_block..=last { + let block_hash = client.get_block_hash(number).await?; + let extrinsics = client.client().blocks().at(block_hash).await?.extrinsics().await?; + let Some(ext) = extrinsics.iter().find(|e| { + e.address_bytes() == Some(expected_signer.as_slice()) && e.call_bytes() == expected_call + }) else { + continue; + }; + let tx_hash = ext.hash(); + log_print!("Found batch {tx_hash:#x} in finalized block #{number}"); + return match common::check_execution_success(client.client(), &block_hash, &tx_hash).await { + Ok(()) => record_paid(manifest, path, tx_hash, Some(block_hash)), + Err(e) => { + log_error!("The batch was included but reverted: {e}"); + reset_to_approved(manifest, path) + }, + }; + } + if finalized_number > expiry { + log_print!( + "No batch found and its mortality window has passed; it can no longer be included." + ); + return reset_to_approved(manifest, path); + } + Err(QuantusError::Generic(format!( + "batch not found in finalized blocks yet; it could still land until block #{expiry} (finalized head is #{finalized_number}). Re-run --recover later" + ))) +} + +enum MarkOutcome { + Marked, + AlreadyMarked, + Failed(String), +} + +fn mark_outcome(status: reqwest::StatusCode, body: &str) -> MarkOutcome { + if status.is_success() { + MarkOutcome::Marked + } else if status == reqwest::StatusCode::CONFLICT { + MarkOutcome::AlreadyMarked + } else { + MarkOutcome::Failed(format_server_error(status, body)) + } +} + +fn admin_token(file: Option) -> Result { + if let Some(path) = file { + return password::read_secret_file(&path, "admin token"); + } + std::env::var(ADMIN_TOKEN_ENV).map_err(|_| { + QuantusError::Generic(format!("provide --admin-token-file or set {ADMIN_TOKEN_ENV}")) + }) +} + +pub(super) async fn handle_mark_paid( + manifest_path: PathBuf, + admin_token_file: Option, +) -> Result<()> { + let mut manifest = Manifest::load(&manifest_path)?; + if manifest.state == State::Marked { + log_print!("Every reward in {} is already marked paid.", manifest_path.display()); + return Ok(()); + } + manifest.ensure_state(State::Paid)?; + let payment = manifest.payment()?; + let tx_hash = payment.tx_hash.clone().ok_or_else(|| { + QuantusError::Generic("manifest is 'paid' but has no transaction hash".into()) + })?; + let paid_at = payment.paid_at; + let token = admin_token(admin_token_file)?; + let client = http_client()?; + let url = format!("{}/mark-paid", manifest.server.trim_end_matches('/')); + let pending: Vec = manifest + .addresses() + .filter(|a| !manifest.marked.iter().any(|m| m == a)) + .map(str::to_string) + .collect(); + log_print!("Marking {} address(es) paid by {tx_hash}", pending.len()); + + let mut failed = 0usize; + for address in pending { + let body = serde_json::json!({ "address": address, "paid_at": paid_at }); + let response = client + .post(&url) + .bearer_auth(&token) + .json(&body) + .send() + .await + .map_err(http_err)?; + let status = response.status(); + let text = response.text().await.map_err(http_err)?; + match mark_outcome(status, &text) { + MarkOutcome::Marked => log_success!("{address} marked paid"), + MarkOutcome::AlreadyMarked => log_print!("{address} was already marked paid"), + MarkOutcome::Failed(message) => { + log_error!("{address}: {message}"); + failed += 1; + continue; + }, + } + manifest.marked.push(address); + manifest.save(&manifest_path)?; + } + if failed > 0 { + return Err(QuantusError::Generic(format!( + "{failed} address(es) not marked; re-run mark-paid to retry them" + ))); + } + manifest.state = State::Marked; + manifest.save(&manifest_path)?; + log_success!("All {} address(es) marked paid", manifest.marked.len()); + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn row(address: &str, to: Option<&str>, amount: u64, status: &str) -> UnpaidRow { + UnpaidRow { + address: address.into(), + claim_account: to.map(Into::into), + amount_hundredths: amount, + status: status.into(), + } + } + + fn rows() -> Vec { + vec![ + row("qzB", Some("qzDest2"), 300, RECORDED), + row("qzA", Some("qzDest1"), 150, RECORDED), + row("qzC", None, 10, "unclaimed"), + row("qzD", Some("qzDest1"), 50, RECORDED), + ] + } + + fn manifest(transfers: Vec) -> Manifest { + Manifest { + version: MANIFEST_VERSION, + server: "http://server".into(), + pulled_at: 0, + state: State::Pulled, + total_hundredths: transfers.iter().map(|t| t.amount_hundredths).sum(), + transfers, + approved_sha256: None, + payment: None, + marked: Vec::new(), + } + } + + #[test] + fn aggregate_groups_recorded_claims_by_destination() { + let transfers = aggregate(&rows(), None).unwrap(); + assert_eq!(transfers.len(), 2); + assert_eq!(transfers[0].to, "qzDest1"); + assert_eq!(transfers[0].amount_hundredths, 200); + assert_eq!( + transfers[0].rewards.iter().map(|r| r.address.as_str()).collect::>(), + ["qzA", "qzD"] + ); + assert_eq!(transfers[1].to, "qzDest2"); + assert_eq!(aggregate(&rows(), Some(1)).unwrap().len(), 1); + } + + #[test] + fn aggregate_rejects_empty_and_destinationless_claims() { + assert!(aggregate(&[row("qzC", None, 10, "unclaimed")], None).is_err()); + assert!(aggregate(&[row("qzE", None, 10, RECORDED)], None).is_err()); + } + + #[test] + fn drift_reports_paid_moved_and_changed_rewards() { + let transfers = aggregate(&rows(), None).unwrap(); + assert!(drift_problems(&transfers, &rows()).is_empty()); + let live = vec![ + row("qzB", Some("qzDest2"), 301, RECORDED), + row("qzA", Some("qzOther"), 150, RECORDED), + ]; + let problems = drift_problems(&transfers, &live); + assert_eq!(problems.len(), 3, "{problems:?}"); + assert!(problems.iter().any(|p| p.starts_with("qzA: destination changed"))); + assert!(problems.iter().any(|p| p.starts_with("qzB: amount changed"))); + assert!(problems.iter().any(|p| p.starts_with("qzD: no longer"))); + } + + #[test] + fn approval_hash_pins_the_transfers() { + let mut m = manifest(aggregate(&rows(), None).unwrap()); + m.state = State::Approved; + m.approved_sha256 = Some(transfers_sha256(&m.transfers).unwrap()); + m.ensure_approved().unwrap(); + m.transfers[0].amount_hundredths += 1; + assert!(m.ensure_approved().is_err()); + m.state = State::Pulled; + assert!(m.ensure_approved().unwrap_err().to_string().contains("not 'approved'")); + } + + #[test] + fn open_manifest_blocks_a_new_pull() { + let dir = tempfile::tempdir().unwrap(); + std::fs::write(dir.path().join("notes.json"), "{\"unrelated\": true}").unwrap(); + ensure_no_open_manifest(dir.path()).unwrap(); + + let mut m = manifest(aggregate(&rows(), None).unwrap()); + m.state = State::Paid; + m.save(&dir.path().join("payout.json")).unwrap(); + let err = ensure_no_open_manifest(dir.path()).unwrap_err().to_string(); + assert!(err.contains("still 'paid'"), "{err}"); + + m.state = State::Marked; + m.save(&dir.path().join("payout.json")).unwrap(); + ensure_no_open_manifest(dir.path()).unwrap(); + assert_eq!(Manifest::load(&dir.path().join("payout.json")).unwrap().state, State::Marked); + } + + #[test] + fn mark_outcome_treats_conflict_as_already_marked() { + assert!(matches!(mark_outcome(reqwest::StatusCode::OK, ""), MarkOutcome::Marked)); + assert!(matches!( + mark_outcome( + reqwest::StatusCode::CONFLICT, + "{\"error\":\"address already marked paid\"}" + ), + MarkOutcome::AlreadyMarked + )); + match mark_outcome(reqwest::StatusCode::NOT_FOUND, "{\"error\":\"no claim\"}") { + MarkOutcome::Failed(msg) => assert_eq!(msg, "server 404 Not Found: no claim"), + _ => panic!("404 must fail"), + } + } + + #[test] + fn cold_payload_budget() { + assert!(fits_cold_payload(MAX_COLD_PAYLOAD - COLD_PAYLOAD_OVERHEAD)); + assert!(!fits_cold_payload(MAX_COLD_PAYLOAD - COLD_PAYLOAD_OVERHEAD + 1)); + } +} diff --git a/src/cli/block.rs b/src/cli/block.rs index 1a7084b..49e44f2 100644 --- a/src/cli/block.rs +++ b/src/cli/block.rs @@ -887,15 +887,7 @@ async fn list_blocks_in_range( for block_num in (start..=end).step_by(step as usize) { // Get block hash for this block number - let block_hash: subxt::utils::H256 = quantus_client - .rpc_client() - .request::("chain_getBlockHash", [block_num]) - .await - .map_err(|e| { - QuantusError::NetworkError(format!( - "Failed to get block hash for block {block_num}: {e:?}" - )) - })?; + let block_hash = quantus_client.get_block_hash(block_num as u64).await?; // Get block data let block_data: serde_json::Value = quantus_client diff --git a/src/cli/mod.rs b/src/cli/mod.rs index 318cd40..f11ef7f 100644 --- a/src/cli/mod.rs +++ b/src/cli/mod.rs @@ -294,7 +294,7 @@ pub enum Commands { #[command(subcommand)] Wormhole(wormhole::WormholeCommands), - /// Claim testnet airdrop rewards + /// Claim testnet airdrop rewards, or pay them out (operators) #[command(subcommand)] Airdrop(airdrop::AirdropCommands), @@ -515,7 +515,8 @@ pub async fn execute_command( Commands::Block(block_cmd) => block::handle_block_command(block_cmd, node_url).await, Commands::Wormhole(wormhole_cmd) => wormhole::handle_wormhole_command(wormhole_cmd, node_url, execution_mode).await, - Commands::Airdrop(airdrop_cmd) => airdrop::handle_airdrop_command(airdrop_cmd).await, + Commands::Airdrop(airdrop_cmd) => + airdrop::handle_airdrop_command(airdrop_cmd, node_url).await, Commands::Multisend { from, addresses_file, diff --git a/src/collect_rewards_lib.rs b/src/collect_rewards_lib.rs index f0d0596..a263882 100644 --- a/src/collect_rewards_lib.rs +++ b/src/collect_rewards_lib.rs @@ -387,21 +387,7 @@ pub async fn collect_rewards( // 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 - use subxt::ext::jsonrpsee::{core::client::ClientT, rpc_params}; - let block_hash: Option = quantus_client - .rpc_client() - .request("chain_getBlockHash", rpc_params![block_num]) - .await - .map_err(|e| { - CollectRewardsError::from(format!( - "Failed to get block hash for block {}: {}", - block_num, e - )) - })?; - let block_hash = block_hash.ok_or_else(|| { - CollectRewardsError::from(format!("Block {} not found", block_num)) - })?; + let block_hash = quantus_client.get_block_hash(block_num as u64).await?; quantus_client .client() .blocks()