move publisher to ingresswriter

Author: wlawt

Cargo.lock

@@ -3,10 +3,8 @@ use clap::Parser;
 use jsonrpsee::server::Server;
 use op_alloy_network::Optimism;
 use rdkafka::ClientConfig;
-use rdkafka::consumer::Consumer;
 use rdkafka::producer::FutureProducer;
 use std::net::IpAddr;
-use tips_audit::{KafkaMempoolEventPublisher, create_kafka_consumer};
 use tracing::{info, warn};
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 use url::Url;
@@ -99,20 +97,14 @@ async fn main() -> anyhow::Result<()> {
         .network::<Optimism>()
         .connect_http(config.mempool_url);
 
-    let kafka_producer: FutureProducer = ClientConfig::new()
-        .set("bootstrap.servers", &config.kafka_brokers)
-        .set("message.timeout.ms", "5000")
-        .create()?;
-
     let queue_producer: FutureProducer = ClientConfig::new()
         .set("bootstrap.servers", &config.kafka_brokers)
         .set("message.timeout.ms", "5000")
         .create()?;
 
-    let publisher = KafkaMempoolEventPublisher::new(kafka_producer, config.kafka_topic);
     let queue = KafkaQueuePublisher::new(queue_producer, config.queue_topic.clone());
 
-    let service = IngressService::new(provider, config.dual_write_mempool, publisher, queue);
+    let service = IngressService::new(provider, config.dual_write_mempool, queue);
     let bind_addr = format!("{}:{}", config.address, config.port);
 
     let server = Server::builder().build(&bind_addr).await?;

crates/ingress/src/main.rs

@@ -11,7 +11,6 @@ use jsonrpsee::{
 use op_alloy_consensus::OpTxEnvelope;
 use op_alloy_network::Optimism;
 use reth_rpc_eth_types::EthApiError;
-use tips_audit::{MempoolEvent, MempoolEventPublisher};
 use tracing::{info, warn};
 
 use crate::queue::QueuePublisher;
@@ -31,33 +30,25 @@ pub trait IngressApi {
     async fn send_raw_transaction(&self, tx: Bytes) -> RpcResult<B256>;
 }
 
-pub struct IngressService<Publisher, Queue> {
+pub struct IngressService<Queue> {
     provider: RootProvider<Optimism>,
     dual_write_mempool: bool,
-    publisher: Publisher,
     queue: Queue,
 }
 
-impl<Publisher, Queue> IngressService<Publisher, Queue> {
-    pub fn new(
-        provider: RootProvider<Optimism>,
-        dual_write_mempool: bool,
-        publisher: Publisher,
-        queue: Queue,
-    ) -> Self {
+impl<Queue> IngressService<Queue> {
+    pub fn new(provider: RootProvider<Optimism>, dual_write_mempool: bool, queue: Queue) -> Self {
         Self {
             provider,
             dual_write_mempool,
-            publisher,
             queue,
         }
     }
 }
 
 #[async_trait]
-impl<Publisher, Queue> IngressApiServer for IngressService<Publisher, Queue>
+impl<Queue> IngressApiServer for IngressService<Queue>
 where
-    Publisher: MempoolEventPublisher + Sync + Send + 'static,
     Queue: QueuePublisher + Sync + Send + 'static,
 {
     async fn send_bundle(&self, _bundle: EthSendBundle) -> RpcResult<EthBundleHash> {

crates/ingress/src/service.rs

@@ -9,6 +9,7 @@ path = "src/main.rs"
 
 [dependencies]
 tips-datastore.workspace = true
+tips-audit.workspace=true
 alloy-rpc-types-mev.workspace = true
 tokio.workspace = true
 tracing.workspace = true

crates/writer/Cargo.toml

@@ -6,15 +6,16 @@ use rdkafka::{
     config::ClientConfig,
     consumer::{Consumer, StreamConsumer},
     message::Message,
+    producer::FutureProducer,
 };
+use tips_audit::{KafkaMempoolEventPublisher, MempoolEvent, MempoolEventPublisher};
 use tips_datastore::{BundleDatastore, postgres::PostgresDatastore};
 use tokio::time::Duration;
 use tracing::{debug, error, info};
 use uuid::Uuid;
 
 #[derive(Parser)]
-#[command(name = "tips-writer")]
-#[command(about = "TIPS Writer Service - Consumes bundles from Kafka and writes to datastore")]
+#[command(author, version, about, long_about = None)]
 struct Args {
     #[arg(long, env = "TIPS_WRITER_DATABASE_URL")]
     database_url: String,
@@ -33,24 +34,28 @@ struct Args {
 }
 
 /// IngressWriter consumes bundles sent from the Ingress service and writes them to the datastore
-pub struct IngressWriter<Store> {
+pub struct IngressWriter<Store, Publisher> {
     queue_consumer: StreamConsumer,
     datastore: Store,
+    publisher: Publisher,
 }
 
-impl<Store> IngressWriter<Store>
+impl<Store, Publisher> IngressWriter<Store, Publisher>
 where
     Store: BundleDatastore + Send + Sync + 'static,
+    Publisher: MempoolEventPublisher + Sync + Send + 'static,
 {
     pub fn new(
         queue_consumer: StreamConsumer,
         queue_topic: String,
         datastore: Store,
+        publisher: Publisher,
     ) -> Result<Self> {
         queue_consumer.subscribe(&[queue_topic.as_str()])?;
         Ok(Self {
             queue_consumer,
             datastore,
+            publisher,
         })
     }
 
@@ -75,7 +80,7 @@ where
                         .map_err(|e| anyhow::anyhow!("Failed to insert bundle: {e}"))
                 };
 
-                insert
+                let bundle_id = insert
                     .retry(
                         &ExponentialBuilder::default()
                             .with_min_delay(Duration::from_millis(100))
@@ -86,7 +91,19 @@ where
                         info!("Retrying to insert bundle {:?} after {:?}", err, dur);
                     })
                     .await
-                    .map_err(|e| anyhow::anyhow!("Failed to insert bundle after retries: {e}"))
+                    .map_err(|e| anyhow::anyhow!("Failed to insert bundle after retries: {e}"))?;
+
+                if let Err(e) = self
+                    .publisher
+                    .publish(MempoolEvent::Created {
+                        bundle_id,
+                        bundle: bundle.clone(),
+                    })
+                    .await
+                {
+                    error!(error = %e, bundle_id = %bundle_id, "Failed to publish MempoolEvent::Created");
+                }
+                Ok(bundle_id)
             }
             Err(e) => {
                 error!(error = %e, "Error receiving message from Kafka");
@@ -114,9 +131,15 @@ async fn main() -> Result<()> {
         .set("session.timeout.ms", "6000")
         .set("enable.auto.commit", "true");
 
+    let kafka_producer: FutureProducer = ClientConfig::new()
+        .set("bootstrap.servers", &args.kafka_brokers)
+        .set("message.timeout.ms", "5000")
+        .create()?;
+
+    let publisher = KafkaMempoolEventPublisher::new(kafka_producer, "tips-audit".to_string());
     let consumer = config.create()?;
     let datastore = PostgresDatastore::connect(args.database_url).await?;
-    let writer = IngressWriter::new(consumer, args.kafka_topic.clone(), datastore)?;
+    let writer = IngressWriter::new(consumer, args.kafka_topic.clone(), datastore, publisher)?;
 
     info!(
         "Ingress Writer service started, consuming from topic: {}",

crates/writer/src/main.rs

@@ -46,5 +46,8 @@ ingress:
 maintenance:
     cargo run --bin tips-maintenance
 
+writer:
+    cargo run --bin tips-writer
+
 ui:
     cd ui && yarn dev