integrate with audit

Author: cody-wang-cb

Cargo.lock

@@ -216,4 +216,6 @@ opentelemetry = { version = "0.31", features = ["trace"] }
 
 # Base Path
 concurrent-queue = "2.5.0"
-tips-core = { git = "https://github.com/base/tips", rev = "c08eaa4fe10c26de8911609b41ddab4918698325", default-features = false }
+tips-core = { path = "/Users/codywang/src/tips/crates/core", default-features = false }
+tips-audit = { path = "/Users/codywang/src/tips/crates/audit" }
+rdkafka = { version = "0.37" }

Cargo.toml

@@ -135,6 +135,8 @@ reqwest = "0.12.23"
 k256 = "0.13.4"
 
 rollup-boost.workspace = true
+tips-audit.workspace = true
+rdkafka.workspace = true
 
 nanoid = { version = "0.4", optional = true }
 reth-ipc = { workspace = true, optional = true }

crates/op-rbuilder/Cargo.toml

@@ -62,6 +62,21 @@ pub struct OpRbuilderArgs {
     #[arg(long = "builder.backrun-bundle-buffer-size", default_value = "10000")]
     pub backrun_bundle_buffer_size: usize,
 
+    /// Path to Kafka properties file for audit events (enables Kafka audit if set)
+    #[arg(
+        long = "builder.audit-kafka-properties",
+        env = "AUDIT_KAFKA_PROPERTIES_FILE"
+    )]
+    pub audit_kafka_properties: Option<String>,
+
+    /// Kafka topic for audit events
+    #[arg(
+        long = "builder.audit-kafka-topic",
+        env = "AUDIT_KAFKA_TOPIC",
+        default_value = "tips-audit"
+    )]
+    pub audit_kafka_topic: String,
+
     /// Path to builder playgorund to automatically start up the node connected to it
     #[arg(
         long = "builder.playground",

crates/op-rbuilder/src/args/op.rs

@@ -36,10 +36,12 @@ use reth_revm::{State, context::Block};
 use reth_transaction_pool::{BestTransactionsAttributes, PoolTransaction};
 use revm::{DatabaseCommit, context::result::ResultAndState, interpreter::as_u64_saturated};
 use std::{sync::Arc, time::Instant};
+use tips_audit::BundleEvent;
 use tokio_util::sync::CancellationToken;
