impl/bridge_catchup_watcher.cpp¶
Implementation of the bridge catch-up scan watcher. More...
Namespaces¶
| Name |
|---|
| sgns |
| sgns::evmwatcher |
Detailed Description¶
Implementation of the bridge catch-up scan watcher.
Date: 2026-07-12 SuperGenius ([email protected]) Copyright 2026 Genius Ventures, Inc. SPDX-License-Identifier: MIT
Source code¶
#include <watcher/impl/bridge_catchup_watcher.hpp>
#include <account/BridgeEventTypes.hpp>
#include <base/parse_utility.hpp>
#include <base/rlp-logger.hpp>
#include <eth/abi_decoder.hpp>
#include <eth/eth_watch_cli.hpp>
#include <eth/json_rpc.hpp>
#include <eth/rpc_http_transport.hpp>
#include <boost/chrono.hpp>
#include <boost/thread.hpp>
#include <algorithm>
#include <set>
namespace sgns::evmwatcher
{
BridgeCatchupWatcher::BridgeCatchupWatcher( const Config &config,
MessageCallback message_callback,
ChainsProvider chains_provider,
RpcUrlResolver rpc_resolver,
BurnProcessor burn_processor ) :
watcher::MessagingWatcher( std::move( message_callback ) ),
config_( config ),
chains_provider_( std::move( chains_provider ) ),
rpc_resolver_( std::move( rpc_resolver ) ),
burn_processor_( std::move( burn_processor ) )
{
}
void BridgeCatchupWatcher::startWatching()
{
auto logger = rlp::base::createLogger( "bridge_catchup_watcher" );
logger->info( "BridgeCatchupWatcher starting: poll_interval={}s, scan_depth={}",
config_.poll_interval.count(),
config_.scan_depth );
watcher::MessagingWatcher::startWatching();
}
void BridgeCatchupWatcher::stopWatching()
{
auto logger = rlp::base::createLogger( "bridge_catchup_watcher" );
logger->info( "BridgeCatchupWatcher stopping" );
watcher::MessagingWatcher::stopWatching();
}
uint64_t BridgeCatchupWatcher::GetLastProcessedBlock( uint64_t chain_id ) const noexcept
{
std::lock_guard lock( mutex_ );
auto it = last_block_per_chain_.find( chain_id );
return ( it != last_block_per_chain_.end() ) ? it->second : 0ULL;
}
void BridgeCatchupWatcher::watch()
{
while ( running )
{
poll_once();
boost::this_thread::sleep_for( boost::chrono::seconds( config_.poll_interval.count() ) );
}
}
void BridgeCatchupWatcher::poll_once()
{
auto logger = rlp::base::createLogger( "bridge_catchup_watcher" );
// ── Compute topic0 hashes (v1 + v2) ──────────────────────────────
static const std::string kEventSigV1( kBridgeSourceBurnedSig );
static const std::string kEventSigV2( kBridgeOutInitiatedSig );
const auto topic0_hash_v1 = eth::abi::event_signature_hash( kEventSigV1 );
std::string topic0_hex_v1 = rlp::base::parse::hex_bytes( topic0_hash_v1.data(), topic0_hash_v1.size() );
const auto topic0_hash_v2 = eth::abi::event_signature_hash( kEventSigV2 );
std::string topic0_hex_v2 = rlp::base::parse::hex_bytes( topic0_hash_v2.data(), topic0_hash_v2.size() );
// ── Snapshot current chains ──────────────────────────────────────
const std::vector<ChainContractPair> chains = chains_provider_();
if ( chains.empty() )
{
return;
}
size_t total_backfilled = 0;
size_t total_skipped = 0;
size_t chains_scanned = 0;
for ( const auto &chain_entry : chains )
{
const std::string chain_id_str = std::to_string( chain_entry.chain_id );
auto rpc_url = rpc_resolver_( chain_id_str );
if ( !rpc_url.has_value() )
{
logger->debug( "CatchUpScan: no RPC endpoint for chain {} (id={}) — skipping",
chain_entry.chain_name,
chain_entry.chain_id );
continue;
}
// RPC transport with 10-second timeout
eth::rpc::RpcHttpTransportOptions opts;
opts.timeout = std::chrono::seconds( 10 );
eth::rpc::RpcHttpTransport transport( *rpc_url, opts );
++chains_scanned;
// Parse contract address
rlp::Address contract_addr{};
if ( !rlp::base::parse::hex_array( chain_entry.contract_address, contract_addr ) )
{
logger->warn( "CatchUpScan: invalid bridge address {} for chain {}",
chain_entry.contract_address,
chain_entry.chain_name );
continue;
}
// Parse topic0 hashes to Hash256 for EventFilter
rlp::Hash256 topic0_hash256_v1{};
if ( !rlp::base::parse::hex_array( topic0_hex_v1, topic0_hash256_v1 ) )
{
logger->warn( "CatchUpScan: invalid v1 topic0 for chain {}", chain_entry.chain_name );
continue;
}
rlp::Hash256 topic0_hash256_v2{};
const bool has_v2_topic0 = rlp::base::parse::hex_array( topic0_hex_v2, topic0_hash256_v2 );
// ── Query current block number ────────────────────────────────
constexpr uint64_t kBlockNumberRequestId = 99;
auto block_number_req = eth::rpc::make_get_block_by_number_request(
eth::rpc::RpcBlockTag::kLatest, kBlockNumberRequestId );
auto block_number_resp = transport.call( block_number_req );
uint64_t current_block = 0;
if ( block_number_resp.has_value() )
{
auto parsed_block = eth::rpc::parse_block_number_response( *block_number_resp );
if ( parsed_block.has_value() )
{
current_block = *parsed_block;
}
}
if ( current_block == 0 )
{
logger->warn( "CatchUpScan: failed to query block number for chain {} — skipping",
chain_entry.chain_name );
continue;
}
// ── Compute block range ──────────────────────────────────────
uint64_t from_block = 0;
{
std::lock_guard lock( mutex_ );
auto it = last_block_per_chain_.find( chain_entry.chain_id );
if ( it != last_block_per_chain_.end() && it->second > 0 )
{
// Subsequent poll: continue from last processed block
from_block = it->second;
}
else
{
// First poll: scan depth from latest
from_block = ( current_block > config_.scan_depth )
? ( current_block - config_.scan_depth )
: 0ULL;
}
}
if ( from_block > current_block )
{
continue; // Nothing new to scan
}
const uint64_t to_block = current_block;
logger->debug( "CatchUpScan: scanning chain {} blocks {}-{} (current={})",
chain_entry.chain_name,
from_block,
to_block,
current_block );
// ── Build v1 EventFilter ──────────────────────────────────────
eth::EventFilter filter_v1;
filter_v1.addresses.push_back( contract_addr );
filter_v1.topics.push_back( topic0_hash256_v1 );
// ── Shared tx_hash dedup across v1 + v2 ──────────────────────
std::set<std::string> seen_tx_hashes;
// ── Helper: process one batch of logs ─────────────────────────
auto process_logs = [&]( const std::vector<eth::rpc::RpcLog> &rpc_logs, bool is_v2 )
{
for ( const auto &rpc_log : rpc_logs )
{
std::string tx_hash_hex = rlp::base::parse::hex_bytes( rpc_log.tx_hash.data(),
rpc_log.tx_hash.size() );
if ( !seen_tx_hashes.insert( tx_hash_hex ).second )
{
++total_skipped;
logger->debug( "CatchUpScan: burn tx {} already seen this scan — skipping", tx_hash_hex );
continue;
}
// Decode the full log entry into ABI values
const std::string &event_sig = is_v2 ? kEventSigV2 : kEventSigV1;
const auto all_params = eth::cli::event_registry().params_for( event_sig );
auto decoded = eth::abi::decode_log( rpc_log.log, event_sig, all_params );
if ( !decoded.has_value() )
{
++total_skipped;
logger->warn( "CatchUpScan: failed to decode log for tx {} — skipping", tx_hash_hex );
continue;
}
try
{
const bool processed = burn_processor_( decoded.value(), tx_hash_hex, chain_id_str );
if ( processed )
{
++total_backfilled;
logger->info( "CatchUpScan: backfilled historical burn {} on chain {}",
tx_hash_hex,
chain_entry.chain_name );
}
else
{
logger->debug( "CatchUpScan: burn processor returned false for tx {} — "
"likely already processed",
tx_hash_hex );
++total_skipped;
}
}
catch ( const std::exception &e )
{
logger->debug( "CatchUpScan: burn processor threw for tx {}: {} — skipping",
tx_hash_hex,
e.what() );
++total_skipped;
}
}
};
// ── v1 query (BridgeSourceBurned) ─────────────────────────────
constexpr uint64_t kV1LogsRequestId = 1;
auto v1_request = eth::rpc::make_get_logs_request( filter_v1,
from_block,
to_block,
kV1LogsRequestId );
auto v1_response = transport.call( v1_request );
if ( !v1_response.has_value() )
{
logger->warn( "CatchUpScan: v1 RPC call failed for chain {} (timeout/refused)",
chain_entry.chain_name );
}
else
{
auto v1_logs = eth::rpc::parse_get_logs_response( *v1_response );
if ( !v1_logs.has_value() )
{
logger->warn( "CatchUpScan: failed to parse v1 getLogs response for chain {}",
chain_entry.chain_name );
logger->debug( "CatchUpScan: v1 response preview (first {} chars): {}",
200,
v1_response->substr( 0, 200 ) );
}
else
{
process_logs( v1_logs.value(), /*is_v2=*/false );
}
}
// ── v2 query (BridgeOutInitiated) ─────────────────────────────
if ( has_v2_topic0 )
{
eth::EventFilter filter_v2;
filter_v2.addresses.push_back( contract_addr );
filter_v2.topics.push_back( topic0_hash256_v2 );
constexpr uint64_t kV2LogsRequestId = 2;
auto v2_request = eth::rpc::make_get_logs_request( filter_v2,
from_block,
to_block,
kV2LogsRequestId );
auto v2_response = transport.call( v2_request );
if ( !v2_response.has_value() )
{
logger->warn( "CatchUpScan: v2 RPC call failed for chain {} (timeout/refused)",
chain_entry.chain_name );
}
else
{
auto v2_logs = eth::rpc::parse_get_logs_response( *v2_response );
if ( !v2_logs.has_value() )
{
logger->warn( "CatchUpScan: failed to parse v2 getLogs response for chain {}",
chain_entry.chain_name );
logger->debug( "CatchUpScan: v2 response preview (first {} chars): {}",
200,
v2_response->substr( 0, 200 ) );
}
else
{
process_logs( v2_logs.value(), /*is_v2=*/true );
}
}
}
// ── Update per-chain last block ──────────────────────────────
{
std::lock_guard lock( mutex_ );
last_block_per_chain_[chain_entry.chain_id] = to_block + 1;
}
}
if ( chains_scanned > 0 )
{
logger->info( "CatchUpScan: scanned {} chains — {} historical burns backfilled, "
"{} skipped (already processed)",
chains_scanned,
total_backfilled,
total_skipped );
}
}
} // namespace sgns::evmwatcher
Updated on 2026-07-12 at 22:42:54 -0700