pezkuwichain/pezkuwi-sdk
1
Fork 0
mirror of https://gitlab.com/pezkuwichain/pezkuwi-sdk.git synced 2026-10-20 18:56:23 +00:00
Code Releases Activity

Fix silent live-subscription stall: force reconnect if no block for 120s

Browse Source 
A real 1 USDT deposit landed on-chain at block #18250967 and was never
detected: subscribe_finalized()'s stream silently stopped yielding new
blocks after startup - no error, no closed stream, the process stayed up
and "Listening for deposits..." was the last line logged for 3+ hours while
the deposit sat undetected the whole time. Checkpoint + backfill (added
earlier) only covers a gap across a process restart; it does nothing for a
stall with no restart at all, which is exactly what happened here.

Both listen_deposits and listen_withdrawals now wrap their subscribe loop in
an outer reconnect loop: each blocks.next() call is bounded by a 120s
timeout (Asset Hub blocks land every ~6-12s, so this is a wide margin), and
a timeout, a closed stream, or a stream-yielded error all trigger a full
reconnect (fresh client + fresh legacy RPC) rather than silently hanging or
crashing the whole process. The existing checkpoint-vs-head backfill then
catches up whatever was missed during the stall, same as it already does
for an ordinary restart.
This commit is contained in:
Satoshi Qazi Muhammed committed 2026-07-15 09:56:59 -07:00
1 parent 17de8a171d
commit f02f831edf
1 file changed
+174 -104
Download Patch File Download Diff File Expand all files Collapse all files
tools/usdt-bridge/src/main.rs
+174 -104
View File
@@ -35,6 +35,7 @@ use serde::{Deserialize, Serialize};
use sp_core::{crypto::Ss58Codec, sr25519, Pair}; use sp_core::{crypto::Ss58Codec, sr25519, Pair};
use std::path::PathBuf; use std::path::PathBuf;
use std::sync::LazyLock; use std::sync::LazyLock;
use std::time::Duration;
use subxt::{OnlineClient, SubstrateConfig}; use subxt::{OnlineClient, SubstrateConfig};
use subxt::backend::legacy::LegacyRpcMethods; use subxt::backend::legacy::LegacyRpcMethods;
use subxt::backend::rpc::RpcClient; 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. // specifically so it CAN be loaded programmatically; that is its whole purpose.
const AUTOMATION_KEY_ADDRESS: &str = "5GQu4PFUb1f3MTJ7i7c1CtLgDk3TVvpSW1VbQCRmfkMoC8cM"; 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 // 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!("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!("mint requires 3 of the 5 signatories to approve via pwap-web.\n");
let client = connect_to_chain(&config.polkadot_rpc).await?; // Outer loop: a full reconnect (fresh client + fresh legacy RPC) on every stall or dropped
let legacy_rpc = connect_legacy_rpc(&config.polkadot_rpc).await?; // 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 // 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 // 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 // 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 // res/validators-tiki.md 2026-07-15). On every (re)start, explicitly scan any gap between
// last persisted checkpoint and the current chain head before relying on the live stream. // 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 last_processed = { BridgeDatabase::load(&config.db_path)?.stats.last_polkadot_block };
let head_hash = legacy_rpc.chain_get_finalized_head().await?; let head_hash = legacy_rpc.chain_get_finalized_head().await?;
let head_block = client.blocks().at(head_hash).await?; let head_block = client.blocks().at(head_hash).await?;
let head_number = head_block.number() as u64; let head_number = head_block.number() as u64;
let mut last_seen = if last_processed == 0 { let mut last_seen = if last_processed == 0 {
info!("No checkpoint found - starting from current head (block {head_number}), not scanning full history"); info!("No checkpoint found - starting from current head (block {head_number}), not scanning full history");
head_number head_number
} else if head_number > last_processed { } else if head_number > last_processed {
info!("Backfilling Polkadot blocks {} to {head_number} before going live (resuming from checkpoint)", last_processed + 1); info!("Backfilling Polkadot blocks {} to {head_number} before going live (resuming from checkpoint)", last_processed + 1);
for n in (last_processed + 1)..=head_number { 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?; 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?; println!("Listening for deposits...\n");
last_seen = block_number;
let _guard = DB_LOCK.lock().await; let mut blocks = client.blocks().subscribe_finalized().await?;
let mut db = BridgeDatabase::load(&config.db_path)?;
db.stats.last_polkadot_block = block_number; loop {
db.save(&config.db_path)?; 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 /// 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!("\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!("detected withdrawal requests are recorded as pending_multisig_approval.\n");
let pezkuwi_client = connect_to_chain(&config.pezkuwi_rpc).await?; // Outer loop: same stall-recovery strategy as listen_deposits - see its comment for why a
let legacy_rpc = connect_legacy_rpc(&config.pezkuwi_rpc).await?; // 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. // 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 last_processed = { BridgeDatabase::load(&config.db_path)?.stats.last_pezkuwi_block };
let head_hash = legacy_rpc.chain_get_finalized_head().await?; let head_hash = legacy_rpc.chain_get_finalized_head().await?;
let head_block = pezkuwi_client.blocks().at(head_hash).await?; let head_block = pezkuwi_client.blocks().at(head_hash).await?;
let head_number = head_block.number() as u64; let head_number = head_block.number() as u64;
let mut last_seen = if last_processed == 0 { let mut last_seen = if last_processed == 0 {
info!("No checkpoint found - starting from current head (block {head_number}), not scanning full history"); info!("No checkpoint found - starting from current head (block {head_number}), not scanning full history");
head_number head_number
} else if head_number > last_processed { } else if head_number > last_processed {
info!("Backfilling Pezkuwi blocks {} to {head_number} before going live (resuming from checkpoint)", last_processed + 1); info!("Backfilling Pezkuwi blocks {} to {head_number} before going live (resuming from checkpoint)", last_processed + 1);
for n in (last_processed + 1)..=head_number { 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?; 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?; println!("Listening for withdrawals...\n");
last_seen = block_number;
let _guard = DB_LOCK.lock().await; let mut blocks = pezkuwi_client.blocks().subscribe_finalized().await?;
let mut db = BridgeDatabase::load(&config.db_path)?;
db.stats.last_pezkuwi_block = block_number; loop {
db.save(&config.db_path)?; 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<()> { async fn process_withdrawals(config: &BridgeConfig) -> Result<()> {