-use tracing::{debug, info, trace};
+use tracing::{debug, info, trace, warn};
 
 use crate::{
+    bundles::AuditSender,
     gas_limiter::AddressGasLimiter,
     metrics::OpRBuilderMetrics,
     primitives::reth::{ExecutionInfo, TxnExecutionResult},
@@ -82,6 +84,8 @@ pub struct OpPayloadBuilderCtx<ExtraCtx: Debug + Default = ()> {
     pub resource_metering: ResourceMetering,
     /// Backrun bundle store for storing backrun transactions
     pub backrun_bundle_store: crate::bundles::BackrunBundleStore,
+    /// Audit event channel for backrun bundle lifecycle tracking
+    pub audit_tx: Option<AuditSender>,
 }
 
 impl<ExtraCtx: Debug + Default> OpPayloadBuilderCtx<ExtraCtx> {
@@ -93,6 +97,19 @@ impl<ExtraCtx: Debug + Default> OpPayloadBuilderCtx<ExtraCtx> {
         Self { extra_ctx, ..self }
     }
 
+    /// Send an audit event if the audit channel is configured
+    fn send_audit_event(&self, event: BundleEvent) {
+        if let Some(ref audit_tx) = self.audit_tx {
+            if let Err(e) = audit_tx.send(event) {
+                warn!(
+                    target: "payload_builder",
+                    error = %e,
+                    "Failed to send audit event"
+                );
+            }
+        }
+    }
+
     /// Returns the parent block the payload will be build on.
     pub fn parent(&self) -> &SealedHeader {
         &self.config.parent_header
@@ -447,6 +464,17 @@ impl<ExtraCtx: Debug + Default> OpPayloadBuilderCtx<ExtraCtx> {
 
             num_txs_considered += 1;
 
+            // Look up bundle_id for this tx (registered via base_txBundleId RPC)
+            let bundle_id = self.backrun_bundle_store.get_tx_bundle_id(&tx_hash);
+
+            // Emit StartExecuting audit event - tx is about to be executed
+            self.send_audit_event(BundleEvent::StartExecuting {
+                bundle_id,
+                tx_hash,
+                block_number: self.block_number(),
+                timestamp_ms: chrono::Utc::now().timestamp_millis(),
+            });
+
             let _resource_usage = self.resource_metering.get(&tx_hash);
 
             // TODO: ideally we should get this from the txpool stream
@@ -607,11 +635,89 @@ impl<ExtraCtx: Debug + Default> OpPayloadBuilderCtx<ExtraCtx> {
 
             info!(message = "Executed transaction", tx_hash = ?tx_hash);
 
+            // Emit Executed audit event - tx successfully executed and committed
+            self.send_audit_event(BundleEvent::Executed {
+                bundle_id,
+                tx_hash,
+                block_number: self.block_number(),
+                gas_used,
+                timestamp_ms: chrono::Utc::now().timestamp_millis(),
+            });
+
             // Execute backrun bundles for this transaction if it succeeded
             if is_success && let Some(backrun_bundles) = self.backrun_bundle_store.get(&tx_hash) {
-                info!(target: "backrun_bundles", target_tx = ?tx_hash, bundle_count = backrun_bundles.len(), "Found backrun bundles for transaction");
+                info!(message = "Found backrun bundles for transaction", tx_hash = ?tx_hash, bundle_count = backrun_bundles.len());
                 self.metrics.backrun_target_txs_found_total.increment(1);
-                // TODO: Execute backrun bundles
+
+                for stored_bundle in backrun_bundles {
+                    let bundle_id = stored_bundle.bundle_id;
+                    for backrun_tx in stored_bundle.backrun_txs {
+                        let ResultAndState { result, state } = match evm.transact(&backrun_tx) {
+                            Ok(res) => res,
+                            Err(err) => {
+                                return Err(PayloadBuilderError::evm(err));
+                            }
+                        };
+
+                        let backrun_gas_used = result.gas_used();
+                        let is_backrun_success = result.is_success();
+
+                        if !is_backrun_success {
+                            info!(message = "Backrun transaction reverted", tx_hash = ?backrun_tx.tx_hash(), bundle_id = %bundle_id);
+
+                            // Emit BackrunBundleExecuted audit event for reverted tx
+                            self.send_audit_event(BundleEvent::BackrunBundleExecuted {
+                                bundle_id,
+                                target_tx_hash: tx_hash,
+                                backrun_tx_hash: backrun_tx.tx_hash(),
+                                block_number: self.block_number(),
+                                gas_used: backrun_gas_used,
+                                success: false,
+                                timestamp_ms: chrono::Utc::now().timestamp_millis(),
+                            });
+
+                            continue;
+                        }
+
+                        info!(message = "Backrun transaction succeeded", tx_hash = ?backrun_tx.tx_hash(), bundle_id = %bundle_id);
+
+                        info.cumulative_gas_used += backrun_gas_used;
+                        info.cumulative_da_bytes_used += backrun_tx.encoded_2718().len() as u64;
+
+                        let ctx = ReceiptBuilderCtx {
+                            tx: backrun_tx.inner(),
+                            evm: &evm,
+                            result,
+                            state: &state,
+                            cumulative_gas_used: info.cumulative_gas_used,
+                        };
+                        info.receipts.push(self.build_receipt(ctx, None));
+
+                        // commit changes
+                        evm.db_mut().commit(state);
+
+                        // Emit BackrunBundleExecuted audit event for successful tx
+                        self.send_audit_event(BundleEvent::BackrunBundleExecuted {
+                            bundle_id,
+                            target_tx_hash: tx_hash,
+                            backrun_tx_hash: backrun_tx.tx_hash(),
+                            block_number: self.block_number(),
+                            gas_used: backrun_gas_used,
+                            success: true,
+                            timestamp_ms: chrono::Utc::now().timestamp_millis(),
+                        });
+
+                        // update add to total fees
+                        let miner_fee = backrun_tx
+                            .effective_tip_per_gas(base_fee)
+                            .expect("fee is always valid; execution succeeded");
+                        info.total_fees += U256::from(miner_fee) * U256::from(backrun_gas_used);
+
+                        // append sender and transaction to the respective lists
+                        info.executed_senders.push(backrun_tx.signer());
+                        info.executed_transactions.push(backrun_tx.into_inner());
+                    }
+                }
             }
         }
 

crates/op-rbuilder/src/builders/context.rs

@@ -1,6 +1,6 @@
 use crate::{
     builders::{BuilderConfig, OpPayloadBuilderCtx, flashblocks::FlashblocksConfig},
-    bundles::BackrunBundleStore,
+    bundles::{AuditSender, BackrunBundleStore},
     gas_limiter::{AddressGasLimiter, args::GasLimiterArgs},
     metrics::OpRBuilderMetrics,
     resource_metering::ResourceMetering,
@@ -35,6 +35,8 @@ pub(super) struct OpPayloadSyncerCtx {
     resource_metering: ResourceMetering,
     /// Backrun bundle store
     backrun_bundle_store: BackrunBundleStore,
+    /// Audit event channel
+    audit_tx: Option<AuditSender>,
 }
 
 impl OpPayloadSyncerCtx {
@@ -48,6 +50,7 @@ impl OpPayloadSyncerCtx {
         Client: ClientBounds,
     {
         let chain_spec = client.chain_spec();
+        let audit_tx = builder_config.backrun_bundle_store.audit_tx();
         Ok(Self {
             evm_config,
             da_config: builder_config.da_config.clone(),
@@ -56,6 +59,7 @@ impl OpPayloadSyncerCtx {
             metrics,
             resource_metering: builder_config.resource_metering,
             backrun_bundle_store: builder_config.backrun_bundle_store,
+            audit_tx,
         })
     }
 
@@ -90,6 +94,7 @@ impl OpPayloadSyncerCtx {
             address_gas_limiter: AddressGasLimiter::new(GasLimiterArgs::default()),
             resource_metering: self.resource_metering.clone(),
             backrun_bundle_store: self.backrun_bundle_store.clone(),
+            audit_tx: self.audit_tx,
         }
     }
 }

crates/op-rbuilder/src/builders/flashblocks/ctx.rs

@@ -284,6 +284,7 @@ where
             address_gas_limiter: self.address_gas_limiter.clone(),
             resource_metering: self.config.resource_metering.clone(),
             backrun_bundle_store: self.config.backrun_bundle_store.clone(),
+            audit_tx: self.config.backrun_bundle_store.audit_tx(),
         })
     }
 

crates/op-rbuilder/src/builders/flashblocks/payload.rs

@@ -253,6 +253,7 @@ where
             address_gas_limiter: self.address_gas_limiter.clone(),
             resource_metering: self.config.resource_metering.clone(),
             backrun_bundle_store: self.config.backrun_bundle_store.clone(),
+            audit_tx: self.config.backrun_bundle_store.audit_tx(),
         };
 
         let builder = OpBuilder::new(best);