Audit tests WIP

Author: danyalprout

CLAUDE.md

@@ -3,7 +3,8 @@
 ## Notes
 - DO NOT ADD COMMENTS UNLESS INSTRUCTED
 - Put imports at the top of the file, never in functions
-- Always run `just ci` before claiming a task is complete and fix any issues
+- For rust tasks always run `just rust-ci` before claiming a task is complete and fix any issues
+- For typescript tasks always run `just ui-ci` before claiming a task is complete and fix any issues
 - Use `just fix` to fix formatting and warnings
 - Always add dependencies to the cargo.toml in the root and reference them in the crate cargo files
 - Always use the latest dependency versions. Use https://crates.io/ to find dependency versions when adding new deps

Cargo.toml

@@ -48,11 +48,12 @@ serde_json = "1.0.143"
 dotenvy = "0.15.7"
 testcontainers = { version = "0.23.1", features = ["blocking"] }
 testcontainers-modules = { version = "0.11.2", features = ["postgres", "kafka", "minio"] }
+futures-util = "0.3.32"
 
 # Kafka and S3 dependencies
 rdkafka = { version = "0.37.0", features = ["libz-static"] }
-aws-sdk-s3 = "1.62.0"
-aws-config = "1.6.0"
+aws-config = "1.1.7"
+aws-sdk-s3 = "1.106.0"
 bytes = { version = "1.8.0", features = ["serde"] }
 md5 = "0.7.0"
 base64 = "0.22.1"

crates/audit/Cargo.toml

@@ -22,15 +22,14 @@ clap = { workspace = true }
 
 # Additional dependencies for Kafka and S3
 rdkafka = { workspace = true }
-aws-sdk-s3 = { workspace = true }
 aws-config = { workspace = true }
+aws-sdk-s3 = { workspace = true }
 bytes = { workspace = true }
 md5 = { workspace = true }
-base64 = { workspace = true }
+futures = "0.3.31"
 
 [dev-dependencies]
 tokio-test = "0.4.4"
 testcontainers = { workspace = true }
 testcontainers-modules = { workspace = true }
-eyre = { workspace = true }
-aws-credential-types = "1.2"
+eyre = { workspace = true }

crates/audit/src/archiver.rs

@@ -1,17 +1,16 @@
-use std::time::Duration;
-use crate::reader::{KafkaMempoolReader, MempoolEventReader, TimestampedEvent};
+use crate::reader::{KafkaMempoolReader, MempoolEventReader};
 use crate::storage::{MempoolEventWriter, S3MempoolEventWriter};
 use anyhow::Result;
+use aws_sdk_s3::Client as S3Client;
+use std::time::Duration;
 use tokio::time::sleep;
 use tracing::{error, info};
 
-
 pub struct KafkaMempoolArchiver {
     reader: KafkaMempoolReader,
     writer: S3MempoolEventWriter,
 }
 
-
 impl KafkaMempoolArchiver {
     pub async fn new(
         kafka_brokers: &str,
@@ -20,7 +19,10 @@ impl KafkaMempoolArchiver {
         bucket: String,
     ) -> Result<Self> {
         let reader = KafkaMempoolReader::new(kafka_brokers, topic, group_id).await?;
-        let writer = S3MempoolEventWriter::new(bucket).await?;
+
+        let config = aws_config::load_defaults(aws_config::BehaviorVersion::latest()).await;
+        let s3_client = S3Client::new(&config);
+        let writer = S3MempoolEventWriter::new(s3_client, bucket);
 
         Ok(Self { reader, writer })
     }

crates/audit/src/reader.rs

@@ -122,4 +122,4 @@ impl MempoolEventReader for KafkaMempoolReader {
         }
         Ok(())
     }
-}
+}

crates/audit/src/storage.rs

@@ -1,9 +1,8 @@
 use crate::types::{BundleId, MempoolEvent, TransactionId};
 use anyhow::Result;
 use async_trait::async_trait;
-use aws_config::BehaviorVersion;
-use aws_sdk_s3::{primitives::ByteStream, Client as S3Client};
-use base64;
+use aws_sdk_s3::primitives::ByteStream;
+use aws_sdk_s3::Client as S3Client;
 use serde::{Deserialize, Serialize};
 use std::collections::HashSet;
 use std::fmt;
@@ -51,14 +50,7 @@ pub struct S3MempoolEventWriter {
 }
 
 impl S3MempoolEventWriter {
-    pub async fn new(bucket: String) -> Result<Self> {
-        let config = aws_config::defaults(BehaviorVersion::latest()).load().await;
-        let s3_client = S3Client::new(&config);
-
-        Ok(Self { s3_client, bucket })
-    }
-
-    pub fn with_client(s3_client: S3Client, bucket: String) -> Self {
+    pub fn new(s3_client: S3Client, bucket: String) -> Self {
         Self { s3_client, bucket }
     }
 
@@ -197,7 +189,6 @@ impl S3MempoolEventWriter {
     async fn put_object_idempotent(&self, key: &str, data: Vec<u8>) -> Result<()> {
         let md5_digest = md5::compute(&data);
         let content_hash_hex = format!("{:x}", md5_digest);
-        let content_hash_base64 = base64::encode(&md5_digest[..]);
 
         if let Ok(existing) = self.get_object_etag(key).await {
             if existing.trim_matches('"') == content_hash_hex {
@@ -211,14 +202,13 @@ impl S3MempoolEventWriter {
         }
 
         let data_size = data.len();
-        let stream = ByteStream::from(data);
+        let body = ByteStream::from(data);
 
         self.s3_client
             .put_object()
             .bucket(&self.bucket)
             .key(key)
-            .body(stream)
-            .content_md5(&content_hash_base64)
+            .body(body)
             .send()
             .await?;
 
@@ -240,7 +230,7 @@ impl S3MempoolEventWriter {
             .send()
             .await?;
 
-        Ok(response.e_tag().unwrap_or_default().to_string())
+        Ok(response.e_tag().unwrap_or("").to_string())
     }
 
     async fn get_bundle_transaction_hashes(&self, bundle_id: BundleId) -> Result<Vec<String>> {
@@ -292,7 +282,7 @@ impl S3MempoolEventWriter {
             .await?;
 
         let body = response.body.collect().await?;
-        Ok(String::from_utf8(body.to_vec())?)
+        Ok(String::from_utf8(body.into_bytes().to_vec())?)
     }
 }