nits

Author: wlawt

crates/ingress/src/main.rs

@@ -102,7 +102,7 @@ async fn main() -> anyhow::Result<()> {
         .set("message.timeout.ms", "5000")
         .create()?;
 
-    let queue = KafkaQueuePublisher::new(queue_producer, config.queue_topic.clone());
+    let queue = KafkaQueuePublisher::new(queue_producer, config.queue_topic);
 
     let service = IngressService::new(provider, config.dual_write_mempool, queue);
     let bind_addr = format!("{}:{}", config.address, config.port);

crates/ingress/src/service.rs

@@ -3,7 +3,6 @@ use alloy_primitives::{B256, Bytes};
 use alloy_provider::network::eip2718::Decodable2718;
 use alloy_provider::{Provider, RootProvider};
 use alloy_rpc_types_mev::{EthBundleHash, EthCancelBundle, EthSendBundle};
-use jsonrpsee::types::ErrorObject;
 use jsonrpsee::{
     core::{RpcResult, async_trait},
     proc_macros::rpc,
@@ -95,11 +94,6 @@ where
         let sender = transaction.signer();
         if let Err(e) = self.queue.publish(&bundle, sender).await {
             warn!(message = "Failed to publish Queue::enqueue_bundle", sender = %sender, error = %e);
-            return Err(ErrorObject::owned(
-                12,
-                "Failed to queue bundle",
-                Some(sender.to_string()),
-            ));
         }
 
         info!(message="queued singleton bundle", txn_hash=%transaction.tx_hash());

crates/writer/src/main.rs

@@ -138,8 +138,11 @@ async fn main() -> Result<()> {
 
     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, publisher)?;
+
+    let bundle_store = PostgresDatastore::connect(args.database_url).await?;
+    bundle_store.run_migrations().await?;
+
+    let writer = IngressWriter::new(consumer, args.kafka_topic.clone(), bundle_store, publisher)?;
 
     info!(
         "Ingress Writer service started, consuming from topic: {}",