diff --git a/Cargo.lock b/Cargo.lock index d0f0f7e68e9..7ee515b0b4d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -10620,6 +10620,34 @@ dependencies = [ "tracing", ] +[[package]] +name = "reth-xpl-webhook" +version = "1.11.3" +dependencies = [ + "alloy-consensus", + "alloy-evm", + "alloy-primitives", + "clap", + "eyre", + "futures", + "hmac", + "metrics", + "rand 0.9.2", + "reqwest 0.13.2", + "reth-ethereum", + "reth-exex", + "reth-libmdbx", + "reth-revm", + "reth-storage-api", + "revm-inspectors", + "serde", + "serde_json", + "sha2", + "tempfile", + "tokio", + "tracing", +] + [[package]] name = "reth-zstd-compressors" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index f0753d588c1..bc3901cd452 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,6 +10,7 @@ exclude = [".github/"] [workspace] members = [ "bin/reth-bench/", + "bin/reth-xpl-webhook/", "bin/reth/", "crates/storage/rpc-provider/", "crates/chain-state/", diff --git a/bin/reth-xpl-webhook/Cargo.toml b/bin/reth-xpl-webhook/Cargo.toml new file mode 100644 index 00000000000..24055dbd1d4 --- /dev/null +++ b/bin/reth-xpl-webhook/Cargo.toml @@ -0,0 +1,54 @@ +[package] +name = "reth-xpl-webhook" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +homepage.workspace = true +repository.workspace = true +exclude.workspace = true +publish = false + +[lints] +workspace = true + +[[bin]] +name = "reth-xpl-webhook" +path = "src/main.rs" + +[dependencies] +# reth +reth-ethereum = { workspace = true, features = ["full", "cli", "node-api"] } +reth-exex.workspace = true +reth-libmdbx.workspace = true +reth-revm.workspace = true +reth-storage-api.workspace = true + +# revm / evm +revm-inspectors.workspace = true +alloy-evm.workspace = true + +# alloy +alloy-primitives.workspace = true +alloy-consensus.workspace = true + +# crypto +hmac.workspace = true +sha2.workspace = true + +# async + http +tokio = { workspace = true, features = ["sync", "rt-multi-thread", "macros", "time"] } +futures.workspace = true +reqwest = { workspace = true, features = ["json"] } + +# misc +eyre.workspace = true +tracing.workspace = true +metrics.workspace = true +serde = { workspace = true, features = ["derive"] } +serde_json = { workspace = true, features = ["std"] } +clap = { workspace = true, features = ["derive", "env"] } +rand.workspace = true + +[dev-dependencies] +tempfile = "3" diff --git a/bin/reth-xpl-webhook/src/args.rs b/bin/reth-xpl-webhook/src/args.rs new file mode 100644 index 00000000000..d1d3a3a6356 --- /dev/null +++ b/bin/reth-xpl-webhook/src/args.rs @@ -0,0 +1,44 @@ +use alloy_primitives::Address; +use clap::Parser; +use std::path::PathBuf; + +/// CLI extension flags appended to the standard `reth` argument set. Each +/// argument ID is namespaced with the `xpl_webhook_` prefix so it doesn't +/// collide with the built-in node flags (e.g. era `--url`). +#[derive(Debug, Clone, Parser)] +pub struct XplWebhookArgs { + /// Webhook URL that receives `{ result: Transfer[] }` POSTs. + #[arg(id = "xpl_webhook_url", long = "xpl-webhook-url", env = "XPL_WEBHOOK_URL")] + pub url: String, + + /// HMAC-SHA256 token used to sign requests. Read from `XPL_WEBHOOK_TOKEN` + /// when the flag is omitted. + #[arg(id = "xpl_webhook_token", long = "xpl-webhook-token", env = "XPL_WEBHOOK_TOKEN")] + pub token: String, + + /// Address emitted in the `address` field of every transfer row. Defaults + /// to the zero address (native XPL marker). + #[arg( + id = "xpl_webhook_native_address", + long = "xpl-webhook-native-address", + default_value = "0x0000000000000000000000000000000000000000" + )] + pub native_address: Address, + + /// Path to the MDBX outbox directory. Defaults to `/xpl-webhook`. + #[arg(id = "xpl_webhook_outbox_path", long = "xpl-webhook-outbox-path")] + pub outbox_path: Option, + + /// Maximum number of HTTP delivery attempts per batch before backing off + /// to the configured ceiling. + #[arg(id = "xpl_webhook_max_retries", long = "xpl-webhook-max-retries", default_value_t = 8)] + pub max_retries: u32, + + /// HTTP request timeout in milliseconds. + #[arg( + id = "xpl_webhook_http_timeout_ms", + long = "xpl-webhook-http-timeout-ms", + default_value_t = 5_000 + )] + pub http_timeout_ms: u64, +} diff --git a/bin/reth-xpl-webhook/src/exex.rs b/bin/reth-xpl-webhook/src/exex.rs new file mode 100644 index 00000000000..43db1130da9 --- /dev/null +++ b/bin/reth-xpl-webhook/src/exex.rs @@ -0,0 +1,92 @@ +use crate::{metrics as m, outbox::Outbox, replay::replay_block}; +use alloy_consensus::BlockHeader; +use alloy_primitives::Address; +use futures::TryStreamExt; +use reth_ethereum::{ + exex::{ExExContext, ExExEvent, ExExNotification}, + node::api::FullNodeComponents, +}; +use tracing::{debug, error, info, warn}; + +/// The ExEx future. Loops over canonical chain notifications, replays each +/// committed block, persists rows to the outbox, and signals +/// [`ExExEvent::FinishedHeight`] only after the rows are durable. +pub async fn xpl_webhook_exex( + mut ctx: ExExContext, + outbox: Outbox, + native_address: Address, +) -> eyre::Result<()> { + info!(target: "xpl-webhook", "ExEx started"); + + while let Some(notification) = ctx.notifications.try_next().await? { + match ¬ification { + ExExNotification::ChainCommitted { new } => { + for block in new.blocks_iter() { + let block_number = block.number(); + match replay_block(&ctx, block, native_address) { + Ok(rows) => { + metrics::counter!(m::BLOCKS_PROCESSED).increment(1); + if rows.is_empty() { + continue; + } + metrics::counter!(m::ROWS_EMITTED).increment(rows.len() as u64); + debug!( + target: "xpl-webhook", + block_number, + rows = rows.len(), + "persisting batch", + ); + if let Err(err) = + outbox.append_block_batch(block.hash(), block_number, rows) + { + error!(target: "xpl-webhook", %err, "outbox write failed"); + } + } + Err(err) => { + metrics::counter!(m::REPLAY_ERRORS).increment(1); + error!(target: "xpl-webhook", block_number, %err, "replay failed"); + } + } + } + } + ExExNotification::ChainReorged { old, new } => { + metrics::counter!(m::REORGS).increment(1); + warn!( + target: "xpl-webhook", + old = ?old.range(), + new = ?new.range(), + "reorg: discarding pending rows for old chain segment", + ); + if let Err(err) = outbox.discard_blocks(old.blocks_iter().map(|b| b.hash())) { + error!(target: "xpl-webhook", %err, "failed to discard reorged blocks"); + } + for block in new.blocks_iter() { + let block_number = block.number(); + if let Ok(rows) = replay_block(&ctx, block, native_address) && + !rows.is_empty() + { + let _ = outbox.append_block_batch(block.hash(), block_number, rows); + } + } + } + ExExNotification::ChainReverted { old } => { + metrics::counter!(m::REVERTS).increment(1); + warn!( + target: "xpl-webhook", + old = ?old.range(), + "revert: discarding pending rows", + ); + if let Err(err) = outbox.discard_blocks(old.blocks_iter().map(|b| b.hash())) { + error!(target: "xpl-webhook", %err, "failed to discard reverted blocks"); + } + } + } + + if let Some(committed) = notification.committed_chain() { + ctx.events.send(ExExEvent::FinishedHeight(committed.tip().num_hash()))?; + } + } + + info!(target: "xpl-webhook", "ExEx exiting"); + Ok(()) +} diff --git a/bin/reth-xpl-webhook/src/extractor.rs b/bin/reth-xpl-webhook/src/extractor.rs new file mode 100644 index 00000000000..1d70f4f3d27 --- /dev/null +++ b/bin/reth-xpl-webhook/src/extractor.rs @@ -0,0 +1,260 @@ +//! Walks a [`CallTraceArena`] and emits one [`Transfer`] row per native-value +//! movement, plus an optional synthetic fee row per transaction. + +use crate::payload::{HexQuantity, Transfer, TransferName}; +use alloy_primitives::{Address, B256, U256}; +use revm_inspectors::tracing::{types::CallKind, CallTraceArena}; + +/// Per-block synthetic counter used as the `logIndex` for emitted rows. Native +/// value movements have no real log index, so we keep a deterministic order +/// across the entire block. +#[derive(Debug, Default)] +pub struct LogIndexCounter(u64); + +impl LogIndexCounter { + pub const fn next(&mut self) -> u64 { + let v = self.0; + self.0 += 1; + v + } +} + +/// Inputs needed to build the constant fields of every emitted row for a +/// transaction. +pub struct TxRowContext { + pub native_address: Address, + pub block_hash: B256, + pub block_number: u64, + pub transaction_hash: B256, + pub tx_signer: Address, +} + +/// Push rows for every value-moving frame in `arena`. Skips delegate/static +/// frames and zero-value frames. +pub fn extract_value_transfers( + arena: &CallTraceArena, + ctx: &TxRowContext, + counter: &mut LogIndexCounter, + out: &mut Vec, +) { + for node in arena.nodes() { + let trace = &node.trace; + if trace.value.is_zero() { + continue; + } + // CALLCODE/DELEGATECALL/STATICCALL/AUTHCALL do not move native value + // (DELEGATECALL inherits its parent's value field without transferring). + let (from, to) = match trace.kind { + CallKind::Call | CallKind::Create | CallKind::Create2 => (trace.caller, trace.address), + _ => continue, + }; + out.push(Transfer { + address: ctx.native_address, + block_hash: ctx.block_hash, + block_number: HexQuantity(ctx.block_number), + from, + log_index: HexQuantity(counter.next()), + name: TransferName::Xpl, + to, + transaction_hash: ctx.transaction_hash, + value: trace.value.into(), + }); + } +} + +/// Append a synthetic gas-fee row using `gas_used * effective_gas_price` if +/// non-zero. +pub fn push_fee_row( + ctx: &TxRowContext, + gas_used: u64, + effective_gas_price: u128, + counter: &mut LogIndexCounter, + out: &mut Vec, +) { + let fee = U256::from(gas_used).saturating_mul(U256::from(effective_gas_price)); + if fee.is_zero() { + return; + } + out.push(Transfer { + address: ctx.native_address, + block_hash: ctx.block_hash, + block_number: HexQuantity(ctx.block_number), + from: ctx.tx_signer, + log_index: HexQuantity(counter.next()), + name: TransferName::Fee, + to: ctx.native_address, + transaction_hash: ctx.transaction_hash, + value: fee.into(), + }); +} + +#[cfg(test)] +mod tests { + use super::*; + use alloy_primitives::{address, b256}; + use revm_inspectors::tracing::types::{CallTrace, CallTraceNode}; + + fn node(kind: CallKind, caller: Address, address: Address, value: U256) -> CallTraceNode { + CallTraceNode { + trace: CallTrace { kind, caller, address, value, ..Default::default() }, + ..Default::default() + } + } + + fn ctx() -> TxRowContext { + TxRowContext { + native_address: Address::ZERO, + block_hash: b256!("0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), + block_number: 100, + transaction_hash: b256!( + "0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" + ), + tx_signer: address!("0x1111111111111111111111111111111111111111"), + } + } + + fn arena_with(nodes: Vec) -> CallTraceArena { + let mut arena = CallTraceArena::default(); + let slot = arena.nodes_mut(); + *slot = nodes; + arena + } + + #[test] + fn direct_call_with_value_emits_row() { + let n = node( + CallKind::Call, + address!("0x2222222222222222222222222222222222222222"), + address!("0x3333333333333333333333333333333333333333"), + U256::from(42), + ); + let arena = arena_with(vec![n]); + let mut out = Vec::new(); + let mut counter = LogIndexCounter::default(); + extract_value_transfers(&arena, &ctx(), &mut counter, &mut out); + assert_eq!(out.len(), 1); + assert_eq!(out[0].value.0, U256::from(42)); + assert_eq!(out[0].from, address!("0x2222222222222222222222222222222222222222")); + assert_eq!(out[0].to, address!("0x3333333333333333333333333333333333333333")); + assert_eq!(out[0].name, TransferName::Xpl); + assert_eq!(out[0].log_index, HexQuantity(0)); + } + + #[test] + fn zero_value_call_emits_nothing() { + let n = node(CallKind::Call, Address::repeat_byte(1), Address::repeat_byte(2), U256::ZERO); + let arena = arena_with(vec![n]); + let mut out = Vec::new(); + let mut counter = LogIndexCounter::default(); + extract_value_transfers(&arena, &ctx(), &mut counter, &mut out); + assert!(out.is_empty()); + } + + #[test] + fn delegate_static_callcode_authcall_skipped() { + let arena = arena_with(vec![ + node( + CallKind::DelegateCall, + Address::repeat_byte(1), + Address::repeat_byte(2), + U256::from(1), + ), + node( + CallKind::StaticCall, + Address::repeat_byte(1), + Address::repeat_byte(2), + U256::from(1), + ), + node( + CallKind::CallCode, + Address::repeat_byte(1), + Address::repeat_byte(2), + U256::from(1), + ), + node( + CallKind::AuthCall, + Address::repeat_byte(1), + Address::repeat_byte(2), + U256::from(1), + ), + ]); + let mut out = Vec::new(); + let mut counter = LogIndexCounter::default(); + extract_value_transfers(&arena, &ctx(), &mut counter, &mut out); + assert!(out.is_empty()); + } + + #[test] + fn create_with_endowment_emits_row_to_created_address() { + let n = node( + CallKind::Create, + address!("0x4444444444444444444444444444444444444444"), + address!("0x5555555555555555555555555555555555555555"), + U256::from(99), + ); + let arena = arena_with(vec![n]); + let mut out = Vec::new(); + let mut counter = LogIndexCounter::default(); + extract_value_transfers(&arena, &ctx(), &mut counter, &mut out); + assert_eq!(out.len(), 1); + assert_eq!(out[0].to, address!("0x5555555555555555555555555555555555555555")); + } + + #[test] + fn aa_style_internal_call_emits_wallet_as_from() { + // Bundler → EntryPoint → SmartAccount → recipient (the inner CALL has value). + let bundler = address!("0x9999999999999999999999999999999999999999"); + let entry = address!("0x1111111111111111111111111111111111111111"); + let wallet = address!("0x2222222222222222222222222222222222222222"); + let recipient = address!("0x3333333333333333333333333333333333333333"); + + let arena = arena_with(vec![ + node(CallKind::Call, bundler, entry, U256::ZERO), + node(CallKind::Call, entry, wallet, U256::ZERO), + node(CallKind::Call, wallet, recipient, U256::from(10_u64.pow(18))), + ]); + + let mut out = Vec::new(); + let mut counter = LogIndexCounter::default(); + extract_value_transfers(&arena, &ctx(), &mut counter, &mut out); + + assert_eq!(out.len(), 1, "only the value-moving frame emits a row"); + assert_eq!(out[0].from, wallet, "the wallet is the from, not the bundler"); + assert_eq!(out[0].to, recipient); + } + + #[test] + fn fee_row_uses_gas_used_times_effective_price() { + let mut out = Vec::new(); + let mut counter = LogIndexCounter::default(); + push_fee_row(&ctx(), 21_000, 2_000_000_000, &mut counter, &mut out); + assert_eq!(out.len(), 1); + assert_eq!(out[0].name, TransferName::Fee); + assert_eq!(out[0].value.0, U256::from(42_000_000_000_000_u128)); + assert_eq!(out[0].from, address!("0x1111111111111111111111111111111111111111")); + assert_eq!(out[0].to, Address::ZERO); + } + + #[test] + fn zero_fee_emits_no_row() { + let mut out = Vec::new(); + let mut counter = LogIndexCounter::default(); + push_fee_row(&ctx(), 0, 1_000, &mut counter, &mut out); + assert!(out.is_empty()); + } + + #[test] + fn log_index_is_monotonic_across_calls_and_fee() { + let arena = arena_with(vec![ + node(CallKind::Call, Address::repeat_byte(1), Address::repeat_byte(2), U256::from(1)), + node(CallKind::Call, Address::repeat_byte(3), Address::repeat_byte(4), U256::from(2)), + ]); + let mut out = Vec::new(); + let mut counter = LogIndexCounter::default(); + extract_value_transfers(&arena, &ctx(), &mut counter, &mut out); + push_fee_row(&ctx(), 21_000, 1_000_000_000, &mut counter, &mut out); + assert_eq!(out[0].log_index, HexQuantity(0)); + assert_eq!(out[1].log_index, HexQuantity(1)); + assert_eq!(out[2].log_index, HexQuantity(2)); + } +} diff --git a/bin/reth-xpl-webhook/src/main.rs b/bin/reth-xpl-webhook/src/main.rs new file mode 100644 index 00000000000..31f6603a53b --- /dev/null +++ b/bin/reth-xpl-webhook/src/main.rs @@ -0,0 +1,51 @@ +//! Custom Reth binary that installs an ExEx posting native XPL transfer rows to a +//! QuickNode-compatible webhook endpoint. + +#![warn(unused_crate_dependencies)] +#![allow(unreachable_pub)] + +mod args; +mod exex; +mod extractor; +mod metrics; +mod outbox; +mod payload; +mod replay; +mod webhook; + +use crate::{ + args::XplWebhookArgs, exex::xpl_webhook_exex, outbox::Outbox, webhook::WebhookDispatcher, +}; +use clap::Parser; +use reth_ethereum::{ + cli::{chainspec::EthereumChainSpecParser, interface::Cli}, + node::EthereumNode, +}; + +fn main() -> eyre::Result<()> { + Cli::::parse().run( + async move |builder, webhook_args| { + let outbox_path = webhook_args + .outbox_path + .clone() + .unwrap_or_else(|| builder.config().datadir().data_dir().join("xpl-webhook")); + + let outbox = Outbox::open(&outbox_path)?; + let dispatcher = WebhookDispatcher::spawn(webhook_args.clone(), outbox.clone()); + + let native_address = webhook_args.native_address; + let handle = builder + .node(EthereumNode::default()) + .install_exex("xpl-webhook", { + let outbox = outbox.clone(); + async move |ctx| Ok(xpl_webhook_exex(ctx, outbox, native_address)) + }) + .launch() + .await?; + + handle.wait_for_node_exit().await?; + dispatcher.shutdown().await; + Ok(()) + }, + ) +} diff --git a/bin/reth-xpl-webhook/src/metrics.rs b/bin/reth-xpl-webhook/src/metrics.rs new file mode 100644 index 00000000000..4ea9da8600e --- /dev/null +++ b/bin/reth-xpl-webhook/src/metrics.rs @@ -0,0 +1,11 @@ +//! Shared metric names so callers don't drift on string constants. + +pub const BLOCKS_PROCESSED: &str = "xpl_webhook_blocks_processed"; +pub const ROWS_EMITTED: &str = "xpl_webhook_rows_emitted"; +pub const ROWS_DELIVERED: &str = "xpl_webhook_rows_delivered"; +pub const HTTP_FAILURES: &str = "xpl_webhook_http_failures"; +pub const HTTP_SUCCESS: &str = "xpl_webhook_http_success"; +pub const DLQ: &str = "xpl_webhook_dlq"; +pub const REORGS: &str = "xpl_webhook_reorgs"; +pub const REVERTS: &str = "xpl_webhook_reverts"; +pub const REPLAY_ERRORS: &str = "xpl_webhook_replay_errors"; diff --git a/bin/reth-xpl-webhook/src/outbox.rs b/bin/reth-xpl-webhook/src/outbox.rs new file mode 100644 index 00000000000..97c59fa819e --- /dev/null +++ b/bin/reth-xpl-webhook/src/outbox.rs @@ -0,0 +1,282 @@ +use crate::payload::Transfer; +use alloy_primitives::B256; +use reth_libmdbx::{DatabaseFlags, Environment, EnvironmentKind, Geometry, PageSize, WriteFlags}; +use serde::{Deserialize, Serialize}; +use std::{path::Path, sync::Arc}; +use tokio::sync::Notify; + +const PENDING_TABLE: &str = "pending_batches"; +const CHECKPOINT_TABLE: &str = "delivery_checkpoint"; +const CHECKPOINT_KEY: &[u8] = b"checkpoint"; + +/// Default MDBX geometry: grow up to 16 GiB, start at 64 MiB. +const DB_INITIAL_BYTES: usize = 64 * 1024 * 1024; +const DB_MAX_BYTES: usize = 16 * 1024 * 1024 * 1024; +const DB_GROWTH_STEP: isize = 64 * 1024 * 1024; + +/// One persisted batch — emitted rows for a single block plus delivery +/// bookkeeping. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct PendingBatchRecord { + pub block_number: u64, + pub block_hash: B256, + pub rows: Vec, + pub delivered: bool, + pub attempts: u32, + /// Earliest unix-millis at which the dispatcher should retry this batch. + pub next_attempt_at: u64, +} + +/// Durable webhook outbox backed by a dedicated MDBX environment under the +/// node's datadir. +#[derive(Clone)] +pub struct Outbox { + inner: Arc, +} + +struct OutboxInner { + env: Environment, + pending_dbi: reth_libmdbx::ffi::MDBX_dbi, + checkpoint_dbi: reth_libmdbx::ffi::MDBX_dbi, + wakeup: Arc, +} + +impl Outbox { + /// Open or create the outbox env at `path`. + pub fn open(path: &Path) -> eyre::Result { + std::fs::create_dir_all(path)?; + + let env = Environment::builder() + .set_kind(EnvironmentKind::Default) + .set_max_dbs(8) + .set_geometry(Geometry { + size: Some(DB_INITIAL_BYTES..DB_MAX_BYTES), + growth_step: Some(DB_GROWTH_STEP), + shrink_threshold: None, + page_size: Some(PageSize::MinimalAcceptable), + }) + .open(path)?; + + let (pending_dbi, checkpoint_dbi) = { + let tx = env.begin_rw_txn()?; + let pending = tx.create_db(Some(PENDING_TABLE), DatabaseFlags::default())?; + let checkpoint = tx.create_db(Some(CHECKPOINT_TABLE), DatabaseFlags::default())?; + let p = pending.dbi(); + let c = checkpoint.dbi(); + tx.commit()?; + (p, c) + }; + + Ok(Self { + inner: Arc::new(OutboxInner { + env, + pending_dbi, + checkpoint_dbi, + wakeup: Arc::new(Notify::new()), + }), + }) + } + + /// Returns a handle the dispatcher can `notified()` on to learn that new + /// rows are pending. + pub fn wakeup_handle(&self) -> Arc { + self.inner.wakeup.clone() + } + + /// Persist a non-empty batch of rows for `block_hash` and wake any waiter. + pub fn append_block_batch( + &self, + block_hash: B256, + block_number: u64, + rows: Vec, + ) -> eyre::Result<()> { + let record = PendingBatchRecord { + block_number, + block_hash, + rows, + delivered: false, + attempts: 0, + next_attempt_at: 0, + }; + self.put_record(&record)?; + self.inner.wakeup.notify_waiters(); + Ok(()) + } + + /// Remove pending (not-yet-delivered) batches for the given block hashes. + /// Used to drop reverted blocks on reorg. + pub fn discard_blocks(&self, hashes: impl IntoIterator) -> eyre::Result<()> { + let tx = self.inner.env.begin_rw_txn()?; + for hash in hashes { + let cursor = tx.cursor_with_dbi(self.inner.pending_dbi)?; + if let Some((key, value)) = find_by_hash(cursor, hash)? { + let record: PendingBatchRecord = serde_json::from_slice(&value)?; + if !record.delivered { + tx.del(self.inner.pending_dbi, &key, None)?; + } + } + } + tx.commit()?; + Ok(()) + } + + /// Return the oldest pending batch whose `next_attempt_at <= now_ms`, or + /// `None` if the outbox is empty or only contains future-scheduled retries. + pub fn next_pending(&self, now_ms: u64) -> eyre::Result> { + let tx = self.inner.env.begin_ro_txn()?; + let mut cursor = tx.cursor_with_dbi(self.inner.pending_dbi)?; + let mut entry: Option<(Vec, Vec)> = cursor.first()?; + while let Some((_, value)) = entry.as_ref() { + let record: PendingBatchRecord = serde_json::from_slice(value)?; + if !record.delivered && record.next_attempt_at <= now_ms { + return Ok(Some(record)); + } + entry = cursor.next()?; + } + Ok(None) + } + + /// Mark a batch as delivered and advance the checkpoint to its height. + pub fn mark_delivered(&self, block_hash: B256) -> eyre::Result<()> { + let tx = self.inner.env.begin_rw_txn()?; + let cursor = tx.cursor_with_dbi(self.inner.pending_dbi)?; + if let Some((key, value)) = find_by_hash(cursor, block_hash)? { + let record: PendingBatchRecord = serde_json::from_slice(&value)?; + let checkpoint = serde_json::to_vec(&record.block_number)?; + tx.put(self.inner.checkpoint_dbi, CHECKPOINT_KEY, &checkpoint, WriteFlags::UPSERT)?; + // Once delivered, drop the row to keep the outbox bounded. + tx.del(self.inner.pending_dbi, &key, None)?; + } + tx.commit()?; + Ok(()) + } + + /// Bump the attempt counter and reschedule a failed batch. + pub fn record_attempt(&self, block_hash: B256, next_attempt_at: u64) -> eyre::Result<()> { + let tx = self.inner.env.begin_rw_txn()?; + let cursor = tx.cursor_with_dbi(self.inner.pending_dbi)?; + if let Some((key, value)) = find_by_hash(cursor, block_hash)? { + let mut record: PendingBatchRecord = serde_json::from_slice(&value)?; + record.attempts = record.attempts.saturating_add(1); + record.next_attempt_at = next_attempt_at; + let encoded = serde_json::to_vec(&record)?; + tx.put(self.inner.pending_dbi, key, &encoded, WriteFlags::UPSERT)?; + } + tx.commit()?; + Ok(()) + } + + fn put_record(&self, record: &PendingBatchRecord) -> eyre::Result<()> { + let key = batch_key(record.block_number, record.block_hash); + let encoded = serde_json::to_vec(record)?; + let tx = self.inner.env.begin_rw_txn()?; + tx.put(self.inner.pending_dbi, key, &encoded, WriteFlags::UPSERT)?; + tx.commit()?; + Ok(()) + } +} + +fn batch_key(block_number: u64, block_hash: B256) -> [u8; 40] { + let mut key = [0_u8; 40]; + key[..8].copy_from_slice(&block_number.to_be_bytes()); + key[8..].copy_from_slice(block_hash.as_slice()); + key +} + +fn find_by_hash( + mut cursor: reth_libmdbx::Cursor, + target: B256, +) -> eyre::Result, Vec)>> { + let mut entry: Option<(Vec, Vec)> = cursor.first()?; + while let Some((key, _)) = entry.as_ref() { + if key.len() == 40 && key[8..] == target.as_slice()[..] { + return Ok(entry); + } + entry = cursor.next()?; + } + Ok(None) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::payload::{HexQuantity, TransferName}; + use alloy_primitives::{b256, Address, U256}; + use tempfile::tempdir; + + fn sample_batch(block: u64) -> (B256, u64, Vec) { + let hash = B256::with_last_byte(block as u8); + let row = Transfer { + address: Address::ZERO, + block_hash: hash, + block_number: HexQuantity(block), + from: Address::ZERO, + log_index: HexQuantity(0), + name: TransferName::Xpl, + to: Address::ZERO, + transaction_hash: b256!( + "0xcccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc" + ), + value: U256::from(1).into(), + }; + (hash, block, vec![row]) + } + + #[test] + fn round_trip_append_and_next_pending() { + let dir = tempdir().unwrap(); + let outbox = Outbox::open(dir.path()).unwrap(); + let (hash, num, rows) = sample_batch(7); + outbox.append_block_batch(hash, num, rows).unwrap(); + let pending = outbox.next_pending(0).unwrap().expect("pending exists"); + assert_eq!(pending.block_number, 7); + assert_eq!(pending.block_hash, hash); + assert_eq!(pending.rows.len(), 1); + assert!(!pending.delivered); + } + + #[test] + fn mark_delivered_removes_from_pending() { + let dir = tempdir().unwrap(); + let outbox = Outbox::open(dir.path()).unwrap(); + let (hash, num, rows) = sample_batch(3); + outbox.append_block_batch(hash, num, rows).unwrap(); + outbox.mark_delivered(hash).unwrap(); + assert!(outbox.next_pending(0).unwrap().is_none()); + } + + #[test] + fn discard_blocks_drops_undelivered() { + let dir = tempdir().unwrap(); + let outbox = Outbox::open(dir.path()).unwrap(); + let (hash, num, rows) = sample_batch(1); + outbox.append_block_batch(hash, num, rows).unwrap(); + outbox.discard_blocks([hash]).unwrap(); + assert!(outbox.next_pending(0).unwrap().is_none()); + } + + #[test] + fn next_pending_respects_next_attempt_at() { + let dir = tempdir().unwrap(); + let outbox = Outbox::open(dir.path()).unwrap(); + let (hash, num, rows) = sample_batch(2); + outbox.append_block_batch(hash, num, rows).unwrap(); + outbox.record_attempt(hash, 1_000_000).unwrap(); + assert!(outbox.next_pending(0).unwrap().is_none()); + let r = outbox.next_pending(1_000_000).unwrap().expect("eligible after deadline"); + assert_eq!(r.attempts, 1); + } + + #[test] + fn restart_preserves_pending() { + let dir = tempdir().unwrap(); + let (hash, num, rows) = sample_batch(11); + { + let outbox = Outbox::open(dir.path()).unwrap(); + outbox.append_block_batch(hash, num, rows).unwrap(); + } + let reopened = Outbox::open(dir.path()).unwrap(); + let pending = reopened.next_pending(0).unwrap().expect("survives reopen"); + assert_eq!(pending.block_number, 11); + } +} diff --git a/bin/reth-xpl-webhook/src/payload.rs b/bin/reth-xpl-webhook/src/payload.rs new file mode 100644 index 00000000000..c42c15d9964 --- /dev/null +++ b/bin/reth-xpl-webhook/src/payload.rs @@ -0,0 +1,142 @@ +use alloy_primitives::{Address, B256, U256}; +use serde::{Deserialize, Serialize}; + +/// QuickNode-compatible transfer row. Field set matches the existing +/// `apps/tasks` controller exactly (hex quantity strings, decimal `value`). +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct Transfer { + /// Token contract address. Always the configured native marker + /// (`0x0000...0000`) for XPL. + pub address: Address, + #[serde(rename = "blockHash")] + pub block_hash: B256, + #[serde(rename = "blockNumber")] + pub block_number: HexQuantity, + pub from: Address, + #[serde(rename = "logIndex")] + pub log_index: HexQuantity, + pub name: TransferName, + pub to: Address, + #[serde(rename = "transactionHash")] + pub transaction_hash: B256, + pub value: DecimalU256, +} + +/// Wrapper carrying the controller's outer shape: `{ "result": Transfer[] }`. +#[derive(Debug, Clone, Serialize)] +pub struct WebhookBody<'a> { + pub result: &'a [Transfer], +} + +/// Discriminator emitted in the `name` field. +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +pub enum TransferName { + #[serde(rename = "XPL")] + Xpl, + #[serde(rename = "XPL Fee")] + Fee, +} + +/// `0x`-prefixed hex quantity. Encodes as `"0x0"` for zero. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct HexQuantity(pub u64); + +impl<'de> Deserialize<'de> for HexQuantity { + fn deserialize>(deserializer: D) -> Result { + let s = String::deserialize(deserializer)?; + let stripped = s.strip_prefix("0x").unwrap_or(&s); + let n = u64::from_str_radix(stripped, 16).map_err(serde::de::Error::custom)?; + Ok(Self(n)) + } +} + +impl From for HexQuantity { + fn from(value: u64) -> Self { + Self(value) + } +} + +impl Serialize for HexQuantity { + fn serialize(&self, serializer: S) -> Result { + serializer.collect_str(&format_args!("0x{:x}", self.0)) + } +} + +/// Decimal-encoded `U256`. The controller expects wei as a base-10 string. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct DecimalU256(pub U256); + +impl<'de> Deserialize<'de> for DecimalU256 { + fn deserialize>(deserializer: D) -> Result { + let s = String::deserialize(deserializer)?; + let n = U256::from_str_radix(&s, 10).map_err(serde::de::Error::custom)?; + Ok(Self(n)) + } +} + +impl From for DecimalU256 { + fn from(value: U256) -> Self { + Self(value) + } +} + +impl Serialize for DecimalU256 { + fn serialize(&self, serializer: S) -> Result { + serializer.collect_str(&self.0) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use alloy_primitives::{address, b256}; + + #[test] + fn transfer_serializes_to_quicknode_shape() { + let row = Transfer { + address: Address::ZERO, + block_hash: b256!("0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), + block_number: 123.into(), + from: address!("0x1111111111111111111111111111111111111111"), + log_index: 0.into(), + name: TransferName::Xpl, + to: address!("0x2222222222222222222222222222222222222222"), + transaction_hash: b256!( + "0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" + ), + value: U256::from(1_000_000_000_000_000_000_u128).into(), + }; + + let body = WebhookBody { result: std::slice::from_ref(&row) }; + let json: serde_json::Value = + serde_json::from_str(&serde_json::to_string(&body).unwrap()).expect("body parses back"); + + let row_json = &json["result"][0]; + assert_eq!(row_json["address"], "0x0000000000000000000000000000000000000000"); + assert_eq!(row_json["blockNumber"], "0x7b"); + assert_eq!(row_json["logIndex"], "0x0"); + assert_eq!(row_json["name"], "XPL"); + assert_eq!(row_json["value"], "1000000000000000000"); + assert_eq!(row_json["from"], "0x1111111111111111111111111111111111111111"); + assert_eq!(row_json["to"], "0x2222222222222222222222222222222222222222"); + } + + #[test] + fn fee_row_uses_xpl_fee_name() { + let row = Transfer { + address: Address::ZERO, + block_hash: B256::ZERO, + block_number: 1.into(), + from: Address::ZERO, + log_index: 5.into(), + name: TransferName::Fee, + to: Address::ZERO, + transaction_hash: B256::ZERO, + value: U256::from(21_000).into(), + }; + let s = serde_json::to_string(&row).unwrap(); + assert!(s.contains("\"name\":\"XPL Fee\"")); + assert!(s.contains("\"logIndex\":\"0x5\"")); + assert!(s.contains("\"value\":\"21000\"")); + } +} diff --git a/bin/reth-xpl-webhook/src/replay.rs b/bin/reth-xpl-webhook/src/replay.rs new file mode 100644 index 00000000000..c95159c5976 --- /dev/null +++ b/bin/reth-xpl-webhook/src/replay.rs @@ -0,0 +1,96 @@ +//! Per-transaction replay using a [`TracingInspector`] to recover internal +//! native value transfers that are invisible in receipts. + +use crate::{ + extractor::{extract_value_transfers, push_fee_row, LogIndexCounter, TxRowContext}, + payload::Transfer, +}; +use alloy_consensus::{transaction::TxHashRef, BlockHeader, Transaction}; +use alloy_evm::{block::BlockExecutor, evm::EvmFactoryExt}; +use alloy_primitives::Address; +use eyre::Context; +use reth_ethereum::{ + evm::primitives::ConfigureEvm, + node::api::{FullNodeComponents, FullNodeTypes, NodePrimitives, NodeTypes}, + primitives::RecoveredBlock, +}; +use reth_revm::{database::StateProviderDatabase, db::State, revm::context::Block as RevmBlock}; +use reth_storage_api::StateProviderFactory; +use revm_inspectors::tracing::{TracingInspector, TracingInspectorConfig}; + +type BlockTy = + <<::Types as NodeTypes>::Primitives as NodePrimitives>::Block; + +/// Replay every transaction in `block` against its parent state with a parity +/// tracing inspector. Returns the flat list of native-transfer + fee rows in +/// block order. +pub fn replay_block( + ctx: &reth_exex::ExExContext, + block: &RecoveredBlock>, + native_address: Address, +) -> eyre::Result> +where + Node: FullNodeComponents, +{ + let parent_state = ctx + .provider() + .history_by_block_hash(block.parent_hash()) + .wrap_err("provider missing parent state for committed block")?; + + let mut db = State::builder() + .with_bundle_update() + .with_database(StateProviderDatabase::new(parent_state)) + .build(); + + let evm_env = ctx + .evm_config() + .evm_env(block.sealed_block().sealed_header()) + .map_err(|e| eyre::eyre!("evm_env failed: {e:?}"))?; + let base_fee = RevmBlock::basefee(&evm_env.block_env); + + // Apply pre-block state transitions (beacon root contract, withdrawals queue, etc.). + ctx.evm_config() + .executor_for_block(&mut db, block.sealed_block()) + .map_err(|e| eyre::eyre!("executor_for_block failed: {e:?}"))? + .apply_pre_execution_changes() + .map_err(|e| eyre::eyre!("apply_pre_execution_changes failed: {e:?}"))?; + + let block_hash = block.hash(); + let block_number = block.number(); + let mut rows = Vec::new(); + let mut counter = LogIndexCounter::default(); + + let mut tracer = ctx.evm_config().evm_factory().create_tracer( + &mut db, + evm_env, + TracingInspector::new(TracingInspectorConfig::default_parity()), + ); + + for result in + tracer.try_trace_many(block.transactions_recovered(), |mut tctx| -> eyre::Result<()> { + let tx = tctx.tx; + let tx_hash = *tx.tx_hash(); + let signer = tx.signer(); + let gas_used = tctx.result.gas_used(); + let effective_gas_price = tx.effective_gas_price(Some(base_fee)); + + let row_ctx = TxRowContext { + native_address, + block_hash, + block_number, + transaction_hash: tx_hash, + tx_signer: signer, + }; + + let inspector = tctx.take_inspector(); + let arena = inspector.into_traces(); + extract_value_transfers(&arena, &row_ctx, &mut counter, &mut rows); + push_fee_row(&row_ctx, gas_used, effective_gas_price, &mut counter, &mut rows); + Ok(()) + }) + { + result?; + } + + Ok(rows) +} diff --git a/bin/reth-xpl-webhook/src/webhook.rs b/bin/reth-xpl-webhook/src/webhook.rs new file mode 100644 index 00000000000..65ab70d6c77 --- /dev/null +++ b/bin/reth-xpl-webhook/src/webhook.rs @@ -0,0 +1,221 @@ +use crate::{args::XplWebhookArgs, metrics as m, outbox::Outbox, payload::WebhookBody}; +use hmac::{Hmac, Mac}; +use rand::RngCore; +use sha2::Sha256; +use std::{ + sync::Arc, + time::{Duration, SystemTime, UNIX_EPOCH}, +}; +use tokio::{sync::Notify, task::JoinHandle}; +use tracing::{debug, error, warn}; + +type HmacSha256 = Hmac; + +/// Background HTTP worker draining the [`Outbox`] in block-number order. +pub struct WebhookDispatcher { + shutdown: Arc, + handle: JoinHandle<()>, +} + +impl WebhookDispatcher { + pub fn spawn(args: XplWebhookArgs, outbox: Outbox) -> Self { + let shutdown = Arc::new(Notify::new()); + let notify_clone = shutdown.clone(); + let wakeup = outbox.wakeup_handle(); + + let client = reqwest::Client::builder() + .timeout(Duration::from_millis(args.http_timeout_ms)) + .build() + .expect("reqwest client builds"); + + let handle = tokio::spawn(async move { + run_dispatcher(args, outbox, client, wakeup, notify_clone).await; + }); + + Self { shutdown, handle } + } + + pub async fn shutdown(self) { + self.shutdown.notify_waiters(); + let _ = self.handle.await; + } +} + +async fn run_dispatcher( + args: XplWebhookArgs, + outbox: Outbox, + client: reqwest::Client, + wakeup: Arc, + shutdown: Arc, +) { + loop { + tokio::select! { + biased; + _ = shutdown.notified() => return, + _ = drain_loop(&args, &outbox, &client, &wakeup) => {} + } + } +} + +async fn drain_loop( + args: &XplWebhookArgs, + outbox: &Outbox, + client: &reqwest::Client, + wakeup: &Arc, +) { + let now_ms = unix_ms(); + let batch = match outbox.next_pending(now_ms) { + Ok(Some(batch)) => batch, + Ok(None) => { + // nothing to send: wait for the producer to nudge us + wakeup.notified().await; + return; + } + Err(err) => { + error!(target: "xpl-webhook", %err, "outbox read failed"); + tokio::time::sleep(Duration::from_secs(1)).await; + return; + } + }; + + let body = WebhookBody { result: &batch.rows }; + let raw_body = match serde_json::to_vec(&body) { + Ok(bytes) => bytes, + Err(err) => { + error!(target: "xpl-webhook", %err, "payload serialization failed; dropping batch"); + let _ = outbox.mark_delivered(batch.block_hash); + return; + } + }; + + let nonce = random_nonce_hex(); + let timestamp = unix_ms().to_string(); + let signature = sign_payload(&args.token, &nonce, ×tamp, &raw_body); + + debug!( + target: "xpl-webhook", + block_number = batch.block_number, + rows = batch.rows.len(), + attempt = batch.attempts + 1, + "posting batch", + ); + + let result = client + .post(&args.url) + .header("content-type", "application/json") + .header("x-qn-signature", signature) + .header("x-qn-nonce", nonce) + .header("x-qn-timestamp", timestamp) + .body(raw_body) + .send() + .await; + + match result { + Ok(resp) if resp.status().is_success() => { + ::metrics::counter!(m::HTTP_SUCCESS).increment(1); + ::metrics::counter!(m::ROWS_DELIVERED).increment(batch.rows.len() as u64); + if let Err(err) = outbox.mark_delivered(batch.block_hash) { + error!(target: "xpl-webhook", %err, "failed to mark batch delivered"); + } + } + Ok(resp) => { + warn!( + target: "xpl-webhook", + status = %resp.status(), + block_number = batch.block_number, + "webhook non-2xx", + ); + handle_failure(args, outbox, &batch); + } + Err(err) => { + warn!( + target: "xpl-webhook", + %err, + block_number = batch.block_number, + "webhook request failed", + ); + handle_failure(args, outbox, &batch); + } + } +} + +fn handle_failure( + args: &XplWebhookArgs, + outbox: &Outbox, + batch: &crate::outbox::PendingBatchRecord, +) { + ::metrics::counter!(m::HTTP_FAILURES).increment(1); + let next_attempt_at = unix_ms().saturating_add(backoff_ms(batch.attempts)); + if let Err(err) = outbox.record_attempt(batch.block_hash, next_attempt_at) { + error!(target: "xpl-webhook", %err, "failed to record attempt"); + } + if batch.attempts + 1 >= args.max_retries { + ::metrics::counter!(m::DLQ).increment(1); + warn!( + target: "xpl-webhook", + block_number = batch.block_number, + attempts = batch.attempts + 1, + "batch reached max retries; remains pending for operator intervention", + ); + } +} + +/// Capped exponential backoff in milliseconds. +const BACKOFF_LADDER_MS: &[u64] = &[500, 1_000, 2_000, 5_000, 15_000, 30_000, 60_000, 120_000]; + +fn backoff_ms(attempts: u32) -> u64 { + let idx = (attempts as usize).min(BACKOFF_LADDER_MS.len() - 1); + BACKOFF_LADDER_MS[idx] +} + +fn unix_ms() -> u64 { + SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0) +} + +fn random_nonce_hex() -> String { + let mut bytes = [0_u8; 16]; + rand::rng().fill_bytes(&mut bytes); + let mut out = String::with_capacity(32); + for b in bytes { + use std::fmt::Write; + let _ = write!(&mut out, "{b:02x}"); + } + out +} + +/// HMAC-SHA256 of `nonce || timestamp || raw_body` keyed with `token`, +/// returned as lowercase hex. Matches the controller's verifier in +/// `apps/tasks`. +pub fn sign_payload(token: &str, nonce: &str, timestamp: &str, raw_body: &[u8]) -> String { + let mut mac = HmacSha256::new_from_slice(token.as_bytes()).expect("HMAC accepts any key size"); + mac.update(nonce.as_bytes()); + mac.update(timestamp.as_bytes()); + mac.update(raw_body); + let digest = mac.finalize().into_bytes(); + let mut hex = String::with_capacity(digest.len() * 2); + for byte in digest { + use std::fmt::Write; + let _ = write!(&mut hex, "{byte:02x}"); + } + hex +} + +#[cfg(test)] +mod tests { + use super::sign_payload; + + #[test] + fn hmac_signature_matches_precomputed_digest() { + // Precomputed with: + // echo -n "nonce0123timestamp9999BODY" | openssl dgst -sha256 -hmac "secrettoken" + let sig = sign_payload("secrettoken", "nonce0123", "timestamp9999", b"BODY"); + assert_eq!(sig, "b0f45fe2ba1e62484e1ab5ac9082b205694fff2c8b5352ab0dda5dd3ea47e609"); + } + + #[test] + fn hmac_signature_includes_body_bytes() { + let a = sign_payload("k", "n", "t", b"{\"result\":[]}"); + let b = sign_payload("k", "n", "t", b"{\"result\":[{}]}"); + assert_ne!(a, b); + } +}