From 3ba0c6dd50f6ac3d79f6cc51f7ec14c116d8082a Mon Sep 17 00:00:00 2001 From: Hristo Staykov Date: Thu, 6 Nov 2025 17:22:41 +0200 Subject: [PATCH 1/3] feat(sequencer/eth_send_utils): Add check for DAG inclusion --- .../src/providers/dag_inclusion_tracking.rs | 257 ++++++++++++++++++ apps/sequencer/src/providers/mod.rs | 1 + 2 files changed, 258 insertions(+) create mode 100644 apps/sequencer/src/providers/dag_inclusion_tracking.rs diff --git a/apps/sequencer/src/providers/dag_inclusion_tracking.rs b/apps/sequencer/src/providers/dag_inclusion_tracking.rs new file mode 100644 index 0000000000..775339692a --- /dev/null +++ b/apps/sequencer/src/providers/dag_inclusion_tracking.rs @@ -0,0 +1,257 @@ +use std::cmp::max; +use std::time::{Duration, Instant}; + +use alloy::providers::Provider; +use alloy_primitives::{TxHash, B256}; +use eyre::{eyre, Result, WrapErr}; +use serde::Deserialize; +use serde_json::{json, Value}; +use tokio::time::sleep; +use tracing::debug; + +use crate::providers::provider::{ProviderType, RpcProvider}; + +/// Default number of extra DAG levels we require before declaring a tx +/// "sufficiently included". This value is intentionally small so the caller +/// can override it when wiring the helper. +pub const DEFAULT_REQUIRED_DAG_DEPTH: u64 = 5; +const DAG_INCLUSION_TIMEOUT: Duration = Duration::from_secs(10); +const DAG_POLL_INTERVAL: Duration = Duration::from_millis(500); +const DAG_LOOKBACK_MIN_LEVELS: u64 = 64; +const DAG_LOOKBACK_MARGIN: u64 = 8; + +#[derive(Debug, Clone)] +pub struct DagInclusionStatus { + pub tx_hash: TxHash, + pub dag_block_hash: B256, + pub dag_level: u64, + pub depth_reached: u64, + pub period: Option, + pub elapsed: Duration, +} + +#[derive(Debug, Clone, Deserialize)] +struct TaraxaDagBlock { + pub hash: B256, + #[serde(deserialize_with = "deserialize_hex_u64")] + pub level: u64, + #[serde(default, deserialize_with = "deserialize_optional_hex_u64")] + pub period: Option, + #[serde(default)] + pub transactions: Vec, +} + +/// Waits until the given transaction hash is observed inside a Taraxa DAG block +/// and that block accumulates at least `required_depth` additional DAG levels. +pub async fn wait_for_dag_inclusion( + provider: &RpcProvider, + tx_hash: TxHash, + required_depth: u64, +) -> Result { + if required_depth == 0 { + return Err(eyre!( + "required_depth must be greater than zero when tracking DAG inclusion" + )); + } + + let rpc = &provider.provider; + let start = Instant::now(); + let deadline = start + DAG_INCLUSION_TIMEOUT; + + let mut located_block: Option = None; + + loop { + if Instant::now() >= deadline { + break; + } + + let current_level = fetch_current_dag_level(rpc).await?; + + if located_block.is_none() { + located_block = search_recent_levels(rpc, &tx_hash, current_level, required_depth) + .await + .wrap_err("failed to search DAG levels for transaction")?; + if let Some(block) = &located_block { + debug!( + tx_hash = format!("{tx_hash:?}"), + dag_block = format!("{:?}", block.hash), + dag_level = block.level, + "Transaction observed in DAG block" + ); + } + } + + if let Some(block) = &located_block { + let depth = current_level.saturating_sub(block.level); + if depth >= required_depth { + return finalize_status(rpc, tx_hash, block.clone(), depth, start).await; + } + } + + sleep(DAG_POLL_INTERVAL).await; + } + + Err(eyre!( + "timed out after {:?} waiting for DAG inclusion of tx {tx_hash:?}", + DAG_INCLUSION_TIMEOUT + )) +} + +async fn finalize_status( + rpc: &ProviderType, + tx_hash: TxHash, + block_hint: TaraxaDagBlock, + depth: u64, + started_at: Instant, +) -> Result { + let detailed = fetch_dag_block_by_hash(rpc, block_hint.hash).await?; + let block = detailed.unwrap_or(block_hint); + Ok(DagInclusionStatus { + tx_hash, + dag_block_hash: block.hash, + dag_level: block.level, + depth_reached: depth, + period: block.period, + elapsed: started_at.elapsed(), + }) +} + +async fn search_recent_levels( + rpc: &ProviderType, + tx_hash: &TxHash, + current_level: u64, + required_depth: u64, +) -> Result> { + let lookback = max( + DAG_LOOKBACK_MIN_LEVELS, + required_depth + DAG_LOOKBACK_MARGIN, + ); + let min_level = current_level.saturating_sub(lookback); + let mut level = current_level; + + loop { + let blocks = fetch_blocks_for_level(rpc, level).await?; + if let Some(found) = blocks.into_iter().find(|block| { + block + .transactions + .iter() + .any(|hash_in_block| hash_in_block == tx_hash) + }) { + return Ok(Some(found)); + } + + if level == 0 || level == min_level { + break; + } + level -= 1; + } + + Ok(None) +} + +async fn fetch_current_dag_level(rpc: &ProviderType) -> Result { + let raw: Value = rpc + .raw_request("taraxa_dagBlockLevel".into(), ()) + .await + .wrap_err("taraxa_dagBlockLevel RPC call failed")?; + parse_quantity_value(raw) +} + +async fn fetch_blocks_for_level(rpc: &ProviderType, level: u64) -> Result> { + let params = json!([format_hex_quantity(level), false]); + let blocks: Option> = rpc + .raw_request("taraxa_getDagBlockByLevel".into(), params) + .await + .wrap_err_with(|| format!("taraxa_getDagBlockByLevel failed for level {level}"))?; + Ok(blocks.unwrap_or_default()) +} + +async fn fetch_dag_block_by_hash(rpc: &ProviderType, hash: B256) -> Result> { + let params = json!([hash, false]); + rpc.raw_request("taraxa_getDagBlockByHash".into(), params) + .await + .wrap_err_with(|| format!("taraxa_getDagBlockByHash failed for {hash:?}")) +} + +fn format_hex_quantity(value: u64) -> String { + format!("0x{:x}", value) +} + +fn parse_quantity_value(value: Value) -> Result { + match value { + Value::String(s) => parse_quantity_str(&s), + Value::Number(num) => num + .as_u64() + .ok_or_else(|| eyre!("failed to decode quantity from number {num}")), + other => Err(eyre!("unexpected quantity representation: {other:?}")), + } +} + +fn parse_quantity_str(input: &str) -> Result { + let trimmed = input.trim(); + if trimmed.is_empty() { + return Err(eyre!("empty quantity string")); + } + + if let Some(without_prefix) = trimmed + .strip_prefix("0x") + .or_else(|| trimmed.strip_prefix("0X")) + { + if without_prefix.is_empty() { + return Ok(0); + } + u64::from_str_radix(without_prefix, 16) + .wrap_err_with(|| format!("invalid hex quantity: {trimmed}")) + } else { + trimmed + .parse::() + .wrap_err_with(|| format!("invalid decimal quantity: {trimmed}")) + } +} + +fn deserialize_hex_u64<'de, D>(deserializer: D) -> std::result::Result +where + D: serde::Deserializer<'de>, +{ + let value = Value::deserialize(deserializer)?; + parse_quantity_value(value).map_err(serde::de::Error::custom) +} + +fn deserialize_optional_hex_u64<'de, D>( + deserializer: D, +) -> std::result::Result, D::Error> +where + D: serde::Deserializer<'de>, +{ + let opt = Option::::deserialize(deserializer)?; + match opt { + None => Ok(None), + Some(value) => parse_quantity_value(value) + .map(Some) + .map_err(serde::de::Error::custom), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn parses_hex_and_decimal_quantities() { + assert_eq!(parse_quantity_str("0x10").unwrap(), 16); + assert_eq!(parse_quantity_str("0X2a").unwrap(), 42); + assert_eq!(parse_quantity_str(" 25 ").unwrap(), 25); + } + + #[test] + fn parse_quantity_value_handles_numbers_and_strings() { + assert_eq!(parse_quantity_value(json!("0x1")).unwrap(), 1); + assert_eq!(parse_quantity_value(json!(17_u64)).unwrap(), 17); + } + + #[test] + fn format_hex_quantity_adds_prefix() { + assert_eq!(format_hex_quantity(26), "0x1a"); + } +} diff --git a/apps/sequencer/src/providers/mod.rs b/apps/sequencer/src/providers/mod.rs index 3749fb7618..44ec27d20b 100644 --- a/apps/sequencer/src/providers/mod.rs +++ b/apps/sequencer/src/providers/mod.rs @@ -1,2 +1,3 @@ +pub mod dag_inclusion_tracking; pub mod eth_send_utils; pub mod provider; From ac1caaa7ed0449b17885e2aea6eea4a57c2ec3c5 Mon Sep 17 00:00:00 2001 From: Hristo Staykov Date: Thu, 6 Nov 2025 18:06:47 +0200 Subject: [PATCH 2/3] feat(sequencer/eth_send_utils): Configuration specified tx confirmation (receipt/dag inclusion) --- apps/sequencer/src/http_handlers/admin.rs | 1 + .../sequencer/src/providers/eth_send_utils.rs | 140 +++++++++++++----- libs/config/src/lib.rs | 21 +++ libs/metrics/src/metrics.rs | 12 ++ 4 files changed, 133 insertions(+), 41 deletions(-) diff --git a/apps/sequencer/src/http_handlers/admin.rs b/apps/sequencer/src/http_handlers/admin.rs index ab938fb22b..9946246e61 100644 --- a/apps/sequencer/src/http_handlers/admin.rs +++ b/apps/sequencer/src/http_handlers/admin.rs @@ -950,6 +950,7 @@ mod tests { impersonated_anvil_account: None, publishing_criteria: vec![], should_load_rb_indices: true, + confirmation_method: blocksense_config::ConfirmationMethod::Receipt, contracts, } }); diff --git a/apps/sequencer/src/providers/eth_send_utils.rs b/apps/sequencer/src/providers/eth_send_utils.rs index bb2c11cbea..296a7d2084 100644 --- a/apps/sequencer/src/providers/eth_send_utils.rs +++ b/apps/sequencer/src/providers/eth_send_utils.rs @@ -7,7 +7,7 @@ use alloy::{ rpc::types::{eth::TransactionRequest, TransactionReceipt}, }; use alloy_primitives::{FixedBytes, TxHash}; -use blocksense_config::{FeedStrideAndDecimals, GNOSIS_SAFE_CONTRACT_NAME}; +use blocksense_config::{ConfirmationMethod, FeedStrideAndDecimals, GNOSIS_SAFE_CONTRACT_NAME}; use blocksense_data_feeds::feeds_processing::{BatchedAggregatesToSend, VotedFeedUpdate}; use blocksense_registry::config::FeedConfig; use blocksense_utils::{counter_unbounded_channel::CountedReceiver, FeedId}; @@ -19,9 +19,14 @@ use tokio::{ }; use crate::{ - providers::provider::{ - parse_eth_address, ProviderStatus, ProviderType, ProvidersMetrics, RpcProvider, - SharedRpcProviders, + providers::{ + dag_inclusion_tracking::{ + wait_for_dag_inclusion, DagInclusionStatus, DEFAULT_REQUIRED_DAG_DEPTH, + }, + provider::{ + parse_eth_address, ProviderStatus, ProviderType, ProvidersMetrics, RpcProvider, + SharedRpcProviders, + }, }, sequencer_state::SequencerState, }; @@ -397,8 +402,8 @@ pub async fn eth_batch_send_to_contract( let rpc_handle = &provider.provider; let input = Bytes::from(serialized_updates); - - let receipt; + let confirmation_method = provider_settings.confirmation_method; + let confirmation_outcome; let tx_time = Instant::now(); let (sender_address, is_impersonated) = match &provider_settings.impersonated_anvil_account { @@ -619,7 +624,7 @@ pub async fn eth_batch_send_to_contract( debug!("Retrying for {transaction_retries_count}-th time in network `{net}` block height {block_height} tx: {tx:?}"); } - let tx_receipt = { + let tx_confirmation = { let rpc_impersonated_handle; let send_transaction_future = if is_impersonated { let rpc_impersonated_url = provider.url(); @@ -675,57 +680,110 @@ pub async fn eth_batch_send_to_contract( info!("Successfully posted tx to RPC and got tx_hash in network `{net}` block height {block_height} and address {sender_address} tx_hash = {tx_hash}"); - let tx_get_receipt_start_time = Instant::now(); - let tx_receipt = match await_receipt( - net.as_str(), - rpc_handle, - transaction_retry_timeout_secs, - &tx_hash, - block_height, - &sender_address, - tx_get_receipt_start_time, - receipt_polling_back_off_period_ms, - ) - .await - { - Ok(receipt) => { - inc_metric!(provider_metrics, net, success_get_receipt); - receipt - } - Err(e) => { - warn!("await_receipt: {e}"); - inc_metric!(provider_metrics, net, failed_get_receipt); - inc_retries_with_backoff( + let confirmation = match confirmation_method { + ConfirmationMethod::Receipt => { + let tx_get_receipt_start_time = Instant::now(); + match await_receipt( net.as_str(), - &mut transaction_retries_count, - provider_metrics, - retry_backoff_ms, + rpc_handle, + transaction_retry_timeout_secs, + &tx_hash, + block_height, + &sender_address, + tx_get_receipt_start_time, + receipt_polling_back_off_period_ms, ) - .await; - continue; + .await + { + Ok(receipt) => { + inc_metric!(provider_metrics, net, success_get_receipt); + ConfirmationOutcome::Receipt(Box::new(receipt)) + } + Err(e) => { + warn!("await_receipt: {e}"); + inc_metric!(provider_metrics, net, failed_get_receipt); + inc_retries_with_backoff( + net.as_str(), + &mut transaction_retries_count, + provider_metrics, + retry_backoff_ms, + ) + .await; + continue; + } + } + } + ConfirmationMethod::DagInclusion => { + match wait_for_dag_inclusion(&provider, tx_hash, DEFAULT_REQUIRED_DAG_DEPTH) + .await + { + Ok(status) => { + inc_metric!(provider_metrics, net, dag_inclusion_success); + ConfirmationOutcome::Dag(status) + } + Err(e) => { + warn!("wait_for_dag_inclusion failed for tx {tx_hash:?} in `{net}` block height {block_height}: {e}"); + inc_metric!(provider_metrics, net, dag_inclusion_failure); + inc_retries_with_backoff( + net.as_str(), + &mut transaction_retries_count, + provider_metrics, + retry_backoff_ms, + ) + .await; + continue; + } + } } }; - tx_receipt + confirmation }; - receipt = tx_receipt; + confirmation_outcome = tx_confirmation; break; } let transaction_time = tx_time.elapsed().as_millis(); - info!( - "Successfully recvd transaction receipt that took {transaction_time}ms for {transaction_retries_count} retries in network `{net}` block height {block_height} and sender_address {sender_address}: {receipt:?}" - ); - - log_gas_used(&net, &receipt, transaction_time, provider_metrics).await; + let confirmation_status = match confirmation_outcome { + ConfirmationOutcome::Receipt(receipt) => { + let receipt = *receipt; + info!( + "Successfully recvd transaction receipt that took {transaction_time}ms for {transaction_retries_count} retries in network `{net}` block height {block_height} and sender_address {sender_address}: {receipt:?}" + ); + log_gas_used(&net, &receipt, transaction_time, provider_metrics).await; + receipt.status().to_string() + } + ConfirmationOutcome::Dag(status) => { + info!( + "DAG inclusion confirmed for tx {:?} in network `{net}` block height {block_height} after {:?} (depth={}, dag_block={}, confirmation retries={transaction_retries_count})", + status.tx_hash, + status.elapsed, + status.depth_reached, + status.dag_block_hash + ); + let dag_confirmation_time_ms = status.elapsed.as_millis(); + set_metric!( + provider_metrics, + net, + transaction_confirmation_time, + dag_confirmation_time_ms + ); + "true".to_string() + } + }; provider.update_history(&updates.updates); drop(provider); debug!("Released a read/write lock on provider state in network `{net}` block height {block_height}"); - Ok((receipt.status().to_string(), feeds_to_update_ids)) + Ok((confirmation_status, feeds_to_update_ids)) +} + +enum ConfirmationOutcome { + Receipt(Box), + Dag(DagInclusionStatus), } #[allow(clippy::too_many_arguments)] diff --git a/libs/config/src/lib.rs b/libs/config/src/lib.rs index a2913c3340..3645af74a4 100644 --- a/libs/config/src/lib.rs +++ b/libs/config/src/lib.rs @@ -219,6 +219,8 @@ pub struct Provider { #[serde(default)] pub contracts: Vec, + #[serde(default = "default_confirmation_method")] + pub confirmation_method: ConfirmationMethod, } fn default_is_enabled() -> bool { @@ -235,6 +237,24 @@ fn default_transaction_retry_back_off_ms() -> u64 { 1_000 } +fn default_confirmation_method() -> ConfirmationMethod { + ConfirmationMethod::Receipt +} + +#[derive(Debug, Serialize, Deserialize, Clone, Copy, PartialEq, Eq)] +pub enum ConfirmationMethod { + #[serde(rename = "receipt", alias = "Receipt")] + Receipt, + #[serde( + rename = "DAG inclusion", + alias = "dag_inclusion", + alias = "dag inclusion", + alias = "Dag inclusion", + alias = "DAG Inclusion" + )] + DagInclusion, +} + impl Validated for Provider { fn validate(&self, context: &str) -> anyhow::Result<()> { if self.transaction_retry_timeout_secs == 0 { @@ -524,6 +544,7 @@ pub fn get_test_config_with_multiple_providers( should_load_rb_indices: false, allow_feeds: None, publishing_criteria: vec![], + confirmation_method: default_confirmation_method(), impersonated_anvil_account: None, contracts: vec![ // Gnosis safe contract, if present changes the flow, and no direct updates will be made to the ADFS contract. diff --git a/libs/metrics/src/metrics.rs b/libs/metrics/src/metrics.rs index 981eb0921f..00b4448d3d 100644 --- a/libs/metrics/src/metrics.rs +++ b/libs/metrics/src/metrics.rs @@ -167,6 +167,8 @@ pub struct ProviderMetrics { pub success_get_gas_price: IntCounterVec, pub success_get_max_priority_fee_per_gas: IntCounterVec, pub success_get_chain_id: IntCounterVec, + pub dag_inclusion_success: IntCounterVec, + pub dag_inclusion_failure: IntCounterVec, pub total_timed_out_tx: IntCounterVec, pub total_transaction_retries: IntCounterVec, pub total_mismatched_gnosis_safe_nonce: IntCounterVec, @@ -253,6 +255,16 @@ impl ProviderMetrics { "Total number of successful get_chain_id req-s for network", &["Network"] )?, + dag_inclusion_success: register_int_counter_vec!( + format!("{}dag_inclusion_success", prefix), + "Total number of successful DAG inclusion waits", + &["Network"] + )?, + dag_inclusion_failure: register_int_counter_vec!( + format!("{}dag_inclusion_failure", prefix), + "Total number of failed DAG inclusion waits", + &["Network"] + )?, total_timed_out_tx: register_int_counter_vec!( format!("{}total_timed_out_tx", prefix), "Total number of tx sent that reached the configured timeout before completion for network", From d5b87a4d4b944c20980fa197ced96fc5d3796206 Mon Sep 17 00:00:00 2001 From: Hristo Staykov Date: Thu, 6 Nov 2025 18:32:39 +0200 Subject: [PATCH 3/3] feat(sequencer/nix/config): Add definition for confirmation-method config option --- .../module-opts/sequencer/provider/default.nix | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/nix/modules/blocksense/module-opts/sequencer/provider/default.nix b/nix/modules/blocksense/module-opts/sequencer/provider/default.nix index 8c03f81bf5..e527bfc803 100644 --- a/nix/modules/blocksense/module-opts/sequencer/provider/default.nix +++ b/nix/modules/blocksense/module-opts/sequencer/provider/default.nix @@ -75,5 +75,14 @@ lib: with lib; { default = [ ]; description = mdDoc "List of contracts of various types"; }; + + confirmation-method = mkOption { + type = types.enum [ + "receipt" + "DAG inclusion" + ]; + default = "receipt"; + description = mdDoc "Confirmation strategy: wait for a receipt or stop after DAG inclusion."; + }; }; }