pezkuwichain/pezkuwi-sdk
1
Fork 0
mirror of https://gitlab.com/pezkuwichain/pezkuwi-sdk.git synced 2026-10-21 03:06:29 +00:00
Code Releases Activity

Add checkpoint + backfill: never rely on live subscription alone

Browse Source 
listen_deposits/listen_withdrawals previously used ONLY a live finalized-
block subscription with no persisted cursor - the last_polkadot_block/
last_pezkuwi_block fields existed in BridgeStats but were never read or
written anywhere. A transient subscription gap (no logging, no crash, just
a silently skipped block) could make a real deposit vanish from detection
entirely. Confirmed live: a real user's 1 USDT deposit landed correctly at
the multisig custody address (verified via direct chain balance query) but
was never once logged as detected across 7+ hours of continuous uptime with
zero errors - see res/validators-tiki.md 2026-07-15 for the full incident.

Both listeners now:
1. On startup, compare the persisted checkpoint against current chain head
   and explicitly backfill (fetch block-by-hash via the legacy RPC's
   chain_get_block_hash, then process identically to the live path) any gap
   before starting the live subscription.
2. During live operation, detect if the subscription itself ever jumps ahead
   unexpectedly and backfill that gap too, paging Telegram when it happens.
3. Persist the checkpoint after every block processed, live or backfilled.

process_deposit_block/process_withdrawal_block are the single shared
per-block handlers - backfilled and live blocks run through identical logic,
so there's no way for the two paths to silently diverge in behavior.
This commit is contained in:
Satoshi Qazi Muhammed committed 2026-07-15 05:38:39 -07:00
1 parent c0a982b32c
commit 17de8a171d
1 file changed
+234 -53
Download Patch File Download Diff File Expand all files Collapse all files
tools/usdt-bridge/src/main.rs
+234 -53
View File
@@ -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<SubstrateConfig, OnlineClient<SubstrateConfig>>,
) -> 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<LegacyRpcMethods<SubstrateConfig>> {
let rpc_client = RpcClient::from_url(url).await?;
Ok(LegacyRpcMethods::<SubstrateConfig>::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<SubstrateConfig>,
legacy_rpc: &LegacyRpcMethods<SubstrateConfig>,
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<SubstrateConfig, OnlineClient<SubstrateConfig>>,
) -> 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<SubstrateConfig>,
legacy_rpc: &LegacyRpcMethods<SubstrateConfig>,
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(())