chore(audit): apply workspace lints

Author: refcell

Cargo.lock

@@ -7,13 +7,14 @@ homepage.workspace = true
 repository.workspace = true
 edition.workspace = true
 
+[lints]
+workspace = true
+
 [dependencies]
 bytes.workspace = true
 metrics.workspace = true
-dotenvy.workspace = true
 async-trait.workspace = true
 metrics-derive.workspace = true
-aws-credential-types.workspace = true
 tips-core = { workspace = true, features = ["test-utils"] }
 serde = { workspace = true, features = ["std", "derive"] }
 tokio = { workspace = true, features = ["full"] }
@@ -23,12 +24,7 @@ anyhow = { workspace = true, features = ["std"] }
 serde_json = { workspace = true, features = ["std"] }
 rdkafka = { workspace = true, features = ["tokio", "libz"] }
 alloy-consensus = { workspace = true, features = ["std"] }
-alloy-provider = { workspace = true, features = ["reqwest"] }
-clap = { version = "4.5.47", features = ["std", "derive", "env"] }
-op-alloy-consensus = { workspace = true, features = ["std", "k256", "serde"] }
 alloy-primitives = { workspace = true, features = ["map-foldhash", "serde"] }
-tracing-subscriber = { workspace = true, features = ["std", "fmt", "env-filter", "json"] }
-aws-config = { workspace = true, features = ["default-https-client", "rt-tokio"] }
 aws-sdk-s3 = { workspace = true, features = ["rustls", "default-https-client", "rt-tokio"] }
 
 [dev-dependencies]

crates/audit/Cargo.toml

@@ -2,10 +2,12 @@ use crate::metrics::Metrics;
 use crate::reader::EventReader;
 use crate::storage::EventWriter;
 use anyhow::Result;
+use std::fmt;
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 use tokio::time::sleep;
 use tracing::{error, info};
 
+/// Archives audit events from Kafka to S3 storage.
 pub struct KafkaAuditArchiver<R, W>
 where
     R: EventReader,
@@ -16,11 +18,22 @@ where
     metrics: Metrics,
 }
 
+impl<R, W> fmt::Debug for KafkaAuditArchiver<R, W>
+where
+    R: EventReader,
+    W: EventWriter + Clone + Send + 'static,
+{
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("KafkaAuditArchiver").finish_non_exhaustive()
+    }
+}
+
 impl<R, W> KafkaAuditArchiver<R, W>
 where
     R: EventReader,
     W: EventWriter + Clone + Send + 'static,
 {
+    /// Creates a new archiver with the given reader and writer.
     pub fn new(reader: R, writer: W) -> Self {
         Self {
             reader,
@@ -29,6 +42,7 @@ where
         }
     }
 
+    /// Runs the archiver loop, reading events and writing them to storage.
     pub async fn run(&mut self) -> Result<()> {
         info!("Starting Kafka bundle archiver");
 

crates/audit/src/archiver.rs

@@ -1,19 +1,47 @@
-pub mod archiver;
-pub mod metrics;
-pub mod publisher;
-pub mod reader;
-pub mod storage;
-pub mod types;
+//! Audit library for tracking and archiving bundle and user operation events.
+//!
+//! This crate provides functionality for publishing events to Kafka,
+//! archiving them to S3, and reading event history.
+
+#![doc(issue_tracker_base_url = "https://github.com/base/tips/issues/")]
+#![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))]
+#![cfg_attr(not(test), warn(unused_crate_dependencies))]
+
+mod archiver;
+pub use archiver::KafkaAuditArchiver;
+
+mod metrics;
+pub use metrics::Metrics;
+
+mod publisher;
+pub use publisher::{
+    BundleEventPublisher, KafkaBundleEventPublisher, KafkaUserOpEventPublisher,
+    LoggingBundleEventPublisher, LoggingUserOpEventPublisher, UserOpEventPublisher,
+};
+
+mod reader;
+pub use reader::{
+    Event, EventReader, KafkaAuditLogReader, KafkaUserOpAuditLogReader, UserOpEventReader,
+    UserOpEventWrapper, assign_topic_partition, create_kafka_consumer,
+};
+
+mod storage;
+pub use storage::{
+    BundleEventS3Reader, BundleHistory, BundleHistoryEvent, EventWriter, S3EventReaderWriter,
+    S3Key, TransactionMetadata, UserOpEventS3Reader, UserOpEventWriter, UserOpHistory,
+    UserOpHistoryEvent,
+};
+
+mod types;
+pub use types::{
+    BundleEvent, BundleId, DropReason, Transaction, TransactionId, UserOpDropReason, UserOpEvent,
+    UserOpHash,
+};
 
 use tokio::sync::mpsc;
 use tracing::error;
 
-pub use archiver::*;
-pub use publisher::*;
-pub use reader::*;
-pub use storage::*;
-pub use types::*;
-
+/// Connects a bundle event receiver to a publisher, spawning a task to forward events.
 pub fn connect_audit_to_publisher<P>(event_rx: mpsc::UnboundedReceiver<BundleEvent>, publisher: P)
 where
     P: BundleEventPublisher + 'static,
@@ -28,6 +56,7 @@ where
     });
 }
 
