[ingress] introduce buffer queue

crates/ingress/Cargo.toml

@@ -26,4 +26,7 @@ op-alloy-consensus.workspace = true
 eyre.workspace = true
 dotenvy.workspace = true
 rdkafka.workspace = true
-reth-rpc-eth-types.workspace = true
+reth-rpc-eth-types.workspace = true
+serde.workspace = true
+serde_json.workspace = true
+async-trait.workspace = true

crates/ingress/src/main.rs

@@ -10,7 +10,9 @@ use tracing::{info, warn};
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 use url::Url;
 
+mod queue;
 mod service;
+use queue::KafkaQueuePublisher;
 use service::{IngressApiServer, IngressService};
 use tips_datastore::PostgresDatastore;
 
@@ -49,6 +51,14 @@ struct Config {
     )]
     kafka_topic: String,
 
+    /// Kafka topic for queuing transactions before the DB Writer
+    #[arg(
+        long,
+        env = "TIPS_INGRESS_KAFKA_QUEUE_TOPIC",
+        default_value = "transaction-queue"
+    )]
+    queue_topic: String,
+
     #[arg(long, env = "TIPS_INGRESS_LOG_LEVEL", default_value = "info")]
     log_level: String,
 }
@@ -101,9 +111,21 @@ async fn main() -> anyhow::Result<()> {
         .set("message.timeout.ms", "5000")
         .create()?;
 
-    let publisher = KafkaMempoolEventPublisher::new(kafka_producer, config.kafka_topic);
+    let queue_producer: FutureProducer = ClientConfig::new()
+        .set("bootstrap.servers", &config.kafka_brokers)
+        .set("message.timeout.ms", "5000")
+        .create()?;
 
-    let service = IngressService::new(provider, bundle_store, config.dual_write_mempool, publisher);
+    let publisher = KafkaMempoolEventPublisher::new(kafka_producer, config.kafka_topic);
+    let queue = KafkaQueuePublisher::new(queue_producer, config.queue_topic);
+
+    let service = IngressService::new(
+        provider,
+        bundle_store,
+        config.dual_write_mempool,
+        publisher,
+        queue,
+    );
     let bind_addr = format!("{}:{}", config.address, config.port);
 
     let server = Server::builder().build(&bind_addr).await?;

crates/ingress/src/queue.rs

@@ -0,0 +1,68 @@
+use alloy_primitives::Address;
+use alloy_rpc_types_mev::EthSendBundle;
+use anyhow::{Error, Result};
+use async_trait::async_trait;
+use rdkafka::producer::{FutureProducer, FutureRecord};
+use tracing::{error, info};
+
+/// A queue to buffer transactions
+#[async_trait]
+pub trait QueuePublisher: Send + Sync {
+    async fn publish(&self, bundle: &EthSendBundle, sender: Address) -> Result<()>;
+}
+
+/// A queue to buffer transactions
+pub struct KafkaQueuePublisher {
+    producer: FutureProducer,
+    topic: String,
+}
+
+impl KafkaQueuePublisher {
+    pub fn new(producer: FutureProducer, topic: String) -> Self {
+        Self { producer, topic }
+    }
+
+    pub async fn enqueue_bundle(
+        &self,
+        bundle: &EthSendBundle,
+        sender: Address,
+    ) -> Result<(), Error> {
+        let key = sender.to_string();
+        let payload = serde_json::to_vec(bundle)?;
+
+        let record = FutureRecord::to(&self.topic).key(&key).payload(&payload);
+
+        match self
+            .producer
+            .send(record, tokio::time::Duration::from_secs(5))
+            .await
+        {
+            Ok((partition, offset)) => {
+                info!(
+                    sender = %sender,
+                    partition = partition,
+                    offset = offset,
+                    topic = %self.topic,
+                    "Successfully enqueued bundle"
+                );
+                Ok(())
+            }
+            Err((err, _)) => {
+                error!(
+                    sender = %sender,
+                    error = %err,
+                    topic = %self.topic,
+                    "Failed to enqueue bundle"
+                );
+                Err(anyhow::anyhow!("Failed to enqueue bundle: {}", err))
+            }
+        }
+    }
+}
+
+#[async_trait]
+impl QueuePublisher for KafkaQueuePublisher {
+    async fn publish(&self, bundle: &EthSendBundle, sender: Address) -> Result<()> {
+        self.enqueue_bundle(bundle, sender).await
+    }
+}

crates/ingress/src/service.rs

@@ -15,6 +15,8 @@ use tips_audit::{MempoolEvent, MempoolEventPublisher};
 use tips_datastore::BundleDatastore;
 use tracing::{info, warn};
 
+use crate::queue::QueuePublisher;
+
 #[rpc(server, namespace = "eth")]
 pub trait IngressApi {
     /// `eth_sendBundle` can be used to send your bundles to the builder.
@@ -30,34 +32,38 @@ pub trait IngressApi {
     async fn send_raw_transaction(&self, tx: Bytes) -> RpcResult<B256>;
 }
 
-pub struct IngressService<Store, Publisher> {
+pub struct IngressService<Store, Publisher, Queue> {
     provider: RootProvider<Optimism>,
     datastore: Store,
     dual_write_mempool: bool,
     publisher: Publisher,
+    queue: Queue,
 }
 
-impl<Store, Publisher> IngressService<Store, Publisher> {
+impl<Store, Publisher, Queue> IngressService<Store, Publisher, Queue> {
     pub fn new(
         provider: RootProvider<Optimism>,
         datastore: Store,
         dual_write_mempool: bool,
         publisher: Publisher,
+        queue: Queue,
     ) -> Self {
         Self {
             provider,
             datastore,
             dual_write_mempool,
             publisher,
+            queue,
         }
     }
 }
 
 #[async_trait]
-impl<Store, Publisher> IngressApiServer for IngressService<Store, Publisher>
+impl<Store, Publisher, Queue> IngressApiServer for IngressService<Store, Publisher, Queue>
 where
     Store: BundleDatastore + Sync + Send + 'static,
     Publisher: MempoolEventPublisher + Sync + Send + 'static,
+    Queue: QueuePublisher + Sync + Send + 'static,
 {
     async fn send_bundle(&self, _bundle: EthSendBundle) -> RpcResult<EthBundleHash> {
         warn!(
@@ -99,6 +105,13 @@ where
             ..Default::default()
         };
 
+        // queue the bundle
+        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);
+        }
+
+        // TODO: have DB Writer consume from the queue and move the insert_bundle logic there
         let result = self
             .datastore
             .insert_bundle(bundle.clone())