fix: add back bundleWithMetadata (#52)

* add back bundleWithMetadata

* fix integration test

Author: wlawt

crates/bundle-pool/src/source.rs

@@ -3,7 +3,7 @@ use async_trait::async_trait;
 use rdkafka::consumer::{Consumer, StreamConsumer};
 use rdkafka::{ClientConfig, Message};
 use std::fmt::Debug;
-use tips_core::{Bundle, BundleWithMetadata};
+use tips_core::BundleWithMetadata;
 use tokio::sync::mpsc;
 use tracing::{error, trace};
 
@@ -52,29 +52,22 @@ impl BundleSource for KafkaBundleSource {
                         }
                     };
 
-                    let bundle: Bundle = match serde_json::from_slice(payload) {
-                        Ok(b) => b,
-                        Err(e) => {
-                            error!(error = %e, "Failed to deserialize bundle");
-                            continue;
-                        }
-                    };
+                    let bundle_with_metadata: BundleWithMetadata =
+                        match serde_json::from_slice(payload) {
+                            Ok(b) => b,
+                            Err(e) => {
+                                error!(error = %e, "Failed to deserialize bundle");
+                                continue;
+                            }
+                        };
 
                     trace!(
-                        bundle = ?bundle,
+                        bundle = ?bundle_with_metadata,
                         offset = message.offset(),
                         partition = message.partition(),
                         "Received bundle from Kafka"
                     );
 
-                    let bundle_with_metadata = match BundleWithMetadata::load(bundle) {
-                        Ok(b) => b,
-                        Err(e) => {
-                            error!(error = %e, "Failed to load bundle");
-                            continue;
-                        }
-                    };
-
                     if let Err(e) = self.publisher.send(bundle_with_metadata) {
                         error!(error = ?e, "Failed to publish bundle to queue");
                     }

crates/bundle-pool/tests/integration_tests.rs

@@ -65,7 +65,7 @@ async fn test_kafka_bundle_source_to_pool_integration() -> Result<(), Box<dyn st
     let test_bundle = create_test_bundle(vec![tx1], Some(100), None, None);
     let test_bundle_uuid = *test_bundle.uuid();
 
-    let bundle_payload = serde_json::to_string(test_bundle.bundle())?;
+    let bundle_payload = serde_json::to_string(&test_bundle)?;
 
     kafka_producer
         .send(

crates/core/src/types.rs

@@ -65,7 +65,7 @@ pub struct CancelBundle {
     pub replacement_uuid: String,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct BundleWithMetadata {
     bundle: Bundle,
     uuid: Uuid,

crates/ingress-rpc/src/queue.rs

@@ -3,14 +3,14 @@ use anyhow::Result;
 use async_trait::async_trait;
 use backon::{ExponentialBuilder, Retryable};
 use rdkafka::producer::{FutureProducer, FutureRecord};
-use tips_core::Bundle;
+use tips_core::BundleWithMetadata;
 use tokio::time::Duration;
 use tracing::{error, info};
 
 /// A queue to buffer transactions
 #[async_trait]
 pub trait QueuePublisher: Send + Sync {
-    async fn publish(&self, bundle: &Bundle, bundle_hash: &B256) -> Result<()>;
+    async fn publish(&self, bundle: &BundleWithMetadata, bundle_hash: &B256) -> Result<()>;
 }
 
 /// A queue to buffer transactions
@@ -27,7 +27,7 @@ impl KafkaQueuePublisher {
 
 #[async_trait]
 impl QueuePublisher for KafkaQueuePublisher {
-    async fn publish(&self, bundle: &Bundle, bundle_hash: &B256) -> Result<()> {
+    async fn publish(&self, bundle: &BundleWithMetadata, bundle_hash: &B256) -> Result<()> {
         let key = bundle_hash.to_string();
         let payload = serde_json::to_vec(&bundle)?;
 
@@ -75,7 +75,7 @@ impl QueuePublisher for KafkaQueuePublisher {
 mod tests {
     use super::*;
     use rdkafka::config::ClientConfig;
-    use tips_core::BundleWithMetadata;
+    use tips_core::{Bundle, BundleWithMetadata};
     use tokio::time::{Duration, Instant};
 
     fn create_test_bundle() -> Bundle {
@@ -97,7 +97,7 @@ mod tests {
         let bundle_hash = bundle_with_metadata.bundle_hash();
 
         let start = Instant::now();
-        let result = publisher.publish(&bundle, &bundle_hash).await;
+        let result = publisher.publish(&bundle_with_metadata, &bundle_hash).await;
         let elapsed = start.elapsed();
 
         // the backoff tries at minimum 100ms, so verify we tried at least once

crates/ingress-rpc/src/service.rs

@@ -70,7 +70,7 @@ where
         let bundle_hash = bundle_with_metadata.bundle_hash();
         if let Err(e) = self
             .bundle_queue
-            .publish(bundle_with_metadata.bundle(), &bundle_hash)
+            .publish(&bundle_with_metadata, &bundle_hash)
             .await
         {
             warn!(message = "Failed to publish bundle to queue", bundle_hash = %bundle_hash, error = %e);
@@ -124,7 +124,7 @@ where
 
         if let Err(e) = self
             .bundle_queue
-            .publish(bundle_with_metadata.bundle(), &bundle_hash)
+            .publish(&bundle_with_metadata, &bundle_hash)
             .await
         {
             warn!(message = "Failed to publish Queue::enqueue_bundle", bundle_hash = %bundle_hash, error = %e);