+/// Connects a user operation event receiver to a publisher, spawning a task to forward events.
 pub fn connect_userop_audit_to_publisher<P>(
     event_rx: mpsc::UnboundedReceiver<UserOpEvent>,
     publisher: P,

crates/audit/src/lib.rs

@@ -1,39 +1,51 @@
 use metrics::{Counter, Gauge, Histogram};
 use metrics_derive::Metrics;
 
+/// Metrics for audit operations including Kafka reads, S3 writes, and event processing.
 #[derive(Metrics, Clone)]
 #[metrics(scope = "tips_audit")]
 pub struct Metrics {
+    /// Duration of archive_event operations.
     #[metric(describe = "Duration of archive_event")]
     pub archive_event_duration: Histogram,
 
+    /// Age of event when processed (now - event timestamp).
     #[metric(describe = "Age of event when processed (now - event timestamp)")]
     pub event_age: Histogram,
 
+    /// Duration of Kafka read_event operations.
     #[metric(describe = "Duration of Kafka read_event")]
     pub kafka_read_duration: Histogram,
 
+    /// Duration of Kafka commit operations.
     #[metric(describe = "Duration of Kafka commit")]
     pub kafka_commit_duration: Histogram,
 
+    /// Duration of update_bundle_history operations.
     #[metric(describe = "Duration of update_bundle_history")]
     pub update_bundle_history_duration: Histogram,
 
+    /// Duration of updating all transaction indexes.
     #[metric(describe = "Duration of update all transaction indexes")]
     pub update_tx_indexes_duration: Histogram,
 
+    /// Duration of S3 get_object operations.
     #[metric(describe = "Duration of S3 get_object")]
     pub s3_get_duration: Histogram,
 
+    /// Duration of S3 put_object operations.
     #[metric(describe = "Duration of S3 put_object")]
     pub s3_put_duration: Histogram,
 
+    /// Total events processed.
     #[metric(describe = "Total events processed")]
     pub events_processed: Counter,
 
+    /// Total S3 writes skipped due to deduplication.
     #[metric(describe = "Total S3 writes skipped due to dedup")]
     pub s3_writes_skipped: Counter,
 
+    /// Number of in-flight archive tasks.
     #[metric(describe = "Number of in-flight archive tasks")]
     pub in_flight_archive_tasks: Gauge,
 }

crates/audit/src/metrics.rs

@@ -2,24 +2,36 @@ use crate::types::{BundleEvent, UserOpEvent};
 use anyhow::Result;
 use async_trait::async_trait;
 use rdkafka::producer::{FutureProducer, FutureRecord};
-use serde_json;
 use tracing::{debug, error, info};
 
+/// Trait for publishing bundle events.
 #[async_trait]
 pub trait BundleEventPublisher: Send + Sync {
+    /// Publishes a single bundle event.
     async fn publish(&self, event: BundleEvent) -> Result<()>;
 
+    /// Publishes multiple bundle events.
     async fn publish_all(&self, events: Vec<BundleEvent>) -> Result<()>;
 }
 
+/// Publishes bundle events to Kafka.
 #[derive(Clone)]
 pub struct KafkaBundleEventPublisher {
     producer: FutureProducer,
     topic: String,
 }
 
+impl std::fmt::Debug for KafkaBundleEventPublisher {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("KafkaBundleEventPublisher")
+            .field("topic", &self.topic)
+            .finish_non_exhaustive()
+    }
+}
+
 impl KafkaBundleEventPublisher {
-    pub fn new(producer: FutureProducer, topic: String) -> Self {
+    /// Creates a new Kafka bundle event publisher.
+    pub const fn new(producer: FutureProducer, topic: String) -> Self {
         Self { producer, topic }
     }
 
@@ -71,11 +83,13 @@ impl BundleEventPublisher for KafkaBundleEventPublisher {
     }
 }
 
-#[derive(Clone)]
+/// Publishes bundle events to logs (for testing/debugging).
+#[derive(Clone, Debug)]
 pub struct LoggingBundleEventPublisher;
 
 impl LoggingBundleEventPublisher {
-    pub fn new() -> Self {
+    /// Creates a new logging bundle event publisher.
+    pub const fn new() -> Self {
         Self
     }
 }
@@ -105,21 +119,34 @@ impl BundleEventPublisher for LoggingBundleEventPublisher {
     }
 }
 
+/// Trait for publishing user operation events.
 #[async_trait]
 pub trait UserOpEventPublisher: Send + Sync {
+    /// Publishes a single user operation event.
     async fn publish(&self, event: UserOpEvent) -> Result<()>;
 
+    /// Publishes multiple user operation events.
     async fn publish_all(&self, events: Vec<UserOpEvent>) -> Result<()>;
 }
 
+/// Publishes user operation events to Kafka.
 #[derive(Clone)]
 pub struct KafkaUserOpEventPublisher {
     producer: FutureProducer,
     topic: String,
 }
 
+impl std::fmt::Debug for KafkaUserOpEventPublisher {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("KafkaUserOpEventPublisher")
+            .field("topic", &self.topic)
+            .finish_non_exhaustive()
+    }
+}
+
 impl KafkaUserOpEventPublisher {
-    pub fn new(producer: FutureProducer, topic: String) -> Self {
+    /// Creates a new Kafka user operation event publisher.
+    pub const fn new(producer: FutureProducer, topic: String) -> Self {
         Self { producer, topic }
     }
 
@@ -171,11 +198,13 @@ impl UserOpEventPublisher for KafkaUserOpEventPublisher {
     }
 }
 
-#[derive(Clone)]
+/// Publishes user operation events to logs (for testing/debugging).
+#[derive(Clone, Debug)]
 pub struct LoggingUserOpEventPublisher;
 
 impl LoggingUserOpEventPublisher {
-    pub fn new() -> Self {
+    /// Creates a new logging user operation event publisher.
+    pub const fn new() -> Self {
         Self
     }
 }