Add rate limiting, fix circuit breaker race condition, setup CI

- Fix TOCTOU race in CircuitBreaker by replacing RwLock with Mutex
- Add token-bucket RateLimiter with configurable rate and burst
- Sanitize error messages to avoid leaking token existence info
- Add GitHub Actions CI pipeline (fmt, clippy, test, build)
- Add cargo-deny config for dependency auditing
- Optimize routing with Arc-wrapped snapshots and arena-based BFS
- Add rustdoc to public APIs
- Add integration tests for end-to-end flows

Author: alexanderattar

.github/workflows/ci.yml

@@ -0,0 +1,73 @@
+name: CI
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
+
+env:
+  CARGO_TERM_COLOR: always
+  RUSTFLAGS: "-Dwarnings"
+
+jobs:
+  check:
+    name: Check
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Install Rust
+        run: |
+          rustup update stable
+          rustup default stable
+      - uses: Swatinem/rust-cache@v2
+      - run: cargo check --all-targets
+
+  fmt:
+    name: Format
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Install Rust
+        run: |
+          rustup update stable
+          rustup default stable
+          rustup component add rustfmt
+      - run: cargo fmt --all --check
+
+  clippy:
+    name: Clippy
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Install Rust
+        run: |
+          rustup update stable
+          rustup default stable
+          rustup component add clippy
+      - uses: Swatinem/rust-cache@v2
+      - run: cargo clippy --all-targets -- -D warnings
+
+  test:
+    name: Test
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Install Rust
+        run: |
+          rustup update stable
+          rustup default stable
+      - uses: Swatinem/rust-cache@v2
+      - run: cargo test --all
+
+  build-release:
+    name: Build Release
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Install Rust
+        run: |
+          rustup update stable
+          rustup default stable
+      - uses: Swatinem/rust-cache@v2
+      - run: cargo build --release

.gitignore

@@ -1,2 +1,3 @@
 /target
-.DS_Store
+.DS_Store
+/notes

deny.toml

@@ -0,0 +1,35 @@
+# cargo-deny configuration
+# Run with: cargo deny check
+
+[advisories]
+vulnerability = "deny"
+unmaintained = "warn"
+yanked = "deny"
+notice = "warn"
+
+[licenses]
+unlicensed = "deny"
+allow = [
+    "MIT",
+    "Apache-2.0",
+    "Apache-2.0 WITH LLVM-exception",
+    "BSD-2-Clause",
+    "BSD-3-Clause",
+    "ISC",
+    "Zlib",
+    "Unicode-DFS-2016",
+]
+copyleft = "warn"
+
+[bans]
+multiple-versions = "warn"
+wildcards = "deny"
+highlight = "all"
+
+# Deny specific crates if needed
+deny = []
+
+[sources]
+unknown-registry = "deny"
+unknown-git = "deny"
+allow-registry = ["https://github.com/rust-lang/crates.io-index"]

packages/aggregator/src/core/event_processor.rs

@@ -7,18 +7,33 @@ use crate::core::format_duration;
 use crate::core::state::{AggregatorState, SharedState};
 use crate::traits::EventProcessor;
 
+/// Errors that can occur during event processing.
 #[derive(Debug, thiserror::Error)]
 pub enum Error {
     #[error("invalid orderbook: {0}")]
     InvalidOrderbook(String),
 }
 
+/// Processes incoming orderbook events and updates the shared aggregator state.
+///
+/// The event processor validates each orderbook update (checking for crossed spreads
+/// and empty books) before ingesting it into the state. Invalid orderbooks are
+/// rejected and tracked in metrics.
+///
+/// # Example
+///
+/// ```ignore
+/// let state = create_shared_state();
+/// let processor = DexEventProcessor::new(state);
+/// processor.process_orderbook(orderbook)?;
+/// ```
 #[derive(Debug, Clone)]
 pub struct DexEventProcessor {
     state: SharedState,
 }
 
 impl DexEventProcessor {
+    /// Create a new event processor with the given shared state.
     pub fn new(state: SharedState) -> Self {
         Self { state }
     }

packages/aggregator/src/core/request_processor.rs

@@ -6,18 +6,40 @@ use crate::core::routing::find_best_route;
 use crate::core::state::SharedState;
 use crate::traits::RequestProcessor;
 
+/// Errors that can occur during request processing.
 #[derive(Debug, thiserror::Error)]
 pub enum Error {
     #[error("routing failed: {0}")]
     RoutingFailed(String),
 }
 
+/// Processes swap requests by finding optimal multi-hop routes.
+///
+/// The request processor enforces rate limiting, circuit breaker protection,
+/// and slippage tolerance before executing swaps. It uses BFS-based routing
+/// to find the best path across multiple orderbooks.
+///
+/// # Features
+///
+/// - Rate limiting to prevent abuse
+/// - Circuit breaker for graceful degradation under failures
+/// - Slippage protection based on user-specified minimum output
+/// - Multi-hop routing (up to 3 hops) for optimal execution
+///
+/// # Example
+///
+/// ```ignore
+/// let state = create_shared_state();
+/// let processor = DexRequestProcessor::new(state);
+/// let response = processor.process_request(swap_request).await?;
+/// ```
 #[derive(Debug, Clone)]
 pub struct DexRequestProcessor {
     state: SharedState,
 }
 
 impl DexRequestProcessor {
+    /// Create a new request processor with the given shared state.
     pub fn new(state: SharedState) -> Self {
         Self { state }
     }
@@ -33,6 +55,16 @@ impl RequestProcessor for DexRequestProcessor {
             .requests_total
             .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
 
+        // Rate limit check: reject if too many requests
+        if !self.state.rate_limiter.try_acquire() {
+            warn!("swap rejected: rate limited");
+            self.state
+                .metrics
+                .requests_rate_limited
+                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+            return Ok(SwapResponse::Failure("rate limited".to_string()));
+        }
+
         // Circuit breaker check: reject early if system is degraded
         if !self.state.circuit_breaker.allow_request() {
             warn!(
@@ -62,29 +94,32 @@ impl RequestProcessor for DexRequestProcessor {
             return Ok(SwapResponse::Failure("zero amount".to_string()));
         }
 
-        // Check tokens exist before searching (better error messages)
+        // Check tokens exist before searching
         let input_known = self.state.contains_token(request.input_token);
         let output_known = self.state.contains_token(request.output_token);
 
         // Reject swaps with unknown tokens (infrastructure issue, affects circuit breaker)
+        // Return generic error to client but log details server-side
         if !input_known || !output_known {
-            let reason = match (input_known, output_known) {
-                (false, false) => "unknown tokens",
-                (false, true) => "unknown input token",
-                (true, false) => "unknown output token",
+            let detail = match (input_known, output_known) {
+                (false, false) => "both tokens unknown",
+                (false, true) => "input token unknown",
+                (true, false) => "output token unknown",
                 _ => unreachable!(),
             };
             warn!(
                 input_token = %request.input_token,
                 output_token = %request.output_token,
-                "swap rejected: {reason}"
+                detail,
+                "swap rejected: invalid request"
             );
             self.state
                 .metrics
                 .requests_failed
                 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
             self.state.circuit_breaker.record_failure();
-            return Ok(SwapResponse::Failure(reason.to_string()));
+            // Generic error to avoid leaking token existence info
+            return Ok(SwapResponse::Failure("invalid request".to_string()));
         }
 
         // Find best route (infrastructure issue if fails, affects circuit breaker)