diff --git a/tools/usdt-bridge/src/main.rs b/tools/usdt-bridge/src/main.rs index 30d24020..83f1e705 100644 --- a/tools/usdt-bridge/src/main.rs +++ b/tools/usdt-bridge/src/main.rs @@ -35,6 +35,7 @@ use serde::{Deserialize, Serialize}; use sp_core::{crypto::Ss58Codec, sr25519, Pair}; use std::path::PathBuf; use std::sync::LazyLock; +use std::time::Duration; use subxt::{OnlineClient, SubstrateConfig}; use subxt::backend::legacy::LegacyRpcMethods; use subxt::backend::rpc::RpcClient; @@ -62,6 +63,17 @@ use tracing::{info, warn, error}; // specifically so it CAN be loaded programmatically; that is its whole purpose. const AUTOMATION_KEY_ADDRESS: &str = "5GQu4PFUb1f3MTJ7i7c1CtLgDk3TVvpSW1VbQCRmfkMoC8cM"; +/// A live `subscribe_finalized()` stream can silently stop yielding blocks - no error, no closed +/// stream, the process just stops seeing new blocks forever - while staying up and logging +/// nothing further ("Listening for deposits..." was the last line for 3+ hours while a real 1 +/// USDT deposit sat undetected at block #18250967, ~62 minutes into that silence - see +/// res/validators-tiki.md 2026-07-15). Checkpoint + backfill only covers a gap across a +/// *restart*; this covers a stall with no restart at all. If no new finalized block arrives +/// within this window (Asset Hub blocks land every ~6-12s under normal conditions, so this is a +/// wide margin, not a tight one), treat the subscription as stalled and force a full reconnect - +/// the existing checkpoint-vs-head backfill then catches up whatever was missed during the stall. +const LIVE_SUBSCRIPTION_STALL_TIMEOUT: Duration = Duration::from_secs(120); + // ============================================================================ // Configuration // ============================================================================ @@ -1222,68 +1234,98 @@ async fn listen_deposits(config: &BridgeConfig, automation: &AutomationContext) 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?; + // Outer loop: a full reconnect (fresh client + fresh legacy RPC) on every stall or dropped + // subscription. Each pass re-derives last_seen from the persisted checkpoint and backfills any + // gap before going live again - the same recovery path an ordinary process restart already + // gets, just triggered by a detected stall instead of only by systemd. + loop { + 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; + // 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 (re)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 { + 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)?; } - process_deposit_block(config, automation, &block).await?; - last_seen = block_number; + println!("Listening for deposits...\n"); - 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)?; + let mut blocks = client.blocks().subscribe_finalized().await?; + + loop { + let next = match tokio::time::timeout(LIVE_SUBSCRIPTION_STALL_TIMEOUT, blocks.next()).await { + Ok(next) => next, + Err(_) => { + warn!("No new finalized Polkadot block in {}s - live subscription looks stalled, forcing a reconnect", LIVE_SUBSCRIPTION_STALL_TIMEOUT.as_secs()); + notify_telegram(config, &format!( + "⚠️ usdt-bridge: Polkadot deposit subscription produced nothing for {}s - reconnecting now. Checkpoint backfill will catch up anything missed during the stall.", + LIVE_SUBSCRIPTION_STALL_TIMEOUT.as_secs() + )).await; + break; + } + }; + + let Some(block) = next else { + warn!("Polkadot block subscription ended - reconnecting"); + notify_telegram(config, "⚠️ usdt-bridge: Polkadot deposit subscription ended unexpectedly - reconnecting now.").await; + break; + }; + + let block = match block { + Ok(block) => block, + Err(e) => { + warn!("Polkadot block subscription yielded an error ({e}) - reconnecting"); + notify_telegram(config, &format!("⚠️ usdt-bridge: Polkadot deposit subscription errored ({e}) - reconnecting now.")).await; + break; + } + }; + + 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(()) } /// Watches Pezkuwi Asset Hub for wUSDT sent to the bridge's own Pezkuwi account (a user @@ -1558,64 +1600,92 @@ async fn listen_withdrawals(config: &BridgeConfig, automation: &AutomationContex 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?; + // Outer loop: same stall-recovery strategy as listen_deposits - see its comment for why a + // live subscription alone (even one that's never restarted) isn't a correctness guarantee. + loop { + 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; + // 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 { + 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)?; } - process_withdrawal_block(config, automation, &block).await?; - last_seen = block_number; + println!("Listening for withdrawals...\n"); - 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)?; + let mut blocks = pezkuwi_client.blocks().subscribe_finalized().await?; + + loop { + let next = match tokio::time::timeout(LIVE_SUBSCRIPTION_STALL_TIMEOUT, blocks.next()).await { + Ok(next) => next, + Err(_) => { + warn!("No new finalized Pezkuwi block in {}s - live subscription looks stalled, forcing a reconnect", LIVE_SUBSCRIPTION_STALL_TIMEOUT.as_secs()); + notify_telegram(config, &format!( + "⚠️ usdt-bridge: Pezkuwi withdrawal subscription produced nothing for {}s - reconnecting now. Checkpoint backfill will catch up anything missed during the stall.", + LIVE_SUBSCRIPTION_STALL_TIMEOUT.as_secs() + )).await; + break; + } + }; + + let Some(block) = next else { + warn!("Pezkuwi block subscription ended - reconnecting"); + notify_telegram(config, "⚠️ usdt-bridge: Pezkuwi withdrawal subscription ended unexpectedly - reconnecting now.").await; + break; + }; + + let block = match block { + Ok(block) => block, + Err(e) => { + warn!("Pezkuwi block subscription yielded an error ({e}) - reconnecting"); + notify_telegram(config, &format!("⚠️ usdt-bridge: Pezkuwi withdrawal subscription errored ({e}) - reconnecting now.")).await; + break; + } + }; + + 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(()) } async fn process_withdrawals(config: &BridgeConfig) -> Result<()> {