Working end to end demo

Author: danyalprout

.env.example

@@ -10,7 +10,7 @@ TIPS_INGRESS_LOG_LEVEL=info
 
 # Maintenance
 TIPS_MAINTENANCE_DATABASE_URL=postgresql://postgres:postgres@localhost:5432/postgres
-TIPS_MAINTENANCE_RPC_NODE=http://localhost:8545
+TIPS_MAINTENANCE_RPC_NODE=http://localhost:2222
 TIPS_MAINTENANCE_KAFKA_BROKERS=localhost:9092
 TIPS_MAINTENANCE_KAFKA_TOPIC=tips-audit
 TIPS_MAINTENANCE_POLL_INTERVAL_MS=250
@@ -20,5 +20,25 @@ TIPS_MAINTENANCE_LOG_LEVEL=info
 TIPS_AUDIT_KAFKA_BROKERS=localhost:9092
 TIPS_AUDIT_KAFKA_TOPIC=tips-audit
 TIPS_AUDIT_KAFKA_GROUP_ID=local-audit
-TIPS_AUDIT_S3_BUCKET=tips
 TIPS_AUDIT_LOG_LEVEL=info
+TIPS_AUDIT_S3_BUCKET=tips
+TIPS_AUDIT_S3_CONFIG_TYPE=manual
+TIPS_AUDIT_S3_ENDPOINT=http://localhost:7000
+TIPS_AUDIT_S3_REGION=us-east-1
+TIPS_AUDIT_S3_ACCESS_KEY_ID=minioadmin
+TIPS_AUDIT_S3_SECRET_ACCESS_KEY=minioadmin
+
+# For production AWS (uncomment and remove manual config above)
+# TIPS_AUDIT_S3_CONFIG_TYPE=aws
+
+# TIPS UI
+TIPS_DATABASE_URL=postgresql://postgres:postgres@localhost:5432/postgres
+TIPS_UI_AWS_REGION=us-east-1
+TIPS_UI_S3_BUCKET_NAME=tips
+TIPS_UI_S3_CONFIG_TYPE=manual
+TIPS_UI_S3_ENDPOINT=http://localhost:7000
+TIPS_UI_S3_ACCESS_KEY_ID=minioadmin
+TIPS_UI_S3_SECRET_ACCESS_KEY=minioadmin
+
+# For production AWS (uncomment and remove manual config above)
+# TIPS_UI_S3_CONFIG_TYPE=aws

.gitignore

