chore: setup an in memory bundle store

Author: danyalprout

Cargo.lock

@@ -7,18 +7,18 @@ homepage = "https://github.com/base/tips"
 repository = "https://github.com/base/tips"
 
 [workspace]
-members = ["crates/audit", "crates/ingress-rpc", "crates/common"]
+members = ["crates/audit", "crates/ingress-rpc", "crates/bundle-pool", "crates/core"]
 resolver = "2"
 
 [workspace.dependencies]
-tips-common = { path = "crates/common" }
 tips-audit = { path = "crates/audit" }
+tips-bundle-pool = { path = "crates/bundle-pool" }
+tips-core = { path = "crates/core" }
 
 # Reth
-reth = { git = "https://github.com/paradigmxyz/reth", tag = "v1.8.1" }
-reth-rpc-eth-types = { git = "https://github.com/paradigmxyz/reth", tag = "v1.8.1" }
-reth-optimism-evm = { git = "https://github.com/paradigmxyz/reth", tag = "v1.8.1" }
-base-reth-flashblocks-rpc = { git = "https://github.com/base/node-reth", rev = "a1ae148a36354c88b356f80281fef12dad9f7737" }
+reth = { git = "https://github.com/paradigmxyz/reth", tag = "v1.8.2" }
+reth-rpc-eth-types = { git = "https://github.com/paradigmxyz/reth", tag = "v1.8.2" }
+reth-optimism-evm = { git = "https://github.com/paradigmxyz/reth", tag = "v1.8.2" }
 
 # alloy
 alloy-primitives = { version = "1.3.1", default-features = false, features = [
@@ -28,29 +28,20 @@ alloy-primitives = { version = "1.3.1", default-features = false, features = [
 alloy-rpc-types = { version = "1.0.35", default-features = false }
 alloy-consensus = { version = "1.0.35" }
 alloy-provider = { version = "1.0.35" }
-alloy-rpc-types-mev = "1.0.35"
 
 # op-alloy
-op-alloy-rpc-types = { version = "0.20.0", default-features = false }
-op-alloy-network = { version = "0.20.0", default-features = false }
-op-alloy-consensus = { version = "0.20.0", features = ["k256"] }
+op-alloy-network = { version = "0.21.0", default-features = false }
+op-alloy-consensus = { version = "0.21.0", features = ["k256"] }
+op-alloy-rpc-types = { version = "0.21.0", default-features = true}
 
 tokio = { version = "1.47.1", features = ["full"] }
 tracing = "0.1.41"
 tracing-subscriber = { version = "0.3.20", features = ["env-filter"] }
 anyhow = "1.0.99"
 clap = { version = "4.5.47", features = ["derive", "env"] }
 url = "2.5.7"
-sqlx = { version = "0.8.6", features = [
-    "runtime-tokio-native-tls",
-    "postgres",
-    "uuid",
-    "chrono",
-    "json",
-]}
 uuid = { version = "1.18.1", features = ["v4", "serde"] }
 serde = { version = "1.0.219", features = ["derive"] }
-eyre = "0.6.12"
 async-trait = "0.1.89"
 serde_json = "1.0.143"
 dotenvy = "0.15.7"

Cargo.toml

@@ -12,6 +12,7 @@ name = "tips-audit"
 path = "src/bin/main.rs"
 
 [dependencies]
+tips-core = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true }
@@ -23,7 +24,6 @@ async-trait = { workspace = true }
 alloy-primitives = { workspace = true }
 alloy-consensus = { workspace = true }
 alloy-provider = { workspace = true }
-alloy-rpc-types-mev = { workspace = true }
 op-alloy-consensus = { workspace = true }
 clap = { workspace = true }
 dotenvy = { workspace = true }

crates/audit/Cargo.toml

@@ -3,7 +3,7 @@ use anyhow::Result;
 use async_trait::async_trait;
 use rdkafka::producer::{FutureProducer, FutureRecord};
 use serde_json;
-use tracing::{debug, error};
+use tracing::{debug, error, info};
 
 #[async_trait]
 pub trait BundleEventPublisher: Send + Sync {
@@ -70,3 +70,37 @@ impl BundleEventPublisher for KafkaBundleEventPublisher {
         Ok(())
     }
 }
+
+#[derive(Clone)]
+pub struct LoggingBundleEventPublisher;
+
+impl LoggingBundleEventPublisher {
+    pub fn new() -> Self {
+        Self
+    }
+}
+
+impl Default for LoggingBundleEventPublisher {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[async_trait]
+impl BundleEventPublisher for LoggingBundleEventPublisher {
+    async fn publish(&self, event: BundleEvent) -> Result<()> {
+        info!(
+            bundle_id = %event.bundle_id(),
+            event = ?event,
+            "Received bundle event"
+        );
+        Ok(())
+    }
+
+    async fn publish_all(&self, events: Vec<BundleEvent>) -> Result<()> {
+        for event in events {
+            self.publish(event).await?;
+        }
+        Ok(())
+    }
+}

crates/audit/src/publisher.rs

@@ -1,7 +1,6 @@
 use crate::reader::Event;
 use crate::types::{BundleEvent, BundleId, DropReason, TransactionId};
 use alloy_primitives::TxHash;
-use alloy_rpc_types_mev::EthSendBundle;
 use anyhow::Result;
 use async_trait::async_trait;
 use aws_sdk_s3::Client as S3Client;
@@ -11,6 +10,7 @@ use aws_sdk_s3::primitives::ByteStream;
 use serde::{Deserialize, Serialize};
 use std::fmt;
 use std::fmt::Debug;
+use tips_core::Bundle;
 use tracing::info;
 
 #[derive(Debug)]
@@ -39,12 +39,12 @@ pub enum BundleHistoryEvent {
     Created {
         key: String,
         timestamp: i64,
-        bundle: EthSendBundle,
+        bundle: Bundle,
     },
     Updated {
         key: String,
         timestamp: i64,
-        bundle: EthSendBundle,
+        bundle: Bundle,
     },
     Cancelled {
         key: String,
@@ -376,11 +376,11 @@ mod tests {
     use crate::reader::Event;
     use crate::types::{BundleEvent, DropReason};
     use alloy_primitives::TxHash;
-    use alloy_rpc_types_mev::EthSendBundle;
+    use tips_core::Bundle;
     use uuid::Uuid;
 
-    fn create_test_bundle() -> EthSendBundle {
-        EthSendBundle::default()
+    fn create_test_bundle() -> Bundle {
+        Bundle::default()
     }
 
     fn create_test_event(key: &str, timestamp: i64, bundle_event: BundleEvent) -> Event {
@@ -485,15 +485,15 @@ mod tests {
             block_number: 12345,
             block_hash: TxHash::from([1u8; 32]),
         };
-        let event = create_test_event("test-key-6", 1234567890, bundle_event);
+        let event = create_test_event("test-key-5", 1234567890, bundle_event);
         let result = update_bundle_history_transform(bundle_history.clone(), &event);
         assert!(result.is_some());
 
         let bundle_event = BundleEvent::Dropped {
             bundle_id,
             reason: DropReason::TimedOut,
         };
-        let event = create_test_event("test-key-7", 1234567890, bundle_event);
+        let event = create_test_event("test-key-6", 1234567890, bundle_event);
         let result = update_bundle_history_transform(bundle_history, &event);
         assert!(result.is_some());
     }

crates/audit/src/storage.rs

@@ -1,10 +1,10 @@
 use alloy_consensus::transaction::{SignerRecoverable, Transaction as ConsensusTransaction};
 use alloy_primitives::{Address, TxHash, U256};
 use alloy_provider::network::eip2718::Decodable2718;
-use alloy_rpc_types_mev::EthSendBundle;
 use bytes::Bytes;
 use op_alloy_consensus::OpTxEnvelope;
 use serde::{Deserialize, Serialize};
+use tips_core::Bundle;
 use uuid::Uuid;
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
@@ -19,13 +19,7 @@ pub type BundleId = Uuid;
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub enum DropReason {
     TimedOut,
-}
-
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct Bundle {
-    pub id: BundleId,
-    pub transactions: Vec<Transaction>,
-    pub metadata: serde_json::Value,
+    Reverted,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -39,11 +33,11 @@ pub struct Transaction {
 pub enum BundleEvent {
     Created {
         bundle_id: BundleId,
-        bundle: EthSendBundle,
+        bundle: Bundle,
     },
     Updated {
         bundle_id: BundleId,
-        bundle: EthSendBundle,
+        bundle: Bundle,
     },
     Cancelled {
         bundle_id: BundleId,

crates/audit/src/types.rs

@@ -1,11 +1,11 @@
-use alloy_rpc_types_mev::EthSendBundle;
 use std::time::Duration;
 use tips_audit::{
     KafkaMempoolArchiver, KafkaMempoolReader,
     publisher::{BundleEventPublisher, KafkaBundleEventPublisher},
     storage::{BundleEventS3Reader, S3EventReaderWriter},
     types::{BundleEvent, DropReason},
 };
+use tips_core::Bundle;
 use uuid::Uuid;
 mod common;
 use common::TestHarness;
@@ -23,7 +23,7 @@ async fn test_kafka_publisher_s3_archiver_integration()
     let test_events = vec![
         BundleEvent::Created {
             bundle_id: test_bundle_id,
-            bundle: EthSendBundle::default(),
+            bundle: Bundle::default(),
         },
         BundleEvent::Dropped {
             bundle_id: test_bundle_id,

crates/audit/tests/integration_tests.rs

@@ -1,11 +1,11 @@
 use alloy_primitives::{Bytes, TxHash, b256, bytes};
-use alloy_rpc_types_mev::EthSendBundle;
 use std::sync::Arc;
 use tips_audit::{
     reader::Event,
     storage::{BundleEventS3Reader, EventWriter, S3EventReaderWriter},
     types::BundleEvent,
 };
+use tips_core::Bundle;
 use tokio::task::JoinSet;
 use uuid::Uuid;
 
@@ -19,8 +19,8 @@ const TXN_DATA: Bytes = bytes!(
 const TXN_HASH: TxHash =
     b256!("0x4f7ddfc911f5cf85dd15a413f4cbb2a0abe4f1ff275ed13581958c0bcf043c5e");
 
-fn create_test_bundle() -> EthSendBundle {
-    EthSendBundle {
+fn create_test_bundle() -> Bundle {
+    Bundle {
         txs: vec![TXN_DATA.clone()],
         ..Default::default()
     }

crates/audit/tests/s3_test.rs

@@ -0,0 +1,28 @@
+[package]
+name = "tips-bundle-pool"
+version.workspace = true
+rust-version.workspace = true
+license.workspace = true
+homepage.workspace = true
+repository.workspace = true
+edition.workspace = true
+
+[dependencies]
+tips-core = { workspace = true }
+tips-audit.workspace = true
+uuid.workspace = true
+alloy-primitives.workspace = true
+
+tracing.workspace = true
+tokio.workspace = true
+anyhow.workspace = true
+async-trait.workspace = true
+
+[dev-dependencies]
+tips-core = { workspace = true, features = ["test-utils"] }
+alloy-consensus.workspace = true
+alloy-provider.workspace = true
+alloy-signer = "1.0.41"
+alloy-signer-local = "1.0.41"
+op-alloy-consensus.workspace = true
+op-alloy-rpc-types.workspace = true

crates/bundle-pool/Cargo.toml

@@ -0,0 +1,4 @@
+pub mod pool;
+
+pub use pool::{BundleStore, InMemoryBundlePool};
+pub use tips_core::{Bundle, BundleHash, BundleWithMetadata, CancelBundle};