From 2535ca61712c169f700d5486547a5e1e1e91ff9f Mon Sep 17 00:00:00 2001 From: hellojason3 Date: Sat, 5 Sep 2026 15:05:17 +0800 Subject: [PATCH 1/2] fix(relayer): isolate withdrawal liquidity failures --- .../src/bridge/claim_withdrawals.rs | 21 ++++++++++++++++--- 1 file changed, 18 insertions(+), 3 deletions(-) diff --git a/psy_cli/psy_relayer_cli/src/bridge/claim_withdrawals.rs b/psy_cli/psy_relayer_cli/src/bridge/claim_withdrawals.rs index d5e2790f..4e1d7318 100644 --- a/psy_cli/psy_relayer_cli/src/bridge/claim_withdrawals.rs +++ b/psy_cli/psy_relayer_cli/src/bridge/claim_withdrawals.rs @@ -634,9 +634,24 @@ pub async fn submit_batch( ); if token_addr != Address::ZERO { if !bridge_erc20_liquidity_remaining.contains_key(&token_addr) { - let balance = erc20_balance_of(&provider, token_addr, bridge) - .await - .with_context(|| format!("failed to read bridge ERC20 liquidity for token {token_addr}"))?; + let balance = match erc20_balance_of(&provider, token_addr, bridge).await { + Ok(balance) => balance, + Err(err) => { + let reason = format!( + "failed to read bridge ERC20 liquidity for token {token_addr}: {err}" + ); + failure_reasons.insert(w.leaf_hash.clone(), reason); + tracing::error!( + index = i, + recipient = %recipient_addr, + token = %token_addr, + leaf_hash = %w.leaf_hash, + error = %err, + "failed to read bridge ERC20 liquidity; deferring only this withdrawal" + ); + continue; + } + }; bridge_erc20_liquidity_remaining.insert(token_addr, balance); } let remaining = bridge_erc20_liquidity_remaining From 2b79c131424edc4b22a1ac3fc3a372a2db2e01d2 Mon Sep 17 00:00:00 2001 From: hellojason3 Date: Sat, 5 Sep 2026 17:22:28 +0800 Subject: [PATCH 2/2] fix(relayer): scope withdrawal scans by L1 chain --- psy_cli/psy_relayer_cli/src/bridge/daemon.rs | 7 +- .../src/bridge/propose_withdrawals.rs | 108 ++++++++++++++++-- 2 files changed, 103 insertions(+), 12 deletions(-) diff --git a/psy_cli/psy_relayer_cli/src/bridge/daemon.rs b/psy_cli/psy_relayer_cli/src/bridge/daemon.rs index bd4aa24b..b487355e 100644 --- a/psy_cli/psy_relayer_cli/src/bridge/daemon.rs +++ b/psy_cli/psy_relayer_cli/src/bridge/daemon.rs @@ -2434,11 +2434,16 @@ async fn build_multichain_l2_plan( let global_count = provider .get_withdrawal_tree_global_count(checkpoint, BRIDGE_USER_ID_U64) .await?; - propose_withdrawals::fetch_pending_bridge_withdrawals( + let destination_chain_indexes = chains + .iter() + .map(|chain| u64::from(chain.chain_index)) + .collect::>(); + propose_withdrawals::fetch_pending_bridge_withdrawals_for_chains( propose_args, from_checkpoint.max(1), to_checkpoint.saturating_add(1), global_count, + &destination_chain_indexes, ).await? } else { Vec::new() diff --git a/psy_cli/psy_relayer_cli/src/bridge/propose_withdrawals.rs b/psy_cli/psy_relayer_cli/src/bridge/propose_withdrawals.rs index df42591d..6aefafe0 100644 --- a/psy_cli/psy_relayer_cli/src/bridge/propose_withdrawals.rs +++ b/psy_cli/psy_relayer_cli/src/bridge/propose_withdrawals.rs @@ -1,4 +1,4 @@ -use std::time::{Duration, Instant}; +use std::{collections::BTreeMap, time::{Duration, Instant}}; use clap::Args; use parth_core::pgoldilocks::QHashOut; @@ -265,15 +265,50 @@ async fn fetch_bridge_withdrawals( http: &reqwest::Client, base_url: &str, initial_event_offset: u64, + destination_chain_indexes: &[u64], ) -> anyhow::Result> { - let mut all_withdrawals: Vec = Vec::new(); - let limit: u32 = 10_000; + let chain_filters = if destination_chain_indexes.is_empty() { + vec![None] + } else { + destination_chain_indexes.iter().copied().map(Some).collect() + }; + let mut withdrawals_by_event = BTreeMap::new(); + + for destination_chain_index in chain_filters { + for withdrawal in fetch_bridge_withdrawals_for_chain( + http, + base_url, + initial_event_offset, + destination_chain_index, + ) + .await? + { + withdrawals_by_event.insert(withdrawal.event_id, withdrawal); + } + } + + Ok(withdrawals_by_event.into_values().collect()) +} + +async fn fetch_bridge_withdrawals_for_chain( + http: &reqwest::Client, + base_url: &str, + initial_event_offset: u64, + destination_chain_index: Option, +) -> anyhow::Result> { + let mut all_withdrawals = Vec::new(); + // psy-services caps this endpoint at 1,000 rows even if a larger limit is + // requested. Advance by that server-side page size so a busy withdrawal + // stream cannot be truncated silently. + let limit: u32 = 1_000; let mut offset: u64 = initial_event_offset; loop { - let url = format!( - "{}/api/v1/bridge/withdrawals?limit={}&offset={}", - base_url, limit, offset, + let url = bridge_withdrawals_url( + base_url, + limit, + offset, + destination_chain_index, ); tracing::debug!(url = %url, "fetching bridge withdrawals page"); @@ -308,18 +343,35 @@ async fn fetch_bridge_withdrawals( } let data = resp.data.ok_or_else(|| anyhow::anyhow!("psy-services returned success but no data"))?; - let page_len = data.withdrawals.len() as u64; all_withdrawals.extend(data.withdrawals); - if page_len < limit as u64 || all_withdrawals.len() as i64 >= data.total { + let total = u64::try_from(data.total.max(0)).unwrap_or(u64::MAX); + let next_offset = offset.saturating_add(u64::from(limit)); + if next_offset >= total { break; } - offset += page_len; + offset = next_offset; } Ok(all_withdrawals) } +fn bridge_withdrawals_url( + base_url: &str, + limit: u32, + offset: u64, + destination_chain_index: Option, +) -> String { + let mut url = format!( + "{}/api/v1/bridge/withdrawals?limit={}&offset={}", + base_url, limit, offset, + ); + if let Some(chain_index) = destination_chain_index { + url.push_str(&format!("&destination_chain_index={chain_index}")); + } + url +} + async fn fetch_services_latest_checkpoint(http: &reqwest::Client, base_url: &str) -> anyhow::Result { let url = format!("{}/api/v1/checkpoint/latest", base_url); let resp: ApiResponse = get_services_json(http, &url, "latest_checkpoint").await?; @@ -448,13 +500,35 @@ pub async fn fetch_pending_bridge_withdrawals( _from_checkpoint: u64, _to_checkpoint_exclusive: u64, initial_event_offset: u64, +) -> anyhow::Result> { + fetch_pending_bridge_withdrawals_for_chains( + args, + _from_checkpoint, + _to_checkpoint_exclusive, + initial_event_offset, + &[], + ) + .await +} + +pub async fn fetch_pending_bridge_withdrawals_for_chains( + args: &ProposeWithdrawalsArgs, + _from_checkpoint: u64, + _to_checkpoint_exclusive: u64, + initial_event_offset: u64, + destination_chain_indexes: &[u64], ) -> anyhow::Result> { let psy_config = psy_config::PsyConfigGoldilocks::from_file(&args.rpc_config)?; let http = build_default_http_client()?; let services_url = resolve_services_url(&args.services_url, &psy_config)?; - let mut service_withdrawals = - fetch_bridge_withdrawals(&http, &services_url, initial_event_offset).await?; + let mut service_withdrawals = fetch_bridge_withdrawals( + &http, + &services_url, + initial_event_offset, + destination_chain_indexes, + ) + .await?; service_withdrawals.sort_by_key(|w| w.event_id); tracing::info!( bridge_withdrawals = service_withdrawals.len(), @@ -797,4 +871,16 @@ mod tests { Ok(()) } + + #[test] + fn multichain_withdrawal_query_selects_destination_chain() { + assert_eq!( + bridge_withdrawals_url("https://services.example", 1_000, 42, Some(2)), + "https://services.example/api/v1/bridge/withdrawals?limit=1000&offset=42&destination_chain_index=2" + ); + assert_eq!( + bridge_withdrawals_url("https://services.example", 1_000, 42, None), + "https://services.example/api/v1/bridge/withdrawals?limit=1000&offset=42" + ); + } }