perf: improve audit (#123)

* perf: improve audit

* fix metric

* wip

* add error logs

* worker pool

* make params configurable

* simplify try_join

* try_join_all

* chore: add more audit logs

* perf: deduplicate bundles with the same contents (#137)

* chore: use uuid v5 for determinism

* feat: add ttl

* diffs

* fix: in-flight archive task

* Revert "fix: in-flight archive task"

This reverts commit a759f2b.

* tmp: clear backlog by nooping

* chore: log out meterbundleres

* make ttl configurable

Author: wlawt

.gitignore

@@ -26,4 +26,4 @@ Thumbs.db
 /ui/.claude
 
 # e2e / load tests
-wallets.json
+wallets.json

Cargo.lock

@@ -97,3 +97,5 @@ testcontainers = { version = "0.23.1", default-features = false }
 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 }

Cargo.toml

@@ -53,6 +53,15 @@ struct Args {
 
     #[arg(long, env = "TIPS_AUDIT_METRICS_ADDR", default_value = "0.0.0.0:9002")]
     metrics_addr: SocketAddr,
+
+    #[arg(long, env = "TIPS_AUDIT_WORKER_POOL_SIZE", default_value = "80")]
+    worker_pool_size: usize,
+
+    #[arg(long, env = "TIPS_AUDIT_CHANNEL_BUFFER_SIZE", default_value = "500")]
+    channel_buffer_size: usize,
+
+    #[arg(long, env = "TIPS_AUDIT_NOOP_ARCHIVE", default_value = "false")]
+    noop_archive: bool,
 }
 
 #[tokio::main]
@@ -82,7 +91,13 @@ async fn main() -> Result<()> {
     let s3_bucket = args.s3_bucket.clone();
     let writer = S3EventReaderWriter::new(s3_client, s3_bucket);
 
-    let mut archiver = KafkaAuditArchiver::new(reader, writer);
+    let mut archiver = KafkaAuditArchiver::new(
+        reader,
+        writer,
+        args.worker_pool_size,
+        args.channel_buffer_size,
+        args.noop_archive,
+    );
 
     info!("Audit archiver initialized, starting main loop");
 

bin/tips-audit/src/main.rs

@@ -18,14 +18,15 @@ 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"] }
 rdkafka = { workspace = true, features = ["tokio", "libz", "zstd", "ssl-vendored"] }
 alloy-consensus = { workspace = true, features = ["std"] }
 alloy-primitives = { workspace = true, features = ["map-foldhash", "serde"] }
 aws-sdk-s3 = { workspace = true, features = ["rustls", "default-https-client", "rt-tokio"] }
+futures = { workspace = true }
 
 [dev-dependencies]
 testcontainers = { workspace = true, features = ["blocking"] }