Index private transactions

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.';

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(

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<F>(PhantomData<F>)
 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<UpdateTransactionsSizeCallback>,
+        >(
+            &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

src/indexer/click/models.rs

@@ -1,20 +1,20 @@
 //! 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;
 use rbuilder_utils::clickhouse::backup::primitives::{ClickhouseIndexableData, ClickhouseRowExt};
 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<u8>,
+}
+
+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) -> &<Self as Row>::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 = <TransactionRow as ClickhouseRowExt>::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)]

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,29 +46,33 @@ 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.
 #[derive(Debug, Clone)]
 pub(crate) struct OrderSenders {
     bundle_tx: mpsc::Sender<SystemBundle>,
     bundle_receipt_tx: mpsc::Sender<BundleReceipt>,
+    transaction_tx: mpsc::Sender<SystemTransaction>,
 }
 
 /// The collection of channel receivers to receive data to be indexed.
 #[derive(Debug)]
 pub(crate) struct OrderReceivers {
     bundle_rx: mpsc::Receiver<SystemBundle>,
     bundle_receipt_rx: mpsc::Receiver<BundleReceipt>,
+    transaction_rx: mpsc::Receiver<SystemTransaction>,
 }
 
 impl OrderSenders {
     /// Creates a new set of order indexer channel senders and receivers.
     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)]

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() => {