perf: deduplicate bundles with the same contents

Cargo.toml

@@ -98,3 +98,4 @@ tracing-subscriber = { version = "0.3.20", default-features = false }
 testcontainers-modules = { version = "0.11.2", default-features = false }
 metrics-exporter-prometheus = { version = "0.17.0", default-features = false }
 futures = { version = "0.3.31", default-features = false }
+moka = { version = "0.12.12", default-features = false }

crates/audit/Cargo.toml

@@ -18,7 +18,7 @@ metrics-derive.workspace = true
 tips-core = { workspace = true, features = ["test-utils"] }
 serde = { workspace = true, features = ["std", "derive"] }
 tokio = { workspace = true, features = ["full"] }
-uuid = { workspace = true, features = ["v4", "serde"] }
+uuid = { workspace = true, features = ["v5", "serde"] }
 tracing = { workspace = true, features = ["std"] }
 anyhow = { workspace = true, features = ["std"] }
 serde_json = { workspace = true, features = ["std"] }

crates/audit/src/storage.rs

@@ -601,7 +601,7 @@ mod tests {
     use crate::reader::Event;
     use crate::types::{BundleEvent, DropReason, UserOpDropReason, UserOpEvent};
     use alloy_primitives::{Address, B256, TxHash, U256};
-    use tips_core::test_utils::create_bundle_from_txn_data;
+    use tips_core::{BundleExtensions, test_utils::create_bundle_from_txn_data};
     use uuid::Uuid;
 
     fn create_test_event(key: &str, timestamp: i64, bundle_event: BundleEvent) -> Event {
@@ -616,7 +616,7 @@ mod tests {
     fn test_update_bundle_history_transform_adds_new_event() {
         let bundle_history = BundleHistory { history: vec![] };
         let bundle = create_bundle_from_txn_data();
-        let bundle_id = Uuid::new_v4();
+        let bundle_id = Uuid::new_v5(&Uuid::NAMESPACE_OID, bundle.bundle_hash().as_slice());
         let bundle_event = BundleEvent::Received {
             bundle_id,
             bundle: Box::new(bundle.clone()),
@@ -655,7 +655,7 @@ mod tests {
         };
 
         let bundle = create_bundle_from_txn_data();
-        let bundle_id = Uuid::new_v4();
+        let bundle_id = Uuid::new_v5(&Uuid::NAMESPACE_OID, bundle.bundle_hash().as_slice());
         let bundle_event = BundleEvent::Received {
             bundle_id,
             bundle: Box::new(bundle),
@@ -670,9 +670,9 @@ mod tests {
     #[test]
     fn test_update_bundle_history_transform_handles_all_event_types() {
         let bundle_history = BundleHistory { history: vec![] };
-        let bundle_id = Uuid::new_v4();
-
         let bundle = create_bundle_from_txn_data();
+        let bundle_id = Uuid::new_v5(&Uuid::NAMESPACE_OID, bundle.bundle_hash().as_slice());
+
         let bundle_event = BundleEvent::Received {
             bundle_id,
             bundle: Box::new(bundle),
@@ -717,7 +717,8 @@ mod tests {
     #[test]
     fn test_update_transaction_metadata_transform_adds_new_bundle() {
         let metadata = TransactionMetadata { bundle_ids: vec![] };
-        let bundle_id = Uuid::new_v4();
+        let bundle = create_bundle_from_txn_data();
+        let bundle_id = Uuid::new_v5(&Uuid::NAMESPACE_OID, bundle.bundle_hash().as_slice());
 
         let result = update_transaction_metadata_transform(metadata, bundle_id);
 
@@ -729,7 +730,8 @@ mod tests {
 
     #[test]
     fn test_update_transaction_metadata_transform_skips_existing_bundle() {
-        let bundle_id = Uuid::new_v4();
+        let bundle = create_bundle_from_txn_data();
+        let bundle_id = Uuid::new_v5(&Uuid::NAMESPACE_OID, bundle.bundle_hash().as_slice());
         let metadata = TransactionMetadata {
             bundle_ids: vec![bundle_id],
         };
@@ -741,8 +743,12 @@ mod tests {
 
     #[test]
     fn test_update_transaction_metadata_transform_adds_to_existing_bundles() {
-        let existing_bundle_id = Uuid::new_v4();
-        let new_bundle_id = Uuid::new_v4();
+        // Some different, dummy bundle IDs since create_bundle_from_txn_data() returns the same bundle ID
+        // Even if the same txn is contained across multiple bundles, the bundle ID will be different since the
+        // UUID is based on the bundle hash.
+        let existing_bundle_id = Uuid::parse_str("550e8400-e29b-41d4-a716-446655440000").unwrap();
+        let new_bundle_id = Uuid::parse_str("6ba7b810-9dad-11d1-80b4-00c04fd430c8").unwrap();
+
         let metadata = TransactionMetadata {
             bundle_ids: vec![existing_bundle_id],
         };

crates/audit/src/types.rs

@@ -140,7 +140,11 @@ impl BundleEvent {
                 format!("{bundle_id}-{block_hash}")
             }
             _ => {
-                format!("{}-{}", self.bundle_id(), Uuid::new_v4())
+                format!(
+                    "{}-{}",
+                    self.bundle_id(),
+                    Uuid::new_v5(&Uuid::NAMESPACE_OID, self.bundle_id().as_bytes())
+                )
             }
         }
     }
@@ -200,7 +204,11 @@ impl UserOpEvent {
                 format!("{user_op_hash}-{tx_hash}")
             }
             _ => {
-                format!("{}-{}", self.user_op_hash(), Uuid::new_v4())
+                format!(
+                    "{}-{}",
+                    self.user_op_hash(),
+                    Uuid::new_v5(&Uuid::NAMESPACE_OID, self.user_op_hash().as_slice())
+                )
             }
         }
     }

crates/audit/tests/common/mod.rs

@@ -35,7 +35,10 @@ impl TestHarness {
             .await;
 
         let s3_client = aws_sdk_s3::Client::new(&config);
-        let bucket_name = format!("test-bucket-{}", Uuid::new_v4());
+        let bucket_name = format!(
+            "test-bucket-{}",
+            Uuid::new_v5(&Uuid::NAMESPACE_OID, bundle.bundle_hash().as_slice())
+        );
 
         s3_client
             .create_bucket()

crates/audit/tests/integration_tests.rs

@@ -9,7 +9,7 @@ use tips_audit_lib::{
     storage::{BundleEventS3Reader, S3EventReaderWriter},
     types::{BundleEvent, DropReason, UserOpEvent},
 };
-use tips_core::test_utils::create_bundle_from_txn_data;
+use tips_core::{BundleExtensions, test_utils::create_bundle_from_txn_data};
 use uuid::Uuid;
 mod common;
 use common::TestHarness;
@@ -24,11 +24,12 @@ async fn test_kafka_publisher_s3_archiver_integration()
     let s3_writer =
         S3EventReaderWriter::new(harness.s3_client.clone(), harness.bucket_name.clone());
 
-    let test_bundle_id = Uuid::new_v4();
+    let bundle = create_bundle_from_txn_data();
+    let test_bundle_id = Uuid::new_v5(&Uuid::NAMESPACE_OID, bundle.bundle_hash().as_slice());
     let test_events = [
         BundleEvent::Received {
             bundle_id: test_bundle_id,
-            bundle: Box::new(create_bundle_from_txn_data()),
+            bundle: Box::new(bundle.clone()),
         },
         BundleEvent::Dropped {
             bundle_id: test_bundle_id,
@@ -45,6 +46,8 @@ async fn test_kafka_publisher_s3_archiver_integration()
     let mut consumer = KafkaAuditArchiver::new(
         KafkaAuditLogReader::new(harness.kafka_consumer, topic.to_string())?,
         s3_writer.clone(),
+        1,
+        100,
     );
 
     tokio::spawn(async move {