diff --git a/fixtures/create_transactions_table.sql b/fixtures/create_transactions_table.sql new file mode 100644 index 00000000..269e9ea4 --- /dev/null +++ b/fixtures/create_transactions_table.sql @@ -0,0 +1,28 @@ +CREATE TABLE transactions_from_send_raw_tx +( + `received_at` DateTime64(3, + 'UTC'), + `hash` String, + `chain_id` String, + `tx_type` Int64, + `from` String, + `to` String, + `value` String, + `nonce` String, + `gas` String, + `gas_price` String, + `gas_tip_cap` String, + `gas_fee_cap` String, + `data_size` Int64, + `data_4bytes` String, + `raw_tx` String, + `ver` Int64 MATERIALIZED -toUnixTimestamp(received_at) +) +ENGINE = ReplacingMergeTree(ver) +PARTITION BY toYYYYMM(received_at) +PRIMARY KEY hash +ORDER BY hash +SETTINGS index_granularity = 8192 +COMMENT 'Transaction details, + deduplicated by hash, + will keep the transaction with earliest received_at.'; \ No newline at end of file diff --git a/src/cli.rs b/src/cli.rs index 3f7f73de..2bbc1120 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -8,7 +8,7 @@ use rbuilder_utils::clickhouse::indexer::{ }; use crate::{ - indexer::{BUNDLE_RECEIPTS_TABLE_NAME, BUNDLE_TABLE_NAME}, + indexer::{BUNDLE_RECEIPTS_TABLE_NAME, BUNDLE_TABLE_NAME, TRANSACTIONS_TABLE_NAME}, ingress, primitives::SystemBundleDecoder, }; @@ -69,6 +69,15 @@ pub struct ClickhouseArgs { )] pub bundle_receipts_table_name: String, + /// The clickhouse table name to store transactions data. + #[arg( + long = "indexer.clickhouse.transactions-table-name", + env = "CLICKHOUSE_TRANSACTIONS_TABLE_NAME", + id = "CLICKHOUSE_TRANSACTIONS_TABLE_NAME", + default_value = TRANSACTIONS_TABLE_NAME, + )] + pub transactions_table_name: String, + /// The maximum size in bytes for the in-memory backup in case of of disk-backup failure, for a /// certain data type (bundles or bundle receipts). Defaults to 1GiB. #[arg( diff --git a/src/indexer/click/mod.rs b/src/indexer/click/mod.rs index c40572e6..00d37e36 100644 --- a/src/indexer/click/mod.rs +++ b/src/indexer/click/mod.rs @@ -13,11 +13,11 @@ use std::{ use crate::{ cli::ClickhouseArgs, indexer::{ - click::models::{BundleReceiptRow, BundleRow}, + click::models::{BundleReceiptRow, BundleRow, TransactionRow}, OrderReceivers, TARGET_INDEXER, }, metrics::CLICKHOUSE_METRICS, - primitives::{BundleReceipt, SystemBundle}, + primitives::{BundleReceipt, SystemBundle, SystemTransaction}, }; use rbuilder_utils::{ clickhouse::{ @@ -46,6 +46,7 @@ fn config_from_clickhouse_args(args: &ClickhouseArgs, validation: bool) -> Click pub(crate) struct ClickhouseLocalBackupDiskSize { bundles_size: AtomicU64, bundle_receipts_size: AtomicU64, + transactions_size: AtomicU64, } impl ClickhouseLocalBackupDiskSize { @@ -55,9 +56,13 @@ impl ClickhouseLocalBackupDiskSize { pub(crate) fn set_bundle_receipts_size(&self, size: u64) { self.bundle_receipts_size.store(size, Ordering::Relaxed); } + pub(crate) fn set_transactions_size(&self, size: u64) { + self.transactions_size.store(size, Ordering::Relaxed); + } pub(crate) fn disk_size(&self) -> u64 { self.bundles_size.load(Ordering::Relaxed) + - self.bundle_receipts_size.load(Ordering::Relaxed) + self.bundle_receipts_size.load(Ordering::Relaxed) + + self.transactions_size.load(Ordering::Relaxed) } } @@ -87,6 +92,14 @@ impl DiskBackupSizeCallback for UpdateBundleReceiptsSizeCallback { } } +struct UpdateTransactionsSizeCallback; + +impl DiskBackupSizeCallback for UpdateTransactionsSizeCallback { + fn on_disk_backup_size(size_bytes: u64) { + CLICKHOUSE_LOCAL_BACKUP_DISK_SIZE.set_transactions_size(size_bytes); + } +} + struct MetricsWrapper(PhantomData) where F: DiskBackupSizeCallback; @@ -211,7 +224,28 @@ impl ClickhouseIndexer { end_timeout, TARGET_INDEXER, ); - vec![bundle_inserter_join_handle, bundle_receipt_inserter_join_handle] + + let transaction_inserter_join_handle = spawn_clickhouse_inserter_and_backup::< + SystemTransaction, + TransactionRow, + MetricsWrapper, + >( + &client, + receivers.transaction_rx, + &task_executor, + args.transactions_table_name, + builder_name.clone(), + disk_backup.clone(), + args.backup_memory_max_size_bytes, + send_timeout, + end_timeout, + TARGET_INDEXER, + ); + vec![ + bundle_inserter_join_handle, + bundle_receipt_inserter_join_handle, + transaction_inserter_join_handle, + ] } } @@ -229,6 +263,7 @@ pub(crate) mod tests { }, tests::{bundle_receipt_example, system_bundle_example}, Indexer, OrderSenders, BUNDLE_RECEIPTS_TABLE_NAME, BUNDLE_TABLE_NAME, TARGET_INDEXER, + TRANSACTIONS_TABLE_NAME, }, jsonrpc::{JsonRpcError, JsonRpcResponse, JsonRpcResponseTy, JSONRPC_VERSION_2}, utils::{ @@ -340,6 +375,7 @@ pub(crate) mod tests { password: Some(config.password), bundles_table_name: BUNDLE_TABLE_NAME.to_string(), bundle_receipts_table_name: BUNDLE_RECEIPTS_TABLE_NAME.to_string(), + transactions_table_name: TRANSACTIONS_TABLE_NAME.to_string(), backup_memory_max_size_bytes: 1024 * 1024 * 10, // 10MiB backup_disk_database_path: default_disk_backup_database_path(), backup_disk_max_size_bytes: 1024 * 1024 * 100, // 100MiB diff --git a/src/indexer/click/models.rs b/src/indexer/click/models.rs index b60e8f13..8aad8170 100644 --- a/src/indexer/click/models.rs +++ b/src/indexer/click/models.rs @@ -1,12 +1,12 @@ //! Contains the model used for storing data inside Clickhouse. use crate::{ - indexer::ser::{address, addresses, hash, hashes, u256es}, + indexer::ser::{address, addresses, hash, hashes, raw_bytes, u256es}, primitives::BundleReceipt, }; use alloy_consensus::Transaction; use alloy_eips::Typed2718; -use alloy_primitives::{Address, Keccak256, B256, U256}; +use alloy_primitives::{hex, Address, Keccak256, B256, U256}; use alloy_rlp::Encodable; use clickhouse::Row; use rbuilder_primitives::BundleVersion; @@ -14,7 +14,7 @@ use rbuilder_utils::clickhouse::backup::primitives::{ClickhouseIndexableData, Cl use time::{OffsetDateTime, UtcDateTime}; use uuid::Uuid; -use crate::primitives::{DecodedBundle, SystemBundle}; +use crate::primitives::{DecodedBundle, SystemBundle, SystemTransaction}; /// Model representing Clickhouse bundle row. /// @@ -424,6 +424,102 @@ impl ClickhouseIndexableData for BundleReceipt { } } +/// Model representing a Clickhouse transaction row for individual transactions received via +/// `eth_sendRawTransaction`. +/// +/// NOTE: Make sure the fields are in the same order as the columns in the Clickhouse table. +#[derive(Clone, clickhouse::Row, Debug, serde::Serialize, serde::Deserialize)] +#[cfg_attr(test, derive(PartialEq, Eq))] +pub struct TransactionRow { + #[serde(with = "clickhouse::serde::time::datetime64::millis")] + pub received_at: OffsetDateTime, + pub hash: String, + pub chain_id: String, + pub tx_type: i64, + #[serde(rename = "from")] + pub tx_from: String, + pub to: String, + pub value: String, + pub nonce: String, + pub gas: String, + pub gas_price: String, + pub gas_tip_cap: String, + pub gas_fee_cap: String, + pub data_size: i64, + pub data_4bytes: String, + #[serde(serialize_with = "raw_bytes::serialize")] + pub raw_tx: Vec, +} + +impl ClickhouseRowExt for TransactionRow { + type TraceId = String; + const TABLE_NAME: &'static str = "transaction"; + + fn trace_id(&self) -> Self::TraceId { + self.hash.clone() + } + + fn to_row_ref(row: &Self) -> &::Value<'_> { + row + } +} + +impl From<(SystemTransaction, String)> for TransactionRow { + fn from((system_tx, _builder_name): (SystemTransaction, String)) -> Self { + let tx = &system_tx.transaction; + let millis = system_tx.received_at.utc.millisecond(); + let received_at: OffsetDateTime = system_tx + .received_at + .utc + .replace_millisecond(millis) + .expect("to replace milliseconds") + .into(); + + let input = tx.decoded.input(); + let data_4bytes = if input.len() >= 4 { + format!("0x{}", hex::encode(&input[..4])) + } else { + String::new() + }; + + TransactionRow { + received_at, + hash: format!("{:#x}", tx.decoded.tx_hash()), + chain_id: tx.decoded.chain_id().map(|c| c.to_string()).unwrap_or_default(), + tx_type: tx.decoded.tx_type() as i64, + tx_from: format!("{:#x}", system_tx.tx_sender), + to: tx.decoded.to().map(|a| format!("{:#x}", a)).unwrap_or_default(), + value: tx.decoded.value().to_string(), + nonce: tx.decoded.nonce().to_string(), + gas: tx.decoded.gas_limit().to_string(), + gas_price: tx.decoded.gas_price().map(|p| p.to_string()).unwrap_or_default(), + gas_tip_cap: tx + .decoded + .max_priority_fee_per_gas() + .map(|p| p.to_string()) + .unwrap_or_default(), + gas_fee_cap: tx.decoded.max_fee_per_gas().to_string(), + data_size: input.len() as i64, + data_4bytes, + raw_tx: tx.raw.to_vec(), + } + } +} + +impl ClickhouseIndexableData for SystemTransaction { + type ClickhouseRowType = TransactionRow; + + const DATA_NAME: &'static str = ::TABLE_NAME; + + fn trace_id(&self) -> String { + format!("{:#x}", self.tx_hash()) + } + + fn to_row(self, builder_name: String) -> Self::ClickhouseRowType { + (self, builder_name).into() + } +} + /// Tests to make sure round-trip conversion between raw bundle and clickhouse bundle types is /// feasible. #[cfg(test)] diff --git a/src/indexer/mod.rs b/src/indexer/mod.rs index 0fea37e5..0a6f01ce 100644 --- a/src/indexer/mod.rs +++ b/src/indexer/mod.rs @@ -11,7 +11,7 @@ use crate::{ cli::IndexerArgs, indexer::{click::ClickhouseIndexer, parq::ParquetIndexer}, metrics::IndexerMetrics, - primitives::{BundleReceipt, SystemBundle}, + primitives::{BundleReceipt, SystemBundle, SystemTransaction}, }; pub(crate) mod click; @@ -34,7 +34,7 @@ pub const BUNDLE_TABLE_NAME: &str = "bundles"; pub const BUNDLE_RECEIPTS_TABLE_NAME: &str = "bundle_receipts"; /// The name of the Clickhouse table to store transactions in. -pub const TRANSACTIONS_TABLE_NAME: &str = "transactions"; +pub const TRANSACTIONS_TABLE_NAME: &str = "transactions_from_send_raw_tx"; /// The path of the backup database for storing failed Clickhouse batch insertions pub const BACKUP_DATABASE_PATH: &str = "/var/lib/buildernet-of-proxy/clickhouse-backup.db"; @@ -46,6 +46,7 @@ const TARGET_INDEXER: &str = "indexer"; pub(crate) trait OrderIndexer: Sync + Send { fn index_bundle(&self, system_bundle: SystemBundle); fn index_bundle_receipt(&self, bundle_receipt: BundleReceipt); + fn index_transaction(&self, system_transaction: SystemTransaction); } /// The collection of channel senders to send data to be indexed. @@ -53,6 +54,7 @@ pub(crate) trait OrderIndexer: Sync + Send { pub(crate) struct OrderSenders { bundle_tx: mpsc::Sender, bundle_receipt_tx: mpsc::Sender, + transaction_tx: mpsc::Sender, } /// The collection of channel receivers to receive data to be indexed. @@ -60,6 +62,7 @@ pub(crate) struct OrderSenders { pub(crate) struct OrderReceivers { bundle_rx: mpsc::Receiver, bundle_receipt_rx: mpsc::Receiver, + transaction_rx: mpsc::Receiver, } impl OrderSenders { @@ -67,8 +70,9 @@ impl OrderSenders { pub(crate) fn new() -> (Self, OrderReceivers) { let (bundle_tx, bundle_rx) = mpsc::channel(BUNDLE_INDEXER_BUFFER_SIZE); let (bundle_receipt_tx, bundle_receipt_rx) = mpsc::channel(BUNDLE_INDEXER_BUFFER_SIZE); - let senders = Self { bundle_tx, bundle_receipt_tx }; - let receivers = OrderReceivers { bundle_rx, bundle_receipt_rx }; + let (transaction_tx, transaction_rx) = mpsc::channel(TRANSACTION_INDEXER_BUFFER_SIZE); + let senders = Self { bundle_tx, bundle_receipt_tx, transaction_tx }; + let receivers = OrderReceivers { bundle_rx, bundle_receipt_rx, transaction_rx }; (senders, receivers) } } @@ -161,6 +165,21 @@ impl OrderIndexer for IndexerHandle { } } } + + fn index_transaction(&self, system_transaction: SystemTransaction) { + if let Err(e) = self.senders.transaction_tx.try_send(system_transaction) { + match e { + mpsc::error::TrySendError::Full(tx) => { + tracing::error!(target: TARGET_INDEXER, tx_hash = ?tx.tx_hash(), "CRITICAL: Failed to send transaction to index, channel is full"); + self.metrics.transaction_indexing_failures("Full").inc(); + } + mpsc::error::TrySendError::Closed(_) => { + tracing::error!(target: TARGET_INDEXER, "CRITICAL: Failed to send transaction to index, indexer task is closed"); + self.metrics.transaction_indexing_failures("Closed").inc(); + } + } + } + } } /// A mock indexer that simply drains the channels. @@ -170,9 +189,10 @@ impl MockIndexer { fn run(self, receivers: OrderReceivers, task_executor: TaskExecutor) { tracing::info!(target: TARGET_INDEXER, "Running with mocked indexer"); - let OrderReceivers { mut bundle_rx, mut bundle_receipt_rx } = receivers; + let OrderReceivers { mut bundle_rx, mut bundle_receipt_rx, mut transaction_rx } = receivers; task_executor.spawn(async move { while let Some(_b) = bundle_rx.recv().await {} }); task_executor.spawn(async move { while let Some(_b) = bundle_receipt_rx.recv().await {} }); + task_executor.spawn(async move { while let Some(_t) = transaction_rx.recv().await {} }); } } #[cfg(test)] diff --git a/src/indexer/parq.rs b/src/indexer/parq.rs index 738f0faa..37b5cf7f 100644 --- a/src/indexer/parq.rs +++ b/src/indexer/parq.rs @@ -177,7 +177,7 @@ impl ParquetIndexer { receivers: OrderReceivers, task_executor: TaskExecutor, ) -> io::Result<()> { - let OrderReceivers { mut bundle_rx, bundle_receipt_rx } = receivers; + let OrderReceivers { mut bundle_rx, bundle_receipt_rx, mut transaction_rx } = receivers; let parquet_file = OpenOptions::new().create(true).append(true).open( parquet_args.bundle_receipts_file_path.expect("bundle receipts file path is set"), @@ -203,6 +203,7 @@ impl ParquetIndexer { }; task_executor.spawn(async move { while let Some(_b) = bundle_rx.recv().await {} }); + task_executor.spawn(async move { while let Some(_t) = transaction_rx.recv().await {} }); task_executor.spawn_with_graceful_shutdown_signal(|mut shutdown| async move { tokio::select! { _ = runner.run_loop() => { diff --git a/src/indexer/ser.rs b/src/indexer/ser.rs index aa597776..d85abe12 100644 --- a/src/indexer/ser.rs +++ b/src/indexer/ser.rs @@ -185,6 +185,18 @@ pub(super) mod address { } } +/// serializes `Vec` as raw bytes (Clickhouse `String`) instead of `Array(UInt8)`. +pub(super) mod raw_bytes { + use serde::{ser::Serializer, Serialize as _}; + + pub(crate) fn serialize( + bytes: &Vec, + serializer: S, + ) -> Result { + bytes.as_slice().serialize(serializer) + } +} + pub(super) mod addresses { use alloy_primitives::Address; use serde::{ diff --git a/src/ingress/mod.rs b/src/ingress/mod.rs index 114f41e3..af70f28c 100644 --- a/src/ingress/mod.rs +++ b/src/ingress/mod.rs @@ -819,17 +819,16 @@ impl OrderflowIngress { self.order_cache.insert(tx_hash); - let system_transaction = - SystemTransaction::from_transaction(transaction, signer, received_at, priority); - - let tx = system_transaction.transaction.clone(); + let transaction = Arc::new(transaction); + let tx = transaction.clone(); // Spawn expensive operations like ECDSA recovery and consensus validation. - self.pqueues + let tx_sender = self + .pqueues .spawn_with_priority(priority, move || { validate_transaction(&tx.decoded, received_at.utc.unix_timestamp() as u64)?; - tx.recover_signer()?; - Ok::<(), IngressError>(()) + let sender = tx.recover_signer()?; + Ok::(sender) }) .await .inspect_err(|e| { @@ -837,6 +836,16 @@ impl OrderflowIngress { self.user_metrics.validation_errors(e.to_string()).inc(); })?; + let system_transaction = SystemTransaction::from_arc_transaction( + transaction, + signer, + tx_sender, + received_at, + priority, + ); + + self.indexer_handle.index_transaction(system_transaction.clone()); + // Send request to all forwarders. self.forwarders.broadcast_order(system_transaction.into()).await; diff --git a/src/metrics.rs b/src/metrics.rs index 4bf9aded..955ae6af 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -142,6 +142,9 @@ pub struct IndexerMetrics { /// The total number of bundle receipt indexing failures. #[metric(labels = ["error"])] bundle_receipt_indexing_failures: Counter, + /// The total number of transaction indexing failures. + #[metric(labels = ["error"])] + transaction_indexing_failures: Counter, } #[derive(Debug, Clone)] diff --git a/src/primitives/mod.rs b/src/primitives/mod.rs index 0f663476..d4cc904d 100644 --- a/src/primitives/mod.rs +++ b/src/primitives/mod.rs @@ -353,10 +353,12 @@ pub struct SystemTransaction { /// Ethereum transaction. #[deref] pub transaction: Arc, - /// The original transaction signer. + /// The original transaction signer (from Flashbots signature header). pub signer: Address, + /// The recovered transaction sender (from ECDSA signature recovery). + pub tx_sender: Address, - /// The timestamp at which the bundle has first been seen from the local operator. + /// The timestamp at which the transaction has first been seen from the local operator. pub received_at: UtcInstant, pub priority: Priority, } @@ -366,10 +368,22 @@ impl SystemTransaction { pub fn from_transaction( transaction: EthereumTransaction, signer: Address, + tx_sender: Address, received_at: UtcInstant, priority: Priority, ) -> Self { - Self { transaction: Arc::new(transaction), signer, received_at, priority } + Self { transaction: Arc::new(transaction), signer, tx_sender, received_at, priority } + } + + /// Create a new system transaction from an already-wrapped Arc transaction. + pub fn from_arc_transaction( + transaction: Arc, + signer: Address, + tx_sender: Address, + received_at: UtcInstant, + priority: Priority, + ) -> Self { + Self { transaction, signer, tx_sender, received_at, priority } } /// Encode the system transaction in a JSON-RPC payload with params EIP-2718 encoded bytes.