From 2dab63a471012511f0c06942f5100557e560e066 Mon Sep 17 00:00:00 2001 From: Daniel Xifra Date: Thu, 19 Mar 2026 16:14:04 -0300 Subject: [PATCH 1/8] stores sendRawTx --- fixtures/create_transactions_table.sql | 23 ++++++ src/cli.rs | 11 ++- src/indexer/click/mod.rs | 44 ++++++++++-- src/indexer/click/models.rs | 99 +++++++++++++++++++++++++- src/indexer/mod.rs | 28 ++++++-- src/indexer/parq.rs | 3 +- src/ingress/mod.rs | 23 ++++-- src/metrics.rs | 3 + src/primitives/mod.rs | 20 +++++- 9 files changed, 232 insertions(+), 22 deletions(-) create mode 100644 fixtures/create_transactions_table.sql diff --git a/fixtures/create_transactions_table.sql b/fixtures/create_transactions_table.sql new file mode 100644 index 00000000..a69d7af1 --- /dev/null +++ b/fixtures/create_transactions_table.sql @@ -0,0 +1,23 @@ +CREATE TABLE transactions +( + `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 +) +ENGINE = MergeTree() +PARTITION BY toDate(received_at) +PRIMARY KEY hash +ORDER BY hash +SETTINGS index_granularity = 8192; 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..89dacc55 100644 --- a/src/indexer/click/models.rs +++ b/src/indexer/click/models.rs @@ -6,7 +6,7 @@ use crate::{ }; 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,101 @@ 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, + pub raw_tx: String, +} + +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: format!("0x{}", hex::encode(&tx.raw)), + } + } +} + +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..68319c65 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; @@ -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/ingress/mod.rs b/src/ingress/mod.rs index df99a529..f44bddeb 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. From 1564d08da0e50c634a618fec71148ecec5618051 Mon Sep 17 00:00:00 2001 From: Daniel Xifra Date: Fri, 27 Mar 2026 16:42:13 -0300 Subject: [PATCH 2/8] fixed raw_tx serialization --- src/indexer/click/models.rs | 7 ++++--- src/indexer/ser.rs | 9 +++++++++ 2 files changed, 13 insertions(+), 3 deletions(-) diff --git a/src/indexer/click/models.rs b/src/indexer/click/models.rs index 89dacc55..8aad8170 100644 --- a/src/indexer/click/models.rs +++ b/src/indexer/click/models.rs @@ -1,7 +1,7 @@ //! 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; @@ -447,7 +447,8 @@ pub struct TransactionRow { pub gas_fee_cap: String, pub data_size: i64, pub data_4bytes: String, - pub raw_tx: String, + #[serde(serialize_with = "raw_bytes::serialize")] + pub raw_tx: Vec, } impl ClickhouseRowExt for TransactionRow { @@ -500,7 +501,7 @@ impl From<(SystemTransaction, String)> for TransactionRow { gas_fee_cap: tx.decoded.max_fee_per_gas().to_string(), data_size: input.len() as i64, data_4bytes, - raw_tx: format!("0x{}", hex::encode(&tx.raw)), + raw_tx: tx.raw.to_vec(), } } } diff --git a/src/indexer/ser.rs b/src/indexer/ser.rs index aa597776..a76f8d5a 100644 --- a/src/indexer/ser.rs +++ b/src/indexer/ser.rs @@ -185,6 +185,15 @@ 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::{ From 303168aaaab1f8427b30c2b50744edd8413cd009 Mon Sep 17 00:00:00 2001 From: Daniel Xifra Date: Mon, 30 Mar 2026 13:17:11 -0300 Subject: [PATCH 3/8] better table --- fixtures/create_transactions_table.sql | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/fixtures/create_transactions_table.sql b/fixtures/create_transactions_table.sql index a69d7af1..958e324f 100644 --- a/fixtures/create_transactions_table.sql +++ b/fixtures/create_transactions_table.sql @@ -1,6 +1,7 @@ -CREATE TABLE transactions +CREATE TABLE mempool_dumpster.transactions ( - `received_at` DateTime64(3, 'UTC'), + `received_at` DateTime64(3, + 'UTC'), `hash` String, `chain_id` String, `tx_type` Int64, @@ -14,10 +15,14 @@ CREATE TABLE transactions `gas_fee_cap` String, `data_size` Int64, `data_4bytes` String, - `raw_tx` String + `raw_tx` String, + `ver` Int64 MATERIALIZED -toUnixTimestamp(received_at) ) -ENGINE = MergeTree() -PARTITION BY toDate(received_at) +ENGINE = ReplacingMergeTree() +PARTITION BY toYYYYMM(received_at) PRIMARY KEY hash ORDER BY hash -SETTINGS index_granularity = 8192; +SETTINGS index_granularity = 8192 +COMMENT 'Transaction details, + deduplicated by hash, + will keep the transaction with earliest received_at.'; \ No newline at end of file From 95dd0c2f047894bc171c745fa870d7c9e1d8acaf Mon Sep 17 00:00:00 2001 From: Daniel Xifra Date: Mon, 30 Mar 2026 13:19:14 -0300 Subject: [PATCH 4/8] fixed table --- fixtures/create_transactions_table.sql | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/fixtures/create_transactions_table.sql b/fixtures/create_transactions_table.sql index 958e324f..a945c1d0 100644 --- a/fixtures/create_transactions_table.sql +++ b/fixtures/create_transactions_table.sql @@ -1,4 +1,4 @@ -CREATE TABLE mempool_dumpster.transactions +CREATE TABLE transactions ( `received_at` DateTime64(3, 'UTC'), @@ -18,7 +18,7 @@ CREATE TABLE mempool_dumpster.transactions `raw_tx` String, `ver` Int64 MATERIALIZED -toUnixTimestamp(received_at) ) -ENGINE = ReplacingMergeTree() +ENGINE = ReplacingMergeTree(ver) PARTITION BY toYYYYMM(received_at) PRIMARY KEY hash ORDER BY hash From 41bdef9987ef7aae7ef77f90f205c02c59444cd3 Mon Sep 17 00:00:00 2001 From: Daniel Xifra Date: Mon, 30 Mar 2026 13:21:58 -0300 Subject: [PATCH 5/8] fixed table name --- fixtures/create_transactions_table.sql | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/fixtures/create_transactions_table.sql b/fixtures/create_transactions_table.sql index a945c1d0..269e9ea4 100644 --- a/fixtures/create_transactions_table.sql +++ b/fixtures/create_transactions_table.sql @@ -1,4 +1,4 @@ -CREATE TABLE transactions +CREATE TABLE transactions_from_send_raw_tx ( `received_at` DateTime64(3, 'UTC'), From 86a9a1aa01f44c7e152777332c1dadcc87d0faaf Mon Sep 17 00:00:00 2001 From: Daniel Xifra Date: Tue, 31 Mar 2026 16:21:27 -0300 Subject: [PATCH 6/8] fixed table name --- src/indexer/mod.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/indexer/mod.rs b/src/indexer/mod.rs index 68319c65..64db3834 100644 --- a/src/indexer/mod.rs +++ b/src/indexer/mod.rs @@ -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"; @@ -167,6 +167,7 @@ impl OrderIndexer for IndexerHandle { } fn index_transaction(&self, system_transaction: SystemTransaction) { + tracing::info!("DX index_transaction"); if let Err(e) = self.senders.transaction_tx.try_send(system_transaction) { match e { mpsc::error::TrySendError::Full(tx) => { From 07a2eda7d059192885b6a2348fec5db1e6567ef9 Mon Sep 17 00:00:00 2001 From: Daniel Xifra Date: Tue, 31 Mar 2026 16:40:53 -0300 Subject: [PATCH 7/8] remove trace --- src/indexer/mod.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/src/indexer/mod.rs b/src/indexer/mod.rs index 64db3834..0a6f01ce 100644 --- a/src/indexer/mod.rs +++ b/src/indexer/mod.rs @@ -167,7 +167,6 @@ impl OrderIndexer for IndexerHandle { } fn index_transaction(&self, system_transaction: SystemTransaction) { - tracing::info!("DX index_transaction"); if let Err(e) = self.senders.transaction_tx.try_send(system_transaction) { match e { mpsc::error::TrySendError::Full(tx) => { From aafa052c9ec4530f9bea2f8b4ba2a1f74b86ff4e Mon Sep 17 00:00:00 2001 From: Daniel Xifra Date: Wed, 8 Apr 2026 16:45:31 -0300 Subject: [PATCH 8/8] lint --- src/indexer/ser.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/indexer/ser.rs b/src/indexer/ser.rs index a76f8d5a..d85abe12 100644 --- a/src/indexer/ser.rs +++ b/src/indexer/ser.rs @@ -189,7 +189,10 @@ pub(super) mod address { pub(super) mod raw_bytes { use serde::{ser::Serializer, Serialize as _}; - pub(crate) fn serialize(bytes: &Vec, serializer: S) -> Result { + pub(crate) fn serialize( + bytes: &Vec, + serializer: S, + ) -> Result { bytes.as_slice().serialize(serializer) } }