@@ -31,3 +31,6 @@ ui/.env*.local
 ui/.vercel
 ui/*.tsbuildinfo
 ui/next-env.d.ts
+
+# Local data directories
+data/

Cargo.toml

@@ -26,6 +26,7 @@ alloy-rlp = "0.3.12"
 # op-alloy
 op-alloy-rpc-types = { version = "0.20.0" }
 op-alloy-consensus = { version = "0.20.0", features = ["k256"] }
+op-alloy-network = {version = "0.20.0"}
 
 tokio = { version = "1.47.1", features = ["full"] }
 tracing = "0.1.41"
@@ -55,6 +56,7 @@ futures-util = "0.3.32"
 rdkafka = { version = "0.37.0", features = ["libz-static"] }
 aws-config = "1.1.7"
 aws-sdk-s3 = "1.106.0"
+aws-credential-types = "1.1.7"
 bytes = { version = "1.8.0", features = ["serde"] }
 md5 = "0.7.0"
 base64 = "0.22.1"

README.md

@@ -1,5 +1,5 @@
 # tips
-A prototype of a Transaction Inclusion Pipeline Service for private sequencers that does not use the P2P mempool. The project
+A prototype of a Transaction Inclusion & Prioritization Stack for private sequencers that does not use the P2P mempool. The project
 aims to increase throughput, improve transaction tracking, reduce latency and add support for bundles.
 
 This project is currently at:
@@ -64,7 +64,7 @@ just ui
 ## Dev Notes
 
 ### Services Access
-- **MinIO UI**: http://localhost:9001 (minioadmin/minioadmin)
+- **MinIO UI**: http://localhost:7001 (minioadmin/minioadmin)
 - **PostgreSQL**: localhost:5432 (postgres/postgres)
 - **Kafka**: localhost:9092
 

crates/audit/Cargo.toml

@@ -29,6 +29,7 @@ dotenvy = { workspace = true }
 rdkafka = { workspace = true }
 aws-config = { workspace = true }
 aws-sdk-s3 = { workspace = true }
+aws-credential-types = { workspace = true }
 bytes = { workspace = true }
 md5 = { workspace = true }
 futures = "0.3.31"

crates/audit/src/bin/main.rs

@@ -1,13 +1,21 @@
 use anyhow::Result;
-use aws_sdk_s3::Client as S3Client;
-use clap::Parser;
+use aws_config::{BehaviorVersion, Region};
+use aws_credential_types::Credentials;
+use aws_sdk_s3::{config::Builder as S3ConfigBuilder, Client as S3Client};
+use clap::{Parser, ValueEnum};
 use rdkafka::consumer::Consumer;
 use tips_audit::{
     create_kafka_consumer, KafkaMempoolArchiver, KafkaMempoolReader, S3MempoolEventReaderWriter,
 };
 use tracing::{info, warn};
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 
+#[derive(Debug, Clone, ValueEnum)]
+enum S3ConfigType {
+    Aws,
+    Manual,
+}
+
 #[derive(Parser, Debug)]
 #[command(author, version, about, long_about = None)]
 struct Args {
@@ -25,6 +33,21 @@ struct Args {
 
     #[arg(long, env = "TIPS_AUDIT_LOG_LEVEL", default_value = "info")]
     log_level: String,
+
+    #[arg(long, env = "TIPS_AUDIT_S3_CONFIG_TYPE", default_value = "aws")]
+    s3_config_type: S3ConfigType,
+
+    #[arg(long, env = "TIPS_AUDIT_S3_ENDPOINT")]
+    s3_endpoint: Option<String>,
+
+    #[arg(long, env = "TIPS_AUDIT_S3_REGION", default_value = "us-east-1")]
+    s3_region: String,
+
+    #[arg(long, env = "TIPS_AUDIT_S3_ACCESS_KEY_ID")]
+    s3_access_key_id: Option<String>,
+
+    #[arg(long, env = "TIPS_AUDIT_S3_SECRET_ACCESS_KEY")]
+    s3_secret_access_key: Option<String>,
 }
 
 #[tokio::main]
@@ -67,15 +90,47 @@ async fn main() -> Result<()> {
     let consumer = create_kafka_consumer(&args.kafka_brokers, &args.kafka_group_id)?;
     consumer.subscribe(&[&args.kafka_topic])?;
 
-    let reader = KafkaMempoolReader::new(consumer, args.kafka_topic)?;
+    let reader = KafkaMempoolReader::new(consumer, args.kafka_topic.clone())?;
 
-    let config = aws_config::load_defaults(aws_config::BehaviorVersion::latest()).await;
-    let s3_client = S3Client::new(&config);
-    let writer = S3MempoolEventReaderWriter::new(s3_client, args.s3_bucket);
+    let s3_client = create_s3_client(&args).await?;
+    let s3_bucket = args.s3_bucket.clone();
+    let writer = S3MempoolEventReaderWriter::new(s3_client, s3_bucket);
 
     let mut archiver = KafkaMempoolArchiver::new(reader, writer);
 
     info!("Audit archiver initialized, starting main loop");
 
     archiver.run().await
 }
+
+async fn create_s3_client(args: &Args) -> Result<S3Client> {
+    match args.s3_config_type {
+        S3ConfigType::Manual => {
+            let region = args.s3_region.clone();
+            let mut config_builder =
+                aws_config::defaults(BehaviorVersion::latest()).region(Region::new(region));
+
+            if let Some(endpoint) = &args.s3_endpoint {
+                config_builder = config_builder.endpoint_url(endpoint);
+            }
+
+            if let (Some(access_key), Some(secret_key)) =
+                (&args.s3_access_key_id, &args.s3_secret_access_key)
+            {
+                let credentials = Credentials::new(access_key, secret_key, None, None, "manual");
+                config_builder = config_builder.credentials_provider(credentials);
+            }
+
+            let config = config_builder.load().await;
+            let s3_config_builder = S3ConfigBuilder::from(&config).force_path_style(true);
+
+            info!(message = "manually configuring s3 client");
+            Ok(S3Client::from_conf(s3_config_builder.build()))
+        }
+        S3ConfigType::Aws => {
+            info!(message = "using aws s3 client");
+            let config = aws_config::load_defaults(BehaviorVersion::latest()).await;
+            Ok(S3Client::new(&config))
+        }
+    }
+}

crates/audit/tests/common/mod.rs

@@ -7,7 +7,9 @@ use uuid::Uuid;
 pub struct TestHarness {
     pub s3_client: aws_sdk_s3::Client,
     pub bucket_name: String,
+    #[allow(dead_code)] // TODO is read
     pub kafka_producer: FutureProducer,
+    #[allow(dead_code)] // TODO is read
     pub kafka_consumer: StreamConsumer,
     _minio_container: testcontainers::ContainerAsync<MinIO>,
     _kafka_container: testcontainers::ContainerAsync<Kafka>,

crates/ingress/Cargo.toml

@@ -13,6 +13,7 @@ tips-audit.workspace = true
 jsonrpsee.workspace = true
 alloy-rpc-types-mev.workspace = true
 alloy-primitives.workspace = true
+op-alloy-network.workspace = true
 alloy-provider.workspace = true
 tokio.workspace = true
 tracing.workspace = true

crates/ingress/src/main.rs

@@ -4,6 +4,7 @@ use jsonrpsee::server::Server;
 use rdkafka::ClientConfig;
 use rdkafka::producer::FutureProducer;
 use std::net::IpAddr;
+use op_alloy_network::Optimism;
 use tips_audit::KafkaMempoolEventPublisher;
 use tracing::{info, warn};
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
@@ -87,8 +88,9 @@ async fn main() -> anyhow::Result<()> {
         mempool_url = %config.mempool_url
     );
 
-    let provider: RootProvider = ProviderBuilder::new()
+    let provider: RootProvider<Optimism> = ProviderBuilder::new()
         .disable_recommended_fillers()
+        .network::<Optimism>()
         .connect_http(config.mempool_url);
 
     let bundle_store = PostgresDatastore::connect(config.database_url).await?;

crates/ingress/src/service.rs

@@ -8,6 +8,7 @@ use jsonrpsee::{
     proc_macros::rpc,
 };
 use op_alloy_consensus::OpTxEnvelope;
+use op_alloy_network::Optimism;
 use tips_audit::{MempoolEvent, MempoolEventPublisher};
 use tips_datastore::BundleDatastore;
 use tracing::{info, warn};
@@ -28,15 +29,15 @@ pub trait IngressApi {
 }
 
 pub struct IngressService<Store, Publisher> {
-    provider: RootProvider,
+    provider: RootProvider<Optimism>,
     datastore: Store,
     dual_write_mempool: bool,
     publisher: Publisher,
 }
 
 impl<Store, Publisher> IngressService<Store, Publisher> {
     pub fn new(
-        provider: RootProvider,
+        provider: RootProvider<Optimism>,
         datastore: Store,
         dual_write_mempool: bool,
         publisher: Publisher,