diff --git a/tools/usdt-bridge/src/main.rs b/tools/usdt-bridge/src/main.rs index 280a3d63..30d24020 100644 --- a/tools/usdt-bridge/src/main.rs +++ b/tools/usdt-bridge/src/main.rs @@ -36,6 +36,8 @@ use sp_core::{crypto::Ss58Codec, sr25519, Pair}; use std::path::PathBuf; use std::sync::LazyLock; use subxt::{OnlineClient, SubstrateConfig}; +use subxt::backend::legacy::LegacyRpcMethods; +use subxt::backend::rpc::RpcClient; use subxt::dynamic::{At, Value}; use subxt_signer::bip39::Mnemonic; use subxt_signer::sr25519::Keypair as PolkadotKeypair; @@ -984,38 +986,24 @@ fn calculate_fee(amount: u128, fee_basis_points: u32) -> u128 { amount * fee_basis_points as u128 / 10_000 } -async fn listen_deposits(config: &BridgeConfig, automation: &AutomationContext) -> Result<()> { +/// Processes exactly one block's deposit-relevant events. Called identically whether the block +/// arrived via the live finalized-block subscription or via backfill - this is the ONE place the +/// detection/auto-pay logic lives, so a gap in live coverage can never silently produce +/// different behavior than a backfilled scan of the same block would. +async fn process_deposit_block( + config: &BridgeConfig, + automation: &AutomationContext, + block: &subxt::blocks::Block>, +) -> Result<()> { let polkadot_addr = &config.custody_multisig_polkadot; + let block_number = block.number(); + let events = block.events().await?; - println!("╔══════════════════════════════════════════════════════════════╗"); - println!("║ DEPOSIT LISTENER (detect-only) ║"); - println!("╚══════════════════════════════════════════════════════════════╝\n"); + for event in events.iter() { + let event = event?; - println!("Bridge Address (Polkadot, 3-of-5 multisig): {}", polkadot_addr); - println!("USDT Asset ID: {}", config.polkadot_usdt_asset_id); - println!("Min Deposit: {} USDT", config.min_deposit as f64 / 1_000_000.0); - println!("\nNo key held by this process can mint on the multisig's behalf -"); - println!("detected deposits are recorded as pending_multisig_approval. An actual"); - println!("mint requires 3 of the 5 signatories to approve via pwap-web.\n"); - println!("Listening for deposits...\n"); - - let client = connect_to_chain(&config.polkadot_rpc).await?; - - // Subscribe to new blocks - let mut blocks = client.blocks().subscribe_finalized().await?; - - while let Some(block) = blocks.next().await { - let block = block?; - let block_number = block.number(); - - // Get events for this block - let events = block.events().await?; - - for event in events.iter() { - let event = event?; - - // Check for Assets.Transferred event - if event.pallet_name() == "Assets" && event.variant_name() == "Transferred" { + // Check for Assets.Transferred event + if event.pallet_name() == "Assets" && event.variant_name() == "Transferred" { // Parse event data if let Ok(fields) = event.field_values() { // Fields: asset_id, from, to, amount @@ -1182,6 +1170,117 @@ async fn listen_deposits(config: &BridgeConfig, automation: &AutomationContext) } } } + + Ok(()) +} + +/// Connects to `url` twice: once via the full subxt client (for reading storage/events/blocks), +/// once via the raw legacy RPC methods (needed only for `chain_get_block_hash` - resolving a +/// plain block NUMBER to a hash, which the high-level `Blocks` API doesn't expose directly). Two +/// lightweight connections to the same endpoint is simpler and safer than threading a shared +/// RPC client through `connect_to_chain`'s existing callers. +async fn connect_legacy_rpc(url: &str) -> Result> { + let rpc_client = RpcClient::from_url(url).await?; + Ok(LegacyRpcMethods::::new(rpc_client)) +} + +/// Fetches block `block_number` by hash and runs it through `process_deposit_block`, then +/// persists the checkpoint. Used both for startup backfill (resuming after downtime) and for +/// closing any gap the live subscription itself might open up mid-run. +async fn backfill_deposit_block( + config: &BridgeConfig, + automation: &AutomationContext, + client: &OnlineClient, + legacy_rpc: &LegacyRpcMethods, + block_number: u64, +) -> Result<()> { + let hash = legacy_rpc + .chain_get_block_hash(Some(block_number.into())) + .await? + .ok_or_else(|| anyhow!("no block hash found for Polkadot block #{block_number}"))?; + let block = client.blocks().at(hash).await?; + process_deposit_block(config, automation, &block).await?; + + let _guard = DB_LOCK.lock().await; + let mut db = BridgeDatabase::load(&config.db_path)?; + db.stats.last_polkadot_block = block_number; + db.save(&config.db_path)?; + Ok(()) +} + +async fn listen_deposits(config: &BridgeConfig, automation: &AutomationContext) -> Result<()> { + let polkadot_addr = &config.custody_multisig_polkadot; + + println!("╔══════════════════════════════════════════════════════════════╗"); + println!("║ DEPOSIT LISTENER (checkpoint + backfill) ║"); + println!("╚══════════════════════════════════════════════════════════════╝\n"); + + println!("Bridge Address (Polkadot, 3-of-5 multisig): {}", polkadot_addr); + println!("USDT Asset ID: {}", config.polkadot_usdt_asset_id); + println!("Min Deposit: {} USDT", config.min_deposit as f64 / 1_000_000.0); + println!("\nNo key held by this process can mint on the multisig's behalf -"); + println!("detected deposits are recorded as pending_multisig_approval. An actual"); + println!("mint requires 3 of the 5 signatories to approve via pwap-web.\n"); + + let client = connect_to_chain(&config.polkadot_rpc).await?; + let legacy_rpc = connect_legacy_rpc(&config.polkadot_rpc).await?; + + // Checkpoint + backfill: a live block subscription alone is not a correctness guarantee - a + // transient reconnect or subscription hiccup can silently skip blocks with zero logging (this + // is exactly how a real user deposit went undetected in production - see + // res/validators-tiki.md 2026-07-15). On every start, explicitly scan any gap between the + // last persisted checkpoint and the current chain head before relying on the live stream. + let last_processed = { BridgeDatabase::load(&config.db_path)?.stats.last_polkadot_block }; + let head_hash = legacy_rpc.chain_get_finalized_head().await?; + let head_block = client.blocks().at(head_hash).await?; + let head_number = head_block.number() as u64; + + let mut last_seen = if last_processed == 0 { + info!("No checkpoint found - starting from current head (block {head_number}), not scanning full history"); + head_number + } else if head_number > last_processed { + info!("Backfilling Polkadot blocks {} to {head_number} before going live (resuming from checkpoint)", last_processed + 1); + for n in (last_processed + 1)..=head_number { + backfill_deposit_block(config, automation, &client, &legacy_rpc, n).await?; + } + head_number + } else { + last_processed + }; + + { + let _guard = DB_LOCK.lock().await; + let mut db = BridgeDatabase::load(&config.db_path)?; + db.stats.last_polkadot_block = last_seen; + db.save(&config.db_path)?; + } + + println!("Listening for deposits...\n"); + + let mut blocks = client.blocks().subscribe_finalized().await?; + + while let Some(block) = blocks.next().await { + let block = block?; + let block_number = block.number() as u64; + + if block_number > last_seen + 1 { + warn!("Gap detected in live Polkadot subscription: last processed {last_seen}, current {block_number} - backfilling {} missing block(s)", block_number - last_seen - 1); + notify_telegram(config, &format!( + "⚠️ usdt-bridge: gap detected in live Polkadot block subscription ({} block(s) between #{last_seen} and #{block_number}) - backfilling now, no deposit should be lost, but this is worth knowing about.", + block_number - last_seen - 1 + )).await; + for n in (last_seen + 1)..block_number { + backfill_deposit_block(config, automation, &client, &legacy_rpc, n).await?; + } + } + + process_deposit_block(config, automation, &block).await?; + last_seen = block_number; + + let _guard = DB_LOCK.lock().await; + let mut db = BridgeDatabase::load(&config.db_path)?; + db.stats.last_polkadot_block = block_number; + db.save(&config.db_path)?; } Ok(()) @@ -1194,34 +1293,22 @@ async fn listen_deposits(config: &BridgeConfig, automation: &AutomationContext) /// bridge was missing - a mismatch here means the peg is under-collateralized and must halt, /// not silently short-pay a user), then release on Polkadot. Mirrors listen_deposits() and /// shares its event-matching caveat - see the comment there. -async fn listen_withdrawals(config: &BridgeConfig, automation: &AutomationContext) -> Result<()> { +/// Processes exactly one block's withdrawal-relevant events - see process_deposit_block's doc +/// comment for why this is factored out identically for live and backfilled blocks. +async fn process_withdrawal_block( + config: &BridgeConfig, + automation: &AutomationContext, + block: &subxt::blocks::Block>, +) -> Result<()> { let polkadot_bridge_addr = &config.custody_multisig_polkadot; let pezkuwi_bridge_addr = &config.custody_multisig_pezkuwi; + let block_number = block.number(); + let events = block.events().await?; - println!("╔══════════════════════════════════════════════════════════════╗"); - println!("║ WITHDRAWAL LISTENER (detect-only) ║"); - println!("╚══════════════════════════════════════════════════════════════╝\n"); + for event in events.iter() { + let event = event?; - println!("Bridge Address (Pezkuwi, 3-of-5 multisig): {}", pezkuwi_bridge_addr); - println!("wUSDT Asset ID: {}", config.pezkuwi_wusdt_asset_id); - println!("Min Withdraw: {} USDT", config.min_withdraw as f64 / 1_000_000.0); - println!("\nNo key held by this process can burn/release on the multisig's behalf -"); - println!("detected withdrawal requests are recorded as pending_multisig_approval.\n"); - println!("Listening for withdrawals...\n"); - - let pezkuwi_client = connect_to_chain(&config.pezkuwi_rpc).await?; - - let mut blocks = pezkuwi_client.blocks().subscribe_finalized().await?; - - while let Some(block) = blocks.next().await { - let block = block?; - let block_number = block.number(); - let events = block.events().await?; - - for event in events.iter() { - let event = event?; - - if event.pallet_name() == "Assets" && event.variant_name() == "Transferred" { + if event.pallet_name() == "Assets" && event.variant_name() == "Transferred" { if let Ok(fields) = event.field_values() { let asset_id = fields.at("asset_id") .and_then(|v| v.as_u128()) @@ -1432,6 +1519,100 @@ async fn listen_withdrawals(config: &BridgeConfig, automation: &AutomationContex } } } + + Ok(()) +} + +/// Same pattern as backfill_deposit_block, for the Pezkuwi side. +async fn backfill_withdrawal_block( + config: &BridgeConfig, + automation: &AutomationContext, + client: &OnlineClient, + legacy_rpc: &LegacyRpcMethods, + block_number: u64, +) -> Result<()> { + let hash = legacy_rpc + .chain_get_block_hash(Some(block_number.into())) + .await? + .ok_or_else(|| anyhow!("no block hash found for Pezkuwi block #{block_number}"))?; + let block = client.blocks().at(hash).await?; + process_withdrawal_block(config, automation, &block).await?; + + let _guard = DB_LOCK.lock().await; + let mut db = BridgeDatabase::load(&config.db_path)?; + db.stats.last_pezkuwi_block = block_number; + db.save(&config.db_path)?; + Ok(()) +} + +async fn listen_withdrawals(config: &BridgeConfig, automation: &AutomationContext) -> Result<()> { + let pezkuwi_bridge_addr = &config.custody_multisig_pezkuwi; + + println!("╔══════════════════════════════════════════════════════════════╗"); + println!("║ WITHDRAWAL LISTENER (checkpoint + backfill) ║"); + println!("╚══════════════════════════════════════════════════════════════╝\n"); + + println!("Bridge Address (Pezkuwi, 3-of-5 multisig): {}", pezkuwi_bridge_addr); + println!("wUSDT Asset ID: {}", config.pezkuwi_wusdt_asset_id); + println!("Min Withdraw: {} USDT", config.min_withdraw as f64 / 1_000_000.0); + println!("\nNo key held by this process can burn/release on the multisig's behalf -"); + println!("detected withdrawal requests are recorded as pending_multisig_approval.\n"); + + let pezkuwi_client = connect_to_chain(&config.pezkuwi_rpc).await?; + let legacy_rpc = connect_legacy_rpc(&config.pezkuwi_rpc).await?; + + // Checkpoint + backfill - see listen_deposits for why this replaced pure live-subscription. + let last_processed = { BridgeDatabase::load(&config.db_path)?.stats.last_pezkuwi_block }; + let head_hash = legacy_rpc.chain_get_finalized_head().await?; + let head_block = pezkuwi_client.blocks().at(head_hash).await?; + let head_number = head_block.number() as u64; + + let mut last_seen = if last_processed == 0 { + info!("No checkpoint found - starting from current head (block {head_number}), not scanning full history"); + head_number + } else if head_number > last_processed { + info!("Backfilling Pezkuwi blocks {} to {head_number} before going live (resuming from checkpoint)", last_processed + 1); + for n in (last_processed + 1)..=head_number { + backfill_withdrawal_block(config, automation, &pezkuwi_client, &legacy_rpc, n).await?; + } + head_number + } else { + last_processed + }; + + { + let _guard = DB_LOCK.lock().await; + let mut db = BridgeDatabase::load(&config.db_path)?; + db.stats.last_pezkuwi_block = last_seen; + db.save(&config.db_path)?; + } + + println!("Listening for withdrawals...\n"); + + let mut blocks = pezkuwi_client.blocks().subscribe_finalized().await?; + + while let Some(block) = blocks.next().await { + let block = block?; + let block_number = block.number() as u64; + + if block_number > last_seen + 1 { + warn!("Gap detected in live Pezkuwi subscription: last processed {last_seen}, current {block_number} - backfilling {} missing block(s)", block_number - last_seen - 1); + notify_telegram(config, &format!( + "⚠️ usdt-bridge: gap detected in live Pezkuwi block subscription ({} block(s) between #{last_seen} and #{block_number}) - backfilling now, no withdrawal should be lost, but this is worth knowing about.", + block_number - last_seen - 1 + )).await; + for n in (last_seen + 1)..block_number { + backfill_withdrawal_block(config, automation, &pezkuwi_client, &legacy_rpc, n).await?; + } + } + + process_withdrawal_block(config, automation, &block).await?; + last_seen = block_number; + + let _guard = DB_LOCK.lock().await; + let mut db = BridgeDatabase::load(&config.db_path)?; + db.stats.last_pezkuwi_block = block_number; + db.save(&config.db_path)?; } Ok(())